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

Record native security preparation and owner checkpoints (1/7)
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details
Reject generated-only commits / generated-commit-guard (push) Successful in 9s
Details

Browse source
This commit is contained in:
codex 2026-10-17 04:30:03 +02:00
parent ceaeef1ee7
commit 275e034883
18 changed files with 7262 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

3643
docs/changes/netkingdom-target-2026-10-17/native-gates-r1/native-gate-audit.json Normal file
View file

File diff suppressed because it is too large Load diff

62
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/README.md Normal file
View file

@ -0,0 +1,62 @@
# Original identity recovery preparation — 17 October 2026
This packet advances NK-WP-0049-T33/T35/T36, RCLUSTER-WP-0018-T02 and
SECRETS-WP-0015-T03 through source preparation. No owner or founder assent is
created. No native recovery, private archive read, credential retrieval, token
exchange, OpenBao login, rotation or production action occurred.
The existing consumer's eight-field public selector passes its offline AST-only
checker against the reviewed `service_auth.py` hash. See selector-preflight.json.
The TLS source still names kc.coulomb.social, sso/kc-tls,
ClusterIssuer/letsencrypt-prod and Traefik. The identity archive excludes kc-tls;
same-host encrypted cluster backups do not establish independent original
certificate/key/trust recovery. Fetching today's public endpoint certificate
would not establish the archive-time original or independent private custody.
## Exact next deliverables
1. RCLUSTER-WP-0018-T02: the actual custody/provider owner supplies the fields in
owner-declaration.template.json: independently recoverable original artifact,
public certificate/chain/trust provenance and hashes, private custody entry
point reference and outage prerequisites. Alternatively, the existing owners
and founder explicitly accept a concrete reissue disposition naming the
changed continuity claim, dependencies and outage limitations. Neither path
is decided by this packet. Original TLS and accepted reissue are distinct.
2. SECRETS-WP-0015-T03: review consumer-proof-plan.md and disposition its outage
dependency with the actual service/custody owners. Public source correspondence
is established; exact archive presence/uniqueness and current recovery assent
remain absent. Existing consumer custody depends on attended OpenBao OIDC and
forbids a client-secret file. Do not silently replace that source with an
archive value or injected callback.
3. NetKingdom reviews the narrow adapter change in selector-uniqueness.patch.
It requires exactly one matching registration in the already-restored payload
before further runtime/private inputs. Current KeyCape Go configuration also
rejects duplicate IDs; this change makes the adapter's own check explicit,
and is not evidence of a native duplicate-registration exploit. Original
behavior and focused regression receipts are retained separately.
4. T33 assembles the complete admission and exact six-argument command only when
those prerequisites are reviewed. See next-attempt-proposal.json. Missing
TLS/custody/assent fields remain null and the proposal is non-executable.
Changes to the adapter require a fresh dependency/admission binding. Frozen
historical fixture/attempt receipts are not rewritten as results of this change.
Only the normal owner publication and projection process may publish canonical
source; root coordinates that process. No task is marked done by this packet.
## Foreign waits and authority
Read-only Hub observations are retained in hub-readback.json. The latest cases
observed are `nk49-t35-rcluster18` receipt
`a2e5ed16-0da0-4ab9-bc0a-d82bd0a393f1`, `nk49-t36-secrets15` receipt
`8a73a503-7479-426b-b2ad-e88f32e4c7e8`, and `secrets15-nk49-readiness`
receipt `3f905339-a4c9-428e-9cbd-fa490c144c7a`. All remain waiting. Cluster
publication is already evidenced; the older unpublished-Cluster reason is
historical and superseded. No worker wake, message or Hub write was performed.
Custodian supervision is supervised and grants no authority. The Emergent
Security start retrieval returned package 0.2.6, digest
`8d1d6a9d4d07fec7b619618de7dfe2be2254c1790f85f4198c7fe19db9751c02`,
withheld because upstream sources changed. Current owner instructions,
environment orientation, Security Companion and identity/start card source were
read directly; package guidance is orientation only. Routing discovery is
recorded in routing-discovery.json and supplies no credential authority.

69
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/consumer-proof-plan.md Normal file
View file

@ -0,0 +1,69 @@
# Proposed bounded consumer proof and custody plan
Draft source plan for SECRETS-WP-0015-T03. Actual source-owner review and any
custody/scope disposition remain required; this is not another owner's review.
The selector is secrets-engine-openbao as both client ID and audience,
KEYCAPE_SECRETS_ENGINE_OPENBAO_CLIENT_SECRET as producer environment name,
service:secrets-engine, tenant:platform, role secrets-engine, granted
openbao:login and denied openbao:admin. No new registration, secret or grant is
proposed. Historical 7 October issuer and 10 October consumer proof predate the
16 October ciphertext but cannot prove its private contents or current assent.
## Claims and their actual dependencies
| Claim | Evidence required | Present limit |
| --- | --- | --- |
| Current intended consumer selector | Offline AST constants and strict eight-field selector correspondence | Passed; no credential source or network exercised |
| Original archived registration | Exact admitted ciphertext, private authenticated restoration, exactly one matching registration, exact producer env binding | Not observed; must remain inside owner custody |
| Original isolated issuer behavior | Pinned image/executable, exact issuer and original TLS/trust bindings, native positive signature/claim checks and unrelated-secret/excess-scope refusals | TLS/private admission incomplete; no native action performed |
| Selected secrets-engine consumer continuity | Actual consumer uses its accepted source and validation flow against admitted target; signature verification belongs to receiving OpenBao | Consumer's ordinary source and resource server require OpenBao; not established by issuer verify-client |
| Backend protected operation | Existing SECRETS-WP-0008/0013 entitlement and execution contracts | Separate gate; login does not grant protected work |
| Retained-factor MFA | NK-WP-0049-T19's separate factor/browser proof | Service client cannot prove it |
`engine_auth.login_service_jwt` constructs OpenBaoOidcClientSecretSource,
performs an attended operator OIDC login, reads the exact platform KV field
in memory, exchanges the service secret at KeyCape, then logs in to OpenBao's
keycape-services role. This cyclic dependency does not establish recovery when
ordinary KeyCape and OpenBao are unavailable. `preflight_service_jwt` checks
claims without cryptographic verification; invoking it alone cannot establish
consumer acceptance. Constructor injection of a secret source is not permission
to bypass the production factory's source contract.
The owner must choose a concrete supported continuity scope: admit the bounded
isolated issuer proof while retaining actual consumer continuity as an open
gate, or supply an existing independently recoverable private provider/custody
path compatible with the accepted consumer contract. A new fallback source,
file copy, production redirect or recovered-OpenBao deployment would be a
separately reviewed change. This packet chooses none and narrows no gate.
## Required reviewed attempt binding
Bind original ciphertext SHA-256 and verified transfer; canonical source and
dependency hashes; selected original client and exact-one registration check;
pinned KeyCape image and executable; original issuer, public certificate/chain/
trust hashes and validity/SANs; owner-held TLS/ciphertext locations; versioned
ciphertext-derived private manifest mode; fixed local Docker target and a fresh
safe receipt path. Private manifest digest and credential values stay inside
custody. Do not reuse a digest from the pre-change adapter source closure.
Before any later key prompt, the existing public preflight must refuse missing,
changed or ambiguous public admission, source, target and selector metadata.
After restoration, the adapter's exact-one check must refuse absent, duplicate
or misbound clients before further environment, signer or native-check access.
It cannot inspect private archived registrations before decryption. The native
attempt must independently refuse wrong unrelated secret, openbao:admin,
wrong issuer/trust/executable/source and failed cleanup; transport failure alone
is never credential-denial evidence. Keep generated fixtures separate from
actual archive acceptance and from real predecessor-rotation proof.
Use the existing attempt's 1800-second total deadline and native verifier's
150-second bound only after owners review the complete revision. No retry may
convert failure or uncertainty into success. On binding, startup, native denial,
timeout or cleanup uncertainty, abort, retain only the sanitized receipt and
explicit obligations, and use exact session-scoped cleanup. The isolated attempt
changes no production service, so production rollback is not an implicit action.
Record removal of exact session containers/tmpfs, descriptor/child lifetime,
and all cleanup failures; independent receipt verification precedes acceptance.
Owner/cost/risk/deadline assumptions remain unapproved until the existing review
path resolves them. Founder-deferred rotation stays deferred.

19
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/focused-regressions.stderr.txt Normal file
View file

@ -0,0 +1,19 @@
.E...........................................................................
======================================================================
ERROR: test_manifest_binds_current_candidate_and_existing_fixture_factory (test_isolated_keycape_fixture_plan.FixturePreparationTests.test_manifest_binds_current_candidate_and_existing_fixture_factory)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_fixture_plan.py", line 44, in test_manifest_binds_current_candidate_and_existing_fixture_factory
manifest=f.reviewed_manifest(digest)
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/worsch/net-kingdom/tests/run_isolated_keycape_native_fixture.py", line 48, in reviewed_manifest
require('sha256:'+hashlib.sha256((ROOT/name).read_bytes()).hexdigest()==digest)
File "/home/worsch/net-kingdom/tests/run_isolated_keycape_native_fixture.py", line 40, in require
if not condition:raise RuntimeError('synthetic_fixture_refused')
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
RuntimeError: synthetic_fixture_refused
----------------------------------------------------------------------
Ran 77 tests in 3.045s
FAILED (errors=1)

0
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/focused-regressions.stdout.txt Normal file
View file

143
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/hub-readback.json Normal file
View file

@ -0,0 +1,143 @@
{
"observed_at": "2026-10-17T02:06:06.866758+00:00",
"read_only": true,
"tasks": [
{
"source_id": "NK-WP-0049-T33",
"id": "56c8d469-7a10-5d50-a30e-95cbf3cd4756",
"workplan_id": "17d315f0-829e-57c7-85cc-3a7914152804",
"status": "wait",
"needs_human": false,
"blocking_reason": "Source custody/preflight preparation is evidenced in T34; await original TLS/trust recovery disposition T35 and selected original registration/private-admission metadata T36 before final reviewed command and existing owner process."
},
{
"source_id": "NK-WP-0049-T35",
"id": "1bfd1703-589f-5511-bad5-812d860e4a71",
"workplan_id": "17d315f0-829e-57c7-85cc-3a7914152804",
"status": "progress",
"needs_human": false,
"blocking_reason": null
},
{
"source_id": "NK-WP-0049-T36",
"id": "8b72334b-fc27-5707-a189-bad345eaf3bc",
"workplan_id": "17d315f0-829e-57c7-85cc-3a7914152804",
"status": "progress",
"needs_human": false,
"blocking_reason": null
},
{
"source_id": "RCLUSTER-WP-0018-T02",
"id": "90836435-917f-504c-a711-3f29ece348e3",
"workplan_id": "e0a7813d-c916-5e49-84b2-5e5b30441891",
"status": "todo",
"needs_human": false,
"blocking_reason": null
},
{
"source_id": "SECRETS-WP-0015-T03",
"id": "6262a915-7655-50be-b9f7-481f816d7335",
"workplan_id": "68a0642e-4908-5260-89c9-3adce60e6a4d",
"status": "todo",
"needs_human": false,
"blocking_reason": null
},
{
"source_id": "SECRETS-WP-0015-T04",
"id": "aeec6eb0-39e2-5f7c-8f5d-778133edfc73",
"workplan_id": "68a0642e-4908-5260-89c9-3adce60e6a4d",
"status": "wait",
"needs_human": true,
"blocking_reason": "Await exact consumer proof/custody plan plus original TLS/private admission readiness; actual attended owner path/authority remains required."
}
],
"latest_foreign_waits": [
{
"workplan_id": "17d315f0-829e-57c7-85cc-3a7914152804",
"id": "a2e5ed16-0da0-4ab9-bc0a-d82bd0a393f1",
"topic_id": null,
"task_id": null,
"decision_id": null,
"event_type": "coordination_receipt",
"summary": "Cluster WP18 published and verified with five tasks; T02 still requires exact independent original TLS artifact/provider/custody metadata. Publication and current-baseline maintenance proof give no TLS/native assent.",
"detail": {
"state": "waiting",
"flavor": "implementation",
"reason": "Cluster WP18 published and verified with five tasks; T02 still requires exact independent original TLS artifact/provider/custody metadata. Publication and current-baseline maintenance proof give no TLS/native assent.",
"case_id": "nk49-t35-rcluster18",
"observer": "codex",
"scenario_type": "cross_owner_wait",
"obligor_task_id": "RCLUSTER-WP-0018-T02",
"native_acceptance": false,
"beneficiary_task_id": "NK-WP-0049-T35",
"obligor_workplan_id": "RCLUSTER-WP-0018",
"coordination_context": "CUST-WP-0080",
"human_gate_never_woken": true,
"beneficiary_workplan_id": "NK-WP-0049",
"runtime_worker_injection_enabled": false
},
"author": "codex",
"session_id": null,
"created_at": "2026-10-17T00:39:42.467089Z",
"workstream_id": "17d315f0-829e-57c7-85cc-3a7914152804"
},
{
"workplan_id": "68a0642e-4908-5260-89c9-3adce60e6a4d",
"id": "3f905339-a4c9-428e-9cbd-fa490c144c7a",
"topic_id": null,
"task_id": null,
"decision_id": null,
"event_type": "coordination_receipt",
"summary": "Observed cross-owner case secrets15-nk49-readiness: Foreign source readiness remains incomplete; beneficiary also has a separate attended human gate",
"detail": {
"state": "waiting",
"flavor": "implementation",
"reason": "Foreign source readiness remains incomplete; beneficiary also has a separate attended human gate",
"case_id": "secrets15-nk49-readiness",
"observer": "codex",
"scenario_type": "cross_owner_wait",
"obligor_task_id": "NK-WP-0049-T33",
"native_acceptance": false,
"beneficiary_task_id": "SECRETS-WP-0015-T04",
"obligor_workplan_id": "NK-WP-0049",
"coordination_context": "CUST-WP-0080",
"human_gate_never_woken": true,
"beneficiary_workplan_id": "SECRETS-WP-0015",
"runtime_worker_injection_enabled": false
},
"author": "codex",
"session_id": null,
"created_at": "2026-10-16T23:44:28.741512Z",
"workstream_id": "68a0642e-4908-5260-89c9-3adce60e6a4d"
},
{
"workplan_id": "17d315f0-829e-57c7-85cc-3a7914152804",
"id": "8a73a503-7479-426b-b2ad-e88f32e4c7e8",
"topic_id": null,
"task_id": null,
"decision_id": null,
"event_type": "coordination_receipt",
"summary": "Observed cross-owner case nk49-t36-secrets15: SECRETS-WP-0015 published c59bbf56 and exact4task/2edge bindings observed; T03 remains todo for outage-compatible consumer preparation, no native assent.",
"detail": {
"state": "waiting",
"flavor": "implementation",
"reason": "SECRETS-WP-0015 published c59bbf56 and exact4task/2edge bindings observed; T03 remains todo for outage-compatible consumer preparation, no native assent.",
"case_id": "nk49-t36-secrets15",
"observer": "codex",
"scenario_type": "cross_owner_wait",
"obligor_task_id": "SECRETS-WP-0015-T03",
"native_acceptance": false,
"beneficiary_task_id": "NK-WP-0049-T36",
"obligor_workplan_id": "SECRETS-WP-0015",
"coordination_context": "CUST-WP-0080",
"human_gate_never_woken": true,
"beneficiary_workplan_id": "NK-WP-0049",
"runtime_worker_injection_enabled": false
},
"author": "codex",
"session_id": null,
"created_at": "2026-10-16T23:44:28.363998Z",
"workstream_id": "17d315f0-829e-57c7-85cc-3a7914152804"
}
]
}

38
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/manifest.json Normal file
View file

@ -0,0 +1,38 @@
{
"schema": "custodian.native-identity-source-packet.v1",
"created_at": "2026-10-17T02:08:34.423600+00:00",
"status": "source_prepared_review_pending_no_native_acceptance",
"proposal_id": "nk49-original-identity-attempt-20261017",
"original_revision": 1,
"original_proposal_digest": "sha256:8cc9ee58131d41c9bd2951c233c3d2c5916fb6ea0572e73cc99b287adc61f858",
"canonical_source_paths": [
"tools/bootstrap-recovery/isolated-keycape.py",
"tests/test_isolated_keycape_recovery.py",
"tests/isolated-keycape-native-fixture-manifest.json",
"workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md"
],
"files": {
"README.md": "sha256:fae49e17e652226c024ac0eae9d5fd967a82fa3b37441a448380edfe2ba07879",
"consumer-proof-plan.md": "sha256:0d47336d44c2e354a7e1c862fcb96bb097fdae7042e9c85caf1c978fba314693",
"focused-regressions.stderr.txt": "sha256:c28f2305fc55ba0571b9e23fd7fd7482b50e85e352c25be0677b9d86882f1ec1",
"focused-regressions.stdout.txt": "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
"hub-readback.json": "sha256:f01ff4e13bf8aad12cb042607ed7ca96ec58673e47edb343176addc45467d5f8",
"next-attempt-proposal.json": "sha256:8cc9ee58131d41c9bd2951c233c3d2c5916fb6ea0572e73cc99b287adc61f858",
"original-behavior-regression.txt": "sha256:1da5266b47ce0be90bfd4056a7b019fde31566654180bc1be4bedd6805e6cb3d",
"owner-declaration.template.json": "sha256:57268212a2f1d1c34fdc2f6587a26724468ee2045c9a88d8523b74452f8f3c09",
"owner-source.patch": "sha256:d02de220d28a293ad22e1781b9d0f42b0bf69456de785a30022851eb723b83e7",
"postimages/tests/isolated-keycape-native-fixture-manifest.json": "sha256:936ee435be04b27f391391e7dda02def69e153bde3adcce92d569f33f308db2e",
"postimages/tests/test_isolated_keycape_recovery.py": "sha256:5f729e077db51dd6b17d00dc3eb71c0e652849f55fe68505a07befa663c91dea",
"postimages/tools/bootstrap-recovery/isolated-keycape.py": "sha256:86f37add7221e4c73e3f9d878c09a3cc81061627cbf024ed6887d973d929b718",
"postimages/workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md": "sha256:5830c429f58a0317b28c7b4bcf22a1b46e6d01b41378cdae95f947c6d9317e44",
"preimages/tests/isolated-keycape-native-fixture-manifest.json": "sha256:1dcefbc7ca7b0c2be33135b1b7977b89ba657223ecbb1310cc691d151b3d916e",
"preimages/tests/test_isolated_keycape_recovery.py": "sha256:247e9a87f5be2ee9f20f7edbb4a9f612ab7cb0cd940392c3caeca8b046605f55",
"preimages/tools/bootstrap-recovery/isolated-keycape.py": "sha256:f67627b7d1647d75187dd899874efe22383bc91de5d97dabb9801076668ab3c2",
"preimages/workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md": "sha256:678ec04b73b57c026001791706c05c5d4d5a508a32f17210ae8ad03ab15d53b8",
"routing-discovery.json": "sha256:763a36b7da8a6476d50e4f377eee3415b7e1460deea1d3827064a200955350e4",
"selector-preflight.json": "sha256:b3946e11ce91b7a9d3e24e4e218e21739f12216cf4621fc216fccc2147439d41",
"selector-uniqueness.patch": "sha256:55aa1efad828e6c466ec92430e94eb4366b0b8470bc790ccfa056fb0b3125853",
"source-references.json": "sha256:a6484b77be832733ceba7b11563ca73b472b9f17f02c89002575d9a0e1e3a42b",
"verification.json": "sha256:4d2beb8cd1871c13de7f6b5fe0ffe6c2d20b5ff5c1e39537200f9fc0e8d04b43"
}
}

69
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/next-attempt-proposal.json Normal file
View file

@ -0,0 +1,69 @@
{
"proposal_id": "nk49-original-identity-attempt-20261017",
"revision": 1,
"status": "incomplete_not_executable_not_approved",
"target": {
"docker_endpoint": "unix:///var/run/docker.sock",
"engine_id": "74f93783-6cef-43a8-a3c4-f62586cd3bc1",
"engine_name": "bnt-lap001",
"production_contact": false
},
"archive": {
"ciphertext_sha256": "sha256:7117b3fe1e4840c86b301c94c85e6b16709e5626bb0884bd0580bde630e30e30",
"manifest_binding_mode": "ciphertext-derived-v1",
"verified_transfer_receipt_reference": null
},
"public_original_identity": {
"issuer": null,
"tls_certificate_chain_sha256": null,
"trust_sha256": null,
"owner_originality_provenance_reference": null,
"private_custody_entrypoint_reference": null
},
"candidate_selector": {
"client_id": "secrets-engine-openbao",
"secret_env": "KEYCAPE_SECRETS_ENGINE_OPENBAO_CLIENT_SECRET",
"audience": "secrets-engine-openbao",
"scope": "openbao:login",
"subject": "service:secrets-engine",
"tenant": "tenant:platform",
"roles": "secrets-engine",
"deny_scope": "openbao:admin"
},
"selector_status": "current_source_correspondence_only_no_archive_or_owner_assent",
"current_owner_assent_reference": null,
"reviewed_complete_source_closure_reference": null,
"exact_admission_sha256": null,
"nonexecutable_command_shape": [
"bash",
"tools/bootstrap-recovery/run-isolated-keycape.sh",
"<verified-transfer-receipt>",
"<complete-reviewed-admission>",
"<exact-admission-sha256>",
"<owner-local-original-public-cert-chain>",
"<owner-private-original-TLS-key-path>",
"<owner-local-original-public-trust>"
],
"verification": [
"Pre-input metadata/source/dependency/target checks before hidden prompt",
"Authenticated archive validation and exact-one original selector before further runtime/private reads",
"Pinned original-image issuer/JWKS/TLS/native positive and negative checks",
"Independent sanitized receipt and exact cleanup verification"
],
"rollback_and_abort": "Abort on any refusal/failure/uncertainty; exact session-scoped removal and safe evidence only. No production rollback or broad cleanup authorized.",
"deadline_seconds": 1800,
"native_checker_timeout_seconds": 150,
"cost_estimate_eur": null,
"credible_loss_bound_eur": null,
"estimate_status": "Unknown for actual custody/native operation; supervisor must evaluate concrete completed proposal. No budget or authority granted.",
"supervisor_disposition": null,
"approved_revision_digest": null,
"executed_revision_digest": null,
"actual_outcome": null,
"blocked_on": [
"Original independent TLS public identity/trust provenance and private custody, or explicit accepted reissue/continuity disposition",
"Exact archive-to-selector binding and current actual owner recovery assent",
"Consumer outage-compatible accepted source/proof scope",
"Complete re-reviewed source/dependency closure after adapter guard change"
]
}

65
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/original-behavior-regression.txt Normal file
View file

@ -0,0 +1,65 @@
test_absent_original_selector_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_absent_original_selector_refused_before_further_inputs) ... ERROR
test_conflicting_duplicate_selector_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_conflicting_duplicate_selector_refused_before_further_inputs) ... ERROR
test_empty_registry_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_empty_registry_refused_before_further_inputs) ... ERROR
test_identical_duplicate_selector_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_identical_duplicate_selector_refused_before_further_inputs) ... ERROR
======================================================================
ERROR: test_absent_original_selector_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_absent_original_selector_refused_before_further_inputs)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 77, in test_absent_original_selector_refused_before_further_inputs
self.assert_refused_before_further_inputs([dict(self.client(),clientId='unrelated')])
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 71, in assert_refused_before_further_inputs
with self.assertRaises(a.Refused):a.run_verifier({},admission,payload)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/worsch/the-custodian/docs/changes/netkingdom-target-2026-10-17/native-identity-r1/preimages/tools/bootstrap-recovery/isolated-keycape.py", line 325, in run_verifier
issuer = session['issuer']
~~~~~~~^^^^^^^^^^
KeyError: 'issuer'
======================================================================
ERROR: test_conflicting_duplicate_selector_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_conflicting_duplicate_selector_refused_before_further_inputs)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 81, in test_conflicting_duplicate_selector_refused_before_further_inputs
self.assert_refused_before_further_inputs([self.client(),dict(self.client(),secretRef='env:OTHER')])
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 71, in assert_refused_before_further_inputs
with self.assertRaises(a.Refused):a.run_verifier({},admission,payload)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/worsch/the-custodian/docs/changes/netkingdom-target-2026-10-17/native-identity-r1/preimages/tools/bootstrap-recovery/isolated-keycape.py", line 325, in run_verifier
issuer = session['issuer']
~~~~~~~^^^^^^^^^^
KeyError: 'issuer'
======================================================================
ERROR: test_empty_registry_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_empty_registry_refused_before_further_inputs)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 83, in test_empty_registry_refused_before_further_inputs
self.assert_refused_before_further_inputs([])
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 71, in assert_refused_before_further_inputs
with self.assertRaises(a.Refused):a.run_verifier({},admission,payload)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/worsch/the-custodian/docs/changes/netkingdom-target-2026-10-17/native-identity-r1/preimages/tools/bootstrap-recovery/isolated-keycape.py", line 325, in run_verifier
issuer = session['issuer']
~~~~~~~^^^^^^^^^^
KeyError: 'issuer'
======================================================================
ERROR: test_identical_duplicate_selector_refused_before_further_inputs (selector_regressions.OriginalSelectorTests.test_identical_duplicate_selector_refused_before_further_inputs)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 79, in test_identical_duplicate_selector_refused_before_further_inputs
self.assert_refused_before_further_inputs([self.client(),self.client()])
File "/home/worsch/net-kingdom/tests/test_isolated_keycape_recovery.py", line 71, in assert_refused_before_further_inputs
with self.assertRaises(a.Refused):a.run_verifier({},admission,payload)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/home/worsch/the-custodian/docs/changes/netkingdom-target-2026-10-17/native-identity-r1/preimages/tools/bootstrap-recovery/isolated-keycape.py", line 325, in run_verifier
issuer = session['issuer']
~~~~~~~^^^^^^^^^^
KeyError: 'issuer'
----------------------------------------------------------------------
Ran 4 tests in 0.023s
FAILED (errors=4)

38
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/owner-declaration.template.json Normal file
View file

@ -0,0 +1,38 @@
{
"status": "draft_for_actual_owners_no_assent",
"tls": {
"source_owner": "railiance-cluster",
"namespace_intent_owner": "net-kingdom",
"original_private_custody_owner": null,
"provider": null,
"artifact_reference": null,
"capture_time": null,
"capture_binding_to_archive_attempt": null,
"independent_private_custody_entrypoint_reference": null,
"outage_prerequisites": null,
"public_metadata_receipt": null,
"reissue_disposition_reference": null,
"accepted_continuity_scope": null,
"owner_review_reference": null
},
"consumer": {
"source_owner": "secrets-engine",
"selector": {
"client_id": "secrets-engine-openbao",
"secret_env": "KEYCAPE_SECRETS_ENGINE_OPENBAO_CLIENT_SECRET",
"audience": "secrets-engine-openbao",
"scope": "openbao:login",
"subject": "service:secrets-engine",
"tenant": "tenant:platform",
"roles": "secrets-engine",
"deny_scope": "openbao:admin"
},
"current_recovery_assent_reference": null,
"exact_archived_registration_uniqueness_receipt": null,
"archive_to_consumer_binding_receipt": null,
"outage_compatible_private_custody_plan_reference": null,
"continuity_scope_disposition_reference": null,
"owner_review_reference": null
},
"instruction": "Actual owners fill references through existing processes; agents do not fabricate owner/founder assent. Never include values, private manifest digest or executable approval tokens."
}

136
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/owner-source.patch Normal file
View file

@ -0,0 +1,136 @@
--- a/tools/bootstrap-recovery/isolated-keycape.py
+++ b/tools/bootstrap-recovery/isolated-keycape.py
@@ -322,6 +322,23 @@
def run_verifier(session, admission, payload):
+ # Select from the already-restored config before further private/runtime reads.
+ # This is not public preflight before archive decryption or its key prompt.
+ import yaml
+ cfg = yaml.safe_load(base64.b64decode(payload['files']['/etc/keycape/config.yaml'],validate=True))
+ if not isinstance(cfg,dict) or cfg.get('issuer') != admission['issuer']:
+ raise Refused()
+ consumer = admission['consumer']
+ clients = cfg.get('clients')
+ if not isinstance(clients,list) or any(not isinstance(client,dict) for client in clients):
+ raise Refused()
+ matches = [client for client in clients if client.get('clientId') == consumer['client_id']]
+ if len(matches) != 1:
+ raise Refused()
+ client = matches[0]
+ if (client.get('secretRef') != 'env:'+consumer['secret_env'] or client.get('clientType') != 'confidential'
+ or 'client_credentials' not in client.get('grantTypes',[])):
+ raise Refused()
issuer = session['issuer']
if {v['target'] for v in session.get('volumes',{}).values()}!={'/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls'}:raise Refused()
from urllib.parse import urlsplit
@@ -352,15 +369,7 @@
if durable!={k:v for k,v in files.items() if k.startswith('/var/lib/keycape-policy/')}:
raise Refused()
equal()
- # Reading YAML is data parsing only; no archived executable is evaluated.
- import yaml
- cfg = yaml.safe_load(files['/etc/keycape/config.yaml'])
- if cfg['issuer'] != admission['issuer']:
- raise Refused()
- consumer = admission['consumer']
- client = next(x for x in cfg['clients'] if x['clientId']==consumer['client_id'])
- if (client['secretRef']!='env:'+consumer['secret_env'] or client['clientType']!='confidential'
- or 'client_credentials' not in client['grantTypes'] or not env.get(consumer['secret_env'])):
+ if not env.get(consumer['secret_env']):
raise Refused()
from cryptography.hazmat.primitives import serialization
key=serialization.load_pem_private_key(files['/etc/keycape/key.pem'],password=None)
--- a/tests/test_isolated_keycape_recovery.py
+++ b/tests/test_isolated_keycape_recovery.py
@@ -53,6 +53,34 @@
for row in self.manifest['files']:
if row['path'].endswith('/deployment/keycape.json'):row.update(bytes=len(raw),sha256=hashlib.sha256(raw).hexdigest())
with self.assertRaises(a.Refused):self.payload()
+
+
+class OriginalSelectorTests(unittest.TestCase):
+ """Already-restored public fixture config; no archive/key/native action."""
+ def assert_refused_before_further_inputs(self,clients):
+ import yaml
+ cfg={'issuer':'https://issuer.example','clients':clients}
+ payload={'files':{'/etc/keycape/config.yaml':base64.b64encode(yaml.safe_dump(cfg).encode()).decode(),
+ '/etc/keycape/key.pem':'deliberately-not-base64'}}
+ admission={'issuer':'https://issuer.example','consumer':{'client_id':'existing','secret_env':'KEYCAPE_ORIGINAL_SECRET'}}
+ # Empty session/no env and an undecodable key also expose early private access.
+ from contextlib import ExitStack
+ with ExitStack() as stack:
+ blocked=[stack.enter_context(patch.object(a,name,side_effect=AssertionError('later input or native call')))
+ for name in ('container_file','runtime_file','runtime_files','command','check_jwks','process')]
+ with self.assertRaises(a.Refused):a.run_verifier({},admission,payload)
+ for call in blocked:call.assert_not_called()
+ def client(self):
+ return {'clientId':'existing','clientType':'confidential','secretRef':'env:KEYCAPE_ORIGINAL_SECRET',
+ 'grantTypes':['client_credentials']}
+ def test_absent_original_selector_refused_before_further_inputs(self):
+ self.assert_refused_before_further_inputs([dict(self.client(),clientId='unrelated')])
+ def test_identical_duplicate_selector_refused_before_further_inputs(self):
+ self.assert_refused_before_further_inputs([self.client(),self.client()])
+ def test_conflicting_duplicate_selector_refused_before_further_inputs(self):
+ self.assert_refused_before_further_inputs([self.client(),dict(self.client(),secretRef='env:OTHER')])
+ def test_empty_registry_refused_before_further_inputs(self):
+ self.assert_refused_before_further_inputs([])
class NativeTests(unittest.TestCase):
def setUp(self):
--- a/tests/isolated-keycape-native-fixture-manifest.json
+++ b/tests/isolated-keycape-native-fixture-manifest.json
@@ -7,7 +7,7 @@
"tests/run_isolated_keycape_native_fixture.py": "sha256:ecbadecc475622eaaf20a726110aebef5f4fe7af141110a98aca1640179b4a03",
"tests/test_identity_bundle.py": "sha256:be5ce74097594cc7f7e77dbccd6cb53aa72f263cd1bb442ce5464038aed1116c",
"tests/test_isolated_keycape_fixture_plan.py": "sha256:5945619f4c5957187763276cc900870a9f08f9475b63c240694688e0c2eca7dd",
- "tests/test_isolated_keycape_recovery.py": "sha256:247e9a87f5be2ee9f20f7edbb4a9f612ab7cb0cd940392c3caeca8b046605f55",
+ "tests/test_isolated_keycape_recovery.py": "sha256:5f729e077db51dd6b17d00dc3eb71c0e652849f55fe68505a07befa663c91dea",
"tests/test_isolated_keycape_session.py": "sha256:cbdf51efde26eff337e9303a2c431f80b333b8846d763eb62853f81eb3f13e2c",
"tools/bootstrap-recovery/capture-identity.py": "sha256:47dca7c4dc96fa7eb9fb02c81fca9ca8dc5281aac918979eb7367148168ba705",
"tools/bootstrap-recovery/check-privacyidea-database.py": "sha256:90f454c87998d7f5d625e3c6656c3418f8f56275dc1879a9d19ed605431b94a4",
@@ -25,7 +25,7 @@
"tools/bootstrap-recovery/inventory-scaleway.py": "sha256:a219df205708fa4ed44b763a418585d63c820c25434bc0fd14e805eb5c2f79d3",
"tools/bootstrap-recovery/inventory-storage-share.py": "sha256:673f9b17fb2f3fef6fcdb5c50a2bcca69e3df3157a9e5582a2d995c6b87640b5",
"tools/bootstrap-recovery/isolated-keycape-session.py": "sha256:8fa38efe28167b59a6307f342b2ac8fcda78ecda60e6228f22017424d2343c22",
- "tools/bootstrap-recovery/isolated-keycape.py": "sha256:f67627b7d1647d75187dd899874efe22383bc91de5d97dabb9801076668ab3c2",
+ "tools/bootstrap-recovery/isolated-keycape.py": "sha256:86f37add7221e4c73e3f9d878c09a3cc81061627cbf024ed6887d973d929b718",
"tools/bootstrap-recovery/isolated-lldap.py": "sha256:35bfa8411186ad936a199bd450c7e18c2ad51059301a6db9650e6bd055ab1b81",
"tools/bootstrap-recovery/isolated-privacyidea.py": "sha256:4df7196678aac1ad594f32abf45c68dc21a9d2760d9e8a536a215a4a3c5d277f",
"tools/bootstrap-recovery/prepare-identity-coverage.py": "sha256:cd5f3067d28b4c7cc93ed08b8cf3869218bdecf7c54b5dffed22d697de71dbb1",
@@ -81,7 +81,7 @@
"synthetic_source_root": "private /dev/shm/nk49-keycape-fixture-<suffix>",
"safe_result": "private /tmp/nk49-keycape-native-fixture-review-<suffix>/result.json"
},
- "scope": "candidate itself on complete synthetic format/integrity archive fixture; no native PI/LLDAP/Authelia proof, real custody, real factor or real client credential; successor binds sealed-descriptor custody preparation sources; synthetic fixture itself overrides custody with generated fixture key as before",
+ "scope": "candidate itself on complete synthetic format/integrity archive fixture; no native PI/LLDAP/Authelia proof, real custody, real factor or real client credential; successor binds sealed-descriptor custody preparation sources; synthetic fixture itself overrides custody with generated fixture key as before; unexecuted exact-one-selector adapter successor: no new native outcome inferred from earlier fixtures",
"mutations": [
"conditional creation of exact fresh local containers and local-driver tmpfs volumes",
"private synthetic file import to those volumes",
--- a/workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md
+++ b/workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md
@@ -1130,6 +1130,24 @@
`recovery-owner-handoffs-r1` source review; independent ID/UUID/status/remote
readback is recorded separately when performed.
+2026-10-17 exact-selector source preparation: the adapter now requires exactly
+one selected client in the already-restored archived configuration before further
+runtime environment reads, signer-key parsing or native checker invocation.
+Absent, duplicate (identical or conflicting) and empty selectors refuse early;
+existing unique registration behavior is retained. This check cannot run before
+private archive decryption or its key prompt. Current KeyCape Go configuration
+also rejects duplicate IDs; this is an explicit adapter guard, not a claim that
+the pinned native issuer admitted duplicates. Tests use public/synthetic config
+and mocked native calls only. Changed adapter bytes require a fresh reviewed
+source/dependency and admission binding; old actual fixture receipts are retained.
+The exact eight-field current consumer selector independently passes its offline
+AST-only preflight. TLS public provenance/private custody, archive-time selector
+binding, current owner recovery assent and outage-compatible actual consumer
+proof remain open under T35/T36 and the existing owner tasks. No actual archive,
+key, Secret object, token exchange, OpenBao login, rotation or native action was
+performed. Source evidence and proposed owner declaration/consumer plan:
+Custodian docs/changes/netkingdom-target-2026-10-17/native-identity-r1/.
+
## Verify actual SOPS compatibility and sealed-descriptor read/write custody
```task

124
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/postimages/tests/isolated-keycape-native-fixture-manifest.json Normal file
View file

@ -0,0 +1,124 @@
{
"schema": "nk49.keycape-native-fixture-proposal.v1",
"status": "prepared_not_executed_not_approved",
"source_sha256": {
"sso-mfa/bootstrap/sops-private.py": "sha256:acac68e2936ed38b7d7c9e35eb4619647c99c117a3928a1c5ce704b570906780",
"sso-mfa/k8s/privacyidea/key-custody-remote.py": "sha256:27b2b885c81923d85054e3999c5613e104fe627779b2b50398ed9bed5088f47e",
"tests/run_isolated_keycape_native_fixture.py": "sha256:ecbadecc475622eaaf20a726110aebef5f4fe7af141110a98aca1640179b4a03",
"tests/test_identity_bundle.py": "sha256:be5ce74097594cc7f7e77dbccd6cb53aa72f263cd1bb442ce5464038aed1116c",
"tests/test_isolated_keycape_fixture_plan.py": "sha256:5945619f4c5957187763276cc900870a9f08f9475b63c240694688e0c2eca7dd",
"tests/test_isolated_keycape_recovery.py": "sha256:5f729e077db51dd6b17d00dc3eb71c0e652849f55fe68505a07befa663c91dea",
"tests/test_isolated_keycape_session.py": "sha256:cbdf51efde26eff337e9303a2c431f80b333b8846d763eb62853f81eb3f13e2c",
"tools/bootstrap-recovery/capture-identity.py": "sha256:47dca7c4dc96fa7eb9fb02c81fca9ca8dc5281aac918979eb7367148168ba705",
"tools/bootstrap-recovery/check-privacyidea-database.py": "sha256:90f454c87998d7f5d625e3c6656c3418f8f56275dc1879a9d19ed605431b94a4",
"tools/bootstrap-recovery/fetch-privacyidea-backup.py": "sha256:233160e1206643a5cf7ad05141d733f0a75aff651190e27f77515a124ff0a59f",
"tools/bootstrap-recovery/identity-backup-profile.json": "sha256:40f9e5b5bb957e5b23e03b31701702b850502d16a92a0fa0ebf92618cc3f18a1",
"tools/bootstrap-recovery/identity-bundle.py": "sha256:524a03c065db4307fcde83d33af39503f3de19cb6ef3a8ca72c991d4a8c08ee6",
"tools/bootstrap-recovery/identity-credential-setup.py": "sha256:ddd6127fa331c0185523ec3bb22d743a791edd56a2e6af7ad8d7949b7c051141",
"tools/bootstrap-recovery/identity-custody.json": "sha256:f8aee8487ed01d7940b82676bb4a5c8d61f51b6c598521b8b8a73588aaea4d4f",
"tools/bootstrap-recovery/identity-files.py": "sha256:d0063a9eedd3610e3d45bd2d834aae83cff74e4b3504d138b3548f427e89331d",
"tools/bootstrap-recovery/identity-isolation.py": "sha256:88d1e13112a44c79571825af89d3fe5b01fa490cff2653badef4c62a45c844f1",
"tools/bootstrap-recovery/identity-offsite.py": "sha256:e7c31fa9ee081961af873348ee3b4829a645d40c7dc7f1e22c564aa9ed01bc49",
"tools/bootstrap-recovery/identity-retention.py": "sha256:a41ea4759f69c35548149fc8d0d5ace547f16cf9fe4c9dac10b119a4d0becdec",
"tools/bootstrap-recovery/identity-runtime.py": "sha256:a2c0acf567c204705b90be64d561d0fda2285e0fc71a962d6e1b149f571e1940",
"tools/bootstrap-recovery/install-identity-backup.py": "sha256:5c612a2a51c9b59aad8cc01d418c2f563cbd92c62dc1fa8be994e5fb59df90f1",
"tools/bootstrap-recovery/inventory-scaleway.py": "sha256:a219df205708fa4ed44b763a418585d63c820c25434bc0fd14e805eb5c2f79d3",
"tools/bootstrap-recovery/inventory-storage-share.py": "sha256:673f9b17fb2f3fef6fcdb5c50a2bcca69e3df3157a9e5582a2d995c6b87640b5",
"tools/bootstrap-recovery/isolated-keycape-session.py": "sha256:8fa38efe28167b59a6307f342b2ac8fcda78ecda60e6228f22017424d2343c22",
"tools/bootstrap-recovery/isolated-keycape.py": "sha256:86f37add7221e4c73e3f9d878c09a3cc81061627cbf024ed6887d973d929b718",
"tools/bootstrap-recovery/isolated-lldap.py": "sha256:35bfa8411186ad936a199bd450c7e18c2ad51059301a6db9650e6bd055ab1b81",
"tools/bootstrap-recovery/isolated-privacyidea.py": "sha256:4df7196678aac1ad594f32abf45c68dc21a9d2760d9e8a536a215a4a3c5d277f",
"tools/bootstrap-recovery/prepare-identity-coverage.py": "sha256:cd5f3067d28b4c7cc93ed08b8cf3869218bdecf7c54b5dffed22d697de71dbb1",
"tools/bootstrap-recovery/recovery-kit.py": "sha256:6fa0f1494dccaf003b6c6396c5de8b36faa2b276eeb16badd04383f5efb71e66",
"tools/bootstrap-recovery/retrieve-identity-offline.py": "sha256:3421dbb3f3280a001d33ea2deb12ab609404a0b4254cd43c21518d77d166f972",
"tools/bootstrap-recovery/retrieve-identity.py": "sha256:ec123d47a07305253570dacab7e103f8c193a23f20c9fc87466ea4a95ae0ec13",
"tools/bootstrap-recovery/run-isolated-keycape.sh": "sha256:0d994c9c14ebdffdbf7879bd7562042531083d0e8955424d0ae099898a141668",
"tools/bootstrap-recovery/verify-bundle.py": "sha256:885b05a2011e711228e42e316bda70bbe0d97d481383d61aa8469d0f10452370",
"tools/bootstrap-recovery/verify-identity-delivery.py": "sha256:63658d1976806f8d0ce80ba8f750bbcfdca59da936b6eb6a78a250f0e53710f1",
"tools/bootstrap-recovery/verify-identity.py": "sha256:039864d0141b552887752bf17cfdcf1bef4bdc9126f946a9c0533e4fd54ebf4f",
"sso-mfa/bootstrap/sops-custody-unlock.sh": "sha256:86c527a8ae0d0cca66a2345c566b25fbff5bc52c1e22c725548c0fbb2f86a58e",
"sso-mfa/bootstrap/sops-custody-fd.py": "sha256:df29eaba83ec6b5fc311664dcf4b22d8f3c8ed7274ed8e6c419ada0dcd169c0f",
"tools/bootstrap-recovery/prepare-isolated-keycape.py": "sha256:5098bdfa4e0ad30627c1716e5e355a2c78dca6ccc607bd2148a2e06126632288",
"tests/test_prepare_isolated_keycape.py": "sha256:4b0871f9a11a3940983464f1b40eca1613110b1cb3cb9694f4891d58e14c700a",
"tests/test_sealed_custody.py": "sha256:f0fa07c89e2419464dbee0489462864a4533770417a554b1ab1d1ed5d7b6c49f",
"tests/test_private_manifest_binding.py": "sha256:2e7ee16af65390343435b7210ef191aa2a10e9dcbb8759ed95fa8bf6fb81893b"
},
"fixed_endpoint": "unix:///var/run/docker.sock",
"engine_id": "74f93783-6cef-43a8-a3c4-f62586cd3bc1",
"engine_name": "bnt-lap001",
"images": [
"forgejo.coulomb.social/coulomb/key-cape@sha256:cdec9dd7c515f7e0a6169fc3fa560ee2cc47bf40d899755dec808c362f5ce010",
"ghcr.io/gpappsoft/privacyidea-docker@sha256:af7841adad262f129e0c1d4f553af13f21cb2f4dc713533f316cfe43ed0b4473"
],
"pull_policy": "never",
"synthetic_issuer": "https://nk49-fixture.invalid",
"cases": [
"positive",
"hard-parent-loss",
"deadline",
"nonzero-after-accept"
],
"limits": {
"sequential_cases": true,
"max_live_service_containers": 2,
"max_tmpfs_volumes": 4,
"max_public_probe_containers": 1,
"service_memory_bytes_each": 536870912,
"service_cpus_each": 1,
"tmpfs_volume_bytes_each": 33554432,
"host_tmpfs_fixture_limit_expected_bytes": 8388608,
"execution_deadline_seconds": {
"deadline_case": 20,
"other_cases": 120
},
"driver_wait_seconds_per_case": 150
},
"resources": {
"container_names": "nk49-kc-<fresh-run-id>-{transport,issuer,image-probe}",
"volume_names": "nk49-kc-<fresh-run-id>-v{0,1,2,3}",
"label": "net-kingdom.recovery-run=<fresh-run-id>",
"host_plaintext": "exact private /dev/shm/nk49-keycape-<run-id>-<suffix> registered before writes",
"synthetic_source_root": "private /dev/shm/nk49-keycape-fixture-<suffix>",
"safe_result": "private /tmp/nk49-keycape-native-fixture-review-<suffix>/result.json"
},
"scope": "candidate itself on complete synthetic format/integrity archive fixture; no native PI/LLDAP/Authelia proof, real custody, real factor or real client credential; successor binds sealed-descriptor custody preparation sources; synthetic fixture itself overrides custody with generated fixture key as before; unexecuted exact-one-selector adapter successor: no new native outcome inferred from earlier fixtures",
"mutations": [
"conditional creation of exact fresh local containers and local-driver tmpfs volumes",
"private synthetic file import to those volumes",
"original-image KeyCape service/client exchange within same loopback-only namespace",
"exact owned quiescent removal and bounded safe receipts"
],
"forbidden": [
"Docker execution before independent approval",
"image pulls/builds/pushes",
"production routes/ports/host networking/host directory mounts",
"OpenBao/Kubernetes/provider APIs",
"actual identity archive, age key, TLS key, user password, factor or client secret"
],
"cleanup": "candidate exact ownership/volume-driver/inode checks; quiesce fresh owner SID/PGID via pidfds first; retained inflight uncertainty never accepted even on absent readback",
"oracles": [
"positive issuer-native verify-client must establish original-hostname HTTPS trust; helper alone never suffices",
"same original single RS256 key-1 JWKS and exact config/registry/policy/history/lock/deliveries",
"correct service exchange and explicit wrong-secret/excess-scope denial through existing native verifier",
"hard-parent-loss and tightened deadline both quiesce before exact resource/plaintext cleanup",
"after real Docker run accepts a synthetic container, inject returncode1; require failed recovery and sticky daemon uncertainty after cleanup"
],
"proposed_command": [
"python3",
"tests/run_isolated_keycape_native_fixture.py",
"--execute-reviewed-fixture",
"--expected-manifest-sha256",
"<SHA256-of-this-file>",
"--case",
"positive"
],
"extended_command": [
"python3",
"tests/run_isolated_keycape_native_fixture.py",
"--execute-reviewed-fixture",
"--expected-manifest-sha256",
"<SHA256-of-this-file>",
"--case",
"all"
]
}

262
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/postimages/tests/test_isolated_keycape_recovery.py Normal file
View file

@ -0,0 +1,262 @@
import base64
import hashlib
import importlib.util
import io
import json
import subprocess
import tarfile
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
SOURCE=Path(__file__).parents[1]/'tools/bootstrap-recovery/isolated-keycape.py'
spec=importlib.util.spec_from_file_location('adapter',SOURCE)
a=importlib.util.module_from_spec(spec);spec.loader.exec_module(a)
class ArchiveTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
self.root=Path(self.tmp.name)
config={'data':{k:base64.b64encode(v).decode() for k,v in {'config.yaml':b'issuer: https://issuer.example','key.pem':b'fixture-key'}.items()}}
self.values={'resources/sso/secret/keycape-config.json':json.dumps(config).encode(),
'resources/sso/deployment/keycape.json':json.dumps({'spec':{'template':{'spec':{'containers':[{'name':'keycape','image':a.IMAGE,'env':[
{'name':'KEYCAPE_CONFIG','value':'/etc/keycape/config.yaml'},
{'name':'KEYCAPE_POLICY_PATH','value':'/var/lib/keycape-policy/policy.json'},
{'name':'KEYCAPE_ORIGINAL_SECRET','valueFrom':{'secretKeyRef':{'name':'client','key':'secret'}}}]}]}}}}).encode(),
'resources/sso/secret/client.json':json.dumps({'data':{'secret':base64.b64encode(b'fixture-private').decode()}}).encode(),
'data/keycape/policy.json':b'{}','data/keycape/receipts/one.json':b'{"original":true}'}
for name,value in self.values.items():
p=self.root/name;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(value);p.chmod(0o600)
self.manifest={'profile':{'components':[{'namespace':'sso','deployment':'keycape','secret_refs':['keycape-config','client'],'configmap_refs':[]}]},
'files':[{'path':k,'bytes':len(v),'sha256':hashlib.sha256(v).hexdigest()} for k,v in self.values.items()]}
def payload(self):
digest=hashlib.sha256(json.dumps(self.manifest,sort_keys=True,separators=(',',':')).encode()).hexdigest()
return a.private_payload(self.root,self.manifest,digest)
def test_original_state_and_secret_delivery_retained(self):
p=self.payload();self.assertEqual(p['env']['KEYCAPE_ORIGINAL_SECRET'],'fixture-private')
self.assertEqual(base64.b64decode(p['files']['/var/lib/keycape-policy/receipts/one.json']),b'{"original":true}')
def test_archive_digest_refused(self):
with self.assertRaises(a.Refused):a.private_payload(self.root,self.manifest,'0'*64)
def test_changed_file_refused(self):
(self.root/'data/keycape/policy.json').write_text('{"changed":true}')
with self.assertRaises(a.Refused):self.payload()
def test_symlink_refused(self):
p=self.root/'data/keycape/policy.json';p.unlink();p.symlink_to('/etc/hostname')
with self.assertRaises(Exception):self.payload()
def test_duplicate_manifest_refused(self):
self.manifest['files'].append(self.manifest['files'][0])
with self.assertRaises(a.Refused):self.payload()
def test_image_mismatch_refused(self):
d=json.loads(self.values['resources/sso/deployment/keycape.json']);d['spec']['template']['spec']['containers'][0]['image']='keycape:latest'
raw=json.dumps(d).encode();p=self.root/'resources/sso/deployment/keycape.json';p.write_bytes(raw)
for row in self.manifest['files']:
if row['path'].endswith('/deployment/keycape.json'):row.update(bytes=len(raw),sha256=hashlib.sha256(raw).hexdigest())
with self.assertRaises(a.Refused):self.payload()
class OriginalSelectorTests(unittest.TestCase):
"""Already-restored public fixture config; no archive/key/native action."""
def assert_refused_before_further_inputs(self,clients):
import yaml
cfg={'issuer':'https://issuer.example','clients':clients}
payload={'files':{'/etc/keycape/config.yaml':base64.b64encode(yaml.safe_dump(cfg).encode()).decode(),
'/etc/keycape/key.pem':'deliberately-not-base64'}}
admission={'issuer':'https://issuer.example','consumer':{'client_id':'existing','secret_env':'KEYCAPE_ORIGINAL_SECRET'}}
# Empty session/no env and an undecodable key also expose early private access.
from contextlib import ExitStack
with ExitStack() as stack:
blocked=[stack.enter_context(patch.object(a,name,side_effect=AssertionError('later input or native call')))
for name in ('container_file','runtime_file','runtime_files','command','check_jwks','process')]
with self.assertRaises(a.Refused):a.run_verifier({},admission,payload)
for call in blocked:call.assert_not_called()
def client(self):
return {'clientId':'existing','clientType':'confidential','secretRef':'env:KEYCAPE_ORIGINAL_SECRET',
'grantTypes':['client_credentials']}
def test_absent_original_selector_refused_before_further_inputs(self):
self.assert_refused_before_further_inputs([dict(self.client(),clientId='unrelated')])
def test_identical_duplicate_selector_refused_before_further_inputs(self):
self.assert_refused_before_further_inputs([self.client(),self.client()])
def test_conflicting_duplicate_selector_refused_before_further_inputs(self):
self.assert_refused_before_further_inputs([self.client(),dict(self.client(),secretRef='env:OTHER')])
def test_empty_registry_refused_before_further_inputs(self):
self.assert_refused_before_further_inputs([])
class NativeTests(unittest.TestCase):
def setUp(self):
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
self.key=rsa.generate_private_key(public_exponent=65537,key_size=2048).private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption())
self.cfg=b'issuer: https://issuer.example\nclients:\n- clientId: existing\n clientType: confidential\n secretRef: env:KEYCAPE_ORIGINAL_SECRET\n grantTypes: [client_credentials]\n'
self.files={'/etc/keycape/config.yaml':self.cfg,'/etc/keycape/key.pem':self.key,'/var/lib/keycape-policy/policy.json':b'{}'}
self.payload={'env':{'KEYCAPE_ORIGINAL_SECRET':'fixture-private'},'files':{k:base64.b64encode(v).decode() for k,v in self.files.items()}}
self.admission={'issuer':'https://issuer.example','verifier_sha256':hashlib.sha256(b'binary').hexdigest(),
'trust_path':'/trust.pem','trust_sha256':hashlib.sha256(b'original-trust').hexdigest(),
'consumer':{'client_id':'existing','secret_env':'KEYCAPE_ORIGINAL_SECRET','scope':'read','audience':'consumer',
'subject':'service:existing','tenant':'tenant:platform','roles':'reader','deny_scope':'write'}}
self.session={'issuer':'isolated','volumes':{x:{'target':x} for x in ('/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls')}}
self.output=''.join('PASS '+x+'\n' for x in ['discovery and endpoint origin','exchange and JWKS signature for granted scopes','access-token claim bindings','principal_type is service','sub matches the registration','tenant matches the registration','roles present','granted scope carries no excess grant','predecessor secret refused','excess scope refused: write'])
def file(self,container,path):
return {**self.files,'/etc/hosts':b'127.0.0.1 issuer.example\n','/keycape':b'binary','/trust.pem':b'original-trust'}[path]
def command(self,argv,**kwargs):
if argv[1:3]==['container','inspect']:return json.dumps(['KEYCAPE_ORIGINAL_SECRET=fixture-private','SSL_CERT_FILE=/trust.pem']).encode()
output=io.BytesIO()
with tarfile.open(fileobj=output,mode='w') as t:
m=tarfile.TarInfo('policy.json');m.size=2;t.addfile(m,io.BytesIO(b'{}'))
return output.getvalue()
def verify(self,code=0,output=None):
with patch.object(a,'check_jwks'),patch.object(a,'runtime_file',return_value=b'original-trust'),patch.object(a,'runtime_files',side_effect=lambda session,paths=(),tree=False:({k:v for k,v in self.files.items() if k.startswith('/var/lib/keycape-policy/')} if tree else dict(self.files))),patch.object(a,'container_file',side_effect=self.file),patch.object(a,'command',side_effect=self.command),patch.object(a.subprocess,'run',return_value=subprocess.CompletedProcess([],code,b'',(self.output if output is None else output).encode())) as run:
result=a.run_verifier(self.session,self.admission,self.payload)
argv=run.call_args.args[0]
self.assertNotIn('fixture-private',' '.join(argv))
self.assertIn('-previous-secret-env',argv)
return result
def test_original_registration_and_unrelated_secret(self):
r=self.verify();self.assertTrue(r['wrong_unrelated_secret_explicitly_refused']);self.assertNotIn('fixture-private',json.dumps(r))
def test_transport_failure_never_counts_as_wrong_secret(self):
with self.assertRaises(a.Refused):self.verify(1,'FAIL discovery and endpoint origin: TLS failure\n')
def test_missing_explicit_rejection_never_counts(self):
with self.assertRaises(a.Refused):self.verify(0,self.output.replace('PASS predecessor secret refused\n',''))
def test_fail_line_refused_even_zero_exit(self):
with self.assertRaises(a.Refused):self.verify(0,self.output+'FAIL transport\n')
def test_security_environment_override_refused(self):
original=self.command
def altered(argv,**kwargs):
if argv[1:3]==['container','inspect']:return json.dumps(['KEYCAPE_ORIGINAL_SECRET=fixture-private','SSL_CERT_FILE=/trust.pem','SSL_CERT_DIR=/unreviewed']).encode()
return original(argv,**kwargs)
with patch.object(self,'command',side_effect=altered):
with self.assertRaises(a.Refused):self.verify()
def test_http_issuer_refused(self):
self.admission['issuer']='http://issuer.example'
with self.assertRaises(a.Refused):self.verify()
def test_changed_trust_refused(self):
self.admission['trust_sha256']='0'*64
with self.assertRaises(a.Refused):self.verify()
def test_config_drift_refused(self):
self.payload['files']['/etc/keycape/config.yaml']=base64.b64encode(b'changed').decode()
with self.assertRaises(a.Refused):self.verify()
class PublicReadinessTransportTests(unittest.TestCase):
def test_provider_probe_failure_is_closed_false_data_not_failed_docker_exec(self):
output=io.StringIO()
with patch('sys.stdin',io.StringIO('{}')),patch('sys.stdout',output):
exec(compile(a.JWKS_TRANSPORT,'<public-readiness>','exec',optimize=2),{})
self.assertEqual(json.loads(output.getvalue()),{'original_public_signing_key_in_live_jwks':False})
class BoundaryTests(unittest.TestCase):
def test_remote_docker_refused_before_info(self):
session={'run_id':'a'*32,'containers':['nk49-kc-'+('a'*32)+'-worker'],'worker':'nk49-kc-'+('a'*32)+'-worker','issuer':'nk49-kc-'+('a'*32)+'-worker'}
with patch.object(a,'command',return_value=b'[{"Endpoints":{"docker":{"Host":"tcp://remote:2375"}}}]') as command:
with self.assertRaises(a.Refused):a.inspect_session(session)
self.assertEqual(command.call_count,1)
def test_failed_verifier_cleanup_is_attempted(self):
receipt={};admission={'session':{},'ciphertext_sha256':'a'*64,'manifest_sha256':'b'*64,'verifier_sha256':'c'*64,'trust_sha256':'d'*64,'consumer':{x:'yes' for x in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope')}}
with patch.object(a,'inspect_session'),patch.object(a,'private_payload',return_value={}),patch.object(a,'run_verifier',side_effect=a.Refused),patch.object(a,'cleanup') as cleanup:
with self.assertRaises(a.Refused):a.native_checks(Path('/unused'),{},admission,receipt)
cleanup.assert_called_once();self.assertTrue(receipt['container_cleanup_verified']);self.assertEqual(receipt['status'],'failed')
def test_cleanup_failure_overrides_success(self):
admission={'session':{},'ciphertext_sha256':'a'*64,'manifest_sha256':'b'*64,'verifier_sha256':'c'*64,'trust_sha256':'d'*64,'consumer':{x:'yes' for x in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope')}}
result={x:True for x in ('native_service_registration_verified','wrong_unrelated_secret_explicitly_refused','denied_scopes_refused','original_config_registry_policy_receipts_equal','original_https_issuer_and_trust_preserved','original_public_signing_key_in_live_jwks')}
result['public_signing_spki_sha256']='e'*64;receipt={}
with patch.object(a,'inspect_session'),patch.object(a,'private_payload',return_value={}),patch.object(a,'run_verifier',return_value=result),patch.object(a,'cleanup',side_effect=a.Refused):
with self.assertRaises(a.Refused):a.native_checks(Path('/unused'),{},admission,receipt)
self.assertEqual(receipt['status'],'failed');self.assertFalse(receipt['container_cleanup_verified'])
class ContainmentTests(unittest.TestCase):
def inspect(self,overrides=None,launch=None,mounts=None):
name='nk49-kc-'+('a'*32)+'-issuer'
session={'run_id':'a'*32,'containers':[name],'issuer':name,'worker':name,'images':{name:a.IMAGE}}
host={'NetworkMode':'none','ReadonlyRootfs':True,'Privileged':False,'CapDrop':['ALL'],'SecurityOpt':['no-new-privileges']}
host.update(overrides or {})
def command(argv,**kwargs):
if argv[1:3]==['context','inspect']:return b'[{"Endpoints":{"docker":{"Host":"unix:///var/run/docker.sock"}}}]'
if argv[1]=='info':return b'{"ID":"74f93783-6cef-43a8-a3c4-f62586cd3bc1","Name":"bnt-lap001","OSType":"linux"}'
if argv[-1]=='{{.Id}}':return b'anchor-id'
if '"path"' in argv[-1]:return json.dumps(launch or {'entrypoint':['/keycape'],'cmd':None,'path':'/keycape','args':[]}).encode()
return json.dumps({'host':host,'mounts':mounts or [],'labels':{a.LABEL:'a'*32},'image':a.IMAGE,'user':'1000:1000'}).encode()
with patch.object(a,'command',side_effect=command),patch.dict('os.environ',{'DOCKER_HOST':'unix:///var/run/docker.sock'}):a.inspect_session(session)
def test_minimal_network_none_admitted(self):self.inspect()
def test_image_declared_anonymous_disk_volume_remains_refused(self):
with self.assertRaises(a.Refused):
self.inspect(mounts=[{'Type':'volume','Name':'anonymous-fixture','Destination':'/privacyidea/etc/persistent','RW':True}])
def test_alternate_same_image_command_refused(self):
with self.assertRaises(a.Refused):self.inspect(launch={'entrypoint':['/keycape'],'cmd':['check-config'],'path':'/keycape','args':['check-config']})
def test_cap_add_refused(self):
with self.assertRaises(a.Refused):self.inspect({'CapAdd':['NET_ADMIN']})
def test_devices_refused(self):
with self.assertRaises(a.Refused):self.inspect({'Devices':[{'PathOnHost':'/dev/mem'}]})
def test_device_requests_refused(self):
with self.assertRaises(a.Refused):self.inspect({'DeviceRequests':[{'Driver':'nvidia'}]})
def test_host_namespace_refused(self):
for field in ('PidMode','IpcMode','UTSMode','UsernsMode','CgroupnsMode'):
with self.subTest(field=field),self.assertRaises(a.Refused):self.inspect({field:'host'})
def test_security_override_refused(self):
with self.assertRaises(a.Refused):self.inspect({'SecurityOpt':['no-new-privileges','seccomp=unconfined']})
class CleanupTests(unittest.TestCase):
def test_partial_cleanup_attempts_every_owned_container(self):
session={'run_id':'a'*32,'containers':['first','second','third']};receipt={}
removed=[]
def command(argv,**kwargs):
if argv[1:3]==['container','inspect']:return ('a'*32).encode()
if argv[1]=='rm':
removed.append(argv[-1])
if argv[-1]=='third':raise a.Refused()
return b''
if argv[1:3]==['container','ls']:
name=next(x for x in argv if x.startswith('name=^/')).removeprefix('name=^/').removesuffix('$')
return (name+'\n').encode() if name=='third' or name not in removed else b''
return b''
with patch.object(a,'command',side_effect=command):
with self.assertRaises(a.Refused):a.cleanup(session,receipt)
self.assertEqual(removed,['third','second','first'])
self.assertEqual([r['status'] for r in receipt['container_cleanup_obligations']],['still_present','absence_verified','absence_verified'])
def test_ownership_mismatch_does_not_remove_and_continues(self):
session={'run_id':'a'*32,'containers':['first','second']};receipt={};removed=[]
def command(argv,**kwargs):
if argv[1:3]==['container','inspect']:return (('b' if argv[3]=='second' else 'a')*32).encode()
if argv[1]=='rm':removed.append(argv[-1])
if argv[1:3]==['container','ls']:
name=next(x for x in argv if x.startswith('name=^/')).removeprefix('name=^/').removesuffix('$')
return (name+'\n').encode() if name not in removed else b''
return b''
with patch.object(a,'command',side_effect=command):
with self.assertRaises(a.Refused):a.cleanup(session,receipt)
self.assertEqual(removed,['first'])
self.assertEqual(receipt['container_cleanup_obligations'][0]['status'],'ownership_mismatch_not_removed')
def test_inspect_failure_retains_uncertain_obligation(self):
receipt={}
with patch.object(a,'command',side_effect=a.Refused):
with self.assertRaises(a.Refused):a.cleanup({'run_id':'a'*32,'containers':['exact-owned-name']},receipt)
self.assertEqual(receipt['container_cleanup_obligations'],[{'container':'exact-owned-name','status':'unverified'}])
class JWKSTests(unittest.TestCase):
def execute(self,n='123',endpoint='https://issuer.example/jwks',resolved='127.0.0.1',extra=False,kid='key-1'):
request={'issuer':'https://issuer.example','trust':'public-certificate','n':'123','e':'65537'}
def enc(i):return base64.urlsafe_b64encode(int(i).to_bytes((int(i).bit_length()+7)//8,'big')).decode().rstrip('=')
responses=[{'issuer':'https://issuer.example','jwks_uri':endpoint},{'keys':[{'kty':'RSA','kid':kid,'use':'sig','alg':'RS256','n':enc(n),'e':enc(65537)}]}]
if extra:responses[1]['keys'].append(dict(responses[1]['keys'][0],kid='key-2'))
class Response:
def __init__(self,value):self.value=value
def __enter__(self):return self
def __exit__(self,*args):pass
def read(self,*args):return json.dumps(self.value).encode()
class Opener:
def open(self,*args,**kwargs):return Response(responses.pop(0))
with patch('sys.stdin',io.StringIO(json.dumps(request))),patch('sys.stdout',io.StringIO()),patch('socket.getaddrinfo',return_value=[(2,1,6,'',(resolved,443))]),patch('ssl.create_default_context'),patch('urllib.request.build_opener',return_value=Opener()):
exec(compile(a.JWKS_PROBE,'<jwks-probe>','exec',optimize=2),{})
def test_exact_original_public_key(self):self.execute()
def test_extra_signing_key_refused_under_optimization(self):
with self.assertRaises(ValueError):self.execute(extra=True)
def test_wrong_kid_refused_under_optimization(self):
with self.assertRaises(ValueError):self.execute(kid='replacement')
def test_wrong_live_signing_key_refused(self):
with self.assertRaises(ValueError):self.execute(n='124')
def test_cross_origin_jwks_refused(self):
with self.assertRaises(ValueError):self.execute(endpoint='https://production.example/jwks')
def test_production_dns_refused_before_fetch(self):
with self.assertRaises(ValueError):self.execute(resolved='192.0.2.1')
if __name__=='__main__':unittest.main()

487
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/postimages/tools/bootstrap-recovery/isolated-keycape.py Normal file
View file

@ -0,0 +1,487 @@
#!/usr/bin/env python3
"""T18 native adapter for an owner-admitted, already restored isolated session.
Called after identity-bundle.restore, never with an unvalidated directory. This
module does not provision TLS, grant credentials, or claim downstream acceptance.
The existing decrypt/tmpfs wrapper remains responsible for archive cleanup.
"""
import base64
import hashlib
import importlib.util
import json
import io
import os
import secrets
import tarfile
import time
import re
import subprocess
from pathlib import Path
HERE = Path(__file__).resolve().parent
IMAGE = 'forgejo.coulomb.social/coulomb/key-cape@sha256:cdec9dd7c515f7e0a6169fc3fa560ee2cc47bf40d899755dec808c362f5ce010'
LABEL = 'net-kingdom.recovery-run'
class Refused(Exception):
pass
def load_bundle():
spec = importlib.util.spec_from_file_location('identity_bundle', HERE / 'identity-bundle.py')
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def manifest_binding_mode(admission,transfer):
"""Versioned private-manifest binding; legacy explicit digests stay valid."""
mode=admission.get('manifest_binding_mode','explicit-digest-v1')
if mode not in ('explicit-digest-v1','ciphertext-derived-v1'):raise Refused()
if (not isinstance(admission.get('ciphertext_sha256'),str)
or not re.fullmatch('[a-f0-9]{64}',admission['ciphertext_sha256'])
or admission['ciphertext_sha256']!=transfer.get('ciphertext_sha256')
or transfer.get('independent_owner_retrieval_verified') is not True):raise Refused()
if mode=='explicit-digest-v1':
if not isinstance(admission.get('manifest_sha256'),str) or not re.fullmatch('[a-f0-9]{64}',admission['manifest_sha256']):raise Refused()
elif 'manifest_sha256' in admission:raise Refused()
return mode
def private_manifest_admission(admission,manifest):
# Only called after the fixed ciphertext bytes were hashed, authenticated
# decryption succeeded and bundle.restore checked the complete archive.
# Never return this enriched private object in a safe receipt/checkpoint.
digest=hashlib.sha256(json.dumps(manifest,sort_keys=True,separators=(',',':')).encode()).hexdigest()
mode=admission.get('manifest_binding_mode','explicit-digest-v1')
if mode=='explicit-digest-v1':
if digest!=admission['manifest_sha256']:raise Refused()
elif mode!='ciphertext-derived-v1' or 'manifest_sha256' in admission:raise Refused()
return {**admission,'manifest_sha256':digest}
def private_payload(root, manifest, expected_manifest_sha256):
"""Bind the complete validated archive before selecting KeyCape material.
Private hashes never enter the public receipt. The digest is supplied by the
existing private archive admission, not inferred from an untrusted manifest.
"""
encoded = json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode()
if hashlib.sha256(encoded).hexdigest() != expected_manifest_sha256:
raise Refused()
entries = manifest['files']
if len(entries) != len({e['path'] for e in entries}):
raise Refused()
values = {}
bundle = load_bundle()
for entry in entries:
if not bundle.allowed(entry['path'], manifest['profile']):
raise Refused()
path = root / entry['path']
bundle.private(path)
value = path.read_bytes()
if len(value) != entry['bytes'] or hashlib.sha256(value).hexdigest() != entry['sha256']:
raise Refused()
values[entry['path']] = value
deployment = json.loads(values['resources/sso/deployment/keycape.json'])
main = next(c for c in deployment['spec']['template']['spec']['containers'] if c['name'] == 'keycape')
if main.get('command') or main.get('args') or main.get('envFrom'):
raise Refused()
if main['image'] != IMAGE:
# Mutable-tag/historical mismatches need a separate explicit owner admission.
raise Refused()
secret = json.loads(values['resources/sso/secret/keycape-config.json'])['data']
files = {'/etc/keycape/' + name: base64.b64decode(secret[name], validate=True)
for name in ('config.yaml', 'key.pem')}
for path, value in values.items():
if path.startswith('data/keycape/'):
relative = path.removeprefix('data/keycape/')
files['/var/lib/keycape-policy/' + relative] = value
if '/var/lib/keycape-policy/policy.json' not in files:
raise Refused()
if 'resources/sso/secret/keycape-factor-read.json' in values:
factor=json.loads(values['resources/sso/secret/keycape-factor-read.json'])['data']
files['/etc/keycape-factor/admin-token']=base64.b64decode(factor['admin-token'],validate=True)
# Resolve original delivered environment, including all referenced secrets.
env = {}
for item in main.get('env', []):
if not re.fullmatch(r'KEYCAPE_[A-Z0-9_]+',item['name']):raise Refused()
if item['name'] in env:
raise Refused()
if 'value' in item:
env[item['name']] = item['value']
else:
ref = item['valueFrom']['secretKeyRef']
source = json.loads(values['resources/sso/secret/' + ref['name'] + '.json'])['data']
env[item['name']] = base64.b64decode(source[ref['key']], validate=True).decode()
if env.get('KEYCAPE_CONFIG') != '/etc/keycape/config.yaml' or env.get('KEYCAPE_POLICY_PATH') != '/var/lib/keycape-policy/policy.json':
raise Refused()
return {'files': {p: base64.b64encode(v).decode() for p, v in files.items()}, 'env': env}
ENDPOINT = 'unix:///var/run/docker.sock'
EXECUTION = None
def docker_environment(environment=None):
# Even an admitted daemon exposed by another Unix socket is outside this
# exact session binding. Conflicting ambient routing is refused, not ignored.
for source in (os.environ, environment or {}):
if source.get('DOCKER_HOST') not in (None,'',ENDPOINT):raise Refused()
if source.get('DOCKER_CONTEXT') not in (None,'','default'):raise Refused()
if any(source.get(k) for k in ('DOCKER_TLS_VERIFY','DOCKER_CERT_PATH')):raise Refused()
env=dict(os.environ if environment is None else environment)
for key in ('DOCKER_CONTEXT','DOCKER_TLS_VERIFY','DOCKER_CERT_PATH'):env.pop(key,None)
env['DOCKER_HOST']=ENDPOINT
return env
def process(argv, **kwargs):
marker=None
if EXECUTION is not None and (kwargs.get('start_new_session') or kwargs.get('process_group') not in (None,-1)):raise Refused()
if argv[0]=='docker':
argv=['docker','--host',ENDPOINT,*argv[1:]]
kwargs['env']=docker_environment(kwargs.get('env'))
if EXECUTION is not None:
if (os.getpid()!=EXECUTION['pid'] or os.getsid(0)!=EXECUTION['sid'] or os.getpgrp()!=EXECUTION['pgid']
or time.monotonic()>=EXECUTION['deadline']):raise Refused()
marker=Path(EXECUTION['staging']+'-docker-inflight')
fd=os.open(marker,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
os.close(fd)
try:
result=subprocess.run(argv,**kwargs)
if marker is not None and result.returncode!=0:
fd=os.open(Path(EXECUTION['staging']+'-docker-uncertain'),os.O_WRONLY|os.O_CREAT|os.O_NOFOLLOW,0o600);os.close(fd)
return result
except BaseException:
# Parent loss leaves .docker-inflight; caught timeout/transport failures
# retain a sticky uncertainty marker even if later cleanup can run.
if marker is not None:
fd=os.open(Path(EXECUTION['staging']+'-docker-uncertain'),os.O_WRONLY|os.O_CREAT|os.O_NOFOLLOW,0o600);os.close(fd)
raise
finally:
if marker is not None:
try:marker.unlink()
except FileNotFoundError:pass
def command(argv, **kwargs):
maximum = kwargs.pop('max_output', 32768)
result = process(argv, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
timeout=kwargs.pop('timeout', 180), **kwargs)
if result.returncode or len(result.stdout) > maximum:
raise Refused()
return result.stdout
def inspect_session(session, worker_only=False):
"""No credential environment is read by Docker inspect or returned to caller."""
if not re.fullmatch(r'[a-f0-9]{32}', session['run_id']):
raise Refused()
names = session['containers']
if not names or len(names) > 12 or len(set(names)) != len(names):
raise Refused()
for name in names:
if not re.fullmatch(r'nk49-kc-' + session['run_id'] + r'-[a-z0-9-]{1,24}', name):
raise Refused()
if session['worker'] not in names or session['issuer'] not in names:
raise Refused()
import os
contexts=json.loads(command(['docker','context','inspect']))
endpoints=[contexts[0]['Endpoints']['docker']['Host'],os.environ.get('DOCKER_HOST') or ENDPOINT]
if not all(x==ENDPOINT for x in endpoints):raise Refused()
info = json.loads(command(['docker', 'info', '--format', '{"ID":{{json .ID}},"Name":{{json .Name}},"OSType":{{json .OSType}}}']))
if info != {'ID': '74f93783-6cef-43a8-a3c4-f62586cd3bc1', 'Name': 'bnt-lap001', 'OSType': 'linux'}:
raise Refused()
# The worker is the anchor; all dependencies/TLS/issuer share its network-none namespace.
anchor = command(['docker', 'container', 'inspect', session['worker'], '--format', '{{.Id}}']).decode().strip()
for name,spec in session.get('volumes',{}).items():
volume=json.loads(command(['docker','volume','inspect',name]))
expected={'type':'tmpfs','device':'tmpfs','o':'size=33554432,uid=1000,gid=1000,mode=0700,nosuid,nodev,noexec'}
if len(volume)!=1 or volume[0]['Driver']!='local' or volume[0].get('Scope')!='local' or volume[0].get('Options')!=expected or spec['options']!=expected or volume[0].get('Labels',{}).get(LABEL)!=session['run_id']:raise Refused()
for name in names:
obj = json.loads(command(['docker', 'container', 'inspect', name, '--format',
'{"host":{{json .HostConfig}},"mounts":{{json .Mounts}},"labels":{{json .Config.Labels}},"image":{{json .Config.Image}},"user":{{json .Config.User}},"entrypoint":{{json .Config.Entrypoint}},"cmd":{{json .Config.Cmd}}}']))
host = obj['host']
mounts=obj['mounts']
destinations=[m['Destination'] for m in mounts]
if len(destinations)!=len(set(destinations)):raise Refused()
for volume,expected_spec in session.get('volumes',{}).items():
matches=[m for m in mounts if m['Destination']==expected_spec['target']]
if len(matches)!=1 or matches[0]['Type']!='volume' or matches[0]['Name']!=volume or matches[0]['RW']!=(name==session['worker'] or expected_spec['target']=='/var/lib/keycape-policy'):raise Refused()
if (obj['labels'].get(LABEL) != session['run_id'] or not obj['user'] or obj['user'].split(':')[0] in ('0', 'root')
or not host['ReadonlyRootfs'] or host['Privileged'] or host.get('Binds') or host.get('PortBindings')
or host.get('PidMode') or host.get('IpcMode') == 'host'
or host.get('UTSMode') or host.get('UsernsMode') == 'host' or host.get('CgroupnsMode') == 'host'
or host.get('Sysctls',{}) not in ({},{'net.ipv4.ip_unprivileged_port_start':'0'} if name==session['worker'] else {})
or host.get('CapAdd') or host.get('Devices') or host.get('DeviceRequests') or host.get('VolumesFrom')
or host.get('CapDrop') != ['ALL']
or host.get('SecurityOpt') not in (['no-new-privileges'], ['no-new-privileges:true'])
or any(m['Type']!='tmpfs' and not (m['Type']=='volume' and m['Name'] in session.get('volumes',{}) and m['Destination']==session['volumes'][m['Name']]['target']) for m in obj['mounts'])
or host['NetworkMode'] != ('none' if name == session['worker'] else 'container:' + anchor)
or obj['image'] != session['images'][name] or '@sha256:' not in obj['image']):
raise Refused()
if worker_only:return
if session['images'][session['issuer']] != IMAGE:
raise Refused()
# Known admitted archive has no command/args override; image default /keycape
# must be the actual issuer executable, not an alternate same-image command.
launch=json.loads(command(['docker','container','inspect',session['issuer'],'--format',
'{"entrypoint":{{json .Config.Entrypoint}},"cmd":{{json .Config.Cmd}},"path":{{json .Path}},"args":{{json .Args}}}']))
if launch != {'entrypoint':['/keycape'],'cmd':None,'path':'/keycape','args':[]}:
raise Refused()
def container_file(container, path):
# Docker streams one tar member into memory; never materialize private files.
raw = command(['docker', 'cp', container + ':' + path, '-'], max_output=128*1024*1024)
with tarfile.open(fileobj=io.BytesIO(raw), mode='r:') as archive:
members = archive.getmembers()
if len(members) != 1 or not members[0].isfile() or members[0].size > 96*1024*1024:
raise Refused()
return archive.extractfile(members[0]).read()
JWKS_PROBE = r"""
import base64,ipaddress,json,socket,ssl,sys,urllib.request
from urllib.parse import urlsplit
p=json.load(sys.stdin);issuer=p['issuer'];origin=urlsplit(issuer)
def require(condition):
if not condition:raise ValueError('public_jwks_binding_refused')
require(origin.scheme=='https')
resolved=socket.getaddrinfo(origin.hostname,origin.port or 443,type=socket.SOCK_STREAM)
require(bool(resolved) and all(ipaddress.ip_address(x[4][0]).is_loopback for x in resolved))
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self,*args,**kwargs):return None
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect(),
urllib.request.HTTPSHandler(context=ssl.create_default_context(cadata=p['trust'])))
def get(url):
u=urlsplit(url)
require((u.scheme,u.hostname,u.port or 443)==(origin.scheme,origin.hostname,origin.port or 443))
require(not u.username and not u.password and not u.fragment)
with opener.open(url,timeout=10) as response:
raw=response.read(1024*1024+1);require(len(raw)<=1024*1024)
return json.loads(raw)
d=get(issuer.rstrip('/')+'/.well-known/openid-configuration');require(d['issuer']==issuer)
keys=get(d['jwks_uri'])['keys']
def integer(value):return int.from_bytes(base64.urlsafe_b64decode(value+'='*(-len(value)%4)),'big')
# The exact admitted owner release constructs one key, kid key-1. Reject every
# extra key, including irrelevant keys: a native successful exchange can then
# only have used the original signer through the existing verifier.
require(isinstance(keys,list) and len(keys)==1)
k=keys[0]
require(k.get('kty')=='RSA' and k.get('kid')=='key-1' and k.get('use')=='sig' and k.get('alg')=='RS256')
require(integer(k['n'])==int(p['n']) and integer(k['e'])==int(p['e']))
print('{"original_public_signing_key_in_live_jwks":true}')
"""
# A failed public readiness probe is closed data, not a failing Docker RPC.
# Transport/CLI nonzero remains uncertain; no deliberately failing exec is
# retried as if daemon acceptance had been established.
JWKS_TRANSPORT = ("import contextlib,io\ntry:\n"
" with contextlib.redirect_stdout(io.StringIO()):\n"
" exec(" + repr(JWKS_PROBE) + ")\n"
"except Exception:\n print('{\"original_public_signing_key_in_live_jwks\":false}')\n"
"else:\n print('{\"original_public_signing_key_in_live_jwks\":true}')\n")
def check_jwks(session, admission, public_numbers):
trust=runtime_file(session,admission['trust_path']).decode()
request={'issuer':admission['issuer'],'trust':trust,'n':str(public_numbers.n),'e':str(public_numbers.e)}
result=json.loads(command(['docker','exec','-i',session['worker'],'python3','-c',JWKS_TRANSPORT],input=json.dumps(request).encode()))
if result != {'original_public_signing_key_in_live_jwks': True}:raise Refused()
RUNTIME_FILES = r"""
import base64,json,sys
from pathlib import Path
p=json.load(sys.stdin)
roots=('/etc/keycape/','/var/lib/keycape-policy/','/etc/keycape-factor/','/recovery-tls/')
result={}
paths=[]
if p.get('tree'):
paths=list(Path('/var/lib/keycape-policy').rglob('*'))
else:paths=[Path(x) for x in p['paths']]
for f in paths:
if f.is_dir():continue
if not str(f).startswith(roots) or not f.is_file() or any(x.is_symlink() for x in (f,*f.parents)) or f.stat().st_size>8*1024*1024:raise ValueError()
result[str(f)]=base64.b64encode(f.read_bytes()).decode()
if len(result)>4096:raise ValueError()
print(json.dumps(result))
"""
def runtime_files(session, paths=(), tree=False):
raw=command(['docker','exec','-i',session['worker'],'python3','-c',RUNTIME_FILES],
input=json.dumps({'paths':list(paths),'tree':tree}).encode(),max_output=48*1024*1024)
return {k:base64.b64decode(v,validate=True) for k,v in json.loads(raw).items()}
def runtime_file(session,path):
return runtime_files(session,[path])[path]
def run_verifier(session, admission, payload):
# Select from the already-restored config before further private/runtime reads.
# This is not public preflight before archive decryption or its key prompt.
import yaml
cfg = yaml.safe_load(base64.b64decode(payload['files']['/etc/keycape/config.yaml'],validate=True))
if not isinstance(cfg,dict) or cfg.get('issuer') != admission['issuer']:
raise Refused()
consumer = admission['consumer']
clients = cfg.get('clients')
if not isinstance(clients,list) or any(not isinstance(client,dict) for client in clients):
raise Refused()
matches = [client for client in clients if client.get('clientId') == consumer['client_id']]
if len(matches) != 1:
raise Refused()
client = matches[0]
if (client.get('secretRef') != 'env:'+consumer['secret_env'] or client.get('clientType') != 'confidential'
or 'client_credentials' not in client.get('grantTypes',[])):
raise Refused()
issuer = session['issuer']
if {v['target'] for v in session.get('volumes',{}).values()}!={'/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls'}:raise Refused()
from urllib.parse import urlsplit
import ipaddress
address = urlsplit(admission['issuer'])
if (address.scheme != 'https' or not address.hostname or address.username or address.password
or address.query or address.fragment):
raise Refused()
if hashlib.sha256(container_file(issuer, '/keycape')).hexdigest() != admission['verifier_sha256']:
raise Refused()
if hashlib.sha256(runtime_file(session, admission['trust_path'])).hexdigest() != admission['trust_sha256']:
raise Refused()
# Capture private environment only in owner process memory, never receipt.
environment = json.loads(command(['docker','container','inspect',issuer,'--format','{{json .Config.Env}}']))
env = dict(item.split('=',1) for item in environment)
if len(env)!=len(environment):raise Refused()
if set(env)-set(payload['env'])-{'PATH','HOME','HOSTNAME','SSL_CERT_FILE'}:raise Refused()
if env.get('SSL_CERT_FILE') != admission['trust_path'] or any(k.lower().endswith('_proxy') and v for k,v in env.items()):
raise Refused()
if {k:v for k,v in env.items() if k.startswith('KEYCAPE_')}!={k:v for k,v in payload['env'].items() if k.startswith('KEYCAPE_')}:raise Refused()
for name,value in payload['env'].items():
if env.get(name) != value:
raise Refused()
files = {k:base64.b64decode(v,validate=True) for k,v in payload['files'].items()}
def equal():
if runtime_files(session,files)!=files:raise Refused()
durable=runtime_files(session,tree=True)
if durable!={k:v for k,v in files.items() if k.startswith('/var/lib/keycape-policy/')}:
raise Refused()
equal()
if not env.get(consumer['secret_env']):
raise Refused()
from cryptography.hazmat.primitives import serialization
key=serialization.load_pem_private_key(files['/etc/keycape/key.pem'],password=None)
check_jwks(session,admission,key.public_key().public_numbers())
wrong = secrets.token_urlsafe(48)
if wrong == env[consumer['secret_env']]:raise Refused()
args = ['docker','exec','--env','NK49_UNRELATED_NEGATIVE_SECRET='+wrong,issuer,'/keycape','verify-client',
'-issuer',admission['issuer'],'-previous-secret-env','NK49_UNRELATED_NEGATIVE_SECRET']
flags = {'client_id':'client-id','secret_env':'secret-env','audience':'audience','scope':'scope',
'subject':'expect-subject','tenant':'expect-tenant','roles':'expect-roles','deny_scope':'deny-scope'}
for name,flag in flags.items():args+=['-'+flag,consumer[name]]
# Native command reports on stderr. Do not return or print its raw response.
result = process(args,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=150)
raw = (result.stdout+result.stderr).decode()
required = ['discovery and endpoint origin','exchange and JWKS signature for granted scopes','access-token claim bindings',
'principal_type is service','sub matches the registration','tenant matches the registration','roles present',
'granted scope carries no excess grant','predecessor secret refused']
required += ['excess scope refused: '+scope for scope in consumer['deny_scope'].split()]
if result.returncode or any('PASS '+name+'\n' not in raw for name in required) or 'FAIL ' in raw:
raise Refused()
equal()
from cryptography.hazmat.primitives import serialization
key=serialization.load_pem_private_key(files['/etc/keycape/key.pem'],password=None)
check_jwks(session,admission,key.public_key().public_numbers())
public=key.public_key().public_bytes(serialization.Encoding.DER,serialization.PublicFormat.SubjectPublicKeyInfo)
return {'native_service_registration_verified':True,'wrong_unrelated_secret_explicitly_refused':True,
'denied_scopes_refused':True,'original_config_registry_policy_receipts_equal':True,
'public_signing_spki_sha256':hashlib.sha256(public).hexdigest(),
'original_https_issuer_and_trust_preserved':True,'original_public_signing_key_in_live_jwks':True}
def cleanup(session, receipt=None):
"""Attempt every owned cleanup even when one obligation fails.
Preserve exact safe container names and outcomes so the owner can recover an
uncertain/partial cleanup. An inspect/daemon failure never means absence.
"""
obligations = []
failed = False
for name in reversed(session['containers']):
row = {'container': name, 'status': 'unverified'}
obligations.append(row)
try:
present=command(['docker','container','ls','--all','--filter','name=^/'+name+'$','--format','{{.Names}}']).decode().splitlines()
if not present:
row['status']='absence_verified'
continue
label = command(['docker', 'container', 'inspect', name, '--format',
'{{index .Config.Labels "' + LABEL + '"}}']).decode().strip()
if label != session['run_id']:
row['status'] = 'ownership_mismatch_not_removed'
failed = True
continue
# A failed remove is followed by a successful explicit inventory;
# interrupted remove may actually have completed.
try:
command(['docker', 'rm', '--force', '--volumes', name])
except Exception:
pass
remaining = command(['docker','container','ls','--all','--filter',
'name=^/' + name + '$','--format','{{.Names}}']).decode().splitlines()
if remaining:
row['status'] = 'still_present'
failed = True
else:
row['status'] = 'absence_verified'
except Exception:
failed = True
if receipt is not None:
receipt['container_cleanup_obligations'] = obligations
if failed:
raise Refused()
def native_checks(root, manifest, admission, receipt, cleanup_session=True):
"""Owner wrapper provides a reviewed session after existing restore pipeline.
Admission must bind ciphertext, canonical private manifest digest, executable
and trust bytes, exact image/target list, original issuer and selected consumer.
None of these admission fields grant authority or carry a client secret.
"""
session = admission['session']
# Validate ownership/containment before permitting either exec or cleanup.
inspect_session(session)
receipt.update(status='failed', source_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
ciphertext_sha256=admission['ciphertext_sha256'], production_contacted=False,
human_mfa_proven=False, downstream_consumer_reconnect_proven=False,
container_cleanup_verified=False)
try:
for field in ('ciphertext_sha256', 'manifest_sha256', 'verifier_sha256', 'trust_sha256'):
if not re.fullmatch('[a-f0-9]{64}', admission[field]):
raise Refused()
for field in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope'):
if not isinstance(admission['consumer'].get(field), str) or not admission['consumer'][field].strip():
raise Refused()
payload = private_payload(root, manifest, admission['manifest_sha256'])
result = run_verifier(session, admission, payload)
required = {'native_service_registration_verified','wrong_unrelated_secret_explicitly_refused',
'denied_scopes_refused','original_config_registry_policy_receipts_equal',
'original_https_issuer_and_trust_preserved','original_public_signing_key_in_live_jwks'}
if any(result.get(k) is not True for k in required) or not re.fullmatch('[a-f0-9]{64}', result.get('public_signing_spki_sha256','')):
raise Refused()
# Closed receipt allowlist: never copy arbitrary worker output.
receipt.update({k: True for k in required})
receipt['public_signing_spki_sha256'] = result['public_signing_spki_sha256']
receipt['status'] = 'isolated_keycape_native_registration_checked'
finally:
try:
if cleanup_session:
cleanup(session, receipt)
receipt['container_cleanup_verified'] = True
except Exception:
receipt['status'] = 'failed'
raise

1271
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/postimages/workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md Normal file
View file

File diff suppressed because it is too large Load diff

124
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/preimages/tests/isolated-keycape-native-fixture-manifest.json Normal file
View file

@ -0,0 +1,124 @@
{
"schema": "nk49.keycape-native-fixture-proposal.v1",
"status": "prepared_not_executed_not_approved",
"source_sha256": {
"sso-mfa/bootstrap/sops-private.py": "sha256:acac68e2936ed38b7d7c9e35eb4619647c99c117a3928a1c5ce704b570906780",
"sso-mfa/k8s/privacyidea/key-custody-remote.py": "sha256:27b2b885c81923d85054e3999c5613e104fe627779b2b50398ed9bed5088f47e",
"tests/run_isolated_keycape_native_fixture.py": "sha256:ecbadecc475622eaaf20a726110aebef5f4fe7af141110a98aca1640179b4a03",
"tests/test_identity_bundle.py": "sha256:be5ce74097594cc7f7e77dbccd6cb53aa72f263cd1bb442ce5464038aed1116c",
"tests/test_isolated_keycape_fixture_plan.py": "sha256:5945619f4c5957187763276cc900870a9f08f9475b63c240694688e0c2eca7dd",
"tests/test_isolated_keycape_recovery.py": "sha256:247e9a87f5be2ee9f20f7edbb4a9f612ab7cb0cd940392c3caeca8b046605f55",
"tests/test_isolated_keycape_session.py": "sha256:cbdf51efde26eff337e9303a2c431f80b333b8846d763eb62853f81eb3f13e2c",
"tools/bootstrap-recovery/capture-identity.py": "sha256:47dca7c4dc96fa7eb9fb02c81fca9ca8dc5281aac918979eb7367148168ba705",
"tools/bootstrap-recovery/check-privacyidea-database.py": "sha256:90f454c87998d7f5d625e3c6656c3418f8f56275dc1879a9d19ed605431b94a4",
"tools/bootstrap-recovery/fetch-privacyidea-backup.py": "sha256:233160e1206643a5cf7ad05141d733f0a75aff651190e27f77515a124ff0a59f",
"tools/bootstrap-recovery/identity-backup-profile.json": "sha256:40f9e5b5bb957e5b23e03b31701702b850502d16a92a0fa0ebf92618cc3f18a1",
"tools/bootstrap-recovery/identity-bundle.py": "sha256:524a03c065db4307fcde83d33af39503f3de19cb6ef3a8ca72c991d4a8c08ee6",
"tools/bootstrap-recovery/identity-credential-setup.py": "sha256:ddd6127fa331c0185523ec3bb22d743a791edd56a2e6af7ad8d7949b7c051141",
"tools/bootstrap-recovery/identity-custody.json": "sha256:f8aee8487ed01d7940b82676bb4a5c8d61f51b6c598521b8b8a73588aaea4d4f",
"tools/bootstrap-recovery/identity-files.py": "sha256:d0063a9eedd3610e3d45bd2d834aae83cff74e4b3504d138b3548f427e89331d",
"tools/bootstrap-recovery/identity-isolation.py": "sha256:88d1e13112a44c79571825af89d3fe5b01fa490cff2653badef4c62a45c844f1",
"tools/bootstrap-recovery/identity-offsite.py": "sha256:e7c31fa9ee081961af873348ee3b4829a645d40c7dc7f1e22c564aa9ed01bc49",
"tools/bootstrap-recovery/identity-retention.py": "sha256:a41ea4759f69c35548149fc8d0d5ace547f16cf9fe4c9dac10b119a4d0becdec",
"tools/bootstrap-recovery/identity-runtime.py": "sha256:a2c0acf567c204705b90be64d561d0fda2285e0fc71a962d6e1b149f571e1940",
"tools/bootstrap-recovery/install-identity-backup.py": "sha256:5c612a2a51c9b59aad8cc01d418c2f563cbd92c62dc1fa8be994e5fb59df90f1",
"tools/bootstrap-recovery/inventory-scaleway.py": "sha256:a219df205708fa4ed44b763a418585d63c820c25434bc0fd14e805eb5c2f79d3",
"tools/bootstrap-recovery/inventory-storage-share.py": "sha256:673f9b17fb2f3fef6fcdb5c50a2bcca69e3df3157a9e5582a2d995c6b87640b5",
"tools/bootstrap-recovery/isolated-keycape-session.py": "sha256:8fa38efe28167b59a6307f342b2ac8fcda78ecda60e6228f22017424d2343c22",
"tools/bootstrap-recovery/isolated-keycape.py": "sha256:f67627b7d1647d75187dd899874efe22383bc91de5d97dabb9801076668ab3c2",
"tools/bootstrap-recovery/isolated-lldap.py": "sha256:35bfa8411186ad936a199bd450c7e18c2ad51059301a6db9650e6bd055ab1b81",
"tools/bootstrap-recovery/isolated-privacyidea.py": "sha256:4df7196678aac1ad594f32abf45c68dc21a9d2760d9e8a536a215a4a3c5d277f",
"tools/bootstrap-recovery/prepare-identity-coverage.py": "sha256:cd5f3067d28b4c7cc93ed08b8cf3869218bdecf7c54b5dffed22d697de71dbb1",
"tools/bootstrap-recovery/recovery-kit.py": "sha256:6fa0f1494dccaf003b6c6396c5de8b36faa2b276eeb16badd04383f5efb71e66",
"tools/bootstrap-recovery/retrieve-identity-offline.py": "sha256:3421dbb3f3280a001d33ea2deb12ab609404a0b4254cd43c21518d77d166f972",
"tools/bootstrap-recovery/retrieve-identity.py": "sha256:ec123d47a07305253570dacab7e103f8c193a23f20c9fc87466ea4a95ae0ec13",
"tools/bootstrap-recovery/run-isolated-keycape.sh": "sha256:0d994c9c14ebdffdbf7879bd7562042531083d0e8955424d0ae099898a141668",
"tools/bootstrap-recovery/verify-bundle.py": "sha256:885b05a2011e711228e42e316bda70bbe0d97d481383d61aa8469d0f10452370",
"tools/bootstrap-recovery/verify-identity-delivery.py": "sha256:63658d1976806f8d0ce80ba8f750bbcfdca59da936b6eb6a78a250f0e53710f1",
"tools/bootstrap-recovery/verify-identity.py": "sha256:039864d0141b552887752bf17cfdcf1bef4bdc9126f946a9c0533e4fd54ebf4f",
"sso-mfa/bootstrap/sops-custody-unlock.sh": "sha256:86c527a8ae0d0cca66a2345c566b25fbff5bc52c1e22c725548c0fbb2f86a58e",
"sso-mfa/bootstrap/sops-custody-fd.py": "sha256:df29eaba83ec6b5fc311664dcf4b22d8f3c8ed7274ed8e6c419ada0dcd169c0f",
"tools/bootstrap-recovery/prepare-isolated-keycape.py": "sha256:5098bdfa4e0ad30627c1716e5e355a2c78dca6ccc607bd2148a2e06126632288",
"tests/test_prepare_isolated_keycape.py": "sha256:4b0871f9a11a3940983464f1b40eca1613110b1cb3cb9694f4891d58e14c700a",
"tests/test_sealed_custody.py": "sha256:f0fa07c89e2419464dbee0489462864a4533770417a554b1ab1d1ed5d7b6c49f",
"tests/test_private_manifest_binding.py": "sha256:2e7ee16af65390343435b7210ef191aa2a10e9dcbb8759ed95fa8bf6fb81893b"
},
"fixed_endpoint": "unix:///var/run/docker.sock",
"engine_id": "74f93783-6cef-43a8-a3c4-f62586cd3bc1",
"engine_name": "bnt-lap001",
"images": [
"forgejo.coulomb.social/coulomb/key-cape@sha256:cdec9dd7c515f7e0a6169fc3fa560ee2cc47bf40d899755dec808c362f5ce010",
"ghcr.io/gpappsoft/privacyidea-docker@sha256:af7841adad262f129e0c1d4f553af13f21cb2f4dc713533f316cfe43ed0b4473"
],
"pull_policy": "never",
"synthetic_issuer": "https://nk49-fixture.invalid",
"cases": [
"positive",
"hard-parent-loss",
"deadline",
"nonzero-after-accept"
],
"limits": {
"sequential_cases": true,
"max_live_service_containers": 2,
"max_tmpfs_volumes": 4,
"max_public_probe_containers": 1,
"service_memory_bytes_each": 536870912,
"service_cpus_each": 1,
"tmpfs_volume_bytes_each": 33554432,
"host_tmpfs_fixture_limit_expected_bytes": 8388608,
"execution_deadline_seconds": {
"deadline_case": 20,
"other_cases": 120
},
"driver_wait_seconds_per_case": 150
},
"resources": {
"container_names": "nk49-kc-<fresh-run-id>-{transport,issuer,image-probe}",
"volume_names": "nk49-kc-<fresh-run-id>-v{0,1,2,3}",
"label": "net-kingdom.recovery-run=<fresh-run-id>",
"host_plaintext": "exact private /dev/shm/nk49-keycape-<run-id>-<suffix> registered before writes",
"synthetic_source_root": "private /dev/shm/nk49-keycape-fixture-<suffix>",
"safe_result": "private /tmp/nk49-keycape-native-fixture-review-<suffix>/result.json"
},
"scope": "candidate itself on complete synthetic format/integrity archive fixture; no native PI/LLDAP/Authelia proof, real custody, real factor or real client credential; successor binds sealed-descriptor custody preparation sources; synthetic fixture itself overrides custody with generated fixture key as before",
"mutations": [
"conditional creation of exact fresh local containers and local-driver tmpfs volumes",
"private synthetic file import to those volumes",
"original-image KeyCape service/client exchange within same loopback-only namespace",
"exact owned quiescent removal and bounded safe receipts"
],
"forbidden": [
"Docker execution before independent approval",
"image pulls/builds/pushes",
"production routes/ports/host networking/host directory mounts",
"OpenBao/Kubernetes/provider APIs",
"actual identity archive, age key, TLS key, user password, factor or client secret"
],
"cleanup": "candidate exact ownership/volume-driver/inode checks; quiesce fresh owner SID/PGID via pidfds first; retained inflight uncertainty never accepted even on absent readback",
"oracles": [
"positive issuer-native verify-client must establish original-hostname HTTPS trust; helper alone never suffices",
"same original single RS256 key-1 JWKS and exact config/registry/policy/history/lock/deliveries",
"correct service exchange and explicit wrong-secret/excess-scope denial through existing native verifier",
"hard-parent-loss and tightened deadline both quiesce before exact resource/plaintext cleanup",
"after real Docker run accepts a synthetic container, inject returncode1; require failed recovery and sticky daemon uncertainty after cleanup"
],
"proposed_command": [
"python3",
"tests/run_isolated_keycape_native_fixture.py",
"--execute-reviewed-fixture",
"--expected-manifest-sha256",
"<SHA256-of-this-file>",
"--case",
"positive"
],
"extended_command": [
"python3",
"tests/run_isolated_keycape_native_fixture.py",
"--execute-reviewed-fixture",
"--expected-manifest-sha256",
"<SHA256-of-this-file>",
"--case",
"all"
]
}

234
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/preimages/tests/test_isolated_keycape_recovery.py Normal file
View file

@ -0,0 +1,234 @@
import base64
import hashlib
import importlib.util
import io
import json
import subprocess
import tarfile
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
SOURCE=Path(__file__).parents[1]/'tools/bootstrap-recovery/isolated-keycape.py'
spec=importlib.util.spec_from_file_location('adapter',SOURCE)
a=importlib.util.module_from_spec(spec);spec.loader.exec_module(a)
class ArchiveTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
self.root=Path(self.tmp.name)
config={'data':{k:base64.b64encode(v).decode() for k,v in {'config.yaml':b'issuer: https://issuer.example','key.pem':b'fixture-key'}.items()}}
self.values={'resources/sso/secret/keycape-config.json':json.dumps(config).encode(),
'resources/sso/deployment/keycape.json':json.dumps({'spec':{'template':{'spec':{'containers':[{'name':'keycape','image':a.IMAGE,'env':[
{'name':'KEYCAPE_CONFIG','value':'/etc/keycape/config.yaml'},
{'name':'KEYCAPE_POLICY_PATH','value':'/var/lib/keycape-policy/policy.json'},
{'name':'KEYCAPE_ORIGINAL_SECRET','valueFrom':{'secretKeyRef':{'name':'client','key':'secret'}}}]}]}}}}).encode(),
'resources/sso/secret/client.json':json.dumps({'data':{'secret':base64.b64encode(b'fixture-private').decode()}}).encode(),
'data/keycape/policy.json':b'{}','data/keycape/receipts/one.json':b'{"original":true}'}
for name,value in self.values.items():
p=self.root/name;p.parent.mkdir(parents=True,exist_ok=True);p.write_bytes(value);p.chmod(0o600)
self.manifest={'profile':{'components':[{'namespace':'sso','deployment':'keycape','secret_refs':['keycape-config','client'],'configmap_refs':[]}]},
'files':[{'path':k,'bytes':len(v),'sha256':hashlib.sha256(v).hexdigest()} for k,v in self.values.items()]}
def payload(self):
digest=hashlib.sha256(json.dumps(self.manifest,sort_keys=True,separators=(',',':')).encode()).hexdigest()
return a.private_payload(self.root,self.manifest,digest)
def test_original_state_and_secret_delivery_retained(self):
p=self.payload();self.assertEqual(p['env']['KEYCAPE_ORIGINAL_SECRET'],'fixture-private')
self.assertEqual(base64.b64decode(p['files']['/var/lib/keycape-policy/receipts/one.json']),b'{"original":true}')
def test_archive_digest_refused(self):
with self.assertRaises(a.Refused):a.private_payload(self.root,self.manifest,'0'*64)
def test_changed_file_refused(self):
(self.root/'data/keycape/policy.json').write_text('{"changed":true}')
with self.assertRaises(a.Refused):self.payload()
def test_symlink_refused(self):
p=self.root/'data/keycape/policy.json';p.unlink();p.symlink_to('/etc/hostname')
with self.assertRaises(Exception):self.payload()
def test_duplicate_manifest_refused(self):
self.manifest['files'].append(self.manifest['files'][0])
with self.assertRaises(a.Refused):self.payload()
def test_image_mismatch_refused(self):
d=json.loads(self.values['resources/sso/deployment/keycape.json']);d['spec']['template']['spec']['containers'][0]['image']='keycape:latest'
raw=json.dumps(d).encode();p=self.root/'resources/sso/deployment/keycape.json';p.write_bytes(raw)
for row in self.manifest['files']:
if row['path'].endswith('/deployment/keycape.json'):row.update(bytes=len(raw),sha256=hashlib.sha256(raw).hexdigest())
with self.assertRaises(a.Refused):self.payload()
class NativeTests(unittest.TestCase):
def setUp(self):
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
self.key=rsa.generate_private_key(public_exponent=65537,key_size=2048).private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption())
self.cfg=b'issuer: https://issuer.example\nclients:\n- clientId: existing\n clientType: confidential\n secretRef: env:KEYCAPE_ORIGINAL_SECRET\n grantTypes: [client_credentials]\n'
self.files={'/etc/keycape/config.yaml':self.cfg,'/etc/keycape/key.pem':self.key,'/var/lib/keycape-policy/policy.json':b'{}'}
self.payload={'env':{'KEYCAPE_ORIGINAL_SECRET':'fixture-private'},'files':{k:base64.b64encode(v).decode() for k,v in self.files.items()}}
self.admission={'issuer':'https://issuer.example','verifier_sha256':hashlib.sha256(b'binary').hexdigest(),
'trust_path':'/trust.pem','trust_sha256':hashlib.sha256(b'original-trust').hexdigest(),
'consumer':{'client_id':'existing','secret_env':'KEYCAPE_ORIGINAL_SECRET','scope':'read','audience':'consumer',
'subject':'service:existing','tenant':'tenant:platform','roles':'reader','deny_scope':'write'}}
self.session={'issuer':'isolated','volumes':{x:{'target':x} for x in ('/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls')}}
self.output=''.join('PASS '+x+'\n' for x in ['discovery and endpoint origin','exchange and JWKS signature for granted scopes','access-token claim bindings','principal_type is service','sub matches the registration','tenant matches the registration','roles present','granted scope carries no excess grant','predecessor secret refused','excess scope refused: write'])
def file(self,container,path):
return {**self.files,'/etc/hosts':b'127.0.0.1 issuer.example\n','/keycape':b'binary','/trust.pem':b'original-trust'}[path]
def command(self,argv,**kwargs):
if argv[1:3]==['container','inspect']:return json.dumps(['KEYCAPE_ORIGINAL_SECRET=fixture-private','SSL_CERT_FILE=/trust.pem']).encode()
output=io.BytesIO()
with tarfile.open(fileobj=output,mode='w') as t:
m=tarfile.TarInfo('policy.json');m.size=2;t.addfile(m,io.BytesIO(b'{}'))
return output.getvalue()
def verify(self,code=0,output=None):
with patch.object(a,'check_jwks'),patch.object(a,'runtime_file',return_value=b'original-trust'),patch.object(a,'runtime_files',side_effect=lambda session,paths=(),tree=False:({k:v for k,v in self.files.items() if k.startswith('/var/lib/keycape-policy/')} if tree else dict(self.files))),patch.object(a,'container_file',side_effect=self.file),patch.object(a,'command',side_effect=self.command),patch.object(a.subprocess,'run',return_value=subprocess.CompletedProcess([],code,b'',(self.output if output is None else output).encode())) as run:
result=a.run_verifier(self.session,self.admission,self.payload)
argv=run.call_args.args[0]
self.assertNotIn('fixture-private',' '.join(argv))
self.assertIn('-previous-secret-env',argv)
return result
def test_original_registration_and_unrelated_secret(self):
r=self.verify();self.assertTrue(r['wrong_unrelated_secret_explicitly_refused']);self.assertNotIn('fixture-private',json.dumps(r))
def test_transport_failure_never_counts_as_wrong_secret(self):
with self.assertRaises(a.Refused):self.verify(1,'FAIL discovery and endpoint origin: TLS failure\n')
def test_missing_explicit_rejection_never_counts(self):
with self.assertRaises(a.Refused):self.verify(0,self.output.replace('PASS predecessor secret refused\n',''))
def test_fail_line_refused_even_zero_exit(self):
with self.assertRaises(a.Refused):self.verify(0,self.output+'FAIL transport\n')
def test_security_environment_override_refused(self):
original=self.command
def altered(argv,**kwargs):
if argv[1:3]==['container','inspect']:return json.dumps(['KEYCAPE_ORIGINAL_SECRET=fixture-private','SSL_CERT_FILE=/trust.pem','SSL_CERT_DIR=/unreviewed']).encode()
return original(argv,**kwargs)
with patch.object(self,'command',side_effect=altered):
with self.assertRaises(a.Refused):self.verify()
def test_http_issuer_refused(self):
self.admission['issuer']='http://issuer.example'
with self.assertRaises(a.Refused):self.verify()
def test_changed_trust_refused(self):
self.admission['trust_sha256']='0'*64
with self.assertRaises(a.Refused):self.verify()
def test_config_drift_refused(self):
self.payload['files']['/etc/keycape/config.yaml']=base64.b64encode(b'changed').decode()
with self.assertRaises(a.Refused):self.verify()
class PublicReadinessTransportTests(unittest.TestCase):
def test_provider_probe_failure_is_closed_false_data_not_failed_docker_exec(self):
output=io.StringIO()
with patch('sys.stdin',io.StringIO('{}')),patch('sys.stdout',output):
exec(compile(a.JWKS_TRANSPORT,'<public-readiness>','exec',optimize=2),{})
self.assertEqual(json.loads(output.getvalue()),{'original_public_signing_key_in_live_jwks':False})
class BoundaryTests(unittest.TestCase):
def test_remote_docker_refused_before_info(self):
session={'run_id':'a'*32,'containers':['nk49-kc-'+('a'*32)+'-worker'],'worker':'nk49-kc-'+('a'*32)+'-worker','issuer':'nk49-kc-'+('a'*32)+'-worker'}
with patch.object(a,'command',return_value=b'[{"Endpoints":{"docker":{"Host":"tcp://remote:2375"}}}]') as command:
with self.assertRaises(a.Refused):a.inspect_session(session)
self.assertEqual(command.call_count,1)
def test_failed_verifier_cleanup_is_attempted(self):
receipt={};admission={'session':{},'ciphertext_sha256':'a'*64,'manifest_sha256':'b'*64,'verifier_sha256':'c'*64,'trust_sha256':'d'*64,'consumer':{x:'yes' for x in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope')}}
with patch.object(a,'inspect_session'),patch.object(a,'private_payload',return_value={}),patch.object(a,'run_verifier',side_effect=a.Refused),patch.object(a,'cleanup') as cleanup:
with self.assertRaises(a.Refused):a.native_checks(Path('/unused'),{},admission,receipt)
cleanup.assert_called_once();self.assertTrue(receipt['container_cleanup_verified']);self.assertEqual(receipt['status'],'failed')
def test_cleanup_failure_overrides_success(self):
admission={'session':{},'ciphertext_sha256':'a'*64,'manifest_sha256':'b'*64,'verifier_sha256':'c'*64,'trust_sha256':'d'*64,'consumer':{x:'yes' for x in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope')}}
result={x:True for x in ('native_service_registration_verified','wrong_unrelated_secret_explicitly_refused','denied_scopes_refused','original_config_registry_policy_receipts_equal','original_https_issuer_and_trust_preserved','original_public_signing_key_in_live_jwks')}
result['public_signing_spki_sha256']='e'*64;receipt={}
with patch.object(a,'inspect_session'),patch.object(a,'private_payload',return_value={}),patch.object(a,'run_verifier',return_value=result),patch.object(a,'cleanup',side_effect=a.Refused):
with self.assertRaises(a.Refused):a.native_checks(Path('/unused'),{},admission,receipt)
self.assertEqual(receipt['status'],'failed');self.assertFalse(receipt['container_cleanup_verified'])
class ContainmentTests(unittest.TestCase):
def inspect(self,overrides=None,launch=None,mounts=None):
name='nk49-kc-'+('a'*32)+'-issuer'
session={'run_id':'a'*32,'containers':[name],'issuer':name,'worker':name,'images':{name:a.IMAGE}}
host={'NetworkMode':'none','ReadonlyRootfs':True,'Privileged':False,'CapDrop':['ALL'],'SecurityOpt':['no-new-privileges']}
host.update(overrides or {})
def command(argv,**kwargs):
if argv[1:3]==['context','inspect']:return b'[{"Endpoints":{"docker":{"Host":"unix:///var/run/docker.sock"}}}]'
if argv[1]=='info':return b'{"ID":"74f93783-6cef-43a8-a3c4-f62586cd3bc1","Name":"bnt-lap001","OSType":"linux"}'
if argv[-1]=='{{.Id}}':return b'anchor-id'
if '"path"' in argv[-1]:return json.dumps(launch or {'entrypoint':['/keycape'],'cmd':None,'path':'/keycape','args':[]}).encode()
return json.dumps({'host':host,'mounts':mounts or [],'labels':{a.LABEL:'a'*32},'image':a.IMAGE,'user':'1000:1000'}).encode()
with patch.object(a,'command',side_effect=command),patch.dict('os.environ',{'DOCKER_HOST':'unix:///var/run/docker.sock'}):a.inspect_session(session)
def test_minimal_network_none_admitted(self):self.inspect()
def test_image_declared_anonymous_disk_volume_remains_refused(self):
with self.assertRaises(a.Refused):
self.inspect(mounts=[{'Type':'volume','Name':'anonymous-fixture','Destination':'/privacyidea/etc/persistent','RW':True}])
def test_alternate_same_image_command_refused(self):
with self.assertRaises(a.Refused):self.inspect(launch={'entrypoint':['/keycape'],'cmd':['check-config'],'path':'/keycape','args':['check-config']})
def test_cap_add_refused(self):
with self.assertRaises(a.Refused):self.inspect({'CapAdd':['NET_ADMIN']})
def test_devices_refused(self):
with self.assertRaises(a.Refused):self.inspect({'Devices':[{'PathOnHost':'/dev/mem'}]})
def test_device_requests_refused(self):
with self.assertRaises(a.Refused):self.inspect({'DeviceRequests':[{'Driver':'nvidia'}]})
def test_host_namespace_refused(self):
for field in ('PidMode','IpcMode','UTSMode','UsernsMode','CgroupnsMode'):
with self.subTest(field=field),self.assertRaises(a.Refused):self.inspect({field:'host'})
def test_security_override_refused(self):
with self.assertRaises(a.Refused):self.inspect({'SecurityOpt':['no-new-privileges','seccomp=unconfined']})
class CleanupTests(unittest.TestCase):
def test_partial_cleanup_attempts_every_owned_container(self):
session={'run_id':'a'*32,'containers':['first','second','third']};receipt={}
removed=[]
def command(argv,**kwargs):
if argv[1:3]==['container','inspect']:return ('a'*32).encode()
if argv[1]=='rm':
removed.append(argv[-1])
if argv[-1]=='third':raise a.Refused()
return b''
if argv[1:3]==['container','ls']:
name=next(x for x in argv if x.startswith('name=^/')).removeprefix('name=^/').removesuffix('$')
return (name+'\n').encode() if name=='third' or name not in removed else b''
return b''
with patch.object(a,'command',side_effect=command):
with self.assertRaises(a.Refused):a.cleanup(session,receipt)
self.assertEqual(removed,['third','second','first'])
self.assertEqual([r['status'] for r in receipt['container_cleanup_obligations']],['still_present','absence_verified','absence_verified'])
def test_ownership_mismatch_does_not_remove_and_continues(self):
session={'run_id':'a'*32,'containers':['first','second']};receipt={};removed=[]
def command(argv,**kwargs):
if argv[1:3]==['container','inspect']:return (('b' if argv[3]=='second' else 'a')*32).encode()
if argv[1]=='rm':removed.append(argv[-1])
if argv[1:3]==['container','ls']:
name=next(x for x in argv if x.startswith('name=^/')).removeprefix('name=^/').removesuffix('$')
return (name+'\n').encode() if name not in removed else b''
return b''
with patch.object(a,'command',side_effect=command):
with self.assertRaises(a.Refused):a.cleanup(session,receipt)
self.assertEqual(removed,['first'])
self.assertEqual(receipt['container_cleanup_obligations'][0]['status'],'ownership_mismatch_not_removed')
def test_inspect_failure_retains_uncertain_obligation(self):
receipt={}
with patch.object(a,'command',side_effect=a.Refused):
with self.assertRaises(a.Refused):a.cleanup({'run_id':'a'*32,'containers':['exact-owned-name']},receipt)
self.assertEqual(receipt['container_cleanup_obligations'],[{'container':'exact-owned-name','status':'unverified'}])
class JWKSTests(unittest.TestCase):
def execute(self,n='123',endpoint='https://issuer.example/jwks',resolved='127.0.0.1',extra=False,kid='key-1'):
request={'issuer':'https://issuer.example','trust':'public-certificate','n':'123','e':'65537'}
def enc(i):return base64.urlsafe_b64encode(int(i).to_bytes((int(i).bit_length()+7)//8,'big')).decode().rstrip('=')
responses=[{'issuer':'https://issuer.example','jwks_uri':endpoint},{'keys':[{'kty':'RSA','kid':kid,'use':'sig','alg':'RS256','n':enc(n),'e':enc(65537)}]}]
if extra:responses[1]['keys'].append(dict(responses[1]['keys'][0],kid='key-2'))
class Response:
def __init__(self,value):self.value=value
def __enter__(self):return self
def __exit__(self,*args):pass
def read(self,*args):return json.dumps(self.value).encode()
class Opener:
def open(self,*args,**kwargs):return Response(responses.pop(0))
with patch('sys.stdin',io.StringIO(json.dumps(request))),patch('sys.stdout',io.StringIO()),patch('socket.getaddrinfo',return_value=[(2,1,6,'',(resolved,443))]),patch('ssl.create_default_context'),patch('urllib.request.build_opener',return_value=Opener()):
exec(compile(a.JWKS_PROBE,'<jwks-probe>','exec',optimize=2),{})
def test_exact_original_public_key(self):self.execute()
def test_extra_signing_key_refused_under_optimization(self):
with self.assertRaises(ValueError):self.execute(extra=True)
def test_wrong_kid_refused_under_optimization(self):
with self.assertRaises(ValueError):self.execute(kid='replacement')
def test_wrong_live_signing_key_refused(self):
with self.assertRaises(ValueError):self.execute(n='124')
def test_cross_origin_jwks_refused(self):
with self.assertRaises(ValueError):self.execute(endpoint='https://production.example/jwks')
def test_production_dns_refused_before_fetch(self):
with self.assertRaises(ValueError):self.execute(resolved='192.0.2.1')
if __name__=='__main__':unittest.main()

478
docs/changes/netkingdom-target-2026-10-17/native-identity-r1/preimages/tools/bootstrap-recovery/isolated-keycape.py Normal file
View file

@ -0,0 +1,478 @@
#!/usr/bin/env python3
"""T18 native adapter for an owner-admitted, already restored isolated session.
Called after identity-bundle.restore, never with an unvalidated directory. This
module does not provision TLS, grant credentials, or claim downstream acceptance.
The existing decrypt/tmpfs wrapper remains responsible for archive cleanup.
"""
import base64
import hashlib
import importlib.util
import json
import io
import os
import secrets
import tarfile
import time
import re
import subprocess
from pathlib import Path
HERE = Path(__file__).resolve().parent
IMAGE = 'forgejo.coulomb.social/coulomb/key-cape@sha256:cdec9dd7c515f7e0a6169fc3fa560ee2cc47bf40d899755dec808c362f5ce010'
LABEL = 'net-kingdom.recovery-run'
class Refused(Exception):
pass
def load_bundle():
spec = importlib.util.spec_from_file_location('identity_bundle', HERE / 'identity-bundle.py')
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
def manifest_binding_mode(admission,transfer):
"""Versioned private-manifest binding; legacy explicit digests stay valid."""
mode=admission.get('manifest_binding_mode','explicit-digest-v1')
if mode not in ('explicit-digest-v1','ciphertext-derived-v1'):raise Refused()
if (not isinstance(admission.get('ciphertext_sha256'),str)
or not re.fullmatch('[a-f0-9]{64}',admission['ciphertext_sha256'])
or admission['ciphertext_sha256']!=transfer.get('ciphertext_sha256')
or transfer.get('independent_owner_retrieval_verified') is not True):raise Refused()
if mode=='explicit-digest-v1':
if not isinstance(admission.get('manifest_sha256'),str) or not re.fullmatch('[a-f0-9]{64}',admission['manifest_sha256']):raise Refused()
elif 'manifest_sha256' in admission:raise Refused()
return mode
def private_manifest_admission(admission,manifest):
# Only called after the fixed ciphertext bytes were hashed, authenticated
# decryption succeeded and bundle.restore checked the complete archive.
# Never return this enriched private object in a safe receipt/checkpoint.
digest=hashlib.sha256(json.dumps(manifest,sort_keys=True,separators=(',',':')).encode()).hexdigest()
mode=admission.get('manifest_binding_mode','explicit-digest-v1')
if mode=='explicit-digest-v1':
if digest!=admission['manifest_sha256']:raise Refused()
elif mode!='ciphertext-derived-v1' or 'manifest_sha256' in admission:raise Refused()
return {**admission,'manifest_sha256':digest}
def private_payload(root, manifest, expected_manifest_sha256):
"""Bind the complete validated archive before selecting KeyCape material.
Private hashes never enter the public receipt. The digest is supplied by the
existing private archive admission, not inferred from an untrusted manifest.
"""
encoded = json.dumps(manifest, sort_keys=True, separators=(',', ':')).encode()
if hashlib.sha256(encoded).hexdigest() != expected_manifest_sha256:
raise Refused()
entries = manifest['files']
if len(entries) != len({e['path'] for e in entries}):
raise Refused()
values = {}
bundle = load_bundle()
for entry in entries:
if not bundle.allowed(entry['path'], manifest['profile']):
raise Refused()
path = root / entry['path']
bundle.private(path)
value = path.read_bytes()
if len(value) != entry['bytes'] or hashlib.sha256(value).hexdigest() != entry['sha256']:
raise Refused()
values[entry['path']] = value
deployment = json.loads(values['resources/sso/deployment/keycape.json'])
main = next(c for c in deployment['spec']['template']['spec']['containers'] if c['name'] == 'keycape')
if main.get('command') or main.get('args') or main.get('envFrom'):
raise Refused()
if main['image'] != IMAGE:
# Mutable-tag/historical mismatches need a separate explicit owner admission.
raise Refused()
secret = json.loads(values['resources/sso/secret/keycape-config.json'])['data']
files = {'/etc/keycape/' + name: base64.b64decode(secret[name], validate=True)
for name in ('config.yaml', 'key.pem')}
for path, value in values.items():
if path.startswith('data/keycape/'):
relative = path.removeprefix('data/keycape/')
files['/var/lib/keycape-policy/' + relative] = value
if '/var/lib/keycape-policy/policy.json' not in files:
raise Refused()
if 'resources/sso/secret/keycape-factor-read.json' in values:
factor=json.loads(values['resources/sso/secret/keycape-factor-read.json'])['data']
files['/etc/keycape-factor/admin-token']=base64.b64decode(factor['admin-token'],validate=True)
# Resolve original delivered environment, including all referenced secrets.
env = {}
for item in main.get('env', []):
if not re.fullmatch(r'KEYCAPE_[A-Z0-9_]+',item['name']):raise Refused()
if item['name'] in env:
raise Refused()
if 'value' in item:
env[item['name']] = item['value']
else:
ref = item['valueFrom']['secretKeyRef']
source = json.loads(values['resources/sso/secret/' + ref['name'] + '.json'])['data']
env[item['name']] = base64.b64decode(source[ref['key']], validate=True).decode()
if env.get('KEYCAPE_CONFIG') != '/etc/keycape/config.yaml' or env.get('KEYCAPE_POLICY_PATH') != '/var/lib/keycape-policy/policy.json':
raise Refused()
return {'files': {p: base64.b64encode(v).decode() for p, v in files.items()}, 'env': env}
ENDPOINT = 'unix:///var/run/docker.sock'
EXECUTION = None
def docker_environment(environment=None):
# Even an admitted daemon exposed by another Unix socket is outside this
# exact session binding. Conflicting ambient routing is refused, not ignored.
for source in (os.environ, environment or {}):
if source.get('DOCKER_HOST') not in (None,'',ENDPOINT):raise Refused()
if source.get('DOCKER_CONTEXT') not in (None,'','default'):raise Refused()
if any(source.get(k) for k in ('DOCKER_TLS_VERIFY','DOCKER_CERT_PATH')):raise Refused()
env=dict(os.environ if environment is None else environment)
for key in ('DOCKER_CONTEXT','DOCKER_TLS_VERIFY','DOCKER_CERT_PATH'):env.pop(key,None)
env['DOCKER_HOST']=ENDPOINT
return env
def process(argv, **kwargs):
marker=None
if EXECUTION is not None and (kwargs.get('start_new_session') or kwargs.get('process_group') not in (None,-1)):raise Refused()
if argv[0]=='docker':
argv=['docker','--host',ENDPOINT,*argv[1:]]
kwargs['env']=docker_environment(kwargs.get('env'))
if EXECUTION is not None:
if (os.getpid()!=EXECUTION['pid'] or os.getsid(0)!=EXECUTION['sid'] or os.getpgrp()!=EXECUTION['pgid']
or time.monotonic()>=EXECUTION['deadline']):raise Refused()
marker=Path(EXECUTION['staging']+'-docker-inflight')
fd=os.open(marker,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
os.close(fd)
try:
result=subprocess.run(argv,**kwargs)
if marker is not None and result.returncode!=0:
fd=os.open(Path(EXECUTION['staging']+'-docker-uncertain'),os.O_WRONLY|os.O_CREAT|os.O_NOFOLLOW,0o600);os.close(fd)
return result
except BaseException:
# Parent loss leaves .docker-inflight; caught timeout/transport failures
# retain a sticky uncertainty marker even if later cleanup can run.
if marker is not None:
fd=os.open(Path(EXECUTION['staging']+'-docker-uncertain'),os.O_WRONLY|os.O_CREAT|os.O_NOFOLLOW,0o600);os.close(fd)
raise
finally:
if marker is not None:
try:marker.unlink()
except FileNotFoundError:pass
def command(argv, **kwargs):
maximum = kwargs.pop('max_output', 32768)
result = process(argv, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
timeout=kwargs.pop('timeout', 180), **kwargs)
if result.returncode or len(result.stdout) > maximum:
raise Refused()
return result.stdout
def inspect_session(session, worker_only=False):
"""No credential environment is read by Docker inspect or returned to caller."""
if not re.fullmatch(r'[a-f0-9]{32}', session['run_id']):
raise Refused()
names = session['containers']
if not names or len(names) > 12 or len(set(names)) != len(names):
raise Refused()
for name in names:
if not re.fullmatch(r'nk49-kc-' + session['run_id'] + r'-[a-z0-9-]{1,24}', name):
raise Refused()
if session['worker'] not in names or session['issuer'] not in names:
raise Refused()
import os
contexts=json.loads(command(['docker','context','inspect']))
endpoints=[contexts[0]['Endpoints']['docker']['Host'],os.environ.get('DOCKER_HOST') or ENDPOINT]
if not all(x==ENDPOINT for x in endpoints):raise Refused()
info = json.loads(command(['docker', 'info', '--format', '{"ID":{{json .ID}},"Name":{{json .Name}},"OSType":{{json .OSType}}}']))
if info != {'ID': '74f93783-6cef-43a8-a3c4-f62586cd3bc1', 'Name': 'bnt-lap001', 'OSType': 'linux'}:
raise Refused()
# The worker is the anchor; all dependencies/TLS/issuer share its network-none namespace.
anchor = command(['docker', 'container', 'inspect', session['worker'], '--format', '{{.Id}}']).decode().strip()
for name,spec in session.get('volumes',{}).items():
volume=json.loads(command(['docker','volume','inspect',name]))
expected={'type':'tmpfs','device':'tmpfs','o':'size=33554432,uid=1000,gid=1000,mode=0700,nosuid,nodev,noexec'}
if len(volume)!=1 or volume[0]['Driver']!='local' or volume[0].get('Scope')!='local' or volume[0].get('Options')!=expected or spec['options']!=expected or volume[0].get('Labels',{}).get(LABEL)!=session['run_id']:raise Refused()
for name in names:
obj = json.loads(command(['docker', 'container', 'inspect', name, '--format',
'{"host":{{json .HostConfig}},"mounts":{{json .Mounts}},"labels":{{json .Config.Labels}},"image":{{json .Config.Image}},"user":{{json .Config.User}},"entrypoint":{{json .Config.Entrypoint}},"cmd":{{json .Config.Cmd}}}']))
host = obj['host']
mounts=obj['mounts']
destinations=[m['Destination'] for m in mounts]
if len(destinations)!=len(set(destinations)):raise Refused()
for volume,expected_spec in session.get('volumes',{}).items():
matches=[m for m in mounts if m['Destination']==expected_spec['target']]
if len(matches)!=1 or matches[0]['Type']!='volume' or matches[0]['Name']!=volume or matches[0]['RW']!=(name==session['worker'] or expected_spec['target']=='/var/lib/keycape-policy'):raise Refused()
if (obj['labels'].get(LABEL) != session['run_id'] or not obj['user'] or obj['user'].split(':')[0] in ('0', 'root')
or not host['ReadonlyRootfs'] or host['Privileged'] or host.get('Binds') or host.get('PortBindings')
or host.get('PidMode') or host.get('IpcMode') == 'host'
or host.get('UTSMode') or host.get('UsernsMode') == 'host' or host.get('CgroupnsMode') == 'host'
or host.get('Sysctls',{}) not in ({},{'net.ipv4.ip_unprivileged_port_start':'0'} if name==session['worker'] else {})
or host.get('CapAdd') or host.get('Devices') or host.get('DeviceRequests') or host.get('VolumesFrom')
or host.get('CapDrop') != ['ALL']
or host.get('SecurityOpt') not in (['no-new-privileges'], ['no-new-privileges:true'])
or any(m['Type']!='tmpfs' and not (m['Type']=='volume' and m['Name'] in session.get('volumes',{}) and m['Destination']==session['volumes'][m['Name']]['target']) for m in obj['mounts'])
or host['NetworkMode'] != ('none' if name == session['worker'] else 'container:' + anchor)
or obj['image'] != session['images'][name] or '@sha256:' not in obj['image']):
raise Refused()
if worker_only:return
if session['images'][session['issuer']] != IMAGE:
raise Refused()
# Known admitted archive has no command/args override; image default /keycape
# must be the actual issuer executable, not an alternate same-image command.
launch=json.loads(command(['docker','container','inspect',session['issuer'],'--format',
'{"entrypoint":{{json .Config.Entrypoint}},"cmd":{{json .Config.Cmd}},"path":{{json .Path}},"args":{{json .Args}}}']))
if launch != {'entrypoint':['/keycape'],'cmd':None,'path':'/keycape','args':[]}:
raise Refused()
def container_file(container, path):
# Docker streams one tar member into memory; never materialize private files.
raw = command(['docker', 'cp', container + ':' + path, '-'], max_output=128*1024*1024)
with tarfile.open(fileobj=io.BytesIO(raw), mode='r:') as archive:
members = archive.getmembers()
if len(members) != 1 or not members[0].isfile() or members[0].size > 96*1024*1024:
raise Refused()
return archive.extractfile(members[0]).read()
JWKS_PROBE = r"""
import base64,ipaddress,json,socket,ssl,sys,urllib.request
from urllib.parse import urlsplit
p=json.load(sys.stdin);issuer=p['issuer'];origin=urlsplit(issuer)
def require(condition):
if not condition:raise ValueError('public_jwks_binding_refused')
require(origin.scheme=='https')
resolved=socket.getaddrinfo(origin.hostname,origin.port or 443,type=socket.SOCK_STREAM)
require(bool(resolved) and all(ipaddress.ip_address(x[4][0]).is_loopback for x in resolved))
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self,*args,**kwargs):return None
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),NoRedirect(),
urllib.request.HTTPSHandler(context=ssl.create_default_context(cadata=p['trust'])))
def get(url):
u=urlsplit(url)
require((u.scheme,u.hostname,u.port or 443)==(origin.scheme,origin.hostname,origin.port or 443))
require(not u.username and not u.password and not u.fragment)
with opener.open(url,timeout=10) as response:
raw=response.read(1024*1024+1);require(len(raw)<=1024*1024)
return json.loads(raw)
d=get(issuer.rstrip('/')+'/.well-known/openid-configuration');require(d['issuer']==issuer)
keys=get(d['jwks_uri'])['keys']
def integer(value):return int.from_bytes(base64.urlsafe_b64decode(value+'='*(-len(value)%4)),'big')
# The exact admitted owner release constructs one key, kid key-1. Reject every
# extra key, including irrelevant keys: a native successful exchange can then
# only have used the original signer through the existing verifier.
require(isinstance(keys,list) and len(keys)==1)
k=keys[0]
require(k.get('kty')=='RSA' and k.get('kid')=='key-1' and k.get('use')=='sig' and k.get('alg')=='RS256')
require(integer(k['n'])==int(p['n']) and integer(k['e'])==int(p['e']))
print('{"original_public_signing_key_in_live_jwks":true}')
"""
# A failed public readiness probe is closed data, not a failing Docker RPC.
# Transport/CLI nonzero remains uncertain; no deliberately failing exec is
# retried as if daemon acceptance had been established.
JWKS_TRANSPORT = ("import contextlib,io\ntry:\n"
" with contextlib.redirect_stdout(io.StringIO()):\n"
" exec(" + repr(JWKS_PROBE) + ")\n"
"except Exception:\n print('{\"original_public_signing_key_in_live_jwks\":false}')\n"
"else:\n print('{\"original_public_signing_key_in_live_jwks\":true}')\n")
def check_jwks(session, admission, public_numbers):
trust=runtime_file(session,admission['trust_path']).decode()
request={'issuer':admission['issuer'],'trust':trust,'n':str(public_numbers.n),'e':str(public_numbers.e)}
result=json.loads(command(['docker','exec','-i',session['worker'],'python3','-c',JWKS_TRANSPORT],input=json.dumps(request).encode()))
if result != {'original_public_signing_key_in_live_jwks': True}:raise Refused()
RUNTIME_FILES = r"""
import base64,json,sys
from pathlib import Path
p=json.load(sys.stdin)
roots=('/etc/keycape/','/var/lib/keycape-policy/','/etc/keycape-factor/','/recovery-tls/')
result={}
paths=[]
if p.get('tree'):
paths=list(Path('/var/lib/keycape-policy').rglob('*'))
else:paths=[Path(x) for x in p['paths']]
for f in paths:
if f.is_dir():continue
if not str(f).startswith(roots) or not f.is_file() or any(x.is_symlink() for x in (f,*f.parents)) or f.stat().st_size>8*1024*1024:raise ValueError()
result[str(f)]=base64.b64encode(f.read_bytes()).decode()
if len(result)>4096:raise ValueError()
print(json.dumps(result))
"""
def runtime_files(session, paths=(), tree=False):
raw=command(['docker','exec','-i',session['worker'],'python3','-c',RUNTIME_FILES],
input=json.dumps({'paths':list(paths),'tree':tree}).encode(),max_output=48*1024*1024)
return {k:base64.b64decode(v,validate=True) for k,v in json.loads(raw).items()}
def runtime_file(session,path):
return runtime_files(session,[path])[path]
def run_verifier(session, admission, payload):
issuer = session['issuer']
if {v['target'] for v in session.get('volumes',{}).values()}!={'/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls'}:raise Refused()
from urllib.parse import urlsplit
import ipaddress
address = urlsplit(admission['issuer'])
if (address.scheme != 'https' or not address.hostname or address.username or address.password
or address.query or address.fragment):
raise Refused()
if hashlib.sha256(container_file(issuer, '/keycape')).hexdigest() != admission['verifier_sha256']:
raise Refused()
if hashlib.sha256(runtime_file(session, admission['trust_path'])).hexdigest() != admission['trust_sha256']:
raise Refused()
# Capture private environment only in owner process memory, never receipt.
environment = json.loads(command(['docker','container','inspect',issuer,'--format','{{json .Config.Env}}']))
env = dict(item.split('=',1) for item in environment)
if len(env)!=len(environment):raise Refused()
if set(env)-set(payload['env'])-{'PATH','HOME','HOSTNAME','SSL_CERT_FILE'}:raise Refused()
if env.get('SSL_CERT_FILE') != admission['trust_path'] or any(k.lower().endswith('_proxy') and v for k,v in env.items()):
raise Refused()
if {k:v for k,v in env.items() if k.startswith('KEYCAPE_')}!={k:v for k,v in payload['env'].items() if k.startswith('KEYCAPE_')}:raise Refused()
for name,value in payload['env'].items():
if env.get(name) != value:
raise Refused()
files = {k:base64.b64decode(v,validate=True) for k,v in payload['files'].items()}
def equal():
if runtime_files(session,files)!=files:raise Refused()
durable=runtime_files(session,tree=True)
if durable!={k:v for k,v in files.items() if k.startswith('/var/lib/keycape-policy/')}:
raise Refused()
equal()
# Reading YAML is data parsing only; no archived executable is evaluated.
import yaml
cfg = yaml.safe_load(files['/etc/keycape/config.yaml'])
if cfg['issuer'] != admission['issuer']:
raise Refused()
consumer = admission['consumer']
client = next(x for x in cfg['clients'] if x['clientId']==consumer['client_id'])
if (client['secretRef']!='env:'+consumer['secret_env'] or client['clientType']!='confidential'
or 'client_credentials' not in client['grantTypes'] or not env.get(consumer['secret_env'])):
raise Refused()
from cryptography.hazmat.primitives import serialization
key=serialization.load_pem_private_key(files['/etc/keycape/key.pem'],password=None)
check_jwks(session,admission,key.public_key().public_numbers())
wrong = secrets.token_urlsafe(48)
if wrong == env[consumer['secret_env']]:raise Refused()
args = ['docker','exec','--env','NK49_UNRELATED_NEGATIVE_SECRET='+wrong,issuer,'/keycape','verify-client',
'-issuer',admission['issuer'],'-previous-secret-env','NK49_UNRELATED_NEGATIVE_SECRET']
flags = {'client_id':'client-id','secret_env':'secret-env','audience':'audience','scope':'scope',
'subject':'expect-subject','tenant':'expect-tenant','roles':'expect-roles','deny_scope':'deny-scope'}
for name,flag in flags.items():args+=['-'+flag,consumer[name]]
# Native command reports on stderr. Do not return or print its raw response.
result = process(args,stdout=subprocess.PIPE,stderr=subprocess.PIPE,timeout=150)
raw = (result.stdout+result.stderr).decode()
required = ['discovery and endpoint origin','exchange and JWKS signature for granted scopes','access-token claim bindings',
'principal_type is service','sub matches the registration','tenant matches the registration','roles present',
'granted scope carries no excess grant','predecessor secret refused']
required += ['excess scope refused: '+scope for scope in consumer['deny_scope'].split()]
if result.returncode or any('PASS '+name+'\n' not in raw for name in required) or 'FAIL ' in raw:
raise Refused()
equal()
from cryptography.hazmat.primitives import serialization
key=serialization.load_pem_private_key(files['/etc/keycape/key.pem'],password=None)
check_jwks(session,admission,key.public_key().public_numbers())
public=key.public_key().public_bytes(serialization.Encoding.DER,serialization.PublicFormat.SubjectPublicKeyInfo)
return {'native_service_registration_verified':True,'wrong_unrelated_secret_explicitly_refused':True,
'denied_scopes_refused':True,'original_config_registry_policy_receipts_equal':True,
'public_signing_spki_sha256':hashlib.sha256(public).hexdigest(),
'original_https_issuer_and_trust_preserved':True,'original_public_signing_key_in_live_jwks':True}
def cleanup(session, receipt=None):
"""Attempt every owned cleanup even when one obligation fails.
Preserve exact safe container names and outcomes so the owner can recover an
uncertain/partial cleanup. An inspect/daemon failure never means absence.
"""
obligations = []
failed = False
for name in reversed(session['containers']):
row = {'container': name, 'status': 'unverified'}
obligations.append(row)
try:
present=command(['docker','container','ls','--all','--filter','name=^/'+name+'$','--format','{{.Names}}']).decode().splitlines()
if not present:
row['status']='absence_verified'
continue
label = command(['docker', 'container', 'inspect', name, '--format',
'{{index .Config.Labels "' + LABEL + '"}}']).decode().strip()
if label != session['run_id']:
row['status'] = 'ownership_mismatch_not_removed'
failed = True
continue
# A failed remove is followed by a successful explicit inventory;
# interrupted remove may actually have completed.
try:
command(['docker', 'rm', '--force', '--volumes', name])
except Exception:
pass
remaining = command(['docker','container','ls','--all','--filter',
'name=^/' + name + '$','--format','{{.Names}}']).decode().splitlines()
if remaining:
row['status'] = 'still_present'
failed = True
else:
row['status'] = 'absence_verified'
except Exception:
failed = True
if receipt is not None:
receipt['container_cleanup_obligations'] = obligations
if failed:
raise Refused()
def native_checks(root, manifest, admission, receipt, cleanup_session=True):
"""Owner wrapper provides a reviewed session after existing restore pipeline.
Admission must bind ciphertext, canonical private manifest digest, executable
and trust bytes, exact image/target list, original issuer and selected consumer.
None of these admission fields grant authority or carry a client secret.
"""
session = admission['session']
# Validate ownership/containment before permitting either exec or cleanup.
inspect_session(session)
receipt.update(status='failed', source_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
ciphertext_sha256=admission['ciphertext_sha256'], production_contacted=False,
human_mfa_proven=False, downstream_consumer_reconnect_proven=False,
container_cleanup_verified=False)
try:
for field in ('ciphertext_sha256', 'manifest_sha256', 'verifier_sha256', 'trust_sha256'):
if not re.fullmatch('[a-f0-9]{64}', admission[field]):
raise Refused()
for field in ('client_id','secret_env','audience','scope','subject','tenant','roles','deny_scope'):
if not isinstance(admission['consumer'].get(field), str) or not admission['consumer'][field].strip():
raise Refused()
payload = private_payload(root, manifest, admission['manifest_sha256'])
result = run_verifier(session, admission, payload)
required = {'native_service_registration_verified','wrong_unrelated_secret_explicitly_refused',
'denied_scopes_refused','original_config_registry_policy_receipts_equal',
'original_https_issuer_and_trust_preserved','original_public_signing_key_in_live_jwks'}
if any(result.get(k) is not True for k in required) or not re.fullmatch('[a-f0-9]{64}', result.get('public_signing_spki_sha256','')):
raise Refused()
# Closed receipt allowlist: never copy arbitrary worker output.
receipt.update({k: True for k in required})
receipt['public_signing_spki_sha256'] = result['public_signing_spki_sha256']
receipt['status'] = 'isolated_keycape_native_registration_checked'
finally:
try:
if cleanup_session:
cleanup(session, receipt)
receipt['container_cleanup_verified'] = True
except Exception:
receipt['status'] = 'failed'
raise