diff --git a/backend/secuscan/auth.py b/backend/secuscan/auth.py index b4ef744c..72caf084 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 @@ -221,13 +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 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. @@ -235,13 +239,39 @@ def get_api_key() -> str | None: _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 diff --git a/backend/secuscan/config.py b/backend/secuscan/config.py index e37be854..6659a1fb 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 da7d6844..719669da 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 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 544b6f30..ead51aa3 100644 --- a/testing/backend/unit/test_auth_helpers.py +++ b/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: when X-User-Id header is present +- 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 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:' 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:' 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): diff --git a/testing/backend/unit/test_auth_owner_resolution.py b/testing/backend/unit/test_auth_owner_resolution.py index a1b1a9b2..d854d64f 100644 --- a/testing/backend/unit/test_auth_owner_resolution.py +++ b/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"