fix: record the calling client and only schema-known names in audit entries (SEC-12, SEC-18)

tests/security/test_audit_fail_closed.py

@@ -3,13 +3,16 @@
 from __future__ import annotations
 
 import json
+import re
 from typing import Any
 
 import pytest
+from mcp import MCPError
 from pytest_httpx import HTTPXMock
 
 from aegis_gitea_mcp.audit import get_audit_logger
-from tests.test_server_helpers import API, connected, mock_user
+from tests.test_oauth_helpers import BEARER_TOKEN, browser, build_oauth_app, mcp_client, sign_in
+from tests.test_server_helpers import API, audit_entries, connected, mock_user
 
 REPO_ARGS = {"owner": "octo", "repo": "demo"}
 
@@ -77,3 +80,72 @@ async def test_the_refusal_does_not_leak_the_failure_reason() -> None:
 
     text = result.content[0].text  # type: ignore[union-attr]
     assert "No space left" not in text and "OSError" not in text
+
+
+# SEC-12: who made the call
+
+
+async def test_oauth_call_entry_names_the_client_and_never_a_token_or_session_id(
+    httpx_mock: HTTPXMock,
+) -> None:
+    oauth = build_oauth_app()
+    flow = sign_in(browser(oauth.app), httpx_mock)
+    session_ids = list(oauth.store.sessions)
+    httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={"full_name": "octo/demo"})
+    access = flow.tokens["access_token"]
+
+    async with mcp_client(oauth.app, access) as mcp:
+        result = await mcp.call_tool("gitea_repo_get", REPO_ARGS)
+    assert not result.is_error
+
+    audit_text = get_audit_logger().log_path.read_text(encoding="utf-8")
+    (entry,) = audit_entries()
+    assert entry["client_id"] == flow.client_id
+    assert re.fullmatch(r"[0-9a-f]{12}", entry["session"])
+    assert entry["client_ip"] == "127.0.0.1"
+    assert entry["identity_login"] == "alice"
+    assert access not in audit_text
+    assert flow.tokens["refresh_token"] not in audit_text
+    assert all(session_id not in audit_text for session_id in session_ids)
+
+
+async def test_static_bearer_call_entry_names_the_static_client(httpx_mock: HTTPXMock) -> None:
+    mock_user(httpx_mock)
+    httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={})
+    oauth = build_oauth_app()
+    async with mcp_client(oauth.app, BEARER_TOKEN) as mcp:
+        await mcp.call_tool("gitea_repo_get", REPO_ARGS)
+
+    (entry,) = audit_entries()
+    assert entry["client_id"] == "aegis-static-bearer"
+    assert "session" not in entry
+    assert BEARER_TOKEN not in get_audit_logger().log_path.read_text(encoding="utf-8")
+
+
+# SEC-18: caller-chosen names
+
+TOKEN_LIKE = "0123456789abcdef0123456789abcdef01234567"
+
+
+async def test_extra_argument_names_are_counted_not_recorded() -> None:
+    async with connected() as client:
+        result = await client.call_tool("gitea_repo_get", {**REPO_ARGS, TOKEN_LIKE: "x"})
+
+    assert result.is_error
+    assert TOKEN_LIKE not in result.content[0].text  # type: ignore[union-attr]
+    (entry,) = audit_entries()
+    assert entry["params"] == {"argument_names": ["owner", "repo"], "rejected_argument_count": 1}
+    assert TOKEN_LIKE not in get_audit_logger().log_path.read_text(encoding="utf-8")
+
+
+async def test_unknown_tool_names_are_neither_echoed_nor_recorded() -> None:
+    name = "gitea_" + TOKEN_LIKE
+    async with connected() as client:
+        with pytest.raises(MCPError) as caught:
+            await client.call_tool(name, {TOKEN_LIKE: 1})
+
+    assert TOKEN_LIKE not in caught.value.message
+    (entry,) = audit_entries()
+    assert entry["tool_name"] == f"(unknown tool, {len(name)} chars)"
+    assert entry["params"] == {"argument_names": [], "rejected_argument_count": 1}
+    assert TOKEN_LIKE not in get_audit_logger().log_path.read_text(encoding="utf-8")

tests/test_server.py

@@ -354,9 +354,9 @@ async def test_unknown_tool_is_a_protocol_error_and_is_audited() -> None:
             await client.call_tool("gitea_does_not_exist", {"a": 1})
 
     assert caught.value.code == -32602
-    assert "gitea_does_not_exist" in caught.value.message
+    assert "gitea_does_not_exist" not in caught.value.message
     entry = audit_entries()[-1]
-    assert entry["tool_name"] == "gitea_does_not_exist"
+    assert entry["tool_name"] == "(unknown tool, 20 chars)"
     assert entry["outcome"] == "unknown_tool"