Extension transport

Drive the user’s everyday Chrome through a companion extension 
from fastcore.test import test_eq, ExceptionExpected
import httpx

source

ExtChannel

def ExtChannel():

Accept an extension connection and exchange JSON frames in both directions.

The channel exchanges JSON dictionaries:

request assigns each request an id and waits for its matching reply. Frames without a pending request go into events. ExtCDP ignores frames without a method, including keepalive pings.

Configure a shared, nonempty token for normal use. The extension must present it in the connection URL. The listener rejects websocket peers with a missing or incorrect token. Without this check, another process could connect as the extension, receive commands, or supply false replies. The channel connects to the user’s normal browser. Reserve unauthenticated listeners for disposable tests.

The examples below use a fake extension and never connect to Chrome. It represents tab 99 and echoes command methods back in its replies. This lets us inspect what the channel sent.

def _fake_reply(m):
    "Canned reply for one command frame"
    if m.get('action')=='new-tab': return dict(tabId=99)
    if m.get('action')=='get-targets': return [dict(type='page', tabId=99, url=m.get('url'))]
    if m.get('action')=='close-tab': return {}
    return dict(method=m['method'], tabId=m.get('tabId'))

fake_ext answers requests with _fake_reply. After Page.enable, it emits a Page.loadEventFired event. This simulates the event that a real page emits when it 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)
                await ws.send(json.dumps(dict(id=m['id'], result=_fake_reply(m))))
                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 websocket peer with the wrong token disconnects with code 4003 before exchanging command frames:

bad = await websockets.connect(f'ws://localhost:{chan.port}/?token=wrong')
with ExceptionExpected(websockets.ConnectionClosed, '4003'): await asyncio.wait_for(bad.recv(), 5)

Before opening a websocket, the extension probes the port with HTTP fetch. A listening ExtChannel returns status 200 and fastcdp\n. The probe avoids repeated websocket connection errors in the extension’s console while the listener is unavailable.

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:str=None, # WebSocket URL, populated by connection helpers
    debug:bool=False, # Log protocol events
    command_timeout:float=10, # Seconds before an unanswered command raises TimeoutError
):

Exchange CDP messages with a companion extension over a JSON channel.

ExtCDP.connect uses an already connected channel. ExtCDP.listen starts a local listener and waits for an extension.

Fastcdp addresses CDP sessions with sid. For this transport, sid is the browser’s tab id. _send converts outgoing sessionId fields to tabId. _pump_events converts incoming tabId fields back to sessionId. Inherited helpers can therefore address extension tabs with their existing sid argument.

_action requests tab operations through the extension’s chrome.tabs and chrome.debugger APIs. These use action frames rather than CDP command frames. They include opening, attaching to, listing, and closing tabs.

A channel needs request, events, close, and is_open. It doesn’t need to inherit from ExtChannel. Dialoghelper’s Channel supplies this interface for solvecdp through a solveit relay.

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

source

ExtCDP.attach_page

async def attach_page(
    tid:int | str
):

Attach to a tab by integer tabId or hex target id from pages. Return an ExtPage.

source

ExtCDP.new_page

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

Open a browser tab and return its ExtPage.

source

ExtPage

def ExtPage(
    cdp:fastcdp.core.CDP, t:str, sid:str, owned:bool=False, frame_id:str=None
):

Control a browser tab through the extension’s Page interface.

new_page opens a tab. attach_page connects to an existing tab by integer tabId or hexadecimal target id from pages. Both return an ExtPage, which supports the inherited Page helpers. Closing this page closes the browser tab, including a tab that already existed before attachment.

Subscriptions use the same event format as other CDP connections. Here, Page.loadEventFired reports the fake tab’s id as 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))
EVT: Page.loadEventFired sid=99

source

ExtCDP.active_page

async def active_page():

A Page driving the frontmost tab: the active tab of the last-focused window

source

ExtCDP.pages

async def pages()->__main__.ExtTargets: # Rows with `tabId`, `id`, `title`, `url`, ...; `*` marks debugger-attached

The browser’s open tabs

source

ExtTargets

def ExtTargets(
    *args, **kwargs
):

Extension tab rows, one line per tab

pages lists open browser tabs. In the display, * marks tabs with a debugger attached. active_page attaches to the active tab in the last-focused browser window.

page.close() closes its tab but keeps the connection open. cdp.close() cancels event tasks and closes the channel and its listener. It leaves the browser and any remaining tabs open. This example closes the fake tab before disconnecting:

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