""" Authentication Domain Models SQLAlchemy models for users, roles, groups, API keys, and preferences. All authentication-related database models consolidated in one file. """ import uuid from datetime import datetime from typing import TYPE_CHECKING, List from sqlalchemy import String, Boolean, DateTime, func, ForeignKey, Table, Column from sqlalchemy.dialects.postgresql import UUID, JSONB, ARRAY from sqlalchemy.orm import Mapped, mapped_column, relationship from src.shared.database import Base # ============================================================================= # Association Tables # ============================================================================= user_groups = Table( "user_groups", Base.metadata, Column("user_id", UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), primary_key=True), Column("group_id", UUID(as_uuid=True), ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True), ) group_roles = Table( "group_roles", Base.metadata, Column("group_id", UUID(as_uuid=True), ForeignKey("groups.id", ondelete="CASCADE"), primary_key=True), Column("role_id", UUID(as_uuid=True), ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True), ) # ============================================================================= # User Model # ============================================================================= class User(Base): """ User model synced from Authentik Users are created/updated when they authenticate via OIDC. The authentik_id links to the Authentik user record. """ __tablename__ = "users" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) authentik_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), unique=True, nullable=False, index=True, ) email: Mapped[str] = mapped_column( String(255), unique=True, nullable=False, index=True, ) name: Mapped[str] = mapped_column( String(255), nullable=False, ) avatar_url: Mapped[str | None] = mapped_column( String(500), nullable=True, ) api_keys_enabled: Mapped[bool] = mapped_column( Boolean, default=True, nullable=False, ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False, ) last_login: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, ) # Relationships roles: Mapped[List["Role"]] = relationship( "Role", secondary="user_roles", back_populates="users", lazy="selectin", ) preferences: Mapped["UserPreferences"] = relationship( "UserPreferences", back_populates="user", uselist=False, lazy="selectin", cascade="all, delete-orphan", ) api_keys: Mapped[List["ApiKey"]] = relationship( "ApiKey", back_populates="user", lazy="selectin", cascade="all, delete-orphan", ) def __repr__(self) -> str: return f"" # ============================================================================= # Role Models # ============================================================================= class Role(Base): """ Role model for domain-scoped permissions Permission format: domain.category:action - domain: Main area (control-room, library, media, ai, etc.) - category: Sub-area within domain (general for full access, or specific tools) - action: Permission level (viewer, user, editor, admin) Roles are seeded from configuration, not user-editable. Groups are assigned roles via the group_roles mapping table. Domains: control-room, library, media, ai, housekeeper, developer, documents, gaming, admin Actions: viewer, user, editor, admin (hierarchical) """ __tablename__ = "roles" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) name: Mapped[str] = mapped_column( String(100), unique=True, nullable=False, index=True, comment="Role name in format domain.category:action (e.g., control-room.general:admin)", ) domain: Mapped[str] = mapped_column( String(50), nullable=False, index=True, comment="Permission domain (e.g., control-room, media, ai)", ) category: Mapped[str] = mapped_column( String(50), nullable=False, default="general", comment="Permission category within domain (general for full access, or specific tool)", ) action: Mapped[str] = mapped_column( String(20), nullable=False, comment="Permission action (viewer, user, editor, admin)", ) # Relationships users: Mapped[List["User"]] = relationship( "User", secondary="user_roles", back_populates="roles", lazy="selectin", ) groups: Mapped[List["Group"]] = relationship( "Group", secondary="group_roles", back_populates="roles", lazy="selectin", ) def __repr__(self) -> str: return f"" class UserRole(Base): """ Association table for User-Role many-to-many relationship Synced from Authentik groups during user authentication. """ __tablename__ = "user_roles" user_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), primary_key=True, ) role_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("roles.id", ondelete="CASCADE"), primary_key=True, ) # ============================================================================= # Group Model # ============================================================================= class Group(Base): """ Group model synced from Authentik Groups are fetched from Authentik admin API and cached locally. They represent organizational units for access control. """ __tablename__ = "groups" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) authentik_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), unique=True, nullable=False, index=True, comment="Authentik group UUID", ) name: Mapped[str] = mapped_column( String(255), unique=True, nullable=False, index=True, ) is_superuser: Mapped[bool] = mapped_column( Boolean, default=False, nullable=False, comment="Whether members of this group have superuser privileges", ) parent_name: Mapped[str | None] = mapped_column( String(255), nullable=True, comment="Parent group name for hierarchy", ) member_count: Mapped[int] = mapped_column( default=0, nullable=False, comment="Number of users in this group", ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False, ) synced_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False, comment="Last sync from Authentik", ) # Relationships roles: Mapped[List["Role"]] = relationship( "Role", secondary="group_roles", back_populates="groups", lazy="selectin", ) def __repr__(self) -> str: return f"" # ============================================================================= # User Preferences Model # ============================================================================= class UserPreferences(Base): """ User preferences model Stores user-specific settings that persist across sessions. Extended settings stored in preferences_json for flexibility. """ __tablename__ = "user_preferences" user_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), primary_key=True, ) theme: Mapped[str] = mapped_column( String(20), default="system", nullable=False, comment="Theme preference: system, light, dark", ) default_room: Mapped[str] = mapped_column( String(50), default="front-hall", nullable=False, comment="Default room for housekeeping features", ) preferences_json: Mapped[dict] = mapped_column( JSONB, default=dict, nullable=False, comment="Extended preferences as JSON", ) # Relationships user: Mapped["User"] = relationship( "User", back_populates="preferences", ) def __repr__(self) -> str: return f"" # ============================================================================= # API Key Model # ============================================================================= class ApiKey(Base): """ API Key model for programmatic access API keys provide an alternative to OIDC for: - Local development without SSO - Service-to-service communication - Scripts and automation Keys inherit the user's roles but can optionally be restricted to a subset of scopes. """ __tablename__ = "api_keys" id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, ) user_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True, ) name: Mapped[str] = mapped_column( String(100), nullable=False, comment="Human-readable key name (e.g., 'Dev Laptop', 'CI/CD')", ) key_hash: Mapped[str] = mapped_column( String(255), nullable=False, comment="SHA-256 hash of the API key", ) key_prefix: Mapped[str] = mapped_column( String(8), nullable=False, comment="First 8 chars of key for identification (e.g., 'cak_abc1')", ) scopes: Mapped[List[str] | None] = mapped_column( ARRAY(String), nullable=True, comment="Optional scope restriction (subset of user roles)", ) expires_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, comment="Optional expiration timestamp", ) last_used_at: Mapped[datetime | None] = mapped_column( DateTime(timezone=True), nullable=True, comment="Last time this key was used", ) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now(), nullable=False, ) # Relationships user: Mapped["User"] = relationship( "User", back_populates="api_keys", ) def __repr__(self) -> str: return f"" @property def is_expired(self) -> bool: """Check if the API key has expired""" if self.expires_at is None: return False return datetime.now(self.expires_at.tzinfo) > self.expires_at