diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5be..48df2fae3 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/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()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722e..aafe29f34 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/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,