fix: redact stored webhook authorization headers and other secret fields by key name (SEC-02)

tests/security/test_secret_echo.py

@@ -0,0 +1,145 @@
+"""Secret values that pass through tool calls are never echoed back to the model.
+
+Covers stored webhook authorization headers (SEC-02), Action secrets and
+submitted credential fields. Credential-issuing operations are the documented
+exception: they return the credential they just minted.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+import pytest
+from mcp import Client
+from pytest_httpx import HTTPXMock
+
+from aegis_gitea_mcp.gitea.executor import CREDENTIAL_ISSUING_OPERATIONS
+from tests.test_server_helpers import API, audit_entries, connected, mock_user, server_settings
+
+STORED_HEADER = "Bearer webhook-target-secret-0123456789"
+BASIC_HEADER = "Basic dXNlcjpwYXNzd29yZC1vZi10aGUtd2ViaG9vaw=="
+MODES = ["off", "mask", "block"]
+
+
+@pytest.fixture(autouse=True)
+def _env(mock_env: None) -> None:
+    """The audit logger reads the process-wide settings."""
+
+
+def hook(header: str | None = STORED_HEADER) -> dict[str, Any]:
+    return {
+        "id": 3,
+        "type": "gitea",
+        "active": True,
+        "config": {"url": "https://ci.example.org/hook", "content_type": "json"},
+        "authorization_header": header,
+    }
+
+
+async def call(client: Client, name: str, arguments: dict[str, Any]) -> tuple[str, Any]:
+    result = await client.call_tool(name, arguments)
+    return result.content[0].text, result.structured_content  # type: ignore[union-attr]
+
+
+HOOK_CASES = [
+    ("gitea_hook_repo_list_hooks", {"owner": "o", "repo": "r"}, "GET", "/repos/o/r/hooks", True),
+    (
+        "gitea_hook_repo_get_hook",
+        {"owner": "o", "repo": "r", "id": 3},
+        "GET",
+        "/repos/o/r/hooks/3",
+        False,
+    ),
+    (
+        "gitea_hook_repo_edit_hook",
+        {"owner": "o", "repo": "r", "id": 3, "body": {"active": True}},
+        "PATCH",
+        "/repos/o/r/hooks/3",
+        False,
+    ),
+    ("gitea_hook_org_list_hooks", {"org": "acme"}, "GET", "/orgs/acme/hooks", True),
+    ("gitea_hook_org_get_hook", {"org": "acme", "id": 3}, "GET", "/orgs/acme/hooks/3", False),
+    ("gitea_hook_user_list_hooks", {}, "GET", "/user/hooks", True),
+    ("gitea_hook_user_get_hook", {"id": 3}, "GET", "/user/hooks/3", False),
+    ("gitea_admin_list_hooks", {}, "GET", "/admin/hooks", True),
+    ("gitea_admin_get_hook", {"id": 3}, "GET", "/admin/hooks/3", False),
+    (
+        "gitea_api_request",
+        {"method": "GET", "path": "/repos/o/r/hooks"},
+        "GET",
+        "/repos/o/r/hooks",
+        True,
+    ),
+]
+
+
+@pytest.mark.parametrize("mode", MODES)
+@pytest.mark.parametrize(("tool", "arguments", "method", "path", "is_list"), HOOK_CASES)
+@pytest.mark.parametrize("header", [STORED_HEADER, BASIC_HEADER])
+async def test_stored_webhook_authorization_header_is_never_returned(
+    httpx_mock: HTTPXMock,
+    mode: str,
+    tool: str,
+    arguments: dict[str, Any],
+    method: str,
+    path: str,
+    is_list: bool,
+    header: str,
+) -> None:
+    mock_user(httpx_mock)
+    stored = hook(header)
+    httpx_mock.add_response(
+        method=method, url=f"{API}{path}", json=[stored, hook(None)] if is_list else stored
+    )
+    async with connected(server_settings(secret_detection_mode=mode)) as client:
+        text, structured = await call(client, tool, arguments)
+
+    assert header not in text
+    assert header not in json.dumps(structured)
+    assert "[REDACTED]" in text
+    # The rest of the hook stays readable.
+    assert "ci.example.org" in text
+    assert header not in json.dumps(audit_entries())
+
+
+async def test_empty_authorization_header_stays_visible_as_empty(httpx_mock: HTTPXMock) -> None:
+    mock_user(httpx_mock)
+    httpx_mock.add_response(url=f"{API}/repos/o/r/hooks/3", json=hook(""))
+    async with connected() as client:
+        _, structured = await call(
+            client, "gitea_hook_repo_get_hook", {"owner": "o", "repo": "r", "id": 3}
+        )
+
+    assert structured["data"]["authorization_header"] == ""
+
+
+async def test_nested_secret_fields_are_redacted_by_key_name(httpx_mock: HTTPXMock) -> None:
+    mock_user(httpx_mock)
+    httpx_mock.add_response(
+        url=f"{API}/repos/o/r/hooks",
+        json=[{"config": {"url": "https://x.example", "secret": "plain-shared-secret"}}],
+    )
+    async with connected(server_settings(secret_detection_mode="off")) as client:
+        text, _ = await call(client, "gitea_hook_repo_list_hooks", {"owner": "o", "repo": "r"})
+
+    assert "plain-shared-secret" not in text
+
+
+async def test_credential_issuing_operations_still_return_their_credential(
+    httpx_mock: HTTPXMock,
+) -> None:
+    assert "userCreateToken" in CREDENTIAL_ISSUING_OPERATIONS
+    mock_user(httpx_mock)
+    minted = "a" * 20 + "0123456789abcdef"
+    httpx_mock.add_response(
+        method="POST", url=f"{API}/users/alice/tokens", json={"id": 1, "name": "ci", "sha1": minted}
+    )
+    async with connected() as client:
+        text, _ = await call(
+            client, "gitea_user_create_token", {"username": "alice", "body": {"name": "ci"}}
+        )
+
+    assert minted in text
+    # ... but the audit file never contains it.
+    assert minted not in json.dumps(audit_entries())