fix(sdp): unblock CI — fuzz OOM + crash-log visibility + base SAST remediation

src/sdp/authz.py

@@ -4,6 +4,7 @@
 import os
 from datetime import datetime, timezone
 from typing import Any
+from urllib.parse import urlsplit
 from urllib.request import urlopen
 
 import jwt
@@ -111,8 +112,19 @@ def resolve_oidc_actor_context(
     return ActorContext(subject=str(subject), tenant_id=tenant_id, roles=sorted(roles))
 
 
+_ALLOWED_JWKS_SCHEMES = frozenset({"https", "http"})
+
+
 def _load_jwks_from_url(jwks_url: str) -> dict[str, Any]:
+    # Restrict the JWKS fetch to HTTP(S). urllib honours file:// (and other
+    # schemes), so without this guard a misconfigured SDP_OIDC_JWKS_URL such as
+    # "file:///etc/passwd" would turn an operator misconfiguration into local
+    # file disclosure.
+    scheme = urlsplit(jwks_url).scheme.lower()
+    if scheme not in _ALLOWED_JWKS_SCHEMES:
+        raise ValueError("OIDC JWKS URL must use the http or https scheme")
     timeout = float(os.getenv("SDP_OIDC_JWKS_TIMEOUT_SECONDS", "2"))
+    # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- scheme is allow-listed to http(s) above; JWKS URL is operator config, not request input
     with urlopen(jwks_url, timeout=timeout) as response:
         return json.loads(response.read().decode("utf-8"))
 

src/sdp/demo_smoke.py

@@ -83,5 +83,5 @@ def main() -> int:
     return 0 if summary["ready"] else 1
 
 
-if __name__ == "__main__":
+if __name__ == "__main__":  # pragma: no cover - CLI entry, exercised only on direct execution
     raise SystemExit(main())

src/sdp/graph_store.py

@@ -475,6 +475,7 @@ def _cypher(
         )
         driver_connection = conn.connection.driver_connection
         with driver_connection.cursor() as cursor:
+            # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query -- not raw string SQL: statement is a psycopg sql.Composed of sql.Literal(graph_name)/sql.Literal(query) with the AS-column decl from a closed allow-list map; params are bound as a positional driver parameter
             cursor.execute(statement, (json.dumps(params),))
             return cursor.fetchall()
 

src/sdp/observability.py

@@ -136,6 +136,7 @@ def _export_to_sink(observation: dict[str, Any]) -> None:
             headers={"Content-Type": "application/json"},
             method="POST",
         )
+        # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- scheme is allow-listed to {http,https} at the guard above (file:// et al. raise below)
         with urlopen(request, timeout=timeout_ms / 1000):
             return
 

src/sdp_core/contracts.py

@@ -251,6 +251,7 @@ class DatasetPatchRequest(BaseModel):
     tags: Optional[list[str]] = None
     terms: Optional[list[str]] = None
     related_datasets: Optional[list[str]] = None
+    schema: Optional[list[ColumnMetadata]] = None
     lineage_inputs: Optional[list[str]] = None
     lineage_outputs: Optional[list[str]] = None
     mappings: Optional[list[BusinessMapping]] = None

tests/fuzz/README.md

@@ -47,15 +47,20 @@ PYTHONPATH=src:. FUZZ_SECONDS=60 tests/fuzz/run_atheris.sh
 PYTHONPATH=src:. python tests/fuzz/atheris/fuzz_draft_sql.py -max_total_time=30 tests/fuzz/corpus/draft_sql
 ```
 
-Seed corpora live in `corpus/<target>/`. A reproducing crash is written to a
-`crash-*` file in the working directory; re-run the harness with that file as
-its argument to replay it.
+Seed corpora live in `corpus/<target>/`. A reproducing failure is written to a
+`crash-*`, `oom-*`, or `timeout-*` file in the working directory; re-run the
+corresponding harness with that file as its argument to replay it. On failure
+the runner identifies the new reproducer across all three patterns and prints
+its base64 payload, SHA-256 digest, and replay command directly to the log, so
+the failure cause is diagnosable without downloading the artifact.
 
 ## CI
 
 `.github/workflows/fuzz.yml` runs the property tests plus a **bounded** Atheris
 job (60s/target on PRs, 300s nightly via `schedule`) so fuzzing never blows CI
-cost. A crash fails the job and uploads the `crash-*` artifact.
+cost. A crash, out-of-memory termination, or timeout fails the job, prints the
+reproducer to the run log, and uploads matching `crash-*`, `oom-*`, and
+`timeout-*` files as workflow artifacts.
 
 ## Background
 

tests/fuzz/invariants.py

@@ -10,7 +10,7 @@
 
 from typing import Any
 
-from sdp import catalog, ontology, orchestrator
+from sdp import catalog, evidence, observability, ontology, orchestrator
 from sdp.domain import QueryDraftRequest, QueryExecutionRequest, QueryExecutionResponse
 
 # Anything the query drafter/executor must never let through unescaped.
@@ -24,6 +24,26 @@
 _SQL_METACHARACTERS = set(" \t\r\n'\";()[]{}`*/\\%+-.,=<>!&|@#?:")
 
 
+def reset_in_memory_state() -> None:
+    """Clear the in-memory append-only stores between fuzz iterations.
+
+    ``execute_query`` and ``draft_sql`` record a policy decision plus an audit
+    event on every call. Across a bounded-time coverage-guided run that is
+    hundreds of thousands of iterations, so the unbounded ``list`` accumulators
+    (``catalog._AUDIT_LOG``, ``evidence._POLICY_DECISION_LOG``) grew the process
+    past libFuzzer's rss limit and aborted the run with an ``oom-*`` exit — a
+    fuzz-harness state-accumulation artifact, not a logic bug in the code under
+    test. The pytest suite avoids this with an autouse isolation fixture; the
+    fuzz harnesses have none, so reset here. Only append-only growth is cleared;
+    the seeded catalog datasets (``catalog._DATA``) are left intact so the code
+    under test still has data to operate on.
+    """
+    catalog._AUDIT_LOG.clear()
+    catalog._SCHEMA_HISTORY.clear()
+    evidence._POLICY_DECISION_LOG.clear()
+    observability.reset_request_observability()
+
+
 def check_safe_identifier(value: str) -> None:
     """``_safe_identifier`` must neutralise every SQL metacharacter and all
     whitespace — the core identifier-injection guard — and be idempotent."""
@@ -83,6 +103,7 @@ def check_draft_sql(req: QueryDraftRequest) -> None:
     """``orchestrator.draft_sql``: no crash; if a SQL string is returned it stays
     within bounds and never smuggles a forbidden keyword or SQL metacharacter in
     via a user-controlled identifier (group_by / columns)."""
+    reset_in_memory_state()  # bound append-only evidence growth across iterations
     result = orchestrator.draft_sql(req)
     assert isinstance(result, dict)
     if "query" not in result:
@@ -105,6 +126,7 @@ def check_draft_sql(req: QueryDraftRequest) -> None:
 def check_execute_query(req: QueryExecutionRequest) -> None:
     """``orchestrator.execute_query``: always returns a response; hostile SQL is
     rejected rather than marked SUCCEEDED."""
+    reset_in_memory_state()  # bound append-only evidence growth across iterations
     resp = orchestrator.execute_query(req)
     assert isinstance(resp, QueryExecutionResponse)
     assert resp.status in {"SUCCEEDED", "REJECTED", "DENIED"}

tests/fuzz/run_atheris.sh

@@ -23,6 +23,12 @@ for harness in "${HARNESS_DIR}"/fuzz_*.py; do
   name="$(basename "${harness}" .py)"
   target="${name#fuzz_}"
   corpus="${CORPUS_DIR}/${target}"
+  # Snapshot pre-existing reproducer files so a crash can be attributed to the
+  # target that just produced it (libFuzzer writes crash-/oom-/timeout- files
+  # into the CWD, which persist across targets within one run).
+  shopt -s nullglob
+  before=(crash-* oom-* timeout-*)
+  shopt -u nullglob
   echo "::group::fuzz ${target} (${SECONDS_PER_TARGET}s)"
   # -max_total_time bounds the run; -close_fd_mask keeps libFuzzer output tidy.
   python "${harness}" \
@@ -32,7 +38,28 @@ for harness in "${HARNESS_DIR}"/fuzz_*.py; do
   rc=$?
   echo "::endgroup::"
   if [ "${rc}" -ne 0 ]; then
-    echo "FUZZ FAILURE: ${target} exited with ${rc} (see crash artifact above)"
+    echo "FUZZ FAILURE: ${target} exited with ${rc}"
+    # Surface the reproducer in the log so the failure cause is diagnosable
+    # directly from the run (the crash artifact is also uploaded, but log
+    # visibility means no artifact download is needed to reproduce).
+    shopt -s nullglob
+    after=(crash-* oom-* timeout-*)
+    shopt -u nullglob
+    for repro in "${after[@]}"; do
+      is_new=1
+      for old in "${before[@]}"; do
+        [ "${repro}" = "${old}" ] && is_new=0 && break
+      done
+      [ "${is_new}" -eq 0 ] && continue
+      echo "::group::crash reproducer ${repro} (target ${target})"
+      echo "target=${target} file=${repro} bytes=$(wc -c <"${repro}") sha256=$(sha256sum "${repro}" | cut -d' ' -f1)"
+      replay_pythonpath="$(printf '%q' "${REPO_ROOT}:${REPO_ROOT}/src")"
+      replay_harness="$(printf '%q' "${harness}")"
+      echo "reproduce locally: base64 -d > repro.bin <<'B64' && PYTHONPATH=${replay_pythonpath} python ${replay_harness} repro.bin"
+      base64 "${repro}"
+      echo "B64"
+      echo "::endgroup::"
+    done
     status=1
   fi
 done

tests/test_authz.py

@@ -0,0 +1,175 @@
+"""Tests for OIDC JWKS loading, focused on the URL-scheme hardening."""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from sdp import authz
+
+
+def test_load_jwks_from_url_rejects_non_http_schemes(monkeypatch):
+    """A misconfigured non-http(s) JWKS URL must be rejected before any fetch,
+    so urllib's ``file://`` support cannot be turned into local file disclosure."""
+    unexpected_calls = []
+
+    def _unexpected_urlopen(*args, **kwargs):
+        unexpected_calls.append((args, kwargs))
+        raise AssertionError("urlopen must not be called for a rejected URL scheme")
+
+    monkeypatch.setattr(authz, "urlopen", _unexpected_urlopen)
+    for bad_url in ("file:///etc/passwd", "ftp://host/keys.json", "gopher://x", ""):
+        with pytest.raises(ValueError):
+            authz._load_jwks_from_url(bad_url)
+
+    assert unexpected_calls == []
+
+
+def test_load_jwks_from_url_fetches_over_https(monkeypatch):
+    """An https JWKS URL passes the scheme allow-list and its JSON body is
+    parsed and returned."""
+    payload = {"keys": [{"kid": "abc", "kty": "RSA"}]}
+
+    class _FakeResponse:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+        def read(self):
+            return json.dumps(payload).encode("utf-8")
+
+    captured = {}
+
+    def _fake_urlopen(url, timeout=None):
+        captured["url"] = url
+        captured["timeout"] = timeout
+        return _FakeResponse()
+
+    monkeypatch.delenv("SDP_OIDC_JWKS_TIMEOUT_SECONDS", raising=False)
+    monkeypatch.setattr(authz, "urlopen", _fake_urlopen)
+    result = authz._load_jwks_from_url("https://idp.example/.well-known/jwks.json")
+
+    assert result == payload
+    assert captured["url"] == "https://idp.example/.well-known/jwks.json"
+    assert captured["timeout"] == pytest.approx(2.0)
+
+
+def test_load_jwks_from_url_honours_timeout_override(monkeypatch):
+    """The JWKS fetch timeout is configurable via SDP_OIDC_JWKS_TIMEOUT_SECONDS."""
+    monkeypatch.setenv("SDP_OIDC_JWKS_TIMEOUT_SECONDS", "5")
+
+    class _FakeResponse:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+        def read(self):
+            return b"{}"
+
+    seen = {}
+
+    def _fake_urlopen(url, timeout=None):
+        seen["timeout"] = timeout
+        return _FakeResponse()
+
+    monkeypatch.setattr(authz, "urlopen", _fake_urlopen)
+    assert authz._load_jwks_from_url("http://localhost:8080/jwks") == {}
+    assert seen["timeout"] == pytest.approx(5.0)
+
+
+# --- OIDC claim/role/JWK guard branches (security-critical error paths) ---
+
+import time
+
+import jwt as _jwt
+
+
+def test_claim_values_handles_str_list_none_and_scalar():
+    assert authz._claim_values(None) == []
+    assert authz._claim_values("one") == ["one"]
+    assert authz._claim_values(["a", 2]) == ["a", "2"]
+    assert authz._claim_values(7) == ["7"]  # non-str, non-list scalar
+
+
+def test_load_oidc_role_map_default_and_override(monkeypatch):
+    monkeypatch.delenv("SDP_OIDC_GROUP_ROLE_MAP", raising=False)
+    assert authz.load_oidc_role_map() == authz._DEFAULT_OIDC_GROUP_ROLE_MAP
+
+    monkeypatch.setenv("SDP_OIDC_GROUP_ROLE_MAP", '{"grp": ["data-analyst"]}')
+    assert authz.load_oidc_role_map() == {"grp": ["data-analyst"]}
+
+    monkeypatch.setenv("SDP_OIDC_GROUP_ROLE_MAP", "[]")
+    with pytest.raises(ValueError):
+        authz.load_oidc_role_map()
+
+
+def _valid_claims(**overrides):
+    claims = {
+        "preferred_username": "alice",
+        "tenant_id": "demo",
+        "exp": int(time.time()) + 3600,
+    }
+    claims.update(overrides)
+    return claims
+
+
+def test_validate_oidc_claim_shape_guard_branches():
+    authz.validate_oidc_claim_shape(_valid_claims())  # happy path
+    with pytest.raises(ValueError):  # missing subject
+        authz.validate_oidc_claim_shape({"tenant_id": "d", "exp": int(time.time()) + 60})
+    with pytest.raises(ValueError):  # missing tenant
+        authz.validate_oidc_claim_shape({"sub": "s", "exp": int(time.time()) + 60})
+    with pytest.raises(ValueError):  # missing exp
+        authz.validate_oidc_claim_shape({"sub": "s", "tenant_id": "d"})
+    with pytest.raises(ValueError):  # invalid exp type
+        authz.validate_oidc_claim_shape({"sub": "s", "tenant_id": "d", "exp": "soon"})
+    with pytest.raises(ValueError):  # expired
+        authz.validate_oidc_claim_shape({"sub": "s", "tenant_id": "d", "exp": 1})
+
+
+def test_select_jwk_guard_branches():
+    jwks = {"keys": [{"kid": "k1", "kty": "RSA"}]}
+    assert authz._select_jwk(jwks, "k1")["kid"] == "k1"
+    with pytest.raises(ValueError):  # missing kid
+        authz._select_jwk(jwks, None)
+    with pytest.raises(ValueError):  # keys not a list
+        authz._select_jwk({"keys": "nope"}, "k1")
+    with pytest.raises(ValueError):  # no matching kid
+        authz._select_jwk(jwks, "absent")
+
+
+def test_verify_oidc_jwks_token_config_and_alg_guards(monkeypatch):
+    monkeypatch.delenv("SDP_OIDC_ISSUER", raising=False)
+    monkeypatch.delenv("SDP_OIDC_AUDIENCE", raising=False)
+    monkeypatch.delenv("SDP_OIDC_JWKS_URL", raising=False)
+    with pytest.raises(ValueError):  # missing issuer
+        authz.verify_oidc_jwks_token("t", jwks={"keys": []})
+    with pytest.raises(ValueError):  # missing audience
+        authz.verify_oidc_jwks_token("t", issuer="iss", jwks={"keys": []})
+    with pytest.raises(ValueError):  # jwks is None and no JWKS URL configured
+        authz.verify_oidc_jwks_token("t", issuer="iss", audience="aud")
+
+    # Unsupported algorithm is rejected before signature verification.
+    hs_token = _jwt.encode({"sub": "s"}, "secret", algorithm="HS256")
+    with pytest.raises(ValueError):
+        authz.verify_oidc_jwks_token(hs_token, issuer="iss", audience="aud", jwks={"keys": []})
+
+
+def test_verify_oidc_jwks_token_loads_jwks_from_env_url(monkeypatch):
+    monkeypatch.setenv("SDP_OIDC_JWKS_URL", "https://idp.example/jwks")
+
+    def _fake_load(url):
+        assert url == "https://idp.example/jwks"
+        return {"keys": []}
+
+    monkeypatch.setattr(authz, "_load_jwks_from_url", _fake_load)
+    # Unsupported alg is rejected after the env JWKS is loaded -> wrapped ValueError,
+    # which exercises the `jwks = _load_jwks_from_url(jwks_url)` branch.
+    hs_token = _jwt.encode({"sub": "s"}, "secret", algorithm="HS256")
+    with pytest.raises(ValueError):
+        authz.verify_oidc_jwks_token(hs_token, issuer="iss", audience="aud")