AegisGitea-MCP/tests/security/test_startup_fail_closed.py

"""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