204 lines
7.2 KiB
Python
204 lines
7.2 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Drive the player with Playwright and capture the traffic that mints a token.
|
||
|
|
|
||
|
|
Purpose: find which request produces a stream URL, so the mint can eventually
|
||
|
|
run without a GUI. Playwright sees request AND response bodies natively, so
|
||
|
|
this replaces the userscript's hooks with something far less fragile.
|
||
|
|
|
||
|
|
No password is needed. Your existing Firefox session cookies are injected into
|
||
|
|
a fresh browser context, so this rides the login you already have.
|
||
|
|
|
||
|
|
python discover.py --url "<match page url>"
|
||
|
|
python discover.py --url "..." --headless
|
||
|
|
|
||
|
|
Ports to Camoufox on halogen by swapping the launcher: Camoufox speaks the
|
||
|
|
Playwright protocol, so the flow below is unchanged.
|
||
|
|
"""
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import re
|
||
|
|
import sys
|
||
|
|
from datetime import datetime
|
||
|
|
|
||
|
|
ROOT = os.path.dirname(os.path.abspath(__file__))
|
||
|
|
LOGS = os.path.join(ROOT, "logs")
|
||
|
|
TRACE = os.path.join(LOGS, "pw-trace.jsonl")
|
||
|
|
|
||
|
|
# Only these domains' cookies are handed to the browser. Everything else in
|
||
|
|
# the Firefox jar is irrelevant here and should not be copied around.
|
||
|
|
COOKIE_DOMAINS = ("bristolrovers", "streamamg", "kaltura", "akamaized")
|
||
|
|
|
||
|
|
# Bodies we never want to read: media segments would be dumped into the log.
|
||
|
|
SKIP_BODY = (".ts", ".mp4", ".m4s", ".aac", ".jpg", ".jpeg", ".png", ".gif",
|
||
|
|
".webp", ".svg", ".woff", ".woff2", ".css", ".ico")
|
||
|
|
|
||
|
|
MAX_BODY = 20000
|
||
|
|
|
||
|
|
|
||
|
|
def firefox_cookies():
|
||
|
|
"""Lift the live session from Firefox. Requires no password."""
|
||
|
|
from yt_dlp.cookies import extract_cookies_from_browser
|
||
|
|
import io, contextlib
|
||
|
|
buf = io.StringIO()
|
||
|
|
with contextlib.redirect_stderr(buf):
|
||
|
|
jar = extract_cookies_from_browser("firefox")
|
||
|
|
|
||
|
|
out = []
|
||
|
|
for c in jar:
|
||
|
|
dom = (c.domain or "")
|
||
|
|
if not any(k in dom.lower() for k in COOKIE_DOMAINS):
|
||
|
|
continue
|
||
|
|
out.append({
|
||
|
|
"name": c.name,
|
||
|
|
"value": c.value or "",
|
||
|
|
"domain": dom,
|
||
|
|
"path": c.path or "/",
|
||
|
|
"expires": float(c.expires) if c.expires else -1,
|
||
|
|
"httpOnly": False,
|
||
|
|
"secure": bool(c.secure),
|
||
|
|
"sameSite": "Lax",
|
||
|
|
})
|
||
|
|
return out
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
ap = argparse.ArgumentParser()
|
||
|
|
ap.add_argument("--url", required=True, help="match page URL")
|
||
|
|
ap.add_argument("--headless", action="store_true",
|
||
|
|
help="run without a window (what halogen will do)")
|
||
|
|
ap.add_argument("--seconds", type=int, default=90,
|
||
|
|
help="how long to watch after load")
|
||
|
|
args = ap.parse_args()
|
||
|
|
|
||
|
|
try:
|
||
|
|
from playwright.sync_api import sync_playwright
|
||
|
|
except ImportError:
|
||
|
|
sys.exit("playwright not installed: pip install playwright "
|
||
|
|
"&& python -m playwright install firefox")
|
||
|
|
|
||
|
|
os.makedirs(LOGS, exist_ok=True)
|
||
|
|
cookies = firefox_cookies()
|
||
|
|
print(f"injecting {len(cookies)} session cookies")
|
||
|
|
|
||
|
|
manifests, mints, jwts, seen = [], [], [], set()
|
||
|
|
fh = open(TRACE, "a", encoding="utf-8")
|
||
|
|
|
||
|
|
def write(rec):
|
||
|
|
rec["ts"] = datetime.now().isoformat()
|
||
|
|
fh.write(json.dumps(rec, ensure_ascii=False) + chr(10))
|
||
|
|
fh.flush()
|
||
|
|
|
||
|
|
with sync_playwright() as p:
|
||
|
|
browser = p.firefox.launch(headless=args.headless)
|
||
|
|
ctx = browser.new_context(
|
||
|
|
user_agent=("Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:131.0) "
|
||
|
|
"Gecko/20100101 Firefox/131.0"),
|
||
|
|
viewport={"width": 1440, "height": 900},
|
||
|
|
)
|
||
|
|
if cookies:
|
||
|
|
try:
|
||
|
|
ctx.add_cookies(cookies)
|
||
|
|
except Exception as exc:
|
||
|
|
print(f"cookie injection failed: {exc}")
|
||
|
|
|
||
|
|
page = ctx.new_page()
|
||
|
|
|
||
|
|
def on_response(resp):
|
||
|
|
url = resp.url
|
||
|
|
if url in seen:
|
||
|
|
return
|
||
|
|
low = url.split("?")[0].lower()
|
||
|
|
if ".m3u8" in low:
|
||
|
|
seen.add(url)
|
||
|
|
manifests.append(url)
|
||
|
|
print(f" manifest: {url[:120]}")
|
||
|
|
write({"kind": "manifest", "url": url, "status": resp.status})
|
||
|
|
return
|
||
|
|
if any(low.endswith(e) for e in SKIP_BODY):
|
||
|
|
return
|
||
|
|
try:
|
||
|
|
body = resp.text()
|
||
|
|
except Exception:
|
||
|
|
return
|
||
|
|
if not body:
|
||
|
|
return
|
||
|
|
# The JWT that authorises the mint is issued somewhere on the club
|
||
|
|
# site off the session cookie. Detect it the same way: by output.
|
||
|
|
# A JWT is three base64 segments starting with a b64 '{"alg"'.
|
||
|
|
if re.search(r"eyJ[A-Za-z0-9_-]{8,}\.eyJ[A-Za-z0-9_-]{8,}\.", body):
|
||
|
|
seen.add(url)
|
||
|
|
req = resp.request
|
||
|
|
rec = {
|
||
|
|
"kind": "JWT-ISSUER",
|
||
|
|
"method": req.method,
|
||
|
|
"url": url,
|
||
|
|
"status": resp.status,
|
||
|
|
"reqHeaders": dict(req.headers),
|
||
|
|
"reqBody": (req.post_data or None),
|
||
|
|
"body": body[:MAX_BODY],
|
||
|
|
}
|
||
|
|
jwts.append(rec)
|
||
|
|
write(rec)
|
||
|
|
print(f" *** JWT ISSUER *** {req.method} {url[:100]}")
|
||
|
|
return
|
||
|
|
|
||
|
|
# The mint call is whatever RETURNS a manifest URL, whatever it is
|
||
|
|
# named. Detect by output, not by guessing endpoint names.
|
||
|
|
if "m3u8" in body.lower():
|
||
|
|
seen.add(url)
|
||
|
|
req = resp.request
|
||
|
|
rec = {
|
||
|
|
"kind": "MINT",
|
||
|
|
"method": req.method,
|
||
|
|
"url": url,
|
||
|
|
"status": resp.status,
|
||
|
|
"reqHeaders": dict(req.headers),
|
||
|
|
"reqBody": (req.post_data or None),
|
||
|
|
"body": body[:MAX_BODY],
|
||
|
|
}
|
||
|
|
mints.append(rec)
|
||
|
|
write(rec)
|
||
|
|
print(f" *** MINT *** {req.method} {url[:110]}")
|
||
|
|
|
||
|
|
page.on("response", on_response)
|
||
|
|
|
||
|
|
print(f"loading {args.url}")
|
||
|
|
page.goto(args.url, wait_until="domcontentloaded", timeout=60000)
|
||
|
|
|
||
|
|
# The player needs a click. Muted playback is enough to trigger the
|
||
|
|
# manifest fetch, and muted autoplay is not blocked.
|
||
|
|
page.wait_for_timeout(5000)
|
||
|
|
for sel in (".bmpui-ui-hugeplaybacktogglebutton",
|
||
|
|
".bmpui-ui-playbacktogglebutton",
|
||
|
|
"button[aria-label*='play' i]"):
|
||
|
|
try:
|
||
|
|
el = page.query_selector(sel)
|
||
|
|
if el and el.is_visible():
|
||
|
|
el.click()
|
||
|
|
print(f"clicked {sel}")
|
||
|
|
break
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
try:
|
||
|
|
page.evaluate("() => { const v = document.querySelector('video');"
|
||
|
|
" if (v) { v.muted = true; v.play(); } }")
|
||
|
|
except Exception:
|
||
|
|
pass
|
||
|
|
|
||
|
|
print(f"watching for {args.seconds}s ...")
|
||
|
|
page.wait_for_timeout(args.seconds * 1000)
|
||
|
|
browser.close()
|
||
|
|
|
||
|
|
fh.close()
|
||
|
|
print(f"\nmanifests seen : {len(manifests)}")
|
||
|
|
print(f"mint calls : {len(mints)}")
|
||
|
|
for m in mints:
|
||
|
|
print(f"\n {m['method']} {m['url']}")
|
||
|
|
print(f" status {m['status']} reqBody={str(m['reqBody'])[:200]}")
|
||
|
|
print(f"\nfull trace -> {TRACE}")
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|