""" JWT Middleware Module Provides decorator to protect Flask routes with JWT authentication Automatically checks auth status and validates tokens """ from flask import request, jsonify from functools import wraps from auth_manager import load_auth_config, verify_token, verify_token_full def require_auth(f): """ Decorator to protect Flask routes with JWT authentication Behavior: - If auth is disabled or declined: Allow access (no token required) - If auth is enabled: Require valid JWT token in Authorization header - Returns 401 if auth required but token missing/invalid Usage: @app.route('/api/protected') @require_auth def protected_route(): return jsonify({"data": "secret"}) """ @wraps(f) def decorated_function(*args, **kwargs): # Check if authentication is enabled config = load_auth_config() # If auth is disabled or declined, allow access if not config.get("enabled", False) or config.get("declined", False): return f(*args, **kwargs) # Auth is enabled, require token auth_header = request.headers.get('Authorization') if not auth_header: return jsonify({ "error": "Authentication required", "message": "No authorization header provided" }), 401 # Extract token from "Bearer " format parts = auth_header.split() if len(parts) != 2 or parts[0].lower() != 'bearer': return jsonify({ "error": "Invalid authorization header", "message": "Authorization header must be in format: Bearer " }), 401 token = parts[1] # Verify token username = verify_token(token) if not username: return jsonify({ "error": "Invalid or expired token", "message": "Please log in again" }), 401 # Token is valid, allow access return f(*args, **kwargs) return decorated_function def require_auth_or_ticket(f): """Like `require_auth` but ALSO accepts a single-use `?ticket=...` query parameter (same tickets `/api/terminal/ticket` issues for WebSockets). Use on endpoints that the browser needs to invoke from an tag — anchor tags can't send the Authorization header, so the caller fetches a ticket first and appends it to the URL. Use only for streaming downloads where adding `?ticket=...` is the only practical way to authenticate; everything else stays on plain Bearer-token auth.""" @wraps(f) def decorated_function(*args, **kwargs): config = load_auth_config() if not config.get("enabled", False) or config.get("declined", False): return f(*args, **kwargs) # First try the Bearer header (fetch from JS uses this). auth_header = request.headers.get('Authorization') if auth_header: parts = auth_header.split() if len(parts) == 2 and parts[0].lower() == 'bearer': if verify_token(parts[1]): return f(*args, **kwargs) # Fall through to single-use ticket (works for ). try: from flask_terminal_routes import _consume_terminal_ticket if _consume_terminal_ticket(request.args.get('ticket', '')): return f(*args, **kwargs) except ImportError: pass return jsonify({ "error": "Authentication required", "message": "Provide a Bearer token in the Authorization header or a fresh ?ticket=... from /api/terminal/ticket" }), 401 return decorated_function def require_admin_scope(f): """Like `require_auth` but ALSO requires the token's `scope == full_admin`. Use on mutating routes that should be off-limits to read-only API tokens (e.g. script execution, SSL disable, auth setup). Tokens generated by the session login flow inherit `full_admin` implicitly; long-lived API tokens default to `read_only` unless the caller opted in. Audit Tier 6 — Tokens API JWT 365 días sin scope. """ @wraps(f) def decorated_function(*args, **kwargs): config = load_auth_config() if not config.get("enabled", False) or config.get("declined", False): return f(*args, **kwargs) auth_header = request.headers.get('Authorization') if not auth_header: return jsonify({"error": "Authentication required", "message": "No authorization header provided"}), 401 parts = auth_header.split() if len(parts) != 2 or parts[0].lower() != 'bearer': return jsonify({"error": "Invalid authorization header", "message": "Authorization header must be in format: Bearer "}), 401 username, scope = verify_token_full(parts[1]) if not username: return jsonify({"error": "Invalid or expired token", "message": "Please log in again"}), 401 if scope != 'full_admin': return jsonify({"error": "Insufficient scope", "message": f"This action requires a full_admin token (your token: {scope})"}), 403 return f(*args, **kwargs) return decorated_function def optional_auth(f): """ Decorator for routes that can optionally use auth Passes username if authenticated, None otherwise Usage: @app.route('/api/optional') @optional_auth def optional_route(username=None): if username: return jsonify({"message": f"Hello {username}"}) return jsonify({"message": "Hello guest"}) """ @wraps(f) def decorated_function(*args, **kwargs): config = load_auth_config() username = None if config.get("enabled", False): auth_header = request.headers.get('Authorization') if auth_header: parts = auth_header.split() if len(parts) == 2 and parts[0].lower() == 'bearer': username = verify_token(parts[1]) # Inject username into kwargs kwargs['username'] = username return f(*args, **kwargs) return decorated_function