diff --git a/backend/main.py b/backend/main.py index ae7da7c1f..cbfa4049c 100644 --- a/backend/main.py +++ b/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 diff --git a/backend/services/rbac.py b/backend/services/rbac.py new file mode 100644 index 000000000..b7d6deee9 --- /dev/null +++ b/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") diff --git a/backend/tests/test_rbac.py b/backend/tests/test_rbac.py new file mode 100644 index 000000000..ac794fa77 --- /dev/null +++ b/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()