python - 有没有办法将 Python sshtunnel 传递给 key 而不是文件？

Question

我在本地使用以下库连接到完美运行的远程服务器:

https://pypi.org/project/sshtunnel/

但我需要将我的 Python 函数托管为 Google Cloud 函数。不幸的是，该库似乎只能接受文件，而不是直接作为字符串的键。这是配置:

server = SSHTunnelForwarder(
    SERVER_HOST,
    ssh_username=SSH_USERNAME,
    ssh_pkey="my_filename.pem",
    remote_bind_address=('127.0.0.1', 5412)
)

如果我尝试插入这样的内容:

SSH_KEY = """-----BEGIN RSA PRIVATE KEY-----

-----END RSA PRIVATE KEY-----"""

然后将 ssh_pkey 行修改为:

ssh_pkey=SSH_KEY

我的期望是它会起作用，但看起来图书馆不允许这样做。我在这里查看了源代码，它似乎是 this导致问题。

@staticmethod
def get_keys(logger=None, host_pkey_directories=None, allow_agent=False):
    """
    Load public keys from any available SSH agent or local
    .ssh directory.
    Arguments:
        logger (Optional[logging.Logger])
        host_pkey_directories (Optional[list[str]]):
            List of local directories where host SSH pkeys in the format
            "id_*" are searched. For example, ['~/.ssh']
            .. versionadded:: 0.1.0
        allow_agent (Optional[boolean]):
            Whether or not load keys from agent
            Default: False
    Return:
        list
    """
    keys = SSHTunnelForwarder.get_agent_keys(logger=logger) \
        if allow_agent else []

    if host_pkey_directories is not None:
        paramiko_key_types = {'rsa': paramiko.RSAKey,
                              'dsa': paramiko.DSSKey,
                              'ecdsa': paramiko.ECDSAKey,
                              'ed25519': paramiko.Ed25519Key}
        for directory in host_pkey_directories or [DEFAULT_SSH_DIRECTORY]:
            for keytype in paramiko_key_types.keys():
                ssh_pkey_expanded = os.path.expanduser(
                    os.path.join(directory, 'id_{}'.format(keytype))
                )
                if os.path.isfile(ssh_pkey_expanded):
                    ssh_pkey = SSHTunnelForwarder.read_private_key_file(
                        pkey_file=ssh_pkey_expanded,
                        logger=logger,
                        key_type=paramiko_key_types[keytype]
                    )
                    if ssh_pkey:
                        keys.append(ssh_pkey)
    if logger:
        logger.info('{0} keys loaded from host directory'.format(
            len(keys))
        )

    return keys

我以前从来没有修补过任何东西，所以看看这个，我可以以某种方式手动覆盖它吗？

Answer 1

我就是这样解决的，希望对大家有帮助。首先，我使用 Python 打印出 key 文件“temp_key.pem”的 base64 编码 key :

import base64
with open('temp_key.pem', 'rb') as f:
    blob = base64.b64encode(f.read())

print(blob)
for_google_cloud_function = blob.decode('utf-8')
print(for_google_cloud_function)

这个输出我使用了我的环境变量 SSH_KEY_BLOB。然后在我的 GCP Cloud Function 中添加了这个:

# decode key back into a useable form from base64
SSH_KEY_BLOB_DECODED = base64.b64decode(SSH_KEY_BLOB)
SSH_KEY = SSH_KEY_BLOB_DECODED.decode('utf-8')

# pass key to parmiko to get your pkey
pkey = paramiko.RSAKey.from_private_key(io.StringIO(SSH_KEY))

# setup your SSHTunnel like normal
server = SSHTunnelForwarder(
    remote_server_ip,
    ssh_username=SSH_USERNAME,
    ssh_pkey=pkey,
    remote_bind_address=('127.0.0.1', 27017)
)

这样 key 就不会被硬编码，并且函数可以从文件中自给自足。它们可能是更好的方法，但这对我有用。