Security hardening (maziggy/bambuddy-security #1)

backend/app/api/routes/groups.py

@@ -5,7 +5,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
-from backend.app.core.auth import RequirePermissionIfAuthEnabled
+from backend.app.core.auth import RequireAdminIfAuthEnabled, RequirePermissionIfAuthEnabled
 from backend.app.core.database import get_db
 from backend.app.core.permissions import (
     ALL_PERMISSIONS,
@@ -87,6 +87,7 @@ async def list_groups(
 @router.post("/", response_model=GroupResponse, status_code=status.HTTP_201_CREATED)
 async def create_group(
     group_data: GroupCreate,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.GROUPS_CREATE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -135,7 +136,8 @@ async def get_group(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.GROUPS_READ),
     db: AsyncSession = Depends(get_db),
 ):
-    """Get a group by ID with user list."""
+    """Get a group by ID with user list. Read-only — gated on
+    ``GROUPS_READ`` only."""
     result = await db.execute(select(Group).where(Group.id == group_id).options(selectinload(Group.users)))
     group = result.scalar_one_or_none()
     if not group:
@@ -161,6 +163,7 @@ async def get_group(
 async def update_group(
     group_id: int,
     group_data: GroupUpdate,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.GROUPS_UPDATE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -193,6 +196,16 @@ async def update_group(
         group.description = group_data.description
 
     if group_data.permissions is not None:
+        # System groups (Administrators in particular) have fixed permission
+        # sets that the app depends on — stripping them is a denial-of-
+        # service vector that even admin callers shouldn't trigger by
+        # accident through the generic edit form. Mirrors the rename block
+        # immediately above.
+        if group.is_system:
+            raise HTTPException(
+                status_code=status.HTTP_400_BAD_REQUEST,
+                detail="Cannot modify permissions of system groups",
+            )
         # Validate permissions
         invalid_perms = [p for p in group_data.permissions if p not in ALL_PERMISSIONS]
         if invalid_perms:
@@ -220,6 +233,7 @@ async def update_group(
 @router.delete("/{group_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_group(
     group_id: int,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.GROUPS_DELETE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -246,6 +260,7 @@ async def delete_group(
 async def add_user_to_group(
     group_id: int,
     user_id: int,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.GROUPS_UPDATE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -283,6 +298,7 @@ async def add_user_to_group(
 async def remove_user_from_group(
     group_id: int,
     user_id: int,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.GROUPS_UPDATE),
     db: AsyncSession = Depends(get_db),
 ):

backend/app/api/routes/users.py

@@ -12,6 +12,7 @@ from backend.app.api.routes.settings import get_external_login_url
 from backend.app.core.auth import (
     ALGORITHM,
     SECRET_KEY,
+    RequireAdminIfAuthEnabled,
     RequirePermissionIfAuthEnabled,
     get_current_user_optional,
     get_password_hash,
@@ -66,7 +67,13 @@ async def list_users(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ),
     db: AsyncSession = Depends(get_db),
 ):
-    """List all users."""
+    """List all users.
+
+    Read-only — gated on ``USERS_READ`` only. Operator-visible UIs
+    (Stats filter-by-user, Archives Print Log username column, File
+    Manager username autocomplete) consume this endpoint via custom-
+    group ``users:read`` grants without admin role. The admin-only
+    boundary lives on the write endpoints below."""
     result = await db.execute(select(User).options(selectinload(User.groups)).order_by(User.created_at))
     users = result.scalars().all()
     return [_user_to_response(user) for user in users]
@@ -76,6 +83,7 @@ async def list_users(
 @router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
 async def create_user(
     user_data: UserCreate,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_CREATE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -185,7 +193,7 @@ async def get_user(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ),
     db: AsyncSession = Depends(get_db),
 ):
-    """Get a user by ID."""
+    """Get a user by ID. Read-only — gated on ``USERS_READ`` only."""
     result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups)))
     user = result.scalar_one_or_none()
     if not user:
@@ -201,6 +209,7 @@ async def get_user(
 async def update_user(
     user_id: int,
     user_data: UserUpdate,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_UPDATE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -311,7 +320,8 @@ async def get_user_items_count(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ),
     db: AsyncSession = Depends(get_db),
 ):
-    """Get count of items created by this user."""
+    """Get count of items created by this user. Read-only — gated on
+    ``USERS_READ`` only."""
     # Verify user exists
     result = await db.execute(select(User).where(User.id == user_id))
     if not result.scalar_one_or_none():
@@ -350,6 +360,7 @@ async def get_user_items_count(
 async def delete_user(
     user_id: int,
     delete_items: bool = Query(False, description="Delete all items created by this user"),
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_DELETE),
     db: AsyncSession = Depends(get_db),
 ):

backend/app/core/auth.py

@@ -1019,13 +1019,17 @@ def require_admin_if_auth_enabled():
     key" — the inner ``admin_checker`` then treated ``None`` as auth-
     disabled and admitted the caller. If any route had ever adopted this
     dep, any API key with no scope flags set would have satisfied an
-    admin requirement.
-
-    Today no route uses this dep, but rather than leave the footgun
-    armed, the dep is rewritten to distinguish the two cases by
-    consulting ``is_auth_enabled`` directly and rejecting API-keyed
-    requests with 403. "Admin" requires a user-identity role, which API
-    keys do not carry.
+    admin requirement. The dep distinguishes the two cases by consulting
+    ``is_auth_enabled`` directly and rejecting API-keyed requests with
+    403. "Admin" requires a user-identity role, which API keys do not
+    carry.
+
+    Admin semantics: uses ``User.is_admin`` (``role == "admin"`` OR
+    Administrators-group membership) so a default-install operator who
+    was made admin by being added to Administrators rather than by
+    flipping the legacy role column passes. Earlier this check looked
+    only at ``role`` and would have locked group-only admins out of the
+    user-management routes once those routes started requiring it.
     """
 
     async def admin_checker(
@@ -1091,7 +1095,7 @@ def require_admin_if_auth_enabled():
                     detail="Could not validate credentials",
                     headers={"WWW-Authenticate": "Bearer"},
                 )
-            if user.role != "admin":
+            if not user.is_admin:
                 raise HTTPException(
                     status_code=status.HTTP_403_FORBIDDEN,
                     detail="Requires admin role",

backend/tests/integration/test_users_groups_privilege_escalation.py

@@ -0,0 +1,388 @@
+"""Privilege-escalation regression suite for the users/groups admin boundary.
+
+The intent declared in ``permissions.py`` is that USERS_* / GROUPS_* are
+admin-level capabilities — the comments literally say "(admin-level)".
+The original implementation enforced ONLY the permission, not admin role.
+Any user holding USERS_UPDATE (or USERS_CREATE / GROUPS_UPDATE /
+GROUPS_CREATE) could grant themselves admin via the management routes.
+
+This suite reproduces every attack vector from the disclosure and pins
+the fail-closed behaviour. Each negative test grants the operator the
+minimum permission needed to *reach* the route gate, then asserts the
+admin gate blocks them. A companion positive test verifies the same
+operation succeeds with an admin token (so the admin gate doesn't
+over-block real flows).
+
+Default-install operators do NOT have USERS_* / GROUPS_* (see
+``DEFAULT_GROUPS``), so default deployments were never vulnerable
+unless an admin had explicitly granted the permission to a custom
+group — but anyone in that position would expect the boundary the
+comments described.
+"""
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import select
+
+from backend.app.models.group import Group
+
+
+async def _setup_admin(async_client: AsyncClient, username: str = "secadmin") -> str:
+    await async_client.post(
+        "/api/v1/auth/setup",
+        json={"auth_enabled": True, "admin_username": username, "admin_password": "AdminPass1!"},
+    )
+    login = await async_client.post(
+        "/api/v1/auth/login",
+        json={"username": username, "password": "AdminPass1!"},
+    )
+    return login.json()["access_token"]
+
+
+async def _create_operator_with_perms(
+    async_client: AsyncClient,
+    admin_token: str,
+    db_session,
+    *,
+    username: str,
+    permissions: list[str],
+) -> tuple[str, int]:
+    """Create a non-admin user, drop them in a custom group with exactly
+    the requested permissions, return (token, user_id).
+
+    The operator is intentionally NOT an admin and NOT in the Administrators
+    group — they hold ONLY the listed permission strings. Mirrors the exact
+    deployment shape the security engineer described: an operator gifted
+    one admin-level permission via a custom group ends up able to escalate
+    to full admin without the gate.
+    """
+    headers = {"Authorization": f"Bearer {admin_token}"}
+
+    # Create a custom group carrying just the requested permissions.
+    grp_resp = await async_client.post(
+        "/api/v1/groups/",
+        headers=headers,
+        json={"name": f"escalation_test_{username}", "permissions": permissions},
+    )
+    assert grp_resp.status_code == 201, grp_resp.text
+    gid = grp_resp.json()["id"]
+
+    # Create a regular (role="user") user.
+    user_resp = await async_client.post(
+        "/api/v1/users/",
+        headers=headers,
+        json={"username": username, "password": "OpPass1234!", "role": "user", "group_ids": [gid]},
+    )
+    assert user_resp.status_code == 201, user_resp.text
+    uid = user_resp.json()["id"]
+
+    # Confirm the operator is NOT admin in the response shape.
+    assert user_resp.json()["is_admin"] is False
+
+    login = await async_client.post(
+        "/api/v1/auth/login",
+        json={"username": username, "password": "OpPass1234!"},
+    )
+    assert login.status_code == 200
+    return login.json()["access_token"], uid
+
+
+async def _admin_group_id(db_session) -> int:
+    result = await db_session.execute(select(Group).where(Group.name == "Administrators"))
+    return result.scalar_one().id
+
+
+# ---------------------------------------------------------------------------
+# 1. PATCH /users/{id} {role: "admin"} — USERS_UPDATE holder cannot
+# self-promote
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_users_update_holder_cannot_set_role_to_admin(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    op_token, op_id = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="op1", permissions=["users:update"]
+    )
+
+    resp = await async_client.patch(
+        f"/api/v1/users/{op_id}",
+        headers={"Authorization": f"Bearer {op_token}"},
+        json={"role": "admin"},
+    )
+    assert resp.status_code == 403
+
+    # And the operator is not admin in the DB after the attempted patch.
+    from backend.app.models.user import User
+
+    result = await db_session.execute(select(User).where(User.id == op_id))
+    user = result.scalar_one()
+    assert user.role == "user"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_users_update_holder_cannot_target_other_user(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    op_token, _ = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="op2", permissions=["users:update"]
+    )
+    # Create a separate target user.
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    target = await async_client.post(
+        "/api/v1/users/",
+        headers=headers,
+        json={"username": "target", "password": "TargetPass1!", "role": "user"},
+    )
+    target_id = target.json()["id"]
+
+    # Operator attempts to elevate target to admin.
+    resp = await async_client.patch(
+        f"/api/v1/users/{target_id}",
+        headers={"Authorization": f"Bearer {op_token}"},
+        json={"role": "admin"},
+    )
+    assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# 2. POST /users/ {role: "admin"} — USERS_CREATE holder cannot create admin
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_users_create_holder_cannot_create_admin(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    op_token, _ = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="op3", permissions=["users:create"]
+    )
+
+    resp = await async_client.post(
+        "/api/v1/users/",
+        headers={"Authorization": f"Bearer {op_token}"},
+        json={"username": "newadmin", "password": "NewAdmin1!", "role": "admin"},
+    )
+    assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# 3. PATCH /groups/{id} {permissions: [...]} — GROUPS_UPDATE holder cannot
+# rewrite a group to admin-equivalent
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_groups_update_holder_cannot_rewrite_permissions(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    op_token, _ = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="op4", permissions=["groups:update"]
+    )
+
+    # Admin creates a target group; operator tries to grant it everything.
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    create = await async_client.post(
+        "/api/v1/groups/",
+        headers=headers,
+        json={"name": "innocent", "permissions": ["printers:read"]},
+    )
+    gid = create.json()["id"]
+
+    from backend.app.core.permissions import ALL_PERMISSIONS
+
+    resp = await async_client.patch(
+        f"/api/v1/groups/{gid}",
+        headers={"Authorization": f"Bearer {op_token}"},
+        json={"permissions": ALL_PERMISSIONS},
+    )
+    assert resp.status_code == 403
+
+    # And the group still has its original (narrow) permissions.
+    result = await db_session.execute(select(Group).where(Group.id == gid))
+    assert result.scalar_one().permissions == ["printers:read"]
+
+
+# ---------------------------------------------------------------------------
+# 4. POST /groups/ {permissions: [...]} — GROUPS_CREATE holder cannot create
+# an admin-equivalent group
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_groups_create_holder_cannot_create_admin_equivalent(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    op_token, _ = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="op5", permissions=["groups:create"]
+    )
+    from backend.app.core.permissions import ALL_PERMISSIONS
+
+    resp = await async_client.post(
+        "/api/v1/groups/",
+        headers={"Authorization": f"Bearer {op_token}"},
+        json={"name": "shadowadmins", "permissions": ALL_PERMISSIONS},
+    )
+    assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# 5. POST /groups/{admin_gid}/users/{my_id} — GROUPS_UPDATE holder cannot
+# self-add to Administrators
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_groups_update_holder_cannot_self_add_to_administrators(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    op_token, op_id = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="op6", permissions=["groups:update"]
+    )
+    admin_gid = await _admin_group_id(db_session)
+
+    resp = await async_client.post(
+        f"/api/v1/groups/{admin_gid}/users/{op_id}",
+        headers={"Authorization": f"Bearer {op_token}"},
+    )
+    assert resp.status_code == 403
+
+
+# ---------------------------------------------------------------------------
+# 6. PATCH /groups/{system_gid} — even an admin must not be able to strip
+# the Administrators group's permissions (DoS guard).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_admin_cannot_strip_administrators_group_permissions(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    admin_gid = await _admin_group_id(db_session)
+
+    resp = await async_client.patch(
+        f"/api/v1/groups/{admin_gid}",
+        headers=headers,
+        json={"permissions": []},
+    )
+    assert resp.status_code == 400
+    assert "system groups" in resp.json()["detail"].lower()
+
+    # Untouched in DB.
+    result = await db_session.execute(select(Group).where(Group.id == admin_gid))
+    grp = result.scalar_one()
+    assert len(grp.permissions or []) > 0
+
+
+# ---------------------------------------------------------------------------
+# Positive companions — admin should succeed on each route (the admin gate
+# must not over-block normal admin flows).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_admin_can_still_perform_user_role_change(async_client: AsyncClient, db_session):
+    admin_token = await _setup_admin(async_client)
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    target = await async_client.post(
+        "/api/v1/users/",
+        headers=headers,
+        json={"username": "promoteme", "password": "Promote1!", "role": "user"},
+    )
+    tid = target.json()["id"]
+
+    resp = await async_client.patch(
+        f"/api/v1/users/{tid}",
+        headers=headers,
+        json={"role": "admin"},
+    )
+    assert resp.status_code == 200
+    assert resp.json()["role"] == "admin"
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_administrators_group_member_passes_admin_gate(async_client: AsyncClient, db_session):
+    """A user whose admin status comes from Administrators-group membership
+    rather than the legacy ``role`` column must pass the admin gate. The
+    canonical signal is ``User.is_admin``, not ``role == 'admin'``.
+
+    Uses a write endpoint (PATCH /users/{id} {role}) since the admin gate
+    lives on writes only — reads stay at ``USERS_READ`` so operator UIs
+    (Stats filter-by-user, Archives Print Log, File Manager username
+    autocomplete) keep working for non-admin operators who hold the
+    read permission via a custom group."""
+    admin_token = await _setup_admin(async_client)
+    headers = {"Authorization": f"Bearer {admin_token}"}
+    admin_gid = await _admin_group_id(db_session)
+
+    # Create a regular user, then add them to Administrators.
+    user_resp = await async_client.post(
+        "/api/v1/users/",
+        headers=headers,
+        json={"username": "groupadmin", "password": "GroupAdmin1!", "role": "user"},
+    )
+    uid = user_resp.json()["id"]
+    add = await async_client.post(f"/api/v1/groups/{admin_gid}/users/{uid}", headers=headers)
+    assert add.status_code == 204
+
+    # Also create a separate target user to mutate (cleaner than self-modify).
+    target_resp = await async_client.post(
+        "/api/v1/users/",
+        headers=headers,
+        json={"username": "target_member", "password": "Target1234!", "role": "user"},
+    )
+    target_id = target_resp.json()["id"]
+
+    login = await async_client.post("/api/v1/auth/login", json={"username": "groupadmin", "password": "GroupAdmin1!"})
+    group_admin_token = login.json()["access_token"]
+
+    # Through an admin-gated write route — must succeed.
+    resp = await async_client.patch(
+        f"/api/v1/users/{target_id}",
+        headers={"Authorization": f"Bearer {group_admin_token}"},
+        json={"is_active": False},
+    )
+    assert resp.status_code == 200
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_users_read_remains_delegable_to_non_admin(async_client: AsyncClient, db_session):
+    """Operator-visible UIs (Stats filter-by-user, Archives Print Log
+    username column, File Manager username autocomplete) reach
+    ``GET /users/`` for non-admin operators when a deployment granted
+    them ``users:read`` via a custom group. The admin gate must NOT
+    apply to read endpoints — only to writes."""
+    admin_token = await _setup_admin(async_client)
+    op_token, _ = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="reader", permissions=["users:read"]
+    )
+
+    resp = await async_client.get("/api/v1/users/", headers={"Authorization": f"Bearer {op_token}"})
+    assert resp.status_code == 200
+    # Operator is in the list with is_admin=False — confirms the read is
+    # working AND the operator hasn't escalated.
+    me = next(u for u in resp.json() if u["username"] == "reader")
+    assert me["is_admin"] is False
+
+
+@pytest.mark.asyncio
+@pytest.mark.integration
+async def test_groups_read_remains_delegable_to_non_admin(async_client: AsyncClient, db_session):
+    """Companion to ``users:read``. ``GET /groups/`` + ``GET /groups/
+    permissions`` stay reachable to non-admin operators with the read
+    permission. Used by setup wizards / informational lookups."""
+    admin_token = await _setup_admin(async_client)
+    op_token, _ = await _create_operator_with_perms(
+        async_client, admin_token, db_session, username="greader", permissions=["groups:read"]
+    )
+
+    headers = {"Authorization": f"Bearer {op_token}"}
+    list_resp = await async_client.get("/api/v1/groups/", headers=headers)
+    assert list_resp.status_code == 200
+    perms_resp = await async_client.get("/api/v1/groups/permissions", headers=headers)
+    assert perms_resp.status_code == 200