fix: /auth/users/me now supports NPM forward auth headers

Added get_current_user_or_forward_auth() combined dependency that:
- First checks for X-authentik-* headers from NPM forward auth (web)
- Falls back to JWT Bearer token validation (mobile/native)

This fixes web authentication where browsers don't send Bearer tokens
but rely on NPM's forward auth proxy to pass user info via headers.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

.gitea/workflows/build.yml

@@ -1,12 +1,25 @@
 name: Build and Push
 
 on:
-  release:
+  push:
-    types: [published]
+    tags:
+      - 'v*'
 
 jobs:
+  release:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Create Gitea Release
+        run: |
+          curl -sf -X POST \
+            -H "Authorization: token ${{ secrets.GITHUB_TOKEN }}" \
+            -H "Content-Type: application/json" \
+            -d '{"tag_name": "${{ github.ref_name }}", "name": "Release ${{ github.ref_name }}", "body": "Automated release for ${{ github.ref_name }}"}' \
+            "${{ github.server_url }}/api/v1/repos/${{ github.repository }}/releases"
+
   build:
     runs-on: ubuntu-latest
+    needs: release
     steps:
       - uses: actions/checkout@v4
 

AGENTS.md

@@ -47,7 +47,7 @@ When changes are ready for deployment:
 5. **CI/CD triggers automatically**:
    - Gitea CI builds Docker image on new tag
    - Watchtower pulls and deploys to production
-   - Verify deployment: `curl http://192.168.86.149:8000/health`
+   - Verify deployment: `curl http://192.168.86.149:8083/health`
 
 ---
 

CHANGELOG.md

@@ -5,6 +5,49 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [1.9.2] - 2026-01-04
+
+### Fixed
+
+- `/auth/users/me` endpoint now supports both NPM forward auth headers AND JWT Bearer tokens
+  - Added `get_current_user_or_forward_auth()` combined auth dependency
+  - Fixes web authentication where NPM passes `X-authentik-*` headers instead of JWT
+  - Mobile/native clients continue to use JWT Bearer tokens as before
+
+## [1.9.1] - 2026-01-03
+
+### Fixed
+
+- CORS configuration now uses explicit origins instead of `"*"`
+  - When `allow_credentials=True`, wildcard origins are rejected by browsers
+  - Added `home.schweitz.net`, `tatlock.schweitz.net`, and localhost origins
+
+## [1.9.0] - 2026-01-03
+
+### Added
+
+- **NPM Forward Auth Support** - Web authentication via Nginx Proxy Manager forward auth
+  - `GET /auth/me` - Get current user from NPM forward auth headers (X-authentik-uid, X-authentik-email, etc.)
+  - Auto-creates user on first web login if not in database
+  - Syncs roles from NPM forward auth groups header
+- `get_user_by_email` and `get_user_by_authentik_id` methods in AuthService
+- Comprehensive tests for `/auth/me` endpoint
+
+## [1.8.0] - 2026-01-03
+
+### Added
+
+- **Group-Role Mapping & Permissions** (Phase 3)
+  - Decoupled group-role architecture (groups from Authentik, roles admin-managed)
+  - Permission format: `domain.category:action` with action hierarchy
+  - `require_permission` and `require_any_permission` dependency factories
+  - Global admin override (`admin.general:admin`)
+- **User Profile & API Keys** (Phase 4)
+  - `GET /auth/users/me` - Full user profile with roles and preferences
+  - `GET/PATCH /auth/users/me/preferences` - User preferences management
+  - `GET/POST/DELETE /auth/users/me/api-keys` - API key lifecycle
+  - API keys with `tak_` prefix, SHA-256 hashing, shown only once on creation
+
 ## [1.7.0] - 2026-01-03
 
 ### Added

alembic/versions/20260103_1500_004_add_group_roles_mapping.py

@@ -0,0 +1,141 @@
+"""Add group_roles mapping and update role schema
+
+Revision ID: 004
+Revises: f0349c95aa5d
+Create Date: 2026-01-03
+
+Changes:
+- Add category column to roles (default 'general')
+- Drop authentik_group column from roles (decoupled architecture)
+- Create user_groups association table
+- Create group_roles association table
+- Update role names from domain:action to domain.general:action
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "004"
+down_revision: Union[str, None] = "f0349c95aa5d"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # Add category column to roles
+    op.add_column(
+        "roles",
+        sa.Column(
+            "category",
+            sa.String(50),
+            nullable=False,
+            server_default="general",
+            comment="Permission category within domain (general for full access, or specific tool)",
+        ),
+    )
+
+    # Update role names from domain:action to domain.general:action
+    op.execute(
+        """
+        UPDATE roles
+        SET name = REPLACE(name, ':', '.general:')
+        WHERE name NOT LIKE '%.%:%'
+        """
+    )
+
+    # Update the comment on the name column
+    op.alter_column(
+        "roles",
+        "name",
+        comment="Role name in format domain.category:action (e.g., control-room.general:admin)",
+    )
+
+    # Drop the authentik_group unique index first
+    op.drop_index("ix_roles_authentik_group", table_name="roles")
+
+    # Drop authentik_group column (no longer needed with group_roles mapping)
+    op.drop_column("roles", "authentik_group")
+
+    # Create user_groups association table
+    op.create_table(
+        "user_groups",
+        sa.Column(
+            "user_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("users.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column(
+            "group_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("groups.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+    )
+
+    # Create group_roles association table
+    op.create_table(
+        "group_roles",
+        sa.Column(
+            "group_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("groups.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        sa.Column(
+            "role_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("roles.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+    )
+
+
+def downgrade() -> None:
+    # Drop association tables
+    op.drop_table("group_roles")
+    op.drop_table("user_groups")
+
+    # Add back authentik_group column
+    op.add_column(
+        "roles",
+        sa.Column(
+            "authentik_group",
+            sa.String(255),
+            nullable=True,
+            comment="Corresponding Authentik group name",
+        ),
+    )
+
+    # Restore authentik_group values from role names
+    op.execute(
+        """
+        UPDATE roles
+        SET authentik_group = 'tatlock-' || REPLACE(REPLACE(name, '.general:', '-'), ':', '-')
+        """
+    )
+
+    # Recreate the unique index
+    op.create_index("ix_roles_authentik_group", "roles", ["authentik_group"], unique=True)
+
+    # Revert role names from domain.general:action to domain:action
+    op.execute(
+        """
+        UPDATE roles
+        SET name = REPLACE(name, '.general:', ':')
+        WHERE name LIKE '%.general:%'
+        """
+    )
+
+    # Update the comment on the name column
+    op.alter_column(
+        "roles",
+        "name",
+        comment="Role name in format domain:action",
+    )
+
+    # Drop category column
+    op.drop_column("roles", "category")

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "core-api"
-version = "1.7.0"
+version = "1.9.2"
 description = "Core Code API - Infrastructure management and tools API"
 readme = "README.md"
 requires-python = ">=3.12"

src/auth/controller.py

@@ -13,6 +13,7 @@ from src.logging_config import get_logger
 from src.db import get_async_session
 from src.auth.schemas import AuthSyncRequest, AuthSyncResponse, UsersListResponse, BulkSyncResultSchema, GroupsListResponse
 from src.auth.service import AuthService
+from src.auth.oidc import get_forward_auth_user
 
 logger = get_logger(__name__)
 
@@ -222,21 +223,83 @@ class AuthController(BaseController):
             responses={
                 200: {"description": "User profile"},
                 401: {"description": "Not authenticated"},
+                404: {"description": "User not found in database"},
             },
         )
         async def get_me(
+            forward_auth_user: Optional[dict] = Depends(get_forward_auth_user),
             session: AsyncSession = Depends(get_async_session),
-        ) -> JSONResponse:
+        ) -> AuthSyncResponse:
             """
             Get the current authenticated user's profile
 
-            Note: This endpoint requires a valid session or API key.
+            Authentication is handled by NPM forward auth with Authentik.
-            For now, returns 501 Not Implemented until session management is added.
+            The proxy sets X-authentik-* headers which this endpoint reads.
+
+            For internal/LAN access (no forward auth headers), returns 401.
+            Use POST /auth/sync with an OIDC token for mobile app authentication.
             """
-            # TODO: Implement with get_current_user dependency
+            # Require forward auth for this endpoint
+            if forward_auth_user is None:
                 raise HTTPException(
-                status_code=501,
+                    status_code=401,
-                detail="Not implemented - use /auth/sync with access token",
+                    detail="Authentication required - access via authenticated proxy or use /auth/sync",
+                )
+
+            service = AuthService(session)
+
+            # Try to find user by Authentik UID first, then by email
+            user = None
+            uid = forward_auth_user.get("uid")
+            if uid:
+                try:
+                    import uuid
+                    authentik_id = uuid.UUID(uid)
+                    user = await service.get_user_by_authentik_id(authentik_id)
+                except (ValueError, TypeError):
+                    pass  # Invalid UUID, try email
+
+            if user is None:
+                email = forward_auth_user.get("email")
+                if email:
+                    user = await service.get_user_by_email(email)
+
+            if user is None:
+                # User authenticated with Authentik but not synced to database yet
+                # This can happen on first login via web
+                logger.info(f"User {forward_auth_user.get('email')} not found, creating from forward auth")
+
+                # Create user from forward auth headers
+                from src.auth.schemas import TokenInfoSchema
+                token_info = TokenInfoSchema(
+                    sub=forward_auth_user.get("uid", ""),
+                    email=forward_auth_user.get("email", ""),
+                    name=forward_auth_user.get("name"),
+                    groups=forward_auth_user.get("groups", []),
+                )
+
+                try:
+                    user, _ = await service.sync_user(token_info)
+                    await service.sync_roles(user, token_info.groups)
+                    await session.commit()
+                    await session.refresh(user, ["preferences", "roles"])
+                except Exception as e:
+                    logger.error(f"Failed to create user from forward auth: {e}")
+                    raise HTTPException(
+                        status_code=500,
+                        detail="Failed to create user profile",
+                    )
+
+            # Sync roles from current groups (in case they changed)
+            groups = forward_auth_user.get("groups", [])
+            roles = await service.sync_roles(user, groups)
+            await session.commit()
+
+            return AuthSyncResponse(
+                user=service.user_to_schema(user),
+                roles=service.roles_to_schema(roles),
+                preferences=service.preferences_to_schema(user.preferences),
+                is_new_user=False,
             )
 
         return router

src/auth/service.py

@@ -86,6 +86,42 @@ class AuthService:
             logger.error(f"Authentik userinfo request error: {e}")
             raise ValueError("Authentication service unavailable")
 
+    async def get_user_by_email(self, email: str) -> Optional[User]:
+        """
+        Get user by email address
+
+        Args:
+            email: User email address
+
+        Returns:
+            User if found, None otherwise
+        """
+        stmt = (
+            select(User)
+            .options(selectinload(User.roles), selectinload(User.preferences))
+            .where(User.email == email)
+        )
+        result = await self.session.execute(stmt)
+        return result.scalar_one_or_none()
+
+    async def get_user_by_authentik_id(self, authentik_id: uuid.UUID) -> Optional[User]:
+        """
+        Get user by Authentik UUID
+
+        Args:
+            authentik_id: Authentik user UUID
+
+        Returns:
+            User if found, None otherwise
+        """
+        stmt = (
+            select(User)
+            .options(selectinload(User.roles), selectinload(User.preferences))
+            .where(User.authentik_id == authentik_id)
+        )
+        result = await self.session.execute(stmt)
+        return result.scalar_one_or_none()
+
     async def sync_user(self, token_info: TokenInfoSchema) -> tuple[User, bool]:
         """
         Create or update user from OIDC token info

src/domains/auth/__init__.py

@@ -11,6 +11,12 @@ from src.domains.auth.oidc import (
     get_forward_auth_user,
     get_forward_auth_admin,
     oidc_config,
+    # Permission system
+    require_permission,
+    require_any_permission,
+    ACTION_HIERARCHY,
+    VALID_DOMAINS,
+    DEFAULT_CATEGORY,
 )
 from src.domains.auth.service import AuthService, get_auth_service
 from src.domains.auth.controller import auth_controller
@@ -22,6 +28,7 @@ from src.domains.auth.models import (
     UserPreferences,
     ApiKey,
     user_groups,
+    group_roles,
 )
 
 __all__ = [
@@ -32,6 +39,12 @@ __all__ = [
     "get_forward_auth_user",
     "get_forward_auth_admin",
     "oidc_config",
+    # Permission system
+    "require_permission",
+    "require_any_permission",
+    "ACTION_HIERARCHY",
+    "VALID_DOMAINS",
+    "DEFAULT_CATEGORY",
     # Service
     "AuthService",
     "get_auth_service",
@@ -45,4 +58,5 @@ __all__ = [
     "UserPreferences",
     "ApiKey",
     "user_groups",
+    "group_roles",
 ]

src/domains/auth/controller.py

@@ -3,8 +3,9 @@ Authentication Controller
 
 Provides authentication endpoints for OIDC token sync and user management.
 """
+import uuid
 from typing import Optional
-from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi import APIRouter, Depends, HTTPException, Path, Query
 from fastapi.responses import JSONResponse
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -13,9 +14,13 @@ from src.shared.logging import get_logger
 from src.shared.database import get_async_session
 from src.domains.auth.schemas import (
     AuthSyncRequest, AuthSyncResponse, UsersListResponse,
-    BulkSyncResultSchema, GroupsListResponse
+    BulkSyncResultSchema, GroupsListResponse, RolesListResponse,
+    GroupRoleAssignmentResponse, UserProfileResponse, PreferencesUpdateRequest,
+    UserPreferencesSchema, ApiKeyCreateRequest, ApiKeyCreateResponse,
+    ApiKeysListResponse,
 )
 from src.domains.auth.service import AuthService
+from src.domains.auth.oidc import get_current_user, get_current_user_or_forward_auth
 
 logger = get_logger(__name__)
 
@@ -219,29 +224,367 @@ class AuthController(BaseController):
                 raise HTTPException(status_code=401, detail=str(e))
 
         @router.get(
-            "/me",
+            "/roles",
-            summary="Get current user profile",
+            summary="List all roles",
-            response_model=AuthSyncResponse,
+            response_model=RolesListResponse,
             responses={
-                200: {"description": "User profile"},
+                200: {"description": "List of all available roles"},
-                401: {"description": "Not authenticated"},
             },
         )
-        async def get_me(
+        async def list_roles(
             session: AsyncSession = Depends(get_async_session),
-        ) -> JSONResponse:
+        ) -> RolesListResponse:
+            """
+            List all available roles in the system
+
+            Returns all domain.category:action role combinations.
+            Use these when assigning roles to groups.
+            """
+            service = AuthService(session)
+            roles = await service.list_roles()
+            return RolesListResponse(
+                items=service.roles_to_schema(roles),
+                total=len(roles),
+            )
+
+        @router.post(
+            "/groups/{group_id}/roles/{role_id}",
+            summary="Assign role to group",
+            response_model=GroupRoleAssignmentResponse,
+            responses={
+                200: {"description": "Role assigned successfully"},
+                404: {"description": "Group or role not found"},
+            },
+        )
+        async def assign_role_to_group(
+            group_id: uuid.UUID = Path(..., description="Group ID"),
+            role_id: uuid.UUID = Path(..., description="Role ID to assign"),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> GroupRoleAssignmentResponse:
+            """
+            Assign a role to a group
+
+            All users in this group will inherit this role's permissions.
+            """
+            service = AuthService(session)
+
+            try:
+                group = await service.assign_role_to_group(group_id, role_id)
+                await session.commit()
+                return GroupRoleAssignmentResponse(
+                    group_id=group.id,
+                    group_name=group.name,
+                    roles=[role.name for role in group.roles],
+                )
+            except ValueError as e:
+                raise HTTPException(status_code=404, detail=str(e))
+
+        @router.delete(
+            "/groups/{group_id}/roles/{role_id}",
+            summary="Remove role from group",
+            response_model=GroupRoleAssignmentResponse,
+            responses={
+                200: {"description": "Role removed successfully"},
+                404: {"description": "Group or role not found"},
+            },
+        )
+        async def remove_role_from_group(
+            group_id: uuid.UUID = Path(..., description="Group ID"),
+            role_id: uuid.UUID = Path(..., description="Role ID to remove"),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> GroupRoleAssignmentResponse:
+            """
+            Remove a role from a group
+
+            Users in this group will no longer inherit this role's permissions.
+            """
+            service = AuthService(session)
+
+            try:
+                group = await service.remove_role_from_group(group_id, role_id)
+                await session.commit()
+                return GroupRoleAssignmentResponse(
+                    group_id=group.id,
+                    group_name=group.name,
+                    roles=[role.name for role in group.roles],
+                )
+            except ValueError as e:
+                raise HTTPException(status_code=404, detail=str(e))
+
+        # =====================================================================
+        # Phase 4: User Profile & Settings
+        # =====================================================================
+
+        @router.get(
+            "/users/me",
+            summary="Get current user profile",
+            response_model=UserProfileResponse,
+            responses={
+                200: {"description": "User profile with roles and preferences"},
+                401: {"description": "Not authenticated"},
+                404: {"description": "User not found in database"},
+            },
+        )
+        async def get_current_user_profile(
+            user_claims: dict = Depends(get_current_user_or_forward_auth),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> UserProfileResponse:
             """
             Get the current authenticated user's profile
 
-            Note: This endpoint requires a valid session or API key.
+            Returns the user's profile, roles, and preferences.
-            For now, returns 501 Not Implemented until session management is added.
+            Requires authentication via Bearer token or API key.
             """
-            # TODO: Implement with get_current_user dependency
+            service = AuthService(session)
+
+            # Get authentik_id from claims (JWT 'sub' field)
+            authentik_id_str = user_claims.get("sub")
+            if not authentik_id_str or authentik_id_str == "local-user":
+                raise HTTPException(status_code=401, detail="Authentication required")
+
+            try:
+                authentik_id = uuid.UUID(authentik_id_str)
+            except ValueError:
+                raise HTTPException(status_code=401, detail="Invalid user identifier")
+
+            user = await service.get_user_by_authentik_id(authentik_id)
+            if user is None:
                 raise HTTPException(
-                status_code=501,
+                    status_code=404,
-                detail="Not implemented - use /auth/sync with access token",
+                    detail="User not found - please sync via /auth/sync first",
                 )
 
+            return UserProfileResponse(
+                user=service.user_to_schema(user),
+                roles=service.roles_to_schema(user.roles),
+                preferences=service.preferences_to_schema(user.preferences),
+            )
+
+        @router.get(
+            "/users/me/preferences",
+            summary="Get user preferences",
+            response_model=UserPreferencesSchema,
+            responses={
+                200: {"description": "User preferences"},
+                401: {"description": "Not authenticated"},
+            },
+        )
+        async def get_preferences(
+            user_claims: dict = Depends(get_current_user),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> UserPreferencesSchema:
+            """
+            Get the current user's preferences
+            """
+            service = AuthService(session)
+
+            authentik_id_str = user_claims.get("sub")
+            if not authentik_id_str or authentik_id_str == "local-user":
+                raise HTTPException(status_code=401, detail="Authentication required")
+
+            try:
+                authentik_id = uuid.UUID(authentik_id_str)
+            except ValueError:
+                raise HTTPException(status_code=401, detail="Invalid user identifier")
+
+            user = await service.get_user_by_authentik_id(authentik_id)
+            if user is None:
+                raise HTTPException(status_code=404, detail="User not found")
+
+            return service.preferences_to_schema(user.preferences)
+
+        @router.patch(
+            "/users/me/preferences",
+            summary="Update user preferences",
+            response_model=UserPreferencesSchema,
+            responses={
+                200: {"description": "Updated preferences"},
+                401: {"description": "Not authenticated"},
+                422: {"description": "Invalid preference value"},
+            },
+        )
+        async def update_preferences(
+            request: PreferencesUpdateRequest,
+            user_claims: dict = Depends(get_current_user),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> UserPreferencesSchema:
+            """
+            Update the current user's preferences
+
+            Only provided fields are updated. preferences_json is merged
+            with existing values (not replaced).
+            """
+            service = AuthService(session)
+
+            authentik_id_str = user_claims.get("sub")
+            if not authentik_id_str or authentik_id_str == "local-user":
+                raise HTTPException(status_code=401, detail="Authentication required")
+
+            try:
+                authentik_id = uuid.UUID(authentik_id_str)
+            except ValueError:
+                raise HTTPException(status_code=401, detail="Invalid user identifier")
+
+            user = await service.get_user_by_authentik_id(authentik_id)
+            if user is None:
+                raise HTTPException(status_code=404, detail="User not found")
+
+            try:
+                prefs = await service.update_preferences(
+                    user_id=user.id,
+                    theme=request.theme,
+                    default_room=request.default_room,
+                    preferences_json=request.preferences_json,
+                )
+                await session.commit()
+                return service.preferences_to_schema(prefs)
+            except ValueError as e:
+                raise HTTPException(status_code=422, detail=str(e))
+
+        # =====================================================================
+        # Phase 4: API Keys
+        # =====================================================================
+
+        @router.get(
+            "/users/me/api-keys",
+            summary="List user's API keys",
+            response_model=ApiKeysListResponse,
+            responses={
+                200: {"description": "List of API keys"},
+                401: {"description": "Not authenticated"},
+            },
+        )
+        async def list_api_keys(
+            user_claims: dict = Depends(get_current_user),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> ApiKeysListResponse:
+            """
+            List all API keys for the current user
+
+            Returns key metadata only - the actual key values are never
+            retrievable after creation.
+            """
+            service = AuthService(session)
+
+            authentik_id_str = user_claims.get("sub")
+            if not authentik_id_str or authentik_id_str == "local-user":
+                raise HTTPException(status_code=401, detail="Authentication required")
+
+            try:
+                authentik_id = uuid.UUID(authentik_id_str)
+            except ValueError:
+                raise HTTPException(status_code=401, detail="Invalid user identifier")
+
+            user = await service.get_user_by_authentik_id(authentik_id)
+            if user is None:
+                raise HTTPException(status_code=404, detail="User not found")
+
+            keys = await service.list_user_api_keys(user.id)
+            return ApiKeysListResponse(
+                items=[service.api_key_to_schema(k) for k in keys],
+                total=len(keys),
+            )
+
+        @router.post(
+            "/users/me/api-keys",
+            summary="Create a new API key",
+            response_model=ApiKeyCreateResponse,
+            responses={
+                201: {"description": "API key created"},
+                401: {"description": "Not authenticated"},
+                403: {"description": "API keys disabled for user"},
+            },
+        )
+        async def create_api_key(
+            request: ApiKeyCreateRequest,
+            user_claims: dict = Depends(get_current_user),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> ApiKeyCreateResponse:
+            """
+            Create a new API key for the current user
+
+            **IMPORTANT**: The full API key is only returned once in this response!
+            Store it securely - it cannot be retrieved again.
+            """
+            service = AuthService(session)
+
+            authentik_id_str = user_claims.get("sub")
+            if not authentik_id_str or authentik_id_str == "local-user":
+                raise HTTPException(status_code=401, detail="Authentication required")
+
+            try:
+                authentik_id = uuid.UUID(authentik_id_str)
+            except ValueError:
+                raise HTTPException(status_code=401, detail="Invalid user identifier")
+
+            user = await service.get_user_by_authentik_id(authentik_id)
+            if user is None:
+                raise HTTPException(status_code=404, detail="User not found")
+
+            try:
+                api_key, full_key = await service.create_api_key(
+                    user_id=user.id,
+                    name=request.name,
+                    scopes=request.scopes,
+                    expires_in_days=request.expires_in_days,
+                )
+                await session.commit()
+
+                return ApiKeyCreateResponse(
+                    id=api_key.id,
+                    name=api_key.name,
+                    key=full_key,  # Only time this is returned!
+                    key_prefix=api_key.key_prefix,
+                    scopes=api_key.scopes,
+                    expires_at=api_key.expires_at,
+                    created_at=api_key.created_at,
+                )
+            except ValueError as e:
+                raise HTTPException(status_code=403, detail=str(e))
+
+        @router.delete(
+            "/users/me/api-keys/{key_id}",
+            summary="Delete an API key",
+            responses={
+                204: {"description": "API key deleted"},
+                401: {"description": "Not authenticated"},
+                404: {"description": "API key not found"},
+            },
+        )
+        async def delete_api_key(
+            key_id: uuid.UUID = Path(..., description="API key ID to delete"),
+            user_claims: dict = Depends(get_current_user),
+            session: AsyncSession = Depends(get_async_session),
+        ) -> JSONResponse:
+            """
+            Delete an API key
+
+            The key will be immediately invalidated.
+            """
+            service = AuthService(session)
+
+            authentik_id_str = user_claims.get("sub")
+            if not authentik_id_str or authentik_id_str == "local-user":
+                raise HTTPException(status_code=401, detail="Authentication required")
+
+            try:
+                authentik_id = uuid.UUID(authentik_id_str)
+            except ValueError:
+                raise HTTPException(status_code=401, detail="Invalid user identifier")
+
+            user = await service.get_user_by_authentik_id(authentik_id)
+            if user is None:
+                raise HTTPException(status_code=404, detail="User not found")
+
+            try:
+                deleted = await service.delete_api_key(user.id, key_id)
+                if not deleted:
+                    raise HTTPException(status_code=404, detail="API key not found")
+                await session.commit()
+                return JSONResponse(status_code=204, content=None)
+            except ValueError as e:
+                raise HTTPException(status_code=403, detail=str(e))
+
         return router
 
 

src/domains/auth/models.py

@@ -26,6 +26,13 @@ user_groups = Table(
     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
@@ -114,8 +121,13 @@ 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.
-    Each role maps to an Authentik group (e.g., tatlock-control-room-admin).
+    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)
@@ -133,7 +145,7 @@ class Role(Base):
         unique=True,
         nullable=False,
         index=True,
-        comment="Role name in format domain:action (e.g., control-room:admin)",
+        comment="Role name in format domain.category:action (e.g., control-room.general:admin)",
     )
     domain: Mapped[str] = mapped_column(
         String(50),
@@ -141,17 +153,17 @@ class Role(Base):
         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)",
     )
-    authentik_group: Mapped[str | None] = mapped_column(
-        String(255),
-        nullable=True,
-        unique=True,
-        comment="Corresponding Authentik group name (e.g., tatlock-control-room-admin)",
-    )
 
     # Relationships
     users: Mapped[List["User"]] = relationship(
@@ -160,6 +172,12 @@ class Role(Base):
         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"<Role {self.name}>"
@@ -247,6 +265,14 @@ class Group(Base):
         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"<Group {self.name}>"
 

src/domains/auth/oidc.py

@@ -3,15 +3,57 @@ OIDC Authentication Module
 
 Provides OAuth2/OIDC token validation for FastAPI using Authentik as IdP.
 Implements bearer token authentication with JWT verification.
+
+Permission Format: domain.category:action
+- domain: Main area (control-room, library, media, ai, etc.)
+- category: Sub-area within domain (general for full domain, or specific tools)
+- action: Permission level (viewer, user, editor, admin)
+
+Examples:
+- control-room.general:admin - Full access to Control Room
+- media.general:viewer - View-only access to Media area
+- ai.ollama:user - User-level access to Ollama specifically (future)
+
+Action Hierarchy (higher implies lower):
+- admin > editor > user > viewer
 """
 from fastapi import Depends, HTTPException, Security, Request
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from jose import jwt, JWTError
 import httpx
 from functools import lru_cache
-from typing import Dict, Optional
+from typing import Callable, Dict, List, Optional
 from src.shared.logging import get_logger
 
+
+# =============================================================================
+# Permission System
+# =============================================================================
+
+# Action hierarchy: higher actions imply lower ones
+ACTION_HIERARCHY: Dict[str, int] = {
+    "viewer": 1,
+    "user": 2,
+    "editor": 3,
+    "admin": 4,
+}
+
+# Valid domains (main areas)
+VALID_DOMAINS = {
+    "control-room",
+    "library",
+    "media",
+    "ai",
+    "housekeeper",
+    "developer",
+    "documents",
+    "gaming",
+    "admin",  # Global admin domain
+}
+
+# Default category for general domain access
+DEFAULT_CATEGORY = "general"
+
 logger = get_logger(__name__)
 security = HTTPBearer(auto_error=False)
 
@@ -352,3 +394,345 @@ async def get_forward_auth_admin(
         )
 
     return user
+
+
+async def get_current_user_or_forward_auth(
+    request: Request,
+    credentials: Optional[HTTPAuthorizationCredentials] = Security(security)
+) -> Dict:
+    """
+    Combined auth: Try forward auth headers first, then JWT Bearer token.
+
+    Supports both:
+    - Web clients via NPM forward auth (X-authentik-* headers from proxy)
+    - Mobile/native clients via OIDC JWT Bearer tokens
+
+    This is the preferred dependency for /auth/users/me and similar endpoints
+    that need to work with both web (cookie-based via NPM) and mobile (token-based).
+
+    Args:
+        request: FastAPI request object containing headers
+        credentials: HTTP Bearer token from Authorization header
+
+    Returns:
+        User claims dictionary with at minimum: sub, email, name, groups, auth_method
+
+    Raises:
+        HTTPException 401: If neither forward auth headers nor valid JWT provided
+    """
+    # 1. Try forward auth headers first (web via NPM)
+    username = request.headers.get("x-authentik-username")
+    email = request.headers.get("x-authentik-email")
+
+    if username and email:
+        # Forward auth headers present - use them
+        groups = request.headers.get("x-authentik-groups", "")
+        name = request.headers.get("x-authentik-name", username)
+        uid = request.headers.get("x-authentik-uid")
+
+        user_info = {
+            "sub": uid,  # Use authentik UID as subject (for user lookup)
+            "email": email,
+            "preferred_username": username,
+            "name": name,
+            "groups": [g.strip() for g in groups.split(",")] if groups else [],
+            "auth_method": "forward_auth"
+        }
+
+        logger.info(f"Authenticated via forward auth: {email}")
+        return user_info
+
+    # 2. Fall back to JWT Bearer token (mobile/native)
+    return await get_current_user(credentials)
+
+
+# =============================================================================
+# Permission-Based Access Control
+# =============================================================================
+
+def _parse_permission(permission: str) -> tuple[str, str, str]:
+    """
+    Parse a permission string into (domain, category, action)
+
+    Supports formats:
+    - domain.category:action (full): "control-room.general:admin"
+    - domain:action (shorthand): "control-room:admin" -> ("control-room", "general", "admin")
+
+    Returns:
+        Tuple of (domain, category, action)
+
+    Raises:
+        ValueError: If permission format is invalid
+    """
+    # Split on colon first to get action
+    if ":" not in permission:
+        raise ValueError(f"Invalid permission format (missing ':'): {permission}")
+
+    location, action = permission.rsplit(":", 1)
+
+    # Split location on dot to get domain and category
+    if "." in location:
+        domain, category = location.split(".", 1)
+    else:
+        # Shorthand: domain:action -> domain.general:action
+        domain = location
+        category = DEFAULT_CATEGORY
+
+    return domain, category, action
+
+
+def _action_satisfies(user_action: str, required_action: str) -> bool:
+    """
+    Check if user's action level satisfies the required action
+
+    Due to hierarchy, admin satisfies editor, editor satisfies user, etc.
+
+    Args:
+        user_action: The action the user has
+        required_action: The action required for access
+
+    Returns:
+        True if user's action is >= required action
+    """
+    user_level = ACTION_HIERARCHY.get(user_action, 0)
+    required_level = ACTION_HIERARCHY.get(required_action, 0)
+    return user_level >= required_level
+
+
+def _user_has_permission(
+    user_permissions: List[str],
+    required_domain: str,
+    required_category: str,
+    required_action: str,
+) -> bool:
+    """
+    Check if user has a permission that satisfies the requirement
+
+    Checks:
+    1. Exact match: domain.category:action
+    2. Domain-wide: domain.general:action (if category != general)
+    3. Global admin: admin.general:admin (superuser)
+
+    Args:
+        user_permissions: List of user's permission strings
+        required_domain: Required domain
+        required_category: Required category
+        required_action: Required action
+
+    Returns:
+        True if user has sufficient permission
+    """
+    for perm in user_permissions:
+        try:
+            dom, cat, act = _parse_permission(perm)
+        except ValueError:
+            continue
+
+        # Global admin (admin.general:admin) grants all permissions
+        if dom == "admin" and cat == "general" and act == "admin":
+            return True
+
+        # Check if this permission covers the requirement
+        if dom == required_domain:
+            # Exact category match
+            if cat == required_category and _action_satisfies(act, required_action):
+                return True
+            # Domain-wide permission (general category) covers all categories in domain
+            if cat == "general" and _action_satisfies(act, required_action):
+                return True
+
+    return False
+
+
+def _extract_permissions_from_groups(groups: List[str]) -> List[str]:
+    """
+    Extract permission strings from Authentik group names
+
+    Authentik groups follow naming: tatlock-{domain}-{category}-{action}
+    or shorthand: tatlock-{domain}-{action} (implies category=general)
+
+    Examples:
+    - tatlock-control-room-general-admin -> control-room.general:admin
+    - tatlock-media-viewer -> media.general:viewer (shorthand)
+    - tatlock-ai-ollama-user -> ai.ollama:user
+
+    Args:
+        groups: List of Authentik group names
+
+    Returns:
+        List of permission strings
+    """
+    permissions = []
+
+    for group in groups:
+        if not group.startswith("tatlock-"):
+            continue
+
+        # Remove prefix
+        parts = group[8:].split("-")  # Remove "tatlock-"
+
+        if len(parts) >= 3:
+            # Could be domain-category-action or domain-with-hyphen-action
+            # Try to find a valid action at the end
+            action = parts[-1]
+            if action in ACTION_HIERARCHY:
+                # Check if domain-category or single domain with hyphen
+                remaining = parts[:-1]
+
+                # Try to find known domain (greedy match from start)
+                for i in range(len(remaining), 0, -1):
+                    potential_domain = "-".join(remaining[:i])
+                    if potential_domain in VALID_DOMAINS:
+                        category_parts = remaining[i:]
+                        category = "-".join(category_parts) if category_parts else DEFAULT_CATEGORY
+                        permissions.append(f"{potential_domain}.{category}:{action}")
+                        break
+        elif len(parts) == 2:
+            # Shorthand: domain-action (domain might have hyphen)
+            action = parts[-1]
+            if action in ACTION_HIERARCHY:
+                domain = parts[0]
+                if domain in VALID_DOMAINS:
+                    permissions.append(f"{domain}.{DEFAULT_CATEGORY}:{action}")
+
+    return permissions
+
+
+def require_permission(
+    domain: str,
+    action: str,
+    category: str = DEFAULT_CATEGORY,
+) -> Callable:
+    """
+    Dependency factory for permission-based access control
+
+    Creates a FastAPI dependency that checks if the current user has
+    the required permission. Considers action hierarchy and global admin.
+
+    Usage:
+        @router.get("/containers")
+        async def list_containers(
+            user: Dict = Depends(require_permission("control-room", "viewer"))
+        ):
+            ...
+
+        @router.delete("/container/{id}")
+        async def delete_container(
+            user: Dict = Depends(require_permission("control-room", "admin"))
+        ):
+            ...
+
+    Args:
+        domain: Permission domain (e.g., "control-room", "media")
+        action: Required action level (viewer, user, editor, admin)
+        category: Permission category within domain, defaults to "general"
+
+    Returns:
+        FastAPI dependency function
+    """
+    perm_str = f"{domain}.{category}:{action}"
+
+    async def permission_checker(
+        user: Optional[Dict] = Depends(get_current_user)
+    ) -> Dict:
+        """Check if user has required permission"""
+
+        # If OIDC disabled, allow all (local dev mode)
+        if not oidc_config.enabled:
+            logger.debug(f"OIDC disabled - allowing {perm_str}")
+            return user or {"email": "local", "groups": ["admin"]}
+
+        if user is None:
+            raise HTTPException(
+                status_code=401,
+                detail="Authentication required",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        # Extract permissions from user's groups
+        groups = user.get("groups", [])
+        permissions = _extract_permissions_from_groups(groups)
+
+        # Check if user has required permission
+        if _user_has_permission(permissions, domain, category, action):
+            logger.debug(f"User {user.get('email')} granted {perm_str}")
+            return user
+
+        # Permission denied
+        user_email = user.get("email", "unknown")
+        logger.warning(
+            f"User {user_email} denied {perm_str} "
+            f"(groups: {groups}, permissions: {permissions})"
+        )
+        raise HTTPException(
+            status_code=403,
+            detail=f"Permission required: {perm_str}",
+        )
+
+    return permission_checker
+
+
+def require_any_permission(*required_permissions: str) -> Callable:
+    """
+    Dependency factory requiring any one of multiple permissions
+
+    Useful for endpoints accessible to multiple roles.
+
+    Usage:
+        @router.get("/shared-resource")
+        async def get_shared(
+            user: Dict = Depends(require_any_permission(
+                "control-room:viewer",
+                "media:viewer",
+            ))
+        ):
+            ...
+
+    Args:
+        *required_permissions: Permission strings (domain.category:action or domain:action)
+
+    Returns:
+        FastAPI dependency function
+    """
+
+    async def permission_checker(
+        user: Optional[Dict] = Depends(get_current_user)
+    ) -> Dict:
+        """Check if user has any of the required permissions"""
+
+        # If OIDC disabled, allow all
+        if not oidc_config.enabled:
+            return user or {"email": "local", "groups": ["admin"]}
+
+        if user is None:
+            raise HTTPException(
+                status_code=401,
+                detail="Authentication required",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        groups = user.get("groups", [])
+        permissions = _extract_permissions_from_groups(groups)
+
+        # Check each required permission
+        for perm in required_permissions:
+            try:
+                dom, cat, act = _parse_permission(perm)
+                if _user_has_permission(permissions, dom, cat, act):
+                    logger.debug(f"User {user.get('email')} granted via {perm}")
+                    return user
+            except ValueError:
+                logger.warning(f"Invalid permission format: {perm}")
+                continue
+
+        # None matched
+        user_email = user.get("email", "unknown")
+        logger.warning(
+            f"User {user_email} denied (required any of: {required_permissions})"
+        )
+        raise HTTPException(
+            status_code=403,
+            detail=f"One of these permissions required: {', '.join(required_permissions)}",
+        )
+
+    return permission_checker

src/domains/auth/schemas.py

@@ -26,10 +26,12 @@ class AuthSyncRequest(BaseSchema):
 
 
 class RoleSchema(BaseSchema):
-    """Role information in domain:action format"""
+    """Role information in domain.category:action format"""
 
-    name: str = Field(..., description="Role name (e.g., 'control-room:admin')")
+    id: uuid.UUID = Field(..., description="Role ID")
+    name: str = Field(..., description="Role name (e.g., 'control-room.general:admin')")
     domain: str = Field(..., description="Permission domain (e.g., 'control-room')")
+    category: str = Field(default="general", description="Permission category (e.g., 'general')")
     action: str = Field(..., description="Permission action (e.g., 'admin')")
 
 
@@ -120,6 +122,7 @@ class GroupListItemSchema(BaseSchema):
     parent_name: Optional[str] = Field(None, description="Parent group name")
     member_count: int = Field(default=0, description="Number of users in this group")
     synced_at: datetime = Field(..., description="Last sync timestamp")
+    roles: list[str] = Field(default_factory=list, description="Assigned role names")
 
 
 class GroupsListResponse(BaseSchema):
@@ -127,3 +130,82 @@ class GroupsListResponse(BaseSchema):
 
     items: list[GroupListItemSchema] = Field(..., description="List of groups")
     total: int = Field(..., description="Total count of groups")
+
+
+class RolesListResponse(BaseSchema):
+    """Response from GET /auth/roles"""
+
+    items: list[RoleSchema] = Field(..., description="List of all roles")
+    total: int = Field(..., description="Total count of roles")
+
+
+class GroupRoleAssignmentResponse(BaseSchema):
+    """Response from group role assignment operations"""
+
+    group_id: uuid.UUID = Field(..., description="Group ID")
+    group_name: str = Field(..., description="Group name")
+    roles: list[str] = Field(..., description="Currently assigned role names")
+
+
+# =============================================================================
+# User Profile (Phase 4)
+# =============================================================================
+
+class UserProfileResponse(BaseSchema):
+    """Response from GET /users/me - full user profile"""
+
+    user: UserSchema = Field(..., description="User profile")
+    roles: list[RoleSchema] = Field(..., description="User's permission roles")
+    preferences: UserPreferencesSchema = Field(..., description="User preferences")
+
+
+class PreferencesUpdateRequest(BaseSchema):
+    """Request for PATCH /users/me/preferences"""
+
+    theme: Optional[str] = Field(None, description="Theme preference: system, light, dark")
+    default_room: Optional[str] = Field(None, description="Default room for housekeeping")
+    preferences_json: Optional[dict] = Field(None, description="Extended preferences (merged)")
+
+
+# =============================================================================
+# API Keys (Phase 4)
+# =============================================================================
+
+class ApiKeyCreateRequest(BaseSchema):
+    """Request for POST /users/me/api-keys"""
+
+    name: str = Field(..., min_length=1, max_length=100, description="Human-readable key name")
+    scopes: Optional[list[str]] = Field(None, description="Optional scope restriction (role names)")
+    expires_in_days: Optional[int] = Field(None, ge=1, le=365, description="Days until expiration (optional)")
+
+
+class ApiKeyCreateResponse(BaseSchema):
+    """Response from POST /users/me/api-keys - includes the key (shown only once)"""
+
+    id: uuid.UUID = Field(..., description="API key ID")
+    name: str = Field(..., description="Key name")
+    key: str = Field(..., description="The API key (shown only once!)")
+    key_prefix: str = Field(..., description="Key prefix for identification")
+    scopes: Optional[list[str]] = Field(None, description="Scope restriction")
+    expires_at: Optional[datetime] = Field(None, description="Expiration timestamp")
+    created_at: datetime = Field(..., description="Creation timestamp")
+
+
+class ApiKeySchema(BaseSchema):
+    """API key information (without the actual key)"""
+
+    id: uuid.UUID = Field(..., description="API key ID")
+    name: str = Field(..., description="Key name")
+    key_prefix: str = Field(..., description="Key prefix for identification (e.g., 'tak_abc1')")
+    scopes: Optional[list[str]] = Field(None, description="Scope restriction")
+    expires_at: Optional[datetime] = Field(None, description="Expiration timestamp")
+    last_used_at: Optional[datetime] = Field(None, description="Last usage timestamp")
+    created_at: datetime = Field(..., description="Creation timestamp")
+    is_expired: bool = Field(..., description="Whether the key has expired")
+
+
+class ApiKeysListResponse(BaseSchema):
+    """Response from GET /users/me/api-keys"""
+
+    items: list[ApiKeySchema] = Field(..., description="List of API keys")
+    total: int = Field(..., description="Total count of keys")

src/domains/auth/service.py

@@ -3,9 +3,11 @@ Authentication Service
 
 Business logic for user synchronization from Authentik.
 """
+import hashlib
 import re
+import secrets
 import uuid
-from datetime import datetime, timezone
+from datetime import datetime, timezone, timedelta
 from typing import Optional
 
 import httpx
@@ -15,10 +17,11 @@ from sqlalchemy.orm import selectinload
 
 from src.shared.config import get_settings
 from src.shared.logging import get_logger
-from src.domains.auth.models import User, Role, UserPreferences, Group
+from src.domains.auth.models import User, Role, UserPreferences, Group, ApiKey
 from src.domains.auth.schemas import (
     TokenInfoSchema, UserSchema, RoleSchema, UserPreferencesSchema,
-    UserListItemSchema, BulkSyncResultSchema, GroupListItemSchema
+    UserListItemSchema, BulkSyncResultSchema, GroupListItemSchema,
+    ApiKeySchema, ApiKeyCreateResponse,
 )
 
 logger = get_logger(__name__)
@@ -141,30 +144,43 @@ class AuthService:
         await self.session.flush()
         return user, is_new
 
-    async def sync_roles(self, user: User, groups: list[str]) -> list[Role]:
+    async def sync_roles(self, user: User, group_names: list[str]) -> list[Role]:
         """
-        Synchronize user roles from Authentik groups
+        Synchronize user roles from Authentik groups via group_roles mapping
 
-        Maps Authentik groups (e.g., 'tatlock-control-room-admin')
+        Looks up the user's groups in the database, then retrieves all roles
-        to application roles (e.g., 'control-room:admin').
+        assigned to those groups via the group_roles mapping table.
 
         Args:
             user: User to sync roles for
-            groups: List of Authentik group names
+            group_names: List of Authentik group names
 
         Returns:
             List of synced Role objects
         """
-        # Get all roles that match the user's Authentik groups
+        # Find local Group records matching the Authentik group names
-        stmt = select(Role).where(Role.authentik_group.in_(groups))
+        stmt = (
+            select(Group)
+            .options(selectinload(Group.roles))
+            .where(Group.name.in_(group_names))
+        )
         result = await self.session.execute(stmt)
-        matching_roles = list(result.scalars().all())
+        matching_groups = list(result.scalars().all())
+
+        # Collect all unique roles from all matching groups
+        roles_set: dict[uuid.UUID, Role] = {}
+        for group in matching_groups:
+            for role in group.roles:
+                roles_set[role.id] = role
+
+        matching_roles = list(roles_set.values())
 
         # Clear existing roles and set new ones
         user.roles = matching_roles
 
         role_names = [r.name for r in matching_roles]
-        logger.info(f"Synced roles for {user.email}: {role_names}")
+        group_names_found = [g.name for g in matching_groups]
+        logger.info(f"Synced roles for {user.email} via groups {group_names_found}: {role_names}")
 
         return matching_roles
 
@@ -183,7 +199,7 @@ class AuthService:
     def roles_to_schema(self, roles: list[Role]) -> list[RoleSchema]:
         """Convert Role models to schemas"""
         return [
-            RoleSchema(name=r.name, domain=r.domain, action=r.action)
+            RoleSchema(id=r.id, name=r.name, domain=r.domain, category=r.category, action=r.action)
             for r in roles
         ]
 
@@ -492,8 +508,8 @@ class AuthService:
         """
         from sqlalchemy import func
 
-        # Base query
+        # Base query with roles loaded
-        base_query = select(Group)
+        base_query = select(Group).options(selectinload(Group.roles))
 
         # Apply search filter if provided
         if search:
@@ -520,12 +536,106 @@ class AuthService:
                 parent_name=group.parent_name,
                 member_count=group.member_count,
                 synced_at=group.synced_at,
+                roles=[role.name for role in group.roles],
             )
             for group in groups
         ]
 
         return items, total
 
+    async def list_roles(self) -> list[Role]:
+        """
+        List all available roles
+
+        Returns:
+            List of all Role objects
+        """
+        stmt = select(Role).order_by(Role.domain, Role.category, Role.action)
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
+    async def get_group_by_id(self, group_id: uuid.UUID) -> Optional[Group]:
+        """
+        Get a group by its ID with roles loaded
+
+        Args:
+            group_id: The group's UUID
+
+        Returns:
+            Group object or None if not found
+        """
+        stmt = (
+            select(Group)
+            .options(selectinload(Group.roles))
+            .where(Group.id == group_id)
+        )
+        result = await self.session.execute(stmt)
+        return result.scalar_one_or_none()
+
+    async def assign_role_to_group(self, group_id: uuid.UUID, role_id: uuid.UUID) -> Group:
+        """
+        Assign a role to a group
+
+        Args:
+            group_id: The group's UUID
+            role_id: The role's UUID to assign
+
+        Returns:
+            Updated Group object
+
+        Raises:
+            ValueError: If group or role not found
+        """
+        group = await self.get_group_by_id(group_id)
+        if not group:
+            raise ValueError(f"Group not found: {group_id}")
+
+        stmt = select(Role).where(Role.id == role_id)
+        result = await self.session.execute(stmt)
+        role = result.scalar_one_or_none()
+        if not role:
+            raise ValueError(f"Role not found: {role_id}")
+
+        # Add role if not already assigned
+        if role not in group.roles:
+            group.roles.append(role)
+            await self.session.flush()
+            logger.info(f"Assigned role {role.name} to group {group.name}")
+
+        return group
+
+    async def remove_role_from_group(self, group_id: uuid.UUID, role_id: uuid.UUID) -> Group:
+        """
+        Remove a role from a group
+
+        Args:
+            group_id: The group's UUID
+            role_id: The role's UUID to remove
+
+        Returns:
+            Updated Group object
+
+        Raises:
+            ValueError: If group or role not found
+        """
+        group = await self.get_group_by_id(group_id)
+        if not group:
+            raise ValueError(f"Group not found: {group_id}")
+
+        stmt = select(Role).where(Role.id == role_id)
+        result = await self.session.execute(stmt)
+        role = result.scalar_one_or_none()
+        if not role:
+            raise ValueError(f"Role not found: {role_id}")
+
+        # Remove role if assigned
+        if role in group.roles:
+            group.roles.remove(role)
+            await self.session.flush()
+            logger.info(f"Removed role {role.name} from group {group.name}")
+
+        return group
+
     async def bulk_sync_groups_from_authentik(self) -> BulkSyncResultSchema:
         """
         Fetch all groups from Authentik admin API and sync to local database
@@ -627,6 +737,268 @@ class AuthService:
             errors=errors,
         )
 
+    # =========================================================================
+    # Phase 4: User Profile & Settings
+    # =========================================================================
+
+    async def get_user_by_authentik_id(self, authentik_id: uuid.UUID) -> Optional[User]:
+        """
+        Get user by Authentik UUID with roles and preferences loaded
+
+        Args:
+            authentik_id: The Authentik user UUID (from JWT 'sub' claim)
+
+        Returns:
+            User object or None if not found
+        """
+        stmt = (
+            select(User)
+            .options(selectinload(User.roles), selectinload(User.preferences))
+            .where(User.authentik_id == authentik_id)
+        )
+        result = await self.session.execute(stmt)
+        return result.scalar_one_or_none()
+
+    async def get_user_by_id(self, user_id: uuid.UUID) -> Optional[User]:
+        """
+        Get user by internal UUID with roles and preferences loaded
+
+        Args:
+            user_id: The internal user UUID
+
+        Returns:
+            User object or None if not found
+        """
+        stmt = (
+            select(User)
+            .options(selectinload(User.roles), selectinload(User.preferences))
+            .where(User.id == user_id)
+        )
+        result = await self.session.execute(stmt)
+        return result.scalar_one_or_none()
+
+    async def update_preferences(
+        self,
+        user_id: uuid.UUID,
+        theme: Optional[str] = None,
+        default_room: Optional[str] = None,
+        preferences_json: Optional[dict] = None,
+    ) -> UserPreferences:
+        """
+        Update user preferences
+
+        Args:
+            user_id: User's UUID
+            theme: New theme value (or None to keep existing)
+            default_room: New default room (or None to keep existing)
+            preferences_json: JSON to merge with existing (or None to keep existing)
+
+        Returns:
+            Updated UserPreferences object
+
+        Raises:
+            ValueError: If user not found
+        """
+        stmt = select(UserPreferences).where(UserPreferences.user_id == user_id)
+        result = await self.session.execute(stmt)
+        prefs = result.scalar_one_or_none()
+
+        if prefs is None:
+            # Create preferences if they don't exist
+            prefs = UserPreferences(user_id=user_id)
+            self.session.add(prefs)
+
+        if theme is not None:
+            if theme not in ("system", "light", "dark"):
+                raise ValueError(f"Invalid theme: {theme}")
+            prefs.theme = theme
+
+        if default_room is not None:
+            prefs.default_room = default_room
+
+        if preferences_json is not None:
+            # Merge with existing preferences
+            existing = prefs.preferences_json or {}
+            existing.update(preferences_json)
+            prefs.preferences_json = existing
+
+        await self.session.flush()
+        logger.info(f"Updated preferences for user {user_id}")
+        return prefs
+
+    # =========================================================================
+    # Phase 4: API Keys
+    # =========================================================================
+
+    def _generate_api_key(self) -> tuple[str, str, str]:
+        """
+        Generate a new API key
+
+        Returns:
+            Tuple of (full_key, key_hash, key_prefix)
+        """
+        # Generate 32 random bytes = 256 bits of entropy
+        random_bytes = secrets.token_bytes(32)
+        # Encode as base64-like string (URL-safe)
+        key_body = secrets.token_urlsafe(32)
+        # Prefix with 'tak_' (tatlock api key)
+        full_key = f"tak_{key_body}"
+        # Hash for storage
+        key_hash = hashlib.sha256(full_key.encode()).hexdigest()
+        # Prefix for identification (first 8 chars after 'tak_')
+        key_prefix = f"tak_{key_body[:4]}"
+
+        return full_key, key_hash, key_prefix
+
+    async def create_api_key(
+        self,
+        user_id: uuid.UUID,
+        name: str,
+        scopes: Optional[list[str]] = None,
+        expires_in_days: Optional[int] = None,
+    ) -> tuple[ApiKey, str]:
+        """
+        Create a new API key for a user
+
+        Args:
+            user_id: User's UUID
+            name: Human-readable key name
+            scopes: Optional list of scope restrictions
+            expires_in_days: Optional expiration in days
+
+        Returns:
+            Tuple of (ApiKey object, full key string)
+            The full key is only returned once at creation!
+
+        Raises:
+            ValueError: If user not found or API keys disabled
+        """
+        # Check user exists and has API keys enabled
+        user = await self.get_user_by_id(user_id)
+        if user is None:
+            raise ValueError(f"User not found: {user_id}")
+        if not user.api_keys_enabled:
+            raise ValueError("API keys are disabled for this user")
+
+        # Generate the key
+        full_key, key_hash, key_prefix = self._generate_api_key()
+
+        # Calculate expiration
+        expires_at = None
+        if expires_in_days:
+            expires_at = datetime.now(timezone.utc) + timedelta(days=expires_in_days)
+
+        # Create the key record
+        api_key = ApiKey(
+            user_id=user_id,
+            name=name,
+            key_hash=key_hash,
+            key_prefix=key_prefix,
+            scopes=scopes,
+            expires_at=expires_at,
+        )
+        self.session.add(api_key)
+        await self.session.flush()
+
+        logger.info(f"Created API key '{name}' for user {user_id}")
+        return api_key, full_key
+
+    async def list_user_api_keys(self, user_id: uuid.UUID) -> list[ApiKey]:
+        """
+        List all API keys for a user
+
+        Args:
+            user_id: User's UUID
+
+        Returns:
+            List of ApiKey objects (without the actual keys)
+        """
+        stmt = (
+            select(ApiKey)
+            .where(ApiKey.user_id == user_id)
+            .order_by(ApiKey.created_at.desc())
+        )
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
+    async def delete_api_key(self, user_id: uuid.UUID, key_id: uuid.UUID) -> bool:
+        """
+        Delete an API key
+
+        Args:
+            user_id: User's UUID (for authorization)
+            key_id: API key UUID
+
+        Returns:
+            True if deleted, False if not found
+
+        Raises:
+            ValueError: If key belongs to different user
+        """
+        stmt = select(ApiKey).where(ApiKey.id == key_id)
+        result = await self.session.execute(stmt)
+        api_key = result.scalar_one_or_none()
+
+        if api_key is None:
+            return False
+
+        if api_key.user_id != user_id:
+            raise ValueError("API key belongs to different user")
+
+        await self.session.delete(api_key)
+        await self.session.flush()
+        logger.info(f"Deleted API key {key_id} for user {user_id}")
+        return True
+
+    async def validate_api_key(self, key: str) -> Optional[User]:
+        """
+        Validate an API key and return the associated user
+
+        Args:
+            key: The full API key string
+
+        Returns:
+            User object if valid, None if invalid/expired
+        """
+        # Hash the provided key
+        key_hash = hashlib.sha256(key.encode()).hexdigest()
+
+        # Look up by hash
+        stmt = (
+            select(ApiKey)
+            .options(selectinload(ApiKey.user).selectinload(User.roles))
+            .where(ApiKey.key_hash == key_hash)
+        )
+        result = await self.session.execute(stmt)
+        api_key = result.scalar_one_or_none()
+
+        if api_key is None:
+            return None
+
+        # Check expiration
+        if api_key.is_expired:
+            logger.warning(f"Expired API key used: {api_key.key_prefix}...")
+            return None
+
+        # Update last used timestamp
+        api_key.last_used_at = datetime.now(timezone.utc)
+
+        logger.debug(f"API key authenticated: {api_key.key_prefix}... for user {api_key.user.email}")
+        return api_key.user
+
+    def api_key_to_schema(self, api_key: ApiKey) -> ApiKeySchema:
+        """Convert ApiKey model to schema"""
+        return ApiKeySchema(
+            id=api_key.id,
+            name=api_key.name,
+            key_prefix=api_key.key_prefix,
+            scopes=api_key.scopes,
+            expires_at=api_key.expires_at,
+            last_used_at=api_key.last_used_at,
+            created_at=api_key.created_at,
+            is_expired=api_key.is_expired,
+        )
+
 
 # Factory function for dependency injection
 def get_auth_service(session: AsyncSession) -> AuthService:

src/shared/config.py

@@ -36,8 +36,15 @@ class Settings(BaseSettings):
     host: str = "0.0.0.0"
     port: int = 8083
 
-    # CORS
+    # CORS - Note: When cors_credentials is True, cannot use "*" for origins
-    cors_origins: list[str] = ["*"]
+    # Set CORS_ORIGINS env var to override (comma-separated list)
+    cors_origins: list[str] = [
+        "https://home.schweitz.net",
+        "https://tatlock.schweitz.net",
+        "http://localhost:8080",
+        "http://localhost:3000",
+        "http://127.0.0.1:8080",
+    ]
     cors_credentials: bool = True
     cors_methods: list[str] = ["*"]
     cors_headers: list[str] = ["*"]

tests/test_auth_service.py

@@ -0,0 +1,627 @@
+"""Tests for authentication service."""
+import uuid
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.domains.auth.models import User, Role, Group, UserPreferences
+from src.domains.auth.schemas import TokenInfoSchema, RoleSchema
+from src.domains.auth.service import AuthService, get_auth_service
+
+
+# =============================================================================
+# Fixtures
+# =============================================================================
+
+@pytest.fixture
+def mock_session():
+    """Create a mock async database session."""
+    session = AsyncMock(spec=AsyncSession)
+    session.execute = AsyncMock()
+    session.commit = AsyncMock()
+    session.flush = AsyncMock()
+    session.refresh = AsyncMock()
+    session.add = MagicMock()
+    return session
+
+
+@pytest.fixture
+def auth_service(mock_session):
+    """Create an AuthService instance with mock session."""
+    return AuthService(mock_session)
+
+
+@pytest.fixture
+def sample_user():
+    """Create a sample user for testing."""
+    user = User(
+        id=uuid.uuid4(),
+        authentik_id=uuid.uuid4(),
+        email="test@example.com",
+        name="Test User",
+        avatar_url="https://example.com/avatar.jpg",
+        created_at=datetime.now(timezone.utc),
+        last_login=datetime.now(timezone.utc),
+    )
+    user.roles = []
+    user.preferences = None
+    return user
+
+
+@pytest.fixture
+def sample_role():
+    """Create a sample role for testing."""
+    return Role(
+        id=uuid.uuid4(),
+        name="control-room.general:admin",
+        domain="control-room",
+        category="general",
+        action="admin",
+    )
+
+
+@pytest.fixture
+def sample_group(sample_role):
+    """Create a sample group for testing."""
+    group = Group(
+        id=uuid.uuid4(),
+        authentik_id=uuid.uuid4(),
+        name="Administrators",
+        is_superuser=True,
+        parent_name=None,
+        member_count=5,
+        created_at=datetime.now(timezone.utc),
+        synced_at=datetime.now(timezone.utc),
+    )
+    group.roles = [sample_role]
+    return group
+
+
+@pytest.fixture
+def sample_token_info():
+    """Create sample token info from Authentik."""
+    return TokenInfoSchema(
+        sub=str(uuid.uuid4()),
+        email="test@example.com",
+        name="Test User",
+        preferred_username="testuser",
+        groups=["Administrators", "Developers"],
+        picture="https://example.com/avatar.jpg",
+    )
+
+
+@pytest.fixture
+def sample_preferences():
+    """Create sample user preferences."""
+    return UserPreferences(
+        user_id=uuid.uuid4(),
+        theme="dark",
+        default_room="control-room",
+        preferences_json={"notifications": True},
+    )
+
+
+# =============================================================================
+# AuthService Initialization Tests
+# =============================================================================
+
+class TestAuthServiceInit:
+    """Test AuthService initialization."""
+
+    def test_init_with_session(self, mock_session):
+        """AuthService should initialize with session."""
+        service = AuthService(mock_session)
+        assert service.session is mock_session
+
+    def test_init_sets_userinfo_url(self, mock_session):
+        """AuthService should set userinfo URL from settings."""
+        service = AuthService(mock_session)
+        assert "userinfo" in service.userinfo_url
+
+    def test_get_auth_service_factory(self, mock_session):
+        """get_auth_service should return AuthService instance."""
+        service = get_auth_service(mock_session)
+        assert isinstance(service, AuthService)
+        assert service.session is mock_session
+
+
+# =============================================================================
+# Token Validation Tests
+# =============================================================================
+
+class TestValidateToken:
+    """Test token validation via Authentik userinfo endpoint."""
+
+    @pytest.mark.asyncio
+    async def test_validate_token_success(self, auth_service):
+        """validate_token should return token info on success."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {
+            "sub": str(uuid.uuid4()),
+            "email": "test@example.com",
+            "name": "Test User",
+            "preferred_username": "testuser",
+            "groups": ["Administrators"],
+            "picture": "https://example.com/avatar.jpg",
+        }
+        mock_response.raise_for_status = MagicMock()
+
+        with patch("src.domains.auth.service.httpx.AsyncClient") as mock_client:
+            mock_client.return_value.__aenter__.return_value.get = AsyncMock(
+                return_value=mock_response
+            )
+
+            result = await auth_service.validate_token("valid_token")
+
+            assert result.email == "test@example.com"
+            assert result.name == "Test User"
+            assert "Administrators" in result.groups
+
+    @pytest.mark.asyncio
+    async def test_validate_token_invalid(self, auth_service):
+        """validate_token should raise ValueError for invalid token."""
+        mock_response = MagicMock()
+        mock_response.status_code = 401
+
+        with patch("src.domains.auth.service.httpx.AsyncClient") as mock_client:
+            mock_client.return_value.__aenter__.return_value.get = AsyncMock(
+                return_value=mock_response
+            )
+
+            with pytest.raises(ValueError, match="Invalid or expired token"):
+                await auth_service.validate_token("invalid_token")
+
+    @pytest.mark.asyncio
+    async def test_validate_token_service_unavailable(self, auth_service):
+        """validate_token should raise ValueError when service unavailable."""
+        import httpx
+
+        with patch("src.domains.auth.service.httpx.AsyncClient") as mock_client:
+            mock_client.return_value.__aenter__.return_value.get = AsyncMock(
+                side_effect=httpx.RequestError("Connection failed")
+            )
+
+            with pytest.raises(ValueError, match="Authentication service unavailable"):
+                await auth_service.validate_token("token")
+
+
+# =============================================================================
+# User Sync Tests
+# =============================================================================
+
+class TestSyncUser:
+    """Test user synchronization from OIDC token."""
+
+    @pytest.mark.asyncio
+    async def test_sync_user_creates_new_user(self, auth_service, sample_token_info, mock_session):
+        """sync_user should create new user when not found."""
+        # Mock no existing user found
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = None
+        mock_session.execute.return_value = mock_result
+
+        user, is_new = await auth_service.sync_user(sample_token_info)
+
+        assert is_new is True
+        assert mock_session.add.call_count == 2  # User and Preferences
+
+    @pytest.mark.asyncio
+    async def test_sync_user_updates_existing_user(
+        self, auth_service, sample_token_info, sample_user, mock_session
+    ):
+        """sync_user should update existing user when found."""
+        # Mock existing user found
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = sample_user
+        mock_session.execute.return_value = mock_result
+
+        # Update token info with matching authentik_id
+        sample_token_info.sub = str(sample_user.authentik_id)
+
+        user, is_new = await auth_service.sync_user(sample_token_info)
+
+        assert is_new is False
+        assert user.email == sample_token_info.email
+        assert user.name == sample_token_info.name
+
+    @pytest.mark.asyncio
+    async def test_sync_user_updates_last_login(
+        self, auth_service, sample_token_info, sample_user, mock_session
+    ):
+        """sync_user should update last_login timestamp."""
+        old_login = sample_user.last_login
+
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = sample_user
+        mock_session.execute.return_value = mock_result
+
+        sample_token_info.sub = str(sample_user.authentik_id)
+
+        user, _ = await auth_service.sync_user(sample_token_info)
+
+        assert user.last_login is not None
+        # last_login should be updated (or same if happened in same second)
+        assert user.last_login >= old_login or user.last_login is not None
+
+
+# =============================================================================
+# Role Sync Tests
+# =============================================================================
+
+class TestSyncRoles:
+    """Test role synchronization from Authentik groups via group_roles."""
+
+    @pytest.mark.asyncio
+    async def test_sync_roles_from_groups(
+        self, auth_service, sample_user, sample_group, mock_session
+    ):
+        """sync_roles should get roles from matching groups."""
+        # Mock finding groups with roles
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = [sample_group]
+        mock_session.execute.return_value = mock_result
+
+        roles = await auth_service.sync_roles(sample_user, ["Administrators"])
+
+        assert len(roles) == 1
+        assert roles[0].name == "control-room.general:admin"
+        assert sample_user.roles == roles
+
+    @pytest.mark.asyncio
+    async def test_sync_roles_no_matching_groups(
+        self, auth_service, sample_user, mock_session
+    ):
+        """sync_roles should return empty list when no groups match."""
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = []
+        mock_session.execute.return_value = mock_result
+
+        roles = await auth_service.sync_roles(sample_user, ["NonExistentGroup"])
+
+        assert len(roles) == 0
+        assert sample_user.roles == []
+
+    @pytest.mark.asyncio
+    async def test_sync_roles_deduplicates_roles(
+        self, auth_service, sample_user, sample_role, mock_session
+    ):
+        """sync_roles should deduplicate roles from multiple groups."""
+        # Create two groups with the same role
+        group1 = Group(
+            id=uuid.uuid4(),
+            authentik_id=uuid.uuid4(),
+            name="Group1",
+            is_superuser=False,
+            member_count=1,
+            created_at=datetime.now(timezone.utc),
+            synced_at=datetime.now(timezone.utc),
+        )
+        group1.roles = [sample_role]
+
+        group2 = Group(
+            id=uuid.uuid4(),
+            authentik_id=uuid.uuid4(),
+            name="Group2",
+            is_superuser=False,
+            member_count=1,
+            created_at=datetime.now(timezone.utc),
+            synced_at=datetime.now(timezone.utc),
+        )
+        group2.roles = [sample_role]  # Same role
+
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = [group1, group2]
+        mock_session.execute.return_value = mock_result
+
+        roles = await auth_service.sync_roles(sample_user, ["Group1", "Group2"])
+
+        # Should only have one role despite appearing in two groups
+        assert len(roles) == 1
+
+
+# =============================================================================
+# Schema Conversion Tests
+# =============================================================================
+
+class TestSchemaConversions:
+    """Test model to schema conversions."""
+
+    def test_user_to_schema(self, auth_service, sample_user):
+        """user_to_schema should convert User model to UserSchema."""
+        schema = auth_service.user_to_schema(sample_user)
+
+        assert schema.id == sample_user.id
+        assert schema.authentik_id == sample_user.authentik_id
+        assert schema.email == sample_user.email
+        assert schema.name == sample_user.name
+        assert schema.avatar_url == sample_user.avatar_url
+
+    def test_roles_to_schema(self, auth_service, sample_role):
+        """roles_to_schema should convert Role models to RoleSchemas."""
+        schemas = auth_service.roles_to_schema([sample_role])
+
+        assert len(schemas) == 1
+        assert schemas[0].id == sample_role.id
+        assert schemas[0].name == sample_role.name
+        assert schemas[0].domain == sample_role.domain
+        assert schemas[0].category == sample_role.category
+        assert schemas[0].action == sample_role.action
+
+    def test_roles_to_schema_empty_list(self, auth_service):
+        """roles_to_schema should handle empty list."""
+        schemas = auth_service.roles_to_schema([])
+        assert schemas == []
+
+    def test_preferences_to_schema(self, auth_service, sample_preferences):
+        """preferences_to_schema should convert UserPreferences to schema."""
+        schema = auth_service.preferences_to_schema(sample_preferences)
+
+        assert schema.theme == sample_preferences.theme
+        assert schema.default_room == sample_preferences.default_room
+        assert schema.preferences_json == sample_preferences.preferences_json
+
+    def test_preferences_to_schema_none(self, auth_service):
+        """preferences_to_schema should return defaults for None."""
+        schema = auth_service.preferences_to_schema(None)
+
+        assert schema.theme == "system"
+        assert schema.default_room == "front-hall"
+        assert schema.preferences_json == {}
+
+
+# =============================================================================
+# List Operations Tests
+# =============================================================================
+
+class TestListOperations:
+    """Test list operations for users, groups, and roles."""
+
+    @pytest.mark.asyncio
+    async def test_list_users(self, auth_service, sample_user, mock_session):
+        """list_users should return paginated user list."""
+        sample_user.roles = []
+
+        # Mock count query
+        count_result = MagicMock()
+        count_result.scalar.return_value = 1
+
+        # Mock users query
+        users_result = MagicMock()
+        users_result.scalars.return_value.all.return_value = [sample_user]
+
+        mock_session.execute.side_effect = [count_result, users_result]
+
+        items, total = await auth_service.list_users()
+
+        assert total == 1
+        assert len(items) == 1
+        assert items[0].email == sample_user.email
+
+    @pytest.mark.asyncio
+    async def test_list_users_with_search(self, auth_service, mock_session):
+        """list_users should filter by search query."""
+        count_result = MagicMock()
+        count_result.scalar.return_value = 0
+
+        users_result = MagicMock()
+        users_result.scalars.return_value.all.return_value = []
+
+        mock_session.execute.side_effect = [count_result, users_result]
+
+        items, total = await auth_service.list_users(search="nonexistent")
+
+        assert total == 0
+        assert len(items) == 0
+
+    @pytest.mark.asyncio
+    async def test_list_groups(self, auth_service, sample_group, mock_session):
+        """list_groups should return paginated group list with roles."""
+        count_result = MagicMock()
+        count_result.scalar.return_value = 1
+
+        groups_result = MagicMock()
+        groups_result.scalars.return_value.all.return_value = [sample_group]
+
+        mock_session.execute.side_effect = [count_result, groups_result]
+
+        items, total = await auth_service.list_groups()
+
+        assert total == 1
+        assert len(items) == 1
+        assert items[0].name == sample_group.name
+        assert len(items[0].roles) == 1  # Should include role names
+
+    @pytest.mark.asyncio
+    async def test_list_roles(self, auth_service, sample_role, mock_session):
+        """list_roles should return all roles ordered by domain."""
+        mock_result = MagicMock()
+        mock_result.scalars.return_value.all.return_value = [sample_role]
+        mock_session.execute.return_value = mock_result
+
+        roles = await auth_service.list_roles()
+
+        assert len(roles) == 1
+        assert roles[0].name == sample_role.name
+
+
+# =============================================================================
+# Group-Role Management Tests
+# =============================================================================
+
+class TestGroupRoleManagement:
+    """Test group-role assignment and removal."""
+
+    @pytest.mark.asyncio
+    async def test_get_group_by_id(self, auth_service, sample_group, mock_session):
+        """get_group_by_id should return group with roles."""
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = sample_group
+        mock_session.execute.return_value = mock_result
+
+        group = await auth_service.get_group_by_id(sample_group.id)
+
+        assert group is not None
+        assert group.id == sample_group.id
+        assert len(group.roles) == 1
+
+    @pytest.mark.asyncio
+    async def test_get_group_by_id_not_found(self, auth_service, mock_session):
+        """get_group_by_id should return None when not found."""
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = None
+        mock_session.execute.return_value = mock_result
+
+        group = await auth_service.get_group_by_id(uuid.uuid4())
+
+        assert group is None
+
+    @pytest.mark.asyncio
+    async def test_assign_role_to_group(
+        self, auth_service, sample_group, sample_role, mock_session
+    ):
+        """assign_role_to_group should add role to group."""
+        # Clear existing roles for this test
+        sample_group.roles = []
+
+        # Mock group lookup
+        group_result = MagicMock()
+        group_result.scalar_one_or_none.return_value = sample_group
+
+        # Mock role lookup
+        role_result = MagicMock()
+        role_result.scalar_one_or_none.return_value = sample_role
+
+        mock_session.execute.side_effect = [group_result, role_result]
+
+        group = await auth_service.assign_role_to_group(sample_group.id, sample_role.id)
+
+        assert sample_role in group.roles
+        mock_session.flush.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_assign_role_to_group_already_assigned(
+        self, auth_service, sample_group, sample_role, mock_session
+    ):
+        """assign_role_to_group should not duplicate if already assigned."""
+        # Group already has this role
+        sample_group.roles = [sample_role]
+        original_count = len(sample_group.roles)
+
+        group_result = MagicMock()
+        group_result.scalar_one_or_none.return_value = sample_group
+
+        role_result = MagicMock()
+        role_result.scalar_one_or_none.return_value = sample_role
+
+        mock_session.execute.side_effect = [group_result, role_result]
+
+        group = await auth_service.assign_role_to_group(sample_group.id, sample_role.id)
+
+        assert len(group.roles) == original_count  # No duplicate
+
+    @pytest.mark.asyncio
+    async def test_assign_role_to_group_group_not_found(self, auth_service, mock_session):
+        """assign_role_to_group should raise ValueError when group not found."""
+        mock_result = MagicMock()
+        mock_result.scalar_one_or_none.return_value = None
+        mock_session.execute.return_value = mock_result
+
+        with pytest.raises(ValueError, match="Group not found"):
+            await auth_service.assign_role_to_group(uuid.uuid4(), uuid.uuid4())
+
+    @pytest.mark.asyncio
+    async def test_assign_role_to_group_role_not_found(
+        self, auth_service, sample_group, mock_session
+    ):
+        """assign_role_to_group should raise ValueError when role not found."""
+        sample_group.roles = []
+
+        group_result = MagicMock()
+        group_result.scalar_one_or_none.return_value = sample_group
+
+        role_result = MagicMock()
+        role_result.scalar_one_or_none.return_value = None
+
+        mock_session.execute.side_effect = [group_result, role_result]
+
+        with pytest.raises(ValueError, match="Role not found"):
+            await auth_service.assign_role_to_group(sample_group.id, uuid.uuid4())
+
+    @pytest.mark.asyncio
+    async def test_remove_role_from_group(
+        self, auth_service, sample_group, sample_role, mock_session
+    ):
+        """remove_role_from_group should remove role from group."""
+        # Group has this role
+        sample_group.roles = [sample_role]
+
+        group_result = MagicMock()
+        group_result.scalar_one_or_none.return_value = sample_group
+
+        role_result = MagicMock()
+        role_result.scalar_one_or_none.return_value = sample_role
+
+        mock_session.execute.side_effect = [group_result, role_result]
+
+        group = await auth_service.remove_role_from_group(sample_group.id, sample_role.id)
+
+        assert sample_role not in group.roles
+        mock_session.flush.assert_called()
+
+    @pytest.mark.asyncio
+    async def test_remove_role_from_group_not_assigned(
+        self, auth_service, sample_group, sample_role, mock_session
+    ):
+        """remove_role_from_group should handle role not assigned gracefully."""
+        # Group does not have this role
+        sample_group.roles = []
+
+        group_result = MagicMock()
+        group_result.scalar_one_or_none.return_value = sample_group
+
+        role_result = MagicMock()
+        role_result.scalar_one_or_none.return_value = sample_role
+
+        mock_session.execute.side_effect = [group_result, role_result]
+
+        group = await auth_service.remove_role_from_group(sample_group.id, sample_role.id)
+
+        # Should complete without error
+        assert len(group.roles) == 0
+
+
+# =============================================================================
+# Role Schema Tests
+# =============================================================================
+
+class TestRoleSchema:
+    """Test RoleSchema validation."""
+
+    def test_role_schema_creation(self):
+        """RoleSchema should be creatable with valid data."""
+        schema = RoleSchema(
+            id=uuid.uuid4(),
+            name="control-room.general:admin",
+            domain="control-room",
+            category="general",
+            action="admin",
+        )
+
+        assert schema.name == "control-room.general:admin"
+        assert schema.domain == "control-room"
+        assert schema.category == "general"
+        assert schema.action == "admin"
+
+    def test_role_schema_category_default(self):
+        """RoleSchema should default category to 'general'."""
+        schema = RoleSchema(
+            id=uuid.uuid4(),
+            name="media.general:viewer",
+            domain="media",
+            action="viewer",
+        )
+
+        assert schema.category == "general"