diff --git a/backend/app/api/users.py b/backend/app/api/users.py new file mode 100644 index 0000000..d868f0c --- /dev/null +++ b/backend/app/api/users.py @@ -0,0 +1,146 @@ +"""Admin user management: list, create, delete admin users.""" + +from fastapi import APIRouter, Depends, HTTPException, status +from pydantic import BaseModel, Field +from sqlalchemy import select +from sqlalchemy.orm import joinedload +from sqlalchemy.ext.asyncio import AsyncSession + +from app.auth import get_current_admin_user, hash_password +from app.database import get_session +from app.user_models import User, Role, user_roles + +router = APIRouter() + + +# ── Schemas ──────────────────────────────────────────────────── + +class AdminUserResponse(BaseModel): + id: int + email: str + display_name: str + auth_provider: str + is_admin: bool + + model_config = {"from_attributes": True} + + +class CreateAdminRequest(BaseModel): + email: str = Field(..., pattern=r'^[^@\s]+@[^@>\s]+.[^@\s.]+$') + display_name: str + password: str + + +class UpdateAdminRequest(BaseModel): + display_name: str | None = None + + +# ── Routes ───────────────────────────────────────────────────── + +@router.get("/users", response_model=list[AdminUserResponse]) +async def list_admin_users(session: AsyncSession = Depends(get_session), current_user: User = Depends(get_current_admin_user)): + """List all users with admin access.""" + result = await session.execute( + select(User).options(joinedload(User.roles)) + ) + users = result.scalars().unique().all() + # Filter to only admin users + admins = [u for u in users if u.is_admin_effective] + return admins + + +@router.post("/users", response_model=AdminUserResponse, status_code=status.HTTP_201_CREATED) +async def create_admin_user( + payload: CreateAdminRequest, + session: AsyncSession = Depends(get_session), + current_user: User = Depends(get_current_admin_user), +): + """Create a new admin user. Assigns the 'admin' role automatically.""" + # Check if user already exists + existing = await session.execute(select(User).where(User.email == payload.email)) + if existing.scalar_one_or_none(): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=f"User with email {payload.email} already exists", + ) + + # Ensure admin role exists + role_result = await session.execute(select(Role).where(Role.name == "admin")) + admin_role = role_result.scalar_one_or_none() + if admin_role is None: + admin_role = Role(name="admin", description="Full administrative access") + session.add(admin_role) + await session.flush() + + # Create user with hashed password + user = User( + email=payload.email, + display_name=payload.display_name, + auth_provider="local", + password_hash=hash_password(payload.password), + is_admin=True, + ) + user.roles.append(admin_role) + session.add(user) + await session.commit() + await session.refresh(user) + return user + + +@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT) +async def delete_admin_user( + user_id: int, + session: AsyncSession = Depends(get_session), + current_user: User = Depends(get_current_admin_user), +): + """Delete an admin user. Cannot delete yourself. At least one admin must remain.""" + # Prevent self-deletion + if user_id == current_user.id: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Cannot delete your own account", + ) + + result = await session.execute( + select(User).options(joinedload(User.roles)).where(User.id == user_id) + ) + user = result.unique().scalar_one_or_none() + if user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") + + # Ensure at least one admin remains (don't count the user being deleted) + other_admins = await session.execute( + select(User).options(joinedload(User.roles)).where(User.id != user_id) + ) + admin_count = sum(1 for u in other_admins.scalars().unique().all() if u.is_admin_effective) + if admin_count == 0: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Cannot delete the last admin user", + ) + + # Remove all roles + user.roles.clear() + await session.delete(user) + await session.commit() + + +@router.put("/users/{user_id}", response_model=AdminUserResponse) +async def update_admin_user( + user_id: int, + payload: UpdateAdminRequest, + session: AsyncSession = Depends(get_session), + current_user: User = Depends(get_current_admin_user), +): + """Update admin user details.""" + result = await session.execute(select(User).where(User.id == user_id)) + user = result.scalar_one_or_none() + if user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") + + if payload.display_name is not None: + user.display_name = payload.display_name + + await session.commit() + await session.refresh(user) + return user diff --git a/backend/app/main.py b/backend/app/main.py index 160ae8c..71fcf5d 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -234,6 +234,10 @@ app.include_router(stats.router, prefix="/api/stats", tags=["stats"]) app.include_router(mobile_builds.router, prefix="/api/mobile-builds", tags=["mobile-builds"]) app.include_router(theme.router, prefix="/api/theme", tags=["theme"]) +# Admin user management (available in all environments) +from app.api import users +app.include_router(users.router, prefix="/api/admin", tags=["admin-users"]) + # Dev-only admin router (disabled in production) if settings.ENV != "production": from app.api import admin diff --git a/backend/tests/test_admin_users.py b/backend/tests/test_admin_users.py new file mode 100644 index 0000000..5548c37 --- /dev/null +++ b/backend/tests/test_admin_users.py @@ -0,0 +1,229 @@ +"""Tests for /api/admin/users admin user management endpoints.""" + +import pytest +from httpx import AsyncClient, ASGITransport +from jose import jwt +from datetime import datetime, timedelta, timezone + +from app.auth import hash_password +from app.config import settings +from app.database import async_session +from app.main import app +from app.user_models import User, Role + + +@pytest.fixture +async def admin_token(): + """Create a valid admin JWT.""" + expire = datetime.now(timezone.utc) + timedelta(minutes=30) + payload = { + "sub": "99999", + "is_admin": True, + "exp": expire, + } + return jwt.encode(payload, settings.JWT_SECRET_KEY, algorithm="HS256") + + +@pytest.fixture +async def seed_admin_user(admin_token: str): + """Seed the test admin user that matches the JWT sub claim.""" + async with async_session() as session: + user = User( + id=99999, + email="testadmin@example.com", + display_name="Test Admin", + auth_provider="local", + is_admin=True, + ) + session.add(user) + await session.commit() + return user + + +@pytest.fixture +async def client(): + """Create an AsyncClient for the FastAPI app.""" + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as ac: + yield ac + + +@pytest.mark.asyncio +async def test_list_admin_users_empty(client: AsyncClient, admin_token: str, seed_admin_user: User): + """GET /api/admin/users returns empty list when only JWT user (no others) exists.""" + resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"}) + assert resp.status_code == 200 + # The seed_admin_user IS an admin, so it shows up + data = resp.json() + assert len(data) == 1 + assert data[0]["email"] == "testadmin@example.com" + + +@pytest.mark.asyncio +async def test_list_admin_users_filters_non_admins(client: AsyncClient, admin_token: str, seed_admin_user: User): + """GET /api/admin/users returns only users with admin access.""" + # Add a non-admin user + async with async_session() as session: + regular = User( + id=100, + email="regular@example.com", + display_name="Regular User", + auth_provider="google", + is_admin=False, + ) + session.add(regular) + await session.commit() + + resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"}) + assert resp.status_code == 200 + data = resp.json() + assert len(data) == 1 + assert data[0]["email"] == "testadmin@example.com" + + +@pytest.mark.asyncio +async def test_create_admin_user(client: AsyncClient, admin_token: str, seed_admin_user: User): + """POST /api/admin/users creates a new admin with hashed password.""" + resp = await client.post( + "/api/admin/users", + headers={"Authorization": f"Bearer {admin_token}"}, + json={ + "email": "newadmin@example.com", + "display_name": "New Admin", + "password": "securepass123", + }, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["email"] == "newadmin@example.com" + assert data["display_name"] == "New Admin" + assert data["is_admin"] is True + + +@pytest.mark.asyncio +async def test_create_admin_user_duplicate_email(client: AsyncClient, admin_token: str, seed_admin_user: User): + """POST /api/admin/users returns 409 for duplicate email.""" + # Create first user + await client.post( + "/api/admin/users", + headers={"Authorization": f"Bearer {admin_token}"}, + json={ + "email": "dup@example.com", + "display_name": "Dup Admin", + "password": "pass123", + }, + ) + + # Attempt duplicate + resp = await client.post( + "/api/admin/users", + headers={"Authorization": f"Bearer {admin_token}"}, + json={ + "email": "dup@example.com", + "display_name": "Dup Admin 2", + "password": "pass123", + }, + ) + assert resp.status_code == 409 + + +@pytest.mark.asyncio +async def test_delete_admin_user(client: AsyncClient, admin_token: str, seed_admin_user: User): + """DELETE /api/admin/users/{id} removes the user.""" + # Create another admin via the API + await client.post( + "/api/admin/users", + headers={"Authorization": f"Bearer {admin_token}"}, + json={ + "email": "otheradmin@example.com", + "display_name": "Other Admin", + "password": "pass123", + }, + ) + + resp = await client.delete( + "/api/admin/users/88888", + headers={"Authorization": f"Bearer {admin_token}"}, + ) + # The new user won't have id=88888 in the test DB; we need to get the actual ID + # Let's list users first to find the ID + list_resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"}) + users = list_resp.json() + other_user = next((u for u in users if u["email"] == "otheradmin@example.com"), None) + if other_user: + resp = await client.delete( + f"/api/admin/users/{other_user['id']}", + headers={"Authorization": f"Bearer {admin_token}"}, + ) + assert resp.status_code == 204 + + # Verify only seed admin remains + list_resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"}) + remaining = list_resp.json() + assert len(remaining) == 1 + assert remaining[0]["email"] == "testadmin@example.com" + + +@pytest.mark.asyncio +async def test_delete_self_admin_user_forbidden(client: AsyncClient, admin_token: str, seed_admin_user: User): + """DELETE /api/admin/users/{id} blocks deleting yourself.""" + resp = await client.delete( + "/api/admin/users/99999", + headers={"Authorization": f"Bearer {admin_token}"}, + ) + assert resp.status_code == 400 + assert "own account" in resp.json()["detail"].lower() + + +@pytest.mark.asyncio +async def test_delete_last_admin_forbidden(client: AsyncClient, admin_token: str, seed_admin_user: User): + """DELETE /api/admin/users/{id} blocks if it's the last admin.""" + # Try to delete the only admin (our seed user) - self-delete is blocked + resp = await client.delete( + "/api/admin/users/99999", + headers={"Authorization": f"Bearer {admin_token}"}, + ) + assert resp.status_code == 400 + + +@pytest.mark.asyncio +async def test_update_admin_user(client: AsyncClient, admin_token: str, seed_admin_user: User): + """PUT /api/admin/users/{id} updates display_name.""" + resp = await client.put( + "/api/admin/users/99999", + headers={"Authorization": f"Bearer {admin_token}"}, + json={"display_name": "Updated Name"}, + ) + assert resp.status_code == 200 + assert resp.json()["display_name"] == "Updated Name" + + +@pytest.mark.asyncio +async def test_admin_users_requires_auth(client: AsyncClient): + """GET /api/admin/users returns 401 without auth headers.""" + resp = await client.get("/api/admin/users") + assert resp.status_code == 401 + + +@pytest.mark.asyncio +async def test_create_admin_assigns_admin_role(client: AsyncClient, admin_token: str, seed_admin_user: User): + """POST /api/admin/users assigns the 'admin' role and sets is_admin=True.""" + resp = await client.post( + "/api/admin/users", + headers={"Authorization": f"Bearer {admin_token}"}, + json={ + "email": "roletest@example.com", + "display_name": "Role Test", + "password": "pass123", + }, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["is_admin"] is True + + # Verify via list endpoint that the user shows up as admin + list_resp = await client.get("/api/admin/users", headers={"Authorization": f"Bearer {admin_token}"}) + users = list_resp.json() + role_user = next((u for u in users if u["email"] == "roletest@example.com"), None) + assert role_user is not None + assert role_user["is_admin"] is True diff --git a/src/app/admin/admin-users-tab.component.html b/src/app/admin/admin-users-tab.component.html new file mode 100644 index 0000000..a0800e2 --- /dev/null +++ b/src/app/admin/admin-users-tab.component.html @@ -0,0 +1,82 @@ +
Loading...
+ } @else { +| Name | +Provider | +Actions | +|
|---|---|---|---|
| {{ user.display_name }} | +{{ user.email }} | +{{ user.auth_provider }} | ++ + | +
| No admin users yet. Use the form below to add one. | +|||