Pentest JS probe format
Applies to the JavaScript probes cxg pentest runs — not to .js
templates in a template directory. Those are loaded by the javascript
template engine and use the
annotation header schema.
A probe written to this schema will not load as a template, and a template
will not run as a probe.
A probe is one JavaScript file defining async function cxgProbe(cxg). The
pipeline evaluates the file inside a page that is already authenticated as
one of the identities under test, calls that function, and reads the array
it returns.
The contract, in the pipeline’s own words
Section titled “The contract, in the pipeline’s own words”From pentest/js_engine.py:
JS template engine — runs JavaScript probe templates inside authenticated browser contexts.
Template contract (copy-paste-able into DevTools console, ALSO runnable by cxg):
// @id: my-probe // @vuln_class: idor // @severity: high // @requires_auth_count: 2 // @description: <one-liner> async function cxgProbe(cxg) { // cxg.profile -> {name, label, index} // cxg.profiles -> [{name, label, index}, ...] all identities in this run // cxg.url(path) -> full target URL // cxg.fetch(path, opts) -> fetch in THIS identity (auto-CSRF on mutations) // cxg.fetchAs(idx, path, opts) -> fetch in another identity (idx into cxg.profiles) const findings = []; // ... probe logic ... return findings; // [{id, severity, confirmed, endpoint, description, evidence, payload}] }
For DevTools console fallback (no cxg bridge), the template can detect: const cxg = (typeof window.__cxg !== 'undefined') ? window.__cxg : { fetch, url:(p)=>p, profile:{name:'console',label:'console',index:0}, profiles:[], fetchAs:(i,p,o)=>fetch(p, {...o, credentials:'include'}) };
cxg's runner wires up the real `window.__cxg` and the cxgProbe is called automatically.Header annotations
Section titled “Header annotations”Read from the whole file — not a fixed window — with one regex, and only
// comments count:
re.compile(r"//\s*@(\w+):\s*(.*)")Required: @id, @vuln_class. A probe
missing one is rejected before it runs, unlike a template, where a
missing annotation only warns.
Every annotation the pipeline reads, and what it uses when one is absent:
| Annotation | Absent → |
|---|---|
@destructive_priority |
0 |
@id |
the filename without its extension |
@requires_auth_count |
1 |
@requires_capability |
unset |
@severity |
medium |
@threat_id |
unset |
@vuln_class |
unknown |
@vuln_class
Section titled “@vuln_class”Required. One of these is expected; another value warns and still runs:
idor, csrf, xss, ssrf, command_injection, session_replay, credential_stuffing, rate_limit_bypass, privilege_escalation, sensitive_data_exposure, metrics_info_disclosure, clickjacking, content_sniffing, auth_api, denial_of_service.
@requires_capability
Section titled “@requires_capability”One of http, ipc, host_fs, naming what
the probe needs of the surface it runs against. A probe demanding a
capability the surface lacks is skipped rather than run — an undefined
namespace would throw, and a throw is indistinguishable from a genuine
refutation.
Calling cxg.ipc.* without declaring @requires_capability: ipc is a
rejection, not a warning.
@requires_auth_count
Section titled “@requires_auth_count”How many identities the probe needs. A value that is not an integer is a rejection; an integer outside 1–5 warns.
How the probe is invoked
Section titled “How the probe is invoked”The file is evaluated inside this wrapper, in the page. %s is where the
probe’s own source is substituted:
async () => { try { %s if (typeof cxgProbe !== 'function') { return {__cxg_err: 'template did not define async function cxgProbe(cxg)'}; } const result = await cxgProbe(window.__cxg); return {__cxg_ok: true, findings: result || []}; } catch (e) { return {__cxg_err: String(e && e.stack || e)}; }}A file that does not define cxgProbe is rejected by the validator before
this point. A cxgProbe that throws is reported and contributes no
findings.
What cxg hands the probe
Section titled “What cxg hands the probe”The bridge, installed as a page init script so it survives navigation, from
pentest/targets/bridge.py:
window.__cxg = { profile: PROFILE, profiles: PROFILES, // Which identity is privileged, decided from operator-set --tier rather // than from position in `profiles`. null when the roster cannot be ranked // (no --tier on some identity, or a tie at that extreme) — in that case // report outcome 'unevaluated'; do NOT fall back to an index. lowestPrivilege: LOWEST_PRIVILEGE, highestPrivilege: HIGHEST_PRIVILEGE, url: (p) => p.startsWith('http') ? p : TARGET + (p.startsWith('/') ? p : '/' + p), fetch: cxgFetch, fetchAs: cxgFetchAs, // Cookie API (HttpOnly-aware — uses Playwright cookie jar, not document.cookie) getCookies: cxgGetCookies, getCookiesAs: cxgGetCookiesAs, getCookie: cxgGetCookie, setCookies: cxgSetCookies, addCookies: cxgAddCookies, clearCookies: cxgClearCookies, // Out-of-band callback API for blind-vuln confirmation (SSRF, blind SQLi, // blind XXE, blind cmd injection). host is null when --oast was not // passed; pollable is false unless cxg can read interactions back, and // only a poll() that RETURNS an interaction confirms anything. oast: cxgOast,};Every request must go through cxg.fetch or cxg.fetchAs. A raw fetch(
is a rejection: it bypasses CSRF injection, the scope check, and the audit
log.
On a desktop surface
Section titled “On a desktop surface”An Electron substrate adds an IPC namespace, from
pentest/targets/electron.py:
window.__cxg.ipc = { invoke: (channel, ...args) => window.__cxg_ipc_call(INDEX, channel, args), invokeAs: (idx, channel, ...args) => window.__cxg_ipc_call(idx, channel, args), channels: () => CHANNELS.slice(),};A raw ipcRenderer.invoke(...) is a rejection for the same reason a raw
fetch( is.
What the probe returns
Section titled “What the probe returns”cxgProbe returns an array of plain objects. Each one becomes a finding
with these fields:
| Field | Type |
|---|---|
id |
str |
vuln_class |
str |
severity |
str |
confirmed |
bool |
target |
str |
endpoint |
str |
description |
str |
evidence |
dict |
payload |
Optional[str] |
hypothesis_id |
Optional[str] |
threat_id |
Optional[str] |
Which keys are read off the object, and what each one falls back to when the
probe omits it — tpl is the probe’s own header:
def _finding_from_dict(self, f: dict, tpl: JsTemplate) -> Finding: return Finding( id=f.get("id") or f"{tpl.vuln_class}-finding", vuln_class=f.get("vuln_class") or tpl.vuln_class, severity=f.get("severity") or tpl.severity, confirmed=bool(f.get("confirmed", False)), target=self.target, endpoint=f.get("endpoint", ""), description=f.get("description", ""), evidence=f.get("evidence", {}) or {}, payload=str(f.get("payload")) if f.get("payload") is not None else None, hypothesis_id=tpl.id, threat_id=tpl.threat_id, )confirmed: true with nothing to show for it is not believed:
_has_probe_evidence in pentest/mutator.py
ignores empty values and _-prefixed bookkeeping keys, and triage downgrades
the finding.
What the validator refuses to run
Section titled “What the validator refuses to run”From pentest/validator.py:
Pre-execution validator for AI-generated JS pentest templates.
Catches templates that would: - silently misbehave (no @id, no @vuln_class, no cxgProbe function) - violate the contract (raw fetch() instead of cxg.fetch — credentials lost) - target destructive routes (regex check on string literals in the source) - exceed declared request budget
Static analysis is intentionally conservative — better to flag a benign templatethan to run a bad one. The runtime ScopeGuard provides the backstop.Rejections stop the probe from running at all; warnings do not. More than 12 fetch call sites combined with a loop warns. The checks in full, in the order they run:
Show validate()
def validate(source: str, destructive_ok: bool = False) -> ValidationResult: errors: list[str] = [] warnings: list[str] = [] meta = parse_meta(source)
# 1. Required metadata for k in REQUIRED_META: if k not in meta: errors.append(f"missing required metadata: @{k}")
if "vuln_class" in meta and meta["vuln_class"] not in ALLOWED_VULN_CLASSES: warnings.append(f"unusual vuln_class '{meta['vuln_class']}' (not in standard set)")
# @g.comment -- "Warns rather than rejects on an unrecognised @requires_capability: the capability set grows as new substrates are added, and a typo here should not hard-fail a template the way a missing @id does." if "requires_capability" in meta and meta["requires_capability"] not in ALLOWED_CAPABILITIES: warnings.append( f"unusual requires_capability '{meta['requires_capability']}' " f"(known: {', '.join(sorted(ALLOWED_CAPABILITIES))})")
# @g.comment -- "Hard-rejects raw ipcRenderer calls: they bypass cxg.ipc.invoke/invokeAs and therefore the scope budget and audit log, mirroring the raw fetch() ban below." if _RAW_IPC_PATTERN.search(source): errors.append( "raw ipcRenderer call detected. Templates must use cxg.ipc.invoke(channel, ...) " "or cxg.ipc.invokeAs(idx, channel, ...) — raw ipcRenderer bypasses the scope " "budget and the audit log.")
# @g.comment -- "Hard-rejects a template that calls cxg.ipc.* without declaring the capability, matching the raw fetch() and raw ipcRenderer bans in severity and message style." if _USES_CXG_IPC.search(source) and meta.get("requires_capability") != "ipc": errors.append( "template calls cxg.ipc.* but does not declare '@requires_capability: ipc'. " "Without the declaration the engine's capability gate cannot skip it on a " "substrate that has no IPC, and it will throw at runtime instead.")
# 2. Must define cxgProbe if not re.search(r"\basync\s+function\s+cxgProbe\s*\(", source): errors.append("missing required `async function cxgProbe(cxg)` definition")
# 3. No raw fetch() — must use cxg.fetch or cxg.fetchAs raw_fetch_matches = [] for m in _RAW_FETCH_PATTERN.finditer(source): # Check the char immediately before — if it's `.` it's cxg.fetch, allow. start = m.start() if start > 0 and source[start - 1] in ".$_": continue # If the preceding identifier is `window.__cxg_fetchAs` style, allow. # We've already excluded `.fetch(`. So a bare `fetch(` slipped through — flag it. line_start = source.rfind("\n", 0, start) + 1 line_end = source.find("\n", start) line = source[line_start:line_end if line_end >= 0 else len(source)] raw_fetch_matches.append((start, line.strip())) if raw_fetch_matches: errors.append( f"raw fetch() call detected ({len(raw_fetch_matches)} site(s)). " f"Templates must use cxg.fetch(path, opts) or cxg.fetchAs(idx, path, opts) — " f"raw fetch() bypasses CSRF injection and the audit log. " f"First offending line: {raw_fetch_matches[0][1][:120]}" )
# 4. Destructive literals if not destructive_ok: for pat in _DESTRUCTIVE_LITERAL_PATTERNS: m = re.search(pat, source, re.IGNORECASE) if m: errors.append( f"template contains destructive endpoint literal: {m.group(0)[:80]} — " f"pass --destructive-ok to allow" )
# 5. Cheap explosion heuristic: many fetch sites + a loop = potential runaway callsite_count = len(_FETCH_CALLSITES.findall(source)) has_loop = bool(_LOOP_PATTERN.search(source)) if callsite_count > _MAX_FETCH_CALLSITES and has_loop: warnings.append( f"template has {callsite_count} fetch call sites AND a loop — potential request flood. " f"ScopeGuard budget will halt it, but consider rewriting." )
# 6. requires_auth_count sanity if "requires_auth_count" in meta: try: n = int(meta["requires_auth_count"]) if n < 1 or n > 5: warnings.append(f"requires_auth_count={n} is unusual (expected 1 or 2)") except ValueError: errors.append(f"requires_auth_count must be an int, got: {meta['requires_auth_count']}")
return ValidationResult(ok=not errors, errors=errors, warnings=warnings, meta=meta)
