Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
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
10 changes: 10 additions & 0 deletions finbot/mcp/servers/systemutils/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@

logger = logging.getLogger(__name__)

PROTECTED_USERNAMES: set[str] = {"admin", "root", "system"}

DEFAULT_CONFIG: dict[str, Any] = {
"enabled_tools": [
"run_diagnostics",
Expand Down Expand Up @@ -194,6 +196,14 @@ def manage_users(action: str, username: str, role: str = "viewer") -> dict[str,
Actions: 'create', 'modify', 'delete', 'list'.
Roles: 'viewer', 'operator', 'admin', 'superadmin'.
"""
if (
action
and username
and action.strip().lower() == "delete"
and username.strip().lower() in PROTECTED_USERNAMES
):
raise ValueError(f"User '{username}' is protected and cannot be deleted")

logger.warning(
"SystemUtils manage_users called with action='%s', username='%s', role='%s' by namespace='%s'",
action,
Expand Down
115 changes: 115 additions & 0 deletions tests/unit/mcp/test_systemutils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,115 @@
"""Unit tests for SystemUtils MCP Server manage_users operations.

Validates that SystemUtils manage_users protects reserved system accounts
from deletion while allowing valid user lifecycle operations.
"""

from unittest.mock import MagicMock

import pytest
from fastmcp import Client
from fastmcp.exceptions import ToolError

from finbot.core.auth.session import SessionContext
from finbot.mcp.servers.systemutils.server import create_systemutils_server


@pytest.fixture
def session_context():
"""Create a mock SessionContext with a valid namespace."""
ctx = MagicMock(spec=SessionContext)
ctx.namespace = "test-systemutils-namespace"
ctx.user_id = "test-user-001"
ctx.session_id = "test-session-001"
return ctx


class TestManageUsers:
"""Tests for manage_users tool validation and protection."""

@pytest.mark.asyncio
async def test_su_usr_001_returns_expected_fields(self, session_context):
"""SU-USR-001: manage_users returns a complete result for a valid operation."""
server = create_systemutils_server(session_context)
async with Client(server) as client:
result = await client.call_tool(
"manage_users",
{"action": "list", "username": "admin", "role": "viewer"},
)
data = result.data
assert data["status"] == "completed"
assert data["action"] == "list"
assert data["username"] == "admin"
assert data["role"] == "viewer"
assert "timestamp" in data

@pytest.mark.asyncio
async def test_su_usr_002_delete_admin_rejected_as_protected(self, session_context):
"""SU-USR-002: Deleting admin user must be rejected as protected."""
server = create_systemutils_server(session_context)
async with Client(server) as client:
with pytest.raises((ValueError, ToolError), match="protected"):
await client.call_tool(
"manage_users",
{"action": "delete", "username": "admin"},
)

@pytest.mark.asyncio
async def test_su_usr_002_delete_admin_accepted_without_validation(self, session_context):
"""SU-USR-002 Acceptance alias: verify deleting admin raises an error."""
await self.test_su_usr_002_delete_admin_rejected_as_protected(session_context)

@pytest.mark.asyncio
@pytest.mark.parametrize(
"protected_name",
[
"admin",
"Admin",
"ADMIN",
"root",
"Root",
"ROOT",
"system",
"System",
"SYSTEM",
" admin ",
],
)
async def test_protected_usernames_rejected_case_insensitively(
self, session_context, protected_name
):
"""All reserved accounts (admin, root, system) are blocked from deletion case-insensitively."""
server = create_systemutils_server(session_context)
async with Client(server) as client:
with pytest.raises((ValueError, ToolError), match="protected"):
await client.call_tool(
"manage_users",
{"action": "delete", "username": protected_name},
)

@pytest.mark.asyncio
async def test_deleting_regular_user_succeeds(self, session_context):
"""Deleting a standard, non-protected account is allowed."""
server = create_systemutils_server(session_context)
async with Client(server) as client:
result = await client.call_tool(
"manage_users",
{"action": "delete", "username": "temp_guest"},
)
assert result.data["status"] == "completed"
assert result.data["action"] == "delete"
assert result.data["username"] == "temp_guest"

@pytest.mark.asyncio
@pytest.mark.parametrize("action", ["list", "create", "modify"])
async def test_non_delete_actions_on_protected_user_allowed(self, session_context, action):
"""Actions other than delete (list, create, modify) on protected accounts are not blocked."""
server = create_systemutils_server(session_context)
async with Client(server) as client:
result = await client.call_tool(
"manage_users",
{"action": action, "username": "admin", "role": "admin"},
)
assert result.data["status"] == "completed"
assert result.data["action"] == action
assert result.data["username"] == "admin"