felhom.eu/documentation/audits/storage-identity-2026-10-15/tools/physdisk_probe.py

#!/usr/bin/env python3
"""physdisk_probe.py — READ-ONLY spike for 16-storage-physical-identity.md (2026-10-15).

For every storage place on a Proxmox host (host root, PVE storages, guest mount points, /mnt/* mounts, extra paths
or devices given as arguments) follow the block chain down to WHOLE physical disks:
  partition -> parent disk (/sys/class/block/<p>/partition, the parent sysfs dir = lsblk PKNAME)
  dm / lvm / md -> /sys/class/block/<dm>/slaves/* (recursive; a thin pool reaches its tdata/tmeta PVs)
  loop -> /sys/block/loopN/loop/backing_file -> the filesystem holding that file -> recurse
and read each disk's identity from udev (ID_WWN, ID_SERIAL_SHORT, ID_BUS, ID_VENDOR, ID_MODEL) + sysfs.
identity = proven | unknown, with the reason. Writes nothing; runs only lsblk/findmnt/udevadm/pct/pvesm reads.
"""
import json, os, subprocess, sys

def sh(*a):
    try: return subprocess.run(a, capture_output=True, text=True, timeout=20).stdout.strip()
    except Exception: return ''

VIRT = sh('systemd-detect-virt') or 'unknown'
VIRTUAL_VENDORS = ('QEMU', 'VBOX', 'VMware', 'Msft', 'Red Hat', 'Xen')
BLOCK = '/sys/class/block'

def kname(dev):
    return os.path.basename(os.path.realpath(dev))

def chain(name, via, depth=0):
    """-> list of (disk_name, via-string)"""
    if depth > 12: return [(name, via + ' !depth')]
    p = os.path.join(BLOCK, name)
    if not os.path.exists(p): return [(name, via + ' !nosysfs')]
    if os.path.exists(os.path.join(p, 'partition')):
        parent = os.path.basename(os.path.dirname(os.path.realpath(p)))
        return chain(parent, via + f' -> part {name}', depth + 1)
    if name.startswith('loop'):
        bf = ''
        try: bf = open(f'/sys/block/{name}/loop/backing_file').read().strip()
        except Exception: pass
        if bf:
            src = sh('findmnt', '-n', '-o', 'SOURCE', '-T', bf)
            if src.startswith('/dev/'):
                return chain(kname(src), via + f' -> loop[{bf}] on {src}', depth + 1)
        return [(name, via + ' -> loop(no backing)')]
    sl = os.path.join(p, 'slaves')
    if os.path.isdir(sl) and os.listdir(sl):
        out = []
        dmname = ''
        try: dmname = open(os.path.join(p, 'dm', 'name')).read().strip()
        except Exception: pass
        for s in sorted(os.listdir(sl)):
            out += chain(s, via + f' -> {name}{"(" + dmname + ")" if dmname else ""}', depth + 1)
        return out
    return [(name, via)]

def disk_identity(d):
    props = {}
    for line in sh('udevadm', 'info', '--query=property', f'/dev/{d}').splitlines():
        k, _, v = line.partition('=')
        props[k] = v
    def rd(f):
        try: return open(f'/sys/block/{d}/{f}').read().strip()
        except Exception: return ''
    info = {
        'disk': d,
        'model': props.get('ID_MODEL', rd('device/model')),
        'vendor': props.get('ID_VENDOR', rd('device/vendor')),
        'serial': props.get('ID_SERIAL_SHORT', ''),
        'wwn': props.get('ID_WWN', rd('wwid') or rd('device/wwid')),
        'bus': props.get('ID_BUS', ''),
        'usb_driver': props.get('ID_USB_DRIVER', ''),
        'path': props.get('ID_PATH', ''),
        'size_gb': round(int(rd('size') or 0) * 512 / 1e9, 1),
        'rotational': rd('queue/rotational'),
    }
    why = []
    vend = (info['vendor'] + ' ' + info['model']).upper()
    if VIRT not in ('none', ''): why.append(f'inside a VM ({VIRT}): a virtual disk says nothing about the host disk')
    if any(v.upper() in vend for v in VIRTUAL_VENDORS): why.append('virtual disk vendor/model')
    if d.startswith('vd') and not info['serial']: why.append('virtio disk without a serial')
    if not (info['serial'] or info['wwn']): why.append('no serial and no wwn')
    if info['bus'] == 'usb' and info['serial'] and len(set(info['serial'])) <= 2: why.append('USB bridge serial looks fake')
    info['id'] = ('wwn:' + info['wwn']) if info['wwn'] else (('serial:' + info['serial']) if info['serial'] else '')
    info['identity'] = 'unknown' if why else 'proven'
    info['why_unknown'] = why
    return info

def place(label, target):
    """target: a /dev path, a mounted path, or 'net:' prefix"""
    rec = {'place': label, 'target': target}
    src = target
    if not target.startswith('/dev/'):
        src = sh('findmnt', '-n', '-o', 'SOURCE', '-T', target)
        rec['fstype'] = sh('findmnt', '-n', '-o', 'FSTYPE', '-T', target)
        rec['mountpoint'] = sh('findmnt', '-n', '-o', 'TARGET', '-T', target)
    rec['source'] = src
    if rec.get('fstype') in ('cifs', 'nfs', 'nfs4', 'smb3', 'fuse.sshfs') or (src and not src.startswith('/dev/')):
        rec['physical_disks'] = []
        rec['identity'] = 'unknown'
        rec['why_unknown'] = [f'network or non-block source ({rec.get("fstype")}: {src})']
        return rec
    src = src.split('[')[0]
    ch = chain(kname(src), kname(src))
    disks = []
    seen = set()
    for d, via in ch:
        if d in seen: continue
        seen.add(d)
        di = disk_identity(d); di['via'] = via
        disks.append(di)
    rec['physical_disks'] = [d['id'] or ('dev:' + d['disk']) for d in disks]
    rec['disks'] = disks
    rec['identity'] = 'proven' if disks and all(d['identity'] == 'proven' for d in disks) else 'unknown'
    rec['why_unknown'] = sorted({w for d in disks for w in d['why_unknown']})
    return rec

def places():
    out = [('host root /', '/')]
    if sh('which', 'pvesm'):
        for line in sh('pvesm', 'status').splitlines()[1:]:
            f = line.split()
            if len(f) < 2 or f[2] != 'active': continue
            name, typ = f[0], f[1]
            cfg = sh('pvesh', 'get', f'/storage/{name}', '--output-format', 'json')
            try: c = json.loads(cfg)
            except Exception: c = {}
            if typ == 'dir' and c.get('path'): out.append((f'pve storage {name} (dir)', c['path']))
            elif typ == 'lvmthin': out.append((f'pve storage {name} (lvmthin {c.get("vgname")}/{c.get("thinpool")})', '/dev/mapper/' + c.get('vgname','').replace('-','--') + '-' + c.get('thinpool','').replace('-','--') + '-tpool'))
            elif typ == 'lvm': out.append((f'pve storage {name} (lvm)', f'/dev/{c.get("vgname")}'))
            elif typ in ('nfs', 'cifs') and c.get('path'): out.append((f'pve storage {name} ({typ})', c['path']))
        for vmid in sh('pct', 'list').split('\n')[1:]:
            v = vmid.split()[0] if vmid.split() else ''
            if not v: continue
            for line in sh('pct', 'config', v).splitlines():
                k, _, val = line.partition(': ')
                if k == 'rootfs' or k.startswith('mp'):
                    vol = val.split(',')[0]
                    mp = [x.split('=')[1] for x in val.split(',') if x.startswith('mp=')]
                    lab = f'guest {v} {k} {mp[0] if mp else "/"}'
                    if vol.startswith('/'):
                        out.append((lab + ' (host bind)', vol))
                    else:
                        pth = sh('pvesm', 'path', vol)
                        out.append((lab + f' ({vol})', pth if pth else vol))
    for m in sh('findmnt', '-rn', '-o', 'TARGET').splitlines():
        if m.startswith('/mnt/') and (m.count('/') == 2 or m.startswith('/mnt/felhom-drives/')): out.append((f'host mount {m}', m))
    return out

if __name__ == '__main__':
    extra = sys.argv[1:]
    todo = places() + [(f'arg {a}', a) for a in extra]
    res = {'host': sh('hostname'), 'virt': VIRT, 'places': []}
    for lab, t in todo:
        try: res['places'].append(place(lab, t))
        except Exception as e: res['places'].append({'place': lab, 'target': t, 'error': str(e)})
    print(json.dumps(res, indent=1))