Create ssh_tunneler.py

ssh_tunneler.py

@@ -0,0 +1,355 @@
+#!/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 = []
+        
+        # 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:
+            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 _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.
+        """
+        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)