from flask import Flask, request, Response, jsonify import jinja2 import re import html import time from functools import wraps from collections import defaultdict, deque app = Flask(__name__) env = jinja2.Environment( loader=jinja2.FileSystemLoader('templates'), autoescape=jinja2.select_autoescape(['xml']) ) # Security configuration ALLOWED_HOSTS = { 'autoconfig.mifi.holdings', 'autodiscover.mifi.holdings', 'autoconfig.mifi.com.br', 'autodiscover.mifi.com.br', 'autoconfig.mifi.dev', 'autodiscover.mifi.dev', 'autoconfig.mifi.ventures', 'autodiscover.mifi.ventures', 'autoconfig.mifi.vix.br', 'autodiscover.mifi.vix.br', 'autoconfig.mifi.me', 'autodiscover.mifi.me', 'autoconfig.blackice.vix.br', 'autodiscover.blackice.vix.br', 'autoconfig.fitz.guru', 'autodiscover.fitz.guru', 'autoconfig.umlautpress.com', 'autodiscover.umlautpress.com', 'autoconfig.camilla-rena.com', 'autodiscover.camilla-rena.com', 'autoconfig.officelift.net', 'autodiscover.officelift.net', 'autoconfig.mylocalpro.biz', 'autodiscover.mylocalpro.biz', 'autoconfig.mylocalpro.online', 'autodiscover.mylocalpro.online', 'autoconfig.happybeardedcarpenter.com', 'autodiscover.happybeardedcarpenter.com', 'autoconfig.thenewenglandpalletguy.com', 'autodiscover.thenewenglandpalletguy.com', 'autoconfig.dining-it.com', 'autodiscover.dining-it.com' } # Rate limiting storage (in production, use Redis or similar) # Simple in-memory rate limiting - use Redis in production rate_limit_storage = defaultdict(deque) RATE_LIMIT_REQUESTS = 100 # requests per window RATE_LIMIT_WINDOW = 3600 # 1 hour window def rate_limit(max_requests=RATE_LIMIT_REQUESTS, window=RATE_LIMIT_WINDOW): """Simple rate limiting decorator""" def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): # Get client IP client_ip = request.headers.get('X-Forwarded-For', request.remote_addr) if client_ip: client_ip = client_ip.split(',')[0].strip() current_time = time.time() # Clean old requests outside the window while (rate_limit_storage[client_ip] and rate_limit_storage[client_ip][0] < current_time - window): rate_limit_storage[client_ip].popleft() # Check if limit exceeded if len(rate_limit_storage[client_ip]) >= max_requests: return jsonify({'error': 'Rate limit exceeded'}), 429 # Add current request rate_limit_storage[client_ip].append(current_time) return f(*args, **kwargs) return decorated_function return decorator def validate_host(f): """Decorator to validate Host header""" @wraps(f) def decorated_function(*args, **kwargs): host = request.headers.get('Host', '').lower() # Remove port if present if ':' in host: host = host.split(':')[0] if host not in ALLOWED_HOSTS: return jsonify({'error': 'Forbidden host'}), 403 return f(*args, **kwargs) return decorated_function def validate_request_size(max_size=1024): """Decorator to validate request size""" def decorator(f): @wraps(f) def decorated_function(*args, **kwargs): content_length = request.content_length if content_length and content_length > max_size: return jsonify({'error': 'Request too large'}), 413 return f(*args, **kwargs) return decorated_function return decorator def add_security_headers(response): """Add security headers to all responses""" response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-Frame-Options'] = 'DENY' response.headers['X-XSS-Protection'] = '1; mode=block' response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' response.headers['Content-Security-Policy'] = "default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self'" return response def sanitize_domain(domain): """Sanitize domain to prevent injection attacks""" # Only allow alphanumeric, dots, and hyphens sanitized = re.sub(r'[^a-zA-Z0-9.-]', '', domain) # Prevent empty or invalid domains if not sanitized or len(sanitized) > 253: return None # Basic domain validation if not re.match(r'^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)*$', sanitized): return None return sanitized # Register security headers middleware @app.after_request def after_request(response): return add_security_headers(response) @app.route('/') @rate_limit(max_requests=50, window=3600) # 50 requests per hour for main page @validate_host @validate_request_size(max_size=512) def index(): host = request.headers.get('Host', '').lower() if ':' in host: host = host.split(':')[0] subdomain = host.split('.', 1)[0] if '.' in host else '' domain = host.split('.', 1)[1] if '.' in host else host # Sanitize domain to prevent injection sanitized_domain = sanitize_domain(domain) if not sanitized_domain: return jsonify({'error': 'Invalid domain'}), 400 base_html = """ {title}
{icon}
{service_type}

{title}

Domain: {domain}
{content} 📊 Service Status
""" if subdomain == 'autoconfig': return base_html.format( title="Mail Autoconfig Service", service_type="Mozilla Thunderbird", icon="🔧", domain=sanitized_domain, content=f'''
Thunderbird Autoconfig Endpoint:
https://{host}/mail/config-v1.1.xml
''' ) elif subdomain == 'autodiscover': return base_html.format( title="Mail Autodiscover Service", service_type="Microsoft Outlook", icon="🔍", domain=sanitized_domain, content=f'''
Outlook Autodiscover Endpoint:
https://{host}/Autodiscover/Autodiscover.xml
''' ) else: return base_html.format( title="Mail Configuration Service", service_type="Error", icon="❌", domain="Invalid", content='''
Invalid subdomain!
Please use autoconfig.domain.com or autodiscover.domain.com
''' ), 400 @app.route('/ping') @rate_limit(max_requests=10, window=60) # 10 requests per minute for health check def ping(): return "✅ Mail Autoconfig Service is running." @app.route('/mail/config-v1.1.xml') @rate_limit(max_requests=20, window=3600) # 20 requests per hour for config @validate_host @validate_request_size(max_size=1024) def thunderbird_config(): host = request.headers.get('Host', '').lower() if ':' in host: host = host.split(':')[0] domain = host.split('.', 1)[1] if '.' in host else host sanitized_domain = sanitize_domain(domain) if not sanitized_domain: return jsonify({'error': 'Invalid domain'}), 400 xml = env.get_template('config-v1.1.xml.j2').render(DOMAIN=sanitized_domain) return Response(xml, mimetype='application/xml') @app.route('/Autodiscover/Autodiscover.xml', methods=['POST','GET']) @rate_limit(max_requests=20, window=3600) # 20 requests per hour for autodiscover @validate_host @validate_request_size(max_size=2048) def outlook_autodiscover(): host = request.headers.get('Host', '').lower() if ':' in host: host = host.split(':')[0] domain = host.split('.', 1)[1] if '.' in host else host sanitized_domain = sanitize_domain(domain) if not sanitized_domain: return jsonify({'error': 'Invalid domain'}), 400 xml = env.get_template('Autodiscover.xml.j2').render(DOMAIN=sanitized_domain) return Response(xml, mimetype='text/xml')