Paramikoとは？PythonでSSH接続・コマンド実行・SFTP転送を行う使い方【5.0対応】

Paramiko（パラミコ）は、SSHv2プロトコルをPythonだけで実装したライブラリです。pip install paramiko で入り、Pythonのコードからリモートサーバーへ接続してコマンドを実行したり、SFTPでファイルをやり取りしたりできます。2026年10月時点の最新版は5.0.0（2026年5月9日リリース）で、SHA-1を使う署名と鍵交換が削除されました。

この記事では、macOS上でユーザー権限で起動したOpenSSH 10.3p1のサーバーへ、Python 3.9.6とParamiko 5.0.0（比較用に4.0.0）から接続して確かめた挙動をもとに、接続・認証・コマンド実行・対話シェル・SFTP転送・並列実行・ポート転送の書き方と、詰まりやすい箇所を説明します。SSHそのものの仕組みはSSH接続の仕組みと公開鍵認証の記事で扱っています。

まとめ

• Paramikoは、クライアントとサーバーの両方の機能を持つSSHv2のPython実装です。ライセンスはLGPL-2.1で、Fabricの土台になっています。
• 最新は5.0.0で、Python 3.9以上が必要です。4.0.0でDSA鍵、5.0.0でSHA-1のRSA署名・SHA-1の鍵交換・GSSAPIが削除されました。PKey.from_path() の引数名も passphrase から password に変わっています。
• 接続は SSHClient を作り、known_hostsを読んでから connect() します。既定の RejectPolicy は未登録のホストを拒否します。AutoAddPolicy は初回接続のなりすましを検知できないので、本番では使いません。
• コマンドの標準出力と標準エラーは並行して読み切り、そのあとで recv_exit_status() を呼びます。未読の出力が既定ウィンドウ（2MB）を埋めると処理が止まります。
• プロンプトへの応答が要る操作は invoke_shell() で行います。出力にはプロンプトや制御文字が混ざるので、目印の文字列まで読んで区切ります。
• SFTPには、ディレクトリを丸ごと送るメソッドがありません。mkdir() と put() を組み合わせた関数を自分で書きます。get() は取得元が無いと既存のローカルファイルを0バイトにするので、一時ファイル経由で取得します。
• 複数台へ並列に接続するときは、同じサーバーや踏み台へ接続が集中する場合に限り、8並列を検証開始時の目安とし、接続先ごとの制限に合わせて調整します。OpenSSHの既定（MaxStartups 10:30:100）では、認証前の接続が10本に達すると、新しい接続が30%の確率で拒否され始めます。
• 5.0.0へ上げたあと古い機器へつながらなくなったら、まずサーバー側の鍵とアルゴリズムを更新します。更新できない機器に限り、Paramikoを4.x（DSA鍵なら3.x）に固定します。

Paramikoの位置づけと現行版

公式ドキュメントは、Paramikoを「SSHv2プロトコルのpure-Python実装で、クライアントとサーバーの両方の機能を持つ」と説明しています。OpenSSHの ssh コマンドを裏で呼ぶのではなく、通信をPythonで組み立て、暗号処理は cryptography に任せます。

公式サイトは、リモートでコマンドを実行する、ファイルを転送するといった一般的な用途には、Paramikoの上に作られたFabricを勧めています。Paramikoを直接使う対象は、低水準の操作が必要な人と、Pythonの中でSSHサーバーを動かしたい人です。ほかのライブラリとの選び方は後半の章で扱います。

pip install paramikoでのインストールと5.0の破壊的変更

Paramikoのインストールとバージョン確認

仮想環境を作ってからpipで入れます。依存パッケージの4つも一緒に入ります。

python3 -m venv .venv
. .venv/bin/activate
python -m pip install paramiko
python -c "import paramiko; print(paramiko.__version__)"
# 5.0.0

Python 3.6〜3.8の環境では、pipが選べる版は3.5.1までです（3.5以前のPythonはさらに古い系列になります）。import paramiko の直後に AttributeError: module 'paramiko' has no attribute ... が出るときは、古いサンプルコードやライブラリが削除済みのクラスを呼んでいることを疑います。代表例は paramiko.DSSKey で、5.0.0では実際にこのエラーになります。後述のsshtunnel 0.4.0もこの原因で止まります。

4.0.0と5.0.0で削除・変更された機能

Paramiko公式の変更履歴によると、4.0.0と5.0.0では次の機能が削除・変更されています。

5.0.0で削除された鍵交換は、diffie-hellman-group-exchange-sha1・diffie-hellman-group14-sha1・diffie-hellman-group1-sha1 の3方式です。残った diffie-hellman-group-exchange-sha256 も、受け入れる法の最小サイズが1024ビットから2048ビットに引き上げられました。Paramiko 5.0.0の既定の鍵交換は [email protected] を先頭にした7方式、ホスト鍵は ssh-ed25519 を先頭にした6方式で、SHA-1を使うものは含まれません。鍵の種類の選び方はECDSAとRSAの鍵長・安全性の比較を参照してください。

SSHClientでの接続とexec_commandによるコマンド実行

最小の形は次の8行です。SSHClient を作り、known_hostsを読み、connect() で認証し、exec_command() で実行します。with 文で書くと、ブロックを抜けたときに接続が閉じます。

import paramiko

with paramiko.SSHClient() as client:
    client.load_system_host_keys()  # ~/.ssh/known_hosts を読む
    client.connect("server.example.com", username="deploy",
                   key_filename="/home/deploy/.ssh/id_ed25519", timeout=10)
    stdin, stdout, stderr = client.exec_command("hostname")
    print(stdout.read().decode(), stdout.channel.recv_exit_status())

ローカルのサーバーではホスト名と終了コード0が表示され、with を抜けたあとの client.get_transport() は None でした。この形で済むのは出力が小さいコマンドだけで、出力が大きい場合は後の節の run() を使います。

公開鍵・暗号化鍵・パスワードの渡し方

以降のコードは用途別の部分例です。認証の例は、ホスト鍵を読み込んだ未接続の SSHClient に対して使います。コマンド実行・対話シェル・SFTP・ポート転送の例は、接続済みの client を使い、接続を閉じる前に実行します。

認証情報の渡し方は4通りあります。暗号化された鍵（パスフレーズ付きのEd25519鍵）とPEM形式のRSA鍵を作り、パスワード認証を無効にしたサーバーで試した結果は次のとおりです。

パスフレーズを誤ったときの例外は AuthenticationException ではなく、その親の SSHException で、メッセージは Invalid key だけです。認証の失敗だけを捕まえる書き方では、この誤りを拾えません。password= がパスフレーズを兼ねるのは、SSHClientの公式リファレンスに「passphraseが無いときは秘密鍵の復号にも使う」と書かれている仕様です。パスワード認証とパスフレーズを取り違えないよう、鍵の復号には passphrase= を使います。

鍵をファイルから読み込んでオブジェクトとして渡す場合は、鍵の種類を自動で判定する PKey.from_path() を使います。5.0.0で引数名が passphrase から password に変わり、4.x向けに from_path(path, passphrase=...) と書いたコードは TypeError: from_path() got an unexpected keyword argument 'passphrase' で止まりました。第2引数は位置で渡すと版の違いを吸収できます。公式リファレンスの型表記は str ですが、5.0.0で文字列を渡すと TypeError: password must be bytes になり、bytesにすると読み込めました。connect() の passphrase= は文字列のままで動きます。

key = paramiko.PKey.from_path("/home/deploy/.ssh/id_ed25519", "パスフレーズ".encode("utf-8"))
print(key.get_name(), key.fingerprint)  # ssh-ed25519 SHA256:...
client.connect("server.example.com", username="deploy", pkey=key,
               allow_agent=False, look_for_keys=False)

key_filename も pkey も省くと、ParamikoはSSHエージェントと ~/.ssh/ 配下の鍵ファイルを順に試します。どの鍵で入ったかを固定したいときは、上のように allow_agent=False と look_for_keys=False を渡して探索を止めます。パスワード認証しか使えないサーバーでは password= を渡しますが、パスワードはコードに書かず、環境変数やシークレット管理の仕組みから読み込みます。

exec_commandの戻り値と終了コード

exec_command() は、標準入力・標準出力・標準エラーの3つのファイル風オブジェクトを返します。コマンドが0以外の終了コードで終わっても例外にはなりません（接続や実行要求そのものが失敗したときは例外になります）。コマンドの成否は stdout.channel.recv_exit_status() の戻り値で判定します。echo out; echo err 1>&2; exit 3 を実行すると、出力は out、エラー出力は err、終了コードは3として別々に受け取れました。

exec_command() は呼ぶたびに新しいチャネルを開くため、cd や環境変数の設定は次の呼び出しへ引き継がれません。cd /tmp を実行した次の呼び出しで pwd を実行すると、ホームディレクトリが返りました。同じシェルの状態で続けたい処理は、cd /srv/app && git pull のように1回のコマンドにまとめます。

未読の出力による停止の条件と並行読み取り

公式のChannelの説明には、リモートの出力がウィンドウサイズを超える場合、十分な量を読む前に recv_exit_status() を呼ぶと無期限に止まる可能性があると書かれています。既定のウィンドウサイズは2,097,152バイト（2MB）で、標準出力と標準エラーはこのウィンドウを共有します。

実測では、出力を読まずに recv_exit_status() を先に呼んだ場合、2,000,000バイトの出力では戻りましたが、3,000,000バイトでは戻りませんでした。stdout.read() を先に呼んでも安全とは限りません。標準エラーへ3,000,000バイトを書くコマンドで標準出力だけを先に読むと、標準出力の終わりが届かず止まりました。未読のデータがウィンドウを埋めるとリモート側は書き込みを待ち、手元は続きを待つので、どちらも進まなくなります。

小さな出力では起きないため、ログの一括取得などを本番で流したときに初めて発覚します。次の run() のように標準エラーを別スレッドで読みながら標準出力を読み、両方を読み切ってから終了コードを取ります。

from concurrent.futures import ThreadPoolExecutor


def run(client, command, timeout=30):
    stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
    stdin.close()
    with ThreadPoolExecutor(max_workers=1) as pool:
        err_future = pool.submit(stderr.read)  # 標準エラーは別スレッドで読む
        out = stdout.read()
        err = err_future.result()
    status = stdout.channel.recv_exit_status()  # 読み切ってから終了コードを取る
    return status, out.decode(), err.decode()


status, out, err = run(client, "uname -s; df -h /")

標準エラーを分けて受け取る必要がなければ、stdout.channel.set_combine_stderr(True) で1本にまとめ、stdout.read() だけで読み切る方法もあります。

get_pty・environment・timeoutの挙動

get_pty=True は sudo が端末を要求する場面で使いますが、標準エラーを分けて受け取れなくなります。environment={"FOO": "bar"} は、OpenSSHの既定設定では AcceptEnv に無い変数として黙って捨てられ、例外も出ません。確実に渡すなら FOO=bar command のようにコマンド文字列へ含めます。timeout=2 で sleep 5 を実行すると、約2.1秒で socket.timeout になりました。この値は受信を1回待つ時間の上限で、データが少しずつ届き続ける限り、read() 全体の所要時間は指定秒数を超えます。Python 3.10以降では socket.timeout は TimeoutError の別名なので、どちらで捕まえても同じです。

invoke_shellによる対話シェルの操作と出力の区切り

invoke_shell() は、端末（pty）付きのシェルを1本開きます。exec_command() と違って同じシェルが続くため、cd や export の結果が次のコマンドへ引き継がれます。プロンプトに応答しながら進める操作や、ログイン直後にメニューが出るネットワーク機器に使います。

import re
import time
import uuid

chan = client.invoke_shell(term="vt100", width=200, height=50)


def run_shell(chan, command, timeout=10):
    marker = "__END_" + uuid.uuid4().hex + "__"  # 出力と衝突しない目印を毎回作る
    end_re = re.compile(re.escape(marker.encode()) + rb"(\d+)\r?\n")
    chan.sendall(command + "; echo " + marker + "$?\n")  # 終了コードを目印に付ける
    buf = b""
    end = time.time() + timeout
    while time.time() < end:
        if chan.recv_ready():
            buf += chan.recv(65535)
            m = end_re.search(buf)
            if m:
                return int(m.group(1)), buf[:m.start()]
        else:
            time.sleep(0.05)
    raise TimeoutError(f"目印が届きませんでした: {command}")


run_shell(chan, "cd /tmp && export FOO=bar")       # (0, ...)
status, out = run_shell(chan, "pwd && echo $FOO")  # 0, .../tmp\r\nbar\r\n
run_shell(chan, "false")                              # (1, ...)
chan.sendall("exit\n")
print(chan.recv_exit_status())  # 0

手元のzshで試すと、受け取ったバイト列には送ったコマンドのエコー、\x1b[?2004l のようなエスケープシーケンス、プロンプトが混ざり、改行は \r\n でした。目的の出力 /tmp と bar はその後ろにあります。recv() は届いた分だけ返すので、1回の呼び出しでコマンドの出力が揃う保証はありません。上のように、コマンドの直後に終了コード付きの目印を出し、それが届くまで読み続けて区切ります。目印を毎回ランダムにしているのは、コマンドの出力に同じ文字列が含まれていても誤って区切らないためです。エコーされたコマンド文字列の中の目印は $? のままなので、数字を要求する正規表現には一致しません。

目印の数字は ; の直前までのコマンド全体の終了コードで、false では1、ls /nope でも1が返りました。目印が時間内に届かなければ例外にして、成功と区別します。タイムアウトしたシェルには前のコマンドの出力が後から届くので、そのチャネルは閉じて開き直します。プロンプトの形はシェルや機器ごとに違い、解析の手間は exec_command() より大きくなるので、対話が要る場面だけ invoke_shell() にします。ルーターやスイッチのCLIなら、プロンプト待ちを実装済みのNetmikoのほうが短く書けます。

set_missing_host_key_policyの選び方とknown_hostsの管理

set_missing_host_key_policy() は、手元のknown_hostsに無いサーバーへ接続したときの扱いを決めます。何も指定しなければ RejectPolicy です。3つのポリシーを同じサーバーで試した結果は次のとおりです。

RejectPolicy での拒否は Server '[127.0.0.1]:2222' not found in known_hosts というメッセージになります。ネット上のサンプルの多くは、このエラーを消すために AutoAddPolicy を付けています。

AutoAddPolicyを本番で使わない理由

AutoAddPolicy は、初めて見たホスト鍵を無条件で信用します。最初の接続が偽のサーバーへ誘導されていても気付けません。一方、known_hostsに登録済みの鍵と食い違うときは AutoAddPolicy を指定していても BadHostKeyException で止まることを確認しました。守れるのは2回目以降だけです。

保存先にも注意が要ります。load_system_host_keys() だけを呼んで AutoAddPolicy を使うと、鍵はメモリ上に追加されるだけでknown_hostsには書かれませんでした。load_host_keys(ファイル名) で読んだファイルがある場合だけ、そこへ書き戻されます。毎回「初回接続」になるので、検証の効果はさらに薄れます。

自動化で使うサーバーは、鍵を事前にknown_hostsへ登録し、既定の RejectPolicy のまま接続します。ssh-keyscan は相手を検証しないので、取った鍵はサーバー管理者から受け取ったフィンガープリントと照合してから登録します。

ssh-keyscan -t ed25519 server.example.com > host.pub
ssh-keygen -lf host.pub          # 管理者から受け取った値と照合する
cat host.pub >> ~/.ssh/known_hosts

AutoAddPolicy を使ってよいのは、使い捨ての検証用VMのように、なりすましを気にしない場面だけです。WarningPolicy も未登録の鍵を検証せずに接続を通す点は同じで、違うのは鍵をどこにも記録しないことです。次回もまた未登録として扱われ、警告が出るだけで止まりません。

SFTPによるファイルのアップロード・ダウンロード

open_sftp() で SFTPClient を開き、put() でアップロード、get() でダウンロードします。どちらも callback に関数を渡すと、転送済みバイト数と総バイト数が繰り返し渡されます。次のコードは接続済みの client を使う部分なので、接続を閉じる前に置きます。SFTPというプロトコル自体の仕組みやFTPSとの違いはSFTPの仕組みとFTP/FTPSとの違いの記事で扱っています。

import os

sftp = client.open_sftp()

def progress(done, total):
    print(f"{done}/{total}")

attrs = sftp.put("backup.tar.gz", "/srv/backup/backup.tar.gz", callback=progress)
print(attrs.st_size)  # 転送後にstatで確かめたサイズ

local = "app.log"
tmp = local + ".part"
try:
    sftp.get("/var/log/app/app.log", tmp)
    os.replace(tmp, local)
except Exception:
    if os.path.exists(tmp):
        os.remove(tmp)
    raise
finally:
    sftp.close()

putとgetの引数の違い

put() の confirm=True（既定）は、転送後にリモートのファイルを stat() してサイズを確かめ、結果を SFTPAttributes として返します。3,000,000バイトのファイルでは、コールバックが92回呼ばれ、最後の値は (3000000, 3000000) でした。get() には confirm が無い代わりに、先読みを制御する prefetch と max_concurrent_prefetch_requests があります（SFTPClientの公式リファレンス）。保存先のディレクトリが無いと、put() は FileNotFoundError になります。パーミッションは転送後に chmod(パス, 0o600) で設定できます。

取得元が無いときのgetによる保存先の切り詰め

取得元のファイルが無いとき、get() は FileNotFoundError を出しますが、その前にローカル側のファイルを作成しています。中身が入った既存ファイルを保存先に指定すると、例外のあとにそのファイルは0バイトになっていました。前回取得したログや設定ファイルを上書き指定で取り直す処理では、リモートの削除や名前変更が起きた時点で手元の正常なファイルが消えます。上のコードのように .part の一時ファイルへ取得し、成功してから os.replace() で置き換えると、既存ファイルは残ります。

ディレクトリの作成・一覧と再帰アップロード

この例には接続中の SFTPClient が必要です。前の転送例は最後に sftp.close() を呼んでいるので、続けて実行する場合は sftp = client.open_sftp() で開き直し、終わったら例外時も含めて sftp.close() で閉じます。

Paramiko 5.0.0の SFTPClient が持つディレクトリ関連のメソッドは、mkdir・rmdir・listdir・listdir_attr・listdir_iter・chdir・getcwd などで、ディレクトリを丸ごと転送するメソッドはありません。中身を見ながら1件ずつ put() する関数を自分で書きます。

import os
import stat


def put_dir(sftp, local_dir, remote_dir):
    try:
        sftp.mkdir(remote_dir)
    except OSError:
        # 既にあるときも理由の無いOSErrorになるので、statでディレクトリか確かめる
        if not stat.S_ISDIR(sftp.stat(remote_dir).st_mode):
            raise
    for name in os.listdir(local_dir):
        local_path = os.path.join(local_dir, name)
        remote_path = remote_dir + "/" + name
        if os.path.islink(local_path):
            raise ValueError(f"シンボリックリンクは転送対象外です: {local_path}")
        if os.path.isdir(local_path):
            put_dir(sftp, local_path, remote_path)
        else:
            sftp.put(local_path, remote_path)


put_dir(sftp, "dist", "/srv/app/dist")
for a in sftp.listdir_attr("/srv/app/dist"):
    print(a.filename, stat.S_ISDIR(a.st_mode), a.st_size)

この関数を書くときに引っかかる挙動を、OpenSSHのsftp-serverで確かめました。

mkdir() は、既に存在する場合は理由の付かない Failure（errnoなし）になり、権限不足なら PermissionError になります。Failure だけでは既存のディレクトリかどうか判断できないので、上の関数では stat() で確かめています。rename() は標準のSFTP RENAMEを使い、公式リファレンスでも変更先が既に存在してはならないと規定されています。アップロードした一時ファイルを本番の名前へ差し替えるなど、既存のファイルを置き換える処理には posix_rename() を使います。これはOpenSSHのPOSIX rename拡張を使うメソッドで、拡張に対応していないサーバーでは例外になります。listdir_attr() の結果は順不同なので、並びが必要なら自分でソートします。

SFTPと scp コマンドの違い、コマンドラインでの転送方法はSCPコマンドの基本構文と主要オプションの記事にまとめています。

複数サーバーへの並列実行と接続数の上限

Paramikoは同期APIなので、複数台へ同時に処理するときはスレッドプールを使い、スレッドごとに別の SSHClient を作ります。1秒かかるコマンド（sleep 1）を10回実行したところ、1本ずつでは11.7秒、5並列では2.4秒でした。

from concurrent.futures import ThreadPoolExecutor, as_completed

import paramiko

HOSTS = ["web01", "web02", "web03"]  # 実際は数十台


def check(host):
    with paramiko.SSHClient() as c:  # スレッドごとに接続を作る
        c.load_system_host_keys()
        c.connect(host, username="deploy", timeout=10, banner_timeout=30)
        status, out, err = run(c, "uptime")  # run()は前の章で定義
        return host, status, out.strip()


with ThreadPoolExecutor(max_workers=8) as pool:
    futures = {pool.submit(check, h): h for h in HOSTS}
    for f in as_completed(futures):
        try:
            print(f.result())
        except Exception as e:
            print(futures[f], type(e).__name__, e)

スレッド数を増やせば速くなるわけではありません。同じサーバーへ40本を一度に接続したところ、11本が SSHException: Error reading SSH protocol banner[Errno 54] Connection reset by peer で失敗しました。サーバーのログには drop connection ... Maxstartups が残っていました。sshd_configのマニュアルによると、MaxStartups の既定値は 10:30:100 で、認証前の接続が10本に達すると30%の確率で新しい接続を拒否し、その確率は100本で100%になるまで上がります。同じ40本を8並列で流すと、すべて成功し、5.8秒で終わりました。

スレッドごとに踏み台へ接続し直す書き方では、その接続がすべて踏み台の MaxStartups に数えられます。並列数は8前後から始め、増やすなら相手のsshd設定を確認してからにします。失敗した接続は例外を記録して、あとで再実行します。

1本の接続の中でチャネルを増やす方法にも上限があります。MaxSessions の既定は10で、1つの接続で11本目のセッションを開こうとすると ChannelException(2, 'Connect failed') になりました。数百台規模や長時間の同時接続が必要なら、後述のAsyncSSHを検討します。社内の運用作業をこうしたスクリプトで自動化して引き継ぐ場合は、保守運用・内製化支援のように、運用手順ごと整える方法もあります。

踏み台経由の接続・ポート転送・keepalive

~/.ssh/configの読み込みと踏み台経由の接続

Paramikoは ~/.ssh/config を自動では読みません。公式ドキュメントも、OpenSSH形式の設定ディレクティブをParamiko自身は使わないと明記しています。ssh コマンドでは通る Host の別名、ProxyJump、IdentityFile は、SSHConfig で読み込んで自分で connect() の引数へ渡します。

import os
import paramiko

config = paramiko.SSHConfig.from_path(os.path.expanduser("~/.ssh/config"))
b = config.lookup("bastion")  # HostName・Port・User・IdentityFile を辞書で返す

with paramiko.SSHClient() as jump, paramiko.SSHClient() as target:
    jump.load_system_host_keys()
    jump.connect(b["hostname"], port=int(b.get("port", 22)),
                 username=b["user"], key_filename=b["identityfile"][0])

    # 踏み台から先へのTCP転送チャネルを、次の接続のソケットとして使う
    chan = jump.get_transport().open_channel(
        "direct-tcpip", ("10.0.1.20", 22), ("127.0.0.1", 0))

    target.load_system_host_keys()
    target.connect("10.0.1.20", username="deploy",
                   key_filename="/home/deploy/.ssh/id_ed25519", sock=chan)
    target.get_transport().set_keepalive(30)  # 30秒送信が無ければkeepaliveを送る
    status, out, err = run(target, "hostname")  # run()は接続の章で定義

lookup() は proxyjump の値を辞書に入れて返すだけで、踏み台への接続はしません。connect() の sock 引数に踏み台のチャネルを渡すと、2段目の接続がそのチャネルを通ります。ローカルの検証環境で、この構成で2段目のコマンドが実行できることを確認しました。設定に ProxyCommand がある場合は、paramiko.ProxyCommand で包んだオブジェクトを sock に渡します。踏み台サーバーの設計そのものは踏み台サーバーとSSH多段接続の記事で解説しています。

無通信の時間が長い処理では、途中のファイアウォールやNATがセッションを切ることがあるため、Transport.set_keepalive(秒) を設定します。指定した秒数だけ何も送信しなかったときに、keepaliveのパケットが送られます。

direct-tcpipによるローカルポート転送とsshtunnel

踏み台の先にあるデータベースや管理画面へ手元から接続するには、ssh -L と同じローカルポート転送を組みます。上の踏み台の例と同じ open_channel("direct-tcpip", ...) を、手元で待ち受けたソケットとつなぎます。

import select
import socket
import threading

transport = client.get_transport()
REMOTE = ("127.0.0.1", 5432)  # SSHサーバーから見た転送先


def pipe(sock):
    chan = None
    try:
        chan = transport.open_channel("direct-tcpip", REMOTE, sock.getpeername())
        while True:
            r, _, _ = select.select([sock, chan], [], [])
            if sock in r:
                data = sock.recv(32768)
                if not data:
                    break
                chan.sendall(data)
            if chan in r:
                data = chan.recv(32768)
                if not data:
                    break
                sock.sendall(data)
    finally:  # 転送先へつながらないときも両端を閉じる
        if chan is not None:
            chan.close()
        sock.close()


with socket.socket() as server:
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("127.0.0.1", 15432))  # 手元の待ち受けポート
    server.listen()
    while True:
        s, _ = server.accept()
        threading.Thread(target=pipe, args=(s,), daemon=True).start()

検証では、SSHサーバー側の127.0.0.1で動かしたHTTPサーバーへこの方法で転送し、手元の http://127.0.0.1:15432/ から応答を受け取れました。転送先のポートが閉じていると、open_channel() が ChannelException(2, 'Connect failed') を出します。サーバー側の AllowTcpForwarding が無効でも転送はできません。

以前は同じ処理の見本として公式リポジトリの demos/forward.py が使われていましたが、5.0.0で demos/ フォルダごと削除されました。転送だけが目的なら、Paramikoの上に作られたsshtunnelを使う方法もあります。ただし版に注意が要ります。sshtunnel 0.4.0（2021年）はParamiko 5.0.0と組み合わせると、start() の中で AttributeError: module 'paramiko' has no attribute 'DSSKey' になりました。2026年10月9日に公開された0.5.1は変更履歴に「DSSKeyを削除したparamiko>=4.0に対応」とあり、同じ環境で転送できました。

接続・実行エラーの例外別の切り分け

ローカルの検証環境でそれぞれの失敗を起こしたときの例外は次のとおりです。paramiko.SSHException を1つ捕まえるだけでは原因を区別できないので、型とメッセージで分けて扱います。

BadAuthenticationType と PasswordRequiredException は AuthenticationException のサブクラスです。個別に処理する場合は、これらの except 節を AuthenticationException の節より先に書きます。まとめて処理するだけなら、AuthenticationException だけで捕まえられます。NoValidConnectionsError は SSHException ではなく OSError のサブクラスです。SSHException だけを捕まえる書き方では、接続先に到達できない失敗が漏れます。

5.0.0と旧式サーバーの互換性エラーへの対処

ホスト鍵を ssh-rsa、鍵交換を diffie-hellman-group14-sha1 だけに絞ったOpenSSHサーバーを立てて試しました。Paramiko 5.0.0は IncompatiblePeer: Incompatible ssh peer (no acceptable kex algorithm) で接続できず、4.0.0は同じサーバーに接続できました。鍵交換だけをcurve25519にして ssh-rsa のホスト鍵を残した場合、5.0.0のメッセージは (no acceptable host key) に変わります。括弧内を見れば、鍵交換とホスト鍵のどちらが足りないかがわかります。

まずサーバーやネットワーク機器のファームウェアを更新し、rsa-sha2-256/512やEd25519を使えるようにします。OpenSSHの版ごとの対応状況はOpenSSHの最新バージョンとアップデート手順で確認できます。更新できない機器が残る場合に限って、その機器へ接続するスクリプトだけ版を固定します。

pip install "paramiko<5"   # SHA-1のRSA署名・鍵交換が必要な機器
pip install "paramiko<4"   # DSA鍵しか持たない機器

固定した環境はSHA-1削除の恩恵を受けられないので、対象機器と更新期限を決めてから使います。

AsyncSSH・Netmiko・Fabricとの使い分け

PythonでSSHを扱うライブラリは、Paramikoの上に作られたものと、別実装のものに分かれます。PyPIの公開情報（2026年10月時点）は次のとおりです。Netmiko 4.8.0とAsyncSSH 2.24.1はPython 3.10以上が必要なので、Python 3.9の環境から移る場合はPythonも更新します。

• サーバー数台へコマンドを流して結果を集めるだけなら、Paramikoを直接書かずにFabricを使います。公式サイトもこの用途にはFabricを勧めています。
• ネットワーク機器のCLIを操作するなら、プロンプト待ちやページャー処理を持つNetmikoを使います。Netmikoには、Paramikoを4.x系に固定してインストールする netmiko[par4] という指定も用意されています。
• 数百台以上へ同時に接続するなら、AsyncSSHを選びます。AsyncSSHはasyncioの上で、1スレッドのまま多数の接続を扱えます。ただし相手のsshdの MaxStartups は同じように効くので、同時接続数の制御は必要です。
• サーバーの状態を宣言的にそろえる構成管理なら、SSHライブラリで手続きを書くよりAnsibleが向きます。

Paramikoを直接使うのは、SFTPの転送を細かく制御する、対話シェルを操作する、ポート転送をアプリケーションに組み込む、テスト用のSSHサーバーをPythonで立てる、といったチャネルやトランスポートを自分で扱う場合です。新規実装で必要な操作をFabric・Netmiko・AsyncSSHだけで表現でき、既存コードや依存関係の制約もなければ、それらを優先します。

よくある質問

Paramikoの読み方は？

「パラミコ」と読みます。PythonからSSH接続を扱うためのライブラリで、pip install paramiko でインストールし、import paramiko で読み込みます。

set_missing_host_key_policyにAutoAddPolicyを指定しても問題ありませんか？

本番の自動化では避けます。初回接続で受け取った鍵を検証せずに信用するため、その時点のなりすましを検知できません。known_hostsへ事前に登録して、既定の RejectPolicy のまま接続します。

ParamikoのSFTPとscpコマンドはどう違いますか？

ParamikoのSFTPはPythonの中からファイルを転送する方法で、進捗の取得や例外処理をコードで書けます。scp はOpenSSHのコマンドで、シェルスクリプトから1行で転送するときに使います。Paramiko本体には scp の実装がなく、転送は open_sftp() で行います。

Paramiko 5.0にしたらssh-rsaのサーバーへ接続できなくなりました。なぜですか？

5.0.0でSHA-1によるRSA署名（ssh-rsa）とSHA-1の鍵交換が削除されたためです。例外の括弧内が no acceptable host key ならホスト鍵、no acceptable kex algorithm なら鍵交換が原因です。サーバー側を更新できない場合だけ paramiko<5 に固定します。

asyncioで使うならParamikoとAsyncSSHのどちらがよいですか？

AsyncSSHです。Paramikoは同期APIで、asyncioから使うにはスレッドへ逃がす処理が必要になります。多数のサーバーへ同時に接続する処理は、asyncio上に作られたAsyncSSHのほうが素直に書けます。

関連記事

• SSH接続とは？仕組み・公開鍵認証・ポート22から安全な運用設定まで実装者向けに解説
• SFTPとは？仕組み・FTP/FTPSとの違い・sftpコマンドとchroot設定を実装者向けに解説【2026年版】
• SCPコマンドとは？基本構文・主要オプション・SSH越しの安全なファイル転送を実装者向けに解説【2026年版】
• 踏み台サーバーとは？仕組み・SSH多段接続・AWS構築と採用判断【2026年版】
• OpenSSHとは？最新バージョン10.6の確認方法とアップデート手順【2026年10月】