Skip to content

fix: sanitize parser sandbox error messages to prevent sensitive data leaks #1680

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ionfwsrijan wants to merge 3 commits into
base: main
Choose a base branch
from
Open

fix: sanitize parser sandbox error messages to prevent sensitive data leaks #1680

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c4e1540
fix: sanitize parser sandbox error messages to prevent sensitive data…
ionfwsrijan Jul 10, 2026
473dff7
fix: remove stderr from API-facing error messages, add regression tests
ionfwsrijan Jul 14, 2026
7be9cb4
fix: mock client.stream() as async context manager in TLS verificatio…
ionfwsrijan Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions backend/secuscan/parser_sandbox.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,8 @@
from pathlib import Path
from typing import Any, Dict

from .redaction import redact

logger = logging.getLogger(__name__)

# Defaults — overridden by the Settings values passed at call time.
Expand Down Expand Up @@ -84,10 +86,8 @@ class ParserSandboxError(RuntimeError):
def __init__(self, plugin_id: str, reason: str, stderr: str = "") -> None:
self.plugin_id = plugin_id
self.reason = reason
# Keep stderr private; callers must not surface this to API consumers.
self._stderr_diagnostic: str = stderr
self.stderr_excerpt = stderr[:2000] if stderr else ""
# User-facing message: reason only — no stderr content.
self.stderr_excerpt = redact(stderr[:2000]) if stderr else ""
super().__init__(f"Parser sandbox failed for '{plugin_id}' ({reason})")


Expand Down
34 changes: 34 additions & 0 deletions testing/backend/unit/test_parser_sandbox.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -403,3 +403,37 @@ def test_stderr_excerpt_truncated_to_2000_chars(self):
def test_str_contains_plugin_id(self):
err = ParserSandboxError("my_plugin", "bad thing")
assert "my_plugin" in str(err)

def test_str_does_not_expose_stderr_secrets(self):
"""Regression: API-facing str() must never leak parser stderr contents."""
secret = "Bearer eyJhbGciOiJIUzI1NiJ9.secret_payload_12345"
api_key = "AKIA1234567890ABCDEF"
err = ParserSandboxError(
"plugin_secret",
"unexpected failure",
stderr=f"secret={secret} api_key={api_key}",
)
msg = str(err)
assert secret not in msg
assert api_key not in msg
assert "eyJhbGciOiJIUzI1NiJ9" not in msg
assert "AKIA1234567890ABCDEF" not in msg

def test_str_does_not_expose_arbitrary_stderr(self):
"""Regression: arbitrary parser diagnostic output must not appear in str()."""
arbitrary_output = (
"Password: hunter2\n"
"Internal path: /etc/shadow\n"
"Private key: -----BEGIN RSA PRIVATE KEY-----\nMIIEpAIBAAKCAQ..."
)
err = ParserSandboxError("plugin_arb", "crash", stderr=arbitrary_output)
msg = str(err)
assert "hunter2" not in msg
assert "/etc/shadow" not in msg
assert "BEGIN RSA PRIVATE KEY" not in msg

def test_stderr_excerpt_is_available_for_diagnostics(self):
"""Internal diagnostic attribute is still accessible for logging."""
stderr = "diagnostic: segfault at 0xdeadbeef"
err = ParserSandboxError("plugin_diag", "signal", stderr=stderr)
assert err.stderr_excerpt == stderr
66 changes: 41 additions & 25 deletions testing/backend/unit/test_tls_verification.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,20 +52,39 @@ def test_env_override_true(self, monkeypatch):


class TestCrawlerVerifySsl:
@pytest.mark.asyncio
async def test_crawl_target_passes_verify_ssl(self):
def _make_mock_stream_response(self):
mock_response = MagicMock()
mock_response.text = "<html></html>"
mock_response.url = "https://example.com/"
mock_response.status_code = 200
mock_response.headers = {}
mock_response.headers = MagicMock()
mock_response.headers.get = MagicMock(return_value=None)
mock_response.headers.get_list = MagicMock(return_value=[])
mock_response.history = []
return mock_response

def _make_mock_client(self, mock_response):
mock_client = AsyncMock()
mock_client.get = AsyncMock(return_value=mock_response)
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)

mock_stream_ctx = AsyncMock()
mock_stream_ctx.__aenter__ = AsyncMock(return_value=mock_response)
mock_stream_ctx.__aexit__ = AsyncMock(return_value=False)
mock_client.stream = MagicMock(return_value=mock_stream_ctx)
async def _aiter_bytes():
yield b"<html></html>"

mock_response.aiter_bytes = _aiter_bytes

return mock_client

@pytest.mark.asyncio
async def test_crawl_target_passes_verify_ssl(self):
mock_response = self._make_mock_stream_response()
mock_client = self._make_mock_client(mock_response)

with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=mock_client) as mock_cls:
with patch("backend.secuscan.crawler.settings") as mock_settings:
mock_settings.verify_ssl = True
Expand All @@ -79,17 +98,8 @@ async def test_crawl_target_passes_verify_ssl(self):

@pytest.mark.asyncio
async def test_crawl_target_verify_ssl_false_when_disabled(self):
mock_response = MagicMock()
mock_response.text = "<html></html>"
mock_response.url = "https://example.com/"
mock_response.status_code = 200
mock_response.headers = {}
mock_response.history = []

mock_client = AsyncMock()
mock_client.get = AsyncMock(return_value=mock_response)
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_response = self._make_mock_stream_response()
mock_client = self._make_mock_client(mock_response)

with patch("backend.secuscan.crawler.httpx.AsyncClient", return_value=mock_client) as mock_cls:
with patch("backend.secuscan.crawler.settings") as mock_settings:
Expand Down Expand Up @@ -119,16 +129,19 @@ async def test_api_scanner_passes_verify_ssl(self):
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)

mock_crawl = AsyncMock(return_value={"api_hints": []})

with patch("backend.secuscan.scanners.api_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls:
with patch("backend.secuscan.scanners.api_scanner.settings") as mock_settings:
mock_settings.verify_ssl = True
with patch.object(scanner, "_fetch_spec", return_value=None):
with patch.object(scanner, "_probe_graphql", return_value=([], [])):
await scanner.run("https://example.com", {})
with patch("backend.secuscan.scanners.api_scanner.crawl_target", mock_crawl):
with patch.object(scanner, "_fetch_spec", return_value=None):
with patch.object(scanner, "_probe_graphql", return_value=([], [])):
await scanner.run("https://example.com", {})

mock_cls.assert_called()
_, kwargs = mock_cls.call_args
assert kwargs["verify"] is True
mock_cls.assert_called()
_, kwargs = mock_cls.call_args
assert kwargs["verify"] is True

@pytest.mark.asyncio
async def test_api_scanner_verify_ssl_false_when_disabled(self):
Expand All @@ -141,15 +154,18 @@ async def test_api_scanner_verify_ssl_false_when_disabled(self):
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)

mock_crawl = AsyncMock(return_value={"api_hints": []})

with patch("backend.secuscan.scanners.api_scanner.httpx.AsyncClient", return_value=mock_client) as mock_cls:
with patch("backend.secuscan.scanners.api_scanner.settings") as mock_settings:
mock_settings.verify_ssl = False
with patch.object(scanner, "_fetch_spec", return_value=None):
with patch.object(scanner, "_probe_graphql", return_value=([], [])):
await scanner.run("https://example.com", {})
with patch("backend.secuscan.scanners.api_scanner.crawl_target", mock_crawl):
with patch.object(scanner, "_fetch_spec", return_value=None):
with patch.object(scanner, "_probe_graphql", return_value=([], [])):
await scanner.run("https://example.com", {})

_, kwargs = mock_cls.call_args
assert kwargs["verify"] is False
_, kwargs = mock_cls.call_args
assert kwargs["verify"] is False


# ---------------------------------------------------------------------------
Expand Down
Loading