mirror of
https://github.com/usestrix/strix.git
synced 2026-08-20 10:33:34 +02:00
Wire vulnerability amendment callbacks
This commit is contained in:
+30
-17
@@ -38,6 +38,35 @@ def _resolve_sandbox_image() -> str:
|
||||
return image
|
||||
|
||||
|
||||
def _configure_report_callbacks(report_state: ReportState, console: Console) -> None:
|
||||
def display_vulnerability(report: dict[str, Any], *, updated: bool = False) -> None:
|
||||
report_id = report.get("id", "unknown")
|
||||
|
||||
vuln_text = format_vulnerability_report(report)
|
||||
|
||||
title = (
|
||||
f"[bold yellow]{report_id.upper()} — UPDATED FINDING"
|
||||
if updated
|
||||
else f"[bold red]{report_id.upper()}"
|
||||
)
|
||||
vuln_panel = Panel(
|
||||
vuln_text,
|
||||
title=title,
|
||||
title_align="left",
|
||||
border_style="yellow" if updated else "red",
|
||||
padding=(1, 2),
|
||||
)
|
||||
|
||||
console.print(vuln_panel)
|
||||
console.print()
|
||||
|
||||
report_state.vulnerability_found_callback = display_vulnerability
|
||||
report_state.vulnerability_updated_callback = lambda report: display_vulnerability(
|
||||
report,
|
||||
updated=True,
|
||||
)
|
||||
|
||||
|
||||
async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
console = Console()
|
||||
|
||||
@@ -105,23 +134,7 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
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
|
||||
_configure_report_callbacks(report_state, console)
|
||||
|
||||
def cleanup_on_exit() -> None:
|
||||
report_state.cleanup()
|
||||
|
||||
@@ -102,6 +102,9 @@ class GoTuiRuntime:
|
||||
self.report_state.vulnerability_found_callback = lambda _report: (
|
||||
self.controller.notify_changed()
|
||||
)
|
||||
self.report_state.vulnerability_updated_callback = lambda _report: (
|
||||
self.controller.notify_changed()
|
||||
)
|
||||
self.controller.notify_changed()
|
||||
|
||||
async def start_from_setup(self, verify: bool = True) -> None:
|
||||
|
||||
@@ -441,7 +441,16 @@ async def _do_update( # noqa: PLR0911, PLR0912, PLR0915
|
||||
"endpoint",
|
||||
"method",
|
||||
}:
|
||||
updates[field] = str(value).strip()
|
||||
normalized_value = str(value).strip()
|
||||
if not normalized_value:
|
||||
errors.append(
|
||||
_REQUIRED_FIELDS.get(
|
||||
field,
|
||||
f"{field.replace('_', ' ').capitalize()} cannot be empty",
|
||||
),
|
||||
)
|
||||
else:
|
||||
updates[field] = normalized_value
|
||||
|
||||
if "fix_effort" in raw_updates:
|
||||
normalized_fix_effort = str(fix_effort).strip().lower()
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import Mock
|
||||
|
||||
from rich.console import Console
|
||||
from rich.panel import Panel
|
||||
|
||||
from strix.interface import cli
|
||||
|
||||
|
||||
def test_cli_report_callbacks_render_new_and_updated_findings() -> None:
|
||||
report_state = SimpleNamespace(
|
||||
vulnerability_found_callback=None,
|
||||
vulnerability_updated_callback=None,
|
||||
)
|
||||
console = Mock(spec=Console)
|
||||
|
||||
cli._configure_report_callbacks(cast("Any", report_state), console)
|
||||
|
||||
report = {"id": "vuln-0001", "title": "Unsafe redirect"}
|
||||
report_state.vulnerability_found_callback(report)
|
||||
report_state.vulnerability_updated_callback(report)
|
||||
|
||||
panels = [call.args[0] for call in console.print.call_args_list if call.args]
|
||||
assert all(isinstance(panel, Panel) for panel in panels)
|
||||
assert panels[0].title == "[bold red]VULN-0001"
|
||||
assert panels[1].title == "[bold yellow]VULN-0001 — UPDATED FINDING"
|
||||
@@ -12,6 +12,7 @@ import threading
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -95,6 +96,25 @@ def test_binary_command_ignores_unconstrained_path_sidecar(
|
||||
GoTuiRuntime.binary_command()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_init_run_state_wires_updated_report_callback(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
runtime = GoTuiRuntime(args())
|
||||
notify_changed = Mock()
|
||||
monkeypatch.setattr(runtime.controller, "notify_changed", notify_changed)
|
||||
|
||||
runtime.init_run_state()
|
||||
|
||||
assert runtime.report_state is not None
|
||||
assert runtime.report_state.vulnerability_updated_callback is not None
|
||||
notify_changed.reset_mock()
|
||||
runtime.report_state.vulnerability_updated_callback({"id": "vuln-0001"})
|
||||
notify_changed.assert_called_once_with()
|
||||
|
||||
|
||||
def test_child_environment_excludes_credentials(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("OPENAI_API_KEY", "openai-secret")
|
||||
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "aws-id")
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -107,6 +107,35 @@ async def test_update_requires_nonempty_reason() -> None:
|
||||
assert result == {"success": False, "error": "update_reason cannot be empty"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("field", "value", "expected_error"),
|
||||
[
|
||||
("impact", "", "Impact cannot be empty"),
|
||||
("impact", " ", "Impact cannot be empty"),
|
||||
("endpoint", "", "Endpoint cannot be empty"),
|
||||
("endpoint", " ", "Endpoint cannot be empty"),
|
||||
],
|
||||
)
|
||||
async def test_update_rejects_blank_text_fields(
|
||||
report_state: ReportState,
|
||||
field: str,
|
||||
value: str,
|
||||
expected_error: str,
|
||||
) -> None:
|
||||
original_value = report_state.vulnerability_reports[0][field]
|
||||
update = cast("dict[str, Any]", {field: value})
|
||||
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
update_reason="The source review supplied no content for this field.",
|
||||
**update,
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert expected_error in result["errors"]
|
||||
assert report_state.vulnerability_reports[0][field] == original_value
|
||||
|
||||
|
||||
async def test_update_changes_impact_only(report_state: ReportState) -> None:
|
||||
result = await _do_update(
|
||||
report_id="vuln-0001",
|
||||
|
||||
Reference in New Issue
Block a user