tegwick
994d11c869
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
71 lines
3.5 KiB
Python
71 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent attended owner retrieval of an existing identity ciphertext.
|
|
|
|
Uses Backup-account recovery custody only in memory; no private age key, upload,
|
|
share, grant, deletion or production restore operation.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
from datetime import datetime,timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import tempfile
|
|
import urllib.request
|
|
|
|
HERE=Path(__file__).resolve().parent
|
|
|
|
|
|
def load(name):
|
|
spec=importlib.util.spec_from_file_location(name,HERE/(name+'.py'))
|
|
m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m);return m
|
|
|
|
|
|
inventory=load('inventory-storage-share');offsite=load('identity-offsite')
|
|
|
|
|
|
def retrieve(record,out):
|
|
if (record.get('schema')!='netkingdom.identity-offsite.v1' or record.get('status')!='readback_verified'
|
|
or not offsite.NAME.fullmatch(record.get('artifact','')) or not re.fullmatch('[a-f0-9]{64}',record.get('ciphertext_sha256',''))
|
|
or not 0<record.get('ciphertext_bytes',0)<=offsite.LIMIT):raise offsite.Refused()
|
|
identity,credentials=inventory.caller_credentials()
|
|
auth=base64.b64encode((credentials['BACKUP_USERNAME']+':'+credentials['BACKUP_PASSWORD']).encode()).decode()
|
|
url=offsite.HOST+'/remote.php/dav/files/Backup/railiance-backups/net-kingdom-identity/'+record['artifact']
|
|
h={'Authorization':'Basic '+auth,'Accept-Encoding':'identity'}
|
|
client=offsite.opener()
|
|
with client.open(urllib.request.Request(url,method='HEAD',headers=h),timeout=30) as response:
|
|
version=offsite.etag(response,url,record['ciphertext_bytes'])
|
|
if version!=record['etag']:raise offsite.Refused()
|
|
h['If-Match']=version
|
|
directory=Path(tempfile.mkdtemp(prefix='nk49-identity-fetched-',dir='/tmp'))
|
|
target=directory/record['artifact'];count=0;digest=hashlib.sha256()
|
|
with client.open(urllib.request.Request(url,headers=h),timeout=180) as response:
|
|
if offsite.etag(response,url,record['ciphertext_bytes'])!=version:raise offsite.Refused()
|
|
with target.open('xb') as output:
|
|
while block:=response.read(1024*1024):
|
|
count+=len(block)
|
|
if count>record['ciphertext_bytes']:raise offsite.Refused()
|
|
digest.update(block);output.write(block)
|
|
if count!=record['ciphertext_bytes'] or digest.hexdigest()!=record['ciphertext_sha256']:raise offsite.Refused()
|
|
with client.open(urllib.request.Request(url,method='HEAD',headers=h),timeout=30) as response:
|
|
if offsite.etag(response,url,record['ciphertext_bytes'])!=version:raise offsite.Refused()
|
|
out.update(record);out.update(status='readback_verified',fetched_path=str(target),
|
|
independent_owner_retrieval_verified=True,caller_entity_id=identity.get('entity_id'),
|
|
age_private_key_used=False,secret_values_exported=False,downloaded_at=datetime.now(timezone.utc).isoformat())
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--source-receipt',required=True);p.add_argument('--receipt',required=True)
|
|
a=p.parse_args();os.umask(0o077);fd=os.open(a.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
|
|
r={'status':'failed','source_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}
|
|
try:retrieve(json.loads(Path(a.source_receipt).read_text()),r)
|
|
except Exception:r.update(status='failed',error='contained_owner_retrieval_failed')
|
|
finally:
|
|
with os.fdopen(fd,'w') as out:json.dump(r,out,indent=2);out.write('\n')
|
|
return int(r['status']=='failed')
|
|
|
|
|
|
if __name__=='__main__':raise SystemExit(main())
|