feat: enforce api permissions

backend/apps/accounts/tests/test_permission_enforcement.py

@@ -0,0 +1,67 @@
+from django.contrib.auth import get_user_model
+from django.contrib.auth.models import Group
+from django.core.management import call_command
+
+from apps.audit.models import OperationAuditLog
+from apps.tickets.models import Ticket
+
+
+def user_with_role(role_name: str):
+    call_command("bootstrap_governance")
+    user = get_user_model().objects.create_user(username=f"user-{role_name}", password="pass")
+    user.groups.add(Group.objects.get(name=role_name))
+    return user
+
+
+def test_anonymous_ticket_list_is_denied(client, db):
+    response = client.get("/api/tickets")
+
+    assert response.status_code == 401
+
+
+def test_user_with_ticket_view_can_list_tickets(client, db):
+    Ticket.objects.create(ticket_no="TK202607010001", title="A", description="A")
+    client.force_login(user_with_role("普通建单人"))
+
+    response = client.get("/api/tickets")
+
+    assert response.status_code == 200
+    assert response.json()[0]["ticket_no"] == "TK202607010001"
+
+
+def test_user_without_ticket_create_cannot_create_ticket(client, db):
+    client.force_login(user_with_role("只读审计员"))
+
+    response = client.post(
+        "/api/tickets",
+        data={"title": "A", "description": "A"},
+        content_type="application/json",
+    )
+
+    assert response.status_code == 403
+
+
+def test_user_with_solution_view_can_list_solutions(client, db):
+    client.force_login(user_with_role("普通建单人"))
+
+    response = client.get("/api/solutions")
+
+    assert response.status_code == 200
+
+
+def test_audit_log_requires_audit_permission(client, db):
+    OperationAuditLog.objects.create(
+        module="system",
+        action="health_check",
+        target_type="system",
+        target_id="health",
+        result="success",
+    )
+    client.force_login(user_with_role("普通建单人"))
+
+    denied = client.get("/api/audit/operation-logs")
+    assert denied.status_code == 403
+
+    client.force_login(user_with_role("只读审计员"))
+    allowed = client.get("/api/audit/operation-logs")
+    assert allowed.status_code == 200

backend/apps/audit/api.py

@@ -3,6 +3,7 @@ from typing import Any
 
 from ninja import Router, Schema
 
+from apps.accounts.api import require_permission
 from apps.audit.models import OperationAuditLog
 
 router = Router(tags=["audit"])
@@ -23,4 +24,5 @@ class OperationAuditLogOut(Schema):
 
 @router.get("/operation-logs", response=list[OperationAuditLogOut])
 def list_operation_logs(request):
+    require_permission(request, "system.audit.view")
     return OperationAuditLog.objects.all()[:100]

backend/apps/audit/tests/test_audit_api.py

@@ -1,7 +1,7 @@
 from apps.audit.models import OperationAuditLog
 
 
-def test_list_operation_logs_returns_recent_logs(client, db):
+def test_list_operation_logs_returns_recent_logs(admin_client, db):
     OperationAuditLog.objects.create(
         module="system",
         action="health_check",
@@ -11,7 +11,7 @@ def test_list_operation_logs_returns_recent_logs(client, db):
         request_snapshot={"path": "/api/system/health"},
     )
 
-    response = client.get("/api/audit/operation-logs")
+    response = admin_client.get("/api/audit/operation-logs")
 
     assert response.status_code == 200
     payload = response.json()

backend/apps/solutions/api.py

@@ -1,5 +1,6 @@
 from ninja import Router
 
+from apps.accounts.api import require_permission
 from apps.solutions.models import Solution, TicketSolutionLink
 from apps.solutions.schemas import (
     SolutionIn,
@@ -14,6 +15,7 @@ router = Router(tags=["solutions"])
 
 @router.post("", response=SolutionOut)
 def create_solution(request, payload: SolutionIn):
+    require_permission(request, "solution.manage")
     return Solution.objects.create(
         title=payload.title,
         problem_signature=payload.problem_signature,
@@ -29,6 +31,7 @@ def create_solution(request, payload: SolutionIn):
 
 @router.get("", response=list[SolutionOut])
 def list_solutions(request, status: str | None = None):
+    require_permission(request, "solution.view")
     queryset = Solution.objects.all()
     if status:
         queryset = queryset.filter(status=status)
@@ -37,11 +40,13 @@ def list_solutions(request, status: str | None = None):
 
 @router.get("/{solution_id}", response=SolutionOut)
 def get_solution(request, solution_id: int):
+    require_permission(request, "solution.view")
     return Solution.objects.get(id=solution_id)
 
 
 @router.post("/{solution_id}/link-ticket", response=TicketSolutionLinkOut)
 def link_solution_to_ticket(request, solution_id: int, payload: TicketSolutionLinkIn):
+    require_permission(request, "solution.manage")
     solution = Solution.objects.get(id=solution_id)
     ticket = Ticket.objects.get(id=payload.ticket_id)
     link, _ = TicketSolutionLink.objects.get_or_create(

backend/apps/solutions/tests/test_solution_api.py

@@ -1,8 +1,8 @@
 from apps.tickets.models import Ticket
 
 
-def test_create_and_list_solution(client, db):
-    response = client.post(
+def test_create_and_list_solution(admin_client, db):
+    response = admin_client.post(
         "/api/solutions",
         data={
             "title": "Restart callback worker",
@@ -17,20 +17,20 @@ def test_create_and_list_solution(client, db):
     assert response.status_code == 200
     solution_id = response.json()["id"]
 
-    response = client.get("/api/solutions")
+    response = admin_client.get("/api/solutions")
     assert response.status_code == 200
     assert response.json()[0]["id"] == solution_id
 
 
-def test_link_solution_to_ticket_increments_usage(client, db):
+def test_link_solution_to_ticket_increments_usage(admin_client, db):
     ticket = Ticket.objects.create(ticket_no="TK202606300020", title="A", description="A")
-    solution = client.post(
+    solution = admin_client.post(
         "/api/solutions",
         data={"title": "S", "problem_signature": "P", "root_cause": "R", "solution": "Do S"},
         content_type="application/json",
     ).json()
 
-    response = client.post(
+    response = admin_client.post(
         f"/api/solutions/{solution['id']}/link-ticket",
         data={"ticket_id": ticket.id, "link_type": "adopted"},
         content_type="application/json",

backend/apps/tickets/api.py

@@ -6,6 +6,7 @@ from django.http import HttpResponse
 from django.utils import timezone
 from ninja import Router
 
+from apps.accounts.api import require_permission
 from apps.audit.models import OperationAuditLog
 from apps.tickets.models import Ticket, TicketAttachment, TicketComment, TicketEvent
 from apps.tickets.schemas import (
@@ -31,6 +32,7 @@ def next_ticket_no() -> str:
 
 @router.post("", response=TicketOut)
 def create_ticket(request, payload: TicketIn):
+    require_permission(request, "ticket.create")
     assignee = None
     if payload.assignee_id:
         assignee = get_user_model().objects.filter(id=payload.assignee_id).first()
@@ -55,6 +57,7 @@ def create_ticket(request, payload: TicketIn):
 
 @router.get("", response=list[TicketOut])
 def list_tickets(request, status: str | None = None, q: str | None = None):
+    require_permission(request, "ticket.view")
     queryset = Ticket.objects.all()
     if status:
         queryset = queryset.filter(status=status)
@@ -65,6 +68,7 @@ def list_tickets(request, status: str | None = None, q: str | None = None):
 
 @router.get("/export")
 def export_tickets(request):
+    require_permission(request, "ticket.export")
     response = HttpResponse(content_type="text/csv")
     response["Content-Disposition"] = 'attachment; filename="tickets.csv"'
     writer = csv.writer(response)
@@ -92,11 +96,13 @@ def export_tickets(request):
 
 @router.get("/{ticket_id}", response=TicketOut)
 def get_ticket(request, ticket_id: int):
+    require_permission(request, "ticket.view")
     return Ticket.objects.get(id=ticket_id)
 
 
 @router.patch("/{ticket_id}", response=TicketOut)
 def update_ticket(request, ticket_id: int, payload: TicketUpdateIn):
+    require_permission(request, "ticket.edit")
     ticket = Ticket.objects.get(id=ticket_id)
     before = {
         "title": ticket.title,
@@ -124,6 +130,7 @@ def update_ticket(request, ticket_id: int, payload: TicketUpdateIn):
 
 @router.post("/{ticket_id}/status", response=TicketOut)
 def update_ticket_status(request, ticket_id: int, payload: TicketStatusIn):
+    require_permission(request, "ticket.edit")
     ticket = Ticket.objects.get(id=ticket_id)
     return transition_ticket(
         ticket,
@@ -135,6 +142,7 @@ def update_ticket_status(request, ticket_id: int, payload: TicketStatusIn):
 
 @router.post("/{ticket_id}/comments", response=TicketCommentOut)
 def add_ticket_comment(request, ticket_id: int, payload: TicketCommentIn):
+    require_permission(request, "ticket.edit")
     ticket = Ticket.objects.get(id=ticket_id)
     comment = TicketComment.objects.create(
         ticket=ticket,
@@ -154,6 +162,7 @@ def add_ticket_comment(request, ticket_id: int, payload: TicketCommentIn):
 
 @router.post("/{ticket_id}/attachments", response=TicketAttachmentOut)
 def add_ticket_attachment(request, ticket_id: int, payload: TicketAttachmentIn):
+    require_permission(request, "ticket.edit")
     ticket = Ticket.objects.get(id=ticket_id)
     attachment = TicketAttachment.objects.create(ticket=ticket, **payload.dict())
     TicketEvent.objects.create(

backend/apps/tickets/tests/test_ticket_api.py

@@ -1,8 +1,8 @@
 from apps.tickets.models import Ticket, TicketEvent
 
 
-def test_create_ticket_returns_ticket_detail(client, db):
-    response = client.post(
+def test_create_ticket_returns_ticket_detail(admin_client, db):
+    response = admin_client.post(
         "/api/tickets",
         data={
             "title": "Payment callback fails",
@@ -23,11 +23,11 @@ def test_create_ticket_returns_ticket_detail(client, db):
     assert TicketEvent.objects.filter(ticket_id=payload["id"], event_type="created").exists()
 
 
-def test_list_tickets_filters_by_status(client, db):
+def test_list_tickets_filters_by_status(admin_client, db):
     Ticket.objects.create(ticket_no="TK202606300001", title="A", description="A", status="pending")
     Ticket.objects.create(ticket_no="TK202606300002", title="B", description="B", status="resolved")
 
-    response = client.get("/api/tickets?status=pending")
+    response = admin_client.get("/api/tickets?status=pending")
 
     assert response.status_code == 200
     payload = response.json()
@@ -35,7 +35,7 @@ def test_list_tickets_filters_by_status(client, db):
     assert payload[0]["ticket_no"] == "TK202606300001"
 
 
-def test_transition_ticket_status_creates_event(client, db):
+def test_transition_ticket_status_creates_event(admin_client, db):
     ticket = Ticket.objects.create(
         ticket_no="TK202606300003",
         title="A",
@@ -43,7 +43,7 @@ def test_transition_ticket_status_creates_event(client, db):
         status="pending",
     )
 
-    response = client.post(
+    response = admin_client.post(
         f"/api/tickets/{ticket.id}/status",
         data={"status": "in_progress", "remark": "Start handling"},
         content_type="application/json",

backend/apps/tickets/tests/test_ticket_export.py

@@ -2,10 +2,10 @@ from apps.audit.models import OperationAuditLog
 from apps.tickets.models import Ticket
 
 
-def test_add_comment_to_ticket(client, db):
+def test_add_comment_to_ticket(admin_client, db):
     ticket = Ticket.objects.create(ticket_no="TK202606300010", title="A", description="A")
 
-    response = client.post(
+    response = admin_client.post(
         f"/api/tickets/{ticket.id}/comments",
         data={
             "content": "Checked logs",
@@ -20,10 +20,10 @@ def test_add_comment_to_ticket(client, db):
     assert ticket.comments.count() == 1
 
 
-def test_add_attachment_to_ticket(client, db):
+def test_add_attachment_to_ticket(admin_client, db):
     ticket = Ticket.objects.create(ticket_no="TK202606300011", title="A", description="A")
 
-    response = client.post(
+    response = admin_client.post(
         f"/api/tickets/{ticket.id}/attachments",
         data={
             "file_name": "error.log",
@@ -38,10 +38,10 @@ def test_add_attachment_to_ticket(client, db):
     assert response.json()["file_name"] == "error.log"
 
 
-def test_export_tickets_returns_csv_and_audit_log(client, db):
+def test_export_tickets_returns_csv_and_audit_log(admin_client, db):
     Ticket.objects.create(ticket_no="TK202606300012", title="Export me", description="A")
 
-    response = client.get("/api/tickets/export")
+    response = admin_client.get("/api/tickets/export")
 
     assert response.status_code == 200
     assert response["content-type"].startswith("text/csv")

backend/tests/conftest.py

@@ -1 +1,13 @@
+import pytest
+from django.contrib.auth import get_user_model
 
+
+@pytest.fixture
+def admin_client(client, db):
+    user = get_user_model().objects.create_superuser(
+        username="admin-test",
+        email="admin-test@example.com",
+        password="admin-test-password",
+    )
+    client.force_login(user)
+    return client