Access AWS RDS MySql through bastion box with SSHTunnelForwarder

rds_mysql_tunnel.py

import datetime

# https://docs.aws.amazon.com/lambda/latest/dg/services-rds-tutorial.html

# pip install sshtunnel
from sshtunnel import SSHTunnelForwarder

# pip install pymysql
import pymysql as db
import logging
import os

logger = logging.getLogger()


# NOTE run_query_with WILL HANG... use run_query
def run_query_with(query, params=None):
    conn = None
    results = None
    try:
        with SSHTunnelForwarder(
                (host, 22),
                ssh_username=ssh_username,
                ssh_private_key=ssh_private_key,
                remote_bind_address=(db_host, 3306),

        ) as server:

            conn = db.connect(host='127.0.0.1',
                              port=server.local_bind_port,
                              user=db_user,
                              password=db_password,
                              database=db_name,
                              connect_timeout=5,
                              read_timeout=5)
            cursor = conn.cursor()
            if params:
                cursor.execute(query, params)
            else:
                cursor.execute(query)
            results = cursor.fetchall()
            print(results)
            print("Done")

    except Exception as exc:
        logger.error(f"Could not execute query: Statement: {query} with parameters: {params}.  Exc: {exc}")
    finally:
        if conn:
            conn.close()
        if server:
            server.close()
    return results

def run_query(query, params=None):
    conn = None
    results = None
    try:
        server = SSHTunnelForwarder(
                (host, 22),
                ssh_username=ssh_username,
                ssh_private_key=ssh_private_key,
                remote_bind_address=(db_host, 3306)
        )
        # setting these flags does not appear to be necessary and
        # I am not sure why
        # server.daemon_forward_servers = True
        # server.daemon_transport = True
        server.start()

        conn = db.connect(host='127.0.0.1',
                          port=server.local_bind_port,
                          user=db_user,
                          password=db_password,
                          database=db_name,
                          connect_timeout=5,
                          read_timeout=5)
        cursor = conn.cursor()
        if params:
            cursor.execute(query, params)
        else:
            cursor.execute(query)
        results = cursor.fetchall()

    except Exception as exc:
        logger.error(f"Could not execute query: Statement: {query} with parameters: {params}.  Exc: {exc}")
    finally:
        if conn:
            conn.close()
        if server:
            server.close()
    return results

if __name__ == '__main__':
    from dotenv import load_dotenv
    from pathlib import Path

    env_path = Path('.env')
    if not env_path.exists():
        raise ValueError(".env not found in current directory")

    print(env_path.absolute())
    load_dotenv(dotenv_path=env_path, verbose=True)

    # ssh variables
    host = os.environ['bastion']
    ssh_username = os.environ['ssh_user']
    ssh_private_key = os.environ['ssh_pem']

    # database variables
    db_host = os.environ['db_host']
    db_user = os.environ['db_user']
    db_password = os.environ['db_password']
    db_name = os.environ['db_database']

    results = run_query("SELECT * FROM test_tbl")
    print(results)