469 lines
17 KiB
Python
469 lines
17 KiB
Python
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
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.getenv('SECRET_KEY', 'malias-default-secret-key-please-change') # Consistent secret key
|
|
|
|
# Configure for reverse proxy (Authelia, Zoraxy, Nginx, etc.)
|
|
# Use higher number of proxies to accommodate both Zoraxy and Authelia
|
|
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',
|
|
APPLICATION_ROOT='/',
|
|
)
|
|
|
|
def login_required(f):
|
|
"""Decorator to require login for routes"""
|
|
@wraps(f)
|
|
def decorated_function(*args, **kwargs):
|
|
# Check for Authelia headers first
|
|
remote_user = request.headers.get('Remote-User') or request.headers.get('X-Remote-User')
|
|
remote_groups = request.headers.get('Remote-Groups') or request.headers.get('X-Remote-Groups')
|
|
|
|
# If Authelia authenticated the user, trust it and bypass local auth
|
|
if remote_user:
|
|
if not session.get('logged_in'):
|
|
session.clear()
|
|
session.permanent = True
|
|
session['logged_in'] = True
|
|
session['authelia_user'] = remote_user
|
|
session['user_token'] = secrets.token_urlsafe(32)
|
|
session.modified = True
|
|
return f(*args, **kwargs)
|
|
|
|
# Otherwise, fall back to local auth
|
|
if not session.get('logged_in'):
|
|
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"""
|
|
# Check for Authelia authentication
|
|
remote_user = request.headers.get('Remote-User') or request.headers.get('X-Remote-User')
|
|
|
|
# If Authelia authenticated, redirect to index
|
|
if remote_user:
|
|
session.clear()
|
|
session.permanent = True
|
|
session['logged_in'] = True
|
|
session['authelia_user'] = remote_user
|
|
session['user_token'] = secrets.token_urlsafe(32)
|
|
session.modified = True
|
|
return redirect(url_for('index'))
|
|
|
|
if request.method == 'POST':
|
|
password = request.json.get('password', '')
|
|
if malias_w.verify_password(password):
|
|
session.clear()
|
|
session.permanent = True
|
|
session['logged_in'] = True
|
|
session['user_token'] = secrets.token_urlsafe(32)
|
|
session.modified = True
|
|
|
|
response = jsonify({'status': 'success', 'message': 'Login successful'})
|
|
# Ensure cookies are properly configured for proxied setup
|
|
return response
|
|
else:
|
|
return jsonify({'status': 'error', 'message': 'Invalid password'})
|
|
|
|
if session.get('logged_in'):
|
|
return redirect(url_for('index'))
|
|
|
|
return render_template('login.html')
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
"""Logout"""
|
|
session.pop('logged_in', None)
|
|
session.pop('authelia_user', None)
|
|
|
|
# Determine if we need to redirect to Authelia logout
|
|
remote_user = request.headers.get('Remote-User') or request.headers.get('X-Remote-User')
|
|
|
|
if remote_user:
|
|
# Redirect to Authelia logout if available
|
|
authelia_url = request.headers.get('X-Authelia-URL') or None
|
|
if authelia_url:
|
|
return redirect(f"{authelia_url}/logout")
|
|
|
|
return redirect(url_for('login'))
|
|
|
|
@app.route('/')
|
|
def index():
|
|
"""Main page - requires login"""
|
|
# Check Authelia authentication
|
|
remote_user = request.headers.get('Remote-User') or request.headers.get('X-Remote-User')
|
|
|
|
if remote_user and not session.get('logged_in'):
|
|
# Auto-login users authenticated by Authelia
|
|
session.clear()
|
|
session.permanent = True
|
|
session['logged_in'] = True
|
|
session['authelia_user'] = remote_user
|
|
session['user_token'] = secrets.token_urlsafe(32)
|
|
session.modified = True
|
|
|
|
if not session.get('logged_in'):
|
|
return redirect(url_for('login'))
|
|
|
|
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:
|
|
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:
|
|
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:
|
|
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:
|
|
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:
|
|
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:
|
|
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:
|
|
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:
|
|
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:
|
|
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),
|
|
'version': '1.0.0'
|
|
})
|
|
|
|
# Add a debugging endpoint (only active in debug mode)
|
|
@app.route('/debug')
|
|
def debug_info():
|
|
"""Show debug information about the current request"""
|
|
if not app.debug:
|
|
return jsonify({'status': 'error', 'message': 'Debug mode not enabled'}), 403
|
|
|
|
debug_data = {
|
|
'headers': dict(request.headers),
|
|
'cookies': dict(request.cookies),
|
|
'session': dict(session) 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': request.headers.get('Remote-User') or request.headers.get('X-Remote-User'),
|
|
'remote_groups': request.headers.get('Remote-Groups') or request.headers.get('X-Remote-Groups'),
|
|
}
|
|
|
|
return jsonify(debug_data)
|
|
|
|
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')
|
|
|
|
# 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()
|
|
|
|
app.run(debug=debug_mode, host=host, port=port) |