""" JWT Middleware Module Provides decorator to protect Flask routes with JWT authentication Automatically checks auth status and validates tokens """ from flask import request, jsonify, g 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): # Internal calls (background prewarmers, in-process cache # refresh) bypass auth. Set `g._internal_call = True` inside # an `app.test_request_context()` block before invoking a # decorated handler — the flag lives only for that context so # a real HTTP request can never accidentally inherit it. try: if getattr(g, '_internal_call', False): return f(*args, **kwargs) except RuntimeError: pass # No request context yet — treat as normal auth flow. # 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