mirror of
https://github.com/usestrix/strix.git
synced 2026-11-06 07:47:28 +00:00
Wait for Caido readiness by deadline
This commit is contained in:
maxi-maxima
parent
7cc9fa9faa
commit
18a36520a7
6 changed files with 1027 additions and 880 deletions
|
|
@ -1,169 +1,174 @@
|
|||
---
|
||||
title: "Configuration"
|
||||
description: "Environment variables for Strix"
|
||||
---
|
||||
|
||||
Configure Strix using environment variables or a config file.
|
||||
|
||||
## LLM Configuration
|
||||
|
||||
<ParamField path="STRIX_LLM" type="string" required>
|
||||
Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_API_KEY" type="string">
|
||||
API key for your LLM provider. Not required for local models or cloud provider auth (Vertex AI, AWS Bedrock).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_API_BASE" type="string">
|
||||
Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_EXTRA_HEADERS" type="string">
|
||||
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.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_TIMEOUT" default="300" type="integer">
|
||||
Request timeout in seconds for LLM calls.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_LLM_MAX_RETRIES" default="5" type="integer">
|
||||
Maximum number of retries for LLM API calls on transient failures.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_REASONING_EFFORT" default="high" type="string">
|
||||
Control thinking effort for reasoning models. Valid values: `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max`. Defaults to `medium` for quick scan mode.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_MEMORY_COMPRESSOR_TIMEOUT" default="30" type="integer">
|
||||
Timeout in seconds for memory compression operations (context summarization).
|
||||
</ParamField>
|
||||
|
||||
### 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.
|
||||
|
||||
<ParamField path="STRIX_DEDUPE_MODEL" type="string">
|
||||
Model used to judge whether a candidate finding duplicates an existing report.
|
||||
Falls back to `STRIX_LLM` when unset.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DEDUPE_LLM_API_KEY" type="string">
|
||||
Optional provider key for the deduplication model.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DEDUPE_LLM_API_BASE" type="string">
|
||||
Optional custom API base URL for the deduplication model. Use when the dedupe
|
||||
model runs on a different endpoint than the main model.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DEDUPE_LLM_EXTRA_HEADERS" type="string">
|
||||
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.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_DEDUPE_REASONING_EFFORT" type="string">
|
||||
Reasoning effort for the deduplication model. Defaults to the model's own
|
||||
baseline when unset.
|
||||
</ParamField>
|
||||
|
||||
## Optional Features
|
||||
|
||||
<ParamField path="PERPLEXITY_API_KEY" type="string">
|
||||
API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="POSTMAN_API_KEY" type="string">
|
||||
Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://<collection-uid>`), and Postman environments (`postman://<collection-uid>?env=<environment-uid>`) to resolve collection variables. Not needed when passing a local collection export file.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_TELEMETRY" default="1" type="string">
|
||||
Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="TRACELOOP_BASE_URL" type="string">
|
||||
OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="TRACELOOP_API_KEY" type="string">
|
||||
API key used for remote trace export. Remote export is enabled only when both `TRACELOOP_BASE_URL` and `TRACELOOP_API_KEY` are set.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="TRACELOOP_HEADERS" type="string">
|
||||
Optional custom OTEL headers (JSON object or `key=value,key2=value2`). Useful for Langfuse or custom/self-hosted OTLP gateways.
|
||||
</ParamField>
|
||||
|
||||
When remote OTEL vars are not set, Strix still writes complete run telemetry locally to:
|
||||
|
||||
```bash
|
||||
strix_runs/<run_name>/events.jsonl
|
||||
```
|
||||
|
||||
When remote vars are set, Strix dual-writes telemetry to both local JSONL and the remote OTEL endpoint.
|
||||
|
||||
## Docker Configuration
|
||||
|
||||
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.3.0" type="string">
|
||||
Docker image to use for the sandbox container.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DOCKER_HOST" type="string">
|
||||
Docker daemon socket path. Use for remote Docker hosts or custom configurations.
|
||||
</ParamField>
|
||||
|
||||
---
|
||||
title: "Configuration"
|
||||
description: "Environment variables for Strix"
|
||||
---
|
||||
|
||||
Configure Strix using environment variables or a config file.
|
||||
|
||||
## LLM Configuration
|
||||
|
||||
<ParamField path="STRIX_LLM" type="string" required>
|
||||
Model name in LiteLLM format (e.g., `openai/gpt-5.4`, `anthropic/claude-sonnet-4-6`).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_API_KEY" type="string">
|
||||
API key for your LLM provider. Not required for local models or cloud provider auth (Vertex AI, AWS Bedrock).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_API_BASE" type="string">
|
||||
Custom API base URL. Also accepts `OPENAI_API_BASE`, `LITELLM_BASE_URL`, or `OLLAMA_API_BASE`.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_EXTRA_HEADERS" type="string">
|
||||
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.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="LLM_TIMEOUT" default="300" type="integer">
|
||||
Request timeout in seconds for LLM calls.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_LLM_MAX_RETRIES" default="5" type="integer">
|
||||
Maximum number of retries for LLM API calls on transient failures.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_REASONING_EFFORT" default="high" type="string">
|
||||
Control thinking effort for reasoning models. Valid values: `none`, `minimal`, `low`, `medium`, `high`, `xhigh`, `max`. Defaults to `medium` for quick scan mode.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_MEMORY_COMPRESSOR_TIMEOUT" default="30" type="integer">
|
||||
Timeout in seconds for memory compression operations (context summarization).
|
||||
</ParamField>
|
||||
|
||||
### 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.
|
||||
|
||||
<ParamField path="STRIX_DEDUPE_MODEL" type="string">
|
||||
Model used to judge whether a candidate finding duplicates an existing report.
|
||||
Falls back to `STRIX_LLM` when unset.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DEDUPE_LLM_API_KEY" type="string">
|
||||
Optional provider key for the deduplication model.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DEDUPE_LLM_API_BASE" type="string">
|
||||
Optional custom API base URL for the deduplication model. Use when the dedupe
|
||||
model runs on a different endpoint than the main model.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DEDUPE_LLM_EXTRA_HEADERS" type="string">
|
||||
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.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_DEDUPE_REASONING_EFFORT" type="string">
|
||||
Reasoning effort for the deduplication model. Defaults to the model's own
|
||||
baseline when unset.
|
||||
</ParamField>
|
||||
|
||||
## Optional Features
|
||||
|
||||
<ParamField path="PERPLEXITY_API_KEY" type="string">
|
||||
API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="POSTMAN_API_KEY" type="string">
|
||||
Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://<collection-uid>`), and Postman environments (`postman://<collection-uid>?env=<environment-uid>`) to resolve collection variables. Not needed when passing a local collection export file.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_TELEMETRY" default="1" type="string">
|
||||
Telemetry toggle. Set to `0`, `false`, `no`, or `off` to disable telemetry (PostHog, Scarf, OTEL).
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="TRACELOOP_BASE_URL" type="string">
|
||||
OTLP/Traceloop base URL for remote OpenTelemetry export. If unset, Strix keeps traces local only.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="TRACELOOP_API_KEY" type="string">
|
||||
API key used for remote trace export. Remote export is enabled only when both `TRACELOOP_BASE_URL` and `TRACELOOP_API_KEY` are set.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="TRACELOOP_HEADERS" type="string">
|
||||
Optional custom OTEL headers (JSON object or `key=value,key2=value2`). Useful for Langfuse or custom/self-hosted OTLP gateways.
|
||||
</ParamField>
|
||||
|
||||
When remote OTEL vars are not set, Strix still writes complete run telemetry locally to:
|
||||
|
||||
```bash
|
||||
strix_runs/<run_name>/events.jsonl
|
||||
```
|
||||
|
||||
When remote vars are set, Strix dual-writes telemetry to both local JSONL and the remote OTEL endpoint.
|
||||
|
||||
## Docker Configuration
|
||||
|
||||
<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.3.0" type="string">
|
||||
Docker image to use for the sandbox container.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="DOCKER_HOST" type="string">
|
||||
Docker daemon socket path. Use for remote Docker hosts or custom configurations.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_RUNTIME_BACKEND" default="docker" type="string">
|
||||
Runtime backend for the sandbox environment.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_CAIDO_BOOT_WAIT_S" default="300" type="integer">
|
||||
Maximum time in seconds to wait for the in-sandbox Caido proxy to become ready.
|
||||
Increase this on slow or heavily loaded container hosts.
|
||||
</ParamField>
|
||||
|
||||
## Sandbox Configuration
|
||||
|
||||
<ParamField path="STRIX_SANDBOX_EXECUTION_TIMEOUT" default="120" type="integer">
|
||||
Maximum execution time in seconds for sandbox operations.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_SANDBOX_CONNECT_TIMEOUT" default="10" type="integer">
|
||||
Timeout in seconds for connecting to the sandbox container.
|
||||
</ParamField>
|
||||
|
||||
## 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"
|
||||
|
||||
```
|
||||
|
||||
<ParamField path="STRIX_SANDBOX_EXECUTION_TIMEOUT" default="120" type="integer">
|
||||
Maximum execution time in seconds for sandbox operations.
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="STRIX_SANDBOX_CONNECT_TIMEOUT" default="10" type="integer">
|
||||
Timeout in seconds for connecting to the sandbox container.
|
||||
</ParamField>
|
||||
|
||||
## 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"
|
||||
|
||||
```
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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/<workspace_subdir>`` 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/<workspace_subdir>`` 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)
|
||||
|
|
|
|||
104
tests/test_caido_bootstrap.py
Normal file
104
tests/test_caido_bootstrap.py
Normal file
|
|
@ -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]
|
||||
|
|
@ -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
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue