diff --git a/adapters/README.md b/adapters/README.md index 221e2833..89c8e7d6 100644 --- a/adapters/README.md +++ b/adapters/README.md @@ -50,7 +50,7 @@ Under the hood it drives two complementary layers: **2. Emission conformance** — each adapter's `tests/integrations//test_emission.py` runs the **production adapter** (the real subprocess for Claude Code / Cursor; the real `pre_invoke` / `post_invoke` / lifecycle-observer middleware for NAT) against a validating **CaptureGuardian** and asserts the *exact bytes it sends* pass the canonical schemas. Two layers of checks: -- *Per-event:* each Core method is emitted exactly once, envelope + payload validate against the canonical schemas, and per-envelope invariants hold — UUID `request_id`, RFC 3339 `timestamp`, `agent_id`/`session_id`/`platform` metadata, the `{value: …}` argument wrapper, and an **independently-recomputed** HMAC-SHA256 signature (a from-scratch HKDF+HMAC+JCS verifier in `acs/testing/capture_guardian.py`, *not* the adapters' own `acs._common` code, so a shared signing+verifying bug can't pass both sides; pinned by a frozen known-answer vector). +- *Per-event:* each Core method is emitted exactly once, envelope + payload validate against the canonical schemas, and per-envelope invariants hold — UUID `request_id`, RFC 3339 `timestamp`, `agent_id`/`session_id`/`platform` metadata, the `{value: …}` argument wrapper, and an **independently-recomputed** HMAC-SHA256 signature (a from-scratch HKDF+HMAC+JCS verifier in `acs/testing/capture_guardian.py`, *not* the adapters' own `acs.core` signing code, so a shared signing+verifying bug can't pass both sides; pinned by a frozen known-answer vector). - *Per-session* (`*SequentialSession`, one Guardian + one handshake cache across an ordered sequence): the handshake fires exactly once, all `request_id`s are unique, `toolCallResult.request_id_ref` links back to its `toolCallRequest`, and — the real honesty check — the handshake's advertised `methods_implemented` **equals** the set of methods **actually emitted in the run** (equality, not just subset: catches over- *and* under-advertising), not merely methods that happen to have a schema. Breaking any adapter's method mapping, envelope/payload field, or signing turns the suite red against the schema files themselves. The oracle is proven non-vacuous by `tests/common/test_capture_guardian.py` (feed it broken envelopes, assert it flags each defect class + the signature KAT). Schemas load from the copy bundled in the package, or from `specification/v0.1.0/` in a source checkout (`ACS_SPEC_DIR` overrides); missing schemas are a hard FAIL. Format checking (`uuid`, `date-time`) is enforced via `rfc3339-validator`. @@ -226,8 +226,8 @@ adapters/ ├── pyproject.toml # distribution, extras, console scripts, entry points ├── uv.lock # exact dependency versions, with hashes ├── src/acs/ -│ ├── core/ # the ACS v0.1.0 schemas, bundled into the wheel -│ ├── _common.py # shared signing, handshake, session and audit code +│ ├── core/ # no I/O: envelopes, signing, decisions, error codes, bundled schemas +│ ├── client/ # I/O: the secret, the handshake, transport, session state, audit │ ├── integrations// │ │ ├── adapter.py # the hook, or the NAT middleware │ │ └── wire.py # writes the framework's own config diff --git a/adapters/SECURITY.md b/adapters/SECURITY.md index a6c8b825..534ddfbc 100644 --- a/adapters/SECURITY.md +++ b/adapters/SECURITY.md @@ -2,7 +2,7 @@ This document is the threat model for the three reference adapters (`src/acs/integrations/`), the example Guardian (`src/acs/guardian/`), -and the shared helpers in `src/acs/_common.py`. It says what attacks the code defends +and the shared helpers in `src/acs/core/` and `src/acs/client/`. It says what attacks the code defends against, how, and what is explicitly out of scope. ACS is a security project — the whole point of the standard is to diff --git a/adapters/docs/claude_code/README.md b/adapters/docs/claude_code/README.md index c09e6c46..c1e94718 100644 --- a/adapters/docs/claude_code/README.md +++ b/adapters/docs/claude_code/README.md @@ -234,7 +234,7 @@ python3 -c "import json, os; d=json.load(open(os.path.expanduser('~/.claude/sett echo "=== Guardian responds to system/ping? ===" cd "$ACS_REPO/adapters" && uv run python -c " -from acs._common import ping; import json +from acs.client import ping; import json r = ping('http://127.0.0.1:8787/acs'); print(json.dumps(r, indent=2)[:200] if r else 'no response') " ``` @@ -249,7 +249,7 @@ r = ping('http://127.0.0.1:8787/acs'); print(json.dumps(r, indent=2)[:200] if r - `tests/integrations/claude_code/`: round-trip, schema and live integration tests. - `docs/claude_code/example_payloads.md`: masked real-world payload examples showing exactly what Claude Code emits. -The adapter shares `src/acs/_common.py` with the Cursor and NAT adapters (signing, handshake cache, audit events, URL allowlist). +The adapter shares `acs.core` (signing, decision rules, error codes) and `acs.client` (handshake cache, session state, audit events, URL allowlist) with the Cursor and NAT adapters. ## Configuration diff --git a/adapters/docs/cursor/README.md b/adapters/docs/cursor/README.md index 33742b82..979c31f6 100644 --- a/adapters/docs/cursor/README.md +++ b/adapters/docs/cursor/README.md @@ -242,7 +242,7 @@ python3 -c "import json, os; d=json.load(open(os.path.expanduser('~/.cursor/hook echo "=== Guardian responds to system/ping? ===" cd "$ACS_REPO/adapters" && uv run python -c " -from acs._common import ping; import json +from acs.client import ping; import json r = ping('http://127.0.0.1:8787/acs'); print(json.dumps(r, indent=2)[:200] if r else 'no response') " ``` @@ -258,7 +258,7 @@ r = ping('http://127.0.0.1:8787/acs'); print(json.dumps(r, indent=2)[:200] if r - `docs/cursor/example_payloads.md`: masked real-world payload examples showing exactly what Cursor emits. - `docs/cursor/live_verification.md`: manual reproduction procedure (Cursor has no headless mode). -The adapter shares `src/acs/_common.py` with the Claude Code and NAT adapters (signing, handshake cache, audit events, URL allowlist). +The adapter shares `acs.core` (signing, decision rules, error codes) and `acs.client` (handshake cache, session state, audit events, URL allowlist) with the Claude Code and NAT adapters. ## How it differs from the Claude Code adapter diff --git a/adapters/docs/nat/README.md b/adapters/docs/nat/README.md index 9726e23f..00ae1f54 100644 --- a/adapters/docs/nat/README.md +++ b/adapters/docs/nat/README.md @@ -236,7 +236,7 @@ Have your Guardian log every received envelope's `method` and the source (middle - `tests/integrations/nat/test_failure_modes.py` — transport failures, signature checks, replay, timestamp skew, JSON-RPC errors. - `docs/nat/example_payloads.md` — masked real-world payload examples showing the in-process `InvocationContext` shape and what the adapter sends. -The adapter shares `src/acs/_common.py` with the Claude Code and Cursor adapters (signing, handshake cache, audit events, URL allowlist, JSON-RPC error code → cause mapping). +The adapter shares `acs.core` (signing, decision rules, JSON-RPC error code → cause mapping) and `acs.client` (handshake cache, audit events, URL allowlist) with the Claude Code and Cursor adapters. ## How it differs from the Claude Code / Cursor adapters diff --git a/adapters/src/acs/_common.py b/adapters/src/acs/_common.py deleted file mode 100644 index 2ee5ae1b..00000000 --- a/adapters/src/acs/_common.py +++ /dev/null @@ -1,979 +0,0 @@ -""" -Shared ACS v0.1.0 helpers used by the three reference adapters and the -example Guardian. - -Lives in `adapters/_common/`. Each adapter prepends this directory to -`sys.path` and imports the symbols it needs. - -What's in here: - -- `jcs_canonicalize` — RFC 8785 (JCS) canonicalization via the - `rfc8785` PyPI package; a hard dependency with no fallback, since - §10 permits no alternative canonicalization. -- `derive_session_key` — HKDF-SHA256 per-session key derivation per §10. -- `sign_envelope` / `verify_signature` — HMAC-SHA256 baseline signature - over JCS(envelope with signature field removed), per §10. -- `load_hmac_secret` — read the HMAC secret from `ACS_HMAC_SECRET_FILE` - (preferred: file mode 0600) or `ACS_HMAC_SECRET` (env). File path - beats env var so secrets don't sit in `ps aux` output. -- `iso8601_now` / `coerce_uuid` / `parse_iso8601` — time + ID helpers. -- `audit_event` — structured `ACS_AUDIT` line for §6.4 fail-open bypass. -- `ensure_session_handshake` / `ping` — protocol helpers (§4, §13). - `ensure_session_handshake` is idempotent per session via disk cache; - see its docstring. The old name `do_handshake` is kept as an alias. -- `session_state` — per-session JSON file used by adapters to track - last_step_id, seen_step_ids, etc. across separate hook-process - invocations (shell-stdin adapters spawn one process per hook). -""" -from __future__ import annotations - -import base64 -import binascii -import datetime -import hashlib -import hmac -import json -import os -import re -import stat -import sys -import time -import urllib.error -import urllib.request -import uuid -from pathlib import Path -from typing import Any - -ACS_VERSION = "0.1.0" -DEFAULT_SKEW_WINDOW_MS = 300_000 - -# Maximum bytes the Guardian will read from a single HTTP POST body. -# Matches the handshake's max_payload_size_bytes default. Defends against -# memory exhaustion via a huge Content-Length. -MAX_REQUEST_BODY_BYTES = 1_048_576 # 1 MiB - -# Maximum command string length to scan with destructive-pattern regexes. -# Real shell commands are tiny; longer inputs are either non-shell data -# routed through the wrong tool or a regex-DoS attempt. -DESTRUCTIVE_SCAN_MAX_LEN = 8 * 1024 # 8 KiB - - -# ----- Canonicalization ----- - -try: - import rfc8785 as _rfc8785 # type: ignore[import-not-found] -except ImportError as _rfc8785_import_error: - # HARD dependency, deliberately. Spec §10: "Alternative - # canonicalization is not permitted in v0.1." A sorted-keys - # json.dumps fallback diverges from RFC 8785 on floats (1.0 vs 1), - # -0.0, large exponents, and non-BMP key ordering. The failure mode - # is a silent policy bypass (divergent canonicalization → - # SIGNATURE_INVALID → fail posture), so refuse to import instead. - raise ImportError( - "ACS adapters require the `rfc8785` package for RFC 8785 (JCS) " - "canonical signing — spec §10 permits no alternative " - "canonicalization. Install it: pip install rfc8785" - ) from _rfc8785_import_error - - -def jcs_canonicalize(obj: Any) -> bytes: - """RFC 8785 (JSON Canonicalization Scheme) via the `rfc8785` package. - - A hard dependency — see the import block above for why there is - deliberately no fallback.""" - return _rfc8785.dumps(obj) - - -# ----- Signing (HMAC-SHA256 baseline per §10) ----- - -def derive_session_key(input_key_material: bytes, session_id: str) -> bytes: - """HKDF-SHA256 with session_id as `info`, no salt. - - Spec §10: 'per-session HMAC key is HKDF-derived from deployment-provided - input keying material (a pre-shared secret, or a transport channel - binding such as a TLS exporter) together with the session_id.' - """ - # HKDF-Extract with empty salt - prk = hmac.new(b"\x00" * 32, input_key_material, hashlib.sha256).digest() - # HKDF-Expand to 32 bytes with info = session_id - info = session_id.encode("utf-8") - t = hmac.new(prk, info + b"\x01", hashlib.sha256).digest() - return t # one 32-byte block is enough for HMAC-SHA256 - - -class SecretFilePermissionsError(RuntimeError): - """ACS_HMAC_SECRET_FILE exists but its permissions / ownership leak the secret.""" - - -class SecretFileUnreadableError(RuntimeError): - """ACS_HMAC_SECRET_FILE is configured but cannot be read. - - Raised loudly: silently downgrading to unsigned mode would turn - off the only integrity control ACS-Core mandates.""" - - -def _check_secret_file_perms(path: str) -> None: - """Refuse to read the secret file if anything about its mode or - ownership would expose the key to another local user.""" - # Reject symlinks — a symlink is an attack vector (replace target - # without changing the visible path). - if os.path.islink(path): - raise SecretFilePermissionsError( - f"ACS_HMAC_SECRET_FILE {path!r} is a symlink; refusing to follow") - st = os.stat(path) - mode = stat.S_IMODE(st.st_mode) - if mode & 0o077: - raise SecretFilePermissionsError( - f"ACS_HMAC_SECRET_FILE {path!r} mode {oct(mode)} is too permissive; " - f"must be 0600 or 0400 (no group/other access). " - f"Fix: chmod 600 {path}") - if hasattr(os, "geteuid") and st.st_uid != os.geteuid(): - raise SecretFilePermissionsError( - f"ACS_HMAC_SECRET_FILE {path!r} owned by uid {st.st_uid}, " - f"adapter is running as uid {os.geteuid()}; refusing") - - -def load_hmac_secret() -> bytes: - """Read the HMAC input keying material. - - Resolution order (first hit wins): - 1. `ACS_HMAC_SECRET_FILE` — path to a file containing the secret. - Preferred for production. Permissions MUST be 0600 (or 0400) and - the file MUST be owned by the running user. The file's content - (stripped of trailing whitespace) is the secret. Symlinks are - rejected. Insecure permissions raise SecretFilePermissionsError — - the adapter refuses to use a leaked secret rather than silently - proceed. - 2. `ACS_HMAC_SECRET` — env-var fallback. Quick for dev, less secure - (visible in `ps eauxw`, child-process envs, core dumps). - 3. Empty bytes — caller decides whether that means dev-mode or fail. - - Generate a secret: `openssl rand -hex 32 > /etc/acs/hmac.key && chmod 600 /etc/acs/hmac.key` - """ - path = os.environ.get("ACS_HMAC_SECRET_FILE", "").strip() - if path: - try: - _check_secret_file_perms(path) - with open(path, "rb") as f: - return f.read().rstrip(b"\r\n\t ") - except SecretFilePermissionsError: - raise - except OSError as e: - # Configured-but-unreadable is an incident, not dev mode. - audit_event("hmac_secret_file_unreadable", path=path, error=str(e)) - raise SecretFileUnreadableError( - f"ACS_HMAC_SECRET_FILE {path!r} is configured but cannot " - f"be read ({e}); refusing to continue unsigned" - ) from e - env_val = os.environ.get("ACS_HMAC_SECRET", "") - return env_val.encode("utf-8") if env_val else b"" - - -# Back-compat alias for internal callers. -_signing_secret = load_hmac_secret - - -# ----- URL scheme allowlist (defends against SSRF / file:// disclosure) ----- - -_ALLOWED_GUARDIAN_SCHEMES = frozenset({"http", "https"}) - - -def validate_guardian_url(url: str) -> None: - """Reject Guardian URLs whose scheme is not http/https. - - urllib.request.urlopen happily accepts file://, ftp://, data://, etc. - An attacker who controls ACS_GUARDIAN_URL could use file:// to read - arbitrary files the adapter user has access to, or data:// to feed - a crafted response. The adapter and any other code POSTing to the - Guardian MUST call this before urlopen. - - Optionally also restricts the hostname against an operator-provided - `ACS_GUARDIAN_HOST_ALLOWLIST` (comma-separated). Defense in depth - against env-var attacks that smuggle a real http:// URL to an - internal service the adapter shouldn't reach. - """ - from urllib.parse import urlparse - parsed = urlparse(url) - if parsed.scheme.lower() not in _ALLOWED_GUARDIAN_SCHEMES: - raise ValueError( - f"Guardian URL scheme {parsed.scheme!r} not allowed; " - f"only {sorted(_ALLOWED_GUARDIAN_SCHEMES)} permitted") - allow = os.environ.get("ACS_GUARDIAN_HOST_ALLOWLIST", "").strip() - if allow: - allowed_hosts = {h.strip().lower() for h in allow.split(",") if h.strip()} - host = (parsed.hostname or "").lower() - if host not in allowed_hosts: - raise ValueError( - f"Guardian host {host!r} not in ACS_GUARDIAN_HOST_ALLOWLIST " - f"({sorted(allowed_hosts)})") - - -# ----- Bounded regex scanning (defends against regex DoS / input bombing) ----- - -def scan_destructive_bash_safely(cmd: str, *, max_len: int = DESTRUCTIVE_SCAN_MAX_LEN): - """Run destructive-pattern regex scanning ONLY if cmd is below max_len. - - Returns: - None — command is short and matched no destructive pattern - a re.Pattern — command matched (caller decides what to do) - "input_too_large" — command exceeds max_len; caller MUST treat as - suspicious and MUST NOT silently allow (skipping the scan is - not the same as the scan returning "safe"). - - Caller pattern set is loaded lazily from example_guardian.DESTRUCTIVE_BASH_PATTERNS - to keep the canonical pattern set in one place. - """ - if len(cmd) > max_len: - audit_event("destructive_scan_skipped_oversized", - cmd_length=len(cmd), max_len=max_len) - return "input_too_large" - # Lazy import — _common doesn't directly own the pattern set, the - # Guardian does. Adapters that want to run the scan themselves can - # call this; the Guardian uses its own DESTRUCTIVE_BASH_PATTERNS - # directly. We import here so this function is callable without - # forcing example-guardian onto every adapter's path. - try: - from acs.guardian import example_guardian - for pat in example_guardian.DESTRUCTIVE_BASH_PATTERNS: - if pat.search(cmd): - return pat - return None - except ImportError: - return None - - -_warned_unsigned_mode = False - - -def _warn_unsigned_once() -> None: - """One loud warning per process when operating without key material. - - Unsigned mode is an authentication failure, not just a - confidentiality one: any local process that binds the Guardian port - first becomes the Guardian and allows everything. - Fine for a first local experiment; never fine silently.""" - global _warned_unsigned_mode - if not _warned_unsigned_mode: - _warned_unsigned_mode = True - audit_event( - "unsigned_mode", - detail=( - "no ACS_HMAC_SECRET / ACS_HMAC_SECRET_FILE configured — " - "envelopes are UNSIGNED and any local process that binds " - "the Guardian address is trusted" - ), - ) - - -def _envelope_container_key(envelope: dict) -> str: - """Where the signature lives: params for requests, result for - decision responses, error for error responses (every response is - signed under ACS-Core — including errors, since a spoofable - unsigned error under a fail-open posture is an allow).""" - if "method" in envelope: - return "params" - if "error" in envelope: - return "error" - return "result" - - -def sign_envelope(envelope: dict, *, key: bytes | None = None, - session_id: str | None = None, key_id: str = "default") -> dict: - """Add a signature to the envelope. Returns the envelope unchanged - (after a loud one-time audit event) if no key material is available.""" - if key is None: - ikm = _signing_secret() - if not ikm: - _warn_unsigned_once() - return envelope - if not session_id: - params = envelope.get("params") or envelope.get("result") or {} - meta = params.get("metadata") or {} - session_id = meta.get("session_id") or params.get("request_id") or "" - key = derive_session_key(ikm, session_id) - - container_key = _envelope_container_key(envelope) - container = envelope.get(container_key, {}) - # Strip any existing signature before signing - unsigned_container = {k: v for k, v in container.items() if k != "signature"} - unsigned_envelope = {**envelope, container_key: unsigned_container} - - sig_bytes = hmac.new(key, jcs_canonicalize(unsigned_envelope), hashlib.sha256).digest() - container["signature"] = { - "algorithm": "HMAC-SHA256", - "value": base64.b64encode(sig_bytes).decode("ascii"), - "key_id": key_id, - } - envelope[container_key] = container - return envelope - - -def verify_signature(envelope: dict, *, key: bytes | None = None, - session_id: str | None = None) -> bool: - """Verify a signed envelope (request, decision response, or error - response). Returns True if valid (or if no signature present and no - key material configured — local-dev mode).""" - container_key = _envelope_container_key(envelope) - container = envelope.get(container_key) or {} - sig = container.get("signature") - if sig is None: - # No signature on the wire; valid only if local-dev (no key configured) - return not bool(_signing_secret()) - - if sig.get("algorithm") != "HMAC-SHA256": - return False - expected_b64 = sig.get("value") - if not expected_b64: - return False - - if key is None: - ikm = _signing_secret() - if not ikm: - # Signature present but no key configured: cannot verify - return False - meta = container.get("metadata") or {} - session_id = session_id or meta.get("session_id") or container.get("request_id") or "" - key = derive_session_key(ikm, session_id) - - unsigned_container = {k: v for k, v in container.items() if k != "signature"} - unsigned_envelope = {**envelope, container_key: unsigned_container} - expected_bytes = hmac.new(key, jcs_canonicalize(unsigned_envelope), hashlib.sha256).digest() - # Malformed base64 (truncation, garbage chars, non-string) must NOT - # crash the request path — that turns a bad signature into a 500 / - # uncaught exception instead of the spec's SIGNATURE_INVALID - # (-32004) response. Return False so the caller emits the right - # error code and the audit event carries cause=signature_invalid_*. - try: - provided_bytes = base64.b64decode(expected_b64, validate=True) - except (binascii.Error, ValueError, TypeError): - return False - return hmac.compare_digest(expected_bytes, provided_bytes) - - -# ----- Safe decision normalization ----- - -# The five ACS dispositions (§6.3). Shared so every adapter classifies an -# arrived verdict against the same set. -KNOWN_DECISIONS = frozenset({"allow", "deny", "modify", "ask", "defer"}) - - -def normalize_decision(result: Any) -> tuple[str, str, dict]: - """Coerce a Guardian result's decision fields to safe types, never - raising. - - Reading `decision`/`reasoning`/`modifications` must never throw: - §6.3:146 requires failing closed on an undeterminable Guardian - intent, and an uncaught AttributeError exits 1, which the shell - hosts treat as proceed (fail-open). - - Returns (decision, reasoning, modifications): - - decision: stripped, lower-cased str; "" when the field is not a - usable string, which routes the caller to its unusable→DENY branch. - - reasoning: always a str (JSON repr if the Guardian sent a - non-string, so it can go into a `reason` field unchanged). - - modifications: always a dict ({} when the Guardian sent a - non-object, which makes a MODIFY with no usable mutation - substitute to DENY per §6.3:146). - """ - if not isinstance(result, dict): - return "", "", {} - raw_decision = result.get("decision") - decision = raw_decision.strip().lower() if isinstance(raw_decision, str) else "" - raw_reason = result.get("reasoning", "") - reasoning = (raw_reason if isinstance(raw_reason, str) - else json.dumps(raw_reason, default=str)) - raw_mods = result.get("modifications", {}) - modifications = raw_mods if isinstance(raw_mods, dict) else {} - return decision, reasoning, modifications - - -def modify_composition_violation(modifications: dict) -> str | None: - """Return a reason string if a MODIFY's `modifications` object violates - the §6.3 / modifications.json composition rules, else None. - - The two shapes are mutually exclusive (modifications.json `oneOf`): - wholesale `modified_content`, OR structured edits (`redactions` and/or - `parameter_overrides`) whose targets are DISJOINT — never both shapes, - never overlapping structured targets. Per §6.3:146 the receiver MUST - fail closed (DENY) rather than half-apply one branch.""" - if not isinstance(modifications, dict): - return None # caller treats empty/absent as no-usable-mutation - has_content = "modified_content" in modifications - has_redactions = "redactions" in modifications - has_overrides = "parameter_overrides" in modifications - redactions = modifications.get("redactions") - overrides = modifications.get("parameter_overrides") - if has_content and (has_redactions or has_overrides): - return ("modifications combines wholesale modified_content with " - "structured edits — the two shapes are mutually exclusive " - "(§6.3)") - if has_content and not isinstance(modifications.get("modified_content"), str): - return "modified_content is not a string — Guardian intent is unusable (§6.3)" - if has_redactions and not isinstance(redactions, list): - return "redactions is not an array — Guardian intent is unusable (§6.3)" - if has_overrides and not isinstance(overrides, dict): - return ("parameter_overrides is not an object — Guardian intent is " - "unusable (§6.3)") - - def _pointer_token(value: str) -> str: - """Encode one object-member name as an RFC 6901 token.""" - return value.replace("~", "~0").replace("/", "~1") - - def _paths_overlap(left: str, right: str) -> bool: - """True when equal paths or one JSON Pointer contains the other.""" - return (left == right or left == "" or right == "" - or left.startswith(right + "/") - or right.startswith(left + "/")) - - override_paths = { - f"/{_pointer_token(key)}": key - for key in (overrides or {}) - if isinstance(key, str) - } - if has_redactions: - for r in redactions: - if not isinstance(r, dict): - return ("redactions contains a non-object item — Guardian " - "intent is unusable (§6.3)") - path = r.get("path") - if not isinstance(path, str) or (path and not path.startswith("/")): - return (f"redaction path {path!r} is not an RFC 6901 JSON " - "Pointer — Guardian intent is unusable (§6.3)") - # JSON Pointer permits only ~0 and ~1 escape sequences. Reject a - # malformed pointer rather than compare a different path from the - # one the Guardian intended. - if re.search(r"~(?:[^01]|$)", path): - return (f"redaction path '{path}' has an invalid JSON Pointer " - "escape — Guardian intent is unusable (§6.3)") - if has_overrides: - for override_path, key in override_paths.items(): - if _paths_overlap(path, override_path): - return (f"redaction path '{path}' overlaps " - f"parameter_override '{key}' — structured " - "targets MUST be disjoint (§6.3)") - return None - - -# ----- JSON-RPC error code → audit cause label ----- -# -# Adapters use this when the Guardian returns a JSON-RPC `error` response -# (as opposed to a transport failure). Separating "Guardian rejected this -# envelope" from "I couldn't reach the Guardian" is load-bearing for -# operator triage — same fail-posture under §6.4, completely different -# remediation. Codes are the §17.1 / JSON-RPC reserved set. -GUARDIAN_ERROR_CAUSE: dict[int, str] = { - -32001: "unsupported_version_response", - -32002: "provenance_required_response", - -32004: "signature_invalid_response", # adapter or operator bug - -32005: "replay_detected_response", # duplicate request_id - -32006: "timestamp_out_of_window_response", # clock skew - -32600: "malformed_envelope_response", # non-conformant envelope - -32700: "parse_error_response", -} - - -def guardian_error_cause(code: int | None) -> str: - """Resolve a JSON-RPC error code to a stable audit cause label. - - Returns the generic 'guardian_error_response' for unrecognized codes - so audit consumers always have a non-empty cause string.""" - if code is None: - return "guardian_error_response" - return GUARDIAN_ERROR_CAUSE.get(code, "guardian_error_response") - - -# ----- Guardian refusals fail closed ----- -# -# DELIBERATELY STRICTER THAN v0.1 SPEC TEXT (tracked as spec issue #32). -# §6.4:156 as written routes ANY "error response instead of a decision" -# to the fail posture (default proceed). We do NOT do that for the codes -# below: the Guardian is ALIVE and REFUSED the envelope, and every one of -# them is attacker-reachable — oversize the body past the Guardian's cap -# (-32600 before policy runs), replay a request_id the adapter derives -# deterministically (-32005), strip or corrupt the signature (-32004), -# present a stale captured envelope (-32006). Routing those through a -# fail-OPEN posture turns each into a policy-bypass primitive. So we fail -# CLOSED on them regardless of ACS_DEFAULT_DENY / on_decision_failure. -# This is deliberate hardening ahead of the spec, NOT a claim that v0.1 -# mandates it — #32 proposes the normative text; until it lands, treat -# this as adapter behavior, not a conformance requirement. -GUARDIAN_REFUSAL_CODES = frozenset({ - -32000, # SESSION_REFUSED — policy refuses this session - -32004, # SIGNATURE_INVALID - -32005, # REPLAY_DETECTED - -32006, # TIMESTAMP_OUT_OF_WINDOW - -32600, # Invalid Request (malformed / oversized envelope) - -32700, # Parse error -}) - - -def is_guardian_refusal(code: int | None) -> bool: - """True when the error code means 'Guardian alive and refused this - envelope' — the adapter must fail closed, not apply the §6.4 posture.""" - return code in GUARDIAN_REFUSAL_CODES - - -# ----- Response↔request binding ----- - -def response_matches_request(request: dict, response: dict) -> bool: - """Bind a Guardian response to the request that elicited it. - - The per-session HMAC proves a response came from the Guardian, not - that it answers *this* request — a captured signed ALLOW replays - fine against a different call. Binding is two spec-carried checks: - - 1. JSON-RPC `id` on the response equals the request's `id`. - 2. `result.request_id` equals the request's `params.request_id` - (required on every result per response-envelope.json). - - Error responses carry no result; for those only check 1 applies. - Adapters MUST fail closed when this returns False.""" - if response.get("id") != request.get("id"): - return False - result = response.get("result") - if result is not None: - want = (request.get("params") or {}).get("request_id") - if result.get("request_id") != want: - return False - return True - - -# ----- Time + IDs ----- - -def iso8601_now() -> str: - return ( - datetime.datetime.now(datetime.timezone.utc) - .isoformat(timespec="milliseconds") - .replace("+00:00", "Z") - ) - - -def coerce_uuid(raw: str | None, *, namespace_prefix: str = "acs") -> str: - if not raw: - return str(uuid.uuid4()) - try: - return str(uuid.UUID(raw)) - except (ValueError, AttributeError, TypeError): - return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{namespace_prefix}:{raw}")) - - -def parse_iso8601(ts: str) -> datetime.datetime: - return datetime.datetime.fromisoformat(ts.replace("Z", "+00:00")) - - -# ----- Audit events (§6.4 fail-open recording) ----- - -def audit_event(event_type: str, **fields: Any) -> None: - """Emit a structured audit-event line to stderr and, when - `ACS_AUDIT_FILE` is set, append it to that file (created 0600). - - §6.4 requires every undecided proceed to be recorded; stderr alone - is not durable, so deployments SHOULD set ACS_AUDIT_FILE. Single-line - JSON for line-oriented ingestion. - """ - payload = { - "acs_audit_event": event_type, - "timestamp": iso8601_now(), - **fields, - } - line = "ACS_AUDIT " + json.dumps(payload, sort_keys=True) + "\n" - sys.stderr.write(line) - sys.stderr.flush() - audit_file = os.environ.get("ACS_AUDIT_FILE", "").strip() - if audit_file: - try: - fd = os.open(audit_file, - os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) - with os.fdopen(fd, "a") as f: - f.write(line) - except OSError as e: - # Never crash the hook path over the audit sink, but say so. - sys.stderr.write( - f"ACS_AUDIT_SINK_ERROR cannot append to {audit_file!r}: {e}\n") - sys.stderr.flush() - - -# ----- Handshake (§4) ----- - -_HANDSHAKE_CACHE_DIR = Path( - os.environ.get( - "ACS_HANDSHAKE_CACHE", - os.path.join(os.path.expanduser("~"), ".cache", "acs-adapter-handshake"), - ) -) - - -def _handshake_cache_path(session_id: str, guardian_url: str) -> Path: - # Full SHA-256 (no truncation) for the same reason as state files — - # avoid birthday collisions across the deployment's lifetime. - key = hashlib.sha256((session_id + "|" + guardian_url).encode()).hexdigest() - return _HANDSHAKE_CACHE_DIR / f"{key}.json" - - -# Cache TTL — default 1 hour. A Guardian config change (new -# skew_window_ms, new accepted profiles) propagates to adapters within -# this window. Override with ACS_HANDSHAKE_CACHE_TTL_SECONDS. -_HANDSHAKE_CACHE_TTL_S = int(os.environ.get("ACS_HANDSHAKE_CACHE_TTL_SECONDS", "3600")) - -# Failure cache TTL — default 30s: the first event against a dead -# Guardian pays the handshake timeout, later ones fail fast to the -# startup posture; short so a recovering Guardian is picked up quickly. -_HANDSHAKE_FAILURE_TTL_S = int( - os.environ.get("ACS_HANDSHAKE_FAILURE_CACHE_TTL_SECONDS", "30")) - -# Handshake network timeout — default 5s, configurable per deployment. -_HANDSHAKE_TIMEOUT_S = float(os.environ.get("ACS_HANDSHAKE_TIMEOUT_SECONDS", "5")) - - -def ensure_session_handshake( - *, - guardian_url: str, - session_id: str, - agent_id: str, - platform: str, - methods_implemented: list[str], - wrapped_protocols: list[str] | None = None, - timeout: float | None = None, - profiles_supported: list[str] | None = None, - transports_supported: list[str] | None = None, -) -> dict | None: - """Idempotently ensure a handshake/hello has happened for this session. - - Spec contract (§4): handshake is REQUIRED at session start, ONCE - per session, not per event. The shell-stdin adapters - (claude-code, cursor) spawn a fresh process per hook event, so we - persist the negotiated ServerHello in a small JSON file under - `~/.cache/acs-adapter-handshake/.json`: - - - First event of a session: cache miss → POSTs ClientHello, - receives ServerHello, writes cache file, returns ServerHello. - - Subsequent events same session: cache hit (file fresh, < 1h - old by default) → reads file, returns cached ServerHello. - NO network call. - - Cache files older than the TTL are ignored so operator - Guardian-config changes propagate. - - Returns the ServerHello (cached or freshly fetched), or None on - failure (Guardian unreachable, etc.) — adapters fall to their - startup posture in that case (§4.1). - - Partial adapters can explicitly advertise no profiles and only their - implemented transports. None preserves the existing adapter defaults. - - Hardening: - - The ServerHello signature is verified before caching or return. - - The whole signed envelope is cached and re-verified on every - read — the cache path is writable by the observed agent. - - The response is bound to the ClientHello by JSON-RPC id. - - Failures are negative-cached for a short TTL; every - negative-cache hit is audited (§4.1:77). - """ - if timeout is None: - timeout = _HANDSHAKE_TIMEOUT_S - cache = _handshake_cache_path(session_id, guardian_url) - failure_marker = cache.with_suffix(".failed") - if cache.exists(): - try: - mtime = cache.stat().st_mtime - if (time.time() - mtime) <= _HANDSHAKE_CACHE_TTL_S: - with open(cache) as f: - cached = json.load(f) - # Verify a cached envelope exactly like a fresh one: - # the file is writable by the observed agent, and an - # unverified read lets it flip its own fail posture (§10). - if not isinstance(cached, dict) or "result" not in cached: - # Pre-0.1.2 cache held the bare payload, with no - # signature to check. Treat as a miss and re-handshake. - raise ValueError("legacy unsigned handshake cache entry") - if not verify_signature(cached, session_id=session_id): - audit_event("handshake_cache_signature_invalid", - guardian_url=guardian_url, session_id=session_id, - detail="cached ServerHello failed verification; " - "re-handshaking") - raise ValueError("cached ServerHello signature invalid") - # handshake/hello returns ServerHello directly in result. - # A signed legacy AcsResult wrapper is still the wrong shape: - # treat it as a cache miss and negotiate again. - server_hello = cached.get("result") - if not isinstance(server_hello, dict) or "negotiated_version" not in server_hello: - raise ValueError("legacy or invalid cached ServerHello") - return server_hello - # Else: cache is stale, fall through to re-handshake - except (json.JSONDecodeError, OSError, ValueError): - pass - if failure_marker.exists(): - try: - if (time.time() - failure_marker.stat().st_mtime) <= _HANDSHAKE_FAILURE_TTL_S: - # Recent failure: fail fast, audited per §4.1:77 (an - # unguarded session must be recorded). - audit_event("handshake_negative_cache_hit", - guardian_url=guardian_url, session_id=session_id) - return None - except OSError: - pass - - client_hello = { - "jsonrpc": "2.0", - "id": str(uuid.uuid4()), - "method": "handshake/hello", - "params": { - "acs_version": ACS_VERSION, - "request_id": str(uuid.uuid4()), - "timestamp": iso8601_now(), - "metadata": {"agent_id": agent_id, "session_id": session_id, "platform": platform}, - "payload": { - "acs_versions_supported": [ACS_VERSION], - "methods_implemented": methods_implemented, - "transports_supported": (transports_supported if transports_supported - is not None else ["http", "stdio"]), - "max_payload_size_bytes": 1_000_000, - "provenance_producer": "none", - "wrapped_protocols": wrapped_protocols or [], - # An unsigned session does not meet acs-core's signed- - # envelope floor (§10), so it must not advertise the profile. - "profiles_supported": ( - (profiles_supported if profiles_supported is not None - else ["acs-core"]) if _signing_secret() else [] - ), - "signature_algorithms_supported": ( - ["HMAC-SHA256"] if _signing_secret() else [] - ), - }, - }, - } - sign_envelope(client_hello, session_id=session_id) - try: - validate_guardian_url(guardian_url) - except ValueError: - return None - def _record_failure(cause: str, **fields: Any) -> None: - audit_event("handshake_failed", cause=cause, - guardian_url=guardian_url, session_id=session_id, - **fields) - try: - _HANDSHAKE_CACHE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) - fd = os.open(str(failure_marker), - os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - os.close(fd) - except OSError: - pass - - try: - body = json.dumps(client_hello).encode("utf-8") - req = urllib.request.Request( - guardian_url, data=body, - headers={"Content-Type": "application/json"}, method="POST", - ) - with urllib.request.urlopen(req, timeout=timeout) as resp: - response = json.loads(resp.read().decode("utf-8")) - except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError, ValueError) as e: - _record_failure("transport_failure", error=str(e)) - return None - - # Bind the response to OUR ClientHello — a stray or replayed - # response envelope must not establish a session. - if response.get("id") != client_hello["id"]: - _record_failure("response_id_mismatch") - return None - - # An unverifiable ServerHello is a failure, not a Guardian: caching - # it would let a port squatter hand out its own negotiated config. - if not verify_signature(response, session_id=session_id): - _record_failure("server_hello_signature_invalid") - return None - - # ServerHello is the handshake result, not an AcsResult payload. - # Leave capability/posture validation to the adapter as before; this - # discriminator prevents an old wrapper from becoming a cached hello. - server_hello = response.get("result") - if isinstance(server_hello, dict) and "negotiated_version" in server_hello: - try: - _HANDSHAKE_CACHE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) - try: - os.chmod(_HANDSHAKE_CACHE_DIR, 0o700) - except OSError: - pass - fd = os.open(str(cache), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - with os.fdopen(fd, "w") as f: - # Store the whole signed response, not the bare result, so the - # read path can re-verify it (§10 binds the signature to the - # envelope, so a bare payload cannot be checked at all). - json.dump(response, f) - # A fresh success clears any lingering failure marker. - try: - failure_marker.unlink() - except OSError: - pass - except OSError: - pass - else: - _record_failure("invalid_server_hello") - return None - return server_hello - - -def method_evaluated(server_hello: dict | None, method: str) -> bool: - """Whether the negotiated Guardian evaluates `method`; handshake.json says - clients MUST treat methods missing from methods_evaluated as - ALLOW-by-default. With no negotiated list, every method counts as - evaluated.""" - if not isinstance(server_hello, dict): - return True - evaluated = server_hello.get("methods_evaluated") - if not isinstance(evaluated, list): - return True - return method in evaluated - - -# ----- system/ping (§13) ----- - -def ping(guardian_url: str, *, echo: str = "ping", timeout: float = 2.0) -> dict | None: - """Send a system/ping and return the result, or None on failure. - - Per §13: Guardian MUST always return allow; ping does not participate - in the chain; no signature required. - """ - request = { - "jsonrpc": "2.0", - "id": str(uuid.uuid4()), - "method": "system/ping", - "params": { - "acs_version": ACS_VERSION, - "request_id": str(uuid.uuid4()), - "timestamp": iso8601_now(), - "metadata": {"agent_id": "ping-client", "session_id": str(uuid.uuid4())}, - "payload": {"echo": echo}, - }, - } - try: - validate_guardian_url(guardian_url) - except ValueError: - return None - try: - body = json.dumps(request).encode("utf-8") - req = urllib.request.Request( - guardian_url, data=body, - headers={"Content-Type": "application/json"}, method="POST", - ) - with urllib.request.urlopen(req, timeout=timeout) as resp: - return json.loads(resp.read().decode("utf-8")).get("result") - except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError, ValueError): - return None - - -# ----- Per-session adapter state (cross-invocation) ----- -# -# Shell-stdin adapters (claude-code, cursor) spawn one process per hook -# event. To accumulate state across events in the same session (last -# step_id, step_ids seen, subagent registry, etc.) the adapter persists -# a small JSON file in the cache directory. - -_SESSION_STATE_DIR = Path( - os.environ.get( - "ACS_SESSION_STATE_DIR", - os.path.join(os.path.expanduser("~"), ".cache", "acs-adapter-session"), - ) -) - - -def _session_state_path(session_id: str, *, workspace: str | None = None) -> Path: - """Path to the per-session state file. - - Hash key is full 64-char SHA-256 (not [:16]) to eliminate birthday - collisions over the lifetime of a deployment. When `workspace` is - given, it is folded into the hash so two clients with the same - session_id but different workspaces (e.g., two Cursor windows - using `conv-default` as conversation_id) get distinct state files. - """ - if not session_id: - # Empty session_id — return a path that won't collide with anything real - digest = hashlib.sha256(b"empty").hexdigest() - return _SESSION_STATE_DIR / f"{digest}.json" - if workspace: - digest = hashlib.sha256( - (workspace + "\x00" + session_id).encode() - ).hexdigest() - else: - digest = hashlib.sha256(session_id.encode()).hexdigest() - return _SESSION_STATE_DIR / f"{digest}.json" - - -def load_session_state(session_id: str, *, workspace: str | None = None) -> dict: - """Return the session-state dict for `session_id`, or an empty dict. - - See `_session_state_path` for the workspace-namespacing rationale. - """ - if not session_id: - return {} - path = _session_state_path(session_id, workspace=workspace) - try: - with open(path) as f: - return json.load(f) - except (OSError, json.JSONDecodeError): - return {} - - -def save_session_state(session_id: str, state: dict, *, workspace: str | None = None) -> None: - """Persist the session-state dict atomically. No-op on session_id empty. - - The directory is created with mode 0700 and the file with mode 0600 - so other local users cannot read or poison adapter state. State - files contain step_id histories that an attacker could use to spoof - `parent_step_id` in subagentStart payloads — a security boundary - for the chain integrity properties of §8. - """ - if not session_id: - return - try: - _SESSION_STATE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) - # mkdir(exist_ok=True) does not chmod an existing dir — enforce explicitly - try: - os.chmod(_SESSION_STATE_DIR, 0o700) - except OSError: - pass - path = _session_state_path(session_id, workspace=workspace) - tmp = path.with_suffix(".json.tmp") - # Open with 0o600 from the start so the file is never group/world-readable - fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) - with os.fdopen(fd, "w") as f: - json.dump(state, f) - os.replace(tmp, path) - try: - os.chmod(path, 0o600) - except OSError: - pass - except OSError: - pass - - -def record_step(session_id: str, step_id: str, *, workspace: str | None = None) -> None: - """Append step_id to the session's seen-list and update last_step_id.""" - if not session_id or not step_id: - return - st = load_session_state(session_id, workspace=workspace) - seen = st.setdefault("seen_step_ids", []) - if step_id not in seen: - seen.append(step_id) - # Bound the list so it doesn't grow unbounded across long sessions - if len(seen) > 1000: - del seen[: len(seen) - 1000] - st["last_step_id"] = step_id - save_session_state(session_id, st, workspace=workspace) - - -# ----- Sys-path bootstrap for adapters in sibling directories ----- - -def install_path_for_sibling() -> None: - """Convenience: ensures the _common dir is on sys.path so adapters in - sibling directories can `from acs_common import ...`. No-op if - already on path.""" - here = str(Path(__file__).resolve().parent) - if here not in sys.path: - sys.path.insert(0, here) - - -# ----- Back-compat alias ----- -# `do_handshake` was the original name. Renamed to make the cache -# short-circuit visible at call sites. Old name kept so out-of-tree -# adapter forks aren't broken by the rename. -do_handshake = ensure_session_handshake diff --git a/adapters/src/acs/client/__init__.py b/adapters/src/acs/client/__init__.py new file mode 100644 index 00000000..6eb549fc --- /dev/null +++ b/adapters/src/acs/client/__init__.py @@ -0,0 +1,23 @@ +"""The adapter side's I/O: the configured secret, signing with it, the handshake, transport, session state and audit records.""" +from acs.client._audit import audit_event +from acs.client._handshake import ensure_session_handshake +from acs.client._secrets import SecretFilePermissionsError, SecretFileUnreadableError, load_hmac_secret +from acs.client._signing import sign_envelope, verify_signature +from acs.client._store import load_session_state, record_step, save_session_state +from acs.client._transport import ping +from acs.client._urls import validate_guardian_url + +__all__ = [ + "SecretFilePermissionsError", + "SecretFileUnreadableError", + "audit_event", + "ensure_session_handshake", + "load_hmac_secret", + "load_session_state", + "ping", + "record_step", + "save_session_state", + "sign_envelope", + "validate_guardian_url", + "verify_signature", +] diff --git a/adapters/src/acs/client/_audit.py b/adapters/src/acs/client/_audit.py new file mode 100644 index 00000000..97944d0e --- /dev/null +++ b/adapters/src/acs/client/_audit.py @@ -0,0 +1,41 @@ +"""Structured ACS_AUDIT records, on stderr and in ACS_AUDIT_FILE.""" +from __future__ import annotations + +import json +import os +import sys +from typing import Any + +from acs.core import iso8601_now + + +# ----- Audit events (§6.4 fail-open recording) ----- + +def audit_event(event_type: str, **fields: Any) -> None: + """Emit a structured audit-event line to stderr and, when + `ACS_AUDIT_FILE` is set, append it to that file (created 0600). + + §6.4 requires every undecided proceed to be recorded; stderr alone + is not durable, so deployments SHOULD set ACS_AUDIT_FILE. Single-line + JSON for line-oriented ingestion. + """ + payload = { + "acs_audit_event": event_type, + "timestamp": iso8601_now(), + **fields, + } + line = "ACS_AUDIT " + json.dumps(payload, sort_keys=True) + "\n" + sys.stderr.write(line) + sys.stderr.flush() + audit_file = os.environ.get("ACS_AUDIT_FILE", "").strip() + if audit_file: + try: + fd = os.open(audit_file, + os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + with os.fdopen(fd, "a") as f: + f.write(line) + except OSError as e: + # Never crash the hook path over the audit sink, but say so. + sys.stderr.write( + f"ACS_AUDIT_SINK_ERROR cannot append to {audit_file!r}: {e}\n") + sys.stderr.flush() diff --git a/adapters/src/acs/client/_handshake.py b/adapters/src/acs/client/_handshake.py new file mode 100644 index 00000000..66a5acff --- /dev/null +++ b/adapters/src/acs/client/_handshake.py @@ -0,0 +1,237 @@ +"""The §4 handshake: ClientHello, a verified ServerHello, and its per-session disk cache.""" +from __future__ import annotations + +import hashlib +import json +import os +import time +import urllib.error +import urllib.request +import uuid +from pathlib import Path +from typing import Any + +from acs.client._audit import audit_event +from acs.client._secrets import load_hmac_secret +from acs.client._signing import sign_envelope, verify_signature +from acs.client._urls import validate_guardian_url +from acs.core import ACS_VERSION, iso8601_now + + +# ----- Handshake (§4) ----- + +_HANDSHAKE_CACHE_DIR = Path( + os.environ.get( + "ACS_HANDSHAKE_CACHE", + os.path.join(os.path.expanduser("~"), ".cache", "acs-adapter-handshake"), + ) +) + + +def _handshake_cache_path(session_id: str, guardian_url: str) -> Path: + # Full SHA-256 (no truncation) for the same reason as state files — + # avoid birthday collisions across the deployment's lifetime. + key = hashlib.sha256((session_id + "|" + guardian_url).encode()).hexdigest() + return _HANDSHAKE_CACHE_DIR / f"{key}.json" + + +# Cache TTL — default 1 hour. A Guardian config change (new +# skew_window_ms, new accepted profiles) propagates to adapters within +# this window. Override with ACS_HANDSHAKE_CACHE_TTL_SECONDS. +_HANDSHAKE_CACHE_TTL_S = int(os.environ.get("ACS_HANDSHAKE_CACHE_TTL_SECONDS", "3600")) + +# Failure cache TTL — default 30s: the first event against a dead +# Guardian pays the handshake timeout, later ones fail fast to the +# startup posture; short so a recovering Guardian is picked up quickly. +_HANDSHAKE_FAILURE_TTL_S = int( + os.environ.get("ACS_HANDSHAKE_FAILURE_CACHE_TTL_SECONDS", "30")) + +# Handshake network timeout — default 5s, configurable per deployment. +_HANDSHAKE_TIMEOUT_S = float(os.environ.get("ACS_HANDSHAKE_TIMEOUT_SECONDS", "5")) + + +def ensure_session_handshake( + *, + guardian_url: str, + session_id: str, + agent_id: str, + platform: str, + methods_implemented: list[str], + wrapped_protocols: list[str] | None = None, + timeout: float | None = None, + profiles_supported: list[str] | None = None, + transports_supported: list[str] | None = None, +) -> dict | None: + """Idempotently ensure a handshake/hello has happened for this session. + + Spec contract (§4): handshake is REQUIRED at session start, ONCE + per session, not per event. The shell-stdin adapters + (claude-code, cursor) spawn a fresh process per hook event, so we + persist the negotiated ServerHello in a small JSON file under + `~/.cache/acs-adapter-handshake/.json`: + + - First event of a session: cache miss → POSTs ClientHello, + receives ServerHello, writes cache file, returns ServerHello. + - Subsequent events same session: cache hit (file fresh, < 1h + old by default) → reads file, returns cached ServerHello. + NO network call. + - Cache files older than the TTL are ignored so operator + Guardian-config changes propagate. + + Returns the ServerHello (cached or freshly fetched), or None on + failure (Guardian unreachable, etc.) — adapters fall to their + startup posture in that case (§4.1). + + Partial adapters can explicitly advertise no profiles and only their + implemented transports. None preserves the existing adapter defaults. + + Hardening: + - The ServerHello signature is verified before caching or return. + - The whole signed envelope is cached and re-verified on every + read — the cache path is writable by the observed agent. + - The response is bound to the ClientHello by JSON-RPC id. + - Failures are negative-cached for a short TTL; every + negative-cache hit is audited (§4.1:77). + """ + if timeout is None: + timeout = _HANDSHAKE_TIMEOUT_S + cache = _handshake_cache_path(session_id, guardian_url) + failure_marker = cache.with_suffix(".failed") + if cache.exists(): + try: + mtime = cache.stat().st_mtime + if (time.time() - mtime) <= _HANDSHAKE_CACHE_TTL_S: + with open(cache) as f: + cached = json.load(f) + # Verify a cached envelope exactly like a fresh one: + # the file is writable by the observed agent, and an + # unverified read lets it flip its own fail posture (§10). + if not isinstance(cached, dict) or "result" not in cached: + # Pre-0.1.2 cache held the bare payload, with no + # signature to check. Treat as a miss and re-handshake. + raise ValueError("legacy unsigned handshake cache entry") + if not verify_signature(cached, session_id=session_id): + audit_event("handshake_cache_signature_invalid", + guardian_url=guardian_url, session_id=session_id, + detail="cached ServerHello failed verification; " + "re-handshaking") + raise ValueError("cached ServerHello signature invalid") + # handshake/hello returns ServerHello directly in result. + # A signed legacy AcsResult wrapper is still the wrong shape: + # treat it as a cache miss and negotiate again. + server_hello = cached.get("result") + if not isinstance(server_hello, dict) or "negotiated_version" not in server_hello: + raise ValueError("legacy or invalid cached ServerHello") + return server_hello + # Else: cache is stale, fall through to re-handshake + except (json.JSONDecodeError, OSError, ValueError): + pass + if failure_marker.exists(): + try: + if (time.time() - failure_marker.stat().st_mtime) <= _HANDSHAKE_FAILURE_TTL_S: + # Recent failure: fail fast, audited per §4.1:77 (an + # unguarded session must be recorded). + audit_event("handshake_negative_cache_hit", + guardian_url=guardian_url, session_id=session_id) + return None + except OSError: + pass + + client_hello = { + "jsonrpc": "2.0", + "id": str(uuid.uuid4()), + "method": "handshake/hello", + "params": { + "acs_version": ACS_VERSION, + "request_id": str(uuid.uuid4()), + "timestamp": iso8601_now(), + "metadata": {"agent_id": agent_id, "session_id": session_id, "platform": platform}, + "payload": { + "acs_versions_supported": [ACS_VERSION], + "methods_implemented": methods_implemented, + "transports_supported": (transports_supported if transports_supported + is not None else ["http", "stdio"]), + "max_payload_size_bytes": 1_000_000, + "provenance_producer": "none", + "wrapped_protocols": wrapped_protocols or [], + # An unsigned session does not meet acs-core's signed- + # envelope floor (§10), so it must not advertise the profile. + "profiles_supported": ( + (profiles_supported if profiles_supported is not None + else ["acs-core"]) if load_hmac_secret() else [] + ), + "signature_algorithms_supported": ( + ["HMAC-SHA256"] if load_hmac_secret() else [] + ), + }, + }, + } + sign_envelope(client_hello, session_id=session_id) + try: + validate_guardian_url(guardian_url) + except ValueError: + return None + def _record_failure(cause: str, **fields: Any) -> None: + audit_event("handshake_failed", cause=cause, + guardian_url=guardian_url, session_id=session_id, + **fields) + try: + _HANDSHAKE_CACHE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) + fd = os.open(str(failure_marker), + os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + os.close(fd) + except OSError: + pass + + try: + body = json.dumps(client_hello).encode("utf-8") + req = urllib.request.Request( + guardian_url, data=body, + headers={"Content-Type": "application/json"}, method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + response = json.loads(resp.read().decode("utf-8")) + except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError, ValueError) as e: + _record_failure("transport_failure", error=str(e)) + return None + + # Bind the response to OUR ClientHello — a stray or replayed + # response envelope must not establish a session. + if response.get("id") != client_hello["id"]: + _record_failure("response_id_mismatch") + return None + + # An unverifiable ServerHello is a failure, not a Guardian: caching + # it would let a port squatter hand out its own negotiated config. + if not verify_signature(response, session_id=session_id): + _record_failure("server_hello_signature_invalid") + return None + + # ServerHello is the handshake result, not an AcsResult payload. + # Leave capability/posture validation to the adapter as before; this + # discriminator prevents an old wrapper from becoming a cached hello. + server_hello = response.get("result") + if isinstance(server_hello, dict) and "negotiated_version" in server_hello: + try: + _HANDSHAKE_CACHE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) + try: + os.chmod(_HANDSHAKE_CACHE_DIR, 0o700) + except OSError: + pass + fd = os.open(str(cache), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + # Store the whole signed response, not the bare result, so the + # read path can re-verify it (§10 binds the signature to the + # envelope, so a bare payload cannot be checked at all). + json.dump(response, f) + # A fresh success clears any lingering failure marker. + try: + failure_marker.unlink() + except OSError: + pass + except OSError: + pass + else: + _record_failure("invalid_server_hello") + return None + return server_hello diff --git a/adapters/src/acs/client/_secrets.py b/adapters/src/acs/client/_secrets.py new file mode 100644 index 00000000..915ee257 --- /dev/null +++ b/adapters/src/acs/client/_secrets.py @@ -0,0 +1,75 @@ +"""Loading the HMAC input keying material from ACS_HMAC_SECRET_FILE or ACS_HMAC_SECRET.""" +from __future__ import annotations + +import os +import stat + +from acs.client._audit import audit_event + + +class SecretFilePermissionsError(RuntimeError): + """ACS_HMAC_SECRET_FILE exists but its permissions / ownership leak the secret.""" + + +class SecretFileUnreadableError(RuntimeError): + """ACS_HMAC_SECRET_FILE is configured but cannot be read. + + Raised loudly: silently downgrading to unsigned mode would turn + off the only integrity control ACS-Core mandates.""" + + +def _check_secret_file_perms(path: str) -> None: + """Refuse to read the secret file if anything about its mode or + ownership would expose the key to another local user.""" + # Reject symlinks — a symlink is an attack vector (replace target + # without changing the visible path). + if os.path.islink(path): + raise SecretFilePermissionsError( + f"ACS_HMAC_SECRET_FILE {path!r} is a symlink; refusing to follow") + st = os.stat(path) + mode = stat.S_IMODE(st.st_mode) + if mode & 0o077: + raise SecretFilePermissionsError( + f"ACS_HMAC_SECRET_FILE {path!r} mode {oct(mode)} is too permissive; " + f"must be 0600 or 0400 (no group/other access). " + f"Fix: chmod 600 {path}") + if hasattr(os, "geteuid") and st.st_uid != os.geteuid(): + raise SecretFilePermissionsError( + f"ACS_HMAC_SECRET_FILE {path!r} owned by uid {st.st_uid}, " + f"adapter is running as uid {os.geteuid()}; refusing") + + +def load_hmac_secret() -> bytes: + """Read the HMAC input keying material. + + Resolution order (first hit wins): + 1. `ACS_HMAC_SECRET_FILE` — path to a file containing the secret. + Preferred for production. Permissions MUST be 0600 (or 0400) and + the file MUST be owned by the running user. The file's content + (stripped of trailing whitespace) is the secret. Symlinks are + rejected. Insecure permissions raise SecretFilePermissionsError — + the adapter refuses to use a leaked secret rather than silently + proceed. + 2. `ACS_HMAC_SECRET` — env-var fallback. Quick for dev, less secure + (visible in `ps eauxw`, child-process envs, core dumps). + 3. Empty bytes — caller decides whether that means dev-mode or fail. + + Generate a secret: `openssl rand -hex 32 > /etc/acs/hmac.key && chmod 600 /etc/acs/hmac.key` + """ + path = os.environ.get("ACS_HMAC_SECRET_FILE", "").strip() + if path: + try: + _check_secret_file_perms(path) + with open(path, "rb") as f: + return f.read().rstrip(b"\r\n\t ") + except SecretFilePermissionsError: + raise + except OSError as e: + # Configured-but-unreadable is an incident, not dev mode. + audit_event("hmac_secret_file_unreadable", path=path, error=str(e)) + raise SecretFileUnreadableError( + f"ACS_HMAC_SECRET_FILE {path!r} is configured but cannot " + f"be read ({e}); refusing to continue unsigned" + ) from e + env_val = os.environ.get("ACS_HMAC_SECRET", "") + return env_val.encode("utf-8") if env_val else b"" diff --git a/adapters/src/acs/client/_signing.py b/adapters/src/acs/client/_signing.py new file mode 100644 index 00000000..cc8eb515 --- /dev/null +++ b/adapters/src/acs/client/_signing.py @@ -0,0 +1,73 @@ +"""Signing and verifying with the configured secret, and the one-time unsigned-mode warning.""" +from __future__ import annotations + +from acs.client._audit import audit_event +from acs.client._secrets import load_hmac_secret +from acs.core import container_key, derive_session_key, sign_with_key, verify_with_key + + +_warned_unsigned_mode = False + + +def _warn_unsigned_once() -> None: + """One loud warning per process when operating without key material. + + Unsigned mode is an authentication failure, not just a + confidentiality one: any local process that binds the Guardian port + first becomes the Guardian and allows everything. + Fine for a first local experiment; never fine silently.""" + global _warned_unsigned_mode + if not _warned_unsigned_mode: + _warned_unsigned_mode = True + audit_event( + "unsigned_mode", + detail=( + "no ACS_HMAC_SECRET / ACS_HMAC_SECRET_FILE configured — " + "envelopes are UNSIGNED and any local process that binds " + "the Guardian address is trusted" + ), + ) + + +def sign_envelope(envelope: dict, *, key: bytes | None = None, + session_id: str | None = None, key_id: str = "default") -> dict: + """Add a signature to the envelope. Returns the envelope unchanged + (after a loud one-time audit event) if no key material is available.""" + if key is None: + ikm = load_hmac_secret() + if not ikm: + _warn_unsigned_once() + return envelope + if not session_id: + params = envelope.get("params") or envelope.get("result") or {} + meta = params.get("metadata") or {} + session_id = meta.get("session_id") or params.get("request_id") or "" + key = derive_session_key(ikm, session_id) + return sign_with_key(envelope, key, key_id=key_id) + + +def verify_signature(envelope: dict, *, key: bytes | None = None, + session_id: str | None = None) -> bool: + """Verify a signed envelope (request, decision response, or error + response). Returns True if valid (or if no signature present and no + key material configured — local-dev mode).""" + container = envelope.get(container_key(envelope)) or {} + sig = container.get("signature") + if sig is None: + # No signature on the wire; valid only if local-dev (no key configured) + return not bool(load_hmac_secret()) + + if sig.get("algorithm") != "HMAC-SHA256": + return False + if not sig.get("value"): + return False + + if key is None: + ikm = load_hmac_secret() + if not ikm: + # Signature present but no key configured: cannot verify + return False + meta = container.get("metadata") or {} + session_id = session_id or meta.get("session_id") or container.get("request_id") or "" + key = derive_session_key(ikm, session_id) + return verify_with_key(envelope, key) diff --git a/adapters/src/acs/client/_store.py b/adapters/src/acs/client/_store.py new file mode 100644 index 00000000..826aeeef --- /dev/null +++ b/adapters/src/acs/client/_store.py @@ -0,0 +1,107 @@ +"""Per-session adapter state persisted across hook processes.""" +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path + + +# ----- Per-session adapter state (cross-invocation) ----- +# +# Shell-stdin adapters (claude-code, cursor) spawn one process per hook +# event. To accumulate state across events in the same session (last +# step_id, step_ids seen, subagent registry, etc.) the adapter persists +# a small JSON file in the cache directory. + +_SESSION_STATE_DIR = Path( + os.environ.get( + "ACS_SESSION_STATE_DIR", + os.path.join(os.path.expanduser("~"), ".cache", "acs-adapter-session"), + ) +) + + +def _session_state_path(session_id: str, *, workspace: str | None = None) -> Path: + """Path to the per-session state file. + + Hash key is full 64-char SHA-256 (not [:16]) to eliminate birthday + collisions over the lifetime of a deployment. When `workspace` is + given, it is folded into the hash so two clients with the same + session_id but different workspaces (e.g., two Cursor windows + using `conv-default` as conversation_id) get distinct state files. + """ + if not session_id: + # Empty session_id — return a path that won't collide with anything real + digest = hashlib.sha256(b"empty").hexdigest() + return _SESSION_STATE_DIR / f"{digest}.json" + if workspace: + digest = hashlib.sha256( + (workspace + "\x00" + session_id).encode() + ).hexdigest() + else: + digest = hashlib.sha256(session_id.encode()).hexdigest() + return _SESSION_STATE_DIR / f"{digest}.json" + + +def load_session_state(session_id: str, *, workspace: str | None = None) -> dict: + """Return the session-state dict for `session_id`, or an empty dict. + + See `_session_state_path` for the workspace-namespacing rationale. + """ + if not session_id: + return {} + path = _session_state_path(session_id, workspace=workspace) + try: + with open(path) as f: + return json.load(f) + except (OSError, json.JSONDecodeError): + return {} + + +def save_session_state(session_id: str, state: dict, *, workspace: str | None = None) -> None: + """Persist the session-state dict atomically. No-op on session_id empty. + + The directory is created with mode 0700 and the file with mode 0600 + so other local users cannot read or poison adapter state. State + files contain step_id histories that an attacker could use to spoof + `parent_step_id` in subagentStart payloads — a security boundary + for the chain integrity properties of §8. + """ + if not session_id: + return + try: + _SESSION_STATE_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) + # mkdir(exist_ok=True) does not chmod an existing dir — enforce explicitly + try: + os.chmod(_SESSION_STATE_DIR, 0o700) + except OSError: + pass + path = _session_state_path(session_id, workspace=workspace) + tmp = path.with_suffix(".json.tmp") + # Open with 0o600 from the start so the file is never group/world-readable + fd = os.open(str(tmp), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + json.dump(state, f) + os.replace(tmp, path) + try: + os.chmod(path, 0o600) + except OSError: + pass + except OSError: + pass + + +def record_step(session_id: str, step_id: str, *, workspace: str | None = None) -> None: + """Append step_id to the session's seen-list and update last_step_id.""" + if not session_id or not step_id: + return + st = load_session_state(session_id, workspace=workspace) + seen = st.setdefault("seen_step_ids", []) + if step_id not in seen: + seen.append(step_id) + # Bound the list so it doesn't grow unbounded across long sessions + if len(seen) > 1000: + del seen[: len(seen) - 1000] + st["last_step_id"] = step_id + save_session_state(session_id, st, workspace=workspace) diff --git a/adapters/src/acs/client/_transport.py b/adapters/src/acs/client/_transport.py new file mode 100644 index 00000000..5a060026 --- /dev/null +++ b/adapters/src/acs/client/_transport.py @@ -0,0 +1,46 @@ +"""system/ping (§13).""" +from __future__ import annotations + +import json +import urllib.error +import urllib.request +import uuid + +from acs.client._urls import validate_guardian_url +from acs.core import ACS_VERSION, iso8601_now + + +# ----- system/ping (§13) ----- + +def ping(guardian_url: str, *, echo: str = "ping", timeout: float = 2.0) -> dict | None: + """Send a system/ping and return the result, or None on failure. + + Per §13: Guardian MUST always return allow; ping does not participate + in the chain; no signature required. + """ + request = { + "jsonrpc": "2.0", + "id": str(uuid.uuid4()), + "method": "system/ping", + "params": { + "acs_version": ACS_VERSION, + "request_id": str(uuid.uuid4()), + "timestamp": iso8601_now(), + "metadata": {"agent_id": "ping-client", "session_id": str(uuid.uuid4())}, + "payload": {"echo": echo}, + }, + } + try: + validate_guardian_url(guardian_url) + except ValueError: + return None + try: + body = json.dumps(request).encode("utf-8") + req = urllib.request.Request( + guardian_url, data=body, + headers={"Content-Type": "application/json"}, method="POST", + ) + with urllib.request.urlopen(req, timeout=timeout) as resp: + return json.loads(resp.read().decode("utf-8")).get("result") + except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError, ValueError): + return None diff --git a/adapters/src/acs/client/_urls.py b/adapters/src/acs/client/_urls.py new file mode 100644 index 00000000..181bfa24 --- /dev/null +++ b/adapters/src/acs/client/_urls.py @@ -0,0 +1,39 @@ +"""Guardian URL checks: an http(s) scheme, and the optional host allowlist.""" +from __future__ import annotations + +import os + + +# ----- URL scheme allowlist (defends against SSRF / file:// disclosure) ----- + +_ALLOWED_GUARDIAN_SCHEMES = frozenset({"http", "https"}) + + +def validate_guardian_url(url: str) -> None: + """Reject Guardian URLs whose scheme is not http/https. + + urllib.request.urlopen happily accepts file://, ftp://, data://, etc. + An attacker who controls ACS_GUARDIAN_URL could use file:// to read + arbitrary files the adapter user has access to, or data:// to feed + a crafted response. The adapter and any other code POSTing to the + Guardian MUST call this before urlopen. + + Optionally also restricts the hostname against an operator-provided + `ACS_GUARDIAN_HOST_ALLOWLIST` (comma-separated). Defense in depth + against env-var attacks that smuggle a real http:// URL to an + internal service the adapter shouldn't reach. + """ + from urllib.parse import urlparse + parsed = urlparse(url) + if parsed.scheme.lower() not in _ALLOWED_GUARDIAN_SCHEMES: + raise ValueError( + f"Guardian URL scheme {parsed.scheme!r} not allowed; " + f"only {sorted(_ALLOWED_GUARDIAN_SCHEMES)} permitted") + allow = os.environ.get("ACS_GUARDIAN_HOST_ALLOWLIST", "").strip() + if allow: + allowed_hosts = {h.strip().lower() for h in allow.split(",") if h.strip()} + host = (parsed.hostname or "").lower() + if host not in allowed_hosts: + raise ValueError( + f"Guardian host {host!r} not in ACS_GUARDIAN_HOST_ALLOWLIST " + f"({sorted(allowed_hosts)})") diff --git a/adapters/src/acs/core/__init__.py b/adapters/src/acs/core/__init__.py index 87c0fa0e..a0ae989d 100644 --- a/adapters/src/acs/core/__init__.py +++ b/adapters/src/acs/core/__init__.py @@ -1,4 +1,45 @@ -"""Protocol logic with no I/O: envelopes, signing, decisions and the ACS schemas.""" +"""Protocol logic with no I/O: envelopes, signing, decisions, error codes and the ACS schemas.""" +from acs.core._canonical import jcs_canonicalize +from acs.core._decision import KNOWN_DECISIONS, modify_composition_violation, normalize_decision +from acs.core._envelope import ( + ACS_VERSION, + DEFAULT_SKEW_WINDOW_MS, + MAX_REQUEST_BODY_BYTES, + coerce_uuid, + iso8601_now, + parse_iso8601, + response_matches_request, +) +from acs.core._errors import ( + GUARDIAN_ERROR_CAUSE, + GUARDIAN_REFUSAL_CODES, + guardian_error_cause, + is_guardian_refusal, +) +from acs.core._handshake import method_evaluated from acs.core._schemas import schemas_dir +from acs.core._signing import container_key, derive_session_key, sign_with_key, verify_with_key -__all__ = ["schemas_dir"] +__all__ = [ + "ACS_VERSION", + "DEFAULT_SKEW_WINDOW_MS", + "GUARDIAN_ERROR_CAUSE", + "GUARDIAN_REFUSAL_CODES", + "KNOWN_DECISIONS", + "MAX_REQUEST_BODY_BYTES", + "coerce_uuid", + "container_key", + "derive_session_key", + "guardian_error_cause", + "is_guardian_refusal", + "iso8601_now", + "jcs_canonicalize", + "method_evaluated", + "modify_composition_violation", + "normalize_decision", + "parse_iso8601", + "response_matches_request", + "schemas_dir", + "sign_with_key", + "verify_with_key", +] diff --git a/adapters/src/acs/core/_canonical.py b/adapters/src/acs/core/_canonical.py new file mode 100644 index 00000000..1cddc743 --- /dev/null +++ b/adapters/src/acs/core/_canonical.py @@ -0,0 +1,30 @@ +"""RFC 8785 (JCS) canonicalization, the only form §10 signs.""" +from __future__ import annotations + +from typing import Any + + +# ----- Canonicalization ----- + +try: + import rfc8785 as _rfc8785 # type: ignore[import-not-found] +except ImportError as _rfc8785_import_error: + # HARD dependency, deliberately. Spec §10: "Alternative + # canonicalization is not permitted in v0.1." A sorted-keys + # json.dumps fallback diverges from RFC 8785 on floats (1.0 vs 1), + # -0.0, large exponents, and non-BMP key ordering. The failure mode + # is a silent policy bypass (divergent canonicalization → + # SIGNATURE_INVALID → fail posture), so refuse to import instead. + raise ImportError( + "ACS adapters require the `rfc8785` package for RFC 8785 (JCS) " + "canonical signing — spec §10 permits no alternative " + "canonicalization. Install it: pip install rfc8785" + ) from _rfc8785_import_error + + +def jcs_canonicalize(obj: Any) -> bytes: + """RFC 8785 (JSON Canonicalization Scheme) via the `rfc8785` package. + + A hard dependency — see the import block above for why there is + deliberately no fallback.""" + return _rfc8785.dumps(obj) diff --git a/adapters/src/acs/core/_decision.py b/adapters/src/acs/core/_decision.py new file mode 100644 index 00000000..519ea35e --- /dev/null +++ b/adapters/src/acs/core/_decision.py @@ -0,0 +1,110 @@ +"""Reading a Guardian's decision safely, and the §6.3 MODIFY composition rules.""" +from __future__ import annotations + +import json +import re +from typing import Any + + +# ----- Safe decision normalization ----- + +# The five ACS dispositions (§6.3). Shared so every adapter classifies an +# arrived verdict against the same set. +KNOWN_DECISIONS = frozenset({"allow", "deny", "modify", "ask", "defer"}) + + +def normalize_decision(result: Any) -> tuple[str, str, dict]: + """Coerce a Guardian result's decision fields to safe types, never + raising. + + Reading `decision`/`reasoning`/`modifications` must never throw: + §6.3:146 requires failing closed on an undeterminable Guardian + intent, and an uncaught AttributeError exits 1, which the shell + hosts treat as proceed (fail-open). + + Returns (decision, reasoning, modifications): + - decision: stripped, lower-cased str; "" when the field is not a + usable string, which routes the caller to its unusable→DENY branch. + - reasoning: always a str (JSON repr if the Guardian sent a + non-string, so it can go into a `reason` field unchanged). + - modifications: always a dict ({} when the Guardian sent a + non-object, which makes a MODIFY with no usable mutation + substitute to DENY per §6.3:146). + """ + if not isinstance(result, dict): + return "", "", {} + raw_decision = result.get("decision") + decision = raw_decision.strip().lower() if isinstance(raw_decision, str) else "" + raw_reason = result.get("reasoning", "") + reasoning = (raw_reason if isinstance(raw_reason, str) + else json.dumps(raw_reason, default=str)) + raw_mods = result.get("modifications", {}) + modifications = raw_mods if isinstance(raw_mods, dict) else {} + return decision, reasoning, modifications + + +def modify_composition_violation(modifications: dict) -> str | None: + """Return a reason string if a MODIFY's `modifications` object violates + the §6.3 / modifications.json composition rules, else None. + + The two shapes are mutually exclusive (modifications.json `oneOf`): + wholesale `modified_content`, OR structured edits (`redactions` and/or + `parameter_overrides`) whose targets are DISJOINT — never both shapes, + never overlapping structured targets. Per §6.3:146 the receiver MUST + fail closed (DENY) rather than half-apply one branch.""" + if not isinstance(modifications, dict): + return None # caller treats empty/absent as no-usable-mutation + has_content = "modified_content" in modifications + has_redactions = "redactions" in modifications + has_overrides = "parameter_overrides" in modifications + redactions = modifications.get("redactions") + overrides = modifications.get("parameter_overrides") + if has_content and (has_redactions or has_overrides): + return ("modifications combines wholesale modified_content with " + "structured edits — the two shapes are mutually exclusive " + "(§6.3)") + if has_content and not isinstance(modifications.get("modified_content"), str): + return "modified_content is not a string — Guardian intent is unusable (§6.3)" + if has_redactions and not isinstance(redactions, list): + return "redactions is not an array — Guardian intent is unusable (§6.3)" + if has_overrides and not isinstance(overrides, dict): + return ("parameter_overrides is not an object — Guardian intent is " + "unusable (§6.3)") + + def _pointer_token(value: str) -> str: + """Encode one object-member name as an RFC 6901 token.""" + return value.replace("~", "~0").replace("/", "~1") + + def _paths_overlap(left: str, right: str) -> bool: + """True when equal paths or one JSON Pointer contains the other.""" + return (left == right or left == "" or right == "" + or left.startswith(right + "/") + or right.startswith(left + "/")) + + override_paths = { + f"/{_pointer_token(key)}": key + for key in (overrides or {}) + if isinstance(key, str) + } + if has_redactions: + for r in redactions: + if not isinstance(r, dict): + return ("redactions contains a non-object item — Guardian " + "intent is unusable (§6.3)") + path = r.get("path") + if not isinstance(path, str) or (path and not path.startswith("/")): + return (f"redaction path {path!r} is not an RFC 6901 JSON " + "Pointer — Guardian intent is unusable (§6.3)") + # JSON Pointer permits only ~0 and ~1 escape sequences. Reject a + # malformed pointer rather than compare a different path from the + # one the Guardian intended. + if re.search(r"~(?:[^01]|$)", path): + return (f"redaction path '{path}' has an invalid JSON Pointer " + "escape — Guardian intent is unusable (§6.3)") + if has_overrides: + for override_path, key in override_paths.items(): + if _paths_overlap(path, override_path): + return (f"redaction path '{path}' overlaps " + f"parameter_override '{key}' — structured " + "targets MUST be disjoint (§6.3)") + return None diff --git a/adapters/src/acs/core/_envelope.py b/adapters/src/acs/core/_envelope.py new file mode 100644 index 00000000..36c200e8 --- /dev/null +++ b/adapters/src/acs/core/_envelope.py @@ -0,0 +1,62 @@ +"""Envelope constants, response binding, timestamps and IDs.""" +from __future__ import annotations + +import datetime +import uuid + + +ACS_VERSION = "0.1.0" +DEFAULT_SKEW_WINDOW_MS = 300_000 + +# Maximum bytes the Guardian will read from a single HTTP POST body. +# Matches the handshake's max_payload_size_bytes default. Defends against +# memory exhaustion via a huge Content-Length. +MAX_REQUEST_BODY_BYTES = 1_048_576 # 1 MiB + + +# ----- Response↔request binding ----- + +def response_matches_request(request: dict, response: dict) -> bool: + """Bind a Guardian response to the request that elicited it. + + The per-session HMAC proves a response came from the Guardian, not + that it answers *this* request — a captured signed ALLOW replays + fine against a different call. Binding is two spec-carried checks: + + 1. JSON-RPC `id` on the response equals the request's `id`. + 2. `result.request_id` equals the request's `params.request_id` + (required on every result per response-envelope.json). + + Error responses carry no result; for those only check 1 applies. + Adapters MUST fail closed when this returns False.""" + if response.get("id") != request.get("id"): + return False + result = response.get("result") + if result is not None: + want = (request.get("params") or {}).get("request_id") + if result.get("request_id") != want: + return False + return True + + +# ----- Time + IDs ----- + +def iso8601_now() -> str: + return ( + datetime.datetime.now(datetime.timezone.utc) + .isoformat(timespec="milliseconds") + .replace("+00:00", "Z") + ) + + +def coerce_uuid(raw: str | None, *, namespace_prefix: str = "acs") -> str: + if not raw: + return str(uuid.uuid4()) + try: + return str(uuid.UUID(raw)) + except (ValueError, AttributeError, TypeError): + return str(uuid.uuid5(uuid.NAMESPACE_URL, f"{namespace_prefix}:{raw}")) + + +def parse_iso8601(ts: str) -> datetime.datetime: + return datetime.datetime.fromisoformat(ts.replace("Z", "+00:00")) diff --git a/adapters/src/acs/core/_errors.py b/adapters/src/acs/core/_errors.py new file mode 100644 index 00000000..851eb0e2 --- /dev/null +++ b/adapters/src/acs/core/_errors.py @@ -0,0 +1,63 @@ +"""JSON-RPC error codes: audit cause labels, and the refusals that fail closed.""" +from __future__ import annotations + + + + +# ----- JSON-RPC error code → audit cause label ----- +# +# Adapters use this when the Guardian returns a JSON-RPC `error` response +# (as opposed to a transport failure). Separating "Guardian rejected this +# envelope" from "I couldn't reach the Guardian" is load-bearing for +# operator triage — same fail-posture under §6.4, completely different +# remediation. Codes are the §17.1 / JSON-RPC reserved set. +GUARDIAN_ERROR_CAUSE: dict[int, str] = { + -32001: "unsupported_version_response", + -32002: "provenance_required_response", + -32004: "signature_invalid_response", # adapter or operator bug + -32005: "replay_detected_response", # duplicate request_id + -32006: "timestamp_out_of_window_response", # clock skew + -32600: "malformed_envelope_response", # non-conformant envelope + -32700: "parse_error_response", +} + + +def guardian_error_cause(code: int | None) -> str: + """Resolve a JSON-RPC error code to a stable audit cause label. + + Returns the generic 'guardian_error_response' for unrecognized codes + so audit consumers always have a non-empty cause string.""" + if code is None: + return "guardian_error_response" + return GUARDIAN_ERROR_CAUSE.get(code, "guardian_error_response") + + +# ----- Guardian refusals fail closed ----- +# +# DELIBERATELY STRICTER THAN v0.1 SPEC TEXT (tracked as spec issue #32). +# §6.4:156 as written routes ANY "error response instead of a decision" +# to the fail posture (default proceed). We do NOT do that for the codes +# below: the Guardian is ALIVE and REFUSED the envelope, and every one of +# them is attacker-reachable — oversize the body past the Guardian's cap +# (-32600 before policy runs), replay a request_id the adapter derives +# deterministically (-32005), strip or corrupt the signature (-32004), +# present a stale captured envelope (-32006). Routing those through a +# fail-OPEN posture turns each into a policy-bypass primitive. So we fail +# CLOSED on them regardless of ACS_DEFAULT_DENY / on_decision_failure. +# This is deliberate hardening ahead of the spec, NOT a claim that v0.1 +# mandates it — #32 proposes the normative text; until it lands, treat +# this as adapter behavior, not a conformance requirement. +GUARDIAN_REFUSAL_CODES = frozenset({ + -32000, # SESSION_REFUSED — policy refuses this session + -32004, # SIGNATURE_INVALID + -32005, # REPLAY_DETECTED + -32006, # TIMESTAMP_OUT_OF_WINDOW + -32600, # Invalid Request (malformed / oversized envelope) + -32700, # Parse error +}) + + +def is_guardian_refusal(code: int | None) -> bool: + """True when the error code means 'Guardian alive and refused this + envelope' — the adapter must fail closed, not apply the §6.4 posture.""" + return code in GUARDIAN_REFUSAL_CODES diff --git a/adapters/src/acs/core/_handshake.py b/adapters/src/acs/core/_handshake.py new file mode 100644 index 00000000..ac715f16 --- /dev/null +++ b/adapters/src/acs/core/_handshake.py @@ -0,0 +1,17 @@ +"""Reading a negotiated ServerHello.""" +from __future__ import annotations + + + + +def method_evaluated(server_hello: dict | None, method: str) -> bool: + """Whether the negotiated Guardian evaluates `method`; handshake.json says + clients MUST treat methods missing from methods_evaluated as + ALLOW-by-default. With no negotiated list, every method counts as + evaluated.""" + if not isinstance(server_hello, dict): + return True + evaluated = server_hello.get("methods_evaluated") + if not isinstance(evaluated, list): + return True + return method in evaluated diff --git a/adapters/src/acs/core/_signing.py b/adapters/src/acs/core/_signing.py new file mode 100644 index 00000000..36edebc1 --- /dev/null +++ b/adapters/src/acs/core/_signing.py @@ -0,0 +1,83 @@ +"""The HMAC-SHA256 baseline signature of §10: key derivation, signing and verification with a given key.""" +from __future__ import annotations + +import base64 +import binascii +import hashlib +import hmac + +from acs.core._canonical import jcs_canonicalize + + +# ----- Signing (HMAC-SHA256 baseline per §10) ----- + +def derive_session_key(input_key_material: bytes, session_id: str) -> bytes: + """HKDF-SHA256 with session_id as `info`, no salt. + + Spec §10: 'per-session HMAC key is HKDF-derived from deployment-provided + input keying material (a pre-shared secret, or a transport channel + binding such as a TLS exporter) together with the session_id.' + """ + # HKDF-Extract with empty salt + prk = hmac.new(b"\x00" * 32, input_key_material, hashlib.sha256).digest() + # HKDF-Expand to 32 bytes with info = session_id + info = session_id.encode("utf-8") + t = hmac.new(prk, info + b"\x01", hashlib.sha256).digest() + return t # one 32-byte block is enough for HMAC-SHA256 + + +def container_key(envelope: dict) -> str: + """Where the signature lives: params for requests, result for + decision responses, error for error responses (every response is + signed under ACS-Core — including errors, since a spoofable + unsigned error under a fail-open posture is an allow).""" + if "method" in envelope: + return "params" + if "error" in envelope: + return "error" + return "result" + + +def sign_with_key(envelope: dict, key: bytes, *, key_id: str = "default") -> dict: + """Add an HMAC-SHA256 signature made with `key` over JCS(envelope without its signature), in place.""" + ckey = container_key(envelope) + container = envelope.get(ckey, {}) + # Strip any existing signature before signing + unsigned_container = {k: v for k, v in container.items() if k != "signature"} + unsigned_envelope = {**envelope, ckey: unsigned_container} + + sig_bytes = hmac.new(key, jcs_canonicalize(unsigned_envelope), hashlib.sha256).digest() + container["signature"] = { + "algorithm": "HMAC-SHA256", + "value": base64.b64encode(sig_bytes).decode("ascii"), + "key_id": key_id, + } + envelope[ckey] = container + return envelope + + +def verify_with_key(envelope: dict, key: bytes) -> bool: + """True when the envelope carries an HMAC-SHA256 signature that `key` reproduces.""" + ckey = container_key(envelope) + container = envelope.get(ckey) or {} + sig = container.get("signature") + if sig is None: + return False + if sig.get("algorithm") != "HMAC-SHA256": + return False + expected_b64 = sig.get("value") + if not expected_b64: + return False + unsigned_container = {k: v for k, v in container.items() if k != "signature"} + unsigned_envelope = {**envelope, ckey: unsigned_container} + expected_bytes = hmac.new(key, jcs_canonicalize(unsigned_envelope), hashlib.sha256).digest() + # Malformed base64 (truncation, garbage chars, non-string) must NOT + # crash the request path — that turns a bad signature into a 500 / + # uncaught exception instead of the spec's SIGNATURE_INVALID + # (-32004) response. Return False so the caller emits the right + # error code and the audit event carries cause=signature_invalid_*. + try: + provided_bytes = base64.b64decode(expected_b64, validate=True) + except (binascii.Error, ValueError, TypeError): + return False + return hmac.compare_digest(expected_bytes, provided_bytes) diff --git a/adapters/src/acs/guardian/example_guardian.py b/adapters/src/acs/guardian/example_guardian.py index 46db691f..c02ab4b5 100755 --- a/adapters/src/acs/guardian/example_guardian.py +++ b/adapters/src/acs/guardian/example_guardian.py @@ -66,18 +66,15 @@ from pathlib import Path from typing import Any -from acs._common import ( # noqa: E402 +from acs.client import audit_event, load_hmac_secret, sign_envelope, verify_signature # noqa: E402 +from acs.core import ( # noqa: E402 ACS_VERSION, DEFAULT_SKEW_WINDOW_MS, - DESTRUCTIVE_SCAN_MAX_LEN, MAX_REQUEST_BODY_BYTES, derive_session_key, iso8601_now, jcs_canonicalize, - load_hmac_secret, parse_iso8601, - sign_envelope, - verify_signature, ) import datetime @@ -164,6 +161,35 @@ def _hmac_secret() -> bytes: re.compile(r"\bchmod\s+(-R\s+)?[0-7]*7{2,}[0-7]*\s+(/etc|/usr|/bin|/sbin)", re.IGNORECASE), ) + +# Maximum command string length to scan with destructive-pattern regexes. +# Real shell commands are tiny; longer inputs are either non-shell data +# routed through the wrong tool or a regex-DoS attempt. +DESTRUCTIVE_SCAN_MAX_LEN = 8 * 1024 # 8 KiB + +# ----- Bounded regex scanning (defends against regex DoS / input bombing) ----- + +def scan_destructive_bash_safely(cmd: str, *, max_len: int = DESTRUCTIVE_SCAN_MAX_LEN): + """Run destructive-pattern regex scanning ONLY if cmd is below max_len. + + Returns: + None — command is short and matched no destructive pattern + a re.Pattern — command matched (caller decides what to do) + "input_too_large" — command exceeds max_len; caller MUST treat as + suspicious and MUST NOT silently allow (skipping the scan is + not the same as the scan returning "safe"). + + Scans with DESTRUCTIVE_BASH_PATTERNS above, the one canonical pattern set. + """ + if len(cmd) > max_len: + audit_event("destructive_scan_skipped_oversized", + cmd_length=len(cmd), max_len=max_len) + return "input_too_large" + for pat in DESTRUCTIVE_BASH_PATTERNS: + if pat.search(cmd): + return pat + return None + PROTECTED_PATH_PREFIXES: tuple[str, ...] = ("/etc/", "/usr/", "/bin/", "/sbin/", "/boot/") # Methods this teaching policy acknowledges with unconditional allow. diff --git a/adapters/src/acs/integrations/claude_code/adapter.py b/adapters/src/acs/integrations/claude_code/adapter.py index ebd22884..5d724272 100755 --- a/adapters/src/acs/integrations/claude_code/adapter.py +++ b/adapters/src/acs/integrations/claude_code/adapter.py @@ -62,27 +62,29 @@ from acs import __version__ try: - from acs._common import ( # noqa: E402 + from acs.client import ( + audit_event, + ensure_session_handshake, + load_session_state, + save_session_state, + sign_envelope, + validate_guardian_url, + verify_signature, + ) + from acs.core import ( ACS_VERSION, MAX_REQUEST_BODY_BYTES, - audit_event, coerce_uuid, - ensure_session_handshake, guardian_error_cause, is_guardian_refusal, iso8601_now, - load_session_state, method_evaluated, modify_composition_violation, normalize_decision, response_matches_request, - save_session_state, - sign_envelope, - validate_guardian_url, - verify_signature, ) except ImportError as _bootstrap_error: - # acs_common hard-requires rfc8785 (§10 permits no alternative + # acs.core hard-requires rfc8785 (§10 permits no alternative # canonicalization), so a missing dependency kills this module before # main() runs. Claude Code proceeds on any nonzero exit except 2, so # dying here is an ungoverned step with no audit event (against diff --git a/adapters/src/acs/integrations/cursor/adapter.py b/adapters/src/acs/integrations/cursor/adapter.py index 73f5f8a9..4a1a02bf 100755 --- a/adapters/src/acs/integrations/cursor/adapter.py +++ b/adapters/src/acs/integrations/cursor/adapter.py @@ -69,28 +69,30 @@ from acs import __version__ try: - from acs._common import ( # noqa: E402 + from acs.client import ( + audit_event, + ensure_session_handshake, + load_session_state, + record_step, + save_session_state, + sign_envelope, + validate_guardian_url, + verify_signature, + ) + from acs.core import ( ACS_VERSION, MAX_REQUEST_BODY_BYTES, - audit_event, coerce_uuid, - ensure_session_handshake, guardian_error_cause, is_guardian_refusal, iso8601_now, - load_session_state, method_evaluated, modify_composition_violation, normalize_decision, - record_step, response_matches_request, - save_session_state, - sign_envelope, - validate_guardian_url, - verify_signature, ) except ImportError as _bootstrap_error: - # acs_common hard-requires rfc8785 (§10 permits no alternative + # acs.core hard-requires rfc8785 (§10 permits no alternative # canonicalization), so the import can fail before main() runs; a bare # exit 1 would skip the §6.4:158 audit and ignore ACS_DEFAULT_DENY, # so both are handled here with the standard library only. diff --git a/adapters/src/acs/integrations/nat/adapter.py b/adapters/src/acs/integrations/nat/adapter.py index d30defca..f67dd5db 100644 --- a/adapters/src/acs/integrations/nat/adapter.py +++ b/adapters/src/acs/integrations/nat/adapter.py @@ -63,10 +63,15 @@ ADAPTER_VERSION = __version__ try: - from acs._common import ( # noqa: E402 - MAX_REQUEST_BODY_BYTES, + from acs.client import ( audit_event, ensure_session_handshake, + sign_envelope, + validate_guardian_url, + verify_signature, + ) + from acs.core import ( + MAX_REQUEST_BODY_BYTES, guardian_error_cause, is_guardian_refusal, iso8601_now as _common_iso8601_now, @@ -74,12 +79,9 @@ modify_composition_violation, normalize_decision, response_matches_request, - sign_envelope, - validate_guardian_url, - verify_signature, ) except ImportError as _acs_import_error: - # acs_common hard-requires rfc8785 (§10 permits no alternative + # acs.core hard-requires rfc8785 (§10 permits no alternative # canonicalization); a failed import would leave NAT running ungoverned, # so fail loudly and re-raise to abort middleware registration. _banner = ( diff --git a/adapters/src/acs/testing/capture_guardian.py b/adapters/src/acs/testing/capture_guardian.py index 5ed1f28b..ffde6f40 100644 --- a/adapters/src/acs/testing/capture_guardian.py +++ b/adapters/src/acs/testing/capture_guardian.py @@ -46,7 +46,7 @@ # ---- INDEPENDENT signature verification ---- # -# Deliberately NOT acs_common's implementation — a shared bug would +# Deliberately NOT acs.core's implementation — a shared bug would # pass both sides. From-scratch HKDF-SHA256 + HMAC-SHA256 over RFC 8785, # pinned by a known-answer vector in tests/test_capture_guardian.py. @@ -237,7 +237,7 @@ def _validate_invariants(self, rec: CaptureRecord) -> None: # signature: INDEPENDENTLY recompute (system/ping exempt, §13). # Uses independent_verify() — a from-scratch HMAC/HKDF/JCS - # implementation, NOT the adapter's own acs_common code, so a + # implementation, NOT the adapter's own acs.core code, so a # shared signing+verifying bug cannot pass both sides. if rec.method != "system/ping": sid = meta.get("session_id") or "" diff --git a/adapters/src/acs/testing/harness.py b/adapters/src/acs/testing/harness.py index f7e459f2..8e4870e3 100644 --- a/adapters/src/acs/testing/harness.py +++ b/adapters/src/acs/testing/harness.py @@ -38,7 +38,8 @@ from pathlib import Path from typing import Any, Callable -from acs import _common as acs_common +from acs import client +from acs import core from acs.core import schemas_dir _HERE = Path(__file__).resolve().parent @@ -134,7 +135,7 @@ def make_envelope( "params": { "acs_version": "0.1.0", "request_id": request_id or str(uuid.uuid4()), - "timestamp": timestamp or acs_common.iso8601_now(), + "timestamp": timestamp or core.iso8601_now(), "metadata": { "agent_id": agent_id, "session_id": sid, @@ -145,8 +146,8 @@ def make_envelope( } if sign_with_secret is not None: secret = sign_with_secret.encode() if isinstance(sign_with_secret, str) else sign_with_secret - key = acs_common.derive_session_key(secret, sid) - acs_common.sign_envelope(env, key=key, session_id=sid) + key = core.derive_session_key(secret, sid) + client.sign_envelope(env, key=key, session_id=sid) return env @@ -414,9 +415,9 @@ def do_POST(self_h): # noqa: N802 # known pre-handshake. if method != "system/ping": sid = req.get("params", {}).get("metadata", {}).get("session_id", "") - key = acs_common.derive_session_key( + key = core.derive_session_key( guardian.hmac_secret.encode(), sid) - if not acs_common.verify_signature(req, key=key, session_id=sid): + if not client.verify_signature(req, key=key, session_id=sid): self_h._reply({ "jsonrpc": "2.0", "id": req.get("id"), "error": {"code": -32004, "message": "SIGNATURE_INVALID"}, @@ -433,9 +434,9 @@ def do_POST(self_h): # noqa: N802 "result": result_or_error} if guardian.sign_responses: sid = req["params"]["metadata"]["session_id"] - key = acs_common.derive_session_key( + key = core.derive_session_key( guardian.hmac_secret.encode(), sid) - acs_common.sign_envelope(resp, key=key, session_id=sid) + client.sign_envelope(resp, key=key, session_id=sid) self_h._reply(resp) def _reply(self_h, resp: dict): diff --git a/adapters/tests/common/test_capture_guardian.py b/adapters/tests/common/test_capture_guardian.py index 0014ae54..fbd81de9 100644 --- a/adapters/tests/common/test_capture_guardian.py +++ b/adapters/tests/common/test_capture_guardian.py @@ -13,7 +13,8 @@ class is flagged — envelope schema, payload schema, every invariant. from pathlib import Path -from acs import _common as acs_common # noqa: E402 +from acs import client # noqa: E402 +from acs import core # noqa: E402 from acs.testing.capture_guardian import CaptureGuardian # noqa: E402 SESSION = "11111111-1111-4111-8111-111111111111" @@ -26,7 +27,7 @@ def _signed_tool_call(secret: str, **overrides) -> dict: "params": { "acs_version": "0.1.0", "request_id": str(uuid.uuid4()), - "timestamp": acs_common.iso8601_now(), + "timestamp": core.iso8601_now(), "metadata": {"agent_id": "a", "session_id": SESSION, "platform": "test"}, "payload": {"tool": {"name": "Bash"}, @@ -38,16 +39,16 @@ def _signed_tool_call(secret: str, **overrides) -> dict: env["params"]["payload"] = overrides["payload"] if overrides.get("metadata") is not None: env["params"]["metadata"] = overrides["metadata"] - key = acs_common.derive_session_key(secret.encode(), + key = core.derive_session_key(secret.encode(), env["params"]["metadata"].get("session_id", "")) - acs_common.sign_envelope(env, key=key, + client.sign_envelope(env, key=key, session_id=env["params"]["metadata"].get("session_id", "")) return env class IndependentVerifierKnownAnswer(unittest.TestCase): """A frozen known-answer vector pins the oracle's independent - HKDF+HMAC+JCS implementation (it must not share acs_common's code, + HKDF+HMAC+JCS implementation (it must not share acs.core's code, where a shared bug would pass both sides).""" SECRET = b"known-answer-test-secret" @@ -94,18 +95,19 @@ def test_verify_rejects_wrong_key(self): "value": self.SIG_B64, "key_id": "default"} self.assertFalse(independent_verify(env, b"wrong-secret", self.SID)) - def test_independent_of_acs_common(self): - """The verifier must not delegate to acs_common — that's the - whole point. Cross-check: it AGREES with acs_common on a valid + def test_independent_of_acs_core(self): + """The verifier must not delegate to acs.core — that's the + whole point. Cross-check: it AGREES with acs.core on a valid envelope (two implementations, same answer) but is a distinct code path.""" - from acs import _common as acs_common + from acs import client + from acs import core from acs.testing.capture_guardian import independent_verify env = self._envelope() - key = acs_common.derive_session_key(self.SECRET, self.SID) - acs_common.sign_envelope(env, key=key, session_id=self.SID) + key = core.derive_session_key(self.SECRET, self.SID) + client.sign_envelope(env, key=key, session_id=self.SID) self.assertTrue(independent_verify(env, self.SECRET, self.SID), - "independent verifier and acs_common disagree on a valid envelope") + "independent verifier and acs.core disagree on a valid envelope") class OracleIsNotVacuous(unittest.TestCase): diff --git a/adapters/tests/common/test_edge_cases.py b/adapters/tests/common/test_edge_cases.py index 1df58abe..9c4371a0 100644 --- a/adapters/tests/common/test_edge_cases.py +++ b/adapters/tests/common/test_edge_cases.py @@ -21,7 +21,10 @@ import uuid from pathlib import Path -from acs import _common as acs_common # noqa: E402 +from acs import client # noqa: E402 +from acs import core # noqa: E402 +from acs.client import _handshake as client_handshake # noqa: E402 +from acs.client import _store as client_store # noqa: E402 from acs.testing.harness import launch_guardian # noqa: E402 @@ -49,18 +52,18 @@ class Item01_JcsConsistency(unittest.TestCase): def test_rfc8785_canonical_output_on_divergent_shapes(self) -> None: """Pin RFC 8785 §3 number serialization on the shapes that the deleted fallback mis-canonicalized.""" - from acs import _common as acs_common + from acs import core for sample, expected in self.DIVERGENT: with self.subTest(sample=sample): - self.assertEqual(acs_common.jcs_canonicalize(sample), expected) + self.assertEqual(core.jcs_canonicalize(sample), expected) def test_non_bmp_key_ordering_matches_utf16_rule(self) -> None: """RFC 8785 §3.2.3 sorts keys by UTF-16 code units: a non-BMP key (U+10000, surrogate pair starting 0xD800) sorts BEFORE U+FFFF. json.dumps sorts by code point and gets this backwards.""" - from acs import _common as acs_common + from acs import core sample = {"￿": 1, "\U00010000": 2} - out = acs_common.jcs_canonicalize(sample).decode("utf-8") + out = core.jcs_canonicalize(sample).decode("utf-8") self.assertLess(out.index("\U00010000"), out.index("￿"), f"UTF-16 ordering violated in {out!r}") @@ -105,7 +108,7 @@ def test_huge_bash_command_is_capped_not_scanned(self) -> None: "params": { "acs_version": "0.1.0", "request_id": str(uuid.uuid4()), - "timestamp": acs_common.iso8601_now(), + "timestamp": core.iso8601_now(), "metadata": {"agent_id": "t", "session_id": str(uuid.uuid4())}, "payload": {"tool": {"name": "Bash"}, "arguments": {"command": {"value": huge}}}, @@ -182,18 +185,18 @@ def test_fresh_cache_is_honored(self) -> None: with tempfile.TemporaryDirectory() as tmp: os.environ["ACS_HANDSHAKE_CACHE"] = tmp import importlib - importlib.reload(acs_common) + importlib.reload(client_handshake) url = "http://127.0.0.1:1/dead" fake_payload = {"negotiated_version": "0.1.0", "_synthetic": True, "methods_evaluated": [], "selected_transport": "http", "timeout_config": {"default_ms": 5000}} envelope = {"jsonrpc": "2.0", "id": "cache-test", "result": fake_payload} - cache_path = acs_common._handshake_cache_path("sess1", url) + cache_path = client_handshake._handshake_cache_path("sess1", url) cache_path.parent.mkdir(parents=True, exist_ok=True) cache_path.write_text(json.dumps(envelope)) # Recently-mtimed cache must be honored - result = acs_common.do_handshake( + result = client.ensure_session_handshake( guardian_url=url, session_id="sess1", agent_id="x", platform="t", methods_implemented=[], ) @@ -211,13 +214,13 @@ def test_legacy_bare_payload_cache_is_a_miss(self) -> None: with tempfile.TemporaryDirectory() as tmp: os.environ["ACS_HANDSHAKE_CACHE"] = tmp import importlib - importlib.reload(acs_common) + importlib.reload(client_handshake) url = "http://127.0.0.1:1/dead" - cache_path = acs_common._handshake_cache_path("sess-legacy", url) + cache_path = client_handshake._handshake_cache_path("sess-legacy", url) cache_path.parent.mkdir(parents=True, exist_ok=True) cache_path.write_text(json.dumps( {"negotiated_version": "0.1.0", "on_decision_failure": "proceed"})) - result = acs_common.do_handshake( + result = client.ensure_session_handshake( guardian_url=url, session_id="sess-legacy", agent_id="x", platform="t", methods_implemented=[], ) @@ -230,15 +233,15 @@ def test_stale_cache_is_ignored(self) -> None: with tempfile.TemporaryDirectory() as tmp: os.environ["ACS_HANDSHAKE_CACHE"] = tmp import importlib - importlib.reload(acs_common) + importlib.reload(client_handshake) url = "http://127.0.0.1:1/dead" - cache_path = acs_common._handshake_cache_path("sess2", url) + cache_path = client_handshake._handshake_cache_path("sess2", url) cache_path.parent.mkdir(parents=True, exist_ok=True) cache_path.write_text(json.dumps({"_synthetic": True})) # Backdate the cache file 2 hours two_hours_ago = time.time() - 2 * 3600 os.utime(cache_path, (two_hours_ago, two_hours_ago)) - result = acs_common.do_handshake( + result = client.ensure_session_handshake( guardian_url=url, session_id="sess2", agent_id="x", platform="t", methods_implemented=[], ) @@ -315,22 +318,22 @@ def _make_envelope(self, value): def test_emoji_round_trip(self) -> None: env = self._make_envelope("🚀 from agent — ✨") - key = acs_common.derive_session_key(b"test-secret", "sess") - acs_common.sign_envelope(env, key=key, session_id="sess") - self.assertTrue(acs_common.verify_signature(env, key=key, session_id="sess")) + key = core.derive_session_key(b"test-secret", "sess") + client.sign_envelope(env, key=key, session_id="sess") + self.assertTrue(client.verify_signature(env, key=key, session_id="sess")) def test_null_byte_round_trip(self) -> None: env = self._make_envelope("before\x00after") - key = acs_common.derive_session_key(b"test-secret", "sess") - acs_common.sign_envelope(env, key=key, session_id="sess") - self.assertTrue(acs_common.verify_signature(env, key=key, session_id="sess")) + key = core.derive_session_key(b"test-secret", "sess") + client.sign_envelope(env, key=key, session_id="sess") + self.assertTrue(client.verify_signature(env, key=key, session_id="sess")) def test_bmp_and_supplementary_planes_round_trip(self) -> None: # Hebrew (BMP) + emoji (supplementary plane) env = self._make_envelope("שלום 🌍 こんにちは") - key = acs_common.derive_session_key(b"test-secret", "sess") - acs_common.sign_envelope(env, key=key, session_id="sess") - self.assertTrue(acs_common.verify_signature(env, key=key, session_id="sess")) + key = core.derive_session_key(b"test-secret", "sess") + client.sign_envelope(env, key=key, session_id="sess") + self.assertTrue(client.verify_signature(env, key=key, session_id="sess")) # ===== #8: ISO 8601 parse resilience ===== @@ -357,7 +360,7 @@ class Item08_Iso8601Parse(unittest.TestCase): def test_good_timestamps_parse(self) -> None: for ts in self.GOOD: try: - acs_common.parse_iso8601(ts) + core.parse_iso8601(ts) except Exception as e: self.fail(f"valid timestamp {ts!r} failed to parse: {e}") @@ -365,7 +368,7 @@ def test_bad_timestamps_raise_value_error(self) -> None: for ts in self.BAD: with self.assertRaises((ValueError, AttributeError), msg=f"{ts!r} should be rejected"): - acs_common.parse_iso8601(ts) + core.parse_iso8601(ts) # ===== #9: ACS_GUARDIAN_HOST_ALLOWLIST ===== @@ -386,15 +389,15 @@ def tearDown(self) -> None: def test_allowlist_unset_accepts_any_http(self) -> None: os.environ.pop("ACS_GUARDIAN_HOST_ALLOWLIST", None) - acs_common.validate_guardian_url("http://127.0.0.1:8787/acs") - acs_common.validate_guardian_url("http://anything.example.com/acs") + client.validate_guardian_url("http://127.0.0.1:8787/acs") + client.validate_guardian_url("http://anything.example.com/acs") def test_allowlist_restricts_to_listed_hosts(self) -> None: os.environ["ACS_GUARDIAN_HOST_ALLOWLIST"] = "127.0.0.1,guardian.internal" - acs_common.validate_guardian_url("http://127.0.0.1:8787/acs") - acs_common.validate_guardian_url("https://guardian.internal/acs") + client.validate_guardian_url("http://127.0.0.1:8787/acs") + client.validate_guardian_url("https://guardian.internal/acs") with self.assertRaises(ValueError): - acs_common.validate_guardian_url("http://attacker.example.com/acs") + client.validate_guardian_url("http://attacker.example.com/acs") # ===== #10: Cursor session-state file collision across workspaces ===== @@ -407,8 +410,8 @@ class Item10_CursorStateCollision(unittest.TestCase): def test_same_session_id_different_workspace_distinct_paths(self) -> None: # Two distinct workspaces, same session_id - p1 = acs_common._session_state_path("conv-default", workspace="/a/work1") - p2 = acs_common._session_state_path("conv-default", workspace="/b/work2") + p1 = client_store._session_state_path("conv-default", workspace="/a/work1") + p2 = client_store._session_state_path("conv-default", workspace="/b/work2") self.assertNotEqual(p1, p2, f"FIX REQUIRED: session-state path collides across workspaces: {p1}") @@ -420,7 +423,7 @@ class Item12_StatePathHashLength(unittest.TestCase): birthday collisions become possible. Use full SHA-256 (64 chars).""" def test_state_path_uses_full_sha256(self) -> None: - p = acs_common._session_state_path("any-session-id") + p = client_store._session_state_path("any-session-id") # Path stem (filename without extension) must be 64 hex chars self.assertRegex(p.stem, r"^[0-9a-f]{64}$", f"session-state filename {p.name!r} uses short hash; " @@ -457,7 +460,7 @@ def _envelope(self, sid, rid): "method": "steps/sessionStart", "params": { "acs_version": "0.1.0", "request_id": rid, - "timestamp": acs_common.iso8601_now(), + "timestamp": core.iso8601_now(), "metadata": {"agent_id": "ha", "session_id": sid, "platform": "t"}, "payload": {}, }} @@ -517,11 +520,11 @@ def test_port_dies_with_the_returned_process(self) -> None: self._stop(proc) time.sleep(0.2) self.assertIsNone( - acs_common.ping(f"http://127.0.0.1:{port}/acs", timeout=1.0), + client.ping(f"http://127.0.0.1:{port}/acs", timeout=1.0), "port still answers after its owner was killed — the " "returned port was not bound by the returned process") self.assertIsNotNone( - acs_common.ping(f"http://127.0.0.1:{decoy_port}/acs", timeout=2.0)) + client.ping(f"http://127.0.0.1:{decoy_port}/acs", timeout=2.0)) finally: self._stop(decoy) self._stop(proc) @@ -542,7 +545,7 @@ def test_returned_port_answers_acs_ping(self) -> None: proc, port = launch_guardian(env=self._dev_env()) try: self.assertIsNone(proc.poll(), "returned process must be alive") - result = acs_common.ping(f"http://127.0.0.1:{port}/acs", timeout=2.0) + result = client.ping(f"http://127.0.0.1:{port}/acs", timeout=2.0) self.assertIsNotNone(result, "returned port must answer system/ping") finally: self._stop(proc) diff --git a/adapters/tests/common/test_methods_evaluated.py b/adapters/tests/common/test_methods_evaluated.py index 5d9c35cc..42c8ea0d 100644 --- a/adapters/tests/common/test_methods_evaluated.py +++ b/adapters/tests/common/test_methods_evaluated.py @@ -18,7 +18,8 @@ import uuid from pathlib import Path -from acs import _common as acs_common # noqa: E402 +from acs import client # noqa: E402 +from acs import core # noqa: E402 from acs.testing.harness import ProgrammableGuardian # noqa: E402 ADAPTERS = Path(__file__).resolve().parents[2] @@ -59,9 +60,9 @@ class SigningHandler(base): def _reply(self_h, resp: dict) -> None: if "error" in resp and guardian.received: sid = guardian.received[-1]["params"]["metadata"]["session_id"] - key = acs_common.derive_session_key( + key = core.derive_session_key( guardian.hmac_secret.encode(), sid) - acs_common.sign_envelope(resp, key=key, session_id=sid) + client.sign_envelope(resp, key=key, session_id=sid) super()._reply(resp) return SigningHandler diff --git a/adapters/tests/common/test_security.py b/adapters/tests/common/test_security.py index 456ea54b..4b3d3be3 100644 --- a/adapters/tests/common/test_security.py +++ b/adapters/tests/common/test_security.py @@ -1,5 +1,5 @@ """ -Security tests for the shared `acs_common` helpers and the Guardian +Security tests for the shared `acs.core` and `acs.client` helpers and the Guardian HTTP path. Each test names a specific attack and asserts the mitigation that defeats it. @@ -27,7 +27,11 @@ from pathlib import Path from unittest import mock -from acs import _common as acs_common # noqa: E402 +from acs import client # noqa: E402 +from acs import core # noqa: E402 +from acs.client import _handshake as client_handshake # noqa: E402 +from acs.client import _store as client_store # noqa: E402 +from acs.guardian import example_guardian # noqa: E402 from acs.testing.harness import launch_guardian # noqa: E402 @@ -44,26 +48,26 @@ class GuardianUrlSchemeAllowlist(unittest.TestCase): def test_file_scheme_rejected(self) -> None: with self.assertRaises(ValueError) as cm: - acs_common.validate_guardian_url("file:///etc/passwd") + client.validate_guardian_url("file:///etc/passwd") self.assertIn("scheme", str(cm.exception).lower()) def test_ftp_scheme_rejected(self) -> None: with self.assertRaises(ValueError): - acs_common.validate_guardian_url("ftp://example.com/x") + client.validate_guardian_url("ftp://example.com/x") def test_data_scheme_rejected(self) -> None: with self.assertRaises(ValueError): - acs_common.validate_guardian_url("data:text/plain,evil") + client.validate_guardian_url("data:text/plain,evil") def test_javascript_scheme_rejected(self) -> None: with self.assertRaises(ValueError): - acs_common.validate_guardian_url("javascript:alert(1)") + client.validate_guardian_url("javascript:alert(1)") def test_http_accepted(self) -> None: - acs_common.validate_guardian_url("http://127.0.0.1:8787/acs") # no raise + client.validate_guardian_url("http://127.0.0.1:8787/acs") # no raise def test_https_accepted(self) -> None: - acs_common.validate_guardian_url("https://guardian.internal/acs") + client.validate_guardian_url("https://guardian.internal/acs") # ----- 2. World-readable secret file ----- @@ -96,17 +100,17 @@ def tearDown(self) -> None: def test_mode_0600_accepted(self) -> None: os.chmod(self.secret_path, 0o600) # Returns secret bytes; no raise - self.assertTrue(acs_common.load_hmac_secret().startswith(b"super-secret")) + self.assertTrue(client.load_hmac_secret().startswith(b"super-secret")) def test_world_readable_rejected(self) -> None: os.chmod(self.secret_path, 0o644) - with self.assertRaises(acs_common.SecretFilePermissionsError): - acs_common.load_hmac_secret() + with self.assertRaises(client.SecretFilePermissionsError): + client.load_hmac_secret() def test_group_readable_rejected(self) -> None: os.chmod(self.secret_path, 0o640) - with self.assertRaises(acs_common.SecretFilePermissionsError): - acs_common.load_hmac_secret() + with self.assertRaises(client.SecretFilePermissionsError): + client.load_hmac_secret() def test_symlink_rejected(self) -> None: real = Path(self.tmpdir) / "real.key" @@ -114,8 +118,8 @@ def test_symlink_rejected(self) -> None: os.chmod(real, 0o600) self.secret_path.unlink() os.symlink(real, self.secret_path) - with self.assertRaises(acs_common.SecretFilePermissionsError): - acs_common.load_hmac_secret() + with self.assertRaises(client.SecretFilePermissionsError): + client.load_hmac_secret() # ----- 3. Guardian HTTP DoS via oversized body ----- @@ -182,7 +186,7 @@ def setUp(self) -> None: os.environ["ACS_SESSION_STATE_DIR"] = str(self.tmpdir / "state") # Re-evaluate module-level dir import importlib - importlib.reload(acs_common) + importlib.reload(client_store) def tearDown(self) -> None: import shutil @@ -193,16 +197,16 @@ def tearDown(self) -> None: os.environ["ACS_SESSION_STATE_DIR"] = self._old def test_session_state_file_is_0600(self) -> None: - acs_common.save_session_state("test-session", {"x": 1}) - path = acs_common._session_state_path("test-session") + client.save_session_state("test-session", {"x": 1}) + path = client_store._session_state_path("test-session") self.assertTrue(path.exists()) mode = stat.S_IMODE(os.stat(path).st_mode) self.assertEqual(mode, 0o600, f"session-state file mode is {oct(mode)}, want 0o600") def test_session_state_dir_is_0700(self) -> None: - acs_common.save_session_state("test-session", {"x": 1}) - d = acs_common._SESSION_STATE_DIR + client.save_session_state("test-session", {"x": 1}) + d = client_store._SESSION_STATE_DIR mode = stat.S_IMODE(os.stat(d).st_mode) self.assertEqual(mode, 0o700, f"session-state dir mode is {oct(mode)}, want 0o700") @@ -221,7 +225,7 @@ def test_oversized_command_is_short_circuited(self) -> None: # The destructive matcher should refuse to scan and return a # sentinel that the caller treats as "do not allow without # external review", not a False that allows the command. - result = acs_common.scan_destructive_bash_safely(large_cmd, max_len=8192) + result = example_guardian.scan_destructive_bash_safely(large_cmd, max_len=8192) self.assertEqual(result, "input_too_large", "oversized command MUST be flagged, not scanned, not silently allowed") @@ -235,7 +239,7 @@ def test_small_destructive_command_still_caught(self) -> None: def test_small_safe_command_returns_none(self) -> None: self.assertIsNone( - acs_common.scan_destructive_bash_safely("ls -la", max_len=8192)) + example_guardian.scan_destructive_bash_safely("ls -la", max_len=8192)) class Item13_DestructiveRmFlagVariants(unittest.TestCase): @@ -424,7 +428,7 @@ def test_malformed_signature_returns_false_not_crashes(self) -> None: with self.subTest(signature=bad[:30]): env = self._make_envelope(bad) try: - result = acs_common.verify_signature(env) + result = client.verify_signature(env) except (binascii.Error, ValueError, TypeError) as e: self.fail( f"REGRESSION: malformed signature {bad[:30]!r} " @@ -449,27 +453,27 @@ def test_bound_response_accepted(self) -> None: resp = {"jsonrpc": "2.0", "id": "rpc-1", "result": {"request_id": "11111111-1111-4111-8111-111111111111", "decision": "allow"}} - self.assertTrue(acs_common.response_matches_request(self.REQ, resp)) + self.assertTrue(core.response_matches_request(self.REQ, resp)) def test_wrong_jsonrpc_id_rejected(self) -> None: """A replayed response carries the OLD exchange's id.""" resp = {"jsonrpc": "2.0", "id": "rpc-OTHER", "result": {"request_id": "11111111-1111-4111-8111-111111111111", "decision": "allow"}} - self.assertFalse(acs_common.response_matches_request(self.REQ, resp)) + self.assertFalse(core.response_matches_request(self.REQ, resp)) def test_wrong_request_id_rejected(self) -> None: resp = {"jsonrpc": "2.0", "id": "rpc-1", "result": {"request_id": "99999999-9999-4999-8999-999999999999", "decision": "allow"}} - self.assertFalse(acs_common.response_matches_request(self.REQ, resp)) + self.assertFalse(core.response_matches_request(self.REQ, resp)) def test_result_missing_request_id_rejected(self) -> None: """response-envelope.json requires request_id on every result; a result without one cannot be bound and must not be trusted.""" resp = {"jsonrpc": "2.0", "id": "rpc-1", "result": {"decision": "allow"}} - self.assertFalse(acs_common.response_matches_request(self.REQ, resp)) + self.assertFalse(core.response_matches_request(self.REQ, resp)) def test_error_response_binds_by_id_alone(self) -> None: """Error responses carry no result — the JSON-RPC id is the @@ -478,8 +482,8 @@ def test_error_response_binds_by_id_alone(self) -> None: "error": {"code": -32005, "message": "replay"}} bad = {"jsonrpc": "2.0", "id": "rpc-2", "error": {"code": -32005, "message": "replay"}} - self.assertTrue(acs_common.response_matches_request(self.REQ, ok)) - self.assertFalse(acs_common.response_matches_request(self.REQ, bad)) + self.assertTrue(core.response_matches_request(self.REQ, ok)) + self.assertFalse(core.response_matches_request(self.REQ, bad)) class Item17_GuardianRefusalClassification(unittest.TestCase): @@ -491,14 +495,14 @@ class Item17_GuardianRefusalClassification(unittest.TestCase): def test_refusal_codes(self) -> None: for code in (-32000, -32004, -32005, -32006, -32600, -32700): with self.subTest(code=code): - self.assertTrue(acs_common.is_guardian_refusal(code)) + self.assertTrue(core.is_guardian_refusal(code)) def test_non_refusal_codes_follow_posture(self) -> None: """Version/negotiation errors and unknowns are decision failures, not refusals — they follow the §6.4 posture.""" for code in (None, -32001, -32002, -32003, -32099, 500): with self.subTest(code=code): - self.assertFalse(acs_common.is_guardian_refusal(code)) + self.assertFalse(core.is_guardian_refusal(code)) class Item18_AuditFileSink(unittest.TestCase): @@ -510,8 +514,8 @@ def test_audit_event_appends_to_file_mode_0600(self) -> None: audit_path = os.path.join(d, "audit.log") os.environ["ACS_AUDIT_FILE"] = audit_path try: - acs_common.audit_event("test_sink", foo="bar") - acs_common.audit_event("test_sink_2", baz=1) + client.audit_event("test_sink", foo="bar") + client.audit_event("test_sink_2", baz=1) finally: os.environ.pop("ACS_AUDIT_FILE", None) content = open(audit_path).read() @@ -526,7 +530,7 @@ def test_audit_event_appends_to_file_mode_0600(self) -> None: def test_unwritable_audit_file_does_not_crash_hook_path(self) -> None: os.environ["ACS_AUDIT_FILE"] = "/nonexistent-dir/audit.log" try: - acs_common.audit_event("still_works") # must not raise + client.audit_event("still_works") # must not raise finally: os.environ.pop("ACS_AUDIT_FILE", None) @@ -538,8 +542,8 @@ class Item19_SecretFileUnreadableRaises(unittest.TestCase): def test_missing_secret_file_raises(self) -> None: os.environ["ACS_HMAC_SECRET_FILE"] = "/nonexistent-secret-file" try: - with self.assertRaises(acs_common.SecretFileUnreadableError): - acs_common.load_hmac_secret() + with self.assertRaises(client.SecretFileUnreadableError): + client.load_hmac_secret() finally: os.environ.pop("ACS_HMAC_SECRET_FILE", None) @@ -549,7 +553,7 @@ def test_no_secret_at_all_is_still_devmode_empty(self) -> None: os.environ.pop("ACS_HMAC_SECRET_FILE", None) old = os.environ.pop("ACS_HMAC_SECRET", None) try: - self.assertEqual(acs_common.load_hmac_secret(), b"") + self.assertEqual(client.load_hmac_secret(), b"") finally: if old is not None: os.environ["ACS_HMAC_SECRET"] = old @@ -565,7 +569,7 @@ def test_second_failure_is_fast_and_cached(self) -> None: os.environ["ACS_HANDSHAKE_CACHE"] = d # Force module to re-read the cache dir env import importlib - importlib.reload(acs_common) + importlib.reload(client_handshake) try: kw = dict(guardian_url="http://127.0.0.1:1/acs", session_id="cache-test", agent_id="a", @@ -573,17 +577,17 @@ def test_second_failure_is_fast_and_cached(self) -> None: methods_implemented=["steps/toolCallRequest"], timeout=2.0) t0 = _time.time() - self.assertIsNone(acs_common.ensure_session_handshake(**kw)) + self.assertIsNone(client.ensure_session_handshake(**kw)) first = _time.time() - t0 t1 = _time.time() - self.assertIsNone(acs_common.ensure_session_handshake(**kw)) + self.assertIsNone(client.ensure_session_handshake(**kw)) second = _time.time() - t1 self.assertLess(second, 0.1, f"second handshake attempt took {second:.3f}s — the " f"failure was not negative-cached (first: {first:.3f}s)") finally: os.environ.pop("ACS_HANDSHAKE_CACHE", None) - importlib.reload(acs_common) + importlib.reload(client_handshake) # ----- 21. Degraded bootstrap: missing rfc8785 ----- @@ -667,7 +671,7 @@ def test_cursor_adapter_that_cannot_load_audits_and_honors_the_posture(self) -> "lets the prompt through on adapter load failure") def test_nat_middleware_import_failure_is_loud_not_silent(self) -> None: - """If acs_common (hence rfc8785) can't import, the NAT module + """If acs.core (hence rfc8785) can't import, the NAT module must emit an 'UNGOVERNED' banner AND re-raise — never load cleanly while ungoverned.""" with tempfile.TemporaryDirectory() as tmpdir: @@ -828,11 +832,11 @@ def refuse_after_capture(req, timeout): raise urllib.error.URLError("capture complete") with tempfile.TemporaryDirectory() as d, \ - mock.patch.object(acs_common, "_HANDSHAKE_CACHE_DIR", + mock.patch.object(client_handshake, "_HANDSHAKE_CACHE_DIR", Path(d)), \ - mock.patch.object(acs_common.urllib.request, "urlopen", + mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=refuse_after_capture): - result = acs_common.ensure_session_handshake( + result = client.ensure_session_handshake( guardian_url="http://127.0.0.1:8787/acs", session_id=str(uuid.uuid4()), agent_id="capability-test", diff --git a/adapters/tests/common/test_serverhello_contract.py b/adapters/tests/common/test_serverhello_contract.py index c36f7f11..1d1739f0 100644 --- a/adapters/tests/common/test_serverhello_contract.py +++ b/adapters/tests/common/test_serverhello_contract.py @@ -20,7 +20,9 @@ from pathlib import Path from unittest import mock -from acs import _common as acs_common +from acs import client +from acs import core +from acs.client import _handshake as client_handshake from acs.testing.harness import ( ProgrammableGuardian, build_local_resolver, @@ -50,11 +52,10 @@ def setUp(self) -> None: } clean_environment["ACS_HMAC_SECRET"] = self.SECRET self._stack.enter_context(mock.patch.dict(os.environ, clean_environment, clear=True)) - self._stack.enter_context(mock.patch.object( - acs_common, "_HANDSHAKE_CACHE_DIR", self.cache_directory)) - self._stack.enter_context(mock.patch.object(acs_common, "_HANDSHAKE_CACHE_TTL_S", 3600)) - self._stack.enter_context(mock.patch.object(acs_common, "_HANDSHAKE_FAILURE_TTL_S", 30)) - self.audit = self._stack.enter_context(mock.patch.object(acs_common, "audit_event")) + self._stack.enter_context(mock.patch.object(client_handshake, "_HANDSHAKE_CACHE_DIR", self.cache_directory)) + self._stack.enter_context(mock.patch.object(client_handshake, "_HANDSHAKE_CACHE_TTL_S", 3600)) + self._stack.enter_context(mock.patch.object(client_handshake, "_HANDSHAKE_FAILURE_TTL_S", 30)) + self.audit = self._stack.enter_context(mock.patch.object(client_handshake, "audit_event")) self.requests: list[dict] = [] self.responses: list[dict] = [] self.hello = { @@ -66,7 +67,7 @@ def setUp(self) -> None: } def _run(self, url: str | None = None) -> dict | None: - return acs_common.ensure_session_handshake( + return client.ensure_session_handshake( guardian_url=url or self.URL, session_id=self.SESSION, agent_id="serverhello-regression", platform="test", methods_implemented=["steps/toolCallRequest"], @@ -74,9 +75,9 @@ def _run(self, url: str | None = None) -> dict | None: ) def _sign(self, response: dict) -> dict: - return acs_common.sign_envelope( + return client.sign_envelope( response, session_id=self.SESSION, - key=acs_common.derive_session_key(self.SECRET.encode(), self.SESSION), + key=core.derive_session_key(self.SECRET.encode(), self.SESSION), ) def _respond(self, request, timeout=None): @@ -90,7 +91,7 @@ def _respond(self, request, timeout=None): return io.BytesIO(json.dumps(response).encode()) def _cache(self, url: str | None = None) -> Path: - return acs_common._handshake_cache_path(self.SESSION, url or self.URL) + return client_handshake._handshake_cache_path(self.SESSION, url or self.URL) def _seed(self, result: dict) -> dict: response = self._sign({ @@ -115,7 +116,7 @@ def _assert_audit(self, event: str, cause: str | None = None) -> None: self.assertTrue(matches, self.audit.call_args_list) def test_fresh_direct_result_is_cached_as_the_whole_signed_envelope(self): - with mock.patch.object(acs_common.urllib.request, "urlopen", side_effect=self._respond): + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=self._respond): result = self._run() self.assertIsInstance(result, dict) self._assert_hello_schema(result) @@ -125,26 +126,26 @@ def test_fresh_direct_result_is_cached_as_the_whole_signed_envelope(self): self.assertEqual(cached, self.responses[0]) self.assertEqual(cached["result"], result) self.assertIn("signature", cached["result"]) - self.assertTrue(acs_common.verify_signature(cached, session_id=self.SESSION)) + self.assertTrue(client.verify_signature(cached, session_id=self.SESSION)) self.assertEqual(stat.S_IMODE(self._cache().stat().st_mode), 0o600) self.assertEqual(stat.S_IMODE(self.cache_directory.stat().st_mode), 0o700) self.assertEqual(validate_response_envelope(cached), []) def test_preseeded_direct_cache_is_reverified_without_network(self): response = self._seed(self.hello) - with mock.patch.object(acs_common.urllib.request, "urlopen") as network, \ - mock.patch.object(acs_common, "verify_signature", - wraps=acs_common.verify_signature) as verify: + with mock.patch.object(client_handshake.urllib.request, "urlopen") as network, \ + mock.patch.object(client_handshake, "verify_signature", + wraps=client.verify_signature) as verify: result = self._run() self.assertEqual(result, response["result"]) network.assert_not_called() verify.assert_called_once_with(response, session_id=self.SESSION) def test_two_calls_perform_one_handshake_and_verify_both_reads(self): - with mock.patch.object(acs_common.urllib.request, "urlopen", + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=self._respond) as network, \ - mock.patch.object(acs_common, "verify_signature", - wraps=acs_common.verify_signature) as verify: + mock.patch.object(client_handshake, "verify_signature", + wraps=client.verify_signature) as verify: first = self._run() second = self._run() self.assertIsNotNone(first) @@ -156,7 +157,7 @@ def test_tampered_cached_posture_is_detected_and_not_trusted(self): response = self._seed(self.hello) response["result"]["on_decision_failure"] = "proceed" self._cache().write_text(json.dumps(response)) - with mock.patch.object(acs_common.urllib.request, "urlopen", + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=self._respond) as network: result = self._run() self.assertEqual(network.call_count, 1) @@ -168,7 +169,7 @@ def respond(request, timeout=None): response = json.loads(self._respond(request).read()) response["result"]["on_decision_failure"] = "proceed" return io.BytesIO(json.dumps(response).encode()) - with mock.patch.object(acs_common.urllib.request, "urlopen", side_effect=respond): + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=respond): self.assertIsNone(self._run()) self.assertFalse(self._cache().exists()) self._assert_audit("handshake_failed", "server_hello_signature_invalid") @@ -178,7 +179,7 @@ def respond(request, timeout=None): response = json.loads(self._respond(request).read()) response["result"].pop("signature") return io.BytesIO(json.dumps(response).encode()) - with mock.patch.object(acs_common.urllib.request, "urlopen", side_effect=respond): + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=respond): self.assertIsNone(self._run()) self.assertFalse(self._cache().exists()) self._assert_audit("handshake_failed", "server_hello_signature_invalid") @@ -189,7 +190,7 @@ def respond(request, timeout=None): response["id"] = "a-different-request" self._sign(response) return io.BytesIO(json.dumps(response).encode()) - with mock.patch.object(acs_common.urllib.request, "urlopen", side_effect=respond): + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=respond): self.assertIsNone(self._run()) self.assertFalse(self._cache().exists()) self._assert_audit("handshake_failed", "response_id_mismatch") @@ -200,7 +201,7 @@ def test_signed_legacy_result_wrapper_is_not_a_serverhello(self): "request_id": str(uuid.uuid4()), "decision": "allow", "payload": copy.deepcopy(self.hello), } - with mock.patch.object(acs_common.urllib.request, "urlopen", side_effect=self._respond): + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=self._respond): self.assertIsNone(self._run()) self.assertFalse(self._cache().exists()) self._assert_audit("handshake_failed", "invalid_server_hello") @@ -211,7 +212,7 @@ def test_signed_legacy_wrapped_cache_triggers_a_new_handshake(self): "request_id": str(uuid.uuid4()), "decision": "allow", "payload": copy.deepcopy(self.hello), }) - with mock.patch.object(acs_common.urllib.request, "urlopen", + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=self._respond) as network: result = self._run() self.assertEqual(network.call_count, 1) @@ -223,7 +224,7 @@ def test_unknown_fields_do_not_change_result_extraction(self): # is not a reason to unwrap a valid direct ServerHello. self.hello["payload"] = {"future_extension": True} self.hello["another_extension"] = "preserve" - with mock.patch.object(acs_common.urllib.request, "urlopen", side_effect=self._respond): + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=self._respond): result = self._run() self._assert_hello_schema(result) self.assertEqual(result["payload"], {"future_extension": True}) @@ -231,7 +232,7 @@ def test_unknown_fields_do_not_change_result_extraction(self): def test_empty_result_is_not_cached(self): self.hello = {} - with mock.patch.object(acs_common.urllib.request, "urlopen", side_effect=self._respond): + with mock.patch.object(client_handshake.urllib.request, "urlopen", side_effect=self._respond): self.assertIsNone(self._run()) self.assertFalse(self._cache().exists()) @@ -258,7 +259,7 @@ def test_real_python_guardian_returns_a_signed_direct_result(self): cached = json.loads(self._cache(url).read_text()) self.assertEqual(result, cached["result"]) self.assertEqual(validate_response_envelope(cached), []) - self.assertTrue(acs_common.verify_signature(cached, session_id=self.SESSION)) + self.assertTrue(client.verify_signature(cached, session_id=self.SESSION)) finally: proc.terminate() try: diff --git a/adapters/tests/conformance/test_acs_core_conformance.py b/adapters/tests/conformance/test_acs_core_conformance.py index 00c53529..406df412 100644 --- a/adapters/tests/conformance/test_acs_core_conformance.py +++ b/adapters/tests/conformance/test_acs_core_conformance.py @@ -67,7 +67,8 @@ GUARDIAN_SCRIPT = ADAPTERS_DIR / "src" / "acs" / "guardian" / "example_guardian.py" INTEGRATIONS_DIR = ADAPTERS_DIR / "src" / "acs" / "integrations" -from acs import _common as acs_common # noqa: E402 +from acs import client # noqa: E402 +from acs import core # noqa: E402 from acs.core import schemas_dir # noqa: E402 # Required: the suite fails without the canonical schemas instead of skipping. @@ -80,7 +81,7 @@ # ============================================================================= # Test harness — spawns the Guardian, exchanges signed envelopes. -# Helpers come from adapters/_common/test_harness.py — see that file for +# Helpers come from src/acs/testing/harness.py — see that file for # the canonical implementations of launch_guardian, schema validators, # and ProgrammableGuardian. # ============================================================================= @@ -163,7 +164,7 @@ def _make_envelope(self, method: str, payload: dict | None = None, *, "params": { "acs_version": "0.1.0", "request_id": request_id or str(uuid.uuid4()), - "timestamp": timestamp or acs_common.iso8601_now(), + "timestamp": timestamp or core.iso8601_now(), "metadata": { "agent_id": "conformance-test", "session_id": sid, @@ -173,8 +174,8 @@ def _make_envelope(self, method: str, payload: dict | None = None, *, }, } if sign and self.HMAC_SECRET: - key = acs_common.derive_session_key(self.HMAC_SECRET.encode(), sid) - acs_common.sign_envelope(env, key=key, session_id=sid) + key = core.derive_session_key(self.HMAC_SECRET.encode(), sid) + client.sign_envelope(env, key=key, session_id=sid) return env def _post(self, envelope: dict) -> dict: @@ -1096,7 +1097,7 @@ def test_disjoint_structured_targets_are_not_reported_as_overlap(self) -> None: from jsonschema import Draft202012Validator schema = json.loads((SPEC_DIR / "modifications.json").read_text()) Draft202012Validator(schema).validate(mods) - self.assertIsNone(acs_common.modify_composition_violation(mods), + self.assertIsNone(core.modify_composition_violation(mods), "disjoint structured edits are valid; the overlap guard must not " "turn every combined MODIFY into DENY") @@ -1213,10 +1214,10 @@ def expected(req: dict, prev_hash: str | None) -> str: "step_id": params["request_id"], "step_type": req["method"], "request_hash": hashlib.sha256( - acs_common.jcs_canonicalize(params)).hexdigest(), + core.jcs_canonicalize(params)).hexdigest(), "timestamp": params["timestamp"], } - content_bytes = acs_common.jcs_canonicalize(entry) + content_bytes = core.jcs_canonicalize(entry) prev_bytes = bytes.fromhex(prev_hash) if prev_hash else b"" return hashlib.sha256(content_bytes + prev_bytes).hexdigest() @@ -1385,8 +1386,8 @@ def test_response_is_signed_and_verifies(self) -> None: sig = resp["result"].get("signature") self.assertIsNotNone(sig, "Guardian response missing `signature` field per §10") - key = acs_common.derive_session_key(self.HMAC_SECRET.encode(), sid) - self.assertTrue(acs_common.verify_signature(resp, key=key), + key = core.derive_session_key(self.HMAC_SECRET.encode(), sid) + self.assertTrue(client.verify_signature(resp, key=key), "Guardian's response signature must verify with the " "HKDF-derived per-session key") @@ -1412,13 +1413,13 @@ def test_error_response_is_signed_and_verifies(self) -> None: self.assertIsNotNone(sig, "error response missing `signature` — errors are responses " "too; conformance.md, ACS-Core / Baseline integrity exempts only system/ping") - key = acs_common.derive_session_key(self.HMAC_SECRET.encode(), sid) - self.assertTrue(acs_common.verify_signature(resp, key=key), + key = core.derive_session_key(self.HMAC_SECRET.encode(), sid) + self.assertTrue(client.verify_signature(resp, key=key), "error-response signature must verify with the per-session key") # And a tampered error must NOT verify (the check is real). tampered = json.loads(json.dumps(resp)) tampered["error"]["message"] = "REPLAY_DETECTED but for a different action" - self.assertFalse(acs_common.verify_signature(tampered, key=key), + self.assertFalse(client.verify_signature(tampered, key=key), "tampered error envelope must fail verification") def test_per_session_key_derivation(self) -> None: @@ -1426,12 +1427,12 @@ def test_per_session_key_derivation(self) -> None: key material'. The derived key MUST differ between sessions with the same secret.""" secret = self.HMAC_SECRET.encode() - k1 = acs_common.derive_session_key(secret, "session-A") - k2 = acs_common.derive_session_key(secret, "session-B") + k1 = core.derive_session_key(secret, "session-A") + k2 = core.derive_session_key(secret, "session-B") self.assertNotEqual(k1, k2, "per-session HKDF MUST produce distinct keys for distinct sessions") # Same session_id → same key - k1b = acs_common.derive_session_key(secret, "session-A") + k1b = core.derive_session_key(secret, "session-A") self.assertEqual(k1, k1b, "HKDF must be deterministic for the same (secret, session_id)") diff --git a/adapters/tests/integrations/claude_code/e2e_check.py b/adapters/tests/integrations/claude_code/e2e_check.py index 3ce41b02..70223e18 100644 --- a/adapters/tests/integrations/claude_code/e2e_check.py +++ b/adapters/tests/integrations/claude_code/e2e_check.py @@ -51,7 +51,7 @@ SPEC_DIR_DEFAULT = schemas_dir(os.environ.get("ACS_SPEC_DIR")) -from acs import _common as acs_common # noqa: E402 + from acs.testing.harness import ( # noqa: E402 ProgrammableGuardian, validate_request_envelope, @@ -64,7 +64,7 @@ PER_CALL_TIMEOUT_S = 120.0 -# Shared pretty-printer + helpers; see adapters/_common/e2e_report.py. +# Shared pretty-printer + helpers; see src/acs/testing/e2e_report.py. from acs.testing.e2e_report import ( # noqa: E402 Report, assert_envelopes_signed_and_valid as _assert_envelopes_signed_and_valid, diff --git a/adapters/tests/integrations/cursor/e2e_check.py b/adapters/tests/integrations/cursor/e2e_check.py index 47ba888e..fa11a381 100644 --- a/adapters/tests/integrations/cursor/e2e_check.py +++ b/adapters/tests/integrations/cursor/e2e_check.py @@ -65,7 +65,7 @@ SPEC_DIR_DEFAULT = schemas_dir(os.environ.get("ACS_SPEC_DIR")) -from acs import _common as acs_common # noqa: E402 + from acs.testing.harness import ( # noqa: E402 ProgrammableGuardian, validate_request_envelope, @@ -78,7 +78,7 @@ HMAC_SECRET = "e2e-test-shared-secret-not-for-production" -# Shared pretty-printer + helpers; see adapters/_common/e2e_report.py. +# Shared pretty-printer + helpers; see src/acs/testing/e2e_report.py. from acs.testing.e2e_report import ( # noqa: E402 Report, real_policy_handler as _shared_real_policy_handler, assert_envelopes_signed_and_valid as _assert_envelopes_signed_and_valid, diff --git a/adapters/tests/integrations/nat/e2e_check.py b/adapters/tests/integrations/nat/e2e_check.py index cba9cf5f..96d6f430 100644 --- a/adapters/tests/integrations/nat/e2e_check.py +++ b/adapters/tests/integrations/nat/e2e_check.py @@ -39,7 +39,6 @@ SPEC_DIR_DEFAULT = schemas_dir(os.environ.get("ACS_SPEC_DIR")) -from acs import _common as acs_common # noqa: E402 from acs.testing.harness import ( # noqa: E402 ProgrammableGuardian, validate_request_envelope, diff --git a/adapters/tests/integrations/nat/test_adapter.py b/adapters/tests/integrations/nat/test_adapter.py index 07685480..086bb240 100644 --- a/adapters/tests/integrations/nat/test_adapter.py +++ b/adapters/tests/integrations/nat/test_adapter.py @@ -34,6 +34,7 @@ _NAT_OK = False # Import adapter (it tolerates NAT missing) +from acs.client import _handshake as client_handshake # noqa: E402 from acs.integrations.nat.adapter import ACSMiddleware, ACSGuardianDenied # noqa: E402 if _NAT_OK: @@ -43,7 +44,7 @@ HERE = Path(__file__).resolve().parent -from acs import _common as acs_common # noqa: E402 + from acs.testing.harness import ( # noqa: E402 ProgrammableGuardian, launch_guardian, @@ -133,7 +134,7 @@ def delayed_deny(req: dict) -> dict: os.environ["ACS_HMAC_SECRET"] = g.hmac_secret os.environ.pop("ACS_HMAC_SECRET_FILE", None) with tempfile.TemporaryDirectory() as d, \ - mock.patch.object(acs_common, "_HANDSHAKE_CACHE_DIR", + mock.patch.object(client_handshake, "_HANDSHAKE_CACHE_DIR", Path(d)), g: mw = ACSMiddleware(ACSMiddlewareConfig( guardian_url=g.url(), default_deny=False, @@ -174,7 +175,7 @@ def hello(req: dict) -> dict: os.environ["ACS_HMAC_SECRET"] = g.hmac_secret os.environ.pop("ACS_HMAC_SECRET_FILE", None) with tempfile.TemporaryDirectory() as d, \ - mock.patch.object(acs_common, "_HANDSHAKE_CACHE_DIR", + mock.patch.object(client_handshake, "_HANDSHAKE_CACHE_DIR", Path(d)), g: mw = ACSMiddleware(ACSMiddlewareConfig( guardian_url=g.url(), default_deny=True,