diff --git a/openadapt_flow/qualification.py b/openadapt_flow/qualification.py index 201b5143..9a9c80d6 100644 --- a/openadapt_flow/qualification.py +++ b/openadapt_flow/qualification.py @@ -5478,6 +5478,11 @@ def save_qualified_workflow( raise QualificationError( "governed authorization template changed the bundle content digest" ) + workflow_py = Path(bundle_dir) / "workflow.py" + if workflow_py.is_file(): + from openadapt_flow.compiler.codegen import render_workflow_py + + workflow_py.write_text(render_workflow_py(workflow), encoding="utf-8") return path diff --git a/openadapt_flow/sanitized_artifact.py b/openadapt_flow/sanitized_artifact.py index 33a9f391..73a08df4 100644 --- a/openadapt_flow/sanitized_artifact.py +++ b/openadapt_flow/sanitized_artifact.py @@ -310,9 +310,21 @@ def _bundle_string_is_structural(path: tuple[str | int, ...], value: str) -> boo key = path[-1] if path and isinstance(path[-1], str) else "" if key.endswith(("_sha256", "_digest", "_hash", "_version")): return True + if key == "sha256" and re.fullmatch(r"[0-9a-f]{64}", value): + return True if key.endswith("_at") and value: return True - if key in {"recording_id", "id"} and _OPAQUE_MACHINE_ID.fullmatch(value): + if key in {"recording_id", "id", "step_id"} and _OPAQUE_MACHINE_ID.fullmatch(value): + return True + if key.endswith("_id") and re.fullmatch( + r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", value, re.IGNORECASE + ): + return True + if key == "environment_observer_id" and re.fullmatch( + r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", value + ): + return True + if key == "relation" and value in {"left_of", "right_of", "above", "below"}: return True if key in {"c", "r", "phash", "salt", "structured"} and re.fullmatch( r"[0-9a-f]{16,}", value diff --git a/tests/test_qualification_project.py b/tests/test_qualification_project.py index cf7618a0..d00e7d55 100644 --- a/tests/test_qualification_project.py +++ b/tests/test_qualification_project.py @@ -2448,11 +2448,14 @@ def test_minimum_effect_tier_versions_round_trips_and_invalidates_certification( def test_full_campaign_certifies_through_existing_policy_and_round_trips( tmp_path: Path, ) -> None: + from openadapt_flow.compiler.codegen import render_workflow_py + workflow = _workflow() bundle = tmp_path / "bundle" (bundle / "templates").mkdir(parents=True) (bundle / "templates" / "save.png").write_bytes(_qualification_visual_fixture()[1]) workflow.save(bundle) + (bundle / "workflow.py").write_text(render_workflow_py(workflow), encoding="utf-8") workflow = Workflow.load(bundle) _configure(workflow, tier=VerificationTier.INDEPENDENT_SYSTEM) evidence_root = tmp_path / "evidence" @@ -2472,6 +2475,9 @@ def test_full_campaign_certifies_through_existing_policy_and_round_trips( save_qualified_workflow(workflow, bundle) loaded = Workflow.load(bundle) + assert (bundle / "workflow.py").read_text(encoding="utf-8") == ( + render_workflow_py(loaded) + ) assert loaded.qualification is not None assert loaded.qualification.schema_version == "openadapt.qualification-project/v1" assert loaded.qualification.last_certification is not None diff --git a/tests/test_sanitized_artifact.py b/tests/test_sanitized_artifact.py index 4e6121b6..82fec3ac 100644 --- a/tests/test_sanitized_artifact.py +++ b/tests/test_sanitized_artifact.py @@ -32,6 +32,7 @@ APPROVAL_NAME, MANIFEST_NAME, SanitizationError, + _bundle_string_is_structural, _review_content_length, _valid_review_host, add_manual_image_redaction, @@ -81,6 +82,24 @@ def scrub_image(self, image, fill_color=None): return image +@pytest.mark.parametrize( + ("path", "value"), + [ + (("qualification", "cases", 0, "evidence", 0, "sha256"), "a" * 64), + (("steps", 0, "id"), "step_000"), + (("qualification", "identity_policies", "step_000", "step_id"), "step_000"), + (("qualification", "project_id"), "f6d239d0-0b5e-440f-b794-bbe69cc01115"), + ( + ("qualification", "environment", "environment_observer_id"), + "backend:PlaywrightBackend", + ), + (("steps", 0, "anchor", "landmarks", 0, "relation"), "left_of"), + ], +) +def test_bundle_machine_contract_strings_are_not_sent_to_phi_ner(path, value): + assert _bundle_string_is_structural(path, value) + + @pytest.fixture(autouse=True) def _scrubber(tmp_path, monkeypatch): privacy.set_text_scrubber(FakeScrubber())