Skip to content

Instantly share code, notes, and snippets.

@pahaz
Created October 31, 2020 12:42
Show Gist options
  • Save pahaz/af63b5ab476931da431f47a8cfeeca2c to your computer and use it in GitHub Desktop.
Save pahaz/af63b5ab476931da431f47a8cfeeca2c to your computer and use it in GitHub Desktop.
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}')
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment