2026年10月14日

PythonでSSH接続とコマンド実行（Paramiko）｜鍵認証・ホスト鍵検証・終了コードの扱い

Linux基礎・コマンド

Python から SSH でサーバーにログインしてコマンドを実行するなら、Paramiko の SSHClient を使い、鍵認証で接続して exec_command() で実行し、終了コードで成否を判定するのが基本です。この記事では、ホスト鍵を正しく検証する接続方法、出力と終了コードの受け取り方、SFTP でのファイル転送、複数サーバーへの実行とエラー処理までを、そのまま使えるコードで説明します。

Paramiko を使う場面と準備

Paramiko は SSH プロトコルを Python で実装したライブラリで、外部の ssh コマンドに頼らずに接続・コマンド実行・ファイル転送ができます。サーバーの状態をまとめて確認するスクリプトや、WordPress を置いたサーバーで WP-CLI を定期的に実行する処理などに向いています。

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

この記事のコードは Paramiko 3 系以降を前提にしています（2026年10月現在）。Paramiko 4.0 では古い DSA（ssh-dss）鍵のサポートが削除されているため、これから鍵を作るなら Ed25519 を選んでください。ライブラリのバージョンによって引数や対応する鍵の種類が変わることがあるので、更新したときは公式ドキュメントの変更履歴を確認します。

接続先には、あらかじめ公開鍵を登録して、ターミナルから ssh -i ~/.ssh/id_ed25519 user@host でログインできる状態にしておきます。Python 側で問題が出たときに、SSH の設定の問題かスクリプトの問題かを切り分けやすくなるためです。

最小の接続とコマンド実行

まずは1台に接続して uptime を実行する例です。

import paramiko

HOST = "203.0.113.10"   # 例示用のアドレス。自分のサーバーに置き換える
USER = "deploy"
KEY = "/home/me/.ssh/id_ed25519"

with paramiko.SSHClient() as client:
    client.load_system_host_keys()                       # ~/.ssh/known_hosts を読む
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    client.connect(HOST, port=22, username=USER, key_filename=KEY, timeout=10)

    stdin, stdout, stderr = client.exec_command("uptime", timeout=30)
    out = stdout.read().decode("utf-8", errors="replace")
    err = stderr.read().decode("utf-8", errors="replace")
    status = stdout.channel.recv_exit_status()

print("exit:", status)
print(out)
if err:
    print("stderr:", err)

with を使うと、途中で例外が起きても接続が閉じられます。timeout=10 は接続時のタイムアウト、exec_command() の timeout=30 はコマンド実行中の読み取りのタイムアウトです。どちらも指定しないと、応答のないサーバーでスクリプトが止まったままになることがあります。

ホスト鍵の検証を省略しない

Paramiko の解説では set_missing_host_key_policy(paramiko.AutoAddPolicy()) がよく紹介されています。これは「初めて見るサーバーの鍵を確認せずに受け入れる」設定で、接続先がすり替えられていても気付けません。テスト用の使い捨て環境以外では避けてください。

安全な手順は次のとおりです。

一度ターミナルから ssh user@host で接続し、表示されたフィンガープリントを、サーバー管理画面やサーバー上の ssh-keygen -lf /etc/ssh/ssh_host_ed25519_key.pub の結果と照合する
一致したら接続を続け、~/.ssh/known_hosts に登録させる
Python 側は load_system_host_keys() でその known_hosts を読み、RejectPolicy() で未登録のサーバーへの接続を拒否する

鍵が変わっていた場合は paramiko.BadHostKeyException が発生します。サーバーを再構築したなど理由が分かっている場合だけ known_hosts を更新し、心当たりがなければ接続を止めて調べます。

22番以外のポートで接続する場合、known_hosts には [203.0.113.10]:2222 のような形式で登録されます。ポートを変えたサーバーで「鍵が見つからない」と言われるときは、この形式で登録されているかを見てください。

出力と終了コードを正しく受け取る

exec_command() は stdin・stdout・stderr の3つを返します。成否の判定には文字列ではなく終了コード（recv_exit_status()）を使います。0 なら成功、それ以外は失敗です。

注意点は読み取る順番です。出力が大きいコマンドで、出力を読まずに先に recv_exit_status() を待つと、送受信のバッファが埋まって処理が進まなくなることがあります。先に stdout.read() で読み切ってから終了コードを取るのが安全です。

複数の処理を使い回せるように関数にまとめます。

import shlex
import paramiko


def run(client: paramiko.SSHClient, command: str, timeout: int = 60):
    stdin, stdout, stderr = client.exec_command(command, timeout=timeout)
    stdin.close()
    out = stdout.read().decode("utf-8", errors="replace")
    err = stderr.read().decode("utf-8", errors="replace")
    status = stdout.channel.recv_exit_status()
    return status, out, err


# 外から受け取った値をコマンドに入れるときは shlex.quote で囲む
site_path = "/var/www/example"
status, out, err = run(client, f"wp --path={shlex.quote(site_path)} core version")

ファイル名やユーザー入力など外部の値をコマンド文字列に入れる場合は、shlex.quote() で囲みます。そのまま埋め込むと、空白や記号でコマンドが壊れるだけでなく、意図しないコマンドを実行させられる危険があります。

sudo が必要な処理は、パスワード入力を求められると止まってしまいます。パスワードをスクリプトに書いて流し込むのではなく、必要なコマンドだけをサーバー側の sudoers で許可する方法を検討してください。

SFTP でファイルを転送する

同じ接続から SFTP を開くと、ファイルのアップロードとダウンロードができます。

with client.open_sftp() as sftp:
    sftp.put("backup.sh", "/home/deploy/bin/backup.sh")
    sftp.chmod("/home/deploy/bin/backup.sh", 0o750)
    sftp.get("/var/log/nginx/error.log", "error.log")

put() と get() のパスはファイル名まで指定します。転送先にディレクトリだけを指定するとエラーになります。また、サーバー側のパスは ~ が展開されないので、/home/deploy/... のようにフルパスで書きます。

大きなファイルや多数のファイルを同期したい場合は、Paramiko で一つずつ送るより rsync を使ったほうが速く、差分だけを送れます。

複数サーバーで実行してエラーを分ける

運用で使うときは、サーバーごとに失敗の理由を分けて記録すると原因を追いやすくなります。

import socket
import paramiko

hosts = ["203.0.113.10", "203.0.113.11"]

for host in hosts:
    try:
        with paramiko.SSHClient() as client:
            client.load_system_host_keys()
            client.set_missing_host_key_policy(paramiko.RejectPolicy())
            client.connect(host, username="deploy", key_filename=KEY, timeout=10)
            status, out, err = run(client, "df -h /")
            print(host, "exit", status, out.strip().splitlines()[-1])
    except paramiko.AuthenticationException:
        print(host, "認証失敗: ユーザー名と公開鍵の登録を確認")
    except paramiko.BadHostKeyException:
        print(host, "ホスト鍵が known_hosts と一致しない")
    except paramiko.SSHException as e:
        print(host, "SSH エラー:", e)
    except (socket.timeout, OSError) as e:
        print(host, "接続できない:", e)

よく出るエラーと確認する場所は次のとおりです。

AuthenticationException：公開鍵がサーバーの ~/.ssh/authorized_keys に入っているか、ユーザー名が合っているか。サーバー側の認証ログ（/var/log/auth.log や journalctl -u ssh など、OS によって異なる）に理由が出ます。
SSHException（Server not found in known_hosts など）：RejectPolicy で未登録のサーバーを拒否した状態です。上の手順でホスト鍵を登録します。
タイムアウトや接続拒否：ファイアウォール、SSH のポート番号、サーバーが起動しているかを確認します。

接続や認証の流れを詳しく見たいときは、paramiko.util.log_to_file("paramiko.log") でライブラリのログをファイルに出せます。ログにはパスワードやパスフレーズを出力しないよう、自分で書くログにも秘密情報を含めないでください。

パスフレーズと秘密鍵の扱い

秘密鍵にはパスフレーズを付けるのが基本です。スクリプトにパスフレーズを直接書くのは避け、次のどちらかで渡します。

ssh-agent を使う：SSHClient.connect() は既定で ssh-agent を参照します（allow_agent が既定で有効）。事前に ssh-add で鍵を登録しておけば、スクリプトに秘密情報を書かずに済みます。
環境変数で渡す：cron などで動かす場合は、パスフレーズを環境変数から読み、connect() の passphrase 引数に渡します。

スクリプト専用の鍵を作り、サーバー側の authorized_keys で from= による接続元の制限や、command= による実行コマンドの固定をかけると、鍵が漏れたときの影響を小さくできます。秘密鍵のファイルは所有者だけが読める権限（600）にし、Git リポジトリには入れないでください。

まとめ

Paramiko では SSHClient で接続し、exec_command() でコマンドを実行します。with とタイムアウトの指定で、止まったままの接続を防ぎます。
AutoAddPolicy は使わず、known_hosts に照合済みの鍵を登録して RejectPolicy で接続します。
出力を読み切ってから recv_exit_status() で終了コードを取り、成否は終了コードで判定します。外部の値は shlex.quote() で囲みます。
ファイル転送は open_sftp()、大量の同期は rsync が向いています。
秘密鍵はパスフレーズ付きで ssh-agent か環境変数から渡し、スクリプトやリポジトリに書かないようにします。