| ## This file is part of Scapy |
| ## Copyright (C) 2017 Maxence Tury |
| ## This program is published under a GPLv2 license |
| |
| """ |
| TLS 1.3 key exchange logic. |
| """ |
| |
| import math |
| |
| from scapy.config import conf, crypto_validator |
| from scapy.error import log_runtime, warning |
| from scapy.fields import * |
| from scapy.packet import Packet, Raw, Padding |
| from scapy.layers.tls.cert import PubKeyRSA, PrivKeyRSA |
| from scapy.layers.tls.session import _GenericTLSSessionInheritance |
| from scapy.layers.tls.basefields import _tls_version, _TLSClientVersionField |
| from scapy.layers.tls.extensions import TLS_Ext_Unknown, _tls_ext |
| from scapy.layers.tls.crypto.pkcs1 import pkcs_i2osp, pkcs_os2ip |
| from scapy.layers.tls.crypto.groups import (_tls_named_ffdh_groups, |
| _tls_named_curves, _ffdh_groups, |
| _tls_named_groups) |
| |
| if conf.crypto_valid: |
| from cryptography.hazmat.backends import default_backend |
| from cryptography.hazmat.primitives.asymmetric import dh, ec |
| if conf.crypto_valid_advanced: |
| from cryptography.hazmat.primitives.asymmetric import x25519 |
| |
| |
| class KeyShareEntry(Packet): |
| """ |
| When building from scratch, we create a DH private key, and when |
| dissecting, we create a DH public key. Default group is secp256r1. |
| """ |
| __slots__ = ["privkey", "pubkey"] |
| name = "Key Share Entry" |
| fields_desc = [ShortEnumField("group", None, _tls_named_groups), |
| FieldLenField("kxlen", None, length_of="key_exchange"), |
| StrLenField("key_exchange", "", |
| length_from=lambda pkt: pkt.kxlen) ] |
| |
| def __init__(self, *args, **kargs): |
| self.privkey = None |
| self.pubkey = None |
| super(KeyShareEntry, self).__init__(*args, **kargs) |
| |
| def do_build(self): |
| """ |
| We need this hack, else 'self' would be replaced by __iter__.next(). |
| """ |
| tmp = self.explicit |
| self.explicit = True |
| b = super(KeyShareEntry, self).do_build() |
| self.explicit = tmp |
| return b |
| |
| @crypto_validator |
| def create_privkey(self): |
| """ |
| This is called by post_build() for key creation. |
| """ |
| if self.group in _tls_named_ffdh_groups: |
| params = _ffdh_groups[_tls_named_ffdh_groups[self.group]][0] |
| privkey = params.generate_private_key() |
| self.privkey = privkey |
| pubkey = privkey.public_key() |
| self.key_exchange = pubkey.public_numbers().y |
| elif self.group in _tls_named_curves: |
| if _tls_named_curves[self.group] == "x25519": |
| if conf.crypto_valid_advanced: |
| privkey = x25519.X25519PrivateKey.generate() |
| self.privkey = privkey |
| pubkey = privkey.public_key() |
| self.key_exchange = pubkey.public_bytes() |
| elif _tls_named_curves[self.group] != "x448": |
| curve = ec._CURVE_TYPES[_tls_named_curves[self.group]]() |
| privkey = ec.generate_private_key(curve, default_backend()) |
| self.privkey = privkey |
| pubkey = privkey.public_key() |
| self.key_exchange = pubkey.public_numbers().encode_point() |
| |
| def post_build(self, pkt, pay): |
| if self.group is None: |
| self.group = 23 # secp256r1 |
| |
| if not self.key_exchange: |
| try: |
| self.create_privkey() |
| except ImportError: |
| pass |
| |
| if self.kxlen is None: |
| self.kxlen = len(self.key_exchange) |
| |
| group = struct.pack("!H", self.group) |
| kxlen = struct.pack("!H", self.kxlen) |
| return group + kxlen + self.key_exchange + pay |
| |
| @crypto_validator |
| def register_pubkey(self): |
| if self.group in _tls_named_ffdh_groups: |
| params = _ffdh_groups[_tls_named_ffdh_groups[self.group]][0] |
| pn = params.parameter_numbers() |
| public_numbers = dh.DHPublicNumbers(self.key_exchange, pn) |
| self.pubkey = public_numbers.public_key(default_backend()) |
| elif self.group in _tls_named_curves: |
| if _tls_named_curves[self.group] == "x25519": |
| if conf.crypto_valid_advanced: |
| import_point = x25519.X25519PublicKey.from_public_bytes |
| self.pubkey = import_point(self.key_exchange) |
| elif _tls_named_curves[self.group] != "x448": |
| curve = ec._CURVE_TYPES[_tls_named_curves[self.group]]() |
| import_point = ec.EllipticCurvePublicNumbers.from_encoded_point |
| public_numbers = import_point(curve, self.key_exchange) |
| self.pubkey = public_numbers.public_key(default_backend()) |
| |
| def post_dissection(self, r): |
| try: |
| self.register_pubkey() |
| except ImportError: |
| pass |
| |
| def extract_padding(self, s): |
| return "", s |
| |
| |
| class TLS_Ext_KeyShare_CH(TLS_Ext_Unknown): |
| name = "TLS Extension - Key Share (for ClientHello)" |
| fields_desc = [ShortEnumField("type", 0x28, _tls_ext), |
| ShortField("len", None), |
| FieldLenField("client_shares_len", None, |
| length_of="client_shares"), |
| PacketListField("client_shares", [], KeyShareEntry, |
| length_from=lambda pkt: pkt.client_shares_len) ] |
| |
| def post_build(self, pkt, pay): |
| if not self.tls_session.frozen: |
| privshares = self.tls_session.tls13_client_privshares |
| for kse in self.client_shares: |
| if kse.privkey: |
| if _tls_named_curves[kse.group] in privshares: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: group %s used twice in the same ClientHello [%s]", kse.group, pkt_info) |
| break |
| privshares[_tls_named_groups[kse.group]] = kse.privkey |
| return super(TLS_Ext_KeyShare_CH, self).post_build(pkt, pay) |
| |
| def post_dissection(self, r): |
| if not self.tls_session.frozen: |
| for kse in self.client_shares: |
| if kse.pubkey: |
| pubshares = self.tls_session.tls13_client_pubshares |
| if _tls_named_curves[kse.group] in pubshares: |
| pkt_info = r.firstlayer().summary() |
| log_runtime.info("TLS: group %s used twice in the same ClientHello [%s]", kse.group, pkt_info) |
| break |
| pubshares[_tls_named_curves[kse.group]] = kse.pubkey |
| return super(TLS_Ext_KeyShare_CH, self).post_dissection(r) |
| |
| |
| class TLS_Ext_KeyShare_HRR(TLS_Ext_Unknown): |
| name = "TLS Extension - Key Share (for HelloRetryRequest)" |
| fields_desc = [ShortEnumField("type", 0x28, _tls_ext), |
| ShortField("len", None), |
| ShortEnumField("selected_group", None, _tls_named_groups) ] |
| |
| |
| class TLS_Ext_KeyShare_SH(TLS_Ext_Unknown): |
| name = "TLS Extension - Key Share (for ServerHello)" |
| fields_desc = [ShortEnumField("type", 0x28, _tls_ext), |
| ShortField("len", None), |
| PacketField("server_share", None, KeyShareEntry) ] |
| |
| def post_build(self, pkt, pay): |
| if not self.tls_session.frozen and self.server_share.privkey: |
| # if there is a privkey, we assume the crypto library is ok |
| privshare = self.tls_session.tls13_server_privshare |
| if len(privshare) > 0: |
| pkt_info = pkt.firstlayer().summary() |
| log_runtime.info("TLS: overwriting previous server key share [%s]", pkt_info) |
| group_name = _tls_named_groups[self.server_share.group] |
| privshare[group_name] = self.server_share.privkey |
| |
| if group_name in self.tls_session.tls13_client_pubshares: |
| privkey = self.server_share.privkey |
| pubkey = self.tls_session.tls13_client_pubshares[group_name] |
| if group_name in six.itervalues(_tls_named_ffdh_groups): |
| pms = privkey.exchange(pubkey) |
| elif group_name in six.itervalues(_tls_named_curves): |
| if group_name == "x25519": |
| pms = privkey.exchange(pubkey) |
| else: |
| pms = privkey.exchange(ec.ECDH(), pubkey) |
| self.tls_session.tls13_dhe_secret = pms |
| return super(TLS_Ext_KeyShare_SH, self).post_build(pkt, pay) |
| |
| def post_dissection(self, r): |
| if not self.tls_session.frozen and self.server_share.pubkey: |
| # if there is a pubkey, we assume the crypto library is ok |
| pubshare = self.tls_session.tls13_server_pubshare |
| if len(pubshare) > 0: |
| pkt_info = r.firstlayer().summary() |
| log_runtime.info("TLS: overwriting previous server key share [%s]", pkt_info) |
| group_name = _tls_named_groups[self.server_share.group] |
| pubshare[group_name] = self.server_share.pubkey |
| |
| if group_name in self.tls_session.tls13_client_privshares: |
| pubkey = self.server_share.pubkey |
| privkey = self.tls_session.tls13_client_privshares[group_name] |
| if group_name in six.itervalues(_tls_named_ffdh_groups): |
| pms = privkey.exchange(pubkey) |
| elif group_name in six.itervalues(_tls_named_curves): |
| if group_name == "x25519": |
| pms = privkey.exchange(pubkey) |
| else: |
| pms = privkey.exchange(ec.ECDH(), pubkey) |
| self.tls_session.tls13_dhe_secret = pms |
| return super(TLS_Ext_KeyShare_SH, self).post_dissection(r) |
| |
| |
| _tls_ext_keyshare_cls = { 1: TLS_Ext_KeyShare_CH, |
| 2: TLS_Ext_KeyShare_SH, |
| 6: TLS_Ext_KeyShare_HRR } |
| |
| |
| class Ticket(Packet): |
| name = "Recommended Ticket Construction (from RFC 5077)" |
| fields_desc = [ StrFixedLenField("key_name", None, 16), |
| StrFixedLenField("iv", None, 16), |
| FieldLenField("encstatelen", None, length_of="encstate"), |
| StrLenField("encstate", "", |
| length_from=lambda pkt: pkt.encstatelen), |
| StrFixedLenField("mac", None, 32) ] |
| |
| class TicketField(PacketField): |
| __slots__ = ["length_from"] |
| def __init__(self, name, default, length_from=None, **kargs): |
| self.length_from = length_from |
| PacketField.__init__(self, name, default, Ticket, **kargs) |
| |
| def m2i(self, pkt, m): |
| l = self.length_from(pkt) |
| tbd, rem = m[:l], m[l:] |
| return self.cls(tbd)/Padding(rem) |
| |
| class PSKIdentity(Packet): |
| name = "PSK Identity" |
| fields_desc = [FieldLenField("identity_len", None, |
| length_of="identity"), |
| TicketField("identity", "", |
| length_from=lambda pkt: pkt.identity_len), |
| IntField("obfuscated_ticket_age", 0) ] |
| |
| class PSKBinderEntry(Packet): |
| name = "PSK Binder Entry" |
| fields_desc = [FieldLenField("binder_len", None, fmt="B", |
| length_of="binder"), |
| StrLenField("binder", "", |
| length_from=lambda pkt: pkt.binder_len) ] |
| |
| class TLS_Ext_PreSharedKey_CH(TLS_Ext_Unknown): |
| #XXX define post_build and post_dissection methods |
| name = "TLS Extension - Pre Shared Key (for ClientHello)" |
| fields_desc = [ShortEnumField("type", 0x28, _tls_ext), |
| ShortField("len", None), |
| FieldLenField("identities_len", None, |
| length_of="identities"), |
| PacketListField("identities", [], PSKIdentity, |
| length_from=lambda pkt: pkt.identities_len), |
| FieldLenField("binders_len", None, |
| length_of="binders"), |
| PacketListField("binders", [], PSKBinderEntry, |
| length_from=lambda pkt: pkt.binders_len) ] |
| |
| |
| class TLS_Ext_PreSharedKey_SH(TLS_Ext_Unknown): |
| name = "TLS Extension - Pre Shared Key (for ServerHello)" |
| fields_desc = [ShortEnumField("type", 0x29, _tls_ext), |
| ShortField("len", None), |
| ShortField("selected_identity", None) ] |
| |
| |
| _tls_ext_presharedkey_cls = { 1: TLS_Ext_PreSharedKey_CH, |
| 2: TLS_Ext_PreSharedKey_SH } |
| |
