or run

npx @tessl/cli init

Version

Tile

Overview

Evals

Files

docs

index.md

tile.json

tessl/pypi-sshtunnel

Pure python SSH tunnels for creating secure connections through intermediate servers

Workspace: tessl
Visibility: Public
Created: 3 months ago
Last updated: 3 months ago
Describes: pkg:pypi/sshtunnel@0.4.x

To install, run

npx @tessl/cli install tessl/pypi-sshtunnel@0.4.0

SSH Tunnel

Pure python SSH tunnels for creating secure connections through intermediate servers. This library enables both local and remote port forwarding via SSH, supporting applications that need reliable SSH tunnel functionality without external dependencies beyond paramiko.

Package Information

Package Name: sshtunnel
Language: Python
Installation: pip install sshtunnel

Core Imports

import sshtunnel

Common patterns:

from sshtunnel import SSHTunnelForwarder, open_tunnel

Basic Usage

from sshtunnel import SSHTunnelForwarder

# Create and start SSH tunnel
server = SSHTunnelForwarder(
    ssh_address_or_host='ssh.example.com',
    ssh_username='user',
    ssh_password='secret',
    remote_bind_address=('127.0.0.1', 3306),
    local_bind_address=('127.0.0.1', 3307)
)

server.start()
print(f"Server connected via local port: {server.local_bind_port}")

# Use your tunnel connection here
# Connect to localhost:3307 to reach 127.0.0.1:3306 on remote server

server.stop()

Context manager usage:

from sshtunnel import open_tunnel

with open_tunnel(
    ('ssh.example.com', 22),
    ssh_username='user', 
    ssh_password='secret',
    remote_bind_address=('127.0.0.1', 3306),
    local_bind_address=('127.0.0.1', 3307)
) as tunnel:
    print(f"Tunnel opened on local port: {tunnel.local_bind_port}")
    # Use tunnel connection here
# Tunnel automatically closed when exiting context

Architecture

The sshtunnel package is built around a few key components:

SSHTunnelForwarder: Main class managing SSH connections and tunnel forwarding
Forward servers: Background TCP/UNIX socket servers handling tunnel connections
Handler threads: Individual connection handlers for concurrent tunnel usage
Transport layer: Paramiko-based SSH transport with keepalive and authentication

Capabilities

SSH Tunnel Management

Main functionality for creating and managing SSH tunnels with full lifecycle control.

class SSHTunnelForwarder:
    def __init__(
        self,
        ssh_address_or_host=None,
        ssh_config_file=SSH_CONFIG_FILE,
        ssh_host_key=None,
        ssh_password=None,
        ssh_pkey=None,
        ssh_private_key_password=None,
        ssh_proxy=None,
        ssh_proxy_enabled=True,
        ssh_username=None,
        local_bind_address=None,
        local_bind_addresses=None,
        logger=None,
        mute_exceptions=False,
        remote_bind_address=None,
        remote_bind_addresses=None,
        set_keepalive=5.0,
        threaded=True,
        compression=None,
        allow_agent=True,
        host_pkey_directories=None
    ): ...
    
    def start(self): ...
    def stop(self, force=False): ...
    def close(self): ...
    def restart(self): ...

Tunnel Status and Properties

Access tunnel connection information and status.

class SSHTunnelForwarder:
    @property
    def local_bind_port(self) -> int: ...
    
    @property 
    def local_bind_ports(self) -> list[int]: ...
    
    @property
    def local_bind_host(self) -> str: ...
    
    @property
    def local_bind_hosts(self) -> list[str]: ...
    
    @property
    def local_bind_address(self) -> tuple: ...
    
    @property
    def local_bind_addresses(self) -> list[tuple]: ...
    
    @property
    def tunnel_bindings(self) -> dict: ...
    
    tunnel_is_up: dict  # Instance attribute - tunnel status mapping
    
    @property
    def is_active(self) -> bool: ...
    
    is_alive: bool  # Instance attribute - whether tunnels are alive
    
    def check_tunnels(self): ...
    def local_is_up(self, target: tuple) -> bool: ...  # deprecated

Simplified Tunnel Interface

Convenience function for common tunnel use cases with context manager support.

def open_tunnel(*args, **kwargs) -> SSHTunnelForwarder:
    """
    Open an SSH Tunnel, wrapper for SSHTunnelForwarder.
    
    Parameters:
    - ssh_address_or_host: SSH server address/hostname
    - ssh_username: SSH username
    - ssh_password: SSH password  
    - ssh_pkey: Private key file path or paramiko.PKey object
    - ssh_port: SSH port (default: 22)
    - remote_bind_address: Remote (host, port) tuple
    - remote_bind_addresses: List of remote bind addresses
    - local_bind_address: Local (host, port) tuple
    - local_bind_addresses: List of local bind addresses
    - debug_level: Log level for debugging
    - skip_tunnel_checkup: Disable tunnel status checking
    - Additional SSHTunnelForwarder parameters supported
    
    Returns:
    SSHTunnelForwarder instance configured as context manager
    """

SSH Key Management

Utilities for loading SSH keys from various sources.

class SSHTunnelForwarder:
    @staticmethod
    def get_agent_keys(logger=None) -> list:
        """Load public keys from SSH agent."""
    
    @staticmethod
    def get_keys(
        logger=None, 
        host_pkey_directories=None, 
        allow_agent=False
    ) -> list:
        """Load keys from SSH agent or local directories."""
    
    @staticmethod
    def read_private_key_file(
        pkey_file: str,
        pkey_password=None,
        key_type=None,
        logger=None
    ):
        """Read private key file and return paramiko.PKey object."""

Validation and Logging

Utility functions for address validation and logger configuration.

def check_host(host: str): ...
def check_port(port: int): ...  
def check_address(address): ...
def check_addresses(address_list: list, is_remote=False): ...

def create_logger(
    logger=None,
    loglevel=None, 
    capture_warnings=True,
    add_paramiko_handler=True
) -> logging.Logger:
    """Create or configure logger with console handler."""

def address_to_str(address) -> str:
    """Convert address to string representation."""

def get_connection_id() -> int:
    """Generate unique connection ID for tunnel connections."""

Exception Classes

class BaseSSHTunnelForwarderError(Exception):
    """Base exception for SSHTunnelForwarder errors."""

class HandlerSSHTunnelForwarderError(BaseSSHTunnelForwarderError):
    """Exception for tunnel forwarder errors."""

Constants

__version__: str = '0.4.0'
__author__: str = 'pahaz'

SSH_TIMEOUT: float = 0.1  # Transport socket timeout
TUNNEL_TIMEOUT: float = 10.0  # Tunnel connection timeout
DEFAULT_LOGLEVEL: int = logging.ERROR  # Default log level
TRACE_LEVEL: int = 1  # Trace log level
DEFAULT_SSH_DIRECTORY: str = '~/.ssh'  # Default SSH directory
SSH_CONFIG_FILE: str  # Path to SSH config file

Usage Examples

Multiple Tunnels

from sshtunnel import SSHTunnelForwarder

# Create multiple tunnels at once
server = SSHTunnelForwarder(
    ssh_address_or_host='ssh.example.com',
    ssh_username='user',
    ssh_password='secret',
    remote_bind_addresses=[('127.0.0.1', 3306), ('127.0.0.1', 5432)],
    local_bind_addresses=[('127.0.0.1', 3307), ('127.0.0.1', 5433)]
)

server.start()
print(f"MySQL tunnel: localhost:{server.local_bind_ports[0]}")
print(f"PostgreSQL tunnel: localhost:{server.local_bind_ports[1]}")
server.stop()

SSH Key Authentication

from sshtunnel import SSHTunnelForwarder

# Using SSH key file
server = SSHTunnelForwarder(
    ssh_address_or_host='example.com',
    ssh_username='user',
    ssh_pkey='/home/user/.ssh/id_rsa',
    ssh_private_key_password='key_password',  # if key is encrypted
    remote_bind_address=('192.168.1.100', 22),
    local_bind_address=('127.0.0.1', 2222)
)

# Using SSH agent keys
server = SSHTunnelForwarder(
    ssh_address_or_host='example.com', 
    ssh_username='user',
    allow_agent=True,
    remote_bind_address=('192.168.1.100', 22)
)

Custom Logger Configuration

import logging
from sshtunnel import SSHTunnelForwarder, create_logger

# Create custom logger
logger = create_logger(loglevel=logging.DEBUG)

server = SSHTunnelForwarder(
    ssh_address_or_host='example.com',
    ssh_username='user',
    ssh_password='secret',
    remote_bind_address=('127.0.0.1', 3306),
    logger=logger
)

Error Handling

from sshtunnel import (
    SSHTunnelForwarder, 
    BaseSSHTunnelForwarderError,
    HandlerSSHTunnelForwarderError
)

try:
    server = SSHTunnelForwarder(
        ssh_address_or_host='example.com',
        ssh_username='user', 
        ssh_password='wrong_password',
        remote_bind_address=('127.0.0.1', 3306),
        mute_exceptions=False  # Enable exception raising
    )
    server.start()
except BaseSSHTunnelForwarderError as e:
    print(f"SSH tunnel error: {e}")
except HandlerSSHTunnelForwarderError as e:
    print(f"Tunnel handler error: {e}")

Proxy Support

from sshtunnel import SSHTunnelForwarder

# Using HTTP proxy
server = SSHTunnelForwarder(
    ssh_address_or_host='target.example.com',
    ssh_username='user',
    ssh_password='secret', 
    ssh_proxy=('proxy.example.com', 8080),
    remote_bind_address=('127.0.0.1', 3306)
)

# Using ProxyCommand from SSH config
server = SSHTunnelForwarder(
    ssh_address_or_host='target.example.com',
    ssh_username='user',
    ssh_config_file='~/.ssh/config',  # Contains ProxyCommand
    ssh_proxy_enabled=True,
    remote_bind_address=('127.0.0.1', 3306)
)