diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx index f1542b752..d5678478e 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -1,169 +1,174 @@ ---- -title: "Configuration" -description: "Environment variables for Strix" ---- - -Configure Strix using environment variables or a config file. - -## LLM Configuration - - - Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`). - - - - API key for your LLM provider. Not required for local models or cloud provider auth (Vertex AI, AWS Bedrock). - - - - Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`. - - - - Extra HTTP headers sent on every LLM request, as a JSON object (e.g. - `{"X-Feature-Key":"value","X-Tenant":"acme"}`). Useful for OpenAI-compatible - gateways that require attribution or routing headers in addition to the bearer - token. The bearer token itself still comes from `LLM_API_KEY`. Applies to both - the LiteLLM and native OpenAI routing paths. - - - - Request timeout in seconds for LLM calls. - - - - Maximum number of retries for LLM API calls on transient failures. - - - - Control thinking effort for reasoning models. Valid values: `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max`. Defaults to `medium` for quick scan mode. - - - - Timeout in seconds for memory compression operations (context summarization). - - -### Dedicated deduplication model - -Finding deduplication is a cheap, structured classification task. By default it -runs on the main model, but you can route it to a smaller/cheaper model without -affecting the agents that do the actual testing. - - - Model used to judge whether a candidate finding duplicates an existing report. - Falls back to `STRIX_LLM` when unset. - - - - Optional provider key for the deduplication model. - - - - Optional custom API base URL for the deduplication model. Use when the dedupe - model runs on a different endpoint than the main model. - - - - Optional JSON object of extra HTTP headers sent on every deduplication-model - request, e.g. `{"X-Feature-Key":"value"}`. A dedicated dedupe model never - inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers. - - - - Reasoning effort for the deduplication model. Defaults to the model's own - baseline when unset. - - -## Optional Features - - - API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research. - - - - Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://`), and Postman environments (`postman://?env=`) to resolve collection variables. Not needed when passing a local collection export file. - - - - Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL). - - - - OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only. - - - - API key used for remote trace export. Remote export is enabled only when both `TRACELOOP_BASE_URL` and `TRACELOOP_API_KEY` are set. - - - - Optional custom OTEL headers (JSON object or `key=value,key2=value2`). Useful for Langfuse or custom/self-hosted OTLP gateways. - - -When remote OTEL vars are not set, Strix still writes complete run telemetry locally to: - -```bash -strix_runs//events.jsonl -``` - -When remote vars are set, Strix dual-writes telemetry to both local JSONL and the remote OTEL endpoint. - -## Docker Configuration - - - Docker image to use for the sandbox container. - - - - Docker daemon socket path. Use for remote Docker hosts or custom configurations. - - +--- +title: "Configuration" +description: "Environment variables for Strix" +--- + +Configure Strix using environment variables or a config file. + +## LLM Configuration + + + Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`). + + + + API key for your LLM provider. Not required for local models or cloud provider auth (Vertex AI, AWS Bedrock). + + + + Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`. + + + + Extra HTTP headers sent on every LLM request, as a JSON object (e.g. + `{"X-Feature-Key":"value","X-Tenant":"acme"}`). Useful for OpenAI-compatible + gateways that require attribution or routing headers in addition to the bearer + token. The bearer token itself still comes from `LLM_API_KEY`. Applies to both + the LiteLLM and native OpenAI routing paths. + + + + Request timeout in seconds for LLM calls. + + + + Maximum number of retries for LLM API calls on transient failures. + + + + Control thinking effort for reasoning models. Valid values: `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max`. Defaults to `medium` for quick scan mode. + + + + Timeout in seconds for memory compression operations (context summarization). + + +### Dedicated deduplication model + +Finding deduplication is a cheap, structured classification task. By default it +runs on the main model, but you can route it to a smaller/cheaper model without +affecting the agents that do the actual testing. + + + Model used to judge whether a candidate finding duplicates an existing report. + Falls back to `STRIX_LLM` when unset. + + + + Optional provider key for the deduplication model. + + + + Optional custom API base URL for the deduplication model. Use when the dedupe + model runs on a different endpoint than the main model. + + + + Optional JSON object of extra HTTP headers sent on every deduplication-model + request, e.g. `{"X-Feature-Key":"value"}`. A dedicated dedupe model never + inherits `LLM_EXTRA_HEADERS`; set this when its endpoint needs custom headers. + + + + Reasoning effort for the deduplication model. Defaults to the model's own + baseline when unset. + + +## Optional Features + + + API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research. + + + + Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://`), and Postman environments (`postman://?env=`) to resolve collection variables. Not needed when passing a local collection export file. + + + + Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL). + + + + OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only. + + + + API key used for remote trace export. Remote export is enabled only when both `TRACELOOP_BASE_URL` and `TRACELOOP_API_KEY` are set. + + + + Optional custom OTEL headers (JSON object or `key=value,key2=value2`). Useful for Langfuse or custom/self-hosted OTLP gateways. + + +When remote OTEL vars are not set, Strix still writes complete run telemetry locally to: + +```bash +strix_runs//events.jsonl +``` + +When remote vars are set, Strix dual-writes telemetry to both local JSONL and the remote OTEL endpoint. + +## Docker Configuration + + + Docker image to use for the sandbox container. + + + + Docker daemon socket path. Use for remote Docker hosts or custom configurations. + + Runtime backend for the sandbox environment. + + Maximum time in seconds to wait for the in-sandbox Caido proxy to become ready. + Increase this on slow or heavily loaded container hosts. + + ## Sandbox Configuration - - - Maximum execution time in seconds for sandbox operations. - - - - Timeout in seconds for connecting to the sandbox container. - - -## Config File - -Strix stores configuration in `~/.strix/cli-config.json`. You can also specify a custom config file: - -```bash -strix --target ./app --config /path/to/config.json -``` - -**Config file format:** - -```json -{ - "env": { - "STRIX_LLM": "openai/gpt-5.4", - "LLM_API_KEY": "sk-...", - "STRIX_REASONING_EFFORT": "high" - } -} -``` - -## Example Setup - -```bash -# Required -export STRIX_LLM="openai/gpt-5.4" -export LLM_API_KEY="sk-..." - -# Optional: Enable web search -export PERPLEXITY_API_KEY="pplx-..." - -# Optional: Custom timeouts -export LLM_TIMEOUT="600" -export STRIX_SANDBOX_EXECUTION_TIMEOUT="300" - -``` + + + Maximum execution time in seconds for sandbox operations. + + + + Timeout in seconds for connecting to the sandbox container. + + +## Config File + +Strix stores configuration in `~/.strix/cli-config.json`. You can also specify a custom config file: + +```bash +strix --target ./app --config /path/to/config.json +``` + +**Config file format:** + +```json +{ + "env": { + "STRIX_LLM": "openai/gpt-5.4", + "LLM_API_KEY": "sk-...", + "STRIX_REASONING_EFFORT": "high" + } +} +``` + +## Example Setup + +```bash +# Required +export STRIX_LLM="openai/gpt-5.4" +export LLM_API_KEY="sk-..." + +# Optional: Enable web search +export PERPLEXITY_API_KEY="pplx-..." + +# Optional: Custom timeouts +export LLM_TIMEOUT="600" +export STRIX_SANDBOX_EXECUTION_TIMEOUT="300" + +``` diff --git a/strix/config/settings.py b/strix/config/settings.py index 42a2c97ea..dda0ebf37 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -1,156 +1,157 @@ -"""Strix application settings — pydantic-settings powered.""" - -from __future__ import annotations - -from typing import Literal - -from pydantic import AliasChoices, Field -from pydantic_settings import BaseSettings, SettingsConfigDict - - -ReasoningEffort = Literal["none", "minimal", "low", "medium", "high", "xhigh", "max"] - -DEFAULT_MAX_TURNS = 500 - -_BASE_CONFIG = SettingsConfigDict( - case_sensitive=False, - populate_by_name=True, - extra="ignore", -) - - -class LlmSettings(BaseSettings): - model_config = _BASE_CONFIG - - model: str | None = Field(default=None, alias="STRIX_LLM") - api_key: str | None = Field( - default=None, - validation_alias=AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"), - repr=False, - ) - api_base: str | None = Field( - default=None, - validation_alias=AliasChoices( - "LLM_API_BASE", - "OPENAI_API_BASE", - "OPENAI_BASE_URL", - "LITELLM_BASE_URL", - "OLLAMA_API_BASE", - ), - ) - extra_headers: dict[str, str] | None = Field( - default=None, - alias="LLM_EXTRA_HEADERS", - repr=False, - ) - reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT") - force_required_tool_choice: bool = Field( - default=False, - alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE", - ) - prompt_cache: bool = Field( - default=True, - alias="STRIX_PROMPT_CACHE", - ) - disable_streaming: bool = Field( - default=False, - alias="LLM_DISABLE_STREAMING", - ) - timeout: int = Field(default=300, alias="LLM_TIMEOUT") - stream_idle_timeout: int = Field(default=300, ge=0, alias="LLM_STREAM_IDLE_TIMEOUT") - max_tool_calls_per_turn: int = Field( - default=32, - ge=0, - alias="LLM_MAX_TOOL_CALLS_PER_TURN", - ) - - -class DedupeSettings(BaseSettings): - model_config = _BASE_CONFIG - - model: str | None = Field(default=None, alias="STRIX_DEDUPE_MODEL") - reasoning_effort: ReasoningEffort | None = Field( - default=None, - alias="STRIX_DEDUPE_REASONING_EFFORT", - ) - api_key: str | None = Field(default=None, alias="DEDUPE_LLM_API_KEY", repr=False) - api_base: str | None = Field(default=None, alias="DEDUPE_LLM_API_BASE") - extra_headers: dict[str, str] | None = Field( - default=None, - alias="DEDUPE_LLM_EXTRA_HEADERS", - repr=False, - ) - - -class ContextSettings(BaseSettings): - """Context-window management: per-tool-output caps and history compaction.""" - - model_config = _BASE_CONFIG - - auto_compact: bool = Field(default=True, alias="STRIX_CONTEXT_AUTO_COMPACT") - compact_buffer_tokens: int = Field(default=20_000, gt=0, alias="STRIX_CONTEXT_BUFFER_TOKENS") - keep_tokens: int = Field(default=8_000, gt=0, alias="STRIX_CONTEXT_KEEP_TOKENS") - fallback_context_tokens: int = Field( - default=200_000, gt=0, alias="STRIX_CONTEXT_FALLBACK_TOKENS" - ) - summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS") - tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS") - tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES") - # Floor above the truncation-notice size so a preview always fits. - tool_output_max_bytes: int = Field( - default=50 * 1024, ge=1024, alias="STRIX_TOOL_OUTPUT_MAX_BYTES" - ) - - -class RuntimeSettings(BaseSettings): - model_config = _BASE_CONFIG - - image: str = Field( - default="ghcr.io/usestrix/strix-sandbox:1.3.0", - alias="STRIX_IMAGE", - ) - backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND") - # Max screenshot/image tool outputs kept live per agent context (0 = none). - max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES") - - -class TelemetrySettings(BaseSettings): - model_config = _BASE_CONFIG - - enabled: bool = Field(default=True, alias="STRIX_TELEMETRY") - - -class IntegrationSettings(BaseSettings): - model_config = _BASE_CONFIG - - perplexity_api_key: str | None = Field( - default=None, - alias="PERPLEXITY_API_KEY", - repr=False, - ) - postman_api_key: str | None = Field( - default=None, - alias="POSTMAN_API_KEY", - repr=False, - ) - - -class ViewerSettings(BaseSettings): - model_config = _BASE_CONFIG - - # Base URL of the Strix relay the local viewer proxies to for email - # verification and encrypted report delivery. The browser never talks to - # the relay directly; the local server is the only caller. - app_url: str = Field(default="https://app.strix.ai", alias="STRIX_APP_URL") - - -class Settings(BaseSettings): - model_config = _BASE_CONFIG - - llm: LlmSettings = Field(default_factory=LlmSettings) - dedupe: DedupeSettings = Field(default_factory=DedupeSettings) - runtime: RuntimeSettings = Field(default_factory=RuntimeSettings) - context: ContextSettings = Field(default_factory=ContextSettings) - telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings) - integrations: IntegrationSettings = Field(default_factory=IntegrationSettings) - viewer: ViewerSettings = Field(default_factory=ViewerSettings) +"""Strix application settings — pydantic-settings powered.""" + +from __future__ import annotations + +from typing import Literal + +from pydantic import AliasChoices, Field +from pydantic_settings import BaseSettings, SettingsConfigDict + + +ReasoningEffort = Literal["none", "minimal", "low", "medium", "high", "xhigh", "max"] + +DEFAULT_MAX_TURNS = 500 + +_BASE_CONFIG = SettingsConfigDict( + case_sensitive=False, + populate_by_name=True, + extra="ignore", +) + + +class LlmSettings(BaseSettings): + model_config = _BASE_CONFIG + + model: str | None = Field(default=None, alias="STRIX_LLM") + api_key: str | None = Field( + default=None, + validation_alias=AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"), + repr=False, + ) + api_base: str | None = Field( + default=None, + validation_alias=AliasChoices( + "LLM_API_BASE", + "OPENAI_API_BASE", + "OPENAI_BASE_URL", + "LITELLM_BASE_URL", + "OLLAMA_API_BASE", + ), + ) + extra_headers: dict[str, str] | None = Field( + default=None, + alias="LLM_EXTRA_HEADERS", + repr=False, + ) + reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT") + force_required_tool_choice: bool = Field( + default=False, + alias="STRIX_FORCE_REQUIRED_TOOL_CHOICE", + ) + prompt_cache: bool = Field( + default=True, + alias="STRIX_PROMPT_CACHE", + ) + disable_streaming: bool = Field( + default=False, + alias="LLM_DISABLE_STREAMING", + ) + timeout: int = Field(default=300, alias="LLM_TIMEOUT") + stream_idle_timeout: int = Field(default=300, ge=0, alias="LLM_STREAM_IDLE_TIMEOUT") + max_tool_calls_per_turn: int = Field( + default=32, + ge=0, + alias="LLM_MAX_TOOL_CALLS_PER_TURN", + ) + + +class DedupeSettings(BaseSettings): + model_config = _BASE_CONFIG + + model: str | None = Field(default=None, alias="STRIX_DEDUPE_MODEL") + reasoning_effort: ReasoningEffort | None = Field( + default=None, + alias="STRIX_DEDUPE_REASONING_EFFORT", + ) + api_key: str | None = Field(default=None, alias="DEDUPE_LLM_API_KEY", repr=False) + api_base: str | None = Field(default=None, alias="DEDUPE_LLM_API_BASE") + extra_headers: dict[str, str] | None = Field( + default=None, + alias="DEDUPE_LLM_EXTRA_HEADERS", + repr=False, + ) + + +class ContextSettings(BaseSettings): + """Context-window management: per-tool-output caps and history compaction.""" + + model_config = _BASE_CONFIG + + auto_compact: bool = Field(default=True, alias="STRIX_CONTEXT_AUTO_COMPACT") + compact_buffer_tokens: int = Field(default=20_000, gt=0, alias="STRIX_CONTEXT_BUFFER_TOKENS") + keep_tokens: int = Field(default=8_000, gt=0, alias="STRIX_CONTEXT_KEEP_TOKENS") + fallback_context_tokens: int = Field( + default=200_000, gt=0, alias="STRIX_CONTEXT_FALLBACK_TOKENS" + ) + summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS") + tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS") + tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES") + # Floor above the truncation-notice size so a preview always fits. + tool_output_max_bytes: int = Field( + default=50 * 1024, ge=1024, alias="STRIX_TOOL_OUTPUT_MAX_BYTES" + ) + + +class RuntimeSettings(BaseSettings): + model_config = _BASE_CONFIG + + image: str = Field( + default="ghcr.io/usestrix/strix-sandbox:1.3.0", + alias="STRIX_IMAGE", + ) + backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND") + caido_boot_wait_s: int = Field(default=300, gt=0, alias="STRIX_CAIDO_BOOT_WAIT_S") + # Max screenshot/image tool outputs kept live per agent context (0 = none). + max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES") + + +class TelemetrySettings(BaseSettings): + model_config = _BASE_CONFIG + + enabled: bool = Field(default=True, alias="STRIX_TELEMETRY") + + +class IntegrationSettings(BaseSettings): + model_config = _BASE_CONFIG + + perplexity_api_key: str | None = Field( + default=None, + alias="PERPLEXITY_API_KEY", + repr=False, + ) + postman_api_key: str | None = Field( + default=None, + alias="POSTMAN_API_KEY", + repr=False, + ) + + +class ViewerSettings(BaseSettings): + model_config = _BASE_CONFIG + + # Base URL of the Strix relay the local viewer proxies to for email + # verification and encrypted report delivery. The browser never talks to + # the relay directly; the local server is the only caller. + app_url: str = Field(default="https://app.strix.ai", alias="STRIX_APP_URL") + + +class Settings(BaseSettings): + model_config = _BASE_CONFIG + + llm: LlmSettings = Field(default_factory=LlmSettings) + dedupe: DedupeSettings = Field(default_factory=DedupeSettings) + runtime: RuntimeSettings = Field(default_factory=RuntimeSettings) + context: ContextSettings = Field(default_factory=ContextSettings) + telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings) + integrations: IntegrationSettings = Field(default_factory=IntegrationSettings) + viewer: ViewerSettings = Field(default_factory=ViewerSettings) diff --git a/strix/runtime/caido_bootstrap.py b/strix/runtime/caido_bootstrap.py index 0b9ad5b17..545c29596 100644 --- a/strix/runtime/caido_bootstrap.py +++ b/strix/runtime/caido_bootstrap.py @@ -1,109 +1,137 @@ -"""Caido client bootstrap. - -The Caido CLI runs as an in-container sidecar listening on -``127.0.0.1:48080`` *inside* the sandbox. We grab a guest token by -``session.exec()``-ing curl from inside the container, then construct -a host-side :class:`caido_sdk_client.Client` against the runtime's -exposed-port URL for all subsequent SDK calls. -""" - -from __future__ import annotations - -import asyncio -import contextlib -import json -import logging -from typing import TYPE_CHECKING - -from caido_sdk_client import Client, TokenAuthOptions -from caido_sdk_client.types import CreateProjectOptions - - -if TYPE_CHECKING: - from agents.sandbox.session import BaseSandboxSession - - -logger = logging.getLogger(__name__) - - -_LOGIN_AS_GUEST_BODY = ( - '{"query":"mutation LoginAsGuest { loginAsGuest { token { accessToken } } }"}' -) - - -async def _login_as_guest( - session: BaseSandboxSession, - *, - container_url: str, - attempts: int = 10, -) -> str: - """``session.exec`` curl to fetch a guest token; retry until ready. - - Caido's GraphQL listener may not be up the instant the container - starts. The retry loop also doubles as the Caido readiness probe — - no separate TCP healthcheck needed. - """ - last_err: str | None = None - for i in range(1, attempts + 1): - result = await session.exec( - "curl", - "-fsS", - "-X", - "POST", - "-H", - "Content-Type: application/json", - "-d", - _LOGIN_AS_GUEST_BODY, - f"{container_url}/graphql", - timeout=15, - ) - if result.ok(): - try: - payload = json.loads(result.stdout) - token = ( - payload.get("data", {}) - .get("loginAsGuest", {}) - .get("token", {}) - .get("accessToken") - ) - if token: - return str(token) - last_err = f"loginAsGuest returned no token: {payload}" - except json.JSONDecodeError as exc: - last_err = f"unparseable response: {exc}: {result.stdout!r}" - else: - stderr = result.stderr.decode("utf-8", errors="replace")[:200] - last_err = f"curl exit {result.exit_code}: {stderr}" - logger.debug("loginAsGuest attempt %d/%d failed: %s", i, attempts, last_err) - await asyncio.sleep(min(2.0 * i, 8.0)) - - raise RuntimeError(f"loginAsGuest failed after {attempts} attempts: {last_err}") - - -async def bootstrap_caido( - session: BaseSandboxSession, - *, - host_url: str, - container_url: str, -) -> Client: - """Connect to the in-container Caido sidecar and select a fresh project.""" - logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url) - - access_token = await _login_as_guest(session, container_url=container_url) - - client = Client(host_url, auth=TokenAuthOptions(token=access_token)) - await client.connect() - - try: - project = await client.project.create( - CreateProjectOptions(name="sandbox", temporary=True), - ) - await client.project.select(project.id) - except BaseException: - # The connected client never reaches the session bundle if project - # setup fails, so close it here to avoid leaking the transport. - with contextlib.suppress(Exception): - await client.aclose() - raise - logger.info("Caido project selected: %s", project.id) - return client +"""Caido client bootstrap. + +The Caido CLI runs as an in-container sidecar listening on +``127.0.0.1:48080`` *inside* the sandbox. We grab a guest token by +``session.exec()``-ing curl from inside the container, then construct +a host-side :class:`caido_sdk_client.Client` against the runtime's +exposed-port URL for all subsequent SDK calls. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import json +import logging +import math +import time +from typing import TYPE_CHECKING + +from caido_sdk_client import Client, TokenAuthOptions +from caido_sdk_client.types import CreateProjectOptions + + +if TYPE_CHECKING: + from agents.sandbox.session import BaseSandboxSession + + +logger = logging.getLogger(__name__) + + +_LOGIN_AS_GUEST_BODY = ( + '{"query":"mutation LoginAsGuest { loginAsGuest { token { accessToken } } }"}' +) + + +async def _login_as_guest( + session: BaseSandboxSession, + *, + container_url: str, + wait_timeout_s: float, +) -> str: + """``session.exec`` curl to fetch a guest token until the deadline. + + Caido's GraphQL listener may not be up the instant the container + starts. The retry loop also doubles as the Caido readiness probe — + no separate TCP healthcheck needed. + """ + started_at = time.monotonic() + deadline = started_at + wait_timeout_s + last_err: str | None = None + attempts = 0 + + while attempts == 0 or time.monotonic() < deadline: + attempts += 1 + remaining = deadline - time.monotonic() + exec_timeout = max(1, min(15, math.ceil(remaining))) + result = await session.exec( + "curl", + "-fsS", + "-X", + "POST", + "-H", + "Content-Type: application/json", + "-d", + _LOGIN_AS_GUEST_BODY, + f"{container_url}/graphql", + timeout=exec_timeout, + ) + if result.ok(): + try: + payload = json.loads(result.stdout) + token = ( + payload.get("data", {}) + .get("loginAsGuest", {}) + .get("token", {}) + .get("accessToken") + ) + if token: + return str(token) + last_err = f"loginAsGuest returned no token: {payload}" + except json.JSONDecodeError as exc: + last_err = f"unparseable response: {exc}: {result.stdout!r}" + else: + stderr = result.stderr.decode("utf-8", errors="replace")[:200] + last_err = f"curl exit {result.exit_code}: {stderr}" + remaining = deadline - time.monotonic() + if remaining <= 0: + break + retry_delay = min(2.0 * attempts, 8.0, remaining) + logger.debug( + "loginAsGuest attempt %d failed after %.1fs; retrying in %.1fs (%.1fs remaining): %s", + attempts, + time.monotonic() - started_at, + retry_delay, + remaining, + last_err, + ) + await asyncio.sleep(retry_delay) + + raise RuntimeError( + f"loginAsGuest not ready after {float(wait_timeout_s):g} seconds " + f"({attempts} attempts): {last_err}" + ) + + +async def bootstrap_caido( + session: BaseSandboxSession, + *, + host_url: str, + container_url: str, + wait_timeout_s: float, +) -> Client: + """Connect to the in-container Caido sidecar and select a fresh project.""" + logger.info("Bootstrapping Caido client (host=%s, container=%s)", host_url, container_url) + + access_token = await _login_as_guest( + session, + container_url=container_url, + wait_timeout_s=wait_timeout_s, + ) + + client = Client(host_url, auth=TokenAuthOptions(token=access_token)) + await client.connect() + + try: + project = await client.project.create( + CreateProjectOptions(name="sandbox", temporary=True), + ) + await client.project.select(project.id) + except BaseException: + # The connected client never reaches the session bundle if project + # setup fails, so close it here to avoid leaking the transport. + with contextlib.suppress(Exception): + await client.aclose() + raise + logger.info("Caido project selected: %s", project.id) + return client diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py index 4b61d7351..d3bcd08d7 100644 --- a/strix/runtime/session_manager.py +++ b/strix/runtime/session_manager.py @@ -1,233 +1,235 @@ -"""Per-scan sandbox session lifecycle.""" - -from __future__ import annotations - -import logging -import os -import sys -from pathlib import Path -from typing import TYPE_CHECKING, Any - -from agents.sandbox.entries import BaseEntry, LocalDir -from agents.sandbox.manifest import Environment, Manifest - -from strix.config import load_settings -from strix.runtime.backends import backend_supports_bind_mounts, get_backend -from strix.runtime.caido_bootstrap import bootstrap_caido - - -if TYPE_CHECKING: - from strix.runtime.status import StatusSink - - -logger = logging.getLogger(__name__) - - -# In-container Caido sidecar port (matches the image's caido-cli bind). -_CONTAINER_CAIDO_PORT = 48080 - - -_SESSION_CACHE: dict[str, dict[str, Any]] = {} - -# Manifest root inside the container; entry keys hang off this path. -_WORKSPACE_ROOT = "/workspace" - -_PROTECTED_METADATA_NAMES = (".git", ".agents", ".codex") - - -def _host_identity_env() -> dict[str, str]: - # Read the platform through a local so it is not narrowed to whichever OS is - # type-checking: comparing sys.platform directly makes one of these branches - # statically dead, and which one flips between Linux and macOS. - platform_name: str = sys.platform - if platform_name != "linux": - return {} - # Bind-mount ownership only needs mapping on Linux, where the container uid - # must match the host's. - return {"STRIX_HOST_UID": str(os.getuid()), "STRIX_HOST_GID": str(os.getgid())} - - -def build_bind_mounts(local_sources: list[dict[str, Any]]) -> list[dict[str, Any]]: - bind_mounts: list[dict[str, Any]] = [] - for src in local_sources: - ws_subdir = src.get("workspace_subdir") or "" - host_path = src.get("source_path") or "" - if not ws_subdir or not host_path: - continue - resolved = Path(host_path).expanduser().resolve() - target = f"{_WORKSPACE_ROOT}/{ws_subdir}" - bind_mounts.append({"source": str(resolved), "target": target, "read_only": False}) - if src.get("protect_metadata"): - bind_mounts.extend(_metadata_mounts(resolved, target)) - return bind_mounts - - -def build_manifest_entries(local_sources: list[dict[str, Any]]) -> dict[str | Path, BaseEntry]: - entries: dict[str | Path, BaseEntry] = {} - for src in local_sources: - ws_subdir = src.get("workspace_subdir") or "" - host_path = src.get("source_path") or "" - if not ws_subdir or not host_path: - continue - entries[ws_subdir] = LocalDir(src=Path(host_path).expanduser().resolve()) - return entries - - -def _metadata_mounts(tree: Path, target: str) -> list[dict[str, Any]]: - mounts: list[dict[str, Any]] = [] - for name in _PROTECTED_METADATA_NAMES: - metadata = tree / name - if not metadata.is_dir() and not metadata.is_file(): - continue - if not metadata.resolve().is_relative_to(tree): - continue - mounts.append({"source": str(metadata), "target": f"{target}/{name}", "read_only": True}) - gitdir = _gitdir_from_pointer(metadata) if metadata.is_file() else None - if gitdir is not None and gitdir.exists() and gitdir.is_relative_to(tree): - relative = gitdir.relative_to(tree).as_posix() - mounts.append( - {"source": str(gitdir), "target": f"{target}/{relative}", "read_only": True} - ) - return mounts - - -def _gitdir_from_pointer(git_file: Path) -> Path | None: - try: - content = git_file.read_text(encoding="utf-8", errors="replace") - except OSError: - return None - for line in content.splitlines(): - prefix, _, value = line.partition(":") - if prefix.strip() == "gitdir" and value.strip(): - candidate = Path(value.strip()).expanduser() - if not candidate.is_absolute(): - candidate = git_file.parent / candidate - return candidate.resolve() - return None - - -async def create_or_reuse( - scan_id: str, - *, - image: str, - local_sources: list[dict[str, Any]], - status_sink: StatusSink | None = None, -) -> dict[str, Any]: - """Return the existing session bundle for ``scan_id`` or create a new one. - - Each ``local_sources`` entry exposes its host ``source_path`` at - ``/workspace/`` inside the container. - """ - - def report(phase: str) -> None: - if status_sink is not None: - status_sink(phase) - - cached = _SESSION_CACHE.get(scan_id) - if cached is not None: - logger.info("Reusing existing sandbox session for scan %s", scan_id) - return cached - - backend_name = load_settings().runtime.backend - backend = get_backend(backend_name) - - if backend_supports_bind_mounts(backend_name): - bind_mounts = build_bind_mounts(local_sources) - entries: dict[str | Path, BaseEntry] = {} - else: - bind_mounts = [] - entries = build_manifest_entries(local_sources) - - # Caido runs as an in-container sidecar; HTTP(S) traffic from any - # process started via ``session.exec`` (the SDK's Shell tool, etc.) - # picks up these env vars automatically. ``NO_PROXY`` keeps the - # agent-browser CDP daemon's localhost traffic from looping back - # through Caido. - container_caido_url = f"http://127.0.0.1:{_CONTAINER_CAIDO_PORT}" - manifest = Manifest( - entries=entries, - environment=Environment( - value={ - "PYTHONUNBUFFERED": "1", - "HOST_GATEWAY": "host.docker.internal", - **_host_identity_env(), - "http_proxy": container_caido_url, - "https_proxy": container_caido_url, - "ALL_PROXY": container_caido_url, - "NO_PROXY": "localhost,127.0.0.1", - }, - ), - ) - - logger.info( - "Creating sandbox session for scan %s (backend=%s, image=%s)", - scan_id, - backend_name, - image, - ) - report("Starting sandbox container") - client, session = await backend( - image=image, - manifest=manifest, - exposed_ports=(_CONTAINER_CAIDO_PORT,), - bind_mounts=bind_mounts, - ) - - report("Setting up the proxy") - caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT) - scheme = "https" if caido_endpoint.tls else "http" - host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}" - logger.debug("Caido host endpoint resolved: %s", host_caido_url) - - caido_client = await bootstrap_caido( - session, - host_url=host_caido_url, - container_url=container_caido_url, - ) - - bundle = { - "client": client, - "session": session, - "caido_client": caido_client, - } - _SESSION_CACHE[scan_id] = bundle - logger.info("Sandbox session for scan %s ready and cached", scan_id) - return bundle - - -async def cleanup(scan_id: str) -> None: - """Tear down ``scan_id``'s container and drop its cache entry. - - Best-effort: any error during ``client.delete`` is logged and - swallowed. We never want a cleanup failure to prevent the next - scan from starting; the worst case is a stranded container that - Docker's normal reaping will catch on next ``docker prune``. - """ - bundle = _SESSION_CACHE.pop(scan_id, None) - if bundle is None: - logger.debug("cleanup(%s): no cached session", scan_id) - return - - caido_client = bundle.get("caido_client") - if caido_client is not None: - try: - await caido_client.aclose() - except Exception: # noqa: BLE001 - logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True) - - client = bundle["client"] - try: - await client.delete(bundle["session"]) - logger.info("Cleaned up sandbox session for scan %s", scan_id) - except Exception: - logger.exception( - "cleanup(%s): client.delete raised; container may need manual reaping", - scan_id, - ) - - docker_client = getattr(client, "docker_client", None) - if docker_client is not None: - try: - docker_client.close() - except Exception: # noqa: BLE001 - logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True) +"""Per-scan sandbox session lifecycle.""" + +from __future__ import annotations + +import logging +import os +import sys +from pathlib import Path +from typing import TYPE_CHECKING, Any + +from agents.sandbox.entries import BaseEntry, LocalDir +from agents.sandbox.manifest import Environment, Manifest + +from strix.config import load_settings +from strix.runtime.backends import backend_supports_bind_mounts, get_backend +from strix.runtime.caido_bootstrap import bootstrap_caido + + +if TYPE_CHECKING: + from strix.runtime.status import StatusSink + + +logger = logging.getLogger(__name__) + + +# In-container Caido sidecar port (matches the image's caido-cli bind). +_CONTAINER_CAIDO_PORT = 48080 + + +_SESSION_CACHE: dict[str, dict[str, Any]] = {} + +# Manifest root inside the container; entry keys hang off this path. +_WORKSPACE_ROOT = "/workspace" + +_PROTECTED_METADATA_NAMES = (".git", ".agents", ".codex") + + +def _host_identity_env() -> dict[str, str]: + # Read the platform through a local so it is not narrowed to whichever OS is + # type-checking: comparing sys.platform directly makes one of these branches + # statically dead, and which one flips between Linux and macOS. + platform_name: str = sys.platform + if platform_name != "linux": + return {} + # Bind-mount ownership only needs mapping on Linux, where the container uid + # must match the host's. + return {"STRIX_HOST_UID": str(os.getuid()), "STRIX_HOST_GID": str(os.getgid())} + + +def build_bind_mounts(local_sources: list[dict[str, Any]]) -> list[dict[str, Any]]: + bind_mounts: list[dict[str, Any]] = [] + for src in local_sources: + ws_subdir = src.get("workspace_subdir") or "" + host_path = src.get("source_path") or "" + if not ws_subdir or not host_path: + continue + resolved = Path(host_path).expanduser().resolve() + target = f"{_WORKSPACE_ROOT}/{ws_subdir}" + bind_mounts.append({"source": str(resolved), "target": target, "read_only": False}) + if src.get("protect_metadata"): + bind_mounts.extend(_metadata_mounts(resolved, target)) + return bind_mounts + + +def build_manifest_entries(local_sources: list[dict[str, Any]]) -> dict[str | Path, BaseEntry]: + entries: dict[str | Path, BaseEntry] = {} + for src in local_sources: + ws_subdir = src.get("workspace_subdir") or "" + host_path = src.get("source_path") or "" + if not ws_subdir or not host_path: + continue + entries[ws_subdir] = LocalDir(src=Path(host_path).expanduser().resolve()) + return entries + + +def _metadata_mounts(tree: Path, target: str) -> list[dict[str, Any]]: + mounts: list[dict[str, Any]] = [] + for name in _PROTECTED_METADATA_NAMES: + metadata = tree / name + if not metadata.is_dir() and not metadata.is_file(): + continue + if not metadata.resolve().is_relative_to(tree): + continue + mounts.append({"source": str(metadata), "target": f"{target}/{name}", "read_only": True}) + gitdir = _gitdir_from_pointer(metadata) if metadata.is_file() else None + if gitdir is not None and gitdir.exists() and gitdir.is_relative_to(tree): + relative = gitdir.relative_to(tree).as_posix() + mounts.append( + {"source": str(gitdir), "target": f"{target}/{relative}", "read_only": True} + ) + return mounts + + +def _gitdir_from_pointer(git_file: Path) -> Path | None: + try: + content = git_file.read_text(encoding="utf-8", errors="replace") + except OSError: + return None + for line in content.splitlines(): + prefix, _, value = line.partition(":") + if prefix.strip() == "gitdir" and value.strip(): + candidate = Path(value.strip()).expanduser() + if not candidate.is_absolute(): + candidate = git_file.parent / candidate + return candidate.resolve() + return None + + +async def create_or_reuse( + scan_id: str, + *, + image: str, + local_sources: list[dict[str, Any]], + status_sink: StatusSink | None = None, +) -> dict[str, Any]: + """Return the existing session bundle for ``scan_id`` or create a new one. + + Each ``local_sources`` entry exposes its host ``source_path`` at + ``/workspace/`` inside the container. + """ + + def report(phase: str) -> None: + if status_sink is not None: + status_sink(phase) + + cached = _SESSION_CACHE.get(scan_id) + if cached is not None: + logger.info("Reusing existing sandbox session for scan %s", scan_id) + return cached + + runtime_settings = load_settings().runtime + backend_name = runtime_settings.backend + backend = get_backend(backend_name) + + if backend_supports_bind_mounts(backend_name): + bind_mounts = build_bind_mounts(local_sources) + entries: dict[str | Path, BaseEntry] = {} + else: + bind_mounts = [] + entries = build_manifest_entries(local_sources) + + # Caido runs as an in-container sidecar; HTTP(S) traffic from any + # process started via ``session.exec`` (the SDK's Shell tool, etc.) + # picks up these env vars automatically. ``NO_PROXY`` keeps the + # agent-browser CDP daemon's localhost traffic from looping back + # through Caido. + container_caido_url = f"http://127.0.0.1:{_CONTAINER_CAIDO_PORT}" + manifest = Manifest( + entries=entries, + environment=Environment( + value={ + "PYTHONUNBUFFERED": "1", + "HOST_GATEWAY": "host.docker.internal", + **_host_identity_env(), + "http_proxy": container_caido_url, + "https_proxy": container_caido_url, + "ALL_PROXY": container_caido_url, + "NO_PROXY": "localhost,127.0.0.1", + }, + ), + ) + + logger.info( + "Creating sandbox session for scan %s (backend=%s, image=%s)", + scan_id, + backend_name, + image, + ) + report("Starting sandbox container") + client, session = await backend( + image=image, + manifest=manifest, + exposed_ports=(_CONTAINER_CAIDO_PORT,), + bind_mounts=bind_mounts, + ) + + report("Setting up the proxy") + caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT) + scheme = "https" if caido_endpoint.tls else "http" + host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}" + logger.debug("Caido host endpoint resolved: %s", host_caido_url) + + caido_client = await bootstrap_caido( + session, + host_url=host_caido_url, + container_url=container_caido_url, + wait_timeout_s=runtime_settings.caido_boot_wait_s, + ) + + bundle = { + "client": client, + "session": session, + "caido_client": caido_client, + } + _SESSION_CACHE[scan_id] = bundle + logger.info("Sandbox session for scan %s ready and cached", scan_id) + return bundle + + +async def cleanup(scan_id: str) -> None: + """Tear down ``scan_id``'s container and drop its cache entry. + + Best-effort: any error during ``client.delete`` is logged and + swallowed. We never want a cleanup failure to prevent the next + scan from starting; the worst case is a stranded container that + Docker's normal reaping will catch on next ``docker prune``. + """ + bundle = _SESSION_CACHE.pop(scan_id, None) + if bundle is None: + logger.debug("cleanup(%s): no cached session", scan_id) + return + + caido_client = bundle.get("caido_client") + if caido_client is not None: + try: + await caido_client.aclose() + except Exception: # noqa: BLE001 + logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True) + + client = bundle["client"] + try: + await client.delete(bundle["session"]) + logger.info("Cleaned up sandbox session for scan %s", scan_id) + except Exception: + logger.exception( + "cleanup(%s): client.delete raised; container may need manual reaping", + scan_id, + ) + + docker_client = getattr(client, "docker_client", None) + if docker_client is not None: + try: + docker_client.close() + except Exception: # noqa: BLE001 + logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True) diff --git a/tests/test_caido_bootstrap.py b/tests/test_caido_bootstrap.py new file mode 100644 index 000000000..5e63e18b3 --- /dev/null +++ b/tests/test_caido_bootstrap.py @@ -0,0 +1,104 @@ +"""Tests for the Caido readiness deadline.""" + +# pyright: reportPrivateUsage=false + +from __future__ import annotations + +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, cast + +import pytest + +from strix.runtime import caido_bootstrap + + +if TYPE_CHECKING: + from agents.sandbox.session import BaseSandboxSession + + +@dataclass +class _ExecResult: + success: bool + stdout: str = "" + stderr: bytes = b"connection refused" + exit_code: int = 7 + + def ok(self) -> bool: + return self.success + + +class _Session: + def __init__(self, results: list[_ExecResult]) -> None: + self._results = iter(results) + self.calls: list[tuple[tuple[str, ...], dict[str, Any]]] = [] + + async def exec(self, *args: str, **kwargs: Any) -> _ExecResult: + self.calls.append((args, kwargs)) + return next(self._results) + + +class _Clock: + def __init__(self) -> None: + self.now = 0.0 + self.sleeps: list[float] = [] + + def monotonic(self) -> float: + return self.now + + async def sleep(self, delay: float) -> None: + self.sleeps.append(delay) + self.now += delay + + +def _install_clock(monkeypatch: pytest.MonkeyPatch) -> _Clock: + clock = _Clock() + monkeypatch.setattr("strix.runtime.caido_bootstrap.time.monotonic", clock.monotonic) + monkeypatch.setattr("strix.runtime.caido_bootstrap.asyncio.sleep", clock.sleep) + return clock + + +@pytest.mark.asyncio +async def test_login_as_guest_retries_until_ready( + monkeypatch: pytest.MonkeyPatch, +) -> None: + clock = _install_clock(monkeypatch) + session = _Session( + [ + _ExecResult(success=False), + _ExecResult( + success=True, + stdout='{"data":{"loginAsGuest":{"token":{"accessToken":"token"}}}}', + ), + ] + ) + + token = await caido_bootstrap._login_as_guest( + cast("BaseSandboxSession", session), + container_url="http://127.0.0.1:48080", + wait_timeout_s=10, + ) + + assert token == "token" # noqa: S105 - synthetic token fixture + assert clock.sleeps == [2.0] + assert [call[1]["timeout"] for call in session.calls] == [10, 8] + + +@pytest.mark.asyncio +async def test_login_as_guest_stops_at_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + clock = _install_clock(monkeypatch) + session = _Session([_ExecResult(success=False), _ExecResult(success=False)]) + + with pytest.raises( + RuntimeError, + match=r"loginAsGuest not ready after 5 seconds \(2 attempts\)", + ): + await caido_bootstrap._login_as_guest( + cast("BaseSandboxSession", session), + container_url="http://127.0.0.1:48080", + wait_timeout_s=5, + ) + + assert clock.sleeps == [2.0, 3.0] + assert [call[1]["timeout"] for call in session.calls] == [5, 3] diff --git a/tests/test_config_loader.py b/tests/test_config_loader.py index e83ab1192..53510e9ef 100644 --- a/tests/test_config_loader.py +++ b/tests/test_config_loader.py @@ -1,218 +1,225 @@ -"""Tests for strix.config.loader: JSON overrides, alias resolution, persistence.""" - -from __future__ import annotations - -import json -from typing import TYPE_CHECKING - -import pytest -from pydantic import AliasChoices, Field, ValidationError -from pydantic.fields import FieldInfo - -from strix.config import loader -from strix.config.settings import ContextSettings - - -if TYPE_CHECKING: - from pathlib import Path - - -_LLM_ENV_KEYS = [ - "STRIX_LLM", - "LLM_API_KEY", - "OPENAI_API_KEY", - "LLM_API_BASE", - "OPENAI_API_BASE", - "OPENAI_BASE_URL", - "LITELLM_BASE_URL", - "OLLAMA_API_BASE", - "STRIX_REASONING_EFFORT", - "STRIX_FORCE_REQUIRED_TOOL_CHOICE", - "LLM_TIMEOUT", - "PERPLEXITY_API_KEY", - # RuntimeSettings - "STRIX_IMAGE", - "STRIX_RUNTIME_BACKEND", - # TelemetrySettings - "STRIX_TELEMETRY", -] - - -@pytest.fixture(autouse=True) -def _reset_loader_state(monkeypatch: pytest.MonkeyPatch) -> None: - """Reset module globals and clear known env vars for deterministic runs.""" - for key in _LLM_ENV_KEYS: - monkeypatch.delenv(key, raising=False) - monkeypatch.setattr(loader, "_cached", None) - monkeypatch.setattr(loader, "_override", None) - - -# --------------------------------------------------------------------------- # -# _read_json_overrides -# --------------------------------------------------------------------------- # - - -def test_read_json_overrides_missing_file(tmp_path: Path) -> None: - assert loader._read_json_overrides(tmp_path / "nope.json") == {} - - -def test_read_json_overrides_corrupt_json(tmp_path: Path) -> None: - path = tmp_path / "cli-config.json" - path.write_text("{not valid json", encoding="utf-8") - assert loader._read_json_overrides(path) == {} - - -def test_read_json_overrides_non_dict_env(tmp_path: Path) -> None: - path = tmp_path / "cli-config.json" - path.write_text(json.dumps({"env": ["not", "a", "dict"]}), encoding="utf-8") - assert loader._read_json_overrides(path) == {} - - -def test_read_json_overrides_maps_to_nested_settings(tmp_path: Path) -> None: - path = tmp_path / "cli-config.json" - path.write_text( - json.dumps({"env": {"STRIX_LLM": "my-model", "PERPLEXITY_API_KEY": "pk"}}), - encoding="utf-8", - ) - assert loader._read_json_overrides(path) == { - "llm": {"model": "my-model"}, - "integrations": {"perplexity_api_key": "pk"}, - } - - -def test_read_json_overrides_skips_keys_already_in_environ( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - monkeypatch.setenv("STRIX_LLM", "from-env") - path = tmp_path / "cli-config.json" - path.write_text(json.dumps({"env": {"STRIX_LLM": "from-file"}}), encoding="utf-8") - # env wins -> the JSON value is not surfaced as an init kwarg. - assert loader._read_json_overrides(path) == {} - - -def test_read_json_overrides_env_wins_across_field_aliases( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - # api_key resolves from AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"). The env - # sets one alias while the persisted file holds another. Env must still win, so - # the stale file value must not be surfaced as an init kwarg (which outranks env). - monkeypatch.setenv("OPENAI_API_KEY", "sk-env") - path = tmp_path / "cli-config.json" - path.write_text(json.dumps({"env": {"LLM_API_KEY": "sk-file"}}), encoding="utf-8") - assert loader._read_json_overrides(path) == {} - - -def test_read_json_overrides_env_wins_case_insensitively( - tmp_path: Path, monkeypatch: pytest.MonkeyPatch -) -> None: - # Settings use case_sensitive=False, so a lowercase env var also counts as set. - monkeypatch.setenv("strix_llm", "from-env") - path = tmp_path / "cli-config.json" - path.write_text(json.dumps({"env": {"STRIX_LLM": "from-file"}}), encoding="utf-8") - assert loader._read_json_overrides(path) == {} - - -def test_read_json_overrides_uses_json_when_no_alias_in_environ(tmp_path: Path) -> None: - # No alias of api_key is set in the environment -> the file value is used, even - # when it is stored under a non-first alias. - path = tmp_path / "cli-config.json" - path.write_text(json.dumps({"env": {"OPENAI_API_KEY": "sk-file"}}), encoding="utf-8") - assert loader._read_json_overrides(path) == {"llm": {"api_key": "sk-file"}} - - -def test_tool_output_max_bytes_rejects_sub_notice_values() -> None: - with pytest.raises(ValidationError): - ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=64) - - -def test_tool_output_max_bytes_accepts_floor() -> None: - assert ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=1024).tool_output_max_bytes == 1024 - - -# --------------------------------------------------------------------------- # -# _aliases_for -# --------------------------------------------------------------------------- # - - -def test_aliases_for_simple_alias() -> None: - finfo = FieldInfo(alias="SIMPLE_ALIAS") - assert loader._aliases_for(finfo) == ["SIMPLE_ALIAS"] - - -def test_aliases_for_alias_choices() -> None: - finfo: FieldInfo = Field( # type: ignore[assignment] - default=None, - validation_alias=AliasChoices("FIRST", "SECOND"), - ) - assert loader._aliases_for(finfo) == ["FIRST", "SECOND"] - - -def test_aliases_for_string_validation_alias() -> None: - finfo: FieldInfo = Field(default=None, validation_alias="STR_ALIAS") # type: ignore[assignment] - assert loader._aliases_for(finfo) == ["STR_ALIAS"] - - -def test_aliases_for_no_alias() -> None: - assert loader._aliases_for(FieldInfo()) == [] - - -# --------------------------------------------------------------------------- # -# apply_config_override + load_settings round-trip -# --------------------------------------------------------------------------- # - - -def test_apply_override_and_load_settings_round_trip(tmp_path: Path) -> None: - path = tmp_path / "cli-config.json" - path.write_text( - json.dumps({"env": {"STRIX_LLM": "round-trip-model", "PERPLEXITY_API_KEY": "pk"}}), - encoding="utf-8", - ) - - loader.apply_config_override(path) - settings = loader.load_settings() - - assert settings.llm.model == "round-trip-model" - assert settings.integrations.perplexity_api_key == "pk" - # Second call is memoized -> same object. - assert loader.load_settings() is settings - - -def test_apply_config_override_invalidates_cache(tmp_path: Path) -> None: - first = tmp_path / "first.json" - first.write_text(json.dumps({"env": {"STRIX_LLM": "first-model"}}), encoding="utf-8") - second = tmp_path / "second.json" - second.write_text(json.dumps({"env": {"STRIX_LLM": "second-model"}}), encoding="utf-8") - - loader.apply_config_override(first) - assert loader.load_settings().llm.model == "first-model" - - loader.apply_config_override(second) - assert loader.load_settings().llm.model == "second-model" - - -# --------------------------------------------------------------------------- # -# persist_current -# --------------------------------------------------------------------------- # - - -def test_persist_current_writes_env_block(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("STRIX_LLM", "persisted-model") - target = tmp_path / "sub" / "cli-config.json" - loader.apply_config_override(target) - - loader.persist_current() - - assert target.exists() - assert json.loads(target.read_text(encoding="utf-8")) == { - "env": {"STRIX_LLM": "persisted-model"} - } - - -def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("STRIX_LLM", "persisted-model") - target = tmp_path / "cli-config.json" - loader.apply_config_override(target) - - loader.persist_current() - - assert target.stat().st_mode & 0o777 == 0o600 +"""Tests for strix.config.loader: JSON overrides, alias resolution, persistence.""" + +from __future__ import annotations + +import json +from typing import TYPE_CHECKING + +import pytest +from pydantic import AliasChoices, Field, ValidationError +from pydantic.fields import FieldInfo + +from strix.config import loader +from strix.config.settings import ContextSettings + + +if TYPE_CHECKING: + from pathlib import Path + + +_LLM_ENV_KEYS = [ + "STRIX_LLM", + "LLM_API_KEY", + "OPENAI_API_KEY", + "LLM_API_BASE", + "OPENAI_API_BASE", + "OPENAI_BASE_URL", + "LITELLM_BASE_URL", + "OLLAMA_API_BASE", + "STRIX_REASONING_EFFORT", + "STRIX_FORCE_REQUIRED_TOOL_CHOICE", + "LLM_TIMEOUT", + "PERPLEXITY_API_KEY", + # RuntimeSettings + "STRIX_IMAGE", + "STRIX_RUNTIME_BACKEND", + "STRIX_CAIDO_BOOT_WAIT_S", + # TelemetrySettings + "STRIX_TELEMETRY", +] + + +@pytest.fixture(autouse=True) +def _reset_loader_state(monkeypatch: pytest.MonkeyPatch) -> None: + """Reset module globals and clear known env vars for deterministic runs.""" + for key in _LLM_ENV_KEYS: + monkeypatch.delenv(key, raising=False) + monkeypatch.setattr(loader, "_cached", None) + monkeypatch.setattr(loader, "_override", None) + + +# --------------------------------------------------------------------------- # +# _read_json_overrides +# --------------------------------------------------------------------------- # + + +def test_read_json_overrides_missing_file(tmp_path: Path) -> None: + assert loader._read_json_overrides(tmp_path / "nope.json") == {} + + +def test_read_json_overrides_corrupt_json(tmp_path: Path) -> None: + path = tmp_path / "cli-config.json" + path.write_text("{not valid json", encoding="utf-8") + assert loader._read_json_overrides(path) == {} + + +def test_read_json_overrides_non_dict_env(tmp_path: Path) -> None: + path = tmp_path / "cli-config.json" + path.write_text(json.dumps({"env": ["not", "a", "dict"]}), encoding="utf-8") + assert loader._read_json_overrides(path) == {} + + +def test_read_json_overrides_maps_to_nested_settings(tmp_path: Path) -> None: + path = tmp_path / "cli-config.json" + path.write_text( + json.dumps({"env": {"STRIX_LLM": "my-model", "PERPLEXITY_API_KEY": "pk"}}), + encoding="utf-8", + ) + assert loader._read_json_overrides(path) == { + "llm": {"model": "my-model"}, + "integrations": {"perplexity_api_key": "pk"}, + } + + +def test_read_json_overrides_skips_keys_already_in_environ( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setenv("STRIX_LLM", "from-env") + path = tmp_path / "cli-config.json" + path.write_text(json.dumps({"env": {"STRIX_LLM": "from-file"}}), encoding="utf-8") + # env wins -> the JSON value is not surfaced as an init kwarg. + assert loader._read_json_overrides(path) == {} + + +def test_read_json_overrides_env_wins_across_field_aliases( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # api_key resolves from AliasChoices("LLM_API_KEY", "OPENAI_API_KEY"). The env + # sets one alias while the persisted file holds another. Env must still win, so + # the stale file value must not be surfaced as an init kwarg (which outranks env). + monkeypatch.setenv("OPENAI_API_KEY", "sk-env") + path = tmp_path / "cli-config.json" + path.write_text(json.dumps({"env": {"LLM_API_KEY": "sk-file"}}), encoding="utf-8") + assert loader._read_json_overrides(path) == {} + + +def test_read_json_overrides_env_wins_case_insensitively( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + # Settings use case_sensitive=False, so a lowercase env var also counts as set. + monkeypatch.setenv("strix_llm", "from-env") + path = tmp_path / "cli-config.json" + path.write_text(json.dumps({"env": {"STRIX_LLM": "from-file"}}), encoding="utf-8") + assert loader._read_json_overrides(path) == {} + + +def test_read_json_overrides_uses_json_when_no_alias_in_environ(tmp_path: Path) -> None: + # No alias of api_key is set in the environment -> the file value is used, even + # when it is stored under a non-first alias. + path = tmp_path / "cli-config.json" + path.write_text(json.dumps({"env": {"OPENAI_API_KEY": "sk-file"}}), encoding="utf-8") + assert loader._read_json_overrides(path) == {"llm": {"api_key": "sk-file"}} + + +def test_tool_output_max_bytes_rejects_sub_notice_values() -> None: + with pytest.raises(ValidationError): + ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=64) + + +def test_tool_output_max_bytes_accepts_floor() -> None: + assert ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=1024).tool_output_max_bytes == 1024 + + +def test_runtime_caido_boot_wait_from_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("STRIX_CAIDO_BOOT_WAIT_S", "900") + + assert loader.load_settings().runtime.caido_boot_wait_s == 900 + + +# --------------------------------------------------------------------------- # +# _aliases_for +# --------------------------------------------------------------------------- # + + +def test_aliases_for_simple_alias() -> None: + finfo = FieldInfo(alias="SIMPLE_ALIAS") + assert loader._aliases_for(finfo) == ["SIMPLE_ALIAS"] + + +def test_aliases_for_alias_choices() -> None: + finfo: FieldInfo = Field( # type: ignore[assignment] + default=None, + validation_alias=AliasChoices("FIRST", "SECOND"), + ) + assert loader._aliases_for(finfo) == ["FIRST", "SECOND"] + + +def test_aliases_for_string_validation_alias() -> None: + finfo: FieldInfo = Field(default=None, validation_alias="STR_ALIAS") # type: ignore[assignment] + assert loader._aliases_for(finfo) == ["STR_ALIAS"] + + +def test_aliases_for_no_alias() -> None: + assert loader._aliases_for(FieldInfo()) == [] + + +# --------------------------------------------------------------------------- # +# apply_config_override + load_settings round-trip +# --------------------------------------------------------------------------- # + + +def test_apply_override_and_load_settings_round_trip(tmp_path: Path) -> None: + path = tmp_path / "cli-config.json" + path.write_text( + json.dumps({"env": {"STRIX_LLM": "round-trip-model", "PERPLEXITY_API_KEY": "pk"}}), + encoding="utf-8", + ) + + loader.apply_config_override(path) + settings = loader.load_settings() + + assert settings.llm.model == "round-trip-model" + assert settings.integrations.perplexity_api_key == "pk" + # Second call is memoized -> same object. + assert loader.load_settings() is settings + + +def test_apply_config_override_invalidates_cache(tmp_path: Path) -> None: + first = tmp_path / "first.json" + first.write_text(json.dumps({"env": {"STRIX_LLM": "first-model"}}), encoding="utf-8") + second = tmp_path / "second.json" + second.write_text(json.dumps({"env": {"STRIX_LLM": "second-model"}}), encoding="utf-8") + + loader.apply_config_override(first) + assert loader.load_settings().llm.model == "first-model" + + loader.apply_config_override(second) + assert loader.load_settings().llm.model == "second-model" + + +# --------------------------------------------------------------------------- # +# persist_current +# --------------------------------------------------------------------------- # + + +def test_persist_current_writes_env_block(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("STRIX_LLM", "persisted-model") + target = tmp_path / "sub" / "cli-config.json" + loader.apply_config_override(target) + + loader.persist_current() + + assert target.exists() + assert json.loads(target.read_text(encoding="utf-8")) == { + "env": {"STRIX_LLM": "persisted-model"} + } + + +def test_persist_current_sets_0600_mode(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("STRIX_LLM", "persisted-model") + target = tmp_path / "cli-config.json" + loader.apply_config_override(target) + + loader.persist_current() + + assert target.stat().st_mode & 0o777 == 0o600