PythonのparamikoとscpでSCP接続と操作

スポンサーリンク
スポンサーリンク

はじめに

以下のソースを参考にしました。

私の場合、そのままだとエラーが出たので改修したのと、新たな機能を追加しました。

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
view raw scpwrapper.py hosted with ❤ by GitHub

公式サイト

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()とする必要があります。

コメント

タイトルとURLをコピーしました