SSH Automation

class pxssh(spawn):
    """
    Specialized SSH connection class that extends spawn with SSH-specific methods.
    Handles SSH login, logout, prompt detection, and common SSH scenarios.
    """
    
    def __init__(self, timeout=30, maxread=2000, searchwindowsize=None,
                 logfile=None, cwd=None, env=None, ignore_sighup=False,
                 echo=True, options={}, spawn_local_ssh=True, sync_multiplier=1,
                 check_local_ip=True):
        """
        Initialize SSH connection object.
        
        Parameters:
        - timeout (int): Default timeout for operations
        - maxread (int): Maximum bytes to read at once
        - searchwindowsize (int): Search window size for pattern matching
        - logfile (file-like): File object for logging
        - cwd (str): Working directory (inherited from spawn)
        - env (dict): Environment variables (inherited from spawn)
        - ignore_sighup (bool): Ignore SIGHUP signal
        - echo (bool): Terminal echo setting
        - options (dict): SSH client options
        - spawn_local_ssh (bool): Use local SSH client
        - sync_multiplier (int): Synchronization timing multiplier
        - check_local_ip (bool): Verify local IP for security
        """

def login(self, server, username, password='', terminal_type='ansi',
          original_prompt=r"[#$]", login_timeout=10, port=None,
          auto_prompt_reset=True, ssh_key=None, quiet=True,
          sync_multiplier=1, check_local_ip=True, password_regex=None,
          ssh_tunnels=None, spawn_local_ssh=True, sync_original_prompt=True,
          ssh_config=None, cmd="ssh"):
    """
    Establish SSH connection and login.
    
    Parameters:
    - server (str): Hostname or IP address to connect to
    - username (str): Username for SSH login
    - password (str): Password for authentication (if not using key)
    - terminal_type (str): Terminal type to request
    - original_prompt (str): Regex pattern for shell prompt
    - login_timeout (int): Timeout for login process
    - port (int): SSH port number (default: 22)
    - auto_prompt_reset (bool): Automatically set unique prompt
    - ssh_key (str): Path to SSH private key file
    - quiet (bool): Suppress SSH client messages
    - sync_multiplier (int): Timing adjustment for slow connections
    - check_local_ip (bool): Verify local IP address
    - password_regex (str): Custom regex for password prompt
    - ssh_tunnels (dict): SSH tunnel configurations
    - spawn_local_ssh (bool): Use local SSH client
    - sync_original_prompt (bool): Synchronize with original prompt
    - ssh_config (str): Path to SSH config file
    - cmd (str): SSH command to execute (default: "ssh")
    
    Returns:
    bool: True if login successful, False otherwise
    
    Raises:
    - ExceptionPxssh: If login fails or connection issues occur
    """

def logout(self):
    """
    Logout from SSH session and close connection.
    
    Sends 'exit' command and closes the connection cleanly.
    """

def prompt(self, timeout=-1):
    """
    Wait for shell prompt to appear.
    
    Parameters:
    - timeout (int): Timeout in seconds (-1 for default)
    
    Returns:
    bool: True if prompt found, False on timeout
    
    Raises:
    - TIMEOUT: If timeout exceeded
    - EOF: If connection closed
    """

def set_unique_prompt(self):
    """
    Set a unique shell prompt for reliable pattern matching.
    
    Changes the shell prompt to a unique string that's unlikely to
    appear in command output, making expect operations more reliable.
    
    Returns:
    bool: True if prompt was set successfully
    """

def sync_original_prompt(self, sync_multiplier=1):
    """
    Synchronize with the original shell prompt.
    
    Parameters:
    - sync_multiplier (int): Timing adjustment multiplier
    
    Returns:
    bool: True if synchronized successfully
    """

class ExceptionPxssh(ExceptionPexpect):
    """
    Exception class for pxssh-specific errors.
    
    Raised for SSH connection failures, authentication errors,
    and other SSH-specific problems.
    """

# Common SSH options that can be passed to pxssh
ssh_options = {
    'StrictHostKeyChecking': 'no',    # Skip host key verification
    'UserKnownHostsFile': '/dev/null', # Don't save host keys
    'ConnectTimeout': '10',            # Connection timeout
    'ServerAliveInterval': '60',       # Keep-alive interval
    'ServerAliveCountMax': '3',        # Keep-alive retry count
    'PasswordAuthentication': 'yes',   # Allow password auth
    'PubkeyAuthentication': 'yes',     # Allow key auth
}

ssh = pxssh.pxssh(options=ssh_options)

# Password authentication
ssh = pxssh.pxssh()
ssh.login('server.example.com', 'username', 'password')

# SSH key authentication
ssh = pxssh.pxssh()
ssh.login('server.example.com', 'username', ssh_key='/path/to/private_key')

# Password authentication with custom prompt
ssh = pxssh.pxssh()
ssh.login('server.example.com', 'username', 'password', 
          original_prompt=r'[#$%>] ')

from pexpect import pxssh
import getpass

# Create SSH connection
ssh = pxssh.pxssh()

try:
    # Login with password
    hostname = 'server.example.com'
    username = 'myuser'
    password = getpass.getpass('Password: ')
    
    ssh.login(hostname, username, password)
    
    # Execute commands
    ssh.sendline('ls -la')
    ssh.prompt()
    print(ssh.before.decode())
    
    ssh.sendline('uptime')
    ssh.prompt()
    print(ssh.before.decode())
    
    # Logout cleanly
    ssh.logout()

except pxssh.ExceptionPxssh as e:
    print(f"SSH connection failed: {e}")

from pexpect import pxssh

ssh = pxssh.pxssh()

try:
    # Login with SSH key
    ssh.login('server.example.com', 'username', 
              ssh_key='/home/user/.ssh/id_rsa')
    
    # Run multiple commands
    commands = ['whoami', 'pwd', 'date', 'df -h']
    
    for cmd in commands:
        ssh.sendline(cmd)
        ssh.prompt()
        output = ssh.before.decode().strip()
        print(f"{cmd}: {output}")
    
    ssh.logout()

except pxssh.ExceptionPxssh as e:
    print(f"SSH error: {e}")

from pexpect import pxssh
import time

def robust_ssh_connect(hostname, username, password, max_retries=3):
    """Establish SSH connection with retry logic."""
    
    for attempt in range(max_retries):
        ssh = pxssh.pxssh()
        
        try:
            ssh.login(hostname, username, password, login_timeout=30)
            print(f"Connected to {hostname} on attempt {attempt + 1}")
            return ssh
            
        except pxssh.ExceptionPxssh as e:
            print(f"Attempt {attempt + 1} failed: {e}")
            if attempt < max_retries - 1:
                time.sleep(5)  # Wait before retry
            else:
                raise
    
    return None

# Use robust connection
try:
    ssh = robust_ssh_connect('server.example.com', 'user', 'pass')
    
    # Your SSH operations here
    ssh.sendline('hostname')
    ssh.prompt()
    print(f"Connected to: {ssh.before.decode().strip()}")
    
    ssh.logout()
    
except Exception as e:
    print(f"Failed to establish SSH connection: {e}")

from pexpect import pxssh

# SSH with port forwarding
ssh_tunnels = {
    'local': [
        {'local_port': 8080, 'remote_host': 'localhost', 'remote_port': 80}
    ]
}

ssh = pxssh.pxssh()

try:
    ssh.login('jumphost.example.com', 'username', 'password',
              ssh_tunnels=ssh_tunnels)
    
    # Now port 8080 on local machine forwards to port 80 on remote
    print("SSH tunnel established")
    
    # Keep connection alive while using tunnel
    ssh.sendline('echo "Tunnel active"')
    ssh.prompt()
    
    # Your application can now use localhost:8080
    # to access the remote service
    
finally:
    ssh.logout()

from pexpect import pxssh

# Custom SSH client configuration
ssh = pxssh.pxssh(
    options={
        'StrictHostKeyChecking': 'no',
        'UserKnownHostsFile': '/dev/null',
        'ConnectTimeout': '30',
        'ServerAliveInterval': '120'
    },
    timeout=60,
    sync_multiplier=2  # For slow connections
)

try:
    # Login with custom configuration
    ssh.login('slow-server.example.com', 'username', 'password',
              login_timeout=60,
              terminal_type='xterm-256color')
    
    # Set custom prompt for better reliability
    ssh.set_unique_prompt()
    
    # Execute commands with custom prompt
    ssh.sendline('export TERM=xterm-256color')
    ssh.prompt()
    
    ssh.sendline('ls --color=auto')
    ssh.prompt()
    print(ssh.before.decode())
    
    ssh.logout()

except pxssh.ExceptionPxssh as e:
    print(f"SSH error: {e}")

from pexpect import pxssh
import contextlib

@contextlib.contextmanager
def ssh_session(hostname, username, password):
    """Context manager for SSH sessions."""
    ssh = pxssh.pxssh()
    try:
        ssh.login(hostname, username, password)
        yield ssh
    except pxssh.ExceptionPxssh as e:
        print(f"SSH error: {e}")
        raise
    finally:
        try:
            ssh.logout()
        except:
            pass  # Ignore logout errors

# Use context manager
with ssh_session('server.example.com', 'user', 'pass') as ssh:
    ssh.sendline('whoami')
    ssh.prompt()
    user = ssh.before.decode().strip()
    print(f"Logged in as: {user}")
    
    ssh.sendline('uname -a')
    ssh.prompt()
    system_info = ssh.before.decode().strip()
    print(f"System: {system_info}")
# SSH connection automatically closed