diff --git a/backend/cortex_backend/execution/__init__.py b/backend/cortex_backend/execution/__init__.py index 70e44b9..4efc484 100644 --- a/backend/cortex_backend/execution/__init__.py +++ b/backend/cortex_backend/execution/__init__.py @@ -1,9 +1,25 @@ -"""Phase 1 durable execution primitives. +"""Durable execution primitives and provider-independent safety contracts. -Only the deterministic fake provider is exposed in this phase. Real runtime -providers are intentionally absent until later ADR gates are approved. +Only the deterministic fake provider and transport-neutral broker contract are +exposed in this phase. Native transports and real runtime providers remain absent +until later ADR gates are approved. """ +from .broker import ( + BrokerAclPolicy, + BrokerFrame, + BrokerFrameDecoder, + BrokerMessage, + BrokerPeerPolicy, + BrokerProtocolError, + BrokerSessionKeys, + PeerIdentity, + authorize_message, + decode_frame, + decode_message, + encode_frame, + encode_message, +) from .fake import FakeExecutionPlan, FakeExecutionProvider from .lifecycle import ExecutionLifecycle, LifecycleSnapshot, RuntimeHealth from .manifest import ( @@ -42,6 +58,13 @@ "ArtifactLimitError", "ApprovalPolicyError", "ApprovalTransitionError", + "BrokerAclPolicy", + "BrokerFrame", + "BrokerFrameDecoder", + "BrokerMessage", + "BrokerPeerPolicy", + "BrokerProtocolError", + "BrokerSessionKeys", "ExecutionRepository", "ExecutionRepositoryError", "ExecutionLifecycle", @@ -69,4 +92,10 @@ "parse_image_transform", "verify_bundle_files", "verify_signed_manifest", + "PeerIdentity", + "authorize_message", + "decode_frame", + "decode_message", + "encode_frame", + "encode_message", ] diff --git a/backend/cortex_backend/execution/broker.py b/backend/cortex_backend/execution/broker.py new file mode 100644 index 0000000..aaa8c00 --- /dev/null +++ b/backend/cortex_backend/execution/broker.py @@ -0,0 +1,414 @@ +"""Authenticated, bounded broker contract for a future local executor. + +The broker contract is transport-neutral. It validates framed messages, direction- +specific MAC keys, OS-provided peer identity, and installation/job ownership before a +future named-pipe adapter may dispatch anything. No socket, pipe, process, or +provider is opened by this module. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import hmac +import json +import math +import re +from struct import Struct, error as StructError +from typing import Any, Callable, Literal, Mapping + +from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator + + +MAX_BROKER_PAYLOAD_BYTES = 64 * 1024 +MAX_BROKER_CHUNK_BYTES = 256 * 1024 +MAX_BROKER_BODY_KEYS = 32 +_MAGIC = b"CXBF" +_VERSION = 1 +_TAG_BYTES = hashlib.sha256().digest_size +_HEADER = Struct(">4sBBHQI") +_MAX_SEQUENCE = (1 << 63) - 1 +_SAFE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$") +_PRINCIPAL = re.compile(r"^[0-9a-f]{64}$") +_SID = re.compile(r"^S-1-[0-9]+(?:-[0-9]+)+$") +_FORBIDDEN_BODY_KEYS = frozenset( + {"path", "source", "command", "shell", "executable", "network", "token"} +) + +BrokerDirection = Literal["to_broker", "to_executor"] +BrokerOperation = Literal["prepare", "start", "cancel", "collect"] +IntegrityLevel = Literal["low", "medium", "high", "system"] + + +class BrokerProtocolError(ValueError): + """Stable, non-sensitive protocol failure category.""" + + def __init__(self, code: str) -> None: + if re.fullmatch(r"[a-z][a-z0-9_]{0,63}", code) is None: + raise ValueError("invalid broker protocol code") + self.code = code + super().__init__("The local execution broker rejected the message.") + + +@dataclass(frozen=True, slots=True) +class BrokerSessionKeys: + """Independent MAC keys prevent valid frames being reflected cross-direction.""" + + to_broker: bytes + to_executor: bytes + + def __post_init__(self) -> None: + if ( + not isinstance(self.to_broker, bytes) + or not isinstance(self.to_executor, bytes) + or len(self.to_broker) < 32 + or len(self.to_executor) < 32 + or hmac.compare_digest(self.to_broker, self.to_executor) + ): + raise ValueError("broker direction keys must be distinct 32-byte values") + + def for_direction(self, direction: BrokerDirection) -> bytes: + return self.to_broker if direction == "to_broker" else self.to_executor + + +@dataclass(frozen=True, slots=True) +class BrokerFrame: + sequence: int + payload: bytes + + +def _validate_key(key: bytes) -> None: + if not isinstance(key, bytes) or len(key) < 32: + raise BrokerProtocolError("frame_key_invalid") + + +def _validate_sequence(sequence: int) -> None: + if type(sequence) is not int or not 1 <= sequence <= _MAX_SEQUENCE: + raise BrokerProtocolError("frame_sequence_invalid") + + +def encode_frame(payload: bytes, *, sequence: int, key: bytes) -> bytes: + _validate_key(key) + _validate_sequence(sequence) + if not isinstance(payload, bytes) or len(payload) > MAX_BROKER_PAYLOAD_BYTES: + raise BrokerProtocolError("frame_payload_too_large") + header = _HEADER.pack(_MAGIC, _VERSION, 0, 0, sequence, len(payload)) + tag = hmac.new(key, header + payload, hashlib.sha256).digest() + return header + payload + tag + + +def decode_frame( + encoded: bytes, + *, + key: bytes, + expected_sequence: int | None = None, +) -> BrokerFrame: + _validate_key(key) + if not isinstance(encoded, bytes) or len(encoded) < _HEADER.size + _TAG_BYTES: + raise BrokerProtocolError("frame_truncated") + if expected_sequence is not None: + _validate_sequence(expected_sequence) + try: + magic, version, flags, reserved, sequence, payload_length = _HEADER.unpack_from(encoded) + except StructError: + raise BrokerProtocolError("frame_invalid") from None + if magic != _MAGIC or version != _VERSION or flags != 0 or reserved != 0: + raise BrokerProtocolError("frame_header_invalid") + _validate_sequence(sequence) + if payload_length > MAX_BROKER_PAYLOAD_BYTES: + raise BrokerProtocolError("frame_payload_too_large") + expected_length = _HEADER.size + payload_length + _TAG_BYTES + if len(encoded) != expected_length: + raise BrokerProtocolError("frame_length_invalid") + if expected_sequence is not None and sequence != expected_sequence: + raise BrokerProtocolError("frame_replay") + payload_end = _HEADER.size + payload_length + expected_tag = hmac.new(key, encoded[:payload_end], hashlib.sha256).digest() + if not hmac.compare_digest(expected_tag, encoded[payload_end:]): + raise BrokerProtocolError("frame_authentication_failed") + return BrokerFrame(sequence=sequence, payload=encoded[_HEADER.size:payload_end]) + + +class BrokerFrameDecoder: + """Incremental decoder that bounds buffering and enforces strict sequencing.""" + + def __init__(self, *, key: bytes, first_sequence: int = 1) -> None: + _validate_key(key) + _validate_sequence(first_sequence) + self._key = key + self._next_sequence = first_sequence + self._buffer = bytearray() + + def feed(self, chunk: bytes) -> tuple[BrokerFrame, ...]: + if not isinstance(chunk, bytes) or len(chunk) > MAX_BROKER_CHUNK_BYTES: + raise BrokerProtocolError("frame_chunk_too_large") + self._buffer.extend(chunk) + if len(self._buffer) > MAX_BROKER_CHUNK_BYTES + _HEADER.size + _TAG_BYTES: + raise BrokerProtocolError("frame_buffer_too_large") + frames: list[BrokerFrame] = [] + while len(self._buffer) >= _HEADER.size: + try: + magic, version, flags, reserved, sequence, payload_length = _HEADER.unpack_from( + self._buffer + ) + except StructError: + raise BrokerProtocolError("frame_invalid") from None + if magic != _MAGIC or version != _VERSION or flags != 0 or reserved != 0: + raise BrokerProtocolError("frame_header_invalid") + _validate_sequence(sequence) + if payload_length > MAX_BROKER_PAYLOAD_BYTES: + raise BrokerProtocolError("frame_payload_too_large") + frame_length = _HEADER.size + payload_length + _TAG_BYTES + if len(self._buffer) < frame_length: + break + raw = bytes(self._buffer[:frame_length]) + del self._buffer[:frame_length] + frame = decode_frame( + raw, + key=self._key, + expected_sequence=self._next_sequence, + ) + self._next_sequence += 1 + frames.append(frame) + return tuple(frames) + + +class BrokerMessage(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + + schema_version: Literal["broker.message.v1"] + direction: BrokerDirection + operation: BrokerOperation + request_id: str + job_id: str + installation_principal_id: str + body: dict[str, Any] = Field(max_length=MAX_BROKER_BODY_KEYS) + + @field_validator("request_id", "job_id") + @classmethod + def _safe_ids(cls, value: str) -> str: + if _SAFE_ID.fullmatch(value) is None: + raise ValueError("broker id is invalid") + return value + + @field_validator("installation_principal_id") + @classmethod + def _principal(cls, value: str) -> str: + if _PRINCIPAL.fullmatch(value) is None: + raise ValueError("broker principal is invalid") + return value + + @field_validator("body") + @classmethod + def _safe_body(cls, value: dict[str, Any]) -> dict[str, Any]: + def is_invalid(value: Any, depth: int = 0, seen: set[int] | None = None) -> bool: + if depth > 8: + return True + if seen is None: + seen = set() + if isinstance(value, Mapping): + marker = id(value) + if marker in seen: + return True + seen.add(marker) + try: + return any( + not isinstance(key, str) + or key.casefold() in _FORBIDDEN_BODY_KEYS + or is_invalid(item, depth + 1, seen) + for key, item in value.items() + ) + finally: + seen.remove(marker) + if isinstance(value, (list, tuple)): + marker = id(value) + if marker in seen: + return True + seen.add(marker) + try: + return any(is_invalid(item, depth + 1, seen) for item in value) + finally: + seen.remove(marker) + if value is None or isinstance(value, (str, bool, int)): + return False + if isinstance(value, float): + return not math.isfinite(value) + return True + + if is_invalid(value): + raise ValueError("broker body contains an invalid or forbidden value") + return value + + def canonical_json(self) -> bytes: + # Pydantic's frozen model does not deep-freeze nested dictionaries. Revalidate + # before serialization so a caller cannot mutate the body after construction + # and smuggle authority fields or non-JSON values into a signed frame. + type(self).model_validate(self.model_dump(mode="python")) + return json.dumps( + self.model_dump(mode="json"), + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + allow_nan=False, + ).encode("ascii") + + +def encode_message(message: BrokerMessage, *, keys: BrokerSessionKeys, sequence: int) -> bytes: + try: + payload = message.canonical_json() + except (TypeError, ValueError, ValidationError): + raise BrokerProtocolError("message_invalid") from None + return encode_frame(payload, sequence=sequence, key=keys.for_direction(message.direction)) + + +def decode_message( + frame: BrokerFrame, + *, + direction: BrokerDirection, +) -> BrokerMessage: + try: + payload = json.loads(frame.payload.decode("ascii")) + message = BrokerMessage.model_validate(payload) + except (UnicodeDecodeError, json.JSONDecodeError, ValidationError): + raise BrokerProtocolError("message_invalid") from None + try: + canonical_payload = message.canonical_json() + except (TypeError, ValueError, ValidationError): + raise BrokerProtocolError("message_invalid") from None + if message.direction != direction or canonical_payload != frame.payload: + raise BrokerProtocolError("message_noncanonical") + return message + + +@dataclass(frozen=True, slots=True) +class PeerIdentity: + process_id: int + user_sid: str + app_container_sid: str | None + integrity_level: IntegrityLevel + + def __post_init__(self) -> None: + if ( + type(self.process_id) is not int + or self.process_id <= 0 + or not isinstance(self.user_sid, str) + or _SID.fullmatch(self.user_sid) is None + ): + raise ValueError("peer identity is invalid") + if self.app_container_sid is not None and ( + not isinstance(self.app_container_sid, str) + or _SID.fullmatch(self.app_container_sid) is None + ): + raise ValueError("peer app-container identity is invalid") + if not isinstance(self.integrity_level, str) or self.integrity_level not in { + "low", + "medium", + "high", + "system", + }: + raise ValueError("peer integrity level is invalid") + + +@dataclass(frozen=True, slots=True) +class BrokerAclPolicy: + """Allowlist that a native named-pipe adapter must apply to its DACL.""" + + allowed_user_sids: frozenset[str] + allowed_app_container_sids: frozenset[str] + + def __post_init__(self) -> None: + if not self.allowed_user_sids or not self.allowed_app_container_sids: + raise ValueError("broker ACL must allow user and app-container SIDs") + if any( + not isinstance(sid, str) or _SID.fullmatch(sid) is None + for sid in self.allowed_user_sids + ): + raise ValueError("broker ACL user SID is invalid") + if any( + not isinstance(sid, str) or _SID.fullmatch(sid) is None + for sid in self.allowed_app_container_sids + ): + raise ValueError("broker ACL app-container SID is invalid") + + +@dataclass(frozen=True, slots=True) +class BrokerPeerPolicy: + acl: BrokerAclPolicy + expected_process_id: int | None = None + maximum_integrity: IntegrityLevel = "low" + + def __post_init__(self) -> None: + if self.expected_process_id is not None and ( + type(self.expected_process_id) is not int or self.expected_process_id <= 0 + ): + raise ValueError("expected broker process ID is invalid") + if not isinstance(self.maximum_integrity, str) or self.maximum_integrity not in { + "low", + "medium", + "high", + "system", + }: + raise ValueError("broker integrity policy is invalid") + + def validate(self, identity: PeerIdentity) -> None: + if self.expected_process_id is not None and identity.process_id != self.expected_process_id: + raise BrokerProtocolError("peer_identity_mismatch") + if identity.user_sid not in self.acl.allowed_user_sids: + raise BrokerProtocolError("peer_acl_denied") + if ( + self.acl.allowed_app_container_sids + and identity.app_container_sid not in self.acl.allowed_app_container_sids + ): + raise BrokerProtocolError("peer_acl_denied") + levels = {"low": 0, "medium": 1, "high": 2, "system": 3} + if self.maximum_integrity not in levels: + raise BrokerProtocolError("peer_integrity_policy_invalid") + if levels[identity.integrity_level] > levels[self.maximum_integrity]: + raise BrokerProtocolError("peer_integrity_denied") + + +def authorize_message( + message: BrokerMessage, + *, + peer: PeerIdentity, + peer_policy: BrokerPeerPolicy, + expected_principal_id: str, + owner_for_job: Callable[[str], str | None], +) -> None: + """Bind a verified frame to the trusted installation and job owner.""" + if ( + not isinstance(expected_principal_id, str) + or _PRINCIPAL.fullmatch(expected_principal_id) is None + ): + raise BrokerProtocolError("broker_principal_invalid") + peer_policy.validate(peer) + if message.installation_principal_id != expected_principal_id: + raise BrokerProtocolError("broker_principal_mismatch") + try: + owner = owner_for_job(message.job_id) + except Exception: + raise BrokerProtocolError("broker_owner_lookup_failed") from None + if owner != expected_principal_id: + raise BrokerProtocolError("broker_owner_mismatch") + + +__all__ = [ + "BrokerAclPolicy", + "BrokerDirection", + "BrokerFrame", + "BrokerFrameDecoder", + "BrokerMessage", + "BrokerOperation", + "BrokerPeerPolicy", + "BrokerProtocolError", + "BrokerSessionKeys", + "IntegrityLevel", + "MAX_BROKER_CHUNK_BYTES", + "MAX_BROKER_PAYLOAD_BYTES", + "PeerIdentity", + "authorize_message", + "decode_frame", + "decode_message", + "encode_frame", + "encode_message", +] diff --git a/docs/adr/0001-phase2-broker-contract.md b/docs/adr/0001-phase2-broker-contract.md new file mode 100644 index 0000000..7fef8e4 --- /dev/null +++ b/docs/adr/0001-phase2-broker-contract.md @@ -0,0 +1,81 @@ +# ADR-0001 Phase 2 authenticated broker contract + +- **Status:** Contract implemented and verified; native transport/provider enablement remains blocked +- **Parent:** [Phase 2 signed-manifest gate](0001-phase2-signed-manifest.md) +- **Scope:** Bounded authenticated frames, canonical broker messages, direction keys, + peer ACL/identity policy, and installation/job ownership authorization + +## Decision + +The future local executor broker uses a transport-neutral framed protocol before any +named-pipe adapter is allowed to dispatch work. A frame has a fixed binary header, +version, zero flags/reserved bits, positive monotonic sequence, bounded payload length, +payload bytes, and a 32-byte HMAC-SHA-256 tag over header plus payload. The hard payload +ceiling is 64 KiB and the decoder bounds chunk and buffered input. Unknown header values, +truncated frames, trailing bytes, oversized lengths, invalid keys, and sequence replay +fail closed. + +The broker session has independent 32-byte-or-longer MAC keys for `to_broker` and +`to_executor` traffic. A valid frame cannot be reflected across directions. The key +establishment and named-pipe transport are deliberately not implemented here; a future +native adapter must provision the keys through a reviewed parent/child handshake. +The implementation keeps frame authentication/sequencing separate from message +canonicalization: a message decoder accepts only an already authenticated frame, and +the authorization step remains a distinct trusted-peer/job-owner check. + +## Canonical message contract + +The frame payload is canonical ASCII JSON for `broker.message.v1`: + +- direction: `to_broker` or `to_executor`; +- operation: `prepare`, `start`, `cancel`, or `collect`; +- bounded request and job identifiers; +- the 256-bit installation principal; and +- a bounded body for a later operation-specific schema. + +The body rejects authority-bearing fields recursively: paths, source, commands, shells, +executables, network targets, and tokens cannot cross this generic broker envelope. +Future operation schemas may add only typed artifact IDs and bounded values. Noncanonical +JSON is rejected even when its MAC is valid, so idempotency and audit digests have one +representation. + +## Peer ACL and confused-deputy policy + +The native transport must construct a named-pipe DACL from `BrokerAclPolicy`, allowing +only the configured user SID and AppContainer SID. After connection, it must obtain the +peer process token and provide `PeerIdentity` to `BrokerPeerPolicy`; the policy checks: + +1. expected process ID when a launch binding exists; +2. exact user and AppContainer SIDs; +3. maximum integrity level (low by default); and +4. no unvalidated peer metadata supplied by the wire message. + +`authorize_message` then binds the message's installation principal to the trusted +session principal and asks a trusted repository lookup for the job owner. A request is +rejected if the wire principal, peer ACL, expected process, integrity, or durable owner +does not match. This prevents a broker from becoming a confused deputy for another +installation or job. + +## Failure and lifecycle contract + +Failures expose stable categories only: `frame_*`, `message_*`, `peer_*`, +`broker_principal_*`, `broker_owner_*`. No HMAC key, token, path, payload, or OS error +detail is returned. A native adapter must close the connection on any protocol or +identity error, discard the session sequence, and leave the execution lifecycle +unavailable. The broker contract has no retry, fallback transport, subprocess, or +provider behavior. + +## Explicit non-goals + +This ADR does not create a named pipe, set a Windows security descriptor, query a token, +derive session keys, spawn or supervise an executor, stage artifacts, or execute any +operation. Native DACL construction, peer-token acquisition, handshake design, +copy-in/output validation, and OS sandbox qualification require separate evidence and +review before the lifecycle health gate can become enabled. + +## Verification + +`tests/test_phase2_broker.py` covers MAC tampering, replay/sequence, payload/chunk +limits, incremental frames, direction reflection, canonical JSON, nested authority +fields, exact ACL and integrity checks, expected process binding, principal mismatch, +job-owner mismatch, and safe lookup failures. diff --git a/docs/adr/0001-phase2-evidence.md b/docs/adr/0001-phase2-evidence.md index 11b172b..b703f0f 100644 --- a/docs/adr/0001-phase2-evidence.md +++ b/docs/adr/0001-phase2-evidence.md @@ -1,7 +1,7 @@ # ADR-0001 Phase 2 evidence log - **Phase:** 2 — signed image recipes and calculator/check primitives -- **Status:** Typed contract and signed-manifest verification complete; provider and release gates remain open +- **Status:** Typed contract, signed-manifest verification, and broker contract complete; native provider and release gates remain open - **Scope:** Provider-independent validation and deterministic trusted primitives only - **Source decision:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Contract ADR:** [Phase 2 typed recipe and primitive contract](0001-phase2-recipe-contract.md) @@ -16,6 +16,8 @@ | Canonical plan identity | **Complete** | Validated plans expose stable canonical JSON and SHA-256 digests for future idempotency/signature binding. | | Signed recipe manifest | **Complete (verification only)** | Ed25519 signature verification uses a pinned key-id allowlist; every declared bundle entry is path-, size-, and SHA-256-verified; monotonic updates and explicit rollback authorization are enforced. | | Signed bundle installation/update | **Blocked / next gate** | Verification does not install, load, atomically replace, persist state, rotate keys, or enable a provider. | +| Authenticated broker contract | **Complete (transport-neutral)** | Bounded versioned frames, direction-specific HMAC keys, canonical messages, peer ACL/integrity policy, and owner-scoped authorization are covered by adversarial tests. | +| Native named-pipe adapter/DACL/peer-token binding | **Blocked / next gate** | No pipe, security descriptor, token query, key handshake, or executor connection is enabled. | | Copy-in, image decoding, output validation, publication | **Blocked / next gate** | No codec or provider path has been enabled; Phase 1 artifact storage remains the only publication mechanism. | | Production broker and sandbox provider | **Blocked / next gate** | No named-pipe broker, Wasmtime, AppContainer, Job Object, subprocess, or production execution route was added. | @@ -39,12 +41,18 @@ SHA-256 before any future installation decision; verification does not load it. 9. The Phase 1 application lifecycle remains explicitly disabled for production execution; this stage cannot make a provider visible by itself. +10. Frames are bounded, authenticated with direction-specific keys, canonical, and + strictly sequenced; replay, reflection, truncation, and malformed headers fail + closed. +11. Peer ACL/identity and durable job ownership are checked outside the wire payload; + a principal or job mismatch cannot be used as a confused deputy. ## Re-run target ```powershell python -m pytest tests/test_phase2_recipe_contract.py -q python -m pytest tests/test_phase2_manifest.py -q +python -m pytest tests/test_phase2_broker.py -q python -m compileall -q backend\cortex_backend\execution tests python -m pytest -q python tools/generate_contracts.py @@ -55,7 +63,7 @@ npm.cmd test --prefix frontend -- --run ``` **Validation result (2026-07-21):** 16 Phase 2 contract tests, 9 signed-manifest tests, -and the full Python suite passed (148 tests total) with one pre-existing +7 broker-contract tests, and the full Python suite passed (155 tests total) with one pre-existing `pytest-asyncio` deprecation warning. Frontend lint, typecheck, production build, and all 39 frontend tests passed. Contract generation, compileall, and `git diff --check` passed. No production execution diff --git a/docs/adr/0001-phase2-recipe-contract.md b/docs/adr/0001-phase2-recipe-contract.md index 2556247..9fb1700 100644 --- a/docs/adr/0001-phase2-recipe-contract.md +++ b/docs/adr/0001-phase2-recipe-contract.md @@ -1,6 +1,6 @@ # ADR-0001 Phase 2 typed recipe and primitive contract -- **Status:** Contract and signed-manifest verification implemented and verified; provider enablement remains blocked +- **Status:** Typed contract, signed-manifest verification, and transport-neutral broker contract implemented and verified; native transport and provider enablement remain blocked - **Parent:** [Capability-tiered agentic execution harness](0001-capability-tiered-agentic-execution-harness.md) - **Depends on:** [Phase 1 production lifecycle gate](0001-phase1-production-lifecycle.md) - **Scope:** Typed fixed-function image plans, calculator/check primitives, canonical @@ -54,7 +54,7 @@ This ADR does not authorize: byte verification are implemented separately in [the signed-manifest ADR](0001-phase2-signed-manifest.md); - image codecs, thumbnails, or decompression handling; -- production broker named-pipe ACL, peer identity, framing, or IPC; +- native production broker named-pipe ACL, peer-token acquisition, or IPC; - artifact copy-in, output validation, atomic publication, or source ownership binding; - Wasmtime/WASI, AppContainer/LPAC, Job Object, host process, or any other provider; - model prompt/tool exposure, automatic execution, or application lifecycle enablement. @@ -64,8 +64,9 @@ packaged application remains on the explicitly disabled lifecycle from Phase 1. ## Required next gates -1. Implement the production broker contract with ACL, peer identity, framing, message - limits, and confused-deputy tests. +1. Complete the native broker adapter: named-pipe DACL, peer-token binding, reviewed + key handshake, and connection lifecycle around the verified transport-neutral + contract in [the broker ADR](0001-phase2-broker-contract.md). 2. Implement trusted copy-in/output validation and artifact publication tests, including parser fuzzing and source non-overwrite proofs. 3. Qualify the fixed-function provider inside the OS sandbox and wire it only through diff --git a/tests/test_phase2_broker.py b/tests/test_phase2_broker.py new file mode 100644 index 0000000..65aa326 --- /dev/null +++ b/tests/test_phase2_broker.py @@ -0,0 +1,312 @@ +"""Authenticated broker framing, ACL, and confused-deputy tests.""" + +from __future__ import annotations + +import json + +import pytest + +from cortex_backend.execution.broker import ( + MAX_BROKER_PAYLOAD_BYTES, + BrokerAclPolicy, + BrokerFrameDecoder, + BrokerMessage, + BrokerPeerPolicy, + BrokerProtocolError, + BrokerSessionKeys, + PeerIdentity, + authorize_message, + decode_frame, + decode_message, + encode_frame, + encode_message, +) + + +PRINCIPAL = "a" * 64 +USER_SID = "S-1-5-21-100-200-300-400" +APP_SID = "S-1-15-2-100-200-300-400" + + +def _keys() -> BrokerSessionKeys: + return BrokerSessionKeys(b"b" * 32, b"e" * 32) + + +def _message(*, principal: str = PRINCIPAL, job_id: str = "job_1") -> BrokerMessage: + return BrokerMessage( + schema_version="broker.message.v1", + direction="to_broker", + operation="start", + request_id="request_1", + job_id=job_id, + installation_principal_id=principal, + body={"profile": "artifact.transform.v1"}, + ) + + +def _peer(*, process_id: int = 123, user_sid: str = USER_SID, integrity_level="low"): + return PeerIdentity( + process_id=process_id, + user_sid=user_sid, + app_container_sid=APP_SID, + integrity_level=integrity_level, + ) + + +def _policy(*, process_id: int | None = 123, maximum_integrity="low"): + return BrokerPeerPolicy( + acl=BrokerAclPolicy( + allowed_user_sids=frozenset({USER_SID}), + allowed_app_container_sids=frozenset({APP_SID}), + ), + expected_process_id=process_id, + maximum_integrity=maximum_integrity, + ) + + +def test_frame_mac_length_and_sequence_are_bounded_and_replay_safe(): + keys = _keys() + encoded = encode_frame(b"hello", sequence=1, key=keys.to_broker) + assert decode_frame(encoded, key=keys.to_broker, expected_sequence=1).payload == b"hello" + + tampered = bytearray(encoded) + tampered[-1] ^= 1 + with pytest.raises(BrokerProtocolError) as auth_error: + decode_frame(bytes(tampered), key=keys.to_broker, expected_sequence=1) + assert auth_error.value.code == "frame_authentication_failed" + + with pytest.raises(BrokerProtocolError) as replay_error: + decode_frame(encoded, key=keys.to_broker, expected_sequence=2) + assert replay_error.value.code == "frame_replay" + + with pytest.raises(BrokerProtocolError) as size_error: + encode_frame(b"x" * (MAX_BROKER_PAYLOAD_BYTES + 1), sequence=1, key=keys.to_broker) + assert size_error.value.code == "frame_payload_too_large" + + +def test_malformed_headers_and_transport_chunks_fail_closed(): + key = b"b" * 32 + encoded = encode_frame(b"hello", sequence=1, key=key) + + with pytest.raises(BrokerProtocolError) as truncated_error: + decode_frame(b"", key=key) + assert truncated_error.value.code == "frame_truncated" + + malformed = bytearray(encoded) + malformed[4] = 2 + with pytest.raises(BrokerProtocolError) as header_error: + decode_frame(bytes(malformed), key=key) + assert header_error.value.code == "frame_header_invalid" + + with pytest.raises(BrokerProtocolError) as chunk_error: + BrokerFrameDecoder(key=key).feed(b"x" * (256 * 1024 + 1)) + assert chunk_error.value.code == "frame_chunk_too_large" + + +def test_key_body_and_identity_constructors_are_strictly_bounded(): + with pytest.raises(ValueError): + BrokerSessionKeys(b"same" * 8, b"same" * 8) + with pytest.raises(BrokerProtocolError) as key_error: + decode_frame(b"x" * 52, key=b"short") + assert key_error.value.code == "frame_key_invalid" + + with pytest.raises(ValueError): + BrokerMessage( + schema_version="broker.message.v1", + direction="to_broker", + operation="start", + request_id="request_1", + job_id="job_1", + installation_principal_id=PRINCIPAL, + body={f"k{i}": i for i in range(33)}, + ) + with pytest.raises(ValueError): + BrokerMessage( + schema_version="broker.message.v1", + direction="to_broker", + operation="start", + request_id="request_1", + job_id="job_1", + installation_principal_id=PRINCIPAL, + body={"value": float("nan")}, + ) + with pytest.raises(ValueError): + PeerIdentity( + process_id=True, + user_sid=USER_SID, + app_container_sid=APP_SID, + integrity_level="low", + ) + with pytest.raises(ValueError): + BrokerPeerPolicy(acl=_policy().acl, expected_process_id=0) + + +def test_incremental_decoder_handles_split_frames_and_rejects_direction_reflection(): + keys = _keys() + first = encode_frame(b"one", sequence=1, key=keys.to_broker) + second = encode_frame(b"two", sequence=2, key=keys.to_broker) + decoder = BrokerFrameDecoder(key=keys.to_broker) + + assert decoder.feed(first[:7]) == () + assert decoder.feed(first[7:] + second) == ( + decode_frame(first, key=keys.to_broker, expected_sequence=1), + decode_frame(second, key=keys.to_broker, expected_sequence=2), + ) + + reflected = encode_frame(b"one", sequence=1, key=keys.to_executor) + with pytest.raises(BrokerProtocolError) as direction_error: + decode_frame(reflected, key=keys.to_broker, expected_sequence=1) + assert direction_error.value.code == "frame_authentication_failed" + + +def test_message_is_canonical_and_direction_bound(): + keys = _keys() + message = _message() + encoded = encode_message(message, keys=keys, sequence=1) + frame = decode_frame(encoded, key=keys.to_broker, expected_sequence=1) + assert decode_message(frame, direction="to_broker") == message + + noncanonical_payload = json.dumps( + message.model_dump(mode="json"), + ensure_ascii=True, + separators=(",", ":"), + ).encode("ascii") + noncanonical_frame = encode_frame(noncanonical_payload, sequence=1, key=keys.to_broker) + with pytest.raises(BrokerProtocolError) as canonical_error: + decode_message( + decode_frame(noncanonical_frame, key=keys.to_broker, expected_sequence=1), + direction="to_broker", + ) + assert canonical_error.value.code == "message_noncanonical" + + forbidden_payload = { + **message.model_dump(), + "body": {"path": "C:\\private", "profile": "artifact.transform.v1"}, + } + forbidden_frame = encode_frame( + json.dumps( + forbidden_payload, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("ascii"), + sequence=1, + key=keys.to_broker, + ) + with pytest.raises(BrokerProtocolError) as body_error: + decode_message( + decode_frame(forbidden_frame, key=keys.to_broker, expected_sequence=1), + direction="to_broker", + ) + assert body_error.value.code == "message_invalid" + + nested_payload = { + **message.model_dump(), + "body": {"input": {"metadata": [{"path": "C:\\private"}]}}, + } + nested_frame = encode_frame( + json.dumps( + nested_payload, + ensure_ascii=True, + sort_keys=True, + separators=(",", ":"), + ).encode("ascii"), + sequence=1, + key=keys.to_broker, + ) + with pytest.raises(BrokerProtocolError) as nested_error: + decode_message( + decode_frame(nested_frame, key=keys.to_broker, expected_sequence=1), + direction="to_broker", + ) + assert nested_error.value.code == "message_invalid" + + mutated = _message() + mutated.body["path"] = "C:\\private" + with pytest.raises(BrokerProtocolError) as mutated_error: + encode_message(mutated, keys=keys, sequence=1) + assert mutated_error.value.code == "message_invalid" + + +def test_peer_acl_identity_and_job_owner_are_all_required(): + keys = _keys() + message = _message() + peer = _peer() + policy = _policy() + encoded = encode_message(message, keys=keys, sequence=1) + verified = decode_message( + decode_frame(encoded, key=keys.to_broker, expected_sequence=1), + direction="to_broker", + ) + + authorize_message( + verified, + peer=peer, + peer_policy=policy, + expected_principal_id=PRINCIPAL, + owner_for_job=lambda job_id: PRINCIPAL if job_id == "job_1" else None, + ) + + with pytest.raises(BrokerProtocolError) as principal_error: + authorize_message( + _message(principal="b" * 64), + peer=peer, + peer_policy=policy, + expected_principal_id=PRINCIPAL, + owner_for_job=lambda _job_id: PRINCIPAL, + ) + assert principal_error.value.code == "broker_principal_mismatch" + + with pytest.raises(BrokerProtocolError) as owner_error: + authorize_message( + message, + peer=peer, + peer_policy=policy, + expected_principal_id=PRINCIPAL, + owner_for_job=lambda _job_id: "b" * 64, + ) + assert owner_error.value.code == "broker_owner_mismatch" + + with pytest.raises(BrokerProtocolError) as acl_error: + authorize_message( + message, + peer=_peer(user_sid="S-1-5-21-999-999-999-999"), + peer_policy=policy, + expected_principal_id=PRINCIPAL, + owner_for_job=lambda _job_id: PRINCIPAL, + ) + assert acl_error.value.code == "peer_acl_denied" + + with pytest.raises(BrokerProtocolError) as integrity_error: + authorize_message( + message, + peer=_peer(integrity_level="medium"), + peer_policy=policy, + expected_principal_id=PRINCIPAL, + owner_for_job=lambda _job_id: PRINCIPAL, + ) + assert integrity_error.value.code == "peer_integrity_denied" + + +def test_peer_pid_and_owner_lookup_fail_closed_without_details(): + message = _message() + with pytest.raises(BrokerProtocolError) as pid_error: + authorize_message( + message, + peer=_peer(process_id=999), + peer_policy=_policy(), + expected_principal_id=PRINCIPAL, + owner_for_job=lambda _job_id: PRINCIPAL, + ) + assert pid_error.value.code == "peer_identity_mismatch" + + with pytest.raises(BrokerProtocolError) as lookup_error: + authorize_message( + message, + peer=_peer(), + peer_policy=_policy(), + expected_principal_id=PRINCIPAL, + owner_for_job=lambda _job_id: (_ for _ in ()).throw(RuntimeError("private")), + ) + assert lookup_error.value.code == "broker_owner_lookup_failed" + assert "private" not in str(lookup_error.value)