"""SAB verification: confirm SharedArrayBuffer + Atomics.wait work in a
COOP+COEP Worker context.
This is the go/no-go gate for the SAB channel substrate used by spawn().
If this fails, blocking channel_recv must be replaced with a MessagePort
async fallback - the entire channel ABI changes.
The signer no longer runs in a browser extension; it is an in-page WASM
Worker. The Firefox MV2 extension SAB test has been removed - that question
(SAB in extension origins) is no longer relevant to the architecture.
Run:
python3 -m pytest test/test_sab_verify.py -v
"""
import http.server
import os
import socketserver
import sys
import tempfile
import threading
import time
import pytest
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# HTML + worker JS served with COOP/COEP for the Chrome web-page proxy test.
_SAB_PAGE_HTML = b"""
running...
"""
_SAB_WORKER_JS = b"""
const r = {
crossOriginIsolated: self.crossOriginIsolated,
hasSABType: typeof SharedArrayBuffer !== 'undefined',
canCreate: false,
atomicsWait: null,
error: null
};
try {
const sab = new SharedArrayBuffer(4);
r.canCreate = true;
const i32 = new Int32Array(sab);
r.atomicsWait = Atomics.wait(i32, 0, 0, 0);
} catch(e) {
r.error = e.message;
}
self.postMessage(r);
"""
class _CoopHandler(http.server.BaseHTTPRequestHandler):
def log_message(self, *args): pass
def do_GET(self):
if self.path == "/" or self.path == "/index.html":
body, ctype = _SAB_PAGE_HTML, b"text/html"
elif self.path == "/worker.js":
body, ctype = _SAB_WORKER_JS, b"application/javascript"
else:
self.send_response(404); self.end_headers(); return
self.send_response(200)
self.send_header("Content-Type", ctype.decode())
self.send_header("Content-Length", str(len(body)))
self.send_header("Cross-Origin-Opener-Policy", "same-origin")
self.send_header("Cross-Origin-Embedder-Policy", "require-corp")
self.end_headers()
self.wfile.write(body)
def _start_coop_server():
srv = socketserver.TCPServer(("127.0.0.1", 0), _CoopHandler)
t = threading.Thread(target=srv.serve_forever, daemon=True)
t.start()
return srv, srv.server_address[1]
# ── Chrome test (Playwright + COOP/COEP web page) ─────────────────────────
class TestChromeSAB:
"""Chrome: SAB in a Worker on a COOP+COEP page.
The Chrome MV3 extension manifest declares `cross_origin_opener_policy`
and `cross_origin_embedder_policy` with value "same-origin" / "require-corp".
These keys apply those headers to all extension pages, producing the same
crossOriginIsolated=true context as a web page served with those headers.
This test validates that Chrome grants SAB+Atomics.wait in Workers under
that condition, which is the proxy for what the offscreen document will see.
"""
def test_sab_in_coop_worker(self):
"""SharedArrayBuffer and Atomics.wait work in a Chrome Worker under COOP+COEP."""
from playwright.sync_api import sync_playwright
srv, port = _start_coop_server()
try:
with sync_playwright() as pw:
browser = pw.chromium.launch(headless=True, args=["--no-sandbox"])
page = browser.new_page()
page.goto(f"http://127.0.0.1:{port}/", wait_until="domcontentloaded")
page.wait_for_function("window.__sabResult !== undefined", timeout=8000)
result = page.evaluate("window.__sabResult")
browser.close()
finally:
srv.shutdown()
assert result is not None, "No result from Worker"
assert result.get("error") is None, f"Worker error: {result.get('error')}"
assert result.get("crossOriginIsolated") is True, \
f"crossOriginIsolated not true in Chrome COOP+COEP Worker: {result}"
assert result.get("canCreate") is True, \
f"SharedArrayBuffer creation failed in Chrome COOP+COEP Worker: {result}"
assert result.get("atomicsWait") in ("timed-out", "ok"), \
f"Atomics.wait did not work in Chrome COOP+COEP Worker: {result}"
# The Firefox MV2 extension SAB test was removed. The signer now runs as an
# in-page WASM Worker, not a browser extension. SAB availability in extension
# origins is no longer a constraint for this codebase.