bounty: role-based access controls authentication system using Supabase

backend/main.py

@@ -59,6 +59,10 @@
 from backend.services.ner_service import NERService
 from backend.services.duplicate_service import DuplicateService
 from backend.services.rag_service import RagService
+from backend.services.rbac import (
+    require_agent_or_admin,
+    require_any_authenticated,
+)
 
 
 # ---------------------------------------------------------------------------
@@ -549,7 +553,11 @@ async def get_tickets(company_id: str | None = None):
     return res.data
 
 @app.post("/tickets/save")
-async def save_ticket(request_body: TicketSaveRequest):
+async def save_ticket(
+    request_body: TicketSaveRequest,
+    request: Request,
+    _: str = Depends(require_any_authenticated),
+):
     """
     OFFICIAL PERSISTENCE: Saves the analyzed ticket to Supabase.
     This is called AFTER the user confirms the analysis results.
@@ -664,8 +672,12 @@ async def get_ticket_by_id(ticket_id: str):
 
 
 @app.post("/tickets", response_model=TicketRecord)
-async def create_ticket(ticket: TicketRecord):
-    """Save a new ticket into the system."""
+async def create_ticket(
+    ticket: TicketRecord,
+    request: Request,
+    _: str = Depends(require_any_authenticated),
+):
+    """Save a new ticket into the system (any authenticated role)."""
     # Check for duplicates before adding
     existing = next((t for t in TICKETS_DB if t.ticket_id == ticket.ticket_id), None)
     if existing:
@@ -677,8 +689,13 @@ async def create_ticket(ticket: TicketRecord):
 
 
 @app.patch("/tickets/{ticket_id}", response_model=TicketRecord)
-async def update_ticket(ticket_id: str, updates: dict):
-    """Partially update a ticket's fields (e.g., status, viewed_at)."""
+async def update_ticket(
+    ticket_id: str,
+    updates: dict,
+    request: Request,
+    _: str = Depends(require_agent_or_admin),
+):
+    """Partially update a ticket's fields (admin/agent only)."""
     for i, ticket in enumerate(TICKETS_DB):
         if str(ticket.ticket_id) == str(ticket_id):
             # Convert to dict, update, then back to model

backend/services/rbac.py

@@ -0,0 +1,101 @@
+"""
+Role-based access control (issue #3911).
+
+Enforces strict permissions for the three supported roles — ``admin``, ``agent``
+and ``employee`` — at the endpoint level through FastAPI dependencies. The
+caller's role is resolved from the ``X-User-Role`` header (set by the frontend
+from the authenticated Supabase profile), normalized and checked against a
+central permission matrix before the handler runs.
+
+Run with:  python -m unittest backend.tests.test_rbac -v
+"""
+
+from __future__ import annotations
+
+from fastapi import Depends, HTTPException, Request
+
+ROLE_ADMIN = "admin"
+ROLE_AGENT = "agent"
+ROLE_EMPLOYEE = "employee"
+ALL_ROLES = frozenset({ROLE_ADMIN, ROLE_AGENT, ROLE_EMPLOYEE})
+
+ROLE_HEADER = "x-user-role"
+
+# Central permission matrix: action -> roles allowed to perform it.
+PERMISSIONS: dict[str, frozenset[str]] = {
+    "ticket.read": frozenset({ROLE_ADMIN, ROLE_AGENT, ROLE_EMPLOYEE}),
+    "ticket.create": frozenset({ROLE_ADMIN, ROLE_AGENT, ROLE_EMPLOYEE}),
+    "ticket.update": frozenset({ROLE_ADMIN, ROLE_AGENT}),
+    "ticket.assign": frozenset({ROLE_ADMIN, ROLE_AGENT}),
+    "ticket.delete": frozenset({ROLE_ADMIN}),
+    "admin.users.manage": frozenset({ROLE_ADMIN}),
+    "reports.export": frozenset({ROLE_ADMIN}),
+    "audit.read": frozenset({ROLE_ADMIN}),
+}
+
+
+def normalize_role(value: str | None) -> str | None:
+    """Lower-case and validate a raw role value; returns None if invalid."""
+    if not value:
+        return None
+    role = value.strip().lower()
+    return role if role in ALL_ROLES else None
+
+
+def get_request_role(request: Request) -> str | None:
+    """Resolve the caller's role from the ``X-User-Role`` header."""
+    return normalize_role(request.headers.get(ROLE_HEADER))
+
+
+def has_permission(role: str | None, action: str) -> bool:
+    """True when ``role`` is allowed to perform ``action``."""
+    return role in PERMISSIONS.get(action, frozenset())
+
+
+def require_roles(*required_roles: str):
+    """
+    Build a FastAPI dependency that requires any one of ``required_roles``.
+
+    Raises 401 when the caller is unauthenticated and 403 when the role is
+    present but not permitted.
+    """
+
+    def dependency(request: Request) -> str:
+        role = get_request_role(request)
+        if role is None:
+            raise HTTPException(status_code=401, detail="Authentication required")
+        if role not in required_roles:
+            raise HTTPException(
+                status_code=403,
+                detail=f"Requires one of roles: {', '.join(sorted(required_roles))}",
+            )
+        return role
+
+    return dependency
+
+
+def require_action(action: str):
+    """
+    Build a FastAPI dependency checking the permission matrix for ``action``.
+    """
+
+    def dependency(request: Request) -> str:
+        role = get_request_role(request)
+        if role is None:
+            raise HTTPException(status_code=401, detail="Authentication required")
+        if not has_permission(role, action):
+            raise HTTPException(
+                status_code=403,
+                detail=f"Role '{role}' is not allowed to {action}",
+            )
+        return role
+
+    return dependency
+
+
+# Common pre-built dependencies.
+require_admin = require_roles(ROLE_ADMIN)
+require_agent_or_admin = require_roles(ROLE_ADMIN, ROLE_AGENT)
+require_any_authenticated = require_roles(*ALL_ROLES)
+
+require_admin_action = require_action("admin.users.manage")

backend/tests/test_rbac.py

@@ -0,0 +1,100 @@
+"""
+Unit tests for role-based access control (issue #3911).
+
+Run with:  python -m unittest backend.tests.test_rbac -v
+"""
+
+import unittest
+
+from fastapi import HTTPException
+from starlette.requests import Request
+
+from backend.services.rbac import (
+    ROLE_ADMIN,
+    ROLE_AGENT,
+    ROLE_EMPLOYEE,
+    ROLE_HEADER,
+    get_request_role,
+    has_permission,
+    normalize_role,
+    require_roles,
+)
+
+
+def _make_request(headers: dict | None = None) -> Request:
+    scope = {
+        "type": "http",
+        "method": "GET",
+        "path": "/",
+        "headers": [
+            (k.lower().encode(), v.encode())
+            for k, v in (headers or {}).items()
+        ],
+        "query_string": b"",
+        "server": ("test", 80),
+        "client": ("127.0.0.1", 5000),
+        "scheme": "http",
+    }
+    return Request(scope)
+
+
+class NormalizeRoleTests(unittest.TestCase):
+    def test_normalizes_case(self):
+        self.assertEqual(normalize_role("ADMIN"), "admin")
+
+    def test_invalid_role_returns_none(self):
+        self.assertIsNone(normalize_role("superuser"))
+
+    def test_empty_returns_none(self):
+        self.assertIsNone(normalize_role(""))
+
+
+class GetRequestRoleTests(unittest.TestCase):
+    def test_reads_role_header(self):
+        request = _make_request({ROLE_HEADER: "agent"})
+        self.assertEqual(get_request_role(request), "agent")
+
+    def test_missing_header(self):
+        request = _make_request()
+        self.assertIsNone(get_request_role(request))
+
+    def test_unknown_role_ignored(self):
+        request = _make_request({ROLE_HEADER: "root"})
+        self.assertIsNone(get_request_role(request))
+
+
+class HasPermissionTests(unittest.TestCase):
+    def test_matrix(self):
+        self.assertTrue(has_permission("admin", "ticket.delete"))
+        self.assertFalse(has_permission("agent", "ticket.delete"))
+        self.assertTrue(has_permission("agent", "ticket.update"))
+        self.assertFalse(has_permission("employee", "ticket.update"))
+        self.assertTrue(has_permission("employee", "ticket.create"))
+
+    def test_unknown_action(self):
+        self.assertFalse(has_permission("admin", "nonexistent.action"))
+
+
+class RequireRolesTests(unittest.TestCase):
+    def test_allowed_role_passes(self):
+        request = _make_request({ROLE_HEADER: "admin"})
+        dependency = require_roles(ROLE_ADMIN, ROLE_AGENT)
+        self.assertEqual(dependency(request), "admin")
+
+    def test_disallowed_role_raises_403(self):
+        request = _make_request({ROLE_HEADER: ROLE_EMPLOYEE})
+        dependency = require_roles(ROLE_ADMIN, ROLE_AGENT)
+        with self.assertRaises(HTTPException) as ctx:
+            dependency(request)
+        self.assertEqual(ctx.exception.status_code, 403)
+
+    def test_no_role_raises_401(self):
+        request = _make_request()
+        dependency = require_roles(ROLE_ADMIN)
+        with self.assertRaises(HTTPException) as ctx:
+            dependency(request)
+        self.assertEqual(ctx.exception.status_code, 401)
+
+
+if __name__ == "__main__":
+    unittest.main()