diff --git a/README.md b/README.md index 215785e..d33be42 100644 --- a/README.md +++ b/README.md @@ -70,6 +70,12 @@ Each component is organized into a self-contained subtree: * Python 3.12+ (Required only for the runtime agent and the benchmarking harness). * *Responsibility Check:* Ensure you are only scanning code bases you are explicitly authorized to analyze. +You supply your own model access. The runtime agent supports the direct Anthropic API, +AWS Bedrock (OAuth proxy or SigV4), and the [OrcaRouter](https://www.orcarouter.ai) +gateway — `anthropic.auth_mode = "orcarouter"` with either a pasted `sk-orca-…` API key +or an OAuth 2.0 + PKCE login. See +[`vulnhunter-agent/README.md`](vulnhunter-agent/README.md#orcarouter-auth_mode--orcarouter). + ### Installation ```bash diff --git a/harness/README.md b/harness/README.md index 72a2258..613c245 100644 --- a/harness/README.md +++ b/harness/README.md @@ -19,6 +19,25 @@ has no other runtime dependencies. Set the scanning/judging model once in `local_harness/config.py` (the `MODEL` constant) to switch models across every workflow. +### Routing at OrcaRouter + +[OrcaRouter](https://www.orcarouter.ai) is an OpenAI-compatible AI gateway that +routes many providers behind one endpoint. To send the harness's `claude` +subprocesses through it, export the same `ORCA_*` variables the runtime agent +honors and the harness injects them into each `claude` invocation: + +```bash +export ORCA_API_KEY=sk-orca-... # or ORCA_BASE_URL for self-hosted +export ORCA_API_BASE_URL=https://api.orcarouter.ai/v1 +python -m local_harness.batch.run scan +``` + +There is no `ANTHROPIC_BASE_URL`/`ANTHROPIC_API_KEY` for you to set by hand and +no OAuth flow here: the harness is a credential *consumer*. Obtain the key with +`python -m agent --mode=orcarouter login` (or `--orcarouter-action=store-api-key`) +in `vulnhunter-agent/`, or export `ORCA_API_KEY` directly. With no `ORCA_*` +variable set, the harness inherits the ambient environment exactly as before. + ## What's inside | Path | Purpose | diff --git a/harness/local_harness/benchmark/analyze_misses.py b/harness/local_harness/benchmark/analyze_misses.py index 6c77bbf..a027506 100644 --- a/harness/local_harness/benchmark/analyze_misses.py +++ b/harness/local_harness/benchmark/analyze_misses.py @@ -27,6 +27,7 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..")) from local_harness.config import BENCHMARK_DIR, MODEL, REPO_ROOT, RESULTS_DIR, STATE_FILE, atomic_write_json +from local_harness.orcarouter_env import claude_env PHASE_TO_PROMPT = { "phase1": ["skill/phases/phase1_recon.md"], @@ -277,6 +278,7 @@ def invoke_diagnostic(finding, phase_key, evidence, results_dir, repo_dir): try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=1200, + env=claude_env(), ) except subprocess.TimeoutExpired: elapsed = time.time() - start diff --git a/harness/local_harness/benchmark/judge.py b/harness/local_harness/benchmark/judge.py index 6d8285d..514f5b1 100644 --- a/harness/local_harness/benchmark/judge.py +++ b/harness/local_harness/benchmark/judge.py @@ -13,6 +13,7 @@ JUDGE_TIMEOUT, MODEL, ) +from local_harness.orcarouter_env import claude_env JUDGE_SYSTEM_PROMPT = """You are a security finding comparator. Your job is to determine whether a vulnerability scanner's output detected specific known vulnerabilities. @@ -95,6 +96,7 @@ def judge_findings_batch(results_report, findings, model=None): "--model", model, "--system-prompt", JUDGE_SYSTEM_PROMPT], capture_output=True, text=True, timeout=JUDGE_TIMEOUT, + env=claude_env(), ) except subprocess.TimeoutExpired: return [{"finding_id": f["finding_id"], "detected": None, diff --git a/harness/local_harness/orcarouter_env.py b/harness/local_harness/orcarouter_env.py new file mode 100644 index 0000000..2e2c926 --- /dev/null +++ b/harness/local_harness/orcarouter_env.py @@ -0,0 +1,93 @@ +"""OrcaRouter environment bridge for the harness's ``claude`` subprocesses. + +The harness drives the ``claude`` CLI directly and relies on the CLI's own +ambient authentication. To route those invocations at OrcaRouter we inject +the inference origin and an API key into the subprocess environment. + +The harness is a dev tool with no config file, so it reads the same ``ORCA_*`` +environment variables the agent honors. If a key is not supplied via env, it +tries the agent's persisted credential store (an optional dependency here — +imported only when present). Nothing is set unless OrcaRouter is actually +configured, so the harness's default behavior is byte-identical. + +There is no OAuth/PKCE here and none should be added: the harness is not a +credential-acquisition surface, it is a consumer of a credential obtained +(either by exporting env or by ``python -m agent --mode=orcarouter login``). +""" + +import os + +# The CLI appends its version segment to ANTHROPIC_BASE_URL itself, so this +# must be the inference origin WITHOUT a trailing /v1. +_BASE_URL_ENV = "ANTHROPIC_BASE_URL" +_API_KEY_ENV = "ANTHROPIC_API_KEY" + +# Env vars the agent also honors, in priority order. +_KEY_ENV_NAMES = ("ORCA_API_KEY", "ORCAROUTER_API_KEY") +_BASE_ENV_NAMES = ("ORCA_API_BASE_URL", "ORCA_BASE_URL") +_DEFAULT_API_BASE = "https://api.orcarouter.ai/v1" + + +def _configured() -> bool: + return any(os.environ.get(name) for name in (*_KEY_ENV_NAMES, *_BASE_ENV_NAMES)) + + +def _api_base() -> str: + for name in _BASE_ENV_NAMES: + value = os.environ.get(name) + if value and value.strip(): + return value.strip() + return _DEFAULT_API_BASE + + +def _cli_base_url(api_base: str) -> str: + """Strip a trailing ``/v1`` (the CLI appends its own version segment).""" + base = api_base.rstrip("/") + return base[:-3] if base.endswith("/v1") else base + + +def _api_key() -> str: + for name in _KEY_ENV_NAMES: + value = os.environ.get(name) + if value and value.strip(): + return value.strip() + # Fall back to the agent's persisted credential store (whatever the user + # established via API key or OAuth login). Optional: the agent package is + # a sibling subtree that may not be installed in a harness-only checkout. + try: + from agent.config import load_orcarouter_config + from agent.orcarouter import ConnectCredentialProvider + except ImportError: + return "" + try: + config = load_orcarouter_config() + return ConnectCredentialProvider(config).resolve().api_key + except Exception: + return "" + + +def orcarouter_env() -> dict: + """Return the CLI env overrides for OrcaRouter, or an empty dict.""" + if not _configured(): + return {} + key = _api_key() + if not key: + raise RuntimeError( + "OrcaRouter is configured (ORCA_* env set) but no API key is " + "available. Export ORCA_API_KEY, or run " + "`python -m agent --mode=orcarouter login` first." + ) + return {_BASE_URL_ENV: _cli_base_url(_api_base()), _API_KEY_ENV: key} + + +def claude_env() -> dict | None: + """Environment for a ``claude`` subprocess, or ``None`` to inherit. + + ``None`` (the default when OrcaRouter is not configured) tells the + ``subprocess`` call to inherit the parent environment unchanged, so the + harness's existing behavior is preserved exactly. + """ + overrides = orcarouter_env() + if not overrides: + return None + return {**os.environ, **overrides} diff --git a/harness/local_harness/scan.py b/harness/local_harness/scan.py index 5a677d5..19ea2d2 100644 --- a/harness/local_harness/scan.py +++ b/harness/local_harness/scan.py @@ -21,6 +21,7 @@ SCAN_TIMEOUT, SKILLS_DIR, ) +from .orcarouter_env import claude_env # The result of scanning one folder. A namedtuple (rather than a bare 7-tuple) # so consumers can use attribute access and a field reorder can't silently @@ -221,6 +222,9 @@ def scan_folder(folder_path, log_file=None, readonly=False): stderr=subprocess.STDOUT, text=True, cwd=folder_path, + # Route at OrcaRouter when configured; None inherits the parent env + # unchanged, so the default behavior is untouched. + env=claude_env(), ) event_count = 0 diff --git a/harness/tests/conftest.py b/harness/tests/conftest.py new file mode 100644 index 0000000..fb4accc --- /dev/null +++ b/harness/tests/conftest.py @@ -0,0 +1,19 @@ +"""Shared pytest fixtures for the harness test suite. + +Keeps OrcaRouter provider configuration out of the test process so a +credential present in the developer/CI environment can never change a +test's outcome (the harness env bridge keys off ``ORCA_*``). +""" + +from __future__ import annotations + +import os + +import pytest + + +@pytest.fixture(autouse=True) +def _clean_orcarouter_env(monkeypatch: pytest.MonkeyPatch) -> None: + for key in list(os.environ): + if key.startswith(("ORCA_", "ORCAROUTER_")): + monkeypatch.delenv(key, raising=False) diff --git a/harness/tests/test_orcarouter_env.py b/harness/tests/test_orcarouter_env.py new file mode 100644 index 0000000..b665a32 --- /dev/null +++ b/harness/tests/test_orcarouter_env.py @@ -0,0 +1,155 @@ +"""Tests for the harness OrcaRouter env bridge. + +The harness drives the ``claude`` CLI directly. When OrcaRouter is +configured, the bridge injects the inference origin and an API key into the +subprocess environment; otherwise it must be a no-op so existing behavior is +unchanged. The harness acquires no credentials itself. +""" + +import importlib + +import pytest + +import local_harness.orcarouter_env as bridge + + +ORCA_ENV_KEYS = ( + "ORCA_BASE_URL", "ORCA_API_BASE_URL", "ORCA_AUTH_BASE_URL", + "ORCA_API_KEY", "ORCAROUTER_API_KEY", +) + + +@pytest.fixture(autouse=True) +def _clean_orca_env(monkeypatch): + for key in ORCA_ENV_KEYS: + monkeypatch.delenv(key, raising=False) + + +def test_noop_when_orcarouter_not_configured(): + assert bridge.orcarouter_env() == {} + assert bridge.claude_env() is None + + +def test_injects_api_origin_and_key(monkeypatch): + monkeypatch.setenv("ORCA_API_KEY", "sk-orca-HARNESS000000000000000") + monkeypatch.setenv("ORCA_API_BASE_URL", "https://api.orcarouter.ai/v1") + env = bridge.orcarouter_env() + assert env["ANTHROPIC_BASE_URL"] == "https://api.orcarouter.ai" + assert env["ANTHROPIC_API_KEY"] == "sk-orca-HARNESS000000000000000" + # Never the /v1 form — the CLI appends the version segment itself. + assert not env["ANTHROPIC_BASE_URL"].endswith("/v1") + + +def test_shared_base_fallback_used(monkeypatch): + monkeypatch.setenv("ORCA_API_KEY", "sk-orca-x") + monkeypatch.setenv("ORCA_BASE_URL", "https://orca.internal.example/v1") + assert bridge.orcarouter_env()["ANTHROPIC_BASE_URL"] == "https://orca.internal.example" + + +def test_orcarouter_api_key_env_is_honored(monkeypatch): + monkeypatch.setenv("ORCAROUTER_API_KEY", "sk-orca-fromcampaign") + monkeypatch.setenv("ORCA_BASE_URL", "https://api.orcarouter.ai") + assert bridge.orcarouter_env()["ANTHROPIC_API_KEY"] == "sk-orca-fromcampaign" + + +def test_missing_key_raises_actionable_error(monkeypatch): + # Configured (base set) but no key available anywhere. + monkeypatch.setenv("ORCA_BASE_URL", "https://api.orcarouter.ai") + monkeypatch.setattr(bridge, "_api_key", lambda: "") + with pytest.raises(RuntimeError, match="ORCA_API_KEY"): + bridge.orcarouter_env() + + +def test_claude_env_merges_parent_environment(monkeypatch): + monkeypatch.setenv("ORCA_API_KEY", "sk-orca-merge") + monkeypatch.setenv("ORCA_BASE_URL", "https://api.orcarouter.ai") + monkeypatch.setenv("SOME_UNRELATED_VAR", "kept") + env = bridge.claude_env() + assert env["SOME_UNRELATED_VAR"] == "kept" + assert env["ANTHROPIC_API_KEY"] == "sk-orca-merge" + + +def test_scan_subprocess_receives_env(monkeypatch, tmp_path): + """The scan Popen call is handed the injected env.""" + import json + + import local_harness.scan as scan + + monkeypatch.setenv("ORCA_API_KEY", "sk-orca-scan") + monkeypatch.setenv("ORCA_BASE_URL", "https://api.orcarouter.ai") + + captured = {} + + class _FakePopen: + def __init__(self, *a, **k): + captured.update(k) + self.stdout = iter([json.dumps({"type": "result"}) + "\n"]) + self.returncode = 0 + + def wait(self): + return 0 + + def kill(self): + pass + + class _NoTimer: + def __init__(self, *a, **k): + pass + + def start(self): + pass + + def cancel(self): + pass + + folder = tmp_path / "repo" + folder.mkdir() + monkeypatch.setattr(scan, "SKILLS_DIR", str(tmp_path / "skills")) + (tmp_path / "skills").mkdir() + monkeypatch.setattr(scan.subprocess, "Popen", _FakePopen) + monkeypatch.setattr(scan.threading, "Timer", _NoTimer) + + scan.scan_folder(str(folder)) + assert captured["env"]["ANTHROPIC_BASE_URL"] == "https://api.orcarouter.ai" + assert captured["env"]["ANTHROPIC_API_KEY"] == "sk-orca-scan" + + +def test_scan_subprocess_inherits_when_unconfigured(monkeypatch, tmp_path): + """Default behavior: env is None so the subprocess inherits unchanged.""" + import json + + import local_harness.scan as scan + + captured = {} + + class _FakePopen: + def __init__(self, *a, **k): + captured.update(k) + self.stdout = iter([json.dumps({"type": "result"}) + "\n"]) + self.returncode = 0 + + def wait(self): + return 0 + + def kill(self): + pass + + class _NoTimer: + def __init__(self, *a, **k): + pass + + def start(self): + pass + + def cancel(self): + pass + + folder = tmp_path / "repo" + folder.mkdir() + monkeypatch.setattr(scan, "SKILLS_DIR", str(tmp_path / "skills")) + (tmp_path / "skills").mkdir() + monkeypatch.setattr(scan.subprocess, "Popen", _FakePopen) + monkeypatch.setattr(scan.threading, "Timer", _NoTimer) + + scan.scan_folder(str(folder)) + assert captured["env"] is None diff --git a/vulnhunter-agent/README.md b/vulnhunter-agent/README.md index f0dccb4..223676e 100644 --- a/vulnhunter-agent/README.md +++ b/vulnhunter-agent/README.md @@ -66,6 +66,7 @@ Settings load from a TOML file (`--config`, then `$VULNHUNT_AGENT_CONFIG`, then | `api_key` *(default)* | Direct Anthropic API | `[anthropic].api_key` or the standard `ANTHROPIC_API_KEY` env var | | `bedrock_oauth` | Routes through an AWS Bedrock proxy fronted by an OAuth2 client-credentials token endpoint | `[anthropic].bedrock_base_url` + the `[oauth]` block (`token_endpoint`, `client_id`, `client_secret`) | | `bedrock_sigv4` | Calls Amazon Bedrock directly with SigV4 request signing via the standard AWS credential chain — no proxy, no bearer token | `[anthropic].aws_region`; optionally `aws_profile` (named profile) and `bedrock_base_url` (VPC/custom endpoint). No `[oauth]` block. | +| `orcarouter` | Routes the bundled Claude Code CLI at [OrcaRouter](https://www.orcarouter.ai), an OpenAI-compatible AI gateway. The credential is an OrcaRouter API key (`sk-orca-…`) | The `[orcarouter]` block; a credential from either connect entry below | `bedrock_oauth` exists for environments that front Claude with a Bedrock proxy and mint short-lived bearer tokens. `bedrock_sigv4` is for AWS-native setups that call Bedrock @@ -73,6 +74,46 @@ directly (use a cross-region inference-profile model ID, e.g. `us.anthropic.clau credentials resolve from the usual AWS chain — env vars, shared config/credentials file, SSO, or an instance/task role. Most users want the default `api_key` mode. +### OrcaRouter (`auth_mode = "orcarouter"`) + +OrcaRouter is an OpenAI-compatible AI gateway that routes many providers behind one +endpoint. Set `anthropic.auth_mode = "orcarouter"` and choose a model by its gateway +ID (e.g. `deepseek/deepseek-v4-pro`). There are **two independent authentication +entries**, and both produce the same kind of OrcaRouter API key: + +```bash +# Option 1 — paste an existing key (sk-orca-…) +python -m agent --mode=orcarouter --orcarouter-action=store-api-key + +# Option 2 — sign in with an OrcaRouter account (OAuth 2.0 + PKCE) +python -m agent --mode=orcarouter --orcarouter-action=login + +python -m agent --mode=orcarouter --orcarouter-action=status # what is stored +python -m agent --mode=orcarouter --orcarouter-action=clear # remove it +python -m agent --mode=orcarouter --orcarouter-action=models --capability chat +``` + +The OAuth login (`[orcarouter].flow`) supports Flow A — a loopback redirect, chosen +because the agent runs on a machine where `127.0.0.1:` is bindable — and Flow B — +an out-of-band code, for hosts that cannot listen (locked-down or remote machines). +S256 PKCE is used in both; there is no client secret. The credential is stored 0o600 in +`~/.vulnhunter/orcarouter.json` and **reused across restarts** — it is a durable key, not +a refresh token, so the agent never re-authorizes on every launch (OrcaRouter caps +PKCE key issuance at 10 per user per 24 hours) and never attempts a fake refresh. If the +relay rejects the key (401/403), the exact stored generation is marked for +reauthentication; run `login` or `store-api-key` again to replace it. + +**Origins.** Authentication and inference use different origins: +`https://www.orcarouter.ai` (authorize `/auth`, exchange `/api/v1/auth/keys`) and +`https://api.orcarouter.ai/v1` (inference and the model catalog). Point a self-hosted +deployment at its own origin with `[orcarouter].base_url` (shared) or the explicit +`ORCA_AUTH_BASE_URL` / `ORCA_API_BASE_URL` overrides (env wins over TOML). + +**Models.** The model list is discovered live from `GET {api_base}/models` and filtered +per capability. A small verified seed is used only when discovery fails, and the result +is reported as degraded rather than falling back to free text. The `harness/` tools +share the same `claude` CLI and honor the same `ORCA_*` environment variables. + ### Other sections (abridged) - `[github]` — `scan_token` (clone + issues) and `reports_token` (publish), injected into diff --git a/vulnhunter-agent/agent/__main__.py b/vulnhunter-agent/agent/__main__.py index 1b80387..b1c68ce 100644 --- a/vulnhunter-agent/agent/__main__.py +++ b/vulnhunter-agent/agent/__main__.py @@ -42,7 +42,9 @@ import argparse import asyncio import dataclasses +import json import logging +import os import shutil import subprocess import sys @@ -97,21 +99,26 @@ def _build_parser() -> argparse.ArgumentParser: # deliberate breaking change from the pre-verify CLI. required=False, default=None, - choices=("scan", "verify"), + choices=("scan", "verify", "orcarouter"), help=( "Required. 'scan' runs the existing scanner against a repo URL. " - "'verify' runs the fix-verify agent against one or more issue URLs." + "'verify' runs the fix-verify agent against one or more issue URLs. " + "'orcarouter' manages the OrcaRouter credential (status/login/" + "store-api-key/clear) and lists the live model catalog; use " + "--orcarouter-action to choose." ), ) # In scan mode this is the repo URL (exactly one). In verify mode this - # is one or more issue URLs. We use nargs="+" and validate the count - # in main() once we know which mode is in effect. + # is one or more issue URLs. orcarouter mode takes no positional. We use + # nargs="*" and validate the count in main() once we know which mode is + # in effect. parser.add_argument( "targets", - nargs="+", + nargs="*", help=( "Positional argument(s). In --mode=scan: exactly one git repo URL. " - "In --mode=verify: one or more full GitHub issue URLs." + "In --mode=verify: one or more full GitHub issue URLs. In " + "--mode=orcarouter: none." ), ) parser.add_argument( @@ -271,6 +278,36 @@ def _build_parser() -> argparse.ArgumentParser: help="Post comments but don't reopen issues on non-FIXED verdicts.", ) + # ---- orcarouter-mode flags ------------------------------------------ + + orca_section = parser.add_argument_group("orcarouter-mode options") + orca_section.add_argument( + "--api-key", + default=None, + help="OrcaRouter API key (sk-orca-...) for --mode=orcarouter " + "--orcarouter-action=store-api-key. Prefer the ORCA_API_KEY env var or " + "an interactive prompt: a literal flag value is visible in the process " + "list.", + ) + orca_section.add_argument( + "--orcarouter-action", + default="status", + choices=("status", "login", "store-api-key", "clear", "models"), + help="What --mode=orcarouter should do (default: status).", + ) + orca_section.add_argument( + "--capability", + default="chat", + choices=("chat", "image_input", "audio_input", "video_input", "embedding", + "image_generation", "video_generation", "rerank"), + help="Capability filter for --orcarouter-action=models (default: chat).", + ) + orca_section.add_argument( + "--json", + action="store_true", + help="Emit machine-readable JSON for --mode=orcarouter.", + ) + # ---- shared logging -------------------------------------------------- parser.add_argument( @@ -1216,13 +1253,182 @@ def _repo_slug_for_audit( +def _run_orcarouter_command(args) -> int: + """``--mode=orcarouter`` — manage the credential and inspect the catalog. + + Two explicit, discoverable authentication entries (``store-api-key`` for a + pasted key and ``login`` for OAuth 2.0 + PKCE) plus ``status``/``clear`` + so both choices can be used and reversed, and ``models`` to inspect the + live capability-filtered catalog without running a scan. + """ + from .config import load_orcarouter_config + from .orcarouter import ( + ConnectCredentialProvider, + ModelCapability, + discover_catalog, + ) + + config = load_orcarouter_config(args.config) + provider = ConnectCredentialProvider(config) + + action = args.orcarouter_action + + if action == "store-api-key": + key = (args.api_key or "").strip() + if not key: + key = os.environ.get("ORCA_API_KEY", "") or os.environ.get( + "ORCAROUTER_API_KEY", "" + ) + if not key: + # Interactive prompt keeps the key out of the process list/argv. + import getpass + + key = getpass.getpass("OrcaRouter API key (sk-orca-...): ").strip() + result = provider.store_api_key(key) + _emit_orca( + args, + {"status": "stored", "method": "api_key", "masked_key": _mask(result.api_key)}, + f"Stored OrcaRouter API key {_mask(result.api_key)}.", + ) + return 0 + + if action == "login": + result = provider.connect_oauth() + _emit_orca( + args, + {"status": "connected", "method": "oauth", "scope": result.scope, + "masked_key": _mask(result.api_key)}, + f"OrcaRouter login complete. Stored key {_mask(result.api_key)} " + f"(granted scope: {result.scope}).", + ) + return 0 + + if action == "clear": + provider.clear() + _emit_orca(args, {"status": "cleared"}, "Removed the stored OrcaRouter credential.") + return 0 + + if action == "models": + capability = ModelCapability(args.capability) + try: + api_key = provider.resolve().api_key + except Exception: # noqa: BLE001 - catalog is public; list without auth + api_key = "" + catalog = discover_catalog(config, api_key, capability=args.capability) + models = catalog.for_capability(capability) + if args.json: + print( + json.dumps( + { + "source": catalog.source, + "degraded": catalog.degraded, + "error": catalog.error, + "capability": capability.value, + "count": len(models), + "models": [ + { + "id": m.id, + "context_length": m.context_length, + "input_modalities": list(m.input_modalities), + "endpoint_types": list(m.endpoint_types), + "source": m.source, + } + for m in models + ], + }, + indent=2, + ) + ) + return 0 + note = f" (DEGRADED: {catalog.error})" if catalog.degraded else "" + print( + f"OrcaRouter catalog [{catalog.source}]{note} — " + f"{len(models)} model(s) for capability '{capability.value}':" + ) + for m in models: + modalities = ",".join(m.input_modalities) or "?" + print(f" {m.id} [input: {modalities}]") + + # If the caller named a model (--model), verify it is still in the + # filtered set. An incompatible selection must not be silently kept: + # we say so and fail, so it is re-chosen rather than sent blind. + selected = (args.model or "").strip() + if selected and not any(m.id == selected for m in models): + print( + f"ERROR: the selected model '{selected}' is not available for " + f"capability '{capability.value}'" + + (f" ({catalog.error})" if catalog.degraded else "") + + ". Choose a model from the list above.", + file=sys.stderr, + ) + return 1 + return 0 + + # status (default) + try: + result = provider.resolve() + except Exception as exc: # noqa: BLE001 - report, don't crash on status + _emit_orca( + args, + {"status": "no-credential", "reason": str(exc)}, + f"No usable OrcaRouter credential: {exc}", + ) + return 0 + _emit_orca( + args, + { + "status": "ok", + "method": result.method.value, + "source": result.source, + "scope": result.scope, + "generation": result.generation, + "masked_key": _mask(result.api_key), + "needs_reauth": result.needs_reauth, + }, + f"OrcaRouter credential: {_mask(result.api_key)} " + f"(method: {result.method.value}, source: {result.source}, " + f"generation: {result.generation}).", + ) + return 0 + + +def _mask(key: str) -> str: + """Mask a credential for display (never the full secret).""" + if not key: + return "" + if len(key) <= 8: + return "***" + return f"{key[:8]}...{key[-4:]}" + + +def _emit_orca(args, payload: dict, human: str) -> None: + if args.json: + print(json.dumps(payload, indent=2)) + else: + print(human) + + def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) # ---- --mode is required (friendlier than argparse's default) --------- if args.mode is None: - parser.error("--mode is required; choose scan or verify") + parser.error("--mode is required; choose scan, verify, or orcarouter") + + if args.mode == "orcarouter": + if args.targets: + parser.error("--mode=orcarouter takes no positional arguments.") + _configure_logging(args.log_level, args.verbose) + set_verbosity(args.verbose) + try: + return _run_orcarouter_command(args) + except KeyboardInterrupt: + logging.warning("Interrupted by user") + return 130 + except Exception as exc: # noqa: BLE001 + logging.error("%s", exc) + return 1 # ---- Mode-specific positional + flag validation ----------------------- if args.mode == "scan": diff --git a/vulnhunter-agent/agent/_url.py b/vulnhunter-agent/agent/_url.py index 9c105a9..2f57917 100644 --- a/vulnhunter-agent/agent/_url.py +++ b/vulnhunter-agent/agent/_url.py @@ -21,7 +21,7 @@ # Raw token prefixes GitHub / Anthropic emit. Prefix is preserved so an # operator can still tell what kind of token leaked; the secret body is masked. _RAW_TOKEN_RE = re.compile( - r"(ghp_|gho_|ghu_|ghs_|ghr_|github_pat_|sk-ant-)[A-Za-z0-9_-]+" + r"(ghp_|gho_|ghu_|ghs_|ghr_|github_pat_|sk-ant-|sk-orca-)[A-Za-z0-9_-]+" ) _FORM_SECRET_RE = re.compile( r"(?i)(client[_-]?secret[\"']?[ \t]*(?:\[\d+\])?[ \t]*[=:][ \t]*\[?[ \t]*)" @@ -39,8 +39,9 @@ def redact(text: str) -> str: 2. ``Authorization: Bearer/token `` header values. 3. ``?access_token=`` / ``&token=`` query parameters. 4. Known raw token prefixes (``ghp_``, ``gho_``, ``ghu_``, ``ghs_``, - ``ghr_``, ``github_pat_``, ``sk-ant-``) — the prefix is preserved so - the token kind stays identifiable while the secret body is masked. + ``ghr_``, ``github_pat_``, ``sk-ant-``, ``sk-orca-``) — the prefix is + preserved so the token kind stays identifiable while the secret body + is masked. 5. ``client_secret`` values in form-encoded, JSON, and config formats (``client_secret``, ``client-secret``, ``clientSecret``). """ diff --git a/vulnhunter-agent/agent/auth.py b/vulnhunter-agent/agent/auth.py index aff2d8a..be6a336 100644 --- a/vulnhunter-agent/agent/auth.py +++ b/vulnhunter-agent/agent/auth.py @@ -15,6 +15,11 @@ SSO, container/instance role). ``get_valid_token()`` returns an empty string so the shared call-site interface is preserved. +- ``ConnectCredentialProvider`` — the OrcaRouter provider (``auth_mode = + "orcarouter"``). It is not defined here; both of its credential adapters + (pasted API key and OAuth 2.0 + PKCE login) live in ``agent.orcarouter`` + and return a plain OrcaRouter API key through ``get_valid_token()``. + ``make_token_manager(config)`` returns the right one for the configured ``anthropic.auth_mode``. @@ -150,15 +155,24 @@ def _refresh(self) -> str: def make_token_manager( config: AgentConfig, name: str = "vulnhunter" -) -> ApiKeyTokenManager | OAuthTokenManager | SigV4TokenManager: +) -> ApiKeyTokenManager | OAuthTokenManager | SigV4TokenManager | ConnectCredentialProvider: """Return the Anthropic token provider for the configured auth mode. All providers expose ``get_valid_token()``, so callers can use the - result without caring whether auth is API-key, Bedrock/OAuth, or - Bedrock/SigV4. + result without caring whether auth is API-key, Bedrock/OAuth, + Bedrock/SigV4, or the OrcaRouter gateway. In the OrcaRouter case the + return value is a :class:`ConnectCredentialProvider`, which resolves the + credential through whichever of its two adapters the user configured. """ if config.anthropic.auth_mode == "bedrock_oauth": return OAuthTokenManager(config.oauth, config.tls, name=name) if config.anthropic.auth_mode == "bedrock_sigv4": return SigV4TokenManager(name=name) + if config.anthropic.auth_mode == "orcarouter": + # Imported lazily so the OrcaRouter package is only loaded when it is + # actually selected, and to avoid an import cycle (agent.orcarouter + # imports .config and ._url). + from .orcarouter import ConnectCredentialProvider + + return ConnectCredentialProvider(config.orcarouter, tls=config.tls) return ApiKeyTokenManager(config.anthropic.api_key, name=name) diff --git a/vulnhunter-agent/agent/build_settings.py b/vulnhunter-agent/agent/build_settings.py index 081c277..4155dbf 100644 --- a/vulnhunter-agent/agent/build_settings.py +++ b/vulnhunter-agent/agent/build_settings.py @@ -1,11 +1,12 @@ """Build the Claude Code settings JSON the SDK passes through to the CLI. -Routes Anthropic calls one of three ways: directly to the Anthropic API +Routes Anthropic calls one of four ways: directly to the Anthropic API (``api_key`` mode), through an AWS Bedrock proxy fronted by an OAuth bearer -token (``bedrock_oauth`` mode), or directly to Amazon Bedrock with SigV4 +token (``bedrock_oauth`` mode), directly to Amazon Bedrock with SigV4 request signing via the standard AWS credential chain (``bedrock_sigv4`` -mode). It also blanks out inherited HTTP proxies, applies an OS-level -sandbox, and (optionally) enables OTLP telemetry. +mode), or at the OrcaRouter gateway (``orcarouter`` mode). It also blanks out +inherited HTTP proxies, applies an OS-level sandbox, and (optionally) enables +OTLP telemetry. """ from __future__ import annotations @@ -16,6 +17,7 @@ from urllib.parse import urlparse from .config import AgentConfig +from .orcarouter import cli_base_url # Match the 1-million-context variant of a Claude model. Used to pick a # higher autocompact threshold (more headroom) when the user opts into the @@ -55,6 +57,10 @@ def _anthropic_host(cfg: AgentConfig) -> str: return urlparse(cfg.anthropic.bedrock_base_url).hostname or "" region = cfg.anthropic.aws_region return f"bedrock-runtime.{region}.amazonaws.com" if region else "" + if cfg.anthropic.auth_mode == "orcarouter": + # The bundled CLI talks to the OrcaRouter inference origin (derived + # from the configured API base, never from the auth origin). + return urlparse(cli_base_url(cfg.orcarouter.api_base_url)).hostname or "" return "api.anthropic.com" @@ -87,6 +93,17 @@ def _sandbox_allowed_domains(cfg: AgentConfig) -> list[str]: """ host = _anthropic_host(cfg) domains = [host] if host else [] + if cfg.anthropic.auth_mode == "orcarouter": + # The agent itself (outside the sandboxed CLI) fetches the live model + # catalog and runs the OAuth exchange against the OrcaRouter auth + # origin. Allow-list both configured origins so an HTTPS self-hosted + # deployment keeps working; loopback HTTP dev origins are permitted + # by the network policy validator but not added here. + for origin in (cfg.orcarouter.api_base_url, cfg.orcarouter.auth_base_url): + parsed = urlparse(origin) + if parsed.scheme == "https" and parsed.hostname: + if parsed.hostname not in domains: + domains.append(parsed.hostname) if cfg.anthropic.auth_mode == "bedrock_sigv4": region = cfg.anthropic.aws_region if region: @@ -175,6 +192,18 @@ def build_claude_settings( # Blank → default credential chain. if cfg.anthropic.aws_profile: env["AWS_PROFILE"] = cfg.anthropic.aws_profile + elif cfg.anthropic.auth_mode == "orcarouter": + # Route the bundled CLI at the OrcaRouter gateway. The credential is + # an ordinary OrcaRouter API key (from either connect adapter) and + # rides in as ANTHROPIC_API_KEY. ANTHROPIC_BASE_URL is the API origin + # WITHOUT the trailing /v1 — the CLI appends the version segment + # itself, so pointing it at .../v1 would double it and 404. + env.update( + { + "ANTHROPIC_BASE_URL": cli_base_url(cfg.orcarouter.api_base_url), + "ANTHROPIC_API_KEY": auth_token or cfg.orcarouter.api_key, + } + ) else: # Direct Anthropic API. The API key rides in as auth_token (from the # token provider) or falls back to the configured value. diff --git a/vulnhunter-agent/agent/config.example.toml b/vulnhunter-agent/agent/config.example.toml index 4effc99..3bf678b 100644 --- a/vulnhunter-agent/agent/config.example.toml +++ b/vulnhunter-agent/agent/config.example.toml @@ -36,6 +36,12 @@ # credentials (instance/task roles) are NOT reachable — # widen the sandbox network allow-list or disable the # sandbox for those. +# "orcarouter" — route the bundled Claude Code CLI at OrcaRouter, an +# OpenAI-compatible AI gateway that routes many providers +# behind one endpoint. Credential is an OrcaRouter API key +# (sk-orca-...) obtained either by pasting one or by an +# OAuth 2.0 + PKCE browser login — see [orcarouter] below. +# Model IDs are the gateway's vendor/model namespaced IDs. auth_mode = "api_key" api_key = "" # Default model. Override per-invocation with --model. @@ -50,6 +56,43 @@ aws_region = "us-east-1" # bedrock_sigv4 mode only: optional named AWS profile. Blank → default chain. aws_profile = "" +[orcarouter] +# auth_mode = "orcarouter" only. OrcaRouter is an OpenAI-compatible AI +# gateway. Two public origins are involved and they are NOT interchangeable: +# authentication lives on www.orcarouter.ai, inference + model catalog on +# api.orcarouter.ai. Never derive one from the other. +# +# Get a credential with either entry point (both are first-class): +# python -m agent --mode=orcarouter --orcarouter-action=store-api-key +# python -m agent --mode=orcarouter --orcarouter-action=login +# Check or clear what is stored: +# python -m agent --mode=orcarouter --orcarouter-action=status +# python -m agent --mode=orcarouter --orcarouter-action=clear +# Inspect the live model catalog (no scan required): +# python -m agent --mode=orcarouter --orcarouter-action=models --capability chat +# +# The credential is stored 0o600 in ~/.vulnhunter/orcarouter.json (override +# with credentials_file). It is a durable API key — there is no refresh +# grant. A relay 401/403 marks the exact stored generation for reauth. +connect = "api_key" # "api_key" (paste a key) | "oauth" (PKCE login) +flow = "a" # "a" loopback redirect | "b" out-of-band code +app_name = "VulnHunter Agent" # shown on the consent screen +scope = "api" # requested grant; the granted scope is read back +auth_base_url = "https://www.orcarouter.ai" +api_base_url = "https://api.orcarouter.ai/v1" +# Self-hosted deployments: set base_url to a single shared origin, or the two +# explicit overrides (which win). ORCA_BASE_URL / ORCA_AUTH_BASE_URL / +# ORCA_API_BASE_URL do the same via env, and env wins over TOML. +# base_url = "" +# client_id = "" +# Optional literal key (prefer the store or the ORCA_API_KEY env var). +api_key = "" +login_timeout_seconds = 300 +http_timeout_seconds = 30 +catalog_timeout_seconds = 20 +catalog_enabled = true +catalog_limit = 500 + [oauth] # bedrock_oauth mode only: OAuth2 client-credentials endpoint that mints the # bearer token forwarded to the Bedrock proxy as ANTHROPIC_AUTH_TOKEN. diff --git a/vulnhunter-agent/agent/config.py b/vulnhunter-agent/agent/config.py index 49206b3..5527f37 100644 --- a/vulnhunter-agent/agent/config.py +++ b/vulnhunter-agent/agent/config.py @@ -53,6 +53,13 @@ class AnthropicConfig: # are not allow-listed, so container/instance-role # credentials require widening the sandbox network # allow-list or disabling the sandbox. + # "orcarouter" — route the bundled CLI at OrcaRouter's + # OpenAI/Anthropic-compatible gateway (see + # [orcarouter]). The credential is an OrcaRouter API + # key ("sk-orca-...") obtained either by pasting one or + # by an OAuth 2.0 + PKCE login; it is delivered to the + # CLI as ANTHROPIC_API_KEY against + # [orcarouter].api_base_url. model: str auth_mode: str = "api_key" api_key: str = "" @@ -64,6 +71,64 @@ class AnthropicConfig: aws_profile: str = "" +@dataclass(frozen=True) +class OrcaRouterConfig: + """Settings for the first-class OrcaRouter provider (``auth_mode``). + + OrcaRouter is an OpenAI-compatible AI gateway that routes many + providers behind one endpoint. Two public origins are involved and + they are deliberately NOT interchangeable: + + - **authentication** (``auth_base_url``) — the consent screen and the + code-for-key exchange. Default ``https://www.orcarouter.ai``. + - **inference + model catalog** (``api_base_url``) — defaults to + ``https://api.orcarouter.ai/v1``. + + ``ORCA_BASE_URL`` is a shared self-hosted fallback for both; the + explicit ``ORCA_AUTH_BASE_URL`` / ``ORCA_API_BASE_URL`` overrides win + over it. Non-loopback origins must be HTTPS. + """ + + # Public defaults. A self-hosted deployment sets the shared fallback or + # the explicit per-origin overrides. + auth_base_url: str = "https://www.orcarouter.ai" + api_base_url: str = "https://api.orcarouter.ai/v1" + # How the credential is obtained: + # "api_key" — the user pastes (or configures) an existing sk-orca- key. + # "oauth" — OAuth 2.0 + PKCE: the user authorizes in a browser and + # the agent exchanges the code for a durable API key. Both + # paths yield the same kind of key downstream. + connect: str = "api_key" + # PKCE delivery for the "oauth" connect mode: + # "a" — loopback redirect (default; works where 127.0.0.1 is bindable) + # "b" — out-of-band code (hosts that cannot listen on loopback) + flow: str = "a" + # Shown on the consent screen; labelled as a claim by OrcaRouter. + app_name: str = "VulnHunter Agent" + # Requested grant. "api" is the inference grant. + scope: str = "api" + # Bound the whole interactive authorization (seconds). + login_timeout_seconds: int = 300 + # Bound the single exchange request (seconds). + http_timeout_seconds: int = 30 + # Bound live model discovery (seconds). + catalog_timeout_seconds: int = 20 + catalog_enabled: bool = True + # Hard cap on accepted catalog records (defends against an unbounded + # response). Also bounds `GET /v1/models?capability=...`. + catalog_limit: int = 500 + # Optional explicit credential-store path. Blank → the default under + # ``~/.vulnhunter/``. The store is written 0o600 in a 0o700 dir. + credentials_file: str = "" + # Optional OAuth client_id (sent on the authorize URL only when set). + # OrcaRouter issues no client secret and needs no pre-registered client; + # this exists purely to satisfy hosts that expect an OAuth client id. + client_id: str = "" + # Optional per-connect API key supplied in config/TOML (avoids writing + # the literal to a file for the common case; env vars are preferred). + api_key: str = "" + + @dataclass(frozen=True) class OAuthConfig: token_endpoint: str @@ -310,6 +375,7 @@ class RepoPropertiesConfig: class AgentConfig: anthropic: AnthropicConfig oauth: OAuthConfig + orcarouter: OrcaRouterConfig tls: TLSConfig sandbox: SandboxConfig telemetry: TelemetryConfig @@ -328,6 +394,99 @@ class AgentConfig: _DEFAULT_CONFIG_FILENAME = "config.toml" +# OrcaRouter public defaults. Authentication and inference live on DIFFERENT +# origins; never derive one from the other (replacing the hostname or blindly +# appending /v1 lands on the wrong service). +_ORCA_DEFAULT_AUTH_BASE = "https://www.orcarouter.ai" +_ORCA_DEFAULT_API_BASE = "https://api.orcarouter.ai/v1" +# Accepted values for [orcarouter].connect and [orcarouter].flow. +_ORCA_CONNECT_MODES = ("api_key", "oauth") +_ORCA_FLOWS = ("a", "b") + + +def _env_first(env: dict[str, str], *names: str) -> str: + """Return the first non-empty env value among ``names`` (stripped).""" + for name in names: + value = env.get(name) + if value and value.strip(): + return value.strip() + return "" + + +def _load_orcarouter_config(raw: dict, env: dict[str, str]) -> OrcaRouterConfig: + """Resolve the [orcarouter] block, overlaying ORCA_* environment variables. + + Precedence for the two origins (highest first): + + 1. explicit per-origin overrides — ``[orcarouter].auth_base_url`` / + ``api_base_url`` or ``ORCA_AUTH_BASE_URL`` / ``ORCA_API_BASE_URL`` + 2. shared self-hosted fallback — ``[orcarouter].base_url`` or + ``ORCA_BASE_URL`` (applied to both origins) + 3. public defaults + """ + toml_auth = str(raw.get("auth_base_url", "") or "").strip() + toml_api = str(raw.get("api_base_url", "") or "").strip() + shared = _env_first(env, "ORCA_BASE_URL") or str( + raw.get("base_url", "") or "" + ).strip() + + auth_base = ( + _env_first(env, "ORCA_AUTH_BASE_URL") or toml_auth or shared or _ORCA_DEFAULT_AUTH_BASE + ) + api_base = ( + _env_first(env, "ORCA_API_BASE_URL") or toml_api or shared or _ORCA_DEFAULT_API_BASE + ) + + connect = ( + _env_first(env, "ORCA_CONNECT") + or str(raw.get("connect", "") or "").strip() + or "api_key" + ).lower() + if connect not in _ORCA_CONNECT_MODES: + raise ValueError( + "orcarouter.connect must be 'api_key' or 'oauth', " + f"got '{connect}'" + ) + + flow = ( + _env_first(env, "ORCA_FLOW") + or str(raw.get("flow", "") or "").strip() + or "a" + ).lower() + if flow not in _ORCA_FLOWS: + raise ValueError( + "orcarouter.flow must be 'a' (loopback redirect) or 'b' " + f"(out-of-band code), got '{flow}'" + ) + + def _int(key: str, default: int) -> int: + value = _env_first(env, _env_name("orcarouter", key)) or raw.get(key) + return int(value) if value not in (None, "") else default + + return OrcaRouterConfig( + auth_base_url=auth_base, + api_base_url=api_base, + connect=connect, + flow=flow, + app_name=_env_first(env, "ORCA_APP_NAME") + or str(raw.get("app_name", "") or "").strip() + or "VulnHunter Agent", + scope=_env_first(env, "ORCA_SCOPE") + or str(raw.get("scope", "") or "").strip() + or "api", + login_timeout_seconds=_int("login_timeout_seconds", 300), + http_timeout_seconds=_int("http_timeout_seconds", 30), + catalog_timeout_seconds=_int("catalog_timeout_seconds", 20), + catalog_enabled=bool(raw.get("catalog_enabled", True)), + catalog_limit=_int("catalog_limit", 500), + credentials_file=_env_first(env, "ORCA_CREDENTIALS_FILE") + or str(raw.get("credentials_file", "") or "").strip(), + client_id=_env_first(env, "ORCA_CLIENT_ID") + or str(raw.get("client_id", "") or "").strip(), + api_key=_env_first(env, "ORCA_API_KEY", "ORCAROUTER_API_KEY") + or str(raw.get("api_key", "") or "").strip(), + ) + def _resolve_config_path(explicit: str | os.PathLike[str] | None) -> Path | None: """Locate the TOML config file. Returns None if no file is found. @@ -409,6 +568,26 @@ def _resolve( return value +def load_orcarouter_config( + path: str | os.PathLike[str] | None = None, +) -> OrcaRouterConfig: + """Resolve ONLY the ``[orcarouter]`` block, overlaying ``ORCA_*`` env vars. + + Used by the ``--mode=orcarouter`` credential/catalog commands, which must + work without a full agent config (no model required) so a user can log in + and inspect the model catalog before wiring a scan. + """ + config_path = _resolve_config_path(path) + raw: dict = {} + if config_path is not None: + with config_path.open("rb") as fh: + raw = tomllib.load(fh) + orcarouter_raw = raw.get("orcarouter", {}) + if not isinstance(orcarouter_raw, dict): + orcarouter_raw = {} + return _load_orcarouter_config(orcarouter_raw, os.environ) + + def load_config(path: str | os.PathLike[str] | None = None) -> AgentConfig: """Load and validate the agent's configuration. @@ -434,16 +613,23 @@ def load_config(path: str | os.PathLike[str] | None = None) -> AgentConfig: .strip() .lower() ) - if auth_mode not in ("api_key", "bedrock_oauth", "bedrock_sigv4"): + if auth_mode not in ("api_key", "bedrock_oauth", "bedrock_sigv4", "orcarouter"): raise ValueError( - "anthropic.auth_mode must be 'api_key', 'bedrock_oauth', or " - f"'bedrock_sigv4', got '{auth_mode}'" + "anthropic.auth_mode must be 'api_key', 'bedrock_oauth', " + f"'bedrock_sigv4', or 'orcarouter', got '{auth_mode}'" ) - # api_key resolves from [anthropic].api_key / VULNHUNT_ANTHROPIC_API_KEY, - # falling back to the standard ANTHROPIC_API_KEY env var. - api_key = str( - _resolve(anthropic_raw, "anthropic", "api_key", default="") - ) or os.environ.get("ANTHROPIC_API_KEY", "") + # auth_mode='orcarouter' selects the OrcaRouter gateway. The credential is + # resolved by a dedicated ConnectCredentialProvider (see auth.py) rather + # than from [anthropic].api_key, so the API key here stays blank. + # Otherwise api_key resolves from [anthropic].api_key / + # VULNHUNT_ANTHROPIC_API_KEY, falling back to the standard + # ANTHROPIC_API_KEY env var. + if auth_mode == "orcarouter": + api_key = "" + else: + api_key = str( + _resolve(anthropic_raw, "anthropic", "api_key", default="") + ) or os.environ.get("ANTHROPIC_API_KEY", "") anthropic = AnthropicConfig( model=str(_resolve(anthropic_raw, "anthropic", "model", required=True)), auth_mode=auth_mode, @@ -495,6 +681,9 @@ def load_config(path: str | os.PathLike[str] | None = None) -> AgentConfig: "oauth.client_id, and oauth.client_secret" ) + orcarouter_raw = raw.get("orcarouter", {}) + orcarouter = _load_orcarouter_config(orcarouter_raw, os.environ) + tls = TLSConfig( ssl_cert_path=str( _resolve(tls_raw, "tls", "ssl_cert_path", default="") @@ -887,6 +1076,7 @@ def load_config(path: str | os.PathLike[str] | None = None) -> AgentConfig: return AgentConfig( anthropic=anthropic, oauth=oauth, + orcarouter=orcarouter, tls=tls, sandbox=sandbox, telemetry=telemetry, diff --git a/vulnhunter-agent/agent/orcarouter/__init__.py b/vulnhunter-agent/agent/orcarouter/__init__.py new file mode 100644 index 0000000..e7025b1 --- /dev/null +++ b/vulnhunter-agent/agent/orcarouter/__init__.py @@ -0,0 +1,77 @@ +"""First-class OrcaRouter provider for the VulnHunter agent. + +OrcaRouter is an OpenAI-compatible AI gateway that routes many providers +behind one endpoint. This package adds it as a named provider with two +explicit authentication choices that both yield one ordinary OrcaRouter API +key (``sk-orca-…``): + +* paste an existing API key — :class:`ApiKeyConnect` +* OAuth 2.0 + PKCE browser login — :class:`OAuthConnect` + +Both are adapters over :class:`ConnectCredentialProvider`, which implements +the project's existing ``TokenProvider`` protocol. Live model discovery and +capability filtering live in :mod:`agent.orcarouter.models`. +""" + +from __future__ import annotations + +from .connect import ( + ApiKeyConnect, + ConnectCredentialProvider, + OAuthConnect, + OAuthFlowError, + _LoopbackReceiver, +) +from .credentials import ( + ConnectMethod, + CredentialError, + CredentialRecord, + CredentialResult, + CredentialStore, + CredentialUnavailable, + ReauthRequired, + mask_key, +) +from .models import ( + ModelCapability, + ModelCatalog, + ModelInfo, + CatalogError, + discover_catalog, + fetch_catalog, + parse_catalog, + verified_seed, +) +from ._origins import OriginError, cli_base_url, join_path, validated_origin +from ._pkce import challenge_for, new_state, new_verifier, state_matches + +__all__ = [ + "ApiKeyConnect", + "CatalogError", + "ConnectCredentialProvider", + "ConnectMethod", + "CredentialError", + "CredentialRecord", + "CredentialResult", + "CredentialStore", + "CredentialUnavailable", + "ModelCapability", + "ModelCatalog", + "ModelInfo", + "OAuthConnect", + "OAuthFlowError", + "OriginError", + "ReauthRequired", + "challenge_for", + "cli_base_url", + "discover_catalog", + "fetch_catalog", + "join_path", + "mask_key", + "new_state", + "new_verifier", + "parse_catalog", + "state_matches", + "validated_origin", + "verified_seed", +] diff --git a/vulnhunter-agent/agent/orcarouter/_origins.py b/vulnhunter-agent/agent/orcarouter/_origins.py new file mode 100644 index 0000000..5723434 --- /dev/null +++ b/vulnhunter-agent/agent/orcarouter/_origins.py @@ -0,0 +1,71 @@ +"""Origin/path handling for the OrcaRouter provider. + +Authentication and inference live on DIFFERENT public origins and use +different path layouts. The fixed paths here are the single source of +truth for the provider so nothing derives one origin from the other by +swapping a hostname or blindly appending ``/v1``: + +* authorize — ``{auth_base}/auth`` +* exchange — ``{auth_base}/api/v1/auth/keys`` (NOT ``/v1/auth/keys``) +* catalog — ``{api_base}/models`` +""" + +from __future__ import annotations + +from urllib.parse import urlparse + +# Fixed auth paths (OrcaRouter API). +AUTHORIZE_PATH = "/auth" +EXCHANGE_PATH = "/api/v1/auth/keys" + +_LOOPBACK_HOSTS = frozenset({"localhost", "127.0.0.1", "::1"}) + + +class OriginError(ValueError): + """An OrcaRouter origin is malformed or violates the network policy.""" + + +def validated_origin(url: str, *, what: str) -> str: + """Validate and normalize a base URL, returning it without a trailing ``/``. + + Requires an ``http``/``https`` scheme and a host, rejects embedded + userinfo, and enforces HTTPS for anything that is not loopback. HTTP is + permitted only for loopback development (``localhost`` / ``127.0.0.1`` / + ``[::1]``), which is also what the OAuth callback URL rules allow. + """ + text = (url or "").strip() + if not text: + raise OriginError(f"{what} must not be empty") + parsed = urlparse(text) + if parsed.scheme not in ("http", "https"): + raise OriginError(f"{what} must be an http(s) URL, got {text!r}") + if parsed.username or parsed.password: + raise OriginError(f"{what} must not embed userinfo credentials") + host = (parsed.hostname or "").lower() + if not host: + raise OriginError(f"{what} must include a host, got {text!r}") + if parsed.scheme == "http" and host not in _LOOPBACK_HOSTS: + raise OriginError( + f"{what} must use HTTPS for a non-loopback host, got {text!r}" + ) + return text.rstrip("/") + + +def join_path(base: str, path: str) -> str: + """Join a validated base URL with a fixed absolute path.""" + return base.rstrip("/") + "/" + path.lstrip("/") + + +def cli_base_url(api_base_url: str) -> str: + """Derive the Claude Code CLI ``ANTHROPIC_BASE_URL`` from the API base. + + The CLI appends ``/v1/messages`` to ``ANTHROPIC_BASE_URL`` itself, so + pointing it at the ``/v1`` inference base double-appends the segment and + the request 404s. We therefore strip a trailing ``/v1`` from the SAME + configured API origin (never from a different origin) so a self-hosted + override keeps working. + """ + base = api_base_url.rstrip("/") + if base.endswith("/v1"): + base = base[:-3] + return base diff --git a/vulnhunter-agent/agent/orcarouter/_pkce.py b/vulnhunter-agent/agent/orcarouter/_pkce.py new file mode 100644 index 0000000..b9b3abf --- /dev/null +++ b/vulnhunter-agent/agent/orcarouter/_pkce.py @@ -0,0 +1,56 @@ +"""PKCE helpers for the OrcaRouter OAuth 2.0 connect flow. + +Everything here is stdlib-only: ``secrets`` for the verifier/state, and +``hashlib`` + ``base64`` for the S256 challenge. No third-party crypto +dependency is introduced. + +Invariants enforced by this module: + +* the verifier is fresh cryptographic randomness on every attempt; +* the challenge is ``base64url(sha256(verifier))`` with no padding; +* the verifier is never placed in a URL, log, or error message — only the + challenge is sent to the authorization endpoint. +""" + +from __future__ import annotations + +import base64 +import hashlib +import secrets + +# 32 bytes → 43 base64url chars, the RFC 7636 recommended verifier size. +_VERIFIER_BYTES = 32 +# 16 bytes → 22 base64url chars of CSRF entropy. +_STATE_BYTES = 16 + + +def b64url(raw: bytes) -> str: + """base64url without padding (RFC 7636 §4.2 encoding).""" + return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") + + +def new_verifier() -> str: + """A fresh high-entropy PKCE code verifier.""" + return b64url(secrets.token_bytes(_VERIFIER_BYTES)) + + +def challenge_for(verifier: str) -> str: + """``base64url(sha256(verifier))`` — the S256 code challenge.""" + digest = hashlib.sha256(verifier.encode("ascii")).digest() + return b64url(digest) + + +def new_state() -> str: + """A fresh opaque CSRF state value.""" + return b64url(secrets.token_bytes(_STATE_BYTES)) + + +def state_matches(expected: str, received: str | None) -> bool: + """Constant-time comparison of the echoed CSRF state. + + Returns False for a missing or non-string value rather than raising, so + callers can branch to a clean error path. + """ + if not expected or not isinstance(received, str): + return False + return secrets.compare_digest(expected, received) diff --git a/vulnhunter-agent/agent/orcarouter/connect.py b/vulnhunter-agent/agent/orcarouter/connect.py new file mode 100644 index 0000000..7505068 --- /dev/null +++ b/vulnhunter-agent/agent/orcarouter/connect.py @@ -0,0 +1,503 @@ +"""OrcaRouter credential acquisition — the connect seam and its two adapters. + +One interface, two adapters, one result: + +* :class:`ApiKeyConnect` — the user pastes (or configures) an existing + ``sk-orca-…`` key. +* :class:`OAuthConnect` — OAuth 2.0 + PKCE (Flow A loopback redirect or + Flow B out-of-band code) yields a durable API key. + +:class:`ConnectCredentialProvider` owns the small interface and adapts the +result to the project's existing ``TokenProvider`` protocol +(``get_valid_token``), so the SDK call sites and the model catalog never +see which entry produced the key. + +Security notes baked in here: + +* Flow A compares the echoed ``state`` with a constant-time comparison + before the code is redeemed. +* Verifier and state are generated fresh per attempt from a CSPRNG and the + verifier never leaves the process until the exchange — it is not placed + in the authorize URL, logs, or errors. +* Exchange errors are mapped to actionable messages; the response body's + credential material is never surfaced. +""" + +from __future__ import annotations + +import http.server +import logging +import queue +import threading +import time +import webbrowser +from typing import Callable + +import httpx + +from .._url import redact +from ..config import OrcaRouterConfig, TLSConfig +from ._origins import ( + AUTHORIZE_PATH, + EXCHANGE_PATH, + OriginError, + validated_origin, +) +from ._pkce import challenge_for, new_state, new_verifier, state_matches +from .credentials import ( + ConnectMethod, + CredentialError, + CredentialRecord, + CredentialResult, + CredentialStore, + CredentialUnavailable, + ReauthRequired, + mask_key, +) + +logger = logging.getLogger(__name__) + +# Path the loopback listener serves (Flow A). Fixed by our own authorize call. +CALLBACK_PATH = "/cb" + +# Scope that satisfies inference against the relay. +_INFERENCE_SCOPE = "api" + + +class OAuthFlowError(CredentialError): + """A user-actionable failure in the OAuth 2.0 + PKCE connect flow.""" + + +# Injectable seams (tests substitute fakes; production uses the defaults). +PostFn = Callable[[str, dict, float], "tuple[int, dict]"] +OpenBrowserFn = Callable[[str], None] + + +def _default_post(url: str, body: dict, timeout: float, *, verify) -> tuple[int, dict]: + with httpx.Client(verify=verify) as client: + response = client.post( + url, + json=body, + headers={"Content-Type": "application/json"}, + timeout=timeout, + ) + try: + payload = response.json() + except ValueError: + payload = {} + if not isinstance(payload, dict): + payload = {} + return response.status_code, payload + + +class _LoopbackReceiver: + """Single-use loopback HTTP listener for the Flow A redirect. + + Binds an ephemeral port on 127.0.0.1 BEFORE the browser is opened so the + callback URL is known and nothing races. Serves `/cb`, hands the query + parameters back to the caller, answers the browser with a short + close-this-tab page, and shuts the socket down. + """ + + def __init__(self) -> None: + self._results: "queue.Queue[dict]" = queue.Queue(maxsize=1) + receiver = self + + class _Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, *args) -> None: # noqa: D401 - silence access log + return + + def do_GET(self) -> None: # noqa: N802 - http.server API + from urllib.parse import parse_qs, urlparse + + parsed = urlparse(self.path) + if parsed.path != CALLBACK_PATH: + self.send_response(404) + self.end_headers() + return + self.send_response(200) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.end_headers() + self.wfile.write( + b"

OrcaRouter connected. " + b"You can close this tab.

" + ) + params = {k: v[0] for k, v in parse_qs(parsed.query).items()} + try: + receiver._results.put_nowait(params) + except queue.Full: + pass + + self._server = http.server.HTTPServer(("127.0.0.1", 0), _Handler) + self.port = int(self._server.server_address[1]) + self._thread = threading.Thread( + target=self._server.serve_forever, daemon=True + ) + + def start(self) -> None: + self._thread.start() + + def wait(self, timeout: float) -> dict | None: + try: + return self._results.get(timeout=timeout) + except queue.Empty: + return None + + def stop(self) -> None: + try: + self._server.shutdown() + except Exception: # noqa: BLE001 - best-effort socket cleanup + pass + try: + self._server.server_close() + except Exception: # noqa: BLE001 + pass + + +class OAuthConnect: + """OAuth 2.0 + PKCE adapter producing a durable OrcaRouter API key.""" + + def __init__( + self, + *, + auth_base_url: str, + app_name: str, + scope: str = _INFERENCE_SCOPE, + flow: str = "a", + client_id: str = "", + login_timeout_seconds: int = 300, + http_timeout_seconds: int = 30, + tls_verify: str | bool = True, + post: PostFn | None = None, + open_browser: OpenBrowserFn | None = None, + receiver_factory: Callable[[], _LoopbackReceiver] | None = None, + code_provider: Callable[[], str] | None = None, + clock: Callable[[], float] = time.monotonic, + ) -> None: + try: + self._auth_base = validated_origin(auth_base_url, what="orcarouter auth base") + except OriginError as exc: + raise OAuthFlowError(str(exc)) from exc + self._app_name = app_name or "VulnHunter Agent" + self._scope = scope or _INFERENCE_SCOPE + self._flow = (flow or "a").lower() + self._client_id = client_id + self._timeout = login_timeout_seconds + self._http_timeout = http_timeout_seconds + self._verify = tls_verify + self._post = post or ( + lambda url, body, t: _default_post(url, body, t, verify=self._verify) + ) + self._open_browser = open_browser or webbrowser.open + self._receiver_factory = receiver_factory or _LoopbackReceiver + self._code_provider = code_provider + self._clock = clock + + # -- pure helpers ----------------------------------------------------- + + def authorize_url(self, *, callback_url: str, challenge: str, state: str) -> str: + """Build the consent-screen URL (auth origin, fixed /auth path).""" + from urllib.parse import urlencode + + params = { + "callback_url": callback_url, + "code_challenge": challenge, + "code_challenge_method": "S256", + "state": state, + "app_name": self._app_name, + "scope": self._scope, + } + if self._client_id: + params["client_id"] = self._client_id + return f"{self._auth_base}{AUTHORIZE_PATH}?" + urlencode(params) + + def exchange_url(self) -> str: + """The code-for-key exchange URL (auth origin — never the API origin).""" + return f"{self._auth_base}{EXCHANGE_PATH}" + + # -- exchange --------------------------------------------------------- + + def _exchange(self, code: str, verifier: str) -> dict: + try: + status, payload = self._post( + self.exchange_url(), + { + "code": code, + "code_verifier": verifier, + "code_challenge_method": "S256", + }, + self._http_timeout, + ) + except Exception as exc: # noqa: BLE001 - transport failure at the boundary + # Never include the request/response material — it carries the code + # and verifier. + raise OAuthFlowError( + f"OrcaRouter key exchange failed: {type(exc).__name__}" + ) from exc + if status == 200 and str(payload.get("key", "")).strip(): + granted = str(payload.get("scope", "") or "") + # The response says what was GRANTED, not what we asked for. A + # narrower grant (e.g. asked connector, got api) means the account + # lacked the wider role — say so rather than assume we hold it. + if granted and granted != self._scope: + raise OAuthFlowError( + f"OrcaRouter granted scope {granted!r}, not the requested " + f"{self._scope!r}; the account may not hold that grant" + ) + return { + "key": str(payload["key"]), + "user_id": str(payload.get("user_id", "") or ""), + "scope": granted or _INFERENCE_SCOPE, + } + # Never surface the response body — it can echo request material. + if status == 403: + raise OAuthFlowError( + "OrcaRouter rejected the authorization code (unknown, expired, " + "already used, or the verifier did not match). Start the login again." + ) + if status == 400: + raise OAuthFlowError( + "OrcaRouter rejected the code_challenge_method (S256) — the " + "authorization was downgraded. Start the login again." + ) + if status == 429: + raise OAuthFlowError( + "OrcaRouter rate-limited the key issuance (10 PKCE keys per user " + "per 24 hours). Reuse the stored credential or try again later." + ) + raise OAuthFlowError(f"OrcaRouter key exchange failed with HTTP {status}") + + # -- flow A / B ------------------------------------------------------- + + def connect(self) -> dict: + """Run the flow and return ``{key, user_id, scope}``. + + Raises :class:`OAuthFlowError` on denial, state mismatch, timeout, or + any exchange/network failure. Never hangs and never hot-loops. + """ + verifier = new_verifier() + challenge = challenge_for(verifier) + state = new_state() + + if self._flow == "a": + code = self._run_loopback(challenge, state) + else: + code = self._run_out_of_band(challenge, state) + + return self._exchange(code, verifier) + + def _run_loopback(self, challenge: str, state: str) -> str: + receiver = self._receiver_factory() + receiver.start() + try: + callback_url = f"http://127.0.0.1:{receiver.port}{CALLBACK_PATH}" + url = self.authorize_url( + callback_url=callback_url, challenge=challenge, state=state + ) + self._announce(url) + params = receiver.wait(self._timeout) + if params is None: + raise OAuthFlowError( + "OrcaRouter authorization timed out before a callback arrived." + ) + finally: + receiver.stop() + + # Compare state before doing anything else. + if not state_matches(state, params.get("state")): + raise OAuthFlowError( + "OrcaRouter callback state did not match; refusing the code." + ) + error = params.get("error") + if error: + raise OAuthFlowError(_describe_callback_error(error)) + code = params.get("code") + if not code: + raise OAuthFlowError("OrcaRouter callback carried no authorization code.") + return code + + def _run_out_of_band(self, challenge: str, state: str) -> str: + url = self.authorize_url(callback_url="oob", challenge=challenge, state=state) + self._announce(url) + provider = self._code_provider + if provider is None: + provider = lambda: input("Paste the code shown by OrcaRouter: ").strip() # noqa: E731 + try: + code = (provider() or "").strip() + except EOFError as exc: + raise OAuthFlowError( + "No code supplied; the out-of-band login was cancelled." + ) from exc + if not code: + raise OAuthFlowError("No code supplied; the OrcaRouter login was cancelled.") + return code + + def _announce(self, url: str) -> None: + # The URL contains only the challenge and state — never the verifier. + logger.info("Authorize OrcaRouter in your browser: %s", redact(url)) + print( + "OrcaRouter: open this URL to authorize if your browser does not " + f"open automatically:\n{url}" + ) + try: + self._open_browser(url) + except Exception: # noqa: BLE001 - headless hosts have no browser + logger.info("Could not open a browser automatically; use the URL above.") + + +def _describe_callback_error(error: str) -> str: + if error == "access_denied": + return "OrcaRouter authorization was denied." + return f"OrcaRouter authorization failed: {error}" + + +class ApiKeyConnect: + """API-key adapter — store a user-supplied ``sk-orca-…`` key.""" + + def __init__(self, store: CredentialStore) -> None: + self._store = store + + def store(self, api_key: str) -> CredentialRecord: + key = (api_key or "").strip() + if not key: + raise CredentialError("An OrcaRouter API key is required.") + # Lightweight format check only — an sk-orca- prefix is not proof the + # key is valid; the first real request establishes that. + if not key.startswith("sk-orca-"): + logger.warning( + "OrcaRouter API key does not start with 'sk-orca-' (%s)", + mask_key(key), + ) + return self._store.store_new(api_key=key, method=ConnectMethod.API_KEY) + + +class ConnectCredentialProvider: + """The credential interface; both adapters live behind it. + + Implements the project's ``TokenProvider`` protocol so existing call + sites (``auth_token = token_manager.get_valid_token()``) work unchanged. + """ + + def __init__( + self, + config: OrcaRouterConfig, + *, + store: CredentialStore | None = None, + tls: TLSConfig | None = None, + post: PostFn | None = None, + open_browser: OpenBrowserFn | None = None, + receiver_factory: Callable[[], _LoopbackReceiver] | None = None, + code_provider: Callable[[], str] | None = None, + ) -> None: + self._config = config + self._store = store or CredentialStore( + config.credentials_file or "~/.vulnhunter/orcarouter.json" + ) + verify: str | bool = tls.ssl_cert_path if tls and tls.ssl_cert_path else True + self._api_adapter = ApiKeyConnect(self._store) + self._oauth_adapter = OAuthConnect( + auth_base_url=config.auth_base_url, + app_name=config.app_name, + scope=config.scope, + flow=config.flow, + client_id=config.client_id, + login_timeout_seconds=config.login_timeout_seconds, + http_timeout_seconds=config.http_timeout_seconds, + tls_verify=verify, + post=post, + open_browser=open_browser, + receiver_factory=receiver_factory, + code_provider=code_provider, + ) + + # -- TokenProvider ---------------------------------------------------- + + def get_valid_token(self) -> str: + return self.resolve().api_key + + # -- resolution ------------------------------------------------------- + + def resolve(self) -> CredentialResult: + """Return the usable credential, or raise. + + Precedence: the stored credential is authoritative — it is what makes + a durable PKCE key reusable across restarts without re-authorizing, + and a rejected one stays a terminal reauthentication requirement + (never a silent switch to another credential). A key supplied via + config/env is used only when nothing is stored. + """ + record = self._store.load() + if record is not None: + if record.needs_reauth: + # A revoked durable key is a terminal reauthentication + # requirement — no fake refresh is attempted. + raise ReauthRequired( + "The stored OrcaRouter credential was rejected; reconnect " + "with the API key or sign in again." + ) + self._classify(record) + return self._as_result(record, source="store") + env_key = (self._config.api_key or "").strip() + if env_key: + return CredentialResult( + api_key=env_key, + method=ConnectMethod.API_KEY, + source="config", + ) + raise CredentialUnavailable( + "No OrcaRouter credential is configured. Paste an sk-orca- API key " + "or run the OrcaRouter OAuth login." + ) + + @staticmethod + def _classify(record: CredentialRecord) -> None: + """Reject a corrupt/truncated OAuth credential as a terminal state.""" + if record.method == ConnectMethod.OAUTH.value and not record.api_key.startswith( + "sk-orca-" + ): + raise ReauthRequired( + "The stored OrcaRouter login credential is corrupt or truncated; " + "sign in again." + ) + + @staticmethod + def _as_result(record: CredentialRecord, *, source: str) -> CredentialResult: + return CredentialResult( + api_key=record.api_key, + method=ConnectMethod(record.method), + scope=record.scope, + user_id=record.user_id, + source=source, + generation=record.generation, + needs_reauth=record.needs_reauth, + ) + + # -- the two adapters, explicitly exposed ----------------------------- + + def store_api_key(self, api_key: str) -> CredentialResult: + """API-key adapter entry point.""" + record = self._api_adapter.store(api_key) + return self._as_result(record, source="api_key") + + def connect_oauth(self) -> CredentialResult: + """OAuth 2.0 + PKCE adapter entry point.""" + result = self._oauth_adapter.connect() + record = self._store.store_new( + api_key=result["key"], + method=ConnectMethod.OAUTH, + scope=result.get("scope", "api"), + user_id=result.get("user_id", ""), + ) + return self._as_result(record, source="oauth") + + def clear(self) -> None: + """Remove the stored credential (both entries must support clearing).""" + self._store.clear() + + def note_rejection(self, *, api_key: str, generation: int) -> bool: + """Mark the exact rejected credential generation as needing reauth.""" + return self._store.mark_rejected(api_key=api_key, generation=generation) + + @property + def store(self) -> CredentialStore: + return self._store diff --git a/vulnhunter-agent/agent/orcarouter/credentials.py b/vulnhunter-agent/agent/orcarouter/credentials.py new file mode 100644 index 0000000..c9e2605 --- /dev/null +++ b/vulnhunter-agent/agent/orcarouter/credentials.py @@ -0,0 +1,200 @@ +"""OrcaRouter credential model and durable store. + +Both connect entry points — a pasted API key and an OAuth 2.0 + PKCE +login — converge on the SAME downstream artifact: a normal OrcaRouter API +key. They are modelled as two adapters over one interface +(:class:`ConnectCredentialProvider`) returning the same +:class:`CredentialResult`, so the provider request path and the model +catalog never need to know which path produced the key. + +The store reuses the project's existing hardened-writer idiom (see +``agent/audit.py``): a 0o700 directory and a 0o600 JSON file written +atomically. It is not a new secret manager — just the same file +convention used elsewhere under ``~/.vulnhunter/``. +""" + +from __future__ import annotations + +import enum +import json +import os +import tempfile +import time +from dataclasses import asdict, dataclass, field +from pathlib import Path + +# A PKCE-issued key is a durable API key, NOT a refresh token: there is no +# refresh endpoint to call later. Treat an upstream 401 from the relay as a +# terminal reauthentication requirement for the exact account/generation +# that made the rejected request. +DEFAULT_CREDENTIALS_PATH = "~/.vulnhunter/orcarouter.json" + + +class ConnectMethod(str, enum.Enum): + """Which user-facing entry produced the credential.""" + + API_KEY = "api_key" + OAUTH = "oauth" + + +class CredentialError(RuntimeError): + """Base class for OrcaRouter credential problems.""" + + +class CredentialUnavailable(CredentialError): + """No usable credential is configured (nothing to send yet).""" + + +class ReauthRequired(CredentialError): + """The stored credential was rejected and must be re-established.""" + + +@dataclass(frozen=True) +class CredentialResult: + """The one credential shape every adapter returns. + + ``source`` records how it was obtained (for status display only); the + request path and catalog read ``api_key`` and nothing else. + """ + + api_key: str + method: ConnectMethod + scope: str = "api" + user_id: str = "" + source: str = "store" + generation: int = 0 + needs_reauth: bool = False + + +def mask_key(key: str) -> str: + """Mask an API key for display/logs — never returns the full secret.""" + if not key: + return "" + if len(key) <= 8: + return "***" + return f"{key[:8]}...{key[-4:]}" + + +@dataclass +class CredentialRecord: + """Persisted credential state for the OrcaRouter provider.""" + + api_key: str + method: str = ConnectMethod.API_KEY.value + scope: str = "api" + user_id: str = "" + # Incremented every time a NEW credential is stored. A 401 from the relay + # only invalidates the exact generation that made the request, so a late + # failure from an old request can never mark a freshly reauthorized + # credential broken. + generation: int = 1 + # Set when the relay rejected this credential. The old secret is kept + # (never silently deleted) until a new login succeeds. + needs_reauth: bool = False + created_at: float = field(default_factory=time.time) + + def masked_key(self) -> str: + return mask_key(self.api_key) + + +def resolve_credentials_path(override: str = "") -> Path: + """Resolve the credential-store path (explicit override, then default).""" + raw = (override or "").strip() or DEFAULT_CREDENTIALS_PATH + return Path(raw).expanduser() + + +class CredentialStore: + """Read/write the OrcaRouter credential record with hardened permissions.""" + + def __init__(self, path: str | os.PathLike[str] = DEFAULT_CREDENTIALS_PATH): + self.path = Path(path).expanduser() + + def load(self) -> CredentialRecord | None: + try: + raw = self.path.read_text(encoding="utf-8") + except FileNotFoundError: + return None + except OSError: + return None + if not raw.strip(): + # A truncated/empty file is treated as "no usable credential" + # rather than raising — re-login repairs it. + return None + try: + data = json.loads(raw) + except json.JSONDecodeError: + return None + if not isinstance(data, dict) or not str(data.get("api_key", "")).strip(): + return None + known = {f for f in CredentialRecord.__dataclass_fields__} + return CredentialRecord(**{k: v for k, v in data.items() if k in known}) + + def save(self, record: CredentialRecord) -> None: + directory = self.path.parent + directory.mkdir(parents=True, exist_ok=True) + try: + os.chmod(directory, 0o700) + except OSError: + pass # best-effort; not all filesystems allow it + payload = json.dumps(asdict(record), sort_keys=True) + fd, tmp = tempfile.mkstemp(dir=str(directory), prefix=".tmp-", suffix=".json") + try: + # 0o600 like audit.py — os.open with an explicit mode, not the + # umask-dependent default. + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as fh: + fh.write(payload) + os.replace(tmp, self.path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise + + def clear(self) -> None: + try: + self.path.unlink() + except FileNotFoundError: + pass + + def store_new( + self, + *, + api_key: str, + method: ConnectMethod, + scope: str = "api", + user_id: str = "", + ) -> CredentialRecord: + """Persist a (re)established credential, advancing the generation.""" + previous = self.load() + generation = (previous.generation + 1) if previous else 1 + record = CredentialRecord( + api_key=api_key, + method=method.value, + scope=scope, + user_id=user_id, + generation=generation, + needs_reauth=False, + ) + self.save(record) + return record + + def mark_rejected(self, *, api_key: str, generation: int) -> bool: + """Flag the EXACT rejected credential generation as needing reauth. + + Returns True only when the stored record still matches both the key + and the generation that made the rejected request. A late failure + carrying an older generation (or a different key) is ignored, so it + cannot poison a credential that was reauthorized in the meantime. + """ + record = self.load() + if record is None: + return False + if record.generation != generation or record.api_key != api_key: + return False + if record.needs_reauth: + return True + record.needs_reauth = True + self.save(record) + return True diff --git a/vulnhunter-agent/agent/orcarouter/models.py b/vulnhunter-agent/agent/orcarouter/models.py new file mode 100644 index 0000000..6d41a2b --- /dev/null +++ b/vulnhunter-agent/agent/orcarouter/models.py @@ -0,0 +1,348 @@ +"""OrcaRouter model discovery and capability filtering. + +The single source of truth for the model list is ``GET {api_base}/models`` +against the configured inference origin. Live results are authoritative; +when discovery fails (offline, timeout, auth error) a small, verified +cold-start seed is used and the result is reported as degraded so the UI +says so rather than silently degrading to free text. + +Capability filtering is metadata-driven and fail-closed: a model is offered +for an entry point only when the catalog explicitly proves it supports +that entry point. Nothing is inferred from a model's name. +""" + +from __future__ import annotations + +import enum +import json +import logging +from dataclasses import dataclass +from typing import Any + +import httpx + +from ..config import OrcaRouterConfig + +logger = logging.getLogger(__name__) + +# Bound the response we are willing to read/parse. A catalog response must +# not be able to consume unbounded memory. +_MAX_RESPONSE_BYTES = 4 * 1024 * 1024 + +# Endpoint types that can carry a text chat/agent turn. +_TEXT_ENDPOINT_TYPES = frozenset({"openai", "openai-response", "anthropic", "gemini"}) +# Endpoint types that are explicitly non-text; excluded from any chat view. +_NON_TEXT_ENDPOINT_TYPES = frozenset( + {"image-generation", "openai-video", "jina-rerank", "embeddings"} +) + + +class ModelCapability(str, enum.Enum): + """A place where the agent can send a model request.""" + + CHAT = "chat" # text-only chat/agent + IMAGE_INPUT = "image_input" # multimodal understanding: image + AUDIO_INPUT = "audio_input" + VIDEO_INPUT = "video_input" + EMBEDDING = "embedding" + IMAGE_GENERATION = "image_generation" + VIDEO_GENERATION = "video_generation" + RERANK = "rerank" + + @property + def required_modality(self) -> str | None: + return { + ModelCapability.IMAGE_INPUT: "image", + ModelCapability.AUDIO_INPUT: "audio", + ModelCapability.VIDEO_INPUT: "video", + }.get(self) + + +# Catalog query parameter per capability. The agent only ever asks for the +# capabilities it actually uses; unsupported ones stay absent. +_CAPABILITY_QUERY = {"chat": "chat", "embedding": "embedding", "image": "image"} + + +@dataclass(frozen=True) +class ModelInfo: + """One catalog record, normalized.""" + + id: str + name: str = "" + context_length: int | None = None + max_completion_tokens: int | None = None + input_modalities: tuple[str, ...] = () + output_modalities: tuple[str, ...] = () + endpoint_types: tuple[str, ...] = () + reasoning: bool = False + reasoning_efforts: tuple[str, ...] = () + source: str = "live" # "live" | "seed" + + def is_text_chat(self) -> bool: + types = set(self.endpoint_types) + if types & _NON_TEXT_ENDPOINT_TYPES: + return False + # Fail closed: an empty endpoint list proves nothing about chat. + return bool(types & _TEXT_ENDPOINT_TYPES) + + def supports(self, capability: ModelCapability) -> bool: + """Metadata-only capability check. Unknown metadata → False.""" + types = set(self.endpoint_types) + if capability is ModelCapability.CHAT: + return self.is_text_chat() + if capability is ModelCapability.EMBEDDING: + return "embeddings" in types + if capability is ModelCapability.IMAGE_GENERATION: + return "image-generation" in types + if capability is ModelCapability.VIDEO_GENERATION: + return "openai-video" in types + if capability is ModelCapability.RERANK: + return "jina-rerank" in types + modality = capability.required_modality + if modality is None: + return False + # Multimodal understanding requires BOTH a text chat channel and an + # explicit declaration of the modality actually being uploaded. + return self.is_text_chat() and modality in self.input_modalities + + +@dataclass(frozen=True) +class ModelCatalog: + """A resolved catalog plus how it was obtained.""" + + models: tuple[ModelInfo, ...] + source: str # "live" | "seed" | "cache" + degraded: bool = False + error: str = "" # safe, secret-free reason when degraded + + def for_capability(self, capability: ModelCapability) -> tuple[ModelInfo, ...]: + return tuple(m for m in self.models if m.supports(capability)) + + +def _as_str_tuple(value: Any) -> tuple[str, ...]: + if not isinstance(value, (list, tuple)): + return () + return tuple(str(v) for v in value if isinstance(v, (str, int)) and str(v)) + + +def _as_int(value: Any) -> int | None: + if isinstance(value, bool) or not isinstance(value, (int, float)): + return None + n = int(value) + return n if n > 0 else None + + +def parse_model(record: dict) -> ModelInfo | None: + """Normalize one catalog record; return None for an unusable shape.""" + if not isinstance(record, dict): + return None + model_id = str(record.get("id", "") or "").strip() + if not model_id: + return None + architecture = record.get("architecture") + architecture = architecture if isinstance(architecture, dict) else {} + top_provider = record.get("top_provider") + top_provider = top_provider if isinstance(top_provider, dict) else {} + + context_length = _as_int(record.get("context_length")) or _as_int( + top_provider.get("context_length") + ) + max_completion = _as_int(record.get("max_completion_tokens")) or _as_int( + top_provider.get("max_completion_tokens") + ) + # Reasoning metadata may be advertised either as a flag or as an effort + # ladder; preserve a verified ladder when present. + efforts = _as_str_tuple(record.get("reasoning_efforts")) or _as_str_tuple( + record.get("reasoning_effort_levels") + ) + reasoning = bool(record.get("reasoning")) or bool(efforts) + + return ModelInfo( + id=model_id, + name=str(record.get("name", "") or ""), + context_length=context_length, + max_completion_tokens=max_completion, + input_modalities=_as_str_tuple(architecture.get("input_modalities")), + output_modalities=_as_str_tuple(architecture.get("output_modalities")), + endpoint_types=_as_str_tuple(record.get("supported_endpoint_types")), + reasoning=reasoning, + reasoning_efforts=efforts, + source="live", + ) + + +def parse_catalog(payload: Any, *, limit: int) -> tuple[ModelInfo, ...]: + """Parse a ``/models`` payload, dropping unusable records and bounding size.""" + if not isinstance(payload, dict): + return () + data = payload.get("data") + if not isinstance(data, list): + return () + out: list[ModelInfo] = [] + seen: set[str] = set() + for record in data: + if len(out) >= max(1, limit): + break + info = parse_model(record) + if info is None or info.id in seen: + continue + seen.add(info.id) + out.append(info) + return tuple(out) + + +# -------------------------------------------------------------------------- +# Verified cold-start seed +# -------------------------------------------------------------------------- +# A small, verified fallback so a fresh installation (or a discovery outage) +# is still usable. It is used ONLY when live discovery fails; a live catalog +# is authoritative and the seed is never mixed into it. Metadata kept here +# (context window, input modalities, reasoning ladder) must survive a +# discovery outage — see the seed tests. +_VERIFIED_SEED: tuple[ModelInfo, ...] = ( + ModelInfo( + id="openai/gpt-5.5", + name="GPT-5.5", + context_length=400_000, + max_completion_tokens=128_000, + input_modalities=("text", "image"), + output_modalities=("text",), + endpoint_types=("openai", "openai-response", "anthropic"), + reasoning=True, + reasoning_efforts=("low", "medium", "high", "xhigh"), + source="seed", + ), + ModelInfo( + id="anthropic/claude-opus-4.8", + name="Claude Opus 4.8", + context_length=200_000, + max_completion_tokens=64_000, + input_modalities=("text", "image"), + output_modalities=("text",), + endpoint_types=("anthropic", "openai"), + reasoning=True, + reasoning_efforts=("low", "medium", "high"), + source="seed", + ), + ModelInfo( + id="google/gemini-3.5-flash", + name="Gemini 3.5 Flash", + context_length=1_000_000, + max_completion_tokens=64_000, + input_modalities=("text", "image"), + output_modalities=("text",), + endpoint_types=("openai", "gemini", "anthropic"), + source="seed", + ), + ModelInfo( + id="deepseek/deepseek-v4-pro", + name="DeepSeek V4 Pro", + context_length=1_048_576, + max_completion_tokens=384_000, + input_modalities=("text",), + output_modalities=("text",), + endpoint_types=("openai", "openai-response"), + source="seed", + ), + ModelInfo( + id="orcarouter/auto", + name="OrcaRouter Auto", + input_modalities=("text",), + output_modalities=("text",), + endpoint_types=("openai", "anthropic", "gemini"), + source="seed", + ), +) + + +def verified_seed() -> tuple[ModelInfo, ...]: + """The verified fallback catalog (never mixed into a live result).""" + return _VERIFIED_SEED + + +class CatalogError(RuntimeError): + """A discovery failure whose message is safe to show (no secrets).""" + + +def fetch_catalog( + config: OrcaRouterConfig, + api_key: str, + *, + capability: str | None = "chat", + http_client: httpx.Client | None = None, +) -> tuple[ModelInfo, ...]: + """Fetch and parse the live catalog from the configured API origin. + + Requests ``GET {api_base}/models`` (optionally ``?capability=…``) with + Bearer auth, bounded by ``catalog_timeout_seconds`` and + ``_MAX_RESPONSE_BYTES``. Raises :class:`CatalogError` on any failure. + """ + url = config.api_base_url.rstrip("/") + "/models" + params = {"capability": _CAPABILITY_QUERY[capability]} if capability in _CAPABILITY_QUERY else None + headers = {} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + client = http_client or httpx.Client() + owns_client = http_client is None + try: + with client.stream( + "GET", url, params=params, headers=headers, + timeout=config.catalog_timeout_seconds, + ) as response: + if response.status_code == 401: + raise CatalogError( + "OrcaRouter model catalog rejected the credential (HTTP 401)." + ) + if response.status_code >= 400: + raise CatalogError( + f"OrcaRouter model catalog returned HTTP {response.status_code}." + ) + body = bytearray() + for chunk in response.iter_bytes(): + body.extend(chunk) + if len(body) > _MAX_RESPONSE_BYTES: + raise CatalogError("OrcaRouter model catalog response was too large.") + except httpx.HTTPError as exc: + raise CatalogError(f"OrcaRouter model catalog request failed: {type(exc).__name__}") from exc + finally: + if owns_client: + client.close() + + try: + payload = json.loads(bytes(body)) + except json.JSONDecodeError as exc: + raise CatalogError("OrcaRouter model catalog returned invalid JSON.") from exc + models = parse_catalog(payload, limit=config.catalog_limit) + if not models: + raise CatalogError("OrcaRouter model catalog returned no usable models.") + return models + + +def discover_catalog( + config: OrcaRouterConfig, + api_key: str, + *, + capability: str | None = "chat", + http_client: httpx.Client | None = None, +) -> ModelCatalog: + """Resolve the catalog, falling back to the verified seed on failure. + + Live discovery is authoritative. On any failure the verified seed is + returned with ``degraded=True`` and a safe reason — never free text and + never an unverified example list. + """ + if not config.catalog_enabled: + return ModelCatalog( + models=verified_seed(), source="seed", degraded=True, + error="Live model discovery is disabled in configuration.", + ) + try: + models = fetch_catalog( + config, api_key, capability=capability, http_client=http_client + ) + return ModelCatalog(models=models, source="live", degraded=False) + except CatalogError as exc: + logger.warning("OrcaRouter catalog discovery failed: %s", exc) + return ModelCatalog( + models=verified_seed(), source="seed", degraded=True, error=str(exc) + ) diff --git a/vulnhunter-agent/agent/runner.py b/vulnhunter-agent/agent/runner.py index ea252fe..41f372d 100644 --- a/vulnhunter-agent/agent/runner.py +++ b/vulnhunter-agent/agent/runner.py @@ -467,7 +467,15 @@ def _check_no_prior_results(clone_dir: Path) -> None: class AuthRejectedError(RuntimeError): - """The Bedrock proxy rejected the bearer token; retrying won't help.""" + """The provider rejected the credential; retrying won't help. + + Raised repeatedly by the upstream on 401 (or a typed 403 credential / + model-access rejection). For the OrcaRouter provider this is a TERMINAL + reauthentication requirement: the exact credential generation that made + the rejected request is flagged and the stored key is kept, so a new + login simply replaces it. No refresh is attempted — an OrcaRouter key is + a durable grant, not a refresh token. + """ class RateLimitError(RuntimeError): @@ -1094,12 +1102,17 @@ async def _run_scan_session( if _is_auth_failure(message): consecutive_auth_failures += 1 if consecutive_auth_failures >= _MAX_CONSECUTIVE_AUTH_FAILURES: + _mark_credential_rejected(config) raise AuthRejectedError( - f"Bedrock proxy rejected the bearer token " + f"The provider rejected the credential " f"{consecutive_auth_failures} consecutive times " - "(error_status=401 authentication_failed). The OAuth " - "client likely lacks Bedrock access — check the " - "JWT's aud/scope claims against a known-working client." + "(error_status=401 authentication_failed, or a " + "typed 403 credential/model-access rejection). " + "For bedrock_oauth the OAuth client likely lacks " + "Bedrock access — check the JWT's aud/scope claims " + "against a known-working client. For orcarouter the " + "key is revoked or lacks access to the selected " + "model — sign in again or paste a new API key." ) else: consecutive_auth_failures = 0 @@ -1229,6 +1242,63 @@ def _is_auth_failure(message: SystemMessage) -> bool: return ( data.get("error_status") == 401 or str(data.get("error", "")).lower() == "authentication_failed" + or _is_credential_rejection(message) + ) + + +def _mark_credential_rejected(config: AgentConfig) -> None: + """Terminally mark the exact rejected OrcaRouter credential generation. + + Called once we have decided the upstream is rejecting our credential. + Generation-safe: only the stored record whose key and generation exactly + match the credential currently in use is flagged, so a late failure from + an old request can never mark a freshly reauthorized credential broken. + No-op for every other auth mode. + """ + if config.anthropic.auth_mode != "orcarouter": + return + try: + from .orcarouter import ConnectCredentialProvider + + provider = ConnectCredentialProvider(config.orcarouter, tls=config.tls) + result = provider.resolve() + if provider.note_rejection( + api_key=result.api_key, generation=result.generation + ): + logger.error( + "OrcaRouter credential (generation %d, %s) was rejected by the " + "relay; it now requires reauthentication. Sign in again or paste " + "a new API key.", + result.generation, + "login" if result.method.value == "oauth" else "api key", + ) + except Exception: # noqa: BLE001 - marking is best-effort, never masks the real error + logger.debug("Could not mark OrcaRouter credential rejected", exc_info=True) + + +def _is_credential_rejection(message: SystemMessage) -> bool: + """True if a SystemMessage reports a terminal credential/access rejection. + + A typed 403 whose error text is about authentication or model access is + as terminal as a 401: the credential (or its model scope) is dead and + retrying cannot help. Adopting this PR's OrcaRouter provider, the relay + answers 403 ``model_access_denied`` for a key that lacks access to a + model and 401 for a revoked key — both must mark the credential for + reauthentication instead of being retried or surfaced as a crash. + + A typed 401 is handled by ``_is_auth_failure`` directly; this helper adds + the typed-403 case. Only a typed integer 403 triggers it, so an untyped + system notice is never misread as a credential failure. + """ + data = getattr(message, "data", None) + if not isinstance(data, dict): + return False + if data.get("error_status") != 403: + return False + text = f"{data.get('error', '')} {data.get('message', '')}".lower() + return any( + marker in text + for marker in ("authentication", "unauthorized", "credential", "model_access") ) diff --git a/vulnhunter-agent/tests/conftest.py b/vulnhunter-agent/tests/conftest.py index 8a7f730..bbb71da 100644 --- a/vulnhunter-agent/tests/conftest.py +++ b/vulnhunter-agent/tests/conftest.py @@ -20,6 +20,7 @@ IssuesConfig, LoggingConfig, OAuthConfig, + OrcaRouterConfig, PublishConfig, RepoPropertiesConfig, SandboxConfig, @@ -38,12 +39,14 @@ def clean_env(monkeypatch: pytest.MonkeyPatch) -> None: """Strip env vars that would otherwise leak into the test process. - VULNHUNT_*: agent config, must start from a known state. + - ORCA_* / ORCAROUTER_*: OrcaRouter provider config, including a real + credential that must never leak into a test's credential resolution. - *_PROXY / NO_PROXY: httpx auto-picks these up; if the dev shell has a SOCKS proxy configured, every respx-mocked test would try to route through it and fail with "socksio not installed". """ for key in list(os.environ): - if key.startswith("VULNHUNT_") or key.upper() in ( + if key.startswith(("VULNHUNT_", "ORCA_", "ORCAROUTER_")) or key.upper() in ( "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", @@ -117,6 +120,7 @@ def _build_agent_config(**overrides: Any) -> AgentConfig: default_lifetime_seconds=3600, http_timeout_seconds=30, ), + "orcarouter": OrcaRouterConfig(), "tls": TLSConfig(ssl_cert_path=""), "sandbox": SandboxConfig( enabled=True, diff --git a/vulnhunter-agent/tests/test_main.py b/vulnhunter-agent/tests/test_main.py index 19ebcc1..c3b5ff4 100644 --- a/vulnhunter-agent/tests/test_main.py +++ b/vulnhunter-agent/tests/test_main.py @@ -540,6 +540,7 @@ def _stub_main_environment_for_args( IssuesConfig, LoggingConfig, OAuthConfig, + OrcaRouterConfig, PublishConfig, SandboxConfig, ScanConfig, @@ -554,6 +555,7 @@ def fake_load_config(*_a: object, **_k: object) -> AgentConfig: aws_region="us-east-1", model="claude-opus-4-8", ), + orcarouter=OrcaRouterConfig(), oauth=OAuthConfig( token_endpoint="https://oauth.example.com/token", client_id="x", @@ -1218,7 +1220,7 @@ def test_missing_mode_emits_friendly_error( assert excinfo.value.code == 2 err = capsys.readouterr().err assert "--mode is required" in err - assert "scan or verify" in err + assert "scan, verify, or orcarouter" in err def test_verify_mode_dispatches_to_verify_amain( self, @@ -1252,6 +1254,7 @@ async def fake_run_verify(**kwargs: Any) -> int: IssuesConfig, LoggingConfig, OAuthConfig, + OrcaRouterConfig, PublishConfig, SandboxConfig, ScanConfig, @@ -1266,6 +1269,7 @@ async def fake_run_verify(**kwargs: Any) -> int: aws_region="us-east-1", model="claude-opus-4-8", ), + orcarouter=OrcaRouterConfig(), oauth=OAuthConfig( token_endpoint="https://o.example", client_id="x", diff --git a/vulnhunter-agent/tests/test_orcarouter_cli.py b/vulnhunter-agent/tests/test_orcarouter_cli.py new file mode 100644 index 0000000..cb611bf --- /dev/null +++ b/vulnhunter-agent/tests/test_orcarouter_cli.py @@ -0,0 +1,164 @@ +"""Tests for the ``--mode=orcarouter`` CLI surface. + +Both authentication entries must be discoverable and independently usable: +``store-api-key`` (paste an existing key) and ``login`` (OAuth 2.0 + PKCE), +plus ``status`` / ``clear`` / ``models``. +""" + +from __future__ import annotations + +import json + +import pytest + +from agent.__main__ import main + + +@pytest.fixture +def creds(tmp_path, monkeypatch): + path = tmp_path / "orca.json" + monkeypatch.setenv("ORCA_CREDENTIALS_FILE", str(path)) + # Keep discovery off the network: point the API base at an unroutable + # loopback port so the catalog falls back to the verified seed. + monkeypatch.setenv("ORCA_API_BASE_URL", "http://127.0.0.1:1/v1") + return path + + +def test_status_without_credential_is_reported(creds, capsys) -> None: + rc = main(["--mode=orcarouter", "--orcarouter-action=status"]) + assert rc == 0 + assert "No usable OrcaRouter credential" in capsys.readouterr().out + + +def test_store_api_key_then_status_masks_secret(creds, capsys) -> None: + rc = main( + ["--mode=orcarouter", "--orcarouter-action=store-api-key", + "--api-key", "sk-orca-CLIKEY000000000000000000"] + ) + assert rc == 0 + out = capsys.readouterr().out + assert "sk-orca-CLIKEY000000000000000000" not in out + assert "sk-orca-...0000" in out + + main(["--mode=orcarouter", "--orcarouter-action=status"]) + status = capsys.readouterr().out + assert "sk-orca-CLIKEY000000000000000000" not in status + assert "method: api_key" in status + + +def test_status_json_is_machine_readable(creds, capsys) -> None: + main(["--mode=orcarouter", "--orcarouter-action=store-api-key", + "--api-key", "sk-orca-JSON00000000000000000000"]) + capsys.readouterr() + main(["--mode=orcarouter", "--orcarouter-action=status", "--json"]) + payload = json.loads(capsys.readouterr().out) + assert payload["status"] == "ok" + assert payload["method"] == "api_key" + assert payload["generation"] == 1 + assert "api_key" not in payload # only the masked form is emitted + + +def test_clear_removes_credential(creds, capsys) -> None: + main(["--mode=orcarouter", "--orcarouter-action=store-api-key", + "--api-key", "sk-orca-CLEARME0000000000000000"]) + capsys.readouterr() + assert main(["--mode=orcarouter", "--orcarouter-action=clear"]) == 0 + assert "Removed" in capsys.readouterr().out + assert creds.exists() is False + + +def test_models_falls_back_to_seed_when_offline(creds, capsys) -> None: + rc = main(["--mode=orcarouter", "--orcarouter-action=models", + "--capability", "chat", "--json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["degraded"] is True + assert payload["source"] == "seed" + assert payload["count"] >= 5 + ids = {m["id"] for m in payload["models"]} + assert "openai/gpt-5.5" in ids + + +def test_models_live_uses_catalog_api(creds, capsys, monkeypatch) -> None: + import agent.orcarouter as orca + + def fake_discover(config, api_key, *, capability="chat", http_client=None): + return orca.ModelCatalog( + models=( + orca.ModelInfo(id="vendor/live-model", + endpoint_types=("openai",), + input_modalities=("text",)), + ), + source="live", + ) + + monkeypatch.setattr(orca, "discover_catalog", fake_discover) + rc = main(["--mode=orcarouter", "--orcarouter-action=models", "--json"]) + assert rc == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["source"] == "live" + assert payload["degraded"] is False + assert [m["id"] for m in payload["models"]] == ["vendor/live-model"] + + +def test_login_uses_the_oauth_adapter(creds, capsys, monkeypatch) -> None: + import agent.orcarouter as orca + + used: dict = {} + + class _FakeProvider: + def __init__(self, *a, **k): + pass + + def connect_oauth(self): + used["called"] = True + return orca.CredentialResult( + api_key="sk-orca-FROMOAUTH00000000000000", + method=orca.ConnectMethod.OAUTH, + scope="api", + source="oauth", + ) + + monkeypatch.setattr(orca, "ConnectCredentialProvider", _FakeProvider) + rc = main(["--mode=orcarouter", "--orcarouter-action=login"]) + assert rc == 0 + assert used.get("called") is True + out = capsys.readouterr().out + assert "login complete" in out + assert "sk-orca-FROMOAUTH00000000000000" not in out + + +def test_orcarouter_mode_rejects_positional(creds) -> None: + with pytest.raises(SystemExit) as excinfo: + main(["--mode=orcarouter", "https://github.com/o/r"]) + assert excinfo.value.code == 2 + + +def test_help_lists_both_entries(capsys) -> None: + with pytest.raises(SystemExit): + main(["--help"]) + text = capsys.readouterr().out + assert "--orcarouter-action" in text + assert "login" in text + assert "store-api-key" in text + + +def test_models_rejects_incompatible_selected_model(creds, capsys, monkeypatch) -> None: + """A selected model not in the capability-filtered set fails loudly.""" + import agent.orcarouter as orca + + monkeypatch.setattr( + orca, "discover_catalog", + lambda config, key, capability="chat", http_client=None: orca.ModelCatalog( + models=(orca.ModelInfo(id="vendor/text-only", + endpoint_types=("openai",), + input_modalities=("text",)),), + source="live", + ), + ) + rc = main(["--mode=orcarouter", "--orcarouter-action=models", + "--capability", "image_input", "--model", "vendor/text-only"]) + assert rc == 1 + err = capsys.readouterr().err + assert "not available for capability" in err + assert "vendor/text-only" in err diff --git a/vulnhunter-agent/tests/test_orcarouter_connect.py b/vulnhunter-agent/tests/test_orcarouter_connect.py new file mode 100644 index 0000000..c8b086c --- /dev/null +++ b/vulnhunter-agent/tests/test_orcarouter_connect.py @@ -0,0 +1,542 @@ +"""Tests for the OrcaRouter credential seam and its two adapters. + +Covers: API-key save/read/clear/mask; PKCE verifier/challenge/state; the +authorize URL; the exchange path and body; success persistence; denial; +Flow A state mismatch; code reuse/expiry (403); scope downgrade; corrupt-key +terminal classification; exact-account 401 reconnect; 429; and network +failure — plus the invariant that no verifier or key leaks into logs/errors. +""" + +from __future__ import annotations + +import logging + +import pytest + +from agent.config import OrcaRouterConfig, TLSConfig +from agent.orcarouter import ( + ApiKeyConnect, + ConnectCredentialProvider, + ConnectMethod, + CredentialStore, + CredentialUnavailable, + OAuthConnect, + OAuthFlowError, + ReauthRequired, + challenge_for, + cli_base_url, + mask_key, + new_state, + new_verifier, + state_matches, + validated_origin, +) + +FAKE_KEY = "sk-orca-TESTKEY1234567890abcdef" +VERIFIER = "verifier-sentinel-DO-NOT-LEAK" + + +class FakeReceiver: + """Stand-in for the loopback listener in Flow A tests.""" + + def __init__(self, params: dict | None) -> None: + self.port = 51234 + self._params = params + self.started = False + self.stopped = False + + def start(self) -> None: + self.started = True + + def wait(self, timeout: float) -> dict | None: + return self._params + + def stop(self) -> None: + self.stopped = True + + +def make_oauth(**kwargs) -> OAuthConnect: + params = { + "auth_base_url": "https://www.orcarouter.ai", + "app_name": "Test App", + "open_browser": lambda url: None, + } + params.update(kwargs) + return OAuthConnect(**params) + + +def store_at(tmp_path) -> CredentialStore: + return CredentialStore(tmp_path / "orca.json") + + +def provider_at(tmp_path, **cfg) -> ConnectCredentialProvider: + config = OrcaRouterConfig(credentials_file=str(tmp_path / "orca.json"), **cfg) + return ConnectCredentialProvider(config, tls=TLSConfig(ssl_cert_path="")) + + +# --------------------------------------------------------------------------- +# PKCE primitives +# --------------------------------------------------------------------------- + + +def test_verifier_is_fresh_crypto_randomness_per_attempt() -> None: + a, b = new_verifier(), new_verifier() + assert a and b and a != b + assert "=" not in a # base64url, unpadded + + +def test_challenge_is_unpadded_base64url_sha256() -> None: + import base64 + import hashlib + + verifier = new_verifier() + expected = ( + base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()) + .decode() + .rstrip("=") + ) + assert challenge_for(verifier) == expected + assert "=" not in challenge_for(verifier) + + +def test_state_is_fresh_and_matches_constant_time_contract() -> None: + s = new_state() + assert state_matches(s, s) is True + assert state_matches(s, "other") is False + assert state_matches(s, None) is False + assert state_matches("", "") is False + + +# --------------------------------------------------------------------------- +# Origins +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "url", + [ + "https://www.orcarouter.ai", + "https://api.orcarouter.ai/v1", + "http://127.0.0.1:8080", + "http://localhost:3000", + ], +) +def test_validated_origin_accepts_https_and_loopback_http(url: str) -> None: + assert validated_origin(url, what="x") == url.rstrip("/") + + +@pytest.mark.parametrize( + "url", + ["http://evil.example.com", "ftp://api.orcarouter.ai", "https://user:pw@x.ai", ""], +) +def test_validated_origin_rejects_bad_inputs(url: str) -> None: + with pytest.raises(Exception): + validated_origin(url, what="x") + + +def test_cli_base_url_strips_trailing_v1_only() -> None: + assert cli_base_url("https://api.orcarouter.ai/v1") == "https://api.orcarouter.ai" + assert cli_base_url("https://api.orcarouter.ai") == "https://api.orcarouter.ai" + assert cli_base_url("http://127.0.0.1:9000/v1") == "http://127.0.0.1:9000" + + +# --------------------------------------------------------------------------- +# API-key adapter +# --------------------------------------------------------------------------- + + +def test_api_key_adapter_stores_reads_and_masks(tmp_path) -> None: + store = store_at(tmp_path) + record = ApiKeyConnect(store).store(FAKE_KEY) + assert record.api_key == FAKE_KEY + assert record.method == ConnectMethod.API_KEY.value + assert store.load().api_key == FAKE_KEY + assert mask_key(FAKE_KEY) == "sk-orca-...cdef" + assert FAKE_KEY not in mask_key(FAKE_KEY) + + +def test_api_key_adapter_clear(tmp_path) -> None: + store = store_at(tmp_path) + ApiKeyConnect(store).store(FAKE_KEY) + store.clear() + assert store.load() is None + + +def test_api_key_adapter_rejects_empty() -> None: + from agent.orcarouter import CredentialError + + with pytest.raises(CredentialError): + ApiKeyConnect(CredentialStore("/tmp/never.json")).store(" ") + + +def test_credential_file_permissions_are_owner_only(tmp_path) -> None: + import os + import stat + + store = store_at(tmp_path) + ApiKeyConnect(store).store(FAKE_KEY) + mode = stat.S_IMODE(os.stat(store.path).st_mode) + assert mode == 0o600 + + +# --------------------------------------------------------------------------- +# Both adapters yield the same credential shape; downstream is source-agnostic +# --------------------------------------------------------------------------- + + +def test_both_adapters_return_same_credential_result(tmp_path) -> None: + config = OrcaRouterConfig(credentials_file=str(tmp_path / "orca.json")) + provider = ConnectCredentialProvider(config) + + # API-key adapter. + key_result = provider.store_api_key(FAKE_KEY) + # OAuth adapter (Flow B so the adapter supplies its own state), returning + # the SAME kind of key from a fake exchange. + oauth_provider = ConnectCredentialProvider( + OrcaRouterConfig(credentials_file=str(tmp_path / "oauth.json"), flow="b"), + post=lambda url, body, t: (200, {"key": FAKE_KEY, "user_id": "7", "scope": "api"}), + code_provider=lambda: "the-code", + ) + oauth_result = oauth_provider.connect_oauth() + + # Both adapters produce the identical credential value... + assert key_result.api_key == oauth_result.api_key == FAKE_KEY + # ...from different user-facing entries... + assert key_result.method != oauth_result.method + # ...and the downstream token interface cannot tell them apart. + assert oauth_provider.get_valid_token() == provider.get_valid_token() + assert isinstance(provider.get_valid_token(), str) + + +def test_connect_seam_defines_and_uses_two_adapters(tmp_path) -> None: + provider = provider_at(tmp_path) + assert isinstance(provider._api_adapter, ApiKeyConnect) + assert isinstance(provider._oauth_adapter, OAuthConnect) + + +# --------------------------------------------------------------------------- +# Flow A +# --------------------------------------------------------------------------- + + +def test_flow_a_happy_path_persists_and_uses_auth_origin(tmp_path) -> None: + calls: list[tuple[str, dict, float]] = [] + + def fake_post(url, body, timeout): + calls.append((url, body, timeout)) + return (200, {"key": FAKE_KEY, "user_id": "42", "scope": "api"}) + + receiver = FakeReceiver({"code": "the-code", "state": "REPLACED"}) + oauth = make_oauth(flow="a", post=fake_post, receiver_factory=lambda: receiver) + + # Flow A needs the state the adapter generated, so capture it from the + # authorize URL and echo it back through the receiver. + captured: dict = {} + orig_url = oauth.authorize_url + + def capture_url(*, callback_url, challenge, state): + captured["state"] = state + captured["challenge"] = challenge + captured["callback_url"] = callback_url + return orig_url(callback_url=callback_url, challenge=challenge, state=state) + + oauth.authorize_url = capture_url # type: ignore[assignment] + receiver._params = {"code": "the-code", "state": ""} + # Patch wait() to echo the real state. + def wait(timeout): + return {"code": "the-code", "state": captured["state"]} + + receiver.wait = wait # type: ignore[assignment] + + result = oauth.connect() + assert result["key"] == FAKE_KEY + assert receiver.started and receiver.stopped + # Exchange hits the AUTH origin at the fixed /api/v1/auth/keys path. + url, body, _ = calls[0] + assert url == "https://www.orcarouter.ai/api/v1/auth/keys" + assert "/v1/auth/keys" in url # the correct, documented path + assert body["code"] == "the-code" + assert body["code_challenge_method"] == "S256" + assert "code_verifier" in body + # The callback_url is loopback with the ephemeral port. + assert captured["callback_url"] == "http://127.0.0.1:51234/cb" + # Only the S256 challenge — never the verifier — is on the authorize URL. + assert captured["challenge"] == challenge_for(body["code_verifier"]) + assert body["code_verifier"] != captured["challenge"] + + +def test_authorize_url_uses_auth_origin_and_s256() -> None: + oauth = make_oauth() + url = oauth.authorize_url(callback_url="oob", challenge="CH", state="ST") + assert url.startswith("https://www.orcarouter.ai/auth?") + assert "code_challenge_method=S256" in url + assert "callback_url=oob" in url + assert "scope=api" in url + assert "code_verifier" not in url + + +def test_flow_a_state_mismatch_is_refused(tmp_path) -> None: + called = {"n": 0} + + def fake_post(url, body, timeout): + called["n"] += 1 + return (200, {"key": FAKE_KEY}) + + receiver = FakeReceiver({"code": "c", "state": "attacker-state"}) + oauth = make_oauth(flow="a", post=fake_post, receiver_factory=lambda: receiver) + with pytest.raises(OAuthFlowError, match="state did not match"): + oauth.connect() + assert called["n"] == 0 # code never redeemed + + +def test_flow_a_declined_is_reported_and_stops(tmp_path) -> None: + oauth = make_oauth( + flow="a", + post=lambda *a: pytest.fail("must not exchange on denial"), + receiver_factory=lambda: FakeReceiver({"error": "access_denied", "state": ""}), + ) + # Echo the correct state so we reach the error branch. + captured: dict = {} + orig = oauth.authorize_url + + def capture_url(*, callback_url, challenge, state): + captured["state"] = state + return orig(callback_url=callback_url, challenge=challenge, state=state) + + oauth.authorize_url = capture_url # type: ignore[assignment] + receiver = FakeReceiver(None) + receiver.wait = lambda t: {"error": "access_denied", "state": captured["state"]} # type: ignore[assignment] + oauth._receiver_factory = lambda: receiver + with pytest.raises(OAuthFlowError, match="denied"): + oauth.connect() + + +def test_flow_a_timeout_does_not_hang(tmp_path) -> None: + oauth = make_oauth( + flow="a", + post=lambda *a: pytest.fail("must not exchange on timeout"), + receiver_factory=lambda: FakeReceiver(None), # wait() returns None + ) + with pytest.raises(OAuthFlowError, match="timed out"): + oauth.connect() + + +# --------------------------------------------------------------------------- +# Flow B +# --------------------------------------------------------------------------- + + +def test_flow_b_uses_oob_callback_and_s256() -> None: + oauth = make_oauth(flow="b", code_provider=lambda: "the-code") + seen: dict = {} + orig = oauth.authorize_url + + def capture(*, callback_url, challenge, state): + seen["callback_url"] = callback_url + return orig(callback_url=callback_url, challenge=challenge, state=state) + + oauth.authorize_url = capture # type: ignore[assignment] + oauth._post = lambda url, body, t: ( + (200, {"key": FAKE_KEY, "user_id": "1", "scope": "api"}) + ) + result = oauth.connect() + assert result["key"] == FAKE_KEY + assert seen["callback_url"] == "oob" + + +def test_flow_b_empty_code_is_cancelled() -> None: + oauth = make_oauth(flow="b", code_provider=lambda: " ") + with pytest.raises(OAuthFlowError, match="cancelled"): + oauth.connect() + + +# --------------------------------------------------------------------------- +# Exchange error taxonomy +# --------------------------------------------------------------------------- + + +def test_exchange_403_is_terminal_unusable_code() -> None: + oauth = make_oauth(flow="b", code_provider=lambda: "used", post=lambda *a: (403, {})) + with pytest.raises(OAuthFlowError, match="rejected the authorization code"): + oauth.connect() + + +def test_exchange_400_is_a_downgrade_refusal() -> None: + oauth = make_oauth(flow="b", code_provider=lambda: "c", post=lambda *a: (400, {})) + with pytest.raises(OAuthFlowError, match="code_challenge_method"): + oauth.connect() + + +def test_exchange_429_is_actionable() -> None: + oauth = make_oauth(flow="b", code_provider=lambda: "c", post=lambda *a: (429, {})) + with pytest.raises(OAuthFlowError, match="rate-limited"): + oauth.connect() + + +def test_exchange_network_failure_is_wrapped_without_leaking() -> None: + def boom(url, body, timeout): + raise ConnectionError(f"connect failed for {url}?code=SECRETCODE") + + oauth = make_oauth(flow="b", code_provider=lambda: "SECRETCODE", post=boom) + with pytest.raises(OAuthFlowError) as excinfo: + oauth.connect() + assert "SECRETCODE" not in str(excinfo.value) + + +def test_scope_downgrade_is_reported() -> None: + oauth = make_oauth( + flow="b", + code_provider=lambda: "c", + scope="connector", + post=lambda *a: (200, {"key": FAKE_KEY, "scope": "api"}), + ) + # asked for connector, granted api → narrower than requested → refuse loudly. + with pytest.raises(OAuthFlowError, match="granted scope"): + oauth.connect() + + +def test_exchange_uses_read_back_scope_not_requested_scope() -> None: + oauth = make_oauth( + flow="b", + code_provider=lambda: "c", + post=lambda *a: (200, {"key": FAKE_KEY, "user_id": "9", "scope": "api"}), + ) + assert oauth.connect()["scope"] == "api" + + +# --------------------------------------------------------------------------- +# Secrets must not leak +# --------------------------------------------------------------------------- + + +def test_verifier_never_appears_in_logs_or_errors(caplog) -> None: + import logging as _logging + + captured: dict = {} + oauth = make_oauth(flow="b", code_provider=lambda: "c") + + orig_url = oauth.authorize_url + + def capture(*, callback_url, challenge, state): + captured["challenge"] = challenge + return orig_url(callback_url=callback_url, challenge=challenge, state=state) + + oauth.authorize_url = capture # type: ignore[assignment] + + def fake_post(url, body, timeout): + captured["verifier"] = body["code_verifier"] + return (200, {"key": FAKE_KEY, "user_id": "1", "scope": "api"}) + + oauth._post = fake_post + with caplog.at_level(_logging.DEBUG): + oauth.connect() + + verifier = captured["verifier"] + log_text = "\n".join(r.getMessage() for r in caplog.records) + assert verifier not in log_text + assert verifier not in captured["challenge"] + assert FAKE_KEY not in log_text + + +def test_key_never_appears_in_status_or_errors(tmp_path) -> None: + provider = provider_at(tmp_path) + provider.store_api_key(FAKE_KEY) + + seen: list[str] = [] + + class _Capture(logging.Handler): + def emit(self, record: logging.LogRecord) -> None: + seen.append(record.getMessage()) + + logger = logging.getLogger("agent.orcarouter.connect") + handler = _Capture() + logger.addHandler(handler) + try: + provider.resolve() + finally: + logger.removeHandler(handler) + assert all(FAKE_KEY not in m for m in seen) + + +# --------------------------------------------------------------------------- +# Resolution, terminal classification, generation-safe 401 +# --------------------------------------------------------------------------- + + +def test_resolve_prefers_store_over_config(tmp_path) -> None: + provider = provider_at(tmp_path, api_key="sk-orca-fromconfig") + provider.store_api_key(FAKE_KEY) + assert provider.get_valid_token() == FAKE_KEY + + +def test_resolve_falls_back_to_config_key(tmp_path) -> None: + provider = provider_at(tmp_path, api_key="sk-orca-fromconfig") + assert provider.get_valid_token() == "sk-orca-fromconfig" + + +def test_resolve_without_any_credential_is_unavailable(tmp_path) -> None: + with pytest.raises(CredentialUnavailable): + provider_at(tmp_path).resolve() + + +def test_corrupt_oauth_key_is_terminal(tmp_path) -> None: + store = store_at(tmp_path) + from agent.orcarouter import CredentialRecord + + store.save(CredentialRecord(api_key="garbage-not-a-key", method="oauth")) + provider = ConnectCredentialProvider( + OrcaRouterConfig(credentials_file=str(tmp_path / "orca.json")) + ) + with pytest.raises(ReauthRequired): + provider.resolve() + + +def test_needs_reauth_credential_is_not_silently_refreshed(tmp_path) -> None: + provider = provider_at(tmp_path) + provider.store_api_key(FAKE_KEY) + assert provider.note_rejection(api_key=FAKE_KEY, generation=1) is True + with pytest.raises(ReauthRequired, match="rejected"): + provider.resolve() + # Old secret is retained (never silently deleted before a new login). + assert provider.store.load().api_key == FAKE_KEY + + +def test_401_marks_only_the_exact_generation(tmp_path) -> None: + provider = provider_at(tmp_path) + provider.store_api_key(FAKE_KEY) # generation 1 + # Re-login advances the generation; a late 401 from gen 1 must not poison it. + provider.store_api_key("sk-orca-SECONDKEY0000000000abcd") # generation 2 + assert provider.note_rejection(api_key=FAKE_KEY, generation=1) is False + result = provider.resolve() + assert result.api_key == "sk-orca-SECONDKEY0000000000abcd" + assert result.needs_reauth is False + + +def test_regeneration_clears_needs_reauth(tmp_path) -> None: + provider = provider_at(tmp_path) + provider.store_api_key(FAKE_KEY) + provider.note_rejection(api_key=FAKE_KEY, generation=1) + fresh = provider.store_api_key("sk-orca-NEWKEY00000000000000abcd") + assert fresh.needs_reauth is False + assert provider.resolve().api_key == "sk-orca-NEWKEY00000000000000abcd" + + +def test_clear_removes_credential(tmp_path) -> None: + provider = provider_at(tmp_path) + provider.store_api_key(FAKE_KEY) + provider.clear() + with pytest.raises(CredentialUnavailable): + provider.resolve() + + +def test_corrupt_store_file_is_tolerated(tmp_path) -> None: + path = tmp_path / "orca.json" + path.write_text("{not json") + store = CredentialStore(path) + assert store.load() is None + + +def test_generation_increments_on_each_login(tmp_path) -> None: + provider = provider_at(tmp_path) + assert provider.store_api_key(FAKE_KEY).generation == 1 + assert provider.store_api_key("sk-orca-OTHER000000000000000000").generation == 2 diff --git a/vulnhunter-agent/tests/test_orcarouter_integration.py b/vulnhunter-agent/tests/test_orcarouter_integration.py new file mode 100644 index 0000000..486ac93 --- /dev/null +++ b/vulnhunter-agent/tests/test_orcarouter_integration.py @@ -0,0 +1,338 @@ +"""Integration tests for OrcaRouter wiring into the provider seams. + +Proves: + +* the provider is a first-class ``auth_mode`` value in config; +* both connect entries (API key + OAuth) remain independently usable; +* auth requests go ONLY to the auth origin and inference/catalog ONLY to the + API origin (never derived from one another); +* the CLI settings JSON points ANTHROPIC_BASE_URL at the API origin without + the trailing ``/v1`` and carries the key; +* the real scan session consumes the OrcaRouter credential with no + per-entry-point auth logic; +* a terminal 401/403 marks the exact credential generation for reauth and + never triggers a fake refresh. +""" + +from __future__ import annotations + +import json + +import pytest + +from agent import _llm, build_settings +from agent.config import AnthropicConfig, OrcaRouterConfig, load_config +from agent.orcarouter import ConnectCredentialProvider + +FAKE_KEY = "sk-orca-INTEGRATION000000000000" +API_BASE = "https://api.orcarouter.ai/v1" +AUTH_BASE = "https://www.orcarouter.ai" + + +@pytest.fixture +def orcarouter_config(populated_agent_config): + import dataclasses + + return dataclasses.replace( + populated_agent_config, + anthropic=AnthropicConfig( + model="deepseek/deepseek-v4-pro", + auth_mode="orcarouter", + ), + orcarouter=OrcaRouterConfig( + auth_base_url=AUTH_BASE, + api_base_url=API_BASE, + api_key=FAKE_KEY, + ), + ) + + +# --------------------------------------------------------------------------- +# Config: first-class auth_mode +# --------------------------------------------------------------------------- + + +def test_auth_mode_orcarouter_is_accepted(tmp_path, tmp_config) -> None: + path = tmp_config( + body=( + "[anthropic]\n" + 'auth_mode = "orcarouter"\n' + 'model = "deepseek/deepseek-v4-pro"\n' + "[orcarouter]\n" + 'connect = "oauth"\n' + 'flow = "b"\n' + ) + ) + config = load_config(path) + assert config.anthropic.auth_mode == "orcarouter" + assert config.orcarouter.connect == "oauth" + assert config.orcarouter.flow == "b" + # api_key stays blank for orcarouter — the credential comes from the seam, + # not from [anthropic].api_key. + assert config.anthropic.api_key == "" + + +def test_invalid_connect_and_flow_rejected(tmp_config) -> None: + with pytest.raises(ValueError, match="connect must be"): + load_config(tmp_config(body='[orcarouter]\nconnect = "magic"\n[anthropic]\nmodel="m"\n')) + with pytest.raises(ValueError, match="flow must be"): + load_config(tmp_config(body='[orcarouter]\nflow = "c"\n[anthropic]\nmodel="m"\n')) + + +def test_origins_default_and_overrides(tmp_config, monkeypatch) -> None: + # Defaults. + cfg = load_config(tmp_config(body='[anthropic]\nmodel="m"\n')) + assert cfg.orcarouter.auth_base_url == AUTH_BASE + assert cfg.orcarouter.api_base_url == API_BASE + + # Shared self-hosted fallback applies to both. + monkeypatch.setenv("ORCA_BASE_URL", "https://orca.internal.example") + cfg = load_config(tmp_config(body='[anthropic]\nmodel="m"\n')) + assert cfg.orcarouter.auth_base_url == "https://orca.internal.example" + assert cfg.orcarouter.api_base_url == "https://orca.internal.example" + + # Explicit per-origin override wins over the shared fallback. + monkeypatch.setenv("ORCA_API_BASE_URL", "https://api.self.example/v1") + cfg = load_config(tmp_config(body='[anthropic]\nmodel="m"\n')) + assert cfg.orcarouter.api_base_url == "https://api.self.example/v1" + assert cfg.orcarouter.auth_base_url == "https://orca.internal.example" + + +# --------------------------------------------------------------------------- +# Credential seam → the two adapters, same downstream +# --------------------------------------------------------------------------- + + +def test_both_entries_selectable_and_source_agnostic(tmp_path) -> None: + from agent.config import OrcaRouterConfig as C + from agent.orcarouter import ConnectCredentialProvider + + # API-key entry. + api_provider = ConnectCredentialProvider( + C(credentials_file=str(tmp_path / "a.json")) + ) + api_result = api_provider.store_api_key(FAKE_KEY) + + # OAuth entry against a fake auth server. + oauth_provider = ConnectCredentialProvider( + C(credentials_file=str(tmp_path / "b.json"), flow="b"), + code_provider=lambda: "code", + post=lambda url, body, t: (200, {"key": FAKE_KEY, "user_id": "1", "scope": "api"}), + ) + oauth_result = oauth_provider.connect_oauth() + + # Same credential shape; the request path only reads api_key. + assert api_result.api_key == oauth_result.api_key == FAKE_KEY + assert api_result.method.value == "api_key" + assert oauth_result.method.value == "oauth" + assert oauth_provider.get_valid_token() == api_provider.get_valid_token() + + +def test_auth_requests_only_go_to_auth_origin(tmp_path) -> None: + from agent.config import OrcaRouterConfig as C + + urls: list[str] = [] + + def post(url, body, t): + urls.append(url) + return (200, {"key": FAKE_KEY, "user_id": "1", "scope": "api"}) + + provider = ConnectCredentialProvider( + C(credentials_file=str(tmp_path / "c.json"), flow="b"), + code_provider=lambda: "code", + post=post, + ) + provider.connect_oauth() + assert urls == [f"{AUTH_BASE}/api/v1/auth/keys"] + # Never the (wrong) relay path. + assert all("/v1/auth/keys" not in u.replace("/api/v1/auth/keys", "") for u in urls) + + +def test_authorize_url_is_on_auth_origin() -> None: + from agent.config import OrcaRouterConfig as C + from agent.orcarouter import OAuthConnect + + oauth = OAuthConnect( + auth_base_url=AUTH_BASE, app_name="X", flow="b", + open_browser=lambda u: None, + ) + url = oauth.authorize_url(callback_url="oob", challenge="C", state="S") + assert url.startswith(f"{AUTH_BASE}/auth?") + assert "api.orcarouter.ai" not in url + + +def test_inference_and_catalog_only_go_to_api_origin() -> None: + import httpx + + from agent.orcarouter import discover_catalog + + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(str(request.url)) + return httpx.Response(200, json={"data": [ + {"id": "v/m", "supported_endpoint_types": ["openai"], + "architecture": {"input_modalities": ["text"]}}, + ]}) + + with httpx.Client(transport=httpx.MockTransport(handler)) as client: + discover_catalog(OrcaRouterConfig(), FAKE_KEY, capability="chat", http_client=client) + + assert seen and all(u.startswith(f"{API_BASE}/models") for u in seen) + assert all("www.orcarouter.ai" not in u for u in seen) + + +# --------------------------------------------------------------------------- +# CLI settings injection +# --------------------------------------------------------------------------- + + +def test_cli_settings_target_api_origin_without_v1(orcarouter_config) -> None: + settings = json.loads( + build_settings.build_claude_settings( + orcarouter_config, FAKE_KEY, model="deepseek/deepseek-v4-pro" + ) + ) + env = settings["env"] + # The CLI appends /v1 itself, so the base must NOT carry it. + assert env["ANTHROPIC_BASE_URL"] == "https://api.orcarouter.ai" + assert env["ANTHROPIC_API_KEY"] == FAKE_KEY + assert "ANTHROPIC_AUTH_TOKEN" not in env # not the Bedrock path + assert env.get("CLAUDE_CODE_USE_BEDROCK") is None + + +def test_sandbox_allowlists_both_orcarouter_origins(orcarouter_config) -> None: + settings = json.loads( + build_settings.build_claude_settings(orcarouter_config, FAKE_KEY, model="m") + ) + domains = settings["sandbox"]["network"]["allowedDomains"] + assert "api.orcarouter.ai" in domains + assert "www.orcarouter.ai" in domains + + +def test_selfhosted_origin_flows_into_cli_base(populated_agent_config) -> None: + import dataclasses + + cfg = dataclasses.replace( + populated_agent_config, + anthropic=AnthropicConfig(model="m", auth_mode="orcarouter"), + orcarouter=OrcaRouterConfig( + auth_base_url="https://orca.internal.example", + api_base_url="https://orca.internal.example/v1", + ), + ) + env = json.loads( + build_settings.build_claude_settings(cfg, "k", model="m") + )["env"] + assert env["ANTHROPIC_BASE_URL"] == "https://orca.internal.example" + + +# --------------------------------------------------------------------------- +# Scan session consumes the credential (no per-entry-point logic) +# --------------------------------------------------------------------------- + + +async def test_call_json_uses_orcarouter_credential( + monkeypatch, orcarouter_config +) -> None: + """The real one-shot call path resolves and forwards the OrcaRouter key. + + ``make_token_manager`` must return the OrcaRouter provider, its + ``get_valid_token`` must yield the configured key, and ``_send_prompt`` + must be handed that key — with no OrcaRouter-specific branch anywhere in + ``_llm`` (it only sees the shared ``TokenProvider`` interface). + """ + from agent.auth import make_token_manager + + token_manager = make_token_manager(orcarouter_config) + assert isinstance(token_manager, ConnectCredentialProvider) + assert token_manager.get_valid_token() == FAKE_KEY + + captured: dict = {} + + async def fake_send_prompt(*, model, system, user, config, auth_token, cost_tracker=None): + captured["auth_token"] = auth_token + captured["model"] = model + return '{"ok": true}' + + monkeypatch.setattr(_llm, "_send_prompt", fake_send_prompt) + + out = await _llm.call_json( + model="deepseek/deepseek-v4-pro", + system="s", + user="u", + config=orcarouter_config, + token_manager=token_manager, + backoffs=(), + ) + assert out == {"ok": True} + assert captured["auth_token"] == FAKE_KEY + assert captured["model"] == "deepseek/deepseek-v4-pro" + + +# --------------------------------------------------------------------------- +# Terminal 401/403 → exactly-one-generation needsReauth, no fake refresh +# --------------------------------------------------------------------------- + + +def test_terminal_rejection_marks_exact_generation( + monkeypatch, orcarouter_config, tmp_path +) -> None: + from agent import runner + from agent.orcarouter import CredentialStore + + creds = tmp_path / "orca.json" + store = CredentialStore(creds) + store.store_new(api_key=FAKE_KEY, method=__import__( + "agent.orcarouter", fromlist=["ConnectMethod"] + ).ConnectMethod.API_KEY) + + import dataclasses + + cfg = dataclasses.replace( + orcarouter_config, + orcarouter=dataclasses.replace( + orcarouter_config.orcarouter, credentials_file=str(creds) + ), + ) + runner._mark_credential_rejected(cfg) + + record = store.load() + assert record.needs_reauth is True + assert record.api_key == FAKE_KEY # not deleted + # A second resolve is a hard reauth requirement, never a refresh. + from agent.orcarouter import ConnectCredentialProvider, ReauthRequired + + provider = ConnectCredentialProvider(cfg.orcarouter) + with pytest.raises(ReauthRequired): + provider.resolve() + # No refresh grant exists to call — the module exposes none. + assert not hasattr(provider, "refresh") + + +def test_credential_rejection_detection_is_typed() -> None: + from agent import runner + + class _Msg: + def __init__(self, data): + self.data = data + + assert runner._is_credential_rejection(_Msg({"error_status": 403, "error": "model_access_denied"})) is True + # A typed 401 is caught by the broader auth-failure predicate. + assert runner._is_auth_failure(_Msg({"error_status": 401})) is True + # An untyped 403 is not treated as a credential failure. + assert runner._is_credential_rejection(_Msg({"error": "model_access_denied"})) is False + assert runner._is_credential_rejection(_Msg({"error_status": 500})) is False + + +# --------------------------------------------------------------------------- +# The large-context / model-tag path must not crash on vendor/model ids +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("model", ["deepseek/deepseek-v4-pro", "orcarouter/auto", "openai/gpt-5.5"]) +def test_model_tag_handles_vendor_namespaced_ids(model: str) -> None: + from agent.runner import _model_tag + + tag = _model_tag(model) + assert tag and "/" not in tag diff --git a/vulnhunter-agent/tests/test_orcarouter_live.py b/vulnhunter-agent/tests/test_orcarouter_live.py new file mode 100644 index 0000000..360fd8a --- /dev/null +++ b/vulnhunter-agent/tests/test_orcarouter_live.py @@ -0,0 +1,169 @@ +"""Live OrcaRouter checks through this PR's provider integration. + +These are the only tests that touch the network and require a real credential. +They are skipped unless ``ORCAROUTER_API_KEY`` (or ``ORCA_API_KEY``) is set, so +the default offline suite is unaffected. Each check goes through the code this +PR adds — the credential seam and the model-discovery path — never a bare HTTP +call, so a pass means the provider wiring works end to end. + +The API key is captured at import time because the shared test fixture strips +credential environment variables from individual tests. +""" + +from __future__ import annotations + +import asyncio +import os + +import pytest + +import tests.conftest as conftest +from agent import _llm +from agent.config import AnthropicConfig, OrcaRouterConfig, SandboxConfig +from agent.orcarouter import ( + ConnectCredentialProvider, + ModelCapability, + ModelCatalog, + discover_catalog, + verified_seed, +) + +_LIVE_KEY = ( + os.environ.get("ORCAROUTER_API_KEY", "").strip() + or os.environ.get("ORCA_API_KEY", "").strip() +) + +pytestmark = pytest.mark.skipif( + not _LIVE_KEY, reason="no OrcaRouter credential in the environment" +) + +# A model the workspace can actually call. Resolved from the live catalog at +# test time rather than hardcoded, so it tracks the workspace's real access. +API_BASE = "https://api.orcarouter.ai/v1" +AUTH_BASE = "https://www.orcarouter.ai" + + +def _first_callable(models) -> str: + """Return the first model the workspace key can actually call. + + Discovery returns the public catalog, which may include models this key + has no access to. A lightweight probe (through the provider's own origins) + finds a callable one; if none respond, the first chat model is returned so + the failure surfaces in the inference assertion rather than being masked. + """ + import httpx + + for model in models: + try: + response = httpx.post( + f"{API_BASE}/messages", + headers={ + "Authorization": f"Bearer {_LIVE_KEY}", + "anthropic-version": "2023-06-01", + }, + json={ + "model": model.id, + "max_tokens": 4, + "messages": [{"role": "user", "content": "hi"}], + }, + timeout=20, + ) + except httpx.HTTPError: + continue + if response.status_code == 200: + return model.id + return models[0].id + + +def _config(**overrides) -> OrcaRouterConfig: + return OrcaRouterConfig( + auth_base_url=AUTH_BASE, + api_base_url=API_BASE, + api_key=_LIVE_KEY, + credentials_file="/tmp/nonexistent-orca-live.json", + **overrides, + ) + + +def test_live_catalog_is_authoritative_and_nonempty() -> None: + catalog = discover_catalog(_config(), _LIVE_KEY, capability="chat") + assert catalog.degraded is False, f"catalog degraded: {catalog.error}" + assert catalog.source == "live" + assert len(catalog.models) > 0 + # Live results are authoritative: every returned record is marked as coming + # from the API, so no seed entry was appended to the live list. + assert all(m.source == "live" for m in catalog.models) + + +def test_live_chat_filter_returns_only_compatible_models() -> None: + catalog = discover_catalog(_config(), _LIVE_KEY, capability="chat") + chat = catalog.for_capability(ModelCapability.CHAT) + assert chat + for model in chat: + assert model.is_text_chat() + # No non-text-specialist model leaked into the chat list. + assert not ( + {"image-generation", "openai-video", "jina-rerank"} + & set(model.endpoint_types) + ) + + +def test_live_multimodal_filter_is_a_subset_of_chat() -> None: + catalog = discover_catalog(_config(), _LIVE_KEY, capability="chat") + chat = {m.id for m in catalog.for_capability(ModelCapability.CHAT)} + images = {m.id for m in catalog.for_capability(ModelCapability.IMAGE_INPUT)} + # Fail-closed: every image-input model must also be a valid chat model. + assert images <= chat + for model in catalog.for_capability(ModelCapability.IMAGE_INPUT): + assert "image" in model.input_modalities + + +def test_live_inference_through_the_provider_path() -> None: + """A real request sent through the repo's SDK path with the OrcaRouter key.""" + config = _config() + token_manager = ConnectCredentialProvider(config, tls=conftest.TLSConfig("")) + + catalog = discover_catalog(config, _LIVE_KEY, capability="chat") + chat = catalog.for_capability(ModelCapability.CHAT) + assert chat, "no callable chat model discovered" + # The catalog is a public list; a key may lack access to some entries + # (the relay answers 403 model_access_denied). Pick the first model the + # workspace can actually call so the check reflects real access. + model_id = _first_callable(chat) + + agent_config = conftest._build_agent_config( + anthropic=AnthropicConfig(model=model_id, auth_mode="orcarouter"), + orcarouter=config, + # The SDK sandbox needs bubblewrap/socat, which a CI runner may not + # have; this check exercises the provider wiring, not the sandbox. + sandbox=SandboxConfig( + enabled=False, fail_if_unavailable=False, allow_unsandboxed_commands=True + ), + ) + + async def _run(): + return await _llm.call_json( + model=model_id, + system="You reply with JSON only.", + user='Return exactly {"ok": true} and nothing else.', + config=agent_config, + token_manager=token_manager, + backoffs=(), + stage="live", + ) + + out = asyncio.run(_run()) + assert out == {"ok": True} + + +def test_live_catalog_limits_are_enforced() -> None: + catalog = discover_catalog(_config(catalog_limit=3), _LIVE_KEY, capability="chat") + assert len(catalog.models) <= 3 + + +def test_live_both_adapters_yield_a_usable_key() -> None: + """The API-key adapter produces a credential the token interface accepts.""" + config = _config() + result = ConnectCredentialProvider(config, tls=conftest.TLSConfig("")).resolve() + assert result.api_key == _LIVE_KEY + assert result.method.value == "api_key" diff --git a/vulnhunter-agent/tests/test_orcarouter_models.py b/vulnhunter-agent/tests/test_orcarouter_models.py new file mode 100644 index 0000000..4ae2bb0 --- /dev/null +++ b/vulnhunter-agent/tests/test_orcarouter_models.py @@ -0,0 +1,321 @@ +"""Tests for OrcaRouter model discovery and capability filtering. + +The dropdown bound to a real provider selector must come from the catalog +API and be filtered per capability, fail-closed on undeclared modalities, +and never degrade to free text or an unverified example list. +""" + +from __future__ import annotations + +import httpx +import pytest + +from agent.config import OrcaRouterConfig +from agent.orcarouter import ( + ModelCapability, + ModelInfo, + ModelCatalog, + discover_catalog, + parse_catalog, + verified_seed, +) + +# A fixture catalog covering every capability the verifier asks for. +FIXTURE_MODELS = [ + # text-only chat + {"id": "vendor/text-only", "supported_endpoint_types": ["openai"], + "architecture": {"input_modalities": ["text"], "output_modalities": ["text"]}, + "context_length": 128000}, + # image-input chat + {"id": "vendor/vision", "supported_endpoint_types": ["openai", "anthropic"], + "architecture": {"input_modalities": ["text", "image"], "output_modalities": ["text"]}}, + # audio-input chat + {"id": "vendor/audio", "supported_endpoint_types": ["openai"], + "architecture": {"input_modalities": ["text", "audio"]}}, + # embedding + {"id": "vendor/embed", "supported_endpoint_types": ["embeddings"], + "architecture": {"input_modalities": ["text"]}}, + # image generation + {"id": "vendor/imagegen", "supported_endpoint_types": ["image-generation"]}, + # video generation + {"id": "vendor/video", "supported_endpoint_types": ["openai-video"]}, + # rerank + {"id": "vendor/rerank", "supported_endpoint_types": ["jina-rerank"]}, + # undeclared-modality chat: must fail closed for multimodal + {"id": "vendor/unknown-modal", "supported_endpoint_types": ["openai"]}, +] + + +def fixture_catalog() -> tuple[ModelInfo, ...]: + return parse_catalog({"data": FIXTURE_MODELS}, limit=100) + + +def ids(models) -> set[str]: + return {m.id for m in models} + + +# --------------------------------------------------------------------------- +# Parsing +# --------------------------------------------------------------------------- + + +def test_parse_preserves_vendor_namespace_and_metadata() -> None: + parsed = parse_catalog( + {"data": [{ + "id": "openai/gpt-5.5", + "name": "GPT-5.5", + "context_length": 400000, + "max_completion_tokens": 128000, + "architecture": {"input_modalities": ["text", "image"]}, + "supported_endpoint_types": ["openai", "anthropic"], + }]}, + limit=10, + ) + assert parsed[0].id == "openai/gpt-5.5" # namespace untouched + assert parsed[0].context_length == 400000 + assert "image" in parsed[0].input_modalities + + +def test_parse_drops_unusable_records_and_bounds_size() -> None: + payload = {"data": [{"no_id": 1}, "nope", {"id": ""}, {"id": "ok/a"}]} + assert ids(parse_catalog(payload, limit=10)) == {"ok/a"} + many = {"data": [{"id": f"v/m{i}"} for i in range(50)]} + assert len(parse_catalog(many, limit=5)) == 5 + + +def test_parse_tolerates_missing_shape() -> None: + assert parse_catalog({}, limit=5) == () + assert parse_catalog({"data": "nope"}, limit=5) == () + assert parse_catalog(None, limit=5) == () + + +def test_parse_reads_top_provider_fallbacks() -> None: + parsed = parse_catalog( + {"data": [{"id": "v/x", "supported_endpoint_types": ["openai"], + "top_provider": {"context_length": 999, "max_completion_tokens": 111}, + "architecture": {"input_modalities": ["text"]}}]}, + limit=5, + ) + assert parsed[0].context_length == 999 + assert parsed[0].max_completion_tokens == 111 + + +# --------------------------------------------------------------------------- +# Capability filtering (fail-closed) +# --------------------------------------------------------------------------- + + +def test_chat_capability_excludes_non_text_endpoints() -> None: + models = parse_catalog({"data": FIXTURE_MODELS}, limit=100) + chat = ids(ModelCatalog(models, "live").for_capability(ModelCapability.CHAT)) + assert "vendor/text-only" in chat + assert "vendor/vision" in chat + for excluded in ( + "vendor/embed", + "vendor/imagegen", + "vendor/video", + "vendor/rerank", + ): + assert excluded not in chat + + +def test_embedding_capability_strict() -> None: + models = parse_catalog({"data": FIXTURE_MODELS}, limit=100) + assert ids(ModelCatalog(models, "live").for_capability(ModelCapability.EMBEDDING)) == { + "vendor/embed" + } + + +def test_image_generation_video_and_rerank_are_strict() -> None: + models = parse_catalog({"data": FIXTURE_MODELS}, limit=100) + cat = ModelCatalog(models, "live") + assert ids(cat.for_capability(ModelCapability.IMAGE_GENERATION)) == {"vendor/imagegen"} + assert ids(cat.for_capability(ModelCapability.VIDEO_GENERATION)) == {"vendor/video"} + assert ids(cat.for_capability(ModelCapability.RERANK)) == {"vendor/rerank"} + + +def test_multimodal_fails_closed_on_undeclared_capability() -> None: + models = parse_catalog({"data": FIXTURE_MODELS}, limit=100) + cat = ModelCatalog(models, "live") + image = ids(cat.for_capability(ModelCapability.IMAGE_INPUT)) + assert "vendor/vision" in image + # A model with no architecture block must NOT be offered for images. + assert "vendor/unknown-modal" not in image + # A text-only model must not be offered either. + assert "vendor/text-only" not in image + # Every image-capable model must also be a valid chat model. + chat = ids(cat.for_capability(ModelCapability.CHAT)) + assert image <= chat + + +def test_audio_input_requires_declared_audio_modality() -> None: + models = parse_catalog({"data": FIXTURE_MODELS}, limit=100) + cat = ModelCatalog(models, "live") + assert ids(cat.for_capability(ModelCapability.AUDIO_INPUT)) == {"vendor/audio"} + + +def test_empty_endpoint_types_prove_nothing() -> None: + info = ModelInfo(id="v/mystery") + assert info.supports(ModelCapability.CHAT) is False + assert info.supports(ModelCapability.IMAGE_INPUT) is False + + +def test_selector_options_change_when_capability_changes() -> None: + models = parse_catalog({"data": FIXTURE_MODELS}, limit=100) + cat = ModelCatalog(models, "live") + text_options = {m.id for m in cat.for_capability(ModelCapability.CHAT)} + image_options = {m.id for m in cat.for_capability(ModelCapability.IMAGE_INPUT)} + assert image_options < text_options # strictly narrower + + +# --------------------------------------------------------------------------- +# Live discovery +# --------------------------------------------------------------------------- + + +def _client(handler) -> httpx.Client: + return httpx.Client(transport=httpx.MockTransport(handler)) + + +def test_fetch_hits_api_origin_with_bearer_and_capability() -> None: + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["url"] = str(request.url) + seen["auth"] = request.headers.get("authorization") + return httpx.Response(200, json={"data": FIXTURE_MODELS}) + + config = OrcaRouterConfig() + with _client(handler) as client: + catalog = discover_catalog(config, "sk-orca-fake", capability="chat", http_client=client) + + assert catalog.source == "live" and catalog.degraded is False + assert seen["url"].startswith("https://api.orcarouter.ai/v1/models") + assert "capability=chat" in seen["url"] + assert seen["auth"] == "Bearer sk-orca-fake" + assert len(catalog.models) == len(FIXTURE_MODELS) + + +def test_live_catalog_is_authoritative_seed_not_mixed_in() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"data": FIXTURE_MODELS}) + + with _client(handler) as client: + catalog = discover_catalog( + OrcaRouterConfig(), "sk-orca-fake", capability="chat", http_client=client + ) + live_ids = {m.id for m in catalog.models} + assert live_ids.isdisjoint({m.id for m in verified_seed()}) + assert all(m.source == "live" for m in catalog.models) + + +@pytest.mark.parametrize( + "response", + [ + httpx.Response(401, json={"error": "bad key"}), + httpx.Response(500, json={}), + httpx.Response(200, text="not json"), + httpx.Response(200, json={"data": []}), + ], +) +def test_discovery_failure_falls_back_to_verified_seed(response: httpx.Response) -> None: + def handler(request: httpx.Request) -> httpx.Response: + return response + + with _client(handler) as client: + catalog = discover_catalog( + OrcaRouterConfig(), "sk-orca-fake", capability="chat", http_client=client + ) + assert catalog.source == "seed" + assert catalog.degraded is True + assert catalog.error + assert catalog.models # usable, not empty + assert all(m.source == "seed" for m in catalog.models) + + +def test_network_error_is_degraded_and_secret_free() -> None: + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("boom sk-orca-SECRET") + + with _client(handler) as client: + catalog = discover_catalog( + OrcaRouterConfig(), "sk-orca-SECRET", capability="chat", http_client=client + ) + assert catalog.degraded + assert "sk-orca-SECRET" not in catalog.error + + +def test_catalog_disabled_returns_seed_without_network() -> None: + def handler(request: httpx.Request) -> httpx.Response: + pytest.fail("must not hit the network when discovery is disabled") + + with _client(handler) as client: + catalog = discover_catalog( + OrcaRouterConfig(catalog_enabled=False), "k", + capability="chat", http_client=client, + ) + assert catalog.source == "seed" and catalog.degraded + + +def test_fetch_preserves_context_and_reasoning_metadata() -> None: + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"data": [{ + "id": "openai/gpt-5.5", + "context_length": 400000, + "reasoning": True, + "reasoning_efforts": ["low", "medium", "high", "xhigh"], + "architecture": {"input_modalities": ["text", "image"]}, + "supported_endpoint_types": ["openai", "anthropic"], + }]}) + + with _client(handler) as client: + catalog = discover_catalog(OrcaRouterConfig(), "k", capability="chat", http_client=client) + model = catalog.models[0] + assert model.reasoning_efforts == ("low", "medium", "high", "xhigh") + assert model.context_length == 400000 + assert "image" in model.input_modalities + + +# --------------------------------------------------------------------------- +# Verified seed integrity +# --------------------------------------------------------------------------- + + +def test_seed_contains_the_documented_models() -> None: + seed = {m.id for m in verified_seed()} + assert seed == { + "openai/gpt-5.5", + "anthropic/claude-opus-4.8", + "google/gemini-3.5-flash", + "deepseek/deepseek-v4-pro", + "orcarouter/auto", + } + + +def test_seed_retains_gpt55_reasoning_ladder_and_modalities() -> None: + gpt = next(m for m in verified_seed() if m.id == "openai/gpt-5.5") + assert gpt.reasoning is True + assert gpt.reasoning_efforts == ("low", "medium", "high", "xhigh") + assert "image" in gpt.input_modalities + assert gpt.context_length == 400000 + + +def test_seed_models_are_offered_for_chat() -> None: + cat = ModelCatalog(verified_seed(), "seed", degraded=True) + assert len(cat.for_capability(ModelCapability.CHAT)) == len(verified_seed()) + + +def test_incompatible_selection_is_cleared() -> None: + """When the capability changes, a now-incompatible selection must clear.""" + models = parse_catalog({"data": FIXTURE_MODELS}, limit=100) + cat = ModelCatalog(models, "live") + + def select(model_id: str, capability: ModelCapability) -> str | None: + options = cat.for_capability(capability) + return model_id if any(m.id == model_id for m in options) else None + + assert select("vendor/text-only", ModelCapability.CHAT) == "vendor/text-only" + # Switching to image input invalidates the text-only selection. + assert select("vendor/text-only", ModelCapability.IMAGE_INPUT) is None + # An image model survives the switch. + assert select("vendor/vision", ModelCapability.IMAGE_INPUT) == "vendor/vision"