coulomb/railiance-platform
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Record archive recovery lifecycle and validate receipt provenance
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
This commit is contained in:
codex 2026-09-06 14:53:15 +02:00
parent 445f1361dc
commit a867ec269a
9 changed files with 203 additions and 11 deletions
Download patch file Download diff file Expand all files Collapse all files

18
scripts/restore_forgejo_offsite_locally.py
View file

@ -1,6 +1,7 @@
#!/usr/bin/env python3
"""Restore a fetched Forgejo archive on a disposable, internal Docker network."""
import argparse
from datetime import datetime, timezone
import configparser
import hashlib
import json
@ -189,13 +190,19 @@ def main():
p.add_argument('--archive',required=True,type=Path);p.add_argument('--receipt',required=True)
p.add_argument('--profile',choices=['full','essentials'],default='full')
p.add_argument('--transfer-receipt',required=True,type=Path)
a=p.parse_args();receipt={'schema':'platform.forgejo-isolated-restore.v1','status':'failed','images':[FORGE,POSTGRES]}
a=p.parse_args();receipt={'schema':'platform.forgejo-isolated-restore.v1','status':'failed','started_at':datetime.now(timezone.utc).isoformat(),'images':[FORGE,POSTGRES]}
fd=os.open(a.receipt,os.O_WRONLY|os.O_CREAT|os.O_EXCL,0o600)
try:
transfer=json.loads(a.transfer_receipt.read_text())
nextcloud = (transfer.get('status') == 'offsite_fetched_pending_isolated_restore'
raw=a.transfer_receipt.read_bytes()
transfer=json.loads(raw)
receipt['transfer_receipt_sha256']=hashlib.sha256(raw).hexdigest()
nextcloud = (transfer.get('schema') == 'platform.real-offsite-recovery.v1'
and transfer.get('decrypted') is True
and transfer.get('upload_http_status') == 201
and transfer.get('status') == 'offsite_fetched_pending_isolated_restore'
and transfer.get('download_http_status') == 200)
primary = (transfer.get('status') == 'primary_fetched_pending_application_restore'
primary = (transfer.get('schema') in ('platform.forgejo-primary-archive.v1', 'platform.forgejo-primary-decryption.v1')
and transfer.get('status') == 'primary_fetched_pending_application_restore'
and transfer.get('download_hash_matches') is True and transfer.get('decrypted') is True)
if (not (nextcloud or primary)
or transfer.get('plaintext_sha256') != file_digest(a.archive)):
@ -205,10 +212,13 @@ def main():
receipt['ciphertext_sha256']=transfer['ciphertext_sha256']
if transfer.get('archive_profile','full')!=a.profile: raise ValueError('profile_provenance_mismatch')
run(a.archive,receipt,a.profile)
if receipt.get('cleanup') is not True:
raise ValueError('restore_cleanup_incomplete')
except Exception:
receipt['status']='failed'
receipt['error']='isolated_restore_failed'
finally:
receipt['finished_at']=datetime.now(timezone.utc).isoformat()
with os.fdopen(fd,'w') as f: json.dump(receipt,f,indent=2)
return int(receipt['status'] not in ('restored','restored_essentials') or not receipt.get('cleanup'))