#!/usr/bin/env python3 """Generated standalone Rung CLI. Do not edit; run scripts/build_single_file.py.""" # source-sha256: cf30d047523c5375078a342fbfcc43f0f203b66d027192503c1698d38306a94f import builtins import importlib.abc import importlib.util from types import MappingProxyType _SOURCES = MappingProxyType({"rung":"\"\"\"Rung \u2014 AI Agent Governance Audit Engine.\n\nModular audit engine for scoring repository readiness for AI coding\nagent governance. Produces a deterministic AuditResult v1 from which\nboth free preview and paid PDF projections render.\n\nLicense: MIT\n\"\"\"\n\n__version__ = \"0.3.0\"\n\nfrom rung.audit import run_audit, AuditResult\nfrom rung.scoring import compute_score, GRADE_LABELS\nfrom rung.models import CheckResult, EvidenceState, Confidence, SourceClass, AuthorityLevel\n\n__all__ = [\n \"run_audit\",\n \"AuditResult\",\n \"compute_score\",\n \"GRADE_LABELS\",\n \"CheckResult\",\n \"EvidenceState\",\n \"Confidence\",\n \"SourceClass\",\n \"AuthorityLevel\",\n]\n","rung.__main__":"\"\"\"Allow running as `python3 -m rung`.\"\"\"\nfrom rung.cli import main\nimport sys\n\nif __name__ == \"__main__\":\n sys.exit(main())","rung.audit":"\"\"\"Audit orchestrator \u2014 runs all checks and produces AuditResult v1.\n\nThe free preview and paid PDF both render from the same AuditResult.\nIncludes a canonical report-data digest computed over all fields\nexcept the digest itself.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport hashlib\nimport json\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nfrom typing import Optional\n\nfrom rung.models import CheckResult, AuditResult, AuthorityLevel\nfrom rung.scoring import compute_score, recommend_authority, GRADE_LABELS\nfrom rung.checks import ALL_CHECKS\nfrom rung import __version__ as RUNG_VERSION\n\nSCHEMA_VERSION = \"1.0.0\"\n\n\ndef _compute_digest(result_dict: dict) -> str:\n \"\"\"Compute SHA-256 over canonical JSON of the result, excluding the digest field.\"\"\"\n data = {k: v for k, v in result_dict.items() if k != \"report_data_sha256\"}\n canonical = json.dumps(data, sort_keys=True, ensure_ascii=False, separators=(\",\", \":\"))\n return hashlib.sha256(canonical.encode(\"utf-8\")).hexdigest()\n\n\ndef run_audit(\n root: Path,\n commit_sha: Optional[str] = None,\n repository: Optional[str] = None,\n timestamp: Optional[str] = None,\n) -> AuditResult:\n \"\"\"Run all checks against a repository root and return AuditResult v1.\"\"\"\n root = Path(root).resolve()\n checks = [check(root) for check in ALL_CHECKS]\n score, grade, gate = compute_score(checks)\n authority = recommend_authority(checks)\n timestamp = timestamp or datetime.now(timezone.utc).strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n\n result = AuditResult(\n repository=repository or str(root),\n commit_sha=commit_sha,\n checks=checks,\n score=score,\n grade=grade,\n grade_label=GRADE_LABELS.get(grade, \"\"),\n quality_gate=\"PASS\" if gate else \"FAIL\",\n authority=authority,\n rung_version=RUNG_VERSION,\n schema_version=SCHEMA_VERSION,\n timestamp=timestamp,\n )\n\n result_dict = result_to_dict(result)\n result.report_data_sha256 = _compute_digest(result_dict)\n return result\n\n\ndef result_to_dict(result: AuditResult) -> dict:\n \"\"\"Convert AuditResult to a JSON-serializable dict.\"\"\"\n return {\n \"repository\": result.repository,\n \"commit_sha\": result.commit_sha,\n \"score\": result.score,\n \"grade\": result.grade,\n \"grade_label\": result.grade_label,\n \"quality_gate\": result.quality_gate,\n \"authority\": result.authority.value if isinstance(result.authority, AuthorityLevel) else str(result.authority),\n \"rung_version\": result.rung_version,\n \"schema_version\": result.schema_version,\n \"timestamp\": result.timestamp,\n \"report_data_sha256\": result.report_data_sha256,\n \"checks\": [check_to_dict(c) for c in result.checks],\n }\n\n\ndef check_to_dict(c: CheckResult) -> dict:\n \"\"\"Convert CheckResult to a JSON-serializable dict.\"\"\"\n return {\n \"name\": c.name,\n \"description\": c.description,\n \"weight\": c.weight,\n \"blocking\": c.blocking,\n \"state\": c.state.value if hasattr(c.state, \"value\") else str(c.state),\n \"confidence\": c.confidence.value if hasattr(c.confidence, \"value\") else str(c.confidence),\n \"blocking_for\": c.blocking_for,\n \"evidence\": c.evidence,\n \"limitations\": c.limitations,\n \"remediation\": c.remediation,\n \"source_mappings\": c.source_mappings,\n \"passed\": c.passed,\n }\n","rung.checks":"\"\"\"Rung governance checks.\n\nEach check returns a CheckResult with evidence states instead of\nboolean pass/fail. Checks are registered in ALL_CHECKS.\n\"\"\"\n\nfrom rung.checks.agent_policy import check_agent_policy\nfrom rung.checks.build_commands import check_build_commands\nfrom rung.checks.verification_gate import check_verification_gate\nfrom rung.checks.source_registry import check_source_registry\nfrom rung.checks.evidence_traceability import check_evidence_traceability\nfrom rung.checks.session_ledger import check_session_ledger\nfrom rung.checks.file_size import check_file_size\nfrom rung.checks.agent_attribution import check_agent_attribution\nfrom rung.checks.security_never_rules import check_security_never_rules\nfrom rung.checks.independent_review import check_independent_review\nfrom rung.checks.cyclic_verification import check_cyclic_verification\n\nALL_CHECKS = [\n check_agent_policy,\n check_build_commands,\n check_verification_gate,\n check_source_registry,\n check_evidence_traceability,\n check_session_ledger,\n check_file_size,\n check_agent_attribution,\n check_security_never_rules,\n check_independent_review,\n check_cyclic_verification,\n]","rung.checks.agent_attribution":"\"\"\"Check 8: Agent attribution on commits/PRs.\"\"\"\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import read_text, has_affirmative_pattern\n\n\ndef check_agent_attribution(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Agent attribution\",\n description=\"Convention for attributing AI-assisted commits and PRs (Generated-by / Co-authored-by)\",\n weight=5, blocking=False,\n state=EvidenceState.ABSENT,\n confidence=Confidence.MEDIUM,\n source_mappings=[\n {\"id\": \"apache_airflow\", \"classification\": SOURCES[\"apache_airflow\"][\"classification\"].value},\n ],\n )\n patterns = [r'Generated-by:', r'Co-authored-by:', r'Drafted-by:', r'AI-assisted', r'agent-assisted']\n files_to_check = [root / \"AGENTS.md\", root / \".github\" / \"pull_request_template.md\", root / \"CONTRIBUTING.md\"]\n found = False\n for f in files_to_check:\n if f.exists():\n content = read_text(f)\n if content is None:\n continue\n if has_affirmative_pattern(content, patterns):\n found = True\n r.evidence.append(f\"Attribution convention in {f.relative_to(root)}\")\n if found:\n r.state = EvidenceState.DETECTED\n else:\n r.remediation = [\n \"Add an attribution convention to AGENTS.md or PR template:\",\n \" Generated-by: \",\n ]\n return r\n","rung.checks.agent_policy":"\"\"\"Check 1: Agent policy file exists.\"\"\"\nimport re\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES, IGNORED_PARTS\nfrom rung.evidence import find_file, read_text\n\n\ndef check_agent_policy(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Agent policy file\",\n description=\"AGENTS.md or .github/copilot-instructions.md exists at repo root\",\n weight=15, blocking=True,\n state=EvidenceState.ABSENT,\n confidence=Confidence.HIGH,\n blocking_for=[\"local\", \"pr\", \"merge\", \"release\"],\n source_mappings=[\n {\"id\": \"agents_md\", \"classification\": SOURCES[\"agents_md\"][\"classification\"].value},\n {\"id\": \"github_copilot\", \"classification\": SOURCES[\"github_copilot\"][\"classification\"].value},\n ],\n )\n candidates = [\n root / \"AGENTS.md\",\n root / \".github\" / \"copilot-instructions.md\",\n root / \"CLAUDE.md\",\n root / \"GEMINI.md\",\n ]\n found = find_file(root, candidates)\n found = [p for p in found if all(\n re.search(pattern, read_text(p) or \"\", re.IGNORECASE | re.MULTILINE)\n for pattern in (r\"^#{1,6}\\s+.*(?:build|test)\", r\"^#{1,6}\\s+.*security\")\n )]\n nested = list(root.rglob(\"AGENTS.md\"))\n nested = [p for p in nested if p != root / \"AGENTS.md\" and all(part not in IGNORED_PARTS for part in p.relative_to(root).parts[:-1]) and p.is_file()]\n\n if found:\n r.state = EvidenceState.DETECTED\n r.evidence.append(f\"Found: {found[0].relative_to(root)}\")\n if len(found) > 1:\n r.evidence.append(f\"Also found: {', '.join(str(p.relative_to(root)) for p in found[1:])}\")\n if nested:\n r.evidence.append(f\"Nested AGENTS.md files: {len(nested)} (good for monorepos)\")\n else:\n r.remediation = [\n \"Create an AGENTS.md at the repo root with these sections:\",\n \" 1. Project overview (what the project does)\",\n \" 2. Build/test commands (exact commands agents should run)\",\n \" 3. Code style conventions\",\n \" 4. Testing instructions\",\n \" 5. Security considerations (what agents must never do)\",\n f\"See exemplars: {SOURCES['openai_codex']['url']}\",\n ]\n return r\n","rung.checks.build_commands":"\"\"\"Check 2: Build/test commands declared and runnable.\"\"\"\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import read_text, detect_build_commands\n\n\ndef check_build_commands(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Build & test commands\",\n description=\"AGENTS.md declares concrete build/test commands that exist and run\",\n weight=15, blocking=True,\n state=EvidenceState.ABSENT,\n confidence=Confidence.HIGH,\n blocking_for=[\"local\", \"pr\", \"merge\"],\n source_mappings=[\n {\"id\": \"openai_codex\", \"classification\": SOURCES[\"openai_codex\"][\"classification\"].value},\n {\"id\": \"github_copilot\", \"classification\": SOURCES[\"github_copilot\"][\"classification\"].value},\n ],\n )\n agents_md = root / \"AGENTS.md\"\n if not agents_md.exists():\n r.remediation = [\"First create an AGENTS.md (see Check 1), then declare build/test commands in it.\"]\n return r\n\n content = read_text(agents_md) or \"\"\n found_cmds = detect_build_commands(content)\n found_cmds = [command for command in found_cmds if not (\n command.startswith((\"python \", \"python3 \"))\n and command.split()[-1].endswith(\".py\")\n and not (root / command.split()[-1]).is_file()\n )]\n\n makefile = root / \"Makefile\"\n justfile = root / \"justfile\"\n package_json = root / \"package.json\"\n\n has_runner = makefile.exists() or justfile.exists() or package_json.exists()\n if found_cmds:\n r.state = EvidenceState.DETECTED\n r.evidence.append(f\"Declared commands: {', '.join(found_cmds[:5])}\")\n if has_runner:\n r.evidence.append(\"Build runner found (Makefile/justfile/package.json)\")\n elif has_runner:\n r.state = EvidenceState.CLAIMED\n r.evidence.append(\"Build runner found but not referenced in AGENTS.md\")\n r.limitations.append(\"AGENTS.md should explicitly name the build/test commands agents must use\")\n r.remediation = [\n \"Add a 'Build & Test' section to AGENTS.md with the exact commands, e.g.:\",\n \" npm test # run all tests\",\n \" npm run build # build the project\",\n ]\n else:\n r.remediation = [\n \"Add a Makefile or justfile with 'test' and 'build' targets,\",\n \"then reference them in AGENTS.md.\",\n ]\n return r\n","rung.checks.cyclic_verification":"\"\"\"Check 11: Cyclic verification loop (plan->build->verify->fix->verify).\"\"\"\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import read_text, has_affirmative_pattern\n\n\ndef check_cyclic_verification(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Cyclic verification\",\n description=\"Workflow loops: plan -> build -> verify -> fix -> verify until pass\",\n weight=5, blocking=False,\n state=EvidenceState.ABSENT,\n confidence=Confidence.MEDIUM,\n source_mappings=[\n {\"id\": \"anthropic_multiagent\", \"classification\": SOURCES[\"anthropic_multiagent\"][\"classification\"].value},\n ],\n )\n patterns = [\n r'(?:trycycle|try.cycle|verification.loop|fix.loop)',\n r'(?:plan|build|verify|review|fix).*loop',\n r'rerun.*verification',\n r're-?verify',\n r'(?:if.*fail|on.*fail).*fix.*(?:verify|test)',\n ]\n files_to_check = [root / \"AGENTS.md\", root / \"CONTRIBUTING.md\"]\n found = False\n for f in files_to_check:\n if f.exists():\n content = read_text(f)\n if content is None:\n continue\n if has_affirmative_pattern(content, patterns):\n found = True\n r.evidence.append(f\"Cyclic verification in {f.relative_to(root)}\")\n if found:\n r.state = EvidenceState.DETECTED\n else:\n r.remediation = [\n \"Document a cyclic verification loop in AGENTS.md:\",\n \" plan -> build -> verify -> (if fail) fix -> verify again -> review\",\n ]\n return r\n","rung.checks.evidence_traceability":"\"\"\"Check 5: Evidence and traceability for completed work.\n\nDoes not give credit merely because .github/workflows exists. CI\nworkflow existence alone is detected, not enforced or verified.\n\"\"\"\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import find_file, has_ci_workflows, ci_workflow_runs_tests\n\n\ndef check_evidence_traceability(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Evidence & traceability\",\n description=\"Evidence index, traceability matrix, or CI artifacts linking work to verified outcomes\",\n weight=10, blocking=False,\n state=EvidenceState.ABSENT,\n confidence=Confidence.MEDIUM,\n source_mappings=[\n {\"id\": \"ibm_adlc\", \"classification\": SOURCES[\"ibm_adlc\"][\"classification\"].value},\n {\"id\": \"slsa\", \"classification\": SOURCES[\"slsa\"][\"classification\"].value},\n ],\n )\n candidates = [\n root / \"factory\" / \"evidence\" / \"index.json\",\n root / \"docs\" / \"traceability-matrix.md\",\n root / \"templates\" / \"traceability-matrix.md\",\n root / \"evidence\",\n root / \".evidence\",\n root / \"test-results\",\n root / \".test-results\",\n root / \"CHANGELOG.md\",\n root / \"docs\" / \"CHANGELOG.md\",\n ]\n found = find_file(root, candidates)\n found = [p for p in found if p.is_dir() or (\n len((text := p.read_text(encoding=\"utf-8\", errors=\"ignore\")).splitlines()) >= 3\n and (\"changelog\" in text.lower() or \"trace\" in text.lower() or \"evidence\" in text.lower())\n )]\n ci_exists = has_ci_workflows(root)\n ci_runs_tests = ci_workflow_runs_tests(root)\n\n if ci_exists:\n if ci_runs_tests:\n r.evidence.append(\"CI workflows found that run tests\")\n else:\n r.evidence.append(\"CI workflow files found but do not appear to run tests\")\n r.limitations.append(\"Workflow existence alone does not provide traceability; the workflow must produce test evidence\")\n\n if found:\n r.state = EvidenceState.DETECTED\n if not any(\"CI workflows\" in d for d in r.evidence):\n r.evidence.append(f\"Evidence system: {found[0].relative_to(root)}\")\n else:\n r.remediation = [\n \"Create an evidence trail linking completed work to verification.\",\n \"Options: a CHANGELOG.md, a factory/evidence/ index, CI workflow\",\n \"artifacts, or a traceability matrix.\",\n ]\n return r\n","rung.checks.file_size":"\"\"\"Check 7: File-size discipline (non-scoring maintainability appendix).\n\nThis check has weight=0 and does NOT contribute to the governance score.\nIt appears in the report as a maintainability appendix. The 500/800 LoC\nthresholds come from openai/codex's AGENTS.md engineering convention, not\nan autonomous-agent governance standard.\n\"\"\"\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES, SOURCE_EXTENSIONS, IGNORED_PARTS, WATCH_LOC, SMELL_LOC, DEFECT_LOC\nfrom rung.evidence import count_source_loc, is_test_file\n\n\ndef check_file_size(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"File-size discipline\",\n description=f\"Source files within size thresholds (warn>{WATCH_LOC} LoC, fail>{SMELL_LOC} LoC) \u2014 non-scoring maintainability appendix\",\n weight=0, blocking=False,\n state=EvidenceState.VERIFIED,\n confidence=Confidence.HIGH,\n source_mappings=[\n {\"id\": \"openai_codex\", \"classification\": SOURCES[\"openai_codex\"][\"classification\"].value},\n ],\n limitations=[\n \"File-size thresholds come from openai/codex's AGENTS.md engineering convention, not an autonomous-agent governance standard.\",\n \"This check is non-scoring and appears only in the maintainability appendix.\",\n ],\n )\n large_files = []\n for path in root.rglob(\"*\"):\n if not path.is_file():\n continue\n if any(part in IGNORED_PARTS for part in path.parts):\n continue\n if path.suffix not in SOURCE_EXTENSIONS:\n continue\n if is_test_file(path):\n continue\n loc = count_source_loc(path)\n if loc >= DEFECT_LOC:\n large_files.append((\"DEFECT\", path.relative_to(root), loc))\n elif loc >= SMELL_LOC:\n large_files.append((\"SMELL\", path.relative_to(root), loc))\n elif loc >= WATCH_LOC:\n large_files.append((\"WATCH\", path.relative_to(root), loc))\n\n if not large_files:\n r.evidence.append(\"All source files within thresholds\")\n else:\n r.state = EvidenceState.ABSENT\n for level, rel, loc in large_files[:10]:\n r.evidence.append(f\"[{level}] {rel} ({loc} LoC)\")\n r.remediation = [\n f\"Target: modules under {WATCH_LOC} LoC (excluding tests)\",\n f\"Hard cap: files over {SMELL_LOC} LoC must add new functionality\",\n \"in a new module unless there is a documented reason not to.\",\n \"Consider splitting large files or adding an exemption comment.\",\n ]\n return r","rung.checks.independent_review":"\"\"\"Check 10: Independent review requirement.\"\"\"\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import read_text, has_affirmative_pattern\n\n\ndef check_independent_review(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Independent review\",\n description=\"Documented requirement for independent review before commit\",\n weight=5, blocking=False,\n state=EvidenceState.ABSENT,\n confidence=Confidence.MEDIUM,\n source_mappings=[\n {\"id\": \"anthropic_multiagent\", \"classification\": SOURCES[\"anthropic_multiagent\"][\"classification\"].value},\n {\"id\": \"nist_rmf\", \"classification\": SOURCES[\"nist_rmf\"][\"classification\"].value},\n ],\n )\n patterns = [\n r'(?:independent|rubberduck|peer|code)\\s+review',\n r'review\\s+(?:before|prior\\s+to)\\s+(?:commit|merge|push)',\n r'self-review.*review',\n ]\n files_to_check = [root / \"AGENTS.md\", root / \"CONTRIBUTING.md\", root / \".github\" / \"pull_request_template.md\", root / \"REVIEW.md\"]\n found = False\n for f in files_to_check:\n if f.exists():\n content = read_text(f)\n if content is None:\n continue\n if has_affirmative_pattern(content, patterns):\n found = True\n r.evidence.append(f\"Review requirement in {f.relative_to(root)}\")\n if found:\n r.state = EvidenceState.DETECTED\n else:\n r.remediation = [\n \"Add an independent review requirement to AGENTS.md:\",\n \" 'Before committing: complete self-review, then independent\",\n \" (rubberduck) review, then rerun verification, then commit.'\",\n ]\n return r\n","rung.checks.security_never_rules":"\"\"\"Check 9: Security 'Never' rules in agent policy.\"\"\"\nimport re\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import read_text\n\n\ndef check_security_never_rules(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Security Never-rules\",\n description=\"AGENTS.md or SECURITY.md enumerates hard 'Never' rules for agents\",\n weight=10, blocking=True,\n state=EvidenceState.ABSENT,\n confidence=Confidence.HIGH,\n blocking_for=[\"local\", \"pr\", \"merge\", \"release\"],\n source_mappings=[\n {\"id\": \"nist_rmf\", \"classification\": SOURCES[\"nist_rmf\"][\"classification\"].value},\n {\"id\": \"apache_airflow\", \"classification\": SOURCES[\"apache_airflow\"][\"classification\"].value},\n ],\n )\n files_to_check = [root / \"AGENTS.md\", root / \"SECURITY.md\", root / \".github\" / \"SECURITY.md\"]\n never_rules = set()\n for f in files_to_check:\n if f.exists():\n content = read_text(f)\n if content is None:\n continue\n never_rules.update(\n re.sub(r'\\s+', ' ', line.strip().lower()) for line in content.splitlines()\n if re.match(r'^\\s*(?:(?:[-*]|\\d+[.)])\\s+)?Never\\b', line, re.IGNORECASE)\n and re.search(r'\\b(?:secret|credential|key|socket|git|security|policy|verification|commit|merge|release|destructive|transmit|execute|threshold|external)\\b', line, re.IGNORECASE)\n )\n never_count = len(never_rules)\n if never_count >= 5:\n r.state = EvidenceState.DETECTED\n r.evidence.append(f\"Found {never_count} 'Never' rules across policy files\")\n elif never_count >= 1:\n r.state = EvidenceState.CLAIMED\n r.evidence.append(f\"Found {never_count} 'Never' rule(s) \u2014 recommend at least 5\")\n r.limitations.append(\"Consider adding more explicit 'Never' rules for security boundaries\")\n r.remediation = [\n \"Add more 'Never' rules to AGENTS.md or SECURITY.md, e.g.:\",\n \" Never commit secrets, API keys, or credentials\",\n \" Never expose the Docker socket to construction agents\",\n \" Never use destructive git operations without explicit request\",\n ]\n else:\n r.remediation = [\n \"Add a 'Never' rules section to AGENTS.md or SECURITY.md.\",\n ]\n return r\n","rung.checks.session_ledger":"\"\"\"Check 6: Session ledger / status tracking.\"\"\"\nimport json\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import find_file, has_valid_json\n\n\ndef check_session_ledger(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Session ledger\",\n description=\"Machine-readable status file, work queue, or changelog for tracking agent sessions\",\n weight=5, blocking=False,\n state=EvidenceState.ABSENT,\n confidence=Confidence.MEDIUM,\n source_mappings=[\n {\"id\": \"anthropic_multiagent\", \"classification\": SOURCES[\"anthropic_multiagent\"][\"classification\"].value},\n {\"id\": \"ibm_adlc\", \"classification\": SOURCES[\"ibm_adlc\"][\"classification\"].value},\n ],\n )\n candidates = [\n root / \"factory\" / \"status.json\",\n root / \"factory\" / \"queue.json\",\n root / \"docs\" / \"sessions\" / \"current.md\",\n root / \"sessions\" / \"current.md\",\n root / \"SESSIONS.md\",\n root / \".factory\" / \"sessions\" / \"current.md\",\n root / \"STATUS.md\",\n root / \"docs\" / \"STATUS.md\",\n root / \"CHANGELOG.md\",\n ]\n found = find_file(root, candidates)\n found = [p for p in found if p.suffix != \".json\" or (\n has_valid_json(p)\n and any(key in json.loads(p.read_text(encoding=\"utf-8\")) for key in (\"state\", \"status\", \"active_increment\", \"items\"))\n )]\n if found:\n r.state = EvidenceState.DETECTED\n r.evidence.append(f\"Session/status tracking: {found[0].relative_to(root)}\")\n else:\n r.remediation = [\n \"Create a session ledger or status file (e.g., docs/sessions/current.md,\",\n \"factory/status.json, or CHANGELOG.md) that records the active work\",\n \"item, last update timestamp, and any blockers.\",\n ]\n return r\n","rung.checks.source_registry":"\"\"\"Check 4: Source-of-truth registry for external claims.\"\"\"\nimport json\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import find_file, has_valid_json\n\n\ndef check_source_registry(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Source-of-truth registry\",\n description=\"Machine-readable source registry for external claims and references\",\n weight=10, blocking=False,\n state=EvidenceState.ABSENT,\n confidence=Confidence.MEDIUM,\n source_mappings=[\n {\"id\": \"nist_rmf\", \"classification\": SOURCES[\"nist_rmf\"][\"classification\"].value},\n {\"id\": \"ibm_adlc\", \"classification\": SOURCES[\"ibm_adlc\"][\"classification\"].value},\n ],\n )\n candidates = [\n root / \"docs\" / \"research\" / \"sources.json\",\n root / \"docs\" / \"sources.json\",\n root / \"SOURCES.md\",\n root / \"docs\" / \"SOURCES.md\",\n ]\n found = find_file(root, candidates)\n found = [p for p in found if p.suffix != \".json\" or (\n has_valid_json(p)\n and isinstance((data := json.loads(p.read_text(encoding=\"utf-8\"))).get(\"sources\"), list)\n and bool(data[\"sources\"])\n )]\n if found:\n r.state = EvidenceState.DETECTED\n r.evidence.append(f\"Source registry: {found[0].relative_to(root)}\")\n else:\n r.remediation = [\n \"Create a source registry (e.g., docs/research/sources.json) that\",\n \"records every external source cited in the repo with a SRC-ID,\",\n \"URL, and classification (first_party / preprint / standard).\",\n ]\n return r\n","rung.checks.verification_gate":"\"\"\"Check 3: Verification gate before commit/merge.\n\nDoes not give credit merely for finding any .github/workflows file.\nWorkflow existence is detected at most; enforcement requires semantic\ninspection or is unobservable from public evidence.\n\"\"\"\nfrom pathlib import Path\nfrom rung.models import CheckResult, EvidenceState, Confidence\nfrom rung.sources import SOURCES\nfrom rung.evidence import read_text, has_affirmative_pattern, has_ci_workflows, ci_workflow_runs_tests, ci_workflow_enforces_gate\n\n\ndef check_verification_gate(root: Path) -> CheckResult:\n r = CheckResult(\n name=\"Verification gate\",\n description=\"Documented rule that verification must pass before commit or merge\",\n weight=10, blocking=True,\n state=EvidenceState.ABSENT,\n confidence=Confidence.MEDIUM,\n blocking_for=[\"merge\", \"release\"],\n source_mappings=[\n {\"id\": \"nist_rmf\", \"classification\": SOURCES[\"nist_rmf\"][\"classification\"].value},\n {\"id\": \"anthropic_multiagent\", \"classification\": SOURCES[\"anthropic_multiagent\"][\"classification\"].value},\n ],\n )\n patterns = [\n r'(?:before|prior\\s+to)\\s+(?:commit|push|merge|marking.*complete)',\n r'verification?\\s+must\\s+pass',\n r'(?:never|do\\s+not)\\s+(?:commit|push|merge)\\s+(?:without|until|before)',\n r'(?:githook|pre-commit|husky)',\n ]\n files_to_check = [root / \"AGENTS.md\", root / \".github\" / \"pull_request_template.md\", root / \"CONTRIBUTING.md\"]\n documented = False\n for f in files_to_check:\n if f.exists():\n content = read_text(f)\n if content is None:\n continue\n if has_affirmative_pattern(content, patterns):\n documented = True\n r.evidence.append(f\"Verification gate mentioned in {f.relative_to(root)}\")\n\n githooks = root / \".githooks\"\n husky = root / \".husky\"\n has_hooks = any(path.is_dir() and any(child.is_file() for child in path.iterdir()) for path in (githooks, husky))\n if has_hooks:\n documented = True\n r.evidence.append(\"Git hooks directory found (.githooks or .husky)\")\n\n ci_exists = has_ci_workflows(root)\n ci_runs_tests = ci_workflow_runs_tests(root)\n ci_enforces = ci_workflow_enforces_gate(root)\n\n if ci_exists:\n if ci_enforces:\n r.evidence.append(\"CI workflow references enforcement settings\")\n elif ci_runs_tests:\n r.evidence.append(\"CI workflow detected running tests (not necessarily enforced)\")\n else:\n r.evidence.append(\"CI workflow file found but does not appear to run tests\")\n r.limitations.append(\"Workflow existence alone is not enforcement; the workflow must run tests and be required\")\n\n if documented and ci_runs_tests:\n r.state = EvidenceState.UNOBSERVABLE\n r.limitations.append(\"Actual enforcement (required status checks, branch protection) is not observable from public repository contents without Administration read permission\")\n elif documented or has_hooks:\n r.state = EvidenceState.CLAIMED\n r.limitations.append(\"Verification gate is documented but not confirmed as enforced by CI\")\n elif ci_exists and not ci_runs_tests:\n r.state = EvidenceState.CLAIMED\n r.limitations.append(\"CI exists but does not run tests; this is not a verification gate\")\n else:\n r.remediation = [\n \"Add an explicit rule to AGENTS.md or CONTRIBUTING.md:\",\n \" 'Never commit or merge before verification (npm test) passes.'\",\n \"Set up a pre-commit hook or CI gate that runs the test suite.\",\n ]\n return r\n","rung.cli":"#!/usr/bin/env python3\n\"\"\"Rung CLI \u2014 AI Agent Governance Audit\n\nScores a repository's readiness for AI coding agent governance using 11\nweighted checks derived from published standards and industry best practices.\n\nOutputs a numeric score (0-100), a letter grade (A-E), a quality gate\nverdict (PASS/FAIL), evidence states, and actionable next steps for each gap.\n\nUsage:\n python3 -m rung.cli [--root /path/to/your/repo] [--json]\n\nExit codes:\n 0 \u2014 quality gate passed (all blocking checks passed)\n 1 \u2014 quality gate failed (one or more blocking checks failed)\n\nLicense: MIT\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport json\nimport re\nimport sys\nfrom pathlib import Path\n\nfrom rung.audit import run_audit, result_to_dict\nfrom rung.scoring import GRADE_LABELS\n\n\ndef format_report(result) -> str:\n \"\"\"Format a human-readable text report from AuditResult.\"\"\"\n lines = []\n lines.append(\"=\" * 60)\n lines.append(\" Rung \u2014 AI Agent Governance Audit\")\n lines.append(\"=\" * 60)\n lines.append(f\" Repository: {result.repository}\")\n if result.commit_sha:\n lines.append(f\" Commit: {result.commit_sha}\")\n lines.append(f\" Score: {result.score}/100 Grade: {result.grade} ({result.grade_label})\")\n lines.append(f\" Quality Gate: {result.quality_gate}\")\n lines.append(f\" Authority: {result.authority.value if hasattr(result.authority, 'value') else result.authority}\")\n lines.append(f\" Rung version: {result.rung_version} Schema: {result.schema_version}\")\n lines.append(f\" Timestamp: {result.timestamp}\")\n lines.append(f\" Report digest: {result.report_data_sha256}\")\n lines.append(\"=\" * 60)\n lines.append(\"\")\n\n for c in result.checks:\n status = c.state.value if hasattr(c.state, \"value\") else str(c.state)\n blocking_tag = \" [blocking]\" if c.blocking else \"\"\n weight_tag = f\" ({c.weight} pts)\" if c.weight > 0 else \" (non-scoring)\"\n lines.append(f\" [{status.upper()}] {c.name}{weight_tag}{blocking_tag}\")\n lines.append(f\" {c.description}\")\n for e in c.evidence:\n lines.append(f\" -> {e}\")\n for lim in c.limitations:\n lines.append(f\" ! {lim}\")\n if c.source_mappings:\n src_ids = [sm[\"id\"] for sm in c.source_mappings]\n lines.append(f\" Sources: {', '.join(src_ids)}\")\n if not c.passed and c.remediation:\n lines.append(\" Remediation:\")\n for step in c.remediation:\n lines.append(f\" {step}\")\n lines.append(\"\")\n\n lines.append(\"-\" * 60)\n lines.append(f\" Score: {result.score}/100 Grade: {result.grade} ({result.grade_label})\")\n lines.append(f\" Quality Gate: {result.quality_gate}\")\n lines.append(f\" Authority: {result.authority.value if hasattr(result.authority, 'value') else result.authority}\")\n failed_blocking = [c for c in result.checks if c.blocking and not c.passed]\n if failed_blocking:\n lines.append(f\" Blocking failures: {len(failed_blocking)}\")\n for c in failed_blocking:\n lines.append(f\" - {c.name} [{c.state.value if hasattr(c.state, 'value') else c.state}]\")\n lines.append(\"\")\n return \"\\n\".join(lines)\n\n\ndef main() -> int:\n if len(sys.argv) > 1 and sys.argv[1] in {\"verify\", \"replay\"}:\n command = sys.argv[1]\n parser = argparse.ArgumentParser(prog=f\"rung {command}\")\n parser.add_argument(\"--root\", required=True, help=\"Clean Git checkout root\")\n parser.add_argument(\"--receipt\", required=True, help=\"Verification receipt path\")\n if command == \"replay\":\n parser.add_argument(\"--observation\", required=True, help=\"Replay observation path\")\n args = parser.parse_args(sys.argv[2:])\n from rung.verification import verification_main\n return verification_main(\n command, args.root, args.receipt, getattr(args, \"observation\", None)\n )\n\n parser = argparse.ArgumentParser(description=\"Rung CLI \u2014 AI Agent Governance Audit\")\n parser.add_argument(\"--root\", default=\".\", help=\"Repository root to audit\")\n parser.add_argument(\"--commit-sha\", type=lambda value: value if re.fullmatch(r\"[0-9a-f]{40}\", value) else parser.error(\"--commit-sha must be 40 lowercase hex characters\"), help=\"Exact audited commit SHA\")\n parser.add_argument(\"--repository\", help=\"Stable repository identifier for the result digest\")\n parser.add_argument(\"--timestamp\", help=\"Explicit RFC 3339 report timestamp for deterministic rendering\")\n parser.add_argument(\"--json\", action=\"store_true\", help=\"Output JSON instead of text\")\n parser.add_argument(\"-o\", \"--output\", help=\"Write output to file instead of stdout\")\n args = parser.parse_args()\n\n root = Path(args.root).resolve()\n result = run_audit(root, commit_sha=args.commit_sha, repository=args.repository, timestamp=args.timestamp)\n\n if args.json:\n output = json.dumps(result_to_dict(result), indent=2)\n else:\n output = format_report(result)\n\n if args.output:\n Path(args.output).write_text(output, encoding=\"utf-8\")\n else:\n print(output)\n\n return 0 if result.quality_gate == \"PASS\" else 1\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n","rung.evidence":"\"\"\"Evidence collection helpers for Rung checks.\n\nProvides utilities for reading files, searching for patterns, and\ndetecting CI workflow semantics without giving false credit for\nworkflow existence alone.\n\"\"\"\n\nfrom __future__ import annotations\n\nimport re\nimport json\nfrom pathlib import Path\nfrom typing import Optional\n\nfrom rung.sources import IGNORED_PARTS\n\n\ndef read_text(path: Path) -> Optional[str]:\n \"\"\"Read a file as UTF-8 text, returning None on failure.\"\"\"\n try:\n return path.read_text(encoding=\"utf-8\", errors=\"ignore\")\n except Exception:\n return None\n\n\ndef find_file(root: Path, candidates: list[Path]) -> list[Path]:\n \"\"\"Return existing files from a list of candidate paths.\"\"\"\n return [p for p in candidates if p.is_file() and bool((read_text(p) or \"\").strip())]\n\n\ndef has_valid_json(path: Path) -> bool:\n \"\"\"Return true for a nonempty regular file containing JSON data.\"\"\"\n if not path.is_file():\n return False\n try:\n value = json.loads(path.read_text(encoding=\"utf-8\"))\n return isinstance(value, (dict, list)) and bool(value)\n except (OSError, UnicodeError, json.JSONDecodeError):\n return False\n\n\ndef has_affirmative_pattern(content: str, patterns: list[str]) -> bool:\n \"\"\"Match policy language while rejecting obvious negations and examples.\"\"\"\n for line in content.splitlines():\n normalized = line.strip().lower()\n if not normalized or normalized.startswith((\"# example\", \"example:\")):\n continue\n if re.search(r\"\\b(?:do not|don't|not required|never use|must not|optional|convenient)\\b\", normalized):\n continue\n if any(has_pattern(line, pattern) for pattern in patterns):\n return True\n return False\n\n\ndef count_pattern(content: str, pattern: str) -> int:\n \"\"\"Count regex matches in content (case-insensitive).\"\"\"\n return len(re.findall(pattern, content, re.IGNORECASE))\n\n\ndef has_pattern(content: str, pattern: str) -> bool:\n \"\"\"Check if content matches a regex pattern (case-insensitive).\"\"\"\n return bool(re.search(pattern, content, re.IGNORECASE))\n\n\ndef has_ci_workflows(root: Path) -> bool:\n \"\"\"Check if .github/workflows directory exists and has .yml/.yaml files.\"\"\"\n ci_dir = root / \".github\" / \"workflows\"\n if not ci_dir.exists():\n return False\n return bool(list(ci_dir.glob(\"*.yml\")) or list(ci_dir.glob(\"*.yaml\")))\n\n\ndef ci_workflow_runs_tests(root: Path) -> bool:\n \"\"\"Semantic inspection: do CI workflows actually run tests?\n\n Reads workflow YAML files and checks for test-related step content.\n Finding any .github/workflows/*.yml does NOT automatically give credit.\n The workflow must contain test-related commands.\n \"\"\"\n ci_dir = root / \".github\" / \"workflows\"\n if not ci_dir.exists():\n return False\n test_patterns = [\n r\"(?:npm|yarn|pnpm)\\s+(?:test|run\\s+test)\",\n r\"(?:make|just)\\s+test\",\n r\"(?:go|cargo)\\s+test\",\n r\"python3?\\s+(?:-m\\s+)?pytest\",\n r\"python3?\\s+\\S+\\.py.*test\",\n r\"\\bpytest\\b\",\n r\"\\bunittest\\b\",\n r\"\\bjest\\b\",\n r\"\\bvitest\\b\",\n ]\n for wf_file in list(ci_dir.glob(\"*.yml\")) + list(ci_dir.glob(\"*.yaml\")):\n content = read_text(wf_file)\n if content is None:\n continue\n lines = content.splitlines()\n commands = []\n index = 0\n while index < len(lines):\n command = re.match(r\"^(\\s*)(?:-\\s*)?run:\\s*(.*)$\", lines[index])\n if not command:\n index += 1\n continue\n value = command.group(2).strip()\n if value in (\"|\", \">\", \"\"):\n base_indent = len(command.group(1))\n block = []\n index += 1\n while index < len(lines) and len(lines[index]) - len(lines[index].lstrip()) > base_indent:\n block.append(lines[index].strip())\n index += 1\n commands.extend(block)\n continue\n commands.append(value)\n index += 1\n for command in commands:\n if re.match(r\"^(?:echo|printf)\\b\", command):\n continue\n if any(has_pattern(command, pattern) for pattern in test_patterns):\n return True\n return False\n\n\ndef ci_workflow_enforces_gate(root: Path) -> bool:\n \"\"\"Semantic inspection: does CI enforce required status checks?\n\n Reads workflow YAML for branch protection hints. Note: actual branch\n protection rules are not observable from public repository contents\n without Administration read permission. This function only checks\n whether workflows reference required status checks or protection\n language. It does NOT claim enforcement.\n \"\"\"\n ci_dir = root / \".github\" / \"workflows\"\n if not ci_dir.exists():\n return False\n enforcement_patterns = [\n r\"required_status_checks\",\n r\"branch_protection\",\n r\"enforce_admins\",\n r\"required_pull_request_reviews\",\n ]\n for wf_file in list(ci_dir.glob(\"*.yml\")) + list(ci_dir.glob(\"*.yaml\")):\n content = read_text(wf_file)\n if content is None:\n continue\n for pattern in enforcement_patterns:\n if has_pattern(content, pattern):\n return True\n return False\n\n\ndef detect_build_commands(content: str) -> list[str]:\n \"\"\"Detect build/test commands in AGENTS.md or similar file content.\n\n Recognizes python3 (not just python), python -m, shell scripts, make,\n just, and package-manager commands.\n \"\"\"\n cmd_patterns = [\n r'(?:npm|yarn|pnpm)\\s+(?:test|run\\s+test|run\\s+build)',\n r'(?:make|just)\\s+(?:test|build|verify|all)',\n r'(?:go|cargo|rustc)\\s+(?:test|build)',\n r'python3\\s+(?:-m\\s+)?(?:pytest|unittest)',\n r'python3\\s+\\S+\\.py',\n r'python\\s+(?:-m\\s+)?(?:pytest|unittest)',\n r'python\\s+\\S+\\.py',\n r'\\./\\S+\\s+(?:test|build|verify)',\n r'ruby\\s+(?:-I\\S+\\s+)?\\S+test\\S*',\n r'dotnet\\s+(?:test|build)',\n ]\n found = []\n for line in content.splitlines():\n if re.search(r\"\\b(?:do not|don't|never|must not)\\b\", line, re.IGNORECASE):\n continue\n for pattern in cmd_patterns:\n found.extend(re.findall(pattern, line))\n return sorted(set(found))\n\n\ndef count_source_loc(path: Path) -> int:\n \"\"\"Count lines of code in a source file, excluding tests.\"\"\"\n try:\n with path.open(encoding=\"utf-8\", errors=\"ignore\") as source:\n return sum(1 for _ in source)\n except Exception:\n return 0\n\n\ndef is_test_file(path: Path) -> bool:\n \"\"\"Check if a path is a test file by name conventions.\"\"\"\n name = path.name.lower()\n stem = path.stem.lower()\n return (\n name.startswith(\"test_\")\n or stem.endswith(\"_test\")\n or \"tests\" in path.parts\n or \".test.\" in name\n or name.endswith(\".test.ts\")\n or name.endswith(\".test.js\")\n or name.endswith(\".spec.ts\")\n or name.endswith(\".spec.js\")\n )\n","rung.git_snapshot":"\"\"\"Safe, bounded materialization of regular files from a Git tree.\"\"\"\n\nfrom __future__ import annotations\n\nimport hashlib\nimport os\nimport selectors\nimport stat\nimport subprocess\nimport tempfile\nimport time\nfrom contextlib import contextmanager\nfrom dataclasses import dataclass\nfrom pathlib import Path\nfrom typing import Iterator\n\n\nMAX_TREE_BYTES = 16 * 1024 * 1024\nMAX_COMMIT_BYTES = 1024 * 1024\nMAX_BLOB_BYTES = 16 * 1024 * 1024\nMAX_TOTAL_BYTES = 128 * 1024 * 1024\nMAX_STDERR_BYTES = 64 * 1024\nGIT_TIMEOUT_SECONDS = 30\nGIT_PREFIX = [\n \"git\", \"-c\", \"core.fsmonitor=false\", \"-c\", \"core.hooksPath=/dev/null\",\n \"-c\", \"protocol.ext.allow=never\",\n]\n\n\nclass SnapshotError(Exception):\n \"\"\"Git metadata or content cannot produce a trusted regular-file tree.\"\"\"\n\n\n@dataclass(frozen=True)\nclass SnapshotFile:\n path: str\n mode: str\n oid: str\n data: bytes\n\n\ndef run_git(root: Path, args: tuple[str, ...], max_stdout: int) -> bytes:\n \"\"\"Run a non-interactive local Git command with strictly bounded pipes.\"\"\"\n stdout_buffer = bytearray()\n stderr_buffer = bytearray()\n try:\n process = subprocess.Popen(\n [*GIT_PREFIX, \"-C\", str(root), *args], stdin=subprocess.DEVNULL,\n stdout=subprocess.PIPE, stderr=subprocess.PIPE, env={\n **os.environ, \"GIT_NO_REPLACE_OBJECTS\": \"1\",\n \"GIT_OPTIONAL_LOCKS\": \"0\", \"GIT_TERMINAL_PROMPT\": \"0\", \"LC_ALL\": \"C\",\n },\n )\n except OSError as exc:\n raise SnapshotError(f\"cannot run git: {exc}\") from exc\n selector = selectors.DefaultSelector()\n assert process.stdout is not None and process.stderr is not None\n selector.register(process.stdout, selectors.EVENT_READ, (stdout_buffer, max_stdout))\n selector.register(process.stderr, selectors.EVENT_READ, (stderr_buffer, MAX_STDERR_BYTES))\n deadline = time.monotonic() + GIT_TIMEOUT_SECONDS\n try:\n while selector.get_map():\n remaining = deadline - time.monotonic()\n if remaining <= 0:\n raise SnapshotError(\"git command timed out\")\n events = selector.select(remaining)\n if not events:\n raise SnapshotError(\"git command timed out\")\n for key, _ in events:\n chunk = os.read(key.fd, 65_536)\n buffer, limit = key.data\n if not chunk:\n selector.unregister(key.fileobj)\n else:\n buffer.extend(chunk)\n if len(buffer) > limit:\n raise SnapshotError(\"git output exceeds safety bound\")\n returncode = process.wait(timeout=max(0.1, deadline - time.monotonic()))\n except Exception:\n process.kill()\n process.wait()\n raise\n finally:\n selector.close()\n process.stdout.close()\n process.stderr.close()\n if returncode:\n message = stderr_buffer.decode(\"utf-8\", \"replace\").strip()\n raise SnapshotError(f\"git command failed: {message or args[0]}\")\n if stderr_buffer:\n raise SnapshotError(\"git emitted unexpected diagnostics\")\n return bytes(stdout_buffer)\n\n\ndef _safe_path(raw: bytes) -> tuple[str, list[str]]:\n try:\n path = raw.decode(\"utf-8\", \"strict\")\n except UnicodeDecodeError as exc:\n raise SnapshotError(\"Git tree contains a non-UTF-8 path\") from exc\n parts = path.split(\"/\")\n if not path or path.startswith(\"/\") or \"\\\\\" in path or any(ord(char) < 32 or ord(char) == 127 for char in path) or any(\n part in (\"\", \".\", \"..\") for part in parts\n ):\n raise SnapshotError(\"Git tree contains an unsafe path\")\n return path, parts\n\n\ndef _validate_path(raw: bytes, files: set[str], directories: set[str]) -> str:\n path, parts = _safe_path(raw)\n if path in files or path in directories:\n raise SnapshotError(\"Git tree contains a duplicate or prefix path conflict\")\n parents = [\"/\".join(parts[:index]) for index in range(1, len(parts))]\n if any(parent in files for parent in parents):\n raise SnapshotError(\"Git tree contains a duplicate or prefix path conflict\")\n files.add(path)\n directories.update(parents)\n return path\n\n\ndef _validate_directory(raw: bytes, files: set[str], directories: set[str]) -> None:\n path, parts = _safe_path(raw)\n parents = [\"/\".join(parts[:index]) for index in range(1, len(parts))]\n if path in files or path in directories or any(parent in files for parent in parents):\n raise SnapshotError(\"Git tree contains a duplicate or prefix path conflict\")\n directories.add(path)\n directories.update(parents)\n\n\ndef _verified_object(root: Path, kind: str, oid: str, limit: int) -> bytes:\n data = run_git(root, (\"cat-file\", kind, oid), limit)\n actual = hashlib.sha1(\n kind.encode(\"ascii\") + b\" \" + str(len(data)).encode(\"ascii\") + b\"\\0\" + data\n ).hexdigest()\n if actual != oid:\n raise SnapshotError(f\"Git {kind} object digest mismatch\")\n return data\n\n\ndef load_commit(root: Path, commit_oid: str) -> tuple[str, int]:\n \"\"\"Read one authenticated commit object and return its tree and epoch.\"\"\"\n data = _verified_object(root, \"commit\", commit_oid, MAX_COMMIT_BYTES)\n headers, separator, _ = data.partition(b\"\\n\\n\")\n if not separator:\n raise SnapshotError(\"Git returned malformed commit metadata\")\n tree = None\n committer_epoch = None\n for line in headers.splitlines():\n if line.startswith(b\"tree \"):\n if tree is not None:\n raise SnapshotError(\"Git commit has duplicate tree metadata\")\n candidate = line[5:]\n if len(candidate) != 40 or any(byte not in b\"0123456789abcdef\" for byte in candidate):\n raise SnapshotError(\"Git commit has invalid tree metadata\")\n tree = candidate.decode(\"ascii\")\n elif line.startswith(b\"committer \"):\n if committer_epoch is not None:\n raise SnapshotError(\"Git commit has duplicate committer metadata\")\n fields = line.rsplit(b\" \", 2)\n if len(fields) != 3 or not fields[1].isdigit():\n raise SnapshotError(\"Git commit has invalid committer metadata\")\n committer_epoch = int(fields[1])\n if tree is None or committer_epoch is None:\n raise SnapshotError(\"Git commit is missing required metadata\")\n return tree, committer_epoch\n\n\ndef load_tree(root: Path, tree_oid: str) -> tuple[SnapshotFile, ...]:\n files: set[str] = set()\n directories: set[str] = set()\n snapshot: list[SnapshotFile] = []\n tree_bytes = 0\n content_bytes = 0\n\n def visit(oid: str, prefix: bytes, depth: int) -> None:\n nonlocal tree_bytes, content_bytes\n if depth > 128:\n raise SnapshotError(\"Git tree nesting exceeds safety bound\")\n raw = _verified_object(root, \"tree\", oid, MAX_TREE_BYTES)\n tree_bytes += len(raw)\n if tree_bytes > MAX_TREE_BYTES:\n raise SnapshotError(\"Git tree metadata exceeds safety bound\")\n position = 0\n while position < len(raw):\n space = raw.find(b\" \", position)\n nul = raw.find(b\"\\0\", space + 1) if space >= 0 else -1\n if space <= position or nul <= space + 1 or nul + 21 > len(raw):\n raise SnapshotError(\"Git returned malformed tree metadata\")\n mode = raw[position:space]\n name = raw[space + 1:nul]\n raw_oid = raw[nul + 1:nul + 21]\n position = nul + 21\n if b\"/\" in name:\n raise SnapshotError(\"Git tree contains an unsafe path\")\n child_oid = raw_oid.hex()\n raw_path = prefix + name\n if mode == b\"40000\":\n _validate_directory(raw_path, files, directories)\n visit(child_oid, raw_path + b\"/\", depth + 1)\n continue\n path = _validate_path(raw_path, files, directories)\n if mode in (b\"120000\", b\"160000\"):\n kind = \"symlinks\" if mode == b\"120000\" else \"submodules\"\n raise SnapshotError(f\"tracked {kind} are unsupported in receipt mode: {path}\")\n if mode not in (b\"100644\", b\"100755\"):\n raise SnapshotError(f\"unsupported Git tree entry: {path}\")\n data = _verified_object(root, \"blob\", child_oid, MAX_BLOB_BYTES)\n content_bytes += len(data)\n if content_bytes > MAX_TOTAL_BYTES:\n raise SnapshotError(\"Git tree contents exceed safety bound\")\n snapshot.append(SnapshotFile(path, mode.decode(\"ascii\"), child_oid, data))\n\n visit(tree_oid, b\"\", 0)\n return tuple(snapshot)\n\n\ndef _open_directory(parent_fd: int, name: str) -> int:\n flags = os.O_RDONLY | getattr(os, \"O_DIRECTORY\", 0) | getattr(os, \"O_NOFOLLOW\", 0)\n return os.open(name, flags, dir_fd=parent_fd)\n\n\ndef compare_worktree(root: Path, snapshot: tuple[SnapshotFile, ...]) -> None:\n root_fd = os.open(root, os.O_RDONLY | getattr(os, \"O_DIRECTORY\", 0))\n try:\n for entry in snapshot:\n directory_fd = os.dup(root_fd)\n descriptor = None\n try:\n parts = entry.path.split(\"/\")\n for part in parts[:-1]:\n next_fd = _open_directory(directory_fd, part)\n os.close(directory_fd)\n directory_fd = next_fd\n flags = os.O_RDONLY | getattr(os, \"O_NOFOLLOW\", 0)\n descriptor = os.open(parts[-1], flags, dir_fd=directory_fd)\n metadata = os.fstat(descriptor)\n if not stat.S_ISREG(metadata.st_mode):\n raise SnapshotError(f\"tracked path is not a regular file: {entry.path}\")\n if bool(metadata.st_mode & 0o111) != (entry.mode == \"100755\"):\n raise SnapshotError(f\"tracked file mode differs from HEAD: {entry.path}\")\n with os.fdopen(descriptor, \"rb\") as stream:\n descriptor = None\n data = stream.read(len(entry.data) + 1)\n if data != entry.data:\n raise SnapshotError(f\"tracked file content differs from HEAD: {entry.path}\")\n except OSError as exc:\n raise SnapshotError(f\"cannot safely read tracked file: {entry.path}\") from exc\n finally:\n if descriptor is not None:\n os.close(descriptor)\n os.close(directory_fd)\n finally:\n os.close(root_fd)\n\n\ndef _write_all(descriptor: int, data: bytes) -> None:\n remaining = memoryview(data)\n while remaining:\n written = os.write(descriptor, remaining)\n if written <= 0:\n raise OSError(\"short write\")\n remaining = remaining[written:]\n\n\n@contextmanager\ndef materialize(snapshot: tuple[SnapshotFile, ...]) -> Iterator[Path]:\n with tempfile.TemporaryDirectory(prefix=\"rung-snapshot-\") as directory:\n root = Path(directory)\n root_fd = os.open(root, os.O_RDONLY | getattr(os, \"O_DIRECTORY\", 0))\n created: set[str] = set()\n try:\n for entry in snapshot:\n directory_fd = os.dup(root_fd)\n descriptor = None\n try:\n parts = entry.path.split(\"/\")\n current = []\n for part in parts[:-1]:\n current.append(part)\n key = \"/\".join(current)\n if key not in created:\n os.mkdir(part, 0o700, dir_fd=directory_fd)\n created.add(key)\n next_fd = _open_directory(directory_fd, part)\n os.close(directory_fd)\n directory_fd = next_fd\n flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, \"O_NOFOLLOW\", 0)\n descriptor = os.open(parts[-1], flags, 0o700 if entry.mode == \"100755\" else 0o600, dir_fd=directory_fd)\n _write_all(descriptor, entry.data)\n os.close(descriptor)\n descriptor = None\n finally:\n if descriptor is not None:\n os.close(descriptor)\n os.close(directory_fd)\n except OSError as exc:\n raise SnapshotError(f\"cannot safely materialize Git tree: {exc}\") from exc\n finally:\n os.close(root_fd)\n yield root\n","rung.models":"\"\"\"Core data models for the Rung audit engine.\n\nDefines evidence states, confidence levels, source classifications,\nauthority recommendations, and the CheckResult dataclass that every\ncheck returns. Replaces the boolean-only pass/fail model with a\nsix-state evidence maturity model.\n\"\"\"\n\nfrom __future__ import annotations\n\nfrom dataclasses import dataclass, field\nfrom enum import Enum\nfrom typing import Optional\n\n\nclass EvidenceState(str, Enum):\n \"\"\"Six-state evidence maturity model replacing boolean pass/fail.\"\"\"\n ABSENT = \"absent\"\n CLAIMED = \"claimed\"\n DETECTED = \"detected\"\n ENFORCED = \"enforced\"\n VERIFIED = \"verified\"\n UNOBSERVABLE = \"unobservable\"\n\n\nclass Confidence(str, Enum):\n \"\"\"Per-finding confidence level.\"\"\"\n LOW = \"low\"\n MEDIUM = \"medium\"\n HIGH = \"high\"\n\n\nclass SourceClass(str, Enum):\n \"\"\"Source authority classification.\"\"\"\n STANDARD = \"standard\"\n PLATFORM_CONTROL = \"platform_control\"\n VENDOR_GUIDANCE = \"vendor_guidance\"\n INDUSTRY_PRACTICE = \"industry_practice\"\n EXEMPLAR = \"exemplar\"\n EDOWORKS_POLICY = \"edoworks_policy\"\n\n\nclass AuthorityLevel(str, Enum):\n \"\"\"Maximum recommended agent authority based on public evidence.\n\n A public-only report must never recommend autonomous merge or release\n when enforcement settings are unobservable.\n \"\"\"\n UNSAFE = \"unsafe\"\n LOCAL_ONLY = \"local_only\"\n PR_ONLY_PROVISIONAL = \"pr_only_provisional\"\n OWNER_EVIDENCE_REQUIRED = \"owner_evidence_required\"\n\n\n@dataclass\nclass CheckResult:\n \"\"\"Result of a single governance check.\n\n Replaces the boolean-only model. Every check returns an evidence state,\n confidence, blocking authority, evidence snippets, limitations,\n remediation steps, and source mappings.\n \"\"\"\n name: str\n description: str\n weight: int\n blocking: bool\n state: EvidenceState\n confidence: Confidence = Confidence.MEDIUM\n blocking_for: list[str] = field(default_factory=list)\n evidence: list[str] = field(default_factory=list)\n limitations: list[str] = field(default_factory=list)\n remediation: list[str] = field(default_factory=list)\n source_mappings: list[dict] = field(default_factory=list)\n\n @property\n def passed(self) -> bool:\n \"\"\"Backward compatibility: a check 'passes' if its state is\n detected, enforced, or verified. Absent, claimed, and unobservable\n do not pass. Blocking checks must pass for the quality gate.\"\"\"\n return self.state in (\n EvidenceState.DETECTED,\n EvidenceState.ENFORCED,\n EvidenceState.VERIFIED,\n )\n\n\n@dataclass\nclass AuditResult:\n \"\"\"Canonical immutable audit result (AuditResult v1).\n\n The free preview and paid PDF both render from this same object.\n Includes a report-data digest field that is computed over the\n canonical JSON of all other fields (excluding the digest itself).\n \"\"\"\n repository: str\n commit_sha: Optional[str]\n checks: list[CheckResult]\n score: int\n grade: str\n grade_label: str\n quality_gate: str\n authority: AuthorityLevel\n rung_version: str\n schema_version: str\n timestamp: str\n report_data_sha256: Optional[str] = None","rung.scoring":"\"\"\"Scoring and authority recommendation engine.\n\nComputes a numeric score (0-100), letter grade (A-E), quality gate\nverdict, and maximum recommended agent authority from check results.\nFile-size discipline is excluded from the governance score and appears\nonly in a non-scoring maintainability appendix.\n\"\"\"\n\nfrom rung.models import CheckResult, EvidenceState, AuthorityLevel\n\nGRADE_LABELS = {\n \"A\": \"Governance-Optimized\",\n \"B\": \"Managed\",\n \"C\": \"Defined\",\n \"D\": \"Repeatable\",\n \"E\": \"Initial / Absent\",\n}\n\n\ndef compute_score(results: list[CheckResult]) -> tuple[int, str, bool]:\n \"\"\"Compute score, grade, and gate from check results.\n\n Only checks with a weight > 0 contribute to the score. File-size\n discipline checks should have weight=0 to appear in the report\n without affecting the governance score.\n \"\"\"\n scoring_checks = [r for r in results if r.weight > 0]\n total_weight = sum(r.weight for r in scoring_checks)\n earned = sum(r.weight for r in scoring_checks if r.passed)\n score = round(earned / total_weight * 100) if total_weight > 0 else 0\n\n if score >= 90:\n grade = \"A\"\n elif score >= 80:\n grade = \"B\"\n elif score >= 70:\n grade = \"C\"\n elif score >= 60:\n grade = \"D\"\n else:\n grade = \"E\"\n\n gate_passed = all(r.passed for r in scoring_checks if r.blocking)\n return score, grade, gate_passed\n\n\ndef recommend_authority(results: list[CheckResult]) -> AuthorityLevel:\n \"\"\"Recommend maximum agent authority based on evidence states.\n\n A public-only report must never recommend autonomous merge or release\n when enforcement settings are unobservable. Authority is capped by\n the weakest blocking evidence state.\n \"\"\"\n has_unobservable = False\n has_claimed_only = False\n has_detected = False\n has_enforced = False\n\n for r in results:\n if not r.blocking or r.weight == 0:\n continue\n if r.state == EvidenceState.UNOBSERVABLE:\n has_unobservable = True\n elif r.state == EvidenceState.ABSENT:\n return AuthorityLevel.UNSAFE\n elif r.state == EvidenceState.CLAIMED:\n has_claimed_only = True\n elif r.state == EvidenceState.DETECTED:\n has_detected = True\n elif r.state == EvidenceState.ENFORCED:\n has_enforced = True\n elif r.state == EvidenceState.VERIFIED:\n has_enforced = True\n\n if has_unobservable:\n return AuthorityLevel.OWNER_EVIDENCE_REQUIRED\n if has_claimed_only:\n return AuthorityLevel.UNSAFE\n if has_detected and not has_enforced:\n return AuthorityLevel.PR_ONLY_PROVISIONAL\n if has_enforced:\n return AuthorityLevel.LOCAL_ONLY\n return AuthorityLevel.UNSAFE","rung.sources":"\"\"\"Cited sources for Rung governance checks.\n\nEach source is classified by authority level. Rung is \"informed by\"\nstandards and industry practices, not every check is directly derived\nfrom a standard.\n\"\"\"\n\nfrom rung.models import SourceClass\n\nSOURCES = {\n \"agents_md\": {\n \"name\": \"agents.md (Linux Foundation / AAIF)\",\n \"url\": \"https://agents.md\",\n \"note\": \"Cross-vendor agent policy spec, adopted by 60k+ repos. \"\n \"Defines AGENTS.md as the de-facto agent instruction file.\",\n \"classification\": SourceClass.STANDARD,\n },\n \"github_copilot\": {\n \"name\": \"GitHub Copilot repository custom instructions\",\n \"url\": \"https://docs.github.com/en/copilot/customizing-copilot/adding-repository-custom-instructions-for-github-copilot\",\n \"note\": \"Official GitHub docs for .github/copilot-instructions.md and \"\n \"AGENTS.md precedence (nearest file in directory tree wins).\",\n \"classification\": SourceClass.PLATFORM_CONTROL,\n },\n \"nist_rmf\": {\n \"name\": \"NIST AI Risk Management Framework 1.0\",\n \"url\": \"https://nist.gov/itl/ai-risk-management-framework\",\n \"note\": \"US government framework. Four functions: Govern, Map, \"\n \"Measure, Manage. Voluntary but widely adopted.\",\n \"classification\": SourceClass.STANDARD,\n },\n \"anthropic_multiagent\": {\n \"name\": \"Anthropic \u2014 How we built our multi-agent research system\",\n \"url\": \"https://www.anthropic.com/engineering/multi-agent-research-system\",\n \"note\": \"Production guidance on agent evaluation: LLM-as-judge rubrics, \"\n \"end-state evaluation, durable execution, checkpointing.\",\n \"classification\": SourceClass.VENDOR_GUIDANCE,\n },\n \"openai_codex\": {\n \"name\": \"openai/codex AGENTS.md (exemplar)\",\n \"url\": \"https://github.com/openai/codex/blob/main/AGENTS.md\",\n \"note\": \"322-line root AGENTS.md with 500/800 LoC thresholds, \"\n \"change-size guidance, mandatory integration tests.\",\n \"classification\": SourceClass.EXEMPLAR,\n },\n \"apache_airflow\": {\n \"name\": \"apache/airflow AGENTS.md (exemplar)\",\n \"url\": \"https://github.com/apache/airflow/blob/main/AGENTS.md\",\n \"note\": \"522-line AGENTS.md with Generated-by attribution, \"\n \"Drafted-by footer, Never-rules, apache-magpie framework.\",\n \"classification\": SourceClass.EXEMPLAR,\n },\n \"ibm_adlc\": {\n \"name\": \"IBM \u2014 Agent Development Lifecycle\",\n \"url\": \"https://www.ibm.com/think/topics/agent-development-lifecycle-adlc\",\n \"note\": \"Trace layer concept: every AI agent needs an action \"\n \"accountability trace.\",\n \"classification\": SourceClass.VENDOR_GUIDANCE,\n },\n \"slsa\": {\n \"name\": \"SLSA v1.2 (Supply-chain Levels for Software Artifacts)\",\n \"url\": \"https://slsa.dev/spec/v1.2/\",\n \"note\": \"OpenSSF standard for build provenance. Applies to \"\n \"agent-produced artifacts.\",\n \"classification\": SourceClass.STANDARD,\n },\n \"iso_42001\": {\n \"name\": \"ISO/IEC 42001:2023 \u2014 AI management system standard\",\n \"url\": \"https://www.iso.org/standard/83730.html\",\n \"note\": \"Certifiable AI Management System standard. Plan-do-check-act \"\n \"for AI.\",\n \"classification\": SourceClass.STANDARD,\n },\n \"github_branch_protection\": {\n \"name\": \"GitHub REST API \u2014 Branch Protection\",\n \"url\": \"https://docs.github.com/en/rest/branches/branch-protection\",\n \"note\": \"Branch protection settings require Administration read \"\n \"permission even for read operations on public repos.\",\n \"classification\": SourceClass.PLATFORM_CONTROL,\n },\n \"openssf_scorecard\": {\n \"name\": \"OpenSSF Scorecard\",\n \"url\": \"https://scorecard.dev/\",\n \"note\": \"Each check explains its risk, scoring logic, evidence, \"\n \"remediation, and limitations. Model for Rung's check format.\",\n \"classification\": SourceClass.INDUSTRY_PRACTICE,\n },\n}\n\nSOURCE_EXTENSIONS = {\".py\", \".swift\", \".gd\", \".ts\", \".tsx\", \".js\", \".jsx\", \".go\", \".rs\", \".java\", \".kt\", \".rb\"}\nIGNORED_PARTS = {\".git\", \"build\", \"dist\", \"node_modules\", \"DerivedData\", \".venv\", \"venv\", \".build\", \"Pods\", \"target\", \"__pycache__\"}\n\nWATCH_LOC = 500\nSMELL_LOC = 800\nDEFECT_LOC = 1200","rung.verification":"\"\"\"Reproducible verification receipts and replay observations.\"\"\"\n\nfrom __future__ import annotations\n\nimport builtins\nimport hashlib\nimport json\nimport os\nimport re\nimport secrets\nimport stat\nimport sys\nfrom collections.abc import Mapping\nfrom datetime import datetime, timezone\nfrom pathlib import Path\nfrom urllib.parse import urlparse\n\nfrom rung import __version__\nfrom rung.audit import SCHEMA_VERSION as AUDIT_SCHEMA_VERSION\nfrom rung.audit import run_audit\nfrom rung.git_snapshot import SnapshotError, compare_worktree, load_commit, load_tree, materialize, run_git\n\n\nRECEIPT_SCHEMA = \"RungVerificationReceipt/v1\"\nOBSERVATION_SCHEMA = \"RungReplayObservation/v1\"\nMAX_JSON_BYTES = 1_048_576\nHEX40 = re.compile(r\"[0-9a-f]{40}\\Z\")\nHEX64 = re.compile(r\"[0-9a-f]{64}\\Z\")\nREPOSITORY = re.compile(r\"github\\.com/[a-z0-9_.-]+/[a-z0-9_.-]+\\Z\")\nRFC3339_UTC = re.compile(r\"\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}Z\\Z\")\nVERSION = re.compile(r\"[0-9]+\\.[0-9]+\\.[0-9]+\\Z\")\nGATES = {\"PASS\", \"FAIL\"}\nAUTHORITIES = {\n \"unsafe\", \"local_only\", \"pr_only_provisional\", \"owner_evidence_required\",\n}\nLIMITATIONS = [\n \"Public repository contents cannot establish unobservable hosting-platform controls.\",\n \"Unsigned reproducibility evidence is not attestation, certification, enforcement proof, or correctness proof.\",\n]\nRECEIPT_FIELDS = {\n \"schema_version\", \"repository\", \"commit_sha\", \"tree_sha\", \"commit_timestamp\",\n \"audit_result_schema\", \"audit_result_sha256\", \"quality_gate\", \"authority\",\n \"rung_version\", \"engine_artifact_sha256\", \"argv\", \"limitations\",\n \"receipt_sha256\",\n}\nOBSERVATION_FIELDS = {\n \"schema_version\", \"receipt_sha256\", \"repository\", \"commit_sha\", \"tree_sha\",\n \"commit_timestamp\", \"audit_result_schema\", \"audit_result_sha256\",\n \"quality_gate\", \"authority\", \"rung_version\", \"engine_artifact_sha256\",\n \"matched\", \"mismatch_categories\", \"observation_sha256\",\n}\nMISMATCH_ORDER = (\n \"engine\", \"repository\", \"commit\", \"tree\", \"commit_timestamp\",\n \"audit_schema\", \"audit_digest\", \"quality_gate\", \"authority\",\n)\n\n\nclass VerificationError(Exception):\n \"\"\"Malformed input or a runtime condition that prevents trusted output.\"\"\"\n\n\ndef canonical_json(value: dict) -> bytes:\n return json.dumps(\n value, sort_keys=True, ensure_ascii=False, separators=(\",\", \":\")\n ).encode(\"utf-8\")\n\n\ndef _digest(value: dict, field: str) -> str:\n return hashlib.sha256(canonical_json({k: v for k, v in value.items() if k != field})).hexdigest()\n\n\ndef engine_artifact_sha256() -> str:\n \"\"\"Digest the generator's canonical UTF-8 bundled-source byte stream.\"\"\"\n bundled = getattr(builtins, \"_RUNG_BUNDLED_SOURCES\", None)\n if bundled is None:\n package = Path(__file__).resolve().parent\n sources = {}\n for path in sorted(package.rglob(\"*.py\")):\n source = path.read_text(encoding=\"utf-8\").replace(\"\\r\\n\", \"\\n\")\n relative = path.relative_to(package.parent).with_suffix(\"\")\n parts = list(relative.parts)\n if parts[-1] == \"__init__\":\n parts.pop()\n sources[\".\".join(parts)] = source\n else:\n sources = bundled\n digest = hashlib.sha256()\n if not isinstance(sources, Mapping):\n raise VerificationError(\"invalid bundled engine source map\")\n for name in sorted(sources):\n source = sources[name]\n if not isinstance(name, str) or not isinstance(source, str):\n raise VerificationError(\"invalid bundled engine source map\")\n digest.update(name.encode(\"utf-8\"))\n digest.update(b\"\\0\")\n digest.update(source.encode(\"utf-8\"))\n digest.update(b\"\\0\")\n return digest.hexdigest()\n\n\ndef _git(root: Path, *args: str, max_bytes: int = 262_144) -> str:\n try:\n output = run_git(root, tuple(args), max_bytes)\n except SnapshotError as exc:\n raise VerificationError(str(exc)) from exc\n try:\n return output.decode(\"utf-8\", \"strict\").strip()\n except UnicodeDecodeError as exc:\n raise VerificationError(\"git returned non-UTF-8 metadata\") from exc\n\n\ndef normalize_repository(remote: str) -> str:\n remote = remote.strip()\n if remote.startswith(\"git@github.com:\"):\n path = remote[len(\"git@github.com:\"):]\n elif remote.startswith(\"ssh://\") or remote.startswith(\"https://\"):\n parsed = urlparse(remote)\n if (parsed.hostname or \"\").lower() != \"github.com\":\n raise VerificationError(\"origin must be hosted on github.com\")\n if parsed.port is not None or parsed.query or parsed.fragment:\n raise VerificationError(\"unsupported GitHub origin URL\")\n if parsed.scheme == \"https\" and (parsed.username is not None or parsed.password is not None):\n raise VerificationError(\"credential-bearing GitHub origin is unsupported\")\n if parsed.scheme == \"ssh\" and parsed.username not in (None, \"git\"):\n raise VerificationError(\"unsupported GitHub SSH origin\")\n path = parsed.path.lstrip(\"/\")\n else:\n raise VerificationError(\"missing or unsupported origin URL\")\n if path.endswith(\".git\"):\n path = path[:-4]\n parts = path.split(\"/\")\n if len(parts) != 2 or not all(re.fullmatch(r\"[A-Za-z0-9_.-]+\", p) for p in parts):\n raise VerificationError(\"origin must identify one GitHub owner/repository\")\n return \"github.com/\" + \"/\".join(part.lower() for part in parts)\n\n\ndef _assert_engine_outside(root: Path) -> None:\n if getattr(builtins, \"_RUNG_BUNDLED_SOURCES\", None) is not None:\n engine_path = Path(sys.argv[0]).resolve(strict=True)\n else:\n engine_path = Path(__file__).resolve(strict=True).parent\n if engine_path.is_relative_to(root):\n raise VerificationError(\"Rung must be independently installed outside the audited checkout\")\n\n\ndef inspect_checkout(root_value: str | Path) -> dict:\n root = Path(root_value).resolve(strict=True)\n if not root.is_dir():\n raise VerificationError(\"root is not a directory\")\n top = Path(_git(root, \"rev-parse\", \"--show-toplevel\")).resolve(strict=True)\n if top != root:\n raise VerificationError(\"root must be the Git checkout root\")\n if _git(\n root, \"status\", \"--porcelain=v1\", \"--untracked-files=all\",\n \"--ignored=matching\", \"--ignore-submodules=all\",\n ):\n raise VerificationError(\"checkout must be clean, including untracked files\")\n _assert_engine_outside(root)\n commit = _git(root, \"rev-parse\", \"--verify\", \"HEAD\")\n if not HEX40.fullmatch(commit):\n raise VerificationError(\"Git returned a noncanonical object identity\")\n repository = normalize_repository(_git(root, \"remote\", \"get-url\", \"origin\"))\n try:\n tree, committer_epoch = load_commit(root, commit)\n timestamp = datetime.fromtimestamp(\n committer_epoch, timezone.utc\n ).strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n snapshot = load_tree(root, tree)\n compare_worktree(root, snapshot)\n except SnapshotError as exc:\n raise VerificationError(str(exc)) from exc\n return {\n \"root\": root, \"repository\": repository, \"commit_sha\": commit,\n \"tree_sha\": tree, \"commit_timestamp\": timestamp, \"snapshot\": snapshot,\n }\n\n\ndef _validate_output(path_value: str | Path, root: Path) -> tuple[Path, Path, tuple[int, int]]:\n path = Path(path_value).absolute()\n supplied_parent = path.parent\n try:\n parent = supplied_parent.resolve(strict=True)\n except OSError as exc:\n raise VerificationError(\"output parent must already exist\") from exc\n if not parent.is_dir() or supplied_parent.is_symlink():\n raise VerificationError(\"unsafe output parent\")\n destination = parent / path.name\n if destination.is_relative_to(root):\n raise VerificationError(\"output destination must be outside checkout\")\n if destination.exists() or destination.is_symlink():\n raise VerificationError(\"output destination already exists\")\n metadata = parent.stat(follow_symlinks=False)\n return destination, parent, (metadata.st_dev, metadata.st_ino)\n\n\ndef _open_validated_parent(parent: Path, expected: tuple[int, int]) -> int:\n flags = os.O_RDONLY | getattr(os, \"O_DIRECTORY\", 0) | getattr(os, \"O_NOFOLLOW\", 0)\n descriptor = os.open(parent.anchor, flags)\n try:\n for component in parent.parts[1:]:\n next_descriptor = os.open(component, flags, dir_fd=descriptor)\n os.close(descriptor)\n descriptor = next_descriptor\n metadata = os.fstat(descriptor)\n if (metadata.st_dev, metadata.st_ino) != expected:\n raise VerificationError(\"output parent changed during validation\")\n return descriptor\n except Exception:\n os.close(descriptor)\n raise\n\n\ndef _write_exclusive(path_value: str | Path, root: Path, value: dict) -> None:\n path, parent, parent_identity = _validate_output(path_value, root)\n data = canonical_json(value)\n descriptor = None\n directory_fd = None\n temporary = f\".rung-{secrets.token_hex(16)}\"\n try:\n directory_fd = _open_validated_parent(parent, parent_identity)\n file_flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL\n file_flags |= getattr(os, \"O_NOFOLLOW\", 0)\n descriptor = os.open(temporary, file_flags, 0o600, dir_fd=directory_fd)\n remaining = memoryview(data)\n while remaining:\n written = os.write(descriptor, remaining)\n if written == 0:\n raise OSError(\"short write\")\n remaining = remaining[written:]\n os.fsync(descriptor)\n os.close(descriptor)\n descriptor = None\n os.link(\n temporary, path.name, src_dir_fd=directory_fd,\n dst_dir_fd=directory_fd, follow_symlinks=False,\n )\n os.unlink(temporary, dir_fd=directory_fd)\n temporary = \"\"\n os.fsync(directory_fd)\n except FileExistsError as exc:\n raise VerificationError(\"output destination already exists\") from exc\n except OSError as exc:\n raise VerificationError(f\"cannot safely create output: {exc}\") from exc\n finally:\n if descriptor is not None:\n os.close(descriptor)\n if temporary and directory_fd is not None:\n try:\n os.unlink(temporary, dir_fd=directory_fd)\n except OSError:\n pass\n if directory_fd is not None:\n os.close(directory_fd)\n\n\ndef _pairs(pairs: list[tuple[str, object]]) -> dict:\n value = {}\n for key, item in pairs:\n if key in value:\n raise VerificationError(f\"duplicate JSON member: {key}\")\n value[key] = item\n return value\n\n\ndef _load_json(path_value: str | Path) -> dict:\n path = Path(path_value)\n descriptor = None\n try:\n if path.is_symlink() or not path.is_file():\n raise VerificationError(\"receipt must be a regular non-symlink file\")\n flags = os.O_RDONLY | getattr(os, \"O_NOFOLLOW\", 0)\n descriptor = os.open(path, flags)\n if not stat.S_ISREG(os.fstat(descriptor).st_mode):\n raise VerificationError(\"receipt must be a regular file\")\n with os.fdopen(descriptor, \"rb\") as stream:\n descriptor = None\n data = stream.read(MAX_JSON_BYTES + 1)\n except OSError as exc:\n raise VerificationError(f\"cannot read receipt: {exc}\") from exc\n finally:\n if descriptor is not None:\n os.close(descriptor)\n if len(data) > MAX_JSON_BYTES:\n raise VerificationError(\"receipt exceeds size limit\")\n try:\n text = data.decode(\"utf-8\", \"strict\")\n value = json.loads(text, object_pairs_hook=_pairs)\n except (UnicodeDecodeError, json.JSONDecodeError) as exc:\n raise VerificationError(\"receipt is not strict UTF-8 JSON\") from exc\n if not isinstance(value, dict):\n raise VerificationError(\"receipt must be a JSON object\")\n return value\n\n\ndef validate_receipt(value: dict) -> None:\n if set(value) != RECEIPT_FIELDS:\n raise VerificationError(\"receipt has missing or unknown fields\")\n string_fields = RECEIPT_FIELDS - {\"argv\", \"limitations\"}\n if any(not isinstance(value[field], str) for field in string_fields):\n raise VerificationError(\"receipt string field has invalid type\")\n if value[\"schema_version\"] != RECEIPT_SCHEMA:\n raise VerificationError(\"unsupported receipt schema\")\n if not REPOSITORY.fullmatch(value[\"repository\"]):\n raise VerificationError(\"invalid repository identity\")\n for field in (\"commit_sha\", \"tree_sha\"):\n if not HEX40.fullmatch(value[field]):\n raise VerificationError(f\"invalid {field}\")\n for field in (\"audit_result_sha256\", \"engine_artifact_sha256\", \"receipt_sha256\"):\n if not HEX64.fullmatch(value[field]):\n raise VerificationError(f\"invalid {field}\")\n if not VERSION.fullmatch(value[\"audit_result_schema\"]) or not VERSION.fullmatch(value[\"rung_version\"]):\n raise VerificationError(\"invalid component version\")\n if value[\"quality_gate\"] not in GATES or value[\"authority\"] not in AUTHORITIES:\n raise VerificationError(\"invalid gate or authority\")\n if not RFC3339_UTC.fullmatch(value[\"commit_timestamp\"]):\n raise VerificationError(\"invalid commit timestamp\")\n expected_argv = [\n \"rung\", \"--root\", \"{checkout}\", \"--commit-sha\", value[\"commit_sha\"],\n \"--repository\", value[\"repository\"], \"--timestamp\",\n value[\"commit_timestamp\"], \"--json\",\n ]\n if value[\"argv\"] != expected_argv or value[\"limitations\"] != LIMITATIONS:\n raise VerificationError(\"noncanonical receipt metadata\")\n if value[\"receipt_sha256\"] != _digest(value, \"receipt_sha256\"):\n raise VerificationError(\"receipt self-digest mismatch\")\n\n\ndef _audit(identity: dict):\n try:\n with materialize(identity[\"snapshot\"]) as immutable_root:\n return run_audit(\n immutable_root, commit_sha=identity[\"commit_sha\"],\n repository=identity[\"repository\"], timestamp=identity[\"commit_timestamp\"],\n )\n except SnapshotError as exc:\n raise VerificationError(str(exc)) from exc\n\n\ndef _confirm_unchanged(identity: dict) -> None:\n current = inspect_checkout(identity[\"root\"])\n for field in (\"repository\", \"commit_sha\", \"tree_sha\", \"commit_timestamp\"):\n if current[field] != identity[field]:\n raise VerificationError(\"checkout changed during audit\")\n\n\ndef create_receipt(root_value: str | Path, output: str | Path) -> dict:\n identity = inspect_checkout(root_value)\n result = _audit(identity)\n _confirm_unchanged(identity)\n receipt = {\n \"schema_version\": RECEIPT_SCHEMA,\n \"repository\": identity[\"repository\"],\n \"commit_sha\": identity[\"commit_sha\"],\n \"tree_sha\": identity[\"tree_sha\"],\n \"commit_timestamp\": identity[\"commit_timestamp\"],\n \"audit_result_schema\": result.schema_version,\n \"audit_result_sha256\": result.report_data_sha256,\n \"quality_gate\": result.quality_gate,\n \"authority\": result.authority.value,\n \"rung_version\": __version__,\n \"engine_artifact_sha256\": engine_artifact_sha256(),\n \"argv\": [\n \"rung\", \"--root\", \"{checkout}\", \"--commit-sha\", identity[\"commit_sha\"],\n \"--repository\", identity[\"repository\"], \"--timestamp\",\n identity[\"commit_timestamp\"], \"--json\",\n ],\n \"limitations\": LIMITATIONS,\n \"receipt_sha256\": \"\",\n }\n receipt[\"receipt_sha256\"] = _digest(receipt, \"receipt_sha256\")\n _write_exclusive(output, identity[\"root\"], receipt)\n return receipt\n\n\ndef replay_receipt(root_value: str | Path, receipt_path: str | Path, output: str | Path) -> tuple[dict, bool]:\n receipt = _load_json(receipt_path)\n validate_receipt(receipt)\n identity = inspect_checkout(root_value)\n result = _audit(identity)\n _confirm_unchanged(identity)\n observed = {\n \"repository\": identity[\"repository\"], \"commit_sha\": identity[\"commit_sha\"],\n \"tree_sha\": identity[\"tree_sha\"], \"commit_timestamp\": identity[\"commit_timestamp\"],\n \"audit_result_schema\": result.schema_version,\n \"audit_result_sha256\": result.report_data_sha256,\n \"quality_gate\": result.quality_gate, \"authority\": result.authority.value,\n \"rung_version\": __version__, \"engine_artifact_sha256\": engine_artifact_sha256(),\n }\n comparisons = {\n \"engine\": (receipt[\"rung_version\"], receipt[\"engine_artifact_sha256\"]) == (observed[\"rung_version\"], observed[\"engine_artifact_sha256\"]),\n \"repository\": receipt[\"repository\"] == observed[\"repository\"],\n \"commit\": receipt[\"commit_sha\"] == observed[\"commit_sha\"],\n \"tree\": receipt[\"tree_sha\"] == observed[\"tree_sha\"],\n \"commit_timestamp\": receipt[\"commit_timestamp\"] == observed[\"commit_timestamp\"],\n \"audit_schema\": receipt[\"audit_result_schema\"] == observed[\"audit_result_schema\"],\n \"audit_digest\": receipt[\"audit_result_sha256\"] == observed[\"audit_result_sha256\"],\n \"quality_gate\": receipt[\"quality_gate\"] == observed[\"quality_gate\"],\n \"authority\": receipt[\"authority\"] == observed[\"authority\"],\n }\n mismatches = [category for category in MISMATCH_ORDER if not comparisons[category]]\n observation = {\n \"schema_version\": OBSERVATION_SCHEMA,\n \"receipt_sha256\": receipt[\"receipt_sha256\"],\n **observed,\n \"matched\": not mismatches,\n \"mismatch_categories\": mismatches,\n \"observation_sha256\": \"\",\n }\n if set(observation) != OBSERVATION_FIELDS:\n raise AssertionError(\"observation schema implementation error\")\n observation[\"observation_sha256\"] = _digest(observation, \"observation_sha256\")\n _write_exclusive(output, identity[\"root\"], observation)\n return observation, not mismatches\n\n\ndef verification_main(command: str, root: str, receipt: str, observation: str | None = None) -> int:\n try:\n if command == \"verify\":\n create_receipt(root, receipt)\n return 0\n _, matched = replay_receipt(root, receipt, observation or \"\")\n return 0 if matched else 2\n except VerificationError as exc:\n print(f\"rung {command}: {exc}\", file=sys.stderr)\n return 1\n except Exception as exc:\n print(f\"rung {command}: runtime error: {exc}\", file=sys.stderr)\n return 1\n"}) builtins._RUNG_BUNDLED_SOURCES = _SOURCES class _BundledLoader(importlib.abc.MetaPathFinder, importlib.abc.Loader): def find_spec(self, fullname, path=None, target=None): if fullname not in _SOURCES: return None is_package = fullname == "rung" or any( name.startswith(fullname + ".") for name in _SOURCES ) return importlib.util.spec_from_loader(fullname, self, is_package=is_package) def create_module(self, spec): return None def exec_module(self, module): module.__file__ = "/" + module.__name__.replace(".", "/") + ".py" exec(compile(_SOURCES[module.__name__], module.__file__, "exec"), module.__dict__) import sys sys.meta_path.insert(0, _BundledLoader()) from rung.cli import main if __name__ == "__main__": raise SystemExit(main())