scriptoriumadmin/strix
1
0
Fork 0
mirror of https://github.com/usestrix/strix.git synced 2026-11-14 09:01:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge a0a5b21877 into cb581bf93d

Browse source
This commit is contained in:
armyluki-wq 2026-11-04 15:44:04 +00:00 • committed by GitHub
commit ae9c4e5fc5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 1557 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
README.md
View file

@ -319,6 +319,15 @@ export STRIX_LLM="chatgpt/gpt-5.4" # chatgpt/<model> 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:

7
docs/advanced/configuration.mdx
View file

@ -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).
</ParamField>
<ParamField path="ORCAROUTER_API_KEY" type="string">
API key for `orcarouter/<model>` 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).
</ParamField>
<ParamField path="LLM_API_BASE" type="string">
Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
</ParamField>

1
docs/docs.json
View file

@ -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",

62
docs/llm-providers/orcarouter.mdx Normal file
View file

@ -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/<model id>` 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/<model id>` 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/<model id>`.
## 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://`.

3
docs/llm-providers/overview.mdx
View file

@ -60,6 +60,9 @@ See the [Local Models guide](/llm-providers/local) for setup instructions and re
<Card title="OpenRouter" href="/llm-providers/openrouter">
Access 100+ models through a single API.
</Card>
<Card title="OrcaRouter Cyber Inference" href="/llm-providers/orcarouter">
Scoped, authorized models for security testing. API key or browser sign-in.
</Card>
<Card title="Vercel AI Gateway" href="/llm-providers/vercel-ai-gateway">
Access 100+ models through one OpenAI-compatible endpoint.
</Card>

24
strix/config/models.py
View file

@ -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,
)

469
strix/config/orcarouter.py Normal file
View file

@ -0,0 +1,469 @@
"""OrcaRouter: API-key and OAuth 2.0 + PKCE sign-in, and the ``orcarouter/`` route.
`OrcaRouter <https://www.orcarouter.ai>`_ is an OpenAI-compatible gateway. A
``STRIX_LLM`` of ``orcarouter/<model id>`` (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/<model>`` 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",
]

308
strix/interface/auth_cli.py
View file

@ -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/<model>`` STRIX_LLM runs on the
subscription.
subscription; an ``orcarouter/<model>`` 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'<img class="logo" src="data:image/png;base64,{encoded}" alt="" />'
def _render_callback_html() -> str:
return _CALLBACK_HTML.replace("<!--LOGO-->", _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-->", _logo_img_tag())
.replace("<!--HEADING-->", heading)
.replace("<!--MESSAGE-->", message)
)
_CALLBACK_HTML = """<!doctype html>
@ -396,11 +671,10 @@ _CALLBACK_HTML = """<!doctype html>
<span>Strix</span>
</a>
<div class="brand">Strix</div>
<h1>You're signed in</h1>
<h1><!--HEADING--></h1>
<main class="card">
<div class="badge">✓</div>
<p class="msg">Strix is connected to your ChatGPT subscription. Head back to your
terminal — your security test runs there.</p>
<p class="msg"><!--MESSAGE--></p>
<div class="rule"></div>
<p class="tagline">Autonomous AI hackers that <b>find and fix</b> your app's
vulnerabilities.</p>

10
strix/interface/environment.py
View file

@ -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")

26
strix/interface/main.py
View file

@ -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:

663
tests/test_orcarouter.py Normal file
View file

@ -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