Private
Public Access
xockoandClaude Opus 5.5
0a2ac901bd
- Run network/TLS work on 256 KB-stack threads (net/Threads.java): Android 2.x's ~16 KB default stack overflowed during the bundled TLS handshake - Close pooled TLS connections off the caller thread - Try several Plex music-transcoder URL variants until the server accepts one, and remember the one that worked - Accept pasted server URLs with paths; default to 443 for https - Catch fatal errors and save a crash report (shown on next launch and written to Minamp-error.txt on the SD card) Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
237 lines
11 KiB
Python
237 lines
11 KiB
Python
"""Mock plex.tv + Plex Media Server for exercising Minamp's networking code on a desktop JVM.
|
|
|
|
Serves:
|
|
https://plex.tv:443 pins, user, resources
|
|
https://127-0-0-1.abc123.plex.direct:32443 PMS over TLS (wildcard cert, ECDSA leaf, RSA root)
|
|
http://127.0.0.1:32400 the same PMS in plain HTTP
|
|
"""
|
|
import datetime
|
|
import json
|
|
import os
|
|
import ssl
|
|
import sys
|
|
import threading
|
|
import time
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import urlparse, parse_qs
|
|
|
|
from cryptography import x509
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec, rsa
|
|
from cryptography.x509.oid import NameOID
|
|
|
|
OUT = sys.argv[1]
|
|
os.makedirs(OUT, exist_ok=True)
|
|
NOW = datetime.datetime.now(datetime.timezone.utc)
|
|
|
|
|
|
def name(cn):
|
|
return x509.Name([x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Minamp Test"), x509.NameAttribute(NameOID.COMMON_NAME, cn)])
|
|
|
|
|
|
def cert(subject, issuer, pub, signer, hashalg, ca, sans=None):
|
|
b = (x509.CertificateBuilder().subject_name(subject).issuer_name(issuer).public_key(pub)
|
|
.serial_number(x509.random_serial_number())
|
|
.not_valid_before(NOW - datetime.timedelta(days=1)).not_valid_after(NOW + datetime.timedelta(days=30))
|
|
.add_extension(x509.BasicConstraints(ca=ca, path_length=None), critical=True))
|
|
if sans:
|
|
b = b.add_extension(x509.SubjectAlternativeName([x509.DNSName(s) for s in sans]), critical=False)
|
|
return b.sign(signer, hashalg)
|
|
|
|
|
|
root_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
root = cert(name("Test Root"), name("Test Root"), root_key.public_key(), root_key, hashes.SHA256(), True)
|
|
inter_key = ec.generate_private_key(ec.SECP384R1())
|
|
inter = cert(name("Test ECC Intermediate"), root.subject, inter_key.public_key(), root_key, hashes.SHA256(), True)
|
|
pms_key = ec.generate_private_key(ec.SECP256R1())
|
|
pms = cert(name("*.abc123.plex.direct"), inter.subject, pms_key.public_key(), inter_key, hashes.SHA384(), False,
|
|
["*.abc123.plex.direct"])
|
|
tv_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
|
|
tv = cert(name("plex.tv"), inter.subject, tv_key.public_key(), inter_key, hashes.SHA384(), False, ["plex.tv", "*.plex.tv"])
|
|
|
|
pem = lambda c: c.public_bytes(serialization.Encoding.PEM)
|
|
keypem = lambda k: k.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())
|
|
open(os.path.join(OUT, "root.pem"), "wb").write(pem(root))
|
|
for label, c, k in (("pms", pms, pms_key), ("tv", tv, tv_key)):
|
|
open(os.path.join(OUT, label + "-chain.pem"), "wb").write(pem(c) + pem(inter))
|
|
open(os.path.join(OUT, label + "-key.pem"), "wb").write(keypem(k))
|
|
|
|
MACHINE = "machine-abc123"
|
|
TOKEN = "server-token-xyz"
|
|
ACCOUNT = "account-token-123"
|
|
pins = {}
|
|
log = []
|
|
|
|
TRACKS = [
|
|
{"type": "track", "ratingKey": "101", "key": "/library/metadata/101", "title": "Song One", "grandparentTitle": "The Testers",
|
|
"parentTitle": "First Album", "parentThumb": "/library/metadata/10/thumb/1", "parentRatingKey": "10", "grandparentRatingKey": "1",
|
|
"index": 1, "duration": 200000, "userRating": 10,
|
|
"Media": [{"audioCodec": "mp3", "container": "mp3", "bitrate": 192, "Part": [{"key": "/library/parts/501/1/file.mp3", "size": 300000}]}]},
|
|
{"type": "track", "ratingKey": "102", "key": "/library/metadata/102", "title": "Song Two (FLAC)", "grandparentTitle": "The Testers",
|
|
"parentTitle": "First Album", "parentThumb": "/library/metadata/10/thumb/1", "parentRatingKey": "10", "grandparentRatingKey": "1",
|
|
"index": 2, "duration": 180000,
|
|
"Media": [{"audioCodec": "flac", "container": "flac", "bitrate": 900, "Part": [{"key": "/library/parts/502/1/file.flac", "size": 9000000}]}]},
|
|
]
|
|
MP3 = bytes((i * 7) % 256 for i in range(300000))
|
|
|
|
|
|
class H(BaseHTTPRequestHandler):
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def log_message(self, *a):
|
|
pass
|
|
|
|
def send_json(self, obj, code=200):
|
|
body = json.dumps(obj).encode()
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_POST(self):
|
|
self.route("POST")
|
|
|
|
def do_PUT(self):
|
|
self.route("PUT")
|
|
|
|
def do_GET(self):
|
|
self.route("GET")
|
|
|
|
def do_HEAD(self):
|
|
self.route("HEAD")
|
|
|
|
def route(self, method):
|
|
n = int(self.headers.get("Content-Length") or 0)
|
|
if n:
|
|
self.rfile.read(n)
|
|
u = urlparse(self.path)
|
|
q = parse_qs(u.query)
|
|
p = u.path
|
|
host = self.headers.get("Host", "")
|
|
tok = self.headers.get("X-Plex-Token") or (q.get("X-Plex-Token") or [None])[0]
|
|
log.append((method, host, self.path, tok, self.headers.get("Range")))
|
|
if host.startswith("plex.tv"):
|
|
return self.tv(method, p, tok)
|
|
if p == "/identity":
|
|
return self.send_json({"MediaContainer": {"machineIdentifier": MACHINE, "version": "1.40"}})
|
|
if tok != TOKEN:
|
|
return self.send_json({"error": "unauthorized"}, 401)
|
|
if p == "/library/sections":
|
|
return self.send_json({"MediaContainer": {"Directory": [
|
|
{"key": "1", "type": "artist", "title": "Music"}, {"key": "2", "type": "movie", "title": "Films"}]}})
|
|
if p == "/library/sections/1/all":
|
|
t = q.get("type", ["8"])[0]
|
|
if t == "8":
|
|
start = int(q.get("X-Plex-Container-Start", ["0"])[0])
|
|
size = int(q.get("X-Plex-Container-Size", ["50"])[0])
|
|
arts = [{"type": "artist", "ratingKey": str(1000 + i), "title": "Artist %03d" % i, "thumb": "/a/%d" % i} for i in range(250)]
|
|
return self.send_json({"MediaContainer": {"totalSize": 250, "offset": start, "Metadata": arts[start:start + size]}})
|
|
if t == "10":
|
|
if "userRating>>" in q or any(k.endswith(">") for k in q):
|
|
return self.send_json({"MediaContainer": {"Metadata": TRACKS[:1], "filterSeen": list(q.keys())}})
|
|
return self.send_json({"MediaContainer": {"Metadata": TRACKS}})
|
|
return self.send_json({"MediaContainer": {"Metadata": [{"type": "album", "ratingKey": "10", "title": "First Album",
|
|
"parentTitle": "The Testers", "year": 1999, "thumb": "/library/metadata/10/thumb/1"}]}})
|
|
if p == "/library/metadata/10/children":
|
|
return self.send_json({"MediaContainer": {"Metadata": TRACKS}})
|
|
if p == "/hubs/search":
|
|
return self.send_json({"MediaContainer": {"Hub": [
|
|
{"type": "artist", "title": "Artists", "Directory": [{"type": "artist", "ratingKey": "1", "title": "The Testers"}]},
|
|
{"type": "track", "title": "Tracks", "Metadata": TRACKS[:1]}]}})
|
|
if p.startswith("/library/parts/501"):
|
|
return self.file(method, MP3)
|
|
if p in ("/music/:/transcode/universal/start.mp3", "/audio/:/transcode/universal/start.mp3"):
|
|
extra = (q.get("X-Plex-Client-Profile-Extra") or [""])[0]
|
|
if "scopeName=*" in extra: # pretend this server rejects that variant
|
|
body = b"<html><head><title>Bad Request</title></head><body><h1>400 Bad Request</h1></body></html>"
|
|
self.send_response(400)
|
|
self.send_header("Content-Type", "text/html")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
return self.wfile.write(body)
|
|
return self.chunked(method, q)
|
|
if p == "/photo/:/transcode":
|
|
return self.file(method, b"\xff\xd8\xff\xe0JPEGDATA" * 50, "image/jpeg")
|
|
if p in ("/:/timeline", "/:/scrobble", "/:/rate", "/transcode/universal/stop"):
|
|
return self.send_json({})
|
|
self.send_json({"error": "not found " + p}, 404)
|
|
|
|
def file(self, method, data, ctype="audio/mpeg"):
|
|
rng = self.headers.get("Range")
|
|
start, end = 0, len(data) - 1
|
|
code = 200
|
|
if rng and rng.startswith("bytes="):
|
|
a, b = rng[6:].split("-")
|
|
start = int(a)
|
|
end = int(b) if b else len(data) - 1
|
|
code = 206
|
|
chunk = data[start:end + 1]
|
|
self.send_response(code)
|
|
self.send_header("Content-Type", ctype)
|
|
self.send_header("Accept-Ranges", "bytes")
|
|
self.send_header("Content-Length", str(len(chunk)))
|
|
if code == 206:
|
|
self.send_header("Content-Range", "bytes %d-%d/%d" % (start, end, len(data)))
|
|
self.end_headers()
|
|
if method != "HEAD":
|
|
self.wfile.write(chunk)
|
|
|
|
def chunked(self, method, q):
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "audio/mpeg")
|
|
self.send_header("Transfer-Encoding", "chunked")
|
|
self.end_headers()
|
|
if method == "HEAD":
|
|
return
|
|
offset = q.get("offset", ["0"])[0]
|
|
payload = ("TRANSCODE offset=%s bitrate=%s|" % (offset, q.get("musicBitrate", ["?"])[0])).encode() * 400
|
|
for i in range(0, len(payload), 1000):
|
|
part = payload[i:i + 1000]
|
|
self.wfile.write(b"%x\r\n" % len(part) + part + b"\r\n")
|
|
self.wfile.flush()
|
|
self.wfile.write(b"0\r\n\r\n")
|
|
|
|
def tv(self, method, p, tok):
|
|
if p == "/api/v2/pins" and method == "POST":
|
|
pid = len(pins) + 1
|
|
pins[pid] = {"id": pid, "code": "WXYZ", "authToken": None, "created": time.time()}
|
|
return self.send_json(pins[pid], 201)
|
|
if p.startswith("/api/v2/pins/"):
|
|
pin = pins.get(int(p.rsplit("/", 1)[1]))
|
|
if pin and time.time() - pin["created"] > 1.0:
|
|
pin["authToken"] = ACCOUNT # "user linked it" after a second
|
|
return self.send_json(pin or {}, 200 if pin else 404)
|
|
if tok != ACCOUNT:
|
|
return self.send_json({"error": "unauthorized"}, 401)
|
|
if p == "/api/v2/user":
|
|
return self.send_json({"username": "tester"})
|
|
if p == "/api/v2/resources":
|
|
return self.send_json([
|
|
{"name": "Living Room PC", "provides": "server", "clientIdentifier": MACHINE, "accessToken": TOKEN, "owned": True,
|
|
"connections": [
|
|
{"protocol": "https", "address": "10.255.255.1", "port": 32400, "uri": "https://10-255-255-1.abc123.plex.direct:32400", "local": False, "relay": False},
|
|
{"protocol": "https", "address": "127.0.0.1", "port": 32443, "uri": "https://127-0-0-1.abc123.plex.direct:32443", "local": True, "relay": False},
|
|
{"protocol": "https", "address": "1.2.3.4", "port": 8443, "uri": "https://1-2-3-4.abc123.plex.direct:8443", "local": False, "relay": True},
|
|
]},
|
|
{"name": "Some TV", "provides": "player", "clientIdentifier": "tv"},
|
|
])
|
|
self.send_json({"error": "nf"}, 404)
|
|
|
|
|
|
def serve(port, chain=None, key=None):
|
|
srv = ThreadingHTTPServer(("127.0.0.1", port), H)
|
|
if chain:
|
|
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
|
|
ctx.load_cert_chain(chain, key)
|
|
srv.socket = ctx.wrap_socket(srv.socket, server_side=True)
|
|
threading.Thread(target=srv.serve_forever, daemon=True).start()
|
|
|
|
|
|
serve(32400)
|
|
serve(32443, os.path.join(OUT, "pms-chain.pem"), os.path.join(OUT, "pms-key.pem"))
|
|
serve(443, os.path.join(OUT, "tv-chain.pem"), os.path.join(OUT, "tv-key.pem"))
|
|
print("ready", flush=True)
|
|
while True:
|
|
time.sleep(1)
|