refactor: consolidate login audit logging and add superadmin user audit endpoints

2026-05-08 06:26:32 +00:00
parent 6d794106be
commit 81a221bd2b
6 changed files with 303 additions and 12 deletions
@@ -514,3 +514,235 @@ def remove_user_from_org(user_id, org_id):
            status=500,
            error_type="INTERNAL_ERROR",
        )
+
+
+# ============ User Audit Log Endpoints ============
+
+@superadmin_bp.route("/users/<user_id>/audit-logs", methods=["GET"])
+@superadmin_required
+def get_user_audit_logs(user_id):
+    """Get audit logs for a specific user (superadmin only).
+
+    Query params:
+        page: Page number (default 1)
+        per_page: Items per page (default 50, max 200)
+        action: Filter by action type
+        success: Filter by success (true/false)
+        start_date: Filter by start date (ISO 8601)
+        end_date: Filter by end date (ISO 8601)
+    """
+    try:
+        from gatehouse_app.models.auth.audit_log import AuditLog
+        from gatehouse_app.models.user.user import User
+
+        user = User.query.get(user_id)
+        if not user or user.deleted_at is not None:
+            return api_response(
+                success=False,
+                message="User not found",
+                status=404,
+                error_type="NOT_FOUND",
+            )
+
+        page = max(1, int(request.args.get("page", 1)))
+        per_page = min(200, max(1, int(request.args.get("per_page", 50))))
+
+        query = AuditLog.query.filter_by(user_id=user_id)
+
+        # Filters
+        action_filter = request.args.get("action")
+        if action_filter:
+            query = query.filter(AuditLog.action == action_filter)
+
+        success_filter = request.args.get("success")
+        if success_filter is not None:
+            query = query.filter(AuditLog.success == (success_filter.lower() == "true"))
+
+        start_date = request.args.get("start_date")
+        if start_date:
+            from datetime import datetime, timezone
+            try:
+                start_dt = datetime.fromisoformat(start_date.replace("Z", "+00:00"))
+                query = query.filter(AuditLog.created_at >= start_dt)
+            except ValueError:
+                return api_response(
+                    success=False,
+                    message="Invalid start_date format. Use ISO 8601.",
+                    status=400,
+                    error_type="VALIDATION_ERROR",
+                )
+
+        end_date = request.args.get("end_date")
+        if end_date:
+            from datetime import datetime, timezone
+            try:
+                end_dt = datetime.fromisoformat(end_date.replace("Z", "+00:00"))
+                query = query.filter(AuditLog.created_at <= end_dt)
+            except ValueError:
+                return api_response(
+                    success=False,
+                    message="Invalid end_date format. Use ISO 8601.",
+                    status=400,
+                    error_type="VALIDATION_ERROR",
+                )
+
+        query = query.order_by(AuditLog.created_at.desc())
+        total = query.count()
+        logs = query.offset((page - 1) * per_page).limit(per_page).all()
+
+        def log_to_dict(log):
+            action = log.action
+            return {
+                "id": log.id,
+                "action": action.value if hasattr(action, "value") else action,
+                "user_id": log.user_id,
+                "user": (
+                    {"id": log.user.id, "email": log.user.email, "full_name": log.user.full_name}
+                    if log.user else None
+                ),
+                "organization_id": log.organization_id,
+                "resource_type": log.resource_type,
+                "resource_id": log.resource_id,
+                "ip_address": log.ip_address,
+                "user_agent": log.user_agent,
+                "request_id": log.request_id,
+                "description": log.description,
+                "success": log.success,
+                "error_message": log.error_message,
+                "metadata": log.extra_data,
+                "created_at": log.created_at.isoformat() if log.created_at else None,
+            }
+
+        return api_response(
+            data={
+                "user": {
+                    "id": user.id,
+                    "email": user.email,
+                    "full_name": user.full_name,
+                },
+                "audit_logs": [log_to_dict(log) for log in logs],
+                "count": total,
+                "page": page,
+                "per_page": per_page,
+                "pages": (total + per_page - 1) // per_page,
+            },
+            message="User audit logs retrieved successfully",
+        )
+
+    except Exception as e:
+        logger.error(f"[SuperadminUsers] Get user audit logs error: {e}")
+        return api_response(
+            success=False,
+            message="An error occurred",
+            status=500,
+            error_type="INTERNAL_ERROR",
+        )
+
+
+@superadmin_bp.route("/users/<user_id>/audit-logs/export", methods=["GET"])
+@superadmin_required
+def export_user_audit_logs(user_id):
+    """Export audit logs for a specific user as CSV (superadmin only).
+
+    Query params:
+        action: Filter by action type
+        success: Filter by success (true/false)
+        start_date: Filter by start date (ISO 8601)
+        end_date: Filter by end date (ISO 8601)
+    """
+    try:
+        from gatehouse_app.models.auth.audit_log import AuditLog
+        from gatehouse_app.models.user.user import User
+        import csv
+        from flask import make_response
+        from io import StringIO
+
+        user = User.query.get(user_id)
+        if not user or user.deleted_at is not None:
+            return api_response(
+                success=False,
+                message="User not found",
+                status=404,
+                error_type="NOT_FOUND",
+            )
+
+        query = AuditLog.query.filter_by(user_id=user_id)
+
+        # Apply same filters as get_user_audit_logs
+        action_filter = request.args.get("action")
+        if action_filter:
+            query = query.filter(AuditLog.action == action_filter)
+
+        success_filter = request.args.get("success")
+        if success_filter is not None:
+            query = query.filter(AuditLog.success == (success_filter.lower() == "true"))
+
+        start_date = request.args.get("start_date")
+        if start_date:
+            from datetime import datetime, timezone
+            try:
+                start_dt = datetime.fromisoformat(start_date.replace("Z", "+00:00"))
+                query = query.filter(AuditLog.created_at >= start_dt)
+            except ValueError:
+                return api_response(
+                    success=False,
+                    message="Invalid start_date format. Use ISO 8601.",
+                    status=400,
+                    error_type="VALIDATION_ERROR",
+                )
+
+        end_date = request.args.get("end_date")
+        if end_date:
+            from datetime import datetime, timezone
+            try:
+                end_dt = datetime.fromisoformat(end_date.replace("Z", "+00:00"))
+                query = query.filter(AuditLog.created_at <= end_dt)
+            except ValueError:
+                return api_response(
+                    success=False,
+                    message="Invalid end_date format. Use ISO 8601.",
+                    status=400,
+                    error_type="VALIDATION_ERROR",
+                )
+
+        query = query.order_by(AuditLog.created_at.desc())
+        logs = query.all()
+
+        # Generate CSV
+        output = StringIO()
+        writer = csv.writer(output)
+        writer.writerow([
+            "id", "timestamp", "action", "user_email", "organization_id",
+            "resource_type", "resource_id", "ip_address", "success",
+            "description", "error_message"
+        ])
+
+        for log in logs:
+            action = log.action
+            writer.writerow([
+                log.id,
+                log.created_at.isoformat() if log.created_at else "",
+                action.value if hasattr(action, "value") else action,
+                log.user.email if log.user else "",
+                log.organization_id or "",
+                log.resource_type or "",
+                log.resource_id or "",
+                log.ip_address or "",
+                log.success,
+                log.description or "",
+                log.error_message or "",
+            ])
+
+        response = make_response(output.getvalue())
+        response.headers["Content-Type"] = "text/csv"
+        response.headers["Content-Disposition"] = f"attachment; filename=user_{user_id}_audit_logs.csv"
+        return response
+
+    except Exception as e:
+        logger.error(f"[SuperadminUsers] Export user audit logs error: {e}")
+        return api_response(
+            success=False,
+            message="An error occurred",
+            status=500,
+            error_type="INTERNAL_ERROR",
+        )