refactor(config): pydantic-settings revamp + drop `is_whitebox` plumbing
Replaces 200+ lines of bespoke env-loader / persist / change-detection
machinery with ``pydantic_settings.BaseSettings`` (already a transitive
of ``openai-agents → mcp``, no new direct dep).
What was wrong with ``Config``:
- 14 knobs flat in one namespace, weak grouping by comment-block.
- ``Config._applied_from_default`` and ``Config._config_file_override``
were externally mutated from ``interface/main.py:532-534``. Private
members were part of the public contract.
- Stringly-typed values: every caller had to coerce
(``int(Config.get("llm_timeout") or "300")``,
``... not in {"0", "false", "no", "off"}``).
- Dead knob: ``strix_llm_max_retries`` declared, persisted, listed in
``_LLM_CANONICAL_NAMES`` — zero readers (``DEFAULT_RETRY``
hardcodes ``max_retries=5``). Dropped.
- ``_LLM_CANONICAL_NAMES`` tuple maintained alongside class vars —
duplicate source of truth.
- ``_tracked_names()`` introspected ``vars(cls).items()`` filtered on
``(v is None or isinstance(v, str))`` — fragile.
- Awkward path: ``strix/config/config.py`` inside ``strix/config/``
with ``__init__.py`` just re-exporting.
- Dual access for the same fact: ``web_search`` read
``os.getenv("PERPLEXITY_API_KEY")`` while ``main.py`` read
``Config.get("perplexity_api_key")``.
New shape:
- ``strix/config/settings.py`` — typed dataclass tree:
``Settings.{llm,runtime,telemetry,integrations}``. Each sub-model is
its own ``BaseSettings`` so it reads env independently. Field-level
``alias=`` and ``validation_alias=AliasChoices(...)`` mirror the
existing flat env-var names — user-facing env contract is unchanged.
Bool fields auto-parse ``"0"``/``"false"``/``"no"``/``"off"``;
int fields auto-coerce.
- ``strix/config/loader.py`` — thin ``load_settings()``,
``apply_config_override(path)``, ``persist_current()`` with module
cache. JSON file reader walks aliases to populate sub-models, dropping
entries already covered by env (so env still wins).
- 13 callsites migrated from ``Config.get("...")`` to
``load_settings().<group>.<field>``.
- ``posthog._is_enabled()`` collapses to one line.
- ``--config <path>`` flow simplified: one
``apply_config_override(...)`` call replaces three lines of
class-private mutation.
Drive-by — drop ``is_whitebox`` from ``scan_config`` dict:
- It was being derived as ``bool(args.local_sources)`` in three places
(``cli.py``, ``tui.py``, ``main.py``) and stuffed into the dict for
``entry.py`` to read back. The fact is fully derivable from
``scan_config["targets"]`` — any target with ``type == "local_code"``.
- New helper ``is_whitebox_scan(targets)`` in ``interface/utils.py``
alongside the other target-classification utilities.
- ``entry.py`` computes once; ``main.py``'s posthog start uses the same
helper. Triplicate derivation gone.
Verified: ruff at baseline (3), mypy at baseline (69). Six smoke tests
pass — defaults / JSON-only / env-wins-over-JSON / alias-chain
fallback / bool parsing / ``is_whitebox_scan``.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,12 +1,37 @@
|
||||
from strix.config.config import (
|
||||
Config,
|
||||
apply_saved_config,
|
||||
save_current_config,
|
||||
"""Strix application settings.
|
||||
|
||||
Public surface:
|
||||
|
||||
- :class:`Settings` — composite model. Get via :func:`load_settings`.
|
||||
- :class:`LlmSettings`, :class:`RuntimeSettings`, :class:`TelemetrySettings`,
|
||||
:class:`IntegrationSettings` — sub-models, attribute-accessed off
|
||||
``Settings``.
|
||||
- :func:`load_settings` — memoized resolve (env > JSON file > defaults).
|
||||
- :func:`apply_config_override` — switch the JSON source to a custom path.
|
||||
- :func:`persist_current` — write currently-set env vars to the active file.
|
||||
"""
|
||||
|
||||
from strix.config.loader import (
|
||||
apply_config_override,
|
||||
load_settings,
|
||||
persist_current,
|
||||
)
|
||||
from strix.config.settings import (
|
||||
IntegrationSettings,
|
||||
LlmSettings,
|
||||
RuntimeSettings,
|
||||
Settings,
|
||||
TelemetrySettings,
|
||||
)
|
||||
|
||||
|
||||
__all__ = [
|
||||
"Config",
|
||||
"apply_saved_config",
|
||||
"save_current_config",
|
||||
"IntegrationSettings",
|
||||
"LlmSettings",
|
||||
"RuntimeSettings",
|
||||
"Settings",
|
||||
"TelemetrySettings",
|
||||
"apply_config_override",
|
||||
"load_settings",
|
||||
"persist_current",
|
||||
]
|
||||
|
||||
@@ -1,207 +0,0 @@
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import Any, ClassVar
|
||||
|
||||
|
||||
class Config:
|
||||
"""Configuration Manager for Strix."""
|
||||
|
||||
# LLM Configuration
|
||||
strix_llm = None
|
||||
llm_api_key = None
|
||||
llm_api_base = None
|
||||
openai_api_base = None
|
||||
litellm_base_url = None
|
||||
ollama_api_base = None
|
||||
strix_reasoning_effort = "high"
|
||||
strix_llm_max_retries = "5"
|
||||
llm_timeout = "300"
|
||||
_LLM_CANONICAL_NAMES = (
|
||||
"strix_llm",
|
||||
"llm_api_key",
|
||||
"llm_api_base",
|
||||
"openai_api_base",
|
||||
"litellm_base_url",
|
||||
"ollama_api_base",
|
||||
"strix_reasoning_effort",
|
||||
"strix_llm_max_retries",
|
||||
"llm_timeout",
|
||||
)
|
||||
|
||||
# Tool & Feature Configuration
|
||||
perplexity_api_key = None
|
||||
|
||||
# Runtime Configuration
|
||||
strix_image = "ghcr.io/usestrix/strix-sandbox:0.1.13"
|
||||
strix_runtime_backend = "docker"
|
||||
|
||||
# Telemetry
|
||||
strix_telemetry = "1"
|
||||
strix_posthog_telemetry = None
|
||||
|
||||
# Config file override (set via --config CLI arg)
|
||||
_config_file_override: Path | None = None
|
||||
|
||||
# Tracks env vars set by the initial default-config load so they can be
|
||||
# cleared when a --config override is later applied (avoids leakage).
|
||||
_applied_from_default: ClassVar[dict[str, str]] = {}
|
||||
|
||||
@classmethod
|
||||
def _tracked_names(cls) -> list[str]:
|
||||
return [
|
||||
k
|
||||
for k, v in vars(cls).items()
|
||||
if not k.startswith("_") and k[0].islower() and (v is None or isinstance(v, str))
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def tracked_vars(cls) -> list[str]:
|
||||
return [name.upper() for name in cls._tracked_names()]
|
||||
|
||||
@classmethod
|
||||
def _llm_env_vars(cls) -> set[str]:
|
||||
return {name.upper() for name in cls._LLM_CANONICAL_NAMES}
|
||||
|
||||
@classmethod
|
||||
def _llm_env_changed(cls, saved_env: dict[str, Any]) -> bool:
|
||||
for var_name in cls._llm_env_vars():
|
||||
current = os.getenv(var_name)
|
||||
if current is None:
|
||||
continue
|
||||
if saved_env.get(var_name) != current:
|
||||
return True
|
||||
return False
|
||||
|
||||
@classmethod
|
||||
def get(cls, name: str) -> str | None:
|
||||
env_name = name.upper()
|
||||
default = getattr(cls, name, None)
|
||||
return os.getenv(env_name, default)
|
||||
|
||||
@classmethod
|
||||
def config_dir(cls) -> Path:
|
||||
return Path.home() / ".strix"
|
||||
|
||||
@classmethod
|
||||
def config_file(cls) -> Path:
|
||||
if cls._config_file_override is not None:
|
||||
return cls._config_file_override
|
||||
return cls.config_dir() / "cli-config.json"
|
||||
|
||||
@classmethod
|
||||
def load(cls) -> dict[str, Any]:
|
||||
path = cls.config_file()
|
||||
if not path.exists():
|
||||
return {}
|
||||
try:
|
||||
with path.open("r", encoding="utf-8") as f:
|
||||
data: dict[str, Any] = json.load(f)
|
||||
return data
|
||||
except (json.JSONDecodeError, OSError):
|
||||
return {}
|
||||
|
||||
@classmethod
|
||||
def save(cls, config: dict[str, Any]) -> bool:
|
||||
try:
|
||||
cls.config_dir().mkdir(parents=True, exist_ok=True)
|
||||
config_path = cls.config_dir() / "cli-config.json"
|
||||
with config_path.open("w", encoding="utf-8") as f:
|
||||
json.dump(config, f, indent=2)
|
||||
except OSError:
|
||||
return False
|
||||
with contextlib.suppress(OSError):
|
||||
config_path.chmod(0o600) # may fail on Windows
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def apply_saved(cls, force: bool = False) -> dict[str, str]:
|
||||
saved = cls.load()
|
||||
env_vars = saved.get("env", {})
|
||||
if not isinstance(env_vars, dict):
|
||||
env_vars = {}
|
||||
cleared_vars = {
|
||||
var_name
|
||||
for var_name in cls.tracked_vars()
|
||||
if var_name in os.environ and os.environ.get(var_name) == ""
|
||||
}
|
||||
if cleared_vars:
|
||||
for var_name in cleared_vars:
|
||||
env_vars.pop(var_name, None)
|
||||
if cls._config_file_override is None:
|
||||
cls.save({"env": env_vars})
|
||||
if cls._llm_env_changed(env_vars):
|
||||
for var_name in cls._llm_env_vars():
|
||||
env_vars.pop(var_name, None)
|
||||
if cls._config_file_override is None:
|
||||
cls.save({"env": env_vars})
|
||||
applied = {}
|
||||
|
||||
for var_name, var_value in env_vars.items():
|
||||
if var_name in cls.tracked_vars() and (force or var_name not in os.environ):
|
||||
os.environ[var_name] = var_value
|
||||
applied[var_name] = var_value
|
||||
|
||||
# Record what was applied from the default config so it can be cleared
|
||||
# if a --config override is later provided (prevents leakage).
|
||||
if cls._config_file_override is None and not force:
|
||||
cls._applied_from_default = applied
|
||||
|
||||
return applied
|
||||
|
||||
@classmethod
|
||||
def capture_current(cls) -> dict[str, Any]:
|
||||
env_vars = {}
|
||||
for var_name in cls.tracked_vars():
|
||||
value = os.getenv(var_name)
|
||||
if value:
|
||||
env_vars[var_name] = value
|
||||
return {"env": env_vars}
|
||||
|
||||
@classmethod
|
||||
def save_current(cls) -> bool:
|
||||
existing = cls.load().get("env", {})
|
||||
merged = dict(existing)
|
||||
|
||||
for var_name in cls.tracked_vars():
|
||||
value = os.getenv(var_name)
|
||||
if value is None:
|
||||
pass
|
||||
elif value == "":
|
||||
merged.pop(var_name, None)
|
||||
else:
|
||||
merged[var_name] = value
|
||||
|
||||
return cls.save({"env": merged})
|
||||
|
||||
|
||||
def apply_saved_config(force: bool = False) -> dict[str, str]:
|
||||
return Config.apply_saved(force=force)
|
||||
|
||||
|
||||
def save_current_config() -> bool:
|
||||
return Config.save_current()
|
||||
|
||||
|
||||
def resolve_llm_config() -> tuple[str | None, str | None, str | None]:
|
||||
"""Resolve LLM model, api_key, and api_base.
|
||||
|
||||
Returns ``(model_name, api_key, api_base)``. ``api_base`` falls back
|
||||
through the ``LLM_API_BASE`` / ``OPENAI_API_BASE`` /
|
||||
``LITELLM_BASE_URL`` / ``OLLAMA_API_BASE`` env chain so the user can
|
||||
point at any OpenAI-compatible endpoint without changing the code.
|
||||
"""
|
||||
model = Config.get("strix_llm")
|
||||
if not model:
|
||||
return None, None, None
|
||||
|
||||
api_key = Config.get("llm_api_key")
|
||||
api_base: str | None = (
|
||||
Config.get("llm_api_base")
|
||||
or Config.get("openai_api_base")
|
||||
or Config.get("litellm_base_url")
|
||||
or Config.get("ollama_api_base")
|
||||
)
|
||||
|
||||
return model, api_key, api_base
|
||||
@@ -0,0 +1,123 @@
|
||||
"""Settings loader, override switch, and disk persistence.
|
||||
|
||||
Process-wide module cache so repeated ``load_settings()`` calls in the
|
||||
same scan are free. ``apply_config_override(path)`` invalidates the
|
||||
cache so the next ``load_settings()`` re-resolves with the new file.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import os
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from pydantic import AliasChoices, BaseModel
|
||||
|
||||
from strix.config.settings import Settings
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from pydantic.fields import FieldInfo
|
||||
|
||||
|
||||
_DEFAULT_PATH: Path = Path.home() / ".strix" / "cli-config.json"
|
||||
_override: Path | None = None
|
||||
_cached: Settings | None = None
|
||||
|
||||
|
||||
def load_settings() -> Settings:
|
||||
"""Resolve settings from env + JSON file + defaults. Memoized.
|
||||
|
||||
Precedence: env vars win, then the JSON file, then field defaults.
|
||||
"""
|
||||
global _cached # noqa: PLW0603
|
||||
if _cached is None:
|
||||
init_kwargs: dict[str, Any] = _read_json_overrides(_override or _DEFAULT_PATH)
|
||||
_cached = Settings(**init_kwargs)
|
||||
return _cached
|
||||
|
||||
|
||||
def apply_config_override(path: Path) -> None:
|
||||
"""Switch the JSON source to ``path`` and invalidate the cache."""
|
||||
global _override, _cached # noqa: PLW0603
|
||||
_override = path
|
||||
_cached = None
|
||||
|
||||
|
||||
def persist_current() -> None:
|
||||
"""Write currently-set env vars to the active config file (0o600)."""
|
||||
s = load_settings()
|
||||
target = _override or _DEFAULT_PATH
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
env_block: dict[str, str] = {}
|
||||
for sub_name in s.model_fields:
|
||||
sub_model = getattr(s, sub_name)
|
||||
if not isinstance(sub_model, BaseModel):
|
||||
continue
|
||||
for finfo in type(sub_model).model_fields.values():
|
||||
for alias in _aliases_for(finfo):
|
||||
value = os.environ.get(alias.upper())
|
||||
if value:
|
||||
env_block[alias.upper()] = value
|
||||
break
|
||||
|
||||
target.write_text(json.dumps({"env": env_block}, indent=2), encoding="utf-8")
|
||||
with contextlib.suppress(OSError):
|
||||
target.chmod(0o600)
|
||||
|
||||
|
||||
# --- internals ---------------------------------------------------------
|
||||
|
||||
|
||||
def _aliases_for(finfo: FieldInfo) -> list[str]:
|
||||
"""Collect every env-var name that should populate ``finfo``."""
|
||||
aliases: list[str] = []
|
||||
if finfo.alias:
|
||||
aliases.append(finfo.alias)
|
||||
va = finfo.validation_alias
|
||||
if isinstance(va, AliasChoices):
|
||||
aliases.extend(c for c in va.choices if isinstance(c, str))
|
||||
elif isinstance(va, str):
|
||||
aliases.append(va)
|
||||
return aliases
|
||||
|
||||
|
||||
def _read_json_overrides(path: Path) -> dict[str, dict[str, Any]]:
|
||||
"""Read ``{"env": {...}}`` from ``path`` and remap to nested kwargs.
|
||||
|
||||
Only includes keys whose env var is NOT already set, so env always
|
||||
wins over the persisted file.
|
||||
"""
|
||||
if not path.exists():
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
except (json.JSONDecodeError, OSError):
|
||||
return {}
|
||||
env_block = data.get("env", {}) if isinstance(data, dict) else {}
|
||||
if not isinstance(env_block, dict):
|
||||
return {}
|
||||
|
||||
# Normalize to upper-case keys for matching.
|
||||
env_block_upper = {str(k).upper(): v for k, v in env_block.items()}
|
||||
|
||||
nested: dict[str, dict[str, Any]] = {}
|
||||
for sub_name, sub_finfo in Settings.model_fields.items():
|
||||
sub_cls = sub_finfo.annotation
|
||||
if not (isinstance(sub_cls, type) and issubclass(sub_cls, BaseModel)):
|
||||
continue
|
||||
sub_data: dict[str, Any] = {}
|
||||
for fname, finfo in sub_cls.model_fields.items():
|
||||
for alias in _aliases_for(finfo):
|
||||
key = alias.upper()
|
||||
if key in os.environ:
|
||||
break # env wins; skip JSON for this field
|
||||
if key in env_block_upper:
|
||||
sub_data[fname] = env_block_upper[key]
|
||||
break
|
||||
if sub_data:
|
||||
nested[sub_name] = sub_data
|
||||
return nested
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Strix application settings — pydantic-settings powered.
|
||||
|
||||
Three sources, env-precedence-first:
|
||||
|
||||
1. Environment variables (``STRIX_LLM``, ``LLM_API_KEY``, etc.) — highest.
|
||||
2. ``~/.strix/cli-config.json`` (or ``--config <path>``) — middle.
|
||||
3. Field defaults — lowest.
|
||||
|
||||
Bool fields auto-parse ``"0"``/``"false"``/``"no"``/``"off"`` as falsy
|
||||
and any other non-empty string as truthy. Int fields auto-coerce from
|
||||
string env. The ``api_base`` field walks an alias chain so users can
|
||||
point at any OpenAI-compatible endpoint via whichever env name they
|
||||
prefer (``LLM_API_BASE`` / ``OPENAI_API_BASE`` / ``LITELLM_BASE_URL`` /
|
||||
``OLLAMA_API_BASE``).
|
||||
|
||||
Each sub-model is a :class:`BaseSettings` so it reads env independently
|
||||
— the alternative (one mega-BaseSettings with flat fields) would lose
|
||||
the logical grouping ``s.llm.model`` / ``s.runtime.image`` / etc.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Literal
|
||||
|
||||
from pydantic import AliasChoices, Field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
||||
ReasoningEffort = Literal["low", "medium", "high"]
|
||||
|
||||
_BASE_CONFIG = SettingsConfigDict(
|
||||
case_sensitive=False,
|
||||
populate_by_name=True,
|
||||
extra="ignore",
|
||||
)
|
||||
|
||||
|
||||
class LlmSettings(BaseSettings):
|
||||
"""LLM provider + model + per-call defaults."""
|
||||
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
model: str | None = Field(default=None, alias="STRIX_LLM")
|
||||
api_key: str | None = Field(default=None, alias="LLM_API_KEY")
|
||||
api_base: str | None = Field(
|
||||
default=None,
|
||||
validation_alias=AliasChoices(
|
||||
"LLM_API_BASE",
|
||||
"OPENAI_API_BASE",
|
||||
"LITELLM_BASE_URL",
|
||||
"OLLAMA_API_BASE",
|
||||
),
|
||||
)
|
||||
reasoning_effort: ReasoningEffort = Field(default="high", alias="STRIX_REASONING_EFFORT")
|
||||
timeout: int = Field(default=300, alias="LLM_TIMEOUT")
|
||||
|
||||
|
||||
class RuntimeSettings(BaseSettings):
|
||||
"""Sandbox image + backend selector."""
|
||||
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
image: str = Field(
|
||||
default="ghcr.io/usestrix/strix-sandbox:0.1.13",
|
||||
alias="STRIX_IMAGE",
|
||||
)
|
||||
backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
|
||||
|
||||
|
||||
class TelemetrySettings(BaseSettings):
|
||||
"""Telemetry toggles. ``posthog`` is None → inherit ``master``."""
|
||||
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
master: bool = Field(default=True, alias="STRIX_TELEMETRY")
|
||||
posthog: bool | None = Field(default=None, alias="STRIX_POSTHOG_TELEMETRY")
|
||||
|
||||
@property
|
||||
def posthog_enabled(self) -> bool:
|
||||
"""Effective PostHog toggle: explicit value if set, else ``master``."""
|
||||
return self.master if self.posthog is None else self.posthog
|
||||
|
||||
|
||||
class IntegrationSettings(BaseSettings):
|
||||
"""Third-party integration credentials."""
|
||||
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
perplexity_api_key: str | None = Field(default=None, alias="PERPLEXITY_API_KEY")
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
"""Composite Strix settings. Instantiate via :func:`strix.config.load_settings`."""
|
||||
|
||||
model_config = _BASE_CONFIG
|
||||
|
||||
llm: LlmSettings = Field(default_factory=LlmSettings)
|
||||
runtime: RuntimeSettings = Field(default_factory=RuntimeSettings)
|
||||
telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings)
|
||||
integrations: IntegrationSettings = Field(default_factory=IntegrationSettings)
|
||||
+6
-4
@@ -24,7 +24,7 @@ from agents import Runner
|
||||
from agents.memory import SQLiteSession
|
||||
|
||||
from strix.agents.factory import build_strix_agent, make_child_factory
|
||||
from strix.config.config import Config
|
||||
from strix.config import load_settings
|
||||
from strix.orchestration.bus import AgentMessageBus
|
||||
from strix.orchestration.hooks import StrixOrchestrationHooks
|
||||
from strix.run_config_factory import (
|
||||
@@ -206,8 +206,11 @@ async def run_strix_scan(
|
||||
)
|
||||
|
||||
try:
|
||||
# Lazy: ``strix.interface`` pulls cli→tui→entry which would cycle.
|
||||
from strix.interface.utils import is_whitebox_scan
|
||||
|
||||
scan_mode = str(scan_config.get("scan_mode") or "deep")
|
||||
is_whitebox = bool(scan_config.get("is_whitebox", False))
|
||||
is_whitebox = is_whitebox_scan(scan_config.get("targets") or [])
|
||||
skills = list(scan_config.get("skills") or [])
|
||||
diff_scope = scan_config.get("diff_scope") or None
|
||||
run_id = scan_config.get("run_id") or scan_id
|
||||
@@ -249,9 +252,8 @@ async def run_strix_scan(
|
||||
agent_factory=agent_factory,
|
||||
)
|
||||
|
||||
reasoning = Config.get("strix_reasoning_effort")
|
||||
reasoning_effort: Literal["low", "medium", "high"] | None = (
|
||||
reasoning if reasoning in ("low", "medium", "high") else None # type: ignore[assignment]
|
||||
load_settings().llm.reasoning_effort
|
||||
)
|
||||
run_config = make_run_config(
|
||||
sandbox_session=bundle["session"],
|
||||
|
||||
@@ -13,7 +13,7 @@ from rich.live import Live
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import Config
|
||||
from strix.config import load_settings
|
||||
from strix.entry import run_strix_scan
|
||||
from strix.runtime import session_manager
|
||||
from strix.telemetry.tracer import Tracer, set_global_tracer
|
||||
@@ -25,12 +25,12 @@ from .utils import (
|
||||
|
||||
|
||||
def _resolve_sandbox_image() -> str:
|
||||
image = Config.get("strix_image")
|
||||
image = load_settings().runtime.image
|
||||
if not image:
|
||||
raise RuntimeError(
|
||||
"strix_image is not configured. Set it in ~/.strix/cli-config.json.",
|
||||
)
|
||||
return str(image)
|
||||
return image
|
||||
|
||||
|
||||
def _resolve_sources_path(args: Any) -> Path:
|
||||
@@ -99,7 +99,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
console.print()
|
||||
|
||||
scan_mode = getattr(args, "scan_mode", "deep")
|
||||
is_whitebox = bool(getattr(args, "local_sources", []))
|
||||
|
||||
scan_config: dict[str, Any] = {
|
||||
"scan_id": args.run_name,
|
||||
@@ -108,7 +107,6 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
"run_name": args.run_name,
|
||||
"diff_scope": getattr(args, "diff_scope", {"active": False}),
|
||||
"scan_mode": scan_mode,
|
||||
"is_whitebox": is_whitebox,
|
||||
}
|
||||
|
||||
tracer = Tracer(args.run_name)
|
||||
|
||||
+30
-65
@@ -6,7 +6,6 @@ Strix Agent Interface
|
||||
import argparse
|
||||
import asyncio
|
||||
import logging
|
||||
import os
|
||||
import shutil
|
||||
import sys
|
||||
from pathlib import Path
|
||||
@@ -18,15 +17,10 @@ from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import Config, apply_saved_config, save_current_config
|
||||
from strix.config.config import resolve_llm_config
|
||||
|
||||
|
||||
apply_saved_config()
|
||||
|
||||
from strix.interface.cli import run_cli # noqa: E402
|
||||
from strix.interface.tui import run_tui # noqa: E402
|
||||
from strix.interface.utils import ( # noqa: E402
|
||||
from strix.config import apply_config_override, load_settings, persist_current
|
||||
from strix.interface.cli import run_cli
|
||||
from strix.interface.tui import run_tui
|
||||
from strix.interface.utils import (
|
||||
assign_workspace_subdirs,
|
||||
build_final_stats_text,
|
||||
check_docker_connection,
|
||||
@@ -35,17 +29,18 @@ from strix.interface.utils import ( # noqa: E402
|
||||
generate_run_name,
|
||||
image_exists,
|
||||
infer_target_type,
|
||||
is_whitebox_scan,
|
||||
process_pull_line,
|
||||
resolve_diff_scope_context,
|
||||
rewrite_localhost_targets,
|
||||
validate_config_file,
|
||||
validate_llm_response,
|
||||
)
|
||||
from strix.telemetry import posthog
|
||||
from strix.telemetry.tracer import get_global_tracer
|
||||
|
||||
|
||||
HOST_GATEWAY_HOSTNAME = "host.docker.internal"
|
||||
from strix.telemetry import posthog # noqa: E402
|
||||
from strix.telemetry.tracer import get_global_tracer # noqa: E402
|
||||
|
||||
|
||||
logging.getLogger().setLevel(logging.ERROR)
|
||||
@@ -56,31 +51,20 @@ def validate_environment() -> None:
|
||||
missing_required_vars = []
|
||||
missing_optional_vars = []
|
||||
|
||||
strix_llm = Config.get("strix_llm")
|
||||
if not strix_llm:
|
||||
settings = load_settings()
|
||||
|
||||
if not settings.llm.model:
|
||||
missing_required_vars.append("STRIX_LLM")
|
||||
|
||||
has_base_url = any(
|
||||
[
|
||||
Config.get("llm_api_base"),
|
||||
Config.get("openai_api_base"),
|
||||
Config.get("litellm_base_url"),
|
||||
Config.get("ollama_api_base"),
|
||||
]
|
||||
)
|
||||
|
||||
if not Config.get("llm_api_key"):
|
||||
if not settings.llm.api_key:
|
||||
missing_optional_vars.append("LLM_API_KEY")
|
||||
|
||||
if not has_base_url:
|
||||
if not settings.llm.api_base:
|
||||
missing_optional_vars.append("LLM_API_BASE")
|
||||
|
||||
if not Config.get("perplexity_api_key"):
|
||||
if not settings.integrations.perplexity_api_key:
|
||||
missing_optional_vars.append("PERPLEXITY_API_KEY")
|
||||
|
||||
if not Config.get("strix_reasoning_effort"):
|
||||
missing_optional_vars.append("STRIX_REASONING_EFFORT")
|
||||
|
||||
if missing_required_vars:
|
||||
error_text = Text()
|
||||
error_text.append("MISSING REQUIRED ENVIRONMENT VARIABLES", style="bold red")
|
||||
@@ -207,25 +191,22 @@ async def warm_up_llm() -> None:
|
||||
console = Console()
|
||||
|
||||
try:
|
||||
model_name, api_key, api_base = resolve_llm_config()
|
||||
litellm_model: str | None = model_name
|
||||
llm = load_settings().llm
|
||||
|
||||
test_messages = [
|
||||
{"role": "system", "content": "You are a helpful assistant."},
|
||||
{"role": "user", "content": "Reply with just 'OK'."},
|
||||
]
|
||||
|
||||
llm_timeout = int(Config.get("llm_timeout") or "300")
|
||||
|
||||
completion_kwargs: dict[str, Any] = {
|
||||
"model": litellm_model,
|
||||
"model": llm.model,
|
||||
"messages": test_messages,
|
||||
"timeout": llm_timeout,
|
||||
"timeout": llm.timeout,
|
||||
}
|
||||
if api_key:
|
||||
completion_kwargs["api_key"] = api_key
|
||||
if api_base:
|
||||
completion_kwargs["api_base"] = api_base
|
||||
if llm.api_key:
|
||||
completion_kwargs["api_key"] = llm.api_key
|
||||
if llm.api_base:
|
||||
completion_kwargs["api_base"] = llm.api_base
|
||||
|
||||
response = litellm.completion(**completion_kwargs)
|
||||
|
||||
@@ -486,11 +467,13 @@ def pull_docker_image() -> None:
|
||||
console = Console()
|
||||
client = check_docker_connection()
|
||||
|
||||
if image_exists(client, Config.get("strix_image")): # type: ignore[arg-type]
|
||||
image = load_settings().runtime.image
|
||||
|
||||
if image_exists(client, image):
|
||||
return
|
||||
|
||||
console.print()
|
||||
console.print(f"[dim]Pulling image[/] {Config.get('strix_image')}")
|
||||
console.print(f"[dim]Pulling image[/] {image}")
|
||||
console.print("[dim yellow]This only happens on first run and may take a few minutes...[/]")
|
||||
console.print()
|
||||
|
||||
@@ -499,7 +482,7 @@ def pull_docker_image() -> None:
|
||||
layers_info: dict[str, str] = {}
|
||||
last_update = ""
|
||||
|
||||
for line in client.api.pull(Config.get("strix_image"), stream=True, decode=True):
|
||||
for line in client.api.pull(image, stream=True, decode=True):
|
||||
last_update = process_pull_line(line, layers_info, status, last_update)
|
||||
|
||||
except DockerException as e:
|
||||
@@ -507,7 +490,7 @@ def pull_docker_image() -> None:
|
||||
error_text = Text()
|
||||
error_text.append("FAILED TO PULL IMAGE", style="bold red")
|
||||
error_text.append("\n\n", style="white")
|
||||
error_text.append(f"Could not download: {Config.get('strix_image')}\n", style="white")
|
||||
error_text.append(f"Could not download: {image}\n", style="white")
|
||||
error_text.append(str(e), style="dim red")
|
||||
|
||||
panel = Panel(
|
||||
@@ -526,22 +509,6 @@ def pull_docker_image() -> None:
|
||||
console.print()
|
||||
|
||||
|
||||
def apply_config_override(config_path: str) -> None:
|
||||
# Clear env vars that were automatically applied from the default config file
|
||||
# so they don't leak into the custom config context.
|
||||
for var_name in Config._applied_from_default:
|
||||
os.environ.pop(var_name, None)
|
||||
Config._applied_from_default = {}
|
||||
|
||||
Config._config_file_override = validate_config_file(config_path)
|
||||
apply_saved_config(force=True)
|
||||
|
||||
|
||||
def persist_config() -> None:
|
||||
if Config._config_file_override is None:
|
||||
save_current_config()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if sys.platform == "win32":
|
||||
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
||||
@@ -549,7 +516,7 @@ def main() -> None:
|
||||
args = parse_arguments()
|
||||
|
||||
if args.config:
|
||||
apply_config_override(args.config)
|
||||
apply_config_override(validate_config_file(args.config))
|
||||
|
||||
check_docker_installed()
|
||||
pull_docker_image()
|
||||
@@ -557,7 +524,7 @@ def main() -> None:
|
||||
validate_environment()
|
||||
asyncio.run(warm_up_llm())
|
||||
|
||||
persist_config()
|
||||
persist_current()
|
||||
|
||||
args.run_name = generate_run_name(args.targets_info)
|
||||
|
||||
@@ -602,12 +569,10 @@ def main() -> None:
|
||||
else:
|
||||
args.instruction = diff_scope.instruction_block
|
||||
|
||||
is_whitebox = bool(args.local_sources)
|
||||
|
||||
posthog.start(
|
||||
model=Config.get("strix_llm"),
|
||||
model=load_settings().llm.model,
|
||||
scan_mode=args.scan_mode,
|
||||
is_whitebox=is_whitebox,
|
||||
is_whitebox=is_whitebox_scan(args.targets_info),
|
||||
interactive=not args.non_interactive,
|
||||
has_instructions=bool(args.instruction),
|
||||
)
|
||||
|
||||
@@ -31,7 +31,7 @@ from textual.screen import ModalScreen
|
||||
from textual.widgets import Button, Label, Static, TextArea, Tree
|
||||
from textual.widgets.tree import TreeNode
|
||||
|
||||
from strix.config import Config
|
||||
from strix.config import load_settings
|
||||
from strix.entry import run_strix_scan
|
||||
from strix.interface.tool_components.agent_message_renderer import AgentMessageRenderer
|
||||
from strix.interface.tool_components.registry import get_tool_renderer
|
||||
@@ -763,7 +763,6 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
"run_name": args.run_name,
|
||||
"diff_scope": getattr(args, "diff_scope", {"active": False}),
|
||||
"scan_mode": getattr(args, "scan_mode", "deep"),
|
||||
"is_whitebox": bool(getattr(args, "local_sources", [])),
|
||||
}
|
||||
|
||||
def _setup_cleanup_handlers(self) -> None:
|
||||
@@ -1408,7 +1407,7 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
|
||||
try:
|
||||
if not self._scan_stop_event.is_set():
|
||||
image = Config.get("strix_image") or "strix-sandbox:latest"
|
||||
image = load_settings().runtime.image or "strix-sandbox:latest"
|
||||
sources_path = self._resolve_sources_path()
|
||||
loop.run_until_complete(
|
||||
run_strix_scan(
|
||||
|
||||
@@ -20,7 +20,7 @@ from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
from rich.text import Text
|
||||
|
||||
from strix.config import Config
|
||||
from strix.config import load_settings
|
||||
|
||||
|
||||
# Token formatting utilities
|
||||
@@ -304,7 +304,7 @@ def build_live_stats_text(tracer: Any) -> Text:
|
||||
if not tracer:
|
||||
return stats_text
|
||||
|
||||
model = Config.get("strix_llm") or "unknown"
|
||||
model = load_settings().llm.model or "unknown"
|
||||
stats_text.append("Model ", style="dim")
|
||||
stats_text.append(str(model), style="white")
|
||||
stats_text.append("\n")
|
||||
@@ -375,7 +375,7 @@ def build_tui_stats_text(tracer: Any) -> Text:
|
||||
if not tracer:
|
||||
return stats_text
|
||||
|
||||
model = Config.get("strix_llm") or "unknown"
|
||||
model = load_settings().llm.model or "unknown"
|
||||
stats_text.append(str(model), style="white")
|
||||
|
||||
llm_stats = tracer.get_total_llm_stats()
|
||||
@@ -1190,6 +1190,11 @@ def assign_workspace_subdirs(targets_info: list[dict[str, Any]]) -> None:
|
||||
details["workspace_subdir"] = workspace_subdir
|
||||
|
||||
|
||||
def is_whitebox_scan(targets_info: list[dict[str, Any]]) -> bool:
|
||||
"""True iff any target is a local source tree (whitebox / source-aware)."""
|
||||
return any(t.get("type") == "local_code" for t in targets_info or [])
|
||||
|
||||
|
||||
def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str, str]]:
|
||||
local_sources: list[dict[str, str]] = []
|
||||
|
||||
|
||||
+2
-2
@@ -16,7 +16,7 @@ from agents.model_settings import ModelSettings
|
||||
from agents.models.interface import ModelTracing
|
||||
from openai.types.responses import ResponseOutputMessage
|
||||
|
||||
from strix.config.config import Config
|
||||
from strix.config import load_settings
|
||||
from strix.llm.multi_provider_setup import build_multi_provider
|
||||
from strix.run_config_factory import DEFAULT_RETRY
|
||||
|
||||
@@ -174,7 +174,7 @@ async def check_duplicate(
|
||||
}
|
||||
|
||||
try:
|
||||
model_name = Config.get("strix_llm")
|
||||
model_name = load_settings().llm.model
|
||||
if not model_name:
|
||||
return {
|
||||
"is_duplicate": False,
|
||||
|
||||
@@ -21,7 +21,7 @@ from typing import TYPE_CHECKING, Any
|
||||
from agents.sandbox.entries import LocalDir
|
||||
from agents.sandbox.manifest import Environment, Manifest
|
||||
|
||||
from strix.config.config import Config
|
||||
from strix.config import load_settings
|
||||
from strix.runtime.backends import get_backend
|
||||
from strix.runtime.caido_bootstrap import bootstrap_caido
|
||||
|
||||
@@ -86,7 +86,7 @@ async def create_or_reuse(
|
||||
),
|
||||
)
|
||||
|
||||
backend_name = Config.get("strix_runtime_backend") or "docker"
|
||||
backend_name = load_settings().runtime.backend
|
||||
backend = get_backend(backend_name)
|
||||
|
||||
logger.info(
|
||||
|
||||
@@ -6,7 +6,7 @@ from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from uuid import uuid4
|
||||
|
||||
from strix.config import Config
|
||||
from strix.config import load_settings
|
||||
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -15,18 +15,12 @@ if TYPE_CHECKING:
|
||||
_POSTHOG_PUBLIC_API_KEY = "phc_7rO3XRuNT5sgSKAl6HDIrWdSGh1COzxw0vxVIAR6vVZ"
|
||||
_POSTHOG_HOST = "https://us.i.posthog.com"
|
||||
|
||||
_DISABLED_VALUES = {"0", "false", "no", "off"}
|
||||
|
||||
_SESSION_ID = uuid4().hex[:16]
|
||||
|
||||
|
||||
def _is_enabled() -> bool:
|
||||
"""Master telemetry gate. ``STRIX_POSTHOG_TELEMETRY`` overrides ``STRIX_TELEMETRY``."""
|
||||
explicit = Config.get("strix_posthog_telemetry")
|
||||
if explicit is not None:
|
||||
return explicit.strip().lower() not in _DISABLED_VALUES
|
||||
fallback = Config.get("strix_telemetry") or "1"
|
||||
return fallback.strip().lower() not in _DISABLED_VALUES
|
||||
return load_settings().telemetry.posthog_enabled
|
||||
|
||||
|
||||
def _is_first_run() -> bool:
|
||||
|
||||
@@ -4,12 +4,13 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
from agents import RunContextWrapper, function_tool
|
||||
|
||||
from strix.config import load_settings
|
||||
|
||||
|
||||
_SYSTEM_PROMPT = """You are assisting a cybersecurity agent specialized in vulnerability scanning
|
||||
and security assessment running on Kali Linux. When responding to search queries:
|
||||
@@ -37,7 +38,7 @@ security implications and details."""
|
||||
|
||||
|
||||
def _do_search(query: str) -> dict[str, Any]:
|
||||
api_key = os.getenv("PERPLEXITY_API_KEY")
|
||||
api_key = load_settings().integrations.perplexity_api_key
|
||||
if not api_key:
|
||||
return {
|
||||
"success": False,
|
||||
|
||||
Reference in New Issue
Block a user