pythonからssh-agentを使う。paramikoを拡張して使う。

動機

ssh-agentは、sshのリモート接続の際の認証以外でも、内緒にしたいデータの鍵のキャッシュとしても広く使い道があるのではないか、と思い立って、pythonからssh-agentにアクセスするユーティリティをつくっておこうと思った。

概要・実装したい機能

• Pytonで、ssh-agentへのアクセスも含めたモジュールとして、paramikoというのを見つけたので、これを利用する。
• 暗号関係のライブラリとしては このparamikoが依存しているcryptographyを利用する。
• 通常ssh-agentを使うには、初回にはssh-keygenで秘密鍵と公開鍵をつくって、次にssh-agentを走らせて、シェルで環境変数SSH_AGENT_SOCKを設定して、 ssh-addでパスフレーズを叩いて鍵をssh-agentに登録して...という準備が必要である。こういうのは自動化したい。
  1. ssh-agentに使おうとする名前の鍵セットが登録済みであれば使う
  2. 使おうとする名前の秘密鍵と公開鍵のセットがなければつくる。すでにあれば、既存のものを利用する。
  3. 2.で準備した鍵をssh-agentに登録する。

必要なこと

paramikoで、ssh-agentとのやりとりをするのは、paramiko.agent.AgentKeyというクラスである。が、2024年1月現在(バージョン3.4.0)ではAPI Docやソースを見る限り、クラスのコンストラクタでssh-agentから登録されている鍵情報とその利用を提供するクラスparamiko.agent.AgentKeyのリスト作って保持する機能があるという実にシンプルなものである。また、paramikoでは、暗号化した鍵を保持するためにparamiko.pkey.PKeyという基底クラスにして、暗号方法ごとの("RSA","DSA", "ECDSA", "ED25519")継承のクラスを実装しているが、いずれもopenssh形式のファイルを読み込んで構築するようになっている。

そのため、前述の機能を実現するには、下記の機能を追加する必要がある。

• ssh-addのように、鍵をssh-agentに登録する機能。
• 追加した鍵を利用するために、保持した鍵のリストを更新する機能。
• ssh-keygenに相当する、鍵を新規作成してparamikoの鍵クラスとして読み込む機能。

このうち最初の2点については、あえて継承して新しいクラスをつくるまでもないと思うので、ここなどで紹介されているような方法で、paramiko.agent.AgentKeyにメンバ関数を追加してみる。3点目にかんしては、その他の機能も含めて、新しいクラス(SSHAgentManager)を作ってみることにする。

paramiko.agent.Agentに、鍵をssh-agentに登録する機能を追加する

githubに関係しそうなスレッドがあったのを見つけたのでこれを参考にしてみた。このスレッドでは、"RSA"のみが例になっていたので、他の暗号方法についてもopensshにあるssh-agentプロトコルのドキュメントに従って実装してみる。鍵の種類ごとにプロトコルに従って鍵データを表現するメッセージ(paramiko.message.Messageクラス)を作成してssh-agentに送る、ということらしい。

paramiko.agent.AgentKey.ssh_add_key() を追加する

paramiko.agent.SSH_AGENT_FAILURE = 5
paramiko.agent.SSH_AGENT_SUCCESS = 6
paramiko.agent.cSSH2_AGENTC_ADD_IDENTITY = paramiko.common.byte_chr(17)

def paramiko_agent_agent_ssh_add_key(self, key : paramiko.pkey.PKey, key_comment: str=""):
    """
    Register the private key to ssh-agent
    """
    ptype, result = (None, None)
    if isinstance(key, paramiko.rsakey.RSAKey):
        msg = paramiko.message.Message()
        msg.add_byte(paramiko.agent.cSSH2_AGENTC_ADD_IDENTITY)
        msg.add_string(key.get_name())
        msg.add_mpint(key.public_numbers.n)
        msg.add_mpint(key.public_numbers.e)
        msg.add_mpint(key.key.private_numbers().d)
        msg.add_mpint(key.key.private_numbers().iqmp)
        msg.add_mpint(key.key.private_numbers().p)
        msg.add_mpint(key.key.private_numbers().q)
        msg.add_string(key_comment)
        ptype, result = self._send_message(msg)
    elif isinstance(key, paramiko.ecdsakey.ECDSAKey):
        msg = paramiko.message.Message()
        msg.add_byte(paramiko.agent.cSSH2_AGENTC_ADD_IDENTITY)
        msg.add_string(key.ecdsa_curve.key_format_identifier)
        msg.add_string(key.ecdsa_curve.nist_name)
        msg.add_string(key.verifying_key.public_bytes(encoding=cryptography.hazmat.primitives.serialization.Encoding.X962,
                                                      format=cryptography.hazmat.primitives.serialization.PublicFormat.UncompressedPoint))
        msg.add_mpint(key.signing_key.private_numbers().private_value)
        msg.add_string(key_comment)
        ptype, result = self._send_message(msg)
    elif isinstance(key, paramiko.ed25519key.Ed25519Key):
        msg = paramiko.message.Message()
        msg.add_byte(paramiko.agent.cSSH2_AGENTC_ADD_IDENTITY)
        msg.add_string(key.get_name())
        msg.add_string(key._signing_key.verify_key._key)
        msg.add_string(key._signing_key._seed+key._signing_key.verify_key._key)
        msg.add_string(key_comment)
        ptype, result = self._send_message(msg)
    elif isinstance(key, paramiko.dsskey.DSSKey):
        msg = paramiko.message.Message()
        msg.add_byte(paramiko.agent.cSSH2_AGENTC_ADD_IDENTITY)
        msg.add_string(key.get_name())
        msg.add_mpint(key.p)
        msg.add_mpint(key.q)
        msg.add_mpint(key.g)
        msg.add_mpint(key.y)
        msg.add_mpint(key.x)
        msg.add_string(key_comment)
        ptype, result = self._send_message(msg)
    else:
        raise NotImplementedError("Unknown key type")
    return (ptype, result)
setattr(paramiko.agent.Agent, 'ssh_add_key', paramiko_agent_agent_ssh_add_key)

paramiko.agent.Agentで、ssh-agentから鍵のリストを読み込み直す機能を追加する。

これに関しては、コンストラクタ(__init__(self,...))の該当する部分のみの機能のメソッドをつくればよい。

paramiko.agent.AgentKey.fetch_agent_keylist() を追加する

def paramiko_agent_agent_fetch_agent_keylist(self, verbose=False):
    """
    Fetch the list of the registered keys from ssh-agent
    """
    ptype, result = self._send_message(paramiko.agent.cSSH2_AGENTC_REQUEST_IDENTITIES)
    if ptype != paramiko.agent.SSH2_AGENT_IDENTITIES_ANSWER:
        raise SSHException("could not get keys from ssh-agent")
    keys = []
    for i in range(result.get_int()):
        keys.append(paramiko.agent.AgentKey(agent=self,
                                            blob=result.get_binary(),
                                            comment=result.get_text()))
    self._keys = tuple(keys)
setattr(paramiko.agent.Agent, 'fetch_agent_keylist', paramiko_agent_agent_fetch_agent_keylist)