Available for day contractsFrom 21st September I have availability for day and half day contracts. Please contact for more information.

Contact →
mikepreston.org

Python Paramiko

A powerful SSH2 protocol library for Python that enables secure remote command execution and file transfers.

Python Paramiko Cheatsheet

A powerful SSH2 protocol library for Python that enables secure remote command execution and file transfers.

Overview

Paramiko provides a pure-Python implementation of SSH2, allowing you to connect to remote servers, execute commands, and transfer files securely. It supports both password and key-based authentication.

Remote ServerSSH ConnectionPython ApplicationSSHClientSFTPClientAuthenticationChannelTransportShell/CommandsFile SystemRemote ServerSSH ConnectionPython ApplicationSSHClientSFTPClientAuthenticationChannelTransportShell/CommandsFile System

Installation

uv pip install paramiko

SSH Connection

Key Concepts

  • SSHClient is the primary class for establishing SSH connections
  • Transport provides the underlying SSH protocol implementation
  • AutoAddPolicy automatically adds unknown host keys (use cautiously)
  • Connection parameters include hostname, port, username, and credentials
  • Host key verification prevents man-in-the-middle attacks

Common Patterns

import paramiko

# Create SSH client
client = paramiko.SSHClient()

# Handle host keys
# WARNING: AutoAddPolicy accepts any host key without verification — MITM risk.
# Prefer: client.load_system_host_keys() + RejectPolicy() for production.
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# Connect to server
client.connect(
    hostname='server.example.com',
    port=22,
    username='user',
    password='password'
)

# Always close connection
client.close()

Examples

Basic Connection with Password

import paramiko

def connect_with_password(host, username, password, port=22):
    """Establish SSH connection using password authentication."""
    client = paramiko.SSHClient()

    # Load system host keys
    client.load_system_host_keys()

    # Set policy for unknown hosts
    client.set_missing_host_key_policy(paramiko.WarningPolicy())

    try:
        client.connect(
            hostname=host,
            port=port,
            username=username,
            password=password
        )
        print(f"Connected to {host}")
        return client
    except paramiko.AuthenticationException:
        print("Authentication failed")
        raise
    except paramiko.SSHException as e:
        print(f"SSH error: {e}")
        raise

# Usage
client = connect_with_password('192.168.1.100', 'admin', 'secret123')
# ... perform operations ...
client.close()

Connection with Context Manager

import paramiko
from contextlib import contextmanager

@contextmanager
def ssh_connection(host, username, password=None, key_filename=None, port=22):
    """Context manager for SSH connections with automatic cleanup."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        client.connect(
            hostname=host,
            port=port,
            username=username,
            password=password,
            key_filename=key_filename
        )
        yield client
    finally:
        client.close()

# Usage
with ssh_connection('server.example.com', 'user', password='secret') as client:
    stdin, stdout, stderr = client.exec_command('hostname')
    print(stdout.read().decode())

Using Transport Directly

import paramiko

# Create transport for low-level control
transport = paramiko.Transport(('server.example.com', 22))
transport.connect(username='user', password='password')

# Create channel
channel = transport.open_session()
channel.exec_command('ls -la')

# Read output
output = channel.recv(4096).decode()
print(output)

# Cleanup
channel.close()
transport.close()

Executing Commands

Key Concepts

  • exec_command() executes a single command on the remote server
  • Returns three file-like objects: stdin, stdout, stderr
  • Non-blocking by default - use read() to wait for completion
  • Exit status indicates command success or failure
  • Interactive commands require special handling with invoke_shell()

Common Patterns

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

# Execute command
stdin, stdout, stderr = client.exec_command('ls -la')

# Read output
output = stdout.read().decode('utf-8')
errors = stderr.read().decode('utf-8')

# Get exit status
exit_status = stdout.channel.recv_exit_status()

client.close()

Examples

Basic Command Execution

import paramiko

def run_command(client, command):
    """Execute command and return output, errors, and exit status."""
    stdin, stdout, stderr = client.exec_command(command)

    # Wait for command to complete
    exit_status = stdout.channel.recv_exit_status()

    output = stdout.read().decode('utf-8').strip()
    errors = stderr.read().decode('utf-8').strip()

    return {
        'output': output,
        'errors': errors,
        'exit_status': exit_status
    }

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

result = run_command(client, 'df -h')
print(f"Exit status: {result['exit_status']}")
print(f"Output:\n{result['output']}")

if result['errors']:
    print(f"Errors:\n{result['errors']}")

client.close()

Command with Input

import paramiko

def run_command_with_input(client, command, input_data):
    """Execute command that requires stdin input."""
    stdin, stdout, stderr = client.exec_command(command)

    # Send input data
    stdin.write(input_data)
    stdin.channel.shutdown_write()  # Signal end of input

    # Wait for completion
    exit_status = stdout.channel.recv_exit_status()

    return {
        'output': stdout.read().decode('utf-8'),
        'errors': stderr.read().decode('utf-8'),
        'exit_status': exit_status
    }

# Usage - example with sudo
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

result = run_command_with_input(
    client,
    'sudo -S apt update',
    'password\n'  # Send password for sudo
)
print(result['output'])

client.close()

Running Multiple Commands

import paramiko

def run_multiple_commands(client, commands):
    """Execute multiple commands sequentially."""
    results = []

    for command in commands:
        stdin, stdout, stderr = client.exec_command(command)
        exit_status = stdout.channel.recv_exit_status()

        results.append({
            'command': command,
            'output': stdout.read().decode('utf-8').strip(),
            'errors': stderr.read().decode('utf-8').strip(),
            'exit_status': exit_status
        })

    return results

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

commands = [
    'hostname',
    'uptime',
    'free -m',
    'df -h /'
]

results = run_multiple_commands(client, commands)

for result in results:
    print(f"\n=== {result['command']} ===")
    print(result['output'])

client.close()

Interactive Shell Session

import paramiko
import time

def interactive_shell(client, commands, delay=0.5):
    """Execute commands in an interactive shell session."""
    channel = client.invoke_shell()

    # Wait for shell to initialise
    time.sleep(1)

    # Clear initial output
    if channel.recv_ready():
        channel.recv(4096)

    output = []

    for command in commands:
        # Send command
        channel.send(command + '\n')
        time.sleep(delay)

        # Receive output
        while channel.recv_ready():
            data = channel.recv(4096).decode('utf-8')
            output.append(data)

    channel.close()
    return ''.join(output)

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

commands = ['cd /var/log', 'ls -la', 'pwd']
output = interactive_shell(client, commands)
print(output)

client.close()

SFTP Operations

Key Concepts

  • SFTPClient provides file transfer capabilities over SSH
  • get() downloads files from remote to local
  • put() uploads files from local to remote
  • listdir() lists directory contents
  • stat() retrieves file attributes (size, permissions, timestamps)
  • File operations include chmod, chown, mkdir, remove, rename
SSHClientopen_sftpSFTPClientFile Operationsget - Downloadput - Uploadlistdir - Liststat - Attributesmkdir/rmdirchmod/chownremove/renameSSHClientopen_sftpSFTPClientFile Operationsget - Downloadput - Uploadlistdir - Liststat - Attributesmkdir/rmdirchmod/chownremove/rename

Common Patterns

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

# Open SFTP session
sftp = client.open_sftp()

# Download file
sftp.get('/remote/path/file.txt', '/local/path/file.txt')

# Upload file
sftp.put('/local/path/file.txt', '/remote/path/file.txt')

# List directory
files = sftp.listdir('/remote/path')

# Get file stats
attrs = sftp.stat('/remote/path/file.txt')

sftp.close()
client.close()

Examples

Basic File Transfer

import paramiko
import os

def download_file(client, remote_path, local_path):
    """Download a file from remote server."""
    sftp = client.open_sftp()

    try:
        sftp.get(remote_path, local_path)
        print(f"Downloaded: {remote_path} -> {local_path}")
    finally:
        sftp.close()

def upload_file(client, local_path, remote_path):
    """Upload a file to remote server."""
    sftp = client.open_sftp()

    try:
        sftp.put(local_path, remote_path)
        print(f"Uploaded: {local_path} -> {remote_path}")
    finally:
        sftp.close()

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

download_file(client, '/var/log/syslog', '/tmp/syslog_backup')
upload_file(client, '/tmp/config.yaml', '/etc/app/config.yaml')

client.close()

Directory Operations

import paramiko
from stat import S_ISDIR, S_ISREG

def list_directory(client, remote_path):
    """List directory contents with details."""
    sftp = client.open_sftp()

    try:
        items = []
        for entry in sftp.listdir_attr(remote_path):
            item = {
                'name': entry.filename,
                'size': entry.st_size,
                'modified': entry.st_mtime,
                'is_dir': S_ISDIR(entry.st_mode),
                'permissions': oct(entry.st_mode)[-3:]
            }
            items.append(item)

        return items
    finally:
        sftp.close()

def create_directory(client, remote_path, mode=0o755):
    """Create a directory on the remote server."""
    sftp = client.open_sftp()

    try:
        sftp.mkdir(remote_path, mode)
        print(f"Created directory: {remote_path}")
    finally:
        sftp.close()

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

# List directory
items = list_directory(client, '/var/log')
for item in items[:10]:  # First 10 items
    type_indicator = 'd' if item['is_dir'] else 'f'
    print(f"[{type_indicator}] {item['name']} ({item['size']} bytes)")

# Create directory
create_directory(client, '/tmp/my_app_data')

client.close()

Recursive Directory Transfer

import paramiko
import os
from stat import S_ISDIR

def download_directory(client, remote_path, local_path):
    """Recursively download a directory from remote server."""
    sftp = client.open_sftp()

    try:
        os.makedirs(local_path, exist_ok=True)

        for entry in sftp.listdir_attr(remote_path):
            remote_item = f"{remote_path}/{entry.filename}"
            local_item = os.path.join(local_path, entry.filename)

            if S_ISDIR(entry.st_mode):
                download_directory(client, remote_item, local_item)
            else:
                sftp.get(remote_item, local_item)
                print(f"Downloaded: {remote_item}")
    finally:
        sftp.close()

def upload_directory(client, local_path, remote_path):
    """Recursively upload a directory to remote server."""
    sftp = client.open_sftp()

    try:
        # Create remote directory
        try:
            sftp.mkdir(remote_path)
        except IOError:
            pass  # Directory may already exist

        for item in os.listdir(local_path):
            local_item = os.path.join(local_path, item)
            remote_item = f"{remote_path}/{item}"

            if os.path.isdir(local_item):
                upload_directory(client, local_item, remote_item)
            else:
                sftp.put(local_item, remote_item)
                print(f"Uploaded: {local_item}")
    finally:
        sftp.close()

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

download_directory(client, '/var/www/html', '/tmp/website_backup')
upload_directory(client, '/local/project', '/home/user/project')

client.close()

File Operations with Progress

import paramiko
import os

def transfer_with_progress(transferred, total):
    """Callback function to display transfer progress."""
    percentage = (transferred / total) * 100
    print(f"\rProgress: {percentage:.1f}% ({transferred}/{total} bytes)", end='')

def download_with_progress(client, remote_path, local_path):
    """Download file with progress display."""
    sftp = client.open_sftp()

    try:
        sftp.get(remote_path, local_path, callback=transfer_with_progress)
        print()  # New line after progress
    finally:
        sftp.close()

def upload_with_progress(client, local_path, remote_path):
    """Upload file with progress display."""
    sftp = client.open_sftp()

    try:
        sftp.put(local_path, remote_path, callback=transfer_with_progress)
        print()  # New line after progress
    finally:
        sftp.close()

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

download_with_progress(client, '/var/log/large_file.log', '/tmp/large_file.log')

client.close()

Key-Based Authentication

Key Concepts

  • RSA/Ed25519/ECDSA keys provide more secure authentication than passwords
  • Private key stays on the client; public key goes on the server
  • Passphrase adds extra security layer for encrypted private keys
  • SSH agent manages keys in memory to avoid repeated passphrase entry
  • Host keys verify server identity and prevent MITM attacks
ServerAuthenticationClientPrivate KeySSH AgentKey ChallengeSignatureVerificationauthorised_keysPublic KeyServerAuthenticationClientPrivate KeySSH AgentKey ChallengeSignatureVerificationauthorised_keysPublic Key

Common Patterns

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# Method 1: Key file path
client.connect(
    hostname='server.example.com',
    username='user',
    key_filename='/home/user/.ssh/id_rsa'
)

# Method 2: Key with passphrase
client.connect(
    hostname='server.example.com',
    username='user',
    key_filename='/home/user/.ssh/id_rsa',
    passphrase='key_passphrase'
)

# Method 3: Using SSH agent
client.connect(
    hostname='server.example.com',
    username='user',
    allow_agent=True
)

Examples

Connection with RSA Key

import paramiko
import os

def connect_with_key(host, username, key_path, passphrase=None, port=22):
    """Establish SSH connection using RSA key authentication."""
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.WarningPolicy())

    # Expand ~ in path
    key_path = os.path.expanduser(key_path)

    try:
        client.connect(
            hostname=host,
            port=port,
            username=username,
            key_filename=key_path,
            passphrase=passphrase
        )
        print(f"Connected to {host} using key authentication")
        return client
    except paramiko.AuthenticationException:
        print("Key authentication failed")
        raise
    except FileNotFoundError:
        print(f"Key file not found: {key_path}")
        raise

# Usage
client = connect_with_key(
    'server.example.com',
    'deploy',
    '~/.ssh/id_rsa',
    passphrase='my_key_passphrase'
)
# ... perform operations ...
client.close()

Loading Key from String

import paramiko
from io import StringIO

def connect_with_key_string(host, username, key_string, port=22):
    """Connect using a private key provided as a string."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    # Load key from string — try each type in turn.
    # Note: modern ssh-keygen produces "BEGIN OPENSSH PRIVATE KEY" for ALL key types
    # (RSA, ECDSA, Ed25519), so header-sniffing is unreliable. Try in preference order.
    key_file = StringIO(key_string)
    pkey = None
    for key_class in (paramiko.Ed25519Key, paramiko.ECDSAKey, paramiko.RSAKey):
        try:
            key_file.seek(0)
            pkey = key_class.from_private_key(key_file)
            break
        except paramiko.SSHException:
            continue
    if pkey is None:
        raise ValueError("Unsupported or unrecognised key type")

    client.connect(
        hostname=host,
        port=port,
        username=username,
        pkey=pkey
    )

    return client

# Usage
key_string = """-----BEGIN RSA PRIVATE KEY-----
MIIEpAIBAAKCAQEA...
-----END RSA PRIVATE KEY-----"""

client = connect_with_key_string('server.example.com', 'user', key_string)
client.close()

Using SSH Agent

import paramiko

def connect_with_agent(host, username, port=22):
    """Connect using keys from SSH agent."""
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.WarningPolicy())

    try:
        # allow_agent=True uses keys from ssh-agent
        client.connect(
            hostname=host,
            port=port,
            username=username,
            allow_agent=True,
            look_for_keys=True  # Also check ~/.ssh for keys
        )
        print(f"Connected to {host} via SSH agent")
        return client
    except paramiko.AuthenticationException:
        print("Agent authentication failed")
        raise

# Usage
# First add key to agent: ssh-add ~/.ssh/id_rsa
client = connect_with_agent('server.example.com', 'user')
client.close()

Multiple Authentication Methods

import paramiko
import os

def connect_with_fallback(host, username, password=None, key_path=None, port=22):
    """Try key authentication first, fall back to password."""
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    # Try key authentication first
    if key_path:
        key_path = os.path.expanduser(key_path)
        if os.path.exists(key_path):
            try:
                client.connect(
                    hostname=host,
                    port=port,
                    username=username,
                    key_filename=key_path
                )
                print("Connected using key authentication")
                return client
            except paramiko.AuthenticationException:
                print("Key auth failed, trying password...")

    # Fall back to password
    if password:
        try:
            client.connect(
                hostname=host,
                port=port,
                username=username,
                password=password
            )
            print("Connected using password authentication")
            return client
        except paramiko.AuthenticationException:
            print("Password authentication failed")
            raise

    raise paramiko.AuthenticationException("No valid authentication method")

# Usage
client = connect_with_fallback(
    'server.example.com',
    'user',
    password='backup_password',
    key_path='~/.ssh/id_rsa'
)
client.close()

Error Handling and Timeouts

Key Concepts

  • AuthenticationException raised when credentials are invalid
  • SSHException covers general SSH protocol errors
  • socket.timeout raised when connection or operation times out
  • Banner timeout limits time waiting for SSH banner
  • Connection timeout limits time to establish TCP connection

Common Patterns

import paramiko
import socket

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

try:
    client.connect(
        hostname='server.example.com',
        username='user',
        password='password',
        timeout=10,              # TCP connection timeout
        banner_timeout=15,       # SSH banner timeout
        auth_timeout=15          # Authentication timeout
    )
except paramiko.AuthenticationException:
    print("Authentication failed")
except paramiko.SSHException as e:
    print(f"SSH error: {e}")
except socket.timeout:
    print("Connection timed out")
except socket.error as e:
    print(f"Socket error: {e}")
finally:
    client.close()

Examples

Comprehensive Error Handling

import paramiko
import socket

def safe_ssh_connect(host, username, password=None, key_filename=None,
                     port=22, timeout=30):
    """Connect to SSH server with comprehensive error handling."""
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.WarningPolicy())

    try:
        client.connect(
            hostname=host,
            port=port,
            username=username,
            password=password,
            key_filename=key_filename,
            timeout=timeout,
            banner_timeout=timeout,
            auth_timeout=timeout
        )
        return client

    except paramiko.AuthenticationException:
        print(f"Authentication failed for {username}@{host}")
        raise

    except paramiko.BadHostKeyException as e:
        print(f"Host key verification failed: {e}")
        raise

    except paramiko.SSHException as e:
        print(f"SSH protocol error: {e}")
        raise

    except socket.timeout:
        print(f"Connection to {host}:{port} timed out after {timeout}s")
        raise

    except socket.gaierror as e:
        print(f"DNS resolution failed for {host}: {e}")
        raise

    except socket.error as e:
        print(f"Network error connecting to {host}:{port}: {e}")
        raise

    except Exception as e:
        print(f"Unexpected error: {type(e).__name__}: {e}")
        raise

# Usage
try:
    client = safe_ssh_connect(
        'server.example.com',
        'user',
        password='password',
        timeout=10
    )
    # ... perform operations ...
    client.close()
except Exception as e:
    print(f"Failed to connect: {e}")

Command Execution with Timeout

import paramiko
import socket

def run_command_with_timeout(client, command, timeout=30):
    """Execute command with timeout handling."""
    stdin, stdout, stderr = client.exec_command(command, timeout=timeout)

    # Set channel timeout
    stdout.channel.settimeout(timeout)

    try:
        # Wait for command with timeout
        exit_status = stdout.channel.recv_exit_status()

        output = stdout.read().decode('utf-8')
        errors = stderr.read().decode('utf-8')

        return {
            'success': exit_status == 0,
            'output': output,
            'errors': errors,
            'exit_status': exit_status
        }

    except socket.timeout:
        # Kill the command if it times out
        stdout.channel.close()
        return {
            'success': False,
            'output': '',
            'errors': f'Command timed out after {timeout} seconds',
            'exit_status': -1
        }

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

# Quick command
result = run_command_with_timeout(client, 'hostname', timeout=5)
print(result['output'])

# Long-running command with timeout
result = run_command_with_timeout(client, 'sleep 60 && echo done', timeout=10)
if not result['success']:
    print(f"Command failed: {result['errors']}")

client.close()

Retry Logic for Connections

import paramiko
import socket
import time

def connect_with_retry(host, username, password=None, key_filename=None,
                       port=22, max_retries=3, retry_delay=5):
    """Connect with automatic retry on transient failures."""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    last_exception = None

    for attempt in range(1, max_retries + 1):
        try:
            print(f"Connection attempt {attempt}/{max_retries}...")

            client.connect(
                hostname=host,
                port=port,
                username=username,
                password=password,
                key_filename=key_filename,
                timeout=10
            )

            print(f"Connected to {host}")
            return client

        except (socket.timeout, socket.error, paramiko.SSHException) as e:
            last_exception = e
            print(f"Attempt {attempt} failed: {e}")

            if attempt < max_retries:
                print(f"Retrying in {retry_delay} seconds...")
                time.sleep(retry_delay)
                retry_delay *= 2  # Exponential backoff

        except paramiko.AuthenticationException:
            # Don't retry authentication failures
            raise

    raise last_exception or Exception("Max retries exceeded")

# Usage
try:
    client = connect_with_retry(
        'server.example.com',
        'user',
        password='password',
        max_retries=3,
        retry_delay=5
    )
    # ... perform operations ...
    client.close()
except Exception as e:
    print(f"Failed after all retries: {e}")

SFTP with Error Handling

import paramiko
import os

def safe_sftp_transfer(client, operation, local_path, remote_path):
    """Perform SFTP operation with error handling."""
    sftp = None

    try:
        sftp = client.open_sftp()

        if operation == 'get':
            # Ensure local directory exists
            local_dir = os.path.dirname(local_path)
            if local_dir:
                os.makedirs(local_dir, exist_ok=True)

            sftp.get(remote_path, local_path)
            print(f"Downloaded: {remote_path}")

        elif operation == 'put':
            # Check if local file exists
            if not os.path.exists(local_path):
                raise FileNotFoundError(f"Local file not found: {local_path}")

            # Create remote directory if needed
            remote_dir = os.path.dirname(remote_path)
            try:
                sftp.stat(remote_dir)
            except FileNotFoundError:
                sftp.mkdir(remote_dir)

            sftp.put(local_path, remote_path)
            print(f"Uploaded: {local_path}")

        return True

    except FileNotFoundError as e:
        print(f"File not found: {e}")
        return False

    except PermissionError as e:
        print(f"Permission denied: {e}")
        return False

    except IOError as e:
        print(f"I/O error: {e}")
        return False

    finally:
        if sftp:
            sftp.close()

# Usage
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

success = safe_sftp_transfer(
    client, 'get',
    '/tmp/local_file.txt',
    '/var/log/remote_file.txt'
)

client.close()

Quick Reference

Operation Code Example
Create client client = paramiko.SSHClient()
Auto-add host keys client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
Load system keys client.load_system_host_keys()
Connect with password client.connect(host, username=user, password=pwd)
Connect with key client.connect(host, username=user, key_filename=path)
Connect with timeout client.connect(host, username=user, password=pwd, timeout=10)
Execute command stdin, stdout, stderr = client.exec_command('ls')
Read output output = stdout.read().decode('utf-8')
Get exit status exit_status = stdout.channel.recv_exit_status()
Open SFTP sftp = client.open_sftp()
Download file sftp.get(remote_path, local_path)
Upload file sftp.put(local_path, remote_path)
List directory files = sftp.listdir(path)
List with attrs entries = sftp.listdir_attr(path)
Get file stats attrs = sftp.stat(path)
Create directory sftp.mkdir(path)
Remove file sftp.remove(path)
Rename file sftp.rename(old_path, new_path)
Change permissions sftp.chmod(path, mode)
Close SFTP sftp.close()
Close connection client.close()
Interactive shell channel = client.invoke_shell()
Set channel timeout channel.settimeout(seconds)

Common Issues and Solutions

Issue: Host Key Verification Failed

Problem: paramiko.ssh_exception.SSHException: Server 'host' not found in known_hosts

Solution:

import paramiko

client = paramiko.SSHClient()

# Option 1: Auto-add unknown hosts (convenient but less secure)
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# Option 2: Warn but add unknown hosts
client.set_missing_host_key_policy(paramiko.WarningPolicy())

# Option 3: Load known hosts and reject unknown (most secure)
client.load_system_host_keys()
client.set_missing_host_key_policy(paramiko.RejectPolicy())

client.connect('server.example.com', username='user', password='password')

Issue: Authentication Fails with Key

Problem: Key authentication fails even with correct key

Solution:

import paramiko
import os

# Ensure correct permissions (chmod 600)
key_path = os.path.expanduser('~/.ssh/id_rsa')
os.chmod(key_path, 0o600)

# Specify key type explicitly
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

# Try loading specific key type
pkey = paramiko.RSAKey.from_private_key_file(key_path, password='passphrase')

client.connect(
    'server.example.com',
    username='user',
    pkey=pkey
)

Issue: Command Output Truncated

Problem: Output from exec_command appears incomplete

Solution:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

stdin, stdout, stderr = client.exec_command('large_output_command')

# IMPORTANT: Wait for command to complete before reading
exit_status = stdout.channel.recv_exit_status()

# Now read all output
output = stdout.read().decode('utf-8')
errors = stderr.read().decode('utf-8')

client.close()

Issue: Connection Timeout

Problem: Connection hangs or times out

Solution:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

client.connect(
    'server.example.com',
    username='user',
    password='password',
    timeout=10,              # TCP connection timeout
    banner_timeout=15,       # Time to receive SSH banner
    auth_timeout=15          # Time for authentication
)

# For exec_command timeout
stdin, stdout, stderr = client.exec_command('long_command', timeout=60)
stdout.channel.settimeout(60)

Issue: SFTP Transfer Fails

Problem: IOError: [Errno 2] No such file during transfer

Solution:

import paramiko
import os

def ensure_remote_dir(sftp, remote_path):
    """Create remote directory if it doesn't exist."""
    remote_dir = os.path.dirname(remote_path)
    if remote_dir:
        try:
            sftp.stat(remote_dir)
        except FileNotFoundError:
            # Create directory recursively
            dirs = remote_dir.split('/')
            current = ''
            for d in dirs:
                if d:
                    current += f'/{d}'
                    try:
                        sftp.stat(current)
                    except FileNotFoundError:
                        sftp.mkdir(current)

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

sftp = client.open_sftp()
remote_path = '/new/nested/directory/file.txt'

ensure_remote_dir(sftp, remote_path)
sftp.put('local_file.txt', remote_path)

sftp.close()
client.close()

Issue: Unicode Encoding Errors

Problem: UnicodeDecodeError when reading command output

Solution:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

stdin, stdout, stderr = client.exec_command('command_with_special_chars')

# Use errors='replace' or 'ignore' for problematic characters
output = stdout.read().decode('utf-8', errors='replace')

# Or try different encoding
output = stdout.read().decode('latin-1')

client.close()

Issue: Channel Closed Unexpectedly

Problem: paramiko.ssh_exception.SSHException: Channel closed

Solution:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect('server.example.com', username='user', password='password')

# Check if connection is active before operations
transport = client.get_transport()
if transport and transport.is_active():
    stdin, stdout, stderr = client.exec_command('command')
    output = stdout.read().decode('utf-8')
else:
    print("Connection lost, reconnecting...")
    client.connect('server.example.com', username='user', password='password')

client.close()

Related Topics

  • Fabric - High-level SSH library built on Paramiko for deployment automation
  • Ansible - IT automation platform that uses SSH for remote execution
  • asyncssh - Asynchronous SSH library for asyncio applications
  • scp - Simple secure copy protocol implementation using Paramiko
  • SSH key management - Best practices for generating and managing SSH keys
  • Port forwarding - Local and remote port forwarding with Paramiko