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,