path: root/core/modules/hardware-stats/data/opt/openslx/hardware-stats/collect_hw_info_json.py

from dmiparser import DmiParser
from subprocess import PIPE
from os import listdir, path
import argparse
import json
import re
import requests
import shlex
import subprocess
import sys

__debug = False

# Run dmi command as subprocess and get the stdout
def run_subprocess(_cmd):
    global __debug
    proc = subprocess.run(_cmd, shell=True, stdout=PIPE, stderr=PIPE)
    stdout = proc.stdout
    stderr = proc.stderr

    if __debug:
        eprint(_cmd + ':')
        eprint()
        eprint('Errors:')
        eprint(stderr.decode())
        eprint()
    # stderr len instead of proc.returncode > 0 is used because some have returncode 2 but are still valid
    if len(stderr.decode()) > 0:
        eprint(_cmd + ' Errors:')
        eprint(stderr.decode())
        if proc.returncode != 0:
            eprint('Error Return Code: ' + str(proc.returncode))
            eprint()
            return False
        else:
            return stdout.decode()
    else:
        return stdout.decode() 

# Get and parse dmidecode using dmiparser
def get_dmidecode():
    _dmiraw = run_subprocess('dmidecode')
    _dmiparsed = ''
    if _dmiraw:
        # Parse dmidecode
        _dmiparsed = DmiParser(_dmiraw)
        return json.loads(str(_dmiparsed))
    else:
        return []

# Get the readlink -f output
def get_readlink(link):
    _readlink = run_subprocess('readlink -f ' + link)
    return _readlink

# Try parsing string to json
def parse_to_json(program_name, raw_string):
    parsed = {}
    if isinstance(raw_string, str):
        try:
            parsed = json.loads(raw_string)
        except ValueError as e:
            eprint(program_name + ' output couldn\'t be parsed')
            eprint(e)
            eprint('Output of ' + program_name + 'was:')
            eprint(raw_string)
            eprint()
    return parsed

# Get smartctl output in json format
def get_smartctl(disk_path, disk_name):
    disk_path_full = disk_path + disk_name
    output = run_subprocess('smartctl -x --json ' + disk_path_full)
    smartctl = parse_to_json('smartctl', output)
    return smartctl

# Get sfdisk info in json format
def get_sfdisk(disk_path, disk_name):
    output = run_subprocess('sfdisk --json ' + disk_path + disk_name)
    sfdisk = parse_to_json('sfdisk', output)
    return sfdisk

# Get lsblk info in json format
def get_lsblk(disk_path, disk_name):
    output = run_subprocess('lsblk --json -b --output-all ' + disk_path + disk_name)
    lsblk = parse_to_json('lsblk', output)
    return lsblk

# Get CD/DVD Information
def get_cdrom():
    cdromdir = '/proc/sys/dev/cdrom/'
    cdrom = []
    if path.exists(cdromdir):
        cdrom_raw = run_subprocess('cat ' + cdromdir + 'info')

        # Skip first two entries because of useless information and empty row
        for row in cdrom_raw.split('\n')[2:]:
            if row == '':
                continue
            # Split at one or more tabs
            values = re.split('\t+', row)
            key = values[0][:-1].replace('drive ', '').replace(' ', '_')
            for index, val in enumerate(values[1:]):
                if len(cdrom) < index + 1:
                    cdrom.append({ 'read': [], 'write': [], 'functions': [] })

                if 'Can_read_' in key:
                    if val == '1':
                        cdrom[index]['read'].append(key[9:])
                elif 'Can_write_' in key:
                    cdrom[index]['write'].append(key[10:])
                elif 'Can_' in key:
                    cdrom[index]['functions'].append(key[4:])
                else:
                    cdrom[index][key] = val
    return {cd['name']:cd for cd in cdrom}

# Get informations about the disks
def get_disk_info():
    diskdir = '/dev/disk/by-path/'
    disk_informations = {}
    dupcheck = {}
    cdrom = get_cdrom()

    # Get and filter all disks
    if path.exists(diskdir):
        disks = listdir(diskdir)
        filtered_disks = [i for i in disks if (not '-part' in i) and (not '-usb-' in i)]

        # Call all disk specific tools
        for d in filtered_disks:
            disk_path = diskdir + d
            devpath = get_readlink(disk_path).rstrip()
            # Sometimes there are multiple links to the same disk, e.g. named
            # pci-0000:00:1f.2-ata-1.0 and pci-0000:00:1f.2-ata-1
            if devpath in dupcheck:
                continue
            dupcheck[devpath] = 1
            disk_info = {}
            disk_info['readlink']   = devpath

            # Check if it's a cd/dvd
            if disk_info['readlink'].split('/')[-1] in cdrom.keys():
                disk_info['type'] = 'cdrom'
                disk_info['info'] = cdrom[disk_info['readlink'].split('/')[-1]]
            else:
                disk_info['type'] = 'drive'
            disk_info['smartctl']   = get_smartctl(diskdir, d)
            disk_info['lsblk']      = get_lsblk(diskdir, d)

            if disk_info['type'] != 'cdrom':
                disk_info['sfdisk']     = get_sfdisk(diskdir, d)

            disk_informations[d]    = disk_info
    return disk_informations

# Get and process "lspci -mn" output
def get_lspci():
    lspci = []
    lspci_raw = run_subprocess('lspci -mmn').split('\n')

    # Prepare addition of iommu group
    iommu_groups = {}
    iommu_raw = run_subprocess('find /sys/kernel/iommu_groups/*/devices/*')
    if iommu_raw:
        iommu_split = iommu_raw.split('\n')
        for iommu_path in iommu_split:
            if iommu_path == "":
                continue
            iommu = iommu_path.split('/')
            iommu_groups[iommu[6][5:]] = iommu[4]

    for line in lspci_raw:
        if len(line) <= 0: continue

        # Parsing shell like command parameters
        parse = shlex.split(line)
        lspci_parsed = {}
        arguments = []
        values = []
        for parameter in parse:
            # Split values from arguments
            if parameter.startswith('-'):
                arguments.append(parameter)
            else:
                values.append(parameter)

        # Prepare values positions are in order
        if len(values) >= 6:
            lspci_parsed['slot']                = values[0]

            # The busybox version of lspci has "Class <class>" instead of "<class>"
            lspci_parsed['class']               = values[1].replace("Class ", "")
            lspci_parsed['vendor']              = values[2]
            lspci_parsed['device']              = values[3]
            lspci_parsed['subsystem_vendor']    = values[4]
            lspci_parsed['subsystem']           = values[5]

            # Additional add iommu group
            if values[0] in iommu_groups:
                lspci_parsed['iommu_group'] = iommu_groups[values[0]]

        # Prepare arguments
        if len(arguments) > 0:
            for arg in arguments:
                if arg.startswith('-p'):
                    lspci_parsed['progif'] = arg[2:]
                elif arg.startswith('-r'):
                    lspci_parsed['rev'] = arg[2:]
                else: continue

        lspci.append(lspci_parsed)
    return lspci

# Get ip data in json format
def get_ip():
    result = []
    ip_raw = run_subprocess('ip --json addr show')
    if isinstance(ip_raw, str):
        result = json.loads(ip_raw)
    return result

def get_net_fallback():
    netdir = '/sys/class/net/'
    result = {}

    # Get MAC address and speed
    if path.exists(netdir):
        interfaces = run_subprocess('ls ' + netdir).split('\n')
        for interface in interfaces:
            # Skip local stuff
            if interface == 'lo' or interface == '':
                continue
            net = {}
            speed = run_subprocess('cat ' + netdir + interface + '/speed')
            if isinstance(speed, str) and speed.endswith('\n'):
                net['speed'] = speed[:-1]

            duplex = run_subprocess('cat ' + netdir + interface + '/duplex')
            if isinstance(duplex, str) and duplex.endswith('\n'):
                net['duplex'] = duplex[:-1]

            mac = run_subprocess('cat ' + netdir + interface + '/address')
            if isinstance(mac, str) and mac.endswith('\n'):
                net['mac'] = mac[:-1]
            result[interface] = net

    # Get IP address
    interfaces = run_subprocess('ip -o addr show | awk \'/inet/ {print $2, $3, $4}\'').split('\n')
    for interface in interfaces:
        if interface == '':
            continue
        interf = interface.split(' ')
        if interf[0] == 'lo':
            continue
        else:
            if interf[0] not in result:
                result[interf[0]] = {}
            if interf[1] == 'inet':
                result[interf[0]]['ipv4'] = interf[2]
            elif interf[1] == 'inet6':
                result[interf[0]]['ipv6'] = interf[2]
    return result

# Get and convert EDID data to hex
def get_edid():
    edid = {}
    display_paths = run_subprocess('ls /sys/class/drm/*/edid')
    if display_paths:
        display_paths = display_paths.split('\n')
        for dp in display_paths:
            if dp == '': continue
            edid_hex = open(dp, 'rb').read().hex()
            if len(edid_hex) > 0:
                # The path is always /sys/class/drm/[..]/edid, so cut the first 15 chars and the last 5 chars
                edid[dp[15:-5]] = edid_hex
    return edid

def get_lshw():
    result = []
    lshw_raw = run_subprocess('lshw -json')
    if isinstance(lshw_raw, str):
        result = json.loads(lshw_raw)
    return result

def get_uuid():
    uuid_path = '/sys/class/dmi/id/product_uuid'
    uuid = 'N/A'
    if path.exists(uuid_path):
        uuid_raw = run_subprocess('cat ' + uuid_path)
        # uuid_raw = False if no sudo permission:
        if uuid_raw:
            uuid = uuid_raw.rstrip()
    return uuid

def prepare_location(parent, bay, slot):
    location = {}
    if parent:
        location['parent']  = parent
    if bay:
        location['bay']     = int(bay)
    if slot:
        location['slot']    = int(slot)
    return location

def prepare_contacts(contact_list):
    contacts = []
    if contact_list == None:
        return contacts
    for contact in contact_list:
        contacts.append(contact[0])
    return contacts

def send_post(url, payload):
    # headers = { 'Content-type': 'application/json', 'Accept': 'text/plain' }
    # req = requests.post(url, json=payload, headers=headers)
    req = requests.post(url, json=payload)
    # Print the response
    print("POST-Request Response: \n")
    print(req.text)

def eprint(*args, **kwargs):
    print(*args, file=sys.stderr, **kwargs)

def main():
    global __debug

    # Create and parse arguments
    parser = argparse.ArgumentParser(description='Collects hardware data from different tools and returns it as json.')
    parser.add_argument('-d', '--debug',    action='store_true',    help='Prints all STDERR messages. (Non critical included)')
    parser.add_argument('-u', '--url',      action='append',        help='[multiple] If given, a post request with the generated JSON is sent to the given URLs')
    parser.add_argument('-uu', '--uuidurl', action='append',        help='[multiple] Same as -u but UUID in the url is replaced with the actual uuid of the client')
    parser.add_argument('-p', '--print',    action='store_true',    help='Prints the generated JSON')
    parser.add_argument('-l', '--location', action='store',         help='Room-/Rackname where the client is located')
    parser.add_argument('-s', '--slot',     action='store',         help='The slot number (int) where the client is located in the rack')
    parser.add_argument('-b', '--bay',      action='store',         help='The bay number (int) where the client is located in the slot (segment)')
    parser.add_argument('-c', '--contact',  action='append',        help='[multiple] The idoit_username of the person responsible for this machine', nargs=1)
    parser.add_argument('-n', '--name',     action='store',         help='Name of the client')
    parser.add_argument('-S', '--SERVER',   action='store_true',    help='Defines the type of the client to be a server')
    args = parser.parse_args()

    if args.debug:
        __debug = True

    # Run the tools
    _collecthw = {}
    _collecthw['version']   = 2.0
    _collecthw['dmidecode'] = get_dmidecode()

    # Includes smartctl, lsblk, readlink and sfdisk
    _collecthw['drives']    = get_disk_info()

    # Includes iommu group
    _collecthw['lspci']     = get_lspci()

    _collecthw['ip']        = get_ip()
    _collecthw['edid']      = get_edid()
    _collecthw['lshw']      = get_lshw()
    _collecthw['net']       = get_net_fallback()

    _collecthw['location']  = prepare_location(args.location, args.bay, args.slot)
    _collecthw['contacts']  = prepare_contacts(args.contact)

    if args.name:
        _collecthw['name'] = args.name

    if args.SERVER:
        _collecthw['type'] = 'SERVER'

    collecthw_json = json.dumps(_collecthw)
    if args.url:
        for url in args.url:
            send_post(url, _collecthw)
    if args.uuidurl:
        uuid = get_uuid()
        for uuidurl in args.uuidurl:
            url = uuidurl.replace("UUID", uuid)
            send_post(url, _collecthw)

    # Print out the final json
    if args.print:
        print(json.dumps(_collecthw))

if __name__ == "__main__":
    main()