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.
flowchart LR
subgraph Client["Python Application"]
A[SSHClient]
B[SFTPClient]
end
subgraph Connection["SSH Connection"]
C[Authentication]
D[Channel]
E[Transport]
end
subgraph Server["Remote Server"]
F[Shell/Commands]
G[File System]
end
A --> C
C --> E
E --> D
D --> F
A --> B
B --> E
E --> G
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
flowchart TD
A[SSHClient] --> B[open_sftp]
B --> C[SFTPClient]
C --> D[File Operations]
D --> E[get - Download]
D --> F[put - Upload]
D --> G[listdir - List]
D --> H[stat - Attributes]
D --> I[mkdir/rmdir]
D --> J[chmod/chown]
D --> K[remove/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
flowchart LR
subgraph Client
A[Private Key]
B[SSH Agent]
end
subgraph Authentication
C[Key Challenge]
D[Signature Verification]
end
subgraph Server
E[authorised_keys]
F[Public Key]
end
A --> B
B --> C
C --> D
D --> E
E --> F
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