Zigbee .psd to .pcap conversion

psd2pcap.py

#!/usr/bin/env python3

# Converts SmartRF Zigbee Sniffer .psd format into .pcap using Scapy
# Can be used with Texas Instruments, Inc. CC2531 Dongle
#
# Author: k3a.me
# License: MIT
#
# TIP: On Linux, you can use https://github.com/homewsn/whsniff
# to directly communicate with the dongle and produce pcap

import sys
import struct
import scapy.data
from scapy.utils import PcapWriter
from scapy.layers.dot15d4 import Dot15d4FCS
from scapy.layers.dot15d4 import conf

TIME_DIV = 32
PAYLOAD_DATA_SIZE = 136 # from Help -> About he PSD format

FLAG_FCS_INCLUDED_IN_LENGTH = 1<<0
FLAG_CORRELATION_USED = 1<<1
FLAG_INCOMPLETE_PACKET = 1<<2
FLAG_BUFFER_OVERFLOW = 1<<3
FLAG_GENERIC_PROTO = 1<<4

STATUS_CRC_OK = 1<<7

if len(sys.argv) < 3:
  print(f"Usage {sys.argv[0]} <input.psd> <output.pcap>", file=sys.stderr)
  sys.exit(1)

in_path = sys.argv[1]
out_path = sys.argv[2]

wr = PcapWriter(out_path) # scapy/data.py for DLT_ definitions

conf.dot15d4_protocol = 'zigbee'

with open(in_path, 'rb') as f:
  while True:
    data = f.read(15)
    if not data:
      break
    
    (flags, nr, timestamp, slen) = struct.unpack('<BIQH', data)
    timestamp //= TIME_DIV
    
    flags_str = "["
    if flags & FLAG_FCS_INCLUDED_IN_LENGTH:
      flags_str += "F"
    if flags & FLAG_CORRELATION_USED:
      flags_str += "C"
    if flags & FLAG_INCOMPLETE_PACKET:
      flags_str += "I"
    if flags & FLAG_BUFFER_OVERFLOW:
      flags_str += "O"
    if flags & FLAG_GENERIC_PROTO:
      flags_str += "G"
    flags_str += "]"
    
    data = f.read(PAYLOAD_DATA_SIZE)
    if not data:
      print(f"Unexpected EOF while reading {PAYLOAD_DATA_SIZE} bytes from {in_path}", file=sys.stderr)
      sys.exit(1)

    data_len = data[0]
    data = data[1:slen]
    dlen = len(data)
    
    # per SmartRF sniffer user manual, FCS is replaced by two status bytes
    status_rssi = 0
    status_flags = 0
    status_str = "["
    if dlen >= 2:
      status_rssi = data[dlen-2]
      status_flags = data[dlen-1]
      # compute correct FCS now
      data = data[:-2] + Dot15d4FCS.compute_fcs(None, data[:dlen-2])
      if status_flags & STATUS_CRC_OK:
        status_str += "O"
    status_str += "]"
    
    print("info:", flags_str, flags, " num:", nr, " ts:", timestamp, " len:", data_len, " rssi & status:", status_rssi, status_str) 
    #print(data.hex())
    
    pkt = Dot15d4FCS(data)
    wr.write(pkt)

wr.close()