mirror of
https://github.com/usestrix/strix.git
synced 2026-09-06 08:15:56 +00:00
b948c14472
Adds an explicit ``--resume RUN_NAME`` flag that loads the prior
run's persisted scan state from ``strix_runs/<run_name>/scan_state.json``
and replays it (targets, scan_mode, instruction, local_sources,
diff_scope, scope_mode, diff_base) so the user never has to retype
their original args.
The exit panel now suggests ``strix --resume <run_name>`` instead of
``--run-name``. Same single-line, same dim-label / coloured-value
styling as ``Target`` / ``Output`` rows, gated on
``not scan_completed``.
CLI contract:
* ``--resume X`` cannot be combined with ``--target`` (parser error).
* ``--resume X`` errors with a clear message if
``strix_runs/X/scan_state.json`` is missing.
* Fresh runs persist scan_state.json once at the end of setup —
after target normalization, repo cloning, local-source
collection, diff-scope resolution, and final instruction
composition. So whatever the agent saw on first run is exactly
what the resumed run sees.
Internally the resume path stays implicit (presence of bus.json
triggers it inside ``run_strix_scan``); ``--resume`` is a UX layer
that:
1. Sets ``args.run_name = args.resume``.
2. Pre-populates ``args.targets_info`` and friends from disk.
3. Skips the fresh-only steps (target re-parse, repo clone,
diff-scope re-resolution) — the persisted values were already
finalized on the first run.
HARNESS_WIKI.md: drop the "delete the run dir to force fresh"
instruction.
737 lines
25 KiB
Python
737 lines
25 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Strix Agent Interface
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import json
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import litellm
|
|
from docker.errors import DockerException
|
|
from rich.console import Console
|
|
from rich.panel import Panel
|
|
from rich.text import Text
|
|
|
|
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,
|
|
clone_repository,
|
|
collect_local_sources,
|
|
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"
|
|
|
|
|
|
import logging # noqa: E402
|
|
|
|
|
|
# Per-scan logging is set up by ``setup_scan_logging`` from inside
|
|
# ``orchestration.scan.run_strix_scan`` once the scan ``run_dir`` is
|
|
# known — that's where ``strix.*`` levels and handlers are owned. Pre-scan
|
|
# work (``main()``, env validation, image pull) emits via the module
|
|
# logger; once setup_scan_logging runs, those records start landing in
|
|
# the file too.
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def validate_environment() -> None:
|
|
logger.info("Validating environment")
|
|
console = Console()
|
|
missing_required_vars = []
|
|
missing_optional_vars = []
|
|
|
|
settings = load_settings()
|
|
|
|
if not settings.llm.model:
|
|
missing_required_vars.append("STRIX_LLM")
|
|
|
|
if not settings.llm.api_key:
|
|
missing_optional_vars.append("LLM_API_KEY")
|
|
|
|
if not settings.llm.api_base:
|
|
missing_optional_vars.append("LLM_API_BASE")
|
|
|
|
if not settings.integrations.perplexity_api_key:
|
|
missing_optional_vars.append("PERPLEXITY_API_KEY")
|
|
|
|
if missing_required_vars:
|
|
error_text = Text()
|
|
error_text.append("MISSING REQUIRED ENVIRONMENT VARIABLES", style="bold red")
|
|
error_text.append("\n\n", style="white")
|
|
|
|
for var in missing_required_vars:
|
|
error_text.append(f"• {var}", style="bold yellow")
|
|
error_text.append(" is not set\n", style="white")
|
|
|
|
if missing_optional_vars:
|
|
error_text.append("\nOptional environment variables:\n", style="dim white")
|
|
for var in missing_optional_vars:
|
|
error_text.append(f"• {var}", style="dim yellow")
|
|
error_text.append(" is not set\n", style="dim white")
|
|
|
|
error_text.append("\nRequired environment variables:\n", style="white")
|
|
for var in missing_required_vars:
|
|
if var == "STRIX_LLM":
|
|
error_text.append("• ", style="white")
|
|
error_text.append("STRIX_LLM", style="bold cyan")
|
|
error_text.append(
|
|
" - Model name to use with litellm (e.g., 'openai/gpt-5.4')\n",
|
|
style="white",
|
|
)
|
|
|
|
if missing_optional_vars:
|
|
error_text.append("\nOptional environment variables:\n", style="white")
|
|
for var in missing_optional_vars:
|
|
if var == "LLM_API_KEY":
|
|
error_text.append("• ", style="white")
|
|
error_text.append("LLM_API_KEY", style="bold cyan")
|
|
error_text.append(
|
|
" - API key for the LLM provider "
|
|
"(not needed for local models, Vertex AI, AWS, etc.)\n",
|
|
style="white",
|
|
)
|
|
elif var == "LLM_API_BASE":
|
|
error_text.append("• ", style="white")
|
|
error_text.append("LLM_API_BASE", style="bold cyan")
|
|
error_text.append(
|
|
" - Custom API base URL if using local models (e.g., Ollama, LMStudio)\n",
|
|
style="white",
|
|
)
|
|
elif var == "PERPLEXITY_API_KEY":
|
|
error_text.append("• ", style="white")
|
|
error_text.append("PERPLEXITY_API_KEY", style="bold cyan")
|
|
error_text.append(
|
|
" - API key for Perplexity AI web search (enables real-time research)\n",
|
|
style="white",
|
|
)
|
|
elif var == "STRIX_REASONING_EFFORT":
|
|
error_text.append("• ", style="white")
|
|
error_text.append("STRIX_REASONING_EFFORT", style="bold cyan")
|
|
error_text.append(
|
|
" - Reasoning effort level: none, minimal, low, medium, high, xhigh "
|
|
"(default: high)\n",
|
|
style="white",
|
|
)
|
|
|
|
error_text.append("\nExample setup:\n", style="white")
|
|
error_text.append("export STRIX_LLM='openai/gpt-5.4'\n", style="dim white")
|
|
|
|
if missing_optional_vars:
|
|
for var in missing_optional_vars:
|
|
if var == "LLM_API_KEY":
|
|
error_text.append(
|
|
"export LLM_API_KEY='your-api-key-here' "
|
|
"# not needed for local models, Vertex AI, AWS, etc.\n",
|
|
style="dim white",
|
|
)
|
|
elif var == "LLM_API_BASE":
|
|
error_text.append(
|
|
"export LLM_API_BASE='http://localhost:11434' "
|
|
"# needed for local models only\n",
|
|
style="dim white",
|
|
)
|
|
elif var == "PERPLEXITY_API_KEY":
|
|
error_text.append(
|
|
"export PERPLEXITY_API_KEY='your-perplexity-key-here'\n", style="dim white"
|
|
)
|
|
elif var == "STRIX_REASONING_EFFORT":
|
|
error_text.append(
|
|
"export STRIX_REASONING_EFFORT='high'\n",
|
|
style="dim white",
|
|
)
|
|
|
|
panel = Panel(
|
|
error_text,
|
|
title="[bold white]STRIX",
|
|
title_align="left",
|
|
border_style="red",
|
|
padding=(1, 2),
|
|
)
|
|
|
|
logger.error("Missing required env vars: %s", missing_required_vars)
|
|
console.print("\n")
|
|
console.print(panel)
|
|
console.print()
|
|
sys.exit(1)
|
|
logger.info(
|
|
"Environment OK (optional missing: %s)",
|
|
missing_optional_vars or "none",
|
|
)
|
|
|
|
|
|
def check_docker_installed() -> None:
|
|
if shutil.which("docker") is None:
|
|
logger.error("Docker CLI not found in PATH")
|
|
console = Console()
|
|
error_text = Text()
|
|
error_text.append("DOCKER NOT INSTALLED", style="bold red")
|
|
error_text.append("\n\n", style="white")
|
|
error_text.append("The 'docker' CLI was not found in your PATH.\n", style="white")
|
|
error_text.append(
|
|
"Please install Docker and ensure the 'docker' command is available.\n\n", style="white"
|
|
)
|
|
|
|
panel = Panel(
|
|
error_text,
|
|
title="[bold white]STRIX",
|
|
title_align="left",
|
|
border_style="red",
|
|
padding=(1, 2),
|
|
)
|
|
console.print("\n", panel, "\n")
|
|
sys.exit(1)
|
|
logger.debug("Docker CLI present")
|
|
|
|
|
|
async def warm_up_llm() -> None:
|
|
console = Console()
|
|
logger.info("Warming up LLM connection")
|
|
|
|
try:
|
|
llm = load_settings().llm
|
|
|
|
test_messages = [
|
|
{"role": "system", "content": "You are a helpful assistant."},
|
|
{"role": "user", "content": "Reply with just 'OK'."},
|
|
]
|
|
|
|
completion_kwargs: dict[str, Any] = {
|
|
"model": llm.model,
|
|
"messages": test_messages,
|
|
"timeout": llm.timeout,
|
|
}
|
|
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)
|
|
|
|
validate_llm_response(response)
|
|
logger.info("LLM warm-up succeeded for model %s", llm.model)
|
|
|
|
except Exception as e:
|
|
logger.exception("LLM warm-up failed")
|
|
error_text = Text()
|
|
error_text.append("LLM CONNECTION FAILED", style="bold red")
|
|
error_text.append("\n\n", style="white")
|
|
error_text.append("Could not establish connection to the language model.\n", style="white")
|
|
error_text.append("Please check your configuration and try again.\n", style="white")
|
|
error_text.append(f"\nError: {e}", style="dim white")
|
|
|
|
panel = Panel(
|
|
error_text,
|
|
title="[bold white]STRIX",
|
|
title_align="left",
|
|
border_style="red",
|
|
padding=(1, 2),
|
|
)
|
|
|
|
console.print("\n")
|
|
console.print(panel)
|
|
console.print()
|
|
sys.exit(1)
|
|
|
|
|
|
def get_version() -> str:
|
|
try:
|
|
from importlib.metadata import version
|
|
|
|
return version("strix-agent")
|
|
except Exception:
|
|
return "unknown"
|
|
|
|
|
|
def parse_arguments() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(
|
|
description="Strix Multi-Agent Cybersecurity Penetration Testing Tool",
|
|
formatter_class=argparse.RawDescriptionHelpFormatter,
|
|
epilog="""
|
|
Examples:
|
|
# Web application penetration test
|
|
strix --target https://example.com
|
|
|
|
# GitHub repository analysis
|
|
strix --target https://github.com/user/repo
|
|
strix --target git@github.com:user/repo.git
|
|
|
|
# Local code analysis
|
|
strix --target ./my-project
|
|
|
|
# Domain penetration test
|
|
strix --target example.com
|
|
|
|
# IP address penetration test
|
|
strix --target 192.168.1.42
|
|
|
|
# Multiple targets (e.g., white-box testing with source and deployed app)
|
|
strix --target https://github.com/user/repo --target https://example.com
|
|
strix --target ./my-project --target https://staging.example.com --target https://prod.example.com
|
|
|
|
# Custom instructions (inline)
|
|
strix --target example.com --instruction "Focus on authentication vulnerabilities"
|
|
|
|
# Custom instructions (from file)
|
|
strix --target example.com --instruction-file ./instructions.txt
|
|
strix --target https://app.com --instruction-file /path/to/detailed_instructions.md
|
|
""",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-v",
|
|
"--version",
|
|
action="version",
|
|
version=f"strix {get_version()}",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-t",
|
|
"--target",
|
|
type=str,
|
|
action="append",
|
|
help="Target to test (URL, repository, local directory path, domain name, or IP address). "
|
|
"Can be specified multiple times for multi-target scans. "
|
|
"Required for fresh runs; loaded from disk when ``--resume`` is set.",
|
|
)
|
|
parser.add_argument(
|
|
"--instruction",
|
|
type=str,
|
|
help="Custom instructions for the penetration test. This can be "
|
|
"specific vulnerability types to focus on (e.g., 'Focus on IDOR and XSS'), "
|
|
"testing approaches (e.g., 'Perform thorough authentication testing'), "
|
|
"test credentials (e.g., 'Use the following credentials to access the app: "
|
|
"admin:password123'), "
|
|
"or areas of interest (e.g., 'Check login API endpoint for security issues').",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--instruction-file",
|
|
type=str,
|
|
help="Path to a file containing detailed custom instructions for the penetration test. "
|
|
"Use this option when you have lengthy or complex instructions saved in a file "
|
|
"(e.g., '--instruction-file ./detailed_instructions.txt').",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-n",
|
|
"--non-interactive",
|
|
action="store_true",
|
|
help=(
|
|
"Run in non-interactive mode (no TUI, exits on completion). "
|
|
"Default is interactive mode with TUI."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-m",
|
|
"--scan-mode",
|
|
type=str,
|
|
choices=["quick", "standard", "deep"],
|
|
default="deep",
|
|
help=(
|
|
"Scan mode: "
|
|
"'quick' for fast CI/CD checks, "
|
|
"'standard' for routine testing, "
|
|
"'deep' for thorough security reviews (default). "
|
|
"Default: deep."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--scope-mode",
|
|
type=str,
|
|
choices=["auto", "diff", "full"],
|
|
default="auto",
|
|
help=(
|
|
"Scope mode for code targets: "
|
|
"'auto' enables PR diff-scope in CI/headless runs, "
|
|
"'diff' forces changed-files scope, "
|
|
"'full' disables diff-scope."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--diff-base",
|
|
type=str,
|
|
help=(
|
|
"Target branch or commit to compare against (e.g., origin/main). "
|
|
"Defaults to the repository's default branch."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--config",
|
|
type=str,
|
|
help="Path to a custom config file (JSON) to use instead of ~/.strix/cli-config.json",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--resume",
|
|
type=str,
|
|
metavar="RUN_NAME",
|
|
help=(
|
|
"Resume a prior scan by its run name (the dir under ./strix_runs/). "
|
|
"Picks up the root + every non-terminal subagent's full LLM history "
|
|
"and bus topology. Skips fresh run-name generation."
|
|
),
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.instruction and args.instruction_file:
|
|
parser.error(
|
|
"Cannot specify both --instruction and --instruction-file. Use one or the other."
|
|
)
|
|
|
|
if args.instruction_file:
|
|
instruction_path = Path(args.instruction_file)
|
|
try:
|
|
with instruction_path.open(encoding="utf-8") as f:
|
|
args.instruction = f.read().strip()
|
|
if not args.instruction:
|
|
parser.error(f"Instruction file '{instruction_path}' is empty")
|
|
except Exception as e:
|
|
parser.error(f"Failed to read instruction file '{instruction_path}': {e}")
|
|
|
|
if args.resume:
|
|
if args.target:
|
|
parser.error(
|
|
"Cannot combine --resume with --target. --resume picks up where "
|
|
"the prior run left off, including the original target list."
|
|
)
|
|
_load_resume_state(args, parser)
|
|
else:
|
|
if not args.target:
|
|
parser.error(
|
|
"the following arguments are required: -t/--target "
|
|
"(or use --resume <run_name> to continue a prior scan)"
|
|
)
|
|
args.targets_info = []
|
|
for target in args.target:
|
|
try:
|
|
target_type, target_dict = infer_target_type(target)
|
|
|
|
if target_type == "local_code":
|
|
display_target = target_dict.get("target_path", target)
|
|
else:
|
|
display_target = target
|
|
|
|
args.targets_info.append(
|
|
{"type": target_type, "details": target_dict, "original": display_target}
|
|
)
|
|
except ValueError:
|
|
parser.error(f"Invalid target '{target}'")
|
|
|
|
assign_workspace_subdirs(args.targets_info)
|
|
rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
|
|
|
|
return args
|
|
|
|
|
|
def _persist_scan_state(args: argparse.Namespace) -> None:
|
|
"""Dump the resolved scan inputs to ``{run_dir}/scan_state.json``.
|
|
|
|
Called once at the end of fresh-run setup. ``--resume <run_name>`` on
|
|
a future invocation reads this file to repopulate targets, scan_mode,
|
|
instruction, local_sources, and diff_scope without the user having to
|
|
retype them.
|
|
"""
|
|
run_dir = Path("strix_runs") / args.run_name
|
|
run_dir.mkdir(parents=True, exist_ok=True)
|
|
state = {
|
|
"targets_info": args.targets_info,
|
|
"scan_mode": args.scan_mode,
|
|
"instruction": args.instruction,
|
|
"non_interactive": args.non_interactive,
|
|
"local_sources": getattr(args, "local_sources", []),
|
|
"diff_scope": getattr(args, "diff_scope", {"active": False}),
|
|
"scope_mode": args.scope_mode,
|
|
"diff_base": args.diff_base,
|
|
}
|
|
(run_dir / "scan_state.json").write_text(
|
|
json.dumps(state, ensure_ascii=False, indent=2, default=str),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
|
|
def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
|
|
"""Populate ``args.targets_info`` and friends from a prior run's scan state.
|
|
|
|
Reads ``strix_runs/<run_name>/scan_state.json`` written at the end of the
|
|
fresh-run setup in ``main()``. Only fields the user did not explicitly
|
|
set on this invocation are restored — passing ``--instruction`` on
|
|
resume, for example, overrides the persisted instruction.
|
|
"""
|
|
state_path = Path("strix_runs") / args.resume / "scan_state.json"
|
|
if not state_path.exists():
|
|
parser.error(
|
|
f"--resume {args.resume}: no such run "
|
|
f"(missing {state_path}; remove --resume for a fresh start)"
|
|
)
|
|
try:
|
|
state = json.loads(state_path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
parser.error(f"--resume {args.resume}: scan_state.json unreadable: {exc}")
|
|
|
|
args.targets_info = state.get("targets_info") or []
|
|
if not args.targets_info:
|
|
parser.error(f"--resume {args.resume}: scan_state.json has no targets_info")
|
|
|
|
if args.instruction is None:
|
|
args.instruction = state.get("instruction")
|
|
if state.get("instruction_file") and args.instruction_file is None:
|
|
args.instruction_file = state.get("instruction_file")
|
|
if state.get("local_sources"):
|
|
args.local_sources = state.get("local_sources")
|
|
if state.get("diff_scope"):
|
|
args.diff_scope = state.get("diff_scope")
|
|
persisted_scan_mode = state.get("scan_mode")
|
|
if persisted_scan_mode and args.scan_mode == "deep":
|
|
# Default scan_mode is "deep"; only override from disk if the user
|
|
# didn't explicitly pass a different one. (Best-effort: argparse
|
|
# can't tell "user passed 'deep'" from "default 'deep'"; if the
|
|
# persisted run was "quick" and user re-runs with an explicit
|
|
# ``-m deep``, we'll honor the persisted mode. Acceptable.)
|
|
args.scan_mode = persisted_scan_mode
|
|
|
|
|
|
def display_completion_message(args: argparse.Namespace, results_path: Path) -> None:
|
|
console = Console()
|
|
tracer = get_global_tracer()
|
|
|
|
scan_completed = False
|
|
if tracer and tracer.scan_results:
|
|
scan_completed = tracer.scan_results.get("scan_completed", False)
|
|
|
|
completion_text = Text()
|
|
if scan_completed:
|
|
completion_text.append("Penetration test completed", style="bold #22c55e")
|
|
else:
|
|
completion_text.append("SESSION ENDED", style="bold #eab308")
|
|
|
|
target_text = Text()
|
|
target_text.append("Target", style="dim")
|
|
target_text.append(" ")
|
|
if len(args.targets_info) == 1:
|
|
target_text.append(args.targets_info[0]["original"], style="bold white")
|
|
else:
|
|
target_text.append(f"{len(args.targets_info)} targets", style="bold white")
|
|
for target_info in args.targets_info:
|
|
target_text.append("\n ")
|
|
target_text.append(target_info["original"], style="white")
|
|
|
|
stats_text = build_final_stats_text(tracer)
|
|
|
|
panel_parts: list[Text | str] = [completion_text, "\n\n", target_text]
|
|
|
|
if stats_text.plain:
|
|
panel_parts.extend(["\n", stats_text])
|
|
|
|
results_text = Text()
|
|
results_text.append("\n")
|
|
results_text.append("Output", style="dim")
|
|
results_text.append(" ")
|
|
results_text.append(str(results_path), style="#60a5fa")
|
|
panel_parts.extend(["\n", results_text])
|
|
|
|
if not scan_completed:
|
|
resume_text = Text()
|
|
resume_text.append("\n")
|
|
resume_text.append("Resume", style="dim")
|
|
resume_text.append(" ")
|
|
resume_text.append(f"strix --resume {args.run_name}", style="#22c55e")
|
|
panel_parts.extend(["\n", resume_text])
|
|
|
|
panel_content = Text.assemble(*panel_parts)
|
|
|
|
border_style = "#22c55e" if scan_completed else "#eab308"
|
|
|
|
panel = Panel(
|
|
panel_content,
|
|
title="[bold white]STRIX",
|
|
title_align="left",
|
|
border_style=border_style,
|
|
padding=(1, 2),
|
|
)
|
|
|
|
console.print("\n")
|
|
console.print(panel)
|
|
console.print()
|
|
console.print("[#60a5fa]strix.ai[/] [dim]·[/] [#60a5fa]discord.gg/strix-ai[/]")
|
|
console.print()
|
|
|
|
|
|
def pull_docker_image() -> None:
|
|
console = Console()
|
|
client = check_docker_connection()
|
|
|
|
image = load_settings().runtime.image
|
|
|
|
if image_exists(client, image):
|
|
logger.debug("Docker image already present locally: %s", image)
|
|
return
|
|
|
|
logger.info("Pulling docker image: %s", image)
|
|
console.print()
|
|
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()
|
|
|
|
with console.status("[bold cyan]Downloading image layers...", spinner="dots") as status:
|
|
try:
|
|
layers_info: dict[str, str] = {}
|
|
last_update = ""
|
|
|
|
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:
|
|
logger.exception("Failed to pull docker image %s", image)
|
|
console.print()
|
|
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: {image}\n", style="white")
|
|
error_text.append(str(e), style="dim red")
|
|
|
|
panel = Panel(
|
|
error_text,
|
|
title="[bold white]STRIX",
|
|
title_align="left",
|
|
border_style="red",
|
|
padding=(1, 2),
|
|
)
|
|
console.print(panel, "\n")
|
|
sys.exit(1)
|
|
|
|
logger.info("Docker image %s ready", image)
|
|
success_text = Text()
|
|
success_text.append("Docker image ready", style="#22c55e")
|
|
console.print(success_text)
|
|
console.print()
|
|
|
|
|
|
def main() -> None:
|
|
if sys.platform == "win32":
|
|
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
|
|
|
args = parse_arguments()
|
|
|
|
if args.config:
|
|
apply_config_override(validate_config_file(args.config))
|
|
|
|
check_docker_installed()
|
|
pull_docker_image()
|
|
|
|
validate_environment()
|
|
asyncio.run(warm_up_llm())
|
|
|
|
persist_current()
|
|
|
|
args.run_name = args.resume or generate_run_name(args.targets_info)
|
|
|
|
if not args.resume:
|
|
for target_info in args.targets_info:
|
|
if target_info["type"] == "repository":
|
|
repo_url = target_info["details"]["target_repo"]
|
|
dest_name = target_info["details"].get("workspace_subdir")
|
|
cloned_path = clone_repository(repo_url, args.run_name, dest_name)
|
|
target_info["details"]["cloned_repo_path"] = cloned_path
|
|
|
|
args.local_sources = collect_local_sources(args.targets_info)
|
|
try:
|
|
diff_scope = resolve_diff_scope_context(
|
|
local_sources=args.local_sources,
|
|
scope_mode=args.scope_mode,
|
|
diff_base=args.diff_base,
|
|
non_interactive=args.non_interactive,
|
|
)
|
|
except ValueError as e:
|
|
console = Console()
|
|
error_text = Text()
|
|
error_text.append("DIFF SCOPE RESOLUTION FAILED", style="bold red")
|
|
error_text.append("\n\n", style="white")
|
|
error_text.append(str(e), style="white")
|
|
|
|
panel = Panel(
|
|
error_text,
|
|
title="[bold white]STRIX",
|
|
title_align="left",
|
|
border_style="red",
|
|
padding=(1, 2),
|
|
)
|
|
console.print("\n")
|
|
console.print(panel)
|
|
console.print()
|
|
sys.exit(1)
|
|
|
|
args.diff_scope = diff_scope.metadata
|
|
if diff_scope.instruction_block:
|
|
if args.instruction:
|
|
args.instruction = f"{diff_scope.instruction_block}\n\n{args.instruction}"
|
|
else:
|
|
args.instruction = diff_scope.instruction_block
|
|
|
|
# Persist the fully-resolved scan state so a future
|
|
# ``--resume <run_name>`` invocation can pick up without
|
|
# re-supplying targets / instructions / scope.
|
|
_persist_scan_state(args)
|
|
|
|
posthog.start(
|
|
model=load_settings().llm.model,
|
|
scan_mode=args.scan_mode,
|
|
is_whitebox=is_whitebox_scan(args.targets_info),
|
|
interactive=not args.non_interactive,
|
|
has_instructions=bool(args.instruction),
|
|
)
|
|
|
|
exit_reason = "user_exit"
|
|
try:
|
|
if args.non_interactive:
|
|
asyncio.run(run_cli(args))
|
|
else:
|
|
asyncio.run(run_tui(args))
|
|
except KeyboardInterrupt:
|
|
exit_reason = "interrupted"
|
|
except Exception as e:
|
|
exit_reason = "error"
|
|
posthog.error("unhandled_exception", str(e))
|
|
raise
|
|
finally:
|
|
tracer = get_global_tracer()
|
|
if tracer:
|
|
posthog.end(tracer, exit_reason=exit_reason)
|
|
|
|
results_path = Path("strix_runs") / args.run_name
|
|
display_completion_message(args, results_path)
|
|
|
|
if args.non_interactive:
|
|
tracer = get_global_tracer()
|
|
if tracer and tracer.vulnerability_reports:
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|