feat(api): opt-in CORS allowlist for browser clients

contextual_orchestrator/__main__.py

@@ -74,6 +74,15 @@ def main() -> None:
     parser.add_argument("--admin-token", default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_ADMIN_TOKEN", ""))
     parser.add_argument("--inference-token", default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_INFERENCE_TOKEN", ""))
     parser.add_argument("--allow-public-bind", action="store_true")
+    parser.add_argument(
+        "--cors-allow-origin",
+        action="append",
+        default=[],
+        help=(
+            "Exact browser Origin allowed for CORS (repeatable). Empty disables CORS. "
+            "Use a single '*' only when deliberately opening all origins."
+        ),
+    )
     parser.add_argument("--insecure-disable-auth", action="store_true", help="Deprecated; API auth is always required.")
     parser.add_argument("--expose-trace-by-default", action="store_true")
     parser.add_argument("--clearfolio-url", default=os.environ.get("CONTEXTUAL_ORCHESTRATOR_CLEARFOLIO_URL") or None,
@@ -129,6 +138,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,
+                cors_allow_origins=tuple(args.cors_allow_origin or ()),
             ),
             clearfolio_url=args.clearfolio_url,
         )

contextual_orchestrator/cost_ledger.py

@@ -583,12 +583,12 @@ 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,),
             )
             if cur.fetchone() is None:
-                cur.execute(
+                cur.execute(  # nosemgrep -- sqlalchemy-execute-raw-query FP: only DB-API placeholder chars are interpolated; values are bound.
                     "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),
@@ -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()]
 
 

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,

contextual_orchestrator/server.py

@@ -90,15 +90,38 @@
     rate_limit_requests: int = 60
     rate_limit_window_seconds: int = 60
     max_concurrent_runs: int = 8
+    # Explicit browser origins allowed to call the API (exact match; empty = CORS off).
+    cors_allow_origins: 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)
 
     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")
+        cleaned: list[str] = []
+        for origin in self.cors_allow_origins:
+            value = str(origin).strip()
+            if value:
+                cleaned.append(value)
+        self.cors_allow_origins = tuple(cleaned)
         self._run_semaphore = threading.BoundedSemaphore(self.max_concurrent_runs)
 
+    def cors_origin_allowed(self, origin: str) -> str | None:
+        """Return the origin echo value when it is on the allowlist, else None.
+
+        Only exact string matches are accepted (including a single ``*`` entry when
+        operators deliberately open browser cross-origin access). Empty allowlist
+        disables CORS headers entirely.
+        """
+        if not self.cors_allow_origins or not origin:
+            return None
+        if "*" in self.cors_allow_origins:
+            return "*"
+        if origin in self.cors_allow_origins:
+            return origin
+        return None
+
     def check_bind(self, host: str) -> None:
         """Require explicit opt-in before binding the API to public interfaces."""
         if host in {"0.0.0.0", "::", ""} and not self.allow_public_bind:  # nosec B104 - comparison rejects public bind unless explicitly opted in.
@@ -151,6 +174,8 @@
             "rate_limit_requests": self.rate_limit_requests,
             "rate_limit_window_seconds": self.rate_limit_window_seconds,
             "max_concurrent_runs": self.max_concurrent_runs,
+            "cors_allow_origin_count": len(self.cors_allow_origins),
+            "cors_allow_any_origin": "*" in self.cors_allow_origins,
         }
 
 
@@ -327,6 +352,31 @@
         clearfolio_url = clearfolio_url.rstrip("/")
 
     class Handler(BaseHTTPRequestHandler):
+        def do_OPTIONS(self) -> None:  # noqa: N802
+            """CORS preflight for browser OpenAI-compatible clients."""
+            try:
+                origin = (self.headers.get("origin") or "").strip()
+                allowed = security.cors_origin_allowed(origin)
+                if not allowed:
+                    self._send_error(403, "cors_origin_not_allowed", "origin is not on the CORS allowlist")
+                    return
+                # Preflight is unauthenticated; actual verbs still require bearer tokens.
+                self._cors_allow_origin = allowed
+                self.send_response(204)
+                self._send_security_headers()
+                self.send_header("access-control-allow-methods", "GET, POST, PATCH, DELETE, OPTIONS")
+                self.send_header(
+                    "access-control-allow-headers",
+                    "authorization, content-type, x-request-id",
+                )
+                self.send_header("access-control-max-age", "600")
+                self.send_header("content-length", "0")
+                self.end_headers()
+            except RequestError as exc:
+                self._send_error(exc.status, exc.code, exc.message, exc.detail)
+            except Exception:
+                self._send_error(500, "internal_error", "internal server error")
+
         def do_GET(self) -> None:  # noqa: N802
             parsed = urllib.parse.urlparse(self.path)
             path = parsed.path
@@ -1054,6 +1104,18 @@
             self.send_header("referrer-policy", "no-referrer")
             self.send_header("cache-control", "no-store")
             self.send_header("x-frame-options", "DENY")
+            # CORS: only when the request Origin is on the operator allowlist.
+            allowed = getattr(self, "_cors_allow_origin", None)
+            if not allowed:
+                origin = (self.headers.get("origin") or "").strip()
+                allowed = security.cors_origin_allowed(origin)
+            if allowed:
+                self.send_header("access-control-allow-origin", allowed)
+                self.send_header("vary", "Origin")
+                # Browser admin may use cookie session on same-site; credentials
+                # are allowed only for exact-origin echoes (never for *).
+                if allowed != "*":
+                    self.send_header("access-control-allow-credentials", "true")
 
     return ThreadingHTTPServer((host, port), Handler)
 

docs/architecture.md

@@ -38,6 +38,14 @@ This repository implements the interface and control plane, not the trained coor
 - `ModelClient`: OpenAI-compatible HTTP client, with `mock://` for local checks.
 - `contextual_orchestrator.server`: small `/v1/chat/completions` HTTP server.
 
+## CORS for browser clients
+
+Cross-origin browser access is **off by default**. Operators opt in with exact
+`cors_allow_origins` entries (CLI `--cors-allow-origin`, repeatable). Allowed
+origins receive `Access-Control-Allow-Origin` (+ credentials for non-`*`) and
+`OPTIONS` preflight replies. See Fetch CORS (WHATWG) and browser same-origin
+policy; wildcards are never implied.
+
 The deliberate simplification is the policy. The paper systems learn routing and topology from rewards; this lab uses deterministic keyword scoring so the repo runs without training data, GPUs, or vendor credentials.
 
 Add learned routing only when there is an evaluation set and logs proving the heuristic policy is the bottleneck.

tests/test_cors_allowlist.py

@@ -0,0 +1,140 @@
+"""CORS allowlist for browser OpenAI-compatible clients."""
+
+from __future__ import annotations
+
+import json
+import threading
+import urllib.error
+import urllib.request
+from pathlib import Path
+import sys
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from contextual_orchestrator import ModelAgent, TaskOrchestrator  # noqa: E402
+from contextual_orchestrator.server import SecurityConfig, build_server  # noqa: E402
+
+_TEST_AUTH_TOKEN = "secret_token"  # noqa: S105
+_ALLOWED_ORIGIN = "https://app.example.com"
+
+
+def build() -> TaskOrchestrator:
+    return TaskOrchestrator(
+        [ModelAgent("general_agent", "mock-generalist", tags=("reasoning", "writing"))]
+    )
+
+
+def test_cors_origin_allowed_matches_exact_and_star() -> None:
+    security = SecurityConfig(auth_token=_TEST_AUTH_TOKEN, cors_allow_origins=(_ALLOWED_ORIGIN,))
+    assert security.cors_origin_allowed(_ALLOWED_ORIGIN) == _ALLOWED_ORIGIN
+    assert security.cors_origin_allowed("https://evil.example") is None
+    assert security.cors_origin_allowed("") is None
+    star = SecurityConfig(auth_token=_TEST_AUTH_TOKEN, cors_allow_origins=("*",))
+    assert star.cors_origin_allowed("https://any.example") == "*"
+    off = SecurityConfig(auth_token=_TEST_AUTH_TOKEN)
+    assert off.cors_origin_allowed(_ALLOWED_ORIGIN) is None
+
+
+def test_preflight_and_authenticated_response_echo_allowlisted_origin() -> None:
+    server = build_server(
+        build(),
+        port=0,
+        security=SecurityConfig(
+            auth_token=_TEST_AUTH_TOKEN,
+            cors_allow_origins=(_ALLOWED_ORIGIN,),
+        ),
+    )
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    port = server.server_address[1]
+    try:
+        preflight = urllib.request.Request(
+            f"http://127.0.0.1:{port}/v1/chat/completions",
+            method="OPTIONS",
+            headers={
+                "origin": _ALLOWED_ORIGIN,
+                "access-control-request-method": "POST",
+                "connection": "close",
+            },
+        )
+        with urllib.request.urlopen(preflight, timeout=5) as response:
+            assert response.status == 204
+            allow_origin = response.headers.get("Access-Control-Allow-Origin") or response.headers.get(
+                "access-control-allow-origin"
+            )
+            allow_methods = response.headers.get("Access-Control-Allow-Methods") or response.headers.get(
+                "access-control-allow-methods"
+            )
+            credentials = response.headers.get("Access-Control-Allow-Credentials") or response.headers.get(
+                "access-control-allow-credentials"
+            )
+            assert allow_origin == _ALLOWED_ORIGIN
+            assert "POST" in (allow_methods or "")
+            assert credentials == "true"
+
+        denied = urllib.request.Request(
+            f"http://127.0.0.1:{port}/v1/chat/completions",
+            method="OPTIONS",
+            headers={"origin": "https://evil.example", "connection": "close"},
+        )
+        try:
+            urllib.request.urlopen(denied, timeout=5)
+            raise AssertionError("expected 403 for non-allowlisted origin")
+        except urllib.error.HTTPError as exc:
+            assert exc.code == 403
+
+        chat = urllib.request.Request(
+            f"http://127.0.0.1:{port}/v1/chat/completions",
+            data=json.dumps({"messages": [{"role": "user", "content": "hi"}]}).encode("utf-8"),
+            headers={
+                "content-type": "application/json",
+                "authorization": f"Bearer {_TEST_AUTH_TOKEN}",
+                "origin": _ALLOWED_ORIGIN,
+                "connection": "close",
+            },
+            method="POST",
+        )
+        with urllib.request.urlopen(chat, timeout=5) as response:
+            assert response.status == 200
+            allow_origin = response.headers.get("Access-Control-Allow-Origin") or response.headers.get(
+                "access-control-allow-origin"
+            )
+            assert allow_origin == _ALLOWED_ORIGIN
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+
+def test_cors_disabled_by_default_omits_allow_origin_header() -> None:
+    server = build_server(build(), port=0, security=SecurityConfig(auth_token=_TEST_AUTH_TOKEN))
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    port = server.server_address[1]
+    try:
+        chat = urllib.request.Request(
+            f"http://127.0.0.1:{port}/v1/chat/completions",
+            data=json.dumps({"messages": [{"role": "user", "content": "hi"}]}).encode("utf-8"),
+            headers={
+                "content-type": "application/json",
+                "authorization": f"Bearer {_TEST_AUTH_TOKEN}",
+                "origin": _ALLOWED_ORIGIN,
+                "connection": "close",
+            },
+            method="POST",
+        )
+        with urllib.request.urlopen(chat, timeout=5) as response:
+            assert response.status == 200
+            allow_origin = response.headers.get("Access-Control-Allow-Origin") or response.headers.get(
+                "access-control-allow-origin"
+            )
+            assert not allow_origin
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+
+if __name__ == "__main__":
+    test_cors_origin_allowed_matches_exact_and_star()
+    test_preflight_and_authenticated_response_echo_allowlisted_origin()
+    test_cors_disabled_by_default_omits_allow_origin_header()
+    print("ok")