feat(detection): cap request-body inspection by Content-Length (S4)

guard_core/models.py

@@ -445,6 +445,19 @@ class SecurityConfig(BaseModel):
         le=100000,
     )
 
+    detection_max_body_inspect_bytes: int = Field(
+        default=262144,
+        description=(
+            "Maximum request body size in bytes read and inspected for penetration "
+            "detection. When the request's Content-Length exceeds this, the body is "
+            "not read or scanned and the request proceeds, bounding memory on the "
+            "detection hot path. Distinct from detection_max_content_length (the regex "
+            "scan window) and max_request_size (the 413 size gate)."
+        ),
+        ge=1024,
+        le=10485760,
+    )
+
     detection_preserve_attack_patterns: bool = Field(
         default=True,
         description="Preserve attack patterns during content truncation",

guard_core/sync/utils.py

@@ -818,6 +818,20 @@ def _build_detection_miss() -> DetectionResult:
     return DetectionResult(is_threat=False, trigger_info="")
 
 
+def _body_exceeds_inspection_cap(
+    request: SyncGuardRequest, config: "SecurityConfig | None"
+) -> bool:
+    if config is None:
+        return False
+    content_length = request.headers.get("content-length")
+    if content_length is None:
+        return False
+    try:
+        return int(content_length) > config.detection_max_body_inspect_bytes
+    except ValueError:
+        return False
+
+
 def detect_penetration_attempt(
     request: SyncGuardRequest,
     config: "SecurityConfig | None" = None,
@@ -859,6 +873,9 @@ def detect_penetration_attempt(
     if detected:
         return _build_detection_hit(trigger, threats)
 
+    if _body_exceeds_inspection_cap(request, config):
+        return _build_detection_miss()
+
     try:
         raw_body = (request.body()).decode()
     except Exception:

guard_core/utils.py

@@ -818,6 +818,20 @@ def _build_detection_miss() -> DetectionResult:
     return DetectionResult(is_threat=False, trigger_info="")
 
 
+def _body_exceeds_inspection_cap(
+    request: GuardRequest, config: "SecurityConfig | None"
+) -> bool:
+    if config is None:
+        return False
+    content_length = request.headers.get("content-length")
+    if content_length is None:
+        return False
+    try:
+        return int(content_length) > config.detection_max_body_inspect_bytes
+    except ValueError:
+        return False
+
+
 async def detect_penetration_attempt(
     request: GuardRequest,
     config: "SecurityConfig | None" = None,
@@ -859,6 +873,9 @@ async def detect_penetration_attempt(
     if detected:
         return _build_detection_hit(trigger, threats)
 
+    if _body_exceeds_inspection_cap(request, config):
+        return _build_detection_miss()
+
     try:
         raw_body = (await request.body()).decode()
     except Exception:

tests/test_sync/test_utils/test_body_inspection_cap.py

@@ -0,0 +1,66 @@
+from typing import cast
+
+from guard_core.models import SecurityConfig
+from guard_core.sync.protocols.request_protocol import SyncGuardRequest
+from guard_core.sync.utils import detect_penetration_attempt
+
+_SQLI_BODY = b'{"q": "1 OR 1=1 UNION SELECT password FROM users--"}'
+
+
+class _BodyRequest:
+    def __init__(self, body: bytes = b"", content_length: int | None = None) -> None:
+        self._body = body
+        self.query_params: dict[str, str] = {}
+        self.headers: dict[str, str] = {}
+        if content_length is not None:
+            self.headers["content-length"] = str(content_length)
+        self.url_path = "/"
+        self.method = "POST"
+        self.client_host = "127.0.0.1"
+        self.state = type("S", (), {})()
+        self.body_read = False
+
+    def body(self) -> bytes:
+        self.body_read = True
+        return self._body
+
+
+def test_over_cap_body_is_not_read_or_scanned() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=10_000_000)
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = detect_penetration_attempt(cast(SyncGuardRequest, request), config)
+
+    assert request.body_read is False
+    assert result.is_threat is False
+
+
+def test_at_cap_body_is_still_read_and_scanned() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=1024)
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = detect_penetration_attempt(cast(SyncGuardRequest, request), config)
+
+    assert request.body_read is True
+    assert result.is_threat is True
+
+
+def test_missing_content_length_still_scans() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=None)
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = detect_penetration_attempt(cast(SyncGuardRequest, request), config)
+
+    assert request.body_read is True
+    assert result.is_threat is True
+
+
+def test_malformed_content_length_falls_back_to_scanning() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=None)
+    request.headers["content-length"] = "not-a-number"
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = detect_penetration_attempt(cast(SyncGuardRequest, request), config)
+
+    assert request.body_read is True
+    assert result.is_threat is True

tests/test_utils/test_body_inspection_cap.py

@@ -0,0 +1,66 @@
+from typing import cast
+
+from guard_core.models import SecurityConfig
+from guard_core.protocols.request_protocol import GuardRequest
+from guard_core.utils import detect_penetration_attempt
+
+_SQLI_BODY = b'{"q": "1 OR 1=1 UNION SELECT password FROM users--"}'
+
+
+class _BodyRequest:
+    def __init__(self, body: bytes = b"", content_length: int | None = None) -> None:
+        self._body = body
+        self.query_params: dict[str, str] = {}
+        self.headers: dict[str, str] = {}
+        if content_length is not None:
+            self.headers["content-length"] = str(content_length)
+        self.url_path = "/"
+        self.method = "POST"
+        self.client_host = "127.0.0.1"
+        self.state = type("S", (), {})()
+        self.body_read = False
+
+    async def body(self) -> bytes:
+        self.body_read = True
+        return self._body
+
+
+async def test_over_cap_body_is_not_read_or_scanned() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=10_000_000)
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = await detect_penetration_attempt(cast(GuardRequest, request), config)
+
+    assert request.body_read is False
+    assert result.is_threat is False
+
+
+async def test_at_cap_body_is_still_read_and_scanned() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=1024)
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = await detect_penetration_attempt(cast(GuardRequest, request), config)
+
+    assert request.body_read is True
+    assert result.is_threat is True
+
+
+async def test_missing_content_length_still_scans() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=None)
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = await detect_penetration_attempt(cast(GuardRequest, request), config)
+
+    assert request.body_read is True
+    assert result.is_threat is True
+
+
+async def test_malformed_content_length_falls_back_to_scanning() -> None:
+    request = _BodyRequest(body=_SQLI_BODY, content_length=None)
+    request.headers["content-length"] = "not-a-number"
+    config = SecurityConfig(detection_max_body_inspect_bytes=1024)
+
+    result = await detect_penetration_attempt(cast(GuardRequest, request), config)
+
+    assert request.body_read is True
+    assert result.is_threat is True