Add attended signer adapter with phase-aware read-only recovery

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786

scripts/flex_auth_signer_custody.py

@@ -58,6 +58,10 @@ class SignerWriter:
                 or '-dev' not in process.args or '-dev-no-store-token' not in process.args
                 or '-dev-listen-address=127.0.0.1:'+str(admission.port) not in process.args):
             raise Refused('live owner-spawned disposable dev process required')
+        self.admission=admission
+        self._initialize_receipt(directory,operation_id,kid=kid,expected_version=expected_version,bindings=bindings)
+
+    def _initialize_receipt(self,directory,operation_id,*,kid,expected_version,bindings):
         if (type(operation_id) is not str or not re.fullmatch(r'[a-zA-Z0-9_-]{1,80}',operation_id)
                 or type(kid) is not str or not re.fullmatch(r'[a-zA-Z0-9_-]{1,80}',kid)
                 or type(expected_version) is not int or not 0<=expected_version<2**31
@@ -66,7 +70,7 @@ class SignerWriter:
             raise Refused('exact signer operation contract required')
         if directory.absolute()!=directory.resolve() or directory.is_symlink():raise Refused('indirect receipt custody refused')
         Journal._private(directory.stat(),directory=True)
-        self.admission=admission;self.directory=directory;self.path=directory/(operation_id+'.json')
+        self.directory=directory;self.path=directory/(operation_id+'.json')
         self.expected_version=expected_version;self.kid=kid;self.attempted=False
         self.receipt={'schema':'railiance-platform.signer-cas-receipt.v1','operation_id':operation_id,
             'bindings':dict(bindings),'writer_sha256':'sha256:'+hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),

scripts/flex_auth_signer_native.py

@@ -0,0 +1,417 @@
+#!/usr/bin/env python3
+"""Silent attended retained-lifecycle signer custody; never a credential front door.
+
+Invoke only through the reviewed warden platform-admin envelope. The externally
+reviewed admission SHA is an exact-command input, not approval by JSON shape.
+No private value crosses stdout/stderr, argv, receipt or persistent local storage.
+"""
+from __future__ import annotations
+import argparse
+import base64
+from datetime import datetime,timezone
+import fcntl
+import hashlib
+import hmac
+import http.client
+import importlib.util
+import json
+import os
+from pathlib import Path
+import re
+import selectors
+import shlex
+import subprocess
+import sys
+import time
+
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from flex_auth_signer_custody import SignerWriter,PATH,FIELDS,BINDINGS,encoded,b64
+from task_execution_journal import Journal,Refused
+from task_execution_native_fixture import _absolute_budget
+from task_execution_recovery import _json_unique,_invalid_constant,_policy_absent,_same_policy
+
+ROOT=Path(__file__).resolve().parents[1]
+META='platform/metadata/workloads/flex-auth/envelope-signing'
+CCR='credential-change-requests/CCR-2026-0050-flex-auth-envelope-signing.yaml'
+PLAN='docs/credential-lane-designs/flex-auth-envelope-signing-native-plan.json'
+CLUSTER='a553c742-0115-43d4-99a4-a5ca56fe0786'
+BAO_CLUSTER='fd28df5d-98ec-57dd-42ec-9b3e4f4e53bf'
+NAME='flex-auth-secrets-engine'
+SECRET='flex-auth-secrets-engine-envelope-signing'
+ROLE='external-secrets-flex-auth-envelope-signing'
+LIFECYCLE='sha256:bd11c5fe77ce6439c65fea225ad6b71d2110efc5e7b5bc9b499c59cd0a53b8b4'
+REVIEWS={'founder','railiance-platform-owner','access-engine-owner','secrets-engine-owner','cluster-gitops-owner'}
+SOURCE_FILES={'scripts/flex_auth_signer_native.py','scripts/flex_auth_signer_custody.py','scripts/task_execution_native_fixture.py','scripts/task_execution_journal.py','scripts/task_execution_recovery.py','scripts/credential-change.py'}
+
+
+def require(ok,reason):
+    if not ok:raise Refused(reason)
+
+
+def digest(raw):return 'sha256:'+hashlib.sha256(raw).hexdigest()
+def is_digest(value):return type(value) is str and re.fullmatch(r'sha256:[0-9a-f]{64}',value) is not None
+
+def private_read(path,limit=65536):
+    path=Path(path)
+    require(path.is_absolute() and path.absolute()==path.resolve() and not path.is_symlink(),'indirect private input')
+    Journal._private(path.parent.stat(),directory=True)
+    fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK)
+    try:
+        before=os.fstat(fd);Journal._private(before)
+        raw=os.read(fd,limit+1);after=os.fstat(fd)
+        require((before.st_dev,before.st_ino,before.st_size,before.st_mtime_ns,before.st_ctime_ns)==
+                (after.st_dev,after.st_ino,after.st_size,after.st_mtime_ns,after.st_ctime_ns)
+                and len(raw)==before.st_size and len(raw)<=limit,'private input changed or oversized')
+        return raw
+    finally:os.close(fd)
+
+
+def parsed(raw):
+    try:return json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant)
+    except Exception:raise Refused('invalid closed JSON input') from None
+
+
+def validate_admission(value):
+    fields={'schema','action','operation_id','kid','expires_at','founder_entity_id','ccr_sha256','source_bundle','bindings',
+            'producer_image','caller_binding','deployment_projection_sha256','authority','routing','owner_reviews',
+            'exclusive_window_sha256','consumer_trust_sha256','original_receipt_sha256','original_execution'}
+    require(type(value) is dict and set(value)==fields,'exact admission schema required')
+    require(value['schema']=='railiance-platform.signer-native-admission.v1' and value['action'] in {'bootstrap','recover'},'native admission action refused')
+    require(type(value['operation_id']) is str and re.fullmatch(r'[A-Za-z0-9_-]{1,80}',value['operation_id']) and
+            type(value['kid']) is str and re.fullmatch(r'[A-Za-z0-9_-]{1,80}',value['kid']),'native operation identity required')
+    require(type(value['founder_entity_id']) is str and re.fullmatch(r'[a-f0-9-]{36}',value['founder_entity_id']),'explicit attended identity required')
+    require(all(is_digest(value[x]) for x in ['ccr_sha256','deployment_projection_sha256','exclusive_window_sha256','consumer_trust_sha256']),'explicit custody/image/trust selections required')
+    require(type(value['producer_image']) is str and re.fullmatch(r'forgejo\.coulomb\.social/coulomb/flex-auth@sha256:[a-f0-9]{64}',value['producer_image']),'immutable producer image required')
+    require(value['caller_binding']=='secrets-engine=system:serviceaccount:secrets-engine:secrets-engine','retained caller binding required')
+    require(value['authority']=='founder-attended-exact-source-scope; no standing grant','independent attended authority required')
+    route=value['routing'];require(type(route) is dict and set(route)=={'lane_id','verdict','founder_act','receipt_sha256'} and
+        route['lane_id']=='openbao-platform-admin-login' and route['verdict']=='founder_required' and route['founder_act']=='oidc_login' and is_digest(route['receipt_sha256']),'routing mismatch refused')
+    reviews=value['owner_reviews'];require(type(reviews) is dict and set(reviews)==REVIEWS,'distinct owner reviews required')
+    for role,review in reviews.items():
+        require(type(review) is dict and set(review)=={'approved_admission_sha256','approval_ref'} and
+            is_digest(review['approved_admission_sha256']) and type(review['approval_ref']) is str and
+            0<len(review['approval_ref'])<=200,'exact external review required')
+    # Dispositions bind a claim-free scope digest, avoiding self-referential JSON.
+    scope={k:v for k,v in value.items() if k!='owner_reviews'};scope_digest=digest(encoded(scope))
+    require(all(x['approved_admission_sha256']==scope_digest for x in reviews.values()),'owner review scope mismatch')
+    require(type(value['source_bundle']) is dict and set(value['source_bundle'])==SOURCE_FILES and
+            all(is_digest(x) for x in value['source_bundle'].values()),'exact execution source bundle required')
+    require(type(value['bindings']) is dict and set(value['bindings'])==BINDINGS and all(is_digest(x) for x in value['bindings'].values()),'exact execution bindings required')
+    require(value['bindings']['policy_sha256']==LIFECYCLE and value['bindings']['image_sha256']=='sha256:'+value['producer_image'].rsplit('sha256:',1)[1]
+            and value['bindings']['trust_sha256']==value['consumer_trust_sha256']
+            and value['bindings']['deployment_sha256']==value['deployment_projection_sha256']
+            and value['bindings']['executed_revision_sha256']==digest(encoded(value['source_bundle'])),'immutable lifecycle/image/trust/source mismatch')
+    require((value['action']=='bootstrap' and value['original_receipt_sha256'] is None) or
+            (value['action']=='recover' and is_digest(value['original_receipt_sha256'])),'separate recovery disposition required')
+    original=value['original_execution']
+    if value['action']=='bootstrap':require(original is None,'bootstrap cannot import execution lineage')
+    else:
+        require(type(original) is dict and set(original)=={'bindings','writer_sha256','source_ccr_sha256'} and
+                is_digest(original['writer_sha256']) and is_digest(original['source_ccr_sha256']) and
+                type(original['bindings']) is dict and set(original['bindings'])==BINDINGS and
+                all(is_digest(x) for x in original['bindings'].values()),'independently reviewed original execution lineage required')
+        require(all(original['bindings'][key]==value['bindings'][key] for key in BINDINGS-{'proposal_sha256','executed_revision_sha256'}),
+                'recovery cannot change original image/deployment/policy/trust authority')
+    try:expires=datetime.fromisoformat(value['expires_at'].replace('Z','+00:00'));remaining=(expires-datetime.now(timezone.utc)).total_seconds()
+    except Exception:raise Refused('invalid authority expiry') from None
+    require(expires.utcoffset().total_seconds()==0 and 5<remaining<=120,'bounded fresh attended window required')
+    return remaining
+
+
+class NativeIO:
+    """All tools capture/discard private bytes; fixed transport and no shell."""
+    def captured(self,argv,payload=None,*,allowed=(0,)):
+        env=os.environ.copy();env.update(BAO_ADDR='http://127.0.0.1:18200',VAULT_ADDR='http://127.0.0.1:18200')
+        for key in ('BAO_LOG_LEVEL','VAULT_LOG_LEVEL'):
+            require(env.get(key,'').lower() not in {'debug','trace'},'verbose custody environment refused')
+        # A hostile inherited CLI config is not an alternate identity/endpoint.
+        require(not env.get('BAO_NAMESPACE') and not env.get('VAULT_NAMESPACE') and not env.get('VAULT_SKIP_VERIFY'),'ambient backend override refused')
+        proc=subprocess.Popen(argv,stdin=subprocess.PIPE,stdout=subprocess.PIPE,stderr=subprocess.PIPE,env=env)
+        try:
+            if payload is not None:proc.stdin.write(encoded(payload));proc.stdin.flush()
+            proc.stdin.close()
+            sel=selectors.DefaultSelector();data={proc.stdout:bytearray(),proc.stderr:bytearray()}
+            try:
+                for stream in data:sel.register(stream,selectors.EVENT_READ)
+                deadline=time.monotonic()+2
+                while sel.get_map():
+                    remaining=deadline-time.monotonic();require(remaining>0,'absolute custody command timeout')
+                    for item,_ in sel.select(remaining):
+                        chunk=os.read(item.fileobj.fileno(),65536)
+                        if not chunk:sel.unregister(item.fileobj)
+                        else:
+                            data[item.fileobj].extend(chunk)
+                            require(sum(map(len,data.values()))<=1048576,'custody command output bound')
+                code=proc.wait(timeout=max(.001,deadline-time.monotonic()))
+                require(code in allowed,'custody command unconfirmed')
+                return code,bytes(data[proc.stdout])
+            finally:sel.close()
+        except BaseException:
+            proc.kill();proc.wait(timeout=2);raise Refused('custody command unconfirmed') from None
+        finally:
+            proc.stdout.close();proc.stderr.close()
+    def bao(self,args,payload=None,allowed=(0,)):
+        with _absolute_budget(2):
+            code,raw=self.captured(['bao',*args],payload,allowed=allowed)
+            return code,parsed(raw)
+    def kube(self,resource,name=None):
+        args=['kubectl','--request-timeout=2s']
+        if resource!='namespace':args+=['-n','flex-auth']
+        args+=['get',resource]
+        if name:args.append(name)
+        args+=['-o','json']
+        with _absolute_budget(2):
+            _,raw=self.captured(['ssh','-o','BatchMode=yes','-o','StrictHostKeyChecking=yes','-o','ConnectTimeout=2','railiance01',shlex.join(args)])
+            return parsed(raw)
+    def http(self,method,path,body=None):
+        require(method in {'GET','POST'} and path in {META,PATH,PATH+'?version=1'},'fixed custody path required')
+        require(type(getattr(self,'_token',None)) is str and self._token,'contained attended session required')
+        with _absolute_budget(2):
+            conn=http.client.HTTPConnection('127.0.0.1',18200,timeout=2)
+            try:
+                conn.request(method,'/v1/'+path,body=None if body is None else encoded(body),
+                             headers={'X-Vault-Token':self._token,'Content-Type':'application/json'})
+                response=conn.getresponse();raw=response.read(1048577)
+                require(len(raw)<=1048576,'custody response bound')
+                return response.status,parsed(raw) if raw else {}
+            finally:conn.close()
+
+
+def deployment_projection(deployment):
+    spec=deployment['spec']['template']['spec'];containers=spec.get('containers',[])
+    require(type(containers) is list and len(containers)==1 and not spec.get('initContainers'),'single fixed signer container required')
+    item=containers[0]
+    # Never hash arbitrary env values or an entire Secret/spec response.
+    return {'service_account':spec.get('serviceAccountName'),'image':item.get('image'),
+            'args':item.get('args'),'volume_mounts':item.get('volumeMounts',[]),
+            'volumes':spec.get('volumes',[]),'security_context':spec.get('securityContext')}
+
+
+def fence(io,contract):
+    namespace=io.kube('namespace','kube-system')
+    require(namespace.get('metadata',{}).get('uid')==CLUSTER,'wrong target cluster')
+    deployment=io.kube('deployment',NAME);spec=deployment.get('spec',{});meta=deployment.get('metadata',{})
+    require(type(spec.get('replicas')) is int and spec['replicas']==0 and
+            deployment.get('status',{}).get('observedGeneration',-1)==meta.get('generation') and
+            not deployment.get('status',{}).get('replicas',0),'all signer replicas must be stopped')
+    projection=deployment_projection(deployment)
+    require(digest(encoded(projection))==contract['deployment_projection_sha256'] and
+            projection['service_account']==NAME and projection['image']==contract['producer_image'],'fenced deployment source/image mismatch')
+    args=projection['args'];require(type(args) is list and all(type(x) is str for x in args),'exact signer args required')
+    def option(flag):
+        require(args.count(flag)==1 and args.index(flag)+1<len(args),'fixed signer flag required')
+        return args[args.index(flag)+1]
+    require(option('--caller-auth-mode')=='enforce' and option('--caller-audience')=='flex-auth' and
+            option('--caller-binding')==contract['caller_binding'],'retained caller guard required')
+    require(not any(x.startswith(flag+'=') for x in args for flag in ['--caller-auth-mode','--caller-binding','--caller-audience','--envelope-signing-key-file','--envelope-signing-kid']),'ambiguous protected args refused')
+    require(option('--envelope-signing-key-file')=='/var/run/secrets/decision-envelope-signing/seed' and
+            option('--envelope-signing-kid')==contract['kid'],'exact seed mount and startup kid required')
+    volumes=projection['volumes'];mounts=projection['volume_mounts']
+    matching=[x for x in volumes if x.get('name')=='flex-auth-envelope-signing']
+    require(len(matching)==1 and matching[0].get('secret')=={'secretName':SECRET,'defaultMode':288,'items':[{'key':'seed','path':'seed'}]},'exact seed volume required')
+    require(any(x=={'name':'flex-auth-envelope-signing','mountPath':'/var/run/secrets/decision-envelope-signing','readOnly':True} for x in mounts),'readonly seed mount required')
+    uid=meta.get('uid');require(type(uid) is str and uid,'deployment UID required')
+    sets=io.kube('replicasets').get('items');pods=io.kube('pods').get('items');hpas=io.kube('hpa').get('items')
+    require(type(sets) is list and type(pods) is list and type(hpas) is list,'complete fence observations required')
+    rsids=[]
+    for item in sets:
+        if any(x.get('uid')==uid for x in item.get('metadata',{}).get('ownerReferences',[])):
+            require(type(item.get('spec',{}).get('replicas')) is int and item['spec']['replicas']==0 and
+                    type(item.get('status',{}).get('replicas',0)) is int and not item.get('status',{}).get('replicas',0),'owned ReplicaSet still running')
+            rsids.append(item['metadata']['uid'])
+    for item in pods:
+        podspec=item.get('spec',{})
+        uses_secret=any(x.get('secret',{}).get('secretName')==SECRET for x in podspec.get('volumes',[]))
+        owned=any(x.get('uid') in [uid,*rsids] for x in item.get('metadata',{}).get('ownerReferences',[]))
+        require(not owned and podspec.get('serviceAccountName')!=NAME and not uses_secret,'live or terminating signer pod refused')
+    require(not any(x.get('spec',{}).get('scaleTargetRef',{}).get('name')==NAME for x in hpas),'signer autoscaler refused')
+    return True
+
+
+def checked_sources(contract):
+    for name,pin in contract['source_bundle'].items():
+        path=ROOT/name
+        require(path.resolve()==path.absolute() and not path.is_symlink() and path.stat().st_mode&0o022==0
+                and path.is_file() and path.stat().st_size<=1048576 and digest(path.read_bytes())==pin,'exact admitted execution source required')
+    planraw=(ROOT/PLAN).read_bytes();require(digest(planraw)==contract['bindings']['proposal_sha256'],'exact native action proposal required')
+    plan=parsed(planraw);require(plan.get('bootstrap',{}).get('source_primitive_sha256')==contract['source_bundle']['scripts/flex_auth_signer_custody.py'],'proposal executed primitive lineage mismatch')
+    require(plan.get('proposal_id')=='RPF35-PDP-SIGNER-NATIVE-20261016-R1' and plan.get('bootstrap',{}).get('kid')==contract['kid'],'exact proposed bootstrap kid required')
+    ccrraw=(ROOT/CCR).read_bytes();require(digest(ccrraw)==contract['ccr_sha256'],'reviewed read CCR source mismatch')
+    spec=importlib.util.spec_from_file_location('native_signer_ccr',ROOT/'scripts/credential-change.py');module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+    ccr,errors,_=module.validate_ccr(ROOT/CCR)
+    require(not errors and ccr['status'] in {'approved','applied','verified','active'} and
+            ccr['openbao']['auth']['bound_claims_confirmed'] is True,'approved exact read CCR required')
+    scope=ccr['openbao'];require(scope['kv_path']=='platform/workloads/flex-auth/envelope-signing' and
+        scope['fields']==['ED25519_PRIVATE_KEY'] and scope['metadata_read'] is False and scope['policy_name']=='workload-kv-read-flex-auth-envelope-signing' and
+        scope['auth']['role']==ROLE and scope['auth']['bound_claims']=={'service_account_names':['flex-auth-envelope-signing-eso'],'service_account_namespaces':['flex-auth']},'read scope correspondence refused')
+    require((ROOT/scope['policy_file']).read_text()==module.generated_policy_hcl(ccr),'read policy source drift')
+
+
+def attended(io,contract):
+    _,status=io.bao(['status','-format=json']);require(status.get('cluster_id')==BAO_CLUSTER and status.get('sealed') is False,'backend cluster pin required')
+    _,lookup=io.bao(['token','lookup','-format=json']);who=lookup.get('data',{})
+    require('platform-admin' in who.get('policies',[]) and 'root' not in who.get('policies',[]) and
+        type(who.get('path')) is str and who['path'].startswith('auth/netkingdom/') and who.get('entity_id')==contract['founder_entity_id'],'exact attended non-root founder required')
+    _,audit=io.bao(['audit','list','-format=json']);require(type(audit) is dict and 0<len(audit)<=32,'enabled backend audit required')
+    # Token lookup-self is inside the contained warden child, never an agent
+    # retrieval. Its identifier stays in custody-process memory for fixed API I/O.
+    token=who.get('id');require(type(token) is str and 0<len(token)<=4096,'contained session handle unavailable')
+    io._token=token
+    _,policy=io.bao(['read','-format=json','sys/policies/acl/workload-kv-read-flex-auth-envelope-signing'])
+    expected_policy=(ROOT/'openbao/policies/workload-kv-read-flex-auth-envelope-signing.hcl').read_text()
+    require(_same_policy(policy.get('data',{}).get('policy'),expected_policy),'native reader policy drift')
+    _,role=io.bao(['read','-format=json','auth/kubernetes/role/'+ROLE]);role=role.get('data',{})
+    required={'bound_service_account_names':['flex-auth-envelope-signing-eso'],
+        'bound_service_account_namespaces':['flex-auth'],'bound_service_account_namespace_selector':'',
+        'audience':'openbao','token_policies':['workload-kv-read-flex-auth-envelope-signing'],
+        'token_ttl':900,'token_max_ttl':900,'token_explicit_max_ttl':900,'token_no_default_policy':True,
+        'token_period':0,'token_num_uses':0,'token_bound_cidrs':[]}
+    require(all(type(role.get(key)) is type(expected) and role.get(key)==expected for key,expected in required.items()),'native delivery role drift')
+
+
+class NativeWriter(SignerWriter):
+    def __init__(self,io,contract,directory):
+        self.io=io;self.contract=contract;self.final_ready=False
+        self._initialize_receipt(directory,contract['operation_id'],kid=contract['kid'],expected_version=0,bindings=contract['bindings'])
+    def _save(self,*,initial=False):
+        if not self.final_ready and self.receipt['status']=='custody-verified-pending-independent-public-admission':
+            self.receipt['status']='seed-verified-pending-metadata'
+        self.receipt.update(actor='founder-attended-platform-admin',native_custody=True,
+                            admission_sha256=digest(encoded(self.contract)),source_ccr_sha256=self.contract['ccr_sha256'])
+        return super()._save(initial=initial)
+    def _http(self,method,path,body=None):
+        if method=='POST':
+            validate_admission(self.contract);checked_sources(self.contract);fence(self.io,self.contract)
+        return self.io.http(method,path,body)
+
+
+def exact_metadata(status,value,*,created):
+    require(status==200 and type(value) is dict and 'errors' not in value and type(value.get('data')) is dict,'metadata readback unconfirmed')
+    data=value['data'];require(type(data.get('cas_required')) is bool and data['cas_required'] is True and
+        type(data.get('max_versions')) is int and data['max_versions']==2 and data.get('delete_version_after')=='0s' and
+        type(data.get('current_version')) is int and data['current_version']==(1 if created else 0),'metadata contract drift')
+
+
+def reconcile(io,contract,directory,original_path):
+    # A fresh independently approved recover action may read/reconcile only;
+    # original write receipt remains untouched and no writer/keygen is created.
+    raw=private_read(original_path);require(digest(raw)==contract['original_receipt_sha256'],'exact original receipt required')
+    receipt=parsed(raw)
+    receipt_fields={'schema','operation_id','bindings','writer_sha256','actor','created_at','path','expected_version','kid','status','production_admission','rollback','public_key','public_fingerprint','kv_version','exact_readback','native_custody','admission_sha256','source_ccr_sha256','phase'}
+    require(type(receipt) is dict and set(receipt)==receipt_fields and receipt.get('schema')=='railiance-platform.signer-cas-receipt.v1' and
+            receipt.get('status') in {'write-intent-unknown','seed-verified-pending-metadata'} and receipt.get('path')==PATH and
+            receipt.get('kid')==contract['kid'] and type(receipt.get('expected_version')) is int and receipt['expected_version']==0 and
+            (receipt.get('kv_version') is None or (type(receipt['kv_version']) is int and receipt['kv_version']==1)) and
+            type(receipt.get('exact_readback')) is bool and receipt.get('production_admission') is False and
+            receipt.get('source_ccr_sha256')==contract['original_execution']['source_ccr_sha256'] and receipt.get('writer_sha256')==contract['original_execution']['writer_sha256'] and
+            receipt.get('native_custody') is True and receipt.get('bindings')==contract['original_execution']['bindings'] and
+            receipt.get('phase') in {'metadata-setup','seed-cas'},'original custody receipt mismatch')
+    has_public=(type(receipt.get('public_key')) is str and re.fullmatch(r'[A-Za-z0-9_-]{43}',receipt['public_key']) is not None
+                and receipt.get('public_fingerprint')==digest(receipt['public_key'].encode()))
+    no_public=(receipt.get('public_key') is None and receipt.get('public_fingerprint') is None)
+    require((receipt['phase']=='metadata-setup' and no_public and receipt['kv_version'] is None and receipt['exact_readback'] is False)
+            or (receipt['phase']=='seed-cas' and has_public),'original phase/public lineage mismatch')
+    # Reserve recovery evidence before private native reads. A crash or failure
+    # leaves this independent attempt unconfirmed and cannot replay its ID.
+    recovery={'schema':'railiance-platform.signer-cas-recovery.v1','operation_id':contract['operation_id'],
+              'original_receipt_sha256':digest(raw),'original_execution':contract['original_execution'],
+              'admission_sha256':digest(encoded(contract)),
+              'status':'recovery-read-intent-unconfirmed','original_phase':receipt['phase'],
+              'kid':contract['kid'],'kv_version':None,'public_key':None,'public_fingerprint':None,
+              'key_custody_verified':False,'historical_effect':'unresolved',
+              'next_action':'fresh independently reviewed exact action required; no generation/write/retry authority',
+              'no_write_or_remint':True,'original_unknown_preserved':True}
+    store=object.__new__(SignerWriter);store.directory=directory;store.path=directory/(contract['operation_id']+'.json');store.receipt=recovery
+    store._save(initial=True)
+    status,value=io.http('GET',META)
+    if _policy_absent(status,value):
+        absent=True
+    else:
+        require(status==200 and type(value) is dict and 'errors' not in value and type(value.get('data')) is dict,'recovery metadata unconfirmed')
+        current=value['data'].get('current_version')
+        require(type(current) is int and current in {0,1},'recovery metadata version ambiguous')
+        exact_metadata(status,value,created=current==1)
+        absent=False
+        if current==1:require(has_public and receipt['phase']=='seed-cas','unexpected key custody after metadata-only phase')
+    if absent or current==0:
+        status,value=io.http('GET',PATH+'?version=1')
+        require(_policy_absent(status,value),'recovery data absence unconfirmed')
+        store.receipt.update(status='current-custody-absent' if absent else 'metadata-only-confirmed')
+        store._save()
+        return dict(store.receipt)
+    status,value=io.http('GET',PATH+'?version=1');require(status==200 and type(value) is dict and 'errors' not in value,'recovery private readback unconfirmed')
+    data=value.get('data',{});document=data.get('data');metadata=data.get('metadata',{})
+    require(type(document) is dict and set(document)==FIELDS and all(type(x) is str for x in document.values()) and
+            type(metadata.get('version')) is int and metadata['version']==1 and document['kid']==contract['kid'],'recovery version/shape mismatch')
+    try:
+        seed=base64.urlsafe_b64decode(document['ED25519_PRIVATE_KEY']+'=')
+        require(len(seed)==32 and b64(seed)==document['ED25519_PRIVATE_KEY'],'noncanonical recovery seed')
+        public=b64(Ed25519PrivateKey.from_private_bytes(seed).public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.Raw))
+    except Exception:raise Refused('recovery key correspondence unconfirmed') from None
+    require(hmac.compare_digest(public,document['ED25519_PUBLIC_KEY']) and hmac.compare_digest(public,receipt['public_key']) and
+            receipt.get('public_fingerprint')==digest(public.encode()),'recovery public correspondence unconfirmed')
+    store.receipt.update(status='custody-reconciled-pending-independent-public-admission',kv_version=1,public_key=public,public_fingerprint=digest(public.encode()),key_custody_verified=True)
+    store._save()
+    return dict(store.receipt)
+
+
+def execute(contract,directory,*,io=None,original_receipt=None):
+    remaining=validate_admission(contract);checked_sources(contract)
+    require(isinstance(directory,Path) and directory.absolute()==directory.resolve(),'exact private receipt directory required')
+    Journal._private(directory.stat(),directory=True)
+    plan=parsed((ROOT/PLAN).read_bytes())
+    require(str(directory)==plan['receipts']['directory'],'independently selected native receipt custody required')
+    if original_receipt is not None:require(Path(original_receipt).parent==directory,'original receipt custody mismatch')
+    # One native owner window; lock is custody concurrency only, not cluster CAS.
+    lock=os.open(directory/'.signer-window.lock',os.O_RDWR|os.O_CREAT|os.O_NOFOLLOW,0o600)
+    try:
+        Journal._private(os.fstat(lock));fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB)
+        with _absolute_budget(remaining):
+            io=NativeIO() if io is None else io
+            attended(io,contract);fence(io,contract)
+            if contract['action']=='recover':
+                require(original_receipt is not None,'independent original receipt required')
+                return reconcile(io,contract,directory,original_receipt)
+            require(original_receipt is None,'bootstrap cannot import recovery/private values')
+            writer=NativeWriter(io,contract,directory)
+            status,value=io.http('GET',META)
+            require(_policy_absent(status,value),'bootstrap exact metadata absence required')
+            writer.receipt.update(status='write-intent-unknown',phase='metadata-setup');writer._save()
+            try:
+                validate_admission(contract);checked_sources(contract);fence(io,contract)
+                status,value=io.http('POST',META,{'cas_required':True,'max_versions':2,'delete_version_after':'0s'})
+                require(status in {200,204},'metadata write unconfirmed')
+                exact_metadata(*io.http('GET',META),created=False)
+                writer.receipt['phase']='seed-cas'
+                writer.write()
+                exact_metadata(*io.http('GET',META),created=True)
+                writer.final_ready=True
+                writer.receipt.update(status='custody-verified-pending-independent-public-admission',exact_readback=True)
+                writer._save()
+                return dict(writer.receipt)
+            except BaseException:
+                writer.receipt.update(status='write-intent-unknown',exact_readback=False);writer._save()
+                raise Refused('native custody outcome unconfirmed') from None
+    finally:os.close(lock)
+
+
+class SilentParser(argparse.ArgumentParser):
+    def error(self,message):raise Refused('invalid custody command arguments')
+
+
+def main(argv=None):
+    parser=SilentParser(description=__doc__,add_help=False)
+    parser.add_argument('--admission',required=True);parser.add_argument('--admission-sha256',required=True)
+    parser.add_argument('--receipt-directory',required=True);parser.add_argument('--original-receipt')
+    # argparse output is captured by warden; all execution exceptions are silent.
+    try:
+        require(not sys.flags.optimize,'optimized custody interpreter refused')
+        args=parser.parse_args(argv);raw=private_read(args.admission)
+        require(is_digest(args.admission_sha256) and digest(raw)==args.admission_sha256,'independently reviewed admission digest required')
+        execute(parsed(raw),Path(args.receipt_directory),original_receipt=args.original_receipt)
+        return 0
+    except BaseException:return 1
+
+
+if __name__=='__main__':sys.exit(main())