From bd692378262e9db8da4e6c9b9f0552292b8e0846 Mon Sep 17 00:00:00 2001 From: Hannes Zietsman Date: Mon, 2 Nov 2026 12:25:10 +0200 Subject: [PATCH] fix: reconcile native service recovery before fleet updates --- .traceweaver/intent-contract.yml | 17 + docs/native-service-recovery.md | 42 ++ requirements.md | 19 + src/cryptolabs_proxy/update_recovery.py | 471 +++++++++++++++++++++ src/cryptolabs_proxy/updates.py | 331 +++++++++++++-- tests/test_proxy_customer_route_repairs.py | 2 + tests/test_update_recovery.py | 308 ++++++++++++++ tests/test_updates.py | 308 ++++++++++++++ traceability-matrix.md | 17 +- 9 files changed, 1477 insertions(+), 38 deletions(-) create mode 100644 docs/native-service-recovery.md create mode 100644 src/cryptolabs_proxy/update_recovery.py create mode 100644 tests/test_update_recovery.py diff --git a/.traceweaver/intent-contract.yml b/.traceweaver/intent-contract.yml index fc4c74e..c0b8f71 100644 --- a/.traceweaver/intent-contract.yml +++ b/.traceweaver/intent-contract.yml @@ -71,3 +71,20 @@ } ] } + +service_recovery_amendment_projection: + date: 2026-11-02 + authority_plan: ../../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/architecture/continuation-amendment.json + plan_sha256: 159af8421e4dde61e4ecc4f47849a1ddd7f07cd6fb158d247256f0bb36ed316e + owner_approval: ../../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/owner-approval.json + implementation_contract_sha256: 91b873c548da7d10a113c8fbb74f09540559ba51339514c738e081d0dc75cf01 + authority_gate: ../../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/execution/go/recovery-source-go.json + reviewed_vv: ../../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/execution/go/recovery-vv-root-reviewed.json + requirement: REQ-FLEETUPDATE-001 + trace: TRACE-FLEETUPDATE-001 + verification: VER-FLEETUPDATE-001 + validation: VAL-FLEETUPDATE-001 + source_only: true + native_acceptance: held + production_writes: disabled + publication: held diff --git a/docs/native-service-recovery.md b/docs/native-service-recovery.md new file mode 100644 index 0000000..6b78289 --- /dev/null +++ b/docs/native-service-recovery.md @@ -0,0 +1,42 @@ + +# Native IPMI service restoration evidence + +Install the reviewed exact DC Overview and CryptoLabs Proxy host wheels into +the same existing CLI environment. Keep the original proxy serving. Before +`release select --record-cli` or controller adoption, run locally as Linux root: + +```sh +dc-overview release reconcile-service --service ipmi-monitor --job +``` + +The command accepts only this service and a 32-character lowercase hex native +job ID. It accepts no path, image, script, force flag or caller observation. It +resolves the serving proxy's local writable `/data/auth` mount, then holds Fleet +lifecycle, dispatch and operation exclusion in that order. It does not invoke +status, replay a worker, stop a container, clear a marker or edit job history. + +It verifies exactly one trusted retained pre-job bundle below the fixed +`/var/backups/cryptolabs/ipmi132` root, its private artifact hashes, original +container/image/runtime/config/network identity, terminal helper, fresh `/health` +and read-only stable SQLite/account/link/inventory fingerprints. Missing, +ambiguous or changed evidence holds. It creates no new baseline or DB copy. + +The only new durable updater record is a private `service-recovery-JOB.json` +receipt. The old desired deployment remains failed. Retry revalidates current +evidence and returns the same receipt without changing its timestamp. Private +fingerprints and configuration values never appear in the command response. + +After separately approved compatible selection and proxy adoption, new native +submits require the live `native_service_recovery: 1` capability and carry the +job/receipt hash binding into their worker. Workers check before pulls and again +before the first lifecycle effect. Blocked successors retain the dependency. +Only a verified successful IPMI update retires that original dependency; archived +receipts are retained without permanently pinning future runtimes. + +New jobs persist bounded intent/completion diagnostics around Engine effects. +Unknown outcomes retain only safe categories and re-observe owned IDs before +restoration. No historical phases or incident cause are synthesized. + +Local source tests are not operational clearance. Exact packaged legacy +bootstrap, protected release, fresh target custody, actual signed admission and +site/browser acceptance are separate gates. diff --git a/requirements.md b/requirements.md index 9cdb220..3151841 100644 --- a/requirements.md +++ b/requirements.md @@ -60,3 +60,22 @@ Verification: `VER-PAIR-005`; validation: `VAL-PAIR-005`. See scoped authority a The implementation shall preserve existing working links and legacy clients, keep the current pinned WordPress package/publisher boundary, and deliver installed product updates through accepted DC Overview Fleet Management. Deployment and paid-service acceptance shall remain held until observed in the intended environment. Verification: `VER-PAIR-006`; validation: `VAL-PAIR-006`. See scoped authority and protocol for executable acceptance boundaries. + + +## 2 November 2026 native service recovery amendment + +The existing administrative REQ/TRACE/VER/VAL-FLEETUPDATE-001 labels also +project the exact owner-approved [continuation amendment](../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/architecture/continuation-amendment.json), +SHA256 `159af8421e4dde61e4ecc4f47849a1ddd7f07cd6fb158d247256f0bb36ed316e`, +and its [owner receipt](../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/owner-approval.json). The frozen implementation +contract is [contract.json](../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/execution/contract/contract.json), SHA256 +`91b873c548da7d10a113c8fbb74f09540559ba51339514c738e081d0dc75cf01`. +This is a linked projection, not a new approval or requirement identity. + +The fixed root host command verifies retained pre-job evidence and the exact +original service before publishing a private job-bound `original_restored` +receipt. The failed job and active pointer stay intact. Selector, native submit +and worker revalidate restoration and retain predecessor dependencies through +blocked successors. New-job phase diagnostics contain only allowlisted facts. +Source tests do not establish actual Linux bootstrap, release, target or browser +acceptance; those gates and publication remain held. diff --git a/src/cryptolabs_proxy/update_recovery.py b/src/cryptolabs_proxy/update_recovery.py new file mode 100644 index 0000000..4a3c5c4 --- /dev/null +++ b/src/cryptolabs_proxy/update_recovery.py @@ -0,0 +1,471 @@ +# TRACEWEAVER: file-role=native-service-recovery; req=REQ-FLEETUPDATE-001; trace=TRACE-FLEETUPDATE-001; ver=VER-FLEETUPDATE-001 +"""Private proof of restoration of an exact failed native service job. + +No lifecycle, Flask, updater or DC imports. Observations are produced by the +installed fixed observer, never accepted from command-line or HTTP callers. +""" +from contextlib import contextmanager +from dataclasses import dataclass +from datetime import datetime, timezone +import hashlib +import json +import math +import os +from pathlib import Path +import re +import stat +import time +import uuid + +from . import release_selection + +MAX_RECEIPT = 16384 +MAX_JOB = 8 * 1024 * 1024 +BASELINE_ROOT = Path('/var/backups/cryptolabs/ipmi132') +HASH_KEYS = {'runtime_config_sha256', 'host_config_sha256', 'mounts_sha256', + 'network_sha256', 'schema_sha256', 'settings_sha256', + 'stable_cloud_sha256', 'stable_inventory_sha256', + 'saved_identity_sha256', 'declared_config_sha256'} +CONTINUITY_KEYS = HASH_KEYS | {'projection', 'database_device', 'database_inode', + 'saved_identity_field_count', 'stored_key_digest_matches', 'sync_enabled'} +RECEIPT_KEYS = {'schema_version', 'kind', 'service', 'job_id', 'job_sha256', + 'active_sha256', 'selection_id', 'selection_sha256', 'manifest_sha256', + 'original_container_id', 'original_image_id', 'helper_name', + 'baseline_receipt_sha256', 'continuity', 'outcome', 'verified_at'} + + +def hold(): + raise ValueError('native_service_recovery_required') + + +def require(condition): + if not condition: + hold() + + +def encoded(value): + return json.dumps(value, sort_keys=True, separators=(',', ':'), + ensure_ascii=True, allow_nan=False).encode() + + +def sha256(raw): + return hashlib.sha256(raw).hexdigest() + + +def fingerprint(value): + return sha256(encoded(value)) + + +def _pairs(pairs): + value = {} + for key, item in pairs: + require(key not in value) + value[key] = item + return value + + +def _finite_float(value): + number = float(value) + require(math.isfinite(number)) + return number + + +def strict_json(raw, maximum=MAX_RECEIPT): + try: + require(isinstance(raw, bytes) and 0 < len(raw) <= maximum) + return json.loads(raw, object_pairs_hook=_pairs, parse_constant=lambda _: hold(), + parse_float=lambda value: _finite_float(value)) + except (UnicodeError, ValueError, TypeError, RecursionError): + hold() + + +def _pattern(value, pattern): + return type(value) is str and re.fullmatch(pattern, value) is not None + + +def validate_continuity(value): + require(type(value) is dict and set(value) == CONTINUITY_KEYS) + require(value['projection'] == 'ipmi-original-continuity-v1') + for key in HASH_KEYS: + require(_pattern(value[key], '[0-9a-f]{64}')) + require(type(value['database_device']) is int and value['database_device'] >= 0) + require(type(value['database_inode']) is int and value['database_inode'] > 0) + require(type(value['saved_identity_field_count']) is int and value['saved_identity_field_count'] == 5) + require(value['stored_key_digest_matches'] is True and value['sync_enabled'] is False) + return value + + +# TRACEWEAVER: entrypoint=validate_receipt; req=REQ-FLEETUPDATE-001; trace=TRACE-FLEETUPDATE-001; ver=VER-FLEETUPDATE-001 +def validate_receipt(value): + if isinstance(value, bytes): + value = strict_json(value) + require(type(value) is dict and set(value) == RECEIPT_KEYS) + require(type(value['schema_version']) is int and value['schema_version'] == 1) + require(value['kind'] == 'native-service-original-restored' and value['service'] == 'ipmi-monitor') + require(value['outcome'] == 'original_restored') + for key in ('job_sha256', 'active_sha256', 'selection_sha256', 'manifest_sha256', + 'baseline_receipt_sha256', 'original_container_id'): + require(_pattern(value[key], '[0-9a-f]{64}')) + require(_pattern(value['job_id'], '[0-9a-f]{32}')) + require(_pattern(value['selection_id'], '[A-Za-z0-9][A-Za-z0-9_.-]{0,79}')) + require(_pattern(value['original_image_id'], 'sha256:[0-9a-f]{64}')) + require(value['helper_name'] == 'cryptolabs-update-' + value['job_id']) + require(_pattern(value['verified_at'], r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?Z')) + try: + datetime.fromisoformat(value['verified_at'].replace('Z', '+00:00')) + except ValueError: + hold() + validate_continuity(value['continuity']) + return value + + +def _private(info, *, directory=False, root=False): + require(info.st_uid == (0 if root else os.geteuid())) + require(stat.S_IMODE(info.st_mode) == (0o700 if directory else 0o600)) + require(stat.S_ISDIR(info.st_mode) if directory else stat.S_ISREG(info.st_mode) and info.st_nlink == 1) + + +@contextmanager +def open_existing_directory(path, *, root=False): + """Never create stores; traverse no-follow descriptors and trusted ancestors.""" + path = Path(path) + require(path.is_absolute() and '..' not in path.parts) + fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY) + try: + for index, part in enumerate(path.parts[1:]): + child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd) + os.close(fd) + fd = child + info = os.fstat(fd) + if index == len(path.parts) - 2: + _private(info, directory=True, root=root) + elif root: + require(info.st_uid == 0 and not info.st_mode & 0o022) + yield fd + except (OSError, KeyError, TypeError): + hold() + finally: + os.close(fd) + + +def read_private(directory_fd, name, maximum=MAX_RECEIPT, *, root=False): + require(type(name) is str and '/' not in name and name not in ('.', '..')) + fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory_fd) + try: + before = os.fstat(fd) + _private(before, root=root) + require(before.st_size <= maximum) + raw = b'' + while len(raw) <= maximum: + chunk = os.read(fd, min(65536, maximum + 1 - len(raw))) + if not chunk: + break + raw += chunk + after = os.fstat(fd) + current = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + require(len(raw) <= maximum and (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) == + (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns)) + require((before.st_dev, before.st_ino) == (current.st_dev, current.st_ino)) + return raw + finally: + os.close(fd) + + +def validate_job(raw, job_id): + require(_pattern(job_id, '[0-9a-f]{32}')) + job = strict_json(raw, MAX_JOB) + try: + require(job['id'] == job_id and job['helper'] == 'cryptolabs-update-' + job_id) + require(job['service'] == 'ipmi-monitor' and job['action'] == 'update') + require(job['state'] == 'failed' and job['success'] is False) + require(job['results']['ipmi-monitor']['state'] == 'recovery_required') + original = job['sources']['ipmi-monitor'] + require(_pattern(original['container_id'], '[0-9a-f]{64}')) + require(_pattern(original['image_id'], 'sha256:[0-9a-f]{64}')) + stored = job['selection_snapshot'] + snapshot = release_selection.validate(stored['selection_bytes'].encode(), stored['manifest_bytes'].encode(), + selection_id=job['selection_id'], selection_sha256=job['selection_sha256']) + require(snapshot['manifest_sha256'] == job['manifest_sha256']) + target = job['targets']['ipmi-monitor'] + require(type(target) is dict and target.get('owner') == 'proxy-updater') + expected = release_selection.targets(snapshot, target['platform']) + require(expected['ipmi-monitor'] == target) + except (KeyError, TypeError, AttributeError, release_selection.SelectionError): + hold() + return job + + +def load_bound_job(directory_fd, job_id): + require(_pattern(job_id, '[0-9a-f]{32}')) + raw = read_private(directory_fd, job_id + '.json', MAX_JOB) + active = read_private(directory_fd, 'active.json') + pointer = strict_json(active) + require(type(pointer) is dict and set(pointer) == {'id'} and pointer['id'] == job_id) + validate_job(raw, job_id) + return raw, active + + +@dataclass(frozen=True) +class OriginalObservation: + container_id: str + image_id: str + helper_terminated: bool + healthy: bool + continuity: dict + + +def validate_original_restored(job_raw, receipt_raw, observation): + value = validate_receipt(receipt_raw) + job = validate_job(job_raw, value['job_id']) + require(isinstance(observation, OriginalObservation)) + require(value['job_sha256'] == sha256(job_raw)) + for key in ('selection_id', 'selection_sha256', 'manifest_sha256'): + require(value[key] == job[key]) + original = job['sources']['ipmi-monitor'] + require(value['original_container_id'] == original['container_id'] == observation.container_id) + require(value['original_image_id'] == original['image_id'] == observation.image_id) + require(observation.helper_terminated is True and observation.healthy is True) + require(validate_continuity(observation.continuity) == value['continuity']) + return value + + +def publish_original_restored(directory_fd, receipt): + """Both caller locks must be held. No replace of a conflicting final file.""" + validate_receipt(receipt) + _private(os.fstat(directory_fd), directory=True) + name = 'service-recovery-' + receipt['job_id'] + '.json' + raw = encoded(receipt) + b'\n' + require(len(raw) <= MAX_RECEIPT) + try: + existing = read_private(directory_fd, name) + except FileNotFoundError: + existing = None + if existing is not None: + require(validate_receipt(existing) == receipt) + os.fsync(directory_fd) + require(read_private(directory_fd, name) == existing) + return existing + temporary = '.service-recovery-' + uuid.uuid4().hex + '.tmp' + descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory_fd) + try: + os.fchmod(descriptor, 0o600) + with os.fdopen(descriptor, 'wb') as stream: + stream.write(raw) + stream.flush() + os.fsync(stream.fileno()) + # link is atomic create-if-absent; replace would overwrite a competing final. + os.link(temporary, name, src_dir_fd=directory_fd, dst_dir_fd=directory_fd, follow_symlinks=False) + os.unlink(temporary, dir_fd=directory_fd) + os.fsync(directory_fd) + require(read_private(directory_fd, name) == raw) + return raw + finally: + try: + os.unlink(temporary, dir_fd=directory_fd) + except FileNotFoundError: + pass + + +def runtime_fingerprints(runtime): + try: + require(type(runtime['Config']) is dict and type(runtime['HostConfig']) is dict and + type(runtime['Mounts']) is list and type(runtime['NetworkSettings']['Networks']) is dict) + mounts = sorted(runtime['Mounts'], key=lambda m: (m.get('Destination', ''), m.get('Source', ''), encoded(m))) + networks = {name: {'NetworkID': entry['NetworkID'], 'Aliases': sorted(entry.get('Aliases') or [])} + for name, entry in runtime['NetworkSettings']['Networks'].items()} + return {'runtime_config_sha256': fingerprint(runtime['Config']), + 'host_config_sha256': fingerprint(runtime['HostConfig']), + 'mounts_sha256': fingerprint(mounts), + 'network_sha256': fingerprint({'mode': runtime['HostConfig'].get('NetworkMode'), 'networks': networks})} + except (KeyError, TypeError, ValueError): + hold() + + +# Static packaged read-only program. Its only entrypoint uses the fixed service +# DB/config/health paths. No caller SQL, path, shell, or source code is accepted. +DATABASE_PROGRAM = r''' +import base64 +import hashlib +import json +import os +from pathlib import Path +import sqlite3 +import stat +import time + +def _encode(value): + return json.dumps(value,sort_keys=True,separators=(',',':'),allow_nan=False, + default=lambda v: {'sqlite_blob_base64':base64.b64encode(v).decode()} if isinstance(v,bytes) else str(v)).encode() + +def _fingerprint(value): + return hashlib.sha256(_encode(value)).hexdigest() + +def _need(value): + if not value: raise ValueError('native_service_recovery_required') + +def database_fingerprints(database): + database=Path(database) + before=database.lstat() + _need(stat.S_ISREG(before.st_mode) and not database.is_symlink()) + deadline=time.monotonic()+5 + connection=sqlite3.connect(database.as_uri()+'?mode=ro',uri=True,timeout=5) + connection.set_progress_handler(lambda:int(time.monotonic()>deadline),500) + try: + connection.execute('PRAGMA query_only=ON') + _need(connection.execute('PRAGMA query_only').fetchone()[0]==1) + connection.execute('BEGIN') + schema=[];schema_size=0 + for item in connection.execute('SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY type,name'): + schema_size+=len(_encode(item));_need(schema_size<=8*1024*1024 and time.monotonic()<=deadline) + schema.append(item) + tables={};size=0 + for name in ('system_settings','cloud_sync','server'): + definition=connection.execute("SELECT sql FROM sqlite_master WHERE type='table' AND name=?",(name,)).fetchone() + _need(definition and definition[0] and not definition[0].lstrip().upper().startswith('CREATE VIRTUAL TABLE')) + info=connection.execute('PRAGMA table_xinfo("'+name+'")').fetchall() + _need(info and all(r[6]==0 for r in info)) + columns=[r[1] for r in info] + primary=[r[1] for r in sorted(info,key=lambda r:r[5]) if r[5]] + quote=lambda n:'"'+n.replace('"','""')+'"' + order=','.join(map(quote,primary)) if primary else 'rowid' + rows=[] + for row in connection.execute('SELECT '+','.join(map(quote,columns))+' FROM "'+name+'" ORDER BY '+order): + size+=len(_encode(row));_need(size<=8*1024*1024 and time.monotonic()<=deadline) + rows.append(row) + tables[name]=(columns,rows) + settings=[dict(zip(tables['system_settings'][0],r)) for r in tables['system_settings'][1]] + keys=('ai_link_email','ai_link_user_id','ai_link_customer_id','ai_link_linked_at','ai_link_key_digest') + saved={r['key']:r['value'] for r in settings if r['key'] in keys} + _need(set(saved)==set(keys) and all(type(v) is str and v for v in saved.values())) + columns,rows=tables['cloud_sync'];_need(len(rows)==1) + cloud=dict(zip(columns,rows[0]));credential=cloud.get('license_key') + _need(type(credential) is str and credential and cloud.get('sync_enabled') in (0,False)) + _need(saved['ai_link_key_digest']==hashlib.sha256(credential.encode()).hexdigest()) + stable_cloud_columns=[n for n in columns if n not in ('last_sync','last_sync_status','last_sync_message','updated_at')] + stable_cloud_rows=[tuple(dict(zip(columns,r))[n] for n in stable_cloud_columns) for r in rows] + columns,rows=tables['server'] + stable_inventory_columns=[n for n in columns if n not in ('status','updated_at')] + stable_inventory_rows=[tuple(dict(zip(columns,r))[n] for n in stable_inventory_columns) for r in rows] + result={'schema_sha256':_fingerprint(schema), + 'settings_sha256':_fingerprint(sorted((r['key'],r['value']) for r in settings)), + 'stable_cloud_sha256':_fingerprint([stable_cloud_columns,stable_cloud_rows]), + 'stable_inventory_sha256':_fingerprint([stable_inventory_columns,stable_inventory_rows]), + 'saved_identity_sha256':_fingerprint(saved),'saved_identity_field_count':5, + 'stored_key_digest_matches':True,'sync_enabled':False} + finally: + connection.rollback();connection.close() + after=database.lstat() + _need((before.st_dev,before.st_ino)==(after.st_dev,after.st_ino)) + result.update(database_device=before.st_dev,database_inode=before.st_ino) + return result + +def fixed_observation(): + from urllib.request import urlopen + database=Path('/app/data/ipmi_events.db') + result=database_fingerprints(database) + info=database.lstat() + _need(result['database_device']==info.st_dev and result['database_inode']==info.st_ino) + declared={} + for name in ('/app/config/servers.yaml','/app/data/servers.json'): + path=Path(name) + try: + fd=os.open(path,os.O_RDONLY|os.O_NOFOLLOW|os.O_NONBLOCK) + except FileNotFoundError: + declared[name]={'present':False};continue + try: + before=os.fstat(fd);_need(stat.S_ISREG(before.st_mode) and before.st_size<=8*1024*1024) + raw=b'' + while len(raw)<=8*1024*1024: + part=os.read(fd,65536) + if not part:break + raw+=part + after=os.fstat(fd);current=path.lstat() + _need(len(raw)<=8*1024*1024 and (before.st_dev,before.st_ino,before.st_size,before.st_mtime_ns)==(after.st_dev,after.st_ino,after.st_size,after.st_mtime_ns)) + _need((before.st_dev,before.st_ino)==(current.st_dev,current.st_ino)) + declared[name]={'present':True,'sha256':hashlib.sha256(raw).hexdigest()} + finally:os.close(fd) + result['declared_config_sha256']=_fingerprint(declared) + with urlopen('http://127.0.0.1:5000/health',timeout=3) as response: + _need(response.status==200) + health=json.loads(response.read(16385));_need(type(health) is dict) + _need(health.get('status')=='healthy' and health.get('checks',{}).get('database')=='ok') + from datetime import datetime,timezone + last=datetime.fromisoformat(health['last_collection'].replace('Z','+00:00')) + if last.tzinfo is None:last=last.replace(tzinfo=timezone.utc) + _need(0<=(datetime.now(timezone.utc)-last).total_seconds()<=180) + return {'continuity':result,'last_collection':last.isoformat()} +''' +_database_namespace = {} +exec(compile(DATABASE_PROGRAM, '', 'exec'), _database_namespace) +database_fingerprints = _database_namespace['database_fingerprints'] +FIXED_PROBE = DATABASE_PROGRAM + "\ntry:\n print(json.dumps(fixed_observation(),sort_keys=True))\nexcept Exception:\n raise SystemExit(1)\n" + + +def _timestamp(value): + parsed = datetime.fromisoformat(value.replace('Z', '+00:00')) + require(parsed.tzinfo is not None) + return parsed.timestamp() + + +def discover_baseline(job, runtime, observed, *, deadline=None): + """Verify one pre-job private bundle; never create or refresh a baseline.""" + original = job['sources']['ipmi-monitor'] + matches = [] + deadline = time.monotonic() + 90 if deadline is None else deadline + with open_existing_directory(BASELINE_ROOT, root=True) as parent_fd: + names = [] + with os.scandir(parent_fd) as entries: + for entry in entries: + names.append(entry.name) + require(len(names) <= 64) + for name in names: + require(time.monotonic() <= deadline) + info = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) + require(stat.S_ISDIR(info.st_mode)) + with open_existing_directory(BASELINE_ROOT / name, root=True) as fd: + raw = read_private(fd, 'receipt.json', MAX_JOB, root=True) + baseline = strict_json(raw, MAX_JOB) + require(baseline.get('status') == 'verified-bounded-selected-state-protection') + artifacts = baseline.get('artifacts') + require(type(artifacts) is list and 2 <= len(artifacts) <= 4) + records = {} + for entry in artifacts: + require(time.monotonic() <= deadline) + artifact = Path(entry['path']).name + require(artifact in {'runtime-config.private.json', 'selected-state.private.sqlite', 'config-0.private', 'config-1.private'} and artifact not in records) + maximum = 16 * 1024 * 1024 if artifact.endswith('.sqlite') else MAX_JOB + data = read_private(fd, artifact, maximum, root=True) + require(type(entry.get('bytes')) is int and len(data) == entry['bytes'] and sha256(data) == entry['sha256']) + records[artifact] = data + require({'runtime-config.private.json', 'selected-state.private.sqlite'} <= records.keys()) + retained = strict_json(records['runtime-config.private.json'], MAX_JOB) + if retained.get('Id') != original['container_id'] or retained.get('Image') != original['image_id']: + continue + require(type(job['created_at']) in (int, float) and math.isfinite(job['created_at'])) + require(_timestamp(baseline['utc']) < job['created_at']) + require(baseline['source_sql_read_only'] is True and baseline['new_full_database_snapshot'] is False) + require(baseline['source']['device'] == observed['database_device'] and baseline['source']['inode'] == observed['database_inode']) + require(runtime_fingerprints(retained) == runtime_fingerprints(runtime)) + expected = {key: baseline['baseline'][key] for key in ('settings_sha256', 'stable_cloud_sha256', 'stable_inventory_sha256', 'saved_identity_sha256', 'saved_identity_field_count', 'stored_key_digest_matches', 'sync_enabled')} + # Recompute stable selected-state inputs from the actual retained + # SQLite, whose exact bytes were verified above. No live copy. + sqlite_path = BASELINE_ROOT / name / 'selected-state.private.sqlite' + projected = database_fingerprints(sqlite_path) + require(all(projected[key] == value for key, value in expected.items())) + require(read_private(fd, 'selected-state.private.sqlite', 16 * 1024 * 1024, root=True) == records['selected-state.private.sqlite']) + expected['schema_sha256'] = baseline['source']['schema_sha256'] + require(all(observed[key] == value for key, value in expected.items())) + declared = {} + require(set(baseline['declared_config']) == {'/app/config/servers.yaml', '/app/data/servers.json'}) + for index, (path, config) in enumerate(sorted(baseline['declared_config'].items())): + require(type(config.get('present')) is bool) + declared[path] = {'present': config['present']} + if config['present']: + artifact = 'config-' + str(index) + '.private' + require(artifact in records and config['backup']['sha256'] == sha256(records[artifact])) + declared[path]['sha256'] = config['backup']['sha256'] + require(fingerprint(declared) == observed['declared_config_sha256']) + require(time.monotonic() <= deadline) + matches.append(sha256(raw)) + require(len(matches) == 1) + return matches[0] diff --git a/src/cryptolabs_proxy/updates.py b/src/cryptolabs_proxy/updates.py index 8eb6dec..f72eda5 100644 --- a/src/cryptolabs_proxy/updates.py +++ b/src/cryptolabs_proxy/updates.py @@ -24,6 +24,7 @@ from .migration import DockerEngine, MigrationError, _mount_request, endpoint_configurations from . import release_selection as selection from . import controller_recovery +from . import update_recovery as recovery SERVICES = { @@ -62,6 +63,7 @@ def get_capabilities(): 'supported_selection_contracts': [1, 2], 'release_components': 1, 'release_idempotency': 1, 'proxy_selection_preview': 1, 'proxy_adoption_recovery': 1, + 'native_service_recovery': 1, 'proxy_replacement_policy': PROXY_REPLACEMENT_POLICY, 'controller_adoption_guard': dict(controller_recovery.CAPABILITY)} @@ -88,6 +90,131 @@ def _require_selected_updater(engine, source, contract=1, *, scoped=False): raise UpdateError('Selected helper requires verified live updater capabilities. '+PROXY_ADOPTION_MESSAGE) from None +def _require_terminated_helper(engine, job): + helper = 'cryptolabs-update-' + job['id'] + recovery.require(job['helper'] == helper) + try: + helper_state = engine.inspect(helper)['State'] + recovery.require(helper_state.get('Status') in ('exited', 'dead') and + all(helper_state.get(k) is False for k in ('Running', 'Paused', 'Restarting'))) + except MigrationError as error: + if not str(error).endswith('HTTP 404'): + recovery.hold() + + +# TRACEWEAVER: entrypoint=observe_original; req=REQ-FLEETUPDATE-001; trace=TRACE-FLEETUPDATE-001; ver=VER-FLEETUPDATE-001 +def observe_original(engine, job): + """Observe exact original, terminal helper and bounded private continuity.""" + try: + original = job['sources']['ipmi-monitor'] + current = engine.inspect('ipmi-monitor') + state = current['State'] + recovery.require(current['Id'] == original['container_id'] and current['Image'] == original['image_id']) + recovery.require(current['Name'] == '/ipmi-monitor' and state.get('Running') is True) + recovery.require(all(state.get(k) is False for k in ('Paused', 'Restarting', 'Dead'))) + recovery.require(state.get('Status') == 'running') + _require_terminated_helper(engine, job) + proof = engine.recovery_probe(original['container_id']) + recovery.require(type(proof) is dict and set(proof) == {'continuity', 'last_collection'}) + recovery.require(recovery._timestamp(proof['last_collection']) > recovery._timestamp(state['StartedAt'])) + continuity = dict(proof['continuity'], projection='ipmi-original-continuity-v1', + **recovery.runtime_fingerprints(current)) + recovery.validate_continuity(continuity) + # Inspect once more after the read-only probe; same-name replacement is + # never evidence for the original. No helper or service is stopped here. + final = engine.inspect('ipmi-monitor') + recovery.require(final['Id'] == current['Id'] and final['Image'] == current['Image'] and + all(final['State'].get(k) == state.get(k) for k in ('Running','Status','Paused','Restarting','Dead','StartedAt')) and + recovery.runtime_fingerprints(final) == recovery.runtime_fingerprints(current)) + _require_terminated_helper(engine, job) + return recovery.OriginalObservation(current['Id'], current['Image'], True, True, continuity) + except (KeyError, TypeError, OSError, MigrationError, ValueError): + recovery.hold() + + +def _needs_service_receipt(job): + return any(type(result) is dict and result.get('state') == 'recovery_required' + for result in job.get('results', {}).values()) + + +def _retains_predecessor(job): + result = job.get('results', {}).get('ipmi-monitor', {}) + verified = (job.get('action') == 'update' and result.get('success') is True and + result.get('state') in ('completed', 'unchanged')) + return 'recovery_predecessor' in job and not verified + + +def _require_recovery_capability(engine): + source = engine.inspect('cryptolabs-proxy') + caps = engine.updater_capabilities(source['Id']) + recovery.require(source.get('State', {}).get('Running') is True and type(caps) is dict and + type(caps.get('native_service_recovery')) is int and caps['native_service_recovery'] == 1) + final = engine.inspect('cryptolabs-proxy') + recovery.require(final['Id'] == source['Id'] and final['Image'] == source['Image'] and + final.get('State', {}).get('Running') is True) + + +def _validate_predecessor(fd, binding, engine, seen=None): + """Read-only bounded dependency validation; never replay an old worker.""" + seen = set() if seen is None else seen + recovery.require(type(binding) is dict and set(binding) == {'job_id', 'job_sha256', 'receipt_sha256'}) + old_id = binding['job_id'] + recovery.require(recovery._pattern(old_id, '[0-9a-f]{32}') and old_id not in seen and len(seen) < 16) + seen.add(old_id) + for key in ('job_sha256', 'receipt_sha256'): + recovery.require(recovery._pattern(binding[key], '[0-9a-f]{64}')) + raw = recovery.read_private(fd, old_id + '.json', recovery.MAX_JOB) + receipt_raw = recovery.read_private(fd, 'service-recovery-' + old_id + '.json') + recovery.require(recovery.sha256(raw) == binding['job_sha256'] and recovery.sha256(receipt_raw) == binding['receipt_sha256']) + old = recovery.validate_job(raw, old_id) + recovery.validate_original_restored(raw, receipt_raw, observe_original(engine, old)) + if _retains_predecessor(old): + _validate_predecessor(fd, old['recovery_predecessor'], engine, seen) + return binding + + +# TRACEWEAVER: entrypoint=service_recovery_dependency; req=REQ-FLEETUPDATE-001; trace=TRACE-FLEETUPDATE-001; ver=VER-FLEETUPDATE-001 +def service_recovery_dependency(directory, previous, engine, *, require_capability=True): + """Called under dispatch/selector exclusion before creating any new work.""" + if not _needs_service_receipt(previous) and not _retains_predecessor(previous): + return None + if require_capability: + _require_recovery_capability(engine) + with recovery.open_existing_directory(directory) as fd: + inherited = None + if _retains_predecessor(previous): + inherited = _validate_predecessor(fd, previous['recovery_predecessor'], engine) + if not _needs_service_receipt(previous): + return inherited + raw, active = recovery.load_bound_job(fd, previous['id']) + receipt_raw = recovery.read_private(fd, 'service-recovery-' + previous['id'] + '.json') + receipt = recovery.validate_original_restored(raw, receipt_raw, observe_original(engine, previous)) + recovery.require(receipt['active_sha256'] == recovery.sha256(active)) + return {'job_id': previous['id'], 'job_sha256': recovery.sha256(raw), + 'receipt_sha256': recovery.sha256(receipt_raw)} + + +def _check_worker_predecessor(directory, job, engine): + if 'recovery_predecessor' not in job: + return + _require_recovery_capability(engine) + with recovery.open_existing_directory(directory) as fd: + pointer = recovery.strict_json(recovery.read_private(fd, 'active.json')) + recovery.require(pointer == {'id': job['id']}) + _validate_predecessor(fd, job['recovery_predecessor'], engine) + + +def _worker_mutation_guard(directory, job, engine): + """Check once at the actual first effect; later service updates advance state.""" + started = False + def check(): + nonlocal started + if not started: + _check_worker_predecessor(directory, job, engine) + started = True + return check + + def _proxy_replacement_required(source, candidate, policy=None): return candidate['Id'] != source['Image'] or bool(policy and (source['Config'].get('Labels') or {}).get(WATCHTOWER_LABEL) != 'false') @@ -131,6 +258,17 @@ def updater_capabilities(self, source_id): except (OSError, subprocess.SubprocessError, ValueError): raise UpdateError('Live updater capabilities unavailable.') from None + def recovery_probe(self, original_id): + """Fixed read-only Python; never execute caller code or pass private data.""" + recovery.require(recovery._pattern(original_id, '[0-9a-f]{64}')) + try: + result = subprocess.run(['docker', '--host', 'unix:///var/run/docker.sock', 'exec', original_id, 'python3', '-c', recovery.FIXED_PROBE], + capture_output=True, timeout=15) + recovery.require(result.returncode == 0 and len(result.stdout) <= recovery.MAX_RECEIPT) + return recovery.strict_json(result.stdout) + except (OSError, subprocess.SubprocessError, ValueError): + recovery.hold() + def selection_bytes(self): return self._release_bytes('selection.json') @@ -389,64 +527,152 @@ def wait_ready(engine, name, image_id, timeout=150): time.sleep(1) -def replace_container(engine, name, image, job_id, *, timeout=150, expected_id=None, policy=None, expected_image=None): - """Swap without destroying the previous runtime; prove recovery on failure.""" +def phase_recorder(directory, job, service): + """Persist allowlisted intent before an Engine effect on new jobs only.""" + started = time.monotonic() + allowed = {'phase', 'operation', 'error_category', 'http_status', 'elapsed', 'deadline', + 'recovery_phase', 'recovery_outcome'} + def record(**event): + if set(event) - allowed: + raise UpdateError('invalid_phase_diagnostic') + event['elapsed'] = round(time.monotonic() - started, 3) + history = job.setdefault('diagnostics', {}).setdefault(service, []) + if len(history) >= 64: + raise UpdateError('phase_diagnostic_limit') + history.append(event) + write_job(directory, job) + return record + + +def replace_container(engine, name, image, job_id, *, timeout=150, expected_id=None, policy=None, expected_image=None, record_phase=None, before_mutation=None): + """Swap and restore only inspected owned identities, including lost replies.""" backup = f'{name}.rollback-{job_id}' - detached = [] - started_stop = renamed = created = False - source = None + started_stop = False + source = candidate = None + def record(**event): + if record_phase is not None: + record_phase(**event) + def mutate(operation, function, *args): + record(phase='intent', operation=operation, deadline=timeout) + value = function(*args) + record(phase='completed', operation=operation, deadline=timeout) + return value + def inspect_optional(identity): + try: + return engine.inspect(identity) + except MigrationError as error: + if str(error).endswith('HTTP 404'): + return None + raise try: source = engine.inspect(name) if expected_id and source['Id'] != expected_id: - raise UpdateError('Container changed during update preparation.') + raise UpdateError('runtime_changed') if not source.get('State', {}).get('Running'): - raise UpdateError('Service is stopped; use its installer to start it.') + raise UpdateError('service_stopped') candidate = engine.image(image) if expected_image and candidate['Id'] != expected_image: - raise UpdateError('Candidate changed during update preparation.') + raise UpdateError('candidate_changed') if policy and (image not in candidate.get('RepoDigests', []) or _platform(candidate) != policy['platform']): - raise UpdateError('Candidate digest or platform verification failed.') + raise UpdateError('artifact_identity_mismatch') if candidate['Id'] == source['Image'] and (not policy or (source['Config'].get('Labels') or {}).get(WATCHTOWER_LABEL) == 'false'): return {'success': True, 'state': 'unchanged', 'image_id': source['Image'], 'message': 'Already running the selected image.'} - if name == 'cryptolabs-proxy': return _proxy_hold() + if name == 'cryptolabs-proxy': + return _proxy_hold() + # A preexisting rollback name is not owned by this attempted switch. + if inspect_optional(backup) is not None: + raise UpdateError('rollback_name_occupied') request = build_update_request(source, engine.image(source['Image']), candidate) + request['Config']['Labels']['io.cryptolabs.update.job'] = job_id if policy: request['Config']['Labels'].update({WATCHTOWER_LABEL: 'false', 'io.cryptolabs.update.origin': policy['origin'], 'io.cryptolabs.update.reference': image}) if not image.startswith('sha256:'): request['Config']['Labels'].update({CHANNEL_LABEL: image, IMAGE_LABEL: candidate['Id']}) endpoints = request['NetworkingConfig']['EndpointsConfig'] + if before_mutation is not None: + try: + before_mutation() + except Exception: + record(phase='blocked', operation='update', error_category='internal_error') + return {'success': False, 'state': 'blocked', 'reason': 'native_service_recovery_required', + 'message': 'native_service_recovery_required'} + # Intent is durable before marking a possibly effective stop. + record(phase='intent', operation='stop', deadline=timeout) started_stop = True - engine.stop(name) + engine.stop(source['Id']) + record(phase='completed', operation='stop', deadline=timeout) for network in endpoints: - engine.disconnect(name, network) - detached.append(network) - engine.rename(name, backup) - renamed = True - engine.create(name, request) - created = True - engine.start(name) - if not wait_ready(engine, name, candidate['Id'], timeout): - raise UpdateError('Replacement failed application health or image verification.') + mutate('disconnect', engine.disconnect, source['Id'], network) + mutate('rename', engine.rename, source['Id'], backup) + mutate('create', engine.create, name, request) + created = engine.inspect(name) + if created['Image'] != candidate['Id'] or (created['Config'].get('Labels') or {}).get('io.cryptolabs.update.job') != job_id: + raise UpdateError('candidate_identity_mismatch') + mutate('start', engine.start, created['Id']) + if not wait_ready(engine, name, candidate['Id'], timeout) or engine.inspect(name)['Id'] != created['Id']: + raise UpdateError('health_failed') return {'success': True, 'state': 'completed', 'image_id': candidate['Id'], 'rollback_container': backup, 'message': 'Updated and health verified.'} except Exception as error: - message = str(error) if isinstance(error, UpdateError) else 'Container update failed.' + # Never export exception text (including exceptions of our public type + # raised by a transport adapter). Only a small fixed diagnostic survives. + category = 'engine_error' if isinstance(error, MigrationError) else 'internal_error' + diagnostic = {'phase': 'failed', 'operation': 'update', 'error_category': category} + status = re.search(r'HTTP ([1-5][0-9]{2})$', str(error)) if isinstance(error, MigrationError) else None + if status: + diagnostic['http_status'] = int(status.group(1)) + try: + record(**diagnostic) + except Exception: + return {'success': False, 'state': 'recovery_required' if started_stop else 'failed', + 'message': 'Update evidence unavailable; operator inspection required.'} if not started_stop: - return {'success': False, 'state': 'failed', 'message': message} + return {'success': False, 'state': 'failed', 'message': 'Container update failed.'} try: - if created: - engine.remove(name, ignore_missing=True) - if renamed: - engine.rename(backup, name) - for network in detached: - engine.connect(name, network, endpoints[network]) - engine.start(name) - if not wait_ready(engine, name, source['Image'], timeout): - raise UpdateError('Original service did not recover.') - return {'success': False, 'state': 'rolled_back', 'message': message + ' Original service restored and health verified.'} + # Replies may be lost after stop/disconnect/rename/create/start took + # effect. Re-observe names and full IDs before ANY restoration effect. + current = inspect_optional(name) + retained = inspect_optional(backup) + originals = [item for item in (current, retained) if item and item['Id'] == source['Id'] and item['Image'] == source['Image']] + if len(originals) != 1: + raise UpdateError('original_identity_unavailable') + original = originals[0] + if current and current['Id'] != source['Id']: + if (current['Image'] != candidate['Id'] or + (current['Config'].get('Labels') or {}).get('io.cryptolabs.update.job') != job_id): + raise UpdateError('unowned_candidate') + record(phase='intent', operation='remove', recovery_phase='remove_owned_candidate') + engine.remove(current['Id']) + record(phase='completed', operation='remove', recovery_phase='remove_owned_candidate') + elif current and current['Image'] != source['Image']: + raise UpdateError('original_identity_changed') + if retained and retained['Id'] == source['Id']: + record(phase='intent', operation='rename', recovery_phase='restore_original_name') + engine.rename(source['Id'], name) + record(phase='completed', operation='rename', recovery_phase='restore_original_name') + restored = engine.inspect(name) + if restored['Id'] != source['Id'] or restored['Image'] != source['Image']: + raise UpdateError('original_identity_changed') + for network, endpoint in endpoints.items(): + if network not in restored.get('NetworkSettings', {}).get('Networks', {}): + record(phase='intent', operation='connect', recovery_phase='restore_original_network') + engine.connect(source['Id'], network, endpoint) + record(phase='completed', operation='connect', recovery_phase='restore_original_network') + record(phase='intent', operation='start', recovery_phase='restore_original_running') + engine.start(source['Id']) + record(phase='completed', operation='start', recovery_phase='restore_original_running') + if not wait_ready(engine, name, source['Image'], timeout) or engine.inspect(name)['Id'] != source['Id']: + raise UpdateError('original_health_unavailable') + record(phase='completed', operation='recovery', recovery_outcome='original_restored') + return {'success': False, 'state': 'rolled_back', 'message': 'Container update failed. Original service restored and health verified.'} except Exception: - return {'success': False, 'state': 'recovery_required', 'message': message + ' Automatic recovery was not verified; operator recovery required.', 'rollback_container': backup if renamed else name} + try: + record(phase='failed', operation='recovery', recovery_outcome='recovery_required') + except Exception: + pass + return {'success': False, 'state': 'recovery_required', + 'message': 'Container update failed. Automatic recovery was not verified; operator recovery required.', 'rollback_container': backup} def _job_path(directory, job_id): @@ -667,9 +893,14 @@ def _submit_job_locked(service, branch, action, *, directory, engine, selection_ except controller_recovery.RecoveryError: raise UpdateError('controller_recovery_required') from None active = Path(directory) / 'active.json' + predecessor = None if active.exists(): previous = read_job(directory, json.loads(active.read_text())['id']) if previous.get('state') == 'recovery-required': raise UpdateBusy(previous) + try: + predecessor = service_recovery_dependency(directory, previous, engine) + except Exception: + raise UpdateBusy(previous) from None if previous.get('state') in ACTIVE_STATES: try: running = engine.inspect(previous['helper'])['State']['Running'] @@ -709,6 +940,8 @@ def _submit_job_locked(service, branch, action, *, directory, engine, selection_ job = {'id': job_id, 'helper': helper, 'service': service, 'branch': branch, 'action': action, 'state': 'queued', 'results': {}, 'created_at': time.time(), 'scope': 'proxy-updater', 'suite_complete': False, 'external_owners': []} + if predecessor is not None: + job['recovery_predecessor'] = predecessor if snapshot: job.update(selection_id=snapshot['selection_id'], selection_sha256=snapshot['selection_sha256'], manifest_sha256=snapshot['manifest_sha256'], selection_snapshot=snapshot, targets=targets, @@ -754,6 +987,7 @@ def pull_target(target): def _run_selected_job(directory, job, engine): """Stage all exact artifacts and clone requests before stopping any service.""" try: + _check_worker_predecessor(directory, job, engine) stored = job['selection_snapshot'] snapshot = selection.validate(stored['selection_bytes'].encode(), stored['manifest_bytes'].encode(), selection_id=job['selection_id'], selection_sha256=job['selection_sha256']) @@ -797,10 +1031,12 @@ def _run_selected_job(directory, job, engine): # complete plan before the first stop, rather than allowing partial work. if selection.sha256(engine.selection_bytes()) != job['selection_sha256'] or selection.sha256(engine.manifest_bytes(job['manifest_sha256'])) != job['manifest_sha256'] or _selection_plan(snapshot, engine, job.get('components'))[1] != sources: raise UpdateError('selection_or_runtime_changed') + _check_worker_predecessor(directory, job, engine) for name, check in checked.items(): _write_json(_check_path(directory, name, job['branch']), check) except Exception as error: - reason = str(error) if isinstance(error, (UpdateError, selection.SelectionError)) else 'preflight_unavailable' + reason = ('native_service_recovery_required' if 'recovery_predecessor' in job and not isinstance(error, (UpdateError, selection.SelectionError)) + else str(error) if isinstance(error, (UpdateError, selection.SelectionError)) else 'preflight_unavailable') job.update(state='blocked', success=False, reason=reason, finished_at=time.time(), message=PROXY_ADOPTION_MESSAGE if reason=='proxy_host_adoption_required' else reason, results={name: {'success': False, 'state': 'blocked', 'reason': reason, @@ -814,6 +1050,7 @@ def _run_selected_job(directory, job, engine): job['results'][name] = {'success': False, 'state': 'blocked', 'owner': target['owner'], 'desired': target.get('reference', target.get('sha256')), 'observed': None, 'origin': target['origin'], 'reason': 'external_owner_required'} + before_mutation = _worker_mutation_guard(directory, job, engine) for name in names: target = targets[name] if job['action'] == 'pull': @@ -821,10 +1058,16 @@ def _run_selected_job(directory, job, engine): 'update_available': staged[name] != sources[name]['image_id']} else: result = replace_container(engine, name, target['reference'], job['id'], policy=target, - expected_id=sources[name]['container_id'], expected_image=staged[name]) + expected_id=sources[name]['container_id'], expected_image=staged[name], + record_phase=phase_recorder(directory, job, name), + before_mutation=before_mutation) result.update(owner=target['owner'], origin=target['origin'], desired=target['reference'], - observed=result.get('image_id'), reason=None if result['success'] else 'replacement_failed') + observed=result.get('image_id'), reason=None if result['success'] else result.get('reason', 'replacement_failed')) job['results'][name] = result + if result.get('state') == 'blocked' and result.get('reason') == 'native_service_recovery_required': + job.update(state='blocked', success=False, reason='native_service_recovery_required', finished_at=time.time()) + write_job(directory, job) + return job write_job(directory, job) if not result['success']: break @@ -848,6 +1091,13 @@ def run_job(directory, job_id, *, engine=None): finished_at=time.time(), results={}) write_job(directory, job) return job + try: + _check_worker_predecessor(directory, job, engine) + except Exception: + job.update(state='blocked', success=False, reason='native_service_recovery_required', + finished_at=time.time(), results={}) + write_job(directory, job) + return job job.update(state='running', results={}) write_job(directory, job) if job['action'] == 'pull': @@ -892,6 +1142,7 @@ def run_job(directory, job_id, *, engine=None): message=PROXY_ADOPTION_MESSAGE if reason=='proxy_host_adoption_required' else reason, finished_at=time.time()) write_job(directory,job); return job + before_mutation = _worker_mutation_guard(directory, job, engine) for name in names: if job['action'] == 'update' and SERVICES[name].get('update_supported') is False: continue @@ -915,7 +1166,9 @@ def run_job(directory, job_id, *, engine=None): result = {'success': True, 'state': 'checked', 'image_id': image['Id'], 'update_available': image['Id'] != source['Image'], 'message': 'Image checked; running service unchanged.'} else: - result = replace_container(engine, name, target, job_id, expected_id=source['Id']) + result = replace_container(engine, name, target, job_id, expected_id=source['Id'], + record_phase=phase_recorder(directory, job, name), + before_mutation=before_mutation) except Exception as error: missing = isinstance(error, MigrationError) and str(error).endswith('HTTP 404') if missing and job['service'] == 'all': @@ -923,6 +1176,10 @@ def run_job(directory, job_id, *, engine=None): else: result = {'success': False, 'state': 'failed', 'message': 'Could not inspect or prepare service; no update started.'} job['results'][name] = result + if result.get('state') == 'blocked' and result.get('reason') == 'native_service_recovery_required': + job.update(state='blocked', success=False, reason='native_service_recovery_required', finished_at=time.time()) + write_job(directory, job) + return job write_job(directory, job) if result.get('state') == 'recovery_required': break diff --git a/tests/test_proxy_customer_route_repairs.py b/tests/test_proxy_customer_route_repairs.py index 2d8eb06..58a2bab 100644 --- a/tests/test_proxy_customer_route_repairs.py +++ b/tests/test_proxy_customer_route_repairs.py @@ -171,9 +171,11 @@ def test_additive_capabilities_getter_matches_cli_contract(capsys): capabilities=updates.get_capabilities() assert capabilities=={'selection_contract':1,'supported_selection_contracts':[1,2], 'release_components':1,'release_idempotency':1,'proxy_selection_preview':1,'proxy_adoption_recovery':1, + 'native_service_recovery':1, 'proxy_replacement_policy':'host-adoption-required-v1', 'controller_adoption_guard':{'schema_version':1,'api':'AdoptionEngine.controller_guard', 'durable_updater_exclusion':True}} + assert type(capabilities['native_service_recovery']) is int assert type(capabilities['controller_adoption_guard']['schema_version']) is int assert capabilities['controller_adoption_guard']['durable_updater_exclusion'] is True assert updates.main(['capabilities'])==0 diff --git a/tests/test_update_recovery.py b/tests/test_update_recovery.py new file mode 100644 index 0000000..a1143f2 --- /dev/null +++ b/tests/test_update_recovery.py @@ -0,0 +1,308 @@ +# TRACEWEAVER: file-role=verification; req=REQ-FLEETUPDATE-001; trace=TRACE-FLEETUPDATE-001; ver=VER-FLEETUPDATE-001 +# TRACEWEAVER: verifies=VER-FLEETUPDATE-001; req=REQ-FLEETUPDATE-001; trace=TRACE-FLEETUPDATE-001 +"""Approved shared receipt contract; synthetic private filesystem only. + +Initial RED is missing-module evidence, not proof that the future rejection, +durability or host bootstrap behavior has been exercised. +""" +import importlib +import json +import os +from copy import deepcopy + +import pytest + + +def receipt(): + """Exact versioned shape; deliberately synthetic IDs and fingerprints.""" + continuity = { + 'projection': 'ipmi-original-continuity-v1', + 'database_device': 1, 'database_inode': 2, + 'saved_identity_field_count': 5, + 'stored_key_digest_matches': True, 'sync_enabled': False, + } + for key in ('runtime_config', 'host_config', 'mounts', 'network', 'schema', + 'settings', 'stable_cloud', 'stable_inventory', 'saved_identity', + 'declared_config'): + continuity[key + '_sha256'] = 'b' * 64 + return { + 'schema_version': 1, 'kind': 'native-service-original-restored', + 'service': 'ipmi-monitor', 'job_id': 'a' * 32, + 'job_sha256': 'c' * 64, 'active_sha256': 'd' * 64, + 'selection_id': 'synthetic-scoped-selection', + 'selection_sha256': 'e' * 64, 'manifest_sha256': 'f' * 64, + 'original_container_id': '1' * 64, 'original_image_id': 'sha256:' + '2' * 64, + 'helper_name': 'cryptolabs-update-' + 'a' * 32, + 'baseline_receipt_sha256': '3' * 64, + 'continuity': continuity, 'outcome': 'original_restored', + 'verified_at': '2026-11-02T08:00:00Z', + } + + +def test_shared_receipt_schema_and_private_atomic_publication(tmp_path, monkeypatch): + recovery = importlib.import_module('cryptolabs_proxy.update_recovery') + for name in ('validate_receipt', 'load_bound_job', 'validate_original_restored', + 'publish_original_restored'): + assert callable(getattr(recovery, name)) + value = receipt() + recovery.validate_receipt(value) + for change in ('extra-key', 'bool-version', 'foreign-helper', 'false-continuity'): + bad = deepcopy(value) + if change == 'extra-key': bad['force'] = True + if change == 'bool-version': bad['schema_version'] = True + if change == 'foreign-helper': bad['helper_name'] = 'foreign-helper' + if change == 'false-continuity': bad['continuity']['stored_key_digest_matches'] = False + with pytest.raises(ValueError): + recovery.validate_receipt(bad) + + tmp_path.chmod(0o700) + # The publisher must not create, rewrite or remove historical job/pointer files. + job = tmp_path / ('a' * 32 + '.json') + active = tmp_path / 'active.json' + job.write_bytes(b'{"synthetic":"historical failed job"}\n') + active.write_bytes(b'{"id":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"}\n') + job.chmod(0o600); active.chmod(0o600) + before = (job.read_bytes(), active.read_bytes()) + final = tmp_path / ('service-recovery-' + value['job_id'] + '.json') + descriptor = os.open(tmp_path, os.O_RDONLY | os.O_DIRECTORY) + try: + original_fsync = os.fsync + def interrupted_fsync(fd): + raise OSError('synthetic disk interruption before durable file') + with monkeypatch.context() as patch: + patch.setattr(os, 'fsync', interrupted_fsync) + with pytest.raises(OSError): + recovery.publish_original_restored(descriptor, value) + assert not final.exists() + assert (job.read_bytes(), active.read_bytes()) == before + calls = [] + def record_fsync(fd): + calls.append(os.fstat(fd).st_mode) + return original_fsync(fd) + with monkeypatch.context() as patch: + patch.setattr(os, 'fsync', record_fsync) + recovery.publish_original_restored(descriptor, value) + stored = final.read_bytes() + assert json.loads(stored) == value + assert final.stat().st_mode & 0o777 == 0o600 + assert final.stat().st_nlink == 1 + assert len(calls) >= 2 # file and parent directory durability + recovery.publish_original_restored(descriptor, value) + assert final.read_bytes() == stored + assert (job.read_bytes(), active.read_bytes()) == before + finally: + os.close(descriptor) + + +def historical_snapshot(snapshot): + from cryptolabs_proxy import release_selection + manifest=deepcopy(snapshot['manifest']) + manifest['components']['ipmi-monitor']={'release':'synthetic-legacy','owner':'proxy-updater', + 'artifacts':{'linux/amd64':{'kind':'oci','reference':'ghcr.io/cryptolabsza/ipmi-monitor@sha256:'+'c'*64}}} + manifest_raw=json.dumps(manifest).encode() + selected=deepcopy(snapshot['selection']) + selected['components']['ipmi-monitor']={'enabled':True,'origin':'baseline'} + selected['manifest_sha256']=release_selection.sha256(manifest_raw) + return release_selection.validate(json.dumps(selected).encode(),manifest_raw) + + +@pytest.fixture +def bound(tmp_path): + from tests.test_release_selection import selection_files + from cryptolabs_proxy import release_selection + directory, digest = selection_files(tmp_path) + snapshot = historical_snapshot(release_selection.load(directory)) + digest = snapshot['selection_sha256'] + value = receipt() + value.update(selection_id=snapshot['selection_id'], selection_sha256=digest, + manifest_sha256=snapshot['manifest_sha256']) + job = {'id': value['job_id'], 'helper': value['helper_name'], 'service': 'ipmi-monitor', + 'branch': 'main', 'action': 'update', 'state': 'failed', 'success': False, + 'results': {'ipmi-monitor': {'state': 'recovery_required', 'success': False}}, + 'sources': {'ipmi-monitor': {'container_id': value['original_container_id'], + 'image_id': value['original_image_id']}}, + 'selection_id': value['selection_id'], 'selection_sha256': value['selection_sha256'], + 'manifest_sha256': value['manifest_sha256'], 'selection_snapshot': snapshot, + 'targets': release_selection.targets(snapshot,'linux/amd64')} + raw = json.dumps(job).encode() + value['job_sha256'] = __import__('hashlib').sha256(raw).hexdigest() + return job, raw, value + + +def test_shared_job_binding_and_observation_are_not_clearance_tokens(bound): + from cryptolabs_proxy import update_recovery as recovery + job, raw, value = bound + observation = recovery.OriginalObservation( + container_id=value['original_container_id'], image_id=value['original_image_id'], + helper_terminated=True, healthy=True, continuity=value['continuity']) + encoded = json.dumps(value).encode() + assert recovery.validate_original_restored(raw, encoded, observation) == value + for changes in ({'healthy': False}, {'helper_terminated': False}, + {'container_id': '9' * 64}, {'continuity': {}}): + from dataclasses import replace + with pytest.raises(ValueError): + recovery.validate_original_restored(raw, encoded, replace(observation, **changes)) + with pytest.raises(ValueError): + recovery.validate_original_restored(raw + b' ', encoded, observation) + with pytest.raises(ValueError): + recovery.validate_receipt(encoded[:-1] + b',"job_id":"' + b'a'*32 + b'"}') + for bad in (dict(value, verified_at='tomorrow'), dict(value, job_sha256=True)): + with pytest.raises(ValueError): recovery.validate_receipt(bad) + + +def test_existing_receipt_conflict_and_untrusted_records_fail_closed(tmp_path, bound): + from cryptolabs_proxy import update_recovery as recovery + job, raw, value = bound + tmp_path.chmod(0o700) + active = json.dumps({'id': job['id']}).encode() + for name, data in ((job['id']+'.json', raw), ('active.json', active)): + (tmp_path/name).write_bytes(data); (tmp_path/name).chmod(0o600) + fd = os.open(tmp_path, os.O_RDONLY | os.O_DIRECTORY) + try: + assert recovery.load_bound_job(fd, job['id']) == (raw, active) + recovery.publish_original_restored(fd, value) + with pytest.raises(ValueError): + recovery.publish_original_restored(fd, dict(value, baseline_receipt_sha256='4'*64)) + (tmp_path/(job['id']+'.json')).chmod(0o644) + with pytest.raises(ValueError): recovery.load_bound_job(fd, job['id']) + finally: os.close(fd) + + +def test_runtime_fingerprint_retains_duplicate_mounts_and_ignores_only_volatile_network(): + from cryptolabs_proxy import update_recovery as recovery + runtime = {'Config': {'Env': ['PRIVATE=synthetic']}, 'HostConfig': {'NetworkMode': 'private'}, + 'Mounts': [{'Destination': '/data', 'Source': '/synthetic', 'RW': True}], + 'NetworkSettings': {'Networks': {'private': {'NetworkID': 'x', 'Aliases': ['b','a'], 'EndpointID': 'old', 'IPAddress':'192.0.2.1'}}}} + original = recovery.runtime_fingerprints(runtime) + changed = deepcopy(runtime) + changed['NetworkSettings']['Networks']['private'].update(EndpointID='new', IPAddress='192.0.2.2', Aliases=['a','b']) + assert recovery.runtime_fingerprints(changed) == original + changed['Mounts'] *= 2 + assert recovery.runtime_fingerprints(changed)['mounts_sha256'] != original['mounts_sha256'] + changed = deepcopy(runtime); changed['Config']['Env'].append('NEW=drift') + assert recovery.runtime_fingerprints(changed)['runtime_config_sha256'] != original['runtime_config_sha256'] + + +def make_database(path): + import sqlite3 + connection = sqlite3.connect(path) + connection.executescript('CREATE TABLE system_settings (key TEXT PRIMARY KEY,value TEXT);' + 'CREATE TABLE cloud_sync (id INTEGER PRIMARY KEY,license_key TEXT,sync_enabled INTEGER,last_sync TEXT);' + 'CREATE TABLE server (id INTEGER PRIMARY KEY,name TEXT,status TEXT,updated_at TEXT);') + digest = __import__('hashlib').sha256(b'synthetic-private-key').hexdigest() + for key, value in zip(('ai_link_email','ai_link_user_id','ai_link_customer_id','ai_link_linked_at','ai_link_key_digest'), + ('synthetic@example.test','user','customer','2020-01-01',digest)): + connection.execute('INSERT INTO system_settings VALUES (?,?)',(key,value)) + connection.execute('INSERT INTO cloud_sync VALUES (1,?,0,?)',('synthetic-private-key','old')) + connection.execute('INSERT INTO server VALUES (1,?,?,?)',('synthetic-server','up','old')) + connection.commit(); connection.close() + + +def test_actual_sqlite_projection_ignores_volatile_fields_but_preserves_identity(tmp_path): + from cryptolabs_proxy import update_recovery as recovery + import sqlite3 + db = tmp_path/'selected.sqlite'; make_database(db) + before = recovery.database_fingerprints(db) + connection=sqlite3.connect(db) + connection.execute("UPDATE cloud_sync SET last_sync='new'") + connection.execute("UPDATE server SET status='down',updated_at='new'") + connection.commit() + assert recovery.database_fingerprints(db) == before + connection.execute("UPDATE server SET name='drift'");connection.commit() + assert recovery.database_fingerprints(db)['stable_inventory_sha256'] != before['stable_inventory_sha256'] + connection.execute("UPDATE cloud_sync SET license_key='changed'");connection.commit();connection.close() + with pytest.raises(ValueError): recovery.database_fingerprints(db) + + +@pytest.fixture +def retained_bundle(tmp_path, monkeypatch): + from cryptolabs_proxy import update_recovery as recovery + from datetime import datetime,timezone + live=tmp_path/'live.sqlite';make_database(live) + root=tmp_path/'baselines';root.mkdir(mode=0o700) + operation=root/'synthetic-pre-job';operation.mkdir(mode=0o700) + runtime={'Id':'1'*64,'Image':'sha256:'+'2'*64,'Config':{'Env':['PRIVATE=synthetic']}, + 'HostConfig':{'NetworkMode':'private'},'Mounts':[{'Destination':'/app/data','Source':'/synthetic','RW':True}], + 'NetworkSettings':{'Networks':{'private':{'NetworkID':'3'*64,'Aliases':['ipmi-monitor']}}}} + state=recovery.database_fingerprints(live) + continuity=dict(state,**recovery.runtime_fingerprints(runtime),projection='ipmi-original-continuity-v1', + database_device=live.stat().st_dev,database_inode=live.stat().st_ino, + declared_config_sha256=recovery.fingerprint({'/app/config/servers.yaml':{'present':True,'sha256':recovery.sha256(b'synthetic')}, + '/app/data/servers.json':{'present':False}})) + artifacts=[] + for name,raw in (('runtime-config.private.json',json.dumps(runtime).encode()), + ('selected-state.private.sqlite',live.read_bytes()),('config-0.private',b'synthetic')): + path=operation/name;path.write_bytes(raw);path.chmod(0o600) + artifacts.append({'path':str(path),'bytes':len(raw),'sha256':recovery.sha256(raw)}) + value={'status':'verified-bounded-selected-state-protection','utc':'2020-01-01T00:00:00Z', + 'source_sql_read_only':True,'new_full_database_snapshot':False, + 'source':{'device':live.stat().st_dev,'inode':live.stat().st_ino,'schema_sha256':state['schema_sha256']}, + 'baseline':state,'artifacts':artifacts,'declared_config':{ + '/app/config/servers.yaml':{'present':True,'backup':{'sha256':recovery.sha256(b'synthetic')}}, + '/app/data/servers.json':{'present':False}}} + raw=json.dumps(value).encode();(operation/'receipt.json').write_bytes(raw);(operation/'receipt.json').chmod(0o600) + job={'sources':{'ipmi-monitor':{'container_id':runtime['Id'],'image_id':runtime['Image']}}, + 'created_at':datetime(2020,1,2,tzinfo=timezone.utc).timestamp()} + monkeypatch.setattr(recovery,'BASELINE_ROOT',root) + private=recovery._private + monkeypatch.setattr(recovery,'_private',lambda info,**kw:private(info,directory=kw.get('directory',False))) + opened=recovery.open_existing_directory + monkeypatch.setattr(recovery,'open_existing_directory',lambda path,**kw:opened(path)) + return job,runtime,continuity,operation,raw + + +def test_retained_bundle_uses_actual_private_artifacts_not_new_baseline(retained_bundle): + from cryptolabs_proxy import update_recovery as recovery + job,runtime,continuity,operation,raw=retained_bundle + before={p.name:p.read_bytes() for p in operation.iterdir()} + assert recovery.discover_baseline(job,runtime,continuity)==recovery.sha256(raw) + assert {p.name:p.read_bytes() for p in operation.iterdir()}==before + (operation/'config-0.private').write_bytes(b'corrupted') + with pytest.raises(ValueError):recovery.discover_baseline(job,runtime,continuity) + + +@pytest.mark.parametrize('drift',['runtime','database','identity','multiple','post-job']) +def test_retained_bundle_rejects_mismatch_ambiguity_and_post_job_inputs(retained_bundle,drift): + from cryptolabs_proxy import update_recovery as recovery + job,runtime,continuity,operation,raw=retained_bundle + if drift=='runtime':runtime['Config']['Env'].append('DRIFT=1') + if drift=='database':continuity['database_inode']+=1 + if drift=='identity':continuity['saved_identity_sha256']='8'*64 + if drift=='multiple': + import shutil + shutil.copytree(operation,operation.parent/'another-pre-job') + if drift=='post-job':job['created_at']=1 + with pytest.raises(ValueError):recovery.discover_baseline(job,runtime,continuity) + + +def test_directory_fsync_failure_never_acknowledges_and_retry_durably_rechecks(tmp_path,monkeypatch): + from cryptolabs_proxy import update_recovery as recovery + tmp_path.chmod(0o700) + fd=os.open(tmp_path,os.O_RDONLY|os.O_DIRECTORY) + fsync=os.fsync;calls=[] + def fail_parent(descriptor): + calls.append(descriptor) + if descriptor==fd:raise OSError('synthetic directory fsync failure') + return fsync(descriptor) + try: + with monkeypatch.context() as patch: + patch.setattr(os,'fsync',fail_parent) + with pytest.raises(OSError):recovery.publish_original_restored(fd,receipt()) + saved=(tmp_path/('service-recovery-'+'a'*32+'.json')).read_bytes() + calls.clear() + with monkeypatch.context() as patch: + patch.setattr(os,'fsync',lambda descriptor:(calls.append(descriptor),fsync(descriptor))[1]) + recovery.publish_original_restored(fd,receipt()) + assert fd in calls + assert (tmp_path/('service-recovery-'+'a'*32+'.json')).read_bytes()==saved + finally:os.close(fd) + + +@pytest.mark.parametrize('wrong',['absent','platform','reference']) +def test_historical_target_is_bound_to_embedded_selection(bound,wrong): + from cryptolabs_proxy import update_recovery as recovery + job,raw,value=bound + if wrong=='absent':job['targets'].pop('ipmi-monitor') + if wrong=='platform':job['targets']['ipmi-monitor']['platform']='linux/unknown' + if wrong=='reference':job['targets']['ipmi-monitor']['reference']='ghcr.io/cryptolabsza/ipmi-monitor@sha256:'+'9'*64 + with pytest.raises(ValueError):recovery.validate_job(json.dumps(job).encode(),job['id']) diff --git a/tests/test_updates.py b/tests/test_updates.py index 233c329..ac868e5 100644 --- a/tests/test_updates.py +++ b/tests/test_updates.py @@ -88,7 +88,11 @@ def __init__(self, failure=None): self.calls = [] self.failure = failure + def resolve(self, name): + return next((key for key,value in self.containers.items() if value.get('Id')==name),name) + def inspect(self, name): + name=self.resolve(name) if self.failure == 'inspect': raise RuntimeError('private detail') if name not in self.containers: @@ -99,16 +103,22 @@ def image(self, image): return {'Id': 'sha256:old' if image == 'sha256:old' else 'sha256:new', 'Config': {}} def stop(self, name, **kwargs): + name=self.resolve(name) self.calls.append(('stop', name)) self.containers[name]['State']['Running'] = False def disconnect(self, name, network): + name=self.resolve(name) self.calls.append(('disconnect', name, network)) + self.containers[name]['NetworkSettings']['Networks'].pop(network,None) def connect(self, name, network, endpoint): + name=self.resolve(name) self.calls.append(('connect', name, network)) + self.containers[name]['NetworkSettings']['Networks'][network]=deepcopy(endpoint) def rename(self, name, new): + name=self.resolve(name) self.calls.append(('rename', name, new)) self.containers[new] = self.containers.pop(name) @@ -117,10 +127,13 @@ def create(self, name, request): if self.failure == 'create': raise RuntimeError('private detail') self.containers[name] = deepcopy(source()) + self.containers[name]['Id'] = 'b'*64 + self.containers[name]['NetworkSettings']['Networks']=deepcopy(request.get('NetworkingConfig',{}).get('EndpointsConfig',{})) self.containers[name]['Config'] = request['Config'] self.containers[name]['Image'] = request['Config']['Image'] def start(self, name): + name=self.resolve(name) self.calls.append(('start', name)) current = self.containers[name] current['State']['Running'] = True @@ -128,6 +141,7 @@ def start(self, name): current['State']['Health']['Status'] = 'unhealthy' def remove(self, name, **kwargs): + name=self.resolve(name) self.calls.append(('remove', name)) self.containers.pop(name, None) @@ -252,6 +266,35 @@ def test_duplicate_submission_does_not_launch_second_helper(monkeypatch, tmp_pat assert not engine.calls +@pytest.mark.parametrize('action', ['pull', 'update']) +def test_terminal_service_recovery_hold_preserves_job_and_active_pointer(tmp_path, action): + """A failed outcome is not permission to bypass unresolved service recovery.""" + engine = Engine() + engine.current['Mounts'].append({ + 'Type': 'volume', 'Name': 'fleet-data', 'Destination': str(tmp_path.parent), 'RW': True, + }) + engine.containers['cryptolabs-proxy'] = engine.current + previous = { + 'id': 'a' * 32, 'state': 'failed', 'success': False, + 'helper': 'cryptolabs-update-' + 'a' * 32, + 'service': 'ipmi-monitor', 'action': 'update', 'branch': 'main', + 'results': {'ipmi-monitor': {'state': 'recovery_required', 'success': False, + 'reason': 'replacement_failed'}}, + } + updates.write_job(tmp_path, previous) + active = tmp_path / 'active.json' + active.write_text(json.dumps({'id': previous['id']})) + job_path = tmp_path / (previous['id'] + '.json') + before = {path.name: path.read_bytes() for path in (active, job_path)} + + with pytest.raises(updates.UpdateBusy): + updates.submit_job('ipmi-monitor', 'main', action, directory=tmp_path, engine=engine) + + assert not engine.calls + assert {path.name: path.read_bytes() for path in (active, job_path)} == before + assert set(tmp_path.glob('[0-9a-f]' * 32 + '.json')) == {job_path} + + def test_missing_helper_is_recorded_as_interrupted_not_left_active(tmp_path): job = {'id': 'a' * 32, 'state': 'running', 'helper': 'missing-helper'} updates.write_job(tmp_path, job) @@ -497,3 +540,268 @@ def create(name,request): monkeypatch.setattr(updates,'pull_target',lambda _:True) engine.selection=(directory/'selection.json').read_bytes() assert updates.run_job(tmp_path/'jobs',result['id'],engine=engine)['state']=='completed' + + +@pytest.fixture +def recovered_predecessor(tmp_path, monkeypatch): + from tests.test_release_selection import selection_files, SelectedEngine + from tests.test_update_recovery import receipt, historical_snapshot + from cryptolabs_proxy import release_selection + directory,digest=selection_files(tmp_path) + snapshot=historical_snapshot(release_selection.load(directory)) + engine=SelectedEngine(tmp_path) + engine.updater_capabilities=lambda _:updates.get_capabilities() + jobs=tmp_path/'jobs' + prior=receipt() + prior.update(selection_id=snapshot['selection_id'],selection_sha256=snapshot['selection_sha256'],manifest_sha256=snapshot['manifest_sha256']) + job={'id':prior['job_id'],'helper':prior['helper_name'],'service':'ipmi-monitor','branch':'main','action':'update', + 'state':'failed','success':False,'results':{'ipmi-monitor':{'state':'recovery_required','success':False}}, + 'selection_id':prior['selection_id'],'selection_sha256':prior['selection_sha256'],'manifest_sha256':prior['manifest_sha256'], + 'selection_snapshot':snapshot,'sources':{'ipmi-monitor':{'container_id':prior['original_container_id'],'image_id':prior['original_image_id']}}, + 'targets':release_selection.targets(snapshot,'linux/amd64')} + updates.write_job(jobs,job);updates._write_json(jobs/'active.json',{'id':job['id']}) + from cryptolabs_proxy import update_recovery as recovery + for name in ('dispatch.lock','operation.lock'):(jobs/name).touch(mode=0o600) + prior.update(job_sha256=recovery.sha256((jobs/(job['id']+'.json')).read_bytes()),active_sha256=recovery.sha256((jobs/'active.json').read_bytes())) + observation=recovery.OriginalObservation(prior['original_container_id'],prior['original_image_id'],True,True,prior['continuity']) + monkeypatch.setattr(updates,'observe_original',lambda *a:observation,raising=False) + fd=__import__('os').open(jobs,__import__('os').O_RDONLY|__import__('os').O_DIRECTORY) + try:recovery.publish_original_restored(fd,prior) + finally:__import__('os').close(fd) + return jobs,directory,digest,engine,job,prior,observation + + +def submit_successor(fixture, operation='12345678-1234-4234-8234-123456789abc'): + jobs,directory,digest,engine,*_=fixture + return updates.submit_job('prometheus','main','pull',directory=jobs,engine=engine, + selection_id='synthetic-selection',selection_sha256=digest,selection_directory=directory, + components=['prometheus'],operation_id=operation,request_fingerprint='a'*64) + + +def test_receipt_bound_dispatch_rechecks_worker_and_preserves_dependency_after_block(recovered_predecessor,monkeypatch): + jobs,directory,digest,engine,old,receipt,observation=recovered_predecessor + oldraw=(jobs/(old['id']+'.json')).read_bytes() + successor=submit_successor(recovered_predecessor) + assert successor['recovery_predecessor']=={'job_id':old['id'],'job_sha256':receipt['job_sha256'], + 'receipt_sha256':__import__('hashlib').sha256((jobs/('service-recovery-'+old['id']+'.json')).read_bytes()).hexdigest()} + # Source drift after dispatch blocks before pulls/lifecycle, retaining old raw evidence. + from dataclasses import replace + monkeypatch.setattr(updates,'observe_original',lambda *a:replace(observation,healthy=False)) + monkeypatch.setattr(updates,'pull_target',lambda *a:pytest.fail('pull before recovery revalidation')) + result=updates.run_job(jobs,successor['id'],engine=engine) + assert result['state']=='blocked' and result['reason']=='native_service_recovery_required' + calls=list(engine.calls) + with pytest.raises(updates.UpdateBusy): + submit_successor(recovered_predecessor,'22345678-1234-4234-8234-123456789abc') + assert engine.calls==calls + assert (jobs/(old['id']+'.json')).read_bytes()==oldraw + # Exact operation replay is read only even while drift prevents a new launch. + assert submit_successor(recovered_predecessor)['id']==successor['id'] + + +@pytest.mark.parametrize('failure',['missing','wrong-job','capability']) +def test_recovery_receipt_or_capability_failure_blocks_before_dispatch_writes(recovered_predecessor,failure): + jobs,directory,digest,engine,old,receipt,observation=recovered_predecessor + path=jobs/('service-recovery-'+old['id']+'.json') + if failure=='missing':path.unlink() + elif failure=='wrong-job': + value=json.loads(path.read_bytes());value['job_sha256']='9'*64;path.write_text(json.dumps(value)) + else:engine.updater_capabilities=lambda _:dict(updates.get_capabilities(),native_service_recovery=0) + before={p.name:p.read_bytes() for p in jobs.iterdir() if p.is_file()} + with pytest.raises(updates.UpdateBusy):submit_successor(recovered_predecessor) + assert {p.name:p.read_bytes() for p in jobs.iterdir() if p.is_file()}==before + assert not engine.calls + + +def test_fixed_original_observer_requires_terminal_helper_and_fresh_exact_identity(monkeypatch): + from cryptolabs_proxy import update_recovery as recovery + engine=Engine();runtime=deepcopy(source()) + runtime.update(Id='1'*64,Image='sha256:'+'2'*64,Name='/ipmi-monitor') + runtime['State'].update(Status='running',Paused=False,Restarting=False,Dead=False,StartedAt='2026-11-02T00:00:00Z') + runtime['NetworkSettings']['Networks']['cryptolabs']['NetworkID']='3'*64 + engine.containers['ipmi-monitor']=runtime + proof={'continuity':{k:v for k,v in __import__('tests.test_update_recovery',fromlist=['receipt']).receipt()['continuity'].items() + if k not in ('projection','runtime_config_sha256','host_config_sha256','mounts_sha256','network_sha256')}, + 'last_collection':'2026-11-02T00:00:01Z'} + engine.recovery_probe=lambda _:proof + job={'id':'a'*32,'helper':'cryptolabs-update-'+'a'*32,'sources':{'ipmi-monitor':{'container_id':runtime['Id'],'image_id':runtime['Image']}}} + assert updates.observe_original(engine,job).healthy is True + engine.containers[job['helper']]={'State':{'Running':False,'Paused':False,'Restarting':False,'Status':'created'}} + with pytest.raises(ValueError):updates.observe_original(engine,job) + engine.containers[job['helper']]['State']['Status']='exited' + assert updates.observe_original(engine,job).helper_terminated is True + engine.containers['ipmi-monitor']['Id']='8'*64 + with pytest.raises(ValueError):updates.observe_original(engine,job) + + +@pytest.mark.parametrize('lost',['stop','rename','create','start']) +def test_new_phase_diagnostics_survive_lost_engine_reply_without_secrets(tmp_path,lost): + engine=Engine() + job={'id':'d'*32,'service':'prometheus','state':'running','results':{}} + updates.write_job(tmp_path,job) + events=[] + original=getattr(engine,lost) + once=[] + def lost_reply(*args,**kwargs): + original(*args,**kwargs) + if not once: + once.append(True) + raise updates.UpdateError('SECRET_SENTINEL-do-not-export') + setattr(engine,lost,lost_reply) + recorder=updates.phase_recorder(tmp_path,job,'prometheus') + def record(**event): + events.append(event) + recorder(**event) + result=updates.replace_container(engine,'prometheus','sha256:new',job['id'],timeout=0,record_phase=record) + assert result['state']=='rolled_back' + assert engine.inspect('prometheus')['Id']==source()['Id'] + stored=updates.read_job(tmp_path,job['id']) + assert 'SECRET_SENTINEL' not in json.dumps([stored,result,events]) + phases=stored['diagnostics']['prometheus'] + assert any(event.get('operation')==lost and event.get('phase')=='intent' for event in phases) + assert all(set(event)<={'phase','operation','error_category','http_status','elapsed','deadline','recovery_phase','recovery_outcome'} for event in phases) + assert phases[-1]['recovery_outcome']=='original_restored' + + +def test_worker_second_predecessor_check_blocks_drift_during_pull(recovered_predecessor,monkeypatch): + from dataclasses import replace + jobs,directory,digest,engine,old,receipt,observation=recovered_predecessor + successor=submit_successor(recovered_predecessor) + engine.selection=(directory/'selection.json').read_bytes() + pulled=[] + def pull(*a): + pulled.append(True) + monkeypatch.setattr(updates,'observe_original',lambda *a:replace(observation,healthy=False)) + return True + monkeypatch.setattr(updates,'pull_target',pull) + result=updates.run_job(jobs,successor['id'],engine=engine) + assert pulled and result['state']=='blocked' + assert result['reason']=='native_service_recovery_required' + assert not [c for c in engine.calls if c[0]=='stop'] + + +def test_original_helper_restarting_during_probe_cannot_publish_observation(): + engine=Engine();runtime=deepcopy(source()) + runtime.update(Id='1'*64,Image='sha256:'+'2'*64,Name='/ipmi-monitor') + runtime['State'].update(Status='running',Paused=False,Restarting=False,Dead=False,StartedAt='2026-11-02T00:00:00Z') + runtime['NetworkSettings']['Networks']['cryptolabs']['NetworkID']='3'*64 + engine.containers['ipmi-monitor']=runtime + from tests.test_update_recovery import receipt + job={'id':'a'*32,'helper':'cryptolabs-update-'+'a'*32,'sources':{'ipmi-monitor':{'container_id':runtime['Id'],'image_id':runtime['Image']}}} + def probe(_): + engine.containers[job['helper']]={'State':{'Status':'running','Running':True,'Paused':False,'Restarting':False}} + return {'continuity':{k:v for k,v in receipt()['continuity'].items() if k not in + ('projection','runtime_config_sha256','host_config_sha256','mounts_sha256','network_sha256')}, + 'last_collection':'2026-11-02T00:00:01Z'} + engine.recovery_probe=probe + with pytest.raises(ValueError):updates.observe_original(engine,job) + + +def test_lost_effect_never_removes_same_name_foreign_container(): + engine=Engine() + create=engine.create + def foreign_create(name,request): + create(name,request) + engine.containers[name]['Config']['Labels'].pop('io.cryptolabs.update.job',None) + raise OSError('uncertain creation') + engine.create=foreign_create + result=updates.replace_container(engine,'prometheus','sha256:new','d'*32,timeout=0) + assert result['state']=='recovery_required' + assert not [call for call in engine.calls if call[0]=='remove'] + assert engine.inspect('prometheus.rollback-'+'d'*32)['Id']==source()['Id'] + + +def test_engine_diagnostic_keeps_only_http_status_not_error_body(tmp_path): + engine=Engine() + engine.create=lambda *a:(_ for _ in ()).throw(updates.MigrationError('SECRET_SENTINEL returned HTTP 503')) + job={'id':'d'*32,'state':'running','results':{}} + updates.write_job(tmp_path,job) + result=updates.replace_container(engine,'prometheus','sha256:new',job['id'],timeout=0, + record_phase=updates.phase_recorder(tmp_path,job,'prometheus')) + assert result['state']=='rolled_back' + event=next(e for e in job['diagnostics']['prometheus'] if e.get('error_category')=='engine_error') + assert event['http_status']==503 + assert 'SECRET_SENTINEL' not in json.dumps(job) + + +def test_worker_revalidates_at_actual_first_stop_boundary(recovered_predecessor,monkeypatch): + from dataclasses import replace + jobs,directory,digest,engine,old,receipt,observation=recovered_predecessor + successor=submit_successor(recovered_predecessor) + successor['action']='update';updates.write_job(jobs,successor) + engine.selection=(directory/'selection.json').read_bytes() + monkeypatch.setattr(updates,'pull_target',lambda *a:True) + image=engine.image;observed=[] + def image_with_late_drift(reference): + if reference=='sha256:old': + observed.append(reference) + if len(observed)==4: + monkeypatch.setattr(updates,'observe_original',lambda *a:replace(observation,healthy=False)) + return image(reference) + engine.image=image_with_late_drift + result=updates.run_job(jobs,successor['id'],engine=engine) + assert len(observed)>=4 and result['state']=='blocked' + assert result['reason']=='native_service_recovery_required' + assert not [call for call in engine.calls if call[0]=='stop'] + + +@pytest.mark.parametrize('later_failure',[None,'create','health']) +def test_successful_service_update_does_not_revalidate_archived_original_mid_bundle(recovered_predecessor,monkeypatch,later_failure): + from cryptolabs_proxy import release_selection, update_recovery as recovery + jobs,directory,digest,engine,old,receipt,observation=recovered_predecessor + # Docker container IDs are unique; the shared old fixture aliases proxy + # and Prometheus IDs, which would make exact-ID rollback address the proxy. + engine.containers['cryptolabs-proxy']['Id']='e'*64 + historical=old['selection_snapshot'] + selected=deepcopy(historical['selection']) + selected['components']={'ipmi-monitor':selected['components']['ipmi-monitor'], + 'prometheus':selected['components']['prometheus']} + selected_raw=json.dumps(selected).encode() + manifest_raw=historical['manifest_bytes'].encode() + (directory/'selection.json').write_bytes(selected_raw) + (directory/'manifests'/(historical['manifest_sha256']+'.json')).write_bytes(manifest_raw) + digest=recovery.sha256(selected_raw) + ipmi=deepcopy(source());ipmi.update(Id=receipt['original_container_id'],Image=receipt['original_image_id'],Name='/ipmi-monitor') + engine.containers['ipmi-monitor']=ipmi + reference=release_selection.targets(historical,'linux/amd64')['ipmi-monitor']['reference'] + image=engine.image + def selected_image(value): + if value==reference:return {'Id':'sha256:'+'f'*64,'Config':{},'RepoDigests':[reference],'Os':'linux','Architecture':'amd64'} + return image(value) + engine.image=selected_image + # Give every created container a unique real-style ID so exact-ID effects + # can be distinguished across the helper and both service replacements. + create=engine.create + def unique_create(name,request): + if later_failure=='create' and name=='prometheus': + raise updates.MigrationError('synthetic later-service create failure') + create(name,request) + engine.containers[name]['Id']=__import__('hashlib').sha256(name.encode()).hexdigest() + engine.create=unique_create + def observe(*a): + if engine.inspect('ipmi-monitor')['Id']!=receipt['original_container_id']:recovery.hold() + return observation + monkeypatch.setattr(updates,'observe_original',observe) + successor=updates.submit_job('ipmi-monitor','main','update',directory=jobs,engine=engine, + selection_id=selected['selection_id'],selection_sha256=digest,selection_directory=directory, + components=['ipmi-monitor','prometheus'],operation_id='12345678-1234-4234-8234-123456789abc',request_fingerprint='a'*64) + engine.selection=selected_raw + monkeypatch.setattr(updates,'pull_target',lambda *a:True) + if later_failure=='health':engine.failure='health' + old_raw=(jobs/(old['id']+'.json')).read_bytes() + result=updates.run_job(jobs,successor['id'],engine=engine) + assert result['state']==('completed' if later_failure is None else 'failed') + assert result['success'] is (later_failure is None) + assert result['results']['ipmi-monitor']['state']=='completed' + assert result['results']['ipmi-monitor']['success'] is True + assert engine.inspect('ipmi-monitor')['Id']!=receipt['original_container_id'] + assert result['results']['prometheus']['state']==('completed' if later_failure is None else 'rolled_back') + assert engine.inspect('prometheus')['State']['Running'] is True + assert (jobs/(old['id']+'.json')).read_bytes()==old_raw + # This is the selector-equivalent dependency gate, followed by the actual + # submit path: unrelated verified rollback cannot re-pin the old IPMI ID. + assert updates.service_recovery_dependency(jobs,result,engine) is None + next_job=updates.submit_job('prometheus','main','pull',directory=jobs,engine=engine, + selection_id=selected['selection_id'],selection_sha256=digest,selection_directory=directory, + components=['prometheus'],operation_id='22345678-1234-4234-8234-123456789abc',request_fingerprint='b'*64) + assert next_job['state']=='queued' and 'recovery_predecessor' not in next_job diff --git a/traceability-matrix.md b/traceability-matrix.md index 50537d5..2ae8d20 100644 --- a/traceability-matrix.md +++ b/traceability-matrix.md @@ -8,7 +8,7 @@ implementation and installed customer acceptance are separate. | Trace | Requirement | Implementation | Verification | Validation | Acceptance status | Status | |---|---|---|---|---|---|---| -| TRACE-FLEETUPDATE-001 | REQ-FLEETUPDATE-001 | `src/cryptolabs_proxy/auth.py`, `src/cryptolabs_proxy/config.py`, `src/cryptolabs_proxy/release_client.py`, `src/cryptolabs_proxy/release_selection.py`, `src/cryptolabs_proxy/standard_adoption.py`, `src/cryptolabs_proxy/updates.py`, `scripts/health-api.py` | VER-FLEETUPDATE-001: affected existing suites and shared preimplementation RED artifacts | VAL-FLEETUPDATE-001: shared validation-definition.json | Local implementation candidate; installed/native/browser/owner acceptance held | approved | +| TRACE-FLEETUPDATE-001 | REQ-FLEETUPDATE-001 | `src/cryptolabs_proxy/auth.py`, `src/cryptolabs_proxy/config.py`, `src/cryptolabs_proxy/release_client.py`, `src/cryptolabs_proxy/release_selection.py`, `src/cryptolabs_proxy/standard_adoption.py`, `src/cryptolabs_proxy/updates.py`, `scripts/health-api.py`, `src/cryptolabs_proxy/update_recovery.py`, `tests/test_update_recovery.py`, `tests/test_updates.py`, `docs/native-service-recovery.md` | VER-FLEETUPDATE-001: affected existing suites and shared preimplementation RED artifacts | VAL-FLEETUPDATE-001: shared validation-definition.json | Local implementation candidate; installed/native/browser/owner acceptance held | approved | | TRACE-PAIR-001 | REQ-PAIR-001 | src/cryptolabs_proxy/auth.py; src/cryptolabs_proxy/account_link.py; tests/test_account_link.py; nginx/nginx.conf; src/cryptolabs_proxy/templates/nginx.conf.j2 | VER-PAIR-001 | VAL-PAIR-001 | Source verification only; installed/public/paid acceptance held | Owner-approved source authority; reviewed scope | | TRACE-PAIR-002 | REQ-PAIR-002 | src/cryptolabs_proxy/auth.py; src/cryptolabs_proxy/account_link.py; tests/test_account_link.py; nginx/nginx.conf; src/cryptolabs_proxy/templates/nginx.conf.j2 | VER-PAIR-002 | VAL-PAIR-002 | Source verification only; installed/public/paid acceptance held | Owner-approved source authority; reviewed scope | | TRACE-PAIR-003 | REQ-PAIR-003 | src/cryptolabs_proxy/auth.py; src/cryptolabs_proxy/account_link.py; tests/test_account_link.py; nginx/nginx.conf; src/cryptolabs_proxy/templates/nginx.conf.j2 | VER-PAIR-003 | VAL-PAIR-003 | Source verification only; installed/public/paid acceptance held | Owner-approved source authority; reviewed scope | @@ -43,6 +43,11 @@ Implementation candidate; all installed/runtime/native/browser/requirement closu | src/cryptolabs_proxy/standard_adoption.py | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | entrypoint | fleet-release-recovery | | src/cryptolabs_proxy/config.py | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | file-role | fleet-release-compose-consumer | | src/cryptolabs_proxy/config.py | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | entrypoint | fleet-release-compose-consumer | +| src/cryptolabs_proxy/update_recovery.py | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | file-role | native-service-recovery | +| src/cryptolabs_proxy/update_recovery.py | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | entrypoint | native-service-recovery | +| src/cryptolabs_proxy/updates.py | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | entrypoint | fleet-release-implementation | +| docs/native-service-recovery.md | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | file-role | recovery-operations-documentation | +| tests/test_update_recovery.py | REQ-FLEETUPDATE-001 | TRACE-FLEETUPDATE-001 | VER-FLEETUPDATE-001 | verifies | verification | | src/cryptolabs_proxy/account_link.py | REQ-PAIR-003 | TRACE-PAIR-003 | VER-PAIR-003 | file-role | pairing-store | | src/cryptolabs_proxy/auth.py | REQ-PAIR-004 | TRACE-PAIR-004 | VER-PAIR-004 | file-role | account-pairing-routes | @@ -98,3 +103,13 @@ These are generated raw data copies within the approved verification scope. Inli | nginx/nginx.conf | 5a6f8bbd6c9121fb0b1a2fd4884205b996f52c332ae2e0ea9b026fc5a748ed7d | TRACE-FLEETUPDATE-001; TRACE-PAIR-001 | proxy-integration/implementation/format-verification.json and affected.log; separate local Unix socket rerun; protected CI and installed/native/browser acceptance held | | src/cryptolabs_proxy/templates/nginx.conf.j2 | d14273f48d5370f480870068b8bcf1ebb2a84a868979079ff97549ef8fcdd7ed | TRACE-FLEETUPDATE-001; TRACE-PAIR-001 | proxy-integration/implementation/format-verification.json and affected.log; separate local Unix socket rerun; protected CI and installed/native/browser acceptance held | | docs/release-selection.md | 1fb846663fce84ae693409de61d222ac031fa6dd77e3eec1d27326ecaf2abbd0 | TRACE-FLEETUPDATE-001 | proxy-integration/implementation/format-verification.json and affected.log; separate local Unix socket rerun; protected CI and installed/native/browser acceptance held | + + +## Recovery amendment source evidence (2 November 2026) + +Authority is the exact approved amendment `159af842…` and frozen contract +`91b873c5…` linked from requirements.md. Focused RED/GREEN and candidate +hashes are retained at `../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/execution/build/manifest.json`. +Shared pre-code V&V is `../artifacts/ipmi-readiness-20261026/fleet-ui-20261102/execution/go/recovery-vv-root-reviewed.json`. +Native installed-wheel bootstrap, source publication, production reconciliation, +site continuity, signed admission and UI acceptance remain held.