Security hardening (maziggy/bambuddy-security #40)

Removing a user's 2FA and searching or provisioning LDAP users are
admin-only, like the other user-account routes. Removing 2FA for an
unknown user returns 404.

backend/app/api/routes/auth.py

@@ -18,6 +18,7 @@ from backend.app.core.auth import (
     ALGORITHM,
     SECRET_KEY,
     Permission,
+    RequireAdminIfAuthEnabled,
     RequirePermissionIfAuthEnabled,
     _is_token_fresh,
     _validate_api_key,
@@ -1607,6 +1608,7 @@ async def get_ldap_status(db: AsyncSession = Depends(get_db)):
 @router.get("/ldap/search", response_model=list[LDAPSearchResultResponse])
 async def search_ldap_directory(
     q: str,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_CREATE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -1680,6 +1682,7 @@ async def search_ldap_directory(
 @router.post("/ldap/provision", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
 async def provision_ldap_user(
     payload: LDAPProvisionRequest,
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_CREATE),
     db: AsyncSession = Depends(get_db),
 ):
@@ -1691,7 +1694,7 @@ async def provision_ldap_user(
     login path (`_provision_ldap_user`), so behavior stays identical regardless
     of whether the user was created here or on first login.
 
-    Requires USERS_CREATE.
+    Admin-only with USERS_CREATE, like users.create_user.
     """
     from sqlalchemy import func as sa_func
 

backend/app/api/routes/mfa.py

@@ -41,6 +41,7 @@ from sqlalchemy.orm import selectinload, undefer
 from backend.app.api.routes._oidc_helpers import assert_safe_public_https_url
 from backend.app.api.routes.settings import get_setting, set_setting
 from backend.app.core.auth import (
+    RequireAdminIfAuthEnabled,
     RequirePermissionIfAuthEnabled,
     create_access_token,
     get_current_active_user,
@@ -1279,11 +1280,14 @@ async def verify_2fa(
 async def admin_disable_2fa(
     user_id: int,
     body: AdminDisable2FARequest = Body(default_factory=AdminDisable2FARequest),
+    _admin: User | None = RequireAdminIfAuthEnabled(),
     current_user: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_UPDATE),
     db: AsyncSession = Depends(get_db),
 ) -> dict:
     """Admin endpoint: disable all 2FA for a given user.
 
+    Admin-only, like every other change to a user account (users.update_user).
+
     Nit 3: Requires the admin's own password as a re-auth step (matching how
     disable_email_otp protects a user's own 2FA removal). OIDC/LDAP-only admins
     (no local password_hash) are exempt.
@@ -1293,6 +1297,10 @@ async def admin_disable_2fa(
         if not body.admin_password or not verify_password(body.admin_password, current_user.password_hash):
             raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Admin password required")
 
+    target_user = (await db.execute(select(User).where(User.id == user_id))).scalar_one_or_none()
+    if target_user is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
+
     # Delete TOTP record
     await db.execute(delete(UserTOTP).where(UserTOTP.user_id == user_id))
 
@@ -1308,21 +1316,18 @@ async def admin_disable_2fa(
 
     # I2: Invalidate existing JWTs for the target user by bumping password_changed_at.
     # Without this, a stolen token remains valid after 2FA removal.
-    target_user = (await db.execute(select(User).where(User.id == user_id))).scalar_one_or_none()
-    if target_user:
-        target_user.password_changed_at = datetime.now(timezone.utc)
-        # The WebSocket and camera tokens too: reusable for an hour, they would
-        # reconnect a closed dashboard straight away.
-        from backend.app.api.routes.users import drop_pending_tokens
+    target_user.password_changed_at = datetime.now(timezone.utc)
+    # The WebSocket and camera tokens too: reusable for an hour, they would
+    # reconnect a closed dashboard straight away.
+    from backend.app.api.routes.users import drop_pending_tokens
 
-        await drop_pending_tokens(db, target_user.username)
+    await drop_pending_tokens(db, target_user.username)
 
     await db.commit()
-    if target_user:
-        # And their open dashboards: they are never signed in again.
-        from backend.app.core.websocket import ws_manager
+    # And their open dashboards: they are never signed in again.
+    from backend.app.core.websocket import ws_manager
 
-        await ws_manager.close_user(user_id)
+    await ws_manager.close_user(user_id)
     actor = current_user.username if current_user else "anonymous"
     logger.info("Admin %s disabled all 2FA for user_id=%d", actor, user_id)
     return {"message": "2FA disabled for user"}

backend/tests/integration/test_ldap_provision.py

@@ -531,3 +531,60 @@ class TestLdapFirstLoginProvisionsOnce:
             second = await async_client.post("/api/v1/auth/login", json={"username": "tofm", "password": "irrelevant"})
         assert second.status_code == 200
         assert second.json()["user"]["email"] == "new@test.com"
+
+
+class TestLdapUserCreationIsAdminOnly:
+    """Creating a user is admin-only (users.create_user); the LDAP way in is too."""
+
+    @staticmethod
+    async def _creator_token(async_client: AsyncClient, admin_token: str) -> str:
+        auth = {"Authorization": f"Bearer {admin_token}"}
+        group = await async_client.post(
+            "/api/v1/groups", json={"name": "User creators", "permissions": ["users:create"]}, headers=auth
+        )
+        assert group.status_code == 201, group.text
+        # pragma: allowlist secret — test fixture only, not a real credential
+        password = "Creator1pass!"  # noqa: S105
+        created = await async_client.post(
+            "/api/v1/users",
+            json={"username": "ldapcreator", "password": password, "group_ids": [group.json()["id"]]},
+            headers=auth,
+        )
+        assert created.status_code == 201, created.text
+        login = await async_client.post("/api/v1/auth/login", json={"username": "ldapcreator", "password": password})
+        return login.json()["access_token"]
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_provision_refuses_a_non_admin(
+        self, async_client: AsyncClient, admin_token: str, db_session: AsyncSession
+    ):
+        await _seed_ldap_settings(db_session)
+        token = await self._creator_token(async_client, admin_token)
+        fake_ldap = LDAPUserInfo(username="jdoe", email="jdoe@test.com", display_name=None, groups=[])
+
+        with patch("backend.app.services.ldap_service.lookup_ldap_user", return_value=fake_ldap):
+            response = await async_client.post(
+                "/api/v1/auth/ldap/provision",
+                json={"username": "jdoe"},
+                headers={"Authorization": f"Bearer {token}"},
+            )
+
+        assert response.status_code == 403
+        assert (await db_session.execute(select(User).where(User.username == "jdoe"))).scalar_one_or_none() is None
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_search_refuses_a_non_admin(
+        self, async_client: AsyncClient, admin_token: str, db_session: AsyncSession
+    ):
+        await _seed_ldap_settings(db_session)
+        token = await self._creator_token(async_client, admin_token)
+
+        with patch("backend.app.services.ldap_service.search_ldap_users", return_value=[]) as search:
+            response = await async_client.get(
+                "/api/v1/auth/ldap/search?q=jdoe", headers={"Authorization": f"Bearer {token}"}
+            )
+
+        assert response.status_code == 403
+        search.assert_not_called()

backend/tests/integration/test_mfa_api.py

@@ -654,19 +654,15 @@ class TestAdminDisable2FA:
 
     @pytest.mark.asyncio
     @pytest.mark.integration
-    async def test_admin_disable_requires_admin(self, async_client: AsyncClient):
-        """Only admins can use the admin disable endpoint."""
-        # The only user in a fresh setup IS admin, so just check the 404 path
+    async def test_admin_disable_unknown_user_is_404(self, async_client: AsyncClient):
         token = await _setup_and_login(async_client, "admincheck", "admincheck123")
-        # Try to disable for a non-existent user_id — should get 200 (no-op) or 404
         response = await async_client.request(
             "DELETE",
             "/api/v1/auth/2fa/admin/99999",
             json={"admin_password": _norm_pw("admincheck123")},
             headers=_auth_header(token),
         )
-        # Admin users succeed regardless (returns 200 even if user doesn't exist)
-        assert response.status_code == 200
+        assert response.status_code == 404
 
     @pytest.mark.asyncio
     @pytest.mark.integration
@@ -1156,6 +1152,37 @@ class TestAdminDisableNonAdminRejection:
         )
         assert resp.status_code == 403
 
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_users_update_permission_alone_is_not_enough(self, async_client: AsyncClient):
+        """Editing users is admin-only; so is removing their 2FA."""
+        admin_token = await _setup_and_login(async_client, "adminperm2fa", "adminperm2fa1")
+        admin_id = (await async_client.get("/api/v1/auth/me", headers=_auth_header(admin_token))).json()["id"]
+
+        group = await async_client.post(
+            "/api/v1/groups",
+            json={"name": "User managers", "permissions": ["users:read", "users:update"]},
+            headers=_auth_header(admin_token),
+        )
+        assert group.status_code == 201, group.text
+        created = await async_client.post(
+            "/api/v1/users",
+            json={"username": "usermgr2fa", "password": "Usermgr2fa1!", "group_ids": [group.json()["id"]]},
+            headers=_auth_header(admin_token),
+        )
+        assert created.status_code == 201, created.text
+        login = await async_client.post(LOGIN_URL, json={"username": "usermgr2fa", "password": "Usermgr2fa1!"})
+        manager_token = login.json()["access_token"]
+
+        resp = await async_client.request(
+            "DELETE",
+            f"/api/v1/auth/2fa/admin/{admin_id}",
+            json={"admin_password": "Usermgr2fa1!"},
+            headers=_auth_header(manager_token),
+        )
+
+        assert resp.status_code == 403
+
 
 # ===========================================================================
 # Regenerate backup codes