a9c1da4bd7
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
65 lines
2.9 KiB
Python
65 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Remove only the rejected WP-0029 drill copy after successful real recovery."""
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import urllib.error
|
|
import urllib.request
|
|
from migrate_nextcloud_backup_account import HOST, OPERATOR_PATH, NoRedirect, quota, require
|
|
from state_hub_preflight_lane import bao, data
|
|
|
|
ARTIFACT = 'wp0029-recovery-forgejo-dump-20260904T001507Z.zip.age'
|
|
SIZE = 147171923
|
|
|
|
|
|
def run(restore, receipt):
|
|
proof = json.loads(restore.read_text())
|
|
require(proof.get('status') == 'restored' and proof.get('cleanup') is True,
|
|
'successful_replacement_recovery_required')
|
|
require(proof.get('offsite_artifact') == 'wp0029-recovery-forgejo-dump-20260905-verified.zip.age',
|
|
'unexpected_replacement_artifact')
|
|
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'])
|
|
headers = {'Authorization': 'Basic '+base64.b64encode((auth[0]+':'+auth[1]).encode()).decode()}
|
|
url = HOST+'/remote.php/dav/files/Backup/railiance-backups/forgejo/'+ARTIFACT
|
|
opener = urllib.request.build_opener(NoRedirect())
|
|
receipt['quota_before'] = quota(auth)
|
|
try:
|
|
with opener.open(urllib.request.Request(url, method='HEAD', headers=headers), timeout=60) as response:
|
|
require(response.status == 200 and int(response.headers['Content-Length']) == SIZE,
|
|
'rejected_fixture_identity_mismatch')
|
|
etag = response.headers.get('ETag')
|
|
require(bool(etag) and not etag.startswith('W/'), 'strong_fixture_etag_required')
|
|
except urllib.error.HTTPError as error:
|
|
if error.code != 404:
|
|
raise
|
|
receipt.update(status='already_absent', artifact=ARTIFACT)
|
|
return
|
|
headers['If-Match'] = etag
|
|
with opener.open(urllib.request.Request(url, method='DELETE', headers=headers), timeout=60) as response:
|
|
require(response.status == 204, 'fixture_cleanup_failed')
|
|
receipt.update(status='removed', artifact=ARTIFACT, delete_http_status=204, quota_after=quota(auth))
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument('--restore-receipt', required=True, type=Path)
|
|
p.add_argument('--receipt', required=True)
|
|
args = p.parse_args()
|
|
receipt = {'schema':'platform.rejected-recovery-fixture-cleanup.v1', 'status':'failed'}
|
|
fd = os.open(args.receipt, os.O_WRONLY|os.O_CREAT|os.O_EXCL, 0o600)
|
|
try:
|
|
run(args.restore_receipt, receipt)
|
|
except Exception:
|
|
receipt['error'] = 'bounded_fixture_cleanup_failed'
|
|
finally:
|
|
with os.fdopen(fd, 'w') as output:
|
|
json.dump(receipt, output, indent=2)
|
|
return int(receipt['status'] == 'failed')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|