tun packet decoder

tuntap_tun.py

import struct

import bitstream
import dotdict
import numpy as np
import pytun

def read_uint(stream, n=None):
	integer = 0
	for _ in range(n):
		integer <<= 1
		if stream.read(bool):
			integer += 1
	return integer
bitstream.register(int, read_uint)

def hexprint(buf):
	pbuf = buf
	while True:
		if len(pbuf) == 0:
			break

		linesize = min(len(pbuf), 8)
		for i in range(8):
			if i < linesize:
				print("%02x " % (pbuf[i],), end="")
			else:
				print("   ", end="")
		for i in range(linesize):
			print("%s" % (chr(pbuf[i]) if 32 <= pbuf[i] <= 127 else '.'), end="")
		print("")

		pbuf = pbuf[linesize:]

def decode_ip(bs):
	ip = dotdict.dotdict()

	ip.version = bs.read(int, 4)
	ip.ihl = bs.read(int, 4) * 4
	ip.tos = str(bs.read(n=8))
	ip.length = bs.read(np.uint16)

	ip.id = bs.read(np.uint16)
	ip.flags = bs.read(int, 3)
	ip.frag = bs.read(int, 13) * 8

	ip.ttl = bs.read(np.uint8)
	ip.proto = bs.read(np.uint8)
	ip.checksum = bs.read(np.uint16)

	ip.src = '.'.join(bs.read(np.uint8, 4).astype(str))
	ip.dst = '.'.join(bs.read(np.uint8, 4).astype(str))

	ip.options_padding = bs.read(n=ip.ihl-20)

	print("IPv%d %s -> %s (%d+%d)" % (ip.version, ip.src, ip.dst, ip.ihl, ip.length-ip.ihl))

	return ip

def decode_udp(bs):
	udp = dotdict.dotdict()

	udp.sport = bs.read(np.uint16)
	udp.dport = bs.read(np.uint16)
	udp.length = bs.read(np.uint16) - 8
	udp.checksum = bs.read(np.uint16)
	udp.data = bs.read(bytes, udp.length)

	print("UDP %d -> %d (8+%d)" % (udp.sport, udp.dport, udp.length))

	return udp

def main():
	tun = pytun.TunTapDevice('test1')

	while True:
		buf = tun.read(4096)

		print("---")

		bs = bitstream.BitStream(buf)
		tuntap_flags = bs.read(np.uint16)
		tuntap_proto = bs.read(np.uint16)
		print("Flags: %04x Proto: %04x" % (tuntap_flags, tuntap_proto))

		print("-")
		hexprint(buf[4:])
		print("-")

		ip = decode_ip(bs)
		if ip.proto == 17:
			udp = decode_udp(bs)
			data = udp.data

		print("-")
		hexprint(data)

if __name__ == '__main__':
	main()