xocko/plexamp-lite
Private
Public Access
1
0 Fork 0
Code Activity
Files
a4b875ef34074710ad258eeb22fa969462b9e926
plexamp-lite/test/mock_plex.py
T
xockoandClaude Opus 5.5 a4b875ef34 Minamp 1.2 
- Transcoding: send client identities Plex has a profile for (unknown ones get
  400 "unable to find a matching profile"), with a fresh session id per variant
- Stream proxy: fetch transcodes once into a temp file and serve every player
  request from it, so reconnects no longer restart the transcode (503s, tracks
  cut short); original files still pass Range requests through
- Sign-in: remember the pending link code so it survives Minamp being closed
  while the code is entered elsewhere; add "I've entered it - check now"
- Diagnostics: in-app connection log (DiagLog) with a streaming test and save
  to SD card; tokens are never logged

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

260 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)
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.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