Last active
January 24, 2018 12:45
-
-
Save vog3lm/8b837f10ad602c44af044af3fbc28205 to your computer and use it in GitHub Desktop.
paramiko ssh respondable remote shell access
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import paramiko | |
| if __name__ == '__main__': | |
| host = 'REMOTE_HOST_IP' | |
| port = 'REMOTE_HOST_PORT' # int() | |
| user = 'USER_LOGIN_NAME' | |
| code = 'USER_LOGIN_PASSWORD' | |
| command = 'pss-shell -c help' | |
| try: | |
| client = paramiko.SSHClient() | |
| client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) | |
| client.connect(host,port,user,code) | |
| stdin, stdout, stderr = client.exec_command(command, get_pty=True) | |
| stdin.close() | |
| while not stdout.channel.exit_status_ready(): | |
| if stdout.channel.recv_ready(): | |
| for line in stdout: | |
| print line.strip('\n') | |
| except paramiko.AuthenticationException as e: | |
| raise InternalRemoteMachineException(["Auth Error! %s"%(str(e))]) | |
| except ValueError as e: | |
| raise InternalRemoteMachineException(['Value Error! %s'%(str(e))]) | |
| except IOError as e: | |
| raise InternalRemoteMachineException(['IO Error! %s'%(str(e))]) | |
| finally: | |
| client.close() |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/python | |
| import os, sys, subprocess, logging | |
| formatter = logging.Formatter(fmt='%(asctime)18s %(levelname)7s :: %(module)s.%(funcName)s():%(lineno)d :- %(message)s',datefmt='%Y-%m-%d %H:%M:%S') | |
| logger = logging.getLogger() | |
| if(logger.handlers): | |
| logger.handlers.pop() | |
| # stream = StringIO.StringIO() # pipe to string | |
| stream = sys.stdout # pipe to shell | |
| handler = logging.StreamHandler(stream) | |
| handler.setFormatter(formatter) | |
| logger.addHandler(handler) | |
| logger.setLevel(logging.INFO) | |
| os.chdir('/root/paramiko-ssh-shell') | |
| cmd = 'python server.py %s'%(' '.join(sys.argv[1:])) | |
| logging.info('Secure Shell Connection: Shell Env Var Hello') | |
| # Synchronous Calls | |
| # response = os.popen(cmd).read() | |
| # exitCode = os.system(cmd) | |
| subprocess.call([cmd], stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr, shell=True, close_fds=True) | |
| # Asynchronous Calls | |
| # process = subprocess.Popen([cmd], stdin=sys.stdin, stdout=sys.stdout, stderr=sys.stderr, shell=True, close_fds=True) | |
| logging.info('Secure Shell Connection: Shell Env Var Good Bye') |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import logging, sys, getopt | |
| class Executable(object): | |
| def __init__(self): | |
| self.fail = 'unset' | |
| self.errors = [] | |
| self.args = {'cmd':self.fail} | |
| def parameterize(self,parameter): | |
| for key in parameter.keys(): | |
| if(key in self.args.keys()): | |
| self.args[key] = parameter[key] | |
| return self | |
| def validate(self): | |
| self.errors = [] | |
| for k in self.args.keys(): | |
| if(self.fail == self.args[k]): | |
| self.errors.append('%s argument Error! %s not Found!'%(self.args['cmd'],k)) | |
| return self | |
| def execute(self): | |
| logging.error('No Executable Code Implemented.') | |
| return self | |
| class Help(Executable): | |
| def __init__(self): | |
| Executable.__init__(self) | |
| self.args.update({'cmd':'help'}) | |
| def execute(self): | |
| logging.info('Help Command Call.') | |
| return self | |
| class Operation(Executable): | |
| def __init__(self): | |
| Executable.__init__(self) | |
| self.args.update({'cmd':'ops'}) | |
| def execute(self): | |
| logging.info('Operation Command Call.') | |
| return self | |
| def logger(): | |
| formatter = logging.Formatter(fmt='%(asctime)18s %(levelname)7s :: %(module)s.%(funcName)s():%(lineno)d :-- %(message)s',datefmt='%Y-%m-%d %H:%M:%S') | |
| logger = logging.getLogger() | |
| if(logger.handlers): | |
| logger.handlers.pop() | |
| # stream = StringIO.StringIO() # pipe to string | |
| stream = sys.stdout # pipe to shell | |
| handler = logging.StreamHandler(stream) | |
| handler.setFormatter(formatter) | |
| logger.addHandler(handler) | |
| logger.setLevel(logging.INFO) | |
| def receive(argv): | |
| try: | |
| argv = argv[1:] | |
| opts, args = getopt.getopt(argv,shortopts='c:',longopts='cmd=') | |
| return opts | |
| except getopt.GetoptError as e: | |
| logging.error('Parse Options Exception. Kill Application. [msg='+e.msg+']') | |
| sys.exit(2) | |
| def dispatcher(key): | |
| executables = {'help':Help | |
| ,'ops':Operation} | |
| if(key in executables.keys()): | |
| return executables[key]() | |
| return None | |
| def dispatch(opts): | |
| options = {} | |
| for o, a in opts: | |
| if o == "-c": options['cmd'] = a | |
| elif o == "--cmd": options['cmd'] = a | |
| if('cmd' in options.keys()): | |
| executable = dispatcher(options['cmd']) | |
| if(executable): | |
| executable.parameterize(options) | |
| executable.execute() | |
| sys.exit(0) | |
| logging.error('Dispatcher Failure! No executeable found. %s'%(str(opts))) | |
| sys.exit(2) | |
| if __name__ == '__main__': | |
| logger() | |
| logging.info('Secure Shell Connection: Server Hello') | |
| opts = receive(sys.argv) | |
| dispatch(opts) | |
| # response | |
| logging.info('Secure Shell Connection: Server Good Bye') | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment