diff --git a/backend/secuscan/cache.py b/backend/secuscan/cache.py index 177b49675..46b81c2c8 100644 --- a/backend/secuscan/cache.py +++ b/backend/secuscan/cache.py @@ -1,5 +1,5 @@ """ -In-memory cache helpers for API responses. +In-memory and Redis-based cache helpers for API responses. """ import json @@ -15,9 +15,15 @@ SWEEP_EVICT_FRACTION = 0.25 OPPORTUNISTIC_SWEEP_INTERVAL = 50 +try: + from redis.asyncio import Redis, ConnectionError as RedisConnectionError +except ImportError: + Redis = None + RedisConnectionError = Exception + class CacheClient: - """In-memory dictionary based cache client with TTL, size limit, and LRU eviction.""" + """Cache client supporting Redis with an in-memory dictionary fallback.""" def __init__(self, url: Optional[str] = None, max_entries: int = DEFAULT_MAX_ENTRIES): self.url = url @@ -28,11 +34,27 @@ def __init__(self, url: Optional[str] = None, max_entries: int = DEFAULT_MAX_ENT self._eviction_count = 0 self._sweep_count = 0 self._write_count = 0 + self.client: Optional[Redis] = None async def connect(self): - pass + if self.url and Redis is not None: + try: + self.client = Redis.from_url(self.url, decode_responses=True) + await self.client.ping() + logger.info("✓ Connected to Redis cache at %s", self.url) + except RedisConnectionError as e: + logger.warning("Failed to connect to Redis, falling back to in-memory: %s", e) + self.client = None + else: + self.client = None async def disconnect(self): + if self.client: + try: + await self.client.aclose() + except Exception: + pass + self.client = None self._data.clear() self._expires.clear() self._access_order.clear() @@ -60,7 +82,14 @@ def _evict_lru(self): self._eviction_count += evict_count async def get_json(self, key: str) -> Optional[Any]: - """Retrieve and parse JSON from memory, respecting TTL.""" + """Retrieve and parse JSON from cache, respecting TTL.""" + if self.client: + try: + val = await self.client.get(key) + return json.loads(val) if val is not None else None + except Exception as e: + logger.warning("Redis get_json error (falling back to in-memory): %s", e) + now = time.time() expiry = self._expires.get(key) @@ -76,12 +105,20 @@ async def get_json(self, key: str) -> Optional[Any]: return self._data.get(key) async def set_json(self, key: str, value: Any, ttl: Optional[int] = None): - """Store value in memory with optional TTL.""" + """Store value in cache with optional TTL.""" + actual_ttl = ttl or settings.cache_ttl_seconds + + if self.client: + try: + await self.client.set(key, json.dumps(value), ex=actual_ttl) + return + except Exception as e: + logger.warning("Redis set_json error (falling back to in-memory): %s", e) + if len(self._data) >= self.max_entries and key not in self._data: self._evict_lru() self._data[key] = value - actual_ttl = ttl or settings.cache_ttl_seconds self._expires[key] = time.time() + actual_ttl self._access_order[key] = time.time() self._write_count += 1 @@ -91,6 +128,17 @@ async def set_json(self, key: str, value: Any, ttl: Optional[int] = None): async def delete_prefix(self, prefix: str): """Delete all keys starting with prefix.""" + if self.client: + try: + keys = [] + async for key in self.client.scan_iter(match=f"{prefix}*"): + keys.append(key) + if keys: + await self.client.delete(*keys) + return + except Exception as e: + logger.warning("Redis delete_prefix error (falling back to in-memory): %s", e) + to_delete = [k for k in self._data.keys() if k.startswith(prefix)] for k in to_delete: self._data.pop(k, None) diff --git a/backend/secuscan/executor.py b/backend/secuscan/executor.py index a8523a245..0c815b49a 100644 --- a/backend/secuscan/executor.py +++ b/backend/secuscan/executor.py @@ -170,6 +170,83 @@ def _row_value(row: Any, key: str, default: Any = None) -> Any: return default +def generate_scan_cache_key( + owner_id: str, + plugin_id: str, + target: str, + inputs: Dict[str, Any], + execution_context: Dict[str, Any], + safe_mode: bool, +) -> tuple[str, str, str]: + """Generate target hash, dependency hash, and an owner-scoped cache key. + + Returns: + tuple: (target_hash, dependency_hash, cache_key) + """ + import hashlib + import subprocess + from pathlib import Path + + target_hash = None + if target and os.path.isdir(target): + try: + res = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=target, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + timeout=5, + ) + if res.returncode == 0: + target_hash = res.stdout.strip() + except Exception: + pass + + if not target_hash: + target_hash = hashlib.sha256(str(target or "").encode("utf-8")).hexdigest() + + dependency_files = [ + "package-lock.json", + "poetry.lock", + "Cargo.lock", + "go.sum", + "requirements.txt", + "Pipfile.lock", + "pnpm-lock.yaml", + "yarn.lock", + "gemfile.lock", + ] + hasher = hashlib.sha256() + found_any = False + + if target and os.path.isdir(target): + p = Path(target) + for dep_file in sorted(dependency_files): + file_path = p / dep_file + if file_path.exists() and file_path.is_file(): + try: + hasher.update(dep_file.encode("utf-8")) + hasher.update(file_path.read_bytes()) + found_any = True + except Exception: + pass + + if not found_any: + dependency_hash = "no_deps" + else: + dependency_hash = hasher.hexdigest() + + inputs_str = json.dumps(inputs, sort_keys=True) + inputs_hash = hashlib.sha256(inputs_str.encode("utf-8")).hexdigest() + + context_str = json.dumps(execution_context, sort_keys=True) + context_hash = hashlib.sha256(context_str.encode("utf-8")).hexdigest() + + cache_key = f"scan_cache:{owner_id}:{plugin_id}:{int(safe_mode)}:{target_hash}:{dependency_hash}:{inputs_hash}:{context_hash}" + return target_hash, dependency_hash, cache_key + + class TaskExecutor: """Executes security scanning tasks in isolated environments""" @@ -654,12 +731,13 @@ async def _execute_standard_scanner( await self._broadcast_phase(task_id, ScanPhase.REPORTING.value) return final_status, duration, exit_code - async def execute_task(self, task_id: str) -> None: + async def execute_task(self, task_id: str, bypass_cache: bool = False) -> None: """ Execute a task asynchronously. Args: task_id: Task identifier + bypass_cache: Whether to ignore cached results """ db = await get_db() self.running_tasks[task_id] = asyncio.current_task() @@ -726,6 +804,92 @@ async def execute_task(self, task_id: str) -> None: "Exploit-level plugins require an execution context with validation_mode set to 'proof' or 'controlled_extract'." ) + # Check cache if not bypassed + cached_result = None + cache_key = None + if target and not bypass_cache: + try: + target_hash, dependency_hash, cache_key = await asyncio.to_thread( + generate_scan_cache_key, + owner_id=owner_id, + plugin_id=plugin_id, + target=target, + inputs=inputs, + execution_context=execution_context, + safe_mode=safe_mode, + ) + cache_client = await get_cache() + cached_result = await cache_client.get_json(cache_key) + except Exception as cache_exc: + logger.warning("Failed to query scan cache: %s", cache_exc) + + if cached_result is not None: + logger.info(f"Scan cache hit for task {task_id} using key {cache_key}") + status = cached_result.get("status", TaskStatus.COMPLETED.value) + duration = cached_result.get("duration_seconds", 0.0) + exit_code = cached_result.get("exit_code", 0) + error_message = cached_result.get("error_message") + structured_data = cached_result.get("structured", {}) + raw_output = cached_result.get("raw_output", "") + + # Save raw output if present + raw_path = Path(settings.raw_output_dir) / f"{task_id}.txt" + with open(raw_path, "w") as f: + f.write(raw_output) + + await db.execute( + """ + UPDATE tasks SET + status = ?, + completed_at = ?, + duration_seconds = ?, + exit_code = ?, + raw_output_path = ?, + command_used = ?, + error_message = ? + WHERE id = ? + """, + ( + status, + datetime.now().isoformat(), + duration, + exit_code, + str(raw_path), + f"[CACHED] {plugin_id}", + error_message, + task_id, + ), + ) + + await self._broadcast_phase(task_id, ScanPhase.PARSING.value) + is_modular = plugin_id in MODULAR_SCANNERS + report_name = f"{plugin.name if hasattr(plugin, 'name') else plugin_id} Report" + await self._persist_cached_result( + db, + task_id=task_id, + owner_id=owner_id, + plugin_id=plugin_id, + target=target, + status=status, + result_dict=structured_data, + is_modular=is_modular, + report_name=report_name, + ) + + await self._dispatch_task_notifications(db, task_id) + await self._broadcast_phase(task_id, ScanPhase.FINISHED.value) + await self._broadcast(task_id, "status", status) + await self._invalidate_cached_views() + + await db.log_audit( + "task_completed", + f"Task completed from cache in {duration:.2f}s", + context={"task_id": task_id, "exit_code": exit_code, "cache_hit": True}, + task_id=task_id, + plugin_id=plugin_id + ) + return + if plugin_id in MODULAR_SCANNERS: final_status, duration = await self._execute_modular_scanner( db=db, @@ -766,6 +930,35 @@ async def execute_task(self, task_id: str) -> None: logger.info(f"Task {task_id} completed in {duration:.2f}s") + # Save to cache if successful + if final_status == TaskStatus.COMPLETED.value and cache_key: + try: + task_row = await db.fetchone( + "SELECT structured_json, error_message, exit_code FROM tasks WHERE id = ?", + (task_id,) + ) + if task_row and task_row["structured_json"]: + structured_data = json.loads(task_row["structured_json"]) + raw_path = Path(settings.raw_output_dir) / f"{task_id}.txt" + raw_output = "" + if raw_path.exists(): + with open(raw_path, "r") as f: + raw_output = f.read() + + cache_payload = { + "status": final_status, + "duration_seconds": duration, + "exit_code": exit_code, + "error_message": task_row["error_message"], + "structured": structured_data, + "raw_output": raw_output, + } + cache_client = await get_cache() + await cache_client.set_json(cache_key, cache_payload, ttl=settings.cache_ttl_seconds) + logger.info(f"Scan result saved to cache under key {cache_key}") + except Exception as cache_exc: + logger.warning("Failed to save scan to cache: %s", cache_exc) + except asyncio.CancelledError: duration = (time.time() - start_time) if 'start_time' in locals() else 0 await db.execute( @@ -1665,6 +1858,84 @@ async def _persist_result_resources( services=[item for item in asset_services if isinstance(item, dict)], ) + async def _persist_cached_result( + self, + db, + *, + task_id: str, + owner_id: str, + plugin_id: str, + target: str, + status: str, + result_dict: Dict[str, Any], + is_modular: bool, + report_name: str, + ) -> None: + """Persist cached scan findings and report records into SQLite.""" + structured_result, previous_findings, asset_services = await self._build_result_contract( + db, + task_id=task_id, + owner_id=owner_id, + plugin_id=plugin_id, + target=target, + result=result_dict, + ) + findings_data: List[Dict[str, Any]] = [] + async with db.transaction(): + for finding in structured_result.get("findings", []): + findings_data.append( + await self._persist_finding( + db, + owner_id=owner_id, + task_id=task_id, + plugin_id=plugin_id, + target=target, + finding=finding, + ) + ) + + structured_result["findings"] = findings_data + structured_result["severity_counts"] = self._build_severity_counts(findings_data) + structured_result["finding_groups"] = build_finding_groups(findings_data) + structured_result["asset_summary"] = build_asset_summary(findings_data, asset_services) + structured_result["scan_diff"] = build_scan_diff(findings_data, previous_findings) + + await db.execute( + "UPDATE tasks SET structured_json = ? WHERE id = ?", + (json.dumps(structured_result), task_id) + ) + + await db.execute( + """ + INSERT INTO reports ( + id, owner_id, task_id, name, type, generated_at, status, findings, pages + ) VALUES (?, ?, ?, ?, ?, (datetime('now')), ?, ?, ?) + ON CONFLICT (id) DO UPDATE SET + status = EXCLUDED.status, + findings = EXCLUDED.findings, + pages = EXCLUDED.pages + """, + ( + f"report:{task_id}", + owner_id, + task_id, + report_name, + "professional" if is_modular else "technical", + "ready" if status == TaskStatus.COMPLETED.value else "failed", + len(findings_data), + 2 if is_modular else 1, + ), + ) + + await self._persist_result_resources( + db, + owner_id=owner_id, + task_id=task_id, + plugin_id=plugin_id, + target=target, + result=structured_result, + ) + def _parse_results(self, plugin, output: str) -> Dict[str, Any]: """Route to appropriate parser based on plugin metadata.""" parser_type = plugin.output.get("parser") diff --git a/backend/secuscan/main.py b/backend/secuscan/main.py index 66282c415..d7a98b07b 100644 --- a/backend/secuscan/main.py +++ b/backend/secuscan/main.py @@ -70,8 +70,8 @@ async def lifespan(app: FastAPI): await init_db(settings.database_path) logger.info("✓ SQLite connected") - await init_cache() - logger.info("✓ In-memory cache initialized") + await init_cache(settings.redis_url) + logger.info("✓ Cache initialized") # ─── RATE LIMITER SETUP ────────────────────────────────────────────── # Initialize rate limiter with Redis client from cache diff --git a/backend/secuscan/parser_sandbox.py b/backend/secuscan/parser_sandbox.py index 1b5d7477b..66e3412cd 100644 --- a/backend/secuscan/parser_sandbox.py +++ b/backend/secuscan/parser_sandbox.py @@ -79,10 +79,14 @@ 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 - self.stderr_excerpt = stderr[:2000] if stderr else "" # User-facing message: reason only — no stderr content. super().__init__(f"Parser sandbox failed for '{plugin_id}' ({reason})") + @property + def stderr_excerpt(self) -> str: + """Helper property for backward compatibility with diagnostics/tests.""" + return self._stderr_diagnostic[:2000] if self._stderr_diagnostic else "" + # --------------------------------------------------------------------------- # Bootstrap script injected into the child process via -c @@ -249,6 +253,7 @@ def _read_stderr() -> None: raise ParserSandboxError( plugin_id, f"output exceeded {max_output_bytes // (1024 * 1024)} MB limit", + stderr_text, ) if timed_out: @@ -311,6 +316,7 @@ def _read_stderr() -> None: raise ParserSandboxError( plugin_id, f"parser returned unexpected type {type(parsed).__name__}; expected dict or list", + stderr_text, ) 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 f4b3289c9..7ddb5d4de 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -312,6 +312,7 @@ async def start_task( request: TaskCreateRequest, background_tasks: BackgroundTasks, raw_request: Request, + bypass_cache: bool = Query(False), owner: str = Depends(get_current_owner), ): """ @@ -486,7 +487,7 @@ async def start_task( # Use BackgroundTasks so the response can be sent without waiting in real # ASGI servers, while tests using TestClient still execute the task to keep # contract tests deterministic. - background_tasks.add_task(executor.execute_task, task_id) + background_tasks.add_task(executor.execute_task, task_id, bypass_cache=bypass_cache) await invalidate_view_cache() return { diff --git a/testing/backend/conftest.py b/testing/backend/conftest.py index da7d68448..f70696122 100644 --- a/testing/backend/conftest.py +++ b/testing/backend/conftest.py @@ -22,6 +22,12 @@ @pytest.fixture(autouse=True) def setup_test_environment(monkeypatch): """Override settings for tests to ensure isolated execution.""" + try: + from backend.secuscan import cache as cache_module + cache_module.cache = None + except ImportError: + pass + temp_dir = tempfile.TemporaryDirectory(ignore_cleanup_errors=True) temp_path = temp_dir.name diff --git a/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py b/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py index ca1b18ce3..84e604e70 100644 --- a/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py +++ b/testing/backend/unit/test_parser_sandbox_timeout_cleanup.py @@ -28,6 +28,7 @@ def test_timeout_expired_raises_parser_sandbox_error(self, tmp_path): from backend.secuscan.parser_sandbox import run_parser_in_sandbox, _sanitised_env from backend.secuscan.parser_sandbox import _BOOTSTRAP_TEMPLATE import json + import sys # Write a parser that sleeps long enough to be killed. parser = tmp_path / "parser.py" @@ -35,7 +36,7 @@ def test_timeout_expired_raises_parser_sandbox_error(self, tmp_path): with pytest.raises(subprocess.TimeoutExpired): proc = subprocess.Popen( - ["python3", "-c", _BOOTSTRAP_TEMPLATE.safe_substitute( + [sys.executable, "-c", _BOOTSTRAP_TEMPLATE.safe_substitute( parser_path_repr=repr(str(parser)), max_input_bytes=1024, )], @@ -55,6 +56,7 @@ 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 + import time parser = tmp_path / "parser.py" parser.write_text("import time; time.sleep(60)\n", encoding="utf-8") @@ -87,7 +89,7 @@ def test_timeout_error_includes_stderr(self, tmp_path): 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", + "import sys, time; sys.stderr.write('early error\n'); time.sleep(60)\n", encoding="utf-8", ) diff --git a/testing/backend/unit/test_scan_cache.py b/testing/backend/unit/test_scan_cache.py new file mode 100644 index 000000000..e1180157a --- /dev/null +++ b/testing/backend/unit/test_scan_cache.py @@ -0,0 +1,311 @@ +import os +import json +import shutil +import tempfile +import pytest +import asyncio +from unittest.mock import AsyncMock, patch, MagicMock, ANY + +from backend.secuscan.executor import generate_scan_cache_key, TaskExecutor +from backend.secuscan.cache import init_cache, get_cache +from backend.secuscan.models import TaskStatus +from backend.secuscan.execution_context import normalize_execution_context + + +@pytest.fixture +def temp_repo(): + # Create a temporary directory structure representing a project + temp_dir = tempfile.mkdtemp() + yield temp_dir + shutil.rmtree(temp_dir) + + +def test_generate_scan_cache_key_no_repo(temp_repo): + # If no git or dependency files exist, it hashes target string + target_hash, dep_hash, key = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo}, + execution_context={}, + safe_mode=False, + ) + assert len(target_hash) == 64 + assert dep_hash == "no_deps" + assert key.startswith("scan_cache:owner_1:test_plugin:0:") + + +def test_generate_scan_cache_key_with_deps(temp_repo): + # Create package-lock.json + dep_file = os.path.join(temp_repo, "package-lock.json") + with open(dep_file, "w") as f: + f.write("npm-deps-v1") + + target_hash, dep_hash, key = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo}, + execution_context={}, + safe_mode=False, + ) + assert len(target_hash) == 64 + assert dep_hash != "no_deps" + + # Modify package-lock.json -> dependency hash changes! + with open(dep_file, "w") as f: + f.write("npm-deps-v2") + + target_hash_2, dep_hash_2, key_2 = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo}, + execution_context={}, + safe_mode=False, + ) + assert dep_hash != dep_hash_2 + assert key != key_2 + + +def test_cache_key_tenant_isolation(temp_repo): + # Same inputs/target, different owners -> different cache keys! + _, _, key_owner1 = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo, "flag": "x"}, + execution_context={"profile": "admin"}, + safe_mode=False, + ) + _, _, key_owner2 = generate_scan_cache_key( + owner_id="owner_2", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo, "flag": "x"}, + execution_context={"profile": "admin"}, + safe_mode=False, + ) + assert key_owner1 != key_owner2 + + +def test_cache_key_inputs_isolation(temp_repo): + # Same target/owner, different inputs/flags -> different cache keys! + _, _, key_inputs1 = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo, "wordlist": "common.txt"}, + execution_context={}, + safe_mode=False, + ) + _, _, key_inputs2 = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo, "wordlist": "deep.txt"}, + execution_context={}, + safe_mode=False, + ) + assert key_inputs1 != key_inputs2 + + +def test_cache_key_safe_mode_isolation(temp_repo): + # Same inputs/owner, safe_mode toggled -> different cache keys! + _, _, key_safe = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo}, + execution_context={}, + safe_mode=True, + ) + _, _, key_unsafe = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs={"target": temp_repo}, + execution_context={}, + safe_mode=False, + ) + assert key_unsafe != key_safe + + +@pytest.mark.asyncio +async def test_execute_task_cache_hit(temp_repo): + # Initialize in-memory cache + await init_cache() + + # We will mock the database and task run details using a SQL-inspecting side effect + mock_db = AsyncMock() + mock_db.transaction = MagicMock(return_value=AsyncMock()) + + async def db_fetchone_mock(query, params=()): + query_lower = query.lower() + if "select owner_id, plugin_id" in query_lower: + return { + "owner_id": "owner_1", + "plugin_id": "test_plugin", + "inputs_json": json.dumps({"target": temp_repo}), + "execution_context_json": "{}", + "safe_mode": False, + } + return None + + mock_db.fetchone = AsyncMock(side_effect=db_fetchone_mock) + + executor = TaskExecutor() + + # Pre-populate cache for this target/owner/inputs/context/safe_mode + # Note: inputs is hydrated inside execute_task to contain normalized execution_context + execution_context = normalize_execution_context({}) + inputs = {"target": temp_repo, "__execution_context": execution_context} + target_hash, dep_hash, cache_key = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs=inputs, + execution_context=execution_context, + safe_mode=False, + ) + cache_client = await get_cache() + + cached_data = { + "status": TaskStatus.COMPLETED.value, + "duration_seconds": 1.5, + "exit_code": 0, + "error_message": None, + "raw_output": "cached output text", + "structured": { + "findings": [ + { + "title": "Cached Finding", + "category": "Code Security", + "severity": "high", + "description": "Cached desc", + } + ] + }, + } + await cache_client.set_json(cache_key, cached_data) + + # We mock internal helper methods + executor._persist_finding = AsyncMock(return_value={"id": "finding_1"}) + executor._persist_result_resources = AsyncMock() + executor._dispatch_task_notifications = AsyncMock() + executor._invalidate_cached_views = AsyncMock() + executor._execute_command = AsyncMock() + + with ( + patch("backend.secuscan.executor.get_db", return_value=mock_db), + patch("backend.secuscan.executor.get_plugin_manager") as mock_pm, + ): + mock_plugin = MagicMock() + mock_plugin.name = "Test Plugin" + mock_pm.return_value.get_plugin.return_value = mock_plugin + + await executor.execute_task("task_id_123", bypass_cache=False) + + # Verify _execute_command was never called (regression test for cache bypass) + executor._execute_command.assert_not_called() + + mock_db.execute.assert_any_call( + """ + UPDATE tasks SET + status = ?, + completed_at = ?, + duration_seconds = ?, + exit_code = ?, + raw_output_path = ?, + command_used = ?, + error_message = ? + WHERE id = ? + """, + (TaskStatus.COMPLETED.value, ANY, 1.5, 0, ANY, "[CACHED] test_plugin", None, "task_id_123"), + ) + + # Verify it persisted the cached findings and updated structured_json + executor._persist_finding.assert_called_once() + mock_db.execute.assert_any_call( + "UPDATE tasks SET structured_json = ? WHERE id = ?", (ANY, "task_id_123") + ) + + +@pytest.mark.asyncio +async def test_execute_task_transient_failure_not_cached(temp_repo): + # Initialize in-memory cache + await init_cache() + + mock_db = AsyncMock() + + async def db_fetchone_mock(query, params=()): + query_lower = query.lower() + if "select owner_id, plugin_id" in query_lower: + return { + "owner_id": "owner_1", + "plugin_id": "test_plugin", + "inputs_json": json.dumps({"target": temp_repo}), + "execution_context_json": "{}", + "safe_mode": False, + } + if "select status, duration_seconds" in query_lower: + return { + "status": TaskStatus.FAILED.value, + "duration_seconds": 2.0, + "exit_code": 1, + "error_message": "Transient network timeout", + "structured_json": None, + "raw_output_path": None, + } + return None + + mock_db.fetchone = AsyncMock(side_effect=db_fetchone_mock) + + executor = TaskExecutor() + executor._persist_findings_and_report_common = AsyncMock() + executor._dispatch_task_notifications = AsyncMock() + executor._invalidate_cached_views = AsyncMock() + + # Stub the actually executed command to fail + async def fake_command(*args, **kwargs): + return "Network timeout", 1 + + execution_context = normalize_execution_context({}) + inputs = {"target": temp_repo, "__execution_context": execution_context} + _, _, cache_key = generate_scan_cache_key( + owner_id="owner_1", + plugin_id="test_plugin", + target=temp_repo, + inputs=inputs, + execution_context=execution_context, + safe_mode=False, + ) + + with ( + patch("backend.secuscan.executor.get_db", return_value=mock_db), + patch.object(executor, "_execute_command", side_effect=fake_command), + patch("backend.secuscan.executor.get_plugin_manager") as mock_pm, + ): + mock_plugin = MagicMock() + mock_plugin.name = "Test Plugin" + mock_plugin.presets = {} + mock_plugin.docker_image = None + mock_plugin.output = {"parser": "builtin_nmap", "format": "text"} + mock_plugin.category = "Network" + mock_plugin.id = "test_plugin" + + mock_pm.return_value.get_plugin.return_value = mock_plugin + mock_pm.return_value.build_command.return_value = ["ping", temp_repo] + mock_pm.return_value.plugins_dir = MagicMock() + mock_pm.return_value.plugins_dir.__truediv__ = MagicMock( + return_value=MagicMock( + __truediv__=MagicMock(return_value=MagicMock(exists=lambda: False)) + ) + ) + + await executor.execute_task("task_id_456", bypass_cache=False) + + # The cache should be empty for this key because the task status is FAILED + cache_client = await get_cache() + cached_val = await cache_client.get_json(cache_key) + assert cached_val is None