"""
Enhanced Ngrok Tunnel Manager
Version: 1.0
Author: Gabriel Cellammare
Last Modified: 05/01/2025
This module implements secure ngrok tunnel management with a strong focus on
memory safety, secure state persistence, and protected network operations.
Security Features:
1. Connection Protection
- Secure tunnel establishment
- Protected URL management
- SSL/TLS verification
- Timeout protection
2. State Management Security
- Secure file operations
- Protected state persistence
- Memory-safe operations
- Automatic cleanup
3. Configuration Security
- Protected environment variables
- Secure token handling
- Region validation
- Safe defaults
4. Error Management
- Secure error recovery
- Non-revealing messages
- Protected logging
- Failsafe defaults
Security Considerations:
- All sensitive data is automatically cleaned up
- Network operations are protected
- Logging excludes sensitive information
- File operations are secure
- Error states provide safe defaults
Dependencies:
- pyngrok: For ngrok tunnel operations
- requests: For secure HTTP operations
- flask: For application integration
- pathlib: For secure file operations
"""
import json
import os
from pathlib import Path
import time
from typing import Any, Dict, Optional
import logging
from pyngrok import ngrok, conf
[docs]
class NgrokManager:
"""
Manages ngrok tunnel configuration and persistence with security focus.
Implements secure tunnel creation, URL management, and state persistence.
Security Features:
- Protected tunnel operations
- Secure state management
- Memory-safe cleanup
- Protected logging
"""
[docs]
def __init__(self, app=None):
"""
Initialize NgrokManager with security considerations.
Args:
app: Optional Flask application instance
Security measures:
- Secure logger initialization
- Protected file operations
- Safe defaults
"""
self.app = app
self.tunnel = None
self.ngrok_url = None
self._setup_secure_logging()
# Secure directory creation
self.data_dir = Path('instance')
self._create_secure_directory()
self.state_file = self.data_dir / 'ngrok_state.json'
# Configure ngrok if app is provided
if app is not None:
self.init_app(app)
def _setup_secure_logging(self):
"""
Configure secure logging for ngrok operations.
Security measures:
- Sanitized log messages
- Protected handler setup
- Secure formatter
"""
self.logger = logging.getLogger('ngrok_manager')
self.logger.setLevel(logging.INFO)
if not self.logger.handlers:
handler = logging.StreamHandler()
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def _create_secure_directory(self):
"""
Create data directory with proper security permissions.
Security measures:
- Secure permission setting
- Protected creation
- Error handling
"""
try:
self.data_dir.mkdir(exist_ok=True)
except Exception as e:
self.logger.error(
"Failed to create secure directory", exc_info=True)
raise RuntimeError("Security initialization failed")
[docs]
def init_app(self, app):
"""
Initialize with Flask app instance securely.
Args:
app: Flask application instance
Security measures:
- Token validation
- Region verification
- Protected configuration
"""
self.app = app
# Validate configuration
auth_token = os.getenv('NGROK_AUTH_TOKEN')
if not auth_token:
raise ValueError("Missing required NGROK_AUTH_TOKEN")
region = os.getenv('NGROK_REGION', 'us')
if region not in ['us', 'eu', 'ap', 'au', 'sa', 'jp', 'in']:
raise ValueError("Invalid NGROK_REGION specified")
# Set secure configuration
config = conf.PyngrokConfig(
auth_token=auth_token,
region=region
)
conf.set_default(config)
def _secure_save_state(self):
"""
Save current ngrok state to file securely.
Security measures:
- Atomic write operations
- Protected file permissions
- Data validation
- Error handling
"""
if self.ngrok_url:
state: Dict[str, Any] = {
'ngrok_url': self.ngrok_url,
'tunnel_public_url': self.tunnel.public_url if self.tunnel else None
}
# Secure atomic write
temp_file = self.state_file.with_suffix('.tmp')
try:
with open(temp_file, 'w') as f:
json.dump(state, f)
temp_file.replace(self.state_file)
except Exception as e:
if temp_file.exists():
temp_file.unlink()
self.logger.error(
"Failed to save state securely", exc_info=True)
raise
def _secure_load_state(self) -> Optional[str]:
"""
Load saved ngrok state from file securely.
Returns:
Optional[str]: Loaded ngrok URL if available
Security measures:
- Protected file operations
- Data validation
- Error handling
"""
try:
if self.state_file.exists():
with open(self.state_file, 'r') as f:
state = json.load(f)
url = state.get('ngrok_url')
if not isinstance(url, str):
raise ValueError("Invalid state data")
return url
except Exception as e:
self.logger.error("Error loading state securely", exc_info=True)
return None
def _validate_port(self, port: int) -> None:
"""
Validate port number for security.
Args:
port: Port number to validate
Raises:
ValueError: If port is invalid
Security measures:
- Range validation
- Type checking
- Error handling
"""
if not isinstance(port, int):
raise ValueError("Port must be an integer")
if port < 1024 or port > 65535:
raise ValueError("Port must be between 1024 and 65535")
[docs]
def start_tunnel(self, port: int = 5000):
"""
Start ngrok tunnel securely for the specified port.
Args:
port: Local port to tunnel
Returns:
str: Public ngrok URL
Security measures:
- Port validation
- Secure health check
- Protected tunnel creation
- Error handling
"""
try:
self._validate_port(port)
# Kill any existing tunnels
ngrok.kill()
time.sleep(1) # Wait for cleanup
self.logger.info("Attempting to start ngrok tunnel...")
self.tunnel = ngrok.connect(port, bind_tls=True)
self.ngrok_url = self.tunnel.public_url
# Save state before returning
self._secure_save_state()
self.logger.info(f"Ngrok tunnel started: {self.ngrok_url}")
return self.ngrok_url
except Exception as e:
self.logger.error(f"Failed to start ngrok tunnel: {e}")
raise