Flipper Zero

#!/usr/bin/env python3
"""Convert WPA/WPA2 EAPOL pairs from an 802.11 PCAP to Hashcat 22000.

This is a small, dependency-free converter intended for captures produced by
Marauder. It supports classic PCAP files containing raw IEEE 802.11 frames or
radiotap-wrapped IEEE 802.11 frames. PCAPNG and PMKID extraction are outside
its scope.
"""

from __future__ import annotations

import argparse
import struct
import sys
from collections import defaultdict
from dataclasses import dataclass
from pathlib import Path
from typing import BinaryIO, Iterable


DLT_IEEE802_11 = 105
DLT_IEEE802_11_RADIO = 127
EAPOL_SNAP = b"\xaa\xaa\x03\x00\x00\x00\x88\x8e"
ZERO_NONCE = bytes(32)
ZERO_MIC = bytes(16)
BROADCAST = bytes.fromhex("ffffffffffff")


class ConversionError(Exception):
    pass


@dataclass(frozen=True)
class PcapInfo:
    endian: str
    nanosecond_timestamps: bool
    linktype: int


@dataclass(frozen=True)
class EapolMessage:
    packet_number: int
    timestamp: float
    number: int
    ap: bytes
    client: bytes
    replay_counter: int
    key_info: int
    nonce: bytes
    mic: bytes
    eapol: bytes


@dataclass(frozen=True)
class HashRecord:
    line: str
    ap: bytes
    client: bytes
    ssid: bytes
    pair_name: str
    packet_numbers: tuple[int, int]


@dataclass
class Capture:
    packet_count: int
    eapol_count: int
    ssids: dict[bytes, set[bytes]]
    messages: dict[tuple[bytes, bytes], list[EapolMessage]]


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Convert Marauder WPA/WPA2 EAPOL PCAP files to Hashcat 22000."
    )
    parser.add_argument("input", type=Path, help="Input .pcap file")
    parser.add_argument(
        "output",
        type=Path,
        nargs="?",
        help="Output .22000 file. Default is next to the input file.",
    )
    parser.add_argument("--ssid", help="Only export this exact SSID")
    parser.add_argument("--bssid", help="Only export this AP MAC address")
    parser.add_argument(
        "--list",
        action="store_true",
        help="List networks with usable EAPOL pairs and do not write a file",
    )
    parser.add_argument(
        "--max-gap",
        type=float,
        default=5.0,
        metavar="SECONDS",
        help="Maximum time between paired EAPOL messages. Default is 5 seconds.",
    )
    parser.add_argument(
        "--force", action="store_true", help="Overwrite an existing output file"
    )
    return parser.parse_args()


def format_mac(value: bytes) -> str:
    return ":".join(f"{part:02X}" for part in value)


def normalize_mac(value: str) -> bytes:
    compact = value.replace(":", "").replace("-", "").strip()
    if len(compact) != 12:
        raise ConversionError(f"Invalid BSSID: {value}")
    try:
        return bytes.fromhex(compact)
    except ValueError as exc:
        raise ConversionError(f"Invalid BSSID: {value}") from exc


def display_ssid(value: bytes) -> str:
    return value.decode("utf-8", "backslashreplace")


def read_pcap_header(handle: BinaryIO) -> PcapInfo:
    header = handle.read(24)
    if len(header) != 24:
        raise ConversionError("The file is too short to contain a PCAP header")

    magic = header[:4]
    formats = {
        b"\xd4\xc3\xb2\xa1": ("<", False),
        b"\xa1\xb2\xc3\xd4": (">", False),
        b"\x4d\x3c\xb2\xa1": ("<", True),
        b"\xa1\xb2\x3c\x4d": (">", True),
    }
    if magic not in formats:
        raise ConversionError("Unsupported file format. Classic PCAP is required")

    endian, nanosecond_timestamps = formats[magic]
    linktype = struct.unpack(f"{endian}I", header[20:24])[0]
    if linktype not in (DLT_IEEE802_11, DLT_IEEE802_11_RADIO):
        raise ConversionError(
            f"Unsupported PCAP link type {linktype}. Expected raw 802.11 or radiotap"
        )
    return PcapInfo(endian, nanosecond_timestamps, linktype)


def unwrap_80211(packet: bytes, linktype: int) -> bytes | None:
    if linktype == DLT_IEEE802_11:
        return packet
    if len(packet) < 8:
        return None
    radiotap_length = int.from_bytes(packet[2:4], "little")
    if radiotap_length < 8 or radiotap_length > len(packet):
        return None
    return packet[radiotap_length:]


def tagged_ssid(frame: bytes, start: int) -> bytes | None:
    position = start
    while position + 2 <= len(frame):
        tag_number = frame[position]
        tag_length = frame[position + 1]
        position += 2
        end = position + tag_length
        if end > len(frame):
            return None
        if tag_number == 0:
            value = frame[position:end]
            return value if 1 <= len(value) <= 32 else None
        position = end
    return None


def collect_ssid(frame: bytes, ssids: dict[bytes, set[bytes]]) -> None:
    if len(frame) < 24:
        return
    frame_control = int.from_bytes(frame[0:2], "little")
    frame_type = (frame_control >> 2) & 0x03
    subtype = (frame_control >> 4) & 0x0F
    if frame_type != 0:
        return

    if subtype in (5, 8) and len(frame) >= 36:
        bssid = frame[16:22]
        value = tagged_ssid(frame, 36)
    elif subtype == 0 and len(frame) >= 28:
        bssid = frame[4:10]
        value = tagged_ssid(frame, 28)
    elif subtype == 2 and len(frame) >= 34:
        bssid = frame[4:10]
        value = tagged_ssid(frame, 34)
    elif subtype == 4 and len(frame) >= 24 and frame[4:10] != BROADCAST:
        bssid = frame[4:10]
        value = tagged_ssid(frame, 24)
    else:
        return

    if value is not None:
        ssids[bssid].add(value)


def classify_eapol(frame: bytes, packet_number: int, timestamp: float) -> EapolMessage | None:
    if len(frame) < 24:
        return None

    snap_position = frame.find(EAPOL_SNAP, 20, min(len(frame), 128))
    if snap_position < 0:
        return None

    eapol_position = snap_position + len(EAPOL_SNAP)
    if eapol_position + 99 > len(frame):
        return None
    if frame[eapol_position + 1] != 3:
        return None

    body_length = int.from_bytes(frame[eapol_position + 2 : eapol_position + 4], "big")
    eapol_length = body_length + 4
    if body_length < 95 or eapol_length > 256:
        return None
    if eapol_position + eapol_length > len(frame):
        return None

    eapol = frame[eapol_position : eapol_position + eapol_length]
    key_info = int.from_bytes(eapol[5:7], "big")
    if not key_info & 0x0008:
        return None

    key_ack = bool(key_info & 0x0080)
    key_mic = bool(key_info & 0x0100)
    secure = bool(key_info & 0x0200)
    if key_ack and not key_mic:
        message_number = 1
    elif not key_ack and key_mic and not secure:
        message_number = 2
    elif key_ack and key_mic:
        message_number = 3
    elif not key_ack and key_mic and secure:
        message_number = 4
    else:
        return None

    receiver = frame[4:10]
    transmitter = frame[10:16]
    if key_ack:
        ap, client = transmitter, receiver
    else:
        ap, client = receiver, transmitter

    return EapolMessage(
        packet_number=packet_number,
        timestamp=timestamp,
        number=message_number,
        ap=ap,
        client=client,
        replay_counter=int.from_bytes(eapol[9:17], "big"),
        key_info=key_info,
        nonce=eapol[17:49],
        mic=eapol[81:97],
        eapol=eapol,
    )


def parse_capture(path: Path) -> Capture:
    ssids: dict[bytes, set[bytes]] = defaultdict(set)
    messages: dict[tuple[bytes, bytes], list[EapolMessage]] = defaultdict(list)
    packet_count = 0
    eapol_count = 0

    with path.open("rb", buffering=1024 * 1024) as handle:
        info = read_pcap_header(handle)
        timestamp_divisor = 1_000_000_000 if info.nanosecond_timestamps else 1_000_000

        while True:
            packet_header = handle.read(16)
            if not packet_header:
                break
            if len(packet_header) != 16:
                raise ConversionError("Truncated PCAP packet header")
            seconds, fraction, captured_length, _ = struct.unpack(
                f"{info.endian}IIII", packet_header
            )
            if captured_length > 16 * 1024 * 1024:
                raise ConversionError("Unreasonable packet size in PCAP")
            packet = handle.read(captured_length)
            if len(packet) != captured_length:
                raise ConversionError("Truncated packet data in PCAP")

            packet_count += 1
            frame = unwrap_80211(packet, info.linktype)
            if frame is None:
                continue
            collect_ssid(frame, ssids)
            timestamp = seconds + fraction / timestamp_divisor
            message = classify_eapol(frame, packet_count, timestamp)
            if message is not None:
                messages[(message.ap, message.client)].append(message)
                eapol_count += 1

    return Capture(packet_count, eapol_count, dict(ssids), dict(messages))


def zero_mic(eapol: bytes) -> bytes:
    result = bytearray(eapol)
    result[81:97] = ZERO_MIC
    return bytes(result)


def make_hash_record(
    m2: EapolMessage,
    nonce_source: EapolMessage,
    ssid: bytes,
    pair_code: str,
    pair_name: str,
) -> HashRecord:
    if m2.number != 2 or nonce_source.number not in (1, 3):
        raise ConversionError("Internal EAPOL pairing error")
    if m2.mic == ZERO_MIC or m2.nonce == ZERO_NONCE or nonce_source.nonce == ZERO_NONCE:
        raise ConversionError("EAPOL pair contains an empty nonce or MIC")

    line = "*".join(
        (
            "WPA",
            "02",
            m2.mic.hex(),
            m2.ap.hex(),
            m2.client.hex(),
            ssid.hex(),
            nonce_source.nonce.hex(),
            zero_mic(m2.eapol).hex(),
            pair_code,
        )
    )
    validate_hash_line(line)
    return HashRecord(
        line=line,
        ap=m2.ap,
        client=m2.client,
        ssid=ssid,
        pair_name=pair_name,
        packet_numbers=tuple(sorted((nonce_source.packet_number, m2.packet_number))),
    )


def validate_hash_line(line: str) -> None:
    fields = line.split("*")
    if len(fields) != 9 or fields[0:2] != ["WPA", "02"]:
        raise ConversionError("Generated an invalid 22000 field layout")
    mic, ap, client, ssid_hex, anonce, eapol_hex, pair_code = fields[2:]
    expected_lengths = ((mic, 32), (ap, 12), (client, 12), (anonce, 64), (pair_code, 2))
    if any(len(value) != length for value, length in expected_lengths):
        raise ConversionError("Generated an invalid 22000 field length")
    if not 2 <= len(ssid_hex) <= 64 or len(ssid_hex) % 2:
        raise ConversionError("Generated an invalid SSID field")
    try:
        eapol = bytes.fromhex(eapol_hex)
        bytes.fromhex("".join((mic, ap, client, ssid_hex, anonce, pair_code)))
    except ValueError as exc:
        raise ConversionError("Generated a non-hexadecimal 22000 field") from exc
    if len(eapol) < 99 or len(eapol) > 256:
        raise ConversionError("Generated an invalid EAPOL length")
    if int.from_bytes(eapol[2:4], "big") + 4 != len(eapol):
        raise ConversionError("Generated EAPOL length does not match its header")
    if eapol[81:97] != ZERO_MIC:
        raise ConversionError("Generated EAPOL data contains a non-zero MIC")
    if pair_code not in ("00", "02"):
        raise ConversionError("Generated an unsupported EAPOL message pair")


def find_pair_records(
    messages: Iterable[EapolMessage], ssid: bytes, max_gap: float
) -> list[HashRecord]:
    ordered = sorted(messages, key=lambda item: (item.timestamp, item.packet_number))
    m1_messages = [item for item in ordered if item.number == 1]
    m2_messages = [item for item in ordered if item.number == 2]
    m3_messages = [item for item in ordered if item.number == 3]
    authorized: list[HashRecord] = []
    challenge: list[HashRecord] = []

    # A busy capture can contain several overlapping attempts with replay
    # counters reset to the same values. Pair each M3 only with the closest
    # preceding M2 so an old M2 cannot borrow a later session's ANonce.
    for m3 in m3_messages:
        matching_m2 = [
            m2
            for m2 in m2_messages
            if m3.replay_counter == m2.replay_counter + 1
            and 0 <= m3.timestamp - m2.timestamp <= max_gap
        ]
        if matching_m2:
            m2 = min(matching_m2, key=lambda item: m3.timestamp - item.timestamp)
            following_m3 = [
                candidate
                for candidate in m3_messages
                if candidate.replay_counter == m2.replay_counter + 1
                and 0 <= candidate.timestamp - m2.timestamp <= max_gap
            ]
            nearest_m3 = min(
                following_m3, key=lambda item: item.timestamp - m2.timestamp
            )
            if nearest_m3 == m3:
                authorized.append(make_hash_record(m2, m3, ssid, "02", "M2+M3"))

    if authorized:
        preferred = authorized
    else:
        for m2 in m2_messages:
            matching_m1 = [
                m1
                for m1 in m1_messages
                if m1.replay_counter == m2.replay_counter
                and 0 <= m2.timestamp - m1.timestamp <= max_gap
            ]
            if matching_m1:
                m1 = min(matching_m1, key=lambda item: m2.timestamp - item.timestamp)
                challenge.append(make_hash_record(m2, m1, ssid, "00", "M1+M2"))
        preferred = challenge

    unique: dict[str, HashRecord] = {}
    for record in preferred:
        unique.setdefault(record.line, record)
    return list(unique.values())


def network_records(capture: Capture, max_gap: float) -> dict[tuple[bytes, bytes], list[HashRecord]]:
    result: dict[tuple[bytes, bytes], list[HashRecord]] = defaultdict(list)
    for (ap, _client), messages in capture.messages.items():
        for ssid in capture.ssids.get(ap, set()):
            records = find_pair_records(messages, ssid, max_gap)
            if records:
                result[(ap, ssid)].extend(records)
    return dict(result)


def print_networks(records: dict[tuple[bytes, bytes], list[HashRecord]]) -> list[tuple[bytes, bytes]]:
    networks = sorted(records, key=lambda item: (display_ssid(item[1]).casefold(), item[0]))
    if not networks:
        print("No networks with usable M1+M2 or M2+M3 pairs were found")
        return []
    print("Networks with usable EAPOL pairs")
    for index, (ap, ssid) in enumerate(networks, 1):
        pair_names = ", ".join(sorted({record.pair_name for record in records[(ap, ssid)]}))
        clients = len({record.client for record in records[(ap, ssid)]})
        print(
            f"[{index}] {display_ssid(ssid)}  {format_mac(ap)}  "
            f"clients={clients}  pairs={pair_names}"
        )
    return networks


def select_networks(
    records: dict[tuple[bytes, bytes], list[HashRecord]],
    ssid_filter: str | None,
    bssid_filter: str | None,
    list_only: bool,
) -> list[tuple[bytes, bytes]]:
    networks = print_networks(records)
    if list_only or not networks:
        return []

    selected = networks
    if ssid_filter is not None:
        wanted_ssid = ssid_filter.encode("utf-8")
        selected = [network for network in selected if network[1] == wanted_ssid]
    if bssid_filter is not None:
        wanted_bssid = normalize_mac(bssid_filter)
        selected = [network for network in selected if network[0] == wanted_bssid]

    if ssid_filter is not None or bssid_filter is not None:
        if not selected:
            raise ConversionError("The requested SSID or BSSID has no usable EAPOL pair")
        return selected

    if len(networks) == 1:
        return networks
    if not sys.stdin.isatty():
        raise ConversionError("Several networks were found. Specify --ssid or --bssid")

    while True:
        answer = input("Select your network number: ").strip()
        try:
            index = int(answer)
        except ValueError:
            continue
        if 1 <= index <= len(networks):
            return [networks[index - 1]]


def write_records(path: Path, records: list[HashRecord], force: bool) -> None:
    if path.exists() and not force:
        raise ConversionError(f"Output file already exists: {path}. Use --force to replace it")
    path.parent.mkdir(parents=True, exist_ok=True)
    content = "\n".join(record.line for record in records) + "\n"
    path.write_text(content, encoding="ascii", newline="\n")


def main() -> int:
    args = parse_args()
    if args.max_gap <= 0:
        raise ConversionError("--max-gap must be greater than zero")
    input_path = args.input.resolve()
    if not input_path.is_file():
        raise ConversionError(f"Input file does not exist: {input_path}")

    capture = parse_capture(input_path)
    records_by_network = network_records(capture, args.max_gap)
    print(
        f"Read {capture.packet_count} packets and {capture.eapol_count} EAPOL key messages"
    )
    selected_networks = select_networks(
        records_by_network, args.ssid, args.bssid, args.list
    )
    if args.list or not selected_networks:
        return 0 if records_by_network else 2

    selected_records: list[HashRecord] = []
    for network in selected_networks:
        selected_records.extend(records_by_network[network])
    unique_records = list({record.line: record for record in selected_records}.values())
    output_path = (args.output or input_path.with_suffix(".22000")).resolve()
    write_records(output_path, unique_records, args.force)

    print(f"Wrote {len(unique_records)} Hashcat 22000 record(s) to {output_path}")
    for record in unique_records:
        packets = ",".join(str(number) for number in record.packet_numbers)
        print(
            f"{display_ssid(record.ssid)}  {format_mac(record.ap)}  "
            f"{format_mac(record.client)}  {record.pair_name}  packets={packets}"
        )
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (ConversionError, OSError) as error:
        print(f"Error: {error}", file=sys.stderr)
        raise SystemExit(1)

@echo off
setlocal

if "%~1"=="" (
  echo Drag a Marauder PCAP file onto this CMD file.
  pause
  exit /b 1
)

py -3 "%~dp0pcap_to_22000.py" "%~1"
set "result=%errorlevel%"
echo.
if not "%result%"=="0" echo Conversion failed with code %result%.
pause
exit /b %result%

@echo off
setlocal

set "HASHCAT_DIR=D:\Flipper\Marauder\Hashcat\hashcat-7.1.2"
set "HASHCAT=%HASHCAT_DIR%\hashcat.exe"
set "WORDLIST=D:\Flipper\Marauder\wordlist.txt"
set "HASHFILE=%~1"

if not defined HASHFILE set "HASHFILE=D:\Flipper\Marauder\logs\sniffpmkid_5.22000"

if not exist "%HASHCAT%" (
  echo Hashcat was not found at
  echo %HASHCAT%
  pause
  exit /b 2
)

if not exist "%HASHFILE%" (
  echo Hash file was not found at
  echo %HASHFILE%
  pause
  exit /b 2
)

if not exist "%WORDLIST%" (
  echo Wordlist was not found at
  echo %WORDLIST%
  pause
  exit /b 2
)

echo Hash file
echo %HASHFILE%
echo.
echo Wordlist
echo %WORDLIST%
echo.

pushd "%HASHCAT_DIR%"
"%HASHCAT%" -m 22000 -a 0 "%HASHFILE%" "%WORDLIST%" --status --status-timer=5
set "RESULT=%ERRORLEVEL%"

echo.
if "%RESULT%"=="0" echo Hashcat reports a recovered password.
if "%RESULT%"=="1" echo The wordlist is exhausted. No new password was found.
if not "%RESULT%"=="0" if not "%RESULT%"=="1" echo Hashcat stopped with code %RESULT%.

echo.
echo Saved result
"%HASHCAT%" -m 22000 "%HASHFILE%" --show
popd

echo.
pause
exit /b %RESULT%

@echo off
setlocal

set "HASHCAT_DIR=D:\Flipper\Marauder\Hashcat\hashcat-7.1.2"
set "HASHCAT=%HASHCAT_DIR%\hashcat.exe"
set "HASHFILE=%~1"

if not defined HASHFILE set "HASHFILE=D:\Flipper\Marauder\logs\sniffpmkid_10.22000"

if not exist "%HASHCAT%" (
  echo Hashcat was not found at
  echo %HASHCAT%
  pause
  exit /b 2
)

if not exist "%HASHFILE%" (
  echo Hash file was not found at
  echo %HASHFILE%
  pause
  exit /b 2
)

echo Hash file
echo %HASHFILE%
echo.
echo Mask
echo Eight digits
echo.
echo Keyspace
echo 100000000 candidates
echo.

pushd "%HASHCAT_DIR%"
"%HASHCAT%" -m 22000 -a 3 "%HASHFILE%" "?d?d?d?d?d?d?d?d" --status --status-timer=5
set "RESULT=%ERRORLEVEL%"

echo.
if "%RESULT%"=="0" echo Hashcat reports a recovered password.
if "%RESULT%"=="1" echo The entire mask is exhausted. No new password was found.
if not "%RESULT%"=="0" if not "%RESULT%"=="1" echo Hashcat stopped with code %RESULT%.

echo.
echo Saved result
"%HASHCAT%" -m 22000 "%HASHFILE%" --show
popd

echo.
pause
exit /b %RESULT%