""" Authentication router — login, token refresh, profile, first-run setup, and optional OIDC (Authentik) sign-in. """ import os import re import secrets import logging from typing import Optional from urllib.parse import urlencode, urlparse from fastapi import APIRouter, Depends, HTTPException, Request, status from fastapi.responses import RedirectResponse from pydantic import BaseModel from sqlalchemy import select, func as sa_func from sqlalchemy.ext.asyncio import AsyncSession from app.auth import hash_password, verify_password, create_access_token, create_refresh_token, decode_token from app.auth_oidc import get_client as get_oidc_client, is_enabled as oidc_is_enabled, provider_label, PROVIDER_NAME from app.database import get_db from app.dependencies import get_current_user from app.models.user import User from app.models.folders import SourceRoot from app.services.gravatar import gravatar_url from app.config import settings logger = logging.getLogger(__name__) router = APIRouter() # --------------------------------------------------------------------------- # Request / response schemas # --------------------------------------------------------------------------- class LoginRequest(BaseModel): username: str password: str class TokenResponse(BaseModel): access_token: str refresh_token: str token_type: str = "bearer" class RefreshRequest(BaseModel): refresh_token: str class UserResponse(BaseModel): id: str username: str email: Optional[str] role: str is_active: bool created_at: Optional[str] avatar_url: Optional[str] = None display_name: Optional[str] = None # Nextcloud integration — username override (defaults to OIDC # preferred_username) and a flag for whether the user has stored # an app password. Cleartext passwords are never serialized. nextcloud_username: Optional[str] = None has_nextcloud_app_password: bool = False class UpdateMeRequest(BaseModel): """PATCH /me payload. Every field is optional — only what's set gets touched. Setting `nextcloud_app_password` to "" clears it.""" nextcloud_username: Optional[str] = None nextcloud_app_password: Optional[str] = None class SetupRequest(BaseModel): username: str password: str class ChangePasswordRequest(BaseModel): current_password: str new_password: str class OidcConfig(BaseModel): enabled: bool label: str login_url: str class AuthConfigResponse(BaseModel): # None when OIDC is disabled / not configured — the frontend uses # that to hide the SSO button. oidc: Optional[OidcConfig] = None # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def serialize_user(user: User) -> UserResponse: """Build the user-facing payload, computing the avatar URL with the `provider picture > Gravatar > None` fallback chain.""" avatar = user.avatar_url or gravatar_url(user.email) return UserResponse( id=user.id, username=user.username, email=user.email, role=user.role, is_active=user.is_active, created_at=user.created_at.isoformat() if user.created_at else None, avatar_url=avatar, display_name=user.display_name, nextcloud_username=user.nextcloud_username, has_nextcloud_app_password=bool(user.nextcloud_app_password_enc), ) # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @router.get("/config", response_model=AuthConfigResponse) async def auth_config(): """Public — tells the frontend which login options to render. Returns `oidc: null` when OIDC is disabled or not fully configured, so the login page can hide the SSO button without a round-trip to the IdP. The `login_url` is browser-navigable (full redirect); it starts the Authlib flow that sets the PKCE cookie. """ if not oidc_is_enabled(): return AuthConfigResponse(oidc=None) return AuthConfigResponse( oidc=OidcConfig( enabled=True, label=provider_label(), login_url="/api/v1/auth/oidc/login", ) ) @router.post("/login", response_model=TokenResponse) async def login(body: LoginRequest, db: AsyncSession = Depends(get_db)): """Authenticate with username + password, receive JWT tokens.""" result = await db.execute( select(User).where(User.username == body.username) ) user = result.scalar_one_or_none() # OIDC-only users (hashed_password IS NULL) can't sign in via this # endpoint; they must go through the SSO flow. Treat as auth failure # so we don't leak account existence. if ( user is None or not user.hashed_password or not verify_password(body.password, user.hashed_password) ): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid username or password", ) if not user.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Account is deactivated", ) return TokenResponse( access_token=create_access_token(user.id, user.role), refresh_token=create_refresh_token(user.id), ) @router.post("/refresh", response_model=TokenResponse) async def refresh_token(body: RefreshRequest, db: AsyncSession = Depends(get_db)): """Exchange a valid refresh token for a new access + refresh pair.""" try: payload = decode_token(body.refresh_token) if payload.get("type") != "refresh": raise ValueError("not a refresh token") user_id = payload["sub"] except Exception: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid or expired refresh token", ) result = await db.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None or not user.is_active: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found or deactivated", ) return TokenResponse( access_token=create_access_token(user.id, user.role), refresh_token=create_refresh_token(user.id), ) @router.get("/me", response_model=UserResponse) async def get_me(current_user: User = Depends(get_current_user)): """Return the authenticated user's profile.""" return serialize_user(current_user) _NC_USERNAME_RE = re.compile(r"^[a-zA-Z0-9._@-]{1,64}$") @router.patch("/me", response_model=UserResponse) async def update_me( body: UpdateMeRequest, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Update the authenticated user's Nextcloud integration settings. `nextcloud_username` overrides the OIDC `preferred_username` default so e.g. the local mule-image user `dtoro` can map to Nextcloud user `admin`. `nextcloud_app_password` is encrypted at rest via the Fernet helper in `services/secrets.py`; passing an empty string clears it. """ from app.services.secrets import encrypt changed = False if body.nextcloud_username is not None: candidate = body.nextcloud_username.strip() if candidate and not _NC_USERNAME_RE.match(candidate): raise HTTPException(status_code=400, detail="Invalid Nextcloud username") current_user.nextcloud_username = candidate or None changed = True if body.nextcloud_app_password is not None: if body.nextcloud_app_password == "": current_user.nextcloud_app_password_enc = None else: current_user.nextcloud_app_password_enc = encrypt(body.nextcloud_app_password) changed = True if changed: await db.commit() await db.refresh(current_user) return serialize_user(current_user) @router.post("/change-password") async def change_password( body: ChangePasswordRequest, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Change the authenticated user's password.""" if not current_user.hashed_password or not verify_password( body.current_password, current_user.hashed_password ): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Current password is incorrect", ) current_user.hashed_password = hash_password(body.new_password) await db.commit() return {"status": "ok"} @router.post("/setup", response_model=TokenResponse, status_code=201) async def setup(body: SetupRequest, db: AsyncSession = Depends(get_db)): """First-run only: create the initial admin account. Returns 409 if any user already exists. This endpoint is unauthenticated by design — it can only run once. """ count = (await db.execute(select(sa_func.count(User.id)))).scalar() if count > 0: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail="Setup already completed — users exist", ) if len(body.username.strip()) < 2: raise HTTPException(status_code=400, detail="Username must be at least 2 characters") if len(body.password) < 6: raise HTTPException(status_code=400, detail="Password must be at least 6 characters") # Every user — including the initial admin — gets their own subfolder # under the photo mount root. Nobody owns the root directory itself. media_path = os.path.join(settings.photo_dirs, body.username.strip()) os.makedirs(media_path, exist_ok=True) user = User( username=body.username.strip(), hashed_password=hash_password(body.password), role="admin", media_path=media_path, ) db.add(user) await db.flush() # get user.id before creating source root source_root = SourceRoot( name=f"{user.username}'s Library", path=media_path, user_id=user.id, ) db.add(source_root) await db.commit() logger.info(f"Initial admin account created: {user.username}") return TokenResponse( access_token=create_access_token(user.id, user.role), refresh_token=create_refresh_token(user.id), ) @router.get("/status") async def auth_status(db: AsyncSession = Depends(get_db)): """Public endpoint: returns whether setup has been completed. The frontend calls this to decide whether to show the setup page or the login page. """ count = (await db.execute(select(sa_func.count(User.id)))).scalar() return {"setup_completed": count > 0} # --------------------------------------------------------------------------- # OIDC (Authentik) sign-in # --------------------------------------------------------------------------- _USERNAME_SANITIZER = re.compile(r"[^a-zA-Z0-9_.-]+") def _derive_username(claims: dict, existing_usernames: set[str]) -> str: """Pick a local username from OIDC claims. Order of preference: 1. `preferred_username` claim (Authentik's usual choice) 2. local-part of `email` 3. `sub` claim (always present) Strips characters the rest of the app doesn't like in paths/URLs, trims to 50 chars (User.username column limit), and appends a short suffix on collision so two Authentik users can't land on the same local row. """ raw = ( claims.get("preferred_username") or (claims.get("email") or "").split("@", 1)[0] or claims.get("sub") or "user" ) base = _USERNAME_SANITIZER.sub("", str(raw)).strip("._-") or "user" base = base[:40] candidate = base suffix = 0 while candidate in existing_usernames: suffix += 1 candidate = f"{base}-{suffix}"[:50] return candidate def _frontend_origin(request: Request) -> str: """Best guess at where the SPA lives so the callback redirect lands back on the app origin. Uses the configured redirect URI's scheme + host (strips /api/... path) when available, falling back to the request's own origin.""" if settings.oidc_redirect_uri: parsed = urlparse(settings.oidc_redirect_uri) return f"{parsed.scheme}://{parsed.netloc}" return f"{request.url.scheme}://{request.url.netloc}" @router.get("/oidc/login") async def oidc_login(request: Request): """Start the OIDC flow — redirect to Authentik's authorization URL.""" if not oidc_is_enabled(): raise HTTPException(status_code=404, detail="OIDC login is not enabled") client = get_oidc_client() if client is None: raise HTTPException(status_code=500, detail="OIDC client not configured") redirect_uri = settings.oidc_redirect_uri return await client.authorize_redirect(request, redirect_uri) @router.get("/oidc/callback") async def oidc_callback(request: Request, db: AsyncSession = Depends(get_db)): """Handle the OIDC redirect — exchange code, provision/link user, issue our own JWTs, bounce back to the SPA.""" if not oidc_is_enabled(): raise HTTPException(status_code=404, detail="OIDC login is not enabled") client = get_oidc_client() if client is None: raise HTTPException(status_code=500, detail="OIDC client not configured") try: token = await client.authorize_access_token(request) except Exception as exc: logger.warning("OIDC callback: authorize_access_token failed: %s", exc) return _oidc_error_redirect(request, "oidc_exchange_failed") # `parse_id_token` verifies signature + nonce; `userinfo` fills in # claims some IdPs don't put in the ID token (e.g. picture). We # merge both, preferring userinfo when both are present. claims = dict(token.get("userinfo") or {}) if not claims: try: claims = dict(await client.userinfo(token=token)) except Exception: claims = {} id_token_claims = token.get("id_token_claims") or {} for k, v in id_token_claims.items(): claims.setdefault(k, v) sub = claims.get("sub") if not sub: logger.warning("OIDC callback: claims missing `sub` — %r", claims) return _oidc_error_redirect(request, "oidc_missing_sub") issuer = claims.get("iss") or (settings.oidc_issuer or "").rstrip("/") email = claims.get("email") display_name = claims.get("name") or claims.get("preferred_username") picture = claims.get("picture") groups = claims.get("groups") or [] if isinstance(groups, str): groups = [groups] admin_groups = set(settings.oidc_admin_group_list) role = "admin" if admin_groups and admin_groups.intersection(groups) else "user" # 1. Match by (issuer, sub) first — stable identity key. user = (await db.execute( select(User).where( User.oidc_issuer == issuer, User.oidc_sub == sub, ) )).scalar_one_or_none() # 2. Fall back to email so a pre-existing local account can be # linked on first SSO login (homelab admin keeps their row). if user is None and email: user = (await db.execute( select(User).where(User.email == email) )).scalar_one_or_none() # 3. Last-resort link by preferred_username. Off by default; only # used in trusted single-tenant setups where local accounts # predate OIDC and never collected email (the app has no UI for # it). Guarded by OIDC_LINK_BY_USERNAME to avoid hijacking # accounts in shared instances. if user is None and settings.oidc_link_by_username: preferred = claims.get("preferred_username") if preferred: user = (await db.execute( select(User).where(User.username == preferred) )).scalar_one_or_none() if user is not None: logger.info( "OIDC linked existing user %s by preferred_username", preferred, ) if user is None: if not settings.oidc_allow_signup: logger.info("OIDC signup disabled — rejecting unknown sub=%s email=%s", sub, email) return _oidc_error_redirect(request, "oidc_signup_disabled") # JIT provision. existing = { u for (u,) in (await db.execute(select(User.username))).all() } username = _derive_username(claims, existing) media_path = os.path.join(settings.photo_dirs, username) os.makedirs(media_path, exist_ok=True) user = User( username=username, email=email, hashed_password=None, role=role, is_active=True, media_path=media_path, oidc_issuer=issuer, oidc_sub=sub, avatar_url=picture, display_name=display_name, # Default the Nextcloud username from preferred_username so # the common case "same name on both sides" needs zero # configuration. Override is exposed in Settings for the # mismatch case (e.g. authentik dtoro ↔ Nextcloud admin). nextcloud_username=(claims.get("preferred_username") or None), ) db.add(user) await db.flush() db.add(SourceRoot( name=f"{user.username}'s Library", path=media_path, user_id=user.id, )) await db.commit() logger.info("OIDC JIT-created user %s (role=%s)", user.username, role) else: # Refresh profile bits + link identity if needed. We do *not* # demote admins created locally; only touch role when admin # group mapping is configured. changed = False if user.oidc_sub != sub or user.oidc_issuer != issuer: user.oidc_issuer = issuer user.oidc_sub = sub changed = True if email and user.email != email: user.email = email changed = True if display_name and user.display_name != display_name: user.display_name = display_name changed = True if picture and user.avatar_url != picture: user.avatar_url = picture changed = True # Backfill nextcloud_username on first OIDC login for users that # predate the column. NEVER overwrites a value the user already # set in Settings — once the override is non-null, it wins. if not user.nextcloud_username: preferred = claims.get("preferred_username") if preferred: user.nextcloud_username = preferred changed = True if admin_groups: new_role = "admin" if admin_groups.intersection(groups) else "user" if user.role != new_role: user.role = new_role changed = True if not user.is_active: # Don't resurrect a deactivated account — surface an error. logger.info("OIDC login rejected — user %s is deactivated", user.username) return _oidc_error_redirect(request, "oidc_deactivated") if changed: await db.commit() # Mint our own JWTs and bounce back to the SPA. Tokens ride in the # URL fragment-free for simplicity; the frontend callback page # strips them from the location bar immediately. access = create_access_token(user.id, user.role) refresh = create_refresh_token(user.id) params = urlencode({"access_token": access, "refresh_token": refresh}) target = f"{_frontend_origin(request)}/auth/callback?{params}" return RedirectResponse(url=target, status_code=302) def _oidc_error_redirect(request: Request, code: str) -> RedirectResponse: """Bounce back to the SPA with an `error=` query so the login page can render something meaningful instead of a stack trace.""" target = f"{_frontend_origin(request)}/auth/callback?error={code}" return RedirectResponse(url=target, status_code=302)