#!/usr/bin/env python3 """Project a minimal, synthetic proxy-trial evidence packet. This standard-library example has no network or file input. It demonstrates allow-list projection and rejection checks; it does not scan real logs or guarantee that a packet is free of every secret or sensitive datum. """ import json import re import sys from datetime import datetime, timezone ALLOWED = ( "packet_id", "purpose", "permission_ref", "target_class", "window_utc", "client_build", "route_label", "outcome", "status_family", "duration_bucket_ms", "validator_version", "validator_result", "limitations", ) FORBIDDEN_MARKERS = ( "CREDENTIAL_URL_CANARY", "AUTH_HEADER_CANARY", "COOKIE_CANARY", "QUERY_SECRET_CANARY", "PRIVATE_URL_CANARY", "RAW_ERROR_CANARY", ) PACKET_VALUES = { "purpose": {"authorized-trial-evidence-review"}, "permission_ref": {"SYNTHETIC-PERMISSION-REF"}, "target_class": {"approved-product-detail"}, "client_build": {"synthetic-client-1.0"}, "route_label": {"trial-route-A"}, "outcome": {"completed", "failed", "unknown"}, "status_family": {"1xx", "2xx", "3xx", "4xx", "5xx", "none"}, "duration_bucket_ms": {"under-100", "100-249", "250-999", "1000-plus", "unknown"}, "validator_version": {"synthetic-validator-1"}, "validator_result": {"required-fields-present", "required-fields-missing", "unknown"}, "limitations": {"synthetic-only; no provider, target, or credential was tested"}, } PACKET_ID = re.compile(r"SYN-\d{8}-\d{3}\Z") WINDOW = re.compile(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z/\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z\Z") def contains_marker(value): if isinstance(value, str): return any(marker in value for marker in FORBIDDEN_MARKERS) if isinstance(value, dict): return any(contains_marker(item) for item in value.values()) if isinstance(value, (list, tuple)): return any(contains_marker(item) for item in value) return False def valid_ordered_window(value): """Validate a real UTC interval after its fixed textual form has matched.""" if not WINDOW.fullmatch(value): return False try: start_text, end_text = value.split("/") start = datetime.strptime(start_text, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) end = datetime.strptime(end_text, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) except ValueError: return False return end > start def project_event(event): """Accept only a complete, scalar, synthetic teaching packet schema. Real logs and raw request/response blobs are intentionally not an input type. Errors are static so an invalid value cannot be reflected into an error message. """ if not isinstance(event, dict): raise TypeError("packet must be a mapping") if set(event) != set(ALLOWED): raise ValueError("packet schema mismatch") if any(not isinstance(event[key], str) for key in ALLOWED): raise ValueError("packet values must be scalar strings") if contains_marker(event): raise ValueError("packet contains a prohibited value") if not PACKET_ID.fullmatch(event["packet_id"]) or not valid_ordered_window(event["window_utc"]): raise ValueError("packet identifier or window is invalid") if any(event[key] not in values for key, values in PACKET_VALUES.items()): raise ValueError("packet value is outside the synthetic vocabulary") return dict(event) def synthetic_event(): """One complete, explicitly synthetic packet; it contains no raw log data.""" return { "packet_id": "SYN-20261031-001", "purpose": "authorized-trial-evidence-review", "permission_ref": "SYNTHETIC-PERMISSION-REF", "target_class": "approved-product-detail", "window_utc": "2026-10-31T01:00:00Z/2026-10-31T01:05:00Z", "client_build": "synthetic-client-1.0", "route_label": "trial-route-A", "outcome": "completed", "status_family": "2xx", "duration_bucket_ms": "100-249", "validator_version": "synthetic-validator-1", "validator_result": "required-fields-present", "limitations": "synthetic-only; no provider, target, or credential was tested", } def self_test(): event = synthetic_event() packet = project_event(event) assert tuple(packet) == ALLOWED assert not contains_marker(packet) encoded = json.dumps(packet, sort_keys=True) assert all(marker not in encoded for marker in FORBIDDEN_MARKERS) for key in ALLOWED: for value in ("PREVIOUSLY_UNSEEN_SECRET_CANARY", {"raw": "PREVIOUSLY_UNSEEN_SECRET_CANARY"}, "x" * 400): bad = dict(event); bad[key] = value try: project_event(bad) except ValueError: pass else: raise AssertionError("unbounded packet field accepted") for invalid in (True, "9xx", "250", "unreviewed"): bad = dict(event); bad["status_family"] = invalid try: project_event(bad) except ValueError: pass else: raise AssertionError("invalid status accepted") completed_invalid = dict(event); completed_invalid["validator_result"] = "required-fields-missing" assert project_event(completed_invalid)["status_family"] == "2xx" assert project_event(completed_invalid)["validator_result"] == "required-fields-missing" unknown = dict(event); unknown.update(outcome="unknown", status_family="none", duration_bucket_ms="unknown", validator_result="unknown") assert project_event(unknown)["outcome"] == "unknown" for window in ( "2026-13-31T01:00:00Z/2026-10-31T01:05:00Z", "2026-02-30T01:00:00Z/2026-10-31T01:05:00Z", "2026-10-31T25:00:00Z/2026-10-31T01:05:00Z", "2026-10-31T01:05:00Z/2026-10-31T01:00:00Z", "2026-10-31T01:00:00Z/2026-10-31T01:00:00Z", "20261031T010000Z/20261031T010500Z", ): bad = dict(event); bad["window_utc"] = window try: project_event(bad) except ValueError as error: assert str(error) == "packet identifier or window is invalid" else: raise AssertionError("invalid window accepted") for field in ("raw_log", "proxy_url", "request_headers", "cookie", "private_url", "raw_error"): bad = dict(event); bad[field] = "CREDENTIAL_URL_CANARY" try: project_event(bad) except ValueError as error: assert str(error) == "packet schema mismatch" else: raise AssertionError("raw input field accepted") return packet def main(): if len(sys.argv) > 2 or (len(sys.argv) == 2 and sys.argv[1] != "--self-test"): print("usage: proxy-trial-evidence-sharing.py [--self-test]", file=sys.stderr) return 2 print(json.dumps(self_test(), indent=2, sort_keys=True)) print("PASS: synthetic allow-list projection rejects packet canaries; no network, credentials, provider, or target was used.") print("LIMIT: human review and a restricted destination remain required; this example cannot guarantee secret-free files.") return 0 if __name__ == "__main__": raise SystemExit(main())