Latte
517c675048
Require consent before sending the token over plain http to a non-loopback host, strip control characters from server-supplied text, restrict env values and tokens to printable ASCII, validate the URLs like other env values and fsync the directory after the atomic replace.
845 lines
31 KiB
Python
845 lines
31 KiB
Python
"""Tests for ``aegis-gitea-mcp setup`` (HTTP mocked with pytest-httpx, files in tmp_path)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import stat
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from pytest_httpx import HTTPXMock
|
||
|
||
from aegis_gitea_mcp import setup_cmd
|
||
from aegis_gitea_mcp.cli import main
|
||
from aegis_gitea_mcp.config import Settings
|
||
from aegis_gitea_mcp.setup_cmd import (
|
||
EnvEntry,
|
||
missing_scopes,
|
||
parse_env_text,
|
||
render_env,
|
||
write_env_atomic,
|
||
)
|
||
from aegis_gitea_mcp.setup_template import ENV_TEMPLATE
|
||
|
||
GITEA = "https://git.test"
|
||
API = f"{GITEA}/api/v1"
|
||
PUBLIC = "https://mcp.example.com"
|
||
TOKEN = "tok_SECRET_0123456789abcdef"
|
||
CLIENT_SECRET = "gto_CLIENTSECRET_9876543210"
|
||
OLD_SECRET = "gto_OLDSECRET_1111111111"
|
||
ALL_WRITE = [f"write:{c}" for c in setup_cmd.GITEA_SCOPE_CATEGORIES]
|
||
LIST_URL = re.compile(re.escape(f"{API}/user/applications/oauth2") + r"\?page=\d+&limit=\d+")
|
||
|
||
|
||
def app_json(
|
||
app_id: int = 7,
|
||
client_id: str = "cid-7",
|
||
redirects: list[str] | None = None,
|
||
secret: str = "",
|
||
confidential: bool = True,
|
||
name: str = "aegis-gitea-mcp",
|
||
) -> dict[str, Any]:
|
||
return {
|
||
"id": app_id,
|
||
"name": name,
|
||
"client_id": client_id,
|
||
"client_secret": secret,
|
||
"redirect_uris": redirects if redirects is not None else [f"{PUBLIC}/oauth/callback"],
|
||
"confidential_client": confidential,
|
||
"skip_secondary_authorization": False,
|
||
}
|
||
|
||
|
||
def mock_instance(
|
||
httpx_mock: HTTPXMock,
|
||
*,
|
||
scopes: list[str] | None = None,
|
||
apps: list[dict[str, Any]] | None = None,
|
||
token_status: int = 200,
|
||
) -> None:
|
||
"""Register the version, user (admin), token and app-list endpoints."""
|
||
httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"})
|
||
httpx_mock.add_response(url=f"{API}/user", json={"login": "root", "is_admin": True})
|
||
if token_status == 200:
|
||
httpx_mock.add_response(
|
||
url=f"{API}/token",
|
||
json={"id": 1, "name": "setup", "scopes": ["all"] if scopes is None else scopes},
|
||
)
|
||
else:
|
||
httpx_mock.add_response(url=f"{API}/token", status_code=token_status, json={})
|
||
if apps is not None:
|
||
httpx_mock.add_response(url=LIST_URL, json=apps, headers={"x-total-count": str(len(apps))})
|
||
|
||
|
||
def mock_create(httpx_mock: HTTPXMock, **overrides: Any) -> None:
|
||
httpx_mock.add_response(
|
||
method="POST",
|
||
url=f"{API}/user/applications/oauth2",
|
||
status_code=201,
|
||
json=app_json(secret=CLIENT_SECRET, **overrides),
|
||
)
|
||
|
||
|
||
def run_setup(*extra: str, public: bool = True, oauth: bool = True) -> int:
|
||
argv = ["setup", "--gitea-url", GITEA, "--yes", *extra]
|
||
if public:
|
||
argv += ["--public-base-url", PUBLIC]
|
||
if not oauth:
|
||
argv.append("--no-oauth")
|
||
return main(argv)
|
||
|
||
|
||
def read_env(path: Path) -> dict[str, str]:
|
||
return {k: e.value for k, e in parse_env_text(path.read_text()).items()}
|
||
|
||
|
||
def write_existing_env(path: Path, **values: str) -> None:
|
||
path.write_text("".join(f"{k}={v}\n" for k, v in values.items()))
|
||
path.chmod(0o600)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def token_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
||
monkeypatch.setenv("GITEA_TOKEN", TOKEN)
|
||
|
||
|
||
def assert_no_secrets(out: str, *extra: str) -> None:
|
||
for secret in (TOKEN, CLIENT_SECRET, OLD_SECRET, *extra):
|
||
assert secret not in out
|
||
|
||
|
||
# --- happy path --------------------------------------------------------------
|
||
|
||
|
||
def test_happy_path_creates_app_and_env(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock, apps=[])
|
||
mock_create(httpx_mock)
|
||
assert run_setup() == 0
|
||
|
||
post = next(r for r in httpx_mock.get_requests() if r.method == "POST")
|
||
assert json.loads(post.content) == {
|
||
"name": "aegis-gitea-mcp",
|
||
"redirect_uris": [f"{PUBLIC}/oauth/callback"],
|
||
"confidential_client": True,
|
||
"skip_secondary_authorization": False,
|
||
}
|
||
assert all(r.headers["authorization"] == f"token {TOKEN}" for r in httpx_mock.get_requests())
|
||
|
||
env_path = tmp_path / ".env"
|
||
assert stat.S_IMODE(env_path.stat().st_mode) == 0o600
|
||
env = read_env(env_path)
|
||
assert env["GITEA_URL"] == GITEA
|
||
assert env["GITEA_TOKEN"] == TOKEN
|
||
assert env["GITEA_OAUTH_CLIENT_ID"] == "cid-7"
|
||
assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET
|
||
assert env["OAUTH_ENABLED"] == "true"
|
||
assert env["PUBLIC_BASE_URL"] == PUBLIC
|
||
assert len(env["MCP_BEARER_TOKEN"]) >= 32
|
||
assert len(env["TOKEN_ENCRYPTION_KEY"]) >= 32
|
||
assert env["MCP_BEARER_TOKEN"] != env["TOKEN_ENCRYPTION_KEY"]
|
||
text = env_path.read_text()
|
||
assert text.startswith("# AegisGitea MCP configuration.")
|
||
assert "# --- Gitea" in text
|
||
assert text.index("GITEA_URL=") < text.index("MCP_TRANSPORT=") < text.index("OAUTH_ENABLED=")
|
||
|
||
captured = capsys.readouterr()
|
||
for stream in (captured.out, captured.err):
|
||
assert_no_secrets(stream, env["MCP_BEARER_TOKEN"], env["TOKEN_ENCRYPTION_KEY"])
|
||
assert "Gitea reachable: version 1.99.0" in captured.out
|
||
assert "root (site administrator)" in captured.out
|
||
assert "cid-7" in captured.out
|
||
assert "grep '^MCP_BEARER_TOKEN=' " in captured.out
|
||
|
||
|
||
def test_written_env_is_accepted_by_settings(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
|
||
mock_instance(httpx_mock, apps=[])
|
||
mock_create(httpx_mock)
|
||
assert run_setup() == 0
|
||
settings = Settings(_env_file=str(tmp_path / ".env")) # type: ignore[call-arg]
|
||
assert settings.oauth_enabled
|
||
assert settings.oauth_config_errors() == []
|
||
assert settings.http_auth_configured()
|
||
assert settings.gitea_url == GITEA
|
||
|
||
|
||
def test_printed_configs(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||
mock_instance(httpx_mock, apps=[])
|
||
mock_create(httpx_mock)
|
||
assert run_setup() == 0
|
||
out = capsys.readouterr().out
|
||
assert f"--env GITEA_URL={GITEA} --env GITEA_TOKEN=<your-gitea-token>" in out
|
||
assert "uvx aegis-gitea-mcp" in out
|
||
assert '"mcpServers"' in out
|
||
assert (
|
||
f"claude mcp add --transport http aegis-gitea {PUBLIC}/mcp "
|
||
'--header "Authorization: Bearer <MCP_BEARER_TOKEN>"'
|
||
) in out
|
||
assert "claude.ai custom connector" in out
|
||
assert "Mistral Le Chat" in out
|
||
assert "git.hiddenden.cafe/hiddenden/aegisgitea-mcp:dev" in out
|
||
assert "MCP_HOST: 0.0.0.0" in out
|
||
assert '"127.0.0.1:8080:8080"' in out
|
||
assert "no-new-privileges:true" in out
|
||
assert "mcp.example.com {" in out
|
||
assert "reverse_proxy 127.0.0.1:8080" in out
|
||
assert "proxy_buffering off;" in out
|
||
assert f"curl {PUBLIC}/health" in out
|
||
assert "aegis-gitea-mcp serve" in out
|
||
assert "MCP_TRANSPORT=http aegis-gitea-mcp serve" in out
|
||
|
||
|
||
# --- verification failures -----------------------------------------------------
|
||
|
||
|
||
def test_non_admin_token_fails(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"})
|
||
httpx_mock.add_response(url=f"{API}/user", json={"login": "bob", "is_admin": False})
|
||
assert run_setup() == 1
|
||
err = capsys.readouterr().err
|
||
assert "not a site administrator" in err
|
||
assert "Settings > Applications > Generate New Token" in err
|
||
assert not (tmp_path / ".env").exists()
|
||
assert_no_secrets(err)
|
||
|
||
|
||
def test_rejected_token_fails(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||
httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"})
|
||
httpx_mock.add_response(url=f"{API}/user", status_code=401, json={"message": "bad token"})
|
||
assert run_setup() == 1
|
||
assert "token rejected" in capsys.readouterr().err
|
||
|
||
|
||
def test_missing_scopes_listed(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock, scopes=["write:repository", "read:user", "write:issue"])
|
||
assert run_setup() == 1
|
||
err = capsys.readouterr().err
|
||
listed = err.split("lacks required scopes: ")[1].split(". ")[0].split(", ")
|
||
assert "write:admin" in listed
|
||
assert "write:user" in listed
|
||
assert "write:repository" not in listed
|
||
assert "write:issue" not in listed
|
||
assert not (tmp_path / ".env").exists()
|
||
|
||
|
||
@pytest.mark.parametrize("scopes", [["all"], ["ALL", "read:user"], ALL_WRITE])
|
||
def test_full_scopes_accepted(httpx_mock: HTTPXMock, scopes: list[str]) -> None:
|
||
mock_instance(httpx_mock, scopes=scopes)
|
||
assert run_setup(oauth=False) == 0
|
||
|
||
|
||
def test_missing_scopes_helper() -> None:
|
||
assert missing_scopes(["all"]) == []
|
||
assert missing_scopes(ALL_WRITE) == []
|
||
assert missing_scopes(["read:user"]) == list(setup_cmd.REQUIRED_SCOPES)
|
||
|
||
|
||
def test_token_endpoint_missing_warns_and_continues(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock, token_status=404)
|
||
assert run_setup(oauth=False) == 0
|
||
out = capsys.readouterr().out
|
||
assert "Warning: could not read the token scopes" in out
|
||
assert "not verified" in out
|
||
|
||
|
||
def test_empty_scope_list_warns(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||
mock_instance(httpx_mock, scopes=[])
|
||
assert run_setup(oauth=False) == 0
|
||
assert "could not be verified" in capsys.readouterr().out
|
||
|
||
|
||
def test_public_only_scope_warns(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||
mock_instance(httpx_mock, scopes=["all", "public-only"])
|
||
assert run_setup(oauth=False) == 0
|
||
assert "public-only" in capsys.readouterr().out
|
||
|
||
|
||
def test_unreachable_instance(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
httpx_mock.add_exception(httpx.ConnectError("boom"))
|
||
assert run_setup() == 1
|
||
err = capsys.readouterr().err
|
||
assert "Gitea is not reachable" in err
|
||
assert "git.test" in err
|
||
assert not (tmp_path / ".env").exists()
|
||
|
||
|
||
def test_timeout_message(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||
httpx_mock.add_exception(httpx.ReadTimeout("slow"))
|
||
assert run_setup() == 1
|
||
assert "did not answer within 15 seconds" in capsys.readouterr().err
|
||
|
||
|
||
def test_not_gitea_response(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||
httpx_mock.add_response(url=f"{API}/version", text="<html>login</html>")
|
||
assert run_setup() == 1
|
||
assert "did not return Gitea API JSON" in capsys.readouterr().err
|
||
|
||
|
||
def test_redirect_is_not_followed(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
httpx_mock.add_response(
|
||
url=f"{API}/version", status_code=301, headers={"location": "https://other.test/x"}
|
||
)
|
||
assert run_setup() == 1
|
||
err = capsys.readouterr().err
|
||
assert "redirect to https://other.test/x" in err
|
||
assert len(httpx_mock.get_requests()) == 1
|
||
|
||
|
||
def test_gitea_error_text_is_redacted(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99.0"})
|
||
httpx_mock.add_response(
|
||
url=f"{API}/user", status_code=500, json={"message": f"boom {TOKEN} boom"}
|
||
)
|
||
assert run_setup() == 1
|
||
err = capsys.readouterr().err
|
||
assert "boom *** boom" in err
|
||
assert_no_secrets(err)
|
||
|
||
|
||
# --- OAuth app -------------------------------------------------------------------
|
||
|
||
|
||
def test_oauth_app_reused_keeps_secret(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
write_existing_env(
|
||
tmp_path / ".env",
|
||
GITEA_OAUTH_CLIENT_ID="cid-7",
|
||
GITEA_OAUTH_CLIENT_SECRET=OLD_SECRET,
|
||
MCP_BEARER_TOKEN="b" * 40,
|
||
TOKEN_ENCRYPTION_KEY="k" * 40,
|
||
)
|
||
mock_instance(httpx_mock, apps=[app_json(), app_json(app_id=8, client_id="x", name="other")])
|
||
assert run_setup() == 0
|
||
assert all(r.method == "GET" for r in httpx_mock.get_requests())
|
||
env = read_env(tmp_path / ".env")
|
||
assert env["GITEA_OAUTH_CLIENT_SECRET"] == OLD_SECRET
|
||
assert env["GITEA_OAUTH_CLIENT_ID"] == "cid-7"
|
||
assert env["OAUTH_ENABLED"] == "true"
|
||
out = capsys.readouterr().out
|
||
assert "reused" in out
|
||
assert_no_secrets(out)
|
||
|
||
|
||
def test_oauth_app_redirect_differs_is_patched(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
write_existing_env(
|
||
tmp_path / ".env", GITEA_OAUTH_CLIENT_ID="cid-7", GITEA_OAUTH_CLIENT_SECRET=OLD_SECRET
|
||
)
|
||
mock_instance(httpx_mock, apps=[app_json(redirects=["https://old.example.com/oauth/callback"])])
|
||
httpx_mock.add_response(
|
||
method="PATCH",
|
||
url=f"{API}/user/applications/oauth2/7",
|
||
json=app_json(secret=CLIENT_SECRET),
|
||
)
|
||
assert run_setup() == 0
|
||
patch = next(r for r in httpx_mock.get_requests() if r.method == "PATCH")
|
||
assert json.loads(patch.content)["redirect_uris"] == [f"{PUBLIC}/oauth/callback"]
|
||
env = read_env(tmp_path / ".env")
|
||
assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET
|
||
out = capsys.readouterr().out
|
||
assert "secret regenerated" in out
|
||
assert "removed 1 other redirect URI" in out
|
||
assert_no_secrets(out)
|
||
|
||
|
||
def test_oauth_app_unknown_secret_is_rotated(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock, apps=[app_json()])
|
||
httpx_mock.add_response(
|
||
method="PATCH",
|
||
url=f"{API}/user/applications/oauth2/7",
|
||
json=app_json(secret=CLIENT_SECRET),
|
||
)
|
||
assert run_setup() == 0
|
||
env = read_env(tmp_path / ".env")
|
||
assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET
|
||
assert env["GITEA_OAUTH_CLIENT_ID"] == "cid-7"
|
||
assert "secret rotated" in capsys.readouterr().out
|
||
|
||
|
||
def test_rotation_requires_yes_when_not_interactive(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock, apps=[app_json()])
|
||
assert main(["setup", "--gitea-url", GITEA, "--public-base-url", PUBLIC]) == 1
|
||
assert "Re-run with --yes" in capsys.readouterr().err
|
||
assert not (tmp_path / ".env").exists()
|
||
|
||
|
||
def test_stale_client_id_creates_new_app_when_missing(
|
||
httpx_mock: HTTPXMock, tmp_path: Path
|
||
) -> None:
|
||
write_existing_env(
|
||
tmp_path / ".env", GITEA_OAUTH_CLIENT_ID="gone", GITEA_OAUTH_CLIENT_SECRET=OLD_SECRET
|
||
)
|
||
mock_instance(httpx_mock, apps=[])
|
||
mock_create(httpx_mock, client_id="fresh")
|
||
assert run_setup() == 0
|
||
env = read_env(tmp_path / ".env")
|
||
assert env["GITEA_OAUTH_CLIENT_ID"] == "fresh"
|
||
assert env["GITEA_OAUTH_CLIENT_SECRET"] == CLIENT_SECRET
|
||
|
||
|
||
def test_oauth_apps_are_paginated(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
|
||
mock_instance(httpx_mock)
|
||
httpx_mock.add_response(
|
||
url=f"{API}/user/applications/oauth2?page=1&limit=50",
|
||
json=[app_json(app_id=1, client_id="a", name="x")],
|
||
)
|
||
httpx_mock.add_response(
|
||
url=f"{API}/user/applications/oauth2?page=2&limit=50",
|
||
json=[app_json(app_id=2, client_id="b", name="y")],
|
||
)
|
||
httpx_mock.add_response(url=f"{API}/user/applications/oauth2?page=3&limit=50", json=[])
|
||
mock_create(httpx_mock, app_id=3, client_id="c")
|
||
assert run_setup() == 0
|
||
assert read_env(tmp_path / ".env")["GITEA_OAUTH_CLIENT_ID"] == "c"
|
||
|
||
|
||
def test_oauth_create_forbidden_explains_scope(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock, apps=[])
|
||
httpx_mock.add_response(
|
||
method="POST", url=f"{API}/user/applications/oauth2", status_code=403, json={}
|
||
)
|
||
assert run_setup() == 1
|
||
assert "write:user" in capsys.readouterr().err
|
||
|
||
|
||
def test_no_oauth_flag(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
|
||
mock_instance(httpx_mock)
|
||
assert run_setup(oauth=False) == 0
|
||
assert not any("applications" in str(r.url) for r in httpx_mock.get_requests())
|
||
env = read_env(tmp_path / ".env")
|
||
assert env["OAUTH_ENABLED"] == "false"
|
||
assert env["PUBLIC_BASE_URL"] == PUBLIC
|
||
assert env["GITEA_OAUTH_CLIENT_ID"] == ""
|
||
|
||
|
||
def test_no_public_url_skips_oauth(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock)
|
||
assert run_setup(public=False) == 0
|
||
assert read_env(tmp_path / ".env")["OAUTH_ENABLED"] == "false"
|
||
out = capsys.readouterr().out
|
||
assert "no public base URL given" in out
|
||
assert "http://127.0.0.1:8080/mcp" in out
|
||
assert "Caddyfile" not in out
|
||
|
||
|
||
# --- env file handling ----------------------------------------------------------------
|
||
|
||
|
||
def test_existing_values_preserved(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
|
||
bearer, key = "b" * 40, "k" * 40
|
||
(tmp_path / ".env").write_text(
|
||
"MAX_TOOL_RESPONSE_ITEMS=5\n"
|
||
f"MCP_BEARER_TOKEN={bearer}\n"
|
||
f"TOKEN_ENCRYPTION_KEY={key}\n"
|
||
'MY_CUSTOM_VAR="hello world"\n'
|
||
"READ_ONLY=true # keep it safe\n"
|
||
)
|
||
mock_instance(httpx_mock)
|
||
assert run_setup(oauth=False) == 0
|
||
text = (tmp_path / ".env").read_text()
|
||
env = read_env(tmp_path / ".env")
|
||
assert env["MAX_TOOL_RESPONSE_ITEMS"] == "5"
|
||
assert env["MCP_BEARER_TOKEN"] == bearer
|
||
assert env["TOKEN_ENCRYPTION_KEY"] == key
|
||
assert env["MY_CUSTOM_VAR"] == "hello world"
|
||
assert 'MY_CUSTOM_VAR="hello world"' in text
|
||
assert "READ_ONLY=true # keep it safe" in text
|
||
assert text.count("MAX_TOOL_RESPONSE_ITEMS=") == 1
|
||
|
||
|
||
def test_short_existing_secret_warns(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
write_existing_env(tmp_path / ".env", MCP_BEARER_TOKEN="short")
|
||
mock_instance(httpx_mock)
|
||
assert run_setup(oauth=False) == 0
|
||
assert read_env(tmp_path / ".env")["MCP_BEARER_TOKEN"] == "short"
|
||
assert "MCP_BEARER_TOKEN in the env file is shorter" in capsys.readouterr().out
|
||
|
||
|
||
def test_url_and_token_taken_from_env_file(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.delenv("GITEA_TOKEN")
|
||
write_existing_env(tmp_path / ".env", GITEA_URL=GITEA, GITEA_TOKEN=TOKEN)
|
||
mock_instance(httpx_mock)
|
||
assert main(["setup", "--yes", "--no-oauth"]) == 0
|
||
assert read_env(tmp_path / ".env")["GITEA_TOKEN"] == TOKEN
|
||
|
||
|
||
def test_symlink_env_file_refused(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
target = tmp_path / "real.env"
|
||
target.write_text("KEEP=1\n")
|
||
(tmp_path / ".env").symlink_to(target)
|
||
assert run_setup() == 1
|
||
assert "symlink" in capsys.readouterr().err
|
||
assert target.read_text() == "KEEP=1\n"
|
||
assert httpx_mock.get_requests() == []
|
||
|
||
|
||
def test_loose_permissions_are_tightened(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
env_path = tmp_path / ".env"
|
||
env_path.write_text("READ_ONLY=true\n")
|
||
env_path.chmod(0o644)
|
||
mock_instance(httpx_mock)
|
||
assert run_setup(oauth=False) == 0
|
||
assert stat.S_IMODE(env_path.stat().st_mode) == 0o600
|
||
assert "permissions are now 0600" in capsys.readouterr().out
|
||
|
||
|
||
def test_custom_env_file_and_template(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
|
||
template = tmp_path / "tpl"
|
||
template.write_text("# my template\nGITEA_URL=\nOAUTH_ENABLED=false\nOTHER=default\n")
|
||
mock_instance(httpx_mock)
|
||
target = tmp_path / "out.env"
|
||
assert run_setup("--env-file", str(target), "--template", str(template), oauth=False) == 0
|
||
text = target.read_text()
|
||
assert text.startswith("# my template\n")
|
||
assert "OTHER=default" in text
|
||
assert f"GITEA_URL={GITEA}" in text
|
||
|
||
|
||
def test_missing_explicit_template_fails(capsys: pytest.CaptureFixture[str]) -> None:
|
||
assert run_setup("--template", "/nonexistent/tpl") == 1
|
||
assert "cannot read template" in capsys.readouterr().err
|
||
|
||
|
||
def test_falls_back_to_embedded_template(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
|
||
assert not (tmp_path / ".env.example").exists()
|
||
mock_instance(httpx_mock)
|
||
assert run_setup(oauth=False) == 0
|
||
assert (tmp_path / ".env").read_text().startswith(ENV_TEMPLATE.splitlines()[0])
|
||
|
||
|
||
def test_uses_local_env_example_when_present(httpx_mock: HTTPXMock, tmp_path: Path) -> None:
|
||
(tmp_path / ".env.example").write_text("# local template\nGITEA_URL=\n")
|
||
mock_instance(httpx_mock)
|
||
assert run_setup(oauth=False) == 0
|
||
assert (tmp_path / ".env").read_text().startswith("# local template\n")
|
||
|
||
|
||
def test_missing_env_dir_fails(capsys: pytest.CaptureFixture[str]) -> None:
|
||
assert run_setup("--env-file", "/nonexistent-dir-xyz/.env") == 1
|
||
assert "does not exist" in capsys.readouterr().err
|
||
|
||
|
||
# --- inputs ----------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("url", "needle"),
|
||
[
|
||
("ftp://git.test", "http:// or https://"),
|
||
("https://user:hunter2@git.test", "credentials"),
|
||
("https://git.test?x=1", "query string"),
|
||
("git.test", "http:// or https://"),
|
||
],
|
||
)
|
||
def test_invalid_gitea_url(url: str, needle: str, capsys: pytest.CaptureFixture[str]) -> None:
|
||
assert main(["setup", "--yes", "--gitea-url", url]) == 1
|
||
err = capsys.readouterr().err
|
||
assert "invalid Gitea URL" in err
|
||
assert needle in err
|
||
assert "hunter2" not in err
|
||
|
||
|
||
def test_invalid_public_url(capsys: pytest.CaptureFixture[str]) -> None:
|
||
assert main(["setup", "--yes", "--gitea-url", GITEA, "--public-base-url", "http://mcp.x"]) == 1
|
||
err = capsys.readouterr().err
|
||
assert "invalid public base URL" in err
|
||
assert "https" in err
|
||
|
||
|
||
def test_gitea_url_api_suffix_normalised(httpx_mock: HTTPXMock) -> None:
|
||
mock_instance(httpx_mock)
|
||
assert main(["setup", "--yes", "--no-oauth", "--gitea-url", f"{GITEA}/api/v1/"]) == 0
|
||
|
||
|
||
def test_token_prompt_uses_getpass(
|
||
httpx_mock: HTTPXMock,
|
||
tmp_path: Path,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
capsys: pytest.CaptureFixture[str],
|
||
) -> None:
|
||
monkeypatch.delenv("GITEA_TOKEN")
|
||
prompts: list[str] = []
|
||
|
||
def fake_getpass(prompt: str = "") -> str:
|
||
prompts.append(prompt)
|
||
return f" {TOKEN} "
|
||
|
||
monkeypatch.setattr(setup_cmd.getpass, "getpass", fake_getpass)
|
||
mock_instance(httpx_mock)
|
||
assert main(["setup", "--gitea-url", GITEA, "--no-oauth"]) == 0
|
||
assert len(prompts) == 1
|
||
assert read_env(tmp_path / ".env")["GITEA_TOKEN"] == TOKEN
|
||
assert_no_secrets(capsys.readouterr().out)
|
||
|
||
|
||
def test_url_prompt(httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||
monkeypatch.setattr("builtins.input", lambda _prompt="": GITEA)
|
||
mock_instance(httpx_mock)
|
||
assert main(["setup", "--no-oauth"]) == 0
|
||
assert read_env(tmp_path / ".env")["GITEA_URL"] == GITEA
|
||
|
||
|
||
def test_missing_token_with_yes_fails(
|
||
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
monkeypatch.delenv("GITEA_TOKEN")
|
||
assert main(["setup", "--yes", "--gitea-url", GITEA]) == 1
|
||
assert "no Gitea token" in capsys.readouterr().err
|
||
|
||
|
||
def test_missing_url_with_yes_fails(capsys: pytest.CaptureFixture[str]) -> None:
|
||
assert main(["setup", "--yes"]) == 1
|
||
assert "no Gitea URL" in capsys.readouterr().err
|
||
|
||
|
||
def test_unsafe_token_rejected_without_echo(
|
||
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
monkeypatch.setenv("GITEA_TOKEN", "abc def#ghi")
|
||
assert main(["setup", "--yes", "--gitea-url", GITEA]) == 1
|
||
captured = capsys.readouterr()
|
||
assert "abc" not in captured.err + captured.out
|
||
|
||
|
||
def test_placeholder_url_in_env_file_is_ignored(
|
||
tmp_path: Path, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
write_existing_env(tmp_path / ".env", GITEA_URL="https://gitea.example.com")
|
||
assert main(["setup", "--yes"]) == 1
|
||
assert "no Gitea URL" in capsys.readouterr().err
|
||
|
||
|
||
# --- env helpers --------------------------------------------------------------------------------
|
||
|
||
|
||
def test_parse_env_text_handles_quotes_comments_and_case() -> None:
|
||
text = "# c\nA=1\nb = \"two words\"\nC='x'\nD=val # note\nexport E=5\nnot a line\n"
|
||
parsed = parse_env_text(text)
|
||
assert {k: e.value for k, e in parsed.items()} == {
|
||
"A": "1",
|
||
"B": "two words",
|
||
"C": "x",
|
||
"D": "val",
|
||
"E": "5",
|
||
}
|
||
|
||
|
||
def test_render_env_precedence_and_extras() -> None:
|
||
template = "# head\nA=tpl\nB=tpl\n# C=commented\nD=tpl\n"
|
||
existing = {
|
||
"B": EnvEntry("B", '"keep me"', "keep me"),
|
||
"X": EnvEntry("X", "9", "9"),
|
||
"A": EnvEntry("A", "old", "old"),
|
||
}
|
||
out = render_env(template, {"A": "new", "Z": "1"}, existing)
|
||
lines = out.splitlines()
|
||
assert lines[:5] == ["# head", "A=new", 'B="keep me"', "# C=commented", "D=tpl"]
|
||
assert "Z=1" in lines
|
||
assert "X=9" in lines
|
||
assert out.endswith("\n")
|
||
|
||
|
||
def test_write_env_atomic_mode_and_cleanup(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||
path = tmp_path / "x.env"
|
||
write_env_atomic(path, "A=1\n")
|
||
assert stat.S_IMODE(path.stat().st_mode) == 0o600
|
||
assert path.read_text() == "A=1\n"
|
||
|
||
def boom(*_args: Any, **_kwargs: Any) -> None:
|
||
raise OSError("disk full")
|
||
|
||
monkeypatch.setattr(os, "replace", boom)
|
||
with pytest.raises(OSError, match="disk full"):
|
||
write_env_atomic(path, "A=2\n")
|
||
assert path.read_text() == "A=1\n"
|
||
assert [p.name for p in tmp_path.iterdir() if p.name != "x.env"] == []
|
||
|
||
|
||
def test_write_env_failure_reports_rerun_hint(
|
||
httpx_mock: HTTPXMock, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
mock_instance(httpx_mock)
|
||
|
||
def boom(*_args: Any, **_kwargs: Any) -> None:
|
||
raise OSError("nope")
|
||
|
||
monkeypatch.setattr(os, "replace", boom)
|
||
assert run_setup(oauth=False) == 1
|
||
assert "run setup again" in capsys.readouterr().err
|
||
|
||
|
||
def test_http_gitea_url_warns(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]) -> None:
|
||
httpx_mock.add_response(url="http://git.test/api/v1/version", json={"version": "1"})
|
||
httpx_mock.add_response(
|
||
url="http://git.test/api/v1/user", json={"login": "root", "is_admin": True}
|
||
)
|
||
httpx_mock.add_response(url="http://git.test/api/v1/token", json={"scopes": ["all"]})
|
||
assert main(["setup", "--yes", "--no-oauth", "--gitea-url", "http://git.test"]) == 0
|
||
assert "plain http" in capsys.readouterr().out
|
||
|
||
|
||
# --- hardening ------------------------------------------------------------------------------
|
||
|
||
|
||
def test_plain_http_non_loopback_needs_consent_before_any_request(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
# No response is registered: any outgoing request would fail the test.
|
||
assert main(["setup", "--no-oauth", "--gitea-url", "http://git.test"]) == 1
|
||
captured = capsys.readouterr()
|
||
assert "plain http" in captured.out
|
||
assert "refusing to send the token over plain http" in captured.err
|
||
assert httpx_mock.get_requests() == []
|
||
assert_no_secrets(captured.out + captured.err)
|
||
|
||
|
||
def test_plain_http_interactive_decline_and_accept(
|
||
httpx_mock: HTTPXMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
monkeypatch.setattr("sys.stdin.isatty", lambda: True, raising=False)
|
||
monkeypatch.setattr("builtins.input", lambda _prompt="": "n")
|
||
assert main(["setup", "--no-oauth", "--gitea-url", "http://git.test"]) == 1
|
||
assert httpx_mock.get_requests() == []
|
||
|
||
monkeypatch.setattr("builtins.input", lambda _prompt="": "y")
|
||
httpx_mock.add_response(url="http://git.test/api/v1/version", json={"version": "1"})
|
||
httpx_mock.add_response(
|
||
url="http://git.test/api/v1/user", json={"login": "root", "is_admin": True}
|
||
)
|
||
httpx_mock.add_response(url="http://git.test/api/v1/token", json={"scopes": ["all"]})
|
||
assert main(["setup", "--no-oauth", "--gitea-url", "http://git.test"]) == 0
|
||
|
||
|
||
def test_plain_http_loopback_needs_no_consent(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
base = "http://127.0.0.1:3000/api/v1"
|
||
httpx_mock.add_response(url=f"{base}/version", json={"version": "1"})
|
||
httpx_mock.add_response(url=f"{base}/user", json={"login": "root", "is_admin": True})
|
||
httpx_mock.add_response(url=f"{base}/token", json={"scopes": ["all"]})
|
||
assert main(["setup", "--no-oauth", "--gitea-url", "http://127.0.0.1:3000"]) == 0
|
||
assert "plain http" not in capsys.readouterr().out
|
||
|
||
|
||
def test_server_supplied_version_and_login_are_sanitised(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
httpx_mock.add_response(url=f"{API}/version", json={"version": "1.99\x1b[31mRED\x07\n\x9bX"})
|
||
httpx_mock.add_response(url=f"{API}/user", json={"login": "ro\x1b[2Jot", "is_admin": True})
|
||
httpx_mock.add_response(url=f"{API}/token", json={"scopes": ["all"]})
|
||
assert run_setup(oauth=False) == 0
|
||
out = capsys.readouterr().out
|
||
for bad in ("\x1b", "\x07", "", "\x9b"):
|
||
assert bad not in out
|
||
assert "1.99" in out
|
||
|
||
|
||
def test_version_made_only_of_control_characters_is_rejected(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
httpx_mock.add_response(url=f"{API}/version", json={"version": "\x1b\x07"})
|
||
assert run_setup(oauth=False) == 1
|
||
assert "printable" in capsys.readouterr().err
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"token",
|
||
["tokén_0123456789abcdef", "tok\x1b[0m_0123456789", "tok_0123456789abcdef"],
|
||
)
|
||
def test_non_ascii_or_control_tokens_are_rejected_without_echo(
|
||
token: str, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
monkeypatch.setenv("GITEA_TOKEN", token)
|
||
assert run_setup(oauth=False) == 1
|
||
captured = capsys.readouterr()
|
||
assert "printable ASCII" in captured.err
|
||
assert token not in captured.out + captured.err
|
||
assert token.split("_")[0] not in captured.out + captured.err
|
||
|
||
|
||
def test_unsafe_urls_are_rejected_before_any_request(
|
||
httpx_mock: HTTPXMock, capsys: pytest.CaptureFixture[str]
|
||
) -> None:
|
||
assert main(["setup", "--yes", "--no-oauth", "--gitea-url", "https://git.test/a'b"]) == 1
|
||
assert "the Gitea URL" in capsys.readouterr().err
|
||
assert (
|
||
main(["setup", "--yes", "--gitea-url", GITEA, "--public-base-url", "https://mcp.x/a$b"])
|
||
== 1
|
||
)
|
||
assert "the public base URL" in capsys.readouterr().err
|
||
assert httpx_mock.get_requests() == []
|
||
|
||
|
||
def test_env_file_directory_is_fsynced_after_replace(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
calls: list[tuple[str, bool]] = []
|
||
real_fsync = os.fsync
|
||
|
||
def recording_fsync(fd: int) -> None:
|
||
calls.append(("fsync", stat.S_ISDIR(os.fstat(fd).st_mode)))
|
||
real_fsync(fd)
|
||
|
||
real_replace = os.replace
|
||
|
||
def recording_replace(src: Any, dst: Any) -> None:
|
||
calls.append(("replace", False))
|
||
real_replace(src, dst)
|
||
|
||
monkeypatch.setattr(os, "fsync", recording_fsync)
|
||
monkeypatch.setattr(os, "replace", recording_replace)
|
||
write_env_atomic(tmp_path / "x.env", "A=1\n")
|
||
|
||
assert calls == [("fsync", False), ("replace", False), ("fsync", True)]
|
||
assert (tmp_path / "x.env").read_text() == "A=1\n"
|
||
|
||
|
||
def test_directory_fsync_failure_is_not_fatal(
|
||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
real_open = os.open
|
||
|
||
def failing_open(path: Any, flags: int, *args: Any, **kwargs: Any) -> int:
|
||
if Path(path) == tmp_path:
|
||
raise OSError("no directory fds here")
|
||
return real_open(path, flags, *args, **kwargs)
|
||
|
||
monkeypatch.setattr(os, "open", failing_open)
|
||
write_env_atomic(tmp_path / "x.env", "A=1\n")
|
||
assert (tmp_path / "x.env").read_text() == "A=1\n"
|