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
2 changes: 1 addition & 1 deletion docs/configuration/detection-tuning.md
Original file line number Diff line number Diff line change
Expand Up @@ -271,4 +271,4 @@ Every pattern category runs only on the request contexts it is scoped to. A payl
| `xml` | `header`, `query_param`, `request_body`, `url_path` |
| `xss` | `header`, `query_param`, `request_body`, `url_path` |

Every category also runs on the `unknown` context, which is what a direct `SusPatternsManager.detect()` call without a context uses. Multipart parts, form fields and JSON values scan as `request_body`; a cookie value scans as `header`; a matrix parameter or path segment scans as `url_path`. 4.0.0 reviewed the whole matrix pair by pair with the per-context benchmark: nineteen (category, context) pairs that were never scanned gained coverage where the malicious corpus fired and the benign corpus stayed clean (three of them extend an already-disclosed false-positive class into one more context, with the corpus ids named in the benchmark), `sqli` gained `header` and `url_path` with three noise-prone rows kept narrow at pattern level, and every pair still excluded (`cmd_injection` on paths beyond the substitution patterns, `file_upload` on paths, `cms_probing`, `recon` and `sensitive_file` on headers) is one where the malicious corpus gains nothing on that context. That benchmark (`tests/test_sus_patterns/test_detection_benchmark.py`, run by the Detection Gate workflow) now measures recall and false positives per context for every category, so a change to any set shows up as a number, not a surprise.
Every category also runs on the `unknown` context, which is what a direct `SusPatternsManager.detect()` call without a context uses. Multipart parts, form fields and JSON values scan as `request_body`; the `Cookie` header scans one `name=value` pair at a time, as `header`; a matrix parameter or path segment scans as `url_path`. 4.0.0 reviewed the whole matrix pair by pair with the per-context benchmark: nineteen (category, context) pairs that were never scanned gained coverage where the malicious corpus fired and the benign corpus stayed clean (three of them extend an already-disclosed false-positive class into one more context, with the corpus ids named in the benchmark), `sqli` gained `header` and `url_path` with three noise-prone rows kept narrow at pattern level, and every pair still excluded (`cmd_injection` on paths beyond the substitution patterns, `file_upload` on paths, `cms_probing`, `recon` and `sensitive_file` on headers) is one where the malicious corpus gains nothing on that context. That benchmark (`tests/test_sus_patterns/test_detection_benchmark.py`, run by the Detection Gate workflow) now measures recall and false positives per context for every category, so a change to any set shows up as a number, not a surprise.
14 changes: 13 additions & 1 deletion guard_core/_utils/body_content_scan.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from collections.abc import Iterator, Mapping

from guard_core._utils.body_form_scan import (
_multipart_part_entries,
_multipart_text_parts,
Expand Down Expand Up @@ -291,6 +293,16 @@ async def _scan_sensitive_header(
return hit


def _header_items_to_scan(headers: Mapping[str, str]) -> Iterator[tuple[str, str]]:
for key, value in headers.items():
if key.strip().lower() != "cookie":
yield key, value
continue
for pair in value.split(";"):
if stripped := pair.strip(" "):
yield key, stripped


async def _scan_headers(
request: GuardRequest,
excluded_headers: set[str],
Expand All @@ -303,7 +315,7 @@ async def _scan_headers(
excluded_body_fields: frozenset[str] = frozenset(),
sensitive_params: frozenset[str] = frozenset(),
) -> tuple[bool, str, list[dict]]:
for key, value in request.headers.items():
for key, value in _header_items_to_scan(request.headers):
if key.strip().lower() in sensitive_headers:
hit = await _scan_sensitive_header(
key,
Expand Down
14 changes: 13 additions & 1 deletion guard_core/sync/_utils/body_content_scan.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
from collections.abc import Iterator, Mapping

from guard_core.sync._utils.body_form_scan import (
_multipart_part_entries,
_multipart_text_parts,
Expand Down Expand Up @@ -293,6 +295,16 @@ def _scan_sensitive_header(
return hit


def _header_items_to_scan(headers: Mapping[str, str]) -> Iterator[tuple[str, str]]:
for key, value in headers.items():
if key.strip().lower() != "cookie":
yield key, value
continue
for pair in value.split(";"):
if stripped := pair.strip(" "):
yield key, stripped


def _scan_headers(
request: SyncGuardRequest,
excluded_headers: set[str],
Expand All @@ -305,7 +317,7 @@ def _scan_headers(
excluded_body_fields: frozenset[str] = frozenset(),
sensitive_params: frozenset[str] = frozenset(),
) -> tuple[bool, str, list[dict]]:
for key, value in request.headers.items():
for key, value in _header_items_to_scan(request.headers):
if key.strip().lower() in sensitive_headers:
hit = _scan_sensitive_header(
key,
Expand Down
72 changes: 72 additions & 0 deletions tests/test_detection/test_cookie_header_pairs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
import base64
import json

import pytest

from guard_core._utils.body_content_scan import _header_items_to_scan
from guard_core.handlers.suspatterns_handler import sus_patterns_handler
from guard_core.models import SecurityConfig
from guard_core.utils import detect_penetration_attempt
from tests.conftest import MockGuardRequest

_CONFIG = SecurityConfig()


@pytest.fixture(autouse=True)
def _force_enhanced_detection_singleton() -> None:
sus_patterns_handler.configure(SecurityConfig())


def _b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def _session_jwt() -> str:
claims = {
"sub": "3f2c1b9e-8a47-4d2e-9c1f-5b6a7d8e9f01",
"name": "Jane Doe (Acme)",
"email": "jane.doe@example.com",
"roles": ["admin"],
"iat": 1792000000,
"exp": 1792000900,
}
return ".".join(
[
_b64url(json.dumps({"alg": "ES256", "typ": "JWT"}).encode()),
_b64url(json.dumps(claims).encode()),
_b64url(bytes(range(64))),
]
)


def test_cookie_header_is_split_into_pairs() -> None:
headers = {"Accept": "text/html", "Cookie": "a=1; b=2;;\r\nc=3"}
assert list(_header_items_to_scan(headers)) == [
("Accept", "text/html"),
("Cookie", "a=1"),
("Cookie", "b=2"),
("Cookie", "\r\nc=3"),
]


@pytest.mark.asyncio
async def test_logged_in_cookie_header_is_not_a_threat() -> None:
cookie = f"session={_session_jwt()}; theme=system; _ga=GA1.1.123456789.1792000000"
request = MockGuardRequest(headers={"Cookie": cookie})
result = await detect_penetration_attempt(request, _CONFIG)
assert result.is_threat is False


@pytest.mark.asyncio
@pytest.mark.parametrize(
"cookie",
[
pytest.param("theme=dark; q=<script>alert(1)</script>", id="xss"),
pytest.param("theme=dark; id=1' OR 1=1--", id="sqli"),
pytest.param("theme=dark;\r\nSet-Cookie: admin=1", id="crlf"),
],
)
async def test_attack_in_one_cookie_is_still_detected(cookie: str) -> None:
request = MockGuardRequest(headers={"Cookie": cookie})
result = await detect_penetration_attempt(request, _CONFIG)
assert result.is_threat is True
70 changes: 70 additions & 0 deletions tests/test_sync/test_detection/test_cookie_header_pairs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,70 @@
import base64
import json

import pytest

from guard_core.models import SecurityConfig
from guard_core.sync._utils.body_content_scan import _header_items_to_scan
from guard_core.sync.handlers.suspatterns_handler import sus_patterns_handler
from guard_core.sync.utils import detect_penetration_attempt
from tests.test_sync.conftest import SyncMockGuardRequest

_CONFIG = SecurityConfig()


@pytest.fixture(autouse=True)
def _force_enhanced_detection_singleton() -> None:
sus_patterns_handler.configure(SecurityConfig())


def _b64url(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()


def _session_jwt() -> str:
claims = {
"sub": "3f2c1b9e-8a47-4d2e-9c1f-5b6a7d8e9f01",
"name": "Jane Doe (Acme)",
"email": "jane.doe@example.com",
"roles": ["admin"],
"iat": 1792000000,
"exp": 1792000900,
}
return ".".join(
[
_b64url(json.dumps({"alg": "ES256", "typ": "JWT"}).encode()),
_b64url(json.dumps(claims).encode()),
_b64url(bytes(range(64))),
]
)


def test_cookie_header_is_split_into_pairs() -> None:
headers = {"Accept": "text/html", "Cookie": "a=1; b=2;;\r\nc=3"}
assert list(_header_items_to_scan(headers)) == [
("Accept", "text/html"),
("Cookie", "a=1"),
("Cookie", "b=2"),
("Cookie", "\r\nc=3"),
]


def test_logged_in_cookie_header_is_not_a_threat() -> None:
cookie = f"session={_session_jwt()}; theme=system; _ga=GA1.1.123456789.1792000000"
request = SyncMockGuardRequest(headers={"Cookie": cookie})
result = detect_penetration_attempt(request, _CONFIG)
assert result.is_threat is False


@pytest.mark.parametrize(
"cookie",
[
pytest.param("theme=dark; q=<script>alert(1)</script>", id="xss"),
pytest.param("theme=dark; id=1' OR 1=1--", id="sqli"),
pytest.param("theme=dark;\r\nSet-Cookie: admin=1", id="crlf"),
],
)
def test_attack_in_one_cookie_is_still_detected(cookie: str) -> None:
request = SyncMockGuardRequest(headers={"Cookie": cookie})
result = detect_penetration_attempt(request, _CONFIG)
assert result.is_threat is True
Loading