Source code for dpkt.ssl

# $Id: ssl.py 90 2014-04-02 22:06:23Z [email protected] $
# Portion Copyright 2012 Google Inc. All rights reserved.
# -*- coding: utf-8 -*-
"""Secure Sockets Layer / Transport Layer Security."""
from __future__ import absolute_import

import struct
import binascii

from . import dpkt
from . import ssl_ciphersuites

#
# Note from April 2011: [email protected] added code that parses SSL3/TLS messages more in depth.
#
# Jul 2012: [email protected] modified and extended SSL support further.
#

# SSL 2.0 is deprecated in RFC 6176
[docs]class SSL2(dpkt.Packet): __hdr__ = ( ('len', 'H', 0), ('msg', 's', ''), ('pad', 's', ''), )
[docs] def unpack(self, buf): dpkt.Packet.unpack(self, buf) if self.len & 0x8000: n = self.len = self.len & 0x7FFF self.msg, self.data = self.data[:n], self.data[n:] else: n = self.len = self.len & 0x3FFF padlen = ord(self.data[0]) self.msg = self.data[1:1 + n] self.pad = self.data[1 + n:1 + n + padlen] self.data = self.data[1 + n + padlen:]
# SSL 3.0 is deprecated in RFC 7568 # Use class TLS for >= SSL 3.0
[docs]class TLS(dpkt.Packet): __hdr__ = ( ('type', 'B', ''), ('version', 'H', ''), ('len', 'H', ''), ) def __init__(self, *args, **kwargs): self.records = [] dpkt.Packet.__init__(self, *args, **kwargs)
[docs] def unpack(self, buf): dpkt.Packet.unpack(self, buf) pointer = 0 while len(self.data[pointer:]) > 0: end = pointer + 5 + struct.unpack("!H", buf[pointer+3:pointer+5])[0] self.records.append(TLSRecord(buf[pointer:end])) pointer = end self.data = self.data[pointer:]
# SSLv3/TLS versions SSL3_V = 0x0300 TLS1_V = 0x0301 TLS11_V = 0x0302 TLS12_V = 0x0303 ssl3_versions_str = { SSL3_V: 'SSL3', TLS1_V: 'TLS 1.0', TLS11_V: 'TLS 1.1', TLS12_V: 'TLS 1.2' } SSL3_VERSION_BYTES = set((b'\x03\x00', b'\x03\x01', b'\x03\x02', b'\x03\x03')) # Alert levels SSL3_AD_WARNING = 1 SSL3_AD_FATAL = 2 alert_level_str = { SSL3_AD_WARNING: 'SSL3_AD_WARNING', SSL3_AD_FATAL: 'SSL3_AD_FATAL' } # SSL3 alert descriptions SSL3_AD_CLOSE_NOTIFY = 0 SSL3_AD_UNEXPECTED_MESSAGE = 10 # fatal SSL3_AD_BAD_RECORD_MAC = 20 # fatal SSL3_AD_DECOMPRESSION_FAILURE = 30 # fatal SSL3_AD_HANDSHAKE_FAILURE = 40 # fatal SSL3_AD_NO_CERTIFICATE = 41 SSL3_AD_BAD_CERTIFICATE = 42 SSL3_AD_UNSUPPORTED_CERTIFICATE = 43 SSL3_AD_CERTIFICATE_REVOKED = 44 SSL3_AD_CERTIFICATE_EXPIRED = 45 SSL3_AD_CERTIFICATE_UNKNOWN = 46 SSL3_AD_ILLEGAL_PARAMETER = 47 # fatal # TLS1 alert descriptions TLS1_AD_DECRYPTION_FAILED = 21 TLS1_AD_RECORD_OVERFLOW = 22 TLS1_AD_UNKNOWN_CA = 48 # fatal TLS1_AD_ACCESS_DENIED = 49 # fatal TLS1_AD_DECODE_ERROR = 50 # fatal TLS1_AD_DECRYPT_ERROR = 51 TLS1_AD_EXPORT_RESTRICTION = 60 # fatal TLS1_AD_PROTOCOL_VERSION = 70 # fatal TLS1_AD_INSUFFICIENT_SECURITY = 71 # fatal TLS1_AD_INTERNAL_ERROR = 80 # fatal TLS1_AD_USER_CANCELLED = 90 TLS1_AD_NO_RENEGOTIATION = 100 # /* codes 110-114 are from RFC3546 */ TLS1_AD_UNSUPPORTED_EXTENSION = 110 TLS1_AD_CERTIFICATE_UNOBTAINABLE = 111 TLS1_AD_UNRECOGNIZED_NAME = 112 TLS1_AD_BAD_CERTIFICATE_STATUS_RESPONSE = 113 TLS1_AD_BAD_CERTIFICATE_HASH_VALUE = 114 TLS1_AD_UNKNOWN_PSK_IDENTITY = 115 # fatal # Mapping alert types to strings alert_description_str = { SSL3_AD_CLOSE_NOTIFY: 'SSL3_AD_CLOSE_NOTIFY', SSL3_AD_UNEXPECTED_MESSAGE: 'SSL3_AD_UNEXPECTED_MESSAGE', SSL3_AD_BAD_RECORD_MAC: 'SSL3_AD_BAD_RECORD_MAC', SSL3_AD_DECOMPRESSION_FAILURE: 'SSL3_AD_DECOMPRESSION_FAILURE', SSL3_AD_HANDSHAKE_FAILURE: 'SSL3_AD_HANDSHAKE_FAILURE', SSL3_AD_NO_CERTIFICATE: 'SSL3_AD_NO_CERTIFICATE', SSL3_AD_BAD_CERTIFICATE: 'SSL3_AD_BAD_CERTIFICATE', SSL3_AD_UNSUPPORTED_CERTIFICATE: 'SSL3_AD_UNSUPPORTED_CERTIFICATE', SSL3_AD_CERTIFICATE_REVOKED: 'SSL3_AD_CERTIFICATE_REVOKED', SSL3_AD_CERTIFICATE_EXPIRED: 'SSL3_AD_CERTIFICATE_EXPIRED', SSL3_AD_CERTIFICATE_UNKNOWN: 'SSL3_AD_CERTIFICATE_UNKNOWN', SSL3_AD_ILLEGAL_PARAMETER: 'SSL3_AD_ILLEGAL_PARAMETER', TLS1_AD_DECRYPTION_FAILED: 'TLS1_AD_DECRYPTION_FAILED', TLS1_AD_RECORD_OVERFLOW: 'TLS1_AD_RECORD_OVERFLOW', TLS1_AD_UNKNOWN_CA: 'TLS1_AD_UNKNOWN_CA', TLS1_AD_ACCESS_DENIED: 'TLS1_AD_ACCESS_DENIED', TLS1_AD_DECODE_ERROR: 'TLS1_AD_DECODE_ERROR', TLS1_AD_DECRYPT_ERROR: 'TLS1_AD_DECRYPT_ERROR', TLS1_AD_EXPORT_RESTRICTION: 'TLS1_AD_EXPORT_RESTRICTION', TLS1_AD_PROTOCOL_VERSION: 'TLS1_AD_PROTOCOL_VERSION', TLS1_AD_INSUFFICIENT_SECURITY: 'TLS1_AD_INSUFFICIENT_SECURITY', TLS1_AD_INTERNAL_ERROR: 'TLS1_AD_INTERNAL_ERROR', TLS1_AD_USER_CANCELLED: 'TLS1_AD_USER_CANCELLED', TLS1_AD_NO_RENEGOTIATION: 'TLS1_AD_NO_RENEGOTIATION', TLS1_AD_UNSUPPORTED_EXTENSION: 'TLS1_AD_UNSUPPORTED_EXTENSION', TLS1_AD_CERTIFICATE_UNOBTAINABLE: 'TLS1_AD_CERTIFICATE_UNOBTAINABLE', TLS1_AD_UNRECOGNIZED_NAME: 'TLS1_AD_UNRECOGNIZED_NAME', TLS1_AD_BAD_CERTIFICATE_STATUS_RESPONSE: 'TLS1_AD_BAD_CERTIFICATE_STATUS_RESPONSE', TLS1_AD_BAD_CERTIFICATE_HASH_VALUE: 'TLS1_AD_BAD_CERTIFICATE_HASH_VALUE', TLS1_AD_UNKNOWN_PSK_IDENTITY: 'TLS1_AD_UNKNOWN_PSK_IDENTITY' } # struct format strings for parsing buffer lengths # don't forget, you have to pad a 3-byte value with \x00 _SIZE_FORMATS = ['!B', '!H', '!I', '!I']
[docs]def parse_variable_array(buf, lenbytes): """ Parse an array described using the 'Type name<x..y>' syntax from the spec Read a length at the start of buf, and returns that many bytes after, in a tuple with the TOTAL bytes consumed (including the size). This does not check that the array is the right length for any given datatype. """ # first have to figure out how to parse length assert lenbytes <= 4 # pretty sure 4 is impossible, too size_format = _SIZE_FORMATS[lenbytes - 1] padding = b'\x00' if lenbytes == 3 else b'' # read off the length size = struct.unpack(size_format, padding + buf[:lenbytes])[0] # read the actual data data = buf[lenbytes:lenbytes + size] # if len(data) != size: insufficient data return data, size + lenbytes
[docs]def parse_extensions(buf): """ Parse TLS extensions in passed buf. Returns an ordered list of extension tuples with ordinal extension type as first value and extension data as second value. Passed buf must start with the 2-byte extensions length TLV. http://www.iana.org/assignments/tls-extensiontype-values/tls-extensiontype-values.xhtml """ extensions_length = struct.unpack('!H', buf[:2])[0] extensions = [] pointer = 2 while pointer < extensions_length: ext_type = struct.unpack('!H', buf[pointer:pointer+2])[0] pointer += 2 ext_data, parsed = parse_variable_array(buf[pointer:], 2) extensions.append((ext_type, ext_data)) pointer += parsed return extensions
[docs]class SSL3Exception(Exception): pass
[docs]class TLSRecord(dpkt.Packet): """ SSLv3 or TLSv1+ packet. In addition to the fields specified in the header, there are compressed and decrypted fields, indicating whether, in the language of the spec, this is a TLSPlaintext, TLSCompressed, or TLSCiphertext. The application will have to figure out when it's appropriate to change these values. """ __hdr__ = ( ('type', 'B', 0), ('version', 'H', 0), ('length', 'H', 0), ) def __init__(self, *args, **kwargs): # assume plaintext unless specified otherwise in arguments self.compressed = kwargs.pop('compressed', False) self.encrypted = kwargs.pop('encrypted', False) # parent constructor dpkt.Packet.__init__(self, *args, **kwargs) # make sure length and data are consistent self.length = len(self.data)
[docs] def unpack(self, buf): dpkt.Packet.unpack(self, buf) header_length = self.__hdr_len__ self.data = buf[header_length:header_length + self.length] # make sure buffer was long enough if len(self.data) != self.length: raise dpkt.NeedData('TLSRecord data was too short.') # assume compressed and encrypted when it's been parsed from # raw data self.compressed = True self.encrypted = True
[docs]class TLSChangeCipherSpec(dpkt.Packet): """ ChangeCipherSpec message is just a single byte with value 1 """ __hdr__ = (('type', 'B', 1),)
[docs]class TLSAppData(str): """ As far as TLSRecord is concerned, AppData is just an opaque blob. """ pass
[docs]class TLSAlert(dpkt.Packet): __hdr__ = ( ('level', 'B', 1), ('description', 'B', 0), )
[docs]class TLSHelloRequest(dpkt.Packet): __hdr__ = tuple()
[docs]class TLSClientHello(dpkt.Packet): __hdr__ = ( ('version', 'H', 0x0301), ('random', '32s', '\x00' * 32), ) # the rest is variable-length and has to be done manually
[docs] def unpack(self, buf): dpkt.Packet.unpack(self, buf) # now session, cipher suites, extensions are in self.data self.session_id, pointer = parse_variable_array(self.data, 1) # print 'pointer',pointer # handle ciphersuites ciphersuites, parsed = parse_variable_array(self.data[pointer:], 2) pointer += parsed self.num_ciphersuites = len(ciphersuites) / 2 # check len(ciphersuites) % 2 == 0 ? # compression methods compression_methods, parsed = parse_variable_array( self.data[pointer:], 1) pointer += parsed self.num_compression_methods = parsed - 1 self.compression_methods = map(ord, compression_methods) # Parse extensions if present if len(self.data[pointer:]) >= 6: self.extensions = parse_extensions(self.data[pointer:])
[docs]class TLSServerHello(dpkt.Packet): __hdr__ = ( ('version', 'H', '0x0301'), ('random', '32s', '\x00' * 32), ) # session is variable, forcing rest to be manual
[docs] def unpack(self, buf): try: dpkt.Packet.unpack(self, buf) self.session_id, pointer = parse_variable_array(self.data, 1) # single cipher suite self.cipher_suite = struct.unpack('!H', self.data[pointer:pointer + 2])[0] pointer += 2 # single compression method self.compression = struct.unpack('!B', self.data[pointer:pointer + 1])[0] pointer += 1 # Parse extensions if present if len(self.data[pointer:]) >= 6: self.extensions = parse_extensions(self.data[pointer:]) except struct.error: # probably data too short raise dpkt.NeedData
[docs]class TLSCertificate(dpkt.Packet): __hdr__ = tuple()
[docs] def unpack(self, buf): try: dpkt.Packet.unpack(self, buf) all_certs, all_certs_len = parse_variable_array(self.data, 3) self.certificates = [] pointer = 3 while pointer < all_certs_len: cert, parsed = parse_variable_array(self.data[pointer:], 3) self.certificates.append((cert)) pointer += parsed except struct.error: raise dpkt.NeedData
[docs]class TLSUnknownHandshake(dpkt.Packet): __hdr__ = tuple()
TLSServerKeyExchange = TLSUnknownHandshake TLSCertificateRequest = TLSUnknownHandshake TLSServerHelloDone = TLSUnknownHandshake TLSCertificateVerify = TLSUnknownHandshake TLSClientKeyExchange = TLSUnknownHandshake TLSFinished = TLSUnknownHandshake # mapping of handshake type ids to their names # and the classes that implement them HANDSHAKE_TYPES = { 0: ('HelloRequest', TLSHelloRequest), 1: ('ClientHello', TLSClientHello), 2: ('ServerHello', TLSServerHello), 11: ('Certificate', TLSCertificate), 12: ('ServerKeyExchange', TLSServerKeyExchange), 13: ('CertificateRequest', TLSCertificateRequest), 14: ('ServerHelloDone', TLSServerHelloDone), 15: ('CertificateVerify', TLSCertificateVerify), 16: ('ClientKeyExchange', TLSClientKeyExchange), 20: ('Finished', TLSFinished), }
[docs]class TLSHandshake(dpkt.Packet): """ A TLS Handshake message This goes for all messages encapsulated in the Record layer, but especially important for handshakes and app data: A message may be spread across a number of TLSRecords, in addition to the possibility of there being more than one in a given Record. You have to put together the contents of TLSRecord's yourself. """ # struct.unpack can't handle the 3-byte int, so we parse it as bytes # (and store it as bytes so dpkt doesn't get confused), and turn it into # an int in a user-facing property __hdr__ = ( ('type', 'B', 0), ('length_bytes', '3s', 0), )
[docs] def unpack(self, buf): dpkt.Packet.unpack(self, buf) # Wait, might there be more than one message of self.type? embedded_type = HANDSHAKE_TYPES.get(self.type, None) if embedded_type is None: raise SSL3Exception('Unknown or invalid handshake type %d' % self.type) # only take the right number of bytes self.data = self.data[:self.length] if len(self.data) != self.length: raise dpkt.NeedData # get class out of embedded_type tuple self.data = embedded_type[1](self.data)
@property def length(self): return struct.unpack('!I', b'\x00' + self.length_bytes)[0]
RECORD_TYPES = { 20: TLSChangeCipherSpec, 21: TLSAlert, 22: TLSHandshake, 23: TLSAppData, }
[docs]class SSLFactory(object): def __new__(cls, buf): v = buf[1:3] if v in SSL3_VERSION_BYTES: return TLSRecord(buf) # SSL2 has no characteristic header or magic bytes, so we just assume # that the msg is an SSL2 msg if it is not detected as SSL3+ return SSL2(buf)
[docs]def tls_multi_factory(buf): """ Attempt to parse one or more TLSRecord's out of buf Args: buf: string containing SSL/TLS messages. May have an incomplete record on the end Returns: [TLSRecord] int, total bytes consumed, != len(buf) if an incomplete record was left at the end. Raises SSL3Exception. """ i, n = 0, len(buf) msgs = [] while i + 5 <= n: v = buf[i + 1:i + 3] if v in SSL3_VERSION_BYTES: try: msg = TLSRecord(buf[i:]) msgs.append(msg) except dpkt.NeedData: break else: raise SSL3Exception('Bad TLS version in buf: %r' % buf[i:i + 5]) i += len(msg) return msgs, i
_hexdecode = binascii.a2b_hex
[docs]class TestTLS(object): """ Test basic TLS functionality. Test that each TLSRecord is correctly discovered and added to TLS.records """
[docs] @classmethod def setup_class(cls): cls.p = TLS(_hexdecode('1603000206010002020303585c2ff72a6599498771f59514f10af68c68f9ef30d0dadc9e1af64d1091476a000084c02bc02cc086c087c009c023c00ac024c072c073c008c007c02fc030c08ac08bc013c027c014c028c076c077c012c011009c009dc07ac07b002f003c0035003d004100ba008400c0000a00050004009e009fc07cc07d003300670039006b004500be008800c4001600a200a3c080c081003200400038006a004400bd008700c3001300660100015500050005010000000000000011000f00000c7777772e69616e612e6f7267ff0100010000230000000a000c000a00130015001700180019000b00020100000d001c001a0401040204030501050306010603030103020303020102020203001500f400f20000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000'))
[docs] def test_records_length(self): assert (len(self.p.records) == 1)
[docs] def test_record_type(self): assert (self.p.records[0].type == 22)
[docs] def test_record_version(self): assert (self.p.records[0].version == 768)
[docs]class TestTLSRecord(object): """ Test basic TLSRecord functionality For this test, the contents of the record doesn't matter, since we're not parsing the next layer. """
[docs] @classmethod def setup_class(cls): # add some extra data, to make sure length is parsed correctly cls.p = TLSRecord(b'\x17\x03\x01\x00\x08abcdefghzzzzzzzzzzz')
[docs] def test_content_type(self): assert (self.p.type == 23)
[docs] def test_version(self): assert (self.p.version == 0x0301)
[docs] def test_length(self): assert (self.p.length == 8)
[docs] def test_data(self): assert (self.p.data == b'abcdefgh')
[docs] def test_initial_flags(self): assert (self.p.compressed == True) assert (self.p.encrypted == True)
[docs] def test_repack(self): p2 = TLSRecord(type=23, version=0x0301, data=b'abcdefgh') assert (p2.type == 23) assert (p2.version == 0x0301) assert (p2.length == 8) assert (p2.data == b'abcdefgh') assert (p2.pack() == self.p.pack())
[docs] def test_total_length(self): # that len(p) includes header assert (len(self.p) == 13)
[docs] def test_raises_need_data_when_buf_is_short(self): import pytest pytest.raises(dpkt.NeedData, TLSRecord, b'\x16\x03\x01\x00\x10abc')
[docs]class TestTLSChangeCipherSpec(object): """It's just a byte. This will be quick, I promise"""
[docs] @classmethod def setup_class(cls): cls.p = TLSChangeCipherSpec(b'\x01')
[docs] def test_parses(self): assert (self.p.type == 1)
[docs] def test_total_length(self): assert (len(self.p) == 1)
[docs]class TestTLSAppData(object): """AppData is basically just a string"""
[docs] def test_value(self): d = TLSAppData('abcdefgh') assert (d == 'abcdefgh')
[docs]class TestTLSHandshake(object):
[docs] @classmethod def setup_class(cls): cls.h = TLSHandshake(b'\x00\x00\x00\x01\xff')
[docs] def test_created_inside_message(self): assert (isinstance(self.h.data, TLSHelloRequest) == True)
[docs] def test_length(self): assert (self.h.length == 0x01)
[docs] def test_raises_need_data(self): import pytest pytest.raises(dpkt.NeedData, TLSHandshake, b'\x00\x00\x01\x01')
[docs]class TestClientHello(object): """This data is extracted from and verified by Wireshark"""
[docs] @classmethod def setup_class(cls): cls.data = _hexdecode( b"01000199" # handshake header b"0301" # version b"5008220ce5e0e78b6891afe204498c9363feffbe03235a2d9e05b7d990eb708d" # rand b"2009bc0192e008e6fa8fe47998fca91311ba30ddde14a9587dc674b11c3d3e5ed1" # session id # cipher suites b"005400ffc00ac0140088008700390038c00fc00500840035c007c009c011c0130045004400330032c00cc00ec002c0040096004100050004002fc008c01200160013c00dc003feff000ac006c010c00bc00100020001" b"0100" # compresssion methods # extensions b"00fc0000000e000c0000096c6f63616c686f7374000a00080006001700180019000b00020100002300d0a50b2e9f618a9ea9bf493ef49b421835cd2f6b05bbe1179d8edf70d58c33d656e8696d36d7e7e0b9d3ecc0e4de339552fa06c64c0fcb550a334bc43944e2739ca342d15a9ebbe981ac87a0d38160507d47af09bdc16c5f0ee4cdceea551539382333226048a026d3a90a0535f4a64236467db8fee22b041af986ad0f253bc369137cd8d8cd061925461d7f4d7895ca9a4181ab554dad50360ac31860e971483877c9335ac1300c5e78f3e56f3b8e0fc16358fcaceefd5c8d8aaae7b35be116f8832856ca61144fcdd95e071b94d0cf7233740000" b"FFFFFFFFFFFFFFFF") # random garbage cls.p = TLSHandshake(cls.data)
[docs] def test_client_hello_constructed(self): """Make sure the correct class was constructed""" # print self.p assert (isinstance(self.p.data, TLSClientHello) == True)
# def testClientDateCorrect(self): # self.assertEqual(self.p.random_unixtime, 1342710284)
[docs] def test_client_random_correct(self): assert (self.p.data.random == _hexdecode(b'5008220ce5e0e78b6891afe204498c9363feffbe03235a2d9e05b7d990eb708d'))
[docs] def test_cipher_suite_length(self): # we won't bother testing the identity of each cipher suite in the list. assert (self.p.data.num_ciphersuites == 42)
# self.assertEqual(len(self.p.ciphersuites), 42)
[docs] def test_session_id(self): assert (self.p.data.session_id == _hexdecode(b'09bc0192e008e6fa8fe47998fca91311ba30ddde14a9587dc674b11c3d3e5ed1'))
[docs] def test_compression_methods(self): assert (self.p.data.num_compression_methods == 1)
[docs] def test_total_length(self): assert (len(self.p) == 413)
[docs]class TestServerHello(object): """Again, from Wireshark"""
[docs] @classmethod def setup_class(cls): cls.data = _hexdecode( b'0200004d03015008220c8ec43c5462315a7c99f5d5b6bff009ad285b51dc18485f352e9fdecd2009bc0192e008e6fa8fe47998fca91311ba30ddde14a9587dc674b11c3d3e5ed10002000005ff01000100') cls.p = TLSHandshake(cls.data)
[docs] def test_constructed(self): assert (isinstance(self.p.data, TLSServerHello) == True)
# def testDateCorrect(self): # self.assertEqual(self.p.random_unixtime, 1342710284)
[docs] def test_random_correct(self): assert (self.p.data.random == _hexdecode(b'5008220c8ec43c5462315a7c99f5d5b6bff009ad285b51dc18485f352e9fdecd'))
[docs] def test_cipher_suite(self): assert (ssl_ciphersuites.BY_CODE[self.p.data.cipher_suite].name == 'TLS_RSA_WITH_NULL_SHA')
[docs] def test_total_length(self): assert (len(self.p) == 81)
[docs]class TestTLSCertificate(object): """We use a 2016 certificate record from iana.org as test data."""
[docs] @classmethod def setup_class(cls): cls.p = TLSHandshake(_hexdecode('0b000b45000b42000687308206833082056ba003020102021009cabbe2191c8f569dd4b6dd250f21d8300d06092a864886f70d01010b05003070310b300906035504061302555331153013060355040a130c446967694365727420496e6331193017060355040b13107777772e64696769636572742e636f6d312f302d06035504031326446967694365727420534841322048696768204173737572616e636520536572766572204341301e170d3134313032373030303030305a170d3138303130333132303030305a3081a3310b3009060355040613025553311330110603550408130a43616c69666f726e6961311430120603550407130b4c6f7320416e67656c6573313c303a060355040a1333496e7465726e657420436f72706f726174696f6e20666f722041737369676e6564204e616d657320616e64204e756d6265727331163014060355040b130d4954204f7065726174696f6e733113301106035504030c0a2a2e69616e612e6f726730820222300d06092a864886f70d01010105000382020f003082020a02820201009dbdfddeb5cae53a559747e2fda63728e4aba60f18b79a69f03310bf0164e5ee7db6b15bf56df23fddbae6a1bb38449b8c883f18102bbd8bb655ac0e2dac2ee3ed5cf4315868d2c598068284854b24894dcd4bd37811f0ad3a282cd4b4e599ffd07d8d2d3f2478554f81020b320ee12f44948e2ea1edbc990b830ca5cca6b4a839fb27b51850c9847eac74f26609eb24365b9751fb1c3208f56913bacbcae49201347c78b7e54a9d99979404c37f00fb65db849fd75e3a68770c30f2abe65b33256fb59b450050b00d8139d4d80d36f7bc46daf303e48f0f0791b2fdd72ec60b2cb3ad533c3f288c9c194e49337a69c496731f086d4f1f9825900713e2a551d05cb6057567850d91e6001c4ce27176f0957873a95b880acbec19e7bd9bcf1286d0452b73789c41905dd470971cd73aea52c77b080cd779af58234f337225c26f87a8c13e2a65e9dd4e03a5b41d7e06b3353f38129b2327a531ec9627a21dc423733aa029d4989448ba3322891c1a5690ddf2d25c8ec8aaa894b14aa92130c6b6d969a21ff671b60c4c923a94a93ea1dd0492c93393ca6edd61f33ca77e9208d01d6bd15107662ec088733df4c876a7e1608b82973a0f7592e84ed15579d181e79024ae8a7e4b9f0078eb2005b23f9d09a1df1bbc7de2a5a6085a3646d9fadb0e9da273a5f403cdd42831ce6f0ca46889585602bb8bc36bb3be861ff6d1a62e350203010001a38201e3308201df301f0603551d230418301680145168ff90af0207753cccd9656462a212b859723b301d0603551d0e04160414c7d0acef898b20e4b914668933032394f6bf3a61301f0603551d1104183016820a2a2e69616e612e6f7267820869616e612e6f7267300e0603551d0f0101ff0404030205a0301d0603551d250416301406082b0601050507030106082b0601050507030230750603551d1f046e306c3034a032a030862e687474703a2f2f63726c332e64696769636572742e636f6d2f736861322d68612d7365727665722d67332e63726c3034a032a030862e687474703a2f2f63726c342e64696769636572742e636f6d2f736861322d68612d7365727665722d67332e63726c30420603551d20043b3039303706096086480186fd6c0101302a302806082b06010505070201161c68747470733a2f2f7777772e64696769636572742e636f6d2f43505330818306082b0601050507010104773075302406082b060105050730018618687474703a2f2f6f6373702e64696769636572742e636f6d304d06082b060105050730028641687474703a2f2f636163657274732e64696769636572742e636f6d2f446967694365727453484132486967684173737572616e636553657276657243412e637274300c0603551d130101ff04023000300d06092a864886f70d01010b0500038201010070314c38e7c02fd80810500b9df6dae85de9b23e29fbd68bfdb5f23411c89acfaf9ae05af9123a8aa6bce6954a4e68dc7cfc480a65d76f229c4bd5f5674b0c9ac6d06a37a1a1c145c3956120b8efe67c887ab4ff7d6aa950ff3698f27c4a19d59d93a39aca5a7b6d6c75e34974e50f5a590005b3cb665ddbd7074f9fcbcbf9c50228d5e25596b64ada160b48f77a93aaced22617bfe005e00fe20a532a0adcb818c878dc5d6649277777ca1a814e21d0b53308af4078be4554715e4ce4828b012f25ffa13a6ceb30d20a75deba8a344e41d627fa638feff38a3063a0187519b39b053f7134d9cd83e6091accf5d2e3a05edfa1dfbe181a87ad86ba24fe6b97fe0004b5308204b130820399a003020102021004e1e7a4dc5cf2f36dc02b42b85d159f300d06092a864886f70d01010b0500306c310b300906035504061302555331153013060355040a130c446967694365727420496e6331193017060355040b13107777772e64696769636572742e636f6d312b30290603550403132244696769436572742048696768204173737572616e636520455620526f6f74204341301e170d3133313032323132303030305a170d3238313032323132303030305a3070310b300906035504061302555331153013060355040a130c446967694365727420496e6331193017060355040b13107777772e64696769636572742e636f6d312f302d06035504031326446967694365727420534841322048696768204173737572616e63652053657276657220434130820122300d06092a864886f70d01010105000382010f003082010a0282010100b6e02fc22406c86d045fd7ef0a6406b27d22266516ae42409bcedc9f9f76073ec330558719b94f940e5a941f5556b4c2022aafd098ee0b40d7c4d03b72c8149eef90b111a9aed2c8b8433ad90b0bd5d595f540afc81ded4d9c5f57b786506899f58adad2c7051fa897c9dca4b182842dc6ada59cc71982a6850f5e44582a378ffd35f10b0827325af5bb8b9ea4bd51d027e2dd3b4233a30528c4bb28cc9aac2b230d78c67be65e71b74a3e08fb81b71616a19d23124de5d79208ac75a49cbacd17b21e4435657f532539d11c0a9a631b199274680a37c2c25248cb395aa2b6e15dc1dda020b821a293266f144a2141c7ed6d9bf2482ff303f5a26892532f5ee30203010001a38201493082014530120603551d130101ff040830060101ff020100300e0603551d0f0101ff040403020186301d0603551d250416301406082b0601050507030106082b06010505070302303406082b0601050507010104283026302406082b060105050730018618687474703a2f2f6f6373702e64696769636572742e636f6d304b0603551d1f044430423040a03ea03c863a687474703a2f2f63726c342e64696769636572742e636f6d2f4469676943657274486967684173737572616e63654556526f6f7443412e63726c303d0603551d200436303430320604551d2000302a302806082b06010505070201161c68747470733a2f2f7777772e64696769636572742e636f6d2f435053301d0603551d0e041604145168ff90af0207753cccd9656462a212b859723b301f0603551d23041830168014b13ec36903f8bf4701d498261a0802ef63642bc3300d06092a864886f70d01010b05000382010100188a958903e66ddf5cfc1d68ea4a8f83d6512f8d6b44169eac63f5d26e6c84998baa8171845bed344eb0b7799229cc2d806af08e20e179a4fe034713eaf586ca59717df404966bd359583dfed331255c183884a3e69f82fd8c5b98314ecd789e1afd85cb49aaf2278b9972fc3eaad5410bdad536a1bf1c6e47497f5ed9487c03d9fd8b49a098264240ebd69211a4640a5754c4f51dd6025e6baceec4809a1272fa5693d7ffbf30850630bf0b7f4eff57059d24ed85c32bfba675a8ac2d16ef7d7927b2ebc29d0b07eaaa85d301a3202841594328d281e3aaf6ec7b3b77b640628005414501ef17063edec0339b67d3612e7287e469fc120057401e70f51ec9b4'))
[docs] def test_num_certs(self): assert (len(self.p.data.certificates) == 2)
[docs]class TestTLSMultiFactory(object): """Made up test data"""
[docs] @classmethod def setup_class(cls): cls.data = _hexdecode(b'1703010010' # header 1 b'AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA' # data 1 b'1703010010' # header 2 b'BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB' # data 2 b'1703010010' # header 3 b'CCCCCCCC') # data 3 (incomplete) cls.msgs, cls.bytes_parsed = tls_multi_factory(cls.data)
[docs] def test_num_messages(self): # only complete messages should be parsed, incomplete ones left # in buffer assert (len(self.msgs) == 2)
[docs] def test_bytes_parsed(self): assert (self.bytes_parsed == (5 + 16) * 2)
[docs] def test_first_msg_data(self): assert (self.msgs[0].data == _hexdecode(b'AA' * 16))
[docs] def test_second_msg_data(self): assert (self.msgs[1].data == _hexdecode(b'BB' * 16))
[docs] def test_incomplete(self): msgs, n = tls_multi_factory(_hexdecode(b'17')) assert (len(msgs) == 0) assert (n == 0) msgs, n = tls_multi_factory(_hexdecode(b'1703')) assert (len(msgs) == 0) assert (n == 0) msgs, n = tls_multi_factory(_hexdecode(b'170301')) assert (len(msgs) == 0) assert (n == 0) msgs, n = tls_multi_factory(_hexdecode(b'17030100')) assert (len(msgs) == 0) assert (n == 0) msgs, n = tls_multi_factory(_hexdecode(b'1703010000')) assert (len(msgs) == 1) assert (n == 5)