Merge 23fe591631 into b8cb1bc983

This section contains a predefined set of HWIDs and the corresponding compatibles to be used in dtb matching

src/ukify/ukify.py

@@ -238,7 +238,9 @@ class UkifyConfig:
     all: bool
     cmdline: Union[str, Path, None]
     devicetree: Path
+    devicetree_auto: list[Path]
     efi_arch: str
+    hwids_directory: Path
     initrd: list[Path]
     join_profiles: list[Path]
     json: Union[Literal['pretty'], Literal['short'], Literal['off']]
@@ -363,6 +365,8 @@ DEFAULT_SECTIONS_TO_SHOW = {
     '.ucode':   'binary',
     '.splash':  'binary',
     '.dtb':     'binary',
+    '.dtbauto': 'binary',
+    '.hwids':   'binary',
     '.cmdline': 'text',
     '.osrel':   'text',
     '.uname':   'text',
@@ -445,7 +449,7 @@ class UKI:
             if s.name == '.profile':
                 start = i + 1
 
-        if any(section.name == s.name for s in self.sections[start:]):
+        if any(section.name == s.name for s in self.sections[start:] if s.name != '.dtbauto'):
             raise ValueError(f'Duplicate section {section.name}')
 
         self.sections += [section]
@@ -575,7 +579,10 @@ def check_inputs(opts: UkifyConfig) -> None:
 
         if isinstance(value, Path):
             # Open file to check that we can read it, or generate an exception
-            value.open().close()
+            if not value.is_dir():
+                value.open().close()
+            else:
+                value.iterdir()
         elif isinstance(value, list):
             for item in value:
                 if isinstance(item, Path):
@@ -658,7 +665,16 @@ def call_systemd_measure(uki: UKI, opts: UkifyConfig, profile_start: int = 0) ->
     # PCR measurement
 
     # First, pick up either the base sections or the profile specific sections we shall measure now
-    to_measure = {s.name: s for s in uki.sections[profile_start:] if s.measure}
+    unique_to_measure = {
+        s.name: s for s in uki.sections[profile_start:] if s.measure and s.name != '.dtbauto'
+    }
+
+    dtbauto_to_measure: list[Optional[Section]] = [
+        s for s in uki.sections[profile_start:] if s.measure and s.name == '.dtbauto'
+    ]
+
+    if len(dtbauto_to_measure) == 0:
+        dtbauto_to_measure = [None]
 
     # Then, if we're measuring a profile, lookup the missing sections from the base image.
     if profile_start != 0:
@@ -672,53 +688,64 @@ def call_systemd_measure(uki: UKI, opts: UkifyConfig, profile_start: int = 0) ->
                 continue
 
             # Check if this is a section we already covered above
-            if section.name in to_measure:
+            if section.name in unique_to_measure:
                 continue
 
-            to_measure[section.name] = section
+            unique_to_measure[section.name] = section
 
     if opts.measure:
-        pp_groups = opts.phase_path_groups or []
+        to_measure = dict((k, v) for k, v in unique_to_measure.items())
 
-        cmd = [
+        for dtbauto in dtbauto_to_measure:
-            measure_tool,
+            if dtbauto is not None:
-            'calculate',
+                to_measure[dtbauto.name] = dtbauto
-            *(f"--{s.name.removeprefix('.')}={s.content}" for s in to_measure.values()),
-            *(f'--bank={bank}' for bank in banks),
-            # For measurement, the keys are not relevant, so we can lump all the phase paths
-            # into one call to systemd-measure calculate.
-            *(f'--phase={phase_path}' for phase_path in itertools.chain.from_iterable(pp_groups)),
-        ]
 
-        print('+', shell_join(cmd))
+            pp_groups = opts.phase_path_groups or []
-        subprocess.check_call(cmd)
+
+            cmd = [
+                measure_tool,
+                'calculate',
+                *(f"--{s.name.removeprefix('.')}={s.content}" for s in to_measure.values()),
+                *(f'--bank={bank}' for bank in banks),
+                # For measurement, the keys are not relevant, so we can lump all the phase paths
+                # into one call to systemd-measure calculate.
+                *(f'--phase={phase_path}' for phase_path in itertools.chain.from_iterable(pp_groups)),
+            ]
+
+            print('+', shell_join(cmd))
+            subprocess.check_call(cmd)
 
     # PCR signing
 
     if opts.pcr_private_keys:
         pcrsigs = []
+        to_measure = dict((k, v) for k, v in unique_to_measure.items())
 
-        cmd = [
+        for dtbauto in dtbauto_to_measure:
-            measure_tool,
+            if dtbauto is not None:
-            'sign',
+                to_measure[dtbauto.name] = dtbauto
-            *(f"--{s.name.removeprefix('.')}={s.content}" for s in to_measure.values()),
-            *(f'--bank={bank}' for bank in banks),
-        ]
 
-        for priv_key, pub_key, group in key_path_groups(opts):
+            cmd = [
-            extra = [f'--private-key={priv_key}']
+                measure_tool,
-            if opts.signing_engine is not None:
+                'sign',
-                assert pub_key
+                *(f"--{s.name.removeprefix('.')}={s.content}" for s in to_measure.values()),
-                extra += [f'--private-key-source=engine:{opts.signing_engine}']
+                *(f'--bank={bank}' for bank in banks),
-                extra += [f'--certificate={pub_key}']
+            ]
-            elif pub_key:
-                extra += [f'--public-key={pub_key}']
-            extra += [f'--phase={phase_path}' for phase_path in group or ()]
 
-            print('+', shell_join(cmd + extra))  # type: ignore
+            for priv_key, pub_key, group in key_path_groups(opts):
-            pcrsig = subprocess.check_output(cmd + extra, text=True)  # type: ignore
+                extra = [f'--private-key={priv_key}']
-            pcrsig = json.loads(pcrsig)
+                if opts.signing_engine is not None:
-            pcrsigs += [pcrsig]
+                    assert pub_key
+                    extra += [f'--private-key-source=engine:{opts.signing_engine}']
+                    extra += [f'--certificate={pub_key}']
+                elif pub_key:
+                    extra += [f'--public-key={pub_key}']
+                extra += [f'--phase={phase_path}' for phase_path in group or ()]
+
+                print('+', shell_join(cmd + extra))  # type: ignore
+                pcrsig = subprocess.check_output(cmd + extra, text=True)  # type: ignore
+                pcrsig = json.loads(pcrsig)
+                pcrsigs += [pcrsig]
 
         combined = combine_signatures(pcrsigs)
         uki.add_section(Section.create('.pcrsig', combined))
@@ -849,7 +876,7 @@ def pe_add_sections(uki: UKI, output: str) -> None:
         # the one from the kernel to it. It should be small enough to fit in the existing section, so just
         # swap the data.
         for i, s in enumerate(pe.sections[:n_original_sections]):
-            if pe_strip_section_name(s.Name) == section.name:
+            if pe_strip_section_name(s.Name) == section.name and section.name != '.dtbauto':
                 if new_section.Misc_VirtualSize > s.SizeOfRawData:
                     raise PEError(f'Not enough space in existing section {section.name} to append new data.')
 
@@ -921,6 +948,117 @@ def merge_sbat(input_pe: list[Path], input_text: list[str]) -> str:
     )
 
 
+# Keep in sync with EFI_GUID (src/boot/efi/efi.h)
+# uint32_t Data1, uint16_t Data2, uint16_t Data3, uint8_t Data4[8]
+EFI_GUID = tuple[int, int, int, tuple[int, int, int, int, int, int, int, int]]
+EFI_GUID_STRUCT_SIZE = 4 + 2 + 2 + 1 * 8
+
+# Keep in sync with Device (src/boot/efi/chid.h)
+# uint32_t struct_size, uint32_t name_offset, uint32_t compatible_offset, EFI_GUID chid
+DEVICE_STRUCT_SIZE = 4 + 4 + 4 + EFI_GUID_STRUCT_SIZE
+NULL_DEVICE = b'\0' * DEVICE_STRUCT_SIZE
+
+
+def pack_device(offsets: dict[str, int], name: str, compatible: str, chids: set[EFI_GUID]) -> bytes:
+    data = b''
+
+    for data1, data2, data3, data4 in chids:
+        data += struct.pack(
+            '<IIIIHH8B', DEVICE_STRUCT_SIZE, offsets[name], offsets[compatible], data1, data2, data3, *data4
+        )
+
+    assert len(data) == DEVICE_STRUCT_SIZE * len(chids)
+    return data
+
+
+def hex_pairs_list(string: str) -> list[int]:
+    return [int(string[i : i + 2], 16) for i in range(0, len(string), 2)]
+
+
+def pack_strings(strings: set[str], base: int) -> tuple[bytes, dict[str, int]]:
+    blob = b''
+    offsets = {}
+
+    for string in sorted(strings):
+        offsets[string] = base + len(blob)
+        blob += string.encode('utf-8') + b'\00'
+
+    return (blob, offsets)
+
+
+def parse_hwid_dir(path: Path) -> bytes:
+    hwid_files = path.rglob('*.txt')
+
+    strings: set[str] = set()
+    devices: dict[tuple[str, str], set[EFI_GUID]] = dict()
+
+    uuid_regexp = re.compile(
+        '\\{[0-9a-f]{8}-[0-9a-f]{4}-[0-5][0-9a-f]{3}-[089ab][0-9a-f]{3}-[0-9a-f]{12}\\}', re.I
+    )
+
+    for hwid_file in hwid_files:
+        content = []
+        with open(hwid_file) as f:
+            content = f.readlines()
+
+        data: dict[str, str] = {
+            'Manufacturer': '',
+            'Family': '',
+            'Compatible': '',
+        }
+        uuids: set[EFI_GUID] = set()
+
+        for line in content:
+            for k in data:
+                if line.startswith(k):
+                    data[k] = line.split(':')[1].strip()
+                    break
+            else:
+                uuid = uuid_regexp.match(line)
+                if uuid is not None:
+                    d1, d2, d3, d4, d5 = uuid.group(0)[1:-1].split('-')
+
+                    data1 = int(d1, 16)
+                    data2 = int(d2, 16)
+                    data3 = int(d3, 16)
+                    data4 = cast(
+                        tuple[int, int, int, int, int, int, int, int],
+                        tuple(hex_pairs_list(d4) + hex_pairs_list(d5)),
+                    )
+
+                    uuids |= set([(data1, data2, data3, data4)])
+
+        for k, v in data.items():
+            if not v:
+                raise ValueError(f'hwid description file "{hwid_file}" does not contain "{k}"')
+
+        name = data['Manufacturer'] + ' ' + data['Family']
+        compatible = data['Compatible']
+
+        strings |= set([name, compatible])
+
+        # (compatible, name) pair uniquely identifies the device
+        key = (compatible, name)
+
+        if key not in devices:
+            devices[key] = set()
+        devices[key] |= uuids
+
+    total_device_structs = 1
+    for dev, uuids in devices.items():
+        total_device_structs += len(uuids)
+
+    strings_blob, offsets = pack_strings(strings, total_device_structs * DEVICE_STRUCT_SIZE)
+
+    devices_blob = b''
+    for (compatible, name), uuids in devices.items():
+        devices_blob += pack_device(offsets, name, compatible, uuids)
+
+    devices_blob += NULL_DEVICE
+
+    return devices_blob + strings_blob
+
+
 STUB_SBAT = """\
 sbat,1,SBAT Version,sbat,1,https://github.com/rhboot/shim/blob/main/SBAT.md
 uki,1,UKI,uki,1,https://uapi-group.org/specifications/specs/unified_kernel_image/
@@ -934,7 +1072,6 @@ uki-addon,1,UKI Addon,addon,1,https://www.freedesktop.org/software/systemd/man/l
 
 def make_uki(opts: UkifyConfig) -> None:
     assert opts.output is not None
-
     # kernel payload signing
 
     sign_args_present = opts.sb_key or opts.sb_cert_name
@@ -992,11 +1129,18 @@ def make_uki(opts: UkifyConfig) -> None:
                 format=serialization.PublicFormat.SubjectPublicKeyInfo,
             )
 
+    hwids = None
+
+    if opts.hwids_directory is not None:
+        hwids = parse_hwid_dir(opts.hwids_directory)
+
     sections = [
         # name,      content,         measure?
         ('.osrel',   opts.os_release, True),
         ('.cmdline', opts.cmdline,    True),
         ('.dtb',     opts.devicetree, True),
+        *(('.dtbauto', dtb, True) for dtb in opts.devicetree_auto),
+        ('.hwids',   hwids,           True),
         ('.uname',   opts.uname,      True),
         ('.splash',  opts.splash,     True),
         ('.pcrpkey', pcrpkey,         True),
@@ -1440,10 +1584,10 @@ class ConfigItem:
         else:
             conv = lambda s: s  # noqa: E731
 
-        # This is a bit ugly, but --initrd is the only option which is specified
+        # This is a bit ugly, but --initrd and --devicetree-auto are the only options
         # with multiple args on the command line and a space-separated list in the
         # config file.
-        if self.name == '--initrd':
+        if self.name in ['--initrd', '--devicetree-auto']:
             value = [conv(v) for v in value.split()]
         else:
             value = conv(value)
@@ -1554,6 +1698,24 @@ CONFIG_ITEMS = [
         help='Device Tree file [.dtb section]',
         config_key='UKI/DeviceTree',
     ),
+    ConfigItem(
+        '--devicetree-auto',
+        metavar='PATH',
+        type=Path,
+        action='append',
+        help='Device Tree file for automatic selection [.dtbauto section]',
+        default=[],
+        config_key='UKI/DeviceTreeAuto',
+        config_push=ConfigItem.config_list_prepend,
+    ),
+    ConfigItem(
+        '--hwids-directory',
+        metavar='PATH',
+        type=Path,
+        action='append',
+        help='Directory with HWID text files [.hwids section]',
+        config_key='UKI/HWIDsDirectory',
+    ),
     ConfigItem(
         '--uname',
         metavar='VERSION',