feat: add RBAC enforcement for MCP tools and API endpoints

backend/api/routes/admin.py

@@ -10,6 +10,7 @@ from backend.api.storage.rules_store import RulesStore
 from backend.api.storage.analytics_store import AnalyticsStore
 from backend.api.services.rule_enhancer import RuleEnhancer
 from backend.api.services.document_ingestion import extract_text_from_file_bytes
+from ..utils.access_control import require_api_permission
 
 router = APIRouter()
 logger = logging.getLogger(__name__)
@@ -132,7 +133,8 @@ async def add_redflag_rule(
     payload: Optional[RulePayload] = None,
     rule: Optional[str] = None,
     x_tenant_id: str = Header(None),
-    enhance: bool = Query(True, description="Use LLM to enhance the rule before saving")
+    enhance: bool = Query(True, description="Use LLM to enhance the rule before saving"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Adds a new red-flag rule to this tenant.
@@ -148,6 +150,7 @@ async def add_redflag_rule(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "manage_rules")
 
     rule_value = payload.rule if payload else rule
     if not rule_value:
@@ -235,7 +238,8 @@ async def add_redflag_rule(
 async def add_redflag_rules_bulk(
     payload: BulkRulePayload,
     x_tenant_id: str = Header(None),
-    enhance: bool = Query(True, description="Use LLM to enhance rules before saving")
+    enhance: bool = Query(True, description="Use LLM to enhance rules before saving"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Adds multiple rules in one call.
@@ -247,6 +251,7 @@ async def add_redflag_rules_bulk(
     """
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "manage_rules")
 
     if not payload.rules:
         raise HTTPException(status_code=400, detail="No rules provided")
@@ -337,7 +342,8 @@ async def add_redflag_rules_bulk(
 @router.delete("/rules/{rule}")
 async def delete_redflag_rule(
     rule: str,
-    x_tenant_id: str = Header(None)
+    x_tenant_id: str = Header(None),
+    x_user_role: str = Header("viewer")
 ):
     """
     Deletes a red-flag rule for this tenant.
@@ -345,6 +351,7 @@ async def delete_redflag_rule(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "manage_rules")
 
     deleted = rules_store.delete_rule(x_tenant_id, rule)
     if not deleted:
@@ -425,7 +432,8 @@ async def get_tool_logs(
 async def upload_rules_from_file(
     file: UploadFile = File(...),
     x_tenant_id: str = Header(None),
-    enhance: bool = Query(True, description="Use LLM to enhance rules before saving")
+    enhance: bool = Query(True, description="Use LLM to enhance rules before saving"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Upload rules from a file (TXT, PDF, DOC, DOCX).
@@ -433,6 +441,7 @@ async def upload_rules_from_file(
     """
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "manage_rules")
     
     if not file.filename:
         raise HTTPException(status_code=400, detail="No file provided")

backend/api/routes/analytics.py

@@ -3,6 +3,7 @@ from typing import Optional
 from datetime import datetime, timedelta
 
 from ..storage.analytics_store import AnalyticsStore
+from ..utils.access_control import require_api_permission
 
 router = APIRouter()
 
@@ -13,7 +14,8 @@ analytics_store = AnalyticsStore()
 @router.get("/overview")
 async def analytics_overview(
     x_tenant_id: str = Header(None),
-    days: int = Query(30, description="Number of days to look back")
+    days: int = Query(30, description="Number of days to look back"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Returns an overview of analytics for the dashboard.
@@ -22,6 +24,7 @@ async def analytics_overview(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "view_analytics")
 
     since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
     
@@ -45,7 +48,8 @@ async def analytics_overview(
 @router.get("/tool-usage")
 async def analytics_tool_usage(
     x_tenant_id: str = Header(None),
-    days: int = Query(30, description="Number of days to look back")
+    days: int = Query(30, description="Number of days to look back"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Returns how often each tool (RAG, Web, Admin, LLM) was used with detailed stats.
@@ -54,6 +58,7 @@ async def analytics_tool_usage(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "view_analytics")
 
     since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
     tool_usage = analytics_store.get_tool_usage_stats(x_tenant_id, since_timestamp)
@@ -69,7 +74,8 @@ async def analytics_tool_usage(
 async def analytics_redflags(
     x_tenant_id: str = Header(None),
     limit: int = Query(50, description="Maximum number of violations to return"),
-    days: int = Query(30, description="Number of days to look back")
+    days: int = Query(30, description="Number of days to look back"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Returns red-flag violations for this tenant.
@@ -78,6 +84,7 @@ async def analytics_redflags(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "view_analytics")
 
     since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
     redflags = analytics_store.get_redflag_violations(x_tenant_id, limit, since_timestamp)
@@ -97,7 +104,8 @@ async def analytics_redflags(
 @router.get("/activity")
 async def analytics_activity(
     x_tenant_id: str = Header(None),
-    days: int = Query(30, description="Number of days to look back")
+    days: int = Query(30, description="Number of days to look back"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Returns general tenant activity statistics.
@@ -106,6 +114,7 @@ async def analytics_activity(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "view_analytics")
 
     since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
     activity = analytics_store.get_activity_summary(x_tenant_id, since_timestamp)
@@ -120,7 +129,8 @@ async def analytics_activity(
 @router.get("/rag-quality")
 async def analytics_rag_quality(
     x_tenant_id: str = Header(None),
-    days: int = Query(30, description="Number of days to look back")
+    days: int = Query(30, description="Number of days to look back"),
+    x_user_role: str = Header("viewer")
 ):
     """
     Returns RAG quality metrics including recall/precision indicators.
@@ -129,6 +139,7 @@ async def analytics_rag_quality(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "view_analytics")
 
     since_timestamp = int((datetime.now() - timedelta(days=days)).timestamp()) if days else None
     rag_quality = analytics_store.get_rag_quality_metrics(x_tenant_id, since_timestamp)

backend/api/routes/rag.py

@@ -9,6 +9,7 @@ from api.services.document_ingestion import (
     normalize_text,
     extract_text_from_file_bytes
 )
+from ..utils.access_control import require_api_permission
 
 router = APIRouter()
 rag_client = RAGClient()
@@ -58,7 +59,8 @@ async def rag_search(
 @router.post("/ingest")
 async def rag_ingest(
     req: IngestRequest,
-    x_tenant_id: str = Header(None)
+    x_tenant_id: str = Header(None),
+    x_user_role: str = Header("viewer")
 ):
     """
     Legacy ingestion endpoint - simple content ingestion.
@@ -67,6 +69,7 @@ async def rag_ingest(
 
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "ingest_documents")
 
     try:
         result = await rag_client.ingest(req.content, x_tenant_id)
@@ -82,7 +85,8 @@ async def rag_ingest(
 @router.post("/ingest-document")
 async def rag_ingest_document(
     req: DocumentIngestRequest,
-    x_tenant_id: Optional[str] = Header(None)
+    x_tenant_id: Optional[str] = Header(None),
+    x_user_role: str = Header("viewer")
 ):
     """
     Enhanced document ingestion endpoint matching the system prompt specification.
@@ -110,6 +114,7 @@ async def rag_ingest_document(
     tenant_id = req.tenant_id or x_tenant_id
     if not tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "ingest_documents")
 
     try:
         # Prepare ingestion payload (async for URL fetching)
@@ -141,7 +146,8 @@ async def rag_ingest_document(
 async def rag_ingest_file(
     file: UploadFile = File(...),
     x_tenant_id: Optional[str] = Header(None),
-    tenant_id: Optional[str] = Form(None)
+    tenant_id: Optional[str] = Form(None),
+    x_user_role: str = Header("viewer")
 ):
     """
     File upload endpoint for binary files (PDF, DOCX, TXT, MD).
@@ -159,6 +165,7 @@ async def rag_ingest_file(
     tenant_id_value = tenant_id or x_tenant_id
     if not tenant_id_value:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "ingest_documents")
 
     try:
         # Read file bytes
@@ -225,13 +232,15 @@ async def rag_list(
 @router.delete("/delete/{document_id}")
 async def rag_delete(
     document_id: int,
-    x_tenant_id: str = Header(None)
+    x_tenant_id: str = Header(None),
+    x_user_role: str = Header("viewer")
 ):
     """
     Delete a specific document by ID from tenant knowledge base.
     """
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "delete_documents")
 
     try:
         result = await rag_client.delete_document(x_tenant_id, document_id)
@@ -253,13 +262,15 @@ async def rag_delete(
 
 @router.delete("/delete-all")
 async def rag_delete_all(
-    x_tenant_id: str = Header(None)
+    x_tenant_id: str = Header(None),
+    x_user_role: str = Header("viewer")
 ):
     """
     Delete all documents for a tenant.
     """
     if not x_tenant_id:
         raise HTTPException(status_code=400, detail="Missing tenant ID")
+    require_api_permission(x_user_role, "delete_documents")
 
     try:
         result = await rag_client.delete_all_documents(x_tenant_id)

backend/api/utils/access_control.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from fastapi import HTTPException
+
+from backend.mcp_server.common import access_control as shared_access
+
+
+def require_api_permission(role_header: str | None, action: str) -> str:
+    """
+    Normalize the caller role from headers and ensure it can perform the action.
+    Raises HTTPException 403 if not permitted.
+    Returns the normalized role for downstream logging if needed.
+    """
+    role = shared_access.normalize_role(role_header)
+    if not shared_access.role_allows(role, action):
+        allowed_roles = shared_access.describe_allowed_roles(action)
+        raise HTTPException(
+            status_code=403,
+            detail=f"Role '{role}' lacks permission for '{action}'. Allowed roles: {allowed_roles}."
+        )
+    return role
+
+

backend/mcp_server/common/access_control.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from typing import Optional, Set
+
+# Role hierarchy used across MCP server and FastAPI routes
+VALID_ROLES = ("viewer", "editor", "admin", "owner")
+ROLE_ORDER = {role: idx for idx, role in enumerate(VALID_ROLES)}
+
+# Permission matrix defining which roles can perform which enterprise actions
+PERMISSIONS: dict[str, Set[str]] = {
+    "manage_rules": {"owner", "admin"},
+    "ingest_documents": {"owner", "admin", "editor"},
+    "delete_documents": {"owner", "admin"},
+    "view_analytics": {"owner", "admin"},
+}
+
+# Mapping of MCP tool names to enterprise actions
+TOOL_PERMISSION_MAP: dict[str, str] = {
+    "admin.addRule": "manage_rules",
+    "admin.deleteRule": "manage_rules",
+    "rag.ingest": "ingest_documents",
+    "rag.delete": "delete_documents",
+}
+
+
+def normalize_role(raw_value: Optional[str]) -> str:
+    """
+    Normalize an inbound role string. Defaults to 'viewer' when undefined or invalid.
+    """
+    if not raw_value:
+        return "viewer"
+    value = raw_value.strip().lower()
+    if value not in VALID_ROLES:
+        return "viewer"
+    return value
+
+
+def allowed_roles_for(action: str) -> Set[str]:
+    """
+    Return the set of roles that can execute the given action.
+    If the action is unknown, all roles are allowed.
+    """
+    return PERMISSIONS.get(action, set(VALID_ROLES))
+
+
+def role_allows(role: str, action: str) -> bool:
+    """
+    Check whether the supplied role has permission for the action.
+    Unknown actions default to allow-all to avoid accidental lockouts.
+    """
+    allowed = allowed_roles_for(action)
+    return role in allowed if allowed else True
+
+
+def describe_allowed_roles(action: str) -> str:
+    """
+    Return a human-friendly description of which roles are allowed for an action.
+    """
+    allowed = sorted(allowed_roles_for(action), key=lambda r: ROLE_ORDER.get(r, 0))
+    return ", ".join(allowed)
+
+
+def get_required_action_for_tool(tool_name: str) -> Optional[str]:
+    """
+    Look up the enterprise action that applies to a tool name, if any.
+    """
+    return TOOL_PERMISSION_MAP.get(tool_name)
+
+

backend/mcp_server/common/tenant.py

@@ -4,6 +4,8 @@ import re
 from dataclasses import dataclass
 from typing import Any, Mapping, Optional
 
+from .access_control import normalize_role
+
 
 class TenantValidationError(ValueError):
     """Raised when tenant metadata is missing or malformed."""
@@ -17,6 +19,7 @@ class TenantContext:
     tenant_id: str
     user_id: Optional[str] = None
     metadata: Optional[dict[str, Any]] = None
+    role: str = "viewer"
 
 
 def _extract_tenant_id(payload: Mapping[str, Any]) -> str:
@@ -43,6 +46,7 @@ def build_tenant_context(payload: Mapping[str, Any]) -> TenantContext:
     tenant_id = _normalize_tenant_id(_extract_tenant_id(payload))
     user_id: Optional[str] = None
     metadata: Optional[dict[str, Any]] = None
+    role: str = "viewer"
 
     for key in ("user_id", "userId"):
         if key in payload and isinstance(payload[key], str):
@@ -53,5 +57,18 @@ def build_tenant_context(payload: Mapping[str, Any]) -> TenantContext:
     if isinstance(meta_candidate, dict):
         metadata = meta_candidate
 
-    return TenantContext(tenant_id=tenant_id, user_id=user_id, metadata=metadata)
+    # Extract role from payload or metadata (if provided)
+    role_candidates = [
+        payload.get("role"),
+        payload.get("user_role"),
+        payload.get("userRole"),
+    ]
+    if metadata:
+        role_candidates.append(metadata.get("role"))
+    for candidate in role_candidates:
+        if isinstance(candidate, str):
+            role = normalize_role(candidate)
+            break
+
+    return TenantContext(tenant_id=tenant_id, user_id=user_id, metadata=metadata, role=role)
 

backend/mcp_server/common/utils.py

@@ -6,6 +6,7 @@ from typing import Any, Awaitable, Callable, Mapping, Optional
 from .logging import log_tool_usage
 from .tenant import TenantContext, TenantValidationError, build_tenant_context
 from . import memory
+from . import access_control
 
 
 class ToolValidationError(ValueError):
@@ -16,6 +17,10 @@ class ToolExecutionError(RuntimeError):
     """Raised for unexpected runtime failures."""
 
 
+class AuthorizationError(ToolValidationError):
+    """Raised when the caller request payload lacks required permissions."""
+
+
 Payload = Mapping[str, Any]
 ToolHandler = Callable[[TenantContext, Payload], Awaitable[dict[str, Any]] | dict[str, Any]]
 
@@ -110,6 +115,16 @@ async def execute_tool(
     try:
         # Tenant context still comes from the original payload
         context = build_tenant_context(payload)
+
+        # Enforce role-based permissions for sensitive tool actions
+        required_action = access_control.get_required_action_for_tool(tool_name)
+        if required_action and not access_control.role_allows(context.role, required_action):
+            allowed_roles = access_control.describe_allowed_roles(required_action)
+            raise AuthorizationError(
+                f"Role '{context.role}' is not permitted to perform '{required_action}'. "
+                f"Allowed roles: {allowed_roles}."
+            )
+
         result = await maybe_await(handler(context, mutable_payload))
         latency_ms = int((time.perf_counter() - start) * 1000)
 

backend/tests/test_access_control.py

@@ -0,0 +1,55 @@
+import sys
+from pathlib import Path
+import pytest
+
+# Ensure backend package is importable
+backend_dir = Path(__file__).parent.parent
+sys.path.insert(0, str(backend_dir))
+
+from mcp_server.common import access_control
+from mcp_server.common.utils import execute_tool
+
+
+@pytest.mark.asyncio
+async def test_execute_tool_denies_without_permission():
+    async def handler(context, payload):
+        return {"ok": True}
+
+    payload = {
+        "tenant_id": "tenant123",
+        "session_id": "s1",
+        "role": "viewer",
+    }
+
+    result = await execute_tool("rag.ingest", payload, handler)
+    assert result["status"] == "error"
+    assert result["error_type"] == "validation_error"
+    assert "not permitted" in result["message"]
+
+
+@pytest.mark.asyncio
+async def test_execute_tool_allows_authorized_role():
+    async def handler(context, payload):
+        return {"ok": True}
+
+    payload = {
+        "tenant_id": "tenant123",
+        "session_id": "s1",
+        "role": "admin",
+    }
+
+    result = await execute_tool("rag.ingest", payload, handler)
+    assert result["status"] == "ok"
+    assert result["data"]["ok"] is True
+
+
+def test_normalize_role_defaults_to_viewer():
+    assert access_control.normalize_role(None) == "viewer"
+    assert access_control.normalize_role("ADMIN") == "admin"
+    assert access_control.normalize_role("unknown") == "viewer"
+
+
+def test_role_allows_matrix():
+    assert access_control.role_allows("owner", "manage_rules")
+    assert not access_control.role_allows("viewer", "manage_rules")
+

backend/tests/test_api_endpoints.py

@@ -34,7 +34,7 @@ def test_analytics_overview_endpoint(client):
     """Test /analytics/overview endpoint."""
     response = client.get(
         "/analytics/overview",
-        headers={"x-tenant-id": "test_tenant"},
+        headers={"x-tenant-id": "test_tenant", "x-user-role": "owner"},
         params={"days": 30}
     )
     
@@ -51,7 +51,7 @@ def test_analytics_tool_usage_endpoint(client):
     """Test /analytics/tool-usage endpoint."""
     response = client.get(
         "/analytics/tool-usage",
-        headers={"x-tenant-id": "test_tenant"},
+        headers={"x-tenant-id": "test_tenant", "x-user-role": "owner"},
         params={"days": 30}
     )
     
@@ -66,7 +66,7 @@ def test_analytics_rag_quality_endpoint(client):
     """Test /analytics/rag-quality endpoint."""
     response = client.get(
         "/analytics/rag-quality",
-        headers={"x-tenant-id": "test_tenant"},
+        headers={"x-tenant-id": "test_tenant", "x-user-role": "owner"},
         params={"days": 30}
     )
     
@@ -80,7 +80,7 @@ def test_admin_rules_with_regex(client):
     """Test adding admin rule with regex pattern and severity."""
     response = client.post(
         "/admin/rules",
-        headers={"x-tenant-id": "test_tenant"},
+        headers={"x-tenant-id": "test_tenant", "x-user-role": "owner"},
         json={
             "rule": "Block password queries",
             "pattern": ".*password.*",
@@ -200,3 +200,23 @@ def test_admin_tenants_endpoints(client):
     response = client.delete("/admin/tenants/new_tenant")
     assert response.status_code == 200
 
+
+def test_analytics_requires_admin_role(client):
+    """Ensure analytics endpoints enforce RBAC."""
+    response = client.get(
+        "/analytics/overview",
+        headers={"x-tenant-id": "test_tenant", "x-user-role": "viewer"},
+        params={"days": 7}
+    )
+    assert response.status_code == 403
+
+
+def test_admin_rules_requires_admin_role(client):
+    """Ensure rule uploads enforce RBAC."""
+    response = client.post(
+        "/admin/rules",
+        headers={"x-tenant-id": "test_tenant", "x-user-role": "viewer"},
+        json={"rule": "No passwords"}
+    )
+    assert response.status_code == 403
+