159 lines
4.4 KiB
Python
159 lines
4.4 KiB
Python
"""Flask application factory and middleware setup."""
|
|
|
|
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import socket
|
|
from datetime import datetime, timezone
|
|
from logging.handlers import RotatingFileHandler
|
|
|
|
from flask import Flask, request, Response
|
|
from flask_cors import CORS
|
|
from flask_compress import Compress
|
|
|
|
from config import load_config
|
|
|
|
# Bunyan level mapping
|
|
_BUNYAN_LEVELS = {
|
|
logging.DEBUG: 20,
|
|
logging.INFO: 30,
|
|
logging.WARNING: 40,
|
|
logging.ERROR: 50,
|
|
logging.CRITICAL: 60,
|
|
}
|
|
|
|
# Reverse mapping: string name → Python level
|
|
_LEVEL_NAMES = {
|
|
'trace': logging.DEBUG,
|
|
'debug': logging.DEBUG,
|
|
'info': logging.INFO,
|
|
'warn': logging.WARNING,
|
|
'error': logging.ERROR,
|
|
'fatal': logging.CRITICAL,
|
|
}
|
|
|
|
|
|
class BunyanFormatter(logging.Formatter):
|
|
"""Produce bunyan-style JSON log lines.
|
|
|
|
Each log line is a single JSON object with keys:
|
|
name, hostname, pid, level, msg, time, v
|
|
Matches the format expected by the existing test suite.
|
|
"""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.hostname = socket.gethostname()
|
|
self.pid = os.getpid()
|
|
|
|
def format(self, record):
|
|
log_entry = {
|
|
'name': 'cteward-st-lexware',
|
|
'hostname': self.hostname,
|
|
'pid': self.pid,
|
|
'level': _BUNYAN_LEVELS.get(record.levelno, 30),
|
|
'msg': record.getMessage(),
|
|
'time': datetime.now(timezone.utc).isoformat(),
|
|
'v': 0,
|
|
}
|
|
# Attach request context if available
|
|
if hasattr(record, 'username'):
|
|
log_entry['username'] = record.username
|
|
if hasattr(record, 'method'):
|
|
log_entry['method'] = record.method
|
|
if hasattr(record, 'url'):
|
|
log_entry['url'] = record.url
|
|
return json.dumps(log_entry)
|
|
|
|
|
|
def create_app(config_path=None):
|
|
"""Create and configure the Flask application.
|
|
|
|
Replaces the restify server + middleware chain from startup.js.
|
|
"""
|
|
app = Flask(__name__)
|
|
app.config.from_mapping(
|
|
SECRET_KEY='cteward-st-lexware',
|
|
)
|
|
|
|
# Load runtime config (JSON file)
|
|
app.cteward_config = load_config(config_path)
|
|
|
|
_setup_logging(app)
|
|
_setup_extensions(app)
|
|
_register_prehandlers(app)
|
|
_register_blueprints(app)
|
|
|
|
return app
|
|
|
|
|
|
def _setup_logging(app):
|
|
"""Setup structured JSON logging similar to bunyan."""
|
|
log_level_name = app.cteward_config.get('loglevel', 'info').lower()
|
|
logfile = app.cteward_config.get('logfile')
|
|
|
|
log_level = _LEVEL_NAMES.get(log_level_name, logging.INFO)
|
|
|
|
handler = (
|
|
RotatingFileHandler(logfile)
|
|
if logfile
|
|
else logging.StreamHandler()
|
|
)
|
|
handler.setLevel(log_level)
|
|
handler.setFormatter(BunyanFormatter())
|
|
|
|
app.logger.handlers.clear()
|
|
app.logger.addHandler(handler)
|
|
app.logger.setLevel(log_level)
|
|
|
|
|
|
def _setup_extensions(app):
|
|
"""Enable CORS and gzip compression."""
|
|
CORS(app)
|
|
Compress(app)
|
|
|
|
|
|
def _extract_basic_username(headers):
|
|
"""Extract username from Basic auth header for logging."""
|
|
auth_header = headers.get('Authorization', '')
|
|
if not auth_header:
|
|
return 'anonymous'
|
|
parts = auth_header.split(' ', 1)
|
|
if len(parts) == 2 and parts[0] == 'Basic':
|
|
try:
|
|
decoded = base64.b64decode(parts[1]).decode('utf-8')
|
|
return decoded.split(':')[0]
|
|
except Exception:
|
|
return 'anonymous'
|
|
return 'anonymous'
|
|
|
|
|
|
def _register_prehandlers(app):
|
|
"""Before-request hooks: extract username for logging."""
|
|
|
|
@app.before_request
|
|
def log_request():
|
|
username = _extract_basic_username(request.headers)
|
|
extra = {'username': username, 'method': request.method, 'url': request.url}
|
|
# Attach extra fields to the log record so BunyanFormatter picks them up
|
|
app.logger.info(
|
|
'%s %s %s', username, request.method, request.url,
|
|
extra=extra,
|
|
)
|
|
|
|
@app.after_request
|
|
def www_authenticate(response):
|
|
"""Add WWW-Authenticate header when no auth is present."""
|
|
if 'Authorization' not in request.headers:
|
|
realm = app.cteward_config.get('server', {}).get(
|
|
'realm', 'cteward API access'
|
|
)
|
|
response.headers['WWW-Authenticate'] = f'Basic realm="{realm}"'
|
|
return response
|
|
|
|
|
|
def _register_blueprints(app):
|
|
"""Register all route blueprints."""
|
|
from views import legacy_bp
|
|
app.register_blueprint(legacy_bp, url_prefix='/legacy')
|