# $Id: sctp.py 23 2006-11-08 15:45:33Z dugsong $ # -*- coding: utf-8 -*- """Stream Control Transmission Protocol.""" from __future__ import print_function from __future__ import absolute_import from . import dpkt from . import crc32c # Stream Control Transmission Protocol # http://tools.ietf.org/html/rfc2960 # Chunk Types DATA = 0 INIT = 1 INIT_ACK = 2 SACK = 3 HEARTBEAT = 4 HEARTBEAT_ACK = 5 ABORT = 6 SHUTDOWN = 7 SHUTDOWN_ACK = 8 ERROR = 9 COOKIE_ECHO = 10 COOKIE_ACK = 11 ECNE = 12 CWR = 13 SHUTDOWN_COMPLETE = 14 class SCTP(dpkt.Packet): """Stream Control Transmission Protocol. TODO: Longer class information.... Attributes: __hdr__: Header fields of SCTP. TODO. """ __hdr__ = ( ('sport', 'H', 0), ('dport', 'H', 0), ('vtag', 'I', 0), ('sum', 'I', 0) ) def unpack(self, buf): dpkt.Packet.unpack(self, buf) l_ = [] while self.data: chunk = Chunk(self.data) l_.append(chunk) if len(chunk) == 0: self.data = b'' break self.data = self.data[len(chunk):] self.chunks = l_ def __len__(self): return self.__hdr_len__ + sum(len(x) for x in self.chunks) def __bytes__(self): l_ = [bytes(x) for x in self.chunks] if self.sum == 0: s = crc32c.add(0xffffffff, self.pack_hdr()) for x in l_: s = crc32c.add(s, x) self.sum = crc32c.done(s) return self.pack_hdr() + b''.join(l_) class Chunk(dpkt.Packet): __hdr__ = ( ('type', 'B', INIT), ('flags', 'B', 0), ('len', 'H', 0) ) def unpack(self, buf): dpkt.Packet.unpack(self, buf) self.data = self.data[:self.len - self.__hdr_len__] self.padding = b'' # optional padding for DATA chunks # SCTP DATA Chunked is padded, 4-bytes aligned if self.type == DATA and self.len % 4: plen = 4 - self.len % 4 # padded length if plen: pos = self.__hdr_len__ + len(self.data) # end of data in buf self.padding = buf[pos:pos + plen] def __len__(self): return self.len + len(self.padding) def __bytes__(self): return self.pack_hdr() + bytes(self.data) + self.padding __s = (b'\x80\x44\x00\x50\x00\x00\x00\x00\x30\xba\xef\x54\x01\x00\x00\x3c\x3b\xb9\x9c\x46\x00\x01' b'\xa0\x00\x00\x0a\xff\xff\x2b\x2d\x7e\xb2\x00\x05\x00\x08\x9b\xe6\x18\x9b\x00\x05\x00\x08' b'\x9b\xe6\x18\x9c\x00\x0c\x00\x06\x00\x05\x00\x00\x80\x00\x00\x04\xc0\x00\x00\x04\xc0\x06' b'\x00\x08\x00\x00\x00\x00') def test_sctp_pack(): sctp = SCTP(__s) assert (__s == bytes(sctp)) sctp.sum = 0 assert (__s == bytes(sctp)) def test_sctp_unpack(): sctp = SCTP(__s) assert (sctp.sport == 32836) assert (sctp.dport == 80) assert (len(sctp.chunks) == 1) assert (len(sctp) == 72) chunk = sctp.chunks[0] assert (chunk.type == INIT) assert (chunk.len == 60) def test_sctp_data_chunk(): # https://github.com/kbandla/dpkt/issues/499 # packet 5 from 'sctp-www.cap' downloaded from https://wiki.wireshark.org/SampleCaptures # chunk len == 419 so requires padding to a 4-byte boundary d = (b'\x80\x44\x00\x50\xd2\x6a\xc1\xe5\x70\xe5\x5b\x4c\x00\x03\x01\xa3\x2b\x2d\x7e\xb2\x00\x00' b'\x00\x00\x00\x00\x00\x00\x47\x45\x54\x20\x2f\x20\x48\x54\x54\x50\x2f\x31\x2e\x31\x0d\x0a' b'\x48\x6f\x73\x74\x3a\x20\x32\x30\x33\x2e\x32\x35\x35\x2e\x32\x35\x32\x2e\x31\x39\x34\x0d' b'\x0a\x55\x73\x65\x72\x2d\x41\x67\x65\x6e\x74\x3a\x20\x4d\x6f\x7a\x69\x6c\x6c\x61\x2f\x35' b'\x2e\x30\x20\x28\x58\x31\x31\x3b\x20\x55\x3b\x20\x4c\x69\x6e\x75\x78\x20\x69\x36\x38\x36' b'\x3b\x20\x6b\x6f\x2d\x4b\x52\x3b\x20\x72\x76\x3a\x31\x2e\x37\x2e\x31\x32\x29\x20\x47\x65' b'\x63\x6b\x6f\x2f\x32\x30\x30\x35\x31\x30\x30\x37\x20\x44\x65\x62\x69\x61\x6e\x2f\x31\x2e' b'\x37\x2e\x31\x32\x2d\x31\x0d\x0a\x41\x63\x63\x65\x70\x74\x3a\x20\x74\x65\x78\x74\x2f\x78' b'\x6d\x6c\x2c\x61\x70\x70\x6c\x69\x63\x61\x74\x69\x6f\x6e\x2f\x78\x6d\x6c\x2c\x61\x70\x70' b'\x6c\x69\x63\x61\x74\x69\x6f\x6e\x2f\x78\x68\x74\x6d\x6c\x2b\x78\x6d\x6c\x2c\x74\x65\x78' b'\x74\x2f\x68\x74\x6d\x6c\x3b\x71\x3d\x30\x2e\x39\x2c\x74\x65\x78\x74\x2f\x70\x6c\x61\x69' b'\x6e\x3b\x71\x3d\x30\x2e\x38\x2c\x69\x6d\x61\x67\x65\x2f\x70\x6e\x67\x2c\x2a\x2f\x2a\x3b' b'\x71\x3d\x30\x2e\x35\x0d\x0a\x41\x63\x63\x65\x70\x74\x2d\x4c\x61\x6e\x67\x75\x61\x67\x65' b'\x3a\x20\x6b\x6f\x2c\x65\x6e\x2d\x75\x73\x3b\x71\x3d\x30\x2e\x37\x2c\x65\x6e\x3b\x71\x3d' b'\x30\x2e\x33\x0d\x0a\x41\x63\x63\x65\x70\x74\x2d\x45\x6e\x63\x6f\x64\x69\x6e\x67\x3a\x20' b'\x67\x7a\x69\x70\x2c\x64\x65\x66\x6c\x61\x74\x65\x0d\x0a\x41\x63\x63\x65\x70\x74\x2d\x43' b'\x68\x61\x72\x73\x65\x74\x3a\x20\x45\x55\x43\x2d\x4b\x52\x2c\x75\x74\x66\x2d\x38\x3b\x71' b'\x3d\x30\x2e\x37\x2c\x2a\x3b\x71\x3d\x30\x2e\x37\x0d\x0a\x4b\x65\x65\x70\x2d\x41\x6c\x69' b'\x76\x65\x3a\x20\x33\x30\x30\x0d\x0a\x43\x6f\x6e\x6e\x65\x63\x74\x69\x6f\x6e\x3a\x20\x6b' b'\x65\x65\x70\x2d\x61\x6c\x69\x76\x65\x0d\x0a\x0d\x0a\x00') # <-- ends with \x00 padding sctp = SCTP(d) assert sctp.chunks assert len(sctp.chunks) == 1 ch = sctp.chunks[0] assert ch.type == DATA assert ch.len == 419 assert len(ch) == 420 # 419 +1 byte padding assert ch.data[-14:] == b'keep-alive\r\n\r\n' # no padding byte at the end # no remaining sctp data assert sctp.data == b'' # test packing of the padded chunk assert bytes(ch) == d[SCTP.__hdr_len__:] def test_malformed_sctp_data_chunk(): # packet 7964 from '4.pcap' downloaded from https://research.unsw.edu.au/projects/unsw-nb15-dataset d = (b'\x27\x0f\xe1\xc3\xc2\x73\x4d\x32\x4f\x54\x27\x8c' #header b'\x0b\x00\x00\x04' #chunk 0, COOKIE_ACK chunk b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00') #chunk 1, malformed DATA chunk, size labeled as 0 sctp = SCTP(d) assert sctp.chunks assert len(sctp.chunks) == 2 ch = sctp.chunks[1] assert ch.type == DATA assert ch.len == 0 assert len(ch) == 0 assert ch.data == b'\x00\x00' # no remaining sctp data assert sctp.data == b''