diff --git a/.github/workflows/command-center-invariants.yml b/.github/workflows/command-center-invariants.yml index 9e2db4b..93f6ae6 100644 --- a/.github/workflows/command-center-invariants.yml +++ b/.github/workflows/command-center-invariants.yml @@ -1,28 +1,13 @@ name: command-center-invariants on: - pull_request: - paths: - - "README.md" - - "profile/**" - - "architecture/**" - - "governance/**" - - "wiki/**" - - ".github/pull_request_template.md" - - ".github/workflows/command-center-invariants.yml" - - "scripts/verify-command-center-invariants.py" + pull_request: {} push: branches: - main - paths: - - "README.md" - - "profile/**" - - "architecture/**" - - "governance/**" - - "wiki/**" - - ".github/pull_request_template.md" - - ".github/workflows/command-center-invariants.yml" - - "scripts/verify-command-center-invariants.py" + workflow_dispatch: {} + schedule: + - cron: "23 7 * * 1" permissions: contents: read @@ -31,8 +16,248 @@ jobs: command-center-invariants: runs-on: ubuntu-latest steps: - - name: Checkout - uses: actions/checkout@v4 + - name: Checkout command-center authority + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 + with: + persist-credentials: false + fetch-depth: 0 + + - name: Set up Python + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + with: + python-version: "3.12" + + - name: Install structural verifier dependency + run: python -m pip install --disable-pip-version-check PyYAML==6.0.2 - name: Verify command-center invariants - run: python scripts/verify-command-center-invariants.py + run: python scripts/verify-command-center-invariants.py --self-test + + - name: Run hostile command-center unit tests + run: python -B -m unittest discover -s tests + + - name: Verify patch whitespace + shell: bash + run: | + set -euo pipefail + if [[ "${{ github.event_name }}" == "pull_request" ]]; then + git diff --check "${{ github.event.pull_request.base.sha }}...${{ github.event.pull_request.head.sha }}" + else + git show --check --format= HEAD + fi + + seven-repository-convergence: + needs: command-center-invariants + runs-on: ubuntu-latest + env: + PYTHONDONTWRITEBYTECODE: "1" + steps: + - name: Checkout workflow authority at the event revision + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + path: source-set/.github + fetch-depth: 0 + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + with: + python-version: "3.12" + + - name: Set up Node + uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 + with: + node-version: "20" + + - name: Install bounded verifier dependencies + run: python -m pip install --disable-pip-version-check PyYAML==6.0.2 pytest==8.3.5 + + - name: Resolve governance/CONVERGENCE_SOURCE_MANIFEST.json + shell: bash + run: | + set -euo pipefail + mkdir -p verification-work verification-artifacts + python -B source-set/.github/scripts/verify-command-center-invariants.py \ + --emit-source-manifest verification-work/resolved-source-manifest.json \ + --event-sha "${{ github.event.pull_request.head.sha || github.sha }}" + + - name: Checkout six immutable sibling revisions without credentials + shell: bash + run: | + set -euo pipefail + python - verification-work/resolved-source-manifest.json > verification-work/sibling-revisions.tsv <<'PY' + import json + import sys + + value = json.load(open(sys.argv[1], encoding="utf-8")) + for entry in value["repositories"]: + if entry["repository"] != ".github": + print(f'{entry["repository"]}\t{entry["revision"]}') + PY + while IFS=$'\t' read -r repo revision; do + test -n "$repo" + test -n "$revision" + mkdir "source-set/$repo" + git -C "source-set/$repo" init --quiet + git -C "source-set/$repo" remote add origin "https://github.com/HawkinsOperations/$repo.git" + fetch_complete=0 + for attempt in 1 2 3 4 5 6; do + if + git -C "source-set/$repo" fetch --quiet origin "$revision" + then + fetch_complete=1 + break + fi + if [ "$attempt" -lt 6 ]; then + sleep 5 + fi + done + test "$fetch_complete" -eq 1 + git -C "source-set/$repo" checkout --quiet --detach "$revision" + test "$(git -C "source-set/$repo" rev-parse HEAD)" = "$revision" + done < verification-work/sibling-revisions.tsv + + - name: Verify the exact clean detached source set + shell: bash + run: | + set -euo pipefail + python -B source-set/.github/scripts/verify-command-center-invariants.py \ + --verify-source-set source-set \ + --resolved-manifest verification-work/resolved-source-manifest.json \ + --source-revisions-output verification-artifacts/source-revisions.json + readarray -t authority_shas < <(python - verification-work/resolved-source-manifest.json <<'PY' + import json + import sys + + value = json.load(open(sys.argv[1], encoding="utf-8")) + by_repo = {entry["repository"]: entry["revision"] for entry in value["repositories"]} + print(by_repo[".github"]) + print(by_repo["hawkinsoperations-proof"]) + print(by_repo["hawkinsoperations-platform"]) + PY + ) + test "${#authority_shas[@]}" -eq 3 + printf 'HAWKINS_COMMAND_CENTER_IMMUTABLE_OBSERVED_SHA=%s\n' "${authority_shas[0]}" >> "$GITHUB_ENV" + printf 'HAWKINS_PROOF_IMMUTABLE_MANIFEST_SHA=%s\n' "${authority_shas[1]}" >> "$GITHUB_ENV" + printf 'HAWKINS_PLATFORM_IMMUTABLE_OBSERVED_SHA=%s\n' "${authority_shas[2]}" >> "$GITHUB_ENV" + + - name: Detect durable sibling main-content drift + if: github.event_name != 'pull_request' + run: >- + python -B source-set/.github/scripts/verify-command-center-invariants.py + --verify-remote-main-content source-set + --resolved-manifest verification-work/resolved-source-manifest.json + + - name: Verify detection authority and hostile paths + shell: bash + run: | + set -euo pipefail + python -B source-set/hawkinsoperations-detections/scripts/verify_detection_contract.py + python -B source-set/hawkinsoperations-detections/scripts/verify_detection_promotion_matrix.py \ + --validation-registry source-set/hawkinsoperations-validation/validation/VALIDATION_REGISTRY.yml \ + --proof-index source-set/hawkinsoperations-proof/proof/indexes/DETECTION_PROOF_STATUS_INDEX.yml \ + --require-sibling-handoffs + python -B -m unittest discover -s source-set/hawkinsoperations-detections/tests + + - name: Verify validation authority and fail-closed parity + shell: bash + run: | + set -euo pipefail + python -B source-set/hawkinsoperations-validation/scripts/verify_validation_registry.py --detections-root source-set/hawkinsoperations-detections --detections-ref "$(git -C source-set/hawkinsoperations-detections rev-parse HEAD)" --source-manifest source-set/hawkinsoperations-validation/validation/SOURCE_AUTHORITY_MANIFEST.json + python -B source-set/hawkinsoperations-validation/scripts/verify_all_validation_packages.py --source-contract required + python -B source-set/hawkinsoperations-validation/scripts/verify_validation_contract.py + python -B source-set/hawkinsoperations-validation/scripts/verify_wazuh_logtest_registry.py + python -B source-set/hawkinsoperations-validation/scripts/verify_ho_lab_wazuh_001.py + python -B source-set/hawkinsoperations-validation/scripts/verify_cross_repo_claim_parity.py --repo-root source-set --enforce + PYTHONPATH="$GITHUB_WORKSPACE/source-set/hawkinsoperations-validation" python -B -m unittest discover -s source-set/hawkinsoperations-validation/tests + + - name: Verify proof authority and reverse inventory + shell: bash + run: | + set -euo pipefail + python -B source-set/hawkinsoperations-proof/scripts/verify_detection_proof_status_index.py + python -B source-set/hawkinsoperations-proof/scripts/verify_proof_integrity.py + python -B -m unittest discover -s source-set/hawkinsoperations-proof/tests + + - name: Verify platform source contract and seven-source convergence + shell: bash + run: | + set -euo pipefail + python -B source-set/hawkinsoperations-platform/scripts/verify-public-status-source-contract.py --format json + python -B source-set/hawkinsoperations-platform/scripts/ho_factory.py \ + hoxline-case-growth-convergence-verify --repo-root source-set --format json + python -B -m unittest discover -s source-set/hawkinsoperations-platform/tests + + - name: Install Hoxline from the checked immutable source + run: python -m pip install --disable-pip-version-check -e source-set/hoxline + + - name: Verify Hoxline Case Growth pair and replay integrity + shell: bash + run: | + set -euo pipefail + python -B -m compileall -q source-set/hoxline/src source-set/hoxline/tests + python -B -m unittest discover -s source-set/hoxline/tests + python -B -m pytest -q source-set/hoxline/tests + python -B -m hoxline.cli case-growth index \ + --repo-root source-set \ + --format json \ + --paired-output-base verification-work/current-case-growth-index + python -B -m hoxline.cli case-growth verify \ + --repo-root source-set \ + --snapshot verification-work/current-case-growth-index.json + python -B -m hoxline.cli case-growth verify \ + --repo-root source-set \ + --snapshot source-set/hoxline/examples/case-growth/current-case-growth-index.json + python -B -m hoxline review batch run \ + --index source-set/hoxline/examples/review/multi-artifact-review-index-v1.json \ + --output verification-work/batch \ + --force \ + --format json + python -B -m hoxline review batch verify \ + --run verification-work/batch/batch-machine-state.json + + - name: Install Website dependencies from the checked lockfile + run: npm ci --prefix source-set/hawkinsoperations-website + + - name: Verify Website rendering-only status plane and static build + shell: bash + run: | + set -euo pipefail + npm --prefix source-set/hawkinsoperations-website run public-status:generate:check + npm --prefix source-set/hawkinsoperations-website run public-status:verify + npm --prefix source-set/hawkinsoperations-website run public-status:self-test + npm --prefix source-set/hawkinsoperations-website run public-status:owner-self-test + npm --prefix source-set/hawkinsoperations-website run public-status:source-checkout-test + npm --prefix source-set/hawkinsoperations-website run public-status:freshness-reachability-test + npm --prefix source-set/hawkinsoperations-website run public-status:dirty-provenance-test + npm --prefix source-set/hawkinsoperations-website run public-status:nested-claim-test + npm --prefix source-set/hawkinsoperations-website run public-status:strict-json-test + npm --prefix source-set/hawkinsoperations-website run public-status:eol-self-test + npm --prefix source-set/hawkinsoperations-website run check:site + npm --prefix source-set/hawkinsoperations-website run typecheck + npm --prefix source-set/hawkinsoperations-website run build + + - name: Write closed-schema verification summary + run: >- + python -B source-set/.github/scripts/verify-command-center-invariants.py + --write-verification-summary + verification-artifacts/verification-summary.json + + - name: Validate upload artifacts + run: >- + python -B source-set/.github/scripts/verify-command-center-invariants.py + --validate-artifacts + verification-artifacts + --artifact-source-set + source-set + + - name: Upload sanitized convergence records + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 + with: + name: seven-repository-convergence-${{ github.run_id }} + path: | + verification-artifacts/source-revisions.json + verification-artifacts/verification-summary.json + if-no-files-found: error + retention-days: 14 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..43ae0e2 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.py[cod] diff --git a/governance/COMMAND_CENTER_INVARIANTS.json b/governance/COMMAND_CENTER_INVARIANTS.json index dbcb12d..623e3f9 100644 --- a/governance/COMMAND_CENTER_INVARIANTS.json +++ b/governance/COMMAND_CENTER_INVARIANTS.json @@ -13,7 +13,19 @@ "governance/PR_REVIEW_AUTHORITY.md", "governance/CROSS_REPO_PROMOTION_MAP.md", "wiki/11_ORG_SYSTEM_MAP.md", - ".github/pull_request_template.md" + ".github/pull_request_template.md", + ".github/workflows/command-center-invariants.yml", + "scripts/verify-command-center-invariants.py", + "governance/CONVERGENCE_SOURCE_MANIFEST.json" + ], + "cross_repo_repositories": [ + ".github", + "hawkinsoperations-detections", + "hawkinsoperations-validation", + "hawkinsoperations-platform", + "hawkinsoperations-proof", + "hawkinsoperations-website", + "hoxline" ], "invariants": { "github_repo_role": ".github is reviewer routing and governance shell only", @@ -26,6 +38,7 @@ "ledger_public_safe_status": "NOT_PUBLIC_SAFE", "reviewer_metrics_pipeline": "Reviewer metrics pipeline keeps Lifetime Governed Cases separate from detection activity, validation cases, proof records, blocked claims, and Project Board reconciliation status", "reviewer_metrics_counts": "Reviewer metrics values are authority-owned snapshots in proof/platform records; front-door text must route to those records instead of copying changing counts", + "cross_repo_convergence": "Read-only verification checks exactly seven repositories at an explicit immutable PR-head matrix, records and verifies every checked revision, fails closed on drift, and does not write main, merge, mutate the Lifetime Case Ledger, or promote proof/public status", "ho_det_001_public_ceiling": "CONTROLLED_TEST_VALIDATED", "runtime_signal_public_promotions": "runtime-active, signal-observed, evidence-linked public proof, public-safe, production-ready, fleet-wide, AWS-live, Cribl-routed, Wazuh-routed, autonomous SOC, AI-approved, AI-decided, analyst-approved, and live Splunk claims remain blocked unless separately proven and approved", "standing_controls": ".github#8 and .github#10 remain standing controls", diff --git a/governance/CONVERGENCE_SOURCE_MANIFEST.json b/governance/CONVERGENCE_SOURCE_MANIFEST.json new file mode 100644 index 0000000..eafbb19 --- /dev/null +++ b/governance/CONVERGENCE_SOURCE_MANIFEST.json @@ -0,0 +1,64 @@ +{ + "schema": "hawkinsoperations-convergence-source-manifest-v1", + "manifest_id": "HAWKINSOPERATIONS_SEVEN_SOURCE_PR_HEAD_MATRIX_V1", + "repositories": [ + { + "repository": ".github", + "canonical_repository": "HawkinsOperations/.github", + "revision_source": "github_event_sha", + "authority_content_revision": "6e6763a81d6af09c2e4588462b56117ce82c2f88", + "tree_source": "github_event_tree" + }, + { + "repository": "hawkinsoperations-detections", + "canonical_repository": "HawkinsOperations/hawkinsoperations-detections", + "revision": "9e01f43fb350de3370f8c01a323dcdcdf2e33147", + "authority_content_revision": "f8bc0a0925113ca815bf5692081b5216162cc918", + "reviewed_tree_sha": "4135fc6fafb3bc842096ee58fa7dd53d24172b65" + }, + { + "repository": "hawkinsoperations-validation", + "canonical_repository": "HawkinsOperations/hawkinsoperations-validation", + "revision": "677b704150b0f5f333c27913dd481b4be6a78ab7", + "authority_content_revision": "ebf52f7c6c9b78de767272cc56fccdc584f5c4e0", + "reviewed_tree_sha": "b4a12cfe4a67e5b66171f73cd228c2691789dcc6" + }, + { + "repository": "hawkinsoperations-platform", + "canonical_repository": "HawkinsOperations/hawkinsoperations-platform", + "revision": "d2901f303a2047436d1ada2d97f2eb4310380585", + "authority_content_revision": "a667c4de8b478fe165c3ec612e642bbd5d879492", + "reviewed_tree_sha": "0ad3ad8ff804e6b0fff9c4bf5eb1c913a76f602c" + }, + { + "repository": "hawkinsoperations-proof", + "canonical_repository": "HawkinsOperations/hawkinsoperations-proof", + "revision": "77b7874dd753369792330508fa3438cf397cd050", + "authority_content_revision": "042a918ad4a8473cd5abcfd575072fc094639682", + "reviewed_tree_sha": "68fc8c604ffebb392e2fa6b205a920e9560b424b" + }, + { + "repository": "hawkinsoperations-website", + "canonical_repository": "HawkinsOperations/hawkinsoperations-website", + "revision": "ee30ae81d31e8f27fa779ddb42470f7d27db1f33", + "authority_content_revision": "5856f8e69527b5e61c3953b88a2ad4c088268655", + "reviewed_tree_sha": "672da08e0e5c42d7be543cb9bdcfa45ccb59daa2" + }, + { + "repository": "hoxline", + "canonical_repository": "HawkinsOperations/hoxline", + "revision": "cd797da491f07b9a0130278d7245f51962dd82c0", + "authority_content_revision": "1cb97efc45ffe753389105645c25ed7fe57cf9e5", + "reviewed_tree_sha": "29dc0d921bbc342f30d8be46cf88f049ea591f0a" + } + ], + "constraints": { + "exact_repository_count": 7, + "read_only": true, + "default_branch_fallback": false, + "require_detached_exact_revision": true, + "record_checked_revisions": true, + "consumer_outputs_are_not_authority": true, + "proof_ceiling": "CONTROLLED_REPO_CONVERGENCE_AND_LOCAL_FIXTURE_REVIEW_ONLY" + } +} diff --git a/scripts/verify-command-center-invariants.py b/scripts/verify-command-center-invariants.py index e25af3b..a722bb8 100644 --- a/scripts/verify-command-center-invariants.py +++ b/scripts/verify-command-center-invariants.py @@ -1,17 +1,187 @@ #!/usr/bin/env python3 -"""Fail-closed checks for the HawkinsOperations .github command center.""" +"""Fail-closed checks for the HawkinsOperations command-center workflow.""" from __future__ import annotations +import argparse +import hashlib import json +import os import re +import subprocess import sys -from pathlib import Path +import unicodedata +from pathlib import Path, PurePosixPath, PureWindowsPath +from typing import Any +from urllib.parse import unquote + +try: + import yaml +except ImportError: # pragma: no cover - reported as a deterministic verifier failure + yaml = None ROOT = Path(__file__).resolve().parents[1] MANIFEST_PATH = ROOT / "governance" / "COMMAND_CENTER_INVARIANTS.json" +SOURCE_MANIFEST_PATH = ROOT / "governance" / "CONVERGENCE_SOURCE_MANIFEST.json" +WORKFLOW_PATH = ROOT / ".github" / "workflows" / "command-center-invariants.yml" TEXT_SCOPES = ["README.md", "profile", "architecture", "governance", "wiki", ".github"] +EXACT_REPOSITORIES = [ + ".github", + "hawkinsoperations-detections", + "hawkinsoperations-validation", + "hawkinsoperations-platform", + "hawkinsoperations-proof", + "hawkinsoperations-website", + "hoxline", +] +CANONICAL_ORIGINS = { + repository: f"https://github.com/HawkinsOperations/{repository}.git" + for repository in EXACT_REPOSITORIES +} +def sanitized_git_environment() -> dict[str, str]: + environment = { + key: value + for key, value in os.environ.items() + if not key.casefold().startswith("git_") + } + environment["GIT_NO_REPLACE_OBJECTS"] = "1" + environment["GIT_TERMINAL_PROMPT"] = "0" + return environment +CANONICAL_AUTHORITY_PATHS = { + ".github": "governance/COMMAND_CENTER_INVARIANTS.json", + "hawkinsoperations-detections": "detections/DETECTION_PROMOTION_MATRIX.yml", + "hawkinsoperations-validation": "validation/VALIDATION_REGISTRY.yml", + "hawkinsoperations-platform": "contracts/public-status-source-contract-v1.json", + "hawkinsoperations-proof": "proof/indexes/DETECTION_PROOF_STATUS_INDEX.yml", + "hawkinsoperations-website": "schemas/public-status-v0.schema.json", + "hoxline": "src/hoxline/case_growth/collector.py", +} +PINNED_ACTIONS = { + "actions/checkout": "11d5960a326750d5838078e36cf38b85af677262", + "actions/setup-python": "a26af69be951a213d495a4c3e4e4022e16d87065", + "actions/setup-node": "49933ea5288caeca8642d1e84afbd3f7d6820020", + "actions/upload-artifact": "ea165f8d65b6e75b540449e92b4886f43607fa02", +} +EXPECTED_ARTIFACT_FILES = {"source-revisions.json", "verification-summary.json"} +PROOF_CEILING = "CONTROLLED_REPO_CONVERGENCE_AND_LOCAL_FIXTURE_REVIEW_ONLY" +EXPECTED_VERIFICATION_CHECKS = [ + "command_center_invariants", + "command_center_hostile_workflow_tests", + "exact_seven_source_checkout", + "detection_contract", + "detection_promotion_matrix", + "detection_reverse_inventory_and_hostile_tests", + "validation_registry", + "validation_package_sweep", + "validation_source_and_report_parity", + "validation_claim_boundary", + "proof_status_index", + "proof_reverse_inventory", + "proof_integrity", + "platform_public_status_source_contract", + "platform_case_growth_convergence", + "platform_mutation_boundary", + "hoxline_case_growth_pair", + "hoxline_expanded_batch", + "hoxline_replay", + "hoxline_hostile_tests", + "website_source_owner_and_freshness", + "website_nested_claim_and_eol_tests", + "website_static_build", +] +EXPECTED_RUN_SHA256_BY_STEP = { + "Install structural verifier dependency": "6777f50efc1a4de7a52454974ba0da5a7adda9e7a84e251b3f3fe93912fdd695", + "Verify command-center invariants": "7457407dbbf6fc6c710590149da3c3a7be1358b567f31ffda84cb8fa4fcd2e46", + "Run hostile command-center unit tests": "16792c22d70f184d03660b7d7641f13a305e9227f1d31be6950732ca2a80a5d3", + "Verify patch whitespace": "c6cab74e9117e643b9234134c46cfcf6cf1733747c76d31399834e41cee706a4", + "Install bounded verifier dependencies": "4e24c9f627196734440d7af0f88696d5c78bcabf31951f052d6f5b8c0d5913b2", + "Resolve governance/CONVERGENCE_SOURCE_MANIFEST.json": "2cc7ec88e5f15e3ce2005c2a7d69d9612b88cd4832d3b6f7ebfc900d326530e8", + "Checkout six immutable sibling revisions without credentials": "457e61f1280506ee49cce8d2c796031a25b7874bb3cffeb76441f788cfdd1942", + "Verify the exact clean detached source set": "3dbfdd7ea15772914b827f395b09fe23aa61c0d31e91703b05cc3ef8c4476e57", + "Detect durable sibling main-content drift": "fa27754b70cd171cac072a868b3632405e5b2c5b8744e34139cf8e64dbcd7a53", + "Verify detection authority and hostile paths": "a55bb68d511268423e7ed392184dab55f3864c8411d74ff725c54f776daa4d4d", + "Verify validation authority and fail-closed parity": "d4b3bf07e7ae50adb8ef1d385be77ad61a11f339f724fc061eeb992a022346c0", + "Verify proof authority and reverse inventory": "f07b841030269d74bd563238e59e4ff695e2d140108321f9d33ea364a0d850b8", + "Verify platform source contract and seven-source convergence": "698d9e4b0035d4581a87c889bff1c7bb7ef53db957d75688f1364a713be76ee5", + "Install Hoxline from the checked immutable source": "f6954cbb94cbc30f5110c536a4c73b71f985953e71e4c08385cf46b7eb8fea0c", + "Verify Hoxline Case Growth pair and replay integrity": "979ca538d7eb872fa2da00afbf8e73606ab874d94cc134475d12e2b86b36886b", + "Install Website dependencies from the checked lockfile": "4be0617fbf64515a837109e98174e687edc18a027a37368223fff95cb33d8f94", + "Verify Website rendering-only status plane and static build": "4430722c68c09465e87a12d612e0d721ef0f7b9494b1e77a8b0d8ba3dc2649c8", + "Write closed-schema verification summary": "056a198f6f178d60e01d0a0eecda11181372ced82d68793b24fc0676a8cb15b7", + "Validate upload artifacts": "e0b9d66521ae1e69ce51a89d1380fcb70f6edbaa8c9a62a7a78aa1c6a4e6ac0f", +} +EXPECTED_ACTION_BY_STEP = { + "Checkout command-center authority": { + "uses": f"actions/checkout@{PINNED_ACTIONS['actions/checkout']}", + "with": {"persist-credentials": "false", "fetch-depth": 0}, + }, + "Checkout workflow authority at the event revision": { + "uses": f"actions/checkout@{PINNED_ACTIONS['actions/checkout']}", + "with": { + "ref": "${{ github.event.pull_request.head.sha || github.sha }}", + "path": "source-set/.github", + "fetch-depth": 0, + "persist-credentials": "false", + }, + }, + "Set up Python": { + "uses": f"actions/setup-python@{PINNED_ACTIONS['actions/setup-python']}", + "with": {"python-version": "3.12"}, + }, + "Set up Node": { + "uses": f"actions/setup-node@{PINNED_ACTIONS['actions/setup-node']}", + "with": {"node-version": "20"}, + }, + "Upload sanitized convergence records": { + "uses": f"actions/upload-artifact@{PINNED_ACTIONS['actions/upload-artifact']}", + "with": { + "name": "seven-repository-convergence-${{ github.run_id }}", + "path": ( + "verification-artifacts/source-revisions.json\n" + "verification-artifacts/verification-summary.json\n" + ), + "if-no-files-found": "error", + "retention-days": 14, + }, + }, +} +EXPECTED_BASH_STEPS = { + "Verify patch whitespace", + "Resolve governance/CONVERGENCE_SOURCE_MANIFEST.json", + "Checkout six immutable sibling revisions without credentials", + "Verify the exact clean detached source set", + "Verify detection authority and hostile paths", + "Verify validation authority and fail-closed parity", + "Verify proof authority and reverse inventory", + "Verify platform source contract and seven-source convergence", + "Verify Hoxline Case Growth pair and replay integrity", + "Verify Website rendering-only status plane and static build", +} +EXPECTED_MANIFEST_ROOT_KEYS = { + "schema", + "scope", + "required_route_files", + "cross_repo_repositories", + "invariants", +} +EXPECTED_INVARIANT_KEYS = { + "github_repo_role", + "project_2_role", + "project_1_boundary", + "project_metadata_boundary", + "rendering_boundary", + "proof_authority_repo", + "command_center_proof_ceiling", + "ledger_public_safe_status", + "reviewer_metrics_pipeline", + "reviewer_metrics_counts", + "cross_repo_convergence", + "ho_det_001_public_ceiling", + "runtime_signal_public_promotions", + "standing_controls", + "standing_control_replacement", +} REQUIRED_TEXT = { "README.md": [ @@ -71,7 +241,6 @@ "analyst-approved", "live Splunk", ] - BOUNDARY_WORDS = ( "blocked", "blocked_claim", @@ -105,6 +274,10 @@ ) +class ValidationError(ValueError): + """Raised when a machine-readable control fails closed.""" + + def fail(message: str, errors: list[str]) -> None: errors.append(message) @@ -113,7 +286,11 @@ def read_text(path: Path, errors: list[str]) -> str: if not path.exists(): fail(f"missing file: {path.relative_to(ROOT).as_posix()}", errors) return "" - return path.read_text(encoding="utf-8") + try: + return path.read_text(encoding="utf-8") + except (OSError, UnicodeError) as exc: + fail(f"cannot read {path.relative_to(ROOT).as_posix()}: {exc}", errors) + return "" def iter_text_files() -> list[Path]: @@ -124,43 +301,1284 @@ def iter_text_files() -> list[Path]: files.append(path) elif path.is_dir(): files.extend( - p - for p in path.rglob("*") - if p.is_file() and p.suffix.lower() in {".md", ".json", ".yml", ".yaml"} + candidate + for candidate in path.rglob("*") + if candidate.is_file() + and candidate.suffix.lower() in {".md", ".json", ".yml", ".yaml"} ) return sorted(set(files)) -def load_manifest(errors: list[str]) -> dict: - text = read_text(MANIFEST_PATH, errors) - if not text: - return {} +def normalize_vocabulary_security_text(value: str) -> str: + """Collapse Unicode token-splitting characters for security matching only.""" + normalized = unicodedata.normalize("NFKD", value) + return "".join( + character + for character in normalized + if not unicodedata.category(character).startswith(("C", "M")) + ) + + +def tracked_vocabulary_findings(repo_root: Path = ROOT) -> list[str]: + retired = "".join(("syn", "thetic")) + binary_extensions = frozenset( + { + ".7z", ".avif", ".avi", ".bz2", ".dll", ".dylib", ".eot", ".exe", ".gif", + ".gz", ".ico", ".jpeg", ".jpg", ".mov", ".mp3", ".mp4", ".pdf", + ".png", ".pyc", ".so", ".tar", ".tgz", ".ttf", ".wasm", ".webp", + ".woff", ".woff2", ".xz", ".zip", + } + ) + findings: list[str] = [] + listed = subprocess.run( + ["git", "-C", str(repo_root), "ls-files", "-z"], + capture_output=True, + check=False, + env=sanitized_git_environment(), + ) + if listed.returncode != 0: + return ["tracked-source vocabulary check could not enumerate Git-tracked files"] try: - manifest = json.loads(text) - except json.JSONDecodeError as exc: - fail(f"manifest JSON parse failed: {exc}", errors) + tracked_paths = listed.stdout.decode("utf-8").split("\0") + except UnicodeDecodeError: + return ["tracked-source vocabulary filename inventory is not valid UTF-8"] + for relative in filter(None, tracked_paths): + if retired in normalize_vocabulary_security_text(relative).casefold(): + findings.append( + f"retired fixture vocabulary appears in tracked filename: {relative}" + ) + if PurePosixPath(relative).suffix.casefold() in binary_extensions: + continue + scanned = subprocess.run( + ["git", "-C", str(repo_root), "show", f":{relative}"], + capture_output=True, + check=False, + env=sanitized_git_environment(), + ) + if scanned.returncode != 0: + findings.append( + f"tracked-source vocabulary check could not read indexed content: {relative}" + ) + continue + if b"\0" in scanned.stdout: + findings.append( + f"tracked non-binary content contains NUL: {relative}" + ) + continue + try: + text = scanned.stdout.decode("utf-8") + except UnicodeDecodeError: + findings.append( + f"tracked non-binary content is not UTF-8: {relative}" + ) + continue + if retired in normalize_vocabulary_security_text(text).casefold(): + findings.append( + f"retired fixture vocabulary appears in tracked content: {relative}" + ) + return findings + + +def reject_duplicate_object_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + normalized: set[str] = set() + for key, value in pairs: + if not isinstance(key, str): + raise ValidationError("JSON object keys must be strings") + folded = key.casefold() + if folded in normalized: + raise ValidationError(f"duplicate JSON key: {key}") + normalized.add(folded) + result[key] = value + return result + + +def load_json_strict(path: Path) -> dict[str, Any]: + try: + value = json.loads( + path.read_text(encoding="utf-8"), + object_pairs_hook=reject_duplicate_object_pairs, + ) + except (OSError, UnicodeError, json.JSONDecodeError, ValidationError) as exc: + raise ValidationError(f"{path.name}: invalid JSON: {exc}") from exc + if not isinstance(value, dict): + raise ValidationError(f"{path.name}: top-level value must be an object") + return value + + +def load_yaml_strict(text: str) -> dict[str, Any]: + if yaml is None: + raise ValidationError("PyYAML is required for structural workflow validation") + + class UniqueKeyLoader(yaml.SafeLoader): + pass + + # GitHub uses YAML 1.2 semantics for the ``on`` key. PyYAML's legacy 1.1 + # boolean resolver would otherwise turn it into True. + for initial, resolvers in list(UniqueKeyLoader.yaml_implicit_resolvers.items()): + UniqueKeyLoader.yaml_implicit_resolvers[initial] = [ + resolver + for resolver in resolvers + if resolver[0] != "tag:yaml.org,2002:bool" + ] + + def construct_mapping( + loader: UniqueKeyLoader, node: Any, deep: bool = False + ) -> dict[str, Any]: + pairs = loader.construct_pairs(node, deep=deep) + result: dict[str, Any] = {} + normalized: set[str] = set() + for key, value in pairs: + if not isinstance(key, str): + raise ValidationError("workflow mapping keys must be strings") + folded = key.casefold() + if folded in normalized: + raise ValidationError(f"duplicate workflow key: {key}") + normalized.add(folded) + result[key] = value + return result + + UniqueKeyLoader.add_constructor( + yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_mapping + ) + try: + value = yaml.load(text, Loader=UniqueKeyLoader) + except ValidationError: + raise + except yaml.YAMLError as exc: + raise ValidationError(f"workflow YAML parse failed: {exc}") from exc + if not isinstance(value, dict): + raise ValidationError("workflow top-level value must be an object") + return value + + +def load_manifest(errors: list[str]) -> dict[str, Any]: + try: + manifest = load_json_strict(MANIFEST_PATH) + except ValidationError as exc: + fail(str(exc), errors) return {} if manifest.get("schema") != "hawkinsoperations-command-center-invariants-v1": fail("manifest schema mismatch", errors) - if not isinstance(manifest.get("invariants"), dict): + if set(manifest) != EXPECTED_MANIFEST_ROOT_KEYS: + fail("manifest root shape is not closed", errors) + invariants = manifest.get("invariants") + if not isinstance(invariants, dict): fail("manifest invariants must be an object", errors) + elif set(invariants) != EXPECTED_INVARIANT_KEYS: + fail("manifest invariant shape is not closed", errors) + elif any(not isinstance(value, str) or not value.strip() for value in invariants.values()): + fail("manifest invariant values must be non-empty strings", errors) + if not isinstance(manifest.get("scope"), str) or not manifest["scope"].strip(): + fail("manifest scope must be a non-empty string", errors) + if not isinstance(manifest.get("required_route_files"), list): + fail("manifest required_route_files must be an array", errors) + if manifest.get("cross_repo_repositories") != EXACT_REPOSITORIES: + fail("manifest cross-repository list is not canonical", errors) return manifest -def check_required_files(manifest: dict, errors: list[str]) -> None: +def validate_source_manifest(value: dict[str, Any]) -> list[str]: + errors: list[str] = [] + allowed_root = {"schema", "manifest_id", "repositories", "constraints"} + if set(value) != allowed_root: + errors.append( + f"source manifest root keys must be exactly {sorted(allowed_root)}" + ) + if value.get("schema") != "hawkinsoperations-convergence-source-manifest-v1": + errors.append("source manifest schema mismatch") + if value.get("manifest_id") != "HAWKINSOPERATIONS_SEVEN_SOURCE_PR_HEAD_MATRIX_V1": + errors.append("source manifest ID mismatch") + entries = value.get("repositories") + if not isinstance(entries, list): + return [*errors, "source manifest repositories must be an ordered list"] + if len(entries) != 7: + errors.append("source manifest must contain exactly seven entries") + seen: set[str] = set() + observed: list[str] = [] + for index, entry in enumerate(entries): + if not isinstance(entry, dict): + errors.append(f"source manifest entry {index} must be an object") + continue + repository = entry.get("repository") + if not isinstance(repository, str): + errors.append(f"source manifest entry {index} repository must be a string") + continue + folded = repository.casefold() + if folded in seen: + errors.append(f"source manifest repository duplicated: {repository}") + seen.add(folded) + observed.append(repository) + expected_full = f"HawkinsOperations/{repository}" + if entry.get("canonical_repository") != expected_full: + errors.append(f"source manifest canonical owner mismatch: {repository}") + if repository == ".github": + if set(entry) != { + "repository", + "canonical_repository", + "revision_source", + "authority_content_revision", + "tree_source", + }: + errors.append(".github source entry has an unsupported shape") + if entry.get("revision_source") != "github_event_sha": + errors.append(".github source entry must use github_event_sha") + if entry.get("tree_source") != "github_event_tree": + errors.append(".github source entry must use github_event_tree") + if re.fullmatch( + r"[0-9a-f]{40}", + str(entry.get("authority_content_revision", "")), + ) is None: + errors.append( + ".github source entry authority content revision is not immutable" + ) + else: + if set(entry) != { + "repository", + "canonical_repository", + "revision", + "authority_content_revision", + "reviewed_tree_sha", + }: + errors.append(f"source manifest entry has an unsupported shape: {repository}") + if re.fullmatch(r"[0-9a-f]{40}", str(entry.get("revision", ""))) is None: + errors.append(f"source manifest revision is not immutable: {repository}") + if re.fullmatch( + r"[0-9a-f]{40}", + str(entry.get("authority_content_revision", "")), + ) is None: + errors.append( + f"source manifest authority content revision is not immutable: {repository}" + ) + if re.fullmatch( + r"[0-9a-f]{40}", str(entry.get("reviewed_tree_sha", "")) + ) is None: + errors.append(f"source manifest reviewed tree is not immutable: {repository}") + if observed != EXACT_REPOSITORIES: + errors.append("source manifest repositories must equal the exact canonical order") + constraints = value.get("constraints") + expected_constraints = { + "exact_repository_count": 7, + "read_only": True, + "default_branch_fallback": False, + "require_detached_exact_revision": True, + "record_checked_revisions": True, + "consumer_outputs_are_not_authority": True, + "proof_ceiling": PROOF_CEILING, + } + if constraints != expected_constraints: + errors.append("source manifest constraints do not match the fail-closed contract") + return errors + + +def load_source_manifest(errors: list[str]) -> dict[str, Any]: + try: + manifest = load_json_strict(SOURCE_MANIFEST_PATH) + except ValidationError as exc: + fail(str(exc), errors) + return {} + for error in validate_source_manifest(manifest): + fail(error, errors) + return manifest + + +def check_required_files(manifest: dict[str, Any], errors: list[str]) -> None: required = manifest.get("required_route_files", []) if not isinstance(required, list) or not required: fail("manifest required_route_files must be a non-empty list", errors) return + if "governance/CONVERGENCE_SOURCE_MANIFEST.json" not in required: + fail("command-center manifest must require the convergence source manifest", errors) for item in required: - rel = Path(str(item)) - if rel.is_absolute() or ".." in rel.parts: + rel = PurePosixPath(str(item)) + if rel.is_absolute() or ".." in rel.parts or "\\" in str(item): fail(f"invalid required route path: {item}", errors) continue - if not (ROOT / rel).is_file(): + if not (ROOT / Path(*rel.parts)).is_file(): fail(f"missing required route file: {item}", errors) +def walk(value: Any, path: tuple[str, ...] = ()): + yield path, value + if isinstance(value, dict): + for key, nested in value.items(): + yield from walk(nested, (*path, str(key))) + elif isinstance(value, list): + for index, nested in enumerate(value): + yield from walk(nested, (*path, str(index))) + + +def scalar_is_false(value: Any) -> bool: + return value is False or (isinstance(value, str) and value.casefold() == "false") + + +def unsafe_workflow_findings(text: str) -> list[str]: + findings: list[str] = [] + try: + workflow = load_yaml_strict(text) + except ValidationError as exc: + return [str(exc)] + + allowed_root_keys = {"name", "on", "permissions", "jobs"} + if set(workflow) != allowed_root_keys: + findings.append("workflow root shape is not closed") + triggers = workflow.get("on") + if not isinstance(triggers, dict): + findings.append("workflow trigger declaration must be an object") + else: + allowed_triggers = {"pull_request", "push", "workflow_dispatch", "schedule"} + if set(triggers) != allowed_triggers: + findings.append("workflow triggers differ from the approved read-only set") + if "pull_request_target" in triggers: + findings.append("pull_request_target is forbidden") + schedule = triggers.get("schedule") + if not isinstance(schedule, list) or not schedule: + findings.append("scheduled drift detection is required") + if not isinstance(triggers.get("workflow_dispatch"), dict): + findings.append("manual read-only dispatch is required") + pull_request = triggers.get("pull_request") + push = triggers.get("push") + if pull_request != {}: + findings.append( + "pull_request trigger must be unrestricted so every tracked path is scanned" + ) + if not isinstance(push, dict) or set(push) != {"branches"}: + findings.append("push trigger shape must contain only branches") + else: + if push.get("branches") != ["main"]: + findings.append("push trigger must govern main exactly") + + if workflow.get("permissions") != {"contents": "read"}: + findings.append("root permissions must be exactly contents: read") + jobs = workflow.get("jobs") + if not isinstance(jobs, dict) or set(jobs) != { + "command-center-invariants", + "seven-repository-convergence", + }: + findings.append("workflow jobs must be the exact approved pair") + else: + expected_step_names = { + "command-center-invariants": [ + "Checkout command-center authority", + "Set up Python", + "Install structural verifier dependency", + "Verify command-center invariants", + "Run hostile command-center unit tests", + "Verify patch whitespace", + ], + "seven-repository-convergence": [ + "Checkout workflow authority at the event revision", + "Set up Python", + "Set up Node", + "Install bounded verifier dependencies", + "Resolve governance/CONVERGENCE_SOURCE_MANIFEST.json", + "Checkout six immutable sibling revisions without credentials", + "Verify the exact clean detached source set", + "Detect durable sibling main-content drift", + "Verify detection authority and hostile paths", + "Verify validation authority and fail-closed parity", + "Verify proof authority and reverse inventory", + "Verify platform source contract and seven-source convergence", + "Install Hoxline from the checked immutable source", + "Verify Hoxline Case Growth pair and replay integrity", + "Install Website dependencies from the checked lockfile", + "Verify Website rendering-only status plane and static build", + "Write closed-schema verification summary", + "Validate upload artifacts", + "Upload sanitized convergence records", + ], + } + for job_name, expected_names in expected_step_names.items(): + job = jobs.get(job_name) + if not isinstance(job, dict): + findings.append(f"{job_name} job must be an object") + continue + expected_job_keys = ( + {"needs", "runs-on", "env", "steps"} + if job_name == "seven-repository-convergence" + else {"runs-on", "steps"} + ) + if set(job) != expected_job_keys: + findings.append(f"{job_name} job shape is not closed") + if job.get("runs-on") != "ubuntu-latest": + findings.append(f"{job_name} runner must be ubuntu-latest") + if ( + job_name == "seven-repository-convergence" + and job.get("needs") != "command-center-invariants" + ): + findings.append( + "seven-repository convergence must depend on command-center invariants" + ) + if job_name == "seven-repository-convergence" and job.get("env") != { + "PYTHONDONTWRITEBYTECODE": "1" + }: + findings.append("seven-repository job environment is not exact") + if "if" in job: + findings.append(f"{job_name} mandatory job must not be conditional") + steps = job.get("steps") + if not isinstance(steps, list): + findings.append(f"{job_name} steps must be an array") + continue + names = [step.get("name") if isinstance(step, dict) else None for step in steps] + if names != expected_names: + findings.append(f"{job_name} step order differs from the approved contract") + for step in steps: + if not isinstance(step, dict): + findings.append(f"{job_name} contains a non-object step") + continue + name = step.get("name") + if not isinstance(name, str): + findings.append(f"{job_name} step name must be a string") + continue + if "run" in step: + expected_keys = {"name", "run"} + if name in EXPECTED_BASH_STEPS: + expected_keys.add("shell") + if name == "Detect durable sibling main-content drift": + expected_keys.add("if") + if set(step) != expected_keys: + findings.append(f"run step shape is not closed: {name}") + run = step.get("run") + if not isinstance(run, str): + findings.append(f"run step command must be a string: {name}") + else: + expected_digest = EXPECTED_RUN_SHA256_BY_STEP.get(name) + actual_digest = hashlib.sha256(run.encode("utf-8")).hexdigest() + if expected_digest is None or actual_digest != expected_digest: + findings.append( + f"run step differs from exact command allowlist: {name}" + ) + if name in EXPECTED_BASH_STEPS and step.get("shell") != "bash": + findings.append(f"multiline step shell must be exactly bash: {name}") + elif name not in EXPECTED_BASH_STEPS and "shell" in step: + findings.append(f"shell override is forbidden: {name}") + elif "uses" in step: + if set(step) != {"name", "uses", "with"}: + findings.append(f"action step shape is not closed: {name}") + expected_action = EXPECTED_ACTION_BY_STEP.get(name) + if expected_action is None or { + "uses": step.get("uses"), + "with": step.get("with"), + } != expected_action: + findings.append(f"action step differs from exact allowlist: {name}") + else: + findings.append(f"step must use one exact action or run block: {name}") + condition = step.get("if") + if name == "Detect durable sibling main-content drift": + if condition != "github.event_name != 'pull_request'": + findings.append("durable main observation condition is not exact") + elif condition is not None: + findings.append(f"mandatory step is conditional: {name}") + convergence_steps = jobs["seven-repository-convergence"].get("steps", []) + if isinstance(convergence_steps, list): + names = [step.get("name") for step in convergence_steps if isinstance(step, dict)] + try: + validate_index = names.index("Validate upload artifacts") + upload_index = names.index("Upload sanitized convergence records") + if upload_index != validate_index + 1: + findings.append("artifact validation must be immediately before upload") + except ValueError: + findings.append("artifact validation/upload steps are missing") + + checkout_count = 0 + source_set_checkout = False + upload_count = 0 + for path, value in walk(workflow): + key = path[-1].casefold() if path else "" + if key == "continue-on-error": + findings.append("continue-on-error is forbidden") + if key == "defaults": + findings.append("workflow and job defaults are forbidden") + if key == "permissions": + if path != ("permissions",): + findings.append("job or step permission override is forbidden") + if key in {"contents", "actions", "checks", "issues", "packages", "pages", + "pull-requests", "security-events", "statuses", "id-token"}: + if isinstance(value, str) and value.casefold() == "write": + findings.append(f"write permission is forbidden at {'/'.join(path)}") + if key == "if" and isinstance(value, str): + if re.search(r"\balways\s*\(\s*\)", value, re.IGNORECASE): + findings.append("always() is forbidden because it can neutralize failure ordering") + if key == "run" and isinstance(value, str) and "\n" in value: + if "set -euo pipefail" not in value: + findings.append("multiline shell steps must enable strict exit propagation") + if key == "run" and isinstance(value, str): + if re.search(r"(?im)^\s*(?:echo|printf)\b.*\b(?:python|git)\b", value): + findings.append("required command may not be replaced by inert output") + if key == "uses" and isinstance(value, str): + match = re.fullmatch(r"([^@]+)@([0-9a-f]{40})", value) + if match is None: + findings.append(f"action must be pinned to an immutable SHA: {value}") + elif PINNED_ACTIONS.get(match.group(1)) != match.group(2): + findings.append(f"action SHA is not allowlisted: {value}") + if value.startswith("actions/checkout@"): + checkout_count += 1 + step_path = path[:-1] + step: Any = workflow + for component in step_path: + step = step[int(component)] if isinstance(step, list) else step[component] + checkout_with = step.get("with") if isinstance(step, dict) else None + if not isinstance(checkout_with, dict) or not scalar_is_false( + checkout_with.get("persist-credentials") + ): + findings.append("checkout must set persist-credentials: false") + if isinstance(checkout_with, dict) and checkout_with.get("path") == "source-set/.github": + source_set_checkout = True + expected_ref = "${{ github.event.pull_request.head.sha || github.sha }}" + if checkout_with.get("ref") != expected_ref: + findings.append( + "source-set .github checkout must use the immutable event SHA" + ) + if value.startswith("actions/upload-artifact@"): + upload_count += 1 + step_path = path[:-1] + step = workflow + for component in step_path: + step = step[int(component)] if isinstance(step, list) else step[component] + upload_with = step.get("with") if isinstance(step, dict) else None + expected_paths = ( + "verification-artifacts/source-revisions.json\n" + "verification-artifacts/verification-summary.json\n" + ) + if ( + not isinstance(upload_with, dict) + or upload_with.get("path") != expected_paths + or upload_with.get("if-no-files-found") != "error" + ): + findings.append( + "artifact upload must use the exact sanitized two-file allowlist" + ) + + if checkout_count != 2: + findings.append( + "workflow must perform exactly one credential-bounded checkout in each job" + ) + if not source_set_checkout: + findings.append("seven-source job must checkout .github under source-set/.github") + if upload_count != 1: + findings.append("workflow must contain exactly one sanitized artifact upload") + + forbidden_text_patterns = { + "pull_request_target": r"(?m)^\s*pull_request_target\s*:", + "direct push": r"\bgit\s+push\b", + "remote mutation": r"\bgit\s+(?:commit|tag)\b", + "PR mutation": r"\b(?:gh\s+pr\s+(?:create|merge|ready|review)|gh\s+api[^\n]*(?:POST|PATCH|PUT|DELETE))\b", + "HTTP PR mutation": r"\bcurl\b[^\n]*(?:-X|--request)\s*(?:POST|PATCH|PUT|DELETE)[^\n]*(?:api\.github\.com|/pulls\b)", + "auto-merge": r"\bauto-merge\b", + "ledger mutation": r"\b(?:lifetime|ledger)[^\n]*(?:append|correct|mutate|write)\b", + "runtime mutation": r"\b(?:runtime|endpoint|wazuh|splunk|cribl)[^\n]*(?:mutate|deploy|configure|restart|write)\b", + "proof promotion": r"\b(?:proof|public.safe)[^\n]*(?:promote|publish|approve)\b", + "swallowed failure": ( + r"(?:\|\|\s*(?::(?:\s|$)|true\b|echo\b|printf\b|exit\s+0\b|" + r"\{[^\n}]*\bexit\s+0\b)|;\s*(?:true\b|exit\s+0\b)|" + r"\bset\s+\+e\b|\btrap\b[^\n]*\bexit\s+0\b)" + ), + "no-op command prefix": r"(?m)^\s*:\s+(?:python|git|npm|npx)\b", + "command function override": ( + r"(?m)^\s*(?:python|python3|git|npm|npx)\s*\(\s*\)\s*\{" + ), + "command alias override": ( + r"(?m)^\s*alias\s+(?:python|python3|git|npm|npx)\s*=" + ), + "command path shadowing": ( + r"(?m)^\s*(?:PATH\s*=|export\s+PATH\s*=|" + r"(?:function\s+)?(?:python|python3|git|npm|npx)\s*=)" + ), + "backgrounded command": r"(?m)(? None: + workflow_text = read_text(WORKFLOW_PATH, errors) + declared = manifest.get("cross_repo_repositories") + if declared != EXACT_REPOSITORIES: + fail( + "manifest cross_repo_repositories must list the exact seven repositories in canonical order", + errors, + ) + if [entry.get("repository") for entry in source_manifest.get("repositories", [])] != EXACT_REPOSITORIES: + fail("source manifest and invariant repository order disagree", errors) + for finding in unsafe_workflow_findings(workflow_text): + fail(f"cross-repo workflow permits unsafe behavior: {finding}", errors) + + +def check_workflow_hostile_self_test(errors: list[str]) -> None: + base = WORKFLOW_PATH.read_text(encoding="utf-8") + hostile_cases = { + "write permission": base.replace("contents: read", "issues: write", 1), + "pull_request_target": base.replace("pull_request:", "pull_request_target:", 1), + "continue-on-error": base.replace( + "run: python scripts/verify-command-center-invariants.py --self-test", + "continue-on-error: true\n run: python scripts/verify-command-center-invariants.py --self-test", + 1, + ), + "persisted credentials": base.replace( + "persist-credentials: false", "persist-credentials: true", 1 + ), + "mutable action": base.replace( + f"actions/checkout@{PINNED_ACTIONS['actions/checkout']}", + "actions/checkout@v4", + 1, + ), + "swallowed failure": base.replace("set -euo pipefail", "set -euo pipefail\n false || true", 1), + "direct push": base.replace("set -euo pipefail", "set -euo pipefail\n git push origin main", 1), + "always step": base.replace( + "- name: Validate upload artifacts", + "- name: Validate upload artifacts\n if: always()", + 1, + ), + } + for name, hostile in hostile_cases.items(): + if not unsafe_workflow_findings(hostile): + fail(f"workflow hostile self-test accepted {name}", errors) + + try: + source = load_json_strict(SOURCE_MANIFEST_PATH) + except ValidationError as exc: + fail(f"source-manifest hostile self-test precondition failed: {exc}", errors) + return + for name, mutate in { + "missing repository": lambda value: value["repositories"].pop(), + "duplicate repository": lambda value: value["repositories"].append( + dict(value["repositories"][0]) + ), + "mutable revision": lambda value: value["repositories"][1].update( + {"revision": "main"} + ), + "owner spoof": lambda value: value["repositories"][1].update( + {"canonical_repository": "NotHawkinsOperations/hawkinsoperations-detections"} + ), + "fallback enabled": lambda value: value["constraints"].update( + {"default_branch_fallback": True} + ), + }.items(): + candidate = json.loads(json.dumps(source)) + mutate(candidate) + if not validate_source_manifest(candidate): + fail(f"source-manifest hostile self-test accepted {name}", errors) + + +def canonical_origin(value: str) -> str: + normalized = value.strip().rstrip("/").casefold() + if normalized.startswith("git@github.com:"): + normalized = "https://github.com/" + normalized.removeprefix("git@github.com:") + elif normalized.startswith("ssh://git@github.com/"): + normalized = "https://github.com/" + normalized.removeprefix( + "ssh://git@github.com/" + ) + if not normalized.endswith(".git"): + normalized += ".git" + return normalized + + +def stored_origin(repo: Path) -> str: + result = subprocess.run( + [ + "git", + "-C", + str(repo), + "config", + "--local", + "--get-all", + "remote.origin.url", + ], + check=False, + capture_output=True, + text=True, + env=sanitized_git_environment(), + ) + values = [value.strip() for value in result.stdout.splitlines()] + if result.returncode != 0 or len(values) != 1 or not values[0]: + raise ValidationError( + f"{repo.name}: stored origin must contain exactly one nonempty local URL" + ) + return values[0] + + +def git(repo: Path, *args: str) -> str: + result = subprocess.run( + ["git", "-C", str(repo), *args], + check=False, + capture_output=True, + text=True, + env=sanitized_git_environment(), + ) + if result.returncode: + raise ValidationError( + f"{repo.name}: git {' '.join(args)} failed: {result.stderr.strip()}" + ) + return result.stdout.strip() + + +def resolved_source_manifest( + manifest: dict[str, Any], event_sha: str, event_tree_sha: str | None = None +) -> dict[str, Any]: + if re.fullmatch(r"[0-9a-f]{40}", event_sha) is None: + raise ValidationError("event SHA must be a lowercase 40-character Git SHA") + if validate_source_manifest(manifest): + raise ValidationError("cannot resolve an invalid source manifest") + if event_tree_sha is None: + event_tree_sha = git(ROOT, "rev-parse", f"{event_sha}^{{tree}}") + if re.fullmatch(r"[0-9a-f]{40}", event_tree_sha) is None: + raise ValidationError("event tree SHA must be a lowercase 40-character Git SHA") + entries = [] + for entry in manifest["repositories"]: + revision = event_sha if entry["repository"] == ".github" else entry["revision"] + reviewed_tree = ( + event_tree_sha + if entry["repository"] == ".github" + else entry["reviewed_tree_sha"] + ) + entries.append( + { + "repository": entry["repository"], + "canonical_repository": entry["canonical_repository"], + "revision": revision, + "authority_content_revision": entry["authority_content_revision"], + "reviewed_tree_sha": reviewed_tree, + } + ) + payload = { + "schema": "hawkinsoperations-resolved-convergence-source-set-v1", + "manifest_id": manifest["manifest_id"], + "repositories": entries, + "constraints": manifest["constraints"], + } + payload["manifest_sha256"] = hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + return payload + + +def validate_resolved_manifest(value: dict[str, Any]) -> list[str]: + errors: list[str] = [] + allowed = { + "schema", + "manifest_id", + "repositories", + "constraints", + "manifest_sha256", + } + if set(value) != allowed: + errors.append("resolved source manifest has unsupported fields") + return errors + if value.get("schema") != "hawkinsoperations-resolved-convergence-source-set-v1": + errors.append("resolved source manifest schema mismatch") + repositories = value.get("repositories") + if not isinstance(repositories, list) or len(repositories) != 7: + errors.append("resolved source manifest must contain exactly seven entries") + return errors + observed: list[str] = [] + for entry in repositories: + if not isinstance(entry, dict) or set(entry) != { + "repository", + "canonical_repository", + "revision", + "authority_content_revision", + "reviewed_tree_sha", + }: + errors.append("resolved source entry has unsupported shape") + continue + repository = entry.get("repository") + observed.append(str(repository)) + if entry.get("canonical_repository") != f"HawkinsOperations/{repository}": + errors.append(f"resolved source owner mismatch: {repository}") + if re.fullmatch(r"[0-9a-f]{40}", str(entry.get("revision", ""))) is None: + errors.append(f"resolved source revision invalid: {repository}") + if re.fullmatch( + r"[0-9a-f]{40}", str(entry.get("authority_content_revision", "")) + ) is None: + errors.append(f"resolved authority content revision invalid: {repository}") + if re.fullmatch( + r"[0-9a-f]{40}", str(entry.get("reviewed_tree_sha", "")) + ) is None: + errors.append(f"resolved source reviewed tree invalid: {repository}") + if observed != EXACT_REPOSITORIES or len(set(observed)) != 7: + errors.append("resolved source repositories differ from exact canonical set") + if value.get("constraints") != { + "exact_repository_count": 7, + "read_only": True, + "default_branch_fallback": False, + "require_detached_exact_revision": True, + "record_checked_revisions": True, + "consumer_outputs_are_not_authority": True, + "proof_ceiling": PROOF_CEILING, + }: + errors.append("resolved source constraints mismatch") + unsigned = {key: nested for key, nested in value.items() if key != "manifest_sha256"} + expected = hashlib.sha256( + json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + if value.get("manifest_sha256") != expected: + errors.append("resolved source manifest digest mismatch") + return errors + + +def verify_source_set( + source_set: Path, resolved: dict[str, Any] +) -> tuple[list[dict[str, Any]], list[str]]: + errors = validate_resolved_manifest(resolved) + if errors: + return [], errors + if not source_set.is_dir() or source_set.is_symlink(): + return [], ["source-set root must be a real directory"] + actual_names = sorted( + path.name + for path in source_set.iterdir() + if path.is_dir() and not path.is_symlink() + ) + if actual_names != sorted(EXACT_REPOSITORIES): + errors.append("source-set directory inventory must equal exactly seven repositories") + return [], errors + records: list[dict[str, Any]] = [] + for entry in resolved["repositories"]: + repository = entry["repository"] + repo_path = source_set / repository + try: + if repo_path.resolve().parent != source_set.resolve(): + raise ValidationError(f"{repository}: repository path escapes source-set root") + head = git(repo_path, "rev-parse", "HEAD") + if head != entry["revision"]: + raise ValidationError( + f"{repository}: checked HEAD {head} differs from manifest {entry['revision']}" + ) + tree = git(repo_path, "rev-parse", "HEAD^{tree}") + if tree != entry["reviewed_tree_sha"]: + raise ValidationError( + f"{repository}: checked tree {tree} differs from reviewed content " + f"{entry['reviewed_tree_sha']}" + ) + branch = git(repo_path, "rev-parse", "--abbrev-ref", "HEAD") + if branch != "HEAD": + raise ValidationError(f"{repository}: checkout must be detached at exact revision") + origin = stored_origin(repo_path) + if canonical_origin(origin) != canonical_origin(CANONICAL_ORIGINS[repository]): + raise ValidationError(f"{repository}: canonical origin mismatch") + status = git(repo_path, "status", "--porcelain=v1", "--untracked-files=all") + if status: + raise ValidationError(f"{repository}: source checkout is dirty") + authority_path = CANONICAL_AUTHORITY_PATHS[repository] + content_revision = entry["authority_content_revision"] + try: + git(repo_path, "cat-file", "-e", f"{content_revision}^{{commit}}") + except ValidationError as exc: + raise ValidationError( + f"{repository}: authority content revision is not reachable" + ) from exc + try: + current_blob = git(repo_path, "rev-parse", f"HEAD:{authority_path}") + content_blob = git( + repo_path, + "rev-parse", + f"{content_revision}:{authority_path}", + ) + except ValidationError as exc: + raise ValidationError( + f"{repository}: canonical authority path is absent at the " + "current or content revision" + ) from exc + if current_blob != content_blob: + raise ValidationError( + f"{repository}: authority content revision does not identify " + f"the current blob at {authority_path}" + ) + records.append( + { + "repository": repository, + "canonical_repository": entry["canonical_repository"], + "checked_sha": head, + "checked_tree_sha": tree, + "authority_path": authority_path, + "authority_content_revision": content_revision, + "authority_git_blob_sha": current_blob, + "detached": True, + "clean": True, + } + ) + except ValidationError as exc: + errors.append(str(exc)) + return records, errors + + +def compare_observed_main_trees( + resolved: dict[str, Any], observed: dict[str, str] +) -> list[str]: + errors = validate_resolved_manifest(resolved) + if errors: + return errors + expected = { + entry["repository"]: entry["reviewed_tree_sha"] + for entry in resolved["repositories"] + if entry["repository"] != ".github" + } + if set(observed) != set(expected): + return ["remote main observations must cover exactly the six sibling repositories"] + for repository, expected_tree in expected.items(): + actual_tree = observed.get(repository) + if actual_tree != expected_tree: + errors.append( + f"{repository}: current main content tree {actual_tree} differs from " + f"reviewed tree {expected_tree}; refresh the reviewed source matrix" + ) + return errors + + +def verify_remote_main_content( + source_set: Path, resolved: dict[str, Any] +) -> list[str]: + errors = validate_resolved_manifest(resolved) + if errors: + return errors + observed: dict[str, str] = {} + for entry in resolved["repositories"]: + repository = entry["repository"] + if repository == ".github": + continue + url = CANONICAL_ORIGINS[repository] + result = subprocess.run( + ["git", "ls-remote", "--exit-code", url, "refs/heads/main"], + check=False, + capture_output=True, + text=True, + env=sanitized_git_environment(), + ) + fields = result.stdout.strip().split() + if result.returncode != 0 or len(fields) != 2 or fields[1] != "refs/heads/main": + errors.append(f"{repository}: current main observation is unavailable") + continue + main_sha = fields[0] + if re.fullmatch(r"[0-9a-f]{40}", main_sha) is None: + errors.append(f"{repository}: current main observation is malformed") + continue + repo_path = source_set / repository + fetch = subprocess.run( + ["git", "-C", str(repo_path), "fetch", "--quiet", "--depth=1", "origin", main_sha], + check=False, + capture_output=True, + text=True, + env=sanitized_git_environment(), + ) + if fetch.returncode: + errors.append(f"{repository}: current main content cannot be fetched") + continue + try: + observed[repository] = git(repo_path, "rev-parse", f"{main_sha}^{{tree}}") + except ValidationError as exc: + errors.append(str(exc)) + if errors: + return errors + return compare_observed_main_trees(resolved, observed) + + +def is_private_scalar(value: str) -> bool: + decoded = value + for _ in range(3): + next_value = unquote(decoded) + if next_value == decoded: + break + decoded = next_value + variants = {value, decoded, decoded.replace("\\", "/")} + for candidate in variants: + lowered = candidate.casefold() + if ( + PureWindowsPath(candidate).is_absolute() + or PurePosixPath(candidate).is_absolute() + or re.match(r"^[a-z]:[^/\\]", lowered) + or lowered.startswith(("\\\\", "//", "file:", "~", "$home", "${home}")) + or "../" in lowered + or "/users/" in lowered + or "/home/" in lowered + or "/raylee/" in lowered + ): + return True + if re.search( + r"(?:github[_-]?pat_|ghp_|begin (?:rsa |openssh )?private key|" + r"\bAKIA[0-9A-Z]{16}\b|\bbearer\s+[a-z0-9._~+/=-]{12,}\b|" + r"\beyJ[a-z0-9_-]{8,}\.[a-z0-9_-]{8,}\.[a-z0-9_-]{8,}\b|" + r"\b(?:10|127)\.\d{1,3}\.\d{1,3}\.\d{1,3}\b|" + r"\b172\.(?:1[6-9]|2\d|3[0-1])\.\d{1,3}\.\d{1,3}\b|" + r"\b192\.168\.\d{1,3}\.\d{1,3}\b|" + r"@[a-z0-9.-]+\.[a-z]{2,}\b|" + r"\b(?:customer|mufg)\b)", + lowered, + ): + return True + return False + + +def validate_artifact_payloads( + directory: Path, source_set: Path | None = None +) -> list[str]: + errors: list[str] = [] + if not directory.is_dir() or directory.is_symlink(): + return ["artifact path must be a real directory"] + files = {path.name for path in directory.iterdir() if path.is_file()} + if files != EXPECTED_ARTIFACT_FILES: + errors.append( + f"artifact file set must be exactly {sorted(EXPECTED_ARTIFACT_FILES)}" + ) + return errors + for path in directory.iterdir(): + if path.is_symlink() or not path.is_file(): + errors.append(f"artifact directory contains unsupported entry: {path.name}") + payloads: dict[str, dict[str, Any]] = {} + for name in EXPECTED_ARTIFACT_FILES: + try: + payloads[name] = load_json_strict(directory / name) + except ValidationError as exc: + errors.append(str(exc)) + revisions = payloads.get("source-revisions.json") + if revisions is not None: + resolved_fields = { + key: value + for key, value in revisions.items() + if key != "checked_repositories" + } + for error in validate_resolved_manifest(resolved_fields): + errors.append(f"source-revisions.json: {error}") + checked = revisions.get("checked_repositories") + expected_by_repo = { + entry["repository"]: entry["revision"] + for entry in revisions.get("repositories", []) + if isinstance(entry, dict) + } + if not isinstance(checked, list) or len(checked) != 7: + errors.append("source-revisions.json: checked_repositories must contain seven records") + else: + observed_checked: list[str] = [] + for entry in checked: + if not isinstance(entry, dict) or set(entry) != { + "repository", + "canonical_repository", + "checked_sha", + "checked_tree_sha", + "authority_path", + "authority_content_revision", + "authority_git_blob_sha", + "detached", + "clean", + }: + errors.append( + "source-revisions.json: checked repository record has unsupported fields" + ) + continue + repository = entry.get("repository") + observed_checked.append(str(repository)) + if entry.get("canonical_repository") != f"HawkinsOperations/{repository}": + errors.append( + f"source-revisions.json: checked owner mismatch: {repository}" + ) + if entry.get("checked_sha") != expected_by_repo.get(str(repository)): + errors.append( + f"source-revisions.json: checked SHA mismatch: {repository}" + ) + reviewed_trees = { + item["repository"]: item.get("reviewed_tree_sha") + for item in revisions.get("repositories", []) + if isinstance(item, dict) + } + if entry.get("checked_tree_sha") != reviewed_trees.get(str(repository)): + errors.append( + f"source-revisions.json: checked tree mismatch: {repository}" + ) + resolved_entries = { + item["repository"]: item + for item in revisions.get("repositories", []) + if isinstance(item, dict) + } + resolved_entry = resolved_entries.get(str(repository), {}) + if entry.get("authority_path") != CANONICAL_AUTHORITY_PATHS.get( + str(repository) + ): + errors.append( + f"source-revisions.json: authority path mismatch: {repository}" + ) + if entry.get("authority_content_revision") != resolved_entry.get( + "authority_content_revision" + ): + errors.append( + f"source-revisions.json: authority content revision mismatch: " + f"{repository}" + ) + if re.fullmatch( + r"[0-9a-f]{40}", + str(entry.get("authority_git_blob_sha", "")), + ) is None: + errors.append( + f"source-revisions.json: authority blob is invalid: {repository}" + ) + if entry.get("detached") is not True or entry.get("clean") is not True: + errors.append( + f"source-revisions.json: checked state is not clean and detached: {repository}" + ) + if observed_checked != EXACT_REPOSITORIES: + errors.append( + "source-revisions.json: checked repositories differ from exact order" + ) + if source_set is not None and not errors: + actual_records, source_errors = verify_source_set( + source_set, resolved_fields + ) + errors.extend( + f"source-revisions.json: {error}" for error in source_errors + ) + if not source_errors and checked != actual_records: + errors.append( + "source-revisions.json: checked authority records differ from " + "the exact current source set" + ) + summary = payloads.get("verification-summary.json") + if summary is not None: + allowed = { + "schema", + "status", + "repository_count", + "repositories", + "checks", + "mutation_boundary", + "proof_ceiling", + } + if set(summary) != allowed: + errors.append("verification summary has unsupported fields") + if summary.get("schema") != "hawkinsoperations-convergence-verification-summary-v1": + errors.append("verification summary schema mismatch") + if summary.get("status") != "PASS": + errors.append("verification summary status must be PASS") + if summary.get("repository_count") != 7: + errors.append("verification summary repository count must be seven") + if summary.get("repositories") != EXACT_REPOSITORIES: + errors.append("verification summary repository set mismatch") + checks = summary.get("checks") + if checks != EXPECTED_VERIFICATION_CHECKS: + errors.append("verification summary check list differs from the exact approved checks") + if summary.get("mutation_boundary") != { + "repository_writes": False, + "pull_request_mutation": False, + "merge": False, + "ledger_mutation": False, + "runtime_mutation": False, + "proof_promotion": False, + }: + errors.append("verification summary mutation boundary mismatch") + if summary.get("proof_ceiling") != PROOF_CEILING: + errors.append("verification summary proof ceiling mismatch") + for name, payload in payloads.items(): + for path, value in walk(payload): + if isinstance(value, str) and is_private_scalar(value): + errors.append( + f"{name}: private or unsafe scalar at {'/'.join(path) or ''}" + ) + return errors + + +def write_json_atomic(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(path.name + ".tmp") + temporary.write_text( + json.dumps(value, indent=2, sort_keys=False) + "\n", encoding="utf-8", newline="\n" + ) + temporary.replace(path) + + +def write_verification_summary(path: Path) -> None: + write_json_atomic( + path, + { + "schema": "hawkinsoperations-convergence-verification-summary-v1", + "status": "PASS", + "repository_count": 7, + "repositories": EXACT_REPOSITORIES, + "checks": EXPECTED_VERIFICATION_CHECKS, + "mutation_boundary": { + "repository_writes": False, + "pull_request_mutation": False, + "merge": False, + "ledger_mutation": False, + "runtime_mutation": False, + "proof_promotion": False, + }, + "proof_ceiling": PROOF_CEILING, + }, + ) + + def check_required_text(errors: list[str]) -> None: for rel, needles in REQUIRED_TEXT.items(): text = read_text(ROOT / rel, errors) @@ -185,7 +1603,6 @@ def check_project_boundaries(all_text: str, errors: list[str]) -> None: for needle in required: if needle.lower() not in lowered: fail(f"missing project boundary wording: {needle}", errors) - forbidden = [ r"Project #1\s+is\s+an\s+active\s+reviewer\s+route", r"Project #1.{0,80}canonical", @@ -212,7 +1629,6 @@ def check_ceiling_boundaries(all_text: str, errors: list[str]) -> None: for needle in required: if needle.lower() not in lowered: fail(f"missing proof-boundary wording: {needle}", errors) - forbidden_patterns = [ r"\brendering\s+is\s+proof\b", r"\bGitHub rendering\s+is\s+proof\b", @@ -229,19 +1645,30 @@ def check_standing_controls(all_text: str, errors: list[str]) -> None: for issue in ("#8", "#10"): if issue not in all_text: fail(f"missing standing control issue reference: {issue}", errors) - if "Do not close unless Raylee explicitly approves replacing the standing-control role" not in all_text: + required = ( + "Do not close unless Raylee explicitly approves replacing " + "the standing-control role" + ) + if required not in all_text: fail("missing explicit replacement-approval boundary for .github#8/#10", errors) def check_exposure(text_files: list[Path], errors: list[str]) -> None: token_prefixes = ["AK" + "IA", "ghp" + "_", "github" + "_pat" + "_"] - private_ip = re.compile(r"\b(10\.\d{1,3}\.\d{1,3}\.\d{1,3}|172\.(?:1[6-9]|2\d|3[0-1])\.\d{1,3}\.\d{1,3}|192\.168\.\d{1,3}\.\d{1,3})\b") + private_ip = re.compile( + r"\b(10\.\d{1,3}\.\d{1,3}\.\d{1,3}|" + r"172\.(?:1[6-9]|2\d|3[0-1])\.\d{1,3}\.\d{1,3}|" + r"192\.168\.\d{1,3}\.\d{1,3})\b" + ) drive_path = re.compile(r"\b[A-Za-z]:\\") private_key = re.compile(r"BEGIN (?:RSA |OPENSSH )?PRIVATE KEY") for path in text_files: rel = path.relative_to(ROOT).as_posix() text = path.read_text(encoding="utf-8", errors="ignore") for line_no, line in enumerate(text.splitlines(), start=1): + # Defensive test literals in this verifier are never uploaded or public data. + if rel == "scripts/verify-command-center-invariants.py": + continue if drive_path.search(line): fail(f"{rel}:{line_no} exposes a local Windows path", errors) if private_ip.search(line): @@ -253,47 +1680,149 @@ def check_exposure(text_files: list[Path], errors: list[str]) -> None: fail(f"{rel}:{line_no} exposes a token-looking prefix", errors) -def check_identity_and_claim_context(text_files: list[Path], errors: list[str]) -> None: +def check_identity_and_claim_context( + text_files: list[Path], errors: list[str] +) -> None: for path in text_files: rel = path.relative_to(ROOT).as_posix() lines = path.read_text(encoding="utf-8", errors="ignore").splitlines() for line_no, line in enumerate(lines, start=1): lowered = line.lower() - if "hawkinsops" in lowered and not any(marker in lowered for marker in ("legacy", "reference", "v1", "prior", "not current")): - fail(f"{rel}:{line_no} uses HawkinsOps outside legacy/reference context", errors) + if "hawkinsops" in lowered and not any( + marker in lowered + for marker in ("legacy", "reference", "v1", "prior", "not current") + ): + fail( + f"{rel}:{line_no} uses HawkinsOps outside legacy/reference context", + errors, + ) for phrase in BLOCKED_CLAIMS: - phrase_pattern = re.compile(rf"(? int: +def run_full_verification(self_test: bool) -> list[str]: errors: list[str] = [] + errors.extend(tracked_vocabulary_findings()) manifest = load_manifest(errors) + source_manifest = load_source_manifest(errors) check_required_files(manifest, errors) + check_cross_repo_workflow(manifest, source_manifest, errors) + if self_test: + check_workflow_hostile_self_test(errors) check_required_text(errors) - text_files = iter_text_files() - all_text = "\n".join(path.read_text(encoding="utf-8", errors="ignore") for path in text_files) + all_text = "\n".join( + path.read_text(encoding="utf-8", errors="ignore") for path in text_files + ) check_project_boundaries(all_text, errors) check_ceiling_boundaries(all_text, errors) check_standing_controls(all_text, errors) check_exposure(text_files, errors) check_identity_and_claim_context(text_files, errors) + return errors + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument("--self-test", action="store_true") + parser.add_argument("--emit-source-manifest", type=Path) + parser.add_argument("--event-sha") + parser.add_argument("--verify-source-set", type=Path) + parser.add_argument("--verify-remote-main-content", type=Path) + parser.add_argument("--resolved-manifest", type=Path) + parser.add_argument("--source-revisions-output", type=Path) + parser.add_argument("--write-verification-summary", type=Path) + parser.add_argument("--validate-artifacts", type=Path) + parser.add_argument("--artifact-source-set", type=Path) + return parser.parse_args() + + +def main() -> int: + args = parse_args() + errors: list[str] = [] + + if args.emit_source_manifest is not None: + try: + source_manifest = load_json_strict(SOURCE_MANIFEST_PATH) + resolved = resolved_source_manifest(source_manifest, str(args.event_sha or "")) + write_json_atomic(args.emit_source_manifest, resolved) + except ValidationError as exc: + errors.append(str(exc)) + elif args.verify_source_set is not None: + if args.resolved_manifest is None or args.source_revisions_output is None: + errors.append( + "--verify-source-set requires --resolved-manifest and --source-revisions-output" + ) + else: + try: + resolved = load_json_strict(args.resolved_manifest) + records, source_errors = verify_source_set(args.verify_source_set, resolved) + errors.extend(source_errors) + if not errors: + output = dict(resolved) + output["checked_repositories"] = records + # The uploaded source-revisions artifact deliberately omits origins + # and local paths; only canonical owner and immutable revisions remain. + write_json_atomic(args.source_revisions_output, output) + except ValidationError as exc: + errors.append(str(exc)) + elif args.verify_remote_main_content is not None: + if args.resolved_manifest is None: + errors.append("--verify-remote-main-content requires --resolved-manifest") + else: + try: + resolved = load_json_strict(args.resolved_manifest) + errors.extend( + verify_remote_main_content(args.verify_remote_main_content, resolved) + ) + except ValidationError as exc: + errors.append(str(exc)) + elif args.write_verification_summary is not None: + write_verification_summary(args.write_verification_summary) + elif args.validate_artifacts is not None: + if args.artifact_source_set is None: + errors.append( + "--validate-artifacts requires --artifact-source-set" + ) + else: + errors.extend( + validate_artifact_payloads( + args.validate_artifacts, args.artifact_source_set + ) + ) + else: + errors.extend(run_full_verification(args.self_test)) if errors: print("COMMAND_CENTER_INVARIANTS=FAIL") for error in errors: print(f"- {error}") return 1 - print("COMMAND_CENTER_INVARIANTS=PASS") - print(f"checked_files={len(text_files)}") + if args.emit_source_manifest: + print("resolved_source_manifest=written") + elif args.verify_source_set: + print("checked_repositories=7") + elif args.verify_remote_main_content: + print("reviewed_main_content=6") + elif args.write_verification_summary: + print("verification_summary=written") + elif args.validate_artifacts: + print("sanitized_artifacts=2") + else: + print(f"checked_files={len(iter_text_files())}") return 0 diff --git a/tests/test_command_center_workflow_safety.py b/tests/test_command_center_workflow_safety.py new file mode 100644 index 0000000..85a2394 --- /dev/null +++ b/tests/test_command_center_workflow_safety.py @@ -0,0 +1,941 @@ +from __future__ import annotations + +import importlib.util +import json +import os +import subprocess +import tempfile +import unittest +from pathlib import Path +from unittest import mock + + +REPO_ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = REPO_ROOT / "scripts" / "verify-command-center-invariants.py" +SPEC = importlib.util.spec_from_file_location("command_center_invariants", MODULE_PATH) +assert SPEC is not None and SPEC.loader is not None +VERIFIER = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(VERIFIER) + + +class WorkflowSafetyTests(unittest.TestCase): + @classmethod + def setUpClass(cls) -> None: + cls.workflow = VERIFIER.WORKFLOW_PATH.read_text(encoding="utf-8") + cls.source_manifest = VERIFIER.load_json_strict(VERIFIER.SOURCE_MANIFEST_PATH) + + def assert_rejected(self, value: str, label: str) -> None: + self.assertTrue( + VERIFIER.unsafe_workflow_findings(value), + f"hostile workflow was accepted: {label}", + ) + + def test_tracked_vocabulary_guard_rejects_content_and_filename(self) -> None: + retired = "".join(("syn", "thetic")) + fullwidth = "".join(chr(ord(character) + 0xFEE0) for character in retired) + zero_width = retired[:3] + "\u200b" + retired[3:] + combining = retired[:3] + "\u034f" + retired[3:] + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + subprocess.run( + ["git", "init", "--quiet"], + cwd=root, + check=True, + capture_output=True, + ) + content_path = root / "content-fixture.txt" + filename_path = root / f"fixture-{fullwidth}.txt" + utf16_path = root / "utf16-fixture.md" + content_path.write_text( + ( + f"controlled-test boundary rejects {fullwidth}\n" + f"controlled-test boundary rejects {zero_width}\n" + f"controlled-test boundary rejects {combining}\n" + ), + encoding="utf-8", + ) + filename_path.write_text( + "controlled-test boundary\n", + encoding="utf-8", + ) + zero_width_filename_path = root / f"fixture-{zero_width}.txt" + combining_filename_path = root / f"fixture-{combining}.txt" + zero_width_filename_path.write_text( + "controlled-test boundary\n", + encoding="utf-8", + ) + combining_filename_path.write_text( + "controlled-test boundary\n", + encoding="utf-8", + ) + utf16_path.write_bytes( + f"controlled-test {retired}\n".encode("utf-16-le") + ) + subprocess.run( + [ + "git", + "add", + "--", + content_path.name, + filename_path.name, + zero_width_filename_path.name, + combining_filename_path.name, + utf16_path.name, + ], + cwd=root, + check=True, + capture_output=True, + ) + findings = VERIFIER.tracked_vocabulary_findings(root) + self.assertTrue(any("tracked content" in item for item in findings)) + self.assertGreaterEqual( + sum("tracked filename" in item for item in findings), + 3, + ) + self.assertTrue(any("utf16-fixture.md" in item for item in findings)) + + def test_vocabulary_security_view_preserves_benign_unicode_semantics(self) -> None: + normalized = VERIFIER.normalize_vocabulary_security_text( + "Café résumé – review 👩‍💻 only" + ) + retired = "".join(("syn", "thetic")) + self.assertNotIn(retired, normalized.casefold()) + self.assertIn("Cafe resume", normalized) + + def test_tracked_vocabulary_guard_fails_on_indexed_read_error(self) -> None: + listed = subprocess.CompletedProcess( + args=["git", "ls-files"], + returncode=0, + stdout=b"fixture.md\0", + stderr=b"", + ) + unreadable = subprocess.CompletedProcess( + args=["git", "show"], + returncode=128, + stdout=b"", + stderr=b"unreadable", + ) + with mock.patch.object( + VERIFIER.subprocess, + "run", + side_effect=(listed, unreadable), + ): + findings = VERIFIER.tracked_vocabulary_findings(REPO_ROOT) + self.assertTrue(any("could not read indexed content" in item for item in findings)) + + def test_current_workflow_is_structurally_safe(self) -> None: + self.assertEqual([], VERIFIER.unsafe_workflow_findings(self.workflow)) + + def test_permission_trigger_credential_and_action_attacks_fail(self) -> None: + mutations = { + "repository write token": self.workflow.replace( + "contents: read", "contents: write", 1 + ), + "issues write token": self.workflow.replace( + "contents: read", "issues: write", 1 + ), + "pull_request_target": self.workflow.replace( + "pull_request:", "pull_request_target:", 1 + ), + "persisted credentials": self.workflow.replace( + "persist-credentials: false", "persist-credentials: true", 1 + ), + "mutable action": self.workflow.replace( + f"actions/checkout@{VERIFIER.PINNED_ACTIONS['actions/checkout']}", + "actions/checkout@v4", + 1, + ), + } + for label, value in mutations.items(): + with self.subTest(label=label): + self.assert_rejected(value, label) + + def test_mutation_and_exit_neutralization_attacks_fail(self) -> None: + marker = "set -euo pipefail" + hostile_lines = { + "direct push": "git push origin main", + "PR creation": "gh pr create --title unsafe", + "HTTP PR creation": "curl -X POST https://api.github.com/repos/x/y/pulls", + "merge": "gh pr merge 1", + "ledger mutation": "python ho_factory.py lifetime-ledger-append", + "runtime mutation": "python tool.py runtime mutate", + "proof promotion": "python tool.py proof promote", + "swallowed failure": "false || true", + "set plus e": "set +e", + "unconditional success": "exit 0", + "backgrounded command": "python unsafe.py &", + "background and wait": "python unsafe.py & wait", + "swallowed echo": "false || echo ignored", + } + for label, hostile in hostile_lines.items(): + with self.subTest(label=label): + self.assert_rejected( + self.workflow.replace(marker, f"{marker}\n {hostile}", 1), + label, + ) + self.assert_rejected( + self.workflow.replace( + "run: python scripts/verify-command-center-invariants.py --self-test", + "continue-on-error: true\n run: python scripts/verify-command-center-invariants.py --self-test", + 1, + ), + "continue-on-error", + ) + self.assert_rejected( + self.workflow.replace( + "- name: Validate upload artifacts", + "- name: Validate upload artifacts\n if: always()", + 1, + ), + "always", + ) + + def test_required_commands_cannot_be_echoed_or_conditionally_disabled(self) -> None: + mutations = { + "echo detection verifier": self.workflow.replace( + "python -B source-set/hawkinsoperations-detections/scripts/verify_detection_contract.py", + "echo python -B source-set/hawkinsoperations-detections/scripts/verify_detection_contract.py", + 1, + ), + "echo sibling fetch": self.workflow.replace( + 'git -C "source-set/$repo" fetch --quiet origin "$revision"', + 'echo git -C "source-set/$repo" fetch --quiet origin "$revision"', + 1, + ), + "validation detached source omitted": self.workflow.replace( + ' --detections-root source-set/hawkinsoperations-detections --detections-ref "$(git -C source-set/hawkinsoperations-detections rev-parse HEAD)" --source-manifest source-set/hawkinsoperations-validation/validation/SOURCE_AUTHORITY_MANIFEST.json', + "", + 1, + ), + "validation unit import root omitted": self.workflow.replace( + 'PYTHONPATH="$GITHUB_WORKSPACE/source-set/hawkinsoperations-validation" ', + "", + 1, + ), + "platform observed SHA omitted": self.workflow.replace( + "HAWKINS_PLATFORM_IMMUTABLE_OBSERVED_SHA", + "HAWKINS_PLATFORM_OBSERVATION_OMITTED", + 1, + ), + "command-center observed SHA omitted": self.workflow.replace( + "HAWKINS_COMMAND_CENTER_IMMUTABLE_OBSERVED_SHA", + "HAWKINS_COMMAND_CENTER_OBSERVATION_OMITTED", + 1, + ), + "conditional job": self.workflow.replace( + " seven-repository-convergence:\n needs: command-center-invariants\n runs-on:", + " seven-repository-convergence:\n needs: command-center-invariants\n if: false\n runs-on:", + 1, + ), + "conditional principal step": self.workflow.replace( + " - name: Verify command-center invariants\n run:", + " - name: Verify command-center invariants\n if: false\n run:", + 1, + ), + } + for label, value in mutations.items(): + with self.subTest(label=label): + self.assert_rejected(value, label) + + def test_convergence_summary_cannot_outlive_owning_invariant_job(self) -> None: + self.assertIn( + " seven-repository-convergence:\n" + " needs: command-center-invariants\n" + " runs-on:", + self.workflow, + ) + self.assert_rejected( + self.workflow.replace( + " needs: command-center-invariants\n", + "", + 1, + ), + "missing owning-job dependency", + ) + + def test_whitespace_check_covers_committed_event_revision(self) -> None: + for fragment in ( + "fetch-depth: 0", + 'git diff --check "${{ github.event.pull_request.base.sha }}...' + '${{ github.event.pull_request.head.sha }}"', + "git show --check --format= HEAD", + ): + self.assertIn(fragment, self.workflow) + self.assert_rejected( + self.workflow.replace( + " run: |\n" + " set -euo pipefail\n" + ' if [[ "${{ github.event_name }}" == "pull_request" ]]; then\n' + ' git diff --check "${{ github.event.pull_request.base.sha }}...' + '${{ github.event.pull_request.head.sha }}"\n' + " else\n" + " git show --check --format= HEAD\n" + " fi\n", + " run: git diff --check\n", + 1, + ), + "working-tree-only whitespace check", + ) + + def test_sibling_fetch_retry_is_bounded_and_fails_closed(self) -> None: + for fragment in ( + "for attempt in 1 2 3 4 5 6; do", + "fetch_complete=1", + 'if [ "$attempt" -lt 6 ]; then', + "sleep 5", + 'test "$fetch_complete" -eq 1', + ): + self.assertIn(fragment, self.workflow) + + self.assert_rejected( + self.workflow.replace('test "$fetch_complete" -eq 1', "true", 1), + "unconditional success", + ) + + def test_exact_run_allowlist_rejects_command_laundering(self) -> None: + command = ( + "python -B source-set/hawkinsoperations-detections/scripts/" + "verify_detection_contract.py" + ) + mutations = { + "or colon": self.workflow.replace(command, f"{command} || :", 1), + "or exit zero": self.workflow.replace(command, f"{command} || exit 0", 1), + "semicolon true": self.workflow.replace(command, f"{command}; true", 1), + "compound swallowed exit": self.workflow.replace( + command, + f"{command} || {{ echo swallowed; exit 0; }}", + 1, + ), + "no-op command prefix": self.workflow.replace(command, f": {command}", 1), + "function override": self.workflow.replace( + command, + f"python() {{ :; }}\n {command}", + 1, + ), + "alias override": self.workflow.replace( + command, + f"alias python=:\n {command}", + 1, + ), + "PATH shadow": self.workflow.replace( + command, + f"PATH=/tmp/hostile:$PATH\n {command}", + 1, + ), + } + for label, value in mutations.items(): + with self.subTest(label=label): + self.assert_rejected(value, label) + + def test_shell_and_job_default_overrides_fail_closed(self) -> None: + mutations = { + "shell suffix": self.workflow.replace( + "shell: bash", "shell: bash {0}; true", 1 + ), + "shell nested exit": self.workflow.replace( + "shell: bash", "shell: bash -c '$0; exit 0' {0}", 1 + ), + "job default shell": self.workflow.replace( + " command-center-invariants:\n runs-on: ubuntu-latest", + " command-center-invariants:\n" + " defaults:\n" + " run:\n" + " shell: bash {0}; true\n" + " runs-on: ubuntu-latest", + 1, + ), + "step working directory": self.workflow.replace( + " - name: Verify command-center invariants\n run:", + " - name: Verify command-center invariants\n" + " working-directory: /tmp\n" + " run:", + 1, + ), + } + for label, value in mutations.items(): + with self.subTest(label=label): + self.assert_rejected(value, label) + + def test_trigger_neutralization_and_tracked_path_narrowing_fail(self) -> None: + mutations = { + "closed-only PR": self.workflow.replace( + " pull_request: {}", + " pull_request:\n types: [closed]", + 1, + ), + "ignored main": self.workflow.replace( + " pull_request: {}", + " pull_request:\n branches-ignore: [main]", + 1, + ), + "tracked vocabulary surface narrowed": self.workflow.replace( + " pull_request: {}", + ' pull_request:\n paths: ["governance/**"]', + 1, + ), + } + for label, value in mutations.items(): + with self.subTest(label=label): + self.assert_rejected(value, label) + + def test_artifact_validation_must_be_immediately_before_upload(self) -> None: + hostile = self.workflow.replace( + " - name: Upload sanitized convergence records", + " - name: Corrupt artifact after validation\n" + " run: echo invalid > verification-artifacts/verification-summary.json\n\n" + " - name: Upload sanitized convergence records", + 1, + ) + self.assert_rejected(hostile, "post-validation artifact mutation") + + def test_duplicate_yaml_key_fails_closed(self) -> None: + hostile = self.workflow.replace( + "permissions:\n contents: read", + "permissions:\n contents: read\npermissions:\n issues: write", + 1, + ) + findings = VERIFIER.unsafe_workflow_findings(hostile) + self.assertTrue(any("duplicate workflow key" in value for value in findings)) + + def test_missing_checkout_or_unsanitized_upload_fails(self) -> None: + self.assert_rejected( + self.workflow.replace( + 'git -C "source-set/$repo" fetch --quiet origin "$revision"', + 'printf "%s\\n" "$revision"', + 1, + ), + "missing sibling fetch", + ) + self.assert_rejected( + self.workflow.replace( + "verification-artifacts/source-revisions.json\n" + " verification-artifacts/verification-summary.json", + "verification-artifacts/", + 1, + ), + "broad upload", + ) + + def test_source_manifest_is_exact_closed_and_immutable(self) -> None: + self.assertEqual( + [], VERIFIER.validate_source_manifest(self.source_manifest) + ) + attacks = [] + missing = json.loads(json.dumps(self.source_manifest)) + missing["repositories"].pop() + attacks.append(missing) + duplicate = json.loads(json.dumps(self.source_manifest)) + duplicate["repositories"].append(dict(duplicate["repositories"][0])) + attacks.append(duplicate) + mutable = json.loads(json.dumps(self.source_manifest)) + mutable["repositories"][1]["revision"] = "main" + attacks.append(mutable) + missing_self_content = json.loads(json.dumps(self.source_manifest)) + missing_self_content["repositories"][0].pop("authority_content_revision") + attacks.append(missing_self_content) + malformed_tree = json.loads(json.dumps(self.source_manifest)) + malformed_tree["repositories"][1]["reviewed_tree_sha"] = "not-a-tree" + attacks.append(malformed_tree) + spoofed = json.loads(json.dumps(self.source_manifest)) + spoofed["repositories"][1]["canonical_repository"] = ( + "HawkinsOperations/hawkinsoperations-detections-suffix" + ) + attacks.append(spoofed) + fallback = json.loads(json.dumps(self.source_manifest)) + fallback["constraints"]["default_branch_fallback"] = True + attacks.append(fallback) + unknown = json.loads(json.dumps(self.source_manifest)) + unknown["repositories"][1]["extension"] = "laundered" + attacks.append(unknown) + for candidate in attacks: + with self.subTest(candidate=candidate): + self.assertTrue(VERIFIER.validate_source_manifest(candidate)) + + def test_resolved_manifest_digest_rejects_tampering(self) -> None: + resolved = VERIFIER.resolved_source_manifest( + self.source_manifest, "1" * 40, "2" * 40 + ) + self.assertEqual([], VERIFIER.validate_resolved_manifest(resolved)) + resolved["repositories"][1]["revision"] = "2" * 40 + self.assertIn( + "resolved source manifest digest mismatch", + VERIFIER.validate_resolved_manifest(resolved), + ) + + def test_main_content_observation_is_tree_bound_not_commit_bound(self) -> None: + resolved = VERIFIER.resolved_source_manifest( + self.source_manifest, "1" * 40, "2" * 40 + ) + observed = { + entry["repository"]: entry["reviewed_tree_sha"] + for entry in resolved["repositories"] + if entry["repository"] != ".github" + } + self.assertEqual([], VERIFIER.compare_observed_main_trees(resolved, observed)) + observed["hoxline"] = "3" * 40 + errors = VERIFIER.compare_observed_main_trees(resolved, observed) + self.assertTrue(any("hoxline" in error for error in errors)) + + def test_duplicate_json_key_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as temp: + path = Path(temp) / "duplicate.json" + path.write_text('{"schema":"one","SCHEMA":"two"}\n', encoding="utf-8") + with self.assertRaises(VERIFIER.ValidationError): + VERIFIER.load_json_strict(path) + + def test_command_center_manifest_shape_is_closed(self) -> None: + original = VERIFIER.load_json_strict(VERIFIER.MANIFEST_PATH) + for mutation in ("root", "invariant"): + with self.subTest(mutation=mutation): + candidate = json.loads(json.dumps(original)) + if mutation == "root": + candidate["extension"] = {"ai_authority": True} + else: + candidate["invariants"]["ai_authority"] = True + with tempfile.TemporaryDirectory() as temp: + path = Path(temp) / "manifest.json" + path.write_text(json.dumps(candidate), encoding="utf-8") + prior = VERIFIER.MANIFEST_PATH + try: + VERIFIER.MANIFEST_PATH = path + errors = [] + VERIFIER.load_manifest(errors) + finally: + VERIFIER.MANIFEST_PATH = prior + self.assertTrue(errors) + + +class SourceSetTests(unittest.TestCase): + def run_git(self, repo: Path, *args: str) -> str: + result = subprocess.run( + ["git", "-C", str(repo), *args], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + def create_source_set(self, root: Path) -> dict: + entries = [] + for repository in VERIFIER.EXACT_REPOSITORIES: + repo = root / repository + repo.mkdir() + self.run_git(repo, "init", "--quiet") + self.run_git(repo, "config", "user.name", "Command Center Test") + self.run_git(repo, "config", "user.email", "test@invalid.example") + authority_path = VERIFIER.CANONICAL_AUTHORITY_PATHS[repository] + authority_file = repo / Path(authority_path) + authority_file.parent.mkdir(parents=True, exist_ok=True) + authority_file.write_text(repository + "\n", encoding="utf-8") + self.run_git(repo, "add", authority_path) + self.run_git(repo, "commit", "--quiet", "-m", "fixture") + sha = self.run_git(repo, "rev-parse", "HEAD") + tree = self.run_git(repo, "rev-parse", "HEAD^{tree}") + self.run_git( + repo, "remote", "add", "origin", VERIFIER.CANONICAL_ORIGINS[repository] + ) + self.run_git(repo, "checkout", "--quiet", "--detach", sha) + entries.append( + { + "repository": repository, + "canonical_repository": f"HawkinsOperations/{repository}", + "revision": sha, + "authority_content_revision": sha, + "reviewed_tree_sha": tree, + } + ) + payload = { + "schema": "hawkinsoperations-resolved-convergence-source-set-v1", + "manifest_id": "HAWKINSOPERATIONS_SEVEN_SOURCE_PR_HEAD_MATRIX_V1", + "repositories": entries, + "constraints": { + "exact_repository_count": 7, + "read_only": True, + "default_branch_fallback": False, + "require_detached_exact_revision": True, + "record_checked_revisions": True, + "consumer_outputs_are_not_authority": True, + "proof_ceiling": VERIFIER.PROOF_CEILING, + }, + } + payload["manifest_sha256"] = VERIFIER.hashlib.sha256( + json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest() + return payload + + def test_exact_clean_detached_source_set_passes(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) / "source-set" + root.mkdir() + resolved = self.create_source_set(root) + records, errors = VERIFIER.verify_source_set(root, resolved) + self.assertEqual([], errors) + self.assertEqual(7, len(records)) + + def test_dirty_wrong_origin_branch_and_missing_repo_fail(self) -> None: + for attack in ("dirty", "origin", "branch", "missing", "extra"): + with self.subTest(attack=attack), tempfile.TemporaryDirectory() as temp: + root = Path(temp) / "source-set" + root.mkdir() + resolved = self.create_source_set(root) + target = root / "hawkinsoperations-detections" + if attack == "dirty": + (target / "untracked.txt").write_text("dirty\n", encoding="utf-8") + elif attack == "origin": + self.run_git( + target, + "remote", + "set-url", + "origin", + "https://github.com/Other/hawkinsoperations-detections.git", + ) + elif attack == "branch": + self.run_git(target, "switch", "--quiet", "-c", "main") + elif attack == "missing": + target.rename(root / "missing") + else: + (root / "eighth-repository").mkdir() + _, errors = VERIFIER.verify_source_set(root, resolved) + self.assertTrue(errors) + + def test_origin_rewrite_cannot_launder_wrong_stored_origin(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) / "source-set" + root.mkdir() + resolved = self.create_source_set(root) + repository = "hawkinsoperations-detections" + target = root / repository + canonical = VERIFIER.CANONICAL_ORIGINS[repository] + wrong = "https://local.invalid/hawkinsoperations-detections.git" + self.run_git(target, "remote", "set-url", "origin", wrong) + rewrite_env = { + "GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": f"url.{canonical}.insteadOf", + "GIT_CONFIG_VALUE_0": wrong, + } + with mock.patch.dict(os.environ, rewrite_env, clear=False): + self.assertEqual( + canonical, + self.run_git(target, "remote", "get-url", "origin"), + "attack precondition: interpreted Git URL must look canonical", + ) + _, errors = VERIFIER.verify_source_set(root, resolved) + self.assertTrue( + any("canonical origin mismatch" in error for error in errors), + errors, + ) + + def test_git_environment_scrub_rejects_every_ambient_git_control(self) -> None: + hostile = { + "GIT_DIR": "decoy", + "GIT_WORK_TREE": "decoy", + "GIT_COMMON_DIR": "decoy", + "GIT_INDEX_FILE": "decoy", + "GIT_OBJECT_DIRECTORY": "decoy", + "GIT_ALTERNATE_OBJECT_DIRECTORIES": "decoy", + "GIT_CONFIG": "decoy", + "GIT_CONFIG_GLOBAL": "decoy", + "GIT_CONFIG_SYSTEM": "decoy", + "GIT_CONFIG_NOSYSTEM": "0", + "GIT_CONFIG_COUNT": "1", + "GIT_CONFIG_KEY_0": "core.repositoryformatversion", + "GIT_CONFIG_VALUE_0": "1", + "GIT_CEILING_DIRECTORIES": "decoy", + "GIT_DISCOVERY_ACROSS_FILESYSTEM": "1", + "GIT_SHALLOW_FILE": "decoy", + "GIT_NAMESPACE": "decoy", + "GIT_REPLACE_REF_BASE": "refs/decoy", + "GIT_IMPLICIT_WORK_TREE": "1", + "GIT_NO_REPLACE_OBJECTS": "0", + "GIT_TERMINAL_PROMPT": "1", + } + with mock.patch.dict(os.environ, hostile, clear=False): + sanitized = VERIFIER.sanitized_git_environment() + self.assertEqual("1", sanitized["GIT_NO_REPLACE_OBJECTS"]) + self.assertEqual("0", sanitized["GIT_TERMINAL_PROMPT"]) + self.assertEqual( + {"git_no_replace_objects", "git_terminal_prompt"}, + { + key.casefold() + for key in sanitized + if key.casefold().startswith("git_") + }, + ) + + def test_git_dir_decoy_cannot_redirect_stored_origin_authority(self) -> None: + with tempfile.TemporaryDirectory() as temp: + base = Path(temp) + root = base / "source-set" + root.mkdir() + resolved = self.create_source_set(root) + repository = "hawkinsoperations-detections" + target = root / repository + canonical = VERIFIER.CANONICAL_ORIGINS[repository] + wrong = "https://local.invalid/hawkinsoperations-detections.git" + self.run_git(target, "remote", "set-url", "origin", wrong) + decoy = base / "decoy" + decoy.mkdir() + self.run_git(decoy, "init", "--quiet") + self.run_git(decoy, "remote", "add", "origin", canonical) + raw_env = os.environ.copy() + raw_env["GIT_DIR"] = str(decoy / ".git") + interpreted = subprocess.run( + [ + "git", + "-C", + str(target), + "config", + "--local", + "--get-all", + "remote.origin.url", + ], + check=True, + capture_output=True, + text=True, + env=raw_env, + ).stdout.strip() + self.assertEqual(canonical, interpreted) + with mock.patch.dict( + os.environ, {"GIT_DIR": str(decoy / ".git")}, clear=False + ): + self.assertEqual(wrong, VERIFIER.stored_origin(target)) + _, errors = VERIFIER.verify_source_set(root, resolved) + self.assertTrue( + any("canonical origin mismatch" in error for error in errors), + errors, + ) + + def test_git_index_file_cannot_hide_staged_dirty_authority(self) -> None: + with tempfile.TemporaryDirectory() as temp: + base = Path(temp) + root = base / "source-set" + root.mkdir() + resolved = self.create_source_set(root) + repository = "hawkinsoperations-detections" + target = root / repository + clean_index = base / "clean.index" + alternate_env = os.environ.copy() + alternate_env["GIT_INDEX_FILE"] = str(clean_index) + subprocess.run( + ["git", "-C", str(target), "read-tree", "HEAD"], + check=True, + capture_output=True, + env=alternate_env, + ) + authority_file = target / VERIFIER.CANONICAL_AUTHORITY_PATHS[repository] + original = authority_file.read_text(encoding="utf-8") + authority_file.write_text("staged contradiction\n", encoding="utf-8") + self.run_git( + target, + "add", + VERIFIER.CANONICAL_AUTHORITY_PATHS[repository], + ) + authority_file.write_text(original, encoding="utf-8") + hidden = subprocess.run( + [ + "git", + "-C", + str(target), + "status", + "--porcelain=v1", + "--untracked-files=all", + ], + check=True, + capture_output=True, + text=True, + env=alternate_env, + ).stdout.strip() + self.assertEqual("", hidden, "attack precondition: alternate index is clean") + with mock.patch.dict( + os.environ, {"GIT_INDEX_FILE": str(clean_index)}, clear=False + ): + _, errors = VERIFIER.verify_source_set(root, resolved) + self.assertTrue( + any("source checkout is dirty" in error for error in errors), + errors, + ) + + def test_missing_empty_or_multiple_stored_origins_fail_closed(self) -> None: + for attack in ("missing", "empty", "multiple"): + with self.subTest(attack=attack), tempfile.TemporaryDirectory() as temp: + root = Path(temp) / "source-set" + root.mkdir() + resolved = self.create_source_set(root) + target = root / "hawkinsoperations-detections" + self.run_git(target, "config", "--unset-all", "remote.origin.url") + if attack == "empty": + self.run_git(target, "config", "--add", "remote.origin.url", "") + elif attack == "multiple": + self.run_git( + target, + "config", + "--add", + "remote.origin.url", + VERIFIER.CANONICAL_ORIGINS["hawkinsoperations-detections"], + ) + self.run_git( + target, + "config", + "--add", + "remote.origin.url", + "https://local.invalid/hawkinsoperations-detections.git", + ) + _, errors = VERIFIER.verify_source_set(root, resolved) + self.assertTrue( + any("exactly one nonempty local URL" in error for error in errors), + errors, + ) + + def test_authority_content_revision_is_bound_to_canonical_current_blob(self) -> None: + for attack in ("unreachable", "wrong-blob"): + with self.subTest(attack=attack), tempfile.TemporaryDirectory() as temp: + root = Path(temp) / "source-set" + root.mkdir() + resolved = self.create_source_set(root) + target = root / "hawkinsoperations-detections" + entry = resolved["repositories"][1] + if attack == "unreachable": + entry["authority_content_revision"] = "f" * 40 + else: + authority_file = target / Path( + VERIFIER.CANONICAL_AUTHORITY_PATHS[ + "hawkinsoperations-detections" + ] + ) + authority_file.write_text("contradictory authority\n", encoding="utf-8") + self.run_git(target, "add", authority_file.relative_to(target).as_posix()) + self.run_git(target, "commit", "--quiet", "-m", "contradiction") + entry["authority_content_revision"] = self.run_git( + target, "rev-parse", "HEAD" + ) + self.run_git(target, "checkout", "--quiet", "--detach", entry["revision"]) + unsigned = { + key: value + for key, value in resolved.items() + if key != "manifest_sha256" + } + resolved["manifest_sha256"] = VERIFIER.hashlib.sha256( + json.dumps( + unsigned, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + ).hexdigest() + _, errors = VERIFIER.verify_source_set(root, resolved) + self.assertTrue(errors) + + def test_uploaded_authority_blob_record_is_reverified_against_source_set( + self, + ) -> None: + with tempfile.TemporaryDirectory() as temp: + base = Path(temp) + source_root = base / "source-set" + source_root.mkdir() + resolved = self.create_source_set(source_root) + records, errors = VERIFIER.verify_source_set(source_root, resolved) + self.assertEqual([], errors) + artifacts = base / "artifacts" + artifacts.mkdir() + source_record = dict(resolved) + source_record["checked_repositories"] = records + VERIFIER.write_json_atomic( + artifacts / "source-revisions.json", source_record + ) + VERIFIER.write_verification_summary( + artifacts / "verification-summary.json" + ) + self.assertEqual( + [], VERIFIER.validate_artifact_payloads(artifacts, source_root) + ) + path = artifacts / "source-revisions.json" + tampered = VERIFIER.load_json_strict(path) + tampered["checked_repositories"][0][ + "authority_git_blob_sha" + ] = "f" * 40 + VERIFIER.write_json_atomic(path, tampered) + errors = VERIFIER.validate_artifact_payloads( + artifacts, source_root + ) + self.assertTrue( + any("exact current source set" in error for error in errors) + ) + + +class ArtifactSanitizerTests(unittest.TestCase): + def create_valid_artifacts(self, root: Path) -> None: + source = VERIFIER.resolved_source_manifest( + VERIFIER.load_json_strict(VERIFIER.SOURCE_MANIFEST_PATH), + "1" * 40, + "2" * 40, + ) + source["checked_repositories"] = [ + { + "repository": entry["repository"], + "canonical_repository": entry["canonical_repository"], + "checked_sha": entry["revision"], + "checked_tree_sha": entry["reviewed_tree_sha"], + "authority_path": VERIFIER.CANONICAL_AUTHORITY_PATHS[ + entry["repository"] + ], + "authority_content_revision": entry[ + "authority_content_revision" + ], + "authority_git_blob_sha": "a" * 40, + "detached": True, + "clean": True, + } + for entry in source["repositories"] + ] + VERIFIER.write_json_atomic(root / "source-revisions.json", source) + VERIFIER.write_verification_summary(root / "verification-summary.json") + + def test_closed_sanitized_artifact_pair_passes(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + self.create_valid_artifacts(root) + self.assertEqual([], VERIFIER.validate_artifact_payloads(root)) + + def test_private_and_unsupported_artifacts_fail(self) -> None: + hostile_values = [ + r"C:\private\output", + r"\\server\share\output", + "/home/operator/output", + "%2fhome%2foperator%2foutput", + "ghp_example", + "192.168.1.12", + "private@example.com", + "customer evidence", + "AKIAIOSFODNN7EXAMPLE", + "Bearer abcdefghijklmnopqrstuvwxyz", + ] + for hostile in hostile_values: + with self.subTest(hostile=hostile), tempfile.TemporaryDirectory() as temp: + root = Path(temp) + self.create_valid_artifacts(root) + path = root / "verification-summary.json" + value = json.loads(path.read_text(encoding="utf-8")) + value["checks"][0] = hostile + VERIFIER.write_json_atomic(path, value) + self.assertTrue(VERIFIER.validate_artifact_payloads(root)) + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + self.create_valid_artifacts(root) + (root / "raw.log").write_text("not approved\n", encoding="utf-8") + self.assertTrue(VERIFIER.validate_artifact_payloads(root)) + + def test_fabricated_check_set_fails(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + self.create_valid_artifacts(root) + path = root / "verification-summary.json" + value = json.loads(path.read_text(encoding="utf-8")) + value["checks"] = [f"fabricated_check_{index}" for index in range(23)] + VERIFIER.write_json_atomic(path, value) + self.assertTrue(VERIFIER.validate_artifact_payloads(root)) + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + self.create_valid_artifacts(root) + (root / "verification-summary.json").write_text("{", encoding="utf-8") + self.assertTrue(VERIFIER.validate_artifact_payloads(root)) + + +if __name__ == "__main__": + unittest.main()