`year` was appended to the album directory name AFTER safe() had run, so it
was the one path component reaching the filesystem unchecked:
POST /api/request {"artist":"E","title":"p","year":"../../../../tmp/pwned"}
-> library/E/Singles (../../../../tmp/pwned)/p.flac
-> normalises to ../tmp/pwned)/p.flac
Any authenticated user could do it, guests included, and it wrote a real
audio file to an arbitrary path the process could reach. Existing files were
not overwritten -- tag_and_file returns early when the destination exists --
so this created new files and directories rather than clobbering.
Fixed in two layers, because the first is a property of five call sites that
must each stay correct and the second cannot be forgotten:
1. safe_year() whitelists \d{4}, applied in enqueue_job (the choke point for
all three request paths) and again in tag_and_file.
2. tag_and_file resolves the destination and refuses anything that is not
under the library root, before mkdir rather than after.
Also from the same pass:
- Cap import at 500 pasted lines / 1000 expanded tracks. URL expansion runs
BEFORE the quota check -- it has to, since a playlist URL is one line and
200 tracks -- so without a cap a paste of 10,000 album URLs was 10,000
requests to Deezer from this IP before any quota was consulted.
- Clamp search limit (1-50), jobs limit (1-1000) and activity days (1-365).
search limit fanned out one /album/{id} enrichment per distinct album.
- safe_cover() requires https and <=300 chars. img-src permits data:, so
the field was free storage for megabytes of base64 per job.
- Truncate filename components to 150 chars; a long title used to fail the
import with ENAMETOOLONG after the download had already completed.
- Treat Deezer's 403 as no-results. Its WAF rejects injection-looking
queries, which surfaced to the user as a 502 "service broken".
Add tests/: 42 security assertions and 35 authorization assertions covering
the role matrix, last-admin protection, mid-session revocation, job scoping
and OAuth CSRF. All 77 pass.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1896 lines
80 KiB
Python
1896 lines
80 KiB
Python
"""
|
||
muse — request a song, get it in Navidrome.
|
||
|
||
The gap this fills: Lidarr thinks in MusicBrainz *releases*, so the smallest
|
||
thing it can want is an album. Soulseek trades individual files. Nothing in
|
||
between existed, so requesting one song meant either hoovering a whole
|
||
discography through Lidarr or hand-filing downloads out of slskd's UI.
|
||
|
||
The trick that makes this simple: the user PICKS a specific track before
|
||
anything is downloaded. That means we hold ground-truth metadata — artist,
|
||
album, year, track number — before the file exists. Taggers like beets have to
|
||
infer all of it afterwards from whatever a stranger named the file; we never
|
||
guess. We tag what was asked for.
|
||
|
||
Flow:
|
||
/api/search Deezer track search -> the user picks one
|
||
/api/request /track/{id} for exact metadata, then slskd search ->
|
||
score candidates -> enqueue best
|
||
worker poll slskd -> on completion: tag, then hardlink into
|
||
/data/media/music/<Artist>/<Album> (<Year>)/NN - Title.ext
|
||
|
||
Hardlink, not copy: the download stays in slskd's completed dir so it keeps
|
||
being shared, while the library gets its own name for the same bytes. One
|
||
file on disk, two names — the same trick the *arrs use.
|
||
"""
|
||
|
||
import asyncio
|
||
import base64
|
||
import difflib
|
||
import hashlib
|
||
import logging
|
||
import os
|
||
import re
|
||
import secrets
|
||
import sqlite3
|
||
import uuid
|
||
from contextlib import asynccontextmanager
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from urllib.parse import urlencode
|
||
|
||
import httpx
|
||
from fastapi import FastAPI, HTTPException, Query, Request
|
||
from fastapi.responses import (FileResponse, HTMLResponse, JSONResponse,
|
||
RedirectResponse)
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel
|
||
|
||
log = logging.getLogger("muse")
|
||
logging.basicConfig(
|
||
level=os.getenv("MUSE_LOG_LEVEL", "INFO"),
|
||
format="[%(levelname)s|%(name)s] %(asctime)s: %(message)s",
|
||
datefmt="%Y-%m-%dT%H:%M:%S%z",
|
||
)
|
||
|
||
SLSKD_URL = os.getenv("SLSKD_URL", "http://192.168.1.87:5030").rstrip("/")
|
||
SLSKD_API_KEY = os.getenv("SLSKD_API_KEY", "")
|
||
DOWNLOAD_DIR = Path(os.getenv("MUSE_DOWNLOAD_DIR", "/data/torrents/soulseek/complete"))
|
||
LIBRARY_DIR = Path(os.getenv("MUSE_LIBRARY_DIR", "/data/media/music"))
|
||
DB_PATH = Path(os.getenv("MUSE_DB", "/config/muse.db"))
|
||
|
||
# Ordered best-first. A candidate's extension must appear here or it is
|
||
# rejected outright — this is what keeps .m4a rips and stray .cue files out.
|
||
FORMATS = [f.strip() for f in os.getenv("MUSE_FORMATS", "flac,mp3").split(",") if f.strip()]
|
||
|
||
# Below this, the remote filename doesn't plausibly contain the track title.
|
||
# Soulseek filenames are chaotic ("08-radiohead-creep_(accoustic)-ent.mp3"),
|
||
# so this compares against a normalised form, not the raw string.
|
||
MIN_MATCH = float(os.getenv("MUSE_MIN_MATCH", "0.45"))
|
||
|
||
# How many Soulseek searches may run at once.
|
||
#
|
||
# This is a hard cap, not a suggestion. Each search takes 10-40s of polling
|
||
# and real load on slskd and the Soulseek server, so firing one task per
|
||
# request — which is what an earlier version did — means pasting a 50-track
|
||
# list launches 50 concurrent searches and drowns the client. Everything goes
|
||
# through a bounded queue instead: requests return instantly, work drains at a
|
||
# civil rate.
|
||
WORKERS = int(os.getenv("MUSE_WORKERS", "3"))
|
||
|
||
# Search uses Deezer's public API: no key, no OAuth, and vastly better
|
||
# relevance than MusicBrainz for the way people actually type.
|
||
#
|
||
# MusicBrainz was tried first and rejected on measurements, not taste. It is a
|
||
# superb database but a poor search engine: "radiohead creep" returned eight
|
||
# covers and mashups before the Radiohead recording, and "bohemian rhapsody"
|
||
# never surfaced Queen at all across 90 results — because its text index
|
||
# matches "radiohead" inside a cover's *title* as readily as in the artist
|
||
# field. Client-side re-ranking can only reorder what came back, so it could
|
||
# not fix a result set that lacked the right row. Deezer returns
|
||
# "Creep / Radiohead / Pablo Honey" first for the same query.
|
||
#
|
||
# Deezer search omits track number and release date, so those are fetched from
|
||
# /track/{id} at request time — one extra call per request, not per result.
|
||
DEEZER = "https://api.deezer.com"
|
||
|
||
# Derivative versions to push down when the user did not ask for them.
|
||
# Matched on WORD BOUNDARIES, not as substrings: a bare "live" is needed to
|
||
# catch "Come As You Are (Live)", but substring matching would also penalise
|
||
# "Deliver Me" and "Olive". Same trap for "demo" vs "Demolition".
|
||
#
|
||
# A term only counts against a result when the user did NOT type it, so asking
|
||
# for "oasis live forever" or "nirvana unplugged demo" still works.
|
||
DERIVATIVE = ("cover", "covers", "karaoke", "tribute", "remix", "remixed",
|
||
"instrumental", "acapella", "originally performed",
|
||
"made popular by", "in the style of", "demo", "rehearsal",
|
||
"live", "unplugged", "re recorded", "rerecorded",
|
||
"sped up", "slowed", "reverb", "8d", "nightcore")
|
||
DERIVATIVE_RE = re.compile(r"\b(" + "|".join(DERIVATIVE) + r")\b")
|
||
|
||
# ─────────────────────────── auth ───────────────────────────
|
||
#
|
||
# OAuth against GitHub and/or Google, built so this can face the internet.
|
||
#
|
||
# FAILS CLOSED. Auth is on unless MUSE_AUTH is explicitly set to "off", and if
|
||
# it is on with incomplete config the app refuses to start. The failure mode
|
||
# worth engineering against is not a broken login page — it is a misconfigured
|
||
# deploy quietly serving an unauthenticated download client to the internet.
|
||
# You should have to *mean* it to run without auth.
|
||
#
|
||
# Sessions are server-side: the cookie carries a 256-bit random token and
|
||
# nothing else, and the database stores only its SHA-256. There is no signed
|
||
# payload to get wrong, a stolen database yields no usable cookies, and logout
|
||
# and revocation actually revoke.
|
||
#
|
||
# WHO MAY LOG IN is two rules, checked in this order:
|
||
# 1. MUSE_ALLOWED_USERS — the bootstrap allowlist. Always admitted. This is
|
||
# how the very first admin gets in when the users table is empty, and it
|
||
# is the recovery path if someone disables themselves.
|
||
# 2. A row in `users` that is not disabled — how everyone invited later gets
|
||
# in, without an env change or a redeploy.
|
||
# Anyone matching neither is rejected and the attempt is logged.
|
||
AUTH_MODE = os.getenv("MUSE_AUTH", "github").strip().lower()
|
||
ENABLED_PROVIDERS = [p for p in (x.strip() for x in AUTH_MODE.split(","))
|
||
if p and p != "off"]
|
||
|
||
# Absolute external URL, e.g. https://muse.aryan-singh.dev. Used to build the
|
||
# OAuth callback. Taken from config rather than from X-Forwarded-* headers on
|
||
# purpose: those are attacker-controllable and this is a redirect target.
|
||
BASE_URL = os.getenv("MUSE_BASE_URL", "").rstrip("/")
|
||
ALLOWED_USERS = {u.strip().lower()
|
||
for u in os.getenv("MUSE_ALLOWED_USERS", "").split(",") if u.strip()}
|
||
COOKIE_SECURE = os.getenv("MUSE_COOKIE_SECURE", "true").lower() != "false"
|
||
SESSION_DAYS = int(os.getenv("MUSE_SESSION_DAYS", "7"))
|
||
|
||
SESSION_COOKIE = "muse_session"
|
||
STATE_COOKIE = "muse_oauth_state"
|
||
|
||
# One callback URL for every provider — the provider is carried in the state
|
||
# COOKIE, not the query string, so adding Google needs no change to the GitHub
|
||
# OAuth app that is already registered against /auth/callback.
|
||
PROVIDERS: dict[str, dict] = {
|
||
"github": {
|
||
"label": "GitHub",
|
||
"authorize": "https://github.com/login/oauth/authorize",
|
||
"token": "https://github.com/login/oauth/access_token",
|
||
"scope": "read:user",
|
||
"client_id": os.getenv("MUSE_GITHUB_CLIENT_ID", "").strip(),
|
||
"client_secret": os.getenv("MUSE_GITHUB_CLIENT_SECRET", "").strip(),
|
||
"id_env": "MUSE_GITHUB_CLIENT_ID",
|
||
"secret_env": "MUSE_GITHUB_CLIENT_SECRET",
|
||
},
|
||
"google": {
|
||
"label": "Google",
|
||
"authorize": "https://accounts.google.com/o/oauth2/v2/auth",
|
||
"token": "https://oauth2.googleapis.com/token",
|
||
"scope": "openid email profile",
|
||
"client_id": os.getenv("MUSE_GOOGLE_CLIENT_ID", "").strip(),
|
||
"client_secret": os.getenv("MUSE_GOOGLE_CLIENT_SECRET", "").strip(),
|
||
"id_env": "MUSE_GOOGLE_CLIENT_ID",
|
||
"secret_env": "MUSE_GOOGLE_CLIENT_SECRET",
|
||
},
|
||
}
|
||
|
||
GH_API = "https://api.github.com/user"
|
||
GOOGLE_API = "https://openidconnect.googleapis.com/v1/userinfo"
|
||
|
||
# Paths reachable without a session. Deliberately tiny: /healthz reports a bare
|
||
# ok so the container healthcheck works without leaking configuration.
|
||
PUBLIC_PATHS = {"/healthz", "/auth/login", "/auth/callback", "/auth/logout"}
|
||
|
||
ROLES = ("admin", "user", "guest")
|
||
|
||
|
||
def auth_preflight():
|
||
if AUTH_MODE == "off":
|
||
log.warning("=" * 62)
|
||
log.warning("MUSE_AUTH=off — every endpoint is UNAUTHENTICATED.")
|
||
log.warning("Never expose this to the internet in this state.")
|
||
log.warning("=" * 62)
|
||
return
|
||
if not ENABLED_PROVIDERS:
|
||
raise RuntimeError(f"MUSE_AUTH must list providers or be 'off', got {AUTH_MODE!r}")
|
||
|
||
unknown = [p for p in ENABLED_PROVIDERS if p not in PROVIDERS]
|
||
if unknown:
|
||
raise RuntimeError(
|
||
f"MUSE_AUTH names unknown provider(s) {unknown}. "
|
||
f"Valid: {', '.join(PROVIDERS)}, or 'off'.")
|
||
|
||
missing: list[str] = []
|
||
for p in ENABLED_PROVIDERS:
|
||
cfg = PROVIDERS[p]
|
||
if not cfg["client_id"]:
|
||
missing.append(cfg["id_env"])
|
||
if not cfg["client_secret"]:
|
||
missing.append(cfg["secret_env"])
|
||
if not BASE_URL:
|
||
missing.append("MUSE_BASE_URL")
|
||
# The bootstrap allowlist is only mandatory while nobody exists yet —
|
||
# otherwise a working multi-user install would be held hostage by an env
|
||
# var it has outgrown. But an EMPTY database and no allowlist means nobody
|
||
# can ever log in, which is a misconfiguration worth refusing to start on.
|
||
if not ALLOWED_USERS and not _any_users():
|
||
missing.append("MUSE_ALLOWED_USERS")
|
||
if missing:
|
||
raise RuntimeError(
|
||
"Auth is enabled but " + ", ".join(sorted(set(missing))) + " not set. "
|
||
"Refusing to start: set them, or set MUSE_AUTH=off to run without "
|
||
"authentication (LAN only).")
|
||
|
||
if not BASE_URL.startswith("https://") and COOKIE_SECURE:
|
||
log.warning("MUSE_BASE_URL is not https but cookies are Secure — the "
|
||
"browser will refuse to send them. Set MUSE_COOKIE_SECURE=false "
|
||
"only if you are testing over plain http on a LAN.")
|
||
log.info("auth: providers=%s bootstrap_allowlist=%d",
|
||
",".join(ENABLED_PROVIDERS), len(ALLOWED_USERS))
|
||
|
||
|
||
def _any_users() -> bool:
|
||
try:
|
||
with db() as c:
|
||
return c.execute("SELECT 1 FROM users LIMIT 1").fetchone() is not None
|
||
except sqlite3.Error:
|
||
return False
|
||
|
||
|
||
def _hash(tok: str) -> str:
|
||
return hashlib.sha256(tok.encode()).hexdigest()
|
||
|
||
|
||
def new_session(username: str) -> str:
|
||
token = secrets.token_urlsafe(32)
|
||
exp = datetime.now(timezone.utc) + timedelta(days=SESSION_DAYS)
|
||
with db() as c:
|
||
c.execute("INSERT INTO sessions (token_hash,username,expires_at) VALUES (?,?,?)",
|
||
(_hash(token), username, exp.isoformat()))
|
||
# Opportunistic sweep; no scheduler needed for a handful of rows.
|
||
c.execute("DELETE FROM sessions WHERE expires_at < ?",
|
||
(datetime.now(timezone.utc).isoformat(),))
|
||
return token
|
||
|
||
|
||
def may_login(username: str, row: sqlite3.Row | None) -> bool:
|
||
"""The single authorization rule, used at login AND on every request.
|
||
|
||
Bootstrap allowlist first, then a live, non-disabled user row. Deliberately
|
||
one function: an authorization check that runs only at login is not a
|
||
revocation mechanism, it is a delay — disabling someone would leave their
|
||
session valid for up to MUSE_SESSION_DAYS.
|
||
"""
|
||
if username.lower() in ALLOWED_USERS:
|
||
return True
|
||
return bool(row) and not row["disabled"]
|
||
|
||
|
||
def current_user(request: Request) -> sqlite3.Row | None:
|
||
"""The signed-in user's row, or None. Re-validated on every request."""
|
||
tok = request.cookies.get(SESSION_COOKIE)
|
||
if not tok:
|
||
return None
|
||
with db() as c:
|
||
row = c.execute(
|
||
"SELECT username, expires_at FROM sessions WHERE token_hash=?",
|
||
(_hash(tok),)).fetchone()
|
||
if not row:
|
||
return None
|
||
try:
|
||
if datetime.fromisoformat(row["expires_at"]) < datetime.now(timezone.utc):
|
||
return None
|
||
except ValueError:
|
||
return None
|
||
u = c.execute("SELECT * FROM users WHERE username=?",
|
||
(row["username"],)).fetchone()
|
||
if not may_login(row["username"], u):
|
||
return None
|
||
if u is None:
|
||
# Allowlisted but no row — an admin deleted the bootstrap account while
|
||
# its session was still live. The allowlist is exactly the recovery
|
||
# path for that, so put the row back rather than half-authenticating
|
||
# them into a state where every role check fails.
|
||
u = upsert_user(row["username"])
|
||
log.info("auth: restored deleted allowlisted user %s", row["username"])
|
||
return u
|
||
|
||
|
||
def session_user(request: Request) -> str | None:
|
||
u = current_user(request)
|
||
return u["username"] if u else None
|
||
|
||
|
||
def user_role(request: Request) -> str:
|
||
u = current_user(request)
|
||
if u:
|
||
return u["role"]
|
||
# With auth off there is nobody to be, so everyone is an admin — otherwise
|
||
# the admin panel would be unreachable on a LAN-only install.
|
||
return "admin" if AUTH_MODE == "off" else "guest"
|
||
|
||
|
||
def require_admin(request: Request) -> str:
|
||
if AUTH_MODE == "off":
|
||
return "local"
|
||
u = current_user(request)
|
||
if not u:
|
||
raise HTTPException(401, "unauthenticated")
|
||
if u["role"] != "admin":
|
||
log.warning("admin: DENIED %s (role=%s) -> %s",
|
||
u["username"], u["role"], request.url.path)
|
||
raise HTTPException(403, "admin only")
|
||
return u["username"]
|
||
|
||
|
||
SAFE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||
NOISE = re.compile(r"[^a-z0-9]+")
|
||
YEAR = re.compile(r"^\d{4}$")
|
||
|
||
# Most filesystems cap a path component at 255 bytes. Track titles can be far
|
||
# longer than that once a stranger has appended their own notes, and the
|
||
# resulting ENAMETOOLONG would fail the import after the download had finished.
|
||
MAX_COMPONENT = 150
|
||
|
||
|
||
def safe(name: str) -> str:
|
||
"""Filesystem-safe path component. Trailing dots/spaces break on some FS."""
|
||
return SAFE.sub("_", name).strip(". ")[:MAX_COMPONENT].strip() or "Unknown"
|
||
|
||
|
||
def safe_cover(url) -> str | None:
|
||
"""Artwork URL, stored and later rendered as an <img src>.
|
||
|
||
The CSP already stops a foreign host loading, so this is not an XSS gate —
|
||
it stops the field being used as free storage. `data:` is permitted by
|
||
img-src, so without a check any client could park megabytes of base64 in
|
||
the jobs table on every request.
|
||
"""
|
||
u = str(url or "").strip()
|
||
return u if u.startswith("https://") and len(u) <= 300 else None
|
||
|
||
|
||
def safe_year(year) -> str | None:
|
||
"""A year, or nothing. This lands in a directory name, so it is checked
|
||
against a whitelist rather than merely scrubbed."""
|
||
y = str(year or "").strip()
|
||
return y if YEAR.match(y) else None
|
||
|
||
|
||
def norm(s: str) -> str:
|
||
return NOISE.sub(" ", s.lower()).strip()
|
||
|
||
|
||
# ─────────────────────────── storage ───────────────────────────
|
||
|
||
def db():
|
||
c = sqlite3.connect(DB_PATH, timeout=30)
|
||
c.row_factory = sqlite3.Row
|
||
return c
|
||
|
||
|
||
def init_db():
|
||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||
with db() as c:
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS jobs (
|
||
id TEXT PRIMARY KEY,
|
||
state TEXT NOT NULL, -- searching|queued|downloading|imported|failed
|
||
detail TEXT,
|
||
artist TEXT, title TEXT, album TEXT,
|
||
year TEXT, track INTEGER,
|
||
recording_mbid TEXT, release_mbid TEXT,
|
||
username TEXT, remote_file TEXT, size INTEGER,
|
||
imported_path TEXT,
|
||
created_at TEXT DEFAULT (datetime('now'))
|
||
)
|
||
""")
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS sessions (
|
||
token_hash TEXT PRIMARY KEY, -- sha256 of the cookie value
|
||
username TEXT NOT NULL,
|
||
created_at TEXT DEFAULT (datetime('now')),
|
||
expires_at TEXT NOT NULL
|
||
)
|
||
""")
|
||
# Users exist as rows even while the allowlist is a single name in an
|
||
# env var. Roles and quotas are read from here from the start, so
|
||
# opening this up to other people later is an INSERT and a config
|
||
# change rather than a schema migration against live job history.
|
||
c.execute("""
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
username TEXT PRIMARY KEY COLLATE NOCASE,
|
||
provider TEXT NOT NULL DEFAULT 'github',
|
||
role TEXT NOT NULL DEFAULT 'user', -- admin|user|guest
|
||
display TEXT,
|
||
avatar TEXT,
|
||
quota INTEGER NOT NULL DEFAULT 0, -- requests/month, 0 = unlimited
|
||
disabled INTEGER NOT NULL DEFAULT 0,
|
||
created_at TEXT DEFAULT (datetime('now')),
|
||
last_seen TEXT
|
||
)
|
||
""")
|
||
# Added after the first release, so existing databases need it grafted
|
||
# on rather than recreated -- job history is the whole point of the file.
|
||
cols = {r["name"] for r in c.execute("PRAGMA table_info(jobs)").fetchall()}
|
||
for col, decl in (("requested_by", "TEXT"),
|
||
("cover", "TEXT"),
|
||
("genre", "TEXT"),
|
||
("duration", "INTEGER")):
|
||
if col not in cols:
|
||
c.execute(f"ALTER TABLE jobs ADD COLUMN {col} {decl}")
|
||
ucols = {r["name"] for r in c.execute("PRAGMA table_info(users)").fetchall()}
|
||
for col, decl in (("storage_quota", "INTEGER NOT NULL DEFAULT 0"),
|
||
("invited_by", "TEXT")):
|
||
if col not in ucols:
|
||
c.execute(f"ALTER TABLE users ADD COLUMN {col} {decl}")
|
||
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_user ON jobs (requested_by)")
|
||
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_state ON jobs (state)")
|
||
c.execute("CREATE INDEX IF NOT EXISTS jobs_by_created ON jobs (created_at)")
|
||
|
||
|
||
def upsert_user(username: str, avatar: str = "", display: str = "",
|
||
provider: str = "github") -> sqlite3.Row:
|
||
"""Record the sign-in.
|
||
|
||
The first user to ever log in becomes admin — no bootstrap ceremony, no
|
||
default password to forget. Everyone after that is a plain user until an
|
||
admin says otherwise; role is never touched on subsequent logins, or every
|
||
sign-in would silently undo a demotion.
|
||
"""
|
||
with db() as c:
|
||
first = c.execute("SELECT COUNT(*) n FROM users").fetchone()["n"] == 0
|
||
c.execute(
|
||
"INSERT INTO users (username,provider,role,display,avatar,last_seen) "
|
||
"VALUES (?,?,?,?,?,datetime('now')) "
|
||
"ON CONFLICT(username) DO UPDATE SET "
|
||
" last_seen=datetime('now'), provider=excluded.provider, "
|
||
" avatar=COALESCE(NULLIF(excluded.avatar,''), users.avatar), "
|
||
" display=COALESCE(NULLIF(excluded.display,''), users.display)",
|
||
(username, provider, "admin" if first else "user", display, avatar))
|
||
return c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
|
||
|
||
def set_state(job_id: str, state: str, detail: str = "", **cols):
|
||
sets = ", ".join(f"{k}=?" for k in cols)
|
||
sql = f"UPDATE jobs SET state=?, detail=?{', ' + sets if sets else ''} WHERE id=?"
|
||
with db() as c:
|
||
c.execute(sql, [state, detail, *cols.values(), job_id])
|
||
log.info("job %s -> %s %s", job_id[:8], state, detail)
|
||
|
||
|
||
# ─────────────────────────── slskd ───────────────────────────
|
||
|
||
def slskd_client(timeout: float = 30.0) -> httpx.AsyncClient:
|
||
"""30s suits searches and transfer polling, which are legitimately slow.
|
||
|
||
Status probes pass something much shorter: they run on a UI poll, and a
|
||
probe that outlasts its own poll interval stacks requests and leaves the
|
||
admin panel on skeletons for the whole timeout — precisely when slskd is
|
||
down and you opened the panel to find out why.
|
||
"""
|
||
return httpx.AsyncClient(
|
||
base_url=SLSKD_URL,
|
||
headers={"X-API-Key": SLSKD_API_KEY},
|
||
timeout=httpx.Timeout(timeout),
|
||
)
|
||
|
||
|
||
# slskd is on the LAN, so a reachable instance answers in milliseconds. Long
|
||
# enough to absorb a slow moment, short enough to stay under the 5s UI poll.
|
||
PROBE_TIMEOUT = 4.0
|
||
|
||
|
||
def errstr(e: Exception) -> str:
|
||
"""httpx's timeout and connection errors often str() to an empty string,
|
||
which would render as 'slskd unreachable — ' with nothing after it."""
|
||
return (str(e) or type(e).__name__)[:120]
|
||
|
||
|
||
def ext_of(remote_path: str) -> str:
|
||
# Soulseek paths are Windows-style; the API's `extension` field comes back
|
||
# empty in practice, so derive it ourselves.
|
||
return remote_path.rsplit(".", 1)[-1].lower() if "." in remote_path else ""
|
||
|
||
|
||
def score(candidate: dict, peer: dict, want: str) -> float | None:
|
||
"""Rank one remote file. None = reject.
|
||
|
||
Ordering intent: correct format first, then actually-gettable (a peer with
|
||
a free slot and a short queue beats a marginally better file behind 200
|
||
people), then filename plausibility, then bitrate.
|
||
"""
|
||
ext = ext_of(candidate["filename"])
|
||
if ext not in FORMATS:
|
||
return None
|
||
|
||
base = candidate["filename"].replace("\\", "/").rsplit("/", 1)[-1]
|
||
ratio = difflib.SequenceMatcher(None, norm(base), want).ratio()
|
||
# A short title inside a long scene filename scores badly on raw ratio, so
|
||
# give credit for simple containment too.
|
||
if want in norm(base):
|
||
ratio = max(ratio, 0.9)
|
||
if ratio < MIN_MATCH:
|
||
return None
|
||
|
||
s = 0.0
|
||
s += (len(FORMATS) - FORMATS.index(ext)) * 100 # format preference
|
||
s += 60 if peer.get("hasFreeUploadSlot") else 0
|
||
s += max(0, 40 - peer.get("queueLength", 0)) # short queues
|
||
s += min(peer.get("uploadSpeed", 0) / 50_000, 20) # capped speed bonus
|
||
s += ratio * 50
|
||
s += min(candidate.get("bitRate") or 0, 320) / 32 # 0-10
|
||
return s
|
||
|
||
|
||
async def pick_and_enqueue(job_id: str, artist: str, title: str):
|
||
"""Search slskd for one track and queue the best candidate."""
|
||
want = norm(f"{artist} {title}")
|
||
search_id = str(uuid.uuid4())
|
||
async with slskd_client() as sc:
|
||
try:
|
||
r = await sc.post("/api/v0/searches",
|
||
json={"id": search_id, "searchText": f"{artist} {title}"})
|
||
r.raise_for_status()
|
||
except Exception as e:
|
||
set_state(job_id, "failed", f"slskd search failed: {e}")
|
||
return
|
||
|
||
# Poll until slskd calls it done, or we run out of patience. Soulseek
|
||
# searches trickle in; results at t=5s are much worse than at t=15s.
|
||
results = {}
|
||
for _ in range(20):
|
||
await asyncio.sleep(2)
|
||
try:
|
||
rr = await sc.get(f"/api/v0/searches/{search_id}",
|
||
params={"includeResponses": "true"})
|
||
rr.raise_for_status()
|
||
results = rr.json()
|
||
except Exception:
|
||
continue
|
||
if str(results.get("state", "")).startswith("Completed"):
|
||
break
|
||
|
||
best, best_score = None, -1.0
|
||
for peer in results.get("responses", []):
|
||
for f in peer.get("files", []):
|
||
if f.get("isLocked"):
|
||
continue
|
||
sc_ = score(f, peer, want)
|
||
if sc_ is not None and sc_ > best_score:
|
||
best, best_score = (peer, f), sc_
|
||
|
||
# Tidy up so slskd's search history doesn't grow without bound.
|
||
try:
|
||
await sc.delete(f"/api/v0/searches/{search_id}")
|
||
except Exception:
|
||
pass
|
||
|
||
if not best:
|
||
set_state(job_id, "failed",
|
||
f"no {'/'.join(FORMATS)} candidate matched (searched {results.get('fileCount', 0)} files)")
|
||
return
|
||
|
||
peer, f = best
|
||
try:
|
||
er = await sc.post(f"/api/v0/transfers/downloads/{peer['username']}",
|
||
json=[{"filename": f["filename"], "size": f["size"]}])
|
||
er.raise_for_status()
|
||
except Exception as e:
|
||
set_state(job_id, "failed", f"enqueue failed: {e}")
|
||
return
|
||
|
||
set_state(job_id, "downloading",
|
||
f"from {peer['username']} (score {best_score:.0f})",
|
||
username=peer["username"], remote_file=f["filename"], size=f["size"])
|
||
|
||
|
||
# ─────────────────────────── import ───────────────────────────
|
||
|
||
def tag_and_file(job: sqlite3.Row, src: Path) -> Path:
|
||
"""Write known-good tags, then hardlink into the library."""
|
||
from mutagen import File as MFile
|
||
from mutagen.easyid3 import EasyID3
|
||
from mutagen.flac import FLAC
|
||
from mutagen.mp3 import MP3
|
||
|
||
try:
|
||
if src.suffix.lower() == ".flac":
|
||
audio = FLAC(src)
|
||
else:
|
||
try:
|
||
audio = EasyID3(src)
|
||
except Exception:
|
||
m = MP3(src)
|
||
m.add_tags()
|
||
m.save()
|
||
audio = EasyID3(src)
|
||
|
||
audio["title"] = job["title"]
|
||
audio["artist"] = job["artist"]
|
||
audio["albumartist"] = job["artist"]
|
||
if job["album"]:
|
||
audio["album"] = job["album"]
|
||
if job["year"]:
|
||
audio["date"] = str(job["year"])
|
||
if job["track"]:
|
||
audio["tracknumber"] = str(job["track"])
|
||
# Deliberately NOT writing the Deezer id into musicbrainz_trackid --
|
||
# that field means a specific thing and a wrong value there would
|
||
# mislead Picard, beets, or anything else that later reads this file.
|
||
audio.save()
|
||
except Exception as e:
|
||
# A tagging failure must not strand the file — Navidrome can still
|
||
# read whatever tags it shipped with.
|
||
log.warning("tagging failed for %s: %s", src.name, e)
|
||
|
||
album = f"{safe(job['album'] or 'Singles')}"
|
||
year = safe_year(job["year"])
|
||
if year:
|
||
album += f" ({year})"
|
||
dest_dir = LIBRARY_DIR / safe(job["artist"]) / album
|
||
|
||
num = f"{int(job['track']):02d} - " if job["track"] else ""
|
||
dest = dest_dir / f"{num}{safe(job['title'])}{src.suffix.lower()}"
|
||
|
||
# Belt and braces. Every component above is either safe()'d or matched
|
||
# against a whitelist, but that is a property of five call sites that must
|
||
# each stay correct; this is one check that cannot be forgotten. It exists
|
||
# because `year` was for a while appended AFTER safe() ran, so a request
|
||
# with year="../../../../tmp/x" wrote outside the library.
|
||
root = LIBRARY_DIR.resolve()
|
||
if not Path(os.path.normpath(dest)).resolve().is_relative_to(root):
|
||
raise ValueError(f"refusing to write outside the library: {dest}")
|
||
|
||
dest_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
if dest.exists():
|
||
return dest
|
||
try:
|
||
os.link(src, dest) # same filesystem -> free, and keeps sharing
|
||
except OSError:
|
||
import shutil
|
||
shutil.copy2(src, dest) # cross-device fallback
|
||
return dest
|
||
|
||
|
||
def find_downloaded(remote_file: str) -> Path | None:
|
||
"""Locate the completed file. slskd nests it under the remote's parent
|
||
directory name, so match on basename rather than guessing the path."""
|
||
base = remote_file.replace("\\", "/").rsplit("/", 1)[-1]
|
||
for p in DOWNLOAD_DIR.rglob("*"):
|
||
if p.is_file() and p.name == base:
|
||
return p
|
||
return None
|
||
|
||
|
||
# ─────────────────────────── search queue ───────────────────────────
|
||
|
||
work_q: "asyncio.Queue[str]" = asyncio.Queue()
|
||
|
||
|
||
async def search_worker(n: int):
|
||
"""Drain the search queue, one job at a time, WORKERS of these running."""
|
||
while True:
|
||
job_id = await work_q.get()
|
||
try:
|
||
with db() as c:
|
||
job = c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()
|
||
if job and job["state"] in ("queued", "searching"):
|
||
set_state(job_id, "searching", "")
|
||
await pick_and_enqueue(job_id, job["artist"], job["title"])
|
||
except Exception as e:
|
||
log.warning("worker %d: %s", n, e)
|
||
try:
|
||
set_state(job_id, "failed", str(e)[:200])
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
work_q.task_done()
|
||
|
||
|
||
async def import_worker():
|
||
"""Poll slskd for in-flight downloads and import the finished ones."""
|
||
while True:
|
||
await asyncio.sleep(10)
|
||
try:
|
||
with db() as c:
|
||
jobs = c.execute(
|
||
"SELECT * FROM jobs WHERE state='downloading'").fetchall()
|
||
if not jobs:
|
||
continue
|
||
|
||
async with slskd_client() as sc:
|
||
r = await sc.get("/api/v0/transfers/downloads")
|
||
r.raise_for_status()
|
||
payload = r.json()
|
||
|
||
# Flatten slskd's user -> directories -> files nesting.
|
||
states: dict[str, str] = {}
|
||
for user in payload:
|
||
for d in user.get("directories", []):
|
||
for f in d.get("files", []):
|
||
states[f.get("filename", "")] = str(f.get("state", ""))
|
||
|
||
for job in jobs:
|
||
st = states.get(job["remote_file"], "")
|
||
if "Completed, Succeeded" in st:
|
||
src = find_downloaded(job["remote_file"])
|
||
if not src:
|
||
continue # still being moved out of incomplete
|
||
dest = tag_and_file(job, src)
|
||
# Record the real size, not the size the peer advertised —
|
||
# storage quotas are computed from this column.
|
||
try:
|
||
actual = src.stat().st_size
|
||
except OSError:
|
||
actual = job["size"]
|
||
set_state(job["id"], "imported", str(dest),
|
||
imported_path=str(dest), size=actual)
|
||
elif "Completed" in st and "Succeeded" not in st:
|
||
set_state(job["id"], "failed", f"transfer {st}")
|
||
except Exception as e:
|
||
log.warning("worker: %s", e)
|
||
|
||
|
||
# ─────────────────────────── api ───────────────────────────
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(_: FastAPI):
|
||
# init_db first: the preflight consults the users table to decide whether
|
||
# the bootstrap allowlist is still mandatory.
|
||
init_db()
|
||
auth_preflight() # raises and stops the container if misconfigured
|
||
|
||
# The queue lives in memory, so a restart mid-import would otherwise
|
||
# strand every unstarted job in 'queued' forever with nothing to pick it
|
||
# up. Anything not yet handed to slskd goes back on the queue.
|
||
with db() as c:
|
||
orphans = [r["id"] for r in c.execute(
|
||
"SELECT id FROM jobs WHERE state IN ('queued','searching')").fetchall()]
|
||
for jid in orphans:
|
||
work_q.put_nowait(jid)
|
||
if orphans:
|
||
log.info("requeued %d job(s) left over from the last run", len(orphans))
|
||
|
||
tasks = [asyncio.create_task(import_worker())]
|
||
tasks += [asyncio.create_task(search_worker(i)) for i in range(WORKERS)]
|
||
log.info("muse up — slskd=%s library=%s formats=%s workers=%d",
|
||
SLSKD_URL, LIBRARY_DIR, ",".join(FORMATS), WORKERS)
|
||
yield
|
||
for t in tasks:
|
||
t.cancel()
|
||
|
||
|
||
# ─────────────────────── static front end / CSP ───────────────────────
|
||
#
|
||
# The UI is a Next.js static export: `next build` writes plain files into
|
||
# web/out and this process serves them, so the deployment stays one container,
|
||
# one process, one healthcheck, with no Node runtime in the image.
|
||
#
|
||
# That export contains inline <script> blocks (React's hydration payload),
|
||
# which `script-src 'self'` blocks outright. The usual fix is a per-request
|
||
# nonce, but a STATIC export has a better option: the scripts never change
|
||
# between requests, so their SHA-256 can go straight into the policy. Hashing
|
||
# them at startup means the CSP needs no per-request work, no HTML rewriting,
|
||
# and no maintenance when the bundle changes -- rebuild and the hashes follow.
|
||
WEB = Path(__file__).parent / "web" / "out"
|
||
|
||
_INLINE_SCRIPT = re.compile(rb"<script(?![^>]*\bsrc=)[^>]*>(.*?)</script>", re.S)
|
||
|
||
|
||
def script_hashes() -> list[str]:
|
||
out: set[str] = set()
|
||
for page in ("index.html", "404.html"):
|
||
f = WEB / page
|
||
if not f.exists():
|
||
continue
|
||
for body in _INLINE_SCRIPT.findall(f.read_bytes()):
|
||
digest = base64.b64encode(hashlib.sha256(body).digest()).decode()
|
||
out.add(f"'sha256-{digest}'")
|
||
return sorted(out)
|
||
|
||
|
||
SCRIPT_HASHES = " ".join(script_hashes())
|
||
|
||
# Framer Motion writes its pre-hydration `initial` state into style="" on the
|
||
# server-rendered markup, so those attributes have to be allowed or every
|
||
# animated element flashes in at full opacity before hydration corrects it.
|
||
#
|
||
# style-src-attr is the narrow lever for that: it permits style ATTRIBUTES
|
||
# while style-src 'self' still blocks injected <style> blocks and remote
|
||
# stylesheets. It is not a free pass either -- CSS-based exfiltration goes
|
||
# through url(), and img-src below already limits where those can point.
|
||
CSP = (
|
||
"default-src 'self'; "
|
||
f"script-src 'self' {SCRIPT_HASHES}; "
|
||
"style-src 'self'; style-src-attr 'unsafe-inline'; "
|
||
"img-src 'self' data: https://*.dzcdn.net https://*.githubusercontent.com; "
|
||
"media-src 'self' https://*.dzcdn.net; "
|
||
"connect-src 'self'; font-src 'self'; object-src 'none'; "
|
||
"frame-ancestors 'none'; base-uri 'none'; form-action 'self'"
|
||
)
|
||
|
||
|
||
app = FastAPI(
|
||
title="muse",
|
||
lifespan=lifespan,
|
||
# No interactive docs on an internet-facing app: they enumerate every
|
||
# endpoint and schema for anyone who reaches the login page.
|
||
docs_url=None, redoc_url=None, openapi_url=None,
|
||
)
|
||
|
||
|
||
@app.middleware("http")
|
||
async def guard(request: Request, call_next):
|
||
"""Require a session for everything except PUBLIC_PATHS, and stamp
|
||
security headers on every response."""
|
||
# /_next/* is NOT public. Unlike the old hand-written page there is no
|
||
# local login screen to style — an unauthenticated visitor is redirected
|
||
# straight to GitHub and never receives our HTML — so the bundle can sit
|
||
# behind the session too, and not hand out a map of the app to anyone who
|
||
# finds the URL. It loads fine after the callback, cookie in hand.
|
||
path = request.url.path
|
||
if AUTH_MODE == "off" or path in PUBLIC_PATHS:
|
||
resp = await call_next(request)
|
||
elif session_user(request):
|
||
resp = await call_next(request)
|
||
elif path.startswith("/api/"):
|
||
# An API caller gets a status code, not an HTML redirect it can't use.
|
||
resp = JSONResponse({"error": "unauthenticated"}, status_code=401)
|
||
else:
|
||
resp = RedirectResponse("/auth/login", status_code=302)
|
||
|
||
# See CSP above. img-src and media-src name Deezer's CDN explicitly rather
|
||
# than allowing https: wholesale — under a blanket https: any injected
|
||
# markup could beacon to an arbitrary host by requesting an image from it,
|
||
# which is a working exfiltration channel even with connect-src locked to
|
||
# 'self'. Artwork comes from dzcdn.net and avatars from githubusercontent.
|
||
resp.headers["Content-Security-Policy"] = CSP
|
||
resp.headers["X-Content-Type-Options"] = "nosniff"
|
||
resp.headers["X-Frame-Options"] = "DENY"
|
||
resp.headers["Referrer-Policy"] = "no-referrer"
|
||
resp.headers["Permissions-Policy"] = "geolocation=(), microphone=(), camera=()"
|
||
if COOKIE_SECURE:
|
||
resp.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
|
||
return resp
|
||
|
||
|
||
def _set_cookie(resp, name, value, max_age):
|
||
resp.set_cookie(name, value, max_age=max_age, httponly=True,
|
||
secure=COOKIE_SECURE, samesite="lax", path="/")
|
||
|
||
|
||
@app.get("/healthz")
|
||
async def healthz():
|
||
"""Unauthenticated on purpose, for the container healthcheck. Returns
|
||
nothing but a literal ok — no versions, no config, no counts."""
|
||
return {"ok": True}
|
||
|
||
|
||
CHOOSER = """<!doctype html><html lang=en><head><meta charset=utf-8>
|
||
<meta name=viewport content="width=device-width,initial-scale=1"><title>muse</title>
|
||
</head><body style="margin:0;min-height:100vh;display:grid;place-items:center;
|
||
background:#08090d;color:#f2f4f9;font:15px/1.5 -apple-system,BlinkMacSystemFont,
|
||
ui-sans-serif,system-ui,sans-serif">
|
||
<div style="text-align:center;padding:32px">
|
||
<div style="font-size:26px;font-weight:700;letter-spacing:-.5px">muse</div>
|
||
<div style="color:#9aa3b6;font-size:13px;margin:6px 0 26px">Sign in to continue</div>
|
||
{buttons}</div></body></html>"""
|
||
|
||
_BTN = ('<a href="/auth/login?provider={p}" style="display:block;margin:10px auto;'
|
||
'padding:12px 26px;min-width:230px;border-radius:12px;text-decoration:none;'
|
||
'background:rgb(255 255 255/.06);border:1px solid rgb(255 255 255/.14);'
|
||
'color:#f2f4f9;font-weight:600">Continue with {label}</a>')
|
||
|
||
|
||
@app.get("/auth/login")
|
||
async def auth_login(provider: str = ""):
|
||
"""Start OAuth. With several providers enabled and none named, offer a
|
||
choice — a tiny self-contained page that uses only style attributes, which
|
||
the CSP already permits, so it needs no stylesheet and no extra hash."""
|
||
if AUTH_MODE == "off":
|
||
return RedirectResponse("/", status_code=302)
|
||
|
||
if not provider:
|
||
if len(ENABLED_PROVIDERS) == 1:
|
||
provider = ENABLED_PROVIDERS[0]
|
||
else:
|
||
buttons = "".join(
|
||
_BTN.format(p=p, label=PROVIDERS[p]["label"]) for p in ENABLED_PROVIDERS)
|
||
return HTMLResponse(CHOOSER.replace("{buttons}", buttons))
|
||
|
||
if provider not in ENABLED_PROVIDERS:
|
||
raise HTTPException(404, "unknown provider")
|
||
|
||
cfg = PROVIDERS[provider]
|
||
nonce = secrets.token_urlsafe(24)
|
||
params = {
|
||
"client_id": cfg["client_id"],
|
||
"redirect_uri": f"{BASE_URL}/auth/callback",
|
||
"scope": cfg["scope"],
|
||
"state": nonce,
|
||
"response_type": "code",
|
||
}
|
||
if provider == "github":
|
||
params["allow_signup"] = "false"
|
||
if provider == "google":
|
||
# Force the account chooser; otherwise Google silently reuses whichever
|
||
# account the browser last used, which on a shared machine signs you in
|
||
# as someone else with no visible prompt.
|
||
params["prompt"] = "select_account"
|
||
|
||
resp = RedirectResponse(f"{cfg['authorize']}?{urlencode(params)}", status_code=302)
|
||
# CSRF defence: the nonce is echoed back by the provider and must match
|
||
# this cookie, so a callback the user did not initiate is rejected. The
|
||
# provider name rides along in the COOKIE, never the query string, so the
|
||
# callback cannot be steered to a different provider by the redirect.
|
||
_set_cookie(resp, STATE_COOKIE, f"{provider}|{nonce}", 600)
|
||
return resp
|
||
|
||
|
||
async def fetch_identity(provider: str, token: str) -> tuple[str, str, str]:
|
||
"""(username, display, avatar) for a provider access token."""
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
if provider == "github":
|
||
r = await c.get(GH_API, headers={"Authorization": f"Bearer {token}",
|
||
"Accept": "application/vnd.github+json"})
|
||
if r.status_code != 200:
|
||
raise HTTPException(401, "github user lookup failed")
|
||
p = r.json() or {}
|
||
return p.get("login") or "", p.get("name") or "", p.get("avatar_url") or ""
|
||
|
||
r = await c.get(GOOGLE_API, headers={"Authorization": f"Bearer {token}"})
|
||
if r.status_code != 200:
|
||
raise HTTPException(401, "google user lookup failed")
|
||
p = r.json() or {}
|
||
# Unverified Google emails must never become an identity: anyone can
|
||
# put someone else's address on an account they control, and the
|
||
# allowlist is keyed on exactly this string.
|
||
if not p.get("email") or not p.get("email_verified"):
|
||
raise HTTPException(403, "google account has no verified email")
|
||
return p["email"], p.get("name") or "", p.get("picture") or ""
|
||
|
||
|
||
@app.get("/auth/callback")
|
||
async def auth_callback(request: Request, code: str = "", state: str = ""):
|
||
if AUTH_MODE == "off":
|
||
return RedirectResponse("/", status_code=302)
|
||
|
||
cookie = request.cookies.get(STATE_COOKIE) or ""
|
||
provider, _, nonce = cookie.partition("|")
|
||
if (not code or not state or not nonce
|
||
or provider not in ENABLED_PROVIDERS
|
||
or not secrets.compare_digest(state, nonce)):
|
||
log.warning("auth: rejected callback with bad or missing state")
|
||
return JSONResponse({"error": "invalid state"}, status_code=400)
|
||
|
||
cfg = PROVIDERS[provider]
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
t = await c.post(cfg["token"], headers={"Accept": "application/json"},
|
||
data={"client_id": cfg["client_id"],
|
||
"client_secret": cfg["client_secret"],
|
||
"code": code,
|
||
"grant_type": "authorization_code",
|
||
"redirect_uri": f"{BASE_URL}/auth/callback"})
|
||
token = (t.json() or {}).get("access_token") if t.status_code == 200 else None
|
||
if not token:
|
||
log.warning("auth: %s token exchange failed (%s)", provider, t.status_code)
|
||
return JSONResponse({"error": "token exchange failed"}, status_code=401)
|
||
|
||
username, display, avatar = await fetch_identity(provider, token)
|
||
if not username:
|
||
return JSONResponse({"error": "no username from provider"}, status_code=401)
|
||
|
||
with db() as c:
|
||
existing = c.execute("SELECT * FROM users WHERE username=?",
|
||
(username,)).fetchone()
|
||
if not may_login(username, existing):
|
||
# Log the attempt: on an internet-facing box this is the signal that
|
||
# someone found the login page.
|
||
why = "disabled" if existing else "not invited"
|
||
log.warning("auth: DENIED %s user %r (%s)", provider, username, why)
|
||
return JSONResponse({"error": "not authorised"}, status_code=403)
|
||
|
||
row = upsert_user(username, avatar, display, provider)
|
||
log.info("auth: signed in %s via %s (%s)", username, provider, row["role"])
|
||
resp = RedirectResponse("/", status_code=302)
|
||
_set_cookie(resp, SESSION_COOKIE, new_session(username), SESSION_DAYS * 86400)
|
||
resp.delete_cookie(STATE_COOKIE, path="/")
|
||
return resp
|
||
|
||
|
||
@app.get("/auth/logout")
|
||
async def auth_logout(request: Request):
|
||
tok = request.cookies.get(SESSION_COOKIE)
|
||
if tok:
|
||
with db() as c:
|
||
c.execute("DELETE FROM sessions WHERE token_hash=?", (_hash(tok),))
|
||
resp = RedirectResponse("/auth/login", status_code=302)
|
||
resp.delete_cookie(SESSION_COOKIE, path="/")
|
||
return resp
|
||
|
||
|
||
# ─────────────────────────── quotas ───────────────────────────
|
||
#
|
||
# Two limits, both per user, both 0 = unlimited:
|
||
# quota requests per CALENDAR month (resets on the 1st)
|
||
# storage_quota bytes of imported audio attributable to this user
|
||
#
|
||
# Calendar month rather than a rolling 30-day window on purpose: "you have 12
|
||
# left until the 1st" is something a person can reason about, where a rolling
|
||
# window silently drips capacity back at times nobody can predict.
|
||
|
||
def month_start() -> str:
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-01 00:00:00")
|
||
|
||
|
||
def usage_for(username: str | None) -> dict:
|
||
"""Requests this month and bytes imported, for one user."""
|
||
if not username:
|
||
return {"requests": 0, "storage": 0}
|
||
with db() as c:
|
||
n = c.execute(
|
||
"SELECT COUNT(*) n FROM jobs WHERE requested_by=? AND created_at >= ?",
|
||
(username, month_start())).fetchone()["n"]
|
||
b = c.execute(
|
||
"SELECT COALESCE(SUM(size),0) b FROM jobs "
|
||
"WHERE requested_by=? AND state='imported'", (username,)).fetchone()["b"]
|
||
return {"requests": n, "storage": b}
|
||
|
||
|
||
def check_quota(request: Request, adding: int = 1):
|
||
"""Raise 429 if this request would exceed the caller's monthly allowance.
|
||
|
||
Admins are exempt — someone has to be able to fix the library when a quota
|
||
is what is in the way.
|
||
"""
|
||
u = current_user(request)
|
||
if not u or u["role"] == "admin":
|
||
return
|
||
used = usage_for(u["username"])
|
||
if u["quota"] and used["requests"] + adding > u["quota"]:
|
||
raise HTTPException(429, (
|
||
f"Monthly request quota reached ({used['requests']}/{u['quota']}). "
|
||
f"Resets on the 1st."))
|
||
if u["storage_quota"] and used["storage"] >= u["storage_quota"]:
|
||
raise HTTPException(429, (
|
||
f"Storage quota reached ({used['storage'] // 2**30} GB of "
|
||
f"{u['storage_quota'] // 2**30} GB)."))
|
||
|
||
|
||
@app.get("/api/me")
|
||
async def me(request: Request):
|
||
u = current_user(request)
|
||
if not u:
|
||
return {"user": None, "auth": AUTH_MODE,
|
||
"role": "admin" if AUTH_MODE == "off" else None,
|
||
"providers": ENABLED_PROVIDERS}
|
||
return {
|
||
"user": u["username"], "auth": AUTH_MODE, "role": u["role"],
|
||
"display": u["display"], "avatar": u["avatar"], "provider": u["provider"],
|
||
"quota": u["quota"], "storage_quota": u["storage_quota"],
|
||
"usage": usage_for(u["username"]),
|
||
"providers": ENABLED_PROVIDERS,
|
||
}
|
||
|
||
|
||
class RequestBody(BaseModel):
|
||
artist: str
|
||
title: str
|
||
album: str | None = None
|
||
year: str | None = None
|
||
track: int | None = None
|
||
deezer_id: int | None = None
|
||
cover: str | None = None
|
||
genre: str | None = None
|
||
length: int | None = None
|
||
|
||
|
||
# Album metadata (release year, genre, label) is not in the track search
|
||
# payload, so it costs one extra call per distinct album. Cached forever on
|
||
# purpose: a release date and a genre are immutable, so there is nothing to
|
||
# invalidate, and repeat searches for the same artist stop hitting Deezer at
|
||
# all. Trimmed only to stop it growing without bound on a long-lived process.
|
||
_album_cache: dict[int, dict] = {}
|
||
_album_lock = asyncio.Semaphore(6) # Deezer rate-limits ~50 req/5s per IP
|
||
|
||
|
||
async def album_meta(album_id) -> dict:
|
||
if not album_id:
|
||
return {}
|
||
key = int(album_id)
|
||
if key in _album_cache:
|
||
return _album_cache[key]
|
||
async with _album_lock:
|
||
if key in _album_cache: # filled while we waited for the slot
|
||
return _album_cache[key]
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
r = await c.get(f"{DEEZER}/album/{key}")
|
||
d = r.json() if r.status_code == 200 else {}
|
||
except Exception as e:
|
||
log.debug("album %s lookup failed: %s", key, e)
|
||
return {}
|
||
genres = [g.get("name") for g in (d.get("genres") or {}).get("data", []) if g.get("name")]
|
||
meta = {
|
||
"year": (d.get("release_date") or "")[:4] or None,
|
||
"genre": genres[0] if genres else None,
|
||
"genres": genres,
|
||
"label": d.get("label"),
|
||
"nb_tracks": d.get("nb_tracks"),
|
||
"record_type": d.get("record_type"),
|
||
}
|
||
if len(_album_cache) > 5000:
|
||
for k in list(_album_cache)[:1000]:
|
||
_album_cache.pop(k, None)
|
||
_album_cache[key] = meta
|
||
return meta
|
||
|
||
|
||
def track_row(t: dict) -> dict:
|
||
"""One Deezer track object -> the shape the UI and the job table use."""
|
||
artist = t.get("artist") or {}
|
||
album = t.get("album") or {}
|
||
return {
|
||
"deezer_id": t.get("id"),
|
||
"title": t.get("title") or "",
|
||
"artist": artist.get("name") or "",
|
||
"artist_id": artist.get("id"),
|
||
"artist_pic": artist.get("picture_medium"),
|
||
"album": album.get("title") or "",
|
||
"album_id": album.get("id"),
|
||
"cover": album.get("cover_medium"),
|
||
"cover_big": album.get("cover_big") or album.get("cover_medium"),
|
||
"length": (t.get("duration") or 0) * 1000,
|
||
# 30-second clip from Deezer's CDN. This is what lets you confirm you
|
||
# picked the right recording BEFORE committing a Soulseek search to it
|
||
# -- the single most useful thing in the payload for avoiding a wrong
|
||
# import, and it costs nothing extra.
|
||
"preview": t.get("preview") or None,
|
||
"explicit": bool(t.get("explicit_lyrics")),
|
||
"isrc": t.get("isrc"),
|
||
"rank": t.get("rank") or 0,
|
||
"year": None, "genre": None, "track": None,
|
||
}
|
||
|
||
|
||
async def add_album_meta(rows: list[dict]) -> list[dict]:
|
||
"""Fill year and genre on a result list, concurrently and de-duplicated."""
|
||
ids = {r["album_id"] for r in rows if r.get("album_id")}
|
||
if not ids:
|
||
return rows
|
||
metas = dict(zip(ids, await asyncio.gather(*(album_meta(i) for i in ids))))
|
||
for r in rows:
|
||
m = metas.get(r.get("album_id")) or {}
|
||
r["year"] = r["year"] or m.get("year")
|
||
r["genre"] = r["genre"] or m.get("genre")
|
||
return rows
|
||
|
||
|
||
async def deezer_search(q: str, limit: int = 25) -> list[dict]:
|
||
"""Ranked Deezer track search. Shared by /api/search and /api/import."""
|
||
if not q.strip():
|
||
return []
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
r = await c.get(f"{DEEZER}/search", params={"q": q, "limit": limit * 2})
|
||
# Deezer's WAF answers 403 to anything that looks like an injection
|
||
# attempt -- "'; DROP TABLE" and friends. That is a statement about the
|
||
# query, not an upstream outage, so it belongs on the empty-results
|
||
# path; a 502 would tell the user the service is broken when it is not.
|
||
if r.status_code == 403:
|
||
log.info("deezer refused the query %r", q[:60])
|
||
return []
|
||
if r.status_code != 200:
|
||
raise HTTPException(502, f"deezer returned {r.status_code}")
|
||
data = r.json()
|
||
|
||
qn = norm(q)
|
||
scored = []
|
||
for t in data.get("data", []):
|
||
title = t.get("title") or ""
|
||
artist = (t.get("artist") or {}).get("name") or ""
|
||
|
||
s = float(t.get("rank") or 0) / 10_000 # Deezer popularity, 0-~100
|
||
|
||
# An artist match separates the real track from a covers compilation.
|
||
at = set(norm(artist).split())
|
||
if norm(artist) and norm(artist) in qn:
|
||
s += 300
|
||
|
||
# Title relevance. This has to dominate everything else: without it a
|
||
# completely different song by the right artist competes on equal
|
||
# footing, and "Oasis - Live Forever" resolved to "Cloudburst" because
|
||
# a variant penalty was enough to tip between two unrelated tracks.
|
||
# Whatever query words the artist name did not explain should show up
|
||
# in the title.
|
||
rest = set(qn.split()) - at
|
||
if rest:
|
||
s += 400 * len(rest & set(norm(title).split())) / len(rest)
|
||
for w in set(DERIVATIVE_RE.findall(norm(title))):
|
||
if not re.search(rf"\b{re.escape(w)}\b", qn):
|
||
s -= 150
|
||
# Catch-all for variants the keyword list will never cover in full
|
||
# ("Devonshire Mix", "Radio Edit", "2011 Version"): a plain title beats
|
||
# a parenthesised one. Deliberately small -- it should break ties
|
||
# between versions of the SAME song, never outweigh title relevance
|
||
# above and pick a different song.
|
||
if ("(" in title or "[" in title) and "(" not in q:
|
||
s -= 40
|
||
scored.append((s, track_row(t)))
|
||
|
||
seen, out = set(), []
|
||
for _, item in sorted(scored, key=lambda x: -x[0]):
|
||
key = (norm(item["artist"]), norm(item["title"]))
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append(item)
|
||
return out[:limit]
|
||
|
||
|
||
@app.get("/api/search")
|
||
async def search(q: str, limit: int = Query(25, ge=1, le=50)):
|
||
"""Track search. Returns individual songs, not albums — the whole point.
|
||
|
||
Year and genre are grafted on afterwards because Deezer's track search
|
||
omits both. Enrichment happens only here, not in deezer_search itself, so
|
||
the bulk-import path stays one HTTP call per line.
|
||
"""
|
||
return {"results": await add_album_meta(await deezer_search(q, limit))}
|
||
|
||
|
||
@app.get("/api/suggest")
|
||
async def suggest(q: str):
|
||
"""Autocomplete. Deliberately does NOT enrich or re-rank: this fires on
|
||
every few keystrokes, so it has to be one upstream call and nothing else.
|
||
Deezer's own autocomplete ordering is good enough for a dropdown, and the
|
||
real ranking still happens when the search is actually submitted."""
|
||
q = q.strip()
|
||
if len(q) < 2:
|
||
return {"tracks": [], "albums": [], "artists": []}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=8) as c:
|
||
r = await c.get(f"{DEEZER}/search/autocomplete",
|
||
params={"q": q, "limit": 5})
|
||
d = r.json() if r.status_code == 200 else {}
|
||
except Exception:
|
||
return {"tracks": [], "albums": [], "artists": []}
|
||
|
||
def rows(key):
|
||
return (d.get(key) or {}).get("data", []) or []
|
||
|
||
return {
|
||
"tracks": [{"deezer_id": t.get("id"), "title": t.get("title"),
|
||
"artist": (t.get("artist") or {}).get("name"),
|
||
"cover": (t.get("album") or {}).get("cover_small"),
|
||
"preview": t.get("preview")}
|
||
for t in rows("tracks")[:5]],
|
||
"albums": [{"album_id": a.get("id"), "title": a.get("title"),
|
||
"artist": (a.get("artist") or {}).get("name"),
|
||
"cover": a.get("cover_small"),
|
||
"nb_tracks": a.get("nb_tracks")}
|
||
for a in rows("albums")[:3]],
|
||
"artists": [{"artist_id": a.get("id"), "name": a.get("name"),
|
||
"cover": a.get("picture_small"),
|
||
"nb_album": a.get("nb_album")}
|
||
for a in rows("artists")[:3]],
|
||
}
|
||
|
||
|
||
@app.get("/api/album/{album_id}")
|
||
async def album_detail(album_id: int):
|
||
"""Full track list for an album, so 'Get all' can queue it in one go."""
|
||
# Two calls, not one. /album/{id} embeds a track list, but those nested
|
||
# objects have no track_position -- and taking the list index instead would
|
||
# misnumber every multi-disc release, writing "13 - ..." for disc 2 track 1.
|
||
# /album/{id}/tracks is the only source that carries track_position and
|
||
# disk_number, and those numbers end up in the filename and the tags.
|
||
try:
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
r, tr = await asyncio.gather(
|
||
c.get(f"{DEEZER}/album/{album_id}"),
|
||
c.get(f"{DEEZER}/album/{album_id}/tracks", params={"limit": 300}))
|
||
except Exception as e:
|
||
raise HTTPException(502, f"deezer: {e}")
|
||
if r.status_code != 200:
|
||
raise HTTPException(404, "album not found")
|
||
d = r.json()
|
||
|
||
artist = (d.get("artist") or {}).get("name") or ""
|
||
year = (d.get("release_date") or "")[:4] or None
|
||
genres = [g.get("name") for g in (d.get("genres") or {}).get("data", []) if g.get("name")]
|
||
raw = (tr.json().get("data", []) if tr.status_code == 200
|
||
else (d.get("tracks") or {}).get("data", []))
|
||
tracks = []
|
||
for t in raw:
|
||
row = track_row(t)
|
||
# These nested track objects carry no album of their own, so inherit
|
||
# what the parent already told us rather than looking it up again.
|
||
row.update(album=d.get("title") or "", album_id=d.get("id"),
|
||
cover=d.get("cover_medium"), cover_big=d.get("cover_big"),
|
||
artist=row["artist"] or artist, year=year,
|
||
genre=genres[0] if genres else None,
|
||
track=t.get("track_position"),
|
||
disc=t.get("disk_number"))
|
||
tracks.append(row)
|
||
|
||
return {"album": {"album_id": d.get("id"), "title": d.get("title"),
|
||
"artist": artist, "artist_id": (d.get("artist") or {}).get("id"),
|
||
"cover_big": d.get("cover_big"), "cover": d.get("cover_medium"),
|
||
"year": year, "genres": genres, "label": d.get("label"),
|
||
"nb_tracks": d.get("nb_tracks"),
|
||
"record_type": d.get("record_type"),
|
||
"duration": d.get("duration")},
|
||
"tracks": tracks}
|
||
|
||
|
||
@app.get("/api/artist/{artist_id}")
|
||
async def artist_detail(artist_id: int):
|
||
"""Artist header plus their top tracks and albums."""
|
||
try:
|
||
async with httpx.AsyncClient(timeout=20) as c:
|
||
info, top, albums = await asyncio.gather(
|
||
c.get(f"{DEEZER}/artist/{artist_id}"),
|
||
c.get(f"{DEEZER}/artist/{artist_id}/top", params={"limit": 15}),
|
||
c.get(f"{DEEZER}/artist/{artist_id}/albums", params={"limit": 24}))
|
||
except Exception as e:
|
||
raise HTTPException(502, f"deezer: {e}")
|
||
if info.status_code != 200:
|
||
raise HTTPException(404, "artist not found")
|
||
a = info.json()
|
||
tracks = [track_row(t) for t in (top.json().get("data", []) if top.status_code == 200 else [])]
|
||
await add_album_meta(tracks)
|
||
return {
|
||
"artist": {"artist_id": a.get("id"), "name": a.get("name"),
|
||
"picture": a.get("picture_big"), "nb_fan": a.get("nb_fan"),
|
||
"nb_album": a.get("nb_album")},
|
||
"tracks": tracks,
|
||
"albums": [{"album_id": x.get("id"), "title": x.get("title"),
|
||
"cover": x.get("cover_medium"), "cover_big": x.get("cover_big"),
|
||
"year": (x.get("release_date") or "")[:4] or None,
|
||
"record_type": x.get("record_type"),
|
||
"nb_tracks": x.get("nb_tracks")}
|
||
for x in (albums.json().get("data", []) if albums.status_code == 200 else [])],
|
||
}
|
||
|
||
|
||
async def enqueue_job(artist: str, title: str, album=None, year=None,
|
||
track=None, deezer_id=None, requested_by=None,
|
||
cover=None, genre=None, duration=None) -> str:
|
||
"""Create a job and hand it to the search queue."""
|
||
job_id = str(uuid.uuid4())
|
||
# The single choke point for all three request paths, so normalise here.
|
||
# year reaches a directory name, and artist/title/album reach filenames.
|
||
year = safe_year(year)
|
||
cover = safe_cover(cover)
|
||
artist, title = artist[:MAX_COMPONENT], title[:MAX_COMPONENT]
|
||
album = album[:MAX_COMPONENT] if album else album
|
||
genre = genre[:80] if genre else genre
|
||
with db() as c:
|
||
c.execute(
|
||
"INSERT INTO jobs (id,state,detail,artist,title,album,year,track,"
|
||
"recording_mbid,release_mbid,requested_by,cover,genre,duration) "
|
||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
|
||
(job_id, "queued", "", artist, title, album, year, track,
|
||
str(deezer_id or ""), None, requested_by, cover, genre, duration))
|
||
await work_q.put(job_id)
|
||
return job_id
|
||
|
||
|
||
async def enrich(deezer_id) -> dict:
|
||
"""Deezer's search payload has no track number or release date, and both
|
||
go into the filename and tags. One lookup per request, not per result."""
|
||
if not deezer_id:
|
||
return {}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=15) as c:
|
||
r = await c.get(f"{DEEZER}/track/{deezer_id}")
|
||
if r.status_code != 200:
|
||
return {}
|
||
d = r.json()
|
||
rd = d.get("release_date") or (d.get("album") or {}).get("release_date")
|
||
return {
|
||
"track": d.get("track_position"),
|
||
"year": (rd or "")[:4] or None,
|
||
"album": (d.get("album") or {}).get("title"),
|
||
"artist": (d.get("artist") or {}).get("name"),
|
||
}
|
||
except Exception as e:
|
||
log.warning("deezer track lookup failed: %s", e)
|
||
return {}
|
||
|
||
|
||
@app.post("/api/request")
|
||
async def request_track(body: RequestBody, request: Request):
|
||
check_quota(request)
|
||
e = await enrich(body.deezer_id)
|
||
return {"id": await enqueue_job(
|
||
e.get("artist") or body.artist, body.title,
|
||
e.get("album") or body.album, e.get("year") or body.year,
|
||
e.get("track") or body.track, body.deezer_id,
|
||
requested_by=session_user(request), cover=body.cover,
|
||
genre=body.genre, duration=body.length)}
|
||
|
||
|
||
class BatchBody(BaseModel):
|
||
tracks: list[RequestBody]
|
||
|
||
|
||
@app.post("/api/request/batch")
|
||
async def request_batch(body: BatchBody, request: Request):
|
||
"""Queue a whole album at once.
|
||
|
||
Unlike /api/request this skips the per-track /track/{id} enrichment: the
|
||
caller got these rows from /api/album/{id}, which already carries the
|
||
album, year and track position for every one of them. Re-fetching would be
|
||
nb_tracks extra calls to learn what we were just told.
|
||
"""
|
||
# Checked against the whole batch, not one track: queueing a 30-track album
|
||
# when 5 requests remain should be refused up front, not accepted and then
|
||
# blocked halfway through.
|
||
check_quota(request, adding=len(body.tracks))
|
||
who = session_user(request)
|
||
ids = [await enqueue_job(t.artist, t.title, t.album, t.year, t.track,
|
||
t.deezer_id, requested_by=who, cover=t.cover,
|
||
genre=t.genre, duration=t.length)
|
||
for t in body.tracks]
|
||
log.info("batch: queued %d track(s) for %s", len(ids), who or "anon")
|
||
return {"queued": len(ids), "ids": ids, "pending": work_q.qsize()}
|
||
|
||
|
||
class ImportBody(BaseModel):
|
||
text: str
|
||
|
||
|
||
# "Artist - Title", "Artist – Title" (en dash), "Title by Artist". Anything
|
||
# that doesn't split just goes to Deezer whole — its search copes fine with
|
||
# free text, and guessing wrong is worse than not guessing.
|
||
SPLIT = re.compile(r"\s+[-–—|]\s+|\s+by\s+", re.I)
|
||
DEEZER_URL = re.compile(r"deezer\.com/(?:[a-z]{2}/)?(album|playlist)/(\d+)")
|
||
|
||
# Generous for a human pasting an album or a year-end list, small enough that
|
||
# one request cannot turn into thousands of upstream calls.
|
||
MAX_IMPORT_LINES = 500
|
||
MAX_IMPORT_TRACKS = 1000
|
||
|
||
|
||
async def expand_url(kind: str, ident: str) -> list[str]:
|
||
"""Turn a Deezer album/playlist URL into 'Artist - Title' lines."""
|
||
out = []
|
||
try:
|
||
async with httpx.AsyncClient(timeout=25) as c:
|
||
r = await c.get(f"{DEEZER}/{kind}/{ident}")
|
||
if r.status_code != 200:
|
||
return []
|
||
for t in (r.json().get("tracks") or {}).get("data", []):
|
||
a = (t.get("artist") or {}).get("name") or ""
|
||
out.append(f"{a} - {t.get('title')}" if a else t.get("title", ""))
|
||
except Exception as e:
|
||
log.warning("expanding %s/%s failed: %s", kind, ident, e)
|
||
return [x for x in out if x]
|
||
|
||
|
||
@app.post("/api/import")
|
||
async def import_list(body: ImportBody, request: Request):
|
||
"""Paste a list, get it all queued.
|
||
|
||
Each line is resolved against Deezer and the top-ranked hit is taken
|
||
automatically — there is no picker in bulk mode. Resolved artist/album are
|
||
recorded on every job so a wrong auto-pick is visible in the list rather
|
||
than silently landing the wrong song in the library.
|
||
"""
|
||
lines, queued, skipped = [], [], []
|
||
# Capped before expansion, not after. Each URL line costs an upstream call,
|
||
# and the quota check below cannot run until expansion has finished — so
|
||
# without a cap here a paste of 10,000 album URLs is 10,000 requests to
|
||
# Deezer from this IP before anyone's quota is consulted.
|
||
raw_lines = body.text.splitlines()
|
||
if len(raw_lines) > MAX_IMPORT_LINES:
|
||
raise HTTPException(413, f"paste at most {MAX_IMPORT_LINES} lines at a time "
|
||
f"(got {len(raw_lines)})")
|
||
for raw in raw_lines:
|
||
line = raw.strip().lstrip("-•*0123456789. )").strip()
|
||
if not line:
|
||
continue
|
||
m = DEEZER_URL.search(line)
|
||
lines.extend(await expand_url(*m.groups()) if m else [line])
|
||
if len(lines) > MAX_IMPORT_TRACKS:
|
||
raise HTTPException(413, f"that expands to more than "
|
||
f"{MAX_IMPORT_TRACKS} tracks")
|
||
|
||
# After expansion — a single playlist URL can be 200 tracks, and the point
|
||
# of the check is the real number, not the number of lines pasted.
|
||
check_quota(request, adding=len(lines))
|
||
|
||
for line in lines:
|
||
# Search the line as typed; the ranker already weights an artist-name
|
||
# match heavily, so "Artist - Title" needs no special handling.
|
||
try:
|
||
# Ask for a real candidate pool, not one row. deezer_search ranks
|
||
# what it is given, so limit=1 fetched two results and handed back
|
||
# the better of a demo and a rehearsal take while the studio
|
||
# version -- 6x the popularity -- sat further down the list unseen.
|
||
hits = await deezer_search(SPLIT.sub(" ", line), limit=10)
|
||
except Exception as e:
|
||
skipped.append({"line": line, "why": errstr(e)})
|
||
continue
|
||
if not hits:
|
||
skipped.append({"line": line, "why": "no match on Deezer"})
|
||
continue
|
||
h = hits[0]
|
||
e = await enrich(h["deezer_id"])
|
||
jid = await enqueue_job(
|
||
e.get("artist") or h["artist"], h["title"],
|
||
e.get("album") or h["album"], e.get("year"), e.get("track"),
|
||
h["deezer_id"], requested_by=session_user(request),
|
||
cover=h.get("cover"), duration=h.get("length"))
|
||
queued.append({"id": jid, "line": line, "cover": h.get("cover"),
|
||
"matched": f"{h['artist']} — {h['title']}"})
|
||
|
||
log.info("import: %d queued, %d skipped", len(queued), len(skipped))
|
||
return {"queued": queued, "skipped": skipped, "pending": work_q.qsize()}
|
||
|
||
|
||
@app.get("/api/jobs")
|
||
async def jobs(request: Request, limit: int = Query(200, ge=1, le=1000),
|
||
mine: bool = False):
|
||
"""Own requests by default; admins see everyone's.
|
||
|
||
Scoped rather than global because on a shared instance one person's
|
||
request history is a list of what they listen to, which is nobody else's
|
||
business by default.
|
||
"""
|
||
u = current_user(request)
|
||
everyone = (AUTH_MODE == "off") or (u is not None and u["role"] == "admin" and not mine)
|
||
who = None if everyone else (u["username"] if u else None)
|
||
|
||
where, args = ("", []) if everyone else ("WHERE requested_by IS ?", [who])
|
||
with db() as c:
|
||
rows = c.execute(
|
||
f"SELECT * FROM jobs {where} ORDER BY created_at DESC, rowid DESC LIMIT ?",
|
||
[*args, limit]).fetchall()
|
||
counts = {r["state"]: r["n"] for r in c.execute(
|
||
f"SELECT state, COUNT(*) n FROM jobs {where} GROUP BY state",
|
||
args).fetchall()}
|
||
return {"jobs": [dict(r) for r in rows], "counts": counts,
|
||
"pending": work_q.qsize(), "scope": "all" if everyone else "mine"}
|
||
|
||
|
||
@app.post("/api/retry/{job_id}")
|
||
async def retry(job_id: str, request: Request):
|
||
"""Put a failed job back on the queue. Soulseek failures are usually the
|
||
peer going offline, not the request being wrong, so retrying later often
|
||
just works."""
|
||
u = current_user(request)
|
||
with db() as c:
|
||
row = c.execute("SELECT id, requested_by FROM jobs WHERE id=?",
|
||
(job_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "no such job")
|
||
# Same 404 rather than 403 for someone else's job: a distinct error would
|
||
# confirm the id exists, which is a small oracle over other people's
|
||
# requests.
|
||
if AUTH_MODE != "off" and u and u["role"] != "admin" \
|
||
and row["requested_by"] != u["username"]:
|
||
raise HTTPException(404, "no such job")
|
||
set_state(job_id, "queued", "")
|
||
await work_q.put(job_id)
|
||
return {"ok": True}
|
||
|
||
|
||
# ─────────────────────────── admin ───────────────────────────
|
||
|
||
class NewUser(BaseModel):
|
||
username: str
|
||
provider: str = "github"
|
||
role: str = "user"
|
||
quota: int = 0
|
||
storage_quota_gb: float = 0
|
||
|
||
|
||
class PatchUser(BaseModel):
|
||
role: str | None = None
|
||
quota: int | None = None
|
||
storage_quota_gb: float | None = None
|
||
disabled: bool | None = None
|
||
|
||
|
||
def user_dict(r: sqlite3.Row) -> dict:
|
||
u = usage_for(r["username"])
|
||
return {
|
||
"username": r["username"], "provider": r["provider"], "role": r["role"],
|
||
"display": r["display"], "avatar": r["avatar"],
|
||
"quota": r["quota"], "storage_quota": r["storage_quota"],
|
||
"disabled": bool(r["disabled"]), "created_at": r["created_at"],
|
||
"last_seen": r["last_seen"], "invited_by": r["invited_by"],
|
||
"usage": u,
|
||
# Surfaced so the UI doesn't have to know the allowlist rule: these are
|
||
# the accounts that cannot actually be locked out by disabling them.
|
||
"bootstrap": r["username"].lower() in ALLOWED_USERS,
|
||
}
|
||
|
||
|
||
@app.get("/api/admin/users")
|
||
async def admin_users(request: Request):
|
||
require_admin(request)
|
||
with db() as c:
|
||
rows = c.execute("SELECT * FROM users ORDER BY role='admin' DESC, username").fetchall()
|
||
return {"users": [user_dict(r) for r in rows], "roles": list(ROLES)}
|
||
|
||
|
||
@app.post("/api/admin/users")
|
||
async def admin_add_user(body: NewUser, request: Request):
|
||
"""Invite someone. There is no email or token — the row IS the invitation:
|
||
once it exists, that identity can sign in with the matching provider."""
|
||
who = require_admin(request)
|
||
name = body.username.strip()
|
||
if not name:
|
||
raise HTTPException(400, "username required")
|
||
if body.role not in ROLES:
|
||
raise HTTPException(400, f"role must be one of {', '.join(ROLES)}")
|
||
if body.provider not in PROVIDERS:
|
||
raise HTTPException(400, f"provider must be one of {', '.join(PROVIDERS)}")
|
||
with db() as c:
|
||
if c.execute("SELECT 1 FROM users WHERE username=?", (name,)).fetchone():
|
||
raise HTTPException(409, "that user already exists")
|
||
c.execute(
|
||
"INSERT INTO users (username,provider,role,quota,storage_quota,invited_by) "
|
||
"VALUES (?,?,?,?,?,?)",
|
||
(name, body.provider, body.role, max(0, body.quota),
|
||
int(max(0.0, body.storage_quota_gb) * 2**30), who))
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (name,)).fetchone()
|
||
log.info("admin: %s invited %s as %s", who, name, body.role)
|
||
return user_dict(r)
|
||
|
||
|
||
@app.patch("/api/admin/users/{username}")
|
||
async def admin_patch_user(username: str, body: PatchUser, request: Request):
|
||
who = require_admin(request)
|
||
with db() as c:
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
if not r:
|
||
raise HTTPException(404, "no such user")
|
||
|
||
# An install with no admin cannot be administered, and nothing in the
|
||
# UI can undo it. Refuse to remove the last one — whether by demotion
|
||
# or by disabling them.
|
||
losing_admin = r["role"] == "admin" and (
|
||
(body.role is not None and body.role != "admin") or body.disabled is True)
|
||
if losing_admin:
|
||
others = c.execute(
|
||
"SELECT COUNT(*) n FROM users "
|
||
"WHERE role='admin' AND disabled=0 AND username<>?",
|
||
(username,)).fetchone()["n"]
|
||
if others == 0:
|
||
raise HTTPException(
|
||
409, "this is the only active admin — promote someone else first")
|
||
|
||
sets, vals = [], []
|
||
if body.role is not None:
|
||
if body.role not in ROLES:
|
||
raise HTTPException(400, f"role must be one of {', '.join(ROLES)}")
|
||
sets.append("role=?"); vals.append(body.role)
|
||
if body.quota is not None:
|
||
sets.append("quota=?"); vals.append(max(0, body.quota))
|
||
if body.storage_quota_gb is not None:
|
||
sets.append("storage_quota=?")
|
||
vals.append(int(max(0.0, body.storage_quota_gb) * 2**30))
|
||
if body.disabled is not None:
|
||
sets.append("disabled=?"); vals.append(1 if body.disabled else 0)
|
||
if not sets:
|
||
raise HTTPException(400, "nothing to change")
|
||
|
||
c.execute(f"UPDATE users SET {', '.join(sets)} WHERE username=?",
|
||
[*vals, username])
|
||
# Disabling must take effect now. session_user re-checks on every
|
||
# request anyway, but dropping the rows makes it unambiguous.
|
||
if body.disabled:
|
||
c.execute("DELETE FROM sessions WHERE username=?", (username,))
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
log.info("admin: %s updated %s (%s)", who, username, ", ".join(sets))
|
||
return user_dict(r)
|
||
|
||
|
||
@app.delete("/api/admin/users/{username}")
|
||
async def admin_delete_user(username: str, request: Request):
|
||
who = require_admin(request)
|
||
with db() as c:
|
||
r = c.execute("SELECT * FROM users WHERE username=?", (username,)).fetchone()
|
||
if not r:
|
||
raise HTTPException(404, "no such user")
|
||
if r["role"] == "admin":
|
||
others = c.execute(
|
||
"SELECT COUNT(*) n FROM users "
|
||
"WHERE role='admin' AND disabled=0 AND username<>?",
|
||
(username,)).fetchone()["n"]
|
||
if others == 0:
|
||
raise HTTPException(409, "this is the only active admin")
|
||
c.execute("DELETE FROM sessions WHERE username=?", (username,))
|
||
c.execute("DELETE FROM users WHERE username=?", (username,))
|
||
# Their jobs are deliberately kept: the history is a record of what is in
|
||
# the library and why, and deleting a person should not rewrite it.
|
||
log.info("admin: %s deleted user %s (jobs retained)", who, username)
|
||
return {"ok": True, "jobs_retained": True}
|
||
|
||
|
||
@app.get("/api/admin/activity")
|
||
async def admin_activity(request: Request, days: int = Query(30, ge=1, le=365)):
|
||
require_admin(request)
|
||
days = max(1, min(days, 365))
|
||
since = (datetime.now(timezone.utc) - timedelta(days=days)).strftime("%Y-%m-%d")
|
||
with db() as c:
|
||
daily = c.execute(
|
||
"SELECT substr(created_at,1,10) d, COUNT(*) n, "
|
||
" SUM(state='imported') imported, SUM(state='failed') failed "
|
||
"FROM jobs WHERE created_at >= ? GROUP BY d ORDER BY d", (since,)).fetchall()
|
||
per_user = c.execute(
|
||
"SELECT COALESCE(requested_by,'(unattributed)') u, COUNT(*) n, "
|
||
" COALESCE(SUM(CASE WHEN state='imported' THEN size END),0) bytes "
|
||
"FROM jobs WHERE created_at >= ? GROUP BY u ORDER BY n DESC",
|
||
(since,)).fetchall()
|
||
top = c.execute(
|
||
"SELECT artist, COUNT(*) n FROM jobs WHERE created_at >= ? "
|
||
"GROUP BY artist ORDER BY n DESC LIMIT 8", (since,)).fetchall()
|
||
|
||
# Zero-fill: a bar chart with gaps for quiet days misreads as missing data
|
||
# rather than as nothing having happened.
|
||
by_day = {r["d"]: r for r in daily}
|
||
today = datetime.now(timezone.utc).date()
|
||
series = []
|
||
for i in range(days - 1, -1, -1):
|
||
d = (today - timedelta(days=i)).isoformat()
|
||
r = by_day.get(d)
|
||
series.append({"date": d, "n": r["n"] if r else 0,
|
||
"imported": (r["imported"] or 0) if r else 0,
|
||
"failed": (r["failed"] or 0) if r else 0})
|
||
return {"days": days, "series": series,
|
||
"per_user": [dict(r) for r in per_user],
|
||
"top_artists": [dict(r) for r in top]}
|
||
|
||
|
||
def _proc_stat() -> tuple[int, int]:
|
||
"""(busy, total) jiffies from /proc/stat."""
|
||
with open("/proc/stat") as f:
|
||
parts = [int(x) for x in f.readline().split()[1:]]
|
||
idle = parts[3] + (parts[4] if len(parts) > 4 else 0)
|
||
return sum(parts) - idle, sum(parts)
|
||
|
||
|
||
_cpu_prev: tuple[int, int] | None = None
|
||
|
||
|
||
def system_stats() -> dict:
|
||
"""Host CPU/RAM plus disk for the paths that matter.
|
||
|
||
Reads /proc directly rather than adding psutil: it is ~30 lines, keeps the
|
||
image free of a compiled dependency, and /proc in a container reports the
|
||
HOST's CPU and memory — which is what you actually want to see for a
|
||
homelab box. Everything is individually guarded so a non-Linux dev machine
|
||
degrades to nulls instead of 500ing the whole panel.
|
||
"""
|
||
global _cpu_prev
|
||
out: dict = {}
|
||
|
||
try:
|
||
busy, total = _proc_stat()
|
||
if _cpu_prev:
|
||
db_, dt = busy - _cpu_prev[0], total - _cpu_prev[1]
|
||
out["cpu_pct"] = round(100 * db_ / dt, 1) if dt > 0 else None
|
||
else:
|
||
out["cpu_pct"] = None # first call has no baseline to diff
|
||
_cpu_prev = (busy, total)
|
||
except Exception:
|
||
out["cpu_pct"] = None
|
||
|
||
try:
|
||
mem = {}
|
||
with open("/proc/meminfo") as f:
|
||
for line in f:
|
||
k, _, v = line.partition(":")
|
||
mem[k] = int(v.split()[0]) * 1024
|
||
total_b = mem.get("MemTotal", 0)
|
||
avail = mem.get("MemAvailable", 0)
|
||
out["mem_total"] = total_b
|
||
out["mem_used"] = total_b - avail
|
||
out["mem_pct"] = round(100 * (total_b - avail) / total_b, 1) if total_b else None
|
||
except Exception:
|
||
out["mem_total"] = out["mem_used"] = out["mem_pct"] = None
|
||
|
||
try:
|
||
out["load"] = [round(x, 2) for x in os.getloadavg()]
|
||
out["cpus"] = os.cpu_count()
|
||
except Exception:
|
||
out["load"] = None
|
||
|
||
disks = {}
|
||
for label, p in (("library", LIBRARY_DIR), ("downloads", DOWNLOAD_DIR)):
|
||
try:
|
||
s = os.statvfs(p)
|
||
total_b = s.f_blocks * s.f_frsize
|
||
free = s.f_bavail * s.f_frsize
|
||
disks[label] = {"total": total_b, "used": total_b - free, "free": free,
|
||
"pct": round(100 * (total_b - free) / total_b, 1) if total_b else None}
|
||
except Exception:
|
||
disks[label] = None
|
||
out["disks"] = disks
|
||
return out
|
||
|
||
|
||
@app.get("/api/admin/system")
|
||
async def admin_system(request: Request):
|
||
require_admin(request)
|
||
out = {"system": system_stats(), "workers": WORKERS, "pending": work_q.qsize()}
|
||
|
||
with db() as c:
|
||
out["jobs"] = {r["state"]: r["n"] for r in c.execute(
|
||
"SELECT state, COUNT(*) n FROM jobs GROUP BY state").fetchall()}
|
||
out["library"] = {
|
||
"tracks": c.execute(
|
||
"SELECT COUNT(*) n FROM jobs WHERE state='imported'").fetchone()["n"],
|
||
"bytes": c.execute(
|
||
"SELECT COALESCE(SUM(size),0) b FROM jobs "
|
||
"WHERE state='imported'").fetchone()["b"],
|
||
}
|
||
out["users"] = c.execute("SELECT COUNT(*) n FROM users").fetchone()["n"]
|
||
out["sessions"] = c.execute(
|
||
"SELECT COUNT(*) n FROM sessions WHERE expires_at > ?",
|
||
(datetime.now(timezone.utc).isoformat(),)).fetchone()["n"]
|
||
|
||
slskd: dict = {"ok": False}
|
||
try:
|
||
async with slskd_client(PROBE_TIMEOUT) as sc:
|
||
r = await sc.get("/api/v0/application")
|
||
slskd["ok"] = r.status_code == 200
|
||
if r.status_code == 200:
|
||
d = r.json()
|
||
srv = d.get("server") or {}
|
||
slskd["connected"] = bool(srv.get("isConnected"))
|
||
slskd["logged_in"] = bool(srv.get("isLoggedIn"))
|
||
slskd["version"] = d.get("version")
|
||
shares = d.get("shares") or {}
|
||
slskd["shared_files"] = (shares.get("directories") or {}).get("fileCount") \
|
||
or shares.get("fileCount")
|
||
else:
|
||
slskd["error"] = f"HTTP {r.status_code}"
|
||
except Exception as e:
|
||
slskd["error"] = errstr(e)
|
||
out["slskd"] = slskd
|
||
return out
|
||
|
||
|
||
@app.get("/api/health")
|
||
async def health():
|
||
ok, detail = True, {}
|
||
async with slskd_client(PROBE_TIMEOUT) as sc:
|
||
try:
|
||
r = await sc.get("/api/v0/application")
|
||
detail["slskd"] = r.status_code
|
||
ok = ok and r.status_code == 200
|
||
except Exception as e:
|
||
detail["slskd"] = errstr(e)
|
||
ok = False
|
||
detail["library_writable"] = os.access(LIBRARY_DIR, os.W_OK)
|
||
detail["downloads_readable"] = os.access(DOWNLOAD_DIR, os.R_OK)
|
||
return {"ok": ok and detail["library_writable"], **detail}
|
||
|
||
|
||
@app.get("/")
|
||
async def index():
|
||
f = WEB / "index.html"
|
||
if not f.exists():
|
||
raise HTTPException(500, "UI not built — run `npm run build` in web/")
|
||
return FileResponse(f)
|
||
|
||
|
||
# Next.js emits its immutable, content-hashed chunks under /_next/static.
|
||
#
|
||
# Guarded rather than mounted unconditionally so that importing this module
|
||
# does not require build artifacts: CI typechecks the API before the UI has
|
||
# been built, and tests should not have to run npm first. A deploy that really
|
||
# is missing its UI says so at startup and again on the first request, rather
|
||
# than silently serving 404s.
|
||
if (WEB / "_next").is_dir():
|
||
app.mount("/_next", StaticFiles(directory=WEB / "_next"), name="next")
|
||
else:
|
||
log.warning("web/out/_next is missing — the UI was not built. "
|
||
"API endpoints work; the web page will not.")
|