111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
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
|