SSH Automation with Paramiko

Paramiko is a pure-Python library that implements the SSH2 protocol. It lets you connect to SSH servers, execute commands, transfer files, and automate remote server management—all without needing external dependencies like OpenSSH.

Installing Paramiko

Install Paramiko with pip:

pip install paramiko

That is it. No system-level SSH libraries required.

Establishing an SSH Connection

The simplest way to connect to a remote server is with a username and password:

import paramiko

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

client.connect(
    hostname="server.example.com",
    username="deploy",
    password="secret-password"
)

client.close()

The set_missing_host_key_policy() call tells Paramiko to automatically accept unknown host keys. For production use, you would want to load a known hosts file instead:

client.load_system_host_keys()
client.set_missing_host_key_policy(paramiko.RejectPolicy())

This rejects connections to servers whose keys are not already known, protecting against man-in-the-middle attacks.

Connecting with SSH Keys

Password authentication is convenient but key-based authentication is more secure and works better for automation:

import paramiko

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

# Connect using a private key file
client.connect(
    hostname="server.example.com",
    username="deploy",
    key_filename="/home/user/.ssh/id_ed25519"
)

client.close()

Paramiko supports RSA, ECDSA, and Ed25519 keys. You can also load keys directly:

private_key = paramiko.Ed255Key.from_private_key_file("/path/to/key")
client.connect(hostname="...", username="...", pkey=private_key)

Executing Commands

Once connected, execute commands with exec_command():

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname="server.example.com", username="deploy", password="pass")

stdin, stdout, stderr = client.exec_command("uname -a")

# Read the output
output = stdout.read().decode()
error = stderr.read().decode()
exit_code = stdout.channel.recv_exit_status()

print(f"Output: {output}")
print(f"Exit code: {exit_code}")

client.close()

The exec_command() method returns three file-like objects: stdin, stdout, and stderr. The command runs in a non-interactive shell, so environment variables and shell features may not work as expected.

Capturing Real-Time Output

For long-running commands, you might want to see output as it happens:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname="server.example.com", username="deploy", password="pass")

stdin, stdout, stderr = client.exec_command("for i in 1 2 3 4 5; do echo \"Count: $i\"; sleep 1; done")

# Read line by line as output arrives
for line in stdout:
    print(line.rstrip())

client.close()

This pattern is useful for deployment scripts, build processes, or any command that produces output over time.

Working with SFTP

Paramiko includes SFTP for secure file transfers:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname="server.example.com", username="deploy", password="pass")

# Open an SFTP session
sftp = client.open_sftp()

# Download a file
sftp.get("/remote/file.txt", "local_file.txt")

# Upload a file
sftp.put("local_script.py", "/remote/script.py")

# List directory contents
for filename in sftp.listdir("/remote/path"):
    print(filename)

sftp.close()
client.close()

SFTP operations mirror Python file operations closely, making it intuitive for anyone familiar with built-in file handling.

Managing Directories and Files

SFTP provides directory and file management methods:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname="server.example.com", username="deploy", password="pass")

sftp = client.open_sftp()

# Create a directory
sftp.mkdir("/remote/new_directory")

# Remove a directory
sftp.rmdir("/remote/empty_directory")

# Delete a file
sftp.remove("/remote/file_to_delete")

# Rename a file
sftp.rename("/remote/old_name.txt", "/remote/new_name.txt")

# Get file attributes
stat = sftp.stat("/remote/file.txt")
print(f"Size: {stat.st_size}, Permissions: {oct(stat.st_mode)}")

sftp.close()
client.close()

These operations map directly to traditional FTP commands.

Context Manager Pattern

Use context managers to ensure connections close properly, even if errors occur:

import paramiko
from contextlib import contextmanager

@contextmanager
def ssh_connection(hostname, username, password=None, key_filename=None):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        client.connect(
            hostname=hostname,
            username=username,
            password=password,
            key_filename=key_filename
        )
        yield client
    finally:
        client.close()

# Usage
with ssh_connection("server.example.com", "deploy", password="pass") as client:
    stdin, stdout, stderr = client.exec_command("uptime")
    print(stdout.read().decode())

This pattern prevents connection leaks in long-running automation scripts.

Handling Multiple Servers

A common pattern is running the same command across multiple servers:

import paramiko
from concurrent.futures import ThreadPoolExecutor

def run_on_server(server_info, command):
    hostname, username, password = server_info
    
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        client.connect(hostname=hostname, username=username, password=password)
        stdin, stdout, stderr = client.exec_command(command)
        output = stdout.read().decode()
        return f"{hostname}: {output.strip()}"
    finally:
        client.close()

# Run command on multiple servers
servers = [
    ("server1.example.com", "deploy", "password1"),
    ("server2.example.com", "deploy", "password2"),
    ("server3.example.com", "deploy", "password3"),
]

with ThreadPoolExecutor(max_workers=5) as executor:
    results = list(executor.map(lambda s: run_on_server(s, "uptime"), servers))
    
for result in results:
    print(result)

The thread pool lets you run commands in parallel. Paramiko is thread-safe for concurrent connections to different servers.

SSH Tunneling (Port Forwarding)

Paramiko supports SSH tunneling for secure port forwarding:

import paramiko

client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(hostname="jump-server.example.com", username="deploy", password="pass")

# Forward local port 9000 to remote port 5432
transport = client.get_transport()
forwarder = transport.open_channel(
    "direct-tcpip",
    ("localhost", 5432),
    ("127.0.0.1", 9000)
)

forwarder.close()
client.close()

This is useful for accessing services behind a jump server or tunneling database connections securely.

Error Handling

Robust scripts need proper error handling:

import paramiko
from paramiko.ssh_exception import SSHException, AuthenticationException

def safe_ssh_command(hostname, username, password, command):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        client.connect(hostname=hostname, username=username, password=password)
    except AuthenticationException:
        return f"Authentication failed for {hostname}"
    except SSHException as e:
        return f"SSH error for {hostname}: {e}"
    
    try:
        stdin, stdout, stderr = client.exec_command(command)
        exit_code = stdout.channel.recv_exit_status()
        
        if exit_code != 0:
            error = stderr.read().decode()
            return f"Command failed on {hostname}: {error}"
        
        return stdout.read().decode()
    except Exception as e:
        return f"Error executing command: {e}"
    finally:
        client.close()

This handles authentication failures, SSH errors, and command failures separately.

Security Best Practices

When using Paramiko in production:

1. Never hardcode passwords — Use environment variables or a secrets manager
2. Validate host keys — Use RejectPolicy with a known hosts file
3. Use key-based auth — It is more secure than passwords
4. Limit permissions — The SSH user should have only the needed permissions
5. Log access — Track who connected and what commands ran

See Also

• subprocess — Running external commands from Python
• concurrent.futures — Parallel task execution
• Running Shell Commands — Related tutorial on subprocess

Next Steps

You now know how to connect to remote servers, execute commands, transfer files, and handle errors programmatically. Combined with what you learned about subprocess, you can build complete deployment and server management automation.

The next tutorial in this series covers AWS automation with Boto3, which pairs well with Paramiko for infrastructure that spans both servers and cloud services.