Skip to content

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.

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

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.

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.

How many identities the probe needs. A value that is not an integer is a rejection; an integer outside 1–5 warns.

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.

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.

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.

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.

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 template
than 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)