from __future__ import annotations import copy import json import tempfile import unittest from pathlib import Path from scripts.lib import test1_runtime_evidence as evidence class RuntimeSecretSourceTests(unittest.TestCase): def test_classifies_runtime_credential_names_and_safe_diagnostics(self) -> None: for key in ( "JWT_SECRET", "NEXUS_ADMIN_PASSWORD", "GH_TOKEN", "HAPIKEY", "secretValue", "secret_value", "private_key_base64", "AWS_SECRET_ACCESS_KEY", "MYSQL_PWD", "SMTP_USER", "PROD_SMTP_USER", "TIDB_USER", "TIDB_CLOUD_PUBLIC_KEY", "TIDB_CLOUD_PRIVATE_KEY", "TIDB_CLOUD_BRANCH_ID", "GOOGLE_CLIENT_ID", "GOOGLE_CLIENT_SECRET", "Authorization", "Set-Cookie", "authorization_code", "codeVerifier", "passwordConfirmation", "Access_Token__c", "Client_Secret__pc", ): with self.subTest(key=key): self.assertTrue(evidence._is_sensitive_key(key)) for key in ( "tokens", "missing_lease_token_count", "password_reset_count", "secret_name", "token_file", "bearer_fingerprint", "proof_hash_prefix", ): with self.subTest(key=key): self.assertFalse(evidence._is_sensitive_key(key)) class KnownStructuredValueRedactionTests(unittest.TestCase): def test_redacts_classifier_owned_fields_and_exact_siblings(self) -> None: report = { "TIDB_PASSWORD": "runtime-password", "TIDB_CLOUD_BRANCH_ID": 12345, "branch_id": 12345, "JWT_SECRET": {"value": "nested-secret", "source": "runtime.env"}, "details": ( "runtime-password 12345 value source nested-secret runtime.env " "explicit-secret tuple-secret" ), "tuple": ("tuple-secret", {"safe": "unchanged"}), } redacted = evidence.redact_known_structured_values( report, ["explicit-secret", "tuple-secret"], ) self.assertEqual(redacted["TIDB_PASSWORD"], "") self.assertEqual(redacted["TIDB_CLOUD_BRANCH_ID"], "") self.assertEqual(redacted["branch_id"], "") self.assertEqual(redacted["JWT_SECRET"], "") self.assertEqual(redacted["tuple"], ["", {"safe": "unchanged"}]) self.assertEqual(redacted["details"], " value source ") def test_keeps_local_tidb_user_field_scoped_and_propagates_hosted_values(self) -> None: local = evidence.redact_known_structured_values( { "TIDB_USER": "root", "details": "root root_cause bootstrap_root", }, [], ) hosted = evidence.redact_known_structured_values( { "TIDB_USER": {"value": "3bwL9f94faN7dCv.root"}, "details": "connected as 3bwL9f94faN7dCv.root", }, [], ) self.assertEqual(local["TIDB_USER"], "") self.assertEqual(local["details"], "root root_cause bootstrap_root") self.assertEqual(hosted["TIDB_USER"], "") self.assertEqual(hosted["details"], "connected as ") def test_marker_free_replacement_cannot_reintroduce_secret_substrings(self) -> None: redacted = evidence.redact_known_structured_values( { "REDA-key": "REDACTED", "details": "REDA and REDACTED; axb", }, ["REDA", "REDACTED", "x", "ab"], ) self.assertEqual(redacted, {"-key": "", "details": " and ; "}) def passing_connected_work_summary() -> dict[str, object]: commit = "a" * 40 return { "status": f"PASS_CONNECTED_WORK_MATRIX_{commit[:7]}", "targetEnv": "test1", "gates": [ { "gate": evidence.TEST1_RUNTIME_GATE, "status": "PASS_RUNTIME_REF_SCHEMA_GATE", "checks": { "health200": True, "moduleReadbackMatches": True, "schemaAuthorityPresent": True, "workItemClaimEnabled": True, "workerCountExpected": True, "maxProcessingRowsExpected": True, "queueProfileExpected": True, }, "runtime": { "expectedSha": commit, "observed": { "schemaAuthorityPresent": True, "workItemClaimEnabled": True, "workerCountMatches": True, "maxProcessingRowsMatches": True, "queueProfileMatches": True, }, "requirements": { "requireSchemaAuthority": True, "requireWorkItemClaim": True, "requireWorkerCount": True, "requireMaxProcessingRows": True, "requireQueueProfile": True, "workers": 16, "maxProcessingRows": 32, "queueProfile": "default-heavy", }, "readback": { "refs": { "CONNECTORS_MODULE_READBACK_REF": commit, "CONNECTORS_MODULE_VERSION": "v0.1.19", } }, }, }, { "gate": evidence.TEST1_WORKITEM_GATE, "status": f"PASS_{evidence.TEST1_WORKITEM_GATE}", "profile": evidence.TEST1_WORKITEM_PROFILE, "results": [ { "name": evidence.TEST1_WORKITEM_RESULT, "events": 20, "lanes": 20, "workers": 16, "accepted": 20, "materialized": 20, "work_materialized": 20, "claimed": 20, "processed": 20, "failed": 0, "retry": 0, "dead": 0, "nonterminal": 0, "installed_pack_id_sha256": "b" * 64, "generation": 7, "claim_scope_workitem_count": 20, "active_pack_match": True, "candidate_claim_plan": "workitem_row_cas", "explain_available": True, "explain_has_broad_scan": False, "scheduler_lease_read_path_used": False, "lane_ordering_ok": True, "verdict": "PASS", } ], }, ], } class Test1ConnectedWorkEvidenceTests(unittest.TestCase): def test_builds_only_the_allowlisted_certificate_from_live_gate_results(self) -> None: summary = passing_connected_work_summary() summary["unapproved"] = {"Cookie": "browser-secret"} summary["gates"][1]["results"][0]["diagnostics"] = { "Authorization": "Bearer api-secret" } report = evidence.build_test1_connected_work_evidence(summary) serialized = json.dumps(report, sort_keys=True) self.assertEqual( set(report), {"schema_version", "verdict", "target_env", "module_identity", "workitem"}, ) self.assertNotIn("browser-secret", serialized) self.assertNotIn("api-secret", serialized) self.assertEqual(report["workitem"]["claim_path"], "workitem_row_cas") self.assertFalse(report["workitem"]["scheduler_lease_read_path_used"]) def test_rejects_incomplete_or_coerced_live_evidence(self) -> None: for key, value in (("claimed", 19), ("failed", 0.0)): with self.subTest(key=key, value=value): summary = passing_connected_work_summary() summary["gates"][1]["results"][0][key] = value with self.assertRaises(evidence.EvidenceBlocked): evidence.build_test1_connected_work_evidence(summary) def test_rejects_unbounded_or_wrong_queue_runtime_profiles(self) -> None: for key in ("maxProcessingRowsMatches", "queueProfileMatches"): with self.subTest(key=key): summary = passing_connected_work_summary() summary["gates"][0]["runtime"]["observed"][key] = False with self.assertRaises(evidence.EvidenceBlocked): evidence.build_test1_connected_work_evidence(summary) def test_rejects_caller_defined_runtime_profile_expectations(self) -> None: for key, value in ( ("workers", 64), ("maxProcessingRows", 64), ("queueProfile", "unbounded"), ): with self.subTest(key=key, value=value): summary = passing_connected_work_summary() summary["gates"][0]["runtime"]["requirements"][key] = value with self.assertRaises(evidence.EvidenceBlocked): evidence.build_test1_connected_work_evidence(summary) def test_rejects_disabled_runtime_profile_requirements(self) -> None: for key in ( "requireSchemaAuthority", "requireWorkItemClaim", "requireWorkerCount", "requireMaxProcessingRows", "requireQueueProfile", ): with self.subTest(key=key): summary = passing_connected_work_summary() summary["gates"][0]["runtime"]["requirements"][key] = False with self.assertRaises(evidence.EvidenceBlocked): evidence.build_test1_connected_work_evidence(summary) def test_writer_rejects_unknown_fields_before_creating_a_file(self) -> None: report = evidence.build_test1_connected_work_evidence( passing_connected_work_summary() ) report["diagnostics"] = {"raw": "not-approved"} with tempfile.TemporaryDirectory() as raw: output = Path(raw) / "evidence.json" with self.assertRaisesRegex( evidence.EvidenceBlocked, "missing or unapproved fields" ): evidence.write_test1_connected_work_evidence( output, report, sensitive_values=[], ) self.assertFalse(output.exists()) def test_writer_blocks_a_known_secret_in_an_otherwise_valid_field(self) -> None: report = evidence.build_test1_connected_work_evidence( passing_connected_work_summary() ) sensitive_commit = report["module_identity"]["expected_commit"] with tempfile.TemporaryDirectory() as raw: output = Path(raw) / "evidence.json" with self.assertRaisesRegex(evidence.EvidenceBlocked, "sensitive value"): evidence.write_test1_connected_work_evidence( output, copy.deepcopy(report), sensitive_values=[sensitive_commit], ) self.assertFalse(output.exists()) def test_writer_atomically_persists_the_valid_certificate(self) -> None: report = evidence.build_test1_connected_work_evidence( passing_connected_work_summary() ) with tempfile.TemporaryDirectory() as raw: output = Path(raw) / "nested" / "evidence.json" evidence.write_test1_connected_work_evidence( output, report, sensitive_values=[], ) self.assertEqual(json.loads(output.read_text(encoding="utf-8")), report) self.assertEqual(list(output.parent.glob("*.tmp")), []) if __name__ == "__main__": unittest.main()