180 lines
8.7 KiB
Python
180 lines
8.7 KiB
Python
"""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 = ("<p>Show notes café \U0001F3A7 " + "x" * (5000 if long_desc else 50) + "</p>")
|
||
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()
|