fix: require a matching Origin on consent and name the binding cookie per flow (SEC-17, SEC-19)

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

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