fix(cli): expose verification and settle imports

This commit is contained in:
oyasumi 2026-08-27 19:12:48 +00:00
parent 7c7729b3bd
commit 4f9af5761c
6 changed files with 129 additions and 5 deletions

View file

@ -117,6 +117,14 @@ strix (--target <target> | --target-list <path>) [options]
Must be greater than `0`.
</ParamField>
<ParamField path="--verify-findings" type="boolean" default="false">
Run an independent sandbox-backed verification loop before each candidate is
persisted. The verifier creates and executes its own PoC. If it cannot
reproduce the claimed impact, it reviews the complete CVSS vector using the
evidence that remains. This enables verification for the current invocation;
use `STRIX_VERIFY_FINDINGS=true` to make it the configured default.
</ParamField>
## Examples
```bash
@ -132,6 +140,9 @@ strix --target api.example.com --instruction "Focus on IDOR and auth bypass"
# CI/CD mode
strix -n --target ./ --scan-mode quick
# Independently reproduce and rescore candidate findings
strix --target https://example.com --verify-findings
# Cap cost and per-agent turns
strix --target https://example.com --max-budget 25 --max-turns 300

View file

@ -7,7 +7,7 @@ import os
import sys
from pathlib import Path
from strix.config import apply_config_override
from strix.config import apply_config_override, load_settings
from strix.config.settings import DEFAULT_MAX_TURNS
from strix.core.paths import run_dir_for, runtime_state_dir
from strix.interface.scan_setup import attach_workspace_mount, build_targets_info
@ -269,6 +269,15 @@ Examples:
),
)
parser.add_argument(
"--verify-findings",
action="store_true",
help=(
"Independently reproduce each candidate finding in the scan sandbox before "
"reporting it. Unreproduced findings receive an evidence-based CVSS review."
),
)
parser.add_argument(
"--resume",
type=str,
@ -292,6 +301,11 @@ Examples:
if args.config:
apply_config_override(validate_config_file(args.config))
if args.verify_findings:
# Process-local CLI override: unlike STRIX_VERIFY_FINDINGS in the config,
# this must not become sticky when startup persists environment settings.
load_settings().verification.enabled = True
if args.mcp_config:
mcp_config_path = Path(args.mcp_config).expanduser()
if not mcp_config_path.is_file():

View file

@ -437,7 +437,7 @@ def main() -> None:
sys.exit(run_auth(sys.argv[2:]))
from strix.llm.warmup import start_import_warmup
from strix.llm.warmup import start_import_warmup, wait_for_import_warmup
start_import_warmup()
@ -452,6 +452,10 @@ def main() -> None:
check_docker_installed()
pull_docker_image()
# Importing the SDK graph concurrently with the report package can expose
# partially initialized ``agents`` modules. Settle the warm-up first.
wait_for_import_warmup()
# In setup mode the TUI collects the target, then runs prepare_run(),
# warm-up, and telemetry itself once the user starts the scan.
if not args.needs_setup:

View file

@ -5,9 +5,10 @@ Caido SDK, the Docker SDK) costs seconds to import cold, but none of it is
needed until a scan actually starts. Importing it on a daemon thread at CLI
entry overlaps that cost with the I/O-bound startup work that always precedes
a scan (argument parsing, Docker checks, image pull, TUI setup), so by the
time the scan begins the modules are already in ``sys.modules``. Any thread
that needs one of them before the warm-up finishes just blocks on the normal
import lock, so behaviour is unchanged either way.
time the scan begins the modules are already in ``sys.modules``. Startup joins
the thread before importing scan dependencies on the main thread: Python locks
individual modules, not an entire package import graph, so that boundary keeps
concurrent imports from observing partially initialized packages.
"""
from __future__ import annotations
@ -53,3 +54,11 @@ def start_import_warmup(modules: tuple[str, ...] = WARMUP_MODULES) -> threading.
)
_thread.start()
return _thread
def wait_for_import_warmup() -> None:
"""Wait until the background import graph has settled, if it was started."""
with _lock:
thread = _thread
if thread is not None and thread is not threading.current_thread():
thread.join()

View file

@ -0,0 +1,48 @@
from __future__ import annotations
import os
import sys
from typing import TYPE_CHECKING
import pytest
from strix.config import loader
from strix.interface.cli_args import parse_arguments
if TYPE_CHECKING:
from pathlib import Path
@pytest.fixture(autouse=True)
def _isolated_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(loader, "_cached", None)
monkeypatch.setattr(loader, "_override", tmp_path / "missing-config.json")
monkeypatch.delenv("STRIX_VERIFY_FINDINGS", raising=False)
def test_verify_findings_flag_enables_process_setting(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
sys,
"argv",
["strix", "--target", "https://example.com", "--verify-findings"],
)
args = parse_arguments()
assert args.verify_findings is True
assert loader.load_settings().verification.enabled is True
assert "STRIX_VERIFY_FINDINGS" not in os.environ
def test_verify_findings_appears_in_help(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
monkeypatch.setattr(sys, "argv", ["strix", "--help"])
with pytest.raises(SystemExit) as exc_info:
parse_arguments()
assert exc_info.value.code == 0
assert "--verify-findings" in capsys.readouterr().out

View file

@ -0,0 +1,38 @@
from __future__ import annotations
import threading
from typing import TYPE_CHECKING
from strix.llm import warmup
if TYPE_CHECKING:
import pytest
def test_wait_for_import_warmup_joins_active_thread(
monkeypatch: pytest.MonkeyPatch,
) -> None:
entered = threading.Event()
release = threading.Event()
returned = threading.Event()
def blocked_warm(_modules: tuple[str, ...]) -> None:
entered.set()
assert release.wait(timeout=2)
monkeypatch.setattr(warmup, "_thread", None)
monkeypatch.setattr(warmup, "_warm", blocked_warm)
thread = warmup.start_import_warmup(())
assert entered.wait(timeout=1)
waiter = threading.Thread(
target=lambda: (warmup.wait_for_import_warmup(), returned.set()),
)
waiter.start()
assert not returned.wait(timeout=0.05)
release.set()
waiter.join(timeout=1)
thread.join(timeout=1)
assert returned.is_set()