A standalone implementation of PCIe VPD resource traversal for SN extraction

get-netdev-vpd-serials.py

#!/usr/bin/env python3

import libvirt
import os

from lxml import etree


def get_vpd_serial(vpd_raw):
    # See the "Vital Product Data (VPD)" section of PCI Local Bus 2.x
    # specification or the same section in PCIe base spec 4.x or above.
    res_addr = 0
    vpd_len = len(vpd_raw)
    if vpd_len < 5:
        # Does not seem like we have a valid VPD since at least
        # the Identifier String large resource tag (> 4 bytes)
        # and the End Tag small resource tag (1 byte) need to be present.
        return None
    last_addr = vpd_len + 1
    while res_addr < last_addr:
        # 0x80 = 0b10000000 - the large resource data type flag.
        if (vpd_raw[res_addr] & 0x80):
            # Handle a large resource.
            # 1xxxxxxx, where xxxxxxx are the resource bits
            res_tag = vpd_raw[res_addr]
            # 2 length bytes comprise the total data length for large records.
            res_data_len = vpd_raw[res_addr + 1] + (vpd_raw[res_addr + 2] << 8)
            res_data_addr = res_addr + 3
        else:
            # Handle a small resource.
            # 0xxxxyyy & 00000111, where xxxx - resource data type bits,
            # yyy - length bits.
            res_data_len = vpd_raw[res_addr] & 0b00000111
            # 0xxxxyyy >> 3 = 0000xxxx
            res_tag = vpd_raw[res_addr] >> 3
            res_data_addr = res_addr + 1

        # If we encounter the End Tag - that's the end of VPD.
        if res_tag == 0x0F:
            return None

        if res_data_addr + 1 > vpd_len:
            # Went past the vpd_len - looks like VPD data is invalid.
            return None

        next_res_addr = res_data_addr + res_data_len
        # We are looking for the VPD-R tag which is a list of keywords.
        # 0x10 or 0x80 = 0x90
        if res_tag != 0x90:
            # Get the address of the next resource record.
            res_addr = next_res_addr
            continue

        kw_addr = res_data_addr
        SN_KW = 'SN'
        while kw_addr < next_res_addr:
            # Keywords are 2 ASCII bytes per the spec.
            # TODO: handle encoding errors here.
            kw = vpd_raw[kw_addr:kw_addr + 2].decode('ascii')
            kw_len = vpd_raw[kw_addr + 2]
            # Skip the 2 keyword bytes and a length byte.
            kw_data_addr = kw_addr + 2 + 1
            if kw == SN_KW:
                sn = vpd_raw[kw_data_addr:kw_data_addr+kw_len].strip()
                return sn.decode('ascii')
            else:
                kw_addr = kw_data_addr + kw_len
                continue
        res_addr = next_res_addr


def print_netdev_serials():
    conn = libvirt.open('qemu:///system')
    dev_flags = (libvirt.VIR_CONNECT_LIST_NODE_DEVICES_CAP_NET |
                 libvirt.VIR_CONNECT_LIST_NODE_DEVICES_CAP_PCI_DEV)
    devices = {dev.name(): dev for dev in conn.listAllDevices(flags=dev_flags)}
    net_devs = [dev for dev in devices.values() if "net" in dev.listCaps()]

    for dev in net_devs:
        vpd_path = os.path.join(etree.fromstring(
            dev.XMLDesc()).find('path').text, 'device/vpd')
        if os.path.exists(vpd_path):
            with open(vpd_path, 'rb') as vpd_fd:
                vpd_raw = vpd_fd.read()
                print(f'{dev.name()}: {get_vpd_serial(vpd_raw)}')


if __name__ == '__main__':
    print_netdev_serials()