From 1e641e56ce7fb3df5cffe4ac40929f606500db53 Mon Sep 17 00:00:00 2001 From: 0xallam Date: Sat, 25 Apr 2026 16:05:40 -0700 Subject: [PATCH] refactor(config): pydantic-settings revamp + drop ``is_whitebox`` plumbing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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()..``. - ``posthog._is_enabled()`` collapses to one line. - ``--config `` 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) --- pyproject.toml | 2 + strix/config/__init__.py | 39 ++++-- strix/config/config.py | 207 ------------------------------- strix/config/loader.py | 123 ++++++++++++++++++ strix/config/settings.py | 100 +++++++++++++++ strix/entry.py | 10 +- strix/interface/cli.py | 8 +- strix/interface/main.py | 95 +++++--------- strix/interface/tui.py | 5 +- strix/interface/utils.py | 11 +- strix/llm/dedupe.py | 4 +- strix/runtime/session_manager.py | 4 +- strix/telemetry/posthog.py | 10 +- strix/tools/web_search/tool.py | 5 +- uv.lock | 2 + 15 files changed, 317 insertions(+), 308 deletions(-) delete mode 100644 strix/config/config.py create mode 100644 strix/config/loader.py create mode 100644 strix/config/settings.py diff --git a/pyproject.toml b/pyproject.toml index 22cf775..4081c27 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -35,6 +35,7 @@ classifiers = [ dependencies = [ "openai-agents[litellm]==0.14.6", "pydantic>=2.11.3", + "pydantic-settings>=2.13.0", "rich", "docker>=7.1.0", "textual>=6.0.0", @@ -98,6 +99,7 @@ module = [ "cvss.*", "docker.*", "caido_sdk_client.*", + "pydantic_settings.*", ] ignore_missing_imports = true disable_error_code = ["import-untyped"] diff --git a/strix/config/__init__.py b/strix/config/__init__.py index 328c138..fda6860 100644 --- a/strix/config/__init__.py +++ b/strix/config/__init__.py @@ -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", ] diff --git a/strix/config/config.py b/strix/config/config.py deleted file mode 100644 index 92c19f4..0000000 --- a/strix/config/config.py +++ /dev/null @@ -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 diff --git a/strix/config/loader.py b/strix/config/loader.py new file mode 100644 index 0000000..5349e64 --- /dev/null +++ b/strix/config/loader.py @@ -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 diff --git a/strix/config/settings.py b/strix/config/settings.py new file mode 100644 index 0000000..6f59010 --- /dev/null +++ b/strix/config/settings.py @@ -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 ``) — 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) diff --git a/strix/entry.py b/strix/entry.py index 26e40c3..e669420 100644 --- a/strix/entry.py +++ b/strix/entry.py @@ -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"], diff --git a/strix/interface/cli.py b/strix/interface/cli.py index 3c6418c..e6d5906 100644 --- a/strix/interface/cli.py +++ b/strix/interface/cli.py @@ -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) diff --git a/strix/interface/main.py b/strix/interface/main.py index 19e7362..cc562bf 100644 --- a/strix/interface/main.py +++ b/strix/interface/main.py @@ -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), ) diff --git a/strix/interface/tui.py b/strix/interface/tui.py index b3c0e73..d0d22e5 100644 --- a/strix/interface/tui.py +++ b/strix/interface/tui.py @@ -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( diff --git a/strix/interface/utils.py b/strix/interface/utils.py index c209d36..1acd4a7 100644 --- a/strix/interface/utils.py +++ b/strix/interface/utils.py @@ -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]] = [] diff --git a/strix/llm/dedupe.py b/strix/llm/dedupe.py index 291d20a..d720f01 100644 --- a/strix/llm/dedupe.py +++ b/strix/llm/dedupe.py @@ -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, diff --git a/strix/runtime/session_manager.py b/strix/runtime/session_manager.py index c4c48de..e784bc7 100644 --- a/strix/runtime/session_manager.py +++ b/strix/runtime/session_manager.py @@ -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( diff --git a/strix/telemetry/posthog.py b/strix/telemetry/posthog.py index 6fbbf30..1d6ddce 100644 --- a/strix/telemetry/posthog.py +++ b/strix/telemetry/posthog.py @@ -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: diff --git a/strix/tools/web_search/tool.py b/strix/tools/web_search/tool.py index 4c72ef1..8855957 100644 --- a/strix/tools/web_search/tool.py +++ b/strix/tools/web_search/tool.py @@ -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, diff --git a/uv.lock b/uv.lock index da4442a..d457658 100644 --- a/uv.lock +++ b/uv.lock @@ -2043,6 +2043,7 @@ dependencies = [ { name = "docker" }, { name = "openai-agents", extra = ["litellm"] }, { name = "pydantic" }, + { name = "pydantic-settings" }, { name = "requests" }, { name = "rich" }, { name = "textual" }, @@ -2065,6 +2066,7 @@ requires-dist = [ { name = "docker", specifier = ">=7.1.0" }, { name = "openai-agents", extras = ["litellm"], specifier = "==0.14.6" }, { name = "pydantic", specifier = ">=2.11.3" }, + { name = "pydantic-settings", specifier = ">=2.13.0" }, { name = "requests", specifier = ">=2.32.0" }, { name = "rich" }, { name = "textual", specifier = ">=6.0.0" },