diff --git a/docs/security.md b/docs/security.md index 5199634..9f2cb10 100644 --- a/docs/security.md +++ b/docs/security.md @@ -184,8 +184,11 @@ confirmation, `READ_ONLY`, `TOOLSETS`, `DISABLED_TOOLS` and narrow token scopes The server is an OAuth 2.1 authorization server for MCP clients (SDK provider, dynamic client registration, PKCE S256 required) and a confidential PKCE client of Gitea. Flow: `/authorize` parks the request and shows our consent page (client name, redirect URI, warning). Approval -needs the form CSRF token and the matching `SameSite=Strict` cookie and sets a -browser-binding cookie (`__Host-` prefixed and `Secure` on https, `HttpOnly`). The Gitea +needs the form CSRF token, the matching `SameSite=Strict` cookie and a request `Origin` equal +to the public origin (the `Referer` origin when `Origin` is absent; neither, or `null`, is +refused). It sets a browser-binding cookie named per flow (`aegis-oauth-flow-<8 hex of the +state hash>`, `__Host-` prefixed and `Secure` on https, `HttpOnly`), so concurrent sign-ins in +one browser do not overwrite each other. The Gitea callback is completed only in the browser holding that cookie; unknown, expired or replayed states never redirect. Pages send `Cache-Control: no-store`, `Referrer-Policy: no-referrer`, `X-Frame-Options: DENY` and a CSP with `frame-ancestors 'none'`. Redirect URIs must be https, diff --git a/src/aegis_gitea_mcp/oauth/callback.py b/src/aegis_gitea_mcp/oauth/callback.py index 4def927..51e9021 100644 --- a/src/aegis_gitea_mcp/oauth/callback.py +++ b/src/aegis_gitea_mcp/oauth/callback.py @@ -23,7 +23,7 @@ from starlette.responses import HTMLResponse, RedirectResponse, Response from starlette.routing import Route, request_response from starlette.types import ASGIApp, Receive, Scope, Send -from aegis_gitea_mcp.oauth.provider import CONSENT_PATH, GiteaOAuthProvider +from aegis_gitea_mcp.oauth.provider import CONSENT_PATH, GiteaOAuthProvider, flow_tag from aegis_gitea_mcp.oauth.upstream import CALLBACK_PATH COOKIE_TTL_SECONDS = 600 @@ -64,7 +64,11 @@ class _Cookies: self.secure = urlsplit(base_url).scheme == "https" prefix = "__Host-" if self.secure else "" self.csrf = f"{prefix}aegis-oauth-csrf" - self.binding = f"{prefix}aegis-oauth-flow" + self._binding_prefix = f"{prefix}aegis-oauth-flow-" + + def binding(self, tag: str) -> str: + """Name of the binding cookie of the flow with the given tag (one per Gitea state).""" + return f"{self._binding_prefix}{tag}" def set(self, response: Response, name: str, value: str, *, same_site: str) -> None: response.set_cookie( @@ -103,6 +107,26 @@ def _redirect(location: str, status_code: int) -> RedirectResponse: return RedirectResponse(location, status_code=status_code, headers=_SECURITY_HEADERS) +def _request_origin(request: Request) -> str | None: + """Origin of a form POST: the ``Origin`` header, else the origin of ``Referer``. + + Browsers send ``Origin`` on form POSTs. Without it (or with ``Origin: null``) the + ``Referer`` is the fallback; when neither names an origin the result is None + and the request is refused. + """ + origin = request.headers.get("origin") + if origin is not None: + return origin + referer = request.headers.get("referer") + if not referer: + return None + try: + parts = urlsplit(referer) + except ValueError: + return None + return f"{parts.scheme}://{parts.netloc}" if parts.scheme and parts.netloc else None + + def _text(value: Any, limit: int = 200) -> str: return value[:limit] if isinstance(value, str) else "" @@ -164,8 +188,7 @@ def build_browser_routes(provider: GiteaOAuthProvider) -> list[Route]: return response async def consent_submit(request: Request) -> Response: - origin = request.headers.get("origin") - if origin is not None and origin != expected_origin: + if _request_origin(request) != expected_origin: return _invalid() try: form = await request.form( @@ -183,21 +206,28 @@ def build_browser_routes(provider: GiteaOAuthProvider) -> list[Route]: return _invalid() response = _redirect(decision.redirect, 303) cookies.clear(response, cookies.csrf, same_site="strict") - if decision.binding_token is not None: + if decision.binding_token is not None and decision.flow_tag is not None: # Lax: the cookie must come back on Gitea's top-level redirect to us. - cookies.set(response, cookies.binding, decision.binding_token, same_site="lax") + cookies.set( + response, + cookies.binding(decision.flow_tag), + decision.binding_token, + same_site="lax", + ) return response async def callback(request: Request) -> Response: params = request.query_params + state = _text(params.get("state")) + cookie_name = cookies.binding(flow_tag(state)) location = await provider.complete_callback( - _text(params.get("state")), - _text(request.cookies.get(cookies.binding)), + state, + _text(request.cookies.get(cookie_name)), _text(params.get("code"), 4096), _text(params.get("error")), ) response = _invalid() if location is None else _redirect(location, 302) - cookies.clear(response, cookies.binding, same_site="lax") + cookies.clear(response, cookie_name, same_site="lax") return response return [ diff --git a/src/aegis_gitea_mcp/oauth/provider.py b/src/aegis_gitea_mcp/oauth/provider.py index e954695..b5baf3f 100644 --- a/src/aegis_gitea_mcp/oauth/provider.py +++ b/src/aegis_gitea_mcp/oauth/provider.py @@ -129,6 +129,16 @@ def same_resource(left: str, right: str) -> bool: return normalized[0] == normalized[1] +def flow_tag(state: str) -> str: + """Eight hex characters derived from a Gitea ``state`` (names the browser-binding cookie). + + Concurrent sign-ins in one browser each get their own cookie, so a second flow + cannot overwrite the binding of the first. The tag is public (it is a hash + prefix of a value that is in the callback URL anyway) and grants nothing. + """ + return token_hash(state)[:8] + + def s256(verifier: str) -> str: """PKCE S256 challenge for ``verifier``.""" digest = hashlib.sha256(verifier.encode("ascii")).digest() @@ -214,6 +224,8 @@ class ConsentDecision: redirect: str = field(repr=False) binding_token: str | None = field(default=None, repr=False) + #: Short public tag of the Gitea state, used to name this flow's binding cookie. + flow_tag: str | None = None class GiteaOAuthProvider( @@ -438,7 +450,7 @@ class GiteaOAuthProvider( location = self._upstream.authorize_url( state=state, code_challenge=s256(pending.code_verifier) ) - return ConsentDecision(redirect=location, binding_token=binding) + return ConsentDecision(redirect=location, binding_token=binding, flow_tag=flow_tag(state)) async def complete_callback( self, state: str, binding: str, code: str, error: str diff --git a/tests/security/test_oauth_csrf.py b/tests/security/test_oauth_csrf.py new file mode 100644 index 0000000..9f7a6d2 --- /dev/null +++ b/tests/security/test_oauth_csrf.py @@ -0,0 +1,238 @@ +"""CSRF, origin and browser-binding checks of the consent and callback routes (SEC-17, SEC-19).""" + +from __future__ import annotations + +import pytest +from pytest_httpx import HTTPXMock +from starlette.testclient import TestClient + +from aegis_gitea_mcp.oauth.provider import PENDING_TTL +from tests.test_oauth_helpers import ( + BASE, + CLIENT_STATE, + CSRF_FIELD, + GITEA_CODE, + REDIRECT, + FakeClock, + browser, + build_oauth_app, + exchange_code, + gitea_requests, + mock_gitea_login, + pkce, + query_of, + register, + start_flow, +) + + +@pytest.fixture(autouse=True) +def _env(mock_env: None) -> None: + """The audit logger and the rate limiter read the process-wide settings.""" + + +def open_consent(client: TestClient, registration: dict[str, str]) -> tuple[dict[str, str], str]: + """Register and call /authorize, open the consent page; return (form fields, consent URL).""" + authorize = client.get( + "/authorize", + params={ + "client_id": registration["client_id"], + "redirect_uri": REDIRECT, + "response_type": "code", + "code_challenge": pkce()[1], + "code_challenge_method": "S256", + "state": CLIENT_STATE, + }, + ) + consent_url = authorize.headers["location"] + match = CSRF_FIELD.search(client.get(consent_url).text) + assert match is not None + form = {"request": query_of(consent_url)["request"], "csrf": match.group(1)} + return {**form, "decision": "approve"}, consent_url + + +def assert_refused(response: object) -> None: + assert response.status_code == 400 # type: ignore[attr-defined] + assert "location" not in response.headers # type: ignore[attr-defined] + + +# Consent POST (SEC-17) + + +@pytest.mark.parametrize( + "headers", + [ + {}, + {"Origin": "null"}, + {"Origin": "https://evil.example"}, + {"Origin": "http://mcp.example.com"}, + {"Origin": "https://mcp.example.com.evil.example"}, + {"Referer": "https://evil.example/page"}, + {"Referer": "not a url"}, + {"Origin": "https://evil.example", "Referer": f"{BASE}/oauth/consent"}, + ], + ids=lambda headers: ",".join(f"{k}={v}" for k, v in headers.items()) or "no-origin-no-referer", +) +def test_consent_post_without_a_matching_origin_is_refused(headers: dict[str, str]) -> None: + oauth = build_oauth_app() + client = browser(oauth.app) + form, _ = open_consent(client, register(client).json()) + + response = client.post("/oauth/consent", data=form, headers=headers) + + assert_refused(response) + # The request was not consumed: the real form still works afterwards. + assert client.post("/oauth/consent", data=form, headers={"Origin": BASE}).status_code == 303 + + +def test_consent_post_falls_back_to_the_referer_origin() -> None: + client = browser(build_oauth_app().app) + form, consent_url = open_consent(client, register(client).json()) + + response = client.post("/oauth/consent", data=form, headers={"Referer": consent_url}) + + assert response.status_code == 303 + + +def test_consent_post_without_the_csrf_cookie_is_refused() -> None: + oauth = build_oauth_app() + client = browser(oauth.app) + form, _ = open_consent(client, register(client).json()) + + assert_refused(browser(oauth.app).post("/oauth/consent", data=form, headers={"Origin": BASE})) + + +def test_consent_post_with_a_mismatched_csrf_token_is_refused() -> None: + client = browser(build_oauth_app().app) + form, _ = open_consent(client, register(client).json()) + + assert_refused( + client.post("/oauth/consent", data={**form, "csrf": "x" * 43}, headers={"Origin": BASE}) + ) + assert_refused( + client.post("/oauth/consent", data={**form, "csrf": ""}, headers={"Origin": BASE}) + ) + missing = {key: value for key, value in form.items() if key != "csrf"} + assert_refused(client.post("/oauth/consent", data=missing, headers={"Origin": BASE})) + + +def test_a_valid_consent_still_works_and_is_single_use() -> None: + client = browser(build_oauth_app().app) + form, _ = open_consent(client, register(client).json()) + + approved = client.post("/oauth/consent", data=form, headers={"Origin": BASE}) + replay = client.post("/oauth/consent", data=form, headers={"Origin": BASE}) + + assert approved.status_code == 303 + assert approved.headers["location"].startswith( + "https://gitea.example.com/login/oauth/authorize" + ) + assert_refused(replay) + + +# Callback + + +def test_callback_without_the_binding_cookie_is_refused_and_audited( + httpx_mock: HTTPXMock, +) -> None: + oauth = build_oauth_app() + client = browser(oauth.app) + gitea_url = start_flow(client, register(client).json(), pkce()[1]) + + response = browser(oauth.app).get( + "/oauth/callback", params={"code": GITEA_CODE, "state": query_of(gitea_url)["state"]} + ) + + assert_refused(response) + assert gitea_requests(httpx_mock, "https://gitea.example.com") == [] + audit = oauth.settings.resolved_audit_log_path.read_text("utf-8") + assert "oauth_callback_binding_mismatch" in audit + + +def test_callback_with_another_flows_cookie_is_refused(httpx_mock: HTTPXMock) -> None: + oauth = build_oauth_app() + client = browser(oauth.app) + registration = register(client).json() + first = start_flow(client, registration, pkce()[1]) + cookies = dict(client.cookies) + second_client = browser(oauth.app) + second = start_flow(second_client, register(second_client).json(), pkce()[1]) + + # The cookie of flow one is presented for flow two's state. + attacker = browser(oauth.app) + for name, value in cookies.items(): + attacker.cookies.set(name, value) + response = attacker.get( + "/oauth/callback", params={"code": GITEA_CODE, "state": query_of(second)["state"]} + ) + + assert_refused(response) + assert query_of(first)["state"] != query_of(second)["state"] + assert gitea_requests(httpx_mock, "https://gitea.example.com") == [] + + +def test_callback_with_a_replayed_state_is_refused(httpx_mock: HTTPXMock) -> None: + oauth = build_oauth_app() + client = browser(oauth.app) + state = query_of(start_flow(client, register(client).json(), pkce()[1]))["state"] + mock_gitea_login(httpx_mock) + + first = client.get("/oauth/callback", params={"code": GITEA_CODE, "state": state}) + replay = client.get("/oauth/callback", params={"code": GITEA_CODE, "state": state}) + + assert first.status_code == 302 + assert_refused(replay) + + +def test_callback_with_an_expired_or_unknown_state_is_refused() -> None: + clock = FakeClock() + oauth = build_oauth_app(clock=clock) + client = browser(oauth.app) + state = query_of(start_flow(client, register(client).json(), pkce()[1]))["state"] + clock.advance(PENDING_TTL + 1) + + assert_refused(client.get("/oauth/callback", params={"code": GITEA_CODE, "state": state})) + assert_refused(client.get("/oauth/callback", params={"code": GITEA_CODE, "state": "unknown"})) + assert_refused(client.get("/oauth/callback", params={"code": GITEA_CODE})) + assert_refused(client.get("/oauth/callback")) + + +def test_callback_never_echoes_its_parameters() -> None: + client = browser(build_oauth_app().app) + + response = client.get( + "/oauth/callback", params={"code": "code-value-123", "state": "state-value-123"} + ) + + assert "code-value-123" not in response.text and "state-value-123" not in response.text + + +# Concurrent sign-ins in one browser (SEC-19) + + +def test_two_interleaved_flows_in_one_browser_both_complete(httpx_mock: HTTPXMock) -> None: + oauth = build_oauth_app() + client = browser(oauth.app) + registration = register(client).json() + verifier_one, challenge_one = pkce() + verifier_two, challenge_two = pkce() + first = start_flow(client, registration, challenge_one) + second = start_flow(client, registration, challenge_two) + assert len({name for name in client.cookies if "oauth-flow" in name}) == 2 + + mock_gitea_login(httpx_mock, login="alice") + mock_gitea_login(httpx_mock, login="alice") + done_first = client.get( + "/oauth/callback", params={"code": GITEA_CODE, "state": query_of(first)["state"]} + ) + done_second = client.get( + "/oauth/callback", params={"code": GITEA_CODE, "state": query_of(second)["state"]} + ) + + assert done_first.status_code == 302 and done_second.status_code == 302 + for done, verifier in ((done_first, verifier_one), (done_second, verifier_two)): + code = query_of(done.headers["location"])["code"] + assert exchange_code(client, registration, code, verifier).status_code == 200 + audit = oauth.settings.resolved_audit_log_path.read_text("utf-8") + assert "oauth_callback_binding_mismatch" not in audit diff --git a/tests/test_oauth_flow_security.py b/tests/test_oauth_flow_security.py index 050b144..0b0f467 100644 --- a/tests/test_oauth_flow_security.py +++ b/tests/test_oauth_flow_security.py @@ -393,6 +393,7 @@ def test_callback_in_a_browser_that_did_not_start_the_flow_is_refused( "csrf": match.group(1), "decision": "approve", }, + headers={"Origin": BASE}, ) state = query_of(approved.headers["location"])["state"] @@ -440,20 +441,21 @@ def test_consent_post_needs_the_csrf_cookie_token_and_origin() -> None: request_id, csrf = consent_form(client) form = {"request": request_id, "csrf": csrf, "decision": "approve"} - cross_site = browser(oauth.app).post("/oauth/consent", data=form) # no Strict cookie - wrong_token = client.post("/oauth/consent", data={**form, "csrf": "nope"}) + origin = {"Origin": BASE} + cross_site = browser(oauth.app).post("/oauth/consent", data=form, headers=origin) # no cookie + wrong_token = client.post("/oauth/consent", data={**form, "csrf": "nope"}, headers=origin) bad_origin = client.post( "/oauth/consent", data=form, headers={"Origin": "https://evil.example"} ) - unknown = client.post("/oauth/consent", data={**form, "request": "unknown"}) - good = client.post("/oauth/consent", data=form, headers={"Origin": BASE}) - again = client.post("/oauth/consent", data=form) + unknown = client.post("/oauth/consent", data={**form, "request": "unknown"}, headers=origin) + good = client.post("/oauth/consent", data=form, headers=origin) + again = client.post("/oauth/consent", data=form, headers=origin) assert [r.status_code for r in (cross_site, wrong_token, bad_origin, unknown)] == [400] * 4 assert good.status_code == 303 assert again.status_code == 400 # single use binding = good.headers["set-cookie"].lower() - assert "__host-aegis-oauth-flow=" in binding and "samesite=lax" in binding + assert "__host-aegis-oauth-flow-" in binding and "samesite=lax" in binding def test_unknown_consent_request_page_is_a_400() -> None: @@ -468,7 +470,9 @@ def test_denying_consent_returns_access_denied_to_the_client() -> None: request_id, csrf = consent_form(client) denied = client.post( - "/oauth/consent", data={"request": request_id, "csrf": csrf, "decision": "deny"} + "/oauth/consent", + data={"request": request_id, "csrf": csrf, "decision": "deny"}, + headers={"Origin": BASE}, ) assert denied.status_code == 303