From 1039bc24df68826da5c47974460f46875c026833 Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Tue, 14 Jul 2026 11:09:49 +0530 Subject: [PATCH 1/2] fix(security): validate X-User-Id against trusted owner whitelist to prevent BOLA The X-User-Id header that scopes all data access (vault secrets, tasks, findings, reports) was accepted without verification, allowing any caller with the shared API key to read another workspace's data by spoofing the header. This fix adds a configurable trusted_owner_ids whitelist: - Empty (default): X-User-Id is ignored, all requests use DEFAULT_OWNER_ID - Populated: only listed owner IDs are accepted; others fall back to default Resolves: #2021 --- backend/secuscan/auth.py | 44 +++++++++- backend/secuscan/config.py | 7 ++ testing/backend/conftest.py | 2 + testing/backend/unit/test_auth_helpers.py | 39 ++++++-- .../unit/test_auth_owner_resolution.py | 88 ++++++++++++------- 5 files changed, 138 insertions(+), 42 deletions(-) diff --git a/backend/secuscan/auth.py b/backend/secuscan/auth.py index b4ef744cb..158384fb1 100644 --- a/backend/secuscan/auth.py +++ b/backend/secuscan/auth.py @@ -15,6 +15,7 @@ import base64 import hmac import json +import logging import os import secrets import time @@ -229,19 +230,58 @@ def get_api_key() -> str | None: # 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. # +# SECURITY FIX (Issue #2021): The X-User-Id header is now validated against a +# configurable whitelist (``trusted_owner_ids``). When the whitelist is empty +# (the default), the header is ignored and all requests use DEFAULT_OWNER_ID, +# preventing cross-workspace BOLA attacks. Operators who need multi-user +# isolation must explicitly populate the whitelist with the allowed workspace +# IDs. +# # 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" +_logger = logging.getLogger(__name__) + 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 its value appears in the + ``trusted_owner_ids`` configuration list. An unrecognised value is + rejected with a logged warning and the request is scoped to the default + owner — this prevents an attacker who knows the shared API key from + reading another workspace's vault secrets, tasks, and reports (BOLA). + """ 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()}" + user_id = user_id.strip() + # Lazy-import to avoid circular dependency at module level. + from .config import settings + + trusted = settings.trusted_owner_ids + if not trusted: + _logger.warning( + "Ignoring X-User-Id header %r — no trusted_owner_ids " + "are configured; all requests are scoped to the default " + "owner. Set SECUSCAN_TRUSTED_OWNER_IDS to enable " + "multi-user isolation.", + user_id, + ) + return DEFAULT_OWNER_ID + + if user_id not in trusted: + _logger.warning( + "Rejected X-User-Id header %r — not in trusted_owner_ids " + "whitelist; request scoped to default owner.", + user_id, + ) + return DEFAULT_OWNER_ID + + return f"user:{user_id}" return DEFAULT_OWNER_ID diff --git a/backend/secuscan/config.py b/backend/secuscan/config.py index e37be8543..6659a1fb6 100644 --- a/backend/secuscan/config.py +++ b/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 diff --git a/testing/backend/conftest.py b/testing/backend/conftest.py index da7d68448..9408f2c30 100644 --- a/testing/backend/conftest.py +++ b/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 alice/bob as trusted owner IDs for multi-user integration tests + monkeypatch.setattr(settings, "trusted_owner_ids", ["alice", "bob"]) settings.ensure_directories() diff --git a/testing/backend/unit/test_auth_helpers.py b/testing/backend/unit/test_auth_helpers.py index 544b6f303..4737e787f 100644 --- a/testing/backend/unit/test_auth_helpers.py +++ b/testing/backend/unit/test_auth_helpers.py @@ -4,12 +4,13 @@ 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: when X-User-Id header is present +- resolve_owner_id returns user: when X-User-Id header is present and trusted - resolve_owner_id strips whitespace from user ID +- resolve_owner_id rejects untrusted X-User-Id values (BOLA fix) - get_api_key returns the current API key or None when not initialised """ -from unittest.mock import MagicMock +from unittest.mock import MagicMock, patch from backend.secuscan import auth @@ -34,19 +35,41 @@ def test_returns_default_when_header_empty(self): result = auth.resolve_owner_id(mock_request) assert result == auth.DEFAULT_OWNER_ID - def test_returns_user_prefix_when_header_present(self): - """resolve_owner_id returns 'user:' when X-User-Id is set.""" + def test_returns_user_prefix_when_header_present_and_trusted(self): + """resolve_owner_id returns 'user:' when X-User-Id is set and trusted.""" mock_request = MagicMock() mock_request.headers = {"x-user-id": "alice"} - result = auth.resolve_owner_id(mock_request) - assert result == "user:alice" + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = ["alice", "bob"] + result = auth.resolve_owner_id(mock_request) + 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) - assert result == "user:bob" + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = ["bob"] + result = auth.resolve_owner_id(mock_request) + assert result == "user:bob" + + def test_returns_default_when_user_not_in_trusted_list(self): + """resolve_owner_id returns DEFAULT_OWNER_ID for untrusted X-User-Id.""" + mock_request = MagicMock() + mock_request.headers = {"x-user-id": "attacker"} + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = ["alice", "bob"] + result = auth.resolve_owner_id(mock_request) + assert result == auth.DEFAULT_OWNER_ID + + def test_returns_default_when_trusted_list_is_empty(self): + """resolve_owner_id ignores X-User-Id when trusted_owner_ids is empty.""" + mock_request = MagicMock() + mock_request.headers = {"x-user-id": "alice"} + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = [] + result = auth.resolve_owner_id(mock_request) + assert result == auth.DEFAULT_OWNER_ID class TestGetApiKey: diff --git a/testing/backend/unit/test_auth_owner_resolution.py b/testing/backend/unit/test_auth_owner_resolution.py index a1b1a9b23..4c1888cb9 100644 --- a/testing/backend/unit/test_auth_owner_resolution.py +++ b/testing/backend/unit/test_auth_owner_resolution.py @@ -1,9 +1,11 @@ """ Unit tests for auth.py owner-resolution helpers. -Covers: resolve_owner_id, DEFAULT_OWNER_ID +Covers: resolve_owner_id, DEFAULT_OWNER_ID, trusted_owner_ids whitelist """ +from unittest.mock import patch + from backend.secuscan.auth import resolve_owner_id, DEFAULT_OWNER_ID @@ -17,52 +19,41 @@ def test_default_owner_id_value(): # ── resolve_owner_id ────────────────────────────────────────────────────────── -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 +class MockRequest: + def __init__(self, headers): + self.headers = headers + +def test_resolve_owner_id_with_x_user_id_header(): + """X-User-Id header with value returns prefixed owner ID when trusted.""" request = MockRequest({"x-user-id": "alice"}) - assert resolve_owner_id(request) == "user:alice" + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = ["alice", "bob"] + 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 "}) - assert resolve_owner_id(request) == "user:bob" + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = ["bob"] + 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 +64,46 @@ 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 trusted.""" for user_id in ["alice", "bob", "test-user-123", "UPPERCASE"]: request = MockRequest({"x-user-id": user_id}) - result = resolve_owner_id(request) - assert result.startswith("user:"), f"failed for {user_id}" - assert result == f"user:{user_id.strip()}" + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = [user_id] + result = resolve_owner_id(request) + assert result.startswith("user:"), f"failed for {user_id}" + assert result == f"user:{user_id.strip()}" + + +# ── trusted_owner_ids whitelist (Issue #2021 BOLA fix) ──────────────────────── + + +def test_resolve_owner_id_rejects_untrusted_user(): + """Untrusted X-User-Id falls back to DEFAULT_OWNER_ID.""" + request = MockRequest({"x-user-id": "attacker"}) + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = ["alice", "bob"] + assert resolve_owner_id(request) == DEFAULT_OWNER_ID + + +def test_resolve_owner_id_empty_whitelist_ignores_header(): + """Empty trusted_owner_ids means X-User-Id is always ignored.""" + request = MockRequest({"x-user-id": "alice"}) + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = [] + assert resolve_owner_id(request) == DEFAULT_OWNER_ID + + +def test_resolve_owner_id_no_whitelist_config_ignores_header(): + """Missing trusted_owner_ids config means X-User-Id is always ignored.""" + request = MockRequest({"x-user-id": "alice"}) + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = None + assert resolve_owner_id(request) == DEFAULT_OWNER_ID + + +def test_resolve_owner_id_trusted_user_accepted(): + """Trusted X-User-Id is accepted and returned with user: prefix.""" + request = MockRequest({"x-user-id": "team-alpha"}) + with patch("backend.secuscan.config.settings") as mock_settings: + mock_settings.trusted_owner_ids = ["team-alpha", "team-beta"] + assert resolve_owner_id(request) == "user:team-alpha" From 8dce321c527f17e80e60080bfb982df6d3ca8f8d Mon Sep 17 00:00:00 2001 From: Naman Singh Date: Wed, 15 Jul 2026 13:59:35 +0530 Subject: [PATCH 2/2] fix(security): bind X-User-Id to trusted proxy IP instead of client-controlled whitelist Replace the trusted_owner_ids whitelist approach with trusted-proxy validation. The whitelist was bypassable because any client holding the shared API key could spoof X-User-Id: team-alpha. Now resolve_owner_id() only accepts X-User-Id when request.client.host is in settings.trusted_proxies. This binds owner identity to an authenticated infrastructure component (reverse proxy / SSO layer) rather than trusting a client-controlled header. Changes: - auth.py: Add _is_trusted_proxy(), validate source before accepting header - test_auth_owner_resolution.py: Add proxy-based tests + attacker regression test - test_auth_helpers.py: Update tests for trusted-proxy model - conftest.py: Add 'testclient' to trusted_proxies for integration tests --- backend/secuscan/auth.py | 76 ++++++------ testing/backend/conftest.py | 4 +- testing/backend/unit/test_auth_helpers.py | 76 ++++++------ .../unit/test_auth_owner_resolution.py | 110 ++++++++++-------- 4 files changed, 133 insertions(+), 133 deletions(-) diff --git a/backend/secuscan/auth.py b/backend/secuscan/auth.py index 158384fb1..72caf0844 100644 --- a/backend/secuscan/auth.py +++ b/backend/secuscan/auth.py @@ -222,20 +222,16 @@ 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 now validated against a -# configurable whitelist (``trusted_owner_ids``). When the whitelist is empty -# (the default), the header is ignored and all requests use DEFAULT_OWNER_ID, -# preventing cross-workspace BOLA attacks. Operators who need multi-user -# isolation must explicitly populate the whitelist with the allowed workspace -# IDs. +# 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. @@ -243,45 +239,39 @@ def get_api_key() -> str | None: _OWNER_HEADER = "x-user-id" -_logger = logging.getLogger(__name__) +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. - The ``X-User-Id`` header is only honoured when its value appears in the - ``trusted_owner_ids`` configuration list. An unrecognised value is - rejected with a logged warning and the request is scoped to the default - owner — this prevents an attacker who knows the shared API key from - reading another workspace's vault secrets, tasks, and reports (BOLA). + 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(): - user_id = user_id.strip() - # Lazy-import to avoid circular dependency at module level. - from .config import settings - - trusted = settings.trusted_owner_ids - if not trusted: - _logger.warning( - "Ignoring X-User-Id header %r — no trusted_owner_ids " - "are configured; all requests are scoped to the default " - "owner. Set SECUSCAN_TRUSTED_OWNER_IDS to enable " - "multi-user isolation.", - user_id, - ) - return DEFAULT_OWNER_ID - - if user_id not in trusted: - _logger.warning( - "Rejected X-User-Id header %r — not in trusted_owner_ids " - "whitelist; request scoped to default owner.", - user_id, - ) - return DEFAULT_OWNER_ID - - return f"user:{user_id}" + 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 diff --git a/testing/backend/conftest.py b/testing/backend/conftest.py index 9408f2c30..719669dac 100644 --- a/testing/backend/conftest.py +++ b/testing/backend/conftest.py @@ -38,8 +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 alice/bob as trusted owner IDs for multi-user integration tests - monkeypatch.setattr(settings, "trusted_owner_ids", ["alice", "bob"]) + # 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() diff --git a/testing/backend/unit/test_auth_helpers.py b/testing/backend/unit/test_auth_helpers.py index 4737e787f..ead51aa3e 100644 --- a/testing/backend/unit/test_auth_helpers.py +++ b/testing/backend/unit/test_auth_helpers.py @@ -4,17 +4,26 @@ 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: when X-User-Id header is present and trusted +- resolve_owner_id returns user: when X-User-Id header is present and from trusted proxy - resolve_owner_id strips whitespace from user ID -- resolve_owner_id rejects untrusted X-User-Id values (BOLA fix) +- 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 """ -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock 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.""" @@ -23,53 +32,42 @@ 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_and_trusted(self): - """resolve_owner_id returns 'user:' when X-User-Id is set and trusted.""" - mock_request = MagicMock() - mock_request.headers = {"x-user-id": "alice"} - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = ["alice", "bob"] - result = auth.resolve_owner_id(mock_request) - assert result == "user:alice" + def test_returns_user_prefix_when_header_present_and_trusted_proxy(self): + """resolve_owner_id returns 'user:' 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 "} - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = ["bob"] - result = auth.resolve_owner_id(mock_request) - assert result == "user:bob" + 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_user_not_in_trusted_list(self): - """resolve_owner_id returns DEFAULT_OWNER_ID for untrusted X-User-Id.""" - mock_request = MagicMock() - mock_request.headers = {"x-user-id": "attacker"} - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = ["alice", "bob"] - result = auth.resolve_owner_id(mock_request) - assert result == auth.DEFAULT_OWNER_ID + 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_trusted_list_is_empty(self): - """resolve_owner_id ignores X-User-Id when trusted_owner_ids is empty.""" - mock_request = MagicMock() - mock_request.headers = {"x-user-id": "alice"} - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = [] - result = auth.resolve_owner_id(mock_request) - 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: diff --git a/testing/backend/unit/test_auth_owner_resolution.py b/testing/backend/unit/test_auth_owner_resolution.py index 4c1888cb9..d854d64fc 100644 --- a/testing/backend/unit/test_auth_owner_resolution.py +++ b/testing/backend/unit/test_auth_owner_resolution.py @@ -1,14 +1,19 @@ """ Unit tests for auth.py owner-resolution helpers. -Covers: resolve_owner_id, DEFAULT_OWNER_ID, trusted_owner_ids whitelist +Covers: resolve_owner_id, DEFAULT_OWNER_ID, trusted-proxy validation (BOLA fix) """ -from unittest.mock import patch - 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 ────────────────────────────────────────────────────────── @@ -19,25 +24,16 @@ def test_default_owner_id_value(): # ── resolve_owner_id ────────────────────────────────────────────────────────── -class MockRequest: - def __init__(self, headers): - self.headers = headers - - def test_resolve_owner_id_with_x_user_id_header(): - """X-User-Id header with value returns prefixed owner ID when trusted.""" - request = MockRequest({"x-user-id": "alice"}) - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = ["alice", "bob"] - assert resolve_owner_id(request) == "user: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.""" - request = MockRequest({"x-user-id": " bob "}) - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = ["bob"] - assert resolve_owner_id(request) == "user: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(): @@ -64,46 +60,62 @@ def test_resolve_owner_id_no_request(): def test_resolve_owner_id_prefix_format(): - """Resolved owner ID always starts with 'user:' prefix when trusted.""" + """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}) - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = [user_id] - result = resolve_owner_id(request) - assert result.startswith("user:"), f"failed for {user_id}" - assert result == f"user:{user_id.strip()}" + 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_owner_ids whitelist (Issue #2021 BOLA fix) ──────────────────────── +# ── trusted-proxy validation (Issue #2021 BOLA fix) ─────────────────────────── -def test_resolve_owner_id_rejects_untrusted_user(): - """Untrusted X-User-Id falls back to DEFAULT_OWNER_ID.""" - request = MockRequest({"x-user-id": "attacker"}) - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = ["alice", "bob"] - assert resolve_owner_id(request) == DEFAULT_OWNER_ID +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_empty_whitelist_ignores_header(): - """Empty trusted_owner_ids means X-User-Id is always ignored.""" - request = MockRequest({"x-user-id": "alice"}) - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = [] - assert resolve_owner_id(request) == DEFAULT_OWNER_ID +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_whitelist_config_ignores_header(): - """Missing trusted_owner_ids config means X-User-Id is always ignored.""" - request = MockRequest({"x-user-id": "alice"}) - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = None - assert resolve_owner_id(request) == DEFAULT_OWNER_ID + +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_trusted_user_accepted(): - """Trusted X-User-Id is accepted and returned with user: prefix.""" - request = MockRequest({"x-user-id": "team-alpha"}) - with patch("backend.secuscan.config.settings") as mock_settings: - mock_settings.trusted_owner_ids = ["team-alpha", "team-beta"] - assert resolve_owner_id(request) == "user:team-alpha" +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"