[uss_qualifier] Add flag to ensure redaction of sensitive information

monitoring/monitorlib/dicts.py

@@ -1,5 +1,5 @@
 import json
-from typing import List, Tuple, Any, Union
+from typing import List, Tuple, Any, Union, Iterator
 
 from implicitdict import ImplicitDict
 
@@ -62,6 +62,17 @@ def get_element(obj: dict, element: Union[JSONAddress, List[str]], pop=False) ->
         raise ValueError(f"{str(e)} at {element}")
 
 
+def get_element_or_default(
+    obj: dict, element: Union[JSONAddress, List[str]], default_value: Any
+) -> Any:
+    try:
+        return get_element(obj, element)
+    except KeyError:
+        return default_value
+    except TypeError:
+        return default_value
+
+
 class RemovedElement(ImplicitDict):
     address: JSONAddress
     value = None

monitoring/uss_qualifier/main.py

@@ -10,7 +10,7 @@
 from loguru import logger
 import yaml
 
-from monitoring.monitorlib.dicts import remove_elements
+from monitoring.monitorlib.dicts import remove_elements, get_element_or_default
 from monitoring.monitorlib.versioning import get_code_version, get_commit_hash
 from monitoring.uss_qualifier.configurations.configuration import (
     USSQualifierConfiguration,
@@ -73,6 +73,13 @@ def parseArgs() -> argparse.Namespace:
         help="JSON string containing runtime metadata to record in the test run report (if specified).",
     )
 
+    parser.add_argument(
+        "--disallow-unredacted",
+        type=bool,
+        default=False,
+        help="When true, do not run a test configuration which would produce unredacted sensitive information in its artifacts",
+    )
+
     return parser.parse_args()
 
 
@@ -140,13 +147,32 @@ def execute_test_run(
     )
 
 
+def raise_for_unredacted_information(config: USSQualifierConfiguration) -> None:
+    """Raises a ValueError if the provided configuration would produce or display unredacted information."""
+
+    required_values = {
+        "v1.artifacts.globally_expanded_report.redact_access_tokens": True,
+        "v1.artifacts.raw_report.redact_access_tokens": True,
+        "v1.artifacts.report_html.redact_access_tokens": True,
+        "v1.artifacts.sequence_view.redact_access_tokens": True,
+    }
+
+    for json_address, required_value in required_values.items():
+        actual_value = get_element_or_default(config, json_address, required_value)
+        if actual_value != required_value:
+            raise ValueError(
+                f"Configuration element {json_address} must be {required_value} to disallow unredacted information, but was instead set to {actual_value}"
+            )
+
+
 def run_config(
     config_name: str,
     config_output: str,
     skip_validation: bool,
     exit_before_execution: bool,
     output_path: Optional[str],
     runtime_metadata: Optional[dict],
+    disallow_unredacted: bool,
 ):
     config_src = load_dict_with_references(config_name)
 
@@ -192,6 +218,9 @@ def run_config(
         logger.info("Exiting because --exit-before-execution specified.")
         return
 
+    if disallow_unredacted:
+        raise_for_unredacted_information(whole_config)
+
     config: USSQualifierConfigurationV1 = whole_config.v1
 
     if config.artifacts and not output_path:
@@ -206,7 +235,7 @@ def run_config(
         report.runtime_metadata = runtime_metadata
 
     if config.artifacts:
-        generate_artifacts(report, config.artifacts, output_path)
+        generate_artifacts(report, config.artifacts, output_path, disallow_unredacted)
 
     if "validation" in config and config.validation:
         logger.info(f"Validating test run report for configuration '{config_name}'")
@@ -228,6 +257,8 @@ def main() -> int:
     if runtime_metadata is not None and not isinstance(runtime_metadata, dict):
         raise ValueError("--runtime-metadata must specify a JSON dictionary")
 
+    disallow_unredacted = args.disallow_unredacted
+
     config_names = str(args.config).split(",")
 
     if args.config_output:
@@ -254,6 +285,7 @@ def main() -> int:
             args.exit_before_execution,
             output_path,
             runtime_metadata,
+            disallow_unredacted,
         )
         if exit_code != os.EX_OK:
             return exit_code

monitoring/uss_qualifier/make_artifacts.py

@@ -88,7 +88,7 @@ def main() -> int:
                 output_path = default_output_path(report_name)
             else:
                 output_path = default_output_path(config_name)
-            generate_artifacts(report, config.artifacts, output_path)
+            generate_artifacts(report, config.artifacts, output_path, False)
         else:
             output_path = "nowhere"
             logger.warning(f"No artifacts to generate for {config_name}")

monitoring/uss_qualifier/reports/artifacts.py

@@ -34,12 +34,18 @@ def generate_artifacts(
     report: TestRunReport,
     artifacts: ArtifactsConfiguration,
     output_path: str,
+    disallow_unredacted: bool,
 ):
     logger.debug(f"Writing artifacts to {os.path.abspath(output_path)}")
     os.makedirs(output_path, exist_ok=True)
 
     def _should_redact(cfg) -> bool:
-        return "redact_access_tokens" in cfg and cfg.redact_access_tokens
+        result = "redact_access_tokens" in cfg and cfg.redact_access_tokens
+        if disallow_unredacted and not result:
+            raise RuntimeError(
+                "The option to disallow unredacted information was set, but the configuration specified unredacted information any way"
+            )
+        return result
 
     logger.info(f"Redacting access tokens from report")
     redacted_report = ImplicitDict.parse(json.loads(json.dumps(report)), TestRunReport)