coulomb/the-custodian
5
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

Record reviewed case source evidence (12/13)
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 5s
Details
Reject generated-only commits / generated-commit-guard (push) Successful in 14s
Details

Browse source
This commit is contained in:
codex 2026-10-17 22:36:09 +02:00
parent df5765e0e1
commit e9dc1a0f90
20 changed files with 2477 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

287
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-integration-r1/projection.json Normal file
View file

@ -0,0 +1,287 @@
{
"ok": true,
"status": "applied",
"repo_slug": "ops-warden",
"commit": "b593c879831378db7dc62adfabdfedf2d44ddd38",
"requests": 2,
"elapsed_ms": 4502.2,
"git": {
"upstream": "origin/main",
"behind": 0,
"ahead": 0
},
"identifiers": {
"schema": "repo-manager.identifier-assignment.v1",
"ok": true,
"executed": true,
"repo": "ops-warden",
"namespace": "helixforge",
"assignments": [],
"historical_invalid_identifiers": [],
"historical_identity_collisions": [],
"files_changed": []
},
"receipt": {
"schema": "state-hub.repository-projection-reconcile.v1",
"instance_role": "primary",
"instance_label": "railiance01",
"expected_commit": "b593c879831378db7dc62adfabdfedf2d44ddd38",
"derived_commit": "b593c879831378db7dc62adfabdfedf2d44ddd38",
"outcome": {
"schema": "state-hub.projection-reset.v1",
"repo_slug": "ops-warden",
"commit": "b593c879831378db7dc62adfabdfedf2d44ddd38",
"status": "applied",
"counts": {
"created": 0,
"updated": 51,
"retired": 0,
"refused": 0,
"released": 0,
"created_tasks": 0,
"updated_tasks": 0,
"cancelled_tasks": 0,
"created_edges": 0,
"removed_edges": 0,
"unresolved_edges": 0,
"external_edges": 17
},
"created": [],
"updated": [
"WARDEN-WP-0001",
"WARDEN-WP-0002",
"WARDEN-WP-0003",
"WARDEN-WP-0004",
"WARDEN-WP-0005",
"WARDEN-WP-0006",
"WARDEN-WP-0007",
"WARDEN-WP-0008",
"WARDEN-WP-0009",
"WARDEN-WP-0010",
"WARDEN-WP-0011",
"WARDEN-WP-0012",
"WARDEN-WP-0013",
"WARDEN-WP-0014",
"WARDEN-WP-0015",
"WARDEN-WP-0016",
"WARDEN-WP-0017",
"WARDEN-WP-0018",
"WARDEN-WP-0019",
"WARDEN-WP-0020",
"WARDEN-WP-0021",
"WARDEN-WP-0022",
"WARDEN-WP-0023",
"WARDEN-WP-0024",
"WARDEN-WP-0025",
"WARDEN-WP-0026",
"WARDEN-WP-0027",
"WARDEN-WP-0028",
"WARDEN-WP-0029",
"WARDEN-WP-0030",
"WARDEN-WP-0031",
"WARDEN-WP-0032",
"WARDEN-WP-0033",
"WARDEN-WP-0034",
"WARDEN-WP-0035",
"WARDEN-WP-0036",
"WARDEN-WP-0037",
"WARDEN-WP-0038",
"WARDEN-WP-0039",
"WARDEN-WP-0040",
"WARDEN-WP-0041",
"WARDEN-WP-0042",
"WARDEN-WP-0043",
"WARDEN-WP-0044",
"WARDEN-WP-ADHOC-2026-06-27",
"WARDEN-WP-ADHOC-2026-06-29",
"WARDEN-WP-ADHOC-2026-07-07",
"WARDEN-WP-ADHOC-2026-08-11",
"WARDEN-WP-ADHOC-2026-08-17",
"WARDEN-WP-ADHOC-2026-09-08",
"WARDEN-WP-ADHOC-2026-10-03"
],
"retired": [],
"released": [],
"created_tasks": [],
"updated_tasks": [],
"cancelled_tasks": [],
"created_edges": [],
"removed_edges": [],
"unresolved_edges": [],
"external_edges": [
{
"workplan": "WARDEN-WP-0032",
"from": "WARDEN-WP-0032",
"target": "WARDEN-WP-0031",
"relationship_type": "blocks",
"from_workplan_id": "38c6a5f3-fb0d-5230-be85-f9e3ffc850f6",
"from_task_id": null,
"to_workplan_id": "739bad25-2345-5f4f-aaa3-cc4cd8c71f6c",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0037",
"from": "WARDEN-WP-0037-T03",
"target": "SECRETS-WP-0006-T06",
"relationship_type": "starts_after",
"from_workplan_id": "42a097db-1c24-558e-a724-030bb2b4443e",
"from_task_id": "a8b1b855-ab34-5835-b9fd-5f48bc0b6817",
"to_workplan_id": null,
"to_task_id": "1431edae-5791-5892-8c4b-829419b537d2"
},
{
"workplan": "WARDEN-WP-0033",
"from": "WARDEN-WP-0033",
"target": "WARDEN-WP-0030",
"relationship_type": "blocks",
"from_workplan_id": "4627d89b-4b00-562a-81e9-76e96f90fa7e",
"from_task_id": null,
"to_workplan_id": "da3367d5-890c-52c6-aa54-1bdd0f277342",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0013",
"from": "WARDEN-WP-0013",
"target": "WARDEN-WP-0011",
"relationship_type": "blocks",
"from_workplan_id": "4678c41a-c1d0-48cd-9988-4ea0380e8258",
"from_task_id": null,
"to_workplan_id": "0a520f8e-01b4-48f1-9af3-2f3f69fd0672",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0013",
"from": "WARDEN-WP-0013",
"target": "WARDEN-WP-0009",
"relationship_type": "blocks",
"from_workplan_id": "4678c41a-c1d0-48cd-9988-4ea0380e8258",
"from_task_id": null,
"to_workplan_id": "9213b262-e2f5-480e-a5bc-56635d5eb4c9",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0013",
"from": "WARDEN-WP-0013",
"target": "WARDEN-WP-0008",
"relationship_type": "blocks",
"from_workplan_id": "4678c41a-c1d0-48cd-9988-4ea0380e8258",
"from_task_id": null,
"to_workplan_id": "a174963a-4ff1-4565-b19f-896cd4ff14a0",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0013",
"from": "WARDEN-WP-0013",
"target": "WARDEN-WP-0010",
"relationship_type": "blocks",
"from_workplan_id": "4678c41a-c1d0-48cd-9988-4ea0380e8258",
"from_task_id": null,
"to_workplan_id": "e93de9fd-0192-4d02-bb7c-5e859fb76b9b",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0031",
"from": "WARDEN-WP-0031",
"target": "WARDEN-WP-0007",
"relationship_type": "blocks",
"from_workplan_id": "739bad25-2345-5f4f-aaa3-cc4cd8c71f6c",
"from_task_id": null,
"to_workplan_id": "3718ac07-2fa2-47d0-a02a-c9a7b83a5ba9",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0038",
"from": "WARDEN-WP-0038",
"target": "WARDEN-WP-0029",
"relationship_type": "blocks",
"from_workplan_id": "751ad530-e44b-52c4-b70e-cb47568a0179",
"from_task_id": null,
"to_workplan_id": "bbb3d9ec-d88d-5088-b4c0-55bfba0a10cf",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0023",
"from": "WARDEN-WP-0023",
"target": "WARDEN-WP-0022",
"relationship_type": "blocks",
"from_workplan_id": "7bad1ec4-a7c2-4980-b8f9-49a7f5408574",
"from_task_id": null,
"to_workplan_id": "fc8afa28-68a7-4250-a19e-9754829f0cd5",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0034",
"from": "WARDEN-WP-0034",
"target": "WARDEN-WP-0030",
"relationship_type": "blocks",
"from_workplan_id": "ae3ff76f-883d-5e2f-b6aa-144d61e8fdef",
"from_task_id": null,
"to_workplan_id": "da3367d5-890c-52c6-aa54-1bdd0f277342",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0040",
"from": "WARDEN-WP-0040",
"target": "WARDEN-WP-0032",
"relationship_type": "blocks",
"from_workplan_id": "c8ee441e-1be1-5219-910c-e79ff23cc9ec",
"from_task_id": null,
"to_workplan_id": "38c6a5f3-fb0d-5230-be85-f9e3ffc850f6",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0040",
"from": "WARDEN-WP-0040",
"target": "WARDEN-WP-0034",
"relationship_type": "blocks",
"from_workplan_id": "c8ee441e-1be1-5219-910c-e79ff23cc9ec",
"from_task_id": null,
"to_workplan_id": "ae3ff76f-883d-5e2f-b6aa-144d61e8fdef",
"to_task_id": null
},
{
"workplan": "WARDEN-WP-0040",
"from": "WARDEN-WP-0040-T03",
"target": "WARDEN-WP-0040-T01",
"relationship_type": "starts_after",
"from_workplan_id": "c8ee441e-1be1-5219-910c-e79ff23cc9ec",
"from_task_id": "93eaf1f2-daaa-5e22-8804-85ca9433571c",
"to_workplan_id": null,
"to_task_id": "611f0901-9fb5-5954-8dd0-a860c5f0cbec"
},
{
"workplan": "WARDEN-WP-0044",
"from": "WARDEN-WP-0044-T04",
"target": "WARDEN-WP-0044-T03",
"relationship_type": "starts_after",
"from_workplan_id": "f7e81013-6868-5d99-8528-ba75a442eca9",
"from_task_id": "4a0fa4a7-5ca9-518a-ab92-a79a5e2cf93e",
"to_workplan_id": null,
"to_task_id": "bcbf6c91-2c04-5c16-ae8c-5d6c06fd6b89"
},
{
"workplan": "WARDEN-WP-0044",
"from": "WARDEN-WP-0044-T03",
"target": "MASON-WP-0010-T03",
"relationship_type": "starts_after",
"from_workplan_id": "f7e81013-6868-5d99-8528-ba75a442eca9",
"from_task_id": "bcbf6c91-2c04-5c16-ae8c-5d6c06fd6b89",
"to_workplan_id": null,
"to_task_id": "56cea6be-c613-586c-80cd-125340b374da"
},
{
"workplan": "WARDEN-WP-0044",
"from": "WARDEN-WP-0044-T03",
"target": "WARDEN-WP-0044-T02",
"relationship_type": "starts_after",
"from_workplan_id": "f7e81013-6868-5d99-8528-ba75a442eca9",
"from_task_id": "bcbf6c91-2c04-5c16-ae8c-5d6c06fd6b89",
"to_workplan_id": null,
"to_task_id": "5f68c991-f490-58f8-a121-92f74073d272"
}
],
"refused": [],
"notes": []
}
}
}

0
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-integration-r1/projection.stderr.txt Normal file
View file

65
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-integration-r1/publication-commit.json Normal file
View file

@ -0,0 +1,65 @@
{
"exit_code": 0,
"stdout": "[detached HEAD b593c87] test: isolate ordinary unit runs from ambient credentials and provider I/O\n 14 files changed, 592 insertions(+), 6 deletions(-)\n create mode 100644 docs/development/safe-unit-tests.md\n create mode 100644 scripts/run_unit_tests.py\n create mode 100644 tests/fixtures/workload-coverage/provenance.json\n create mode 100644 tests/fixtures/workload-coverage/rapp-core-hub/declarations/rapp.yaml\n create mode 100644 tests/fixtures/workload-coverage/rapp-issue-core/declarations/rapp.yaml\n create mode 100644 tests/fixtures/workload-coverage/rapp-qonto/declarations/rapp.yaml\n create mode 100644 tests/test_safe_unit_runner.py\n create mode 100644 tests/unit_isolation_probe.py\n",
"stderr": "",
"commit": "b593c879831378db7dc62adfabdfedf2d44ddd38",
"base": "922c26da0f3fcb0182352ea83e6ae06f29878580",
"files": [
{
"path": "Makefile",
"sha256": "8cf08ec680ed6bd323a1ce8475ade731dbeebcd64ffd43ecd169308f8a727940"
},
{
"path": "scripts/run_unit_tests.py",
"sha256": "86880094b33114fdefb26691140b5c9a516cdba5fe6ddabd70fd5b7f18e9d2aa"
},
{
"path": "tests/test_safe_unit_runner.py",
"sha256": "5408e7face98b01b92da077b0b2d115fcb370ec46e572ad5a69e3d0c10df8433"
},
{
"path": "tests/unit_isolation_probe.py",
"sha256": "726d3e01ae21c1379588e1f08459449908cc8a75ffea1b337c33a1a888c66f59"
},
{
"path": "tests/test_layer_conformance.py",
"sha256": "3edf0a0d49c3560278a564c477834335cfaf5243341cb9345f0d4bf8994db56c"
},
{
"path": "tests/test_workload_join.py",
"sha256": "8ef0891daf4b7ac520430a514690364917110628edc760911a850195cffbdd4a"
},
{
"path": "tests/test_case_plan.py",
"sha256": "944bc2a725ca1b7a10f4516c3ceed65f99627057dd15a859da4f8ffb3d97e71f"
},
{
"path": "tests/test_case_type.py",
"sha256": "5a217a172e7e2ffbea66f1beee9a1c18bf5e97f9fb25b83cb96b6ba976cbb2a5"
},
{
"path": "docs/development/safe-unit-tests.md",
"sha256": "b157c3e6956a41dcd9fc3912d0c7d09a664f11f08f0f8a8e4cae57fac9747cf7"
},
{
"path": "workplans/WARDEN-WP-0044-security-case-follow-through.md",
"sha256": "44f88a5a84894d951add888206fe823d3d13019aca39a572ad8dfa15d510e05d"
},
{
"path": "tests/fixtures/workload-coverage/provenance.json",
"sha256": "a4a203c1bd7bc5b8039d1b2525bea65dbae288a3f552969e4218cd5abda96990"
},
{
"path": "tests/fixtures/workload-coverage/rapp-core-hub/declarations/rapp.yaml",
"sha256": "9ce902da0c77809f1094a634244652835b162f59682e9fa6923d8c8aa4ed4a3c"
},
{
"path": "tests/fixtures/workload-coverage/rapp-issue-core/declarations/rapp.yaml",
"sha256": "a1956deecf2f0fbca44549e5ea5e8f1c4b74ebf91a9482c64c982491af0d923b"
},
{
"path": "tests/fixtures/workload-coverage/rapp-qonto/declarations/rapp.yaml",
"sha256": "15aa9809e8d6a120c516be5ce9adfb72c00382140c171e791083ad117f503b3f"
}
]
}

63
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-integration-r1/publication-inputs.json Normal file
View file

@ -0,0 +1,63 @@
{
"owner": "ops-warden",
"base_revision": "922c26da0f3fcb0182352ea83e6ae06f29878580",
"message": "test: isolate ordinary unit runs from ambient credentials and provider I/O",
"files": [
{
"path": "Makefile",
"sha256": "8cf08ec680ed6bd323a1ce8475ade731dbeebcd64ffd43ecd169308f8a727940"
},
{
"path": "scripts/run_unit_tests.py",
"sha256": "86880094b33114fdefb26691140b5c9a516cdba5fe6ddabd70fd5b7f18e9d2aa"
},
{
"path": "tests/test_safe_unit_runner.py",
"sha256": "5408e7face98b01b92da077b0b2d115fcb370ec46e572ad5a69e3d0c10df8433"
},
{
"path": "tests/unit_isolation_probe.py",
"sha256": "726d3e01ae21c1379588e1f08459449908cc8a75ffea1b337c33a1a888c66f59"
},
{
"path": "tests/test_layer_conformance.py",
"sha256": "3edf0a0d49c3560278a564c477834335cfaf5243341cb9345f0d4bf8994db56c"
},
{
"path": "tests/test_workload_join.py",
"sha256": "8ef0891daf4b7ac520430a514690364917110628edc760911a850195cffbdd4a"
},
{
"path": "tests/test_case_plan.py",
"sha256": "944bc2a725ca1b7a10f4516c3ceed65f99627057dd15a859da4f8ffb3d97e71f"
},
{
"path": "tests/test_case_type.py",
"sha256": "5a217a172e7e2ffbea66f1beee9a1c18bf5e97f9fb25b83cb96b6ba976cbb2a5"
},
{
"path": "docs/development/safe-unit-tests.md",
"sha256": "b157c3e6956a41dcd9fc3912d0c7d09a664f11f08f0f8a8e4cae57fac9747cf7"
},
{
"path": "workplans/WARDEN-WP-0044-security-case-follow-through.md",
"sha256": "44f88a5a84894d951add888206fe823d3d13019aca39a572ad8dfa15d510e05d"
},
{
"path": "tests/fixtures/workload-coverage/provenance.json",
"sha256": "a4a203c1bd7bc5b8039d1b2525bea65dbae288a3f552969e4218cd5abda96990"
},
{
"path": "tests/fixtures/workload-coverage/rapp-core-hub/declarations/rapp.yaml",
"sha256": "9ce902da0c77809f1094a634244652835b162f59682e9fa6923d8c8aa4ed4a3c"
},
{
"path": "tests/fixtures/workload-coverage/rapp-issue-core/declarations/rapp.yaml",
"sha256": "a1956deecf2f0fbca44549e5ea5e8f1c4b74ebf91a9482c64c982491af0d923b"
},
{
"path": "tests/fixtures/workload-coverage/rapp-qonto/declarations/rapp.yaml",
"sha256": "15aa9809e8d6a120c516be5ce9adfb72c00382140c171e791083ad117f503b3f"
}
]
}

18
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-integration-r1/root-disposition.json Normal file
View file

@ -0,0 +1,18 @@
{
"original_candidate": "rejected as reviewed",
"corrected_candidate": "accepted for source revalidation",
"fixes": [
"Classify O_TRUNC and O_APPEND opens as writes before the OS open.",
"Check closed fixture projection set and every recorded byte digest in default unit selection."
],
"candidate_sha256": "1fe65076ce7e633be304fc779bf2c87b7ee1e4e460de55ca74355b5842d036f5",
"correction_sha256": "9407289f8da41ed6fff367baf566ad30f5510a2627c14e0ac0cf971bf9b465c9",
"remaining_limits": [
"No full-suite recertification; author783-pass final run is attributed to author, not adopted as independent full-suite evidence.",
"Source roots/interpreters/libraries remain trusted and directory-readable by design. This is no OS isolation or grant.",
"Projection digest check proves shipped fixture consistency, not live estate freshness or source-owner authenticity.",
"No native acceptance,1.3 pin,draft promotion,credential investigation or external owner assent.",
"Original author initial failure logs remain intact."
],
"native_acceptance": false
}

60
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/README.md Normal file
View file

@ -0,0 +1,60 @@
# Warden safe unit runner — implementation candidate
Prepared against actual owner source `922c26da0f3fcb0182352ea83e6ae06f29878580`,
preserving the normal allocator commit after `ae7bc73`. This is an implementation
packet, not an independent review or canonical integration. `candidate.patch`
applies to that exact archive and all 14 resulting file hashes match
`implementation.json`. No canonical checkout, credential, catalog or pin changed.
The substantive change is `scripts/run_unit_tests.py` plus the ordinary
`make test` entry. It selects an already installed interpreter directly, constructs
the environment/temporary HOME and installs the guard before pytest collection.
No uv/dependency resolution occurs. The supplied documentation lists the closed
Python, Bash/Git fixture and one synthetic-loopback exceptions and their limits.
Unexpected processes, provider commands, sockets and ambient file opens refuse.
Allowed Python children install guards, including the exact reverified BrowserInputs
validator; nested execution refuses. Child HOME/environment are reconstructed
even when a test monkeypatches synthetic token and HOME variables.
Existing tests depended on workstation state. The candidate makes their public
workload declaration snapshots explicit with provenance; default-type/case tests
use the already established explicit NetKingdom source option. It preserves every
test assertion about counts, masking, file modes, original intent, validator
identity and parent/child containment. Existing fixed `printf` transport fixtures
and exact-byte contract doubles remain real local fixture processes rather than
mocked success. Native SSH-keygen integration tests remain deselected and denied
by this unit entry if selected without separate support.
Validation:
- Final ordinary `make test`: **783 passed, four deselected**, 62.70 seconds,
with synthetic input BAO_TOKEN/VAULT_TOKEN/HOME and hostile synthetic pytest/Python
environment options. No real token or home file was read or copied.
- Corrected affected focus: **117 passed**. New runner and regression code pass Ruff.
- Initial focused **136 passed/two failed** and initial full **13 failed/768
passed/two skipped** are retained. They exposed explicit fixture dependencies;
failures were corrected rather than excluded. The first full was an intermediate
diagnostic state, with one configuration edit made while it ran; it is not
evidence for the stable final candidate. See `implementation.json`.
- Exact patch application/hashes verified in a second disposable archive.
One local replay, using existing installed dependencies and explicit source roots:
```bash
git clone --shared --no-hardlinks /home/worsch/ops-warden /tmp/warden-unit-review
git -C /tmp/warden-unit-review checkout --detach 922c26da0f3fcb0182352ea83e6ae06f29878580
git -C /tmp/warden-unit-review apply /home/worsch/the-custodian/docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate.patch
make -C /tmp/warden-unit-review test UNIT_PYTHON=/home/worsch/ops-warden/.venv/bin/python PHASE_MEMORY_REPO=/home/worsch/phase-memory TEST_ARGS='--netkingdom-root /home/worsch/net-kingdom -q'
```
Use a fresh scratch path; existing paths should be inspected, not reset. Tests need
local Git history for the existing generated-catalog provenance assertion.
The existing `WARDEN-WP-0044-T03` paragraph is appended without altering its IDs,
status or dependencies. T03 remains progress and native acceptance is incomplete.
This is an accident guard for trusted unit tests. It is not a hostile-code OS
sandbox, a production isolation grant or proof of native owner provenance.
Installed libraries/source roots remain trusted. Native Bash/Git fixture binaries
do not carry Python audit hooks. Already-open descriptors, native extensions and
all non-open filesystem syscalls are not fully contained. Direct pytest bypasses
the launcher. These limits are also recorded in the owner documentation.

660
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate.patch Normal file
View file

@ -0,0 +1,660 @@
--- a/Makefile
+++ b/Makefile
@@ -1,5 +1,6 @@
# ops-warden development and install targets
ROOT := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
+UNIT_PYTHON ?= $(ROOT)/.venv/bin/python
PHASE_MEMORY_REPO ?= $(abspath $(ROOT)/../phase-memory)
.DEFAULT_GOAL := help
@@ -43,7 +44,7 @@
@warden memory status
test:
- PYTHONPATH=src:$(PHASE_MEMORY_REPO)/src uv run pytest
+ "$(UNIT_PYTHON)" -I scripts/run_unit_tests.py --phase-memory-root "$(PHASE_MEMORY_REPO)" $(TEST_ARGS)
lint:
uv run ruff check .
--- /dev/null
+++ b/scripts/run_unit_tests.py
@@ -0,0 +1,283 @@
+"""Run trusted unit tests with no inherited credentials or external I/O.
+
+This is an accident guard for this suite, not an OS sandbox for hostile Python.
+Direct pytest bypasses it. Use make test (already installed dependencies only).
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import runpy
+import socket
+import subprocess
+import sys
+import tempfile
+
+ROOT = Path(__file__).resolve().parents[1]
+SELF = Path(__file__).resolve()
+SHELL_FIXTURES = {
+ "printf 'the-secret-value'", "printf 'sekret'", "printf 'new'",
+ "printf '%s' '{\"wrap_info\":{\"token\":\"hvs.WRAP\"}}'",
+ "printf 'not-json'",
+}
+JOURNAL_CODE_SHA256 = '69aa18f302fd698104cb2253b4af0c3a077e72a0a427183e3dce1444c4926df9'
+
+
+class IsolationDenied(PermissionError, subprocess.SubprocessError):
+ """A unit test attempted an unreviewed external operation."""
+
+
+def clean_environment(home: Path, temporary: Path) -> dict[str, str]:
+ # Construct, never copy/inventory the caller's environment.
+ return {'HOME': str(home), 'TMPDIR': str(temporary), 'PATH': '/usr/bin:/bin',
+ 'LANG': 'C.UTF-8', 'PYTHONDONTWRITEBYTECODE': '1',
+ 'PYTEST_DISABLE_PLUGIN_AUTOLOAD': '1'}
+
+
+def inside(path: Path, roots: tuple[Path, ...]) -> bool:
+ return any(path == root or root in path.parents for root in roots)
+
+
+class Guard:
+ def __init__(self, temporary: Path, read_roots: list[Path], child: bool = False, git_metadata: bool = False):
+ self.temporary = temporary.resolve()
+ self.read_roots = tuple(p.resolve() for p in read_roots)
+ self.child = child
+ self.git_metadata = git_metadata
+ self.loopback_test = False
+ self.endpoints: set[tuple[str, int]] = set()
+ self.launch = None
+ self.original_popen = subprocess.Popen
+
+ def fixture_path(self, value: str) -> Path:
+ path = Path(value).resolve()
+ if not inside(path, (self.temporary,)):
+ raise IsolationDenied('unit isolation: non-fixture path')
+ return path
+
+ def audit(self, event, args):
+ if event == 'open' and not isinstance(args[0], int):
+ path = Path(os.fsdecode(args[0])).resolve()
+ mode, flags = args[1:3]
+ write = ((isinstance(mode, str) and any(c in mode for c in 'wax+'))
+ or (isinstance(flags, int) and flags & (os.O_WRONLY | os.O_RDWR | os.O_CREAT)))
+ roots = (self.temporary,) if write else self.read_roots
+ if path != Path('/dev/null') and not inside(path, roots):
+ raise IsolationDenied('unit isolation: file outside reviewed roots')
+ if event == 'subprocess.Popen':
+ if self.launch is None or tuple(args[1]) != self.launch:
+ raise IsolationDenied('unit isolation: unexpected process')
+ if event in {'os.system', 'os.exec', 'os.posix_spawn', 'os.fork', 'os.forkpty'}:
+ raise IsolationDenied('unit isolation: native execution disabled')
+ if event.startswith('socket.'):
+ if not self.loopback_test or self.child:
+ raise IsolationDenied('unit isolation: sockets disabled')
+ if event == 'socket.__new__':
+ if args[1] != socket.AF_INET or args[2] & 15 != socket.SOCK_STREAM:
+ raise IsolationDenied('unit isolation: socket family/type denied')
+ elif event == 'socket.bind':
+ if args[1] != ('127.0.0.1', 0):
+ raise IsolationDenied('unit isolation: only ephemeral fixture bind')
+ elif event == 'socket.connect':
+ if args[1] not in self.endpoints:
+ raise IsolationDenied('unit isolation: only this fixture endpoint')
+ elif event not in {'socket.gethostname'}:
+ raise IsolationDenied('unit isolation: DNS/socket operation denied')
+
+ def python_payload(self, argv):
+ """Closed subprocess inventory; every allowed Python child installs guard."""
+ from warden import case_inputs
+ if (len(argv) == 5 and argv[:3] == [sys.executable, '-I', '-c']
+ and argv[3] == case_inputs._VALIDATE):
+ directory = self.fixture_path(argv[4])
+ if not directory.name.startswith('warden-case-validator-'):
+ raise IsolationDenied('unit isolation: validator fixture directory differs')
+ # Reverify copied modules immediately before launch, independently of caller.
+ for relative, digest in case_inputs.SOURCE_PINS.items():
+ if hashlib.sha256((directory / relative).read_bytes()).hexdigest() != digest:
+ raise IsolationDenied('unit isolation: validator source pin differs')
+ return {'code': argv[3], 'args': [str(directory)]}
+ if len(argv) == 3 and argv[:2] == [sys.executable, '-c'] and argv[2] == 'print("SYNTHETIC-SECRET")':
+ return {'code': argv[2], 'args': []}
+ if (len(argv) == 6 and argv[:3] == [sys.executable, '-B', '-c']
+ and hashlib.sha256(argv[3].encode()).hexdigest() == JOURNAL_CODE_SHA256):
+ self.fixture_path(argv[4])
+ if len(argv[5]) > 192:
+ raise IsolationDenied('unit isolation: journal source bound')
+ return {'code': argv[3], 'args': argv[4:]}
+ if len(argv) >= 2 and argv[0] in (sys.executable, 'python3'):
+ script = Path(argv[1]).resolve()
+ args = argv[2:]
+ if script == ROOT / 'tests/unit_isolation_probe.py' and len(args) == 1 and args[0] in {'state', 'socket', 'provider', 'home', 'nested'}:
+ return {'script': str(script), 'args': args}
+ if script == ROOT / 'scripts/check_layer_conformance.py' and not args:
+ return {'script': str(script), 'args': args}
+ if script == ROOT / 'scripts/emit_high_risk_paths.py' and args == ['--check']:
+ return {'script': str(script), 'args': args, 'git_metadata': True}
+ if script == ROOT / 'scripts/check_principals_drift.py' and len(args) == 4 and args[::2] == ['--inventory', '--infra']:
+ for value in args[1::2]:
+ self.fixture_path(value)
+ return {'script': str(script), 'args': args}
+ if script == ROOT / 'scripts/build_flex_auth_registry.py':
+ if (len(args) in (3, 5) and Path(args[0]).resolve() == ROOT / 'examples/inventory.seed.yaml'
+ and args[-2] == '-o' and (len(args) == 3 or args[1] == '--zone-resolutions')):
+ self.fixture_path(args[-1])
+ if len(args) == 5:
+ self.fixture_path(args[2])
+ return {'script': str(script), 'args': args}
+ raise IsolationDenied('unit isolation: command not in reviewed subprocess inventory')
+
+ def popen(self, argv, *pos, **kw):
+ if pos or kw.get('executable') or kw.get('preexec_fn') or kw.get('pass_fds') or kw.get('close_fds') is False:
+ raise IsolationDenied('unit isolation: unexpected process options')
+ if kw.pop('shell', False):
+ if self.child or not isinstance(argv, str) or argv not in SHELL_FIXTURES:
+ raise IsolationDenied('unit isolation: unreviewed shell command')
+ return self.spawn(['/bin/bash', '--noprofile', '--norc', '-c', argv], kw)
+ if not isinstance(argv, (list, tuple)):
+ raise IsolationDenied('unit isolation: argv list required')
+ argv = [os.fspath(x) for x in argv]
+ # Never run a real provider: the only shell fixture is an exact-byte double.
+ from warden.doubles import _DOUBLES
+ metadata_commands = (['git', 'log', '-1', '--format=%H %cI', '--', str(ROOT / 'registry/routing/catalog.yaml')],
+ ['git', 'status', '--porcelain', '--', str(ROOT / 'registry/routing/catalog.yaml')])
+ if self.child and not (self.git_metadata and argv in metadata_commands):
+ raise IsolationDenied('unit isolation: nested process disabled')
+ candidate = Path(argv[0])
+ if not candidate.is_absolute():
+ import shutil
+ resolved = shutil.which(argv[0], path=kw.get('env', os.environ).get('PATH', ''))
+ candidate = Path(resolved) if resolved else candidate
+ if self.child and self.git_metadata:
+ if Path(kw.get('cwd', '.')).resolve() != ROOT:
+ raise IsolationDenied('unit isolation: metadata cwd differs')
+ command = ['/usr/bin/git', '-c', 'core.fsmonitor=false', '-c', 'core.hooksPath=/dev/null', '--no-pager', *argv[1:]]
+ elif candidate.is_absolute() and inside(candidate.resolve(), (self.temporary,)) and candidate.name in _DOUBLES:
+ if candidate.read_text() != _DOUBLES[candidate.name].script:
+ raise IsolationDenied('unit isolation: double bytes differ')
+ tail = argv[1:]
+ if (candidate.name == 'bao' and tail not in (
+ ['kv', 'get', '-field=NPM_AUTH_TOKEN', 'platform/x/y'],
+ ['kv', 'get', '-field=API_KEY', 'platform/x/y/z'],
+ ['login', '-method=oidc'], ['write', 'secret/x'])) or (candidate.name == 'key-cape' and tail != ['login']):
+ raise IsolationDenied('unit isolation: double arguments differ')
+ command = ['/bin/bash', '--noprofile', '--norc', str(candidate), *tail]
+ else:
+ payload = self.python_payload(argv)
+ payload.update(temporary=str(self.temporary), read_roots=[str(x) for x in self.read_roots])
+ command = [sys.executable, '-I', str(SELF), '--child', json.dumps(payload)]
+ return self.spawn(command, kw)
+
+ def spawn(self, command, kw):
+ cwd = Path(kw.get('cwd') or ROOT).resolve()
+ if not inside(cwd, (ROOT, self.temporary)):
+ raise IsolationDenied('unit isolation: non-fixture working directory')
+ kw['cwd'] = str(cwd)
+ # env supplied by tests is never propagated: even synthetic token mutations
+ # remain in the parent. All approved children have only this fresh environment.
+ kw['env'] = clean_environment(self.temporary / 'home', self.temporary)
+ kw['env'].update(GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null', GIT_OPTIONAL_LOCKS='0')
+ self.launch = tuple(command)
+ try:
+ return self.original_popen(command, **kw)
+ finally:
+ self.launch = None
+
+ def install(self):
+ sys.addaudithook(self.audit)
+ subprocess.Popen = self.popen
+ guard = self
+ original_socket = socket.socket
+ class FixtureSocket(original_socket):
+ def bind(self, address):
+ super().bind(address)
+ guard.endpoints.add(self.getsockname())
+ socket.socket = FixtureSocket
+ def numeric_addresses(host, port, family=0, type=0, proto=0, flags=0):
+ if not guard.loopback_test or (host, port) not in guard.endpoints:
+ raise IsolationDenied('unit isolation: DNS/external endpoint denied')
+ return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', (host, port))]
+ socket.getaddrinfo = numeric_addresses
+ def local_name(host=''):
+ if not guard.loopback_test or host != '127.0.0.1':
+ raise IsolationDenied('unit isolation: hostname lookup denied')
+ return 'localhost'
+ socket.getfqdn = local_name
+
+ def pytest_configure(self, config):
+ config._unit_guard = self
+
+ # Exact single synthetic HTTP test only; no other test gets a loopback lease.
+ def pytest_runtest_setup(self, item):
+ self.loopback_test = item.nodeid == 'tests/test_desk.py::test_approve_flow_http_dry'
+ self.endpoints.clear()
+
+ def pytest_runtest_teardown(self, item):
+ self.loopback_test = False
+ self.endpoints.clear()
+
+
+def worker(temporary: Path, arguments: list[str], netkingdom: Path | None, phase: Path | None):
+ sys.dont_write_bytecode = True
+ tempfile.tempdir = str(temporary)
+ paths = [ROOT / 'src']
+ if phase:
+ paths.append(phase / 'src')
+ # Explicit trusted source roots only, never ambient PYTHONPATH/plugins.
+ sys.path[:0] = [str(p) for p in paths]
+ roots = [ROOT, *paths, Path(sys.prefix), Path(sys.base_prefix), temporary]
+ if phase:
+ roots.append(phase / 'tests/fixtures/ops-warden-coordination-profile.json')
+ if netkingdom:
+ roots.append(netkingdom)
+ os.environ['WARDEN_TEST_NETKINGDOM_ROOT'] = str(netkingdom)
+ guard = Guard(temporary, roots)
+ guard.install()
+ # Pytest itself and collection run only after installation.
+ import pytest
+ raise SystemExit(pytest.main(['-p', 'no:cacheprovider', '--basetemp', str(temporary / 'pytest'), *arguments], plugins=[guard]))
+
+
+def main():
+ if sys.argv[1:2] == ['--child']:
+ payload = json.loads(sys.argv[2])
+ temporary = Path(payload['temporary'])
+ sys.dont_write_bytecode = True
+ tempfile.tempdir = str(temporary)
+ sys.path.insert(0, str(ROOT / 'src'))
+ guard = Guard(temporary, [Path(p) for p in payload['read_roots']], child=True,
+ git_metadata=payload.get('git_metadata', False))
+ guard.install()
+ sys.argv = [payload.get('script', '-c'), *payload['args']]
+ if 'script' in payload:
+ runpy.run_path(payload['script'], run_name='__main__')
+ else:
+ exec(compile(payload['code'], '<reviewed-unit-child>', 'exec'), {'__name__': '__main__'})
+ return
+ if sys.argv[1:2] == ['--worker']:
+ temporary = Path(sys.argv[2])
+ netkingdom = Path(sys.argv[3]) if sys.argv[3] else None
+ phase = Path(sys.argv[4]) if sys.argv[4] else None
+ worker(temporary, sys.argv[5:], netkingdom, phase)
+ return
+ import argparse
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument('--netkingdom-root', type=Path, default=ROOT.parent / 'net-kingdom')
+ parser.add_argument('--phase-memory-root', type=Path, default=ROOT.parent / 'phase-memory')
+ options, arguments = parser.parse_known_args()
+ netkingdom = options.netkingdom_root.resolve() if options.netkingdom_root.is_dir() else None
+ phase = options.phase_memory_root.resolve() if (options.phase_memory_root / 'src').is_dir() else None
+ with tempfile.TemporaryDirectory(prefix='warden-unit-') as directory:
+ temporary = Path(directory)
+ home = temporary / 'home'
+ home.mkdir()
+ result = subprocess.run([sys.executable, '-I', str(SELF), '--worker', str(temporary),
+ str(netkingdom) if netkingdom else '', str(phase) if phase else '', *arguments],
+ cwd=ROOT, env=clean_environment(home, temporary))
+ raise SystemExit(result.returncode)
+
+
+if __name__ == '__main__':
+ main()
--- /dev/null
+++ b/tests/test_safe_unit_runner.py
@@ -0,0 +1,81 @@
+"""Exercise installed parent and nested child controls, never real credentials."""
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import sys
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+@pytest.fixture
+def guarded(request):
+ guard = getattr(request.config, '_unit_guard', None)
+ if guard is None:
+ pytest.skip('requires make test / scripts/run_unit_tests.py')
+ return guard
+
+
+def test_parent_environment_is_allowlisted_and_home_is_temporary(guarded):
+ # The external regression invocation sets synthetic BAO_TOKEN/VAULT_TOKEN/HOME.
+ assert 'BAO_TOKEN' not in os.environ
+ assert 'VAULT_TOKEN' not in os.environ
+ assert Path(os.environ['TMPDIR']) in Path.home().parents
+ assert 'PYTHONPATH' not in os.environ
+
+
+@pytest.mark.parametrize('command', [
+ ['bao', 'token', 'lookup'], ['ssh', 'railiance01'], ['curl', 'http://127.0.0.1:8000'],
+ [sys.executable, '-c', 'import os; print(os.environ)'],
+])
+def test_unexpected_process_refused_before_launch(guarded, command):
+ with pytest.raises(PermissionError, match='unit isolation'):
+ subprocess.run(command, check=True)
+
+
+def test_parent_socket_denies_even_unregistered_loopback(guarded):
+ with pytest.raises(PermissionError, match='unit isolation'):
+ socket.socket().connect(('127.0.0.1', 8000))
+
+
+@pytest.mark.parametrize('path', ['/home/synthetic-ambient-unit-fixture/.vault-token',
+ '/root/.kube/config', '/proc/self/environ',
+ '/tmp/warden-synthetic-ambient-home/.vault-token'])
+def test_ambient_paths_refused_without_opening_them(guarded, path):
+ with pytest.raises(PermissionError, match='unit isolation'):
+ Path(path).read_bytes()
+
+
+def test_symlink_cannot_escape_fixture_root(guarded, tmp_path):
+ link = tmp_path / 'ambient'
+ link.symlink_to('/home/synthetic-ambient-unit-fixture/.vault-token')
+ with pytest.raises(PermissionError, match='unit isolation'):
+ link.read_bytes()
+
+
+def test_child_drops_even_synthetic_token_environment(guarded, monkeypatch):
+ monkeypatch.setenv('BAO_TOKEN', 'synthetic-only-never-a-real-token')
+ monkeypatch.setenv('VAULT_TOKEN', 'synthetic-only-never-a-real-token')
+ monkeypatch.setenv('HOME', '/home/synthetic-ambient-unit-fixture')
+ result = subprocess.run([sys.executable, str(ROOT / 'tests/unit_isolation_probe.py'), 'state'],
+ capture_output=True, text=True, check=True, env=dict(os.environ))
+ assert json.loads(result.stdout) == {'bao_token_present': False,
+ 'vault_token_present': False, 'home_in_temporary': True}
+
+
+@pytest.mark.parametrize('mode', ['socket', 'provider', 'home', 'nested'])
+def test_guard_exists_in_each_allowed_python_child(guarded, mode):
+ result = subprocess.run([sys.executable, str(ROOT / 'tests/unit_isolation_probe.py'), mode],
+ capture_output=True, text=True, check=True)
+ assert result.stdout == 'denied\n'
+
+
+def test_forged_provider_double_refused(guarded, tmp_path):
+ provider = tmp_path / 'bao'
+ provider.write_text('#!/bin/bash\necho not-the-reviewed-fixture\n')
+ provider.chmod(0o700)
+ with pytest.raises(PermissionError, match='double bytes differ'):
+ subprocess.run([str(provider), 'token', 'lookup'], check=True)
--- /dev/null
+++ b/tests/unit_isolation_probe.py
@@ -0,0 +1,29 @@
+"""Closed child regression probes; executed only under the unit runner guard."""
+import json
+import os
+from pathlib import Path
+import socket
+import subprocess
+import sys
+
+mode = sys.argv[1]
+if mode == 'state':
+ print(json.dumps({'bao_token_present': 'BAO_TOKEN' in os.environ,
+ 'vault_token_present': 'VAULT_TOKEN' in os.environ,
+ 'home_in_temporary': Path(os.environ['TMPDIR']) in Path.home().parents}))
+else:
+ try:
+ if mode == 'socket':
+ socket.socket().connect(('127.0.0.1', 8000))
+ elif mode == 'provider':
+ subprocess.run(['bao', 'token', 'lookup'], check=True)
+ elif mode == 'home':
+ Path('/home/synthetic-ambient-unit-fixture/.vault-token').read_text()
+ elif mode == 'nested':
+ subprocess.run([sys.executable, '-c', 'print("SYNTHETIC-SECRET")'], check=True)
+ else:
+ raise AssertionError('unknown probe')
+ except PermissionError:
+ print('denied')
+ else:
+ raise AssertionError('external operation was not refused')
--- a/tests/test_layer_conformance.py
+++ b/tests/test_layer_conformance.py
@@ -273,7 +273,11 @@
# --- classification coverage (v0.8 §6.4 obligation 3) -------------------------
def test_published_coverage_equals_measured_coverage():
- """The published figure must equal what the repo actually measures.
+ """The published figure must equal its explicit public declaration snapshot.
+
+ This is a deterministic source check, not a claim about current estate state.
+ Live measurement remains scripts/report_coverage.py with an explicit estate
+ root; tests never treat the developer HOME as an authority or fixture.
ops-warden asked gate-house for §13.1's Coverage column and its figures are
that column's first entries, so their accuracy is ours to hold. The register
@@ -295,7 +299,7 @@
published = yaml.safe_load((repo / "pep-stance.yaml").read_text())[
"classification_coverage"
]
- measured = module.measure()
+ measured = module.measure(repo / "tests/fixtures/workload-coverage")
for population in ("signing_targets", "routing_lanes"):
assert published[population] == measured[population], population
--- a/tests/test_workload_join.py
+++ b/tests/test_workload_join.py
@@ -10,7 +10,8 @@
def test_repo_catalog_uses_only_explicit_workload_references():
- report = build(ROOT / "registry/routing/catalog.yaml", Path.home())
+ report = build(ROOT / "registry/routing/catalog.yaml",
+ ROOT / "tests/fixtures/workload-coverage")
assert report["ok"] is True
assert len(report["resolved"]) == 4
assert len(report["unknown"]) == 22
--- a/tests/test_case_plan.py
+++ b/tests/test_case_plan.py
@@ -2,6 +2,7 @@
from __future__ import annotations
import copy
+import os
from pathlib import Path
import pytest
@@ -12,7 +13,8 @@
from warden.cli import app
from warden.routing.catalog import load_catalog
-LIVE_CASE = Path.home() / "net-kingdom/security-cases/forgejo-netkingdom-oidc.yaml"
+LIVE_CASE = Path(os.environ.get("WARDEN_TEST_NETKINGDOM_ROOT",
+ Path(__file__).resolve().parents[2] / "net-kingdom")) / "security-cases/forgejo-netkingdom-oidc.yaml"
TYPE = {
"apiVersion": "netkingdom.io/security-case-type/v0.1",
--- a/tests/test_case_type.py
+++ b/tests/test_case_type.py
@@ -2,6 +2,7 @@
from __future__ import annotations
import copy
+import os
from pathlib import Path
import pytest
@@ -12,7 +13,8 @@
from warden.cli import app
from warden.routing.catalog import load_catalog
-LIVE_TYPE = Path.home() / "net-kingdom/security-case-types/oidc-browser-relying-party/type.yaml"
+LIVE_TYPE = Path(os.environ.get("WARDEN_TEST_NETKINGDOM_ROOT",
+ Path(__file__).resolve().parents[2] / "net-kingdom")) / "security-case-types/oidc-browser-relying-party/type.yaml"
BASE = {
"apiVersion": "netkingdom.io/security-case-type/v0.1",
--- /dev/null
+++ b/docs/development/safe-unit-tests.md
@@ -0,0 +1,64 @@
+# Safe unit-test entry point
+
+Use `make test`, or `.venv/bin/python -I scripts/run_unit_tests.py`. Dependencies
+must already be installed; this path does not run uv, synchronize or download.
+`UNIT_PYTHON` may select another existing interpreter. `PHASE_MEMORY_REPO` remains
+an explicit local source option. Pass pytest options with `TEST_ARGS`, for example
+`make test TEST_ARGS='-q tests/test_safe_unit_runner.py'`.
+
+The launcher creates an empty temporary HOME and temporary fixture root and starts
+an isolated (`-I`) Python interpreter with a constructed environment. Caller token,
+proxy, Python, pytest, XDG and Warden configuration variables are never copied or
+inventoried. Third-party pytest plugin autoload is disabled. Tests can explicitly
+introduce synthetic variables with monkeypatch, but allowed children always get
+the original temporary HOME and a fresh environment, never those variables.
+
+Before pytest import/collection, a Python audit hook and subprocess wrapper:
+
+- deny file opens outside the selected source roots, installed interpreter/library
+ roots and fixture root; writes through open are limited to the fixture root;
+- deny sockets, DNS and process creation except the closed fixture inventory;
+- refuse shell, executable, preexec and inherited-FD overrides outside that inventory;
+- install the same file/socket/process denial before executing every allowed
+ Python child, including BrowserInputs' exact validator code and reverified copied
+ module hashes. Validator children cannot start further processes.
+
+The process inventory in `scripts/run_unit_tests.py` is intentionally small:
+BrowserInputs validation, four metadata-checker scripts with fixed argument shapes,
+the exact original-intent restart probe, the synthetic containment print, and
+closed guard-regression probes. Metadata checkers receive only fixture output
+paths. Only the generated-artifact checker child can run the exact read-only Git
+log/status queries for this repository's catalog, with global/system config and
+optional locks disabled, hooks disabled and fsmonitor disabled. Those local Git
+commands are not provider operations.
+
+Two reviewed non-Python cases remain: exact-byte, temporary contract-double scripts
+with a closed argv list, and five exact fixed `printf` commands used by existing
+transport tests. They run Bash without profile/rc startup and without inherited
+environment; no arbitrary shell text, provider binary, SSH, kubectl or OpenBao
+command is permitted. Default unit selection excludes four integration-marked
+SSH-keygen tests. Selecting them explicitly does not grant their native commands.
+
+Only `test_desk.py::test_approve_flow_http_dry` receives a network lease. It may
+bind `127.0.0.1:0` and connect to the endpoint actually allocated to that test's
+socket. DNS is replaced by numeric fixture resolution. Another loopback service
+(e.g. the real Hub on port 8000), Unix sockets and remote endpoints stay denied.
+The lease and registered endpoints are cleared after the test.
+
+Coverage tests now use the explicit public declaration projection snapshot in
+`tests/fixtures/workload-coverage`, with provenance hashes. They compare the
+published source snapshot rather than silently consulting the workstation HOME.
+Run the existing coverage report separately with an explicit estate root when
+measuring current estate state. Tests of the acknowledged NetKingdom source use
+`--netkingdom-root` (default: sibling source checkout), never a HOME symlink. The
+optional Phase Memory library's one existing public coordination-profile fixture
+is explicitly readable alongside its source; no runtime memory store is inherited.
+
+This is an accident guard for trusted unit tests, not a security sandbox for
+hostile Python, native extensions or hostile installed dependencies. It does not
+revoke already-open descriptors, prevent every filesystem mutation syscall, or
+provide OS process/network isolation. Fixed Bash/Git fixture processes rely on
+the reviewed executable and source/argument shapes; Python audit hooks do not run
+inside those binaries. Do not run untrusted test code under a privileged runtime.
+Direct `pytest` bypasses this launcher and is not equivalent validation. No
+production Warden boundary, grant, catalog, source pin or owner credential changes.
--- a/workplans/WARDEN-WP-0044-security-case-follow-through.md
+++ b/workplans/WARDEN-WP-0044-security-case-follow-through.md
@@ -354,3 +354,25 @@
No credential lookup, live investigation or rotation is authorized by this note.
T03 remains progress; contained sessions, authenticated owner provenance,
materialization, exact type acknowledgment/pin and first native case remain open.
+
+
+2026-10-17 safe unit-runner source candidate: `make test` now selects an existing
+interpreter and `scripts/run_unit_tests.py`, which constructs the environment and
+empty temporary HOME before pytest collection, denies unexpected process/socket
+use and ambient file opens, and guards the pinned BrowserInputs child too.
+Closed fixture exceptions cover the synthetic desk endpoint, metadata checkers,
+exact-byte contract doubles and fixed transport prints; they confer no provider
+authority. Explicit public declaration fixtures replace ambient HOME source
+lookups. See `docs/development/safe-unit-tests.md` and Custodian packet
+`docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1`.
+Synthetic token, ambient-path, forged-provider and nested-child regressions are
+retained alongside the initial failed isolation runs and corrected results. This
+is a source candidate for the existing safe-test handoff, not an OS sandbox or
+native case acceptance. T03 remains progress; contained sessions, authenticated
+owner provenance, materialization, exact type acknowledgment/pin and first native
+case remain open. No credential lookup, live investigation or rotation occurred.
+
+Candidate validation: ordinary `make test` under synthetic ambient token/HOME
+controls passes 783 tests with four integration-marked tests deselected. Affected
+focused checks pass 117 tests; new runner code/regressions pass Ruff. Retained
+initial failures exposed fixture dependencies and were corrected, not suppressed.
--- /dev/null
+++ b/tests/fixtures/workload-coverage/provenance.json
@@ -0,0 +1,20 @@
+[
+ {
+ "repo": "rapp-core-hub",
+ "source": "retained public snapshot from warden-1.3-kernel-fixes-r1 test preparation",
+ "source_sha256": "98e81fb1116aa0ac9dc1a3e7aef9dd87c6d2110d98d46fbfca2f76eec8f3a77b",
+ "projection_sha256": "9ce902da0c77809f1094a634244652835b162f59682e9fa6923d8c8aa4ed4a3c"
+ },
+ {
+ "repo": "rapp-issue-core",
+ "source": "retained public snapshot from warden-1.3-kernel-fixes-r1 test preparation",
+ "source_sha256": "0425c9fe84e26b359b2a35fc37ef8fcb781de1aaa6229c59fcd246ee523514a5",
+ "projection_sha256": "a1956deecf2f0fbca44549e5ea5e8f1c4b74ebf91a9482c64c982491af0d923b"
+ },
+ {
+ "repo": "rapp-qonto",
+ "source": "retained public snapshot from warden-1.3-kernel-fixes-r1 test preparation",
+ "source_sha256": "6279dc2f3ab09519dffa1ceeedb2095f1182960e0106109a87aa5c0052446883",
+ "projection_sha256": "15aa9809e8d6a120c516be5ce9adfb72c00382140c171e791083ad117f503b3f"
+ }
+]
--- /dev/null
+++ b/tests/fixtures/workload-coverage/rapp-core-hub/declarations/rapp.yaml
@@ -0,0 +1,15 @@
+# Public declaration projection for the published coverage snapshot; not live evidence.
+rapp_id: rapp-core-hub
+workload_identity:
+ name: core-hub
+composition:
+ member_repos:
+ - deployables:
+ - core-hub-api
+ - core-hub-migrate
+ - hub-core-candidate
+ - hub-core-candidate-migrate
+ - repository-classification-publisher
+data_classification: confidential
+criticality: high
+readiness_state: verified
--- /dev/null
+++ b/tests/fixtures/workload-coverage/rapp-issue-core/declarations/rapp.yaml
@@ -0,0 +1,11 @@
+# Public declaration projection for the published coverage snapshot; not live evidence.
+rapp_id: rapp-issue-core
+workload_identity:
+ name: issue-core
+composition:
+ member_repos:
+ - deployables:
+ - issue-core
+data_classification: confidential
+criticality: high
+readiness_state: verified
--- /dev/null
+++ b/tests/fixtures/workload-coverage/rapp-qonto/declarations/rapp.yaml
@@ -0,0 +1,11 @@
+# Public declaration projection for the published coverage snapshot; not live evidence.
+rapp_id: rapp-qonto
+workload_identity:
+ name: qonto
+composition:
+ member_repos:
+ - deployables:
+ - rapp-qonto
+data_classification: restricted
+criticality: critical
+readiness_state: verified

50
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/Makefile Normal file
View file

@ -0,0 +1,50 @@
# ops-warden development and install targets
ROOT := $(dir $(abspath $(lastword $(MAKEFILE_LIST))))
UNIT_PYTHON ?= $(ROOT)/.venv/bin/python
PHASE_MEMORY_REPO ?= $(abspath $(ROOT)/../phase-memory)
.DEFAULT_GOAL := help
.PHONY: help sync install install-warden install-memory install-all check-memory verify-memory test lint
help:
@echo "ops-warden make targets"
@echo ""
@echo " make sync # uv sync (dev dependencies)"
@echo " make install # install warden CLI (SSH only, no phase-memory)"
@echo " make install-memory # reinstall warden with editable phase-memory"
@echo " make install-all # sync + warden + phase-memory (recommended)"
@echo " make check-memory # fail unless phase-memory is importable by warden"
@echo " make verify-memory # check-memory + warden memory status"
@echo " make test # unit tests"
@echo " make lint # ruff check"
@echo ""
@echo "Override phase-memory path: make install-all PHASE_MEMORY_REPO=/path/to/phase-memory"
sync:
uv sync
install-warden:
uv tool install . --force --reinstall-package ops-warden
install-memory:
@test -d "$(PHASE_MEMORY_REPO)" || (echo "ERROR: phase-memory not found at $(PHASE_MEMORY_REPO). Clone it or set PHASE_MEMORY_REPO."; exit 1)
@test -f "$(PHASE_MEMORY_REPO)/pyproject.toml" || (echo "ERROR: $(PHASE_MEMORY_REPO) is not a phase-memory checkout."; exit 1)
uv tool install . --with-editable "$(PHASE_MEMORY_REPO)" --force --reinstall-package ops-warden
install: install-warden
install-all: sync install-memory
check-memory:
@warden memory status --json >/dev/null 2>&1 || (echo "ERROR: phase-memory is not available to warden. Run: make install-memory"; exit 1)
@echo "phase-memory: ok (warden memory status succeeded)"
verify-memory: check-memory
@warden memory status
test:
"$(UNIT_PYTHON)" -I scripts/run_unit_tests.py --phase-memory-root "$(PHASE_MEMORY_REPO)" $(TEST_ARGS)
lint:
uv run ruff check .

64
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/docs/development/safe-unit-tests.md Normal file
View file

@ -0,0 +1,64 @@
# Safe unit-test entry point
Use `make test`, or `.venv/bin/python -I scripts/run_unit_tests.py`. Dependencies
must already be installed; this path does not run uv, synchronize or download.
`UNIT_PYTHON` may select another existing interpreter. `PHASE_MEMORY_REPO` remains
an explicit local source option. Pass pytest options with `TEST_ARGS`, for example
`make test TEST_ARGS='-q tests/test_safe_unit_runner.py'`.
The launcher creates an empty temporary HOME and temporary fixture root and starts
an isolated (`-I`) Python interpreter with a constructed environment. Caller token,
proxy, Python, pytest, XDG and Warden configuration variables are never copied or
inventoried. Third-party pytest plugin autoload is disabled. Tests can explicitly
introduce synthetic variables with monkeypatch, but allowed children always get
the original temporary HOME and a fresh environment, never those variables.
Before pytest import/collection, a Python audit hook and subprocess wrapper:
- deny file opens outside the selected source roots, installed interpreter/library
roots and fixture root; writes through open are limited to the fixture root;
- deny sockets, DNS and process creation except the closed fixture inventory;
- refuse shell, executable, preexec and inherited-FD overrides outside that inventory;
- install the same file/socket/process denial before executing every allowed
Python child, including BrowserInputs' exact validator code and reverified copied
module hashes. Validator children cannot start further processes.
The process inventory in `scripts/run_unit_tests.py` is intentionally small:
BrowserInputs validation, four metadata-checker scripts with fixed argument shapes,
the exact original-intent restart probe, the synthetic containment print, and
closed guard-regression probes. Metadata checkers receive only fixture output
paths. Only the generated-artifact checker child can run the exact read-only Git
log/status queries for this repository's catalog, with global/system config and
optional locks disabled, hooks disabled and fsmonitor disabled. Those local Git
commands are not provider operations.
Two reviewed non-Python cases remain: exact-byte, temporary contract-double scripts
with a closed argv list, and five exact fixed `printf` commands used by existing
transport tests. They run Bash without profile/rc startup and without inherited
environment; no arbitrary shell text, provider binary, SSH, kubectl or OpenBao
command is permitted. Default unit selection excludes four integration-marked
SSH-keygen tests. Selecting them explicitly does not grant their native commands.
Only `test_desk.py::test_approve_flow_http_dry` receives a network lease. It may
bind `127.0.0.1:0` and connect to the endpoint actually allocated to that test's
socket. DNS is replaced by numeric fixture resolution. Another loopback service
(e.g. the real Hub on port 8000), Unix sockets and remote endpoints stay denied.
The lease and registered endpoints are cleared after the test.
Coverage tests now use the explicit public declaration projection snapshot in
`tests/fixtures/workload-coverage`, with provenance hashes. They compare the
published source snapshot rather than silently consulting the workstation HOME.
Run the existing coverage report separately with an explicit estate root when
measuring current estate state. Tests of the acknowledged NetKingdom source use
`--netkingdom-root` (default: sibling source checkout), never a HOME symlink. The
optional Phase Memory library's one existing public coordination-profile fixture
is explicitly readable alongside its source; no runtime memory store is inherited.
This is an accident guard for trusted unit tests, not a security sandbox for
hostile Python, native extensions or hostile installed dependencies. It does not
revoke already-open descriptors, prevent every filesystem mutation syscall, or
provide OS process/network isolation. Fixed Bash/Git fixture processes rely on
the reviewed executable and source/argument shapes; Python audit hooks do not run
inside those binaries. Do not run untrusted test code under a privileged runtime.
Direct `pytest` bypasses this launcher and is not equivalent validation. No
production Warden boundary, grant, catalog, source pin or owner credential changes.

283
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/scripts/run_unit_tests.py Normal file
View file

@ -0,0 +1,283 @@
"""Run trusted unit tests with no inherited credentials or external I/O.
This is an accident guard for this suite, not an OS sandbox for hostile Python.
Direct pytest bypasses it. Use make test (already installed dependencies only).
"""
from __future__ import annotations
import hashlib
import json
import os
from pathlib import Path
import runpy
import socket
import subprocess
import sys
import tempfile
ROOT = Path(__file__).resolve().parents[1]
SELF = Path(__file__).resolve()
SHELL_FIXTURES = {
"printf 'the-secret-value'", "printf 'sekret'", "printf 'new'",
"printf '%s' '{\"wrap_info\":{\"token\":\"hvs.WRAP\"}}'",
"printf 'not-json'",
}
JOURNAL_CODE_SHA256 = '69aa18f302fd698104cb2253b4af0c3a077e72a0a427183e3dce1444c4926df9'
class IsolationDenied(PermissionError, subprocess.SubprocessError):
"""A unit test attempted an unreviewed external operation."""
def clean_environment(home: Path, temporary: Path) -> dict[str, str]:
# Construct, never copy/inventory the caller's environment.
return {'HOME': str(home), 'TMPDIR': str(temporary), 'PATH': '/usr/bin:/bin',
'LANG': 'C.UTF-8', 'PYTHONDONTWRITEBYTECODE': '1',
'PYTEST_DISABLE_PLUGIN_AUTOLOAD': '1'}
def inside(path: Path, roots: tuple[Path, ...]) -> bool:
return any(path == root or root in path.parents for root in roots)
class Guard:
def __init__(self, temporary: Path, read_roots: list[Path], child: bool = False, git_metadata: bool = False):
self.temporary = temporary.resolve()
self.read_roots = tuple(p.resolve() for p in read_roots)
self.child = child
self.git_metadata = git_metadata
self.loopback_test = False
self.endpoints: set[tuple[str, int]] = set()
self.launch = None
self.original_popen = subprocess.Popen
def fixture_path(self, value: str) -> Path:
path = Path(value).resolve()
if not inside(path, (self.temporary,)):
raise IsolationDenied('unit isolation: non-fixture path')
return path
def audit(self, event, args):
if event == 'open' and not isinstance(args[0], int):
path = Path(os.fsdecode(args[0])).resolve()
mode, flags = args[1:3]
write = ((isinstance(mode, str) and any(c in mode for c in 'wax+'))
or (isinstance(flags, int) and flags & (os.O_WRONLY | os.O_RDWR | os.O_CREAT)))
roots = (self.temporary,) if write else self.read_roots
if path != Path('/dev/null') and not inside(path, roots):
raise IsolationDenied('unit isolation: file outside reviewed roots')
if event == 'subprocess.Popen':
if self.launch is None or tuple(args[1]) != self.launch:
raise IsolationDenied('unit isolation: unexpected process')
if event in {'os.system', 'os.exec', 'os.posix_spawn', 'os.fork', 'os.forkpty'}:
raise IsolationDenied('unit isolation: native execution disabled')
if event.startswith('socket.'):
if not self.loopback_test or self.child:
raise IsolationDenied('unit isolation: sockets disabled')
if event == 'socket.__new__':
if args[1] != socket.AF_INET or args[2] & 15 != socket.SOCK_STREAM:
raise IsolationDenied('unit isolation: socket family/type denied')
elif event == 'socket.bind':
if args[1] != ('127.0.0.1', 0):
raise IsolationDenied('unit isolation: only ephemeral fixture bind')
elif event == 'socket.connect':
if args[1] not in self.endpoints:
raise IsolationDenied('unit isolation: only this fixture endpoint')
elif event not in {'socket.gethostname'}:
raise IsolationDenied('unit isolation: DNS/socket operation denied')
def python_payload(self, argv):
"""Closed subprocess inventory; every allowed Python child installs guard."""
from warden import case_inputs
if (len(argv) == 5 and argv[:3] == [sys.executable, '-I', '-c']
and argv[3] == case_inputs._VALIDATE):
directory = self.fixture_path(argv[4])
if not directory.name.startswith('warden-case-validator-'):
raise IsolationDenied('unit isolation: validator fixture directory differs')
# Reverify copied modules immediately before launch, independently of caller.
for relative, digest in case_inputs.SOURCE_PINS.items():
if hashlib.sha256((directory / relative).read_bytes()).hexdigest() != digest:
raise IsolationDenied('unit isolation: validator source pin differs')
return {'code': argv[3], 'args': [str(directory)]}
if len(argv) == 3 and argv[:2] == [sys.executable, '-c'] and argv[2] == 'print("SYNTHETIC-SECRET")':
return {'code': argv[2], 'args': []}
if (len(argv) == 6 and argv[:3] == [sys.executable, '-B', '-c']
and hashlib.sha256(argv[3].encode()).hexdigest() == JOURNAL_CODE_SHA256):
self.fixture_path(argv[4])
if len(argv[5]) > 192:
raise IsolationDenied('unit isolation: journal source bound')
return {'code': argv[3], 'args': argv[4:]}
if len(argv) >= 2 and argv[0] in (sys.executable, 'python3'):
script = Path(argv[1]).resolve()
args = argv[2:]
if script == ROOT / 'tests/unit_isolation_probe.py' and len(args) == 1 and args[0] in {'state', 'socket', 'provider', 'home', 'nested'}:
return {'script': str(script), 'args': args}
if script == ROOT / 'scripts/check_layer_conformance.py' and not args:
return {'script': str(script), 'args': args}
if script == ROOT / 'scripts/emit_high_risk_paths.py' and args == ['--check']:
return {'script': str(script), 'args': args, 'git_metadata': True}
if script == ROOT / 'scripts/check_principals_drift.py' and len(args) == 4 and args[::2] == ['--inventory', '--infra']:
for value in args[1::2]:
self.fixture_path(value)
return {'script': str(script), 'args': args}
if script == ROOT / 'scripts/build_flex_auth_registry.py':
if (len(args) in (3, 5) and Path(args[0]).resolve() == ROOT / 'examples/inventory.seed.yaml'
and args[-2] == '-o' and (len(args) == 3 or args[1] == '--zone-resolutions')):
self.fixture_path(args[-1])
if len(args) == 5:
self.fixture_path(args[2])
return {'script': str(script), 'args': args}
raise IsolationDenied('unit isolation: command not in reviewed subprocess inventory')
def popen(self, argv, *pos, **kw):
if pos or kw.get('executable') or kw.get('preexec_fn') or kw.get('pass_fds') or kw.get('close_fds') is False:
raise IsolationDenied('unit isolation: unexpected process options')
if kw.pop('shell', False):
if self.child or not isinstance(argv, str) or argv not in SHELL_FIXTURES:
raise IsolationDenied('unit isolation: unreviewed shell command')
return self.spawn(['/bin/bash', '--noprofile', '--norc', '-c', argv], kw)
if not isinstance(argv, (list, tuple)):
raise IsolationDenied('unit isolation: argv list required')
argv = [os.fspath(x) for x in argv]
# Never run a real provider: the only shell fixture is an exact-byte double.
from warden.doubles import _DOUBLES
metadata_commands = (['git', 'log', '-1', '--format=%H %cI', '--', str(ROOT / 'registry/routing/catalog.yaml')],
['git', 'status', '--porcelain', '--', str(ROOT / 'registry/routing/catalog.yaml')])
if self.child and not (self.git_metadata and argv in metadata_commands):
raise IsolationDenied('unit isolation: nested process disabled')
candidate = Path(argv[0])
if not candidate.is_absolute():
import shutil
resolved = shutil.which(argv[0], path=kw.get('env', os.environ).get('PATH', ''))
candidate = Path(resolved) if resolved else candidate
if self.child and self.git_metadata:
if Path(kw.get('cwd', '.')).resolve() != ROOT:
raise IsolationDenied('unit isolation: metadata cwd differs')
command = ['/usr/bin/git', '-c', 'core.fsmonitor=false', '-c', 'core.hooksPath=/dev/null', '--no-pager', *argv[1:]]
elif candidate.is_absolute() and inside(candidate.resolve(), (self.temporary,)) and candidate.name in _DOUBLES:
if candidate.read_text() != _DOUBLES[candidate.name].script:
raise IsolationDenied('unit isolation: double bytes differ')
tail = argv[1:]
if (candidate.name == 'bao' and tail not in (
['kv', 'get', '-field=NPM_AUTH_TOKEN', 'platform/x/y'],
['kv', 'get', '-field=API_KEY', 'platform/x/y/z'],
['login', '-method=oidc'], ['write', 'secret/x'])) or (candidate.name == 'key-cape' and tail != ['login']):
raise IsolationDenied('unit isolation: double arguments differ')
command = ['/bin/bash', '--noprofile', '--norc', str(candidate), *tail]
else:
payload = self.python_payload(argv)
payload.update(temporary=str(self.temporary), read_roots=[str(x) for x in self.read_roots])
command = [sys.executable, '-I', str(SELF), '--child', json.dumps(payload)]
return self.spawn(command, kw)
def spawn(self, command, kw):
cwd = Path(kw.get('cwd') or ROOT).resolve()
if not inside(cwd, (ROOT, self.temporary)):
raise IsolationDenied('unit isolation: non-fixture working directory')
kw['cwd'] = str(cwd)
# env supplied by tests is never propagated: even synthetic token mutations
# remain in the parent. All approved children have only this fresh environment.
kw['env'] = clean_environment(self.temporary / 'home', self.temporary)
kw['env'].update(GIT_CONFIG_NOSYSTEM='1', GIT_CONFIG_GLOBAL='/dev/null', GIT_OPTIONAL_LOCKS='0')
self.launch = tuple(command)
try:
return self.original_popen(command, **kw)
finally:
self.launch = None
def install(self):
sys.addaudithook(self.audit)
subprocess.Popen = self.popen
guard = self
original_socket = socket.socket
class FixtureSocket(original_socket):
def bind(self, address):
super().bind(address)
guard.endpoints.add(self.getsockname())
socket.socket = FixtureSocket
def numeric_addresses(host, port, family=0, type=0, proto=0, flags=0):
if not guard.loopback_test or (host, port) not in guard.endpoints:
raise IsolationDenied('unit isolation: DNS/external endpoint denied')
return [(socket.AF_INET, socket.SOCK_STREAM, 6, '', (host, port))]
socket.getaddrinfo = numeric_addresses
def local_name(host=''):
if not guard.loopback_test or host != '127.0.0.1':
raise IsolationDenied('unit isolation: hostname lookup denied')
return 'localhost'
socket.getfqdn = local_name
def pytest_configure(self, config):
config._unit_guard = self
# Exact single synthetic HTTP test only; no other test gets a loopback lease.
def pytest_runtest_setup(self, item):
self.loopback_test = item.nodeid == 'tests/test_desk.py::test_approve_flow_http_dry'
self.endpoints.clear()
def pytest_runtest_teardown(self, item):
self.loopback_test = False
self.endpoints.clear()
def worker(temporary: Path, arguments: list[str], netkingdom: Path | None, phase: Path | None):
sys.dont_write_bytecode = True
tempfile.tempdir = str(temporary)
paths = [ROOT / 'src']
if phase:
paths.append(phase / 'src')
# Explicit trusted source roots only, never ambient PYTHONPATH/plugins.
sys.path[:0] = [str(p) for p in paths]
roots = [ROOT, *paths, Path(sys.prefix), Path(sys.base_prefix), temporary]
if phase:
roots.append(phase / 'tests/fixtures/ops-warden-coordination-profile.json')
if netkingdom:
roots.append(netkingdom)
os.environ['WARDEN_TEST_NETKINGDOM_ROOT'] = str(netkingdom)
guard = Guard(temporary, roots)
guard.install()
# Pytest itself and collection run only after installation.
import pytest
raise SystemExit(pytest.main(['-p', 'no:cacheprovider', '--basetemp', str(temporary / 'pytest'), *arguments], plugins=[guard]))
def main():
if sys.argv[1:2] == ['--child']:
payload = json.loads(sys.argv[2])
temporary = Path(payload['temporary'])
sys.dont_write_bytecode = True
tempfile.tempdir = str(temporary)
sys.path.insert(0, str(ROOT / 'src'))
guard = Guard(temporary, [Path(p) for p in payload['read_roots']], child=True,
git_metadata=payload.get('git_metadata', False))
guard.install()
sys.argv = [payload.get('script', '-c'), *payload['args']]
if 'script' in payload:
runpy.run_path(payload['script'], run_name='__main__')
else:
exec(compile(payload['code'], '<reviewed-unit-child>', 'exec'), {'__name__': '__main__'})
return
if sys.argv[1:2] == ['--worker']:
temporary = Path(sys.argv[2])
netkingdom = Path(sys.argv[3]) if sys.argv[3] else None
phase = Path(sys.argv[4]) if sys.argv[4] else None
worker(temporary, sys.argv[5:], netkingdom, phase)
return
import argparse
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--netkingdom-root', type=Path, default=ROOT.parent / 'net-kingdom')
parser.add_argument('--phase-memory-root', type=Path, default=ROOT.parent / 'phase-memory')
options, arguments = parser.parse_known_args()
netkingdom = options.netkingdom_root.resolve() if options.netkingdom_root.is_dir() else None
phase = options.phase_memory_root.resolve() if (options.phase_memory_root / 'src').is_dir() else None
with tempfile.TemporaryDirectory(prefix='warden-unit-') as directory:
temporary = Path(directory)
home = temporary / 'home'
home.mkdir()
result = subprocess.run([sys.executable, '-I', str(SELF), '--worker', str(temporary),
str(netkingdom) if netkingdom else '', str(phase) if phase else '', *arguments],
cwd=ROOT, env=clean_environment(home, temporary))
raise SystemExit(result.returncode)
if __name__ == '__main__':
main()

20
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/fixtures/workload-coverage/provenance.json vendored Normal file
View file

@ -0,0 +1,20 @@
[
{
"repo": "rapp-core-hub",
"source": "retained public snapshot from warden-1.3-kernel-fixes-r1 test preparation",
"source_sha256": "98e81fb1116aa0ac9dc1a3e7aef9dd87c6d2110d98d46fbfca2f76eec8f3a77b",
"projection_sha256": "9ce902da0c77809f1094a634244652835b162f59682e9fa6923d8c8aa4ed4a3c"
},
{
"repo": "rapp-issue-core",
"source": "retained public snapshot from warden-1.3-kernel-fixes-r1 test preparation",
"source_sha256": "0425c9fe84e26b359b2a35fc37ef8fcb781de1aaa6229c59fcd246ee523514a5",
"projection_sha256": "a1956deecf2f0fbca44549e5ea5e8f1c4b74ebf91a9482c64c982491af0d923b"
},
{
"repo": "rapp-qonto",
"source": "retained public snapshot from warden-1.3-kernel-fixes-r1 test preparation",
"source_sha256": "6279dc2f3ab09519dffa1ceeedb2095f1182960e0106109a87aa5c0052446883",
"projection_sha256": "15aa9809e8d6a120c516be5ce9adfb72c00382140c171e791083ad117f503b3f"
}
]

15
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/fixtures/workload-coverage/rapp-core-hub/declarations/rapp.yaml vendored Normal file
View file

@ -0,0 +1,15 @@
# Public declaration projection for the published coverage snapshot; not live evidence.
rapp_id: rapp-core-hub
workload_identity:
name: core-hub
composition:
member_repos:
- deployables:
- core-hub-api
- core-hub-migrate
- hub-core-candidate
- hub-core-candidate-migrate
- repository-classification-publisher
data_classification: confidential
criticality: high
readiness_state: verified

11
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/fixtures/workload-coverage/rapp-issue-core/declarations/rapp.yaml vendored Normal file
View file

@ -0,0 +1,11 @@
# Public declaration projection for the published coverage snapshot; not live evidence.
rapp_id: rapp-issue-core
workload_identity:
name: issue-core
composition:
member_repos:
- deployables:
- issue-core
data_classification: confidential
criticality: high
readiness_state: verified

11
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/fixtures/workload-coverage/rapp-qonto/declarations/rapp.yaml vendored Normal file
View file

@ -0,0 +1,11 @@
# Public declaration projection for the published coverage snapshot; not live evidence.
rapp_id: rapp-qonto
workload_identity:
name: qonto
composition:
member_repos:
- deployables:
- rapp-qonto
data_classification: restricted
criticality: critical
readiness_state: verified

205
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/test_case_plan.py Normal file
View file

@ -0,0 +1,205 @@
"""WARDEN-WP-0044-T02: plan a KnownSecurityCase run."""
from __future__ import annotations
import copy
import os
from pathlib import Path
import pytest
import yaml
from typer.testing import CliRunner
from warden.case_plan import CaseError, plan_case
from warden.cli import app
from warden.routing.catalog import load_catalog
LIVE_CASE = Path(os.environ.get("WARDEN_TEST_NETKINGDOM_ROOT",
Path(__file__).resolve().parents[2] / "net-kingdom")) / "security-cases/forgejo-netkingdom-oidc.yaml"
TYPE = {
"apiVersion": "netkingdom.io/security-case-type/v0.1",
"kind": "SecurityCaseType",
"spec": {
"id": "t",
"version": "1.0",
"status": "acknowledged",
"parameters": [
{"name": "client_id", "case_settable": True, "sensitivity": "operational"},
{"name": "client_type", "case_settable": True, "sensitivity": "operational"},
{"name": "consumer", "case_settable": True, "sensitivity": "operational"},
{"name": "issuer", "case_settable": False, "default": "https://kc.example"},
],
"follow_through": [
{"id": "proxy", "owner": "o", "entry_point": "ensure {client_id}", "auth": "caller",
"lane": "key-cape-oidc-login", "warden_executes": True, "attended": False,
"value_handling": "none"},
{"id": "mint", "owner": "o", "entry_point": "mint {client_id}", "auth": "caller",
"lane": "key-cape-oidc-login", "warden_executes": True, "attended": False,
"value_handling": "generated_in_custody", "when": "client_type == confidential"},
{"id": "login", "owner": "o", "entry_point": "apply {client_id} env KEYCAPE_{client_id:env}_X",
"auth": "operator",
"lane": "key-cape-oidc-login", "warden_executes": False, "attended": True,
"value_handling": "none", "attended_justification": "live registry is a Secret"},
{"id": "handover", "owner": "{consumer}", "entry_point": "use {issuer}", "auth": "a",
"warden_executes": False, "attended": False, "value_handling": "none",
"ready_check": "{consumer} ready"},
],
"verification": [
{"id": "v", "check": "GET {issuer} for {client_id}", "expect": "present",
"reads_secret_value": False},
{"id": "vc", "check": "custody {client_id}", "expect": "deny",
"reads_secret_value": False, "when": "client_type == confidential"},
],
"rollback": [{"reverses": s} for s in ("proxy", "mint", "login", "handover")],
},
}
CASE = {
"apiVersion": "netkingdom.io/security-case/v0.1",
"kind": "SecurityCase",
"spec": {
"id": "c",
"type": "t@1.0",
"status": "materialized",
"acceptance": {"at": "2026-10-09", "record": "r"},
"parameters": {"client_id": "forgejo", "client_type": "public", "consumer": "railiance-apps"},
},
}
@pytest.fixture(scope="module")
def catalog():
return load_catalog()
def _case(**spec):
doc = copy.deepcopy(CASE)
doc["spec"].update(spec)
return doc
def _type():
return copy.deepcopy(TYPE)
def test_plan_resolves_modes_params_and_defaults(catalog):
plan = plan_case(_case(), _type(), catalog)
modes = {s.id: s.mode for s in plan.steps}
assert modes == {"proxy": "warden", "login": "attended", "handover": "owner"}
assert plan.skipped == ["mint"]
assert plan.attended_sessions == ["key-cape-oidc-login"]
handover = next(s for s in plan.steps if s.id == "handover")
assert handover.owner == "railiance-apps"
assert handover.entry_point == "use https://kc.example"
assert handover.ready_check == "railiance-apps ready"
assert [v["id"] for v in plan.verification] == ["v"]
assert plan.verification[0]["check"] == "GET https://kc.example for forgejo"
def test_condition_true_includes_step_and_check(catalog):
params = dict(CASE["spec"]["parameters"], client_type="confidential")
plan = plan_case(_case(parameters=params), _type(), catalog)
assert "mint" in [s.id for s in plan.steps] and not plan.skipped
assert [v["id"] for v in plan.verification] == ["v", "vc"]
@pytest.mark.parametrize("status,needle", [
("proposed", "not accepted"),
("accepted", "not materialized"),
("live", "already live"),
])
def test_refuse_case_not_materialized(catalog, status, needle):
with pytest.raises(CaseError, match=needle):
plan_case(_case(status=status), _type(), catalog)
def test_refuse_case_without_acceptance(catalog):
with pytest.raises(CaseError, match="acceptance"):
plan_case(_case(acceptance=None), _type(), catalog)
def test_refuse_type_mismatch_and_unacknowledged_type(catalog):
with pytest.raises(CaseError, match="type file is"):
plan_case(_case(type="t@2.0"), _type(), catalog)
doc = _type()
doc["spec"]["status"] = "proposed"
with pytest.raises(CaseError, match="not acknowledged"):
plan_case(_case(), doc, catalog)
def test_refuse_missing_catalog_lane(catalog):
doc = _type()
doc["spec"]["follow_through"][0]["lane"] = "no-such-lane"
with pytest.raises(CaseError, match="LANE-MISSING"):
plan_case(_case(), doc, catalog)
def test_refuse_unreviewed_catalog_lane(catalog):
draft = next((e.id for e in catalog.entries if not (e.is_active and e.reviewed)), None)
if draft is None:
pytest.skip("catalog has no inactive or unreviewed lane")
doc = _type()
doc["spec"]["follow_through"][0]["lane"] = draft
with pytest.raises(CaseError, match="LANE-UNREVIEWED"):
plan_case(_case(), doc, catalog)
def test_refuse_bad_parameters(catalog):
with pytest.raises(CaseError, match="does not declare"):
plan_case(_case(parameters=dict(CASE["spec"]["parameters"], extra="x")), _type(), catalog)
with pytest.raises(CaseError, match="fixes"):
plan_case(_case(parameters=dict(CASE["spec"]["parameters"], issuer="https://evil")), _type(), catalog)
params = {k: v for k, v in CASE["spec"]["parameters"].items() if k != "consumer"}
with pytest.raises(CaseError, match="not set by the case"):
plan_case(_case(parameters=params), _type(), catalog)
def test_refuse_secret_parameter(catalog):
doc = _type()
doc["spec"]["parameters"].append({"name": "pw", "case_settable": True, "sensitivity": "secret"})
with pytest.raises(CaseError, match="secret-sensitivity"):
plan_case(_case(parameters=dict(CASE["spec"]["parameters"], pw="hunter2")), doc, catalog)
def test_env_format_spec_renders_like_keycape_client(catalog):
params = dict(CASE["spec"]["parameters"], client_id="my-app")
plan = plan_case(_case(parameters=params), _type(), catalog)
login = next(s for s in plan.steps if s.id == "login")
assert login.entry_point == "apply my-app env KEYCAPE_MY_APP_X"
@pytest.mark.parametrize("template", ["{client_id:upper}", "{client_id:>9}", "{client_id!r}"])
def test_refuse_unknown_format_spec_or_conversion(catalog, template):
doc = _type()
doc["spec"]["follow_through"][0]["entry_point"] = template
with pytest.raises(CaseError, match="format spec|conversions"):
plan_case(_case(), doc, catalog)
def test_refuse_unsupported_condition(catalog):
doc = _type()
doc["spec"]["follow_through"][0]["when"] = "client_type in (a, b)"
with pytest.raises(CaseError, match="unsupported condition"):
plan_case(_case(), doc, catalog)
def test_cli_plan_shows_attended_and_exit_codes(tmp_path):
case_file, type_file = tmp_path / "case.yaml", tmp_path / "type.yaml"
case_file.write_text(yaml.safe_dump(CASE))
type_file.write_text(yaml.safe_dump(TYPE))
runner = CliRunner()
ok = runner.invoke(app, ["case", "plan", str(case_file), "--type", str(type_file)])
assert ok.exit_code == 0, ok.output
assert "ATTENDED" in ok.output and "nothing executed" in ok.output
import hashlib
digest = hashlib.sha256(case_file.read_bytes()).hexdigest()
assert f"case-ref: c@sha256:{digest}" in ok.output
case_file.write_text(yaml.safe_dump(_case(status="accepted")))
refused = runner.invoke(app, ["case", "plan", str(case_file), "--type", str(type_file)])
assert refused.exit_code == 1
@pytest.mark.skipif(not LIVE_CASE.exists(), reason="net-kingdom checkout absent")
def test_live_forgejo_case_is_refused_until_materialized():
result = CliRunner().invoke(app, ["case", "plan", str(LIVE_CASE)])
assert result.exit_code == 1

130
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/test_case_type.py Normal file
View file

@ -0,0 +1,130 @@
"""WARDEN-WP-0044-T01: SecurityCaseType follow-through check."""
from __future__ import annotations
import copy
import os
from pathlib import Path
import pytest
import yaml
from typer.testing import CliRunner
from warden.case_type import check_follow_through, load_type
from warden.cli import app
from warden.routing.catalog import load_catalog
LIVE_TYPE = Path(os.environ.get("WARDEN_TEST_NETKINGDOM_ROOT",
Path(__file__).resolve().parents[2] / "net-kingdom")) / "security-case-types/oidc-browser-relying-party/type.yaml"
BASE = {
"apiVersion": "netkingdom.io/security-case-type/v0.1",
"kind": "SecurityCaseType",
"spec": {
"id": "t",
"version": "1.0",
"follow_through": [
{"id": "run", "owner": "o", "entry_point": "e", "auth": "a", "lane": "key-cape-oidc-login",
"warden_executes": True, "attended": False, "value_handling": "none"},
{"id": "login", "owner": "o", "entry_point": "e", "auth": "a", "lane": "key-cape-oidc-login",
"warden_executes": False, "attended": True, "value_handling": "none",
"attended_justification": "live registry is a Secret"},
{"id": "handover", "owner": "o", "entry_point": "e", "auth": "a",
"warden_executes": False, "attended": False, "value_handling": "none"},
],
"verification": [{"id": "v", "check": "c", "expect": "present", "reads_secret_value": False}],
"rollback": [{"reverses": "run"}, {"reverses": "login"}, {"reverses": "handover"}],
},
}
@pytest.fixture(scope="module")
def catalog():
return load_catalog()
def _doc(**step_overrides):
doc = copy.deepcopy(BASE)
doc["spec"]["follow_through"][0].update(step_overrides)
return doc
def _codes(report):
return {f.code for f in report.findings}
def test_conforming_type_is_plannable(catalog):
report = check_follow_through(_doc(), catalog)
assert report.plannable, report.findings
assert report.warden_steps == ["run"]
assert report.attended_lanes == ["key-cape-oidc-login"]
def test_executed_step_without_lane_refused(catalog):
doc = _doc()
del doc["spec"]["follow_through"][0]["lane"]
assert "FT-LANE" in _codes(check_follow_through(doc, catalog))
def test_unknown_lane_is_catalog_finding_not_schema(catalog):
report = check_follow_through(_doc(lane="no-such-lane"), catalog)
assert report.schema_ok and not report.plannable
assert _codes(report) == {"LANE-MISSING"}
def test_value_handling_required_and_enumerated(catalog):
assert "FT-VALUE" in _codes(check_follow_through(_doc(value_handling="returned"), catalog))
doc = _doc()
del doc["spec"]["follow_through"][0]["value_handling"]
assert "FT-FIELD" in _codes(check_follow_through(doc, catalog))
def test_attended_needs_justification_and_excludes_execution(catalog):
doc = copy.deepcopy(BASE)
del doc["spec"]["follow_through"][1]["attended_justification"]
doc["spec"]["follow_through"][1]["warden_executes"] = True
msgs = [f.message for f in check_follow_through(doc, catalog).findings if f.code == "FT-ATTEND"]
assert len(msgs) == 2
def test_attended_step_needs_login_lane(catalog):
doc = copy.deepcopy(BASE)
secret_lane = next(e.id for e in catalog.entries if e.is_active and e.reviewed and e.lane == "secret")
doc["spec"]["follow_through"][1]["lane"] = secret_lane
assert "LANE-KIND" in _codes(check_follow_through(doc, catalog))
def test_generated_value_never_in_attended_session(catalog):
doc = copy.deepcopy(BASE)
doc["spec"]["follow_through"][1]["value_handling"] = "generated_in_custody"
assert "FT-VALUE" in _codes(check_follow_through(doc, catalog))
def test_verification_must_not_read_values(catalog):
doc = _doc()
doc["spec"]["verification"][0]["reads_secret_value"] = True
assert "VER-VALUE" in _codes(check_follow_through(doc, catalog))
def test_every_step_reversible(catalog):
doc = _doc()
doc["spec"]["rollback"] = doc["spec"]["rollback"][1:]
assert "RB-MISSING" in _codes(check_follow_through(doc, catalog))
def test_cli_exit_codes(tmp_path):
good, bad = tmp_path / "good.yaml", tmp_path / "bad.yaml"
good.write_text(yaml.safe_dump(_doc()))
bad.write_text(yaml.safe_dump(_doc(lane="no-such-lane")))
runner = CliRunner()
assert runner.invoke(app, ["case", "check-type", str(good), "--json"]).exit_code == 0
assert runner.invoke(app, ["case", "check-type", str(bad), "--json"]).exit_code == 1
@pytest.mark.skipif(not LIVE_TYPE.exists(), reason="net-kingdom checkout not present")
def test_live_oidc_rp_type_schema_conforms(catalog):
"""The acknowledged 1.0 type is plannable: every lane it names is active (2026-10-10)."""
report = check_follow_through(load_type(LIVE_TYPE), catalog)
assert report.schema_ok, report.findings
assert report.plannable, report.findings
assert report.attended_lanes == ["key-cape-oidc-login"]
assert report.warden_steps == ["mint-secret"]

355
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/test_layer_conformance.py Normal file
View file

@ -0,0 +1,355 @@
"""Layer-model conformance (security-layer-model_v0.8 §5, §11).
Two things are checked here. §11 makes one of them mechanical: every direct
Tooling client maps to a declared shape. §5.2 asks for the other: the conduit's
supplied-authority property covered by a test.
Deliberately absent: any assertion on a §5.3 review date. A date-triggered
failure breaks the build on a calendar day with no code change, punishing
whoever commits next rather than whoever owns the gap — the same reasoning
recorded in WARDEN-WP-0033-T05 for blocker staleness.
"""
from __future__ import annotations
import os
import subprocess
import sys
from pathlib import Path
import yaml
ROOT = Path(__file__).resolve().parents[1]
def _decl() -> dict:
return yaml.safe_load((ROOT / "layer.yaml").read_text())
def _intent_frontmatter() -> dict:
lines = (ROOT / "INTENT.md").read_text().splitlines()
assert lines[0].strip() == "---", "INTENT.md must carry frontmatter — it is the declaration"
end = next(i for i, ln in enumerate(lines[1:], 1) if ln.strip() == "---")
return yaml.safe_load("\n".join(lines[1:end]))
def _checker():
import importlib.util
spec = importlib.util.spec_from_file_location(
"check_layer_conformance", ROOT / "scripts" / "check_layer_conformance.py"
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def _fold(value: str) -> str:
return str(value).strip().encode("ascii", "ignore").decode().lower()
class TestDeclaration:
def test_declares_staff_layer_in_its_own_voice(self):
d = _decl()
assert d["repository"] == "ops-warden"
assert _fold(d["layer"]) == "staff"
# §11: "only the repository's own file, in its own voice, conforms."
assert d["declared_by"] == "docs/adr/ADR-0010"
def test_intent_md_carries_the_governing_declaration(self):
"""GH-DEC-2026-017 §1 / A11: INTENT.md's frontmatter is the declaration."""
assert _fold(_intent_frontmatter()["layer"]) == "staff"
def test_sidecar_is_marked_derived_and_names_its_source(self):
"""§11's derived-artifact rule, applied to the layer sidecar."""
d = _decl()
assert d["derived"] is True
assert d["derived_from"] == "INTENT.md"
def test_the_two_forms_agree_once_case_is_folded(self):
"""A11: the derived form must agree; A9: comparison folds case.
`Staff` in INTENT.md and `staff` in layer.yaml are the SAME value. This
test is deliberately a fold rather than an equality: the ruling asked
nobody to re-spell anything, and an equality assertion here would be this
repository quietly doing the re-spelling the ruling declined to order.
The next *real* divergence — a different layer — still fails.
"""
assert _fold(_decl()["layer"]) == _fold(_intent_frontmatter()["layer"])
def test_layer_is_in_section_3_closed_vocabulary(self):
"""A9: {Taxonomy, Tooling, Engine, Staff}, closed, case-insensitive."""
vocabulary = {"taxonomy", "tooling", "engine", "staff"}
assert _fold(_intent_frontmatter()["layer"]) in vocabulary
assert _fold(_decl()["layer"]) in vocabulary
def test_no_declaration_carries_a_standard_version(self):
"""GH-DEC-2026-017 §5 / A12 — and the regression guard on its return.
The field was removed from the estate's reference form, not just from
this file. A field that is present will be branched on, so absence is
asserted rather than trusted.
"""
assert "standard_version" not in _decl()
assert "standard_version" not in _intent_frontmatter()
def test_no_version_anywhere_in_either_declaration(self):
"""A12 r2 / GH-DEC-2026-020 §1-§2: content, not a key name.
A versioned `standard:` path or a `companion_version` is the same pin as
`standard_version`, so the guard walks every key and value of both forms.
"""
checker = _checker()
assert checker.find_version_pins(_intent_frontmatter()) == []
assert checker.find_version_pins(_decl()) == []
assert not str(_intent_frontmatter()["standard"]).endswith(".md")
def test_checker_catches_a_versioned_standard_path(self):
checker = _checker()
pins = checker.find_version_pins(
{"layer": "Staff", "standard": "net-kingdom/canon/standards/security-layer-model_v0.7.md"}
)
assert pins and pins[0].startswith("standard")
def test_checker_catches_a_companion_version(self):
checker = _checker()
assert checker.find_version_pins({"layer": "Staff", "companion_version": "0.2"})
assert checker.find_version_pins({"nested": {"standard_version": "0.7"}})
def test_version_token_in_identity_value_is_a_pin(self):
"""GH-DEC-2026-021 §3: any `v?N.N` in a standard:/companion: value is a pin."""
checker = _checker()
pins = checker.find_version_pins({"standard": "security-layer-model v0.7"})
assert pins and pins[0].startswith("standard")
assert checker.find_version_pins({"companion": "SECURITY-COMPANION 0.2"})
assert checker.find_version_pins({"nested": {"standard": ["security-layer-model v0.8"]}})
def test_prose_citation_and_intent_version_are_not_reached(self):
"""GH-DEC-2026-021 §1 (A12 r3): prose provenance and `intent_version` pass."""
checker = _checker()
assert checker.find_version_pins(
{"layer": "Staff", "note": "Outside §5 by the v0.5 scope rule", "intent_version": "0.1.0"}
) == []
def test_schema_version_is_not_reached(self):
assert _checker().find_version_pins({"schema_version": "0.2", "layer": "Staff"}) == []
def test_stance_map_is_outside_the_run(self):
"""GH-DEC-2026-020 §3: a stance map keeps its version; the run must not read it."""
stance = yaml.safe_load((ROOT / "pep-stance.yaml").read_text())
assert "standard_version" in stance, "pep-stance.yaml keeps its clause-scoped version"
assert "pep-stance" not in _checker().SCOPE
def test_every_run_states_version_and_scope(self):
"""GH-DEC-2026-020 §4: the version belongs to the run, printed every time."""
checker = _checker()
out = subprocess.run(
[sys.executable, str(ROOT / "scripts" / "check_layer_conformance.py")],
capture_output=True,
text=True,
).stdout
assert f"validated against: {checker.VALIDATED_AGAINST}" in out
assert f"scope: {checker.SCOPE}" in out
pass_line = next(ln for ln in out.splitlines() if ln.startswith("PASS"))
assert checker.VALIDATED_AGAINST in pass_line
def test_every_tooling_contact_maps_to_a_declared_shape(self):
"""§11 mechanical check — the guard against a new undeclared client."""
result = subprocess.run(
[sys.executable, str(ROOT / "scripts" / "check_layer_conformance.py")],
capture_output=True,
text=True,
)
assert result.returncode == 0, (
f"undeclared Tooling contact — a finding under §11, not a tracked gap:\n"
f"{result.stdout}{result.stderr}"
)
def test_declared_gaps_carry_all_four_fields(self):
"""§5.3 is machine-readable or it is prose wearing a schema."""
for c in _decl()["tooling_contacts"]:
if c["shape"] == "5.3":
for field in ("capability", "intended_owner", "blocked_on", "review"):
assert c.get(field), f"{c['id']} missing {field}"
def test_gaps_are_not_counted_as_conformance(self):
"""§11: a declared gap is tracked non-conformance. Keep that visible."""
text = (ROOT / "layer.yaml").read_text()
assert "TRACKED NON-CONFORMANCE" in text.upper()
class TestConduitSuppliesNoAuthority:
"""§5.2: 'MUST NOT present its own credential, MUST NOT widen what the
caller could already do.' The standard says this SHOULD be covered by a
test; this is that test."""
def test_conduit_supplies_no_authority_of_its_own(self, monkeypatch):
from warden import proxy
monkeypatch.setenv("VAULT_TOKEN", "caller-own-token")
monkeypatch.setenv("HOME", "/home/nobody")
before = dict(os.environ)
env = proxy._caller_env()
# The child environment IS the caller's environment — nothing added,
# nothing removed, no ops-warden credential injected.
assert env == before, (
"conduit altered the caller's environment; §5.2 requires it to "
"supply no authority of its own"
)
assert env["VAULT_TOKEN"] == "caller-own-token"
def test_conduit_declares_supplied_authority_none(self):
conduits = [c for c in _decl()["tooling_contacts"] if c["shape"] == "5.2"]
assert conduits, "no §5.2 conduit declared — proxy.py is one"
for c in conduits:
assert c["supplied_authority"] == "none"
def test_proxy_holds_no_credential_constant(self):
"""A conduit that presents its own token is not a conduit (§5.2)."""
src = (ROOT / "src" / "warden" / "proxy.py").read_text()
# It may name token ENV VARS to detect caller auth; it must not carry a
# token value or mint one.
for forbidden in ("X-Vault-Token", "auth/approle/login", "token create"):
assert forbidden not in src, (
f"proxy.py references {forbidden!r} — that is presenting or "
f"minting authority, not conducting the caller's"
)
class TestPepStanceMap:
"""§6.4: every PEP-shaped consumer MUST publish its unreachable-engine
stance map, total and per zone, 'published rather than held in code'.
ADR-0009 is named as the reference shape, so it should actually hold."""
def _stance(self) -> dict:
return yaml.safe_load((ROOT / "pep-stance.yaml").read_text())
def test_published_map_equals_shipped_behaviour(self):
"""The whole point. A published map that may drift from the code is
worse than none, because it invites reliance it cannot support."""
from warden.config import PolicyConfig
assert self._stance()["stance"] == PolicyConfig().failure_modes
def test_stance_is_total_over_the_zone_model(self):
"""§6.4 obligation 3: total, no implicit default."""
stance = self._stance()["stance"]
required = {
"z0-experimental", "z1-operational", "z2-protected",
"z2-continuity", "z3-critical", "unknown", "not-applicable",
}
assert required <= set(stance), f"stance not total; missing {required - set(stance)}"
assert set(stance.values()) <= {"fail_open", "fail_closed"}
def test_critical_zone_fails_closed(self):
"""ADR-0009's one non-negotiable row."""
assert self._stance()["stance"]["z3-critical"] == "fail_closed"
def test_verdict_is_never_cached(self):
"""§6.4 obligation 2: caching an input claim is permitted; caching the
answer is a second decision point deciding early (§6.1)."""
assert self._stance()["verdict_caching"] == "none"
def test_revocation_visibility_deadline_equals_enforced_ttl_policy(self):
"""§9.7.2: a published replay window must not drift from issuance."""
from warden.models import ActorType, MAX_TTL_HOURS
published = self._stance()["revocation_visibility"]
expected = {actor.value: MAX_TTL_HOURS[actor] for actor in ActorType}
assert published["deadline_hours"] == expected
assert published["mechanism"] == "ttl_expiry"
assert published["revocation_channel"] == "none"
def test_attributive_emission_cadence_deferral_carries_measurement(self):
cadence = self._stance()["emission_cadence"]
assert cadence["classification"] == "attributive"
assert cadence["status"] == "deferred"
assert cadence["observed_window"]["signature_records"] == 3
assert cadence["observed_window"]["active_signature_days"] == 2
assert cadence["reason"]
# --- classification coverage (v0.8 §6.4 obligation 3) -------------------------
def test_published_coverage_equals_measured_coverage():
"""The published figure must equal its explicit public declaration snapshot.
This is a deterministic source check, not a claim about current estate state.
Live measurement remains scripts/report_coverage.py with an explicit estate
root; tests never treat the developer HOME as an authority or fixture.
ops-warden asked gate-house for §13.1's Coverage column and its figures are
that column's first entries, so their accuracy is ours to hold. The register
explicitly does not compute anyone's coverage, and a stale number beside a
marked cell is worse than a blank -- a blank at least reads as "not reported".
This is the same property that makes the stance map worth publishing (the map
equals PolicyConfig.failure_modes by test), applied one level up.
"""
import importlib.util
repo = Path(__file__).resolve().parents[1]
spec = importlib.util.spec_from_file_location(
"report_coverage", repo / "scripts" / "report_coverage.py"
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
published = yaml.safe_load((repo / "pep-stance.yaml").read_text())[
"classification_coverage"
]
measured = module.measure(repo / "tests/fixtures/workload-coverage")
for population in ("signing_targets", "routing_lanes"):
assert published[population] == measured[population], population
def test_the_unknown_cell_is_marked_as_a_declared_gap():
"""A non-conformant cell must say so where it is declared, not only in a review.
§11's marking obligation, which ops-warden argued for in the v0.6 round and
then acquired a marked cell under. If the cell is ever flipped to fail_closed
this test fails, which is the correct time to remove the marking.
"""
repo = Path(__file__).resolve().parents[1]
text = (repo / "pep-stance.yaml").read_text()
stance = yaml.safe_load(text)["stance"]
if stance["unknown"] == "fail_open":
assert "DECLARED GAP" in text
assert "WARDEN-WP-0040" in text
else:
assert stance["unknown"] == "fail_closed"
def test_accepted_v08_run_provenance_and_catalog_scope():
checker = _checker()
assert "security-layer-model_v0.8.md" in checker.VALIDATED_AGAINST
assert "sha256:3d8298560069ef471fba28d2b1245fb2309328f231dad1a667cdf0ce5a5a7c8b" in checker.VALIDATED_AGAINST
assert "single repository ops-warden (catalogued in §4)" in checker.SCOPE
def test_reference_semantics_propagate_identity_through_nested_sequences():
checker = _checker()
for data in (
{"nested": {"standard": ["Security Layer Model v0.8"]}},
{"companion": [["Security Companion 0.3"]]},
{"standard": {"alternate": ["Security Layer Model v0.7"]}},
):
# A dictionary begins a fresh identity key; only sequence descendants
# inherit the immediate standard:/companion: identity in the reference.
if isinstance(data.get("standard"), dict):
assert checker.find_version_pins(data) == []
else:
assert checker.find_version_pins(data), data
assert checker.find_version_pins({"rationale": "Assented at v0.7; migrated at v0.8."}) == []
assert checker.find_version_pins({"intent_version": "0.2.0", "schema_version": "0.1"}) == []
def test_ascii_folding_never_discards_non_ascii_letters():
checker = _checker()
assert checker._fold("Staff") == "staff"
assert checker._fold("Engine") == "engine"
# Deleting non-ASCII bytes would turn this fifth token into Staff.
assert checker._fold("Stαaff") not in checker.LAYER_VOCABULARY

81
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/test_safe_unit_runner.py Normal file
View file

@ -0,0 +1,81 @@
"""Exercise installed parent and nested child controls, never real credentials."""
import json
import os
from pathlib import Path
import socket
import subprocess
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
@pytest.fixture
def guarded(request):
guard = getattr(request.config, '_unit_guard', None)
if guard is None:
pytest.skip('requires make test / scripts/run_unit_tests.py')
return guard
def test_parent_environment_is_allowlisted_and_home_is_temporary(guarded):
# The external regression invocation sets synthetic BAO_TOKEN/VAULT_TOKEN/HOME.
assert 'BAO_TOKEN' not in os.environ
assert 'VAULT_TOKEN' not in os.environ
assert Path(os.environ['TMPDIR']) in Path.home().parents
assert 'PYTHONPATH' not in os.environ
@pytest.mark.parametrize('command', [
['bao', 'token', 'lookup'], ['ssh', 'railiance01'], ['curl', 'http://127.0.0.1:8000'],
[sys.executable, '-c', 'import os; print(os.environ)'],
])
def test_unexpected_process_refused_before_launch(guarded, command):
with pytest.raises(PermissionError, match='unit isolation'):
subprocess.run(command, check=True)
def test_parent_socket_denies_even_unregistered_loopback(guarded):
with pytest.raises(PermissionError, match='unit isolation'):
socket.socket().connect(('127.0.0.1', 8000))
@pytest.mark.parametrize('path', ['/home/synthetic-ambient-unit-fixture/.vault-token',
'/root/.kube/config', '/proc/self/environ',
'/tmp/warden-synthetic-ambient-home/.vault-token'])
def test_ambient_paths_refused_without_opening_them(guarded, path):
with pytest.raises(PermissionError, match='unit isolation'):
Path(path).read_bytes()
def test_symlink_cannot_escape_fixture_root(guarded, tmp_path):
link = tmp_path / 'ambient'
link.symlink_to('/home/synthetic-ambient-unit-fixture/.vault-token')
with pytest.raises(PermissionError, match='unit isolation'):
link.read_bytes()
def test_child_drops_even_synthetic_token_environment(guarded, monkeypatch):
monkeypatch.setenv('BAO_TOKEN', 'synthetic-only-never-a-real-token')
monkeypatch.setenv('VAULT_TOKEN', 'synthetic-only-never-a-real-token')
monkeypatch.setenv('HOME', '/home/synthetic-ambient-unit-fixture')
result = subprocess.run([sys.executable, str(ROOT / 'tests/unit_isolation_probe.py'), 'state'],
capture_output=True, text=True, check=True, env=dict(os.environ))
assert json.loads(result.stdout) == {'bao_token_present': False,
'vault_token_present': False, 'home_in_temporary': True}
@pytest.mark.parametrize('mode', ['socket', 'provider', 'home', 'nested'])
def test_guard_exists_in_each_allowed_python_child(guarded, mode):
result = subprocess.run([sys.executable, str(ROOT / 'tests/unit_isolation_probe.py'), mode],
capture_output=True, text=True, check=True)
assert result.stdout == 'denied\n'
def test_forged_provider_double_refused(guarded, tmp_path):
provider = tmp_path / 'bao'
provider.write_text('#!/bin/bash\necho not-the-reviewed-fixture\n')
provider.chmod(0o700)
with pytest.raises(PermissionError, match='double bytes differ'):
subprocess.run([str(provider), 'token', 'lookup'], check=True)

70
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/test_workload_join.py Normal file
View file

@ -0,0 +1,70 @@
"""Explicit lane-to-workload join tests (WARDEN-WP-0032 / RMGR-WP-0010-T06)."""
from pathlib import Path
import yaml
from scripts.report_workload_join import build
ROOT = Path(__file__).resolve().parents[1]
def test_repo_catalog_uses_only_explicit_workload_references():
report = build(ROOT / "registry/routing/catalog.yaml",
ROOT / "tests/fixtures/workload-coverage")
assert report["ok"] is True
assert len(report["resolved"]) == 4
assert len(report["unknown"]) == 22
# 15 since 2026-09-21: the layer-declaration route is repository conformance
# guidance, not a workload delivery lane, and the two NetKingdom SSO lanes
# (c374d41) remain provider/control-plane credentials. Lanes added after
# 2026-09-09 moved the denominator without moving `resolved`; these counts
# and pep-stance.yaml's classification_coverage measure the same populations
# and must be updated together. 16 since 2026-10-05: the
# openbao-structure-survey pointer (WARDEN-WP-0042) is a builder survey
# under a founder-minted grant, not a workload credential. 17 since
# 2026-10-08: platform-supplier-credentials (WARDEN-WP-0043) is a generic
# supplier-account path pattern; concrete consumers keep their own lanes.
# unknown 22 since 2026-10-09: fluid-telegram-operator (MASON-WP-0005-T06)
# names fluid-telegram, which has published no workload declaration.
# not_applicable 18 since 2026-10-10: keycape-client-secret-generate is a
# per-case generation lane; each case's consumer gets its own entry.
# not_applicable 22 since 2026-10-17: the four CATALOG-1 draft lanes
# (nk48 1.3 owner review) are per-case lanes like the one above.
assert len(report["not_applicable"]) == 22
assert {row["lane"] for row in report["resolved"]} == {
"ops-warden-warden-sign-token",
"issue-core-ingestion-api-key",
"rapp-qonto-keycape-client",
"core-hub-forgejo-source-read", # CCR-2026-0031, 2026-10-03
}
def test_invalid_exact_deployable_resolves_unknown(tmp_path):
rapp = tmp_path / "rapp-x" / "declarations"
rapp.mkdir(parents=True)
(rapp / "rapp.yaml").write_text(yaml.safe_dump({
"rapp_id": "rapp-x",
"workload_identity": {"name": "x"},
"composition": {"member_repos": [{"deployables": ["api"]}]},
}))
catalog_dir = tmp_path / "ops-warden" / "registry" / "routing"
catalog_dir.mkdir(parents=True)
catalog = catalog_dir / "catalog.yaml"
catalog.write_text(yaml.safe_dump({"entries": [{
"id": "x",
"workload_ref": {
"applicability": "applicable",
"rapp_id": "rapp-x",
"name": "x",
"deployable": "missing",
},
}]}))
posture = catalog_dir.parent / "policy"
posture.mkdir()
(posture / "security-posture.yaml").write_text("dataclass_floor: {}\n")
report = build(catalog, tmp_path)
assert not report["resolved"]
assert report["unknown"][0]["lane"] == "x"
assert "deployable" in report["unknown"][0]["reason"]

29
docs/changes/netkingdom-target-2026-10-17/warden-safe-unit-runner-r1/candidate/tests/unit_isolation_probe.py Normal file
View file

@ -0,0 +1,29 @@
"""Closed child regression probes; executed only under the unit runner guard."""
import json
import os
from pathlib import Path
import socket
import subprocess
import sys
mode = sys.argv[1]
if mode == 'state':
print(json.dumps({'bao_token_present': 'BAO_TOKEN' in os.environ,
'vault_token_present': 'VAULT_TOKEN' in os.environ,
'home_in_temporary': Path(os.environ['TMPDIR']) in Path.home().parents}))
else:
try:
if mode == 'socket':
socket.socket().connect(('127.0.0.1', 8000))
elif mode == 'provider':
subprocess.run(['bao', 'token', 'lookup'], check=True)
elif mode == 'home':
Path('/home/synthetic-ambient-unit-fixture/.vault-token').read_text()
elif mode == 'nested':
subprocess.run([sys.executable, '-c', 'print("SYNTHETIC-SECRET")'], check=True)
else:
raise AssertionError('unknown probe')
except PermissionError:
print('denied')
else:
raise AssertionError('external operation was not refused')