Ethernet II

https://zh.wikipedia.org/wiki/%E4%BB%A5%E5%A4%AA%E7%BD%91%E5%B8%A7%E6%A0%BC%E5%BC%8F#Ethernet_II

• 以太网帧

Struct	Size(Bytes)	Usage	Example
Destination	6	目的地址	33:33:00:01:00:03
Source	6	源地址	00:50:56:c0:00:08
Type	2	类型	0x8600

Type	Description
0x0800	IPV4
0x0806	ARP
0x0835	RARP
0x86DD	IPV6

IPv4

Internet Protocol

https://zh.wikipedia.org/zh-hans/IPv4

Struct	Size(bits)	Description	Example
Version	4	IP版本（IPv4必为4）	0100
Header Length	4	头部32位长个数（最小结构为5）	0101
Differentiated Service	6	区分服务（实时数据流时使用）	000000
Explicit Congestion Notification	2	显式阻塞通告	00
Total Length	16	全长（首部+数据）	0x003c
Identification	16	识别码（全局，自动加一）	0xc181
Flags	3	控制识别分片	000
Fragment Offset	13	每个分片相对于原始报文开头的偏移量，以8字节作单位	0000000000000
Time to Live	8	存活时长（经过路由器减一）	0x80
Protocol	8	协议类型	0x01
Header Checksum	16	头部校验和	0x4769
Source Address	32	源地址	0xc0a85801
Destinstaion	32	目的地址	0xc0a85884

Index	Description
0	保留，必须为0
1	为1时禁止分片，为0时允许分片
2	为1时后续还有分片，为0时为最后一个分片

Protocol详情：https://zh.wikipedia.org/zh-hans/IP%E5%8D%8F%E8%AE%AE%E5%8F%B7%E5%88%97%E8%A1%A8

• 手动构造

def checksum(data: bytes) -> bytes:
    data = data.hex()
    c = 0
    for i in range(len(data) // 4):
        c += int(data[i * 4: (i + 1) * 4], 16)
        c = (c & 0xffff) + 1 if c > 0xffff else c
    c = hex(~c & 0xffff)[2:]
    c = c.rjust((len(c) // 2 + 1) * 2, "0") if len(c) % 2 else c
    return bytes.fromhex(c)


something = b""  # such as icmp

ip = b""
ip += b"\x45"  # version + header length
ip += b"\x00"  # differentiated services
ip += b"\x00\x3c"  # total length
ip += b"\xc1\x81"  # identification
ip += b"\x00\x00"  # flags + fragment offset
ip += b"\x80"  # ttl
ip += b"\x01"  # protocol
ip += b"\x00\x00"  # header checksum
ip += bytes.fromhex(''.join([hex(int(i))[2:].rjust(2, '0') for i in "192.168.88.1".split(".")]))  # source
ip += bytes.fromhex(''.join([hex(int(i))[2:].rjust(2, '0') for i in "192.168.88.132".split(".")]))  # destination
ip += something  # such as icmp
ip = ip[:3] + bytes.fromhex(hex(len(ip))[2:].rjust(4, "0")) + ip[5:]  # reset total length
ip = ip[:10] + checksum(ip[:20]) + ip[12:]  # calc header checksum
print(ip)

• 分块构造

from tools import send, checksum


def ether():
    e = b""
    e += bytes.fromhex("000c29e57f05")  # destination mac
    e += bytes.fromhex("005056c00008")  # source mac
    e += b"\x08\x00"  # type
    return e


def icmp(data: bytes):
    i = b""
    i += b"\x08"  # type
    i += b"\x00"  # code
    i += b"\x00\x00"  # checksum
    i += b"\x12\x34"  # identifier
    i += b"\x56\x78"  # sequence number
    i += data  # data
    i = i[:2] + checksum(i) + i[4:]
    return i


def ipv4(data: bytes):
    blocks = [data[i * 1480: (i + 1) * 1480] for i in range(len(data) // 1480 + 1)]
    blocks = blocks if blocks[-1] else blocks[:-1]
    for i in range(len(blocks)):
        ip = b""
        ip += b"\x45"  # version + header length
        ip += b"\x00"  # differentiated services
        ip += b"\x00\x3c"  # total length
        ip += b"\x00\x00"  # identification
        if i == len(blocks) - 1:
            ip += bytes.fromhex(hex(int("000" + bin(i * 1480 // 8)[2:].rjust(13, "0"), 2))[2:].rjust(4, "0"))
        else:
            ip += bytes.fromhex(hex(int("001" + bin(i * 1480 // 8)[2:].rjust(13, "0"), 2))[2:].rjust(4, "0"))
        ip += b"\x80"  # ttl
        ip += b"\x01"  # protocol
        ip += b"\x00\x00"  # header checksum
        ip += bytes.fromhex(''.join([hex(int(i))[2:].rjust(2, '0') for i in "192.168.88.1".split(".")]))  # source
        ip += bytes.fromhex(''.join([hex(int(i))[2:].rjust(2, '0') for i in "192.168.88.132".split(".")]))  # destination
        ip += blocks[i]  # such as icmp
        ip = ip[:2] + bytes.fromhex(hex(len(ip))[2:].rjust(4, "0")) + ip[4:]  # reset total length
        ip = ip[:10] + checksum(ip[:20]) + ip[12:]  # calc header checksum
        ping = ether() + ip
        send(ping)


a = icmp(b"test" * 789)
ipv4(a)

ICMP

Internet Control Message Protocol

https://zh.wikipedia.org/wiki/%E4%BA%92%E8%81%94%E7%BD%91%E6%8E%A7%E5%88%B6%E6%B6%88%E6%81%AF%E5%8D%8F%E8%AE%AE#ICMP%E6%8A%A5%E6%96%87%E7%B5%90%E6%A7%8B

https://datatracker.ietf.org/doc/html/rfc1071

虽然ICMP是包含在IP数据包中的，但是对ICMP消息通常会特殊处理，会和一般IP数据包的处理不同，而不是作为IP的一个子协议来处理

Package = Ethernet II + IP + ICMP

Struct	Size(Bytes)	Usage	Example	可选
Type	1	类型	0x08
Code	1	代码	0x00
Checksum	2	校验和	0xe9c4
Identifier	2	进程pid（随机）	0x0002
Squence Number	2	序列（递增）	0x6395
TimeStamp	8	时间戳（小端序）	0x1a5bad6400000000	√
Data	32	数据	….

• Checksum：校验和置零，每 2 Bytes 做二进制加法，取反

a = "08000000000263956162636465666768696a6b6c6d6e6f7071727374757677616263646566676869"
c = 0
for i in range(len(a) // 4):
    c += int(a[i * 4: (i + 1) * 4], 16)
    c = (c & 0xffff) + 1 if c > 0xffff else c
icmp = a[:4] + hex(~c & 0xffff)[2:] + a[8:]
print(icmp)

• 手动构造

# Windows ping.exe

package = b""
package += b"\x08"  
package += b"\x00"
package += b"\x00\x00"
package += b"\x12\x34"
package += b"\x56\x78"
package += b"test".rjust(32, b"\x00")[:32]


def checksum(data: bytes) -> bytes:
    data = data.hex()
    c = 0
    for i in range(len(data) // 4):
        c += int(data[i * 4: (i + 1) * 4], 16)
        c = (c & 0xffff) + 1 if c > 0xffff else c
    c = hex(~c & 0xffff)[2:]
    c = c.rjust((len(c) // 2 + 1) * 2, "0") if len(c) % 2 else c
    return bytes.fromhex(c)


print(checksum(package))

# Linux ping
# 在 checksum 后增添小端序的8字节时间戳

from time import time
from test import send
from struct import Struct
from ctypes import create_string_buffer
from numpy import frombuffer, uint8


def little_endian(num: int):
    buffer = create_string_buffer(Struct('<1i').size)
    Struct('<1i').pack_into(buffer, 0, num)
    data = frombuffer(buffer, dtype=uint8)
    return bytes.fromhex(''.join(list(map(lambda x: hex(x)[2:], data))).ljust(16, "0"))


def checksum(data: bytes) -> bytes:
    data = data.hex()
    c = 0
    for i in range(len(data) // 4):
        c += int(data[i * 4: (i + 1) * 4], 16)
        c = (c & 0xffff) + 1 if c > 0xffff else c
    c = hex(~c & 0xffff)[2:]
    c = c.rjust((len(c) // 2 + 1) * 2, "0") if len(c) % 2 else c
    return bytes.fromhex(c)


package = b""
package += b"\x08"  # type
package += b"\x00"  # code
package += b"\x00\x00"  # checksum
package += b"\x12\x34"  # identifier
package += b"\x56\x78"  # sequence
package += little_endian(round(time()))  # timestamp
package += b"test".rjust(48, b"\x00")[:48]  # data

print(checksum(package))

ARP

Address Resolution Protocol

https://zh.wikipedia.org/wiki/%E5%9C%B0%E5%9D%80%E8%A7%A3%E6%9E%90%E5%8D%8F%E8%AE%AE

Package = Ethernet II + ARP

目标以太网地址：目标MAC地址。FF:FF:FF:FF:FF:FF （二进制全1）为广播地址

未知MAC地址：00:00:00:00:00:00

Struct	Size(bytes)	Description	Example
Hardware Type	2	硬件类型	0x0001
Protocol Type	2	协议类型	0x0800
Hardware Size	1	硬件地址长度（Bytes）	0x06
Protocol Size	1	协议地址长度（Bytes）	0x04
Opcode	2	操作码	0x0001
Sender Mac Address	6	源MAC	…
Sender IP Address	4	源IP	…
Target Mac Address	6	目标MAC	…
Target IP Address	4	目标IP	…

Opcode	Description
0x0001	ARP Request
0x0002	ARP Replay
0x0003	RARP Request
0x0004	RARP Replay

• 手动构造

arp = b""
arp += b"\x00\x01"  # hardware type
arp += b"\x08\x00"  # protocol type
arp += b"\x06"  # hardware size
arp += b"\x04"  # protocol size
arp += b"\x00\x01"  # opcode
arp += bytes.fromhex("00:0c:29:e5:7f:05".replace(":", ""))  # sender mac
arp += bytes([int(i) for i in "192.168.88.1".split(".")])  # sender ip
arp += b"\x00\x00\x00\x00\x00\x00"  # target mac
arp += bytes([int(i) for i in "192.168.88.132".split(".")])  # target ip

print(arp)

Test Script

• send.py

from ctypes import create_string_buffer, c_ubyte, POINTER, byref
import libpcap as pcap


class NotExistAdapterError(Exception):

    pass


def finddevs_by_name(name: str):
    alldevs = POINTER(pcap.pcap_if_t)()
    pcap.findalldevs(byref(alldevs), create_string_buffer(256))
    adapters = []
    while alldevs[0].name:
        adapters.append(alldevs[0].description)
        if name.lower() in alldevs[0].description.decode().lower():
            adapter = alldevs[0].name
            pcap.freealldevs(alldevs)
            return adapter
        try:
            alldevs[0] = alldevs[0].next[0]
        except ValueError:
            break
    pcap.freealldevs(alldevs)
    raise NotExistAdapterError(f"No such network adapter: `{name}` in {adapters}")

def send(package: bytes):
    p = pcap.open_live(
        create_string_buffer(finddevs_by_name("VMnet8")),
        65535,
        True,
        100,
        create_string_buffer(256)
    )
    package = (c_ubyte * len(package)).from_buffer(bytearray(package))
    pcap.sendpacket(p, package, len(package))

• ping.py

from send import send


def checksum(data: bytes) -> bytes:
    data = data.hex()
    c = 0
    for i in range(len(data) // 4):
        c += int(data[i * 4: (i + 1) * 4], 16)
        c = (c & 0xffff) + 1 if c > 0xffff else c
    c = hex(~c & 0xffff)[2:]
    c = c.rjust((len(c) // 2 + 1) * 2, "0") if len(c) % 2 else c
    return bytes.fromhex(c)


ether = b""
ether += bytes.fromhex("000c29e57f05")  # destination
ether += bytes.fromhex("005056c00008")  # source
ether += b"\x08\x00"  # type

icmp = b""
icmp += b"\x08"  # type
icmp += b"\x00"  # code
icmp += b"\x00\x00"  # checksum
icmp += b"\x12\x34"  # identifier
icmp += b"\x56\x78"  # sequence number
icmp += b"test".rjust(32, b"\x00")[:32]  # data
icmp = icmp[:2] + checksum(icmp) + icmp[4:]

ip = b""
ip += b"\x45"  # version + header length
ip += b"\x00"  # differentiated services
ip += b"\x00\x3c"  # total length
ip += b"\xb4\x5e"  # identification
ip += b"\x00\x00"  # flags + fragment offset
ip += b"\x80"  # ttl
ip += b"\x01"  # protocol
ip += b"\x00\x00"  # header checksum
ip += bytes.fromhex(''.join([hex(int(i))[2:].rjust(2, '0') for i in "192.168.88.1".split(".")]))  # source
ip += bytes.fromhex(''.join([hex(int(i))[2:].rjust(2, '0') for i in "192.168.88.132".split(".")]))  # destination
ip += icmp  # such as icmp
ip = ip[:2] + bytes.fromhex(hex(len(ip))[2:].rjust(4, "0")) + ip[4:]  # reset total length
ip = ip[:10] + checksum(ip[:20]) + ip[12:]  # calc header checksum

ping = ether + ip
send(ping)

• arp.py

from send import send


ether = b""
ether += bytes.fromhex("ffffffffffff")  # destination mac
ether += bytes.fromhex("005056c00008")  # source mac
ether += b"\x08\x06"  # type

arp = b""
arp += b"\x00\x01"  # hardware type
arp += b"\x08\x00"  # protocol type
arp += b"\x06"  # hardware size
arp += b"\x04"  # protocol size
arp += b"\x00\x01"  # opcode
arp += bytes.fromhex("00:0c:29:e5:7f:05".replace(":", ""))  # sender mac
arp += bytes([int(i) for i in "192.168.88.1".split(".")])  # sender ip
arp += b"\xff\xff\xff\xff\xff\xff"  # target mac
arp += bytes([int(i) for i in "192.168.88.132".split(".")])  # target ip

arp = ether + arp
send(arp)