Fix BOLA by enforcing per-user ownership for tasks and reports

backend/secuscan/auth.py

@@ -83,3 +83,41 @@ async def require_api_key(
 def get_api_key() -> str | None:
     """Return the current API key, or None if not yet initialised."""
     return _api_key
+
+
+# ── Per-user / per-workspace ownership ──────────────────────────────────────
+#
+# SecuScan authenticates the deployment with a single shared API key (above).
+# That gate does not, by itself, distinguish between the different users or
+# workspaces that share a deployment, which is what allowed any caller to read,
+# delete, or export any task/report by guessing its ID (BOLA, issue #401).
+#
+# ``resolve_owner_id`` derives a stable owner identity for the request and is
+# persisted as ``owner_id`` on tasks/findings/reports at creation time and
+# compared on every read/delete/report access. It deliberately prioritises the
+# explicit authenticated-user header (``X-User-Id``) — the same header
+# ``resolve_client_identity`` already treats as the authenticated user — so that
+# multiple workspaces sharing the deployment API key remain isolated. In a
+# production deployment the header is expected to be set by an upstream auth
+# proxy / SSO layer; deployments that do not send it fall back to a single
+# shared ``DEFAULT_OWNER_ID`` and keep their existing (single-user) behaviour.
+#
+# This value is duplicated as the SQL column default ('default') in
+# database.py — keep the two in sync.
+DEFAULT_OWNER_ID = "default"
+
+_OWNER_HEADER = "x-user-id"
+
+
+def resolve_owner_id(request: Request | None) -> str:
+    """Resolve the owning user/workspace identity for the current request."""
+    if request is not None:
+        user_id = request.headers.get(_OWNER_HEADER)
+        if user_id and user_id.strip():
+            return f"user:{user_id.strip()}"
+    return DEFAULT_OWNER_ID
+
+
+async def get_current_owner(request: Request) -> str:
+    """FastAPI dependency yielding the owner identity for the request."""
+    return resolve_owner_id(request)

backend/secuscan/database.py

@@ -54,6 +54,7 @@ async def _create_schema(self):
             """
             CREATE TABLE IF NOT EXISTS tasks (
                 id TEXT PRIMARY KEY,
+                owner_id TEXT NOT NULL DEFAULT 'default',
                 plugin_id TEXT NOT NULL,
                 tool_name TEXT NOT NULL,
                 target TEXT NOT NULL,
@@ -95,6 +96,7 @@ async def _create_schema(self):
 
             CREATE TABLE IF NOT EXISTS findings (
                 id TEXT PRIMARY KEY,
+                owner_id TEXT NOT NULL DEFAULT 'default',
                 task_id TEXT REFERENCES tasks(id) ON DELETE SET NULL,
                 plugin_id TEXT NOT NULL,
                 title TEXT NOT NULL,
@@ -113,6 +115,7 @@ async def _create_schema(self):
 
             CREATE TABLE IF NOT EXISTS reports (
                 id TEXT PRIMARY KEY,
+                owner_id TEXT NOT NULL DEFAULT 'default',
                 task_id TEXT REFERENCES tasks(id) ON DELETE SET NULL,
                 name TEXT NOT NULL,
                 type TEXT NOT NULL DEFAULT 'technical',
@@ -200,6 +203,8 @@ async def _create_schema(self):
             CREATE INDEX IF NOT EXISTS idx_tasks_plugin ON tasks(plugin_id);
             -- Composite index for dashboard running tasks query
             CREATE INDEX IF NOT EXISTS idx_tasks_status_created ON tasks(status, created_at DESC);
+            -- Owner scoping (BOLA prevention, issue #401)
+            CREATE INDEX IF NOT EXISTS idx_tasks_owner ON tasks(owner_id);
 
             -- Findings indexes (new)
             CREATE INDEX IF NOT EXISTS idx_findings_severity ON findings(severity);
@@ -209,11 +214,15 @@ async def _create_schema(self):
             CREATE INDEX IF NOT EXISTS idx_findings_target ON findings(target);
             -- Composite index for severity counting by task
             CREATE INDEX IF NOT EXISTS idx_findings_task_severity ON findings(task_id, severity);
+            -- Owner scoping (BOLA prevention, issue #401)
+            CREATE INDEX IF NOT EXISTS idx_findings_owner ON findings(owner_id);
 
             -- Reports indexes (new)
             CREATE INDEX IF NOT EXISTS idx_reports_task_id ON reports(task_id);
             CREATE INDEX IF NOT EXISTS idx_reports_generated_at ON reports(generated_at DESC);
             CREATE INDEX IF NOT EXISTS idx_reports_status ON reports(status);
+            -- Owner scoping (BOLA prevention, issue #401)
+            CREATE INDEX IF NOT EXISTS idx_reports_owner ON reports(owner_id);
 
             -- Audit log indexes (new)
             CREATE INDEX IF NOT EXISTS idx_audit_timestamp ON audit_log(timestamp DESC);
@@ -235,6 +244,10 @@ async def _create_schema(self):
         existing_cols = {col["name"] for col in tasks_columns}
 
         needed_cols = {
+            # Per-user ownership for BOLA prevention (issue #401). NOT NULL with a
+            # constant default backfills every existing row to the shared default
+            # owner, preserving single-user deployments' access to their history.
+            "owner_id": "TEXT NOT NULL DEFAULT 'default'",
             "exit_code": "INTEGER",
             "scan_phase": "TEXT",
             "structured_json": "TEXT",
@@ -272,6 +285,8 @@ async def _create_schema(self):
             "asset_exposure": "TEXT",
             "risk_score": "REAL",
             "risk_factors_json": "TEXT NOT NULL DEFAULT '[]'",
+            # Per-user ownership for BOLA prevention (issue #401).
+            "owner_id": "TEXT NOT NULL DEFAULT 'default'",
         }
         for col_name, col_type in risk_cols.items():
             if col_name not in existing_finding_cols:
@@ -281,6 +296,18 @@ async def _create_schema(self):
                 except Exception as e:
                     print(f"Failed to add column {col_name}: {e}")
 
+        # Reports table migration: ensure owner_id exists (issue #401)
+        reports_columns = await self.fetchall("PRAGMA table_info(reports)")
+        existing_report_cols = {col["name"] for col in reports_columns}
+        if "owner_id" not in existing_report_cols:
+            try:
+                await self.execute(
+                    "ALTER TABLE reports ADD COLUMN owner_id TEXT NOT NULL DEFAULT 'default'"
+                )
+                print("Added missing column 'owner_id' to reports table.")
+            except Exception as e:
+                print(f"Failed to add 'owner_id' to reports: {e}")
+
     async def _run_migrations(self):
         migrations_dir = Path(__file__).parent / "migrations"
 

backend/secuscan/executor.py

@@ -13,6 +13,7 @@
 import logging
 import re
 
+from .auth import DEFAULT_OWNER_ID
 from .redaction import redact
 from .cache import get_cache
 from .config import settings
@@ -145,17 +146,22 @@ async def create_task(
         inputs: Dict[str, Any],
         safe_mode: bool,
         preset: Optional[str] = None,
-        consent_granted: bool = False
+        consent_granted: bool = False,
+        owner_id: str = DEFAULT_OWNER_ID,
     ) -> str:
         """
         Create a new scan task.
-        
+
         Args:
             plugin_id: Plugin identifier
             inputs: User input values
             preset: Optional preset name
             consent_granted: Whether user granted consent
-
+            owner_id: Owning user/workspace identity used to scope later
+                access (issue #401). Defaults to the shared default owner for
+                internal callers (workflows, scheduler, CLI) that are not tied
+                to a request.
+
         Returns:
             Task ID
         """
@@ -177,12 +183,13 @@ async def create_task(
         await db.execute(
             """
             INSERT INTO tasks (
-                id, plugin_id, tool_name, target, inputs_json, preset,
+                id, owner_id, plugin_id, tool_name, target, inputs_json, preset,
                 status, scan_phase, consent_granted, safe_mode
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             """,
             (
                 task_id,
+                owner_id,
                 plugin_id,
                 plugin.name,
                 extract_target(inputs),
@@ -261,13 +268,14 @@ async def execute_task(self, task_id: str):
 
             # Get task details
             task_row = await db.fetchone(
-                "SELECT plugin_id, inputs_json, safe_mode FROM tasks WHERE id = ?",
+                "SELECT owner_id, plugin_id, inputs_json, safe_mode FROM tasks WHERE id = ?",
                 (task_id,)
             )
 
             if not task_row:
                 raise ValueError(f"Task not found: {task_id}")
 
+            owner_id = task_row["owner_id"]
             plugin_id = task_row["plugin_id"]
             inputs = json.loads(task_row["inputs_json"])
             safe_mode = bool(task_row["safe_mode"])
@@ -383,6 +391,7 @@ async def execute_task(self, task_id: str):
                 await self._upsert_findings_and_report_from_scanner(
                     db=db,
                     task_id=task_id,
+                    owner_id=owner_id,
                     scanner=scanner,
                     plugin_id=plugin_id,
                     target=target,
@@ -520,6 +529,7 @@ async def execute_task(self, task_id: str):
                 await self._upsert_findings_and_report(
                     db=db,
                     task_id=task_id,
+                    owner_id=owner_id,
                     plugin=plugin,
                     plugin_id=plugin_id,
                     target=target,
@@ -850,7 +860,7 @@ async def get_task_status(self, task_id: str) -> Optional[Dict]:
             "pending_count": pending_count,
         }
 
-    async def _upsert_findings_and_report(self, db, task_id: str, plugin, plugin_id: str, target: str, status: str, output: str = ""):
+    async def _upsert_findings_and_report(self, db, task_id: str, owner_id: str, plugin, plugin_id: str, target: str, status: str, output: str = ""):
         """Persist derived findings and report records into SQLite."""
         parsed = self._parse_results(plugin, output)
         findings_data = parsed.get("findings", [])
@@ -890,16 +900,17 @@ async def _upsert_findings_and_report(self, db, task_id: str, plugin, plugin_id:
             await db.execute(
                 """
                 INSERT INTO findings (
-                    id, task_id, plugin_id, title, category, severity,
+                    id, owner_id, task_id, plugin_id, title, category, severity,
                     target, description, remediation, proof, cvss, cve,
                     metadata_json, discovered_at,
                     exploitability, confidence, asset_exposure,
                     risk_score, risk_factors_json
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                           ?, ?, ?, ?, ?)
                 """,
                 (
                     finding_id,
+                    owner_id,
                     task_id,
                     plugin_id,
                     finding["title"],
@@ -924,15 +935,16 @@ async def _upsert_findings_and_report(self, db, task_id: str, plugin, plugin_id:
         await db.execute(
             """
             INSERT INTO reports (
-                id, task_id, name, type, generated_at, status, findings, pages
-            ) VALUES (?, ?, ?, ?, (datetime('now')), ?, ?, ?)
+                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",
@@ -942,7 +954,7 @@ async def _upsert_findings_and_report(self, db, task_id: str, plugin, plugin_id:
             ),
         )
 
-    async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, scanner: Any, plugin_id: str, target: str, status: str, result: Dict[str, Any]):
+    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."""
         findings_data = result.get("findings", [])
 
@@ -975,16 +987,17 @@ async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, scann
             await db.execute(
                 """
                 INSERT INTO findings (
-                    id, task_id, plugin_id, title, category, severity,
+                    id, owner_id, task_id, plugin_id, title, category, severity,
                     target, description, remediation, proof, cvss, cve,
                     metadata_json, discovered_at,
                     exploitability, confidence, asset_exposure,
                     risk_score, risk_factors_json
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                           ?, ?, ?, ?, ?)
                 """,
                 (
                     finding_id,
+                    owner_id,
                     task_id,
                     plugin_id,
                     finding["title"],
@@ -1010,15 +1023,16 @@ async def _upsert_findings_and_report_from_scanner(self, db, task_id: str, scann
         await db.execute(
             """
             INSERT INTO reports (
-                id, task_id, name, type, generated_at, status, findings, pages
-            ) VALUES (?, ?, ?, ?, (datetime('now')), ?, ?, ?)
+                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",

backend/secuscan/migrations/003_add_owner_id.sql

@@ -0,0 +1,26 @@
+-- Migration: 003_add_owner_id
+-- Introduces per-user / per-workspace ownership for tasks, findings, and
+-- reports to close the Broken Object Level Authorization (BOLA) gap where any
+-- caller could read, delete, or export any task/report by guessing its ID
+-- (issue #401).
+--
+-- The owner_id columns themselves are added idempotently in database.py
+-- (_create_schema), using PRAGMA table_info checks so re-running startup is
+-- safe — SQLite has no "ALTER TABLE ... ADD COLUMN IF NOT EXISTS". This file
+-- only contains statements that are safe to re-run on every startup:
+--
+--   1. A defensive backfill of any NULL owner_id to the shared default owner.
+--      (New columns are added as NOT NULL DEFAULT 'default', so existing rows
+--      are already backfilled; this guards against rows created by an older
+--      build that may have added the column as nullable.)
+--   2. Indexes used to keep owner-scoped list queries fast.
+--
+-- Keep the 'default' literal in sync with auth.DEFAULT_OWNER_ID.
+
+UPDATE tasks    SET owner_id = 'default' WHERE owner_id IS NULL;
+UPDATE findings SET owner_id = 'default' WHERE owner_id IS NULL;
+UPDATE reports  SET owner_id = 'default' WHERE owner_id IS NULL;
+
+CREATE INDEX IF NOT EXISTS idx_tasks_owner    ON tasks(owner_id);
+CREATE INDEX IF NOT EXISTS idx_findings_owner ON findings(owner_id);
+CREATE INDEX IF NOT EXISTS idx_reports_owner  ON reports(owner_id);