macsec.py

#!/usr/bin/env python3

#    _____ ___ ___ ___ ___ ___   ___ _ _
#   |     | .'|  _|_ -| -_|  _|_| . | | |
#   |_|_|_|__,|___|___|___|___|_|  _|_  |
#                               |_| |___|
#
# Author:  Nathaniel Williams.
# License: MIT.
# Purpose: Provides functions for parsing MACsec frames in PCAP files.

# === Imports. === #
import dpkt                     # Reading and writing PCAP files.
import sys                      # Writing bytes to stdout like a file.
from Crypto.Cipher import AES   # Encryption and decryption with AES.

def normalize_mac_address(mac_address: str):
    """Normalize a hexadecimal string MAC address by stripping whitespace,
    converting to lowercase, and removing colons.
    """
    return mac_address.strip().lower().replace(":", "")

class MACsecPeerTXSAK:
    """Parameters for a MACsec peer's TX SAK."""
    def __init__(self, key: str, port_identifier=None):
        """Stores the parameters for a MACsec peer's TX SAK.

        key                  - The key as a hexadecimal string.
        port_identifier=None - The port identifier as an integer. This is
                               optional if the MACsecFrame object being
                               decrypted had the SCI included. Otherwise, this
                               must be set. It's probably a good idea to have
                               this explicitly set just in case.

        Returns a new MACsecPeerTXSAK object.
        """
        self.key = bytes.fromhex(key.strip())
        if port_identifier == None: self.port_identifier = None
        else:
            self.port_identifier = port_identifier.to_bytes(2, byteorder="big")

class MACsecPeer:
    """Parameters for a MACsec peer."""
    def __init__(self, mac_address: str):
        """Stores the parameters of a MACsec peer.

        mac_address - The MAC address of the peer as a hexadecimal string.

        Returns a new MACsecPeer object.
        """
        self.mac_address = bytes.fromhex(normalize_mac_address(mac_address))
        self.tx_saks = []

    def add_tx_sak(self, key: str, port_identifier=None):
        """Add a TX SAK used by this peer.

        tx_sak - The 128-bit AES key used by the peer's TX SA (TX SAK).

        Returns nothing.
        """
        self.tx_saks.append(MACsecPeerTXSAK(key, port_identifier))

class MACsecFrame:
    """Deserialized MACsec frame."""
    MACSEC_ETHERTYPE_BYTES = bytearray(b"\x88\xE5")
    MACSEC_ETHERTYPE_INT = int.from_bytes(MACSEC_ETHERTYPE_BYTES, "big")

    def __init__(self, destination: bytes, source: bytes, data: bytes):
        """Deserialize a MACsec frame.

        destination - The destination MAC address of the frame.
        source      - The source MAC address of the frame.
        data        - The rest of the frame, including the SecTag, data, and
                      ICV.

        Returns a new MACsecFrame object.
        """
        self.destination = destination
        self.source = source
        self.tci_an = data[0].to_bytes(1, "big")
        self.includes_sci = bool(self.tci_an[0] & 0x08)
        self.short_length = data[1].to_bytes(1, "big")
        self.packet_number = data[2:6]
        if self.includes_sci:
            self.sci = data[6:12]
            self.port_identifier = data[12:14]
            self.data = data[14:-16]
        else:
            # If the frame doesn't include the SCI, we can just use the source
            # MAC address. However, we wouldn't know the port identifier. This
            # would have to be provided by the user. Otherwise, the frame can't
            # be authenticated or decrypted.
            self.sci = self.source
            self.port_identifier = bytearray(b"")
            self.data = data[6:-16]
        self.icv = data[-16:]

    @staticmethod
    def from_ethernet(frame: dpkt.ethernet.Ethernet):
        """Deserialize a MACsec frame from a dpkt.ethernet.Ethernet object.

        frame - The dpkt.ethernet.Ethernet object.

        Returns a new MACsecFrame object.
        """
        return MACsecFrame(frame.dst, frame.src, frame.data)

    def get_nonce_and_ad(self, peer_tx_sak=None):
        """Get the nonce and Associated Data (AD) of this frame.

        peer_tx_sak=None - The MACsecPeerTXSAK object to get the port
                           identifier from. This is optional if the frame
                           includes the SCI already. If not, then it must be
                           supplied through this.

        Returns the tuple (nonce, ad). Otherwise, returns None.
        """
        # If the SCI wasn't included with the frame, that means the port
        # identifier wasn't either. We have to get the port identifier from the
        # peers CSV file (provided by the user). Otherwise, we can't decrypt
        # the frame.
        port_identifier = self.port_identifier
        if not self.includes_sci:
            if not peer_tx_sak or peer_tx_sak.port_identifier == None: return None
            port_identifier = peer_tx_sak.port_identifier
        nonce = self.sci + port_identifier + self.packet_number
        ad = self.destination + self.source + self.MACSEC_ETHERTYPE_BYTES + \
            self.tci_an + self.short_length + self.packet_number + \
            self.sci + port_identifier
        return (nonce, ad)

    def decrypt(self, peer: MACsecPeer):
        """Decrypt the data of this MACsec frame using the parameters of the
        provided peer (the MAC address and TX SAK).

        peer - The peer whose parameters will be used for decryption.

        Returns a bytesarray of the decrypted data if successful. Otherwise,
        returns None.

        Source: https://stackoverflow.com/questions/61937514/decrypt-macsec-frame-python-aes-gcm
        """
        data = None
        for tx_sak in peer.tx_saks:
            try:
                nonce, ad = self.get_nonce_and_ad(peer_tx_sak=tx_sak)
                cipher = AES.new(tx_sak.key, AES.MODE_GCM, nonce=nonce)
                cipher.update(ad)
                data = cipher.decrypt_and_verify(self.data, self.icv)
                break
            except ValueError:
                print(f"Couldn't decrypt MACsec frame with ICV {self.icv}", file=sys.stderr)
                continue
        return data

    def decrypt_to_ethernet(self, peer: MACsecPeer):
        """Decrypt the data of this MACsec frame using the parameters of the
        provided peer (the MAC address and TX SAKs) and then create a new
        dpkt.ethernet.Ethernet object with the decrpyted data.

        peer - The peer whose parameters will be used for decryption.

        Returns a new dpkt.ethernet.Ethernet object if successful. Otherwise,
        returns None.
        """
        if (data := self.decrypt(peer)) == None: return None
        return dpkt.ethernet.Ethernet(self.destination + self.source + data)

    def __str__(self):
        """Returns the MACsecFrame object as a printable string."""
        str_ = ""
        for attribute, value in vars(self).items():
            str_ += attribute + ": "
            if isinstance(value, bytes): str_ += value.hex()
            else: str_ += value
            str_ += "\n"
        return str_

# === Reading. === #
def read_pcap_file(file_path: str):
    """Reads a PCAP file using dpkt.pcap.Reader.

    file_path - The path to the .pcap file being read.

    Returns a dictionary of string timestamps mapped to dpkt.ethernet.Ethernet
    objects.
    """
    frames = {}
    with open(file_path, "rb") as pcap_file:
        pcap_reader = dpkt.pcap.Reader(pcap_file)
        for timestamp, buffer in pcap_reader:
            frames[str(timestamp)] = dpkt.ethernet.Ethernet(buffer)
    return frames

def read_peers_file(file_path: str):
    """Reads a CSV file containing peer configuration information.

    file_path - The path to the .csv file being read.

    Returns a dictionary of hexadecimal string MAC addresses mapped to
    MACsecPeer objects.
    """
    peers = {}
    with open(file_path, "r") as peer_file:
        for mac_address_text, key_text, port_identifier_text in [line.strip().split(",") for line in peer_file.readlines()]:
            mac_address = normalize_mac_address(mac_address_text)
            key = key_text.strip()
            port_identifier = None
            if port_identifier_text != "":
                port_identifier = int(port_identifier_text)
            if (existing_peer := peers.get(mac_address)) == None:
                peers[mac_address] = MACsecPeer(mac_address)
            peers[mac_address].add_tx_sak(key, port_identifier=port_identifier)
    return peers

# === Parsing. === #
def parse_frames(frames: dict, peers: dict, output_all=False):
    """Parse frames read from a PCAP file.

    frames           - A dictionary of timestamp strings mapped to
                       dpkt.ethernet.Ethernet frames to be written.
    peers            - A dictionary of hexadecimal string MAC addresses mapped
                       to MACsecPeer objects.
    output_all=False - Optionally include non-MACsec frames in the output.

    Returns a dictionary of timestamp strings mapped to dpkt.ethernet.Ethernet
    objects, but with MACsec frames replaced with their decrypted versions (if
    decryption was successful).
    """
    output_frames = {}
    for timestamp, frame in frames.items():
        if frame.type == MACsecFrame.MACSEC_ETHERTYPE_INT:
            frame = MACsecFrame.from_ethernet(frame)
            frame_source = frame.source.hex().lower()
            if (peer := peers.get(frame_source)) == None:
                print(f"No peer configuration was found for MAC address {frame_source}.", file=sys.stderr)
                continue
            if (decrypted_frame := frame.decrypt_to_ethernet(peer)) == None:
                print(f"No valid key was found to decrypt this frame from the peer {frame_source}.", file=sys.stderr)
                continue
            output_frames[timestamp] = decrypted_frame
        elif output_all: output_frames[timestamp] = frame
    return output_frames

# === Writing. === #
def write_pcap_file(file_path: str, frames: dict, pcap_file=None):
    """Write a PCAP file using dpkt.pcap.Writer. This will erase all data if a
    file exists at the target file path.

    file_path      - The path to the .pcap file being written.
    frames         - A dictionary of string timestamps mapped to
                     dpkt.ethernet.Ethernet frames to be written.
    pcap_file=None - Optionally set a different file-like object to write to.

    Returns nothing.
    """
    new_file_object = False
    if pcap_file == None:
        pcap_file = open(file_path, "wb")
        new_file_object = True
    pcap_writer = dpkt.pcap.Writer(pcap_file)
    for timestamp, frame in frames.items(): pcap_writer.writepkt(frame, ts=float(timestamp))
    if new_file_object: pcap_file.close()

def write_pcap_stdout(frames: dict):
    """Write a PCAP file to stdout using dpkt.pcap.Writer.

    frames - A dictionary of string timestamps mapped to dpkt.ethernet.Ethernet
             frames to be written.

    Returns nothing.
    """
    write_pcap_file("", frames, sys.stdout.buffer)

def write_hex_dump_file(file_path: str, frames: dict, hex_dump_file=None):
    """Write a hexdump file.

    file_path         - The path to the hex dump file being written.
    frames            - A dictionary of string timestamps mapped to
                        dpkt.ethernet.Ethernet frames to be written.
    hexdump_file=None - Optionally set a different file-like object to write to.

    Returns nothing.
    """
    new_file_object = False
    if hex_dump_file == None:
        hex_dump_file = open(file_path, "w")
        new_file_object = True
    for timestamp, frame in frames.items():
        frame_bytes = frame.pack()
        for offset in range(0, len(frame_bytes), 16):
            chunk = frame_bytes[offset:offset+16]
            hex_bytes = " ".join(f"{b:02x}" for b in chunk)
            hex_dump_file.write(f"{offset:04x}   {hex_bytes}\n")
        hex_dump_file.write("\n")
    if new_file_object: hex_dump_file.close()

def write_hex_dump_stdout(frames: dict):
    """Write a hexdump file to stdout.

    frames - A dictionary of string timestamps mapped to dpkt.ethernet.Ethernet
             frames to be written.

    Returns nothing.
    """
    write_hex_dump_file("", frames, hex_dump_file=sys.stdout)

# === Running directly with Python. === #
if __name__ == "__main__":

    from argparse import ArgumentParser

    parser = ArgumentParser(
        description="Decrypt/authenticate MACsec frames from a PCAP file.",
        epilog="example:\n python macsec.py --all macsec.pcap peers.csv - | wireshark -k -i -"
    )
    parser.add_argument("-a","--all",    help="include all non-MACsec frames in the output",                   action="store_true")
    parser.add_argument("-d","--hex-dump",help="output frames as a hexdump that can be imported into Wireshark",action="store_true")
    parser.add_argument("input",    help="the PCAP file containing the encrypted MAACsec frames")
    parser.add_argument("peers_csv",     help="the CSV file containing the peer MAC addresses and their respective TX SAKs")
    parser.add_argument("output",        help="where to output the parsed frames (a file path or \"-\" for stdout)")
    args = parser.parse_args()

    # Read frames from the input PCAP file.
    try:
        input_frames = read_pcap_file(args.input)
    except FileNotFoundError:
        print(f"The file '{args.input}' couldn't be found.", file=sys.stderr)
        exit(1)

    # Parse CSV file containing peer configuration (MAC addresses and TX SAKs).
    try:
        peers = read_peers_file(args.peers_csv)
    except FileNotFoundError:
        print(f"The file '{args.peers_csv}' couldn't be found.", file=sys.stderr)
        exit(1)
    except ValueError:
        print("There must be three columns (exactly two commas): <MAC address>,<TX SAK>,[Port Identifier]", file=sys.stderr)
        exit(1)

    # Parse input frames.
    output_frames = parse_frames(input_frames, peers, output_all=args.all)

    # Write the decrypted MACsec frames.
    if args.output == "-":  # To stdout
        if args.hex_dump:
            write_hex_dump_stdout(output_frames)
        else:
            write_pcap_stdout(output_frames)
    else:                   # To a file
        if args.hex_dump:
            write_hex_dump_file(args.output, output_frames)
        else:
            write_pcap_file(args.output, output_frames)