feat: add role-based authorization with admin endpoint

README.md

@@ -107,6 +107,14 @@ This requires the API to be running locally (or set `OPENAPI_URL` to point to a
 | POST   | `/auth/refresh`     | Rotate refresh token, reissue access token |
 | GET    | `/auth/me`          | Get current authenticated user      |
 
+### Admin
+
+Admin endpoints require the `admin` role (superusers also have access).
+
+| Method | Endpoint                       | Description              |
+| ------ | ------------------------------ | ------------------------ |
+| PATCH  | `/admin/users/{id}/role`       | Update a user's role     |
+
 ### Notes (Example CRUD)
 
 All note endpoints require authentication. Users can only access their own notes.
@@ -128,6 +136,23 @@ Authentication uses httpOnly cookies with short-lived access tokens and rotating
 - **Token rotation**: Each refresh issues a new token in the same family; reuse of an old token revokes the entire family (theft detection)
 - **Rate limiting**: Login (5/min), registration (3/min), refresh (30/min)
 
+### Role-Based Access Control
+
+Users have a `role` field (default: `user`). Roles are defined as a `StrEnum` in `app/auth/roles.py`:
+
+- **user** — default role for all registered users
+- **admin** — can access admin endpoints (e.g. updating user roles)
+
+Superusers (`is_superuser=True`) bypass all role checks. Roles are read-only via `GET /auth/me` and can only be changed by admins via `PATCH /admin/users/{id}/role`. The `require_role()` dependency factory can be used to gate any route:
+
+```python
+from app.auth import require_role
+
+@router.get("/admin-only")
+async def admin_only(user: User = Depends(require_role("admin"))):
+    ...
+```
+
 ### Security Features
 
 - **Cookie auth**: httpOnly, Secure (in production), SameSite
@@ -265,13 +290,15 @@ api-template/
 │   ├── auth/
 │   │   ├── backend.py          # Cookie transport + JWT strategy
 │   │   ├── refresh.py          # Refresh token create/rotate/revoke
+│   │   ├── roles.py            # UserRole enum + require_role() dependency
 │   │   ├── security_logging.py # Structured security event logging
 │   │   └── users.py            # UserManager with login/failure hooks
 │   ├── models/
 │   │   ├── note.py             # Note model (example CRUD entity)
 │   │   ├── refresh_token.py    # Refresh token model
 │   │   └── user.py             # User model (FastAPI-Users)
 │   ├── routers/
+│   │   ├── admin.py            # Admin endpoints (role management)
 │   │   ├── auth_refresh.py     # /auth/refresh and /auth/jwt/logout
 │   │   └── notes.py            # Notes CRUD (user-scoped)
 │   ├── schemas/
@@ -289,7 +316,8 @@ api-template/
 │   └── env.py                  # Alembic configuration
 ├── tests/
 │   ├── conftest.py             # Fixtures (client, session, users)
-│   └── test_notes.py           # Notes CRUD + isolation tests
+│   ├── test_notes.py           # Notes CRUD + isolation tests
+│   └── test_roles.py           # Role-based access control tests
 ├── .env.example                # Environment template
 ├── .pre-commit-config.yaml
 ├── .python-version             # pyenv Python version

alembic/versions/b3c7a1d9e2f4_add_user_role_column.py

@@ -0,0 +1,31 @@
+"""add user role column
+
+Revision ID: b3c7a1d9e2f4
+Revises: af85030565fd
+Create Date: 2026-02-02 00:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "b3c7a1d9e2f4"
+down_revision: str | Sequence[str] | None = "af85030565fd"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Add role column to user table."""
+    op.add_column(
+        "user", sa.Column("role", sa.String(length=50), server_default="user", nullable=False)
+    )
+
+
+def downgrade() -> None:
+    """Remove role column from user table."""
+    op.drop_column("user", "role")

app/auth/__init__.py

@@ -1,3 +1,10 @@
+from app.auth.roles import UserRole, require_role
 from app.auth.users import auth_backend, current_active_user, fastapi_users
 
-__all__ = ["auth_backend", "current_active_user", "fastapi_users"]
+__all__ = [
+    "UserRole",
+    "auth_backend",
+    "current_active_user",
+    "fastapi_users",
+    "require_role",
+]

app/auth/roles.py

@@ -0,0 +1,31 @@
+from enum import StrEnum
+
+from fastapi import Depends, HTTPException, status
+
+from app.auth.users import current_active_user
+from app.models.user import User
+
+
+class UserRole(StrEnum):
+    USER = "user"
+    ADMIN = "admin"
+
+
+def require_role(*allowed_roles: str):
+    """Dependency factory that checks the user has one of the allowed roles.
+
+    Superusers bypass the role check. Returns the authenticated User so
+    routes don't need a separate ``Depends(current_active_user)``.
+    """
+
+    async def _check_role(user: User = Depends(current_active_user)) -> User:
+        if user.is_superuser:
+            return user
+        if user.role not in allowed_roles:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Insufficient permissions",
+            )
+        return user
+
+    return _check_role

app/main.py

@@ -16,7 +16,7 @@
 from app.features import router as features_router
 from app.logging import setup_logging
 from app.models.user import User
-from app.routers import notes_router
+from app.routers import admin_router, notes_router
 from app.routers.auth_refresh import router as auth_refresh_router
 from app.schemas.user import UserCreate, UserRead
 from app.telemetry import setup_telemetry
@@ -135,6 +135,7 @@ async def request_logging_middleware(request: Request, call_next) -> Response:
 
 
 # API routes
+app.include_router(admin_router)
 app.include_router(notes_router)
 app.include_router(features_router)
 

app/models/user.py

@@ -1,4 +1,6 @@
 from fastapi_users.db import SQLAlchemyBaseUserTableUUID
+from sqlalchemy import String
+from sqlalchemy.orm import Mapped, mapped_column
 
 from app.database import Base
 
@@ -13,9 +15,8 @@ class User(SQLAlchemyBaseUserTableUUID, Base):
     - is_active: whether user can authenticate
     - is_superuser: admin privileges
     - is_verified: email verification status
-
-    Add custom fields below if needed, e.g.:
-        display_name: Mapped[str | None] = mapped_column(String(100))
     """
 
-    pass
+    role: Mapped[str] = mapped_column(
+        String(50), default="user", server_default="user", nullable=False
+    )

app/routers/__init__.py

@@ -1,3 +1,4 @@
+from app.routers.admin import router as admin_router
 from app.routers.notes import router as notes_router
 
-__all__ = ["notes_router"]
+__all__ = ["admin_router", "notes_router"]

app/routers/admin.py

@@ -0,0 +1,47 @@
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from pydantic import BaseModel
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.auth.roles import UserRole, require_role
+from app.database import get_async_session
+from app.models.user import User
+from app.schemas.user import UserRead
+
+router = APIRouter(prefix="/admin", tags=["admin"])
+
+
+class RoleUpdate(BaseModel):
+    role: str
+
+
+@router.patch("/users/{user_id}/role", response_model=UserRead)
+async def update_user_role(
+    user_id: UUID,
+    body: RoleUpdate,
+    session: AsyncSession = Depends(get_async_session),
+    _admin: User = Depends(require_role("admin")),
+):
+    """Update a user's role. Requires admin role."""
+    # Validate role value against enum
+    try:
+        UserRole(body.role)
+    except ValueError:
+        valid = [r.value for r in UserRole]
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
+            detail=f"Invalid role '{body.role}'. Must be one of: {valid}",
+        ) from None
+
+    target = await session.get(User, user_id)
+    if not target:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="User not found",
+        )
+
+    target.role = body.role
+    await session.commit()
+    await session.refresh(target)
+    return target

app/schemas/user.py

@@ -6,7 +6,7 @@
 class UserRead(schemas.BaseUser[UUID]):
     """Schema for reading user data."""
 
-    pass
+    role: str = "user"
 
 
 class UserCreate(schemas.BaseUserCreate):

tests/conftest.py

@@ -64,6 +64,18 @@ def other_user() -> User:
     return User(id=uuid4(), email="other@example.com", hashed_password="fake")
 
 
+@pytest.fixture
+def admin_user() -> User:
+    """A user with the admin role."""
+    return User(id=uuid4(), email="admin@example.com", hashed_password="fake", role="admin")
+
+
+@pytest.fixture
+def superuser() -> User:
+    """A superuser (bypasses role checks)."""
+    return User(id=uuid4(), email="super@example.com", hashed_password="fake", is_superuser=True)
+
+
 @pytest.fixture
 async def auth_client(client: AsyncClient, test_user: User) -> AsyncGenerator[AsyncClient, None]:
     """Client that authenticates as test_user via dependency override."""
@@ -72,3 +84,13 @@ async def auth_client(client: AsyncClient, test_user: User) -> AsyncGenerator[As
         yield client
     finally:
         app.dependency_overrides.pop(current_active_user, None)
+
+
+@pytest.fixture
+async def admin_client(client: AsyncClient, admin_user: User) -> AsyncGenerator[AsyncClient, None]:
+    """Client that authenticates as admin_user via dependency override."""
+    app.dependency_overrides[current_active_user] = lambda: admin_user
+    try:
+        yield client
+    finally:
+        app.dependency_overrides.pop(current_active_user, None)