From 93500b8a1ed1fb19cfb0e0a2507374b6072f50e3 Mon Sep 17 00:00:00 2001 From: opencode Date: Sat, 1 Aug 2026 19:51:20 +0530 Subject: [PATCH] feat(auth): audit log triggers on privilege elevation --- backend/main.py | 65 +++++++++++++++++ backend/services/audit_log.py | 88 +++++++++++++++++++++++ backend/tests/test_audit_log.py | 119 ++++++++++++++++++++++++++++++++ 3 files changed, 272 insertions(+) create mode 100644 backend/services/audit_log.py create mode 100644 backend/tests/test_audit_log.py diff --git a/backend/main.py b/backend/main.py index ae7da7c1f..03c99741e 100644 --- a/backend/main.py +++ b/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: diff --git a/backend/services/audit_log.py b/backend/services/audit_log.py new file mode 100644 index 000000000..7f81f01cc --- /dev/null +++ b/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) diff --git a/backend/tests/test_audit_log.py b/backend/tests/test_audit_log.py new file mode 100644 index 000000000..741f675c0 --- /dev/null +++ b/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()