Dev #72
No files matched your search
@@ -34,6 +34,11 @@ GITEA_OAUTH_CLIENT_SECRET=
|
|||||||
# Encrypts stored user tokens. At least 32 characters. Generate with:
|
# Encrypts stored user tokens. At least 32 characters. Generate with:
|
||||||
# openssl rand -hex 32
|
# openssl rand -hex 32
|
||||||
TOKEN_ENCRYPTION_KEY=
|
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,
|
# Granular Gitea scopes only (read:/write: activitypub, admin, issue, misc,
|
||||||
# notification, organization, package, repository, user). Unknown values are rejected.
|
# 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
|
OAUTH_SCOPES=write:activitypub,write:admin,write:issue,write:misc,write:notification,write:organization,write:package,write:repository,write:user
|
||||||
|
|||||||
@@ -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. |
|
| `TOKEN_ENCRYPTION_KEY` | unset | At least 32 characters. Encrypts stored user tokens. Losing it invalidates every stored login. |
|
||||||
| `OAUTH_SCOPES` | all `write:<category>` 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_SCOPES` | all `write:<category>` 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
|
`PUBLIC_BASE_URL` is also required for OAuth. The Gitea OAuth application's redirect URI is
|
||||||
`<PUBLIC_BASE_URL>/oauth/callback`.
|
`<PUBLIC_BASE_URL>/oauth/callback`.
|
||||||
|
|
||||||
|
|||||||
+6
-3
@@ -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`) |
|
| 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` |
|
| 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` |
|
| 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` |
|
| Code or refresh token replay | Single-use codes; replay or reuse of a rotated refresh token revokes the grant | `oauth/provider.py` |
|
||||||
|
|
||||||
Tampering
|
Tampering
|
||||||
@@ -127,8 +127,10 @@ Elevation of privilege
|
|||||||
Gitea token committed to a file) is returned unless it matches a pattern.
|
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
|
- SSRF through Gitea features (migrations, webhooks, mirrors) is Gitea's responsibility; this
|
||||||
server only ever contacts `GITEA_URL`.
|
server only ever contacts `GITEA_URL`.
|
||||||
- The consent page is the only barrier against phishing with a self-registered client
|
- With the default `OAUTH_REDIRECT_ALLOWLIST` (empty), the consent page is the only barrier
|
||||||
(SEC-05).
|
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
|
- Key rotation for `TOKEN_ENCRYPTION_KEY` is not supported: a new key means moving the store
|
||||||
away and signing every user out.
|
away and signing every user out.
|
||||||
- The audit chain is tamper-evident, not tamper-proof: anyone with write access to the file can
|
- 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-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-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-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. |
|
||||||
@@ -88,6 +88,64 @@ def _split_list(value: Any) -> Any:
|
|||||||
return value
|
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:
|
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 without query, fragment or trailing slash."""
|
||||||
candidate = value.strip()
|
candidate = value.strip()
|
||||||
@@ -142,6 +200,7 @@ class Settings(BaseSettings):
|
|||||||
oauth_scopes: Annotated[list[str], NoDecode] = Field(
|
oauth_scopes: Annotated[list[str], NoDecode] = Field(
|
||||||
default_factory=lambda: DEFAULT_OAUTH_SCOPES.split(",")
|
default_factory=lambda: DEFAULT_OAUTH_SCOPES.split(",")
|
||||||
)
|
)
|
||||||
|
oauth_redirect_allowlist: Annotated[list[str], NoDecode] = Field(default_factory=list)
|
||||||
|
|
||||||
# State
|
# State
|
||||||
data_dir: Path = Field(default_factory=default_data_dir)
|
data_dir: Path = Field(default_factory=default_data_dir)
|
||||||
@@ -250,7 +309,14 @@ class Settings(BaseSettings):
|
|||||||
)
|
)
|
||||||
return value
|
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
|
@classmethod
|
||||||
def _parse_list(cls, value: Any) -> Any:
|
def _parse_list(cls, value: Any) -> Any:
|
||||||
"""Accept comma or whitespace separated strings for list settings."""
|
"""Accept comma or whitespace separated strings for list settings."""
|
||||||
@@ -271,6 +337,18 @@ class Settings(BaseSettings):
|
|||||||
raise ValueError("OAUTH_SCOPES must not be empty")
|
raise ValueError("OAUTH_SCOPES must not be empty")
|
||||||
return list(dict.fromkeys(value))
|
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")
|
@field_validator("toolsets")
|
||||||
@classmethod
|
@classmethod
|
||||||
def _validate_toolsets(cls, value: list[str]) -> list[str]:
|
def _validate_toolsets(cls, value: list[str]) -> list[str]:
|
||||||
|
|||||||
@@ -59,7 +59,7 @@ from pydantic import AnyHttpUrl, ValidationError
|
|||||||
|
|
||||||
from aegis_gitea_mcp.audit import AuditLogger, get_audit_logger
|
from aegis_gitea_mcp.audit import AuditLogger, get_audit_logger
|
||||||
from aegis_gitea_mcp.auth import MCP_SCOPE
|
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 (
|
from aegis_gitea_mcp.oauth.store import (
|
||||||
ClientRecord,
|
ClientRecord,
|
||||||
OAuthStore,
|
OAuthStore,
|
||||||
@@ -87,6 +87,7 @@ CONSENT_PATH = "/oauth/consent"
|
|||||||
|
|
||||||
_LOOPBACK_REDIRECT_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
|
_LOOPBACK_REDIRECT_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"})
|
||||||
_CLIENT_AUTH_METHODS = frozenset({"none", "client_secret_post", "client_secret_basic"})
|
_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_RULE = (
|
||||||
"redirect URIs must use https://, or http:// on localhost, 127.0.0.1 or [::1], "
|
"redirect URIs must use https://, or http:// on localhost, 127.0.0.1 or [::1], "
|
||||||
"without credentials or a fragment"
|
"without credentials or a fragment"
|
||||||
@@ -96,8 +97,12 @@ V = TypeVar("V")
|
|||||||
Clock = Callable[[], float]
|
Clock = Callable[[], float]
|
||||||
|
|
||||||
|
|
||||||
def redirect_uri_problem(uri: str) -> str | None:
|
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."""
|
"""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:
|
try:
|
||||||
parts = urlsplit(uri)
|
parts = urlsplit(uri)
|
||||||
parts.port # noqa: B018 (accessing .port validates it)
|
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()
|
host = (parts.hostname or "").lower()
|
||||||
if not host:
|
if not host:
|
||||||
return _REDIRECT_RULE
|
return _REDIRECT_RULE
|
||||||
if parts.scheme == "https":
|
|
||||||
return None
|
|
||||||
if parts.scheme == "http" and host in _LOOPBACK_REDIRECT_HOSTS:
|
if parts.scheme == "http" and host in _LOOPBACK_REDIRECT_HOSTS:
|
||||||
return None
|
return None
|
||||||
|
if parts.scheme == "https":
|
||||||
|
return None if redirect_matches_allowlist(uri, allowlist or []) else _NOT_ALLOWED
|
||||||
return _REDIRECT_RULE
|
return _REDIRECT_RULE
|
||||||
|
|
||||||
|
|
||||||
@@ -306,7 +311,8 @@ class GiteaOAuthProvider(
|
|||||||
async def register_client(self, client_info: OAuthClientInformationFull) -> None:
|
async def register_client(self, client_info: OAuthClientInformationFull) -> None:
|
||||||
"""Validate and persist a dynamically registered client.
|
"""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
|
:data:`MAX_CLIENTS` is reached, the least recently used client without
|
||||||
any session is evicted; when every client has a session, registration
|
any session is evicted; when every client has a session, registration
|
||||||
is refused.
|
is refused.
|
||||||
@@ -320,7 +326,7 @@ class GiteaOAuthProvider(
|
|||||||
"invalid_redirect_uri", f"between 1 and {MAX_REDIRECT_URIS} redirect URIs required"
|
"invalid_redirect_uri", f"between 1 and {MAX_REDIRECT_URIS} redirect URIs required"
|
||||||
)
|
)
|
||||||
for uri in uris:
|
for uri in uris:
|
||||||
problem = redirect_uri_problem(uri)
|
problem = redirect_uri_problem(uri, self._settings.oauth_redirect_allowlist)
|
||||||
if problem is not None:
|
if problem is not None:
|
||||||
raise RegistrationError("invalid_redirect_uri", problem)
|
raise RegistrationError("invalid_redirect_uri", problem)
|
||||||
if client_info.token_endpoint_auth_method not in _CLIENT_AUTH_METHODS:
|
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):
|
if params.resource is not None and not same_resource(params.resource, self.resource):
|
||||||
raise AuthorizeError("invalid_target", "unknown 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)
|
request_id = secrets.token_urlsafe(32)
|
||||||
pending = PendingAuthorization(
|
pending = PendingAuthorization(
|
||||||
client_id=client.client_id,
|
client_id=client.client_id,
|
||||||
|
|||||||
@@ -44,6 +44,11 @@ GITEA_OAUTH_CLIENT_SECRET=
|
|||||||
# Encrypts stored user tokens. At least 32 characters. Generate with:
|
# Encrypts stored user tokens. At least 32 characters. Generate with:
|
||||||
# openssl rand -hex 32
|
# openssl rand -hex 32
|
||||||
TOKEN_ENCRYPTION_KEY=
|
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,
|
# Granular Gitea scopes only (read:/write: activitypub, admin, issue, misc,
|
||||||
# notification, organization, package, repository, user). Unknown values are rejected.
|
# 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
|
OAUTH_SCOPES=write:activitypub,write:admin,write:issue,write:misc,write:notification,write:organization,write:package,write:repository,write:user
|
||||||
|
|||||||
@@ -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
|
||||||
Reference in new issue
Block a user