feat: Piece 01: add RBAC core models and permissions

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

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

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}}

backend/src/modules/__init__.py

@@ -2,14 +2,21 @@
 
 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",
     "RateLimit",
     "APIKey",
     "KeyUsage",
     "KeyPermission",
+    "Role",
+    "RolePermission",
+    "UserRole",
 ]

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

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

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