Quickstart: SSH and SFTP (Safe Defaults)

import paramiko, socket

def ssh_client(host, user, password=None, pkey_path=None, timeout=8):
    ssh = paramiko.SSHClient()
    ssh.load_system_host_keys()
    ssh.set_missing_host_key_policy(paramiko.RejectPolicy())  # safer default
    try:
        if pkey_path:
            key = paramiko.RSAKey.from_private_key_file(pkey_path)
            ssh.connect(hostname=host, username=user, pkey=key,
                        look_for_keys=False, allow_agent=False, timeout=timeout)
        else:
            ssh.connect(hostname=host, username=user, password=password,
                        look_for_keys=False, allow_agent=False, timeout=timeout)
        return ssh
    except (paramiko.SSHException, socket.error) as e:
        raise RuntimeError(f"SSH connect failed: {host}") from e

def run(host, user, cmd, **kw):
    with ssh_client(host, user, **kw) as ssh:
        stdin, stdout, stderr = ssh.exec_command(cmd, timeout=15)
        rc = stdout.channel.recv_exit_status()
        return rc, stdout.read().decode(), stderr.read().decode()

Key Authentication & Host Keys

# Load key types
from paramiko import RSAKey, ECDSAKey, Ed25519Key
rsa = RSAKey.from_private_key_file("~/.ssh/id_rsa")
ed  = Ed25519Key.from_private_key_file("~/.ssh/id_ed25519")

# Known hosts & strict host key checking
ssh = paramiko.SSHClient()
ssh.load_system_host_keys()  # ~/.ssh/known_hosts
ssh.set_missing_host_key_policy(paramiko.RejectPolicy())

Command Execution Patterns

def run_checked(ssh, cmd, timeout=20):
    stdin, stdout, stderr = ssh.exec_command(cmd, timeout=timeout)
    rc = stdout.channel.recv_exit_status()
    out, err = stdout.read().decode(), stderr.read().decode()
    if rc != 0:
        raise RuntimeError(f"cmd failed rc={rc}: {cmd}\n{err}")
    return out

cmd = "cd /var/log && env FOO=1 BAR=2 ./script.sh"
run_checked(ssh, cmd)

# Use a PTY to handle sudo prompts
chan = ssh.get_transport().open_session()
chan.get_pty()
chan.exec_command("sudo -S ls /root")
chan.send("your_sudo_password\\n")
out = chan.makefile("r").read()

shell = ssh.invoke_shell()
shell.send("uname -a\\n")
resp = shell.recv(65535).decode()

SFTP: Upload, Download, Permissions

def sftp_session(ssh):
    return ssh.open_sftp()

with ssh_client("server", "user", pkey_path="~/.ssh/id_ed25519") as ssh:
    with sftp_session(ssh) as sftp:
        sftp.put("local.bin", "/remote/path/app.bin")   # upload
        sftp.get("/remote/path/log.txt", "log.txt")     # download
        sftp.chdir("/remote/path")
        sftp.mkdir("archive", mode=0o750)
        sftp.chmod("/remote/path/app.bin", 0o750)
        # Atomic upload: write temp then rename
        tmp = "/remote/path/.app.bin.tmp"
        sftp.put("local.bin", tmp)
        sftp.rename(tmp, "/remote/path/app.bin")

Port Forwarding & Tunnels

# Local port forward: localhost:9000 -> remote:5432
from paramiko.forward import forward_tunnel
import threading

def forward(db_host="127.0.0.1", db_port=5432, local_port=9000):
    ssh = ssh_client("server", "user", pkey_path="~/.ssh/id_ed25519")
    t = threading.Thread(target=forward_tunnel,
                         args=(local_port, db_host, db_port, ssh.get_transport()),
                         daemon=True)
    t.start()
    return ssh, t  # keep ssh open while forwarding

Bulk Operations & Concurrency

from concurrent.futures import ThreadPoolExecutor, as_completed

def run_remote(host, user, cmd, pkey_path):
    ssh = ssh_client(host, user, pkey_path=pkey_path)
    try:
        out = run_checked(ssh, cmd)
        return host, 0, out
    except Exception as e:
        return host, 1, str(e)
    finally:
        ssh.close()

hosts = ["srv1","srv2","srv3","srv4"]
with ThreadPoolExecutor(max_workers=8) as ex:
    futures = [ex.submit(run_remote, h, "ops", " ~/.ssh/id_ed25519", "uptime") for h in hosts]
    for fut in as_completed(futures):
        host, rc, msg = fut.result()
        print(host, rc, msg[:120])

Logging & Auditing

import logging, json, sys, time
logging.basicConfig(level=logging.INFO, format="%(message)s")
log = logging.getLogger("paramiko_ops")

def jlog(event, **kw):
    log.info(json.dumps({"ts":time.time(),"event":event,**kw}))

jlog("ssh_connect", host="srv1")
# ...
jlog("cmd", host="srv1", cmd="uptime", rc=0)

Security Hardening Checklist

• Use key authentication; disable password auth on servers where possible.
• Strict host key checking with RejectPolicy; manage known_hosts centrally.
• Always set timeouts (connect, exec, SFTP). Fail fast and retry with backoff.
• Least privilege: dedicate users, limit sudoers, restrict file permissions.
• Rotate keys; protect private keys (chmod 600) and use passphrases or agent.
• Log commands, exit codes, and transfer checksums; retain for audits.
• For jump hosts/bastions, prefer ProxyJump or SSH tunnels rather than exposing ports.

Troubleshooting & Common Errors

# Test manually with OpenSSH for clarity
# ssh -i ~/.ssh/id_ed25519 user@host -vvv

ssh-keyscan -H host.example.com >> ~/.ssh/known_hosts

ssh.connect(..., timeout=8)
ssh.exec_command("cmd", timeout=20)

t = ssh.get_transport()
t.set_keepalive(30)  # seconds

Frequently Asked Questions

How do I run commands that require a TTY (sudo)?

Use get_pty() on a SessionChannel, send the password to stdin, and read both stdout and stderr. Consider NOPASSWD for non‑interactive automation when policy allows.

What’s the best way to deploy files safely?

Upload to a temporary path, verify checksum, set permissions, then rename to the final path (atomic move). Roll back by keeping a backup and swapping names.

How do I go through a bastion/jump host?

Create an SSH connection to the bastion, open a direct-tcpip channel to the target, then create a second Paramiko Transport over that channel; or use local port forwarding and connect the inner session to localhost.