#!/usr/bin/env python3 """Generate deterministic, signed NON-PRODUCTION acceptance fixtures.""" import argparse, base64, hashlib, importlib.util, json, pathlib, shutil, subprocess, tempfile ROOT = pathlib.Path(__file__).resolve().parent STAMP = "2026-08-23T00:00:00Z" VERSION = "1.0.0" TEST_KEY_LABEL = "NON-PRODUCTION TEST KEY" OWNER_MARKER = ".auth-split-fixture-generator-v1" def prepare_output(path): path = path.absolute() protected = {pathlib.Path('/'), pathlib.Path.home().resolve(), ROOT.resolve(), ROOT.parent.resolve()} if path.resolve(strict=False) in protected or path in ROOT.resolve().parents or path in pathlib.Path.home().resolve().parents: raise SystemExit("GENERATOR=BLOCKED reason=unsafe-output-path") cursor = path while cursor != cursor.parent: if cursor.exists() and cursor.is_symlink(): raise SystemExit("GENERATOR=BLOCKED reason=unsafe-output-path") cursor = cursor.parent if path.exists(): marker=path/OWNER_MARKER if not path.is_dir() or not marker.is_file() or marker.read_text(encoding='utf-8')!='auth-split-fixture-generator-v1\n': raise SystemExit("GENERATOR=BLOCKED reason=output-not-generator-owned") shutil.rmtree(path) path.mkdir(parents=True) (path/OWNER_MARKER).write_text('auth-split-fixture-generator-v1\n',encoding='utf-8') def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() def sha_bytes(value): return hashlib.sha256(value).hexdigest() def write_json(path, value): path.write_text(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n", encoding="utf-8") def write_ndjson(path, rows): path.write_text("".join(json.dumps(row, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + "\n" for row in rows), encoding="utf-8") class OpenSSLSigner: """Deterministic Ed25519 fixture signer backed by system OpenSSL.""" def __init__(self, seed): self.private_der = bytes.fromhex("302e020100300506032b657004220420") + seed try: result = subprocess.run( ["/usr/bin/openssl", "pkey", "-inform", "DER", "-pubout", "-outform", "DER"], input=self.private_der, capture_output=True, check=True, ) except (FileNotFoundError, subprocess.CalledProcessError) as exc: raise SystemExit("GENERATOR=BLOCKED reason=ed25519-validator-unavailable") from exc prefix = bytes.fromhex("302a300506032b6570032100") if not result.stdout.startswith(prefix) or len(result.stdout) != len(prefix) + 32: raise SystemExit("GENERATOR=BLOCKED reason=ed25519-validator-unavailable") self.public_raw = result.stdout[len(prefix):] def sign(self, payload): with tempfile.TemporaryDirectory(prefix="auth-split-sign-") as directory: key = pathlib.Path(directory) / "fixture-key.der" message = pathlib.Path(directory) / "message.bin" key.write_bytes(self.private_der) message.write_bytes(payload) try: result = subprocess.run( ["/usr/bin/openssl", "pkeyutl", "-sign", "-rawin", "-keyform", "DER", "-inkey", str(key), "-in", str(message)], capture_output=True, check=True, ) except (FileNotFoundError, subprocess.CalledProcessError) as exc: raise SystemExit("GENERATOR=BLOCKED reason=ed25519-validator-unavailable") from exc if len(result.stdout) != 64: raise SystemExit("GENERATOR=BLOCKED reason=ed25519-validator-unavailable") return result.stdout def keypair(label): seed = hashlib.sha256(label.encode()).digest() try: from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey except ImportError: private = OpenSSLSigner(seed) public = private.public_raw else: private = Ed25519PrivateKey.from_private_bytes(seed) public = private.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) return private, base64.b64encode(public).decode() def verifier_module(): spec = importlib.util.spec_from_file_location("auth_split_verify", ROOT / "verify.py") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def signed_manifest(disconnected=False): verifier = verifier_module() owner_private, owner_public = keypair("auth-split-owner-fixture-v1") security_private, security_public = keypair("auth-split-security-fixture-v1") classes = sorted(verifier.HOP_CLASSES) entries = [] split = len(classes) // 2 for index, hop_class in enumerate(classes): component_start = index == 0 or (disconnected and index == split) component_end = index == len(classes) - 1 or (disconnected and index == split - 1) entry = { "hop_id": f"hop-{index + 1:02d}-{hop_class}", "class": hop_class, "product": "deterministic-fixture", "version": "1", "environment": "production", "owner": f"Acceptance fixture owner ({TEST_KEY_LABEL})", "owner_public_key": owner_public, "upstream": [] if component_start else [f"hop-{index:02d}-{classes[index - 1]}"], "downstream": [] if component_end else [f"hop-{index + 2:02d}-{classes[index + 1]}"], "data_classes": ["cookie"], "capture_policy": "redacted", "control_revision": "fixture-v1", "retention": "fixture-only", "region": "fixture", "access_policy": "fixture-only", "probe_id": f"probe-{index + 1:02d}", "enabled": True, "disabled_reason": None, } entry["owner_signature"] = base64.b64encode(owner_private.sign(canonical(entry))).decode() entries.append(entry) manifest = { "schema_version": VERSION, "manifest_revision": "fixture-v1", "generated_at": STAMP, "signature_algorithm": "Ed25519", "entries": entries, "security_owner": f"Acceptance fixture security owner ({TEST_KEY_LABEL})", "security_public_key": security_public, "closure_assertion": "Every byte path from browser creation or ingress through processing, observability, support, backup, and deletion is represented; there is no uninventoryed hop.", } manifest["security_signature"] = base64.b64encode(security_private.sign(canonical(manifest))).decode() return manifest def outcome(): return { "status": 200, "canonical_body_sha256": sha_bytes(b"fixture-body"), "headers_present": {"content-type": "application/json"}, "headers_absent": ["set-cookie"], "cookie_metadata": [], "retry_outcome": "not-applicable", "reuse_outcome": "not-applicable", "generation_outcome": "unchanged", } def generate(out, scenario): verifier = verifier_module() publication = out.with_name(out.name + "-publication") for directory in (out, publication): prepare_output(directory) (out / "traces").mkdir() (out / "evidence").mkdir() trace = b"deterministic sanitized fixture trace\n" evidence = b"deterministic sanitized fixture evidence\n" (out / "traces" / "fixture.trace").write_bytes(trace) (out / "evidence" / "fixture.json").write_bytes(evidence) manifest = signed_manifest(scenario == "disconnected-graph") write_json(out / "auth-sensitive-data-flow.json", manifest) manifest_digest = sha_bytes((out / "auth-sensitive-data-flow.json").read_bytes()) run_id = f"non-production-{scenario}-fixture" digest = "sha256:" + "1" * 64 environment = { "schema_version": VERSION, "production_equivalent": True, "run_id": run_id, "started_at": STAMP, "git_sha": "1" * 40, "dirty": False, "api_image_digest": digest, "web_image_digest": digest, "package_versions": {"auth": "fixture", "auth-react": "fixture"}, "migration_ids": ["fixture"], "origins": {"api": "https://api.press.zone", "web": "https://tools.press.zone"}, "tls_fingerprints": [digest, "sha256:" + "2" * 64], "cookie_policy_revision": "fixture-v1", "auth_revision": "fixture-v1", "db": {"identity": "fixture", "region": "fixture"}, "network_hops": [{"instance": "fixture", "config_digest": digest}], "telemetry_hops": [{"instance": "fixture", "config_digest": digest}], "browser": {"name": "fixture", "version": "1", "os": "fixture"}, "runtime_digest": digest, "feature_flags": {"fixture": True}, "crash_seam_revision": "fixture-v1", "redaction_manifest": {"path": "auth-sensitive-data-flow.json", "sha256": "sha256:" + manifest_digest}, "deployment_id": f"fixture-{scenario}", "fixture_environment": {"mode":"fixture","registry_id":"auth-split-fixture-v1"}, } write_json(out / "environment.json", environment) write_json(out / "sink-inventory.json", {"schema_version": VERSION, "run_id": run_id, "manifest_sha256": manifest_digest, "entries": manifest["entries"]}) searches = [{ "schema_version": VERSION, "run_id": run_id, "probe_id": entry["probe_id"], "hop_id": entry["hop_id"], "forms_searched": ["raw", "url", "base64", "hex"], "raw_hits": 0, "encoded_hits": 0, "marker_hits": len(entry["data_classes"]), "verdict": "PASS", "marker_classes": entry["data_classes"], } for entry in manifest["entries"]] write_ndjson(out / "sink-search.ndjson", searches) identities = sorted(verifier.expected_identities(), key=lambda row: tuple("" if value is None else value for value in row)) if scenario == "fabricated-minimal": identities = identities[:102] cases, dbrows = [], [] observed = outcome(); observed_digest = sha_bytes(canonical(observed)) probe_ids = [entry["probe_id"] for entry in manifest["entries"]] for index, (route, phase, category, seam) in enumerate(identities, 1): case_id = f"case-{index:03d}"; assertion_id = f"db-{index:03d}" context = {"owner_id": f"owner-{index:03d}", "session_id": f"session-{index:03d}", "family_id": f"family-{index:03d}", "generation_before": 0, "generation_after": 0} cases.append({ "schema_version": VERSION, "run_id": run_id, "case_id": case_id, "route": route, "phase": phase, "category": category, "request_fixture_id": f"request-{index:03d}", "oracle_id": f"oracle-{index:03d}", "observed": observed, "observed_side_effects": {"family_generation": 0}, "db_assertions": [assertion_id], "sink_probes": probe_ids, "crash_seam": seam, "lock_transport": "web-locks" if phase == "browser" else "not-applicable", "started_at": STAMP, "ended_at": STAMP, "verdict": "PASS", "canonical_outcome_sha256": observed_digest, "trace_ref": "traces/fixture.trace", "trace_sha256": sha_bytes(trace), "evidence_refs": [{"path": "evidence/fixture.json", "sha256": sha_bytes(evidence)}], "db_context": context, }) dbrows.append({"schema_version": VERSION, "run_id": run_id, "case_id": case_id, "assertion_id": assertion_id, "before_digest": sha_bytes(f"before-{index}".encode()), "after_digest": sha_bytes(f"after-{index}".encode()), "observed_delta": {"family_generation": 0}, "verdict": "PASS", "action": category, **context}) write_ndjson(out / "cases.ndjson", cases) write_ndjson(out / "db-assertions.ndjson", dbrows) case_by_seam = {row["crash_seam"]: row["case_id"] for row in cases if row["phase"] == "crash"} crashrows = [{"schema_version": VERSION, "run_id": run_id, "case_id": case_by_seam[seam], "seam": seam, "seam_revision": "fixture-v1", "fire_count": 1, "resolution": "exact-committed-outcome", "blind_retry": False, "verdict": "PASS"} for seam in verifier.SEAMS if seam in case_by_seam] if not crashrows: # Keep the fabricated attack schema-valid and non-empty. seam = verifier.SEAMS[0] crashrows = [{"schema_version": VERSION, "run_id": run_id, "case_id": cases[0]["case_id"], "seam": seam, "seam_revision": "fixture-v1", "fire_count": 1, "resolution": "exact-committed-outcome", "blind_retry": False, "verdict": "PASS"}] write_ndjson(out / "crash-injection.ndjson", crashrows) record_count = len(cases) + len(searches) + len(dbrows) + len(crashrows) payload_names = sorted(p.relative_to(out).as_posix() for p in out.rglob("*") if p.is_file() and p.name not in {"summary.json", "artifact-manifest.json", "SHA256SUMS"}) files = [{"path": name, "sha256": sha_bytes((out / name).read_bytes()), "bytes": (out / name).stat().st_size} for name in payload_names] case_ids = [row["case_id"] for row in cases] summary_files = [{**item, "producer_version": "auth-split-fixture-v1", "case_ids": case_ids} for item in files] write_json(out / "artifact-manifest.json", {"schema_version": VERSION, "run_id": run_id, "generated_at": STAMP, "files": files}) write_json(out / "summary.json", {"schema_version": VERSION, "run_id": run_id, "environment_valid": True, "production_equivalent": True, "counts": {"pass": record_count, "fail": 0, "blocked": 0}, "verdict": "PASS", "artifacts": summary_files}) retained = sorted(p.relative_to(out).as_posix() for p in out.rglob("*") if p.is_file() and p.name != "SHA256SUMS") (out / "SHA256SUMS").write_text("".join(f"{sha_bytes((out / name).read_bytes())} {name}\n" for name in retained), encoding="utf-8") shutil.copytree(out, publication, dirs_exist_ok=True) print(f"FIXTURE=GENERATED scenario={scenario} cases={len(cases)} sinks={len(searches)} input={out} publication={publication}") def main(): parser = argparse.ArgumentParser() parser.add_argument("--out", required=True, type=pathlib.Path) parser.add_argument("--scenario", required=True, choices=("valid", "fabricated-minimal", "disconnected-graph")) args = parser.parse_args() generate(args.out.resolve(), args.scenario) if __name__ == "__main__": main()