import hmac import hashlib import time import json import base64 import urllib.request import urllib.error from typing import Optional, Tuple from fastapi import Request, HTTPException, status from fastapi.responses import RedirectResponse from app.config import settings COOKIE_NAME = "sync_session" SESSION_DURATION_DAYS = 30 def verify_credentials(username: str, password: str) -> bool: """Verify credentials against Navidrome (/auth/login) or fallback admin password.""" if not username or not password: return False # 1. Check fallback password if configured if settings.ADMIN_PASSWORD and password == settings.ADMIN_PASSWORD: return True # 2. Check Navidrome native /auth/login endpoint try: url = f"{settings.NAVIDROME_INTERNAL_URL}/auth/login" data = json.dumps({"username": username, "password": password}).encode("utf-8") req = urllib.request.Request( url, data=data, headers={ "Content-Type": "application/json", "User-Agent": "NavidromeMusicSync/1.0" } ) with urllib.request.urlopen(req, timeout=5) as resp: if resp.status == 200: return True except urllib.error.HTTPError as he: # 401 Unauthorized or other auth errors return False except Exception as e: print(f"[Auth] Navidrome auth check failed: {e}") return False def create_session_token(username: str, duration_days: int = SESSION_DURATION_DAYS) -> str: """Generate a signed HMAC token containing username and expiry timestamp.""" expiry = int(time.time()) + (duration_days * 86400) payload = f"{username}:{expiry}" sig = hmac.new(settings.SECRET_KEY.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest() return f"{payload}:{sig}" def verify_session_token(token: str) -> Optional[str]: """Verify signed session token. Returns username if valid, None otherwise.""" if not token or ":" not in token: return None parts = token.split(":") if len(parts) != 3: return None username, expiry_str, sig = parts try: expiry = int(expiry_str) if time.time() > expiry: return None payload = f"{username}:{expiry}" expected_sig = hmac.new(settings.SECRET_KEY.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256).hexdigest() if hmac.compare_digest(sig, expected_sig): return username except Exception: return None return None def get_authenticated_user(request: Request) -> Optional[str]: """Extract authenticated username from cookies, Basic Auth, or Bearer header.""" if not settings.AUTH_ENABLED: return "anonymous" # 1. Cookie session token = request.cookies.get(COOKIE_NAME) if token: user = verify_session_token(token) if user: return user # 2. HTTP Basic Auth auth_header = request.headers.get("Authorization") if auth_header and auth_header.startswith("Basic "): try: raw = base64.b64decode(auth_header[6:]).decode("utf-8") if ":" in raw: u, p = raw.split(":", 1) if verify_credentials(u, p): return u except Exception: pass return None def require_auth(request: Request) -> str: """FastAPI dependency to require authentication on protected endpoints.""" user = get_authenticated_user(request) if not user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Authentication required. Please log in.", headers={"WWW-Authenticate": "Basic realm=\"Navidrome Music Sync\""} ) return user