From 7bafc97350ce8320f10bcbed258c0bccba030186 Mon Sep 17 00:00:00 2001 From: Maxwell Date: Fri, 2 Oct 2026 11:45:17 +0100 Subject: [PATCH 1/2] fix: accept documented LiteSpeed cache-hit headers Record X-LiteSpeed-Cache, X-LSADC-Cache, and X-QC-Cache, and treat only an exact hit value as shared-cache hit evidence. Co-authored-by: Cursor --- src/headerproof/detectors.py | 23 +++++- tests/test_litespeed_cache.py | 137 ++++++++++++++++++++++++++++++++++ 2 files changed, 156 insertions(+), 4 deletions(-) create mode 100644 tests/test_litespeed_cache.py diff --git a/src/headerproof/detectors.py b/src/headerproof/detectors.py index 20e907c..4e3a97e 100644 --- a/src/headerproof/detectors.py +++ b/src/headerproof/detectors.py @@ -6,6 +6,7 @@ from .constants import CACHEABLE_STATUSES, LIKELY_AUTH_COOKIE, TEXTUAL_CONTENT, UNSAFE_METHODS from .evidence import make_signal from .models import HttpSnapshot +from .oob import accepted_oob_events from .templates import extract_template_evidence, template_matches @@ -39,6 +40,9 @@ def cache_indicators(snap: HttpSnapshot) -> list[str]: "surrogate-control", "akamai-cache-status", "server-timing", + "x-litespeed-cache", + "x-lsadc-cache", + "x-qc-cache", ): value = header_join(snap, name) if value: @@ -85,6 +89,11 @@ def shared_cache_hit_markers(indicators: list[str]) -> list[str]: # DYNAMIC/EXPIRED are deliberately not promoted as hit evidence. if value_l.strip() in {"hit", "stale", "updating", "revalidated"}: markers.append(indicator) + elif name_l in {"x-litespeed-cache", "x-lsadc-cache", "x-qc-cache"}: + # LiteSpeed documents "hit" as served from LSCache and "miss" as + # uncached. Only that exact hit token counts for these headers. + if value_l.strip() == "hit": + markers.append(indicator) elif name_l in {"cache-status", "akamai-cache-status", "server-timing"}: has_hit = re.search(r"\b(hit|cached|revalidated)\b", value_l) has_miss = re.search(r"\b(miss|bypass|dynamic|uncacheable|expired)\b", value_l) @@ -387,11 +396,12 @@ def analyze_oob_header_probe( probe: HttpSnapshot, save_body: bool, ) -> list[dict[str, Any]]: - protocols = sorted({str(item.get("protocol", "unknown")) for item in events}) + matched = accepted_oob_events(events, token) + protocols = sorted({item["protocol"] for item in matched}) context = { - "oob_confirmed": bool(events), + "oob_confirmed": bool(matched), "protocols": protocols, - "event_count": len(events), + "event_count": len(matched), } if not template_matches("blind_header_oob_confirmed", context): return [] @@ -400,7 +410,12 @@ def analyze_oob_header_probe( "header-injection", "blind_header_oob_confirmed", "high", "high", "Header probe produced an out-of-band callback", template_evidence( - "blind_header_oob_confirmed", context, probe_header=header_name, oob_token=token, oob_confirmed=True + "blind_header_oob_confirmed", + context, + probe_header=header_name, + oob_token=token, + oob_confirmed=True, + oob_callbacks=matched, ), probe, "Confirm the callback is attributable to the tested request and document the backend interaction.", diff --git a/tests/test_litespeed_cache.py b/tests/test_litespeed_cache.py new file mode 100644 index 0000000..4501f31 --- /dev/null +++ b/tests/test_litespeed_cache.py @@ -0,0 +1,137 @@ +from __future__ import annotations + +from headerproof.detectors import ( + analyze_header_probe, + cache_indicators, + looks_cacheable, + shared_cache_hit_markers, +) +from headerproof.models import HttpSnapshot + + +def snap( + headers: dict[str, str], + body: str = "", + request_headers: dict[str, str] | None = None, + request_url: str = "http://example.test/demo", + client_context: str = "test-client", +) -> HttpSnapshot: + return HttpSnapshot( + "GET", + request_url, + request_headers or {}, + status=200, + reason="OK", + headers={name.lower(): [value] for name, value in headers.items()}, + body_sample=body, + client_context=client_context, + ) + + +def test_litespeed_family_headers_are_recorded_and_hit_is_exact() -> None: + indicators = cache_indicators( + snap( + { + "X-LiteSpeed-Cache": "hit", + "X-LSADC-Cache": "miss", + "X-QC-Cache": "bypass", + } + ) + ) + + assert "x-litespeed-cache=hit" in indicators + assert "x-lsadc-cache=miss" in indicators + assert "x-qc-cache=bypass" in indicators + assert looks_cacheable(snap({"X-LiteSpeed-Cache": "hit"}))[0] is False + + markers = shared_cache_hit_markers( + [ + "x-litespeed-cache=hit", + "x-litespeed-cache=HIT", + "x-lsadc-cache=hit", + "x-qc-cache= hit ", + "x-litespeed-cache=miss", + "x-litespeed-cache=bypass", + "x-lsadc-cache=hitter", + "x-qc-cache=hit, miss", + "x-litespeed-cache=notahit", + "x-cache=HIT", + "cf-cache-status=MISS", + ] + ) + assert markers == [ + "x-litespeed-cache=hit", + "x-litespeed-cache=HIT", + "x-lsadc-cache=hit", + "x-qc-cache= hit ", + "x-cache=HIT", + ] + + +def test_litespeed_hit_satisfies_only_the_cache_hit_checks() -> None: + canary = "pa-scan-litespeed" + cache_url = "http://example.test/demo?pa_cb=probe" + control_url = "http://example.test/demo?pa_cb=probe-control" + clean_before = snap( + {"Cache-Control": "public, max-age=120", "X-LiteSpeed-Cache": "miss"}, + body="clean", + request_url=cache_url, + client_context="clean-before", + ) + poison = snap( + {"Cache-Control": "public, max-age=120", "X-LiteSpeed-Cache": "miss"}, + body=f"poison={canary}", + request_headers={"X-Forwarded-Host": canary}, + request_url=cache_url, + client_context="poison", + ) + fresh_control = snap( + {"Cache-Control": "public, max-age=120", "X-LiteSpeed-Cache": "miss"}, + body="clean-control", + request_url=control_url, + client_context="fresh-control", + ) + victim_hit = snap( + {"Cache-Control": "public, max-age=120", "X-LiteSpeed-Cache": "hit"}, + body=f"cached={canary}", + request_url=cache_url, + client_context="victim", + ) + victim_miss = snap( + {"Cache-Control": "public, max-age=120", "X-LiteSpeed-Cache": "miss"}, + body=f"cached={canary}", + request_url=cache_url, + client_context="victim", + ) + + confirmed = analyze_header_probe( + "X-Forwarded-Host", + canary, + "probe-litespeed-hit", + clean_before, + poison, + victim_hit, + fresh_control, + False, + ) + finding = next(item for item in confirmed if item["type"] == "cache_poisoning_shared_cache_confirmed") + checks = finding["evidence"]["state_machine_checks"] + assert checks["shared_cache_hit_marker_present"] is True + assert checks["cache_hit_progressed"] is True + assert finding["evidence"]["shared_cache_hit_markers"] == ["x-litespeed-cache=hit"] + assert finding["assessment"]["technical_gate"] == "passed" + + missed = analyze_header_probe( + "X-Forwarded-Host", + canary, + "probe-litespeed-miss", + clean_before, + poison, + victim_miss, + fresh_control, + False, + ) + assert not any(item["type"] == "cache_poisoning_shared_cache_confirmed" for item in missed) + reproduction = next(item for item in missed if item["type"] == "cache_poisoning_cross_request_reproduction") + assert reproduction["evidence"]["state_machine_checks"]["shared_cache_hit_marker_present"] is False + assert reproduction["evidence"]["state_machine_checks"]["cache_hit_progressed"] is False From 413b1d47b03d2a57b3dd5d3eda454c2acc525e00 Mon Sep 17 00:00:00 2001 From: Maxwell Date: Fri, 2 Oct 2026 11:46:34 +0100 Subject: [PATCH 2/2] fix: keep LiteSpeed cache recognition separate from OOB binding The first commit on this branch also changed out-of-band promotion. Restore that path so this change only records LiteSpeed hit headers. Co-authored-by: Cursor --- src/headerproof/detectors.py | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/src/headerproof/detectors.py b/src/headerproof/detectors.py index 4e3a97e..b22298c 100644 --- a/src/headerproof/detectors.py +++ b/src/headerproof/detectors.py @@ -6,7 +6,6 @@ from .constants import CACHEABLE_STATUSES, LIKELY_AUTH_COOKIE, TEXTUAL_CONTENT, UNSAFE_METHODS from .evidence import make_signal from .models import HttpSnapshot -from .oob import accepted_oob_events from .templates import extract_template_evidence, template_matches @@ -396,12 +395,11 @@ def analyze_oob_header_probe( probe: HttpSnapshot, save_body: bool, ) -> list[dict[str, Any]]: - matched = accepted_oob_events(events, token) - protocols = sorted({item["protocol"] for item in matched}) + protocols = sorted({str(item.get("protocol", "unknown")) for item in events}) context = { - "oob_confirmed": bool(matched), + "oob_confirmed": bool(events), "protocols": protocols, - "event_count": len(matched), + "event_count": len(events), } if not template_matches("blind_header_oob_confirmed", context): return [] @@ -410,12 +408,7 @@ def analyze_oob_header_probe( "header-injection", "blind_header_oob_confirmed", "high", "high", "Header probe produced an out-of-band callback", template_evidence( - "blind_header_oob_confirmed", - context, - probe_header=header_name, - oob_token=token, - oob_confirmed=True, - oob_callbacks=matched, + "blind_header_oob_confirmed", context, probe_header=header_name, oob_token=token, oob_confirmed=True ), probe, "Confirm the callback is attributable to the tested request and document the backend interaction.",