From 70ecee5acb93a739dab9f85a1e6855a8ca0d9d5d Mon Sep 17 00:00:00 2001 From: ionfwsrijan Date: Sun, 21 Jun 2026 19:35:37 +0530 Subject: [PATCH 1/7] fix(#1137): add transaction context manager for atomic multi-write operations - Add Database.transaction() async context manager (begin/commit/rollback) - Wrap _upsert_findings_and_report in transaction (findings + report + resources) - Wrap _upsert_findings_and_report_from_scanner in transaction - Wrap cancel_task status update + audit log in transaction - Wrap replace_asset_services delete+insert in transaction - Wrap delete_task_records in transaction (replacing manual begin/commit/rollback) - Wrap upsert_vault_secret select-then-upsert in transaction - Keep existing execute() auto-commit behavior for backward compatibility --- backend/secuscan/database.py | 24 +++- backend/secuscan/executor.py | 157 +++++++++++++------------ backend/secuscan/platform_resources.py | 5 +- backend/secuscan/routes.py | 51 +++----- 4 files changed, 121 insertions(+), 116 deletions(-) diff --git a/backend/secuscan/database.py b/backend/secuscan/database.py index 0b2d02798..42bb36b80 100644 --- a/backend/secuscan/database.py +++ b/backend/secuscan/database.py @@ -3,10 +3,11 @@ """ import asyncio +import contextlib import json import sqlite3 from pathlib import Path -from typing import Any, Optional, List, Dict +from typing import Any, Optional, List, Dict, AsyncIterator import aiosqlite from .config import settings @@ -701,6 +702,27 @@ async def _backfill_risk_scores(self): ) print(f"Backfilled risk scores for {len(rows)} existing finding(s).") + @contextlib.asynccontextmanager + async def transaction(self) -> AsyncIterator["Database"]: + """Context manager for atomic transactions. + + Usage:: + + async with db.transaction(): + await db.execute("INSERT INTO ...") + await db.execute("UPDATE ...") + + If any statement raises, the entire transaction is rolled back. + On success the transaction is committed automatically. + """ + await self.begin() + try: + yield self + await self.commit() + except Exception: + await self.rollback() + raise + async def execute(self, query: str, params: tuple = ()): """Execute a write query and return the cursor (so callers can inspect rowcount).""" cursor = await self.connection.execute(query, params) diff --git a/backend/secuscan/executor.py b/backend/secuscan/executor.py index 3cb44e9de..94946e5db 100644 --- a/backend/secuscan/executor.py +++ b/backend/secuscan/executor.py @@ -1003,20 +1003,21 @@ async def cancel_task(self, task_id: str) -> bool: logger.error(f"Failed to kill docker container for {task_id}: {e}") db = await get_db() - await db.execute( - "UPDATE tasks SET status = ?, completed_at = ? WHERE id = ?", - (TaskStatus.CANCELLED.value, datetime.now().isoformat(), task_id) - ) + async with db.transaction(): + await db.execute( + "UPDATE tasks SET status = ?, completed_at = ? WHERE id = ?", + (TaskStatus.CANCELLED.value, datetime.now().isoformat(), task_id) + ) + + await db.log_audit( + "task_cancelled", + "Task cancelled by user", + task_id=task_id + ) await self._broadcast(task_id, "status", TaskStatus.CANCELLED.value) await self._invalidate_cached_views() - await db.log_audit( - "task_cancelled", - "Task cancelled by user", - task_id=task_id - ) - return True async def get_task_status(self, task_id: str) -> Optional[Dict]: @@ -1424,41 +1425,42 @@ async def _upsert_findings_and_report(self, db, task_id: str, owner_id: str, plu 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) - ) + async with db.transaction(): + 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, - f"{plugin.name} Report", - "technical", - "ready" if status == TaskStatus.COMPLETED.value else "failed", - len(findings_data), - 1, - ), - ) + 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, + f"{plugin.name} Report", + "technical", + "ready" if status == TaskStatus.COMPLETED.value else "failed", + len(findings_data), + 1, + ), + ) - await self._persist_result_resources( - db, - owner_id=owner_id, - task_id=task_id, - plugin_id=plugin_id, - target=target, - result=structured_result, - ) + await self._persist_result_resources( + db, + owner_id=owner_id, + task_id=task_id, + plugin_id=plugin_id, + target=target, + result=structured_result, + ) async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, owner_id: str, scanner: Any, plugin_id: str, target: str, status: str, result: Dict[str, Any]): """Persist modular scanner results into findings, and reports.""" @@ -1489,42 +1491,43 @@ async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, owner 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) - ) + async with db.transaction(): + await db.execute( + "UPDATE tasks SET structured_json = ? WHERE id = ?", + (json.dumps(structured_result), task_id) + ) - # Create/Update report - 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, - f"{scanner.name} Report", - "professional" if status == TaskStatus.COMPLETED.value else "failed", - "ready" if status == TaskStatus.COMPLETED.value else "failed", - len(findings_data), - 2, # Professional reports are typically multi-page - ), - ) + # Create/Update report + 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, + f"{scanner.name} Report", + "professional" if status == TaskStatus.COMPLETED.value else "failed", + "ready" if status == TaskStatus.COMPLETED.value else "failed", + len(findings_data), + 2, # Professional reports are typically multi-page + ), + ) - await self._persist_result_resources( - db, - owner_id=owner_id, - task_id=task_id, - plugin_id=plugin_id, - target=target, - result=structured_result, - ) + await self._persist_result_resources( + db, + owner_id=owner_id, + task_id=task_id, + plugin_id=plugin_id, + target=target, + result=structured_result, + ) async def _persist_result_resources( self, diff --git a/backend/secuscan/platform_resources.py b/backend/secuscan/platform_resources.py index 8b84ac00f..1116ab8b9 100644 --- a/backend/secuscan/platform_resources.py +++ b/backend/secuscan/platform_resources.py @@ -114,8 +114,9 @@ async def replace_asset_services( target: str, services: Iterable[Dict[str, Any]], ) -> None: - await db.execute("DELETE FROM asset_services WHERE task_id = ?", (task_id,)) - for item in services: + async with db.transaction(): + await db.execute("DELETE FROM asset_services WHERE task_id = ?", (task_id,)) + for item in services: metadata = item.get("metadata", {}) if isinstance(item.get("metadata"), dict) else {} host = str(item.get("host") or target) port = item.get("port") diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index 22acc6f1c..53acc882b 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -1258,8 +1258,7 @@ async def delete_task_records(task_ids: List[str]): all_task_rows.extend(rows) # Delete associated records in chunks, atomic within a transaction - await db.begin() - try: + async with db.transaction(): # Re-check running status inside the transaction to prevent the # race where a task starts running between the check and the delete. for i in range(0, len(task_ids), SQLITE_CHUNK_SIZE): @@ -1300,13 +1299,6 @@ async def delete_task_records(task_ids: List[str]): await db.execute_no_commit( f"DELETE FROM tasks WHERE id IN ({placeholders})", tuple(chunk) ) - await db.commit() - except HTTPException: - await db.rollback() - raise - except Exception: - await db.rollback() - raise # Cleanup files on disk (outside the transaction — file deletion is not # transactional; a failure here does not leave the DB in an inconsistent @@ -1490,34 +1482,21 @@ async def upsert_vault_secret( encrypted = crypto.encrypt(value) secret_id = str(uuid.uuid4()) - existing = await db.fetchone( - """ - SELECT id - FROM credential_vault - WHERE owner_id = ? AND name = ? - """, - (owner, name), - ) - - if existing: - await db.execute( - """ - UPDATE credential_vault - SET encrypted_value = ?, updated_at = datetime('now') - WHERE owner_id = ? AND name = ? - """, - (encrypted, owner, name), - ) - else: - await db.execute( - """ - INSERT INTO credential_vault - (id, owner_id, name, encrypted_value) - VALUES (?, ?, ?, ?) - """, - (secret_id, owner, name, encrypted), + async with db.transaction(): + existing = await db.fetchone( + "SELECT id FROM credential_vault WHERE owner_id = ? AND name = ?", + (owner, name), ) - + if existing: + await db.execute( + "UPDATE credential_vault SET encrypted_value = ?, updated_at = datetime('now') WHERE owner_id = ? AND name = ?", + (encrypted, owner, name), + ) + else: + await db.execute( + "INSERT INTO credential_vault (id, owner_id, name, encrypted_value) VALUES (?, ?, ?, ?)", + (secret_id, owner, name, encrypted), + ) return {"name": name, "stored": True} @router.get("/vault/{name}", dependencies=[Depends(vault_limiter)]) From 145a52a9de06fb7ecb5d899389a4c948659440bd Mon Sep 17 00:00:00 2001 From: ionfwsrijan Date: Sun, 21 Jun 2026 19:56:09 +0530 Subject: [PATCH 2/7] fix(#1137): fix indentation and ruff formatting - Fix indentation in replace_asset_services for loop body after wrapping in async with db.transaction() - Run ruff format on all 4 changed files --- backend/secuscan/database.py | 102 +++++++++++------- backend/secuscan/platform_resources.py | 139 +++++++++++++++---------- 2 files changed, 148 insertions(+), 93 deletions(-) diff --git a/backend/secuscan/database.py b/backend/secuscan/database.py index 42bb36b80..d48667348 100644 --- a/backend/secuscan/database.py +++ b/backend/secuscan/database.py @@ -28,7 +28,9 @@ def __init__(self, db_path: str): def connection(self) -> aiosqlite.Connection: """Get the active database connection, raising an error if it's not connected.""" if self._connection is None: - raise RuntimeError("Database not connected. Did you forget to await connect()?") + raise RuntimeError( + "Database not connected. Did you forget to await connect()?" + ) return self._connection async def connect(self): @@ -418,13 +420,15 @@ async def _create_schema(self): "inputs_json": "TEXT NOT NULL DEFAULT '{}'", "execution_context_json": "TEXT NOT NULL DEFAULT '{}'", "preset": "TEXT", - "safe_mode": "BOOLEAN NOT NULL DEFAULT 1" + "safe_mode": "BOOLEAN NOT NULL DEFAULT 1", } for col_name, col_type in needed_cols.items(): if col_name not in existing_cols: try: - await self.execute(f"ALTER TABLE tasks ADD COLUMN {col_name} {col_type}") + await self.execute( + f"ALTER TABLE tasks ADD COLUMN {col_name} {col_type}" + ) print(f"Added missing column {col_name} to tasks table.") except Exception as e: print(f"Failed to add column {col_name}: {e}") @@ -468,7 +472,9 @@ async def _create_schema(self): for col_name, col_type in risk_cols.items(): if col_name not in existing_finding_cols: try: - await self.execute(f"ALTER TABLE findings ADD COLUMN {col_name} {col_type}") + await self.execute( + f"ALTER TABLE findings ADD COLUMN {col_name} {col_type}" + ) print(f"Added missing column {col_name} to findings table.") except Exception as e: print(f"Failed to add column {col_name}: {e}") @@ -488,7 +494,9 @@ async def _create_schema(self): for col_name, col_type in asset_service_needed.items(): if col_name not in existing_asset_service_cols: try: - await self.execute(f"ALTER TABLE asset_services ADD COLUMN {col_name} {col_type}") + await self.execute( + f"ALTER TABLE asset_services ADD COLUMN {col_name} {col_type}" + ) print(f"Added missing column {col_name} to asset_services table.") except Exception as e: print(f"Failed to add column {col_name} to asset_services: {e}") @@ -616,7 +624,9 @@ async def _create_schema(self): ALTER TABLE workflows_new RENAME TO workflows; """) await self.connection.commit() - print("Replaced workflows UNIQUE(name) constraint with UNIQUE(owner_id, name).") + print( + "Replaced workflows UNIQUE(name) constraint with UNIQUE(owner_id, name)." + ) finally: if old_fk: await self.execute("PRAGMA foreign_keys = ON") @@ -651,9 +661,9 @@ async def _run_migrations(self): if not migrations_dir.exists(): raise RuntimeError( - f"Migrations directory not found at {migrations_dir} — " - "ensure the backend package is installed correctly." - ) + f"Migrations directory not found at {migrations_dir} — " + "ensure the backend package is installed correctly." + ) for migration_file in sorted(migrations_dir.glob("*.sql")): sql = migration_file.read_text(encoding="utf-8") @@ -669,6 +679,7 @@ async def _run_migrations(self): async def _backfill_risk_scores(self): """Compute risk scores for existing findings that have none.""" from datetime import datetime, timezone + rows = await self.fetchall( "SELECT id, severity, exploitability, confidence, asset_exposure, discovered_at, risk_score FROM findings WHERE risk_score IS NULL" ) @@ -804,7 +815,6 @@ async def log_audit( ), ) - async def snapshot_workflow_version( self, workflow_id: str, @@ -858,17 +868,21 @@ async def get_workflow_versions(self, workflow_id: str) -> List[Dict]: defn = json.loads(row["definition_json"]) except (json.JSONDecodeError, TypeError): defn = {} - result.append({ - "id": row["id"], - "workflow_id": row["workflow_id"], - "version_number": row["version_number"], - "definition": defn, - "created_at": row["created_at"], - "created_by": row["created_by"], - }) + result.append( + { + "id": row["id"], + "workflow_id": row["workflow_id"], + "version_number": row["version_number"], + "definition": defn, + "created_at": row["created_at"], + "created_by": row["created_by"], + } + ) return result - async def get_workflow_version(self, workflow_id: str, version_number: int) -> Optional[Dict]: + async def get_workflow_version( + self, workflow_id: str, version_number: int + ) -> Optional[Dict]: """Return a specific version record or None if it does not exist.""" row = await self.fetchone( "SELECT id, workflow_id, version_number, definition_json, created_at, created_by " @@ -904,11 +918,20 @@ async def record_workflow_run( "INSERT INTO workflow_runs " "(id, workflow_id, version_id, version_number, triggered_by, status, task_ids_json) " "VALUES (?, ?, ?, ?, ?, 'queued', ?)", - (run_id, workflow_id, version_id, version_number, triggered_by, json.dumps(task_ids)), + ( + run_id, + workflow_id, + version_id, + version_number, + triggered_by, + json.dumps(task_ids), + ), ) return run_id - async def finalize_workflow_run(self, run_id: str, status: str, error_message: Optional[str] = None) -> None: + async def finalize_workflow_run( + self, run_id: str, status: str, error_message: Optional[str] = None + ) -> None: """Mark a workflow run as completed, failed, or cancelled with a timestamp. status must be one of: completed | failed | cancelled. @@ -929,7 +952,9 @@ async def check_workflow_run_tasks(self, run_id: str) -> Optional[str]: 'cancelled' if any task was cancelled and none are still running/queued. None if tasks are still in progress. """ - run_row = await self.fetchone("SELECT task_ids_json FROM workflow_runs WHERE id = ?", (run_id,)) + run_row = await self.fetchone( + "SELECT task_ids_json FROM workflow_runs WHERE id = ?", (run_id,) + ) if run_row is None: return None try: @@ -954,10 +979,13 @@ async def check_workflow_run_tasks(self, run_id: str) -> Optional[str]: return "cancelled" return "failed" - async def get_workflow_runs(self, workflow_id: str, limit: int = 50, offset: int = 0) -> Dict: + async def get_workflow_runs( + self, workflow_id: str, limit: int = 50, offset: int = 0 + ) -> Dict: """Return paginated run history for a workflow.""" count_row = await self.fetchone( - "SELECT COUNT(*) AS total FROM workflow_runs WHERE workflow_id = ?", (workflow_id,) + "SELECT COUNT(*) AS total FROM workflow_runs WHERE workflow_id = ?", + (workflow_id,), ) total = count_row["total"] if count_row else 0 rows = await self.fetchall( @@ -971,18 +999,20 @@ async def get_workflow_runs(self, workflow_id: str, limit: int = 50, offset: int task_ids = json.loads(row["task_ids_json"] or "[]") except (json.JSONDecodeError, TypeError): task_ids = [] - entries.append({ - "id": row["id"], - "workflow_id": row["workflow_id"], - "version_id": row["version_id"], - "version_number": row["version_number"], - "triggered_by": row["triggered_by"], - "status": row["status"], - "task_ids": task_ids, - "started_at": row["started_at"], - "completed_at": row["completed_at"], - "error_message": row["error_message"], - }) + entries.append( + { + "id": row["id"], + "workflow_id": row["workflow_id"], + "version_id": row["version_id"], + "version_number": row["version_number"], + "triggered_by": row["triggered_by"], + "status": row["status"], + "task_ids": task_ids, + "started_at": row["started_at"], + "completed_at": row["completed_at"], + "error_message": row["error_message"], + } + ) return {"total": total, "runs": entries} diff --git a/backend/secuscan/platform_resources.py b/backend/secuscan/platform_resources.py index 1116ab8b9..2c2c309dd 100644 --- a/backend/secuscan/platform_resources.py +++ b/backend/secuscan/platform_resources.py @@ -30,7 +30,9 @@ def _stable_asset_id(target: str, host: Any, port: Any, protocol: Any) -> str: return f"asset:{digest}" -async def get_target_policy(db: Database, owner_id: str, policy_id: str | None) -> Optional[Dict[str, Any]]: +async def get_target_policy( + db: Database, owner_id: str, policy_id: str | None +) -> Optional[Dict[str, Any]]: if not policy_id: return None row = await db.fetchone( @@ -40,7 +42,9 @@ async def get_target_policy(db: Database, owner_id: str, policy_id: str | None) return _deserialize_resource_row(row) -async def get_credential_profile(db: Database, owner_id: str, profile_id: str | None) -> Optional[Dict[str, Any]]: +async def get_credential_profile( + db: Database, owner_id: str, profile_id: str | None +) -> Optional[Dict[str, Any]]: if not profile_id: return None row = await db.fetchone( @@ -50,7 +54,9 @@ async def get_credential_profile(db: Database, owner_id: str, profile_id: str | return _deserialize_resource_row(row) -async def get_session_profile(db: Database, owner_id: str, profile_id: str | None) -> Optional[Dict[str, Any]]: +async def get_session_profile( + db: Database, owner_id: str, profile_id: str | None +) -> Optional[Dict[str, Any]]: if not profile_id: return None row = await db.fetchone( @@ -117,66 +123,85 @@ async def replace_asset_services( async with db.transaction(): await db.execute("DELETE FROM asset_services WHERE task_id = ?", (task_id,)) for item in services: - metadata = item.get("metadata", {}) if isinstance(item.get("metadata"), dict) else {} - host = str(item.get("host") or target) - port = item.get("port") - protocol = item.get("protocol") - asset_id = str(item.get("asset_id") or _stable_asset_id(target, host, port, protocol)) - cert_sans = item.get("cert_san") or item.get("cert_sans") or metadata.get("cert_san") or metadata.get("cert_sans") or [] - if not isinstance(cert_sans, list): - cert_sans = [cert_sans] - service_fingerprint = item.get("service_fingerprint") - if not service_fingerprint: - service_fingerprint = " ".join( - str(part).strip() - for part in ( + metadata = ( + item.get("metadata", {}) + if isinstance(item.get("metadata"), dict) + else {} + ) + host = str(item.get("host") or target) + port = item.get("port") + protocol = item.get("protocol") + asset_id = str( + item.get("asset_id") or _stable_asset_id(target, host, port, protocol) + ) + cert_sans = ( + item.get("cert_san") + or item.get("cert_sans") + or metadata.get("cert_san") + or metadata.get("cert_sans") + or [] + ) + if not isinstance(cert_sans, list): + cert_sans = [cert_sans] + service_fingerprint = item.get("service_fingerprint") + if not service_fingerprint: + service_fingerprint = ( + " ".join( + str(part).strip() + for part in ( + item.get("product"), + item.get("version"), + item.get("service"), + item.get("title"), + ) + if str(part or "").strip() + ) + or None + ) + await db.execute( + """ + INSERT INTO asset_services ( + id, owner_id, task_id, plugin_id, target, asset_id, host, ip, port, protocol, + service, product, version, cpe, confidence, title, banner, cert_subject, + cert_san_json, cert_expiry, service_fingerprint, metadata_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + str(uuid.uuid4()), + owner_id, + task_id, + plugin_id, + target, + asset_id, + host, + item.get("ip"), + item.get("port"), + item.get("protocol"), + item.get("service"), item.get("product"), item.get("version"), - item.get("service"), + item.get("cpe"), + item.get("confidence"), item.get("title"), - ) - if str(part or "").strip() - ) or None - await db.execute( - """ - INSERT INTO asset_services ( - id, owner_id, task_id, plugin_id, target, asset_id, host, ip, port, protocol, - service, product, version, cpe, confidence, title, banner, cert_subject, - cert_san_json, cert_expiry, service_fingerprint, metadata_json - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - """, - ( - str(uuid.uuid4()), - owner_id, - task_id, - plugin_id, - target, - asset_id, - host, - item.get("ip"), - item.get("port"), - item.get("protocol"), - item.get("service"), - item.get("product"), - item.get("version"), - item.get("cpe"), - item.get("confidence"), - item.get("title"), - item.get("banner"), - item.get("cert_subject"), - json.dumps(cert_sans), - item.get("cert_expiry"), - service_fingerprint, - json.dumps(metadata), - ), - ) - - -def serialize_execution_context(context: ExecutionContext | Dict[str, Any] | None) -> str: + item.get("banner"), + item.get("cert_subject"), + json.dumps(cert_sans), + item.get("cert_expiry"), + service_fingerprint, + json.dumps(metadata), + ), + ) + + +def serialize_execution_context( + context: ExecutionContext | Dict[str, Any] | None, +) -> str: return json.dumps(normalize_execution_context(context or {})) -def _deserialize_resource_row(row: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]: +def _deserialize_resource_row( + row: Optional[Dict[str, Any]], +) -> Optional[Dict[str, Any]]: if row is None: return None item = dict(row) From db9982c8e830cfad3a5ac831c1a9c1644db66e7e Mon Sep 17 00:00:00 2001 From: opencode-bot Date: Wed, 24 Jun 2026 21:00:46 +0530 Subject: [PATCH 3/7] fix: add missing import os and fix syntax error in routes.py --- backend/secuscan/config.py | 1 + backend/secuscan/routes.py | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/backend/secuscan/config.py b/backend/secuscan/config.py index 5e7b7bacd..44d6a51de 100644 --- a/backend/secuscan/config.py +++ b/backend/secuscan/config.py @@ -8,6 +8,7 @@ from pydantic_settings import BaseSettings import base64 import hashlib +import os PROJECT_ROOT = Path(__file__).resolve().parent.parent diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index 53acc882b..bbc6e2a81 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -1838,7 +1838,7 @@ async def _verify_workflow_owner(db, workflow_id: str, owner: str): return row -@router.post("/workflows/{workflow_id}/run") , dependencies=[Depends(check_scan_rate_limit)] +@router.post("/workflows/{workflow_id}/run", dependencies=[Depends(check_scan_rate_limit)]) async def run_workflow_once(workflow_id: str, owner: str = Depends(get_current_owner)): db = await get_db() row = await _verify_workflow_owner(db, workflow_id, owner) From 8779244331cd7c285abc4cdf9f573cbd0c464c19 Mon Sep 17 00:00:00 2001 From: opencode-bot Date: Wed, 24 Jun 2026 21:05:49 +0530 Subject: [PATCH 4/7] fix: add missing check_scan_rate_limit dependency function to rate_limiter.py --- backend/secuscan/rate_limiter.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/backend/secuscan/rate_limiter.py b/backend/secuscan/rate_limiter.py index 60f5091aa..d105b30ec 100644 --- a/backend/secuscan/rate_limiter.py +++ b/backend/secuscan/rate_limiter.py @@ -183,3 +183,14 @@ def make_scan_rate_limiter( burst_limit=burst_limit, burst_window=burst_window, ) + + +async def check_scan_rate_limit(request: Request) -> None: + """FastAPI dependency that checks scan rate limits for scan-triggering endpoints. + + Retrieves the ``ScanRateLimiter`` instance from ``request.app.state`` + (initialized during app startup) and delegates to its ``check`` method. + """ + limiter = getattr(request.app.state, "scan_rate_limiter", None) + if limiter: + await limiter.check(request) From 592bfd30a49343056958695ba5fe828f1e8d94c8 Mon Sep 17 00:00:00 2001 From: opencode-bot Date: Wed, 24 Jun 2026 21:09:45 +0530 Subject: [PATCH 5/7] fix: add missing RateLimitExceeded exception class to rate_limiter.py --- backend/secuscan/rate_limiter.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/backend/secuscan/rate_limiter.py b/backend/secuscan/rate_limiter.py index d105b30ec..4c83dc9b3 100644 --- a/backend/secuscan/rate_limiter.py +++ b/backend/secuscan/rate_limiter.py @@ -25,6 +25,10 @@ logger = logging.getLogger(__name__) +class RateLimitExceeded(HTTPException): + """Raised when a rate limit is exceeded. Caught by a global exception handler.""" + + class ScanRateLimiter: """ Sliding window rate limiter for scan execution endpoints. From e22d23af22e5255d0c74ffdaa03b11d48b246f7a Mon Sep 17 00:00:00 2001 From: opencode-bot Date: Wed, 24 Jun 2026 21:16:39 +0530 Subject: [PATCH 6/7] fix: remove invalid @router.exception_handler - APIRouter lacks this method, handler already in main.py --- backend/secuscan/routes.py | 18 +----------------- 1 file changed, 1 insertion(+), 17 deletions(-) diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index bbc6e2a81..a6b583ea2 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -104,7 +104,7 @@ def _json_payload(value: Any, fallback: str) -> str: resolve_client_identity, admin_limiter, scheduler_tick_limiter, ) -from .rate_limiter import check_scan_rate_limit, RateLimitExceeded +from .rate_limiter import check_scan_rate_limit from .validation import validate_target, validate_task_start_payload, validate_url from .reporting import reporting from .vault import VaultCrypto @@ -196,22 +196,6 @@ async def get_or_set_cached(key: str, builder): return value -from fastapi.responses import JSONResponse -from starlette.status import HTTP_429_TOO_MANY_REQUESTS -from .rate_limiter import RateLimitExceeded - -@router.exception_handler(RateLimitExceeded) -async def rate_limit_exception_handler(request: Request, exc: RateLimitExceeded): - return JSONResponse( - status_code=HTTP_429_TOO_MANY_REQUESTS, - content={ - "error": str(exc.detail) if hasattr(exc, 'detail') else "Too Many Requests", - "retry_after": getattr(exc, 'retry_after', 60), - }, - headers={"Retry-After": str(getattr(exc, 'retry_after', 60))}, - ) - - async def require_owned_task(db, task_id: str, owner: str, columns: str = "owner_id") -> Dict[str, Any]: """Fetch a task and enforce that it belongs to ``owner`` (issue #401). From 75863df7198345bc440d44d51611a9d9b61c2a8d Mon Sep 17 00:00:00 2001 From: opencode-bot Date: Wed, 24 Jun 2026 21:31:48 +0530 Subject: [PATCH 7/7] fix: configure mock_db.transaction as async context manager in tests, preserve rate-limit headers in 429 handler --- backend/secuscan/main.py | 17 +++++++++++++---- .../unit/test_bulk_delete_sqlite_limit.py | 2 ++ .../unit/test_process_tree_cancellation.py | 2 ++ 3 files changed, 17 insertions(+), 4 deletions(-) diff --git a/backend/secuscan/main.py b/backend/secuscan/main.py index c1b650093..66282c415 100644 --- a/backend/secuscan/main.py +++ b/backend/secuscan/main.py @@ -238,17 +238,26 @@ async def rate_limit_exceeded_handler(request: Request, exc: RateLimitExceeded): async def generic_rate_limit_handler(request: Request, exc: Exception): """ Generic handler for 429 status code exceptions. + + Merges headers from the original exception (e.g. X-RateLimit-Limit, + X-RateLimit-Remaining, Retry-After) with default headers, so + callers always receive accurate rate-limit metadata. """ + exc_headers = getattr(exc, "headers", None) or {} + headers = { + "X-Request-ID": getattr(request.state, "request_id", get_request_id()), + **exc_headers, + } + if "Retry-After" not in headers: + headers["Retry-After"] = "60" + return JSONResponse( status_code=HTTP_429_TOO_MANY_REQUESTS, content={ "error": "Too Many Requests", "message": "Rate limit exceeded. Please try again later." }, - headers={ - "Retry-After": "60", - "X-Request-ID": getattr(request.state, "request_id", get_request_id()), - }, + headers=headers, ) # ─── END CUSTOM 429 HANDLER ────────────────────────────────────────────────── diff --git a/testing/backend/unit/test_bulk_delete_sqlite_limit.py b/testing/backend/unit/test_bulk_delete_sqlite_limit.py index 8ac034078..e68c638c3 100644 --- a/testing/backend/unit/test_bulk_delete_sqlite_limit.py +++ b/testing/backend/unit/test_bulk_delete_sqlite_limit.py @@ -190,6 +190,7 @@ async def test_single_chunk_does_not_exceed_sqlite_limit(self): mock_db.begin = AsyncMock() mock_db.commit = AsyncMock() mock_db.rollback = AsyncMock() + mock_db.transaction = MagicMock(return_value=AsyncMock()) async def capture_execute(sql, params=()): captured_sql.append(sql) mock_db.execute_no_commit = capture_execute @@ -226,6 +227,7 @@ async def test_multi_chunk_splits_correctly(self): mock_db.begin = AsyncMock() mock_db.commit = AsyncMock() mock_db.rollback = AsyncMock() + mock_db.transaction = MagicMock(return_value=AsyncMock()) async def capture_execute(sql, params=()): captured_sql.append(sql) captured_params.append(params) diff --git a/testing/backend/unit/test_process_tree_cancellation.py b/testing/backend/unit/test_process_tree_cancellation.py index a4470cf45..2f5b61f5e 100644 --- a/testing/backend/unit/test_process_tree_cancellation.py +++ b/testing/backend/unit/test_process_tree_cancellation.py @@ -250,6 +250,7 @@ async def test_cancel_task_terminates_process_group(self): mock_get_db.return_value = mock_db mock_db.execute = AsyncMock() mock_db.log_audit = AsyncMock() + mock_db.transaction = MagicMock(return_value=AsyncMock()) await executor.cancel_task("task-pg") mock_term.assert_awaited_once_with(7001, "task-pg") @@ -273,6 +274,7 @@ async def test_cancel_task_no_pid_still_cancels_asyncio_task(self): mock_get_db.return_value = mock_db mock_db.execute = AsyncMock() mock_db.log_audit = AsyncMock() + mock_db.transaction = MagicMock(return_value=AsyncMock()) await executor.cancel_task("task-nopid") mock_term.assert_not_awaited()