diff --git a/backend/app/api/auth.py b/backend/app/api/auth.py index ea1c97b..1f8f80c 100644 --- a/backend/app/api/auth.py +++ b/backend/app/api/auth.py @@ -57,8 +57,18 @@ async def login(payload: LoginRequest): # This is handled in the auth router to keep it self-contained from app.database import async_session + from app.user_models import Role async with async_session() as session: + # Ensure the 'admin' role exists + result = await session.execute(select(Role).where(Role.name == "admin")) + admin_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.commit() + await session.refresh(admin_role) + # Find or create the bootstrap admin user result = await session.execute( select(User).where( @@ -74,11 +84,16 @@ async def login(payload: LoginRequest): auth_provider="local", is_admin=True, ) + user.roles.append(admin_role) session.add(user) - await session.commit() - await session.refresh(user) + else: + # Ensure existing bootstrap user also has the role + if admin_role not in user.roles: + user.roles.append(admin_role) + await session.commit() + await session.refresh(user) - token = create_access_token(user.id, user.is_admin) + token = create_access_token(user.id, user.is_admin_effective) return TokenResponse(access_token=token) @@ -102,7 +117,7 @@ async def login_with_google(payload: GoogleLoginRequest, session: AsyncSession = await session.commit() await session.refresh(user) - token = create_access_token(user.id, user.is_admin) + token = create_access_token(user.id, user.is_admin_effective) return TokenResponse(access_token=token) diff --git a/backend/app/auth.py b/backend/app/auth.py index c7e7d3a..a917903 100644 --- a/backend/app/auth.py +++ b/backend/app/auth.py @@ -98,8 +98,11 @@ async def get_current_user( async def get_current_admin_user( current_user: User = Depends(get_current_user), ) -> User: - """Raise 403 if the current user is not an admin.""" - if not current_user.is_admin: + """Raise 403 if the current user is not an admin. + + Checks both the legacy is_admin flag and the new role-based admin role. + """ + if not current_user.is_admin_effective: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin access required") return current_user diff --git a/backend/app/user_models.py b/backend/app/user_models.py index f17a9c8..a878079 100644 --- a/backend/app/user_models.py +++ b/backend/app/user_models.py @@ -1,11 +1,47 @@ +"""User, Role, and UserRole models — supports multiple admin roles.""" + from datetime import datetime -from sqlalchemy import Column, Integer, String, Boolean, DateTime -from sqlalchemy.orm import DeclarativeBase +from sqlalchemy import ( + Boolean, + Column, + DateTime, + ForeignKey, + Integer, + String, + Table, + Text, + UniqueConstraint, +) +from sqlalchemy.orm import DeclarativeBase, relationship from app.models import Base +# ── Many-to-many: users ↔ roles ────────────────────────────────────────────── + +user_roles = Table( + "user_roles", + Base.metadata, + Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), primary_key=True), + Column("role_id", Integer, ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True), +) + + +# ── Role ───────────────────────────────────────────────────────────────────── + +class Role(Base): + __tablename__ = "roles" + + id = Column(Integer, primary_key=True, autoincrement=True) + name = Column(String(50), unique=True, nullable=False) + description = Column(Text, nullable=True) + + users = relationship("User", secondary=user_roles, back_populates="roles") + + +# ── User ───────────────────────────────────────────────────────────────────── + class User(Base): __tablename__ = "users" @@ -14,5 +50,16 @@ class User(Base): display_name = Column(String(100), nullable=False) avatar_url = Column(String(500), nullable=True) auth_provider = Column(String(20), nullable=False, default="google") + password_hash = Column(String(255), nullable=True) is_admin = Column(Boolean, default=False) created_at = Column(DateTime, nullable=False, default=datetime.utcnow) + + # Role-based admin (many-to-many) + roles = relationship("Role", secondary=user_roles, back_populates="users") + + @property + def is_admin_effective(self) -> bool: + """True when is_admin=True (legacy) OR the user has an 'admin' role.""" + if self.is_admin: + return True + return any(role.name == "admin" for role in self.roles) diff --git a/backend/migrate_roles.py b/backend/migrate_roles.py new file mode 100644 index 0000000..9010bfb --- /dev/null +++ b/backend/migrate_roles.py @@ -0,0 +1,60 @@ +""" +Migration seed: create the 'admin' role and assign it to existing is_admin=True users. + +Run once after deploying the updated schema. Idempotent — safe to run multiple times. + +Usage: + KMTN_DATABASE_URL=... python migrate_roles.py + (run from the backend/ directory) +""" + +import asyncio +from sqlalchemy import select, text + +from app.database import async_session +from app.user_models import User, Role, user_roles + + +async def migrate() -> None: + """Ensure 'admin' role exists and assign to existing is_admin=True users.""" + async with async_session() as session: + # 1) Create 'admin' role if it doesn't exist + result = await session.execute(select(Role).where(Role.name == "admin")) + admin_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.commit() + await session.refresh(admin_role) + print(f" ✓ Created 'admin' role (id={admin_role.id})") + else: + print(f" ✓ 'admin' role already exists (id={admin_role.id})") + + # 2) Assign 'admin' role to all existing is_admin=True users + result = await session.execute( + select(User).where(User.is_admin == True) # noqa: E712 + ) + admin_users = result.scalars().all() + migrated = 0 + for user in admin_users: + if admin_role not in user.roles: + user.roles.append(admin_role) + migrated += 1 + if migrated: + await session.commit() + print(f" ✓ Assigned 'admin' role to {migrated} existing admin user(s)") + else: + print(" ✓ No new role assignments needed") + + # 3) Summary + result = await session.execute( + select(User).where(user_roles.c.role_id == admin_role.id) + ) + role_admins = result.scalars().all() + print(f" → {len(role_admins)} user(s) now have the 'admin' role") + + +if __name__ == "__main__": + print("Running role migration...") + asyncio.run(migrate()) + print("Done.") diff --git a/backend/tests/test_user_models.py b/backend/tests/test_user_models.py new file mode 100644 index 0000000..deee643 --- /dev/null +++ b/backend/tests/test_user_models.py @@ -0,0 +1,87 @@ +"""Tests for app.user_models — User, Role, and UserRole relationships.""" + +import pytest + +from app.user_models import User, Role, user_roles + + +class TestRoleModel: + """Verify Role model definition.""" + + def test_role_table_name(self): + assert Role.__tablename__ == "roles" + + def test_role_has_name_column(self): + col = Role.__table__.columns["name"] + assert col.type.__visit_name__ == "VARCHAR" + + def test_role_name_is_unique(self): + col = Role.__table__.columns["name"] + assert col.unique is True + + def test_role_name_not_nullable(self): + col = Role.__table__.columns["name"] + assert col.nullable is False + + +class TestUserModel: + """Verify User model still works with new role fields.""" + + def test_user_table_name(self): + assert User.__tablename__ == "users" + + def test_user_has_is_admin_column(self): + col = User.__table__.columns["is_admin"] + assert col.default.arg is False + + def test_user_has_password_hash_column(self): + col = User.__table__.columns["password_hash"] + assert col.nullable is True + + def test_user_roles_relationship(self): + assert hasattr(User, "roles") + + +class TestUserRoleAssociation: + """Verify the user_roles association table.""" + + def test_user_roles_table_name(self): + assert user_roles.name == "user_roles" + + def test_user_roles_has_user_id(self): + assert "user_id" in user_roles.columns + + def test_user_roles_has_role_id(self): + assert "role_id" in user_roles.columns + + def test_user_roles_composite_primary_key(self): + pk = [col for col in user_roles.columns if col.primary_key] + assert len(pk) == 2 + + +class TestIsAdminEffective: + """Test the is_admin_effective property.""" + + def test_legacy_is_admin_true(self): + user = User(email="a@test.com", display_name="A", is_admin=True) + assert user.is_admin_effective is True + + def test_legacy_is_admin_false_no_roles(self): + user = User(email="b@test.com", display_name="B", is_admin=False) + assert user.is_admin_effective is False + + def test_role_based_admin(self): + user = User(email="c@test.com", display_name="C", is_admin=False) + role = Role(name="admin") + user.roles.append(role) + assert user.is_admin_effective is True + + def test_non_admin_role_does_not_grant_admin(self): + user = User(email="d@test.com", display_name="D", is_admin=False) + role = Role(name="editor") + user.roles.append(role) + assert user.is_admin_effective is False + + def test_legacy_is_admin_true_overrides_empty_roles(self): + user = User(email="e@test.com", display_name="E", is_admin=True) + assert user.is_admin_effective is True