test: security suite for auth, startup, redaction; fix: mask secrets in object keys

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