b9eaa20060
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a140c4-354a-7823-8f0d-8f717397f786
336 lines
18 KiB
Python
336 lines
18 KiB
Python
"""Owner-only cleanup recovery for the disposable native fixture.
|
|
|
|
Private, separately installed custody persists accessors, never bearer tokens.
|
|
This is a fixture implementation, not production custody/isolation admission.
|
|
The recovery owner supplies a fresh capability and credential independently of
|
|
public request/journal data. No business adapter or resumed admission is built.
|
|
"""
|
|
from __future__ import annotations
|
|
from dataclasses import asdict, dataclass, field
|
|
import hashlib
|
|
import hmac
|
|
import http.client
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import subprocess
|
|
import time
|
|
import uuid
|
|
|
|
from task_execution_journal import Binding, Cleanup, Journal, Refused
|
|
from task_execution_plan import cleanup_specs
|
|
|
|
|
|
def _encoded(value):
|
|
return json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()
|
|
|
|
|
|
def _digest(value):
|
|
return hashlib.sha256(_encoded(value)).hexdigest()
|
|
|
|
|
|
def _json_unique(pairs):
|
|
result={}
|
|
for key,value in pairs:
|
|
if key in result: raise Refused('duplicate private custody field')
|
|
result[key]=value
|
|
return result
|
|
|
|
|
|
def _invalid_constant(value):
|
|
raise Refused('nonfinite private custody value')
|
|
|
|
|
|
def _policy_absent(status, response):
|
|
# Exactly the native OpenBao empty-errors 404. Denial/malformed/error-shaped
|
|
# bodies never prove absence, including after a successful delete response.
|
|
return (status==404 and type(response) is dict and set(response)=={'errors'}
|
|
and type(response['errors']) is list and not response['errors'])
|
|
|
|
|
|
def _same_policy(actual, expected):
|
|
# OpenBao may omit one terminal LF from HCL source. Only that equivalence
|
|
# is admitted: spaces, tabs, CR and additional blank lines remain meaningful.
|
|
if type(actual) is not str or type(expected) is not str: return False
|
|
strip_one=lambda value:value[:-1] if value.endswith('\n') else value
|
|
return strip_one(actual)==strip_one(expected)
|
|
|
|
|
|
class FixtureRecoveryCustody:
|
|
"""Independent owner-private persistent fixture custody; no public API.
|
|
|
|
Mode/ownership and no-follow checks protect against accidental exposure.
|
|
Same-UID hostile code is NOT isolated. Production must supply independently
|
|
enforced owner storage/executor authority; this implementation refuses estate
|
|
backends and is never evidence of a production grant.
|
|
"""
|
|
def __init__(self, directory: Path, *, integrity_key: bytes):
|
|
if type(integrity_key) is not bytes or len(integrity_key)!=32:
|
|
raise Refused('independent owner capsule integrity key required')
|
|
self._integrity_key=integrity_key
|
|
self.directory = directory
|
|
if directory.is_symlink() or directory.absolute() != directory.resolve():
|
|
raise Refused('indirect recovery custody directory')
|
|
Journal._private(directory.stat(), directory=True)
|
|
|
|
def _path(self, binding):
|
|
if type(binding) is not Binding: raise Refused('exact recovery binding required')
|
|
binding.validate()
|
|
return self.directory / (binding.operation_id + '.private.json')
|
|
|
|
def _sync(self):
|
|
fd = os.open(self.directory, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try: os.fsync(fd)
|
|
finally: os.close(fd)
|
|
|
|
def read(self, binding):
|
|
try: fd = os.open(self._path(binding), os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
|
|
except OSError: raise Refused('private custody unavailable') from None
|
|
try:
|
|
before=os.fstat(fd)
|
|
Journal._private(before)
|
|
with os.fdopen(fd, 'rb', closefd=False) as stream: raw = stream.read(65537)
|
|
after=os.fstat(fd)
|
|
identity=lambda value:(value.st_dev,value.st_ino,value.st_size,value.st_mtime_ns,value.st_ctime_ns,value.st_mode,value.st_nlink)
|
|
if identity(before)!=identity(after) or len(raw)!=before.st_size:
|
|
raise Refused('private custody changed during read')
|
|
finally: os.close(fd)
|
|
if len(raw) > 65536: raise Refused('oversize private custody record')
|
|
try: capsule = json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant)
|
|
except Exception: raise Refused('invalid private custody record') from None
|
|
if (type(capsule) is not dict or set(capsule)!={'payload','mac'} or type(capsule['mac']) is not str
|
|
or len(capsule['mac'])!=64 or any(x not in '0123456789abcdef' for x in capsule['mac'])):
|
|
raise Refused('invalid authenticated private capsule')
|
|
expected_mac=hmac.new(self._integrity_key,_encoded(capsule['payload']),hashlib.sha256).hexdigest()
|
|
if not hmac.compare_digest(capsule['mac'],expected_mac):
|
|
raise Refused('private capsule integrity mismatch')
|
|
value=capsule['payload']
|
|
expected={'binding','cleanup','port','policy_hcl','identity_accessor','task_accessor','phase','revision'}
|
|
if type(value) is not dict or set(value)!=expected:
|
|
raise Refused('invalid private custody schema')
|
|
stored=value['binding']
|
|
if (type(stored) is not dict or set(stored)!=set(asdict(binding))
|
|
or any(type(v) is not str for v in stored.values())):
|
|
raise Refused('invalid private custody binding schema')
|
|
restored=Binding(**stored)
|
|
restored.validate()
|
|
if asdict(restored)!=asdict(binding): raise Refused('private custody binding mismatch')
|
|
if (type(value['port']) is not int or not 1<=value['port']<=65535
|
|
or type(value['policy_hcl']) is not str or not value['policy_hcl']
|
|
or type(value['identity_accessor']) is not str or not value['identity_accessor']
|
|
or type(value['phase']) is not str or value['phase'] not in {'prepared','mint-intent','issued'}
|
|
or type(value['revision']) is not int or value['revision']!={'prepared':0,'mint-intent':1,'issued':2}[value['phase']]
|
|
or (value['phase']=='issued' and (type(value['task_accessor']) is not str or not value['task_accessor']))
|
|
or (value['phase']!='issued' and value['task_accessor'] is not None)
|
|
or type(value['cleanup']) is not list or len(value['cleanup'])!=3):
|
|
raise Refused('invalid private custody schema')
|
|
for item in value['cleanup']:
|
|
if type(item) is not dict or set(item)!={'kind','call_digest'}:
|
|
raise Refused('invalid private cleanup schema')
|
|
Cleanup(**item).validate()
|
|
expected_cleanup=[{'kind':item['kind'],'call_digest':item['call_digest']}
|
|
for item in cleanup_specs(binding.operation_id)]
|
|
if value['cleanup']!=expected_cleanup:
|
|
raise Refused('private cleanup differs from deterministic owner commitment')
|
|
return value
|
|
|
|
def _write(self, binding, value, initial=False):
|
|
capsule={'payload':value,'mac':hmac.new(self._integrity_key,_encoded(value),hashlib.sha256).hexdigest()}
|
|
raw = _encoded(capsule)
|
|
if len(raw) > 65536: raise Refused('oversize private custody record')
|
|
path = self._path(binding)
|
|
if initial:
|
|
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
else:
|
|
# Existing owner-private record checked before replacement. Unique
|
|
# temp + fsync + replace + directory fsync never truncates old state.
|
|
self.read(binding)
|
|
tmp = self.directory / ('.pending-' + secrets.token_hex(16))
|
|
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
try:
|
|
with os.fdopen(fd, 'wb', closefd=False) as stream:
|
|
stream.write(raw); stream.flush(); os.fsync(fd)
|
|
finally: os.close(fd)
|
|
if not initial: os.replace(tmp, path)
|
|
self._sync()
|
|
|
|
def prepare(self, binding, cleanup, port, policy_hcl, identity_accessor):
|
|
value = {'binding': asdict(binding), 'cleanup': [asdict(x) for x in cleanup],
|
|
'port': port, 'policy_hcl': policy_hcl,
|
|
'identity_accessor': identity_accessor, 'task_accessor': None,
|
|
'phase': 'prepared', 'revision': 0}
|
|
self._write(binding, value, initial=True)
|
|
|
|
def checkpoint(self, binding, phase, task_accessor=None):
|
|
value = self.read(binding)
|
|
if (value['phase'], phase) not in {('prepared', 'mint-intent'), ('mint-intent', 'issued')}:
|
|
raise Refused('invalid protected recovery checkpoint')
|
|
if phase == 'issued' and (type(task_accessor) is not str or not task_accessor):
|
|
raise Refused('protected issued accessor required')
|
|
value.update(phase=phase, task_accessor=task_accessor, revision=value['revision'] + 1)
|
|
self._write(binding, value)
|
|
|
|
def revoke(self, permit):
|
|
if type(permit) is not FixtureRecoveryPermit or permit.custody is not self:
|
|
raise Refused('exact owner recovery permit required')
|
|
uuid.UUID(permit.attempt_id)
|
|
path=self.directory / (permit.attempt_id+'.revoked')
|
|
fd=os.open(path,os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW,0o600)
|
|
try:os.fsync(fd)
|
|
finally:os.close(fd)
|
|
self._sync()
|
|
|
|
def admit(self, binding, *, owner_token, server_process, valid_seconds=30):
|
|
"""Trusted recovery-owner action, not deserialization of caller flags.
|
|
|
|
Fresh admission seals the current private custody revision. The owner
|
|
supplies its credential and retained backend process separately. Neither
|
|
token nor private capsule digest is emitted to the public journal.
|
|
"""
|
|
value = self.read(binding)
|
|
if type(valid_seconds) is not int or not 1 <= valid_seconds <= 60:
|
|
raise Refused('bounded recovery admission required')
|
|
if type(owner_token) is not str or not owner_token:
|
|
raise Refused('independent recovery owner credential required')
|
|
return FixtureRecoveryPermit(str(uuid.uuid4()), binding, self, _digest(value),
|
|
owner_token, server_process, time.monotonic()+valid_seconds)
|
|
|
|
|
|
@dataclass(frozen=True, repr=False)
|
|
class FixtureRecoveryPermit:
|
|
attempt_id: str
|
|
binding: Binding
|
|
custody: FixtureRecoveryCustody
|
|
private_revision_digest: str = field(repr=False)
|
|
owner_token: str = field(repr=False)
|
|
server_process: subprocess.Popen = field(repr=False)
|
|
deadline: float
|
|
|
|
|
|
class _Recorder:
|
|
def __init__(self, journal, permit, value):
|
|
self.journal, self.permit = journal, permit
|
|
b = permit.binding
|
|
if journal._admission is not None or journal._consumed_admission is not None:
|
|
raise Refused('recovery cannot share a business admission')
|
|
if journal.observe(b) is None: raise Refused('original operation absent')
|
|
current = journal.cleanup_status(b)['obligations']
|
|
if [(x['kind'], x['call_digest']) for x in current] != [(x['kind'], x['call_digest']) for x in value['cleanup']]:
|
|
raise Refused('original cleanup commitment mismatch')
|
|
if all(x['outcome'] == 'confirmed' for x in current):
|
|
raise Refused('cleanup already confirmed')
|
|
self.current = current
|
|
# An authenticated older prepared capsule must not erase later issuance
|
|
# uncertainty. Without an issued accessor, any native call or state at
|
|
# the grant boundary makes absence unprovable. Recovery is conservative.
|
|
self.no_mint_proved=(journal.observe(b)['state'] in {'reserved','consume-confirmed','authenticating'}
|
|
and journal.db.execute('SELECT COUNT(*) FROM calls WHERE operation=?',(b.operation_id,)).fetchone()[0]==0)
|
|
db = journal.db
|
|
try: db.execute('INSERT INTO recovery_attempts VALUES (?,?,?)', (permit.attempt_id,b.operation_id,json.dumps(current,sort_keys=True)))
|
|
except Exception: raise Refused('recovery admission already used') from None
|
|
journal._sync_directory()
|
|
|
|
def before(self, ordinal, step, method, path):
|
|
digest = _digest({'kind': self.current[ordinal]['kind'], 'commitment':self.current[ordinal]['call_digest'], 'method':method,'path':path,'step':step})
|
|
self.journal.db.execute("INSERT INTO recovery_calls VALUES (?,?,?,?,'pending',NULL)", (self.permit.attempt_id,ordinal,step,digest))
|
|
self.journal._sync_directory()
|
|
return digest
|
|
|
|
def after(self, ordinal, step, status):
|
|
self.journal.db.execute("UPDATE recovery_calls SET outcome='complete', evidence=? WHERE attempt=? AND obligation=? AND step=? AND outcome='pending'", (_digest({'http_status':status}),self.permit.attempt_id,ordinal,step))
|
|
self.journal._sync_directory()
|
|
|
|
def finish(self, ordinal, confirmed):
|
|
evidence = _digest({'attempt':self.permit.attempt_id,'obligation':ordinal,'confirmed':confirmed})
|
|
db = self.journal.db
|
|
try:
|
|
db.execute('BEGIN IMMEDIATE')
|
|
pending = db.execute("SELECT COUNT(*) FROM recovery_calls WHERE attempt=? AND obligation=? AND outcome='pending'",(self.permit.attempt_id,ordinal)).fetchone()[0]
|
|
if confirmed and pending: raise Refused('pending recovery cannot confirm')
|
|
db.execute('INSERT INTO recovery_results VALUES (?,?,?,?)',(self.permit.attempt_id,ordinal,int(confirmed),evidence))
|
|
if confirmed:
|
|
item=self.current[ordinal]
|
|
db.execute("UPDATE cleanup SET outcome='confirmed', evidence=? WHERE operation=? AND ordinal=? AND kind=? AND call_digest=?",(evidence,self.permit.binding.operation_id,ordinal,item['kind'],item['call_digest']))
|
|
db.execute('COMMIT');self.journal._sync_directory()
|
|
except BaseException:
|
|
if db.in_transaction: db.execute('ROLLBACK')
|
|
raise
|
|
|
|
|
|
def recover_fixture_cleanup(journal: Journal, permit: FixtureRecoveryPermit):
|
|
"""Fresh independent owner admission: fixed cleanup only, no business retry.
|
|
|
|
Each attempt preserves original pending/unconfirmed rows in recovery_attempts
|
|
and native pending call rows unchanged. A fresh owner permit is required for
|
|
another attempt after failure; never autonomously retry uncertain mutations.
|
|
"""
|
|
if type(permit) is not FixtureRecoveryPermit or type(permit.custody) is not FixtureRecoveryCustody:
|
|
raise Refused('independent typed recovery admission required')
|
|
if str(uuid.UUID(permit.attempt_id)) != permit.attempt_id:
|
|
raise Refused('invalid recovery attempt identity')
|
|
value = permit.custody.read(permit.binding)
|
|
if _digest(value) != permit.private_revision_digest:
|
|
raise Refused('private custody revision changed')
|
|
proc=permit.server_process
|
|
if (type(proc) is not subprocess.Popen or proc.poll() is not None or type(proc.args) is not list
|
|
or '-dev' not in proc.args or '-dev-no-store-token' not in proc.args
|
|
or '-dev-listen-address=127.0.0.1:'+str(value['port']) not in proc.args):
|
|
raise Refused('same live disposable backend required')
|
|
def fresh():
|
|
if (permit.custody.directory / (permit.attempt_id+'.revoked')).exists():
|
|
raise Refused('recovery permit revoked')
|
|
if time.monotonic() >= permit.deadline: raise Refused('recovery admission expired')
|
|
if proc.poll() is not None: raise Refused('disposable backend lost')
|
|
if _digest(permit.custody.read(permit.binding)) != permit.private_revision_digest:
|
|
raise Refused('private custody changed during recovery')
|
|
fresh()
|
|
with journal.locked():
|
|
record = _Recorder(journal,permit,value)
|
|
for ordinal,item in enumerate(record.current):
|
|
if item['outcome']=='confirmed': continue
|
|
step=0
|
|
def call(method,path,body=None):
|
|
nonlocal step
|
|
fresh(); index=step; step+=1
|
|
record.before(ordinal,index,method,path)
|
|
conn=http.client.HTTPConnection('127.0.0.1',value['port'],timeout=2)
|
|
try:
|
|
conn.request(method,'/v1/'+path,body=_encoded(body) if body is not None else None,
|
|
headers={'X-Vault-Token':permit.owner_token,'Content-Type':'application/json'})
|
|
response=conn.getresponse();raw=response.read(1048577)
|
|
if len(raw)>1048576: raise Refused('oversize recovery response')
|
|
data=json.loads(raw,object_pairs_hook=_json_unique,parse_constant=_invalid_constant) if raw else {}
|
|
if type(data) is not dict: raise Refused('invalid recovery response')
|
|
status=response.status
|
|
except Exception: raise Refused('recovery transport or response unconfirmed') from None
|
|
finally:conn.close()
|
|
record.after(ordinal,index,status)
|
|
return status,data
|
|
confirmed=False
|
|
try:
|
|
if item['kind'] in {'task-token-revoke','identity-token-revoke'}:
|
|
accessor=value['task_accessor' if ordinal==0 else 'identity_accessor']
|
|
if accessor is None:
|
|
confirmed=ordinal==0 and value['phase']=='prepared' and record.no_mint_proved
|
|
else:
|
|
status,data=call('POST','auth/token/revoke-accessor',{'accessor':accessor})
|
|
# Revoke is idempotent for an already absent accessor;
|
|
# actual absence must still be independently read back.
|
|
if status in {200,204,400}:
|
|
status,data=call('POST','auth/token/lookup-accessor',{'accessor':accessor})
|
|
confirmed=status==400 and data.get('errors') in (['invalid accessor'],['1 error occurred:\n\t* invalid accessor\n\n'])
|
|
elif item['kind']=='grant-policy-delete':
|
|
path='sys/policies/acl/task-exec-'+permit.binding.operation_id
|
|
status,data=call('GET',path)
|
|
if _policy_absent(status,data):confirmed=True
|
|
elif (status==200 and not data.get('errors') and type(data.get('data')) is dict
|
|
and _same_policy(data['data'].get('policy'),value['policy_hcl'])):
|
|
status,data=call('DELETE',path)
|
|
if status in {200,204}:
|
|
status,data=call('GET',path);confirmed=_policy_absent(status,data)
|
|
except Exception:
|
|
# No backend errors/response/handle hashes enter public evidence.
|
|
confirmed=False
|
|
record.finish(ordinal,confirmed)
|
|
return journal.cleanup_status(permit.binding)
|