fix(orchestrator): close SQL source-table allowlist bypass via comma joins

docs/doctoring/dependency-security.md

@@ -0,0 +1,34 @@
+# Dependency security decisions
+
+## Cryptography 50.0.0 security floor
+
+### Decision
+
+All installable Python dependency surfaces must resolve `cryptography==50.0.0` or a later explicitly reviewed replacement. The runtime metadata, test resolver input, and generated hash-locked files are kept aligned so local development, CI, and production installation cannot silently reintroduce an affected release.
+
+### Security rationale
+
+GitHub Advisory Database entry GHSA-g6cj-pr64-35w5 describes a Bleichenbacher-style oracle in PKCS#7 EnvelopedData decryption caused by distinguishable errors and timing. The affected range is `cryptography>=44.0.0,<50.0.0`; version 50.0.0 is the first patched release. The repository previously resolved 49.0.0 through the `PyJWT[crypto]` dependency graph, so the direct security floor is intentional rather than relying on a transitive resolver choice.
+
+### Verification contract
+
+`tests/test_dependency_security.py` verifies that:
+
+1. `pyproject.toml` and `requirements-test.in` declare the patched floor;
+2. each installable hash-locked dependency file contains exactly one `cryptography==50.0.0` entry; and
+3. no stale affected pin remains in those installation surfaces.
+
+Regenerate the locks with the repository's pinned `uv` compiler and review the resulting graph before merge:
+
+```bash
+uv pip compile pyproject.toml --generate-hashes -o requirements.txt
+uv pip compile pyproject.toml --extra dev --generate-hashes -o requirements-dev.txt
+uv pip compile --generate-hashes --universal --python-version 3.12 \
+  requirements-test.in -o requirements-test.txt
+```
+
+Security scanning, tests, and exact-current-head review remain required after lock regeneration. A successful scan on an earlier head is not evidence for the regenerated graph.
+
+## Reference
+
+GitHub. (2026, August 3). *Cryptography: PKCS#7 EnvelopedData decryption exposes a Bleichenbacher oracle through distinguishable errors and timing (GHSA-g6cj-pr64-35w5).* GitHub Advisory Database. https://github.com/advisories/GHSA-g6cj-pr64-35w5

pyproject.toml

@@ -9,6 +9,7 @@ description = "Ontology-driven, graph-backed semantic data catalog with vector r
 readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
+  "cryptography==50.0.0",
   "fastapi==0.139.0",
   "psycopg[binary]==3.3.4",
   "PyJWT[crypto]==2.13.0",

requirements-test.in

@@ -1,5 +1,6 @@
 # Runtime pins mirror [project.dependencies] and the dev extra in
 # pyproject.toml; keep them in sync when bumping versions there.
+cryptography==50.0.0
 fastapi==0.139.0
 psycopg[binary]==3.3.4
 PyJWT[crypto]==2.13.0

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/orchestrator.py

@@ -11,6 +11,66 @@
 
 _FORBIDDEN_KEYWORDS = {"drop", "delete", "truncate", "alter", "insert", "update", "merge", "exec", "union"}
 
+# Keywords that terminate a FROM clause's relation list. Everything between a
+# ``FROM`` and the first of these is the (possibly comma-separated) table list.
+_CLAUSE_BOUNDARY = re.compile(
+    r"\b(where|group\s+by|order\s+by|having|limit|offset|window|"
+    r"union|intersect|except|fetch)\b",
+    re.IGNORECASE,
+)
+
+
+def _split_top_level_commas(value: str) -> list[str]:
+    """Split on commas outside parenthesized SQL regions.
+
+    This exposes implicit joins without misreading projection-list or function
+    argument commas inside a derived relation as table separators.
+    """
+
+    segments: list[str] = []
+    segment_start = 0
+    parenthesis_depth = 0
+    for index, character in enumerate(value):
+        if character == "(":
+            parenthesis_depth += 1
+        elif character == ")":
+            parenthesis_depth = max(0, parenthesis_depth - 1)
+        elif character == "," and parenthesis_depth == 0:
+            segments.append(value[segment_start:index])
+            segment_start = index + 1
+    segments.append(value[segment_start:])
+    return segments
+
+
+def _from_clause_tables(sql: str) -> list[str]:
+    """Return every relation named in each FROM clause of ``sql``.
+
+    Unlike a bare ``FROM <table>`` / ``JOIN <table>`` match, this also captures
+    comma-separated (implicit-join) relations such as the ``customer`` in
+    ``FROM crm, customer``, which would otherwise slip past the source-table
+    allowlist. Iterating over every ``FROM`` occurrence also covers subquery
+    FROM clauses, matching the previous scan's coverage.
+    """
+
+    tables: list[str] = []
+    for from_match in re.finditer(r"\bfrom\b", sql, re.IGNORECASE):
+        region = sql[from_match.end():]
+        boundary = _CLAUSE_BOUNDARY.search(region)
+        if boundary:
+            region = region[: boundary.start()]
+        # Split on JOIN so an ON-condition never swallows the next relation, then
+        # split each segment on top-level commas to expose implicit-join relations.
+        for join_segment in re.split(r"\bjoin\b", region, flags=re.IGNORECASE):
+            relation_part = re.split(r"\bon\b", join_segment, maxsplit=1, flags=re.IGNORECASE)[0]
+            for candidate in _split_top_level_commas(relation_part):
+                stripped_candidate = candidate.strip()
+                if not stripped_candidate or stripped_candidate.startswith("("):
+                    continue
+                identifier = re.match(r"[A-Za-z_][\w.]*", stripped_candidate.split()[0])
+                if identifier:
+                    tables.append(identifier.group(0))
+    return tables
+
 
 def _safe_identifier(value: str) -> str:
     return "".join(ch if ch.isalnum() or ch == "_" else "_" for ch in value)
@@ -45,10 +105,7 @@ def validate_sql_query(sql: str, *, source_system: str) -> list[str]:
             warnings.append("forbidden_keyword_detected")
             break
 
-    referenced = [
-        next(value for value in match if value)
-        for match in re.findall(r"\bfrom\s+([A-Za-z_][\w.]*)|\bjoin\s+([A-Za-z_][\w.]*)", stripped, re.IGNORECASE)
-    ]
+    referenced = _from_clause_tables(stripped)
     if not referenced:
         warnings.append("missing_source_table")
         return warnings

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_api.py

@@ -1508,6 +1508,38 @@ def test_browse_query_rejects_table_outside_dataset_binding():
     )
 
 
+def test_browse_query_rejects_comma_joined_unauthorized_table():
+    """A comma-separated (implicit-join) table outside the dataset binding must
+    be rejected by the source-table allowlist, exactly like an explicit
+    unauthorized FROM/JOIN target. The allowlisted table for crm-event is
+    ``crm`` (source_system s3://analytics/events/crm); smuggling a second table
+    via ``FROM crm, customer`` must not pass validation."""
+    response = client.post(
+        "/browse/query",
+        json={
+            "user": "analyst",
+            "purpose": "analysis",
+            "dataset_ids": ["crm-event"],
+            "language": "SQL",
+            "query": "SELECT count(*) AS active_count FROM crm, customer",
+        },
+    )
+    assert response.status_code == 400
+    assert "unauthorized_table_reference" in response.json()["detail"]["warnings"]
+
+
+def test_validate_sql_query_flags_comma_joined_table_outside_allowlist():
+    """Unit-level guard: validate_sql_query must flag a comma-joined relation
+    that is not the allowlisted source table."""
+    from sdp.orchestrator import validate_sql_query
+
+    warnings = validate_sql_query(
+        "SELECT count(*) FROM crm, customer",
+        source_system="s3://analytics/events/crm",
+    )
+    assert "unauthorized_table_reference" in warnings
+
+
 def test_browse_query_rejects_literal_tautology_injection():
     response = client.post(
         "/browse/query",