#!/usr/bin/env python3 """ HTTP Tunnel over SSH using Paramiko with improved monitoring and reliability This script establishes an HTTP tunnel over SSH using the Paramiko library with additional monitoring capabilities and automatic reconnection. """ import socket import select import threading import logging import paramiko import time class SSHTunnel: """ A class that establishes an HTTP tunnel over SSH using Paramiko with improved monitoring. This creates a secure tunnel that forwards traffic from a local port to a remote HTTP port through an SSH connection and automatically reconnects if needed. """ def __init__(self, ssh_host, remote_port=80, local_port=80, ssh_port=22, username=None, password=None, key_filename=None, reconnect_interval=30, keep_alive_interval=15): """ Initialize the SSH tunnel with the given parameters. Args: ssh_host (str): The SSH server hostname or IP address remote_port (int): The remote HTTP port to forward to (default: 80) local_port (int): The local port to listen on (default: 80) ssh_port (int): The SSH server port (default: 22) username (str): The SSH username password (str): The SSH password (optional if using key_filename) key_filename (str): Path to private key file (optional if using password) reconnect_interval (int): Seconds to wait before reconnecting (default: 30) keep_alive_interval (int): Seconds between keep-alive packets (default: 15) """ self.ssh_host = ssh_host self.remote_port = remote_port self.local_port = local_port self.ssh_port = ssh_port self.username = username self.password = password self.key_filename = key_filename self.reconnect_interval = reconnect_interval self.keep_alive_interval = keep_alive_interval self.server_socket = None self.transport = None self.client = None self.is_running = False self.should_reconnect = True self.last_activity_time = time.time() self.connection_status = "disconnected" self.connection_error = None self.threads = [] self.last_check_time = 0 self.check_interval = 10 # Check tunnel every 10 seconds # Set up logging self.logger = logging.getLogger('ssh_tunnel') handler = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) self.logger.addHandler(handler) self.logger.setLevel(logging.INFO) def start(self): """ Start the SSH tunnel with monitoring threads. This method establishes the SSH connection and starts listening for incoming connections on the local port, with additional threads for monitoring. Returns: bool: True if the tunnel was started successfully, False otherwise. """ if self.is_running: self.logger.warning("Tunnel is already running") return True self.should_reconnect = True self.connect() # Start the monitoring thread monitor_thread = threading.Thread(target=self._monitor_connection) monitor_thread.daemon = True monitor_thread.start() self.threads.append(monitor_thread) # Start the keep-alive thread keepalive_thread = threading.Thread(target=self._send_keepalive) keepalive_thread.daemon = True keepalive_thread.start() self.threads.append(keepalive_thread) return self.is_running def connect(self): """ Establish the SSH connection and start the server socket. Returns: bool: True if connection was successful, False otherwise. """ try: # Create an SSH client self.client = paramiko.SSHClient() self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # Connect to the SSH server self.logger.info(f"Connecting to SSH server {self.ssh_host}:{self.ssh_port}") connect_kwargs = { 'hostname': self.ssh_host, 'port': self.ssh_port, 'username': self.username, } if self.password: connect_kwargs['password'] = self.password if self.key_filename: connect_kwargs['key_filename'] = self.key_filename self.client.connect(**connect_kwargs) # Get the transport layer self.transport = self.client.get_transport() self.transport.set_keepalive(self.keep_alive_interval) # Start a server socket to listen for incoming connections self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try: self.server_socket.bind(('', self.local_port)) self.server_socket.listen(5) except OSError as e: self.logger.error(f"Could not bind to port {self.local_port}: {str(e)}") self.client.close() return False self.is_running = True self.connection_status = "connected" self.connection_error = None self.last_activity_time = time.time() self.logger.info(f"SSH tunnel established. Forwarding local port {self.local_port} " f"to remote port {self.remote_port} via {self.ssh_host}") # Start the main thread to handle incoming connections connection_thread = threading.Thread(target=self._handle_connections) connection_thread.daemon = True connection_thread.start() self.threads.append(connection_thread) return True except Exception as e: self.logger.error(f"Failed to connect to SSH server: {str(e)}") self.connection_status = "error" self.connection_error = str(e) self.is_running = False return False def _monitor_connection(self): """ Monitor the connection and reconnect if necessary. """ while self.should_reconnect: current_time = time.time() # Only check tunnel every check_interval seconds if current_time - self.last_check_time >= self.check_interval: self.last_check_time = current_time # Perform active check of tunnel status is_active = self._check_tunnel_active() if not is_active and self.is_running: self.logger.warning("Active check detected tunnel is down despite is_running=True") self.stop(reconnect=True) continue if not self.is_running: self.logger.info("Connection is down, attempting to reconnect...") successful = self.connect() if not successful: self.logger.warning(f"Reconnection failed, waiting {self.reconnect_interval} seconds...") time.sleep(self.reconnect_interval) # Check if transport is still active elif self.transport and not self.transport.is_active(): self.logger.warning("Transport is no longer active") self.stop(reconnect=True) # Check for activity timeout (in case keepalive fails) elif time.time() - self.last_activity_time > self.reconnect_interval * 2: self.logger.warning("No activity detected, reconnecting...") self.stop(reconnect=True) time.sleep(5) # Check connection status every 5 seconds def _check_tunnel_active(self): """ Actively check if the tunnel is working by attempting a simple operation. Returns: bool: True if the tunnel is active, False otherwise """ if not self.transport or not self.client: return False try: # Try to execute a simple command to test connection if self.transport.is_active(): # Try to open a test channel test_channel = self.transport.open_channel("session") if test_channel: test_channel.close() return True return False except Exception as e: self.logger.warning(f"Active tunnel check failed: {str(e)}") return False def _send_keepalive(self): """ Send keepalive packets to maintain the SSH connection. """ while self.should_reconnect: if self.is_running and self.transport and self.transport.is_active(): try: self.transport.send_ignore() self.last_activity_time = time.time() self.logger.debug("Sent keepalive packet") except Exception as e: self.logger.warning(f"Failed to send keepalive: {str(e)}") time.sleep(self.keep_alive_interval) def check_status(self): """ Check the current status of the tunnel. Returns: dict: Status information including whether the tunnel is running, connection status, and any error messages. """ # Do an active check if it's been more than check_interval since last check current_time = time.time() if current_time - self.last_check_time >= self.check_interval: self.last_check_time = current_time is_active = self._check_tunnel_active() if not is_active and self.is_running: self.logger.warning("Status check: tunnel appears down but marked as running") self.is_running = False self.connection_status = "disconnected" self.connection_error = "Tunnel appears to be down" return { "is_running": self.is_running, "status": self.connection_status, "error": self.connection_error, "last_activity": time.time() - self.last_activity_time } def _handle_connections(self): """ Handle incoming connections on the local port. This method accepts incoming connections and creates a new thread to handle each connection. """ try: while self.is_running: try: client_socket, client_addr = self.server_socket.accept() self.logger.info(f"New connection from {client_addr[0]}:{client_addr[1]}") self.last_activity_time = time.time() thread = threading.Thread( target=self._forward_traffic, args=(client_socket, client_addr) ) thread.daemon = True thread.start() self.threads.append(thread) except (socket.timeout, socket.error) as e: if self.is_running: self.logger.error(f"Socket error: {str(e)}") except Exception as e: if self.is_running: self.logger.error(f"Error accepting connection: {str(e)}") except Exception as e: self.logger.error(f"Error in connection handler: {str(e)}") if self.is_running: self.stop(reconnect=True) def _forward_traffic(self, client_socket, client_addr): """ Forward traffic between the local client and the remote server. Args: client_socket (socket.socket): The socket for the local client connection client_addr (tuple): The address of the local client """ channel = None try: # Create a channel to the remote host through the SSH transport remote_addr = ('127.0.0.1', self.remote_port) channel = self.transport.open_channel( "direct-tcpip", remote_addr, client_addr ) if channel is None: self.logger.error(f"Failed to open channel to {remote_addr[0]}:{remote_addr[1]}") client_socket.close() return self.logger.info(f"Channel opened to {remote_addr[0]}:{remote_addr[1]}") self.last_activity_time = time.time() # Forward data in both directions while True: r, w, e = select.select([client_socket, channel], [], [], 1) if client_socket in r: data = client_socket.recv(4096) if len(data) == 0: break channel.send(data) self.last_activity_time = time.time() if channel in r: data = channel.recv(4096) if len(data) == 0: break client_socket.send(data) self.last_activity_time = time.time() except Exception as e: self.logger.error(f"Error forwarding traffic: {str(e)}") finally: client_socket.close() if channel: channel.close() self.logger.info(f"Connection from {client_addr[0]}:{client_addr[1]} closed") def stop(self, reconnect=False): """ Stop the SSH tunnel. Args: reconnect (bool): If True, the tunnel will attempt to reconnect after stopping. """ self.is_running = False if self.server_socket: try: self.server_socket.close() except Exception: pass self.server_socket = None if self.transport: try: self.transport.close() except Exception: pass self.transport = None if self.client: try: self.client.close() except Exception: pass self.client = None if not reconnect: self.should_reconnect = False self.logger.info("SSH tunnel stopped permanently") else: self.logger.info("SSH tunnel stopped, will attempt to reconnect") def __enter__(self): """ Enter the context manager. Returns: SSHTunnel: The SSH tunnel instance. """ self.start() return self def __exit__(self, exc_type, exc_val, exc_tb): """ Exit the context manager. """ self.stop(reconnect=False)