From 9a02721e53a74a582e631ac63958f489fa7484b3 Mon Sep 17 00:00:00 2001 From: Amr Gaber Date: Mon, 2 Feb 2026 02:47:11 -0600 Subject: [PATCH] feat: add role-based authorization with admin endpoint --- README.md | 30 +++- .../b3c7a1d9e2f4_add_user_role_column.py | 31 ++++ app/auth/__init__.py | 9 +- app/auth/roles.py | 31 ++++ app/main.py | 3 +- app/models/user.py | 9 +- app/routers/__init__.py | 3 +- app/routers/admin.py | 47 ++++++ app/schemas/user.py | 2 +- tests/conftest.py | 22 +++ tests/test_roles.py | 137 ++++++++++++++++++ 11 files changed, 315 insertions(+), 9 deletions(-) create mode 100644 alembic/versions/b3c7a1d9e2f4_add_user_role_column.py create mode 100644 app/auth/roles.py create mode 100644 app/routers/admin.py create mode 100644 tests/test_roles.py diff --git a/README.md b/README.md index 8deb71c..1db310a 100644 --- a/README.md +++ b/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,6 +290,7 @@ 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/ @@ -272,6 +298,7 @@ api-template/ │ │ ├── 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 diff --git a/alembic/versions/b3c7a1d9e2f4_add_user_role_column.py b/alembic/versions/b3c7a1d9e2f4_add_user_role_column.py new file mode 100644 index 0000000..6f84312 --- /dev/null +++ b/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") diff --git a/app/auth/__init__.py b/app/auth/__init__.py index 031a632..2e2b9f8 100644 --- a/app/auth/__init__.py +++ b/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", +] diff --git a/app/auth/roles.py b/app/auth/roles.py new file mode 100644 index 0000000..952d4cf --- /dev/null +++ b/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 diff --git a/app/main.py b/app/main.py index 732b551..ba4e744 100644 --- a/app/main.py +++ b/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) diff --git a/app/models/user.py b/app/models/user.py index 09a0fa9..ab217eb 100644 --- a/app/models/user.py +++ b/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 + ) diff --git a/app/routers/__init__.py b/app/routers/__init__.py index ff63447..958073c 100644 --- a/app/routers/__init__.py +++ b/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"] diff --git a/app/routers/admin.py b/app/routers/admin.py new file mode 100644 index 0000000..e3464b4 --- /dev/null +++ b/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 diff --git a/app/schemas/user.py b/app/schemas/user.py index 00c1502..5439228 100644 --- a/app/schemas/user.py +++ b/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): diff --git a/tests/conftest.py b/tests/conftest.py index ddf1564..aa507c2 100644 --- a/tests/conftest.py +++ b/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) diff --git a/tests/test_roles.py b/tests/test_roles.py new file mode 100644 index 0000000..8d16dc7 --- /dev/null +++ b/tests/test_roles.py @@ -0,0 +1,137 @@ +from uuid import uuid4 + +from httpx import AsyncClient + +from app.auth import current_active_user +from app.main import app +from app.models.user import User + + +class TestRoleAccess: + """Test that role-based access control works correctly.""" + + async def test_admin_can_access_admin_route( + self, admin_client: AsyncClient, admin_user: User, session + ): + """Admin users can access admin-gated endpoints.""" + # Create a target user in the DB to update + target = User(id=uuid4(), email="target@example.com", hashed_password="fake") + session.add(target) + await session.commit() + + response = await admin_client.patch( + f"/admin/users/{target.id}/role", + json={"role": "admin"}, + ) + assert response.status_code == 200 + + async def test_regular_user_gets_403(self, auth_client: AsyncClient, test_user: User): + """Regular users are forbidden from admin endpoints.""" + response = await auth_client.patch( + f"/admin/users/{uuid4()}/role", + json={"role": "admin"}, + ) + assert response.status_code == 403 + assert response.json()["detail"] == "Insufficient permissions" + + async def test_superuser_bypasses_role_check( + self, client: AsyncClient, superuser: User, session + ): + """Superusers can access admin endpoints regardless of their role.""" + app.dependency_overrides[current_active_user] = lambda: superuser + try: + target = User(id=uuid4(), email="target2@example.com", hashed_password="fake") + session.add(target) + await session.commit() + + response = await client.patch( + f"/admin/users/{target.id}/role", + json={"role": "admin"}, + ) + assert response.status_code == 200 + finally: + app.dependency_overrides.pop(current_active_user, None) + + +class TestRoleUpdate: + """Test the PATCH /admin/users/{user_id}/role endpoint.""" + + async def test_update_role_success(self, admin_client: AsyncClient, admin_user: User, session): + """Successfully update a user's role.""" + target = User(id=uuid4(), email="target3@example.com", hashed_password="fake") + session.add(target) + await session.commit() + + response = await admin_client.patch( + f"/admin/users/{target.id}/role", + json={"role": "admin"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["role"] == "admin" + assert data["email"] == "target3@example.com" + + async def test_invalid_role_returns_422( + self, admin_client: AsyncClient, admin_user: User, session + ): + """Invalid role value returns 422.""" + target = User(id=uuid4(), email="target4@example.com", hashed_password="fake") + session.add(target) + await session.commit() + + response = await admin_client.patch( + f"/admin/users/{target.id}/role", + json={"role": "supervillain"}, + ) + assert response.status_code == 422 + + async def test_user_not_found_returns_404(self, admin_client: AsyncClient, admin_user: User): + """Updating a non-existent user returns 404.""" + response = await admin_client.patch( + f"/admin/users/{uuid4()}/role", + json={"role": "admin"}, + ) + assert response.status_code == 404 + assert response.json()["detail"] == "User not found" + + +class TestAuthMeIncludesRole: + """Test that /auth/me returns the role field.""" + + async def test_me_returns_role_for_regular_user(self, client: AsyncClient): + user = User( + id=uuid4(), + email="me@example.com", + hashed_password="fake", + role="user", + is_active=True, + is_superuser=False, + is_verified=False, + ) + app.dependency_overrides[current_active_user] = lambda: user + try: + response = await client.get("/auth/me") + assert response.status_code == 200 + data = response.json() + assert data["role"] == "user" + finally: + app.dependency_overrides.pop(current_active_user, None) + + async def test_me_returns_role_for_admin(self, client: AsyncClient): + user = User( + id=uuid4(), + email="meadmin@example.com", + hashed_password="fake", + role="admin", + is_active=True, + is_superuser=False, + is_verified=False, + ) + app.dependency_overrides[current_active_user] = lambda: user + try: + response = await client.get("/auth/me") + assert response.status_code == 200 + data = response.json() + assert data["role"] == "admin" + finally: + app.dependency_overrides.pop(current_active_user, None)