Prepare contained real-backup offsite retrieval for WP-0029

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/verify_nextcloud_offsite_restore.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Silent attended transfer of a real encrypted backup for isolated recovery."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+from migrate_nextcloud_backup_account import HOST, OPERATOR_PATH, LANE, request, quota, require
+from state_hub_preflight_lane import bao, data, LaneError
+
+
+def run(source, directory, receipt):
+    require(source.is_file() and source.name.endswith('.zip.age'), 'encrypted_backup_required')
+    entry = data(bao(['read', '-format=json', OPERATOR_PATH]))['data']['data']
+    require(entry['BACKUP_USERNAME'] == 'Backup', 'backup_owner_required')
+    auth = (entry['BACKUP_USERNAME'], entry['BACKUP_PASSWORD'])
+    receipt['quota_before'] = quota(auth)
+    lane = data(bao(['read', '-format=json', LANE]))['data']
+    values = lane['data']
+    require(source.stat().st_size < receipt['quota_before']['available_bytes'], 'insufficient_quota')
+    name = 'wp0029-recovery-' + source.name
+    receipt.update(artifact=name, source_backup=source.name, kv_version=lane['metadata']['version'])
+    payload = source.read_bytes()
+    code, _ = request(values['NC_WEBDAV_URL']+'/forgejo/'+name, 'PUT', payload,
+                      (values['NC_WEBDAV_TOKEN'], ''), {'Content-Type':'application/octet-stream','If-None-Match':'*'})
+    require(code == 201, 'new_offsite_upload_required')
+    receipt['upload_http_status'] = code
+    # Recovery must fetch from the owner path; never decrypt the local cache.
+    code, fetched = request(HOST+'/remote.php/dav/files/Backup/railiance-backups/forgejo/'+name, auth=auth)
+    require(code == 200 and fetched == payload, 'offsite_download_mismatch')
+    receipt.update(download_http_status=code, ciphertext_bytes=len(fetched), ciphertext_sha256=hashlib.sha256(fetched).hexdigest())
+    directory.mkdir(mode=0o700)
+    encrypted = directory/'fetched.zip.age'
+    encrypted.write_bytes(fetched)
+    encrypted.chmod(0o600)
+    plain = directory/'fetched.zip'
+    with plain.open('xb') as output:
+        plain.chmod(0o600)
+        result = subprocess.run(['age','-d','-i','/dev/stdin',str(encrypted)],
+            input=(values['AGE_PRIVATE_KEY'].strip()+'\n').encode(), stdout=output, stderr=subprocess.PIPE, timeout=180)
+    require(result.returncode == 0, 'offsite_decryption_failed')
+    receipt.update(decrypted=True, quota_after=quota(auth), status='offsite_fetched_pending_isolated_restore')
+
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__)
+    p.add_argument('--source',required=True,type=Path)
+    p.add_argument('--directory',required=True,type=Path)
+    p.add_argument('--receipt',required=True)
+    a=p.parse_args()
+    fd=os.open(a.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
+    receipt={'schema':'platform.real-offsite-recovery.v1','status':'failed'}
+    try: run(a.source,a.directory,receipt)
+    except Exception as error: receipt['error']=str(error) if isinstance(error,LaneError) else 'internal_error'
+    finally:
+        with os.fdopen(fd,'w') as f: json.dump(receipt,f,indent=2)
+    return int(receipt['status']=='failed')
+
+if __name__=='__main__': raise SystemExit(main())