Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions src/headerproof/detectors.py
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,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:
Expand Down Expand Up @@ -85,6 +88,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)
Expand Down
137 changes: 137 additions & 0 deletions tests/test_litespeed_cache.py
Original file line number Diff line number Diff line change
@@ -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
Loading