brfc-stream-pirate/discover.py

#!/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()