diff --git a/README.md b/README.md index ab7cf31..c10c76d 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,24 @@ The auth server reads the service token from `API_SERVICE_TOKEN` and your app re `RequireUser` answers 401 unless the request carries a session, and returns the `User`. It answers 403 for a recovery session (see [Recovery sessions](#recovery-sessions)) unless you pass -`RequireUser(auth, allow_recovery=True)`. +`RequireUser(auth, allow_recovery=True)`. `RequireUser(auth, roles=["admin"])` also requires a role +(see [Roles](#roles)). + +`OptionalUser(auth)` is for routes that serve signed-in and anonymous visitors alike. It returns +the `User` exactly when `RequireUser(auth)` would admit the request, and `None` otherwise (no +session, a recovery session, or a cross-site state change on the cookie). It never answers an +error. + +```python +from seamless_auth.fastapi import OptionalUser + +maybe_user = OptionalUser(auth) + + +@app.get("/") +def home(user: User | None = Depends(maybe_user)): + return {"signed_in": user is not None} +``` ## Django @@ -93,7 +110,9 @@ def me(request): ``` `@require_auth(allow_recovery=True)` admits a recovery session too (see -[Recovery sessions](#recovery-sessions)). +[Recovery sessions](#recovery-sessions)), and `@require_auth(roles=["admin"])` also requires a role +(see [Roles](#roles)). `user_of(request)` reads the `User` the guard put on the request; call it +only under `require_auth`, as it raises `AttributeError` on an unguarded request. `SEAMLESS_AUTH` takes the keyword arguments of `Adapter` (below). `deliver` and `resolve_client_ip` may be dotted paths to callables. `require_auth` works on sync and async @@ -126,7 +145,38 @@ while cookies are `SameSite=None` and only for methods other than `GET`, `HEAD` - 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. +A bearer token is never attached by a browser, so it is not checked. For the same reason the auth +routes skip the check on a bearer-transport request (`x-seamless-auth-transport: bearer`), which +reads no cookies. + +`User.organization_id` is the active organization, from the `org_id` claim of the verified access +token, or `None` when the session has none. Like `User.roles`, it never comes from the cookie. + +### Roles + +`roles` on the guard (`RequireUser(auth, roles=[...])`, `@require_auth(roles=[...])`) admits a +session when any one of the user's roles grants any one listed role, by `role_grants_access` (the +port of `roleGrantsAccess` in `@seamless-auth/types`): `org:*` grants `org` and `org:billing`, an +unscoped `org` grants `org:billing`, and `projects:write` grants `projects:read`. The roles come +from the verified access token. A session granted none answers 403 +`{ "error": "insufficient_role" }` (FastAPI: `{ "detail": "insufficient_role" }`), after the +session, recovery and cross-site checks, and the body never names the roles. An empty `roles` +list raises `ValueError` when the guard is built. + +```python +@app.get("/api/admin") +def admin(user: User = Depends(RequireUser(auth, roles=["admin"]))): + return {"id": user.id} +``` + +```python +@require_auth(roles=["admin", "billing:write"]) +def billing(request): + return JsonResponse({"org": user_of(request).organization_id}) +``` + +`Adapter.authenticate_request(method, headers, roles=[...])` and `Adapter.authenticate(headers, +roles=[...])` raise `InsufficientRole`; `has_any_role(user, roles)` is the check on its own. ### Recovery sessions diff --git a/src/seamless_auth/__init__.py b/src/seamless_auth/__init__.py index 7e5daa3..6021556 100644 --- a/src/seamless_auth/__init__.py +++ b/src/seamless_auth/__init__.py @@ -13,7 +13,15 @@ :mod:`seamless_auth.django`. """ -from ._adapter import Adapter, CrossSiteRequest, EnrollmentRequired, Unauthenticated, User +from ._adapter import ( + Adapter, + CrossSiteRequest, + EnrollmentRequired, + InsufficientRole, + Unauthenticated, + User, + has_any_role, +) from ._client_ip import TrustedProxies from ._delivery import Delivery from ._http import AuthRequest, AuthResponse, Headers, ResponseCookie @@ -44,6 +52,7 @@ "EnrollmentRequired", "Guard", "Headers", + "InsufficientRole", "Manifest", "ManifestError", "ManifestRoute", @@ -53,6 +62,7 @@ "Unauthenticated", "User", "VerifiedUser", + "has_any_role", "require_role", "require_scope", "require_step_up", diff --git a/src/seamless_auth/_adapter.py b/src/seamless_auth/_adapter.py index 6ad36fb..6ff8120 100644 --- a/src/seamless_auth/_adapter.py +++ b/src/seamless_auth/_adapter.py @@ -35,6 +35,7 @@ from ._jwt import Claims, sign_hs256, str_claim, unverified_claims from ._manifest import ManifestRoute, ManifestSource, RouteMatch from ._refresh import Refresher, RefreshOutcome +from ._resource_server import as_names, role_grants_access if TYPE_CHECKING: from ._console import Console @@ -96,6 +97,8 @@ class User: #: A recovery session: signed in with a recovery code or self-service recovery, #: and not yet enrolled in a new passkey or authenticator app. recovery: bool = False + #: The active organization, from the verified access token's ``org_id`` claim. + organization_id: str | None = None #: The auth API access token, for calling the API on the user's behalf. token: str = field(default="", repr=False) @@ -113,6 +116,23 @@ class CrossSiteRequest(Exception): """A state-changing request from another site, carrying a session cookie.""" +class InsufficientRole(Exception): + """A session granted none of the roles a guard requires. Answer it 403 + ``insufficient_role``, without naming the roles.""" + + +def required_roles(roles: Iterable[str] | None) -> tuple[str, ...] | None: + """Validates a guard's ``roles`` at setup: ``None`` means no role check, and an + empty list is refused because it could only ever deny or be a mistake.""" + return None if roles is None else as_names("roles", roles) + + +def has_any_role(user: User, roles: Iterable[str]) -> bool: + """Whether any of the user's roles grants any listed role, by + :func:`~seamless_auth.role_grants_access`.""" + return any(role_grants_access(g, r) for r in roles for g in user.roles) + + @dataclass class Config: auth_server_url: str @@ -306,11 +326,14 @@ def handle(self, request: AuthRequest) -> AuthResponse: return _json_response(500, {"error": "internal_error"}) def _handle(self, request: AuthRequest) -> AuthResponse: - blocked = self._check_origin(request.method, request.headers) - if blocked is not None: - return self._write(blocked, bearer=False) - bearer = (request.headers.get(TRANSPORT_HEADER) or "").lower() == "bearer" + # Bearer transport reads no cookies, so a cross-site request has nothing of the + # user's to ride on. A custom header also needs a CORS preflight to be sent. + if not bearer: + blocked = self._check_origin(request.method, request.headers) + if blocked is not None: + return self._write(blocked, bearer=False) + forwarded = self._forwarded(request) method = request.method.upper() # Matched on segments, as manifest routes are, so /logout/ and //logout are @@ -832,7 +855,11 @@ def _logout_route( # The guard. def authenticate( - self, headers: Headers | Iterable[tuple[str, str]], *, allow_recovery: bool = False + self, + headers: Headers | Iterable[tuple[str, str]], + *, + allow_recovery: bool = False, + roles: Iterable[str] | None = None, ) -> User: """Verifies a request's session: the adapter's access cookie, or an auth API access token in ``Authorization: Bearer`` for clients with no cookie jar. @@ -840,8 +867,12 @@ def authenticate( to the auth routes, and a bearer client refreshes through ``POST /auth/refresh`` itself. Raises :class:`Unauthenticated`, or its subclass :class:`EnrollmentRequired` for a recovery session unless - ``allow_recovery`` is set.""" - return self._authenticate(_as_headers(headers), allow_recovery)[0] + ``allow_recovery`` is set. With ``roles``, raises :class:`InsufficientRole` + unless the session is granted any one of them.""" + required = required_roles(roles) + user = self._authenticate(_as_headers(headers), allow_recovery)[0] + _check_roles(user, required) + return user def authenticate_request( self, @@ -849,16 +880,29 @@ def authenticate_request( headers: Headers | Iterable[tuple[str, str]], *, allow_recovery: bool = False, + roles: Iterable[str] | None = None, ) -> User: """:meth:`authenticate` for a guarded route. A session from the cookie also gets the auth routes' cross-site check, or a form on another site could act as the user: raises :class:`CrossSiteRequest` for one.""" + required = required_roles(roles) headers = _as_headers(headers) user, by_cookie = self._authenticate(headers, allow_recovery) if by_cookie and self._check_origin(method, headers) is not None: raise CrossSiteRequest + _check_roles(user, required) return user + def optional_user( + self, method: str, headers: Headers | Iterable[tuple[str, str]] + ) -> User | None: + """The user exactly when :meth:`authenticate_request` with its defaults would + return one, otherwise ``None``. A recovery session yields ``None``.""" + try: + return self.authenticate_request(method, headers) + except (Unauthenticated, CrossSiteRequest): + return None + def _authenticate(self, headers: Headers, allow_recovery: bool) -> tuple[User, bool]: user, by_cookie = self._session_of(headers) if user.recovery and not allow_recovery: @@ -885,6 +929,7 @@ def _session_of(self, headers: Headers) -> tuple[User, bool]: email=str_claim(cookie, "email") or None, phone=str_claim(cookie, "phone") or None, recovery=claims.get("recovery") is True, + organization_id=str_claim(claims, "org_id") or None, token=token, ), True, @@ -952,6 +997,11 @@ def _is_allowed_origin(self, origin: str) -> bool: _NO_REFRESH_COOKIE = "no refresh cookie" +def _check_roles(user: User, required: tuple[str, ...] | None) -> None: + if required is not None and not has_any_role(user, required): + raise InsufficientRole + + def _as_headers(headers: Headers | Iterable[tuple[str, str]]) -> Headers: return headers if isinstance(headers, Headers) else Headers(headers) @@ -1078,6 +1128,7 @@ def _user_from(claims: Claims, token: str, recovery: bool) -> User: email=str_claim(claims, "email") or None, phone=str_claim(claims, "phone") or None, recovery=recovery, + organization_id=str_claim(claims, "org_id") or None, token=token, ) diff --git a/src/seamless_auth/_resource_server.py b/src/seamless_auth/_resource_server.py index 5b27d57..9918df2 100644 --- a/src/seamless_auth/_resource_server.py +++ b/src/seamless_auth/_resource_server.py @@ -453,6 +453,13 @@ def role_grants_access(granted_role: str, required_role: str) -> bool: ) +def as_names(what: str, values: Iterable[str]) -> tuple[str, ...]: + """``values`` as a non-empty tuple of non-empty strings, or a setup error.""" + names = _as_tuple(values) + _check_names(what, names) + return names + + def _as_tuple(values: Iterable[str]) -> tuple[str, ...]: if isinstance(values, str): raise TypeError("seamless-auth: pass a list of values, not a string") diff --git a/src/seamless_auth/django.py b/src/seamless_auth/django.py index 1c6a98a..f9e9be5 100644 --- a/src/seamless_auth/django.py +++ b/src/seamless_auth/django.py @@ -14,12 +14,16 @@ urlpatterns = [path("auth/", include("seamless_auth.django"))] # views.py - from seamless_auth.django import require_auth + from seamless_auth.django import require_auth, user_of @require_auth def me(request): return JsonResponse({"id": request.seamless_user.id}) + @require_auth(roles=["admin"]) + def admin(request): + return JsonResponse({"id": user_of(request).id}) + ``SEAMLESS_AUTH`` takes the keyword arguments of :class:`seamless_auth.Adapter`. ``deliver`` and ``resolve_client_ip`` may be dotted paths to callables. @@ -48,7 +52,7 @@ def orders(request): import functools import inspect import threading -from collections.abc import Callable +from collections.abc import Callable, Iterable from typing import Any, Literal, TypeVar, cast, overload from urllib.parse import quote, unquote @@ -65,8 +69,10 @@ def orders(request): Adapter, CrossSiteRequest, EnrollmentRequired, + InsufficientRole, Unauthenticated, User, + required_roles, ) from ._http import AuthRequest, AuthResponse, Headers from ._resource_server import ( @@ -94,6 +100,7 @@ def orders(request): "require_scope", "require_step_up", "urlpatterns", + "user_of", ] _PATH_SAFE = "/:@!$&'()*+,;=-._~" @@ -182,27 +189,42 @@ def console_view(request: HttpRequest, path: str = "") -> HttpResponse: @overload def require_auth(view: F) -> F: ... @overload -def require_auth(*, allow_recovery: bool = False) -> Callable[[F], F]: ... +def require_auth( + *, allow_recovery: bool = False, roles: Iterable[str] | None = None +) -> Callable[[F], F]: ... -def require_auth(view: F | None = None, *, allow_recovery: bool = False) -> Any: +def require_auth( + view: F | None = None, + *, + allow_recovery: bool = False, + roles: Iterable[str] | None = None, +) -> Any: """Answers 401 unless the request carries a valid session, and puts the :class:`~seamless_auth.User` on ``request.seamless_user``. A session from the cookie also gets the auth routes' cross-site check (403 on a state change from another site). A recovery session (a user who still has to enroll a new passkey or authenticator app) answers 403 ``enrollment_required`` unless - ``@require_auth(allow_recovery=True)``. Works on sync and async views.""" + ``@require_auth(allow_recovery=True)``. With ``roles``, a session granted none + of them (by :func:`~seamless_auth.role_grants_access`) answers 403 + ``insufficient_role``. Works on sync and async views.""" + required = required_roles(roles) if view is None: - return lambda inner: _guard(inner, allow_recovery) - return _guard(view, allow_recovery) + return lambda inner: _guard(inner, allow_recovery, required) + return _guard(view, allow_recovery, required) -def _guard(view: F, allow_recovery: bool) -> F: +def _guard(view: F, allow_recovery: bool, roles: tuple[str, ...] | None) -> F: def check(request: HttpRequest) -> HttpResponse | None: try: user = get_adapter().authenticate_request( - request.method or "GET", _headers(request), allow_recovery=allow_recovery + request.method or "GET", + _headers(request), + allow_recovery=allow_recovery, + roles=roles, ) + except InsufficientRole: + return JsonResponse({"error": "insufficient_role"}, status=403) except EnrollmentRequired: return JsonResponse({"error": "enrollment_required"}, status=403) except CrossSiteRequest: @@ -326,7 +348,9 @@ def _bearer_response(err: BearerError) -> JsonResponse: def user_of(request: HttpRequest) -> User: - """The user :func:`require_auth` put on the request.""" + """The user :func:`require_auth` put on the request. Call it only under that + guard: it is not an optional-user lookup and raises ``AttributeError`` on an + unguarded request.""" return cast(User, request.seamless_user) # type: ignore[attr-defined] diff --git a/src/seamless_auth/fastapi.py b/src/seamless_auth/fastapi.py index 91e9b75..520e569 100644 --- a/src/seamless_auth/fastapi.py +++ b/src/seamless_auth/fastapi.py @@ -16,6 +16,10 @@ def me(user: User = Depends(current_user)): return {"id": user.id} + @app.get("/api/admin") + def admin(user: User = Depends(RequireUser(auth, roles=["admin"]))): + return {"id": user.id} + Verify-only, for an API that only receives OpenID Provider access tokens:: from seamless_auth import ResourceServer, VerifiedUser, require_scope @@ -32,6 +36,7 @@ def orders(user: VerifiedUser = Depends(RequireBearer(rs, require_scope("orders: from __future__ import annotations +from collections.abc import Iterable from urllib.parse import quote from fastapi import APIRouter, FastAPI, HTTPException, Request @@ -44,14 +49,17 @@ def orders(user: VerifiedUser = Depends(RequireBearer(rs, require_scope("orders: Adapter, CrossSiteRequest, EnrollmentRequired, + InsufficientRole, Unauthenticated, User, + required_roles, ) from ._http import AuthRequest, AuthResponse, Headers from ._resource_server import BearerError, BearerRequest, Guard, ResourceServer, VerifiedUser __all__ = [ "BearerHTTPException", + "OptionalUser", "RequireBearer", "RequireUser", "add_bearer_error_handler", @@ -101,21 +109,34 @@ class RequireUser: A recovery session (a user who still has to enroll a new passkey or authenticator app) answers 403 ``enrollment_required`` unless - ``allow_recovery`` is set. + ``allow_recovery`` is set. With ``roles``, a session granted none of them (by + :func:`~seamless_auth.role_grants_access`) answers 403 ``insufficient_role``. A plain function, so FastAPI runs it on its threadpool: the first sight of a key may fetch the auth API's key set. """ - def __init__(self, adapter: Adapter, *, allow_recovery: bool = False) -> None: + def __init__( + self, + adapter: Adapter, + *, + allow_recovery: bool = False, + roles: Iterable[str] | None = None, + ) -> None: self._adapter = adapter self._allow_recovery = allow_recovery + self._roles = required_roles(roles) def __call__(self, request: Request) -> User: try: return self._adapter.authenticate_request( - request.method, _headers(request), allow_recovery=self._allow_recovery + request.method, + _headers(request), + allow_recovery=self._allow_recovery, + roles=self._roles, ) + except InsufficientRole: + raise HTTPException(status_code=403, detail="insufficient_role") from None except EnrollmentRequired: raise HTTPException(status_code=403, detail="enrollment_required") from None except CrossSiteRequest: @@ -124,6 +145,19 @@ def __call__(self, request: Request) -> User: raise HTTPException(status_code=401, detail="unauthenticated") from None +class OptionalUser: + """A dependency that returns the :class:`~seamless_auth.User` exactly when + :class:`RequireUser` with its defaults would, and ``None`` otherwise (no session, + a recovery session, or a cross-site state change). It never answers an error. + """ + + def __init__(self, adapter: Adapter) -> None: + self._adapter = adapter + + def __call__(self, request: Request) -> User | None: + return self._adapter.optional_user(request.method, _headers(request)) + + class BearerHTTPException(HTTPException): """A verify-only refusal. :func:`bearer_exception_handler` answers it with the exact JSON body; without the handler FastAPI still sends its status and diff --git a/tests/test_adapter.py b/tests/test_adapter.py index 36a00f0..abfe8b8 100644 --- a/tests/test_adapter.py +++ b/tests/test_adapter.py @@ -531,6 +531,22 @@ def test_cross_site_state_changes_are_blocked(api: FakeApi) -> None: assert r.status == 400 +def test_bearer_transport_skips_the_origin_check(api: FakeApi) -> None: + api.on("POST", "/login", respond(400, {"error": "invalid_request"})) + adapter = api.adapter(allowed_origins=["https://app.example"]) + for headers in ({"sec-fetch-site": "cross-site"}, {"origin": "null"}): + r = call( + adapter, + "POST", + "/login", + body="{}", + headers={"x-seamless-auth-transport": "bearer", **headers}, + ) + assert r.status == 400, r.text + r = call(adapter, "POST", "/login", body="{}", headers=headers) + assert (r.status, r.body) == (403, {"error": "cross_site_request_blocked"}) + + APP = "https://app.example" ORIGIN_CASES = [ # (allowlist, same_site, method, sec-fetch-site, origin, blocked) diff --git a/tests/test_django.py b/tests/test_django.py index 0ae9723..af5d20d 100644 --- a/tests/test_django.py +++ b/tests/test_django.py @@ -224,3 +224,56 @@ def test_the_django_console(api: FakeApi, client: Client) -> None: assert b"".join(r.streaming_content) == b"js" # type: ignore[attr-defined] assert client.get("/console/%2e%2e/jwks.json").status_code == 400 assert client.post("/console/x").status_code == 405 + + +def bearer_auth(**claims: Any) -> str: + return f"Bearer {api_token('u1', 'access', **claims)}" + + +def test_the_role_guard(client: Client) -> None: + for view in ("/api/admin", "/api/admin-async"): + r = client.get(view) + assert (r.status_code, r.json()) == (401, {"error": "unauthenticated"}), view + r = client.get(view, HTTP_AUTHORIZATION=bearer_auth(roles=["user"])) + assert (r.status_code, r.json()) == (403, {"error": "insufficient_role"}), view + r = client.get(view, HTTP_AUTHORIZATION=bearer_auth(roles=["admin"], recovery=True)) + assert (r.status_code, r.json()) == (403, {"error": "enrollment_required"}), view + r = client.get(view, HTTP_AUTHORIZATION=bearer_auth(roles=["admin"])) + assert r.status_code == 200, view + + for roles in (["billing:*"], ["billing"]): + r = client.get("/api/admin", HTTP_AUTHORIZATION=bearer_auth(roles=roles, org_id="org-1")) + assert (r.status_code, r.json()) == (200, {"id": "u1", "org": "org-1"}), roles + + forged_roles = signed_cookie( + "seamless-access", {"sub": "u1", "token": api_token("u1", "access"), "roles": ["admin"]} + ) + r = client.get("/api/admin", HTTP_COOKIE=forged_roles) + assert (r.status_code, r.json()) == (403, {"error": "insufficient_role"}) + admin_cookie = signed_cookie( + "seamless-access", + { + "sub": "u1", + "token": api_token("u1", "access", roles=["admin"]), + "organizationId": "org-forged", + }, + ) + r = client.post("/api/admin", HTTP_COOKIE=admin_cookie, HTTP_SEC_FETCH_SITE="cross-site") + assert (r.status_code, r.json()) == (403, {"error": "cross_site_request_blocked"}) + r = client.get("/api/admin", HTTP_COOKIE=admin_cookie) + assert (r.status_code, r.json()) == (200, {"id": "u1", "org": None}) + + +def test_the_role_guard_refuses_no_roles_at_setup() -> None: + from seamless_auth.django import require_auth + + with pytest.raises(ValueError, match="at least one"): + require_auth(roles=[]) + with pytest.raises(TypeError): + require_auth(roles="admin") + + +def test_user_of_is_exported() -> None: + import seamless_auth.django as binding + + assert "user_of" in binding.__all__ diff --git a/tests/test_django_urls.py b/tests/test_django_urls.py index e086563..ffecd06 100644 --- a/tests/test_django_urls.py +++ b/tests/test_django_urls.py @@ -33,12 +33,25 @@ def transfer(request: HttpRequest) -> JsonResponse: return JsonResponse({"id": user_of(request).id}) +@csrf_exempt +@require_auth(roles=["admin", "billing:write"]) +def admin(request: HttpRequest) -> JsonResponse: + return JsonResponse({"id": user_of(request).id, "org": user_of(request).organization_id}) + + +@require_auth(roles=["admin"]) +async def admin_async(request: HttpRequest) -> JsonResponse: + return JsonResponse({"id": user_of(request).id}) + + urlpatterns = [ path("api/me", me), path("api/me-async", me_async), path("api/enroll", enroll), path("api/enroll-async", enroll_async), path("api/transfer", transfer), + path("api/admin", admin), + path("api/admin-async", admin_async), path("auth/", include("seamless_auth.django")), path("console/", include(console_urlpatterns)), ] diff --git a/tests/test_fastapi.py b/tests/test_fastapi.py index d772383..0196ec3 100644 --- a/tests/test_fastapi.py +++ b/tests/test_fastapi.py @@ -1,6 +1,7 @@ from __future__ import annotations import httpx2 +import pytest from conftest import ( API_URL, FakeApi, @@ -16,7 +17,7 @@ from fastapi.testclient import TestClient from seamless_auth import Adapter, Delivery, User -from seamless_auth.fastapi import RequireUser, auth_router +from seamless_auth.fastapi import OptionalUser, RequireUser, auth_router def app_for(adapter: Adapter) -> TestClient: @@ -38,6 +39,18 @@ def enroll( def transfer(user: User = Depends(current_user)) -> dict[str, str]: # noqa: B008 return {"id": user.id} + @app.api_route("/api/admin", methods=["GET", "POST"]) + def admin( + user: User = Depends(RequireUser(adapter, roles=["admin", "billing:write"])), # noqa: B008 + ) -> dict[str, str]: + return {"id": user.id} + + @app.api_route("/api/maybe", methods=["GET", "POST"]) + def maybe( + user: User | None = Depends(OptionalUser(adapter)), # noqa: B008 + ) -> dict[str, object]: + return {"user": None if user is None else {"id": user.id, "org": user.organization_id}} + return TestClient(app, base_url="https://testserver") @@ -177,3 +190,56 @@ async def deliver(delivery: Delivery) -> None: assert r.json() == {"message": "sent"} assert got[0].to == "a@b.test" assert API_URL + + +def bearer_header(**claims: object) -> dict[str, str]: + return {"authorization": f"Bearer {api_token('u1', 'access', **claims)}"} + + +def test_the_role_guard(api: FakeApi) -> None: + client = app_for(api.adapter()) + + r = client.get("/api/admin") + assert (r.status_code, r.json()) == (401, {"detail": "unauthenticated"}) + r = client.get("/api/admin", headers=bearer_header(roles=["user"])) + assert (r.status_code, r.json()) == (403, {"detail": "insufficient_role"}) + r = client.get( + "/api/admin", headers=bearer_header(roles=["admin", "secret-role"], recovery=True) + ) + assert (r.status_code, r.json()) == (403, {"detail": "enrollment_required"}) + for roles in (["admin"], ["billing:*"], ["billing"]): + r = client.get("/api/admin", headers=bearer_header(roles=roles)) + assert (r.status_code, r.json()) == (200, {"id": "u1"}), roles + + cookie = signed_cookie( + "seamless-access", {"sub": "u1", "token": api_token("u1", "access"), "roles": ["admin"]} + ) + r = client.get("/api/admin", headers={"cookie": cookie}) + assert (r.status_code, r.json()) == (403, {"detail": "insufficient_role"}) + admin_cookie = signed_cookie( + "seamless-access", {"sub": "u1", "token": api_token("u1", "access", roles=["admin"])} + ) + r = client.post("/api/admin", headers={"cookie": admin_cookie, "sec-fetch-site": "cross-site"}) + assert (r.status_code, r.json()) == (403, {"detail": "cross_site_request_blocked"}) + + +def test_the_role_guard_refuses_no_roles_at_setup(api: FakeApi) -> None: + with pytest.raises(ValueError, match="at least one"): + RequireUser(api.adapter(), roles=[]) + with pytest.raises(TypeError): + RequireUser(api.adapter(), roles="admin") + + +def test_the_optional_user(api: FakeApi) -> None: + client = app_for(api.adapter()) + + assert client.get("/api/maybe").json() == {"user": None} + r = client.get("/api/maybe", headers=bearer_header(org_id="org-1")) + assert (r.status_code, r.json()) == (200, {"user": {"id": "u1", "org": "org-1"}}) + r = client.get("/api/maybe", headers=bearer_header(recovery=True)) + assert (r.status_code, r.json()) == (200, {"user": None}) + cookie = signed_cookie("seamless-access", {"sub": "u1", "token": api_token("u1", "access")}) + r = client.post("/api/maybe", headers={"cookie": cookie, "sec-fetch-site": "cross-site"}) + assert (r.status_code, r.json()) == (200, {"user": None}) + forged = signed_cookie("seamless-access", {"sub": "u1", "token": hs256_access_token()}) + assert client.get("/api/maybe", headers={"cookie": forged}).json() == {"user": None} diff --git a/tests/test_guard.py b/tests/test_guard.py index 7df63d9..ba1ded9 100644 --- a/tests/test_guard.py +++ b/tests/test_guard.py @@ -17,7 +17,14 @@ ) from cryptography.hazmat.primitives.asymmetric import rsa -from seamless_auth import CrossSiteRequest, Unauthenticated +from seamless_auth import ( + CrossSiteRequest, + EnrollmentRequired, + InsufficientRole, + Unauthenticated, + User, + has_any_role, +) def bearer(token: str) -> tuple[str, str]: @@ -359,3 +366,90 @@ def post(*headers: tuple[str, str]) -> str: adapter.authenticate_request("POST", [bearer(ACCESS), ("sec-fetch-site", "same-site")]).id == "u1" ) + + +ROLE_CASES = [ + # (granted, required, ok) + (["admin"], ["admin"], True), + (["user"], ["admin"], False), + (["user"], ["admin", "user"], True), + ([], ["admin"], False), + (["org:*"], ["org:billing"], True), + (["org"], ["org:billing"], True), + (["projects:write"], ["projects:read"], True), + (["projects:read"], ["projects:write"], False), + (["org:billing"], ["org"], False), +] + + +@pytest.mark.parametrize("case", ROLE_CASES) +def test_a_role_guard_admits_any_granted_role( + api: FakeApi, case: tuple[list[str], list[str], bool] +) -> None: + granted, required, ok = case + adapter = api.adapter() + token = api_token("u1", "access", roles=granted) + for headers in ([bearer(token)], [session_cookie(token)]): + if ok: + assert adapter.authenticate_request("GET", headers, roles=required).id == "u1" + assert adapter.authenticate(headers, roles=required).id == "u1" + else: + with pytest.raises(InsufficientRole): + adapter.authenticate_request("GET", headers, roles=required) + with pytest.raises(InsufficientRole): + adapter.authenticate(headers, roles=required) + assert has_any_role(User(id="u1", roles=tuple(granted)), required) is ok + + +def test_a_role_guard_takes_roles_from_the_verified_token_not_the_cookie(api: FakeApi) -> None: + token = api_token("u1", "access", roles=["user"]) + with pytest.raises(InsufficientRole): + api.adapter().authenticate_request( + "GET", [session_cookie(token, roles=["admin"])], roles=["admin"] + ) + + +def test_a_role_guard_checks_the_session_and_origin_first(api: FakeApi) -> None: + adapter = api.adapter() + with pytest.raises(Unauthenticated): + adapter.authenticate_request("GET", [], roles=["admin"]) + with pytest.raises(EnrollmentRequired): + adapter.authenticate_request( + "GET", [bearer(api_token("u1", "access", recovery=True))], roles=["admin"] + ) + with pytest.raises(CrossSiteRequest): + adapter.authenticate_request( + "POST", + [session_cookie(ACCESS), ("sec-fetch-site", "cross-site")], + roles=["admin"], + ) + + +@pytest.mark.parametrize("roles", [[], (), [""], "admin"]) +def test_a_role_guard_refuses_unusable_roles(api: FakeApi, roles: Any) -> None: + with pytest.raises((ValueError, TypeError)): + api.adapter().authenticate_request("GET", [bearer(ACCESS)], roles=roles) + + +def test_the_organization_comes_from_the_verified_token(api: FakeApi) -> None: + adapter = api.adapter() + token = api_token("u1", "access", org_id="org-1") + assert adapter.authenticate([bearer(token)]).organization_id == "org-1" + cookie_user = adapter.authenticate([session_cookie(token, organizationId="org-forged")]) + assert cookie_user.organization_id == "org-1" + unscoped = adapter.authenticate([session_cookie(ACCESS, organizationId="org-forged")]) + assert unscoped.organization_id is None + assert ( + adapter.authenticate([bearer(api_token("u1", "access", org_id=7))]).organization_id is None + ) + + +def test_the_optional_user(api: FakeApi) -> None: + adapter = api.adapter() + assert adapter.optional_user("GET", []) is None + assert adapter.optional_user("GET", [bearer(ACCESS)]) == adapter.authenticate([bearer(ACCESS)]) + assert adapter.optional_user("GET", [bearer(api_token("u1", "ephemeral"))]) is None + assert adapter.optional_user("GET", [bearer(api_token("u1", "access", recovery=True))]) is None + cross_site = [session_cookie(ACCESS), ("sec-fetch-site", "cross-site")] + assert adapter.optional_user("POST", cross_site) is None + assert adapter.optional_user("GET", cross_site) is not None