Utility script for decrypting CCMP in a pcapng capture

decrypt_ccmp.py

#!/usr/bin/env python3

import argparse
import binascii

from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from scapy.all import *


# HEY! Please don't use this for any production/security purpose!
def decrypt_ccm(aes, nonce, data, ad=b'', m=8, l=2):
    assert 1+len(nonce)+l == 16

    # Decryption:
    iv = bytes([l-1]) + nonce + b'\0'*l
    ciph = Cipher(aes, modes.CTR(iv)).decryptor()
    mac = ciph.update(data[-m:] + b'\0'*(16-m))[:m]
    msg = ciph.update(data[:-m]) + ciph.finalize()

    # Now authentication:
    ciph = Cipher(aes, modes.CBC(b'\0'*16)).encryptor()

    f  = (l-1)
    f |= ((m-2)//2) << 3
    if ad:
        if len(ad) < (1<<16) - (1<<8):
            la = len(ad).to_bytes(2, 'big')
        elif len(ad) < (1<<32):
            la = b'\xff\xfe' + len(ad).to_bytes(4, 'big')
        else:
            la = b'\xff\xff' + len(ad).to_bytes(8, 'big')
        f |= 64 # Adata bit
    else:
        la = b''

    lm = len(msg).to_bytes(l, 'big')
    b0 = bytes([f]) + nonce + lm
    assert len(b0) == 16
    ciph.update(b0)

    amsg  = la + ad
    amsg += b'\0'*((16-len(amsg))&15)
    amsg += msg
    amsg += b'\0'*((16-len(amsg))&15)

    if not ciph.update(amsg)[-16:].startswith(mac):
        return None
    ciph.finalize()

    return msg


def compute_ad(d11):
    hdr_len = 22
    if d11.subtype == 8:
        hdr_len += 2

    ad = bytearray(hdr_len)

    # Masked FC
    ad[0] = d11.proto | (d11.type << 2) | ((d11.subtype&0x8) << 4)
    ad[1] = 0x40 | int(d11.FCfield&0x87)

    ad[ 2: 8] = binascii.unhexlify(d11.addr1.replace(':',''))
    ad[ 8:14] = binascii.unhexlify(d11.addr2.replace(':',''))
    ad[14:20] = binascii.unhexlify(d11.addr3.replace(':',''))

    ad[20] = d11.SC & 0xF
    ad[21] = 0

    if Dot11QoS in d11:
        ad[22] = d11[Dot11QoS].TID

    return bytes(ad)


def compute_nonce(d11):
    f = 0
    if Dot11QoS in d11:
        f = d11[Dot11QoS].TID
    nonce = bytes([f]) + binascii.unhexlify(d11.addr2.replace(':',''))
    ccmp = d11[Dot11CCMP]
    nonce += bytes([ccmp.PN5, ccmp.PN4, ccmp.PN3,
                    ccmp.PN2, ccmp.PN1, ccmp.PN0])

    return nonce


def filter_pkt(pkt, aeses):
    if Dot11CCMP not in pkt:
        return pkt

    ad = compute_ad(pkt)
    nonce = compute_nonce(pkt)

    for i, aes in enumerate(aeses):
        raw_data = decrypt_ccm(aes, nonce, pkt[Dot11CCMP].data, ad)
        if raw_data:
            pkt = pkt.copy()
            pkt.len = None
            if Dot11FCS in pkt:
                pkt[Dot11FCS].fcs = None

            del pkt[Dot11CCMP]
            pkt = pkt/raw_data
            pkt.FCfield.protected = False

            break

    else:
        return pkt

    if i:
        # Let's move this AES object so we try it first next time
        aeses.insert(0, aeses.pop(i))

    return pkt


def main():
    key = lambda x: algorithms.AES(binascii.unhexlify(x))

    parser = argparse.ArgumentParser()
    parser.add_argument('input')
    parser.add_argument('output')
    parser.add_argument('--key', '-K', action='append', type=key)

    args = parser.parse_args()
    aeses = list(args.key)

    with PcapNgReader(args.input) as i:
        with PcapWriter(args.output) as o:
            for pkt in i:
                if pkt.time > (1<<32):
                    pkt.time /= 1000 # Workaround for reader bug
                o.write(filter_pkt(pkt, aeses))


if __name__ == '__main__':
    main()