add macsec support

scapy/contrib/macsec.py

+# 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
+
+if conf.crypto_valid:
+    from cryptography.exceptions import InvalidTag
+    from cryptography.hazmat.backends import default_backend
+    from cryptography.hazmat.primitives import interfaces
+    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 type(sci) == type(int()) or type(sci) == type(long()):
+            self.sci = struct.pack('!Q', sci)
+        elif type(sci) == type(str()):
+            self.sci = sci
+        else:
+            raise TypeError("SCI must be either str 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 = str(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(str(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)

scapy/data.py

@@ -31,6 +31,7 @@ ETH_P_ALL = 3
 ETH_P_IP = 0x800
 ETH_P_ARP = 0x806
 ETH_P_IPV6 = 0x86dd
+ETH_P_MACSEC = 0x88e5
 
 # From net/if_arp.h
 ARPHDR_ETHER = 1

scapy/layers/l2.py

@@ -143,6 +143,7 @@ class ARPSourceMACField(SourceMACField):
 ### Layers
 
 ETHER_TYPES['802_AD'] = 0x88a8
+ETHER_TYPES['802_1AE'] = ETH_P_MACSEC
 
 class Ether(Packet):
     name = "Ethernet"
@@ -410,6 +411,7 @@ class Loopback(Packet):
 class Dot1AD(Dot1Q):
     name = '802_1AD'
 
+
 bind_layers( Dot3,          LLC,           )
 bind_layers( Ether,         LLC,           type=122)
 bind_layers( Ether,         LLC,           type=34928)