Add backward compatibility to secret masking for output schema

CHANGELOG.rst

@@ -23,7 +23,6 @@ Fixed
 
   Contributed by @bharath-orchestral
 
-
 * Fixed ``st2client/st2client/base.py`` file to use ``https_proxy``(not ``http_proxy``) to check HTTPS_PROXY environment variables.
 
   Contributed by @wfgydbu
@@ -39,10 +38,16 @@ Fixed
 * Fixed generation of `st2.conf.sample` to show correct syntax for `[sensorcontainer].partition_provider` (space separated `key:value` pairs). #5710
   Contributed by @cognifloyd
 
-* Fix access to key-value pairs in workflow and action execution where RBAC rules did not get applied
+* Fix access to key-value pairs in workflow and action execution where RBAC rules did not get applied #5764
 
   Contributed by @m4dcoder
 
+* Add backward compatibility to secret masking introduced in #5319 to prevent security-relative issues.
+  Migration to the new schema is required to take advantage of the full output schema validation. #5783
+
+  Contributed by @m4dcoder
+
+
 Added
 ~~~~~
 

st2common/st2common/util/output_schema.py

@@ -36,19 +36,39 @@ def _output_schema_is_valid(_schema):
     if not isinstance(_schema, Mapping):
         # malformed schema
         return False
+
+    if "type" not in _schema:
+        # legacy schema format
+        return False
+
     try:
+        # the validator is smart enough to handle
+        # schema that is similar to the input schema
         schema.validate(
             _schema,
             schema.get_action_output_schema(),
             cls=schema.get_validator("custom"),
         )
     except jsonschema.ValidationError as e:
         LOG.debug("output_schema not valid: %s", e)
-        # likely a legacy partial object schema (only defines properties)
         return False
+
     return True
 
 
+def _normalize_legacy_output_schema(_schema):
+    if not isinstance(_schema, Mapping):
+        return _schema
+
+    _normalized_schema = {
+        "type": "object",
+        "properties": _schema,
+        "additionalProperties": True,
+    }
+
+    return _normalized_schema
+
+
 def _validate_runner(runner_schema, result):
     LOG.debug("Validating runner output: %s", runner_schema)
 
@@ -183,15 +203,31 @@ def mask_secret_output(ac_ex, output_value):
         or output_key not in output_value
         # no action output_schema defined
         or not output_schema
-        # malformed action output_schema
-        or not _output_schema_is_valid(output_schema)
     ):
         # nothing to mask
         return output_value
 
+    # backward compatibility for legacy output_schema so secrets stay masked
+    if not _output_schema_is_valid(output_schema):
+        # normalized the legacy schema to a full JSON schema and check if it is valid
+        normalized_output_schema = _normalize_legacy_output_schema(output_schema)
+
+        if not _output_schema_is_valid(normalized_output_schema):
+            # nothing to mask
+            return output_value
+
+        # mask secret for the legacy output schema
+        output_value[output_key] = _get_masked_value(
+            normalized_output_schema, output_value[output_key]
+        )
+
+        return output_value
+
+    # mask secret for the output schema
     output_value[output_key] = _get_masked_value(
         output_schema, output_value[output_key]
     )
+
     return output_value
 
 

st2common/tests/unit/test_util_output_schema.py

@@ -389,7 +389,7 @@ def test_mask_secret_output_noop(self):
         masked_output = output_schema.mask_secret_output(ac_ex, ac_ex_result)
         self.assertDictEqual(masked_output, expected_masked_output)
 
-    def test_mask_secret_output_noop_legacy_schema(self):
+    def test_mask_secret_output_with_legacy_schema(self):
         ac_ex = {
             "action": {
                 "output_schema": LEGACY_ACTION_OUTPUT_SCHEMA,
@@ -399,10 +399,17 @@ def test_mask_secret_output_noop_legacy_schema(self):
                 "output_schema": RUNNER_OUTPUT_SCHEMA,
             },
         }
-        ac_ex_result = {"output_1": "foobar"}
-        expected_masked_output = {"output_1": "foobar"}
 
-        # Legacy schemas should be ignored since they aren't full json schemas.
+        ac_ex_result = {OUTPUT_KEY: {"output_1": "foobar", "output_3": "fubar"}}
+
+        expected_masked_output = {
+            OUTPUT_KEY: {
+                "output_1": "foobar",
+                "output_3": MASKED_ATTRIBUTE_VALUE,
+            }
+        }
+
+        # Legacy schemas are not full json schemas, but they should still mask secrets.
         masked_output = output_schema.mask_secret_output(ac_ex, ac_ex_result)
         self.assertDictEqual(masked_output, expected_masked_output)