11322a5256
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
80 lines
4.2 KiB
Python
80 lines
4.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Silent attended transfer of a real encrypted backup for isolated recovery."""
|
|
import argparse
|
|
import base64
|
|
import urllib.request
|
|
import shutil
|
|
import hashlib
|
|
import json
|
|
import zipfile
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
from migrate_nextcloud_backup_account import HOST, OPERATOR_PATH, LANE, NoRedirect, 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'])
|
|
directory.mkdir(mode=0o700)
|
|
encrypted = directory/'fetched.zip.age'
|
|
def stream(url, method, credentials, source_file=None, target=None):
|
|
require(url.startswith(HOST+'/'), 'unapproved_provider_origin')
|
|
headers={'Authorization':'Basic '+base64.b64encode((credentials[0]+':'+credentials[1]).encode()).decode()}
|
|
if source_file:
|
|
headers.update({'Content-Length':str(source.stat().st_size),'Content-Type':'application/octet-stream','If-None-Match':'*'})
|
|
req=urllib.request.Request(url,data=source_file,method=method,headers=headers)
|
|
with urllib.request.build_opener(NoRedirect()).open(req,timeout=1200) as response:
|
|
if target:
|
|
with target.open('xb') as out:
|
|
target.chmod(0o600)
|
|
shutil.copyfileobj(response,out,1024*1024)
|
|
else: response.read()
|
|
return response.status
|
|
with source.open('rb') as payload:
|
|
code=stream(values['NC_WEBDAV_URL']+'/forgejo/'+name,'PUT',
|
|
(values['NC_WEBDAV_TOKEN'],''),source_file=payload)
|
|
require(code == 201, 'new_offsite_upload_required')
|
|
receipt['upload_http_status']=code
|
|
code=stream(HOST+'/remote.php/dav/files/Backup/railiance-backups/forgejo/'+name,'GET',auth,target=encrypted)
|
|
def sha(path):
|
|
with path.open('rb') as f: return hashlib.file_digest(f,'sha256').hexdigest()
|
|
digest=sha(encrypted)
|
|
require(code==200 and digest==sha(source), 'offsite_download_mismatch')
|
|
receipt.update(download_http_status=code,ciphertext_bytes=encrypted.stat().st_size,ciphertext_sha256=digest)
|
|
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=1200)
|
|
require(result.returncode == 0, 'offsite_decryption_failed')
|
|
with zipfile.ZipFile(plain) as archive:
|
|
profile=json.loads(archive.read('backup-manifest.json'))['profile'] if 'backup-manifest.json' in archive.namelist() else 'full'
|
|
require(profile in ('full','essentials'), 'unknown_archive_profile')
|
|
receipt.update(archive_profile=profile,decrypted=True, plaintext_sha256=sha(plain), 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())
|