diff --git a/README.md b/README.md index 2cabdb1c9..3ac2656dc 100644 --- a/README.md +++ b/README.md @@ -319,6 +319,15 @@ export STRIX_LLM="chatgpt/gpt-5.4" # chatgpt/ runs on the subscription strix auth status # show the active sign-in, or logout to forget it ``` +#### Sign in with OrcaRouter + +Use an [OrcaRouter](https://www.orcarouter.ai) API key, or sign in with your OrcaRouter account instead of copying one. OrcaRouter's [Cyber Inference tier](https://www.orcarouter.ai/cyber) adds scoped, authorized models for security testing: + +```bash +strix auth login orcarouter # browser sign-in (add --manual to paste a code) +export STRIX_LLM="orcarouter/orcarouter/auto" # or: export ORCAROUTER_API_KEY="sk-orca-..." +``` + #### Use the managed platform: `strix cloud` Run scans on [app.strix.ai](https://app.strix.ai) from the terminal, without Docker or an LLM key: diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx index 29be4c871..c577e7444 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -15,6 +15,13 @@ Configure Strix using environment variables or a config file. API key for your LLM provider. Not required for local models or cloud provider auth (Vertex AI, AWS Bedrock). + + API key for `orcarouter/` models. Takes precedence over a key stored by + `strix auth login orcarouter`, which in turn takes precedence over `LLM_API_KEY`. + `ORCA_BASE_URL`, `ORCA_AUTH_BASE_URL`, and `ORCA_API_BASE_URL` point OrcaRouter + at a self-hosted gateway. See [OrcaRouter](/llm-providers/orcarouter). + + Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`. diff --git a/docs/docs.json b/docs/docs.json index f9cbf26c6..303f1c6c4 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -37,6 +37,7 @@ "llm-providers/anthropic", "llm-providers/vercel-ai-gateway", "llm-providers/openrouter", + "llm-providers/orcarouter", "llm-providers/vertex", "llm-providers/bedrock", "llm-providers/azure", diff --git a/docs/llm-providers/orcarouter.mdx b/docs/llm-providers/orcarouter.mdx new file mode 100644 index 000000000..61e3e6b58 --- /dev/null +++ b/docs/llm-providers/orcarouter.mdx @@ -0,0 +1,62 @@ +--- +title: "OrcaRouter Cyber Inference" +description: "Run Strix on OrcaRouter Cyber Inference — scoped, authorized models for security testing" +--- + +[OrcaRouter](https://www.orcarouter.ai) is an OpenAI-compatible gateway to models from OpenAI, Anthropic, Google, DeepSeek, and others through a single API key. Its [Cyber Inference tier](https://www.orcarouter.ai/cyber) is built for authorized security work: models a security team can switch on for scoped, authorized engagements, so an agent like Strix isn't stopped by a general-purpose model's blanket refusals in the middle of a sanctioned test. + +## Setup + +Use the `orcarouter/` format and pick one of two ways to authenticate. + +### API key + +```bash +export STRIX_LLM="orcarouter/orcarouter/auto" +export ORCAROUTER_API_KEY="sk-orca-..." +``` + +`LLM_API_KEY` works too, like every other provider. + +### Sign in with your OrcaRouter account + +No key to copy: Strix opens the OrcaRouter consent screen in your browser (OAuth 2.0 with PKCE) and stores the API key it issues in `~/.strix/subscription-auth.json`. + +```bash +strix auth login orcarouter # approve in the browser +export STRIX_LLM="orcarouter/orcarouter/auto" +strix auth status orcarouter # show which credential is in use +``` + +On a machine where the browser can't reach back to the terminal (SSH, containers), add `--manual`: OrcaRouter shows a code on the consent screen and you paste it into the terminal. + +The issued key has no expiry and is reused on every run until you revoke it at [Authorized apps](https://www.orcarouter.ai/console/authorized-apps). If OrcaRouter rejects it, Strix asks you to run `strix auth login orcarouter` again. `strix auth logout orcarouter` removes the stored key from this machine. + +When more than one credential is present, Strix uses `ORCAROUTER_API_KEY` first, then the signed-in key, then `LLM_API_KEY`. + +## Models for security testing + +Every id uses the `orcarouter/` form. Browse the full [Cyber Inference catalog](https://www.orcarouter.ai/models?tier=cyber): + +| Tier | Model | `STRIX_LLM` | +|------|-------|-------------| +| Default — works with any key | Adaptive routing | `orcarouter/orcarouter/auto` | +| Dedicated cyber | OrcaCyber Ultra | `orcarouter/orcacyber-ultra-1.0` | +| Dedicated cyber | OrcaCyber Zero | `orcarouter/orcacyber-zero-1.5` | +| General fallback | OrcaRouter Fusion | `orcarouter/orcarouter/fusion` | + +- **`orcarouter/auto`** is the zero-config default: it routes each request to a capable upstream model and works with any OrcaRouter key. +- **OrcaCyber Ultra** and **OrcaCyber Zero** are the dedicated [Cyber Inference](https://www.orcarouter.ai/cyber) models, tuned for authorized red-team and blue-team workflows. A security team unlocks them once OrcaRouter has verified the engagement is scoped and authorized. +- **`orcarouter/fusion`** is a general-purpose fallback for when a cyber-tier model isn't enabled. + +For agentic scans, pin a model with tool-calling support. Any chat model in the [OrcaRouter catalog](https://api.orcarouter.ai/v1/models) also works as `orcarouter/`. + +## Get API Key + +1. Go to [orcarouter.ai](https://www.orcarouter.ai) +2. Sign in and open the console +3. Create a new API key + +## Self-hosted gateways + +`ORCA_BASE_URL` points both sign-in and inference at one origin. `ORCA_AUTH_BASE_URL` and `ORCA_API_BASE_URL` (including `/v1`) override each side separately and take precedence. Remote origins must use `https://`. diff --git a/docs/llm-providers/overview.mdx b/docs/llm-providers/overview.mdx index c00351271..8423c50c7 100644 --- a/docs/llm-providers/overview.mdx +++ b/docs/llm-providers/overview.mdx @@ -60,6 +60,9 @@ See the [Local Models guide](/llm-providers/local) for setup instructions and re Access 100+ models through a single API. + + Scoped, authorized models for security testing. API key or browser sign-in. + Access 100+ models through one OpenAI-compatible endpoint. diff --git a/strix/config/models.py b/strix/config/models.py index 102d132cf..5e73933b3 100644 --- a/strix/config/models.py +++ b/strix/config/models.py @@ -39,7 +39,7 @@ from openai.types.responses.response_usage import ResponseUsage from openai.types.shared import Reasoning from strix.agents.prompt import CACHE_POINT -from strix.config import codex +from strix.config import codex, orcarouter from strix.config.loader import load_settings from strix.config.tool_call_arguments import repair_input from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input @@ -608,8 +608,22 @@ class StrixProvider(MultiProvider): ) if prefix == "ollama" and stripped_model_name: return self._get_fallback_provider("litellm"), f"ollama_chat/{stripped_model_name}" + if prefix.lower() == orcarouter.PROVIDER and stripped_model_name: + return self._orcarouter_provider(), orcarouter.litellm_model_name(stripped_model_name) return self._get_fallback_provider("litellm"), original_model_name + def _orcarouter_provider(self) -> ModelProvider: + """OrcaRouter's OpenAI-compatible API, with a key from either sign-in method.""" + if self._override_api_key: + api_key = self._override_api_key + else: + credential = orcarouter.resolve_credential(load_settings().llm.api_key) + orcarouter.remember_active(credential) + api_key = credential.key + return _CredentialedLitellmProvider( + api_key, self._override_base_url or orcarouter.api_base() + ) + def get_model(self, model_name: str | None) -> Model: llm = load_settings().llm slug = codex.subscription_model(model_name) @@ -637,11 +651,15 @@ class StrixProvider(MultiProvider): if _routes_via_litellm(model): # LiteLLM's callbacks log every reply; only a cancelled attempt # (stream idle timeout, abandoned turn) escapes them. + is_orcarouter = orcarouter.route_model(resolved_name) is not None model = request_log.RequestLoggingModel( model, model_name=resolved_name, - provider=_litellm_provider(resolved_name), - base_url=self._override_base_url or llm.api_base, + provider=( + orcarouter.PROVIDER if is_orcarouter else _litellm_provider(resolved_name) + ), + base_url=self._override_base_url + or (orcarouter.api_base() if is_orcarouter else llm.api_base), route="litellm", abandoned_only=True, ) diff --git a/strix/config/orcarouter.py b/strix/config/orcarouter.py new file mode 100644 index 000000000..b680169a2 --- /dev/null +++ b/strix/config/orcarouter.py @@ -0,0 +1,469 @@ +"""OrcaRouter: API-key and OAuth 2.0 + PKCE sign-in, and the ``orcarouter/`` route. + +`OrcaRouter `_ is an OpenAI-compatible gateway. A +``STRIX_LLM`` of ``orcarouter/`` (e.g. ``orcarouter/openai/gpt-5.5``) +runs through LiteLLM's OpenAI-compatible client against the OrcaRouter API. + +Two ways to supply the credential, both ending in the same ordinary +``sk-orca-…`` key: + +- an existing API key (``ORCAROUTER_API_KEY``, or ``LLM_API_KEY``); +- ``strix auth login orcarouter``, which runs OAuth 2.0 + PKCE in the browser and + stores the issued key next to the ChatGPT sign-in in + ``~/.strix/subscription-auth.json``. + +The key issued by the login flow is durable: there is no refresh token and no +refresh grant. It is reused until the user revokes it; a ``401`` marks that +exact stored key as needing a new sign-in. +""" + +from __future__ import annotations + +import contextlib +import hmac +import ipaddress +import json +import logging +import os +import threading +import time +import urllib.parse +from dataclasses import dataclass +from typing import Any, cast + +import requests + +from strix.config import codex +from strix.utils.secret_files import write_secret_text + + +logger = logging.getLogger(__name__) + + +PROVIDER = "orcarouter" +MODEL_PREFIX = "orcarouter/" + +DEFAULT_AUTH_BASE = "https://www.orcarouter.ai" +DEFAULT_API_BASE = "https://api.orcarouter.ai/v1" +AUTHORIZE_PATH = "/auth" +EXCHANGE_PATH = "/api/v1/auth/keys" +OOB_CALLBACK = "oob" +CALLBACK_PATH = "/callback" +SCOPE = "api" +APP_NAME = "Strix" + +HOMEPAGE_URL = "https://www.orcarouter.ai" +AUTHORIZED_APPS_URL = "https://www.orcarouter.ai/console/authorized-apps" + +API_KEY_ENV = "ORCAROUTER_API_KEY" +SHARED_BASE_ENV = "ORCA_BASE_URL" +AUTH_BASE_ENV = "ORCA_AUTH_BASE_URL" +API_BASE_ENV = "ORCA_API_BASE_URL" + +# Credential sources, in resolution order. +SOURCE_ENV = "env" +SOURCE_LOGIN = "login" +SOURCE_LLM_API_KEY = "llm_api_key" + +# Used when the live catalog is slow or unavailable, so a fresh install can +# still suggest a working model. +SEED_MODELS = ( + "openai/gpt-5.5", + "anthropic/claude-opus-4.8", + "google/gemini-3.5-flash", + "deepseek/deepseek-v4-pro", + "orcarouter/auto", +) + +_EXCHANGE_TIMEOUT_S = 30 +_CATALOG_TIMEOUT_S = 10 +_CATALOG_MAX_BYTES = 4 * 1024 * 1024 +_CATALOG_MAX_ITEMS = 2000 +_MODEL_ID_MAX_LEN = 200 + +_store_lock = threading.Lock() + + +class OrcaRouterAuthError(Exception): + def __init__(self, code: str, message: str | None = None) -> None: + self.code = code + super().__init__(message or code) + + +@dataclass(frozen=True) +class Credential: + """An OrcaRouter API key and where it came from.""" + + key: str = "" + source: str = "" + + def __repr__(self) -> str: # never render the key + return f"Credential(source={self.source!r})" + + +# --- Model route ------------------------------------------------------------- + + +def route_model(model_name: str | None) -> str | None: + """The OrcaRouter model id behind an ``orcarouter/`` STRIX_LLM, or None.""" + name = (model_name or "").strip() + if not name.lower().startswith(MODEL_PREFIX): + return None + return name[len(MODEL_PREFIX) :] or None + + +def litellm_model_name(model_id: str) -> str: + """LiteLLM's name for an OrcaRouter model: its OpenAI-compatible client.""" + return f"openai/{model_id}" + + +# --- Origins ----------------------------------------------------------------- + + +def _clean_env(name: str) -> str: + return (os.environ.get(name) or "").strip().rstrip("/") + + +def _require_secure_origin(url: str, env_name: str) -> str: + parsed = urllib.parse.urlparse(url) + if parsed.scheme == "https" and parsed.hostname: + return url + if parsed.scheme == "http" and _is_loopback(parsed.hostname): + return url + raise OrcaRouterAuthError( + "insecure_origin", + f"{env_name} must be an https:// URL (http:// is allowed for loopback only)", + ) + + +def _is_loopback(host: str | None) -> bool: + if not host: + return False + if host == "localhost": + return True + try: + return ipaddress.ip_address(host).is_loopback + except ValueError: + return False + + +def auth_base() -> str: + """Origin of the consent screen and code exchange (``www``, not ``api``).""" + if explicit := _clean_env(AUTH_BASE_ENV): + return _require_secure_origin(explicit, AUTH_BASE_ENV) + if shared := _clean_env(SHARED_BASE_ENV): + return _require_secure_origin(shared, SHARED_BASE_ENV) + return DEFAULT_AUTH_BASE + + +def api_base() -> str: + """Base URL of the OpenAI-compatible inference API, including ``/v1``.""" + if explicit := _clean_env(API_BASE_ENV): + return _require_secure_origin(explicit, API_BASE_ENV) + if shared := _clean_env(SHARED_BASE_ENV): + return _require_secure_origin(shared, SHARED_BASE_ENV) + "/v1" + return DEFAULT_API_BASE + + +# --- OAuth 2.0 + PKCE -------------------------------------------------------- + + +def generate_pkce() -> tuple[str, str]: + """A fresh ``(verifier, S256 challenge)`` pair from the OS CSPRNG.""" + return codex.generate_pkce() + + +def create_state() -> str: + return codex.create_state() + + +def build_authorize_url(challenge: str, state: str, callback_url: str) -> str: + """The consent-screen URL. ``callback_url`` is a loopback URL or ``"oob"``. + + Only the challenge travels on the URL; the verifier stays in this process. + """ + params = { + "callback_url": callback_url, + "code_challenge": challenge, + "code_challenge_method": "S256", + "state": state, + "app_name": APP_NAME, + "scope": SCOPE, + } + return f"{auth_base()}{AUTHORIZE_PATH}?{urllib.parse.urlencode(params)}" + + +def state_matches(returned: str | None, expected: str) -> bool: + return returned is not None and hmac.compare_digest(returned, expected) + + +_EXCHANGE_ERRORS = { + 400: ("bad_request", "OrcaRouter rejected the sign-in request. Start a new sign-in."), + 403: ( + "invalid_code", + "The authorization code is invalid, expired, or was already used. " + "Run `strix auth login orcarouter` again.", + ), + 429: ( + "rate_limited", + "Too many OrcaRouter sign-ins in the last 24 hours. Reuse the existing sign-in, " + "or use an API key (ORCAROUTER_API_KEY) instead.", + ), +} + + +def exchange_code(code: str, verifier: str) -> dict[str, Any]: + """Redeem an authorization code for a durable OrcaRouter API key record.""" + try: + with requests.post( + f"{auth_base()}{EXCHANGE_PATH}", + json={"code": code, "code_verifier": verifier, "code_challenge_method": "S256"}, + headers={"Accept": "application/json"}, + timeout=_EXCHANGE_TIMEOUT_S, + ) as response: + status_code = response.status_code + body = response.content + except requests.RequestException as exc: + # The exception text can echo the request; keep only its type. + raise OrcaRouterAuthError( + "unavailable", f"could not reach OrcaRouter ({type(exc).__name__})" + ) from None + if status_code in _EXCHANGE_ERRORS: + raise OrcaRouterAuthError(*_EXCHANGE_ERRORS[status_code]) + if status_code >= 400: + raise OrcaRouterAuthError("exchange_http_error", f"key exchange failed: HTTP {status_code}") + try: + data = json.loads(body or b"{}") + except (ValueError, UnicodeDecodeError): + raise OrcaRouterAuthError("bad_response", "key exchange returned invalid JSON") from None + if not isinstance(data, dict): + raise OrcaRouterAuthError("bad_response", "key exchange returned a non-object") + grant = cast("dict[str, Any]", data) + key = grant.get("key") + if not isinstance(key, str) or not key.strip(): + raise OrcaRouterAuthError("bad_response", "key exchange response has no key") + # Read back what was granted, not what was asked for. + scope = grant.get("scope") or SCOPE + if scope != SCOPE: + raise OrcaRouterAuthError( + "scope_mismatch", f"OrcaRouter granted scope {scope!r}; Strix needs {SCOPE!r}" + ) + user_id = grant.get("user_id") + return { + "type": "api_key", + "provider": PROVIDER, + "key": key.strip(), + "scope": scope, + "user_id": str(user_id) if user_id is not None else None, + "created_at": time.time(), + "needs_reauth": False, + } + + +# --- Stored sign-in ---------------------------------------------------------- +# Shares the ChatGPT sign-in file (one entry per provider) rather than adding a +# second secret store. + + +def _read_store() -> dict[str, Any]: + try: + data = json.loads(codex.AUTH_PATH.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return cast("dict[str, Any]", data) if isinstance(data, dict) else {} + + +def _write_store(data: dict[str, Any]) -> None: + write_secret_text(codex.AUTH_PATH, json.dumps(data, indent=2)) + + +def read_record() -> dict[str, Any] | None: + record = _read_store().get(PROVIDER) + if not isinstance(record, dict): + return None + entry = cast("dict[str, Any]", record) + key = entry.get("key") + if entry.get("type") != "api_key" or not isinstance(key, str) or not key: + return None + return entry + + +def save_record(record: dict[str, Any]) -> None: + with _store_lock: + data = _read_store() + data[PROVIDER] = record + _write_store(data) + + +def logout() -> bool: + """Forget the stored sign-in. Returns whether there was one.""" + with _store_lock: + data = _read_store() + if PROVIDER not in data: + return False + del data[PROVIDER] + if data: + _write_store(data) + else: + with contextlib.suppress(OSError): + codex.AUTH_PATH.unlink() + return True + + +def mark_needs_reauth(rejected_key: str) -> bool: + """Flag the stored sign-in as revoked, if it still holds ``rejected_key``. + + Compares the exact key so a late ``401`` from an old credential never + flags a newer sign-in. The key is kept until a new sign-in replaces it. + """ + with _store_lock: + data = _read_store() + record = data.get(PROVIDER) + if not isinstance(record, dict): + return False + entry = cast("dict[str, Any]", record) + stored_key = entry.get("key") + if not isinstance(stored_key, str) or not hmac.compare_digest(stored_key, rejected_key): + return False + entry["needs_reauth"] = True + _write_store(data) + return True + + +# --- Credential resolution --------------------------------------------------- + + +def resolve_credential(llm_api_key: str | None = None) -> Credential: + """The key an ``orcarouter/`` request runs with. + + ``ORCAROUTER_API_KEY`` wins, then a usable ``strix auth login orcarouter`` + sign-in, then ``LLM_API_KEY``. The sign-in outranks ``LLM_API_KEY`` because + that variable is persisted across runs and often still holds another + provider's key. + """ + if env_key := (os.environ.get(API_KEY_ENV) or "").strip(): + return Credential(env_key, SOURCE_ENV) + record = read_record() + if record is not None and not record.get("needs_reauth"): + return Credential(record["key"], SOURCE_LOGIN) + if explicit := (llm_api_key or "").strip(): + return Credential(explicit, SOURCE_LLM_API_KEY) + if record is not None: + raise OrcaRouterAuthError( + "needs_reauth", + "Your OrcaRouter sign-in was revoked or expired. Sign in again:\n" + " strix auth login orcarouter", + ) + raise OrcaRouterAuthError( + "not_authenticated", + "No OrcaRouter credential. Either sign in:\n" + " strix auth login orcarouter\n" + f"or set an API key: export {API_KEY_ENV}='sk-orca-...'", + ) + + +_active: Credential | None = None + + +def remember_active(credential: Credential) -> None: + """Record the credential the current run's requests are sent with.""" + global _active # noqa: PLW0603 + _active = credential + + +def handle_unauthorized() -> str: + """React to a ``401`` from OrcaRouter and return an actionable hint.""" + credential = _active + if credential is not None and credential.source == SOURCE_LOGIN: + mark_needs_reauth(credential.key) + return ( + "OrcaRouter rejected the signed-in key (it was likely revoked). Sign in again:\n" + " strix auth login orcarouter" + ) + variable = API_KEY_ENV if credential and credential.source == SOURCE_ENV else "LLM_API_KEY" + return ( + f"OrcaRouter rejected the API key in {variable}. Check or replace it at " + f"{HOMEPAGE_URL}, or sign in instead with: strix auth login orcarouter" + ) + + +# --- Model catalog ----------------------------------------------------------- + + +def fetch_models(api_key: str | None = None) -> list[str]: + """Chat-capable model ids from the live catalog, or ``[]`` if unavailable. + + The request is bounded in time and size; entries that are malformed or + cannot be served over the OpenAI-compatible chat API are dropped. + """ + headers = {"Accept": "application/json"} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + try: + with requests.get( + f"{api_base()}/models", headers=headers, timeout=_CATALOG_TIMEOUT_S, stream=True + ) as response: + if response.status_code != 200: + return [] + body = response.raw.read(_CATALOG_MAX_BYTES + 1, decode_content=True) + except (requests.RequestException, OrcaRouterAuthError, OSError): + return [] + if len(body) > _CATALOG_MAX_BYTES: + return [] + try: + data = json.loads(body) + except (ValueError, UnicodeDecodeError): + return [] + items: Any = cast("dict[str, Any]", data).get("data") if isinstance(data, dict) else None + if not isinstance(items, list): + return [] + entries = cast("list[object]", items) + models: list[str] = [] + for raw in entries[:_CATALOG_MAX_ITEMS]: + if not isinstance(raw, dict): + continue + item = cast("dict[str, Any]", raw) + model_id = item.get("id") + if not isinstance(model_id, str) or not 0 < len(model_id) <= _MODEL_ID_MAX_LEN: + continue + endpoint_types = item.get("supported_endpoint_types") + if isinstance(endpoint_types, list) and "openai" not in endpoint_types: + continue + models.append(model_id) + return models + + +def available_models(api_key: str | None = None) -> tuple[list[str], bool]: + """``(model ids, is_live)``: the live catalog, or the verified seed.""" + live = fetch_models(api_key) + if live: + return live, True + return list(SEED_MODELS), False + + +def mask_key(key: str) -> str: + return f"{key[:8]}…{key[-4:]}" if len(key) > 16 else "…" + + +__all__ = [ + "API_KEY_ENV", + "MODEL_PREFIX", + "PROVIDER", + "SEED_MODELS", + "Credential", + "OrcaRouterAuthError", + "api_base", + "auth_base", + "available_models", + "build_authorize_url", + "exchange_code", + "fetch_models", + "handle_unauthorized", + "litellm_model_name", + "logout", + "mark_needs_reauth", + "read_record", + "resolve_credential", + "route_model", + "save_record", +] diff --git a/strix/interface/auth_cli.py b/strix/interface/auth_cli.py index 51e6b9fe9..9e495033e 100644 --- a/strix/interface/auth_cli.py +++ b/strix/interface/auth_cli.py @@ -1,14 +1,16 @@ -"""`strix auth` — ChatGPT subscription sign-in (login / status / logout). +"""`strix auth` — ChatGPT subscription and OrcaRouter sign-in (login / status / logout). -Signing in only stores OAuth tokens (``~/.strix/subscription-auth.json``); model +Signing in only stores credentials (``~/.strix/subscription-auth.json``); model selection stays with ``STRIX_LLM``. A ``chatgpt/`` STRIX_LLM runs on the -subscription. +subscription; an ``orcarouter/`` STRIX_LLM runs on OrcaRouter with the key +issued by ``strix auth login orcarouter`` (or an ``ORCAROUTER_API_KEY``). """ from __future__ import annotations import argparse import base64 +import contextlib import logging import threading import webbrowser @@ -21,7 +23,7 @@ from rich.console import Console from rich.panel import Panel from rich.text import Text -from strix.config import codex, load_settings +from strix.config import codex, load_settings, orcarouter if TYPE_CHECKING: @@ -37,8 +39,15 @@ _CALLBACK_TIMEOUT_S = 300 # command and messaging say. ``codex`` is accepted as an alias. LOGIN_PROVIDER = "chatgpt" _ACCEPTED_PROVIDERS = frozenset({LOGIN_PROVIDER, codex.PROVIDER}) +ORCAROUTER_PROVIDER = orcarouter.PROVIDER -_USAGE = "Usage:\n strix auth login chatgpt [--manual]\n strix auth status\n strix auth logout" +_USAGE = ( + "Usage:\n" + " strix auth login chatgpt [--manual]\n" + " strix auth login orcarouter [--manual]\n" + " strix auth status [chatgpt|orcarouter]\n" + " strix auth logout [chatgpt|orcarouter]" +) def run_auth(argv: list[str]) -> int: @@ -49,20 +58,20 @@ def run_auth(argv: list[str]) -> int: rest = argv[1:] if subcommand in ("-h", "--help", "help"): - console.print(_USAGE) + console.print(_USAGE, markup=False) return 0 handlers: dict[str, Callable[[], int]] = { "login": lambda: _login(console, rest), - "status": lambda: _status(console), - "logout": lambda: _logout(console), + "status": lambda: _status_command(console, rest), + "logout": lambda: _logout_command(console, rest), } handler = handlers.get(subcommand) if handler is not None: return handler() console.print(f"[red]Unknown auth command:[/] {subcommand}\n") - console.print(_USAGE) + console.print(_USAGE, markup=False) return 2 @@ -77,17 +86,23 @@ def _login(console: Console, argv: list[str]) -> int: parser.add_argument( "--manual", action="store_true", - help="Skip the local callback server and paste the redirect URL by hand.", + help=( + "Skip the local callback server and paste the redirect URL by hand " + "(for OrcaRouter, the code shown on the consent screen)." + ), ) try: args = parser.parse_args(argv) except SystemExit as exc: # argparse already printed the message return int(exc.code or 2) + if args.provider.lower() == ORCAROUTER_PROVIDER: + return _login_orcarouter(console, manual=args.manual) if args.provider.lower() not in _ACCEPTED_PROVIDERS: console.print( f"[red]Unsupported provider:[/] {args.provider}. " - f"Only '{LOGIN_PROVIDER}' (ChatGPT subscription) is supported." + f"Supported: '{LOGIN_PROVIDER}' (ChatGPT subscription) and " + f"'{ORCAROUTER_PROVIDER}' (OrcaRouter)." ) return 2 @@ -242,6 +257,254 @@ def _first(query: dict[str, list[str]], key: str) -> str | None: return values[0] if values else None +# --- OrcaRouter -------------------------------------------------------------- + + +def _login_orcarouter(console: Console, *, manual: bool) -> int: + console.print() + console.print("[bold]Signing in with OrcaRouter[/] [dim](provider: orcarouter)[/]") + console.print( + "[dim]Authorizes Strix in your browser and stores the OrcaRouter API key it issues. " + f"Already have a key? Set {orcarouter.API_KEY_ENV} instead.[/]" + ) + console.print() + try: + record = _run_orcarouter_flow(console, manual=manual) + except orcarouter.OrcaRouterAuthError as exc: + return _fail(console, exc) + except KeyboardInterrupt: + console.print("\n[yellow]Sign-in cancelled.[/]") + return 130 + + orcarouter.save_record(record) + _print_orcarouter_success(console, record["key"]) + return 0 + + +def _run_orcarouter_flow(console: Console, *, manual: bool) -> dict[str, Any]: + """Loopback redirect, or an out-of-band code the user pastes back. + + Every attempt gets a fresh verifier and state; only the S256 challenge + leaves this process before the code exchange. + """ + verifier, challenge = orcarouter.generate_pkce() + state = orcarouter.create_state() + server = None if manual else _try_start_orcarouter_callback_server() + try: + callback_url = ( + f"http://127.0.0.1:{server.port}{orcarouter.CALLBACK_PATH}" + if server is not None + else orcarouter.OOB_CALLBACK + ) + authorize_url = orcarouter.build_authorize_url(challenge, state, callback_url) + + console.print("Open this URL in your browser to authorize:") + console.print(f"[cyan]{authorize_url}[/]") + console.print() + with contextlib.suppress(Exception): # opening a browser is best-effort + webbrowser.open(authorize_url) + + if server is not None: + console.print("[dim]Waiting for you to approve in the browser…[/]") + result = server.wait(_CALLBACK_TIMEOUT_S) + if result is not None: + code, returned_state, error = result + # The redirect always carries state; anything else is forged. + if not orcarouter.state_matches(returned_state, state): + raise orcarouter.OrcaRouterAuthError( + "state_mismatch", "state did not match; possible CSRF" + ) + if error: + message = "Sign-in was denied." if error == "access_denied" else error + raise orcarouter.OrcaRouterAuthError(error, message) + return _orcarouter_exchange(code, verifier) + console.print( + "[yellow]Timed out waiting for the browser. " + "Paste the code or the redirect URL instead.[/]" + ) + finally: + if server is not None: + server.shutdown() + + console.print() + try: + pasted = console.input("Paste the code shown by OrcaRouter (or the redirect URL): ") + except EOFError as exc: + raise orcarouter.OrcaRouterAuthError("no_input", "no code provided") from exc + code, returned_state = codex.parse_redirect_input(pasted.strip()) + if returned_state is not None and not orcarouter.state_matches(returned_state, state): + raise orcarouter.OrcaRouterAuthError("state_mismatch", "state did not match; possible CSRF") + return _orcarouter_exchange(code, verifier) + + +def _orcarouter_exchange(code: str | None, verifier: str) -> dict[str, Any]: + if not code: + raise orcarouter.OrcaRouterAuthError("no_code", "no authorization code was provided") + return orcarouter.exchange_code(code, verifier) + + +class _PortCallbackServer(_CallbackServer): + @property + def port(self) -> int: + return int(self._httpd.server_address[1]) + + +def _try_start_orcarouter_callback_server() -> _PortCallbackServer | None: + event = threading.Event() + holder: dict[str, Any] = {} + + class Handler(BaseHTTPRequestHandler): + def log_message(self, format: str, *args: Any) -> None: # noqa: A002 - the query carries the code + pass + + def do_GET(self) -> None: + parsed = urlparse(self.path) + if parsed.path != orcarouter.CALLBACK_PATH or event.is_set(): + self.send_response(404) + self.end_headers() + return + query = parse_qs(parsed.query) + holder["code"] = _first(query, "code") + holder["state"] = _first(query, "state") + holder["error"] = _first(query, "error") + body = _render_orcarouter_callback_html(denied=bool(holder["error"])).encode() + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + event.set() + + try: + # OrcaRouter accepts any loopback port, so let the OS pick a free one. + httpd = HTTPServer(("127.0.0.1", 0), Handler) + except OSError: + logger.debug("could not bind an OrcaRouter callback port", exc_info=True) + return None + return _PortCallbackServer(httpd, event, holder) + + +_ORCAROUTER_SOURCES = { + orcarouter.SOURCE_ENV: f"API key from {orcarouter.API_KEY_ENV}", + orcarouter.SOURCE_LOGIN: "signed in with `strix auth login orcarouter`", + orcarouter.SOURCE_LLM_API_KEY: "API key from LLM_API_KEY", +} + + +def _orcarouter_status(console: Console) -> int: + settings = load_settings() + try: + credential = orcarouter.resolve_credential(settings.llm.api_key) + except orcarouter.OrcaRouterAuthError as exc: + console.print(f"[yellow]OrcaRouter: no usable credential.[/] {exc}") + return 1 + console.print(f"[green]OrcaRouter[/]: {_ORCAROUTER_SOURCES[credential.source]}.") + console.print(f" Key: [bold]{orcarouter.mask_key(credential.key)}[/]") + record = orcarouter.read_record() + if record is not None and record.get("needs_reauth"): + console.print( + " [yellow]The stored sign-in was rejected by OrcaRouter.[/] " + "Run [cyan]strix auth login orcarouter[/] to replace it." + ) + console.print(f" API: [bold]{orcarouter.api_base()}[/]") + if orcarouter.route_model(settings.llm.model): + console.print(f" Runs use OrcaRouter (STRIX_LLM=[bold]{settings.llm.model}[/]).") + else: + console.print( + " [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. " + f"[cyan]{orcarouter.MODEL_PREFIX}{orcarouter.SEED_MODELS[0]}[/] to run on OrcaRouter." + ) + return 0 + + +def _orcarouter_logout(console: Console) -> None: + if orcarouter.logout(): + console.print( + "[green]Signed out of OrcaRouter.[/] The stored key was removed from this machine; " + f"revoke it at {orcarouter.AUTHORIZED_APPS_URL}" + ) + + +def _suggested_orcarouter_model(api_key: str) -> str: + models, _live = orcarouter.available_models(api_key) + for seed in orcarouter.SEED_MODELS: + if seed in models: + return seed + return models[0] + + +def _print_orcarouter_success(console: Console, api_key: str) -> None: + model = f"{orcarouter.MODEL_PREFIX}{_suggested_orcarouter_model(api_key)}" + text = Text() + text.append("Signed in with OrcaRouter", style="bold #22c55e") + text.append("\n\n", style="white") + text.append("Set ", style="white") + text.append("STRIX_LLM", style="bold white") + text.append(" to an ", style="white") + text.append(orcarouter.MODEL_PREFIX, style="bold cyan") + text.append(" model, e.g. ", style="white") + text.append(model, style="bold cyan") + text.append("\n\n", style="white") + text.append("The key is reused until you revoke it at ", style="white") + text.append(orcarouter.AUTHORIZED_APPS_URL, style="cyan") + console.print() + console.print( + Panel( + text, + title="[bold white]STRIX", + title_align="left", + border_style="#22c55e", + padding=(1, 2), + ) + ) + console.print() + + +def _render_orcarouter_callback_html(*, denied: bool) -> str: + if denied: + return _render_callback_html( + heading="Sign-in cancelled", + message="OrcaRouter access was not granted. Head back to your terminal to retry.", + ) + return _render_callback_html( + message="Strix is connected to OrcaRouter. Head back to your terminal — " + "your security test runs there." + ) + + +def _provider_arg(argv: list[str]) -> str | None: + return argv[0].lower() if argv else None + + +def _status_command(console: Console, argv: list[str]) -> int: + provider = _provider_arg(argv) + if provider == ORCAROUTER_PROVIDER: + return _orcarouter_status(console) + if provider is not None and provider not in _ACCEPTED_PROVIDERS: + console.print(f"[red]Unknown provider:[/] {provider}\n") + console.print(_USAGE, markup=False) + return 2 + code = _status(console) + if provider is None and orcarouter.read_record() is not None: + console.print() + code = min(code, _orcarouter_status(console)) + return code + + +def _logout_command(console: Console, argv: list[str]) -> int: + provider = _provider_arg(argv) + if provider == ORCAROUTER_PROVIDER: + if orcarouter.read_record() is None: + console.print("[yellow]Not signed in to OrcaRouter.[/]") + _orcarouter_logout(console) + return 0 + if provider is not None and provider not in _ACCEPTED_PROVIDERS: + console.print(f"[red]Unknown provider:[/] {provider}\n") + console.print(_USAGE, markup=False) + return 2 + return _logout(console) + + def _status(console: Console) -> int: record = codex.read_record() if record is None: @@ -266,7 +529,7 @@ def _logout(console: Console) -> int: return 0 -def _fail(console: Console, exc: codex.CodexAuthError) -> int: +def _fail(console: Console, exc: codex.CodexAuthError | orcarouter.OrcaRouterAuthError) -> int: error_text = Text() error_text.append("SIGN-IN FAILED", style="bold red") error_text.append("\n\n", style="white") @@ -329,8 +592,20 @@ def _logo_img_tag() -> str: return f'' -def _render_callback_html() -> str: - return _CALLBACK_HTML.replace("", _logo_img_tag()) +_CHATGPT_CALLBACK_MESSAGE = ( + "Strix is connected to your ChatGPT subscription. Head back to your\n" + " terminal — your security test runs there." +) + + +def _render_callback_html( + *, heading: str = "You're signed in", message: str = _CHATGPT_CALLBACK_MESSAGE +) -> str: + return ( + _CALLBACK_HTML.replace("", _logo_img_tag()) + .replace("", heading) + .replace("", message) + ) _CALLBACK_HTML = """ @@ -396,11 +671,10 @@ _CALLBACK_HTML = """ Strix
Strix
-

You're signed in

+

✓
-

Strix is connected to your ChatGPT subscription. Head back to your - terminal — your security test runs there.

+

Autonomous AI hackers that find and fix your app's vulnerabilities.

diff --git a/strix/interface/environment.py b/strix/interface/environment.py index 522067df8..e3b5ff2b8 100644 --- a/strix/interface/environment.py +++ b/strix/interface/environment.py @@ -8,7 +8,7 @@ from rich.console import Console from rich.panel import Panel from rich.text import Text -from strix.config import IntegrationSettings, codex, load_settings +from strix.config import IntegrationSettings, codex, load_settings, orcarouter from strix.interface.utils import ( check_docker_connection, image_exists, @@ -50,6 +50,14 @@ def validate_environment() -> None: logger.info("Environment OK (ChatGPT subscription)") return + if orcarouter.route_model(settings.llm.model): + try: + orcarouter.resolve_credential(settings.llm.api_key) + except orcarouter.OrcaRouterAuthError as exc: + console.print(f"[red]STRIX_LLM={settings.llm.model} runs on OrcaRouter.[/] {exc}") + report_error(f"orcarouter_{exc.code}") + sys.exit(1) + if not settings.llm.model: missing_required_vars.append("STRIX_LLM") diff --git a/strix/interface/main.py b/strix/interface/main.py index 846a35da9..bc6457238 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -15,7 +15,7 @@ from rich.console import Console from rich.panel import Panel from rich.text import Text -from strix.config import codex, load_settings, persist_current +from strix.config import codex, load_settings, orcarouter, persist_current from strix.core.paths import RUNS_DIR_NAME, run_dir_for from strix.interface.cli_args import ( FAIL_ON_SEVERITIES, @@ -62,7 +62,7 @@ logger = logging.getLogger(__name__) _ROOT_SUBCOMMAND_HELP = """ Additional commands: strix cloud ... Use the managed Strix platform - strix auth ... Manage model-subscription sign-in + strix auth ... Manage ChatGPT / OrcaRouter sign-in strix view [RUN] View a completed or running scan strix completions SHELL Generate zsh, bash, or fish tab completion """ @@ -312,16 +312,34 @@ def _print_error_panel(title: str, message: str) -> None: console.print() -def _print_model_connection_error(exc: BaseException) -> None: +def _orcarouter_error_hint(exc: BaseException, model_name: str) -> str | None: + """Return a re-authentication hint when OrcaRouter rejected the credential.""" + if orcarouter.route_model(model_name) is None: + return None + if isinstance(exc, orcarouter.OrcaRouterAuthError): + return str(exc) + joined = " ".join(_exception_messages(exc)).lower() + if "error code: 401" in joined or "authenticationerror" in joined: + return orcarouter.handle_unauthorized() + return None + + +def _print_model_connection_error(exc: BaseException, model_name: str) -> None: console = Console() error_text = Text() sub_hint = _subscription_error_hint(exc) + orca_hint = _orcarouter_error_hint(exc, model_name) if sub_hint is not None: border_style = "yellow" error_text.append("MODEL NOT AVAILABLE ON SUBSCRIPTION", style="bold yellow") error_text.append("\n\n", style="white") error_text.append(f"{sub_hint}\n", style="white") error_text.append(f"\nDetails: {exc}", style="dim white") + elif orca_hint is not None: + border_style = "yellow" + error_text.append("ORCAROUTER CREDENTIAL REJECTED", style="bold yellow") + error_text.append("\n\n", style="white") + error_text.append(f"{orca_hint}\n", style="white") else: border_style = "red" error_text.append("LLM CONNECTION FAILED", style="bold red") @@ -406,7 +424,7 @@ def _bootstrap_scan(args: argparse.Namespace) -> None: asyncio.run(warm_up_llm()) except ModelConnectionError as exc: report_error("model_connection_failed", exc) - _print_model_connection_error(exc) + _print_model_connection_error(exc, exc.model_name) sys.exit(1) persist_current() try: diff --git a/tests/test_orcarouter.py b/tests/test_orcarouter.py new file mode 100644 index 000000000..5789404bb --- /dev/null +++ b/tests/test_orcarouter.py @@ -0,0 +1,663 @@ +"""Tests for the OrcaRouter provider: API-key and OAuth 2.0 + PKCE sign-in, routing, +credential lifecycle, and the model catalog. + +A local fake of the OrcaRouter auth and API origins stands in for the network, +so the sign-in runs end to end: authorize URL -> loopback callback or pasted +code -> key exchange -> stored key -> an ``orcarouter/`` model bound to it. +""" + +from __future__ import annotations + +import base64 +import hashlib +import json +import threading +import urllib.parse +import urllib.request +from http.server import BaseHTTPRequestHandler, HTTPServer +from typing import TYPE_CHECKING, Any + +import pytest + +from strix.config import codex, loader, orcarouter +from strix.config.models import StrixProvider +from strix.config.settings import DedupeSettings +from strix.interface import auth_cli, environment +from strix.interface.main import _orcarouter_error_hint +from strix.report.dedupe import resolve_dedupe_model + + +if TYPE_CHECKING: + from collections.abc import Iterator + from pathlib import Path + + +ISSUED_KEY = "sk-orca-issued-by-login-0001" +ENV_KEY = "sk-orca-from-env-0002" +LLM_KEY = "sk-orca-from-llm-api-key-0003" + +_ENV_VARS = ( + "STRIX_LLM", + "LLM_API_KEY", + "OPENAI_API_KEY", + "LLM_API_BASE", + "OPENAI_API_BASE", + "OPENAI_BASE_URL", + "LITELLM_BASE_URL", + "OLLAMA_API_BASE", + orcarouter.API_KEY_ENV, + orcarouter.SHARED_BASE_ENV, + orcarouter.AUTH_BASE_ENV, + orcarouter.API_BASE_ENV, +) + + +def _b64url_sha256(value: str) -> str: + return base64.urlsafe_b64encode(hashlib.sha256(value.encode()).digest()).rstrip(b"=").decode() + + +class FakeOrcaRouter: + """Both OrcaRouter origins on one loopback server.""" + + def __init__(self) -> None: + self.requests: list[tuple[str, str, dict[str, Any]]] = [] + self.pending: dict[str, str] = {} # code -> challenge + self.exchange_status: int | None = None + self.exchange_body: dict[str, Any] | None = None + self.catalog: Any = { + "data": [ + {"id": "openai/gpt-5.5", "supported_endpoint_types": ["openai", "anthropic"]}, + {"id": "deepseek/deepseek-v4-pro", "supported_endpoint_types": ["openai"]}, + {"id": "vendor/image-only", "supported_endpoint_types": ["image-generation"]}, + {"id": "", "supported_endpoint_types": ["openai"]}, + {"id": 42}, + "not-an-object", + ] + } + fake = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, *args: Any) -> None: + pass + + def _reply(self, status: int, payload: Any) -> None: + body = payload if isinstance(payload, bytes) else json.dumps(payload).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + fake.requests.append(("GET", self.path, dict(self.headers))) + if self.path == "/v1/models": + self._reply(200, fake.catalog) + return + self._reply(404, {"error": "not found"}) + + def do_POST(self) -> None: + length = int(self.headers.get("Content-Length") or 0) + body = json.loads(self.rfile.read(length) or b"{}") + fake.requests.append(("POST", self.path, body)) + if self.path != orcarouter.EXCHANGE_PATH: + self._reply(404, {"error": "not found"}) + return + if fake.exchange_status is not None: + self._reply(fake.exchange_status, fake.exchange_body or {}) + return + challenge = fake.pending.pop(body.get("code"), None) # single use + if challenge is None or body.get("code_challenge_method") != "S256": + self._reply(403, {"error": "invalid code"}) + return + if _b64url_sha256(body.get("code_verifier", "")) != challenge: + self._reply(403, {"error": "verifier mismatch"}) + return + self._reply(200, {"key": ISSUED_KEY, "user_id": 42, "scope": "api"}) + + self._httpd = HTTPServer(("127.0.0.1", 0), Handler) + self.base = f"http://127.0.0.1:{self._httpd.server_address[1]}" + self._thread = threading.Thread(target=self._httpd.serve_forever, daemon=True) + self._thread.start() + + def approve(self, authorize_url: str, code: str = "auth-code-1") -> dict[str, str]: + """What the consent screen does: mint a code bound to the challenge.""" + parsed = urllib.parse.urlparse(authorize_url) + assert f"{parsed.scheme}://{parsed.netloc}" == self.base + assert parsed.path == orcarouter.AUTHORIZE_PATH + params = {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query).items()} + self.pending[code] = params["code_challenge"] + return params + + def close(self) -> None: + self._httpd.shutdown() + self._httpd.server_close() + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + for name in _ENV_VARS: + monkeypatch.delenv(name, raising=False) + monkeypatch.setattr(codex, "AUTH_PATH", tmp_path / ".strix" / "subscription-auth.json") + monkeypatch.setattr(loader, "_cached", None) + monkeypatch.setattr(loader, "_override", tmp_path / "no-cli-config.json") + monkeypatch.setattr(orcarouter, "_active", None) + + +@pytest.fixture +def fake(monkeypatch: pytest.MonkeyPatch) -> Iterator[FakeOrcaRouter]: + server = FakeOrcaRouter() + monkeypatch.setenv(orcarouter.AUTH_BASE_ENV, server.base) + monkeypatch.setenv(orcarouter.API_BASE_ENV, f"{server.base}/v1") + try: + yield server + finally: + server.close() + + +def _stored_record(key: str = ISSUED_KEY, **extra: Any) -> dict[str, Any]: + return {"type": "api_key", "provider": "orcarouter", "key": key, "scope": "api", **extra} + + +def _unwrap(model: object) -> object: + while hasattr(model, "_inner"): + model = model._inner + return model + + +# --- Route and origins ------------------------------------------------------- + + +@pytest.mark.parametrize( + ("model", "expected"), + [ + ("orcarouter/openai/gpt-5.5", "openai/gpt-5.5"), + ("OrcaRouter/orcarouter/auto", "orcarouter/auto"), + ("orcarouter/", None), + ("openrouter/openai/gpt-5.5", None), + (None, None), + ], +) +def test_route_model(model: str | None, expected: str | None) -> None: + assert orcarouter.route_model(model) == expected + + +def test_default_origins_keep_auth_and_inference_apart() -> None: + assert orcarouter.auth_base() == "https://www.orcarouter.ai" + assert orcarouter.api_base() == "https://api.orcarouter.ai/v1" + + +def test_shared_base_with_explicit_overrides_taking_precedence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(orcarouter.SHARED_BASE_ENV, "https://orca.internal/") + assert orcarouter.auth_base() == "https://orca.internal" + assert orcarouter.api_base() == "https://orca.internal/v1" + monkeypatch.setenv(orcarouter.AUTH_BASE_ENV, "https://auth.internal") + monkeypatch.setenv(orcarouter.API_BASE_ENV, "https://api.internal/v1") + assert orcarouter.auth_base() == "https://auth.internal" + assert orcarouter.api_base() == "https://api.internal/v1" + + +@pytest.mark.parametrize("url", ["http://orca.example.com", "ftp://127.0.0.1", "orca"]) +def test_remote_origins_must_be_https(url: str, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(orcarouter.AUTH_BASE_ENV, url) + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.auth_base() + assert exc.value.code == "insecure_origin" + + +@pytest.mark.parametrize("url", ["http://127.0.0.1:8080", "http://localhost:3000", "http://[::1]"]) +def test_http_allowed_for_loopback(url: str, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(orcarouter.API_BASE_ENV, url) + assert orcarouter.api_base() == url + + +# --- PKCE -------------------------------------------------------------------- + + +def test_pkce_is_fresh_s256_and_unpadded() -> None: + first = orcarouter.generate_pkce() + second = orcarouter.generate_pkce() + assert first != second + for verifier, challenge in (first, second): + assert 43 <= len(verifier) <= 128 + assert "=" not in challenge + assert challenge == _b64url_sha256(verifier) + assert orcarouter.create_state() != orcarouter.create_state() + + +def test_authorize_url_uses_auth_origin_and_carries_only_the_challenge() -> None: + verifier, challenge = orcarouter.generate_pkce() + url = orcarouter.build_authorize_url(challenge, "st4te", "oob") + parsed = urllib.parse.urlparse(url) + params = {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query).items()} + assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == "https://www.orcarouter.ai/auth" + assert params == { + "callback_url": "oob", + "code_challenge": challenge, + "code_challenge_method": "S256", + "state": "st4te", + "app_name": "Strix", + "scope": "api", + } + assert verifier not in url + + +def test_state_comparison() -> None: + assert orcarouter.state_matches("abc", "abc") + assert not orcarouter.state_matches("abd", "abc") + assert not orcarouter.state_matches(None, "abc") + + +# --- Key exchange ------------------------------------------------------------ + + +def test_exchange_posts_code_and_verifier_to_the_auth_origin(fake: FakeOrcaRouter) -> None: + verifier, challenge = orcarouter.generate_pkce() + fake.pending["c1"] = challenge + record = orcarouter.exchange_code("c1", verifier) + assert record["key"] == ISSUED_KEY + assert record["scope"] == "api" + assert record["user_id"] == "42" + assert record["needs_reauth"] is False + method, path, body = fake.requests[-1] + assert (method, path) == ("POST", "/api/v1/auth/keys") + assert body == {"code": "c1", "code_verifier": verifier, "code_challenge_method": "S256"} + + +def test_reused_code_is_rejected(fake: FakeOrcaRouter) -> None: + verifier, challenge = orcarouter.generate_pkce() + fake.pending["c1"] = challenge + orcarouter.exchange_code("c1", verifier) + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.exchange_code("c1", verifier) + assert exc.value.code == "invalid_code" + + +def test_wrong_verifier_is_rejected_without_leaking_it(fake: FakeOrcaRouter) -> None: + _verifier, challenge = orcarouter.generate_pkce() + fake.pending["c1"] = challenge + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.exchange_code("c1", "wrong-verifier-value") + assert exc.value.code == "invalid_code" + assert "wrong-verifier-value" not in str(exc.value) + + +@pytest.mark.parametrize( + ("status", "code"), + [ + (400, "bad_request"), + (403, "invalid_code"), + (429, "rate_limited"), + (500, "exchange_http_error"), + ], +) +def test_exchange_http_errors(status: int, code: str, fake: FakeOrcaRouter) -> None: + fake.exchange_status = status + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.exchange_code("c1", "verifier") + assert exc.value.code == code + + +@pytest.mark.parametrize( + ("body", "code"), + [ + ({"key": ISSUED_KEY, "scope": "connector-lite"}, "scope_mismatch"), + ({"scope": "api"}, "bad_response"), + ({"key": " "}, "bad_response"), + ], +) +def test_exchange_rejects_unusable_grants( + body: dict[str, Any], code: str, fake: FakeOrcaRouter +) -> None: + fake.exchange_status = 200 + fake.exchange_body = body + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.exchange_code("c1", "verifier") + assert exc.value.code == code + assert ISSUED_KEY not in str(exc.value) + + +def test_exchange_network_failure_hides_request_details(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(orcarouter.AUTH_BASE_ENV, "http://127.0.0.1:9") # discard port + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.exchange_code("c1", "secret-verifier") + assert exc.value.code == "unavailable" + assert "secret-verifier" not in str(exc.value) + + +# --- Stored sign-in and credential resolution -------------------------------- + + +def test_store_coexists_with_the_chatgpt_sign_in() -> None: + codex_record = {"type": "oauth", "access": "a", "refresh": "r", "account_id": "acct"} + codex.save_record(codex_record) + orcarouter.save_record(_stored_record()) + assert orcarouter.read_record() == _stored_record() + assert codex.read_record() == codex_record + assert oct(codex.AUTH_PATH.stat().st_mode & 0o777) == "0o600" + + assert orcarouter.logout() is True + assert orcarouter.read_record() is None + assert codex.read_record() == codex_record + assert orcarouter.logout() is False + + +def test_credential_resolution_order(monkeypatch: pytest.MonkeyPatch) -> None: + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.resolve_credential(None) + assert exc.value.code == "not_authenticated" + assert "strix auth login orcarouter" in str(exc.value) + assert orcarouter.API_KEY_ENV in str(exc.value) + + assert orcarouter.resolve_credential(LLM_KEY) == orcarouter.Credential(LLM_KEY, "llm_api_key") + orcarouter.save_record(_stored_record()) + assert orcarouter.resolve_credential(LLM_KEY) == orcarouter.Credential(ISSUED_KEY, "login") + monkeypatch.setenv(orcarouter.API_KEY_ENV, ENV_KEY) + assert orcarouter.resolve_credential(LLM_KEY) == orcarouter.Credential(ENV_KEY, "env") + + +def test_credential_repr_hides_the_key() -> None: + assert ENV_KEY not in repr(orcarouter.Credential(ENV_KEY, "env")) + assert ENV_KEY not in orcarouter.mask_key(ENV_KEY) + + +def test_revoked_sign_in_requires_reauth_but_keeps_the_key() -> None: + orcarouter.save_record(_stored_record()) + assert orcarouter.mark_needs_reauth(ISSUED_KEY) is True + assert orcarouter.read_record()["key"] == ISSUED_KEY # type: ignore[index] + with pytest.raises(orcarouter.OrcaRouterAuthError) as exc: + orcarouter.resolve_credential(None) + assert exc.value.code == "needs_reauth" + # An explicit API key still works while the sign-in is broken. + assert orcarouter.resolve_credential(LLM_KEY).source == "llm_api_key" + + +def test_stale_rejection_never_flags_a_newer_sign_in() -> None: + orcarouter.save_record(_stored_record("sk-orca-new-generation-9999")) + assert orcarouter.mark_needs_reauth(ISSUED_KEY) is False + assert not orcarouter.read_record().get("needs_reauth") # type: ignore[union-attr] + + +def test_unauthorized_marks_only_the_signed_in_key() -> None: + orcarouter.save_record(_stored_record()) + orcarouter.remember_active(orcarouter.Credential(ENV_KEY, "env")) + assert orcarouter.API_KEY_ENV in orcarouter.handle_unauthorized() + assert not orcarouter.read_record().get("needs_reauth") # type: ignore[union-attr] + + orcarouter.remember_active(orcarouter.Credential(ISSUED_KEY, "login")) + assert "strix auth login orcarouter" in orcarouter.handle_unauthorized() + assert orcarouter.read_record()["needs_reauth"] is True # type: ignore[index] + + +# --- Routing ----------------------------------------------------------------- + + +def test_orcarouter_models_route_through_the_openai_compatible_client( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv(orcarouter.API_KEY_ENV, ENV_KEY) + model = _unwrap(StrixProvider().get_model("orcarouter/openai/gpt-5.5")) + assert model.model == "openai/openai/gpt-5.5" # type: ignore[attr-defined] + assert model.base_url == "https://api.orcarouter.ai/v1" # type: ignore[attr-defined] + assert model.api_key == ENV_KEY # type: ignore[attr-defined] + + +def test_signed_in_key_is_used_over_a_stale_llm_api_key(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LLM_API_KEY", "sk-or-another-providers-key") + orcarouter.save_record(_stored_record()) + model = _unwrap(StrixProvider().get_model("orcarouter/deepseek/deepseek-v4-pro")) + assert model.api_key == ISSUED_KEY # type: ignore[attr-defined] + assert orcarouter._active == orcarouter.Credential(ISSUED_KEY, "login") + + +def test_api_key_in_llm_api_key_works(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LLM_API_KEY", LLM_KEY) + model = _unwrap(StrixProvider().get_model("orcarouter/orcarouter/auto")) + assert model.api_key == LLM_KEY # type: ignore[attr-defined] + assert model.model == "openai/orcarouter/auto" # type: ignore[attr-defined] + + +def test_dedupe_model_on_orcarouter_uses_its_own_key(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(orcarouter.API_KEY_ENV, ENV_KEY) + dedupe = DedupeSettings( + STRIX_DEDUPE_MODEL="orcarouter/deepseek/deepseek-v4-flash", + DEDUPE_LLM_API_KEY="sk-orca-dedupe-0004", + ) + model = _unwrap(resolve_dedupe_model(dedupe, "orcarouter/deepseek/deepseek-v4-flash")) + assert model.api_key == "sk-orca-dedupe-0004" # type: ignore[attr-defined] + assert model.base_url == "https://api.orcarouter.ai/v1" # type: ignore[attr-defined] + + +def test_other_prefixes_are_unchanged() -> None: + model = _unwrap(StrixProvider().get_model("openrouter/z-ai/glm-5.3")) + assert model.model == "openrouter/z-ai/glm-5.3" # type: ignore[attr-defined] + + +# --- Model catalog ----------------------------------------------------------- + + +def test_catalog_keeps_only_well_formed_chat_models(fake: FakeOrcaRouter) -> None: + assert orcarouter.fetch_models("sk-orca-catalog") == [ + "openai/gpt-5.5", + "deepseek/deepseek-v4-pro", + ] + _method, path, headers = fake.requests[-1] + assert path == "/v1/models" + assert headers["Authorization"] == "Bearer sk-orca-catalog" + + +def test_catalog_falls_back_to_the_seed(fake: FakeOrcaRouter) -> None: + assert orcarouter.available_models() == ( + ["openai/gpt-5.5", "deepseek/deepseek-v4-pro"], + True, + ) + fake.catalog = {"data": "nope"} + assert orcarouter.available_models() == (list(orcarouter.SEED_MODELS), False) + + +@pytest.mark.usefixtures("fake") +def test_oversized_catalog_is_discarded(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(orcarouter, "_CATALOG_MAX_BYTES", 64) + assert orcarouter.fetch_models() == [] + + +def test_unreachable_catalog_returns_nothing(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv(orcarouter.API_BASE_ENV, "http://127.0.0.1:9/v1") + assert orcarouter.fetch_models() == [] + + +def test_seed_keeps_the_verified_models() -> None: + assert orcarouter.SEED_MODELS == ( + "openai/gpt-5.5", + "anthropic/claude-opus-4.8", + "google/gemini-3.5-flash", + "deepseek/deepseek-v4-pro", + "orcarouter/auto", + ) + + +# --- `strix auth login orcarouter` ------------------------------------------- + + +def _browser_that_approves(fake: FakeOrcaRouter, seen: dict[str, Any], **overrides: str) -> Any: + def _open(url: str) -> bool: + params = fake.approve(url) + seen["params"] = params + query = {"code": "auth-code-1", "state": params["state"], **overrides} + callback = f"{params['callback_url']}?{urllib.parse.urlencode(query)}" + with urllib.request.urlopen(callback, timeout=5) as response: # noqa: S310 + seen["page"] = response.read().decode() + return True + + return _open + + +def test_login_via_loopback_redirect_stores_the_key( + fake: FakeOrcaRouter, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + seen: dict[str, Any] = {} + monkeypatch.setattr(auth_cli.webbrowser, "open", _browser_that_approves(fake, seen)) + + assert auth_cli.run_auth(["login", "orcarouter"]) == 0 + + callback = urllib.parse.urlparse(seen["params"]["callback_url"]) + assert (callback.scheme, callback.hostname, callback.path) == ("http", "127.0.0.1", "/callback") + assert seen["params"]["code_challenge_method"] == "S256" + assert "connected to OrcaRouter" in seen["page"] + assert orcarouter.read_record()["key"] == ISSUED_KEY # type: ignore[index] + exchange = next(body for method, path, body in fake.requests if method == "POST") + assert _b64url_sha256(exchange["code_verifier"]) == seen["params"]["code_challenge"] + + out = capsys.readouterr().out + assert "orcarouter/openai/gpt-5.5" in out + assert ISSUED_KEY not in out + assert exchange["code_verifier"] not in out + + +def test_login_denied_in_the_browser_stores_nothing( + fake: FakeOrcaRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + seen: dict[str, Any] = {} + browser = _browser_that_approves(fake, seen, code="", error="access_denied") + monkeypatch.setattr(auth_cli.webbrowser, "open", browser) + assert auth_cli.run_auth(["login", "orcarouter"]) == 1 + assert "Sign-in cancelled" in seen["page"] + assert orcarouter.read_record() is None + assert not [r for r in fake.requests if r[0] == "POST"] + + +def test_login_rejects_a_forged_state( + fake: FakeOrcaRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + seen: dict[str, Any] = {} + monkeypatch.setattr( + auth_cli.webbrowser, "open", _browser_that_approves(fake, seen, state="forged") + ) + assert auth_cli.run_auth(["login", "orcarouter"]) == 1 + assert orcarouter.read_record() is None + assert not [r for r in fake.requests if r[0] == "POST"] + + +def test_login_with_manual_out_of_band_code( + fake: FakeOrcaRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + seen: dict[str, Any] = {} + + def _open(url: str) -> bool: + seen["params"] = fake.approve(url, code="shown-code") + return True + + monkeypatch.setattr(auth_cli.webbrowser, "open", _open) + monkeypatch.setattr(auth_cli.Console, "input", lambda *_a, **_k: " shown-code \n") + + assert auth_cli.run_auth(["login", "orcarouter", "--manual"]) == 0 + assert seen["params"]["callback_url"] == "oob" + assert seen["params"]["code_challenge_method"] == "S256" + assert orcarouter.read_record()["key"] == ISSUED_KEY # type: ignore[index] + + +def test_login_timeout_falls_back_to_pasting( + fake: FakeOrcaRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(auth_cli, "_CALLBACK_TIMEOUT_S", 0.05) + monkeypatch.setattr(auth_cli.webbrowser, "open", lambda url: bool(fake.approve(url))) + monkeypatch.setattr(auth_cli.Console, "input", lambda *_a, **_k: "auth-code-1") + assert auth_cli.run_auth(["login", "orcarouter"]) == 0 + assert orcarouter.read_record() is not None + + +@pytest.mark.usefixtures("fake") +def test_login_with_an_expired_code_fails_cleanly(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(auth_cli.webbrowser, "open", lambda _url: True) + monkeypatch.setattr(auth_cli.Console, "input", lambda *_a, **_k: "never-issued") + assert auth_cli.run_auth(["login", "orcarouter", "--manual"]) == 1 + assert orcarouter.read_record() is None + + +def test_login_rate_limited(fake: FakeOrcaRouter, monkeypatch: pytest.MonkeyPatch) -> None: + fake.exchange_status = 429 + monkeypatch.setattr(auth_cli.webbrowser, "open", lambda _url: True) + monkeypatch.setattr(auth_cli.Console, "input", lambda *_a, **_k: "any") + assert auth_cli.run_auth(["login", "orcarouter", "--manual"]) == 1 + + +def test_login_replaces_a_revoked_sign_in( + fake: FakeOrcaRouter, monkeypatch: pytest.MonkeyPatch +) -> None: + orcarouter.save_record(_stored_record("sk-orca-revoked-0005", needs_reauth=True)) + seen: dict[str, Any] = {} + monkeypatch.setattr(auth_cli.webbrowser, "open", _browser_that_approves(fake, seen)) + assert auth_cli.run_auth(["login", "orcarouter"]) == 0 + assert orcarouter.resolve_credential(None) == orcarouter.Credential(ISSUED_KEY, "login") + + +# --- status / logout --------------------------------------------------------- + + +def test_status_reports_the_active_source_without_the_key( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + assert auth_cli.run_auth(["status", "orcarouter"]) == 1 + orcarouter.save_record(_stored_record()) + assert auth_cli.run_auth(["status", "orcarouter"]) == 0 + out = capsys.readouterr().out + assert "strix auth login orcarouter" in out + assert ISSUED_KEY not in out + + monkeypatch.setenv(orcarouter.API_KEY_ENV, ENV_KEY) + assert auth_cli.run_auth(["status", "orcarouter"]) == 0 + out = capsys.readouterr().out + assert orcarouter.API_KEY_ENV in out + assert ENV_KEY not in out + + +def test_bare_status_includes_orcarouter_when_signed_in() -> None: + assert auth_cli.run_auth(["status"]) == 1 + orcarouter.save_record(_stored_record()) + assert auth_cli.run_auth(["status"]) == 0 + + +def test_logout_orcarouter_leaves_chatgpt_alone() -> None: + codex_record = {"type": "oauth", "access": "a", "refresh": "r", "account_id": "acct"} + codex.save_record(codex_record) + orcarouter.save_record(_stored_record()) + assert auth_cli.run_auth(["logout", "orcarouter"]) == 0 + assert orcarouter.read_record() is None + assert codex.read_record() == codex_record + assert auth_cli.run_auth(["logout", "gemini"]) == 2 + + +# --- Startup checks and error hints ------------------------------------------ + + +def test_environment_requires_an_orcarouter_credential( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + monkeypatch.setenv("STRIX_LLM", "orcarouter/openai/gpt-5.5") + monkeypatch.setattr(environment, "report_error", lambda *_a, **_k: None) + with pytest.raises(SystemExit): + environment.validate_environment() + assert "strix auth login orcarouter" in capsys.readouterr().out + + +def test_401_hint_flags_the_signed_in_key_for_reauth() -> None: + orcarouter.save_record(_stored_record()) + orcarouter.remember_active(orcarouter.Credential(ISSUED_KEY, "login")) + exc = RuntimeError("litellm.AuthenticationError: Error code: 401 - Invalid API key") + hint = _orcarouter_error_hint(exc, "orcarouter/openai/gpt-5.5") + assert hint is not None + assert "strix auth login orcarouter" in hint + assert orcarouter.read_record()["needs_reauth"] is True # type: ignore[index] + + +def test_error_hint_ignores_other_providers() -> None: + exc = RuntimeError("Error code: 401") + assert _orcarouter_error_hint(exc, "openrouter/z-ai/glm-5.3") is None + + +def test_usage_lists_both_providers_verbatim(capsys: pytest.CaptureFixture[str]) -> None: + assert auth_cli.run_auth(["--help"]) == 0 + out = capsys.readouterr().out + assert "strix auth login orcarouter [--manual]" in out + assert "strix auth status [chatgpt|orcarouter]" in out