From 88edbde14fda818d083cc9a88ee6ec77231e006a Mon Sep 17 00:00:00 2001 From: Steve Brown Date: Thu, 15 Oct 2026 05:52:32 -0400 Subject: [PATCH] feat(receiver): bridge signed phases through the fixed Shuttle CLI --- docs/demos/shuttle-process.log | 8 ++ docs/demos/shuttle-process.timing | 2 + docs/scheduled-receiver.md | 36 +++++++++ pyproject.toml | 2 +- scripts/demo-shuttle-process.py | 46 +++++++++++ shuttle_process_port.py | 95 ++++++++++++++++++++++ test_shuttle_process_port.py | 126 ++++++++++++++++++++++++++++++ 7 files changed, 314 insertions(+), 1 deletion(-) create mode 100644 docs/demos/shuttle-process.log create mode 100644 docs/demos/shuttle-process.timing create mode 100644 scripts/demo-shuttle-process.py create mode 100644 shuttle_process_port.py create mode 100644 test_shuttle_process_port.py diff --git a/docs/demos/shuttle-process.log b/docs/demos/shuttle-process.log new file mode 100644 index 0000000..fbe56dd --- /dev/null +++ b/docs/demos/shuttle-process.log @@ -0,0 +1,8 @@ +Script started on 2026-10-15 05:52:16-04:00 [COMMAND="stty -onlcr; python scripts/demo-shuttle-process.py" ] +Actual subprocess and durable slot ledger; CLI receipts are a fixture. +Held result: skipped-by-hold +Delivery attempts: 2; execution callbacks: 1 +Repeat result identical: True +Changed pinned config: REFUSED before CLI invocation + +Script done on 2026-10-15 05:52:16-04:00 [COMMAND_EXIT_CODE="0"] diff --git a/docs/demos/shuttle-process.timing b/docs/demos/shuttle-process.timing new file mode 100644 index 0000000..1725bd9 --- /dev/null +++ b/docs/demos/shuttle-process.timing @@ -0,0 +1,2 @@ +0.244914 175 +0.000139 53 diff --git a/docs/scheduled-receiver.md b/docs/scheduled-receiver.md index 319eb3c..20e1dda 100644 --- a/docs/scheduled-receiver.md +++ b/docs/scheduled-receiver.md @@ -51,3 +51,39 @@ publication the second time. The small recording is retained with this source proposal, replayed using `scriptreplay --log-out scheduled-receiver.log --log-timing scheduled-receiver.timing`. It requires downloading both files; no inline video playback is claimed. + +## Separate Shuttle environment + +`shuttle_process_port.ShuttleProcessPort` implements the same four-phase port +through a fixed absolute Shuttle CLI path. Supply an absolute local config path, +its SHA256, the admitted `ExecutionScope` and UTC slot. Each call uses +`--require-existing-keys scheduled `, JSON on stdin and a bounded timeout; +a delivery cannot choose executable, arguments, config, graph query or key. +The config must be a regular file owned by the current Unix user, at most 64KiB, +and not writable by group or others. The final component must not be a symlink. +The hashed opened descriptor is passed to Shuttle through `/dev/fd`, so replacing +the config pathname does not substitute different bytes. The executable and +parent directories require operator custody; this is not isolation from another +process running as the same user or an executable hash admission mechanism. + +The CLI must return its exact verified phase receipt. Nonzero exit, timeout, +oversized output, an empty success, wrong phase or malformed receipt is UNKNOWN +and cannot authorize execution. The receiver's existing claim/replay behavior +still applies: failure after claim does not rerun the child. A successful +`publish` means the Shuttle SDK accepted its exact registered-signature read-back, +not merely that this process returned zero. + +This library does not launch a background worker or make a long child safe to +execute in a synchronous command sink. An asynchronous domain execution handoff +and installed child-survival acceptance are separate runtime prerequisites. + +Run `python -m pytest -q test_shuttle_process_port.py` for real process and file +descriptor controls. Its CLI is explicitly a fixture that records packets and +returns phase receipts; it does not prove real keys, registration, graph export, +macOS acceptance or production scheduling. Those require the installed Shuttle +SDK and the scoped runtime smoke. + +Process bridge demo: `python scripts/demo-shuttle-process.py` shows two deliveries, +one execution callback and config drift refusal using actual subprocesses and +a durable ledger. [Recording](demos/shuttle-process.log) and +[timing](demos/shuttle-process.timing) retain that fixture-only limitation. diff --git a/pyproject.toml b/pyproject.toml index 4d639aa..c114451 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ chaski-once = "once:main" py-modules = [ "alertmanager_sink", "change_stream", "chaski", "command_sink", "consumer_lease", "demo", "emitter", "external_event", "failure_family_adapter", - "graph_sink", "incremental", "once", "scheduled_receiver", "slot_receiver", + "graph_sink", "incremental", "once", "scheduled_receiver", "shuttle_process_port", "slot_receiver", ] [tool.pytest.ini_options] diff --git a/scripts/demo-shuttle-process.py b/scripts/demo-shuttle-process.py new file mode 100644 index 0000000..61e1ebc --- /dev/null +++ b/scripts/demo-shuttle-process.py @@ -0,0 +1,46 @@ +"""Real subprocess bridge and slot ledger, fixture CLI receipts, no real keys/graph.""" +import hashlib +from pathlib import Path +import sys +import tempfile + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +from scheduled_receiver import ExecutionScope, receive +from shuttle_process_port import PortRefused, ShuttleProcessPort +from slot_receiver import SlotLedger + +with tempfile.TemporaryDirectory() as root: + root = Path(root) + config, cli = root/'config.json', root/'fixture-cli' + config.write_text('{}') + config.chmod(0o600) + cli.write_text('#!'+sys.executable+'\n'+'''import json,sys +from pathlib import Path +assert sys.argv[1:3] == ['--require-existing-keys', 'scheduled'] +assert Path(sys.argv[5]).read_text() == '{}' +json.load(sys.stdin) +print(json.dumps({'action': sys.argv[3], 'result': 'verified'})) +''') + cli.chmod(0o700) + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + port = ShuttleProcessPort(cli, config, hashlib.sha256(config.read_bytes()).hexdigest(), scope, 600) + calls = [] + def execute(): + calls.append(1) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + packet = {**scope.__dict__, 'scheduled_at': 600} + ledger = SlotLedger(root/'slots.sqlite') + first = receive(ledger, scope, 600, packet, port, execute) + second = receive(ledger, scope, 600, packet, port, execute) + print('Actual subprocess and durable slot ledger; CLI receipts are a fixture.') + print('Held result:', first['outcome']) + print('Delivery attempts: 2; execution callbacks:', len(calls)) + print('Repeat result identical:', first == second) + config.write_text('{"changed": true}') + try: + port.admit(scope, 600) + except PortRefused: + print('Changed pinned config: REFUSED before CLI invocation') + else: + raise AssertionError('changed config accepted') diff --git a/shuttle_process_port.py b/shuttle_process_port.py new file mode 100644 index 0000000..48b6ce2 --- /dev/null +++ b/shuttle_process_port.py @@ -0,0 +1,95 @@ +"""Fixed local Shuttle CLI bridge; no scheduler, launcher or child executor. + +The operator pins the executable and config digest. Each call passes the same +opened config descriptor to Shuttle, so replacement of the pathname cannot +change the admitted configuration. Parent-directory custody remains local. +""" +from __future__ import annotations + +from dataclasses import asdict +import hashlib +import json +import os +from pathlib import Path +import stat +import subprocess +import tempfile + +from scheduled_receiver import ExecutionScope, run_id + + +class PortRefused(RuntimeError): + """No verified phase receipt; never authorizes child execution or replay.""" + + +class ShuttleProcessPort: + def __init__(self, executable: Path, config: Path, config_sha256: str, + scope: ExecutionScope, scheduled_at: int, *, timeout_s: float = 30): + scope.validate() + self.run = run_id(scope, scheduled_at) + if not Path(executable).is_absolute() or not Path(config).is_absolute(): + raise ValueError("absolute operator-owned executable and config required") + if (type(config_sha256) is not str or len(config_sha256) != 64 or + any(c not in '0123456789abcdef' for c in config_sha256)): + raise ValueError("pinned config SHA256 required") + if type(timeout_s) not in (int, float) or not 0 < timeout_s <= 60: + raise ValueError("bounded phase timeout required") + self.executable, self.config = str(executable), Path(config) + self.digest, self.scope, self.slot = config_sha256, scope, scheduled_at + self.timeout_s = timeout_s + + def _call(self, action, scope, run=None, outcome=None): + if type(scope) is not ExecutionScope or scope != self.scope: + raise PortRefused("scope differs from pinned delivery") + if action != 'admit' and run != self.run: + raise PortRefused("run differs from pinned slot") + packet = {'scope': asdict(scope), 'scheduled_at': self.slot} + if action != 'admit': + packet['run'] = run + if outcome is not None: + packet['outcome'] = outcome + try: + fd = os.open(self.config, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + with os.fdopen(fd, 'rb') as config: + info = os.fstat(config.fileno()) + if (not stat.S_ISREG(info.st_mode) or info.st_size > 65536 or + info.st_uid != os.getuid() or info.st_mode & 0o022): + raise PortRefused("config must be bounded, owned and not writable by others") + raw = config.read(65537) + if len(raw) > 65536 or hashlib.sha256(raw).hexdigest() != self.digest: + raise PortRefused("config differs from pinned bytes") + config.seek(0) + # Temporary files bound memory use even if a broken SDK emits + # more output than its small receipt contract permits. + with tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err: + result = subprocess.run( + [self.executable, '--require-existing-keys', 'scheduled', action, + '--config', f'/dev/fd/{config.fileno()}'], + input=json.dumps(packet, allow_nan=False).encode(), + stdout=out, stderr=err, pass_fds=(config.fileno(),), + timeout=self.timeout_s, check=False) + if result.returncode or out.tell() > 4096 or err.tell() > 4096: + raise PortRefused("Shuttle phase failed or exceeded receipt bound") + out.seek(0) + receipt = out.read(4097) + expected = {'action': action, 'result': 'verified'} + # Exact canonical receipt rejects duplicate keys, trailing output, + # wrong action, and a generic exit-zero without verification. + if receipt.strip() != json.dumps(expected).encode(): + raise PortRefused("Shuttle did not return the exact phase receipt") + except (OSError, subprocess.TimeoutExpired, ValueError) as exc: + raise PortRefused(f"Shuttle phase unavailable: {type(exc).__name__}") from None + + def admit(self, scope, scheduled_at): + if type(scheduled_at) is not int or scheduled_at != self.slot: + raise PortRefused("delivery slot differs from pinned slot") + self._call('admit', scope) + + def started(self, run, scope): + self._call('started', scope, run) + + def finished(self, run, scope, outcome): + self._call('finished', scope, run, outcome) + + def publish(self, run, scope, outcome): + self._call('publish', scope, run, outcome) diff --git a/test_shuttle_process_port.py b/test_shuttle_process_port.py new file mode 100644 index 0000000..c15fe02 --- /dev/null +++ b/test_shuttle_process_port.py @@ -0,0 +1,126 @@ +"""Real process/descriptor controls; subprocess is a fixture CLI, not Shuttle crypto.""" +from dataclasses import replace +import hashlib +import json +import sys + +import pytest + +from scheduled_receiver import ExecutionScope, receive, run_id +from shuttle_process_port import PortRefused, ShuttleProcessPort +from slot_receiver import SlotLedger + + +@pytest.fixture +def setup(tmp_path): + trace = tmp_path / 'trace.jsonl' + config = tmp_path / 'config.json' + config.write_text(json.dumps({'trace': str(trace), 'mode': 'ok'})) + config.chmod(0o600) + executable = tmp_path / 'shuttle-fixture' + executable.write_text('#!' + sys.executable + '\n' + '''import json,sys,time +from pathlib import Path +assert sys.argv[1:3] == ['--require-existing-keys', 'scheduled'] +action = sys.argv[3] +assert sys.argv[4] == '--config' +config = json.loads(Path(sys.argv[5]).read_text()) +packet = json.load(sys.stdin) +with open(config['trace'], 'a') as out: + out.write(json.dumps({'action': action, 'packet': packet})+'\\n') +mode = config['mode'] +if mode == 'timeout': time.sleep(10) +if mode == 'fail': sys.exit(3) +if mode == 'large': print('x'*4097) +elif mode == 'duplicate': print('{"action":"bad","action":"'+action+'","result":"verified"}') +elif mode == 'wrong': print(json.dumps({'action': 'wrong', 'result': 'verified'})) +elif mode == 'empty': pass +else: print(json.dumps({'action': action, 'result': 'verified'})) +''') + executable.chmod(0o700) + scope = ExecutionScope('urn:example:job', 'r1', 'urn:example:host', 'a'*64, + 'verifier', 'urn:example:workflow') + def port(mode='ok', timeout=5): + config.write_text(json.dumps({'trace': str(trace), 'mode': mode})) + return ShuttleProcessPort(executable, config, hashlib.sha256(config.read_bytes()).hexdigest(), + scope, 600, timeout_s=timeout) + return port, scope, trace, config, executable + + +def test_real_process_calls_all_phases_and_duplicate_only_publishes(setup, tmp_path): + factory, scope, trace, _, _ = setup + port = factory() + delivery = {**scope.__dict__, 'scheduled_at': 600} + calls = [] + def execute(): + calls.append(1) + return {'outcome': 'skipped-by-hold', 'exit_code': 0} + ledger = SlotLedger(tmp_path/'ledger.sqlite') + result = receive(ledger, scope, 600, delivery, port, execute) + assert receive(ledger, scope, 600, delivery, port, execute) == result + assert calls == [1] + events = [json.loads(x) for x in trace.read_text().splitlines()] + assert [e['action'] for e in events] == ['admit', 'started', 'finished', 'publish', 'admit', 'publish'] + assert events[2]['packet']['outcome'] == {'outcome': 'skipped-by-hold', 'exit_code': 0} + assert events[2]['packet']['run'] == run_id(scope, 600) + + +@pytest.mark.parametrize('mode', ['fail', 'large', 'duplicate', 'wrong', 'empty', 'timeout']) +def test_failed_or_unverified_admission_never_executes_child(setup, tmp_path, mode): + factory, scope, _, _, _ = setup + port = factory(mode, timeout=.1 if mode == 'timeout' else 5) + with pytest.raises(PortRefused): + receive(SlotLedger(tmp_path/'ledger.sqlite'), scope, 600, + {**scope.__dict__, 'scheduled_at': 600}, port, + lambda: pytest.fail('unverified admission executed child')) + + +def test_pinned_config_drift_refuses_before_process(setup): + factory, scope, trace, config, _ = setup + port = factory() + config.write_text('{}') + with pytest.raises(PortRefused, match='pinned bytes'): + port.admit(scope, 600) + assert not trace.exists() + + +@pytest.mark.parametrize('change', ['symlink', 'writable', 'fifo']) +def test_config_custody_refuses_without_blocking_or_launch(setup, change, tmp_path): + factory, scope, trace, config, _ = setup + port = factory() + if change == 'writable': + config.chmod(0o666) + else: + config.unlink() + if change == 'symlink': + config.symlink_to(tmp_path/'elsewhere') + else: + import os + os.mkfifo(config) + with pytest.raises(PortRefused): + port.admit(scope, 600) + assert not trace.exists() + + +def test_scope_slot_and_run_mismatch_refuse_before_process(setup): + factory, scope, trace, _, _ = setup + port = factory() + for callback in [lambda: port.admit(scope, True), lambda: port.admit(scope, 601), + lambda: port.admit(replace(scope, host='urn:other'), 600), + lambda: port.started('other', scope)]: + with pytest.raises(PortRefused): + callback() + assert not trace.exists() + + +def test_replaced_config_path_does_not_replace_opened_pinned_config(setup, monkeypatch): + import subprocess + factory, scope, trace, config, _ = setup + port = factory() + run = subprocess.run + def replace_path(*args, **kwargs): + config.rename(config.with_suffix('.original')) + config.write_text('{}') + return run(*args, **kwargs) + monkeypatch.setattr(subprocess, 'run', replace_path) + port.admit(scope, 600) + assert json.loads(trace.read_text())['action'] == 'admit'