diff --git a/src/aegis_gitea_mcp/gitea/executor.py b/src/aegis_gitea_mcp/gitea/executor.py index cc3789b..1591493 100644 --- a/src/aegis_gitea_mcp/gitea/executor.py +++ b/src/aegis_gitea_mcp/gitea/executor.py @@ -14,7 +14,8 @@ Security properties (all covered by tests): 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 + webhook's stored ``authorization_header``, and ``data`` on Action secret operations) 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 @@ -26,6 +27,7 @@ Security properties (all covered by tests): from __future__ import annotations import base64 +import itertools import json from collections.abc import Iterable, Mapping from dataclasses import dataclass, field @@ -171,6 +173,8 @@ class _Call: sudo: str | None = None paginated: bool = False credential_issuing: bool = False + #: Action secret operations: a ``data`` field in the response is a secret value. + action_secret: bool = False header_data: str | None = None call_secrets: tuple[str, ...] = () @@ -183,6 +187,13 @@ class _Call: return next((value for key, value in self.query if key == name), None) +def _is_action_secret_path(segments: list[str]) -> bool: + """True for ``.../actions/secrets[/...]`` raw paths (an Action secrets endpoint).""" + return any( + first == "actions" and second == "secrets" for first, second in itertools.pairwise(segments) + ) + + def _classify(media_type: str) -> str: """Return ``json``, ``text`` or ``binary`` for a response media type.""" if not media_type or media_type == "application/json" or media_type.endswith("+json"): @@ -358,6 +369,7 @@ class GiteaExecutor: sudo=sudo, paginated=operation.paginated, credential_issuing=credential_issuing, + action_secret="Secret" in operation.operation_id, header_data=HEADER_DATA_OPERATIONS.get(operation.operation_id), call_secrets=tuple( sensitive_body_values(submitted, secret_data="Secret" in operation.operation_id) @@ -415,6 +427,7 @@ class GiteaExecutor: call.body = RequestBody(json=body, has_json=True) segments = call.display_path.split("/") call.call_secrets = tuple(sensitive_body_values(body, secret_data="secrets" in segments)) + call.action_secret = _is_action_secret_path(segments) call.paginated = any(key in {"page", "limit"} for key, _ in call.query) return await self._send(call, identity) @@ -624,8 +637,15 @@ class GiteaExecutor: 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)} + # masking cannot recognise them and SECRET_DETECTION_MODE may be off. Gitea + # never returns an Action secret's value; should a response carry a ``data`` + # field on a secret operation, it is the value and is redacted too. + keys = ( + SENSITIVE_RESPONSE_KEYS | {"data"} + if call.action_secret + else SENSITIVE_RESPONSE_KEYS + ) + payload = {**payload, "data": redact_fields(payload["data"], keys)} mode = self._settings.secret_detection_mode if mode != "off" and not exempt: if mode == "block": diff --git a/tests/security/test_secret_echo.py b/tests/security/test_secret_echo.py index e154729..6433d1b 100644 --- a/tests/security/test_secret_echo.py +++ b/tests/security/test_secret_echo.py @@ -143,3 +143,268 @@ async def test_credential_issuing_operations_still_return_their_credential( assert minted in text # ... but the audit file never contains it. assert minted not in json.dumps(audit_entries()) + + +# Action secrets: names and metadata only, never values + +VALUE = "Sup3r-Secret-Action-Value!" +SHORT_VALUE = "pw" + + +def listing_with_values() -> list[dict[str, Any]]: + """What Gitea sends (names and creation times), plus a hostile ``data`` field.""" + return [ + {"name": "DEPLOY_KEY", "created": "2026-01-01T00:00:00Z", "data": VALUE}, + {"name": "NPM_TOKEN", "created": "2026-01-02T00:00:00Z", "description": "ci"}, + ] + + +SECRET_LISTINGS = [ + ("gitea_actions_repo_list_secrets", {"owner": "o", "repo": "r"}, "/repos/o/r/actions/secrets"), + ("gitea_actions_org_list_secrets", {"org": "acme"}, "/orgs/acme/actions/secrets"), + ( + "gitea_api_request", + {"method": "GET", "path": "/repos/o/r/actions/secrets"}, + "/repos/o/r/actions/secrets", + ), + ( + "gitea_api_request", + {"method": "GET", "path": "/user/actions/secrets"}, + "/user/actions/secrets", + ), +] + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize(("tool", "arguments", "path"), SECRET_LISTINGS) +async def test_listing_secrets_returns_names_and_metadata_never_values( + httpx_mock: HTTPXMock, mode: str, tool: str, arguments: dict[str, Any], path: str +) -> None: + mock_user(httpx_mock) + httpx_mock.add_response(url=f"{API}{path}", json=listing_with_values()) + async with connected(server_settings(secret_detection_mode=mode)) as client: + text, structured = await call(client, tool, arguments) + + assert VALUE not in text and VALUE not in json.dumps(structured) + names = [item["name"] for item in structured["data"]] + assert names == ["DEPLOY_KEY", "NPM_TOKEN"] + assert structured["data"][0]["created"] == "2026-01-01T00:00:00Z" + assert structured["data"][1]["description"] == "ci" + + +async def test_a_secret_path_in_a_repository_is_not_mistaken_for_action_secrets( + httpx_mock: HTTPXMock, +) -> None: + mock_user(httpx_mock) + httpx_mock.add_response( + url=f"{API}/repos/o/r/contents/secrets", json=[{"name": "a", "data": "kept"}] + ) + async with connected() as client: + _, structured = await call( + client, "gitea_api_request", {"method": "GET", "path": "/repos/o/r/contents/secrets"} + ) + + assert structured["data"][0]["data"] == "kept" + + +SECRET_WRITES = [ + ( + "gitea_actions_update_repo_secret", + {"owner": "o", "repo": "r", "secretname": "DEPLOY_KEY"}, + "/repos/o/r/actions/secrets/DEPLOY_KEY", + ), + ( + "gitea_actions_update_org_secret", + {"org": "acme", "secretname": "DEPLOY_KEY"}, + "/orgs/acme/actions/secrets/DEPLOY_KEY", + ), + ( + "gitea_actions_update_user_secret", + {"secretname": "DEPLOY_KEY"}, + "/user/actions/secrets/DEPLOY_KEY", + ), +] + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize("value", [VALUE, SHORT_VALUE]) +@pytest.mark.parametrize(("tool", "arguments", "path"), SECRET_WRITES) +async def test_updating_a_secret_never_echoes_the_submitted_value( + httpx_mock: HTTPXMock, + caplog: pytest.LogCaptureFixture, + mode: str, + value: str, + tool: str, + arguments: dict[str, Any], + path: str, +) -> None: + caplog.set_level("DEBUG") + mock_user(httpx_mock) + # Success echoing the value, then a validation error repeating it (as Gitea 422s do). + httpx_mock.add_response( + method="PUT", + url=f"{API}{path}", + status_code=201, + json={"echo": value, "name": "DEPLOY_KEY"}, + ) + httpx_mock.add_response( + method="PUT", + url=f"{API}{path}", + status_code=422, + json={"message": f"invalid secret data: {value}", "errors": [f"data {value} too long"]}, + ) + async with connected(server_settings(secret_detection_mode=mode)) as client: + ok = await call(client, tool, {**arguments, "body": {"data": value}}) + failed = await call(client, tool, {**arguments, "body": {"data": value}}) + + for text, structured in (ok, failed): + assert value not in text + assert value not in json.dumps(structured) + assert "[REDACTED]" in ok[0] and "[REDACTED]" in failed[0] + assert value not in json.dumps(audit_entries()) and value not in caplog.text + + +async def test_the_raw_tool_also_hides_a_submitted_secret_value(httpx_mock: HTTPXMock) -> None: + mock_user(httpx_mock) + httpx_mock.add_response( + method="PUT", + url=f"{API}/repos/o/r/actions/secrets/K", + status_code=422, + json={"message": f"bad {VALUE}"}, + ) + async with connected() as client: + text, _ = await call( + client, + "gitea_api_request", + { + "method": "PUT", + "path": "/repos/o/r/actions/secrets/K", + "body": {"data": VALUE}, + }, + ) + + assert VALUE not in text and "[REDACTED]" in text + + +# Other submitted credential fields + +CREDENTIAL_BODIES = [ + ( + "gitea_admin_create_user", + {}, + "POST", + "/admin/users", + {"username": "bob", "email": "bob@example.com", "password": "Pa55word-Of-Bob"}, + ), + ( + "gitea_hook_repo_create_hook", + {"owner": "o", "repo": "r"}, + "POST", + "/repos/o/r/hooks", + {"type": "gitea", "authorization_header": "Bearer hook-target-token-123", "config": {}}, + ), + ( + "gitea_repo_migrate", + {}, + "POST", + "/repos/migrate", + { + "clone_addr": "https://example.org/x.git", + "repo_name": "x", + "auth_token": "migration-source-token-123", + "auth_password": "migration-source-pass-123", + "aws_secret_access_key": "aws-secret-access-key-value-123", + }, + ), + ( + "gitea_api_request", + {}, + "POST", + "/orgs/acme/hooks", + {"config": {"secret": "shared-hook-secret-123", "url": "https://x.example"}}, + ), +] + + +def secret_values(body: Any) -> list[str]: + found: list[str] = [] + if isinstance(body, dict): + for key, item in body.items(): + if isinstance(item, str) and key in { + "password", + "authorization_header", + "auth_token", + "auth_password", + "aws_secret_access_key", + "secret", + }: + found.append(item) + else: + found.extend(secret_values(item)) + return found + + +@pytest.mark.parametrize("mode", MODES) +@pytest.mark.parametrize(("tool", "arguments", "method", "path", "body"), CREDENTIAL_BODIES) +async def test_submitted_credentials_are_redacted_from_results( + httpx_mock: HTTPXMock, + mode: str, + tool: str, + arguments: dict[str, Any], + method: str, + path: str, + body: dict[str, Any], +) -> None: + mock_user(httpx_mock) + secrets = secret_values(body) + assert secrets + httpx_mock.add_response( + method=method, + url=f"{API}{path}", + status_code=422, + json={"message": "rejected: " + " ".join(secrets), "errors": secrets}, + ) + request = ( + {"method": method, "path": path, "body": body} + if tool == "gitea_api_request" + else {**arguments, "body": body} + ) + async with connected(server_settings(secret_detection_mode=mode)) as client: + text, structured = await call(client, tool, request) + + for secret in secrets: + assert secret not in text and secret not in json.dumps(structured) + assert secret not in json.dumps(audit_entries()) + assert "[REDACTED]" in text + + +# Credential-issuing operations: the credential goes to the caller, nowhere else + + +async def test_issued_credentials_reach_the_caller_but_not_the_audit_file_or_the_logs( + httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture +) -> None: + caplog.set_level("DEBUG") + mock_user(httpx_mock) + token = "9f8e7d6c5b4a39281706f5e4d3c2b1a099887766" + registration = "runner-registration-token-0123456789ABCDEF" + httpx_mock.add_response( + method="POST", url=f"{API}/users/alice/tokens", json={"id": 1, "name": "ci", "sha1": token} + ) + httpx_mock.add_response( + method="POST", + url=f"{API}/admin/actions/runners/registration-token", + headers={"token": registration}, + status_code=200, + ) + async with connected(server_settings(secret_detection_mode="block")) as client: + created, _ = await call( + client, "gitea_user_create_token", {"username": "alice", "body": {"name": "ci"}} + ) + runner, _ = await call(client, "gitea_admin_create_runner_registration_token", {}) + + assert token in created and registration in runner # the documented exemption (W-07) + audit_text = json.dumps(audit_entries()) + for secret in (token, registration): + assert secret not in audit_text + assert secret not in caplog.text