mirror of https://github.com/Aidaho12/haproxy-wi
241 lines
8.1 KiB
Python
241 lines
8.1 KiB
Python
"""
|
|
Error handling module for Roxy-WI.
|
|
|
|
This module provides a unified way to handle errors across the application.
|
|
It includes functions for handling exceptions, logging errors, and returning
|
|
appropriate HTTP responses.
|
|
"""
|
|
from typing import Any, Dict, Tuple
|
|
|
|
from flask import jsonify, request, render_template, g, redirect, url_for
|
|
from werkzeug.exceptions import HTTPException
|
|
from flask_pydantic.exceptions import ValidationError
|
|
|
|
from app.modules.roxywi.class_models import ErrorResponse
|
|
from app.modules.roxywi.exception import (
|
|
RoxywiResourceNotFound,
|
|
RoxywiGroupMismatch,
|
|
RoxywiGroupNotFound,
|
|
RoxywiPermissionError,
|
|
RoxywiConflictError,
|
|
RoxywiValidationError,
|
|
RoxywiCheckLimits
|
|
)
|
|
import app.modules.roxywi.common as roxywi_common
|
|
from app.modules.roxywi import logger
|
|
from app.middleware import get_user_params
|
|
|
|
# Map exception types to HTTP status codes
|
|
ERROR_CODE_MAPPING = {
|
|
RoxywiResourceNotFound: 404,
|
|
RoxywiGroupNotFound: 404,
|
|
RoxywiGroupMismatch: 404,
|
|
RoxywiPermissionError: 403,
|
|
RoxywiConflictError: 409,
|
|
RoxywiValidationError: 400,
|
|
RoxywiCheckLimits: 402,
|
|
KeyError: 400,
|
|
ValueError: 400,
|
|
Exception: 500
|
|
}
|
|
|
|
# Map exception types to error messages
|
|
ERROR_MESSAGE_MAPPING = {
|
|
RoxywiResourceNotFound: "Resource not found",
|
|
RoxywiGroupNotFound: "Group not found",
|
|
RoxywiGroupMismatch: "Resource not found in group",
|
|
RoxywiPermissionError: "You do not have permission to access this resource",
|
|
RoxywiConflictError: "Conflict with existing resource",
|
|
RoxywiValidationError: "Validation error",
|
|
RoxywiCheckLimits: "You have reached your plan limits",
|
|
KeyError: "Missing required field",
|
|
ValueError: "Invalid value provided"
|
|
}
|
|
|
|
|
|
def log_error(exception: Exception, server_ip: str = "Roxy-WI server",
|
|
additional_info: str = "", keep_history: bool = False,
|
|
service: str = None) -> None:
|
|
"""
|
|
Log an error with detailed information.
|
|
|
|
Args:
|
|
exception: The exception that was raised
|
|
server_ip: The IP of the server where the error occurred
|
|
additional_info: Additional information to include in the log
|
|
keep_history: Whether to keep the error in the action history
|
|
service: The service where the error occurred
|
|
"""
|
|
error_message = str(exception)
|
|
if additional_info:
|
|
error_message = f"{additional_info}: {error_message}"
|
|
|
|
# Log the error with structured context
|
|
logger.exception(
|
|
error_message,
|
|
exc=exception,
|
|
server_ip=server_ip,
|
|
service=service
|
|
)
|
|
|
|
# Keep history if requested
|
|
if keep_history and service:
|
|
try:
|
|
# Get user information if available
|
|
user_id = None
|
|
user_ip = request.remote_addr if request else 'unknown'
|
|
|
|
if hasattr(g, 'user'):
|
|
user_id = getattr(g.user, 'user_id', None)
|
|
|
|
if user_id:
|
|
roxywi_common.keep_action_history(service, error_message, server_ip, user_id, user_ip)
|
|
except Exception as e:
|
|
logger.error(f"Failed to keep error history: {e}", server_ip=server_ip)
|
|
|
|
|
|
def handle_exception(exception: Exception, server_ip: str = "Roxy-WI server",
|
|
additional_info: str = "", keep_history: bool = False,
|
|
service: str = None) -> Tuple[Dict[str, Any], int]:
|
|
"""
|
|
Handle an exception and return an appropriate HTTP response.
|
|
|
|
Args:
|
|
exception: The exception that was raised
|
|
server_ip: The IP of the server where the error occurred
|
|
additional_info: Additional information to include in the response
|
|
keep_history: Whether to keep the error in the action history
|
|
service: The service where the error occurred
|
|
|
|
Returns:
|
|
A tuple containing the error response and HTTP status code
|
|
"""
|
|
# Log the error
|
|
log_error(exception, server_ip, additional_info, keep_history, service)
|
|
|
|
# Determine the exception type and get the appropriate status code and message
|
|
for exception_type, status_code in ERROR_CODE_MAPPING.items():
|
|
if isinstance(exception, exception_type):
|
|
message = ERROR_MESSAGE_MAPPING.get(exception_type, str(exception))
|
|
if additional_info:
|
|
message = f"{additional_info}: {message}"
|
|
|
|
# Create the error response
|
|
error_response = ErrorResponse(error=message).model_dump(mode='json')
|
|
return error_response, status_code
|
|
|
|
# If we get here, we don't have a specific handler for this exception type
|
|
error_response = ErrorResponse(error=str(exception)).model_dump(mode='json')
|
|
return error_response, 500
|
|
|
|
|
|
def register_error_handlers(app):
|
|
"""
|
|
Register error handlers for the Flask application.
|
|
|
|
Args:
|
|
app: The Flask application
|
|
"""
|
|
@app.errorhandler(Exception)
|
|
def handle_exception_error(e):
|
|
"""Handle all unhandled exceptions."""
|
|
if isinstance(e, HTTPException):
|
|
# Pass through HTTP exceptions
|
|
return e
|
|
|
|
# Log the error
|
|
log_error(e)
|
|
|
|
# Return a JSON response
|
|
error_response, status_code = handle_exception(e)
|
|
return jsonify(error_response), status_code
|
|
|
|
# Register handlers for specific HTTP errors
|
|
@app.errorhandler(ValidationError)
|
|
def handle_pydantic_validation_errors(e):
|
|
"""Handle validation errors from pydantic."""
|
|
errors = []
|
|
if e.body_params:
|
|
req_type = e.body_params
|
|
elif e.form_params:
|
|
req_type = e.form_params
|
|
elif e.path_params:
|
|
req_type = e.path_params
|
|
else:
|
|
req_type = e.query_params
|
|
for er in req_type:
|
|
if len(er["loc"]) > 0:
|
|
errors.append(f'{er["loc"][0]}: {er["msg"]}')
|
|
else:
|
|
errors.append(er["msg"])
|
|
return ErrorResponse(error=errors).model_dump(mode='json'), 400
|
|
|
|
@app.errorhandler(400)
|
|
def bad_request(e):
|
|
"""Handle 400 Bad Request errors."""
|
|
return jsonify(ErrorResponse(error="Bad request").model_dump(mode='json')), 400
|
|
|
|
@app.errorhandler(401)
|
|
def unauthorized(e):
|
|
"""Handle 401 Unauthorized errors."""
|
|
if 'api' in request.url:
|
|
return jsonify(ErrorResponse(error=str(e)).model_dump(mode='json')), 401
|
|
return redirect(url_for('login_page', next=request.full_path))
|
|
|
|
@app.errorhandler(403)
|
|
@get_user_params()
|
|
def forbidden(e):
|
|
"""Handle 403 Forbidden errors."""
|
|
if 'api' in request.url:
|
|
return jsonify(ErrorResponse(error=str(e)).model_dump(mode='json')), 403
|
|
|
|
kwargs = {
|
|
'user_params': g.user_params,
|
|
'title': e,
|
|
'e': e
|
|
}
|
|
return render_template('error.html', **kwargs), 403
|
|
|
|
@app.errorhandler(404)
|
|
@get_user_params()
|
|
def not_found(e):
|
|
"""Handle 404 Not Found errors."""
|
|
if 'api' in request.url:
|
|
return jsonify(ErrorResponse(error=str(e)).model_dump(mode='json')), 404
|
|
|
|
kwargs = {
|
|
'user_params': g.user_params,
|
|
'title': e,
|
|
'e': e
|
|
}
|
|
return render_template('error.html', **kwargs), 404
|
|
|
|
@app.errorhandler(405)
|
|
def method_not_allowed(e):
|
|
"""Handle 405 Method Not Allowed errors."""
|
|
return jsonify(ErrorResponse(error=str(e)).model_dump(mode='json')), 405
|
|
|
|
@app.errorhandler(415)
|
|
def unsupported_media_type(e):
|
|
"""Handle 415 Unsupported Media Type errors."""
|
|
return jsonify(ErrorResponse(error="Unsupported Media Type").model_dump(mode='json')), 415
|
|
|
|
@app.errorhandler(429)
|
|
def too_many_requests(e):
|
|
"""Handle 429 Too Many Requests errors."""
|
|
return jsonify(ErrorResponse(error="Too many requests").model_dump(mode='json')), 429
|
|
|
|
@app.errorhandler(500)
|
|
@get_user_params()
|
|
def internal_server_error(e):
|
|
"""Handle 500 Internal Server Error errors."""
|
|
if 'api' in request.url:
|
|
return jsonify(ErrorResponse(error=str(e)).model_dump(mode='json')), 500
|
|
|
|
kwargs = {
|
|
'user_params': g.user_params,
|
|
'title': e,
|
|
'e': e
|
|
}
|
|
return render_template('error.html', **kwargs), 500
|