Hypercorn Configuration
import asyncio
import logging
import os
import signal
import ssl
from datetime import datetime
from typing import Dict, List, Optional, Set, Tuple
from hypercorn.asyncio import serve
from hypercorn.config import Config
from hypercorn.middleware import DispatcherMiddleware, ProxyFixMiddleware
from hypercorn.typing import ASGIFramework
from utils.load_env import LOADENV
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(),
logging.FileHandler(
f"hypercorn_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"
),
],
)
logger = logging.getLogger("HypercornConfig")
LOADENV() # I have implemented my Own function to load ENV
class SecureHypercornConfig:
def __init__(
self,
app: ASGIFramework,
host: str = "0.0.0.0",
port: int = 8000,
http2: bool = True,
workers: Optional[int] = None,
ssl_enabled: bool = True,
ssl_keyfile: Optional[str] = None,
ssl_certfile: Optional[str] = None,
ssl_ca_certs: Optional[str] = None,
access_log: bool = True,
error_log: bool = True,
graceful_timeout: float = 30.0,
keep_alive_timeout: float = 75.0,
max_requests: int = 10000,
reload: bool = False,
max_requests_jitter: int = 1000,
backlog: int = 2048,
max_app_queue_size: int = 100,
include_server_header: bool = False,
include_date_header: bool = True,
root_path: str = "",
websocket_ping_interval: Optional[float] = 20.0,
websocket_max_message_size: int = 16 * 1024 * 1024,
timeout: float = 30.0,
startup_timeout: float = 60.0,
shutdown_timeout: float = 60.0,
log_level: str = "info",
log_format: str = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s"',
# ssl.CERT_REQUIRED (value 2) forces the client (e.g., Postman) to present a certificate,
# which it doesn't do by default, causing the handshake to fail.
cert_reqs: int = ssl.CERT_NONE,
pid_file: Optional[str] = None,
umask: Optional[int] = None,
user: Optional[int] = None,
group: Optional[int] = None,
bind_to_multiple: Optional[List[str]] = None,
additional_headers: Optional[List[Tuple[str, str]]] = None,
behind_proxy: bool = False,
proxy_mode: str = "legacy",
trusted_proxies: Optional[List[str]] = None,
trusted_hops: int = 1,
mount_points: Optional[Dict[str, ASGIFramework]] = None,
server_names: Optional[List[str]] = None,
hsts_max_age: int = 63072000,
hsts_include_subdomains: bool = True,
hsts_preload: bool = True,
):
# Application configuration
self.app = app
self.host = host
self.port = port
self.http2 = http2
# Worker configuration
self.workers = workers or max(1, (os.cpu_count() or 1) * 2 + 1)
# SSL/TLS configuration
self.ssl_enabled = ssl_enabled
self.ssl_keyfile = ssl_keyfile
self.ssl_certfile = ssl_certfile
self.ssl_ca_certs = ssl_ca_certs
self.cert_reqs = cert_reqs
# Logging configuration
self.access_log = access_log
self.error_log = error_log
self.log_level = log_level
self.log_format = log_format
# Performance tuning
self.graceful_timeout = graceful_timeout
self.keep_alive_timeout = keep_alive_timeout
self.max_requests = max_requests
self.max_requests_jitter = max_requests_jitter
self.backlog = backlog
self.max_app_queue_size = max_app_queue_size
self.timeout = timeout
self.startup_timeout = startup_timeout
self.shutdown_timeout = shutdown_timeout
# Security headers
self.include_server_header = include_server_header
self.include_date_header = include_date_header
self.root_path = root_path
# WebSocket settings
self.websocket_ping_interval = websocket_ping_interval
self.websocket_max_message_size = websocket_max_message_size
# Process management
self.reload = reload
self.pid_file = pid_file
self.umask = umask
self.user = user
self.group = group
# Multiple binding support
self.bind_to_multiple = bind_to_multiple or [f"{host}:{port}"]
# Proxy configuration
self.behind_proxy = behind_proxy
self.proxy_mode = proxy_mode
self.trusted_proxies = trusted_proxies or []
self.trusted_hops = trusted_hops
# Mount points for multiple applications
self.mount_points = mount_points
# Server names configuration for DNS rebinding protection
self.server_names: Optional[Set[str]] = (
set(server_names) if server_names else None
)
# HSTS configuration
self.hsts_max_age = hsts_max_age
self.hsts_include_subdomains = hsts_include_subdomains
self.hsts_preload = hsts_preload
# Additional headers
self.additional_headers = additional_headers or []
# Initialize shutdown event
self.shutdown_event = asyncio.Event()
# Wrap application with appropriate middleware
self.app = self._wrap_application(app)
def _wrap_application(self, app: ASGIFramework) -> ASGIFramework:
"""Wrap application with appropriate middleware"""
wrapped_app = app
# Add proxy middleware if behind proxy
if self.behind_proxy:
wrapped_app = self._wrap_with_proxy_middleware(wrapped_app)
# Add dispatcher middleware if mount points exist
if self.mount_points:
wrapped_app = self._wrap_with_dispatcher_middleware(wrapped_app)
return wrapped_app
def _wrap_with_proxy_middleware(self, app: ASGIFramework) -> ASGIFramework:
"""Wrap application with appropriate proxy middleware"""
logger.info(f"Configuring proxy middleware with mode: {self.proxy_mode}")
logger.info(f"Trusted hops: {self.trusted_hops}")
if self.trusted_proxies:
logger.info(f"Trusted proxies: {self.trusted_proxies}")
return ProxyFixMiddleware(
app,
# Remember: "legacy" for reddis "modren" for other e.g Tarifik
mode=self.proxy_mode, # type: ignore
trusted_hops=self.trusted_hops,
)
def _wrap_with_dispatcher_middleware(self, app: ASGIFramework) -> ASGIFramework:
"""Wrap application with dispatcher middleware for multiple mount points"""
if self.mount_points:
logger.info(
f"Configuring dispatcher with mount points: {list(self.mount_points.keys())}"
)
return DispatcherMiddleware(self.mount_points)
return app
def _get_hsts_header_value(self) -> str:
"""Generate HSTS header value based on configuration"""
hsts_parts = [f"max-age={self.hsts_max_age}"]
if self.hsts_include_subdomains:
hsts_parts.append("includeSubDomains")
if self.hsts_preload:
hsts_parts.append("preload")
return "; ".join(hsts_parts)
def createConfig(self) -> Config:
"""Create and configure Hypercorn Config with security settings"""
config = Config()
# Set number of workers
config.workers = self.workers
# Basic configuration
config.bind = self.bind_to_multiple
config.alpn_protocols = ["h2", "http/1.1"] if self.http2 else ["http/1.1"]
# SSL/TLS configuration
if self.ssl_enabled and self.ssl_keyfile and self.ssl_certfile:
if os.path.isfile(self.ssl_keyfile):
config.keyfile = self.ssl_keyfile
else:
logger.error(f"SSL keyfile not found: {self.ssl_keyfile}")
if os.path.isfile(self.ssl_certfile):
config.certfile = self.ssl_certfile
else:
logger.error(f"SSL certfile not found: {self.ssl_certfile}")
if self.ssl_ca_certs and os.path.isfile(self.ssl_ca_certs):
config.ca_certs = self.ssl_ca_certs
elif self.ssl_ca_certs:
logger.warning(f"SSL CA certs file not found: {self.ssl_ca_certs}")
config.cert_reqs = self.cert_reqs
# Modern, secure cipher suite
config.ciphers = "ECDHE-ECDSA-AES256-GCM-SHA384:ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305:ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256"
# DNS rebinding protection
if self.server_names:
config.server_names = list(self.server_names)
logger.info(f"Server names whitelist: {config.server_names}")
else:
logger.warning(
"Server names not configured - DNS rebinding protection is disabled"
)
# Logging configuration
config.accesslog = (
logging.getLogger("hypercorn.access") if self.access_log else None
)
config.errorlog = (
logging.getLogger("hypercorn.error") if self.error_log else None
)
config.loglevel = self.log_level
config.access_log_format = self.log_format
# Performance tuning
config.graceful_timeout = self.graceful_timeout
config.keep_alive_timeout = self.keep_alive_timeout
config.max_requests = self.max_requests
config.max_requests_jitter = self.max_requests_jitter
config.backlog = self.backlog
config.max_app_queue_size = self.max_app_queue_size
config.startup_timeout = self.startup_timeout
config.shutdown_timeout = self.shutdown_timeout
# Security headers
config.include_server_header = self.include_server_header
config.include_date_header = self.include_date_header
config.root_path = self.root_path
# WebSocket configuration
config.websocket_ping_interval = self.websocket_ping_interval
config.websocket_max_message_size = self.websocket_max_message_size
# Process management
config.pid_path = self.pid_file
config.umask = self.umask
config.user = self.user
config.group = self.group
config.use_reloader = self.reload
return config
def get_security_headers(self) -> List[Tuple[str, str]]:
"""Return security headers for middleware integration"""
headers: List[Tuple[str, str]] = [
("X-Frame-Options", "DENY"),
("X-Content-Type-Options", "nosniff"),
("X-XSS-Protection", "1; mode=block"),
(
"Content-Security-Policy",
"default-src 'self'; script-src 'self' 'unsafe-inline'",
),
("Referrer-Policy", "strict-origin-when-cross-origin"),
("Permissions-Policy", "camera=(), microphone=(), geolocation=()"),
("Cross-Origin-Embedder-Policy", "require-corp"),
("Cross-Origin-Opener-Policy", "same-origin"),
("Cross-Origin-Resource-Policy", "same-origin"),
]
if self.ssl_enabled:
hsts_value = self._get_hsts_header_value()
headers.append(("Strict-Transport-Security", hsts_value))
if self.additional_headers:
headers.extend(self.additional_headers)
return headers
async def run(self) -> None:
"""Run the Hypercorn server with the configured settings"""
config = self.createConfig()
logger.info(f"Starting Hypercorn server on {', '.join(self.bind_to_multiple)}")
logger.info(f"Workers: {self.workers}")
logger.info(f"HTTP/2: {self.http2}")
logger.info(f"SSL Enabled: {self.ssl_enabled}")
logger.info(f"Log Level: {self.log_level}")
# Setup signal handlers for graceful shutdown
loop = asyncio.get_running_loop()
for sig in (signal.SIGINT, signal.SIGTERM):
try:
loop.add_signal_handler(
sig, lambda s=sig: asyncio.create_task(self.shutdown(s))
)
except (NotImplementedError, ValueError):
# logger.warning(
# f"Signal handler for {sig.name} not supported on this platform."
# )
logger.warning("Signal handler for not supported on this platform.")
try:
await serve(self.app, config, shutdown_trigger=self.shutdown_event.wait)
except Exception as e:
logger.error(f"Server error: {e}", exc_info=True)
raise
finally:
await self.cleanup()
async def shutdown(self, sig: Optional[signal.Signals] = None) -> None:
"""Initiate graceful shutdown"""
if not self.shutdown_event.is_set():
if sig:
logger.info(
f"Received signal {sig.name}, initiating graceful shutdown..."
)
else:
logger.info("Initiating graceful shutdown...")
self.shutdown_event.set()
async def cleanup(self) -> None:
"""Cleanup resources before exit"""
logger.info("Server has shut down.")
# Factory function for production RUN
def prodHypercornConfiguration(
app: ASGIFramework,
host: Optional[str] = None,
port: Optional[int] = None,
workers: Optional[int] = None,
bind_to_multiple: Optional[List[str]] = None,
) -> SecureHypercornConfig:
# Parse trusted proxies from environment
trusted_proxies: Optional[List[str]] = None
trusted_proxies_env = os.getenv("TRUSTED_PROXIES")
if trusted_proxies_env:
trusted_proxies = [proxy.strip() for proxy in trusted_proxies_env.split(",")]
# Parse server names from environment
server_names: Optional[List[str]] = None
server_names_env = os.getenv("SERVER_NAMES")
if server_names_env:
server_names = [name.strip() for name in server_names_env.split(",")]
elif os.getenv("DOMAIN"): # Backward compatibility
domain_env = os.getenv("DOMAIN")
if domain_env:
server_names = [domain_env.strip()]
# Handle workers parameter
workers_value: Optional[int] = None
if workers is not None:
workers_value = workers
else:
workers_env = os.getenv("WORKERS", "0")
if workers_env and workers_env != "0":
workers_value = int(workers_env)
# Handle umask
umask_value: Optional[int] = None
umask_env = os.getenv("UMASK")
if umask_env:
umask_value = int(umask_env)
# Handle user
user_value: Optional[int] = None
user_env = os.getenv("USER_ID")
if user_env:
user_value = int(user_env)
# Handle group
group_value: Optional[int] = None
group_env = os.getenv("GROUP_ID")
if group_env:
group_value = int(group_env)
return SecureHypercornConfig(
app=app,
host=host or os.getenv("HOST", "0.0.0.0"),
port=port or int(os.getenv("PORT", "8000")),
workers=workers_value,
http2=os.getenv("HTTP2", "true").lower() == "true",
ssl_enabled=os.getenv("SSL_ENABLED", "true").lower() == "true",
# Path to your certs where you created
ssl_keyfile=os.getenv("SSL_KEYFILE") or "./certs/localhost+1-key.pem",
ssl_certfile=os.getenv("SSL_CERTFILE") or "./certs/localhost+1.pem",
ssl_ca_certs=os.getenv("SSL_CA_CERTS")
or None,
access_log=os.getenv("ACCESS_LOG", "false").lower() == "true",
error_log=os.getenv("ERROR_LOG", "true").lower() == "true",
max_requests=int(os.getenv("MAX_REQUESTS", "10000")),
max_requests_jitter=int(os.getenv("MAX_REQUESTS_JITTER", "1000")),
reload=os.getenv("RELOAD", "false").lower() == "true",
log_level=os.getenv("LOG_LEVEL", "info"),
include_server_header=os.getenv("INCLUDE_SERVER_HEADER", "false").lower()
== "true",
timeout=float(os.getenv("TIMEOUT", "30.0")),
startup_timeout=float(os.getenv("STARTUP_TIMEOUT", "60.0")),
shutdown_timeout=float(os.getenv("SHUTDOWN_TIMEOUT", "60.0")),
bind_to_multiple=bind_to_multiple
or [b.strip() for b in os.getenv("BIND", "").split(",") if b.strip()]
or None,
pid_file=os.getenv("PID_FILE")
or None,
umask=umask_value,
user=user_value,
group=group_value,
behind_proxy=os.getenv("BEHIND_PROXY", "false").lower() == "true",
proxy_mode=os.getenv("PROXY_MODE", "legacy"),
trusted_proxies=trusted_proxies,
trusted_hops=int(os.getenv("TRUSTED_HOPS", "1")),
server_names=server_names,
hsts_max_age=int(os.getenv("HSTS_MAX_AGE", "63072000")),
hsts_include_subdomains=os.getenv("HSTS_INCLUDE_SUBDOMAINS", "true").lower()
== "true",
hsts_preload=os.getenv("HSTS_PRELOAD", "true").lower() == "true",
)
# Factory function for development RUN - for testing
def devHypercornConfiguration(
app: ASGIFramework,
host: str = "127.0.0.1",
port: int = 8000,
reload: bool = True,
) -> SecureHypercornConfig:
return SecureHypercornConfig(
app=app,
host=host,
port=port,
ssl_enabled=True,
reload=reload,
access_log=True,
include_server_header=True,
log_level="debug",
workers=1,
server_names=["localhost", "127.0.0.1", "[::1]"],
behind_proxy=False,
)
Main File
async def main():
# NOTE: Ensure the certificate paths are correct relative to where you run the script.
# You might need to generate them first with:
# mkcert -install
# mkcert localhost 127.0.0.1 ::1
# Then move the generated files to a './certs' directory.
# Create directories if they don't exist
if not os.path.exists("./certs"):
os.makedirs("./certs")
print(
"Created './certs' directory. Please add your mkcert key and cert files there."
)
print("Example: './certs/localhost+2-key.pem' and './certs/localhost+2.pem'")
# For demonstration, we'll point to where mkcert commonly places files
# but you should adjust these paths to your actual file locations.
keyfile = "./certs/localhost+1-key.pem" # Adjust if your filename is different
certfile = "./certs/localhost+1.pem" # Adjust if your filename is different
config = SecureHypercornConfig(
app=cast(ASGIFramework, app),
host="127.0.0.1",
port=8000,
ssl_enabled=True,
ssl_keyfile=keyfile,
ssl_certfile=certfile,
http2=True,
reload=True,
access_log=True,
log_level="info",
workers=4,
)
print(f"Starting server with HTTPS/2 on https://{config.host}:{config.port}")
print("Press Ctrl+C to stop")
await config.run()
if __name__ == "__main__":
print("Starting FastAPI app with secure Hypercorn configuration...")
try:
asyncio.run(main())
except KeyboardInterrupt:
print("\nServer Stopped by User.")
except Exception as e:
print(f"An error occurred: {e}")
For Production
import asyncio
import os
from typing import cast
from hypercorn.typing import ASGIFramework
from app import app
from utils.hypercorn_config import SecureHypercornConfig
async def main():
# --- PRODUCTION CONFIGURATION ---
# In production, get settings from environment variables for security and flexibility.
# Your domain should be set in an environment variable.
domain = os.getenv("DOMAIN_NAME", "api.yourdomain.com")
# Let's Encrypt certificate paths.
# These are the standard locations after running Certbot.
keyfile = f"/etc/letsencrypt/live/{domain}/privkey.pem"
certfile = f"/etc/letsencrypt/live/{domain}/fullchain.pem"
# Check if the certificate files exist before starting
if not os.path.exists(keyfile) or not os.path.exists(certfile):
print(f"ERROR: SSL certificate files not found for domain {domain}.")
print(f"Expected keyfile: {keyfile}")
print(f"Expected certfile: {certfile}")
print("Please ensure you have run Certbot to generate certificates.")
return # Exit if certs are not found
config = SecureHypercornConfig(
app=cast(ASGIFramework, app),
host=os.getenv("HOST", "0.0.0.0"), # Listen on all interfaces in production
port=int(os.getenv("PORT", "8000")),
ssl_enabled=True,
ssl_keyfile=keyfile,
ssl_certfile=certfile,
http2=True,
reload=False, # Reload should be disabled in production
access_log=True,
log_level="info",
workers=int(os.getenv("WORKERS", "4")),
server_names=[domain], # Important for security (DNS rebinding)
)
print(
f"Starting server with HTTPS/2 on https://{config.host}:{config.port} for domain {domain}"
)
print("This is a production server. Ensure your DNS is configured correctly.")
print("Press Ctrl+C to stop")
await config.run()
if __name__ == "__main__":
print("Starting FastAPI app with secure Hypercorn configuration...")
# You would run this script on your server, e.g., using a process manager like systemd.
# Example environment variable setup before running:
# export DOMAIN_NAME="api.your-actual-domain.com"
# python main.py
try:
asyncio.run(main())
except PermissionError:
print("\nERROR: Permission denied. You might need to run this script with sudo")
print("to access the Let's Encrypt certificate files in /etc/letsencrypt/.")
except KeyboardInterrupt:
print("\nServer Stopped by User.")
except Exception as e:
print(f"An error occurred: {e}")
Script
This should be based on your server
#!/bin/bash
# ==============================================================================
# Script to Obtain Let's Encrypt SSL Certificates for a Production Server
# ==============================================================================
#
# This script automates the process of installing Certbot and obtaining a
# certificate using the standalone method. This method requires port 80 to be
# temporarily free.
#
# USAGE:
# 1. Make the script executable: chmod +x setup_prod_certs.sh
# 2. Run the script with your domain name and email address:
# sudo ./setup_prod_certs.sh your-api.yourdomain.com your-email@yourdomain.com
#
# ==============================================================================
# --- Configuration ---
# Exit immediately if a command exits with a non-zero status.
set -e
# --- Input Validation ---
if [ "$#" -ne 2 ]; then
echo "ERROR: Missing arguments."
echo "Usage: $0 <YOUR_DOMAIN> <YOUR_EMAIL>"
exit 1
fi
DOMAIN_NAME="$1"
EMAIL_ADDRESS="$2"
echo "------------------------------------------------"
echo "Starting SSL Certificate Setup for: $DOMAIN_NAME"
echo "------------------------------------------------"
# --- Step 1: Update System and Install Certbot ---
echo "[Step 1/5] Updating package lists and installing Certbot..."
sudo apt-get update
sudo apt-get install certbot -y
echo "Certbot installation complete."
echo ""
# --- Step 2: Stop any running web server ---
echo "[Step 2/5] Checking port 80..."
# The standalone method requires port 80 to be free.
# If you have a service (like Nginx, Apache, or your own app) running, it must be stopped first.
# The command `lsof` checks if the port is in use.
if sudo lsof -i :80 -t >/dev/null; then
echo "A service is currently running on port 80. Attempting to stop it."
# --- IMPORTANT ---
# Replace 'nginx' with the name of your service if it's not Nginx.
# If your Python app is running directly on port 80, you would stop it here.
# For example: sudo systemctl stop your-python-app.service
sudo systemctl stop nginx || echo "Could not stop nginx, it might not be running. Continuing..."
sleep 5 # Give the service time to shut down
else
echo "Port 80 is free. Continuing..."
fi
echo ""
# --- Step 3: Obtain the SSL Certificate ---
echo "[Step 3/5] Requesting a certificate from Let's Encrypt..."
sudo certbot certonly \
--standalone \
--non-interactive \
--agree-tos \
-m "$EMAIL_ADDRESS" \
-d "$DOMAIN_NAME"
echo "Certificate obtained successfully!"
echo ""
# --- Step 4: Verify Automatic Renewal ---
echo "[Step 4/5] Verifying Certbot's auto-renewal process..."
# Certbot automatically creates a systemd timer or cron job for renewal.
# This dry run simulates renewal without actually changing certificates.
sudo certbot renew --dry-run
echo "Auto-renewal is configured correctly."
echo ""
# --- Step 5: Final Instructions ---
echo "[Step 5/5] Setup Complete!"
echo ""
echo "------------------------------------------------------------------"
echo "SUCCESS! Your SSL certificate is ready."
echo "------------------------------------------------------------------"
echo ""
echo "The certificate files are located in:"
echo "/etc/letsencrypt/live/$DOMAIN_NAME/"
echo ""
echo "To run your Python application from the Canvas, use the following commands on your server:"
echo ""
echo " export DOMAIN_NAME=\"$DOMAIN_NAME\""
echo " export WORKERS=4 # Or your desired number of workers"
echo " export PORT=443 # Standard HTTPS port"
echo ""
echo " sudo python3 main.py"
echo ""
echo "You need 'sudo' to bind to port 443 and to access the certificate files."
echo "Remember to restart your web service if you stopped one in Step 2."
echo "For example: sudo systemctl start nginx"
RUN
chmod +x cert.sh
3. **Run the Script:** Execute the script with `sudo`, providing your actual domain name and email address as arguments.
sudo ./setup_prod_certs.sh api.your-actual-domain.com <your-email@provider.com>
Top comments (0)