path: root/scripts/qmp_helper.py

#!/usr/bin/env python3
#
# pylint: disable=C0103,E0213,E1135,E1136,E1137,R0902,R0903,R0912,R0913,R0917
# SPDX-License-Identifier: GPL-2.0-or-later
#
# Copyright (C) 2024-2025 Mauro Carvalho Chehab <mchehab+huawei@kernel.org>

"""
Helper classes to be used by ghes_inject command classes.
"""

import json
import sys

from datetime import datetime
from os import path as os_path

try:
    qemu_dir = os_path.abspath(os_path.dirname(os_path.dirname(__file__)))
    sys.path.append(os_path.join(qemu_dir, 'python'))

    from qemu.qmp.legacy import QEMUMonitorProtocol

except ModuleNotFoundError as exc:
    print(f"Module '{exc.name}' not found.")
    print("Try export PYTHONPATH=top-qemu-dir/python or run from top-qemu-dir")
    sys.exit(1)

from base64 import b64encode

class util:
    """
    Ancillary functions to deal with bitmaps, parse arguments,
    generate GUID and encode data on a bytearray buffer.
    """

    #
    # Helper routines to handle multiple choice arguments
    #
    def get_choice(name, value, choices, suffixes=None, bitmask=True):
        """Produce a list from multiple choice argument"""

        new_values = 0

        if not value:
            return new_values

        for val in value.split(","):
            val = val.lower()

            if suffixes:
                for suffix in suffixes:
                    val = val.removesuffix(suffix)

            if val not in choices.keys():
                if suffixes:
                    for suffix in suffixes:
                        if val + suffix in choices.keys():
                            val += suffix
                            break

            if val not in choices.keys():
                sys.exit(f"Error on '{name}': choice '{val}' is invalid.")

            val = choices[val]

            if bitmask:
                new_values |= val
            else:
                if new_values:
                    sys.exit(f"Error on '{name}': only one value is accepted.")

                new_values = val

        return new_values

    def get_array(name, values, max_val=None):
        """Add numbered hashes from integer lists into an array"""

        array = []

        for value in values:
            for val in value.split(","):
                try:
                    val = int(val, 0)
                except ValueError:
                    sys.exit(f"Error on '{name}': {val} is not an integer")

                if val < 0:
                    sys.exit(f"Error on '{name}': {val} is not unsigned")

                if max_val and val > max_val:
                    sys.exit(f"Error on '{name}': {val} is too little")

                array.append(val)

        return array

    def get_mult_array(mult, name, values, allow_zero=False, max_val=None):
        """Add numbered hashes from integer lists"""

        if not allow_zero:
            if not values:
                return
        else:
            if values is None:
                return

            if not values:
                i = 0
                if i not in mult:
                    mult[i] = {}

                mult[i][name] = []
                return

        i = 0
        for value in values:
            for val in value.split(","):
                try:
                    val = int(val, 0)
                except ValueError:
                    sys.exit(f"Error on '{name}': {val} is not an integer")

                if val < 0:
                    sys.exit(f"Error on '{name}': {val} is not unsigned")

                if max_val and val > max_val:
                    sys.exit(f"Error on '{name}': {val} is too little")

                if i not in mult:
                    mult[i] = {}

                if name not in mult[i]:
                    mult[i][name] = []

                mult[i][name].append(val)

            i += 1


    def get_mult_choices(mult, name, values, choices,
                        suffixes=None, allow_zero=False):
        """Add numbered hashes from multiple choice arguments"""

        if not allow_zero:
            if not values:
                return
        else:
            if values is None:
                return

        i = 0
        for val in values:
            new_values = util.get_choice(name, val, choices, suffixes)

            if i not in mult:
                mult[i] = {}

            mult[i][name] = new_values
            i += 1


    def get_mult_int(mult, name, values, allow_zero=False):
        """Add numbered hashes from integer arguments"""
        if not allow_zero:
            if not values:
                return
        else:
            if values is None:
                return

        i = 0
        for val in values:
            try:
                val = int(val, 0)
            except ValueError:
                sys.exit(f"Error on '{name}': {val} is not an integer")

            if val < 0:
                sys.exit(f"Error on '{name}': {val} is not unsigned")

            if i not in mult:
                mult[i] = {}

            mult[i][name] = val
            i += 1


    #
    # Data encode helper functions
    #
    def bit(b):
        """Simple macro to define a bit on a bitmask"""
        return 1 << b


    def data_add(data, value, num_bytes):
        """Adds bytes from value inside a bitarray"""

        data.extend(value.to_bytes(num_bytes, byteorder="little"))  # pylint: disable=E1101

    def dump_bytearray(name, data):
        """Does an hexdump of a byte array, grouping in bytes"""

        print(f"{name} ({len(data)} bytes):")

        for ln_start in range(0, len(data), 16):
            ln_end = min(ln_start + 16, len(data))
            print(f"      {ln_start:08x}  ", end="")
            for i in range(ln_start, ln_end):
                print(f"{data[i]:02x} ", end="")
            for i in range(ln_end, ln_start + 16):
                print("   ", end="")
            print("  ", end="")
            for i in range(ln_start, ln_end):
                if data[i] >= 32 and data[i] < 127:
                    print(chr(data[i]), end="")
                else:
                    print(".", end="")

            print()
        print()

    def time(string):
        """Handle BCD timestamps used on Generic Error Data Block"""

        time = None

        # Formats to be used when parsing time stamps
        formats = [
            "%Y-%m-%d %H:%M:%S",
        ]

        if string == "now":
            time = datetime.now()

        if time is None:
            for fmt in formats:
                try:
                    time = datetime.strptime(string, fmt)
                    break
                except ValueError:
                    pass

            if time is None:
                raise ValueError("Invalid time format")

        return time

class guid:
    """
    Simple class to handle GUID fields.
    """

    def __init__(self, time_low, time_mid, time_high, nodes):
        """Initialize a GUID value"""

        assert len(nodes) == 8

        self.time_low = time_low
        self.time_mid = time_mid
        self.time_high = time_high
        self.nodes = nodes

    @classmethod
    def UUID(cls, guid_str):
        """Initialize a GUID using a string on its standard format"""

        if len(guid_str) != 36:
            print("Size not 36")
            raise ValueError('Invalid GUID size')

        # It is easier to parse without separators. So, drop them
        guid_str = guid_str.replace('-', '')

        if len(guid_str) != 32:
            print("Size not 32", guid_str, len(guid_str))
            raise ValueError('Invalid GUID hex size')

        time_low = 0
        time_mid = 0
        time_high = 0
        nodes = []

        for i in reversed(range(16, 32, 2)):
            h = guid_str[i:i + 2]
            value = int(h, 16)
            nodes.insert(0, value)

        time_high = int(guid_str[12:16], 16)
        time_mid = int(guid_str[8:12], 16)
        time_low = int(guid_str[0:8], 16)

        return cls(time_low, time_mid, time_high, nodes)

    def __str__(self):
        """Output a GUID value on its default string representation"""

        clock = self.nodes[0] << 8 | self.nodes[1]

        node = 0
        for i in range(2, len(self.nodes)):
            node = node << 8 | self.nodes[i]

        s = f"{self.time_low:08x}-{self.time_mid:04x}-"
        s += f"{self.time_high:04x}-{clock:04x}-{node:012x}"
        return s

    def to_bytes(self):
        """Output a GUID value in bytes"""

        data = bytearray()

        util.data_add(data, self.time_low, 4)
        util.data_add(data, self.time_mid, 2)
        util.data_add(data, self.time_high, 2)
        data.extend(bytearray(self.nodes))

        return data

class qmp:
    """
    Opens a connection and send/receive QMP commands.
    """

    def send_cmd(self, command, args=None, may_open=False, return_error=True):
        """Send a command to QMP, optinally opening a connection"""

        if may_open:
            self._connect()
        elif not self.connected:
            return False

        msg = { 'execute': command }
        if args:
            msg['arguments'] = args

        try:
            obj = self.qmp_monitor.cmd_obj(msg)
        # Can we use some other exception class here?
        except Exception as e:                         # pylint: disable=W0718
            print(f"Command: {command}")
            print(f"Failed to inject error: {e}.")
            return None

        if "return" in obj:
            if isinstance(obj.get("return"), dict):
                if obj["return"]:
                    return obj["return"]
                return "OK"

            return obj["return"]

        if isinstance(obj.get("error"), dict):
            error = obj["error"]
            if return_error:
                print(f"Command: {msg}")
                print(f'{error["class"]}: {error["desc"]}')
        else:
            print(json.dumps(obj))

        return None

    def _close(self):
        """Shutdown and close the socket, if opened"""
        if not self.connected:
            return

        self.qmp_monitor.close()
        self.connected = False

    def _connect(self):
        """Connect to a QMP TCP/IP port, if not connected yet"""

        if self.connected:
            return True

        try:
            self.qmp_monitor.connect(negotiate=True)
        except ConnectionError:
            sys.exit(f"Can't connect to QMP host {self.host}:{self.port}")

        self.connected = True

        return True

    BLOCK_STATUS_BITS = {
        "uncorrectable":            util.bit(0),
        "correctable":              util.bit(1),
        "multi-uncorrectable":      util.bit(2),
        "multi-correctable":        util.bit(3),
    }

    ERROR_SEVERITY = {
        "recoverable":  0,
        "fatal":        1,
        "corrected":    2,
        "none":         3,
    }

    VALIDATION_BITS = {
        "fru-id":       util.bit(0),
        "fru-text":     util.bit(1),
        "timestamp":    util.bit(2),
    }

    GEDB_FLAGS_BITS = {
        "recovered":    util.bit(0),
        "prev-error":   util.bit(1),
        "simulated":    util.bit(2),
    }

    GENERIC_DATA_SIZE = 72

    def argparse(parser):
        """Prepare a parser group to query generic error data"""

        block_status_bits = ",".join(qmp.BLOCK_STATUS_BITS.keys())
        error_severity_enum = ",".join(qmp.ERROR_SEVERITY.keys())
        validation_bits = ",".join(qmp.VALIDATION_BITS.keys())
        gedb_flags_bits = ",".join(qmp.GEDB_FLAGS_BITS.keys())

        g_gen = parser.add_argument_group("Generic Error Data")  # pylint: disable=E1101
        g_gen.add_argument("--block-status",
                           help=f"block status bits: {block_status_bits}")
        g_gen.add_argument("--raw-data", nargs="+",
                        help="Raw data inside the Error Status Block")
        g_gen.add_argument("--error-severity", "--severity",
                           help=f"error severity: {error_severity_enum}")
        g_gen.add_argument("--gen-err-valid-bits",
                           "--generic-error-validation-bits",
                           help=f"validation bits: {validation_bits}")
        g_gen.add_argument("--fru-id", type=guid.UUID,
                           help="GUID representing a physical device")
        g_gen.add_argument("--fru-text",
                           help="ASCII string identifying the FRU hardware")
        g_gen.add_argument("--timestamp", type=util.time,
                           help="Time when the error info was collected")
        g_gen.add_argument("--precise", "--precise-timestamp",
                           action='store_true',
                           help="Marks the timestamp as precise if --timestamp is used")
        g_gen.add_argument("--gedb-flags",
                           help=f"General Error Data Block flags: {gedb_flags_bits}")

    def set_args(self, args):
        """Set the arguments optionally defined via self.argparse()"""

        if args.block_status:
            self.block_status = util.get_choice(name="block-status",
                                                value=args.block_status,
                                                choices=self.BLOCK_STATUS_BITS,
                                                bitmask=False)
        if args.raw_data:
            self.raw_data = util.get_array("raw-data", args.raw_data,
                                           max_val=255)
            print(self.raw_data)

        if args.error_severity:
            self.error_severity = util.get_choice(name="error-severity",
                                                  value=args.error_severity,
                                                  choices=self.ERROR_SEVERITY,
                                                  bitmask=False)

        if args.fru_id:
            self.fru_id = args.fru_id.to_bytes()
            if not args.gen_err_valid_bits:
                self.validation_bits |= self.VALIDATION_BITS["fru-id"]

        if args.fru_text:
            text = bytearray(args.fru_text.encode('ascii'))
            if len(text) > 20:
                sys.exit("FRU text is too big to fit")

            self.fru_text = text
            if not args.gen_err_valid_bits:
                self.validation_bits |= self.VALIDATION_BITS["fru-text"]

        if args.timestamp:
            time = args.timestamp
            century = int(time.year / 100)

            bcd = bytearray()
            util.data_add(bcd, (time.second // 10) << 4 | (time.second % 10), 1)
            util.data_add(bcd, (time.minute // 10) << 4 | (time.minute % 10), 1)
            util.data_add(bcd, (time.hour // 10) << 4 | (time.hour % 10), 1)

            if args.precise:
                util.data_add(bcd, 1, 1)
            else:
                util.data_add(bcd, 0, 1)

            util.data_add(bcd, (time.day // 10) << 4 | (time.day % 10), 1)
            util.data_add(bcd, (time.month // 10) << 4 | (time.month % 10), 1)
            util.data_add(bcd,
                          ((time.year % 100) // 10) << 4 | (time.year % 10), 1)
            util.data_add(bcd, ((century % 100) // 10) << 4 | (century % 10), 1)

            self.timestamp = bcd
            if not args.gen_err_valid_bits:
                self.validation_bits |= self.VALIDATION_BITS["timestamp"]

        if args.gen_err_valid_bits:
            self.validation_bits = util.get_choice(name="validation",
                                                   value=args.gen_err_valid_bits,
                                                   choices=self.VALIDATION_BITS)

    def __init__(self, host, port, debug=False):
        """Initialize variables used by the QMP send logic"""

        self.connected = False
        self.host = host
        self.port = port
        self.debug = debug

        # ACPI 6.1: 18.3.2.7.1 Generic Error Data: Generic Error Status Block
        self.block_status = self.BLOCK_STATUS_BITS["uncorrectable"]
        self.raw_data = []
        self.error_severity = self.ERROR_SEVERITY["recoverable"]

        # ACPI 6.1: 18.3.2.7.1 Generic Error Data: Generic Error Data Entry
        self.validation_bits = 0
        self.flags = 0
        self.fru_id = bytearray(16)
        self.fru_text = bytearray(20)
        self.timestamp = bytearray(8)

        self.qmp_monitor = QEMUMonitorProtocol(address=(self.host, self.port))

    #
    # Socket QMP send command
    #
    def send_cper_raw(self, cper_data):
        """Send a raw CPER data to QEMU though QMP TCP socket"""

        data = b64encode(bytes(cper_data)).decode('ascii')

        cmd_arg = {
            'cper': data
        }

        self._connect()

        if self.send_cmd("inject-ghes-v2-error", cmd_arg):
            print("Error injected.")

    def send_cper(self, notif_type, payload):
        """Send commands to QEMU though QMP TCP socket"""

        # Fill CPER record header

        # NOTE: bits 4 to 13 of block status contain the number of
        # data entries in the data section. This is currently unsupported.

        cper_length = len(payload)
        data_length = cper_length + len(self.raw_data) + self.GENERIC_DATA_SIZE

        #  Generic Error Data Entry
        gede = bytearray()

        gede.extend(notif_type.to_bytes())
        util.data_add(gede, self.error_severity, 4)
        util.data_add(gede, 0x300, 2)
        util.data_add(gede, self.validation_bits, 1)
        util.data_add(gede, self.flags, 1)
        util.data_add(gede, cper_length, 4)
        gede.extend(self.fru_id)
        gede.extend(self.fru_text)
        gede.extend(self.timestamp)

        # Generic Error Status Block
        gebs = bytearray()

        if self.raw_data:
            raw_data_offset = len(gebs)
        else:
            raw_data_offset = 0

        util.data_add(gebs, self.block_status, 4)
        util.data_add(gebs, raw_data_offset, 4)
        util.data_add(gebs, len(self.raw_data), 4)
        util.data_add(gebs, data_length, 4)
        util.data_add(gebs, self.error_severity, 4)

        cper_data = bytearray()
        cper_data.extend(gebs)
        cper_data.extend(gede)
        cper_data.extend(bytearray(self.raw_data))
        cper_data.extend(bytearray(payload))

        if self.debug:
            print(f"GUID: {notif_type}")

            util.dump_bytearray("Generic Error Status Block", gebs)
            util.dump_bytearray("Generic Error Data Entry", gede)

            if self.raw_data:
                util.dump_bytearray("Raw data", bytearray(self.raw_data))

            util.dump_bytearray("Payload", payload)

        self.send_cper_raw(cper_data)


    def search_qom(self, path, prop, regex):
        """
        Return a list of devices that match path array like:

            /machine/unattached/device
            /machine/peripheral-anon/device
            ...
        """

        found = []

        i = 0
        while 1:
            dev = f"{path}[{i}]"
            args = {
                'path': dev,
                'property': prop
            }
            ret = self.send_cmd("qom-get", args, may_open=True,
                                return_error=False)
            if not ret:
                break

            if isinstance(ret, str):
                if regex.search(ret):
                    found.append(dev)

            i += 1
            if i > 10000:
                print("Too many objects returned by qom-get!")
                break

        return found

class cper_guid:
    """
    Contains CPER GUID, as per:
    https://uefi.org/specs/UEFI/2.10/Apx_N_Common_Platform_Error_Record.html
    """

    CPER_PROC_GENERIC =  guid(0x9876CCAD, 0x47B4, 0x4bdb,
                              [0xB6, 0x5E, 0x16, 0xF1,
                               0x93, 0xC4, 0xF3, 0xDB])

    CPER_PROC_X86 = guid(0xDC3EA0B0, 0xA144, 0x4797,
                         [0xB9, 0x5B, 0x53, 0xFA,
                          0x24, 0x2B, 0x6E, 0x1D])

    CPER_PROC_ITANIUM = guid(0xe429faf1, 0x3cb7, 0x11d4,
                             [0xbc, 0xa7, 0x00, 0x80,
                              0xc7, 0x3c, 0x88, 0x81])

    CPER_PROC_ARM = guid(0xE19E3D16, 0xBC11, 0x11E4,
                         [0x9C, 0xAA, 0xC2, 0x05,
                          0x1D, 0x5D, 0x46, 0xB0])

    CPER_PLATFORM_MEM = guid(0xA5BC1114, 0x6F64, 0x4EDE,
                             [0xB8, 0x63, 0x3E, 0x83,
                              0xED, 0x7C, 0x83, 0xB1])

    CPER_PLATFORM_MEM2 = guid(0x61EC04FC, 0x48E6, 0xD813,
                              [0x25, 0xC9, 0x8D, 0xAA,
                               0x44, 0x75, 0x0B, 0x12])

    CPER_PCIE = guid(0xD995E954, 0xBBC1, 0x430F,
                     [0xAD, 0x91, 0xB4, 0x4D,
                      0xCB, 0x3C, 0x6F, 0x35])

    CPER_PCI_BUS = guid(0xC5753963, 0x3B84, 0x4095,
                        [0xBF, 0x78, 0xED, 0xDA,
                         0xD3, 0xF9, 0xC9, 0xDD])

    CPER_PCI_DEV = guid(0xEB5E4685, 0xCA66, 0x4769,
                        [0xB6, 0xA2, 0x26, 0x06,
                         0x8B, 0x00, 0x13, 0x26])

    CPER_FW_ERROR = guid(0x81212A96, 0x09ED, 0x4996,
                         [0x94, 0x71, 0x8D, 0x72,
                          0x9C, 0x8E, 0x69, 0xED])

    CPER_DMA_GENERIC = guid(0x5B51FEF7, 0xC79D, 0x4434,
                            [0x8F, 0x1B, 0xAA, 0x62,
                             0xDE, 0x3E, 0x2C, 0x64])

    CPER_DMA_VT = guid(0x71761D37, 0x32B2, 0x45cd,
                       [0xA7, 0xD0, 0xB0, 0xFE,
                        0xDD, 0x93, 0xE8, 0xCF])

    CPER_DMA_IOMMU = guid(0x036F84E1, 0x7F37, 0x428c,
                         [0xA7, 0x9E, 0x57, 0x5F,
                          0xDF, 0xAA, 0x84, 0xEC])

    CPER_CCIX_PER = guid(0x91335EF6, 0xEBFB, 0x4478,
                         [0xA6, 0xA6, 0x88, 0xB7,
                          0x28, 0xCF, 0x75, 0xD7])

    CPER_CXL_PROT_ERR = guid(0x80B9EFB4, 0x52B5, 0x4DE3,
                             [0xA7, 0x77, 0x68, 0x78,
                              0x4B, 0x77, 0x10, 0x48])