#!/usr/bin/env python3
import argparse
import html
import http.server
import json
import re
import socket
import subprocess
import sys
import threading
import time
import urllib.request
from datetime import datetime
from pathlib import Path
CACHE_DIR = "/var/lib/mopidy/.local/share/mopidy/spotify/credentials-cache"
CREDS_NAME = "credentials.json"
GET_CREDS = Path.home() / ".cargo" / "bin" / "get_creds"
TEST_TRACK = "spotify:track:4uLU6hMCjMI75M1A2tKUQC"
DEFAULT_TUNNEL = "powellc@192.168.40.186"
INSTALL_CMD = (
"cargo install --git https://github.com/librespot-org/librespot "
"--no-default-features --features rustls-tls-native-roots --example get_creds"
)
RPC_URL = "http://localhost:6680/mopidy/rpc"
def load_env(path):
env = {}
if not path.is_file():
return env
for line in path.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
env[key.strip()] = value.strip().strip('"').strip("'")
return env
def docker(*args, check=True):
return subprocess.run(["docker", *args], check=check, capture_output=True, text=True)
def container_running(container):
result = docker("inspect", "-f", "{{.State.Running}}", container, check=False)
return result.returncode == 0 and result.stdout.strip() == "true"
def cache_owner(container):
result = docker("exec", container, "stat", "-c", "%u:%g", CACHE_DIR, check=False)
owner = result.stdout.strip()
return owner if result.returncode == 0 and owner else ""
def pick_port(preferred):
for port in range(preferred, preferred + 20):
with socket.socket() as sock:
try:
sock.bind(("0.0.0.0", port))
return port
except OSError:
continue
raise RuntimeError("no free port")
class PageHandler(http.server.BaseHTTPRequestHandler):
state = {}
tunnel = DEFAULT_TUNNEL
def do_GET(self):
if self.path not in ("/", "/index.html"):
self.send_error(404)
return
state = self.state
url = html.escape(state.get("url") or "", quote=True)
status = html.escape(state.get("status", ""))
detail = html.escape(state.get("detail", ""))
waiting_since = state.get("waiting_since")
tunnel = ""
if waiting_since and time.time() - waiting_since > 30:
tunnel = (
"Still waiting. Is the SSH tunnel up on your laptop?
"
f"ssh -N -L 8898:localhost:8898 {self.tunnel}
"
)
button = f'Authorize with Spotify
' if url else ""
body = f"""
Mopidy Spotify credentials
Mopidy Spotify credentials
- Start the tunnel on your laptop:
ssh -N -L 8898:localhost:8898 {self.tunnel}
- Click the button and approve the login.
{button}
{tunnel}
{status}
{detail}
"""
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Refresh", "2")
self.end_headers()
self.wfile.write(body.encode())
def log_message(self, *args):
pass
def start_server(port, state, tunnel):
handler = type("Handler", (PageHandler,), {"state": state, "tunnel": tunnel})
server = http.server.ThreadingHTTPServer(("0.0.0.0", port), handler)
threading.Thread(target=server.serve_forever, daemon=True).start()
return server
def start_get_creds(creds_dir):
creds_dir.mkdir(parents=True, exist_ok=True)
stale = creds_dir / CREDS_NAME
if stale.exists():
stale.rename(creds_dir / f"{CREDS_NAME}.old-{datetime.now():%Y%m%d-%H%M%S}")
proc = subprocess.Popen(
[str(GET_CREDS), str(creds_dir)],
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
)
pattern = re.compile(r"^Browse to: (\S+)\s*$")
url = None
deadline = time.time() + 30
while time.time() < deadline:
line = proc.stdout.readline()
if not line and proc.poll() is not None:
break
match = pattern.match(line.strip())
if match:
url = match.group(1)
break
if url is None:
proc.kill()
raise RuntimeError("could not parse the authorization URL from get_creds")
threading.Thread(target=proc.stdout.read, daemon=True).start()
return proc, url
def wait_for_blob(path, proc, timeout):
deadline = time.time() + timeout
while time.time() < deadline:
if path.exists() and path.stat().st_size > 0:
return True
if proc is not None and proc.poll() is not None:
return path.exists() and path.stat().st_size > 0
time.sleep(0.5)
return False
def backup_existing(container, backup_dir):
backup_dir.mkdir(parents=True, exist_ok=True)
target = backup_dir / f"credentials-{datetime.now():%Y%m%d-%H%M%S}.json"
result = docker("cp", f"{container}:{CACHE_DIR}/{CREDS_NAME}", str(target), check=False)
return target if result.returncode == 0 else None
def install_blob(container, blob):
owner = cache_owner(container)
docker("exec", container, "mkdir", "-p", CACHE_DIR)
docker("cp", str(blob), f"{container}:{CACHE_DIR}/{CREDS_NAME}")
if owner:
docker("exec", container, "chown", owner, f"{CACHE_DIR}/{CREDS_NAME}")
docker("exec", container, "chmod", "600", f"{CACHE_DIR}/{CREDS_NAME}")
def rpc(method, params=None, timeout=10):
payload = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params or {}}).encode()
request = urllib.request.Request(RPC_URL, data=payload, headers={"Content-Type": "application/json"})
with urllib.request.urlopen(request, timeout=timeout) as response:
return json.load(response).get("result")
def wait_for_mopidy(timeout=90):
deadline = time.time() + timeout
while time.time() < deadline:
try:
rpc("core.get_version", timeout=5)
return True
except Exception:
time.sleep(2)
return False
def verify_playback(container, keep_playing):
rpc("core.tracklist.clear")
rpc("core.tracklist.add", {"uris": [TEST_TRACK]})
rpc("core.playback.play")
time.sleep(10)
state = rpc("core.playback.get_state")
logs = docker("logs", "--since", "90s", container, check=False)
combined = (logs.stdout or "") + (logs.stderr or "")
errors = [
line
for line in combined.splitlines()
if "GStreamer error" in line or "track is not available" in line
]
if not keep_playing:
rpc("core.playback.stop")
rpc("core.tracklist.clear")
return state, errors
def main():
parser = argparse.ArgumentParser(
description="Regenerate the Spotify credentials blob used by mopidy-spotify.",
epilog=(
"Reads .env for CONTAINER_NAME, COMPOSE_PROJECT_NAME and HTTP_PORT. "
"The interactive login needs an SSH tunnel from the browser machine."
),
)
parser.add_argument("--env-file", type=Path, default=Path(".env"), help="compose env file (default: .env)")
parser.add_argument("--container", help="container to install into (default from .env)")
parser.add_argument("--rpc-url", help="Mopidy JSON-RPC URL (default from .env HTTP_PORT)")
parser.add_argument("--tunnel", default=DEFAULT_TUNNEL, help=f"ssh user@host for the tunnel (default: {DEFAULT_TUNNEL})")
parser.add_argument("--port", type=int, default=8901, help="port for the setup web page (default: 8901)")
parser.add_argument("--creds-dir", type=Path, default=Path.home() / "creds-new", help="where get_creds stores the new blob")
parser.add_argument("--from-file", type=Path, help="skip the Spotify login and install an existing blob")
parser.add_argument("--install-helper", action="store_true", help="build get_creds with cargo if it is missing")
parser.add_argument("--no-restart", action="store_true", help="install the blob but do not restart the container")
parser.add_argument("--no-verify", action="store_true", help="skip the playback verification")
parser.add_argument("--keep-playing", action="store_true", help="do not stop playback after verification")
parser.add_argument("--backup-dir", type=Path, default=Path.home() / "creds-backups", help="where to back up the current blob")
args = parser.parse_args()
global RPC_URL
env = load_env(args.env_file)
container = args.container or env.get("CONTAINER_NAME")
if not container:
project = env.get("COMPOSE_PROJECT_NAME") or Path.cwd().name
container = f"{project}-mopidy-1"
RPC_URL = args.rpc_url or f"http://localhost:{env.get('HTTP_PORT', '6680')}/mopidy/rpc"
page_host = args.tunnel.split("@")[-1]
if not container_running(container):
sys.exit(f"error: container {container} is not running")
print(f"target: container={container} rpc={RPC_URL}")
server = None
proc = None
state = {"status": "starting", "detail": ""}
try:
if args.from_file:
blob = args.from_file
if not blob.is_file():
sys.exit(f"error: {blob} not found")
print(f"using existing blob: {blob}")
else:
if not GET_CREDS.exists():
if args.install_helper:
print("building get_creds ...")
subprocess.run(INSTALL_CMD, shell=True, check=True)
else:
sys.exit(f"error: {GET_CREDS} not found; run:\n {INSTALL_CMD}\nor pass --install-helper")
proc, url = start_get_creds(args.creds_dir)
state.update(url=url, status="waiting for Spotify login", waiting_since=time.time())
port = pick_port(args.port)
server = start_server(port, state, args.tunnel)
print(f"Open http://{page_host}:{port}/ on your laptop")
print(f"Tunnel: ssh -N -L 8898:localhost:8898 {args.tunnel}")
blob = args.creds_dir / CREDS_NAME
if not wait_for_blob(blob, proc, timeout=600):
state.update(status="timed out", detail="no credentials file was produced")
sys.exit("error: timed out waiting for the Spotify login")
print(f"login captured: {blob}")
state.update(status="login captured", detail=str(blob))
backup = backup_existing(container, args.backup_dir)
state.update(status="installing credentials", detail=f"backup: {backup}" if backup else "")
install_blob(container, blob)
print("credentials installed")
if not args.no_restart:
state.update(status="restarting mopidy", detail="")
docker("restart", container)
if not wait_for_mopidy():
state.update(status="mopidy did not come back up", detail="")
sys.exit("error: mopidy did not come back up")
print("mopidy restarted")
if not args.no_verify:
state.update(status="verifying playback", detail="")
playback_state, errors = verify_playback(container, args.keep_playing)
if playback_state == "playing" and not errors:
state.update(status="done", detail="playback verified, no GStreamer errors")
print("OK: playback verified, no GStreamer errors")
else:
state.update(status="verification failed", detail=f"state={playback_state} errors={len(errors)}")
print(f"FAIL: state={playback_state}")
for line in errors[:5]:
print(f" {line}")
sys.exit(1)
else:
state.update(status="done", detail="credentials installed, verification skipped")
print("OK: credentials installed")
except KeyboardInterrupt:
print("\ninterrupted")
sys.exit(130)
finally:
if proc is not None and proc.poll() is None:
proc.terminate()
if server is not None:
time.sleep(8)
server.shutdown()
if __name__ == "__main__":
main()