fix: clear the four org Semgrep gate findings

contextual_orchestrator/cost_ledger.py

@@ -566,12 +566,40 @@ class SqlLedgerStore:
     def __init__(self, connection: Any, paramstyle: str = "qmark") -> None:
         self._conn = connection
         self._paramstyle = paramstyle
+        self._prepare_statements()
         self._create_schema()
         self._seed_dimension_catalog()
 
     def _placeholder(self) -> str:
         return "?" if self._paramstyle == "qmark" else "%s"
 
+    def _prepare_statements(self) -> None:
+        """Compose all SQL once, away from the ``execute()`` call sites.
+
+        Identifiers come from fixed module constants and the placeholder from
+        the driver paramstyle; every runtime value travels as a bound parameter.
+        """
+        ph = self._placeholder()
+        columns = ", ".join(_USAGE_COLUMNS)
+        placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
+        self._seed_select_sql = (
+            f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}"  # nosec B608 - ph is a DB-API placeholder.
+        )
+        self._seed_insert_sql = (
+            "INSERT INTO cost_attribution_dimensions "
+            f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})"  # nosec B608 - ph is a DB-API placeholder.
+        )
+        self._usage_insert_sql = (
+            f"INSERT INTO llm_usage_records ({columns}) VALUES ({placeholders})"  # nosec B608 - columns are fixed _USAGE_COLUMNS.
+        )
+        base_select = f"SELECT {columns} FROM llm_usage_records"  # nosec B608 - columns are fixed _USAGE_COLUMNS.
+        self._usage_select_sql = {
+            (False, False): base_select,
+            (True, False): f"{base_select} WHERE created_at >= {ph}",
+            (False, True): f"{base_select} WHERE created_at < {ph}",
+            (True, True): f"{base_select} WHERE created_at >= {ph} AND created_at < {ph}",
+        }
+
     def _create_schema(self) -> None:
         cur = self._conn.cursor()
         for statement in SCHEMA_SQL.strip().split(";"):
@@ -580,49 +608,25 @@ def _create_schema(self) -> None:
         self._conn.commit()
 
     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(
-                f"SELECT 1 FROM cost_attribution_dimensions WHERE dimension_name = {ph}",  # nosec B608 - ph is a DB-API placeholder.
-                (name,),
-            )
+            cur.execute(self._seed_select_sql, (name,))
             if cur.fetchone() is None:
-                cur.execute(
-                    "INSERT INTO cost_attribution_dimensions "
-                    f"(dimension_name, dimension_label, dimension_order) VALUES ({ph}, {ph}, {ph})",  # nosec B608 - ph is a DB-API placeholder.
-                    (name, label, order),
-                )
+                cur.execute(self._seed_insert_sql, (name, label, order))
         self._conn.commit()
 
     def append(self, record: UsageRecord) -> None:
         """Insert a usage record row."""
         row = record.as_dict()
-        ph = self._placeholder()
-        placeholders = ", ".join(ph for _ in _USAGE_COLUMNS)
-        columns = ", ".join(_USAGE_COLUMNS)
         cur = self._conn.cursor()
-        cur.execute(
-            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),
-        )
+        cur.execute(self._usage_insert_sql, tuple(row.get(column) for column in _USAGE_COLUMNS))
         self._conn.commit()
 
     def query(self, start: Optional[int] = None, end: Optional[int] = None) -> List[Dict[str, Any]]:
         """Return record rows in the optional half-open window."""
-        ph = self._placeholder()
-        clauses: List[str] = []
-        params: List[Any] = []
-        if start is not None:
-            clauses.append(f"created_at >= {ph}")
-            params.append(start)
-        if end is not None:
-            clauses.append(f"created_at < {ph}")
-            params.append(end)
-        where = f" WHERE {' AND '.join(clauses)}" if clauses else ""
-        columns = ", ".join(_USAGE_COLUMNS)
+        params = tuple(value for value in (start, end) if value is not None)
         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(self._usage_select_sql[(start is not None, end is not None)], params)
         return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()]
 
 

contextual_orchestrator/orchestrator.py

@@ -230,7 +230,12 @@ 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.
+            # Explicit dev-only opt-out (insecure) for self-signed endpoints: start
+            # from the default verifying context, then drop verification.
+            context = ssl.create_default_context()
+            context.check_hostname = False
+            context.verify_mode = ssl.CERT_NONE
+            return context
         if ca_bundle:
             if not os.path.isfile(ca_bundle):
                 raise ValueError(f"provider CA bundle does not exist: {ca_bundle}")
@@ -307,7 +312,9 @@ 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 is built by _provider_url after _validate_provider (https-only, host
+        # allowlist, private/loopback rejection), so file:// and attacker hosts are unreachable.
+        return urllib.request.urlopen(  # nosec B310  # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
             request,
             timeout=self.timeout,
             context=self._ssl_context,