Search code examples
pythonssh-tunnel

Is there a way to pass Python sshtunnel a key rather than a file?


I am using the following library locally to connect to a remote server which works perfectly:

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

But I need to host my Python function as a Google Cloud Function. Unfortunately the library only appears to be able to accept a file, not a key directly as a string. This is the config:

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

If I try to insert something like this:

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

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

Then amend the ssh_pkey line to:

ssh_pkey=SSH_KEY

My expectation would be that it would work but it looks to be like the library doesn't allow this. I've looked at the source code here and it appears that it's this is causing the issue.

@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

I've never monkey patched anything before so looking at this, could I somehow override this manually?


Solution

  • This is how I solved it, hope it helps someone. First I used Python to print out a base64 encoded key of my key file 'temp_key.pem':

    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)
    

    The output of this I used an my environment variable SSH_KEY_BLOB. In my GCP Cloud Function I then added this:

    # 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)
    )
    

    That way the key is not hard coded and the function is self sufficient from the file. They may be better ways but this worked for me.