| # This file is part of Scapy |
| ## See http://www.secdev.org/projects/scapy for more informations |
| ## Copyright (C) Sabrina Dubroca <sd@queasysnail.net> |
| ## This program is published under a GPLv2 license |
| |
| """ |
| Classes and functions for MACsec. |
| """ |
| |
| from __future__ import absolute_import |
| from __future__ import print_function |
| import struct |
| |
| from scapy.config import conf |
| from scapy.fields import * |
| from scapy.packet import Packet, Raw, bind_layers |
| from scapy.layers.l2 import Ether, Dot1AD, Dot1Q |
| from scapy.layers.eap import MACsecSCI |
| from scapy.layers.inet import IP |
| from scapy.layers.inet6 import IPv6 |
| import scapy.modules.six as six |
| |
| if conf.crypto_valid: |
| from cryptography.exceptions import InvalidTag |
| from cryptography.hazmat.backends import default_backend |
| from cryptography.hazmat.primitives.ciphers import ( |
| Cipher, |
| algorithms, |
| modes, |
| ) |
| else: |
| log_loading.info("Can't import python-cryptography v1.7+. " |
| "Disabled MACsec encryption/authentication.") |
| |
| |
| NOSCI_LEN = 14 + 6 |
| SCI_LEN = 8 |
| DEFAULT_ICV_LEN = 16 |
| |
| |
| class MACsecSA(object): |
| """Representation of a MACsec Secure Association |
| |
| Provides encapsulation, decapsulation, encryption, and decryption |
| of MACsec frames |
| """ |
| def __init__(self, sci, an, pn, key, icvlen, encrypt, send_sci): |
| if isinstance(sci, six.integer_types): |
| self.sci = struct.pack('!Q', sci) |
| elif isinstance(sci, bytes): |
| self.sci = sci |
| else: |
| raise TypeError("SCI must be either bytes or int") |
| self.an = an |
| self.pn = pn |
| self.key = key |
| self.icvlen = icvlen |
| self.do_encrypt = encrypt |
| self.send_sci = send_sci |
| |
| def make_iv(self, pkt): |
| """generate an IV for the packet""" |
| return self.sci + struct.pack('!I', pkt[MACsec].pn) |
| |
| @staticmethod |
| def split_pkt(pkt, assoclen, icvlen=0): |
| """ |
| split the packet into associated data, plaintext or ciphertext, and |
| optional ICV |
| """ |
| data = raw(pkt) |
| assoc = data[:assoclen] |
| if icvlen: |
| icv = data[-icvlen:] |
| enc = data[assoclen:-icvlen] |
| else: |
| icv = b'' |
| enc = data[assoclen:] |
| return assoc, enc, icv |
| |
| def e_bit(self): |
| """returns the value of the E bit for packets sent through this SA""" |
| return self.do_encrypt |
| |
| def c_bit(self): |
| """returns the value of the C bit for packets sent through this SA""" |
| return self.do_encrypt or self.icvlen != DEFAULT_ICV_LEN |
| |
| @staticmethod |
| def shortlen(pkt): |
| """determine shortlen for a raw packet (not encapsulated yet)""" |
| datalen = len(pkt) - 2*6 |
| if datalen < 48: |
| return datalen |
| return 0 |
| |
| def encap(self, pkt): |
| """encapsulate a frame using this Secure Association""" |
| if pkt.name != Ether().name: |
| raise TypeError('cannot encapsulate packet in MACsec, must be Ethernet') |
| hdr = copy.deepcopy(pkt) |
| payload = hdr.payload |
| del hdr.payload |
| tag = MACsec(sci=self.sci, an=self.an, |
| SC=self.send_sci, |
| E=self.e_bit(), C=self.c_bit(), |
| shortlen=MACsecSA.shortlen(pkt), |
| pn=self.pn, type=pkt.type) |
| hdr.type = ETH_P_MACSEC |
| return hdr/tag/payload |
| |
| # this doesn't really need to be a method, but for symmetry with |
| # encap(), it is |
| def decap(self, orig_pkt): |
| """decapsulate a MACsec frame""" |
| if orig_pkt.name != Ether().name or orig_pkt.payload.name != MACsec().name: |
| raise TypeError('cannot decapsulate MACsec packet, must be Ethernet/MACsec') |
| packet = copy.deepcopy(orig_pkt) |
| prev_layer = packet[MACsec].underlayer |
| prev_layer.type = packet[MACsec].type |
| next_layer = packet[MACsec].payload |
| del prev_layer.payload |
| if prev_layer.name == Ether().name: |
| return Ether(raw(prev_layer/next_layer)) |
| return prev_layer/next_layer |
| |
| def encrypt(self, orig_pkt, assoclen=None): |
| """encrypt a MACsec frame for this Secure Association""" |
| hdr = copy.deepcopy(orig_pkt) |
| del hdr[MACsec].payload |
| del hdr[MACsec].type |
| pktlen = len(orig_pkt) |
| if self.send_sci: |
| hdrlen = NOSCI_LEN + SCI_LEN |
| else: |
| hdrlen = NOSCI_LEN |
| if assoclen is None or not self.do_encrypt: |
| if self.do_encrypt: |
| assoclen = hdrlen |
| else: |
| assoclen = pktlen |
| iv = self.make_iv(orig_pkt) |
| assoc, pt, _ = MACsecSA.split_pkt(orig_pkt, assoclen) |
| encryptor = Cipher( |
| algorithms.AES(self.key), |
| modes.GCM(iv), |
| backend=default_backend() |
| ).encryptor() |
| encryptor.authenticate_additional_data(assoc) |
| ct = encryptor.update(pt) + encryptor.finalize() |
| hdr[MACsec].payload = Raw(assoc[hdrlen:assoclen] + ct + encryptor.tag) |
| return hdr |
| |
| def decrypt(self, orig_pkt, assoclen=None): |
| """decrypt a MACsec frame for this Secure Association""" |
| hdr = copy.deepcopy(orig_pkt) |
| del hdr[MACsec].payload |
| pktlen = len(orig_pkt) |
| if self.send_sci: |
| hdrlen = NOSCI_LEN + SCI_LEN |
| else: |
| hdrlen = NOSCI_LEN |
| if assoclen is None or not self.do_encrypt: |
| if self.do_encrypt: |
| assoclen = hdrlen |
| else: |
| assoclen = pktlen - self.icvlen |
| iv = self.make_iv(hdr) |
| assoc, ct, icv = MACsecSA.split_pkt(orig_pkt, assoclen, self.icvlen) |
| decryptor = Cipher( |
| algorithms.AES(self.key), |
| modes.GCM(iv, icv), |
| backend=default_backend() |
| ).decryptor() |
| decryptor.authenticate_additional_data(assoc) |
| pt = assoc[hdrlen:assoclen] |
| pt += decryptor.update(ct) |
| pt += decryptor.finalize() |
| hdr[MACsec].type = struct.unpack('!H', pt[0:2])[0] |
| hdr[MACsec].payload = Raw(pt[2:]) |
| return hdr |
| |
| |
| class MACsec(Packet): |
| """representation of one MACsec frame""" |
| name = '802.1AE' |
| fields_desc = [BitField('Ver', 0, 1), |
| BitField('ES', 0, 1), |
| BitField('SC', 0, 1), |
| BitField('SCB', 0, 1), |
| BitField('E', 0, 1), |
| BitField('C', 0, 1), |
| BitField('an', 0, 2), |
| BitField('reserved', 0, 2), |
| BitField('shortlen', 0, 6), |
| IntField("pn", 1), |
| ConditionalField(PacketField("sci", None, MACsecSCI), lambda pkt: pkt.SC), |
| ConditionalField(XShortEnumField("type", None, ETHER_TYPES), |
| lambda pkt: pkt.type is not None)] |
| |
| def mysummary(self): |
| summary = self.sprintf("an=%MACsec.an%, pn=%MACsec.pn%") |
| if self.SC: |
| summary += self.sprintf(", sci=%MACsec.sci%") |
| if self.type is not None: |
| summary += self.sprintf(", %MACsec.type%") |
| return summary |
| |
| |
| bind_layers(MACsec, IP, type=ETH_P_IP) |
| bind_layers(MACsec, IPv6, type=ETH_P_IPV6) |
| |
| bind_layers( Dot1AD, MACsec, type=ETH_P_MACSEC) |
| bind_layers( Dot1Q, MACsec, type=ETH_P_MACSEC) |
| bind_layers( Ether, MACsec, type=ETH_P_MACSEC) |
