mirror of
https://github.com/usestrix/strix.git
synced 2026-08-28 05:25:00 +00:00
770 lines
27 KiB
Python
770 lines
27 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.scan_store import get_global_scan_store
|
|
|
|
|
|
HOST_GATEWAY_HOSTNAME = "host.docker.internal"
|
|
|
|
|
|
import logging # noqa: E402
|
|
|
|
|
|
# Per-scan logging is set up by ``setup_scan_logging`` from inside
|
|
# ``orchestration.runner.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 agent 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}")
|
|
|
|
# Capture before ``_load_resume_state`` overrides — used by the resume
|
|
# path in ``run_strix_scan`` to decide whether to inject the new
|
|
# instruction into the root's SDK session after replay.
|
|
args.user_explicit_instruction = args.instruction if args.resume else None
|
|
|
|
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)
|
|
agents_path = Path("strix_runs") / args.resume / "agents.json"
|
|
if not agents_path.exists():
|
|
parser.error(
|
|
f"--resume {args.resume}: missing {agents_path}. The run was "
|
|
f"persisted but never reached its first agent snapshot — "
|
|
f"there's nothing to resume from. Pick a fresh --run-name "
|
|
f"or remove --resume to start over with the same targets."
|
|
)
|
|
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")
|
|
|
|
# Validate any persisted ``cloned_repo_path`` still exists on disk.
|
|
# The resume path skips re-cloning, so a missing dir would mean the
|
|
# container mounts an empty source tree and agents silently scan
|
|
# nothing.
|
|
for target in args.targets_info:
|
|
if not isinstance(target, dict):
|
|
continue
|
|
details = target.get("details") or {}
|
|
if target.get("type") != "repository":
|
|
continue
|
|
cloned = details.get("cloned_repo_path")
|
|
if not cloned:
|
|
continue
|
|
if not Path(cloned).expanduser().exists():
|
|
parser.error(
|
|
f"--resume {args.resume}: cloned repo at {cloned} is missing. "
|
|
f"It was deleted between runs. Pick a fresh --run-name to "
|
|
f"re-clone, or restore the directory before resuming."
|
|
)
|
|
|
|
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()
|
|
scan_store = get_global_scan_store()
|
|
|
|
scan_completed = False
|
|
if scan_store and scan_store.scan_results:
|
|
scan_completed = scan_store.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(scan_store)
|
|
|
|
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:
|
|
scan_store = get_global_scan_store()
|
|
if scan_store:
|
|
posthog.end(scan_store, exit_reason=exit_reason)
|
|
|
|
results_path = Path("strix_runs") / args.run_name
|
|
display_completion_message(args, results_path)
|
|
|
|
if args.non_interactive:
|
|
scan_store = get_global_scan_store()
|
|
if scan_store and scan_store.vulnerability_reports:
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|