diff --git a/src/aegis_gitea_mcp/gitea/encoding.py b/src/aegis_gitea_mcp/gitea/encoding.py index a7f32fb..dc11a5d 100644 --- a/src/aegis_gitea_mcp/gitea/encoding.py +++ b/src/aegis_gitea_mcp/gitea/encoding.py @@ -96,6 +96,14 @@ _SENSITIVE_BODY_KEYS = frozenset( ) _MIN_SENSITIVE_VALUE_CHARS = 4 +#: Response fields whose non-empty string values are replaced with ``[REDACTED]`` +#: in every result except those of credential-issuing operations (whose purpose +#: is to return a credential). Gitea fills ``authorization_header`` on every +#: webhook read, so a stored header would otherwise reach the model. Keep this +#: next to :data:`_SENSITIVE_BODY_KEYS` so a new secret-bearing field is added +#: in one place. +SENSITIVE_RESPONSE_KEYS = _SENSITIVE_BODY_KEYS + class ArgumentError(ValueError): """An argument cannot be turned into a safe request (message never echoes the value).""" diff --git a/src/aegis_gitea_mcp/gitea/executor.py b/src/aegis_gitea_mcp/gitea/executor.py index 48ce832..cc3789b 100644 --- a/src/aegis_gitea_mcp/gitea/executor.py +++ b/src/aegis_gitea_mcp/gitea/executor.py @@ -13,6 +13,9 @@ Security properties (all covered by tests): * the identity token and ``Identity.redact``, the configured Gitea token, the MCP bearer token, the OAuth client secret, ``extra_secrets`` and secret values submitted in a request body are replaced with ``[REDACTED]`` in every result, always; +* secret-bearing response fields (:data:`~.encoding.SENSITIVE_RESPONSE_KEYS`, for example a + webhook's stored ``authorization_header``) are redacted by key name in every result except + those of :data:`CREDENTIAL_ISSUING_OPERATIONS`, whatever ``SECRET_DETECTION_MODE`` says; * secret-pattern masking/blocking follows ``SECRET_DETECTION_MODE``, except for :data:`CREDENTIAL_ISSUING_OPERATIONS` whose purpose is to return a freshly minted credential (exact redaction still applies there); @@ -38,6 +41,7 @@ from aegis_gitea_mcp.config import Settings from aegis_gitea_mcp.errors import kind_for_status, sanitize_message from aegis_gitea_mcp.gitea.encoding import ( RAW_METHODS, + SENSITIVE_RESPONSE_KEYS, ArgumentError, PayloadTooLargeError, RequestBody, @@ -50,7 +54,12 @@ from aegis_gitea_mcp.gitea.encoding import ( validate_sudo, ) from aegis_gitea_mcp.gitea.pagination import build_pagination -from aegis_gitea_mcp.gitea.redaction import normalize_secrets, redact_exact, redact_text +from aegis_gitea_mcp.gitea.redaction import ( + normalize_secrets, + redact_exact, + redact_fields, + redact_text, +) from aegis_gitea_mcp.identity import Identity from aegis_gitea_mcp.security import sanitize_data @@ -613,6 +622,10 @@ class GiteaExecutor: """Redact, mask or block, and bound the payload size.""" secrets = self._secrets(identity, call) payload = redact_exact(payload, secrets) + if not exempt and "data" in payload: + # Stored secrets (webhook authorization headers, ...) by field name: pattern + # masking cannot recognise them and SECRET_DETECTION_MODE may be off. + payload = {**payload, "data": redact_fields(payload["data"], SENSITIVE_RESPONSE_KEYS)} mode = self._settings.secret_detection_mode if mode != "off" and not exempt: if mode == "block": diff --git a/src/aegis_gitea_mcp/gitea/redaction.py b/src/aegis_gitea_mcp/gitea/redaction.py index 0a0dea1..1df8a8f 100644 --- a/src/aegis_gitea_mcp/gitea/redaction.py +++ b/src/aegis_gitea_mcp/gitea/redaction.py @@ -7,7 +7,7 @@ they have, in every string of a payload (values and dict keys). from __future__ import annotations -from collections.abc import Iterable +from collections.abc import Collection, Iterable from typing import Any REDACTED = "[REDACTED]" @@ -41,3 +41,36 @@ def redact_exact(value: Any, secrets: tuple[str, ...]) -> Any: if isinstance(value, (list, tuple)): return [redact_exact(item, secrets) for item in value] return value + + +def redact_fields(value: Any, keys: Collection[str], *, max_depth: int = 128) -> Any: + """Replace the string values of secret-bearing fields, by key name, at any depth. + + A non-empty string under a key in ``keys`` (compared lower-cased) becomes + ``[REDACTED]``; an empty value stays so the presence of the field is still + visible. Containers under such a key are searched like any other value. + Nesting deeper than ``max_depth`` is returned unchanged (the executor bounds + JSON depth before this runs). + """ + + def walk(item: Any, depth: int) -> Any: + if depth > max_depth: + return item + if isinstance(item, dict): + return { + key: ( + REDACTED + if isinstance(key, str) and key.lower() in keys and _is_secret_value(sub) + else walk(sub, depth + 1) + ) + for key, sub in item.items() + } + if isinstance(item, (list, tuple)): + return [walk(sub, depth + 1) for sub in item] + return item + + return walk(value, 0) + + +def _is_secret_value(value: Any) -> bool: + return isinstance(value, str) and bool(value) diff --git a/tests/security/test_secret_echo.py b/tests/security/test_secret_echo.py new file mode 100644 index 0000000..e154729 --- /dev/null +++ b/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())