xocko/plexamp-lite
Private
Public Access
1
0 Fork 0
Code Activity
Files
read-ahead-2
plexamp-lite/test/mock_plex.py
T
XoséandClaude Opus 5.5 7eebb0d881 Plexamp Lite 1.11: Minamp is now Plexamp Lite 
Renames the app (label, texts, log files, Plex product name, User-Agent, APK). The package
name, preferences and signing key stay, so it installs over Minamp and keeps its settings.
tools/strings.py catches up with the strings added straight to the XML in 9440536.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-19 16:57:47 +02:00

269 lines
13 KiB
Python
Raw Permalink Blame History

"""Mock plex.tv + Plex Media Server for exercising Plexamp Lite'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, "Plexamp Lite 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