from flask import Flask, render_template, request, jsonify, redirect, url_for, session, make_response from functools import wraps from werkzeug.middleware.proxy_fix import ProxyFix from datetime import timedelta import malias_wrapper as malias_w import os import argparse import sys import secrets import logging # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger('mailcow-alias-manager') app = Flask(__name__) app.secret_key = os.getenv('SECRET_KEY', 'malias-default-secret-key-please-change') # Consistent secret key # Configure for reverse proxy - modified to handle multiple proxy layers including Authelia # Increasing the number of proxies to handle Zoraxy->Authelia->App chain app.wsgi_app = ProxyFix( app.wsgi_app, x_for=2, # Trust X-Forwarded-For with 2 proxies (Zoraxy + Authelia) x_proto=2, # Trust X-Forwarded-Proto (http/https) x_host=2, # Trust X-Forwarded-Host x_prefix=2 # Trust X-Forwarded-Prefix ) # Initialize database on startup malias_w.init_database() # Session configuration optimized for reverse proxy with Authelia app.config.update( PERMANENT_SESSION_LIFETIME=timedelta(hours=24), SESSION_COOKIE_NAME='malias_session', # Unique name to avoid conflicts with Authelia SESSION_COOKIE_SECURE=True, # Always use secure cookies with Authelia SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE='None', # Required for authentication proxies SESSION_COOKIE_PATH='/', SESSION_COOKIE_DOMAIN=None, # Let browser auto-set domain SESSION_REFRESH_EACH_REQUEST=True, # Keep session alive PREFERRED_URL_SCHEME='https' ) @app.after_request def add_security_headers(response): """Add security headers to every response""" # Add SameSite=None explicitly for all cookies if Auth proxy is detected if request.headers.get('Remote-User') or request.headers.get('X-Remote-User'): cookie_header = response.headers.get('Set-Cookie', '') if cookie_header and 'SameSite=' not in cookie_header: cookie_header = cookie_header.replace('HttpOnly', 'HttpOnly; SameSite=None; Secure') response.headers['Set-Cookie'] = cookie_header return response def get_authelia_user(): """Helper to get authenticated user from Authelia headers""" # Check several possible header variations auth_headers = [ 'Remote-User', 'X-Remote-User', 'X-Authelia-Username', 'X-Forwarded-User', 'REMOTE_USER' ] for header in auth_headers: user = request.headers.get(header) if user: logger.info(f"Authelia user detected via {header}: {user}") return user return None def login_required(f): """Decorator to require login for routes""" @wraps(f) def decorated_function(*args, **kwargs): # Check for Authelia authentication authelia_user = get_authelia_user() # If Authelia authenticated the user, update local session if authelia_user: # Log all headers for debugging if app.debug: logger.info(f"Headers for authenticated request: {dict(request.headers)}") if not session.get('logged_in') or session.get('authelia_user') != authelia_user: logger.info(f"Auto-login via Authelia for user: {authelia_user}") session.clear() session.permanent = True session['logged_in'] = True session['authelia_user'] = authelia_user session['user_token'] = secrets.token_urlsafe(32) session['auth_method'] = 'authelia' session.modified = True return f(*args, **kwargs) # Regular session check if not session.get('logged_in'): logger.warning("Access denied: User not authenticated") return jsonify({'status': 'error', 'message': 'Not authenticated', 'redirect': '/login'}), 401 return f(*args, **kwargs) return decorated_function @app.route('/login', methods=['GET', 'POST']) def login(): """Login page""" # First, try Authelia authentication authelia_user = get_authelia_user() # Debug logging for all requests if app.debug: logger.info(f"Login route: method={request.method}, headers={dict(request.headers)}") # If Authelia authenticated, login and redirect to index if authelia_user: logger.info(f"Login via Authelia for user: {authelia_user}") session.clear() session.permanent = True session['logged_in'] = True session['authelia_user'] = authelia_user session['user_token'] = secrets.token_urlsafe(32) session['auth_method'] = 'authelia' session.modified = True return redirect(url_for('index')) # Handle form submission for local authentication if request.method == 'POST': password = request.json.get('password', '') logger.info("Login attempt with password (redacted)") if malias_w.verify_password(password): logger.info("Login successful with local password") session.clear() session.permanent = True session['logged_in'] = True session['user_token'] = secrets.token_urlsafe(32) session['auth_method'] = 'local' session.modified = True response = jsonify({'status': 'success', 'message': 'Login successful'}) return response else: logger.warning("Login failed: Invalid password") return jsonify({'status': 'error', 'message': 'Invalid password'}) # If already logged in, redirect to index if session.get('logged_in'): return redirect(url_for('index')) # Show login form return render_template('login.html') @app.route('/logout') def logout(): """Logout""" # Get auth method before clearing session auth_method = session.get('auth_method') authelia_user = session.get('authelia_user') # Clear local session session.clear() # If user was authenticated via Authelia, try to redirect to Authelia logout if auth_method == 'authelia' or authelia_user: # Look for Authelia URL in headers authelia_url = request.headers.get('X-Authelia-URL') # If found, redirect to Authelia logout if authelia_url: logger.info(f"Redirecting to Authelia logout: {authelia_url}/logout") return redirect(f"{authelia_url}/logout") # Try some common authelia URLs based on the request if request.host: domain_parts = request.host.split('.') if len(domain_parts) >= 2: base_domain = '.'.join(domain_parts[1:]) # e.g., extract 'example.com' from 'app.example.com' common_authelia_urls = [ f"https://auth.{base_domain}/logout", f"https://authelia.{base_domain}/logout", f"https://sso.{base_domain}/logout" ] # Try the first one as a fallback logger.info(f"No Authelia URL header, trying fallback: {common_authelia_urls[0]}") return redirect(common_authelia_urls[0]) # Default case: redirect to login page return redirect(url_for('login')) @app.route('/') def index(): """Main page - requires login""" # Try to auto-login with Authelia authelia_user = get_authelia_user() if authelia_user and not session.get('logged_in'): # Auto-login for users authenticated by Authelia logger.info(f"Auto-login via Authelia for user: {authelia_user}") session.clear() session.permanent = True session['logged_in'] = True session['authelia_user'] = authelia_user session['user_token'] = secrets.token_urlsafe(32) session['auth_method'] = 'authelia' session.modified = True # Check if logged in if not session.get('logged_in'): return redirect(url_for('login')) # Show main page return render_template('index.html') @app.route('/list_aliases', methods=['POST']) @login_required def list_aliases(): """List all aliases from Mailcow with pagination""" try: connection = malias_w.get_settings_from_db() if not connection: return jsonify({'status': 'error', 'message': 'Please configure Mailcow server first'}) # Get page number from request, default to 1 page = request.json.get('page', 1) if request.json else 1 result = malias_w.get_all_aliases(page=page, per_page=20) return jsonify({'status': 'success', 'data': result}) except Exception as e: logger.exception("Error listing aliases") return jsonify({'status': 'error', 'message': f"Error: {str(e)}"}) @app.route('/sync_aliases', methods=['POST']) @login_required def sync_aliases(): """Sync aliases from server to local DB""" try: count = malias_w.update_aliases() result = f"Aliases synchronized successfully! Added {count} new aliases to local DB." return jsonify({'status': 'success', 'message': result}) except Exception as e: logger.exception("Error syncing aliases") return jsonify({'status': 'error', 'message': f"Error syncing: {str(e)}"}) @app.route('/get_domains', methods=['POST']) @login_required def get_domains(): """Get all mail domains""" try: domains = malias_w.get_domains() domain_list = [d['domain_name'] for d in domains] result = f"Domains: {', '.join(domain_list)}" return jsonify({'status': 'success', 'message': result, 'domains': domain_list}) except Exception as e: logger.exception("Error getting domains") return jsonify({'status': 'error', 'message': f"Error: {str(e)}"}) @app.route('/create_alias', methods=['POST']) @login_required def create_alias(): """Create a new alias""" try: data = request.json alias = data.get('alias', '') goto = data.get('goto', '') if not alias or not goto: return jsonify({'status': 'error', 'message': 'Both alias and destination are required'}) malias_w.create_alias(alias, goto) result = f"Alias {alias} created successfully for {goto}" return jsonify({'status': 'success', 'message': result}) except Exception as e: logger.exception("Error creating alias") return jsonify({'status': 'error', 'message': f"Error creating alias: {str(e)}"}) @app.route('/delete_alias', methods=['POST']) @login_required def delete_alias_route(): """Delete an alias""" try: data = request.json alias = data.get('alias', '') if not alias: return jsonify({'status': 'error', 'message': 'Alias is required'}) malias_w.delete_alias(alias) result = f"Alias {alias} deleted successfully" return jsonify({'status': 'success', 'message': result}) except Exception as e: logger.exception("Error deleting alias") return jsonify({'status': 'error', 'message': f"Error deleting alias: {str(e)}"}) @app.route('/delete_aliases_bulk', methods=['POST']) @login_required def delete_aliases_bulk(): """Delete multiple aliases at once""" try: data = request.json aliases = data.get('aliases', []) if not aliases or not isinstance(aliases, list): return jsonify({'status': 'error', 'message': 'Aliases array is required'}) if len(aliases) == 0: return jsonify({'status': 'error', 'message': 'No aliases provided'}) # Delete multiple aliases result = malias_w.delete_multiple_aliases(aliases) # Build response message success_count = result['success_count'] failed_count = result['failed_count'] total_count = success_count + failed_count if failed_count == 0: message = f"{success_count} of {total_count} aliases deleted successfully" return jsonify({ 'status': 'success', 'message': message, 'details': result }) elif success_count == 0: # All failed failed_list = ', '.join([f['alias'] for f in result['failed_aliases'][:3]]) if len(result['failed_aliases']) > 3: failed_list += f" and {len(result['failed_aliases']) - 3} more" message = f"Failed to delete aliases: {failed_list}" return jsonify({ 'status': 'error', 'message': message, 'details': result }) else: # Partial success failed_list = ', '.join([f['alias'] for f in result['failed_aliases'][:3]]) if len(result['failed_aliases']) > 3: failed_list += f" and {len(result['failed_aliases']) - 3} more" message = f"{success_count} of {total_count} aliases deleted successfully. Failed: {failed_list}" return jsonify({ 'status': 'partial', 'message': message, 'details': result }) except Exception as e: logger.exception("Error deleting aliases bulk") return jsonify({'status': 'error', 'message': f"Error deleting aliases: {str(e)}"}) @app.route('/create_timed_alias', methods=['POST']) @login_required def create_timed_alias(): """Create a time-limited alias""" try: data = request.json username = data.get('username', '') domain = data.get('domain', '') if not username or not domain: return jsonify({'status': 'error', 'message': 'Both username and domain are required'}) malias_w.create_timed_alias(username, domain) result = f"Timed alias created for {username} on domain {domain}" return jsonify({'status': 'success', 'message': result}) except Exception as e: logger.exception("Error creating timed alias") return jsonify({'status': 'error', 'message': f"Error: {str(e)}"}) @app.route('/search', methods=['POST']) @login_required def search(): """Search for aliases""" query = request.json.get('query', '') if not query: return jsonify({'status': 'error', 'message': 'No search query provided'}) try: results = malias_w.search_aliases(query) if results: result_list = [f"{alias} → {goto}" for alias, goto in results] message = f"Found {len(results)} alias(es):\n" + "\n".join(result_list) else: message = f"No aliases found matching '{query}'" return jsonify({'status': 'success', 'message': message, 'query': query}) except Exception as e: logger.exception("Error searching aliases") return jsonify({'status': 'error', 'message': f"Search error: {str(e)}"}) @app.route('/config', methods=['GET', 'POST']) @login_required def config_page(): """Configuration management for Mailcow connection""" if request.method == 'POST': mailcow_server = request.json.get('mailcow_server', '') mailcow_api_key = request.json.get('mailcow_api_key', '') if mailcow_server and mailcow_api_key: malias_w.set_connection_info(mailcow_server, mailcow_api_key) return jsonify({'status': 'success', 'message': 'Mailcow configuration saved successfully'}) else: return jsonify({'status': 'error', 'message': 'Server and API key are required'}) # GET request - return current config try: current_config = malias_w.get_config() return jsonify(current_config) except Exception as e: return jsonify({'mailcow_server': '', 'mailcow_api_key': ''}) @app.route('/change_password', methods=['POST']) @login_required def change_password_route(): """Change password""" try: data = request.json old_password = data.get('old_password', '') new_password = data.get('new_password', '') confirm_password = data.get('confirm_password', '') if not old_password or not new_password or not confirm_password: return jsonify({'status': 'error', 'message': 'All fields are required'}) if new_password != confirm_password: return jsonify({'status': 'error', 'message': 'New passwords do not match'}) if len(new_password) < 6: return jsonify({'status': 'error', 'message': 'Password must be at least 6 characters'}) malias_w.change_password(old_password, new_password) return jsonify({'status': 'success', 'message': 'Password changed successfully'}) except Exception as e: logger.exception("Error changing password") return jsonify({'status': 'error', 'message': str(e)}) # Add a health check endpoint for proxies @app.route('/health') def health_check(): """Health check endpoint for proxies""" return jsonify({ 'status': 'ok', 'authenticated': session.get('logged_in', False), 'authelia_user': session.get('authelia_user', None), 'auth_method': session.get('auth_method'), 'version': '1.0.1' }) # Add a debugging endpoint @app.route('/debug') def debug_info(): """Show debug information about the current request""" # Always allow this in production to help with troubleshooting debug_data = { 'headers': dict(request.headers), 'cookies': {k: '***' for k in request.cookies.keys()}, # Don't expose cookie values 'session': {k: ('***' if k in ['user_token', 'csrf_token'] else v) for k, v in session.items()} if session else {}, 'remote_addr': request.remote_addr, 'scheme': request.scheme, 'host': request.host, 'path': request.path, 'is_secure': request.is_secure, 'x_forwarded_for': request.headers.get('X-Forwarded-For'), 'x_forwarded_proto': request.headers.get('X-Forwarded-Proto'), 'x_forwarded_host': request.headers.get('X-Forwarded-Host'), 'x_forwarded_prefix': request.headers.get('X-Forwarded-Prefix'), 'remote_user': get_authelia_user(), } # Additional server information debug_data['server_info'] = { 'cookie_settings': { 'SESSION_COOKIE_NAME': app.config['SESSION_COOKIE_NAME'], 'SESSION_COOKIE_SECURE': app.config['SESSION_COOKIE_SECURE'], 'SESSION_COOKIE_HTTPONLY': app.config['SESSION_COOKIE_HTTPONLY'], 'SESSION_COOKIE_SAMESITE': app.config['SESSION_COOKIE_SAMESITE'], 'SESSION_COOKIE_PATH': app.config['SESSION_COOKIE_PATH'], }, 'app_config': { 'DEBUG': app.debug, 'PREFERRED_URL_SCHEME': app.config['PREFERRED_URL_SCHEME'], } } response = jsonify(debug_data) response.headers['Cache-Control'] = 'no-store, must-revalidate' response.headers['Pragma'] = 'no-cache' response.headers['Expires'] = '0' return response # Dedicated endpoint for header diagnostics @app.route('/headers') def show_headers(): """Show all request headers - useful for debugging proxies""" return jsonify({ 'headers': dict(request.headers), 'remote_addr': request.remote_addr, 'authelia_user': get_authelia_user() }) if __name__ == '__main__': # Parse command-line arguments parser = argparse.ArgumentParser( description='Mailcow Alias Manager - Web interface for managing mail aliases', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=''' Examples: python3 app.py Run in normal mode on port 5172 python3 app.py --debug Run in debug mode with auto-reload python3 app.py --port 8080 Run on custom port 8080 python3 app.py --debug --port 8080 Run in debug mode on port 8080 python3 app.py --host 127.0.0.1 Bind to localhost only Environment Variables (Docker): FLASK_DEBUG Set to 'True' for debug mode (default: False) FLASK_PORT Port to run on (default: 5172) FLASK_HOST Host to bind to (default: 0.0.0.0) FLASK_ENV Set to 'production' or 'development' ''' ) parser.add_argument( '--debug', action='store_true', help='Enable debug mode with auto-reload and detailed error messages' ) parser.add_argument( '--port', type=int, default=None, help='Port to run on (default: 5172 or FLASK_PORT env var)' ) parser.add_argument( '--host', type=str, default=None, help='Host to bind to (default: 0.0.0.0 or FLASK_HOST env var)' ) args = parser.parse_args() # Priority: Command-line args > Environment variables > Defaults # Debug mode if args.debug: debug_mode = True else: debug_mode = os.getenv('FLASK_DEBUG', 'False').lower() in ('true', '1', 'yes') # Set log level based on debug mode logger.setLevel(logging.DEBUG if debug_mode else logging.INFO) # Port if args.port is not None: port = args.port else: port = int(os.getenv('FLASK_PORT', '5172')) # Host if args.host is not None: host = args.host else: host = os.getenv('FLASK_HOST', '0.0.0.0') # Print startup info print('=' * 60) print(' Mailcow Alias Manager - Starting...') print('=' * 60) print(f' Mode: {"DEBUG (Development)" if debug_mode else "NORMAL (Production)"}') print(f' Host: {host}') print(f' Port: {port}') print(f' URL: http://{host}:{port}') print('=' * 60) if debug_mode: print(' ⚠️ WARNING: Debug mode is enabled!') print(' - Auto-reload on code changes') print(' - Detailed error messages shown') print(' - NOT suitable for production use') print('=' * 60) print() logger.info(f"Starting Mailcow Alias Manager - Debug: {debug_mode}, Host: {host}, Port: {port}") app.run(debug=debug_mode, host=host, port=port)