"""One approved test-host restart-policy change; never starts or stops services.""" import argparse import copy import datetime import hashlib import json import os from pathlib import Path import sys TOOLS = Path('/usr/local/lib/contextra-maintenance/portable-20261001T182138') ROOT = Path(__file__).resolve().parent sys.path.insert(0, str(TOOLS)) import compose_backend as cb import field_profiles as fp import server_context as server from apply_source import normalize from data_safety import atomic, lock STATE = Path('/workspace/contextra-runtime-config/ct/runtime/state') POLICY = 'unless-stopped' def stamp(): return datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=9))).isoformat() def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def policy(row): p = row['HostConfig']['RestartPolicy'] return p['Name'] + (':' + str(p['MaximumRetryCount']) if p['Name'] == 'on-failure' and p['MaximumRetryCount'] else '') def invariants(rows): out = {} for row in rows: host = copy.deepcopy(row['HostConfig']) host.pop('RestartPolicy') out[row['Name']] = dict( id=row['Id'], image=row['Image'], pid=row['State']['Pid'], started=row['State']['StartedAt'], status=row['State']['Status'], oom=row['State']['OOMKilled'], config=cb.canonical(row['Config']), host=cb.canonical(host), mounts=cb.canonical(sorted(row['Mounts'], key=lambda m: m['Destination'])), networks=cb.canonical(row['NetworkSettings']['Networks'])) return out def channels(): code = '''import json,urllib.request with open('/run/bridge/control-token') as f: token=f.read().strip() r=urllib.request.Request('http://127.0.0.1:5000/internal/secure-control/runtime-status',headers={'Authorization':'Bearer '+token}) with urllib.request.urlopen(r,timeout=4) as s: d=json.load(s) print(json.dumps({'counts':d['streamCounts'],'streams':d['streams']}))''' return json.loads(cb.execute(['docker', 'exec', 'contextra-analytics-1', 'python3', '-c', code], timeout=8)) def inspect(): if os.geteuid() != 0 or server.project() != 'contextra' or cb.state_path() != STATE: raise RuntimeError('Wrong host/project or effective user') a, site, cp, approved, live, source = cb.inspect() if approved != live or set(live['services']) != set(server.ROLES): raise RuntimeError('Unreviewed Compose drift') for name in ('apply-pending.json', 'network-pending.json', 'pending.json'): if (fp.STATE / name).exists(): raise RuntimeError('Pending ct operation: ' + name) if json.loads(Path(a['physical']).read_text()) != server.physical(live): raise RuntimeError('Logical/physical Compose mismatch') rows = cb.containers() roles = {row['Config']['Labels']['com.docker.compose.service'] for row in rows} if len(rows) != 5 or roles != {server.service(x) for x in server.ROLES}: raise RuntimeError('Expected exactly five existing target containers') if any(row['State']['Status'] != 'running' for row in rows): raise RuntimeError('Only an in-place update of running containers is approved') return a, live, source, rows def make_plan(): a, live, source, rows = inspect() if (ROOT / 'plan.json').exists(): raise RuntimeError('Preserve existing plan') tx = 'autostart-' + datetime.datetime.now( datetime.timezone(datetime.timedelta(hours=9))).strftime('%Y%m%dT%H%M%S') plan = dict(time=stamp(), active_before=a, active_hash=digest(STATE/'active.json'), logical_hash=digest(Path(a['compose'])), physical_hash=digest(Path(a['physical'])), source_hash=cb.canonical(source), transaction=tx, invariants=invariants(rows), channels_before=channels(), policies_before={row['Name']:policy(row) for row in rows}, policy_after=POLICY) atomic(ROOT/'plan.json', plan) print(json.dumps(dict(status='PLANNED', transaction=tx, before=plan['policies_before'], after=POLICY))) def apply(): p = json.loads((ROOT/'plan.json').read_text()) if (ROOT/'receipt.json').exists(): raise RuntimeError('Existing receipt: inspect before any retry') a, live, source, rows = inspect() checks = [a == p['active_before'], digest(STATE/'active.json') == p['active_hash'], digest(Path(a['compose'])) == p['logical_hash'], digest(Path(a['physical'])) == p['physical_hash'], cb.canonical(source) == p['source_hash'], invariants(rows) == p['invariants'], channels() == p['channels_before']] if not all(checks): raise RuntimeError('Pre-apply drift: no mutation') candidate = copy.deepcopy(live) for cfg in candidate['services'].values(): cfg['restart'] = POLICY restored = copy.deepcopy(candidate) for role, cfg in live['services'].items(): if 'restart' in cfg: restored['services'][role]['restart'] = cfg['restart'] else: restored['services'][role].pop('restart', None) if restored != live: raise RuntimeError('A field other than restart was modified') tx = STATE/'transactions'/p['transaction'] tx.mkdir(mode=0o700) for name, value in (('compose.json', candidate), ('approved.json', candidate), ('compose.physical.json', server.physical(candidate))): atomic(tx/name, value) rendered = json.loads(cb.execute(cb.compose_args(tx/'compose.physical.json') + ['config', '--format', 'json'])) if normalize(server.logical(rendered)) != normalize(candidate): raise RuntimeError('Compose validation failed; active pointer unchanged') cb.guard_images(candidate) updated = dict(a, transaction=tx.name, compose=str(tx/'compose.json'), physical=str(tx/'compose.physical.json'), compose_semantic_sha256=cb.canonical(candidate)) receipt = dict(time=stamp(), phase='PREPARED', before=a, after=updated, policies_before=p['policies_before'], policy_after=POLICY, updated_containers=[], services_restarted=False, channels_changed=False) atomic(ROOT/'receipt.json', receipt) if digest(STATE/'active.json') != p['active_hash']: raise RuntimeError('Active pointer changed before commit') atomic(STATE/'active.json', updated) receipt['phase'] = 'PERSISTENT_APPLIED' atomic(ROOT/'receipt.json', receipt) cb.inspect() for row in rows: if policy(row) == POLICY: continue current = json.loads(cb.execute(['docker', 'inspect', row['Id']]))[0] if invariants([current]) != invariants([row]) or policy(current) != policy(row): raise RuntimeError('Live target changed; stop without automatic rollback') cb.execute(['docker', 'update', '--restart', POLICY, row['Id']], timeout=15) receipt['updated_containers'].append(row['Name']) atomic(ROOT/'receipt.json', receipt) final_a, final_doc, final_source, final_rows = inspect() verification = dict( persistent_policy=all(v['restart'] == POLICY for v in final_doc['services'].values()), runtime_policy=all(policy(r) == POLICY for r in final_rows), containers_unchanged=invariants(final_rows) == p['invariants'], channels_unchanged=channels() == p['channels_before'], input_source_unchanged=cb.canonical(final_source) == p['source_hash'], old_compose_unchanged=digest(Path(a['compose'])) == p['logical_hash'], old_physical_unchanged=digest(Path(a['physical'])) == p['physical_hash'], ct_drift_guard_passed=True) receipt.update(phase='VERIFIED' if all(verification.values()) else 'VERIFY_FAILED', finished=stamp(), verification=verification) atomic(ROOT/'receipt.json', receipt) if receipt['phase'] != 'VERIFIED': raise RuntimeError('Verification failed; inspect receipt') print(json.dumps(dict(status=receipt['phase'], verification=verification, updated_containers=receipt['updated_containers']))) if __name__ == '__main__': os.umask(0o077) parser = argparse.ArgumentParser() parser.add_argument('mode', choices=('plan', 'apply')) args = parser.parse_args() with lock(fp.STATE/'data-maintenance.lock'): (make_plan if args.mode == 'plan' else apply)()