Protocol Reverse Engineering

from scapy.all import *
# Read pcappackets = rdpcap("capture.pcap")
# Analyze packetsfor pkt in packets:    if pkt.haslayer(TCP):        print(f"Src: {pkt[IP].src}:{pkt[TCP].sport}")        print(f"Dst: {pkt[IP].dst}:{pkt[TCP].dport}")        if pkt.haslayer(Raw):            print(f"Data: {pkt[Raw].load[:50]}")
# Filter packetshttp_packets = [p for p in packets if p.haslayer(TCP)                and (p[TCP].sport == 80 or p[TCP].dport == 80)]
# Create custom packetspkt = IP(dst="target")/TCP(dport=80)/Raw(load="GET / HTTP/1.1\r\n")send(pkt)

# Common patterns in binary protocols
# Length-prefixed messagestruct Message {    uint32_t length;      # Total message length    uint16_t msg_type;    # Message type identifier    uint8_t  flags;       # Flags/options    uint8_t  reserved;    # Padding/alignment    uint8_t  payload[];   # Variable-length payload};
# Type-Length-Value (TLV)struct TLV {    uint8_t  type;        # Field type    uint16_t length;      # Field length    uint8_t  value[];     # Field data};
# Fixed header + variable payloadstruct Packet {    uint8_t  magic[4];    # "ABCD" signature    uint32_t version;    uint32_t payload_len;    uint32_t checksum;    # CRC32 or similar    uint8_t  payload[];};

import structfrom dataclasses import dataclass
@dataclassclass MessageHeader:    magic: bytes    version: int    msg_type: int    length: int
    @classmethod    def from_bytes(cls, data: bytes):        magic, version, msg_type, length = struct.unpack(            ">4sHHI", data[:12]        )        return cls(magic, version, msg_type, length)
def parse_messages(data: bytes):    offset = 0    messages = []
    while offset < len(data):        header = MessageHeader.from_bytes(data[offset:])        payload = data[offset+12:offset+12+header.length]        messages.append((header, payload))        offset += 12 + header.length
    return messages
# Parse TLV structuredef parse_tlv(data: bytes):    fields = []    offset = 0
    while offset < len(data):        field_type = data[offset]        length = struct.unpack(">H", data[offset+1:offset+3])[0]        value = data[offset+3:offset+3+length]        fields.append((field_type, value))        offset += 3 + length
    return fields

def hexdump(data: bytes, width: int = 16):    """Format binary data as hex dump."""    lines = []    for i in range(0, len(data), width):        chunk = data[i:i+width]        hex_part = ' '.join(f'{b:02x}' for b in chunk)        ascii_part = ''.join(            chr(b) if 32 <= b < 127 else '.'            for b in chunk        )        lines.append(f'{i:08x}  {hex_part:<{width*3}}  {ascii_part}')    return '\n'.join(lines)
# Example output:# 00000000  48 54 54 50 2f 31 2e 31  20 32 30 30 20 4f 4b 0d  HTTP/1.1 200 OK.# 00000010  0a 43 6f 6e 74 65 6e 74  2d 54 79 70 65 3a 20 74  .Content-Type: t

# Entropy analysis - high entropy suggests encryption/compressionimport mathfrom collections import Counter
def entropy(data: bytes) -> float:    if not data:        return 0.0    counter = Counter(data)    probs = [count / len(data) for count in counter.values()]    return -sum(p * math.log2(p) for p in probs)
# Entropy thresholds:# < 6.0: Likely plaintext or structured data# 6.0-7.5: Possibly compressed# > 7.5: Likely encrypted or random
# Common encryption indicators# - High, uniform entropy# - No obvious structure or patterns# - Length often multiple of block size (16 for AES)# - Possible IV at start (16 bytes for AES-CBC)

from boofuzz import *
def main():    session = Session(        target=Target(            connection=TCPSocketConnection("target", 8888)        )    )
    # Define protocol structure    s_initialize("HELLO")    s_static(b"\x50\x52\x4f\x54")  # Magic    s_word(1, name="version")       # Version    s_word(0x01, name="type")       # Type (HELLO)    s_size("payload", length=4)     # Length field    s_block_start("payload")    s_dword(0x12345678, name="client_id")    s_word(0, name="flags")    s_block_end()
    session.connect(s_get("HELLO"))    session.fuzz()
if __name__ == "__main__":    main()

from scapy.all import *
# Replay captured trafficpackets = rdpcap("capture.pcap")for pkt in packets:    if pkt.haslayer(TCP) and pkt[TCP].dport == 8888:        send(pkt)
# Modify and replayfor pkt in packets:    if pkt.haslayer(Raw):        # Modify payload        original = pkt[Raw].load        modified = original.replace(b"client", b"CLIENT")        pkt[Raw].load = modified        # Recalculate checksums        del pkt[IP].chksum        del pkt[TCP].chksum        send(pkt)