You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
82 lines
2.5 KiB
82 lines
2.5 KiB
from services.log import access_logger |
|
import time |
|
from typing import Optional |
|
from fastapi import Request |
|
from starlette.middleware.base import BaseHTTPMiddleware |
|
from nicegui import app |
|
import uuid |
|
|
|
|
|
|
|
def get_client_ip(request: Request) -> str: |
|
"""Extract client IP address with X-Forwarded-For support""" |
|
# Check X-Forwarded-For header first (for reverse proxies) |
|
forwarded_for = request.headers.get('X-Forwarded-For') |
|
if forwarded_for: |
|
# Take the first IP in the chain (original client) |
|
return forwarded_for.split(',')[0].strip() |
|
|
|
# Check X-Real-IP header (nginx) |
|
real_ip = request.headers.get('X-Real-IP') |
|
if real_ip: |
|
return real_ip.strip() |
|
|
|
# Fall back to direct connection IP |
|
return request.client.host if request.client else 'unknown' |
|
|
|
|
|
def get_user_agent(request: Request) -> str: |
|
"""Extract user agent string""" |
|
return request.headers.get('User-Agent', 'unknown') |
|
|
|
|
|
def get_current_user() -> Optional[str]: |
|
"""Get current authenticated user""" |
|
if app.storage.user.get('authenticated', False): |
|
return app.storage.user.get('username', 'authenticated_user') |
|
return None |
|
|
|
|
|
|
|
|
|
class AccessLoggingMiddleware(BaseHTTPMiddleware): |
|
"""Middleware to log all HTTP requests with client IP and user info""" |
|
|
|
async def dispatch(self, request: Request, call_next): |
|
start_time = time.time() |
|
request_id = str(uuid.uuid4()) # Generate a unique request ID |
|
client_ip = get_client_ip(request) |
|
user_agent = get_user_agent(request) |
|
user = get_current_user() |
|
|
|
|
|
|
|
# Process the request |
|
response = await call_next(request) |
|
|
|
# Calculate processing time |
|
process_time = time.time() - start_time |
|
|
|
# Log the access |
|
log_message = ( |
|
f"{request.method} {request.url.path} " |
|
f"{response.status_code} " |
|
f"\"{user_agent}\"" |
|
) |
|
access_logger.extra = { |
|
'request_id': request_id, |
|
'client_ip': client_ip, |
|
'user_agent': user_agent, |
|
'user': user if user else 'anonymous', |
|
'process_time': f"{process_time:.3f}s", |
|
} |
|
|
|
# Use different log levels based on status code |
|
if response.status_code >= 500: |
|
access_logger.error(log_message) |
|
elif response.status_code >= 400: |
|
access_logger.warning(log_message) |
|
else: |
|
access_logger.info(log_message) |
|
|
|
return response
|
|
|