first commit

This commit is contained in:
Zakaria
2026-07-02 12:31:49 -04:00
commit 1c7784cae9
189 changed files with 32427 additions and 0 deletions
+4
View File
@@ -0,0 +1,4 @@
from .main import main
__all__ = ["main"]
+687
View File
@@ -0,0 +1,687 @@
Screen {
background: #000000;
color: #d4d4d4;
}
.screen--selection {
background: #2d3d2f;
color: #e5e5e5;
}
ToastRack {
dock: top;
align: right top;
margin-bottom: 0;
margin-top: 1;
}
Toast {
width: 25;
background: #000000;
border-left: outer #22c55e;
}
Toast.-information .toast--title {
color: #22c55e;
}
#splash_screen {
height: 100%;
width: 100%;
background: #000000;
color: #22c55e;
align: center middle;
content-align: center middle;
text-align: center;
}
#splash_content {
width: auto;
height: auto;
background: transparent;
text-align: center;
content-align: center middle;
padding: 2;
}
#main_container {
height: 100%;
padding: 0;
margin: 0;
background: #000000;
}
#content_container {
height: 1fr;
padding: 0;
background: transparent;
}
#sidebar {
width: 20%;
background: transparent;
margin-left: 1;
}
#sidebar.-hidden {
display: none;
}
#agents_tree {
height: 1fr;
background: transparent;
border: round #333333;
border-title-color: #a8a29e;
border-title-style: bold;
padding: 1;
margin-bottom: 0;
}
#stats_scroll {
height: auto;
max-height: 15;
background: transparent;
padding: 0;
margin: 0;
border: round #333333;
scrollbar-size: 0 0;
}
#stats_display {
height: auto;
background: transparent;
padding: 0 1;
margin: 0;
}
#vulnerabilities_panel {
height: auto;
max-height: 12;
background: transparent;
padding: 0;
margin: 0;
border: round #333333;
overflow-y: auto;
scrollbar-background: #000000;
scrollbar-color: #333333;
scrollbar-corner-color: #000000;
scrollbar-size-vertical: 1;
}
#vulnerabilities_panel.hidden {
display: none;
}
.vuln-item {
height: auto;
width: 100%;
padding: 0 1;
background: transparent;
color: #d4d4d4;
}
.vuln-item:hover {
background: #1a1a1a;
color: #fafaf9;
}
VulnerabilityDetailScreen {
align: center middle;
background: #000000 80%;
}
#vuln_detail_dialog {
grid-size: 1;
grid-gutter: 1;
grid-rows: 1fr auto;
padding: 2 3;
width: 85%;
max-width: 110;
height: 85%;
max-height: 45;
border: solid #262626;
background: #0a0a0a;
}
#vuln_detail_scroll {
height: 1fr;
background: transparent;
scrollbar-background: #0a0a0a;
scrollbar-color: #404040;
scrollbar-corner-color: #0a0a0a;
scrollbar-size: 1 1;
padding-right: 1;
}
#vuln_detail_content {
width: 100%;
background: transparent;
padding: 0;
}
#vuln_detail_buttons {
width: 100%;
height: auto;
align: right middle;
padding-top: 1;
margin: 0;
border-top: solid #1a1a1a;
}
#copy_vuln_detail {
width: auto;
min-width: 12;
height: auto;
background: transparent;
color: #525252;
border: none;
text-style: none;
margin: 0 1;
padding: 0 2;
}
#close_vuln_detail {
width: auto;
min-width: 10;
height: auto;
background: transparent;
color: #a3a3a3;
border: none;
text-style: none;
margin: 0;
padding: 0 2;
}
#copy_vuln_detail:hover, #copy_vuln_detail:focus {
background: transparent;
color: #22c55e;
border: none;
}
#close_vuln_detail:hover, #close_vuln_detail:focus {
background: transparent;
color: #ffffff;
border: none;
}
#chat_area_container {
width: 80%;
background: transparent;
}
#chat_area_container.-full-width {
width: 100%;
}
#chat_history {
height: 1fr;
background: transparent;
border: round #0a0a0a;
padding: 0;
margin-bottom: 0;
margin-right: 0;
scrollbar-background: #000000;
scrollbar-color: #1a1a1a;
scrollbar-corner-color: #000000;
scrollbar-size: 1 1;
}
#agent_status_display {
height: 1;
background: transparent;
margin: 0;
padding: 0 1;
}
#agent_status_display.hidden {
display: none;
}
#status_text {
width: 1fr;
height: 100%;
background: transparent;
color: #a3a3a3;
text-align: left;
content-align: left middle;
text-style: none;
margin: 0;
padding: 0;
}
#keymap_indicator {
width: auto;
height: 100%;
background: transparent;
color: #737373;
text-align: right;
content-align: right middle;
text-style: none;
margin: 0;
padding: 0;
}
#chat_input_container {
height: 3;
background: transparent;
border: round #333333;
margin-right: 0;
padding: 0;
layout: horizontal;
align-vertical: top;
}
#chat_input_container:focus-within {
border: round #22c55e;
}
#chat_input_container:focus-within #chat_prompt {
color: #22c55e;
text-style: bold;
}
#chat_prompt {
width: auto;
height: 100%;
padding: 0 0 0 1;
color: #737373;
content-align-vertical: top;
}
#chat_history:focus {
border: round #22c55e;
}
#chat_input {
width: 1fr;
height: 100%;
background: transparent;
border: none;
color: #d4d4d4;
padding: 0;
margin: 0;
}
#chat_input:focus {
border: none;
}
#chat_input .text-area--cursor-line {
background: transparent;
}
#chat_input:focus .text-area--cursor-line {
background: transparent;
}
#chat_input > .text-area--placeholder {
color: #525252;
text-style: italic;
}
#chat_input > .text-area--cursor {
color: #22c55e;
background: #22c55e;
}
.chat-placeholder {
width: 100%;
height: 100%;
content-align: center middle;
text-align: center;
color: #737373;
text-style: italic;
}
.chat-content {
margin: 0 !important;
margin-top: 0 !important;
margin-bottom: 0 !important;
padding: 0 1;
background: transparent;
width: 100%;
}
.chat-message {
margin-bottom: 0;
padding: 0;
background: transparent;
width: 100%;
}
.user-message {
color: #e5e5e5;
border-left: thick #3b82f6;
padding-left: 1;
margin-bottom: 1;
}
.tool-call {
margin-top: 1;
margin-bottom: 0;
padding: 0 1;
background: transparent;
border: none;
width: 100%;
}
.tool-call.status-completed {
background: transparent;
margin-top: 1;
margin-bottom: 0;
}
.tool-call.status-running {
background: transparent;
margin-top: 1;
margin-bottom: 0;
}
.tool-call.status-failed,
.tool-call.status-error {
background: transparent;
margin-top: 1;
margin-bottom: 0;
}
.browser-tool,
.terminal-tool,
.agents-graph-tool,
.file-edit-tool,
.proxy-tool,
.notes-tool,
.thinking-tool,
.web-search-tool,
.scan-info-tool,
.subagent-info-tool {
margin-top: 1;
margin-bottom: 0;
background: transparent;
}
.finish-tool,
.reporting-tool {
margin-top: 1;
margin-bottom: 0;
background: transparent;
}
.browser-tool.status-completed,
.browser-tool.status-running,
.terminal-tool.status-completed,
.terminal-tool.status-running,
.agents-graph-tool.status-completed,
.agents-graph-tool.status-running,
.file-edit-tool.status-completed,
.file-edit-tool.status-running,
.proxy-tool.status-completed,
.proxy-tool.status-running,
.notes-tool.status-completed,
.notes-tool.status-running,
.thinking-tool.status-completed,
.thinking-tool.status-running,
.web-search-tool.status-completed,
.web-search-tool.status-running,
.scan-info-tool.status-completed,
.scan-info-tool.status-running,
.subagent-info-tool.status-completed,
.subagent-info-tool.status-running {
background: transparent;
margin-top: 1;
margin-bottom: 0;
}
.finish-tool.status-completed,
.finish-tool.status-running,
.reporting-tool.status-completed,
.reporting-tool.status-running {
background: transparent;
margin-top: 1;
margin-bottom: 0;
}
Tree {
background: transparent;
color: #e7e5e4;
scrollbar-background: transparent;
scrollbar-color: #404040;
scrollbar-corner-color: transparent;
scrollbar-size: 1 1;
}
Tree > .tree--label {
text-style: bold;
color: #a8a29e;
background: transparent;
padding: 0 1;
margin-bottom: 1;
border-bottom: solid #1a1a1a;
text-align: center;
}
.tree--node {
height: 1;
padding: 0;
margin: 0;
}
.tree--node-label {
color: #d6d3d1;
background: transparent;
text-style: none;
padding: 0 1;
margin: 0 1;
}
.tree--node:hover .tree--node-label {
background: transparent;
color: #fafaf9;
text-style: bold;
border-left: solid #a8a29e;
}
.tree--node.-selected .tree--node-label {
background: transparent;
color: #fafaf9;
text-style: bold;
border-left: heavy #d6d3d1;
}
.tree--node.-expanded .tree--node-label {
text-style: bold;
color: #fafaf9;
background: transparent;
border-left: solid #78716c;
}
Tree:focus {
border: round #1a1a1a;
}
Tree:focus > .tree--label {
color: #fafaf9;
text-style: bold;
background: transparent;
}
.tree--node .tree--node .tree--node-label {
color: #a8a29e;
padding-left: 2;
border: none;
background: transparent;
margin-left: 1;
}
.tree--node .tree--node:hover .tree--node-label {
background: transparent;
color: #e7e5e4;
}
.tree--node .tree--node .tree--node .tree--node-label {
color: #78716c;
padding-left: 3;
text-style: none;
border: none;
background: transparent;
margin-left: 2;
}
StopAgentScreen {
align: center middle;
background: $background 0%;
}
#stop_agent_dialog {
grid-size: 1;
grid-gutter: 1;
grid-rows: auto auto;
padding: 1;
width: 30;
height: auto;
border: round #a3a3a3;
background: #000000 98%;
}
#stop_agent_title {
color: #a3a3a3;
text-style: bold;
text-align: center;
width: 100%;
margin-bottom: 0;
}
#stop_agent_buttons {
grid-size: 2;
grid-gutter: 1;
grid-columns: 1fr 1fr;
width: 100%;
height: 1;
}
#stop_agent_buttons Button {
height: 1;
min-height: 1;
border: none;
text-style: bold;
}
#stop_agent {
background: transparent;
color: #ef4444;
border: none;
}
#stop_agent:hover, #stop_agent:focus {
background: #ef4444;
color: #ffffff;
border: none;
}
#cancel_stop {
background: transparent;
color: #737373;
border: none;
}
#cancel_stop:hover, #cancel_stop:focus {
background:rgb(54, 54, 54);
color: #ffffff;
border: none;
}
QuitScreen {
align: center middle;
background: $background 0%;
}
#quit_dialog {
grid-size: 1;
grid-gutter: 1;
grid-rows: auto auto;
padding: 1;
width: 24;
height: auto;
border: round #333333;
background: #000000 98%;
}
#quit_title {
color: #d4d4d4;
text-style: bold;
text-align: center;
width: 100%;
margin-bottom: 0;
}
#quit_buttons {
grid-size: 2;
grid-gutter: 1;
grid-columns: 1fr 1fr;
width: 100%;
height: 1;
}
#quit_buttons Button {
height: 1;
min-height: 1;
border: none;
text-style: bold;
}
#quit {
background: transparent;
color: #ef4444;
border: none;
}
#quit:hover, #quit:focus {
background: #ef4444;
color: #ffffff;
border: none;
}
#cancel {
background: transparent;
color: #737373;
border: none;
}
#cancel:hover, #cancel:focus {
background:rgb(54, 54, 54);
color: #ffffff;
border: none;
}
HelpScreen {
align: center middle;
background: $background 0%;
}
#dialog {
grid-size: 1;
grid-gutter: 0 1;
grid-rows: auto auto;
padding: 1 2;
width: 40;
height: auto;
border: round #22c55e;
background: #000000 98%;
}
#help_title {
color: #22c55e;
text-style: bold;
text-align: center;
width: 100%;
margin-bottom: 1;
}
#help_content {
color: #d4d4d4;
text-align: left;
width: 100%;
margin-bottom: 1;
padding: 0;
background: transparent;
text-style: none;
}
+217
View File
@@ -0,0 +1,217 @@
import atexit
import contextlib
import logging
import signal
import sys
import threading
import time
from typing import Any
from rich.console import Console
from rich.live import Live
from rich.panel import Panel
from rich.text import Text
from strix.config import load_settings
from strix.core.runner import run_strix_scan
from strix.report.state import ReportState, set_global_report_state
from strix.runtime import session_manager
from .utils import (
build_live_stats_text,
format_vulnerability_report,
)
logger = logging.getLogger(__name__)
def _resolve_sandbox_image() -> str:
image = load_settings().runtime.image
if not image:
raise RuntimeError(
"strix_image is not configured. Set it in ~/.strix/cli-config.json.",
)
return image
async def run_cli(args: Any) -> None: # noqa: PLR0915
console = Console()
start_text = Text()
start_text.append("Penetration test initiated", style="bold #22c55e")
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")
results_text = Text()
results_text.append("Output", style="dim")
results_text.append(" ")
results_text.append(f"strix_runs/{args.run_name}", style="#60a5fa")
note_text = Text()
note_text.append("\n\n", style="dim")
note_text.append("Vulnerabilities will be displayed in real-time.", style="dim")
startup_panel = Panel(
Text.assemble(
start_text,
"\n\n",
target_text,
"\n",
results_text,
note_text,
),
title="[bold white]STRIX",
title_align="left",
border_style="#22c55e",
padding=(1, 2),
)
console.print("\n")
console.print(startup_panel)
console.print()
scan_mode = getattr(args, "scan_mode", "deep")
scan_config: dict[str, Any] = {
"scan_id": args.run_name,
"targets": args.targets_info,
"user_instructions": args.instruction or "",
"run_name": args.run_name,
"diff_scope": getattr(args, "diff_scope", {"active": False}),
"scan_mode": scan_mode,
"non_interactive": bool(getattr(args, "non_interactive", False)),
"local_sources": getattr(args, "local_sources", None) or [],
"scope_mode": getattr(args, "scope_mode", "auto"),
"diff_base": getattr(args, "diff_base", None),
"resume_instruction": getattr(args, "user_explicit_instruction", None) or "",
}
report_state = ReportState(args.run_name)
report_state.hydrate_from_run_dir()
report_state.set_scan_config(scan_config)
report_state.save_run_data()
def display_vulnerability(report: dict[str, Any]) -> None:
report_id = report.get("id", "unknown")
vuln_text = format_vulnerability_report(report)
vuln_panel = Panel(
vuln_text,
title=f"[bold red]{report_id.upper()}",
title_align="left",
border_style="red",
padding=(1, 2),
)
console.print(vuln_panel)
console.print()
report_state.vulnerability_found_callback = display_vulnerability
def cleanup_on_exit() -> None:
report_state.cleanup()
def signal_handler(_signum: int, _frame: Any) -> None:
report_state.cleanup(status="interrupted")
sys.exit(1)
atexit.register(cleanup_on_exit)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
if hasattr(signal, "SIGHUP"):
signal.signal(signal.SIGHUP, signal_handler)
set_global_report_state(report_state)
def create_live_status() -> Panel:
status_text = Text()
status_text.append("Penetration test in progress", style="bold #22c55e")
status_text.append("\n\n")
stats_text = build_live_stats_text(report_state)
if stats_text:
status_text.append(stats_text)
return Panel(
status_text,
title="[bold white]STRIX",
title_align="left",
border_style="#22c55e",
padding=(1, 2),
)
try:
console.print()
with Live(
create_live_status(), console=console, refresh_per_second=2, transient=False
) as live:
stop_updates = threading.Event()
def update_status() -> None:
while not stop_updates.is_set():
try:
live.update(create_live_status())
time.sleep(2)
except Exception:
break
update_thread = threading.Thread(target=update_status, daemon=True)
update_thread.start()
try:
logger.info(
"CLI launching scan: run_name=%s targets=%d interactive=%s",
args.run_name,
len(scan_config.get("targets") or []),
bool(getattr(args, "interactive", False)),
)
await run_strix_scan(
scan_config=scan_config,
scan_id=args.run_name,
image=_resolve_sandbox_image(),
local_sources=getattr(args, "local_sources", None) or [],
interactive=bool(getattr(args, "interactive", False)),
max_budget_usd=getattr(args, "max_budget_usd", None),
)
finally:
stop_updates.set()
update_thread.join(timeout=1)
with contextlib.suppress(Exception):
await session_manager.cleanup(args.run_name)
except Exception as e:
console.print(f"[bold red]Error during penetration test:[/] {e}")
raise
if report_state.final_scan_result:
console.print()
final_report_text = Text()
final_report_text.append("Penetration test summary", style="bold #60a5fa")
final_report_panel = Panel(
Text.assemble(
final_report_text,
"\n\n",
report_state.final_scan_result,
),
title="[bold white]STRIX",
title_align="left",
border_style="#60a5fa",
padding=(1, 2),
)
console.print(final_report_panel)
console.print()
+849
View File
@@ -0,0 +1,849 @@
#!/usr/bin/env python3
"""
Strix Agent Interface
"""
import argparse
import asyncio
import shutil
import sys
from datetime import UTC, datetime
from pathlib import Path
from agents.model_settings import ModelSettings
from agents.models.interface import ModelTracing
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.config.models import (
StrixProvider,
configure_sdk_model_defaults,
is_known_openai_bare_model,
)
from strix.core.paths import run_dir_for, runtime_state_dir
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,
build_mount_targets_info,
check_docker_connection,
clone_repository,
collect_local_sources,
dedupe_local_targets,
find_oversized_local_targets,
generate_run_name,
image_exists,
infer_target_type,
is_whitebox_scan,
process_pull_line,
resolve_diff_scope_context,
rewrite_localhost_targets,
validate_config_file,
)
from strix.report.state import get_global_report_state
from strix.report.writer import read_run_record, write_run_record
from strix.telemetry import posthog, scarf
from strix.telemetry.logging import configure_dependency_logging
HOST_GATEWAY_HOSTNAME = "host.docker.internal"
import logging # noqa: E402
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 (e.g., 'openai/gpt-5.4' or "
"'anthropic/claude-opus-4-7')\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:
settings = load_settings()
configure_sdk_model_defaults(settings)
llm = settings.llm
raw_model = (llm.model or "").strip()
if (
raw_model
and "/" not in raw_model
and not is_known_openai_bare_model(raw_model)
and not llm.api_base
):
warn_text = Text()
warn_text.append("UNKNOWN MODEL NAME", style="bold yellow")
warn_text.append("\n\n", style="white")
warn_text.append(f"'{raw_model}'", style="bold cyan")
warn_text.append(
" is not a known OpenAI model. Bare names route to OpenAI by default.\n"
"If you meant a non-OpenAI provider, use the '",
style="white",
)
warn_text.append("<provider>/<model>", style="bold cyan")
warn_text.append(
"' form, e.g. 'anthropic/claude-opus-4-7', 'deepseek/deepseek-v4-pro'.",
style="white",
)
console.print(
Panel(
warn_text,
title="[bold white]STRIX",
title_align="left",
border_style="yellow",
padding=(1, 2),
),
)
sys.exit(1)
model = StrixProvider().get_model(raw_model)
await asyncio.wait_for(
model.get_response(
system_instructions="You are a helpful assistant.",
input="Reply with just 'OK'.",
model_settings=ModelSettings(),
tools=[],
output_schema=None,
handoffs=[],
tracing=ModelTracing.DISABLED,
previous_response_id=None,
conversation_id=None,
prompt=None,
),
timeout=llm.timeout,
)
logger.info("LLM warm-up succeeded for model %s", (llm.model or "").strip())
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 _positive_budget(value: str) -> float:
try:
budget = float(value)
except ValueError as exc:
raise argparse.ArgumentTypeError(f"invalid float value: {value!r}") from exc
import math
if not math.isfinite(budget) or budget <= 0:
raise argparse.ArgumentTypeError("must be a finite number greater than 0")
return budget
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
# Large local repository (bind-mounted read-only instead of copied)
strix --mount ./huge-monorepo
# 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(
"--mount",
type=str,
action="append",
metavar="PATH",
help="Bind-mount a local directory into the sandbox (read-only) instead of "
"copying it file-by-file. Use this for large repositories that are too big to "
"stream into the container. Can be specified multiple times.",
)
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(
"--max-budget-usd",
type=_positive_budget,
default=None,
help="Maximum LLM cost in USD (> 0). The scan stops cleanly when this limit is reached.",
)
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}")
args.user_explicit_instruction = args.instruction if args.resume else None
if args.resume:
if args.target or args.mount:
parser.error(
"Cannot combine --resume with --target/--mount. --resume picks up where "
"the prior run left off, including the original target list."
)
_load_resume_state(args, parser)
agents_path = runtime_state_dir(run_dir_for(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 and not args.mount:
parser.error(
"the following arguments are required: -t/--target or --mount "
"(or use --resume <run_name> to continue a prior scan)"
)
args.targets_info = []
for target in args.target or []:
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}'")
try:
args.targets_info.extend(build_mount_targets_info(args.mount or []))
except ValueError as e:
parser.error(str(e))
args.targets_info = dedupe_local_targets(args.targets_info)
assign_workspace_subdirs(args.targets_info)
rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
max_local_copy_mb = load_settings().runtime.max_local_copy_mb
max_copy_bytes = max_local_copy_mb * 1024 * 1024
oversized = find_oversized_local_targets(args.targets_info, max_copy_bytes)
if oversized:
details = "; ".join(
f"{path} ({size / (1024 * 1024):.0f} MB)" for path, size in oversized
)
parser.error(
f"Local target too large to stream into the sandbox: {details}. "
f"The limit is {max_local_copy_mb} MB "
"(set STRIX_MAX_LOCAL_COPY_MB to change it). Re-run with "
"--mount <path> to bind-mount the directory instead of copying it."
)
return args
def _persist_run_record(args: argparse.Namespace) -> None:
run_dir = run_dir_for(args.run_name)
run_dir.mkdir(parents=True, exist_ok=True)
run_record = {
"run_id": args.run_name,
"run_name": args.run_name,
"status": "running",
"start_time": datetime.now(UTC).isoformat(),
"end_time": None,
"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,
}
write_run_record(run_dir, run_record)
def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser) -> None:
"""Populate ``args.targets_info`` and friends from a prior run's run.json."""
run_dir = run_dir_for(args.resume)
state_path = run_dir / "run.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 = read_run_record(run_dir)
except RuntimeError as exc:
parser.error(f"--resume {args.resume}: run.json unreadable: {exc}")
args.targets_info = state.get("targets_info") or []
if not args.targets_info:
parser.error(f"--resume {args.resume}: run.json has no targets_info")
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("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":
args.scan_mode = persisted_scan_mode
def display_completion_message(args: argparse.Namespace, results_path: Path) -> None:
console = Console()
report_state = get_global_report_state()
scan_completed = False
if report_state:
scan_completed = report_state.run_record.get("status") == "completed"
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(report_state)
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]docs.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:
configure_dependency_logging()
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_run_record(args)
_telemetry_start_kwargs = {
"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),
}
posthog.start(**_telemetry_start_kwargs)
scarf.start(**_telemetry_start_kwargs)
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:
exit_reason = "error"
posthog.error("unhandled_exception")
scarf.error("unhandled_exception")
raise
finally:
report_state = get_global_report_state()
if report_state:
status = {"interrupted": "interrupted", "error": "failed"}.get(
exit_reason,
"stopped",
)
report_state.cleanup(status=status)
posthog.end(report_state, exit_reason=exit_reason)
scarf.end(report_state, exit_reason=exit_reason)
results_path = run_dir_for(args.run_name)
display_completion_message(args, results_path)
if args.non_interactive:
report_state = get_global_report_state()
if report_state and report_state.vulnerability_reports:
sys.exit(2)
if __name__ == "__main__":
main()
+6
View File
@@ -0,0 +1,6 @@
"""Textual TUI interface."""
from strix.interface.tui.app import StrixTUIApp, run_tui
__all__ = ["StrixTUIApp", "run_tui"]
File diff suppressed because it is too large Load Diff
+60
View File
@@ -0,0 +1,60 @@
"""Historical SDK session loading for the TUI."""
from __future__ import annotations
import json
import logging
import sqlite3
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from strix.core.paths import runtime_state_dir
if TYPE_CHECKING:
from pathlib import Path
logger = logging.getLogger(__name__)
def load_session_history(run_dir: Path, agent_ids: Any) -> list[tuple[str, dict[str, Any], str]]:
agents_db = runtime_state_dir(run_dir) / "agents.db"
session_ids = [aid for aid in agent_ids if isinstance(aid, str)]
if not agents_db.exists() or not session_ids:
return []
session_id_set = set(session_ids)
try:
with sqlite3.connect(agents_db) as conn:
rows = conn.execute(
"select id, session_id, message_data, created_at from agent_messages order by id"
).fetchall()
except sqlite3.Error:
logger.exception("Failed to hydrate TUI history from %s", agents_db)
return []
items: list[tuple[str, dict[str, Any], str]] = []
for row_id, agent_id, message_data, created_at in rows:
if agent_id not in session_id_set:
continue
try:
item = json.loads(message_data)
except (TypeError, json.JSONDecodeError):
logger.debug("Skipping unreadable SDK session item %s for %s", row_id, agent_id)
continue
if isinstance(item, dict):
items.append((str(agent_id), item, _sqlite_timestamp_to_iso(created_at)))
return items
def _sqlite_timestamp_to_iso(value: Any) -> str:
if not isinstance(value, str) or not value.strip():
return datetime.now(UTC).isoformat()
text = value.strip()
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return text
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=UTC)
return parsed.astimezone(UTC).isoformat()
+356
View File
@@ -0,0 +1,356 @@
"""TUI-owned projection of SDK session history and stream events."""
from __future__ import annotations
import json
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from pathlib import Path
from strix.core.paths import runtime_state_dir
from strix.interface.tui.history import load_session_history
class TuiLiveView:
def __init__(self) -> None:
self.agents: dict[str, dict[str, Any]] = {}
self.events: list[dict[str, Any]] = []
self._next_event_id = 1
self._open_assistant_event_by_agent: dict[str, dict[str, Any]] = {}
self._tool_event_by_call_id: dict[str, dict[str, Any]] = {}
def hydrate_from_run_dir(self, run_dir: Path) -> None:
state_dir = runtime_state_dir(run_dir)
agents_path = state_dir / "agents.json"
if not agents_path.exists():
return
try:
agents_data = json.loads(agents_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return
statuses = agents_data.get("statuses") or {}
names = agents_data.get("names") or {}
parent_of = agents_data.get("parent_of") or {}
if not isinstance(statuses, dict):
return
for agent_id, status in statuses.items():
if not isinstance(agent_id, str):
continue
self.upsert_agent(
agent_id,
name=names.get(agent_id, agent_id) if isinstance(names, dict) else agent_id,
parent_id=parent_of.get(agent_id) if isinstance(parent_of, dict) else None,
status=str(status),
)
self._hydrate_sdk_session_history(run_dir, statuses.keys())
def _hydrate_sdk_session_history(self, run_dir: Path, agent_ids: Any) -> None:
for agent_id, item, timestamp in load_session_history(run_dir, agent_ids):
self._ingest_session_history_item(
agent_id,
item,
timestamp=timestamp,
)
def upsert_agent(
self,
agent_id: str,
*,
name: str | None = None,
parent_id: str | None = None,
status: str | None = None,
error_message: str | None = None,
) -> None:
now = datetime.now(UTC).isoformat()
current = self.agents.setdefault(
agent_id,
{
"id": agent_id,
"name": name or agent_id,
"parent_id": parent_id,
"status": status or "running",
"created_at": now,
"updated_at": now,
},
)
if name is not None:
current["name"] = name
if parent_id is not None or "parent_id" not in current:
current["parent_id"] = parent_id
if status is not None:
current["status"] = status
if error_message:
current["error_message"] = error_message
current["updated_at"] = now
def record_user_message(self, agent_id: str, content: str) -> None:
self._append_event(
agent_id,
"chat",
{
"role": "user",
"content": content,
"metadata": {"source": "tui_user"},
},
)
def ingest_sdk_event(self, agent_id: str, event: Any) -> None:
event_type = getattr(event, "type", "")
if event_type == "raw_response_event":
self._ingest_raw_response_event(agent_id, getattr(event, "data", None))
return
if event_type != "run_item_stream_event":
return
item = getattr(event, "item", None)
item_type = getattr(item, "type", "")
if item_type == "message_output_item":
self._record_assistant_message(agent_id, _sdk_message_text(item), final=True)
elif item_type == "tool_call_item":
self._record_tool_call(agent_id, item)
elif item_type == "tool_call_output_item":
self._record_tool_output(agent_id, item)
def events_for_agent(self, agent_id: str) -> list[dict[str, Any]]:
return [event for event in self.events if event.get("agent_id") == agent_id]
def has_events_for_agent(self, agent_id: str) -> bool:
return any(event.get("agent_id") == agent_id for event in self.events)
def _ingest_raw_response_event(self, agent_id: str, data: Any) -> None:
data_type = getattr(data, "type", "")
if data_type == "response.output_text.delta":
delta = getattr(data, "delta", "")
if delta:
self._record_assistant_message(agent_id, str(delta), final=False)
def _ingest_session_history_item(
self,
agent_id: str,
item: dict[str, Any],
*,
timestamp: str,
) -> None:
item_type = item.get("type")
role = item.get("role")
if role in {"user", "assistant"} and (item_type in {None, "message"}):
content = _session_message_text(item)
if content:
self._append_event(
agent_id,
"chat",
{
"role": role,
"content": content,
"metadata": {"source": "sdk_session"},
},
timestamp=timestamp,
)
return
if item_type == "function_call":
self._record_tool_call_data(
agent_id,
{
"call_id": str(item.get("call_id") or item.get("id") or ""),
"tool_name": str(item.get("name") or "tool"),
"args": _parse_json_object(item.get("arguments")),
},
timestamp=timestamp,
)
return
if item_type == "function_call_output":
self._record_tool_output_data(
agent_id,
{
"call_id": str(item.get("call_id") or item.get("id") or ""),
"tool_name": "tool",
"output": item.get("output"),
},
timestamp=timestamp,
)
def _record_assistant_message(self, agent_id: str, content: str, *, final: bool) -> None:
if not content:
return
existing = self._open_assistant_event_by_agent.get(agent_id)
if existing is None:
event = self._append_event(
agent_id,
"chat",
{
"role": "assistant",
"content": content,
"metadata": {"source": "sdk_stream", "streaming": not final},
},
)
if not final:
self._open_assistant_event_by_agent[agent_id] = event
return
data = existing["data"]
if final:
data["content"] = content
data["metadata"]["streaming"] = False
self._open_assistant_event_by_agent.pop(agent_id, None)
else:
data["content"] = f"{data.get('content', '')}{content}"
self._bump_event(existing)
def _record_tool_call(self, agent_id: str, item: Any) -> None:
self._record_tool_call_data(agent_id, _sdk_tool_call_data(item))
def _record_tool_call_data(
self,
agent_id: str,
call: dict[str, Any],
*,
timestamp: str | None = None,
) -> None:
call_id = call["call_id"]
existing = self._tool_event_by_call_id.get(call_id)
tool_data = {
"tool_name": call["tool_name"],
"args": call["args"],
"status": "running",
"agent_id": agent_id,
"call_id": call_id,
}
if existing is None:
event = self._append_event(agent_id, "tool", tool_data, timestamp=timestamp)
self._tool_event_by_call_id[call_id] = event
else:
existing["data"].update(tool_data)
self._bump_event(existing, timestamp=timestamp)
def _record_tool_output(self, agent_id: str, item: Any) -> None:
self._record_tool_output_data(agent_id, _sdk_tool_output_data(item))
def _record_tool_output_data(
self,
agent_id: str,
output: dict[str, Any],
*,
timestamp: str | None = None,
) -> None:
call_id = output["call_id"]
event = self._tool_event_by_call_id.get(call_id)
if event is None:
event = self._append_event(
agent_id,
"tool",
{
"tool_name": output["tool_name"],
"args": {},
"status": "completed",
"agent_id": agent_id,
"call_id": call_id,
},
timestamp=timestamp,
)
self._tool_event_by_call_id[call_id] = event
result = _parse_json_value(output["output"])
event["data"]["result"] = result
event["data"]["status"] = _tool_status_from_result(result)
self._bump_event(event, timestamp=timestamp)
def _append_event(
self,
agent_id: str,
event_type: str,
data: dict[str, Any],
*,
timestamp: str | None = None,
) -> dict[str, Any]:
event = {
"id": f"{event_type}_{self._next_event_id}",
"type": event_type,
"agent_id": agent_id,
"timestamp": timestamp or datetime.now(UTC).isoformat(),
"version": 0,
"data": data,
}
self._next_event_id += 1
self.events.append(event)
return event
@staticmethod
def _bump_event(event: dict[str, Any], *, timestamp: str | None = None) -> None:
event["version"] = int(event.get("version", 0)) + 1
event["timestamp"] = timestamp or datetime.now(UTC).isoformat()
def _sdk_tool_call_data(item: Any) -> dict[str, Any]:
raw = getattr(item, "raw_item", None)
call_id = str(_raw_field(raw, "call_id") or _raw_field(raw, "id") or id(item))
tool_name = str(
_raw_field(raw, "name") or _raw_field(raw, "type") or getattr(item, "title", None) or "tool"
)
return {
"call_id": call_id,
"tool_name": tool_name,
"args": _parse_json_object(_raw_field(raw, "arguments")),
}
def _sdk_tool_output_data(item: Any) -> dict[str, Any]:
raw = getattr(item, "raw_item", None)
call_id = str(_raw_field(raw, "call_id") or _raw_field(raw, "id") or id(item))
return {
"call_id": call_id,
"tool_name": str(_raw_field(raw, "name") or _raw_field(raw, "type") or "tool"),
"output": getattr(item, "output", _raw_field(raw, "output")),
}
def _sdk_message_text(item: Any) -> str:
raw = getattr(item, "raw_item", None)
return _message_content_text(_raw_field(raw, "content", []))
def _session_message_text(item: dict[str, Any]) -> str:
return _message_content_text(item.get("content", ""))
def _message_content_text(content: Any) -> str:
parts: list[str] = []
content_items = content if isinstance(content, list) else [content]
for part in content_items:
if isinstance(part, str):
parts.append(part)
continue
text = _raw_field(part, "text")
if isinstance(text, str):
parts.append(text)
return "".join(parts)
def _raw_field(raw: Any, key: str, default: Any = None) -> Any:
if isinstance(raw, dict):
return raw.get(key, default)
return getattr(raw, key, default)
def _parse_json_object(value: Any) -> dict[str, Any]:
parsed = _parse_json_value(value)
return parsed if isinstance(parsed, dict) else {}
def _parse_json_value(value: Any) -> Any:
if not isinstance(value, str):
return value
try:
return json.loads(value)
except json.JSONDecodeError:
return value
def _tool_status_from_result(result: Any) -> str:
if isinstance(result, dict) and result.get("success") is False:
return "failed"
return "completed"
+43
View File
@@ -0,0 +1,43 @@
"""Message delivery bridge from TUI input to SDK-backed agents."""
from __future__ import annotations
import asyncio
import logging
from typing import Any
logger = logging.getLogger(__name__)
def send_user_message_to_agent(
*,
coordinator: Any,
loop: asyncio.AbstractEventLoop | None,
live_view: Any,
target_agent_id: str,
message: str,
) -> bool:
if loop is None or loop.is_closed():
return False
live_view.record_user_message(target_agent_id, message)
future = asyncio.run_coroutine_threadsafe(
coordinator.send(
target_agent_id,
{"from": "user", "content": message, "type": "instruction"},
),
loop,
)
future.add_done_callback(_log_delivery_failure)
return True
def _log_delivery_failure(future: Any) -> None:
try:
delivered = bool(future.result())
except Exception:
logger.exception("TUI user message delivery failed")
return
if not delivered:
logger.warning("TUI user message was not persisted to the SDK session")
+30
View File
@@ -0,0 +1,30 @@
from . import (
agents_graph_renderer,
filesystem_renderer,
finish_renderer,
load_skill_renderer,
notes_renderer,
proxy_renderer,
reporting_renderer,
shell_renderer,
thinking_renderer,
todo_renderer,
web_search_renderer,
)
from .registry import render_tool_widget
__all__ = [
"agents_graph_renderer",
"filesystem_renderer",
"finish_renderer",
"load_skill_renderer",
"notes_renderer",
"proxy_renderer",
"render_tool_widget",
"reporting_renderer",
"shell_renderer",
"thinking_renderer",
"todo_renderer",
"web_search_renderer",
]
@@ -0,0 +1,171 @@
import re
from functools import cache
from typing import Any
from pygments.lexers import get_lexer_by_name, guess_lexer
from pygments.styles import get_style_by_name
from pygments.util import ClassNotFound
from rich.text import Text
_BLANK_LINE_RUNS = re.compile(r"\n\s*\n")
_HEADER_STYLES = [
("###### ", 7, "bold #4ade80"),
("##### ", 6, "bold #22c55e"),
("#### ", 5, "bold #16a34a"),
("### ", 4, "bold #15803d"),
("## ", 3, "bold #22c55e"),
("# ", 2, "bold #4ade80"),
]
@cache
def _get_style_colors() -> dict[Any, str]:
style = get_style_by_name("native")
return {token: f"#{style_def['color']}" for token, style_def in style if style_def["color"]}
def _get_token_color(token_type: Any) -> str | None:
colors = _get_style_colors()
while token_type:
if token_type in colors:
return colors[token_type]
token_type = token_type.parent
return None
def _highlight_code(code: str, language: str | None = None) -> Text:
text = Text()
try:
lexer = get_lexer_by_name(language) if language else guess_lexer(code)
except ClassNotFound:
text.append(code, style="#d4d4d4")
return text
for token_type, token_value in lexer.get_tokens(code):
if not token_value:
continue
color = _get_token_color(token_type)
text.append(token_value, style=color)
return text
def _try_parse_header(line: str) -> tuple[str, str] | None:
for prefix, strip_len, style in _HEADER_STYLES:
if line.startswith(prefix):
return (line[strip_len:], style)
return None
def _apply_markdown_styles(text: str) -> Text: # noqa: PLR0912
result = Text()
lines = text.split("\n")
in_code_block = False
code_block_lang: str | None = None
code_block_lines: list[str] = []
for i, line in enumerate(lines):
if i > 0 and not in_code_block:
result.append("\n")
if line.startswith("```"):
if not in_code_block:
in_code_block = True
code_block_lang = line[3:].strip() or None
code_block_lines = []
if i > 0:
result.append("\n")
else:
in_code_block = False
code_content = "\n".join(code_block_lines)
if code_content:
result.append_text(_highlight_code(code_content, code_block_lang))
code_block_lines = []
code_block_lang = None
continue
if in_code_block:
code_block_lines.append(line)
continue
header = _try_parse_header(line)
if header:
result.append(header[0], style=header[1])
elif line.startswith("> "):
result.append("", style="#22c55e")
result.append_text(_process_inline_formatting(line[2:]))
elif line.startswith(("- ", "* ")):
result.append("", style="#22c55e")
result.append_text(_process_inline_formatting(line[2:]))
elif len(line) > 2 and line[0].isdigit() and line[1:3] in (". ", ") "):
result.append(line[0] + ". ", style="#22c55e")
result.append_text(_process_inline_formatting(line[2:]))
elif line.strip() in ("---", "***", "___"):
result.append("" * 40, style="#22c55e")
else:
result.append_text(_process_inline_formatting(line))
if in_code_block and code_block_lines:
code_content = "\n".join(code_block_lines)
result.append_text(_highlight_code(code_content, code_block_lang))
return result
def _process_inline_formatting(line: str) -> Text:
result = Text()
i = 0
n = len(line)
while i < n:
if i + 1 < n and line[i : i + 2] in ("**", "__"):
marker = line[i : i + 2]
end = line.find(marker, i + 2)
if end != -1:
result.append(line[i + 2 : end], style="bold #4ade80")
i = end + 2
continue
if i + 1 < n and line[i : i + 2] == "~~":
end = line.find("~~", i + 2)
if end != -1:
result.append(line[i + 2 : end], style="strike #525252")
i = end + 2
continue
if line[i] == "`":
end = line.find("`", i + 1)
if end != -1:
result.append(line[i + 1 : end], style="bold #22c55e on #0a0a0a")
i = end + 1
continue
if line[i] in ("*", "_"):
marker = line[i]
if i + 1 < n and line[i + 1] != marker:
end = line.find(marker, i + 1)
if end != -1 and (end + 1 >= n or line[end + 1] != marker):
result.append(line[i + 1 : end], style="italic #86efac")
i = end + 1
continue
result.append(line[i])
i += 1
return result
class AgentMessageRenderer:
@classmethod
def render_simple(cls, content: str) -> Text:
if not content:
return Text()
cleaned = _BLANK_LINE_RUNS.sub("\n\n", content).strip()
if not cleaned:
return Text()
return _apply_markdown_styles(cleaned)
@@ -0,0 +1,175 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
@register_tool_renderer
class ViewAgentGraphRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "view_agent_graph"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
status = tool_data.get("status", "unknown")
text = Text()
text.append("", style="#a78bfa")
text.append("viewing agents graph", style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class CreateAgentRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_agent"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
task = args.get("task", "")
name = args.get("name", "Agent")
text = Text()
text.append("", style="#a78bfa")
text.append("spawning ", style="dim")
text.append(name, style="bold #a78bfa")
if task:
text.append("\n ")
text.append(task, style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class SendMessageToAgentRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "send_message_to_agent"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
message = args.get("message", "")
target_agent_id = args.get("target_agent_id", "")
text = Text()
text.append("", style="#60a5fa")
if target_agent_id:
text.append(f"to {target_agent_id}", style="dim")
else:
text.append("sending message", style="dim")
if message:
text.append("\n ")
text.append(message, style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class AgentFinishRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "agent_finish"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
result_summary = args.get("result_summary", "")
findings = args.get("findings", [])
success = args.get("success", True)
text = Text()
if success:
text.append("", style="#22c55e")
text.append("Agent completed", style="bold #22c55e")
else:
text.append("", style="#ef4444")
text.append("Agent failed", style="bold #ef4444")
if result_summary:
text.append("\n ")
text.append(result_summary, style="bold")
if findings and isinstance(findings, list):
for finding in findings:
text.append("\n")
text.append(str(finding), style="dim")
else:
text.append("\n ")
text.append("Completing task...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class WaitForMessageRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "wait_for_message"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
reason = args.get("reason", "")
text = Text()
text.append("", style="#6b7280")
text.append("waiting", style="dim")
if reason:
text.append("\n ")
text.append(reason, style="dim")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class StopAgentRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "stop_agent"
css_classes: ClassVar[list[str]] = ["tool-call", "agents-graph-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "unknown")
target_agent_id = args.get("target_agent_id", "")
cascade = args.get("cascade", True)
reason = args.get("reason", "")
text = Text()
text.append("", style="#ef4444")
text.append("stopping", style="dim")
if target_agent_id:
text.append(f" {target_agent_id}", style="bold #ef4444")
if cascade:
text.append(" + descendants", style="dim italic")
if reason:
text.append("\n ")
text.append(reason, style="dim")
if isinstance(result, dict) and result.get("success") is False and result.get("error"):
text.append("\n ")
text.append(str(result["error"]), style="#ef4444")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@@ -0,0 +1,30 @@
from abc import ABC, abstractmethod
from typing import Any, ClassVar
from textual.widgets import Static
class BaseToolRenderer(ABC):
tool_name: ClassVar[str] = ""
css_classes: ClassVar[list[str]] = ["tool-call"]
@classmethod
@abstractmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
pass
@classmethod
def status_icon(cls, status: str) -> tuple[str, str]:
icons = {
"running": ("● In progress...", "#f59e0b"),
"completed": ("✓ Done", "#22c55e"),
"failed": ("✗ Failed", "#dc2626"),
"error": ("✗ Error", "#dc2626"),
}
return icons.get(status, ("○ Unknown", "dim"))
@classmethod
def get_css_classes(cls, status: str) -> str:
base_classes = cls.css_classes.copy()
base_classes.append(f"status-{status}")
return " ".join(base_classes)
@@ -0,0 +1,266 @@
from __future__ import annotations
import json
from functools import cache
from typing import Any, ClassVar
from pygments.lexers import get_lexer_by_name, get_lexer_for_filename
from pygments.styles import get_style_by_name
from pygments.util import ClassNotFound
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
_ADD_FILE = "*** Add File: "
_DELETE_FILE = "*** Delete File: "
_UPDATE_FILE = "*** Update File: "
_BEGIN_PATCH = "*** Begin Patch"
_END_PATCH = "*** End Patch"
_VIEW_IMAGE_ERROR_PREFIXES = (
"image path ",
"unable to read image",
"manifest path",
"exceeded the allowed size",
)
@cache
def _get_style_colors() -> dict[Any, str]:
style = get_style_by_name("native")
return {token: f"#{style_def['color']}" for token, style_def in style if style_def["color"]}
def _get_lexer_for_file(path: str) -> Any:
try:
return get_lexer_for_filename(path)
except ClassNotFound:
return get_lexer_by_name("text")
def _get_token_color(token_type: Any) -> str | None:
colors = _get_style_colors()
while token_type:
if token_type in colors:
return colors[token_type]
token_type = token_type.parent
return None
def _highlight_code(code: str, path: str) -> Text:
lexer = _get_lexer_for_file(path)
text = Text()
for token_type, token_value in lexer.get_tokens(code):
if not token_value:
continue
color = _get_token_color(token_type)
text.append(token_value, style=color)
return text
def _extract_patch_text(args: dict[str, Any]) -> str:
"""apply_patch input arrives as either {"patch": text} or raw text in
the "input" field, depending on whether the tool is wrapped as a
chat-completions FunctionTool or routed through as a CustomTool.
"""
raw = args.get("patch")
if isinstance(raw, str):
return raw
if isinstance(raw, dict):
inner = raw.get("patch")
if isinstance(inner, str):
return inner
fallback = args.get("input") if isinstance(args, dict) else None
if isinstance(fallback, str):
return fallback
return ""
def _parse_patch_operations(
patch_text: str,
) -> list[tuple[str, str, list[str], list[str]]]:
"""Return [(kind, path, old_lines, new_lines), ...] for each file op."""
ops: list[tuple[str, str, list[str], list[str]]] = []
current_kind: str | None = None
current_path: str | None = None
old_lines: list[str] = []
new_lines: list[str] = []
def flush() -> None:
nonlocal current_kind, current_path, old_lines, new_lines
if current_kind and current_path is not None:
ops.append((current_kind, current_path, old_lines, new_lines))
current_kind = None
current_path = None
old_lines = []
new_lines = []
for line in patch_text.splitlines():
if line in (_BEGIN_PATCH, _END_PATCH):
continue
if line.startswith(_ADD_FILE):
flush()
current_kind = "add"
current_path = line[len(_ADD_FILE) :].strip()
elif line.startswith(_UPDATE_FILE):
flush()
current_kind = "update"
current_path = line[len(_UPDATE_FILE) :].strip()
elif line.startswith(_DELETE_FILE):
flush()
current_kind = "delete"
current_path = line[len(_DELETE_FILE) :].strip()
elif current_kind == "update":
if line.startswith("@@"):
continue
if line.startswith("-") and not line.startswith("---"):
old_lines.append(line[1:])
elif line.startswith("+") and not line.startswith("+++"):
new_lines.append(line[1:])
elif current_kind == "add":
if line.startswith("+"):
new_lines.append(line[1:])
elif line.strip():
new_lines.append(line)
flush()
return ops
_OP_LABEL = {
"add": "create",
"update": "edit",
"delete": "delete",
}
def _render_operation(text: Text, kind: str, path: str, old: list[str], new: list[str]) -> None:
label = _OP_LABEL.get(kind, "file")
text.append("", style="#10b981")
text.append(label, style="dim")
if path:
path_display = path[-60:] if len(path) > 60 else path
text.append(" ")
text.append(path_display, style="dim")
if kind == "update":
if old:
highlighted_old = _highlight_code("\n".join(old), path)
for line in highlighted_old.plain.split("\n"):
text.append("\n")
text.append("-", style="#ef4444")
text.append(" ")
text.append(line)
if new:
highlighted_new = _highlight_code("\n".join(new), path)
for line in highlighted_new.plain.split("\n"):
text.append("\n")
text.append("+", style="#22c55e")
text.append(" ")
text.append(line)
elif kind == "add" and new:
text.append("\n")
text.append_text(_highlight_code("\n".join(new), path))
@register_tool_renderer
class ApplyPatchRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "apply_patch"
css_classes: ClassVar[list[str]] = ["tool-call", "file-edit-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "completed")
patch_text = _extract_patch_text(args)
ops = _parse_patch_operations(patch_text)
text = Text()
if not ops:
text.append("", style="#10b981")
text.append("patch", style="dim")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif not result:
text.append(" ")
text.append("Processing...", style="dim")
return Static(text, classes=cls.get_css_classes(status))
for i, (kind, path, old, new) in enumerate(ops):
if i > 0:
text.append("\n")
_render_operation(text, kind, path, old, new)
if status == "failed" and isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="#ef4444")
return Static(text, classes=cls.get_css_classes(status))
def _is_image_success(result: Any) -> bool:
if isinstance(result, dict) and result.get("type") == "image":
return True
if isinstance(result, str):
stripped = result.lstrip()
if stripped.startswith("data:image/"):
return True
try:
obj = json.loads(stripped)
except (TypeError, ValueError):
return False
return isinstance(obj, dict) and obj.get("type") == "image"
return False
def _image_error_text(result: Any) -> str | None:
if not isinstance(result, str):
return None
stripped = result.strip()
if not stripped:
return None
lower = stripped.lower()
if lower.startswith(_VIEW_IMAGE_ERROR_PREFIXES) or "not a supported image" in lower:
return stripped
return None
@register_tool_renderer
class ViewImageRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "view_image"
css_classes: ClassVar[list[str]] = ["tool-call", "file-edit-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "completed")
path = str(args.get("path", "")).strip()
text = Text()
text.append("", style="#10b981")
text.append("view image", style="dim")
if path:
path_display = path[-60:] if len(path) > 60 else path
text.append(" ")
text.append(path_display, style="dim")
err = _image_error_text(result)
if err is not None:
text.append("\n ")
text.append(err, style="#ef4444")
elif _is_image_success(result):
text.append(" ")
text.append("", style="#22c55e")
return Static(text, classes=cls.get_css_classes(status))
@@ -0,0 +1,65 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
FIELD_STYLE = "bold #4ade80"
@register_tool_renderer
class FinishScanRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "finish_scan"
css_classes: ClassVar[list[str]] = ["tool-call", "finish-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
executive_summary = args.get("executive_summary", "")
methodology = args.get("methodology", "")
technical_analysis = args.get("technical_analysis", "")
recommendations = args.get("recommendations", "")
text = Text()
text.append("", style="#22c55e")
text.append("Penetration test completed", style="bold #22c55e")
if executive_summary:
text.append("\n\n")
text.append("Executive Summary", style=FIELD_STYLE)
text.append("\n")
text.append(executive_summary)
if methodology:
text.append("\n\n")
text.append("Methodology", style=FIELD_STYLE)
text.append("\n")
text.append(methodology)
if technical_analysis:
text.append("\n\n")
text.append("Technical Analysis", style=FIELD_STYLE)
text.append("\n")
text.append(technical_analysis)
if recommendations:
text.append("\n\n")
text.append("Recommendations", style=FIELD_STYLE)
text.append("\n")
text.append(recommendations)
if not (executive_summary or methodology or technical_analysis or recommendations):
text.append("\n ")
text.append("Generating final report...", style="dim")
padded = Text()
padded.append("\n\n")
padded.append_text(text)
padded.append("\n\n")
css_classes = cls.get_css_classes("completed")
return Static(padded, classes=css_classes)
@@ -0,0 +1,37 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
@register_tool_renderer
class LoadSkillRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "load_skill"
css_classes: ClassVar[list[str]] = ["tool-call", "load-skill-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "completed")
raw_skills = args.get("skills", "")
if isinstance(raw_skills, list):
requested = ", ".join(str(s) for s in raw_skills)
else:
requested = str(raw_skills)
text = Text()
text.append("", style="#10b981")
text.append("loading skill", style="dim")
if requested:
text.append(" ")
text.append(requested, style="#10b981")
elif not tool_data.get("result"):
text.append("\n ")
text.append("Loading...", style="dim")
return Static(text, classes=cls.get_css_classes(status))
@@ -0,0 +1,167 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
@register_tool_renderer
class CreateNoteRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_note"
css_classes: ClassVar[list[str]] = ["tool-call", "notes-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
title = args.get("title", "")
content = args.get("content", "")
category = args.get("category", "general")
text = Text()
text.append("", style="#fbbf24")
text.append("note", style="dim")
text.append(" ")
text.append(f"({category})", style="dim")
if title:
text.append("\n ")
text.append(title.strip())
if content:
text.append("\n ")
text.append(content.strip(), style="dim")
if not title and not content:
text.append("\n ")
text.append("Capturing...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class DeleteNoteRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "delete_note"
css_classes: ClassVar[list[str]] = ["tool-call", "notes-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: ARG003
text = Text()
text.append("", style="#fbbf24")
text.append("note removed", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class UpdateNoteRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "update_note"
css_classes: ClassVar[list[str]] = ["tool-call", "notes-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
title = args.get("title")
content = args.get("content")
text = Text()
text.append("", style="#fbbf24")
text.append("note updated", style="dim")
if title:
text.append("\n ")
text.append(title)
if content:
text.append("\n ")
text.append(content.strip(), style="dim")
if not title and not content:
text.append("\n ")
text.append("Updating...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class ListNotesRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "list_notes"
css_classes: ClassVar[list[str]] = ["tool-call", "notes-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("", style="#fbbf24")
text.append("notes", style="dim")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif result and isinstance(result, dict) and result.get("success"):
count = result.get("total_count", 0)
notes = result.get("notes", []) or []
if count == 0:
text.append("\n ")
text.append("No notes", style="dim")
else:
for note in notes:
title = note.get("title", "").strip() or "(untitled)"
category = note.get("category", "general")
note_content = note.get("content", "").strip()
if not note_content:
note_content = note.get("content_preview", "").strip()
text.append("\n - ")
text.append(title)
text.append(f" ({category})", style="dim")
if note_content:
text.append("\n ")
text.append(note_content, style="dim")
else:
text.append("\n ")
text.append("Loading...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class GetNoteRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "get_note"
css_classes: ClassVar[list[str]] = ["tool-call", "notes-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("", style="#fbbf24")
text.append("note read", style="dim")
if result and isinstance(result, dict) and result.get("success"):
note = result.get("note", {}) or {}
title = str(note.get("title", "")).strip() or "(untitled)"
category = note.get("category", "general")
content = str(note.get("content", "")).strip()
text.append("\n ")
text.append(title)
text.append(f" ({category})", style="dim")
if content:
text.append("\n ")
text.append(content, style="dim")
else:
text.append("\n ")
text.append("Loading...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@@ -0,0 +1,536 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
PROXY_ICON = "<~>"
MAX_REQUESTS_DISPLAY = 20
MAX_LINE_LENGTH = 200
def _truncate(text: str, max_len: int = 80) -> str:
return text[: max_len - 3] + "..." if len(text) > max_len else text
def _sanitize(text: str, max_len: int = 150) -> str:
clean = text.replace("\n", " ").replace("\r", "").replace("\t", " ")
return _truncate(clean, max_len)
def _status_style(code: int | None) -> str:
if code is None:
return "dim"
if 200 <= code < 300:
return "#22c55e" # green
if 300 <= code < 400:
return "#eab308" # yellow
if 400 <= code < 500:
return "#f97316" # orange
if code >= 500:
return "#ef4444" # red
return "dim"
@register_tool_renderer
class ListRequestsRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "list_requests"
css_classes: ClassVar[list[str]] = ["tool-call", "proxy-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "running")
httpql_filter = args.get("httpql_filter")
sort_by = args.get("sort_by")
sort_order = args.get("sort_order")
scope_id = args.get("scope_id")
text = Text()
text.append(PROXY_ICON, style="dim")
text.append(" listing requests", style="#06b6d4")
if httpql_filter:
text.append(f" where {_truncate(httpql_filter, 150)}", style="dim italic")
meta_parts = []
if sort_by and sort_by != "timestamp":
meta_parts.append(f"by:{sort_by}")
if sort_order and sort_order != "desc":
meta_parts.append(sort_order)
if scope_id and isinstance(scope_id, str):
meta_parts.append(f"scope:{scope_id[:8]}")
if meta_parts:
text.append(f" ({', '.join(meta_parts)})", style="dim")
if status == "completed" and isinstance(result, dict):
if "error" in result:
text.append(f" error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
else:
entries = result.get("entries", [])
page_info = result.get("page_info") or {}
has_more = (
bool(page_info.get("has_next_page")) if isinstance(page_info, dict) else False
)
count_suffix = "+" if has_more else ""
text.append(f" [{len(entries)}{count_suffix} found]", style="dim")
if entries and isinstance(entries, list):
text.append("\n")
for i, entry in enumerate(entries[:MAX_REQUESTS_DISPLAY]):
if not isinstance(entry, dict):
continue
req = entry.get("request") or {}
resp = entry.get("response") or {}
method = req.get("method", "?") if isinstance(req, dict) else "?"
host = req.get("host", "") if isinstance(req, dict) else ""
path = req.get("path", "/") if isinstance(req, dict) else "/"
code = resp.get("status_code") if isinstance(resp, dict) else None
text.append(" ")
text.append(f"{method:6}", style="#a78bfa")
text.append(f" {_truncate(host + path, 180)}", style="dim")
if code:
text.append(f" {code}", style=_status_style(code))
if i < min(len(entries), MAX_REQUESTS_DISPLAY) - 1:
text.append("\n")
if len(entries) > MAX_REQUESTS_DISPLAY:
text.append("\n")
text.append(
f" ... +{len(entries) - MAX_REQUESTS_DISPLAY} more",
style="dim italic",
)
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class ViewRequestRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "view_request"
css_classes: ClassVar[list[str]] = ["tool-call", "proxy-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "running")
request_id = args.get("request_id", "")
part = args.get("part", "request")
search_pattern = args.get("search_pattern")
text = Text()
text.append(PROXY_ICON, style="dim")
action = "searching" if search_pattern else "viewing"
text.append(f" {action} {part}", style="#06b6d4")
if request_id:
text.append(f" #{request_id}", style="dim")
if search_pattern:
text.append(f" /{_truncate(search_pattern, 100)}/", style="dim italic")
if status == "completed" and isinstance(result, dict):
if "error" in result:
text.append(f" error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
elif "hits" in result:
hits = result.get("hits", [])
total = result.get("total_hits", len(hits))
text.append(f" [{total} matches]", style="dim")
if hits and isinstance(hits, list):
text.append("\n")
for i, m in enumerate(hits[:5]):
if not isinstance(m, dict):
continue
before = m.get("before", "") or ""
match_text = m.get("match", "") or ""
after = m.get("after", "") or ""
before = before.replace("\n", " ").replace("\r", "")[-100:]
after = after.replace("\n", " ").replace("\r", "")[:100]
text.append(" ")
if before:
text.append(f"...{before}", style="dim")
text.append(match_text, style="#22c55e bold")
if after:
text.append(f"{after}...", style="dim")
if i < min(len(hits), 5) - 1:
text.append("\n")
if len(hits) > 5:
text.append("\n")
text.append(f" ... +{len(hits) - 5} more matches", style="dim italic")
elif "content" in result:
page = result.get("page", 1)
total_lines = result.get("total_lines", 0)
has_more = result.get("has_more", False)
content = result.get("content", "")
text.append(f" [page {page}, {total_lines} lines]", style="dim")
if content and isinstance(content, str):
lines = content.split("\n")[:15]
text.append("\n")
for i, line in enumerate(lines):
text.append(" ")
text.append(_truncate(line, MAX_LINE_LENGTH), style="dim")
if i < len(lines) - 1:
text.append("\n")
if has_more or len(lines) > 15:
text.append("\n")
text.append(" ... more content available", style="dim italic")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class RepeatRequestRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "repeat_request"
css_classes: ClassVar[list[str]] = ["tool-call", "proxy-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "running")
request_id = args.get("request_id", "")
modifications = args.get("modifications")
text = Text()
text.append(PROXY_ICON, style="dim")
text.append(" repeating request", style="#06b6d4")
if request_id:
text.append(f" #{request_id}", style="dim")
if modifications and isinstance(modifications, dict):
text.append("\n modifications:", style="dim italic")
if "url" in modifications:
text.append("\n")
text.append(" >> ", style="#3b82f6")
text.append(f"url: {_truncate(str(modifications['url']), 180)}", style="dim")
if "headers" in modifications and isinstance(modifications["headers"], dict):
for k, v in list(modifications["headers"].items())[:5]:
text.append("\n")
text.append(" >> ", style="#3b82f6")
text.append(f"{k}: {_sanitize(str(v), 150)}", style="dim")
if "cookies" in modifications and isinstance(modifications["cookies"], dict):
for k, v in list(modifications["cookies"].items())[:5]:
text.append("\n")
text.append(" >> ", style="#3b82f6")
text.append(f"cookie {k}={_sanitize(str(v), 100)}", style="dim")
if "params" in modifications and isinstance(modifications["params"], dict):
for k, v in list(modifications["params"].items())[:5]:
text.append("\n")
text.append(" >> ", style="#3b82f6")
text.append(f"param {k}={_sanitize(str(v), 100)}", style="dim")
if "body" in modifications and isinstance(modifications["body"], str):
text.append("\n")
text.append(" >> ", style="#3b82f6")
body_lines = modifications["body"].split("\n")[:4]
for i, line in enumerate(body_lines):
if i > 0:
text.append("\n")
text.append(" ", style="dim")
text.append(_truncate(line, MAX_LINE_LENGTH), style="dim")
if len(modifications["body"].split("\n")) > 4:
text.append(" ...", style="dim italic")
elif modifications and isinstance(modifications, str):
text.append(f"\n {_truncate(modifications, 200)}", style="dim italic")
if status == "completed" and isinstance(result, dict):
if not result.get("success", True) and result.get("error"):
text.append(f"\n error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
else:
elapsed_ms = result.get("elapsed_ms")
response = result.get("response") or {}
code = response.get("status_code") if isinstance(response, dict) else None
body = response.get("body", "") if isinstance(response, dict) else ""
body_truncated = (
bool(response.get("body_truncated")) if isinstance(response, dict) else False
)
text.append("\n")
text.append(" << ", style="#22c55e")
if code:
text.append(f"{code}", style=_status_style(code))
else:
text.append("(no response)", style="dim")
if elapsed_ms:
text.append(f" ({elapsed_ms}ms)", style="dim")
if body and isinstance(body, str):
lines = body.split("\n")[:5]
for line in lines:
text.append("\n")
text.append(" << ", style="#22c55e")
text.append(_truncate(line, MAX_LINE_LENGTH - 5), style="dim")
if body_truncated or len(body.split("\n")) > 5:
text.append("\n")
text.append(" ...", style="dim italic")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class ListSitemapRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "list_sitemap"
css_classes: ClassVar[list[str]] = ["tool-call", "proxy-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "running")
parent_id = args.get("parent_id")
scope_id = args.get("scope_id")
depth = args.get("depth")
text = Text()
text.append(PROXY_ICON, style="dim")
text.append(" listing sitemap", style="#06b6d4")
if parent_id:
text.append(f" under #{_truncate(str(parent_id), 20)}", style="dim")
meta_parts = []
if scope_id and isinstance(scope_id, str):
meta_parts.append(f"scope:{scope_id[:8]}")
if depth and depth != "DIRECT":
meta_parts.append(depth.lower())
if meta_parts:
text.append(f" ({', '.join(meta_parts)})", style="dim")
if status == "completed" and isinstance(result, dict):
if "error" in result:
text.append(f" error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
else:
total = result.get("total_count", 0)
entries = result.get("entries", [])
text.append(f" [{total} entries]", style="dim")
if entries and isinstance(entries, list):
text.append("\n")
for i, entry in enumerate(entries[:MAX_REQUESTS_DISPLAY]):
if not isinstance(entry, dict):
continue
kind = entry.get("kind") or "?"
label = entry.get("label") or "?"
has_children = entry.get("has_descendants", False)
req = entry.get("request") or {}
kind_style = {
"DOMAIN": "#f59e0b",
"DIRECTORY": "#3b82f6",
"REQUEST": "#22c55e",
}.get(kind, "dim")
text.append(" ")
kind_abbr = kind[:3] if isinstance(kind, str) else "?"
text.append(f"{kind_abbr:3}", style=kind_style)
text.append(f" {_truncate(label, 150)}", style="dim")
if req:
method = req.get("method", "")
code = req.get("status_code")
if method:
text.append(f" {method}", style="#a78bfa")
if code:
text.append(f" {code}", style=_status_style(code))
if has_children:
text.append(" +", style="dim italic")
if i < min(len(entries), MAX_REQUESTS_DISPLAY) - 1:
text.append("\n")
if len(entries) > MAX_REQUESTS_DISPLAY:
text.append("\n")
text.append(
f" ... +{len(entries) - MAX_REQUESTS_DISPLAY} more", style="dim italic"
)
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class ViewSitemapEntryRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "view_sitemap_entry"
css_classes: ClassVar[list[str]] = ["tool-call", "proxy-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "running")
entry_id = args.get("entry_id", "")
text = Text()
text.append(PROXY_ICON, style="dim")
text.append(" viewing sitemap", style="#06b6d4")
if entry_id:
text.append(f" #{_truncate(str(entry_id), 20)}", style="dim")
if status == "completed" and isinstance(result, dict):
if "error" in result:
text.append(f" error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
elif "entry" in result:
entry = result.get("entry") or {}
if not isinstance(entry, dict):
entry = {}
kind = entry.get("kind", "")
label = entry.get("label", "")
related = entry.get("related_requests") or {}
related_reqs = related.get("requests", []) if isinstance(related, dict) else []
total_related = related.get("total_count", 0) if isinstance(related, dict) else 0
if kind and label:
text.append(f" {kind}: {_truncate(label, 120)}", style="dim")
if total_related:
text.append(f" [{total_related} requests]", style="dim")
if related_reqs and isinstance(related_reqs, list):
text.append("\n")
for i, req in enumerate(related_reqs[:10]):
if not isinstance(req, dict):
continue
method = req.get("method", "?")
path = req.get("path", "/")
code = req.get("status_code")
text.append(" ")
text.append(f"{method:6}", style="#a78bfa")
text.append(f" {_truncate(path, 180)}", style="dim")
if code:
text.append(f" {code}", style=_status_style(code))
if i < min(len(related_reqs), 10) - 1:
text.append("\n")
if len(related_reqs) > 10:
text.append("\n")
text.append(f" ... +{len(related_reqs) - 10} more", style="dim italic")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
@register_tool_renderer
class ScopeRulesRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "scope_rules"
css_classes: ClassVar[list[str]] = ["tool-call", "proxy-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result")
status = tool_data.get("status", "running")
action = args.get("action", "")
scope_name = args.get("scope_name", "")
scope_id = args.get("scope_id", "")
allowlist = args.get("allowlist")
denylist = args.get("denylist")
text = Text()
text.append(PROXY_ICON, style="dim")
action_map = {
"get": "getting",
"list": "listing",
"create": "creating",
"update": "updating",
"delete": "deleting",
}
action_text = action_map.get(action, action + "ing" if action else "managing")
text.append(f" {action_text} proxy scope", style="#06b6d4")
if scope_name:
text.append(f" '{_truncate(scope_name, 50)}'", style="dim italic")
if scope_id and isinstance(scope_id, str):
text.append(f" #{scope_id[:8]}", style="dim")
if allowlist and isinstance(allowlist, list):
allow_str = ", ".join(_truncate(str(a), 40) for a in allowlist[:4])
text.append(f"\n allow: {allow_str}", style="dim")
if len(allowlist) > 4:
text.append(f" +{len(allowlist) - 4}", style="dim italic")
if denylist and isinstance(denylist, list):
deny_str = ", ".join(_truncate(str(d), 40) for d in denylist[:4])
text.append(f"\n deny: {deny_str}", style="dim")
if len(denylist) > 4:
text.append(f" +{len(denylist) - 4}", style="dim italic")
if status == "completed" and isinstance(result, dict):
if "error" in result:
text.append(f" error: {_sanitize(str(result['error']), 150)}", style="#ef4444")
elif "scopes" in result:
scopes = result.get("scopes", [])
text.append(f" [{len(scopes)} scopes]", style="dim")
if scopes and isinstance(scopes, list):
text.append("\n")
for i, scope in enumerate(scopes[:5]):
if not isinstance(scope, dict):
continue
name = scope.get("name", "?")
allow = scope.get("allowlist") or []
text.append(" ")
text.append(_truncate(str(name), 40), style="#22c55e")
if allow and isinstance(allow, list):
allow_str = ", ".join(_truncate(str(a), 30) for a in allow[:3])
text.append(f" {allow_str}", style="dim")
if len(allow) > 3:
text.append(f" +{len(allow) - 3}", style="dim italic")
if i < min(len(scopes), 5) - 1:
text.append("\n")
elif "scope" in result:
scope = result.get("scope") or {}
if isinstance(scope, dict):
allow = scope.get("allowlist") or []
deny = scope.get("denylist") or []
if allow and isinstance(allow, list):
allow_str = ", ".join(_truncate(str(a), 40) for a in allow[:5])
text.append(f"\n allow: {allow_str}", style="dim")
if deny and isinstance(deny, list):
deny_str = ", ".join(_truncate(str(d), 40) for d in deny[:5])
text.append(f"\n deny: {deny_str}", style="dim")
elif "message" in result:
text.append(f" {result['message']}", style="#22c55e")
css_classes = cls.get_css_classes(status)
return Static(text, classes=css_classes)
+71
View File
@@ -0,0 +1,71 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
class ToolTUIRegistry:
_renderers: ClassVar[dict[str, type[BaseToolRenderer]]] = {}
@classmethod
def register(cls, renderer_class: type[BaseToolRenderer]) -> None:
if not renderer_class.tool_name:
raise ValueError(f"Renderer {renderer_class.__name__} must define tool_name")
cls._renderers[renderer_class.tool_name] = renderer_class
@classmethod
def get_renderer(cls, tool_name: str) -> type[BaseToolRenderer] | None:
return cls._renderers.get(tool_name)
def register_tool_renderer(renderer_class: type[BaseToolRenderer]) -> type[BaseToolRenderer]:
ToolTUIRegistry.register(renderer_class)
return renderer_class
def get_tool_renderer(tool_name: str) -> type[BaseToolRenderer] | None:
return ToolTUIRegistry.get_renderer(tool_name)
def render_tool_widget(tool_data: dict[str, Any]) -> Static:
tool_name = tool_data.get("tool_name", "")
renderer = get_tool_renderer(tool_name)
if renderer:
return renderer.render(tool_data)
return _render_default_tool_widget(tool_data)
def _render_default_tool_widget(tool_data: dict[str, Any]) -> Static:
tool_name = tool_data.get("tool_name", "Unknown Tool")
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
result = tool_data.get("result")
text = Text()
text.append("→ Using tool ", style="dim")
text.append(tool_name, style="bold blue")
text.append("\n")
for k, v in list(args.items()):
str_v = str(v)
text.append(" ")
text.append(k, style="dim")
text.append(": ")
text.append(str_v)
text.append("\n")
if status in ["completed", "failed", "error"] and result is not None:
result_str = str(result)
text.append("Result: ", style="bold")
text.append(result_str)
else:
icon, color = BaseToolRenderer.status_icon(status)
text.append(icon, style=color)
css_classes = BaseToolRenderer.get_css_classes(status)
return Static(text, classes=css_classes)
@@ -0,0 +1,258 @@
from functools import cache
from typing import Any, ClassVar
from pygments.lexers import PythonLexer
from pygments.styles import get_style_by_name
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
def _coerce_dict(value: Any) -> dict[str, Any]:
if isinstance(value, dict):
return value
return {}
def _coerce_list_of_dicts(value: Any) -> list[dict[str, Any]]:
if isinstance(value, list):
return [item for item in value if isinstance(item, dict)]
return []
@cache
def _get_style_colors() -> dict[Any, str]:
style = get_style_by_name("native")
return {token: f"#{style_def['color']}" for token, style_def in style if style_def["color"]}
FIELD_STYLE = "bold #4ade80"
DIM_STYLE = "dim"
FILE_STYLE = "bold #60a5fa"
LINE_STYLE = "#facc15"
LABEL_STYLE = "italic #a1a1aa"
CODE_STYLE = "#e2e8f0"
BEFORE_STYLE = "#ef4444"
AFTER_STYLE = "#22c55e"
@register_tool_renderer
class CreateVulnerabilityReportRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_vulnerability_report"
css_classes: ClassVar[list[str]] = ["tool-call", "reporting-tool"]
SEVERITY_COLORS: ClassVar[dict[str, str]] = {
"critical": "#dc2626",
"high": "#ea580c",
"medium": "#d97706",
"low": "#65a30d",
"info": "#0284c7",
}
@classmethod
def _get_token_color(cls, token_type: Any) -> str | None:
colors = _get_style_colors()
while token_type:
if token_type in colors:
return colors[token_type]
token_type = token_type.parent
return None
@classmethod
def _highlight_python(cls, code: str) -> Text:
lexer = PythonLexer()
text = Text()
for token_type, token_value in lexer.get_tokens(code):
if not token_value:
continue
color = cls._get_token_color(token_type)
text.append(token_value, style=color)
return text
@classmethod
def _get_cvss_color(cls, cvss_score: float) -> str:
if cvss_score >= 9.0:
return "#dc2626"
if cvss_score >= 7.0:
return "#ea580c"
if cvss_score >= 4.0:
return "#d97706"
if cvss_score >= 0.1:
return "#65a30d"
return "#6b7280"
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static: # noqa: PLR0912, PLR0915
args = tool_data.get("args", {})
result = tool_data.get("result", {})
title = args.get("title", "")
description = args.get("description", "")
impact = args.get("impact", "")
target = args.get("target", "")
technical_analysis = args.get("technical_analysis", "")
poc_description = args.get("poc_description", "")
poc_script_code = args.get("poc_script_code", "")
remediation_steps = args.get("remediation_steps", "")
cvss_breakdown = _coerce_dict(args.get("cvss_breakdown"))
code_locations = _coerce_list_of_dicts(args.get("code_locations"))
endpoint = args.get("endpoint", "")
method = args.get("method", "")
cve = args.get("cve", "")
cwe = args.get("cwe", "")
severity = ""
cvss_score = None
if isinstance(result, dict):
severity = result.get("severity", "")
cvss_score = result.get("cvss_score")
text = Text()
text.append("🐞 ")
text.append("Vulnerability Report", style="bold #ea580c")
if title:
text.append("\n\n")
text.append("Title: ", style=FIELD_STYLE)
text.append(title)
if severity:
text.append("\n\n")
text.append("Severity: ", style=FIELD_STYLE)
severity_color = cls.SEVERITY_COLORS.get(severity.lower(), "#6b7280")
text.append(severity.upper(), style=f"bold {severity_color}")
if cvss_score is not None:
text.append("\n\n")
text.append("CVSS Score: ", style=FIELD_STYLE)
cvss_color = cls._get_cvss_color(cvss_score)
text.append(str(cvss_score), style=f"bold {cvss_color}")
if target:
text.append("\n\n")
text.append("Target: ", style=FIELD_STYLE)
text.append(target)
if endpoint:
text.append("\n\n")
text.append("Endpoint: ", style=FIELD_STYLE)
text.append(endpoint)
if method:
text.append("\n\n")
text.append("Method: ", style=FIELD_STYLE)
text.append(method)
if cve:
text.append("\n\n")
text.append("CVE: ", style=FIELD_STYLE)
text.append(cve)
if cwe:
text.append("\n\n")
text.append("CWE: ", style=FIELD_STYLE)
text.append(cwe)
if cvss_breakdown:
text.append("\n\n")
cvss_parts = []
for key, prefix in [
("attack_vector", "AV"),
("attack_complexity", "AC"),
("privileges_required", "PR"),
("user_interaction", "UI"),
("scope", "S"),
("confidentiality", "C"),
("integrity", "I"),
("availability", "A"),
]:
val = cvss_breakdown.get(key)
if val:
cvss_parts.append(f"{prefix}:{val}")
text.append("CVSS Vector: ", style=FIELD_STYLE)
text.append("/".join(cvss_parts), style=DIM_STYLE)
if description:
text.append("\n\n")
text.append("Description", style=FIELD_STYLE)
text.append("\n")
text.append(description)
if impact:
text.append("\n\n")
text.append("Impact", style=FIELD_STYLE)
text.append("\n")
text.append(impact)
if technical_analysis:
text.append("\n\n")
text.append("Technical Analysis", style=FIELD_STYLE)
text.append("\n")
text.append(technical_analysis)
if code_locations:
text.append("\n\n")
text.append("Code Locations", style=FIELD_STYLE)
for i, loc in enumerate(code_locations):
text.append("\n\n")
text.append(f" Location {i + 1}: ", style=DIM_STYLE)
text.append(loc.get("file", "unknown"), style=FILE_STYLE)
start = loc.get("start_line")
end = loc.get("end_line")
if start is not None:
if end and end != start:
text.append(f":{start}-{end}", style=LINE_STYLE)
else:
text.append(f":{start}", style=LINE_STYLE)
if loc.get("label"):
text.append(f"\n {loc['label']}", style=LABEL_STYLE)
if loc.get("snippet"):
text.append("\n ")
text.append(loc["snippet"], style=CODE_STYLE)
if loc.get("fix_before") or loc.get("fix_after"):
text.append("\n ")
text.append("Fix:", style=DIM_STYLE)
if loc.get("fix_before"):
text.append("\n ")
text.append("- ", style=BEFORE_STYLE)
text.append(loc["fix_before"], style=BEFORE_STYLE)
if loc.get("fix_after"):
text.append("\n ")
text.append("+ ", style=AFTER_STYLE)
text.append(loc["fix_after"], style=AFTER_STYLE)
if poc_description:
text.append("\n\n")
text.append("PoC Description", style=FIELD_STYLE)
text.append("\n")
text.append(poc_description)
if poc_script_code:
text.append("\n\n")
text.append("PoC Code", style=FIELD_STYLE)
text.append("\n")
text.append_text(cls._highlight_python(poc_script_code))
if remediation_steps:
text.append("\n\n")
text.append("Remediation", style=FIELD_STYLE)
text.append("\n")
text.append(remediation_steps)
if not title:
text.append("\n ")
text.append("Creating report...", style="dim")
padded = Text()
padded.append("\n\n")
padded.append_text(text)
padded.append("\n\n")
css_classes = cls.get_css_classes("completed")
return Static(padded, classes=css_classes)
@@ -0,0 +1,266 @@
import re
from functools import cache
from typing import Any, ClassVar
from pygments.lexers import get_lexer_by_name
from pygments.styles import get_style_by_name
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
MAX_OUTPUT_LINES = 50
MAX_LINE_LENGTH = 200
STRIP_PATTERNS = [
r"^Chunk ID: [0-9a-f]+\s*$",
r"^Wall time: [\d.]+ seconds\s*$",
r"^Process exited with code -?\d+\s*$",
r"^Process running with session ID \d+\s*$",
r"^Original token count: \d+\s*$",
]
_EXIT_RE = re.compile(r"Process exited with code (-?\d+)")
_SESSION_RE = re.compile(r"Process running with session ID (\d+)")
_OUTPUT_HEADER = "\nOutput:\n"
_CONTROL_BYTES_TO_DROP = dict.fromkeys(
[b for b in range(0x20) if b not in (0x09, 0x0A)] + [0x7F],
None,
)
@cache
def _get_style_colors() -> dict[Any, str]:
style = get_style_by_name("native")
return {token: f"#{style_def['color']}" for token, style_def in style if style_def["color"]}
def _parse_sdk_shell_result(result: Any) -> dict[str, Any]:
"""Translate the SDK's terminal-output string into the dict shape the
renderer's `_append_output` helper expects.
The SDK returns a header-prefixed string ending with `Output:\\n<actual>`.
We extract `content`, `exit_code`, and `session_id`; anything else (or a
non-string result) flows through unchanged so renderers can handle errors.
"""
if isinstance(result, dict):
return result
if not isinstance(result, str):
return {"content": "" if result is None else str(result)}
exit_match = _EXIT_RE.search(result)
session_match = _SESSION_RE.search(result)
idx = result.find(_OUTPUT_HEADER)
content = result[idx + len(_OUTPUT_HEADER) :] if idx >= 0 else result
parsed: dict[str, Any] = {"content": content}
if exit_match:
parsed["exit_code"] = int(exit_match.group(1))
if session_match:
parsed["session_id"] = int(session_match.group(1))
return parsed
def _truncate_line(line: str) -> str:
if len(line) > MAX_LINE_LENGTH:
return line[: MAX_LINE_LENGTH - 3] + "..."
return line
def _clean_output(output: str) -> str:
cleaned = Text.from_ansi(output).plain.translate(_CONTROL_BYTES_TO_DROP)
for pattern in STRIP_PATTERNS:
cleaned = re.sub(pattern, "", cleaned, flags=re.MULTILINE)
if cleaned.strip():
lines = cleaned.splitlines()
filtered_lines: list[str] = []
for line in lines:
if not filtered_lines and not line.strip():
continue
if line.strip() == "Output:":
continue
filtered_lines.append(line)
while filtered_lines and not filtered_lines[-1].strip():
filtered_lines.pop()
cleaned = "\n".join(filtered_lines)
return cleaned.strip()
def _format_output(output: str) -> Text:
text = Text()
lines = output.splitlines()
total_lines = len(lines)
head_count = MAX_OUTPUT_LINES // 2
tail_count = MAX_OUTPUT_LINES - head_count - 1
if total_lines <= MAX_OUTPUT_LINES:
display_lines = lines
truncated = False
hidden_count = 0
else:
display_lines = lines[:head_count]
truncated = True
hidden_count = total_lines - head_count - tail_count
for i, line in enumerate(display_lines):
text.append(" ")
text.append(_truncate_line(line), style="dim")
if i < len(display_lines) - 1 or truncated:
text.append("\n")
if truncated:
text.append(f" ... {hidden_count} lines truncated ...", style="dim italic")
text.append("\n")
tail_lines = lines[-tail_count:]
for i, line in enumerate(tail_lines):
text.append(" ")
text.append(_truncate_line(line), style="dim")
if i < len(tail_lines) - 1:
text.append("\n")
return text
def _get_token_color(token_type: Any) -> str | None:
colors = _get_style_colors()
while token_type:
if token_type in colors:
return colors[token_type]
token_type = token_type.parent
return None
def _highlight_bash(code: str) -> Text:
lexer = get_lexer_by_name("bash")
text = Text()
for token_type, token_value in lexer.get_tokens(code):
if not token_value:
continue
color = _get_token_color(token_type)
text.append(token_value, style=color)
return text
def _append_output(text: Text, parsed: dict[str, Any], tool_status: str) -> None:
raw_output = parsed.get("content", "") or ""
output = _clean_output(raw_output) if isinstance(raw_output, str) else ""
exit_code = parsed.get("exit_code")
if tool_status == "running":
if output:
text.append("\n")
text.append_text(_format_output(output))
return
if not output:
if exit_code is not None and exit_code != 0:
text.append("\n")
text.append(f" exit {exit_code}", style="dim #ef4444")
return
text.append("\n")
text.append_text(_format_output(output))
if exit_code is not None and exit_code != 0:
text.append("\n")
text.append(f" exit {exit_code}", style="dim #ef4444")
def _build_terminal_content(
*,
prompt: str,
prompt_style: str,
command: str,
parsed_result: dict[str, Any] | None,
tool_status: str,
meta: str | None = None,
) -> Text:
text = Text()
text.append(">_", style="dim")
text.append(" ")
if not command.strip():
text.append("getting logs...", style="dim")
else:
text.append(prompt, style=prompt_style)
text.append(" ")
text.append_text(_highlight_bash(command))
if meta:
text.append(f" {meta}", style="dim")
if parsed_result is not None:
_append_output(text, parsed_result, tool_status)
return text
@register_tool_renderer
class ExecCommandRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "exec_command"
css_classes: ClassVar[list[str]] = ["tool-call", "terminal-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
result = tool_data.get("result")
cmd = str(args.get("cmd", ""))
workdir = args.get("workdir")
tty = bool(args.get("tty"))
meta_parts: list[str] = []
if workdir:
meta_parts.append(f"cwd:{workdir}")
if tty:
meta_parts.append("tty")
meta = ", ".join(meta_parts) if meta_parts else None
parsed = _parse_sdk_shell_result(result) if result is not None else None
content = _build_terminal_content(
prompt="$",
prompt_style="#22c55e",
command=cmd,
parsed_result=parsed,
tool_status=status,
meta=meta,
)
return Static(content, classes=cls.get_css_classes(status))
@register_tool_renderer
class WriteStdinRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "write_stdin"
css_classes: ClassVar[list[str]] = ["tool-call", "terminal-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
status = tool_data.get("status", "unknown")
result = tool_data.get("result")
chars = str(args.get("chars", ""))
session_id = args.get("session_id")
meta = f"session #{session_id}" if session_id is not None else None
parsed = _parse_sdk_shell_result(result) if result is not None else None
content = _build_terminal_content(
prompt=">>>",
prompt_style="#3b82f6",
command=chars,
parsed_result=parsed,
tool_status=status,
meta=meta,
)
return Static(content, classes=cls.get_css_classes(status))
@@ -0,0 +1,31 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
@register_tool_renderer
class ThinkRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "think"
css_classes: ClassVar[list[str]] = ["tool-call", "thinking-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
thought = args.get("thought", "")
text = Text()
text.append("🧠 ")
text.append("Thinking", style="bold #a855f7")
text.append("\n ")
if thought:
text.append(thought, style="italic dim")
else:
text.append("Thinking...", style="italic dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@@ -0,0 +1,225 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
STATUS_MARKERS: dict[str, str] = {
"pending": "[ ]",
"in_progress": "[~]",
"done": "[•]",
}
def _format_todo_lines(text: Text, result: dict[str, Any]) -> None:
todos = result.get("todos")
if not isinstance(todos, list) or not todos:
text.append("\n ")
text.append("No todos", style="dim")
return
for todo in todos:
status = todo.get("status", "pending")
marker = STATUS_MARKERS.get(status, STATUS_MARKERS["pending"])
title = todo.get("title", "").strip() or "(untitled)"
text.append("\n ")
text.append(marker)
text.append(" ")
if status == "done":
text.append(title, style="dim strike")
elif status == "in_progress":
text.append(title, style="italic")
else:
text.append(title)
@register_tool_renderer
class CreateTodoRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "create_todo"
css_classes: ClassVar[list[str]] = ["tool-call", "todo-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("📋 ")
text.append("Todo", style="bold #a78bfa")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif result and isinstance(result, dict):
if result.get("success"):
_format_todo_lines(text, result)
else:
error = result.get("error", "Failed to create todo")
text.append("\n ")
text.append(error, style="#ef4444")
else:
text.append("\n ")
text.append("Creating...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class ListTodosRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "list_todos"
css_classes: ClassVar[list[str]] = ["tool-call", "todo-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("📋 ")
text.append("Todos", style="bold #a78bfa")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif result and isinstance(result, dict):
if result.get("success"):
_format_todo_lines(text, result)
else:
error = result.get("error", "Unable to list todos")
text.append("\n ")
text.append(error, style="#ef4444")
else:
text.append("\n ")
text.append("Loading...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class UpdateTodoRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "update_todo"
css_classes: ClassVar[list[str]] = ["tool-call", "todo-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("📋 ")
text.append("Todo Updated", style="bold #a78bfa")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif result and isinstance(result, dict):
if result.get("success"):
_format_todo_lines(text, result)
else:
error = result.get("error", "Failed to update todo")
text.append("\n ")
text.append(error, style="#ef4444")
else:
text.append("\n ")
text.append("Updating...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class MarkTodoDoneRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "mark_todo_done"
css_classes: ClassVar[list[str]] = ["tool-call", "todo-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("📋 ")
text.append("Todo Completed", style="bold #a78bfa")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif result and isinstance(result, dict):
if result.get("success"):
_format_todo_lines(text, result)
else:
error = result.get("error", "Failed to mark todo done")
text.append("\n ")
text.append(error, style="#ef4444")
else:
text.append("\n ")
text.append("Marking done...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class MarkTodoPendingRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "mark_todo_pending"
css_classes: ClassVar[list[str]] = ["tool-call", "todo-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("📋 ")
text.append("Todo Reopened", style="bold #f59e0b")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif result and isinstance(result, dict):
if result.get("success"):
_format_todo_lines(text, result)
else:
error = result.get("error", "Failed to reopen todo")
text.append("\n ")
text.append(error, style="#ef4444")
else:
text.append("\n ")
text.append("Reopening...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@register_tool_renderer
class DeleteTodoRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "delete_todo"
css_classes: ClassVar[list[str]] = ["tool-call", "todo-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
result = tool_data.get("result")
text = Text()
text.append("📋 ")
text.append("Todo Removed", style="bold #94a3b8")
if isinstance(result, str) and result.strip():
text.append("\n ")
text.append(result.strip(), style="dim")
elif result and isinstance(result, dict):
if result.get("success"):
_format_todo_lines(text, result)
else:
error = result.get("error", "Failed to remove todo")
text.append("\n ")
text.append(error, style="#ef4444")
else:
text.append("\n ")
text.append("Removing...", style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
@@ -0,0 +1,29 @@
from rich.text import Text
class UserMessageRenderer:
@classmethod
def render_simple(cls, content: str) -> Text:
if not content:
return Text()
return cls._format_user_message(content)
@classmethod
def _format_user_message(cls, content: str) -> Text:
text = Text()
text.append("", style="#3b82f6")
text.append(" ")
text.append("You:", style="bold")
text.append("\n")
lines = content.split("\n")
for i, line in enumerate(lines):
if i > 0:
text.append("\n")
text.append("", style="#3b82f6")
text.append(" ")
text.append(line)
return text
@@ -0,0 +1,29 @@
from typing import Any, ClassVar
from rich.text import Text
from textual.widgets import Static
from .base_renderer import BaseToolRenderer
from .registry import register_tool_renderer
@register_tool_renderer
class WebSearchRenderer(BaseToolRenderer):
tool_name: ClassVar[str] = "web_search"
css_classes: ClassVar[list[str]] = ["tool-call", "web-search-tool"]
@classmethod
def render(cls, tool_data: dict[str, Any]) -> Static:
args = tool_data.get("args", {})
query = args.get("query", "")
text = Text()
text.append("🌐 ")
text.append("Searching the web...", style="bold #60a5fa")
if query:
text.append("\n ")
text.append(query, style="dim")
css_classes = cls.get_css_classes("completed")
return Static(text, classes=css_classes)
File diff suppressed because it is too large Load Diff