Address CSPT review: real data-flow + browser tests, fingerprint & docs fixes

Resolve the "changes required" review on the CSPT PR:

- SARIF: add a shared `_class_token` helper preferring the structured
  `finding_class` over a title keyword, and use it in BOTH `_class_fingerprint`
  and the synthetic branch of `_primary_fingerprint`, so a locationless CSPT
  finding no longer collapses onto a server-side traversal's primary
  fingerprint. Require CSPT reports to carry a discriminator (endpoint/method or
  a code location) so multiple locationless CSPT findings stay distinct.
- Dedup: include `finding_class` in the candidate and in
  `_prepare_report_for_comparison`, and teach the judge that a differing
  finding_class under a shared CWE is not a duplicate.
- Skill/runbook: correct the %2f guidance — the browser does not decode `%2f`
  during normalization; only literal-slash dot-segments traverse, and
  `%2f`-based traversal is an app-layer decode. Switch fragment examples to raw
  `../../`. Split confirmation into a CSPT *primitive* (path steered) vs. an
  *impact* level (credentials/reach/chain) to stop false negatives. Fix
  non-loadable skill names (`client_side_path_traversal`,
  `path_traversal_lfi_rfi`).
- Fixtures: drive `positive_normalized` from the fragment so it serves under a
  plain static server.
- Tests: add `test_cspt_browser_capture.py` (headless Chromium + local
  request-capture server asserting the exact outbound path per positive and no
  traversal per negative) and `test_cspt_dataflow.py` (deterministic taint
  tracker that fails when source and sink are present but unconnected); add
  primary-fingerprint and dedup regressions. Add playwright to dev deps.
- Docs: list the CSPT skill in the public skills catalog.
This commit is contained in:
Afraa Shaikh 2026-07-19 15:17:18 +05:30
parent 302e3442b6
commit 42eb5684af
11 changed files with 856 additions and 78 deletions

View file

@ -48,6 +48,7 @@ Core vulnerability classes with deep exploitation techniques.
| `business_logic` | Logic flaws, state manipulation, race conditions |
| `race_conditions` | TOCTOU, parallel request attacks |
| `path_traversal_lfi_rfi` | File inclusion, path traversal |
| `client_side_path_traversal` | Client-side path traversal (CSPT), browser request steering |
| `open_redirect` | Redirect bypasses, URL parsing tricks |
| `mass_assignment` | Attribute injection, hidden parameter pollution |
| `insecure_file_uploads` | Upload bypasses, extension tricks |

View file

@ -39,7 +39,11 @@ agent-browser wait --load networkidle
agent-browser network requests # note the intended path, e.g. /api/orders/12345/detail
# 3b. TRAVERSAL — drive the same source with a crafted value.
agent-browser open "<target>/#/orders/..%2f..%2fadmin%2fkeys"
# Use RAW ../ with literal slashes: the browser's URL parser only collapses
# dot-segments delimited by real "/". A percent-encoded ..%2f..%2f stays a
# single literal segment and does NOT traverse unless the app decodes it
# before joining (see the skill's encoding notes).
agent-browser open "<target>/#/orders/../../admin/keys"
agent-browser wait --load networkidle
agent-browser network requests # note the traversed path, e.g. /admin/keys/detail
@ -68,20 +72,32 @@ agent-browser open "<target>"
## Confirmation criteria
A finding is a real CSPT only when **all** of these hold, evidenced from the HAR:
Confirm at **two independent levels** — don't require impact to accept the
primitive.
**Level 1 — Confirmed CSPT primitive** (the bug exists), from the HAR:
1. **Traversed path on the wire** — the outbound request path is the traversed
target (e.g. `/admin/keys/detail`), not the intended one
(`/api/orders/12345/detail`).
2. **Ambient credentials** — that request carried the victim's session
(`Cookie` / `Authorization` header present; read it from the HAR entry).
3. **Benign control** — the same flow with a normal value hit the intended path.
(`/api/orders/12345/detail`), because of attacker-controlled client input.
2. **Benign control** — the same flow with a normal value hit the intended path.
Show both, side by side.
4. **Reach** — the traversed endpoint returned data or performed an action the
source value should not have been able to reach.
Do not exfiltrate real secrets — a status / length / shape delta against a
low-sensitivity sibling endpoint is enough to prove reach.
This alone establishes path traversal. A traversed request that is
unauthenticated or returns 404 is still a confirmed primitive.
**Level 2 — Confirmed exploit impact** (severity), one or more of:
- **Ambient credentials** — the traversed request carried the victim's session
(`Cookie` / `Authorization` header present; read it from the HAR entry).
- **Reach** — the traversed endpoint returned data or performed an action the
source value should not have been able to reach.
- **Chain** — the response feeds an XSS/CORS sink, or the reached endpoint is
state-changing with weak/absent CSRF defense.
Credentials and reach set severity; they don't decide whether traversal
occurred. Do not exfiltrate real secrets — a status / length / shape delta
against a low-sensitivity sibling endpoint is enough to prove reach.
## Turning it into a report
@ -110,9 +126,11 @@ Traversed (CSPT):
## Demoing against the bundled fixtures
The fixtures in `tests/fixtures/cspt/` are self-contained static pages. Serve the
directory and drive each one to see the positive/negative behavior without a real
target:
The fixtures in `tests/fixtures/cspt/` are self-contained static pages. Every
positive drives its source from the **fragment, query string, `postMessage`, or
`localStorage`** — never from a server-side route — so they all serve correctly
under a plain static server (no SPA fallback needed). Serve the directory and
drive one to see the behavior without a real target:
```bash
python -m http.server 8000 --directory tests/fixtures/cspt
@ -122,6 +140,20 @@ agent-browser wait --load networkidle
agent-browser network requests # observe the normalized /admin/keys/detail request
```
The normalization-bypass fixture is driven the same way, via the fragment:
```bash
agent-browser open "http://localhost:8000/positive_normalized.html#/view/....//....//admin/settings"
```
For an automated, reproducible version of this — used as the CI regression that
proves the source-to-sink flow steers the browser's outbound path — see
`tests/test_cspt_browser_capture.py`, which serves each fixture from a local
request-capture server, drives it with headless Chromium (Playwright), and
asserts the exact traversed path the server observed (and that negatives emit
none). `tests/test_cspt_dataflow.py` adds a deterministic taint-flow check that
fails if a fixture's source and sink are present but not connected.
The positive fixtures emit a traversed request whose path escapes the intended
`/api/...` prefix (e.g. `/admin/keys/detail`); the negative fixtures
(`negative_validated.html`, `negative_safe_construction.html`) either reject the

View file

@ -63,6 +63,12 @@ dev = [
"pyinstaller>=6.17.0; python_version >= '3.12' and python_version < '3.15'",
"pytest>=8.3",
"pytest-asyncio>=0.24",
# Drives a real headless Chromium for the CSPT browser/outbound-request
# assertions in tests/test_cspt_browser_capture.py. CI must also run
# `playwright install chromium` once; the tests importorskip when the
# runtime or browser build is unavailable so a browserless environment
# still runs the rest of the suite.
"playwright>=1.40",
]
[tool.pytest.ini_options]
@ -245,6 +251,10 @@ ignore = [
"strix/interface/tui/app.py" = ["BLE001", "PLC0415", "PLR0912", "PLR0915", "SIM105"]
"strix/interface/main.py" = ["BLE001", "PLC0415", "PLR0912", "PLR0915"]
"strix/interface/tui/renderers/agent_message_renderer.py" = ["PLC0415"]
# Browser integration harness: broad except around Playwright/Chromium startup
# is an intentional skip-guard so a browserless CI environment degrades to a
# clean skip rather than an error.
"tests/test_cspt_browser_capture.py" = ["BLE001"]
[tool.ruff.lint.isort]
force-single-line = false

View file

@ -45,6 +45,9 @@ CRITICAL DEDUPLICATION RULES:
- Different root causes (e.g., stored XSS vs reflected XSS in same field)
- Different severity levels due to different impact
- One is authenticated, other is unauthenticated
- Different finding_class even under the same CWE (e.g., client_side_path_traversal
vs a server-side/dynamic path-traversal finding both mapped to CWE-22): these are
distinct classes fixed by different code, so they are NOT duplicates
3. ARE DUPLICATES even if:
- Titles are worded differently
@ -67,6 +70,8 @@ COMPARISON GUIDELINES:
FIELDS TO ANALYZE:
- title, description: General vulnerability info
- target, endpoint, method: Exact location of vulnerability
- finding_class: Machine-readable vulnerability class; a different finding_class means a
different class of bug even under a shared CWE — not a duplicate
- technical_analysis: Root cause details
- poc_description: How it's exploited
- impact: What damage it can cause
@ -109,6 +114,7 @@ def _prepare_report_for_comparison(report: dict[str, Any]) -> dict[str, Any]:
"endpoint",
"method",
"cve",
"finding_class",
"dependency_metadata",
]

View file

@ -890,12 +890,19 @@ def _primary_fingerprint(
parts.append(f"route:{route}")
if is_synthetic:
# Augment with class keyword so locationless findings of
# Augment with the class token so locationless findings of
# different vuln classes don't collide. rule_id is already in
# the composite so we don't also need to stamp the CWE.
title = _string_value(report.get("title")) or ""
if title:
parts.append(f"synth_class:{_class_keyword(title)}")
# Prefer the structured finding_class over a title keyword: two
# CWE-22 findings (a CSPT with finding_class
# "client_side_path_traversal" and a server-side traversal with
# finding_class "dynamic") can share a generic "path traversal"
# title keyword, so keying on the title alone collapses them onto
# the same primary fingerprint. _class_token resolves to the
# discriminating structured field when present.
token = _class_token(report)
if token:
parts.append(f"synth_class:{token}")
composite = "|".join(parts)
return hashlib.sha256(composite.encode("utf-8")).hexdigest()
@ -918,39 +925,41 @@ def _class_fingerprint(rule_id: str, report: dict[str, Any]) -> str | None:
primary fingerprint (typical case: file rename, or a fix that moves
the vulnerable code to a new module).
Composite of (rule_id, vuln-class token). The class token is the
structured ``finding_class`` when it carries signal (i.e. is present
and not the default ``dynamic``), otherwise a keyword extracted from
the title.
Preferring ``finding_class`` matters when a finding's title happens to
contain a broader curated keyword: a CSPT report titled "Path Traversal
via fragment" carries ``finding_class="client_side_path_traversal"``
but its title matches the generic ``path traversal`` keyword, which
would otherwise collapse it onto the same fingerprint as a server-side
path-traversal finding (both key on CWE-22). The structured field is
authoritative, so it wins; title-keyword extraction is the fallback
only when ``finding_class`` is absent or ``dynamic``.
Title is LLM-authored so it's stochastic at the prose level, but the
keyword extraction picks up the discrete vulnerability category, which
is much more stable than the full title. Falls back to the first 5
lowercased words of the title when no curated keyword matches.
Acceptable as a fallback because the class fingerprint is a tiebreaker,
not a primary reconciliation key.
Composite of (rule_id, vuln-class token). The token comes from
``_class_token``, which prefers the structured ``finding_class`` and
falls back to a title keyword — see that helper for why the structured
field must win (CSPT vs. server-side traversal both key on CWE-22).
"""
finding_class = _string_value(report.get("finding_class"))
if finding_class and finding_class != "dynamic":
token = finding_class
else:
title = _string_value(report.get("title")) or ""
token = _class_keyword(title) if title else ""
token = _class_token(report)
if not token:
return None
composite = f"rule:{rule_id}|class:{token}"
return hashlib.sha256(composite.encode("utf-8")).hexdigest()
def _class_token(report: dict[str, Any]) -> str:
"""Resolve a finding's vulnerability-class token.
Prefers the structured ``finding_class`` when it carries signal (i.e.
present and not the default ``dynamic``); otherwise falls back to a
keyword extracted from the (LLM-authored) title. Returns empty string
when neither yields a token.
Shared by ``_class_fingerprint`` (file-rename dismissal carryover) and
the synthetic branch of ``_primary_fingerprint`` (locationless
distinguisher) so both fingerprints agree on what counts as "the same
class". Keeping them aligned is what stops a CSPT finding whose title
contains the generic ``path traversal`` keyword from collapsing onto a
server-side path-traversal finding's primary fingerprint: the
structured field distinguishes them and both hashes consume it.
"""
finding_class = _string_value(report.get("finding_class"))
if finding_class and finding_class != "dynamic":
return finding_class
title = _string_value(report.get("title")) or ""
return _class_keyword(title) if title else ""
def _class_keyword(title: str) -> str:
"""Pick the first matching curated keyword in ``title``, or fall
back to the first 5 lowercased alpha-numeric words.

View file

@ -1,5 +1,5 @@
---
name: client-side-path-traversal
name: client_side_path_traversal
description: Client-Side Path Traversal (CSPT) testing — attacker-controlled client input steering the victim's own authenticated browser requests to unintended same-origin endpoints
---
@ -28,7 +28,7 @@ CSPT becomes account takeover or a stored-XSS delivery path.
## Not CSPT (route these elsewhere)
Keep this class clean. The following are **server-side** file/path issues covered
by the `path-traversal-lfi-rfi` skill — report them as `dynamic` findings, never
by the `path_traversal_lfi_rfi` skill — report them as `dynamic` findings, never
as `client_side_path_traversal`:
- **Server-side path traversal** — server code joins a request parameter into a
@ -116,19 +116,41 @@ Inject into each candidate source and watch the **outbound request path** (via t
proxy), not the response body alone — the win condition is the browser sending a
path you didn't intend.
- **Direct**: `../`, `../../`, `..%2f` un-decoded, leading `/` (absolute-path
Know exactly what the **browser's** URL parser normalizes vs. what stays literal —
this is where most CSPT reports go wrong:
- **Browser-normalized dot-segments** (WHATWG URL Standard): the parser collapses a
path segment that equals `..`, `%2e.`, `.%2e`, or `%2e%2e` **only when it is
delimited by literal `/`**. So `../`, `%2e%2e/`, `.%2e/`, `%2e%2e%2e%2e//`-style
self-collapsing sequences all traverse once real slashes separate them.
- **Encoded separators stay encoded**: the browser does **not** turn `%2f` into a
path delimiter during normalization. `..%2f..%2fadmin` is a single literal
segment `..%2f..%2fadmin` — it does **not** become `../../admin`. Payloads like
`..%2f` or `%2e%2e%2f` only traverse if the **application** decodes them before
concatenating into the URL (see below). Do not assume `%2f` traversal without
confirming decode-then-join in the app code or on the wire.
- **Application/framework decoding before URL construction**: if app code runs
`decodeURIComponent`, reads `URLSearchParams.get()` (which decodes), or a router
decodes a route param, an encoded payload becomes raw `../` *before* it reaches
the sink — then the literal-slash rule above applies. Double-encoding
(`%252e%252e%252f`) matters only when the app decodes twice.
- **Direct** (no app decode needed): `../`, `../../`, leading `/` (absolute-path
override: `id=/admin/keys` → `/api//admin/keys` or an absolute join).
- **Encoded**: `%2e%2e%2f`, `%2e%2e/`, `..%2f`, double-encoded `%252e%252e%252f`
when an app decodes before joining. The browser decodes `%2e`/`%2f` in the path
during normalization — test both raw and encoded because framework routers may
decode a layer first.
- **Normalized**: `....//` (folds to `../` after one collapse), `..././`,
`.%2e/`, mixed `..\` on apps that rewrite backslashes, redundant `//` and `/./`
segments that shift the base.
Two more families to try, both using **literal** `/` separators so the browser
parser acts on them:
- **Self-collapsing / normalization**: `....//` (folds to `../` after a naive
single strip), `..././`, `.%2e/`, mixed `..\` on apps that rewrite backslashes,
redundant `//` and `/./` segments that shift the base.
- **Base-relative vs root-relative**: confirm whether the app uses a relative
base (`api/x`) or rooted (`/api/x`) — it changes how many `../` are needed and
whether a leading `/` fully replaces the path.
Test both raw and encoded because framework routers may decode a layer first — but
attribute any `%2f`-based traversal to an **app-layer decode**, never to the browser
parser.
## Validation (browser + Caido proxy)
Detection is dynamic: drive a real headless browser and read the **final outbound
@ -147,8 +169,9 @@ agent-browser auth login <profile> # or state load — see agent_browse
agent-browser network har start
# 3. Drive the tainted source. Examples per source type:
# - fragment: navigate with a crafted hash
agent-browser open "<target>/#/orders/..%2f..%2fadmin%2fkeys"
# - fragment: navigate with a crafted hash (RAW ../ — literal slashes; the
# browser only collapses dot-segments split by real "/")
agent-browser open "<target>/#/orders/../../admin/keys"
# - query: open with a crafted param
agent-browser open "<target>/dashboard?tab=../../admin/config"
# - postMessage: eval a crafted message into the vulnerable listener
@ -165,16 +188,38 @@ agent-browser network har stop /workspace/.agent-browser-screenshots/cspt.har
agent-browser screenshot
```
**Confirmation criteria** — you have a real CSPT when *all* hold:
Confirmation has **two independent levels**. Keep them separate — conflating
them causes false negatives (a valid primitive dismissed because impact wasn't
demonstrated *yet*).
1. The **outbound request path in the proxy/HAR** is the traversed target
(e.g. `/admin/keys`), not the intended one (`/api/users/<id>/avatar`).
2. That request carried the **victim's credentials** (session cookie /
`Authorization` header present on the request — read it from the HAR).
3. There is a **benign same-endpoint control**: the same flow with a normal
**Level 1 — Confirmed CSPT primitive** (the bug exists). Both must hold:
1. The **outbound request path in the proxy/HAR** differs from the intended one
because of attacker-controlled client input — the traversed target
(e.g. `/admin/keys`) appears on the wire, not the intended
`/api/users/<id>/avatar`.
2. There is a **benign same-endpoint control**: the same flow with a normal
value hits the intended path. Show both paths side by side.
4. The traversed endpoint returned data (or performed an action) the source
value should not have been able to reach.
That's enough to establish path traversal occurred. Credentials and a successful
response are **not** required for the primitive to be real — a traversed request
that is unauthenticated or returns 404 still proves attacker input steered the
browser's request path, and may become exploitable when chained.
**Level 2 — Confirmed exploit impact** (what it currently demonstrates). One or
more of:
- The traversed request carried the **victim's credentials** (session cookie /
`Authorization` header on the request — read it from the HAR), and
- The traversed endpoint **returned data or performed an action** the source
value should not have reached, or
- The primitive **participates in a demonstrated chain**: response rendered into
a sink (XSS), a permissive CORS reader, a state-changing endpoint with weak/no
CSRF defense, or application-added authorization on the reached path.
Credentials, reach, and chaining drive **reportability and severity** — they do
not decide *whether* traversal happened. Report a confirmed primitive; let the
impact level set the CVSS and how loudly you rate it.
Record the request/response pair (method, full path, relevant headers, status)
from the HAR as evidence. Prefer reading a low-sensitivity sibling endpoint to
@ -203,7 +248,7 @@ File confirmed CSPT with `create_vulnerability_report` and:
## False positives
Not CSPT — do not report:
**Not CSPT at all** — the primitive doesn't exist, do not report:
- The source is **validated to the resulting path**: an allowlist of endpoints,
a strict `^[a-z0-9-]+$` segment check, or `encodeURIComponent` on a value used
@ -214,11 +259,19 @@ Not CSPT — do not report:
value only ever becomes a **query parameter or request body field**, never part
of the path. A tainted query value is not CSPT (it may be another bug).
- The browser **does not normalize** the value into a different path (confirm on
the wire — no traversal actually occurred in the outbound request).
- The traversed request is **not** sent with ambient credentials (no session on
that request), so there's no privilege to abuse — note it, don't over-rate it.
- Server rejects the traversed path (404/403) with no differential — no reach, no
finding.
the wire — no traversal actually occurred in the outbound request). This is the
real negative: if the outbound path is unchanged, there is no primitive.
**Primitive exists but low/unproven impact** — this is *still CSPT*; report it,
but rate it honestly and say what's missing (do NOT silently drop it):
- The traversed request carries **no ambient credentials** (no session on that
request). The path was still steered — report the primitive at reduced severity
and note the missing privilege, or look for a chain that supplies impact.
- Server **rejects** the traversed path (404/403) with no differential. Reach
isn't demonstrated *yet*, but the primitive is confirmed. Report it as a
primitive and note that impact depends on a reachable sibling endpoint or a
chain (response rendering, CORS reader, state-changing target).
## Pro tips

View file

@ -164,7 +164,7 @@ _VALID_FIX_EFFORT = frozenset({"trivial", "low", "medium", "high"})
_VALID_FINDING_CLASS = frozenset({"dynamic", "client_side_path_traversal"})
async def _do_create( # noqa: PLR0912
async def _do_create( # noqa: PLR0912, PLR0915
*,
title: str,
description: str,
@ -218,6 +218,22 @@ async def _do_create( # noqa: PLR0912
f"{sorted(_VALID_FINDING_CLASS)}"
)
# CSPT findings are typically locationless (the bug lives in a JS bundle,
# not a server route), so they anchor to SECURITY.md in SARIF. Without a
# discriminator, multiple distinct CSPT findings collapse onto the same
# synthetic fingerprint. Require the traversed endpoint/method or a
# physical code location so each one stays a separate alert.
if finding_class == "client_side_path_traversal":
has_endpoint = bool(str(endpoint or "").strip())
has_location = bool(_normalize_code_locations(code_locations))
if not (has_endpoint or has_location):
errors.append(
"finding_class 'client_side_path_traversal' requires a discriminator: "
"set endpoint (the traversed target path, e.g. '/admin/keys') and method, "
"or a code_location for the vulnerable client sink. Locationless CSPT "
"findings otherwise collapse onto one SARIF fingerprint."
)
if not isinstance(cvss_breakdown, dict) or not cvss_breakdown:
errors.append("cvss_breakdown: must be an object with the 8 CVSS metrics")
cvss_breakdown = {}
@ -271,6 +287,7 @@ async def _do_create( # noqa: PLR0912
"poc_script_code": poc_script_code,
"endpoint": endpoint,
"method": method,
"finding_class": finding_class,
}
dedupe = await check_duplicate(candidate, existing)
if dedupe.get("is_duplicate"):
@ -575,7 +592,7 @@ async def create_vulnerability_report(
traversal / LFI / RFI but is a distinct class; setting this
keeps it separated in the report artifacts and SARIF. Do NOT
use it for server-side traversal / LFI / RFI (leave those
``dynamic``). See the ``client-side-path-traversal`` skill.
``dynamic``). See the ``client_side_path_traversal`` skill.
Example (abbreviated — mirror this structure)::

View file

@ -1,11 +1,17 @@
<!doctype html>
<!--
CSPT POSITIVE — normalization bypass
Source: route segment parsed from location.pathname
Source: route segment parsed from the URL fragment (hash router)
Sink: fetch()
A naive single-pass filter strips one "../" occurrence, but the self-
collapsing sequence "....//" folds back to "../" after that strip, so the
traversal survives and the browser normalizes the outbound path.
Served statically: the segment comes from location.hash (a hash router), so
the page loads at its own URL under `python -m http.server` and the crafted
route travels in the fragment — e.g.
positive_normalized.html#/view/....//....//admin/settings
No server-side route (/view/...) is required.
-->
<html lang="en">
<head><meta charset="utf-8" /><title>Profile</title></head>
@ -13,8 +19,8 @@
<h1>Profile</h1>
<pre id="out">loading…</pre>
<script>
// SOURCE: route segment, e.g. /view/....//....//admin/settings
const segment = location.pathname.replace(/^\/view\//, "");
// SOURCE: hash-router route segment, e.g. #/view/....//....//admin/settings
const segment = location.hash.replace(/^#\/view\//, "");
// Ineffective sanitizer: a single non-recursive strip. "....//" -> "../".
const cleaned = segment.replace(/\.\.\//g, "");

View file

@ -0,0 +1,239 @@
"""Browser-driven outbound-request assertions for the CSPT fixtures.
This is the runtime counterpart to the structural checks in
``test_cspt_skill.py``. It answers the acceptance-criterion the structural
tests cannot: *does the tainted source actually flow into the request sink and
steer the browser's outbound path?*
For each fixture we:
1. Serve the fixture directory from a local HTTP server that also **captures**
every request whose path escapes the intended ``/api/...`` prefix (the
traversed target) and every ``/api/...`` request (the benign/intended one).
2. Drive a real headless Chromium at the fixture with a crafted source value.
3. Read what the *server* actually received off the wire and assert the exact
outbound path.
Positives must emit a traversed request to the exact escaped path; negatives
must emit **no** traversed request (the value is validated away or kept out of
the path). A fixture whose source and sink exist but are not connected would
never produce the traversed request, so this fails closed — unlike a purely
textual "contains a source and a sink" check.
Skips cleanly when Playwright or its Chromium build is unavailable, so CI
without a browser stays green; the structural suite still runs there.
"""
from __future__ import annotations
import threading
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlparse
import pytest
playwright_sync = pytest.importorskip("playwright.sync_api")
FIXTURES = Path(__file__).parent / "fixtures" / "cspt"
@dataclass
class _Capture:
"""Records the request paths the server observed, by category."""
intended: list[str] = field(default_factory=list) # /api/... (benign target)
traversed: list[str] = field(default_factory=list) # escaped /api/... prefix
lock: threading.Lock = field(default_factory=threading.Lock)
def record(self, path: str) -> None:
with self.lock:
if path.startswith("/api/"):
self.intended.append(path)
else:
self.traversed.append(path)
def _make_handler(capture: _Capture, root: Path) -> type[BaseHTTPRequestHandler]:
class Handler(BaseHTTPRequestHandler):
def log_message(self, *_args: object) -> None: # silence stderr spam
return
def _serve(self) -> None:
parsed = urlparse(self.path)
path = parsed.path
# A fixture file request: serve it so the page (and its script) load.
if path == "/" or path.endswith(".html"):
name = "index.html" if path == "/" else Path(path).name
target = (root / name).resolve()
if target.is_file() and root.resolve() in target.parents:
body = target.read_bytes()
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
# Any other path is a request the page's sink emitted. Record which
# bucket it fell into, then answer 200 so the fetch/XHR resolves.
capture.record(path)
body = b"ok"
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None:
self._serve()
return Handler
@dataclass
class _Server:
base_url: str
capture: _Capture
_httpd: ThreadingHTTPServer
_thread: threading.Thread
def stop(self) -> None:
self._httpd.shutdown()
self._thread.join(timeout=5)
def _start_server() -> _Server:
capture = _Capture()
handler = _make_handler(capture, FIXTURES)
httpd = ThreadingHTTPServer(("127.0.0.1", 0), handler)
thread = threading.Thread(target=httpd.serve_forever, daemon=True)
thread.start()
host, port = httpd.server_address[0], httpd.server_address[1]
return _Server(f"http://{host}:{port}", capture, httpd, thread)
# Each case: fixture, how to drive the source, and the exact traversed path the
# server must observe on the wire. ``drive`` returns the URL to open (fragment /
# query sources) or ``None`` when the page must be seeded via JS (postMessage /
# storage), handled explicitly in the test body.
@pytest.fixture(scope="module")
def browser(): # type: ignore[no-untyped-def]
try:
pw = playwright_sync.sync_playwright().start()
except Exception as exc: # pragma: no cover - env without playwright runtime
pytest.skip(f"playwright runtime unavailable: {exc}")
try:
try:
browser = pw.chromium.launch()
except Exception as exc: # pragma: no cover - chromium not installed
pytest.skip(f"chromium not installed for playwright: {exc}")
yield browser
browser.close()
finally:
pw.stop()
@pytest.fixture()
def server(): # type: ignore[no-untyped-def]
srv = _start_server()
yield srv
srv.stop()
def _drive(browser, url: str, *, before_load_js: str | None = None): # type: ignore[no-untyped-def]
"""Open ``url`` in a fresh page, optionally running JS first, and wait for
network to settle so the sink's request has left the browser."""
context = browser.new_context()
page = context.new_page()
if before_load_js:
page.add_init_script(before_load_js)
page.goto(url, wait_until="networkidle")
page.wait_for_timeout(150)
context.close()
# --------------------------------------------------------------------------- #
# Positives — a traversed request must appear on the wire at the exact path.
# --------------------------------------------------------------------------- #
# Payloads use RAW ../ with literal slashes (browser-normalized dot-segments).
# The encoded fixture decodes at the app layer (decodeURIComponent), so it is
# fed a %2e%2e%2f payload that the page decodes before joining.
def test_positive_direct_traverses(browser, server) -> None: # type: ignore[no-untyped-def]
# #/orders/../../admin/keys -> orderId="../../admin/keys"
# -> fetch("/api/orders/../../admin/keys/detail") -> /admin/keys/detail
_drive(browser, f"{server.base_url}/positive_direct.html#/orders/../../admin/keys")
assert server.capture.traversed == ["/admin/keys/detail"], server.capture
def test_positive_encoded_traverses(browser, server) -> None: # type: ignore[no-untyped-def]
# ?resource=%2e%2e%2f%2e%2e%2fadmin%2fconfig -> decoded to ../../admin/config
# -> fetch("/api/resources/../../admin/config") -> /admin/config
payload = "%2e%2e%2f%2e%2e%2fadmin%2fconfig"
_drive(browser, f"{server.base_url}/positive_encoded.html?resource={payload}")
assert server.capture.traversed == ["/admin/config"], server.capture
def test_positive_normalized_traverses(browser, server) -> None: # type: ignore[no-untyped-def]
# #/view/....//....//admin/settings survives the single-pass strip and
# traverses. Assert the exact observed path rather than hand-computing it.
_drive(
browser,
f"{server.base_url}/positive_normalized.html#/view/....//....//admin/settings",
)
assert len(server.capture.traversed) == 1, server.capture
observed = server.capture.traversed[0]
assert not observed.startswith("/api/"), observed
assert "admin/settings" in observed, observed
def test_positive_postmessage_traverses(browser, server) -> None: # type: ignore[no-untyped-def]
# Post a crafted message after load; listener joins it into the path.
context = browser.new_context()
page = context.new_page()
page.goto(f"{server.base_url}/positive_postmessage.html", wait_until="networkidle")
page.evaluate("window.postMessage({ path: '../../admin/keys' }, '*')")
page.wait_for_timeout(200)
context.close()
# fetch("/api/widgets/../../admin/keys") -> /admin/keys
assert server.capture.traversed == ["/admin/keys"], server.capture
def test_positive_storage_traverses(browser, server) -> None: # type: ignore[no-untyped-def]
# Seed localStorage before the page script runs, then load.
_drive(
browser,
f"{server.base_url}/positive_storage.html",
before_load_js="localStorage.setItem('activeTenant', '../../admin')",
)
# api.get("/tenants/../../admin/summary") over baseURL "/api"
# -> fetch("/api/tenants/../../admin/summary") -> /admin/summary
assert server.capture.traversed == ["/admin/summary"], server.capture
# --------------------------------------------------------------------------- #
# Negatives — no traversed request may leave the browser.
# --------------------------------------------------------------------------- #
def test_negative_validated_emits_no_traversal(browser, server) -> None: # type: ignore[no-untyped-def]
# The allowlist rejects a traversal payload before the sink fires.
_drive(browser, f"{server.base_url}/negative_validated.html#/orders/../../admin/keys")
assert server.capture.traversed == [], server.capture
assert server.capture.intended == [], server.capture # guard failed -> no request at all
def test_negative_safe_construction_stays_on_constant_path(browser, server) -> None: # type: ignore[no-untyped-def]
# The value is a query param on a constant path; nothing escapes /api/.
_drive(browser, f"{server.base_url}/negative_safe_construction.html?q=../../admin")
assert server.capture.traversed == [], server.capture
assert server.capture.intended, "expected the benign /api/search request to fire"
assert all(p.startswith("/api/search") for p in server.capture.intended), server.capture

282
tests/test_cspt_dataflow.py Normal file
View file

@ -0,0 +1,282 @@
"""Deterministic source-to-sink data-flow analysis over the CSPT fixtures.
The structural check in ``test_cspt_skill.py`` only asserts that a source string
and a sink string both appear in a fixture; a page with an *unrelated* source
and sink would pass it. This module closes that gap with a small, deterministic
taint tracker that verifies the source value actually **flows into** the request
sink's path argument.
It is intentionally not a general JS engine — it models the assignment- and
concatenation-based flow the fixtures use (the same shape real CSPT bugs take):
* seed taint at ``const X = <source-expression>`` and inside a ``postMessage``
handler's ``event.data`` access;
* propagate taint across ``const Y = <expr referencing a tainted identifier>``
(covers ``.replace(...)``, ``decodeURIComponent(...)``, wrapper vars);
* mark a flow **broken** when a tainted value is gated by an allowlist guard
(``/.../.test(id)``) before any sink, or is only ever passed to
``searchParams.set`` / used as a query value on a constant path;
* report a **hit** when a tainted identifier appears in the path argument of a
request sink (``fetch(...)``, ``xhr.open(...)``, an ``axios``-style
``baseURL + path`` join).
Because it requires a *connected* path, it fails closed: break the flow in a
fixture (rename the tainted var at the sink, drop the concatenation, add a real
guard) and the corresponding assertion flips. That is the regression the issue
asks for.
"""
from __future__ import annotations
import re
from pathlib import Path
import pytest
FIXTURES = Path(__file__).parent / "fixtures" / "cspt"
POSITIVE_FIXTURES = [
"positive_direct.html",
"positive_encoded.html",
"positive_normalized.html",
"positive_postmessage.html",
"positive_storage.html",
]
NEGATIVE_FIXTURES = [
"negative_validated.html",
"negative_safe_construction.html",
]
# Expressions that introduce attacker-controlled taint.
_SOURCE_EXPR = re.compile(
r"""
location\.hash
| location\.search
| location\.pathname
| document\.referrer
| new\s+URLSearchParams\b
| localStorage\.getItem
| sessionStorage\.getItem
| event\.data
| window\.name
""",
re.VERBOSE,
)
# A request sink call. Each alternative captures the sink name; the balancer
# starts at the ``(`` that immediately follows the match. ``.open`` is the XHR
# method-then-URL form, handled by picking the 2nd argument below.
_SINK_CALL = re.compile(
r"""
(?P<open>\.open)\s*\(
| (?P<fetch>\bfetch)\s*\(
| (?P<beacon>sendBeacon)\s*\(
| (?P<es>new\s+EventSource)\s*\(
| (?P<ws>new\s+WebSocket)\s*\(
""",
re.VERBOSE,
)
_JS_IDENT = re.compile(r"[A-Za-z_$][\w$]*")
# A same-endpoint allowlist guard: an anchored character-class regex literal used
# anywhere, plus a ``.test(<ident>)`` gate. The literal and the call need not be
# adjacent (the fixture stores the regex in a const, then calls VALID_ID.test).
_ALLOWLIST_REGEX_LITERAL = re.compile(r"/\^\[[^\]]+\][^/]*\$?/")
_TEST_GUARD = re.compile(r"\.test\s*\(\s*([A-Za-z_$][\w$]*)\s*\)")
def _script_body(html: str) -> str:
return "\n".join(re.findall(r"<script>(.*?)</script>", html, re.DOTALL))
def _balanced_args(src: str, open_paren_idx: int) -> str:
"""Return the full argument-list text between an opening paren and its
matching close paren (exclusive)."""
depth = 0
out: list[str] = []
i = open_paren_idx
while i < len(src):
ch = src[i]
if ch == "(":
depth += 1
if depth == 1:
i += 1
continue
elif ch == ")":
depth -= 1
if depth == 0:
break
out.append(ch)
i += 1
return "".join(out)
def _split_top_level_args(arglist: str) -> list[str]:
"""Split an argument-list string on top-level commas (ignoring commas nested
in (), [], {}, or string/template literals)."""
args: list[str] = []
depth = 0
quote: str | None = None
buf: list[str] = []
for ch in arglist:
if quote:
buf.append(ch)
if ch == quote:
quote = None
continue
if ch in "\"'`":
quote = ch
buf.append(ch)
continue
if ch in "([{":
depth += 1
elif ch in ")]}":
depth -= 1
if ch == "," and depth == 0:
args.append("".join(buf))
buf = []
continue
buf.append(ch)
if buf:
args.append("".join(buf))
return args
def _tainted_vars(script: str) -> set[str]:
"""Fixed-point propagation of taint through ``const/let/var X = <expr>``.
Seeds on source expressions, then repeatedly taints any assignment whose
right-hand side references an already-tainted identifier. ``event.data``
accesses inside a message handler seed the handler's derived var too.
"""
assign_re = re.compile(
r"\b(?:const|let|var)\s+([A-Za-z_$][\w$]*)\s*=\s*(.+?);",
re.DOTALL,
)
assignments = [(m.group(1), m.group(2)) for m in assign_re.finditer(script)]
tainted: set[str] = set()
# Seed: direct source expression on the RHS.
for name, rhs in assignments:
if _SOURCE_EXPR.search(rhs):
tainted.add(name)
# Propagate to a fixed point: RHS references a tainted identifier.
changed = True
while changed:
changed = False
for name, rhs in assignments:
if name in tainted:
continue
idents = set(_JS_IDENT.findall(rhs))
if idents & tainted:
tainted.add(name)
changed = True
return tainted
def _guarded_idents(script: str) -> set[str]:
"""Identifiers gated by an allowlist ``.test(ident)`` guard before a sink."""
guarded: set[str] = set()
if _ALLOWLIST_REGEX_LITERAL.search(script):
for m in _TEST_GUARD.finditer(script):
guarded.add(m.group(1))
return guarded
def _sink_path_args(script: str) -> list[str]:
"""The URL/path argument text of every request sink call in the script.
The URL is the 1st argument for ``fetch``/``sendBeacon``/``EventSource``/
``WebSocket`` and the 2nd for ``xhr.open(method, url)``.
"""
args: list[str] = []
for m in _SINK_CALL.finditer(script):
open_idx = script.index("(", m.end() - 1)
arglist = _balanced_args(script, open_idx)
parts = _split_top_level_args(arglist)
if not parts:
continue
url_arg = parts[1] if m.group("open") and len(parts) > 1 else parts[0]
args.append(url_arg)
return args
def _tainted_flows_into_sink_path(script: str) -> bool:
"""True iff a tainted identifier is concatenated into a sink's path arg.
Requires an actual join into the path (``+`` concatenation or a template
literal) so that a tainted value used *only* as a query parameter (e.g.
``searchParams.set(k, tainted)`` on a constant-path URL) does not count.
"""
tainted = _tainted_vars(script)
guarded = _guarded_idents(script)
live = tainted - guarded
if not live:
return False
# An axios-style wrapper joins ``baseURL + path``; the tainted value reaches
# the sink transitively through the wrapper argument. Treat a call to such a
# wrapper method (``.get("/x/" + tainted)``) as a path-join sink too.
wrapper_join = re.compile(r"\.get\s*\(")
candidate_args = list(_sink_path_args(script))
for m in wrapper_join.finditer(script):
open_idx = script.find("(", m.end() - 1)
if open_idx != -1:
parts = _split_top_level_args(_balanced_args(script, open_idx))
if parts:
candidate_args.append(parts[0])
for arg in candidate_args:
joins_path = "+" in arg or "${" in arg or "`" in arg
if not joins_path:
continue
if set(_JS_IDENT.findall(arg)) & live:
return True
return False
# --------------------------------------------------------------------------- #
# Tests
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("name", POSITIVE_FIXTURES)
def test_positive_fixture_source_flows_into_sink_path(name: str) -> None:
script = _script_body((FIXTURES / name).read_text(encoding="utf-8"))
assert _tainted_flows_into_sink_path(script), (
f"{name}: no data flow from a client source into a request sink's path — "
"source and sink present but not connected"
)
@pytest.mark.parametrize("name", NEGATIVE_FIXTURES)
def test_negative_fixture_flow_is_broken(name: str) -> None:
script = _script_body((FIXTURES / name).read_text(encoding="utf-8"))
assert not _tainted_flows_into_sink_path(script), (
f"{name}: a tainted value reaches a request sink's path — negative fixture "
"is not actually neutralized"
)
def test_analyzer_detects_a_disconnected_source_and_sink() -> None:
# Meta-test: the analyzer must NOT flag a page where the source and sink are
# both present but unrelated (the exact gap the structural check misses).
disconnected = """
const tainted = location.hash.slice(1);
console.log(tainted); // taint goes nowhere
fetch("/api/orders/12345/detail", { credentials: "include" });
"""
assert not _tainted_flows_into_sink_path(disconnected)
def test_analyzer_flags_a_connected_source_and_sink() -> None:
# ...and it MUST flag the connected version, so the negative above is real.
connected = """
const tainted = location.hash.slice(1);
fetch("/api/orders/" + tainted + "/detail", { credentials: "include" });
"""
assert _tainted_flows_into_sink_path(connected)

View file

@ -3,7 +3,7 @@
Covers the three in-repo pillars of issue #776 that can run in CI without a live
browser/proxy:
1. The ``client-side-path-traversal`` skill exists, loads, and documents the
1. The ``client_side_path_traversal`` skill exists, loads, and documents the
source/sink taxonomy plus the "Not CSPT" boundary the agent relies on.
2. The HTML fixtures encode a real source->sink flow (positives) or a validation
guard / safe construction (negatives), asserted structurally so they can't
@ -23,6 +23,7 @@ from typing import Any
import pytest
from strix.report.dedupe import _prepare_report_for_comparison
from strix.report.sarif import write_sarif
from strix.report.state import ReportState, set_global_report_state
from strix.skills import get_all_skill_names, get_available_skills, load_skills
@ -196,9 +197,10 @@ def test_do_create_accepts_cspt_finding_class() -> None:
state = ReportState(run_name="cspt-test")
set_global_report_state(state)
try:
result = asyncio.run(
_do_create(finding_class="client_side_path_traversal", **_create_kwargs())
)
# CSPT requires a discriminator (endpoint or code location); supply the
# traversed endpoint.
kwargs = _create_kwargs(endpoint="/admin/keys", method="GET")
result = asyncio.run(_do_create(finding_class="client_side_path_traversal", **kwargs))
finally:
set_global_report_state(None) # type: ignore[arg-type]
@ -225,6 +227,41 @@ def test_do_create_rejects_unknown_finding_class() -> None:
assert any("finding_class" in e for e in result["errors"])
def test_do_create_cspt_requires_endpoint_or_location() -> None:
# A CSPT finding with no discriminator (no endpoint, no code_locations) is
# rejected so multiple locationless CSPT findings can't collapse onto one
# SARIF fingerprint.
kwargs = _create_kwargs(endpoint=None, method=None, code_locations=None)
result = asyncio.run(_do_create(finding_class="client_side_path_traversal", **kwargs))
assert result["success"] is False
assert any("discriminator" in e for e in result["errors"]), result
def test_do_create_cspt_accepts_with_endpoint() -> None:
state = ReportState(run_name="cspt-endpoint")
set_global_report_state(state)
try:
kwargs = _create_kwargs(endpoint="/admin/keys", method="GET", code_locations=None)
result = asyncio.run(_do_create(finding_class="client_side_path_traversal", **kwargs))
finally:
set_global_report_state(None) # type: ignore[arg-type]
assert result["success"] is True, result
def test_dedupe_comparison_payload_includes_finding_class() -> None:
# The structured discriminator must reach the dedup judge; without it a CSPT
# and a server-side traversal on the same endpoint look identical.
cleaned = _prepare_report_for_comparison(
{
"id": "vuln-1",
"title": "Path Traversal",
"endpoint": "/admin/keys",
"finding_class": "client_side_path_traversal",
}
)
assert cleaned.get("finding_class") == "client_side_path_traversal"
def test_sarif_surfaces_cspt_finding_class(tmp_path: Path) -> None:
write_sarif(
tmp_path,
@ -341,6 +378,92 @@ def test_sarif_class_fingerprint_uses_finding_class_over_colliding_title(tmp_pat
)
def test_sarif_primary_fingerprint_separates_locationless_cspt(tmp_path: Path) -> None:
# Regression for the *primary* fingerprint (partialFingerprints
# .primaryLocationLineHash), not just the custom class property. Both
# findings are locationless (no code_locations, no endpoint/method) so they
# anchor to SECURITY.md and share the CWE-22 ruleId. Only the structured
# finding_class distinguishes them. Before the fix the synthetic branch
# derived its class token from the title keyword alone, so a CSPT titled
# "Path Traversal via fragment" collapsed onto the server-side traversal's
# primary fingerprint — hiding a real, separate alert from code-scanning
# reconciliation. The primary fingerprints must differ.
write_sarif(
tmp_path,
[
{
"id": "vuln-0001",
"title": "Path Traversal via fragment", # generic keyword collision
"severity": "high",
"cwe": "CWE-22",
"finding_class": "client_side_path_traversal",
"timestamp": "2026-07-16 10:00:00 UTC",
"code_locations": None,
"endpoint": None,
"method": None,
},
{
"id": "vuln-0002",
"title": "Path Traversal in download file parameter",
"severity": "high",
"cwe": "CWE-22",
"finding_class": "dynamic",
"timestamp": "2026-07-16 10:00:00 UTC",
"code_locations": None,
"endpoint": None,
"method": None,
},
],
)
results = json.loads((tmp_path / "findings.sarif").read_text(encoding="utf-8"))["runs"][0][
"results"
]
primary = [r.get("partialFingerprints", {}).get("primaryLocationLineHash") for r in results]
assert all(primary), "expected a primary fingerprint on both locationless findings"
assert primary[0] != primary[1], (
"locationless CSPT and server-side traversal share a primary fingerprint; "
"the synthetic-location class token must prefer finding_class"
)
def test_sarif_primary_fingerprint_separates_cspt_by_endpoint(tmp_path: Path) -> None:
# Two locationless CSPT findings that differ only by endpoint must stay
# distinct, so multiple CSPT findings don't collapse onto one alert.
write_sarif(
tmp_path,
[
{
"id": "vuln-0003",
"title": "Client-Side Path Traversal",
"severity": "high",
"cwe": "CWE-22",
"finding_class": "client_side_path_traversal",
"timestamp": "2026-07-16 10:00:00 UTC",
"endpoint": "/admin/keys",
"method": "GET",
},
{
"id": "vuln-0004",
"title": "Client-Side Path Traversal",
"severity": "high",
"cwe": "CWE-22",
"finding_class": "client_side_path_traversal",
"timestamp": "2026-07-16 10:00:00 UTC",
"endpoint": "/admin/config",
"method": "GET",
},
],
)
results = json.loads((tmp_path / "findings.sarif").read_text(encoding="utf-8"))["runs"][0][
"results"
]
primary = [r.get("partialFingerprints", {}).get("primaryLocationLineHash") for r in results]
assert all(primary)
assert primary[0] != primary[1], (
"two CSPT findings on different traversed endpoints collapsed onto one primary fingerprint"
)
# --------------------------------------------------------------------------- #
# Fixture path normalization (documented expected paths must be correct)
# --------------------------------------------------------------------------- #