From 1b135ae65cece1402f3e088a8d62fbff8b4458c2 Mon Sep 17 00:00:00 2001 From: Brandon Corbett Date: Fri, 6 Nov 2026 00:13:31 -0500 Subject: [PATCH] feat(guard): add an organization-aware permission guard RequireUser and require_auth take permissions, permission_mode, organization and organization_param. Matching is exact. A token with permissions_omitted is checked against a live read of /users/me/permissions, which fails closed with 503 permissions_unavailable. require_permission is the verify-only form, and denies provider tokens until seamless-auth-oidc-provider#47. User and VerifiedUser gain permissions, organization_roles, organization_permissions and permissions_omitted. The reference apps gain the four routes of the pending permission-guards contract. Closes #32 --- AGENTS.md | 2 + README.md | 84 +++ conformance/django_app.py | 24 + conformance/fastapi_app.py | 28 + src/seamless_auth/__init__.py | 9 + src/seamless_auth/_adapter.py | 108 ++- src/seamless_auth/_permissions.py | 95 +++ src/seamless_auth/_resource_server.py | 45 +- src/seamless_auth/django.py | 85 ++- src/seamless_auth/fastapi.py | 69 +- tests/test_permissions.py | 944 ++++++++++++++++++++++++++ tests/test_permissions_django_urls.py | 84 +++ 12 files changed, 1564 insertions(+), 13 deletions(-) create mode 100644 src/seamless_auth/_permissions.py create mode 100644 tests/test_permissions.py create mode 100644 tests/test_permissions_django_urls.py diff --git a/AGENTS.md b/AGENTS.md index 5112c35..c3f9cd8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -66,6 +66,8 @@ current Django, so both are covered. Do not use language or standard library fea - `src/seamless_auth/_adapter.py`: `Adapter`, the request pipeline, the upstream call, manifest routes, credential resolution (with silent refresh), session verification, refresh, logout, the guard, the cross-site check. +- `src/seamless_auth/_permissions.py`: the permission rule, shared by the session guards and the + verify-only guard. - `src/seamless_auth/_console.py`: the admin dashboard proxy (`Adapter.console`). - `src/seamless_auth/_refresh.py`: refresh sharing across threads (one rotation per refresh token while it runs, and for cookie transport only, its result for 5 seconds after). diff --git a/README.md b/README.md index b62202b..915380c 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,8 @@ bearer`) included, as the other adapters do. A native client sends neither `Sec- `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. +`User.permissions`, `User.organization_roles`, `User.organization_permissions` and +`User.permissions_omitted` come from that token in the same way (see [Permissions](#permissions)). ### Roles @@ -185,6 +187,73 @@ def billing(request): `Adapter.authenticate_request(method, headers, roles=[...])` and `Adapter.authenticate(headers, roles=[...])` raise `InsufficientRole`; `has_any_role(user, roles)` is the check on its own. +### Permissions + +`permissions` on the guard (`RequireUser(auth, permissions=[...])`, +`@require_auth(permissions=[...])`) admits a session that holds the listed permissions. It needs +the auth API at v0.23, which issues the claims. Lead with the form bound to the route, because +without it a token for organization A passes on `/orgs/B/...`: + +```python +# FastAPI +@app.get("/orgs/{organization_id}/projects") +def projects( + user: User = Depends( + RequireUser(auth, permissions=["projects:read"], organization_param="organization_id") + ), +): + return {"id": user.id} +``` + +```python +# Django +@require_auth(permissions=["projects:read"], organization_param="organization_id") +def projects(request, organization_id): + return JsonResponse({"id": user_of(request).id}) +``` + +| Option | Effect | +| --- | --- | +| `permissions=[...]` | What the route requires. An empty list allows nobody, in any mode | +| `permission_mode="any"` | The default: at least one is held | +| `permission_mode="all"` | Every one is held, each from either claim | +| `organization=True` | Only `org_permissions` count, and the token must have a non-empty `org_id` | +| `organization_param="name"` | `organization=True`, and the route parameter `name` must equal `org_id` exactly. A missing parameter is refused | + +Without `organization` or `organization_param`, a permission is held if it is in `permissions` or +`org_permissions`. + +Matching is exact and case-sensitive, unlike roles: no trimming, no wildcards, no implied grants. +`invoices:write` does not satisfy `invoices:read`. Organization roles never count for `roles`. + +| Situation | Answer | +| --- | --- | +| Not authenticated | The guard's 401 | +| Recovery session | The guard's 403 `enrollment_required`, unless `allow_recovery` is set | +| Requirement not met | 403 `{ "error": "insufficient_permission" }`, naming no permissions | +| Token omits its permissions and the live read fails | 503 `{ "error": "permissions_unavailable" }` | +| Live read answers 401 | The guard's 401 | + +These two bodies are the same on FastAPI and Django, with no setup. A dependency can only raise, so +on FastAPI the guard registers a handler for its `PermissionHTTPException` the first time it refuses +a request. To answer differently, register your own for that class. The guard's 401 and +`enrollment_required` are the same as `RequireUser`'s, so on FastAPI they stay under `detail`. + +A large permission set does not fit in a token, so the API sends `permissions_omitted: true` in +place of `permissions`, `org_roles` and `org_permissions`. `User.permissions` is then empty, which +means unknown rather than none. The guard reads `GET {auth_server_url}/users/me/permissions` with +the caller's access token and applies the same rules to `permissions`, `organizationPermissions` +and `organizationId`, with no cache and no retry. Any other status than 2xx or 401, a timeout, or a +malformed body fails closed. Under FastAPI and async Django the guard already runs on a worker +thread, so the read does not block the event loop. + +In the permission claims, a value that is not a list reads as empty and entries that are not +strings are ignored. A `permissions` that is a bare string, or a name that is not a non-empty +string, raises when the guard is built, as do the other options without `permissions`. + +`Adapter.authenticate_request(method, headers, permissions=PermissionRule(...), route_params=...)` +and `Adapter.authenticate` raise `InsufficientPermission` or `PermissionsUnavailable`. + ### Recovery sessions A user who signs in with a recovery code, or completes self-service account recovery, gets a @@ -490,6 +559,7 @@ step-up guards run after. | --- | --- | --- | | `require_scope(*scopes)` | every scope is in `scopes`, exactly | 403 `{"error":"insufficient_scope","scope":"a b"}`, `WWW-Authenticate: Bearer error="insufficient_scope", scope="a b"` | | `require_role(*roles)` | any role is granted (`x:*` grants `x` and under, `billing` grants `billing:...`, `p:write` grants `p:read`) | 403 `{"error":"insufficient_role"}` | +| `require_permission(*permissions, mode=, organization=, organization_param=)` | the token meets the rule in [Permissions](#permissions) | 403 `{"error":"insufficient_permission"}` | | `require_step_up(acr_values=, amr=, max_age=)` | `acr` is one of `acr_values` (`phr` also accepts `phrh`), every `amr` is present, and at most `max_age` seconds since `auth_time` | 401 `insufficient_user_authentication` (RFC 9470) | A missing token, a scheme other than `Bearer` or more than one token answers 401 @@ -504,6 +574,17 @@ WWW-Authenticate: Bearer error="insufficient_user_authentication", with the same fields in the JSON body (`max_age` as a number). `role_grants_access(granted, required)` is the role matching on its own. +`require_permission` matches the session guard, but **it denies every token today**: provider tokens +do not carry `permissions`, `org_permissions` or `permissions_omitted` until +[seamless-auth-oidc-provider#47](https://github.com/fells-code/seamless-auth-oidc-provider/issues/47). +A token with `permissions_omitted` answers 403 rather than a live read, because the first-party +endpoint refuses provider tokens. An empty call allows nobody. On FastAPI its refusal is +`{"error":"insufficient_permission"}` with or without `add_bearer_error_handler`. On Django it goes +under `@require_bearer`, and `organization_param` names a URL parameter. + +`VerifiedUser` carries the same `permissions`, `organization_roles`, `organization_permissions` and +`permissions_omitted`, and keeps its `org_id`. + ### Sender-constrained tokens A token with a `cnf` claim is bound to a key the client must prove it holds (DPoP), so it is @@ -572,6 +653,9 @@ PORT=8080 AUTH_SERVER_URL=http://localhost:5312 AUTH_SERVER_ISSUER=http://auth-a SEAMLESS_API_DIR=../seamless-auth-api seamless verify --adapter-url=http://localhost:8080 ``` +The `/api/permissions/*` routes serve the pending `permission-guards` contract. It needs the auth +API at v0.23: add `SEAMLESS_VERIFY_CONTRACTS=permission-guards` to run those specs. + ## Status Pre-1.0. The public API may change between minor versions until 1.0. diff --git a/conformance/django_app.py b/conformance/django_app.py index 1893a70..eebf9fc 100644 --- a/conformance/django_app.py +++ b/conformance/django_app.py @@ -54,10 +54,34 @@ def me(request: HttpRequest) -> JsonResponse: return JsonResponse({"id": user_of(request).id}) +@require_auth(permissions=["conformance:read", "conformance:other"]) +def permissions_any(request: HttpRequest) -> JsonResponse: + return JsonResponse({"id": user_of(request).id}) + + +@require_auth(permissions=["conformance:read", "conformance:write"], permission_mode="all") +def permissions_all(request: HttpRequest) -> JsonResponse: + return JsonResponse({"id": user_of(request).id}) + + +@require_auth(permissions=["conformance:org"], organization=True) +def permissions_org(request: HttpRequest) -> JsonResponse: + return JsonResponse({"id": user_of(request).id}) + + +@require_auth(permissions=["conformance:org"], organization_param="organization_id") +def permissions_org_bound(request: HttpRequest, organization_id: str) -> JsonResponse: + return JsonResponse({"id": user_of(request).id}) + + urlpatterns = [ path("", health), path("__captured/", captured), path("api/me", me), + path("api/permissions/any", permissions_any), + path("api/permissions/all", permissions_all), + path("api/permissions/org", permissions_org), + path("api/permissions/org/", permissions_org_bound), path("auth/", include("seamless_auth.django")), ] diff --git a/conformance/fastapi_app.py b/conformance/fastapi_app.py index 573b754..fe76f47 100644 --- a/conformance/fastapi_app.py +++ b/conformance/fastapi_app.py @@ -23,6 +23,14 @@ auth = Adapter(**_shared.adapter_options()) current_user = RequireUser(auth) +any_of = RequireUser(auth, permissions=["conformance:read", "conformance:other"]) +all_of = RequireUser( + auth, permissions=["conformance:read", "conformance:write"], permission_mode="all" +) +in_organization = RequireUser(auth, permissions=["conformance:org"], organization=True) +in_route_organization = RequireUser( + auth, permissions=["conformance:org"], organization_param="organization_id" +) app = FastAPI() app.include_router(auth_router(auth)) @@ -43,6 +51,26 @@ def me(user: User = Depends(current_user)) -> dict[str, str]: # noqa: B008 return {"id": user.id} +@app.get("/api/permissions/any") +def permissions_any(user: User = Depends(any_of)) -> dict[str, str]: # noqa: B008 + return {"id": user.id} + + +@app.get("/api/permissions/all") +def permissions_all(user: User = Depends(all_of)) -> dict[str, str]: # noqa: B008 + return {"id": user.id} + + +@app.get("/api/permissions/org") +def permissions_org(user: User = Depends(in_organization)) -> dict[str, str]: # noqa: B008 + return {"id": user.id} + + +@app.get("/api/permissions/org/{organization_id}") +def permissions_org_bound(user: User = Depends(in_route_organization)) -> dict[str, str]: # noqa: B008 + return {"id": user.id} + + if __name__ == "__main__": print(f"seamless-auth FastAPI reference app listening on :{_shared.port()}") uvicorn.run(app, host="0.0.0.0", port=_shared.port(), log_level="warning") diff --git a/src/seamless_auth/__init__.py b/src/seamless_auth/__init__.py index 6021556..85667a4 100644 --- a/src/seamless_auth/__init__.py +++ b/src/seamless_auth/__init__.py @@ -17,7 +17,9 @@ Adapter, CrossSiteRequest, EnrollmentRequired, + InsufficientPermission, InsufficientRole, + PermissionsUnavailable, Unauthenticated, User, has_any_role, @@ -26,6 +28,7 @@ from ._delivery import Delivery from ._http import AuthRequest, AuthResponse, Headers, ResponseCookie from ._manifest import MANIFEST_PATH, Manifest, ManifestError, ManifestRoute +from ._permissions import PermissionMode, PermissionRule from ._resource_server import ( BearerError, BearerRequest, @@ -33,6 +36,7 @@ Guard, ResourceServer, VerifiedUser, + require_permission, require_role, require_scope, require_step_up, @@ -52,10 +56,14 @@ "EnrollmentRequired", "Guard", "Headers", + "InsufficientPermission", "InsufficientRole", "Manifest", "ManifestError", "ManifestRoute", + "PermissionMode", + "PermissionRule", + "PermissionsUnavailable", "ResourceServer", "ResponseCookie", "TrustedProxies", @@ -63,6 +71,7 @@ "User", "VerifiedUser", "has_any_role", + "require_permission", "require_role", "require_scope", "require_step_up", diff --git a/src/seamless_auth/_adapter.py b/src/seamless_auth/_adapter.py index ddc6fd0..a9922a5 100644 --- a/src/seamless_auth/_adapter.py +++ b/src/seamless_auth/_adapter.py @@ -9,7 +9,7 @@ import re import threading import time -from collections.abc import Awaitable, Callable, Iterable, Iterator +from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any from urllib.parse import urlsplit @@ -34,6 +34,7 @@ from ._jwks import BodyTooLarge, JwksCache, TokenError, read_limited from ._jwt import Claims, sign_hs256, str_claim, unverified_claims from ._manifest import ManifestRoute, ManifestSource, RouteMatch +from ._permissions import PermissionRule, permission_names from ._refresh import Refresher, RefreshOutcome from ._resource_server import as_names, role_grants_access @@ -102,6 +103,16 @@ class User: 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) + #: Instance permissions, from the verified access token's ``permissions`` claim. + permissions: tuple[str, ...] = () + #: The user's roles in the active organization. They never count as ``roles``. + organization_roles: tuple[str, ...] = () + #: The user's permissions in the active organization. + organization_permissions: tuple[str, ...] = () + #: The token left its permissions out because they would not fit. The three + #: fields above are then empty, which means unknown, not none. A permission guard + #: reads the live set from the auth API. + permissions_omitted: bool = False class Unauthenticated(Exception): @@ -122,6 +133,16 @@ class InsufficientRole(Exception): ``insufficient_role``, without naming the roles.""" +class InsufficientPermission(Exception): + """A session that does not meet a guard's permission rule. Answer it 403 + ``insufficient_permission``, without naming the permissions.""" + + +class PermissionsUnavailable(Exception): + """A token that omits its permissions, and a live read of them that failed. Answer + it 503 ``permissions_unavailable``: a guard fails closed.""" + + 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.""" @@ -907,6 +928,8 @@ def authenticate( *, allow_recovery: bool = False, roles: Iterable[str] | None = None, + permissions: PermissionRule | None = None, + route_params: Mapping[str, Any] | 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. @@ -915,10 +938,15 @@ def authenticate( ``POST /auth/refresh`` itself. Raises :class:`Unauthenticated`, or its subclass :class:`EnrollmentRequired` for a recovery session unless ``allow_recovery`` is set. With ``roles``, raises :class:`InsufficientRole` - unless the session is granted any one of them.""" + unless the session is granted any one of them. With ``permissions``, raises + :class:`InsufficientPermission` unless the session meets the rule, whose + ``organization_param`` is looked up in ``route_params``. A token that omits its + permissions is checked against a live read of them, which is blocking and + raises :class:`PermissionsUnavailable` if it fails.""" required = required_roles(roles) user = self._authenticate(_as_headers(headers), allow_recovery)[0] _check_roles(user, required) + self._check_permissions(user, permissions, route_params) return user def authenticate_request( @@ -928,6 +956,8 @@ def authenticate_request( *, allow_recovery: bool = False, roles: Iterable[str] | None = None, + permissions: PermissionRule | None = None, + route_params: Mapping[str, Any] | 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 @@ -938,6 +968,7 @@ def authenticate_request( if by_cookie and self._check_origin(method, headers) is not None: raise CrossSiteRequest _check_roles(user, required) + self._check_permissions(user, permissions, route_params) return user def optional_user( @@ -950,6 +981,51 @@ def optional_user( except (Unauthenticated, CrossSiteRequest): return None + def _check_permissions( + self, user: User, rule: PermissionRule | None, route_params: Mapping[str, Any] | None + ) -> None: + if rule is None: + return + # An empty rule can only deny, so it asks the auth API nothing. + if not rule.permissions: + raise InsufficientPermission + if user.permissions_omitted: + held = self._live_permissions(user.token) + else: + held = (user.permissions, user.organization_permissions, user.organization_id) + if not rule.allows(*held, route_params): + raise InsufficientPermission + + def _live_permissions(self, token: str) -> tuple[tuple[str, ...], tuple[str, ...], str | None]: + """The caller's permissions, organization permissions and active organization, + read from the auth API. Not cached and not retried, so a change applies here + before the token catches up.""" + try: + res = self._call( + _Forwarded(None, None), + "GET", + "/users/me/permissions", + authorization=f"Bearer {token}", + stream=True, + ) + except _UpstreamError as err: + log.warning("[seamless-auth] The live permissions read failed: %s", err) + raise PermissionsUnavailable from None + status = res.status_code + if status == 401: + res.close() + raise Unauthenticated + if not 200 <= status < 300: + res.close() + log.warning("[seamless-auth] The live permissions read answered %s", status) + raise PermissionsUnavailable + body, _ = _read_json(res) + held = _live_permission_set(body) + if held is None: + log.warning("[seamless-auth] The live permissions read answered a malformed body") + raise PermissionsUnavailable + return held + 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: @@ -978,6 +1054,10 @@ def _session_of(self, headers: Headers) -> tuple[User, bool]: recovery=claims.get("recovery") is True, organization_id=str_claim(claims, "org_id") or None, token=token, + permissions=permission_names(claims.get("permissions")), + organization_roles=permission_names(claims.get("org_roles")), + organization_permissions=permission_names(claims.get("org_permissions")), + permissions_omitted=claims.get("permissions_omitted") is True, ), True, ) @@ -1173,6 +1253,30 @@ def _user_from(claims: Claims, token: str, recovery: bool) -> User: recovery=recovery, organization_id=str_claim(claims, "org_id") or None, token=token, + permissions=permission_names(claims.get("permissions")), + organization_roles=permission_names(claims.get("org_roles")), + organization_permissions=permission_names(claims.get("org_permissions")), + permissions_omitted=claims.get("permissions_omitted") is True, + ) + + +def _live_permission_set(body: Any) -> tuple[tuple[str, ...], tuple[str, ...], str | None] | None: + """The live read's answer, or ``None`` for a malformed one. Stricter than a claim: + a list that is missing here could otherwise read as holding nothing, when the + answer to give is that the set is unknown.""" + if not isinstance(body, dict): + return None + permissions = body.get("permissions") + organization_permissions = body.get("organizationPermissions") + organization_id = body.get("organizationId") + if not isinstance(permissions, list) or not isinstance(organization_permissions, list): + return None + if organization_id is not None and not isinstance(organization_id, str): + return None + return ( + permission_names(permissions), + permission_names(organization_permissions), + organization_id, ) diff --git a/src/seamless_auth/_permissions.py b/src/seamless_auth/_permissions.py new file mode 100644 index 0000000..843f7ff --- /dev/null +++ b/src/seamless_auth/_permissions.py @@ -0,0 +1,95 @@ +"""Permission rules, shared by the session guards and the verify-only guard. + +Matching is exact and case-sensitive, unlike roles: no trimming, no wildcards, no +implied grants. ``invoices:write`` does not satisfy ``invoices:read``. +""" + +from __future__ import annotations + +from collections.abc import Collection, Iterable, Mapping +from dataclasses import dataclass +from typing import Any, Literal + +PermissionMode = Literal["any", "all"] + + +def permission_names(claim: Any) -> tuple[str, ...]: + """A permission claim as names. A claim that is not a list reads as empty and + entries that are not strings are ignored, so a malformed claim never fails.""" + return tuple(p for p in claim if isinstance(p, str)) if isinstance(claim, list) else () + + +@dataclass(frozen=True) +class PermissionRule: + """What a guard requires. + + ``permissions`` may be empty, and an empty rule allows nobody, in any mode. + ``mode`` is ``"any"`` (at least one is held) or ``"all"`` (every one is held). + With ``organization``, only the active organization's permissions count and the + token must have an active organization. ``organization_param`` implies that, and + also names the route parameter that must equal the active organization's id. + Without it, a token for organization A passes on a route for organization B. + """ + + permissions: tuple[str, ...] = () + mode: PermissionMode = "any" + organization: bool = False + organization_param: str | None = None + + def __post_init__(self) -> None: + if isinstance(self.permissions, str): + raise TypeError("seamless-auth: pass a list of permissions, not a string") + names = tuple(self.permissions) + if not all(isinstance(p, str) and p for p in names): + raise ValueError("seamless-auth: permissions must be non-empty strings") + if self.mode not in ("any", "all"): + raise ValueError('seamless-auth: permission mode must be "any" or "all"') + param = self.organization_param + if param is not None and (not isinstance(param, str) or not param): + raise ValueError("seamless-auth: organization_param must be a non-empty string") + object.__setattr__(self, "permissions", names) + + def allows( + self, + permissions: Collection[str], + organization_permissions: Collection[str], + organization_id: str | None, + route_params: Mapping[str, Any] | None = None, + ) -> bool: + """Whether a holder of these permissions meets the rule.""" + if not self.permissions: + return False + if self.organization or self.organization_param is not None: + if not organization_id or not self._route_matches(organization_id, route_params): + return False + held = set(organization_permissions) + else: + held = {*permissions, *organization_permissions} + pick = all if self.mode == "all" else any + return pick(p in held for p in self.permissions) + + def _route_matches(self, organization_id: str, route_params: Mapping[str, Any] | None) -> bool: + if self.organization_param is None: + return True + # str() because a route converter may have parsed the value, into a UUID say. + value = (route_params or {}).get(self.organization_param) + return value is not None and str(value) == organization_id + + +def required_permissions( + permissions: Iterable[str] | None, + *, + mode: PermissionMode = "any", + organization: bool = False, + organization_param: str | None = None, +) -> PermissionRule | None: + """A guard's permission options as a rule: ``None`` means no permission check. + The other options without ``permissions`` are refused, because that guard would + look scoped and check nothing.""" + if permissions is None: + if mode != "any" or organization or organization_param is not None: + raise ValueError("seamless-auth: permission options need permissions") + return None + if isinstance(permissions, str): + raise TypeError("seamless-auth: pass a list of permissions, not a string") + return PermissionRule(tuple(permissions), mode, organization, organization_param) diff --git a/src/seamless_auth/_resource_server.py b/src/seamless_auth/_resource_server.py index 786a5cf..476ed02 100644 --- a/src/seamless_auth/_resource_server.py +++ b/src/seamless_auth/_resource_server.py @@ -11,7 +11,7 @@ import re import threading import time -from collections.abc import Callable, Iterable, Sequence +from collections.abc import Callable, Iterable, Mapping, Sequence from dataclasses import dataclass, field from typing import Any, TypeGuard from urllib.parse import urlsplit @@ -22,6 +22,7 @@ from ._http import Headers from ._jwks import FETCH_TIMEOUT, JwksCache, TokenError, read_limited from ._jwt import CLOCK_SKEW_SECONDS +from ._permissions import PermissionMode, PermissionRule, permission_names log = logging.getLogger("seamless_auth") @@ -87,6 +88,12 @@ class VerifiedUser: verified_at: float = 0 claims: dict[str, Any] = field(default_factory=dict, repr=False, compare=False) token: str = field(default="", repr=False) + permissions: tuple[str, ...] = () + organization_roles: tuple[str, ...] = () + organization_permissions: tuple[str, ...] = () + #: The token left its permissions out. The verify-only guard denies then: the + #: first-party endpoint that holds the live set refuses provider tokens. + permissions_omitted: bool = False @property def is_client(self) -> bool: @@ -297,6 +304,10 @@ def _verify(self, token: str, request: BearerRequest | None) -> VerifiedUser: verified_at=now, claims=claims, token=token, + permissions=permission_names(claims.get("permissions")), + organization_roles=permission_names(claims.get("org_roles")), + organization_permissions=permission_names(claims.get("org_permissions")), + permissions_omitted=claims.get("permissions_omitted") is True, ) def _confirmed(self, cnf: Any, request: BearerRequest | None) -> bool: @@ -326,6 +337,10 @@ class Guard: def check(self, user: VerifiedUser) -> None: raise NotImplementedError + def check_route(self, user: VerifiedUser, route_params: Mapping[str, Any]) -> None: + """:meth:`check`, for a guard that also reads the route. The bindings call this.""" + self.check(user) + class _ScopeGuard(Guard): def __init__(self, scopes: tuple[str, ...]) -> None: @@ -350,6 +365,20 @@ def check(self, user: VerifiedUser) -> None: raise BearerError(403, {"error": "insufficient_role"}, None) +class _PermissionGuard(Guard): + def __init__(self, rule: PermissionRule) -> None: + self.rule = rule + + def check(self, user: VerifiedUser) -> None: + self.check_route(user, {}) + + def check_route(self, user: VerifiedUser, route_params: Mapping[str, Any]) -> None: + if user.permissions_omitted or not self.rule.allows( + user.permissions, user.organization_permissions, user.org_id, route_params + ): + raise BearerError(403, {"error": "insufficient_permission"}, None) + + class _StepUpGuard(Guard): def __init__( self, @@ -410,6 +439,20 @@ def require_role(*roles: str) -> Guard: return _RoleGuard(tuple(roles)) +def require_permission( + *permissions: str, + mode: PermissionMode = "any", + organization: bool = False, + organization_param: str | None = None, +) -> Guard: + """The token must meet a :class:`~seamless_auth.PermissionRule`. Otherwise 403 + ``insufficient_permission``, without naming the permissions. No permissions at + all allows nobody. Provider tokens do not carry these claims yet, so this denies + them until fells-code/seamless-auth-oidc-provider#47. A token with + ``permissions_omitted`` is denied, not read live.""" + return _PermissionGuard(PermissionRule(permissions, mode, organization, organization_param)) + + def require_step_up( *, acr_values: Sequence[str] | None = None, diff --git a/src/seamless_auth/django.py b/src/seamless_auth/django.py index 870444a..236c16f 100644 --- a/src/seamless_auth/django.py +++ b/src/seamless_auth/django.py @@ -24,6 +24,10 @@ def me(request): def admin(request): return JsonResponse({"id": user_of(request).id}) + @require_auth(permissions=["projects:read"], organization_param="organization_id") + def projects(request, organization_id): + 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. @@ -52,7 +56,7 @@ def orders(request): import functools import inspect import threading -from collections.abc import Callable, Iterable +from collections.abc import Callable, Iterable, Mapping from typing import Any, Literal, TypeVar, cast, overload from urllib.parse import quote, unquote @@ -69,12 +73,15 @@ def orders(request): Adapter, CrossSiteRequest, EnrollmentRequired, + InsufficientPermission, InsufficientRole, + PermissionsUnavailable, Unauthenticated, User, required_roles, ) from ._http import AuthRequest, AuthResponse, Headers +from ._permissions import PermissionMode, PermissionRule, required_permissions from ._resource_server import ( BearerError, BearerRequest, @@ -83,6 +90,7 @@ def orders(request): VerifiedUser, missing_token, ) +from ._resource_server import require_permission as _permission_guard from ._resource_server import require_role as _role_guard from ._resource_server import require_scope as _scope_guard from ._resource_server import require_step_up as _step_up_guard @@ -96,6 +104,7 @@ def orders(request): "get_resource_server", "require_auth", "require_bearer", + "require_permission", "require_role", "require_scope", "require_step_up", @@ -190,7 +199,13 @@ def console_view(request: HttpRequest, path: str = "") -> HttpResponse: def require_auth(view: F) -> F: ... @overload def require_auth( - *, allow_recovery: bool = False, roles: Iterable[str] | None = None + *, + allow_recovery: bool = False, + roles: Iterable[str] | None = None, + permissions: Iterable[str] | None = None, + permission_mode: PermissionMode = "any", + organization: bool = False, + organization_param: str | None = None, ) -> Callable[[F], F]: ... @@ -199,6 +214,10 @@ def require_auth( *, allow_recovery: bool = False, roles: Iterable[str] | None = None, + permissions: Iterable[str] | None = None, + permission_mode: PermissionMode = "any", + organization: bool = False, + organization_param: 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 @@ -207,14 +226,30 @@ def require_auth( or authenticator app) answers 403 ``enrollment_required`` unless ``@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.""" + ``insufficient_role``. With ``permissions``, a session that does not meet the + :class:`~seamless_auth.PermissionRule` answers 403 ``insufficient_permission``, and + when a token omits its permissions and the live read fails, 503 + ``permissions_unavailable``. An empty list allows nobody. ``organization_param`` + names the URL parameter that must equal the active organization's id. Works on + sync and async views.""" required = required_roles(roles) + rule = required_permissions( + permissions, + mode=permission_mode, + organization=organization, + organization_param=organization_param, + ) if view is None: - return lambda inner: _guard(inner, allow_recovery, required) - return _guard(view, allow_recovery, required) + return lambda inner: _guard(inner, allow_recovery, required, rule) + return _guard(view, allow_recovery, required, rule) -def _guard(view: F, allow_recovery: bool, roles: tuple[str, ...] | None) -> F: +def _guard( + view: F, + allow_recovery: bool, + roles: tuple[str, ...] | None, + permissions: PermissionRule | None, +) -> F: def check(request: HttpRequest) -> HttpResponse | None: try: user = get_adapter().authenticate_request( @@ -222,7 +257,13 @@ def check(request: HttpRequest) -> HttpResponse | None: _headers(request), allow_recovery=allow_recovery, roles=roles, + permissions=permissions, + route_params=_route_params(request), ) + except InsufficientPermission: + return JsonResponse({"error": "insufficient_permission"}, status=403) + except PermissionsUnavailable: + return JsonResponse({"error": "permissions_unavailable"}, status=503) except InsufficientRole: return JsonResponse({"error": "insufficient_role"}, status=403) except EnrollmentRequired: @@ -266,8 +307,9 @@ def require_bearer( ``missing_token`` or ``invalid_token`` otherwise. With ``allow_clients=False`` a client credentials token answers 403 ``user_required``. Uses ``server``, or the one ``settings.SEAMLESS_RESOURCE_SERVER`` configures. Put it above the - :func:`require_scope`, :func:`require_role` and :func:`require_step_up` - decorators; they run top to bottom after it. Works on sync and async views.""" + :func:`require_scope`, :func:`require_role`, :func:`require_permission` and + :func:`require_step_up` decorators; they run top to bottom after it. Works on sync + and async views.""" def check(request: HttpRequest) -> HttpResponse | None: values = request.headers.get("Authorization") @@ -300,6 +342,26 @@ def require_role(*roles: str) -> Callable[[F], F]: return _guarded(_role_guard(*roles)) +def require_permission( + *permissions: str, + mode: PermissionMode = "any", + organization: bool = False, + organization_param: str | None = None, +) -> Callable[[F], F]: + """Under :func:`require_bearer`: the token must meet the rule, else 403 + ``insufficient_permission``. Provider tokens do not carry these claims until + fells-code/seamless-auth-oidc-provider#47, so this denies them today. For a session, + use ``@require_auth(permissions=[...])``.""" + return _guarded( + _permission_guard( + *permissions, + mode=mode, + organization=organization, + organization_param=organization_param, + ) + ) + + def require_step_up( *, acr_values: list[str] | tuple[str, ...] | None = None, @@ -322,7 +384,7 @@ def check(request: HttpRequest) -> HttpResponse | None: try: if not isinstance(user, VerifiedUser): raise missing_token() - guard.check(user) + guard.check_route(user, _route_params(request)) except BearerError as err: return _bearer_response(err) return None @@ -361,6 +423,11 @@ def user_of(request: HttpRequest) -> User: return cast(User, request.seamless_user) # type: ignore[attr-defined] +def _route_params(request: HttpRequest) -> Mapping[str, Any]: + match = request.resolver_match + return match.kwargs if match is not None else {} + + def _headers(request: HttpRequest) -> Headers: return Headers(request.headers.items()) diff --git a/src/seamless_auth/fastapi.py b/src/seamless_auth/fastapi.py index 829348d..6b97661 100644 --- a/src/seamless_auth/fastapi.py +++ b/src/seamless_auth/fastapi.py @@ -20,6 +20,16 @@ def me(user: User = Depends(current_user)): def admin(user: User = Depends(RequireUser(auth, roles=["admin"]))): return {"id": user.id} + @app.get("/orgs/{organization_id}/projects") + def projects( + user: User = Depends( + RequireUser( + auth, permissions=["projects:read"], organization_param="organization_id" + ) + ), + ): + return {"id": user.id} + Verify-only, for an API that only receives OpenID Provider access tokens:: from seamless_auth import ResourceServer, VerifiedUser, require_scope @@ -37,6 +47,7 @@ def orders(user: VerifiedUser = Depends(RequireBearer(rs, require_scope("orders: from __future__ import annotations from collections.abc import Iterable +from typing import Any from urllib.parse import quote, urlsplit from fastapi import APIRouter, FastAPI, HTTPException, Request @@ -49,17 +60,21 @@ def orders(user: VerifiedUser = Depends(RequireBearer(rs, require_scope("orders: Adapter, CrossSiteRequest, EnrollmentRequired, + InsufficientPermission, InsufficientRole, + PermissionsUnavailable, Unauthenticated, User, required_roles, ) from ._http import AuthRequest, AuthResponse, Headers +from ._permissions import PermissionMode, required_permissions from ._resource_server import BearerError, BearerRequest, Guard, ResourceServer, VerifiedUser __all__ = [ "BearerHTTPException", "OptionalUser", + "PermissionHTTPException", "RequireBearer", "RequireUser", "add_bearer_error_handler", @@ -112,6 +127,13 @@ class RequireUser: ``allow_recovery`` is set. With ``roles``, a session granted none of them (by :func:`~seamless_auth.role_grants_access`) answers 403 ``insufficient_role``. + With ``permissions``, a session that does not meet the + :class:`~seamless_auth.PermissionRule` answers 403 + ``{"error": "insufficient_permission"}``, and when a token omits its permissions + and the live read fails, 503 ``{"error": "permissions_unavailable"}``. An empty + list allows nobody. ``organization_param`` names the path parameter that must + equal the active organization's id. + A plain function, so FastAPI runs it on its threadpool: the first sight of a key may fetch the auth API's key set. """ @@ -122,10 +144,20 @@ def __init__( *, allow_recovery: bool = False, roles: Iterable[str] | None = None, + permissions: Iterable[str] | None = None, + permission_mode: PermissionMode = "any", + organization: bool = False, + organization_param: str | None = None, ) -> None: self._adapter = adapter self._allow_recovery = allow_recovery self._roles = required_roles(roles) + self._permissions = required_permissions( + permissions, + mode=permission_mode, + organization=organization, + organization_param=organization_param, + ) def __call__(self, request: Request) -> User: try: @@ -134,7 +166,13 @@ def __call__(self, request: Request) -> User: _headers(request), allow_recovery=self._allow_recovery, roles=self._roles, + permissions=self._permissions, + route_params=request.path_params, ) + except InsufficientPermission: + raise _permission_refusal(request, 403, "insufficient_permission") from None + except PermissionsUnavailable: + raise _permission_refusal(request, 503, "permissions_unavailable") from None except InsufficientRole: raise HTTPException(status_code=403, detail="insufficient_role") from None except EnrollmentRequired: @@ -145,6 +183,33 @@ def __call__(self, request: Request) -> User: raise HTTPException(status_code=401, detail="unauthenticated") from None +class PermissionHTTPException(HTTPException): + """A permission guard's refusal, answered ``{"error": ...}`` as on every other + framework and adapter, not FastAPI's ``{"detail": ...}``. To answer it differently, + register a handler for this class.""" + + def __init__(self, status_code: int, error: str) -> None: + super().__init__(status_code, detail=error) + self.body = {"error": error} + + +def _permission_exception_handler(request: Request, exc: Exception) -> Response: + if not isinstance(exc, PermissionHTTPException): + raise exc + return JSONResponse(exc.body, status_code=exc.status_code) + + +def _permission_refusal(request: Request, status: int, error: str) -> PermissionHTTPException: + # A dependency can only raise, and FastAPI's handler for an HTTPException always + # wraps the body in "detail". An add_exception_handler call would be a second step + # that, once forgotten, changes the response shape, so the guard registers the + # handler itself. A handler the application registered wins. + handlers: Any = request.scope.get("starlette.exception_handlers") + if handlers is not None: + handlers[0].setdefault(PermissionHTTPException, _permission_exception_handler) + return PermissionHTTPException(status, error) + + class OptionalUser: """A dependency that returns the :class:`~seamless_auth.User` exactly when :class:`RequireUser` with its defaults would, and ``None`` otherwise (no session, @@ -238,8 +303,10 @@ def __call__(self, request: Request) -> VerifiedUser: allow_clients=self._allow_clients, ) for guard in self._guards: - guard.check(user) + guard.check_route(user, request.path_params) except BearerError as err: + if err.error == "insufficient_permission": + raise _permission_refusal(request, err.status, err.error) from None raise BearerHTTPException(err) from None return user diff --git a/tests/test_permissions.py b/tests/test_permissions.py new file mode 100644 index 0000000..ae7596f --- /dev/null +++ b/tests/test_permissions.py @@ -0,0 +1,944 @@ +"""The permission guard, through the real FastAPI and Django stacks. + +Every row of the guard contract runs on each framework and on both transports, so a +difference between them fails here rather than in the conformance suite. +""" + +from __future__ import annotations + +from collections.abc import Callable, Iterator +from dataclasses import dataclass +from typing import Any +from uuid import UUID + +import httpx2 +import jwt +import pytest +from conftest import ( + _KEY, + API_URL, + JWKS, + TEST_SECRET, + TEST_SERVICE, + FakeApi, + api_token, + respond, + signed_cookie, +) +from django.conf import settings + +if not settings.configured: + settings.configure( + SECRET_KEY="django-tests-only", + ALLOWED_HOSTS=["*"], + ROOT_URLCONF="test_django_urls", + MIDDLEWARE=["django.middleware.csrf.CsrfViewMiddleware"], + INSTALLED_APPS=[], + ) + import django + + django.setup() + +from django.test import Client, override_settings +from fastapi import Depends, FastAPI, Request +from fastapi.testclient import TestClient +from starlette.responses import JSONResponse, Response +from test_permissions_django_urls import GUARDS +from verify_only_support import TOKENS, Provider + +from seamless_auth import ( + Adapter, + CrossSiteRequest, + EnrollmentRequired, + InsufficientPermission, + InsufficientRole, + PermissionRule, + PermissionsUnavailable, + ResourceServer, + Unauthenticated, + User, + VerifiedUser, + require_permission, + require_role, +) +from seamless_auth.django import _reset_adapter, _reset_resource_server, require_auth +from seamless_auth.fastapi import ( + PermissionHTTPException, + RequireBearer, + RequireUser, + add_bearer_error_handler, +) + +ORG = "6f1c2c9e-2c5e-4b7a-9d39-0d3f6a1f0a11" +OTHER_ORG = "0b4f3f0e-8a55-4b0a-8d6a-52c8c27c1e77" +LIVE = "/users/me/permissions" +DENIED = (403, {"error": "insufficient_permission"}) +UNAVAILABLE = (503, {"error": "permissions_unavailable"}) +ALLOWED = (200, {"id": "u1"}) + + +def _route_template(route: str) -> str: + return route.replace("", "{organization_id}").replace( + "", "{organization_id:uuid}" + ) + + +def fastapi_app(adapter: Adapter) -> FastAPI: + app = FastAPI() + for route, options in GUARDS.items(): + guard = RequireUser(adapter, **options) + + def handler(user: User = Depends(guard)) -> dict[str, str]: # noqa: B008 + return {"id": user.id} + + app.add_api_route(f"/sync/{_route_template(route)}", handler, methods=["GET", "POST"]) + return app + + +@dataclass +class Stack: + """One framework's routes, and the bodies that framework gives the existing + refusals, which the permission guard leaves as they are.""" + + get: Callable[..., Any] + prefix: str + unauthenticated: dict[str, str] + enrollment_required: dict[str, str] + + def status(self, route: str, headers: dict[str, str] | None = None) -> tuple[int, Any]: + r = self.get(f"/{self.prefix}/{route}", headers=headers or {}) + return r.status_code, r.json() + + +@pytest.fixture(params=["fastapi", "django", "django-async"]) +def stack(request: pytest.FixtureRequest, api: FakeApi) -> Iterator[Stack]: + if request.param == "fastapi": + client = TestClient(fastapi_app(api.adapter()), base_url="https://testserver") + yield Stack( + client.get, "sync", {"detail": "unauthenticated"}, {"detail": "enrollment_required"} + ) + return + options = { + "auth_server_url": API_URL, + "cookie_secret": TEST_SECRET, + "service_secret": TEST_SERVICE, + "jwks_kid": "test-main", + "disable_manifest_fetch": True, + "http_client": api.client(), + } + _reset_adapter() + with override_settings(SEAMLESS_AUTH=options, ROOT_URLCONF="test_permissions_django_urls"): + yield Stack( + Client(enforce_csrf_checks=True).get, + "async" if request.param == "django-async" else "sync", + {"error": "unauthenticated"}, + {"error": "enrollment_required"}, + ) + _reset_adapter() + + +def bearer(**claims: Any) -> dict[str, str]: + return {"authorization": f"Bearer {api_token('u1', 'access', **claims)}"} + + +def cookie(**claims: Any) -> dict[str, str]: + token = api_token("u1", "access", **claims) + return {"cookie": signed_cookie("seamless-access", {"sub": "u1", "token": token})} + + +@pytest.fixture(params=[bearer, cookie], ids=["bearer", "cookie"]) +def session(request: pytest.FixtureRequest) -> Callable[..., dict[str, str]]: + transport: Callable[..., dict[str, str]] = request.param + return transport + + +def live_body(**fields: Any) -> dict[str, Any]: + return { + "roles": [], + "permissions": [], + "organizationId": None, + "organizationRoles": [], + "organizationPermissions": [], + **fields, + } + + +# Not authenticated, and recovery sessions. + + +@pytest.mark.parametrize("route", ["any", "all", "org", f"org/{ORG}", "empty"]) +def test_no_session_is_the_existing_401(stack: Stack, route: str) -> None: + assert stack.status(route) == (401, stack.unauthenticated) + + +def test_a_bad_token_is_the_existing_401(stack: Stack) -> None: + assert stack.status("any", {"authorization": "Bearer nope"}) == (401, stack.unauthenticated) + + +def test_a_recovery_session_is_the_existing_rejection( + stack: Stack, api: FakeApi, session: Callable[..., dict[str, str]] +) -> None: + for omitted in ({"permissions": ["reports:read"]}, {"permissions_omitted": True}): + assert stack.status("any", session(recovery=True, **omitted)) == ( + 403, + stack.enrollment_required, + ) + assert api.calls_to("GET", LIVE) == [] + + +def test_allow_recovery_still_checks_permissions( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + assert stack.status("recovery", session(recovery=True)) == DENIED + allowed = session(recovery=True, permissions=["reports:read"]) + assert stack.status("recovery", allowed) == ALLOWED + + +# Modes. + + +def test_any_needs_one(stack: Stack, session: Callable[..., dict[str, str]]) -> None: + assert stack.status("any", session(permissions=["reports:read"])) == ALLOWED + assert stack.status("any", session(permissions=["reports:export", "x"])) == ALLOWED + assert stack.status("any", session(permissions=["x"])) == DENIED + assert stack.status("any", session()) == DENIED + + +def test_all_needs_every_one(stack: Stack, session: Callable[..., dict[str, str]]) -> None: + assert stack.status("all", session(permissions=["reports:read"])) == DENIED + both = ["reports:read", "reports:write"] + assert stack.status("all", session(permissions=both)) == ALLOWED + + +def test_all_may_draw_from_either_claim( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + split = session(permissions=["reports:read"], org_id=ORG, org_permissions=["reports:write"]) + assert stack.status("all", split) == ALLOWED + + +def test_a_refusal_names_no_permissions(stack: Stack) -> None: + r = stack.get(f"/{stack.prefix}/any", headers=bearer(permissions=["x"])) + assert r.status_code == 403 + assert r.json() == DENIED[1] + assert "reports" not in r.content.decode() + + +# Matching. + + +@pytest.mark.parametrize( + "held", + [ + "REPORTS:READ", + "Reports:read", + " reports:read", + "reports:read ", + "reports:*", + "*", + "reports", + "reports:read:all", + "reports:write", + ], +) +def test_matching_is_exact(stack: Stack, session: Callable[..., dict[str, str]], held: str) -> None: + assert stack.status("any", session(permissions=[held])) == DENIED + + +def test_a_role_is_not_a_permission(stack: Stack) -> None: + assert stack.status("any", bearer(roles=["reports:read", "reports:*"])) == DENIED + org_roles = bearer(org_id=ORG, org_roles=["reports:read"]) + assert stack.status("any", org_roles) == DENIED + + +# Scope. + + +def test_by_default_an_organization_permission_counts( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + held = session(org_id=ORG, org_permissions=["reports:read"]) + assert stack.status("any", held) == ALLOWED + + +def test_organization_scope_counts_only_organization_permissions( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + instance = session(org_id=ORG, permissions=["projects:write"]) + assert stack.status("org", instance) == DENIED + org = session(org_id=ORG, org_permissions=["projects:write"]) + assert stack.status("org", org) == ALLOWED + + +@pytest.mark.parametrize( + "claims", + [{}, {"org_id": ""}, {"org_id": None}, {"org_id": 7}, {"org_id": ["x"]}], + ids=["absent", "empty", "null", "number", "list"], +) +def test_organization_scope_needs_an_organization( + stack: Stack, session: Callable[..., dict[str, str]], claims: dict[str, Any] +) -> None: + headers = session(org_permissions=["projects:write"], **claims) + assert stack.status("org", headers) == DENIED + assert stack.status(f"org/{ORG}", headers) == DENIED + + +def test_organization_all_needs_every_one_from_the_organization( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + split = session(org_id=ORG, permissions=["projects:read"], org_permissions=["projects:write"]) + assert stack.status("org-all", split) == DENIED + both = session(org_id=ORG, org_permissions=["projects:read", "projects:write"]) + assert stack.status("org-all", both) == ALLOWED + + +def test_a_bound_route_needs_that_organization( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + held = session(org_id=ORG, org_permissions=["projects:write"]) + assert stack.status(f"org/{ORG}", held) == ALLOWED + assert stack.status(f"org/{OTHER_ORG}", held) == DENIED + assert stack.status(f"org/{ORG.upper()}", held) == DENIED + + +def test_a_bound_route_counts_only_organization_permissions( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + held = session(org_id=ORG, permissions=["projects:write"]) + assert stack.status(f"org/{ORG}", held) == DENIED + + +def test_a_route_without_the_parameter_denies( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + held = session(org_id=ORG, org_permissions=["projects:write"]) + assert stack.status("unbound", held) == DENIED + + +def test_a_parsed_route_parameter_compares_as_its_text( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + held = session(org_id=ORG, org_permissions=["projects:write"]) + assert stack.status(f"org-uuid/{UUID(ORG)}", held) == ALLOWED + assert stack.status(f"org-uuid/{UUID(OTHER_ORG)}", held) == DENIED + + +# An empty list. + + +@pytest.mark.parametrize("route", ["empty", "empty-all", "empty-org"]) +def test_an_empty_list_never_opens_a_route( + stack: Stack, api: FakeApi, session: Callable[..., dict[str, str]], route: str +) -> None: + api.on("GET", LIVE, respond(200, live_body(permissions=["x"], organizationId=ORG))) + held = session(permissions=["x"], org_id=ORG, org_permissions=["x"]) + assert stack.status(route, held) == DENIED + assert stack.status(route, session(permissions_omitted=True, org_id=ORG)) == DENIED + assert api.calls_to("GET", LIVE) == [] + + +# Malformed claims. + + +@pytest.mark.parametrize( + "value", + ["reports:read", {"reports:read": True}, 1, True, None], + ids=["string", "object", "number", "bool", "null"], +) +def test_a_claim_that_is_not_a_list_reads_as_empty( + stack: Stack, session: Callable[..., dict[str, str]], value: Any +) -> None: + assert stack.status("any", session(permissions=value)) == DENIED + assert stack.status("org", session(org_id=ORG, org_permissions=value)) == DENIED + + +def test_entries_that_are_not_strings_are_ignored( + stack: Stack, session: Callable[..., dict[str, str]] +) -> None: + noisy = [1, None, {"a": 1}, ["reports:read"], True] + assert stack.status("any", session(permissions=noisy)) == DENIED + assert stack.status("any", session(permissions=[*noisy, "reports:read"])) == ALLOWED + + +@pytest.mark.parametrize("value", ["true", 1, [True], False, None]) +def test_only_a_literal_true_means_omitted( + stack: Stack, api: FakeApi, session: Callable[..., dict[str, str]], value: Any +) -> None: + api.on("GET", LIVE, respond(200, live_body(permissions=["reports:read"]))) + assert stack.status("any", session(permissions_omitted=value)) == DENIED + allowed = session(permissions_omitted=value, permissions=["reports:read"]) + assert stack.status("any", allowed) == ALLOWED + assert api.calls_to("GET", LIVE) == [] + + +# The live read. + + +def test_a_token_with_its_permissions_is_never_read_live(stack: Stack, api: FakeApi) -> None: + assert stack.status("any", bearer(permissions=["reports:read"])) == ALLOWED + assert api.calls_to("GET", LIVE) == [] + + +def test_the_live_read_decides_when_omitted( + stack: Stack, api: FakeApi, session: Callable[..., dict[str, str]] +) -> None: + api.on("GET", LIVE, respond(200, live_body(permissions=["reports:read"]))) + omitted = session(permissions_omitted=True) + assert stack.status("any", omitted) == ALLOWED + assert stack.status("all", omitted) == DENIED + + +def test_the_live_read_ignores_what_the_token_says(stack: Stack, api: FakeApi) -> None: + api.on("GET", LIVE, respond(200, live_body())) + held = bearer(permissions_omitted=True, permissions=["reports:read"]) + assert stack.status("any", held) == DENIED + + +def test_the_live_read_sends_only_the_callers_token(stack: Stack, api: FakeApi) -> None: + api.on("GET", LIVE, respond(200, live_body(permissions=["reports:read"]))) + token = api_token("u1", "access", permissions_omitted=True) + + assert stack.status("any", {"authorization": f"Bearer {token}"}) == ALLOWED + + [call] = api.calls_to("GET", LIVE) + assert call.headers["authorization"] == f"Bearer {token}" + assert "x-seamless-service-token" not in call.headers + + +def test_the_live_read_uses_the_token_inside_the_cookie(stack: Stack, api: FakeApi) -> None: + api.on("GET", LIVE, respond(200, live_body(permissions=["reports:read"]))) + token = api_token("u1", "access", permissions_omitted=True) + held = {"cookie": signed_cookie("seamless-access", {"sub": "u1", "token": token})} + + assert stack.status("any", held) == ALLOWED + + [call] = api.calls_to("GET", LIVE) + assert call.headers["authorization"] == f"Bearer {token}" + + +def test_the_live_read_is_not_cached(stack: Stack, api: FakeApi) -> None: + api.on("GET", LIVE, respond(200, live_body(permissions=["reports:read"]))) + omitted = bearer(permissions_omitted=True) + assert stack.status("any", omitted) == ALLOWED + + api.on("GET", LIVE, respond(200, live_body())) + assert stack.status("any", omitted) == DENIED + assert len(api.calls_to("GET", LIVE)) == 2 + + +def test_the_live_read_goes_to_the_configured_server_only(stack: Stack, api: FakeApi) -> None: + urls: list[str] = [] + + def record(request: httpx2.Request) -> httpx2.Response: + urls.append(str(request.url)) + return httpx2.Response(200, json=live_body(permissions=["reports:read"])) + + api.on("GET", LIVE, record) + claims = {"aud": API_URL, "auth_server_url": "http://evil.test"} + assert stack.status("any", bearer(permissions_omitted=True, **claims)) == ALLOWED + assert urls == [f"{API_URL}{LIVE}"] + + +def test_organization_scope_reads_the_live_organization(stack: Stack, api: FakeApi) -> None: + omitted = bearer(permissions_omitted=True, org_id=ORG) + + api.on( + "GET", + LIVE, + respond(200, live_body(organizationId=ORG, organizationPermissions=["projects:write"])), + ) + assert stack.status("org", omitted) == ALLOWED + assert stack.status(f"org/{ORG}", omitted) == ALLOWED + assert stack.status(f"org/{OTHER_ORG}", omitted) == DENIED + + # The organization the live read names decides, not the one in the token. + api.on( + "GET", + LIVE, + respond( + 200, live_body(organizationId=OTHER_ORG, organizationPermissions=["projects:write"]) + ), + ) + assert stack.status(f"org/{ORG}", omitted) == DENIED + assert stack.status(f"org/{OTHER_ORG}", omitted) == ALLOWED + + api.on("GET", LIVE, respond(200, live_body(organizationPermissions=["projects:write"]))) + assert stack.status("org", omitted) == DENIED + + api.on("GET", LIVE, respond(200, live_body(organizationId=ORG, permissions=["projects:write"]))) + assert stack.status("org", omitted) == DENIED + + +def test_a_live_401_is_the_existing_401(stack: Stack, api: FakeApi) -> None: + api.on("GET", LIVE, respond(401, {"error": "unauthorized"})) + assert stack.status("any", bearer(permissions_omitted=True)) == (401, stack.unauthenticated) + + +@pytest.mark.parametrize("status", [204, 301, 302, 400, 403, 404, 429, 500, 502, 503]) +def test_any_other_live_status_fails_closed(stack: Stack, api: FakeApi, status: int) -> None: + api.on("GET", LIVE, lambda _req: httpx2.Response(status)) + assert stack.status("any", bearer(permissions_omitted=True)) == UNAVAILABLE + + +@pytest.mark.parametrize( + "error", + [httpx2.ReadTimeout("slow"), httpx2.ConnectTimeout("slow"), httpx2.ConnectError("down")], + ids=["read-timeout", "connect-timeout", "connect-error"], +) +def test_an_unreachable_live_read_fails_closed( + stack: Stack, api: FakeApi, error: Exception +) -> None: + def fail(_req: httpx2.Request) -> httpx2.Response: + raise error + + api.on("GET", LIVE, fail) + assert stack.status("any", bearer(permissions_omitted=True)) == UNAVAILABLE + + +@pytest.mark.parametrize( + "content", + [ + b"", + b"not json", + b"null", + b"[]", + b'"reports:read"', + b"{}", + b'{"permissions":["reports:read"]}', + b'{"organizationPermissions":[]}', + b'{"permissions":"reports:read","organizationPermissions":[]}', + b'{"permissions":["reports:read"],"organizationPermissions":{}}', + b'{"permissions":["reports:read"],"organizationPermissions":[],"organizationId":7}', + b"x" * ((1 << 20) + 1), + ], + ids=[ + "empty", + "text", + "null", + "array", + "string", + "no-fields", + "no-org-permissions", + "no-permissions", + "permissions-string", + "org-permissions-object", + "org-id-number", + "too-large", + ], +) +def test_a_malformed_live_body_fails_closed(stack: Stack, api: FakeApi, content: bytes) -> None: + api.on("GET", LIVE, lambda _req: httpx2.Response(200, content=content)) + assert stack.status("any", bearer(permissions_omitted=True)) == UNAVAILABLE + + +def test_entries_of_a_live_list_that_are_not_strings_are_ignored( + stack: Stack, api: FakeApi +) -> None: + api.on("GET", LIVE, respond(200, live_body(permissions=[1, None]))) + assert stack.status("any", bearer(permissions_omitted=True)) == DENIED + api.on("GET", LIVE, respond(200, live_body(permissions=[1, "reports:read"]))) + assert stack.status("any", bearer(permissions_omitted=True)) == ALLOWED + + +def test_a_failed_live_read_is_logged_without_the_token( + stack: Stack, api: FakeApi, caplog: pytest.LogCaptureFixture +) -> None: + api.on("GET", LIVE, lambda _req: httpx2.Response(500)) + token = api_token("u1", "access", permissions_omitted=True) + + assert stack.status("any", {"authorization": f"Bearer {token}"}) == UNAVAILABLE + + assert "live permissions read answered 500" in caplog.text + assert token not in caplog.text + + +# The user type. + + +def rule(*permissions: str, **options: Any) -> PermissionRule: + return PermissionRule(permissions, **options) + + +def test_the_user_carries_the_claims(api: FakeApi) -> None: + claims = { + "permissions": ["reports:read", 1], + "org_id": ORG, + "org_roles": ["admin", None], + "org_permissions": ["projects:write"], + } + for headers in (bearer(**claims), cookie(**claims)): + user = api.adapter().authenticate(list(headers.items())) + assert user.permissions == ("reports:read",) + assert user.organization_id == ORG + assert user.organization_roles == ("admin",) + assert user.organization_permissions == ("projects:write",) + assert user.permissions_omitted is False + + +def test_the_user_says_when_the_token_omits_them(api: FakeApi) -> None: + for headers in ( + bearer(permissions_omitted=True, org_id=ORG), + cookie(permissions_omitted=True, org_id=ORG), + ): + user = api.adapter().authenticate(list(headers.items())) + assert user.permissions_omitted is True + assert user.permissions == () + assert user.organization_id == ORG + + +def test_cookie_payload_claims_are_not_read(api: FakeApi) -> None: + held = { + "cookie": signed_cookie( + "seamless-access", + { + "sub": "u1", + "token": api_token("u1", "access"), + "permissions": ["reports:read"], + "org_id": ORG, + "org_permissions": ["projects:write"], + "permissions_omitted": True, + }, + ) + } + user = api.adapter().authenticate(list(held.items())) + assert user.permissions == () + assert user.organization_permissions == () + assert user.organization_id is None + assert user.permissions_omitted is False + + +def test_organization_roles_do_not_count_for_the_role_guard(api: FakeApi) -> None: + headers = bearer(org_id=ORG, org_roles=["admin"], permissions=["admin"]) + with pytest.raises(InsufficientRole): + api.adapter().authenticate(list(headers.items()), roles=["admin"]) + + +# The core, without a framework. + + +def test_the_core_raises_what_the_bindings_answer(api: FakeApi) -> None: + adapter = api.adapter() + required = rule("reports:read") + + with pytest.raises(InsufficientPermission): + adapter.authenticate(list(bearer().items()), permissions=required) + + api.on("GET", LIVE, lambda _req: httpx2.Response(500)) + omitted = list(bearer(permissions_omitted=True).items()) + with pytest.raises(PermissionsUnavailable): + adapter.authenticate(omitted, permissions=required) + + api.on("GET", LIVE, lambda _req: httpx2.Response(401)) + with pytest.raises(Unauthenticated): + adapter.authenticate(omitted, permissions=required) + + with pytest.raises(EnrollmentRequired): + adapter.authenticate(list(bearer(recovery=True).items()), permissions=required) + + +def test_the_core_binds_the_route_it_is_given(api: FakeApi) -> None: + adapter = api.adapter() + headers = list(bearer(org_id=ORG, org_permissions=["p"]).items()) + bound = rule("p", organization_param="org") + + assert adapter.authenticate(headers, permissions=bound, route_params={"org": ORG}).id == "u1" + for params in (None, {}, {"org": OTHER_ORG}, {"org": None}, {"other": ORG}): + with pytest.raises(InsufficientPermission): + adapter.authenticate(headers, permissions=bound, route_params=params) + + +def test_the_cross_site_check_comes_before_permissions(api: FakeApi) -> None: + api.on("GET", LIVE, respond(200, live_body())) + headers = {**cookie(permissions_omitted=True), "sec-fetch-site": "cross-site"} + with pytest.raises(CrossSiteRequest): + api.adapter().authenticate_request( + "POST", list(headers.items()), permissions=rule("reports:read") + ) + assert api.calls_to("GET", LIVE) == [] + + +def test_roles_and_permissions_together_both_apply(api: FakeApi) -> None: + adapter = api.adapter() + required = rule("reports:read") + with pytest.raises(InsufficientRole): + adapter.authenticate( + list(bearer(permissions=["reports:read"]).items()), + roles=["admin"], + permissions=required, + ) + with pytest.raises(InsufficientPermission): + adapter.authenticate( + list(bearer(roles=["admin"]).items()), roles=["admin"], permissions=required + ) + + +@pytest.mark.parametrize( + ("options", "error"), + [ + ({"permissions": "reports:read"}, TypeError), + ({"permissions": [""]}, ValueError), + ({"permissions": [1]}, ValueError), + ({"permissions": ["a"], "permission_mode": "some"}, ValueError), + ({"permissions": ["a"], "organization_param": ""}, ValueError), + # Each of these would look scoped and check nothing. + ({"permission_mode": "all"}, ValueError), + ({"organization": True}, ValueError), + ({"organization_param": "org"}, ValueError), + ], +) +def test_a_guard_set_up_wrong_is_refused( + api: FakeApi, options: dict[str, Any], error: type[Exception] +) -> None: + with pytest.raises(error): + RequireUser(api.adapter(), **options) + with pytest.raises(error): + require_auth(**options) + + +def test_an_iterable_of_permissions_is_kept(api: FakeApi) -> None: + client = _one_route(api, RequireUser(api.adapter(), permissions=iter(["reports:read"]))) + assert client.get("/x", headers=bearer(permissions=["reports:read"])).status_code == 200 + assert client.get("/x", headers=bearer(permissions=["reports:read"])).status_code == 200 + + +# FastAPI's response shape. + + +def _one_route(api: FakeApi, guard: RequireUser, app: FastAPI | None = None) -> TestClient: + app = app or FastAPI() + + @app.get("/x") + def x(user: User = Depends(guard)) -> dict[str, str]: # noqa: B008 + return {"id": user.id} + + return TestClient(app) + + +def test_fastapi_needs_no_setup_for_the_permission_body(api: FakeApi) -> None: + client = _one_route(api, RequireUser(api.adapter(), permissions=["reports:read"])) + + r = client.get("/x", headers=bearer()) + assert (r.status_code, r.json()) == DENIED + + api.on("GET", LIVE, lambda _req: httpx2.Response(500)) + r = client.get("/x", headers=bearer(permissions_omitted=True)) + assert (r.status_code, r.json()) == UNAVAILABLE + + +def test_fastapi_a_mounted_app_gets_the_permission_body(api: FakeApi) -> None: + inner = FastAPI() + _one_route(api, RequireUser(api.adapter(), permissions=["reports:read"]), inner) + outer = FastAPI() + outer.mount("/inner", inner) + + r = TestClient(outer).get("/inner/x", headers=bearer()) + assert (r.status_code, r.json()) == DENIED + + +def test_fastapi_an_application_handler_wins(api: FakeApi) -> None: + app = FastAPI() + + def mine(request: Request, exc: Exception) -> Response: + return JSONResponse({"mine": True}, status_code=418) + + app.add_exception_handler(PermissionHTTPException, mine) + client = _one_route(api, RequireUser(api.adapter(), permissions=["reports:read"]), app) + + r = client.get("/x", headers=bearer()) + assert (r.status_code, r.json()) == (418, {"mine": True}) + + +def test_fastapi_the_session_role_guard_still_answers_under_detail(api: FakeApi) -> None: + """Records the role guard's existing difference from every other guard: its body is + ``{"detail": ...}``. The permission guard does not share it. See the README, Roles.""" + client = _one_route(api, RequireUser(api.adapter(), roles=["admin"])) + + r = client.get("/x", headers=bearer()) + assert (r.status_code, r.json()) == (403, {"detail": "insufficient_role"}) + + +# Verify-only. + +ISSUER = "https://login.test" +AUDIENCE = "https://api.test" +JWKS_URI = f"{ISSUER}/jwks" + + +def provider_token(**claims: Any) -> str: + now = 2_000_000_000 + payload = { + "iss": ISSUER, + "aud": AUDIENCE, + "sub": "u1", + "client_id": "app", + "iat": now, + "exp": now + 300, + **claims, + } + return jwt.encode(payload, _KEY, algorithm="RS256", headers={"kid": "k1", "typ": "at+jwt"}) + + +def bearer_token_header(**claims: Any) -> dict[str, str]: + return {"authorization": f"Bearer {provider_token(**claims)}"} + + +@dataclass +class KeyServer: + paths: list[str] + + def handle(self, request: httpx2.Request) -> httpx2.Response: + self.paths.append(request.url.path) + if str(request.url) == JWKS_URI: + return httpx2.Response(200, json=JWKS) + return httpx2.Response(404) + + def options(self) -> dict[str, Any]: + return { + "issuer": ISSUER, + "audience": AUDIENCE, + "jwks_uri": JWKS_URI, + "http_client": httpx2.Client(transport=httpx2.MockTransport(self.handle)), + "clock": lambda: 2_000_000_000.0, + } + + +def fastapi_bearer_app(server: ResourceServer) -> FastAPI: + app = FastAPI() + add_bearer_error_handler(app) + for route, options in GUARDS.items(): + if "allow_recovery" in options: + continue + guard = require_permission( + *options["permissions"], + mode=options.get("permission_mode", "any"), + organization=options.get("organization", False), + organization_param=options.get("organization_param"), + ) + dependency = RequireBearer(server, guard) + + def handler(user: VerifiedUser = Depends(dependency)) -> dict[str, str]: # noqa: B008 + return {"id": user.sub} + + app.add_api_route(f"/bearer/{_route_template(route)}", handler, methods=["GET"]) + return app + + +@pytest.fixture +def keys() -> KeyServer: + return KeyServer([]) + + +@pytest.fixture(params=["fastapi", "django", "django-async"]) +def bearer_stack(request: pytest.FixtureRequest, keys: KeyServer) -> Iterator[Stack]: + if request.param == "fastapi": + client = TestClient(fastapi_bearer_app(ResourceServer(**keys.options()))) + yield Stack(client.get, "bearer", {"error": "missing_token"}, {}) + return + _reset_resource_server() + with override_settings( + SEAMLESS_RESOURCE_SERVER=keys.options(), ROOT_URLCONF="test_permissions_django_urls" + ): + prefix = "bearer-async" if request.param == "django-async" else "bearer" + yield Stack(Client().get, prefix, {"error": "missing_token"}, {}) + _reset_resource_server() + + +def test_verify_only_no_token_is_the_existing_401(bearer_stack: Stack) -> None: + assert bearer_stack.status("any") == (401, bearer_stack.unauthenticated) + + +def test_verify_only_denies_a_token_without_the_claims(bearer_stack: Stack) -> None: + held = bearer_token_header(roles=["reports:read"], scope="reports:read") + assert bearer_stack.status("any", held) == DENIED + + +def test_verify_only_modes_and_scopes(bearer_stack: Stack) -> None: + assert bearer_stack.status("any", bearer_token_header(permissions=["reports:read"])) == ALLOWED + assert bearer_stack.status("all", bearer_token_header(permissions=["reports:read"])) == DENIED + split = bearer_token_header( + permissions=["reports:read"], org_id=ORG, org_permissions=["reports:write"] + ) + assert bearer_stack.status("all", split) == ALLOWED + + instance = bearer_token_header(org_id=ORG, permissions=["projects:write"]) + assert bearer_stack.status("org", instance) == DENIED + no_org = bearer_token_header(org_permissions=["projects:write"]) + assert bearer_stack.status("org", no_org) == DENIED + + org = bearer_token_header(org_id=ORG, org_permissions=["projects:write"]) + assert bearer_stack.status("org", org) == ALLOWED + assert bearer_stack.status(f"org/{ORG}", org) == ALLOWED + assert bearer_stack.status(f"org/{OTHER_ORG}", org) == DENIED + assert bearer_stack.status("unbound", org) == DENIED + + +@pytest.mark.parametrize("route", ["empty", "empty-all", "empty-org"]) +def test_verify_only_an_empty_list_denies(bearer_stack: Stack, route: str) -> None: + held = bearer_token_header(permissions=["x"], org_id=ORG, org_permissions=["x"]) + assert bearer_stack.status(route, held) == DENIED + + +def test_verify_only_malformed_claims_read_as_empty(bearer_stack: Stack) -> None: + for value in ("reports:read", {"a": 1}, 1, [1, None]): + assert bearer_stack.status("any", bearer_token_header(permissions=value)) == DENIED + mixed = bearer_token_header(permissions=[1, "reports:read"]) + assert bearer_stack.status("any", mixed) == ALLOWED + + +def test_verify_only_omitted_denies_and_reads_nothing(bearer_stack: Stack, keys: KeyServer) -> None: + held = bearer_token_header(permissions_omitted=True, permissions=["reports:read"]) + assert bearer_stack.status("any", held) == DENIED + assert set(keys.paths) == {"/jwks"} + + +def test_verify_only_exposes_the_fields(keys: KeyServer) -> None: + server = ResourceServer(**keys.options()) + user = server.authenticate( + "Bearer " + + provider_token( + permissions=["a", 1], + org_id=ORG, + org_roles=["admin"], + org_permissions=["b"], + permissions_omitted=True, + ) + ) + assert user.permissions == ("a",) + assert user.organization_roles == ("admin",) + assert user.organization_permissions == ("b",) + assert user.permissions_omitted is True + assert user.org_id == ORG + + bare = server.authenticate("Bearer " + provider_token()) + assert bare.permissions == bare.organization_roles == bare.organization_permissions == () + assert bare.permissions_omitted is False + + +def test_verify_only_a_real_provider_token_denies_today() -> None: + server = Provider().server() + guard = require_permission("orders:read") + user = server.authenticate(f"Bearer {TOKENS['user_passkey']}") + with pytest.raises(Exception, match="insufficient_permission") as refused: + guard.check(user) + assert refused.value.status == 403 # type: ignore[attr-defined] + assert refused.value.body == {"error": "insufficient_permission"} # type: ignore[attr-defined] + + +def test_verify_only_a_bound_guard_checked_without_a_route_denies(keys: KeyServer) -> None: + server = ResourceServer(**keys.options()) + user = server.authenticate("Bearer " + provider_token(org_id=ORG, org_permissions=["p"])) + with pytest.raises(Exception, match="insufficient_permission"): + require_permission("p", organization_param="org").check(user) + + +def test_verify_only_setup_is_checked() -> None: + with pytest.raises(ValueError, match="mode"): + require_permission("a", mode="some") # type: ignore[arg-type] + with pytest.raises(ValueError, match="non-empty"): + require_permission("") + + +def test_guards_without_route_needs_are_unchanged(keys: KeyServer) -> None: + server = ResourceServer(**keys.options()) + user = server.authenticate("Bearer " + provider_token(roles=["admin"])) + require_role("admin").check_route(user, {"anything": 1}) diff --git a/tests/test_permissions_django_urls.py b/tests/test_permissions_django_urls.py new file mode 100644 index 0000000..89fee02 --- /dev/null +++ b/tests/test_permissions_django_urls.py @@ -0,0 +1,84 @@ +"""The URLconf the Django permission guard tests run against: each route as a sync +and as an async view, and the verify-only guard under ``bearer/``.""" + +from typing import Any + +from django.http import HttpRequest, JsonResponse +from django.urls import path + +from seamless_auth.django import ( + bearer_user_of, + require_auth, + require_bearer, + require_permission, + user_of, +) + + +def ok(request: HttpRequest, **_: Any) -> JsonResponse: + return JsonResponse({"id": user_of(request).id}) + + +async def ok_async(request: HttpRequest, **_: Any) -> JsonResponse: + return JsonResponse({"id": user_of(request).id}) + + +def bearer_ok(request: HttpRequest, **_: Any) -> JsonResponse: + return JsonResponse({"id": bearer_user_of(request).sub}) + + +async def bearer_ok_async(request: HttpRequest, **_: Any) -> JsonResponse: + return JsonResponse({"id": bearer_user_of(request).sub}) + + +GUARDS: dict[str, dict[str, Any]] = { + "any": {"permissions": ["reports:read", "reports:export"]}, + "all": {"permissions": ["reports:read", "reports:write"], "permission_mode": "all"}, + "org": {"permissions": ["projects:write"], "organization": True}, + "org-all": { + "permissions": ["projects:read", "projects:write"], + "permission_mode": "all", + "organization": True, + }, + "org/": { + "permissions": ["projects:write"], + "organization_param": "organization_id", + }, + "org-uuid/": { + "permissions": ["projects:write"], + "organization_param": "organization_id", + }, + "unbound": {"permissions": ["projects:write"], "organization_param": "organization_id"}, + "empty": {"permissions": []}, + "empty-all": {"permissions": [], "permission_mode": "all"}, + "empty-org": {"permissions": [], "organization": True}, + "recovery": {"permissions": ["reports:read"], "allow_recovery": True}, +} + + +def _bearer_guard(options: dict[str, Any]) -> Any: + return require_permission( + *options["permissions"], + mode=options.get("permission_mode", "any"), + organization=options.get("organization", False), + organization_param=options.get("organization_param"), + ) + + +urlpatterns = [ + *(path(f"sync/{route}", require_auth(**options)(ok)) for route, options in GUARDS.items()), + *( + path(f"async/{route}", require_auth(**options)(ok_async)) + for route, options in GUARDS.items() + ), + *( + path(f"bearer/{route}", require_bearer(_bearer_guard(options)(bearer_ok))) + for route, options in GUARDS.items() + if "allow_recovery" not in options + ), + *( + path(f"bearer-async/{route}", require_bearer(_bearer_guard(options)(bearer_ok_async))) + for route, options in GUARDS.items() + if "allow_recovery" not in options + ), +]