#!/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 [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.request
from http.cookies import SimpleCookie
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
# 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 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", "")
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())
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())
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("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(sorted(users)) or "(no users yet; run: python3 server.py adduser NAME)")
# ---------- 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()
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})
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_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 == "/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 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):
oa_user, oa_pass = oa_login()
req = urllib.request.Request(self.upstream + self.path)
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])
try:
resp = urllib.request.urlopen(req, timeout=30)
except urllib.error.HTTPError as e:
resp = e
except Exception as e:
return self.json(502, {"error": f"can't reach OpenAudible: {e}"})
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"})
self.send_response(code)
for h in RESP_HEADERS:
if resp.headers.get(h):
self.send_header(h, resp.headers[h])
self.end_headers()
try:
while chunk := resp.read(64 * 1024):
self.wfile.write(chunk)
except (BrokenPipeError, ConnectionResetError):
pass # player seeked or closed the stream
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)
if not read(F_USERS, {}):
sys.exit("No accounts yet. Run: python3 server.py adduser YOURNAME")
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] in ("-h", "--help"):
print(__doc__)
else:
serve(args[1:] if args and args[0] == "serve" else args)