Helper classes for variable attributes and policies (#128)

uefi: Add helper class for Variable Attributes
uefi/edk2: Add helper classes to work with VariablePolicy structures

Co-authored-by: Bret Barkelew <[email protected]>

edk2toollib/uefi/edk2/variable_policy.py

@@ -0,0 +1,188 @@
+# @file
+# Module contains helper classes and functions to work with Variable Policy structures
+# and substructures.
+#
+# Copyright (c) Microsoft Corporation
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+##
+
+import uuid
+import struct
+from edk2toollib.uefi.uefi_multi_phase import EfiVariableAttributes
+
+
+class VariableLockOnVarStatePolicy(object):
+    # typedef struct {
+    #     EFI_GUID Namespace;
+    #     UINT8    Value;
+    #     UINT8    Reserved;
+    #     // CHAR16   Name[];           // Variable Length Field
+    # } VARIABLE_LOCK_ON_VAR_STATE_POLICY;
+    _HdrStructFormat = "<16sBB"
+    _HdrStructSize = struct.calcsize(_HdrStructFormat)
+
+    def __init__(self):
+        self.Namespace = uuid.UUID(bytes=b'\x00' * 16)
+        self.Value = 0
+        self.Name = None
+
+    def __str__(self):
+        return "VARIABLE_LOCK_ON_VAR_STATE_POLICY(%s, %d, %s)" % (self.Namespace, self.Value, self.Name)
+
+    def decode(self, buffer):
+        """
+        load this object from a bytes buffer
+
+        return any remaining buffer
+        """
+        (_namespace, self.Value, _) = struct.unpack(
+            self._HdrStructFormat, buffer[:self._HdrStructSize])
+
+        self.Namespace = uuid.UUID(bytes_le=_namespace)
+
+        # Scan the rest of the buffer for a \x00\x00 to terminate the string.
+        buffer = buffer[self._HdrStructSize:]
+        if len(buffer) < 4:
+            raise ValueError("Buffer too short!")
+
+        string_end = None
+        for i in range(0, len(buffer), 2):
+            if buffer[i] == 0 and buffer[i + 1] == 0:
+                string_end = i + 2
+                break
+
+        if string_end is None:
+            raise ValueError("String end not detected!")
+
+        self.Name = buffer[:string_end].decode('utf-16').strip('\x00')
+
+        return buffer[string_end:]
+
+
+class VariablePolicyEntry(object):
+    # typedef struct {
+    #     UINT32   Version;
+    #     UINT16   Size;
+    #     UINT16   OffsetToName;
+    #     EFI_GUID Namespace;
+    #     UINT32   MinSize;
+    #     UINT32   MaxSize;
+    #     UINT32   AttributesMustHave;
+    #     UINT32   AttributesCantHave;
+    #     UINT8    LockPolicyType;
+    #     UINT8    Reserved[3];
+    #     // UINT8    LockPolicy[];     // Variable Length Field
+    #     // CHAR16   Name[]            // Variable Length Field
+    # } VARIABLE_POLICY_ENTRY;
+    _HdrStructFormat = "<IHH16sIIIIB3s"             # spell-checker:disable-line
+    _HdrStructSize = struct.calcsize(_HdrStructFormat)
+
+    ENTRY_REVISION = 0x0001_0000
+
+    NO_MIN_SIZE = 0
+    NO_MAX_SIZE = 0xFFFF_FFFF
+    NO_MUST_ATTR = 0
+    NO_CANT_ATTR = 0
+
+    TYPE_NO_LOCK = 0
+    TYPE_LOCK_NOW = 1
+    TYPE_LOCK_ON_CREATE = 2
+    TYPE_LOCK_ON_VAR_STATE = 3
+
+    LOCK_POLICY_STRING_MAP = {
+        TYPE_NO_LOCK: "NONE",
+        TYPE_LOCK_NOW: "NOW",
+        TYPE_LOCK_ON_CREATE: "ON_CREATE",
+        TYPE_LOCK_ON_VAR_STATE: "ON_VAR_STATE",
+    }
+
+    def __init__(self):
+        self.Version = VariablePolicyEntry.ENTRY_REVISION
+        self.Size = VariablePolicyEntry._HdrStructSize
+        self.OffsetToName = self.Size
+        self.Namespace = uuid.UUID(bytes=b'\x00' * 16)
+        self.MinSize = VariablePolicyEntry.NO_MIN_SIZE
+        self.MaxSize = VariablePolicyEntry.NO_MAX_SIZE
+        self.AttributesMustHave = VariablePolicyEntry.NO_MUST_ATTR
+        self.AttributesCantHave = VariablePolicyEntry.NO_CANT_ATTR
+        self.LockPolicyType = VariablePolicyEntry.TYPE_NO_LOCK
+        self.LockPolicy = None
+        self.Name = None
+
+    def __str__(self):
+        result = "VARIABLE_POLICY_ENTRY(%s, %s)\n" % (self.Namespace, self.Name)
+
+        if self.LockPolicyType in (VariablePolicyEntry.TYPE_NO_LOCK,
+                                   VariablePolicyEntry.TYPE_LOCK_NOW,
+                                   VariablePolicyEntry.TYPE_LOCK_ON_CREATE):
+            result += "\tLock        = %s\n" % VariablePolicyEntry.LOCK_POLICY_STRING_MAP[self.LockPolicyType]
+        elif self.LockPolicyType is VariablePolicyEntry.TYPE_LOCK_ON_VAR_STATE:
+            result += "\tLock        = %s\n" % self.LockPolicy
+
+        result += "\tMin = 0x%08X, Max = 0x%08X, Must = 0x%08X, Cant = 0x%08X\n" % (
+            self.MinSize, self.MaxSize, self.AttributesMustHave, self.AttributesCantHave)
+
+        return result
+
+    @staticmethod
+    def csv_header():
+        """returns a list containing the names of the ordered columns that are produced by csv_row()"""
+        return ['Namespace', 'Name', 'LockPolicyType', 'VarStateNamespace', 'VarStateName',
+                'VarStateValue', 'MinSize', 'MaxSize', 'AttributesMustHave', 'AttributesCantHave']
+
+    def csv_row(self, guid_xref: dict = None):
+        """
+        returns a list containing the elements of this structure (in the same order as the csv_header)
+        ready to be written to a csv file
+
+        guid_xref - a dictionary of GUID/name substitutions where the key is a uuid object
+                    and the value is a string
+        """
+        if guid_xref is None:
+            guid_xref = {}
+
+        result = [guid_xref.get(self.Namespace, self.Namespace),
+                  self.Name, VariablePolicyEntry.LOCK_POLICY_STRING_MAP[self.LockPolicyType]]
+
+        if self.LockPolicyType in (VariablePolicyEntry.TYPE_NO_LOCK,
+                                   VariablePolicyEntry.TYPE_LOCK_NOW,
+                                   VariablePolicyEntry.TYPE_LOCK_ON_CREATE):
+            result += ['N/A', 'N/A', 'N/A']
+        elif self.LockPolicyType is VariablePolicyEntry.TYPE_LOCK_ON_VAR_STATE:
+            result += [guid_xref.get(self.LockPolicy.Namespace, self.LockPolicy.Namespace),
+                       self.LockPolicy.Name, self.LockPolicy.Value]
+
+        result += ["0x%08X" % self.MinSize,
+                   "0x%08X" % self.MaxSize,
+                   str(EfiVariableAttributes(self.AttributesMustHave)),
+                   str(EfiVariableAttributes(self.AttributesCantHave))]
+
+        return result
+
+    def decode(self, buffer):
+        """
+        load this object from a bytes buffer
+
+        return any remaining buffer
+        """
+        (self.Version, self.Size, self.OffsetToName, _namespace,
+            self.MinSize, self.MaxSize, self.AttributesMustHave,
+            self.AttributesCantHave, self.LockPolicyType, _) = struct.unpack(
+                self._HdrStructFormat, buffer[:self._HdrStructSize])
+
+        if self.Version != VariablePolicyEntry.ENTRY_REVISION:
+            raise ValueError("Unknown structure version!")
+        if self.LockPolicyType not in VariablePolicyEntry.LOCK_POLICY_STRING_MAP:
+            raise ValueError("Unknown LockPolicyType!")
+
+        self.Namespace = uuid.UUID(bytes_le=_namespace)
+
+        if self.OffsetToName != self.Size:
+            self.Name = buffer[self.OffsetToName:self.Size].decode('utf-16').strip('\x00')
+
+        if self.LockPolicyType == VariablePolicyEntry.TYPE_LOCK_ON_VAR_STATE:
+            self.LockPolicy = VariableLockOnVarStatePolicy()
+            self.LockPolicy.decode(buffer[self._HdrStructSize:self.OffsetToName])
+
+        return buffer[self.Size:]

edk2toollib/uefi/edk2/variable_policy_test.py

@@ -0,0 +1,115 @@
+# @file
+# Unit test harness for the VariablePolicy module/classes.
+#
+# Copyright (c) Microsoft Corporation
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+##
+
+
+import unittest
+import uuid
+from edk2toollib.uefi.edk2.variable_policy import VariableLockOnVarStatePolicy, VariablePolicyEntry
+
+
+TEST_GUID_1 = uuid.UUID("48B5F961-3F7D-4B88-9BEE-D305ED8256DA")
+TEST_GUID_2 = uuid.UUID("65D16747-FCBC-4FAE-A727-7B679A7B23F9")
+
+TEST_POLICY_ENTRY = b''.fromhex("000001006A004600E222FFB0EA4A2547A6E55317FB8FD39C00000000FFFFFFFF000000000000000003AFAFAFC690F5ECF9F887438422486E3CCD8B2001AF45004F00440000004C0061007300740041007400740065006D00700074005300740061007400750073000000")                 # noqa
+TEST_POLICY_ENTRY_BAD_VERSION = b''.fromhex("010001006A004600E222FFB0EA4A2547A6E55317FB8FD39C00000000FFFFFFFF000000000000000003AFAFAFC690F5ECF9F887438422486E3CCD8B2001AF45004F00440000004C0061007300740041007400740065006D00700074005300740061007400750073000000")     # noqa
+TEST_POLICY_ENTRY_BAD_LOCK_TYPE = b''.fromhex("000001006A004600E222FFB0EA4A2547A6E55317FB8FD39C00000000FFFFFFFF000000000000000004AFAFAFC690F5ECF9F887438422486E3CCD8B2001AF45004F00440000004C0061007300740041007400740065006D00700074005300740061007400750073000000")   # noqa
+TEST_POLICY_ENTRY_GUID = uuid.UUID("B0FF22E2-4AEA-4725-A6E5-5317FB8FD39C")
+
+
+class TestVariableLockOnVarStatePolicy(unittest.TestCase):
+    def test_remaining_buffer(self):
+        test_vpl = VariableLockOnVarStatePolicy()
+        test_remainder = b'123'
+        test_buffer = TEST_GUID_2.bytes_le + b'\x00\x00' + b'\x00A\x00\x00' + test_remainder
+
+        self.assertEqual(test_remainder, test_vpl.decode(test_buffer))
+
+    def test_missing_name(self):
+        test_vpl = VariableLockOnVarStatePolicy()
+
+        # Test with no Name field at all.
+        test1 = TEST_GUID_1.bytes_le + b'\x00\x00'
+        with self.assertRaises(Exception):
+            test_vpl.decode(test1)
+
+        # Test with an empty string.
+        test2 = test1 + b'\x00\x00'
+        with self.assertRaises(Exception):
+            test_vpl.decode(test2)
+
+        # Test successful.
+        test3 = test1 + b'\x00A\x00\x00'
+        _ = test_vpl.decode(test3)
+
+    def test_malformed_name(self):
+        test_vpl = VariableLockOnVarStatePolicy()
+
+        # Test with no termination.
+        test1 = TEST_GUID_1.bytes_le + b'\x00\x00' + b'\x00A\x00B'
+        with self.assertRaises(Exception):
+            test_vpl.decode(test1)
+
+        # Test with an unaligned termination.
+        test2 = TEST_GUID_1.bytes_le + b'\x00\x00' + b'A\x00B\x00' + b'C' + b'\x00\x00'
+        with self.assertRaises(Exception):
+            test_vpl.decode(test2)
+
+    def test_to_string(self):
+        test_vpl = VariableLockOnVarStatePolicy()
+        test_buffer = TEST_GUID_2.bytes_le + b'\x00\x00' + b'A\x00B\x00C\x00\x00\x00'
+
+        test_vpl.decode(test_buffer)
+
+        self.assertEqual(test_vpl.Name, "ABC")
+
+
+class TestVariablePolicyEntry(unittest.TestCase):
+    def test_create_and_to_string(self):
+        test_vp = VariablePolicyEntry()
+        to_string = str(test_vp)
+
+        # Check for the LockType string.
+        self.assertIn("NONE", to_string)
+
+        test_vp.LockPolicyType = VariablePolicyEntry.TYPE_LOCK_ON_CREATE
+        to_string = str(test_vp)
+
+        # Check for the new LockType string.
+        self.assertIn("CREATE", to_string)
+
+    def test_csv_formatting(self):
+        header_row = VariablePolicyEntry.csv_header()
+        self.assertIn("Namespace", header_row)
+        self.assertIn("LockPolicyType", header_row)
+
+        test_vp = VariablePolicyEntry()
+        test_vp.LockPolicyType = VariablePolicyEntry.TYPE_LOCK_ON_CREATE
+        csv_row = test_vp.csv_row()
+        self.assertEqual(len(header_row), len(csv_row))
+        self.assertIn("ON_CREATE", csv_row)
+
+    def test_decoding(self):
+        test_vp = VariablePolicyEntry()
+        test_vp.decode(TEST_POLICY_ENTRY)
+
+        self.assertEqual(test_vp.Namespace, TEST_POLICY_ENTRY_GUID)
+        self.assertEqual(test_vp.LockPolicyType, VariablePolicyEntry.TYPE_LOCK_ON_VAR_STATE)
+        self.assertEqual(test_vp.Name, "LastAttemptStatus")
+        self.assertEqual(test_vp.LockPolicy.Name, "EOD")
+
+        to_string = str(test_vp)
+        self.assertIn("VAR_STATE", to_string)
+        self.assertIn("EOD", to_string)
+        self.assertIn("LastAttemptStatus", to_string)
+
+    def test_decoding_errors(self):
+        test_vp = VariablePolicyEntry()
+
+        with self.assertRaises(ValueError):
+            test_vp.decode(TEST_POLICY_ENTRY_BAD_VERSION)
+        with self.assertRaises(ValueError):
+            test_vp.decode(TEST_POLICY_ENTRY_BAD_LOCK_TYPE)

edk2toollib/uefi/uefi_multi_phase.py

@@ -6,10 +6,38 @@
 # SPDX-License-Identifier: BSD-2-Clause-Patent
 ##
 
+import struct
+
 EFI_VARIABLE_NON_VOLATILE = 0x00000001
 EFI_VARIABLE_BOOTSERVICE_ACCESS = 0x00000002
 EFI_VARIABLE_RUNTIME_ACCESS = 0x00000004
 EFI_VARIABLE_HARDWARE_ERROR_RECORD = 0x00000008
 EFI_VARIABLE_AUTHENTICATED_WRITE_ACCESS = 0x00000010
 EFI_VARIABLE_TIME_BASED_AUTHENTICATED_WRITE_ACCESS = 0x00000020
 EFI_VARIABLE_APPEND_WRITE = 0x00000040
+
+
+class EfiVariableAttributes(object):
+    # UINT32
+    _StructFormat = "<I"
+    _StructSize = struct.calcsize(_StructFormat)
+
+    STRING_MAP = {
+        EFI_VARIABLE_APPEND_WRITE: "EFI_VARIABLE_APPEND_WRITE",
+        EFI_VARIABLE_TIME_BASED_AUTHENTICATED_WRITE_ACCESS: "EFI_VARIABLE_TIME_BASED_AUTHENTICATED_WRITE_ACCESS",
+        EFI_VARIABLE_AUTHENTICATED_WRITE_ACCESS: "EFI_VARIABLE_AUTHENTICATED_WRITE_ACCESS",
+        EFI_VARIABLE_HARDWARE_ERROR_RECORD: "EFI_VARIABLE_HARDWARE_ERROR_RECORD",
+        EFI_VARIABLE_RUNTIME_ACCESS: "EFI_VARIABLE_RUNTIME_ACCESS",
+        EFI_VARIABLE_BOOTSERVICE_ACCESS: "EFI_VARIABLE_BOOTSERVICE_ACCESS",
+        EFI_VARIABLE_NON_VOLATILE: "EFI_VARIABLE_NON_VOLATILE",
+    }
+
+    def __init__(self, attributes=0x0000_0000):
+        self.Attributes = attributes
+
+    def __str__(self):
+        result = []
+        for key in EfiVariableAttributes.STRING_MAP:
+            if self.Attributes & key:
+                result.append(EfiVariableAttributes.STRING_MAP[key])
+        return ",".join(result)

edk2toollib/uefi/uefi_multi_phase_test.py

@@ -0,0 +1,23 @@
+# @file
+# Code to test UEFI MultiPhase module
+#
+# Copyright (c) Microsoft Corporation
+#
+# SPDX-License-Identifier: BSD-2-Clause-Patent
+##
+import unittest
+from edk2toollib.uefi.uefi_multi_phase import (EfiVariableAttributes,
+                                               EFI_VARIABLE_NON_VOLATILE, EFI_VARIABLE_RUNTIME_ACCESS,
+                                               EFI_VARIABLE_BOOTSERVICE_ACCESS)
+
+
+class TestUefiMultiphase (unittest.TestCase):
+
+    def test_StringConversion(self):
+        attr = EfiVariableAttributes(EFI_VARIABLE_NON_VOLATILE
+                                     | EFI_VARIABLE_RUNTIME_ACCESS | EFI_VARIABLE_BOOTSERVICE_ACCESS)
+        string = str(attr)
+
+        self.assertTrue("EFI_VARIABLE_RUNTIME_ACCESS" in string)
+        self.assertTrue("EFI_VARIABLE_NON_VOLATILE" in string)
+        self.assertTrue("EFI_VARIABLE_BOOTSERVICE_ACCESS" in string)