ssh_tunnel_test.py

ssh_tunnel_test.py

import logging
import threading
import time
import sshtunnel
from sshtunnel import HandlerSSHTunnelForwarderError, SSHTunnelForwarder

sshtunnel.DEFAULT_LOGLEVEL = 1
logging.basicConfig(filename='example.log', filemode='w', level=sshtunnel.DEFAULT_LOGLEVEL)
logging.warning('Start example! v %s', sshtunnel.__version__)

print(f'Creating SSHTunnelForwarder... (v{sshtunnel.__version__})')
tunnel = SSHTunnelForwarder(
    ('165.22.65.44', 22),
    ssh_username='root',
    ssh_pkey='~/.ssh/id_rsa',
    remote_bind_address=('127.0.0.1', 1500),
    local_bind_address=('127.0.0.1', 53826),
)

try:
    print(f'Trying to open ssh tunnel...')
    tunnel.start()
except HandlerSSHTunnelForwarderError as e:
    logging.exception('Tunnel start exception: {}'.format(e))

i = 30
logging.warning(f'Sleeping for {i} second...')
while i:
    time.sleep(1)
    if i % 10 == 0:
        print(f'Runing tunnel.check_tunnels... (i={i})')
        tunnel.check_tunnels()
        print(f'Check result: {tunnel.tunnel_is_up} (i={i})')
        if not (tunnel.is_active):
            print(f'Tunnel is DOWN! restarting ...')
            tunnel.restart()
    i -= 1

current_threads = threading.enumerate()
print(f'1 Unclosed threads list: {current_threads}')

try:
    print(f'Trying to close resources...')
    tunnel.stop()
except Exception as e:
    logging.exception(f'Resources closing exception: {e}')


i = 0
print(f'Sleeping for {i} second...')
time.sleep(i)
current_threads = threading.enumerate()

print(f'2 Unclosed threads list: {current_threads}')