fix(ci): stop the nightly Atheris OOM + clear 3 reviewed Semgrep base findings

src/sdp/authz.py

@@ -113,7 +113,7 @@ def resolve_oidc_actor_context(
 
 def _load_jwks_from_url(jwks_url: str) -> dict[str, Any]:
     timeout = float(os.getenv("SDP_OIDC_JWKS_TIMEOUT_SECONDS", "2"))
-    with urlopen(jwks_url, timeout=timeout) as response:
+    with urlopen(jwks_url, timeout=timeout) as response:  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- jwks_url is operator config (SDP_OIDC_JWKS_URL from the KV/env), never attacker-controlled; OIDC discovery requires the operator to set an https endpoint
         return json.loads(response.read().decode("utf-8"))
 
 

src/sdp/graph_store.py

@@ -475,7 +475,7 @@ def _cypher(
         )
         driver_connection = conn.connection.driver_connection
         with driver_connection.cursor() as cursor:
-            cursor.execute(statement, (json.dumps(params),))
+            cursor.execute(statement, (json.dumps(params),))  # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- not raw-string SQL: statement is a psycopg.sql.SQL composition with graph_name/query as Literal()-escaped values and result_columns drawn from a closed allowlist map; the agtype params are bound via the %s driver parameter
             return cursor.fetchall()
 
     def upsert_node(

src/sdp/observability.py

@@ -136,7 +136,7 @@ def _export_to_sink(observation: dict[str, Any]) -> None:
             headers={"Content-Type": "application/json"},
             method="POST",
         )
-        with urlopen(request, timeout=timeout_ms / 1000):
+        with urlopen(request, timeout=timeout_ms / 1000):  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- reached only after the scheme is validated to http/https above (the file:// scheme is handled separately and any other scheme raises); sink_url is operator config (SDP_LOG_SINK_URL), not attacker-controlled
             return
 
     raise ValueError(f"unsupported SDP_LOG_SINK_URL scheme: {scheme}")

tests/fuzz/atheris/fuzz_draft_sql.py

@@ -26,6 +26,7 @@
 
 
 def TestOneInput(data: bytes) -> None:
+    invariants.reset_shared_state()  # isolate each iteration: keep audit/decision logs bounded
     fdp = atheris.FuzzedDataProvider(data)
     try:
         req = QueryDraftRequest(

tests/fuzz/atheris/fuzz_execute_query.py

@@ -26,6 +26,7 @@
 
 
 def TestOneInput(data: bytes) -> None:
+    invariants.reset_shared_state()  # isolate each iteration: keep audit/decision logs bounded
     fdp = atheris.FuzzedDataProvider(data)
     ds = [
         fdp.PickValueInList(_DATASETS) for _ in range(fdp.ConsumeIntInRange(1, 3))

tests/fuzz/atheris/fuzz_resolve_terms.py

@@ -17,6 +17,7 @@
 
 
 def TestOneInput(data: bytes) -> None:
+    invariants.reset_shared_state()  # isolate each iteration: keep audit/decision logs bounded
     fdp = atheris.FuzzedDataProvider(data)
     text = fdp.ConsumeUnicodeNoSurrogates(fdp.remaining_bytes())
     invariants.check_resolve_terms(text)

tests/fuzz/atheris/fuzz_search_catalog.py

@@ -19,6 +19,7 @@
 
 
 def TestOneInput(data: bytes) -> None:
+    invariants.reset_shared_state()  # isolate each iteration: keep audit/decision logs bounded
     fdp = atheris.FuzzedDataProvider(data)
     query = fdp.ConsumeUnicodeNoSurrogates(fdp.ConsumeIntInRange(0, 200))
     tags = [

tests/fuzz/invariants.py

@@ -10,12 +10,34 @@
 
 from typing import Any
 
-from sdp import catalog, ontology, orchestrator
+from sdp import catalog, evidence, ontology, orchestrator
 from sdp.domain import QueryDraftRequest, QueryExecutionRequest, QueryExecutionResponse
 
 # Anything the query drafter/executor must never let through unescaped.
 FORBIDDEN_KEYWORDS = orchestrator._FORBIDDEN_KEYWORDS
 
+
+def reset_shared_state() -> None:
+    """Clear the process-global in-memory accumulators the query paths append to.
+
+    Every ``orchestrator.execute_query`` call records an audit event
+    (``catalog._AUDIT_LOG``) and a policy decision (``evidence._POLICY_DECISION_LOG``
+    via ``policy.evaluate``); ``draft_sql`` records the policy decision. Those are
+    unbounded module-level lists. A coverage-guided Atheris run drives these
+    functions millions of times against that shared state, so without a
+    per-iteration reset the lists grow until libFuzzer kills the run on its RSS
+    limit — an out-of-memory that is a harness-isolation artifact, not a defect in
+    the code under test (it only reproduces at the long nightly budget, never at
+    the short PR budget). Resetting per iteration mirrors the
+    ``isolate_in_memory_app_state`` fixture the API tests already use, and keeps
+    the crash-oracle focused on input handling rather than accumulated memory.
+
+    ``list.clear()`` keeps each list's object identity, so any module that closed
+    over the original list still observes the reset.
+    """
+    catalog._AUDIT_LOG.clear()
+    evidence._POLICY_DECISION_LOG.clear()
+
 # The genuine injection-relevant contract: an emitted identifier must contain no
 # ASCII whitespace and no SQL metacharacter. (NB: fuzzing surfaced that
 # ``str.isalnum()`` is Unicode-aware, so non-ASCII letters/digits such as "²" or

tests/fuzz/test_fuzz_properties.py

@@ -27,6 +27,17 @@
 )
 from tests.fuzz import invariants
 
+
+@pytest.fixture(autouse=True)
+def _isolate_fuzz_shared_state():
+    """Keep the process-global audit / policy-decision logs from accumulating
+    across property examples and leaking into other test modules — the same
+    isolation the Atheris harnesses now apply per iteration."""
+    invariants.reset_shared_state()
+    yield
+    invariants.reset_shared_state()
+
+
 # A generous but bounded settings profile: enough cases to explore the surface
 # without slowing the unit suite down.
 FUZZ_SETTINGS = settings(
@@ -191,3 +202,36 @@ def test_dto_parsing_only_raises_validation_error(model, payload):
         # e.g. a payload key collides with a positional-only/duplicate arg;
         # still a controlled parse failure, not a logic crash.
         pass
+
+
+# --------------------------------------------------------------------------- #
+# Regression: the nightly Atheris `fuzz` job was OOM-killed (RSS > 2 GiB) because
+# every execute_query/draft_sql call appends to unbounded module-level logs and
+# the coverage-guided run drives them millions of times. The harnesses now reset
+# that state per iteration; assert the reset actually keeps the logs bounded so
+# a future refactor that drops it fails here (fast) instead of only in nightly CI.
+# --------------------------------------------------------------------------- #
+def test_reset_shared_state_bounds_accumulators_across_a_call_loop():
+    from sdp import catalog, evidence
+
+    req = QueryExecutionRequest(
+        language="SQL",
+        user="analyst",
+        purpose="analysis",
+        dataset_ids=["crm-customer-master"],
+        query="select 1",
+        dry_run=True,
+    )
+    for _ in range(64):
+        invariants.reset_shared_state()
+        invariants.check_execute_query(req)
+
+    # With a reset per iteration only the final iteration's records survive; a
+    # single execute_query records one audit event and one policy decision.
+    assert len(catalog._AUDIT_LOG) <= 4
+    assert len(evidence._POLICY_DECISION_LOG) <= 4
+
+    # And the reset itself must fully clear both logs.
+    invariants.reset_shared_state()
+    assert catalog._AUDIT_LOG == []
+    assert evidence._POLICY_DECISION_LOG == []