"""A small mock of the Audiobookshelf 2.x API, shaped like the real server's replies. Usage: python3 mock_abs.py PORT [legacy] legacy: /login returns only the old long-lived user.token (pre-2.26 servers). Test-only endpoints: /__expire (invalidate the access token), /__revoke (invalidate the refresh token too), /__log (requests seen so far). """ import base64 import json import sys import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer PORT = int(sys.argv[1]) LEGACY = len(sys.argv) > 2 and sys.argv[2] == "legacy" lock = threading.Lock() state = {"access": None, "refresh": None, "n": 0, "log": [], "progress": {}, "refreshes": 0} def b64(d): return base64.urlsafe_b64encode(json.dumps(d).encode()).rstrip(b"=").decode() def new_tokens(): state["n"] += 1 now = int(time.time()) state["access"] = b64({"alg": "HS256"}) + "." + b64({"userId": "u1", "iat": now, "exp": now + 3600, "n": state["n"]}) + ".sig" state["refresh"] = "refresh-%d" % state["n"] def episode(i, item="pod1", long_desc=False): desc = ("

Show notes café \U0001F3A7 " + "x" * (5000 if long_desc else 50) + "

") return { "libraryItemId": item, "podcastId": "p-" + item, "id": "ep%d" % i, "index": i, "title": "Episode %d – ¿Qué tal?" % i, "subtitle": "", "description": desc, "publishedAt": 1700000000000 + i * 86400000, "pubDate": "x", "chapters": [{"id": c, "start": c * 60, "end": c * 60 + 60, "title": "ch"} for c in range(30)], "enclosure": {"url": "http://feed/ep%d.mp3" % i, "type": "audio/mpeg", "length": "123"}, "audioFile": {"index": 1, "ino": "ino%d" % i, "duration": 1800.5 + i, "mimeType": "audio/mpeg", "metadata": {"filename": "ep%d.mp3" % i, "ext": ".mp3", "size": 30000000 + i}, "metaTags": {"tagAlbum": "x"}}, } def podcast_media(n_eps, item="pod1"): return {"id": "p-" + item, "metadata": {"title": "Mini Show", "author": "Ana Álvarez"}, "episodes": [episode(i, item, long_desc=True) for i in range(n_eps)]} class H(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def log_message(self, *a): pass def send(self, code, body=None, ctype="application/json"): data = b"" if body is not None: data = (json.dumps(body) if ctype == "application/json" else body).encode() self.send_response(code) self.send_header("Content-Type", ctype + "; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Connection", "close") self.end_headers() self.wfile.write(data) def body(self): n = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(n) if n else b"" return json.loads(raw) if raw else None def authed(self): auth = self.headers.get("Authorization", "") tok = auth[7:] if auth.startswith("Bearer ") else None if tok is None and "token=" in self.path: tok = self.path.split("token=")[1].split("&")[0] return tok is not None and tok == state["access"] def handle_any(self, method): path = self.path.split("?")[0] b = self.body() if method in ("POST", "PATCH") else None with lock: state["log"].append({"m": method, "p": self.path, "b": b, "rt": self.headers.get("x-return-tokens"), "xr": self.headers.get("x-refresh-token")}) if path == "/__log": return self.send(200, state["log"]) if path == "/__refreshes": return self.send(200, {"n": state["refreshes"]}) if path == "/__expire": state["access"] = "expired" return self.send(200, {}) if path == "/__revoke": state["access"] = "expired" state["refresh"] = "revoked" return self.send(200, {}) if path == "/status": return self.send(200, {"app": "audiobookshelf", "serverVersion": "2.30.0", "isInit": True, "authMethods": ["local"]}) if path == "/login" and method == "POST": if not b or b.get("username") != "demo" or b.get("password") != "pw": return self.send(401, "Unauthorized", "text/plain") new_tokens() user = {"id": "u1", "username": "demo", "token": state["access"] if LEGACY else "legacy-tok"} if LEGACY: user["token"] = state["access"] elif self.headers.get("x-return-tokens") == "true": user["accessToken"] = state["access"] user["refreshToken"] = state["refresh"] return self.send(200, {"user": user, "userDefaultLibraryId": "lib-pod", "serverSettings": {}}) if path == "/auth/refresh" and method == "POST": if self.headers.get("x-refresh-token") != state["refresh"]: return self.send(401, "Unauthorized", "text/plain") state["refreshes"] += 1 new_tokens() return self.send(200, {"user": {"id": "u1", "accessToken": state["access"], "refreshToken": state["refresh"]}}) if not self.authed(): # Real ABS answers 401 without a WWW-Authenticate header return self.send(401, "Unauthorized", "text/plain") if path == "/api/libraries": return self.send(200, {"libraries": [ {"id": "lib-book", "name": "Books", "mediaType": "book"}, {"id": "lib-pod", "name": "Podcasts", "mediaType": "podcast"}]}) if path == "/api/libraries/lib-pod/recent-episodes": eps = [] for i in range(20): e = episode(100 + i, "pod%d" % (i % 3)) e["podcast"] = podcast_media(40, "pod%d" % (i % 3)) # the real server nests the whole podcast e["podcast"]["libraryItem"] = {"id": "pod%d" % (i % 3), "libraryId": "lib-pod"} eps.append(e) return self.send(200, {"episodes": eps, "limit": 20, "page": 0}) if path == "/api/me/items-in-progress": return self.send(200, {"libraryItems": [ {"id": "pod1", "libraryId": "lib-pod", "mediaType": "podcast", "media": {"metadata": {"title": "Mini Show", "author": "Ana"}, "numEpisodes": 300}, "recentEpisode": episode(7), "progressLastUpdate": 1}, {"id": "book9", "libraryId": "lib-book", "mediaType": "book", "media": {"metadata": {"title": "B"}}}]}) if path == "/api/me": mp = [{"libraryItemId": "pod1", "episodeId": "ep7", "currentTime": 600.0, "duration": 1807.5, "progress": 0.33, "isFinished": False}, {"libraryItemId": "pod1", "episodeId": "ep8", "currentTime": 1808.5, "duration": 1808.5, "progress": 1, "isFinished": True}, {"libraryItemId": "book9", "episodeId": None, "currentTime": 5, "isFinished": False}] return self.send(200, {"id": "u1", "username": "demo", "mediaProgress": mp, "bookmarks": []}) if path == "/api/libraries/lib-pod/items": res = [{"id": "pod%d" % i, "libraryId": "lib-pod", "media": {"metadata": {"title": "Show %d" % i, "author": "A"}, "numEpisodes": i}} for i in range(3)] return self.send(200, {"results": res, "total": 3, "limit": 100, "page": 0}) if path == "/api/items/pod1": item = {"id": "pod1", "libraryId": "lib-pod", "media": podcast_media(300), "libraryFiles": [{"ino": "x%d" % i, "metadata": {"path": "/p" * 20}} for i in range(300)]} return self.send(200, item) if path.startswith("/api/items/pod1/play/") and method == "POST": return self.send(200, {"id": "sess1", "libraryItemId": "pod1", "episodeId": path.split("/")[-1], "currentTime": 600.0, "duration": 1807.5, "playMethod": 0, "audioTracks": [{"index": 1, "contentUrl": "/api/items/pod1/file/ino7", "mimeType": "audio/mpeg", "duration": 1807.5}]}) if path.startswith("/api/session/") and method == "POST": return self.send(200, "OK", "text/plain") if path.startswith("/api/me/progress/") and method == "PATCH": state["progress"][path] = b return self.send(200, "OK", "text/plain") return self.send(404, "Not Found", "text/plain") def do_GET(self): self.handle_any("GET") def do_POST(self): self.handle_any("POST") def do_PATCH(self): self.handle_any("PATCH") ThreadingHTTPServer(("127.0.0.1", PORT), H).serve_forever()