From 8be859243df1026d2efc03dfb7ea6a64fae30a40 Mon Sep 17 00:00:00 2001 From: latte Date: Wed, 14 Oct 2026 17:24:41 +0200 Subject: [PATCH] fix: add OAUTH_REDIRECT_ALLOWLIST for client registration and waive the open default (SEC-05) --- .env.example | 5 + docs/configuration.md | 2 + docs/security.md | 9 +- src/aegis_gitea_mcp/config.py | 80 +++++++++++- src/aegis_gitea_mcp/oauth/provider.py | 26 ++-- src/aegis_gitea_mcp/setup_template.py | 5 + tests/security/test_oauth_redirects.py | 167 +++++++++++++++++++++++++ 7 files changed, 283 insertions(+), 11 deletions(-) create mode 100644 tests/security/test_oauth_redirects.py diff --git a/.env.example b/.env.example index 0ebe6bc..9a92cd7 100644 --- a/.env.example +++ b/.env.example @@ -34,6 +34,11 @@ GITEA_OAUTH_CLIENT_SECRET= # Encrypts stored user tokens. At least 32 characters. Generate with: # openssl rand -hex 32 TOKEN_ENCRYPTION_KEY= +# Optional allowlist of redirect URIs clients may register (comma separated): exact +# URIs, or https://host/ prefixes. Empty accepts any https URI (and http on +# loopback); the consent page is then the only check. For a public deployment list the +# hosted clients you use, e.g. https://claude.ai/,https://chat.mistral.ai/ +OAUTH_REDIRECT_ALLOWLIST= # Granular Gitea scopes only (read:/write: activitypub, admin, issue, misc, # notification, organization, package, repository, user). Unknown values are rejected. OAUTH_SCOPES=write:activitypub,write:admin,write:issue,write:misc,write:notification,write:organization,write:package,write:repository,write:user diff --git a/docs/configuration.md b/docs/configuration.md index 94cc6eb..1750efd 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -62,6 +62,8 @@ when OAuth is on. | `TOKEN_ENCRYPTION_KEY` | unset | At least 32 characters. Encrypts stored user tokens. Losing it invalidates every stored login. | | `OAUTH_SCOPES` | all `write:` scopes | Granular Gitea scopes only (`read:` or `write:` plus activitypub, admin, issue, misc, notification, organization, package, repository, user). Anything else is rejected, because Gitea grants full access for unknown scope strings. Must not be empty. | +| `OAUTH_REDIRECT_ALLOWLIST` | empty | Comma list of redirect URIs MCP clients may register: exact URIs (`https://client.example/cb`) or prefixes ending in `/` that allow a whole host (`https://claude.ai/`). A bare origin without a path is refused. Empty accepts any https URI (and http on `localhost`, `127.0.0.1`, `[::1]`). With a list, https URIs must match it; loopback URIs stay allowed for native clients. Checked at registration and again at `/authorize`. | + `PUBLIC_BASE_URL` is also required for OAuth. The Gitea OAuth application's redirect URI is `/oauth/callback`. diff --git a/docs/security.md b/docs/security.md index 8869b54..5199634 100644 --- a/docs/security.md +++ b/docs/security.md @@ -60,7 +60,7 @@ Spoofing | Calling the endpoint without credentials | Non-loopback bind refuses to start without a bearer or OAuth; SDK `RequireAuthMiddleware` answers 401 | `transports.py` (`http_startup_errors`, `build_http_app`) | | Guessing the bearer | 32 character minimum; constant-time comparison of SHA-256 digests; per-IP and per-token rate limit before auth | `config.py`, `auth.py` (`StaticBearerVerifier`), `rate_limit.py` | | Using a token minted for another resource | RFC 8707 resource binding and required scope `gitea` enforced by the SDK; `/authorize` with a foreign `resource` gets `invalid_target` | `transports.py` (`_auth_settings`), `oauth/provider.py` | -| Confused deputy through the shared Gitea OAuth app | Own consent page naming client and redirect URI, CSRF double submit with `SameSite=Strict` cookie, browser-binding cookie on the callback | `oauth/callback.py`, `oauth/provider.py` (residual: SEC-05) | +| Confused deputy through the shared Gitea OAuth app | Own consent page naming client and redirect URI, CSRF double submit with `SameSite=Strict` cookie, browser-binding cookie on the callback | `oauth/callback.py`, `oauth/provider.py` (residual: W-10, optional `OAUTH_REDIRECT_ALLOWLIST`) | | Code or refresh token replay | Single-use codes; replay or reuse of a rotated refresh token revokes the grant | `oauth/provider.py` | Tampering @@ -127,8 +127,10 @@ Elevation of privilege Gitea token committed to a file) is returned unless it matches a pattern. - SSRF through Gitea features (migrations, webhooks, mirrors) is Gitea's responsibility; this server only ever contacts `GITEA_URL`. -- The consent page is the only barrier against phishing with a self-registered client - (SEC-05). +- With the default `OAUTH_REDIRECT_ALLOWLIST` (empty), the consent page is the only barrier + against phishing with a self-registered client (waiver W-10). For a public deployment set the + allowlist to the hosted clients you use, for example + `OAUTH_REDIRECT_ALLOWLIST=https://claude.ai/,https://chat.mistral.ai/`. - Key rotation for `TOKEN_ENCRYPTION_KEY` is not supported: a new key means moving the store away and signing every user out. - The audit chain is tamper-evident, not tamper-proof: anyone with write access to the file can @@ -233,3 +235,4 @@ changeset. | W-07 | Credential-issuing operations bypass pattern masking | `gitea/executor.py` (`CREDENTIAL_ISSUING_OPERATIONS`) | Their purpose is to return a new credential; exact redaction of the caller's own tokens still applies. | | W-08 | Unauthenticated `/health` and `/` report version and tool count, `/health` reports Gitea reachability | `transports.py` | Needed by container health checks and clients; reveals no configuration or secret. The Gitea probe is cached for 10 seconds. | | W-09 | Pattern masking keeps the first and last 4 characters of a matched secret | `security.py` (`mask_secret`) | Applies only to pattern matches, which are at least 16 characters; the remainder keeps enough entropy, and the context helps users recognise the value. | +| W-10 | Any https redirect URI can be registered when `OAUTH_REDIRECT_ALLOWLIST` is empty (SEC-05) | `oauth/provider.py` (`redirect_uri_problem`), `config.py` | Owner decision: the default stays open so any MCP client can register without configuration. The mitigation is the consent page (client name, redirect host and URI, explicit approval, CSRF and browser binding) and ADR 0004. The optional allowlist closes the gap: exact URIs or `https://host/` prefixes, checked at registration and at `/authorize`; recommended for every public deployment. | diff --git a/src/aegis_gitea_mcp/config.py b/src/aegis_gitea_mcp/config.py index 40b3546..7992f31 100644 --- a/src/aegis_gitea_mcp/config.py +++ b/src/aegis_gitea_mcp/config.py @@ -88,6 +88,64 @@ def _split_list(value: Any) -> Any: return value +def redirect_allowlist_entry_problem(entry: str) -> str | None: + """Return why ``entry`` cannot be an ``OAUTH_REDIRECT_ALLOWLIST`` item, or None when it can. + + An item is an exact redirect URI (``https://client.example/cb``) or a prefix + that ends in ``/`` (``https://claude.ai/``, which allows every path on that + host). A bare origin without a path is refused as ambiguous. + """ + if any(ch.isspace() or not ch.isprintable() for ch in entry): + return "whitespace or control characters" + try: + parts = urlsplit(entry) + parts.port # noqa: B018 (accessing .port validates it) + except ValueError: + return "not a valid URL" + host = parts.hostname or "" + if not host: + return "no host" + if parts.username is not None or parts.password is not None: + return "credentials in the URL" + if parts.query or parts.fragment or "?" in entry or "#" in entry: + return "query string or fragment" + if parts.scheme == "http" and not is_loopback_host(host): + return "http is only allowed for loopback hosts" + if parts.scheme not in {"http", "https"}: + return "scheme must be https (or http for loopback)" + if not parts.path: + return f"no path: write {entry}/ to allow the whole host or give the full callback URL" + return None + + +def redirect_matches_allowlist(uri: str, allowlist: list[str]) -> bool: + """True when ``uri`` equals an allowlist item or lies under one of its ``/`` prefixes. + + An empty allowlist allows every acceptable redirect URI. The comparison of a + prefix is by scheme, host (case-insensitive) and path prefix, so + ``https://claude.ai/`` never matches ``https://claude.ai.evil.example/``. + """ + if not allowlist: + return True + try: + target = urlsplit(uri) + except ValueError: + return False + for entry in allowlist: + if uri == entry: + return True + if not entry.endswith("/"): + continue + prefix = urlsplit(entry) + if ( + target.scheme == prefix.scheme + and (target.netloc.lower() == prefix.netloc.lower()) + and target.path.startswith(prefix.path) + ): + return True + return False + + def _normalize_http_url(value: str, *, field: str) -> str: """Validate an http(s) URL and return it without query, fragment or trailing slash.""" candidate = value.strip() @@ -142,6 +200,7 @@ class Settings(BaseSettings): oauth_scopes: Annotated[list[str], NoDecode] = Field( default_factory=lambda: DEFAULT_OAUTH_SCOPES.split(",") ) + oauth_redirect_allowlist: Annotated[list[str], NoDecode] = Field(default_factory=list) # State data_dir: Path = Field(default_factory=default_data_dir) @@ -250,7 +309,14 @@ class Settings(BaseSettings): ) return value - @field_validator("allowed_hosts", "oauth_scopes", "toolsets", "disabled_tools", mode="before") + @field_validator( + "allowed_hosts", + "oauth_scopes", + "oauth_redirect_allowlist", + "toolsets", + "disabled_tools", + mode="before", + ) @classmethod def _parse_list(cls, value: Any) -> Any: """Accept comma or whitespace separated strings for list settings.""" @@ -271,6 +337,18 @@ class Settings(BaseSettings): raise ValueError("OAUTH_SCOPES must not be empty") return list(dict.fromkeys(value)) + @field_validator("oauth_redirect_allowlist") + @classmethod + def _validate_redirect_allowlist(cls, value: list[str]) -> list[str]: + """Accept exact redirect URIs and ``https://host/`` prefixes, nothing looser.""" + checked: list[str] = [] + for entry in value: + problem = redirect_allowlist_entry_problem(entry) + if problem is not None: + raise ValueError(f"OAUTH_REDIRECT_ALLOWLIST entry is not acceptable: {problem}") + checked.append(entry) + return list(dict.fromkeys(checked)) + @field_validator("toolsets") @classmethod def _validate_toolsets(cls, value: list[str]) -> list[str]: diff --git a/src/aegis_gitea_mcp/oauth/provider.py b/src/aegis_gitea_mcp/oauth/provider.py index cfce6bf..e954695 100644 --- a/src/aegis_gitea_mcp/oauth/provider.py +++ b/src/aegis_gitea_mcp/oauth/provider.py @@ -59,7 +59,7 @@ from pydantic import AnyHttpUrl, ValidationError from aegis_gitea_mcp.audit import AuditLogger, get_audit_logger from aegis_gitea_mcp.auth import MCP_SCOPE -from aegis_gitea_mcp.config import Settings +from aegis_gitea_mcp.config import Settings, redirect_matches_allowlist from aegis_gitea_mcp.oauth.store import ( ClientRecord, OAuthStore, @@ -87,6 +87,7 @@ CONSENT_PATH = "/oauth/consent" _LOOPBACK_REDIRECT_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) _CLIENT_AUTH_METHODS = frozenset({"none", "client_secret_post", "client_secret_basic"}) +_NOT_ALLOWED = "redirect URI is not on this server's allowlist (OAUTH_REDIRECT_ALLOWLIST)" _REDIRECT_RULE = ( "redirect URIs must use https://, or http:// on localhost, 127.0.0.1 or [::1], " "without credentials or a fragment" @@ -96,8 +97,12 @@ V = TypeVar("V") Clock = Callable[[], float] -def redirect_uri_problem(uri: str) -> str | None: - """Return why ``uri`` is not an acceptable redirect URI, or None when it is.""" +def redirect_uri_problem(uri: str, allowlist: list[str] | None = None) -> str | None: + """Return why ``uri`` is not an acceptable redirect URI, or None when it is. + + With a non-empty ``allowlist`` an https URI must also match it; http loopback + URIs (native clients on the user's own machine) stay allowed. + """ try: parts = urlsplit(uri) parts.port # noqa: B018 (accessing .port validates it) @@ -108,10 +113,10 @@ def redirect_uri_problem(uri: str) -> str | None: host = (parts.hostname or "").lower() if not host: return _REDIRECT_RULE - if parts.scheme == "https": - return None if parts.scheme == "http" and host in _LOOPBACK_REDIRECT_HOSTS: return None + if parts.scheme == "https": + return None if redirect_matches_allowlist(uri, allowlist or []) else _NOT_ALLOWED return _REDIRECT_RULE @@ -306,7 +311,8 @@ class GiteaOAuthProvider( async def register_client(self, client_info: OAuthClientInformationFull) -> None: """Validate and persist a dynamically registered client. - Redirect URIs must be https, or http on a loopback host. When + Redirect URIs must be https, or http on a loopback host, and match + ``OAUTH_REDIRECT_ALLOWLIST`` when that is set. When :data:`MAX_CLIENTS` is reached, the least recently used client without any session is evicted; when every client has a session, registration is refused. @@ -320,7 +326,7 @@ class GiteaOAuthProvider( "invalid_redirect_uri", f"between 1 and {MAX_REDIRECT_URIS} redirect URIs required" ) for uri in uris: - problem = redirect_uri_problem(uri) + problem = redirect_uri_problem(uri, self._settings.oauth_redirect_allowlist) if problem is not None: raise RegistrationError("invalid_redirect_uri", problem) if client_info.token_endpoint_auth_method not in _CLIENT_AUTH_METHODS: @@ -368,6 +374,12 @@ class GiteaOAuthProvider( """ if params.resource is not None and not same_resource(params.resource, self.resource): raise AuthorizeError("invalid_target", "unknown resource") + # A client registered before the allowlist was set must not slip through it. + if ( + redirect_uri_problem(str(params.redirect_uri), self._settings.oauth_redirect_allowlist) + is not None + ): + raise AuthorizeError("invalid_request", "redirect URI is not allowed") request_id = secrets.token_urlsafe(32) pending = PendingAuthorization( client_id=client.client_id, diff --git a/src/aegis_gitea_mcp/setup_template.py b/src/aegis_gitea_mcp/setup_template.py index 1bd66fa..999e3cd 100644 --- a/src/aegis_gitea_mcp/setup_template.py +++ b/src/aegis_gitea_mcp/setup_template.py @@ -44,6 +44,11 @@ GITEA_OAUTH_CLIENT_SECRET= # Encrypts stored user tokens. At least 32 characters. Generate with: # openssl rand -hex 32 TOKEN_ENCRYPTION_KEY= +# Optional allowlist of redirect URIs clients may register (comma separated): exact +# URIs, or https://host/ prefixes. Empty accepts any https URI (and http on +# loopback); the consent page is then the only check. For a public deployment list the +# hosted clients you use, e.g. https://claude.ai/,https://chat.mistral.ai/ +OAUTH_REDIRECT_ALLOWLIST= # Granular Gitea scopes only (read:/write: activitypub, admin, issue, misc, # notification, organization, package, repository, user). Unknown values are rejected. OAUTH_SCOPES=write:activitypub,write:admin,write:issue,write:misc,write:notification,write:organization,write:package,write:repository,write:user diff --git a/tests/security/test_oauth_redirects.py b/tests/security/test_oauth_redirects.py new file mode 100644 index 0000000..2602c1d --- /dev/null +++ b/tests/security/test_oauth_redirects.py @@ -0,0 +1,167 @@ +"""Redirect URIs accepted at dynamic client registration (SEC-05 and the base rules).""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError +from starlette.testclient import TestClient + +from aegis_gitea_mcp.config import redirect_matches_allowlist +from tests.test_oauth_helpers import browser, build_oauth_app, oauth_settings, register + +ALLOWLIST = ["https://claude.ai/", "https://chat.mistral.ai/", "https://client.example/cb"] + + +@pytest.fixture(autouse=True) +def _env(mock_env: None) -> None: + """The audit logger and the rate limiter read the process-wide settings.""" + + +def web(**settings: object) -> TestClient: + return browser(build_oauth_app(oauth_settings(**settings)).app) + + +def error_of(response: object) -> str: + assert response.status_code == 400 # type: ignore[attr-defined] + return str(response.json()["error"]) # type: ignore[attr-defined] + + +@pytest.mark.parametrize( + "uri", + [ + "https://evil.example/cb", + "https://claude.ai.evil.example/api/mcp/auth_callback", + "https://evilclaude.ai/cb", + "https://claude.ai@evil.example/cb", + "https://client.example/cb2", + "https://client.example/other", + "http://claude.ai/api/mcp/auth_callback", + ], +) +def test_redirect_outside_the_allowlist_is_refused(uri: str) -> None: + response = register(web(oauth_redirect_allowlist=ALLOWLIST), [uri]) + + assert error_of(response) == "invalid_redirect_uri" + + +@pytest.mark.parametrize( + "uri", + [ + "https://claude.ai/api/mcp/auth_callback", + "https://CLAUDE.ai/api/mcp/auth_callback", + "https://chat.mistral.ai/oauth/callback", + "https://client.example/cb", + "http://localhost:3000/callback", + "http://127.0.0.1:8123/cb", + "http://[::1]:8123/cb", + ], +) +def test_allowlisted_and_loopback_redirects_register(uri: str) -> None: + response = register(web(oauth_redirect_allowlist=ALLOWLIST), [uri]) + + assert response.status_code == 201, response.text + + +def test_one_bad_uri_among_good_ones_refuses_the_registration() -> None: + response = register( + web(oauth_redirect_allowlist=ALLOWLIST), + ["https://claude.ai/cb", "https://evil.example/cb"], + ) + + assert error_of(response) == "invalid_redirect_uri" + + +def test_default_allowlist_is_empty_and_accepts_any_https_redirect() -> None: + assert oauth_settings().oauth_redirect_allowlist == [] + assert register(web(), ["https://anything.example/cb"]).status_code == 201 + + +@pytest.mark.parametrize( + "uri", + [ + "http://example.com/cb", + "https://example.com/cb#frag", + "https://user:pass@example.com/cb", + "javascript:alert(1)", + "custom-app://callback", + "file:///etc/passwd", + "data:text/html,x", + "ftp://example.com/cb", + ], +) +@pytest.mark.parametrize("allowlist", [[], ALLOWLIST]) +def test_the_base_redirect_rules_hold_with_and_without_an_allowlist( + uri: str, allowlist: list[str] +) -> None: + response = register(web(oauth_redirect_allowlist=allowlist), [uri]) + + assert response.status_code == 400, response.text + + +@pytest.mark.parametrize( + "entry", + [ + "https://claude.ai", + "claude.ai/", + "http://claude.ai/", + "https://user@claude.ai/", + "https://claude.ai/?x=1", + "https://claude.ai/#f", + "https://claude .ai/", + "ftp://claude.ai/", + "https:///", + ], +) +def test_malformed_allowlist_entries_stop_startup(entry: str) -> None: + with pytest.raises(ValidationError): + oauth_settings(oauth_redirect_allowlist=[entry]) + + +def test_allowlist_is_read_from_a_comma_separated_variable( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from aegis_gitea_mcp.config import Settings + + monkeypatch.setenv("OAUTH_REDIRECT_ALLOWLIST", "https://claude.ai/, https://chat.mistral.ai/") + monkeypatch.setenv("GITEA_URL", "https://gitea.example.com") + + assert Settings().oauth_redirect_allowlist == ["https://claude.ai/", "https://chat.mistral.ai/"] + + +def test_prefix_matching_is_by_host_not_by_string_prefix() -> None: + assert redirect_matches_allowlist("https://claude.ai/x", ["https://claude.ai/"]) + assert not redirect_matches_allowlist( + "https://claude.ai.evil.example/x", ["https://claude.ai/"] + ) + assert not redirect_matches_allowlist("https://claude.ai:8443/x", ["https://claude.ai/"]) + assert redirect_matches_allowlist("https://anything/x", []) + + +def test_a_client_registered_before_the_allowlist_cannot_authorize() -> None: + from tests.test_oauth_helpers import pkce + + unrestricted = build_oauth_app(oauth_settings()) + client = browser(unrestricted.app) + registration = register(client, ["https://evil.example/cb"]).json() + assert registration["client_id"] + + # Same store, allowlist now set: /authorize must refuse the old client. + restricted = build_oauth_app( + oauth_settings(oauth_redirect_allowlist=ALLOWLIST), clock=unrestricted.clock + ) + restricted.store.clients.update(unrestricted.store.clients) + response = browser(restricted.app).get( + "/authorize", + params={ + "client_id": registration["client_id"], + "redirect_uri": "https://evil.example/cb", + "response_type": "code", + "code_challenge": pkce()[1], + "code_challenge_method": "S256", + }, + ) + + location = response.headers["location"] + assert response.status_code == 302 + assert "error=invalid_request" in location + assert "/oauth/consent" not in location and "code=" not in location