feat(auth): set up audit log triggers on privilege access elevations

backend/main.py

@@ -59,6 +59,7 @@
 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_admin
 
 
 # ---------------------------------------------------------------------------
@@ -1150,6 +1151,70 @@ class SignupBody(BaseModel):
     role: str | None = "user"
     company: str | None = None
 
+
+class PrivilegeChangeBody(BaseModel):
+    target_user_id: str
+    role: str
+    action: str = "privilege.elevation"
+
+
+@app.post("/admin/users/{user_id}/privilege")
+async def admin_privilege_change(
+    user_id: str,
+    body: PrivilegeChangeBody,
+    request: Request,
+    _actor_role: str = Depends(require_admin),
+):
+    """
+    Admin-only endpoint that updates a target user's role.
+
+    Every call triggers a security audit record (issue #3906) capturing the
+    acting admin, the target user, request metadata, and a timestamp.
+    """
+    from backend.services.audit_log import log_privilege_change
+    from backend.services.rbac import get_request_role
+
+    if not supabase:
+        raise HTTPException(status_code=503, detail="Database connection offline")
+
+    new_role = body.role.strip().lower()
+    if new_role not in ("admin", "agent", "employee"):
+        raise HTTPException(status_code=400, detail="Invalid target role")
+
+    res = (
+        supabase.table("profiles")
+        .select("id, role, company_id")
+        .eq("id", user_id)
+        .single()
+        .execute()
+    )
+    if not res.data:
+        raise HTTPException(status_code=404, detail="Target user not found")
+    previous_role = res.data.get("role")
+
+    updated = (
+        supabase.table("profiles")
+        .update({"role": new_role})
+        .eq("id", user_id)
+        .execute()
+    )
+    if not updated.data:
+        raise HTTPException(status_code=500, detail="Role update failed")
+
+    log_privilege_change(
+        supabase,
+        actor_id=request.headers.get("x-user-id") or "unknown",
+        actor_role=get_request_role(request) or "admin",
+        target_user_id=user_id,
+        target_role=new_role,
+        action=body.action,
+        ip_address=request.client.host if request.client else None,
+        user_agent=request.headers.get("user-agent"),
+        meta={"previous_role": previous_role, "company_id": res.data.get("company_id")},
+    )
+
+    return {"ok": True, "target_user_id": user_id, "role": new_role, "previous_role": previous_role}
+
 @app.post("/auth/login")
 async def auth_login(body: LoginBody, response: Response):
     if not supabase:

backend/services/audit_log.py

@@ -0,0 +1,88 @@
+"""
+Security audit logging (issue #3906).
+
+Privilege-sensitive operations (admin role updates, user login elevation)
+persist a detailed audit record capturing the acting operator, the target
+user, and request metadata. Writes are deliberately non-blocking: an audit
+insert failure is logged but never fails the originating request.
+
+Run with:  python -m unittest backend.tests.test_audit_log -v
+"""
+
+import datetime
+import json
+import logging
+
+logger = logging.getLogger(__name__)
+
+AUDIT_TABLE = "audit_logs"
+
+PRIVILEGE_ACTIONS = ("privilege.elevation", "privilege.revocation", "role.update")
+
+
+def build_audit_payload(
+    *,
+    actor_id: str,
+    actor_role: str,
+    action: str,
+    target_user_id: str | None = None,
+    target_role: str | None = None,
+    ip_address: str | None = None,
+    user_agent: str | None = None,
+    meta: dict | None = None,
+) -> dict:
+    """
+    Assemble the normalized audit record. ``meta`` is serialized to JSON so
+    arbitrary context (previous role, changed fields) is preserved.
+    """
+    return {
+        "actor_id": actor_id,
+        "actor_role": actor_role,
+        "action": action,
+        "target_user_id": target_user_id,
+        "target_role": target_role,
+        "ip_address": ip_address,
+        "user_agent": user_agent,
+        "meta": json.dumps(meta) if meta else None,
+        "created_at": datetime.datetime.utcnow().isoformat() + "Z",
+    }
+
+
+def log_privilege_change(
+    supabase,
+    *,
+    actor_id: str,
+    actor_role: str,
+    target_user_id: str,
+    target_role: str,
+    action: str = "privilege.elevation",
+    ip_address: str | None = None,
+    user_agent: str | None = None,
+    meta: dict | None = None,
+) -> None:
+    """
+    Record a privilege change on ``target_user_id`` performed by ``actor_id``.
+
+    Never raises: if the audit insert fails, the error is logged and the
+    request continues (security logging must not be a single point of failure).
+    """
+    if action not in PRIVILEGE_ACTIONS:
+        action = "privilege.elevation"
+    if supabase is None:
+        logger.warning("[AuditLog] Supabase unavailable; audit event dropped")
+        return
+
+    payload = build_audit_payload(
+        actor_id=actor_id,
+        actor_role=actor_role,
+        action=action,
+        target_user_id=target_user_id,
+        target_role=target_role,
+        ip_address=ip_address,
+        user_agent=user_agent,
+        meta=meta,
+    )
+    try:
+        supabase.table(AUDIT_TABLE).insert(payload).execute()
+    except Exception as exc:  # noqa: BLE001 - audit must never break the request
+        logger.warning("[AuditLog] Failed to persist audit event: %s", exc)

backend/tests/test_audit_log.py

@@ -0,0 +1,119 @@
+"""
+Unit tests for security audit logging (issue #3906).
+
+Run with:  python -m unittest backend.tests.test_audit_log -v
+"""
+
+import unittest
+
+from backend.services.audit_log import (
+    AUDIT_TABLE,
+    build_audit_payload,
+    log_privilege_change,
+)
+
+
+class FakeTable:
+    def __init__(self, name):
+        self.name = name
+        self.pending = None
+
+    def insert(self, payload):
+        self.pending = payload
+        return self
+
+    def execute(self):
+        return None
+
+
+class FakeSupabase:
+    def __init__(self):
+        self.tables = {}
+
+    def table(self, name):
+        if name not in self.tables:
+            self.tables[name] = FakeTable(name)
+        return self.tables[name]
+
+
+class BuildAuditPayloadTests(unittest.TestCase):
+    def test_payload_contains_required_fields(self):
+        payload = build_audit_payload(
+            actor_id="a-1",
+            actor_role="admin",
+            action="privilege.elevation",
+            target_user_id="u-9",
+            target_role="agent",
+            ip_address="10.0.0.1",
+            user_agent="Mozilla/5.0",
+        )
+        self.assertEqual(payload["actor_id"], "a-1")
+        self.assertEqual(payload["actor_role"], "admin")
+        self.assertEqual(payload["action"], "privilege.elevation")
+        self.assertEqual(payload["target_user_id"], "u-9")
+        self.assertEqual(payload["target_role"], "agent")
+        self.assertEqual(payload["ip_address"], "10.0.0.1")
+        self.assertEqual(payload["user_agent"], "Mozilla/5.0")
+        self.assertTrue(payload["created_at"].endswith("Z"))
+
+    def test_meta_serialized_to_json(self):
+        payload = build_audit_payload(
+            actor_id="a",
+            actor_role="admin",
+            action="role.update",
+            meta={"previous_role": "agent", "changed_fields": ["role"]},
+        )
+        self.assertIn("previous_role", payload["meta"])
+        self.assertIn("agent", payload["meta"])
+
+
+class LogPrivilegeChangeTests(unittest.TestCase):
+    def test_writes_audit_record(self):
+        fake = FakeSupabase()
+        log_privilege_change(
+            fake,
+            actor_id="a-1",
+            actor_role="admin",
+            target_user_id="u-9",
+            target_role="admin",
+        )
+        table = fake.tables.get(AUDIT_TABLE)
+        self.assertIsNotNone(table)
+        self.assertEqual(table.pending["target_user_id"], "u-9")
+        self.assertEqual(table.pending["target_role"], "admin")
+
+    def test_unknown_action_normalized(self):
+        fake = FakeSupabase()
+        log_privilege_change(
+            fake,
+            actor_id="a",
+            actor_role="admin",
+            target_user_id="u",
+            target_role="agent",
+            action="hack.the.thing",
+        )
+        self.assertEqual(fake.tables[AUDIT_TABLE].pending["action"], "privilege.elevation")
+
+    def test_no_supabase_is_safe(self):
+        log_privilege_change(None, actor_id="a", actor_role="admin", target_user_id="u", target_role="agent")
+
+
+class BrokenSupabase:
+    def table(self, name):
+        raise RuntimeError("db down")
+
+
+class AuditFailureTests(unittest.TestCase):
+    def test_db_failure_does_not_raise(self):
+        # Audit logging must never break the originating request.
+        log_privilege_change(
+            BrokenSupabase(),
+            actor_id="a",
+            actor_role="admin",
+            target_user_id="u",
+            target_role="agent",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()