Extension transport

Drive the user’s everyday Chrome through a companion extension

from fastcore.test import test_eq
import httpx

---

source

ExtChannel

def ExtChannel():

Listen for the extension to dial in, then speak duplex JSON frames with it

Frames are JSON dicts, in four kinds. Commands carry a correlation id plus method/params (and tabId for tab-scoped ones); the extension answers each with {id, result} or {id, error}. Lifecycle actions ({id, action}: new-tab, attach, get-targets, detach, close-tab) cover operations chrome.debugger doesn’t expose as CDP. Debugger events arrive as {method, params, tabId} with no id, and keepalive pings have neither id nor method and are ignored by consumers. request handles the correlation; anything that isn’t a reply lands in events.

token is the trust check: the listener hands full browser control to whoever dials in, so outside of throwaway settings the extension should be configured with a shared secret, and only matching dials are accepted.

To exercise everything without a browser, a scripted stand-in plays the extension: it dials the listener and answers frames the way the real one would, faking a single tab 99. Method frames are echoed back so tests can see exactly what arrived, and enabling the Page domain emits one debugger event, as the real extension does when a page loads.

async def fake_ext(url):
    "A stand-in extension: one fake tab 99, canned replies, an event on `Page.enable`"
    try:
        async with websockets.connect(url) as ws:
            async for frame in ws:
                m = json.loads(frame)
                if m.get('action')=='new-tab': r = dict(tabId=99)
                elif m.get('action')=='get-targets': r = [dict(type='page', tabId=99, url=m.get('url'))]
                elif m.get('action')=='close-tab': r = {}
                else: r = dict(method=m['method'], tabId=m.get('tabId'))
                await ws.send(json.dumps(dict(id=m['id'], result=r)))
                if m.get('method')=='Page.enable':
                    await ws.send(json.dumps(dict(method='Page.loadEventFired', params={}, tabId=m['tabId'])))
    except websockets.ConnectionClosed: pass

chan = await ExtChannel.listen(port=0, token='s3cret')
fake = asyncio.create_task(fake_ext(f'ws://localhost:{chan.port}/?token=s3cret'))
await chan.wait_peer(5)
assert chan.is_open

A dial with the wrong token is refused before any frame flows:

bad = await websockets.connect(f'ws://localhost:{chan.port}/?token=wrong')
try: await asyncio.wait_for(bad.recv(), 5); raise AssertionError('expected close')
except websockets.ConnectionClosed as e: test_eq(e.rcvd.code, 4003)

The listener also answers plain HTTP: the extension probes the port with a cheap fetch before dialing the websocket, so a port with nothing listening never spams its console with failed connections.

async with httpx.AsyncClient() as c: r = await c.get(f'http://localhost:{chan.port}/')
test_eq((r.status_code, r.text), (200, 'fastcdp\n'))

---

source

ExtCDP

def ExtCDP(
    wsconn:NoneType=None, debug:bool=False
):

CDP spoken with a companion extension, as JSON frames over a channel to it

connect wraps any connected channel; listen is the common case of waiting for the extension locally. _send swaps fastcdp’s sessionId for the extension’s tabId on the way out, and _pump_events swaps it back on the way in, so every inherited helper that threads sid works against tabs directly. _action sends the lifecycle frames the extension performs with the chrome.tabs/chrome.debugger APIs: opening, attaching, listing, and closing tabs.

The chan given to connect just needs the ExtChannel surface (request, events, close, is_open), so other transports plug in without subclassing ceremony: dialoghelper’s Channel is one, which is how solvecdp drives the extension through a solveit relay.

---

source

ExtCDP.pages

async def pages():

Call self as a function.

---

source

ExtCDP.attach_page

async def attach_page(
    tid
):

Attach to an existing tab and return a Page driving it

---

source

ExtCDP.new_page

async def new_page(
    url:str='about:blank', active:bool=False
):

Open a new browser tab and return a Page driving it

---

source

ExtPage

def ExtPage(
    cdp, t, sid, owned:bool=False
):

A Page whose tab is closed through the extension

new_page, attach_page, and pages mirror fastcdp’s tab lifecycle through extension actions, returning Page proxies so goto, eval, ax_tree, and friends all work unchanged. Continuing with the fake extension from above: a command sent with a sid reaches the extension with that tabId:

cdp = await ExtCDP.connect(chan)
r = await cdp('Runtime.evaluate', sid=99, expression='1')
test_eq((r['method'], r['tabId']), ('Runtime.evaluate', 99))

Debugger events cross as {method, params, tabId} frames and land in the inherited subscription machinery, with the tabId restored to a sessionId:

async with cdp.on('Page.loadEventFired') as q:
    page = await cdp.new_page()
    e = await asyncio.wait_for(q.get(), 5)
test_eq((page.t, e['sessionId']), (99, 99))

pages lists the extension’s targets, and closing tears down the tab, the pump, and the listener:

test_eq((await cdp.pages)[0]['type'], 'page')
await page.close()
await cdp.close()
await fake

---

source

cdp_yolo

def cdp_yolo():

Allow all CDP classes in safepyrun, including the extension transport