| ## This file is part of Scapy |
| ## Copyright (C) 2007, 2008, 2009 Arnaud Ebalard |
| ## 2015, 2016, 2017 Maxence Tury |
| ## This program is published under a GPLv2 license |
| |
| """ |
| TLS handshake fields & logic. |
| |
| This module covers the handshake TLS subprotocol, except for the key exchange |
| mechanisms which are addressed with keyexchange.py. |
| """ |
| |
| from __future__ import absolute_import |
| import math |
| |
| from scapy.error import log_runtime, warning |
| from scapy.fields import * |
| from scapy.compat import * |
| from scapy.packet import Packet, Raw, Padding |
| from scapy.utils import repr_hex |
| from scapy.layers.x509 import OCSP_Response |
| from scapy.layers.tls.cert import Cert, PrivKey, PubKey |
| from scapy.layers.tls.basefields import (_tls_version, _TLSVersionField, |
| _TLSClientVersionField) |
| from scapy.layers.tls.extensions import (_ExtensionsLenField, _ExtensionsField, |
| _cert_status_type, TLS_Ext_SupportedVersions) |
| from scapy.layers.tls.keyexchange import (_TLSSignature, _TLSServerParamsField, |
| _TLSSignatureField, ServerRSAParams, |
| SigAndHashAlgsField, _tls_hash_sig, |
| SigAndHashAlgsLenField) |
| from scapy.layers.tls.keyexchange_tls13 import TicketField |
| from scapy.layers.tls.session import (_GenericTLSSessionInheritance, |
| readConnState, writeConnState) |
| from scapy.layers.tls.crypto.compression import (_tls_compression_algs, |
| _tls_compression_algs_cls, |
| Comp_NULL, _GenericComp, |
| _GenericCompMetaclass) |
| from scapy.layers.tls.crypto.suites import (_tls_cipher_suites, |
| _tls_cipher_suites_cls, |
| _GenericCipherSuite, |
| _GenericCipherSuiteMetaclass) |
| |
| |
| ############################################################################### |
| ### Generic TLS Handshake message ### |
| ############################################################################### |
| |
| _tls_handshake_type = { 0: "hello_request", 1: "client_hello", |
| 2: "server_hello", 3: "hello_verify_request", |
| 4: "session_ticket", 6: "hello_retry_request", |
| 8: "encrypted_extensions", 11: "certificate", |
| 12: "server_key_exchange", 13: "certificate_request", |
| 14: "server_hello_done", 15: "certificate_verify", |
| 16: "client_key_exchange", 20: "finished", |
| 21: "certificate_url", 22: "certificate_status", |
| 23: "supplemental_data", 24: "key_update" } |
| |
| |
| class _TLSHandshake(_GenericTLSSessionInheritance): |
| """ |
| Inherited by other Handshake classes to get post_build(). |
| Also used as a fallback for unknown TLS Handshake packets. |
| """ |
| name = "TLS Handshake Generic message" |
| fields_desc = [ ByteEnumField("msgtype", None, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| StrLenField("msg", "", |
| length_from=lambda pkt: pkt.msglen) ] |
| |
| def post_build(self, p, pay): |
| l = len(p) |
| if self.msglen is None: |
| l2 = l - 4 |
| p = struct.pack("!I", (orb(p[0]) << 24) | l2) + p[4:] |
| return p + pay |
| |
| def guess_payload_class(self, p): |
| return conf.padding_layer |
| |
| def tls_session_update(self, msg_str): |
| """ |
| Covers both post_build- and post_dissection- context updates. |
| """ |
| self.tls_session.handshake_messages.append(msg_str) |
| self.tls_session.handshake_messages_parsed.append(self) |
| |
| |
| ############################################################################### |
| ### HelloRequest ### |
| ############################################################################### |
| |
| class TLSHelloRequest(_TLSHandshake): |
| name = "TLS Handshake - Hello Request" |
| fields_desc = [ ByteEnumField("msgtype", 0, _tls_handshake_type), |
| ThreeBytesField("msglen", None) ] |
| |
| def tls_session_update(self, msg_str): |
| """ |
| Message should not be added to the list of handshake messages |
| that will be hashed in the finished and certificate verify messages. |
| """ |
| return |
| |
| |
| ############################################################################### |
| ### ClientHello fields ### |
| ############################################################################### |
| |
| class _GMTUnixTimeField(UTCTimeField): |
| """ |
| "The current time and date in standard UNIX 32-bit format (seconds since |
| the midnight starting Jan 1, 1970, GMT, ignoring leap seconds)." |
| """ |
| def i2h(self, pkt, x): |
| if x is not None: |
| return x |
| return 0 |
| |
| class _TLSRandomBytesField(StrFixedLenField): |
| def i2repr(self, pkt, x): |
| if x is None: |
| return repr(x) |
| return repr_hex(self.i2h(pkt,x)) |
| |
| |
| class _SessionIDField(StrLenField): |
| """ |
| opaque SessionID<0..32>; section 7.4.1.2 of RFC 4346 |
| """ |
| pass |
| |
| |
| class _CipherSuitesField(StrLenField): |
| __slots__ = ["itemfmt", "itemsize", "i2s", "s2i"] |
| islist = 1 |
| def __init__(self, name, default, dico, length_from=None, itemfmt="!H"): |
| StrLenField.__init__(self, name, default, length_from=length_from) |
| self.itemfmt = itemfmt |
| self.itemsize = struct.calcsize(itemfmt) |
| i2s = self.i2s = {} |
| s2i = self.s2i = {} |
| for k in six.iterkeys(dico): |
| i2s[k] = dico[k] |
| s2i[dico[k]] = k |
| |
| def any2i_one(self, pkt, x): |
| if (isinstance(x, _GenericCipherSuite) or |
| isinstance(x, _GenericCipherSuiteMetaclass)): |
| x = x.val |
| if isinstance(x, bytes): |
| x = self.s2i[x] |
| return x |
| |
| def i2repr_one(self, pkt, x): |
| fmt = "0x%%0%dx" % self.itemsize |
| return self.i2s.get(x, fmt % x) |
| |
| def any2i(self, pkt, x): |
| if x is None: |
| return None |
| if not isinstance(x, list): |
| x = [x] |
| return [self.any2i_one(pkt, z) for z in x] |
| |
| def i2repr(self, pkt, x): |
| if x is None: |
| return "None" |
| l = [self.i2repr_one(pkt, z) for z in x] |
| if len(l) == 1: |
| l = l[0] |
| else: |
| l = "[%s]" % ", ".join(l) |
| return l |
| |
| def i2m(self, pkt, val): |
| if val is None: |
| val = [] |
| return b"".join(struct.pack(self.itemfmt, x) for x in val) |
| |
| def m2i(self, pkt, m): |
| res = [] |
| itemlen = struct.calcsize(self.itemfmt) |
| while m: |
| res.append(struct.unpack(self.itemfmt, m[:itemlen])[0]) |
| m = m[itemlen:] |
| return res |
| |
| def i2len(self, pkt, i): |
| if i is None: |
| return 0 |
| return len(i)*self.itemsize |
| |
| |
| class _CompressionMethodsField(_CipherSuitesField): |
| |
| def any2i_one(self, pkt, x): |
| if (isinstance(x, _GenericComp) or |
| isinstance(x, _GenericCompMetaclass)): |
| x = x.val |
| if isinstance(x, str): |
| x = self.s2i[x] |
| return x |
| |
| |
| ############################################################################### |
| ### ClientHello ### |
| ############################################################################### |
| |
| class TLSClientHello(_TLSHandshake): |
| """ |
| TLS ClientHello, with abilities to handle extensions. |
| |
| The Random structure follows the RFC 5246: while it is 32-byte long, |
| many implementations use the first 4 bytes as a gmt_unix_time, and then |
| the remaining 28 byts should be completely random. This was designed in |
| order to (sort of) mitigate broken RNGs. If you prefer to show the full |
| 32 random bytes without any GMT time, just comment in/out the lines below. |
| """ |
| name = "TLS Handshake - Client Hello" |
| fields_desc = [ ByteEnumField("msgtype", 1, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _TLSClientVersionField("version", None, _tls_version), |
| |
| #_TLSRandomBytesField("random_bytes", None, 32), |
| _GMTUnixTimeField("gmt_unix_time", None), |
| _TLSRandomBytesField("random_bytes", None, 28), |
| |
| FieldLenField("sidlen", None, fmt="B", length_of="sid"), |
| _SessionIDField("sid", "", |
| length_from=lambda pkt:pkt.sidlen), |
| |
| FieldLenField("cipherslen", None, fmt="!H", |
| length_of="ciphers"), |
| _CipherSuitesField("ciphers", None, |
| _tls_cipher_suites, itemfmt="!H", |
| length_from=lambda pkt: pkt.cipherslen), |
| |
| FieldLenField("complen", None, fmt="B", length_of="comp"), |
| _CompressionMethodsField("comp", [0], |
| _tls_compression_algs, |
| itemfmt="B", |
| length_from= |
| lambda pkt: pkt.complen), |
| |
| _ExtensionsLenField("extlen", None, length_of="ext"), |
| _ExtensionsField("ext", None, |
| length_from=lambda pkt: (pkt.msglen - |
| (pkt.sidlen or 0) - |
| (pkt.cipherslen or 0) - |
| (pkt.complen or 0) - |
| 40)) ] |
| |
| def post_build(self, p, pay): |
| if self.random_bytes is None: |
| p = p[:10] + randstring(28) + p[10+28:] |
| |
| # if no ciphersuites were provided, we add a few usual, supported |
| # ciphersuites along with the appropriate extensions |
| if self.ciphers is None: |
| cipherstart = 39 + (self.sidlen or 0) |
| s = b"001ac02bc023c02fc027009e0067009c003cc009c0130033002f000a" |
| p = p[:cipherstart] + bytes_hex(s) + p[cipherstart+2:] |
| if self.ext is None: |
| ext_len = b'\x00\x2c' |
| ext_reneg = b'\xff\x01\x00\x01\x00' |
| ext_sn = b'\x00\x00\x00\x0f\x00\r\x00\x00\nsecdev.org' |
| ext_sigalg = b'\x00\r\x00\x08\x00\x06\x04\x03\x04\x01\x02\x01' |
| ext_supgroups = b'\x00\n\x00\x04\x00\x02\x00\x17' |
| p += ext_len + ext_reneg + ext_sn + ext_sigalg + ext_supgroups |
| |
| return super(TLSClientHello, self).post_build(p, pay) |
| |
| def tls_session_update(self, msg_str): |
| """ |
| Either for parsing or building, we store the client_random |
| along with the raw string representing this handshake message. |
| """ |
| super(TLSClientHello, self).tls_session_update(msg_str) |
| |
| self.tls_session.advertised_tls_version = self.version |
| self.random_bytes = msg_str[10:38] |
| self.tls_session.client_random = (struct.pack('!I', |
| self.gmt_unix_time) + |
| self.random_bytes) |
| if self.ext: |
| for e in self.ext: |
| if isinstance(e, TLS_Ext_SupportedVersions): |
| if self.tls_session.tls13_early_secret is None: |
| # this is not recomputed if there was a TLS 1.3 HRR |
| self.tls_session.compute_tls13_early_secrets() |
| break |
| |
| ############################################################################### |
| ### ServerHello ### |
| ############################################################################### |
| |
| class TLSServerHello(TLSClientHello): |
| """ |
| TLS ServerHello, with abilities to handle extensions. |
| |
| The Random structure follows the RFC 5246: while it is 32-byte long, |
| many implementations use the first 4 bytes as a gmt_unix_time, and then |
| the remaining 28 byts should be completely random. This was designed in |
| order to (sort of) mitigate broken RNGs. If you prefer to show the full |
| 32 random bytes without any GMT time, just comment in/out the lines below. |
| """ |
| name = "TLS Handshake - Server Hello" |
| fields_desc = [ ByteEnumField("msgtype", 2, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _TLSVersionField("version", None, _tls_version), |
| |
| #_TLSRandomBytesField("random_bytes", None, 32), |
| _GMTUnixTimeField("gmt_unix_time", None), |
| _TLSRandomBytesField("random_bytes", None, 28), |
| |
| FieldLenField("sidlen", None, length_of="sid", fmt="B"), |
| _SessionIDField("sid", "", |
| length_from = lambda pkt: pkt.sidlen), |
| |
| EnumField("cipher", None, _tls_cipher_suites), |
| _CompressionMethodsField("comp", [0], |
| _tls_compression_algs, |
| itemfmt="B", |
| length_from=lambda pkt: 1), |
| |
| _ExtensionsLenField("extlen", None, length_of="ext"), |
| _ExtensionsField("ext", None, |
| length_from=lambda pkt: (pkt.msglen - |
| (pkt.sidlen or 0) - |
| 38)) ] |
| #40)) ] |
| |
| @classmethod |
| def dispatch_hook(cls, _pkt=None, *args, **kargs): |
| if _pkt and len(_pkt) >= 6: |
| version = struct.unpack("!H", _pkt[4:6])[0] |
| if version == 0x0304 or version > 0x7f00: |
| return TLS13ServerHello |
| return TLSServerHello |
| |
| def post_build(self, p, pay): |
| if self.random_bytes is None: |
| p = p[:10] + randstring(28) + p[10+28:] |
| return super(TLSClientHello, self).post_build(p, pay) |
| |
| def tls_session_update(self, msg_str): |
| """ |
| Either for parsing or building, we store the server_random |
| along with the raw string representing this handshake message. |
| We also store the session_id, the cipher suite (if recognized), |
| the compression method, and finally we instantiate the pending write |
| and read connection states. Usually they get updated later on in the |
| negotiation when we learn the session keys, and eventually they |
| are committed once a ChangeCipherSpec has been sent/received. |
| """ |
| super(TLSClientHello, self).tls_session_update(msg_str) |
| |
| self.tls_session.tls_version = self.version |
| self.random_bytes = msg_str[10:38] |
| self.tls_session.server_random = (struct.pack('!I', |
| self.gmt_unix_time) + |
| self.random_bytes) |
| self.tls_session.sid = self.sid |
| |
| cs_cls = None |
| if self.cipher: |
| cs_val = self.cipher |
| if cs_val not in _tls_cipher_suites_cls: |
| warning("Unknown cipher suite %d from ServerHello" % cs_val) |
| # we do not try to set a default nor stop the execution |
| else: |
| cs_cls = _tls_cipher_suites_cls[cs_val] |
| |
| comp_cls = Comp_NULL |
| if self.comp: |
| comp_val = self.comp[0] |
| if comp_val not in _tls_compression_algs_cls: |
| err = "Unknown compression alg %d from ServerHello" % comp_val |
| warning(err) |
| comp_val = 0 |
| comp_cls = _tls_compression_algs_cls[comp_val] |
| |
| connection_end = self.tls_session.connection_end |
| self.tls_session.pwcs = writeConnState(ciphersuite=cs_cls, |
| compression_alg=comp_cls, |
| connection_end=connection_end, |
| tls_version=self.version) |
| self.tls_session.prcs = readConnState(ciphersuite=cs_cls, |
| compression_alg=comp_cls, |
| connection_end=connection_end, |
| tls_version=self.version) |
| |
| |
| class TLS13ServerHello(TLSClientHello): |
| """ TLS 1.3 ServerHello """ |
| name = "TLS 1.3 Handshake - Server Hello" |
| fields_desc = [ ByteEnumField("msgtype", 2, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _TLSVersionField("version", None, _tls_version), |
| _TLSRandomBytesField("random_bytes", None, 32), |
| EnumField("cipher", None, _tls_cipher_suites), |
| _ExtensionsLenField("extlen", None, length_of="ext"), |
| _ExtensionsField("ext", None, |
| length_from=lambda pkt: (pkt.msglen - |
| 38)) ] |
| |
| def tls_session_update(self, msg_str): |
| """ |
| Either for parsing or building, we store the server_random along with |
| the raw string representing this handshake message. We also store the |
| cipher suite (if recognized), and finally we instantiate the write and |
| read connection states. |
| """ |
| super(TLSClientHello, self).tls_session_update(msg_str) |
| |
| s = self.tls_session |
| s.tls_version = self.version |
| s.server_random = self.random_bytes |
| |
| cs_cls = None |
| if self.cipher: |
| cs_val = self.cipher |
| if cs_val not in _tls_cipher_suites_cls: |
| warning("Unknown cipher suite %d from ServerHello" % cs_val) |
| # we do not try to set a default nor stop the execution |
| else: |
| cs_cls = _tls_cipher_suites_cls[cs_val] |
| |
| connection_end = s.connection_end |
| s.pwcs = writeConnState(ciphersuite=cs_cls, |
| connection_end=connection_end, |
| tls_version=self.version) |
| s.triggered_pwcs_commit = True |
| s.prcs = readConnState(ciphersuite=cs_cls, |
| connection_end=connection_end, |
| tls_version=self.version) |
| s.triggered_prcs_commit = True |
| |
| if self.tls_session.tls13_early_secret is None: |
| # In case the connState was not pre-initialized, we could not |
| # compute the early secrets at the ClientHello, so we do it here. |
| self.tls_session.compute_tls13_early_secrets() |
| s.compute_tls13_handshake_secrets() |
| |
| |
| ############################################################################### |
| ### HelloRetryRequest ### |
| ############################################################################### |
| |
| class TLSHelloRetryRequest(_TLSHandshake): |
| name = "TLS 1.3 Handshake - Hello Retry Request" |
| fields_desc = [ ByteEnumField("msgtype", 6, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _TLSVersionField("version", None, _tls_version), |
| _ExtensionsLenField("extlen", None, length_of="ext"), |
| _ExtensionsField("ext", None, |
| length_from=lambda pkt: pkt.msglen - 4) ] |
| |
| |
| ############################################################################### |
| ### EncryptedExtensions ### |
| ############################################################################### |
| |
| class TLSEncryptedExtensions(_TLSHandshake): |
| name = "TLS 1.3 Handshake - Encrypted Extensions" |
| fields_desc = [ ByteEnumField("msgtype", 8, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _ExtensionsLenField("extlen", None, length_of="ext"), |
| _ExtensionsField("ext", None, |
| length_from=lambda pkt: pkt.msglen - 2) ] |
| |
| |
| ############################################################################### |
| ### Certificate ### |
| ############################################################################### |
| |
| #XXX It might be appropriate to rewrite this mess with basic 3-byte FieldLenField. |
| |
| class _ASN1CertLenField(FieldLenField): |
| """ |
| This is mostly a 3-byte FieldLenField. |
| """ |
| def __init__(self, name, default, length_of=None, adjust=lambda pkt, x: x): |
| self.length_of = length_of |
| self.adjust = adjust |
| Field.__init__(self, name, default, fmt="!I") |
| |
| def i2m(self, pkt, x): |
| if x is None: |
| if self.length_of is not None: |
| fld,fval = pkt.getfield_and_val(self.length_of) |
| f = fld.i2len(pkt, fval) |
| x = self.adjust(pkt, f) |
| return x |
| |
| def addfield(self, pkt, s, val): |
| return s + struct.pack(self.fmt, self.i2m(pkt,val))[1:4] |
| |
| def getfield(self, pkt, s): |
| return s[3:], self.m2i(pkt, struct.unpack(self.fmt, b"\x00" + s[:3])[0]) |
| |
| |
| class _ASN1CertListField(StrLenField): |
| islist = 1 |
| def i2len(self, pkt, i): |
| if i is None: |
| return 0 |
| return len(self.i2m(pkt, i)) |
| |
| def getfield(self, pkt, s): |
| """ |
| Extract Certs in a loop. |
| XXX We should provide safeguards when trying to parse a Cert. |
| """ |
| l = None |
| if self.length_from is not None: |
| l = self.length_from(pkt) |
| |
| lst = [] |
| ret = b"" |
| m = s |
| if l is not None: |
| m, ret = s[:l], s[l:] |
| while m: |
| clen = struct.unpack("!I", b'\x00' + m[:3])[0] |
| lst.append((clen, Cert(m[3:3 + clen]))) |
| m = m[3 + clen:] |
| return m + ret, lst |
| |
| def i2m(self, pkt, i): |
| def i2m_one(i): |
| if isinstance(i, str): |
| return i |
| if isinstance(i, Cert): |
| s = i.der |
| l = struct.pack("!I", len(s))[1:4] |
| return l + s |
| |
| (l, s) = i |
| if isinstance(s, Cert): |
| s = s.der |
| return struct.pack("!I", l)[1:4] + s |
| |
| if i is None: |
| return b"" |
| if isinstance(i, str): |
| return i |
| if isinstance(i, Cert): |
| i = [i] |
| return b"".join(i2m_one(x) for x in i) |
| |
| def any2i(self, pkt, x): |
| return x |
| |
| class _ASN1CertField(StrLenField): |
| def i2len(self, pkt, i): |
| if i is None: |
| return 0 |
| return len(self.i2m(pkt, i)) |
| |
| def getfield(self, pkt, s): |
| l = None |
| if self.length_from is not None: |
| l = self.length_from(pkt) |
| ret = b"" |
| m = s |
| if l is not None: |
| m, ret = s[:l], s[l:] |
| clen = struct.unpack("!I", b'\x00' + m[:3])[0] |
| len_cert = (clen, Cert(m[3:3 + clen])) |
| m = m[3 + clen:] |
| return m + ret, len_cert |
| |
| def i2m(self, pkt, i): |
| def i2m_one(i): |
| if isinstance(i, str): |
| return i |
| if isinstance(i, Cert): |
| s = i.der |
| l = struct.pack("!I", len(s))[1:4] |
| return l + s |
| |
| (l, s) = i |
| if isinstance(s, Cert): |
| s = s.der |
| return struct.pack("!I", l)[1:4] + s |
| |
| if i is None: |
| return b"" |
| return i2m_one(i) |
| |
| def any2i(self, pkt, x): |
| return x |
| |
| |
| class TLSCertificate(_TLSHandshake): |
| """ |
| XXX We do not support RFC 5081, i.e. OpenPGP certificates. |
| """ |
| name = "TLS Handshake - Certificate" |
| fields_desc = [ ByteEnumField("msgtype", 11, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _ASN1CertLenField("certslen", None, length_of="certs"), |
| _ASN1CertListField("certs", [], |
| length_from = lambda pkt: pkt.certslen) ] |
| |
| @classmethod |
| def dispatch_hook(cls, _pkt=None, *args, **kargs): |
| if _pkt: |
| tls_session = kargs.get("tls_session", None) |
| if tls_session and (tls_session.tls_version or 0) >= 0x0304: |
| return TLS13Certificate |
| return TLSCertificate |
| |
| def post_dissection_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| connection_end = self.tls_session.connection_end |
| if connection_end == "client": |
| self.tls_session.server_certs = [x[1] for x in self.certs] |
| else: |
| self.tls_session.client_certs = [x[1] for x in self.certs] |
| |
| |
| class _ASN1CertAndExt(_GenericTLSSessionInheritance): |
| name = "Certificate and Extensions" |
| fields_desc = [ _ASN1CertField("cert", ""), |
| FieldLenField("extlen", None, length_of="ext"), |
| _ExtensionsField("ext", [], |
| length_from=lambda pkt: pkt.extlen) ] |
| def extract_padding(self, s): |
| return b"", s |
| |
| class _ASN1CertAndExtListField(PacketListField): |
| def m2i(self, pkt, m): |
| return self.cls(m, tls_session=pkt.tls_session) |
| |
| class TLS13Certificate(_TLSHandshake): |
| name = "TLS 1.3 Handshake - Certificate" |
| fields_desc = [ ByteEnumField("msgtype", 11, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| FieldLenField("cert_req_ctxt_len", None, fmt="B", |
| length_of="cert_req_ctxt"), |
| StrLenField("cert_req_ctxt", "", |
| length_from=lambda pkt: pkt.cert_req_ctxt_len), |
| _ASN1CertLenField("certslen", None, length_of="certs"), |
| _ASN1CertAndExtListField("certs", [], _ASN1CertAndExt, |
| length_from=lambda pkt: pkt.certslen) ] |
| |
| def post_dissection_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| connection_end = self.tls_session.connection_end |
| if connection_end == "client": |
| if self.certs: |
| sc = [x.cert[1] for x in self.certs] |
| self.tls_session.server_certs = sc |
| else: |
| if self.certs: |
| cc = [x.cert[1] for x in self.certs] |
| self.tls_session.client_certs = cc |
| |
| |
| ############################################################################### |
| ### ServerKeyExchange ### |
| ############################################################################### |
| |
| class TLSServerKeyExchange(_TLSHandshake): |
| name = "TLS Handshake - Server Key Exchange" |
| fields_desc = [ ByteEnumField("msgtype", 12, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _TLSServerParamsField("params", None, |
| length_from=lambda pkt: pkt.msglen), |
| _TLSSignatureField("sig", None, |
| length_from=lambda pkt: pkt.msglen - len(pkt.params)) ] |
| |
| def build(self, *args, **kargs): |
| """ |
| We overload build() method in order to provide a valid default value |
| for params based on TLS session if not provided. This cannot be done by |
| overriding i2m() because the method is called on a copy of the packet. |
| |
| The 'params' field is built according to key_exchange.server_kx_msg_cls |
| which should have been set after receiving a cipher suite in a |
| previous ServerHello. Usual cases are: |
| - None: for RSA encryption or fixed FF/ECDH. This should never happen, |
| as no ServerKeyExchange should be generated in the first place. |
| - ServerDHParams: for ephemeral FFDH. In that case, the parameter to |
| server_kx_msg_cls does not matter. |
| - ServerECDH*Params: for ephemeral ECDH. There are actually three |
| classes, which are dispatched by _tls_server_ecdh_cls_guess on |
| the first byte retrieved. The default here is b"\03", which |
| corresponds to ServerECDHNamedCurveParams (implicit curves). |
| |
| When the Server*DHParams are built via .fill_missing(), the session |
| server_kx_privkey will be updated accordingly. |
| """ |
| fval = self.getfieldval("params") |
| if fval is None: |
| s = self.tls_session |
| if s.pwcs: |
| if s.pwcs.key_exchange.export: |
| cls = ServerRSAParams(tls_session=s) |
| else: |
| cls = s.pwcs.key_exchange.server_kx_msg_cls(b"\x03") |
| cls = cls(tls_session=s) |
| try: |
| cls.fill_missing() |
| except: |
| pass |
| else: |
| cls = Raw() |
| self.params = cls |
| |
| fval = self.getfieldval("sig") |
| if fval is None: |
| s = self.tls_session |
| if s.pwcs: |
| if not s.pwcs.key_exchange.anonymous: |
| p = self.params |
| if p is None: |
| p = b"" |
| m = s.client_random + s.server_random + raw(p) |
| cls = _TLSSignature(tls_session=s) |
| cls._update_sig(m, s.server_key) |
| else: |
| cls = Raw() |
| else: |
| cls = Raw() |
| self.sig = cls |
| |
| return _TLSHandshake.build(self, *args, **kargs) |
| |
| def post_dissection(self, pkt): |
| """ |
| While previously dissecting Server*DHParams, the session |
| server_kx_pubkey should have been updated. |
| |
| XXX Add a 'fixed_dh' OR condition to the 'anonymous' test. |
| """ |
| s = self.tls_session |
| if s.prcs and s.prcs.key_exchange.no_ske: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: useless ServerKeyExchange [%s]", pkt_info) |
| if (s.prcs and |
| not s.prcs.key_exchange.anonymous and |
| s.client_random and s.server_random and |
| s.server_certs and len(s.server_certs) > 0): |
| m = s.client_random + s.server_random + raw(self.params) |
| sig_test = self.sig._verify_sig(m, s.server_certs[0]) |
| if not sig_test: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: invalid ServerKeyExchange signature [%s]", pkt_info) |
| |
| |
| ############################################################################### |
| ### CertificateRequest ### |
| ############################################################################### |
| |
| _tls_client_certificate_types = { 1: "rsa_sign", |
| 2: "dss_sign", |
| 3: "rsa_fixed_dh", |
| 4: "dss_fixed_dh", |
| 5: "rsa_ephemeral_dh_RESERVED", |
| 6: "dss_ephemeral_dh_RESERVED", |
| 20: "fortezza_dms_RESERVED", |
| 64: "ecdsa_sign", |
| 65: "rsa_fixed_ecdh", |
| 66: "ecdsa_fixed_ecdh" } |
| |
| |
| class _CertTypesField(_CipherSuitesField): |
| pass |
| |
| class _CertAuthoritiesField(StrLenField): |
| """ |
| XXX Rework this with proper ASN.1 parsing. |
| """ |
| islist = 1 |
| |
| def getfield(self, pkt, s): |
| l = self.length_from(pkt) |
| return s[l:], self.m2i(pkt, s[:l]) |
| |
| def m2i(self, pkt, m): |
| res = [] |
| while len(m) > 1: |
| l = struct.unpack("!H", m[:2])[0] |
| if len(m) < l + 2: |
| res.append((l, m[2:])) |
| break |
| dn = m[2:2+l] |
| res.append((l, dn)) |
| m = m[2+l:] |
| return res |
| |
| def i2m(self, pkt, i): |
| return b"".join(map(lambda x_y: struct.pack("!H", x_y[0]) + x_y[1], i)) |
| |
| def addfield(self, pkt, s, val): |
| return s + self.i2m(pkt, val) |
| |
| def i2len(self, pkt, val): |
| if val is None: |
| return 0 |
| else: |
| return len(self.i2m(pkt, val)) |
| |
| |
| class TLSCertificateRequest(_TLSHandshake): |
| name = "TLS Handshake - Certificate Request" |
| fields_desc = [ ByteEnumField("msgtype", 13, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| FieldLenField("ctypeslen", None, fmt="B", |
| length_of="ctypes"), |
| _CertTypesField("ctypes", [1, 64], |
| _tls_client_certificate_types, |
| itemfmt="!B", |
| length_from=lambda pkt: pkt.ctypeslen), |
| SigAndHashAlgsLenField("sig_algs_len", None, |
| length_of="sig_algs"), |
| SigAndHashAlgsField("sig_algs", [0x0403, 0x0401, 0x0201], |
| EnumField("hash_sig", None, _tls_hash_sig), |
| length_from=lambda pkt: pkt.sig_algs_len), |
| FieldLenField("certauthlen", None, fmt="!H", |
| length_of="certauth"), |
| _CertAuthoritiesField("certauth", [], |
| length_from=lambda pkt: pkt.certauthlen) ] |
| |
| |
| ############################################################################### |
| ### ServerHelloDone ### |
| ############################################################################### |
| |
| class TLSServerHelloDone(_TLSHandshake): |
| name = "TLS Handshake - Server Hello Done" |
| fields_desc = [ ByteEnumField("msgtype", 14, _tls_handshake_type), |
| ThreeBytesField("msglen", None) ] |
| |
| |
| ############################################################################### |
| ### CertificateVerify ### |
| ############################################################################### |
| |
| class TLSCertificateVerify(_TLSHandshake): |
| name = "TLS Handshake - Certificate Verify" |
| fields_desc = [ ByteEnumField("msgtype", 15, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _TLSSignatureField("sig", None, |
| length_from=lambda pkt: pkt.msglen) ] |
| |
| def build(self, *args, **kargs): |
| sig = self.getfieldval("sig") |
| if sig is None: |
| s = self.tls_session |
| m = b"".join(s.handshake_messages) |
| if s.tls_version >= 0x0304: |
| if s.connection_end == "client": |
| context_string = "TLS 1.3, client CertificateVerify" |
| elif s.connection_end == "server": |
| context_string = "TLS 1.3, server CertificateVerify" |
| m = b"\x20"*64 + context_string + b"\x00" + s.wcs.hash.digest(m) |
| self.sig = _TLSSignature(tls_session=s) |
| if s.connection_end == "client": |
| self.sig._update_sig(m, s.client_key) |
| elif s.connection_end == "server": |
| # should be TLS 1.3 only |
| self.sig._update_sig(m, s.server_key) |
| return _TLSHandshake.build(self, *args, **kargs) |
| |
| def post_dissection(self, pkt): |
| s = self.tls_session |
| m = b"".join(s.handshake_messages) |
| if s.tls_version >= 0x0304: |
| if s.connection_end == "client": |
| context_string = b"TLS 1.3, server CertificateVerify" |
| elif s.connection_end == "server": |
| context_string = b"TLS 1.3, client CertificateVerify" |
| m = b"\x20"*64 + context_string + b"\x00" + s.rcs.hash.digest(m) |
| |
| if s.connection_end == "server": |
| if s.client_certs and len(s.client_certs) > 0: |
| sig_test = self.sig._verify_sig(m, s.client_certs[0]) |
| if not sig_test: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: invalid CertificateVerify signature [%s]", pkt_info) |
| elif s.connection_end == "client": |
| # should be TLS 1.3 only |
| if s.server_certs and len(s.server_certs) > 0: |
| sig_test = self.sig._verify_sig(m, s.server_certs[0]) |
| if not sig_test: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: invalid CertificateVerify signature [%s]", pkt_info) |
| |
| |
| ############################################################################### |
| ### ClientKeyExchange ### |
| ############################################################################### |
| |
| class _TLSCKExchKeysField(PacketField): |
| __slots__ = ["length_from"] |
| holds_packet = 1 |
| def __init__(self, name, length_from=None, remain=0): |
| self.length_from = length_from |
| PacketField.__init__(self, name, None, None, remain=remain) |
| |
| def m2i(self, pkt, m): |
| """ |
| The client_kx_msg may be either None, EncryptedPreMasterSecret |
| (for RSA encryption key exchange), ClientDiffieHellmanPublic, |
| or ClientECDiffieHellmanPublic. When either one of them gets |
| dissected, the session context is updated accordingly. |
| """ |
| l = self.length_from(pkt) |
| tbd, rem = m[:l], m[l:] |
| |
| s = pkt.tls_session |
| cls = None |
| |
| if s.prcs and s.prcs.key_exchange: |
| cls = s.prcs.key_exchange.client_kx_msg_cls |
| |
| if cls is None: |
| return Raw(tbd)/Padding(rem) |
| |
| return cls(tbd, tls_session=s)/Padding(rem) |
| |
| |
| class TLSClientKeyExchange(_TLSHandshake): |
| """ |
| This class mostly works like TLSServerKeyExchange and its 'params' field. |
| """ |
| name = "TLS Handshake - Client Key Exchange" |
| fields_desc = [ ByteEnumField("msgtype", 16, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _TLSCKExchKeysField("exchkeys", |
| length_from = lambda pkt: pkt.msglen) ] |
| |
| def build(self, *args, **kargs): |
| fval = self.getfieldval("exchkeys") |
| if fval is None: |
| s = self.tls_session |
| if s.prcs: |
| cls = s.prcs.key_exchange.client_kx_msg_cls |
| cls = cls(tls_session=s) |
| else: |
| cls = Raw() |
| self.exchkeys = cls |
| return _TLSHandshake.build(self, *args, **kargs) |
| |
| |
| ############################################################################### |
| ### Finished ### |
| ############################################################################### |
| |
| class _VerifyDataField(StrLenField): |
| def getfield(self, pkt, s): |
| if pkt.tls_session.tls_version == 0x0300: |
| sep = 36 |
| elif pkt.tls_session.tls_version >= 0x0304: |
| sep = pkt.tls_session.rcs.hash.hash_len |
| else: |
| sep = 12 |
| return s[sep:], s[:sep] |
| |
| class TLSFinished(_TLSHandshake): |
| name = "TLS Handshake - Finished" |
| fields_desc = [ ByteEnumField("msgtype", 20, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| _VerifyDataField("vdata", None) ] |
| |
| def build(self, *args, **kargs): |
| fval = self.getfieldval("vdata") |
| if fval is None: |
| s = self.tls_session |
| handshake_msg = b"".join(s.handshake_messages) |
| con_end = s.connection_end |
| if s.tls_version < 0x0304: |
| ms = s.master_secret |
| self.vdata = s.wcs.prf.compute_verify_data(con_end, "write", |
| handshake_msg, ms) |
| else: |
| self.vdata = s.compute_tls13_verify_data(con_end, "write") |
| return _TLSHandshake.build(self, *args, **kargs) |
| |
| def post_dissection(self, pkt): |
| s = self.tls_session |
| if not s.frozen: |
| handshake_msg = b"".join(s.handshake_messages) |
| if s.tls_version < 0x0304 and s.master_secret is not None: |
| ms = s.master_secret |
| con_end = s.connection_end |
| verify_data = s.rcs.prf.compute_verify_data(con_end, "read", |
| handshake_msg, ms) |
| if self.vdata != verify_data: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: invalid Finished received [%s]", pkt_info) |
| elif s.tls_version >= 0x0304: |
| con_end = s.connection_end |
| verify_data = s.compute_tls13_verify_data(con_end, "read") |
| if self.vdata != verify_data: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: invalid Finished received [%s]", pkt_info) |
| |
| def post_build_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| s = self.tls_session |
| if s.tls_version >= 0x0304: |
| s.pwcs = writeConnState(ciphersuite=type(s.wcs.ciphersuite), |
| connection_end=s.connection_end, |
| tls_version=s.tls_version) |
| s.triggered_pwcs_commit = True |
| if s.connection_end == "server": |
| s.compute_tls13_traffic_secrets() |
| elif s.connection_end == "client": |
| s.compute_tls13_traffic_secrets_end() |
| s.compute_tls13_resumption_secret() |
| |
| def post_dissection_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| s = self.tls_session |
| if s.tls_version >= 0x0304: |
| s.prcs = readConnState(ciphersuite=type(s.rcs.ciphersuite), |
| connection_end=s.connection_end, |
| tls_version=s.tls_version) |
| s.triggered_prcs_commit = True |
| if s.connection_end == "client": |
| s.compute_tls13_traffic_secrets() |
| elif s.connection_end == "server": |
| s.compute_tls13_traffic_secrets_end() |
| s.compute_tls13_resumption_secret() |
| |
| |
| ## Additional handshake messages |
| |
| ############################################################################### |
| ### HelloVerifyRequest ### |
| ############################################################################### |
| |
| class TLSHelloVerifyRequest(_TLSHandshake): |
| """ |
| Defined for DTLS, see RFC 6347. |
| """ |
| name = "TLS Handshake - Hello Verify Request" |
| fields_desc = [ ByteEnumField("msgtype", 21, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| FieldLenField("cookielen", None, |
| fmt="B", length_of="cookie"), |
| StrLenField("cookie", "", |
| length_from=lambda pkt: pkt.cookielen) ] |
| |
| |
| ############################################################################### |
| ### CertificateURL ### |
| ############################################################################### |
| |
| _tls_cert_chain_types = { 0: "individual_certs", |
| 1: "pkipath" } |
| |
| class URLAndOptionalHash(Packet): |
| name = "URLAndOptionHash structure for TLSCertificateURL" |
| fields_desc = [ FieldLenField("urllen", None, length_of="url"), |
| StrLenField("url", "", |
| length_from=lambda pkt: pkt.urllen), |
| FieldLenField("hash_present", None, |
| fmt="B", length_of="hash", |
| adjust=lambda pkt,x: int(math.ceil(x/20.))), |
| StrLenField("hash", "", |
| length_from=lambda pkt: 20*pkt.hash_present) ] |
| def guess_payload_class(self, p): |
| return Padding |
| |
| class TLSCertificateURL(_TLSHandshake): |
| """ |
| Defined in RFC 4366. PkiPath structure of section 8 is not implemented yet. |
| """ |
| name = "TLS Handshake - Certificate URL" |
| fields_desc = [ ByteEnumField("msgtype", 21, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| ByteEnumField("certchaintype", None, _tls_cert_chain_types), |
| FieldLenField("uahlen", None, length_of="uah"), |
| PacketListField("uah", [], URLAndOptionalHash, |
| length_from=lambda pkt: pkt.uahlen) ] |
| |
| |
| ############################################################################### |
| ### CertificateStatus ### |
| ############################################################################### |
| |
| class ThreeBytesLenField(FieldLenField): |
| def __init__(self, name, default, length_of=None, adjust=lambda pkt, x:x): |
| FieldLenField.__init__(self, name, default, length_of=length_of, |
| fmt='!I', adjust=adjust) |
| def i2repr(self, pkt, x): |
| if x is None: |
| return 0 |
| return repr(self.i2h(pkt,x)) |
| def addfield(self, pkt, s, val): |
| return s+struct.pack(self.fmt, self.i2m(pkt,val))[1:4] |
| def getfield(self, pkt, s): |
| return s[3:], self.m2i(pkt, struct.unpack(self.fmt, b"\x00"+s[:3])[0]) |
| |
| _cert_status_cls = { 1: OCSP_Response } |
| |
| class _StatusField(PacketField): |
| def m2i(self, pkt, m): |
| idtype = pkt.status_type |
| cls = self.cls |
| if idtype in _cert_status_cls: |
| cls = _cert_status_cls[idtype] |
| return cls(m) |
| |
| class TLSCertificateStatus(_TLSHandshake): |
| name = "TLS Handshake - Certificate Status" |
| fields_desc = [ ByteEnumField("msgtype", 22, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| ByteEnumField("status_type", 1, _cert_status_type), |
| ThreeBytesLenField("responselen", None, |
| length_of="response"), |
| _StatusField("response", None, Raw) ] |
| |
| |
| ############################################################################### |
| ### SupplementalData ### |
| ############################################################################### |
| |
| class SupDataEntry(Packet): |
| name = "Supplemental Data Entry - Generic" |
| fields_desc = [ ShortField("sdtype", None), |
| FieldLenField("len", None, length_of="data"), |
| StrLenField("data", "", |
| length_from=lambda pkt:pkt.len) ] |
| def guess_payload_class(self, p): |
| return Padding |
| |
| class UserMappingData(Packet): |
| name = "User Mapping Data" |
| fields_desc = [ ByteField("version", None), |
| FieldLenField("len", None, length_of="data"), |
| StrLenField("data", "", |
| length_from=lambda pkt: pkt.len)] |
| def guess_payload_class(self, p): |
| return Padding |
| |
| class SupDataEntryUM(Packet): |
| name = "Supplemental Data Entry - User Mapping" |
| fields_desc = [ ShortField("sdtype", None), |
| FieldLenField("len", None, length_of="data", |
| adjust=lambda pkt, x: x+2), |
| FieldLenField("dlen", None, length_of="data"), |
| PacketListField("data", [], UserMappingData, |
| length_from=lambda pkt:pkt.dlen) ] |
| def guess_payload_class(self, p): |
| return Padding |
| |
| class TLSSupplementalData(_TLSHandshake): |
| name = "TLS Handshake - Supplemental Data" |
| fields_desc = [ ByteEnumField("msgtype", 23, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| ThreeBytesLenField("sdatalen", None, length_of="sdata"), |
| PacketListField("sdata", [], SupDataEntry, |
| length_from=lambda pkt: pkt.sdatalen) ] |
| |
| |
| ############################################################################### |
| ### NewSessionTicket ### |
| ############################################################################### |
| |
| class TLSNewSessionTicket(_TLSHandshake): |
| """ |
| XXX When knowing the right secret, we should be able to read the ticket. |
| """ |
| name = "TLS Handshake - New Session Ticket" |
| fields_desc = [ ByteEnumField("msgtype", 4, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| IntField("lifetime", 0xffffffff), |
| FieldLenField("ticketlen", None, length_of="ticket"), |
| StrLenField("ticket", "", |
| length_from=lambda pkt: pkt.ticketlen) ] |
| |
| @classmethod |
| def dispatch_hook(cls, _pkt=None, *args, **kargs): |
| s = kargs.get("tls_session", None) |
| if s and s.tls_version >= 0x0304: |
| return TLS13NewSessionTicket |
| return TLSNewSessionTicket |
| |
| def post_dissection_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| if self.tls_session.connection_end == "client": |
| self.tls_session.client_session_ticket = self.ticket |
| |
| |
| class TLS13NewSessionTicket(_TLSHandshake): |
| """ |
| Uncomment the TicketField line for parsing a RFC 5077 ticket. |
| """ |
| name = "TLS Handshake - New Session Ticket" |
| fields_desc = [ ByteEnumField("msgtype", 4, _tls_handshake_type), |
| ThreeBytesField("msglen", None), |
| IntField("ticket_lifetime", 0xffffffff), |
| IntField("ticket_age_add", 0), |
| FieldLenField("ticketlen", None, length_of="ticket"), |
| #TicketField("ticket", "", |
| StrLenField("ticket", "", |
| length_from=lambda pkt: pkt.ticketlen), |
| _ExtensionsLenField("extlen", None, length_of="ext"), |
| _ExtensionsField("ext", None, |
| length_from=lambda pkt: (pkt.msglen - |
| (pkt.ticketlen or 0) - |
| 12)) ] |
| |
| def post_dissection_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| if self.tls_session.connection_end == "client": |
| self.tls_session.client_session_ticket = self.ticket |
| |
| |
| ############################################################################### |
| ### All handshake messages defined in this module ### |
| ############################################################################### |
| |
| _tls_handshake_cls = { 0: TLSHelloRequest, 1: TLSClientHello, |
| 2: TLSServerHello, 3: TLSHelloVerifyRequest, |
| 4: TLSNewSessionTicket, 6: TLSHelloRetryRequest, |
| 8: TLSEncryptedExtensions, 11: TLSCertificate, |
| 12: TLSServerKeyExchange, 13: TLSCertificateRequest, |
| 14: TLSServerHelloDone, 15: TLSCertificateVerify, |
| 16: TLSClientKeyExchange, 20: TLSFinished, |
| 21: TLSCertificateURL, 22: TLSCertificateStatus, |
| 23: TLSSupplementalData } |
| |
