From 613ee4b292c4701ae8466f06e3e8c670f0e427f2 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Tue, 4 Aug 2026 17:37:24 +0900 Subject: [PATCH] fix(security): clear the central Semgrep SAST gate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Suppresses the 5 blocking Medium+ findings with inline justifications — all are false positives or documented, guarded opt-ins; no behavior change: - orchestrator.ModelClient._build_ssl_context: unverified-ssl-context is reachable only via the explicit verify_tls=False constructor opt-out (default True), the documented insecure dev-only escape hatch. - orchestrator.ModelClient._open_provider: dynamic-urllib audit hit — the URL passed _validate_provider (https-only, host allowlist, private-IP rejection) and _provider_url (scheme + path checks). - cost_ledger.SqlLedgerStore (x3): sqlalchemy-execute-raw-query on plain DB-API cursors where the f-strings interpolate only the paramstyle placeholder and the fixed _USAGE_COLUMNS constant; all values are bound driver parameters (each site already carries nosec B608). Verified locally with semgrep 1.169.0 and the CI flags: 0 findings, rc=0. Co-Authored-By: Claude Fable 5 --- contextual_orchestrator/cost_ledger.py | 6 +++--- contextual_orchestrator/orchestrator.py | 9 +++++++-- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5be..e22a7eeb7 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -583,7 +583,7 @@ def _seed_dimension_catalog(self) -> None: ph = self._placeholder() cur = self._conn.cursor() for order, (name, label, _column) in enumerate(ATTRIBUTION_DIMENSION_CATALOG): - cur.execute( + cur.execute( # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}", # nosec B608 - ph is a DB-API placeholder. (name,), ) @@ -602,7 +602,7 @@ def append(self, record: UsageRecord) -> None: placeholders = ", ".join(ph for _ in _USAGE_COLUMNS) columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute( + cur.execute( # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})", # nosec B608 - columns are fixed _USAGE_COLUMNS. tuple(row.get(column) for column in _USAGE_COLUMNS), ) @@ -622,7 +622,7 @@ def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[ where = f" WHERE {' AND '.join(clauses)}" if clauses else "" columns = ", ".join(_USAGE_COLUMNS) cur = self._conn.cursor() - cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. + cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. # nosemgrep: python.sqlalchemy.security.sqlalchemy-execute-raw-query.sqlalchemy-execute-raw-query return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722e..d084a3240 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -230,7 +230,9 @@ def __init__( @staticmethod def _build_ssl_context(ca_bundle: str | None, verify_tls: bool) -> ssl.SSLContext: if not verify_tls: - return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. + # Reached only via the explicit verify_tls=False constructor opt-out + # (default is True); documented insecure dev-only escape hatch. + return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. # nosemgrep: python.lang.security.unverified-ssl-context.unverified-ssl-context if ca_bundle: if not os.path.isfile(ca_bundle): raise ValueError(f"provider CA bundle does not exist: {ca_bundle}") @@ -307,7 +309,10 @@ def _send(self, agent: ModelAgent, payload: dict[str, Any]) -> str: def _open_provider(self, request: urllib.request.Request) -> Any: """Open a provider request built from a validated provider URL.""" - return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. + # The URL passed _validate_provider (https-only, host allowlist, private-IP + # rejection) and _provider_url (scheme + path checks), so file:// and + # attacker-chosen hosts are unreachable. + return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected request, timeout=self.timeout, context=self._ssl_context,