feat: add identification page

app/utils/auth.py

@@ -0,0 +1,104 @@
+import base64
+import hashlib
+import hmac
+import json
+import os
+import time
+from dataclasses import dataclass
+
+from app.utils.logger import logger
+
+
+def _parse_bool(value: str | None, default: bool = False) -> bool:
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "y", "on"}
+
+
+def _b64url_encode(data: bytes) -> str:
+    return base64.urlsafe_b64encode(data).decode("utf-8").rstrip("=")
+
+
+def _b64url_decode(data: str) -> bytes:
+    padding = "=" * (-len(data) % 4)
+    return base64.urlsafe_b64decode((data + padding).encode("utf-8"))
+
+
+@dataclass(frozen=True)
+class AuthConfig:
+    enabled: bool
+    username: str
+    password: str
+    token_secret: bytes
+    token_ttl_seconds: int
+
+
+def load_auth_config() -> AuthConfig:
+    enabled = _parse_bool(os.environ.get("PLEXPLAYLISTSYNC_AUTH_ENABLED"), default=False)
+    username = os.environ.get("PLEXPLAYLISTSYNC_AUTH_USERNAME", "").strip()
+    password = os.environ.get("PLEXPLAYLISTSYNC_AUTH_PASSWORD", "")
+    ttl = int(os.environ.get("PLEXPLAYLISTSYNC_AUTH_TOKEN_TTL_SECONDS", "86400"))
+    ttl = max(60, ttl)
+
+    secret_env = os.environ.get("PLEXPLAYLISTSYNC_AUTH_TOKEN_SECRET", "").strip()
+    if secret_env:
+        token_secret = secret_env.encode("utf-8")
+    elif enabled:
+        # If auth is enabled but no explicit secret is set, fall back to an ephemeral secret.
+        # This means tokens become invalid after restart, which is acceptable for this project.
+        token_secret = os.urandom(32)
+        logger.warning(
+            "PLEXPLAYLISTSYNC_AUTH_TOKEN_SECRET not set; using ephemeral secret (tokens reset on restart)."
+        )
+    else:
+        token_secret = b""
+
+    if enabled:
+        if not username or not password:
+            raise RuntimeError(
+                "Auth enabled but missing credentials: please set PLEXPLAYLISTSYNC_AUTH_USERNAME and PLEXPLAYLISTSYNC_AUTH_PASSWORD."
+            )
+
+    return AuthConfig(
+        enabled=enabled,
+        username=username,
+        password=password,
+        token_secret=token_secret,
+        token_ttl_seconds=ttl,
+    )
+
+
+def issue_token(config: AuthConfig, username: str) -> str:
+    now = int(time.time())
+    payload = {"u": username, "exp": now + config.token_ttl_seconds}
+    payload_bytes = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+    payload_b64 = _b64url_encode(payload_bytes)
+
+    sig = hmac.new(config.token_secret, payload_b64.encode("utf-8"), hashlib.sha256).digest()
+    sig_b64 = _b64url_encode(sig)
+    return f"{payload_b64}.{sig_b64}"
+
+
+def verify_token(config: AuthConfig, token: str) -> dict | None:
+    try:
+        payload_b64, sig_b64 = token.split(".", 1)
+    except ValueError:
+        return None
+
+    try:
+        expected_sig = hmac.new(
+            config.token_secret, payload_b64.encode("utf-8"), hashlib.sha256
+        ).digest()
+        actual_sig = _b64url_decode(sig_b64)
+        if not hmac.compare_digest(expected_sig, actual_sig):
+            return None
+
+        payload = json.loads(_b64url_decode(payload_b64).decode("utf-8"))
+        exp = int(payload.get("exp", 0))
+        if exp <= int(time.time()):
+            return None
+        if not payload.get("u"):
+            return None
+        return payload
+    except Exception:
+        return None