a867ec269a
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883
107 lines
5.4 KiB
Python
107 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Bounded primary archive multipart PUT/GET using existing backup custody."""
|
|
import argparse
|
|
import base64
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import subprocess
|
|
from state_hub_preflight_lane import assert_cluster
|
|
|
|
ENDPOINT='https://s3.nl-ams.scw.cloud'
|
|
BUCKET='railiance-platform-pg-backup'
|
|
PREFIX='platform-pg/application-archives/forgejo/'
|
|
PART_SIZE=64*1024**2
|
|
|
|
|
|
def digest(path):
|
|
with path.open('rb') as stream:
|
|
return hashlib.file_digest(stream,'sha256').hexdigest()
|
|
|
|
|
|
def transfer(client, source, output, receipt, checkpoint=lambda: None):
|
|
expected=digest(source)
|
|
key=PREFIX+datetime.now(timezone.utc).strftime('%Y/%m/%d/%H%M%S-')+secrets.token_hex(12)+'.zip.age'
|
|
receipt.update(destination='s3://'+BUCKET+'/'+key, ciphertext_sha256=expected,
|
|
ciphertext_bytes=source.stat().st_size, stage='multipart_upload', uploaded_bytes=0)
|
|
checkpoint()
|
|
upload=None
|
|
try:
|
|
upload=client.create_multipart_upload(Bucket=BUCKET,Key=key,ContentType='application/octet-stream',Metadata={'sha256':expected})['UploadId']
|
|
parts=[]
|
|
with source.open('rb') as stream:
|
|
while block:=stream.read(PART_SIZE):
|
|
number=len(parts)+1
|
|
result=client.upload_part(Bucket=BUCKET,Key=key,UploadId=upload,PartNumber=number,Body=block)
|
|
parts.append({'PartNumber':number,'ETag':result['ETag']})
|
|
receipt['uploaded_bytes']+=len(block);checkpoint()
|
|
completed=client.complete_multipart_upload(Bucket=BUCKET,Key=key,UploadId=upload,MultipartUpload={'Parts':parts})
|
|
upload=None
|
|
version=completed.get('VersionId')
|
|
receipt.update(stage='primary_download',multipart_completed=True,version_pinned=bool(version),downloaded_bytes=0)
|
|
checkpoint()
|
|
args={'Bucket':BUCKET,'Key':key}
|
|
if version: args['VersionId']=version
|
|
response=client.get_object(**args)
|
|
if response['ContentLength']!=source.stat().st_size: raise ValueError('primary_length_mismatch')
|
|
with response['Body'] as body, output.open('xb') as target:
|
|
output.chmod(0o600)
|
|
while block:=body.read(8*1024**2):
|
|
target.write(block);receipt['downloaded_bytes']+=len(block);checkpoint()
|
|
if digest(output)!=expected: raise ValueError('primary_digest_mismatch')
|
|
receipt.update(status='primary_fetched_pending_application_restore',stage='transfer_verified',download_hash_matches=True)
|
|
checkpoint()
|
|
finally:
|
|
if upload:
|
|
try:
|
|
client.abort_multipart_upload(Bucket=BUCKET,Key=key,UploadId=upload)
|
|
receipt['multipart_aborted']=True
|
|
except Exception:
|
|
receipt['multipart_aborted']=False
|
|
checkpoint()
|
|
|
|
|
|
def main():
|
|
p=argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument('--source',required=True,type=Path);p.add_argument('--source-receipt',required=True,type=Path)
|
|
p.add_argument('--output',required=True,type=Path);p.add_argument('--receipt',required=True,type=Path)
|
|
p.add_argument('--kubeconfig',required=True)
|
|
a=p.parse_args()
|
|
receipt={'schema':'platform.forgejo-primary-archive.v1','status':'running','started_at':datetime.now(timezone.utc).isoformat()}
|
|
fd=os.open(a.receipt,os.O_CREAT|os.O_EXCL|os.O_WRONLY,0o600);os.close(fd)
|
|
def checkpoint(): a.receipt.write_text(json.dumps(receipt,indent=2)+'\n')
|
|
try:
|
|
receipt['stage']='source_validation';checkpoint()
|
|
source=json.loads(a.source_receipt.read_text())
|
|
if (source.get('status')!='offsite_fetched_pending_isolated_restore'
|
|
or not a.source.name.endswith('.zip.age') or a.output.exists()
|
|
or not 0<a.source.stat().st_size<=20*1024**3
|
|
or source.get('ciphertext_sha256')!=digest(a.source)):
|
|
raise ValueError('verified_source_required')
|
|
receipt['stage']='cluster_identity';checkpoint()
|
|
k=['kubectl','--kubeconfig',a.kubeconfig];assert_cluster(k)
|
|
# Existing governed databases delivery, captured only in process memory.
|
|
receipt['stage']='existing_credential_delivery';checkpoint()
|
|
r=subprocess.run(k+['-n','databases','get','secret','platform-pg-backup-s3','-o','json'],capture_output=True,timeout=30,check=True)
|
|
values=json.loads(r.stdout)['data']
|
|
receipt['stage']='client_setup';checkpoint()
|
|
import boto3
|
|
from botocore.config import Config
|
|
client=boto3.client('s3',endpoint_url=ENDPOINT,region_name='nl-ams',
|
|
aws_access_key_id=base64.b64decode(values['ACCESS_KEY_ID']).decode(),
|
|
aws_secret_access_key=base64.b64decode(values['ACCESS_SECRET_KEY']).decode(),
|
|
config=Config(request_checksum_calculation='when_required',response_checksum_validation='when_required',signature_version='s3v4',connect_timeout=15,read_timeout=180,retries={'max_attempts':3},s3={'addressing_style':'path'}))
|
|
transfer(client,a.source,a.output,receipt,checkpoint)
|
|
except Exception as error:
|
|
receipt['exception_type']=type(error).__name__
|
|
receipt.update(status='failed',error='bounded_primary_archive_transfer_failed');checkpoint()
|
|
return 1
|
|
finally:
|
|
receipt['finished_at']=datetime.now(timezone.utc).isoformat()
|
|
checkpoint()
|
|
return 0
|
|
|
|
if __name__=='__main__': raise SystemExit(main())
|