""" Jellyfin AuthService — validates Jellyfin credentials and stores the session token. Two authentication flows: 1. Quick Connect (primary): user enters a short code on their Jellyfin app. - initiate_quick_connect() → {code, secret} - poll_quick_connect(secret) → "Active" | "Authorized" | "Expired" - authenticate_quick_connect(secret) → AuthResult with token 2. Username/password (legacy): renders an HTML form, called via the REST API. - render_login_form(token, discord_id) → HTML string - authenticate(form_data) → AuthResult """ from __future__ import annotations import logging from dataclasses import dataclass from typing import Optional import httpx from auth import AuthService, AuthResult, register_auth_service from core.config import get_config logger = logging.getLogger("auth.jellyfin") # Emby-style authorization header required by Jellyfin's AuthenticateByName _EMBY_HEADER = ( 'MediaBrowser Client="AgentBot",' 'Device="DiscordBot",' 'DeviceId="agent-bot",' 'Version="1.0"' ) @dataclass class QuickConnectResult: """Result of a Quick Connect initiation.""" secret: str code: str device_id: str device_name: str class JellyfinAuth(AuthService): name = "jellyfin" display_name = "Jellyfin" # ------------------------------------------------------------------ # Quick Connect helpers # ------------------------------------------------------------------ def _qc_headers(self) -> dict[str, str]: """Return headers used by all Quick Connect API calls.""" return { "X-Emby-Authorization": ( 'MediaBrowser Client="AgentBot",' 'Device="DiscordBot",' 'DeviceId="agent-bot-qc",' 'Version="1.0"' ) } async def _resolve_url(self) -> str | None: """ Resolve the Jellyfin server URL. 1. Check JELLYFIN_URL env var (used in deployment). 2. Check if user already has a stored auth with a URL (from legacy login). Returns None if no URL is configured. """ # First: explicit env var env_url = get_config("JELLYFIN_URL") if env_url: return env_url.strip().rstrip("/") return None # ------------------------------------------------------------------ # Phase 1a: initiate Quick Connect # ------------------------------------------------------------------ async def initiate_quick_connect(self, url: str | None = None) -> QuickConnectResult | None: """ Call Jellyfin's POST /QuickConnect/Initiate. Returns a QuickConnectResult with {secret, code} or None on failure. The *code* is what the user enters on their Jellyfin page. The *secret* is used internally to poll/authenticate. """ base_url = url or await self._resolve_url() if not base_url: logger.error("QuickConnect failed — no JELLYFIN_URL configured.") return None logger.info("Initiating Quick Connect on %s", base_url) async with httpx.AsyncClient(timeout=10) as client: try: resp = await client.post( f"{base_url}/QuickConnect/Initiate", headers=self._qc_headers(), json={}, ) if resp.status_code != 200: logger.warning( "QuickConnect init failed: HTTP %s — %s", resp.status_code, resp.text[:200], ) return None data = resp.json() secret = data.get("Secret", "") code = data.get("Code", "") device_id = data.get("DeviceId", "") device_name = data.get("DeviceName", "") if not secret or not code: logger.warning("QuickConnect init returned unexpected payload: %s", data) return None logger.info( "Quick Connect initiated: code=%s device=%s", code, device_name, ) return QuickConnectResult( secret=secret, code=code, device_id=device_id, device_name=device_name, ) except httpx.TimeoutException: logger.error("QuickConnect init timed out reaching %s", base_url) return None except httpx.ConnectError: logger.error("QuickConnect init — cannot connect to %s", base_url) return None except Exception: logger.exception("Unexpected error during QuickConnect init") return None # ------------------------------------------------------------------ # Phase 1b: poll Quick Connect status # ------------------------------------------------------------------ async def poll_quick_connect(self, secret: str, url: str | None = None) -> str: """ Call Jellyfin's GET /QuickConnect/Connect?secret=. Returns one of: - "Active" → user hasn't entered the code yet - "Authorized" → user entered code AND approved - "Expired" → code expired / unknown secret - "Error" → network or unexpected failure """ base_url = url or await self._resolve_url() if not base_url: logger.error("QuickConnect poll failed — no JELLYFIN_URL configured.") return "Error" async with httpx.AsyncClient(timeout=10) as client: try: resp = await client.get( f"{base_url}/QuickConnect/Connect", params={"secret": secret}, headers=self._qc_headers(), ) if resp.status_code == 404: return "Expired" if resp.status_code == 200: data = resp.json() # Jellyfin returns "Authenticated" (not "Authorized") if data.get("Authenticated") is True: return "Authorized" # "Authenticated" is false, missing, or null → still active return "Active" logger.warning( "QuickConnect poll unexpected: HTTP %s — %s", resp.status_code, resp.text[:200], ) return "Error" except (httpx.TimeoutException, httpx.ConnectError): logger.warning("QuickConnect poll network error") return "Error" except Exception: logger.exception("Unexpected error during QuickConnect poll") return "Error" # ------------------------------------------------------------------ # Phase 1c: exchange secret for token # ------------------------------------------------------------------ async def authenticate_quick_connect( self, secret: str, url: str | None = None ) -> AuthResult: """ After poll_quick_connect returns "Authorized", call POST /Users/AuthenticateWithQuickConnect to exchange the secret for a real access token. Returns AuthResult with token, user_id, username on success. """ base_url = url or await self._resolve_url() if not base_url: return AuthResult( success=False, error_message="No Jellyfin server URL configured.", ) logger.info("Exchanging QuickConnect secret for token on %s", base_url) async with httpx.AsyncClient(timeout=10) as client: try: resp = await client.post( f"{base_url}/Users/AuthenticateWithQuickConnect", json={"Secret": secret}, headers=self._qc_headers(), ) if resp.status_code != 200: logger.warning( "QuickConnect auth exchange failed: HTTP %s", resp.status_code, ) return AuthResult( success=False, error_message="Quick Connect authentication failed. The code may have expired.", ) data = resp.json() user = data.get("User", {}) token = data.get("AccessToken", "") if not token: return AuthResult( success=False, error_message="Jellyfin returned an unexpected response.", ) logger.info( "QuickConnect linked: user=%s (%s)", user.get("Name", "?"), user.get("Id", "?"), ) return AuthResult( success=True, external_user_id=user.get("Id", ""), external_name=user.get("Name", "?"), credentials={ "token": token, "url": base_url, "user_id": user.get("Id", ""), }, ) except httpx.TimeoutException: return AuthResult( success=False, error_message=f"Could not reach {base_url} — connection timed out.", ) except httpx.ConnectError: return AuthResult( success=False, error_message=f"Could not connect to {base_url}. Is the server running?", ) except Exception: logger.exception("Unexpected error during QuickConnect auth exchange") return AuthResult( success=False, error_message="An unexpected error occurred during authentication.", ) # ------------------------------------------------------------------ # Login form (legacy — used by the REST API) # ------------------------------------------------------------------ def render_login_form(self, token: str, discord_id: int) -> str: return f""" Link Jellyfin

🔗 Link Jellyfin to Discord

Enter your Jellyfin server URL and credentials to link your account.

""" # ------------------------------------------------------------------ # Authentication # ------------------------------------------------------------------ async def authenticate(self, form_data: dict) -> AuthResult: url = form_data.get("jellyfin_url", "").strip().rstrip("/") username = form_data.get("username", "").strip() password = form_data.get("password", "").strip() if not url or not username or not password: return AuthResult( success=False, error_message="All fields are required (URL, username, password).", ) logger.info("Attempting Jellyfin login for '%s' on %s", username, url) async with httpx.AsyncClient(timeout=10) as client: try: resp = await client.post( f"{url}/Users/AuthenticateByName", json={"Username": username, "Pw": password}, headers={"X-Emby-Authorization": _EMBY_HEADER}, ) if resp.status_code != 200: logger.warning( "Jellyfin login failed for '%s': HTTP %s", username, resp.status_code ) return AuthResult( success=False, error_message=f"Login failed — check your server URL and credentials.", ) data = resp.json() user = data.get("User", {}) token = data.get("AccessToken", "") if not token: return AuthResult( success=False, error_message="Jellyfin returned an unexpected response.", ) logger.info( "Jellyfin login OK: user=%s (%s)", user.get("Name", "?"), user.get("Id", "?"), ) return AuthResult( success=True, external_user_id=user.get("Id", ""), external_name=user.get("Name", username), credentials={ "token": token, "url": url, "user_id": user.get("Id", ""), }, ) except httpx.TimeoutException: return AuthResult( success=False, error_message=f"Could not reach {url} — connection timed out. Check the URL.", ) except httpx.ConnectError: return AuthResult( success=False, error_message=f"Could not connect to {url}. Is the server running?", ) except Exception as exc: logger.exception("Unexpected error during Jellyfin login") return AuthResult( success=False, error_message=f"An unexpected error occurred. Please try again.", ) # Self-register at import time register_auth_service(JellyfinAuth())