a252447058
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a1029d-2c93-7500-a0e0-e3e089668b29
133 lines
7.3 KiB
Python
133 lines
7.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent attended current-source and governed provider history recovery proof."""
|
|
import argparse
|
|
import configparser
|
|
import hmac
|
|
from urllib.parse import urlsplit,unquote
|
|
from datetime import datetime,timezone
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
|
|
from root_grant_attended import Bao, RECORD_ID, atomic_json, check_baseline, digest, require_attended
|
|
from migrate_nextcloud_backup_account import HOST, OPERATOR_PATH, LANE, quota, request
|
|
|
|
VALIDATOR_DIGEST='9156e399c8d92a45b817bfc3738c89843487ce57e1718ed747f804f95cc719bb'
|
|
SCHEMA_DIGEST='bb6cd6da5dc48eddcf118ae78c2cacfafd5f860098fd8a8368fe6bc845776d07'
|
|
|
|
|
|
def checked_bundle(state_path, validator):
|
|
state=json.loads(state_path.read_text())
|
|
if state.get('pending') or validator.validate(state['document']) or state['record_id']!=RECORD_ID or digest(state['document'])!=state['digest']:
|
|
raise ValueError('retained_state_invalid')
|
|
history=[]
|
|
for path in sorted((state_path.parent/'history').glob('*.json')):
|
|
if path.is_symlink() or path.stat().st_size>262144:
|
|
raise ValueError('history_invalid')
|
|
receipt=json.loads(path.read_text())
|
|
if validator.validate(receipt['document']) or receipt['record_id']!=RECORD_ID or receipt['digest']!=digest(receipt['document']):
|
|
raise ValueError('history_invalid')
|
|
history.append(receipt)
|
|
if state not in history:
|
|
raise ValueError('latest_history_missing')
|
|
return state,json.dumps({'baseline':state,'history':history},sort_keys=True,separators=(',',':')).encode()
|
|
|
|
|
|
def historical_config_comparison(config_path, values):
|
|
if not config_path.exists():
|
|
return {"historical_config_present":False,"historical_share_matches_current_lane":None}
|
|
try:
|
|
config=configparser.ConfigParser(interpolation=None)
|
|
config.read(config_path)
|
|
remote=config['nextcloud-backup']
|
|
candidates=[remote.get('user','')]+unquote(urlsplit(remote.get('url','')).path).split('/')
|
|
matched=any(hmac.compare_digest(value,values['NC_WEBDAV_TOKEN']) for value in candidates if value)
|
|
return {"historical_config_present":True,"historical_share_matches_current_lane":matched,
|
|
"historical_credential_status":"current-lane-incident" if matched else "historical-owner-and-revocation-unverified"}
|
|
except Exception:
|
|
return {"historical_config_present":True,"historical_share_matches_current_lane":None,
|
|
"historical_credential_status":"comparison-unavailable"}
|
|
|
|
|
|
def roundtrip(bundle, lane, operator, network=request, get_quota=quota):
|
|
values=lane['data']
|
|
if operator['BACKUP_USERNAME']!='Backup':
|
|
raise ValueError('governed_owner_mismatch')
|
|
auth=(operator['BACKUP_USERNAME'],operator['BACKUP_PASSWORD'])
|
|
upload=values['NC_WEBDAV_URL']
|
|
if not upload.startswith(HOST+'/public.php/dav/filesdrop/') or values['NC_WEBDAV_TOKEN']!=upload.rsplit('/',1)[-1]:
|
|
raise ValueError('current_lane_contract_mismatch')
|
|
key=(values['AGE_PRIVATE_KEY'].strip()+'\n').encode()
|
|
def process(command,payload):
|
|
result=subprocess.run(command,input=payload,capture_output=True,timeout=30)
|
|
if result.returncode:
|
|
raise ValueError('protected_process_failed')
|
|
return result.stdout
|
|
recipient=process(['age-keygen','-y','/dev/stdin'],key).decode().strip()
|
|
encrypted=process(['age','-r',recipient],bundle)
|
|
if len(encrypted)>1048576 or get_quota(auth)['available_bytes']<len(encrypted):
|
|
raise ValueError('backup_size_or_quota_invalid')
|
|
name='root-custody-history-'+datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')+'-'+uuid.uuid4().hex+'.json.age'
|
|
status,_=network(upload+'/forgejo/'+name,'PUT',encrypted,(values['NC_WEBDAV_TOKEN'],''),{'Content-Type':'application/octet-stream','If-None-Match':'*'})
|
|
if status!=201:
|
|
raise ValueError('new_artifact_upload_failed')
|
|
status,fetched=network(HOST+'/remote.php/dav/files/Backup/railiance-backups/forgejo/'+name,'GET',auth=auth)
|
|
if status!=200 or hashlib.sha256(fetched).digest()!=hashlib.sha256(encrypted).digest():
|
|
raise ValueError('provider_readback_mismatch')
|
|
with tempfile.TemporaryDirectory(prefix='root-custody-recovery-') as tmp:
|
|
cipher=Path(tmp)/'history.json.age'
|
|
cipher.write_bytes(fetched)
|
|
cipher.chmod(0o600)
|
|
restored=process(['age','-d','-i','/dev/stdin',str(cipher)],key)
|
|
if restored!=bundle:
|
|
raise ValueError('decrypted_history_mismatch')
|
|
return {'provider':'existing-governed-nextcloud-backup','artifact':name,'upload_http_status':201,'readback_http_status':200,'ciphertext_sha256':hashlib.sha256(encrypted).hexdigest(),'decrypted_history_matches':True,'lane_kv_version':lane['metadata']['version'],'local_historical_rclone_used':False}
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser()
|
|
p.add_argument('--state',type=Path,required=True)
|
|
p.add_argument('--net-kingdom',type=Path,required=True)
|
|
p.add_argument('--receipt',type=Path,required=True)
|
|
args=p.parse_args()
|
|
try:
|
|
if args.receipt.exists() or args.receipt.is_symlink():
|
|
return 1
|
|
require_attended()
|
|
if os.getenv('BAO_ADDR')!='http://127.0.0.1:18200':
|
|
return 1
|
|
source=args.net_kingdom/'tools/custody-record/validate.py'
|
|
schema=args.net_kingdom/'canon/schemas/custody-record_v0.1.schema.json'
|
|
if hashlib.sha256(source.read_bytes()).hexdigest()!=VALIDATOR_DIGEST or hashlib.sha256(schema.read_bytes()).hexdigest()!=SCHEMA_DIGEST:
|
|
return 1
|
|
spec=importlib.util.spec_from_file_location('custody_validator',source)
|
|
validator=importlib.util.module_from_spec(spec);spec.loader.exec_module(validator)
|
|
import fcntl
|
|
with (args.state.parent/'publisher.lock').open('a') as lock:
|
|
fcntl.flock(lock,fcntl.LOCK_EX)
|
|
state,bundle=checked_bundle(args.state,validator)
|
|
bao=Bao()
|
|
check_baseline(bao.read('platform/data/custody/net-kingdom/platform-root'),state,validator)
|
|
# Only current owner-held lane; no cached rclone config or caller token.
|
|
lane=bao.read(LANE)
|
|
operator=bao.read(OPERATOR_PATH)['data']
|
|
comparison=historical_config_comparison(Path('/home/worsch/.config/rclone/rclone.conf'),lane['data'])
|
|
if comparison['historical_share_matches_current_lane'] is True:
|
|
atomic_json(args.receipt,{'ok':False,'current_live_baseline_matches':True,**comparison,
|
|
'error':'current_backup_lane_requires_coordinated_existing_incident_recovery','recorded_at':datetime.now(timezone.utc).isoformat()})
|
|
return 1
|
|
receipt={**roundtrip(bundle,lane,operator),**comparison}
|
|
atomic_json(args.receipt,{'ok':True,'scope':'current-owner-read-and-governed-history-provider-roundtrip','record_id':RECORD_ID,'revision':state['revision'],'kv_version':state['kv_version'],'digest':state['digest'],'current_live_baseline_matches':True,'history_entries':len(json.loads(bundle)['history']),'recorded_at':datetime.now(timezone.utc).isoformat(),**receipt})
|
|
return 0
|
|
except Exception:
|
|
# Server/credential/HTTP/process diagnostics are never disclosed.
|
|
return 1
|
|
|
|
|
|
if __name__=='__main__':
|
|
raise SystemExit(main())
|