From 5cb9634c517ab1ed71289ae0d0e56ddda2c09be7 Mon Sep 17 00:00:00 2001 From: CC#3 Kora Runtime Date: Sat, 23 May 2026 13:55:31 -0700 Subject: [PATCH] =?UTF-8?q?feat(kora):=20KR-MCP-AUDIT-ON-DENIAL=20?= =?UTF-8?q?=E2=80=94=20emit=20audit=20at=20cap-gate=20denial?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Activates CC#1's pinned-no-alert rule from KR-ALERTS-PANEL-FLIP (#145). Two denial paths now write JSONL audit rows BEFORE returning the -32001 envelope, so the capability_denied_24h alert rule surfaces real warnings when callers are misconfigured. Two helpers in kora_cli/listeners/mcp.py: - _emit_capability_denied_audit — fires at the cap-gate (when _check_cap_gate returns a -32001 envelope). Writes seam=mcp.tool_called with details.result="capability_denied" plus tool_name, tool_kind="mutating", caller_actor_kind, caller_actor_id, required_capability, duration_ms=0, tool_status="not_allowed". - _emit_actor_id_required_audit — fires at the actor_id-required gate inside kora__request_stop (the extra gate from KR-MCP-STOP-CONTROL ST2). Same shape but details.result="actor_id_required" — distinct discriminator so the alert rule's detail_match doesn't conflate two different operator-fix paths (cap-not-granted vs actor_id-field-absent in mcp_callers.yaml). Both helpers are best-effort emit: any exception from the audit sink is caught + logged but does NOT mask the denial response to the caller (verified by test_cap_gate_denial_audit_emitted_ before_envelope). Reason text NEVER in the denial audit (caller-supplied string might leak context) — same posture as the stop-tool's success-path audit from #147. Documentation: - kora_cli/alerts/aggregator.py docstring: removed the forward-compat "rule emits zero today" block; replaced with a description of the now-live wiring + the two result discriminators. Tests: - tests/kora_cli/test_listeners/test_mcp_audit_on_denial.py: 8 new tests covering cap-gate denial audit shape, missing actor_id audit shape, audit-emitted-before-envelope (sink failure doesn't mask denial), executor-not-invoked-on- denial, no-denial-audit-on-success, and the SECURITY sweep (bearer token never in any audit row). - tests/kora_cli/alerts/test_aggregator.py: renamed test_capability_denied_today_no_alert_since_audit_doesnt_emit_denials → test_capability_denied_today_emits_alert_when_threshold_exceeded. Flipped behavior to assert the alert fires when 11+ capability_denied audit rows accumulate in 24h. The new test also seeds 7 "ok" + 5 "actor_id_required" rows to verify the rule's detail_match correctly excludes them (alert count is 11, not 23). Full regression: 9097 passed; 47 failed identical to ST2 baseline = zero new regressions. All failures are pre-existing in tests/agent/test_anthropic_adapter.py + tests/kora_cli/test_web_server*.py — unrelated to this PR. After this lands, the alerts panel surfaces real "N capability_denied responses in 24h" warnings when callers are misconfigured. Co-Authored-By: Claude Opus 4.7 (1M context) --- kora_cli/alerts/aggregator.py | 40 +- kora_cli/listeners/mcp.py | 134 +++++ tests/kora_cli/alerts/test_aggregator.py | 58 +- .../test_mcp_audit_on_denial.py | 524 ++++++++++++++++++ 4 files changed, 725 insertions(+), 31 deletions(-) create mode 100644 tests/kora_cli/test_listeners/test_mcp_audit_on_denial.py diff --git a/kora_cli/alerts/aggregator.py b/kora_cli/alerts/aggregator.py index af412cb4a46f..2834567f26c9 100644 --- a/kora_cli/alerts/aggregator.py +++ b/kora_cli/alerts/aggregator.py @@ -50,17 +50,17 @@ ``heartbeat_probes/types.py:28``: {healthy, degraded, unhealthy, unknown}. -# Forward-compat note: capability_denied - -The ``capability_denied_24h`` rule is forward-looking: today the -``mcp.tool_called`` audit emit at -``kora_cli/listeners/mcp_tools.py:714`` is reached AFTER the -capability gate (``listeners/mcp.py:181``), so denial responses -are NOT currently audit-logged. This rule's matcher uses -``details.result == "capability_denied"`` so when a follow-on -bucket adds audit-on-denial the rule activates automatically; in -the meantime it emits zero alerts (no false negatives — the data -genuinely isn't there). +# capability_denied rule — live as of KR-MCP-AUDIT-ON-DENIAL + +The ``capability_denied_24h`` rule consumes JSONL audit rows that +``listeners/mcp.py:_emit_capability_denied_audit`` writes BEFORE +returning the -32001 envelope at the cap gate. Discriminator is +``details.result == "capability_denied"``. The companion path — +``actor_id_required_for_stop`` from the ``kora__request_stop`` +tool's extra gate — writes +``details.result == "actor_id_required"`` so it's queryable +separately (future ``actor_id_required_24h`` rule can grep it +without conflating two distinct operator-fix surfaces). """ from __future__ import annotations @@ -338,14 +338,16 @@ def _rules_from_webhook_audit() -> List[Alert]: def _rules_from_capability_denied_audit() -> List[Alert]: - """Forward-looking — see module docstring's forward-compat note. - - Today the audit log at ``mcp.tool_called`` only records successful - invocations (capability gate at ``listeners/mcp.py:181`` returns - BEFORE the audit emit at ``listeners/mcp_tools.py:714``). This - rule's matcher is forward-compatible: when a follow-on bucket - adds audit-on-denial it'll start firing without an aggregator - edit. + """Surface a warning when too many cap-gate denials in 24h. + + KR-MCP-AUDIT-ON-DENIAL wired the cap-gate at + ``listeners/mcp.py`` to call + ``_emit_capability_denied_audit`` BEFORE returning the + -32001 envelope, so denial responses now land in the audit + JSONL with ``details.result == "capability_denied"``. This + rule fires when more than + :data:`CAPABILITY_DENIED_24H_THRESHOLD` denials accumulate + in the trailing 24h. """ since = datetime.now(timezone.utc) - timedelta(hours=24) count = _count_audit_entries_in_window( diff --git a/kora_cli/listeners/mcp.py b/kora_cli/listeners/mcp.py index 57e4ef3010fa..a742bb29e483 100644 --- a/kora_cli/listeners/mcp.py +++ b/kora_cli/listeners/mcp.py @@ -203,6 +203,119 @@ def _check_cap_gate( } +# --------------------------------------------------------------------------- +# KR-MCP-AUDIT-ON-DENIAL — JSONL audit emit for the two denial paths +# --------------------------------------------------------------------------- +# +# Both helpers write to the same ``mcp.tool_called`` seam used by the +# success-path audit in ``mcp_tools._emit_audit``. They run BEFORE the +# JSON-RPC error envelope is returned so a denied call leaves a JSONL +# row even though no executor ran. The ``details.result`` field is the +# alert-rule discriminator — values match the patterns CC#1's +# ``capability_denied_24h`` rule (+ future actor_id_required rules) +# already grep for. +# +# Why not call into ``mcp_tools._emit_audit``: that helper takes a +# ``result`` STRING the executor builds from its own state-change +# vocabulary (e.g. ``"active->paused"``) and embeds ``args_keys`` from +# the executor's input. Denial paths don't have either — the request +# never reached the executor. Keeping a separate helper here avoids +# pretzeling _emit_audit's contract for two callers with different +# inputs. + + +def _emit_capability_denied_audit( + *, tool_name: Optional[str], caller: Caller +) -> None: + """Write a JSONL audit row for a cap-gate denial. + + Reason text NEVER in the audit (caller-supplied; might leak + sensitive context — same posture as the stop-tool's audit + omission). Captures only operator-actionable fields: which tool, + which caller actor_kind, which capability was required, and the + discriminator literal CC#1's alert rule matches on. + """ + # Best-effort emit — never raise from this path. An audit-sink + # failure must NOT mask the denial response to the caller. + try: + from kora_cli.audit import emit_audit + + emit_audit( + seam="mcp.tool_called", + details={ + "tool_name": tool_name or "", + "tool_kind": "mutating", + "caller_actor_kind": caller.actor_kind, + "caller_actor_id": caller.actor_id, + "required_capability": tool_name or "", + "duration_ms": 0, + "tool_status": "not_allowed", + "result": "capability_denied", + }, + source="mcp_http", + ) + # Also emit the structured-log line for operator grep — mirrors + # the dual-write pattern in mcp_tools._emit_audit. + logger.info( + "[kora.mcp.tool_denied] tool=%s caller_actor_kind=%s " + "result=capability_denied", + tool_name, + caller.actor_kind, + ) + except Exception: # pragma: no cover — sink failure must not mask denial + logger.exception( + "[kora.mcp.tool_denied] emit_audit failed for " + "tool=%s caller_actor_kind=%s — denial response still " + "returned, but JSONL row missing", + tool_name, + caller.actor_kind, + ) + + +def _emit_actor_id_required_audit( + *, tool_name: Optional[str], caller: Caller +) -> None: + """Write a JSONL audit row for an actor_id-required denial. + + Distinct ``result`` literal so CC#1's ``capability_denied_24h`` + alert rule's ``detail_match`` doesn't conflate this with a cap + denial — different operator-fix path (the cap IS granted; the + actor_id field is missing in mcp_callers.yaml). A future + ``actor_id_required_24h`` alert rule can grep this separately. + """ + try: + from kora_cli.audit import emit_audit + + emit_audit( + seam="mcp.tool_called", + details={ + "tool_name": tool_name or "", + "tool_kind": "mutating", + "caller_actor_kind": caller.actor_kind, + "caller_actor_id": caller.actor_id, + "required_capability": tool_name or "", + "duration_ms": 0, + "tool_status": "not_allowed", + "result": "actor_id_required", + }, + source="mcp_http", + ) + logger.info( + "[kora.mcp.tool_denied] tool=%s caller_actor_kind=%s " + "result=actor_id_required", + tool_name, + caller.actor_kind, + ) + except Exception: # pragma: no cover + logger.exception( + "[kora.mcp.tool_denied] emit_audit failed for " + "tool=%s caller_actor_kind=%s — denial response still " + "returned, but JSONL row missing", + tool_name, + caller.actor_kind, + ) + + def _execute_daemon_status() -> Dict[str, Any]: """Body for ``kora__daemon_status``. Returns the JSON dict.""" coord = current_coordinator() @@ -311,6 +424,15 @@ async def post_jsonrpc( # caller's allowed_caps must include the tool name. gate_err = _check_cap_gate(req_id, tool_name, caller) if gate_err is not None: + # KR-MCP-AUDIT-ON-DENIAL — emit a JSONL audit row for the + # denial BEFORE returning. The cap-gate is the only gate + # that needs this explicit emit; the success-path audit + # at mcp_tools._emit_audit runs from inside each executor + # AFTER dispatch, so it never sees denied calls. CC#1's + # capability_denied_24h alert rule consumes these rows. + _emit_capability_denied_audit( + tool_name=tool_name, caller=caller + ) return gate_err if tool_name == "kora__daemon_status": @@ -376,6 +498,18 @@ async def post_jsonrpc( # caller still lacks the actor_id field needed for # substrate attribution. Operator-fix path is in the # error message. + # + # KR-MCP-AUDIT-ON-DENIAL — emit symmetrically with the + # cap-gate denial so the alerts panel can surface + # actor_id-required misconfigurations the same way. + # Same seam + same tool_kind; result discriminator is + # "actor_id_required" (distinct from + # "capability_denied" so the alert rule's + # detail_match doesn't conflate the two — they're + # different operator-fix paths). + _emit_actor_id_required_audit( + tool_name=tool_name, caller=caller + ) return _jsonrpc_error( req_id, -32001, diff --git a/tests/kora_cli/alerts/test_aggregator.py b/tests/kora_cli/alerts/test_aggregator.py index a6cf181bf7d0..b9c8d9ac87b3 100644 --- a/tests/kora_cli/alerts/test_aggregator.py +++ b/tests/kora_cli/alerts/test_aggregator.py @@ -330,19 +330,46 @@ def fake_read(seam=None, since=None): assert str(CAPABILITY_DENIED_24H_THRESHOLD + 1) in matching[0].title -def test_capability_denied_today_no_alert_since_audit_doesnt_emit_denials( +def test_capability_denied_today_emits_alert_when_threshold_exceeded( patch_sources, ): - """Forward-compat: today the audit emit at mcp_tools.py:714 runs - AFTER the cap-gate, so denial responses aren't logged. Rule - emits zero alerts in the current state — documented in the - aggregator module docstring.""" + """KR-MCP-AUDIT-ON-DENIAL — the cap-gate's denial path now writes + ``mcp.tool_called`` rows with ``details.result == + "capability_denied"``. With more than + :data:`CAPABILITY_DENIED_24H_THRESHOLD` such rows in the + trailing 24h, the alert rule fires exactly one warning so the + panel can prompt the operator to review mcp_callers.yaml. + + Previously this test pinned the OPPOSITE behavior (forward-compat + "no alert because audit doesn't emit denials yet") — that + contract closed when audit-on-denial landed. The test is kept + under a renamed identity to lock in the new behavior in the + same slot so a future regression can't quietly drop the alert.""" sources, _ = patch_sources - # Only "ok" entries — no capability_denied results. - entries = [ - _make_audit_entry("mcp.tool_called", details={"result": "ok"}) - for _ in range(CAPABILITY_DENIED_24H_THRESHOLD + 50) - ] + # 11 denials + a mix of OK + actor_id_required entries that should + # NOT count against the capability_denied threshold. + entries = ( + [ + _make_audit_entry( + "mcp.tool_called", + details={"result": "capability_denied"}, + ) + for _ in range(CAPABILITY_DENIED_24H_THRESHOLD + 1) + ] + + [ + _make_audit_entry( + "mcp.tool_called", details={"result": "ok"} + ) + for _ in range(7) + ] + + [ + _make_audit_entry( + "mcp.tool_called", + details={"result": "actor_id_required"}, + ) + for _ in range(5) + ] + ) def fake_read(seam=None, since=None): return entries if seam == "mcp.tool_called" else [] @@ -352,8 +379,15 @@ def fake_read(seam=None, since=None): side_effect=fake_read, ): alerts = compute_active_alerts() - ids = [a.id for a in alerts] - assert "capability_denied_24h" not in ids + matching = [a for a in alerts if a.id == "capability_denied_24h"] + assert len(matching) == 1, ( + "expected exactly one capability_denied_24h alert when " + "11 denials exceed the threshold of 10" + ) + assert matching[0].severity == "info" + # Title surfaces the matched-only count (11 — not the mixed-bag + # total of 11 + 7 + 5 = 23). + assert str(CAPABILITY_DENIED_24H_THRESHOLD + 1) in matching[0].title def test_reasoning_errors_fires_when_over_threshold(patch_sources): diff --git a/tests/kora_cli/test_listeners/test_mcp_audit_on_denial.py b/tests/kora_cli/test_listeners/test_mcp_audit_on_denial.py new file mode 100644 index 000000000000..a65ff3573262 --- /dev/null +++ b/tests/kora_cli/test_listeners/test_mcp_audit_on_denial.py @@ -0,0 +1,524 @@ +"""Tests for KR-MCP-AUDIT-ON-DENIAL — emit JSONL audit row at cap-gate denials. + +Two denial paths land in the audit stream: + - cap-gate denial → ``details.result == "capability_denied"`` + - actor_id-required denial (kora__request_stop only) → + ``details.result == "actor_id_required"`` + +The two ``result`` literals are CC#1's alert-rule discriminators. + +Covers spec §2(c): + - cap-gate denial: AuditEntry written with result=capability_denied + + required_capability + tool_name + caller_actor_kind + + caller_actor_id + - actor_id-required denial: AuditEntry written with result= + actor_id_required + - Successful tool call: no denial-result audit (only success-path) + - Audit emit happens BEFORE the JSON-RPC envelope is returned + - Audit-sink failure does NOT mask the denial response + - Walk-payload security: bearer tokens never in audit details +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import pytest +import yaml +from fastapi.testclient import TestClient + + +def _sha256(tok: str) -> str: + return "sha256:" + hashlib.sha256(tok.encode("utf-8")).hexdigest() + + +def _write_callers_yaml(path: Path, entries: list) -> None: + path.write_text(yaml.safe_dump({"callers": entries}), encoding="utf-8") + + +_OPERATOR_ACTOR_UUID = "8d50b3aa-1111-4222-9333-cafebabe1234" + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _reset_caller_cache(): + from kora_cli.listeners import mcp_caller_auth + + mcp_caller_auth._reset_cache_for_tests() + yield + mcp_caller_auth._reset_cache_for_tests() + + +@pytest.fixture +def empty_caps_token(monkeypatch, tmp_path): + """Caller authenticated but with NO caps — every mutating tool denies.""" + token = "no-caps-tok" + callers_path = tmp_path / "mcp_callers.yaml" + _write_callers_yaml( + callers_path, + [ + { + "token_hash": _sha256(token), + "actor_kind": "claude_pm_uncfg", + "actor_id": _OPERATOR_ACTOR_UUID, + "allowed_caps": [], + } + ], + ) + from kora_cli.listeners import mcp_caller_auth + + monkeypatch.setattr( + mcp_caller_auth, "DEFAULT_CALLERS_PATH", callers_path + ) + monkeypatch.delenv("KORA_MCP_BEARER_TOKEN", raising=False) + return token + + +@pytest.fixture +def stop_cap_no_actor_id_token(monkeypatch, tmp_path): + """Caller has stop cap but NO actor_id — actor_id_required path.""" + token = "stop-no-aid-tok" + callers_path = tmp_path / "mcp_callers.yaml" + _write_callers_yaml( + callers_path, + [ + { + "token_hash": _sha256(token), + "actor_kind": "claude_pm_no_actor_id", + # actor_id omitted + "allowed_caps": ["kora__request_stop"], + } + ], + ) + from kora_cli.listeners import mcp_caller_auth + + monkeypatch.setattr( + mcp_caller_auth, "DEFAULT_CALLERS_PATH", callers_path + ) + monkeypatch.delenv("KORA_MCP_BEARER_TOKEN", raising=False) + return token + + +@pytest.fixture +def full_caps_token(monkeypatch, tmp_path): + """Caller fully authorized — success path (no denial audit).""" + token = "full-caps-tok" + callers_path = tmp_path / "mcp_callers.yaml" + _write_callers_yaml( + callers_path, + [ + { + "token_hash": _sha256(token), + "actor_kind": "claude_pm_full", + "actor_id": _OPERATOR_ACTOR_UUID, + "allowed_caps": [ + "kora__request_pause", + "kora__request_resume", + ], + } + ], + ) + from kora_cli.listeners import mcp_caller_auth + + monkeypatch.setattr( + mcp_caller_auth, "DEFAULT_CALLERS_PATH", callers_path + ) + monkeypatch.delenv("KORA_MCP_BEARER_TOKEN", raising=False) + return token + + +@pytest.fixture +def client(): + from kora_cli.web_server import app + + return TestClient(app) + + +@pytest.fixture +def captured_audit_calls(monkeypatch): + """Patch the lazy ``from kora_cli.audit import emit_audit`` inside + the denial helpers. The helpers do ``from kora_cli.audit import + emit_audit`` at call time; the import resolves via + kora_cli/audit/__init__.py which re-exports from jsonl_sink. + + We patch the underlying ``emit_audit`` in the jsonl_sink module + + the re-export. Either resolves into the same callable in this + process.""" + calls: list[dict] = [] + + def fake_emit_audit( + seam: str, + details: dict, + *, + caller_session_id: Any = None, + source: Any = None, + log_path: Any = None, + ) -> None: + calls.append( + { + "seam": seam, + "details": dict(details), + "caller_session_id": caller_session_id, + "source": source, + } + ) + + monkeypatch.setattr( + "kora_cli.audit.jsonl_sink.emit_audit", fake_emit_audit + ) + monkeypatch.setattr("kora_cli.audit.emit_audit", fake_emit_audit) + return calls + + +# --------------------------------------------------------------------------- +# Cap-gate denial path +# --------------------------------------------------------------------------- + + +def test_cap_gate_denial_writes_audit_row( + client, empty_caps_token, captured_audit_calls +): + r = client.post( + "/mcp", + headers={"Authorization": f"Bearer {empty_caps_token}"}, + json={ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "kora__request_pause", + "arguments": {"reason": "denied-test"}, + }, + }, + ) + # Denial envelope returned to caller. + err = r.json()["error"] + assert err["code"] == -32001 + assert err["message"] == "capability_denied" + + # Exactly one denial audit row written. + denial_calls = [ + c + for c in captured_audit_calls + if c["details"].get("result") == "capability_denied" + ] + assert len(denial_calls) == 1 + row = denial_calls[0] + assert row["seam"] == "mcp.tool_called" + assert row["source"] == "mcp_http" + d = row["details"] + assert d["tool_name"] == "kora__request_pause" + assert d["tool_kind"] == "mutating" + assert d["caller_actor_kind"] == "claude_pm_uncfg" + assert d["caller_actor_id"] == _OPERATOR_ACTOR_UUID + assert d["required_capability"] == "kora__request_pause" + assert d["duration_ms"] == 0 + assert d["tool_status"] == "not_allowed" + assert d["result"] == "capability_denied" + + +def test_cap_gate_denial_caller_actor_id_none_when_absent( + client, monkeypatch, tmp_path, captured_audit_calls +): + """Caller without actor_id field still gets a clean audit row + (caller_actor_id: None) — actor_id absence is its own state and + shouldn't be masked into 'no audit'.""" + token = "no-aid-tok" + callers_path = tmp_path / "mcp_callers.yaml" + _write_callers_yaml( + callers_path, + [ + { + "token_hash": _sha256(token), + "actor_kind": "claude_pm_legacy", + # actor_id absent + "allowed_caps": [], + } + ], + ) + from kora_cli.listeners import mcp_caller_auth + + monkeypatch.setattr( + mcp_caller_auth, "DEFAULT_CALLERS_PATH", callers_path + ) + monkeypatch.delenv("KORA_MCP_BEARER_TOKEN", raising=False) + + r = client.post( + "/mcp", + headers={"Authorization": f"Bearer {token}"}, + json={ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "kora__request_pause", + "arguments": {"reason": "x"}, + }, + }, + ) + assert r.json()["error"]["code"] == -32001 + denial_calls = [ + c + for c in captured_audit_calls + if c["details"].get("result") == "capability_denied" + ] + assert len(denial_calls) == 1 + assert denial_calls[0]["details"]["caller_actor_id"] is None + assert denial_calls[0]["details"]["caller_actor_kind"] == "claude_pm_legacy" + + +def test_cap_gate_denial_audit_emitted_before_envelope( + client, empty_caps_token, monkeypatch +): + """Verify ordering: audit emit MUST run BEFORE the envelope returns. + + The audit fixture isn't enough — it captures regardless of order. + Patch emit_audit to raise inside; the response should still be + a -32001 envelope (audit failure must not mask the denial).""" + from kora_cli.listeners import mcp as mcp_mod + + def _failing_emit(*args, **kwargs): + raise RuntimeError("audit sink down") + + # Patch BOTH bound emit_audit references so the lazy import inside + # the helper resolves to the failing function. + monkeypatch.setattr( + "kora_cli.audit.jsonl_sink.emit_audit", _failing_emit + ) + monkeypatch.setattr( + "kora_cli.audit.emit_audit", _failing_emit + ) + + r = client.post( + "/mcp", + headers={"Authorization": f"Bearer {empty_caps_token}"}, + json={ + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": "kora__request_pause", + "arguments": {"reason": "x"}, + }, + }, + ) + # Denial response still surfaces cleanly. + assert r.status_code == 200 + assert r.json()["error"]["code"] == -32001 + assert r.json()["error"]["message"] == "capability_denied" + + +def test_cap_gate_denial_does_not_run_executor( + client, empty_caps_token, captured_audit_calls, monkeypatch +): + """Denied calls must NOT reach _ST2_DISPATCH executors. Verify by + spying on the dispatch table.""" + from kora_cli.listeners import mcp_tools + + dispatched: list = [] + original = mcp_tools.ST2_TOOL_DISPATCH["kora__request_pause"] + + async def _spy(params, caller): + dispatched.append((params, caller)) + return await original(params, caller) + + monkeypatch.setitem( + mcp_tools.ST2_TOOL_DISPATCH, "kora__request_pause", _spy + ) + + r = client.post( + "/mcp", + headers={"Authorization": f"Bearer {empty_caps_token}"}, + json={ + "jsonrpc": "2.0", + "id": 4, + "method": "tools/call", + "params": { + "name": "kora__request_pause", + "arguments": {"reason": "x"}, + }, + }, + ) + assert r.json()["error"]["code"] == -32001 + assert dispatched == [], ( + "denied call must not reach the executor; cap-gate fires first" + ) + + +# --------------------------------------------------------------------------- +# actor_id_required denial path +# --------------------------------------------------------------------------- + + +def test_actor_id_required_denial_writes_audit_row( + client, + stop_cap_no_actor_id_token, + captured_audit_calls, + monkeypatch, +): + """Caller with kora__request_stop cap but no actor_id triggers + the actor_id-required gate — second denial discriminator.""" + # Install a dummy coordinator so confirm_token can be read. + from kora_cli.daemon import DaemonCoordinator + from kora_cli.listeners import mcp as mcp_mod + + coord = DaemonCoordinator() + monkeypatch.setattr(mcp_mod, "current_coordinator", lambda: coord) + + r = client.post( + "/mcp", + headers={"Authorization": f"Bearer {stop_cap_no_actor_id_token}"}, + json={ + "jsonrpc": "2.0", + "id": 10, + "method": "tools/call", + "params": { + "name": "kora__request_stop", + "arguments": { + "reason": "trying without actor_id", + "level": 1, + "confirm_token": coord.daemon_session_id, + }, + }, + }, + ) + err = r.json()["error"] + assert err["code"] == -32001 + assert err["message"] == "actor_id_required_for_stop" + + denial_calls = [ + c + for c in captured_audit_calls + if c["details"].get("result") == "actor_id_required" + ] + assert len(denial_calls) == 1 + d = denial_calls[0]["details"] + assert d["tool_name"] == "kora__request_stop" + assert d["tool_kind"] == "mutating" + assert d["caller_actor_kind"] == "claude_pm_no_actor_id" + assert d["caller_actor_id"] is None # caller has no actor_id field + assert d["required_capability"] == "kora__request_stop" + assert d["duration_ms"] == 0 + assert d["tool_status"] == "not_allowed" + # The two denial discriminators are MUTUALLY EXCLUSIVE so the + # alert rule's detail_match doesn't conflate them. + cap_denials = [ + c + for c in captured_audit_calls + if c["details"].get("result") == "capability_denied" + ] + assert cap_denials == [] + + +# --------------------------------------------------------------------------- +# Success path — no denial audit +# --------------------------------------------------------------------------- + + +def test_successful_call_does_not_emit_denial_audit( + client, full_caps_token, captured_audit_calls, monkeypatch +): + """Authorized call → pause executor runs → success-path audit only. + + Verify the cap-gate doesn't accidentally fire on authorized calls.""" + # Put the holder into ACTIVE so the pause succeeds without + # exercising the InvalidStateTransitionError path. + from agent.operational_state import OperationalState, PrimaryState + from agent.operational_state_holder import OperationalStateHolder + from agent import operational_state_holder as h_mod + + h_mod._HOLDER = OperationalStateHolder( + OperationalState(primary_state=PrimaryState.ACTIVE) + ) + + r = client.post( + "/mcp", + headers={"Authorization": f"Bearer {full_caps_token}"}, + json={ + "jsonrpc": "2.0", + "id": 20, + "method": "tools/call", + "params": { + "name": "kora__request_pause", + "arguments": {"reason": "authorized"}, + }, + }, + ) + assert "result" in r.json(), r.json() + + # No capability_denied / actor_id_required entries. + denial_calls = [ + c + for c in captured_audit_calls + if c["details"].get("result") in ("capability_denied", "actor_id_required") + ] + assert denial_calls == [] + # There SHOULD be at least one success-path audit row from the + # pause executor's _emit_audit (its result is "active->paused"). + success_calls = [ + c + for c in captured_audit_calls + if "->" in str(c["details"].get("result", "")) + ] + assert len(success_calls) >= 1 + + +# --------------------------------------------------------------------------- +# Security sweep — bearer tokens never in audit details +# --------------------------------------------------------------------------- + + +def test_audit_details_never_contain_bearer_token( + client, monkeypatch, tmp_path, captured_audit_calls +): + """Walk every key+value in every captured audit row; the bearer + token marker MUST NOT appear anywhere.""" + token_marker = "kora-mcp-bearer-MUST-NOT-LEAK-IN-AUDIT-987" + callers_path = tmp_path / "mcp_callers.yaml" + _write_callers_yaml( + callers_path, + [ + { + "token_hash": _sha256(token_marker), + "actor_kind": "secure_caller", + "actor_id": _OPERATOR_ACTOR_UUID, + "allowed_caps": [], + } + ], + ) + from kora_cli.listeners import mcp_caller_auth + + monkeypatch.setattr( + mcp_caller_auth, "DEFAULT_CALLERS_PATH", callers_path + ) + monkeypatch.delenv("KORA_MCP_BEARER_TOKEN", raising=False) + + r = client.post( + "/mcp", + headers={"Authorization": f"Bearer {token_marker}"}, + json={ + "jsonrpc": "2.0", + "id": 30, + "method": "tools/call", + "params": { + "name": "kora__request_pause", + "arguments": {"reason": "token-leak-check"}, + }, + }, + ) + assert r.json()["error"]["code"] == -32001 + + for call in captured_audit_calls: + # Serialize the entire captured row + walk for the marker. + serialized = json.dumps(call, default=str) + assert token_marker not in serialized, ( + f"bearer token surfaced in audit row: {call!r}" + )