[Solved]-How to open an SSH tunnel using python?

Try use sshtunnel package.

This is simple:

pip install sshtunnel
python -m sshtunnel -U vagrant -P vagrant -L :3306 -R 127.0.0.1:3306 -p 2222 localhost

Disclosure: I’m the author and maintainer of this package.

Here is a code snippet for Python3 (but you should be able to retrofit it into Python2 without difficulty). It runs an SSH tunnel in a separate thread; then the main thread does something to get network traffic over the SSH tunnel.

In this example, the ssh tunnel forwards local port 2222 to port 80 on localhost. The main activity consists of running

curl http://localhost:2222

ie., fetching a webpage but from port 2222.

The class SshTunnel is initialized with 4 parameters, the local and remote port, the remote user, and the remote host. All it does, is start SSH in the following way:

ssh -N -L localport:remotehost:remoteport remoteuser@remotehost

In order to make this work, you’ll need a password-less login for remoteuser@remotehost (via ~/.ssh/id_rsa.pub that’s known on the remote server).
The thus running ssh tunnel is on one thread; the main task must be in another one. The ssh tunnel thread is marked as daemon so that it will automatically stop once the main activity terminates.

I didn’t put in a full MySQL connectivity example because it should be self-explanatory. Once SshTunnel sets up a local TCP port, you can connect to it – be it via your MySQL client, curl, or whatever.

import subprocess
import time
import threading

class SshTunnel(threading.Thread):
    def __init__(self, localport, remoteport, remoteuser, remotehost):
        threading.Thread.__init__(self)
        self.localport = localport      # Local port to listen to
        self.remoteport = remoteport    # Remote port on remotehost
        self.remoteuser = remoteuser    # Remote user on remotehost
        self.remotehost = remotehost    # What host do we send traffic to
        self.daemon = True              # So that thread will exit when
                                        # main non-daemon thread finishes

    def run(self):
        if subprocess.call([
            'ssh', '-N',
                   '-L', str(self.localport) + ':' + self.remotehost + ':' + str(self.remoteport),
                   self.remoteuser + '@' + self.remotehost ]):
            raise Exception ('ssh tunnel setup failed')


if __name__ == '__main__':
    tunnel = SshTunnel(2222, 80, 'karel', 'localhost')
    tunnel.start()
    time.sleep(1)
    subprocess.call(['curl', 'http://localhost:2222'])

Here’s a little class that you can drop into your code:

import subprocess
import random
import tempfile

class SSHTunnel:

    def __init__(self, host, user, port, key, remote_port):
        self.host = host
        self.user = user
        self.port = port
        self.key = key
        self.remote_port = remote_port
        # Get a temporary file name
        tmpfile = tempfile.NamedTemporaryFile()
        tmpfile.close()
        self.socket = tmpfile.name
        self.local_port = random.randint(10000, 65535)
        self.local_host = '127.0.0.1'
        self.open = False

    def start(self):
        exit_status = subprocess.call(['ssh', '-MfN',
            '-S', self.socket,
            '-i', self.key,
            '-p', self.port,
            '-l', self.user,
            '-L', '{}:{}:{}'.format(self.local_port, self.local_host, self.remote_port),
            '-o', 'ExitOnForwardFailure=True',
            self.host
        ])
        if exit_status != 0:
            raise Exception('SSH tunnel failed with status: {}'.format(exit_status))
        if self.send_control_command('check') != 0:
            raise Exception('SSH tunnel failed to check')
        self.open = True

    def stop(self):
        if self.open:
            if self.send_control_command('exit') != 0:
                raise Exception('SSH tunnel failed to exit')
            self.open = False

    def send_control_command(self, cmd):
        return subprocess.check_call(['ssh', '-S', self.socket, '-O', cmd, '-l', self.user, self.host])

    def __enter__(self):
        self.start()
        return self

    def __exit__(self, type, value, traceback):
        self.stop()

And here’s how you could use it, for example with MySQL (port 3306 usually):

with SSHTunnel('database.server.com', 'you', '22', '/path/to/private_key', '3306') as tunnel:
    print "Connected on port {} at {}".format(tunnel.local_port, tunnel.local_host)

Might I suggest trying something like pyngrok to programmatically manage an ngrok tunnel for you? Full disclosure, I am the developer of it. SSH example here, and Django example here, but it’s as easy as installing pyngrok:

pip install pyngrok

and using it:

from pyngrok import ngrok

# <NgrokTunnel: "tcp://0.tcp.ngrok.io:12345" -> "localhost:22">
ssh_tunnel = ngrok.connect(22, "tcp")

Or to expose your MySQL database:

# <NgrokTunnel: "tcp://0.tcp.ngrok.io:12346" -> "localhost:3306">
mysql_tunnel = ngrok.connect(3306, "tcp")

Or to expose to Django dev server:

# <NgrokTunnel: "http://<public_sub>.ngrok.io" -> "http://localhost:8000">
http_tunnel = ngrok.connect(8000)