import hashlib import json import os from pathlib import Path import subprocess import sys import tarfile import time ROOT = Path(__file__).resolve().parent.parent REPO = Path('/workspace/contextra-analytics-runtime-lifecycle-refactor') PRIOR = ROOT.parent/'2026-11-03_contextra-offline-full-prep_r01' sys.path.insert(0, str(REPO)) from deployment_kit.core import KitError, write_json from deployment_kit.offline import image_export from prepare_evidence import snapshot LOCK = Path('/tmp/codex/ltb1021/2026-11-03_offline-export_r01/export-lock') REPORTS = ROOT/'reports'/sys.argv[1] if len(sys.argv)>1 else ROOT/'reports' IMAGES = ROOT/'images'/sys.argv[1] if len(sys.argv)>1 else ROOT/'images' def verify_archive(path, expected, image): """Read every tar member and trailing byte without extracting or loading an image.""" hashes = {} manifests = {} total = 0 outer = hashlib.sha256() class Reader: def __init__(self, stream): self.stream = stream def read(self, n=-1): data = self.stream.read(n); outer.update(data); return data with path.open('rb') as raw: reader = Reader(raw) with tarfile.open(fileobj=reader, mode='r|') as archive: for member in archive: name = member.name.removeprefix('./') assert name and not name.startswith('/') and '..' not in Path(name).parts assert name not in hashes, 'Duplicate member' if member.isdir(): continue assert member.isfile(), 'Unexpected link/special member' h = hashlib.sha256() data = bytearray() capture = name == 'manifest.json' or name == expected.removeprefix('sha256:')+'.json' or name == 'blobs/sha256/'+expected.removeprefix('sha256:') count = 0 with archive.extractfile(member) as stream: while True: block = stream.read(1024**2) if not block: break h.update(block); count += len(block) if capture: assert len(data)+len(block) <= 8*1024**2 data.extend(block) assert count == member.size total += count hashes[name] = h.hexdigest() if name.startswith('blobs/sha256/'): assert h.hexdigest() == name.split('/')[-1], 'OCI blob digest mismatch' if capture: manifests[name] = json.loads(data) while reader.read(1024**2): pass manifest = manifests['manifest.json'] matching = [m for m in manifest if hashes[m['Config']] == expected.removeprefix('sha256:')] assert len(matching) == 1 m = matching[0]; cfg = manifests[m['Config']] assert cfg['rootfs']['diff_ids'] == image['RootFS']['Layers'] assert [hashes[n] for n in m['Layers']] == [d.removeprefix('sha256:') for d in cfg['rootfs']['diff_ids']] return dict(sha256=outer.hexdigest(), members=len(hashes), payload_bytes=total, image=expected, layers=len(m['Layers']), full_read=True, config_id_match=True, layer_digests_match=True) def foreign_save(): found = [] for p in Path('/proc').iterdir(): if not p.name.isdigit(): continue try: argv = (p/'cmdline').read_bytes().split(b'\0') if argv and Path(os.fsdecode(argv[0])).name == 'docker' and b'save' in argv: ppid = int(next(s.split()[1] for s in (p/'status').read_text().splitlines() if s.startswith('PPid:'))) if ppid != os.getpid(): found.append(int(p.name)) except (OSError, StopIteration): continue return found def main(): if REPORTS != ROOT/'reports': assert sys.argv[1].startswith('r') and sys.argv[1][1:].isdigit() and len(sys.argv[1]) == 3 REPORTS.mkdir(mode=0o700); IMAGES.mkdir(mode=0o700) baseline = snapshot() spec = json.loads((PRIOR/'package-spec.pending.json').read_bytes()) assert baseline['files']['/etc/contextra-maintenance/server.json'] == spec['target']['definition_sha256'] current = {x['name']: x for x in baseline['containers']} for role, ident in spec['before_images'].items(): assert current['contextra-'+role+'-1']['image'] == ident jobs = [(role, ident) for role, ident in spec['before_images'].items() if role != 'analytics'] jobs += [('analytics-previous', spec['before_images']['analytics']), ('analytics-candidate', spec['after_images']['analytics'])] images = {role: json.loads(subprocess.check_output(['docker','image','inspect',ident],timeout=30))[0] for role, ident in jobs} records = [] last = [0] def guard(): if time.monotonic()-last[0] < 15: return last[0] = time.monotonic() if foreign_save(): raise KitError('FOREIGN_EXPORT', '다른 반출 작업이 있어 이번 작업을 중단합니다.') observed = snapshot() if observed['containers'] != baseline['containers'] or observed['files'] != baseline['files']: write_json(REPORTS/'observed-change.json', observed, new=True) raise KitError('OPERATIONAL_CHANGE', '운영 상태/설정 변경을 관측해 반출을 중단합니다. 자동복구하지 않습니다.') write_json(REPORTS/'export-baseline.json', baseline, new=True) write_json(REPORTS/'export-plans.json', {role:image_export.plan([ident],IMAGES/f'{role}.tar',test_preparation=True) for role,ident in jobs},new=True) try: for role, ident in jobs: last[0] = 0 path = IMAGES/f'{role}.tar' print(json.dumps(dict(stage='EXPORT', role=role, image=ident)), flush=True) result = image_export.export([ident], path, test_preparation=True, lock_directory=LOCK, guard=guard) print(json.dumps(dict(stage='VERIFY_ARCHIVE', role=role, bytes=result['bytes'], min_root_gib=result['min_root_free']/2**30)),flush=True) validation = verify_archive(path, ident, images[role]) assert validation['sha256'] == result['sha256'] write_json(REPORTS/f'archive-{role}.json', validation, new=True) records.append(dict(role=role, image=ident, path=str(path), **result, validation=validation)) write_json(REPORTS/'export-progress.json', dict(status='RUNNING', completed=records)) print(json.dumps(dict(stage='COMPLETE', role=role, sha256=result['sha256'])),flush=True) guard() except BaseException as exc: write_json(REPORTS/'export-progress.json', dict(status='BLOCKED', code=getattr(exc,'code',type(exc).__name__), completed=records)) raise write_json(REPORTS/'export-progress.json', dict(status='COMPLETE', completed=records)) write_json(REPORTS/'export-after.json', snapshot(), new=True) if __name__ == '__main__': main()