157 lines
6.4 KiB
Python
157 lines
6.4 KiB
Python
from flask import request, session
|
|
import time
|
|
from flask_security import current_user
|
|
import json
|
|
|
|
|
|
def log_request_middleware(app):
|
|
# @app.before_request
|
|
# def log_request_info():
|
|
# start_time = time.time()
|
|
# app.logger.debug(f"Request URL: {request.url}")
|
|
# app.logger.debug(f"Request Method: {request.method}")
|
|
# app.logger.debug(f"Request Headers: {request.headers}")
|
|
# app.logger.debug(f"Time taken for logging request info: {time.time() - start_time} seconds")
|
|
# try:
|
|
# app.logger.debug(f"Request Body: {request.get_data()}")
|
|
# except Exception as e:
|
|
# app.logger.error(f"Error reading request body: {e}")
|
|
# app.logger.debug(f"Time taken for logging request body: {time.time() - start_time} seconds")
|
|
|
|
# @app.before_request
|
|
# def check_csrf_token():
|
|
# start_time = time.time()
|
|
# if request.method == "POST":
|
|
# csrf_token = request.form.get("csrf_token")
|
|
# app.logger.debug(f"CSRF Token: {csrf_token}")
|
|
# app.logger.debug(f"Time taken for logging CSRF token: {time.time() - start_time} seconds")
|
|
|
|
# @app.before_request
|
|
# def log_user_info():
|
|
# if current_user and current_user.is_authenticated:
|
|
# app.logger.debug(f"Before: User ID: {current_user.id}")
|
|
# app.logger.debug(f"Before: User Email: {current_user.email}")
|
|
# app.logger.debug(f"Before: User Roles: {current_user.roles}")
|
|
# else:
|
|
# app.logger.debug("After: No user logged in")
|
|
|
|
@app.before_request
|
|
def log_session_state_before():
|
|
pass
|
|
|
|
# @app.after_request
|
|
# def log_response_info(response):
|
|
# start_time = time.time()
|
|
# app.logger.debug(f"Response Status: {response.status}")
|
|
# app.logger.debug(f"Response Headers: {response.headers}")
|
|
#
|
|
# app.logger.debug(f"Time taken for logging response info: {time.time() - start_time} seconds")
|
|
# return response
|
|
|
|
# @app.after_request
|
|
# def log_user_after_request(response):
|
|
# if current_user and current_user.is_authenticated:
|
|
# app.logger.debug(f"After: User ID: {current_user.id}")
|
|
# app.logger.debug(f"after: User Email: {current_user.email}")
|
|
# app.logger.debug(f"After: User Roles: {current_user.roles}")
|
|
# else:
|
|
# app.logger.debug("After: No user logged in")
|
|
|
|
@app.after_request
|
|
def log_session_state_after(response):
|
|
return response
|
|
|
|
|
|
def register_request_debugger(app):
|
|
@app.before_request
|
|
def debug_request_info():
|
|
"""Log consolidated request information for debugging"""
|
|
# Skip health check endpoints
|
|
if request.path.startswith('/_healthz') or request.path.startswith('/healthz'):
|
|
return
|
|
|
|
# Gather all request information in a structured way
|
|
debug_info = {
|
|
"basic_info": {
|
|
"method": request.method,
|
|
"path": request.path,
|
|
"content_type": request.content_type,
|
|
"content_length": request.content_length
|
|
},
|
|
"environment": {
|
|
"remote_addr": request.remote_addr,
|
|
"user_agent": str(request.user_agent)
|
|
}
|
|
}
|
|
|
|
# Add headers (excluding sensitive ones)
|
|
safe_headers = {k: v for k, v in request.headers.items()
|
|
if k.lower() not in ('authorization', 'cookie', 'x-api-key')}
|
|
debug_info["headers"] = safe_headers
|
|
|
|
# Add authentication info (presence only)
|
|
auth_header = request.headers.get('Authorization', '')
|
|
debug_info["auth_info"] = {
|
|
"has_auth_header": bool(auth_header),
|
|
"auth_type": auth_header.split(' ')[0] if auth_header else None,
|
|
"token_length": len(auth_header.split(' ')[1]) if auth_header and len(auth_header.split(' ')) > 1 else 0,
|
|
"header_format": 'Valid format' if auth_header.startswith('Bearer ') else 'Invalid format',
|
|
"raw_header": auth_header[:10] + '...' if auth_header else None # Show first 10 chars only
|
|
}
|
|
|
|
# Add request data based on type
|
|
if request.is_json:
|
|
try:
|
|
json_data = request.get_json()
|
|
if isinstance(json_data, dict):
|
|
# Remove sensitive fields from logging
|
|
safe_json = {k: v for k, v in json_data.items()
|
|
if not any(sensitive in k.lower()
|
|
for sensitive in ['password', 'token', 'secret', 'key'])}
|
|
debug_info["request_data"] = {
|
|
"type": "json",
|
|
"content": safe_json
|
|
}
|
|
except Exception as e:
|
|
debug_info["request_data"] = {
|
|
"type": "json",
|
|
"error": str(e)
|
|
}
|
|
elif request.form:
|
|
safe_form = {k: v for k, v in request.form.items()
|
|
if not any(sensitive in k.lower()
|
|
for sensitive in ['password', 'token', 'secret', 'key'])}
|
|
debug_info["request_data"] = {
|
|
"type": "form",
|
|
"content": safe_form
|
|
}
|
|
|
|
# Add file information if present
|
|
if request.files:
|
|
debug_info["files"] = {
|
|
name: {
|
|
"filename": f.filename,
|
|
"content_type": f.content_type,
|
|
"content_length": f.content_length if hasattr(f, 'content_length') else None
|
|
}
|
|
for name, f in request.files.items()
|
|
}
|
|
|
|
# Add CORS information if present
|
|
cors_headers = {
|
|
"origin": request.headers.get('Origin'),
|
|
"request_method": request.headers.get('Access-Control-Request-Method'),
|
|
"request_headers": request.headers.get('Access-Control-Request-Headers')
|
|
}
|
|
if any(cors_headers.values()):
|
|
debug_info["cors"] = {k: v for k, v in cors_headers.items() if v is not None}
|
|
|
|
# Format the debug info as a pretty-printed JSON string with indentation
|
|
formatted_debug_info = json.dumps(debug_info, indent=2, sort_keys=True)
|
|
|
|
# Log everything in a single statement
|
|
app.logger.debug(
|
|
"Request Debug Information\n",
|
|
extra={"request_debug\n": formatted_debug_info}
|
|
)
|