tegwick
a1939a7d06
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a13eab-1e6a-7b33-9c37-372e90ada3a2
88 lines
4.1 KiB
Python
88 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Fetch one fixed ciphertext through Platform's attended Backup-account lane.
|
|
|
|
No key read, plaintext, redirects, proxies, grants, deletion or production writes.
|
|
The object and byte count are bound to the reviewed inventory. Provider diagnostics
|
|
and credentials remain private. Output is a fresh private directory under /tmp.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import urllib.request
|
|
|
|
HERE=Path(__file__).resolve().parent
|
|
spec=importlib.util.spec_from_file_location('inventory',HERE/'inventory-storage-share.py')
|
|
owner=importlib.util.module_from_spec(spec);spec.loader.exec_module(owner)
|
|
OBJECT='r01-net-kingdom-pg/net-kingdom-pg-privacyidea_db-20261015T023000Z.dump.age'
|
|
SIZE=89129309
|
|
|
|
|
|
def metadata(response,url):
|
|
etag=response.headers.get('ETag','')
|
|
if (response.status!=200 or response.url!=url or
|
|
response.headers.get('Content-Length')!=str(SIZE) or
|
|
len(etag)>200 or not etag.startswith('"') or not etag.endswith('"') or
|
|
any(ord(c)<32 for c in etag)):
|
|
raise owner.Refused()
|
|
return etag
|
|
|
|
|
|
def transfer(opener,auth,destination):
|
|
url=owner.HOST+owner.BASE+OBJECT
|
|
headers={'Authorization':'Basic '+auth,'Accept-Encoding':'identity'}
|
|
with opener.open(urllib.request.Request(url,method='HEAD',headers=headers),timeout=30) as response:
|
|
etag=metadata(response,url)
|
|
headers['If-Match']=etag
|
|
digest=hashlib.sha256();count=0
|
|
with opener.open(urllib.request.Request(url,headers=headers),timeout=60) as response:
|
|
if metadata(response,url)!=etag:raise owner.Refused()
|
|
fd=os.open(destination,os.O_WRONLY|os.O_CREAT|os.O_EXCL|os.O_NOFOLLOW,0o600)
|
|
with os.fdopen(fd,'wb') as out:
|
|
while chunk:=response.read(1024*1024):
|
|
count+=len(chunk)
|
|
if count>SIZE:raise owner.Refused()
|
|
digest.update(chunk);out.write(chunk)
|
|
out.flush();os.fsync(out.fileno())
|
|
if count!=SIZE:raise owner.Refused()
|
|
with opener.open(urllib.request.Request(url,method='HEAD',headers=headers),timeout=30) as response:
|
|
if metadata(response,url)!=etag:raise owner.Refused()
|
|
# ETag used for request binding only; it is not asserted to be a content hash.
|
|
return {'ciphertext_bytes':count,'ciphertext_sha256':digest.hexdigest(),
|
|
'object_unchanged_during_transfer':True}
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__);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={'schema':'netkingdom.privacyidea-offhost-transfer.v1','status':'failed',
|
|
'started_at':datetime.now(timezone.utc).isoformat(),'provider':'Storage Share',
|
|
'object':OBJECT,'age_keys_read':False,'plaintext_created':False,
|
|
'production_writes':False,'source_sha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
|
|
'owner_accessor_sha256':hashlib.sha256((HERE/'inventory-storage-share.py').read_bytes()).hexdigest()}
|
|
destination=None
|
|
try:
|
|
identity,credentials=owner.caller_credentials()
|
|
r['caller_auth_path']=identity['path'];r['caller_entity_id']=identity.get('entity_id')
|
|
auth=base64.b64encode((credentials['BACKUP_USERNAME']+':'+credentials['BACKUP_PASSWORD']).encode()).decode()
|
|
destination=Path(tempfile.mkdtemp(prefix='nk49-offhost-',dir='/tmp'))/'privacyidea.dump.age'
|
|
opener=urllib.request.build_opener(urllib.request.ProxyHandler({}),owner.NoRedirect())
|
|
r.update(transfer(opener,auth,destination))
|
|
r.update(status='ciphertext_fetched',ciphertext_path=str(destination))
|
|
except Exception:
|
|
r['error']='bounded_transfer_failed_diagnostics_suppressed'
|
|
if destination is not None:
|
|
destination.unlink(missing_ok=True);destination.parent.rmdir()
|
|
finally:
|
|
r['finished_at']=datetime.now(timezone.utc).isoformat()
|
|
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())
|