xocko/plexamp-lite
Private
Public Access
1
0 Fork 0
Code Activity
Files
3caabd942c9f1e0e8aeacbd6811b1ba48ed9e538
plexamp-lite/test/mock_plex.py
T
xockoandClaude Opus 5.5 58d21b6d68 Minamp 1.3 
- Fix NoSuchMethodError on Android 2.x: Dalvik before 4.0 can't link Object
  methods (toString/equals/...) called through an interface. Patch the bundled
  Bouncy Castle (tools/patch_bc_dalvik.py) and our own CertVerifier, and make
  build.sh fail if tools/scan_dalvik.py finds any such call
- Treat a manually entered private address (10.x, 172.16-31.x, 192.168.x,
  .local) as home, so LAN servers that allow access without sign-in work
- Send Plex client headers on media requests (files and transcodes)
- Explain a 503 on a file as the server being unable to open it (drive or
  folder not available)
- Connection log: home/away, token presence, Plex's file check and a direct
  file request in the streaming test

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-14 20:56:27 +02:00

269 lines
13 KiB
Python
Raw Blame History

"""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 = []
starts = []
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, keep_blank_values=True)
p = u.path
host = self.headers.get("Host", "")
tok = self.headers.get("X-Plex-Token") or (q.get("X-Plex-Token") or [None])[0]
empty_token_param = "X-Plex-Token" in q and q["X-Plex-Token"][0] == "" and not self.headers.get("X-Plex-Token")
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"}})
# Like a server with "allowed without auth" for its LAN: no token at all is fine from here,
# but media requests carrying an *empty* token are refused (503 files / 400 transcoder).
lan_no_auth = tok is None and self.server.server_address[1] == 32400
if empty_token_param:
if p.startswith("/library/parts/"):
return self.send_json({"error": "Service Unavailable"}, 503)
if "/transcode/" in p:
return self.send_json({"error": "Bad Request"}, 400)
if tok != TOKEN and not lan_no_auth:
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.endswith("/transcode/universal/decision"):
if not self.profile_ok(q):
return self.send_json({"error": "unable to find a matching profile"}, 400)
return self.send_json({"MediaContainer": {"generalDecisionCode": 1001, "generalDecisionText": "Direct play not available; Conversion OK.",
"transcodeDecisionCode": 1001, "transcodeDecisionText": "Transcode OK."}})
if p in ("/music/:/transcode/universal/start.mp3", "/audio/:/transcode/universal/start.mp3", "/audio/:/transcode/universal/start"):
# like a real server: unknown client identity -> 400; a session started twice -> 503
if not self.profile_ok(q):
body = b"<html><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)
sess = (q.get("session") or [""])[0]
starts.append(sess)
if starts.count(sess) > 1:
return self.send_json({"error": "session busy"}, 503)
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 profile_ok(self, q):
platform = (q.get("X-Plex-Platform") or [""])[0]
device = (q.get("X-Plex-Device") or [""])[0]
return platform in ("Chrome", "Linux") and device in ("", "Sonos")
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)
View Git Blame Copy Permalink