はじめに
以下のソースを参考にしました。
私の場合、そのままだとエラーが出たので改修したのと、新たな機能を追加しました。
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
class SCPClientWrapper(object): | |
def __init__(self, login): | |
self.login = login | |
def __enter__(self): | |
from paramiko import SSHClient, AutoAddPolicy | |
self.client = SSHClient() | |
self.client.set_missing_host_key_policy(AutoAddPolicy()) | |
self.client.connect( | |
login["host"], | |
port = login["port"], | |
username = login["user"], | |
password = login["pass"]) | |
return self | |
def __exit__(self, exception_type, exception_value, traceback): | |
self.client.close() | |
def find(self, remote_path): | |
stdin, stdout, stderr = self.client.exec_command("find %s -type f -maxdepth 1" % remote_path) | |
return stdout.read().strip("\n").split("\n") | |
def ls(self, remote_path): | |
stdin, stdout, stderr = self.client.exec_command("cd %s && ls" % remote_path) | |
return stdout.read().strip("\n").split("\n") | |
def cp(self, remote_path, local_path): | |
from scp import SCPClient | |
with SCPClient(self.client.get_transport()) as scp: | |
scp.get(remote_path, local_path) | |
return self | |
def rm(self, remote_path): | |
self.client.exec_command("rm %s" % remote_path) | |
return self |
公式サイト
paramiko
Welcome to Paramiko! — Paramiko documentation
scp
scp
scp module for paramiko
SCPClientのラッパークラス
以下が変更内容です。
find
:私の場合、取得した値がバイトのようでそのまま文字列として変換しようといてエラーになっていたので改修しました。ls
:私の場合、取得した値がバイトのようでそのまま文字列として変換しようといてエラーになっていたので改修しました。__exit__
:引数が不明で、なくても切断できるようなので引数をなくしました。up
:アップロードの機能を追加しました。
class SCPClientWrapper(object):
def __init__(self, login):
self.login = login
def __enter__(self):
from paramiko import SSHClient, AutoAddPolicy
self.client = SSHClient()
self.client.set_missing_host_key_policy(AutoAddPolicy())
self.client.connect(
login["host"],
port = login["port"],
username = login["user"],
password = login["pass"])
return self
def __exit__(self, exception_type, exception_value, traceback):
self.client.close()
def find(self, remote_path):
stdin, stdout, stderr = self.client.exec_command("find %s -type f -maxdepth 1" % remote_path)
re = stdout.read().decode().strip("\n").split("\n")
return re
def ls(self, remote_path):
stdin, stdout, stderr = self.client.exec_command("cd %s && ls" % remote_path)
re = stdout.read().decode().strip("\n").split("\n")
return re
def up(self, local_path, remote_path):
from scp import SCPClient
with SCPClient(self.client.get_transport()) as scp:
scp.put(local_path, remote_path)
return self
def cp(self, remote_path, local_path):
from scp import SCPClient
with SCPClient(self.client.get_transport()) as scp:
scp.get(remote_path, local_path)
return self
def rm(self, remote_path):
self.client.exec_command("rm %s" % remote_path)
return self
login = {}
login["host"] = host
login["port"] = port
login["user"] = username
login["pass"] = password
# インスタンス作成
scp = SCPClientWrapper(login)
# 接続
scp.__enter__()
# 処理
scp.up("scp.html", "/home/xxx/www/scp.html")
class SCPClientWrapper(object):
def __init__(self, login):
self.login = login
def __enter__(self):
from paramiko import SSHClient, AutoAddPolicy
self.client = SSHClient()
self.client.set_missing_host_key_policy(AutoAddPolicy())
self.client.connect(
login["host"],
port = login["port"],
username = login["user"],
password = login["pass"])
return self
def __exit__(self, exception_type, exception_value, traceback):
self.client.close()
def find(self, remote_path):
stdin, stdout, stderr = self.client.exec_command("find %s -type f -maxdepth 1" % remote_path)
re = stdout.read().decode().strip("\n").split("\n")
return re
def ls(self, remote_path):
stdin, stdout, stderr = self.client.exec_command("cd %s && ls" % remote_path)
re = stdout.read().decode().strip("\n").split("\n")
return re
def up(self, local_path, remote_path):
from scp import SCPClient
with SCPClient(self.client.get_transport()) as scp:
scp.put(local_path, remote_path)
return self
def cp(self, remote_path, local_path):
from scp import SCPClient
with SCPClient(self.client.get_transport()) as scp:
scp.get(remote_path, local_path)
return self
def rm(self, remote_path):
self.client.exec_command("rm %s" % remote_path)
return self
login = {}
login["host"] = host
login["port"] = port
login["user"] = username
login["pass"] = password
# インスタンス作成
scp = SCPClientWrapper(login)
# 接続
scp.__enter__()
# 処理
scp.up("scp.html", "/home/xxx/www/scp.html")
class SCPClientWrapper(object): def __init__(self, login): self.login = login def __enter__(self): from paramiko import SSHClient, AutoAddPolicy self.client = SSHClient() self.client.set_missing_host_key_policy(AutoAddPolicy()) self.client.connect( login["host"], port = login["port"], username = login["user"], password = login["pass"]) return self def __exit__(self, exception_type, exception_value, traceback): self.client.close() def find(self, remote_path): stdin, stdout, stderr = self.client.exec_command("find %s -type f -maxdepth 1" % remote_path) re = stdout.read().decode().strip("\n").split("\n") return re def ls(self, remote_path): stdin, stdout, stderr = self.client.exec_command("cd %s && ls" % remote_path) re = stdout.read().decode().strip("\n").split("\n") return re def up(self, local_path, remote_path): from scp import SCPClient with SCPClient(self.client.get_transport()) as scp: scp.put(local_path, remote_path) return self def cp(self, remote_path, local_path): from scp import SCPClient with SCPClient(self.client.get_transport()) as scp: scp.get(remote_path, local_path) return self def rm(self, remote_path): self.client.exec_command("rm %s" % remote_path) return self login = {} login["host"] = host login["port"] = port login["user"] = username login["pass"] = password # インスタンス作成 scp = SCPClientWrapper(login) # 接続 scp.__enter__() # 処理 scp.up("scp.html", "/home/xxx/www/scp.html")
関数の種類
- __enter__:SCPで接続します。
- __exit__:SCPを切断します。
- find:ファイルを検索します。
- ls:ファイルやディレクトリの一覧を取得します。
- up:ファイルをアップロードします。
- cp:ファイルをダウンロードします。
- rm:ファイルを削除します。
おまけ:SCPで接続できない場合の対応
上記のラッパーを使用せずに、公式サイトのように接続しようとすると以下のようなエラーが出ました。
SSHException: Server 'www.xxx.com' not found in known_hosts
本ラッパーにもあるように、以下のポリシーを追加するとエラーがなくなり接続できます。
client.set_missing_host_key_policy(AutoAddPolicy())
AutoAddPolicy()
はparamikoですので、インポートの仕方によってはparamiko.AutoAddPolicy()
とする必要があります。
コメント