| ## This file is part of Scapy |
| ## Copyright (C) 2017 Maxence Tury |
| ## This program is published under a GPLv2 license |
| |
| """ |
| SSLv2 handshake fields & logic. |
| """ |
| |
| import math |
| |
| from scapy.error import log_runtime, warning |
| from scapy.fields import * |
| from scapy.packet import Packet, Raw, Padding |
| from scapy.layers.tls.cert import Cert, PrivKey, PubKey |
| from scapy.layers.tls.basefields import _tls_version, _TLSVersionField |
| from scapy.layers.tls.handshake import _CipherSuitesField |
| from scapy.layers.tls.keyexchange import _TLSSignatureField, _TLSSignature |
| from scapy.layers.tls.session import (_GenericTLSSessionInheritance, |
| readConnState, writeConnState) |
| from scapy.layers.tls.crypto.suites import (_tls_cipher_suites, |
| _tls_cipher_suites_cls, |
| _GenericCipherSuite, |
| _GenericCipherSuiteMetaclass, |
| get_usable_ciphersuites, |
| SSL_CK_DES_192_EDE3_CBC_WITH_MD5) |
| |
| |
| ############################################################################### |
| ### Generic SSLv2 Handshake message ### |
| ############################################################################### |
| |
| _sslv2_handshake_type = { 0: "error", 1: "client_hello", |
| 2: "client_master_key", 3: "client_finished", |
| 4: "server_hello", 5: "server_verify", |
| 6: "server_finished", 7: "request_certificate", |
| 8: "client_certificate" } |
| |
| |
| class _SSLv2Handshake(_GenericTLSSessionInheritance): |
| """ |
| Inherited by other Handshake classes to get post_build(). |
| Also used as a fallback for unknown TLS Handshake packets. |
| """ |
| name = "SSLv2 Handshake Generic message" |
| fields_desc = [ ByteEnumField("msgtype", None, _sslv2_handshake_type) ] |
| |
| def guess_payload_class(self, p): |
| return Padding |
| |
| 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) |
| |
| |
| ############################################################################### |
| ### Error ### |
| ############################################################################### |
| |
| _tls_error_code = { 1: "no_cipher", 2: "no_certificate", |
| 4: "bad_certificate", 6: "unsupported_certificate_type" } |
| |
| class SSLv2Error(_SSLv2Handshake): |
| """ |
| SSLv2 Error. |
| """ |
| name = "SSLv2 Handshake - Error" |
| fields_desc = [ ByteEnumField("msgtype", 0, _sslv2_handshake_type), |
| ShortEnumField("code", None, _tls_error_code) ] |
| |
| |
| ############################################################################### |
| ### ClientHello ### |
| ############################################################################### |
| |
| class _SSLv2CipherSuitesField(_CipherSuitesField): |
| def __init__(self, name, default, dico, length_from=None): |
| _CipherSuitesField.__init__(self, name, default, dico, |
| length_from=length_from) |
| self.itemfmt = b"" |
| self.itemsize = 3 |
| |
| def i2m(self, pkt, val): |
| if val is None: |
| val2 = [] |
| val2 = [(x >> 16, x & 0x00ffff) for x in val] |
| return b"".join([struct.pack(">BH", x[0], x[1]) for x in val2]) |
| |
| def m2i(self, pkt, m): |
| res = [] |
| while m: |
| res.append(struct.unpack("!I", b"\x00" + m[:3])[0]) |
| m = m[3:] |
| return res |
| |
| |
| class SSLv2ClientHello(_SSLv2Handshake): |
| """ |
| SSLv2 ClientHello. |
| """ |
| name = "SSLv2 Handshake - Client Hello" |
| fields_desc = [ ByteEnumField("msgtype", 1, _sslv2_handshake_type), |
| _TLSVersionField("version", 0x0002, _tls_version), |
| |
| FieldLenField("cipherslen", None, fmt="!H", |
| length_of="ciphers"), |
| FieldLenField("sidlen", None, fmt="!H", |
| length_of="sid"), |
| FieldLenField("challengelen", None, fmt="!H", |
| length_of="challenge"), |
| |
| XStrLenField("sid", b"", |
| length_from=lambda pkt:pkt.sidlen), |
| _SSLv2CipherSuitesField("ciphers", |
| [SSL_CK_DES_192_EDE3_CBC_WITH_MD5], |
| _tls_cipher_suites, |
| length_from=lambda pkt: pkt.cipherslen), |
| XStrLenField("challenge", b"", |
| length_from=lambda pkt:pkt.challengelen) ] |
| |
| def tls_session_update(self, msg_str): |
| super(SSLv2ClientHello, self).tls_session_update(msg_str) |
| self.tls_session.advertised_tls_version = self.version |
| self.tls_session.sslv2_common_cs = self.ciphers |
| self.tls_session.sslv2_challenge = self.challenge |
| |
| |
| ############################################################################### |
| ### ServerHello ### |
| ############################################################################### |
| |
| class _SSLv2CertDataField(StrLenField): |
| def getfield(self, pkt, s): |
| l = 0 |
| if self.length_from is not None: |
| l = self.length_from(pkt) |
| try: |
| certdata = Cert(s[:l]) |
| except: |
| certdata = s[:l] |
| return s[l:], certdata |
| |
| def i2len(self, pkt, i): |
| if isinstance(i, Cert): |
| return len(i.der) |
| return len(i) |
| |
| def i2m(self, pkt, i): |
| if isinstance(i, Cert): |
| return i.der |
| return i |
| |
| |
| class SSLv2ServerHello(_SSLv2Handshake): |
| """ |
| SSLv2 ServerHello. |
| """ |
| name = "SSLv2 Handshake - Server Hello" |
| fields_desc = [ ByteEnumField("msgtype", 4, _sslv2_handshake_type), |
| |
| ByteField("sid_hit", 0), |
| ByteEnumField("certtype", 1, {1: "x509_cert"}), |
| _TLSVersionField("version", 0x0002, _tls_version), |
| |
| FieldLenField("certlen", None, fmt="!H", |
| length_of="cert"), |
| FieldLenField("cipherslen", None, fmt="!H", |
| length_of="ciphers"), |
| FieldLenField("connection_idlen", None, fmt="!H", |
| length_of="connection_id"), |
| |
| _SSLv2CertDataField("cert", b"", |
| length_from=lambda pkt: pkt.certlen), |
| _SSLv2CipherSuitesField("ciphers", [], _tls_cipher_suites, |
| length_from=lambda pkt: pkt.cipherslen), |
| XStrLenField("connection_id", b"", |
| length_from=lambda pkt: pkt.connection_idlen) ] |
| |
| def tls_session_update(self, msg_str): |
| """ |
| XXX Something should be done about the session ID here. |
| """ |
| super(SSLv2ServerHello, self).tls_session_update(msg_str) |
| |
| s = self.tls_session |
| client_cs = s.sslv2_common_cs |
| css = [cs for cs in client_cs if cs in self.ciphers] |
| s.sslv2_common_cs = css |
| s.sslv2_connection_id = self.connection_id |
| s.tls_version = self.version |
| if self.cert is not None: |
| s.server_certs = [self.cert] |
| |
| |
| ############################################################################### |
| ### ClientMasterKey ### |
| ############################################################################### |
| |
| class _SSLv2CipherSuiteField(EnumField): |
| def __init__(self, name, default, dico): |
| EnumField.__init__(self, name, default, dico) |
| |
| def i2m(self, pkt, val): |
| if val is None: |
| return b"" |
| val2 = (val >> 16, val & 0x00ffff) |
| return struct.pack(">BH", val2[0], val2[1]) |
| |
| def addfield(self, pkt, s, val): |
| return s + self.i2m(pkt, val) |
| |
| def m2i(self, pkt, m): |
| return struct.unpack("!I", b"\x00" + m[:3])[0] |
| |
| def getfield(self, pkt, s): |
| return s[3:], self.m2i(pkt, s) |
| |
| class _SSLv2EncryptedKeyField(XStrLenField): |
| def i2repr(self, pkt, x): |
| s = super(_SSLv2EncryptedKeyField, self).i2repr(pkt, x) |
| if pkt.decryptedkey is not None: |
| dx = pkt.decryptedkey |
| ds = super(_SSLv2EncryptedKeyField, self).i2repr(pkt, dx) |
| s += " [decryptedkey= %s]" % ds |
| return s |
| |
| class SSLv2ClientMasterKey(_SSLv2Handshake): |
| """ |
| SSLv2 ClientMasterKey. |
| """ |
| __slots__ = ["decryptedkey"] |
| name = "SSLv2 Handshake - Client Master Key" |
| fields_desc = [ ByteEnumField("msgtype", 2, _sslv2_handshake_type), |
| _SSLv2CipherSuiteField("cipher", None, _tls_cipher_suites), |
| |
| FieldLenField("clearkeylen", None, fmt="!H", |
| length_of="clearkey"), |
| FieldLenField("encryptedkeylen", None, fmt="!H", |
| length_of="encryptedkey"), |
| FieldLenField("keyarglen", None, fmt="!H", |
| length_of="keyarg"), |
| |
| XStrLenField("clearkey", "", |
| length_from=lambda pkt: pkt.clearkeylen), |
| _SSLv2EncryptedKeyField("encryptedkey", "", |
| length_from=lambda pkt: pkt.encryptedkeylen), |
| XStrLenField("keyarg", "", |
| length_from=lambda pkt: pkt.keyarglen) ] |
| |
| def __init__(self, *args, **kargs): |
| """ |
| When post_building, the packets fields are updated (this is somewhat |
| non-standard). We might need these fields later, but calling __str__ |
| on a new packet (i.e. not dissected from a raw string) applies |
| post_build to an object different from the original one... unless |
| we hackishly always set self.explicit to 1. |
| """ |
| if "decryptedkey" in kargs: |
| self.decryptedkey = kargs["decryptedkey"] |
| del kargs["decryptedkey"] |
| else: |
| self.decryptedkey = b"" |
| super(SSLv2ClientMasterKey, self).__init__(*args, **kargs) |
| self.explicit = 1 |
| |
| def pre_dissect(self, s): |
| clearkeylen = struct.unpack("!H", s[4:6])[0] |
| encryptedkeylen = struct.unpack("!H", s[6:8])[0] |
| encryptedkeystart = 10 + clearkeylen |
| encryptedkey = s[encryptedkeystart:encryptedkeystart+encryptedkeylen] |
| if self.tls_session.server_rsa_key: |
| self.decryptedkey = \ |
| self.tls_session.server_rsa_key.decrypt(encryptedkey) |
| else: |
| self.decryptedkey = None |
| return s |
| |
| def post_build(self, pkt, pay): |
| cs_val = None |
| if self.cipher is None: |
| common_cs = self.tls_session.sslv2_common_cs |
| cs_vals = get_usable_ciphersuites(common_cs, "SSLv2") |
| if len(cs_vals) == 0: |
| warning("No known common cipher suite between SSLv2 Hellos.") |
| cs_val = 0x0700c0 |
| cipher = b"\x07\x00\xc0" |
| else: |
| cs_val = cs_vals[0] #XXX choose the best one |
| cipher = struct.pack(">BH", cs_val >> 16, cs_val & 0x00ffff) |
| cs_cls = _tls_cipher_suites_cls[cs_val] |
| self.cipher = cs_val |
| else: |
| cipher = pkt[1:4] |
| cs_val = struct.unpack("!I", b"\x00" + cipher)[0] |
| if cs_val not in _tls_cipher_suites_cls: |
| warning("Unknown ciphersuite %d from ClientMasterKey" % cs_val) |
| cs_cls = None |
| else: |
| cs_cls = _tls_cipher_suites_cls[cs_val] |
| |
| if cs_cls: |
| if (self.encryptedkey == b"" and |
| len(self.tls_session.server_certs) > 0): |
| # else, the user is responsible for export slicing & encryption |
| key = randstring(cs_cls.cipher_alg.key_len) |
| |
| if self.clearkey == b"" and cs_cls.kx_alg.export: |
| self.clearkey = key[:-5] |
| |
| if self.decryptedkey == b"": |
| if cs_cls.kx_alg.export: |
| self.decryptedkey = key[-5:] |
| else: |
| self.decryptedkey = key |
| |
| pubkey = self.tls_session.server_certs[0].pubKey |
| self.encryptedkey = pubkey.encrypt(self.decryptedkey) |
| |
| if self.keyarg == b"" and cs_cls.cipher_alg.type == "block": |
| self.keyarg = randstring(cs_cls.cipher_alg.block_size) |
| |
| clearkey = self.clearkey or b"" |
| if self.clearkeylen is None: |
| self.clearkeylen = len(clearkey) |
| clearkeylen = struct.pack("!H", self.clearkeylen) |
| |
| encryptedkey = self.encryptedkey or b"" |
| if self.encryptedkeylen is None: |
| self.encryptedkeylen = len(encryptedkey) |
| encryptedkeylen = struct.pack("!H", self.encryptedkeylen) |
| |
| keyarg = self.keyarg or b"" |
| if self.keyarglen is None: |
| self.keyarglen = len(keyarg) |
| keyarglen = struct.pack("!H", self.keyarglen) |
| |
| s = (chb(pkt[0]) + cipher |
| + clearkeylen + encryptedkeylen + keyarglen |
| + clearkey + encryptedkey + keyarg) |
| return s + pay |
| |
| def tls_session_update(self, msg_str): |
| super(SSLv2ClientMasterKey, self).tls_session_update(msg_str) |
| |
| s = self.tls_session |
| cs_val = self.cipher |
| if cs_val not in _tls_cipher_suites_cls: |
| warning("Unknown cipher suite %d from ClientMasterKey" % cs_val) |
| cs_cls = None |
| else: |
| cs_cls = _tls_cipher_suites_cls[cs_val] |
| |
| tls_version = s.tls_version or 0x0002 |
| connection_end = s.connection_end |
| wcs_seq_num = s.wcs.seq_num |
| s.pwcs = writeConnState(ciphersuite=cs_cls, |
| connection_end=connection_end, |
| seq_num=wcs_seq_num, |
| tls_version=tls_version) |
| rcs_seq_num = s.rcs.seq_num |
| s.prcs = readConnState(ciphersuite=cs_cls, |
| connection_end=connection_end, |
| seq_num=rcs_seq_num, |
| tls_version=tls_version) |
| |
| if self.decryptedkey is not None: |
| s.master_secret = self.clearkey + self.decryptedkey |
| s.compute_sslv2_km_and_derive_keys() |
| |
| if s.pwcs.cipher.type == "block": |
| s.pwcs.cipher.iv = self.keyarg |
| if s.prcs.cipher.type == "block": |
| s.prcs.cipher.iv = self.keyarg |
| |
| s.triggered_prcs_commit = True |
| s.triggered_pwcs_commit = True |
| |
| |
| ############################################################################### |
| ### ServerVerify ### |
| ############################################################################### |
| |
| class SSLv2ServerVerify(_SSLv2Handshake): |
| """ |
| In order to parse a ServerVerify, the exact message string should be |
| fed to the class. This is how SSLv2 defines the challenge length... |
| """ |
| name = "SSLv2 Handshake - Server Verify" |
| fields_desc = [ ByteEnumField("msgtype", 5, _sslv2_handshake_type), |
| XStrField("challenge", "") ] |
| |
| def build(self, *args, **kargs): |
| fval = self.getfieldval("challenge") |
| if fval is None: |
| self.challenge = self.tls_session.sslv2_challenge |
| return super(SSLv2ServerVerify, self).build(*args, **kargs) |
| |
| def post_dissection(self, pkt): |
| s = self.tls_session |
| if s.sslv2_challenge is not None: |
| if self.challenge != s.sslv2_challenge: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: invalid ServerVerify received [%s]", pkt_info) |
| |
| |
| ############################################################################### |
| ### RequestCertificate ### |
| ############################################################################### |
| |
| class SSLv2RequestCertificate(_SSLv2Handshake): |
| """ |
| In order to parse a RequestCertificate, the exact message string should be |
| fed to the class. This is how SSLv2 defines the challenge length... |
| """ |
| name = "SSLv2 Handshake - Request Certificate" |
| fields_desc = [ ByteEnumField("msgtype", 7, _sslv2_handshake_type), |
| ByteEnumField("authtype", 1, {1: "md5_with_rsa"}), |
| XStrField("challenge", "") ] |
| |
| def tls_session_update(self, msg_str): |
| super(SSLv2RequestCertificate, self).tls_session_update(msg_str) |
| self.tls_session.sslv2_challenge_clientcert = self.challenge |
| |
| |
| ############################################################################### |
| ### ClientCertificate ### |
| ############################################################################### |
| |
| class SSLv2ClientCertificate(_SSLv2Handshake): |
| """ |
| SSLv2 ClientCertificate. |
| """ |
| name = "SSLv2 Handshake - Client Certificate" |
| fields_desc = [ ByteEnumField("msgtype", 8, _sslv2_handshake_type), |
| |
| ByteEnumField("certtype", 1, {1: "x509_cert"}), |
| FieldLenField("certlen", None, fmt="!H", |
| length_of="certdata"), |
| FieldLenField("responselen", None, fmt="!H", |
| length_of="responsedata"), |
| |
| _SSLv2CertDataField("certdata", b"", |
| length_from=lambda pkt: pkt.certlen), |
| _TLSSignatureField("responsedata", None, |
| length_from=lambda pkt: pkt.responselen) ] |
| |
| def build(self, *args, **kargs): |
| s = self.tls_session |
| sig = self.getfieldval("responsedata") |
| test = (sig is None and |
| s.sslv2_key_material is not None and |
| s.sslv2_challenge_clientcert is not None and |
| len(s.server_certs) > 0) |
| if test: |
| s = self.tls_session |
| m = (s.sslv2_key_material + |
| s.sslv2_challenge_clientcert + |
| s.server_certs[0].der) |
| self.responsedata = _TLSSignature(tls_session=s) |
| self.responsedata._update_sig(m, s.client_key) |
| else: |
| self.responsedata = b"" |
| return super(SSLv2ClientCertificate, self).build(*args, **kargs) |
| |
| def post_dissection_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| |
| s = self.tls_session |
| test = (len(s.client_certs) > 0 and |
| s.sslv2_key_material is not None and |
| s.sslv2_challenge_clientcert is not None and |
| len(s.server_certs) > 0) |
| if test: |
| m = (s.sslv2_key_material + |
| s.sslv2_challenge_clientcert + |
| s.server_certs[0].der) |
| sig_test = self.responsedata._verify_sig(m, s.client_certs[0]) |
| if not sig_test: |
| pkt_info = self.firstlayer().summary() |
| log_runtime.info("TLS: invalid client CertificateVerify signature [%s]", pkt_info) |
| |
| def tls_session_update(self, msg_str): |
| super(SSLv2ClientCertificate, self).tls_session_update(msg_str) |
| if self.certdata: |
| self.tls_session.client_certs = [self.certdata] |
| |
| |
| ############################################################################### |
| ### Finished ### |
| ############################################################################### |
| |
| class SSLv2ClientFinished(_SSLv2Handshake): |
| """ |
| In order to parse a ClientFinished, the exact message string should be fed |
| to the class. SSLv2 does not offer any other way to know the c_id length. |
| """ |
| name = "SSLv2 Handshake - Client Finished" |
| fields_desc = [ ByteEnumField("msgtype", 3, _sslv2_handshake_type), |
| XStrField("connection_id", "") ] |
| |
| def build(self, *args, **kargs): |
| fval = self.getfieldval("connection_id") |
| if fval == b"": |
| self.connection_id = self.tls_session.sslv2_connection_id |
| return super(SSLv2ClientFinished, self).build(*args, **kargs) |
| |
| def post_dissection(self, pkt): |
| s = self.tls_session |
| if s.sslv2_connection_id is not None: |
| if self.connection_id != s.sslv2_connection_id: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: invalid client Finished received [%s]", pkt_info) |
| |
| |
| class SSLv2ServerFinished(_SSLv2Handshake): |
| """ |
| In order to parse a ServerFinished, the exact message string should be fed |
| to the class. SSLv2 does not offer any other way to know the sid length. |
| """ |
| name = "SSLv2 Handshake - Server Finished" |
| fields_desc = [ ByteEnumField("msgtype", 6, _sslv2_handshake_type), |
| XStrField("sid", "") ] |
| |
| def build(self, *args, **kargs): |
| fval = self.getfieldval("sid") |
| if fval == b"": |
| self.sid = self.tls_session.sid |
| return super(SSLv2ServerFinished, self).build(*args, **kargs) |
| |
| def post_dissection_tls_session_update(self, msg_str): |
| self.tls_session_update(msg_str) |
| self.tls_session.sid = self.sid |
| |
| |
| ############################################################################### |
| ### All handshake messages defined in this module ### |
| ############################################################################### |
| |
| _sslv2_handshake_cls = { 0: SSLv2Error, 1: SSLv2ClientHello, |
| 2: SSLv2ClientMasterKey, 3: SSLv2ClientFinished, |
| 4: SSLv2ServerHello, 5: SSLv2ServerVerify, |
| 6: SSLv2ServerFinished, 7: SSLv2RequestCertificate, |
| 8: SSLv2ClientCertificate } |
| |
