#!/usr/bin/env python3 """Node-side quiesced coulomb-social snapshot. Binary output stays private. Run on railiance01 after reviewed live approval. Does not restore production. Default is server dry-run; --execute briefly pauses the app and captures a paired pg_dump/content archive. A separate isolated restore must validate it. """ from __future__ import annotations import argparse import fcntl import hashlib import json import os import signal import stat import subprocess import sys import time from pathlib import Path NS = 'coulomb-social' IMAGE = ('forgejo.coulomb.social/coulomb/coulomb-social@sha256:' 'b58d858abd5b4acbd6b893c3f622f7a9fdb5e606164ab9866eb3cfaa04df1a0e') CLUSTER_UID = 'a553c742-0115-43d4-99a4-a5ca56fe0786' LOCK_PATH = Path('/var/lib/coulomb-social-backup/snapshot.lock') class Refused(RuntimeError): pass def run(args: list[str], *, payload: bytes | None = None, timeout: int = 180) -> bytes: p = subprocess.run(args, input=payload, capture_output=True, timeout=timeout) if p.returncode: raise Refused('command_failed') return p.stdout def k(*args: str, payload: bytes | None = None) -> bytes: return run(['kubectl', *args], payload=payload) def document(*args: str) -> dict: return json.loads(k(*args, '-o', 'json')) def private_write(path: Path, value: bytes) -> None: fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, 'wb') as f: f.write(value) def snapshot() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--execute', action='store_true') parser.add_argument('--destination', type=Path, required=True) args = parser.parse_args() os.environ['KUBECONFIG'] = '/etc/rancher/k3s/k3s.yaml' started = time.monotonic() if document('get', 'namespace', 'kube-system')['metadata']['uid'] != CLUSTER_UID: raise Refused('wrong_cluster') dep = document('get', 'deployment', 'coulomb-social', '-n', NS) spec = dep['spec']['template']['spec'] if dep['spec']['replicas'] != 1 or spec['containers'][0]['image'] != IMAGE: raise Refused('unexpected_release') if document('get', 'pvc', 'coulomb-social-content', '-n', NS)['status']['phase'] != 'Bound': raise Refused('pvc_not_bound') cluster = document('get', 'cluster', 'apps-pg', '-n', 'databases') primary = cluster['status']['currentPrimary'] name = 'csoc-paired-snapshot-' + time.strftime('%Y%m%d%H%M%S', time.gmtime()) pod = { 'apiVersion': 'v1', 'kind': 'Pod', 'metadata': {'name': name, 'namespace': NS, 'labels': {'csoc-recovery': name}}, 'spec': { 'restartPolicy': 'Never', 'automountServiceAccountToken': False, 'securityContext': {'runAsNonRoot': True, 'runAsUser': 10001, 'runAsGroup': 10001, 'fsGroup': 10001, 'fsGroupChangePolicy': 'OnRootMismatch'}, 'containers': [{'name': 'snapshot', 'image': IMAGE, 'command': ['python', '-c', 'import time;time.sleep(1800)'], 'resources': {'requests': {'cpu': '10m', 'memory': '64Mi'}, 'limits': {'cpu': '100m', 'memory': '256Mi'}}, 'securityContext': {'allowPrivilegeEscalation': False, 'readOnlyRootFilesystem': True, 'capabilities': {'drop': ['ALL']}}, 'volumeMounts': [{'name': 'content', 'mountPath': '/snapshot', 'readOnly': True}]}], 'volumes': [{'name': 'content', 'persistentVolumeClaim': {'claimName': 'coulomb-social-content', 'readOnly': True}}], }, } body = json.dumps(pod).encode() k('create', '--dry-run=server', '-f', '-', payload=body) if not args.execute: print(json.dumps({'status': 'reviewed', 'server_dry_run': True, 'image_pinned': True, 'pvc_read_only': True, 'db_scope': 'coulomb_social_db', 'replicas_restored_to': 1})) return 0 dest = args.destination if not dest.is_absolute() or dest.exists(): raise Refused('destination_must_be_new_absolute_directory') dest.mkdir(mode=0o700) paused = False created = False receipt = {'status': 'incomplete', 'app': 'coulomb-social', 'database': 'coulomb_social_db', 'image': IMAGE, 'started_at': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())} try: paused = True k('scale', 'deployment/coulomb-social', '-n', NS, '--replicas=0') k('wait', '--for=delete', 'pod', '-n', NS, '-l', 'app.kubernetes.io/instance=coulomb-social', '--timeout=120s') created = True k('create', '-f', '-', payload=body) k('wait', '--for=condition=Ready', 'pod/' + name, '-n', NS, '--timeout=120s') dump = k('exec', '-n', 'databases', primary, '-c', 'postgres', '--', 'pg_dump', '-U', 'postgres', '-d', 'coulomb_social_db', '-Fc') if not dump.startswith(b'PGDMP'): raise Refused('invalid_pg_dump') private_write(dest / 'database.dump', dump) archive_code = ('import io,sys,tarfile; b=io.BytesIO(); ' 't=tarfile.open(fileobj=b,mode="w:gz"); ' 't.add("/snapshot",arcname="content"); ' 't.close();sys.stdout.buffer.write(b.getvalue())') content = k('exec', '-n', NS, name, '--', 'python', '-c', archive_code) private_write(dest / 'content.tar.gz', content) receipt.update(status='captured', writers_quiesced=True, database_sha256=hashlib.sha256(dump).hexdigest(), content_sha256=hashlib.sha256(content).hexdigest(), database_bytes=len(dump), content_bytes=len(content)) finally: try: if created: k('delete', 'pod', name, '-n', NS, '--wait=true', '--timeout=120s') finally: if paused: k('scale', 'deployment/coulomb-social', '-n', NS, '--replicas=1') k('rollout', 'status', 'deployment/coulomb-social', '-n', NS, '--timeout=180s') receipt['app_resumed'] = True receipt['elapsed_seconds'] = round(time.monotonic() - started, 2) private_write(dest / 'receipt.json', json.dumps(receipt, indent=2).encode()) print(json.dumps(receipt)) return 0 def main() -> int: if '--execute' not in sys.argv: return snapshot() LOCK_PATH.parent.mkdir(mode=0o700, parents=True, exist_ok=True) parent = LOCK_PATH.parent.lstat() if (not stat.S_ISDIR(parent.st_mode) or parent.st_uid != os.geteuid() or parent.st_mode & 0o077): raise Refused('lock_directory_not_private') fd = os.open(LOCK_PATH, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) with os.fdopen(fd, 'w') as lock: try: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise Refused('snapshot_already_running') from None def terminate(signum, frame): raise Refused('snapshot_interrupted') previous = signal.signal(signal.SIGTERM, terminate) try: return snapshot() finally: signal.signal(signal.SIGTERM, previous) if __name__ == '__main__': try: sys.exit(main()) except Exception as exc: # Never expose child output, kubectl objects, credentials or raw exception messages. print(json.dumps({'status': 'failed', 'failure_type': type(exc).__name__})) sys.exit(1)