Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 42 additions & 2 deletions backend/secuscan/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
import base64
import hmac
import json
import logging
import os
import secrets
import time
Expand Down Expand Up @@ -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


Expand Down
7 changes: 7 additions & 0 deletions backend/secuscan/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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] = []
Comment thread
namann5 marked this conversation as resolved.

# Sandbox
docker_enabled: bool = False
sandbox_timeout: int = 600 # seconds
Expand Down
2 changes: 2 additions & 0 deletions testing/backend/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down
39 changes: 31 additions & 8 deletions testing/backend/unit/test_auth_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:<id> when X-User-Id header is present
- resolve_owner_id returns user:<id> 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

Expand All @@ -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:<id>' when X-User-Id is set."""
def test_returns_user_prefix_when_header_present_and_trusted(self):
"""resolve_owner_id returns 'user:<id>' 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:
Expand Down
88 changes: 56 additions & 32 deletions testing/backend/unit/test_auth_owner_resolution.py
Original file line number Diff line number Diff line change
@@ -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


Expand All @@ -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

Expand All @@ -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"
Loading