105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
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
|