feat(security): implement secure demo mode with automatic user creation and route protection

Author: Lidianacosta

AGENTS.md

@@ -82,6 +82,15 @@ Certifique-se de que todos os passos passem localmente antes de enviar seu códi
 
 ---
 
+## 🛡️ Modo Demonstração (Demo Mode)
+
+O sistema possui uma trava de segurança para contas de demonstração:
+- **Identificação:** Usuários com a flag `is_demo: true` no banco de dados.
+- **Restrição:** Bloqueio automático de atualizações de perfil (nome, email, senha) via dependência `forbid_demo_user`.
+- **Automação:** Criação automática de usuário demo via variáveis `DEMO_USER_USERNAME` e `DEMO_USER_PASSWORD`.
+
+---
+
 ## 🚀 Deployment
 
 O deploy da aplicação é realizado na plataforma **Render** utilizando **Docker**.

migrations/versions/cbaf2fee6e9b_add_is_demo_field_to_user_model.py

@@ -0,0 +1,36 @@
+"""add is_demo field to user model
+
+Revision ID: cbaf2fee6e9b
+Revises: 0de60751eea2
+Create Date: 2026-05-11 15:15:59.722259
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "cbaf2fee6e9b"
+down_revision: str | Sequence[str] | None = "0de60751eea2"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("user", schema=None) as batch_op:
+        batch_op.add_column(sa.Column("is_demo", sa.Boolean(), nullable=False))
+
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    with op.batch_alter_table("user", schema=None) as batch_op:
+        batch_op.drop_column("is_demo")
+
+    # ### end Alembic commands ###

render.yaml

@@ -17,6 +17,10 @@ services:
         sync: false
       - key: FIRST_SUPERUSER_PASSWORD
         sync: false
+      - key: DEMO_USER_USERNAME
+        sync: false
+      - key: DEMO_USER_PASSWORD
+        sync: false
 
 databases:
   - name: fastapibank-db

src/commands/init_db.py

@@ -1,6 +1,6 @@
 """Database initialization script.
 
-Checks if an initial superuser should be created and creates it if needed.
+Checks if an initial superuser and a demo user should be created.
 """
 
 import asyncio
@@ -19,41 +19,60 @@
 logger = logging.getLogger(__name__)
 
 
-async def init_db() -> None:
-    """Create the initial superuser if configured."""
-    if (
-        not settings.first_superuser_username
-        or not settings.first_superuser_password
-    ):
-        logger.info(
-            "First superuser credentials not configured. Skipping initialization."
+async def create_user_if_not_exists(
+    session: AsyncSession,
+    username: str,
+    password: str,
+    is_demo: bool = False,
+    description: str = "user",
+) -> None:
+    """Create a user if they don't already exist in the database."""
+    statement = select(User).where(User.username == username)
+    result = await session.exec(statement)
+    user = result.first()
+
+    if user:
+        logger.info("%s '%s' already exists. Skipping.", description, username)
+    else:
+        logger.info("Creating %s '%s'.", description, username)
+        user_in = UserIn(
+            username=username,
+            plain_password=password,
         )
-        return
+        new_user = User(
+            **user_in.model_dump(exclude={"plain_password"}), is_demo=is_demo
+        )
+        new_user.hashed_password = get_password_hash(user_in.plain_password)
+        session.add(new_user)
+        await session.commit()
+        logger.info("%s '%s' created successfully.", description, username)
+
 
+async def init_db() -> None:
+    """Initialize the database with default users."""
     engine = create_async_engine(settings.database_url)
 
     async with AsyncSession(engine) as session:
-        statement = select(User).where(
-            User.username == settings.first_superuser_username
-        )
-        result = await session.exec(statement)
-        user = result.first()
-
-        if user:
-            logger.info("Superuser already exists. Skipping creation.")
-        else:
-            logger.info("Creating initial superuser.")
-            user_in = UserIn(
-                username=settings.first_superuser_username,
-                plain_password=settings.first_superuser_password,
+        if (
+            settings.first_superuser_username
+            and settings.first_superuser_password
+        ):
+            await create_user_if_not_exists(
+                session,
+                settings.first_superuser_username,
+                settings.first_superuser_password,
+                is_demo=False,
+                description="Superuser",
             )
-            new_user = User(**user_in.model_dump(exclude={"plain_password"}))
-            new_user.hashed_password = get_password_hash(
-                user_in.plain_password
+
+        if settings.demo_user_username and settings.demo_user_password:
+            await create_user_if_not_exists(
+                session,
+                settings.demo_user_username,
+                settings.demo_user_password,
+                is_demo=True,
+                description="Demo user",
             )
-            session.add(new_user)
-            await session.commit()
-            logger.info("Initial superuser created successfully.")
 
     await engine.dispose()
 

src/controllers/user.py

@@ -11,6 +11,7 @@
 from src.schemas.user import User, UserDB, UserUpdateIn
 from src.services.user import UserServiceDep
 from src.utils.security import (
+    forbid_demo_user,
     get_current_active_user,
 )
 
@@ -36,7 +37,7 @@ async def read_user_me(
 @router.patch("/me/", response_model=User)
 async def update_user_me(
     user_update_in: UserUpdateIn,
-    current_user: Annotated[UserDB, Depends(get_current_active_user)],
+    current_user: Annotated[UserDB, Depends(forbid_demo_user)],
     user_service: UserServiceDep,
 ):
     """Update the profile of the currently authenticated user.

src/core/config.py

@@ -48,6 +48,9 @@ def assemble_db_url(cls, v: str) -> str:
     first_superuser_username: str | None = Field(default=None)
     first_superuser_password: str | None = Field(default=None)
 
+    demo_user_username: str | None = Field(default=None)
+    demo_user_password: str | None = Field(default=None)
+
     model_config = SettingsConfigDict(env_file=".env")
 
 

src/models/user.py

@@ -25,3 +25,4 @@ class User(Base, table=True):
     full_name: str | None = None
     disabled: bool | None = None
     hashed_password: str
+    is_demo: bool = False

src/utils/security.py

@@ -131,3 +131,28 @@ def get_current_active_user(
     if current_user.disabled:
         raise HTTPException(status_code=400, detail="Inactive user")
     return current_user
+
+
+def forbid_demo_user(
+    current_user: Annotated[User, Depends(get_current_active_user)],
+):
+    """Ensure the current user is not a demo user.
+
+    Used to protect sensitive operations from being performed by demo accounts.
+
+    Args:
+        current_user: The authenticated active user.
+
+    Returns:
+        The current user if they are not a demo user.
+
+    Raises:
+        HTTPException: 403 if the user is a demo user.
+
+    """
+    if current_user.is_demo:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Action not permitted in demonstration mode.",
+        )
+    return current_user

tests/conftest.py

@@ -90,3 +90,33 @@ async def access_token(client: AsyncClient):
         )
 
     return response.json()["access_token"]
+
+
+@pytest_asyncio.fixture
+async def demo_access_token(client: AsyncClient):
+    async with AsyncSession(async_engine) as session:
+        test_password = secrets.token_urlsafe(16)
+        test_user = User(
+            username="demo_user",
+            hashed_password=get_password_hash(test_password),
+            is_demo=True,
+        )
+        session.add(test_user)
+        await session.commit()
+        await session.refresh(test_user)
+        await session.close()
+
+    response = await client.post(
+        "/api/auth/token",
+        data={
+            "username": test_user.username,
+            "password": test_password,
+            "grant_type": "password",
+        },
+        headers={
+            "Authorization": "Bearer token",
+            "Content-Type": "application/x-www-form-urlencoded",
+        },
+    )
+
+    return response.json()["access_token"]

tests/integration/controllers/test_user.py

@@ -55,3 +55,19 @@ async def test_update_user_me_password_success(
 async def test_user_me_unauthorized(client: AsyncClient):
     response = await client.get("/api/users/me/")
     assert response.status_code == codes.UNAUTHORIZED
+
+
+async def test_update_user_me_fail_for_demo_user(
+    client: AsyncClient, demo_access_token: str
+):
+    response = await client.patch(
+        "/api/users/me/",
+        json={"full_name": "Should Fail"},
+        headers={"Authorization": f"Bearer {demo_access_token}"},
+    )
+
+    assert response.status_code == codes.FORBIDDEN
+    assert (
+        response.json()["detail"]
+        == "Action not permitted in demonstration mode."
+    )