Merge pull request #31 from fells-code/feat/first-party-role-guard

feat: first-party role guard, optional user and organization on User

Author: Bccorb

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
 

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",

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,33 +855,54 @@ 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.
         The cookie wins when present. It does not refresh: silent refresh belongs
         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,
         method: str,
         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,
     )
 

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")

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]