Adding ability to authenticate jump host using certificate and some other helpful functions

README.md

@@ -36,7 +36,7 @@ $ kafkatunnel aws ec2-user@awsjumphost
 * manual passing your remote zookeeper/kafka and optionally schema registry IPs
 
 ```bash
-$ kafkatunnel manual 10.11.85.128,10.11.82.30,10.11.83.9 10.11.80.7,10.11.80.123,10.11.81.13 10.11.82.10,10.11.82.11
+$ kafkatunnel manual ec2-user@awsjumphost 10.11.85.128,10.11.82.30,10.11.83.9 10.11.80.7,10.11.80.123,10.11.81.13 10.11.82.10,10.11.82.11
 ```
 afterwards you have to provide your root password in order to create the interfaces
 

kafkatunnel.py

@@ -2,6 +2,7 @@
 import click
 import sys
 import subprocess
+import signal
 
 from Instance import ManualInstances, AWSInstances
 
@@ -15,14 +16,30 @@ def cli():
 @click.option('-kp','--kafka_port',default='9092')
 @click.option('-r','--region',default='ap-southeast-2')
 @click.option('-p','--profile',default='default')
-def aws(jump_host,zookeeper_port,kafka_port,region,profile):
+@click.option('-jc','--jump_host_cert',default=None)
+def aws(jump_host,zookeeper_port,kafka_port,region,profile,jump_host_cert):
     instances=[]
     click.echo(' * retrieving ip\'s from AWS ({},{}) zookeeper/kafka ec2 instances by tag_name ...'.format(profile,region))
     aws = AWSInstances(profile,region)
     instances += aws.getIps('zookeeper',zookeeper_port)
     instances += aws.getIps('kafka',kafka_port)
-    connect(jump_host,instances)
+    connect(jump_host,instances,jump_host_cert)
 
+@cli.command(help='clean up interfaces after ungraceful exit from AWS)')
+@click.argument('jump_host')
+@click.option('-zp','--zookeeper_port',default='2181')
+@click.option('-kp','--kafka_port',default='9092')
+@click.option('-r','--region',default='ap-southeast-2')
+@click.option('-p','--profile',default='default')
+@click.option('-jc','--jump_host_cert',default=None)
+def awsclean(jump_host,zookeeper_port,kafka_port,region,profile,jump_host_cert):
+    instances=[]
+    click.echo(' * retrieving ip\'s from AWS ({},{}) zookeeper/kafka ec2 instances by tag_name ...'.format(profile,region))
+    aws = AWSInstances(profile,region)
+    instances += aws.getIps('zookeeper',zookeeper_port)
+    instances += aws.getIps('kafka',kafka_port)
+    click.echo(' * cleaning up interfaces ...')
+    connect(jump_host,instances,jump_host_cert)
 
 @cli.command(help='provide the IP\'s of your zookeeper/kafka')
 @click.argument('jump_host')
@@ -32,21 +49,54 @@ def aws(jump_host,zookeeper_port,kafka_port,region,profile):
 @click.option('-zp','--zookeeper_port',default='2181')
 @click.option('-kp','--kafka_port',default='9092')
 @click.option('-sp','--schemaregistry_port',default='8081')
-def manual(jump_host,zookeeper_ips, kafka_ips, schemaregistry_ips, zookeeper_port, kafka_port, schemaregistry_port):
+@click.option('-jc','--jump_host_cert',default=None)
+def manual(jump_host, zookeeper_ips, kafka_ips, schemaregistry_ips, zookeeper_port, kafka_port, schemaregistry_port, jump_host_cert):
     instances=[]
     click.echo(' * using manual ip\'s ...')
     man = ManualInstances()
     instances += man.getIps('zookeeper',zookeeper_ips, zookeeper_port)
     instances += man.getIps('kafka',kafka_ips, kafka_port)
     if schemaregistry_ips:
         instances += man.getIps('schemareg', schemaregistry_ips, schemaregistry_port)
-    connect(jump_host,instances)
+    connect(jump_host, instances, jump_host_cert)
+
+@cli.command(help='clean up interfaces after ungraceful exit from manual')
+@click.argument('jump_host')
+@click.argument('zookeeper_ips')
+@click.argument('kafka_ips')
+@click.argument('schemaregistry_ips',default='')
+@click.option('-zp','--zookeeper_port',default='2181')
+@click.option('-kp','--kafka_port',default='9092')
+@click.option('-sp','--schemaregistry_port',default='8081')
+@click.option('-jc','--jump_host_cert',default=None)
+def manualclean(jump_host, zookeeper_ips, kafka_ips, schemaregistry_ips, zookeeper_port, kafka_port, schemaregistry_port,jump_host_cert):
+    instances=[]
+    click.echo(' * using manual ip\'s ...')
+    man = ManualInstances()
+    instances += man.getIps('zookeeper',zookeeper_ips, zookeeper_port)
+    instances += man.getIps('kafka',kafka_ips, kafka_port)
+    if schemaregistry_ips:
+        instances += man.getIps('schemareg', schemaregistry_ips, schemaregistry_port)
+    click.echo(' * cleaning up interfaces ...')
+    remove_local_interfaces(instances)
 
 
-def connect(jump_host,instances):
+def connect(jump_host,instances,jump_host_cert):
     print_instances(instances)
     add_local_interfaces(instances)
-    connect_ssh_tunnel(jump_host,instances)
+
+    # clean up if script it terminated
+    def receiveSignal(signalNumber, frame):
+        click.echo(' * received signal {} ...'.format(signalNumber))
+        click.echo(' * cleaning up interfaces ...')
+        remove_local_interfaces(instances)
+        raise SystemExit(' * successfully cleaned up')
+
+    signal.signal(signal.SIGINT, receiveSignal)
+    signal.signal(signal.SIGQUIT, receiveSignal)
+    signal.signal(signal.SIGHUP, receiveSignal)
+
+    connect_ssh_tunnel(jump_host,instances,jump_host_cert)
     remove_local_interfaces(instances)
 
 def add_local_interfaces(instances):
@@ -73,11 +123,14 @@ def print_instances(instances):
         click.echo('{:<10} on {:<15} port {:>5}'.format(i.name,i.ip,i.port))
     click.echo('')
 
-def connect_ssh_tunnel(jump_host,instances):
+def connect_ssh_tunnel(jump_host,instances,jump_host_cert):
     click.echo(' * connecting to jump host ' + jump_host)
-    opts = []
+    opts = ['-N']
+    if jump_host_cert is not None:
+        opts += ['-i', jump_host_cert]
     for i in instances:
         opts += ['-L','{ip}:{port}:{ip}:{port}'.format(ip=i.ip,port=i.port)]
+    click.echo(' * ' + "using {}".format(['ssh'] + opts + [jump_host]))
     subprocess.call(['ssh'] + opts + [jump_host])
 
 if __name__ == '__main__':