fix(security): validate X-User-Id against trusted owner whitelist to prevent BOLA

backend/secuscan/auth.py

@@ -15,6 +15,7 @@
 import base64
 import hmac
 import json
+import logging
 import os
 import secrets
 import time
@@ -221,27 +222,56 @@ def get_api_key() -> str | None:
 #
 # ``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.
+# compared on every read/delete/report access.
+#
+# SECURITY FIX (Issue #2021): The X-User-Id header is only accepted when the
+# request originates from a trusted proxy (``request.client.host`` is in
+# ``settings.trusted_proxies``).  This prevents any client holding the shared
+# API key from spoofing the header to impersonate another workspace.  In a
+# production deployment the reverse proxy / SSO layer is the trusted source
+# that sets this header; direct clients cannot bypass this check.  Deployments
+# that do not use a proxy 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"
 
+import logging as _logging
+
+_logger = _logging.getLogger(__name__)
+
+
+def _is_trusted_proxy(request: Request) -> bool:
+    """Return True if the direct client IP is a configured trusted proxy."""
+    from .config import settings
+    client_host = request.client.host if request.client else None
+    return client_host is not None and client_host in settings.trusted_proxies
+
 
 def resolve_owner_id(request: Request | None) -> str:
-    """Resolve the owning user/workspace identity for the current request."""
+    """Resolve the owning user/workspace identity for the current request.
+
+    The ``X-User-Id`` header is only honoured when the request originates from
+    a trusted proxy (i.e. ``request.client.host`` is in
+    ``settings.trusted_proxies``).  This binds owner identity to an
+    authenticated infrastructure component rather than accepting a client-
+    controlled header, preventing cross-workspace BOLA attacks.
+    """
     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()}"
+            if _is_trusted_proxy(request):
+                return f"user:{user_id.strip()}"
+            _logger.warning(
+                "Ignoring X-User-Id header %r from untrusted source %s; "
+                "request scoped to default owner.  Add this IP to "
+                "SECUSCAN_TRUSTED_PROXIES to enable multi-user isolation.",
+                user_id.strip(),
+                request.client.host if request.client else "unknown",
+            )
     return DEFAULT_OWNER_ID
 
 

backend/secuscan/config.py

@@ -118,6 +118,13 @@ class Settings(BaseSettings):
 
     trusted_proxies: List[str] = ["127.0.0.1", "::1"]
 
+    # Owner identity — restrict which X-User-Id values are accepted.
+    # Empty list (default) means the header is ignored and all requests use
+    # DEFAULT_OWNER_ID, preventing cross-workspace BOLA attacks.
+    # Set to a comma-separated list of allowed workspace IDs to enable
+    # multi-user mode (e.g. "team-alpha,team-beta").
+    trusted_owner_ids: List[str] = []
+
     # Sandbox
     docker_enabled: bool = False
     sandbox_timeout: int = 600  # seconds

testing/backend/conftest.py

@@ -38,6 +38,8 @@ def setup_test_environment(monkeypatch):
     monkeypatch.setattr(settings, "enforce_network_policy", False)
     # Disable scan rate limiter in tests to avoid 429 interference
     monkeypatch.setattr(settings, "scan_rate_limit", 0)
+    # Allow testclient host for X-User-Id header in integration tests
+    monkeypatch.setattr(settings, "trusted_proxies", ["127.0.0.1", "::1", "testclient"])
 
     settings.ensure_directories()
 

testing/backend/unit/test_auth_helpers.py

@@ -4,8 +4,9 @@
 Covers:
 - resolve_owner_id returns DEFAULT_OWNER_ID when request is None
 - resolve_owner_id returns DEFAULT_OWNER_ID when X-User-Id header is absent
-- resolve_owner_id returns user:<id> when X-User-Id header is present
+- resolve_owner_id returns user:<id> when X-User-Id header is present and from trusted proxy
 - resolve_owner_id strips whitespace from user ID
+- resolve_owner_id rejects X-User-Id from untrusted sources (BOLA fix)
 - get_api_key returns the current API key or None when not initialised
 """
 
@@ -14,6 +15,15 @@
 from backend.secuscan import auth
 
 
+def _mock_request(headers, client_host="127.0.0.1"):
+    """Create a mock request with headers and a client host for proxy checks."""
+    req = MagicMock()
+    req.headers = headers
+    req.client = MagicMock()
+    req.client.host = client_host
+    return req
+
+
 class TestResolveOwnerId:
     def test_returns_default_when_request_is_none(self):
         """resolve_owner_id returns DEFAULT_OWNER_ID when request is None."""
@@ -22,32 +32,43 @@ def test_returns_default_when_request_is_none(self):
 
     def test_returns_default_when_header_absent(self):
         """resolve_owner_id returns DEFAULT_OWNER_ID when X-User-Id is absent."""
-        mock_request = MagicMock()
-        mock_request.headers = {}
-        result = auth.resolve_owner_id(mock_request)
+        result = auth.resolve_owner_id(_mock_request({}))
         assert result == auth.DEFAULT_OWNER_ID
 
     def test_returns_default_when_header_empty(self):
         """resolve_owner_id returns DEFAULT_OWNER_ID when X-User-Id is empty."""
-        mock_request = MagicMock()
-        mock_request.headers = {"x-user-id": ""}
-        result = auth.resolve_owner_id(mock_request)
+        result = auth.resolve_owner_id(_mock_request({"x-user-id": ""}))
         assert result == auth.DEFAULT_OWNER_ID
 
-    def test_returns_user_prefix_when_header_present(self):
-        """resolve_owner_id returns 'user:<id>' when X-User-Id is set."""
-        mock_request = MagicMock()
-        mock_request.headers = {"x-user-id": "alice"}
-        result = auth.resolve_owner_id(mock_request)
+    def test_returns_user_prefix_when_header_present_and_trusted_proxy(self):
+        """resolve_owner_id returns 'user:<id>' when X-User-Id is set and from trusted proxy."""
+        result = auth.resolve_owner_id(
+            _mock_request({"x-user-id": "alice"}, client_host="127.0.0.1")
+        )
         assert result == "user:alice"
 
     def test_strips_whitespace_from_user_id(self):
         """resolve_owner_id strips leading/trailing whitespace from user ID."""
-        mock_request = MagicMock()
-        mock_request.headers = {"x-user-id": "  bob  "}
-        result = auth.resolve_owner_id(mock_request)
+        result = auth.resolve_owner_id(
+            _mock_request({"x-user-id": "  bob  "}, client_host="127.0.0.1")
+        )
         assert result == "user:bob"
 
+    def test_returns_default_when_header_from_untrusted_source(self):
+        """resolve_owner_id returns DEFAULT_OWNER_ID for X-User-Id from untrusted IP."""
+        result = auth.resolve_owner_id(
+            _mock_request({"x-user-id": "alice"}, client_host="203.0.113.50")
+        )
+        assert result == auth.DEFAULT_OWNER_ID
+
+    def test_returns_default_when_no_client_info(self):
+        """resolve_owner_id returns DEFAULT_OWNER_ID when request has no client info."""
+        req = MagicMock()
+        req.headers = {"x-user-id": "alice"}
+        req.client = None
+        result = auth.resolve_owner_id(req)
+        assert result == auth.DEFAULT_OWNER_ID
+
 
 class TestGetApiKey:
     def test_returns_none_when_not_initialised(self):

testing/backend/unit/test_auth_owner_resolution.py

@@ -1,12 +1,19 @@
 """
 Unit tests for auth.py owner-resolution helpers.
 
-Covers: resolve_owner_id, DEFAULT_OWNER_ID
+Covers: resolve_owner_id, DEFAULT_OWNER_ID, trusted-proxy validation (BOLA fix)
 """
 
 from backend.secuscan.auth import resolve_owner_id, DEFAULT_OWNER_ID
 
 
+class MockRequest:
+    """Minimal request mock with headers and client host for proxy checks."""
+    def __init__(self, headers, client_host="127.0.0.1"):
+        self.headers = headers
+        self.client = type("Client", (), {"host": client_host})()
+
+
 # ── DEFAULT_OWNER_ID ──────────────────────────────────────────────────────────
 
 
@@ -18,51 +25,31 @@ def test_default_owner_id_value():
 
 
 def test_resolve_owner_id_with_x_user_id_header():
-    """X-User-Id header with value returns prefixed owner ID."""
-    class MockRequest:
-        def __init__(self, headers):
-            self.headers = headers
-
-    request = MockRequest({"x-user-id": "alice"})
+    """X-User-Id header with value returns prefixed owner ID when from trusted proxy."""
+    request = MockRequest({"x-user-id": "alice"}, client_host="127.0.0.1")
     assert resolve_owner_id(request) == "user:alice"
 
 
 def test_resolve_owner_id_trims_whitespace():
     """Leading/trailing whitespace in X-User-Id is stripped."""
-    class MockRequest:
-        def __init__(self, headers):
-            self.headers = headers
-
-    request = MockRequest({"x-user-id": "  bob  "})
+    request = MockRequest({"x-user-id": "  bob  "}, client_host="127.0.0.1")
     assert resolve_owner_id(request) == "user:bob"
 
 
 def test_resolve_owner_id_whitespace_only():
     """Whitespace-only X-User-Id falls back to DEFAULT_OWNER_ID."""
-    class MockRequest:
-        def __init__(self, headers):
-            self.headers = headers
-
     request = MockRequest({"x-user-id": "   "})
     assert resolve_owner_id(request) == DEFAULT_OWNER_ID
 
 
 def test_resolve_owner_id_empty_header():
     """Empty X-User-Id falls back to DEFAULT_OWNER_ID."""
-    class MockRequest:
-        def __init__(self, headers):
-            self.headers = headers
-
     request = MockRequest({"x-user-id": ""})
     assert resolve_owner_id(request) == DEFAULT_OWNER_ID
 
 
 def test_resolve_owner_id_missing_header():
     """Missing X-User-Id falls back to DEFAULT_OWNER_ID."""
-    class MockRequest:
-        def __init__(self, headers):
-            self.headers = headers
-
     request = MockRequest({})
     assert resolve_owner_id(request) == DEFAULT_OWNER_ID
 
@@ -73,13 +60,62 @@ def test_resolve_owner_id_no_request():
 
 
 def test_resolve_owner_id_prefix_format():
-    """Resolved owner ID always starts with 'user:' prefix."""
-    class MockRequest:
-        def __init__(self, headers):
-            self.headers = headers
-
+    """Resolved owner ID always starts with 'user:' prefix when from trusted proxy."""
     for user_id in ["alice", "bob", "test-user-123", "UPPERCASE"]:
-        request = MockRequest({"x-user-id": user_id})
+        request = MockRequest({"x-user-id": user_id}, client_host="127.0.0.1")
         result = resolve_owner_id(request)
         assert result.startswith("user:"), f"failed for {user_id}"
         assert result == f"user:{user_id.strip()}"
+
+
+# ── trusted-proxy validation (Issue #2021 BOLA fix) ───────────────────────────
+
+
+def test_resolve_owner_id_rejects_untrusted_proxy():
+    """X-User-Id from a non-trusted IP is rejected — request scoped to default."""
+    request = MockRequest({"x-user-id": "alice"}, client_host="203.0.113.50")
+    assert resolve_owner_id(request) == DEFAULT_OWNER_ID
+
+
+def test_resolve_owner_id_accepts_trusted_proxy():
+    """X-User-Id from a trusted proxy IP is accepted."""
+    request = MockRequest({"x-user-id": "alice"}, client_host="127.0.0.1")
+    assert resolve_owner_id(request) == "user:alice"
+
+
+def test_resolve_owner_id_accepts_ipv6_trusted_proxy():
+    """X-User-Id from an IPv6 loopback trusted proxy is accepted."""
+    request = MockRequest({"x-user-id": "alice"}, client_host="::1")
+    assert resolve_owner_id(request) == "user:alice"
+
+
+def test_resolve_owner_id_no_client_always_default():
+    """Request with no client info always falls back to DEFAULT_OWNER_ID."""
+    class NoClientRequest:
+        def __init__(self, headers):
+            self.headers = headers
+            self.client = None
+
+    request = NoClientRequest({"x-user-id": "alice"})
+    assert resolve_owner_id(request) == DEFAULT_OWNER_ID
+
+
+def test_resolve_owner_id_attacker_impersonation_blocked():
+    """REGRESSION: attacker with shared API key cannot spoof X-User-Id
+    to access another workspace's data.  The header is ignored because
+    the attacker's IP is not a trusted proxy."""
+    # Attacker sends X-User-Id from their own machine (not a proxy)
+    attacker_request = MockRequest(
+        {"x-user-id": "team-alpha"}, client_host="198.51.100.23"
+    )
+    result = resolve_owner_id(attacker_request)
+    assert result == DEFAULT_OWNER_ID, (
+        "Attacker should NOT be able to impersonate team-alpha via X-User-Id"
+    )
+
+    # Legitimate request from the trusted proxy succeeds
+    proxy_request = MockRequest(
+        {"x-user-id": "team-alpha"}, client_host="127.0.0.1"
+    )
+    result = resolve_owner_id(proxy_request)
+    assert result == "user:team-alpha"