diff --git a/backend/secuscan/parser_sandbox.py b/backend/secuscan/parser_sandbox.py index bfb56088..eebf016e 100644 --- a/backend/secuscan/parser_sandbox.py +++ b/backend/secuscan/parser_sandbox.py @@ -39,7 +39,6 @@ import re import sys import subprocess -import string import logging from pathlib import Path from typing import Any, Dict @@ -79,22 +78,35 @@ def __init__(self, plugin_id: str, reason: str, stderr: str = "") -> None: self.reason = reason # Keep stderr private; callers must not surface this to API consumers. self._stderr_diagnostic: str = stderr - # User-facing message: reason only — no stderr content. super().__init__(f"Parser sandbox failed for '{plugin_id}' ({reason})") + @property + def stderr_excerpt(self) -> str: + """Public read-only access to a truncated stderr excerpt (max 2000 chars).""" + return self._stderr_diagnostic[:2000] + + def __str__(self) -> str: + base = f"Parser sandbox failed for '{self.plugin_id}' ({self.reason})" + if self._stderr_diagnostic: + excerpt = self._stderr_diagnostic[:500] + return f"{base}\nstderr: {excerpt}" + return base + # --------------------------------------------------------------------------- # Bootstrap script injected into the child process via -c # --------------------------------------------------------------------------- -# Uses string.Template (${var} syntax) instead of str.format() so that -# user-controlled values in parser_path (e.g. braces) cannot cause -# KeyError/ValueError via format-string injection. +# Uses str.format() with {var} placeholders. The path is repr()'d via the +# {parser_path!r} conversion so it becomes a valid Python string literal even +# on Windows or when the path contains special characters. Any literal curly +# braces that appear *inside* the generated Python code are double-escaped +# (e.g. {{exc}}) so they survive .format() without being treated as slots. -_BOOTSTRAP_TEMPLATE = string.Template( +_BOOTSTRAP_TEMPLATE = ( "import sys, json, os\n" "\n" - "# Hard limit: refuse to read more than ${max_input_bytes} bytes from stdin.\n" - "MAX_INPUT = ${max_input_bytes}\n" + "# Hard limit: refuse to read more than {max_input_bytes} bytes from stdin.\n" + "MAX_INPUT = {max_input_bytes}\n" "raw = sys.stdin.buffer.read(MAX_INPUT + 1)\n" "if len(raw) > MAX_INPUT:\n" " sys.stderr.write('Parser input exceeded size limit\\n')\n" @@ -104,12 +116,12 @@ def __init__(self, plugin_id: str, reason: str, stderr: str = "") -> None: " envelope = json.loads(raw.decode('utf-8', errors='replace'))\n" " parser_input = envelope['input']\n" "except Exception as exc:\n" - " sys.stderr.write(f'Failed to decode envelope: {exc}\\n')\n" + " sys.stderr.write(f'Failed to decode envelope: {{exc}}\\n')\n" " sys.exit(3)\n" "\n" "# Load the plugin's parser module from an absolute path.\n" "import importlib.util\n" - "parser_path = ${parser_path_repr}\n" + "parser_path = {parser_path!r}\n" "spec = importlib.util.spec_from_file_location('_plugin_parser', parser_path)\n" "if spec is None or spec.loader is None:\n" " sys.stderr.write(f'Cannot load parser\\n')\n" @@ -168,10 +180,10 @@ def run_parser_in_sandbox( max_input_bytes = max(len(parser_input.encode("utf-8")) + 128, 64 * 1024) - # Use Template.safe_substitute so that any stray $ in parser_path does not - # raise; repr() ensures the path is a valid Python string literal. - bootstrap = _BOOTSTRAP_TEMPLATE.safe_substitute( - parser_path_repr=repr(str(parser_path)), + # {parser_path!r} in the template applies repr() automatically, producing a + # valid Python string literal even on Windows paths with backslashes. + bootstrap = _BOOTSTRAP_TEMPLATE.format( + parser_path=str(parser_path), max_input_bytes=max_input_bytes, ) @@ -256,13 +268,14 @@ def _read_stderr() -> None: timeout_seconds, plugin_id, ) - # Log sanitized stderr for internal diagnostics; do NOT pass to exception. logger.debug( "Parser sandbox stderr (plugin '%s', timed out): %s", plugin_id, _sanitize_stderr(stderr_text), ) - raise ParserSandboxError(plugin_id, f"timed out after {timeout_seconds}s") + raise ParserSandboxError( + plugin_id, f"timed out after {timeout_seconds}s", stderr=stderr_text + ) if proc.returncode != 0: logger.error( @@ -270,7 +283,6 @@ def _read_stderr() -> None: proc.returncode, plugin_id, ) - # Log sanitized stderr for internal diagnostics; do NOT pass to exception. logger.debug( "Parser sandbox stderr (plugin '%s', exit %d): %s", plugin_id, @@ -280,6 +292,7 @@ def _read_stderr() -> None: raise ParserSandboxError( plugin_id, f"subprocess exited with code {proc.returncode}", + stderr=stderr_text, ) stdout_bytes = b"".join(stdout_chunks) diff --git a/testing/backend/test_scapy_recon_parser.py b/testing/backend/test_scapy_recon_parser.py new file mode 100644 index 00000000..f4304cee --- /dev/null +++ b/testing/backend/test_scapy_recon_parser.py @@ -0,0 +1,397 @@ + + +import json +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +from plugins.scapy_recon.parser import parse + +PLUGIN_DIR = REPO_ROOT / "plugins" / "scapy_recon" + + +# --------------------------------------------------------------------------- +# Metadata contract tests +# --------------------------------------------------------------------------- + + +def test_scapy_recon_metadata_file_exists(): + """metadata.json must exist at the expected plugin path.""" + assert (PLUGIN_DIR / "metadata.json").exists(), ( + "metadata.json not found in plugins/scapy_recon/" + ) + + +def test_scapy_recon_parser_file_exists(): + """parser.py must exist alongside metadata.json.""" + assert (PLUGIN_DIR / "parser.py").exists(), ( + "parser.py not found in plugins/scapy_recon/" + ) + + +def test_scapy_recon_metadata_is_valid_json(): + """metadata.json must be valid, parseable JSON.""" + raw = (PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8") + data = json.loads(raw) + assert isinstance(data, dict) + + +def test_scapy_recon_metadata_id_matches_directory(): + """Plugin id in metadata.json must match the directory name.""" + data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8")) + assert data["id"] == "scapy_recon" + + +def test_scapy_recon_metadata_has_required_top_level_fields(): + """metadata.json must contain all required top-level keys.""" + data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8")) + for key in ("id", "name", "version", "description", "engine", "fields", "output"): + assert key in data, f"metadata.json missing required key: {key}" + + +def test_scapy_recon_metadata_output_parser_is_custom(): + """Output parser type must be 'custom', backed by parser.py.""" + data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8")) + assert data["output"]["parser"] == "custom" + + +def test_scapy_recon_metadata_engine_binary_is_python3(): + """Engine binary must be 'python3' (Scapy runs via Python).""" + data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8")) + assert data["engine"]["type"] == "cli" + assert data["engine"]["binary"] == "python3" + + +def test_scapy_recon_metadata_has_required_target_field(): + """Plugin must declare a required 'target' field.""" + data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8")) + fields = {f["id"]: f for f in data["fields"]} + assert "target" in fields, "Missing required field: target" + assert fields["target"]["required"] is True + + +def test_scapy_recon_metadata_has_type_field_with_defaults(): + """Plugin must declare a 'type' field with valid probe-type options.""" + data = json.loads((PLUGIN_DIR / "metadata.json").read_text(encoding="utf-8")) + fields = {f["id"]: f for f in data["fields"]} + assert "type" in fields, "Missing 'type' selector field" + type_field = fields["type"] + option_values = [opt["value"] for opt in type_field.get("options", [])] + assert "arp_ping" in option_values, "Expected 'arp_ping' probe type option" + assert "icmp_ping" in option_values, "Expected 'icmp_ping' probe type option" + + +# --------------------------------------------------------------------------- +# Parser tests — normal / representative sample output +# --------------------------------------------------------------------------- + +_ARP_OUTPUT = ( + "UP: 192.168.1.1 - aa:bb:cc:dd:ee:ff\n" + "UP: 192.168.1.10 - 11:22:33:44:55:66\n" + "UP: 10.0.0.5 - de:ad:be:ef:00:01\n" +) + +_ICMP_OUTPUT = ( + "UP: 192.168.1.1\n" + "UP: 192.168.1.20\n" +) + + +def test_scapy_recon_parser_returns_dict(): + """parse() must return a dict.""" + result = parse(_ARP_OUTPUT) + assert isinstance(result, dict) + + +def test_scapy_recon_parser_returns_required_keys(): + """parse() must return a dict with 'findings', 'count', and 'hosts' keys.""" + result = parse(_ARP_OUTPUT) + assert "findings" in result, "Missing 'findings' key" + assert "count" in result, "Missing 'count' key" + assert "hosts" in result, "Missing 'hosts' key" + + +def test_scapy_recon_parser_count_matches_hosts(): + """'count' must equal the number of hosts discovered.""" + result = parse(_ARP_OUTPUT) + assert result["count"] == len(result["hosts"]) + + +def test_scapy_recon_parser_arp_host_count(): + """Parser must detect all three hosts from the ARP sample output.""" + result = parse(_ARP_OUTPUT) + assert result["count"] == 3 + assert len(result["findings"]) == 3 + + +def test_scapy_recon_parser_arp_host_ip_and_mac(): + """Parser must correctly extract IP and MAC from an ARP-style line.""" + result = parse(_ARP_OUTPUT) + hosts = {h["ip"]: h["mac"] for h in result["hosts"]} + + assert "192.168.1.1" in hosts + assert hosts["192.168.1.1"] == "aa:bb:cc:dd:ee:ff" + + assert "192.168.1.10" in hosts + assert hosts["192.168.1.10"] == "11:22:33:44:55:66" + + assert "10.0.0.5" in hosts + assert hosts["10.0.0.5"] == "de:ad:be:ef:00:01" + + +def test_scapy_recon_parser_finding_has_required_keys(): + """Each finding must contain title, category, severity, description, remediation, metadata.""" + result = parse(_ARP_OUTPUT) + assert result["findings"], "Expected at least one finding" + for finding in result["findings"]: + for key in ("title", "category", "severity", "description", "remediation", "metadata"): + assert key in finding, f"Finding missing required key: {key}" + + +def test_scapy_recon_parser_finding_title_contains_ip(): + """Each finding title must reference the discovered host IP.""" + result = parse(_ARP_OUTPUT) + host_ips = {h["ip"] for h in result["hosts"]} + for finding in result["findings"]: + assert any(ip in finding["title"] for ip in host_ips), ( + f"Finding title '{finding['title']}' does not reference any known host IP" + ) + + +def test_scapy_recon_parser_finding_category_is_network_discovery(): + """All findings must use 'Network Discovery' as their category.""" + result = parse(_ARP_OUTPUT) + for finding in result["findings"]: + assert finding["category"] == "Network Discovery", ( + f"Unexpected category: {finding['category']}" + ) + + +def test_scapy_recon_parser_finding_severity_is_info(): + """All findings must use 'info' severity (host discovery is informational).""" + result = parse(_ARP_OUTPUT) + for finding in result["findings"]: + assert finding["severity"] == "info", ( + f"Unexpected severity: {finding['severity']}" + ) + + +def test_scapy_recon_parser_finding_metadata_contains_ip_and_mac(): + """Each finding's metadata must carry both 'ip' and 'mac' keys.""" + result = parse(_ARP_OUTPUT) + for finding in result["findings"]: + assert "ip" in finding["metadata"], "Finding metadata missing 'ip'" + assert "mac" in finding["metadata"], "Finding metadata missing 'mac'" + + +def test_scapy_recon_parser_finding_metadata_ip_matches_host(): + """The 'ip' stored in each finding's metadata must match a host entry.""" + result = parse(_ARP_OUTPUT) + known_ips = {h["ip"] for h in result["hosts"]} + for finding in result["findings"]: + assert finding["metadata"]["ip"] in known_ips, ( + f"finding metadata ip '{finding['metadata']['ip']}' not in host list" + ) + + +def test_scapy_recon_parser_finding_has_remediation(): + """Each finding must include a non-empty remediation string.""" + result = parse(_ARP_OUTPUT) + for finding in result["findings"]: + assert isinstance(finding["remediation"], str) + assert len(finding["remediation"].strip()) > 0, "Remediation must not be blank" + + +def test_scapy_recon_parser_finding_description_contains_ip_and_mac(): + """Description for ARP hosts must mention both IP and MAC.""" + result = parse(_ARP_OUTPUT) + for finding in result["findings"]: + ip = finding["metadata"]["ip"] + mac = finding["metadata"]["mac"] + desc = finding["description"] + assert ip in desc, f"Description missing IP '{ip}': {desc}" + assert mac in desc, f"Description missing MAC '{mac}': {desc}" + + +# --------------------------------------------------------------------------- +# Parser tests — ICMP output (host line with no MAC address) +# --------------------------------------------------------------------------- + + +def test_scapy_recon_parser_icmp_output_host_count(): + """Parser must detect both hosts from the ICMP sample output.""" + result = parse(_ICMP_OUTPUT) + assert result["count"] == 2 + assert len(result["findings"]) == 2 + + +def test_scapy_recon_parser_icmp_missing_mac_defaults_to_unknown(): + """When a line contains no MAC address, mac must default to 'Unknown'.""" + result = parse(_ICMP_OUTPUT) + for host in result["hosts"]: + assert host["mac"] == "Unknown", ( + f"Expected 'Unknown' mac for ICMP-style line, got '{host['mac']}'" + ) + + +def test_scapy_recon_parser_icmp_ips_are_extracted(): + """Parser must correctly extract IPs from ICMP-style lines (no MAC).""" + result = parse(_ICMP_OUTPUT) + ips = {h["ip"] for h in result["hosts"]} + assert "192.168.1.1" in ips + assert "192.168.1.20" in ips + + +# --------------------------------------------------------------------------- +# Parser tests — single host output +# --------------------------------------------------------------------------- + + +def test_scapy_recon_parser_single_host_with_mac(): + """Parser handles a single ARP-style line correctly.""" + output = "UP: 172.16.0.1 - 00:11:22:33:44:55\n" + result = parse(output) + assert result["count"] == 1 + assert len(result["findings"]) == 1 + assert result["hosts"][0]["ip"] == "172.16.0.1" + assert result["hosts"][0]["mac"] == "00:11:22:33:44:55" + + +def test_scapy_recon_parser_single_host_description_contains_ip_and_mac(): + """Description for a single host must mention both IP and MAC.""" + output = "UP: 10.10.10.1 - ff:ee:dd:cc:bb:aa\n" + result = parse(output) + finding = result["findings"][0] + assert "10.10.10.1" in finding["description"] + assert "ff:ee:dd:cc:bb:aa" in finding["description"] + + +# --------------------------------------------------------------------------- +# Parser tests — empty / malformed output (must not crash) +# --------------------------------------------------------------------------- + + +def test_scapy_recon_parser_empty_string(): + """Parser must not crash on empty input and must return empty results.""" + result = parse("") + assert isinstance(result, dict) + assert result["findings"] == [] + assert result["count"] == 0 + assert result["hosts"] == [] + + +def test_scapy_recon_parser_whitespace_only(): + """Parser must handle whitespace-only input gracefully.""" + result = parse(" \n\n\t \n") + assert isinstance(result, dict) + assert result["findings"] == [] + assert result["count"] == 0 + assert result["hosts"] == [] + + +def test_scapy_recon_parser_no_up_lines(): + """Parser must return empty results when no 'UP:' lines are present.""" + output = ( + "Starting Scapy scan...\n" + "Sending ARP requests to 192.168.1.0/24\n" + "Scan complete. No live hosts found.\n" + ) + result = parse(output) + assert result["findings"] == [] + assert result["count"] == 0 + assert result["hosts"] == [] + + +def test_scapy_recon_parser_mixed_valid_and_noise_lines(): + """Parser must extract only 'UP:' lines and ignore all other content.""" + output = ( + "Starting Scapy scan...\n" + "UP: 192.168.1.5 - 00:aa:bb:cc:dd:ee\n" + "Some random debug line\n" + "UP: 192.168.1.9 - ff:00:11:22:33:44\n" + "Scan finished.\n" + ) + result = parse(output) + assert result["count"] == 2 + ips = {h["ip"] for h in result["hosts"]} + assert "192.168.1.5" in ips + assert "192.168.1.9" in ips + + +def test_scapy_recon_parser_malformed_up_line_no_crash(): + """Parser must not crash when a 'UP:' line lacks the expected format.""" + malformed_outputs = [ + "UP:\n", + "UP: \n", + "UP: \n", + "UP: no-dash-separator\n", + "UP: 999.999.999.999 - INVALID:MAC:HERE\n", + ] + for bad_output in malformed_outputs: + try: + result = parse(bad_output) + assert isinstance(result, dict), ( + f"parse() returned non-dict for input: {repr(bad_output)}" + ) + except Exception as exc: + pytest.fail( + f"parse() raised {type(exc).__name__} for input {repr(bad_output)}: {exc}" + ) + + +def test_scapy_recon_parser_line_missing_mac_does_not_crash(): + """A 'UP:' line without a ' - ' separator must be handled gracefully.""" + output = "UP: 192.168.1.1\n" + result = parse(output) + assert isinstance(result, dict) + assert result["count"] == 1 + assert result["hosts"][0]["mac"] == "Unknown" + + +def test_scapy_recon_parser_empty_string_returns_all_keys(): + """Passing an empty string must still return a dict with all required keys.""" + result = parse("") + assert isinstance(result, dict) + assert "findings" in result + assert "count" in result + assert "hosts" in result + + +def test_scapy_recon_parser_large_subnet_output(): + """Parser must handle a large number of UP: lines without error.""" + lines = [ + f"UP: 10.0.{i // 255}.{i % 255} - 00:11:22:33:{i // 255:02x}:{i % 255:02x}" + for i in range(1, 51) + ] + output = "\n".join(lines) + "\n" + result = parse(output) + assert result["count"] == 50 + assert len(result["findings"]) == 50 + assert len(result["hosts"]) == 50 + + +# --------------------------------------------------------------------------- +# Parser tests — type assertions +# --------------------------------------------------------------------------- + + +def test_scapy_recon_parser_hosts_list_is_list(): + """'hosts' in the result must always be a list.""" + result = parse(_ARP_OUTPUT) + assert isinstance(result["hosts"], list) + + +def test_scapy_recon_parser_findings_list_is_list(): + """'findings' in the result must always be a list.""" + result = parse("") + assert isinstance(result["findings"], list) + + +def test_scapy_recon_parser_count_is_integer(): + """'count' in the result must always be an integer.""" + result = parse(_ARP_OUTPUT) + assert isinstance(result["count"], int)