Dev #72
No files matched your search
@@ -36,7 +36,7 @@ when OAuth is on.
|
||||
|
||||
| Variable | Default | Rules |
|
||||
|---|---|---|
|
||||
| `GITEA_URL` | required | `http` or `https`, must have a host, no credentials, query or fragment. A trailing `/api/v1` is removed. |
|
||||
| `GITEA_URL` | required | `http` or `https`, must have a host, no credentials, query or fragment, and no whitespace, backslash or control characters. A trailing `/api/v1` is removed and the URL is rebuilt from its parsed parts. Plain `http` to a host that is not loopback starts, but `serve` prints a warning on stderr because every token then travels unencrypted. |
|
||||
| `GITEA_TOKEN` | unset | Token for stdio and for admin-token mode on HTTP. Its scopes bound what the server can do. Required for stdio (exit code 2 without it). |
|
||||
|
||||
## Transport
|
||||
|
||||
+4
-1
@@ -177,7 +177,10 @@ confirmation, `READ_ONLY`, `TOOLSETS`, `DISABLED_TOOLS` and narrow token scopes
|
||||
MCP path and the unauthenticated OAuth paths (`/register`, `/authorize`, `/token`, `/revoke`,
|
||||
consent, callback), before authentication. Client IPs come from `X-Forwarded-For` only for
|
||||
`FORWARDED_ALLOW_IPS`; a forwarded header from any other peer is ignored and logged once.
|
||||
- `PUBLIC_BASE_URL` must be https unless loopback. Gitea redirects are never followed.
|
||||
- `PUBLIC_BASE_URL` must be https unless loopback. `GITEA_URL` and `PUBLIC_BASE_URL` reject
|
||||
whitespace, backslashes and control characters and are rebuilt from their parsed parts;
|
||||
plain http `GITEA_URL` to a non-loopback host prints a startup warning. Gitea redirects are
|
||||
never followed.
|
||||
- stdio: stdout carries JSON-RPC only; all logging goes to stderr.
|
||||
|
||||
### Per-user OAuth design
|
||||
|
||||
@@ -109,6 +109,13 @@ def run_serve(args: argparse.Namespace) -> int:
|
||||
if settings is None:
|
||||
return 2
|
||||
configure_logging(settings.log_level)
|
||||
if settings.insecure_gitea_url():
|
||||
print(
|
||||
"aegis-gitea-mcp: warning: GITEA_URL uses plain http to a host that is not loopback: "
|
||||
"the Gitea token, the OAuth client secret and every user's token travel unencrypted. "
|
||||
"Use https",
|
||||
file=sys.stderr,
|
||||
)
|
||||
if settings.mcp_transport == "http":
|
||||
return transports.run_http(settings)
|
||||
return transports.run_stdio(settings)
|
||||
|
||||
@@ -15,7 +15,7 @@ import sys
|
||||
from collections.abc import Mapping
|
||||
from pathlib import Path
|
||||
from typing import Annotated, Any, Literal
|
||||
from urllib.parse import urlsplit
|
||||
from urllib.parse import urlsplit, urlunsplit
|
||||
|
||||
from pydantic import Field, SecretStr, field_validator
|
||||
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
|
||||
@@ -147,8 +147,16 @@ def redirect_matches_allowlist(uri: str, allowlist: list[str]) -> bool:
|
||||
|
||||
|
||||
def _normalize_http_url(value: str, *, field: str) -> str:
|
||||
"""Validate an http(s) URL and return it without query, fragment or trailing slash."""
|
||||
"""Validate an http(s) URL and return it rebuilt without query, fragment or trailing slash.
|
||||
|
||||
The result is assembled from the parsed parts, so the string that is validated
|
||||
is the string that is used. ``urlsplit`` silently drops tabs and newlines
|
||||
before parsing, so any whitespace, backslash or non-printable character
|
||||
(control and bidi characters included) is rejected up front.
|
||||
"""
|
||||
candidate = value.strip()
|
||||
if any(ch.isspace() or not ch.isprintable() or ch == "\\" for ch in candidate):
|
||||
raise ValueError(f"{field} must not contain whitespace, backslashes or control characters")
|
||||
parts = urlsplit(candidate)
|
||||
if parts.scheme not in {"http", "https"}:
|
||||
raise ValueError(f"{field} must start with http:// or https://")
|
||||
@@ -162,7 +170,7 @@ def _normalize_http_url(value: str, *, field: str) -> str:
|
||||
parts.port # noqa: B018 (accessing .port validates it)
|
||||
except ValueError as exc:
|
||||
raise ValueError(f"{field} has an invalid port") from exc
|
||||
return candidate.rstrip("/")
|
||||
return urlunsplit((parts.scheme, parts.netloc, parts.path.rstrip("/"), "", ""))
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
@@ -403,6 +411,11 @@ class Settings(BaseSettings):
|
||||
"""Audit log path, defaulting to ``DATA_DIR/audit.log``."""
|
||||
return self.audit_log_path or (self.data_dir / "audit.log")
|
||||
|
||||
def insecure_gitea_url(self) -> bool:
|
||||
"""True when ``GITEA_URL`` is plain http to a host that is not loopback."""
|
||||
parts = urlsplit(self.gitea_url)
|
||||
return parts.scheme == "http" and not is_loopback_host(parts.hostname or "")
|
||||
|
||||
def is_loopback_bind(self) -> bool:
|
||||
"""Return True when ``MCP_HOST`` is a loopback address (127.0.0.0/8, ::1, localhost)."""
|
||||
return is_loopback_host(self.mcp_host)
|
||||
|
||||
@@ -0,0 +1,256 @@
|
||||
"""``GITEA_URL``, ``PUBLIC_BASE_URL`` and the other path-like settings are validated strictly (SEC-10)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from aegis_gitea_mcp import transports
|
||||
from aegis_gitea_mcp.cli import main
|
||||
from aegis_gitea_mcp.config import Settings
|
||||
from aegis_gitea_mcp.oauth.provider import redirect_uri_problem
|
||||
|
||||
|
||||
def make(**values: Any) -> Settings:
|
||||
base: dict[str, Any] = {"gitea_url": "https://gitea.example.com"}
|
||||
base.update(values)
|
||||
return Settings(**base)
|
||||
|
||||
|
||||
BAD_CHARACTERS = [
|
||||
("tab", "\t"),
|
||||
("newline", "\n"),
|
||||
("carriage return", "\r"),
|
||||
("space", " "),
|
||||
("nul", "\x00"),
|
||||
("vertical tab", "\x0b"),
|
||||
("line separator", chr(0x2028)),
|
||||
("no-break space", chr(0xA0)),
|
||||
("bidi override", chr(0x202E)),
|
||||
("bidi isolate", chr(0x2066)),
|
||||
("zero width space", chr(0x200B)),
|
||||
("backslash", "\\"),
|
||||
]
|
||||
POSITIONS = [
|
||||
"https://gi{c}tea.example.com",
|
||||
"https://gitea.example.com/a{c}b",
|
||||
"https://gitea.example.com:44{c}3",
|
||||
"https://gitea.example.com/{c}x",
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("template", POSITIONS)
|
||||
@pytest.mark.parametrize(("name", "char"), BAD_CHARACTERS, ids=[n for n, _ in BAD_CHARACTERS])
|
||||
def test_gitea_url_rejects_whitespace_and_control_characters(
|
||||
template: str, name: str, char: str
|
||||
) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(gitea_url=template.format(c=char))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("template", POSITIONS)
|
||||
@pytest.mark.parametrize(("name", "char"), BAD_CHARACTERS, ids=[n for n, _ in BAD_CHARACTERS])
|
||||
def test_public_base_url_rejects_whitespace_and_control_characters(
|
||||
template: str, name: str, char: str
|
||||
) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(public_base_url=template.format(c=char).replace("gitea", "mcp"))
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"ftp://gitea.example.com",
|
||||
"file:///etc/passwd",
|
||||
"javascript:alert(1)",
|
||||
"gitea.example.com",
|
||||
"//gitea.example.com",
|
||||
"https://",
|
||||
"https:///path",
|
||||
"https://user:pass@gitea.example.com",
|
||||
"https://user@gitea.example.com",
|
||||
"https://gitea.example.com?x=1",
|
||||
"https://gitea.example.com/?",
|
||||
"https://gitea.example.com#frag",
|
||||
"https://gitea.example.com:99999",
|
||||
"https://gitea.example.com:abc",
|
||||
"",
|
||||
],
|
||||
)
|
||||
def test_gitea_url_rejects_bad_schemes_credentials_query_fragment_port_and_missing_host(
|
||||
url: str,
|
||||
) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(gitea_url=url)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"http://mcp.example.com",
|
||||
"http://203.0.113.5",
|
||||
"ftp://mcp.example.com",
|
||||
"https://user:pass@mcp.example.com",
|
||||
"https://mcp.example.com?x=1",
|
||||
"https://mcp.example.com#f",
|
||||
"https://mcp.example.com:99999",
|
||||
],
|
||||
)
|
||||
def test_public_base_url_rejects_http_on_public_hosts_and_the_shared_url_rules(url: str) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(public_base_url=url)
|
||||
|
||||
|
||||
def test_http_public_base_url_is_only_allowed_on_loopback() -> None:
|
||||
assert make(public_base_url="http://localhost:8080").public_base_url == "http://localhost:8080"
|
||||
assert make(public_base_url="http://127.0.0.1:8080").public_base_url == "http://127.0.0.1:8080"
|
||||
assert make(public_base_url="http://[::1]:8080").public_base_url == "http://[::1]:8080"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("given", "used"),
|
||||
[
|
||||
("https://gitea.example.com/", "https://gitea.example.com"),
|
||||
("https://gitea.example.com/api/v1", "https://gitea.example.com"),
|
||||
("https://gitea.example.com/api/v1/", "https://gitea.example.com"),
|
||||
("https://gitea.example.com/git/api/v1", "https://gitea.example.com/git"),
|
||||
(" https://gitea.example.com ", "https://gitea.example.com"),
|
||||
("HTTPS://gitea.example.com:3000/sub/", "https://gitea.example.com:3000/sub"),
|
||||
("http://127.0.0.1:3000", "http://127.0.0.1:3000"),
|
||||
],
|
||||
)
|
||||
def test_the_validated_url_is_the_url_that_is_used(given: str, used: str) -> None:
|
||||
settings = make(gitea_url=given)
|
||||
|
||||
assert settings.gitea_url == used
|
||||
assert settings.gitea_api_url == f"{used}/api/v1"
|
||||
|
||||
|
||||
def test_serve_warns_about_plain_http_gitea_on_a_non_loopback_host(
|
||||
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
from aegis_gitea_mcp import cli
|
||||
|
||||
monkeypatch.setenv("GITEA_URL", "http://gitea.internal")
|
||||
monkeypatch.setenv("GITEA_TOKEN", "t" * 20)
|
||||
monkeypatch.setattr(cli, "configure_logging", lambda level: None)
|
||||
monkeypatch.setattr(transports, "run_stdio", lambda settings: 0)
|
||||
|
||||
assert main(["serve"]) == 0
|
||||
|
||||
err = capsys.readouterr().err
|
||||
assert "plain http" in err and "unencrypted" in err
|
||||
assert "t" * 20 not in err
|
||||
|
||||
|
||||
@pytest.mark.parametrize("url", ["https://gitea.example.com", "http://localhost:3000"])
|
||||
def test_serve_does_not_warn_for_https_or_loopback(
|
||||
monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], url: str
|
||||
) -> None:
|
||||
from aegis_gitea_mcp import cli
|
||||
|
||||
monkeypatch.setenv("GITEA_URL", url)
|
||||
monkeypatch.setenv("GITEA_TOKEN", "t" * 20)
|
||||
monkeypatch.setattr(cli, "configure_logging", lambda level: None)
|
||||
monkeypatch.setattr(transports, "run_stdio", lambda settings: 0)
|
||||
|
||||
assert main(["serve"]) == 0
|
||||
|
||||
assert "plain http" not in capsys.readouterr().err
|
||||
|
||||
|
||||
# Other path-like and list settings
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"path",
|
||||
["mcp", "//mcp", "/", "//", "/mcp?x=1", "/mcp#f", "/m cp", ""],
|
||||
)
|
||||
def test_mcp_path_rejections(path: str) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(mcp_path=path)
|
||||
|
||||
|
||||
def test_mcp_path_trailing_slash_is_normalized() -> None:
|
||||
assert make(mcp_path="/mcp/").mcp_path == "/mcp"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"value",
|
||||
["not-an-ip", "10.0.0.0/33", "*,10.0.0.1", "10.0.0.1,*", "10.0.0.1;10.0.0.2", "10.0.0.256"],
|
||||
)
|
||||
def test_forwarded_allow_ips_rejections(value: str) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(forwarded_allow_ips=value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"scopes",
|
||||
[
|
||||
["all"],
|
||||
["read:unknown"],
|
||||
["repo"],
|
||||
["write:repository", "sudo"],
|
||||
["admin"],
|
||||
[""],
|
||||
[],
|
||||
],
|
||||
)
|
||||
def test_oauth_scopes_reject_unknown_scopes_because_gitea_would_grant_everything(
|
||||
scopes: list[str],
|
||||
) -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(oauth_scopes=scopes)
|
||||
|
||||
|
||||
def test_oauth_scopes_accept_granular_scopes_only() -> None:
|
||||
assert make(oauth_scopes=["read:repository", "write:issue"]).oauth_scopes == [
|
||||
"read:repository",
|
||||
"write:issue",
|
||||
]
|
||||
|
||||
|
||||
def test_secret_length_minimums() -> None:
|
||||
with pytest.raises(ValidationError):
|
||||
make(mcp_bearer_token="short")
|
||||
with pytest.raises(ValidationError):
|
||||
make(token_encryption_key="x" * 31)
|
||||
|
||||
|
||||
# Redirect URIs at client registration
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"uri",
|
||||
[
|
||||
"http://example.com/cb",
|
||||
"http://localhost.evil.example/cb",
|
||||
"https://example.com/cb#frag",
|
||||
"https://user:pass@example.com/cb",
|
||||
"javascript:alert(1)",
|
||||
"data:text/html,hi",
|
||||
"file:///etc/passwd",
|
||||
"myapp://callback",
|
||||
"claude://callback",
|
||||
"https:///cb",
|
||||
"https://example.com:99999/cb",
|
||||
"",
|
||||
],
|
||||
)
|
||||
def test_redirect_uri_rules(uri: str) -> None:
|
||||
assert redirect_uri_problem(uri) is not None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"uri",
|
||||
[
|
||||
"https://claude.ai/api/mcp/auth_callback",
|
||||
"http://localhost:3000/callback",
|
||||
"http://127.0.0.1/cb",
|
||||
"http://[::1]:5000/cb",
|
||||
],
|
||||
)
|
||||
def test_acceptable_redirect_uris(uri: str) -> None:
|
||||
assert redirect_uri_problem(uri) is None
|
||||
Reference in new issue
Block a user