Source code for dpkt.ip

# $Id: ip.py 87 2013-03-05 19:41:04Z [email protected] $
# -*- coding: utf-8 -*-
"""Internet Protocol."""
from __future__ import print_function
from __future__ import absolute_import

from . import dpkt
from .decorators import deprecated
from .compat import iteritems

[docs]class IP(dpkt.Packet): """Internet Protocol. TODO: Longer class information.... Attributes: __hdr__: Header fields of IP. TODO. """ __hdr__ = ( ('_v_hl', 'B', (4 << 4) | (20 >> 2)), ('tos', 'B', 0), ('len', 'H', 20), ('id', 'H', 0), ('off', 'H', 0), ('ttl', 'B', 64), ('p', 'B', 0), ('sum', 'H', 0), ('src', '4s', b'\x00' * 4), ('dst', '4s', b'\x00' * 4) ) _protosw = {} opts = b'' def __init__(self, *args, **kwargs): super(IP, self).__init__(*args, **kwargs) # If IP packet is not initialized by string and the len field has # been rewritten. if not args and 'len' not in kwargs: self.len = self.__len__() @property def v(self): return self._v_hl >> 4 @v.setter def v(self, v): self._v_hl = (v << 4) | (self._v_hl & 0xf) @property def hl(self): return self._v_hl & 0xf @hl.setter def hl(self, hl): self._v_hl = (self._v_hl & 0xf0) | hl @property def rf(self): return (self.off >> 15) & 0x1 @rf.setter def rf(self, rf): self.off = (self.off & ~IP_RF) | (rf << 15) @property def df(self): return (self.off >> 14) & 0x1 @df.setter def df(self, df): self.off = (self.off & ~IP_DF) | (df << 14) @property def mf(self): return (self.off >> 13) & 0x1 @mf.setter def mf(self, mf): self.off = (self.off & ~IP_MF) | (mf << 13) @property def offset(self): return (self.off & IP_OFFMASK) << 3 @offset.setter def offset(self, offset): self.off = (self.off & ~IP_OFFMASK) | (offset >> 3) def __len__(self): return self.__hdr_len__ + len(self.opts) + len(self.data) def __bytes__(self): self.len = self.__len__() if self.sum == 0: self.sum = dpkt.in_cksum(self.pack_hdr() + bytes(self.opts)) if (self.p == 6 or self.p == 17) and (self.off & (IP_MF | IP_OFFMASK)) == 0 and \ isinstance(self.data, dpkt.Packet) and self.data.sum == 0: # Set zeroed TCP and UDP checksums for non-fragments. p = bytes(self.data) s = dpkt.struct.pack('>4s4sxBH', self.src, self.dst, self.p, len(p)) s = dpkt.in_cksum_add(0, s) s = dpkt.in_cksum_add(s, p) self.data.sum = dpkt.in_cksum_done(s) if self.p == 17 and self.data.sum == 0: self.data.sum = 0xffff # RFC 768 # XXX - skip transports which don't need the pseudoheader return self.pack_hdr() + bytes(self.opts) + bytes(self.data)
[docs] def unpack(self, buf): dpkt.Packet.unpack(self, buf) ol = ((self._v_hl & 0xf) << 2) - self.__hdr_len__ if ol < 0: raise dpkt.UnpackError('invalid header length') self.opts = buf[self.__hdr_len__:self.__hdr_len__ + ol] if self.len: buf = buf[self.__hdr_len__ + ol:self.len] else: # very likely due to TCP segmentation offload buf = buf[self.__hdr_len__ + ol:] try: self.data = self._protosw[self.p](buf) if self.offset == 0 else buf setattr(self, self.data.__class__.__name__.lower(), self.data) except (KeyError, dpkt.UnpackError): self.data = buf
[docs] @classmethod def set_proto(cls, p, pktclass): cls._protosw[p] = pktclass
[docs] @classmethod def get_proto(cls, p): return cls._protosw[p]
# IP Headers IP_ADDR_LEN = 0x04 IP_ADDR_BITS = 0x20 IP_HDR_LEN = 0x14 IP_OPT_LEN = 0x02 IP_OPT_LEN_MAX = 0x28 IP_HDR_LEN_MAX = IP_HDR_LEN + IP_OPT_LEN_MAX IP_LEN_MAX = 0xffff IP_LEN_MIN = IP_HDR_LEN # Reserved Addresses IP_ADDR_ANY = "\x00\x00\x00\x00" # 0.0.0.0 IP_ADDR_BROADCAST = "\xff\xff\xff\xff" # 255.255.255.255 IP_ADDR_LOOPBACK = "\x7f\x00\x00\x01" # 127.0.0.1 IP_ADDR_MCAST_ALL = "\xe0\x00\x00\x01" # 224.0.0.1 IP_ADDR_MCAST_LOCAL = "\xe0\x00\x00\xff" # 224.0.0.255 # Type of service (ip_tos), RFC 1349 ("obsoleted by RFC 2474") IP_TOS_DEFAULT = 0x00 # default IP_TOS_LOWDELAY = 0x10 # low delay IP_TOS_THROUGHPUT = 0x08 # high throughput IP_TOS_RELIABILITY = 0x04 # high reliability IP_TOS_LOWCOST = 0x02 # low monetary cost - XXX IP_TOS_ECT = 0x02 # ECN-capable transport IP_TOS_CE = 0x01 # congestion experienced # IP precedence (high 3 bits of ip_tos), hopefully unused IP_TOS_PREC_ROUTINE = 0x00 IP_TOS_PREC_PRIORITY = 0x20 IP_TOS_PREC_IMMEDIATE = 0x40 IP_TOS_PREC_FLASH = 0x60 IP_TOS_PREC_FLASHOVERRIDE = 0x80 IP_TOS_PREC_CRITIC_ECP = 0xa0 IP_TOS_PREC_INTERNETCONTROL = 0xc0 IP_TOS_PREC_NETCONTROL = 0xe0 # Fragmentation flags (ip_off) IP_RF = 0x8000 # reserved IP_DF = 0x4000 # don't fragment IP_MF = 0x2000 # more fragments (not last frag) IP_OFFMASK = 0x1fff # mask for fragment offset # Time-to-live (ip_ttl), seconds IP_TTL_DEFAULT = 64 # default ttl, RFC 1122, RFC 1340 IP_TTL_MAX = 255 # maximum ttl # Protocol (ip_p) - http://www.iana.org/assignments/protocol-numbers IP_PROTO_IP = 0 # dummy for IP IP_PROTO_HOPOPTS = IP_PROTO_IP # IPv6 hop-by-hop options IP_PROTO_ICMP = 1 # ICMP IP_PROTO_IGMP = 2 # IGMP IP_PROTO_GGP = 3 # gateway-gateway protocol IP_PROTO_IPIP = 4 # IP in IP IP_PROTO_ST = 5 # ST datagram mode IP_PROTO_TCP = 6 # TCP IP_PROTO_CBT = 7 # CBT IP_PROTO_EGP = 8 # exterior gateway protocol IP_PROTO_IGP = 9 # interior gateway protocol IP_PROTO_BBNRCC = 10 # BBN RCC monitoring IP_PROTO_NVP = 11 # Network Voice Protocol IP_PROTO_PUP = 12 # PARC universal packet IP_PROTO_ARGUS = 13 # ARGUS IP_PROTO_EMCON = 14 # EMCON IP_PROTO_XNET = 15 # Cross Net Debugger IP_PROTO_CHAOS = 16 # Chaos IP_PROTO_UDP = 17 # UDP IP_PROTO_MUX = 18 # multiplexing IP_PROTO_DCNMEAS = 19 # DCN measurement IP_PROTO_HMP = 20 # Host Monitoring Protocol IP_PROTO_PRM = 21 # Packet Radio Measurement IP_PROTO_IDP = 22 # Xerox NS IDP IP_PROTO_TRUNK1 = 23 # Trunk-1 IP_PROTO_TRUNK2 = 24 # Trunk-2 IP_PROTO_LEAF1 = 25 # Leaf-1 IP_PROTO_LEAF2 = 26 # Leaf-2 IP_PROTO_RDP = 27 # "Reliable Datagram" proto IP_PROTO_IRTP = 28 # Inet Reliable Transaction IP_PROTO_TP = 29 # ISO TP class 4 IP_PROTO_NETBLT = 30 # Bulk Data Transfer IP_PROTO_MFPNSP = 31 # MFE Network Services IP_PROTO_MERITINP = 32 # Merit Internodal Protocol IP_PROTO_SEP = 33 # Sequential Exchange proto IP_PROTO_3PC = 34 # Third Party Connect proto IP_PROTO_IDPR = 35 # Interdomain Policy Route IP_PROTO_XTP = 36 # Xpress Transfer Protocol IP_PROTO_DDP = 37 # Datagram Delivery Proto IP_PROTO_CMTP = 38 # IDPR Ctrl Message Trans IP_PROTO_TPPP = 39 # TP++ Transport Protocol IP_PROTO_IL = 40 # IL Transport Protocol IP_PROTO_IP6 = 41 # IPv6 IP_PROTO_SDRP = 42 # Source Demand Routing IP_PROTO_ROUTING = 43 # IPv6 routing header IP_PROTO_FRAGMENT = 44 # IPv6 fragmentation header IP_PROTO_RSVP = 46 # Reservation protocol IP_PROTO_GRE = 47 # General Routing Encap IP_PROTO_MHRP = 48 # Mobile Host Routing IP_PROTO_ENA = 49 # ENA IP_PROTO_ESP = 50 # Encap Security Payload IP_PROTO_AH = 51 # Authentication Header IP_PROTO_INLSP = 52 # Integated Net Layer Sec IP_PROTO_SWIPE = 53 # SWIPE IP_PROTO_NARP = 54 # NBMA Address Resolution IP_PROTO_MOBILE = 55 # Mobile IP, RFC 2004 IP_PROTO_TLSP = 56 # Transport Layer Security IP_PROTO_SKIP = 57 # SKIP IP_PROTO_ICMP6 = 58 # ICMP for IPv6 IP_PROTO_NONE = 59 # IPv6 no next header IP_PROTO_DSTOPTS = 60 # IPv6 destination options IP_PROTO_ANYHOST = 61 # any host internal proto IP_PROTO_CFTP = 62 # CFTP IP_PROTO_ANYNET = 63 # any local network IP_PROTO_EXPAK = 64 # SATNET and Backroom EXPAK IP_PROTO_KRYPTOLAN = 65 # Kryptolan IP_PROTO_RVD = 66 # MIT Remote Virtual Disk IP_PROTO_IPPC = 67 # Inet Pluribus Packet Core IP_PROTO_DISTFS = 68 # any distributed fs IP_PROTO_SATMON = 69 # SATNET Monitoring IP_PROTO_VISA = 70 # VISA Protocol IP_PROTO_IPCV = 71 # Inet Packet Core Utility IP_PROTO_CPNX = 72 # Comp Proto Net Executive IP_PROTO_CPHB = 73 # Comp Protocol Heart Beat IP_PROTO_WSN = 74 # Wang Span Network IP_PROTO_PVP = 75 # Packet Video Protocol IP_PROTO_BRSATMON = 76 # Backroom SATNET Monitor IP_PROTO_SUNND = 77 # SUN ND Protocol IP_PROTO_WBMON = 78 # WIDEBAND Monitoring IP_PROTO_WBEXPAK = 79 # WIDEBAND EXPAK IP_PROTO_EON = 80 # ISO CNLP IP_PROTO_VMTP = 81 # Versatile Msg Transport IP_PROTO_SVMTP = 82 # Secure VMTP IP_PROTO_VINES = 83 # VINES IP_PROTO_TTP = 84 # TTP IP_PROTO_NSFIGP = 85 # NSFNET-IGP IP_PROTO_DGP = 86 # Dissimilar Gateway Proto IP_PROTO_TCF = 87 # TCF IP_PROTO_EIGRP = 88 # EIGRP IP_PROTO_OSPF = 89 # Open Shortest Path First IP_PROTO_SPRITERPC = 90 # Sprite RPC Protocol IP_PROTO_LARP = 91 # Locus Address Resolution IP_PROTO_MTP = 92 # Multicast Transport Proto IP_PROTO_AX25 = 93 # AX.25 Frames IP_PROTO_IPIPENCAP = 94 # yet-another IP encap IP_PROTO_MICP = 95 # Mobile Internet Ctrl IP_PROTO_SCCSP = 96 # Semaphore Comm Sec Proto IP_PROTO_ETHERIP = 97 # Ethernet in IPv4 IP_PROTO_ENCAP = 98 # encapsulation header IP_PROTO_ANYENC = 99 # private encryption scheme IP_PROTO_GMTP = 100 # GMTP IP_PROTO_IFMP = 101 # Ipsilon Flow Mgmt Proto IP_PROTO_PNNI = 102 # PNNI over IP IP_PROTO_PIM = 103 # Protocol Indep Multicast IP_PROTO_ARIS = 104 # ARIS IP_PROTO_SCPS = 105 # SCPS IP_PROTO_QNX = 106 # QNX IP_PROTO_AN = 107 # Active Networks IP_PROTO_IPCOMP = 108 # IP Payload Compression IP_PROTO_SNP = 109 # Sitara Networks Protocol IP_PROTO_COMPAQPEER = 110 # Compaq Peer Protocol IP_PROTO_IPXIP = 111 # IPX in IP IP_PROTO_VRRP = 112 # Virtual Router Redundancy IP_PROTO_PGM = 113 # PGM Reliable Transport IP_PROTO_ANY0HOP = 114 # 0-hop protocol IP_PROTO_L2TP = 115 # Layer 2 Tunneling Proto IP_PROTO_DDX = 116 # D-II Data Exchange (DDX) IP_PROTO_IATP = 117 # Interactive Agent Xfer IP_PROTO_STP = 118 # Schedule Transfer Proto IP_PROTO_SRP = 119 # SpectraLink Radio Proto IP_PROTO_UTI = 120 # UTI IP_PROTO_SMP = 121 # Simple Message Protocol IP_PROTO_SM = 122 # SM IP_PROTO_PTP = 123 # Performance Transparency IP_PROTO_ISIS = 124 # ISIS over IPv4 IP_PROTO_FIRE = 125 # FIRE IP_PROTO_CRTP = 126 # Combat Radio Transport IP_PROTO_CRUDP = 127 # Combat Radio UDP IP_PROTO_SSCOPMCE = 128 # SSCOPMCE IP_PROTO_IPLT = 129 # IPLT IP_PROTO_SPS = 130 # Secure Packet Shield IP_PROTO_PIPE = 131 # Private IP Encap in IP IP_PROTO_SCTP = 132 # Stream Ctrl Transmission IP_PROTO_FC = 133 # Fibre Channel IP_PROTO_RSVPIGN = 134 # RSVP-E2E-IGNORE IP_PROTO_RAW = 255 # Raw IP packets IP_PROTO_RESERVED = IP_PROTO_RAW # Reserved IP_PROTO_MAX = 255 # XXX - auto-load IP dispatch table from IP_PROTO_* definitions def __load_protos(): g = globals() for k, v in iteritems(g): if k.startswith('IP_PROTO_'): name = k[9:].lower() try: mod = __import__(name, g, level=1) IP.set_proto(v, getattr(mod, name.upper())) except (ImportError, AttributeError): continue def _mod_init(): """Post-initialization called when all dpkt modules are fully loaded""" if not IP._protosw: __load_protos()
[docs]def test_ip(): from . import udp s = b'E\x00\x00"\x00\x00\x00\x00@\x11r\xc0\x01\x02\x03\x04\x01\x02\x03\x04\x00o\x00\xde\x00\x0e\xbf5foobar' ip = IP(id=0, src=b'\x01\x02\x03\x04', dst=b'\x01\x02\x03\x04', p=17) u = udp.UDP(sport=111, dport=222) u.data = b'foobar' u.ulen += len(u.data) ip.data = u ip.len += len(u) assert (bytes(ip) == s) assert (ip.v == 4) assert (ip.hl == 5) ip = IP(s) assert (bytes(ip) == s) assert (ip.udp.sport == 111) assert (ip.udp.data == b'foobar')
[docs]def test_hl(): # Todo chack this test method s = b'BB\x03\x00\x00\x00\x00\x00\x00\x00\xd0\x00\xec\xbc\xa5\x00\x00\x00\x03\x80\x00\x00\xd0\x01\xf2\xac\xa5"0\x01\x00\x14\x00\x02\x00\x0f\x00\x00\x00\x00\x00\x00\x00\x00\x00' try: IP(s) except dpkt.UnpackError: pass
[docs]def test_opt(): s = b'\x4f\x00\x00\x3c\xae\x08\x00\x00\x40\x06\x18\x10\xc0\xa8\x0a\x26\xc0\xa8\x0a\x01\x07\x27\x08\x01\x02\x03\x04\x01\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00' ip = IP(s) ip.sum = 0 assert (bytes(ip) == s)
[docs]def test_zerolen(): from . import tcp d = b'X' * 2048 s = b'E\x00\x00\x004\xce@\x00\x80\x06\x00\x00\x7f\x00\x00\x01\x7f\x00\x00\x01\xccN\x0c8`\xff\xc6N_\x8a\x12\x98P\x18@):\xa3\x00\x00' + d ip = IP(s) assert (isinstance(ip.data, tcp.TCP)) assert (ip.tcp.data == d)
[docs]def test_constuctor(): ip1 = IP(data = b"Hello world!") ip2 = IP(data = b"Hello world!", len = 0) ip3 = IP(bytes(ip1)) ip4 = IP(bytes(ip2)) assert (bytes(ip1) == bytes(ip3)) assert (bytes(ip1) == b'E\x00\x00 \x00\x00\x00\x00@\x00z\xdf\x00\x00\x00\x00\x00\x00\x00\x00Hello world!') assert (bytes(ip2) == bytes(ip4)) assert (bytes(ip2) == b'E\x00\x00 \x00\x00\x00\x00@\x00z\xdf\x00\x00\x00\x00\x00\x00\x00\x00Hello world!')
[docs]def test_frag(): from . import ethernet s = b"\x00\x23\x20\xd4\x2a\x8c\x00\x23\x20\xd4\x2a\x8c\x08\x00\x45\x00\x00\x54\x00\x00\x40\x00\x40\x01\x25\x8d\x0a\x00\x00\x8f\x0a\x00\x00\x8e\x08\x00\x2e\xa0\x01\xff\x23\x73\x20\x48\x4a\x4d\x00\x00\x00\x00\x78\x85\x02\x00\x00\x00\x00\x00\x10\x11\x12\x13\x14\x15\x16\x17\x18\x19\x1a\x1b\x1c\x1d\x1e\x1f\x20\x21\x22\x23\x24\x25\x26\x27\x28\x29\x2a\x2b\x2c\x2d\x2e\x2f\x30\x31\x32\x33\x34\x35\x36\x37" ip = ethernet.Ethernet(s).ip assert (ip.rf == 0) assert (ip.df == 1) assert (ip.mf == 0) assert (ip.offset == 0) # test setters of fragmentation related attributes. ip.rf = 1 ip.df = 0 ip.mf = 1 ip.offset = 1480 assert (ip.rf == 1) assert (ip.df == 0) assert (ip.mf == 1) assert (ip.offset == 1480)
if __name__ == '__main__': test_ip() test_hl() test_opt() test_zerolen() test_constuctor() test_frag() print('Tests Successful...')