Inteligent parser for handshake messages.
| 17 | |
| 18 | |
| 19 | class HandshakeParser(object): |
| 20 | """Inteligent parser for handshake messages.""" |
| 21 | |
| 22 | def __init__(self, version=(3, 1), |
| 23 | cipher_suite=CipherSuite.TLS_DHE_RSA_WITH_AES_128_CBC_SHA, |
| 24 | certificate_type=CertificateType.x509): |
| 25 | """Initialize parser object.""" |
| 26 | self.version = version |
| 27 | self.cipher_suite = cipher_suite |
| 28 | self.certificate_type = certificate_type |
| 29 | |
| 30 | def parse(self, parser): |
| 31 | """Parse a handshake message.""" |
| 32 | hs_type = parser.get(1) |
| 33 | if hs_type == HandshakeType.server_hello: |
| 34 | msg = ServerHello().parse(parser) |
| 35 | self.version = msg.server_version |
| 36 | self.cipher_suite = msg.cipher_suite |
| 37 | self.certificate_type = msg.certificate_type |
| 38 | return msg |
| 39 | elif hs_type == HandshakeType.certificate: |
| 40 | msg = Certificate(self.certificate_type) |
| 41 | elif hs_type == HandshakeType.server_key_exchange: |
| 42 | msg = ServerKeyExchange(self.cipher_suite, self.version) |
| 43 | elif hs_type == HandshakeType.certificate_request: |
| 44 | msg = CertificateRequest(self.version) |
| 45 | elif hs_type == HandshakeType.next_protocol: |
| 46 | msg = NextProtocol().parse(parser) |
| 47 | elif hs_type == HandshakeType.server_hello_done: |
| 48 | msg = ServerHelloDone() |
| 49 | elif hs_type == HandshakeType.session_ticket: |
| 50 | msg = NewSessionTicket() |
| 51 | elif hs_type == HandshakeType.certificate_status: |
| 52 | msg = CertificateStatus() |
| 53 | else: |
| 54 | raise ValueError("Unknown handshake type: {0}".format(hs_type)) |
| 55 | |
| 56 | # don't abort when we can't parse a message, save it as unparsed |
| 57 | try: |
| 58 | msg.parse(parser) |
| 59 | except SyntaxError: |
| 60 | msg = Message(ContentType.handshake, parser.bytes) |
| 61 | return msg |
| 62 | |
| 63 | |
| 64 | class Scanner(object): |