#!/usr/bin/env python3
"""Audiobook player server: user accounts + a safe proxy to OpenAudible.
python3 server.py setup one-time: store the OpenAudible login the server uses
python3 server.py adduser NAME create an account (prompts for a password)
python3 server.py passwd NAME change a password
python3 server.py deluser NAME remove an account
python3 server.py users list accounts
python3 server.py check test the OpenAudible connection and whether its files can stream to a phone
python3 server.py [serve] run (--port 8080 --upstream http://172.16.16.106:8000)
Each person signs in with their own account and gets their own listening progress.
Only the server knows the OpenAudible password; it never reaches the browser.
"""
import argparse
import base64
import getpass
import hashlib
import hmac
import json
import os
import re
import secrets
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.cookies import SimpleCookie
from urllib.parse import unquote
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.environ.get("DATA_DIR", HERE) # where accounts/progress live (a volume in Docker)
F_CONFIG = os.path.join(DATA, "config.json") # OpenAudible login (chmod 600)
F_USERS = os.path.join(DATA, "users.json") # name -> salted scrypt hash
F_SESSIONS = os.path.join(DATA, "sessions.json") # token hash -> {user, exp}
F_PROGRESS = os.path.join(DATA, "progress_users.json")
F_LEGACY = os.path.join(DATA, "progress.json") # single-user file from the first version
COOKIE = "oa_session"
SESSION_DAYS = 90
# Files the browser fetches without a session (home-screen icon and manifest).
PUBLIC = {"/manifest.webmanifest": "application/manifest+json", "/sw.js": "text/javascript"}
ICON = re.compile(r"^/icon-[a-z0-9]+\.png$")
AUDIO_TYPES = {".mp3": "audio/mpeg", ".m4b": "audio/mp4", ".m4a": "audio/mp4", ".aac": "audio/aac",
".ogg": "audio/ogg", ".opus": "audio/ogg", ".flac": "audio/flac", ".wav": "audio/wav"}
AUDIO_PATH = re.compile(r"^/api/books/[^/]+/files/([^/]+)$")
RANGE = re.compile(r"^bytes=(\d*)-(\d*)$")
# Only read-only endpoints needed for listening are reachable; admin commands/import are not.
ALLOWED = [re.compile(p) for p in (
r"^/api/books$",
r"^/api/books/[^/]+$",
r"^/api/books/[^/]+/(chapters|chapters/flat|cover)$",
r"^/api/books/[^/]+/files/[^/]+$",
r"^/admin/api/books/[^/]+/files$",
)]
PASS_HEADERS = ("Range", "If-None-Match", "If-Modified-Since")
RESP_HEADERS = ("Content-Type", "Content-Length", "Content-Range", "Accept-Ranges", "ETag", "Last-Modified", "Cache-Control")
lock = threading.Lock()
# ---------- tiny JSON stores ----------
def log(*a):
print(time.strftime("%H:%M:%S"), *a, flush=True)
def read(path, default):
try:
with open(path) as f:
return json.load(f)
except (OSError, ValueError):
return default
def write(path, data, private=False):
tmp = path + ".tmp"
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600 if private else 0o644)
with os.fdopen(fd, "w") as f:
json.dump(data, f, indent=1)
os.replace(tmp, path)
def hash_pw(password, salt=None):
salt = salt or secrets.token_bytes(16)
h = hashlib.scrypt(password.encode(), salt=salt, n=2 ** 14, r=8, p=1)
return {"salt": salt.hex(), "hash": h.hex()}
def check_pw(password, rec):
h = hash_pw(password, bytes.fromhex(rec["salt"]))["hash"]
return hmac.compare_digest(h, rec["hash"])
def oa_login():
"""OpenAudible credentials: env vars win over config.json (handy for containers)."""
cfg = read(F_CONFIG, {})
return os.environ.get("OPENAUDIBLE_USER") or cfg.get("user"), os.environ.get("OPENAUDIBLE_PASSWORD") or cfg.get("password", "")
USERNAME = re.compile(r"^[a-z0-9_.-]{1,32}$")
def revoke_sessions(name):
write(F_SESSIONS, {k: v for k, v in read(F_SESSIONS, {}).items() if v["user"] != name}, private=True)
def sha(token):
return hashlib.sha256(token.encode()).hexdigest()
# ---------- command-line account management ----------
def ask_password(confirm=True):
while True:
p = getpass.getpass("Password: ")
if len(p) < 6:
print("Use at least 6 characters.")
elif confirm and getpass.getpass("Again: ") != p:
print("Those didn't match.")
else:
return p
def cli(argv):
cmd = argv[0]
users = read(F_USERS, {})
if cmd == "setup":
user = input("OpenAudible username: ")
pw = getpass.getpass("OpenAudible password: ")
write(F_CONFIG, {"user": user, "password": pw}, private=True)
print("Saved to config.json (readable only by you). All player accounts share this login.")
elif cmd == "adduser":
name = argv[1].lower()
if name in users:
sys.exit("That user already exists (use passwd).")
users[name] = hash_pw(ask_password())
if len(users) == 1 or "--admin" in argv:
users[name]["admin"] = True
write(F_USERS, users, private=True)
if len(users) == 1 and os.path.exists(F_LEGACY): # carry over progress from before accounts
legacy = read(F_LEGACY, {})
if legacy and all(isinstance(v, dict) and "pos" in v for v in legacy.values()):
write(F_PROGRESS, {name: legacy})
print(f"Imported your earlier listening progress into '{name}'.")
print(f"Created '{name}'.")
elif cmd == "passwd":
name = argv[1].lower()
if name not in users:
sys.exit("No such user.")
users[name] = {**hash_pw(ask_password()), **({"admin": True} if users[name].get("admin") else {})}
write(F_USERS, users, private=True)
revoke_sessions(name)
print("Password changed; that user was signed out of all devices.")
elif cmd == "deluser":
name = argv[1].lower()
users.pop(name, None)
write(F_USERS, users, private=True)
write(F_SESSIONS, {k: v for k, v in read(F_SESSIONS, {}).items() if v["user"] != name}, private=True)
print(f"Removed '{name}'.")
elif cmd == "users":
print("\n".join(n + (" (admin)" if users[n].get("admin") else "") for n in sorted(users)) or "(no users yet; run: python3 server.py adduser NAME)")
def check(argv):
"""Ask OpenAudible what it has and whether it can stream it to a phone."""
ap = argparse.ArgumentParser(prog="server.py check")
ap.add_argument("--upstream", default=os.environ.get("OPENAUDIBLE_URL", "http://172.16.16.106:8000"))
ap.add_argument("--books", type=int, default=3, help="how many books to test")
a = ap.parse_args(argv)
base = a.upstream.rstrip("/")
user, pw = oa_login()
auth = {"Authorization": "Basic " + base64.b64encode(f"{user}:{pw}".encode()).decode()} if user else {}
def fetch(path, **headers):
return urllib.request.urlopen(urllib.request.Request(base + path, headers={**auth, **headers}), timeout=30)
def step(label, fn):
try:
return fn()
except urllib.error.HTTPError as e:
sys.exit(f"{label}: HTTP {e.code} ({'wrong OpenAudible login? run: python3 server.py setup' if e.code == 401 else e.reason})")
except Exception as e:
sys.exit(f"{label}: {e}")
print(f"OpenAudible at {base}")
books = step("library", lambda: json.load(fetch("/api/books")))
books = books if isinstance(books, list) else list(books.values())
print(f" {len(books)} books")
problems = []
for b in books[: a.books]:
bid = b.get("id") or b.get("asin") or b.get("ASIN") or b.get("book_id")
print(f"\n{b.get('title', bid)} [{bid}]")
files = step("file list", lambda: json.load(fetch(f"/admin/api/books/{bid}/files")))
names = [f if isinstance(f, str) else (f.get("name") or f.get("filename") or f.get("file") or f.get("path") or "") for f in (files if isinstance(files, list) else files.get("files", []))]
names = [n.split("/")[-1] for n in names if n]
if not names:
print(" no files (not downloaded in OpenAudible?)")
problems.append(f"{b.get('title', bid)}: no files")
for n in names:
ext = os.path.splitext(n)[1].lower()
if ext not in AUDIO_TYPES:
print(f" {n}: not playable in a browser ({ext}); convert it to MP3/M4B in OpenAudible")
continue
try:
r = fetch(f"/api/books/{bid}/files/{urllib.parse.quote(n)}", Range="bytes=0-1")
got = r.headers
ranged = r.status == 206 and bool(got.get("Content-Range"))
print(f" {n}: HTTP {r.status}, {got.get('Content-Type')}, Content-Range={got.get('Content-Range')}, Accept-Ranges={got.get('Accept-Ranges')}")
print(" byte ranges:", "supported" if ranged else "NOT supported by OpenAudible; the player server will emulate them (seeking is slower on big files)")
if got.get("Content-Type", "").startswith(("application/octet", "binary/")):
print(" content type is generic; the player server will correct it")
except Exception as e:
print(f" {n}: stream failed: {e}")
problems.append(f"{n}: {e}")
print("\n" + ("Problems found:\n " + "\n ".join(problems) if problems else "OK: OpenAudible is reachable and has playable files."))
# ---------- login page ----------
LOGIN = """
Sign in
"""
class Handler(BaseHTTPRequestHandler):
upstream = ""
fails = {} # ip -> [timestamps] for simple login throttling
# ----- helpers -----
def send_body(self, code, body, ctype, extra=()):
if isinstance(body, str):
body = body.encode()
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
for k, v in extra:
self.send_header(k, v)
self.end_headers()
self.wfile.write(body)
def json(self, code, obj, extra=()):
self.send_body(code, json.dumps(obj), "application/json", extra)
def user(self):
c = SimpleCookie(self.headers.get("Cookie", ""))
if COOKIE not in c:
return None
s = read(F_SESSIONS, {}).get(sha(c[COOKIE].value))
return s["user"] if s and s["exp"] > time.time() else None
def deny(self, path):
if self.command == "GET" and path in ("/", "/index.html"):
self.send_response(302)
self.send_header("Location", "/login")
self.end_headers()
else:
self.json(401, {"error": "not signed in"})
# ----- routes -----
def do_GET(self):
path = self.path.split("?")[0]
if path == "/login":
return self.send_body(200, LOGIN, "text/html; charset=utf-8")
if path == "/logout":
return self.logout()
if path in PUBLIC or (ICON.match(path) and os.path.exists(os.path.join(HERE, path[1:]))):
with open(os.path.join(HERE, path[1:]), "rb") as f:
return self.send_body(200, f.read(), PUBLIC.get(path, "image/png"), [("Cache-Control", "no-cache" if path in PUBLIC else "public, max-age=86400")])
user = self.user()
if not user:
return self.deny(path)
if path in ("/", "/index.html"):
with open(os.path.join(HERE, "index.html"), "rb") as f:
return self.send_body(200, f.read(), "text/html; charset=utf-8", [("Cache-Control", "no-cache")])
if path == "/me":
return self.json(200, {"user": user, "admin": bool(read(F_USERS, {}).get(user, {}).get("admin"))})
if path in ("/admin", "/admin/users"):
if not read(F_USERS, {}).get(user, {}).get("admin"):
return self.json(403, {"error": "admins only"}) if path == "/admin/users" else self.send_body(403, "Admins only.", "text/plain")
if path == "/admin":
with open(os.path.join(HERE, "admin.html"), "rb") as f:
return self.send_body(200, f.read(), "text/html; charset=utf-8", [("Cache-Control", "no-cache")])
return self.json(200, self.user_list())
if path == "/sync/progress":
return self.json(200, read(F_PROGRESS, {}).get(user, {}))
if any(p.match(path) for p in ALLOWED):
return self.proxy()
self.send_error(404)
def do_HEAD(self):
path = self.path.split("?")[0]
if not self.user():
self.send_response(401)
return self.end_headers()
if any(p.match(path) for p in ALLOWED):
return self.proxy(head=True)
self.send_error(404)
def do_POST(self):
path = self.path.split("?")[0]
n = int(self.headers.get("Content-Length", 0) or 0)
if n > 1_000_000:
return self.send_error(413)
try:
body = json.loads(self.rfile.read(n) or b"{}")
except ValueError:
return self.json(400, {"error": "bad json"})
if path == "/login":
return self.login(body)
user = self.user()
if not user:
return self.deny(path)
if path.startswith("/admin/users"):
if not read(F_USERS, {}).get(user, {}).get("admin"):
return self.json(403, {"error": "admins only"})
if not self.headers.get("Content-Type", "").startswith("application/json"):
return self.json(415, {"error": "json only"}) # blocks cross-site form posts
with lock:
return self.admin_api(user, path, body)
if path == "/sync/progress":
with lock:
allp = read(F_PROGRESS, {})
mine = allp.setdefault(user, {})
for k, v in body.items(): # newest timestamp wins
if isinstance(v, dict) and v.get("t", 0) >= mine.get(k, {}).get("t", 0):
mine[k] = {"pos": v.get("pos", 0), "dur": v.get("dur", 0), "t": v.get("t", 0)}
write(F_PROGRESS, allp)
return self.json(200, mine)
self.send_error(404)
def user_list(self):
prog = read(F_PROGRESS, {})
out = []
for name, rec in sorted(read(F_USERS, {}).items()):
p = prog.get(name, {})
out.append({"name": name, "admin": bool(rec.get("admin")), "books": len(p), "last": max((v.get("t", 0) for v in p.values()), default=0)})
return out
def admin_api(self, me, path, body):
users = read(F_USERS, {})
bad = lambda msg, code=400: self.json(code, {"error": msg})
if path == "/admin/users": # create
name = str(body.get("username", "")).strip().lower()
pw = str(body.get("password", ""))
if not USERNAME.match(name):
return bad("Username: 1-32 characters, letters, numbers, . _ -")
if name in users:
return bad("That username is taken.")
if len(pw) < 6:
return bad("Password must be at least 6 characters.")
users[name] = hash_pw(pw)
if body.get("admin"):
users[name]["admin"] = True
write(F_USERS, users, private=True)
return self.json(200, {"ok": True})
m = re.match(r"^/admin/users/([^/]+)/(password|admin|delete)$", path)
if not m:
return self.send_error(404)
name, action = unquote(m.group(1)).lower(), m.group(2)
if name not in users:
return bad("No such user.", 404)
others_admin = any(r.get("admin") for n, r in users.items() if n != name)
if action == "password":
pw = str(body.get("password", ""))
if len(pw) < 6:
return bad("Password must be at least 6 characters.")
users[name] = {**hash_pw(pw), **({"admin": True} if users[name].get("admin") else {})}
write(F_USERS, users, private=True)
revoke_sessions(name) # signs them out everywhere
elif action == "admin":
flag = bool(body.get("admin"))
if not flag and not others_admin:
return bad("There must be at least one admin.")
users[name]["admin"] = flag
write(F_USERS, users, private=True)
else: # delete
if name == me:
return bad("You can't delete your own account.")
users.pop(name)
write(F_USERS, users, private=True)
revoke_sessions(name)
prog = read(F_PROGRESS, {})
if prog.pop(name, None) is not None:
write(F_PROGRESS, prog)
self.json(200, {"ok": True})
def login(self, body):
ip = self.client_address[0]
now = time.time()
recent = [t for t in self.fails.get(ip, []) if now - t < 60]
if len(recent) >= 8:
return self.json(429, {"error": "slow down"})
name = str(body.get("username", "")).strip().lower()
rec = read(F_USERS, {}).get(name)
ok = bool(rec) and check_pw(str(body.get("password", "")), rec)
if not ok:
self.fails[ip] = recent + [now]
return self.json(401, {"error": "wrong"})
token = secrets.token_urlsafe(32)
with lock:
sessions = {k: v for k, v in read(F_SESSIONS, {}).items() if v["exp"] > now}
sessions[sha(token)] = {"user": name, "exp": now + SESSION_DAYS * 86400}
write(F_SESSIONS, sessions, private=True)
cookie = f"{COOKIE}={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age={SESSION_DAYS * 86400}"
self.json(200, {"user": name}, [("Set-Cookie", cookie)])
def logout(self):
c = SimpleCookie(self.headers.get("Cookie", ""))
if COOKIE in c:
with lock:
s = read(F_SESSIONS, {})
s.pop(sha(c[COOKIE].value), None)
write(F_SESSIONS, s, private=True)
self.send_response(302)
self.send_header("Set-Cookie", f"{COOKIE}=; Path=/; Max-Age=0")
self.send_header("Location", "/login")
self.end_headers()
def proxy(self, head=False):
path = self.path.split("?")[0]
m = AUDIO_PATH.match(path)
name = unquote(m.group(1)) if m else ""
oa_user, oa_pass = oa_login()
parts = urllib.parse.urlsplit(self.path) # drop the player's own cache-busting parameter before forwarding
q = [kv for kv in urllib.parse.parse_qsl(parts.query, keep_blank_values=True) if kv[0] != "_r"]
up = parts.path + ("?" + urllib.parse.urlencode(q) if q else "")
req = urllib.request.Request(self.upstream + up, method="HEAD" if head else "GET")
if oa_user:
cred = base64.b64encode(f"{oa_user}:{oa_pass}".encode()).decode()
req.add_header("Authorization", "Basic " + cred)
for h in PASS_HEADERS:
if self.headers.get(h):
req.add_header(h, self.headers[h])
resp = None
for attempt in (1, 2): # one quick retry for a flaky connection
try:
resp = urllib.request.urlopen(req, timeout=60 if m else 30)
break
except urllib.error.HTTPError as e:
resp = e
break
except Exception as e:
if attempt == 2:
log("upstream unreachable:", self.path, e)
return self.json(502, {"error": f"can't reach OpenAudible: {e}"})
time.sleep(0.4)
code = getattr(resp, "status", None) or resp.code
if code == 401: # the server-side OpenAudible login is wrong; not the listener's fault
return self.json(502, {"error": "OpenAudible login rejected; run: python3 server.py setup"})
if code >= 400 and m:
log("upstream", code, "for", self.command, path, "range:", self.headers.get("Range"))
h = resp.headers
out = [(k, h[k]) for k in RESP_HEADERS if h.get(k)]
skip, length = 0, None
if m:
# iPhones refuse to play audio unless the server answers byte-range requests.
ctype = (h.get("Content-Type") or "").lower()
if not ctype or ctype.startswith(("application/octet-stream", "binary/", "text/")):
ext = os.path.splitext(name)[1].lower()
if ext in AUDIO_TYPES:
out = [(k, v) for k, v in out if k != "Content-Type"] + [("Content-Type", AUDIO_TYPES[ext])]
if not any(k == "Accept-Ranges" for k, _ in out):
out.append(("Accept-Ranges", "bytes"))
rng = self.headers.get("Range")
cl = h.get("Content-Length")
total = int(cl) if cl and cl.isdigit() else None
if rng and code == 200 and not h.get("Content-Range") and not head:
# OpenAudible ignored the Range header: answer it ourselves by skipping ahead in the stream.
mm = RANGE.match(rng.strip())
if mm and total:
a, b = mm.groups()
if a == "" and b != "":
start, end = max(0, total - int(b)), total - 1
else:
start, end = int(a or 0), min(int(b), total - 1) if b else total - 1
if start >= total:
self.send_response(416)
self.send_header("Content-Range", f"bytes */{total}")
self.send_header("Content-Length", "0")
self.end_headers()
return resp.close()
code, skip, length = 206, start, end - start + 1
out = [(k, v) for k, v in out if k != "Content-Length"] + [
("Content-Range", f"bytes {start}-{end}/{total}"), ("Content-Length", str(length))]
if start:
log("emulating Range for", name, f"(from byte {start}; OpenAudible doesn't support Range)")
self.send_response(code)
for k, v in out:
self.send_header(k, v)
self.end_headers()
if head:
return resp.close()
try:
while skip > 0:
d = resp.read(min(65536, skip))
if not d:
return
skip -= len(d)
while length is None or length > 0:
d = resp.read(65536 if length is None else min(65536, length))
if not d:
break
self.wfile.write(d)
if length is not None:
length -= len(d)
except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError, TimeoutError, OSError):
pass # player seeked, closed the stream, or the connection dropped
finally:
resp.close()
def log_message(self, *a):
pass
def serve(argv):
ap = argparse.ArgumentParser()
ap.add_argument("--upstream", default=os.environ.get("OPENAUDIBLE_URL", "http://172.16.16.106:8000"))
ap.add_argument("--port", type=int, default=int(os.environ.get("PORT", 8080)))
ap.add_argument("--host", default="0.0.0.0")
a = ap.parse_args(argv)
users = read(F_USERS, {})
du, dp = os.environ.get("PLAYER_USER", "").strip().lower(), os.environ.get("PLAYER_PASSWORD", "")
if du and dp and du not in users: # default account for containers; never overwrites an existing one
users[du] = hash_pw(dp)
write(F_USERS, users, private=True)
if len(users) == 1 and os.path.exists(F_LEGACY):
legacy = read(F_LEGACY, {})
if legacy and all(isinstance(v, dict) and "pos" in v for v in legacy.values()):
write(F_PROGRESS, {du: legacy})
print(f"Created default account '{du}'.")
if du in users and not users[du].get("admin"): # the default account is always an admin
users[du]["admin"] = True
write(F_USERS, users, private=True)
if users and not any(r.get("admin") for r in users.values()): # accounts from before roles existed
first = next(iter(users))
users[first]["admin"] = True
write(F_USERS, users, private=True)
print(f"No admin found; made '{first}' an admin.")
if not users:
sys.exit("No accounts yet. Run: python3 server.py adduser NAME (or set PLAYER_USER and PLAYER_PASSWORD)")
if not oa_login()[0]:
print("Note: no OpenAudible login set. Run `python3 server.py setup` or set OPENAUDIBLE_USER/OPENAUDIBLE_PASSWORD.")
Handler.upstream = a.upstream.rstrip("/")
print(f"Player: http://localhost:{a.port} (OpenAudible at {Handler.upstream})")
ThreadingHTTPServer((a.host, a.port), Handler).serve_forever()
if __name__ == "__main__":
args = sys.argv[1:]
if args and args[0] in ("setup", "adduser", "passwd", "deluser", "users"):
if args[0] in ("adduser", "passwd", "deluser") and len(args) < 2:
sys.exit(f"Usage: python3 server.py {args[0]} NAME")
cli(args)
elif args and args[0] == "check":
check(args[1:])
elif args and args[0] in ("-h", "--help"):
print(__doc__)
else:
serve(args[1:] if args and args[0] == "serve" else args)