fix(job-analysis): bound pre-authentication request metadata

services/job-analysis-api/README.md

@@ -23,7 +23,17 @@ POST requires `Authorization`, `Content-Type: application/json`,
 types receive `415 unsupported_media_type` before the body is read. Optional
 `position_record_id` and `criterion_blueprint_id` may be included and are bound
 through foreign keys that fail closed when the parent is missing. GET requires
-`purpose` and returns the persisted snapshot document.
+`Authorization` and `X-Purpose-Code`, accepts no query parameters, and returns
+the persisted snapshot document.
+
+Attacker-controlled request metadata is bounded before bearer authentication.
+The transport rejects paths longer than 256 characters before splitting route
+segments or parsing UUIDs, accepts at most 64 ASGI header frames, and accepts at
+most 16 KiB of aggregate header-name and header-value bytes before lower-casing
+or dictionary allocation. Requests above a header budget fail closed as
+authentication failures, while an oversized path fails route recognition; none
+reaches the identity provider, authorization policy, request body, or persistence
+boundary.
 
 Posted evidence is bounded and unambiguous. The transport stops reading once the
 cumulative chunked body exceeds 1 MiB and rejects duplicate JSON member names at

services/job-analysis-api/src/orgmetra_job_analysis_api/http.py

@@ -35,6 +35,9 @@
 _PURPOSE_PATTERN = re.compile(r"^[a-z][a-z0-9]*(?:_[a-z0-9]+)*$")
 _MAX_UUID_INT = (1 << 128) - 1
 _MAX_REQUEST_BODY_BYTES = 1 << 20
+_MAX_REQUEST_PATH_CHARACTERS = 256
+_MAX_REQUEST_HEADERS = 64
+_MAX_REQUEST_HEADER_BYTES = 16384
 _ERROR_NEXT_ACTION = {
     "route_not_found": "Use the documented tenant-scoped job-analysis route and retry.",
     "method_not_allowed": "Use POST to persist or GET to read one job-analysis snapshot.",
@@ -291,7 +294,9 @@ async def __call__(self, scope: Mapping[str, object], receive: AsgiReceive, send
 
 
 def _looks_like_snapshot_route(path: str) -> bool:
-    """Recognize collection or item snapshot routes before parsing identifiers."""
+    """Recognize only one bounded collection or item snapshot route."""
+    if len(path) > _MAX_REQUEST_PATH_CHARACTERS:
+        return False
     parts = path.strip("/").split("/")
     if len(parts) not in {4, 5}:
         return False
@@ -314,17 +319,21 @@ def _optional_uuid(value: object) -> UUID | None:
 
 
 def _typed_headers(scope: Mapping[str, object]) -> dict[bytes, bytes]:
-    """Return lower-cased singleton headers or reject malformed header frames."""
+    """Return one bounded set of lower-cased singleton request headers."""
     raw_headers = scope.get("headers", ())
-    if not isinstance(raw_headers, Sequence):
+    if not isinstance(raw_headers, Sequence) or len(raw_headers) > _MAX_REQUEST_HEADERS:
         raise AuthenticationFailed("request headers are invalid")
     headers: dict[bytes, bytes] = {}
+    total_header_bytes = 0
     for header in raw_headers:
         if not isinstance(header, Sequence) or len(header) != 2:
             raise AuthenticationFailed("request headers are invalid")
         name, value = header
         if not isinstance(name, bytes) or not isinstance(value, bytes):
             raise AuthenticationFailed("request headers are invalid")
+        total_header_bytes += len(name) + len(value)
+        if total_header_bytes > _MAX_REQUEST_HEADER_BYTES:
+            raise AuthenticationFailed("request headers exceed the accepted size")
         key = name.lower()
         if key in headers:
             raise AuthenticationFailed("duplicate request header")

services/job-analysis-api/tests/test_http_request_budgets.py

@@ -0,0 +1,36 @@
+"""Regression contracts for bounded pre-authentication Job Analysis HTTP metadata."""
+
+from __future__ import annotations
+
+import unittest
+
+from orgmetra_job_analysis_api.auth import AuthenticationFailed
+from orgmetra_job_analysis_api.http import _looks_like_snapshot_route, _typed_headers
+
+
+class JobAnalysisHttpRequestBudgetTests(unittest.TestCase):
+    """Keep attacker-controlled request metadata bounded before authentication."""
+
+    def test_rejects_excessive_header_count_before_normalization(self) -> None:
+        """Reject a request with more than the reviewed header-frame budget."""
+        headers = [(f"x-padding-{index}".encode("ascii"), b"x") for index in range(65)]
+
+        with self.assertRaises(AuthenticationFailed):
+            _typed_headers({"headers": headers})
+
+    def test_rejects_excessive_aggregate_header_bytes_before_normalization(self) -> None:
+        """Reject one oversized header block before lower-casing or authentication."""
+        headers = [(b"x-padding", b"x" * 16384)]
+
+        with self.assertRaises(AuthenticationFailed):
+            _typed_headers({"headers": headers})
+
+    def test_rejects_excessive_route_path_before_split_or_uuid_parsing(self) -> None:
+        """Reject an oversized route-shaped path before allocating split segments."""
+        path = f"/v1/tenants/{'x' * 257}/job-analysis-snapshots"
+
+        self.assertFalse(_looks_like_snapshot_route(path))
+
+
+if __name__ == "__main__":
+    unittest.main()