152 lines
5.5 KiB
Python
152 lines
5.5 KiB
Python
"""The audit log is fail closed after startup too (SEC-03) and names the client (SEC-12)."""
|
|
|
|
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_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"}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _env(mock_env: None) -> None:
|
|
"""The audit logger reads the process-wide settings."""
|
|
|
|
|
|
class FailingFile:
|
|
"""Stands in for the open audit file and fails every write like a full disk."""
|
|
|
|
def __init__(self) -> None:
|
|
self.attempts = 0
|
|
|
|
def write(self, data: str) -> int:
|
|
self.attempts += 1
|
|
raise OSError(28, "No space left on device")
|
|
|
|
def flush(self) -> None: # pragma: no cover - never reached
|
|
raise OSError(28, "No space left on device")
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
|
|
def payload_of(result: Any) -> dict[str, Any]:
|
|
decoded: dict[str, Any] = json.loads(result.content[0].text)
|
|
return decoded
|
|
|
|
|
|
async def test_calls_stop_when_the_audit_log_stops_accepting_writes(
|
|
httpx_mock: HTTPXMock,
|
|
) -> None:
|
|
mock_user(httpx_mock)
|
|
httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={"name": "demo"}, is_reusable=True)
|
|
async with connected() as client:
|
|
first = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
assert not first.is_error
|
|
sent_before = len(httpx_mock.get_requests())
|
|
|
|
audit = get_audit_logger()
|
|
healthy_file = audit._log_file
|
|
audit._log_file = FailingFile() # type: ignore[assignment]
|
|
second = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
mutation = await client.call_tool(
|
|
"gitea_issue_create_issue", {**REPO_ARGS, "body": {"title": "x"}}
|
|
)
|
|
|
|
assert second.is_error and mutation.is_error
|
|
assert payload_of(second)["error"]["kind"] == "audit_unavailable"
|
|
assert payload_of(mutation)["error"]["kind"] == "audit_unavailable"
|
|
# Gitea was not contacted at all while the log was broken.
|
|
assert len(httpx_mock.get_requests()) == sent_before
|
|
|
|
audit._log_file = healthy_file
|
|
resumed = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
assert not resumed.is_error
|
|
|
|
|
|
async def test_the_refusal_does_not_leak_the_failure_reason() -> None:
|
|
async with connected() as client:
|
|
get_audit_logger()._log_file = FailingFile() # type: ignore[assignment]
|
|
result = await client.call_tool("gitea_repo_get", REPO_ARGS)
|
|
|
|
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")
|