From aa11347d5e17aedd251d3fb9dfd579e23a1e85e7 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 17:14:08 +0000 Subject: [PATCH 1/3] Add installed CLI command-key acceptance adapters --- scripts/acceptance/command-keys/README.md | 94 ++++ scripts/acceptance/command-keys/cli_216.py | 415 ++++++++++++++++++ scripts/acceptance/command-keys/legacy_cli.py | 251 +++++++++++ 3 files changed, 760 insertions(+) create mode 100644 scripts/acceptance/command-keys/README.md create mode 100644 scripts/acceptance/command-keys/cli_216.py create mode 100644 scripts/acceptance/command-keys/legacy_cli.py diff --git a/scripts/acceptance/command-keys/README.md b/scripts/acceptance/command-keys/README.md new file mode 100644 index 0000000..db00802 --- /dev/null +++ b/scripts/acceptance/command-keys/README.md @@ -0,0 +1,94 @@ +# Installed CLI command-key experiments + +These manual adapters exercise shared [#216](https://github.com/durable-workflow/.github/issues/216) +with the published CLI 2.8.0 binary and native Server 3.9.2, linux/amd64. +They build no package and use only disposable synthetic SQLite state. Run one +adapter at a time after the normal host-capacity/owner check. Docker runtime +limits are one CPU, 1 GiB and 256 PIDs, with a loopback-only ephemeral port. + +The modern adapter covers admission, committed-response loss, explicit retry, +concurrency, Unicode limits, terminal runs and namespace fences. Its second +phase restarts the actual Server process on the same database and checks the +previously lost receipts and explicitly selected terminal runs. The legacy +adapter makes current-producer API admissions, stops that Server, transforms a +separate copy through the accepted helper, and checks legacy readback and +ambiguity refusals. **The synthetic transform is not a physical 3.8 upgrade.** +Physical upgrade input and the producer's accepted seed procedure have their +own source/qualification gate. Neither adapter claims another SDK's Worker +acceptance. + +Prepare an empty, canonical directory owned by the invoking user, mode `0700`, +on disk-backed scratch. Put the checksum/attestation-verified published binary +at `bin/dw` inside it. Expected binary SHA256: +`da366c79c90978a3c0a92d723c66a96a2d0963f9dccbd41c8207f2c41848f814`. +The binary must be executable. The runtime image is pulled separately, without +registry credentials, using the exact index in `cli_216.py`. Independently +verify the index/platform manifest and source revision before execution; do +not equate an engine's image ID with the raw OCI config digest. + +Stage these public, reviewed source files in the experiment directory: + +| File | Accepted source | SHA256 | +| --- | --- | --- | +| `committed_response_loss.py` | `.github` commit `117118c83093822469b3824ba9587ff4068ed29f`, `conformance/command-keys/committed_response_loss.py` | `bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725` | +| `sqlite_legacy.py` | `sdk-rust` commit `4add871b175517f2031f71c868ce278bbfbba486`, `scripts/acceptance/command-keys/sqlite_legacy.py` | `6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad` | +| `command-key-legacy-v1-vectors.json` | `server` commit `d66e33bfb5c76f94c5b202f8c9f7a9e8cd75cf1c`, `resources/platform-protocol-specs/command-key-legacy-v1-vectors.json` | `da1990000c611d7377690c7526487d6415fa4f9cb42d299f59585a7c2eebf769` | + +The helper reads the adjacent manifest. Both are used only by `legacy_cli.py`; +the proxy is shared by both adapters. Their identities are checked before import. +Do not replace them with a working tree overlay. + +Use a separately verified installed official Python SDK 2.17.1 environment only +for Avro fixture encoding/decoding. Its PyPI wheel SHA256 is +`90040cb28906985b227988815b1b53aa3079378673e35c689392ac71edaf7b35`. +Check installed package files against that wheel and retain module-origin and +dependency evidence before using its Python executable. This is a codec +fixture, not Python Worker qualification. Generate `worker-envelopes.json` in +the private experiment directory with that interpreter: + +```python +import json +from durable_workflow import serializer + +print(json.dumps({ + "empty": serializer.envelope(None, codec="avro"), + "one": serializer.envelope(1, codec="avro"), +})) +``` + +Run the source from this directory with isolated host Python. Supply an +absolute verified codec interpreter path in `CLI_216_CODEC_PYTHON` and the +private directory in `CLI_216_WORKDIR`. Provide a clean environment without +cloud, GitHub, Composer or registry credentials. The adapters create synthetic +runtime authentication privately and remove its environment file on exit. + +```bash +env -i PATH=/usr/bin:/bin \ + CLI_216_WORKDIR="$experiment" \ + CLI_216_PROXY_SOURCE="$experiment/committed_response_loss.py" \ + CLI_216_CODEC_PYTHON="$codec_python" \ + timeout 180s python3 -I -B scripts/acceptance/command-keys/cli_216.py +``` + +For the restart phase, preserve the first phase's `admissions.json` as +`modern-admissions.json`, preserve its result/wire records separately, then run +the same command with `CLI_216_PHASE=selected-restart`. Preserve that phase's +records separately too. `CLI_216_PHASE=fresh-selected` adds direct fresh +explicit-run literal-key markers (the loss cells intentionally withhold those +initial response bodies). Use a **new** private +directory for `legacy_cli.py`, with the same staged inputs and environment. +The legacy adapter generates its own admissions and stopped-copy database. + +An outer experiment owner must also enforce the timeout and remove only the +exact owned container if the process is interrupted. Names are +`cli-216-392-modern` and `cli-216-392-legacy`; remove their anonymous volumes +too, and verify absence. No listener or production service is retained. + +Keep generated databases, response bodies and invocation records private and +bounded in retention. Publish sanitized tuple/source hashes, timestamps, +scenario outcomes, snapshot fingerprints, nonpasses and cleanup to #216. +The wire fixture records only metadata and hashes. A script exit alone does +not classify a failed assertion as a product defect: preserve the first result +and diagnose runner/source/cleanup failures explicitly. A later corrected run +does not erase it. The adapters are manual evidence producers, with no workflow +activation, release authority, task scheduler or automatic retry service. diff --git a/scripts/acceptance/command-keys/cli_216.py b/scripts/acceptance/command-keys/cli_216.py new file mode 100644 index 0000000..d5647fa --- /dev/null +++ b/scripts/acceptance/command-keys/cli_216.py @@ -0,0 +1,415 @@ +"""Installed CLI 2.8 / native 3.9.2 synthetic admission cohort. No builds.""" +import concurrent.futures +import datetime +import hashlib +import json +import os +from pathlib import Path +import secrets +import shutil +import sqlite3 +import subprocess +import threading +import time +import urllib.error +import urllib.request +import importlib.util + +PROXY_SOURCE = Path(os.environ['CLI_216_PROXY_SOURCE']).resolve() +PROXY_SHA256 = 'bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725' +if hashlib.sha256(PROXY_SOURCE.read_bytes()).hexdigest() != PROXY_SHA256: + raise SystemExit('Shared proxy bytes do not match the independently reviewed fixture') +spec = importlib.util.spec_from_file_location('cli216_shared_proxy', PROXY_SOURCE) +shared_proxy = importlib.util.module_from_spec(spec) +spec.loader.exec_module(shared_proxy) +CommittedResponseLoss = shared_proxy.CommittedResponseLoss + +ROOT = Path(os.environ['CLI_216_WORKDIR']) +if not ROOT.is_absolute() or ROOT.resolve() != ROOT or ROOT.is_symlink(): + raise SystemExit('Require an absolute canonical private experiment directory') +owned = ROOT.stat() +if owned.st_uid != os.getuid() or owned.st_mode & 0o777 != 0o700: + raise SystemExit('Experiment directory must be owned by this user and mode 0700') +CODEC_PYTHON = Path(os.environ['CLI_216_CODEC_PYTHON']) +if not CODEC_PYTHON.is_absolute(): + raise SystemExit('Provide the verified installed codec Python as an absolute path') +IMAGE = 'ghcr.io/durable-workflow/server@sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9' +NAME = 'cli-216-392-modern' +TOKEN = secrets.token_hex(32) +STARTED = datetime.datetime.now(datetime.timezone.utc).isoformat() +CHECKS, INVOCATIONS, ADMISSIONS = [], [], [] +BASE = None +PROXY = None +DB = ROOT / 'data/database.sqlite' +ENV = {'PATH': str(ROOT / 'bin') + ':/usr/bin:/bin', 'HOME': str(ROOT / 'home'), + 'DW_CONFIG_HOME': str(ROOT / 'config'), 'TMPDIR': str(ROOT / 'tmp'), + 'DURABLE_WORKFLOW_AUTH_TOKEN': TOKEN} +for directory in ('data', 'home', 'config', 'tmp', 'docker-empty'): + (ROOT / directory).mkdir(exist_ok=True) +(ROOT / 'data').chmod(0o700) + + +def require(condition, label, detail=None): + if not condition: + raise AssertionError(label + (': ' + json.dumps(detail, default=str) if detail is not None else '')) + CHECKS.append(label) + print(json.dumps({'check': label, 'outcome': 'pass'}), flush=True) + + +def docker(*args): + p = subprocess.run(['docker', '--config', str(ROOT / 'docker-empty'), *args], capture_output=True, text=True, timeout=50, env={'PATH': os.defpath, 'HOME': str(ROOT / 'home')}) + if p.returncode: + raise RuntimeError('docker operation failed: ' + p.stderr.replace(TOKEN, '[redacted]')[:2000]) + return p.stdout.strip() + + +def request(method, path, body=None, namespace='default'): + headers = {'Authorization': 'Bearer ' + TOKEN, 'X-Namespace': namespace, 'Content-Type': 'application/json'} + worker = path.startswith('/api/worker/') + headers['X-Durable-Workflow-Protocol-Version' if worker else 'X-Durable-Workflow-Control-Plane-Version'] = '1.20' if worker else '2' + r = urllib.request.Request(BASE + path, method=method, headers=headers, + data=None if body is None else json.dumps(body).encode()) + try: + with urllib.request.urlopen(r, timeout=15) as response: + return response.status, json.load(response) + except urllib.error.HTTPError as e: + return e.code, json.load(e) + + +def cli(*args, namespace='default', success=True): + p = subprocess.run(['dw', *args, '--server=' + PROXY.url, '--namespace=' + namespace, '--json'], + env=ENV, capture_output=True, text=True, timeout=45) + try: + body = json.loads(p.stdout) + except ValueError: + body = {'stdout': p.stdout, 'stderr': p.stderr} + INVOCATIONS.append({'arguments': args, 'namespace': namespace, 'exit': p.returncode, 'body': body}) + if (p.returncode == 0) != success: + raise AssertionError(json.dumps({'args': args, 'exit': p.returncode, 'body': body, 'stderr': p.stderr})) + return body + + +def message(kind, workflow, key=None, value=1, name=None, run=None, namespace='default', success=True): + args = ['workflow:' + kind, workflow, name or ('finish' if kind == 'signal' else 'set'), '--input=' + json.dumps([value])] + if key is not None: + args.append('--request-id=' + key) + if run is not None: + args.append('--run-id=' + run) + return cli(*args, namespace=namespace, success=success) + + +def path_for(kind, fixture, name=None, run=None): + path = '/api/workflows/' + fixture['workflow'] + if run: + path += '/runs/' + run + return path + '/' + kind + '/' + (name or ('finish' if kind == 'signal' else 'set')) + + +def poll(f): + status, body = request('POST', '/api/worker/workflow-tasks/poll', + {'worker_id': f['worker'], 'task_queue': f['queue'], 'timeout_seconds': 0}, f['namespace']) + if status != 200: + raise AssertionError(('poll', status, body)) + return body.get('task') + + +def finish(f, task, commands): + if not task: + raise AssertionError('expected a leased workflow task') + status, body = request('POST', '/api/worker/workflow-tasks/' + task['task_id'] + '/complete', + {'lease_owner': task['lease_owner'], 'workflow_task_attempt': task['workflow_task_attempt'], + 'commands': commands}, f['namespace']) + if status != 200: + raise AssertionError(('complete', status, body)) + return body + + +def fixture(kind, tag, namespace='default', workflow=None): + wf = workflow or 'cli216-' + tag + '-' + secrets.token_hex(4) + f = {'workflow': wf, 'worker': wf + '-w', 'queue': wf + '-q', 'type': wf + '-t', 'namespace': namespace} + param = {'name': 'value', 'type': 'int', 'required': True, 'variadic': False, + 'default_available': False, 'allows_null': False} + contracts = {'signals': ['finish'], 'signal_contracts': [{'name': 'finish', 'parameters': [param]}], + 'updates': ['set'], 'update_validators': [], 'update_contracts': [{'name': 'set', 'parameters': [param]}]} + status, body = request('POST', '/api/worker/register', { + 'worker_id': f['worker'], 'task_queue': f['queue'], 'runtime': 'python', + 'supported_workflow_types': [f['type']], 'supported_activity_types': [], 'capabilities': ['workflow_updates'], + 'capability_manifest': {name: {'supported': False, 'minimum_protocol_version': '1.18'} + for name in ('local_activities', 'worker_sessions', 'sticky_execution')}, + 'workflow_command_contracts': {f['type']: contracts}}, namespace) + if status != 201: + raise AssertionError(('register', status, body)) + started = cli('workflow:start', '--workflow-id=' + wf, '--type=' + f['type'], '--task-queue=' + f['queue'], '--input=[]', namespace=namespace) + f['run'] = started['run_id'] + f['empty_envelope'] = json.loads((ROOT / 'worker-envelopes.json').read_text())['empty'] + if kind == 'update': + finish(f, poll(f), [{'type': 'open_signal_wait', 'signal_name': 'finish'}]) + return f + + +def history(f): + status, body = request('GET', '/api/workflows/' + f['workflow'] + '/runs/' + f['run'] + '/history', namespace=f['namespace']) + if status != 200: + raise AssertionError(('history', status, body)) + return body['events'] + + +def snapshot(): + with sqlite3.connect('file:' + str(DB) + '?mode=ro', uri=True) as db: + tables = [r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")] + data = {} + for table in tables: + quoted = '"' + table.replace('"', '""') + '"' + rows = db.execute('SELECT * FROM ' + quoted).fetchall() + data[table] = {'count': len(rows), 'sha256': hashlib.sha256(repr(sorted(rows, key=repr)).encode()).hexdigest()} + return data + + +def original(kind, f, key=None, lost=False, run=None): + path = path_for(kind, f, run=run) + before = len(PROXY.records) + if lost: + PROXY.arm('POST', path) + answer = message(kind, f['workflow'], key, run=run, namespace=f['namespace'], success=not lost) + sent = [r for r in PROXY.records[before:] if r['method'] == 'POST' and r['path'] == path] + require(len(sent) == 1, kind + ': exactly one admission POST; no implicit resend') + wire = sent[0] + require(wire['upstream_status'] == 202 and wire['upstream_complete'], + kind + ': verified complete upstream admission before any injected loss') + chosen = answer.get('request_id') + require(isinstance(chosen, str) and bool(chosen), kind + ': CLI exposes a nonempty chosen command key') + if key is not None: + require(chosen == key, kind + ': CLI reports the explicit caller key') + if lost: + require(wire['dropped'] and wire['upstream_body_bytes'] > 0, + kind + ': complete upstream receipt dropped before downstream status') + admitted = snapshot() + recovery_start = len(PROXY.records) + receipt = message(kind, f['workflow'], chosen, run=run, namespace=f['namespace']) + recovered = [r for r in PROXY.records[recovery_start:] if r['method'] == 'POST' and r['path'] == path] + require(len(recovered) == 1, kind + ': one explicit recovery POST') + require(recovered[0]['request_body_sha256'] == wire['request_body_sha256'], + kind + ': explicit recovery sends identical request bytes without retaining their contents') + require(receipt.get('retained') is True and receipt.get('intent_verified') is True, + kind + ': explicit recovery verifies retained intent') + require(snapshot() == admitted, kind + ': first explicit recovery changes no database row') + else: + receipt = answer + require(receipt.get('retained') is False and receipt.get('intent_verified') is True, + kind + ': fresh receipt explicitly reports verified non-retained intent') + fields = ('command_id', 'run_id', 'command_sequence', 'accepted_at', 'update_id', 'signal_name') + owner = {'request_id': chosen, 'receipt': {k: receipt[k] for k in fields if k in receipt}, + 'path': path, 'dropped': lost, 'wire': dict(wire)} + ADMISSIONS.append(owner) + return owner + + +def retained(kind, f, record, value=1, run=None): + before = snapshot() + answer = message(kind, f['workflow'], record['request_id'], value=value, run=run, namespace=f['namespace']) + receipt = record['receipt'] + for key in ('command_id', 'run_id', 'command_sequence', 'accepted_at', 'update_id', 'signal_name'): + if key in receipt: + require(answer.get(key) == receipt[key], kind + ': retained original ' + key, answer) + require(answer.get('retained') is True and answer.get('intent_verified') is True, kind + ': verified retained intent') + require(snapshot() == before, kind + ': retained retry changes no database row') + return answer + + +def conflict(kind, f, record, part, **options): + before = snapshot() + answer = message(kind, f['workflow'], record['request_id'], namespace=f['namespace'], success=False, **options) + # CLI error JSON nests the Server's typed body under response on some paths. + candidates = [answer] + [v for v in answer.values() if isinstance(v, dict)] + typed = next((v for v in candidates if v.get('reason') == 'request_id_conflict'), None) + require(typed is not None, kind + ': typed conflict for ' + part, answer) + require(typed.get('conflict') == [('target_or_arguments' if options.get('run') is not None else 'arguments') if part == 'target' else ('arguments' if kind == 'update' and part == 'name' else part)] and typed.get('retained_command_id') == record['receipt']['command_id'], kind + ': conflict identifies original intent component ' + part, typed) + require(snapshot() == before, kind + ': conflict changes no database row') + + +def decoded_result(answer): + envelope = answer.get('result_envelope') + p = subprocess.run([str(CODEC_PYTHON), '-I', '-B', '-c', + 'import json,sys; from durable_workflow import serializer; print(json.dumps(serializer.decode_envelope(json.load(sys.stdin))))'], + input=json.dumps(envelope), capture_output=True, text=True, timeout=10, + env={'PATH': '/usr/bin:/bin', 'HOME': str(ROOT / 'home'), 'TMPDIR': str(ROOT / 'tmp')}) + if p.returncode: + raise AssertionError('official Avro fixture decode failed') + return json.loads(p.stdout) + + +def run_modern(): + require(shutil.which('dw', path=ENV['PATH']) == str(ROOT / 'bin/dw'), 'ordinary scoped PATH selects installed release') + require(hashlib.sha256((ROOT / 'bin/dw').read_bytes()).hexdigest() == 'da366c79c90978a3c0a92d723c66a96a2d0963f9dccbd41c8207f2c41848f814', 'published CLI binary SHA matches approved release') + status, info = request('GET', '/api/cluster/info') + require(status == 200 and info['version'] == '3.9.2' and info['capabilities']['command_request_keys'] is True, 'published Server 3.9.2 advertises command_request_keys', info.get('version')) + require(info['worker_protocol']['version'] == '1.20', 'worker protocol 1.20') + for kind in ('signal', 'update'): + f = fixture(kind, kind + '-loss') + record = original(kind, f, lost=True) + retained(kind, f, record) + conflict(kind, f, record, 'arguments', value=2) + conflict(kind, f, record, 'name', name='other') + conflict(kind, f, record, 'target', run=f['run']) + key = kind + '-concurrent' + with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool: + replies = list(pool.map(lambda _: message(kind, f['workflow'], key), range(4))) + require(len({r['command_id'] for r in replies}) == 1 and sum(r.get('retained') is True for r in replies) == 3, kind + ': four simultaneous retries admit exactly one command') + event_type = 'SignalReceived' if kind == 'signal' else 'UpdateAccepted' + require(sum(e['event_type'] == event_type for e in history(f)) == 2, kind + ': history has one event per original and concurrent key') + long_record = original(kind, f, key='k' * 255) + retained(kind, f, long_record) + for label, unicode_key in (('two-byte', 'é' * 255), ('four-byte', '🙂' * 255)): + unicode_record = original(kind, f, key=unicode_key) + retained(kind, f, unicode_record) + require(len(unicode_key) == 255, kind + ': admitted 255 code points for ' + label) + refused_before = snapshot() + wire_before = len(PROXY.records) + message(kind, f['workflow'], key='é' * 256, namespace=f['namespace'], success=False) + require(not any(r['method'] == 'POST' for r in PROXY.records[wire_before:]), + kind + ': 256-code-point key refused before admission POST') + require(snapshot() == refused_before, kind + ': overlength refusal changes no row') + a = message(kind, f['workflow'], key='CaseSensitive') + b = message(kind, f['workflow'], key='casesensitive') + require(a['command_id'] != b['command_id'], kind + ': key identity is case sensitive') + for terminal in ('complete_workflow', 'continue_as_new'): + f = fixture(kind, kind + '-' + terminal) + record = original(kind, f, key=kind + '-' + terminal, lost=True) + commands = [] + if kind == 'update': + commands.append({'type': 'complete_update', 'update_id': record['receipt']['update_id'], 'result': json.loads((ROOT / 'worker-envelopes.json').read_text())['one']}) + commands.append({'type': terminal, 'result': f['empty_envelope']}) + terminal_event = 'WorkflowCompleted' if terminal == 'complete_workflow' else 'WorkflowContinuedAsNew' + for _ in range(10): + finish(f, poll(f), commands) + if any(e['event_type'] == terminal_event for e in history(f)): + break + require(any(e['event_type'] == terminal_event for e in history(f)), kind + ': terminal transition recorded before retry') + answer = retained(kind, f, record) + if kind == 'update': + require(answer.get('update_status') == 'completed' and decoded_result(answer) == 1, 'update: original completed typed result survives ' + terminal, answer) + require(any(e['event_type'] == ('WorkflowCompleted' if terminal == 'complete_workflow' else 'WorkflowContinuedAsNew') for e in history(f)), kind + ': original history proves ' + terminal) + conflict(kind, f, record, 'arguments', value=2) + conflict(kind, f, record, 'target', run='00000000000000000000000000') + f = fixture(kind, kind + '-selected') + record = original(kind, f, key=kind + '-explicit-run', run=f['run'], lost=True) + retained(kind, f, record, run=f['run']) + conflict(kind, f, record, 'target') + request('POST', '/api/namespaces', {'name': 'cli216-other'}) + for kind in ('signal', 'update'): + common = 'cli216-ns-' + kind + a = fixture(kind, common, workflow=common) + b = fixture(kind, common, namespace='cli216-other', workflow=common + '-other') + ra = message(kind, a['workflow'], key='same-key') + rb = message(kind, b['workflow'], key='same-key', namespace=b['namespace']) + require(ra['command_id'] != rb['command_id'] and ra['namespace'] != rb['namespace'], kind + ': same caller key admits independently in each namespace') + before = snapshot() + foreign = message(kind, a['workflow'], key='same-key', namespace=b['namespace'], success=False) + require(foreign.get('reason') in ('instance_not_found', 'workflow_not_found'), kind + ': foreign namespace cannot recover another workflow receipt', foreign) + require(foreign.get('command_id') is None and foreign.get('update_id') is None, kind + ': foreign namespace leaks no retained identity') + require(snapshot() == before, kind + ': foreign namespace refusal changes no database row') + + +def run_selected_and_restart(): + status, info = request('GET', '/api/cluster/info') + require(status == 200 and info['version'] == '3.9.2', 'same published Server version after process restart') + previous = json.loads((ROOT / 'modern-admissions.json').read_text()) + recovered = 0 + for record in previous: + if not record['dropped']: + continue + parts = record['path'].split('/') + kind, workflow = parts[-2], parts[3] + f = {'workflow': workflow, 'run': record['receipt']['run_id'], 'namespace': 'default'} + selected = parts[5] if 'runs' in parts else None + retained(kind, f, record, run=selected) + recovered += 1 + require(recovered == 8, 'all eight previous lost-response receipts survive actual runtime process restart') + for kind in ('signal', 'update'): + for terminal in ('complete_workflow', 'continue_as_new'): + f = fixture(kind, 'selected-' + kind + '-' + terminal) + record = original(kind, f, key='selected-' + kind + '-' + terminal, run=f['run'], lost=True) + commands = [] + if kind == 'update': + commands.append({'type': 'complete_update', 'update_id': record['receipt']['update_id'], + 'result': json.loads((ROOT / 'worker-envelopes.json').read_text())['one']}) + commands.append({'type': terminal, 'result': f['empty_envelope']}) + terminal_event = 'WorkflowCompleted' if terminal == 'complete_workflow' else 'WorkflowContinuedAsNew' + for _ in range(10): + finish(f, poll(f), commands) + if any(e['event_type'] == terminal_event for e in history(f)): + break + require(any(e['event_type'] == terminal_event for e in history(f)), kind + ': selected-run terminal transition recorded') + answer = retained(kind, f, record, run=f['run']) + if kind == 'update': + require(answer.get('update_status') == 'completed' and decoded_result(answer) == 1, + 'selected update: original typed result survives ' + terminal) + conflict(kind, f, record, 'target') + conflict(kind, f, record, 'target', run='00000000000000000000000000') + + +def run_fresh_selected(): + for kind in ('signal', 'update'): + f = fixture(kind, 'fresh-selected-' + kind) + record = original(kind, f, key='🧪' * 255, run=f['run']) + require(record['receipt']['run_id'] == f['run'], kind + ': fresh selected-run receipt names the explicit original run') + retained(kind, f, record, run=f['run']) + + +if __name__ == '__main__': + outcome, error = 'runner-blocked', None + try: + observed = json.loads(docker('image', 'inspect', IMAGE, '--format', '{{json .}}')) + require(observed['Architecture'] == 'amd64' and observed['Os'] == 'linux', 'observed native platform is linux/amd64') + require(observed['Config']['Labels']['org.opencontainers.image.revision'] == '5614274d2629d03f92932a07dfee0fb10349ddd9', 'observed image source matches qualified producer') + envfile = ROOT / 'runtime.env' + envfile.write_text('DW_AUTH_TOKEN=' + TOKEN + '\nDW_RUST_MAINTENANCE_LOOP=0\n') + envfile.chmod(0o600) + docker('run', '--rm', '-d', '--name', NAME, '--platform=linux/amd64', '--init', '--cpus=1', '--memory=1g', '--pids-limit=256', + '--read-only', '--cap-drop=ALL', '--security-opt=no-new-privileges', '--env-file', str(envfile), + '--mount', 'type=bind,source=' + str(ROOT / 'data') + ',target=/app/database', + '--tmpfs', '/app/storage:rw,uid=1000,gid=1000,mode=750,size=64m', + '--tmpfs', '/tmp:rw,size=16m', '-p', '127.0.0.1::8080', IMAGE) + binding = json.loads(docker('inspect', NAME, '--format', '{{json .NetworkSettings.Ports}}'))['8080/tcp'][0] + BASE = 'http://127.0.0.1:' + binding['HostPort'] + for _ in range(100): + try: + if request('GET', '/api/cluster/info')[0] == 200: + break + except (OSError, ValueError): + time.sleep(.1) + else: + raise RuntimeError('runtime did not become ready') + PROXY = CommittedResponseLoss(BASE) + threading.Thread(target=PROXY.serve_forever, daemon=True).start() + if os.environ.get('CLI_216_PHASE') == 'selected-restart': + run_selected_and_restart() + elif os.environ.get('CLI_216_PHASE') == 'fresh-selected': + run_fresh_selected() + else: + run_modern() + outcome = 'pass' + except Exception as e: + error = (type(e).__name__ + ': ' + str(e)).replace(TOKEN, '[redacted]') + print(json.dumps({'outcome': outcome, 'error': error}), flush=True) + finally: + if PROXY: + PROXY.shutdown() + PROXY.server_close() + try: + docker('rm', '-fv', NAME) + cleanup = 'container and anonymous volumes removed' + except Exception: + cleanup = 'container removal needs investigation' + (ROOT / 'runtime.env').unlink(missing_ok=True) + report = {'started_at': STARTED, 'finished_at': datetime.datetime.now(datetime.timezone.utc).isoformat(), + 'outcome': outcome, 'checks': CHECKS, 'error': error, 'cleanup': cleanup, + 'runner_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + 'proxy_sha256': hashlib.sha256(PROXY_SOURCE.read_bytes()).hexdigest(), 'server_index': IMAGE, 'server_source': '5614274d2629d03f92932a07dfee0fb10349ddd9', 'phase': os.environ.get('CLI_216_PHASE', 'modern')} + (ROOT / 'result.json').write_text(json.dumps(report, indent=2)) + (ROOT / 'invocations.json').write_text(json.dumps(INVOCATIONS, indent=2).replace(TOKEN, '[redacted]')) + (ROOT / 'admissions.json').write_text(json.dumps(ADMISSIONS, indent=2)) + if PROXY: + (ROOT / 'wire.json').write_text(json.dumps(PROXY.records, indent=2).replace(TOKEN, '[redacted]')) + print(json.dumps(report), flush=True) + raise SystemExit(0 if outcome == 'pass' else 1) diff --git a/scripts/acceptance/command-keys/legacy_cli.py b/scripts/acceptance/command-keys/legacy_cli.py new file mode 100644 index 0000000..905c436 --- /dev/null +++ b/scripts/acceptance/command-keys/legacy_cli.py @@ -0,0 +1,251 @@ +"""CLI-owned synthetic legacy cells; never a physical old-image upgrade.""" +import concurrent.futures +from contextlib import closing +import datetime +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import sqlite3 +import threading +import time + +SOURCE = Path(__file__).resolve().parent +ROOT = Path(os.environ["CLI_216_WORKDIR"]) + + +def bound_module(name, filename, expected): + path = SOURCE / filename if filename == "cli_216.py" else ROOT / filename + if hashlib.sha256(path.read_bytes()).hexdigest() != expected: + raise ValueError("fixture source identity differs: " + filename) + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +d = bound_module("cli216_owner", "cli_216.py", "d3833b37e96dab863c8652bfe65c09ca5f632dcdee985637b443c4f66a56898e") +legacy = bound_module("cli216_legacy", "sqlite_legacy.py", "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad") +d.NAME = "cli-216-392-legacy" +manifest = legacy.manifest() +cases, selections, windows = [], [], [] +started = datetime.datetime.now(datetime.timezone.utc).isoformat() + + +def full_snapshot(): + with closing(sqlite3.connect(d.DB.as_uri() + "?mode=ro", uri=True, timeout=2)) as db: + return legacy.snapshot(db) + + +def start(): + envfile = ROOT / "runtime.env" + envfile.write_text("DW_AUTH_TOKEN=" + d.TOKEN + "\nDW_RUST_MAINTENANCE_LOOP=0\n") + envfile.chmod(0o600) + d.docker("run", "--rm", "-d", "--name", d.NAME, "--platform=linux/amd64", "--init", + "--cpus=1", "--memory=1g", "--pids-limit=256", "--read-only", "--cap-drop=ALL", + "--security-opt=no-new-privileges", "--env-file", str(envfile), + "--mount", "type=bind,source=" + str(ROOT / "data") + ",target=/app/database", + "--tmpfs", "/app/storage:rw,uid=1000,gid=1000,mode=750,size=64m", + "--tmpfs", "/tmp:rw,size=16m", "-p", "127.0.0.1::8080", d.IMAGE) + binding = json.loads(d.docker("inspect", d.NAME, "--format", "{{json .NetworkSettings.Ports}}"))["8080/tcp"][0] + d.BASE = "http://127.0.0.1:" + binding["HostPort"] + for _ in range(100): + try: + status, info = d.request("GET", "/api/cluster/info") + if status == 200: + break + except (OSError, ValueError): + pass + time.sleep(.1) + else: + raise RuntimeError("Server did not become ready") + d.require(info["version"] == "3.9.2" and info["worker_protocol"]["version"] == "1.20", + "published corrected Server version/protocol") + d.PROXY = d.CommittedResponseLoss(d.BASE) + threading.Thread(target=d.PROXY.serve_forever, daemon=True).start() + + +def stop(): + if d.PROXY: + d.PROXY.shutdown() + d.PROXY.server_close() + d.PROXY = None + d.docker("stop", "--time=10", d.NAME) + for _ in range(50): + if not d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$"): + break + time.sleep(.1) + else: + raise RuntimeError("owned automatic removal did not finish within five seconds") + d.require(True, "exact owned runtime is absent before offline copy") + + +def select(kind, fixture, record, stored): + selections.append({"command_id": record["receipt"]["command_id"], "command_type": kind, + "workflow_instance_id": fixture["workflow"], "workflow_run_id": fixture["run"], + "caller_key": record["request_id"], "legacy_request_id": stored}) + + +def terminal(kind, f, record, action): + commands = [] + if kind == "update": + commands.append({"type": "complete_update", "update_id": record["receipt"]["update_id"], + "result": f["empty_envelope"]}) + commands.append({"type": action, "result": f["empty_envelope"]}) + event = "WorkflowCompleted" if action == "complete_workflow" else "WorkflowContinuedAsNew" + for _ in range(10): + d.finish(f, d.poll(f), commands) + if any(row["event_type"] == event for row in d.history(f)): + break + d.require(any(row["event_type"] == event for row in d.history(f)), kind + ": original " + action + " recorded") + + +def seed(): + for kind in manifest["legacy_readback"]["types"]: + for state, action in [("open_after_restart", None), ("completed_after_restart", "complete_workflow"), + ("continued_after_restart", "continue_as_new")]: + f = d.fixture(kind, "legacy-" + kind + "-" + state) + record = d.original(kind, f, key=manifest["keys"]["original"]) + if action: + terminal(kind, f, record, action) + select(kind, f, record, record["request_id"]) + cases.append({"kind": kind, "fixture": f, "record": record, "state": state, "shape": "readback"}) + for shape in ["two-rows", "legacy-token", "modern-raw"]: + f = d.fixture(kind, "ambiguity-" + kind + "-" + shape) + record = d.original(kind, f, key=manifest["keys"]["original"]) + token = manifest["keys"]["tokens"][kind] + if shape == "two-rows": + second = d.original(kind, f, key=manifest["keys"]["second"]) + select(kind, f, second, manifest["keys"]["original"]) + elif shape == "legacy-token": + select(kind, f, record, token) + cases.append({"kind": kind, "fixture": f, "record": record, "shape": shape, + "retry_key": token if shape == "modern-raw" else manifest["keys"]["original"]}) + f = d.fixture(kind, "case-" + kind) + record = d.original(kind, f, key=manifest["keys"]["case_exact"]) + select(kind, f, record, record["request_id"]) + cases.append({"kind": kind, "fixture": f, "record": record, "shape": "case"}) + + +def readback(case, *, run=None): + kind, f, record = case["kind"], case["fixture"], case["record"] + answer = d.message(kind, f["workflow"], record["request_id"], value=manifest["legacy_readback"]["retry_input"], run=run) + for field, value in record["receipt"].items(): + d.require(answer.get(field) == value, kind + ": legacy original " + field) + d.require(answer.get("retained") is True and answer.get("intent_verified") is False, + kind + ": legacy readback is retained and explicitly unverified") + d.require(answer.get("request_id") == record["request_id"], kind + ": literal legacy caller key") + if kind == "update" and case.get("state") in ("completed_after_restart", "continued_after_restart"): + d.require(answer.get("update_status") == "completed" and "result" in answer and answer["result"] is None, + "legacy update remains completed null after " + case["state"]) + return answer + + +def typed_error(answer, reason): + values = [v for v in answer.values() if isinstance(v, dict)] + [answer] + result = next((v for v in values if v.get("reason") == reason), None) + d.require(result is not None, "typed " + reason) + return result + + +def unchanged(before, label): + after = full_snapshot() + d.require(before == after, label + ": complete logical schema/rows unchanged") + windows.append({"cell": label, "before_sha256": legacy.fingerprint(before), + "after_sha256": legacy.fingerprint(after)}) + + +def exercise(): + for case in cases: + kind, f, record = case["kind"], case["fixture"], case["record"] + label = kind + ":" + case["shape"] + ":" + case.get("state", "open") + before = full_snapshot() + if case["shape"] in ("readback", "case"): + with concurrent.futures.ThreadPoolExecutor(max_workers=manifest["legacy_readback"]["concurrency"]) as pool: + replies = list(pool.map(lambda _: readback(case), range(4))) + d.require(len({r["command_id"] for r in replies}) == 1, label + ": four callers retain one original identity") + readback(case, run=f["run"]) + for options in [{"name": "other"}, {"run": "00000000000000000000000000"}]: + reply = d.message(kind, f["workflow"], record["request_id"], value=999, success=False, **options) + typed = typed_error(reply, "request_id_conflict") + d.require(typed.get("retained_command_id") == record["receipt"]["command_id"], + label + ": name/run refusal identifies the original command") + d.require(typed.get("conflict") == ["name" if "name" in options else "target"], + label + ": correct conflict component") + unchanged(before, label) + if case["shape"] == "case": + answer = d.message(kind, f["workflow"], manifest["keys"]["case_other"]) + d.require(answer.get("command_id") != record["receipt"]["command_id"] and answer.get("retained") is False, + kind + ": SQLite case-different key admits a distinct new command; writes expected") + else: + reply = d.message(kind, f["workflow"], case["retry_key"], success=False) + typed = typed_error(reply, "request_id_ambiguous") + d.require(typed.get("retained") is True and typed.get("intent_verified") is False + and typed.get("retryable") is False, label + ": ambiguity fails closed and unverified") + d.require(all(typed.get(field) is None for field in ("command_id", "update_id", "retained_command_id")), + label + ": ambiguity selects no receipt") + unchanged(before, label) + + +if __name__ == "__main__": + outcome, error, phase, transform = "runner-blocked", None, "preflight", None + seed_wire = [] + try: + d.require(hashlib.sha256((ROOT / "bin/dw").read_bytes()).hexdigest() == + "da366c79c90978a3c0a92d723c66a96a2d0963f9dccbd41c8207f2c41848f814", "installed CLI binary identity") + observed = json.loads(d.docker("image", "inspect", d.IMAGE, "--format", "{{json .}}")) + d.require(observed["Architecture"] == "amd64" and observed["Os"] == "linux" and + observed["Config"]["Labels"]["org.opencontainers.image.revision"] == + "5614274d2629d03f92932a07dfee0fb10349ddd9", "published Server platform/source identity") + start() + phase = "API seed" + seed() + seed_wire = list(d.PROXY.records) + stop() + phase = "offline accepted transform" + # This is a current-producer fixture. Checkpoint only after observed + # process absence, then freeze the helper's untouched source inode. + with closing(sqlite3.connect(d.DB, timeout=2)) as db: + db.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone() + original = d.DB.with_name("seed.sqlite") + d.DB.rename(original) + transform = legacy.transform_copy(original, d.DB, selections, stopped=True) + d.require(transform["transformed"] == 12, "exact twelve approved legacy transforms; all other schema/rows preserved") + start() + phase = "installed legacy readback" + exercise() + outcome = "pass" + except Exception as exc: + error = (type(exc).__name__ + ": " + str(exc)).replace(d.TOKEN, "[redacted]") + print(json.dumps({"outcome": outcome, "phase": phase, "error": error}), flush=True) + finally: + wire = list(d.PROXY.records) if d.PROXY else [] + if d.PROXY: + d.PROXY.shutdown() + d.PROXY.server_close() + names = d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$") + if names: + try: + d.docker("rm", "-fv", d.NAME) + except RuntimeError: + pass + for _ in range(50): + if not d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$"): + break + time.sleep(.1) + (ROOT / "runtime.env").unlink(missing_ok=True) + absent = not d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$") + report = {"cell": "synthetic-legacy-shape", "outcome": outcome, "error": error, + "phase": phase, "checks": len(d.CHECKS), "started_at": started, + "finished_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "driver_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + "base_driver_sha256": hashlib.sha256((SOURCE / "cli_216.py").read_bytes()).hexdigest(), + "helper_sha256": hashlib.sha256((ROOT / "sqlite_legacy.py").read_bytes()).hexdigest(), + "manifest_sha256": legacy.MANIFEST_SHA256, "server_index": d.IMAGE, + "transform": transform, "windows": windows, + "cleanup": "exact container/volumes/env removed" if absent else "container cleanup failed"} + (ROOT / "legacy-result.json").write_text(json.dumps(report, indent=2)) + (ROOT / "legacy-wire.json").write_text(json.dumps({"seed": seed_wire, "readback": wire}, indent=2)) + print(json.dumps(report), flush=True) + raise SystemExit(0 if outcome == "pass" and absent else 1) From 0fddae2e1cb6c81bcbd68ca8aed3088c854483cc Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 17:30:27 +0000 Subject: [PATCH 2/3] Hold acceptance inputs and clean only created container IDs --- scripts/acceptance/command-keys/README.md | 69 +++-- scripts/acceptance/command-keys/boundaries.py | 120 +++++++++ scripts/acceptance/command-keys/cli_216.py | 167 +++++++----- scripts/acceptance/command-keys/legacy_cli.py | 72 ++++-- .../command-keys/test_boundaries.py | 238 ++++++++++++++++++ 5 files changed, 563 insertions(+), 103 deletions(-) create mode 100644 scripts/acceptance/command-keys/boundaries.py create mode 100644 scripts/acceptance/command-keys/test_boundaries.py diff --git a/scripts/acceptance/command-keys/README.md b/scripts/acceptance/command-keys/README.md index db00802..56b6261 100644 --- a/scripts/acceptance/command-keys/README.md +++ b/scripts/acceptance/command-keys/README.md @@ -7,7 +7,8 @@ adapter at a time after the normal host-capacity/owner check. Docker runtime limits are one CPU, 1 GiB and 256 PIDs, with a loopback-only ephemeral port. The modern adapter covers admission, committed-response loss, explicit retry, -concurrency, Unicode limits, terminal runs and namespace fences. Its second +concurrency, Unicode limits, terminal runs and namespace fences. Its typed phase +covers int64, null, Unicode, reordered maps, integer/float and signed-zero intent. Its second phase restarts the actual Server process on the same database and checks the previously lost receipts and explicitly selected terminal runs. The legacy adapter makes current-producer API admissions, stops that Server, transforms a @@ -34,39 +35,52 @@ Stage these public, reviewed source files in the experiment directory: | `sqlite_legacy.py` | `sdk-rust` commit `4add871b175517f2031f71c868ce278bbfbba486`, `scripts/acceptance/command-keys/sqlite_legacy.py` | `6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad` | | `command-key-legacy-v1-vectors.json` | `server` commit `d66e33bfb5c76f94c5b202f8c9f7a9e8cd75cf1c`, `resources/platform-protocol-specs/command-key-legacy-v1-vectors.json` | `da1990000c611d7377690c7526487d6415fa4f9cb42d299f59585a7c2eebf769` | -The helper reads the adjacent manifest. Both are used only by `legacy_cli.py`; -the proxy is shared by both adapters. Their identities are checked before import. -Do not replace them with a working tree overlay. +The complete-snapshot helper and manifest are used by both adapters. Its +transform is used only by `legacy_cli.py`. The proxy is shared too. Inputs must +be canonical, owned, single-link regular files, with no group/world write bits. +The runner holds their descriptors, executes/imports sealed or in-memory checked +bytes and rechecks identities after use. Do not use working tree overlays or +symlink/hardlink staging. Container operations use the exact ID returned by the +runner's creation call; an existing or reused foreign name grants no authority. -Use a separately verified installed official Python SDK 2.17.1 environment only -for Avro fixture encoding/decoding. Its PyPI wheel SHA256 is +Prepare the official-codec fixture with a separately verified installed Python +SDK 2.17.1. Its PyPI wheel SHA256 is `90040cb28906985b227988815b1b53aa3079378673e35c689392ac71edaf7b35`. -Check installed package files against that wheel and retain module-origin and -dependency evidence before using its Python executable. This is a codec -fixture, not Python Worker qualification. Generate `worker-envelopes.json` in -the private experiment directory with that interpreter: +Compare installed package files against that wheel and retain module-origin and +dependency evidence. This is codec preparation, not Python Worker qualification. +Write `worker-envelopes.json` in the experiment directory using that verified +interpreter and the following code (no trailing newline): ```python +import hashlib +import importlib.metadata import json +from pathlib import Path from durable_workflow import serializer -print(json.dumps({ +Path("worker-envelopes.json").write_text(json.dumps({ "empty": serializer.envelope(None, codec="avro"), "one": serializer.envelope(1, codec="avro"), + "serializer_sha256": hashlib.sha256(Path(serializer.__file__).read_bytes()).hexdigest(), + "sdk_version": importlib.metadata.version("durable-workflow"), + "fastavro_version": importlib.metadata.version("fastavro"), })) ``` -Run the source from this directory with isolated host Python. Supply an -absolute verified codec interpreter path in `CLI_216_CODEC_PYTHON` and the -private directory in `CLI_216_WORKDIR`. Provide a clean environment without -cloud, GitHub, Composer or registry credentials. The adapters create synthetic -runtime authentication privately and remove its environment file on exit. +Use official fastavro 1.13.1 for this frozen fixture. Expected file SHA256 is +`68dac0826e251d0450d458224fa3d4d8714aba8e432df7e241909b8c4b877767`. +The runner holds these exact bytes and compares the actual retained Avro result +with this known fixture. It never executes an external codec program or accepts +its self-reported decoded result. + +Run source with isolated host Python and the private directory in +`CLI_216_WORKDIR`. Provide a clean environment without cloud, GitHub, Composer +or registry credentials. Runtime authentication is synthetic and private. ```bash env -i PATH=/usr/bin:/bin \ CLI_216_WORKDIR="$experiment" \ CLI_216_PROXY_SOURCE="$experiment/committed_response_loss.py" \ - CLI_216_CODEC_PYTHON="$codec_python" \ timeout 180s python3 -I -B scripts/acceptance/command-keys/cli_216.py ``` @@ -75,14 +89,16 @@ For the restart phase, preserve the first phase's `admissions.json` as the same command with `CLI_216_PHASE=selected-restart`. Preserve that phase's records separately too. `CLI_216_PHASE=fresh-selected` adds direct fresh explicit-run literal-key markers (the loss cells intentionally withhold those -initial response bodies). Use a **new** private +initial response bodies). `CLI_216_PHASE=typed` runs the separate typed-intent +cases. Use a **new** private directory for `legacy_cli.py`, with the same staged inputs and environment. The legacy adapter generates its own admissions and stopped-copy database. An outer experiment owner must also enforce the timeout and remove only the -exact owned container if the process is interrupted. Names are +exact created container ID if the process is interrupted. Names are `cli-216-392-modern` and `cli-216-392-legacy`; remove their anonymous volumes -too, and verify absence. No listener or production service is retained. +too, and verify that ID is absent. Never remove a container by a reused name. +No listener or production service is retained. Keep generated databases, response bodies and invocation records private and bounded in retention. Publish sanitized tuple/source hashes, timestamps, @@ -92,3 +108,16 @@ not classify a failed assertion as a product defect: preserve the first result and diagnose runner/source/cleanup failures explicitly. A later corrected run does not erase it. The adapters are manual evidence producers, with no workflow activation, release authority, task scheduler or automatic retry service. + +The offline boundary controls use no Docker, network or credentials: + +```bash +CLI_216_TEST_HELPER="$experiment/sqlite_legacy.py" \ + python3 -I -B scripts/acceptance/command-keys/test_boundaries.py -v +``` + +They cover executable/module substitution, symlink/hardlink and same-byte inode +replacement, a forged codec, sealed-content writes, foreign-name survival and +schema-only mutations that the predecessor's row-only comparison accepted. +Modern and legacy zero-write checks use the same bounded typed logical snapshot, +including catalog SQL, columns, indexes, triggers, views and all table rows. diff --git a/scripts/acceptance/command-keys/boundaries.py b/scripts/acceptance/command-keys/boundaries.py new file mode 100644 index 0000000..146f1e9 --- /dev/null +++ b/scripts/acceptance/command-keys/boundaries.py @@ -0,0 +1,120 @@ +"""Held inputs and exact container ownership for the manual Linux experiment.""" +import fcntl +import hashlib +import json +import os +from pathlib import Path +import re +import stat +import sys +import types + + +class HeldInput: + def __init__(self, path, expected, *, maximum=16 * 1024 * 1024): + self.path = Path(path) + if not self.path.is_absolute() or self.path.resolve() != self.path: + raise ValueError("input must be an absolute path without symlinks") + self.fd = os.open(self.path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + self.sealed_fd = None + try: + self.identity = self._identity(os.fstat(self.fd)) + value = os.fstat(self.fd) + if (not stat.S_ISREG(value.st_mode) or value.st_uid != os.getuid() + or value.st_nlink != 1 or value.st_mode & 0o022 + or not 0 < value.st_size <= maximum): + raise ValueError("input must be an owned single-link non-writable-by-others regular file") + self.data = os.pread(self.fd, maximum + 1, 0) + if len(self.data) != value.st_size or hashlib.sha256(self.data).hexdigest() != expected: + raise ValueError("input content identity differs") + self.digest = expected + self.recheck() + self.sealed_fd = os.memfd_create("cli-216-input", os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING) + with os.fdopen(os.dup(self.sealed_fd), "wb") as output: + output.write(self.data) + os.fchmod(self.sealed_fd, 0o500) + fcntl.fcntl(self.sealed_fd, fcntl.F_ADD_SEALS, + fcntl.F_SEAL_WRITE | fcntl.F_SEAL_GROW | fcntl.F_SEAL_SHRINK | fcntl.F_SEAL_SEAL) + except BaseException: + self.close() + raise + + @staticmethod + def _identity(value): + return (value.st_dev, value.st_ino, value.st_mode, value.st_uid, value.st_gid, + value.st_nlink, value.st_size, value.st_mtime_ns, value.st_ctime_ns) + + def recheck(self): + if (self.path.resolve() != self.path or self._identity(self.path.lstat()) != self.identity + or self._identity(os.fstat(self.fd)) != self.identity): + raise ValueError("held input path or descriptor identity changed") + current = os.pread(self.fd, len(self.data) + 1, 0) + if hashlib.sha256(current).hexdigest() != self.digest: + raise ValueError("held input content changed") + + def module(self, name): + self.recheck() + module = types.ModuleType(name) + module.__file__ = str(self.path) + sys.modules[name] = module + exec(compile(self.data, str(self.path), "exec"), module.__dict__) + self.recheck() + return module + + @property + def executable(self): + return "/proc/self/fd/" + str(self.sealed_fd) + + def close(self): + for name in ("fd", "sealed_fd"): + descriptor = getattr(self, name, None) + if descriptor is not None: + os.close(descriptor) + setattr(self, name, None) + + +def private_directory(path): + path = Path(path) + if not path.is_absolute() or path.resolve() != path: + raise ValueError("private directory must be absolute and have no symlink component") + path.mkdir(mode=0o700, exist_ok=True) + value = path.lstat() + if not stat.S_ISDIR(value.st_mode) or value.st_uid != os.getuid() or stat.S_IMODE(value.st_mode) != 0o700: + raise ValueError("private directory must be owned and mode 0700") + return path + + +class OwnedContainer: + def __init__(self, docker, name): + self.docker, self.name, self.id = docker, name, None + self.created_ids = [] + + def run(self, *arguments): + if self.exists() or self.docker("ps", "-aq", "--filter", "name=^" + self.name + "$"): + raise ValueError("container name already exists; no cleanup authority acquired") + created = self.docker("run", *arguments) + if not re.fullmatch(r"[0-9a-f]{64}", created, re.ASCII): + raise ValueError("Docker did not return one canonical created container ID") + self.id = created + self.created_ids.append(created) + return self.inspect() + + def exists(self): + return self.id is not None and self.docker("ps", "-aq", "--no-trunc", "--filter", "id=" + self.id) == self.id + + def inspect(self): + if not self.exists(): + raise ValueError("created container no longer exists") + value = json.loads(self.docker("inspect", self.id, "--format", "{{json .}}")) + if value["Id"] != self.id or value["Name"] != "/" + self.name: + raise ValueError("created container ID/name changed") + return value + + def stop(self): + self.inspect() + self.docker("stop", "--time=10", self.id) + + def remove(self): + if self.exists(): + self.inspect() + self.docker("rm", "-fv", self.id) diff --git a/scripts/acceptance/command-keys/cli_216.py b/scripts/acceptance/command-keys/cli_216.py index d5647fa..7905d90 100644 --- a/scripts/acceptance/command-keys/cli_216.py +++ b/scripts/acceptance/command-keys/cli_216.py @@ -13,26 +13,51 @@ import time import urllib.error import urllib.request -import importlib.util -PROXY_SOURCE = Path(os.environ['CLI_216_PROXY_SOURCE']).resolve() +SOURCE = Path(__file__).resolve().parent +BOUNDARY_SHA = '664b0765774063c50bb82162be70fa4f91ca35abc3305a38498364cbdd51b3d8' +# Bootstrap the local reviewed boundary module from the same bytes that are +# checked. Subsequent inputs use its held descriptors and sealed memory files. +_boundary_path = SOURCE / 'boundaries.py' +_boundary_fd = os.open(_boundary_path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) +try: + import stat + import types + _identity = os.fstat(_boundary_fd) + if (not stat.S_ISREG(_identity.st_mode) or _identity.st_uid != os.getuid() + or _identity.st_nlink != 1 or _identity.st_mode & 0o022 or _identity.st_size > 65536): + raise ValueError('boundary module owner/type/mode differs') + _boundary_bytes = os.pread(_boundary_fd, 65537, 0) + if hashlib.sha256(_boundary_bytes).hexdigest() != BOUNDARY_SHA: + raise ValueError('boundary module source differs') + boundary = types.ModuleType('cli216_boundaries') + boundary.__file__ = str(_boundary_path) + exec(compile(_boundary_bytes, str(_boundary_path), 'exec'), boundary.__dict__) + if boundary.HeldInput._identity(_boundary_path.lstat()) != boundary.HeldInput._identity(_identity): + raise ValueError('boundary module changed while importing held bytes') +finally: + os.close(_boundary_fd) + +ROOT = boundary.private_directory(Path(os.environ['CLI_216_WORKDIR'])) +PROXY_SOURCE = Path(os.environ['CLI_216_PROXY_SOURCE']) PROXY_SHA256 = 'bd1e8ccbbf76f63fc2eafa278a023eb181d405b68d9c30016392e395a1eff725' -if hashlib.sha256(PROXY_SOURCE.read_bytes()).hexdigest() != PROXY_SHA256: - raise SystemExit('Shared proxy bytes do not match the independently reviewed fixture') -spec = importlib.util.spec_from_file_location('cli216_shared_proxy', PROXY_SOURCE) -shared_proxy = importlib.util.module_from_spec(spec) -spec.loader.exec_module(shared_proxy) -CommittedResponseLoss = shared_proxy.CommittedResponseLoss +HELD = [boundary.HeldInput(_boundary_path, BOUNDARY_SHA)] + +def held(path, digest, maximum=16 * 1024 * 1024): + value = boundary.HeldInput(path, digest, maximum=maximum) + HELD.append(value) + return value -ROOT = Path(os.environ['CLI_216_WORKDIR']) -if not ROOT.is_absolute() or ROOT.resolve() != ROOT or ROOT.is_symlink(): - raise SystemExit('Require an absolute canonical private experiment directory') -owned = ROOT.stat() -if owned.st_uid != os.getuid() or owned.st_mode & 0o777 != 0o700: - raise SystemExit('Experiment directory must be owned by this user and mode 0700') -CODEC_PYTHON = Path(os.environ['CLI_216_CODEC_PYTHON']) -if not CODEC_PYTHON.is_absolute(): - raise SystemExit('Provide the verified installed codec Python as an absolute path') +PROXY_INPUT = held(PROXY_SOURCE, PROXY_SHA256) +shared_proxy = PROXY_INPUT.module('cli216_shared_proxy') +CommittedResponseLoss = shared_proxy.CommittedResponseLoss +HELPER_INPUT = held(ROOT / 'sqlite_legacy.py', '6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad') +legacy_helper = HELPER_INPUT.module('cli216_snapshot_helper') +MANIFEST_INPUT = held(ROOT / 'command-key-legacy-v1-vectors.json', 'da1990000c611d7377690c7526487d6415fa4f9cb42d299f59585a7c2eebf769') +legacy_helper.MANIFEST = Path(MANIFEST_INPUT.executable) +ENVELOPE_INPUT = held(ROOT / 'worker-envelopes.json', '68dac0826e251d0450d458224fa3d4d8714aba8e432df7e241909b8c4b877767') +ENVELOPES = json.loads(ENVELOPE_INPUT.data) +CLI_INPUT = held(ROOT / 'bin/dw', 'da366c79c90978a3c0a92d723c66a96a2d0963f9dccbd41c8207f2c41848f814', 32 * 1024 * 1024) IMAGE = 'ghcr.io/durable-workflow/server@sha256:8ffcdb2a36af3129a1987849aafe8715fe0d3bf9a9648467f0b5cb8eeef43ba9' NAME = 'cli-216-392-modern' TOKEN = secrets.token_hex(32) @@ -45,8 +70,9 @@ 'DW_CONFIG_HOME': str(ROOT / 'config'), 'TMPDIR': str(ROOT / 'tmp'), 'DURABLE_WORKFLOW_AUTH_TOKEN': TOKEN} for directory in ('data', 'home', 'config', 'tmp', 'docker-empty'): - (ROOT / directory).mkdir(exist_ok=True) -(ROOT / 'data').chmod(0o700) + boundary.private_directory(ROOT / directory) +HTTP = urllib.request.build_opener(urllib.request.ProxyHandler({})) + def require(condition, label, detail=None): @@ -70,15 +96,17 @@ def request(method, path, body=None, namespace='default'): r = urllib.request.Request(BASE + path, method=method, headers=headers, data=None if body is None else json.dumps(body).encode()) try: - with urllib.request.urlopen(r, timeout=15) as response: + with HTTP.open(r, timeout=15) as response: return response.status, json.load(response) except urllib.error.HTTPError as e: return e.code, json.load(e) def cli(*args, namespace='default', success=True): + CLI_INPUT.recheck() p = subprocess.run(['dw', *args, '--server=' + PROXY.url, '--namespace=' + namespace, '--json'], - env=ENV, capture_output=True, text=True, timeout=45) + executable=CLI_INPUT.executable, env=ENV, capture_output=True, text=True, timeout=45, pass_fds=(CLI_INPUT.sealed_fd,)) + CLI_INPUT.recheck() try: body = json.loads(p.stdout) except ValueError: @@ -124,11 +152,11 @@ def finish(f, task, commands): return body -def fixture(kind, tag, namespace='default', workflow=None): +def fixture(kind, tag, namespace='default', workflow=None, parameter_type='int'): wf = workflow or 'cli216-' + tag + '-' + secrets.token_hex(4) f = {'workflow': wf, 'worker': wf + '-w', 'queue': wf + '-q', 'type': wf + '-t', 'namespace': namespace} - param = {'name': 'value', 'type': 'int', 'required': True, 'variadic': False, - 'default_available': False, 'allows_null': False} + param = {'name': 'value', 'type': parameter_type, 'required': True, 'variadic': False, + 'default_available': False, 'allows_null': parameter_type == 'mixed'} contracts = {'signals': ['finish'], 'signal_contracts': [{'name': 'finish', 'parameters': [param]}], 'updates': ['set'], 'update_validators': [], 'update_contracts': [{'name': 'set', 'parameters': [param]}]} status, body = request('POST', '/api/worker/register', { @@ -141,7 +169,7 @@ def fixture(kind, tag, namespace='default', workflow=None): raise AssertionError(('register', status, body)) started = cli('workflow:start', '--workflow-id=' + wf, '--type=' + f['type'], '--task-queue=' + f['queue'], '--input=[]', namespace=namespace) f['run'] = started['run_id'] - f['empty_envelope'] = json.loads((ROOT / 'worker-envelopes.json').read_text())['empty'] + f['empty_envelope'] = ENVELOPES['empty'] if kind == 'update': finish(f, poll(f), [{'type': 'open_signal_wait', 'signal_name': 'finish'}]) return f @@ -155,22 +183,17 @@ def history(f): def snapshot(): - with sqlite3.connect('file:' + str(DB) + '?mode=ro', uri=True) as db: - tables = [r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")] - data = {} - for table in tables: - quoted = '"' + table.replace('"', '""') + '"' - rows = db.execute('SELECT * FROM ' + quoted).fetchall() - data[table] = {'count': len(rows), 'sha256': hashlib.sha256(repr(sorted(rows, key=repr)).encode()).hexdigest()} - return data + from contextlib import closing + with closing(sqlite3.connect(DB.as_uri() + '?mode=ro', uri=True, timeout=2)) as db: + return legacy_helper.snapshot(db) -def original(kind, f, key=None, lost=False, run=None): +def original(kind, f, key=None, lost=False, run=None, value=1): path = path_for(kind, f, run=run) before = len(PROXY.records) if lost: PROXY.arm('POST', path) - answer = message(kind, f['workflow'], key, run=run, namespace=f['namespace'], success=not lost) + answer = message(kind, f['workflow'], key, value=value, run=run, namespace=f['namespace'], success=not lost) sent = [r for r in PROXY.records[before:] if r['method'] == 'POST' and r['path'] == path] require(len(sent) == 1, kind + ': exactly one admission POST; no implicit resend') wire = sent[0] @@ -185,7 +208,7 @@ def original(kind, f, key=None, lost=False, run=None): kind + ': complete upstream receipt dropped before downstream status') admitted = snapshot() recovery_start = len(PROXY.records) - receipt = message(kind, f['workflow'], chosen, run=run, namespace=f['namespace']) + receipt = message(kind, f['workflow'], chosen, value=value, run=run, namespace=f['namespace']) recovered = [r for r in PROXY.records[recovery_start:] if r['method'] == 'POST' and r['path'] == path] require(len(recovered) == 1, kind + ': one explicit recovery POST') require(recovered[0]['request_body_sha256'] == wire['request_body_sha256'], @@ -227,15 +250,10 @@ def conflict(kind, f, record, part, **options): require(snapshot() == before, kind + ': conflict changes no database row') -def decoded_result(answer): - envelope = answer.get('result_envelope') - p = subprocess.run([str(CODEC_PYTHON), '-I', '-B', '-c', - 'import json,sys; from durable_workflow import serializer; print(json.dumps(serializer.decode_envelope(json.load(sys.stdin))))'], - input=json.dumps(envelope), capture_output=True, text=True, timeout=10, - env={'PATH': '/usr/bin:/bin', 'HOME': str(ROOT / 'home'), 'TMPDIR': str(ROOT / 'tmp')}) - if p.returncode: - raise AssertionError('official Avro fixture decode failed') - return json.loads(p.stdout) +def result_is_one(answer): + # Compare the actual retained Avro result with the held official-codec + # fixture. No external interpreter can self-certify a decoded value. + return answer.get('result_envelope') == ENVELOPES['one'] def run_modern(): @@ -277,7 +295,7 @@ def run_modern(): record = original(kind, f, key=kind + '-' + terminal, lost=True) commands = [] if kind == 'update': - commands.append({'type': 'complete_update', 'update_id': record['receipt']['update_id'], 'result': json.loads((ROOT / 'worker-envelopes.json').read_text())['one']}) + commands.append({'type': 'complete_update', 'update_id': record['receipt']['update_id'], 'result': ENVELOPES['one']}) commands.append({'type': terminal, 'result': f['empty_envelope']}) terminal_event = 'WorkflowCompleted' if terminal == 'complete_workflow' else 'WorkflowContinuedAsNew' for _ in range(10): @@ -287,7 +305,7 @@ def run_modern(): require(any(e['event_type'] == terminal_event for e in history(f)), kind + ': terminal transition recorded before retry') answer = retained(kind, f, record) if kind == 'update': - require(answer.get('update_status') == 'completed' and decoded_result(answer) == 1, 'update: original completed typed result survives ' + terminal, answer) + require(answer.get('update_status') == 'completed' and result_is_one(answer), 'update: original completed typed result survives ' + terminal, answer) require(any(e['event_type'] == ('WorkflowCompleted' if terminal == 'complete_workflow' else 'WorkflowContinuedAsNew') for e in history(f)), kind + ': original history proves ' + terminal) conflict(kind, f, record, 'arguments', value=2) conflict(kind, f, record, 'target', run='00000000000000000000000000') @@ -332,7 +350,7 @@ def run_selected_and_restart(): commands = [] if kind == 'update': commands.append({'type': 'complete_update', 'update_id': record['receipt']['update_id'], - 'result': json.loads((ROOT / 'worker-envelopes.json').read_text())['one']}) + 'result': ENVELOPES['one']}) commands.append({'type': terminal, 'result': f['empty_envelope']}) terminal_event = 'WorkflowCompleted' if terminal == 'complete_workflow' else 'WorkflowContinuedAsNew' for _ in range(10): @@ -342,7 +360,7 @@ def run_selected_and_restart(): require(any(e['event_type'] == terminal_event for e in history(f)), kind + ': selected-run terminal transition recorded') answer = retained(kind, f, record, run=f['run']) if kind == 'update': - require(answer.get('update_status') == 'completed' and decoded_result(answer) == 1, + require(answer.get('update_status') == 'completed' and result_is_one(answer), 'selected update: original typed result survives ' + terminal) conflict(kind, f, record, 'target') conflict(kind, f, record, 'target', run='00000000000000000000000000') @@ -356,8 +374,31 @@ def run_fresh_selected(): retained(kind, f, record, run=f['run']) +def run_typed(): + vectors = [ + ('int64', 9223372036854775806, '9223372036854775806'), + ('null', None, False), + ('unicode', 'é🙂', 'e\u0301🙂'), + ('map', {'b': 2, 'a': [None, 'é', 9007199254740993]}, + {'b': '2', 'a': [None, 'é', 9007199254740993]}), + ('integer-float', 1, 1.0), + ('signed-zero', 0.0, -0.0), + ] + for kind in ('signal', 'update'): + f = fixture(kind, 'typed-' + kind, parameter_type='mixed') + for label, value, other in vectors: + record = original(kind, f, key='typed-' + label, value=value, lost=True) + equal = {'a': value['a'], 'b': value['b']} if label == 'map' else value + retained(kind, f, record, value=equal) + conflict(kind, f, record, 'arguments', value=other) + event = 'SignalReceived' if kind == 'signal' else 'UpdateAccepted' + require(sum(row['event_type'] == event for row in history(f)) == len(vectors), + kind + ': typed recovery/conflict admits one event per original key') + + if __name__ == '__main__': outcome, error = 'runner-blocked', None + runtime = boundary.OwnedContainer(docker, NAME) try: observed = json.loads(docker('image', 'inspect', IMAGE, '--format', '{{json .}}')) require(observed['Architecture'] == 'amd64' and observed['Os'] == 'linux', 'observed native platform is linux/amd64') @@ -365,12 +406,12 @@ def run_fresh_selected(): envfile = ROOT / 'runtime.env' envfile.write_text('DW_AUTH_TOKEN=' + TOKEN + '\nDW_RUST_MAINTENANCE_LOOP=0\n') envfile.chmod(0o600) - docker('run', '--rm', '-d', '--name', NAME, '--platform=linux/amd64', '--init', '--cpus=1', '--memory=1g', '--pids-limit=256', + runtime.run('--rm', '-d', '--name', NAME, '--platform=linux/amd64', '--init', '--cpus=1', '--memory=1g', '--pids-limit=256', '--read-only', '--cap-drop=ALL', '--security-opt=no-new-privileges', '--env-file', str(envfile), '--mount', 'type=bind,source=' + str(ROOT / 'data') + ',target=/app/database', '--tmpfs', '/app/storage:rw,uid=1000,gid=1000,mode=750,size=64m', '--tmpfs', '/tmp:rw,size=16m', '-p', '127.0.0.1::8080', IMAGE) - binding = json.loads(docker('inspect', NAME, '--format', '{{json .NetworkSettings.Ports}}'))['8080/tcp'][0] + binding = runtime.inspect()['NetworkSettings']['Ports']['8080/tcp'][0] BASE = 'http://127.0.0.1:' + binding['HostPort'] for _ in range(100): try: @@ -384,6 +425,8 @@ def run_fresh_selected(): threading.Thread(target=PROXY.serve_forever, daemon=True).start() if os.environ.get('CLI_216_PHASE') == 'selected-restart': run_selected_and_restart() + elif os.environ.get('CLI_216_PHASE') == 'typed': + run_typed() elif os.environ.get('CLI_216_PHASE') == 'fresh-selected': run_fresh_selected() else: @@ -397,15 +440,25 @@ def run_fresh_selected(): PROXY.shutdown() PROXY.server_close() try: - docker('rm', '-fv', NAME) - cleanup = 'container and anonymous volumes removed' - except Exception: - cleanup = 'container removal needs investigation' + runtime.remove() + for _ in range(50): + if not runtime.exists(): + break + time.sleep(.1) + if runtime.exists(): + raise RuntimeError('created container removal did not complete') + for item in HELD: + item.recheck() + cleanup = 'created container and anonymous volumes removed; held inputs unchanged' + except Exception as failure: + cleanup = 'cleanup or input identity requires investigation' + outcome = 'runner-blocked' + error = (type(failure).__name__ + ': ' + str(failure)).replace(TOKEN, '[redacted]') (ROOT / 'runtime.env').unlink(missing_ok=True) report = {'started_at': STARTED, 'finished_at': datetime.datetime.now(datetime.timezone.utc).isoformat(), - 'outcome': outcome, 'checks': CHECKS, 'error': error, 'cleanup': cleanup, + 'outcome': outcome, 'created_container_ids': runtime.created_ids, 'checks': CHECKS, 'error': error, 'cleanup': cleanup, 'runner_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), - 'proxy_sha256': hashlib.sha256(PROXY_SOURCE.read_bytes()).hexdigest(), 'server_index': IMAGE, 'server_source': '5614274d2629d03f92932a07dfee0fb10349ddd9', 'phase': os.environ.get('CLI_216_PHASE', 'modern')} + 'proxy_sha256': PROXY_INPUT.digest, 'server_index': IMAGE, 'server_source': '5614274d2629d03f92932a07dfee0fb10349ddd9', 'phase': os.environ.get('CLI_216_PHASE', 'modern')} (ROOT / 'result.json').write_text(json.dumps(report, indent=2)) (ROOT / 'invocations.json').write_text(json.dumps(INVOCATIONS, indent=2).replace(TOKEN, '[redacted]')) (ROOT / 'admissions.json').write_text(json.dumps(ADMISSIONS, indent=2)) diff --git a/scripts/acceptance/command-keys/legacy_cli.py b/scripts/acceptance/command-keys/legacy_cli.py index 905c436..275d6cb 100644 --- a/scripts/acceptance/command-keys/legacy_cli.py +++ b/scripts/acceptance/command-keys/legacy_cli.py @@ -3,7 +3,9 @@ from contextlib import closing import datetime import hashlib -import importlib.util +import stat +import types +import sys import json import os from pathlib import Path @@ -15,19 +17,33 @@ ROOT = Path(os.environ["CLI_216_WORKDIR"]) -def bound_module(name, filename, expected): - path = SOURCE / filename if filename == "cli_216.py" else ROOT / filename - if hashlib.sha256(path.read_bytes()).hexdigest() != expected: - raise ValueError("fixture source identity differs: " + filename) - spec = importlib.util.spec_from_file_location(name, path) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module +def bound_driver(): + path = SOURCE / "cli_216.py" + descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) + try: + observed = os.fstat(descriptor) + if (not stat.S_ISREG(observed.st_mode) or observed.st_uid != os.getuid() + or observed.st_nlink != 1 or observed.st_mode & 0o022 or observed.st_size > 131072): + raise ValueError("base driver owner/type/mode differs") + data = os.pread(descriptor, 131073, 0) + if hashlib.sha256(data).hexdigest() != BASE_SHA: + raise ValueError("base driver source identity differs") + module = types.ModuleType("cli216_owner") + module.__file__ = str(path) + exec(compile(data, str(path), "exec"), module.__dict__) + item = module.held(path, BASE_SHA) + if item.identity != item._identity(observed): + raise ValueError("base driver changed during held import") + return module + finally: + os.close(descriptor) -d = bound_module("cli216_owner", "cli_216.py", "d3833b37e96dab863c8652bfe65c09ca5f632dcdee985637b443c4f66a56898e") -legacy = bound_module("cli216_legacy", "sqlite_legacy.py", "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad") +BASE_SHA = "0f2afebcfeea0a728a203a59025bf0a94669581ccd0d84bf3f53abc2d971c653" +d = bound_driver() +legacy = d.legacy_helper d.NAME = "cli-216-392-legacy" +runtime = d.boundary.OwnedContainer(d.docker, d.NAME) manifest = legacy.manifest() cases, selections, windows = [], [], [] started = datetime.datetime.now(datetime.timezone.utc).isoformat() @@ -42,13 +58,13 @@ def start(): envfile = ROOT / "runtime.env" envfile.write_text("DW_AUTH_TOKEN=" + d.TOKEN + "\nDW_RUST_MAINTENANCE_LOOP=0\n") envfile.chmod(0o600) - d.docker("run", "--rm", "-d", "--name", d.NAME, "--platform=linux/amd64", "--init", + runtime.run("--rm", "-d", "--name", d.NAME, "--platform=linux/amd64", "--init", "--cpus=1", "--memory=1g", "--pids-limit=256", "--read-only", "--cap-drop=ALL", "--security-opt=no-new-privileges", "--env-file", str(envfile), "--mount", "type=bind,source=" + str(ROOT / "data") + ",target=/app/database", "--tmpfs", "/app/storage:rw,uid=1000,gid=1000,mode=750,size=64m", "--tmpfs", "/tmp:rw,size=16m", "-p", "127.0.0.1::8080", d.IMAGE) - binding = json.loads(d.docker("inspect", d.NAME, "--format", "{{json .NetworkSettings.Ports}}"))["8080/tcp"][0] + binding = runtime.inspect()["NetworkSettings"]["Ports"]["8080/tcp"][0] d.BASE = "http://127.0.0.1:" + binding["HostPort"] for _ in range(100): try: @@ -71,9 +87,9 @@ def stop(): d.PROXY.shutdown() d.PROXY.server_close() d.PROXY = None - d.docker("stop", "--time=10", d.NAME) + runtime.stop() for _ in range(50): - if not d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$"): + if not runtime.exists(): break time.sleep(.1) else: @@ -224,27 +240,31 @@ def exercise(): if d.PROXY: d.PROXY.shutdown() d.PROXY.server_close() - names = d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$") - if names: - try: - d.docker("rm", "-fv", d.NAME) - except RuntimeError: - pass + cleanup_error = None + try: + runtime.remove() for _ in range(50): - if not d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$"): + if not runtime.exists(): break time.sleep(.1) + if runtime.exists(): + raise RuntimeError("created container removal did not complete") + for item in d.HELD: + item.recheck() + except Exception as failure: + cleanup_error = (type(failure).__name__ + ": " + str(failure)).replace(d.TOKEN, "[redacted]") + outcome = "runner-blocked" (ROOT / "runtime.env").unlink(missing_ok=True) - absent = not d.docker("ps", "-aq", "--filter", "name=^" + d.NAME + "$") - report = {"cell": "synthetic-legacy-shape", "outcome": outcome, "error": error, - "phase": phase, "checks": len(d.CHECKS), "started_at": started, + absent = not runtime.exists() + report = {"cell": "synthetic-legacy-shape", "outcome": outcome, "created_container_ids": runtime.created_ids, "error": error, + "cleanup_error": cleanup_error, "phase": phase, "checks": len(d.CHECKS), "started_at": started, "finished_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "driver_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), "base_driver_sha256": hashlib.sha256((SOURCE / "cli_216.py").read_bytes()).hexdigest(), "helper_sha256": hashlib.sha256((ROOT / "sqlite_legacy.py").read_bytes()).hexdigest(), "manifest_sha256": legacy.MANIFEST_SHA256, "server_index": d.IMAGE, "transform": transform, "windows": windows, - "cleanup": "exact container/volumes/env removed" if absent else "container cleanup failed"} + "cleanup": "created container/volumes/env removed; held inputs unchanged" if absent and cleanup_error is None else "cleanup/input identity failed"} (ROOT / "legacy-result.json").write_text(json.dumps(report, indent=2)) (ROOT / "legacy-wire.json").write_text(json.dumps({"seed": seed_wire, "readback": wire}, indent=2)) print(json.dumps(report), flush=True) diff --git a/scripts/acceptance/command-keys/test_boundaries.py b/scripts/acceptance/command-keys/test_boundaries.py new file mode 100644 index 0000000..5642414 --- /dev/null +++ b/scripts/acceptance/command-keys/test_boundaries.py @@ -0,0 +1,238 @@ +"""Offline counterexamples for the manual acceptance runner's trust boundary.""" +import ast +from contextlib import closing +import hashlib +import importlib.util +import json +import os +from pathlib import Path +import sqlite3 +import subprocess +import sys +import tempfile +import unittest +from unittest.mock import patch + +SOURCE = Path(__file__).resolve().parent +spec = importlib.util.spec_from_file_location("tested_boundary", SOURCE / "boundaries.py") +boundary = importlib.util.module_from_spec(spec) +spec.loader.exec_module(boundary) + + +class InputTests(unittest.TestCase): + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.root = Path(self.temporary.name).resolve() + self.addCleanup(self.temporary.cleanup) + + def file(self, value=b"original"): + path = self.root / "input" + path.write_bytes(value) + path.chmod(0o600) + return path + + def hold(self, path): + result = boundary.HeldInput(path, hashlib.sha256(path.read_bytes()).hexdigest()) + self.addCleanup(result.close) + return result + + def test_symlink_refused(self): + original = self.file() + link = self.root / "link" + link.symlink_to(original) + with self.assertRaises(ValueError): + self.hold(link) + + def test_hardlink_refused(self): + original = self.file() + os.link(original, self.root / "second") + with self.assertRaises(ValueError): + self.hold(original) + + def test_same_bytes_new_inode_refused(self): + path = self.file() + held = self.hold(path) + path.rename(self.root / "old") + self.file() + with self.assertRaises(ValueError): + held.recheck() + + def test_inplace_change_refused(self): + path = self.file() + held = self.hold(path) + path.write_bytes(b"modified") + with self.assertRaises(ValueError): + held.recheck() + + def test_sealed_content_cannot_be_changed(self): + held = self.hold(self.file()) + with self.assertRaises(OSError): + os.pwrite(held.sealed_fd, b"evil", 0) + + def test_executable_path_swap_never_executes_replacement(self): + path = self.file(Path(sys.executable).resolve().read_bytes()) + path.chmod(0o700) + held = self.hold(path) + path.rename(self.root / "verified") + marker = self.root / "replacement-executed" + replacement = self.file(("#!/bin/sh\ntouch " + str(marker) + "\nexit 9\n").encode()) + replacement.chmod(0o700) + # The predecessor re-resolved the pathname and executes the marker. + self.assertEqual(subprocess.run([str(path)], check=False).returncode, 9) + self.assertTrue(marker.exists()) + marker.unlink() + self.assertEqual(subprocess.run(["python3", "-I", "-S", "-c", "raise SystemExit(0)"], + executable=held.executable, pass_fds=(held.sealed_fd,), check=False).returncode, 0) + self.assertFalse(marker.exists()) + with self.assertRaises(ValueError): + held.recheck() + + def test_module_swap_after_check_uses_held_bytes_and_refuses_result(self): + path = self.file(b"value = 'reviewed'\n") + held = self.hold(path) + marker = self.root / "unreviewed-executed" + original_check = held.recheck + + def replace_after_check(): + original_check() + path.write_text("from pathlib import Path\nPath(" + repr(str(marker)) + ").touch()\n") + + count = 0 + + def guarded_check(): + nonlocal count + count += 1 + (replace_after_check if count == 1 else original_check)() + + with patch.object(held, "recheck", side_effect=guarded_check): + with self.assertRaises(ValueError): + held.module("held_race_module") + self.assertFalse(marker.exists()) + self.assertEqual(sys.modules["held_race_module"].value, "reviewed") + + def test_forged_codec_cannot_certify_a_result(self): + marker = self.root / "codec-executed" + forged = self.root / "codec" + forged.write_text("#!/bin/sh\ntouch " + str(marker) + "\nprintf '1\\n'\n") + forged.chmod(0o700) + # The predecessor accepted any absolute codec program as authority. + self.assertEqual(subprocess.check_output([str(forged)], text=True).strip(), "1") + marker.unlink() + tree = ast.parse((SOURCE / "cli_216.py").read_text()) + function = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == "result_is_one") + scope = {"ENVELOPES": {"one": {"codec": "avro", "blob": "reviewed"}}} + exec(compile(ast.Module(body=[function], type_ignores=[]), "result-check", "exec"), scope) + with patch.dict(os.environ, {"CLI_216_CODEC_PYTHON": str(forged)}): + self.assertFalse(scope["result_is_one"]({"result_envelope": {"codec": "avro", "blob": "forged"}})) + self.assertFalse(marker.exists()) + + +class FakeDocker: + def __init__(self): + self.objects, self.calls, self.sequence = {}, [], 0 + + def __call__(self, *args): + self.calls.append(args) + if args[0] == "ps": + expression = args[-1] + if expression.startswith("name=^"): + return next((key for key, name in self.objects.items() if expression == "name=^" + name + "$"), "") + value = expression.removeprefix("id=") + return value if value in self.objects else "" + if args[0] == "run": + value = hex(10 + self.sequence)[2:] * 64 + self.sequence += 1 + self.objects[value] = "cli-test" + return value + if args[0] == "inspect": + return json.dumps({"Id": args[1], "Name": "/" + self.objects[args[1]]}) + if args[0] == "rm": + target = args[-1] + if target not in self.objects: + target = next(key for key, name in self.objects.items() if name == target) + del self.objects[target] + return target + raise AssertionError(args) + + +class OwnershipTests(unittest.TestCase): + def test_preexisting_name_survives_failed_launch_cleanup(self): + predecessor = FakeDocker() + predecessor.objects["b" * 64] = "cli-test" + predecessor("rm", "-fv", "cli-test") + self.assertFalse(predecessor.objects) + docker = FakeDocker() + docker.objects["b" * 64] = "cli-test" + owned = boundary.OwnedContainer(docker, "cli-test") + with self.assertRaises(ValueError): + owned.run("--name", "cli-test") + owned.remove() + self.assertIn("b" * 64, docker.objects) + self.assertFalse(any(call[0] == "rm" for call in docker.calls)) + + def test_reused_foreign_name_survives_cleanup(self): + docker = FakeDocker() + owned = boundary.OwnedContainer(docker, "cli-test") + owned.run("--name", "cli-test") + docker.objects.clear() + docker.objects["b" * 64] = "cli-test" + owned.remove() + self.assertIn("b" * 64, docker.objects) + self.assertFalse(any(call[0] == "rm" for call in docker.calls)) + + def test_cleanup_uses_only_created_id(self): + docker = FakeDocker() + owned = boundary.OwnedContainer(docker, "cli-test") + owned.run("--name", "cli-test") + owned.remove() + self.assertIn(("rm", "-fv", "a" * 64), docker.calls) + + + def test_stopped_owned_id_can_start_a_new_process(self): + docker = FakeDocker() + owned = boundary.OwnedContainer(docker, "cli-test") + owned.run("--name", "cli-test") + docker.objects.clear() + owned.run("--name", "cli-test") + self.assertEqual(owned.created_ids, ["a" * 64, "b" * 64]) + owned.remove() + self.assertIn(("rm", "-fv", "b" * 64), docker.calls) + + +class SnapshotTests(unittest.TestCase): + def test_schema_only_changes_do_not_pass_zero_write_comparison(self): + path = Path(os.environ["CLI_216_TEST_HELPER"]) + held = boundary.HeldInput(path, "6fc6054ce087c136a86d7b6b222daeefaa87c58fa0b613b155a5df39e6a26fad") + self.addCleanup(held.close) + helper = held.module("tested_snapshot_helper") + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + database = Path(temporary.name).resolve() / "snapshot.sqlite" + tree = ast.parse((SOURCE / "cli_216.py").read_text()) + function = next(n for n in tree.body if isinstance(n, ast.FunctionDef) and n.name == "snapshot") + scope = {"DB": database, "sqlite3": sqlite3, "legacy_helper": helper} + exec(compile(ast.Module(body=[function], type_ignores=[]), "snapshot-check", "exec"), scope) + with closing(sqlite3.connect(database)) as db: + db.execute("CREATE TABLE fixture(value INTEGER)") + db.execute("INSERT INTO fixture VALUES (1)") + + def predecessor(): + # aa11347 snapshot compared only table names and row hashes. + return {name: db.execute('SELECT * FROM "' + name + '"').fetchall() + for name, in db.execute("SELECT name FROM sqlite_master WHERE type='table'")} + + db.commit() + old = predecessor() + current = scope["snapshot"]() + for statement in ["CREATE INDEX fixture_index ON fixture(value)", + "CREATE VIEW fixture_view AS SELECT value FROM fixture", + "CREATE TRIGGER fixture_trigger AFTER INSERT ON fixture BEGIN SELECT 1; END", + "ALTER TABLE fixture RENAME COLUMN value TO renamed"]: + db.execute(statement) + db.commit() + self.assertEqual(old, predecessor()) + self.assertNotEqual(current, scope["snapshot"]()) + + +if __name__ == "__main__": + unittest.main() From 9147be7e8f36ac748377c8c20666d7e472afb4b2 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Fri, 13 Nov 2026 17:34:50 +0000 Subject: [PATCH 3/3] Capture entrypoint identity before acceptance runs --- scripts/acceptance/command-keys/boundaries.py | 7 +++++-- scripts/acceptance/command-keys/cli_216.py | 5 +++-- scripts/acceptance/command-keys/legacy_cli.py | 9 +++++---- scripts/acceptance/command-keys/test_boundaries.py | 13 +++++++++++++ 4 files changed, 26 insertions(+), 8 deletions(-) diff --git a/scripts/acceptance/command-keys/boundaries.py b/scripts/acceptance/command-keys/boundaries.py index 146f1e9..2680401 100644 --- a/scripts/acceptance/command-keys/boundaries.py +++ b/scripts/acceptance/command-keys/boundaries.py @@ -25,9 +25,12 @@ def __init__(self, path, expected, *, maximum=16 * 1024 * 1024): or not 0 < value.st_size <= maximum): raise ValueError("input must be an owned single-link non-writable-by-others regular file") self.data = os.pread(self.fd, maximum + 1, 0) - if len(self.data) != value.st_size or hashlib.sha256(self.data).hexdigest() != expected: + actual = hashlib.sha256(self.data).hexdigest() + if len(self.data) != value.st_size or (expected is not None and actual != expected): raise ValueError("input content identity differs") - self.digest = expected + # None is used only to capture an already-running entrypoint. + # Every external executable/module/fixture has a literal digest. + self.digest = actual self.recheck() self.sealed_fd = os.memfd_create("cli-216-input", os.MFD_CLOEXEC | os.MFD_ALLOW_SEALING) with os.fdopen(os.dup(self.sealed_fd), "wb") as output: diff --git a/scripts/acceptance/command-keys/cli_216.py b/scripts/acceptance/command-keys/cli_216.py index 7905d90..9e1b052 100644 --- a/scripts/acceptance/command-keys/cli_216.py +++ b/scripts/acceptance/command-keys/cli_216.py @@ -15,7 +15,7 @@ import urllib.request SOURCE = Path(__file__).resolve().parent -BOUNDARY_SHA = '664b0765774063c50bb82162be70fa4f91ca35abc3305a38498364cbdd51b3d8' +BOUNDARY_SHA = 'd377e9191ebf79dc990db159629c303c10531a2f27d3ae20933fabc87e3b3525' # Bootstrap the local reviewed boundary module from the same bytes that are # checked. Subsequent inputs use its held descriptors and sealed memory files. _boundary_path = SOURCE / 'boundaries.py' @@ -48,6 +48,7 @@ def held(path, digest, maximum=16 * 1024 * 1024): HELD.append(value) return value +ENTRY_INPUT = held(Path(__file__).absolute(), None, 131072) PROXY_INPUT = held(PROXY_SOURCE, PROXY_SHA256) shared_proxy = PROXY_INPUT.module('cli216_shared_proxy') CommittedResponseLoss = shared_proxy.CommittedResponseLoss @@ -457,7 +458,7 @@ def run_typed(): (ROOT / 'runtime.env').unlink(missing_ok=True) report = {'started_at': STARTED, 'finished_at': datetime.datetime.now(datetime.timezone.utc).isoformat(), 'outcome': outcome, 'created_container_ids': runtime.created_ids, 'checks': CHECKS, 'error': error, 'cleanup': cleanup, - 'runner_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), + 'runner_sha256': ENTRY_INPUT.digest, 'proxy_sha256': PROXY_INPUT.digest, 'server_index': IMAGE, 'server_source': '5614274d2629d03f92932a07dfee0fb10349ddd9', 'phase': os.environ.get('CLI_216_PHASE', 'modern')} (ROOT / 'result.json').write_text(json.dumps(report, indent=2)) (ROOT / 'invocations.json').write_text(json.dumps(INVOCATIONS, indent=2).replace(TOKEN, '[redacted]')) diff --git a/scripts/acceptance/command-keys/legacy_cli.py b/scripts/acceptance/command-keys/legacy_cli.py index 275d6cb..d568929 100644 --- a/scripts/acceptance/command-keys/legacy_cli.py +++ b/scripts/acceptance/command-keys/legacy_cli.py @@ -39,8 +39,9 @@ def bound_driver(): os.close(descriptor) -BASE_SHA = "0f2afebcfeea0a728a203a59025bf0a94669581ccd0d84bf3f53abc2d971c653" +BASE_SHA = "c6b7897a6b5c5208a236d296438fc56fa97d4b15ef28329ac719fade852dc562" d = bound_driver() +ENTRY_INPUT = d.held(Path(__file__).absolute(), None, 131072) legacy = d.legacy_helper d.NAME = "cli-216-392-legacy" runtime = d.boundary.OwnedContainer(d.docker, d.NAME) @@ -259,9 +260,9 @@ def exercise(): report = {"cell": "synthetic-legacy-shape", "outcome": outcome, "created_container_ids": runtime.created_ids, "error": error, "cleanup_error": cleanup_error, "phase": phase, "checks": len(d.CHECKS), "started_at": started, "finished_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), - "driver_sha256": hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), - "base_driver_sha256": hashlib.sha256((SOURCE / "cli_216.py").read_bytes()).hexdigest(), - "helper_sha256": hashlib.sha256((ROOT / "sqlite_legacy.py").read_bytes()).hexdigest(), + "driver_sha256": ENTRY_INPUT.digest, + "base_driver_sha256": d.ENTRY_INPUT.digest, + "helper_sha256": d.HELPER_INPUT.digest, "manifest_sha256": legacy.MANIFEST_SHA256, "server_index": d.IMAGE, "transform": transform, "windows": windows, "cleanup": "created container/volumes/env removed; held inputs unchanged" if absent and cleanup_error is None else "cleanup/input identity failed"} diff --git a/scripts/acceptance/command-keys/test_boundaries.py b/scripts/acceptance/command-keys/test_boundaries.py index 5642414..d84ae68 100644 --- a/scripts/acceptance/command-keys/test_boundaries.py +++ b/scripts/acceptance/command-keys/test_boundaries.py @@ -57,6 +57,19 @@ def test_same_bytes_new_inode_refused(self): with self.assertRaises(ValueError): held.recheck() + def test_entrypoint_replacement_cannot_relabel_terminal_digest(self): + path = self.file(b"reviewed entrypoint") + captured = boundary.HeldInput(path, None) + self.addCleanup(captured.close) + expected = hashlib.sha256(path.read_bytes()).hexdigest() + path.rename(self.root / "loaded") + self.file(b"replacement source") + # The predecessor's terminal pathname hash relabeled the run. + self.assertNotEqual(hashlib.sha256(path.read_bytes()).hexdigest(), expected) + self.assertEqual(captured.digest, expected) + with self.assertRaises(ValueError): + captured.recheck() + def test_inplace_change_refused(self): path = self.file() held = self.hold(path)