From 517c6750486eb115e571420e3667df054ae8fbfa Mon Sep 17 00:00:00 2001 From: latte Date: Wed, 14 Oct 2026 16:11:01 +0200 Subject: [PATCH] fix: harden setup against plain http, untrusted output and unsafe values 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. --- src/aegis_gitea_mcp/setup_cmd.py | 78 +++++++++++++++--- tests/test_setup.py | 131 +++++++++++++++++++++++++++++++ 2 files changed, 198 insertions(+), 11 deletions(-) diff --git a/src/aegis_gitea_mcp/setup_cmd.py b/src/aegis_gitea_mcp/setup_cmd.py index 5ca60eb..21e3301 100644 --- a/src/aegis_gitea_mcp/setup_cmd.py +++ b/src/aegis_gitea_mcp/setup_cmd.py @@ -18,6 +18,7 @@ import ssl import stat import sys import tempfile +import unicodedata from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -45,7 +46,23 @@ PLACEHOLDER_GITEA_HOSTS = frozenset({"gitea.example.com"}) REQUIRED_SCOPES = tuple(f"write:{category}" for category in GITEA_SCOPE_CATEGORIES) _ASSIGNMENT = re.compile(r"^(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)\s*=(.*)$") -_SAFE_VALUE = re.compile(r"^[^\s\"'#$\\]+$") +#: Printable ASCII without whitespace and without the characters a dotenv parser or +#: a shell would treat specially. Anything else is refused instead of being quoted. +_SAFE_VALUE = re.compile(r"^(?!.*[\"'#$\\])[!-~]+$") + + +def printable(text: str, limit: int = 200) -> str: + """Return ``text`` made safe to print: control and format characters become spaces. + + Text that comes from the Gitea server is untrusted. Control characters (including + escape sequences' ESC byte, C1 controls, bidi overrides and line separators) must not + reach the operator's terminal. + """ + cleaned = "".join( + " " if unicodedata.category(char)[0] == "C" or char in "\u2028\u2029" else char + for char in text + ) + return cleaned.strip()[:limit] class SetupError(Exception): @@ -162,6 +179,21 @@ def write_env_atomic(path: Path, content: str) -> None: except BaseException: tmp.unlink(missing_ok=True) raise + _fsync_directory(path.parent) + + +def _fsync_directory(directory: Path) -> None: + """Flush the directory entry of a just-renamed file (best effort where unsupported).""" + try: + fd = os.open(directory, os.O_RDONLY) + except OSError: + return + try: + os.fsync(fd) + except OSError: + pass + finally: + os.close(fd) # --------------------------------------------------------------------------- @@ -236,7 +268,7 @@ class GiteaApi: message = body["message"] except ValueError: message = "" - message = re.sub(r"[\x00-\x1f\x7f]", " ", self.redact(message)).strip()[:200] + message = printable(self.redact(message)) return f"HTTP {response.status_code}" + (f": {message}" if message else "") def json_object(self, response: httpx.Response, what: str) -> dict[str, Any]: @@ -311,9 +343,10 @@ def _say(message: str = "") -> None: def _validate_gitea_url(value: str) -> str: try: - return Settings._validate_gitea_url(value) + validated = Settings._validate_gitea_url(value) except ValueError as exc: raise SetupError(f"invalid Gitea URL: {exc}") from None + return _check_value_text("the Gitea URL", validated) def _validate_public_url(value: str) -> str: @@ -322,7 +355,7 @@ def _validate_public_url(value: str) -> str: except ValueError as exc: raise SetupError(f"invalid public base URL: {exc}") from None assert validated is not None # noqa: S101 (value is a non-empty string here) - return validated + return _check_value_text("the public base URL", validated) def _validate_token_text(token: str) -> str: @@ -331,15 +364,18 @@ def _validate_token_text(token: str) -> str: raise SetupError("no Gitea token given") if not _SAFE_VALUE.match(token): raise SetupError( - "the Gitea token contains whitespace or one of the characters " - "\" ' # $ \\ that cannot be stored safely in an env file" + "the Gitea token must be printable ASCII without whitespace and without any of " + "the characters \" ' # $ \\ (it cannot be stored safely in an env file)" ) return token def _check_value_text(name: str, value: str) -> str: if not _SAFE_VALUE.match(value): - raise SetupError(f"{name} contains characters that cannot be stored safely in an env file") + raise SetupError( + f"{name} contains characters that cannot be stored safely in an env file " + "(only printable ASCII without whitespace or any of \" ' # $ \\ is allowed)" + ) return value @@ -390,7 +426,9 @@ def verify_instance(api: GiteaApi, report: Report) -> None: version = api.json_object(response, "GET /api/v1/version").get("version") if not isinstance(version, str) or not version: raise SetupError("GET /api/v1/version did not report a Gitea version") - report.version = version[:64] + report.version = printable(version, 64) + if not report.version: + raise SetupError("GET /api/v1/version did not report a printable Gitea version") _say(f"Gitea reachable: version {report.version}") response = api.request("GET", "/user") @@ -403,7 +441,7 @@ def verify_instance(api: GiteaApi, report: Report) -> None: raise SetupError(f"GET /api/v1/user failed ({api.detail(response)})") user = api.json_object(response, "GET /api/v1/user") login = user.get("login") - report.login = login if isinstance(login, str) else "?" + report.login = (printable(login, 64) or "?") if isinstance(login, str) else "?" if user.get("is_admin") is not True: raise SetupError( f"the token belongs to '{report.login}', who is not a site administrator. " @@ -767,10 +805,9 @@ def _run(args: argparse.Namespace) -> int: report = Report() report.gitea_url = _resolve_gitea_url(args, existing) + _confirm_plain_http(report.gitea_url, yes=args.yes) token = _resolve_token(args, existing) public_url = _resolve_public_url(args, existing) - if urlsplit(report.gitea_url).scheme == "http" and not _is_loopback(report.gitea_url): - report.warnings.append("the Gitea URL uses plain http, so the token travels unencrypted.") oauth_wanted = not args.no_oauth and public_url is not None _say(f"Using {report.gitea_url}") @@ -837,6 +874,25 @@ def _run(args: argparse.Namespace) -> int: return 0 +def _confirm_plain_http(url: str, *, yes: bool) -> None: + """Warn about a plain http Gitea URL on a non-loopback host and require consent. + + Runs before the first request, so the token never travels unencrypted unless the + operator agreed with ``--yes`` or an interactive answer. + """ + if urlsplit(url).scheme != "http" or _is_loopback(url): + return + _say( + f"Warning: {url} uses plain http. The Gitea admin token and the OAuth client secret " + "would travel unencrypted over the network." + ) + if not _confirm("Send the token over plain http anyway?", yes=yes): + raise SetupError( + "refusing to send the token over plain http to a non-loopback host. " + "Use an https URL, or re-run with --yes to accept the risk." + ) + + def _is_loopback(url: str) -> bool: return is_loopback_host(urlsplit(url).hostname or "") diff --git a/tests/test_setup.py b/tests/test_setup.py index 9104f0e..f76b2bc 100644 --- a/tests/test_setup.py +++ b/tests/test_setup.py @@ -711,3 +711,134 @@ def test_http_gitea_url_warns(httpx_mock: HTTPXMock, capsys: pytest.CaptureFixtu 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"