mirror of
https://github.com/usestrix/strix.git
synced 2026-08-18 17:52:32 +02:00
Real-time display panel for agent stats (#134)
Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>
This commit is contained in:
committed by
GitHub
co-authored by
Ahmed Allam
parent
ea0b457e6a
commit
976faaaebd
+70
-11
@@ -1,9 +1,12 @@
|
||||
import atexit
|
||||
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
|
||||
|
||||
@@ -11,7 +14,7 @@ from strix.agents.StrixAgent import StrixAgent
|
||||
from strix.llm.config import LLMConfig
|
||||
from strix.telemetry.tracer import Tracer, set_global_tracer
|
||||
|
||||
from .utils import get_severity_color
|
||||
from .utils import build_final_stats_text, build_live_stats_text, get_severity_color
|
||||
|
||||
|
||||
async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
@@ -130,24 +133,80 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
|
||||
set_global_tracer(tracer)
|
||||
|
||||
def create_live_status() -> Panel:
|
||||
status_text = Text()
|
||||
status_text.append("🦉 ", style="bold white")
|
||||
status_text.append("Running penetration test...", style="bold #22c55e")
|
||||
status_text.append("\n\n")
|
||||
|
||||
stats_text = build_live_stats_text(tracer)
|
||||
if stats_text:
|
||||
status_text.append(stats_text)
|
||||
|
||||
return Panel(
|
||||
status_text,
|
||||
title="[bold #22c55e]🔍 Live Penetration Test Status",
|
||||
title_align="center",
|
||||
border_style="#22c55e",
|
||||
padding=(1, 2),
|
||||
)
|
||||
|
||||
try:
|
||||
console.print()
|
||||
with console.status("[bold cyan]Running penetration test...", spinner="dots") as status:
|
||||
agent = StrixAgent(agent_config)
|
||||
result = await agent.execute_scan(scan_config)
|
||||
status.stop()
|
||||
|
||||
if isinstance(result, dict) and not result.get("success", True):
|
||||
error_msg = result.get("error", "Unknown error")
|
||||
console.print()
|
||||
console.print(f"[bold red]❌ Penetration test failed:[/] {error_msg}")
|
||||
console.print()
|
||||
sys.exit(1)
|
||||
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: # noqa: BLE001
|
||||
break
|
||||
|
||||
update_thread = threading.Thread(target=update_status, daemon=True)
|
||||
update_thread.start()
|
||||
|
||||
try:
|
||||
agent = StrixAgent(agent_config)
|
||||
result = await agent.execute_scan(scan_config)
|
||||
|
||||
if isinstance(result, dict) and not result.get("success", True):
|
||||
error_msg = result.get("error", "Unknown error")
|
||||
console.print()
|
||||
console.print(f"[bold red]❌ Penetration test failed:[/] {error_msg}")
|
||||
console.print()
|
||||
sys.exit(1)
|
||||
finally:
|
||||
stop_updates.set()
|
||||
update_thread.join(timeout=1)
|
||||
|
||||
except Exception as e:
|
||||
console.print(f"[bold red]Error during penetration test:[/] {e}")
|
||||
raise
|
||||
|
||||
console.print()
|
||||
final_stats_text = Text()
|
||||
final_stats_text.append("📊 ", style="bold cyan")
|
||||
final_stats_text.append("PENETRATION TEST COMPLETED", style="bold green")
|
||||
final_stats_text.append("\n\n")
|
||||
|
||||
stats_text = build_final_stats_text(tracer)
|
||||
if stats_text:
|
||||
final_stats_text.append(stats_text)
|
||||
|
||||
final_stats_panel = Panel(
|
||||
final_stats_text,
|
||||
title="[bold green]✅ Final Statistics",
|
||||
title_align="center",
|
||||
border_style="green",
|
||||
padding=(1, 2),
|
||||
)
|
||||
console.print(final_stats_panel)
|
||||
|
||||
if tracer.final_scan_result:
|
||||
console.print()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user