Jeroen SchweitzerandClaude Opus 4.5
397a47c8fc
Build and Push / build (release) Successful in 1m10s
Add user profile, preferences, and API key management endpoints: - 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 use tak_ prefix, SHA-256 hashing, and are shown only once on creation. Preferences support partial updates with JSON merge behavior. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
212 lines
8.6 KiB
Python
212 lines
8.6 KiB
Python
"""
|
|
Authentication Schemas
|
|
|
|
Pydantic models for auth request/response payloads.
|
|
"""
|
|
import uuid
|
|
from datetime import datetime
|
|
from typing import Optional
|
|
from pydantic import Field
|
|
|
|
from src.shared.base import BaseSchema
|
|
|
|
|
|
class AuthSyncRequest(BaseSchema):
|
|
"""
|
|
Request payload for POST /auth/sync
|
|
|
|
The client sends this after obtaining an OIDC token from Authentik.
|
|
The access_token is validated against Authentik's userinfo endpoint.
|
|
"""
|
|
|
|
access_token: str = Field(
|
|
...,
|
|
description="OIDC access token from Authentik",
|
|
)
|
|
|
|
|
|
class RoleSchema(BaseSchema):
|
|
"""Role information in domain.category:action format"""
|
|
|
|
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')")
|
|
|
|
|
|
class UserPreferencesSchema(BaseSchema):
|
|
"""User preferences"""
|
|
|
|
theme: str = Field(default="system", description="Theme preference: system, light, dark")
|
|
default_room: str = Field(default="front-hall", description="Default room for housekeeping")
|
|
preferences_json: dict = Field(default_factory=dict, description="Extended preferences")
|
|
|
|
|
|
class UserSchema(BaseSchema):
|
|
"""User information returned from sync"""
|
|
|
|
id: uuid.UUID = Field(..., description="Internal user ID")
|
|
authentik_id: uuid.UUID = Field(..., description="Authentik user ID")
|
|
email: str = Field(..., description="User email")
|
|
name: str = Field(..., description="Display name")
|
|
avatar_url: Optional[str] = Field(None, description="Profile picture URL")
|
|
created_at: datetime = Field(..., description="Account creation timestamp")
|
|
last_login: Optional[datetime] = Field(None, description="Last login timestamp")
|
|
|
|
|
|
class AuthSyncResponse(BaseSchema):
|
|
"""
|
|
Response from POST /auth/sync
|
|
|
|
Contains the synced user profile, roles, and preferences.
|
|
"""
|
|
|
|
user: UserSchema = Field(..., description="User profile")
|
|
roles: list[RoleSchema] = Field(..., description="User's permission roles")
|
|
preferences: UserPreferencesSchema = Field(..., description="User preferences")
|
|
is_new_user: bool = Field(..., description="True if user was just created")
|
|
|
|
|
|
class TokenInfoSchema(BaseSchema):
|
|
"""
|
|
Token information from Authentik userinfo endpoint
|
|
|
|
This is what Authentik returns when validating an access token.
|
|
"""
|
|
|
|
sub: str = Field(..., description="Subject (Authentik user ID)")
|
|
email: str = Field(..., description="User email")
|
|
name: Optional[str] = Field(None, description="Display name")
|
|
preferred_username: Optional[str] = Field(None, description="Username")
|
|
groups: list[str] = Field(default_factory=list, description="Group memberships")
|
|
picture: Optional[str] = Field(None, description="Profile picture URL")
|
|
|
|
|
|
class UserListItemSchema(BaseSchema):
|
|
"""User item for list display"""
|
|
|
|
id: uuid.UUID = Field(..., description="Internal user ID")
|
|
email: str = Field(..., description="User email")
|
|
name: str = Field(..., description="Display name")
|
|
avatar_url: Optional[str] = Field(None, description="Profile picture URL")
|
|
created_at: datetime = Field(..., description="Account creation timestamp")
|
|
last_login: Optional[datetime] = Field(None, description="Last login timestamp")
|
|
roles: list[str] = Field(default_factory=list, description="Role names")
|
|
|
|
|
|
class UsersListResponse(BaseSchema):
|
|
"""Response from GET /auth/users"""
|
|
|
|
items: list[UserListItemSchema] = Field(..., description="List of users")
|
|
total: int = Field(..., description="Total count of users")
|
|
|
|
|
|
class BulkSyncResultSchema(BaseSchema):
|
|
"""Result from bulk sync operation"""
|
|
|
|
created: int = Field(..., description="Number of users created")
|
|
updated: int = Field(..., description="Number of users updated")
|
|
failed: int = Field(..., description="Number of users that failed to sync")
|
|
total_in_authentik: int = Field(..., description="Total users in Authentik")
|
|
errors: list[str] = Field(default_factory=list, description="Error messages for failed syncs")
|
|
|
|
|
|
class GroupListItemSchema(BaseSchema):
|
|
"""Group item for list display"""
|
|
|
|
id: uuid.UUID = Field(..., description="Internal group ID")
|
|
authentik_id: uuid.UUID = Field(..., description="Authentik group ID")
|
|
name: str = Field(..., description="Group name")
|
|
is_superuser: bool = Field(default=False, description="Whether group has superuser privileges")
|
|
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):
|
|
"""Response from GET /auth/groups"""
|
|
|
|
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")
|