mirror of
https://github.com/usestrix/strix.git
synced 2026-08-16 17:27:26 +02:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
399c15627b | ||
|
|
dce70c643a |
@@ -315,6 +315,18 @@ strix auth status # show the active sign-in
|
||||
strix auth logout # forget the sign-in
|
||||
```
|
||||
|
||||
#### Sign in with an OpenCode subscription
|
||||
|
||||
You can also run Strix on [OpenCode Zen](https://opencode.ai/docs/zen/) credits or an [OpenCode Go](https://opencode.ai/docs/go/) subscription:
|
||||
|
||||
```bash
|
||||
strix auth login opencode # paste your API key from opencode.ai/auth
|
||||
|
||||
export STRIX_LLM="opencode/claude-sonnet-5" # opencode/<model> runs on Zen credits
|
||||
export STRIX_LLM="opencode-go/kimi-k3" # opencode-go/<model> runs on the Go subscription
|
||||
strix --target ./app-directory
|
||||
```
|
||||
|
||||
**Recommended models for best results:**
|
||||
|
||||
- [OpenAI GPT-5.4](https://openai.com/api/) - `openai/gpt-5.4`
|
||||
|
||||
@@ -280,6 +280,7 @@ ignore = [
|
||||
# a runtime ``Callable`` annotation on ``vulnerability_found_callback``.
|
||||
"strix/report/state.py" = ["TC003", "PLR0912", "PLR0915", "E501", "PERF401", "PLC0415"]
|
||||
"strix/report/usage.py" = ["PLC0415"]
|
||||
"strix/report/pricing.py" = ["PLC0415"]
|
||||
# Lazy import of strix.config.models avoids a circular dependency between the
|
||||
# report pipeline and the config layer.
|
||||
"strix/report/dedupe.py" = ["PLC0415"]
|
||||
|
||||
+27
-13
@@ -72,8 +72,32 @@ def _write_store(data: dict[str, Any]) -> None:
|
||||
write_secret_text(AUTH_PATH, json.dumps(data, indent=2))
|
||||
|
||||
|
||||
def read_provider_record(provider: str) -> dict[str, Any] | None:
|
||||
"""Raw record for *provider* from the shared subscription-auth store."""
|
||||
record = _read_store().get(provider)
|
||||
return record if isinstance(record, dict) else None
|
||||
|
||||
|
||||
def save_provider_record(provider: str, record: dict[str, Any]) -> None:
|
||||
data = _read_store()
|
||||
data[provider] = record
|
||||
_write_store(data)
|
||||
|
||||
|
||||
def remove_provider_record(provider: str) -> None:
|
||||
data = _read_store()
|
||||
if provider not in data:
|
||||
return
|
||||
del data[provider]
|
||||
if data:
|
||||
_write_store(data)
|
||||
return
|
||||
with contextlib.suppress(OSError):
|
||||
AUTH_PATH.unlink()
|
||||
|
||||
|
||||
def read_record() -> dict[str, Any] | None:
|
||||
record = _read_store().get(PROVIDER)
|
||||
record = read_provider_record(PROVIDER)
|
||||
if not isinstance(record, dict) or record.get("type") != "oauth":
|
||||
return None
|
||||
if not (record.get("access") and record.get("refresh") and record.get("account_id")):
|
||||
@@ -86,21 +110,11 @@ def is_authenticated() -> bool:
|
||||
|
||||
|
||||
def save_record(record: dict[str, Any]) -> None:
|
||||
data = _read_store()
|
||||
data[PROVIDER] = record
|
||||
_write_store(data)
|
||||
save_provider_record(PROVIDER, record)
|
||||
|
||||
|
||||
def logout() -> None:
|
||||
data = _read_store()
|
||||
if PROVIDER not in data:
|
||||
return
|
||||
del data[PROVIDER]
|
||||
if data:
|
||||
_write_store(data)
|
||||
return
|
||||
with contextlib.suppress(OSError):
|
||||
AUTH_PATH.unlink()
|
||||
remove_provider_record(PROVIDER)
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
|
||||
+41
-9
@@ -20,6 +20,7 @@ from agents.model_settings import ModelSettings
|
||||
from agents.models.fake_id import FAKE_RESPONSES_ID
|
||||
from agents.models.interface import Model
|
||||
from agents.models.multi_provider import MultiProvider
|
||||
from agents.models.openai_chatcompletions import OpenAIChatCompletionsModel
|
||||
from agents.models.openai_responses import OpenAIResponsesModel
|
||||
from agents.retry import (
|
||||
ModelRetryBackoffSettings,
|
||||
@@ -36,7 +37,7 @@ from openai.types.responses import (
|
||||
from openai.types.responses.response_usage import ResponseUsage
|
||||
from openai.types.shared import Reasoning
|
||||
|
||||
from strix.config import codex
|
||||
from strix.config import codex, opencode
|
||||
from strix.config.loader import load_settings
|
||||
from strix.config.tool_call_ids import TurnCallIdRewriter, dedupe_input
|
||||
from strix.config.tool_call_limits import TurnToolCallLimiter
|
||||
@@ -79,7 +80,12 @@ def _retry_statusless_provider_errors(context: RetryPolicyContext) -> bool:
|
||||
|
||||
|
||||
class _CodexResponsesModel(OpenAIResponsesModel):
|
||||
"""Responses model for the ChatGPT subscription backend (always streamed, stateless)."""
|
||||
"""Responses model for stateless subscription gateways (always streamed).
|
||||
|
||||
Used for the ChatGPT subscription backend and for Responses-served models on
|
||||
the OpenCode gateway: neither stores responses server-side, so reasoning is
|
||||
carried inline via ``reasoning.encrypted_content``.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -471,6 +477,7 @@ class StrixProvider(MultiProvider):
|
||||
def get_model(self, model_name: str | None) -> Model:
|
||||
llm = load_settings().llm
|
||||
slug = codex.subscription_model(model_name)
|
||||
oc = opencode.subscription_model(model_name)
|
||||
idle_timeout = float(llm.stream_idle_timeout)
|
||||
if slug:
|
||||
# The ChatGPT subscription backend is always streamed; it has no
|
||||
@@ -481,6 +488,19 @@ class StrixProvider(MultiProvider):
|
||||
codex.get_subscription_client(),
|
||||
reasoning_effort=llm.reasoning_effort,
|
||||
)
|
||||
elif oc and oc.uses_responses:
|
||||
model = _CodexResponsesModel(
|
||||
oc.slug,
|
||||
opencode.get_subscription_client(oc.base_url),
|
||||
reasoning_effort=llm.reasoning_effort,
|
||||
)
|
||||
elif oc:
|
||||
model = OpenAIChatCompletionsModel(
|
||||
oc.slug, opencode.get_subscription_client(oc.base_url)
|
||||
)
|
||||
if llm.disable_streaming:
|
||||
model = _NonStreamingModel(model)
|
||||
idle_timeout = 0.0
|
||||
else:
|
||||
model = super().get_model(model_name)
|
||||
if llm.disable_streaming:
|
||||
@@ -540,15 +560,24 @@ RECOMMENDED_MODEL_NAMES = (
|
||||
_RECOMMENDED_MODEL_NAME_SET = frozenset(name.lower() for name in RECOMMENDED_MODEL_NAMES)
|
||||
|
||||
FRONTIER_MODEL_FAMILIES = (
|
||||
(("azure", "azure_ai", "bedrock_mantle", "chatgpt", "openai"), ("gpt-5",)),
|
||||
(("azure", "azure_ai", "bedrock_mantle", "chatgpt", "openai", "opencode"), ("gpt-5",)),
|
||||
(
|
||||
("anthropic", "azure_ai", "bedrock", "claude", "databricks", "snowflake", "vertex_ai"),
|
||||
(
|
||||
"anthropic",
|
||||
"azure_ai",
|
||||
"bedrock",
|
||||
"claude",
|
||||
"databricks",
|
||||
"opencode",
|
||||
"snowflake",
|
||||
"vertex_ai",
|
||||
),
|
||||
("claude-fable-5", "claude-opus-5", "claude-opus-4", "claude-sonnet-5", "claude-sonnet-4"),
|
||||
),
|
||||
(("google", "gemini", "vertex_ai"), ("gemini-3",)),
|
||||
(("deepseek",), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")),
|
||||
(("alibaba", "dashscope", "qwen"), ("qwen3.8", "qwen3.7", "qwen3-max")),
|
||||
(("moonshot", "moonshotai", "kimi"), ("kimi-k3", "kimi-k2.7", "kimi-k2.6")),
|
||||
(("google", "gemini", "opencode", "vertex_ai"), ("gemini-3",)),
|
||||
(("deepseek", "opencode"), ("deepseek-v4", "deepseek-r1", "deepseek-reasoner")),
|
||||
(("alibaba", "dashscope", "opencode", "qwen"), ("qwen3.8", "qwen3.7", "qwen3-max")),
|
||||
(("kimi", "moonshot", "moonshotai", "opencode"), ("kimi-k3", "kimi-k2.7", "kimi-k2.6")),
|
||||
)
|
||||
|
||||
|
||||
@@ -556,7 +585,7 @@ def configure_sdk_model_defaults(settings: Settings) -> None:
|
||||
"""Apply Strix config to SDK-native defaults."""
|
||||
llm = settings.llm
|
||||
set_tracing_disabled(True)
|
||||
if codex.subscription_model(llm.model):
|
||||
if codex.subscription_model(llm.model) or opencode.subscription_model(llm.model):
|
||||
return
|
||||
_configure_litellm_compatibility()
|
||||
_configure_openrouter_attribution(llm.model)
|
||||
@@ -741,6 +770,9 @@ def uses_chat_completions_tool_schema(model_name: str, settings: Settings) -> bo
|
||||
"""Return whether the resolved SDK route can only receive JSON function tools."""
|
||||
if codex.subscription_model(model_name):
|
||||
return False
|
||||
oc = opencode.subscription_model(model_name)
|
||||
if oc:
|
||||
return not oc.uses_responses
|
||||
model = model_name.strip().lower()
|
||||
if "/" in model and not model.startswith("openai/"):
|
||||
return True
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
"""OpenCode subscription auth: API-key sign-in and the OpenAI clients that
|
||||
route inference through the OpenCode gateway.
|
||||
|
||||
Covers both OpenCode offerings — Zen (pay-as-you-go credits) and Go (the
|
||||
monthly subscription) — which share one account and API key but live behind
|
||||
different gateway base URLs. Unlike the ChatGPT subscription there is no
|
||||
OAuth: the user copies a plain API key from https://opencode.ai/auth, and
|
||||
using the gateway from other agents is officially supported.
|
||||
|
||||
Model routing follows the endpoint each model is served on (see
|
||||
https://opencode.ai/docs/zen/): GPT models use the Responses API, everything
|
||||
else the OpenAI-compatible Chat Completions API.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
import requests
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
from strix.config import codex
|
||||
|
||||
|
||||
PROVIDER = "opencode"
|
||||
|
||||
ZEN_BASE_URL = "https://opencode.ai/zen/v1"
|
||||
GO_BASE_URL = "https://opencode.ai/zen/go/v1"
|
||||
|
||||
# ``opencode/<model>`` runs on Zen credits; ``opencode-go/<model>`` on the Go
|
||||
# subscription (matching OpenCode's own ``opencode-go/`` model ids).
|
||||
ZEN_PREFIX = "opencode/"
|
||||
GO_PREFIX = "opencode-go/"
|
||||
|
||||
AUTH_CONSOLE_URL = "https://opencode.ai/auth"
|
||||
|
||||
_KEY_CHECK_TIMEOUT = 30
|
||||
|
||||
|
||||
class OpencodeAuthError(Exception):
|
||||
def __init__(self, code: str, message: str | None = None) -> None:
|
||||
self.code = code
|
||||
super().__init__(message or code)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SubscriptionModel:
|
||||
slug: str
|
||||
base_url: str
|
||||
uses_responses: bool
|
||||
|
||||
|
||||
def _uses_responses(slug: str, base_url: str) -> bool:
|
||||
lowered = slug.lower()
|
||||
if lowered.startswith("gpt-"):
|
||||
return True
|
||||
# Grok is served via Responses on Zen but Chat Completions on Go.
|
||||
return lowered.startswith("grok") and base_url == ZEN_BASE_URL
|
||||
|
||||
|
||||
def subscription_model(model_name: str | None) -> SubscriptionModel | None:
|
||||
"""The gateway model behind an ``opencode/`` or ``opencode-go/`` STRIX_LLM."""
|
||||
name = (model_name or "").strip()
|
||||
lowered = name.lower()
|
||||
for prefix, base_url in ((GO_PREFIX, GO_BASE_URL), (ZEN_PREFIX, ZEN_BASE_URL)):
|
||||
if lowered.startswith(prefix):
|
||||
slug = name[len(prefix) :]
|
||||
if not slug:
|
||||
return None
|
||||
return SubscriptionModel(slug, base_url, _uses_responses(slug, base_url))
|
||||
return None
|
||||
|
||||
|
||||
def read_record() -> dict[str, Any] | None:
|
||||
record = codex.read_provider_record(PROVIDER)
|
||||
if not isinstance(record, dict) or record.get("type") != "api_key":
|
||||
return None
|
||||
key = record.get("key")
|
||||
if not isinstance(key, str) or not key:
|
||||
return None
|
||||
return record
|
||||
|
||||
|
||||
def is_authenticated() -> bool:
|
||||
return read_record() is not None
|
||||
|
||||
|
||||
def save_api_key(key: str) -> None:
|
||||
codex.save_provider_record(PROVIDER, {"type": "api_key", "provider": PROVIDER, "key": key})
|
||||
|
||||
|
||||
def logout() -> None:
|
||||
codex.remove_provider_record(PROVIDER)
|
||||
|
||||
|
||||
def get_api_key() -> str:
|
||||
record = read_record()
|
||||
if record is None:
|
||||
raise OpencodeAuthError(
|
||||
"not_authenticated", "not signed in; run: strix auth login opencode"
|
||||
)
|
||||
return str(record["key"])
|
||||
|
||||
|
||||
def validate_api_key(key: str) -> None:
|
||||
"""Check the key against the gateway's models endpoint; raise if rejected."""
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{ZEN_BASE_URL}/models",
|
||||
headers={"Authorization": f"Bearer {key}"},
|
||||
timeout=_KEY_CHECK_TIMEOUT,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
raise OpencodeAuthError("unavailable", str(exc)) from exc
|
||||
if response.status_code in (401, 403):
|
||||
raise OpencodeAuthError(
|
||||
"invalid_key", f"OpenCode rejected the API key (HTTP {response.status_code})"
|
||||
)
|
||||
if response.status_code >= 400:
|
||||
raise OpencodeAuthError("http_error", f"HTTP {response.status_code}: {response.text[:300]}")
|
||||
|
||||
|
||||
def build_openai_client(base_url: str) -> AsyncOpenAI:
|
||||
return AsyncOpenAI(
|
||||
api_key=get_api_key(),
|
||||
base_url=base_url,
|
||||
http_client=httpx.AsyncClient(timeout=httpx.Timeout(600.0, connect=30.0)),
|
||||
)
|
||||
|
||||
|
||||
_subscription_clients: dict[str, AsyncOpenAI] = {}
|
||||
|
||||
|
||||
def get_subscription_client(base_url: str) -> AsyncOpenAI:
|
||||
client = _subscription_clients.get(base_url)
|
||||
if client is None:
|
||||
client = build_openai_client(base_url)
|
||||
_subscription_clients[base_url] = client
|
||||
return client
|
||||
|
||||
|
||||
def auth_mode(model_name: str | None) -> str:
|
||||
"""Return "subscription" when STRIX_LLM runs on any subscription
|
||||
(OpenCode or ChatGPT), else "api_key"."""
|
||||
if subscription_model(model_name) or codex.subscription_model(model_name):
|
||||
return "subscription"
|
||||
return "api_key"
|
||||
|
||||
|
||||
def subscription_provider(model_name: str | None) -> str | None:
|
||||
"""The subscription behind STRIX_LLM: "opencode", "chatgpt", or None."""
|
||||
if subscription_model(model_name):
|
||||
return PROVIDER
|
||||
if codex.subscription_model(model_name):
|
||||
return "chatgpt"
|
||||
return None
|
||||
@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, Any
|
||||
from agents.model_settings import ModelSettings
|
||||
from openai.types.shared import Reasoning
|
||||
|
||||
from strix.config import opencode
|
||||
from strix.config.models import (
|
||||
DEFAULT_MODEL_RETRY,
|
||||
OPENROUTER_ATTRIBUTION_HEADERS,
|
||||
@@ -272,6 +273,10 @@ def _prompt_cache_extra_args(model_name: str) -> dict[str, Any] | None:
|
||||
"""
|
||||
if not is_claude_model(model_name):
|
||||
return None
|
||||
# OpenCode routes use the raw OpenAI SDK, which rejects this LiteLLM-only
|
||||
# argument; the gateway applies Anthropic prompt caching itself.
|
||||
if opencode.subscription_model(model_name):
|
||||
return None
|
||||
if is_bedrock_route(model_name) and not bedrock_route_supports_prompt_caching(model_name):
|
||||
return None
|
||||
|
||||
|
||||
+105
-19
@@ -1,8 +1,9 @@
|
||||
"""`strix auth` — ChatGPT subscription sign-in (login / status / logout).
|
||||
"""`strix auth` — subscription sign-in (login / status / logout).
|
||||
|
||||
Signing in only stores OAuth tokens (``~/.strix/subscription-auth.json``); model
|
||||
Signing in only stores credentials (``~/.strix/subscription-auth.json``); model
|
||||
selection stays with ``STRIX_LLM``. A ``chatgpt/<model>`` STRIX_LLM runs on the
|
||||
subscription.
|
||||
ChatGPT subscription; ``opencode/<model>`` (Zen credits) or
|
||||
``opencode-go/<model>`` (Go subscription) run on OpenCode.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -21,7 +22,7 @@ from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import codex, load_settings
|
||||
from strix.config import codex, load_settings, opencode
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -32,13 +33,20 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
_CALLBACK_TIMEOUT_S = 300
|
||||
|
||||
# CLI-facing name for the login provider. Internally this is the Codex OAuth
|
||||
# flow (``codex.PROVIDER``), but users know it as ChatGPT, so that's what the
|
||||
# command and messaging say. ``codex`` is accepted as an alias.
|
||||
# CLI-facing name for the default login provider. Internally this is the Codex
|
||||
# OAuth flow (``codex.PROVIDER``), but users know it as ChatGPT, so that's what
|
||||
# the command and messaging say. ``codex`` is accepted as an alias.
|
||||
LOGIN_PROVIDER = "chatgpt"
|
||||
_ACCEPTED_PROVIDERS = frozenset({LOGIN_PROVIDER, codex.PROVIDER})
|
||||
_OPENCODE_PROVIDERS = frozenset({opencode.PROVIDER, "opencode-go", "zen"})
|
||||
|
||||
_USAGE = "Usage:\n strix auth login chatgpt [--manual]\n strix auth status\n strix auth logout"
|
||||
_USAGE = (
|
||||
"Usage:\n"
|
||||
" strix auth login chatgpt [--manual]\n"
|
||||
" strix auth login opencode\n"
|
||||
" strix auth status\n"
|
||||
" strix auth logout [chatgpt|opencode]"
|
||||
)
|
||||
|
||||
|
||||
def run_auth(argv: list[str]) -> int:
|
||||
@@ -55,7 +63,7 @@ def run_auth(argv: list[str]) -> int:
|
||||
handlers: dict[str, Callable[[], int]] = {
|
||||
"login": lambda: _login(console, rest),
|
||||
"status": lambda: _status(console),
|
||||
"logout": lambda: _logout(console),
|
||||
"logout": lambda: _logout(console, rest),
|
||||
}
|
||||
handler = handlers.get(subcommand)
|
||||
if handler is not None:
|
||||
@@ -84,10 +92,14 @@ def _login(console: Console, argv: list[str]) -> int:
|
||||
except SystemExit as exc: # argparse already printed the message
|
||||
return int(exc.code or 2)
|
||||
|
||||
if args.provider.lower() in _OPENCODE_PROVIDERS:
|
||||
return _login_opencode(console)
|
||||
|
||||
if args.provider.lower() not in _ACCEPTED_PROVIDERS:
|
||||
console.print(
|
||||
f"[red]Unsupported provider:[/] {args.provider}. "
|
||||
f"Only '{LOGIN_PROVIDER}' (ChatGPT subscription) is supported."
|
||||
f"Supported: '{LOGIN_PROVIDER}' (ChatGPT subscription) and "
|
||||
f"'{opencode.PROVIDER}' (OpenCode Zen/Go)."
|
||||
)
|
||||
return 2
|
||||
|
||||
@@ -115,6 +127,63 @@ def _login(console: Console, argv: list[str]) -> int:
|
||||
return 0
|
||||
|
||||
|
||||
def _login_opencode(console: Console) -> int:
|
||||
console.print()
|
||||
console.print("[bold]Signing in with OpenCode[/] [dim](provider: opencode)[/]")
|
||||
console.print(
|
||||
"[dim]This uses your OpenCode Zen credits or Go subscription for inference.\n"
|
||||
f"Get your API key at {opencode.AUTH_CONSOLE_URL}[/]"
|
||||
)
|
||||
console.print()
|
||||
try:
|
||||
key = console.input("Paste your OpenCode API key: ", password=True).strip()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
console.print("\n[yellow]Sign-in cancelled.[/]")
|
||||
return 130
|
||||
if not key:
|
||||
console.print("[red]No API key provided.[/]")
|
||||
return 2
|
||||
try:
|
||||
opencode.validate_api_key(key)
|
||||
except opencode.OpencodeAuthError as exc:
|
||||
console.print(f"[red]SIGN-IN FAILED:[/] {exc}")
|
||||
return 1
|
||||
opencode.save_api_key(key)
|
||||
_print_opencode_success(console)
|
||||
return 0
|
||||
|
||||
|
||||
def _print_opencode_success(console: Console) -> None:
|
||||
text = Text()
|
||||
text.append("Signed in with your OpenCode account", style="bold #22c55e")
|
||||
text.append("\n\n", style="white")
|
||||
text.append("Set ", style="white")
|
||||
text.append("STRIX_LLM", style="bold white")
|
||||
text.append(" to an ", style="white")
|
||||
text.append("opencode/", style="bold cyan")
|
||||
text.append(" model (e.g. ", style="white")
|
||||
text.append("opencode/claude-sonnet-5", style="bold cyan")
|
||||
text.append(") to run on Zen credits, or ", style="white")
|
||||
text.append("opencode-go/", style="bold cyan")
|
||||
text.append(" (e.g. ", style="white")
|
||||
text.append("opencode-go/kimi-k3", style="bold cyan")
|
||||
text.append(") to run on the Go subscription.", style="white")
|
||||
text.append("\n\n", style="white")
|
||||
text.append("Run a scan as usual, e.g. ", style="white")
|
||||
text.append("strix --target https://example.com", style="bold cyan")
|
||||
console.print()
|
||||
console.print(
|
||||
Panel(
|
||||
text,
|
||||
title="[bold white]STRIX",
|
||||
title_align="left",
|
||||
border_style="#22c55e",
|
||||
padding=(1, 2),
|
||||
)
|
||||
)
|
||||
console.print()
|
||||
|
||||
|
||||
def _run_oauth_flow(
|
||||
console: Console,
|
||||
authorize_url: str,
|
||||
@@ -244,24 +313,41 @@ def _first(query: dict[str, list[str]], key: str) -> str | None:
|
||||
|
||||
def _status(console: Console) -> int:
|
||||
record = codex.read_record()
|
||||
if record is None:
|
||||
console.print("[yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] to sign in.")
|
||||
opencode_signed_in = opencode.is_authenticated()
|
||||
if record is None and not opencode_signed_in:
|
||||
console.print(
|
||||
"[yellow]Not signed in.[/] Run [cyan]strix auth login chatgpt[/] or "
|
||||
"[cyan]strix auth login opencode[/] to sign in."
|
||||
)
|
||||
return 1
|
||||
settings = load_settings()
|
||||
console.print("[green]Signed in[/] with a ChatGPT subscription.")
|
||||
console.print(f" Account: [bold]{record.get('account_id')}[/]")
|
||||
if codex.subscription_model(settings.llm.model):
|
||||
if record is not None:
|
||||
console.print("[green]Signed in[/] with a ChatGPT subscription.")
|
||||
console.print(f" Account: [bold]{record.get('account_id')}[/]")
|
||||
if opencode_signed_in:
|
||||
console.print("[green]Signed in[/] with an OpenCode account.")
|
||||
if codex.subscription_model(settings.llm.model) or opencode.subscription_model(
|
||||
settings.llm.model
|
||||
):
|
||||
console.print(f" Runs use the subscription (STRIX_LLM=[bold]{settings.llm.model}[/]).")
|
||||
else:
|
||||
console.print(
|
||||
" [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. [cyan]chatgpt/gpt-5.4[/] "
|
||||
"to run on the subscription."
|
||||
" [yellow]Note:[/] set [cyan]STRIX_LLM[/] to e.g. [cyan]chatgpt/gpt-5.4[/] or "
|
||||
"[cyan]opencode/claude-sonnet-5[/] to run on a subscription."
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _logout(console: Console) -> int:
|
||||
codex.logout()
|
||||
def _logout(console: Console, argv: list[str] | None = None) -> int:
|
||||
target = (argv[0].lower() if argv else "") or "all"
|
||||
if target in _ACCEPTED_PROVIDERS or target == "all":
|
||||
codex.logout()
|
||||
if target in _OPENCODE_PROVIDERS or target == "all":
|
||||
opencode.logout()
|
||||
if target != "all" and target not in _ACCEPTED_PROVIDERS | _OPENCODE_PROVIDERS:
|
||||
console.print(f"[red]Unknown provider:[/] {target}\n")
|
||||
console.print(_USAGE)
|
||||
return 2
|
||||
console.print("[green]Signed out.[/] Stored subscription credentials removed.")
|
||||
return 0
|
||||
|
||||
|
||||
@@ -8,7 +8,7 @@ from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import codex, load_settings
|
||||
from strix.config import codex, load_settings, opencode
|
||||
from strix.interface.utils import (
|
||||
check_docker_connection,
|
||||
image_exists,
|
||||
@@ -37,6 +37,16 @@ def validate_environment() -> None:
|
||||
logger.info("Environment OK (ChatGPT subscription)")
|
||||
return
|
||||
|
||||
if opencode.subscription_model(settings.llm.model):
|
||||
if not opencode.is_authenticated():
|
||||
console.print(
|
||||
f"[red]STRIX_LLM={settings.llm.model} uses your OpenCode subscription, "
|
||||
"but you're not signed in.[/] Run [cyan]strix auth login opencode[/] first."
|
||||
)
|
||||
sys.exit(1)
|
||||
logger.info("Environment OK (OpenCode subscription)")
|
||||
return
|
||||
|
||||
if not settings.llm.model:
|
||||
missing_required_vars.append("STRIX_LLM")
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import codex, load_settings, persist_current
|
||||
from strix.config import codex, load_settings, opencode, persist_current
|
||||
from strix.core.paths import run_dir_for
|
||||
from strix.interface.cli_args import parse_arguments
|
||||
from strix.interface.environment import (
|
||||
@@ -104,8 +104,14 @@ def _provider_import_hint(exc: BaseException, model: str) -> str | None:
|
||||
|
||||
|
||||
def _subscription_error_hint(exc: BaseException) -> str | None:
|
||||
"""Return an actionable hint for a known ChatGPT-subscription error, or None."""
|
||||
if not codex.subscription_model(load_settings().llm.model):
|
||||
"""Return an actionable hint for a known subscription error, or None."""
|
||||
model = load_settings().llm.model
|
||||
if opencode.subscription_model(model):
|
||||
joined = " ".join(_exception_messages(exc)).lower()
|
||||
if "error code: 401" in joined or "http 401" in joined or "unauthorized" in joined:
|
||||
return "Your OpenCode API key was rejected. Sign in again:\n strix auth login opencode"
|
||||
return None
|
||||
if not codex.subscription_model(model):
|
||||
return None
|
||||
joined = " ".join(_exception_messages(exc)).lower()
|
||||
if "not supported when using codex with a chatgpt account" in joined:
|
||||
|
||||
@@ -14,7 +14,7 @@ import logging
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from strix.config import Settings, codex, load_settings
|
||||
from strix.config import Settings, load_settings, opencode
|
||||
from strix.core.paths import run_dir_for
|
||||
from strix.interface.utils import (
|
||||
assign_workspace_subdirs,
|
||||
@@ -226,7 +226,7 @@ def telemetry_start(args: argparse.Namespace) -> None:
|
||||
model = load_settings().llm.model
|
||||
kwargs = {
|
||||
"model": model,
|
||||
"auth_mode": codex.auth_mode(model),
|
||||
"auth_mode": opencode.auth_mode(model),
|
||||
"scan_mode": args.scan_mode,
|
||||
"is_whitebox": is_whitebox_scan(args.targets_info),
|
||||
"interactive": not args.non_interactive,
|
||||
@@ -247,7 +247,8 @@ def _persist_run_record(args: argparse.Namespace) -> None:
|
||||
"status": "running",
|
||||
"start_time": datetime.now(UTC).isoformat(),
|
||||
"end_time": None,
|
||||
"auth_mode": codex.auth_mode(load_settings().llm.model),
|
||||
"auth_mode": opencode.auth_mode(load_settings().llm.model),
|
||||
"subscription_provider": opencode.subscription_provider(load_settings().llm.model),
|
||||
"targets_info": args.targets_info,
|
||||
"scan_mode": args.scan_mode,
|
||||
"instruction": args.instruction,
|
||||
|
||||
@@ -24,7 +24,7 @@ from strix.interface.tui.backend.projection import (
|
||||
sanitize_terminal_text,
|
||||
terminal_projection,
|
||||
)
|
||||
from strix.interface.utils import is_subscription_run
|
||||
from strix.interface.utils import is_subscription_run, subscription_label
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -164,6 +164,10 @@ class TuiController:
|
||||
subscription = False
|
||||
with contextlib.suppress(Exception):
|
||||
subscription = is_subscription_run(self.report_state)
|
||||
label = ""
|
||||
if subscription:
|
||||
with contextlib.suppress(Exception):
|
||||
label = subscription_label()
|
||||
model_warning = ""
|
||||
if model and not is_recommended_or_frontier_model(model):
|
||||
model_warning = (
|
||||
@@ -200,6 +204,7 @@ class TuiController:
|
||||
],
|
||||
"usage": terminal_projection(usage, max_string=256, max_items=20),
|
||||
"subscription": subscription,
|
||||
"subscription_label": label,
|
||||
"viewer_status": self.viewer_status,
|
||||
"viewer_url": terminal_projection(self.viewer_url, max_string=1024),
|
||||
"error": terminal_projection(self.error, max_string=2 * 1024),
|
||||
|
||||
@@ -596,7 +596,11 @@ func (m Model) statsView() string {
|
||||
if b.Len() > 0 {
|
||||
b.WriteString("\n")
|
||||
}
|
||||
b.WriteString(lipgloss.NewStyle().Foreground(green).Render("ChatGPT subscription"))
|
||||
label := m.snapshot.SubscriptionLabel
|
||||
if label == "" {
|
||||
label = "ChatGPT subscription"
|
||||
}
|
||||
b.WriteString(lipgloss.NewStyle().Foreground(green).Render(label))
|
||||
}
|
||||
total := numberValue(m.snapshot.Usage["total_tokens"])
|
||||
if total > 0 {
|
||||
|
||||
@@ -68,6 +68,7 @@ type Snapshot struct {
|
||||
Vulnerabilities []map[string]any `json:"-"`
|
||||
Usage map[string]any `json:"usage"`
|
||||
Subscription bool `json:"subscription"`
|
||||
SubscriptionLabel string `json:"subscription_label"`
|
||||
ViewerStatus string `json:"viewer_status"`
|
||||
ViewerURL *string `json:"viewer_url"`
|
||||
Error *string `json:"error"`
|
||||
|
||||
@@ -262,9 +262,19 @@ def is_subscription_run(report_state: Any) -> bool:
|
||||
record = getattr(report_state, "run_record", None)
|
||||
if isinstance(record, dict) and record.get("auth_mode"):
|
||||
return record.get("auth_mode") == "subscription"
|
||||
from strix.config import codex
|
||||
from strix.config import opencode
|
||||
|
||||
return codex.auth_mode(load_settings().llm.model) == "subscription"
|
||||
return opencode.auth_mode(load_settings().llm.model) == "subscription"
|
||||
|
||||
|
||||
def subscription_label() -> str:
|
||||
"""Display name of the subscription behind the configured model."""
|
||||
from strix.config import opencode
|
||||
|
||||
model = load_settings().llm.model
|
||||
if opencode.subscription_model(model):
|
||||
return "OpenCode subscription"
|
||||
return "ChatGPT subscription"
|
||||
|
||||
|
||||
def _int_stat(usage: dict[str, Any], key: str) -> int:
|
||||
@@ -368,7 +378,7 @@ def build_live_stats_text(report_state: Any) -> Text:
|
||||
stats_text.append(str(model), style="white")
|
||||
if is_subscription_run(report_state):
|
||||
stats_text.append(" · ", style="dim white")
|
||||
stats_text.append("ChatGPT subscription", style="#22c55e")
|
||||
stats_text.append(subscription_label(), style="#22c55e")
|
||||
stats_text.append("\n")
|
||||
|
||||
vuln_count = len(report_state.vulnerability_reports)
|
||||
@@ -414,7 +424,7 @@ def build_tui_stats_text(report_state: Any) -> Text:
|
||||
subscription = is_subscription_run(report_state)
|
||||
if subscription:
|
||||
stats_text.append("\n")
|
||||
stats_text.append("ChatGPT subscription", style="#22c55e")
|
||||
stats_text.append(subscription_label(), style="#22c55e")
|
||||
|
||||
usage = _llm_usage(report_state)
|
||||
if usage and _int_stat(usage, "total_tokens") > 0:
|
||||
|
||||
@@ -101,6 +101,11 @@ export function RunDetails({
|
||||
const totalTokens = num(usage.total_tokens);
|
||||
const cost = num(usage.cost);
|
||||
const subscription = str(raw.auth_mode) === "subscription";
|
||||
const subscriptionProvider =
|
||||
str(raw.subscription_provider) ??
|
||||
(models.some((m) => m.toLowerCase().startsWith("opencode")) ? "opencode" : "chatgpt");
|
||||
const subscriptionLabel =
|
||||
subscriptionProvider === "opencode" ? "OpenCode subscription" : "ChatGPT subscription";
|
||||
|
||||
const sub = (n: number, word: string) => (
|
||||
<span className="text-[#666]"> ({formatNumber(n)} {word})</span>
|
||||
@@ -180,7 +185,7 @@ export function RunDetails({
|
||||
<Field label="Provider">
|
||||
<span className="inline-flex items-center gap-1.5">
|
||||
<span className="rounded-full border border-[#22c55e]/40 bg-[#22c55e]/10 px-2 py-0.5 text-[11px] text-[#22c55e]">
|
||||
ChatGPT subscription
|
||||
{subscriptionLabel}
|
||||
</span>
|
||||
</span>
|
||||
</Field>
|
||||
|
||||
+24
-24
File diff suppressed because one or more lines are too long
@@ -6,7 +6,7 @@
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
|
||||
<meta name="color-scheme" content="dark" />
|
||||
<title>Strix Results</title>
|
||||
<script type="module" crossorigin src="./assets/index-DBJ-RJqo.js"></script>
|
||||
<script type="module" crossorigin src="./assets/index-1LIW3rcB.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="./assets/index-DKbLYAbP.css">
|
||||
</head>
|
||||
<body>
|
||||
|
||||
@@ -20,6 +20,8 @@ logger = logging.getLogger(__name__)
|
||||
_STRIPPABLE_PREFIXES = (
|
||||
"openai/",
|
||||
"chatgpt/",
|
||||
"opencode-go/",
|
||||
"opencode/",
|
||||
"litellm/",
|
||||
"any-llm/",
|
||||
"ollama/",
|
||||
@@ -48,7 +50,11 @@ def _model_info(model: str) -> dict[str, int]:
|
||||
lookup_key = _lookup_key(model)
|
||||
# Provider-qualified ChatGPT lookups may start a synchronous device-login
|
||||
# poll. LiteLLM keys the metadata by the underlying model slug.
|
||||
candidates = (lookup_key,) if model.startswith("chatgpt/") else (model, lookup_key)
|
||||
candidates = (
|
||||
(lookup_key,)
|
||||
if model.startswith(("chatgpt/", "opencode/", "opencode-go/"))
|
||||
else (model, lookup_key)
|
||||
)
|
||||
for candidate in candidates:
|
||||
info = _safe_get_model_info(candidate)
|
||||
if info is not None:
|
||||
|
||||
@@ -11,7 +11,7 @@ from uuid import uuid4
|
||||
|
||||
from agents.usage import Usage
|
||||
|
||||
from strix.config import codex
|
||||
from strix.config import opencode
|
||||
from strix.config.loader import load_settings
|
||||
from strix.core.paths import run_dir_for
|
||||
from strix.report.pricing import resolve_litellm_model
|
||||
@@ -123,7 +123,7 @@ class ReportState:
|
||||
self.scan_results: dict[str, Any] | None = None
|
||||
self.scan_config: dict[str, Any] | None = None
|
||||
self._llm_usage = LLMUsageLedger()
|
||||
auth_mode = codex.auth_mode(load_settings().llm.model)
|
||||
auth_mode = opencode.auth_mode(load_settings().llm.model)
|
||||
self._llm_usage.zero_cost = auth_mode == "subscription"
|
||||
self.run_record: dict[str, Any] = {
|
||||
"run_id": self.run_id,
|
||||
@@ -132,6 +132,7 @@ class ReportState:
|
||||
"end_time": None,
|
||||
"status": "running",
|
||||
"auth_mode": auth_mode,
|
||||
"subscription_provider": opencode.subscription_provider(load_settings().llm.model),
|
||||
"targets_info": [],
|
||||
"llm_usage": self._build_llm_usage_record(),
|
||||
}
|
||||
|
||||
+59
-1
@@ -2,11 +2,12 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import pytest
|
||||
|
||||
from strix.config import codex
|
||||
from strix.config import codex, opencode
|
||||
from strix.interface import auth_cli
|
||||
|
||||
|
||||
@@ -104,3 +105,60 @@ def test_login_accepts_provider_aliases(provider: str, monkeypatch: pytest.Monke
|
||||
|
||||
assert auth_cli.run_auth(["login", provider]) == 0
|
||||
assert reached["flow"] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("provider", ["opencode", "OpenCode", "opencode-go", "zen"])
|
||||
def test_login_accepts_opencode_aliases(provider: str, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
reached = {"login": False}
|
||||
|
||||
def _fake_login(_console: Any) -> int:
|
||||
reached["login"] = True
|
||||
return 0
|
||||
|
||||
monkeypatch.setattr(auth_cli, "_login_opencode", _fake_login)
|
||||
assert auth_cli.run_auth(["login", provider]) == 0
|
||||
assert reached["login"] is True
|
||||
|
||||
|
||||
def test_login_opencode_validates_and_saves(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
saved: dict[str, str] = {}
|
||||
monkeypatch.setattr("rich.console.Console.input", lambda _self, *_a, **_k: " sk-oc-test ")
|
||||
monkeypatch.setattr(opencode, "validate_api_key", lambda key: saved.setdefault("checked", key))
|
||||
monkeypatch.setattr(opencode, "save_api_key", lambda key: saved.setdefault("key", key))
|
||||
|
||||
assert auth_cli.run_auth(["login", "opencode"]) == 0
|
||||
assert saved == {"checked": "sk-oc-test", "key": "sk-oc-test"}
|
||||
|
||||
|
||||
def test_login_opencode_rejects_bad_key(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr("rich.console.Console.input", lambda _self, *_a, **_k: "bad")
|
||||
|
||||
def _reject(_key: str) -> None:
|
||||
raise opencode.OpencodeAuthError("invalid_key")
|
||||
|
||||
monkeypatch.setattr(opencode, "validate_api_key", _reject)
|
||||
assert auth_cli.run_auth(["login", "opencode"]) == 1
|
||||
assert opencode.is_authenticated() is False
|
||||
|
||||
|
||||
def test_logout_provider_scoped() -> None:
|
||||
codex.save_record(
|
||||
{
|
||||
"type": "oauth",
|
||||
"provider": "codex",
|
||||
"access": "a",
|
||||
"refresh": "r",
|
||||
"account_id": "acct",
|
||||
"expires_at": time.time() + 3600,
|
||||
}
|
||||
)
|
||||
opencode.save_api_key("sk-oc-test")
|
||||
|
||||
assert auth_cli.run_auth(["logout", "opencode"]) == 0
|
||||
assert opencode.is_authenticated() is False
|
||||
assert codex.is_authenticated() is True
|
||||
|
||||
assert auth_cli.run_auth(["logout"]) == 0
|
||||
assert codex.is_authenticated() is False
|
||||
|
||||
assert auth_cli.run_auth(["logout", "bogus"]) == 2
|
||||
|
||||
@@ -111,6 +111,13 @@ def test_make_model_settings_no_prompt_cache_for_non_claude(model_name: str) ->
|
||||
assert make_model_settings(None, model_name=model_name).extra_args is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize("model_name", ["opencode/claude-sonnet-5", "opencode-go/claude-sonnet-5"])
|
||||
def test_no_prompt_cache_for_opencode_claude(model_name: str) -> None:
|
||||
# The OpenCode route uses the raw OpenAI SDK, whose create() rejects the
|
||||
# LiteLLM-only cache_control_injection_points argument.
|
||||
assert _cache_points(model_name) is None
|
||||
|
||||
|
||||
def test_no_prompt_cache_for_unmapped_bedrock_claude_model(monkeypatch: Any) -> None:
|
||||
# A Bedrock Claude model LiteLLM hasn't mapped must run uncached, not crash.
|
||||
unmapped = "bedrock/global.anthropic.claude-brand-new-9"
|
||||
|
||||
@@ -66,6 +66,11 @@ def test_recommended_models_are_matched_case_insensitively() -> None:
|
||||
"moonshot/kimi-k2.6",
|
||||
"kimi-k2.7-code",
|
||||
"moonshot/kimi-k3",
|
||||
"opencode/gpt-5.4",
|
||||
"opencode/claude-sonnet-5",
|
||||
"opencode-go/kimi-k3",
|
||||
"opencode-go/deepseek-v4-flash",
|
||||
"opencode-go/qwen3.8-max",
|
||||
],
|
||||
)
|
||||
def test_frontier_model_families_are_accepted(model_name: str) -> None:
|
||||
|
||||
@@ -0,0 +1,133 @@
|
||||
"""Tests for OpenCode (Zen/Go) subscription auth: prefix parsing and key store."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
from strix.config import codex, opencode
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _tmp_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||||
path = tmp_path / "home" / ".strix" / "subscription-auth.json"
|
||||
monkeypatch.setattr(codex, "AUTH_PATH", path)
|
||||
return path
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("model", "slug", "base_url", "uses_responses"),
|
||||
[
|
||||
("opencode/claude-sonnet-5", "claude-sonnet-5", opencode.ZEN_BASE_URL, False),
|
||||
("opencode/gpt-5.4", "gpt-5.4", opencode.ZEN_BASE_URL, True),
|
||||
("opencode/grok-4.5", "grok-4.5", opencode.ZEN_BASE_URL, True),
|
||||
("OpenCode/Kimi-K3", "Kimi-K3", opencode.ZEN_BASE_URL, False),
|
||||
("opencode-go/kimi-k3", "kimi-k3", opencode.GO_BASE_URL, False),
|
||||
("opencode-go/gpt-5.6-luna", "gpt-5.6-luna", opencode.GO_BASE_URL, True),
|
||||
("opencode-go/grok-4.5", "grok-4.5", opencode.GO_BASE_URL, False),
|
||||
],
|
||||
)
|
||||
def test_subscription_model_parses_prefixes(
|
||||
model: str, slug: str, base_url: str, uses_responses: bool
|
||||
) -> None:
|
||||
parsed = opencode.subscription_model(model)
|
||||
assert parsed is not None
|
||||
assert parsed.slug == slug
|
||||
assert parsed.base_url == base_url
|
||||
assert parsed.uses_responses == uses_responses
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"model",
|
||||
["openai/gpt-5.4", "chatgpt/gpt-5.4", "opencode/", "opencode-go/", "opencode", "", None],
|
||||
)
|
||||
def test_subscription_model_rejects_non_opencode(model: str | None) -> None:
|
||||
assert opencode.subscription_model(model) is None
|
||||
|
||||
|
||||
def test_store_roundtrip_and_logout() -> None:
|
||||
assert opencode.read_record() is None
|
||||
assert opencode.is_authenticated() is False
|
||||
|
||||
opencode.save_api_key("sk-oc-test")
|
||||
record = opencode.read_record()
|
||||
assert record is not None
|
||||
assert record["key"] == "sk-oc-test"
|
||||
assert opencode.is_authenticated() is True
|
||||
assert opencode.get_api_key() == "sk-oc-test"
|
||||
|
||||
opencode.logout()
|
||||
assert opencode.read_record() is None
|
||||
opencode.logout() # no-op when already gone
|
||||
|
||||
|
||||
def test_store_coexists_with_chatgpt_record() -> None:
|
||||
codex.save_record({"type": "oauth", "access": "a", "refresh": "r", "account_id": "acct"})
|
||||
opencode.save_api_key("sk-oc-test")
|
||||
|
||||
assert codex.read_record() is not None
|
||||
assert opencode.get_api_key() == "sk-oc-test"
|
||||
|
||||
opencode.logout()
|
||||
assert codex.read_record() is not None
|
||||
assert opencode.read_record() is None
|
||||
|
||||
|
||||
def test_get_api_key_raises_when_not_signed_in() -> None:
|
||||
with pytest.raises(opencode.OpencodeAuthError) as exc:
|
||||
opencode.get_api_key()
|
||||
assert exc.value.code == "not_authenticated"
|
||||
|
||||
|
||||
def test_auth_mode_covers_both_subscriptions() -> None:
|
||||
assert opencode.auth_mode("opencode/claude-sonnet-5") == "subscription"
|
||||
assert opencode.auth_mode("opencode-go/kimi-k3") == "subscription"
|
||||
assert opencode.auth_mode("chatgpt/gpt-5.4") == "subscription"
|
||||
assert opencode.auth_mode("openai/gpt-5.4") == "api_key"
|
||||
assert opencode.auth_mode(None) == "api_key"
|
||||
|
||||
|
||||
def test_subscription_provider() -> None:
|
||||
assert opencode.subscription_provider("opencode/claude-sonnet-5") == "opencode"
|
||||
assert opencode.subscription_provider("opencode-go/kimi-k3") == "opencode"
|
||||
assert opencode.subscription_provider("chatgpt/gpt-5.4") == "chatgpt"
|
||||
assert opencode.subscription_provider("openai/gpt-5.4") is None
|
||||
assert opencode.subscription_provider(None) is None
|
||||
|
||||
|
||||
def _response(status_code: int, text: str = "") -> mock.MagicMock:
|
||||
response = mock.MagicMock()
|
||||
response.status_code = status_code
|
||||
response.text = text
|
||||
return response
|
||||
|
||||
|
||||
def test_validate_api_key_accepts_ok() -> None:
|
||||
with mock.patch.object(requests, "get", return_value=_response(200)) as get:
|
||||
opencode.validate_api_key("sk-oc-test")
|
||||
assert get.call_args.kwargs["headers"]["Authorization"] == "Bearer sk-oc-test"
|
||||
|
||||
|
||||
def test_validate_api_key_rejects_unauthorized() -> None:
|
||||
with (
|
||||
mock.patch.object(requests, "get", return_value=_response(401)),
|
||||
pytest.raises(opencode.OpencodeAuthError) as exc,
|
||||
):
|
||||
opencode.validate_api_key("bad-key")
|
||||
assert exc.value.code == "invalid_key"
|
||||
|
||||
|
||||
def test_validate_api_key_maps_network_errors() -> None:
|
||||
with (
|
||||
mock.patch.object(requests, "get", side_effect=requests.ConnectionError("boom")),
|
||||
pytest.raises(opencode.OpencodeAuthError) as exc,
|
||||
):
|
||||
opencode.validate_api_key("sk-oc-test")
|
||||
assert exc.value.code == "unavailable"
|
||||
Reference in New Issue
Block a user