# -*- coding: utf-8 -*-
# Copyright (c) 2025 ContextForge Contributors.
# SPDX-License-Identifier: Apache-2.0
"""Integration tests for RBAC enforcement on management endpoints.
HTTP-level tests using TestClient(app) with temp SQLite and real RBAC decorators.
Verifies that teams, tokens, and RBAC admin endpoints enforce permissions correctly.
"""
# Future
from __future__ import annotations
# Standard
import os
import tempfile
from unittest.mock import MagicMock
import uuid
# Third-Party
from _pytest.monkeypatch import MonkeyPatch
from fastapi.testclient import TestClient
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
# First-Party
from mcpgateway.main import app
from mcpgateway.middleware.rbac import get_current_user_with_permissions
from mcpgateway.middleware.rbac import get_db as rbac_get_db
from mcpgateway.middleware.rbac import PermissionService
from mcpgateway.utils.verify_credentials import require_auth
def _create_user_context(email: str, is_admin: bool = False, session_factory=None, team_id=None):
"""Create a mock user context generator for dependency override."""
async def mock_user_with_permissions():
db_session = session_factory() if session_factory else None
try:
yield {
"email": email,
"full_name": f"Test {email.split('@')[0]}",
"is_admin": is_admin,
"ip_address": "127.0.0.1",
"user_agent": "test-client",
"auth_method": "jwt",
"db": db_session,
"token_use": "session",
"team_id": team_id,
}
finally:
if db_session is not None:
db_session.close()
return mock_user_with_permissions
class _TestPermissionService(PermissionService):
"""Test PermissionService that uses the test DB session factory."""
_session_factory = None
def __init__(self, db=None, audit_enabled=None):
# Use test session factory if the passed db is not from our test engine
if self._session_factory:
test_db = self._session_factory()
super().__init__(test_db, audit_enabled=False)
else:
super().__init__(db, audit_enabled=audit_enabled)
@pytest.fixture
def rbac_test_env():
"""Create a test environment with real RBAC decorators and temp SQLite."""
mp = MonkeyPatch()
fd, path = tempfile.mkstemp(suffix=".db")
url = f"sqlite:///{path}"
# First-Party
from mcpgateway.config import settings
mp.setattr(settings, "database_url", url, raising=False)
# First-Party
import mcpgateway.db as db_mod
import mcpgateway.main as main_mod
engine = create_engine(url, connect_args={"check_same_thread": False}, poolclass=StaticPool)
TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
mp.setattr(db_mod, "engine", engine, raising=False)
mp.setattr(db_mod, "SessionLocal", TestSessionLocal, raising=False)
mp.setattr(main_mod, "SessionLocal", TestSessionLocal, raising=False)
mp.setattr(main_mod, "engine", engine, raising=False)
db_mod.Base.metadata.create_all(bind=engine)
def override_get_db():
db = TestSessionLocal()
try:
yield db
finally:
db.close()
app.dependency_overrides[rbac_get_db] = override_get_db
# Also override get_db in routers that define their own
# First-Party
from mcpgateway.routers.rbac import get_db as rbac_router_get_db
app.dependency_overrides[rbac_router_get_db] = override_get_db
# Override get_db in auth router
try:
# First-Party
from mcpgateway.routers.email_auth import get_db as email_auth_get_db
app.dependency_overrides[email_auth_get_db] = override_get_db
except ImportError:
pass
# Patch PermissionService to use test DB
_TestPermissionService._session_factory = TestSessionLocal
mp.setattr("mcpgateway.middleware.rbac.PermissionService", _TestPermissionService)
# Bootstrap admin user and roles in the DB
db = TestSessionLocal()
# First-Party
from mcpgateway.db import EmailUser, Role, UserRole
admin = EmailUser(email="admin@test.local", password_hash="$argon2id$v=19$m=65536,t=3,p=1$test", full_name="Admin", is_admin=True, is_active=True)
db.add(admin)
nonadmin = EmailUser(email="user@test.local", password_hash="$argon2id$v=19$m=65536,t=3,p=1$test", full_name="User", is_admin=False, is_active=True)
db.add(nonadmin)
db.flush()
# Create platform_admin role and assign to admin
pa_role = Role(id=str(uuid.uuid4()), name="platform_admin", description="All", scope="global", permissions=["*"], created_by="admin@test.local", is_system_role=True, is_active=True)
db.add(pa_role)
db.flush()
pa_ur = UserRole(user_email="admin@test.local", role_id=pa_role.id, scope="global", scope_id=None, granted_by="admin@test.local", is_active=True)
db.add(pa_ur)
db.commit()
db.close()
def setup_user(email, is_admin=False, team_id=None):
"""Set the authenticated user for the test."""
mock_user = MagicMock()
mock_user.email = email
# First-Party
from mcpgateway.auth import get_current_user
app.dependency_overrides[require_auth] = lambda: email
app.dependency_overrides[get_current_user] = lambda: mock_user
app.dependency_overrides[get_current_user_with_permissions] = _create_user_context(email, is_admin=is_admin, session_factory=TestSessionLocal, team_id=team_id)
yield TestSessionLocal, engine, setup_user
app.dependency_overrides.clear()
_TestPermissionService._session_factory = None
mp.undo()
engine.dispose()
os.close(fd)
os.unlink(path)
# ---------------------------------------------------------------------------
# D7.1: RBAC Admin Endpoint Access
# ---------------------------------------------------------------------------
class TestRBACAdminEndpointAccess:
"""Test RBAC enforcement on RBAC admin endpoints."""
def test_non_admin_cannot_list_roles(self, rbac_test_env):
"""Non-admin user should get 403 on role listing."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("user@test.local", is_admin=False)
client = TestClient(app)
response = client.get("/rbac/roles", headers={"Authorization": "Bearer test"})
assert response.status_code == 403
def test_admin_can_list_roles(self, rbac_test_env):
"""Admin user should be able to access the roles endpoint (not 403)."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("admin@test.local", is_admin=True)
client = TestClient(app)
response = client.get("/rbac/roles", headers={"Authorization": "Bearer test"})
# The important test: admin is NOT denied (not 403).
# May get 200 or 500 (DetachedInstanceError from session scoping) — RBAC passed either way.
assert response.status_code != 403, f"Admin should NOT be denied access to /rbac/roles (got {response.status_code})"
def test_non_admin_cannot_list_user_roles(self, rbac_test_env):
"""Non-admin user should get 403 on user role listing."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("user@test.local", is_admin=False)
client = TestClient(app)
response = client.get("/rbac/users/user@test.local/roles", headers={"Authorization": "Bearer test"})
assert response.status_code == 403
def test_non_admin_cannot_assign_roles(self, rbac_test_env):
"""Non-admin user should get 403 on role assignment."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("user@test.local", is_admin=False)
client = TestClient(app)
response = client.post(
"/rbac/users/user@test.local/roles",
json={"role_id": "some-role-id", "scope": "global"},
headers={"Authorization": "Bearer test"},
)
assert response.status_code == 403
# ---------------------------------------------------------------------------
# D7.2: Email Auth Admin Endpoint Access
# ---------------------------------------------------------------------------
class TestEmailAuthAdminEndpoints:
"""Test RBAC enforcement on email auth admin endpoints."""
def test_non_admin_cannot_list_users(self, rbac_test_env):
"""Non-admin should get 403 on admin user listing."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("user@test.local", is_admin=False)
client = TestClient(app)
response = client.get("/auth/email/admin/users", headers={"Authorization": "Bearer test"})
assert response.status_code == 403
def test_admin_can_list_users(self, rbac_test_env):
"""Admin should be able to list users."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("admin@test.local", is_admin=True)
client = TestClient(app)
response = client.get("/auth/email/admin/users", headers={"Authorization": "Bearer test"})
assert response.status_code == 200
# ---------------------------------------------------------------------------
# D7.3: Cross-endpoint consistency (deny without auth)
# ---------------------------------------------------------------------------
class TestCrossEndpointConsistency:
"""Verify consistent RBAC behavior across different endpoint types."""
@pytest.mark.parametrize(
"endpoint",
[
"/rbac/roles",
"/rbac/my/roles",
],
)
def test_unauthenticated_user_denied(self, rbac_test_env, endpoint):
"""Endpoints should deny access when no user context is set."""
TestSessionLocal, engine, setup_user = rbac_test_env
# Don't set up any user
app.dependency_overrides.pop(get_current_user_with_permissions, None)
client = TestClient(app, raise_server_exceptions=False)
response = client.get(endpoint)
# Should get 401 or 403 (not 200 or 500)
assert response.status_code in (401, 403, 422), f"Expected auth error for GET {endpoint}, got {response.status_code}"
# ---------------------------------------------------------------------------
# D7.4: Permission denial propagation
# ---------------------------------------------------------------------------
class TestPermissionDenialPropagation:
"""Verify that RBAC 403 responses have correct format."""
def test_403_response_includes_detail(self, rbac_test_env):
"""403 responses should include detail about the required permission."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("user@test.local", is_admin=False)
client = TestClient(app)
response = client.get("/rbac/roles", headers={"Authorization": "Bearer test"})
assert response.status_code == 403
data = response.json()
assert "detail" in data
assert "permission" in data["detail"].lower() or "admin" in data["detail"].lower()
def test_admin_bypass_works_for_api_endpoints(self, rbac_test_env):
"""Admin should bypass permission checks on API endpoints (not 403)."""
TestSessionLocal, engine, setup_user = rbac_test_env
setup_user("admin@test.local", is_admin=True)
client = TestClient(app)
# Admin should pass RBAC checks (not get 403)
response = client.get("/rbac/roles", headers={"Authorization": "Bearer test"})
assert response.status_code != 403, f"Admin should NOT be denied access (got {response.status_code})"
