python で ssh でコマンド送信 (pxssh で "su -" を実行する)
概要
python で pexpect モジュールの pxssh を使うことで su - を含めたコマンド送信処理を自動化できる。その手順を説明する。
最終コード
#!/usr/bin/python3
# pxssh を利用した "su -" のテスト
import sys
import re
from pexpect import pxssh
def login_ssh(server: str, username: str, password: str):
# ログイン情報を設定しSSHサーバーにログイン
ssh = pxssh.pxssh(options={
"StrictHostKeyChecking": "no",
"UserKnownHostsFile": "/dev/null"})
# プロンプトの設定 (オプション)
ssh.UNIQUE_PROMPT = r"[\$\#] "
ssh.PROMPT = ssh.UNIQUE_PROMPT
ssh.PROMPT_SET_SH = r"PS1='\$ '"
ssh.PROMPT_SET_CSH = r"set prompt='\$ '"
ssh.PROMPT_SET_ZSH = "prompt restore;\nPS1='%(!.#.$) '"
ssh.login(server=server,
username=username,
password=password,
port=22)
print(ssh.before.decode(encoding='utf-8'), flush=True, end="")
print(ssh.after.decode(encoding='utf-8'), flush=True, end="")
return ssh
def sendline_and_print(ssh, command: str):
print(f"{command}")
ssh.sendline(command)
def run_command(ssh, command: str):
sendline_and_print(ssh, command)
# プロンプトが表示されるまで待つ
while True:
index = ssh.expect_list(
[
re.compile(r".*(#|\$) ".encode()), # 一般ユーザーまたはroot ユーザーのプロンプト
re.compile(r"\n".encode()),
], timeout = 10)
print(ssh.before.decode(encoding='utf-8'), flush=True, end="")
print(ssh.after.decode(encoding='utf-8'), flush=True, end="")
if index == 0:
break
def send_commands(ssh):
run_command(ssh, "touch test.txt")
run_command(ssh, "ls -l")
def switch_to_root(ssh, root_pass: str):
sendline_and_print(ssh, "su -")
# rootのパスワードを入力するプロンプトが表示されるまで待つ
ssh.expect(r".*: ")
print(ssh.before.decode(encoding='utf-8'), flush=True, end="")
print(ssh.after.decode(encoding='utf-8'), flush=True, end="")
ssh.sendline(root_pass)
# root ユーザーのプロンプトが表示されるまで待つ
index = ssh.expect_list(
[
re.compile(r".*# ".encode()), # root ユーザーのプロンプト
re.compile(r"su: .*".encode()), # "su: Authentication failure" or "su: 認証失敗"
], timeout = 10)
if index != 0:
print("failed to switch to root")
sys.exit(1)
print(ssh.before.decode(encoding='utf-8'), flush=True, end="")
print(ssh.after.decode(encoding='utf-8'), flush=True, end="")
def send_commands_root(ssh):
run_command(ssh, "ls -l")
run_command(ssh, "find /etc")
run_command(ssh, "id")
run_command(ssh, "exit")
if __name__ == "__main__":
if len(sys.argv) < 5:
print(f"Usage: python3 {sys.argv[0]} <server> <username> <password> <root_password>")
sys.exit(1)
server = sys.argv[1]
username = sys.argv[2]
password = sys.argv[3]
root_pass = sys.argv[4]
ssh = login_ssh(server, username, password)
send_commands(ssh)
switch_to_root(ssh, root_pass=root_pass)
send_commands_root(ssh)
ssh.logout()
説明
必要モジュールのインポート
from pexpect import pxssh
pxssh のインスタンス化
pxssh をインタスタンス化する。
この際、オプションを指定する。
ssh = pxssh.pxssh(options={
"StrictHostKeyChecking": "no",
"UserKnownHostsFile": "/dev/null"})
ssh 接続する。
ssh.login(server=server,
username=username,
password=password,
port=22)
コマンド送信 & コマンド実行完了待ち
文字列 command で指定したコマンドを ssh 経由で送信してコマンド実行する。
ssh.sendline(command)
ssh.expect_list で指定したパターンのいずれかが表示されるまで待つ。
一般ユーザーまたは rootユーザーのプロンプトが表示されるか、
あるいは改行コードが表示されるまで待つ。
どのパターンに一致したか戻り値で返すので、それに基づいて場合分けをする。
タイムアウトまでどのパターンにも一致しない場合は -1 を返す。
コマンド自体がいつまでも終わらない場合のタイムアウト処理は必要かも
# プロンプトが表示されるまで待つ
while True:
index = ssh.expect_list(
[
re.compile(r".*(#|\$) ".encode()), # 一般ユーザーまたはroot ユーザーのプロンプト
re.compile(r"\n".encode()),
], timeout = 10)
print(ssh.before.decode(encoding='utf-8'), flush=True, end="")
print(ssh.after.decode(encoding='utf-8'), flush=True, end="")
if index == 0:
break
なお、ssh.before で expect 呼び出しより前の出力を取得できる。
ssh.after で expect 呼び出しより後の出力を取得できる。
su - を実行して root パスワードを送信する。
su - を実行する。
ssh.sendline("su -")
su - を実行後表示されるパスワードの入力確認が表示されるのを待つ
ssh.expect(r".*: ")
root パスワードを送信する。
ssh.sendline(root_pass)
root パスワードを送信結果を確認する。
root パスワードが正しい場合、root 用のプロンプトが表示される。
root パスワードが誤りの場合、su: Authentication failure あるいは su: 認証失敗 が表示される。
これらのいずれかの文字列が表示されるのを待つ。
expect で待つ場合は単に文字列を渡せばよいが、
expect_list で複数のパターンを渡す場合は re.compile で正規表現をコンパイルしたオブジェクトを渡す必要がある。かつ、re.compileに渡すデータは bytes データである必要がある。このため UTF-8 エンコードして bytes データにした配列を渡す。
index = ssh.expect_list(
[
re.compile(r".*# ".encode()), # root ユーザーのプロンプト
re.compile(r"su: .*".encode()), # "su: Authentication failure" or "su: 認証失敗"
], timeout = 10)
if index != 0:
print("failed to switch to root")
sys.exit(1)
パターンマッチした場合、マッチしたインデックスを返すので、rootのプロンプトと一致した場合以外ではエラー終了する。
タイムアウトまでどのパターンにも一致しない場合もエラー終了する。
root シェルの終了
su - で root になった後 exit コマンドを実行して、 root シェルを終了する。
ログアウト
root シェル終了後、ssh.logout() でログアウトする。
コード
Discussion