refactor: move owasp route into small file

Author: makara4code

backend/app/alembic/versions/405b1d47dec0_add_owasp_demo_permissions.py

@@ -0,0 +1,107 @@
+"""add_owasp_demo_permissions
+
+Revision ID: 405b1d47dec0
+Revises: ffdac40437ff
+Create Date: 2025-12-14 20:05:20.434206
+
+"""
+import uuid
+from datetime import datetime, timezone
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '405b1d47dec0'
+down_revision = 'ffdac40437ff'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    """Add OWASP demo permissions and assign them to roles."""
+    permission_table = sa.table(
+        'permission',
+        sa.column('id', sa.Uuid()),
+        sa.column('name', sa.String()),
+        sa.column('description', sa.String()),
+        sa.column('resource', sa.String()),
+        sa.column('action', sa.String()),
+        sa.column('created_at', sa.DateTime()),
+    )
+
+    role_permission_table = sa.table(
+        'role_permission',
+        sa.column('role_id', sa.Uuid()),
+        sa.column('permission_id', sa.Uuid()),
+    )
+
+    now = datetime.now(timezone.utc)
+
+    # OWASP Demo permissions
+    owasp_permissions = [
+        {'id': uuid.uuid4(), 'name': 'documents:read', 'description': 'View all documents (OWASP demo)', 'resource': 'documents', 'action': 'read', 'created_at': now},
+        {'id': uuid.uuid4(), 'name': 'notes:read', 'description': 'View all notes (OWASP demo)', 'resource': 'notes', 'action': 'read', 'created_at': now},
+        {'id': uuid.uuid4(), 'name': 'config:read', 'description': 'View system configuration', 'resource': 'config', 'action': 'read', 'created_at': now},
+        {'id': uuid.uuid4(), 'name': 'audit:write', 'description': 'Write audit log entries', 'resource': 'audit', 'action': 'write', 'created_at': now},
+    ]
+
+    op.bulk_insert(permission_table, owasp_permissions)
+
+    # Get role IDs from database
+    connection = op.get_bind()
+
+    admin_role = connection.execute(
+        sa.text("SELECT id FROM role WHERE name = 'Admin'")
+    ).fetchone()
+    editor_role = connection.execute(
+        sa.text("SELECT id FROM role WHERE name = 'Editor'")
+    ).fetchone()
+    viewer_role = connection.execute(
+        sa.text("SELECT id FROM role WHERE name = 'Viewer'")
+    ).fetchone()
+
+    if not admin_role or not editor_role or not viewer_role:
+        return  # Roles not found, skip permission assignment
+
+    perm_ids = {p['name']: p['id'] for p in owasp_permissions}
+
+    role_permissions = []
+
+    # Admin gets all OWASP permissions
+    for perm in owasp_permissions:
+        role_permissions.append({'role_id': admin_role[0], 'permission_id': perm['id']})
+
+    # Editor gets all OWASP permissions (read + write)
+    for perm_name in ['documents:read', 'notes:read', 'config:read', 'audit:write']:
+        role_permissions.append({'role_id': editor_role[0], 'permission_id': perm_ids[perm_name]})
+
+    # Viewer gets read-only OWASP permissions
+    for perm_name in ['documents:read', 'notes:read']:
+        role_permissions.append({'role_id': viewer_role[0], 'permission_id': perm_ids[perm_name]})
+
+    op.bulk_insert(role_permission_table, role_permissions)
+
+
+def downgrade():
+    """Remove OWASP demo permissions."""
+    connection = op.get_bind()
+
+    # Get permission IDs
+    perm_names = ['documents:read', 'notes:read', 'config:read', 'audit:write']
+
+    for perm_name in perm_names:
+        # Delete role_permission associations first
+        connection.execute(
+            sa.text("""
+                DELETE FROM role_permission
+                WHERE permission_id IN (SELECT id FROM permission WHERE name = :name)
+            """),
+            {'name': perm_name}
+        )
+        # Delete the permission
+        connection.execute(
+            sa.text("DELETE FROM permission WHERE name = :name"),
+            {'name': perm_name}
+        )

backend/app/modules/owasp_demo/routers/__init__.py

@@ -0,0 +1,31 @@
+"""
+OWASP Top 10 2025 Vulnerability Demonstration Routes
+
+Each submodule contains vulnerable and secure endpoint pairs for educational purposes.
+"""
+
+from app.modules.owasp_demo.routers.a01_access_control import router as router_a01
+from app.modules.owasp_demo.routers.a02_security_misconfig import router as router_a02
+from app.modules.owasp_demo.routers.a03_supply_chain import router as router_a03
+from app.modules.owasp_demo.routers.a04_cryptographic import router as router_a04
+from app.modules.owasp_demo.routers.a05_injection import router as router_a05
+from app.modules.owasp_demo.routers.a06_insecure_design import router as router_a06
+from app.modules.owasp_demo.routers.a07_authentication import router as router_a07
+from app.modules.owasp_demo.routers.a08_integrity import router as router_a08
+from app.modules.owasp_demo.routers.a09_logging import router as router_a09
+from app.modules.owasp_demo.routers.a10_exception_handling import router as router_a10
+from app.modules.owasp_demo.routers.summary import router as router_summary
+
+__all__ = [
+    "router_a01",
+    "router_a02",
+    "router_a03",
+    "router_a04",
+    "router_a05",
+    "router_a06",
+    "router_a07",
+    "router_a08",
+    "router_a09",
+    "router_a10",
+    "router_summary",
+]

backend/app/modules/owasp_demo/routers/a01_access_control.py

@@ -0,0 +1,243 @@
+"""
+A01:2025 - Broken Access Control
+
+Demonstrates IDOR, mass assignment, and missing function-level access control vulnerabilities.
+"""
+
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from sqlmodel import select
+
+from app.modules.owasp_demo.models import SecretDocument, UserNote
+from app.modules.owasp_demo.schemas import (
+    DocumentPublic,
+    NotePublic,
+    UserProfileUpdate,
+)
+from app.modules.rbac.service import UserRoleService
+from app.modules.shared import CurrentUser, SessionDep
+from app.modules.users.models import User
+
+router = APIRouter(
+    prefix="/a01",
+    tags=["A01 - Broken Access Control"],
+)
+
+
+@router.get("/vulnerable/documents", response_model=DocumentPublic)
+def get_document_vulnerable(
+    session: SessionDep,
+    doc_id: uuid.UUID | None = None,
+) -> Any:
+    """
+    VULNERABLE: Insecure Direct Object Reference (IDOR)
+
+    Problem: No authentication or authorization check.
+    Anyone can access any document by guessing/enumerating the ID.
+
+    Attack: Simply change the doc_id in the URL to access other users' documents.
+    """
+    if doc_id:
+        document = session.get(SecretDocument, doc_id)
+    else:
+        # Auto-fetch first available document for easy testing
+        document = session.exec(select(SecretDocument).limit(1)).first()
+
+    if not document:
+        raise HTTPException(status_code=404, detail="Document not found")
+    return document
+
+
+@router.get("/secure/documents", response_model=DocumentPublic)
+def get_document_secure(
+    session: SessionDep,
+    current_user: CurrentUser,
+    doc_id: uuid.UUID | None = None,
+) -> Any:
+    """
+    SECURE: Proper access control implementation with RBAC
+
+    Fix:
+    1. Requires authentication (CurrentUser dependency)
+    2. Verifies the user owns the document OR has documents:read permission OR is superuser
+    """
+    if doc_id:
+        document = session.get(SecretDocument, doc_id)
+    else:
+        # Auto-fetch first available document for easy testing
+        document = session.exec(select(SecretDocument).limit(1)).first()
+
+    if not document:
+        raise HTTPException(status_code=404, detail="Document not found")
+
+    if document.owner_id != current_user.id and not current_user.is_superuser:
+        user_role_service = UserRoleService(session)
+        if not user_role_service.user_has_permission(current_user.id, "documents:read"):
+            raise HTTPException(status_code=403, detail="Access denied")
+
+    return document
+
+
+@router.get("/vulnerable/notes", response_model=NotePublic)
+def get_note_idor_vulnerable(
+    session: SessionDep,
+    current_user: CurrentUser,  # noqa: ARG001
+    note_id: int | None = None,
+) -> Any:
+    """
+    VULNERABLE: IDOR with sequential IDs
+
+    Problem: Uses sequential integer IDs that are easy to enumerate.
+    Even with authentication, there's no ownership check.
+
+    Attack: Try note_id=1, note_id=2, etc. to access other users' notes.
+    """
+    if note_id:
+        note = session.get(UserNote, note_id)
+    else:
+        # Auto-fetch first available note for easy testing
+        note = session.exec(select(UserNote).limit(1)).first()
+
+    if not note:
+        raise HTTPException(status_code=404, detail="Note not found")
+    return note
+
+
+@router.get("/secure/notes", response_model=NotePublic)
+def get_note_secure(
+    session: SessionDep,
+    current_user: CurrentUser,
+    note_id: int | None = None,
+) -> Any:
+    """
+    SECURE: Proper IDOR prevention with RBAC
+
+    Fix: Always verify ownership OR check RBAC permissions before returning data.
+    """
+    if note_id:
+        note = session.get(UserNote, note_id)
+    else:
+        # Auto-fetch first available note for easy testing
+        note = session.exec(select(UserNote).limit(1)).first()
+
+    if not note:
+        raise HTTPException(status_code=404, detail="Note not found")
+
+    if note.owner_id != current_user.id and not current_user.is_superuser:
+        user_role_service = UserRoleService(session)
+        if not user_role_service.user_has_permission(current_user.id, "notes:read"):
+            raise HTTPException(status_code=403, detail="Access denied")
+
+    return note
+
+
+@router.patch("/vulnerable/users/profile")
+def update_profile_mass_assignment(
+    session: SessionDep,
+    current_user: CurrentUser,
+    profile: UserProfileUpdate,
+) -> dict[str, Any]:
+    """
+    VULNERABLE: Mass Assignment / Privilege Escalation
+
+    Problem: Accepts is_superuser and is_active fields from user input,
+    allowing users to grant themselves admin privileges.
+
+    Attack: Send {"is_superuser": true} in the request body.
+    """
+    user = session.get(User, current_user.id)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    update_data = profile.model_dump(exclude_unset=True)
+    for field, value in update_data.items():
+        setattr(user, field, value)
+
+    session.add(user)
+    session.commit()
+    session.refresh(user)
+
+    return {
+        "message": "Profile updated",
+        "is_superuser": user.is_superuser,
+        "is_active": user.is_active,
+    }
+
+
+@router.patch("/secure/users/profile")
+def update_profile_secure(
+    session: SessionDep,
+    current_user: CurrentUser,
+    profile: UserProfileUpdate,
+) -> dict[str, Any]:
+    """
+    SECURE: Protected against mass assignment
+
+    Fix: Explicitly whitelist which fields can be updated.
+    Never trust user input for sensitive fields.
+    """
+    user = session.get(User, current_user.id)
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    ALLOWED_FIELDS = {"full_name", "email"}
+
+    update_data = profile.model_dump(exclude_unset=True)
+    for field, value in update_data.items():
+        if field in ALLOWED_FIELDS:
+            setattr(user, field, value)
+
+    session.add(user)
+    session.commit()
+    session.refresh(user)
+
+    return {"message": "Profile updated safely", "full_name": user.full_name}
+
+
+@router.get("/vulnerable/admin/users")
+def list_users_no_auth(session: SessionDep) -> dict[str, Any]:
+    """
+    VULNERABLE: Missing Function Level Access Control
+
+    Problem: Admin endpoint accessible without authentication.
+
+    Attack: Anyone can access /admin/users to see all user data.
+    """
+    statement = select(User)
+    users = session.exec(statement).all()
+    return {
+        "count": len(users),
+        "users": [
+            {"id": str(u.id), "email": u.email, "is_superuser": u.is_superuser}
+            for u in users
+        ],
+    }
+
+
+@router.get("/secure/admin/users")
+def list_users_secure(
+    session: SessionDep,
+    current_user: CurrentUser,
+) -> dict[str, Any]:
+    """
+    SECURE: Proper admin access control with RBAC
+
+    Fix: Require authentication AND appropriate RBAC permission (users:read).
+    Demonstrates proper function-level access control using role-based permissions.
+    """
+    if not current_user.is_superuser:
+        user_role_service = UserRoleService(session)
+        if not user_role_service.user_has_permission(current_user.id, "users:read"):
+            raise HTTPException(status_code=403, detail="Access denied")
+
+    statement = select(User)
+    users = session.exec(statement).all()
+    return {
+        "count": len(users),
+        "users": [
+            {"id": str(u.id), "email": u.email, "is_superuser": u.is_superuser}
+            for u in users
+        ],
+    }