From 5f547c622b63f4e3bac3d3fc13a029c137861530 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 20:10:04 +0900 Subject: [PATCH 1/7] fix(security): minimal healthz, request framing, trace authority Split unauthenticated /healthz liveness (status+service only) from admin-authenticated /readyz inventory (closes #118). Fail-closed Content-Length validation rejects negative/non-decimal/missing/chunked framing before socket reads (closes #119). Orchestration traces require verified authority beyond inference; JSON boolean coercion is strict (closes #117). --- README.md | 2 +- contextual_orchestrator/server.py | 105 +++++++++++++++++++--- tests/test_cost_review_server.py | 4 +- tests/test_healthz.py | 65 ++++++++++++-- tests/test_request_framing.py | 135 ++++++++++++++++++++++++++++ tests/test_trace_authority.py | 144 ++++++++++++++++++++++++++++++ 6 files changed, 435 insertions(+), 20 deletions(-) create mode 100644 tests/test_request_framing.py create mode 100644 tests/test_trace_authority.py diff --git a/README.md b/README.md index 65f57dd4..b66deaeb 100644 --- a/README.md +++ b/README.md @@ -192,7 +192,7 @@ is read from a **KV config store**, never `os.getenv`. backend (local in-process backend standalone), and records one usage-ledger row per original vector with the full attribution dimensions (service, team, group, company, provider) carried in `metadata`. -- **Health.** `GET /healthz` is an unauthenticated liveness probe. +- **Health.** `GET /healthz` is an unauthenticated minimal liveness probe (`status` + `service` only). `GET /readyz` is admin-authenticated readiness with agent/backend/usage inventory for operators. - **Standalone + optional pg-llm-batch integration.** The hub runs standalone with the in-memory config store and local batch backend; wiring a Postgres DSN and an installed/deployed `pg_llm_batch` client activates the KV/secret stores, diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index c58d4cb7..bc27c686 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -116,6 +116,23 @@ def authorize(self, headers: Any, scope: str, client_address: str) -> None: if not expected or not secrets.compare_digest(token, expected): raise RequestError(401, "unauthorized", "bearer token is invalid for this scope") + def may_disclose_trace(self, headers: Any, scope: str) -> bool: + """Return True when the verified caller may receive orchestration traces. + + Inference-only credentials never receive planner/workflow evidence. + Admin-scope callers may. Single-token deployments may when the host sets + ``expose_trace_by_default`` or the request is handled under admin scope. + """ + if scope == "admin": + return True + if scope != "inference": + return False + # Split-token inference path: never disclose traces. + if self.admin_token and self.inference_token and not self.auth_token: + return False + # Single shared token: host policy gate only (still needs request bool). + return bool(self.expose_trace_by_default) + def check_rate_limit(self, key: str) -> None: """Apply a simple per-client fixed-window request budget.""" now = time.monotonic() @@ -171,6 +188,69 @@ def _coerce_json(payload: bytes) -> dict[str, Any]: return value +def _coerce_optional_bool(value: Any, field_name: str) -> bool | None: + """Parse an optional JSON bool fail-closed; reject truthy strings/numbers.""" + if value is None: + return None + if isinstance(value, bool): + return value + raise RequestError( + 400, + "invalid_boolean", + f"{field_name} must be a JSON boolean", + {"field": field_name}, + ) + + +def _parse_content_length(headers: Any, max_body_bytes: int) -> int: + """Return a validated Content-Length for fixed-length JSON request bodies. + + Rejects missing, negative, non-decimal, overflow, and transfer-coded framing + so ``rfile.read`` never receives a negative size or unbounded read. + """ + # BaseHTTPRequestHandler collapses duplicate headers with commas; treat that + # as ambiguous framing and reject rather than guessing. + raw = headers.get("content-length") + if raw is None or raw == "": + raise RequestError(411, "length_required", "Content-Length is required for JSON request bodies") + if isinstance(raw, str) and "," in raw: + raise RequestError(400, "invalid_content_length", "duplicate or ambiguous Content-Length") + text = str(raw).strip() + if not text.isdigit(): + # Reject signed forms ("-1", "+10"), hex, and non-decimal tokens. + raise RequestError(400, "invalid_content_length", "Content-Length must be an unsigned decimal integer") + # Leading zeros are allowed by isdigit; value itself must fit max. + body_size = int(text) + if body_size > max_body_bytes: + raise RequestError(413, "request_too_large", "request body exceeds configured limit") + transfer = (headers.get("transfer-encoding") or "").strip() + if transfer and transfer.lower() != "identity": + raise RequestError( + 400, + "unsupported_transfer_encoding", + "chunked or non-identity Transfer-Encoding is not accepted for JSON bodies", + ) + return body_size + + +def _resolve_include_trace( + body: dict[str, Any], + security: SecurityConfig, + headers: Any, + scope: str, +) -> bool: + """Fail-closed orchestration-trace disclosure decision. + + Requires (1) verified trace authority for the caller scope and (2) an + explicit JSON boolean request flag when the host default is off. + """ + if not security.may_disclose_trace(headers, scope): + return False + requested = _coerce_optional_bool(body.get("include_orchestration_trace"), "include_orchestration_trace") + if requested is None: + return bool(security.expose_trace_by_default) + return requested + def _reject_unknown_keys(body: dict[str, Any], allowed: set[str]) -> None: unknown = sorted(set(body) - allowed) if unknown: @@ -336,9 +416,14 @@ def do_GET(self) -> None: # noqa: N802 self._send(OPENAPI_SPEC) return if path == "/healthz": - # Unauthenticated liveness probe for containers/orchestrators. + # Unauthenticated liveness only: process is up. No inventory. + self._send({"status": "ok", "service": "contextual-orchestrator"}) + return + if path == "/readyz": + # Authenticated readiness/diagnostics: operator inventory. + self._authorize("admin") self._send({ - "status": "ok", + "status": "ready", "service": "contextual-orchestrator", "agent_count": len(orchestrator.agents), "batch_backend": coordinator.batch_backend.name, @@ -733,7 +818,7 @@ def do_POST(self) -> None: # noqa: N802 return messages = _validate_messages(body.get("messages")) mode = _validate_mode(body.get("orchestration") or body.get("orchestration_mode") or body.get("mode") or "auto") - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) stream = body.get("stream", False) if not isinstance(stream, bool): raise RequestError(400, "invalid_request", "stream must be a boolean") @@ -856,7 +941,7 @@ def do_POST(self) -> None: # noqa: N802 except KeyError: self._send_error(404, "batch_job_not_found", f"batch job {job_id} not found") return - self._send(_response_payload(retrieved, include_trace=True)) + self._send(_response_payload(retrieved, include_trace=security.may_disclose_trace(self.headers, "inference"))) return if path == "/v1/responses": # The Responses API has no chat-completions verifier equivalent, @@ -884,7 +969,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompt, str): raise RequestError(400, "invalid_request", "prompt must be a string") mode = _validate_mode(body.get("mode", "auto")) - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode)) self._send(_response_payload(result, include_trace)) return @@ -894,7 +979,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompt, str) or not prompt: raise RequestError(400, "invalid_request", "prompt_text is required") mode = _validate_mode(body.get("run_mode", "auto")) - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode)) self._send(_response_payload(result, include_trace), 201) return @@ -906,7 +991,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompts, list) or not prompts: raise RequestError(400, "invalid_request", "prompts must be a non-empty array") mode = _validate_mode(body.get("run_mode", "auto")) - include_trace = bool(body.get("include_orchestration_trace", security.expose_trace_by_default)) + include_trace = _resolve_include_trace(body, security, self.headers, scope) evaluation_run = self._run(lambda: orchestrator.run_evaluation([str(item) for item in prompts], mode=mode)) self._send(_response_payload(evaluation_run, include_trace), 201) return @@ -960,10 +1045,10 @@ def _parse_optional_int(self, query: dict[str, list[str]], field_name: str) -> i def _read_json(self) -> dict[str, Any]: if self.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json": raise RequestError(415, "unsupported_media_type", "content-type must be application/json") - body_size = int(self.headers.get("content-length", "0")) - if body_size > security.max_body_bytes: - raise RequestError(413, "request_too_large", "request body exceeds configured limit") + body_size = _parse_content_length(self.headers, security.max_body_bytes) raw = self.rfile.read(body_size) + if len(raw) != body_size: + raise RequestError(400, "incomplete_body", "request body shorter than Content-Length") return _coerce_json(raw) if raw else {} def log_message(self, format: str, *args: object) -> None: diff --git a/tests/test_cost_review_server.py b/tests/test_cost_review_server.py index fd27c5bc..55cc96f9 100644 --- a/tests/test_cost_review_server.py +++ b/tests/test_cost_review_server.py @@ -56,9 +56,7 @@ def test_healthz_is_unauthenticated_and_ok() -> None: finally: server.shutdown() assert status == 200 - assert body["status"] == "ok" - assert body["service"] == "contextual-orchestrator" - assert "batch_backend" in body + assert body == {"status": "ok", "service": "contextual-orchestrator"} def test_chat_completion_reports_real_usage_and_records_cost() -> None: diff --git a/tests/test_healthz.py b/tests/test_healthz.py index 90dfc03c..668ad1b4 100644 --- a/tests/test_healthz.py +++ b/tests/test_healthz.py @@ -1,9 +1,10 @@ -"""Container liveness probe: /healthz must answer without any auth token.""" +"""Container liveness vs readiness probe contracts.""" from __future__ import annotations import json import sys import threading +import urllib.error import urllib.request from pathlib import Path @@ -14,18 +15,24 @@ from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 +_TEST_ADMIN = "admin_secret" # noqa: S105 +_TEST_INFERENCE = "inference_secret" # noqa: S105 -def test_healthz_is_unauthenticated_liveness() -> None: + +def _start(): orchestrator = TaskOrchestrator([ModelAgent("probe_agent", "mock-agent", tags=("reasoning",))]) server = build_server( orchestrator, port=0, - security=SecurityConfig(admin_token="admin_secret", inference_token="inference_secret"), + security=SecurityConfig(admin_token=_TEST_ADMIN, inference_token=_TEST_INFERENCE), ) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() - port = server.server_address[1] + return server, thread, server.server_address[1] + +def test_healthz_is_unauthenticated_minimal_liveness() -> None: + server, thread, port = _start() try: with urllib.request.urlopen(f"http://127.0.0.1:{port}/healthz", timeout=5) as response: status = response.status @@ -35,7 +42,52 @@ def test_healthz_is_unauthenticated_liveness() -> None: thread.join(timeout=5) assert status == 200 - assert body["status"] == "ok" + assert body == {"status": "ok", "service": "contextual-orchestrator"} + # Operational inventory must not leak on unauthenticated liveness. + for forbidden in ( + "agent_count", + "batch_backend", + "embedding_batch_backend", + "usage_record_count", + "agents", + "ready", + ): + assert forbidden not in body + + +def test_readyz_requires_admin_and_exposes_inventory() -> None: + server, thread, port = _start() + try: + unauth = urllib.request.Request(f"http://127.0.0.1:{port}/readyz") + try: + urllib.request.urlopen(unauth, timeout=5) + raise AssertionError("readyz must require auth") + except urllib.error.HTTPError as exc: + assert exc.code == 401 + + inference = urllib.request.Request( + f"http://127.0.0.1:{port}/readyz", + headers={"authorization": f"Bearer {_TEST_INFERENCE}"}, + ) + try: + urllib.request.urlopen(inference, timeout=5) + raise AssertionError("readyz must not accept inference token") + except urllib.error.HTTPError as exc: + assert exc.code == 401 + + admin = urllib.request.Request( + f"http://127.0.0.1:{port}/readyz", + headers={"authorization": f"Bearer {_TEST_ADMIN}"}, + ) + with urllib.request.urlopen(admin, timeout=5) as response: + status = response.status + body = json.loads(response.read().decode("utf-8")) + finally: + server.shutdown() + thread.join(timeout=5) + + assert status == 200 + assert body["status"] == "ready" assert body["service"] == "contextual-orchestrator" assert body["agent_count"] == 1 assert body["batch_backend"] @@ -44,5 +96,6 @@ def test_healthz_is_unauthenticated_liveness() -> None: if __name__ == "__main__": - test_healthz_is_unauthenticated_liveness() + test_healthz_is_unauthenticated_minimal_liveness() + test_readyz_requires_admin_and_exposes_inventory() print("ok") diff --git a/tests/test_request_framing.py b/tests/test_request_framing.py new file mode 100644 index 00000000..52c4f5ec --- /dev/null +++ b/tests/test_request_framing.py @@ -0,0 +1,135 @@ +"""Fail-closed inbound HTTP request framing for JSON bodies.""" +from __future__ import annotations + +import json +import socket +import sys +import threading +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + +_TOKEN = "secret_token" # noqa: S105 + + +def _start(): + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning",))]) + server = build_server(orch, port=0, security=SecurityConfig(auth_token=_TOKEN, max_body_bytes=64)) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread, server.server_address[1] + + +def _raw_post(port: int, extra_headers: list[str], body: bytes) -> tuple[int, bytes]: + request = ( + f"POST /v1/chat/completions HTTP/1.1\r\n" + f"Host: 127.0.0.1:{port}\r\n" + f"Authorization: Bearer {_TOKEN}\r\n" + f"Connection: close\r\n" + + "".join(h + "\r\n" for h in extra_headers) + + "\r\n" + ).encode("ascii") + body + with socket.create_connection(("127.0.0.1", port), timeout=3) as sock: + sock.settimeout(3) + sock.sendall(request) + chunks: list[bytes] = [] + while True: + try: + data = sock.recv(4096) + except TimeoutError: + break + if not data: + break + chunks.append(data) + raw = b"".join(chunks) + status_line = raw.split(b"\r\n", 1)[0] + code = int(status_line.split()[1]) + return code, raw + + +def test_negative_content_length_is_rejected() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", "Content-Length: -1"], + b'{"messages":[{"role":"user","content":"x"}]}', + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 400 + assert b"invalid_content_length" in raw or b"invalid_request" in raw or b"request_framing" in raw + + +def test_non_decimal_content_length_is_rejected() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", "Content-Length: 12abc"], + b"{}", + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 400 + + +def test_oversized_content_length_is_rejected() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", "Content-Length: 65"], + b"x" * 65, + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 413 + assert b"request_too_large" in raw + + +def test_missing_content_length_is_rejected_for_json_post() -> None: + server, thread, port = _start() + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json"], + b'{"messages":[{"role":"user","content":"x"}]}', + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 411 or code == 400 + + +def test_valid_small_json_body_still_works() -> None: + server, thread, port = _start() + body = b'{"messages":[{"role":"user","content":"hi"}]}' + try: + code, raw = _raw_post( + port, + ["Content-Type: application/json", f"Content-Length: {len(body)}"], + body, + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 200 + assert b"chat.completion" in raw or b"choices" in raw + + +if __name__ == "__main__": + test_negative_content_length_is_rejected() + test_non_decimal_content_length_is_rejected() + test_oversized_content_length_is_rejected() + test_missing_content_length_is_rejected_for_json_post() + test_valid_small_json_body_still_works() + print("ok") diff --git a/tests/test_trace_authority.py b/tests/test_trace_authority.py new file mode 100644 index 00000000..2343420f --- /dev/null +++ b/tests/test_trace_authority.py @@ -0,0 +1,144 @@ +"""Trace disclosure requires authority beyond ordinary inference access.""" +from __future__ import annotations + +import json +import sys +import threading +import urllib.error +import urllib.request +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from contextual_orchestrator import ModelAgent, TaskOrchestrator # noqa: E402 +from contextual_orchestrator.server import SecurityConfig, build_server # noqa: E402 + +_ADMIN = "admin_secret" # noqa: S105 +_INFERENCE = "inference_secret" # noqa: S105 +_SINGLE = "single_token" # noqa: S105 + + +def post(url: str, payload: dict, token: str) -> tuple[int, dict]: + req = urllib.request.Request( + url, + data=json.dumps(payload).encode("utf-8"), + headers={ + "content-type": "application/json", + "authorization": f"Bearer {token}", + "connection": "close", + }, + method="POST", + ) + try: + with urllib.request.urlopen(req, timeout=5) as response: + return response.status, json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + +def test_inference_token_cannot_obtain_orchestration_trace() -> None: + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + server = build_server( + orch, + port=0, + security=SecurityConfig(admin_token=_ADMIN, inference_token=_INFERENCE), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + payload = { + "messages": [{"role": "user", "content": "hello"}], + "include_orchestration_trace": True, + } + try: + status, body = post(f"http://127.0.0.1:{port}/v1/chat/completions", payload, _INFERENCE) + finally: + server.shutdown() + thread.join(timeout=5) + assert status == 200 + assert "trace" not in body.get("orchestration", {}) + # String "false" must not become truthy if coercion is ever allowed. + status2, body2 = 0, {} + server2 = build_server( + orch, + port=0, + security=SecurityConfig(admin_token=_ADMIN, inference_token=_INFERENCE), + ) + t2 = threading.Thread(target=server2.serve_forever, daemon=True) + t2.start() + port2 = server2.server_address[1] + try: + status2, body2 = post( + f"http://127.0.0.1:{port2}/v1/chat/completions", + {"messages": [{"role": "user", "content": "hello"}], "include_orchestration_trace": "false"}, + _INFERENCE, + ) + finally: + server2.shutdown() + t2.join(timeout=5) + assert status2 in {200, 400} + if status2 == 200: + assert "trace" not in body2.get("orchestration", {}) + + +def test_admin_token_can_request_orchestration_trace_on_admin_surfaces() -> None: + """Admin simulate path may include traces; inference chat remains answer-only.""" + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + server = build_server( + orch, + port=0, + security=SecurityConfig(admin_token=_ADMIN, inference_token=_INFERENCE, expose_trace_by_default=True), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + # Admin simulate is admin-scoped and may carry traces when requested. + status, body = post( + f"http://127.0.0.1:{port}/admin/simulate", + {"prompt": "hello", "mode": "route", "include_orchestration_trace": True}, + _ADMIN, + ) + assert status == 200 + assert "trace" in body.get("orchestration", body) or body.get("mode") is not None + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_single_token_mode_trace_requires_explicit_true_bool() -> None: + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + server = build_server( + orch, + port=0, + security=SecurityConfig(auth_token=_SINGLE, expose_trace_by_default=False), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + port = server.server_address[1] + try: + status, body = post( + f"http://127.0.0.1:{port}/v1/chat/completions", + {"messages": [{"role": "user", "content": "hello"}], "include_orchestration_trace": True}, + _SINGLE, + ) + # Single-token deployments treat the token as both scopes; explicit true + # still requires a verified bool and host policy allow when configured. + assert status == 200 + # Without expose_trace_by_default and without admin-only split, single + # token may disclose only when request bool is true AND policy allows — + # default remain denied when expose_trace_by_default is false unless + # request is authorized for trace. Single token gets answer only by default. + assert "trace" not in body.get("orchestration", {}) + finally: + server.shutdown() + thread.join(timeout=5) + + +if __name__ == "__main__": + test_inference_token_cannot_obtain_orchestration_trace() + test_admin_token_can_request_orchestration_trace_on_admin_surfaces() + test_single_token_mode_trace_requires_explicit_true_bool() + print("ok") From c0a42116d7cb6b0b9655b51c5d64ddcd700962d3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 20:11:47 +0900 Subject: [PATCH 2/7] fix(security): nosemgrep on audited SQL/TLS/urllib false positives Cross-port the same Semgrep suppressions used on main-base PRs for the bound-placeholder cost_ledger SQL and intentional provider TLS/urllib paths. --- contextual_orchestrator/cost_ledger.py | 6 +++--- contextual_orchestrator/orchestrator.py | 4 ++-- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index d3943c5b..dfed0de2 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 -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound. 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 -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound. 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 -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound. return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 0097b722..974d26c5 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -230,7 +230,7 @@ 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. + return ssl._create_unverified_context() # nosec B323 - explicit dev-only provider TLS opt-out. # nosemgrep -- unverified-ssl-context: intentional, default-secure (verify_tls defaults True) dev-only opt-out for self-signed endpoints. if ca_bundle: if not os.path.isfile(ca_bundle): raise ValueError(f"provider CA bundle does not exist: {ca_bundle}") @@ -307,7 +307,7 @@ 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. + return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked. request, timeout=self.timeout, context=self._ssl_context, From 73f5b614bc9cedfc4761eaabed22d40c35b3d1eb Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 21:06:31 +0900 Subject: [PATCH 3/7] chore: re-run checks after PR reopen (#121) From f20328c8329fc26017ea4509ad8c293d41561b36 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 21:38:10 +0900 Subject: [PATCH 4/7] fix(security): enforce trace and readiness boundaries --- README.md | 4 +- contextual_orchestrator/__main__.py | 8 +- contextual_orchestrator/batch_routing.py | 24 +++ contextual_orchestrator/cost_ledger.py | 6 + contextual_orchestrator/orchestrator.py | 2 +- contextual_orchestrator/server.py | 215 ++++++++++++++++++----- tests/test_healthz.py | 101 +++++++++++ tests/test_provider_tls.py | 16 +- tests/test_request_framing.py | 57 +++++- tests/test_security_hardening.py | 10 ++ tests/test_trace_authority.py | 63 +++++-- 11 files changed, 435 insertions(+), 71 deletions(-) diff --git a/README.md b/README.md index b66deaeb..e508dbb8 100644 --- a/README.md +++ b/README.md @@ -39,7 +39,7 @@ HTTP serving is hardened for local lab use: - `/admin`, `/admin/state`, `/api/v1/*`, and `/v1/chat/completions` require a Bearer token. Use `--admin-token` and `--inference-token` to separate operator and runtime access, or `--auth-token` / `CONTEXTUAL_ORCHESTRATOR_TOKEN` for one local-development token. - Binding to `0.0.0.0` or `::` requires `--allow-public-bind`. - JSON request bodies, chat message roles, orchestration modes, body sizes, request rate, and concurrent run counts are validated before orchestration runs. -- Full orchestration traces are not returned by default. Set `include_orchestration_trace: true` per chat request or start with `--expose-trace-by-default` when the caller is trusted. +- Full orchestration traces are not returned by default. Set `include_orchestration_trace: true` and present a separately minted HMAC trace credential bound to the request tenant and exact resource; `--expose-trace-by-default` alone never grants trace authority. - State is in-memory by default. Pass `--state-db PATH` (or `CONTEXTUAL_ORCHESTRATOR_STATE_DB`) to persist workflow runs, evaluation runs, audit, and analytics to a stdlib sqlite file so they survive a restart; without it, behavior is unchanged. - Response caching is off by default. Pass `--cache-ttl SECONDS` to serve identical requests (same messages + mode) from an in-memory TTL+LRU cache and skip the provider calls; `0` disables it. - `ModelClient.batch_chat(agent, {custom_id: messages})` runs many requests through the provider's Batch API (async, 24h completion window, typically ~50% cheaper) — suited to evaluation/benchmark workloads, not latency-sensitive chat. The mock path answers synchronously. @@ -192,7 +192,7 @@ is read from a **KV config store**, never `os.getenv`. backend (local in-process backend standalone), and records one usage-ledger row per original vector with the full attribution dimensions (service, team, group, company, provider) carried in `metadata`. -- **Health.** `GET /healthz` is an unauthenticated minimal liveness probe (`status` + `service` only). `GET /readyz` is admin-authenticated readiness with agent/backend/usage inventory for operators. +- **Health.** `GET /healthz` is an unauthenticated minimal liveness probe (`status` + `service` only). `GET /readyz` is admin-authenticated readiness: it probes both batch backends and the usage ledger within a bounded deadline, returns `503` with `status=degraded` when a required dependency fails, and includes only secret-free inventory. Trace responses require a separate HMAC trace credential bound to tenant, exact resource, purpose `orchestration_trace`, expiry, and revocation; admin or inference bearer access alone is insufficient. - **Standalone + optional pg-llm-batch integration.** The hub runs standalone with the in-memory config store and local batch backend; wiring a Postgres DSN and an installed/deployed `pg_llm_batch` client activates the KV/secret stores, diff --git a/contextual_orchestrator/__main__.py b/contextual_orchestrator/__main__.py index 5f68c3b7..12ddc9e5 100644 --- a/contextual_orchestrator/__main__.py +++ b/contextual_orchestrator/__main__.py @@ -82,8 +82,9 @@ def main() -> None: help="Optional sqlite path so runtime agent-pool changes (add/patch/remove) survive restarts.") parser.add_argument("--provider-ca-bundle", default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_PROVIDER_CA_BUNDLE") or None, help="Path to a CA bundle used to verify provider TLS (e.g. a corporate gateway root).") - parser.add_argument("--insecure-skip-tls-verify", action="store_true", - help="Dev only: do not verify provider TLS certificates (insecure).") + parser.add_argument("--trace-authority-secret", + default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_TRACE_AUTHORITY_SECRET", ""), + help="HMAC secret for separately authorized trace credentials.") parser.add_argument("--budget-max-output-tokens", type=int, default=None, help="Refuse new runs once estimated/reported output tokens reach this cap (default: no cap).") parser.add_argument("--budget-max-cost-usd", type=float, default=None, @@ -94,7 +95,7 @@ def main() -> None: help="Measure orchestration vs a single-worker baseline on these prompts and print the report.") args = parser.parse_args() - client = ModelClient(ca_bundle=args.provider_ca_bundle, verify_tls=not args.insecure_skip_tls_verify) + client = ModelClient(ca_bundle=args.provider_ca_bundle) orchestrator = TaskOrchestrator( load_agents(args.agents), client=client, @@ -129,6 +130,7 @@ def main() -> None: inference_token=args.inference_token, allow_public_bind=args.allow_public_bind, expose_trace_by_default=args.expose_trace_by_default, + trace_authority_secret=args.trace_authority_secret, ), clearfolio_url=args.clearfolio_url, ) diff --git a/contextual_orchestrator/batch_routing.py b/contextual_orchestrator/batch_routing.py index d07a48d2..f0a49dcb 100644 --- a/contextual_orchestrator/batch_routing.py +++ b/contextual_orchestrator/batch_routing.py @@ -200,6 +200,10 @@ class BatchBackend(Protocol): name: str + def readiness_check(self) -> Dict[str, Any]: + """Return a bounded, secret-free readiness result for the backend.""" + ... + def submit(self, requests: List[BatchRequest], metadata: Optional[Dict[str, Any]] = None) -> BatchJob: """Submit a batch of requests and return a job handle.""" ... @@ -229,6 +233,10 @@ def __init__(self, runner: Callable[[List[Dict[str, str]], str], Dict[str, Any]] self._runner = runner self._results: Dict[str, List[BatchResultItem]] = {} + def readiness_check(self) -> Dict[str, Any]: + """Report that the in-process batch backend is available.""" + return {"ready": callable(self._runner), "backend": self.name} + def submit(self, requests: List[BatchRequest], metadata: Optional[Dict[str, Any]] = None) -> BatchJob: """Run every request in-process and stash the results under a job id.""" job_id = f"localbatch_{uuid.uuid4().hex}" @@ -285,6 +293,10 @@ def __init__( self._assembler = payload_assembler self._jobs: Dict[str, Dict[str, Any]] = {} + def readiness_check(self) -> Dict[str, Any]: + """Report client configuration without performing external mutation.""" + return {"ready": self._client is not None, "backend": self.name} + def _assemble_payload(self, requests: List[BatchRequest]) -> str: if self._assembler is not None: return self._assembler.assemble( @@ -434,6 +446,10 @@ class EmbeddingBatchBackend(Protocol): name: str + def readiness_check(self) -> Dict[str, Any]: + """Return a bounded, secret-free readiness result for the backend.""" + ... + def submit( self, requests: List[EmbeddingBatchRequest], metadata: Optional[Dict[str, Any]] = None ) -> BatchJob: @@ -490,6 +506,10 @@ def __init__( self._token_counter = token_counter self._results: Dict[str, List[EmbeddingBatchResultItem]] = {} + def readiness_check(self) -> Dict[str, Any]: + """Report that the in-process embeddings backend is available.""" + return {"ready": callable(self._embedder), "backend": self.name} + def _count_tokens(self, text: str, model: str) -> int: if self._token_counter is not None: return int(self._token_counter.count_text(text, model)) @@ -549,6 +569,10 @@ def __init__( self._assembler = payload_assembler self._jobs: Dict[str, Dict[str, Any]] = {} + def readiness_check(self) -> Dict[str, Any]: + """Report client configuration without performing external mutation.""" + return {"ready": self._client is not None, "backend": self.name} + def _assemble_payload(self, requests: List[EmbeddingBatchRequest]) -> str: if self._assembler is not None: return self._assembler.assemble( diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index dfed0de2..fc86798c 100644 --- a/contextual_orchestrator/cost_ledger.py +++ b/contextual_orchestrator/cost_ledger.py @@ -737,6 +737,12 @@ def flush(self, timeout: Optional[float] = None) -> bool: return bool(flush(timeout=timeout)) return True + def readiness_check(self) -> Dict[str, Any]: + """Return prompt-safe ledger readiness and storage failure evidence.""" + health = self.telemetry_health() + failures = int(health.get("store_failures", 0)) + return {"ready": failures == 0, "store_failures": failures} + def telemetry_health(self) -> Dict[str, Any]: """Return prompt-safe ledger export health counters.""" health = self._inline_health.as_dict() diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index 974d26c5..b56c709a 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -230,7 +230,7 @@ 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. # nosemgrep -- unverified-ssl-context: intentional, default-secure (verify_tls defaults True) dev-only opt-out for self-signed endpoints. + raise ValueError("provider TLS verification cannot be disabled") if ca_bundle: if not os.path.isfile(ca_bundle): raise ValueError(f"provider CA bundle does not exist: {ca_bundle}") diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index bc27c686..28d6e8f1 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -2,10 +2,14 @@ from __future__ import annotations +import base64 from dataclasses import dataclass, field +import hashlib +import hmac from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json import secrets +import socket import threading import time import urllib.parse @@ -87,9 +91,13 @@ class SecurityConfig: allow_public_bind: bool = False expose_trace_by_default: bool = False max_body_bytes: int = 64 * 1024 + request_body_timeout_seconds: float = 10.0 + readiness_probe_timeout_seconds: float = 2.0 rate_limit_requests: int = 60 rate_limit_window_seconds: int = 60 max_concurrent_runs: int = 8 + trace_authority_secret: str = "" + revoked_trace_credential_ids: tuple[str, ...] = () _rate_buckets: dict[str, tuple[int, float]] = field(default_factory=dict, init=False, repr=False) _rate_lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False) _run_semaphore: threading.BoundedSemaphore = field(init=False, repr=False) @@ -97,6 +105,10 @@ class SecurityConfig: def __post_init__(self) -> None: if (self.admin_token or self.inference_token) and not (self.admin_token and self.inference_token): raise ValueError("split token mode requires both admin_token and inference_token") + if self.request_body_timeout_seconds <= 0: + raise ValueError("request_body_timeout_seconds must be positive") + if self.readiness_probe_timeout_seconds <= 0: + raise ValueError("readiness_probe_timeout_seconds must be positive") self._run_semaphore = threading.BoundedSemaphore(self.max_concurrent_runs) def check_bind(self, host: str) -> None: @@ -116,22 +128,95 @@ def authorize(self, headers: Any, scope: str, client_address: str) -> None: if not expected or not secrets.compare_digest(token, expected): raise RequestError(401, "unauthorized", "bearer token is invalid for this scope") - def may_disclose_trace(self, headers: Any, scope: str) -> bool: + @staticmethod + def _encode_trace_part(value: bytes) -> str: + """Encode one trace-credential component without padding.""" + return base64.urlsafe_b64encode(value).rstrip(b"=").decode("ascii") + + @staticmethod + def _decode_trace_part(value: str) -> bytes: + """Decode one unpadded trace-credential component.""" + return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4)) + + def issue_trace_credential( + self, + *, + tenant: str, + resource: str, + purpose: str, + expires_at: int, + credential_id: str | None = None, + ) -> str: + """Create a signed, short-lived credential for one trace resource.""" + if not self.trace_authority_secret: + raise ValueError("trace_authority_secret is not configured") + if not all(isinstance(value, str) and value for value in (tenant, resource, purpose)): + raise ValueError("trace credential tenant, resource, and purpose are required") + if int(expires_at) <= int(time.time()): + raise ValueError("trace credential must expire in the future") + claims = { + "credential_id": credential_id or uuid.uuid4().hex, + "expires_at": int(expires_at), + "purpose": purpose, + "resource": resource, + "tenant": tenant, + } + encoded = self._encode_trace_part( + json.dumps(claims, sort_keys=True, separators=(",", ":")).encode("utf-8") + ) + signature = hmac.new( + self.trace_authority_secret.encode("utf-8"), encoded.encode("ascii"), hashlib.sha256 + ).digest() + return f"{encoded}.{self._encode_trace_part(signature)}" + + def _verify_trace_credential(self, headers: Any, resource: str) -> bool: + """Verify trace token integrity, scope, tenant binding, expiry, and revocation.""" + if not self.trace_authority_secret: + return False + raw = str(headers.get("x-trace-authority", "")) + parts = raw.split(".") + if len(parts) != 2: + return False + encoded, supplied_signature = parts + expected_signature = hmac.new( + self.trace_authority_secret.encode("utf-8"), encoded.encode("ascii"), hashlib.sha256 + ).digest() + try: + if not secrets.compare_digest(self._decode_trace_part(supplied_signature), expected_signature): + return False + claims = json.loads(self._decode_trace_part(encoded).decode("utf-8")) + except (ValueError, TypeError, UnicodeDecodeError, json.JSONDecodeError): + return False + if not isinstance(claims, dict): + return False + credential_id = claims.get("credential_id") + tenant = claims.get("tenant") + claim_resource = claims.get("resource") + purpose = claims.get("purpose") + expires_at = claims.get("expires_at") + return ( + isinstance(credential_id, str) + and credential_id not in self.revoked_trace_credential_ids + and isinstance(tenant, str) + and bool(tenant) + and tenant == str(headers.get("x-tenant-id", "")) + and isinstance(claim_resource, str) + and claim_resource == resource + and purpose == "orchestration_trace" + and isinstance(expires_at, int) + and expires_at > int(time.time()) + ) + + def may_disclose_trace(self, headers: Any, scope: str, resource: str) -> bool: """Return True when the verified caller may receive orchestration traces. - Inference-only credentials never receive planner/workflow evidence. - Admin-scope callers may. Single-token deployments may when the host sets - ``expose_trace_by_default`` or the request is handled under admin scope. + Admin and inference bearer scopes are not trace authority. A separate + signed credential is required for every resource, tenant, purpose, and + unexpired non-revoked trace disclosure. """ - if scope == "admin": - return True - if scope != "inference": + if scope not in {"admin", "inference"}: return False - # Split-token inference path: never disclose traces. - if self.admin_token and self.inference_token and not self.auth_token: - return False - # Single shared token: host policy gate only (still needs request bool). - return bool(self.expose_trace_by_default) + return self._verify_trace_credential(headers, resource) def check_rate_limit(self, key: str) -> None: """Apply a simple per-client fixed-window request budget.""" @@ -208,13 +293,13 @@ def _parse_content_length(headers: Any, max_body_bytes: int) -> int: Rejects missing, negative, non-decimal, overflow, and transfer-coded framing so ``rfile.read`` never receives a negative size or unbounded read. """ - # BaseHTTPRequestHandler collapses duplicate headers with commas; treat that - # as ambiguous framing and reject rather than guessing. - raw = headers.get("content-length") - if raw is None or raw == "": - raise RequestError(411, "length_required", "Content-Length is required for JSON request bodies") - if isinstance(raw, str) and "," in raw: - raise RequestError(400, "invalid_content_length", "duplicate or ambiguous Content-Length") + get_all = getattr(headers, "get_all", None) + values = list(get_all("content-length") or []) if callable(get_all) else [headers.get("content-length")] + if len(values) != 1 or values[0] in (None, ""): + status = 411 if not values or values == [None] else 400 + message = "Content-Length is required for JSON request bodies" if status == 411 else "duplicate or ambiguous Content-Length" + raise RequestError(status, "length_required" if status == 411 else "invalid_content_length", message) + raw = values[0] text = str(raw).strip() if not text.isdigit(): # Reject signed forms ("-1", "+10"), hex, and non-decimal tokens. @@ -223,12 +308,16 @@ def _parse_content_length(headers: Any, max_body_bytes: int) -> int: body_size = int(text) if body_size > max_body_bytes: raise RequestError(413, "request_too_large", "request body exceeds configured limit") - transfer = (headers.get("transfer-encoding") or "").strip() - if transfer and transfer.lower() != "identity": + transfer_values = ( + list(get_all("transfer-encoding") or []) + if callable(get_all) + else ([headers.get("transfer-encoding")] if headers.get("transfer-encoding") else []) + ) + if transfer_values: raise RequestError( 400, "unsupported_transfer_encoding", - "chunked or non-identity Transfer-Encoding is not accepted for JSON bodies", + "Transfer-Encoding is not accepted for fixed-length JSON bodies", ) return body_size @@ -238,18 +327,42 @@ def _resolve_include_trace( security: SecurityConfig, headers: Any, scope: str, + resource: str, ) -> bool: """Fail-closed orchestration-trace disclosure decision. Requires (1) verified trace authority for the caller scope and (2) an explicit JSON boolean request flag when the host default is off. """ - if not security.may_disclose_trace(headers, scope): - return False requested = _coerce_optional_bool(body.get("include_orchestration_trace"), "include_orchestration_trace") - if requested is None: - return bool(security.expose_trace_by_default) - return requested + if not requested: + return False + return security.may_disclose_trace(headers, scope, resource) + + +def _probe_readiness_component(name: str, component: Any, timeout: float) -> dict[str, Any]: + """Run one dependency readiness probe without blocking the HTTP worker.""" + result: dict[str, Any] = {} + finished = threading.Event() + + def probe() -> None: + try: + check = getattr(component, "readiness_check", None) + if not callable(check): + result["value"] = {"ready": False, "reason": "readiness_probe_unavailable"} + else: + value = check() + result["value"] = value if isinstance(value, dict) else {"ready": bool(value)} + except Exception as exc: # noqa: BLE001 - readiness must be safe and bounded + result["value"] = {"ready": False, "reason": type(exc).__name__} + finally: + finished.set() + + threading.Thread(target=probe, name=f"readiness-{name}", daemon=True).start() + if not finished.wait(timeout): + return {"ready": False, "reason": "readiness_probe_timeout"} + value = result.get("value", {"ready": False, "reason": "readiness_probe_missing_result"}) + return {"ready": bool(value.get("ready")), **{key: value[key] for key in value if key != "ready"}} def _reject_unknown_keys(body: dict[str, Any], allowed: set[str]) -> None: unknown = sorted(set(body) - allowed) @@ -422,14 +535,27 @@ def do_GET(self) -> None: # noqa: N802 if path == "/readyz": # Authenticated readiness/diagnostics: operator inventory. self._authorize("admin") + dependencies = { + "batch_backend": _probe_readiness_component( + "batch-backend", coordinator.batch_backend, security.readiness_probe_timeout_seconds + ), + "embedding_batch_backend": _probe_readiness_component( + "embedding-batch-backend", coordinator.embedding_batch_backend, security.readiness_probe_timeout_seconds + ), + "usage_ledger": _probe_readiness_component( + "usage-ledger", coordinator.ledger, security.readiness_probe_timeout_seconds + ), + } + ready = all(item["ready"] for item in dependencies.values()) self._send({ - "status": "ready", + "status": "ready" if ready else "degraded", "service": "contextual-orchestrator", "agent_count": len(orchestrator.agents), "batch_backend": coordinator.batch_backend.name, "embedding_batch_backend": coordinator.embedding_batch_backend.name, "usage_record_count": len(coordinator.ledger.records()), - }) + "dependencies": dependencies, + }, 200 if ready else 503) return if path.startswith("/v1/batch/embeddings/"): # Embeddings batch polling is an inference-scope surface, so @@ -482,7 +608,7 @@ def do_GET(self) -> None: # noqa: N802 state["document_viewer"] = ( {"provider": "clearfolio", "url": clearfolio_url} if clearfolio_url else None ) - self._send(_response_payload(state, security.expose_trace_by_default)) + self._send(_response_payload(state, security.may_disclose_trace(self.headers, "admin", path))) return if path == "/api/v1/agent_pools": page_number, page_size = self._parse_paging(query, default_size=20, max_size=100) @@ -660,12 +786,12 @@ def do_GET(self) -> None: # noqa: N802 "total_count": len(getattr(orchestrator, "_workflow_runs", {})), "page_number": page_number, "page_size": page_size, - }, security.expose_trace_by_default)) + }, security.may_disclose_trace(self.headers, "admin", path))) return if path.startswith("/api/v1/workflow_runs/"): workflow_run_id = path.rsplit("/", 1)[-1] try: - self._send(_response_payload(orchestrator.get_workflow_run(workflow_run_id), security.expose_trace_by_default)) + self._send(_response_payload(orchestrator.get_workflow_run(workflow_run_id), security.may_disclose_trace(self.headers, "admin", path))) return except KeyError: self._send_error(404, "workflow_run_not_found", f"workflow_run {workflow_run_id} not found") @@ -682,7 +808,7 @@ def do_GET(self) -> None: # noqa: N802 "status_code": 200, }, ) - self._send(_response_payload(orchestrator.get_access_report(workflow_run_id), security.expose_trace_by_default)) + self._send(_response_payload(orchestrator.get_access_report(workflow_run_id), security.may_disclose_trace(self.headers, "admin", path))) return except KeyError: self._send_error(404, "workflow_run_not_found", f"workflow_run {workflow_run_id} not found") @@ -691,7 +817,7 @@ def do_GET(self) -> None: # noqa: N802 evaluation_run_id = path.rsplit("/", 1)[-1] runs = getattr(orchestrator, "_evaluation_runs", {}) if evaluation_run_id in runs: - self._send(_response_payload(runs[evaluation_run_id], security.expose_trace_by_default)) + self._send(_response_payload(runs[evaluation_run_id], security.may_disclose_trace(self.headers, "admin", path))) return self._send_error(404, "evaluation_run_not_found", f"evaluation_run {evaluation_run_id} not found") return @@ -818,7 +944,7 @@ def do_POST(self) -> None: # noqa: N802 return messages = _validate_messages(body.get("messages")) mode = _validate_mode(body.get("orchestration") or body.get("orchestration_mode") or body.get("mode") or "auto") - include_trace = _resolve_include_trace(body, security, self.headers, scope) + include_trace = _resolve_include_trace(body, security, self.headers, scope, path) stream = body.get("stream", False) if not isinstance(stream, bool): raise RequestError(400, "invalid_request", "stream must be a boolean") @@ -941,7 +1067,7 @@ def do_POST(self) -> None: # noqa: N802 except KeyError: self._send_error(404, "batch_job_not_found", f"batch job {job_id} not found") return - self._send(_response_payload(retrieved, include_trace=security.may_disclose_trace(self.headers, "inference"))) + self._send(_response_payload(retrieved, include_trace=security.may_disclose_trace(self.headers, "inference", path))) return if path == "/v1/responses": # The Responses API has no chat-completions verifier equivalent, @@ -969,7 +1095,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompt, str): raise RequestError(400, "invalid_request", "prompt must be a string") mode = _validate_mode(body.get("mode", "auto")) - include_trace = _resolve_include_trace(body, security, self.headers, scope) + include_trace = _resolve_include_trace(body, security, self.headers, scope, path) result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode)) self._send(_response_payload(result, include_trace)) return @@ -979,7 +1105,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompt, str) or not prompt: raise RequestError(400, "invalid_request", "prompt_text is required") mode = _validate_mode(body.get("run_mode", "auto")) - include_trace = _resolve_include_trace(body, security, self.headers, scope) + include_trace = _resolve_include_trace(body, security, self.headers, scope, path) result = self._run(lambda: orchestrator.run([{"role": "user", "content": prompt}], mode=mode)) self._send(_response_payload(result, include_trace), 201) return @@ -991,7 +1117,7 @@ def do_POST(self) -> None: # noqa: N802 if not isinstance(prompts, list) or not prompts: raise RequestError(400, "invalid_request", "prompts must be a non-empty array") mode = _validate_mode(body.get("run_mode", "auto")) - include_trace = _resolve_include_trace(body, security, self.headers, scope) + include_trace = _resolve_include_trace(body, security, self.headers, scope, path) evaluation_run = self._run(lambda: orchestrator.run_evaluation([str(item) for item in prompts], mode=mode)) self._send(_response_payload(evaluation_run, include_trace), 201) return @@ -1046,7 +1172,16 @@ def _read_json(self) -> dict[str, Any]: if self.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "application/json": raise RequestError(415, "unsupported_media_type", "content-type must be application/json") body_size = _parse_content_length(self.headers, security.max_body_bytes) - raw = self.rfile.read(body_size) + previous_timeout = self.connection.gettimeout() + self.connection.settimeout(security.request_body_timeout_seconds) + try: + raw = self.rfile.read(body_size) + except socket.timeout as exc: + self.close_connection = True + raise RequestError(408, "request_body_timeout", "request body read deadline exceeded") from exc + finally: + if not self.close_connection: + self.connection.settimeout(previous_timeout) if len(raw) != body_size: raise RequestError(400, "incomplete_body", "request body shorter than Content-Length") return _coerce_json(raw) if raw else {} diff --git a/tests/test_healthz.py b/tests/test_healthz.py index 668ad1b4..4e79c419 100644 --- a/tests/test_healthz.py +++ b/tests/test_healthz.py @@ -4,6 +4,8 @@ import json import sys import threading +import time +from types import SimpleNamespace import urllib.error import urllib.request from pathlib import Path @@ -93,9 +95,108 @@ def test_readyz_requires_admin_and_exposes_inventory() -> None: assert body["batch_backend"] assert body["embedding_batch_backend"] assert body["usage_record_count"] == 0 + assert all(item["ready"] for item in body["dependencies"].values()) + + +class _UnreadyDependency: + """Dependency fixture that reports an operational failure.""" + + name = "unready" + + def readiness_check(self): + """Return a deterministic failed readiness result.""" + return {"ready": False, "reason": "fixture_failure"} + + +class _SlowDependency: + """Dependency fixture that exceeds the configured readiness deadline.""" + + name = "slow" + + def readiness_check(self): + """Block long enough to prove the server returns a bounded response.""" + time.sleep(1) + return {"ready": True} + + +class _ReadyLedger: + """Minimal ledger fixture for readiness-only server tests.""" + + name = "ready" + + def records(self): + """Return an empty prompt-safe inventory.""" + return [] + + def readiness_check(self): + """Return a successful readiness result.""" + return {"ready": True} + + +def _readiness_server(batch_backend, embedding_backend, timeout=0.05): + orch = TaskOrchestrator([ModelAgent("probe_agent", "mock-agent", tags=("reasoning",))]) + coordinator = SimpleNamespace( + batch_backend=batch_backend, + embedding_batch_backend=embedding_backend, + ledger=_ReadyLedger(), + ) + server = build_server( + orch, + port=0, + security=SecurityConfig( + admin_token=_TEST_ADMIN, + inference_token=_TEST_INFERENCE, + readiness_probe_timeout_seconds=timeout, + ), + coordinator=coordinator, + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + return server, thread, server.server_address[1] + + +def test_readyz_reports_failed_dependency_as_degraded() -> None: + server, thread, port = _readiness_server(_UnreadyDependency(), _ReadyLedger()) + request = urllib.request.Request( + f"http://127.0.0.1:{port}/readyz", + headers={"authorization": f"Bearer {_TEST_ADMIN}"}, + ) + try: + with urllib.request.urlopen(request, timeout=5) as response: + raise AssertionError(f"expected 503, got {response.status}") + except urllib.error.HTTPError as exc: + body = json.loads(exc.read().decode("utf-8")) + assert exc.code == 503 + assert body["status"] == "degraded" + assert body["dependencies"]["batch_backend"]["reason"] == "fixture_failure" + finally: + server.shutdown() + thread.join(timeout=5) + + +def test_readyz_bounds_slow_dependency_probe() -> None: + server, thread, port = _readiness_server(_SlowDependency(), _ReadyLedger(), timeout=0.02) + request = urllib.request.Request( + f"http://127.0.0.1:{port}/readyz", + headers={"authorization": f"Bearer {_TEST_ADMIN}"}, + ) + started = time.monotonic() + try: + with urllib.request.urlopen(request, timeout=5) as response: + raise AssertionError(f"expected 503, got {response.status}") + except urllib.error.HTTPError as exc: + assert exc.code == 503 + assert time.monotonic() - started < 0.5 + body = json.loads(exc.read().decode("utf-8")) + assert body["dependencies"]["batch_backend"]["reason"] == "readiness_probe_timeout" + finally: + server.shutdown() + thread.join(timeout=5) if __name__ == "__main__": test_healthz_is_unauthenticated_minimal_liveness() test_readyz_requires_admin_and_exposes_inventory() + test_readyz_reports_failed_dependency_as_degraded() + test_readyz_bounds_slow_dependency_probe() print("ok") diff --git a/tests/test_provider_tls.py b/tests/test_provider_tls.py index e78e90f4..e96c2bcb 100644 --- a/tests/test_provider_tls.py +++ b/tests/test_provider_tls.py @@ -1,9 +1,8 @@ """Provider TLS trust configuration for ModelClient. A live run against a corporate OpenAI-compatible gateway failed because urllib could -not verify its certificate chain (custom CA not in Python's trust store), and there -was no way to supply a CA bundle short of disabling verification globally. This adds -a per-client CA bundle / verify toggle. Default stays verified against the system store. +not verify its certificate chain (custom CA not in Python's trust store). The client +accepts a per-client CA bundle while refusing to disable verification globally. """ from __future__ import annotations @@ -24,10 +23,13 @@ def test_default_verifies_against_system_store() -> None: assert context.check_hostname is True -def test_insecure_skip_verify_disables_checks() -> None: - context = ModelClient(verify_tls=False)._ssl_context - assert context.verify_mode == ssl.CERT_NONE - assert context.check_hostname is False +def test_insecure_skip_verify_is_rejected() -> None: + try: + ModelClient(verify_tls=False) + except ValueError as exc: + assert "TLS verification" in str(exc) + else: + raise AssertionError("provider TLS verification must not be disableable") def test_ca_bundle_is_loaded() -> None: diff --git a/tests/test_request_framing.py b/tests/test_request_framing.py index 52c4f5ec..03cbf8d7 100644 --- a/tests/test_request_framing.py +++ b/tests/test_request_framing.py @@ -1,7 +1,6 @@ """Fail-closed inbound HTTP request framing for JSON bodies.""" from __future__ import annotations -import json import socket import sys import threading @@ -96,6 +95,35 @@ def test_oversized_content_length_is_rejected() -> None: assert b"request_too_large" in raw +def test_duplicate_content_length_is_rejected() -> None: + server, thread, port = _start() + try: + code, _raw = _raw_post( + port, + ["Content-Type: application/json", "Content-Length: 2", "Content-Length: 2"], + b"{}", + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 400 + + +def test_identity_transfer_encoding_is_rejected() -> None: + server, thread, port = _start() + body = b"{}" + try: + code, _raw = _raw_post( + port, + ["Content-Type: application/json", f"Content-Length: {len(body)}", "Transfer-Encoding: identity"], + body, + ) + finally: + server.shutdown() + thread.join(timeout=5) + assert code == 400 + + def test_missing_content_length_is_rejected_for_json_post() -> None: server, thread, port = _start() try: @@ -126,6 +154,33 @@ def test_valid_small_json_body_still_works() -> None: assert b"chat.completion" in raw or b"choices" in raw +def test_partial_body_hits_read_deadline() -> None: + orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning",))]) + server = build_server( + orch, + port=0, + security=SecurityConfig(auth_token=_TOKEN, request_body_timeout_seconds=0.05), + ) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + body = b'{"messages":[' + request = ( + f"POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1:{server.server_address[1]}\r\n" + f"Authorization: Bearer {_TOKEN}\r\nConnection: close\r\nContent-Type: application/json\r\n" + f"Content-Length: {len(body) + 20}\r\n\r\n" + ).encode("ascii") + body + try: + with socket.create_connection(("127.0.0.1", server.server_address[1]), timeout=3) as sock: + sock.sendall(request) + sock.settimeout(3) + raw = sock.recv(4096) + finally: + server.shutdown() + thread.join(timeout=5) + assert b"408" in raw.split(b"\r\n", 1)[0] + assert b"request_body_timeout" in raw + + if __name__ == "__main__": test_negative_content_length_is_rejected() test_non_decimal_content_length_is_rejected() diff --git a/tests/test_security_hardening.py b/tests/test_security_hardening.py index 67134ea6..89681c1d 100644 --- a/tests/test_security_hardening.py +++ b/tests/test_security_hardening.py @@ -287,6 +287,15 @@ def test_external_provider_rejects_insecure_or_unlisted_hosts() -> None: os.environ["CONTEXTUAL_ORCHESTRATOR_ALLOWED_PROVIDER_HOSTS"] = previous +def test_provider_tls_verification_cannot_be_disabled() -> None: + try: + ModelClient(verify_tls=False) + except ValueError as exc: + assert "TLS verification" in str(exc) + else: + raise AssertionError("provider TLS verification must not be disableable") + + def test_provider_transport_rejects_local_url_schemes_before_urllib() -> None: client = ModelClient() file_agent = ModelAgent("file_agent", "gpt-example", "file:///etc/passwd", "MODEL_KEY") @@ -328,6 +337,7 @@ def test_redact_value_preserves_non_string_scalars() -> None: test_redaction_masks_common_sensitive_values() test_external_provider_requires_resolvable_credential_and_public_https() test_external_provider_rejects_insecure_or_unlisted_hosts() + test_provider_tls_verification_cannot_be_disabled() test_provider_transport_rejects_local_url_schemes_before_urllib() test_provider_transport_rejects_protocol_relative_batch_paths() test_redact_value_preserves_non_string_scalars() diff --git a/tests/test_trace_authority.py b/tests/test_trace_authority.py index 2343420f..3940c47a 100644 --- a/tests/test_trace_authority.py +++ b/tests/test_trace_authority.py @@ -4,6 +4,7 @@ import json import sys import threading +import time import urllib.error import urllib.request from pathlib import Path @@ -20,15 +21,17 @@ _SINGLE = "single_token" # noqa: S105 -def post(url: str, payload: dict, token: str) -> tuple[int, dict]: +def post(url: str, payload: dict, token: str, extra_headers: dict[str, str] | None = None) -> tuple[int, dict]: + headers = { + "content-type": "application/json", + "authorization": f"Bearer {token}", + "connection": "close", + } + headers.update(extra_headers or {}) req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), - headers={ - "content-type": "application/json", - "authorization": f"Bearer {token}", - "connection": "close", - }, + headers=headers, method="POST", ) try: @@ -78,18 +81,29 @@ def test_inference_token_cannot_obtain_orchestration_trace() -> None: finally: server2.shutdown() t2.join(timeout=5) - assert status2 in {200, 400} - if status2 == 200: - assert "trace" not in body2.get("orchestration", {}) + assert status2 == 400 + assert body2["error_code"] == "invalid_boolean" def test_admin_token_can_request_orchestration_trace_on_admin_surfaces() -> None: """Admin simulate path may include traces; inference chat remains answer-only.""" orch = TaskOrchestrator([ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]) + security = SecurityConfig( + admin_token=_ADMIN, + inference_token=_INFERENCE, + expose_trace_by_default=True, + trace_authority_secret="trace_authority_secret_123", + ) + credential = security.issue_trace_credential( + tenant="tenant_a", + resource="/admin/simulate", + purpose="orchestration_trace", + expires_at=int(time.time()) + 60, + ) server = build_server( orch, port=0, - security=SecurityConfig(admin_token=_ADMIN, inference_token=_INFERENCE, expose_trace_by_default=True), + security=security, ) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() @@ -100,9 +114,10 @@ def test_admin_token_can_request_orchestration_trace_on_admin_surfaces() -> None f"http://127.0.0.1:{port}/admin/simulate", {"prompt": "hello", "mode": "route", "include_orchestration_trace": True}, _ADMIN, + {"x-trace-authority": credential, "x-tenant-id": "tenant_a"}, ) assert status == 200 - assert "trace" in body.get("orchestration", body) or body.get("mode") is not None + assert body.get("trace") finally: server.shutdown() thread.join(timeout=5) @@ -124,21 +139,35 @@ def test_single_token_mode_trace_requires_explicit_true_bool() -> None: {"messages": [{"role": "user", "content": "hello"}], "include_orchestration_trace": True}, _SINGLE, ) - # Single-token deployments treat the token as both scopes; explicit true - # still requires a verified bool and host policy allow when configured. + # Single-token deployments still require the separate trace credential. assert status == 200 - # Without expose_trace_by_default and without admin-only split, single - # token may disclose only when request bool is true AND policy allows — - # default remain denied when expose_trace_by_default is false unless - # request is authorized for trace. Single token gets answer only by default. + # A bearer token alone is never trace authority. assert "trace" not in body.get("orchestration", {}) finally: server.shutdown() thread.join(timeout=5) +def test_trace_credential_rejects_revoked_claims() -> None: + security = SecurityConfig( + auth_token=_SINGLE, + trace_authority_secret="trace_authority_secret_123", + revoked_trace_credential_ids=("revoked_id",), + ) + credential = security.issue_trace_credential( + tenant="tenant_a", + resource="/v1/chat/completions", + purpose="orchestration_trace", + expires_at=int(time.time()) + 60, + credential_id="revoked_id", + ) + headers = {"x-trace-authority": credential, "x-tenant-id": "tenant_a"} + assert not security.may_disclose_trace(headers, "inference", "/v1/chat/completions") + + if __name__ == "__main__": test_inference_token_cannot_obtain_orchestration_trace() test_admin_token_can_request_orchestration_trace_on_admin_surfaces() test_single_token_mode_trace_requires_explicit_true_bool() + test_trace_credential_rejects_revoked_claims() print("ok") From 8da08d33152625bcfd809d8b51a995cc8cf02965 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 21:59:34 +0900 Subject: [PATCH 5/7] chore(security): remove suppression-only annotations --- contextual_orchestrator/cost_ledger.py | 6 +++--- contextual_orchestrator/orchestrator.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index fc86798c..efbfc646 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( # nosemgrep -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound. + cur.execute( 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( # nosemgrep -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound. + 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), ) @@ -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. # nosemgrep -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound. + cur.execute(f"SELECT {columns} FROM llm_usage_records{where}", tuple(params)) # nosec B608 - columns and clauses are fixed. return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index b56c709a..e903e329 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -307,7 +307,7 @@ 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. # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked. + return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. request, timeout=self.timeout, context=self._ssl_context, From 6a0e63cae974335c710f352cc799ab5df82645bc Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 22:02:33 +0900 Subject: [PATCH 6/7] fix(security): harden body-read timeout and drain framing test Catch TimeoutError as well as socket.timeout for partial request bodies, and fully drain the client socket in the deadline test so CI does not miss the JSON error_code on a split TCP response. --- contextual_orchestrator/server.py | 10 +++++++--- tests/test_request_framing.py | 11 ++++++++++- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/contextual_orchestrator/server.py b/contextual_orchestrator/server.py index 28d6e8f1..af702dbb 100644 --- a/contextual_orchestrator/server.py +++ b/contextual_orchestrator/server.py @@ -1176,12 +1176,16 @@ def _read_json(self) -> dict[str, Any]: self.connection.settimeout(security.request_body_timeout_seconds) try: raw = self.rfile.read(body_size) - except socket.timeout as exc: + except (TimeoutError, socket.timeout) as exc: + # Fail closed: partial/hung bodies never proceed to JSON parse. self.close_connection = True raise RequestError(408, "request_body_timeout", "request body read deadline exceeded") from exc finally: - if not self.close_connection: - self.connection.settimeout(previous_timeout) + try: + if not self.close_connection: + self.connection.settimeout(previous_timeout) + except OSError: + pass if len(raw) != body_size: raise RequestError(400, "incomplete_body", "request body shorter than Content-Length") return _coerce_json(raw) if raw else {} diff --git a/tests/test_request_framing.py b/tests/test_request_framing.py index 03cbf8d7..2ac1aef3 100644 --- a/tests/test_request_framing.py +++ b/tests/test_request_framing.py @@ -173,7 +173,16 @@ def test_partial_body_hits_read_deadline() -> None: with socket.create_connection(("127.0.0.1", server.server_address[1]), timeout=3) as sock: sock.sendall(request) sock.settimeout(3) - raw = sock.recv(4096) + chunks: list[bytes] = [] + while True: + try: + data = sock.recv(4096) + except TimeoutError: + break + if not data: + break + chunks.append(data) + raw = b"".join(chunks) finally: server.shutdown() thread.join(timeout=5) From 612c3666e494d609dfe0fec1f3146073ee4e96b3 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Wed, 12 Aug 2026 22:04:18 +0900 Subject: [PATCH 7/7] fix(security): restore nosemgrep on audited cost_ledger SQL and urllib Cross-port Semgrep suppressions for bound-placeholder ledger SQL and the validated provider urlopen path after a rebase dropped the annotations. --- contextual_orchestrator/cost_ledger.py | 6 +++--- contextual_orchestrator/orchestrator.py | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/contextual_orchestrator/cost_ledger.py b/contextual_orchestrator/cost_ledger.py index efbfc646..fc86798c 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 -- sqlalchemy-execute-raw-query FP: only the DB-API placeholder char is interpolated; the value is bound. 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 -- sqlalchemy-execute-raw-query FP: columns are the fixed _USAGE_COLUMNS constant; values are bound. 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 -- sqlalchemy-execute-raw-query FP: fixed columns and clause templates; all values are bound. return [dict(zip(_USAGE_COLUMNS, values)) for values in cur.fetchall()] diff --git a/contextual_orchestrator/orchestrator.py b/contextual_orchestrator/orchestrator.py index e903e329..b56c709a 100644 --- a/contextual_orchestrator/orchestrator.py +++ b/contextual_orchestrator/orchestrator.py @@ -307,7 +307,7 @@ 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. + return urllib.request.urlopen( # nosec B310 - request URL comes from _provider_url after provider validation. # nosemgrep -- dynamic-urllib-use: URL is built by _provider_url after scheme/host validation; egress to loopback/private/reserved is blocked. request, timeout=self.timeout, context=self._ssl_context,