gistfile1.txt

#!/usr/bin/env python3
"""
CAPWAP Fuzzer - RFC 5415
Objetivo: identificar parsing bugs en la capa de control del WLC (C9800)
y en APs (CW9171/CW9174) via paquetes malformados en UDP 5246/5247.

RFC 5415 secciones clave:
  4.0  - CAPWAP Header Format
  4.6  - Control Messages
  4.7  - Data Channel Keep-Alive
  6.0  - CAPWAP Control Messages (tipos 1-60+)

USO:
  sudo python3 capwap_fuzzer.py --target 192.168.1.1 --mode header
  sudo python3 capwap_fuzzer.py --target 192.168.1.1 --mode msgtype
  sudo python3 capwap_fuzzer.py --target 192.168.1.1 --mode element --log results.json
"""

import socket
import struct
import random
import time
import json
import argparse
import logging
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime

logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s')
log = logging.getLogger(__name__)

# ─── RFC 5415 §4.0 CAPWAP Header ───────────────────────────────────────────
#
#  0                   1                   2                   3
#  0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
# +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
# |HLEN |  RID  |WBID |T|F|L|W|M|K|  Flags  |    Fragment ID    |
# +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
# |         Fragment Offset        |           Reserved           |
# +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
#
# HLEN: header length in 32-bit words
# RID:  radio ID (0-30)
# WBID: wireless binding ID (1=802.11, 2=EPCGlobal)
# T:    type (0=data, 1=control)
# F:    fragment flag
# L:    last fragment
# W:    wireless specific info present
# M:    MIC present
# K:    keep-alive
#
# Control port: UDP 5246
# Data port:    UDP 5247

CAPWAP_CTRL_PORT = 5246
CAPWAP_DATA_PORT = 5247

# RFC 5415 §6 - Control message types
MSG_TYPES = {
    1:  "Discovery Request",
    2:  "Discovery Response",
    3:  "Join Request",
    4:  "Join Response",
    5:  "Configuration Status Request",
    6:  "Configuration Status Response",
    7:  "Configuration Update Request",
    8:  "Configuration Update Response",
    9:  "WTP Event Request",
    10: "WTP Event Response",
    11: "Change State Event Request",
    12: "Change State Event Response",
    13: "Echo Request",
    14: "Echo Response",
    15: "Image Data Request",
    16: "Image Data Response",
    17: "Reset Request",
    18: "Reset Response",
    19: "Primary Discovery Request",
    20: "Primary Discovery Response",
    21: "Data Transfer Request",
    22: "Data Transfer Response",
    23: "Clear Configuration Request",
    24: "Clear Configuration Response",
    25: "Station Configuration Request",
    26: "Station Configuration Response",
}

# RFC 5415 §4.6.1 - Message Element types (campos que parsea el WLC)
# Estos son los más interesantes para fuzzing porque cada uno tiene
# su propio parser en el WLC/AP
MSG_ELEMENTS = {
    1:   "AC Descriptor",
    2:   "AC IPv4 List",
    3:   "AC IPv6 List",
    4:   "AC Name",
    5:   "AC Name with Priority",
    6:   "AC Timestamp",
    7:   "Add MAC ACL Entry",
    8:   "Add Station",
    9:   "Reserved",
    10:  "CAPWAP Control IPV4 Address",
    11:  "CAPWAP Control IPV6 Address",
    12:  "CAPWAP Timers",
    13:  "Data Transfer Data",
    14:  "Data Transfer Mode",
    15:  "Decryption Error Report",
    16:  "Decryption Error Report Period",
    17:  "Delete MAC ACL Entry",
    18:  "Delete Station",
    19:  "Reserved",
    20:  "Discovery Type",
    21:  "Duplicate IPv4 Address",
    22:  "Duplicate IPv6 Address",
    23:  "Idle Timeout",
    24:  "Image Data",
    25:  "Image Identifier",
    26:  "Image Information",
    27:  "Initiate Download",
    28:  "Location Data",
    29:  "Maximum Message Length",
    30:  "CAPWAP Local IPv4 Address",
    31:  "Radio Administrative State",
    32:  "Radio Operational State",
    33:  "Result Code",
    34:  "Returned Message Element",
    35:  "Session ID",
    36:  "Statistics Timer",
    37:  "Vendor Specific Payload",        # ← parser por vendor, muy interesante
    38:  "WTP Board Data",
    39:  "WTP Descriptor",
    40:  "WTP Fallback",
    41:  "WTP Frame Tunnel Mode",
    42:  "Reserved",
    43:  "Reserved",
    44:  "WTP MAC Type",
    45:  "WTP Name",
    46:  "Reserved",
    47:  "WTP Radio Statistics",
    48:  "WTP Reboot Statistics",
    49:  "WTP Static IP Address Information",
    50:  "WTP Radio Information",           # ← parsea configuración de radio
    51:  "Supported Rates",
    52:  "Tx Power",
    53:  "WTP RSSI Threshold",
    54:  "AntennaCoefficients",
    55:  "IEEE 802.11 WTP Radio Information",
    56:  "IEEE 802.11 Supported Rates",
}

@dataclass
class CapwapHeader:
    hlen: int = 2         # header length in 32-bit words (mínimo=2 para control)
    rid: int = 0          # radio ID
    wbid: int = 1         # 1=802.11
    type_bit: int = 1     # 1=control
    frag: int = 0
    last_frag: int = 1
    wireless: int = 0
    mic: int = 0
    keep_alive: int = 0
    fragment_id: int = 0
    fragment_offset: int = 0

    def pack(self) -> bytes:
        # Word 1: HLEN(5) RID(5) WBID(5) T F L W M K Flags(3) FragID(16)
        word1 = (
            ((self.hlen & 0x1F) << 27) |
            ((self.rid & 0x1F) << 22) |
            ((self.wbid & 0x1F) << 17) |
            ((self.type_bit & 0x1) << 16) |
            ((self.frag & 0x1) << 15) |
            ((self.last_frag & 0x1) << 14) |
            ((self.wireless & 0x1) << 13) |
            ((self.mic & 0x1) << 12) |
            ((self.keep_alive & 0x1) << 11) |
            (self.fragment_id & 0xFFFF)  # últimos 16 bits
        )
        word2 = (self.fragment_offset & 0x1FFF) << 3
        return struct.pack(">II", word1, word2)


@dataclass
class CapwapControlHeader:
    msg_type: int = 1     # Discovery Request
    seq_num: int = 0
    msg_element_length: int = 0
    flags: int = 0

    def pack(self) -> bytes:
        # msg_type: 32 bits
        # seq_num: 8 bits
        # msg_element_length: 16 bits  
        # flags: 8 bits
        return struct.pack(">IBHB",
            self.msg_type,
            self.seq_num & 0xFF,
            self.msg_element_length & 0xFFFF,
            self.flags & 0xFF
        )


def build_msg_element(elem_type: int, data: bytes) -> bytes:
    """RFC 5415 §4.6.1: Type(16) + Length(16) + Value"""
    return struct.pack(">HH", elem_type, len(data)) + data


def build_discovery_request(seq: int = 0) -> bytes:
    """
    Paquete Discovery Request legítimo (msg type 1).
    Este es el primer paquete que envía un AP al WLC.
    Útil como baseline para verificar que el target responde.
    """
    # Discovery Type element (type=20): 0=Unknown, 1=Static, 2=DHCP, 3=DNS, 4=AC Referral
    discovery_elem = build_msg_element(20, bytes([1]))  # Static discovery

    ctrl_hdr = CapwapControlHeader(
        msg_type=1,
        seq_num=seq,
        msg_element_length=len(discovery_elem),
        flags=0
    )
    hdr = CapwapHeader(type_bit=1)

    payload = ctrl_hdr.pack() + discovery_elem
    # Reajustar HLEN: header size en words de 32 bits
    # Para paquete sin wireless binding, HLEN=2
    return hdr.pack() + payload


# ─── Estrategias de fuzzing ─────────────────────────────────────────────────

class FuzzStrategy:
    """Genera variantes malformadas de un campo o paquete."""

    @staticmethod
    def interesting_ints_8() -> list:
        return [0, 1, 0x7F, 0x80, 0xFE, 0xFF]

    @staticmethod
    def interesting_ints_16() -> list:
        return [0, 1, 0xFF, 0x100, 0x7FFF, 0x8000, 0xFFFE, 0xFFFF]

    @staticmethod
    def interesting_ints_32() -> list:
        return [0, 1, 0xFF, 0xFFFF, 0x7FFFFFFF, 0x80000000, 0xFFFFFFFE, 0xFFFFFFFF]

    @staticmethod
    def random_bytes(n: int) -> bytes:
        return bytes(random.randint(0, 255) for _ in range(n))

    @staticmethod
    def long_string(n: int = 1024) -> bytes:
        return b'A' * n


class HeaderFuzzer:
    """
    Fuzzea campos del CAPWAP header.
    
    Campos más interesantes por qué:
    - HLEN: si se parsea sin validar, HLEN muy grande puede causar
            lectura OOB al calcular offset del payload
    - Fragment ID + Offset: el reassembler de fragmentos es código
            complejo, históricamente buggy (ver CVE-2021-34704)
    - WBID: valores no estándar pueden activar parsers no probados
    - RID: radio ID fuera de rango (>30) puede indexar array OOB
    """

    def generate(self) -> list[tuple[str, bytes]]:
        cases = []

        # HLEN boundary values (5 bits, 0-31)
        for hlen in FuzzStrategy.interesting_ints_8():
            hdr = CapwapHeader(hlen=hlen & 0x1F, type_bit=1)
            ctrl = CapwapControlHeader(msg_type=1, seq_num=0,
                                        msg_element_length=0)
            cases.append((f"hlen={hlen}", hdr.pack() + ctrl.pack()))

        # RID fuera de rango
        for rid in [0, 15, 30, 31]:
            hdr = CapwapHeader(rid=rid, type_bit=1)
            ctrl = CapwapControlHeader(msg_type=1, seq_num=0,
                                        msg_element_length=0)
            cases.append((f"rid={rid}", hdr.pack() + ctrl.pack()))

        # WBID no estándar (solo 1 y 2 están definidos)
        for wbid in [0, 3, 15, 31]:
            hdr = CapwapHeader(wbid=wbid, type_bit=1)
            ctrl = CapwapControlHeader(msg_type=1, seq_num=0,
                                        msg_element_length=0)
            cases.append((f"wbid={wbid}", hdr.pack() + ctrl.pack()))

        # Fragment con offset incoherente
        hdr = CapwapHeader(frag=1, last_frag=0, fragment_id=0x1234,
                           fragment_offset=0x1FFF, type_bit=1)
        ctrl = CapwapControlHeader(msg_type=1, seq_num=0, msg_element_length=0)
        cases.append(("frag_offset_max", hdr.pack() + ctrl.pack()))

        # Header truncado (menos de 8 bytes)
        for trunc_len in [1, 2, 3, 4, 7]:
            full = CapwapHeader(type_bit=1).pack()
            cases.append((f"header_truncated_{trunc_len}", full[:trunc_len]))

        return cases


class MsgTypeFuzzer:
    """
    Fuzzea el campo msg_type del control header.
    
    Tipos no definidos (>26 en RFC base) pueden llegar a código
    con switch/case sin default handler, o activar features
    vendor-specific de Cisco (tipos en rango 0x80xx).
    """

    def generate(self) -> list[tuple[str, bytes]]:
        cases = []
        hdr = CapwapHeader(type_bit=1)

        # Tipos válidos como baseline
        for mt in [1, 3, 5, 13]:
            ctrl = CapwapControlHeader(msg_type=mt, seq_num=0,
                                        msg_element_length=0)
            cases.append((f"msgtype_valid_{mt}_{MSG_TYPES.get(mt,'?')}",
                          hdr.pack() + ctrl.pack()))

        # Tipos fuera del rango RFC (27-255)
        for mt in range(27, 60):
            ctrl = CapwapControlHeader(msg_type=mt, seq_num=0,
                                        msg_element_length=0)
            cases.append((f"msgtype_undef_{mt}", hdr.pack() + ctrl.pack()))

        # Tipos Cisco vendor-specific (rango alto)
        for mt in [0x80000001, 0xFFFFFFFF, 0x00000000]:
            ctrl = CapwapControlHeader(msg_type=mt, seq_num=0,
                                        msg_element_length=0)
            cases.append((f"msgtype_vendor_{hex(mt)}",
                          hdr.pack() + ctrl.pack()))

        return cases


class ElementFuzzer:
    """
    Fuzzea Message Elements (TLVs).
    
    El parsing TLV es donde viven la mayoría de bugs:
    - Length > datos reales → read past buffer
    - Length=0 con código que no lo chequea → div-by-zero o infinite loop  
    - Type desconocido con Length grande → skip logic OOB
    - Elementos duplicados → comportamiento undefined en algunos parsers
    - Anidamiento de vendor-specific (type=37) con sub-TLVs malformados
    """
a
    def generate(self) -> list[tuple[str, bytes]]:
        cases = []
        hdr = CapwapHeader(type_bit=1)

        def make_pkt(elements_bytes: bytes, msg_type: int = 1) -> bytes:
            ctrl = CapwapControlHeader(
                msg_type=msg_type,
                seq_num=random.randint(0, 255),
                msg_element_length=len(elements_bytes)
            )
            return hdr.pack() + ctrl.pack() + elements_bytes

        # Length = 0 para cada tipo de elemento
        for etype in [1, 4, 20, 25, 37, 39, 45, 50]:
            elem = struct.pack(">HH", etype, 0)
            cases.append((f"elem_{etype}_len0", make_pkt(elem)))

        # Length mayor que datos reales (length claim vs actual)
        for etype in [4, 25, 37, 45]:
            # Declaramos length=100 pero solo ponemos 4 bytes de data
            elem = struct.pack(">HH", etype, 100) + b'\x41' * 4
            cases.append((f"elem_{etype}_len_overflow", make_pkt(elem)))

        # Length = 0xFFFF (max uint16)
        for etype in [4, 37, 39]:
            elem = struct.pack(">HH", etype, 0xFFFF) + b'\x00' * 8
            cases.append((f"elem_{etype}_len_max", make_pkt(elem)))

        # Vendor Specific (type=37): sub-TLV con Enterprise Number de Cisco
        # Cisco Enterprise Number = 9 (IANA)
        cisco_en = struct.pack(">I", 9)
        for sub_type in [0, 1, 0xFF, 0xFFFF]:
            sub_data = b'\xDE\xAD\xBE\xEF' * 4
            sub_tlv = struct.pack(">HH", sub_type, len(sub_data)) + sub_data
            vendor_payload = cisco_en + sub_tlv
            elem = build_msg_element(37, vendor_payload)
            cases.append((f"vendor_specific_cisco_sub{sub_type}",
                          make_pkt(elem)))

        # Elementos duplicados (mismo type, dos veces)
        for etype in [4, 35, 45]:
            data = b'\x41' * 8
            elem = build_msg_element(etype, data) + build_msg_element(etype, data)
            cases.append((f"elem_{etype}_duplicate", make_pkt(elem)))

        # Paquete con 0 elementos pero msg_element_length != 0
        ctrl = CapwapControlHeader(msg_type=1, seq_num=0,
                                    msg_element_length=999)
        cases.append(("ctrl_len_mismatch_no_data", hdr.pack() + ctrl.pack()))

        # Paquete gigante (fragmentación forzada)
        big_data = FuzzStrategy.random_bytes(8192)
        elem = build_msg_element(45, big_data)  # WTP Name con 8KB
        cases.append(("elem_wtpname_huge", make_pkt(elem)))

        # Session ID (type=35) con valores boundary
        for sid_len in [0, 1, 15, 16, 17, 255]:
            elem = build_msg_element(35, b'\xAB' * sid_len)
            cases.append((f"session_id_len{sid_len}", make_pkt(elem)))

        # Image Data (type=24) - relevante porque hay parsers de firmware
        for img_len in [0, 1, 4, 100, 1400]:
            elem = build_msg_element(24, FuzzStrategy.random_bytes(img_len))
            cases.append((f"image_data_len{img_len}",
                          make_pkt(elem, msg_type=15)))  # Image Data Request

        return cases


# ─── Runner ─────────────────────────────────────────────────────────────────

class CapwapFuzzer:
    def __init__(self, target: str, port: int = CAPWAP_CTRL_PORT,
                 delay: float = 0.05, timeout: float = 1.0):
        self.target = target
        self.port = port
        self.delay = delay
        self.timeout = timeout
        self.results = []
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.settimeout(timeout)

    def send_and_capture(self, label: str, payload: bytes) -> dict:
        result = {
            "ts": datetime.utcnow().isoformat(),
            "label": label,
            "sent_bytes": len(payload),
            "sent_hex": payload.hex(),
            "response": None,
            "response_hex": None,
            "error": None
        }
        try:
            self.sock.sendto(payload, (self.target, self.port))
            try:
                resp, addr = self.sock.recvfrom(4096)
                result["response"] = len(resp)
                result["response_hex"] = resp[:64].hex()
                log.info(f"  [{label}] → {len(payload)}B sent, {len(resp)}B recv")
            except socket.timeout:
                log.debug(f"  [{label}] → {len(payload)}B sent, no response")
        except Exception as e:
            result["error"] = str(e)
            log.warning(f"  [{label}] error: {e}")

        self.results.append(result)
        time.sleep(self.delay)
        return result

    def baseline(self):
        """Verificar conectividad con Discovery Request legítimo."""
        log.info(f"[*] Baseline check → {self.target}:{self.port}")
        pkt = build_discovery_request(seq=0)
        r = self.send_and_capture("baseline_discovery", pkt)
        if r["response"]:
            log.info("[+] Target responde a Discovery Request — CAPWAP activo")
            return True
        log.warning("[-] Sin respuesta al baseline — ¿puerto cerrado? ¿firewall?")
        return False

    def run(self, mode: str = "all"):
        log.info(f"[*] Iniciando fuzzing en {self.target}:{self.port} modo={mode}")

        fuzzers = {
            "header":  HeaderFuzzer(),
            "msgtype": MsgTypeFuzzer(),
            "element": ElementFuzzer(),
        }

        if mode == "all":
            selected = fuzzers.values()
        elif mode in fuzzers:
            selected = [fuzzers[mode]]
        else:
            log.error(f"Modo desconocido: {mode}")
            return

        if not self.baseline():
            log.warning("[!] Continuando de todas formas...")

        total = 0
        for fuzzer in selected:
            cases = fuzzer.generate()
            log.info(f"[*] {fuzzer.__class__.__name__}: {len(cases)} casos")
            for label, pkt in cases:
                self.send_and_capture(label, pkt)
                total += 1

        log.info(f"[*] Done: {total} paquetes enviados")
        return self.results

    def save(self, path: str):
        with open(path, "w") as f:
            json.dump(self.results, f, indent=2)
        log.info(f"[*] Resultados guardados en {path}")

    def summary(self):
        responded = [r for r in self.results if r["response"] is not None]
        errors    = [r for r in self.results if r["error"] is not None]
        print(f"\n{'='*50}")
        print(f"Total enviados : {len(self.results)}")
        print(f"Con respuesta  : {len(responded)}")
        print(f"Errores        : {len(errors)}")
        print(f"\nCasos con respuesta (posibles crash/reset si luego callaron):")
        for r in responded:
            print(f"  {r['label']}: {r['response']}B ← {r['response_hex'][:32]}")


def main():
    ap = argparse.ArgumentParser(description="CAPWAP Fuzzer - RFC 5415")
    ap.add_argument("--target", required=True, help="IP del WLC o AP")
    ap.add_argument("--port", type=int, default=CAPWAP_CTRL_PORT,
                    help=f"Puerto UDP (default {CAPWAP_CTRL_PORT})")
    ap.add_argument("--mode", choices=["header", "msgtype", "element", "all"],
                    default="all")
    ap.add_argument("--delay", type=float, default=0.05,
                    help="Segundos entre paquetes")
    ap.add_argument("--log", help="Guardar resultados en JSON")
    args = ap.parse_args()

    fuzzer = CapwapFuzzer(args.target, args.port, args.delay)
    fuzzer.run(args.mode)
    fuzzer.summary()
    if args.log:
        fuzzer.save(args.log)


if __name__ == "__main__":
    main()