Hiddenden/AegisGitea-MCP
5
0 Fork 0
Code Issues 2 Pull Requests Actions Packages Projects 1 Releases 1 Wiki Activity
Files
446ec11954245bd1e2dedab903a4b065d0567403
AegisGitea-MCP/tests/security/test_redaction.py
T

389 lines
14 KiB
Python
Raw Blame History

"""Credentials never reach results, errors, logs, the audit file or ``repr`` output.
Every secret the server knows (the Gitea token, the MCP bearer, the OAuth client secret,
the store key, a user's Gitea access and refresh tokens and our own issued tokens) is
planted in Gitea's answers, successful and failed, and in the incoming requests.
"""
from __future__ import annotations
import json
import logging
from typing import Any
import pytest
from pytest_httpx import HTTPXMock
from aegis_gitea_mcp.audit import get_audit_logger
from aegis_gitea_mcp.identity import Identity
from aegis_gitea_mcp.oauth.provider import PendingAuthorization
from aegis_gitea_mcp.oauth.store import SessionRecord
from aegis_gitea_mcp.oauth.upstream import UpstreamTokens
from aegis_gitea_mcp.transports import AccessLogQueryFilter, run_http
from tests.test_oauth_helpers import (
API,
APP_CLIENT_SECRET,
BEARER_TOKEN,
ENCRYPTION_KEY,
GITEA_TOKEN,
USER_REFRESH,
USER_TOKEN,
browser,
build_oauth_app,
mcp_client,
oauth_settings,
sign_in,
)
@pytest.fixture(autouse=True)
def _env(mock_env: None) -> None:
"""The audit logger and the rate limiter read the process-wide settings."""
def static_secrets() -> list[str]:
return [GITEA_TOKEN, BEARER_TOKEN, APP_CLIENT_SECRET, ENCRYPTION_KEY, USER_TOKEN, USER_REFRESH]
def everything_secret(extra: list[str]) -> str:
return " ".join([*static_secrets(), *extra])
def assert_clean(text: str, secrets: list[str]) -> None:
for secret in secrets:
assert secret not in text, f"secret leaked: {secret[:6]}..."
@pytest.mark.parametrize("mode", ["off", "mask", "block"])
async def test_no_secret_reaches_a_result_the_audit_file_or_the_logs(
httpx_mock: HTTPXMock,
caplog: pytest.LogCaptureFixture,
capsys: pytest.CaptureFixture[str],
mode: str,
) -> None:
caplog.set_level(logging.DEBUG)
oauth = build_oauth_app(oauth_settings(secret_detection_mode=mode))
flow = sign_in(browser(oauth.app), httpx_mock)
# The client secret the SDK issues at registration is plain hex without a recognisable
# prefix: it is stored (encrypted) but cannot be pattern masked, and Gitea never sees it.
issued = [flow.tokens["access_token"], flow.tokens["refresh_token"]]
# SECRET_DETECTION_MODE=off disables pattern matching, which is what protects our own
# issued agm_ tokens (only hashes of them live on the server, so they cannot be redacted
# exactly); every other secret is redacted exactly in all modes.
secrets = static_secrets() if mode == "off" else [*static_secrets(), *issued]
# The admin identity is not the OAuth user: Gitea never shows it that user's tokens, so
# only the server's own secrets (and, with pattern masking, our issued tokens) apply.
admin_secrets = [GITEA_TOKEN, BEARER_TOKEN, APP_CLIENT_SECRET, ENCRYPTION_KEY]
blob = everything_secret(issued)
# The admin identity's lookup, then success and failure bodies that echo every secret.
httpx_mock.add_response(url=f"{API}/user", json={"login": "admin"}, is_reusable=True)
echo = {"is_reusable": True}
httpx_mock.add_response(
url=f"{API}/repos/octo/ok", json={"description": blob, blob: "key"}, **echo
)
httpx_mock.add_response(
url=f"{API}/repos/octo/bad",
status_code=422,
json={"message": blob, "errors": [blob]},
**echo,
)
httpx_mock.add_response(
url=f"{API}/repos/octo/text", text=blob, headers={"content-type": "text/plain"}, **echo
)
httpx_mock.add_response(url=f"{API}/repos/octo/boom", status_code=500, text=blob, **echo)
results: list[Any] = []
# One app per identity: the SDK's session manager can only run once per app.
apps = [
(oauth.app, flow.tokens["access_token"]),
(build_oauth_app(oauth.settings).app, BEARER_TOKEN),
]
for app, token in apps:
async with mcp_client(app, token) as mcp:
for repo in ("ok", "bad", "boom"):
results.append(
await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": repo})
)
results.append(
await mcp.call_tool(
"gitea_api_request", {"method": "GET", "path": "/repos/octo/text"}
)
)
texts = [result.content[0].text for result in results]
per_identity = [
(results[:4], secrets),
(results[4:], [*admin_secrets, *(issued if mode != "off" else [])]),
]
for group, forbidden in per_identity:
for result in group:
assert_clean(result.content[0].text, forbidden)
assert_clean(json.dumps(result.structured_content), forbidden)
assert any("[REDACTED]" in text for text in texts) # the redaction happened, not an empty body
everything = [*secrets, *admin_secrets]
assert_clean(get_audit_logger().log_path.read_text(encoding="utf-8"), everything)
assert_clean(caplog.text, everything)
captured = capsys.readouterr()
assert_clean(captured.out + captured.err, everything)
async def test_exceptions_during_a_call_do_not_leak_secrets(
httpx_mock: HTTPXMock, caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch
) -> None:
caplog.set_level(logging.DEBUG)
oauth = build_oauth_app()
secrets = static_secrets()
async def explode(*args: Any, **kwargs: Any) -> None:
raise RuntimeError("failure carrying " + " ".join(secrets))
monkeypatch.setattr(oauth.extras.executor, "execute", explode)
httpx_mock.add_response(url=f"{API}/user", json={"login": "admin"}, is_reusable=True)
async with mcp_client(oauth.app, BEARER_TOKEN) as mcp:
result = await mcp.call_tool("gitea_repo_get", {"owner": "o", "repo": "r"})
assert result.is_error
assert_clean(result.content[0].text, secrets)
assert_clean(caplog.text, secrets)
assert_clean(get_audit_logger().log_path.read_text(encoding="utf-8"), secrets)
assert "RuntimeError" not in result.content[0].text # no exception details by default
async def test_validation_errors_never_contain_argument_values(httpx_mock: HTTPXMock) -> None:
oauth = build_oauth_app()
planted = "sekret-value-" + GITEA_TOKEN
async with mcp_client(oauth.app, BEARER_TOKEN) as mcp:
results = [
await mcp.call_tool("gitea_repo_get", {"owner": planted + "?x#y", "repo": "r"}),
await mcp.call_tool("gitea_repo_get", {"owner": 12345, "repo": planted}),
await mcp.call_tool("gitea_repo_get", {"owner": "o", "repo": "r", "extra": planted}),
await mcp.call_tool("gitea_api_request", {"method": planted, "path": "/x"}),
await mcp.call_tool("gitea_api_request", {"method": "GET", "path": planted}),
await mcp.call_tool(
"gitea_api_request",
{"method": "GET", "path": "/user", "sudo": "not valid " + planted},
),
]
for result in results:
assert result.is_error
assert planted not in result.content[0].text
assert GITEA_TOKEN not in result.content[0].text
assert planted not in get_audit_logger().log_path.read_text(encoding="utf-8")
# Access logs
def uvicorn_access_record(path_with_query: str) -> logging.LogRecord:
return logging.LogRecord(
name="uvicorn.access",
level=logging.INFO,
pathname=__file__,
lineno=1,
msg='%s - "%s %s HTTP/%s" %d',
args=("127.0.0.1:5000", "GET", path_with_query, "1.1", 302),
exc_info=None,
)
@pytest.mark.parametrize(
"path",
[
"/oauth/callback?code=one-time-gitea-code&state=state-secret-value",
"/oauth/consent?request=pending-request-secret",
"/authorize?client_id=c&code_challenge=abc&state=client-state-secret",
],
)
def test_access_log_lines_drop_the_query_string(path: str) -> None:
record = uvicorn_access_record(path)
assert AccessLogQueryFilter().filter(record)
line = record.getMessage()
assert "[redacted]" in line
assert path.split("?")[0] in line
for secret in (
"one-time-gitea-code",
"state-secret-value",
"pending-request-secret",
"client-state-secret",
"code_challenge",
):
assert secret not in line
def test_the_runner_installs_the_access_log_filter(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
from aegis_gitea_mcp import transports
access_logger = logging.getLogger("uvicorn.access")
monkeypatch.setattr(access_logger, "filters", [])
monkeypatch.setattr(transports.uvicorn, "run", lambda *a, **k: None)
caplog.set_level(logging.INFO, logger="uvicorn.access")
assert run_http(oauth_settings()) == 0
access_logger.info(
'%s - "%s %s HTTP/%s" %d',
"127.0.0.1:5000",
"GET",
"/oauth/callback?code=one-time-gitea-code&state=state-secret-value",
"1.1",
302,
)
assert "one-time-gitea-code" not in caplog.text
assert "[redacted]" in caplog.text
# repr
def test_repr_of_settings_and_secret_holding_objects_contains_no_secret() -> None:
settings = oauth_settings()
identity = Identity(login="alice", mode="oauth", token=USER_TOKEN, redact=(USER_REFRESH,))
session = SessionRecord(
session_id="session-id-value",
client_id="c1",
login="alice",
is_admin=False,
gitea_access_token=USER_TOKEN,
gitea_refresh_token=USER_REFRESH,
gitea_expires_at=None,
scopes=["gitea"],
resource="https://mcp.example.com/mcp",
created_at=0.0,
previous_refresh_hash="hash-of-previous-refresh",
)
upstream = UpstreamTokens(access_token=USER_TOKEN, refresh_token=USER_REFRESH, expires_at=None)
pending = PendingAuthorization(
client_id="c1",
client_name="Client",
redirect_uri="https://client.example/cb",
redirect_uri_provided_explicitly=True,
code_challenge="challenge",
client_state="client-state-secret",
scopes=["gitea"],
resource="https://mcp.example.com/mcp",
code_verifier="pkce-verifier-secret",
csrf_hash="csrf-hash-secret",
binding_hash="binding-hash-secret",
)
rendered = [
repr(settings),
str(settings),
repr(identity),
repr(session),
repr(upstream),
repr(pending),
repr(settings.gitea_token),
repr(settings.model_dump()),
]
for text in rendered:
assert_clean(text, [*static_secrets(), "client-state-secret", "pkce-verifier-secret"])
assert "csrf-hash-secret" not in text and "binding-hash-secret" not in text
assert "hash-of-previous-refresh" not in text
assert "**********" in repr(settings.gitea_token)
def test_secrets_in_object_keys_are_masked_like_values() -> None:
from aegis_gitea_mcp.security import sanitize_data
token = "agm_" + "A1b2C3d4E5f6G7h8I9j0K1l2M3n4O5p6Q7r8S9t0U1v"
nested = {"outer": [{token: "value", "plain": token}]}
masked = sanitize_data(nested, mode="mask")
blocked = sanitize_data(nested, mode="block")
assert token not in json.dumps(masked) and token not in json.dumps(blocked)
assert masked["outer"][0]["plain"] != token
assert blocked != nested
# Configuration errors and a real stdio process
@pytest.mark.parametrize(
("variable", "value"),
[
("GITEA_URL", "https://user:" + GITEA_TOKEN + "@gitea.example.com"),
("GITEA_URL", "https://gitea.example.com?token=" + GITEA_TOKEN),
("MCP_BEARER_TOKEN", "short-" + "x" * 5),
("PUBLIC_BASE_URL", "http://mcp.example.com/" + GITEA_TOKEN),
("OAUTH_SCOPES", GITEA_TOKEN),
("TOKEN_ENCRYPTION_KEY", "tooshort-" + ENCRYPTION_KEY[:5]),
("MCP_PORT", GITEA_TOKEN),
("MCP_PATH", GITEA_TOKEN),
("FORWARDED_ALLOW_IPS", GITEA_TOKEN),
("LOG_LEVEL", GITEA_TOKEN),
],
)
def test_configuration_errors_name_the_setting_but_never_echo_its_value(
variable: str,
value: str,
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
from aegis_gitea_mcp import cli
from aegis_gitea_mcp.cli import main
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
monkeypatch.setenv("GITEA_TOKEN", GITEA_TOKEN)
monkeypatch.setenv(variable, value)
monkeypatch.setattr(cli, "configure_logging", lambda level: None)
assert main(["serve", "--transport", "http"]) == 2
err = capsys.readouterr().err
assert variable in err
assert value not in err and GITEA_TOKEN not in err
assert "Traceback" not in err
async def test_a_real_stdio_process_keeps_every_secret_out_of_stdout_stderr_and_the_audit_file(
tmp_path: Any,
) -> None:
import sys
from mcp import Client
from mcp.client.stdio import StdioServerParameters, stdio_client
secrets = [GITEA_TOKEN, BEARER_TOKEN, APP_CLIENT_SECRET, ENCRYPTION_KEY]
env = {
"GITEA_URL": "http://127.0.0.1:9", # nothing listens: every call fails
"GITEA_TOKEN": GITEA_TOKEN,
"MCP_BEARER_TOKEN": BEARER_TOKEN,
"GITEA_OAUTH_CLIENT_SECRET": APP_CLIENT_SECRET,
"TOKEN_ENCRYPTION_KEY": ENCRYPTION_KEY,
"DATA_DIR": str(tmp_path / "state"),
"AUDIT_LOG_PATH": str(tmp_path / "state" / "audit.log"),
"REQUEST_TIMEOUT_SECONDS": "2",
"LOG_LEVEL": "DEBUG",
}
params = StdioServerParameters(
command=sys.executable,
args=["-m", "aegis_gitea_mcp.cli", "serve", "--transport", "stdio"],
env=env,
cwd=str(tmp_path),
)
stderr_path = tmp_path / "stderr.log"
with stderr_path.open("w", encoding="utf-8") as errlog:
async with Client(stdio_client(params, errlog=errlog)) as client:
results = [
await client.call_tool("gitea_repo_get", {"owner": "o", "repo": "r"}),
await client.call_tool("gitea_repo_get", {"owner": 5}),
await client.call_tool("gitea_api_request", {"method": "GET", "path": "/user"}),
]
for result in results:
assert result.is_error
assert_clean(result.content[0].text, secrets) # type: ignore[union-attr]
assert_clean(stderr_path.read_text(encoding="utf-8"), secrets)
assert_clean((tmp_path / "state" / "audit.log").read_text(encoding="utf-8"), secrets)
Reference in new issue View Git Blame Copy Permalink