diff --git a/backend/src/infrastructure/auth/dependencies.py b/backend/src/infrastructure/auth/dependencies.py index cf1d10e1..3ce2df00 100644 --- a/backend/src/infrastructure/auth/dependencies.py +++ b/backend/src/infrastructure/auth/dependencies.py @@ -1,4 +1,4 @@ -"""Auth dependencies: resolve the crudauth ``Principal`` and the dict-compat user. +"""Auth dependencies: resolve the crudauth ``Principal``, the dict-compat user, and permissions. Routes depend on these; they wrap the crudauth ``auth`` singleton so the session engine (validation, CSRF, lockout) lives in crudauth while handlers keep their @@ -7,13 +7,17 @@ contract is unchanged. """ +from collections.abc import Collection from typing import Annotated, Any from crudauth import Principal from crudauth.exceptions import ForbiddenException, UnauthorizedException from fastapi import Depends +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from ...modules.role.models import RolePermission, UserRole +from ...modules.role.permission_registry import all_permissions from ...modules.user.crud import crud_users from ..database.session import async_session from .setup import auth @@ -86,3 +90,132 @@ async def get_current_superuser( raise ForbiddenException("Insufficient privileges") return current_user + + +async def load_permissions(db: AsyncSession, user_id: int, *, is_superuser: bool = False) -> frozenset[str]: + """The permissions a user effectively holds, through the roles assigned to them. + + Reads through the session the request already uses, so it sees the same data + (and the same database under test overrides) as the route it authorizes. + Stored names that are no longer registered are ignored, so removing a + permission from the code can't grant anything. + + Args: + db: Database session for the lookup. + user_id: The user whose roles to read. + is_superuser: When true, every registered permission is granted without a query. + + Returns: + The registered permission names the user holds. + """ + registered = all_permissions() + + if is_superuser: + return registered + + if not registered: + return frozenset() + + statement = ( + select(RolePermission.permission_name) + .join(UserRole, UserRole.role_id == RolePermission.role_id) + .where( + UserRole.user_id == user_id, + RolePermission.permission_name.in_(registered), + ) + ) + result = await db.execute(statement) + + return frozenset(result.scalars().all()) + + +async def get_current_permissions( + principal: Annotated[Principal, Depends(get_current_principal)], + db: Annotated[AsyncSession, Depends(async_session)], +) -> frozenset[str]: + """The current user's effective permissions, resolved once per request. + + FastAPI caches a dependency's result for the length of a request, so however + many ``require_permissions`` guards and route parameters ask for permissions, + the join runs once. + """ + return await load_permissions(db, principal.user_id, is_superuser=principal.is_superuser) + + +def require_permissions(*needed: str) -> Any: + """A dependency that answers 403 unless the caller holds every named permission. + + Superusers pass without a lookup. Unknown names are a programming error and + raise at import time, when the route is declared, rather than at request time. + + Example: + ```python + @router.get("/", dependencies=[require_permissions("user.read")]) + async def get_users(...): ... + ``` + """ + required = frozenset(needed) + unknown = required - all_permissions() + + if unknown: + raise ValueError(f"Unknown permission name(s): {', '.join(sorted(unknown))}") + + async def dependency( + principal: Annotated[Principal, Depends(get_current_principal)], + permissions: Annotated[frozenset[str], Depends(get_current_permissions)], + ) -> Principal: + if principal.is_superuser or required <= permissions: + return principal + + raise ForbiddenException("Insufficient permissions") + + return Depends(dependency) + + +async def can_delegate_permissions( + db: AsyncSession, + principal: Principal, + permissions: Collection[str], +) -> bool: + """Whether a principal may hand out every one of these permissions. + + This is the escalation check only: a caller can't grant what they don't hold. + A route that changes a role must still require ``role.update``, and one that + assigns a role must still require ``role.assign``. + + Unregistered names are never delegable, so a typo or a removed permission + can't slip through. + """ + needed = set(permissions) + + if needed - all_permissions(): + return False + + if principal.is_superuser: + return True + + held = await load_permissions(db, principal.user_id) + + return needed <= held + + +async def can_assign_role(db: AsyncSession, principal: Principal, role_id: int) -> bool: + """Whether a principal may assign this role to someone. + + The escalation check for role assignment: every permission the role carries + must be one the caller already holds. Stored names that are no longer + registered are ignored, the same way ``load_permissions`` ignores them. + """ + if principal.is_superuser: + return True + + statement = select(RolePermission.permission_name).where(RolePermission.role_id == role_id) + result = await db.execute(statement) + carried = set(result.scalars().all()) & all_permissions() + + if not carried: + return True + + held = await load_permissions(db, principal.user_id) + + return carried <= held diff --git a/backend/src/infrastructure/dependencies.py b/backend/src/infrastructure/dependencies.py index 0fda24ee..21338f46 100644 --- a/backend/src/infrastructure/dependencies.py +++ b/backend/src/infrastructure/dependencies.py @@ -1,15 +1,28 @@ from typing import Annotated, Any +from crudauth import Principal from fastapi import Depends from fastapi.security import OAuth2PasswordRequestForm from sqlalchemy.ext.asyncio import AsyncSession -from .auth.dependencies import get_current_superuser, get_current_user, get_optional_user +from .auth.dependencies import ( + get_current_permissions, + get_current_principal, + get_current_superuser, + get_current_user, + get_optional_user, +) from .database.session import async_session # Database AsyncSessionDep = Annotated[AsyncSession, Depends(async_session)] +# Authentication +CurrentPrincipalDep = Annotated[Principal, Depends(get_current_principal)] + +# Authorization (effective permissions, resolved once per request) +CurrentPermissionsDep = Annotated[frozenset[str], Depends(get_current_permissions)] + # Users (dict-compat, resolved by crudauth) CurrentUserDep = Annotated[dict[str, Any], Depends(get_current_user)] CurrentSuperUserDep = Annotated[dict[str, Any], Depends(get_current_superuser)] diff --git a/backend/src/interfaces/admin/views/users.py b/backend/src/interfaces/admin/views/users.py index 13331514..47141d4e 100644 --- a/backend/src/interfaces/admin/views/users.py +++ b/backend/src/interfaces/admin/views/users.py @@ -11,7 +11,7 @@ from ....infrastructure.database.session import local_session from ....modules.user.enums import OAuthProvider from ....modules.user.models import User -from ....modules.user.schemas import UserUpdate +from ....modules.user.schemas import UserAdminUpdate from ....modules.user.service import UserService from ..mixins import DataclassModelMixin @@ -41,7 +41,7 @@ class UserAdmin(DataclassModelMixin, ModelView, model=User): column_labels = {"hashed_password": "Password"} form_create_rules = ["name", "username", "email", "hashed_password", "tier_id", "is_superuser"] - form_edit_rules = [*UserUpdate.model_fields.keys(), "tier_id", "is_superuser"] + form_edit_rules = [*UserAdminUpdate.model_fields.keys(), "tier_id", "is_superuser"] form_overrides = {"oauth_provider": SelectField} form_args = {"oauth_provider": {"choices": OAUTH_PROVIDER_CHOICES}} diff --git a/backend/src/modules/__init__.py b/backend/src/modules/__init__.py index 186d38e1..63583bf1 100644 --- a/backend/src/modules/__init__.py +++ b/backend/src/modules/__init__.py @@ -2,9 +2,13 @@ from .api_keys.models import APIKey, KeyPermission, KeyUsage from .rate_limit.models import RateLimit +from .role.models import Role, RolePermission, UserRole +from .role.permission_registry import discover_permissions from .tier.models import Tier from .user.models import User +discover_permissions() + __all__ = [ "User", "Tier", @@ -12,4 +16,7 @@ "APIKey", "KeyUsage", "KeyPermission", + "Role", + "RolePermission", + "UserRole", ] diff --git a/backend/src/modules/role/__init__.py b/backend/src/modules/role/__init__.py new file mode 100644 index 00000000..52351bba --- /dev/null +++ b/backend/src/modules/role/__init__.py @@ -0,0 +1,30 @@ +"""Role module for role-based access control.""" + +from .constants import PERMISSION_NAME_MAX_LENGTH, ROLE_NAME_MAX_LENGTH +from .models import Role, RolePermission, UserRole +from .permission_registry import ( + all_permissions, + discover_permissions, + is_known_permission, + permission_groups, + register_permissions, +) +from .permissions import RolePermissionName + +__all__ = [ + # Models + "Role", + "RolePermission", + "UserRole", + # Permissions + "RolePermissionName", + # Registry + "all_permissions", + "discover_permissions", + "is_known_permission", + "permission_groups", + "register_permissions", + # Limits + "PERMISSION_NAME_MAX_LENGTH", + "ROLE_NAME_MAX_LENGTH", +] diff --git a/backend/src/modules/role/constants.py b/backend/src/modules/role/constants.py new file mode 100644 index 00000000..683d9952 --- /dev/null +++ b/backend/src/modules/role/constants.py @@ -0,0 +1,4 @@ +"""Shared limits for the RBAC models and the permission registry.""" + +ROLE_NAME_MAX_LENGTH = 50 +PERMISSION_NAME_MAX_LENGTH = 100 diff --git a/backend/src/modules/role/models.py b/backend/src/modules/role/models.py new file mode 100644 index 00000000..f62da551 --- /dev/null +++ b/backend/src/modules/role/models.py @@ -0,0 +1,123 @@ +"""RBAC models: roles, their permissions, and the users assigned to them.""" + +from typing import TYPE_CHECKING + +from sqlalchemy import ForeignKey, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column, relationship, validates + +from ...infrastructure.database.models import TimestampMixin +from ...infrastructure.database.session import Base +from .constants import PERMISSION_NAME_MAX_LENGTH, ROLE_NAME_MAX_LENGTH +from .permission_registry import is_known_permission + +if TYPE_CHECKING: + from ..user.models import User + + +class Role(Base, TimestampMixin): + """Reusable named role that holds a set of permission strings.""" + + __tablename__ = "roles" + + id: Mapped[int] = mapped_column( + Integer, + primary_key=True, + autoincrement=True, + init=False, + ) + name: Mapped[str] = mapped_column( + String(ROLE_NAME_MAX_LENGTH), + nullable=False, + unique=True, + ) + description: Mapped[str | None] = mapped_column(Text, default=None) + + permissions: Mapped[list["RolePermission"]] = relationship( + "RolePermission", + back_populates="role", + lazy="select", + cascade="all, delete-orphan", + passive_deletes=True, + default_factory=list, + init=False, + ) + user_roles: Mapped[list["UserRole"]] = relationship( + "UserRole", + back_populates="role", + lazy="select", + cascade="all, delete-orphan", + passive_deletes=True, + default_factory=list, + init=False, + ) + + def __repr__(self) -> str: + return self.name + + +class RolePermission(Base, TimestampMixin): + """Maps a role to a permission name.""" + + __tablename__ = "role_permissions" + + role_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("roles.id", ondelete="CASCADE"), + primary_key=True, + ) + permission_name: Mapped[str] = mapped_column( + String(PERMISSION_NAME_MAX_LENGTH), + primary_key=True, + ) + + role: Mapped["Role"] = relationship( + "Role", + back_populates="permissions", + lazy="select", + init=False, + ) + + @validates("permission_name") + def validate_permission_name(self, key: str, value: str) -> str: + """Reject permission names that are not registered.""" + + if not is_known_permission(value): + raise ValueError(f"Unknown permission name: {value}") + + return value + + def __repr__(self) -> str: + return f"{self.role_id}:{self.permission_name}" + + +class UserRole(Base, TimestampMixin): + """Maps a user to a role.""" + + __tablename__ = "user_roles" + + user_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("user.id", ondelete="CASCADE"), + primary_key=True, + ) + role_id: Mapped[int] = mapped_column( + Integer, + ForeignKey("roles.id", ondelete="CASCADE"), + primary_key=True, + ) + + user: Mapped["User"] = relationship( + "User", + back_populates="user_roles", + lazy="select", + init=False, + ) + role: Mapped["Role"] = relationship( + "Role", + back_populates="user_roles", + lazy="select", + init=False, + ) + + def __repr__(self) -> str: + return f"user={self.user_id} role={self.role_id}" diff --git a/backend/src/modules/role/permission_registry.py b/backend/src/modules/role/permission_registry.py new file mode 100644 index 00000000..2c86b33b --- /dev/null +++ b/backend/src/modules/role/permission_registry.py @@ -0,0 +1,100 @@ +"""Central registry for module-defined permissions. + +Each module declares its own permissions as a ``StrEnum`` decorated with +``@register_permissions("")``. The registry is what validates a stored +permission name and what the admin UI groups by resource. +""" + +import importlib +import pkgutil +import re +from enum import StrEnum + +from .constants import PERMISSION_NAME_MAX_LENGTH + +PERMISSIONS_PACKAGE = "src.modules" + +RESOURCE_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$") +ACTION_PATTERN = re.compile(r"^[a-z][a-z0-9_]*$") + +_registered_permissions: dict[str, type[StrEnum]] = {} +_known_permissions: frozenset[str] = frozenset() +_discovered = False + + +def _validate(resource: str, enum_class: type[StrEnum]) -> None: + """Reject a registration that the stored permission column can't hold or a check can't match.""" + if not (isinstance(enum_class, type) and issubclass(enum_class, StrEnum)): + raise TypeError(f"Permissions for resource '{resource}' must be a StrEnum subclass.") + + if resource in _registered_permissions: + raise ValueError(f"Permissions for resource '{resource}' are already registered.") + + if not RESOURCE_PATTERN.match(resource): + raise ValueError(f"Resource '{resource}' must match {RESOURCE_PATTERN.pattern}.") + + members = list(enum_class) + if not members: + raise ValueError(f"Permissions for resource '{resource}' must define at least one permission.") + + for permission in members: + value = permission.value + resource_name, separator, action = value.partition(".") + + if not separator or resource_name != resource: + raise ValueError(f"Permission '{value}' must start with '{resource}.'.") + + if not ACTION_PATTERN.match(action): + raise ValueError(f"Permission '{value}' must name one action matching {ACTION_PATTERN.pattern}.") + + if len(value) > PERMISSION_NAME_MAX_LENGTH: + raise ValueError(f"Permission '{value}' is longer than {PERMISSION_NAME_MAX_LENGTH} characters.") + + +def register_permissions(resource: str): + """Register the permission enum for a resource.""" + + def decorator(enum_class: type[StrEnum]) -> type[StrEnum]: + global _known_permissions + + _validate(resource, enum_class) + + _registered_permissions[resource] = enum_class + _known_permissions |= {permission.value for permission in enum_class} + + return enum_class + + return decorator + + +def all_permissions() -> frozenset[str]: + """Return every registered permission name.""" + return _known_permissions + + +def permission_groups() -> dict[str, tuple[str, ...]]: + """Return permissions grouped by resource, for a UI that offers them per resource.""" + return { + resource: tuple(permission.value for permission in enum_class) + for resource, enum_class in _registered_permissions.items() + } + + +def is_known_permission(permission_name: str) -> bool: + """Return whether a permission is registered.""" + return permission_name in _known_permissions + + +def discover_permissions(package_name: str = PERMISSIONS_PACKAGE) -> None: + """Import every permissions module below the given package, once.""" + global _discovered + + if _discovered: + return + + _discovered = True + package = importlib.import_module(package_name) + + for _, module_name, _ in pkgutil.walk_packages(package.__path__, package.__name__ + "."): + if module_name.endswith(".permissions"): + importlib.import_module(module_name) diff --git a/backend/src/modules/role/permissions.py b/backend/src/modules/role/permissions.py new file mode 100644 index 00000000..80474170 --- /dev/null +++ b/backend/src/modules/role/permissions.py @@ -0,0 +1,16 @@ +"""Role module permissions.""" + +from enum import StrEnum + +from .permission_registry import register_permissions + + +@register_permissions("role") +class RolePermissionName(StrEnum): + """Permissions for role resources.""" + + READ = "role.read" + CREATE = "role.create" + UPDATE = "role.update" + DELETE = "role.delete" + ASSIGN = "role.assign" diff --git a/backend/src/modules/tier/permissions.py b/backend/src/modules/tier/permissions.py new file mode 100644 index 00000000..11be4f8e --- /dev/null +++ b/backend/src/modules/tier/permissions.py @@ -0,0 +1,15 @@ +"""Tier module permissions.""" + +from enum import StrEnum + +from ..role.permission_registry import register_permissions + + +@register_permissions("tier") +class TierPermission(StrEnum): + """Permissions for tier resources.""" + + READ = "tier.read" + CREATE = "tier.create" + UPDATE = "tier.update" + DELETE = "tier.delete" diff --git a/backend/src/modules/user/models.py b/backend/src/modules/user/models.py index cb651e03..528596fc 100644 --- a/backend/src/modules/user/models.py +++ b/backend/src/modules/user/models.py @@ -9,6 +9,7 @@ from .constants import NAME_MAX_LENGTH, USERNAME_MAX_LENGTH if TYPE_CHECKING: + from ..role.models import UserRole from ..tier.models import Tier @@ -49,6 +50,15 @@ class User(Base, TimestampMixin, SoftDeleteMixin): oauth_updated_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), default=None) tier: Mapped["Tier | None"] = relationship("Tier", back_populates="users", lazy="selectin", init=False) + user_roles: Mapped[list["UserRole"]] = relationship( + "UserRole", + back_populates="user", + lazy="select", + cascade="all, delete-orphan", + passive_deletes=True, + default_factory=list, + init=False, + ) @property def is_active(self) -> bool: diff --git a/backend/src/modules/user/permissions.py b/backend/src/modules/user/permissions.py new file mode 100644 index 00000000..9ba04e0c --- /dev/null +++ b/backend/src/modules/user/permissions.py @@ -0,0 +1,15 @@ +"""User module permissions.""" + +from enum import StrEnum + +from ..role.permission_registry import register_permissions + + +@register_permissions("user") +class UserPermission(StrEnum): + """Permissions for user resources.""" + + READ = "user.read" + CREATE = "user.create" + UPDATE = "user.update" + DELETE = "user.delete" diff --git a/backend/src/modules/user/routes.py b/backend/src/modules/user/routes.py index 12655f81..78f639af 100644 --- a/backend/src/modules/user/routes.py +++ b/backend/src/modules/user/routes.py @@ -3,8 +3,10 @@ from fastapi import APIRouter from fastcrud import PaginatedListResponse, compute_offset, paginated_response +from ...infrastructure.auth.dependencies import load_permissions, require_permissions from ...infrastructure.dependencies import ( AsyncSessionDep, + CurrentPermissionsDep, CurrentSuperUserDep, CurrentUserDep, ) @@ -55,28 +57,31 @@ async def create_user( @router.get( "/", response_model=PaginatedListResponse[UserRead], - summary="List All Users (Admin)", + summary="List All Users", description=""" Retrieves a paginated list of all users in the system. - This admin-only endpoint provides access to all user accounts and supports - pagination to handle large numbers of users efficiently. The results include - basic profile information for each user. + Requires the `user.read` permission, which a role grants; superusers + always pass. The results include basic profile information for each + user. For security reasons, sensitive information like passwords is never included in the response. """, - responses={401: {"description": "Not authenticated"}, 403: {"description": "Not authorized - requires admin privileges"}}, + responses={ + 401: {"description": "Not authenticated"}, + 403: {"description": "Not authorized - requires the user.read permission"}, + }, response_description="A paginated list of users with total count and pagination metadata", + dependencies=[require_permissions("user.read")], ) async def get_users( db: AsyncSessionDep, - _: CurrentSuperUserDep, user_service: UserServiceDep, page: int = 1, items_per_page: int = 10, ) -> dict[str, Any]: - """Get paginated list of all users (admin only).""" + """Get paginated list of users.""" users_data = await user_service.get_paginated(skip=compute_offset(page, items_per_page), limit=items_per_page, db=db) return paginated_response(crud_data=users_data, page=page, items_per_page=items_per_page) @@ -171,13 +176,17 @@ async def get_active_and_inactive_user_by_username( description=""" Updates a user's profile information. - This endpoint allows users to modify their own profile data or administrators - to modify any user's data. Only the fields provided in the request will be - updated, and all fields are optional. + This endpoint allows users to modify their own profile data, and users + holding the `user.update` permission to modify another user's data. Only + the fields provided in the request will be updated, and all fields are + optional. Permission rules: - Regular users can only update their own profiles - - Administrators can update any user's profile + - Superusers can update any user's profile + - A `user.update` holder can update another user only when that user is + not a superuser and holds no permission the requester lacks, and cannot + change another user's email address - Note: Tier updates are handled by a separate endpoint (/users/{username}/tier) Username and email changes are validated to ensure uniqueness. @@ -195,13 +204,18 @@ async def update_user_profile( username: str, values: UserUpdate, current_user: CurrentUserDep, + permissions: CurrentPermissionsDep, db: AsyncSessionDep, user_service: UserServiceDep, ) -> dict[str, str]: """Update user profile information.""" - await user_service.verify_user_permission(current_user, username, "update profile") + await user_service.verify_update_permission(current_user, username, permissions) user = await user_service.get_by_username(username, db) + if not user_service.is_self_or_superuser(current_user, user["username"]): + target_permissions = await load_permissions(db, user["id"]) + user_service.verify_no_privilege_escalation(user, values, permissions, target_permissions) + await user_service.update(user["id"], values, db) return {"message": "User updated successfully"} diff --git a/backend/src/modules/user/schemas.py b/backend/src/modules/user/schemas.py index bbe4c3dd..1ce1bd12 100644 --- a/backend/src/modules/user/schemas.py +++ b/backend/src/modules/user/schemas.py @@ -133,6 +133,15 @@ class UserUpdate(BaseModel): default=None, ), ] + + +class UserAdminUpdate(UserUpdate): + """Schema for updates only an administrator may make. + + The OAuth identifiers and the verification flag decide who a provider login + resolves to, so they are not part of the public profile update. + """ + google_id: str | None = None github_id: str | None = None oauth_provider: str | None = None @@ -140,7 +149,7 @@ class UserUpdate(BaseModel): oauth_updated_at: datetime | None = None -class UserUpdateInternal(UserUpdate): +class UserUpdateInternal(UserAdminUpdate): """Internal schema for user updates.""" updated_at: datetime diff --git a/backend/src/modules/user/service.py b/backend/src/modules/user/service.py index b62b162e..df12658c 100644 --- a/backend/src/modules/user/service.py +++ b/backend/src/modules/user/service.py @@ -1,3 +1,4 @@ +from collections.abc import Collection from datetime import UTC, datetime from typing import Any, cast @@ -24,6 +25,7 @@ from ..tier.schemas import TierRead from .crud import crud_users from .models import User +from .permissions import UserPermission from .schemas import ( User as UserSchema, ) @@ -341,6 +343,74 @@ async def verify_user_permission( if not has_permission: raise PermissionDeniedError(f"You don't have permission to {action_description} on this user") + async def verify_update_permission( + self, + requester_user: dict[str, Any], + target_username: str, + permissions: Collection[str], + ) -> None: + """Verify the requester may update this profile at all. + + Editing your own profile stays the ownership rule. Editing someone else's + needs the ``user.update`` permission, or superuser. + + Args: + requester_user: User data of the user making the request. + target_username: Username of the user to be updated. + permissions: Effective permissions held by the requester. + + Raises: + PermissionDeniedError: If the requester may not update this profile. + """ + if self.is_self_or_superuser(requester_user, target_username): + return + + if UserPermission.UPDATE.value in permissions: + return + + raise PermissionDeniedError("You don't have permission to update profile on this user") + + def is_self_or_superuser(self, requester_user: dict[str, Any], target_username: str) -> bool: + """Whether the requester owns this profile or may act on any of them.""" + if requester_user.get("is_superuser", False): + return True + + return requester_user.get("username") == target_username + + def verify_no_privilege_escalation( + self, + target_user: dict[str, Any], + values: UserUpdate, + requester_permissions: Collection[str], + target_permissions: Collection[str], + ) -> None: + """Verify a ``user.update`` holder isn't editing their way to more access. + + Editing an account is a way to take it over, so a non-superuser holding + ``user.update`` may only edit accounts weaker than their own: not a + superuser, and holding nothing the requester doesn't already hold. + The email address is how a verified provider login is matched to an + existing account, so only a superuser may change someone else's. + + Args: + target_user: The user being updated. + values: The submitted changes. + requester_permissions: Effective permissions held by the requester. + target_permissions: Effective permissions held by the target. + + Raises: + PermissionDeniedError: If the edit would reach a stronger account. + """ + if target_user.get("is_superuser", False): + raise PermissionDeniedError("You don't have permission to update profile on this user") + + if not set(target_permissions) <= set(requester_permissions): + raise PermissionDeniedError("You don't have permission to update profile on this user") + + submitted = values.model_dump(exclude_unset=True) + if "email" in submitted and submitted["email"] != target_user.get("email"): + raise PermissionDeniedError("Only a superuser can change another user's email address") + async def delete(self, user_id: int, db: AsyncSession) -> None: """Soft delete a user. diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 17f2784a..5fe8479a 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -27,7 +27,7 @@ import pytest_asyncio # noqa: E402 import redis as syncredis # noqa: E402 import redis.asyncio as aioredis # noqa: E402 -from crudauth import get_password_hash # noqa: E402 +from crudauth import Principal, get_password_hash # noqa: E402 from faker import Faker # noqa: E402 from httpx import ASGITransport, AsyncClient # noqa: E402 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine # noqa: E402 @@ -37,7 +37,11 @@ # mypy: disable-error-code="import-untyped" from testcontainers.postgres import PostgresContainer # noqa: E402 -from src.infrastructure.auth.dependencies import get_current_superuser, get_current_user # noqa: E402 +from src.infrastructure.auth.dependencies import ( # noqa: E402 + get_current_principal, + get_current_superuser, + get_current_user, +) from src.infrastructure.config.settings import get_settings # noqa: E402 from src.infrastructure.database.session import Base, async_session # noqa: E402 from src.interfaces.main import app # noqa: E402 @@ -231,6 +235,20 @@ async def test_superuser(db_session: AsyncSession, test_tier: dict): } +def _principal_for(user: dict) -> Principal: + """The crudauth principal the session transport would resolve for this user. + + The auth fixtures override the dict-compat dependencies, so anything reading + the principal directly (permission checks, session routes) needs it too. + """ + return Principal( + user_id=user["id"], + transport="session", + is_superuser=user.get("is_superuser", False), + email_verified=True, + ) + + @pytest_asyncio.fixture async def auth_client(client: AsyncClient, test_user: dict): """Authenticated test client (regular user) — overrides get_current_user dependency.""" @@ -238,7 +256,11 @@ async def auth_client(client: AsyncClient, test_user: dict): async def override_get_current_user(): return test_user + async def override_get_current_principal(): + return _principal_for(test_user) + app.dependency_overrides[get_current_user] = override_get_current_user + app.dependency_overrides[get_current_principal] = override_get_current_principal return client @@ -249,7 +271,11 @@ async def auth_client_2(client: AsyncClient, test_user_2: dict): async def override_get_current_user(): return test_user_2 + async def override_get_current_principal(): + return _principal_for(test_user_2) + app.dependency_overrides[get_current_user] = override_get_current_user + app.dependency_overrides[get_current_principal] = override_get_current_principal return client @@ -260,10 +286,14 @@ async def superuser_auth_client(client: AsyncClient, test_superuser: dict): async def override_get_current_user(): return test_superuser + async def override_get_current_principal(): + return _principal_for(test_superuser) + async def override_get_current_superuser(): return test_superuser app.dependency_overrides[get_current_user] = override_get_current_user + app.dependency_overrides[get_current_principal] = override_get_current_principal app.dependency_overrides[get_current_superuser] = override_get_current_superuser return client diff --git a/backend/tests/integration/api/v1/users/test_permissions.py b/backend/tests/integration/api/v1/users/test_permissions.py new file mode 100644 index 00000000..90bfc3fb --- /dev/null +++ b/backend/tests/integration/api/v1/users/test_permissions.py @@ -0,0 +1,289 @@ +"""RBAC on the user routes, exercised through a real login. + +These sign in through ``/api/v1/auth/login`` instead of overriding the auth +dependencies, so the permission lookup runs against the same session the route +uses and a guard that only holds in a mock can't pass. +""" + +from datetime import UTC, datetime + +import pytest +from crudauth import Principal +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.infrastructure.auth import dependencies as deps +from src.infrastructure.database.session import async_session +from src.infrastructure.dependencies import CurrentPermissionsDep +from src.modules.role.models import Role, RolePermission, UserRole +from src.modules.user.models import User + +pytestmark = pytest.mark.asyncio + + +async def _grant(db: AsyncSession, user_id: int, name: str, *permissions: str) -> Role: + """Give a user a role carrying these permissions.""" + role = Role(name=name) + db.add(role) + await db.flush() + db.add_all([RolePermission(role_id=role.id, permission_name=p) for p in permissions]) + db.add(UserRole(user_id=user_id, role_id=role.id)) + await db.commit() + return role + + +async def _login(client: AsyncClient, user: dict) -> str: + """Sign in for real and return the CSRF token for unsafe requests.""" + response = await client.post( + "/api/v1/auth/login", + data={"username": user["username"], "password": user["password"]}, + ) + assert response.status_code == 200, response.text + return response.json()["csrf_token"] + + +# ============================================================================= +# GET /users/ is behind user.read +# ============================================================================= +async def test_listing_users_needs_the_read_permission(client: AsyncClient, test_user: dict): + await _login(client, test_user) + + response = await client.get("/api/v1/users/") + + assert response.status_code == 403 + + +async def test_a_role_granting_read_lists_users(client: AsyncClient, db_session: AsyncSession, test_user: dict): + await _grant(db_session, test_user["id"], "reader", "user.read") + await _login(client, test_user) + + response = await client.get("/api/v1/users/") + + assert response.status_code == 200 + assert response.json()["data"] + + +async def test_a_superuser_lists_users_without_a_role(client: AsyncClient, test_superuser: dict): + await _login(client, test_superuser) + + response = await client.get("/api/v1/users/") + + assert response.status_code == 200 + + +async def test_listing_users_needs_authentication(client: AsyncClient): + assert (await client.get("/api/v1/users/")).status_code == 401 + + +# ============================================================================= +# PATCH /users/{username}: what user.update may reach +# ============================================================================= +async def test_an_update_holder_cannot_edit_a_superuser( + client: AsyncClient, db_session: AsyncSession, test_user: dict, test_superuser: dict +): + """Editing a superuser's profile would be a route to taking the account over.""" + await _grant(db_session, test_user["id"], "editor", "user.update") + csrf_token = await _login(client, test_user) + + response = await client.patch( + f"/api/v1/users/{test_superuser['username']}", + json={"name": "Pwned Name"}, + headers={"X-CSRF-Token": csrf_token}, + ) + + assert response.status_code == 403 + target = await db_session.get(User, test_superuser["id"]) + await db_session.refresh(target) + assert target.name == test_superuser["name"] + + +async def test_an_update_holder_cannot_edit_someone_holding_more( + client: AsyncClient, db_session: AsyncSession, test_user: dict, test_user_2: dict +): + await _grant(db_session, test_user["id"], "editor", "user.update") + await _grant(db_session, test_user_2["id"], "deleter", "user.delete") + csrf_token = await _login(client, test_user) + + response = await client.patch( + f"/api/v1/users/{test_user_2['username']}", + json={"name": "Pwned Name"}, + headers={"X-CSRF-Token": csrf_token}, + ) + + assert response.status_code == 403 + + +async def test_an_update_holder_can_edit_a_weaker_user( + client: AsyncClient, db_session: AsyncSession, test_user: dict, test_user_2: dict +): + await _grant(db_session, test_user["id"], "editor", "user.update") + csrf_token = await _login(client, test_user) + + response = await client.patch( + f"/api/v1/users/{test_user_2['username']}", + json={"name": "Renamed User"}, + headers={"X-CSRF-Token": csrf_token}, + ) + + assert response.status_code == 200 + target = await db_session.get(User, test_user_2["id"]) + await db_session.refresh(target) + assert target.name == "Renamed User" + + +async def test_an_update_holder_cannot_change_another_users_email( + client: AsyncClient, db_session: AsyncSession, test_user: dict, test_user_2: dict +): + """A verified provider email links a social login to an account, so this is superuser-only.""" + await _grant(db_session, test_user["id"], "editor", "user.update") + csrf_token = await _login(client, test_user) + + response = await client.patch( + f"/api/v1/users/{test_user_2['username']}", + json={"email": "attacker@example.com"}, + headers={"X-CSRF-Token": csrf_token}, + ) + + assert response.status_code == 403 + target = await db_session.get(User, test_user_2["id"]) + await db_session.refresh(target) + assert target.email == test_user_2["email"] + + +@pytest.mark.parametrize( + "payload", + [ + {"google_id": "attacker-google-sub"}, + {"github_id": "attacker-github-id"}, + {"oauth_provider": "google"}, + {"email_verified": True}, + {"oauth_updated_at": "2026-01-01T00:00:00Z"}, + ], +) +async def test_the_public_endpoint_refuses_the_oauth_fields( + client: AsyncClient, db_session: AsyncSession, test_user: dict, payload: dict +): + """Setting an OAuth identifier decides who a provider login resolves to.""" + csrf_token = await _login(client, test_user) + + response = await client.patch( + f"/api/v1/users/{test_user['username']}", + json=payload, + headers={"X-CSRF-Token": csrf_token}, + ) + + assert response.status_code == 422 + owner = await db_session.get(User, test_user["id"]) + await db_session.refresh(owner) + assert owner.google_id is None + assert owner.email_verified is False + + +async def test_a_user_still_edits_their_own_profile(client: AsyncClient, db_session: AsyncSession, test_user: dict): + csrf_token = await _login(client, test_user) + + response = await client.patch( + f"/api/v1/users/{test_user['username']}", + json={"name": "My New Name"}, + headers={"X-CSRF-Token": csrf_token}, + ) + + assert response.status_code == 200 + owner = await db_session.get(User, test_user["id"]) + await db_session.refresh(owner) + assert owner.name == "My New Name" + + +async def test_a_superuser_can_change_another_users_email( + client: AsyncClient, db_session: AsyncSession, test_superuser: dict, test_user_2: dict +): + csrf_token = await _login(client, test_superuser) + + response = await client.patch( + f"/api/v1/users/{test_user_2['username']}", + json={"email": "moved@example.com"}, + headers={"X-CSRF-Token": csrf_token}, + ) + + assert response.status_code == 200 + + +# ============================================================================= +# The permission lookup runs once per request +# ============================================================================= +async def test_permissions_are_loaded_once_per_request( + client: AsyncClient, db_session: AsyncSession, test_user: dict, monkeypatch +): + """``require_permissions`` and the route both ask for permissions; FastAPI caches the answer.""" + await _grant(db_session, test_user["id"], "reader", "user.read") + await _login(client, test_user) + + calls: list[int] = [] + original = deps.load_permissions + + async def counting_load_permissions(db, user_id, *, is_superuser=False): + calls.append(user_id) + return await original(db, user_id, is_superuser=is_superuser) + + monkeypatch.setattr(deps, "load_permissions", counting_load_permissions) + + response = await client.get("/api/v1/users/") + + assert response.status_code == 200 + assert calls == [test_user["id"]] + + +async def test_one_request_resolves_permissions_once_however_many_ask(db_session: AsyncSession, test_user: dict, monkeypatch): + """A guard and a handler that both need permissions share one lookup.""" + await _grant(db_session, test_user["id"], "reader", "user.read") + + calls: list[int] = [] + original = deps.load_permissions + + async def counting_load_permissions(db, user_id, *, is_superuser=False): + calls.append(user_id) + return await original(db, user_id, is_superuser=is_superuser) + + monkeypatch.setattr(deps, "load_permissions", counting_load_permissions) + + app = FastAPI() + + @app.get("/twice", dependencies=[deps.require_permissions("user.read")]) + async def twice(permissions: CurrentPermissionsDep) -> dict[str, int]: + return {"held": len(permissions)} + + app.dependency_overrides[async_session] = lambda: db_session + app.dependency_overrides[deps.get_current_principal] = lambda: Principal(user_id=test_user["id"]) + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as local_client: + response = await local_client.get("/twice") + + assert response.status_code == 200 + assert response.json() == {"held": 1} + assert calls == [test_user["id"]] + + +async def test_a_stale_stored_permission_grants_nothing(client: AsyncClient, db_session: AsyncSession, test_user: dict): + """A permission removed from the code is ignored rather than trusted.""" + role = Role(name="stale") + db_session.add(role) + await db_session.flush() + db_session.add(UserRole(user_id=test_user["id"], role_id=role.id)) + await db_session.commit() + await db_session.execute( + RolePermission.__table__.insert().values( + role_id=role.id, + permission_name="user.retired", + created_at=datetime.now(UTC), + ) + ) + await db_session.commit() + await _login(client, test_user) + + response = await client.get("/api/v1/users/") + + assert response.status_code == 403 + stored = await db_session.scalars(select(RolePermission.permission_name).where(RolePermission.role_id == role.id)) + assert list(stored) == ["user.retired"] diff --git a/backend/tests/unit/infrastructure/auth/test_dependencies.py b/backend/tests/unit/infrastructure/auth/test_dependencies.py index 95c10a4d..e4c8ee11 100644 --- a/backend/tests/unit/infrastructure/auth/test_dependencies.py +++ b/backend/tests/unit/infrastructure/auth/test_dependencies.py @@ -1,18 +1,24 @@ """Unit tests for the crudauth-backed auth dependencies. -The integration fixtures (`auth_client` / `superuser_auth_client`) override these -functions, so the real bodies — the dict re-load, the soft-delete filter, and the -superuser 403 branch — are only exercised here, called directly with a Principal -and a mocked ``crud_users.get``. +The auth fixtures in ``tests/conftest.py`` override ``get_current_user`` and +``get_current_principal`` rather than carrying a real session, so a route that +reads the principal or the caller's permissions keeps working under them. Tests +for the permission rules themselves sign in for real; see +``tests/integration/api/v1/users/test_permissions.py``. """ +from datetime import UTC, datetime from unittest.mock import AsyncMock, MagicMock, patch import pytest from crudauth import Principal from crudauth.exceptions import ForbiddenException, UnauthorizedException +from sqlalchemy import insert +from sqlalchemy.ext.asyncio import AsyncSession from src.infrastructure.auth import dependencies as deps +from src.modules.role.models import Role, RolePermission, UserRole +from src.modules.role.permission_registry import all_permissions @pytest.mark.asyncio @@ -26,15 +32,22 @@ async def test_get_current_user_missing_row_raises(): """A valid principal whose row is gone/soft-deleted re-loads to None → 401.""" with patch.object(deps.crud_users, "get", new=AsyncMock(return_value=None)): with pytest.raises(UnauthorizedException): - await deps.get_current_user(principal=Principal(user_id=1), db=MagicMock()) + await deps.get_current_user( + principal=Principal(user_id=1), + db=MagicMock(), + ) @pytest.mark.asyncio async def test_get_current_user_returns_dict_and_filters_soft_deleted(): user = {"id": 1, "username": "x", "is_superuser": False} mock_get = AsyncMock(return_value=user) + with patch.object(deps.crud_users, "get", new=mock_get): - result = await deps.get_current_user(principal=Principal(user_id=1), db=MagicMock()) + result = await deps.get_current_user( + principal=Principal(user_id=1), + db=MagicMock(), + ) assert result == user assert mock_get.call_args.kwargs.get("is_deleted") is False @@ -42,24 +55,146 @@ async def test_get_current_user_returns_dict_and_filters_soft_deleted(): @pytest.mark.asyncio async def test_get_optional_user_none_principal_returns_none(): - assert await deps.get_optional_user(principal=None, db=MagicMock()) is None + assert ( + await deps.get_optional_user( + principal=None, + db=MagicMock(), + ) + is None + ) @pytest.mark.asyncio async def test_get_optional_user_returns_dict(): user = {"id": 2} - with patch.object(deps.crud_users, "get", new=AsyncMock(return_value=user)): - result = await deps.get_optional_user(principal=Principal(user_id=2), db=MagicMock()) + + with patch.object( + deps.crud_users, + "get", + new=AsyncMock(return_value=user), + ): + result = await deps.get_optional_user( + principal=Principal(user_id=2), + db=MagicMock(), + ) + assert result == user @pytest.mark.asyncio async def test_get_current_superuser_denies_non_superuser(): with pytest.raises(ForbiddenException): - await deps.get_current_superuser(current_user={"id": 1, "is_superuser": False}) + await deps.get_current_superuser( + current_user={"id": 1, "is_superuser": False}, + ) @pytest.mark.asyncio async def test_get_current_superuser_allows_superuser(): user = {"id": 1, "is_superuser": True} + assert await deps.get_current_superuser(current_user=user) == user + + +# ============================================================================= +# Permission loading and the require_permissions gate, against a real database +# ============================================================================= +async def _role_with(db: AsyncSession, name: str, *permissions: str) -> Role: + role = Role(name=name) + db.add(role) + await db.flush() + db.add_all([RolePermission(role_id=role.id, permission_name=p) for p in permissions]) + await db.commit() + return role + + +async def test_load_permissions_reads_the_roles_assigned_to_the_user(db_session: AsyncSession, test_user: dict): + role = await _role_with(db_session, "reader", "user.read", "tier.read") + db_session.add(UserRole(user_id=test_user["id"], role_id=role.id)) + await db_session.commit() + + permissions = await deps.load_permissions(db_session, test_user["id"]) + + assert permissions == {"user.read", "tier.read"} + + +async def test_load_permissions_is_empty_without_a_role(db_session: AsyncSession, test_user: dict): + assert await deps.load_permissions(db_session, test_user["id"]) == frozenset() + + +async def test_load_permissions_grants_a_superuser_everything_without_a_query(db_session: AsyncSession): + permissions = await deps.load_permissions(db_session, 999999, is_superuser=True) + + assert permissions == all_permissions() + + +async def test_load_permissions_ignores_a_stored_name_that_is_no_longer_registered(db_session: AsyncSession, test_user: dict): + """A permission removed from the code must not keep granting anything.""" + role = await _role_with(db_session, "stale", "user.read") + await db_session.execute( + insert(RolePermission).values( + role_id=role.id, + permission_name="user.retired", + created_at=datetime.now(UTC), + ) + ) + db_session.add(UserRole(user_id=test_user["id"], role_id=role.id)) + await db_session.commit() + + assert await deps.load_permissions(db_session, test_user["id"]) == {"user.read"} + + +def test_require_permissions_rejects_unknown_permission(): + with pytest.raises(ValueError, match="Unknown permission name"): + deps.require_permissions("user.reed") + + +# ============================================================================= +# Escalation helpers +# ============================================================================= +async def test_a_principal_can_delegate_what_it_holds(db_session: AsyncSession, test_user: dict): + role = await _role_with(db_session, "editor", "user.read", "user.update") + db_session.add(UserRole(user_id=test_user["id"], role_id=role.id)) + await db_session.commit() + principal = Principal(user_id=test_user["id"]) + + assert await deps.can_delegate_permissions(db_session, principal, ["user.read"]) + assert not await deps.can_delegate_permissions(db_session, principal, ["user.read", "user.delete"]) + + +async def test_a_superuser_can_delegate_anything_registered(db_session: AsyncSession): + principal = Principal(user_id=1, is_superuser=True) + + assert await deps.can_delegate_permissions(db_session, principal, sorted(all_permissions())) + + +async def test_an_unregistered_permission_is_never_delegable(db_session: AsyncSession): + """A typo or a removed permission is refused rather than raising into the route.""" + superuser = Principal(user_id=1, is_superuser=True) + + assert not await deps.can_delegate_permissions(db_session, superuser, ["user.reed"]) + + +async def test_assigning_a_role_needs_every_permission_it_carries(db_session: AsyncSession, test_user: dict): + held = await _role_with(db_session, "held", "user.read") + stronger = await _role_with(db_session, "stronger", "user.read", "user.delete") + weaker = await _role_with(db_session, "weaker", "user.read") + db_session.add(UserRole(user_id=test_user["id"], role_id=held.id)) + await db_session.commit() + principal = Principal(user_id=test_user["id"]) + + assert await deps.can_assign_role(db_session, principal, weaker.id) + assert not await deps.can_assign_role(db_session, principal, stronger.id) + + +async def test_a_role_carrying_nothing_is_assignable(db_session: AsyncSession, test_user: dict): + empty = await _role_with(db_session, "empty") + principal = Principal(user_id=test_user["id"]) + + assert await deps.can_assign_role(db_session, principal, empty.id) + + +async def test_a_superuser_can_assign_any_role(db_session: AsyncSession): + strong = await _role_with(db_session, "strong", "user.delete") + + assert await deps.can_assign_role(db_session, Principal(user_id=1, is_superuser=True), strong.id) diff --git a/backend/tests/unit/modules/role/__init__.py b/backend/tests/unit/modules/role/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/unit/modules/role/test_models.py b/backend/tests/unit/modules/role/test_models.py new file mode 100644 index 00000000..d6159213 --- /dev/null +++ b/backend/tests/unit/modules/role/test_models.py @@ -0,0 +1,149 @@ +"""Unit tests for the RBAC ORM models and permission configuration.""" + +import pytest +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from src.modules.role.models import Role, RolePermission, UserRole +from src.modules.role.permission_registry import all_permissions, is_known_permission +from src.modules.user.models import User + + +def test_registered_permissions_are_known(): + """Registered flat permission names are recognized.""" + permissions = all_permissions() + + assert "user.read" in permissions + assert "role.assign" in permissions + assert "tier.delete" in permissions + + assert is_known_permission("user.read") + assert is_known_permission("role.assign") + assert is_known_permission("tier.delete") + + +def test_unknown_permissions_are_not_known(): + """Unknown permission names are rejected by the registry.""" + assert not is_known_permission("user.reed") + assert not is_known_permission("unknown.permission") + assert not is_known_permission("user") + + +def test_role_permission_rejects_unknown_permission(): + """RolePermission must reject permission names outside the registry.""" + with pytest.raises(ValueError, match="Unknown permission name"): + RolePermission(role_id=1, permission_name="user.reed") + + +def test_role_permission_accepts_registered_permission(): + """RolePermission accepts a registered flat permission name.""" + permission = RolePermission(role_id=1, permission_name="user.read") + + assert permission.role_id == 1 + assert permission.permission_name == "user.read" + + +async def test_role_delete_cascades_to_permissions_and_user_roles( + db_session: AsyncSession, + test_user: dict, +): + """Deleting a role must remove its permission and user-role assignments.""" + role = Role( + name="test-role", + description="Test role", + ) + db_session.add(role) + await db_session.flush() + + role_permission = RolePermission( + role_id=role.id, + permission_name="user.read", + ) + user_role = UserRole( + user_id=test_user["id"], + role_id=role.id, + ) + + db_session.add_all([role_permission, user_role]) + await db_session.commit() + + role_id = role.id + + await db_session.delete(role) + await db_session.commit() + + role_permission_count = await db_session.scalar( + select(func.count()).select_from(RolePermission).where(RolePermission.role_id == role_id) + ) + user_role_count = await db_session.scalar(select(func.count()).select_from(UserRole).where(UserRole.role_id == role_id)) + + assert role_permission_count == 0 + assert user_role_count == 0 + + +async def test_a_role_name_cannot_be_reused(db_session: AsyncSession): + db_session.add(Role(name="duplicate")) + await db_session.commit() + db_session.add(Role(name="duplicate")) + + with pytest.raises(IntegrityError): + await db_session.commit() + + await db_session.rollback() + + +async def test_a_role_cannot_carry_the_same_permission_twice(db_session: AsyncSession): + role = Role(name="carrier") + db_session.add(role) + await db_session.flush() + db_session.add(RolePermission(role_id=role.id, permission_name="user.read")) + await db_session.commit() + db_session.add(RolePermission(role_id=role.id, permission_name="user.read")) + + with pytest.raises(IntegrityError): + await db_session.commit() + + await db_session.rollback() + + +async def test_a_user_cannot_be_assigned_the_same_role_twice(db_session: AsyncSession, test_user: dict): + role = Role(name="assignee") + db_session.add(role) + await db_session.flush() + db_session.add(UserRole(user_id=test_user["id"], role_id=role.id)) + await db_session.commit() + db_session.add(UserRole(user_id=test_user["id"], role_id=role.id)) + + with pytest.raises(IntegrityError): + await db_session.commit() + + await db_session.rollback() + + +async def test_a_role_assignment_needs_a_user_that_exists(db_session: AsyncSession): + role = Role(name="orphan") + db_session.add(role) + await db_session.flush() + db_session.add(UserRole(user_id=999999, role_id=role.id)) + + with pytest.raises(IntegrityError): + await db_session.commit() + + await db_session.rollback() + + +async def test_deleting_a_user_removes_their_role_assignments(db_session: AsyncSession, test_user: dict): + role = Role(name="departing") + db_session.add(role) + await db_session.flush() + db_session.add(UserRole(user_id=test_user["id"], role_id=role.id)) + await db_session.commit() + + user = await db_session.get(User, test_user["id"]) + await db_session.delete(user) + await db_session.commit() + + remaining = await db_session.scalar(select(func.count()).select_from(UserRole).where(UserRole.user_id == test_user["id"])) + + assert remaining == 0 diff --git a/backend/tests/unit/modules/role/test_permission_registry.py b/backend/tests/unit/modules/role/test_permission_registry.py new file mode 100644 index 00000000..d43457f4 --- /dev/null +++ b/backend/tests/unit/modules/role/test_permission_registry.py @@ -0,0 +1,163 @@ +"""The registry is what validates a stored permission name, so it has to be strict.""" + +import subprocess +import sys +from enum import StrEnum +from pathlib import Path + +import pytest + +from src.modules.role.constants import PERMISSION_NAME_MAX_LENGTH +from src.modules.role.permission_registry import ( + all_permissions, + permission_groups, + register_permissions, +) + + +def test_registered_permissions_are_flat_and_grouped(): + permissions = all_permissions() + groups = permission_groups() + + assert permissions + assert permissions == {permission for group in groups.values() for permission in group} + + assert "user.read" in permissions + assert "role.assign" in permissions + assert "tier.update" in permissions + + assert groups["user"] == ( + "user.read", + "user.create", + "user.update", + "user.delete", + ) + + +def test_register_permissions_rejects_wrong_resource_prefix(): + class InvalidPermission(StrEnum): + READ = "other.read" + + with pytest.raises(ValueError, match="must start with 'test_resource.'"): + register_permissions("test_resource")(InvalidPermission) + + +def test_register_permissions_rejects_duplicate_resource(): + class DuplicatePermission(StrEnum): + READ = "user.read" + + with pytest.raises( + ValueError, + match="Permissions for resource 'user' are already registered", + ): + register_permissions("user")(DuplicatePermission) + + +def test_register_permissions_rejects_a_nested_name(): + """Permissions are flat ``resource.action``; a deeper name has no meaning here.""" + + class NestedPermission(StrEnum): + READ = "billing.invoice.read" + + with pytest.raises(ValueError, match="one action"): + register_permissions("billing")(NestedPermission) + + +def test_register_permissions_rejects_a_dotted_resource(): + class DottedPermission(StrEnum): + READ = "billing.invoice.read" + + with pytest.raises(ValueError, match="must match"): + register_permissions("billing.invoice")(DottedPermission) + + +def test_register_permissions_rejects_an_empty_action(): + class EmptyActionPermission(StrEnum): + READ = "empty." + + with pytest.raises(ValueError, match="one action"): + register_permissions("empty")(EmptyActionPermission) + + +def test_register_permissions_rejects_a_name_the_column_cannot_hold(): + action = "a" * PERMISSION_NAME_MAX_LENGTH + + class LongPermission(StrEnum): + READ = f"long.{action}" + + with pytest.raises(ValueError, match="longer than"): + register_permissions("long")(LongPermission) + + +def test_register_permissions_rejects_uppercase_and_spaces(): + class WeirdPermission(StrEnum): + READ = "weird.Do Thing" + + with pytest.raises(ValueError, match="one action"): + register_permissions("weird")(WeirdPermission) + + +def test_register_permissions_rejects_an_empty_enum(): + class NoPermission(StrEnum): + pass + + with pytest.raises(ValueError, match="at least one permission"): + register_permissions("blank")(NoPermission) + + +def test_register_permissions_rejects_a_class_that_is_not_a_str_enum(): + class NotAnEnum: + READ = "plain.read" + + with pytest.raises(TypeError, match="StrEnum"): + register_permissions("plain")(NotAnEnum) # type: ignore[arg-type] + + +def test_no_resource_can_claim_a_name_another_resource_owns(): + """The prefix rule is what keeps one permission name owned by one resource.""" + + class ShadowPermission(StrEnum): + READ = "user.read" + OWN = "shadow.own" + + with pytest.raises(ValueError, match="must start with 'shadow.'"): + register_permissions("shadow")(ShadowPermission) + + +def test_a_failed_registration_leaves_the_registry_untouched(): + before = all_permissions() + + class BadPermission(StrEnum): + READ = "other.read" + + with pytest.raises(ValueError): + register_permissions("leftover")(BadPermission) + + assert all_permissions() == before + assert "leftover" not in permission_groups() + + +def test_a_route_module_imports_on_its_own(): + """``require_permissions`` validates at import time, so discovery has to have run. + + Importing a route module directly is how Alembic, a worker or a script reaches + the app, and it must not depend on something else importing ``src.modules`` first. + """ + backend = Path(__file__).resolve().parents[4] + environment = { + "PATH": "/usr/bin:/bin", + "PYTHONPATH": str(backend), + "SECRET_KEY": "test_secret_key_for_tests", + "SESSION_BACKEND": "memory", + "RATE_LIMITER_BACKEND": "memory", + } + + result = subprocess.run( + [sys.executable, "-c", "import src.modules.user.routes"], + cwd=backend, + capture_output=True, + text=True, + env=environment, + ) + + assert result.returncode == 0, result.stderr diff --git a/backend/tests/unit/modules/user/test_service.py b/backend/tests/unit/modules/user/test_service.py new file mode 100644 index 00000000..5b9c629a --- /dev/null +++ b/backend/tests/unit/modules/user/test_service.py @@ -0,0 +1,97 @@ +"""Unit tests for the user service's authorization rules.""" + +import pytest + +from src.modules.common.exceptions import PermissionDeniedError +from src.modules.user.schemas import UserUpdate +from src.modules.user.service import UserService + +UPDATE = "user.update" + + +@pytest.fixture +def user_service() -> UserService: + """Return a user service instance.""" + return UserService() + + +def _user(username: str, *, is_superuser: bool = False, email: str = "target@example.com") -> dict: + return {"id": 1, "username": username, "email": email, "is_superuser": is_superuser} + + +# ============================================================================= +# Who may update a profile at all +# ============================================================================= +async def test_a_user_can_update_their_own_profile_without_the_permission(user_service: UserService): + await user_service.verify_update_permission(_user("alice"), "alice", frozenset()) + + +async def test_a_user_cannot_update_another_profile_without_the_permission(user_service: UserService): + with pytest.raises(PermissionDeniedError): + await user_service.verify_update_permission(_user("alice"), "bob", frozenset()) + + +async def test_the_update_permission_allows_editing_another_profile(user_service: UserService): + await user_service.verify_update_permission(_user("alice"), "bob", frozenset({UPDATE})) + + +async def test_a_superuser_can_update_another_profile(user_service: UserService): + await user_service.verify_update_permission(_user("alice", is_superuser=True), "bob", frozenset()) + + +async def test_the_update_permission_does_not_grant_the_other_actions(user_service: UserService): + """``user.update`` is about updating; deleting another account stays owner-or-superuser.""" + with pytest.raises(PermissionDeniedError): + await user_service.verify_user_permission(_user("alice"), "bob", "delete this account") + + +# ============================================================================= +# What a user.update holder may do to someone else +# ============================================================================= +def test_a_holder_cannot_edit_a_superuser(user_service: UserService): + with pytest.raises(PermissionDeniedError): + user_service.verify_no_privilege_escalation( + _user("bob", is_superuser=True), + UserUpdate(name="Pwned Name"), + frozenset({UPDATE}), + frozenset(), + ) + + +def test_a_holder_cannot_edit_someone_holding_more(user_service: UserService): + with pytest.raises(PermissionDeniedError): + user_service.verify_no_privilege_escalation( + _user("bob"), + UserUpdate(name="Pwned Name"), + frozenset({UPDATE}), + frozenset({UPDATE, "user.delete"}), + ) + + +def test_a_holder_can_edit_someone_weaker(user_service: UserService): + user_service.verify_no_privilege_escalation( + _user("bob"), + UserUpdate(name="New Name"), + frozenset({UPDATE, "user.delete"}), + frozenset({UPDATE}), + ) + + +def test_a_holder_cannot_change_another_users_email(user_service: UserService): + """A verified provider email resolves a login to an account, so this is a takeover.""" + with pytest.raises(PermissionDeniedError, match="email"): + user_service.verify_no_privilege_escalation( + _user("bob"), + UserUpdate(email="attacker@example.com"), + frozenset({UPDATE}), + frozenset(), + ) + + +def test_submitting_the_email_a_user_already_has_is_not_a_change(user_service: UserService): + user_service.verify_no_privilege_escalation( + _user("bob", email="bob@example.com"), + UserUpdate(email="bob@example.com", name="New Name"), + frozenset({UPDATE}), + frozenset(), + ) diff --git a/docs/user-guide/admin-panel/user-management.md b/docs/user-guide/admin-panel/user-management.md index 4972ddd2..d044c6ab 100644 --- a/docs/user-guide/admin-panel/user-management.md +++ b/docs/user-guide/admin-panel/user-management.md @@ -175,5 +175,5 @@ For richer auditing, write to a dedicated log stream or push events to a SIEM. - **[Configuration](configuration.md)** — Environment variables and cookie behavior - **[Adding Models](adding-models.md)** — Register your own admin views -- **[Permissions](../authentication/permissions.md)** — Application-level superuser checks +- **[Permissions](../authentication/permissions.md)** — Role permissions and application-level superuser checks - **[Production](../production.md)** — Production hardening checklist diff --git a/docs/user-guide/api/endpoints.md b/docs/user-guide/api/endpoints.md index 9c8bbcdf..0f45b2ce 100644 --- a/docs/user-guide/api/endpoints.md +++ b/docs/user-guide/api/endpoints.md @@ -43,6 +43,8 @@ The boilerplate pre-defines aliases for every shared dependency in `infrastructu | `CurrentUserDep` | `Annotated[dict[str, Any], Depends(get_current_user)]` | | `CurrentSuperUserDep` | `Annotated[dict[str, Any], Depends(get_current_superuser)]` | | `OptionalUserDep` | `Annotated[dict[str, Any] \| None, Depends(get_optional_user)]` | +| `CurrentPrincipalDep` | `Annotated[Principal, Depends(get_current_principal)]` | +| `CurrentPermissionsDep` | `Annotated[frozenset[str], Depends(get_current_permissions)]` | | `OAuth2FormDep` | `Annotated[OAuth2PasswordRequestForm, Depends()]` | Per-module service aliases live in `modules//dependencies.py`: @@ -285,6 +287,47 @@ async def hard_delete_widget( The leading underscore on the dependency-only parameter is the convention used across the boilerplate. +### Permission Required + +`require_permissions` returns a dependency that answers 403 unless the caller holds every permission named. Superusers always pass. Because it injects nothing into the handler, it belongs in the decorator's `dependencies` list: + +```python +from ...infrastructure.auth.dependencies import require_permissions + + +@router.get( + "/", + response_model=list[WidgetRead], + dependencies=[require_permissions("widget.read")], +) +async def list_widgets( + db: AsyncSessionDep, + widget_service: WidgetServiceDep, +) -> list[dict[str, Any]]: + return await widget_service.get_all(db) +``` + +When a handler needs the permission set itself — to branch, or to compare it against another user's — take `CurrentPermissionsDep`: + +```python +from ...infrastructure.dependencies import CurrentPermissionsDep + + +@router.patch("/{widget_id}") +async def update_widget( + widget_id: int, + values: WidgetUpdate, + current_user: CurrentUserDep, + permissions: CurrentPermissionsDep, + db: AsyncSessionDep, + widget_service: WidgetServiceDep, +) -> dict[str, str]: + await widget_service.verify_update_permission(current_user, widget_id, permissions, db) + ... +``` + +The permission names must be registered in the `role` module's registry, or `require_permissions` raises at import time. See [Permissions](../authentication/permissions.md#role-based-permissions) for declaring them. + ### API Key Authentication For machine-to-machine clients, see [Authentication](../authentication/index.md). API keys are managed via the `/api/v1/api-keys/*` endpoints in `modules/api_keys/routes.py`. @@ -440,6 +483,7 @@ touch backend/src/modules/widgets/__init__.py | `crud.py` | `crud_widgets: FastCRUD = FastCRUD(Widget)` | | `service.py` | `WidgetService` with `create`, `get_by_id`, `update`, `delete` methods | | `routes.py` | `APIRouter` with the endpoints | +| `permissions.py` | `StrEnum` of the module's permissions, if its routes are permission-gated (see [Permissions](../authentication/permissions.md#role-based-permissions)) | ### 3. Register the Model diff --git a/docs/user-guide/api/index.md b/docs/user-guide/api/index.md index 27111e7e..51f8e00d 100644 --- a/docs/user-guide/api/index.md +++ b/docs/user-guide/api/index.md @@ -69,7 +69,16 @@ async def get_profile( return current_user ``` -For superuser-only endpoints, swap in `get_current_superuser`. See [Authentication](../authentication/index.md) for the full picture. +For superuser-only endpoints, swap in `get_current_superuser`. For endpoints gated on a permission a role grants, add `require_permissions(...)` to the route's `dependencies`: + +```python +from ...infrastructure.auth.dependencies import require_permissions + +@router.get("/", dependencies=[require_permissions("user.read")]) +async def list_users(...): ... +``` + +`infrastructure/dependencies.py` also exports `CurrentPrincipalDep` (the crudauth `Principal`) and `CurrentPermissionsDep` (the caller's effective permissions) for handlers that need them. See [Authentication](../authentication/index.md) for the full picture. ### Easy Pagination @@ -182,7 +191,7 @@ What ships out of the box (40 total routes): | Prefix | Source | Notes | |--------|--------|-------| -| `POST/GET/PATCH/DELETE /api/v1/users/*` | `modules/user/routes.py` | Open create; reads/updates need a session, and a lookup by username returns no email | +| `POST/GET/PATCH/DELETE /api/v1/users/*` | `modules/user/routes.py` | Open create; reads/updates need a session, and a lookup by username returns no email. Listing every user needs the `user.read` permission | | `GET /api/v1/tiers/*` | `modules/tier/routes.py` | Authenticated list + lookup by name | | `GET/PATCH/DELETE /api/v1/rate-limits/*` | `modules/rate_limit/routes.py` | Superuser only | | `POST /api/v1/auth/login`, `logout`, `logout-all`, `refresh-csrf`, `check-auth` | `infrastructure/auth/routes.py` | Session auth | diff --git a/docs/user-guide/api/pagination.md b/docs/user-guide/api/pagination.md index 2f7f5306..19741370 100644 --- a/docs/user-guide/api/pagination.md +++ b/docs/user-guide/api/pagination.md @@ -197,23 +197,26 @@ The boilerplate uses `ge=1, le=100` for the user list endpoint and `ge=1, le=100 From `modules/user/routes.py`: ```python +from ...infrastructure.auth.dependencies import require_permissions + + @router.get( "/", response_model=PaginatedListResponse[UserRead], - summary="List All Users (Admin)", + summary="List All Users", responses={ 401: {"description": "Not authenticated"}, - 403: {"description": "Not authorized - requires admin privileges"}, + 403: {"description": "Not authorized - requires the user.read permission"}, }, + dependencies=[require_permissions("user.read")], ) async def get_users( - db: Annotated[AsyncSession, Depends(async_session)], - _: Annotated[dict[str, Any], Depends(get_current_superuser)], - user_service: Annotated[UserService, Depends(get_user_service)], + db: AsyncSessionDep, + user_service: UserServiceDep, page: int = 1, items_per_page: int = 10, ) -> dict[str, Any]: - """Get paginated list of all users (admin only).""" + """Get paginated list of users.""" users_data = await user_service.get_paginated( skip=compute_offset(page, items_per_page), limit=items_per_page, @@ -222,6 +225,8 @@ async def get_users( return paginated_response(crud_data=users_data, page=page, items_per_page=items_per_page) ``` +The endpoint is gated on the `user.read` permission, which a role grants — not on the superuser flag. Superusers always pass. See [Permissions](../authentication/permissions.md#role-based-permissions). + ## Real Endpoint: API Key Usage History From `modules/api_keys/routes.py` — same pattern, different limit cap: @@ -304,4 +309,4 @@ The User model already indexes `username`, `email`, `tier_id`, `google_id`, and - **[CRUD Operations](../database/crud.md)** — Filter/sort/offset/limit semantics - **[Schemas](../database/schemas.md)** — How `*Read` schemas pair with `schema_to_select` -- **[Authentication](../authentication/index.md)** — Gating list endpoints behind login or admin +- **[Authentication](../authentication/index.md)** — Gating list endpoints behind login, a permission, or admin diff --git a/docs/user-guide/authentication/index.md b/docs/user-guide/authentication/index.md index 53ac35f2..72e81655 100644 --- a/docs/user-guide/authentication/index.md +++ b/docs/user-guide/authentication/index.md @@ -142,6 +142,7 @@ The full key is returned only on creation. Each key has its own permissions, usa ### Permission System +- **Roles** carrying `resource.action` permissions (`modules/role/`) — `require_permissions("user.read")` gates a route, superusers bypass it - **Superuser flag** on `User.is_superuser` for admin-only routes - **Tier-based** access via the `Tier` model — every user belongs to a tier, and rate limits are configured per tier path - **Resource ownership** checks live in services (the route doesn't decide who owns what) @@ -200,6 +201,21 @@ async def gdpr_delete_user( The leading underscore is the codebase's convention for dependency-only parameters. +### Permission Required + +```python +from ...infrastructure.auth.dependencies import require_permissions + +@router.get("/", dependencies=[require_permissions("user.read")]) +async def get_users( + db: AsyncSessionDep, + user_service: UserServiceDep, +) -> dict[str, Any]: + ... +``` + +Returns 403 unless the caller holds every named permission through one of their roles; superusers always pass. `infrastructure/dependencies.py` exports `CurrentPermissionsDep` for handlers that need the permission set itself, and `CurrentPrincipalDep` for the crudauth `Principal`. See [Permissions](permissions.md#role-based-permissions). + ### Resource Ownership Ownership is checked in the service layer, not in the route: diff --git a/docs/user-guide/authentication/permissions.md b/docs/user-guide/authentication/permissions.md index b6ff066f..73714d55 100644 --- a/docs/user-guide/authentication/permissions.md +++ b/docs/user-guide/authentication/permissions.md @@ -1,13 +1,14 @@ # Permissions and Authorization -Authentication answers "who are you?". Authorization answers "what can you do?". This page covers the boilerplate's authorization patterns: superuser flags, resource ownership, tier-based limits, and API key permissions. +Authentication answers "who are you?". Authorization answers "what can you do?". This page covers the boilerplate's authorization patterns: role permissions, superuser flags, resource ownership, tier-based limits, and API key permissions. ## Authorization Patterns -The boilerplate ships four overlapping mechanisms. Pick the one(s) that fit your use case. +The boilerplate ships five overlapping mechanisms. Pick the one(s) that fit your use case. | Pattern | Where it lives | When to use | |---------|----------------|-------------| +| **Role permissions** | `Role` + `RolePermission` + `UserRole` models, `require_permissions` | Granting a named capability to a group of users | | **Superuser flag** | `User.is_superuser` boolean | Admin-only operations | | **Resource ownership** | Service-layer permission checks | "Users can only edit their own X" | | **Tier-based limits** | `Tier` model + `RateLimit` rules | Subscription gating, rate limits | @@ -16,10 +17,128 @@ The boilerplate ships four overlapping mechanisms. Pick the one(s) that fit your These compose. A typical request goes through: 1. **Authentication** — session cookie (or API key) identifies *who* -2. **Coarse access** — superuser flag for admin endpoints +2. **Coarse access** — role permissions, or the superuser flag, for privileged endpoints 3. **Fine-grained access** — service-layer ownership / tier checks 4. **Rate limiting** — tier-based per-route limits (separate concern) +## Role-Based Permissions + +A permission is a flat `resource.action` string — `user.read`, `tier.update`. Permissions are never granted to a user directly: a `Role` carries a set of them, and a user is assigned roles. + +```text +User ──< UserRole >── Role ──< RolePermission +``` + +Three tables in `modules/role/models.py`: + +| Table | Columns | Notes | +|-------|---------|-------| +| `roles` | `id`, `name` (unique), `description` | No soft delete — a deleted role is gone | +| `role_permissions` | `role_id`, `permission_name` (composite PK) | `ON DELETE CASCADE` from `roles` | +| `user_roles` | `user_id`, `role_id` (composite PK) | `ON DELETE CASCADE` from both sides | + +### Declaring a Module's Permissions + +Each module owns its permission names in its own `permissions.py`, as a `StrEnum` decorated with `@register_permissions("")`: + +```python +# modules/user/permissions.py +from enum import StrEnum + +from ..role.permission_registry import register_permissions + + +@register_permissions("user") +class UserPermission(StrEnum): + READ = "user.read" + CREATE = "user.create" + UPDATE = "user.update" + DELETE = "user.delete" +``` + +Registration is validated: the resource must match `^[a-z][a-z0-9_]*$`, every member must be `.` with an action matching the same pattern, and the whole name must fit the 100-character `permission_name` column. A resource can only be registered once. + +`discover_permissions()` walks `src.modules.*.permissions` and imports each one; `modules/__init__.py` calls it at import time, so a new `permissions.py` needs no registration elsewhere. + +The registry in `modules/role/permission_registry.py` is what the rest of the app reads: + +| Function | Returns | +|----------|---------| +| `all_permissions()` | Every registered name, as a `frozenset[str]` | +| `permission_groups()` | `{resource: (names, ...)}` — for a UI that offers permissions per resource | +| `is_known_permission(name)` | Whether a name is registered | + +Unregistered names can't be stored: `RolePermission` validates `permission_name` against the registry and raises. In the other direction, a name that was stored and has since been removed from the code is ignored when permissions are loaded, so deleting a permission from a `StrEnum` can never grant anything. + +### Protecting a Route + +`require_permissions(*names)` returns a dependency that answers 403 unless the caller holds every name. It injects nothing into the handler, so it goes in the route's `dependencies`: + +```python +# modules/user/routes.py +from ...infrastructure.auth.dependencies import require_permissions + + +@router.get( + "/", + response_model=PaginatedListResponse[UserRead], + dependencies=[require_permissions("user.read")], +) +async def get_users( + db: AsyncSessionDep, + user_service: UserServiceDep, + page: int = 1, + items_per_page: int = 10, +) -> dict[str, Any]: + ... +``` + +Unknown names are a programming error, not a runtime one: `require_permissions` raises when the route is declared, so a typo fails at import rather than on the first request. + +**Superusers bypass every permission check.** `require_permissions` passes them without a lookup, and `get_current_permissions` reports them as holding every registered permission — so a superuser needs no roles. + +### Reading the Caller's Permissions + +When the handler itself has to decide, take the permission set instead of a guard. `CurrentPermissionsDep` (from `infrastructure/dependencies.py`) is `get_current_permissions` as an `Annotated` alias; FastAPI resolves it once per request, so several guards and parameters share one query: + +```python +from ...infrastructure.dependencies import CurrentPermissionsDep + + +@router.patch("/{username}") +async def update_user_profile( + username: str, + values: UserUpdate, + current_user: CurrentUserDep, + permissions: CurrentPermissionsDep, + db: AsyncSessionDep, + user_service: UserServiceDep, +) -> dict[str, str]: + await user_service.verify_update_permission(current_user, username, permissions) + ... +``` + +`load_permissions(db, user_id, is_superuser=...)` is the same lookup as a plain function, for code outside a request — or, as in the route above, for reading *another* user's permissions to compare against the caller's. + +### Granting Permissions + +A role is a row in `roles` plus one `role_permissions` row per permission; assigning it is a row in `user_roles`. Write them through the ORM (from a script, a migration, or your own admin tooling): + +```python +from src.modules.role.models import Role, RolePermission, UserRole + +role = Role(name="support", description="Read-only access to user records") +db.add(role) +await db.flush() + +db.add(RolePermission(role_id=role.id, permission_name="user.read")) +db.add(UserRole(user_id=user_id, role_id=role.id)) +await db.commit() +``` + +!!! info "Not shipped yet" + Role and permission CRUD endpoints, admin-panel views for roles, and narrowing an API key to a subset of its owner's permissions are follow-up work. This change ships the models, the registry, and the route guards. + ## Superuser Authorization The User model has an `is_superuser: bool` column. Endpoints that should only be accessible to admins use the `get_current_superuser` dependency: @@ -55,6 +174,8 @@ The leading `_:` is the codebase convention for dependency-only parameters whose - GDPR data anonymization - System configuration changes +Anything a *subset* of staff should be able to do is better expressed as a permission on a role than as another superuser. + ### Bootstrapping the First Superuser The first superuser is created by `scripts/setup_initial_data.py` from `ADMIN_*` env vars on first run: @@ -87,7 +208,9 @@ async def verify_user_permission( raise PermissionDeniedError(f"Cannot {action} for another user") ``` -Routes call this before dispatching the operation: +Routes call this before dispatching the operation. + +Where ownership and a role permission both apply, the service takes the permission set too. `PATCH /api/v1/users/{username}` is the example that ships: ```python # modules/user/routes.py @@ -95,15 +218,31 @@ Routes call this before dispatching the operation: async def update_user_profile( username: str, values: UserUpdate, - current_user: Annotated[dict[str, Any], Depends(get_current_user)], - db: Annotated[AsyncSession, Depends(async_session)], - user_service: Annotated[UserService, Depends(get_user_service)], + current_user: CurrentUserDep, + permissions: CurrentPermissionsDep, + db: AsyncSessionDep, + user_service: UserServiceDep, ) -> dict[str, str]: - try: - await user_service.verify_user_permission(current_user, username, "update profile") - # ...proceed with update... + await user_service.verify_update_permission(current_user, username, permissions) + user = await user_service.get_by_username(username, db) + + if not user_service.is_self_or_superuser(current_user, user["username"]): + target_permissions = await load_permissions(db, user["id"]) + user_service.verify_no_privilege_escalation(user, values, permissions, target_permissions) + + await user_service.update(user["id"], values, db) + return {"message": "User updated successfully"} ``` +The rules this enforces: + +- A user may always edit their own profile. +- A superuser may edit anyone. +- A `user.update` holder may edit *another* user only when that user is not a superuser and holds no permission the requester lacks — editing an account is a way to take it over, so it can't reach a stronger one. +- Only a superuser may change another user's email address: a verified provider email is how an OAuth login is matched to an existing account. + +The public `UserUpdate` schema accepts `name`, `username`, `email` and `profile_image_url` only. `google_id`, `github_id`, `oauth_provider`, `email_verified` and `oauth_updated_at` moved to `UserAdminUpdate`, which the admin panel uses — sending them to the API returns 422. + The exception flows up to the global handler (registered in `infrastructure/app_factory.py`) which translates it via the `EXCEPTION_MAPPING` table — `PermissionDeniedError` → `ForbiddenException` (403). See [Exceptions](../api/exceptions.md) for the full mapping pipeline. ### Generic Ownership Pattern diff --git a/docs/user-guide/authentication/sessions.md b/docs/user-guide/authentication/sessions.md index d30f6cc2..b4185416 100644 --- a/docs/user-guide/authentication/sessions.md +++ b/docs/user-guide/authentication/sessions.md @@ -75,6 +75,31 @@ async def list_products( **`get_current_principal`** — Returns the crudauth `Principal` (session-validated, CSRF-enforced). Use it when you need the session id (`principal.metadata["session_id"]`) or the `user_id` directly rather than the full user dict. `get_optional_principal` is the never-raises variant. +**`get_current_permissions`** — Returns the permissions the caller effectively holds through the roles assigned to them, as a `frozenset[str]`. FastAPI caches it for the length of the request, so the lookup runs once however many guards ask for it. Superusers hold every registered permission. + +```python +@router.patch("/{username}") +async def update_user_profile( + username: str, + permissions: Annotated[frozenset[str], Depends(get_current_permissions)], +) -> dict[str, str]: + if "user.update" in permissions: + ... +``` + +**`require_permissions(*names)`** — Returns a dependency that raises 403 unless the caller holds every named permission. Superusers pass. It injects nothing, so it goes in the route's `dependencies`: + +```python +from ...infrastructure.auth.dependencies import require_permissions + + +@router.get("/", dependencies=[require_permissions("user.read")]) +async def get_users(...) -> dict[str, Any]: + ... +``` + +See [Permissions](permissions.md#role-based-permissions) for declaring the names these take. + ### Protecting Entire Routers Apply auth to every route in a router: diff --git a/docs/user-guide/authentication/user-management.md b/docs/user-guide/authentication/user-management.md index ac10afb3..3faac8c9 100644 --- a/docs/user-guide/authentication/user-management.md +++ b/docs/user-guide/authentication/user-management.md @@ -138,7 +138,7 @@ Returns 404 if not found or soft-deleted. The admin-only `/active-and-inactive/{ ### Update Profile -Users can update their own profile; superusers can update anyone's. Tier updates are gated on a separate endpoint (see [Permissions](permissions.md)). +Users can always update their own profile. Updating someone else's needs superuser, or the `user.update` permission. Tier updates are gated on a separate endpoint (see [Permissions](permissions.md)). ```bash curl -X PATCH http://localhost:8000/api/v1/users/johndoe \ @@ -148,20 +148,30 @@ curl -X PATCH http://localhost:8000/api/v1/users/johndoe \ -d '{"name": "John Updated"}' ``` -The service enforces the ownership rule: +The service enforces the rules: ```python # modules/user/service.py -async def verify_user_permission( - self, current_user: dict[str, Any], target_username: str, action: str, +async def verify_update_permission( + self, + requester_user: dict[str, Any], + target_username: str, + permissions: Collection[str], ) -> None: - if current_user["username"] != target_username and not current_user["is_superuser"]: - raise PermissionDeniedError(f"Cannot {action} for another user") + if self.is_self_or_superuser(requester_user, target_username): + return + + if UserPermission.UPDATE.value in permissions: + return + + raise PermissionDeniedError("You don't have permission to update profile on this user") ``` +A `user.update` holder who isn't a superuser is then held to `verify_no_privilege_escalation`: the target must not be a superuser, must hold no permission the requester lacks, and its email address can't be changed — a verified provider email is how an OAuth login is matched to an account, so only a superuser may change someone else's. See [Permissions](permissions.md#resource-ownership). + If the body changes `username` or `email`, the service also re-checks uniqueness. -The `UserUpdate` schema makes every field optional so clients can send partial updates: +The `UserUpdate` schema makes every field optional so clients can send partial updates. The OAuth identifiers and the verification flag are not part of it — they live on `UserAdminUpdate`, which the admin panel uses, so sending them here returns 422: ```python from .constants import NAME_MAX_LENGTH, USERNAME_MAX_LENGTH, USERNAME_PATTERN @@ -244,11 +254,11 @@ Email is intentionally retained for legal compliance purposes (audit trail, "rig ### List All Users -`GET /api/v1/users/` — superuser only, paginated. +`GET /api/v1/users/` — paginated, gated on the `user.read` permission (`dependencies=[require_permissions("user.read")]`). A role grants it; superusers always pass. See [Permissions](permissions.md#role-based-permissions). ```bash curl "http://localhost:8000/api/v1/users/?page=1&items_per_page=10" \ - -b superuser_cookies.txt + -b cookies.txt ``` Response shape (via `paginated_response`): diff --git a/docs/user-guide/database/models.md b/docs/user-guide/database/models.md index 32870ebe..42757548 100644 --- a/docs/user-guide/database/models.md +++ b/docs/user-guide/database/models.md @@ -9,6 +9,7 @@ Models live in **`backend/src/modules//models.py`** — colocated with ```text backend/src/modules/ ├── user/models.py # User +├── role/models.py # Role, RolePermission, UserRole ├── tier/models.py # Tier ├── rate_limit/models.py # RateLimit └── api_keys/models.py # APIKey, KeyUsage, KeyPermission @@ -65,11 +66,15 @@ Each module's models are imported in `backend/src/modules/__init__.py`: ```python from .api_keys.models import APIKey, KeyPermission, KeyUsage from .rate_limit.models import RateLimit +from .role.models import Role, RolePermission, UserRole +from .role.permission_registry import discover_permissions from .tier.models import Tier from .user.models import User + +discover_permissions() ``` -When you add a new module, **add its models here** so Alembic's `--autogenerate` sees them. +When you add a new module, **add its models here** so Alembic's `--autogenerate` sees them. The same import is what triggers `discover_permissions()`, which imports every module's `permissions.py` — see [Permissions](../authentication/permissions.md#role-based-permissions). ## Relationships diff --git a/docs/user-guide/database/schemas.md b/docs/user-guide/database/schemas.md index 346ed911..d221510c 100644 --- a/docs/user-guide/database/schemas.md +++ b/docs/user-guide/database/schemas.md @@ -139,7 +139,17 @@ class UserUpdate(BaseModel): ] -class UserUpdateInternal(UserUpdate): +# Fields only an administrator may set — the OAuth identifiers decide who a +# provider login resolves to, so they stay out of the public update schema +class UserAdminUpdate(UserUpdate): + google_id: str | None = None + github_id: str | None = None + oauth_provider: str | None = None + email_verified: bool | None = None + oauth_updated_at: datetime | None = None + + +class UserUpdateInternal(UserAdminUpdate): updated_at: datetime # service stamps this before persisting @@ -174,6 +184,7 @@ The schemas follow a consistent vocabulary across modules: | `Create` | API request body for POST | | `CreateInternal` | What the service stores (raw password → hashed_password) | | `Update` | Partial update body for PATCH (all fields optional) | +| `AdminUpdate` | Update body extended with the fields only an administrator may set | | `UpdateInternal` | What the service stores on update (e.g. with stamped `updated_at`) | | `TierUpdate`, `Anonymize`, `Delete`, … | Operation-specific narrow schemas | diff --git a/docs/user-guide/project-structure.md b/docs/user-guide/project-structure.md index af43eb4c..2551d71b 100644 --- a/docs/user-guide/project-structure.md +++ b/docs/user-guide/project-structure.md @@ -122,7 +122,9 @@ modules/ │ ├── crud.py # FastCRUD wrapper (crud_users) │ ├── service.py # Business logic (UserService) │ ├── routes.py # APIRouter with /users endpoints +│ ├── permissions.py # UserPermission StrEnum (user.read, user.update, ...) │ └── enums.py # OAuthProvider, etc. +├── role/ # RBAC: Role, RolePermission, UserRole + the permission registry ├── tier/ # Subscription tiers (model + simple CRUD) ├── rate_limit/ # Per-tier rate limit definitions └── api_keys/ # API keys, key usage, key permissions @@ -139,6 +141,7 @@ Each module is **self-contained**: drop it in, drop it out, with minimal blast r | `crud.py` | FastCRUD instances for the model | | `service.py` | Business logic — orchestrates CRUD calls, applies rules | | `routes.py` | `APIRouter` with the module's endpoints | +| `permissions.py` | `StrEnum` of the module's permissions, registered with `@register_permissions` (optional) | | `enums.py` | StrEnum types if the module needs them (optional) | ## Migrations (`backend/migrations/`) @@ -213,8 +216,11 @@ FastAPI's `Depends` is used throughout: - **Database session** — `Depends(async_session)` from `infrastructure.database.session` - **Current user** — `Depends(get_current_user)` from `infrastructure.auth.dependencies` - **Superuser only** — `Depends(get_current_superuser)` +- **Permission required** — `require_permissions("user.read")` in the route's `dependencies` list - **Service instances** — Each module's `routes.py` defines its own `get__service()` factory +`infrastructure/dependencies.py` collects `Annotated` aliases for the shared ones — including `CurrentPrincipalDep` (the crudauth `Principal`) and `CurrentPermissionsDep` (the caller's effective permissions, resolved once per request). + ### Configuration All configuration lives in `infrastructure/config/settings.py`, loaded from `.env`: