From fa6cb31a683666e20707bc2a0509451eca81d230 Mon Sep 17 00:00:00 2001 From: ionfwsrijan <201338831+ionfwsrijan@users.noreply.github.com> Date: Mon, 6 Jul 2026 00:38:54 +0530 Subject: [PATCH 1/7] fix: apply redaction before SSE broadcast in _execute_command (#1624) - Redact each decoded line before appending to output_lines and before broadcasting to SSE stream listeners - Add test_execute_command_redacts_secrets_before_broadcast to verify that known secret patterns are redacted from stream events - The raw_output_path file redaction (line 599) remains as defense-in-depth --- backend/secuscan/executor.py | 5 +-- testing/backend/unit/test_executor.py | 49 +++++++++++++++++++++++++++ 2 files changed, 52 insertions(+), 2 deletions(-) diff --git a/backend/secuscan/executor.py b/backend/secuscan/executor.py index 787d1ffb4..b1c73c4ea 100644 --- a/backend/secuscan/executor.py +++ b/backend/secuscan/executor.py @@ -883,8 +883,9 @@ async def read_stream(): line = await stdout.readline() if line: decoded_line = line.decode("utf-8", errors="replace") - output_lines.append(decoded_line) - await self._broadcast(task_id, "output", decoded_line) + safe_line = redact(decoded_line) + output_lines.append(safe_line) + await self._broadcast(task_id, "output", safe_line) try: await asyncio.wait_for(read_stream(), timeout=timeout) diff --git a/testing/backend/unit/test_executor.py b/testing/backend/unit/test_executor.py index b685d7e4b..f222fcc0f 100644 --- a/testing/backend/unit/test_executor.py +++ b/testing/backend/unit/test_executor.py @@ -1001,3 +1001,52 @@ async def test_execute_task_aborts_when_task_deleted_before_running(setup_test_e assert task_id not in executor.running_tasks await db.disconnect() + + +@pytest.mark.asyncio +async def test_execute_command_redacts_secrets_before_broadcast(): + executor = TaskExecutor() + task_id = "test-sse-redaction-1624" + queue = executor.subscribe(task_id) + + async def mock_readline(): + for line in [ + b"Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.secret123\n", + b"port 80 is open\n", + b"api_key=supersecretkey12345\n", + None, + ]: + yield line + + mock_stdout = AsyncMock() + mock_stdout.readline = mock_readline().__anext__ + mock_stdout.at_eof = AsyncMock(side_effect=[False, False, False, True]) + + with patch.object(executor, "_process_pids", {}): + process = AsyncMock() + process.stdout = mock_stdout + process.returncode = 0 + + async def fake_create_subprocess(*args, **kwargs): + return process + + with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess): + output, exit_code = await executor._execute_command(["echo", "test"], task_id, timeout=30) + + events = [] + while not queue.empty(): + events.append(queue.get_nowait()) + + output_events = [e for e in events if e["type"] == "output"] + assert len(output_events) == 2 + + assert "[REDACTED]" in output_events[0]["data"] + assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output_events[0]["data"] + assert "Authorization: Bearer" in output_events[0]["data"] + + assert "[REDACTED]" in output_events[1]["data"] + assert "supersecretkey12345" not in output_events[1]["data"] + + assert "[REDACTED]" in output + assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output + assert "supersecretkey12345" not in output From add393c1a464298c4de1519d8fbcebf8aae2a012 Mon Sep 17 00:00:00 2001 From: ionfwsrijan <201338831+ionfwsrijan@users.noreply.github.com> Date: Mon, 6 Jul 2026 01:17:42 +0530 Subject: [PATCH 2/7] checkpoint: reset test_executor.py to upstream/main --- testing/backend/unit/test_executor.py | 49 --------------------------- 1 file changed, 49 deletions(-) diff --git a/testing/backend/unit/test_executor.py b/testing/backend/unit/test_executor.py index 6844fdb30..bf368e33f 100644 --- a/testing/backend/unit/test_executor.py +++ b/testing/backend/unit/test_executor.py @@ -1014,55 +1014,6 @@ async def test_execute_task_aborts_when_task_deleted_before_running(setup_test_e await db.disconnect() -@pytest.mark.asyncio -async def test_execute_command_redacts_secrets_before_broadcast(): - executor = TaskExecutor() - task_id = "test-sse-redaction-1624" - queue = executor.subscribe(task_id) - - async def mock_readline(): - for line in [ - b"Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.secret123\n", - b"port 80 is open\n", - b"api_key=supersecretkey12345\n", - None, - ]: - yield line - - mock_stdout = AsyncMock() - mock_stdout.readline = mock_readline().__anext__ - mock_stdout.at_eof = AsyncMock(side_effect=[False, False, False, True]) - - with patch.object(executor, "_process_pids", {}): - process = AsyncMock() - process.stdout = mock_stdout - process.returncode = 0 - - async def fake_create_subprocess(*args, **kwargs): - return process - - with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess): - output, exit_code = await executor._execute_command(["echo", "test"], task_id, timeout=30) - - events = [] - while not queue.empty(): - events.append(queue.get_nowait()) - - output_events = [e for e in events if e["type"] == "output"] - assert len(output_events) == 2 - - assert "[REDACTED]" in output_events[0]["data"] - assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output_events[0]["data"] - assert "Authorization: Bearer" in output_events[0]["data"] - - assert "[REDACTED]" in output_events[1]["data"] - assert "supersecretkey12345" not in output_events[1]["data"] - - assert "[REDACTED]" in output - assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output - assert "supersecretkey12345" not in output - - # --------------------------------------------------------------------------- # Regression: URL inputs must not be replaced with pinned IP # --------------------------------------------------------------------------- From 95347bd1833f64ea50ab5cfef4bfd7f2d339cdea Mon Sep 17 00:00:00 2001 From: ionfwsrijan <201338831+ionfwsrijan@users.noreply.github.com> Date: Mon, 6 Jul 2026 01:25:21 +0530 Subject: [PATCH 3/7] fix: resolve merge conflict in test_executor.py between pinned-ip and atomicity tests --- testing/backend/unit/test_executor.py | 158 ++++++++++++++++++++++++++ 1 file changed, 158 insertions(+) diff --git a/testing/backend/unit/test_executor.py b/testing/backend/unit/test_executor.py index bf368e33f..a7fc917fc 100644 --- a/testing/backend/unit/test_executor.py +++ b/testing/backend/unit/test_executor.py @@ -1090,6 +1090,164 @@ async def fake_command(*args, **kwargs): await db.disconnect() +# --------------------------------------------------------------------------- +# Atomicity regression tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_transaction_rolls_back_findings_on_failure(setup_test_environment): + await init_db(settings.database_path) + db = await get_db() + + task_id = str(uuid.uuid4()) + owner_id = str(uuid.uuid4()) + await db.execute( + "INSERT INTO tasks (id, owner_id, plugin_id, tool_name, target, inputs_json, status, consent_granted, safe_mode) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (task_id, owner_id, "nmap", "nmap", "127.0.0.1", "{}", TaskStatus.RUNNING.value, 1, 0) + ) + + executor = TaskExecutor() + + mock_plugin = MagicMock() + mock_plugin.id = "nmap" + mock_plugin.name = "Nmap" + mock_plugin.output = {"parser": "builtin_nmap", "format": "text"} + mock_plugin.category = "Network" + + with patch("backend.secuscan.executor.get_plugin_manager") as mock_pm: + mock_pm.return_value.build_command.return_value = ["nmap", "127.0.0.1"] + parser = MagicMock(return_value={"findings": [ + {"title": "F1", "category": "N", "severity": "low", "description": "d1"}, + {"title": "F2", "category": "N", "severity": "low", "description": "d2"}, + ]}) + executor._parse_results = parser + + with patch.object(executor, "_persist_result_resources", + side_effect=ValueError("DB write failed")): + findings_before = await db.fetchall( + "SELECT id FROM findings WHERE task_id = ?", (task_id,) + ) + row_before = await db.fetchone( + "SELECT structured_json FROM tasks WHERE id = ?", (task_id,) + ) + + try: + await executor._upsert_findings_and_report( + db=db, task_id=task_id, owner_id=owner_id, + plugin=mock_plugin, plugin_id="nmap", + target="127.0.0.1", status="failed", output="", + ) + except ValueError: + pass + + findings_after = await db.fetchall( + "SELECT id FROM findings WHERE task_id = ?", (task_id,) + ) + row_after = await db.fetchone( + "SELECT structured_json FROM tasks WHERE id = ?", (task_id,) + ) + + assert len(findings_before) == 0 + assert len(findings_after) == 0 + assert row_before["structured_json"] is None + assert row_after["structured_json"] is None + + await db.disconnect() + + +@pytest.mark.asyncio +async def test_transaction_rolls_back_findings_and_report(setup_test_environment): + await init_db(settings.database_path) + db = await get_db() + + task_id = str(uuid.uuid4()) + owner_id = str(uuid.uuid4()) + await db.execute( + "INSERT INTO tasks (id, owner_id, plugin_id, tool_name, target, inputs_json, status, consent_granted, safe_mode) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (task_id, owner_id, "nmap", "nmap", "127.0.0.1", "{}", TaskStatus.RUNNING.value, 1, 0) + ) + + executor = TaskExecutor() + + mock_plugin = MagicMock() + mock_plugin.id = "nmap" + mock_plugin.name = "Nmap" + mock_plugin.output = {"parser": "builtin_nmap", "format": "text"} + mock_plugin.category = "Network" + + with patch("backend.secuscan.executor.get_plugin_manager") as mock_pm: + mock_pm.return_value.build_command.return_value = ["nmap", "127.0.0.1"] + parser = MagicMock(return_value={"findings": [ + {"title": "F1", "category": "N", "severity": "low", "description": "d1"}, + ]}) + executor._parse_results = parser + + with patch.object(executor, "_persist_result_resources", + side_effect=ValueError("Asset service write failed")): + try: + await executor._upsert_findings_and_report( + db=db, task_id=task_id, owner_id=owner_id, + plugin=mock_plugin, plugin_id="nmap", + target="127.0.0.1", status="failed", output="", + ) + except ValueError: + pass + + findings = await db.fetchall( + "SELECT id FROM findings WHERE task_id = ?", (task_id,) + ) + reports = await db.fetchall( + "SELECT id FROM reports WHERE task_id = ?", (task_id,) + ) + assert len(findings) == 0 + assert len(reports) == 0 + + await db.disconnect() + + +@pytest.mark.asyncio +async def test_nested_transaction_rollback_on_exception(setup_test_environment): + await init_db(settings.database_path) + db = await get_db() + + task_id = str(uuid.uuid4()) + await db.execute( + "INSERT INTO tasks (id, owner_id, plugin_id, tool_name, target, inputs_json, status, consent_granted, safe_mode) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + (task_id, "default", "test", "test", "t", "{}", TaskStatus.QUEUED.value, 1, 1) + ) + + assert db._in_transaction is False + await db.begin() + assert db._in_transaction is True + + await db.execute( + "INSERT INTO findings (id, owner_id, task_id, plugin_id, title, category, severity, target, description) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("f1", "default", task_id, "test", "Outer", "T", "low", "t", "outer finding") + ) + + assert db._in_transaction is True + async with db.transaction(): + assert db._in_transaction is True + await db.execute( + "INSERT INTO findings (id, owner_id, task_id, plugin_id, title, category, severity, target, description) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", + ("f2", "default", task_id, "test", "Nested", "T", "low", "t", "nested finding") + ) + assert db._in_transaction is True + + await db.rollback() + assert db._in_transaction is False + + rows = await db.fetchall("SELECT id FROM findings WHERE task_id = ?", (task_id,)) + assert len(rows) == 0 + + await db.disconnect() + + @pytest.mark.asyncio async def test_pinned_ip_preserves_host_input_for_host_header_scanners(setup_test_environment): await init_db(settings.database_path) From 7f13b36ca4f6ca66dba441ed198c7b6bd969f942 Mon Sep 17 00:00:00 2001 From: ionfwsrijan <201338831+ionfwsrijan@users.noreply.github.com> Date: Mon, 6 Jul 2026 01:26:29 +0530 Subject: [PATCH 4/7] fix: update test files for new ParserSandboxError API (stderr_excerpt -> _stderr_diagnostic, Template.safe_substitute) --- testing/backend/unit/test_parser_sandbox.py | 6 +++--- testing/backend/unit/test_parser_sandbox_timeout_cleanup.py | 2 +- testing/backend/unit/test_plugin_validator.py | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/testing/backend/unit/test_parser_sandbox.py b/testing/backend/unit/test_parser_sandbox.py index 6667b6b9f..2b1ce0c89 100644 --- a/testing/backend/unit/test_parser_sandbox.py +++ b/testing/backend/unit/test_parser_sandbox.py @@ -197,7 +197,7 @@ def parse(output): ) with pytest.raises(ParserSandboxError) as exc_info: run_parser_in_sandbox(p, "verbose_crash", "data") - assert "detailed crash info" in exc_info.value.stderr_excerpt + assert "detailed crash info" in exc_info.value._stderr_diagnostic def test_syntax_error_in_parser_raises(self, tmp_path): p = tmp_path / "parser.py" @@ -396,9 +396,9 @@ def test_reason_stored(self): err = ParserSandboxError("plugin_x", "custom reason") assert err.reason == "custom reason" - def test_stderr_excerpt_truncated_to_2000_chars(self): + def test__stderr_diagnostic_truncated_to_2000_chars(self): err = ParserSandboxError("p", "r", stderr="x" * 5000) - assert len(err.stderr_excerpt) == 2000 + assert len(err._stderr_diagnostic) == 2000 def test_str_contains_plugin_id(self): err = ParserSandboxError("my_plugin", "bad thing") diff --git a/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py b/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py index 9ae679d0e..1bc65dd4d 100644 --- a/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py +++ b/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py @@ -34,7 +34,7 @@ def test_timeout_expired_raises_parser_sandbox_error(self, tmp_path): with pytest.raises(subprocess.TimeoutExpired): proc = subprocess.Popen( - ["python3", "-c", _BOOTSTRAP_TEMPLATE.format( + ["python3", "-c", _BOOTSTRAP_TEMPLATE.safe_substitute( parser_path=str(parser), max_input_bytes=1024, )], diff --git a/testing/backend/unit/test_plugin_validator.py b/testing/backend/unit/test_plugin_validator.py index 4d3db8767..3d42c6e61 100644 --- a/testing/backend/unit/test_plugin_validator.py +++ b/testing/backend/unit/test_plugin_validator.py @@ -709,7 +709,7 @@ def test_sandbox_exec_fails_on_exec_statement(self): plugin_id="forbidden_parser_plugin", parser_input="test input" ) - assert "ValueError" in exc_info.value.stderr_excerpt or "exec" in exc_info.value.stderr_excerpt or "sandbox exec test" in exc_info.value.stderr_excerpt + assert "ValueError" in exc_info.value._stderr_diagnostic or "exec" in exc_info.value._stderr_diagnostic or "sandbox exec test" in exc_info.value._stderr_diagnostic def test_sandbox_strips_environ_secrets(self, monkeypatch): """Parser sandbox environment variables should be stripped to avoid secret leakage.""" From 2c8d7940719fbd73f93b61fc65b8c28761139569 Mon Sep 17 00:00:00 2001 From: ionfwsrijan <201338831+ionfwsrijan@users.noreply.github.com> Date: Mon, 6 Jul 2026 01:34:49 +0530 Subject: [PATCH 5/7] fix: add missing SSE redaction test from original PR #1624 commit --- testing/backend/unit/test_executor.py | 49 +++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/testing/backend/unit/test_executor.py b/testing/backend/unit/test_executor.py index a7fc917fc..fa0d44917 100644 --- a/testing/backend/unit/test_executor.py +++ b/testing/backend/unit/test_executor.py @@ -1014,6 +1014,55 @@ async def test_execute_task_aborts_when_task_deleted_before_running(setup_test_e await db.disconnect() +@pytest.mark.asyncio +async def test_execute_command_redacts_secrets_before_broadcast(): + executor = TaskExecutor() + task_id = "test-sse-redaction-1624" + queue = executor.subscribe(task_id) + + async def mock_readline(): + for line in [ + b"Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.secret123\n", + b"port 80 is open\n", + b"api_key=supersecretkey12345\n", + None, + ]: + yield line + + mock_stdout = AsyncMock() + mock_stdout.readline = mock_readline().__anext__ + mock_stdout.at_eof = AsyncMock(side_effect=[False, False, False, True]) + + with patch.object(executor, "_process_pids", {}): + process = AsyncMock() + process.stdout = mock_stdout + process.returncode = 0 + + async def fake_create_subprocess(*args, **kwargs): + return process + + with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess): + output, exit_code = await executor._execute_command(["echo", "test"], task_id, timeout=30) + + events = [] + while not queue.empty(): + events.append(queue.get_nowait()) + + output_events = [e for e in events if e["type"] == "output"] + assert len(output_events) == 2 + + assert "[REDACTED]" in output_events[0]["data"] + assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output_events[0]["data"] + assert "Authorization: Bearer" in output_events[0]["data"] + + assert "[REDACTED]" in output_events[1]["data"] + assert "supersecretkey12345" not in output_events[1]["data"] + + assert "[REDACTED]" in output + assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output + assert "supersecretkey12345" not in output + + # --------------------------------------------------------------------------- # Regression: URL inputs must not be replaced with pinned IP # --------------------------------------------------------------------------- From 8cd9f55d8049de3aa783084bee68ae98f6fdccb6 Mon Sep 17 00:00:00 2001 From: ionfwsrijan <201338831+ionfwsrijan@users.noreply.github.com> Date: Mon, 6 Jul 2026 01:50:37 +0530 Subject: [PATCH 6/7] restore test files to upstream/main (strip extra tests from other PRs) --- testing/backend/unit/test_executor.py | 207 ------------------ testing/backend/unit/test_parser_sandbox.py | 6 +- .../test_parser_sandbox_timeout_cleanup.py | 156 ------------- testing/backend/unit/test_plugin_validator.py | 2 +- 4 files changed, 4 insertions(+), 367 deletions(-) delete mode 100644 testing/backend/unit/test_parser_sandbox_timeout_cleanup.py diff --git a/testing/backend/unit/test_executor.py b/testing/backend/unit/test_executor.py index fa0d44917..bf368e33f 100644 --- a/testing/backend/unit/test_executor.py +++ b/testing/backend/unit/test_executor.py @@ -1014,55 +1014,6 @@ async def test_execute_task_aborts_when_task_deleted_before_running(setup_test_e await db.disconnect() -@pytest.mark.asyncio -async def test_execute_command_redacts_secrets_before_broadcast(): - executor = TaskExecutor() - task_id = "test-sse-redaction-1624" - queue = executor.subscribe(task_id) - - async def mock_readline(): - for line in [ - b"Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.secret123\n", - b"port 80 is open\n", - b"api_key=supersecretkey12345\n", - None, - ]: - yield line - - mock_stdout = AsyncMock() - mock_stdout.readline = mock_readline().__anext__ - mock_stdout.at_eof = AsyncMock(side_effect=[False, False, False, True]) - - with patch.object(executor, "_process_pids", {}): - process = AsyncMock() - process.stdout = mock_stdout - process.returncode = 0 - - async def fake_create_subprocess(*args, **kwargs): - return process - - with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess): - output, exit_code = await executor._execute_command(["echo", "test"], task_id, timeout=30) - - events = [] - while not queue.empty(): - events.append(queue.get_nowait()) - - output_events = [e for e in events if e["type"] == "output"] - assert len(output_events) == 2 - - assert "[REDACTED]" in output_events[0]["data"] - assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output_events[0]["data"] - assert "Authorization: Bearer" in output_events[0]["data"] - - assert "[REDACTED]" in output_events[1]["data"] - assert "supersecretkey12345" not in output_events[1]["data"] - - assert "[REDACTED]" in output - assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output - assert "supersecretkey12345" not in output - - # --------------------------------------------------------------------------- # Regression: URL inputs must not be replaced with pinned IP # --------------------------------------------------------------------------- @@ -1139,164 +1090,6 @@ async def fake_command(*args, **kwargs): await db.disconnect() -# --------------------------------------------------------------------------- -# Atomicity regression tests -# --------------------------------------------------------------------------- - -@pytest.mark.asyncio -async def test_transaction_rolls_back_findings_on_failure(setup_test_environment): - await init_db(settings.database_path) - db = await get_db() - - task_id = str(uuid.uuid4()) - owner_id = str(uuid.uuid4()) - await db.execute( - "INSERT INTO tasks (id, owner_id, plugin_id, tool_name, target, inputs_json, status, consent_granted, safe_mode) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (task_id, owner_id, "nmap", "nmap", "127.0.0.1", "{}", TaskStatus.RUNNING.value, 1, 0) - ) - - executor = TaskExecutor() - - mock_plugin = MagicMock() - mock_plugin.id = "nmap" - mock_plugin.name = "Nmap" - mock_plugin.output = {"parser": "builtin_nmap", "format": "text"} - mock_plugin.category = "Network" - - with patch("backend.secuscan.executor.get_plugin_manager") as mock_pm: - mock_pm.return_value.build_command.return_value = ["nmap", "127.0.0.1"] - parser = MagicMock(return_value={"findings": [ - {"title": "F1", "category": "N", "severity": "low", "description": "d1"}, - {"title": "F2", "category": "N", "severity": "low", "description": "d2"}, - ]}) - executor._parse_results = parser - - with patch.object(executor, "_persist_result_resources", - side_effect=ValueError("DB write failed")): - findings_before = await db.fetchall( - "SELECT id FROM findings WHERE task_id = ?", (task_id,) - ) - row_before = await db.fetchone( - "SELECT structured_json FROM tasks WHERE id = ?", (task_id,) - ) - - try: - await executor._upsert_findings_and_report( - db=db, task_id=task_id, owner_id=owner_id, - plugin=mock_plugin, plugin_id="nmap", - target="127.0.0.1", status="failed", output="", - ) - except ValueError: - pass - - findings_after = await db.fetchall( - "SELECT id FROM findings WHERE task_id = ?", (task_id,) - ) - row_after = await db.fetchone( - "SELECT structured_json FROM tasks WHERE id = ?", (task_id,) - ) - - assert len(findings_before) == 0 - assert len(findings_after) == 0 - assert row_before["structured_json"] is None - assert row_after["structured_json"] is None - - await db.disconnect() - - -@pytest.mark.asyncio -async def test_transaction_rolls_back_findings_and_report(setup_test_environment): - await init_db(settings.database_path) - db = await get_db() - - task_id = str(uuid.uuid4()) - owner_id = str(uuid.uuid4()) - await db.execute( - "INSERT INTO tasks (id, owner_id, plugin_id, tool_name, target, inputs_json, status, consent_granted, safe_mode) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (task_id, owner_id, "nmap", "nmap", "127.0.0.1", "{}", TaskStatus.RUNNING.value, 1, 0) - ) - - executor = TaskExecutor() - - mock_plugin = MagicMock() - mock_plugin.id = "nmap" - mock_plugin.name = "Nmap" - mock_plugin.output = {"parser": "builtin_nmap", "format": "text"} - mock_plugin.category = "Network" - - with patch("backend.secuscan.executor.get_plugin_manager") as mock_pm: - mock_pm.return_value.build_command.return_value = ["nmap", "127.0.0.1"] - parser = MagicMock(return_value={"findings": [ - {"title": "F1", "category": "N", "severity": "low", "description": "d1"}, - ]}) - executor._parse_results = parser - - with patch.object(executor, "_persist_result_resources", - side_effect=ValueError("Asset service write failed")): - try: - await executor._upsert_findings_and_report( - db=db, task_id=task_id, owner_id=owner_id, - plugin=mock_plugin, plugin_id="nmap", - target="127.0.0.1", status="failed", output="", - ) - except ValueError: - pass - - findings = await db.fetchall( - "SELECT id FROM findings WHERE task_id = ?", (task_id,) - ) - reports = await db.fetchall( - "SELECT id FROM reports WHERE task_id = ?", (task_id,) - ) - assert len(findings) == 0 - assert len(reports) == 0 - - await db.disconnect() - - -@pytest.mark.asyncio -async def test_nested_transaction_rollback_on_exception(setup_test_environment): - await init_db(settings.database_path) - db = await get_db() - - task_id = str(uuid.uuid4()) - await db.execute( - "INSERT INTO tasks (id, owner_id, plugin_id, tool_name, target, inputs_json, status, consent_granted, safe_mode) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - (task_id, "default", "test", "test", "t", "{}", TaskStatus.QUEUED.value, 1, 1) - ) - - assert db._in_transaction is False - await db.begin() - assert db._in_transaction is True - - await db.execute( - "INSERT INTO findings (id, owner_id, task_id, plugin_id, title, category, severity, target, description) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - ("f1", "default", task_id, "test", "Outer", "T", "low", "t", "outer finding") - ) - - assert db._in_transaction is True - async with db.transaction(): - assert db._in_transaction is True - await db.execute( - "INSERT INTO findings (id, owner_id, task_id, plugin_id, title, category, severity, target, description) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", - ("f2", "default", task_id, "test", "Nested", "T", "low", "t", "nested finding") - ) - assert db._in_transaction is True - - await db.rollback() - assert db._in_transaction is False - - rows = await db.fetchall("SELECT id FROM findings WHERE task_id = ?", (task_id,)) - assert len(rows) == 0 - - await db.disconnect() - - @pytest.mark.asyncio async def test_pinned_ip_preserves_host_input_for_host_header_scanners(setup_test_environment): await init_db(settings.database_path) diff --git a/testing/backend/unit/test_parser_sandbox.py b/testing/backend/unit/test_parser_sandbox.py index 2b1ce0c89..6667b6b9f 100644 --- a/testing/backend/unit/test_parser_sandbox.py +++ b/testing/backend/unit/test_parser_sandbox.py @@ -197,7 +197,7 @@ def parse(output): ) with pytest.raises(ParserSandboxError) as exc_info: run_parser_in_sandbox(p, "verbose_crash", "data") - assert "detailed crash info" in exc_info.value._stderr_diagnostic + assert "detailed crash info" in exc_info.value.stderr_excerpt def test_syntax_error_in_parser_raises(self, tmp_path): p = tmp_path / "parser.py" @@ -396,9 +396,9 @@ def test_reason_stored(self): err = ParserSandboxError("plugin_x", "custom reason") assert err.reason == "custom reason" - def test__stderr_diagnostic_truncated_to_2000_chars(self): + def test_stderr_excerpt_truncated_to_2000_chars(self): err = ParserSandboxError("p", "r", stderr="x" * 5000) - assert len(err._stderr_diagnostic) == 2000 + assert len(err.stderr_excerpt) == 2000 def test_str_contains_plugin_id(self): err = ParserSandboxError("my_plugin", "bad thing") diff --git a/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py b/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py deleted file mode 100644 index 1bc65dd4d..000000000 --- a/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py +++ /dev/null @@ -1,156 +0,0 @@ -""" -Unit tests for parser_sandbox subprocess timeout and cleanup paths. - -Covers: - - subprocess.TimeoutExpired: process is killed, ParserSandboxError raised - - thread cleanup after kill: reader threads are joined within timeout - - overflow kill: process killed before full buffer read, ParserSandboxError raised - -These are not happy-path tests (those are in test_parser_sandbox.py); these -focus on the failure/timeout paths that must not leak resources. -""" - -from __future__ import annotations - -import subprocess -import pytest -import time -from pathlib import Path -from unittest.mock import patch, MagicMock - - -class TestParserSandboxTimeoutCleanup: - """Coverage for run_parser_in_sandbox timeout kill and cleanup.""" - - def test_timeout_expired_raises_parser_sandbox_error(self, tmp_path): - """subprocess.TimeoutExpired must be caught and re-raised as ParserSandboxError.""" - from backend.secuscan.parser_sandbox import run_parser_in_sandbox, _sanitised_env - from backend.secuscan.parser_sandbox import _BOOTSTRAP_TEMPLATE - import json - - # Write a parser that sleeps long enough to be killed. - parser = tmp_path / "parser.py" - parser.write_text("import time; time.sleep(60); print('late')\n", encoding="utf-8") - - with pytest.raises(subprocess.TimeoutExpired): - proc = subprocess.Popen( - ["python3", "-c", _BOOTSTRAP_TEMPLATE.safe_substitute( - parser_path=str(parser), - max_input_bytes=1024, - )], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - env=_sanitised_env(), - ) - try: - proc.wait(timeout=1) - except subprocess.TimeoutExpired: - proc.kill() - proc.wait() - raise - - def test_timeout_kill_does_not_hang_join_threads(self, tmp_path): - """Threads reading from killed subprocess must complete within the join timeout.""" - from backend.secuscan.parser_sandbox import run_parser_in_sandbox, _sanitised_env - from backend.secuscan.parser_sandbox import _BOOTSTRAP_TEMPLATE - - parser = tmp_path / "parser.py" - parser.write_text("import time; time.sleep(60)\n", encoding="utf-8") - - start = time.monotonic() - with pytest.raises(Exception): # ParserSandboxError - run_parser_in_sandbox(parser, "slow", "data", timeout_seconds=1) - elapsed = time.monotonic() - start - - # Must not hang; total time must be well under 10 seconds. - assert elapsed < 10, f"ParserSandboxError took {elapsed:.1f}s — threads may be hanging" - - def test_timeout_error_message_contains_duration(self, tmp_path): - """ParserSandboxError message after timeout must include the duration.""" - from backend.secuscan.parser_sandbox import run_parser_in_sandbox - - parser = tmp_path / "parser.py" - parser.write_text("import time; time.sleep(60)\n", encoding="utf-8") - - with pytest.raises(Exception) as exc_info: - run_parser_in_sandbox(parser, "slow_plugin", "data", timeout_seconds=2) - - exc_message = str(exc_info.value) - assert "timed out" in exc_message or "timeout" in exc_message.lower() - - def test_timeout_error_includes_stderr(self, tmp_path): - """ParserSandboxError after timeout must include stderr output.""" - from backend.secuscan.parser_sandbox import run_parser_in_sandbox - - parser = tmp_path / "parser.py" - # Write to stderr before sleeping. - parser.write_text( - "import sys, time; sys.stderr.write('early error\\n'); time.sleep(60)\n", - encoding="utf-8", - ) - - with pytest.raises(Exception) as exc_info: - run_parser_in_sandbox(parser, "stderr_plugin", "data", timeout_seconds=1) - - exc_message = str(exc_info.value) - assert "early error" in exc_message - - def test_multiple_consecutive_timeouts_do_not_leak_resources(self, tmp_path): - """Running multiple timeouts in sequence must not accumulate open file handles.""" - from backend.secuscan.parser_sandbox import run_parser_in_sandbox - import resource - - parser = tmp_path / "parser.py" - parser.write_text("import time; time.sleep(60)\n", encoding="utf-8") - - for _ in range(3): - with pytest.raises(Exception): - run_parser_in_sandbox(parser, "repeat", "data", timeout_seconds=1) - - # Get current process file descriptor count. - soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) - # If we are leaking fds, the count would grow. This is a rough check. - import os as _os - open_fds_before = len(_os.listdir(f"/proc/{_os.getpid()}/fd")) - # Run one more timeout. - with pytest.raises(Exception): - run_parser_in_sandbox(parser, "final", "data", timeout_seconds=1) - open_fds_after = len(_os.listdir(f"/proc/{_os.getpid()}/fd")) - assert open_fds_after <= open_fds_before + 1, "File descriptors may be leaking after timeout" - - -class TestParserSandboxOverflowCleanup: - """Coverage for run_parser_in_sandbox output-overflow kill and cleanup.""" - - def test_overflow_kill_does_not_hang(self, tmp_path): - """A parser that writes more than max_output_bytes must be killed promptly.""" - from backend.secuscan.parser_sandbox import run_parser_in_sandbox - - parser = tmp_path / "parser.py" - # Write a parser that outputs 10 MB of data. - parser.write_text( - f"import sys; sys.stdout.write('x' * (10 * 1024 * 1024))\n", - encoding="utf-8", - ) - - start = time.monotonic() - with pytest.raises(Exception) as exc_info: - run_parser_in_sandbox(parser, "big_output", "data", max_output_bytes=1024) - elapsed = time.monotonic() - start - - # Must complete quickly without waiting for full 10 MB to be written. - assert elapsed < 10, f"Overflow kill took {elapsed:.1f}s — may be reading full output" - assert "limit" in str(exc_info.value).lower() or "output" in str(exc_info.value).lower() - - def test_overflow_error_message_contains_limit(self, tmp_path): - """ParserSandboxError after overflow must reference the size limit.""" - from backend.secuscan.parser_sandbox import run_parser_in_sandbox - - parser = tmp_path / "parser.py" - parser.write_text("import sys; sys.stdout.write('x' * 10_000_000)\n", encoding="utf-8") - - with pytest.raises(Exception) as exc_info: - run_parser_in_sandbox(parser, "overflow", "data", max_output_bytes=512) - - assert "limit" in str(exc_info.value).lower() or "exceeded" in str(exc_info.value).lower() diff --git a/testing/backend/unit/test_plugin_validator.py b/testing/backend/unit/test_plugin_validator.py index 3d42c6e61..4d3db8767 100644 --- a/testing/backend/unit/test_plugin_validator.py +++ b/testing/backend/unit/test_plugin_validator.py @@ -709,7 +709,7 @@ def test_sandbox_exec_fails_on_exec_statement(self): plugin_id="forbidden_parser_plugin", parser_input="test input" ) - assert "ValueError" in exc_info.value._stderr_diagnostic or "exec" in exc_info.value._stderr_diagnostic or "sandbox exec test" in exc_info.value._stderr_diagnostic + assert "ValueError" in exc_info.value.stderr_excerpt or "exec" in exc_info.value.stderr_excerpt or "sandbox exec test" in exc_info.value.stderr_excerpt def test_sandbox_strips_environ_secrets(self, monkeypatch): """Parser sandbox environment variables should be stripped to avoid secret leakage.""" From 7cbe9d20bf7741d8831aa7191e5ada00eaa12e15 Mon Sep 17 00:00:00 2001 From: ionfwsrijan <201338831+ionfwsrijan@users.noreply.github.com> Date: Mon, 6 Jul 2026 01:51:03 +0530 Subject: [PATCH 7/7] fix: add SSE redact test with Mock for at_eof (sync call, not async) --- testing/backend/unit/test_executor.py | 49 +++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/testing/backend/unit/test_executor.py b/testing/backend/unit/test_executor.py index bf368e33f..d4dd1dfda 100644 --- a/testing/backend/unit/test_executor.py +++ b/testing/backend/unit/test_executor.py @@ -1158,3 +1158,52 @@ async def fake_command(*args, **kwargs): assert original_inputs_seen.get("domain") == "example.com" assert original_inputs_seen.get("__pinned_ip") == "93.184.216.34" await db.disconnect() + + +@pytest.mark.asyncio +async def test_execute_command_redacts_secrets_before_broadcast(): + executor = TaskExecutor() + task_id = "test-sse-redaction-1624" + queue = executor.subscribe(task_id) + + async def mock_readline(): + for line in [ + b"Authorization: Bearer eyJhbGciOiJIUzI1NiJ9.secret123\n", + b"port 80 is open\n", + b"api_key=supersecretkey12345\n", + None, + ]: + yield line + + mock_stdout = AsyncMock() + mock_stdout.readline = mock_readline().__anext__ + mock_stdout.at_eof = MagicMock(side_effect=[False, False, False, True]) + + with patch.object(executor, "_process_pids", {}): + process = AsyncMock() + process.stdout = mock_stdout + process.returncode = 0 + + async def fake_create_subprocess(*args, **kwargs): + return process + + with patch("asyncio.create_subprocess_exec", side_effect=fake_create_subprocess): + output, exit_code = await executor._execute_command(["echo", "test"], task_id, timeout=30) + + events = [] + while not queue.empty(): + events.append(queue.get_nowait()) + + output_events = [e for e in events if e["type"] == "output"] + assert len(output_events) == 2 + + assert "[REDACTED]" in output_events[0]["data"] + assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output_events[0]["data"] + assert "Authorization: Bearer" in output_events[0]["data"] + + assert "[REDACTED]" in output_events[1]["data"] + assert "supersecretkey12345" not in output_events[1]["data"] + + assert "[REDACTED]" in output + assert "eyJhbGciOiJIUzI1NiJ9.secret123" not in output + assert "supersecretkey12345" not in output