fix: add defensive guards per Copilot review

- Validate webhook URL scheme (http/https only)
- Guard against tracer=None in all formatters and helpers
- Truncate long strings for Slack (3000 char) and Discord (1024 char) limits
- Add _truncate helper with ellipsis
- Add 11 new tests for None tracer, URL validation, and truncation
This commit is contained in:
ST-2 2026-02-23 16:27:57 -08:00
parent 797dea8320
commit 369f2e6c35
2 changed files with 114 additions and 14 deletions

View file

@ -21,6 +21,18 @@ logger = logging.getLogger(__name__)
WEBHOOK_TIMEOUT = 10
# Platform limits for field truncation
_SLACK_SECTION_TEXT_LIMIT = 3000
_DISCORD_FIELD_VALUE_LIMIT = 1024
_DISCORD_TITLE_LIMIT = 256
def _truncate(text: str, limit: int) -> str:
"""Truncate *text* to *limit* characters, appending an ellipsis if trimmed."""
if len(text) <= limit:
return text
return text[: limit - 1] + "\u2026"
def send_completion_webhook(
webhook_url: str,
@ -36,6 +48,16 @@ def send_completion_webhook(
tracer: The global :class:`Tracer` instance containing scan results.
args: Parsed CLI arguments (used to extract target info and run name).
"""
# Validate URL scheme
parsed = urlparse(webhook_url)
if parsed.scheme not in ("http", "https"):
logger.warning("Invalid webhook URL scheme %r — skipping delivery", parsed.scheme)
return
if not tracer:
logger.warning("No tracer available — skipping webhook delivery")
return
resolved_format = _resolve_format(webhook_url, webhook_format)
formatters: dict[str, Any] = {
@ -92,6 +114,8 @@ def _targets_summary(args: argparse.Namespace) -> str:
def _vulnerability_summary(tracer: Any) -> list[dict[str, Any]]:
"""Return a lightweight list of vulnerability dicts safe for JSON serialisation."""
if not tracer:
return []
return [
{
"id": report.get("id", ""),
@ -108,6 +132,8 @@ def _vulnerability_summary(tracer: Any) -> list[dict[str, Any]]:
def _severity_counts(tracer: Any) -> dict[str, int]:
counts: dict[str, int] = {"critical": 0, "high": 0, "medium": 0, "low": 0, "info": 0}
if not tracer:
return counts
for report in tracer.vulnerability_reports:
severity = report.get("severity", "").lower()
if severity in counts:
@ -130,18 +156,19 @@ def _format_generic(tracer: Any, args: argparse.Namespace) -> dict[str, Any]:
"""Plain JSON payload with full scan data."""
completed = _scan_completed(tracer)
llm_stats = tracer.get_total_llm_stats()["total"] if tracer else {}
vuln_reports = tracer.vulnerability_reports if tracer else []
return {
"event": "scan_completed" if completed else "scan_ended",
"run_name": getattr(args, "run_name", ""),
"targets": _targets_summary(args),
"scan_mode": getattr(args, "scan_mode", ""),
"completed": completed,
"vulnerability_count": len(tracer.vulnerability_reports),
"vulnerability_count": len(vuln_reports),
"severity_counts": _severity_counts(tracer),
"vulnerabilities": _vulnerability_summary(tracer),
"stats": {
"agents": len(tracer.agents),
"tools": tracer.get_real_tool_count(),
"agents": len(tracer.agents) if tracer else 0,
"tools": tracer.get_real_tool_count() if tracer else 0,
"input_tokens": llm_stats.get("input_tokens", 0),
"output_tokens": llm_stats.get("output_tokens", 0),
"cost": llm_stats.get("cost", 0),
@ -152,7 +179,8 @@ def _format_generic(tracer: Any, args: argparse.Namespace) -> dict[str, Any]:
def _format_slack(tracer: Any, args: argparse.Namespace) -> dict[str, Any]:
"""Slack Block Kit payload."""
completed = _scan_completed(tracer)
vuln_count = len(tracer.vulnerability_reports)
vuln_reports = tracer.vulnerability_reports if tracer else []
vuln_count = len(vuln_reports)
counts = _severity_counts(tracer)
status_emoji = ":white_check_mark:" if completed else ":warning:"
@ -191,13 +219,14 @@ def _format_slack(tracer: Any, args: argparse.Namespace) -> dict[str, Any]:
]
# Add top vulnerabilities (max 5)
for report in tracer.vulnerability_reports[:5]:
title = report.get("title", "Untitled")
for report in vuln_reports[:5]:
title = _truncate(report.get("title", "Untitled"), 200)
severity = report.get("severity", "unknown").upper()
endpoint = report.get("endpoint", "")
text = f":rotating_light: *[{severity}]* {title}"
if endpoint:
text += f"\n`{endpoint}`"
text += f"\n`{_truncate(endpoint, 200)}`"
text = _truncate(text, _SLACK_SECTION_TEXT_LIMIT)
blocks.append(
{
"type": "section",
@ -211,7 +240,8 @@ def _format_slack(tracer: Any, args: argparse.Namespace) -> dict[str, Any]:
def _format_discord(tracer: Any, args: argparse.Namespace) -> dict[str, Any]:
"""Discord webhook payload with an embed."""
completed = _scan_completed(tracer)
vuln_count = len(tracer.vulnerability_reports)
vuln_reports = tracer.vulnerability_reports if tracer else []
vuln_count = len(vuln_reports)
counts = _severity_counts(tracer)
color = 0x22C55E if completed else 0xEAB308 # green / yellow
@ -226,27 +256,33 @@ def _format_discord(tracer: Any, args: argparse.Namespace) -> dict[str, Any]:
)
fields: list[dict[str, Any]] = [
{"name": "Target", "value": _targets_summary(args), "inline": True},
{
"name": "Target",
"value": _truncate(_targets_summary(args), _DISCORD_FIELD_VALUE_LIMIT),
"inline": True,
},
{"name": "Scan Mode", "value": getattr(args, "scan_mode", "N/A"), "inline": True},
{"name": "Vulnerabilities", "value": str(vuln_count), "inline": True},
{"name": "Severity Breakdown", "value": severity_line, "inline": False},
]
# Top vulnerabilities (max 5)
for report in tracer.vulnerability_reports[:5]:
title = report.get("title", "Untitled")
for report in vuln_reports[:5]:
title = _truncate(report.get("title", "Untitled"), 200)
severity = report.get("severity", "unknown").upper()
endpoint = report.get("endpoint", "")
value = f"**[{severity}]** {title}"
if endpoint:
value += f"\n`{endpoint}`"
value += f"\n`{_truncate(endpoint, 200)}`"
value = _truncate(value, _DISCORD_FIELD_VALUE_LIMIT)
fields.append({"name": "\u200b", "value": value, "inline": False})
status_text = "Scan Completed" if completed else "Scan Ended"
run_name = _truncate(getattr(args, "run_name", "N/A"), _DISCORD_TITLE_LIMIT)
embed: dict[str, Any] = {
"title": f"\ud83d\udd12 Strix \u2014 {status_text}",
"description": f"Run: **{getattr(args, 'run_name', 'N/A')}**",
"title": _truncate(f"\ud83d\udd12 Strix \u2014 {status_text}", _DISCORD_TITLE_LIMIT),
"description": f"Run: **{run_name}**",
"color": color,
"fields": fields,
"footer": {"text": "Strix Security Scanner"},

View file

@ -15,6 +15,7 @@ from strix.interface.webhooks import (
_resolve_format,
_severity_counts,
_targets_summary,
_truncate,
_vulnerability_summary,
send_completion_webhook,
)
@ -270,3 +271,66 @@ class TestSendCompletionWebhook:
# Should not raise
send_completion_webhook("https://example.com/hook", "generic", tracer, args)
@patch("strix.interface.webhooks.requests.post")
def test_none_tracer_skips_delivery(self, mock_post: MagicMock) -> None:
"""When tracer is None, webhook should not be sent."""
args = _make_args()
send_completion_webhook("https://example.com/hook", "generic", None, args)
mock_post.assert_not_called()
@patch("strix.interface.webhooks.requests.post")
def test_invalid_url_scheme_skips_delivery(self, mock_post: MagicMock) -> None:
"""Non-http(s) URLs should be rejected."""
tracer = _make_tracer()
args = _make_args()
send_completion_webhook("ftp://example.com/hook", "generic", tracer, args)
mock_post.assert_not_called()
class TestTruncate:
"""Tests for the _truncate helper."""
def test_short_string_unchanged(self) -> None:
assert _truncate("hello", 10) == "hello"
def test_exact_limit_unchanged(self) -> None:
assert _truncate("hello", 5) == "hello"
def test_long_string_truncated(self) -> None:
result = _truncate("hello world", 6)
assert len(result) == 6
assert result.endswith("\u2026")
def test_preserves_content_before_ellipsis(self) -> None:
result = _truncate("abcdefghij", 5)
assert result == "abcd\u2026"
class TestNoneTracer:
"""Tests for None tracer handling in helpers and formatters."""
def test_severity_counts_with_none(self) -> None:
counts = _severity_counts(None)
assert all(v == 0 for v in counts.values())
def test_vulnerability_summary_with_none(self) -> None:
assert _vulnerability_summary(None) == []
def test_format_generic_with_none(self) -> None:
args = _make_args()
payload = _format_generic(None, args)
assert payload["vulnerability_count"] == 0
assert payload["stats"]["agents"] == 0
def test_format_slack_with_none(self) -> None:
args = _make_args()
payload = _format_slack(None, args)
assert "blocks" in payload
assert len(payload["blocks"]) == 3 # header + section + severity
def test_format_discord_with_none(self) -> None:
args = _make_args()
payload = _format_discord(None, args)
assert "embeds" in payload
assert payload["embeds"][0]["color"] == 0xEAB308 # yellow (scan not completed)