Azeez Muibi
115 lines
3.7 KiB
Python
115 lines
3.7 KiB
Python
"""
|
|
Middleware module for the Flask application.
|
|
"""
|
|
from flask import Flask, request, jsonify, g, current_app
|
|
from flask.typing import ResponseReturnValue
|
|
from functools import wraps
|
|
import base64
|
|
from typing import Callable, Any, TypeVar, cast
|
|
import logging
|
|
|
|
# Configure logger
|
|
logger = logging.getLogger(__name__)
|
|
|
|
F = TypeVar('F', bound=Callable[..., Any])
|
|
|
|
def register_middleware(app: Flask) -> None:
|
|
"""
|
|
Register middleware with the Flask application.
|
|
|
|
Args:
|
|
app: Flask application instance
|
|
"""
|
|
# Register CORS if needed
|
|
try:
|
|
from flask_cors import CORS
|
|
CORS(app, resources={r"/*": {"origins": app.config.get('CORS_ORIGINS', '*')}})
|
|
except ImportError:
|
|
logger.warning("flask-cors not installed. CORS support disabled.")
|
|
|
|
@app.before_request
|
|
def before_request() -> None:
|
|
"""Process request before it reaches the view function."""
|
|
# Log incoming requests
|
|
logger.debug(f"Received {request.method} request to {request.path}")
|
|
|
|
# You can add more global middleware here if needed
|
|
|
|
def basic_auth_required(f: F) -> F:
|
|
"""
|
|
Decorator for endpoints that require basic authentication.
|
|
|
|
Args:
|
|
f: Function to decorate
|
|
|
|
Returns:
|
|
Decorated function
|
|
"""
|
|
@wraps(f)
|
|
def decorated(*args: Any, **kwargs: Any) -> ResponseReturnValue:
|
|
auth = request.headers.get('Authorization')
|
|
if not auth or not auth.startswith('Basic '):
|
|
logger.warning("Authentication failed: No Basic Auth header")
|
|
return jsonify({
|
|
'resultCode': '01',
|
|
'resultDescription': 'Authentication required'
|
|
}), 401
|
|
|
|
try:
|
|
auth_decoded = base64.b64decode(auth[6:]).decode('utf-8')
|
|
username, password = auth_decoded.split(':', 1)
|
|
|
|
if username != current_app.config['API_USERNAME'] or password != current_app.config['API_PASSWORD']:
|
|
logger.warning(f"Authentication failed: Invalid credentials for user {username}")
|
|
return jsonify({
|
|
'resultCode': '01',
|
|
'resultDescription': 'Invalid credentials'
|
|
}), 401
|
|
|
|
g.user = username
|
|
logger.debug(f"Authentication successful for user {username}")
|
|
return f(*args, **kwargs)
|
|
except Exception as e:
|
|
logger.error(f"Authentication error: {str(e)}")
|
|
return jsonify({
|
|
'resultCode': '01',
|
|
'resultDescription': 'Invalid authentication format'
|
|
}), 401
|
|
|
|
return cast(F, decorated)
|
|
|
|
def api_key_required(f: F) -> F:
|
|
"""
|
|
Decorator for endpoints that require API key authentication.
|
|
|
|
Args:
|
|
f: Function to decorate
|
|
|
|
Returns:
|
|
Decorated function
|
|
"""
|
|
@wraps(f)
|
|
def decorated(*args: Any, **kwargs: Any) -> ResponseReturnValue:
|
|
app_id = request.headers.get('appID')
|
|
api_key = request.headers.get('apiKey')
|
|
|
|
if not app_id or not api_key:
|
|
logger.warning("API key authentication failed: Missing appID or apiKey")
|
|
return jsonify({
|
|
'resultCode': '01',
|
|
'resultDescription': 'API key authentication required'
|
|
}), 401
|
|
|
|
# Validate against configured API keys
|
|
if app_id != current_app.config['SIMBRELLA_APP_ID'] or api_key != current_app.config['SIMBRELLA_API_KEY']:
|
|
logger.warning(f"API key authentication failed: Invalid appID or apiKey")
|
|
return jsonify({
|
|
'resultCode': '01',
|
|
'resultDescription': 'Invalid API keys'
|
|
}), 401
|
|
|
|
g.api_client = 'simbrella'
|
|
logger.debug(f"API key authentication successful")
|
|
return f(*args, **kwargs)
|
|
|
|
return cast(F, decorated) |