coulomb/net-kingdom
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(recovery): add isolated KeyCape restore and loss cleanup proof
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a1468b-63fa-7710-85dc-8f2e16b3ebfa
This commit is contained in:
tegwick 2026-10-16 23:35:54 +02:00
parent 85a54ab9da
commit 5fcd5952ad
9 changed files with 1985 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

118
tests/isolated-keycape-native-fixture-manifest.json Normal file
View file

@ -0,0 +1,118 @@
{
"schema": "nk49.keycape-native-fixture-proposal.v1",
"status": "prepared_not_executed_not_approved",
"source_sha256": {
"sso-mfa/bootstrap/sops-private.py": "sha256:113d17ee4cd7ad405e73cc51e1823dc02bd2e618e7fa67c1b624f0226e334ec6",
"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:ba274034cd8491089048ad54324be735b7869d54b9d916330e987d7c288531f7",
"tools/bootstrap-recovery/isolated-keycape.py": "sha256:1c018fd7fbfd3afcc96ae61e7eafc283ac1dde3916c3bf7eac9eb50b934b57a5",
"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:fba8f6c2247f0ff951663a8d6372aa2276ca21aaac6f77fbfdb7f730f525fefa",
"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"
},
"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",
"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"
]
}

221
tests/run_isolated_keycape_native_fixture.py Normal file
View file

@ -0,0 +1,221 @@
#!/usr/bin/env python3
"""Prepare/run the exact reviewed T18 candidate on disposable synthetic material.
Default --plan performs no Docker calls and generates no key/material. Execution
requires the reviewed fixture manifest hash and source closure. This runner calls
the candidate itself; hooks only pause or inject a post-response transport fault.
"""
import argparse
import base64
from datetime import datetime,timedelta,timezone
import hashlib
import importlib.util
import io
import json
import os
import re
from pathlib import Path
import signal
import subprocess
import sys
import tarfile
import tempfile
import time
import uuid
ROOT=Path(__file__).resolve().parents[1]
MANIFEST=ROOT/'tests/isolated-keycape-native-fixture-manifest.json'
ISSUER='https://nk49-fixture.invalid'
CASES=('positive','hard-parent-loss','deadline','nonzero-after-accept')
def load(name,path):
spec=importlib.util.spec_from_file_location(name,path)
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);return module
s=load('session',ROOT/'tools/bootstrap-recovery/isolated-keycape-session.py')
def require(condition):
if not condition:raise RuntimeError('synthetic_fixture_refused')
def reviewed_manifest(expected):
raw=MANIFEST.read_bytes();require(hashlib.sha256(raw).hexdigest()==expected)
manifest=json.loads(raw)
for name,digest in manifest['source_sha256'].items():
require(isinstance(digest,str) and re.fullmatch('sha256:[a-f0-9]{64}',digest))
require('sha256:'+hashlib.sha256((ROOT/name).read_bytes()).hexdigest()==digest)
return manifest
def private_write(path,value):
path.parent.mkdir(mode=0o700,parents=True,exist_ok=True)
fd=os.open(path,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
with os.fdopen(fd,'wb') as out:out.write(value)
def materials(root):
"""Reuse the existing full archive fixture factory, no real archive accepted."""
from cryptography import x509
from cryptography.hazmat.primitives import hashes,serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.x509.oid import NameOID
import yaml
existing=load('existing_bundle_fixture',ROOT/'tests/test_identity_bundle.py')
payload=existing.stage.__wrapped__(root)
config={'issuer':ISSUER,'port':8080,'privateKeyPem':'/etc/keycape/key.pem','environment':'dev','tokenLifetime':'5m',
'lldap':{'url':'ldap://127.0.0.1:1','bindDN':'cn=fixture','bindPW':'synthetic-only','baseDN':'dc=fixture'},
'authelia':{'baseURL':'http://127.0.0.1:1','clientId':'synthetic','clientSecret':'synthetic','redirectURI':ISSUER+'/authorize/callback'},
'privacyidea':{'baseURL':'http://127.0.0.1:1'},
'clients':[{'clientId':'nk49-fixture-consumer','clientType':'confidential','audience':'nk49-fixture',
'grantTypes':['client_credentials'],'allowedScopes':['fixture:read'],'secretRef':'env:KEYCAPE_FIXTURE_CLIENT_SECRET',
'serviceSubject':'service:nk49-fixture','tenant':'tenant:fixture','roles':['fixture-reader']}]+
[{'clientId':name,'displayName':name,'clientType':'public','grantTypes':['authorization_code'],
'redirectUris':['https://nk49-fixture.invalid/callback'],'allowedScopes':['openid'],'mfaRequired':True}
for name in ('vergabe-demo-company','user-engine-portal')]}
secret_path=payload/'resources/sso/secret/keycape-config.json';secret=json.loads(secret_path.read_text())
secret['data']['config.yaml']=base64.b64encode(yaml.safe_dump(config).encode()).decode();secret_path.write_text(json.dumps(secret))
client_path=payload/'resources/sso/secret/keycape-rapp-qonto-client.json';client=json.loads(client_path.read_text())
client['data']={'client-secret':base64.b64encode(b'synthetic-fixture-client-secret').decode()};client_path.write_text(json.dumps(client))
factor_path=payload/'resources/sso/secret/keycape-factor-read.json';factor=json.loads(factor_path.read_text())
factor['data']={'admin-token':base64.b64encode(b'synthetic-unused-factor-reader').decode()};factor_path.write_text(json.dumps(factor))
dep_path=payload/'resources/sso/deployment/keycape.json';dep=json.loads(dep_path.read_text())
dep['spec']={'template':{'spec':{'containers':[{'name':'keycape','image':s.adapter.IMAGE,'env':[
{'name':'KEYCAPE_CONFIG','value':'/etc/keycape/config.yaml'},
{'name':'KEYCAPE_POLICY_PATH','value':'/var/lib/keycape-policy/policy.json'},
{'name':'KEYCAPE_FIXTURE_CLIENT_SECRET','valueFrom':{'secretKeyRef':{'name':'keycape-rapp-qonto-client','key':'client-secret'}}}]}]}}};dep_path.write_text(json.dumps(dep))
policy={'Revision':1,'Modes':{k:'mandatory' for k in ('vergabe-demo-company','user-engine-portal')},'History':[
{'reference':'synthetic-retained-receipt','actor':'service:fixture','client':'user-engine-portal','before':'mandatory','after':'mandatory','revision':1,'at':1}]}
# JSON tags in the owner store are lowercase; retain an existing lock inode
# representation as part of the archived tree, never synthesize it after import.
policy={k.lower():v for k,v in policy.items()};(payload/'data/keycape/policy.json').write_text(json.dumps(policy))
private_write(payload/'data/keycape/policy.json.lock',b'')
key=rsa.generate_private_key(public_exponent=65537,key_size=2048)
subject=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'nk49-fixture.invalid')]);now=datetime.now(timezone.utc)
cert=(x509.CertificateBuilder().subject_name(subject).issuer_name(subject).public_key(key.public_key()).serial_number(x509.random_serial_number())
.not_valid_before(now-timedelta(minutes=5)).not_valid_after(now+timedelta(days=1))
.add_extension(x509.SubjectAlternativeName([x509.DNSName('nk49-fixture.invalid')]),critical=False)
.add_extension(x509.BasicConstraints(ca=True,path_length=None),critical=True).sign(key,hashes.SHA256()))
tls={'/recovery-tls/server.pem':cert.public_bytes(serialization.Encoding.PEM),
'/recovery-tls/server-key.pem':key.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption()),
'/recovery-tls/trust.pem':cert.public_bytes(serialization.Encoding.PEM)}
for path,value in tls.items():private_write(root/Path(path).name,value)
age=root/'synthetic-age.key';s.run(['age-keygen','-o',str(age)])
recipient=s.run(['age-keygen','-y',str(age)]).decode().strip();profile=s.bundle.profile();profile['public_recipient']=recipient
s.bundle.profile=lambda:profile
artifact=root/'net-kingdom-identity-20261016T063159Z-0123456789ab.tar.age';s.bundle.seal(payload,artifact,{'synthetic_native_keycape_fixture':True})
raw=s.run(['age','-d','-i',str(age),str(artifact)],max_output=s.bundle.LIMIT)
with tarfile.open(fileobj=io.BytesIO(raw)) as archive:manifest=json.load(archive.extractfile('manifest.json'))
transfer={'schema':'netkingdom.identity-offsite.v1','status':'readback_verified','provider':'Storage Share',
'folder':'Backup/railiance-backups/net-kingdom-identity','artifact':artifact.name,'ciphertext_bytes':artifact.stat().st_size,
'ciphertext_sha256':hashlib.sha256(artifact.read_bytes()).hexdigest(),'etag':'"synthetic"','fetched_path':str(artifact),
'independent_owner_retrieval_verified':True}
# Derive public executable identity only from the exact cached admitted image.
probe=s.plan_session();name=probe['public_image_probe']
try:
s.run(['docker','create','--pull=never','--name',name,'--network','none','--read-only','--label',s.adapter.LABEL+'='+probe['run_id'],s.adapter.IMAGE])
binary=s.adapter.container_file(name,'/keycape')
finally:s.adapter.cleanup({**probe,'containers':[name]}, {})
admission={'issuer':ISSUER,'ciphertext_sha256':transfer['ciphertext_sha256'],
'manifest_sha256':hashlib.sha256(json.dumps(manifest,sort_keys=True,separators=(',',':')).encode()).hexdigest(),
'verifier_sha256':hashlib.sha256(binary).hexdigest(),'trust_sha256':hashlib.sha256(tls['/recovery-tls/trust.pem']).hexdigest(),
'tls_cert_sha256':hashlib.sha256(tls['/recovery-tls/server.pem']).hexdigest(),
'consumer':{'client_id':'nk49-fixture-consumer','secret_env':'KEYCAPE_FIXTURE_CLIENT_SECRET','audience':'nk49-fixture','scope':'fixture:read',
'subject':'service:nk49-fixture','tenant':'tenant:fixture','roles':'fixture-reader','deny_scope':'fixture:write'}}
private_write(root/'fixture.json',json.dumps({'schema':'nk49.generated-keycape-fixture.v1','transfer':transfer,'admission':admission,'recipient':recipient}).encode())
def worker(root,case):
execution=s.isolate_execution();execution['deadline']=time.monotonic()+(20 if case=='deadline' else 120)
data=json.loads((root/'fixture.json').read_text());require(data['schema']=='nk49.generated-keycape-fixture.v1')
profile=s.bundle.profile();profile['public_recipient']=data['recipient'];s.bundle.profile=lambda:profile
s.local.verify.custody.private_key_fd=lambda:os.open(root/'synthetic-age.key',os.O_RDONLY|os.O_NOFOLLOW)
tls={'/recovery-tls/'+name:(root/name).read_bytes() for name in ('server.pem','server-key.pem','trust.pem')}
receipt={};output=root/(case+'.json');checkpoint=lambda r:output.write_text(json.dumps(r))
provision=s.provision
def paused(*args):
provision(*args)
if case in ('hard-parent-loss','deadline'):
private_write(root/(case+'.barrier'),b'private-import-and-default-launch-complete')
time.sleep(150)
s.provision=paused
if case=='nonzero-after-accept':
original=s.adapter.subprocess.run;injected=False
def fault(argv,**kwargs):
nonlocal injected
result=original(argv,**kwargs)
if argv[:3]==['docker','--host',s.adapter.ENDPOINT] and 'run' in argv and not injected and result.returncode==0:
injected=True
return subprocess.CompletedProcess(argv,1,b'',b'')
return result
s.adapter.subprocess.run=fault
try:s.execute(data['transfer'],data['admission'],tls,receipt,execution,checkpoint=checkpoint)
except BaseException:receipt['status']='failed'
finally:output.write_text(json.dumps(receipt))
return int(receipt.get('status')!='isolated_keycape_native_registration_checked')
def run_cases(root,summary,cases):
for case in cases:
child=subprocess.Popen([sys.executable,str(Path(__file__).resolve()),'--worker-root',str(root),'--worker-case',case,
'--execute-reviewed-fixture','--expected-manifest-sha256',summary['manifest_sha256']],stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
try:
end=time.monotonic()+150
if case=='hard-parent-loss':
while not (root/(case+'.barrier')).exists() and child.poll() is None and time.monotonic()<end:time.sleep(.05)
require((root/(case+'.barrier')).exists())
execution=json.loads((root/(case+'.json')).read_text())['execution_binding'];require(execution['pid']==child.pid)
fd=os.pidfd_open(child.pid)
try:signal.pidfd_send_signal(fd,signal.SIGKILL)
finally:os.close(fd)
child.wait(timeout=max(1,end-time.monotonic()))
receipt=json.loads((root/(case+'.json')).read_text())
summary['receipts'][case]=receipt
if case=='deadline':require((root/(case+'.barrier')).exists())
if case=='positive':
require(receipt.get('status')=='isolated_keycape_native_registration_checked' and receipt.get('container_cleanup_verified') and receipt.get('temporary_plaintext_cleanup_verified'))
require(receipt.get('native_service_registration_verified') and receipt.get('original_public_signing_key_in_live_jwks'))
summary['actual_issuer_original_hostname_tls_verified']=True
else:
require(receipt.get('status')!='isolated_keycape_native_registration_checked')
outcome=Path(receipt['reaper_outcome_path'])
while not outcome.exists() and time.monotonic()<end:time.sleep(.05)
require(outcome.exists());cleanup=json.loads(outcome.read_text())
require(cleanup.get('execution_quiescence_verified') and cleanup.get('temporary_plaintext_cleanup_verified'))
if case=='nonzero-after-accept':require(cleanup.get('status')=='bounded_cleanup_uncertain' and cleanup.get('in_flight_daemon_request_uncertain'))
else:require(cleanup.get('status')=='bounded_cleanup_verified')
summary['cases'][case]='passed'
finally:
if child.poll() is None:
fd=os.pidfd_open(child.pid)
try:signal.pidfd_send_signal(fd,signal.SIGKILL)
finally:os.close(fd)
child.wait(timeout=5)
def main():
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--plan',action='store_true')
p.add_argument('--execute-reviewed-fixture',action='store_true');p.add_argument('--expected-manifest-sha256')
p.add_argument('--worker-root');p.add_argument('--worker-case',choices=CASES);p.add_argument('--case',choices=(*CASES,'all'),default='positive');args=p.parse_args()
if not args.execute_reviewed_fixture:
print(MANIFEST.read_text());return 0
require(not args.plan);manifest=reviewed_manifest(args.expected_manifest_sha256);os.umask(0o077)
if args.worker_root:
root=Path(args.worker_root);require(root.parent==Path('/dev/shm') and root.name.startswith('nk49-keycape-fixture-'))
s.bundle.private(root,True);return worker(root,args.worker_case)
require(not args.worker_case)
s.local.preflight();s.run(['docker','image','inspect',s.adapter.IMAGE,'--format','{{.Id}}'])
safe=Path(tempfile.mkdtemp(prefix='nk49-keycape-native-fixture-review-',dir='/tmp'))
summary={'schema':'nk49.synthetic-keycape-native-fixture.v1','status':'failed','manifest_sha256':args.expected_manifest_sha256,
'cases':{},'receipts':{},'actual_credentials_used':False,'actual_recovery_proven':False,'docker_executed':True}
print('Safe synthetic fixture result: '+str(safe/'result.json'),flush=True)
try:
with tempfile.TemporaryDirectory(prefix='nk49-keycape-fixture-',dir='/dev/shm') as temporary:
root=Path(temporary);materials(root);run_cases(root,summary,CASES if args.case=='all' else (args.case,))
summary['status']='synthetic_native_fixture_passed'
finally:(safe/'result.json').write_text(json.dumps(summary,indent=2)+'\n')
print('Safe synthetic fixture result: '+str(safe/'result.json'))
return int(summary['status']!='synthetic_native_fixture_passed')
if __name__=='__main__':raise SystemExit(main())

49
tests/test_isolated_keycape_fixture_plan.py Normal file
View file

@ -0,0 +1,49 @@
import contextlib
import hashlib
import importlib.util
import io
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
ROOT=Path(__file__).resolve().parents[1]
spec=importlib.util.spec_from_file_location('fixture_runner',ROOT/'tests/run_isolated_keycape_native_fixture.py')
f=importlib.util.module_from_spec(spec);spec.loader.exec_module(f)
class FixturePreparationTests(unittest.TestCase):
def test_default_plan_neither_runs_docker_nor_generates_materials(self):
output=io.StringIO()
with patch.object(sys,'argv',['fixture','--plan']),patch.object(f.s,'run') as run,patch.object(f,'materials') as materials,patch.object(f.s.local,'preflight') as preflight,contextlib.redirect_stdout(output):
self.assertEqual(f.main(),0)
run.assert_not_called();materials.assert_not_called();preflight.assert_not_called()
manifest=json.loads(output.getvalue())
self.assertEqual(manifest['status'],'prepared_not_executed_not_approved')
self.assertEqual(manifest['fixed_endpoint'],'unix:///var/run/docker.sock')
def test_wrong_reviewed_manifest_hash_stops_before_preflight_or_generation(self):
with patch.object(sys,'argv',['fixture','--execute-reviewed-fixture','--expected-manifest-sha256','0'*64]),patch.object(f.s,'run') as run,patch.object(f,'materials') as materials,patch.object(f.s.local,'preflight') as preflight:
with self.assertRaises(RuntimeError):f.main()
run.assert_not_called();materials.assert_not_called();preflight.assert_not_called()
def test_untyped_or_malformed_public_digest_refused(self):
for invalid in ('0'*64, 'sha256:'+'A'*64, 'sha256:'+'0'*63, None):
with self.subTest(invalid_type=type(invalid).__name__), tempfile.TemporaryDirectory() as temporary:
manifest=json.loads(f.MANIFEST.read_text())
first=next(iter(manifest['source_sha256']))
manifest['source_sha256'][first]=invalid
path=Path(temporary)/'manifest.json'
path.write_text(json.dumps(manifest))
with patch.object(f,'MANIFEST',path), patch.object(f.s,'run') as run:
with self.assertRaises(RuntimeError):
f.reviewed_manifest(hashlib.sha256(path.read_bytes()).hexdigest())
run.assert_not_called()
def test_manifest_binds_current_candidate_and_existing_fixture_factory(self):
digest=hashlib.sha256(f.MANIFEST.read_bytes()).hexdigest()
manifest=f.reviewed_manifest(digest)
self.assertIn('tests/test_identity_bundle.py',manifest['source_sha256'])
self.assertIn('tools/bootstrap-recovery/isolated-keycape-session.py',manifest['source_sha256'])
self.assertIn('sso-mfa/k8s/privacyidea/key-custody-remote.py',manifest['source_sha256'])
if __name__=='__main__':unittest.main()

234
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()

292
tests/test_isolated_keycape_session.py Normal file
View file

@ -0,0 +1,292 @@
import base64
import importlib.util
import json
import io
import os
import tempfile
import signal
import subprocess
import sys
import time
from pathlib import Path
import unittest
from unittest.mock import patch
SOURCE=Path(__file__).parents[1]/'tools/bootstrap-recovery/isolated-keycape-session.py'
spec=importlib.util.spec_from_file_location('session',SOURCE)
s=importlib.util.module_from_spec(spec);spec.loader.exec_module(s)
class SessionTests(unittest.TestCase):
def test_tmpfs_plan_has_four_exact_original_mounts(self):
session=s.plan_session()
self.assertEqual({v['target'] for v in session['volumes'].values()},set(s.RUNROOTS))
self.assertTrue(all(v['options']==s.VOLUME_OPTIONS for v in session['volumes'].values()))
self.assertEqual(session['images'][session['issuer']],s.adapter.IMAGE)
def test_non_tmpfs_driver_options_refused(self):
session=s.plan_session();name,spec=next(iter(session['volumes'].items()))
wrong=[{'Name':name,'Driver':'local','Scope':'local','Options':{'type':'none','device':'/host/data'},'Labels':{s.adapter.LABEL:session['run_id']}}]
with patch.object(s,'run',return_value=json.dumps(wrong).encode()):
with self.assertRaises(s.adapter.Refused):s.verify_volume(name,spec,session['run_id'])
def test_owner_secret_is_not_docker_argv_and_import_is_after_inspection(self):
session=s.plan_session();events=[]
payload={'files':{'/etc/keycape/config.yaml':base64.b64encode(b'original').decode()},'env':{'KEYCAPE_TEST_SECRET':'fixture-sensitive'}}
def run(argv,**kwargs):
events.append(('run',argv,kwargs))
if argv[-1]=='{{.Id}}':return b'anchor'
return b''
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session',side_effect=lambda *args,**kwargs:events.append(('inspect',args,kwargs))):
s.provision(session,payload,{'/recovery-tls/trust.pem':b'original-trust'},'https://issuer.example')
joined=' '.join(' '.join(e[1]) for e in events if e[0]=='run')
self.assertNotIn('fixture-sensitive',joined)
private_index=next(i for i,e in enumerate(events) if e[0]=='run' and s.IMPORT in e[1])
inspect_index=next(i for i,e in enumerate(events) if e[0]=='inspect')
self.assertLess(inspect_index,private_index)
issuer_call=next(e for e in events if e[0]=='run' and e[1][-1]==s.adapter.IMAGE)
self.assertEqual(issuer_call[2]['env']['KEYCAPE_TEST_SECRET'],'fixture-sensitive')
self.assertIn('KEYCAPE_TEST_SECRET',issuer_call[1])
self.assertNotIn('--entrypoint',issuer_call[1])
def test_pinned_helper_image_volume_is_explicitly_masked_with_tmpfs(self):
calls=[]
def run(argv,**kwargs):calls.append(argv);return b'anchor' if argv[-1]=='{{.Id}}' else b''
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session'):
s.provision(s.plan_session(),{'files':{},'env':{}},{},'https://issuer.example')
worker=next(argv for argv in calls if s.WORKER in argv)
mask='/privacyidea/etc/persistent:rw,nosuid,nodev,noexec,size=16777216,uid=1000,gid=1000,mode=0700'
self.assertIn(mask,worker)
self.assertEqual(worker[worker.index(mask)-1],'--tmpfs')
issuer=next(argv for argv in calls if argv[-1]==s.adapter.IMAGE)
self.assertNotIn(mask,issuer)
def test_public_phase_sequence_reports_containment_before_private_import(self):
session=s.plan_session();phases=[]
def run(argv,**kwargs):
return b'anchor' if argv[-1]=='{{.Id}}' else b''
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session'):
s.provision(session,{'files':{},'env':{}},{},'https://issuer.example',phases.append)
self.assertEqual(phases,list(s.PROVISION_PHASES))
def test_worker_containment_refusal_keeps_exact_phase_and_skips_import(self):
phases=[];calls=[]
def run(argv,**kwargs):calls.append(argv);return b''
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session',side_effect=s.adapter.Refused):
with self.assertRaises(s.adapter.Refused):
s.provision(s.plan_session(),{'files':{},'env':{}},{},'https://issuer.example',phases.append)
self.assertEqual(phases[-1],'worker_containment')
self.assertFalse(any(s.IMPORT in argv for argv in calls))
def test_issuer_containment_refusal_does_not_claim_provision_complete(self):
phases=[]
def run(argv,**kwargs):return b'anchor' if argv[-1]=='{{.Id}}' else b''
with patch.object(s,'run',side_effect=run),patch.object(s,'verify_volume'),patch.object(s.adapter,'inspect_session',side_effect=[None,s.adapter.Refused()]):
with self.assertRaises(s.adapter.Refused):
s.provision(s.plan_session(),{'files':{},'env':{}},{},'https://issuer.example',phases.append)
self.assertEqual(phases[-1],'issuer_containment')
self.assertNotIn('complete',phases)
def test_non_original_https_port_refused_before_any_mutation(self):
with patch.object(s,'run') as run:
with self.assertRaises(s.adapter.Refused):s.provision(s.plan_session(),{'files':{},'env':{}},{},'http://issuer.example')
run.assert_not_called()
def test_container_failure_still_attempts_each_volume_cleanup(self):
session=s.plan_session();receipt={};removed=[]
def run(argv,**kwargs):
if argv[1:3]==['volume','rm']:removed.append(argv[-1])
if argv[1:3]==['volume','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(s.adapter,'cleanup',side_effect=s.adapter.Refused),patch.object(s,'verify_volume'),patch.object(s,'run',side_effect=run):
with self.assertRaises(s.adapter.Refused):s.cleanup_all(session,receipt)
self.assertEqual(set(removed),set(session['volumes']))
self.assertFalse(receipt['container_cleanup_verified'])
self.assertEqual(len(receipt['tmpfs_volume_cleanup_obligations']),4)
def test_reaper_inherits_no_custody_or_service_environment(self):
session=s.plan_session();staging={'path':'/dev/shm/registered','run_id':session['run_id']}
with patch.object(s.subprocess,'Popen') as spawn,patch.object(s,'verify_execution'),patch.object(s.os,'pidfd_open',return_value=99),patch.object(s.os,'close'),patch.object(s,'register_workdir',return_value=staging),patch.object(s.select,'select',side_effect=lambda *args:([spawn.return_value.stdout],[],[])):
spawn.return_value.stdout.readline.return_value=b'registered\n'
s.start_reaper(session,staging,{'pid':123})
kwargs=spawn.call_args.kwargs
self.assertEqual(set(kwargs['env']),{'PATH','DOCKER_HOST'})
self.assertTrue(kwargs['close_fds']);self.assertTrue(kwargs['start_new_session']);self.assertEqual(kwargs['pass_fds'],(99,))
class TargetBindingTests(unittest.TestCase):
def test_every_direct_docker_command_is_explicitly_bound(self):
with patch.dict('os.environ',{},clear=True),patch.object(s.adapter.subprocess,'run',return_value=__import__('subprocess').CompletedProcess([],0,b'ok')) as run:
self.assertEqual(s.run(['docker','info']),b'ok')
argv=run.call_args.args[0]
self.assertEqual(argv[:3],['docker','--host',s.adapter.ENDPOINT])
self.assertEqual(run.call_args.kwargs['env']['DOCKER_HOST'],s.adapter.ENDPOINT)
def test_nondefault_unix_socket_is_refused_before_any_docker_call(self):
with patch.dict('os.environ',{'DOCKER_HOST':'unix:///tmp/other-daemon.sock'},clear=True),patch.object(s.adapter.subprocess,'run') as run:
with self.assertRaises(s.adapter.Refused):s.run(['docker','info'])
run.assert_not_called()
def test_conflicting_context_is_refused(self):
with patch.dict('os.environ',{'DOCKER_CONTEXT':'production'},clear=True),patch.object(s.adapter.subprocess,'run') as run:
with self.assertRaises(s.adapter.Refused):s.run(['docker','info'])
run.assert_not_called()
def test_conflicting_per_call_host_is_refused(self):
with patch.dict('os.environ',{},clear=True),patch.object(s.adapter.subprocess,'run') as run:
with self.assertRaises(s.adapter.Refused):s.run(['docker','info'],env={'DOCKER_HOST':'unix:///tmp/other.sock'})
run.assert_not_called()
def test_reused_preflight_uses_same_bound_command(self):
self.assertIs(s.local.command,s.adapter.command)
def test_probe_is_registered_before_session_creation(self):
session=s.plan_session()
self.assertEqual(session['public_image_probe'],'nk49-kc-'+session['run_id']+'-image-probe')
self.assertEqual(session['docker_endpoint'],s.adapter.ENDPOINT)
class PlaintextLifecycleTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory();self.addCleanup(self.tmp.cleanup)
self.root=Path(self.tmp.name)
self.override=patch.object(s,'STAGING_ROOT',self.root);self.override.start();self.addCleanup(self.override.stop)
self.run_id='a'*32
self.work=self.root/('nk49-keycape-'+self.run_id+'-abcdefgh');self.work.mkdir(mode=0o700)
self.staging=s.register_workdir(self.work,self.run_id)
def test_exact_plaintext_directory_removed_and_recorded(self):
(self.work/'archive.tar').write_bytes(b'synthetic-private-archive')
restored=self.work/'restored';restored.mkdir(mode=0o700);(restored/'fixture').write_bytes(b'synthetic')
receipt={};s.cleanup_workdir(self.staging,receipt)
self.assertFalse(self.work.exists());self.assertTrue(receipt['temporary_plaintext_cleanup_verified'])
self.assertEqual(receipt['plaintext_cleanup_obligation']['status'],'absence_verified')
def test_docker_failure_does_not_skip_plaintext_cleanup(self):
(self.work/'archive.tar').write_bytes(b'synthetic')
receipt={}
with patch.object(s,'cleanup_all',side_effect=s.adapter.Refused):
with self.assertRaises(s.adapter.Refused):s.recover_obligations({},self.staging,receipt)
self.assertFalse(self.work.exists());self.assertTrue(receipt['temporary_plaintext_cleanup_verified'])
def test_replacement_inode_is_not_removed(self):
# Preserve original inode while replacing its registered pathname.
old=self.root/'retained-original';self.work.rename(old);self.work.mkdir(mode=0o700)
(self.work/'replacement').write_bytes(b'not-owned-by-this-obligation');receipt={}
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(self.staging,receipt)
self.assertTrue((self.work/'replacement').exists());self.assertTrue(old.exists())
self.assertEqual(receipt['plaintext_cleanup_obligation']['status'],'identity_mismatch_not_removed')
def test_symlink_root_is_not_followed(self):
self.work.rmdir();outside=self.root/'outside';outside.mkdir();(outside/'kept').write_bytes(b'keep')
self.work.symlink_to(outside,target_is_directory=True)
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(self.staging,{})
self.assertTrue((outside/'kept').exists())
def test_descendant_symlink_is_not_followed(self):
outside=self.root/'kept';outside.write_bytes(b'keep');(self.work/'link').symlink_to(outside)
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(self.staging,{})
self.assertEqual(outside.read_bytes(),b'keep')
def test_wrong_owner_identity_is_not_removed(self):
changed={**self.staging,'owner_uid':os.getuid()+1}
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir(changed,{})
self.assertTrue(self.work.exists())
def test_unregistered_path_refused(self):
with self.assertRaises(s.adapter.Refused):s.cleanup_workdir({**self.staging,'path':str(self.root)}, {})
@unittest.skipUnless(hasattr(os,'pidfd_open') and hasattr(signal,'pidfd_send_signal'),'Linux pidfd required')
class ExecutionDeadlineTests(unittest.TestCase):
def start_synthetic_owner(self,root):
late=str(root/'late-creation')
code="""import importlib.util,json,subprocess,sys,time
spec=importlib.util.spec_from_file_location('session',sys.argv[1]);s=importlib.util.module_from_spec(spec);spec.loader.exec_module(s)
execution=s.isolate_execution()
child=subprocess.Popen([sys.executable,'-c',"import pathlib,sys,time;time.sleep(2);pathlib.Path(sys.argv[1]).write_text('late')",sys.argv[2]])
print(json.dumps({'execution':execution,'child_pid':child.pid}),flush=True)
time.sleep(20)
"""
parent=subprocess.Popen([sys.executable,'-c',code,str(SOURCE),late],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL)
message=json.loads(parent.stdout.readline());parent.stdout.close()
def finish():
if parent.poll() is None:parent.kill()
parent.wait(timeout=5)
try:
before=s.process_facts(message['child_pid'])
if before is None or before['sid']!=message['execution']['sid'] or before['pgid']!=message['execution']['pgid']:return
fd=os.pidfd_open(message['child_pid'])
try:
after=s.process_facts(message['child_pid'])
if after is not None and all(after[k]==before[k] for k in ('pid','sid','pgid','start_ticks','uid')):
signal.pidfd_send_signal(fd,signal.SIGKILL)
except ProcessLookupError:pass
finally:os.close(fd)
except ProcessLookupError:pass
self.addCleanup(finish)
return parent,message['execution'],Path(late)
def test_deadline_quiesces_only_fresh_owner_group_before_cleanup(self):
with tempfile.TemporaryDirectory() as directory:
parent,execution,late=self.start_synthetic_owner(Path(directory));fd=os.pidfd_open(parent.pid)
try:
receipt={};s.quiesce_execution(execution,fd,receipt)
self.assertTrue(receipt['execution_quiescence_verified'])
self.assertFalse(s.group_members(execution));self.assertFalse(late.exists())
finally:os.close(fd)
def test_hard_parent_loss_stops_surviving_marker_creator(self):
with tempfile.TemporaryDirectory() as directory:
parent,execution,late=self.start_synthetic_owner(Path(directory));fd=os.pidfd_open(parent.pid)
try:
signal.pidfd_send_signal(fd,signal.SIGKILL)
poll=__import__('select').poll();poll.register(fd,__import__('select').POLLIN);self.assertTrue(poll.poll(2000))
receipt={};s.quiesce_execution(execution,fd,receipt)
self.assertTrue(receipt['execution_quiescence_verified']);self.assertFalse(late.exists())
self.assertFalse(s.group_members(execution))
finally:os.close(fd)
def test_expired_execution_cannot_start_any_docker_command(self):
binding={'pid':os.getpid(),'sid':os.getsid(0),'pgid':os.getpgrp(),'deadline':time.monotonic()-1,'staging':'/tmp/not-created'}
with patch.object(s.adapter,'EXECUTION',binding),patch.object(s.adapter.subprocess,'run') as run:
with self.assertRaises(s.adapter.Refused):s.run(['docker','create','fixed-image'])
run.assert_not_called()
def test_reused_or_foreign_session_refuses_signals(self):
binding={'pid':123,'sid':123,'pgid':123,'start_ticks':10,'uid':os.getuid()}
wrong={**binding,'sid':456,'state':'S'}
with patch.object(s,'process_facts',return_value=wrong),patch.object(s.signal,'pidfd_send_signal') as send:
with self.assertRaises(s.adapter.Refused):s.quiesce_execution(binding,99,{})
send.assert_not_called()
def test_inflight_uncertainty_survives_plaintext_directory_removal(self):
with tempfile.TemporaryDirectory() as directory:
path=Path(directory)/'owned-stage';path.mkdir();marker=Path(str(path)+'-docker-inflight');marker.touch()
staging={'path':str(path)};self.assertTrue(s.daemon_uncertain(staging));path.rmdir()
self.assertTrue(s.daemon_uncertain(staging))
class ReaperOrderingTests(unittest.TestCase):
def exercise(self,quiescence_fails=False):
events=[];records=[]
def quiet(*args):
events.append('quiesce')
if quiescence_fails:raise s.adapter.Refused()
args[-1]['execution_quiescence_verified']=True
def recover(*args):events.append('cleanup')
with patch.object(s,'verify_execution'),patch.object(s,'pidfd_dead',return_value=False),patch.object(s.select,'poll') as poll,patch.object(s,'quiesce_execution',side_effect=quiet),patch.object(s,'daemon_uncertain',return_value=True),patch.object(s,'recover_obligations',side_effect=recover),patch.object(s.os,'close'),patch.object(s.os,'open',return_value=99),patch.object(s.os,'fdopen'),patch.object(s.json,'dump',side_effect=lambda r,*args:records.append(dict(r))),patch('sys.stdout',io.StringIO()):
poll.return_value.poll.return_value=[]
s.reaper_main({}, {'path':'/dev/shm/exact-staging'}, {'deadline':time.monotonic()-1},77)
return events,records[0]
def test_daemon_uncertainty_survives_verified_resource_readback(self):
events,receipt=self.exercise();self.assertEqual(events,['quiesce','cleanup'])
self.assertTrue(receipt['execution_quiescence_verified']);self.assertTrue(receipt['in_flight_daemon_request_uncertain'])
self.assertEqual(receipt['status'],'bounded_cleanup_uncertain')
def test_unverified_quiescence_prevents_any_cleanup(self):
events,receipt=self.exercise(True);self.assertEqual(events,['quiesce'])
self.assertFalse(receipt['execution_quiescence_verified']);self.assertEqual(receipt['status'],'bounded_cleanup_uncertain')
def test_checkpoint_rewrites_one_valid_json_document(self):
with tempfile.TemporaryFile() as stream:
s.persist_receipt(stream.fileno(),{'status':'registered','long':'x'*100})
s.persist_receipt(stream.fileno(),{'status':'failed'})
stream.seek(0);self.assertEqual(json.load(stream),{'status':'failed'})
def test_nonzero_docker_result_retains_uncertainty_before_inflight_clear(self):
with tempfile.TemporaryDirectory() as directory:
staging=str(Path(directory)/'stage');Path(staging).mkdir()
binding={'pid':os.getpid(),'sid':os.getsid(0),'pgid':os.getpgrp(),'deadline':time.monotonic()+10,'staging':staging}
def returned(*args,**kwargs):
self.assertTrue(Path(staging+'-docker-inflight').exists())
return subprocess.CompletedProcess(args[0],1,b'')
with patch.object(s.adapter,'EXECUTION',binding),patch.object(s.adapter.subprocess,'run',side_effect=returned):
with self.assertRaises(s.adapter.Refused):s.run(['docker','create','synthetic-already-accepted'])
self.assertFalse(Path(staging+'-docker-inflight').exists());self.assertTrue(s.daemon_uncertain({'path':staging}))
def test_only_policy_mount_is_writable_in_issuer(self):
plan=s.plan_session();args=s.mounts(plan,readonly=True)
mounts=args[1::2]
for value in mounts:
self.assertEqual(not value.endswith(',readonly'),'target=/var/lib/keycape-policy' in value)
def test_docker_timeout_retains_independent_uncertainty_marker(self):
with tempfile.TemporaryDirectory() as directory:
staging=str(Path(directory)/'stage');Path(staging).mkdir()
binding={'pid':os.getpid(),'sid':os.getsid(0),'pgid':os.getpgrp(),'deadline':time.monotonic()+10,'staging':staging}
def timeout(*args,**kwargs):
self.assertTrue(Path(staging+'-docker-inflight').exists())
raise subprocess.TimeoutExpired('synthetic Docker',1)
with patch.object(s.adapter,'EXECUTION',binding),patch.object(s.adapter.subprocess,'run',side_effect=timeout):
with self.assertRaises(subprocess.TimeoutExpired):s.run(['docker','create','fixed-image'])
self.assertFalse(Path(staging+'-docker-inflight').exists())
self.assertTrue(Path(staging+'-docker-uncertain').exists());self.assertTrue(s.daemon_uncertain({'path':staging}))
if __name__=='__main__':unittest.main()

509
tools/bootstrap-recovery/isolated-keycape-session.py Normal file
View file

@ -0,0 +1,509 @@
#!/usr/bin/env python3
"""Owner-run T18 session assembly using the existing archive/containment pipeline.
Candidate only: native pinned-image tests and review are required before use.
Shared Docker local-driver tmpfs is required because Docker cp does not support
private tmpfs/mount paths, and the original KeyCape image is distroless.
"""
import argparse
import base64
from datetime import datetime, timezone
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import signal
import re
import stat
import select
import subprocess
import sys
import tempfile
import time
import uuid
HERE=Path(__file__).resolve().parent
def load(name):
spec=importlib.util.spec_from_file_location(name,HERE/(name+'.py'))
module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);return module
adapter=load('isolated-keycape')
local=load('isolated-lldap')
bundle=local.bundle
# Bind the reused preflight itself, not only this module's direct Docker calls.
local.command=adapter.command
_original_preflight=local.preflight
def preflight():
contexts=json.loads(adapter.command(['docker','context','inspect']))
if len(contexts)!=1 or contexts[0]['Endpoints']['docker']['Host']!=adapter.ENDPOINT:raise adapter.Refused()
result=_original_preflight();result['docker_endpoint']=adapter.ENDPOINT;return result
local.preflight=preflight
WORKER=local.CLIENT
STAGING_ROOT=Path('/dev/shm')
RUNROOTS=('/etc/keycape','/var/lib/keycape-policy','/etc/keycape-factor','/recovery-tls')
VOLUME_OPTIONS={'type':'tmpfs','device':'tmpfs','o':'size=33554432,uid=1000,gid=1000,mode=0700,nosuid,nodev,noexec'}
IMPORT=r'''
import base64,json,os,sys
from pathlib import Path
os.umask(0o077)
p=json.load(sys.stdin)
allowed=('/etc/keycape/','/var/lib/keycape-policy/','/etc/keycape-factor/','/recovery-tls/')
for name,encoded in p.items():
f=Path(name)
if not name.startswith(allowed) or '..' in f.parts or any(x.is_symlink() for x in (f,*f.parents)):raise ValueError()
f.parent.mkdir(mode=0o700,parents=True,exist_ok=True)
fd=os.open(f,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
with os.fdopen(fd,'wb') as out:out.write(base64.b64decode(encoded,validate=True))
'''
# Small transport inside the already pinned helper; it has only loopback routes.
# No token handling, auth bypass, dependency imitation, reset or notifier path.
TLS=r'''
import select,socket,socketserver,ssl,threading
context=ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.minimum_version=ssl.TLSVersion.TLSv1_2
context.load_cert_chain('/recovery-tls/server.pem','/recovery-tls/server-key.pem')
class Handler(socketserver.BaseRequestHandler):
def handle(self):
try:
with context.wrap_socket(self.request,server_side=True) as incoming:
with socket.create_connection(('127.0.0.1',8080),timeout=10) as outgoing:
incoming.settimeout(15);outgoing.settimeout(15)
while True:
ready,_,_=select.select([incoming,outgoing],[],[],15)
if not ready:return
for source in ready:
data=source.recv(65536)
if not data:return
(outgoing if source is incoming else incoming).sendall(data)
except Exception:pass
class Server(socketserver.ThreadingTCPServer):
allow_reuse_address=True
daemon_threads=True
with Server(('127.0.0.1',443),Handler) as server:server.serve_forever()
'''
def run(argv,**kwargs):return adapter.command(argv,**kwargs)
def plan_session():
run_id=uuid.uuid4().hex
stem='nk49-kc-'+run_id
worker=stem+'-transport';issuer=stem+'-issuer'
volumes={stem+'-v'+str(i):{'target':root,'options':dict(VOLUME_OPTIONS)} for i,root in enumerate(RUNROOTS)}
return {'run_id':run_id,'containers':[worker,issuer], 'worker':worker,'issuer':issuer,
'images':{worker:WORKER,issuer:adapter.IMAGE},'volumes':volumes,
'public_image_probe':stem+'-image-probe','docker_endpoint':adapter.ENDPOINT}
def verify_volume(name,spec,run_id):
obj=json.loads(run(['docker','volume','inspect',name]))
if (len(obj)!=1 or obj[0]['Name']!=name or obj[0]['Driver']!='local' or obj[0].get('Scope')!='local'
or obj[0].get('Options')!=VOLUME_OPTIONS or spec['options']!=VOLUME_OPTIONS
or obj[0].get('Labels',{}).get(adapter.LABEL)!=run_id):raise adapter.Refused()
def base_args(session,name,network):
return ['docker','run','--detach','--pull=never','--name',name,'--network',network,
'--label',adapter.LABEL+'='+session['run_id'],'--read-only','--cap-drop','ALL',
'--security-opt','no-new-privileges','--user','1000:1000','--memory','512m','--cpus','1',
'--pids-limit','64','--log-driver','none','--restart','no',
'--tmpfs','/tmp:rw,nosuid,nodev,noexec,size=16777216,uid=1000,gid=1000,mode=0700']
def mounts(session,readonly=False):
args=[]
for name,spec in session['volumes'].items():
args+=['--mount','type=volume,source='+name+',target='+spec['target']+(',readonly' if readonly and spec['target']!='/var/lib/keycape-policy' else '')]
return args
PROVISION_PHASES=(
'volume_creation_and_verification','worker_launch','worker_containment',
'private_import','issuer_launch','issuer_containment','tls_transport_launch','complete')
def provision(session,payload,tls,issuer,phase=None):
"""Inspect containment and driver options before the first private import."""
from urllib.parse import urlsplit
address=urlsplit(issuer)
if address.scheme!='https' or not address.hostname or address.port not in (None,443) or address.path not in ('','/'):
raise adapter.Refused()
if phase is None:phase=lambda value:None
phase('volume_creation_and_verification')
for name,spec in session['volumes'].items():
run(['docker','volume','create','--driver','local','--label',adapter.LABEL+'='+session['run_id'],
'--opt','type=tmpfs','--opt','device=tmpfs','--opt','o='+VOLUME_OPTIONS['o'],name])
verify_volume(name,spec,session['run_id'])
phase('worker_launch')
args=base_args(session,session['worker'],'none')+mounts(session)
# This pinned helper declares an image VOLUME here. Mask it explicitly so
# Docker cannot allocate an unregistered disk-backed anonymous volume.
args+=['--tmpfs','/privacyidea/etc/persistent:rw,nosuid,nodev,noexec,size=16777216,uid=1000,gid=1000,mode=0700']
args+=['--sysctl','net.ipv4.ip_unprivileged_port_start=0','--add-host',address.hostname+':127.0.0.1',
'--entrypoint','python3',WORKER,'-c','import time;time.sleep(1800)']
run(args)
# Check helper and every private filesystem before touching the ciphertext's
# private contents. Issuer is created only after originals are present.
phase('worker_containment')
adapter.inspect_session({**session,'containers':[session['worker']],'issuer':session['worker']},worker_only=True)
phase('private_import')
files={**payload['files'],**{k:base64.b64encode(v).decode() for k,v in tls.items()}}
run(['docker','exec','-i',session['worker'],'python3','-c',IMPORT],input=json.dumps(files).encode())
anchor=run(['docker','container','inspect',session['worker'],'--format','{{.Id}}']).decode().strip()
phase('issuer_launch')
args=base_args(session,session['issuer'],'container:'+anchor)+mounts(session,readonly=True)
# CLI argv carries environment names only. Original credentials stay in the
# owner subprocess environment and Docker's private container configuration.
env={'PATH':os.environ.get('PATH',os.defpath),'DOCKER_HOST':'unix:///var/run/docker.sock',**payload['env'],
'SSL_CERT_FILE':'/recovery-tls/trust.pem'}
for name in (*payload['env'],'SSL_CERT_FILE'):args+=['--env',name]
args+=[adapter.IMAGE]
run(args,env=env)
phase('issuer_containment')
adapter.inspect_session(session)
phase('tls_transport_launch')
run(['docker','exec','--detach',session['worker'],'python3','-c',TLS])
phase('complete')
def cleanup_all(session,receipt):
failed=False
try:adapter.cleanup({**session,'containers':session['containers']+[session['public_image_probe']]},receipt)
except Exception:failed=True
rows=[]
for name,spec in session['volumes'].items():
row={'volume':name,'status':'unverified'};rows.append(row)
try:
exists=run(['docker','volume','ls','--filter','name=^'+name+'$','--format','{{.Name}}']).decode().splitlines()
if not exists:
row['status']='absence_verified'
continue
verify_volume(name,spec,session['run_id'])
run(['docker','volume','rm',name])
remaining=run(['docker','volume','ls','--filter','name=^'+name+'$','--format','{{.Name}}']).decode().splitlines()
if remaining:row['status']='still_present';failed=True
else:row['status']='absence_verified'
except Exception:failed=True
receipt['tmpfs_volume_cleanup_obligations']=rows
receipt['container_cleanup_verified']=not failed
if failed:raise adapter.Refused()
def register_workdir(work, run_id):
if work.parent!=STAGING_ROOT or not re.fullmatch('nk49-keycape-'+run_id+'-[a-z0-9_]{8}',work.name):raise adapter.Refused()
if any(p.is_symlink() for p in (work,*work.parents)):raise adapter.Refused()
info=work.lstat()
if not stat.S_ISDIR(info.st_mode) or info.st_uid!=os.getuid() or stat.S_IMODE(info.st_mode)!=0o700:raise adapter.Refused()
return {'path':str(work),'run_id':run_id,'device':info.st_dev,'inode':info.st_ino,'owner_uid':info.st_uid}
def cleanup_workdir(staging, receipt):
"""Delete only the registered inode below actual /dev/shm, never a replacement."""
row={'path':staging['path'],'device':staging['device'],'inode':staging['inode'],'status':'unverified'}
receipt['plaintext_cleanup_obligation']=row
work=Path(staging['path'])
if work.parent!=STAGING_ROOT or not re.fullmatch('nk49-keycape-'+staging['run_id']+'-[a-z0-9_]{8}',work.name):raise adapter.Refused()
if not any(line.split()[1:3]==['/dev/shm','tmpfs'] for line in Path('/proc/mounts').read_text().splitlines()):raise adapter.Refused()
if any(p.is_symlink() for p in work.parents):raise adapter.Refused()
parentfd=os.open(work.parent,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW)
try:
try:info=os.stat(work.name,dir_fd=parentfd,follow_symlinks=False)
except FileNotFoundError:
row['status']='absence_verified';receipt['temporary_plaintext_cleanup_verified']=True;return
if (not stat.S_ISDIR(info.st_mode) or info.st_uid!=staging['owner_uid'] or info.st_uid!=os.getuid()
or info.st_dev!=staging['device'] or info.st_ino!=staging['inode'] or stat.S_IMODE(info.st_mode)!=0o700):
row['status']='identity_mismatch_not_removed';raise adapter.Refused()
def remove(parent,name,expected):
fd=os.open(name,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=parent)
try:
current=os.fstat(fd)
if (current.st_dev,current.st_ino)!=(expected.st_dev,expected.st_ino):raise adapter.Refused()
for child in os.listdir(fd):
before=os.stat(child,dir_fd=fd,follow_symlinks=False)
if before.st_uid!=staging['owner_uid'] or before.st_dev!=staging['device']:raise adapter.Refused()
if stat.S_ISDIR(before.st_mode):remove(fd,child,before)
elif stat.S_ISREG(before.st_mode) and before.st_nlink==1:os.unlink(child,dir_fd=fd)
else:raise adapter.Refused() # symlinks/special files/hardlinks are uncertain, never followed
finally:os.close(fd)
after=os.stat(name,dir_fd=parent,follow_symlinks=False)
if (after.st_dev,after.st_ino)!=(expected.st_dev,expected.st_ino):raise adapter.Refused()
os.rmdir(name,dir_fd=parent)
remove(parentfd,work.name,info)
try:os.stat(work.name,dir_fd=parentfd,follow_symlinks=False)
except FileNotFoundError:
row['status']='absence_verified';receipt['temporary_plaintext_cleanup_verified']=True;return
raise adapter.Refused()
finally:os.close(parentfd)
def recover_obligations(session, staging, receipt):
failed=False
try:cleanup_all(session,receipt)
except Exception:failed=True
# The plaintext obligation is attempted even when Docker is unreachable.
try:cleanup_workdir(staging,receipt)
except Exception:failed=True
if failed:raise adapter.Refused()
def process_facts(pid):
raw=Path('/proc')/str(pid)/'stat'
try:
fields=raw.read_text().rsplit(')',1)[1].split()
return {'pid':pid,'state':fields[0],'pgid':int(fields[2]),'sid':int(fields[3]),
'start_ticks':int(fields[19]),'uid':raw.parent.stat().st_uid}
except FileNotFoundError:return None
def isolate_execution():
# Called by main before reading TLS/key inputs. Never reuse an inherited
# shell, custody-helper, user or unrelated process group.
os.setsid()
pid=os.getpid();facts=process_facts(pid)
if os.getsid(0)!=pid or os.getpgrp()!=pid or facts['sid']!=pid or facts['pgid']!=pid:raise adapter.Refused()
return {k:facts[k] for k in ('pid','pgid','sid','start_ticks','uid')}|{'deadline':time.monotonic()+1800}
def verify_execution(execution):
if execution['pid']!=execution['sid'] or execution['pid']!=execution['pgid'] or execution['uid']!=os.getuid():raise adapter.Refused()
current=process_facts(execution['pid'])
if current is not None and any(current[k]!=execution[k] for k in ('pid','pgid','sid','start_ticks','uid')):
# Numeric leader PID reuse is not authority to signal a new group.
raise adapter.Refused()
def group_members(execution):
verify_execution(execution)
members=[]
for path in Path('/proc').iterdir():
if not path.name.isdecimal():continue
facts=process_facts(int(path.name))
if facts is None or facts['pgid']!=execution['pgid']:continue
if facts['sid']!=execution['sid'] or facts['uid']!=execution['uid'] or facts['start_ticks']<execution['start_ticks']:raise adapter.Refused()
if facts['state'] not in ('Z','X'):members.append(facts)
return members
def pidfd_dead(pidfd):
poller=select.poll();poller.register(pidfd,select.POLLIN)
return bool(poller.poll(0))
def quiesce_execution(execution,parent_pidfd,receipt):
"""Stop only this fresh execution session; cleanup needs no live creator.
Exact parent pidfd prevents leader reuse. Each remaining session member gets
its own pidfd after metadata comparison; no inherited/user-group kill occurs.
Unknown membership or uninterruptible survivors refuse cleanup altogether.
"""
verify_execution(execution)
if not pidfd_dead(parent_pidfd):
signal.pidfd_send_signal(parent_pidfd,signal.SIGTERM)
poller=select.poll();poller.register(parent_pidfd,select.POLLIN);poller.poll(2000)
if not pidfd_dead(parent_pidfd):signal.pidfd_send_signal(parent_pidfd,signal.SIGKILL)
for attempt in range(50):
members=group_members(execution)
if pidfd_dead(parent_pidfd) and not members:
receipt['execution_quiescence_verified']=True;return
for original in members:
try:fd=os.pidfd_open(original['pid'])
except ProcessLookupError:continue
try:
current=process_facts(original['pid'])
if current is None or pidfd_dead(fd):continue
if current!=original: # state may naturally change; bind stable identity below
if any(current[k]!=original[k] for k in ('pid','pgid','sid','start_ticks','uid')):raise adapter.Refused()
verify_execution(execution)
signal.pidfd_send_signal(fd,signal.SIGKILL)
except ProcessLookupError:pass
finally:os.close(fd)
time.sleep(.05)
receipt['execution_quiescence_verified']=False
raise adapter.Refused()
def daemon_uncertain(staging):
return any(Path(staging['path']+suffix).exists() or Path(staging['path']+suffix).is_symlink() for suffix in ('-docker-inflight','-docker-uncertain'))
def reaper_main(session,staging,execution,parent_pidfd):
receipt={'status':'failed','execution_quiescence_verified':False}
verify_execution(execution)
if pidfd_dead(parent_pidfd):raise adapter.Refused()
print('registered',flush=True)
poller=select.poll();poller.register(parent_pidfd,select.POLLIN)
remaining=max(0,int((execution['deadline']-time.monotonic())*1000))
parent_lost=bool(poller.poll(remaining))
receipt['cleanup_trigger']='parent_exit' if parent_lost else 'execution_deadline'
try:
# No deletion races a live creator: first stop/wait this exact owner
# session and its children. A daemon request already accepted can still
# complete after CLI loss; retain that uncertainty even after readback.
quiesce_execution(execution,parent_pidfd,receipt)
uncertain=daemon_uncertain(staging)
receipt['in_flight_daemon_request_uncertain']=uncertain
recover_obligations(session,staging,receipt)
receipt['status']='bounded_cleanup_uncertain' if uncertain else 'bounded_cleanup_verified'
except Exception:receipt['status']='bounded_cleanup_uncertain'
finally:
os.close(parent_pidfd)
try:
fd=os.open(staging['path']+'-reaper.json',os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
with os.fdopen(fd,'w') as out:json.dump(receipt,out)
except Exception:pass
def start_reaper(session, staging, execution):
adapter.docker_environment()
verify_execution(execution)
if session.get('docker_endpoint')!=adapter.ENDPOINT or staging.get('run_id')!=session['run_id']:raise adapter.Refused()
if register_workdir(Path(staging['path']),session['run_id'])!=staging:raise adapter.Refused()
parent_pidfd=os.pidfd_open(execution['pid'])
code="""import importlib.util,json,sys
spec=importlib.util.spec_from_file_location('session',sys.argv[1]);m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
m.reaper_main(json.loads(sys.argv[2]),json.loads(sys.argv[3]),json.loads(sys.argv[4]),int(sys.argv[5]))
"""
try:
child=subprocess.Popen([sys.executable,'-c',code,str(Path(__file__).resolve()),json.dumps(session),json.dumps(staging),json.dumps(execution),str(parent_pidfd)],
stdin=subprocess.DEVNULL,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,
env={'PATH':os.environ.get('PATH',os.defpath),'DOCKER_HOST':adapter.ENDPOINT},
close_fds=True,pass_fds=(parent_pidfd,),start_new_session=True)
finally:os.close(parent_pidfd)
ready,_,_=select.select([child.stdout],[],[],10)
if not ready or child.stdout.readline()!=b'registered\n':
child.terminate();child.wait(timeout=5);raise adapter.Refused()
child.stdout.close()
return child
def execute(transfer,admission,tls,receipt,execution,checkpoint=None):
"""Existing private custody/transfer/archive APIs; no second archive format."""
verify_execution(execution)
if execution['pid']!=os.getpid():raise adapter.Refused()
target=local.preflight()
for image in (adapter.IMAGE,):run(['docker','image','inspect',image])
local.offline.validate(transfer)
if transfer.get('independent_owner_retrieval_verified') is not True:raise adapter.Refused()
path=Path(transfer['fetched_path']);bundle.private(path,max_bytes=bundle.LIMIT)
if (path.stat().st_size!=transfer['ciphertext_bytes'] or local.offline.offsite.digest(path)!=transfer['ciphertext_sha256']
or admission['ciphertext_sha256']!=transfer['ciphertext_sha256']):raise adapter.Refused()
if hashlib.sha256(tls['/recovery-tls/trust.pem']).hexdigest()!=admission['trust_sha256']:raise adapter.Refused()
if hashlib.sha256(tls['/recovery-tls/server.pem']).hexdigest()!=admission['tls_cert_sha256']:raise adapter.Refused()
# Validate exact private certificate/key pair and original issuer name before
# opening service containers; do not generate replacement trust or keys.
from cryptography import x509
from cryptography.hazmat.primitives import serialization
cert=x509.load_pem_x509_certificate(tls['/recovery-tls/server.pem'])
key=serialization.load_pem_private_key(tls['/recovery-tls/server-key.pem'],password=None)
if cert.public_key().public_numbers()!=key.public_key().public_numbers():raise adapter.Refused()
keyfd=local.verify.custody.private_key_fd();work=None;session=None;reaper=None;staging=None
receipt.update(status='failed',target=target,session_source_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
source_files_sha256={name:hashlib.sha256((HERE/name).read_bytes()).hexdigest() for name in ('isolated-keycape.py','isolated-keycape-session.py','run-isolated-keycape.sh','isolated-lldap.py','identity-bundle.py','verify-identity.py','retrieve-identity-offline.py','identity-offsite.py')},
temporary_plaintext_cleanup_verified=False,container_cleanup_verified=False,
downstream_consumer_reconnect_proven=False,human_mfa_proven=False)
try:
if run(['age-keygen','-y',f'/proc/self/fd/{keyfd}'],pass_fds=(keyfd,)).strip()!=bundle.profile()['public_recipient'].encode():raise adapter.Refused()
session=plan_session()
work=Path(tempfile.mkdtemp(prefix='nk49-keycape-'+session['run_id']+'-',dir='/dev/shm'))
staging=register_workdir(work,session['run_id'])
reaper=start_reaper(session,staging,execution)
adapter.EXECUTION={**execution,'staging':str(work)}
receipt.update(execution_binding=execution,bounded_cleanup_reaper_started=True,plaintext_cleanup_obligation={**staging,'status':'registered_before_plaintext'},docker_endpoint=adapter.ENDPOINT,
reaper_outcome_path=staging['path']+'-reaper.json',
registered_cleanup={'containers':session['containers']+[session['public_image_probe']],'tmpfs_volumes':list(session['volumes'])})
if checkpoint is not None:checkpoint(receipt)
archive=work/'archive.tar'
with archive.open('xb') as output:
p=adapter.process(['age','-d','-i',f'/proc/self/fd/{keyfd}',str(path)],stdout=output,stderr=subprocess.DEVNULL,pass_fds=(keyfd,),timeout=180)
if p.returncode or archive.stat().st_size>bundle.LIMIT+4*1024*1024:raise adapter.Refused()
root=work/'restored';root.mkdir(mode=0o700)
manifest,_=bundle.restore(archive,root)
payload=adapter.private_payload(root,manifest,admission['manifest_sha256'])
import yaml
cfg=yaml.safe_load(base64.b64decode(payload['files']['/etc/keycape/config.yaml']))
if cfg.get('port')!=8080 or cfg.get('issuer')!=admission['issuer']:raise adapter.Refused()
# Public executable readback uses the daemon archive API only on the
# original image rootfs, never a tmpfs or private user-created mount.
probe=session['public_image_probe']
try:
run(['docker','create','--pull=never','--name',probe,'--network','none','--read-only','--label',adapter.LABEL+'='+session['run_id'],adapter.IMAGE])
binary=adapter.container_file(probe,'/keycape')
if hashlib.sha256(binary).hexdigest()!=admission['verifier_sha256']:raise adapter.Refused()
finally:
receipt['public_image_probe_cleanup_obligations']=[{'container':probe,'status':'unverified'}]
run(['docker','rm',probe])
if run(['docker','container','ls','--all','--filter','name=^/'+probe+'$','--format','{{.Names}}']).strip():raise adapter.Refused()
receipt['public_image_probe_cleanup_obligations'][0]['status']='absence_verified'
def phase(value):
# Public enum only: no private payload, exception text or daemon logs.
if value not in PROVISION_PHASES:raise adapter.Refused()
receipt['provision_phase']=value
if checkpoint is not None:checkpoint(receipt)
provision(session,payload,tls,admission['issuer'],phase)
receipt.update(ciphertext_sha256=transfer['ciphertext_sha256'],all_manifest_hashes_verified=True,manifest_file_count=len(manifest['files']))
checked_admission={**admission,'session':session,'trust_path':'/recovery-tls/trust.pem'}
# Native checker consumes no service secret argument or new grant.
# Retry only bounded public readiness; no account/factor attempt.
for attempt in range(40):
try:
numbers=serialization.load_pem_private_key(base64.b64decode(payload['files']['/etc/keycape/key.pem']),password=None).public_key().public_numbers()
adapter.check_jwks(session,checked_admission,numbers);break
except Exception:
if attempt==39:raise
time.sleep(.25)
adapter.native_checks(root,manifest,checked_admission,receipt,cleanup_session=False)
finally:
os.close(keyfd)
try:
if session is not None and staging is not None:
uncertain=daemon_uncertain(staging)
receipt['in_flight_daemon_request_uncertain']=uncertain
recover_obligations(session,staging,receipt)
if uncertain:raise adapter.Refused()
if reaper is not None:
reaper.terminate();reaper.wait(timeout=10)
except Exception:
receipt['status']='failed'
raise
finally:
adapter.EXECUTION=None
if work is not None and not work.exists():receipt['temporary_plaintext_cleanup_verified']=True
def persist_receipt(fd,receipt):
raw=(json.dumps(receipt,indent=2)+'\n').encode()
os.lseek(fd,0,os.SEEK_SET);os.ftruncate(fd,0)
with os.fdopen(os.dup(fd),'wb') as output:
output.write(raw);output.flush();os.fsync(output.fileno())
def main():
p=argparse.ArgumentParser(description=__doc__)
for name in ('transfer-receipt','admission','admission-sha256','tls-cert','tls-key','trust','receipt'):p.add_argument('--'+name,required=True)
args=p.parse_args();os.umask(0o077)
fd=os.open(args.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
receipt={'schema':'netkingdom.isolated-keycape-recovery.v1','status':'failed','production_contacted':False,'secret_values_exported':False}
try:
execution=isolate_execution()
raw=Path(args.admission).read_bytes()
if hashlib.sha256(raw).hexdigest()!=args.admission_sha256:raise adapter.Refused()
admission=json.loads(raw)
tls={}
for name,path in (('server.pem',args.tls_cert),('server-key.pem',args.tls_key),('trust.pem',args.trust)):
source=Path(path);bundle.private(source,max_bytes=1024*1024);tls['/recovery-tls/'+name]=source.read_bytes()
execute(json.loads(Path(args.transfer_receipt).read_text()),admission,tls,receipt,execution,checkpoint=lambda r:persist_receipt(fd,r))
except (Exception,KeyboardInterrupt):receipt.update(status='failed',error='isolated_keycape_failed_private_diagnostics_suppressed')
finally:
receipt['finished_at']=datetime.now(timezone.utc).isoformat()
try:persist_receipt(fd,receipt)
finally:os.close(fd)
print('OK: isolated KeyCape native checks and cleanup passed.' if receipt['status']=='isolated_keycape_native_registration_checked' else 'ERROR: isolated KeyCape stopped; review safe receipt.')
return int(receipt['status']!='isolated_keycape_native_registration_checked')
if __name__=='__main__':
for sig in (signal.SIGHUP,signal.SIGTERM,signal.SIGINT):signal.signal(sig,local.offline.interrupted)
raise SystemExit(main())

452
tools/bootstrap-recovery/isolated-keycape.py Normal file
View file

@ -0,0 +1,452 @@
#!/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 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

40
tools/bootstrap-recovery/run-isolated-keycape.sh Normal file
View file

@ -0,0 +1,40 @@
#!/usr/bin/env bash
# Candidate owner launcher. Review and native fixture proof required before use.
set +x
set -euo pipefail
ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/../.." && pwd)"
cd -- "$ROOT"
umask 077
[[ $# == 6 ]] || { printf '%s\n' 'Usage: run-isolated-keycape.sh VERIFIED_TRANSFER_RECEIPT ADMISSION ADMISSION_SHA256 TLS_CERT TLS_KEY TRUST'; exit 1; }
[[ "$3" =~ ^[a-f0-9]{64}$ ]] || { printf '%s\n' 'ERROR: exact reviewed admission digest required.'; exit 1; }
for index in 1 2 4 5 6; do
[[ -f "${!index}" && ! -L "${!index}" ]] || { printf '%s\n' 'ERROR: provide regular owner-controlled input files.'; exit 1; }
done
[[ -t 0 && -t 1 && -t 2 ]] || { printf '%s\n' 'ERROR: run in your own interactive terminal.'; exit 1; }
for tool in python3 age age-keygen docker; do
command -v "$tool" >/dev/null || { printf 'ERROR: required tool unavailable: %s\n' "$tool"; exit 1; }
done
python3 - <<'PY'
import importlib.util
from pathlib import Path
p=Path('tools/bootstrap-recovery/isolated-keycape-session.py')
spec=importlib.util.spec_from_file_location('session',p)
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
try:
m.local.preflight()
m.run(['docker','image','inspect',m.adapter.IMAGE])
except Exception:
raise SystemExit('ERROR: exact local Docker/images/tmpfs preflight failed; no key requested.')
print('OK: fixed local target and cached pinned images available.')
PY
NK49_KC_RUN="$(mktemp -d /tmp/nk49-isolated-keycape-XXXXXXXX)"
printf 'Safe evidence directory: %s\n' "$NK49_KC_RUN"
printf '%s\n' 'Owner-reviewed candidate: isolated original issuer/trust and current registered service-client checks only.'
printf '%s\n' 'No browser, downstream consumer or retained-factor MFA acceptance is claimed.'
printf '%s\n' 'Enter only the independently held backup age key at the hidden prompt.'
bash sso-mfa/bootstrap/sops-custody-unlock.sh \
--expected-recipient age1zvryunvjhvpkmasskauga2heeg0ztnte9ymgppvjge36ekumk50syr3tsz \
--prompt -- python3 tools/bootstrap-recovery/isolated-keycape-session.py \
--transfer-receipt "$1" --admission "$2" --admission-sha256 "$3" \
--tls-cert "$4" --tls-key "$5" --trust "$6" --receipt "$NK49_KC_RUN/recovery.json"
printf 'Safe receipt: %s/recovery.json\n' "$NK49_KC_RUN"

72
workplans/NK-WP-0049-bootstrap-secret-free-recovery-handoff.md
View file

@ -559,8 +559,8 @@ id: NK-WP-0049-T18
status: wait
priority: high
assignee: codex
depends_on: [NK-WP-0049-T15, NK-WP-0049-T16, NK-WP-0049-T19]
blocking_reason: "LLDAP recovery T16 passed; await the exact isolated KeyCape adapter/procedure and retained-token MFA proof for the selected flow."
depends_on: [NK-WP-0049-T15, NK-WP-0049-T16, NK-WP-0049-T19, NK-WP-0049-T32, NK-WP-0049-T33]
blocking_reason: "Source and synthetic KeyCape lifecycle proof are delivered in T32; await exact owner-held recovery attempt preparation T33, selected consumer acceptance and retained-token MFA for the selected flow."
state_hub_task_id: "141dd773-c1fa-56c1-a852-bbe2372e5125"
```
@ -910,6 +910,74 @@ not prove the recovered secret authenticates a retained factor; actual positive
MFA and wrong-key/wrong-code authentication refusal remain T19. No token seed,
plaintext, serial or user record belongs in the receipt, Git, State Hub or chat.
## Deliver isolated KeyCape restore source and synthetic lifecycle proof
```task
id: NK-WP-0049-T32
status: done
priority: high
assignee: codex
```
Completed 2026-10-16. This extracts bounded source/fixture delivery from T18;
T18 retains actual original archive, issuer and selected consumer acceptance.
The adapter reuses the existing transfer/archive/custody contracts, binds the
fixed local daemon and cached images, restores private state only to temporary
memory, verifies original issuer/TLS/signing/configuration/policy bytes and runs
the original native service-client checker with wrong-secret/excess-scope denial.
Durable pre-import obligations and exact parent pidfd/session quiescence precede
loss/deadline cleanup. Daemon response uncertainty remains uncertainty after
resource absence; it never authorizes recovery success or replay.
Frozen R6 fixes a demonstrated image-declared anonymous-volume incompatibility
using bounded helper-only tmpfs, preserving anonymous disk-volume refusal. Safe
phase markers carry no private diagnostics. Independently passed 72 focused
checks, then actual pinned-image synthetic positive, hard-parent-loss, deadline
and nonzero-after-accept cases. Independent fixed-daemon inventories and host
mount-namespace checks prove all exact resources and plaintext absent. Reaper
outcomes are retained; initial sandbox receipt lookup failure is preserved and
superseded by device-bound host verification. R1–R5 failures remain retained.
Evidence: Custodian docs/changes/netkingdom-target-2026-10-16/
identity-restore-adapter-next/revisions/r6/, actual-fixture-r6-all.json and
coordinator-r6-all-review.json. Exact source/dependency hashes and candidate
manifest bind the proof. This uses generated test keys only; it proves neither
actual owner-held recovery, human MFA, downstream reconnect nor native authority.
Publication preparation R7 preserves every R6 recovery adapter byte and digest.
The normal scanner flagged eight public source-file SHA-256 values in the R6
manifest. Explicit sha256-qualified checksums resolve those false positives;
the fixture reader strictly validates their type/shape before checking the same
bytes. R7 passes 73 focused checks including malformed/untyped digest refusal.
The frozen R6 actual four-case proof remains the runtime evidence; no repeated
native outcome is claimed for publication encoding. Original scanner refusal is
retained, and scanner policy/exemptions are unchanged.
## Prepare the exact owner-held KeyCape recovery attempt
```task
id: NK-WP-0049-T33
status: todo
priority: high
assignee: codex
depends_on: [NK-WP-0049-T32]
```
**Accountable worker:** NetKingdom codex with KeyCape and selected consumer owners.
Prepare the exact archive/manifest, original issuer/TLS/trust and executable
bindings, selected current registration and dependency graph using public metadata.
Use the existing independently held archive-key and owner-controlled TLS custody
paths; never put secret values in the packet. Validate the attended launcher and
its full dependency closure, execution deadline, recovery obligations and safe
receipt path before asking for any owner-held input. Bind approved source revision
and rollback/abort limits through the existing owner process.
**Done when:** reviewed executable command and metadata-only admission packet
are ready for the existing attended custody path; prerequisites and independent
receipt verification are explicit. T18 still owns actual execution, original
consumer continuity and acceptance. T19 retains retained-factor MFA. This task
neither authorizes a production consumer redirect nor creates a new client grant.
## Accept remaining recovery and caller evidence and close the G08 handoff
```task