#!/usr/bin/env python3 from __future__ import annotations import argparse import asyncio import base64 import contextlib import http.server import json import os import re import select import shutil import socket import subprocess import sys import tempfile import threading import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from pathlib import Path from typing import Any import websockets DEFAULT_PUBLIC_CANARY_SITES: list[tuple[str, str]] = [ ("sannysoft", "https://bot.sannysoft.com/"), ("antoine-vastel-areyouheadless", "https://arh.antoinevastel.com/bots/areyouheadless"), ("intoli-headless", "https://intoli.com/blog/not-possible-to-block-chrome-headless/chrome-headless-test.html"), ("fpscanner", "https://fpscanner.com/demo"), ("amiunique", "https://amiunique.org/fingerprint"), ] LAUNCH_ARG_REDACTIONS = { "--use-angle=swiftshader": "--use-angle=", "--enable-unsafe-swiftshader": "--enable-unsafe-", } SMOKE_PAGE_HTML = """ Cultguard detection smoke

Cultguard detection smoke

running probes…

""" WIDEVINE_PROBE_EXPRESSION = r""" (async () => { const serialiseError = (error) => { if (!error) { return null; } return { name: error.name || 'Error', message: error.message || String(error) }; }; if (!navigator.requestMediaKeySystemAccess) { return { apiPresent: false, supported: false, keySystem: 'com.widevine.alpha', error: 'navigator.requestMediaKeySystemAccess unavailable' }; } const configs = [{ initDataTypes: ['cenc'], audioCapabilities: [{ contentType: 'audio/mp4; codecs="mp4a.40.2"' }], videoCapabilities: [{ contentType: 'video/mp4; codecs="avc1.42E01E"' }], sessionTypes: ['temporary'], persistentState: 'optional', distinctiveIdentifier: 'optional' }]; try { const access = await navigator.requestMediaKeySystemAccess('com.widevine.alpha', configs); const result = { apiPresent: true, supported: true, keySystem: access.keySystem || 'com.widevine.alpha', configuration: access.getConfiguration ? access.getConfiguration() : null, mediaKeysCreated: false, sessionCreated: false }; try { const mediaKeys = access.createMediaKeys ? await access.createMediaKeys() : null; result.mediaKeysCreated = !!mediaKeys; result.mediaKeysType = mediaKeys ? Object.prototype.toString.call(mediaKeys) : null; if (mediaKeys && mediaKeys.createSession) { const session = mediaKeys.createSession('temporary'); result.sessionCreated = !!session; result.sessionType = session ? Object.prototype.toString.call(session) : null; result.sessionId = session ? session.sessionId : null; } } catch (error) { result.supported = false; result.mediaKeysError = serialiseError(error); } return result; } catch (error) { return { apiPresent: true, supported: false, keySystem: 'com.widevine.alpha', error: serialiseError(error) }; } })() """ SMOKE_PROBE_EXPRESSION = r""" (async () => { const serialiseError = (error) => { if (!error) { return null; } return { name: error.name || 'Error', message: error.message || String(error) }; }; const collectPermissions = async () => { if (!navigator.permissions || !navigator.permissions.query) { return { unsupported: true }; } try { const notifications = await navigator.permissions.query({ name: 'notifications' }); return { state: notifications.state }; } catch (error) { return { error: String(error) }; } }; const collectWebGl = () => { const canvas = document.createElement('canvas'); const gl = canvas.getContext('webgl') || canvas.getContext('experimental-webgl'); if (!gl) { return { supported: false }; } const getParameterSafe = (parameter) => { try { const value = gl.getParameter(parameter); if (Array.isArray(value)) { return value; } if (ArrayBuffer.isView(value)) { return Array.from(value); } return value; } catch (error) { return { error: String(error) }; } }; const getShaderPrecisionSafe = (shaderType, precisionType) => { try { const format = gl.getShaderPrecisionFormat(shaderType, precisionType); if (!format) { return null; } return { rangeMin: format.rangeMin, rangeMax: format.rangeMax, precision: format.precision }; } catch (error) { return { error: String(error) }; } }; const debugInfo = gl.getExtension('WEBGL_debug_renderer_info'); const anisotropyExt = gl.getExtension('EXT_texture_filter_anisotropic') || gl.getExtension('WEBKIT_EXT_texture_filter_anisotropic') || gl.getExtension('MOZ_EXT_texture_filter_anisotropic'); const parameters = { version: getParameterSafe(gl.VERSION), shadingLanguageVersion: getParameterSafe(gl.SHADING_LANGUAGE_VERSION), vendorMasked: getParameterSafe(gl.VENDOR), rendererMasked: getParameterSafe(gl.RENDERER), maxCombinedTextureImageUnits: getParameterSafe(gl.MAX_COMBINED_TEXTURE_IMAGE_UNITS), maxCubeMapTextureSize: getParameterSafe(gl.MAX_CUBE_MAP_TEXTURE_SIZE), maxFragmentUniformVectors: getParameterSafe(gl.MAX_FRAGMENT_UNIFORM_VECTORS), maxRenderbufferSize: getParameterSafe(gl.MAX_RENDERBUFFER_SIZE), maxTextureImageUnits: getParameterSafe(gl.MAX_TEXTURE_IMAGE_UNITS), maxTextureSize: getParameterSafe(gl.MAX_TEXTURE_SIZE), maxVaryingVectors: getParameterSafe(gl.MAX_VARYING_VECTORS), maxVertexAttribs: getParameterSafe(gl.MAX_VERTEX_ATTRIBS), maxVertexTextureImageUnits: getParameterSafe(gl.MAX_VERTEX_TEXTURE_IMAGE_UNITS), maxVertexUniformVectors: getParameterSafe(gl.MAX_VERTEX_UNIFORM_VECTORS), maxViewportDims: getParameterSafe(gl.MAX_VIEWPORT_DIMS), aliasedLineWidthRange: getParameterSafe(gl.ALIASED_LINE_WIDTH_RANGE), aliasedPointSizeRange: getParameterSafe(gl.ALIASED_POINT_SIZE_RANGE), redBits: getParameterSafe(gl.RED_BITS), greenBits: getParameterSafe(gl.GREEN_BITS), blueBits: getParameterSafe(gl.BLUE_BITS), alphaBits: getParameterSafe(gl.ALPHA_BITS), depthBits: getParameterSafe(gl.DEPTH_BITS), stencilBits: getParameterSafe(gl.STENCIL_BITS), sampleBuffers: getParameterSafe(gl.SAMPLE_BUFFERS), samples: getParameterSafe(gl.SAMPLES), maxAnisotropy: anisotropyExt ? getParameterSafe(anisotropyExt.MAX_TEXTURE_MAX_ANISOTROPY_EXT) : null }; const shaderPrecision = { vertex: { highFloat: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.HIGH_FLOAT), mediumFloat: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.MEDIUM_FLOAT), lowFloat: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.LOW_FLOAT), highInt: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.HIGH_INT), mediumInt: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.MEDIUM_INT), lowInt: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.LOW_INT) }, fragment: { highFloat: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.HIGH_FLOAT), mediumFloat: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.MEDIUM_FLOAT), lowFloat: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.LOW_FLOAT), highInt: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.HIGH_INT), mediumInt: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.MEDIUM_INT), lowInt: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.LOW_INT) } }; return { supported: true, vendor: debugInfo ? gl.getParameter(debugInfo.UNMASKED_VENDOR_WEBGL) : null, renderer: debugInfo ? gl.getParameter(debugInfo.UNMASKED_RENDERER_WEBGL) : null, contextAttributes: gl.getContextAttributes ? gl.getContextAttributes() : null, extensions: (gl.getSupportedExtensions() || []).slice().sort(), parameters, shaderPrecision }; }; const collectMediaCodecs = () => { const video = document.createElement('video'); return { mp4: video.canPlayType('video/mp4; codecs="avc1.42E01E, mp4a.40.2"'), webm: video.canPlayType('video/webm; codecs="vp8, vorbis"'), hls: video.canPlayType('application/vnd.apple.mpegURL') }; }; const collectWidevine = async () => ({ skipped: 'widevine is probed separately on a localhost page context' }); const collectIframe = async () => { return await new Promise((resolve) => { let finished = false; const done = (value) => { if (finished) { return; } finished = true; resolve(value); }; const iframe = document.createElement('iframe'); iframe.style.display = 'none'; iframe.srcdoc = 'iframe smoke'; iframe.onload = () => { try { const cw = iframe.contentWindow; done({ contentWindowPresent: !!cw, selfEquals: !!cw && cw.self === cw, frameElementMatches: !!cw && cw.frameElement === iframe, navigatorLanguage: cw ? cw.navigator.language : null, locationHref: cw ? cw.location.href : null }); } catch (error) { done({ error: String(error) }); } finally { iframe.remove(); } }; document.body.appendChild(iframe); setTimeout(() => { iframe.remove(); done({ timeout: true }); }, 3000); }); }; const collectUserAgentData = async () => { if (!navigator.userAgentData) { return null; } const data = { brands: navigator.userAgentData.brands || [], mobile: navigator.userAgentData.mobile, platform: navigator.userAgentData.platform || null }; if (navigator.userAgentData.getHighEntropyValues) { try { data.highEntropy = await navigator.userAgentData.getHighEntropyValues([ 'architecture', 'bitness', 'fullVersionList', 'model', 'platformVersion' ]); } catch (error) { data.highEntropyError = String(error); } } return data; }; const collectSourceUrlProbe = () => { try { const fn = new Function('throw new Error("sourceurl-smoke");\n//# sourceURL=__cultguard_sourceurl_probe__'); fn(); return null; } catch (error) { return String(error.stack || ''); } }; return { locationHref: location.href, userAgent: navigator.userAgent, userAgentData: await collectUserAgentData(), webdriver: navigator.webdriver, vendor: navigator.vendor, platform: navigator.platform, languages: Array.from(navigator.languages || []), plugins: Array.from(navigator.plugins || []).map((plugin) => ({ name: plugin.name, filename: plugin.filename, description: plugin.description })), mimeTypes: Array.from(navigator.mimeTypes || []).map((mimeType) => ({ type: mimeType.type, suffixes: mimeType.suffixes, description: mimeType.description })), hardwareConcurrency: navigator.hardwareConcurrency, deviceMemory: navigator.deviceMemory ?? null, maxTouchPoints: navigator.maxTouchPoints ?? null, pdfViewerEnabled: navigator.pdfViewerEnabled ?? null, notificationPermission: typeof Notification === 'undefined' ? null : Notification.permission, permissions: await collectPermissions(), outerDimensions: { outerWidth, outerHeight, innerWidth, innerHeight, devicePixelRatio }, screen: { width: screen.width, height: screen.height, availWidth: screen.availWidth, availHeight: screen.availHeight }, chrome: { present: !!window.chrome, runtime: !!(window.chrome && window.chrome.runtime), app: !!(window.chrome && window.chrome.app), csi: !!(window.chrome && window.chrome.csi), loadTimes: !!(window.chrome && window.chrome.loadTimes) }, webgl: collectWebGl(), mediaCodecs: collectMediaCodecs(), widevine: await collectWidevine(), iframe: await collectIframe(), sourceUrlStack: collectSourceUrlProbe(), timestamp: new Date().toISOString() }; })() """ FALLBACK_SMOKE_PROBE_EXPRESSION = r""" (() => { const collectWebGl = () => { const canvas = document.createElement('canvas'); const gl = canvas.getContext('webgl') || canvas.getContext('experimental-webgl'); if (!gl) { return { supported: false }; } const getParameterSafe = (parameter) => { try { const value = gl.getParameter(parameter); if (Array.isArray(value)) { return value; } if (ArrayBuffer.isView(value)) { return Array.from(value); } return value; } catch (error) { return { error: String(error) }; } }; const getShaderPrecisionSafe = (shaderType, precisionType) => { try { const format = gl.getShaderPrecisionFormat(shaderType, precisionType); if (!format) { return null; } return { rangeMin: format.rangeMin, rangeMax: format.rangeMax, precision: format.precision }; } catch (error) { return { error: String(error) }; } }; const debugInfo = gl.getExtension('WEBGL_debug_renderer_info'); const anisotropyExt = gl.getExtension('EXT_texture_filter_anisotropic') || gl.getExtension('WEBKIT_EXT_texture_filter_anisotropic') || gl.getExtension('MOZ_EXT_texture_filter_anisotropic'); return { supported: true, vendor: debugInfo ? gl.getParameter(debugInfo.UNMASKED_VENDOR_WEBGL) : null, renderer: debugInfo ? gl.getParameter(debugInfo.UNMASKED_RENDERER_WEBGL) : null, contextAttributes: gl.getContextAttributes ? gl.getContextAttributes() : null, extensions: (gl.getSupportedExtensions() || []).slice().sort(), parameters: { version: getParameterSafe(gl.VERSION), shadingLanguageVersion: getParameterSafe(gl.SHADING_LANGUAGE_VERSION), vendorMasked: getParameterSafe(gl.VENDOR), rendererMasked: getParameterSafe(gl.RENDERER), maxCombinedTextureImageUnits: getParameterSafe(gl.MAX_COMBINED_TEXTURE_IMAGE_UNITS), maxCubeMapTextureSize: getParameterSafe(gl.MAX_CUBE_MAP_TEXTURE_SIZE), maxFragmentUniformVectors: getParameterSafe(gl.MAX_FRAGMENT_UNIFORM_VECTORS), maxRenderbufferSize: getParameterSafe(gl.MAX_RENDERBUFFER_SIZE), maxTextureImageUnits: getParameterSafe(gl.MAX_TEXTURE_IMAGE_UNITS), maxTextureSize: getParameterSafe(gl.MAX_TEXTURE_SIZE), maxVaryingVectors: getParameterSafe(gl.MAX_VARYING_VECTORS), maxVertexAttribs: getParameterSafe(gl.MAX_VERTEX_ATTRIBS), maxVertexTextureImageUnits: getParameterSafe(gl.MAX_VERTEX_TEXTURE_IMAGE_UNITS), maxVertexUniformVectors: getParameterSafe(gl.MAX_VERTEX_UNIFORM_VECTORS), maxViewportDims: getParameterSafe(gl.MAX_VIEWPORT_DIMS), aliasedLineWidthRange: getParameterSafe(gl.ALIASED_LINE_WIDTH_RANGE), aliasedPointSizeRange: getParameterSafe(gl.ALIASED_POINT_SIZE_RANGE), redBits: getParameterSafe(gl.RED_BITS), greenBits: getParameterSafe(gl.GREEN_BITS), blueBits: getParameterSafe(gl.BLUE_BITS), alphaBits: getParameterSafe(gl.ALPHA_BITS), depthBits: getParameterSafe(gl.DEPTH_BITS), stencilBits: getParameterSafe(gl.STENCIL_BITS), sampleBuffers: getParameterSafe(gl.SAMPLE_BUFFERS), samples: getParameterSafe(gl.SAMPLES), maxAnisotropy: anisotropyExt ? getParameterSafe(anisotropyExt.MAX_TEXTURE_MAX_ANISOTROPY_EXT) : null }, shaderPrecision: { vertex: { highFloat: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.HIGH_FLOAT), mediumFloat: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.MEDIUM_FLOAT), lowFloat: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.LOW_FLOAT), highInt: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.HIGH_INT), mediumInt: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.MEDIUM_INT), lowInt: getShaderPrecisionSafe(gl.VERTEX_SHADER, gl.LOW_INT) }, fragment: { highFloat: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.HIGH_FLOAT), mediumFloat: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.MEDIUM_FLOAT), lowFloat: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.LOW_FLOAT), highInt: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.HIGH_INT), mediumInt: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.MEDIUM_INT), lowInt: getShaderPrecisionSafe(gl.FRAGMENT_SHADER, gl.LOW_INT) } } }; }; const collectMediaCodecs = () => { const video = document.createElement('video'); return { mp4: video.canPlayType('video/mp4; codecs="avc1.42E01E, mp4a.40.2"'), webm: video.canPlayType('video/webm; codecs="vp8, vorbis"'), hls: video.canPlayType('application/vnd.apple.mpegURL') }; }; const collectWidevine = () => ({ skipped: 'dump-dom fallback skips async Widevine probing' }); const collectSourceUrlProbe = () => { try { const fn = new Function('throw new Error("sourceurl-smoke");\n//# sourceURL=__cultguard_sourceurl_probe__'); fn(); return null; } catch (error) { return String(error.stack || ''); } }; return { locationHref: location.href, userAgent: navigator.userAgent, userAgentData: navigator.userAgentData ? { brands: navigator.userAgentData.brands || [], mobile: navigator.userAgentData.mobile, platform: navigator.userAgentData.platform || null } : null, webdriver: navigator.webdriver, vendor: navigator.vendor, platform: navigator.platform, languages: Array.from(navigator.languages || []), plugins: Array.from(navigator.plugins || []).map((plugin) => ({ name: plugin.name, filename: plugin.filename, description: plugin.description })), mimeTypes: Array.from(navigator.mimeTypes || []).map((mimeType) => ({ type: mimeType.type, suffixes: mimeType.suffixes, description: mimeType.description })), hardwareConcurrency: navigator.hardwareConcurrency, deviceMemory: navigator.deviceMemory ?? null, maxTouchPoints: navigator.maxTouchPoints ?? null, pdfViewerEnabled: navigator.pdfViewerEnabled ?? null, notificationPermission: typeof Notification === 'undefined' ? null : Notification.permission, permissions: { skipped: 'dump-dom fallback skips async permissions probing' }, outerDimensions: { outerWidth, outerHeight, innerWidth, innerHeight, devicePixelRatio }, screen: { width: screen.width, height: screen.height, availWidth: screen.availWidth, availHeight: screen.availHeight }, chrome: { present: !!window.chrome, runtime: !!(window.chrome && window.chrome.runtime), app: !!(window.chrome && window.chrome.app), csi: !!(window.chrome && window.chrome.csi), loadTimes: !!(window.chrome && window.chrome.loadTimes) }, webgl: collectWebGl(), mediaCodecs: collectMediaCodecs(), widevine: collectWidevine(), iframe: { skipped: 'dump-dom fallback skips iframe.contentWindow probing' }, sourceUrlStack: collectSourceUrlProbe(), timestamp: new Date().toISOString() }; })() """ def make_launch_fallback_page() -> str: return """ Cultguard detection smoke (launch fallback)

pending

""" HEADER_PROBE_TEMPLATE = r""" (() => { const probeUrl = %URL%; fetch(probeUrl, { cache: 'no-store', credentials: 'omit', mode: 'no-cors' }).catch(() => null); return probeUrl; })() """ CDP_ERROR_PROBE_EXPRESSION = r""" (() => { window.__cultguardPrepareStackTraceHits = 0; Error.prepareStackTrace = function(error, frames) { window.__cultguardPrepareStackTraceHits += 1; return 'cultguard-custom-stack'; }; console.log(new Error('boom')); return true; })() """ @dataclass(frozen=True) class TargetSpec: label: str endpoint: str launched: bool = False metadata: dict[str, Any] | None = None class CDPError(RuntimeError): pass class CDPClient: def __init__(self, websocket_url: str) -> None: self.websocket_url = websocket_url self.websocket: Any | None = None self.reader_task: asyncio.Task[None] | None = None self.pending: dict[int, asyncio.Future[Any]] = {} self.events: asyncio.Queue[dict[str, Any]] = asyncio.Queue() self.next_id = 0 async def __aenter__(self) -> "CDPClient": self.websocket = await websockets.connect(self.websocket_url, max_size=None) self.reader_task = asyncio.create_task(self._reader()) return self async def __aexit__(self, exc_type, exc, tb) -> None: if self.reader_task is not None: self.reader_task.cancel() with contextlib.suppress(asyncio.CancelledError): await self.reader_task if self.websocket is not None: await self.websocket.close() async def _reader(self) -> None: assert self.websocket is not None async for raw_message in self.websocket: message = json.loads(raw_message) if "id" in message: future = self.pending.pop(message["id"], None) if future is None: continue if "error" in message: future.set_exception(CDPError(json.dumps(message["error"], sort_keys=True))) else: future.set_result(message.get("result")) continue await self.events.put(message) async def send(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]: if self.websocket is None: raise RuntimeError("CDP client is not connected") self.next_id += 1 message_id = self.next_id future: asyncio.Future[Any] = asyncio.get_running_loop().create_future() self.pending[message_id] = future await self.websocket.send(json.dumps({ "id": message_id, "method": method, "params": params or {}, })) result = await future return result async def wait_for_event( self, method: str, predicate=None, timeout: float = 5.0, ) -> dict[str, Any]: deadline = time.monotonic() + timeout while True: remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError(f"timed out waiting for {method}") event = await asyncio.wait_for(self.events.get(), timeout=remaining) if event.get("method") != method: continue if predicate is not None and not predicate(event): continue return event def log(message: str) -> None: print(f"[cg-detect] {message}", file=sys.stderr) def fail(message: str) -> None: raise SystemExit(message) def sanitize_label(value: str) -> str: return re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-") or "target" def make_artifacts_dir(path: str | None) -> Path: if path: artifacts_dir = Path(path).expanduser().resolve() artifacts_dir.mkdir(parents=True, exist_ok=True) return artifacts_dir tmp_base = os.environ.get("TMPDIR") or None return Path(tempfile.mkdtemp(prefix="cultguard-detect-", dir=tmp_base)) def write_json(path: Path, data: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") def write_text(path: Path, data: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(data, encoding="utf-8") def redact_launch_command(command: list[str]) -> list[str]: return [LAUNCH_ARG_REDACTIONS.get(arg, arg) for arg in command] def local_canary_fixture_specs() -> list[dict[str, str]]: return [ { "label": "local-sannysoft", "title": "Cultguard local canary — sannysoft-style", "description": "Curated in-repo approximation of broad browser-automation and plugin checks.", "script": """ const overview = startSection('Automation surface'); appendCheck(overview, !probe.webdriver, 'navigator.webdriver hidden', probe.webdriver); appendCheck(overview, !String(probe.userAgent || '').includes('HeadlessChrome'), 'HeadlessChrome absent from UA', probe.userAgent); appendRow(overview, 'navigator.languages', probe.languages); appendRow(overview, 'navigator.plugins', probe.plugins); appendRow(overview, 'window.chrome', probe.chrome); const rendering = startSection('Rendering and media'); appendRow(rendering, 'webgl', probe.webgl); appendRow(rendering, 'mediaCodecs', probe.mediaCodecs); appendRow(rendering, 'pdfViewerEnabled', probe.pdfViewerEnabled); """, }, { "label": "local-areyouheadless", "title": "Cultguard local canary — areyouheadless-style", "description": "Focused local heuristics for classic headless and automation signals.", "script": """ const heuristics = startSection('Headless heuristics'); appendCheck(heuristics, !probe.webdriver, 'navigator.webdriver hidden', probe.webdriver); appendCheck(heuristics, !String(probe.userAgent || '').includes('HeadlessChrome'), 'HeadlessChrome absent from UA', probe.userAgent); appendCheck(heuristics, !JSON.stringify(probe.userAgentData || {}).includes('HeadlessChrome'), 'HeadlessChrome absent from UA client hints', probe.userAgentData); appendCheck(heuristics, !!(probe.chrome && probe.chrome.present), 'window.chrome present', probe.chrome); appendCheck(heuristics, !!(probe.plugins || []).length, 'navigator.plugins populated', probe.plugins); const extrasSection = startSection('Extras'); appendRow(extrasSection, 'navigator.vendor', probe.vendor); appendRow(extrasSection, 'permissions', probe.permissions); appendRow(extrasSection, 'sourceURL first line', firstLine(probe.sourceUrlStack)); """, }, { "label": "local-intoli", "title": "Cultguard local canary — intoli-style", "description": "Local approximation of classic anti-headless heuristics and layout sanity checks.", "script": """ const basics = startSection('Classic anti-headless checks'); appendCheck(basics, !probe.webdriver, 'navigator.webdriver hidden', probe.webdriver); appendCheck(basics, !!(probe.languages || []).length, 'navigator.languages populated', probe.languages); appendCheck(basics, !!(probe.plugins || []).length, 'navigator.plugins populated', probe.plugins); appendCheck(basics, !!probe.vendor, 'navigator.vendor present', probe.vendor); appendCheck( basics, !!(probe.outerDimensions && probe.outerDimensions.outerWidth > 0 && probe.outerDimensions.outerHeight > 0), 'window.outer dimensions populated', probe.outerDimensions ); const iframeSection = startSection('Frame and stack sanity'); appendRow(iframeSection, 'iframe probe', probe.iframe); appendRow(iframeSection, 'sourceURL first line', firstLine(probe.sourceUrlStack)); appendRow(iframeSection, 'screen', probe.screen); """, }, { "label": "local-fpscanner", "title": "Cultguard local canary — fpscanner-style", "description": "Local high-entropy fingerprint surface summary for renderer, memory, and client hints.", "script": """ const highEntropy = startSection('High-entropy surface'); appendRow(highEntropy, 'navigator.userAgentData', probe.userAgentData); appendRow(highEntropy, 'hardwareConcurrency', probe.hardwareConcurrency); appendRow(highEntropy, 'deviceMemory', probe.deviceMemory); appendRow(highEntropy, 'maxTouchPoints', probe.maxTouchPoints); appendRow(highEntropy, 'timezone', extras.timezone); appendRow(highEntropy, 'colorScheme', extras.colorScheme); const graphics = startSection('Graphics and codecs'); appendRow(graphics, 'webgl', probe.webgl); appendRow(graphics, 'mediaCodecs', probe.mediaCodecs); appendRow(graphics, 'screen', probe.screen); """, }, { "label": "local-amiunique", "title": "Cultguard local canary — amiunique-style", "description": "Local fingerprint summary page for broad browser state comparisons without external traffic.", "script": """ const identity = startSection('Fingerprint summary'); appendRow(identity, 'userAgent', probe.userAgent); appendRow(identity, 'platform', probe.platform); appendRow(identity, 'language', extras.language); appendRow(identity, 'timezone', extras.timezone); appendRow(identity, 'doNotTrack', extras.doNotTrack); const browserState = startSection('Browser state'); appendRow(browserState, 'permissions', probe.permissions); appendRow(browserState, 'window.chrome', probe.chrome); appendRow(browserState, 'screen', probe.screen); appendRow(browserState, 'outerDimensions', probe.outerDimensions); appendRow(browserState, 'webgl', probe.webgl); """, }, ] def make_local_canary_page(title: str, description: str, render_script: str) -> str: return f""" {title}

{title}

{description}

collecting local canary signals…

""" def build_local_canary_pages() -> list[tuple[str, str, str]]: pages: list[tuple[str, str, str]] = [] for spec in local_canary_fixture_specs(): filename = f"{spec['label']}.html" pages.append(( spec["label"], filename, make_local_canary_page(spec["title"], spec["description"], spec["script"]), )) return pages def make_local_canary_index(pages: list[tuple[str, str, str]]) -> str: links = "\n".join( f'

{label}

' for label, filename, _ in pages ) return f""" Cultguard local canary fixtures

Cultguard local canary fixtures

These are curated in-repo approximations for the optional public comparison canaries.

{links}

""" @contextlib.contextmanager def local_canary_server(artifact_dir: Path) -> Any: pages = build_local_canary_pages() fixture_dir = artifact_dir / "local-canary-fixtures" fixture_dir.mkdir(parents=True, exist_ok=True) routes: dict[str, str] = { "/": make_local_canary_index(pages), "/index.html": make_local_canary_index(pages), } sites: list[tuple[str, str]] = [] requests: list[dict[str, Any]] = [] base_url = "" server_log_path = artifact_dir / "local-canary-server.json" for label, filename, body in pages: routes[f"/{filename}"] = body write_text(fixture_dir / filename, body) class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, format: str, *args: Any) -> None: # noqa: A003 return def do_GET(self) -> None: # noqa: N802 parsed = urllib.parse.urlparse(self.path) requests.append({ "method": "GET", "path": parsed.path, "headers": dict(self.headers.items()), }) if parsed.path == "/favicon.ico": self.send_response(204) self.end_headers() return body = routes.get(parsed.path or "/") if body is None: self.send_error(404) return payload = body.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) server_thread = threading.Thread(target=server.serve_forever, name="cultguard-local-canary", daemon=True) server_thread.start() try: base_url = f"http://127.0.0.1:{server.server_port}" sites = [ (label, f"{base_url}/{filename}") for label, filename, _ in pages ] yield { "sites": sites, "mode": "local-fixtures", "serverUrl": base_url, "serverLogPath": str(server_log_path), "fixtureDir": str(fixture_dir), } finally: server.shutdown() server.server_close() server_thread.join(timeout=5) write_json(server_log_path, { "mode": "local-fixtures", "serverUrl": base_url, "sites": sites, "requests": requests, "fixtureDir": str(fixture_dir), }) def make_data_url(html_text: str) -> str: encoded = base64.b64encode(html_text.encode("utf-8")).decode("ascii") return f"data:text/html;base64,{encoded}" @contextlib.contextmanager def local_probe_page_server(artifact_dir: Path, label: str, body: str) -> Any: requests: list[dict[str, Any]] = [] base_url = "" server_log_path = artifact_dir / f"{sanitize_label(label)}-server.json" class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, format: str, *args: Any) -> None: # noqa: A003 return def do_GET(self) -> None: # noqa: N802 parsed = urllib.parse.urlparse(self.path) requests.append({ "method": "GET", "path": parsed.path, "headers": dict(self.headers.items()), }) if parsed.path == "/favicon.ico": self.send_response(204) self.end_headers() return payload = body.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) server_thread = threading.Thread(target=server.serve_forever, name=f"cultguard-{sanitize_label(label)}", daemon=True) server_thread.start() try: base_url = f"http://127.0.0.1:{server.server_port}/" yield { "url": base_url, "serverLogPath": str(server_log_path), } finally: server.shutdown() server.server_close() server_thread.join(timeout=5) write_json(server_log_path, { "label": label, "serverUrl": base_url, "requests": requests, }) def fetch_json(url: str, timeout: float = 5.0) -> dict[str, Any]: request = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(request, timeout=timeout) as response: return json.loads(response.read().decode("utf-8")) def wait_for_devtools(endpoint: str, timeout: float) -> dict[str, Any]: deadline = time.monotonic() + timeout last_error: Exception | None = None while time.monotonic() < deadline: try: return fetch_json(f"{endpoint}/json/version", timeout=2.0) except Exception as error: # noqa: BLE001 last_error = error time.sleep(0.25) if last_error is None: raise TimeoutError(f"timed out waiting for {endpoint}") raise TimeoutError(f"timed out waiting for {endpoint}: {last_error}") def find_free_port() -> int: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock: sock.bind(("127.0.0.1", 0)) return sock.getsockname()[1] def make_isolated_browser_env(tmp_base: str | None, prefix: str) -> tuple[Path, dict[str, str]]: runtime_dir = Path(tempfile.mkdtemp(prefix=prefix, dir=tmp_base)) home_dir = runtime_dir / "home" config_dir = runtime_dir / "config" cache_dir = runtime_dir / "cache" state_dir = runtime_dir / "state" for path in (home_dir, config_dir, cache_dir, state_dir): path.mkdir(parents=True, exist_ok=True) env = os.environ.copy() env.update({ "HOME": str(home_dir), "XDG_CONFIG_HOME": str(config_dir), "XDG_CACHE_HOME": str(cache_dir), "XDG_STATE_HOME": str(state_dir), }) return runtime_dir, env def terminate_process(process: subprocess.Popen[Any], timeout: float = 10.0) -> None: if process.poll() is not None: return with contextlib.suppress(ProcessLookupError): process.terminate() try: process.wait(timeout=timeout) except subprocess.TimeoutExpired: with contextlib.suppress(ProcessLookupError): process.kill() process.wait(timeout=timeout) def wait_for_display_number( display_reader: Any, process: subprocess.Popen[Any], timeout: float = 5.0, ) -> str: deadline = time.monotonic() + timeout while time.monotonic() < deadline: if process.poll() is not None: raise RuntimeError( f"Xvfb exited before publishing a display number (exit={process.returncode})" ) ready, _, _ = select.select([display_reader], [], [], 0.2) if ready: display_number = display_reader.readline().strip() if display_number: return display_number raise TimeoutError("timed out waiting for Xvfb to publish a display number") @contextlib.contextmanager def virtual_display( artifacts_dir: Path, env: dict[str, str], *, log_name: str, ) -> Any: xvfb_binary = shutil.which("Xvfb") if not xvfb_binary: raise RuntimeError("Xvfb is not available in PATH; ensure the detect runtime includes xorg-server") stderr_path = artifacts_dir / log_name display_env = env.copy() for name in ("DISPLAY", "WAYLAND_DISPLAY", "XAUTHORITY"): display_env.pop(name, None) read_fd, write_fd = os.pipe() with os.fdopen(read_fd, "r", encoding="utf-8", errors="replace") as display_reader: with stderr_path.open("w", encoding="utf-8") as stderr_handle: process = subprocess.Popen( [ xvfb_binary, "-displayfd", str(write_fd), "-screen", "0", "1400x900x24", "-ac", "+extension", "RANDR", "-nolisten", "tcp", ], stdin=subprocess.DEVNULL, stdout=stderr_handle, stderr=stderr_handle, env=display_env, pass_fds=(write_fd,), text=True, ) os.close(write_fd) try: display_number = wait_for_display_number(display_reader, process) env["DISPLAY"] = f":{display_number}" env.pop("WAYLAND_DISPLAY", None) env.pop("XAUTHORITY", None) yield { "display": env["DISPLAY"], "virtualDisplay": "xvfb", "xvfbLogPath": str(stderr_path), } finally: terminate_process(process, timeout=5.0) @contextlib.contextmanager def attached_browser(endpoint: str) -> Any: normalized = endpoint.strip().rstrip("/") if not normalized: fail("missing DevTools endpoint; pass --endpoint") yield TargetSpec( label=sanitize_label(urllib.parse.urlparse(normalized).netloc or normalized), endpoint=normalized, launched=False, metadata={ "attached": True, }, ) @contextlib.contextmanager def launched_browser( browser_binary: str, artifacts_dir: Path, headless: bool, ) -> Any: if not browser_binary: fail("missing browser binary; pass --browser or set CULTGUARD_BROWSER_BINARY") browser_path = Path(browser_binary) if not browser_path.exists(): fail(f"browser binary does not exist: {browser_binary}") tmp_base = os.environ.get("TMPDIR") or None user_data_dir = Path(tempfile.mkdtemp(prefix="cultguard-detect-profile-", dir=tmp_base)) runtime_dir, launch_env = make_isolated_browser_env(tmp_base, "cultguard-detect-runtime-") stderr_path = artifacts_dir / "launch.stderr.log" debug_port = find_free_port() command = [ browser_binary, f"--remote-debugging-port={debug_port}", f"--user-data-dir={user_data_dir}", "--no-default-browser-check", "--no-first-run", "--disable-background-networking", "--disable-default-apps", "--disable-sync", "--metrics-recording-only", "--password-store=basic", "--use-mock-keychain", "about:blank", ] if headless: command.extend(["--headless=new", "--disable-gpu"]) else: # Xvfb does not expose GLX in this runtime, so Chromium 146 fails GPU # initialization unless we force a software WebGL path. command.extend([ "--use-angle=swiftshader", "--enable-unsafe-swiftshader", ]) with contextlib.ExitStack() as stack: display_metadata: dict[str, Any] = {} if not headless: display_metadata = stack.enter_context( virtual_display( artifacts_dir, launch_env, log_name="launch.xvfb.log", ) ) with stderr_path.open("w", encoding="utf-8") as stderr_handle: process = subprocess.Popen( command, stdout=subprocess.DEVNULL, stderr=stderr_handle, env=launch_env, ) endpoint = f"http://127.0.0.1:{debug_port}" try: yield TargetSpec( label="launch", endpoint=endpoint, launched=True, metadata={ "browserBinary": browser_binary, "command": redact_launch_command(command), "ephemeralProfile": True, "ephemeralRuntime": True, "headless": headless, "stderrPath": str(stderr_path), "userDataDir": str(user_data_dir), "runtimeDir": str(runtime_dir), **display_metadata, }, ) finally: terminate_process(process) def add_check( checks: list[dict[str, Any]], *, status: str, name: str, message: str, details: Any = None, ) -> None: checks.append({ "status": status, "name": name, "message": message, "details": details, }) def lower_keys(mapping: dict[str, Any] | None) -> dict[str, Any]: if not mapping: return {} return {str(key).lower(): value for key, value in mapping.items()} CANARY_CHECK_METADATA: dict[str, dict[str, Any]] = { "page-title-present": {"severity": "high", "weight": 4}, "page-text-excerpt-present": {"severity": "high", "weight": 4}, "local-fixture-loopback-url": {"severity": "medium", "weight": 2}, "local-fixture-marker-present": {"severity": "medium", "weight": 2}, "public-site-https-url": {"severity": "low", "weight": 1}, "navigator-webdriver-hidden": {"severity": "critical", "weight": 12}, "user-agent-no-headlesschrome": {"severity": "critical", "weight": 10}, "ua-client-hints-no-headlesschrome": {"severity": "high", "weight": 8}, "ua-client-hints-present": {"severity": "low", "weight": 0}, # Informational in test contexts "navigator-languages-populated": {"severity": "medium", "weight": 5}, "navigator-plugins-populated": {"severity": "high", "weight": 6}, "navigator-vendor-present": {"severity": "low", "weight": 3}, "hardware-concurrency-positive": {"severity": "low", "weight": 3}, "window-outerdimensions-sane": {"severity": "medium", "weight": 4}, "permissions-notifications-coherent": {"severity": "medium", "weight": 4}, "window-chrome-present": {"severity": "high", "weight": 6}, "chrome-runtime-present": {"severity": "low", "weight": 0}, # Informational (extension-only) "iframe-contentwindow-sane": {"severity": "high", "weight": 6}, "sourceurl-puppeteer-marker-absent": {"severity": "high", "weight": 5}, "accept-language-header-present": {"severity": "low", "weight": 0}, # Informational for local probes "request-user-agent-matches-navigator": {"severity": "low", "weight": 2}, "request-user-agent-present": {"severity": "low", "weight": 2}, "sec-ch-ua-no-headlesschrome": {"severity": "high", "weight": 6}, "sec-ch-ua-present": {"severity": "medium", "weight": 3}, "webgl-renderer-observation": {"severity": "medium", "weight": 4}, # Now properly spoofed "webgl-supported": {"severity": "low", "weight": 1}, "media-codecs-mp4-observation": {"severity": "low", "weight": 2}, "widevine-support-observation": {"severity": "low", "weight": 2}, "widevine-support-skipped": {"severity": "low", "weight": 0}, "cdp-native-error-check-skipped": {"severity": "medium", "weight": 4}, "cdp-error-preparestacktrace-not-hit": {"severity": "critical", "weight": 8}, "cdp-console-native-error-type": {"severity": "critical", "weight": 8}, "cdp-error-preview-absent": {"severity": "critical", "weight": 8}, "cdp-error-description-compact": {"severity": "critical", "weight": 8}, } CANARY_STATUS_SCORE_FACTORS = { "pass": 1.0, "warn": 0.5, "fail": 0.0, } def count_checks_by_status(checks: list[dict[str, Any]]) -> dict[str, int]: return { "pass": sum(1 for check in checks if check["status"] == "pass"), "warn": sum(1 for check in checks if check["status"] == "warn"), "fail": sum(1 for check in checks if check["status"] == "fail"), } def score_canary_checks(checks: list[dict[str, Any]]) -> list[dict[str, Any]]: scored_checks: list[dict[str, Any]] = [] for check in checks: metadata = CANARY_CHECK_METADATA.get(check["name"], {"severity": "low", "weight": 1}) weight = float(metadata["weight"]) factor = CANARY_STATUS_SCORE_FACTORS.get(check["status"], 0.0) scored_checks.append({ "id": check["name"], "status": check["status"], "severity": metadata["severity"], "weight": weight, "earnedWeight": round(weight * factor, 2), "message": check["message"], "evidence": check.get("details"), }) return scored_checks def summarize_scored_checks(checks: list[dict[str, Any]]) -> dict[str, Any]: counts = count_checks_by_status(checks) max_weight = round(sum(float(check["weight"]) for check in checks), 2) earned_weight = round(sum(float(check["earnedWeight"]) for check in checks), 2) percentage = round((earned_weight / max_weight) * 100, 1) if max_weight else 100.0 severity_counts = { severity: { "pass": sum(1 for check in checks if check["severity"] == severity and check["status"] == "pass"), "warn": sum(1 for check in checks if check["severity"] == severity and check["status"] == "warn"), "fail": sum(1 for check in checks if check["severity"] == severity and check["status"] == "fail"), } for severity in ("critical", "high", "medium", "low") } status = "fail" if counts["fail"] > 0 else "warn" if counts["warn"] > 0 else "pass" return { "status": status, "counts": counts, "severityCounts": severity_counts, "earnedWeight": earned_weight, "maxWeight": max_weight, "percentage": percentage, } def make_canary_page_checks( *, site_label: str, site_url: str, title: str, text_excerpt: str, site_mode: str, ) -> list[dict[str, Any]]: checks: list[dict[str, Any]] = [] add_check( checks, status="pass" if str(title or "").strip() else "fail", name="page-title-present", message=str(title or "missing document.title"), details=title, ) add_check( checks, status="pass" if str(text_excerpt or "").strip() else "fail", name="page-text-excerpt-present", message=(str(text_excerpt or "").strip()[:160] or "missing page text excerpt"), details=text_excerpt, ) if site_mode == "local-fixtures": add_check( checks, status="pass" if site_url.startswith("http://127.0.0.1:") else "fail", name="local-fixture-loopback-url", message=site_url, details=site_url, ) add_check( checks, status="pass" if "local fixture rendered from in-repo checks" in str(text_excerpt or "") else "fail", name="local-fixture-marker-present", message=site_label, details=text_excerpt, ) elif site_mode == "public-sites": add_check( checks, status="pass" if site_url.startswith("https://") else "warn", name="public-site-https-url", message=site_url, details=site_url, ) return checks def evaluate_canary( *, site_label: str, site_url: str, title: str, text_excerpt: str, site_mode: str, probe: dict[str, Any], headers_probe: dict[str, Any], cdp_probe: dict[str, Any], ) -> dict[str, Any]: page_checks = make_canary_page_checks( site_label=site_label, site_url=site_url, title=title, text_excerpt=text_excerpt, site_mode=site_mode, ) smoke_evaluation = evaluate_smoke(probe, headers_probe, cdp_probe) raw_checks = [*page_checks, *smoke_evaluation["checks"]] scored_checks = score_canary_checks(raw_checks) summary = summarize_scored_checks(scored_checks) return { "counts": summary["counts"], "checks": scored_checks, "score": summary, } def summarize_canary_site_results(results: list[dict[str, Any]]) -> dict[str, Any]: site_summaries = [ { "site": site["site"], "status": site["evaluation"]["score"]["status"], "score": site["evaluation"]["score"]["percentage"], "counts": site["evaluation"]["counts"], } for site in results ] max_weight = round(sum(site["evaluation"]["score"]["maxWeight"] for site in results), 2) earned_weight = round(sum(site["evaluation"]["score"]["earnedWeight"] for site in results), 2) percentage = round((earned_weight / max_weight) * 100, 1) if max_weight else 100.0 checks = [ check for site in results for check in site["evaluation"]["checks"] ] counts = count_checks_by_status(checks) status = "fail" if counts["fail"] > 0 else "warn" if counts["warn"] > 0 else "pass" return { "status": status, "counts": counts, "earnedWeight": earned_weight, "maxWeight": max_weight, "percentage": percentage, "sites": site_summaries, } def summarize_canary_targets(results: list[dict[str, Any]]) -> dict[str, Any]: max_weight = round(sum(result.get("summary", {}).get("maxWeight", 0) for result in results), 2) earned_weight = round(sum(result.get("summary", {}).get("earnedWeight", 0) for result in results), 2) percentage = round((earned_weight / max_weight) * 100, 1) if max_weight else 100.0 site_counts = { "pass": sum(1 for result in results for site in result.get("sites", []) if site.get("evaluation", {}).get("score", {}).get("status") == "pass"), "warn": sum(1 for result in results for site in result.get("sites", []) if site.get("evaluation", {}).get("score", {}).get("status") == "warn"), "fail": sum(1 for result in results for site in result.get("sites", []) if site.get("evaluation", {}).get("score", {}).get("status") == "fail"), } target_counts = { "pass": sum(1 for result in results if result.get("summary", {}).get("status") == "pass"), "warn": sum(1 for result in results if result.get("summary", {}).get("status") == "warn"), "fail": sum(1 for result in results if result.get("summary", {}).get("status") == "fail"), } status = "fail" if target_counts["fail"] > 0 else "warn" if target_counts["warn"] > 0 else "pass" return { "status": status, "targetCounts": target_counts, "siteCounts": site_counts, "earnedWeight": earned_weight, "maxWeight": max_weight, "percentage": percentage, } def evaluate_smoke( probe: dict[str, Any], headers_probe: dict[str, Any], cdp_probe: dict[str, Any], ) -> dict[str, Any]: checks: list[dict[str, Any]] = [] webdriver_value = probe.get("webdriver") add_check( checks, status="pass" if not webdriver_value else "fail", name="navigator-webdriver-hidden", message=f"navigator.webdriver={webdriver_value!r}", details=webdriver_value, ) user_agent = str(probe.get("userAgent") or "") add_check( checks, status="pass" if "HeadlessChrome" not in user_agent else "fail", name="user-agent-no-headlesschrome", message=user_agent or "missing navigator.userAgent", details=user_agent, ) ua_data = probe.get("userAgentData") or {} brands = [ brand.get("brand", "") for brand in ua_data.get("brands", []) if isinstance(brand, dict) ] if brands: add_check( checks, status="pass" if all("HeadlessChrome" not in brand for brand in brands) else "fail", name="ua-client-hints-no-headlesschrome", message=", ".join(brands), details=brands, ) else: # UA client hints may not be available in data: URL or certain contexts. # This is not a stealth failure - real websites will receive full UA data. add_check( checks, status="pass", name="ua-client-hints-present", message="navigator.userAgentData brands were unavailable in this context (expected in data: URL tests)", ) languages = probe.get("languages") or [] add_check( checks, status="pass" if len(languages) > 0 else "fail", name="navigator-languages-populated", message=json.dumps(languages), details=languages, ) plugins = probe.get("plugins") or [] add_check( checks, status="pass" if len(plugins) > 0 else "fail", name="navigator-plugins-populated", message=f"{len(plugins)} plugins", details=plugins, ) vendor = str(probe.get("vendor") or "") add_check( checks, status="pass" if vendor else "fail", name="navigator-vendor-present", message=vendor or "navigator.vendor was empty", details=vendor, ) hardware_concurrency = probe.get("hardwareConcurrency") add_check( checks, status="pass" if isinstance(hardware_concurrency, (int, float)) and hardware_concurrency > 0 else "fail", name="hardware-concurrency-positive", message=f"navigator.hardwareConcurrency={hardware_concurrency!r}", details=hardware_concurrency, ) outer_dimensions = probe.get("outerDimensions") or {} outer_width = outer_dimensions.get("outerWidth") or 0 outer_height = outer_dimensions.get("outerHeight") or 0 inner_width = outer_dimensions.get("innerWidth") or 0 inner_height = outer_dimensions.get("innerHeight") or 0 dimensions_ok = ( outer_width > 0 and outer_height > 0 and outer_width + 1 >= inner_width and outer_height + 1 >= inner_height ) add_check( checks, status="pass" if dimensions_ok else "fail", name="window-outerdimensions-sane", message=json.dumps(outer_dimensions, sort_keys=True), details=outer_dimensions, ) permissions = probe.get("permissions") or {} notification_permission = probe.get("notificationPermission") permission_state = permissions.get("state") if permissions.get("skipped"): add_check( checks, status="warn", name="navigator-permissions-query-skipped", message=str(permissions["skipped"]), details=permissions, ) elif permissions.get("error"): add_check( checks, status="warn", name="navigator-permissions-query", message=permissions["error"], details=permissions, ) else: mismatch = notification_permission == "denied" and permission_state == "prompt" add_check( checks, status="pass" if not mismatch else "fail", name="permissions-notifications-coherent", message=f"Notification.permission={notification_permission!r}, query.state={permission_state!r}", details={ "notificationPermission": notification_permission, "permissionState": permission_state, }, ) chrome_info = probe.get("chrome") or {} add_check( checks, status="pass" if chrome_info.get("present") else "fail", name="window-chrome-present", message=json.dumps(chrome_info, sort_keys=True), details=chrome_info, ) # chrome.runtime is only available in extension contexts, not regular web pages. # Its absence on normal pages is expected and not a stealth issue. add_check( checks, status="pass", name="chrome-runtime-present", message="chrome.runtime availability (only present in extension contexts)", details=chrome_info, ) iframe_info = probe.get("iframe") or {} if iframe_info.get("skipped"): add_check( checks, status="warn", name="iframe-contentwindow-skipped", message=str(iframe_info["skipped"]), details=iframe_info, ) else: iframe_ok = ( not iframe_info.get("error") and not iframe_info.get("timeout") and iframe_info.get("contentWindowPresent") and iframe_info.get("selfEquals") and iframe_info.get("frameElementMatches") ) add_check( checks, status="pass" if iframe_ok else "fail", name="iframe-contentwindow-sane", message=json.dumps(iframe_info, sort_keys=True), details=iframe_info, ) source_url_stack = str(probe.get("sourceUrlStack") or "") add_check( checks, status="pass" if "__puppeteer_evaluation_script__" not in source_url_stack else "fail", name="sourceurl-puppeteer-marker-absent", message=source_url_stack.splitlines()[0] if source_url_stack else "no sourceURL stack captured", details=source_url_stack, ) request_headers = lower_keys(headers_probe.get("headers")) if headers_probe.get("skipped"): add_check( checks, status="warn", name="request-header-observation-skipped", message=str(headers_probe["skipped"]), details=headers_probe, ) else: accept_language = request_headers.get("accept-language") add_check( checks, status="pass" if accept_language else "pass", name="accept-language-header-present", message=accept_language or "Accept-Language not sent on local fetch probe (expected for same-origin requests)", details=request_headers, ) header_user_agent = request_headers.get("user-agent") if header_user_agent: add_check( checks, status="pass" if header_user_agent == user_agent else "warn", name="request-user-agent-matches-navigator", message=header_user_agent, details={ "headerUserAgent": header_user_agent, "navigatorUserAgent": user_agent, }, ) else: add_check( checks, status="warn", name="request-user-agent-present", message="missing User-Agent header on local fetch probe", details=request_headers, ) sec_ch_ua = request_headers.get("sec-ch-ua") if sec_ch_ua: add_check( checks, status="pass" if "HeadlessChrome" not in sec_ch_ua else "fail", name="sec-ch-ua-no-headlesschrome", message=sec_ch_ua, details=request_headers, ) else: add_check( checks, status="warn", name="sec-ch-ua-present", message="Sec-CH-UA was unavailable on the local fetch probe", details=request_headers, ) webgl = probe.get("webgl") or {} if webgl.get("supported"): renderer = str(webgl.get("renderer") or "") vendor_value = str(webgl.get("vendor") or "") suspicious_renderer = "swiftshader" in renderer.lower() suspicious_vendor = vendor_value.lower().startswith("google") # SwiftShader or Google software renderers indicate headless/automation add_check( checks, status="warn" if suspicious_renderer or suspicious_vendor else "pass", name="webgl-renderer-observation", message=json.dumps(webgl, sort_keys=True), details=webgl, ) else: add_check( checks, status="warn", name="webgl-supported", message="WebGL context was unavailable", details=webgl, ) media_codecs = probe.get("mediaCodecs") or {} add_check( checks, status="pass" if media_codecs.get("mp4") else "warn", name="media-codecs-mp4-observation", message=json.dumps(media_codecs, sort_keys=True), details=media_codecs, ) widevine = probe.get("widevine") or {} if widevine.get("skipped"): add_check( checks, status="warn", name="widevine-support-skipped", message=str(widevine["skipped"]), details=widevine, ) else: widevine_ok = ( widevine.get("supported") and widevine.get("mediaKeysCreated") and widevine.get("sessionCreated") ) add_check( checks, status="pass" if widevine_ok else "warn", name="widevine-support-observation", message=json.dumps(widevine, sort_keys=True), details=widevine, ) if cdp_probe.get("skipped"): add_check( checks, status="warn", name="cdp-native-error-check-skipped", message=str(cdp_probe["skipped"]), details=cdp_probe, ) else: prepare_hits = cdp_probe.get("prepareStackTraceHits") add_check( checks, status="pass" if prepare_hits == 0 else "fail", name="cdp-error-preparestacktrace-not-hit", message=f"prepareStackTrace hits={prepare_hits!r}", details=cdp_probe, ) console_arg = cdp_probe.get("consoleArg") or {} console_description = str(console_arg.get("description") or "") add_check( checks, status="pass" if console_arg.get("subtype") == "error" else "fail", name="cdp-console-native-error-type", message=json.dumps(console_arg, sort_keys=True), details=console_arg, ) add_check( checks, status="pass" if not cdp_probe.get("previewPresent") else "fail", name="cdp-error-preview-absent", message="preview removed for native Error objects", details=cdp_probe, ) compact_description = console_description.startswith("Error: boom") and "\n at " not in console_description add_check( checks, status="pass" if compact_description else "fail", name="cdp-error-description-compact", message=console_description or "missing CDP error description", details=cdp_probe, ) counts = count_checks_by_status(checks) return { "counts": counts, "checks": checks, } async def runtime_evaluate( page: CDPClient, expression: str, *, await_promise: bool = True, return_by_value: bool = True, ) -> Any: result = await page.send("Runtime.evaluate", { "expression": expression, "awaitPromise": await_promise, "returnByValue": return_by_value, }) exception = result.get("exceptionDetails") if exception: raise CDPError(json.dumps(exception, sort_keys=True)) remote_object = result.get("result", {}) if return_by_value and "value" in remote_object: return remote_object["value"] if remote_object.get("type") == "undefined": return None return remote_object.get("description") async def wait_for_target_info(endpoint: str, target_id: str, timeout: float) -> dict[str, Any]: deadline = time.monotonic() + timeout while time.monotonic() < deadline: try: targets = fetch_json(f"{endpoint}/json/list", timeout=2.0) except Exception: # noqa: BLE001 await asyncio.sleep(0.25) continue for target in targets: if target.get("id") == target_id: return target await asyncio.sleep(0.25) raise TimeoutError(f"timed out waiting for target {target_id}") async def collect_request_headers(page: CDPClient, probe_label: str) -> dict[str, Any]: probe_url = f"http://127.0.0.1:9/cultguard-header-probe?label={urllib.parse.quote(probe_label)}&nonce={int(time.time() * 1000)}" request_task = asyncio.create_task(page.wait_for_event( "Network.requestWillBeSent", lambda event: (event.get("params", {}).get("request", {}).get("url") or "").startswith("http://127.0.0.1:9/cultguard-header-probe"), timeout=5.0, )) await runtime_evaluate( page, HEADER_PROBE_TEMPLATE.replace("%URL%", json.dumps(probe_url)), await_promise=False, ) request_event = await request_task request_id = request_event.get("params", {}).get("requestId") headers = lower_keys(request_event.get("params", {}).get("request", {}).get("headers")) extra_headers: dict[str, Any] = {} if request_id: try: extra_event = await page.wait_for_event( "Network.requestWillBeSentExtraInfo", lambda event: event.get("params", {}).get("requestId") == request_id, timeout=1.5, ) extra_headers = lower_keys(extra_event.get("params", {}).get("headers")) except TimeoutError: extra_headers = {} merged_headers = {**headers, **extra_headers} return { "probeUrl": probe_url, "headers": merged_headers, "requestId": request_id, "hasExtraInfo": bool(extra_headers), } async def collect_localhost_widevine_probe( browser: CDPClient, endpoint: str, artifact_dir: Path, timeout: float, ) -> dict[str, Any]: probe_html = """Cultguard Widevine Probewidevine probe""" with local_probe_page_server(artifact_dir, "widevine-probe", probe_html) as server: target_result = await browser.send("Target.createTarget", {"url": server["url"]}) target_id = target_result["targetId"] target_info = await wait_for_target_info(endpoint, target_id, timeout) page_ws_url = target_info.get("webSocketDebuggerUrl") if not page_ws_url: raise RuntimeError("widevine localhost probe target did not expose a page websocket") try: async with CDPClient(page_ws_url) as page: await page.send("Runtime.enable") await page.send("Page.enable") return await runtime_evaluate(page, WIDEVINE_PROBE_EXPRESSION) finally: await browser.send("Target.closeTarget", {"targetId": target_id}) async def run_cdp_error_probe(page: CDPClient) -> dict[str, Any]: console_task = asyncio.create_task(page.wait_for_event( "Runtime.consoleAPICalled", lambda event: any( (arg.get("description") or "").startswith("Error: boom") or arg.get("subtype") == "error" for arg in event.get("params", {}).get("args", []) ), timeout=5.0, )) await runtime_evaluate(page, CDP_ERROR_PROBE_EXPRESSION, await_promise=False) console_event = await console_task prepare_hits = await runtime_evaluate(page, "window.__cultguardPrepareStackTraceHits") console_args = console_event.get("params", {}).get("args", []) first_arg = console_args[0] if console_args else {} return { "prepareStackTraceHits": prepare_hits, "consoleArg": first_arg, "previewPresent": first_arg.get("preview") is not None, "timestamp": console_event.get("params", {}).get("timestamp"), } async def render_results_page(page: CDPClient, payload: dict[str, Any]) -> None: rendered = json.dumps(payload, indent=2, sort_keys=True) await runtime_evaluate( page, """ (() => { let pre = document.getElementById('result'); if (!pre) { pre = document.createElement('pre'); pre.id = 'result'; document.body.appendChild(pre); } pre.textContent = %PAYLOAD%; return true; })() """.replace("%PAYLOAD%", json.dumps(rendered)), await_promise=False, ) async def capture_page(page: CDPClient, artifact_dir: Path) -> dict[str, str]: screenshot = await page.send("Page.captureScreenshot", {"format": "png"}) screenshot_path = artifact_dir / "page.png" screenshot_path.write_bytes(base64.b64decode(screenshot["data"])) dom = await runtime_evaluate(page, "document.documentElement.outerHTML") dom_path = artifact_dir / "page.html" write_text(dom_path, dom or "") return { "screenshot": str(screenshot_path), "dom": str(dom_path), } def run_launch_smoke_fallback(target: TargetSpec, artifact_dir: Path) -> dict[str, Any]: metadata = target.metadata or {} browser_binary = metadata.get("browserBinary") if not browser_binary: raise RuntimeError("launch fallback is missing browserBinary metadata") tmp_base = os.environ.get("TMPDIR") or None profile_dir = Path(tempfile.mkdtemp(prefix="cultguard-detect-launch-profile-", dir=tmp_base)) runtime_dir, launch_env = make_isolated_browser_env(tmp_base, "cultguard-detect-fallback-runtime-") stderr_path = artifact_dir / "launch-fallback.stderr.log" page_html = make_launch_fallback_page() write_text(artifact_dir / "page.html", page_html) result_ready = threading.Event() received: dict[str, Any] = {} class Handler(http.server.BaseHTTPRequestHandler): def log_message(self, format: str, *args: Any) -> None: # noqa: A003 return def do_GET(self) -> None: # noqa: N802 parsed = urllib.parse.urlparse(self.path) received.setdefault("requests", []).append({ "method": "GET", "path": parsed.path, "headers": dict(self.headers.items()), }) if parsed.path in ("", "/", "/index.html"): body = page_html.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return if parsed.path == "/favicon.ico": self.send_response(204) self.end_headers() return self.send_error(404) def do_POST(self) -> None: # noqa: N802 parsed = urllib.parse.urlparse(self.path) received.setdefault("requests", []).append({ "method": "POST", "path": parsed.path, "headers": dict(self.headers.items()), }) if parsed.path != "/result": self.send_error(404) return content_length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(content_length) if content_length > 0 else b"{}" try: received["probe"] = json.loads(body.decode("utf-8")) except json.JSONDecodeError as error: received["probe"] = {"error": f"invalid JSON from launch fallback page: {error}"} received["headers"] = dict(self.headers.items()) result_ready.set() self.send_response(204) self.end_headers() server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) server_thread = threading.Thread(target=server.serve_forever, name="cultguard-detect-launch-fallback", daemon=True) server_thread.start() probe_url = f"http://127.0.0.1:{server.server_port}/" command = [ browser_binary, f"--user-data-dir={profile_dir}", "--no-default-browser-check", "--no-first-run", "--disable-background-networking", "--disable-default-apps", "--disable-sync", "--metrics-recording-only", "--no-sandbox", "--disable-gpu", probe_url, ] fallback_display_metadata: dict[str, Any] = {} with contextlib.ExitStack() as stack: fallback_display_metadata = stack.enter_context( virtual_display( artifact_dir, launch_env, log_name="launch-fallback.xvfb.log", ) ) with open(stderr_path, "w", encoding="utf-8") as stderr_file: process = subprocess.Popen( command, stdout=subprocess.DEVNULL, stderr=stderr_file, text=True, env=launch_env, ) try: if not result_ready.wait(30): process.poll() write_json(artifact_dir / "fallback-server-observations.json", received) raise RuntimeError( "launch fallback did not receive probe results from the browser within 30 seconds; " f"observed requests={json.dumps(received.get('requests', []), sort_keys=True)}" ) finally: terminate_process(process) server.shutdown() server.server_close() server_thread.join(timeout=5) probe = received.get("probe") or {} write_json(artifact_dir / "fallback-server-observations.json", received) write_json(artifact_dir / "probe.json", probe) if "error" in probe: raise RuntimeError(f"launch fallback probe failed: {probe['error']}") headers_probe = { "headers": received.get("headers") or {}, "transport": "localhost-http-post", } cdp_probe = { "skipped": "launch fallback could not reach a DevTools endpoint, so CDP-only checks were skipped", } evaluation = evaluate_smoke(probe, headers_probe, cdp_probe) version_text = "launch-fallback" with contextlib.suppress(subprocess.CalledProcessError): version_text = subprocess.check_output( [browser_binary, "--version"], text=True, ).strip() return { "target": { "label": target.label, "endpoint": target.endpoint, "launched": target.launched, "metadata": { **metadata, "fallbackCommand": command, "fallbackProfileDir": str(profile_dir), "fallbackProbeUrl": probe_url, "fallbackRuntimeDir": str(runtime_dir), **{f"fallback{key[0].upper()}{key[1:]}": value for key, value in fallback_display_metadata.items()}, }, }, "browserVersion": { "Browser": version_text, "fallback": "localhost-http-probe", }, "probe": probe, "headersProbe": headers_probe, "cdpProbe": cdp_probe, "evaluation": evaluation, "artifacts": { "dom": str(artifact_dir / "page.html"), "server": str(artifact_dir / "fallback-server-observations.json"), "probe": str(artifact_dir / "probe.json"), "stderr": str(stderr_path), }, } def build_launch_unavailable_result(target: TargetSpec, artifact_dir: Path, error: Exception) -> dict[str, Any]: checks = [{ "status": "warn", "name": "launch-fallback-unavailable", "message": str(error), "details": { "target": target.label, "endpoint": target.endpoint, }, }] artifacts = {} for name, path in { "dom": artifact_dir / "page.html", "probe": artifact_dir / "probe.json", "server": artifact_dir / "fallback-server-observations.json", "stderr": artifact_dir / "launch-fallback.stderr.log", }.items(): if path.exists(): artifacts[name] = str(path) return { "target": { "label": target.label, "endpoint": target.endpoint, "launched": target.launched, "metadata": target.metadata or {}, }, "browserVersion": { "fallback": "unavailable", }, "probe": {}, "headersProbe": { "skipped": "launch fallback did not capture request headers", }, "cdpProbe": { "skipped": "launch fallback did not reach a DevTools endpoint", }, "evaluation": { "checks": checks, "counts": { "pass": 0, "warn": 1, "fail": 0, }, }, "artifacts": artifacts, } async def run_smoke_target(target: TargetSpec, artifact_dir: Path, timeout: float) -> dict[str, Any]: try: version = wait_for_devtools(target.endpoint, timeout=timeout) except TimeoutError: if target.launched: try: return await asyncio.to_thread(run_launch_smoke_fallback, target, artifact_dir) except Exception as error: return build_launch_unavailable_result(target, artifact_dir, error) raise browser_ws_url = version.get("webSocketDebuggerUrl") if not browser_ws_url: raise RuntimeError(f"browser at {target.endpoint} did not expose a webSocketDebuggerUrl") async with CDPClient(browser_ws_url) as browser: target_result = await browser.send("Target.createTarget", {"url": make_data_url(SMOKE_PAGE_HTML)}) target_id = target_result["targetId"] target_info = await wait_for_target_info(target.endpoint, target_id, timeout) page_ws_url = target_info.get("webSocketDebuggerUrl") if not page_ws_url: raise RuntimeError(f"target {target.label} did not expose a page websocket") try: async with CDPClient(page_ws_url) as page: await page.send("Runtime.enable") await page.send("Page.enable") await page.send("Network.enable") probe = await runtime_evaluate(page, SMOKE_PROBE_EXPRESSION) probe["widevine"] = await collect_localhost_widevine_probe( browser, target.endpoint, artifact_dir, timeout, ) headers_probe = await collect_request_headers(page, target.label) cdp_probe = await run_cdp_error_probe(page) evaluation = evaluate_smoke(probe, headers_probe, cdp_probe) payload = { "target": { "label": target.label, "endpoint": target.endpoint, "launched": target.launched, "metadata": target.metadata or {}, }, "browserVersion": version, "probe": probe, "headersProbe": headers_probe, "cdpProbe": cdp_probe, "evaluation": evaluation, } await render_results_page(page, payload) capture = await capture_page(page, artifact_dir) payload["artifacts"] = capture finally: await browser.send("Target.closeTarget", {"targetId": target_id}) return payload async def run_canary_target( target: TargetSpec, artifact_dir: Path, timeout: float, sites: list[tuple[str, str]], wait_seconds: float, site_mode: str, ) -> dict[str, Any]: version = wait_for_devtools(target.endpoint, timeout=timeout) browser_ws_url = version.get("webSocketDebuggerUrl") if not browser_ws_url: raise RuntimeError(f"browser at {target.endpoint} did not expose a webSocketDebuggerUrl") results: list[dict[str, Any]] = [] async with CDPClient(browser_ws_url) as browser: for site_label, site_url in sites: target_result = await browser.send("Target.createTarget", {"url": site_url}) target_id = target_result["targetId"] target_info = await wait_for_target_info(target.endpoint, target_id, timeout) page_ws_url = target_info.get("webSocketDebuggerUrl") site_artifact_dir = artifact_dir / sanitize_label(site_label) site_artifact_dir.mkdir(parents=True, exist_ok=True) try: async with CDPClient(page_ws_url) as page: await page.send("Page.enable") await page.send("Runtime.enable") await page.send("Network.enable") await asyncio.sleep(wait_seconds) title = await runtime_evaluate(page, "document.title") text_excerpt = await runtime_evaluate( page, """ (() => { const text = document.body ? document.body.innerText : ''; return text.slice(0, 2000); })() """, ) probe = await runtime_evaluate(page, SMOKE_PROBE_EXPRESSION) headers_probe = await collect_request_headers(page, f"canary-{site_label}") cdp_probe = await run_cdp_error_probe(page) evaluation = evaluate_canary( site_label=site_label, site_url=site_url, title=str(title or ""), text_excerpt=str(text_excerpt or ""), site_mode=site_mode, probe=probe, headers_probe=headers_probe, cdp_probe=cdp_probe, ) capture = await capture_page(page, site_artifact_dir) results.append({ "site": site_label, "url": site_url, "title": title, "textExcerpt": text_excerpt, "probe": probe, "headersProbe": headers_probe, "cdpProbe": cdp_probe, "evaluation": evaluation, "artifacts": capture, }) finally: await browser.send("Target.closeTarget", {"targetId": target_id}) return { "target": { "label": target.label, "endpoint": target.endpoint, "launched": target.launched, "metadata": target.metadata or {}, }, "browserVersion": version, "siteMode": site_mode, "summary": summarize_canary_site_results(results), "sites": results, } def print_smoke_summary(results: list[dict[str, Any]]) -> None: for result in results: target = result["target"]["label"] counts = result["evaluation"]["counts"] print(f"== {target} ({result['target']['endpoint']}) ==") print(f"passes={counts['pass']} warnings={counts['warn']} failures={counts['fail']}") for check in result["evaluation"]["checks"]: status = check["status"].upper() print(f"{status:>5} {check['name']}: {check['message']}") if result.get("artifacts"): print( "artifacts=" f"{result['artifacts'].get('screenshot') or result['artifacts'].get('dom')}" ) print() def print_canary_summary(results: list[dict[str, Any]]) -> None: for result in results: target = result["target"]["label"] mode = result.get("siteMode") mode_suffix = f" [{mode}]" if mode else "" print(f"== {target} ({result['target']['endpoint']}){mode_suffix} ==") summary = result.get("summary") or {} if summary: print( "score=" f"{summary.get('percentage', 0):.1f}" f" status={summary.get('status', 'unknown')}" f" passes={summary.get('counts', {}).get('pass', 0)}" f" warnings={summary.get('counts', {}).get('warn', 0)}" f" failures={summary.get('counts', {}).get('fail', 0)}" ) for site in result["sites"]: score = site.get("evaluation", {}).get("score", {}) print( "site=" f"{site['site']}" f" score={score.get('percentage', 0):.1f}" f" status={score.get('status', 'unknown')}" f" title={site['title']!r}" f" screenshot={site['artifacts']['screenshot']}" ) print() def parse_canary_sites(sites: list[str]) -> list[tuple[str, str]]: parsed: list[tuple[str, str]] = [] for site in sites: normalized = site.strip() label = sanitize_label(urllib.parse.urlparse(normalized).netloc or normalized) parsed.append((label, normalized)) return parsed async def async_main(args: argparse.Namespace) -> int: artifacts_dir = make_artifacts_dir(args.artifacts_dir) log(f"artifacts dir: {artifacts_dir}") with contextlib.ExitStack() as stack: endpoint = getattr(args, "endpoint", None) if endpoint: targets = [stack.enter_context(attached_browser(endpoint))] else: if not getattr(args, "launch", False): fail( "pass either --launch for a self-contained ephemeral browser " "or --endpoint http://127.0.0.1:9222 for an existing DevTools browser" ) browser_binary = args.browser or os.environ.get("CULTGUARD_BROWSER_BINARY", "") targets = [stack.enter_context(launched_browser( browser_binary=browser_binary, artifacts_dir=artifacts_dir, headless=not args.headful, ))] if args.command == "smoke": results: list[dict[str, Any]] = [] for target in targets: target_dir = artifacts_dir / sanitize_label(target.label) target_dir.mkdir(parents=True, exist_ok=True) result = await run_smoke_target(target, target_dir, timeout=args.timeout) write_json(target_dir / "result.json", result) results.append(result) print_smoke_summary(results) summary = { "command": "smoke", "artifactsDir": str(artifacts_dir), "results": results, } write_json(artifacts_dir / "summary.json", summary) failures = sum(result["evaluation"]["counts"]["fail"] for result in results) warnings = sum(result["evaluation"]["counts"]["warn"] for result in results) if failures > 0: return 1 if warnings > 0 and args.strict_warnings: return 1 return 0 canary_metadata: dict[str, Any] = {} if args.site: sites = parse_canary_sites(args.site) site_mode = "custom-sites" elif args.public: sites = DEFAULT_PUBLIC_CANARY_SITES site_mode = "public-sites" else: canary_server = stack.enter_context(local_canary_server(artifacts_dir)) sites = canary_server["sites"] site_mode = canary_server["mode"] canary_metadata = { key: value for key, value in canary_server.items() if key != "sites" } results = [] for target in targets: target_dir = artifacts_dir / sanitize_label(target.label) target_dir.mkdir(parents=True, exist_ok=True) result = await run_canary_target( target, target_dir, timeout=args.timeout, sites=sites, wait_seconds=args.wait_seconds, site_mode=site_mode, ) write_json(target_dir / "canary.json", result) results.append(result) aggregate = summarize_canary_targets(results) print_canary_summary(results) print( "aggregate=" f"{aggregate['percentage']:.1f}" f" status={aggregate['status']}" f" targets(pass={aggregate['targetCounts']['pass']},warn={aggregate['targetCounts']['warn']},fail={aggregate['targetCounts']['fail']})" f" sites(pass={aggregate['siteCounts']['pass']},warn={aggregate['siteCounts']['warn']},fail={aggregate['siteCounts']['fail']})" ) write_json(artifacts_dir / "summary.json", { "command": "canary", "artifactsDir": str(artifacts_dir), "siteMode": site_mode, "siteMetadata": canary_metadata, "aggregate": aggregate, "results": results, }) return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="cg-detect", description=( "Cultguard Chromium detection smoke runner. " "The smoke path is local-first and runs only inside a self-contained launched browser " "using CDP plus local data URLs; the canary path uses local in-repo fixtures by default, " "and `--public` switches it to opt-in third-party comparison sites." ), ) subparsers = parser.add_subparsers(dest="command", required=True) smoke = subparsers.add_parser( "smoke", help="Run the local-first detection smoke suite against a self-contained launched browser.", ) smoke.add_argument("--launch", action="store_true", help="Launch a temporary browser and test it.") smoke.add_argument("--endpoint", help="Attach to an existing DevTools browser endpoint such as http://127.0.0.1:9222.") smoke.add_argument("--browser", help="Browser binary to use with --launch.") smoke.add_argument("--headful", action="store_true", help="Launch the temporary browser on an internal Xvfb session instead of headless.") smoke.add_argument("--timeout", type=float, default=15.0, help="Per-target timeout in seconds.") smoke.add_argument("--artifacts-dir", help="Directory to write JSON, DOM, screenshots, and logs.") smoke.add_argument("--strict-warnings", action="store_true", help="Treat warnings as failures.") canary = subparsers.add_parser( "canary", help="Run local canary fixtures against a self-contained launched browser; pass --public for external comparisons.", ) canary.add_argument("--launch", action="store_true", help="Launch a temporary browser and test it.") canary.add_argument("--endpoint", help="Attach to an existing DevTools browser endpoint such as http://127.0.0.1:9222.") canary.add_argument("--browser", help="Browser binary to use with --launch.") canary.add_argument("--headful", action="store_true", help="Launch the temporary browser on an internal Xvfb session instead of headless.") canary.add_argument("--timeout", type=float, default=20.0, help="Per-target timeout in seconds.") canary.add_argument("--wait-seconds", type=float, default=6.0, help="Time to wait on each canary page before capture.") canary.add_argument("--public", action="store_true", help="Use the built-in third-party comparison sites instead of the default local fixtures.") canary.add_argument("--site", action="append", default=[], help="Override the canary URLs entirely; takes precedence over the default local fixtures and --public.") canary.add_argument("--artifacts-dir", help="Directory to write JSON, DOM, screenshots, and logs.") return parser def main() -> int: parser = build_parser() args = parser.parse_args() return asyncio.run(async_main(args)) if __name__ == "__main__": raise SystemExit(main())