fix: Origin allowlist with Sec-Fetch-Site, and verify the cookie's inner access token

README.md

@@ -105,10 +105,25 @@ are present. It does not refresh: the auth routes refresh a browser session sile
 client calls `POST /auth/refresh` itself. `Adapter.authenticate(headers)` does the same check
 anywhere else.
 
-A request the guard admits on the session cookie also gets the auth routes' cross-site check:
-while cookies are `SameSite=None`, a state-changing request from another site (`Sec-Fetch-Site:
-cross-site`, or an `Origin` outside `allowed_origins`) answers 403. A bearer token is never
-attached by a browser, so it is not checked.
+The session cookie is signed with `cookie_secret`, and the guard also verifies the auth API access
+token inside it against the API's key set, as it does a bearer token: RS256, the expected issuer
+and audience, in date, `typ: access`, and the same `sub` as the cookie. `User.roles` and
+`User.recovery` come from that verified token, not from the cookie, so a forged cookie cannot carry
+claims of its own even if the cookie secret leaks. The first request with a new key fetches the key
+set, and when it cannot be fetched the guard answers 401.
+
+A request the guard admits on the session cookie also gets the auth routes' cross-site check,
+while cookies are `SameSite=None` and only for methods other than `GET`, `HEAD` and `OPTIONS`:
+
+- `Sec-Fetch-Site: cross-site` answers 403.
+- `Sec-Fetch-Site: same-site` also covers a sibling subdomain, so when `allowed_origins` is set the
+  request must carry an `Origin` on that list, or it answers 403. Without `allowed_origins` it
+  passes.
+- `Sec-Fetch-Site: same-origin` (or `none`) passes.
+- With no `Sec-Fetch-Site`, an `Origin: null` answers 403, and when `allowed_origins` is set any
+  other `Origin` must be on it. A request with neither header passes.
+
+A bearer token is never attached by a browser, so it is not checked.
 
 ### Recovery sessions
 

src/seamless_auth/_adapter.py

@@ -868,26 +868,48 @@ def _authenticate(self, headers: Headers, allow_recovery: bool) -> tuple[User, b
     def _session_of(self, headers: Headers) -> tuple[User, bool]:
         name = self._config.access_cookie_name
         if raw_cookie(headers, name):
-            claims = read_cookie(self._config, headers, name, KIND_ACCESS)
-            token = str_claim(claims, "token")
-            if claims is None or not str_claim(claims, "sub") or not _is_access_token(token):
+            cookie = read_cookie(self._config, headers, name, KIND_ACCESS)
+            subject = str_claim(cookie, "sub")
+            token = str_claim(cookie, "token")
+            if cookie is None or not subject or not token:
                 raise Unauthenticated
-            return _user_from(claims, token, _is_recovery_token(token)), True
+            # The cookie secret alone can sign a cookie. Holding the token inside it to
+            # the API's key set means a forged cookie cannot carry claims of its own.
+            claims = self._verified_access_token(token)
+            if str_claim(claims, "sub") != subject:
+                raise Unauthenticated
+            return (
+                User(
+                    id=subject,
+                    roles=_roles(claims),
+                    email=str_claim(cookie, "email") or None,
+                    phone=str_claim(cookie, "phone") or None,
+                    recovery=claims.get("recovery") is True,
+                    token=token,
+                ),
+                True,
+            )
 
         bearer = _bearer_token(headers)
         if bearer is None:
             raise Unauthenticated
+        claims = self._verified_access_token(bearer)
+        return _user_from(claims, bearer, claims.get("recovery") is True), False
+
+    def _verified_access_token(self, token: str) -> Claims:
+        """The claims of an auth API access token, verified against its key set."""
+        issuer, audience = self._config.auth_server_issuer, self._config.audience
+        if not issuer or not audience:
+            raise Unauthenticated
         try:
-            claims = self._jwks.verify(
-                bearer, self._config.auth_server_issuer, self._config.audience
-            )
+            claims = self._jwks.verify(token, issuer, audience)
         except TokenError as err:
             raise Unauthenticated from err
         # An ephemeral sign-in token is signed by the same key, so the type is what
         # keeps it from passing for a session.
         if str_claim(claims, "typ") != "access" or not str_claim(claims, "sub"):
             raise Unauthenticated
-        return _user_from(claims, bearer, claims.get("recovery") is True), False
+        return claims
 
     def _check_origin(self, method: str, headers: Headers) -> _Reply | None:
         """Blocks cross-site state changes while cookies are SameSite=None, which is
@@ -898,22 +920,33 @@ def _check_origin(self, method: str, headers: Headers) -> _Reply | None:
             return None
         blocked = _error(403, "cross_site_request_blocked")
 
+        origin = headers.get("origin")
         site = headers.get("sec-fetch-site")
         if site is not None:
-            return blocked if site.lower() == "cross-site" else None
-        origin = headers.get("origin")
+            site = site.lower()
+            if site == "cross-site":
+                return blocked
+            # same-site also covers a sibling subdomain, so an adopter who listed
+            # allowed origins gets the Origin held to that list.
+            if site != "same-site" or not self._config.allowed_origins:
+                return None
+            return None if origin is not None and self._is_allowed_origin(origin) else blocked
         if origin is None:
             return None
         if origin == "null":
             return blocked
         if not self._config.allowed_origins:
             return None
+        return None if self._is_allowed_origin(origin) else blocked
+
+    def _is_allowed_origin(self, origin: str) -> bool:
         normalized = _normalize_origin(origin)
-        if normalized and any(
+        # Checked so an unparseable Origin cannot match an unparseable allowlist entry.
+        if not normalized:
+            return False
+        return any(
             _normalize_origin(allowed) == normalized for allowed in self._config.allowed_origins
-        ):
-            return None
-        return blocked
+        )
 
 
 _NO_REFRESH_COOKIE = "no refresh cookie"
@@ -1011,13 +1044,6 @@ def _bearer_token(headers: Headers) -> str | None:
     return None
 
 
-def _is_access_token(token: str) -> bool:
-    """Reads the type of the auth API token inside a session cookie. It was verified
-    against the API's key set before the adapter signed it into the cookie, so its
-    claims are read without verifying it again."""
-    return bool(token) and str_claim(unverified_claims(token), "typ") == "access"
-
-
 def _is_recovery_token(token: str) -> bool:
     """Whether an auth API token, verified before it was stored or forwarded, is a
     recovery session's. Only a literal ``true`` counts."""
@@ -1040,11 +1066,15 @@ def _is_placeholder(secret: str) -> bool:
     return secret in PLACEHOLDER_SECRETS or PLACEHOLDER_MARKER in secret.lower()
 
 
-def _user_from(claims: Claims, token: str, recovery: bool) -> User:
+def _roles(claims: Claims) -> tuple[str, ...]:
     roles = claims.get("roles")
+    return tuple(r for r in roles if isinstance(r, str)) if isinstance(roles, list) else ()
+
+
+def _user_from(claims: Claims, token: str, recovery: bool) -> User:
     return User(
         id=str_claim(claims, "sub"),
-        roles=tuple(r for r in roles if isinstance(r, str)) if isinstance(roles, list) else (),
+        roles=_roles(claims),
         email=str_claim(claims, "email") or None,
         phone=str_claim(claims, "phone") or None,
         recovery=recovery,

tests/test_adapter.py

@@ -531,6 +531,60 @@ def test_cross_site_state_changes_are_blocked(api: FakeApi) -> None:
     assert r.status == 400
 
 
+APP = "https://app.example"
+ORIGIN_CASES = [
+    # (allowlist, same_site, method, sec-fetch-site, origin, blocked)
+    ((), "None", "POST", "cross-site", None, True),
+    ((), "None", "POST", "same-origin", None, False),
+    ((), "None", "POST", "same-site", None, False),
+    ((), "None", "POST", "same-site", "https://evil.example", False),
+    ((), "None", "POST", None, None, False),
+    ((), "None", "POST", None, "null", True),
+    ((), "None", "POST", None, "https://evil.example", False),
+    ((), "None", "GET", "cross-site", None, False),
+    ((), "None", "OPTIONS", "cross-site", None, False),
+    ((), "Lax", "POST", "cross-site", None, False),
+    ((APP,), "Lax", "POST", "same-site", "https://evil.example", False),
+    ((APP,), "None", "POST", None, APP, False),
+    ((APP,), "None", "POST", None, "https://evil.example", True),
+    # A sibling subdomain is same-site, so the header alone cannot tell it from the app.
+    ((APP,), "None", "POST", "same-site", "https://evil.example", True),
+    ((APP,), "None", "POST", "same-site", None, True),
+    ((APP,), "None", "POST", "same-site", "null", True),
+    ((APP,), "None", "POST", "same-site", "https://APP.example", False),
+    ((APP,), "None", "POST", "SAME-SITE", APP, False),
+    ((APP,), "None", "POST", "same-origin", "https://adapter.example", False),
+    ((APP,), "None", "POST", "same-origin", None, False),
+    ((APP,), "None", "POST", "none", None, False),
+    ((APP,), "None", "POST", "cross-site", APP, True),
+    ((APP,), "None", "POST", "Cross-Site", APP, True),
+    ((APP,), "None", "GET", "same-site", "https://evil.example", False),
+    # Neither normalizes to an origin, so neither can match the other.
+    (("not an origin",), "None", "POST", "same-site", "also not one", True),
+    (("not an origin",), "None", "POST", None, "also not one", True),
+    (("",), "None", "POST", "same-site", "", True),
+]
+
+
+OriginCase = tuple[tuple[str, ...], str, str, str | None, str | None, bool]
+
+
+@pytest.mark.parametrize("case", ORIGIN_CASES)
+def test_the_origin_check(api: FakeApi, case: OriginCase) -> None:
+    allowlist, same_site, method, site, origin, blocked = case
+    api.on(method, "/login", respond(400, {"error": "invalid_request"}))
+    adapter = api.adapter(allowed_origins=allowlist, same_site=same_site)
+    headers = {k: v for k, v in (("sec-fetch-site", site), ("origin", origin)) if v is not None}
+
+    r = call(adapter, method, "/login", body="{}" if method == "POST" else None, headers=headers)
+
+    if blocked:
+        assert (r.status, r.body) == (403, {"error": "cross_site_request_blocked"})
+        assert not api.calls_to(method, "/login")
+    else:
+        assert r.status != 403, r.text
+
+
 def test_the_live_manifest_adds_routes(api: FakeApi) -> None:
     api.manifest = {
         "schemaVersion": 1,

tests/test_django.py

@@ -12,6 +12,7 @@
     FakeApi,
     access,
     api_token,
+    hs256_access_token,
     pre_auth,
     respond,
     session,
@@ -133,6 +134,13 @@ def test_the_guard(client: Client) -> None:
     r = client.post("/api/transfer", HTTP_COOKIE=session_cookie, HTTP_SEC_FETCH_SITE="same-origin")
     assert r.status_code == 200
 
+    # A valid cookie signature does not make the token inside it the API's.
+    forged = signed_cookie("seamless-access", {"sub": "u1", "token": hs256_access_token()})
+    assert client.get("/api/me", HTTP_COOKIE=forged).status_code == 401
+    other = signed_cookie("seamless-access", {"sub": "u1", "token": api_token("u2", "access")})
+    assert client.get("/api/me", HTTP_COOKIE=other).status_code == 401
+    assert client.get("/api/me-async", HTTP_COOKIE=other).status_code == 401
+
 
 def test_the_guard_refuses_a_recovery_session_unless_allowed(client: Client) -> None:
     recovery = f"Bearer {api_token('u1', 'access', recovery=True)}"

tests/test_fastapi.py

@@ -1,7 +1,17 @@
 from __future__ import annotations
 
 import httpx2
-from conftest import API_URL, FakeApi, access, api_token, pre_auth, respond, session, signed_cookie
+from conftest import (
+    API_URL,
+    FakeApi,
+    access,
+    api_token,
+    hs256_access_token,
+    pre_auth,
+    respond,
+    session,
+    signed_cookie,
+)
 from fastapi import Depends, FastAPI
 from fastapi.testclient import TestClient
 
@@ -126,6 +136,12 @@ def test_the_guard(api: FakeApi) -> None:
     )
     assert r.status_code == 200
 
+    # A valid cookie signature does not make the token inside it the API's.
+    forged = signed_cookie("seamless-access", {"sub": "u1", "token": hs256_access_token()})
+    assert client.get("/api/me", headers={"cookie": forged}).status_code == 401
+    other = signed_cookie("seamless-access", {"sub": "u1", "token": api_token("u2", "access")})
+    assert client.get("/api/me", headers={"cookie": other}).status_code == 401
+
 
 def test_the_guard_refuses_a_recovery_session_unless_allowed(api: FakeApi) -> None:
     client = app_for(api.adapter())