""" Authorization decorators for RateRight Australian Construction Marketplace Provides role-based access control for protected routes """ from functools import wraps from flask import jsonify, flash, redirect, url_for, request, current_app from flask_login import current_user import os def admin_api_key_required(f): """ Decorator to require valid admin API key for route access Checks X-Admin-API-Key header against ADMIN_API_KEY config Usage: @app.route('/api/admin/sms/send') @admin_api_key_required def send_sms(): ... """ @wraps(f) def decorated_function(*args, **kwargs): api_key = request.headers.get('X-Admin-API-Key') if not api_key: return jsonify({'error': 'Admin API key required'}), 401 # Get the admin API key from config or environment valid_key = current_app.config.get('ADMIN_API_KEY') or os.environ.get('ADMIN_API_KEY') if not valid_key: return jsonify({'error': 'Admin API key not configured on server'}), 500 if api_key != valid_key: return jsonify({'error': 'Invalid admin API key'}), 403 return f(*args, **kwargs) return decorated_function def super_admin_required(f): """ Decorator to require super_admin role for route access. Also accepts valid X-API-Key header matching ADMIN_API_KEY. Returns 401 if not authenticated, 403 if not super admin Usage: @app.route('/admin/users') @login_required @super_admin_required def admin_users(): ... """ @wraps(f) def decorated_function(*args, **kwargs): # Check API key first api_key = request.headers.get('X-API-Key') if api_key: admin_api_key = current_app.config.get('ADMIN_API_KEY') if admin_api_key and api_key == admin_api_key: return f(*args, **kwargs) # Fall back to standard admin authentication if not current_user.is_authenticated: if request.is_json: return jsonify({'error': 'Authentication required'}), 401 flash('Please log in to access this page.', 'error') return redirect(url_for('auth.login')) if current_user.role != 'super_admin': if request.is_json: return jsonify({'error': 'Admin access required'}), 403 flash('Access denied. Admin privileges required.', 'error') return redirect(url_for('dashboard')) return f(*args, **kwargs) return decorated_function def role_required(*allowed_roles): """ Flexible decorator to require one of multiple roles Usage: @app.route('/contractor/dashboard') @login_required @role_required('contractor', 'super_admin') def contractor_dashboard(): ... """ def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): if not current_user.is_authenticated: if request.is_json: return jsonify({'error': 'Authentication required'}), 401 flash('Please log in to access this page.', 'error') return redirect(url_for('auth.login')) if current_user.role not in allowed_roles: if request.is_json: return jsonify({ 'error': f'Access denied. Required role: {", ".join(allowed_roles)}' }), 403 flash('Access denied. Insufficient permissions.', 'error') return redirect(url_for('dashboard')) return f(*args, **kwargs) return decorated_function return decorator def worker_required(f): """Decorator to require worker role""" @wraps(f) def decorated_function(*args, **kwargs): if not current_user.is_authenticated: if request.is_json: return jsonify({'error': 'Authentication required'}), 401 flash('Please log in to access this page.', 'error') return redirect(url_for('auth.login')) if current_user.role != 'worker': if request.is_json: return jsonify({'error': 'Worker access required'}), 403 flash('This page is only accessible to workers.', 'error') return redirect(url_for('dashboard')) return f(*args, **kwargs) return decorated_function def contractor_required(f): """Decorator to require contractor role""" @wraps(f) def decorated_function(*args, **kwargs): if not current_user.is_authenticated: if request.is_json: return jsonify({'error': 'Authentication required'}), 401 flash('Please log in to access this page.', 'error') return redirect(url_for('auth.login')) if current_user.role not in ['contractor', 'employer']: if request.is_json: return jsonify({'error': 'Contractor access required'}), 403 flash('This page is only accessible to contractors.', 'error') return redirect(url_for('dashboard')) return f(*args, **kwargs) return decorated_function def onboarding_required(f): """ Decorator to ensure workers have completed profile and payment setup before accessing protected routes. Checks: 1. Profile completion (required fields, insurance, trade) 2. Payment setup (Stripe onboarding) Redirects to appropriate setup page if incomplete. Usage: @app.route('/dashboard') @login_required @onboarding_required def dashboard(): ... """ @wraps(f) def decorated_function(*args, **kwargs): # Only enforce for authenticated workers if not current_user.is_authenticated: if request.is_json: return jsonify({'error': 'Authentication required'}), 401 flash('Please log in to access this page.', 'error') return redirect(url_for('login')) # Only enforce for workers (contractors and admins can access immediately) if current_user.role != 'worker': return f(*args, **kwargs) # Check onboarding status needs_onboarding, missing_steps = current_user.needs_onboarding() if not needs_onboarding: # All set! Continue to the route return f(*args, **kwargs) # Worker needs to complete onboarding if 'profile' in missing_steps: # Profile incomplete - redirect to profile setup if request.is_json: return jsonify({ 'error': 'Profile setup required', 'redirect': '/onboarding/profile' }), 403 flash('Please complete your profile setup to continue.', 'warning') return redirect(url_for('onboarding_profile')) if 'payment' in missing_steps: # Payment setup incomplete - redirect to payment setup if request.is_json: return jsonify({ 'error': 'Payment setup required', 'redirect': '/onboarding/payment' }), 403 flash('Please complete your payment setup to start receiving payments.', 'warning') return redirect(url_for('onboarding_payment')) # Shouldn't reach here, but safety fallback return f(*args, **kwargs) return decorated_function