diff --git a/.echo/certification.json b/.echo/certification.json new file mode 100644 index 0000000..b06c33e --- /dev/null +++ b/.echo/certification.json @@ -0,0 +1,4 @@ +{ + "version": 1, + "journey": ["python3", "-B", "scripts/certforge_journey.py"] +} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 01578a4..bc4b3d2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -13,13 +13,22 @@ permissions: jobs: test: - runs-on: ubuntu-latest + # Release Sentinel requires the authentic Actions check named CI. + name: CI + runs-on: ubuntu-24.04 timeout-minutes: 30 steps: - name: Check out exact source - uses: actions/checkout@v4 + uses: actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Verify candidate identity + env: + EXPECTED_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$EXPECTED_SHA" - name: Configure Python - uses: actions/setup-python@v5 + uses: actions/setup-python@v7 with: python-version: "3.13" cache: pip @@ -37,9 +46,9 @@ jobs: run: python scripts/p7_acceptance.py - name: Upload verification evidence if: always() - uses: actions/upload-artifact@v4 + uses: actions/upload-artifact@v7 with: - name: certification-forge-evidence-${{ github.sha }} + name: certification-forge-evidence-${{ github.event.pull_request.head.sha || github.sha }} path: artifacts/ if-no-files-found: error retention-days: 30 diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml new file mode 100644 index 0000000..446e95b --- /dev/null +++ b/.github/workflows/codeql.yml @@ -0,0 +1,42 @@ +name: certification-forge-codeql + +"on": + push: + branches: + - main + - feat/p5-adapter-execution + pull_request: + workflow_dispatch: + +permissions: + contents: read + +jobs: + analyze: + name: CodeQL + runs-on: ubuntu-24.04 + timeout-minutes: 30 + permissions: + contents: read + security-events: write + steps: + - name: Check out exact source + uses: actions/checkout@v7 + with: + ref: ${{ github.event.pull_request.head.sha || github.sha }} + persist-credentials: false + - name: Verify candidate identity + env: + EXPECTED_SHA: ${{ github.event.pull_request.head.sha || github.sha }} + run: test "$(git rev-parse HEAD)" = "$EXPECTED_SHA" + - name: Initialize Python analysis + uses: github/codeql-action/init@v4 + with: + languages: python + build-mode: none + - name: Analyze and upload results + uses: github/codeql-action/analyze@v4 + with: + category: /language:python + ref: ${{ github.event_name == 'pull_request' && format('refs/pull/{0}/head', github.event.pull_request.number) || github.ref }} + sha: ${{ github.event.pull_request.head.sha || github.sha }} diff --git a/deploy/README.md b/deploy/README.md index ca18a25..3c1d713 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -16,6 +16,14 @@ Staging failures retain the service log under `$STATE_ROOT/deploy-logs/staging-.log`. Listener ownership, signed readiness verification, live-smoke gates, and rollback requirements still apply. +# Source validation and release evidence + +The `CI` Actions job runs the full deterministic suite with coverage and P1, P6, +and P7 verification. The `CodeQL` job performs Python security analysis and uploads +its results. Both check out the immutable PR head (or event SHA on push), verify +that checkout, and bind their output to it. Their names match Release Sentinel's +required authentic checks; neither is a substitute status or a release verdict. + A green source-validation run does not prove current production identity or provide signed production E2E evidence. Every changed candidate needs fresh exact-SHA certification. Historical smoke reports and old certified tags must not diff --git a/deploy/deploy_forge.sh b/deploy/deploy_forge.sh index 01bf226..4785c0c 100644 --- a/deploy/deploy_forge.sh +++ b/deploy/deploy_forge.sh @@ -27,13 +27,19 @@ source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/readiness.sh" configure_readiness_timeout "${CERTFORGE_READY_TIMEOUT_S:-90}" ADAPTER_DIR="${ECHO_CERTFORGE_PROD_ADAPTER_DIR:-$STATE_ROOT/p5}" ADAPTER_MODE="${CERTFORGE_ADAPTER_MODE:-required}" -TRUSTED_MANIFEST_SHA256="${ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-965106b00917268d556b325719f26f5096e6c3746551658ffecb9fd4a95ec342}" +# Canonical-JSON digest of policies/mandatory-rules.v2.json (== run_worker._PRODUCTION_MANIFEST_SHA256). +# The old default (965106b0...) was the raw file-bytes digest, which run_worker rejects. +TRUSTED_MANIFEST_SHA256="${ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-08ba068ceb3e14bfed2690337edbb94c546e3e0a1a89b1321f7657653d8eea43}" UNIT_PATH="/etc/systemd/system/$SERVICE.service" DISPATCH_UNIT_PATH="/etc/systemd/system/$DISPATCH_SERVICE.service" RELEASE_DROPIN="/etc/systemd/system/$SERVICE.service.d/10-release.conf" DISPATCH_RELEASE_DROPIN="/etc/systemd/system/$DISPATCH_SERVICE.service.d/10-release.conf" ENV_FILE="${CERTFORGE_ENV_FILE:-/home/forge/.config/echo/certforge.env}" -GITC=(-c credential.helper= -c credential.helper="store --file=/home/forge/.config/echo/omega_git_creds") +# Source fetch credentials. Default: the FORGE credential store. Repositories on the +# echoomegaprime account are fetched with a repo-scoped GitHub App helper instead +# (CERTFORGE_GIT_CREDENTIAL_HELPER), because that store cannot see them. +GIT_CREDENTIAL_HELPER="${CERTFORGE_GIT_CREDENTIAL_HELPER:-store --file=/home/forge/.config/echo/omega_git_creds}" +GITC=(-c credential.helper= -c credential.helper="$GIT_CREDENTIAL_HELPER") LOCK_FILE="${CERTFORGE_DEPLOY_LOCK:-/run/lock/echo-certforge-deploy.lock}" exec 9>"$LOCK_FILE" @@ -591,7 +597,9 @@ sudo tee "$DISPATCH_UNIT_PATH" >/dev/null <:` route observations. +The collector rechecks admission before every observe and recall, and at +completion. An aggregate health result is not authority for a named route. + +For `observe`, result contains a new actual `receipt_id`. The collector then +requests `recall`; recall result contains the identical ID, exact `content`, and +canonical `content_sha256`. Content schema `echo.finisher.observation.v1` contains +the receipt ID, `mode: live`, exact collection challenge/kind/name, identities and +bindings, a fresh `observed_at`, and the journey result. No receipt or provider +invocation may be reused. The complete collection is bounded to 30 minutes. + +The receipt service must preserve actual authenticated provenance and permit +exact readback. Merely echoing the just-submitted request or returning an +unverifiable success flag does not satisfy this contract. The separately +admitted signer must independently verify the receipt manifest before signing. + +## Finite journey and result contract + +The ordered fronts are `auto`, `cauto`, `gauto`, `grauto`, `CFauto`, `CFcauto`, +`CFgauto`, `CFgrauto`, mapped to claude/codex/gemini/grok respectively. + +1. `front/` performs the authorized real front journey. Its result binds + actual provider/model/account, run/worker/session and native invocation IDs; + observed model/subscription identity; exact resumed session; real scoped MCP, + hook capture and exact capture readback hashes; live heartbeat; uncertain + replay suppression; and process cleanup. It requires zero inference caused + by readonly status. `start` has the actual `singleton_observed` shape with + `already_running`, verified readback and exact target. `status_snapshot` is + the real UI snapshot, with no top-level accepted/applied/request ID. It must + be fresh, online, installation/resource/heartbeat/service verified, and bound + to the actual full Cloudflare target and transport source. `controls` contains + exactly pause/resume/halt, each with distinct actual `request_id` and + `native_request_id`, matching `action`, exact full `target` including run, + `submitted_at`, accepted/applied proof and paused/running/halted `state`. + Each `readback` must independently match that target and transport source, + have fresh `observed_at >= submitted_at`, verified installation/readback, + and `controller.last_control_id` equal to the exact native request ID. + The actual paused/halted flags must agree with the action. Pause/resume + require active service and fresh heartbeat. Halt requires a fresh independently + established `terminal_halt_verified: true`, `lifecycle: halted` and + `online: false` observation after process exit without pretending that the + exited controller still has a live heartbeat. A live controller setting its + halted flag before cleanup is insufficient. A stale + observation, replayed native ID, other run/installation, accepted write, + pending journal or unavailable terminal readback must fail. +2. Immediately after **each** halt, `restoration/` requires + `method: admitted_owner_operation`, the same admission ID, previous run ID, + a new actual owner operation ID, and a fresh healthy exact-target snapshot. + This is a normal independently authorized owner bootstrap, never a synthetic + frontend start receipt. Without an actual admitted restoration route the + collection stops; it does not continue to the next front in a stopped state. +3. `negative/` checks unauthenticated, wrong_owner, wrong_installation, + wrong_source, invalid_token, expired_token, stale_control, replayed_control, + out_of_scope, redirect and forbidden_native_route. Each result must prove + denial, no side effects, and identical bounded before/after state hashes. + Credentials and personal identity never enter the receipt. A redirect check + must refuse rather than follow an unadmitted destination. +4. `rollback/restore-and-recover` requires actual prior and recovered artifact + and source readbacks, both checkpoint source matches, and verified restored + and recovered state. All four actual checkpoint/configuration SHA256 readbacks + must **equal their separately owner-approved immutable policy pins**. A digest + of the right length, a truthful-looking boolean, or only a Worker image + replacement is insufficient. It runs **before** final stability. +5. `stability/0`, `/1`, `/2` establish aggregate health, resource admission, + fresh heartbeat and zero unexpected restarts only. Legacy aggregate + `snapshots` are rejected; they cannot count as eight-front evidence. + Each aggregate probe is followed by **eight separate** owner-issued, + independently recalled `stability-front/:` receipts, each + bound to its exact challenge/kind/name, front and expected provider. Each + route result requires `route_readback_verified: true`, a fresh healthy + `snapshot`, and `snapshot_sha256` equal to the canonical digest of that + snapshot. Reusing another route's receipt/result fails named-route binding. + All routes may legitimately share the single Cloudflare singleton `run_id`; + route receipt provenance, not distinct run IDs, proves eight-front coverage. + The first-to-third aggregate observation interval must be at least 30 seconds. +6. `external/acceptance` requires a distinct independent reviewer, actual Access + session, immutable UI assets, exact HTTPS origin and accepted live behavior. + +Tests use explicitly synthetic authority fixtures to exercise forged, replayed, +stale, mismatched, revoked and incomplete receipts. They also exercise real local +child processes on FORGE for small-pipe/nonreading deadlines, output bounds and +opaque broker state inheritance. None of these tests call a real provider or +produce admission, signing, deployment or production-acceptance evidence. + +## Unchanged signing and deployment boundaries + +Only after real observations and independent review may an already admitted +independent signer add its actual key identity and produce the existing signed +production-E2E envelope. The normal trusted-key verifier must accept that exact +envelope and its source/target/environment/manifest bindings before re-running +certification. Private signing keys never belong in this collector or its worker. + +The existing deployment admission hook remains unchanged: + +```sh +python /home/forge/echo-certification-forge-current/scripts/deployment_admission_hook.py \ + admit --forge-url ADMITTED_FORGE_ORIGIN --tenant ACTUAL_TENANT \ + --artifact-digest ACTUAL_ARTIFACT_BYTE_SHA256 --environment staging \ + --environment-identity-digest ACTUAL_ENVIRONMENT_IDENTITY_DIGEST \ + --rule-manifest-digest ACTUAL_RULE_MANIFEST_DIGEST \ + --deployment-id ACTUAL_DEPLOYMENT_ID --requested-by ACTUAL_ADMITTED_OWNER +``` + +Any credential required by that existing hook stays opaque in its normal owner +environment; do not paste values into this command or handoff. Staging and +production gates remain enforced. Actual outcome registration and exact readback +remain owner responsibilities after an admitted deployment. This collector's +source, tests, draft PR or unsigned output cannot clear those gates. diff --git a/images/anchor/Dockerfile b/images/anchor/Dockerfile index 387e74f..6137490 100644 --- a/images/anchor/Dockerfile +++ b/images/anchor/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df +FROM python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111 ARG SOURCE_REVISION=unknown ARG IMAGE_VARIANT=unknown @@ -9,7 +9,7 @@ LABEL org.opencontainers.image.title="Echo Certification Forge Anchor" \ org.opencontainers.image.source="https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge" \ org.opencontainers.image.revision="${SOURCE_REVISION}" \ org.opencontainers.image.base.name="python:3.12-alpine" \ - org.opencontainers.image.base.digest="sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" \ + org.opencontainers.image.base.digest="sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" \ echo.certforge.image.role="anchor" \ echo.certforge.image.variant="${IMAGE_VARIANT}" \ echo.certforge.admission.policy="${ADMISSION_POLICY_ID}" \ diff --git a/images/custody/Dockerfile b/images/custody/Dockerfile index 6d7f995..85f8ddc 100644 --- a/images/custody/Dockerfile +++ b/images/custody/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df +FROM python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111 ARG SOURCE_REVISION=unknown ARG IMAGE_VARIANT=unknown @@ -9,7 +9,7 @@ LABEL org.opencontainers.image.title="Echo Certification Forge Custody" \ org.opencontainers.image.source="https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge" \ org.opencontainers.image.revision="${SOURCE_REVISION}" \ org.opencontainers.image.base.name="python:3.12-alpine" \ - org.opencontainers.image.base.digest="sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" \ + org.opencontainers.image.base.digest="sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" \ echo.certforge.image.role="custody" \ echo.certforge.image.variant="${IMAGE_VARIANT}" \ echo.certforge.admission.policy="${ADMISSION_POLICY_ID}" \ diff --git a/images/requirements.in b/images/requirements.in index c6bed32..ea9a891 100644 --- a/images/requirements.in +++ b/images/requirements.in @@ -1,4 +1,4 @@ -cryptography==49.0.0 +cryptography==50.0.0 fastapi==0.139.0 pydantic==2.13.4 uvicorn==0.51.0 diff --git a/images/requirements.lock b/images/requirements.lock index 3aa498a..07cfe86 100644 --- a/images/requirements.lock +++ b/images/requirements.lock @@ -126,53 +126,53 @@ colorama==0.4.6 \ --hash=sha256:08695f5cb7ed6e0531a20572697297273c47b8cae5a63ffc6d6ed5c201be6e44 \ --hash=sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6 # via click -cryptography==49.0.0 \ - --hash=sha256:026ac7423e6fa66872d3bf889be5974507da3944f866f704fa200eadacd00001 \ - --hash=sha256:07cab27cc7b7e0fd28e5e26bb9eeedde5c135c868b46de4a27845abe94af6122 \ - --hash=sha256:084ef1af862eb07ec46d25f68689f2102a9fc0e05ce7b80f14f5fe51e4eef0f6 \ - --hash=sha256:0b82e28ee398a386f0807bba7884d30f25218855690f45115831bcce5d90822c \ - --hash=sha256:0e959b578856a3924bc0cbb710fc12c387b9412a951389f3ca61704a9e25f325 \ - --hash=sha256:0f21641cf4b30fca7aee061ced0ec7ad7b073518088b7c9969a297c0ae796c69 \ - --hash=sha256:196ecd6a36e4e9aa10270393bb98d8df88fccee0bf1e5128b91ae4eb4375896d \ - --hash=sha256:2400ef9c9e2299a25614eb1dea3db54a69b1349efd043bfac9c67630d136df36 \ - --hash=sha256:28d8b15e6275f12c8a207dc309dfa957903c927d08d0cc937ee3f63f200693cc \ - --hash=sha256:2afe9051da7ae7bd5905da5a949280c7d2bb75682e188f650a9d0f2756b834c6 \ - --hash=sha256:2eda353d8a27bcbcaa4cbed18994a74ab4d19a2ca897db188ea269ab9b71419b \ - --hash=sha256:32703d93296f5c1f4b53349ad3a250c2cae0fdecd3a3dd5d47e616d8d616af27 \ - --hash=sha256:33cd0565932807baddb67b96dbee92f2c374b5c89dee09fd74079aeb8c8dba61 \ - --hash=sha256:35b151772baff2c74cba7fa290ceaff4c3b11c0c881eb93eb5dbc05a7cfbba18 \ - --hash=sha256:36d1709f992593689b45bda411498d62c6e365f2ca00b84657d4dadd24de16db \ - --hash=sha256:42b0684e0e40cf26122427802486f6d93aea593612603a94fbf260c7eb1e9c1b \ - --hash=sha256:4ae387c9cb68ea569ca17e490d66d8142b81c3cc814bf179974b7d146e490bbb \ - --hash=sha256:53ecee2e23f7169b6117e99fc8a944e5e50f79e69758a83b52a00cb98ab2b2d2 \ - --hash=sha256:66ec79c3904820572d7e987abdf304281f141d37ad9a489b8e97066e7b9b6459 \ - --hash=sha256:67e1d20ad9ef3a563c59ef22e7a8a0b8210bd26604369ea4a30a7c66aefe504e \ - --hash=sha256:6f2debedf9ca60cf1d5bd466475638af5130f89965605cd818484d19987d3a21 \ - --hash=sha256:6fc361c34fb6aac015ce19435876635e5c6d21db31998b0920f675f131e043b8 \ - --hash=sha256:73a205dce83953d131a4aa1e0fd917a2fd1c5b1eef251e9d7152efefcbf5caf7 \ - --hash=sha256:7abcee80084cda3f7691f3eb1ce480d8df49cec637b429aa35986c1de71738aa \ - --hash=sha256:8c25ceb16df5b9435f3f6a9829204985b0e0cbee3b48aacd432c7d2c850b44d9 \ - --hash=sha256:966fe0e9c67490071f14c0d2b1cb2dfb3023c5ce39457343931415f08382f2db \ - --hash=sha256:9e82dcc8e56052715fb18b2429e3bca4823b1629136a2084fc45a9a5cecb9b64 \ - --hash=sha256:b20133d204d2bb56ba047642199603876c872026ca53e79c35b83772ab2cc505 \ - --hash=sha256:b39efa323140595abd3ecca8529d321ae50f55f3aa3ba9cc81ea56a6011953d5 \ - --hash=sha256:b47db11c2c3525083296069b98ac5221907455e989ae0c2e3008bde851921615 \ - --hash=sha256:b87e65d263b3e5d3bb92a57e2a6638e2f31110fa7aa890c7b2dbba42248d0a3f \ - --hash=sha256:b970c6da94d5bb18629db453d14f2a1300f6bf59b61e9b82377931ef95504866 \ - --hash=sha256:be9fcb48a55f023493482827d4f459bd263cc20efde64f204b97c123201850c6 \ - --hash=sha256:c2bc30226390d60ea19d9f82b19db005fe0452154a23c1c410c12ea801e43561 \ - --hash=sha256:c83782480a4a9da4d0feb51950131ba32e12e70813848b3343f6e18c28a66838 \ - --hash=sha256:cbc77da8c523d5abd028635ba850a6966fcee2c82e2bf65a41d1d8afe0f98be9 \ - --hash=sha256:ccac2bfebc306b862133e3bb71f3f6ee8bb525240089b2d952e4144b3a6d5da7 \ - --hash=sha256:d0527ce944105f257f605a827d6ebead966c752038b6e8656abb9c5edee6fc68 \ - --hash=sha256:d8ecde755e2e91bf773fc94e8c9d730cd7f2007004cb492263a794ec3899a1c8 \ - --hash=sha256:e3fb64c420688e5319ae25113a354015abbd8dffbfbc41781a1ea66fc7622ac3 \ - --hash=sha256:e5dfc1e64de5677cec922ffa8da89c546d0415bf6efdf081842e5d44c84e1f0e \ - --hash=sha256:ec5e529fb80935c94fe7b729f9972b50e351a0e6b50aa294fd5cabb109fcc29a \ - --hash=sha256:f37d847238971164fdbc68ade6f6574aecc9c0af714190e2083429ff68f4ce9d \ - --hash=sha256:f78ff2c9ed8dc2d036b0f4d640e22522213d047c1b14e61205a7e55c80a494d4 \ - --hash=sha256:f89660a348f4f78a92366240a61404e337586ef7f5909a2fef59ca88ef505493 \ - --hash=sha256:fc1e275c2f1d97b1a6450b8b0ea3ebfa6e087a611c2b26cb2404d48588abab7b +cryptography==50.0.0 \ + --hash=sha256:031e2d5dd4bb9caa3ca9c82e5a197fd8ae680232cee62603d1a813f3f07e3d03 \ + --hash=sha256:06a32a980526a6ab9a4b9bf8f7385800791e2bb960903cb6b530e4817509a3b7 \ + --hash=sha256:07479a1cb08219ab719147e742e76090c9c773321959bb94946fffdd397a6437 \ + --hash=sha256:07949c449a1abcf60d1ee6e88956d89404c7df3c8258f46589e912988e551987 \ + --hash=sha256:105110f43a471dbd0060b9c9516cb8a6a79233631a04cc2ba16f28323ac6e025 \ + --hash=sha256:11b74db56cdbe3cdee6e3f6982ecb70334fa10dce99ed58bf7894aaaa3b2a037 \ + --hash=sha256:12b9c6996425c76ea6c457ace4f3073e715b8c545add07cd1a8f3a4f90691269 \ + --hash=sha256:1489e263a8048bb8b6a8bac662eb2d402ea5d2b7b4699b72f385f1e2772db105 \ + --hash=sha256:19736989797678c6af1e55cd49055cdbcb55d8f6b5583ac5335f933aba9101dc \ + --hash=sha256:1b4a266766514614f8aa60416e71f2fc6e575d36e7bdc90f644fadb2f4b75b95 \ + --hash=sha256:2a8183b489dc1f7f80f135780fadc1108f14b31b8a40411c7a5b17425f65f28b \ + --hash=sha256:37fdb0d0111f1e2ff07139dfb79f1b49531f8e213c46f1163dd7642979b58c47 \ + --hash=sha256:3f5735ffe4996d28b809371756219f5354864902a3b9e7c0b9ee87041209fc9c \ + --hash=sha256:49e7d93abdbd2990caced757e5fade25302f719c3c8fb6e6fff2dde98999fc41 \ + --hash=sha256:5e34edd123674534acd70147f0ca331eaa2c74e6325fb2028c886aa26ba0b68c \ + --hash=sha256:62598a8a57f815db4c6259a4e97d857dab56697e7de8e8ab02352ab74da1995d \ + --hash=sha256:65c2c3add92b45fd0709db8594536aea39c2a67af0e27ffcf049c498501140b7 \ + --hash=sha256:6ba6a53445bd3cfa809ef3ef5f1589aa6ba08784a1d962bf47d0940e871dab1c \ + --hash=sha256:6e7d61120573a7f2cd94cc095f9e81f6967c61ccdf194285aa143ecec8e0b708 \ + --hash=sha256:7cec5b856506da6defb290f30c9ee687d5f5e8cb0bd3f6459dde43b0b4fa40ef \ + --hash=sha256:80b63928fa35083b33966ce1efb70e5b9607181e49dcd1c22c8c005e319f667f \ + --hash=sha256:82148ec5bddac30b51a5b3c1945075f896fa022cb93f8e4a01e9f6ee95292c5f \ + --hash=sha256:828743d939e9629bc267b8e2d08d8bb67cd4319c771a33d4b18b22dd8fb7440a \ + --hash=sha256:8d89f3976b10b4ce31118de72329025f70d2c6ead14a8217c5514dd2c6d5a78f \ + --hash=sha256:8eb5e1172eb569ea8a872796576e6a67c276351728b6455d5beb01242b027c6a \ + --hash=sha256:900131fafd8aead39ac7dd3a7e833be754c17a95cfd91221636949fe4eb0aa8a \ + --hash=sha256:910d11e1a385c654bf738bf3e6b8e6ed5de0f5610fcae2be9e5b398d8081d20e \ + --hash=sha256:910e1d2668e7de9648f2bcee30e180db2a6b15c30f887d7c4c93ddf96e3992e3 \ + --hash=sha256:9aa87839c383bdbab6ef865787a1fb877af8dd03464c4400322726feaaadfc6d \ + --hash=sha256:a1b30560f2acc95aa8b2e06e716a13dbfc97314747b80d9707e307f77b40d6b3 \ + --hash=sha256:a91296cb61e8df6f86d0c19cc4068228da256bf59bf86049fbd821084565327f \ + --hash=sha256:b42a28c1844fd9de8f3f7d540e36b66f3a9c83fceac7170ebc7a6a19edd9dcae \ + --hash=sha256:bd1c592e4d5974f0d08d4888e432157adba757c66da0246918e43677fafa2d30 \ + --hash=sha256:c87f62a3d3b9888ed0fdde100ec06aa61ca9cd44bad9057d1dff9a516b5f5bb9 \ + --hash=sha256:c99c003e088647b8a5b7c145d6f78c335f6348332b62e142d411c4b63d1460b9 \ + --hash=sha256:ccdc4a71a4dabae05de219404f9f4abc38e3b58422177ff93d0da05967dafa07 \ + --hash=sha256:d24fead1d4d076e1bfb006dcec392074a3cd8d7b4fc8a595aa64073b2b7a96ba \ + --hash=sha256:d58c3db7cd6eed54e6c06744db55456b65ebd7492ddeae9c1e93cfca7aa857d3 \ + --hash=sha256:d764dcf130c428ef66786f866dd750f53182bc608813489915e9fc106bb0c82f \ + --hash=sha256:df2a58a472f332225671c35b0a830208b86d004f82baa8530fa3782c85646533 \ + --hash=sha256:e722f16708d854fe924790e051061f6704a472c3bac347b6fd88033ea8dd0dc5 \ + --hash=sha256:ecfed7367f965a0328cfbdd70da860f15441f002f613185668c6e6ebf5a0ac11 \ + --hash=sha256:eeac2acb5a20ed25e0ad6d1df9891a520b78b404266b6d11778f25d5d691a6c9 \ + --hash=sha256:f59e38625469987d7ef6d495323c55e7db6c212eaf6112267e0d3b565a2e9c9f \ + --hash=sha256:f89831ef99dd7dd169ab06d63a831adb9e20a87aac6d380266bbda5823349169 \ + --hash=sha256:fd9192b7b70c573d7f214eb1ae35e00d359f6f5e4b27c7e21e30de1fc6204645 # via -r images/requirements.in fastapi==0.139.0 \ --hash=sha256:99ab7b2d92223c76d6cf10757ab3f89d45b38267fc20b2a136cf02f6beac3145 \ diff --git a/images/runner/Dockerfile b/images/runner/Dockerfile index 1052310..70a7e41 100644 --- a/images/runner/Dockerfile +++ b/images/runner/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df +FROM python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111 ARG SOURCE_REVISION=unknown ARG IMAGE_VARIANT=unknown @@ -9,7 +9,7 @@ LABEL org.opencontainers.image.title="Echo Certification Forge Runner" \ org.opencontainers.image.source="https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge" \ org.opencontainers.image.revision="${SOURCE_REVISION}" \ org.opencontainers.image.base.name="python:3.12-alpine" \ - org.opencontainers.image.base.digest="sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" \ + org.opencontainers.image.base.digest="sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" \ echo.certforge.image.role="runner" \ echo.certforge.image.variant="${IMAGE_VARIANT}" \ echo.certforge.admission.policy="${ADMISSION_POLICY_ID}" \ diff --git a/images/signer/Dockerfile b/images/signer/Dockerfile index fa7c7ae..183614f 100644 --- a/images/signer/Dockerfile +++ b/images/signer/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df +FROM python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111 ARG SOURCE_REVISION=unknown ARG IMAGE_VARIANT=unknown @@ -9,7 +9,7 @@ LABEL org.opencontainers.image.title="Echo Certification Forge Verdict Signer" \ org.opencontainers.image.source="https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge" \ org.opencontainers.image.revision="${SOURCE_REVISION}" \ org.opencontainers.image.base.name="python:3.12-alpine" \ - org.opencontainers.image.base.digest="sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" \ + org.opencontainers.image.base.digest="sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" \ echo.certforge.image.role="signer" \ echo.certforge.image.variant="${IMAGE_VARIANT}" \ echo.certforge.admission.policy="${ADMISSION_POLICY_ID}" \ diff --git a/images/verifier/Dockerfile b/images/verifier/Dockerfile index 62aa2bb..8d17357 100644 --- a/images/verifier/Dockerfile +++ b/images/verifier/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df +FROM python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111 ARG SOURCE_REVISION=unknown ARG IMAGE_VARIANT=unknown @@ -9,7 +9,7 @@ LABEL org.opencontainers.image.title="Echo Certification Forge Verifier" \ org.opencontainers.image.source="https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge" \ org.opencontainers.image.revision="${SOURCE_REVISION}" \ org.opencontainers.image.base.name="python:3.12-alpine" \ - org.opencontainers.image.base.digest="sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" \ + org.opencontainers.image.base.digest="sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" \ echo.certforge.image.role="verifier" \ echo.certforge.image.variant="${IMAGE_VARIANT}" \ echo.certforge.admission.policy="${ADMISSION_POLICY_ID}" \ diff --git a/images/worker/Dockerfile b/images/worker/Dockerfile index e4389b8..2d6b454 100644 --- a/images/worker/Dockerfile +++ b/images/worker/Dockerfile @@ -1,4 +1,4 @@ -FROM python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df +FROM python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111 ARG SOURCE_REVISION=unknown ARG IMAGE_VARIANT=unknown @@ -9,7 +9,7 @@ LABEL org.opencontainers.image.title="Echo Certification Forge Worker" \ org.opencontainers.image.source="https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge" \ org.opencontainers.image.revision="${SOURCE_REVISION}" \ org.opencontainers.image.base.name="python:3.12-alpine" \ - org.opencontainers.image.base.digest="sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" \ + org.opencontainers.image.base.digest="sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" \ echo.certforge.image.role="worker" \ echo.certforge.image.variant="${IMAGE_VARIANT}" \ echo.certforge.admission.policy="${ADMISSION_POLICY_ID}" \ diff --git a/pyproject.toml b/pyproject.toml index 3f1a10e..0d44c78 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ requires-python = ">=3.12" license = {text = "Proprietary"} authors = [{name = "ECHO-OMEGA-PRIME", email = "bobbymcwilliams@echo-op.com"}] dependencies = [ - "cryptography>=44.0.0", + "cryptography>=50.0.0", "fastapi>=0.115.0", "pydantic>=2.10.0", "uvicorn>=0.34.0", diff --git a/scripts/certforge_journey.py b/scripts/certforge_journey.py new file mode 100644 index 0000000..722977e --- /dev/null +++ b/scripts/certforge_journey.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python3 +"""Certification Forge journey for echo-certification-forge itself (ops-20260924). + +CertForge runs this against the exact acquired commit, offline and stdlib-only, from a +read-only checkout. The product's full suite needs pydantic and cryptography, so it runs in +hosted CI and in P1-P7 acceptance. This journey proves the release contract that has to hold +for any certifiable CertForge build: + +- The production rule manifest's canonical digest equals the pinned + run_worker._PRODUCTION_MANIFEST_SHA256 and the deploy default. +- Every role image and the journey sandbox use one patched base-image digest, and + cryptography is hash-locked at a patched release. +- The dispatcher unit only Wants= the API service (issue #22). +- Every source module compiles. +- The stdlib-runnable release suites pass through scripts/certforge_testkit.py. +""" +from __future__ import annotations + +import hashlib +import json +import re +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import certforge_testkit # noqa: E402 + +SUITES = ("tests/test_release_pins.py", "tests/test_deploy_gate.py") + + +def fail(message: str) -> int: + print(f"CERTFORGE_SELF_JOURNEY_FAILED: {message}", file=sys.stderr) + return 1 + + +def canonical_digest(path: Path) -> str: + value = json.loads(path.read_text(encoding="utf-8")) + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), + ensure_ascii=False).encode("utf-8")).hexdigest() + + +def main() -> int: + cert = json.loads((ROOT / ".echo/certification.json").read_text(encoding="utf-8")) + apps = json.loads((ROOT / ".echo/apps.json").read_text(encoding="utf-8")) + if cert.get("version") != 1 or apps.get("apps", {}).get("certification-forge", {}).get("enabled") is not True: + return fail("opt-in contract invalid") + + manifest_digest = canonical_digest(ROOT / "policies/mandatory-rules.v2.json") + run_worker = (ROOT / "src/echo_certification_forge/run_worker.py").read_text(encoding="utf-8") + pinned = re.search(r'_PRODUCTION_MANIFEST_SHA256 = \(\s*"([0-9a-f]{64})"', run_worker) + deploy = (ROOT / "deploy/deploy_forge.sh").read_text(encoding="utf-8") + deploy_default = re.search(r'TRUSTED_MANIFEST_SHA256="\$\{ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-([0-9a-f]{64})\}"', + deploy) + if not pinned or pinned.group(1) != manifest_digest: + return fail(f"run_worker manifest pin != canonical policy digest {manifest_digest}") + if not deploy_default or deploy_default.group(1) != manifest_digest: + return fail("deploy default trusted manifest digest != canonical policy digest") + + compiled = 0 + for path in sorted((ROOT / "src").rglob("*.py")): + try: + compile(path.read_text(encoding="utf-8"), str(path), "exec") + except SyntaxError as exc: + return fail(f"syntax error: {path.relative_to(ROOT)}: {exc}") + compiled += 1 + print(json.dumps({"manifest_sha256": manifest_digest, "modules_compiled": compiled})) + # The two release suites are self-contained text/contract checks; tests/conftest.py needs + # pydantic/cryptography, which the stdlib-only sandbox does not have. + return certforge_testkit.run(ROOT, SUITES, sys_paths=("tests", "."), budget_s=60.0, + min_passed=15, conftest=False) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/certforge_testkit.py b/scripts/certforge_testkit.py new file mode 100644 index 0000000..ce78108 --- /dev/null +++ b/scripts/certforge_testkit.py @@ -0,0 +1,562 @@ +"""Stdlib-only test runner for Certification Forge journeys (ops-20260924). + +The CertForge sandbox runs a journey in ``python:3.12-alpine`` with no network, no +third-party packages, a read-only checkout and a 90 s wall clock. Most ECHO suites are +written for pytest, so this module provides the small pytest surface they use: +``raises``, ``mark.parametrize`` / ``skip`` / ``skipif`` (other marks are no-ops), ``param``, +``skip()``, ``fail()``, ``importorskip()``, ``approx``, and ``@fixture`` functions (plain or +``yield``) from the suite module and its ``conftest.py`` files, plus the built-in fixtures +``tmp_path``, ``monkeypatch`` and ``capsys``. It also runs ``unittest.TestCase`` classes and +pytest-style ``Test*`` classes. + +Third-party modules that a suite imports but never exercises can be replaced with inert +stubs (``stub_modules``). Stubs are installed only when the real module is missing, and the +report names them. Every failure, error, import error, empty run or blown time budget fails +the journey (fail closed). +""" +from __future__ import annotations + +import contextlib +import importlib +import importlib.util +import inspect +import io +import itertools +import os +import re +import sys +import tempfile +import time +import traceback +import types +import unittest +from pathlib import Path +from typing import Any, Callable, Iterable + + +class Skipped(Exception): + """Raised by ``pytest.skip`` / ``importorskip``.""" + + +class Failed(AssertionError): + """Raised by ``pytest.fail``.""" + + +class BudgetExceeded(Exception): + """The journey ran out of its wall-clock budget (fails the journey).""" + + +# ── pytest compatibility surface ───────────────────────────────────────────────────────── +class _Raises: + def __init__(self, expected: Any, match: str | None = None) -> None: + self.expected = expected + self.match = match + self.value: BaseException | None = None + + def __enter__(self) -> "_Raises": + return self + + def __exit__(self, exc_type, exc, tb) -> bool: + if exc_type is None: + raise Failed(f"DID NOT RAISE {self.expected}") + if not issubclass(exc_type, self.expected): + return False + if self.match is not None and not re.search(self.match, str(exc)): + raise Failed(f"regex {self.match!r} did not match {str(exc)!r}") + self.value = exc + return True + + +class _Param: + def __init__(self, *values: Any, id: str | None = None, marks: Any = ()) -> None: # noqa: A002 + self.values = values + self.id = id + self.marks = marks + + +class _NoopMark: + def __call__(self, *args: Any, **kwargs: Any) -> Any: + if len(args) == 1 and not kwargs and callable(args[0]): + return args[0] + return lambda fn: fn + + +class _Mark: + @staticmethod + def parametrize(names: Any, values: Iterable[Any], **_: Any) -> Callable: + if isinstance(names, str): + names = [n.strip() for n in names.split(",") if n.strip()] + names = list(names) + rows = [] + for value in values: + if isinstance(value, _Param): + value = value.values if len(names) > 1 else value.values[0] + rows.append(tuple(value) if len(names) > 1 else (value,)) + + def deco(fn: Callable) -> Callable: + fn.__dict__.setdefault("_cf_params", []).insert(0, (names, rows)) + return fn + return deco + + @staticmethod + def skip(reason: str = "") -> Callable: + def deco(fn: Callable) -> Callable: + fn._cf_skip = reason or "skip" + return fn + return deco + + @staticmethod + def skipif(condition: Any, reason: str = "") -> Callable: + def deco(fn: Callable) -> Callable: + if condition: + fn._cf_skip = reason or "skipif" + return fn + return deco + + def __getattr__(self, name: str) -> _NoopMark: + return _NoopMark() + + +class _Approx: + def __init__(self, expected: Any, rel: float | None = None, abs: float | None = None) -> None: # noqa: A002 + self.expected, self.rel, self.abs = expected, rel, abs + + def _close(self, a: float, b: float) -> bool: + rel = 1e-6 if self.rel is None else self.rel + tol = max(rel * abs(b), 1e-12 if self.abs is None else self.abs) + return abs(a - b) <= tol + + def __eq__(self, other: Any) -> bool: + if isinstance(self.expected, (list, tuple)): + return len(other) == len(self.expected) and all( + self._close(a, b) for a, b in zip(other, self.expected)) + if isinstance(self.expected, dict): + return other.keys() == self.expected.keys() and all( + self._close(other[k], v) for k, v in self.expected.items()) + return self._close(other, self.expected) + + def __repr__(self) -> str: + return f"approx({self.expected!r})" + + +def _fixture(*args: Any, **kwargs: Any) -> Any: + def mark(fn: Callable) -> Callable: + fn._cf_fixture = True + return fn + if len(args) == 1 and callable(args[0]) and not kwargs: + return mark(args[0]) + return mark + + +def _skip(reason: str = "", **_: Any) -> None: + raise Skipped(reason) + + +def _fail(reason: str = "", **_: Any) -> None: + raise Failed(reason) + + +def _importorskip(name: str, *_: Any, **__: Any) -> Any: + try: + return importlib.import_module(name) + except ImportError as exc: + raise Skipped(f"{name} unavailable: {exc}") from None + + +def pytest_module() -> types.ModuleType: + mod = types.ModuleType("pytest") + mod.__dict__.update( + raises=_Raises, mark=_Mark(), param=_Param, fixture=_fixture, skip=_skip, + fail=_fail, importorskip=_importorskip, approx=_Approx, + __certforge_shim__=True, + ) + mod.skip.Exception = Skipped # type: ignore[attr-defined] + return mod + + +# ── built-in fixtures ──────────────────────────────────────────────────────────────────── +def _resolve_dotted(path: str) -> Any: + """Import the longest importable prefix of ``path``, then walk the remaining attributes.""" + parts = path.split(".") + for cut in range(len(parts), 0, -1): + try: + obj = importlib.import_module(".".join(parts[:cut])) + except ImportError: + continue + for attr in parts[cut:]: + obj = getattr(obj, attr) + return obj + raise ImportError(f"cannot resolve {path!r}") + + +class MonkeyPatch: + _MISSING = object() + + def __init__(self) -> None: + self._undo: list[Callable[[], None]] = [] + + def setattr(self, target: Any, name: Any, value: Any = _MISSING, raising: bool = True) -> None: + if value is self._MISSING: # "pkg.mod.attr" form + value = name + owner_path, _, name = str(target).rpartition(".") + target = _resolve_dotted(owner_path) + old = getattr(target, name, self._MISSING) + if old is self._MISSING and raising: + raise AttributeError(f"{target!r} has no attribute {name!r}") + setattr(target, name, value) + self._undo.append(lambda: delattr(target, name) if old is self._MISSING + else setattr(target, name, old)) + + def delattr(self, target: Any, name: str, raising: bool = True) -> None: + if not hasattr(target, name): + if raising: + raise AttributeError(name) + return + old = getattr(target, name) + delattr(target, name) + self._undo.append(lambda: setattr(target, name, old)) + + def setitem(self, mapping: Any, key: Any, value: Any) -> None: + old = mapping.get(key, self._MISSING) + mapping[key] = value + self._undo.append(lambda: mapping.pop(key, None) if old is self._MISSING + else mapping.__setitem__(key, old)) + + def delitem(self, mapping: Any, key: Any, raising: bool = True) -> None: + if key not in mapping: + if raising: + raise KeyError(key) + return + old = mapping.pop(key) + self._undo.append(lambda: mapping.__setitem__(key, old)) + + def setenv(self, name: str, value: Any, prepend: str | None = None) -> None: + value = str(value) + if prepend and name in os.environ: + value = value + prepend + os.environ[name] + self.setitem(os.environ, name, value) + + def delenv(self, name: str, raising: bool = True) -> None: + self.delitem(os.environ, name, raising) + + def syspath_prepend(self, path: Any) -> None: + sys.path.insert(0, str(path)) + self._undo.append(lambda: sys.path.remove(str(path))) + + def chdir(self, path: Any) -> None: + old = os.getcwd() + os.chdir(path) + self._undo.append(lambda: os.chdir(old)) + + def undo(self) -> None: + while self._undo: + self._undo.pop()() + + +class CapSys: + def __init__(self) -> None: + self._out, self._err = io.StringIO(), io.StringIO() + self._saved = (sys.stdout, sys.stderr) + sys.stdout, sys.stderr = self._out, self._err + + def readouterr(self) -> Any: + out, err = self._out.getvalue(), self._err.getvalue() + self._out.seek(0), self._out.truncate(), self._err.seek(0), self._err.truncate() + return types.SimpleNamespace(out=out, err=err) + + def close(self) -> None: + sys.stdout, sys.stderr = self._saved + + +# ── inert stubs for unexercised third-party imports ────────────────────────────────────── +class _StubObject: + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + def __call__(self, *args: Any, **kwargs: Any) -> Any: + if len(args) == 1 and callable(args[0]) and not kwargs: + return args[0] # used as a decorator + return _StubObject() + + def __getattr__(self, name: str) -> Any: + return _StubObject() + + +def _stub_module(name: str) -> types.ModuleType: + mod = types.ModuleType(name) + mod.__path__ = [] # allow "import pkg.sub" + mod.__certforge_stub__ = True + + def __getattr__(attr: str) -> Any: + if attr.startswith("__"): + raise AttributeError(attr) + base = Exception if attr.endswith(("Error", "Exception")) else _StubObject + value = type(attr, (base,), {"__module__": name}) + setattr(mod, attr, value) + return value + mod.__getattr__ = __getattr__ # type: ignore[attr-defined] + return mod + + +def install_stubs(names: Iterable[str]) -> list[str]: + installed = [] + for name in names: + if name in sys.modules: + continue + try: + importlib.import_module(name) + continue + except ImportError: + pass + sys.modules[name] = _stub_module(name) + parent, _, child = name.rpartition(".") + if parent and parent in sys.modules: + setattr(sys.modules[parent], child, sys.modules[name]) + installed.append(name) + return installed + + +# ── collection and execution ───────────────────────────────────────────────────────────── +class _Session: + def __init__(self, root: Path, deadline: float) -> None: + self.root = root + self.deadline = deadline + self.passed = self.failed = self.skipped = 0 + self.failures: list[str] = [] + + # fixtures ---------------------------------------------------------------------------- + def _resolve(self, name: str, fixtures: dict, cache: dict, teardown: list) -> Any: + if name in cache: + return cache[name] + if name == "tmp_path": + value = Path(tempfile.mkdtemp(prefix="cf-")) + elif name == "monkeypatch": + value = MonkeyPatch() + teardown.append(value.undo) + elif name == "capsys": + value = CapSys() + teardown.append(value.close) + elif name == "request": + value = types.SimpleNamespace(param=None, node=None, config=None) + elif name in fixtures: + fn = fixtures[name] + kwargs = {p: self._resolve(p, fixtures, cache, teardown) + for p in inspect.signature(fn).parameters} + value = fn(**kwargs) + if inspect.isgenerator(value): + gen = value + value = next(gen) + teardown.append(lambda g=gen: next(g, None)) + else: + raise LookupError(f"fixture {name!r} not found") + cache[name] = value + return value + + def _call(self, label: str, fn: Callable, fixtures: dict, bound: dict) -> None: + if time.monotonic() > self.deadline: + raise BudgetExceeded(f"journey time budget exhausted before {label}") + skip = getattr(fn, "_cf_skip", None) + if skip: + self.skipped += 1 + return + cache: dict = dict(bound) + teardown: list = [] + try: + params = [p for p in inspect.signature(fn).parameters if p != "self"] + kwargs = {p: self._resolve(p, fixtures, cache, teardown) for p in params} + fn(**kwargs) + self.passed += 1 + except Skipped: + self.skipped += 1 + except (BudgetExceeded, KeyboardInterrupt): + raise + except BaseException as exc: # noqa: BLE001 - every error (incl. SystemExit) is a failure + self.failed += 1 + self.failures.append(f"{label}: {type(exc).__name__}: {exc}\n" + + "".join(traceback.format_exc(limit=6))[-1500:]) + finally: + for fin in reversed(teardown): + with contextlib.suppress(Exception): + fin() + + def _expand(self, label: str, fn: Callable, fixtures: dict) -> None: + grids = getattr(fn, "_cf_params", []) + if not grids: + self._call(label, fn, fixtures, {}) + return + for i, combo in enumerate(itertools.product(*[rows for _, rows in grids])): + bound: dict = {} + for (names, _), row in zip(grids, combo): + bound.update(zip(names, row)) + self._call(f"{label}[{i}]", fn, fixtures, bound) + + # suites ------------------------------------------------------------------------------ + def _conftest_fixtures(self, suite: Path) -> dict: + fixtures: dict = {} + if not getattr(self, "use_conftest", True): + return fixtures # self-contained suites; conftest needs packages the sandbox lacks + chain = [d for d in [suite.parent, *suite.parent.parents] if self.root in (d, *d.parents)] + for directory in reversed(chain): + conftest = directory / "conftest.py" + if conftest.is_file(): + mod = _load(conftest, "cf_conftest_" + _slug(conftest.relative_to(self.root))) + fixtures.update(_fixtures_of(mod)) + return fixtures + + def run_suite(self, suite: Path) -> None: + fixtures = self._conftest_fixtures(suite) + module = _load(suite, "cf_suite_" + _slug(suite.relative_to(self.root))) + fixtures.update(_fixtures_of(module)) + rel = str(suite.relative_to(self.root)) + for name, obj in list(vars(module).items()): + if isinstance(obj, type) and issubclass(obj, unittest.TestCase): + result = unittest.TestResult() + unittest.defaultTestLoader.loadTestsFromTestCase(obj).run(result) + self.passed += result.testsRun - len(result.failures) - len(result.errors) - len(result.skipped) + self.skipped += len(result.skipped) + for case, tb in result.failures + result.errors: + self.failed += 1 + self.failures.append(f"{rel}::{case.id()}\n{tb[-1500:]}") + elif isinstance(obj, type) and name.startswith("Test") and "__init__" not in vars(obj): + for meth in [m for m in dir(obj) if m.startswith("test")]: + inst = obj() + if hasattr(inst, "setup_method"): + inst.setup_method(getattr(inst, meth)) + try: + self._expand(f"{rel}::{name}::{meth}", getattr(inst, meth), fixtures) + finally: + if hasattr(inst, "teardown_method"): + inst.teardown_method(getattr(inst, meth)) + elif (name.startswith("test") and inspect.isfunction(obj) + and obj.__module__ == module.__name__): + self._expand(f"{rel}::{name}", obj, fixtures) + + +def _slug(path: Any) -> str: + return re.sub(r"[^0-9A-Za-z]+", "_", str(path)) + + +def _load(path: Path, name: str) -> types.ModuleType: + spec = importlib.util.spec_from_file_location(name, path) + if spec is None or spec.loader is None: + raise ImportError(f"cannot load {path}") + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +def _fixtures_of(module: types.ModuleType) -> dict: + return {n: f for n, f in vars(module).items() if callable(f) and getattr(f, "_cf_fixture", False)} + + +def run(root: Path, suites: Iterable[str], *, sys_paths: Iterable[str] = (), + stub_modules: Iterable[str] = (), budget_s: float = 75.0, min_passed: int = 1, + conftest: bool = True) -> int: + """Run ``suites`` (paths relative to ``root``); return a process exit code (0 = PASS).""" + import json + + start = time.monotonic() + sys.dont_write_bytecode = True # read-only checkout + for entry in reversed(list(sys_paths)): + sys.path.insert(0, str(root / entry)) + shim = importlib.util.find_spec("pytest") is None + if shim: + sys.modules["pytest"] = pytest_module() + stubs = install_stubs(stub_modules) + suites = list(suites) + missing = [s for s in suites if not (root / s).is_file()] + if missing: + print("CERTFORGE_TESTKIT_FAILED: missing suites: " + ", ".join(missing), file=sys.stderr) + return 1 + session = _Session(root, start + budget_s) + session.use_conftest = conftest + aborted = None + for suite in suites: + try: + session.run_suite(root / suite) + except BudgetExceeded as exc: + aborted = str(exc) + break + except BaseException as exc: # noqa: BLE001 - an import/collection error fails the suite + session.failed += 1 + session.failures.append(f"{suite}: collection error {type(exc).__name__}: {exc}\n" + + traceback.format_exc(limit=6)[-1500:]) + for failure in session.failures[:20]: + print("FAIL " + failure, file=sys.stderr) + report = { + "suites": len(suites), "passed": session.passed, "failed": session.failed, + "skipped": session.skipped, "pytest_shim": shim, "stubbed_modules": stubs, + "elapsed_s": round(time.monotonic() - start, 2), "aborted": aborted, + } + ok = aborted is None and session.failed == 0 and session.passed >= min_passed + print(("CERTFORGE_TESTKIT_OK " if ok else "CERTFORGE_TESTKIT_FAILED ") + + json.dumps(report, sort_keys=True)) + return 0 if ok else 1 + + +def run_isolated(root: Path, suites: Iterable[dict], *, budget_s: float = 60.0, + per_suite_timeout_s: float = 30.0, min_passed: int = 1) -> int: + """Run each suite in its own interpreter (no module-name collisions between suites). + + ``suites``: dicts with ``path`` plus optional ``sys_paths`` / ``stub_modules``. + """ + import json + import subprocess + + start = time.monotonic() + suites = list(suites) + totals = {"passed": 0, "failed": 0, "skipped": 0} + failed_suites: list[str] = [] + for spec in suites: + remaining = budget_s - (time.monotonic() - start) + if remaining <= 1: + failed_suites.append(f"{spec['path']}: journey time budget exhausted") + break + argv = [sys.executable, "-B", str(Path(__file__).resolve()), "--root", str(root), + "--suite", spec["path"]] + for entry in spec.get("sys_paths", []): + argv += ["--sys-path", entry] + for name in spec.get("stub_modules", []): + argv += ["--stub", name] + try: + proc = subprocess.run(argv, cwd=str(root), capture_output=True, text=True, + timeout=min(per_suite_timeout_s, remaining), check=False) + except subprocess.TimeoutExpired: + failed_suites.append(f"{spec['path']}: timed out") + continue + line = next((ln for ln in reversed(proc.stdout.splitlines()) + if ln.startswith("CERTFORGE_TESTKIT_")), "") + try: + report = json.loads(line.split(" ", 1)[1]) + except (IndexError, ValueError): + report = {} + for key in totals: + totals[key] += int(report.get(key, 0) or 0) + status = "ok" if proc.returncode == 0 else "FAILED" + print(f"suite {status} {spec['path']} passed={report.get('passed')} " + f"failed={report.get('failed')} skipped={report.get('skipped')} " + f"elapsed_s={report.get('elapsed_s')}") + if proc.returncode != 0: + failed_suites.append(spec["path"]) + sys.stderr.write(proc.stderr[-3000:]) + ok = not failed_suites and totals["failed"] == 0 and totals["passed"] >= min_passed + summary = dict(totals, suites=len(suites), failed_suites=failed_suites, + elapsed_s=round(time.monotonic() - start, 2)) + print(("CERTFORGE_SUITES_OK " if ok else "CERTFORGE_SUITES_FAILED ") + + json.dumps(summary, sort_keys=True)) + return 0 if ok else 1 + + +if __name__ == "__main__": # single-suite worker used by run_isolated + import argparse + + parser = argparse.ArgumentParser() + parser.add_argument("--root", required=True) + parser.add_argument("--suite", required=True) + parser.add_argument("--sys-path", action="append", default=[]) + parser.add_argument("--stub", action="append", default=[]) + parser.add_argument("--budget", type=float, default=60.0) + ns = parser.parse_args() + raise SystemExit(run(Path(ns.root), [ns.suite], sys_paths=ns.sys_path, + stub_modules=ns.stub, budget_s=ns.budget)) diff --git a/scripts/collect_finisher_e2e.py b/scripts/collect_finisher_e2e.py new file mode 100644 index 0000000..c82e955 --- /dev/null +++ b/scripts/collect_finisher_e2e.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +"""Run the owner-admitted Finisher collector; output is never a signed verdict.""" +from __future__ import annotations + +import argparse +import hashlib +import json +import logging +import os +from pathlib import Path +import stat +import sys + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) +from echo_certification_forge.finisher_collector import ( # noqa: E402 + CollectorError, FinisherCollector, PinnedOwnerProbe, bounded_document, require, +) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--policy", type=Path, required=True) + parser.add_argument("--policy-sha256", required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + logging.basicConfig(level=logging.INFO, format="%(message)s") + try: + require(os.name == "posix", "collector_platform_unsupported") + require(args.policy.is_absolute() and not args.policy.is_symlink(), "policy_path_invalid") + descriptor = os.open(args.policy, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(descriptor, "rb") as handle: + metadata = os.fstat(handle.fileno()) + require(stat.S_ISREG(metadata.st_mode) and metadata.st_size <= 65536, "policy_file_invalid") + require(metadata.st_uid in (0, os.geteuid()) and not metadata.st_mode & 0o022, "policy_file_permissions") + raw = handle.read(65537) + require(hashlib.sha256(raw).hexdigest() == args.policy_sha256, "policy_pin_mismatch") + policy = bounded_document(raw, 65536) + probe = policy.get("probe") + require(isinstance(probe, dict), "owner_probe_unconfigured") + require(args.output.is_absolute() and not args.output.exists() and not args.output.is_symlink(), "output_path_invalid") + parent = args.output.parent + require(parent.resolve(strict=True) == parent and parent.is_dir(), "output_parent_invalid") + metadata = parent.stat() + require(metadata.st_uid == os.geteuid() and not metadata.st_mode & 0o077, "output_parent_not_private") + owner = PinnedOwnerProbe(probe.get("command"), probe.get("sha256"), + dependencies=probe.get("dependencies"), timeout=probe.get("timeout", 120)) + result = FinisherCollector(policy, owner).collect() + data = (json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n").encode() + descriptor = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + with os.fdopen(descriptor, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + print(json.dumps({"status": "UNSIGNED_CANDIDATE", "release_verdict": "NOT_READY", + "signature_verified": False, "sha256": hashlib.sha256(data).hexdigest(), + "observations": len(result["evidence"])})) + return 0 + except CollectorError as exc: + print(json.dumps({"status": "HELD", "release_verdict": "NOT_READY", "reason": str(exc)})) + return 2 + except (OSError, ValueError, TypeError, KeyError): + print(json.dumps({"status": "HELD", "release_verdict": "NOT_READY", "reason": "collector_input_or_io_failed"})) + return 3 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/p4_hostile_acceptance.py b/scripts/p4_hostile_acceptance.py index b17dd0e..f05e19c 100644 --- a/scripts/p4_hostile_acceptance.py +++ b/scripts/p4_hostile_acceptance.py @@ -136,6 +136,25 @@ def write_json(path: Path, value: Any) -> None: path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") +def write_acceptance_report(path: Path, report: dict[str, Any]) -> None: + """Publish phase completion only after checks and final cleanup succeed.""" + report.pop("completed_phase_gate", None) + checks = report.get("checks") + cleanup = report.get("cleanup") or {} + if ( + report.get("phase") == "P4" + and report.get("passed") is True + and report.get("run_outcome") == "COMPLETE" + and isinstance(checks, dict) + and checks + and all(value is True for value in checks.values()) + and cleanup.get("unrelated_container_ids_preserved") is True + and cleanup.get("ephemeral_private_files_removed") is True + ): + report["completed_phase_gate"] = "P4" + write_json(path, report) + + def docker_json(*arguments: str) -> Any: return json.loads(run(["docker", *arguments]).stdout) @@ -1375,6 +1394,10 @@ def service_health_probe(image: str, role: ImageRole, ownership_token: str, work "ECHO_CERTFORGE_DB": "/workspace/state/certforge.sqlite3", "ECHO_CERTFORGE_EVIDENCE_ROOT": "/workspace/state/evidence", "ECHO_CERTFORGE_TRUSTED_KEYS": "/workspace/state/trusted-public-keys", + # The worker app opens its P6 deployment ledger at import time. The default + # (/var/deployments.sqlite3) is on the read-only image root, so + # the ledger joins the other state paths on the writable workspace. + "ECHO_CERTFORGE_DEPLOYMENT_LEDGER": "/workspace/state/deployments.sqlite3", } ) command = ["-lc", "mkdir -p /workspace/state/trusted-public-keys /workspace/state/evidence && exec uvicorn echo_certification_forge.app:app --host 127.0.0.1 --port 8080 --no-access-log"] @@ -1859,7 +1882,7 @@ def main() -> int: report["error"] = {"type": "CleanupError", "message": "unrelated container identity changed"} return_code = 1 report["completed_at_utc"] = to_utc_iso(datetime.now(UTC)) - write_json(output, report) + write_acceptance_report(output, report) print( json.dumps( { diff --git a/scripts/p4_hostile_acceptance_exact.py b/scripts/p4_hostile_acceptance_exact.py index 0d6d61c..732ad99 100644 --- a/scripts/p4_hostile_acceptance_exact.py +++ b/scripts/p4_hostile_acceptance_exact.py @@ -147,6 +147,10 @@ def enrich_report( if not output_path.is_file(): raise RuntimeError(f"P4 harness did not produce an output report: {output_path}") report = json.loads(output_path.read_text(encoding="utf-8")) + # Enrichment is a further acceptance boundary; invalidate any earlier marker + # before parsing captured evidence, which can raise or be interrupted. + report.pop("completed_phase_gate", None) + output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") report["harness"] = { "path": str(HERE), "sha256": sha256_file(HERE), @@ -205,6 +209,10 @@ def enrich_report( ) return_code = 1 report["wrapper_return_code"] = return_code + if return_code != 0: + report["passed"] = False + report["run_outcome"] = "INFRA_FAILED" + report["release_verdict"] = "NOT_READY" output_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") return report @@ -213,6 +221,15 @@ def main(arguments: list[str] | None = None) -> int: argv = list(sys.argv[1:] if arguments is None else arguments) sealed_manifest_path, output_path = parse_wrapper_paths(argv) harness = load_harness() + publish_report = harness.write_acceptance_report + + def write_pending_report(path: Path, report: dict[str, Any]) -> None: + report.pop("completed_phase_gate", None) + harness.write_json(path, report) + + # The core can finish before the exact verifier evidence is enriched. + # Keep its on-disk result pending until every wrapper layer has returned. + harness.write_acceptance_report = write_pending_report artifacts = resolve_runner_artifacts(harness, sealed_manifest_path) capture: dict[str, Any] = {} harness.execute_container = make_execute_container_override( @@ -234,6 +251,7 @@ def main(arguments: list[str] | None = None) -> int: return_code=return_code, ) final_code = 0 if report.get("passed") is True and report.get("run_outcome") == "COMPLETE" else 1 + publish_report(output_path, report) print( json.dumps( { diff --git a/scripts/run_p4_gate.sh b/scripts/run_p4_gate.sh index 046b33b..21bce96 100644 --- a/scripts/run_p4_gate.sh +++ b/scripts/run_p4_gate.sh @@ -25,7 +25,7 @@ set -Eeuo pipefail REPO_URL="${REPO_URL:-https://github.com/ECHO-OMEGA-PRIME/echo-certification-forge}" BRANCH="${BRANCH:-feat/certforge-r5-negative-controls}" GIT_CRED_FILE="${GIT_CRED_FILE:-/home/forge/.config/echo/omega_git_creds}" -BASE_DIGEST="${BASE_DIGEST:-sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df}" +BASE_DIGEST="${BASE_DIGEST:-sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111}" CLAMAV_IMAGE="${CLAMAV_IMAGE:-clamav/clamav@sha256:7f5389ccaa2368c383fa80e167ccfe44348d71e685f926fce4755eed1757673a}" COSIGN="${COSIGN:-/home/forge/.cache/echo-certforge/p4-9c07eb7/tools/cosign/cosign}" TRIVY="${TRIVY:-/home/forge/.cache/echo-certforge/p4-9c07eb7/tools/trivy/trivy}" diff --git a/src/echo_certification_forge/finisher_collector.py b/src/echo_certification_forge/finisher_collector.py new file mode 100644 index 0000000..90cbf90 --- /dev/null +++ b/src/echo_certification_forge/finisher_collector.py @@ -0,0 +1,482 @@ +"""Finisher observation collection through a separately admitted owner probe. + +This module never signs, deploys, enrolls a worker or changes a release policy. +The owner must independently review/admit the probe and pin its entire closure. +Successful collection produces an UNSIGNED_CANDIDATE, always NOT_READY. +""" +from __future__ import annotations + +from collections.abc import Callable, Mapping +from datetime import UTC, datetime, timedelta +import hashlib +import json +import logging +import os +from pathlib import Path +import re +import selectors +import signal +import subprocess +import time +from typing import Any, Protocol +from urllib.parse import urlsplit +import uuid + +from .canonical import sha256_json +from .production_e2e import BASE_CHECKS, GENERIC_PROFILE, SCHEMA_VERSION, _contains_restricted_value + +FRONTS = {"auto": "claude", "cauto": "codex", "gauto": "gemini", "grauto": "grok", + "CFauto": "claude", "CFcauto": "codex", "CFgauto": "gemini", "CFgrauto": "grok"} +NEGATIVE_CONTROLS = ("unauthenticated", "wrong_owner", "wrong_installation", "wrong_source", + "invalid_token", "expired_token", "stale_control", "replayed_control", + "out_of_scope", "redirect", "forbidden_native_route") +CONTROL_STATES = {"pause": "paused", "resume": "running", "halt": "halted"} +IDENTIFIER = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}\Z") +DIGEST = re.compile(r"[0-9a-f]{64}\Z") +COMMIT = re.compile(r"[0-9a-f]{40}\Z") +LOG = logging.getLogger(__name__) + + +class CollectorError(RuntimeError): + """A fixed nonsecret reason code; raw probe responses are never interpolated.""" + + +class OwnerProbe(Protocol): + def call(self, request: dict[str, Any]) -> dict[str, Any]: ... + + +def require(condition: bool, reason: str) -> None: + if not condition: + raise CollectorError(reason) + + +def identifier(value: Any) -> bool: + return isinstance(value, str) and IDENTIFIER.fullmatch(value) is not None + + +def digest(value: Any) -> bool: + return isinstance(value, str) and DIGEST.fullmatch(value) is not None + + +def timestamp(value: Any) -> datetime: + require(isinstance(value, str) and len(value) <= 64, "time_invalid") + try: + result = datetime.fromisoformat(value.replace("Z", "+00:00")) + require(result.tzinfo is not None, "time_invalid") + return result.astimezone(UTC) + except (ValueError, OverflowError) as exc: + raise CollectorError("time_invalid") from exc + + +def bounded_document(data: bytes, limit: int = 131072) -> dict[str, Any]: + require(len(data) <= limit, "document_size") + def pairs(items): + result = {} + for key, value in items: + require(key not in result, "document_duplicate_key") + result[key] = value + return result + try: + result = json.loads(data, object_pairs_hook=pairs, + parse_constant=lambda _: (_ for _ in ()).throw(CollectorError("document_number"))) + except (ValueError, RecursionError, UnicodeError) as exc: + raise CollectorError("document_invalid") from exc + require(isinstance(result, dict), "document_invalid") + try: + require(not _contains_restricted_value(result), "restricted_content") + except RecursionError as exc: + raise CollectorError("document_depth") from exc + return result + + +class PinnedOwnerProbe: + """Bounded Linux process transport for a previously reviewed owner probe. + + Hashes are integrity pins, NOT administrative admission. The probe must query + the real authenticated control plane for admission and every observation. + No command, shell text, interpreter switch or provider argument is accepted + from an observation. Dependencies must be pinned by the owner as well. + """ + def __init__(self, command: list[str], pins: list[str], *, + dependencies: Mapping[str, str] | None = None, timeout: float = 120): + require(isinstance(command, list) and len(command) == 2 and isinstance(pins, list) and len(pins) == 2, + "probe_command_invalid") + require(type(timeout) in (int, float) and 1 <= timeout <= 120, "probe_timeout_invalid") + try: + self.command = [str(Path(item).resolve(strict=True)) for item in command] + except (OSError, ValueError, TypeError) as exc: + raise CollectorError("probe_command_invalid") from exc + self.pins = dict(zip(self.command, pins, strict=True)) + for path, pin in (dependencies or {}).items(): + resolved = str(Path(path).resolve(strict=True)) + require(resolved not in self.pins, "probe_pin_duplicate") + self.pins[resolved] = pin + self.timeout = timeout + self._verify() + + def _verify(self) -> None: + for name, expected in self.pins.items(): + require(digest(expected), "probe_pin_invalid") + path = Path(name) + require(path.is_file() and not path.is_symlink(), "probe_pin_file") + value = hashlib.sha256() + with path.open("rb") as handle: + for block in iter(lambda: handle.read(65536), b""): + value.update(block) + require(value.hexdigest() == expected, "probe_pin_mismatch") + + def call(self, request: dict[str, Any]) -> dict[str, Any]: + self._verify() + require(os.name == "posix", "probe_platform_unsupported") + data = json.dumps(request, sort_keys=True, separators=(",", ":")).encode() + require(len(data) <= 8192, "probe_request_size") + # Credentials remain opaque in the inherited normal scoped broker path. + allowed = {"PATH", "HOME", "LANG", "LC_ALL", "SSL_CERT_FILE", "SSL_CERT_DIR", + "SOL_BROKER_TOKEN", "SOL_RUN_ID", "SOL_STATE", "SOL_STATE_DB", "SOL_ROLE"} + environment = {name: value for name, value in os.environ.items() if name in allowed} + environment["PYTHONDONTWRITEBYTECODE"] = "1" + deadline = time.monotonic() + self.timeout + process = subprocess.Popen([self.command[0], "-I", self.command[1]], stdin=subprocess.PIPE, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + env=environment, start_new_session=True, close_fds=True, bufsize=0) + selector = selectors.DefaultSelector() + output = bytearray() + try: + # Both directions share one deadline, including a child that never + # reads stdin. No assumption about the actual pipe capacity is made. + os.set_blocking(process.stdin.fileno(), False) + os.set_blocking(process.stdout.fileno(), False) + selector.register(process.stdin, selectors.EVENT_WRITE) + selector.register(process.stdout, selectors.EVENT_READ) + sent = 0 + while selector.get_map(): + require(time.monotonic() < deadline, "probe_timeout") + for key, _ in selector.select(min(.1, max(0, deadline - time.monotonic()))): + try: + if key.fileobj is process.stdin: + sent += os.write(process.stdin.fileno(), data[sent:sent + 4096]) + if sent == len(data): + selector.unregister(process.stdin) + process.stdin.close() + continue + block = os.read(key.fileobj.fileno(), 16384) + except BlockingIOError: + continue + if not block: + selector.unregister(key.fileobj) + continue + output.extend(block) + require(len(output) <= 131072, "probe_output_size") + try: + process.wait(timeout=max(.01, deadline - time.monotonic())) + except subprocess.TimeoutExpired as exc: + raise CollectorError("probe_timeout") from exc + require(process.returncode == 0, "probe_failed") + self._verify() + return bounded_document(bytes(output)) + except (OSError, subprocess.SubprocessError) as exc: + raise CollectorError("probe_transport_failed") from exc + finally: + selector.close() + # The probe owns this process group, including any unsettled children. + try: + os.killpg(process.pid, signal.SIGKILL) + except ProcessLookupError: + pass + process.wait(timeout=5) + process.stdout.close() + process.stdin.close() + + +class FinisherCollector: + def __init__(self, policy: dict[str, Any], probe: OwnerProbe, *, + clock: Callable[[], datetime] = lambda: datetime.now(UTC)): + self.policy = bounded_document(json.dumps(policy, allow_nan=False).encode(), 65536) + self.probe, self.clock = probe, clock + p = self.policy + require(p.get("schema") == "echo.finisher.collector-policy.v1", "policy_schema") + require(all(identifier(p.get(key)) for key in + ("tenant_id", "collector_id", "authority_id", "execution_id", "admission_id")), "policy_identity") + bindings = p.get("bindings") + require(isinstance(bindings, dict), "policy_bindings") + expected_keys = {"target_identity_digest", "environment_identity_digest", "source_commit", + "native_source_commit", "transport_source_commit", "artifact_sha256", + "native_artifact_sha256", "image_digest", "installation_id", "owner_id", "cloudflare_origin", + "account_id", "application_id", "durable_object_id"} + require(set(bindings) == expected_keys, "policy_bindings") + require(all(digest(bindings[key]) for key in + ("target_identity_digest", "environment_identity_digest", "artifact_sha256", "native_artifact_sha256")), "policy_digest") + require(all(isinstance(bindings[key], str) and COMMIT.fullmatch(bindings[key]) for key in + ("source_commit", "native_source_commit", "transport_source_commit")), "policy_source") + require(bindings["source_commit"] in (bindings["native_source_commit"], bindings["transport_source_commit"]), "policy_source") + require(isinstance(bindings["image_digest"], str) and bindings["image_digest"].startswith("sha256:") + and digest(bindings["image_digest"][7:]), "policy_image") + require(identifier(bindings["installation_id"]) and identifier(bindings["owner_id"]), "policy_installation") + require(isinstance(bindings["account_id"], str) and re.fullmatch("[a-f0-9]{32}", bindings["account_id"]) + and digest(bindings["durable_object_id"]) and isinstance(bindings["application_id"], str) + and re.fullmatch("[A-Za-z0-9_-]{1,128}", bindings["application_id"]), "policy_cloudflare_identity") + origin = urlsplit(bindings["cloudflare_origin"]) + require(origin.scheme == "https" and bool(origin.hostname) and not origin.username and not origin.password + and origin.path in ("", "/") and not origin.query and not origin.fragment, "policy_origin") + require(digest(p.get("rollback_artifact_sha256")) and p["rollback_artifact_sha256"] != bindings["artifact_sha256"], "policy_rollback") + require(isinstance(p.get("rollback_source_commit"), str) + and COMMIT.fullmatch(p["rollback_source_commit"]) + and p["rollback_source_commit"] != bindings["source_commit"], "policy_rollback") + # The approved immutable baseline and recovered candidate must pin their + # actual checkpoint/config identities before any rollback evidence is read. + require(all(digest(p.get(key)) for key in + ("rollback_checkpoint_sha256", "rollback_configuration_sha256", + "recovered_checkpoint_sha256", "recovered_configuration_sha256")), "policy_rollback_identity") + require(isinstance(p.get("providers"), dict) and set(p["providers"]) == set(FRONTS.values()), "policy_providers") + for provider in p["providers"].values(): + require(isinstance(provider, dict) and set(provider) == {"model", "account_sha256"} + and identifier(provider["model"]) and digest(provider["account_sha256"]), "policy_provider") + self.policy_hash = sha256_json(p) + self.challenge = uuid.uuid4().hex + self.started = self.clock() + require(self.started.tzinfo is not None, "clock_invalid") + self.ids: set[str] = set() + self.invocations: set[str] = set() + self.control_ids: set[str] = set() + self.native_control_ids: set[str] = set() + self.restoration_ids: set[str] = set() + self.evidence: list[dict[str, str]] = [] + + def _query(self, operation: str, **extra: Any) -> dict[str, Any]: + now = self.clock() + require(0 <= (now - self.started).total_seconds() <= 1800, "collection_deadline") + request = {"schema": "echo.finisher.owner-query.v1", "operation": operation, + "nonce": uuid.uuid4().hex, "policy_sha256": self.policy_hash, + **{key: self.policy[key] for key in + ("admission_id", "execution_id", "tenant_id", "collector_id", "bindings")}, **extra} + LOG.info(json.dumps({"event": "owner_query", "operation": operation, "policy_sha256": self.policy_hash})) + response = self.probe.call(request) + require(isinstance(response, dict), "owner_response_invalid") + response = bounded_document(json.dumps(response, allow_nan=False).encode()) + require(response.get("schema") == "echo.finisher.owner-response.v1" + and response.get("nonce") == request["nonce"] + and response.get("request_sha256") == sha256_json(request) + and response.get("authority_id") == self.policy["authority_id"] + and response.get("authenticated") is True and response.get("authorized") is True, + "owner_response_binding") + observed = timestamp(response.get("observed_at")) + require(now - timedelta(seconds=1) <= observed <= self.clock() + timedelta(seconds=1), "owner_response_time") + require(isinstance(response.get("result"), dict), "owner_response_result") + return response["result"] + + def _admission(self) -> None: + result = self._query("admission") + require(result.get("allowed") is True and result.get("independent") is True, "admission_denied") + require(all(result.get(key) == self.policy[key] for key in + ("admission_id", "collector_id", "execution_id", "tenant_id", "bindings")), "admission_binding") + require(result.get("policy_sha256") == self.policy_hash and result.get("restoration_allowed") is True, + "admission_journey_not_allowed") + expiry = timestamp(result.get("expires_at")) + require(self.clock() < expiry <= self.clock() + timedelta(hours=1), "admission_expired") + require(result.get("operations") == ["observe", "recall"], "admission_operations") + + def _observe(self, kind: str, name: str) -> tuple[dict[str, Any], datetime]: + self._admission() + produced = self._query("observe", challenge=self.challenge, kind=kind, name=name) + receipt_id = produced.get("receipt_id") + require(identifier(receipt_id) and receipt_id not in self.ids, "receipt_replay") + self.ids.add(receipt_id) + self._admission() + recalled = self._query("recall", receipt_id=receipt_id) + content = recalled.get("content") + require(recalled.get("receipt_id") == receipt_id and isinstance(content, dict), "receipt_readback") + require(digest(recalled.get("content_sha256")) and recalled["content_sha256"] == sha256_json(content), "receipt_hash") + require(content.get("schema") == "echo.finisher.observation.v1" + and content.get("receipt_id") == receipt_id and content.get("mode") == "live" + and content.get("challenge") == self.challenge and content.get("kind") == kind + and content.get("name") == name, "observation_binding") + require(all(content.get(key) == self.policy[key] for key in + ("tenant_id", "execution_id", "collector_id", "bindings")), "observation_identity") + observed = timestamp(content.get("observed_at")) + require(self.started <= observed <= self.clock() and (self.clock() - observed).total_seconds() <= 900, "observation_time") + result = content.get("result") + require(isinstance(result, dict), "observation_result") + self.evidence.append({"kind": kind, "name": name, "receipt_id": receipt_id, + "sha256": recalled["content_sha256"], "observed_at": observed.isoformat()}) + LOG.info(json.dumps({"event": "observation_recalled", "kind": kind, "name": name, + "receipt_id": receipt_id, "sha256": recalled["content_sha256"]})) + return result, observed + + def _target(self, target: Any, run_id: str | None = None) -> None: + binding = self.policy["bindings"] + expected = {"provider": "cloudflare", **{key: binding[key] for key in + ("account_id", "application_id", "durable_object_id", "installation_id")}, + "source_sha": binding["transport_source_commit"]} + require(isinstance(target, dict) and set(target) == {*expected, "run_id"} + and all(target.get(key) == value for key, value in expected.items()) + and identifier(target.get("run_id")) and (run_id is None or target["run_id"] == run_id), + "snapshot_target") + + def _observation_time(self, stamp: Any) -> datetime: + if type(stamp) in (int, float): + require(0 <= stamp <= 253402300799, "status_snapshot_time") + stamp = datetime.fromtimestamp(stamp, UTC) + else: + stamp = timestamp(stamp) + require(self.started <= stamp <= self.clock() and (self.clock() - stamp).total_seconds() <= 30, + "status_snapshot_time") + return stamp + + def _snapshot(self, value: Any, run_id: str | None = None) -> None: + require(isinstance(value, dict) and value.get("schema") == "echo.finisher.ui.snapshot.v1" + and not any(key in value for key in ("accepted", "applied", "request_id")), "status_snapshot_invalid") + self._target(value.get("target"), run_id) + require(value.get("source_sha") == self.policy["bindings"]["transport_source_commit"] + and value.get("online") is True and value.get("readback_verified") is True + and value.get("lifecycle") in ("online", "idle", "running", "executing"), "status_snapshot_unhealthy") + checks = value.get("checks") + require(isinstance(checks, dict) and all(checks.get(key) is True for key in + ("service_active", "installation_verified", "heartbeat_fresh", "resource_admitted")), "status_snapshot_unhealthy") + self._observation_time(value.get("observed_at")) + + def _control(self, action: str, control: Any, run_id: str) -> None: + require(isinstance(control, dict) and identifier(control.get("request_id")) + and identifier(control.get("native_request_id")) and control.get("action") == action, + "front_control_invalid") + require(control["request_id"] not in self.control_ids + and control["native_request_id"] not in self.native_control_ids, "control_replay") + self.control_ids.add(control["request_id"]) + self.native_control_ids.add(control["native_request_id"]) + self._target(control.get("target"), run_id) + require(control.get("accepted") is True and control.get("applied") is True + and control.get("state") == CONTROL_STATES[action], "front_control_not_applied") + submitted = self._observation_time(control.get("submitted_at")) + observed = control.get("readback") + require(isinstance(observed, dict) and observed.get("target") == control["target"] + and observed.get("source_sha") == self.policy["bindings"]["transport_source_commit"] + and observed.get("readback_verified") is True, "control_readback_binding") + require(self._observation_time(observed.get("observed_at")) >= submitted, "control_readback_stale") + checks, controller = observed.get("checks"), observed.get("controller") + require(isinstance(checks, dict) and checks.get("installation_verified") is True + and isinstance(controller, dict) and controller.get("last_control_id") == control["native_request_id"], + "control_readback_identity") + require(type(controller.get("paused")) is bool and type(controller.get("halted")) is bool, + "control_readback_state") + expected = {"pause": controller["paused"] and not controller["halted"], + "resume": not controller["paused"] and not controller["halted"], + "halt": controller["halted"]} + require(expected[action], "control_readback_state") + live = checks.get("service_active") is True and checks.get("heartbeat_fresh") is True + terminal = (observed.get("terminal_halt_verified") is True + and observed.get("lifecycle") == "halted" and observed.get("online") is False) + # A legitimate fresh terminal observation proves process exit through + # the owner observer. It must not fabricate a live controller heartbeat. + require(terminal if action == "halt" else live, "control_readback_unavailable") + + def _front(self, name: str, result: dict[str, Any]) -> None: + provider = FRONTS[name] + require(result.get("provider") == provider and all(result.get(key) == value for key, value in + self.policy["providers"][provider].items()), "front_provider_identity") + require(all(result.get(key) is True for key in ("model_observed", "account_verified", "subscription_auth", + "heartbeat_verified", "scoped_mcp_verified", "uncertain_replay_suppressed", "cleanup_verified")), "front_incomplete") + require(type(result.get("status_inference_count")) is int and result["status_inference_count"] == 0, "front_status_inference") + require(all(identifier(result.get(key)) for key in ("run_id", "worker_id", "session_id", + "provider_invocation_id", "hook_receipt_id", "capture_receipt_id")), "front_receipt_missing") + require(result.get("resumed_session_id") == result["session_id"], "front_resume_mismatch") + require(digest(result.get("hook_readback_sha256")) and digest(result.get("capture_readback_sha256")), "front_capture_readback") + require(result["provider_invocation_id"] not in self.invocations, "provider_replay") + self.invocations.add(result["provider_invocation_id"]) + self._snapshot(result.get("status_snapshot"), result["run_id"]) + start = result.get("start") + require(isinstance(start, dict) and start.get("state") == "singleton_observed" + and start.get("already_running") is True and start.get("readback_verified") is True, + "front_start_unverified") + self._target(start.get("target"), result["run_id"]) + controls = result.get("controls") + require(isinstance(controls, dict) and set(controls) == set(CONTROL_STATES), "front_controls_missing") + for action in CONTROL_STATES: + self._control(action, controls[action], result["run_id"]) + + def _restoration(self, name: str, previous_run_id: str) -> None: + result, _ = self._observe("restoration", name) + require(result.get("method") == "admitted_owner_operation" + and result.get("admission_id") == self.policy["admission_id"] + and result.get("previous_run_id") == previous_run_id + and identifier(result.get("operation_id")) + and result["operation_id"] not in self.restoration_ids, "restoration_unverified") + self.restoration_ids.add(result["operation_id"]) + self._snapshot(result.get("snapshot")) + + def _rollback(self) -> None: + rollback, _ = self._observe("rollback", "restore-and-recover") + require(rollback.get("verified") is True and rollback.get("restored_readback_verified") is True + and rollback.get("recovered_readback_verified") is True + and rollback.get("restored_artifact_sha256") == self.policy["rollback_artifact_sha256"] + and rollback.get("recovered_artifact_sha256") == self.policy["bindings"]["artifact_sha256"] + and rollback.get("restored_source_commit") == self.policy["rollback_source_commit"] + and rollback.get("restored_checkpoint_source_commit") == self.policy["rollback_source_commit"] + and rollback.get("recovered_source_commit") == self.policy["bindings"]["source_commit"] + and rollback.get("recovered_checkpoint_source_commit") == self.policy["bindings"]["source_commit"] + and all(rollback.get(observed) == self.policy[expected] for observed, expected in { + "restored_checkpoint_sha256": "rollback_checkpoint_sha256", + "restored_configuration_sha256": "rollback_configuration_sha256", + "recovered_checkpoint_sha256": "recovered_checkpoint_sha256", + "recovered_configuration_sha256": "recovered_configuration_sha256", + }.items()), "rollback_failed") + + def collect(self) -> dict[str, Any]: + require(not self.evidence, "collector_already_used") + self._admission() + for front in FRONTS: + result, _ = self._observe("front", front) + self._front(front, result) + # A halted singleton cannot serve the next front. Its independent + # owner must use an already admitted restoration operation; frontend + # start reconciliation itself is not a bootstrap or reset grant. + self._restoration(front, result["run_id"]) + for name in NEGATIVE_CONTROLS: + result, _ = self._observe("negative", name) + require(result.get("denied") is True and result.get("side_effects") is False + and digest(result.get("before_sha256")) and result.get("before_sha256") == result.get("after_sha256"), "negative_control_failed") + # Restore the prior compatible checkpoint/config and then the candidate + # before measuring final health. A Worker image rollback alone is not + # checkpoint restoration. + self._rollback() + probes = [] + for index in range(3): + result, observed = self._observe("stability", str(index)) + require(all(result.get(key) is True for key in ("healthy", "resource_admitted", "heartbeat_verified")) + and type(result.get("unexpected_restarts")) is int and result["unexpected_restarts"] == 0, "stability_failed") + # Aggregate health is not eight-front evidence. Each route requires + # its own fresh, owner-issued and independently recalled observation. + require("snapshots" not in result, "stability_snapshots_unadmitted") + for front, provider in FRONTS.items(): + route, _ = self._observe("stability-front", f"{index}:{front}") + snapshot = route.get("snapshot") + require(route.get("front") == front and route.get("provider") == provider + and route.get("route_readback_verified") is True + and digest(route.get("snapshot_sha256")) + and route["snapshot_sha256"] == sha256_json(snapshot), "stability_route_binding") + # The Cloudflare singleton run_id may legitimately be shared + # across fronts; identity is the independently recalled route. + self._snapshot(snapshot) + probes.append(observed) + require(probes[0] < probes[1] < probes[2] and (probes[2] - probes[0]).total_seconds() >= 30, "stability_interval") + external, _ = self._observe("external", "acceptance") + require(external.get("accepted") is True and external.get("ui_assets_verified") is True + and external.get("access_session_verified") is True + and identifier(external.get("reviewer_id")) and external["reviewer_id"] != self.policy["collector_id"] + and external.get("origin") == self.policy["bindings"]["cloudflare_origin"], "external_acceptance_failed") + self._admission() + finished = self.clock() + require((finished - self.started).total_seconds() <= 1800, "collection_deadline") + payload = {"schema_version": SCHEMA_VERSION, "profile": GENERIC_PROFILE, + "attestation_id": "finisher-" + self.challenge, + "target_identity_digest": self.policy["bindings"]["target_identity_digest"], + "environment_identity_digest": self.policy["bindings"]["environment_identity_digest"], + "source_commit": self.policy["bindings"]["source_commit"], + "deployment_sha": self.policy["bindings"]["source_commit"], + "observed_at": self.started.isoformat(), + "expires_at": (self.started + timedelta(hours=1)).isoformat(), + "required_checks": sorted(BASE_CHECKS), "checks": {key: True for key in sorted(BASE_CHECKS)}, + "stability_probe_count": len(probes), "collector_execution_id": self.policy["execution_id"], + "observation_manifest_sha256": sha256_json(self.evidence), "policy_sha256": self.policy_hash} + return {"schema": "echo.finisher.unsigned-collection.v1", "status": "UNSIGNED_CANDIDATE", + "release_verdict": "NOT_READY", "signature_verified": False, "owner_signing_required": True, + "attestation_payload": payload, "evidence": self.evidence, + "finished_at": finished.isoformat(), "bindings": self.policy["bindings"]} diff --git a/src/echo_certification_forge/sandbox.py b/src/echo_certification_forge/sandbox.py index a6ed098..96ed7f9 100644 --- a/src/echo_certification_forge/sandbox.py +++ b/src/echo_certification_forge/sandbox.py @@ -25,7 +25,7 @@ from typing import Callable # Pinned minimal Python base (same digest the P4 supply-chain pipeline pins). Override per policy. -DEFAULT_IMAGE = "python:3.12-alpine@sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" +DEFAULT_IMAGE = "python:3.12-alpine@sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" class SandboxError(RuntimeError): diff --git a/tests/test_finisher_collector.py b/tests/test_finisher_collector.py new file mode 100644 index 0000000..cd2bb37 --- /dev/null +++ b/tests/test_finisher_collector.py @@ -0,0 +1,566 @@ +"""Synthetic authority fixtures exercise rejection paths; none is live evidence.""" +from copy import deepcopy +from datetime import UTC, datetime, timedelta +import json +import hashlib +import os +import sys +import subprocess +import time +from pathlib import Path + +import pytest + +from echo_certification_forge.canonical import sha256_json +from echo_certification_forge.finisher_collector import ( + CollectorError, FRONTS, NEGATIVE_CONTROLS, FinisherCollector, PinnedOwnerProbe, +) +from echo_certification_forge.production_e2e import BASE_CHECKS, validate_attestation_trust_metadata + +NOW = datetime(2026, 11, 7, 6, 0, tzinfo=UTC) + + +def policy(): + return { + "schema": "echo.finisher.collector-policy.v1", "tenant_id": "tenant-fixture", + "collector_id": "independent-fixture", "authority_id": "owner-fixture", + "execution_id": "execution-fixture", "admission_id": "admission-fixture", + "bindings": {"target_identity_digest": "1" * 64, + "environment_identity_digest": "2" * 64, + "source_commit": "3" * 40, "native_source_commit": "4" * 40, + "transport_source_commit": "3" * 40, "artifact_sha256": "5" * 64, + "image_digest": "sha256:" + "6" * 64, + "native_artifact_sha256": "7" * 64, + "installation_id": "installation-fixture", "owner_id": "owner-fixture", + "account_id": "a" * 32, "application_id": "application-fixture", + "durable_object_id": "b" * 64, + "cloudflare_origin": "https://finisher.example.test"}, + "rollback_artifact_sha256": "8" * 64, + "rollback_source_commit": "9" * 40, + "rollback_checkpoint_sha256": "1" * 64, + "rollback_configuration_sha256": "3" * 64, + "recovered_checkpoint_sha256": "2" * 64, + "recovered_configuration_sha256": "4" * 64, + "providers": {v: {"model": v + "-fixture-model", "account_sha256": "a" * 64} + for v in set(FRONTS.values())}, + } + + +class OwnerFixture: + """In-memory test authority, not a provider, deployment or signing service.""" + def __init__(self, config): + self.config = deepcopy(config) + self.time = NOW + self.receipts = {} + self.calls = [] + self.modify = lambda value: value + self.admitted = True + self.nonce_override = None + self.shared_run_id = False + + def run_id(self, name): + return "run-shared-singleton" if self.shared_run_id else "run-" + name + + def clock(self): + return self.time + + def snapshot(self, name): + binding = self.config["bindings"] + return {"schema": "echo.finisher.ui.snapshot.v1", "observed_at": self.time.isoformat(), + "source_sha": binding["transport_source_commit"], "online": True, + "lifecycle": "idle", "readback_verified": True, + "checks": {key: True for key in ("installation_verified", "service_active", + "heartbeat_fresh", "resource_admitted")}, + "target": {"provider": "cloudflare", **{key: binding[key] for key in + ("account_id", "application_id", "durable_object_id", "installation_id")}, + "source_sha": binding["transport_source_commit"], "run_id": self.run_id(name)}} + + def call(self, request): + self.calls.append(deepcopy(request)) + self.time += timedelta(seconds=1) + op = request["operation"] + common = {"schema": "echo.finisher.owner-response.v1", "nonce": request["nonce"], + "request_sha256": sha256_json(request), "authority_id": self.config["authority_id"], + "authenticated": True, "authorized": True, + "observed_at": self.time.isoformat()} + if self.nonce_override: + common["nonce"] = self.nonce_override + if op == "admission": + value = {"allowed": self.admitted, "admission_id": self.config["admission_id"], + "collector_id": self.config["collector_id"], + "execution_id": self.config["execution_id"], "tenant_id": self.config["tenant_id"], + "bindings": self.config["bindings"], "expires_at": (self.time + timedelta(minutes=15)).isoformat(), + "operations": ["observe", "recall"], "independent": True, + "policy_sha256": request["policy_sha256"], "restoration_allowed": True} + elif op == "observe": + kind, name = request["kind"], request["name"] + if kind == "stability": + self.time += timedelta(seconds=15) + receipt_id = "receipt-" + str(len(self.receipts)) + value = {"receipt_id": receipt_id} + data = {"schema": "echo.finisher.observation.v1", "receipt_id": receipt_id, + "tenant_id": self.config["tenant_id"], "execution_id": self.config["execution_id"], + "collector_id": self.config["collector_id"], "bindings": deepcopy(self.config["bindings"]), + "challenge": request["challenge"], "kind": kind, "name": name, + "observed_at": self.time.isoformat(), "mode": "live"} + if kind == "front": + provider = FRONTS[name] + data["result"] = { + "provider": provider, **self.config["providers"][provider], + "provider_invocation_id": "invoke-" + name, + "model_observed": True, "account_verified": True, "subscription_auth": True, + "run_id": self.run_id(name), "worker_id": "worker-" + name, + "session_id": "session-" + name, "resumed_session_id": "session-" + name, + "heartbeat_verified": True, "scoped_mcp_verified": True, + "hook_receipt_id": "hook-" + name, "capture_receipt_id": "capture-" + name, + "hook_readback_sha256": "b" * 64, "capture_readback_sha256": "c" * 64, + "uncertain_replay_suppressed": True, "status_inference_count": 0, + "status_snapshot": self.snapshot(name), + "start": {"state": "singleton_observed", "already_running": True, + "readback_verified": True, "target": self.snapshot(name)["target"]}, + "controls": {action: {"request_id": name + "-" + action, "accepted": True, + "applied": True, "state": state, "action": action, + "native_request_id": "native-" + name + "-" + action, + "submitted_at": self.time.isoformat(), + "target": self.snapshot(name)["target"], + "readback": {"observed_at": self.time.isoformat(), + "target": self.snapshot(name)["target"], + "source_sha": self.config["bindings"]["transport_source_commit"], + "readback_verified": True, "terminal_halt_verified": action == "halt", + "lifecycle": "halted" if action == "halt" else "running", + "online": action != "halt", + "checks": {"installation_verified": True, + "service_active": action != "halt", "heartbeat_fresh": action != "halt"}, + "controller": {"paused": action == "pause", "halted": action == "halt", + "last_control_id": "native-" + name + "-" + action}}} + for action, state in {"pause": "paused", "resume": "running", "halt": "halted"}.items()}, + "cleanup_verified": True, + } + elif kind == "restoration": + data["result"] = {"method": "admitted_owner_operation", "operation_id": "restore-" + name, + "admission_id": self.config["admission_id"], + "previous_run_id": self.run_id(name), "snapshot": self.snapshot(name)} + elif kind == "negative": + data["result"] = {"denied": True, "side_effects": False, + "before_sha256": "d" * 64, "after_sha256": "d" * 64} + elif kind == "stability": + data["result"] = {"healthy": True, "resource_admitted": True, + "heartbeat_verified": True, "unexpected_restarts": 0} + elif kind == "stability-front": + probe_index, front = name.split(":", 1) + assert probe_index in ("0", "1", "2") + snapshot = self.snapshot(front) + data["result"] = {"front": front, "provider": FRONTS[front], + "route_readback_verified": True, "snapshot": snapshot, + "snapshot_sha256": sha256_json(snapshot)} + elif kind == "external": + data["result"] = {"accepted": True, "reviewer_id": "other-reviewer", + "ui_assets_verified": True, "access_session_verified": True, + "origin": self.config["bindings"]["cloudflare_origin"]} + elif kind == "rollback": + data["result"] = {"restored_artifact_sha256": self.config["rollback_artifact_sha256"], + "recovered_artifact_sha256": self.config["bindings"]["artifact_sha256"], + "restored_source_commit": self.config["rollback_source_commit"], + "recovered_source_commit": self.config["bindings"]["source_commit"], + "restored_checkpoint_source_commit": self.config["rollback_source_commit"], + "recovered_checkpoint_source_commit": self.config["bindings"]["source_commit"], + "restored_checkpoint_sha256": "1" * 64, + "recovered_checkpoint_sha256": "2" * 64, + "restored_configuration_sha256": "3" * 64, + "recovered_configuration_sha256": "4" * 64, + "restored_readback_verified": True, "recovered_readback_verified": True, + "verified": True} + self.receipts[receipt_id] = self.modify(data) + else: + data = deepcopy(self.receipts[request["receipt_id"]]) + value = {"receipt_id": request["receipt_id"], "content": data, "content_sha256": sha256_json(data)} + return {**common, "result": value} + + +def collect(config=None, fixture=None): + config = config or policy() + fixture = fixture or OwnerFixture(config) + return FinisherCollector(config, fixture, clock=fixture.clock).collect() + + +def test_all_eight_fronts_are_observed_then_independently_recalled(): + config = policy() + owner = OwnerFixture(config) + result = collect(config, owner) + assert result["status"] == "UNSIGNED_CANDIDATE" + assert result["release_verdict"] == "NOT_READY" + assert result["signature_verified"] is False + assert result["owner_signing_required"] is True + assert "signing_key_id" not in result["attestation_payload"] + assert validate_attestation_trust_metadata(result["attestation_payload"]) == (False, "production_e2e_signature_not_verified") + assert set(result["attestation_payload"]["checks"]) == BASE_CHECKS + assert {x["name"] for x in owner.calls if x["operation"] == "observe" and x["kind"] == "front"} == set(FRONTS) + observed = [x for x in owner.calls if x["operation"] == "observe"] + recalled = [x for x in owner.calls if x["operation"] == "recall"] + assert len(observed) == len(recalled) == 16 + len(NEGATIVE_CONTROLS) + 3 + 3 * len(FRONTS) + 2 + journey = [(x["kind"], x["name"]) for x in observed] + for index, name in enumerate(FRONTS): + assert journey[index * 2:index * 2 + 2] == [("front", name), ("restoration", name)] + assert journey.index(("rollback", "restore-and-recover")) < journey.index(("stability", "0")) + assert [(kind, name) for kind, name in journey if kind == "stability-front"] == [ + ("stability-front", f"{index}:{front}") for index in range(3) for front in FRONTS + ] + + +def test_all_eight_routes_may_share_one_authorized_cloudflare_singleton_run(): + config = policy() + owner = OwnerFixture(config) + owner.shared_run_id = True + result = collect(config, owner) + assert result["status"] == "UNSIGNED_CANDIDATE" + assert result["release_verdict"] == "NOT_READY" + assert len([x for x in owner.calls if x["operation"] == "recall"]) == 16 + len(NEGATIVE_CONTROLS) + 3 + 3 * len(FRONTS) + 2 + + +def test_missing_admission_stops_before_any_observation(): + owner = OwnerFixture(policy()) + owner.admitted = False + with pytest.raises(CollectorError, match="admission_denied"): + collect(fixture=owner) + assert [x["operation"] for x in owner.calls] == ["admission"] + + +@pytest.mark.parametrize("field,value", [("nonce", "forged"), ("request_sha256", "0" * 64), + ("authenticated", False), ("authority_id", "candidate-author")]) +def test_forged_owner_response_rejected(field, value): + owner = OwnerFixture(policy()) + original = owner.call + owner.call = lambda request: {**original(request), field: value} + with pytest.raises(CollectorError, match="owner_response"): + collect(fixture=owner) + + +@pytest.mark.parametrize("mutation", [ + lambda d: d.update(mode="fixture"), + lambda d: d.update(challenge="prior-run"), + lambda d: d["bindings"].update(source_commit="f" * 40), + lambda d: d["bindings"].update(durable_object_id="f" * 64), + lambda d: d["result"]["status_snapshot"]["target"].update(account_id="f" * 32), + lambda d: d["result"]["status_snapshot"].update(online=False), + lambda d: d["result"]["status_snapshot"].update(request_id="not-a-mutation"), + lambda d: d["result"]["controls"]["halt"].update(state="stopped"), + lambda d: d.update(observed_at=(NOW - timedelta(hours=2)).isoformat()), + lambda d: d["result"].update(model="invented"), + lambda d: d["result"].update(account_sha256="f" * 64), + lambda d: d["result"].update(subscription_auth=False), + lambda d: d["result"].update(resumed_session_id="latest-other-session"), + lambda d: d["result"]["controls"]["halt"].update(applied=False), + lambda d: d["result"].update(status_inference_count=1), + lambda d: d["result"].update(capture_readback_sha256="bad"), +]) +def test_forged_or_incomplete_front_proof_is_not_promoted(mutation): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + mutation(data) + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) + + +def test_recall_hash_mismatch_is_rejected(): + owner = OwnerFixture(policy()) + original = owner.call + def wrong(request): + value = original(request) + if request["operation"] == "recall": + value["result"]["content_sha256"] = "f" * 64 + return value + owner.call = wrong + with pytest.raises(CollectorError, match="receipt_hash"): + collect(fixture=owner) + + +@pytest.mark.parametrize("kind,field,value", [ + ("negative", "side_effects", True), ("negative", "after_sha256", "e" * 64), + ("stability", "unexpected_restarts", 1), ("stability", "healthy", False), + ("external", "reviewer_id", "independent-fixture"), + ("external", "access_session_verified", False), + ("rollback", "restored_artifact_sha256", "f" * 64), + ("rollback", "restored_checkpoint_source_commit", "f" * 40), + ("rollback", "recovered_readback_verified", False), + ("restoration", "method", "unadmitted_reset"), + ("restoration", "previous_run_id", "wrong-run"), +]) +def test_negative_stability_external_and_rollback_fail_closed(kind, field, value): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == kind: + data["result"][field] = value + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) + + +@pytest.mark.parametrize("field", [ + "rollback_checkpoint_sha256", "rollback_configuration_sha256", + "recovered_checkpoint_sha256", "recovered_configuration_sha256", +]) +def test_missing_owner_approved_rollback_digest_pin_fails_before_observation(field): + config = policy() + del config[field] + with pytest.raises(CollectorError, match="policy_rollback_identity"): + collect(config) + + +@pytest.mark.parametrize("field", [ + "restored_checkpoint_sha256", "restored_configuration_sha256", + "recovered_checkpoint_sha256", "recovered_configuration_sha256", +]) +def test_well_formed_but_wrong_rollback_digest_fails_closed(field): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "rollback": + data["result"][field] = "f" * 64 + return data + owner.modify = change + with pytest.raises(CollectorError, match="rollback_failed"): + collect(fixture=owner) + + +def test_sensitive_receipt_content_is_rejected_without_echoing_value(): + owner = OwnerFixture(policy()) + owner.modify = lambda data: {**data, "authorization": "test-sensitive-value"} + with pytest.raises(CollectorError) as error: + collect(fixture=owner) + assert "test-sensitive-value" not in str(error.value) + + +def test_pinned_probe_rejects_source_drift_before_execution(tmp_path): + path = tmp_path / "probe.py" + path.write_text("raise SystemExit('must not execute')") + with pytest.raises(CollectorError, match="probe_pin"): + PinnedOwnerProbe([sys.executable, str(path)], ["0" * 64, "0" * 64]) + + +def test_policy_rejects_userinfo_origin_and_boolean_counts(): + config = policy() + config["bindings"]["cloudflare_origin"] = "https://name:password@example.test" + with pytest.raises(CollectorError, match="policy"): + collect(config) + + +def test_duplicate_provider_invocation_receipt_is_rejected(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + data["result"]["provider_invocation_id"] = "reused-invocation" + return data + owner.modify = change + with pytest.raises(CollectorError, match="replay"): + collect(fixture=owner) + + +def make_probe(tmp_path, source, timeout=1): + path = tmp_path / "probe.py" + path.write_text(source) + command = [str(Path(sys.executable).resolve()), str(path)] + pins = [hashlib.sha256(Path(item).read_bytes()).hexdigest() for item in command] + return PinnedOwnerProbe(command, pins, timeout=timeout) + + +@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport") +def test_nonreading_probe_has_whole_request_deadline_with_small_pipe(tmp_path, monkeypatch): + import fcntl + probe = make_probe(tmp_path, "import time; time.sleep(4)") + original = subprocess.Popen + def small_pipe(*args, **kwargs): + process = original(*args, **kwargs) + fcntl.fcntl(process.stdin.fileno(), fcntl.F_SETPIPE_SZ, 4096) + return process + monkeypatch.setattr(subprocess, "Popen", small_pipe) + started = time.monotonic() + with pytest.raises(CollectorError, match="probe_timeout"): + probe.call({"padding": "x" * 6000}) + assert time.monotonic() - started < 2.5 + + +@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport") +def test_normal_broker_state_db_is_preserved_and_route_overrides_removed(tmp_path, monkeypatch): + monkeypatch.setenv("SOL_STATE_DB", "/private/fixture.db") + monkeypatch.setenv("GOOGLE_APPLICATION_CREDENTIALS", "/must/not/inherit") + probe = make_probe(tmp_path, "import json,os,sys; json.load(sys.stdin); print(json.dumps({'state_ok':os.environ.get('SOL_STATE_DB')=='/private/fixture.db','override_present':'GOOGLE_APPLICATION_CREDENTIALS' in os.environ}))") + assert probe.call({"request": "fixture"}) == {"state_ok": True, "override_present": False} + + +@pytest.mark.skipif(os.name != "posix", reason="Linux owner-probe transport") +def test_probe_output_overflow_is_bounded(tmp_path): + probe = make_probe(tmp_path, "import sys; sys.stdin.read(); sys.stdout.write('x'*140000); sys.stdout.flush()") + with pytest.raises(CollectorError, match="probe_output_size"): + probe.call({"request": "fixture"}) + + +def test_healthy_boolean_cannot_replace_per_route_stability_readback(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "stability-front" and data["name"] == "2:CFgrauto": + snapshot = data["result"]["snapshot"] + snapshot["online"] = False + # A correct route digest cannot turn an unhealthy snapshot into health. + data["result"]["snapshot_sha256"] = sha256_json(snapshot) + return data + owner.modify = change + with pytest.raises(CollectorError, match="status_snapshot_unhealthy"): + collect(fixture=owner) + + +def test_cloned_owner_route_result_is_rejected_for_another_front(): + owner = OwnerFixture(policy()) + first = {} + def change(data): + if data["kind"] == "stability-front" and data["name"] == "0:auto": + first["result"] = deepcopy(data["result"]) + if data["kind"] == "stability-front" and data["name"] == "0:cauto": + # The singleton target may match; route provenance still must not. + data["result"] = deepcopy(first["result"]) + return data + owner.modify = change + with pytest.raises(CollectorError, match="stability_route_binding"): + collect(fixture=owner) + + +@pytest.mark.parametrize("field,value", [ + ("front", "auto"), ("provider", "claude"), ("route_readback_verified", False), + ("snapshot_sha256", "f" * 64), +]) +def test_stability_route_provenance_tampering_fails_closed(field, value): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "stability-front" and data["name"] == "0:cauto": + data["result"][field] = value + return data + owner.modify = change + with pytest.raises(CollectorError, match="stability_route_binding"): + collect(fixture=owner) + + +def test_legacy_aggregate_snapshots_cannot_count_as_route_receipts(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "stability": + data["result"]["snapshots"] = {front: owner.snapshot(front) for front in FRONTS} + return data + owner.modify = change + with pytest.raises(CollectorError, match="stability_snapshots_unadmitted"): + collect(fixture=owner) + + +@pytest.mark.parametrize("field", ["account_id", "application_id", "durable_object_id"]) +def test_policy_requires_the_complete_cloudflare_identity_tuple(field): + config = policy() + del config["bindings"][field] + with pytest.raises(CollectorError, match="policy_bindings"): + collect(config) + + +def test_admission_revocation_is_checked_before_exact_recall(): + owner = OwnerFixture(policy()) + original = owner.call + def revoke(request): + value = original(request) + if request["operation"] == "observe": + owner.admitted = False + return value + owner.call = revoke + with pytest.raises(CollectorError, match="admission_denied"): + collect(fixture=owner) + assert not any(request["operation"] == "recall" for request in owner.calls) + + +def test_admission_cannot_authorize_a_different_journey_policy(): + owner = OwnerFixture(policy()) + original = owner.call + def wrong(request): + value = original(request) + if request["operation"] == "admission": + value["result"]["policy_sha256"] = "f" * 64 + return value + owner.call = wrong + with pytest.raises(CollectorError, match="admission_journey"): + collect(fixture=owner) + + +@pytest.mark.skipif(os.name != "posix", reason="Linux collector CLI") +@pytest.mark.parametrize("case,reason", [("missing_probe", "owner_probe_unconfigured"), + ("wrong_pin", "policy_pin_mismatch"), + ("symlink", "policy_path_invalid")]) +def test_cli_refuses_missing_authority_or_bad_policy_without_output(tmp_path, case, reason): + tmp_path.chmod(0o700) + path = tmp_path / "policy.json" + raw = json.dumps(policy()).encode() + path.write_bytes(raw) + path.chmod(0o600) + pin = hashlib.sha256(raw).hexdigest() + if case == "wrong_pin": + pin = "0" * 64 + if case == "symlink": + link = tmp_path / "link.json" + link.symlink_to(path) + path = link + output = tmp_path / "unsigned.json" + process = subprocess.run([sys.executable, str(Path(__file__).resolve().parents[1] / "scripts/collect_finisher_e2e.py"), + "--policy", str(path), "--policy-sha256", pin, "--output", str(output)], + capture_output=True, text=True, timeout=5) + assert process.returncode == 2 + assert json.loads(process.stdout) == {"status": "HELD", "release_verdict": "NOT_READY", "reason": reason} + assert not output.exists() + + +@pytest.mark.parametrize("mutation", [ + lambda c: c["pause"]["target"].update(run_id="another-run"), + lambda c: c["pause"]["readback"]["target"].update(durable_object_id="f" * 64), + lambda c: c["pause"]["readback"].update(source_sha="f" * 40), + lambda c: c["resume"]["readback"]["controller"].update(last_control_id="different-request"), + lambda c: c["resume"]["readback"]["controller"].update(paused=True), + lambda c: c["halt"]["readback"]["controller"].update(halted=False), + lambda c: c["halt"]["readback"].update(terminal_halt_verified=False), + lambda c: c["pause"]["readback"]["checks"].update(heartbeat_fresh=False), + lambda c: c["pause"]["readback"].update(observed_at=(NOW - timedelta(seconds=31)).isoformat()), + lambda c: c["pause"].update(submitted_at=(NOW + timedelta(minutes=5)).isoformat()), + lambda c: c["resume"].update(action="pause"), +]) +def test_action_readback_cannot_cross_target_run_time_or_actual_state(mutation): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + mutation(data["result"]["controls"]) + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) + + +def test_native_control_id_cannot_be_replayed_with_new_front_request_id(): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + control = data["result"]["controls"]["pause"] + control["native_request_id"] = "reused-native-id" + control["readback"]["controller"]["last_control_id"] = "reused-native-id" + return data + owner.modify = change + with pytest.raises(CollectorError, match="control_replay"): + collect(fixture=owner) + + +@pytest.mark.parametrize("mutation", [ + lambda d: d.update(terminal_halt_verified=False, checks={"installation_verified": True, "service_active": True, "heartbeat_fresh": True}), + lambda d: d.update(lifecycle="running"), + lambda d: d.update(online=True), +]) +def test_terminal_halt_requires_actual_stopped_cleanup_proof(mutation): + owner = OwnerFixture(policy()) + def change(data): + if data["kind"] == "front": + mutation(data["result"]["controls"]["halt"]["readback"]) + return data + owner.modify = change + with pytest.raises(CollectorError): + collect(fixture=owner) diff --git a/tests/test_p4_master_report_contract.py b/tests/test_p4_master_report_contract.py new file mode 100644 index 0000000..4026343 --- /dev/null +++ b/tests/test_p4_master_report_contract.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import json +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] + + +def load_script(name: str): + spec = importlib.util.spec_from_file_location(name, ROOT / "scripts" / f"{name}.py") + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def completed_report() -> dict: + # Contract fixture only: no real P4 execution or certification is claimed. + return { + "phase": "P4", + "passed": True, + "run_outcome": "COMPLETE", + "release_verdict": "NOT_READY", + "source_commit": "a" * 40, + "checks": {"sealed_twelve_image_manifest_verified": True, "public_verifier_passed": True}, + "cleanup": { + "unrelated_container_ids_preserved": True, + "ephemeral_private_files_removed": True, + }, + } + + +def test_successful_producer_report_is_consumable_without_changing_master_gate(tmp_path: Path) -> None: + producer = load_script("p4_hostile_acceptance") + consumer = load_script("master_acceptance") + report = completed_report() + output = tmp_path / "p4.json" + + producer.write_acceptance_report(output, report) + + gate = consumer._gate("P4", output, consumer._p4) + assert gate == {"state": "COMPLETE", "evidence_sha256": hashlib.sha256(output.read_bytes()).hexdigest()} + emitted = json.loads(output.read_text()) + assert emitted["source_commit"] == "a" * 40 + assert emitted["release_verdict"] == "NOT_READY" + + +@pytest.mark.parametrize("failure", ["checks", "empty_checks", "outcome", "passed", "containers", "private_files"]) +def test_incomplete_producer_report_cannot_satisfy_master_gate(tmp_path: Path, failure: str) -> None: + producer = load_script("p4_hostile_acceptance") + consumer = load_script("master_acceptance") + report = completed_report() + # A stale marker must be removed if a later verification or cleanup fails. + report["completed_phase_gate"] = "P4" + if failure == "checks": + report["checks"]["public_verifier_passed"] = False + elif failure == "empty_checks": + report["checks"] = {} + elif failure == "outcome": + report["run_outcome"] = "INFRA_FAILED" + elif failure == "passed": + report["passed"] = False + elif failure == "containers": + report["cleanup"]["unrelated_container_ids_preserved"] = False + else: + report["cleanup"]["ephemeral_private_files_removed"] = False + output = tmp_path / "p4.json" + + producer.write_acceptance_report(output, report) + + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + assert "completed_phase_gate" not in json.loads(output.read_text()) + + +def test_wrapper_failure_removes_previously_completed_marker(tmp_path: Path) -> None: + producer = load_script("p4_hostile_acceptance") + wrapper = load_script("p4_hostile_acceptance_exact") + consumer = load_script("master_acceptance") + output = tmp_path / "p4.json" + producer.write_acceptance_report(output, completed_report()) + + result = wrapper.enrich_report(output_path=output, harness=producer, capture={}, return_code=1) + + assert result["passed"] is False + assert "completed_phase_gate" not in result + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + + +@pytest.mark.parametrize("scenario", ["success", "nonzero", "malformed", "interrupted"]) +def test_exact_entrypoint_publishes_only_after_complete_enrichment(tmp_path: Path, monkeypatch, scenario: str) -> None: + producer = load_script("p4_hostile_acceptance") + wrapper = load_script("p4_hostile_acceptance_exact") + consumer = load_script("master_acceptance") + output = tmp_path / "p4.json" + capture = { + "result": { + "exit_code": 0, "timed_out": False, "oom_killed": False, + "containment": {"network_none": True}, + "logs": json.dumps({"passed": True, "private_key_loaded": False, + "sbom_valid": True, "admission": {"allowed": True}}), + }, + "artifacts": {}, "execution_limits": {}, + } + if scenario == "malformed": + capture["result"] = None + + def core_main(): + producer.write_acceptance_report(output, completed_report()) + # Even a successful core result is pending while the wrapper is running. + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) + if scenario == "interrupted": + raise RuntimeError("simulated interruption after core serialization") + return 1 if scenario == "nonzero" else 0 + + def capture_override(original, *, artifacts, capture: dict): + capture.update(captured) + return original + + captured = capture + monkeypatch.setattr(producer, "main", core_main) + monkeypatch.setattr(wrapper, "load_harness", lambda: producer) + monkeypatch.setattr(wrapper, "parse_wrapper_paths", lambda argv: (tmp_path / "manifest.json", output)) + monkeypatch.setattr(wrapper, "resolve_runner_artifacts", lambda *args: {}) + monkeypatch.setattr(wrapper, "make_execute_container_override", capture_override) + if scenario in {"malformed", "interrupted"}: + with pytest.raises((AttributeError, RuntimeError)): + wrapper.main([]) + else: + assert wrapper.main([]) == (0 if scenario == "success" else 1) + if scenario == "success": + assert consumer._gate("P4", output, consumer._p4)["state"] == "COMPLETE" + else: + with pytest.raises(ValueError, match="P4 gate is incomplete"): + consumer._gate("P4", output, consumer._p4) diff --git a/tests/test_p4_worker_service_env.py b/tests/test_p4_worker_service_env.py new file mode 100644 index 0000000..db4d071 --- /dev/null +++ b/tests/test_p4_worker_service_env.py @@ -0,0 +1,51 @@ +"""The P4 worker health probe keeps every state path on the writable workspace. + +The hardened probe runs the worker image with a read-only root. Since P6 the worker app opens its +deployment ledger at import time, and its default path lives on the image root, so the probe died with +"Read-only file system: '/opt/var'" before /healthz answered (P4 INFRA_FAILED, ops-20260924). +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / "scripts" / "p4_hostile_acceptance.py" +STATE_PATHS = ( + "ECHO_CERTFORGE_DB", + "ECHO_CERTFORGE_EVIDENCE_ROOT", + "ECHO_CERTFORGE_TRUSTED_KEYS", + "ECHO_CERTFORGE_DEPLOYMENT_LEDGER", +) + + +class _Captured(Exception): + pass + + +def _load_harness(): + spec = importlib.util.spec_from_file_location("p4_hostile_acceptance_worker_env_test", SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_worker_health_probe_keeps_state_on_the_writable_workspace(monkeypatch, tmp_path: Path) -> None: + harness = _load_harness() + captured: dict = {} + + def fake_create_container(**kwargs): + captured.update(kwargs) + raise _Captured() + + monkeypatch.setattr(harness, "create_container", fake_create_container) + with pytest.raises(_Captured): + harness.service_health_probe("image@sha256:" + "0" * 64, harness.ImageRole.WORKER, "token-12345678", tmp_path) + environment = captured["environment"] + assert environment["ECHO_CERTFORGE_DEPLOYMENT_LEDGER"] == "/workspace/state/deployments.sqlite3" + for name in STATE_PATHS: + assert environment[name].startswith("/workspace/state/"), name diff --git a/tests/test_package_dependency_policy.py b/tests/test_package_dependency_policy.py new file mode 100644 index 0000000..bfe83e1 --- /dev/null +++ b/tests/test_package_dependency_policy.py @@ -0,0 +1,17 @@ +"""Package-install policy, separate from the stdlib-only certification journey.""" +from pathlib import Path +import tomllib + +from packaging.requirements import Requirement + + +def test_package_install_rejects_vulnerable_cryptography_versions() -> None: + # The deployment venv installs pyproject.toml, independently of the image lock. + manifest = Path(__file__).resolve().parents[1] / "pyproject.toml" + dependencies = tomllib.loads(manifest.read_text(encoding="utf-8"))["project"]["dependencies"] + requirement = next( + item for item in map(Requirement, dependencies) if item.name == "cryptography" + ) + for vulnerable in ("44.0.0", "49.0.0", "49.0.1"): + assert not requirement.specifier.contains(vulnerable) + assert requirement.specifier.contains("50.0.0") diff --git a/tests/test_release_pins.py b/tests/test_release_pins.py new file mode 100644 index 0000000..152dc36 --- /dev/null +++ b/tests/test_release_pins.py @@ -0,0 +1,69 @@ +"""ops-20260924: the supply-chain pins stay consistent and patched (issue #22 security release). + +P4 image sealing fails closed on HIGH vulnerabilities with an available fix. This guards the +fix: one patched base image digest everywhere, the patched cryptography release hash-locked, +and the dispatcher unit decoupled from API restarts. +""" +from __future__ import annotations + +import re +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +PATCHED_BASE = "sha256:4c47124a8391cb7a9f571164147d154777cf012a4ece5f86097130d7a4478111" +VULNERABLE_BASE = "sha256:6d43704baacd1bfbe7c295d7f13079d5d8104ed33568873133f8fc69980419df" +ROLES = ("runner", "custody", "anchor", "signer", "worker", "verifier") + + +def _text(rel: str) -> str: + return (ROOT / rel).read_text(encoding="utf-8") + + +def test_every_role_image_uses_the_patched_base() -> None: + for role in ROLES: + dockerfile = _text(f"images/{role}/Dockerfile") + assert dockerfile.startswith(f"FROM python:3.12-alpine@{PATCHED_BASE}\n"), role + assert f'org.opencontainers.image.base.digest="{PATCHED_BASE}"' in dockerfile, role + assert VULNERABLE_BASE not in dockerfile, role + + +def test_sandbox_and_p4_gate_share_the_patched_base() -> None: + assert f'DEFAULT_IMAGE = "python:3.12-alpine@{PATCHED_BASE}"' in _text( + "src/echo_certification_forge/sandbox.py") + assert f'BASE_DIGEST="${{BASE_DIGEST:-{PATCHED_BASE}}}"' in _text("scripts/run_p4_gate.sh") + + +def test_cryptography_is_the_patched_hash_locked_release() -> None: + assert "cryptography==50.0.0" in _text("images/requirements.in") + lock = _text("images/requirements.lock").replace("\r\n", "\n") + block = re.search(r"^cryptography==(\S+) \\\n((?: --hash=sha256:[0-9a-f]{64}(?: \\)?\n)+)", lock, re.M) + assert block is not None + major = int(block.group(1).split(".")[0]) + assert major >= 50 + assert len(re.findall(r"sha256:[0-9a-f]{64}", block.group(2))) >= 20 + assert "cryptography==49" not in lock + + +def test_dispatcher_wants_but_does_not_require_the_api_service() -> None: + script = _text("deploy/deploy_forge.sh") + dispatcher = script[script.index('sudo tee "$DISPATCH_UNIT_PATH"'):] + unit = dispatcher[: dispatcher.index("[Service]")] + assert "Wants=$SERVICE.service" in unit + assert "Requires=$SERVICE.service" not in unit + assert "After=network.target $SERVICE.service" in unit + + +def test_source_fetch_credential_helper_is_configurable_with_safe_default() -> None: + script = _text("deploy/deploy_forge.sh") + assert ('GIT_CREDENTIAL_HELPER="${CERTFORGE_GIT_CREDENTIAL_HELPER:-store --file=' + '/home/forge/.config/echo/omega_git_creds}"') in script + assert 'GITC=(-c credential.helper= -c credential.helper="$GIT_CREDENTIAL_HELPER")' in script + + +def test_deploy_default_manifest_digest_matches_the_enforced_pin() -> None: + script = _text("deploy/deploy_forge.sh") + run_worker = _text("src/echo_certification_forge/run_worker.py") + pinned = re.search(r'_PRODUCTION_MANIFEST_SHA256 = \(\s*"([0-9a-f]{64})"', run_worker) + default = re.search(r'TRUSTED_MANIFEST_SHA256="\$\{ECHO_CERTFORGE_TRUSTED_MANIFEST_SHA256:-([0-9a-f]{64})\}"', + script) + assert pinned and default and default.group(1) == pinned.group(1)