diff --git a/src/aegis_gitea_mcp/security.py b/src/aegis_gitea_mcp/security.py index a3bdafe..e8b2bb3 100644 --- a/src/aegis_gitea_mcp/security.py +++ b/src/aegis_gitea_mcp/security.py @@ -90,7 +90,11 @@ def sanitize_data(value: Any, mode: str = "mask") -> Any: Sanitized payload value. """ if isinstance(value, dict): - return {str(key): sanitize_data(item, mode=mode) for key, item in value.items()} + # Keys are data too: a secret can sit in an object key (``{"": 1}``). + return { + sanitize_data(str(key), mode=mode): sanitize_data(item, mode=mode) + for key, item in value.items() + } if isinstance(value, list): return [sanitize_data(item, mode=mode) for item in value] diff --git a/tests/security/test_auth_required.py b/tests/security/test_auth_required.py new file mode 100644 index 0000000..5dec5c5 --- /dev/null +++ b/tests/security/test_auth_required.py @@ -0,0 +1,302 @@ +"""The MCP endpoint rejects every request that does not carry a valid credential.""" + +from __future__ import annotations + +import hashlib +import hmac +from typing import Any + +import pytest +from pytest_httpx import HTTPXMock +from starlette.applications import Starlette +from starlette.routing import Route +from starlette.testclient import TestClient + +from aegis_gitea_mcp import auth as auth_module +from aegis_gitea_mcp.auth import StaticBearerVerifier +from aegis_gitea_mcp.oauth.provider import TOKEN_PREFIX +from aegis_gitea_mcp.transports import build_http_app +from tests.test_oauth_helpers import ( + API, + GITEA_TOKEN, + USER_TOKEN, + FakeClock, + browser, + build_oauth_app, + mcp_client, + mcp_status_lifespan, + sign_in, +) +from tests.test_server_helpers import BEARER_TOKEN, server_settings + +MCP_BODY = {"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}} +HOST = "http://127.0.0.1:8080" + + +@pytest.fixture(autouse=True) +def _env(mock_env: None) -> None: + """The audit logger and the rate limiter read the process-wide settings.""" + + +def bearer_app() -> Starlette: + return build_http_app(server_settings(mcp_bearer_token=BEARER_TOKEN)) + + +def post_mcp(client: TestClient, authorization: str | None) -> Any: + headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"} + if authorization is not None: + headers["Authorization"] = authorization + return client.post("/mcp", json=MCP_BODY, headers=headers) + + +# No credential, wrong credential + + +def test_mcp_without_a_bearer_is_401_with_a_challenge_on_the_bearer_app() -> None: + client = TestClient(bearer_app(), base_url=HOST) + + response = post_mcp(client, None) + + assert response.status_code == 401 + assert response.headers["www-authenticate"].lower().startswith("bearer") + + +def test_mcp_without_a_bearer_is_401_with_a_challenge_on_the_oauth_app() -> None: + client = browser(build_oauth_app().app) + + response = post_mcp(client, None) + + assert response.status_code == 401 + challenge = response.headers["www-authenticate"] + assert challenge.lower().startswith("bearer") + assert "resource_metadata" in challenge # tells clients where to find the OAuth server + + +@pytest.mark.parametrize("method", ["GET", "POST", "DELETE"]) +def test_every_method_on_the_endpoint_needs_authentication(method: str) -> None: + client = TestClient(bearer_app(), base_url=HOST) + + assert client.request(method, "/mcp").status_code == 401 + + +WRONG_BEARERS = { + "same length": "x" * len(BEARER_TOKEN), + "one character off": BEARER_TOKEN[:-1] + ("A" if BEARER_TOKEN[-1] != "A" else "B"), + "shorter": BEARER_TOKEN[:-1], + "longer": BEARER_TOKEN + "x", + "prefix only": BEARER_TOKEN[:8], + "doubled": BEARER_TOKEN * 2, + "upper case": BEARER_TOKEN.upper(), + "trailing space": BEARER_TOKEN + " ", + "trailing newline": BEARER_TOKEN + "\n", + "another deployment": "another-deployments-bearer-0123456789abcdef", + "admin gitea token": GITEA_TOKEN, + "null byte": BEARER_TOKEN + "\x00", +} + + +@pytest.mark.parametrize("name", WRONG_BEARERS) +def test_a_wrong_bearer_is_401(name: str) -> None: + client = TestClient(bearer_app(), base_url=HOST) + + try: + response = post_mcp(client, f"Bearer {WRONG_BEARERS[name]}") + except Exception as exc: # an invalid header value is refused before it leaves the client + assert "header" in str(exc).lower() or isinstance(exc, (ValueError, UnicodeError)) + return + + assert response.status_code == 401 + + +@pytest.mark.parametrize( + "header", + [ + "", + "Bearer", + "Bearer ", + f"Basic {BEARER_TOKEN}", + f"token {BEARER_TOKEN}", + f"Token {BEARER_TOKEN}", + BEARER_TOKEN, + f"Bearer {BEARER_TOKEN} extra", + f"Digest {BEARER_TOKEN}", + ], +) +def test_wrong_scheme_or_shape_of_the_authorization_header_is_401(header: str) -> None: + client = TestClient(bearer_app(), base_url=HOST) + + assert post_mcp(client, header).status_code == 401 + + +def test_the_right_bearer_is_not_401() -> None: + assert mcp_status_lifespan(bearer_app(), BEARER_TOKEN) != 401 + + +def test_the_bearer_is_never_accepted_from_a_query_string_or_a_cookie() -> None: + client = TestClient(bearer_app(), base_url=HOST) + + by_query = client.post(f"/mcp?access_token={BEARER_TOKEN}&token={BEARER_TOKEN}", json=MCP_BODY) + client.cookies.set("token", BEARER_TOKEN) + by_cookie = client.post("/mcp", json=MCP_BODY) + + assert by_query.status_code == 401 and by_cookie.status_code == 401 + + +# Constant-time comparison + + +async def test_the_static_bearer_is_compared_in_constant_time_on_digests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + seen: list[tuple[Any, Any]] = [] + real = hmac.compare_digest + + def spy(a: Any, b: Any) -> bool: + seen.append((a, b)) + return real(a, b) + + monkeypatch.setattr(auth_module.hmac, "compare_digest", spy) + verifier = StaticBearerVerifier(BEARER_TOKEN) + + for candidate in ("", "x", BEARER_TOKEN[:-1], BEARER_TOKEN, BEARER_TOKEN + "x" * 500): + accepted = await verifier.verify_token(candidate) + assert (accepted is not None) == (candidate == BEARER_TOKEN) + + assert len(seen) == 5 + # Digests of equal length whatever the candidate length: nothing about the length leaks. + assert {len(a) for a, _ in seen} == {len(b) for _, b in seen} == {32} + assert all(isinstance(a, bytes) and isinstance(b, bytes) for a, b in seen) + assert seen[3][0] == hashlib.sha256(BEARER_TOKEN.encode()).digest() + + +def test_the_source_has_no_plain_equality_on_the_secret() -> None: + import inspect + + source = inspect.getsource(StaticBearerVerifier) + + assert "compare_digest" in source + assert "== self._token" not in source and "self._token ==" not in source + + +async def test_the_access_token_of_the_static_bearer_never_carries_the_secret() -> None: + access = await StaticBearerVerifier(BEARER_TOKEN).verify_token(BEARER_TOKEN) + + assert access is not None + assert BEARER_TOKEN not in repr(access) and BEARER_TOKEN not in access.model_dump_json() + + +# OAuth tokens + + +def test_unknown_expired_revoked_and_foreign_oauth_tokens_are_401(httpx_mock: HTTPXMock) -> None: + clock = FakeClock() + oauth = build_oauth_app(clock=clock) + web = browser(oauth.app) + flow = sign_in(web, httpx_mock) + access = flow.tokens["access_token"] + assert access.startswith(TOKEN_PREFIX) + assert mcp_status_lifespan(oauth.app, access) != 401 + + unknown = TOKEN_PREFIX + "A" * 43 + assert post_mcp(web, f"Bearer {unknown}").status_code == 401 + assert post_mcp(web, f"Bearer {TOKEN_PREFIX}").status_code == 401 + assert post_mcp(web, f"Bearer {access}x").status_code == 401 + assert post_mcp(web, f"Bearer {access[:-1]}").status_code == 401 + + # Bound to another resource: the token does not work here. + (session,) = oauth.store.sessions.values() + real_resource = session.resource + session.resource = "https://other.example.com/mcp" + assert post_mcp(web, f"Bearer {access}").status_code == 401 + session.resource = real_resource + + # Revoked by its client. + revoked = web.post( + "/revoke", + data={"token": access, "client_id": flow.client_id, "client_secret": flow.client_secret}, + ) + assert revoked.status_code == 200 + assert post_mcp(web, f"Bearer {access}").status_code == 401 + + +def test_an_expired_oauth_token_is_401(httpx_mock: HTTPXMock) -> None: + clock = FakeClock() + oauth = build_oauth_app(clock=clock) + web = browser(oauth.app) + flow = sign_in(web, httpx_mock) + access = flow.tokens["access_token"] + + clock.advance(3600 + 1) + + assert post_mcp(web, f"Bearer {access}").status_code == 401 + + +async def test_an_oauth_token_never_resolves_to_the_admin_identity( + httpx_mock: HTTPXMock, +) -> None: + oauth = build_oauth_app() + flow = sign_in(browser(oauth.app), httpx_mock) + httpx_mock.add_response(url=f"{API}/repos/octo/demo", json={}) + httpx_mock.add_response(url=f"{API}/admin/users", status_code=403, json={"message": "no"}) + + async with mcp_client(oauth.app, flow.tokens["access_token"]) as mcp: + await mcp.call_tool("gitea_repo_get", {"owner": "octo", "repo": "demo"}) + await mcp.call_tool("gitea_admin_search_users", {}) + + calls = [ + r for r in httpx_mock.get_requests() if "/repos/" in str(r.url) or "/admin/" in str(r.url) + ] + assert len(calls) == 2 + for request in calls: + assert request.headers["Authorization"] == f"Bearer {USER_TOKEN}" + assert GITEA_TOKEN not in request.headers["Authorization"] + + +def test_an_oauth_only_deployment_has_no_static_bearer_to_guess(httpx_mock: HTTPXMock) -> None: + from tests.test_oauth_helpers import oauth_settings + + oauth = build_oauth_app(oauth_settings(gitea_token=None, mcp_bearer_token=None)) + web = browser(oauth.app) + + assert post_mcp(web, f"Bearer {BEARER_TOKEN}").status_code == 401 + assert post_mcp(web, None).status_code == 401 + + +# Route inventory + +#: Routes reachable without authentication, with the reason each one exists. +PUBLIC_ROUTES = { + "/": "service banner", + "/health": "container and client health check", + "/.well-known/oauth-protected-resource": "RFC 9728 discovery (bare path, OAuth only)", + "/.well-known/oauth-protected-resource/mcp": "RFC 9728 discovery", + "/.well-known/oauth-authorization-server": "RFC 8414 discovery (OAuth only)", + "/authorize": "OAuth authorization endpoint (OAuth only)", + "/token": "OAuth token endpoint (OAuth only)", + "/register": "OAuth dynamic client registration (OAuth only)", + "/revoke": "OAuth revocation (OAuth only)", + "/oauth/consent": "consent page (OAuth only)", + "/oauth/callback": "Gitea callback (OAuth only)", +} + + +def inventory(app: Starlette) -> tuple[set[str], set[str]]: + """Paths without authentication, and paths guarded by the SDK's bearer middleware.""" + open_paths: set[str] = set() + guarded: set[str] = set() + for route in app.routes: + assert isinstance(route, Route) + endpoint = type(route.endpoint).__name__ + (guarded if endpoint == "RequireAuthMiddleware" else open_paths).add(route.path) + return open_paths, guarded + + +def test_the_routes_reachable_without_authentication_are_exactly_the_known_ones() -> None: + oauth_open, oauth_guarded = inventory(build_oauth_app().app) + bearer_open, bearer_guarded = inventory(bearer_app()) + + assert oauth_open == set(PUBLIC_ROUTES) + assert bearer_open <= set(PUBLIC_ROUTES) + assert {"/", "/health"} <= bearer_open + # The MCP endpoint is the only guarded route, and it is guarded in both modes. + assert oauth_guarded == bearer_guarded == {"/mcp"} diff --git a/tests/security/test_redaction.py b/tests/security/test_redaction.py new file mode 100644 index 0000000..eedf5e8 --- /dev/null +++ b/tests/security/test_redaction.py @@ -0,0 +1,306 @@ +"""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 diff --git a/tests/security/test_startup_fail_closed.py b/tests/security/test_startup_fail_closed.py new file mode 100644 index 0000000..ee948ac --- /dev/null +++ b/tests/security/test_startup_fail_closed.py @@ -0,0 +1,223 @@ +"""The server refuses to start unless it is safe to: authentication off loopback, valid settings.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from aegis_gitea_mcp import cli, transports +from aegis_gitea_mcp.cli import main +from aegis_gitea_mcp.transports import run_http, run_stdio +from tests.test_oauth_helpers import oauth_settings +from tests.test_server_helpers import BEARER_TOKEN, GITEA_TOKEN, server_settings + +WILDCARD_V4 = "0.0.0.0" # noqa: S104 (a bind address used as test input, never bound) +WILDCARD_V6 = "::" +NON_LOOPBACK = [ + WILDCARD_V4, + WILDCARD_V6, + "[::]", + "192.168.1.10", + "10.0.0.5", + "203.0.113.5", + "2001:db8::1", + "mcp.example.com", + "localhost.evil.example", + "gitea.internal", + "127.1", # not a canonical loopback address: refused rather than guessed + "0", + "128.0.0.1", +] +LOOPBACK = ["127.0.0.1", "127.0.0.2", "127.255.255.254", "::1", "[::1]", "localhost", "LOCALHOST"] + + +class Served: + """Replaces ``uvicorn.run`` and records whether anything was started.""" + + def __init__(self) -> None: + self.calls: list[dict[str, Any]] = [] + + def __call__(self, app: Any, **kwargs: Any) -> None: + self.calls.append(kwargs) + + +@pytest.fixture(autouse=True) +def served(mock_env: None, monkeypatch: pytest.MonkeyPatch) -> Served: + spy = Served() + monkeypatch.setattr(transports.uvicorn, "run", spy) + monkeypatch.setattr(cli, "configure_logging", lambda level: None) + return spy + + +@pytest.mark.parametrize("host", NON_LOOPBACK) +def test_a_non_loopback_bind_without_authentication_exits_2( + host: str, served: Served, capsys: pytest.CaptureFixture[str] +) -> None: + assert run_http(server_settings(mcp_host=host)) == 2 + + captured = capsys.readouterr() + assert served.calls == [] + assert "without HTTP authentication" in captured.err + assert captured.out == "" + + +@pytest.mark.parametrize("host", NON_LOOPBACK) +def test_the_serve_command_refuses_the_same_binds( + host: str, served: Served, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") + monkeypatch.setenv("GITEA_TOKEN", GITEA_TOKEN) + + assert main(["serve", "--transport", "http", "--host", host]) == 2 + + assert served.calls == [] + assert "without HTTP authentication" in capsys.readouterr().err + + +@pytest.mark.parametrize("host", NON_LOOPBACK[:4]) +def test_a_bearer_is_enough_to_bind_publicly(host: str, served: Served) -> None: + assert run_http(server_settings(mcp_host=host, mcp_bearer_token=BEARER_TOKEN)) == 0 + + assert len(served.calls) == 1 + + +@pytest.mark.parametrize("host", NON_LOOPBACK[:4]) +def test_complete_oauth_is_enough_to_bind_publicly(host: str, served: Served) -> None: + settings = oauth_settings(mcp_host=host, mcp_bearer_token=None, gitea_token=None) + + assert run_http(settings) == 0 + assert len(served.calls) == 1 + + +@pytest.mark.parametrize("host", LOOPBACK) +def test_loopback_binds_may_run_without_authentication(host: str, served: Served) -> None: + assert run_http(server_settings(mcp_host=host)) == 0 + + assert len(served.calls) == 1 + + +# Half-configured authentication + + +@pytest.mark.parametrize( + "missing", + [ + "gitea_oauth_client_id", + "gitea_oauth_client_secret", + "token_encryption_key", + "public_base_url", + ], +) +def test_half_configured_oauth_exits_2_and_never_falls_back_to_no_auth( + missing: str, served: Served, capsys: pytest.CaptureFixture[str] +) -> None: + settings = oauth_settings(mcp_host=WILDCARD_V4, mcp_bearer_token=None, **{missing: None}) + + assert run_http(settings) == 2 + + assert served.calls == [] + assert missing.upper() in capsys.readouterr().err + + +def test_oauth_with_a_bearer_but_no_admin_token_exits_2( + served: Served, capsys: pytest.CaptureFixture[str] +) -> None: + assert run_http(oauth_settings(gitea_token=None)) == 2 + + assert served.calls == [] + assert "GITEA_TOKEN" in capsys.readouterr().err + + +@pytest.mark.parametrize("bearer", ["", "short", "x" * 31]) +def test_a_bearer_shorter_than_32_characters_is_rejected_at_configuration( + bearer: str, served: Served, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") + monkeypatch.setenv("GITEA_TOKEN", GITEA_TOKEN) + monkeypatch.setenv("MCP_BEARER_TOKEN", bearer) + monkeypatch.setenv("MCP_HOST", WILDCARD_V4) + + code = main(["serve", "--transport", "http"]) + + if bearer == "": + # A blank value counts as unset, so the unauthenticated public bind is refused instead. + assert code == 2 and "without HTTP authentication" in capsys.readouterr().err + else: + assert code == 2 + assert "invalid configuration" in capsys.readouterr().err + assert served.calls == [] + + +def test_an_http_public_base_url_on_a_public_host_is_rejected( + served: Served, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") + monkeypatch.setenv("GITEA_TOKEN", GITEA_TOKEN) + monkeypatch.setenv("PUBLIC_BASE_URL", "http://mcp.example.com") + + assert main(["serve", "--transport", "http"]) == 2 + + assert served.calls == [] + assert "PUBLIC_BASE_URL" in capsys.readouterr().err + + +def test_a_missing_gitea_token_exits_2_for_admin_token_mode_and_stdio( + served: Served, capsys: pytest.CaptureFixture[str] +) -> None: + assert run_http(server_settings(gitea_token=None, mcp_bearer_token=BEARER_TOKEN)) == 2 + assert run_http(server_settings(gitea_token=None)) == 2 + assert run_stdio(server_settings(gitea_token=None)) == 2 + + captured = capsys.readouterr() + assert served.calls == [] + assert "GITEA_TOKEN" in captured.err + assert captured.out == "" # stdout is reserved for JSON-RPC + + +def test_an_unwritable_audit_log_exits_2_before_serving( + tmp_path: Path, + served: Served, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + blocker = tmp_path / "blocker" + blocker.write_text("a file where a directory is needed") + path = blocker / "audit.log" + monkeypatch.setenv("AUDIT_LOG_PATH", str(path)) + + assert run_http(server_settings(audit_log_path=path)) == 2 + assert run_stdio(server_settings(audit_log_path=path)) == 2 + + assert served.calls == [] + assert "audit log" in capsys.readouterr().err + + +def test_a_symlinked_audit_path_exits_2( + tmp_path: Path, + served: Served, + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + target = tmp_path / "elsewhere.log" + target.write_text("") + link = tmp_path / "audit.log" + link.symlink_to(target) + monkeypatch.setenv("AUDIT_LOG_PATH", str(link)) + + assert run_http(server_settings(audit_log_path=link)) == 2 + + assert served.calls == [] + assert "symbolic link" in capsys.readouterr().err + assert target.read_text() == "" + + +def test_unknown_disabled_tools_stop_startup( + served: Served, capsys: pytest.CaptureFixture[str] +) -> None: + assert run_http(server_settings(disabled_tools=["gitea_no_such_tool"])) == 2 + assert run_stdio(server_settings(disabled_tools=["gitea_no_such_tool"])) == 2 + + assert served.calls == [] + assert "gitea_no_such_tool" in capsys.readouterr().err