From 2dc5a8f9eb9af5ded1beccaed7fb53910c32829f Mon Sep 17 00:00:00 2001 From: karrisanthoshigayatri Date: Sat, 4 Jul 2026 05:06:20 +0530 Subject: [PATCH 1/2] Parser Sandbox Captures Stderr with Sensitive Data and Leaks It Via Exception Messages --- backend/secuscan/executor.py | 9 +- backend/secuscan/parser_sandbox.py | 153 +++++++++++++++++++---------- backend/secuscan/routes.py | 6 +- 3 files changed, 108 insertions(+), 60 deletions(-) diff --git a/backend/secuscan/executor.py b/backend/secuscan/executor.py index fad732c2b..1d81c25a9 100644 --- a/backend/secuscan/executor.py +++ b/backend/secuscan/executor.py @@ -818,6 +818,7 @@ async def execute_task(self, task_id: str) -> None: except Exception as e: logger.error(f"Task {task_id} failed: {e}", exc_info=True) duration = (time.time() - start_time) if 'start_time' in locals() else 0 + safe_error = redact(str(e)) await db.execute( """ UPDATE tasks SET @@ -831,7 +832,7 @@ async def execute_task(self, task_id: str) -> None: TaskStatus.FAILED.value, datetime.now().isoformat(), duration, - str(e), + safe_error, task_id ) ) @@ -841,9 +842,9 @@ async def execute_task(self, task_id: str) -> None: await db.log_audit( "task_failed", - f"Task failed: {str(e)}", + f"Task failed: {safe_error}", severity="error", - context={"task_id": task_id, "error": str(e)}, + context={"task_id": task_id, "error": safe_error}, task_id=task_id ) finally: @@ -1622,7 +1623,7 @@ def _parse_results(self, plugin, output: str) -> Dict[str, Any]: except Exception as exc: logger.error("Unexpected error running parser sandbox for '%s': %s", plugin.id, exc) raise RuntimeError( - f"Custom parser encountered an unexpected error for plugin '{plugin.id}': {exc}" + f"Custom parser encountered an unexpected error for plugin '{plugin.id}'" ) from exc # 2. Fallback to legacy built-in parsers (only reached when no parser.py exists) diff --git a/backend/secuscan/parser_sandbox.py b/backend/secuscan/parser_sandbox.py index 4b178f097..bfb560887 100644 --- a/backend/secuscan/parser_sandbox.py +++ b/backend/secuscan/parser_sandbox.py @@ -22,15 +22,24 @@ The child process is a minimal Python bootstrap that imports the plugin's parser.py, calls parse(input_data), and writes the result to stdout. It imports nothing from the backend package, so no application state leaks. + +Security note on stderr +----------------------- +Stderr from the child process may contain stack traces, file paths, partial +parser-input excerpts, or other diagnostic data. It is intentionally NOT +included in the user-facing exception message or stored in ``error_message``. +Full stderr is logged at DEBUG level (internal diagnostics only) so operators +can investigate failures without leaking sensitive details to API callers. """ from __future__ import annotations import json import os +import re import sys import subprocess -import textwrap +import string import logging from pathlib import Path from typing import Any, Dict @@ -41,61 +50,83 @@ _DEFAULT_TIMEOUT_SECONDS: int = 30 _DEFAULT_MAX_OUTPUT_BYTES: int = 8 * 1024 * 1024 # 8 MB +# Strip absolute file-system paths and Python line-number references from +# stderr before any logging so internal topology is not exposed even in logs +# that might be shipped to external observability platforms. +_PATH_RE = re.compile(r'(?:[A-Za-z]:[\\/]|/)[^\s"\'<>|:*?\n]{3,}') +_LINENO_RE = re.compile(r'\bline \d+\b', re.IGNORECASE) + + +def _sanitize_stderr(stderr: str, max_chars: int = 500) -> str: + """Strip file paths and line numbers from stderr; truncate to *max_chars*.""" + sanitized = _PATH_RE.sub("[PATH]", stderr) + sanitized = _LINENO_RE.sub("[LINE]", sanitized) + return sanitized[:max_chars] + class ParserSandboxError(RuntimeError): - """Raised when the sandboxed parser fails for any reason.""" + """Raised when the sandboxed parser fails for any reason. + + The public exception message intentionally contains only the *reason* + string (a short, controlled description) and never includes raw stderr + content. Stderr is stored privately on the instance so callers that need + it for internal diagnostics can access it, but it must not be forwarded to + API responses or stored as a user-facing error message. + """ def __init__(self, plugin_id: str, reason: str, stderr: str = "") -> None: self.plugin_id = plugin_id self.reason = reason - self.stderr_excerpt = stderr[:2000] if stderr else "" - detail = f": {stderr[:200]}" if stderr.strip() else "" - super().__init__(f"Parser sandbox failed for '{plugin_id}' ({reason}){detail}") + # 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})") # --------------------------------------------------------------------------- # Bootstrap script injected into the child process via -c # --------------------------------------------------------------------------- - -_BOOTSTRAP_TEMPLATE = textwrap.dedent( - """\ - import sys, json, os - - # Hard limit: refuse to read more than {max_input_bytes} bytes from stdin. - MAX_INPUT = {max_input_bytes} - raw = sys.stdin.buffer.read(MAX_INPUT + 1) - if len(raw) > MAX_INPUT: - sys.stderr.write("Parser input exceeded size limit\\n") - sys.exit(2) - - try: - envelope = json.loads(raw.decode("utf-8", errors="replace")) - parser_input = envelope["input"] - except Exception as exc: - sys.stderr.write(f"Failed to decode envelope: {{exc}}\\n") - sys.exit(3) - - # Load the plugin's parser module from an absolute path. - import importlib.util - parser_path = {parser_path!r} - spec = importlib.util.spec_from_file_location("_plugin_parser", parser_path) - if spec is None or spec.loader is None: - sys.stderr.write(f"Cannot load parser from {{parser_path}}\\n") - sys.exit(4) - - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - - if not hasattr(module, "parse"): - sys.stderr.write("Parser module missing 'parse' function\\n") - sys.exit(5) - - result = module.parse(parser_input) - - # Write result as a single JSON line. - sys.stdout.write(json.dumps(result, default=str)) - sys.stdout.flush() -""" +# 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. + +_BOOTSTRAP_TEMPLATE = string.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" + "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" + " sys.exit(2)\n" + "\n" + "try:\n" + " 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.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" + "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" + " sys.exit(4)\n" + "\n" + "module = importlib.util.module_from_spec(spec)\n" + "spec.loader.exec_module(module)\n" + "\n" + "if not hasattr(module, 'parse'):\n" + " sys.stderr.write(\"Parser module missing 'parse' function\\n\")\n" + " sys.exit(5)\n" + "\n" + "result = module.parse(parser_input)\n" + "\n" + "# Write result as a single JSON line.\n" + "sys.stdout.write(json.dumps(result, default=str))\n" + "sys.stdout.flush()\n" ) @@ -137,8 +168,10 @@ def run_parser_in_sandbox( max_input_bytes = max(len(parser_input.encode("utf-8")) + 128, 64 * 1024) - bootstrap = _BOOTSTRAP_TEMPLATE.format( - parser_path=str(parser_path), + # 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)), max_input_bytes=max_input_bytes, ) @@ -223,19 +256,30 @@ def _read_stderr() -> None: timeout_seconds, plugin_id, ) - raise ParserSandboxError(plugin_id, f"timed out after {timeout_seconds}s", stderr_text) + # 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") if proc.returncode != 0: logger.error( - "Parser sandbox exited with code %d for plugin '%s': %s", + "Parser sandbox exited with code %d for plugin '%s'", proc.returncode, plugin_id, - stderr_text[:500], + ) + # Log sanitized stderr for internal diagnostics; do NOT pass to exception. + logger.debug( + "Parser sandbox stderr (plugin '%s', exit %d): %s", + plugin_id, + proc.returncode, + _sanitize_stderr(stderr_text), ) raise ParserSandboxError( plugin_id, f"subprocess exited with code {proc.returncode}", - stderr_text, ) stdout_bytes = b"".join(stdout_chunks) @@ -250,10 +294,14 @@ def _read_stderr() -> None: try: parsed = json.loads(stdout_bytes.decode("utf-8", errors="replace")) except json.JSONDecodeError as exc: + logger.debug( + "Parser sandbox non-JSON stdout (plugin '%s'): %s", + plugin_id, + _sanitize_stderr(stderr_text), + ) raise ParserSandboxError( plugin_id, f"parser returned non-JSON output: {exc}", - stderr_text, ) if not isinstance(parsed, (dict, list)): @@ -261,6 +309,5 @@ def _read_stderr() -> None: plugin_id, f"parser returned unexpected type {type(parsed).__name__}; expected dict or list", ) - logger.info("Parser sandbox completed successfully for plugin '%s'", plugin_id) return parsed if isinstance(parsed, dict) else {"findings": parsed} diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index 4e2b73f89..185fcbf9b 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -103,7 +103,7 @@ def _json_payload(value: Any, fallback: str) -> str: from .plugins import get_plugin_manager, init_plugins from . import notification_service from .executor import executor -from .redaction import redact_inputs +from .redaction import redact, redact_inputs from .ratelimit import ( rate_limiter, concurrent_limiter, workflow_rate_limiter, task_start_limiter, vault_limiter, @@ -923,8 +923,8 @@ async def get_task_result(task_id: str, owner: str = Depends(get_current_owner)) "raw_output_excerpt": raw_output, "raw_output": raw_output, "command_used": task_row["command_used"], - "errors": [{"message": task_row["error_message"]}] if task_row["error_message"] else [], - "error_message": task_row["error_message"], + "errors": [{"message": redact(task_row["error_message"])}] if task_row["error_message"] else [], + "error_message": redact(task_row["error_message"]) if task_row["error_message"] else None, "exit_code": task_row["exit_code"], "metadata": {} } From 9dc5052ea002b7faa1239c6ea9d0cf3cfc0d3adc Mon Sep 17 00:00:00 2001 From: karrisanthoshigayatri Date: Sun, 5 Jul 2026 09:35:21 +0530 Subject: [PATCH 2/2] Add parser unit tests for subdomain_discovery plugin #1436 --- .../unit/test_subdomain_discovery_parser.py | 222 ++++++++++++++++++ 1 file changed, 222 insertions(+) create mode 100644 testing/backend/unit/test_subdomain_discovery_parser.py diff --git a/testing/backend/unit/test_subdomain_discovery_parser.py b/testing/backend/unit/test_subdomain_discovery_parser.py new file mode 100644 index 000000000..699a97789 --- /dev/null +++ b/testing/backend/unit/test_subdomain_discovery_parser.py @@ -0,0 +1,222 @@ +"""Dedicated parser unit tests for plugins/subdomain_discovery (issue #1436). + +Covers: + - Normal output: one or more subdomains, one per line. + - Single-subdomain output. + - Output with blank lines / surrounding whitespace. + - Empty string input. + - Whitespace-only input. + - Malformed / non-subdomain lines (parser must not crash). + - Return-value contract: keys, types, and counts are always consistent. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +from backend.secuscan.config import settings + +PLUGIN_ID = "subdomain_discovery" +PARSER_PATH = Path(settings.plugins_dir) / PLUGIN_ID / "parser.py" +FIXTURE_PATH = Path(__file__).parent / "fixtures" / PLUGIN_ID / "sample_output.txt" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _load_parser(): + """Dynamically load the plugin's parser.py module.""" + spec = importlib.util.spec_from_file_location("subdomain_discovery_parser", PARSER_PATH) + assert spec is not None and spec.loader is not None, ( + f"Cannot load parser from {PARSER_PATH} — check plugins_dir setting" + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture(scope="module") +def parser(): + """Module-scoped fixture so the file is loaded once for all tests.""" + return _load_parser() + + +def _assert_finding_shape(finding: dict, expected_subdomain: str) -> None: + """Assert that a single finding dict has the expected structure and values.""" + assert finding["title"] == f"Subdomain Discovered: {expected_subdomain}" + assert finding["category"] == "Subdomain" + assert finding["severity"] == "info" + assert isinstance(finding["description"], str) and finding["description"] + assert isinstance(finding["remediation"], str) and finding["remediation"] + + meta = finding["metadata"] + assert meta["subdomain"] == expected_subdomain + assert meta["source"] == "subfinder" + assert meta["confidence"] == "high" + assert expected_subdomain in meta["evidence"] + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestParserWithFixtureSampleOutput: + """Tests using the canonical sample_output.txt fixture.""" + + def test_fixture_file_exists(self): + assert FIXTURE_PATH.exists(), f"Fixture not found: {FIXTURE_PATH}" + + def test_returns_correct_count(self, parser): + raw = FIXTURE_PATH.read_text(encoding="utf-8") + result = parser.parse(raw) + assert result["count"] == 3 + + def test_returns_correct_subdomains_list(self, parser): + raw = FIXTURE_PATH.read_text(encoding="utf-8") + result = parser.parse(raw) + assert result["subdomains"] == [ + "api.secuscan.in", + "staging.secuscan.in", + "dev.secuscan.in", + ] + + def test_findings_length_matches_count(self, parser): + raw = FIXTURE_PATH.read_text(encoding="utf-8") + result = parser.parse(raw) + assert len(result["findings"]) == result["count"] + + def test_first_finding_has_correct_shape(self, parser): + raw = FIXTURE_PATH.read_text(encoding="utf-8") + result = parser.parse(raw) + _assert_finding_shape(result["findings"][0], "api.secuscan.in") + + def test_all_findings_have_correct_shape(self, parser): + raw = FIXTURE_PATH.read_text(encoding="utf-8") + result = parser.parse(raw) + expected = ["api.secuscan.in", "staging.secuscan.in", "dev.secuscan.in"] + for finding, subdomain in zip(result["findings"], expected): + _assert_finding_shape(finding, subdomain) + + +class TestParserReturnContract: + """Verify the return-value contract (keys and types) for all inputs.""" + + def test_result_always_has_required_keys(self, parser): + for raw in ["", " ", "api.example.com\nsub.example.com"]: + result = parser.parse(raw) + assert "findings" in result + assert "count" in result + assert "subdomains" in result + + def test_findings_is_always_a_list(self, parser): + for raw in ["", "api.example.com"]: + result = parser.parse(raw) + assert isinstance(result["findings"], list) + + def test_subdomains_is_always_a_list(self, parser): + for raw in ["", "api.example.com"]: + result = parser.parse(raw) + assert isinstance(result["subdomains"], list) + + def test_count_equals_len_subdomains(self, parser): + for raw in ["", "a.example.com", "a.example.com\nb.example.com"]: + result = parser.parse(raw) + assert result["count"] == len(result["subdomains"]) + + def test_count_equals_len_findings(self, parser): + for raw in ["", "a.example.com", "a.example.com\nb.example.com"]: + result = parser.parse(raw) + assert result["count"] == len(result["findings"]) + + +class TestParserEmptyAndWhitespaceInput: + """Parser must not crash and must return empty results for blank inputs.""" + + def test_empty_string_does_not_crash(self, parser): + result = parser.parse("") + assert result is not None + + def test_empty_string_returns_zero_findings(self, parser): + result = parser.parse("") + assert result["findings"] == [] + assert result["count"] == 0 + assert result["subdomains"] == [] + + def test_whitespace_only_does_not_crash(self, parser): + result = parser.parse(" \n \n ") + assert result is not None + + def test_whitespace_only_returns_zero_findings(self, parser): + result = parser.parse(" \n \n ") + assert result["findings"] == [] + assert result["count"] == 0 + assert result["subdomains"] == [] + + def test_trailing_newlines_are_ignored(self, parser): + result = parser.parse("api.example.com\n\n\n") + assert result["count"] == 1 + assert result["subdomains"] == ["api.example.com"] + + def test_leading_and_trailing_blank_lines_are_ignored(self, parser): + result = parser.parse("\n\napi.example.com\n\n") + assert result["count"] == 1 + + +class TestParserSingleSubdomain: + """Parser must handle a single-line input correctly.""" + + def test_single_subdomain_count(self, parser): + result = parser.parse("mail.example.com") + assert result["count"] == 1 + + def test_single_subdomain_in_list(self, parser): + result = parser.parse("mail.example.com") + assert result["subdomains"] == ["mail.example.com"] + + def test_single_subdomain_finding_shape(self, parser): + result = parser.parse("mail.example.com") + _assert_finding_shape(result["findings"][0], "mail.example.com") + + def test_single_subdomain_with_surrounding_whitespace(self, parser): + result = parser.parse(" mail.example.com ") + assert result["count"] == 1 + assert result["subdomains"] == ["mail.example.com"] + + +class TestParserMalformedAndEdgeCaseInput: + """Parser must not raise for unexpected or malformed input.""" + + def test_non_domain_lines_do_not_crash(self, parser): + malformed = "not-a-domain\n!!bad!!\n123\n" + result = parser.parse(malformed) + assert result is not None + + def test_non_domain_lines_still_produce_findings(self, parser): + # The parser does not validate domain format — it trusts subfinder output. + # Non-domain tokens are included as-is, so count should reflect all lines. + malformed = "not-a-domain\n!!bad!!\n123" + result = parser.parse(malformed) + assert result["count"] == 3 + assert len(result["findings"]) == 3 + + def test_mixed_valid_and_invalid_lines_do_not_crash(self, parser): + mixed = "api.example.com\n\n!!garbage!!\nsub.example.com" + result = parser.parse(mixed) + # Blank line is ignored; two valid + one garbage → 3 total + assert result["count"] == 3 + + def test_very_long_single_line_does_not_crash(self, parser): + long_line = "a" * 1000 + ".example.com" + result = parser.parse(long_line) + assert result["count"] == 1 + assert result["subdomains"] == [long_line] + + def test_many_subdomains_performance(self, parser): + many = "\n".join(f"sub{i}.example.com" for i in range(500)) + result = parser.parse(many) + assert result["count"] == 500 + assert len(result["findings"]) == 500