|
|
"""Tests for System User Protection in user_manager.py |
|
|
|
|
|
Tests cover: |
|
|
- get_request_user_id(): 1st defense layer - blocks System Users from HTTP headers |
|
|
- get_request_user_filepath(): 2nd defense layer - structural blocking via get_public_user_directory() |
|
|
- add_user(): 3rd defense layer - prevents creation of System User names |
|
|
- Defense layers integration tests |
|
|
""" |
|
|
|
|
|
import pytest |
|
|
from unittest.mock import MagicMock, patch |
|
|
import tempfile |
|
|
|
|
|
import folder_paths |
|
|
from app.user_manager import UserManager |
|
|
|
|
|
|
|
|
@pytest.fixture |
|
|
def mock_user_directory(): |
|
|
"""Create a temporary user directory.""" |
|
|
with tempfile.TemporaryDirectory() as temp_dir: |
|
|
original_dir = folder_paths.get_user_directory() |
|
|
folder_paths.set_user_directory(temp_dir) |
|
|
yield temp_dir |
|
|
folder_paths.set_user_directory(original_dir) |
|
|
|
|
|
|
|
|
@pytest.fixture |
|
|
def user_manager(mock_user_directory): |
|
|
"""Create a UserManager instance for testing.""" |
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
manager = UserManager() |
|
|
|
|
|
manager.users = {"default": "default", "test_user_123": "Test User"} |
|
|
yield manager |
|
|
|
|
|
|
|
|
@pytest.fixture |
|
|
def mock_request(): |
|
|
"""Create a mock request object.""" |
|
|
request = MagicMock() |
|
|
request.headers = {} |
|
|
return request |
|
|
|
|
|
|
|
|
class TestGetRequestUserId: |
|
|
"""Tests for get_request_user_id() - 1st defense layer. |
|
|
|
|
|
Verifies: |
|
|
- System Users (__ prefix) in HTTP header are rejected with KeyError |
|
|
- Public Users pass through successfully |
|
|
""" |
|
|
|
|
|
def test_system_user_raises_error(self, user_manager, mock_request): |
|
|
"""Test System User in header raises KeyError.""" |
|
|
mock_request.headers = {"comfy-user": "__system"} |
|
|
|
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
with pytest.raises(KeyError, match="Unknown user"): |
|
|
user_manager.get_request_user_id(mock_request) |
|
|
|
|
|
def test_system_user_cache_raises_error(self, user_manager, mock_request): |
|
|
"""Test System User cache raises KeyError.""" |
|
|
mock_request.headers = {"comfy-user": "__cache"} |
|
|
|
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
with pytest.raises(KeyError, match="Unknown user"): |
|
|
user_manager.get_request_user_id(mock_request) |
|
|
|
|
|
def test_normal_user_works(self, user_manager, mock_request): |
|
|
"""Test normal user access works.""" |
|
|
mock_request.headers = {"comfy-user": "default"} |
|
|
|
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
user_id = user_manager.get_request_user_id(mock_request) |
|
|
assert user_id == "default" |
|
|
|
|
|
def test_unknown_user_raises_error(self, user_manager, mock_request): |
|
|
"""Test unknown user raises KeyError.""" |
|
|
mock_request.headers = {"comfy-user": "unknown_user"} |
|
|
|
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
with pytest.raises(KeyError, match="Unknown user"): |
|
|
user_manager.get_request_user_id(mock_request) |
|
|
|
|
|
|
|
|
class TestGetRequestUserFilepath: |
|
|
"""Tests for get_request_user_filepath() - 2nd defense layer. |
|
|
|
|
|
Verifies: |
|
|
- Returns None when get_public_user_directory() returns None (System User) |
|
|
- Acts as backup defense if 1st layer is bypassed |
|
|
""" |
|
|
|
|
|
def test_system_user_returns_none(self, user_manager, mock_request, mock_user_directory): |
|
|
"""Test System User returns None (structural blocking).""" |
|
|
|
|
|
|
|
|
|
|
|
mock_request.headers = {"comfy-user": "default"} |
|
|
|
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
|
|
|
with patch.object(folder_paths, 'get_public_user_directory', return_value=None): |
|
|
result = user_manager.get_request_user_filepath(mock_request, "test.txt") |
|
|
assert result is None |
|
|
|
|
|
def test_normal_user_gets_path(self, user_manager, mock_request, mock_user_directory): |
|
|
"""Test normal user gets valid filepath.""" |
|
|
mock_request.headers = {"comfy-user": "default"} |
|
|
|
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
path = user_manager.get_request_user_filepath(mock_request, "test.txt") |
|
|
assert path is not None |
|
|
assert "default" in path |
|
|
assert path.endswith("test.txt") |
|
|
|
|
|
|
|
|
class TestAddUser: |
|
|
"""Tests for add_user() - 3rd defense layer (creation-time blocking). |
|
|
|
|
|
Verifies: |
|
|
- System User name (__ prefix) creation is rejected with ValueError |
|
|
- Sanitized usernames that become System User are also rejected |
|
|
""" |
|
|
|
|
|
def test_system_user_prefix_name_raises(self, user_manager): |
|
|
"""Test System User prefix in name raises ValueError.""" |
|
|
with pytest.raises(ValueError, match="System User prefix not allowed"): |
|
|
user_manager.add_user("__system") |
|
|
|
|
|
def test_system_user_prefix_cache_raises(self, user_manager): |
|
|
"""Test System User cache prefix raises ValueError.""" |
|
|
with pytest.raises(ValueError, match="System User prefix not allowed"): |
|
|
user_manager.add_user("__cache") |
|
|
|
|
|
def test_sanitized_system_user_prefix_raises(self, user_manager): |
|
|
"""Test sanitized name becoming System User prefix raises ValueError (bypass prevention).""" |
|
|
|
|
|
with pytest.raises(ValueError, match="System User prefix not allowed"): |
|
|
user_manager.add_user("__test") |
|
|
|
|
|
def test_normal_user_creation(self, user_manager, mock_user_directory): |
|
|
"""Test normal user creation works.""" |
|
|
user_id = user_manager.add_user("Normal User") |
|
|
assert user_id is not None |
|
|
assert not user_id.startswith("__") |
|
|
assert "Normal-User" in user_id or "Normal_User" in user_id |
|
|
|
|
|
def test_empty_name_raises(self, user_manager): |
|
|
"""Test empty name raises ValueError.""" |
|
|
with pytest.raises(ValueError, match="username not provided"): |
|
|
user_manager.add_user("") |
|
|
|
|
|
def test_whitespace_only_raises(self, user_manager): |
|
|
"""Test whitespace-only name raises ValueError.""" |
|
|
with pytest.raises(ValueError, match="username not provided"): |
|
|
user_manager.add_user(" ") |
|
|
|
|
|
|
|
|
class TestDefenseLayers: |
|
|
"""Integration tests for all three defense layers. |
|
|
|
|
|
Verifies: |
|
|
- Each defense layer blocks System Users independently |
|
|
- System User bypass is impossible through any layer |
|
|
""" |
|
|
|
|
|
def test_layer1_get_request_user_id(self, user_manager, mock_request): |
|
|
"""Test 1st defense layer blocks System Users.""" |
|
|
mock_request.headers = {"comfy-user": "__system"} |
|
|
|
|
|
with patch('app.user_manager.args') as mock_args: |
|
|
mock_args.multi_user = True |
|
|
with pytest.raises(KeyError): |
|
|
user_manager.get_request_user_id(mock_request) |
|
|
|
|
|
def test_layer2_get_public_user_directory(self): |
|
|
"""Test 2nd defense layer blocks System Users.""" |
|
|
result = folder_paths.get_public_user_directory("__system") |
|
|
assert result is None |
|
|
|
|
|
def test_layer3_add_user(self, user_manager): |
|
|
"""Test 3rd defense layer blocks System User creation.""" |
|
|
with pytest.raises(ValueError): |
|
|
user_manager.add_user("__system") |
|
|
|