Project: Packet Header Parser တစ်ခု တည်ဆောက်ခြင်း

import struct
import ipaddress
import binascii

PROTOCOLS = {1: "ICMP", 6: "TCP", 17: "UDP"}

# TCP flag bits, low bit first, as they sit in the 9-bit flags field.
TCP_FLAGS = [
    (0x001, "FIN"), (0x002, "SYN"), (0x004, "RST"), (0x008, "PSH"),
    (0x010, "ACK"), (0x020, "URG"), (0x040, "ECE"), (0x080, "CWR"),
    (0x100, "NS"),
]


def checksum16(data):
    """The standard Internet checksum: one's complement of the one's
    complement sum of all 16-bit words."""
    if len(data) % 2:
        data += b"\x00"
    total = 0
    for (word,) in struct.iter_unpack("!H", data):
        total += word
    while total >> 16:
        total = (total & 0xFFFF) + (total >> 16)
    return ~total & 0xFFFF


def build_ipv4_header(src, dst, payload_len, ttl=64, proto=6, ident=0x1C46):
    version_ihl = (4 << 4) | 5          # version 4, IHL 5 words = 20 bytes
    total_len = 20 + payload_len
    flags_frag = 0x4000                 # Don't Fragment, offset 0
    header = struct.pack(
        "!BBHHHBBH4s4s",
        version_ihl, 0, total_len, ident, flags_frag, ttl, proto, 0,
        ipaddress.IPv4Address(src).packed,
        ipaddress.IPv4Address(dst).packed,
    )
    csum = checksum16(header)           # computed with the checksum field zeroed
    return header[:10] + struct.pack("!H", csum) + header[12:]


def build_tcp_header(sport, dport, seq, ack, flags, window=64240):
    offset_reserved = (5 << 12) | flags   # data offset 5 words = 20 bytes
    return struct.pack("!HHIIHHHH", sport, dport, seq, ack,
                       offset_reserved, window, 0, 0)


def parse_ipv4(data):
    (version_ihl, tos, total_len, ident, flags_frag, ttl, proto,
     csum, src, dst) = struct.unpack("!BBHHHBBH4s4s", data[:20])
    return {
        "version": version_ihl >> 4,
        "ihl_words": version_ihl & 0x0F,
        "tos": tos,
        "total_length": total_len,
        "id": ident,
        "df": bool(flags_frag & 0x4000),
        "mf": bool(flags_frag & 0x2000),
        "ttl": ttl,
        "protocol": proto,
        "checksum": csum,
        "src": str(ipaddress.IPv4Address(src)),
        "dst": str(ipaddress.IPv4Address(dst)),
    }


def parse_tcp(data):
    (sport, dport, seq, ack, offset_flags, window, csum, urg) = \
        struct.unpack("!HHIIHHHH", data[:20])
    return {
        "sport": sport,
        "dport": dport,
        "seq": seq,
        "ack": ack,
        "offset_words": offset_flags >> 12,
        "flags": offset_flags & 0x01FF,
        "window": window,
        "checksum": csum,
        "urgent": urg,
    }


def flag_names(flags):
    return [name for bit, name in TCP_FLAGS if flags & bit]


def report(packet):
    ip = parse_ipv4(packet)
    ip_len = ip["ihl_words"] * 4
    tcp = parse_tcp(packet[ip_len:])

    print("PACKET HEADER REPORT")
    print("=" * 64)
    print("Raw bytes (%d total):" % len(packet))
    hexs = binascii.hexlify(packet).decode()
    for i in range(0, len(hexs), 32):
        print("  %04x  %s" % (i // 2, " ".join(
            hexs[j:j + 4] for j in range(i, min(i + 32, len(hexs)), 4))))
    print()

    print("-- LAYER 3: IPv4 " + "-" * 47)
    print("  Version          : %d" % ip["version"])
    print("  IHL              : %d words (%d bytes)" % (ip["ihl_words"], ip_len))
    print("  Total length     : %d bytes (payload %d)" % (ip["total_length"],
                                                          ip["total_length"] - ip_len))
    print("  Identification   : 0x%04x" % ip["id"])
    print("  Flags            : DF=%d MF=%d" % (ip["df"], ip["mf"]))
    print("  TTL              : %d" % ip["ttl"])
    print("  Protocol         : %d (%s)" % (ip["protocol"],
                                            PROTOCOLS.get(ip["protocol"], "unknown")))
    print("  Header checksum  : 0x%04x" % ip["checksum"])
    # A correct header re-checksums to zero, because the stored value is
    # already the complement of the sum of every other word.
    verify = checksum16(packet[:ip_len])
    print("  Checksum verify  : %s (recomputed 0x%04x)" %
          ("VALID" if verify == 0 else "INVALID", verify))
    print("  Source IP        : %s" % ip["src"])
    print("  Destination IP   : %s" % ip["dst"])
    print()

    print("-- LAYER 4: TCP " + "-" * 48)
    print("  Source port      : %d" % tcp["sport"])
    print("  Destination port : %d" % tcp["dport"])
    print("  Sequence number  : %d (0x%08x)" % (tcp["seq"], tcp["seq"]))
    print("  Ack number       : %d" % tcp["ack"])
    print("  Data offset      : %d words (%d bytes)" % (tcp["offset_words"],
                                                        tcp["offset_words"] * 4))
    print("  Flags            : 0x%03x -> %s" % (tcp["flags"],
                                                 " ".join(flag_names(tcp["flags"])) or "none"))
    for bit, name in TCP_FLAGS:
        print("      %-4s %d" % (name, 1 if tcp["flags"] & bit else 0))
    print("  Window           : %d" % tcp["window"])
    print()

    print("-- THE 4-TUPLE " + "-" * 49)
    print("  %s:%d  ->  %s:%d" % (ip["src"], tcp["sport"], ip["dst"], tcp["dport"]))
    print("  protocol %s" % PROTOCOLS.get(ip["protocol"], "unknown"))


tcp_header = build_tcp_header(50912, 443, 0x9ECF1D2A, 0, flags=0x002)  # SYN
ip_header = build_ipv4_header("192.168.1.10", "93.184.216.34", len(tcp_header))
report(ip_header + tcp_header)