src/ukify/ukify.py

   1 #!/usr/bin/env python3
   2 # SPDX-License-Identifier: LGPL-2.1-or-later
   3 #
   4 # This file is part of systemd.
   5 #
   6 # systemd is free software; you can redistribute it and/or modify it
   7 # under the terms of the GNU Lesser General Public License as published by
   8 # the Free Software Foundation; either version 2.1 of the License, or
   9 # (at your option) any later version.
  10 #
  11 # systemd is distributed in the hope that it will be useful, but
  12 # WITHOUT ANY WARRANTY; without even the implied warranty of
  13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  14 # General Public License for more details.
  15 #
  16 # You should have received a copy of the GNU Lesser General Public License
  17 # along with systemd; If not, see <https://www.gnu.org/licenses/>.
  18
  19 # pylint: disable=missing-docstring,invalid-name,import-outside-toplevel
  20 # pylint: disable=consider-using-with,unspecified-encoding,line-too-long
  21 # pylint: disable=too-many-locals,too-many-statements,too-many-return-statements
  22 # pylint: disable=too-many-branches,fixme
  23
  24 import argparse
  25 import configparser
  26 import collections
  27 import dataclasses
  28 import fnmatch
  29 import itertools
  30 import json
  31 import os
  32 import pathlib
  33 import pprint
  34 import re
  35 import shlex
  36 import shutil
  37 import subprocess
  38 import sys
  39 import tempfile
  40 from typing import (Any,
  41                     Callable,
  42                     IO,
  43                     Optional,
  44                     Union)
  45
  46 import pefile  # type: ignore
  47
  48 __version__ = '{{PROJECT_VERSION}} ({{GIT_VERSION}})'
  49
  50 EFI_ARCH_MAP = {
  51     # host_arch glob : [efi_arch, 32_bit_efi_arch if mixed mode is supported]
  52     'x86_64'       : ['x64', 'ia32'],
  53     'i[3456]86'    : ['ia32'],
  54     'aarch64'      : ['aa64'],
  55     'arm[45678]*l' : ['arm'],
  56     'loongarch32'  : ['loongarch32'],
  57     'loongarch64'  : ['loongarch64'],
  58     'riscv32'      : ['riscv32'],
  59     'riscv64'      : ['riscv64'],
  60 }
  61 EFI_ARCHES: list[str] = sum(EFI_ARCH_MAP.values(), [])
  62
  63 def guess_efi_arch():
  64     arch = os.uname().machine
  65
  66     for glob, mapping in EFI_ARCH_MAP.items():
  67         if fnmatch.fnmatch(arch, glob):
  68             efi_arch, *fallback = mapping
  69             break
  70     else:
  71         raise ValueError(f'Unsupported architecture {arch}')
  72
  73     # This makes sense only on some architectures, but it also probably doesn't
  74     # hurt on others, so let's just apply the check everywhere.
  75     if fallback:
  76         fw_platform_size = pathlib.Path('/sys/firmware/efi/fw_platform_size')
  77         try:
  78             size = fw_platform_size.read_text().strip()
  79         except FileNotFoundError:
  80             pass
  81         else:
  82             if int(size) == 32:
  83                 efi_arch = fallback[0]
  84
  85     print(f'Host arch {arch!r}, EFI arch {efi_arch!r}')
  86     return efi_arch
  87
  88
  89 def shell_join(cmd):
  90     # TODO: drop in favour of shlex.join once shlex.join supports pathlib.Path.
  91     return ' '.join(shlex.quote(str(x)) for x in cmd)
  92
  93
  94 def round_up(x, blocksize=4096):
  95     return (x + blocksize - 1) // blocksize * blocksize
  96
  97
  98 def try_import(modname, name=None):
  99     try:
 100         return __import__(modname)
 101     except ImportError as e:
 102         raise ValueError(f'Kernel is compressed with {name or modname}, but module unavailable') from e
 103
 104
 105 def maybe_decompress(filename):
 106     """Decompress file if compressed. Return contents."""
 107     f = open(filename, 'rb')
 108     start = f.read(4)
 109     f.seek(0)
 110
 111     if start.startswith(b'\x7fELF'):
 112         # not compressed
 113         return f.read()
 114
 115     if start.startswith(b'MZ'):
 116         # not compressed aarch64 and riscv64
 117         return f.read()
 118
 119     if start.startswith(b'\x1f\x8b'):
 120         gzip = try_import('gzip')
 121         return gzip.open(f).read()
 122
 123     if start.startswith(b'\x28\xb5\x2f\xfd'):
 124         zstd = try_import('zstd')
 125         return zstd.uncompress(f.read())
 126
 127     if start.startswith(b'\x02\x21\x4c\x18'):
 128         lz4 = try_import('lz4.frame', 'lz4')
 129         return lz4.frame.decompress(f.read())
 130
 131     if start.startswith(b'\x04\x22\x4d\x18'):
 132         print('Newer lz4 stream format detected! This may not boot!')
 133         lz4 = try_import('lz4.frame', 'lz4')
 134         return lz4.frame.decompress(f.read())
 135
 136     if start.startswith(b'\x89LZO'):
 137         # python3-lzo is not packaged for Fedora
 138         raise NotImplementedError('lzo decompression not implemented')
 139
 140     if start.startswith(b'BZh'):
 141         bz2 = try_import('bz2', 'bzip2')
 142         return bz2.open(f).read()
 143
 144     if start.startswith(b'\x5d\x00\x00'):
 145         lzma = try_import('lzma')
 146         return lzma.open(f).read()
 147
 148     raise NotImplementedError(f'unknown file format (starts with {start})')
 149
 150
 151 class Uname:
 152     # This class is here purely as a namespace for the functions
 153
 154     VERSION_PATTERN = r'(?P<version>[a-z0-9._-]+) \([^ )]+\) (?:#.*)'
 155
 156     NOTES_PATTERN = r'^\s+Linux\s+0x[0-9a-f]+\s+OPEN\n\s+description data: (?P<version>[0-9a-f ]+)\s*$'
 157
 158     # Linux version 6.0.8-300.fc37.ppc64le (mockbuild@buildvm-ppc64le-03.iad2.fedoraproject.org)
 159     # (gcc (GCC) 12.2.1 20220819 (Red Hat 12.2.1-2), GNU ld version 2.38-24.fc37)
 160     # #1 SMP Fri Nov 11 14:39:11 UTC 2022
 161     TEXT_PATTERN = rb'Linux version (?P<version>\d\.\S+) \('
 162
 163     @classmethod
 164     def scrape_x86(cls, filename, opts=None):
 165         # Based on https://gitlab.archlinux.org/archlinux/mkinitcpio/mkinitcpio/-/blob/master/functions#L136
 166         # and https://www.kernel.org/doc/html/latest/x86/boot.html#the-real-mode-kernel-header
 167         with open(filename, 'rb') as f:
 168             f.seek(0x202)
 169             magic = f.read(4)
 170             if magic != b'HdrS':
 171                 raise ValueError('Real-Mode Kernel Header magic not found')
 172             f.seek(0x20E)
 173             offset = f.read(1)[0] + f.read(1)[0]*256  # Pointer to kernel version string
 174             f.seek(0x200 + offset)
 175             text = f.read(128)
 176         text = text.split(b'\0', maxsplit=1)[0]
 177         text = text.decode()
 178
 179         if not (m := re.match(cls.VERSION_PATTERN, text)):
 180             raise ValueError(f'Cannot parse version-host-release uname string: {text!r}')
 181         return m.group('version')
 182
 183     @classmethod
 184     def scrape_elf(cls, filename, opts=None):
 185         readelf = find_tool('readelf', opts=opts)
 186
 187         cmd = [
 188             readelf,
 189             '--notes',
 190             filename,
 191         ]
 192
 193         print('+', shell_join(cmd))
 194         try:
 195             notes = subprocess.check_output(cmd, stderr=subprocess.PIPE, text=True)
 196         except subprocess.CalledProcessError as e:
 197             raise ValueError(e.stderr.strip()) from e
 198
 199         if not (m := re.search(cls.NOTES_PATTERN, notes, re.MULTILINE)):
 200             raise ValueError('Cannot find Linux version note')
 201
 202         text = ''.join(chr(int(c, 16)) for c in m.group('version').split())
 203         return text.rstrip('\0')
 204
 205     @classmethod
 206     def scrape_generic(cls, filename, opts=None):
 207         # import libarchive
 208         # libarchive-c fails with
 209         # ArchiveError: Unrecognized archive format (errno=84, retcode=-30, archive_p=94705420454656)
 210
 211         # Based on https://gitlab.archlinux.org/archlinux/mkinitcpio/mkinitcpio/-/blob/master/functions#L209
 212
 213         text = maybe_decompress(filename)
 214         if not (m := re.search(cls.TEXT_PATTERN, text)):
 215             raise ValueError(f'Cannot find {cls.TEXT_PATTERN!r} in {filename}')
 216
 217         return m.group('version').decode()
 218
 219     @classmethod
 220     def scrape(cls, filename, opts=None):
 221         for func in (cls.scrape_x86, cls.scrape_elf, cls.scrape_generic):
 222             try:
 223                 version = func(filename, opts=opts)
 224                 print(f'Found uname version: {version}')
 225                 return version
 226             except ValueError as e:
 227                 print(str(e))
 228         return None
 229
 230
 231 @dataclasses.dataclass
 232 class Section:
 233     name: str
 234     content: pathlib.Path
 235     tmpfile: Optional[IO] = None
 236     measure: bool = False
 237
 238     @classmethod
 239     def create(cls, name, contents, **kwargs):
 240         if isinstance(contents, (str, bytes)):
 241             mode = 'wt' if isinstance(contents, str) else 'wb'
 242             tmp = tempfile.NamedTemporaryFile(mode=mode, prefix=f'tmp{name}')
 243             tmp.write(contents)
 244             tmp.flush()
 245             contents = pathlib.Path(tmp.name)
 246         else:
 247             tmp = None
 248
 249         return cls(name, contents, tmpfile=tmp, **kwargs)
 250
 251     @classmethod
 252     def parse_arg(cls, s):
 253         try:
 254             name, contents, *rest = s.split(':')
 255         except ValueError as e:
 256             raise ValueError(f'Cannot parse section spec (name or contents missing): {s!r}') from e
 257         if rest:
 258             raise ValueError(f'Cannot parse section spec (extraneous parameters): {s!r}')
 259
 260         if contents.startswith('@'):
 261             contents = pathlib.Path(contents[1:])
 262
 263         return cls.create(name, contents)
 264
 265     def size(self):
 266         return self.content.stat().st_size
 267
 268     def check_name(self):
 269         # PE section names with more than 8 characters are legal, but our stub does
 270         # not support them.
 271         if not self.name.isascii() or not self.name.isprintable():
 272             raise ValueError(f'Bad section name: {self.name!r}')
 273         if len(self.name) > 8:
 274             raise ValueError(f'Section name too long: {self.name!r}')
 275
 276
 277 @dataclasses.dataclass
 278 class UKI:
 279     executable: list[Union[pathlib.Path, str]]
 280     sections: list[Section] = dataclasses.field(default_factory=list, init=False)
 281
 282     def add_section(self, section):
 283         if section.name in [s.name for s in self.sections]:
 284             raise ValueError(f'Duplicate section {section.name}')
 285
 286         self.sections += [section]
 287
 288
 289 def parse_banks(s):
 290     banks = re.split(r',|\s+', s)
 291     # TODO: do some sanity checking here
 292     return banks
 293
 294
 295 KNOWN_PHASES = (
 296     'enter-initrd',
 297     'leave-initrd',
 298     'sysinit',
 299     'ready',
 300     'shutdown',
 301     'final',
 302 )
 303
 304 def parse_phase_paths(s):
 305     # Split on commas or whitespace here. Commas might be hard to parse visually.
 306     paths = re.split(r',|\s+', s)
 307
 308     for path in paths:
 309         for phase in path.split(':'):
 310             if phase not in KNOWN_PHASES:
 311                 raise argparse.ArgumentTypeError(f'Unknown boot phase {phase!r} ({path=})')
 312
 313     return paths
 314
 315
 316 def check_splash(filename):
 317     if filename is None:
 318         return
 319
 320     # import is delayed, to avoid import when the splash image is not used
 321     try:
 322         from PIL import Image
 323     except ImportError:
 324         return
 325
 326     img = Image.open(filename, formats=['BMP'])
 327     print(f'Splash image {filename} is {img.width}×{img.height} pixels')
 328
 329
 330 def check_inputs(opts):
 331     for name, value in vars(opts).items():
 332         if name in {'output', 'tools'}:
 333             continue
 334
 335         if isinstance(value, pathlib.Path):
 336             # Open file to check that we can read it, or generate an exception
 337             value.open().close()
 338         elif isinstance(value, list):
 339             for item in value:
 340                 if isinstance(item, pathlib.Path):
 341                     item.open().close()
 342
 343     check_splash(opts.splash)
 344
 345
 346 def find_tool(name, fallback=None, opts=None):
 347     if opts and opts.tools:
 348         for d in opts.tools:
 349             tool = d / name
 350             if tool.exists():
 351                 return tool
 352
 353     if shutil.which(name) is not None:
 354         return name
 355
 356     if fallback is None:
 357         print(f"Tool {name} not installed!")
 358
 359     return fallback
 360
 361 def combine_signatures(pcrsigs):
 362     combined = collections.defaultdict(list)
 363     for pcrsig in pcrsigs:
 364         for bank, sigs in pcrsig.items():
 365             for sig in sigs:
 366                 if sig not in combined[bank]:
 367                     combined[bank] += [sig]
 368     return json.dumps(combined)
 369
 370
 371 def call_systemd_measure(uki, linux, opts):
 372     measure_tool = find_tool('systemd-measure',
 373                              '/usr/lib/systemd/systemd-measure',
 374                              opts=opts)
 375
 376     banks = opts.pcr_banks or ()
 377
 378     # PCR measurement
 379
 380     if opts.measure:
 381         pp_groups = opts.phase_path_groups or []
 382
 383         cmd = [
 384             measure_tool,
 385             'calculate',
 386             f'--linux={linux}',
 387             *(f"--{s.name.removeprefix('.')}={s.content}"
 388               for s in uki.sections
 389               if s.measure),
 390             *(f'--bank={bank}'
 391               for bank in banks),
 392             # For measurement, the keys are not relevant, so we can lump all the phase paths
 393             # into one call to systemd-measure calculate.
 394             *(f'--phase={phase_path}'
 395               for phase_path in itertools.chain.from_iterable(pp_groups)),
 396         ]
 397
 398         print('+', shell_join(cmd))
 399         subprocess.check_call(cmd)
 400
 401     # PCR signing
 402
 403     if opts.pcr_private_keys:
 404         n_priv = len(opts.pcr_private_keys or ())
 405         pp_groups = opts.phase_path_groups or [None] * n_priv
 406         pub_keys = opts.pcr_public_keys or [None] * n_priv
 407
 408         pcrsigs = []
 409
 410         cmd = [
 411             measure_tool,
 412             'sign',
 413             f'--linux={linux}',
 414             *(f"--{s.name.removeprefix('.')}={s.content}"
 415               for s in uki.sections
 416               if s.measure),
 417             *(f'--bank={bank}'
 418               for bank in banks),
 419         ]
 420
 421         for priv_key, pub_key, group in zip(opts.pcr_private_keys,
 422                                             pub_keys,
 423                                             pp_groups):
 424             extra = [f'--private-key={priv_key}']
 425             if pub_key:
 426                 extra += [f'--public-key={pub_key}']
 427             extra += [f'--phase={phase_path}' for phase_path in group or ()]
 428
 429             print('+', shell_join(cmd + extra))
 430             pcrsig = subprocess.check_output(cmd + extra, text=True)
 431             pcrsig = json.loads(pcrsig)
 432             pcrsigs += [pcrsig]
 433
 434         combined = combine_signatures(pcrsigs)
 435         uki.add_section(Section.create('.pcrsig', combined))
 436
 437
 438 def join_initrds(initrds):
 439     if len(initrds) == 0:
 440         return None
 441     elif len(initrds) == 1:
 442         return initrds[0]
 443
 444     seq = []
 445     for file in initrds:
 446         initrd = file.read_bytes()
 447         n = len(initrd)
 448         padding = b'\0' * (round_up(n, 4) - n)  # pad to 32 bit alignment
 449         seq += [initrd, padding]
 450
 451     return b''.join(seq)
 452
 453
 454 def pairwise(iterable):
 455     a, b = itertools.tee(iterable)
 456     next(b, None)
 457     return zip(a, b)
 458
 459
 460 class PEError(Exception):
 461     pass
 462
 463
 464 def pe_add_sections(uki: UKI, output: str):
 465     pe = pefile.PE(uki.executable, fast_load=True)
 466
 467     # Old stubs do not have the symbol/string table stripped, even though image files should not have one.
 468     if symbol_table := pe.FILE_HEADER.PointerToSymbolTable:
 469         symbol_table_size = 18 * pe.FILE_HEADER.NumberOfSymbols
 470         if string_table_size := pe.get_dword_from_offset(symbol_table + symbol_table_size):
 471             symbol_table_size += string_table_size
 472
 473         # Let's be safe and only strip it if it's at the end of the file.
 474         if symbol_table + symbol_table_size == len(pe.__data__):
 475             pe.__data__ = pe.__data__[:symbol_table]
 476             pe.FILE_HEADER.PointerToSymbolTable = 0
 477             pe.FILE_HEADER.NumberOfSymbols = 0
 478             pe.FILE_HEADER.IMAGE_FILE_LOCAL_SYMS_STRIPPED = True
 479
 480     # Old stubs might have been stripped, leading to unaligned raw data values, so let's fix them up here.
 481     for i, section in enumerate(pe.sections):
 482         oldp = section.PointerToRawData
 483         oldsz = section.SizeOfRawData
 484         section.PointerToRawData = round_up(oldp, pe.OPTIONAL_HEADER.FileAlignment)
 485         section.SizeOfRawData = round_up(oldsz, pe.OPTIONAL_HEADER.FileAlignment)
 486         padp = section.PointerToRawData - oldp
 487         padsz = section.SizeOfRawData - oldsz
 488
 489         for later_section in pe.sections[i+1:]:
 490             later_section.PointerToRawData += padp + padsz
 491
 492         pe.__data__ = pe.__data__[:oldp] + bytes(padp) + pe.__data__[oldp:oldp+oldsz] + bytes(padsz) + pe.__data__[oldp+oldsz:]
 493
 494     # We might not have any space to add new sections. Let's try our best to make some space by padding the
 495     # SizeOfHeaders to a multiple of the file alignment. This is safe because the first section's data starts
 496     # at a multiple of the file alignment, so all space before that is unused.
 497     pe.OPTIONAL_HEADER.SizeOfHeaders = round_up(pe.OPTIONAL_HEADER.SizeOfHeaders, pe.OPTIONAL_HEADER.FileAlignment)
 498     pe = pefile.PE(data=pe.write(), fast_load=True)
 499
 500     warnings = pe.get_warnings()
 501     if warnings:
 502         raise PEError(f'pefile warnings treated as errors: {warnings}')
 503
 504     security = pe.OPTIONAL_HEADER.DATA_DIRECTORY[pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_SECURITY']]
 505     if security.VirtualAddress != 0:
 506         # We could strip the signatures, but why would anyone sign the stub?
 507         raise PEError('Stub image is signed, refusing.')
 508
 509     for section in uki.sections:
 510         new_section = pefile.SectionStructure(pe.__IMAGE_SECTION_HEADER_format__, pe=pe)
 511         new_section.__unpack__(b'\0' * new_section.sizeof())
 512
 513         offset = pe.sections[-1].get_file_offset() + new_section.sizeof()
 514         if offset + new_section.sizeof() > pe.OPTIONAL_HEADER.SizeOfHeaders:
 515             raise PEError(f'Not enough header space to add section {section.name}.')
 516
 517         data = section.content.read_bytes()
 518
 519         new_section.set_file_offset(offset)
 520         new_section.Name = section.name.encode()
 521         new_section.Misc_VirtualSize = len(data)
 522         # Non-stripped stubs might still have an unaligned symbol table at the end, making their size
 523         # unaligned, so we make sure to explicitly pad the pointer to new sections to an aligned offset.
 524         new_section.PointerToRawData = round_up(len(pe.__data__), pe.OPTIONAL_HEADER.FileAlignment)
 525         new_section.SizeOfRawData = round_up(len(data), pe.OPTIONAL_HEADER.FileAlignment)
 526         new_section.VirtualAddress = round_up(
 527             pe.sections[-1].VirtualAddress + pe.sections[-1].Misc_VirtualSize,
 528             pe.OPTIONAL_HEADER.SectionAlignment,
 529         )
 530
 531         new_section.IMAGE_SCN_MEM_READ = True
 532         if section.name == '.linux':
 533             # Old kernels that use EFI handover protocol will be executed inline.
 534             new_section.IMAGE_SCN_CNT_CODE = True
 535         else:
 536             new_section.IMAGE_SCN_CNT_INITIALIZED_DATA = True
 537
 538         pe.__data__ = pe.__data__[:] + bytes(new_section.PointerToRawData - len(pe.__data__)) + data + bytes(new_section.SizeOfRawData - len(data))
 539
 540         pe.FILE_HEADER.NumberOfSections += 1
 541         pe.OPTIONAL_HEADER.SizeOfInitializedData += new_section.Misc_VirtualSize
 542         pe.__structures__.append(new_section)
 543         pe.sections.append(new_section)
 544
 545     pe.OPTIONAL_HEADER.CheckSum = 0
 546     pe.OPTIONAL_HEADER.SizeOfImage = round_up(
 547         pe.sections[-1].VirtualAddress + pe.sections[-1].Misc_VirtualSize,
 548         pe.OPTIONAL_HEADER.SectionAlignment,
 549     )
 550
 551     pe.write(output)
 552
 553 def signer_sign(cmd):
 554     print('+', shell_join(cmd))
 555     subprocess.check_call(cmd)
 556
 557 def find_sbsign(opts=None):
 558     return find_tool('sbsign', opts=opts)
 559
 560 def sbsign_sign(sbsign_tool, input_f, output_f, opts=None):
 561     sign_invocation = [
 562         sbsign_tool,
 563         '--key', opts.sb_key,
 564         '--cert', opts.sb_cert,
 565         input_f,
 566         '--output', output_f,
 567     ]
 568     if opts.signing_engine is not None:
 569         sign_invocation += ['--engine', opts.signing_engine]
 570     signer_sign(sign_invocation)
 571
 572 def find_pesign(opts=None):
 573     return find_tool('pesign', opts=opts)
 574
 575 def pesign_sign(pesign_tool, input_f, output_f, opts=None):
 576     sign_invocation = [
 577         pesign_tool, '-s', '--force',
 578         '-n', opts.sb_certdir,
 579         '-c', opts.sb_cert_name,
 580         '-i', input_f,
 581         '-o', output_f,
 582     ]
 583     signer_sign(sign_invocation)
 584
 585 SBVERIFY = {
 586     'name': 'sbverify',
 587     'option': '--list',
 588     'output': 'No signature table present',
 589 }
 590
 591 PESIGCHECK = {
 592     'name': 'pesign',
 593     'option': '-i',
 594     'output': 'No signatures found.',
 595     'flags': '-S'
 596 }
 597
 598 def verify(tool, opts):
 599     verify_tool = find_tool(tool['name'], opts=opts)
 600     cmd = [
 601         verify_tool,
 602         tool['option'],
 603         opts.linux,
 604     ]
 605     if 'flags' in tool:
 606         cmd.append(tool['flags'])
 607
 608     print('+', shell_join(cmd))
 609     info = subprocess.check_output(cmd, text=True)
 610
 611     return tool['output'] in info
 612
 613 def make_uki(opts):
 614     # kernel payload signing
 615
 616     sign_tool = None
 617     if opts.signtool == 'sbsign':
 618         sign_tool = find_sbsign(opts=opts)
 619         sign = sbsign_sign
 620         verify_tool = SBVERIFY
 621     else:
 622         sign_tool = find_pesign(opts=opts)
 623         sign = pesign_sign
 624         verify_tool = PESIGCHECK
 625
 626     sign_args_present = opts.sb_key or opts.sb_cert_name
 627
 628     if sign_tool is None and sign_args_present:
 629         raise ValueError(f'{opts.signtool}, required for signing, is not installed')
 630
 631     sign_kernel = opts.sign_kernel
 632     if sign_kernel is None and opts.linux is not None and sign_args_present:
 633         # figure out if we should sign the kernel
 634         sign_kernel = verify(verify_tool, opts)
 635
 636     if sign_kernel:
 637         linux_signed = tempfile.NamedTemporaryFile(prefix='linux-signed')
 638         linux = pathlib.Path(linux_signed.name)
 639         sign(sign_tool, opts.linux, linux, opts=opts)
 640     else:
 641         linux = opts.linux
 642
 643     if opts.uname is None and opts.linux is not None:
 644         print('Kernel version not specified, starting autodetection 😖.')
 645         opts.uname = Uname.scrape(opts.linux, opts=opts)
 646
 647     uki = UKI(opts.stub)
 648     initrd = join_initrds(opts.initrd)
 649
 650     # TODO: derive public key from opts.pcr_private_keys?
 651     pcrpkey = opts.pcrpkey
 652     if pcrpkey is None:
 653         if opts.pcr_public_keys and len(opts.pcr_public_keys) == 1:
 654             pcrpkey = opts.pcr_public_keys[0]
 655
 656     sections = [
 657         # name,      content,         measure?
 658         ('.osrel',   opts.os_release, True ),
 659         ('.cmdline', opts.cmdline,    True ),
 660         ('.dtb',     opts.devicetree, True ),
 661         ('.uname',   opts.uname,      True ),
 662         ('.splash',  opts.splash,     True ),
 663         ('.pcrpkey', pcrpkey,         True ),
 664         ('.initrd',  initrd,          True ),
 665
 666         # linux shall be last to leave breathing room for decompression.
 667         # We'll add it later.
 668     ]
 669
 670     for name, content, measure in sections:
 671         if content:
 672             uki.add_section(Section.create(name, content, measure=measure))
 673
 674     # systemd-measure doesn't know about those extra sections
 675     for section in opts.sections:
 676         uki.add_section(section)
 677
 678     # PCR measurement and signing
 679
 680     call_systemd_measure(uki, linux, opts=opts)
 681
 682     # UKI or addon creation - addons don't use the stub so we add SBAT manually
 683
 684     if linux is not None:
 685         uki.add_section(Section.create('.linux', linux, measure=True))
 686     elif opts.sbat:
 687         uki.add_section(Section.create('.sbat', opts.sbat, measure=False))
 688
 689     if sign_args_present:
 690         unsigned = tempfile.NamedTemporaryFile(prefix='uki')
 691         output = unsigned.name
 692     else:
 693         output = opts.output
 694
 695     pe_add_sections(uki, output)
 696
 697     # UKI signing
 698
 699     if sign_args_present:
 700         sign(sign_tool, unsigned.name, opts.output, opts=opts)
 701
 702         # We end up with no executable bits, let's reapply them
 703         os.umask(umask := os.umask(0))
 704         os.chmod(opts.output, 0o777 & ~umask)
 705
 706     print(f"Wrote {'signed' if sign_args_present else 'unsigned'} {opts.output}")
 707
 708
 709 @dataclasses.dataclass(frozen=True)
 710 class ConfigItem:
 711     @staticmethod
 712     def config_list_prepend(
 713             namespace: argparse.Namespace,
 714             group: Optional[str],
 715             dest: str,
 716             value: Any,
 717     ) -> None:
 718         "Prepend value to namespace.<dest>"
 719
 720         assert not group
 721
 722         old = getattr(namespace, dest, [])
 723         setattr(namespace, dest, value + old)
 724
 725     @staticmethod
 726     def config_set_if_unset(
 727             namespace: argparse.Namespace,
 728             group: Optional[str],
 729             dest: str,
 730             value: Any,
 731     ) -> None:
 732         "Set namespace.<dest> to value only if it was None"
 733
 734         assert not group
 735
 736         if getattr(namespace, dest) is None:
 737             setattr(namespace, dest, value)
 738
 739     @staticmethod
 740     def config_set_group(
 741             namespace: argparse.Namespace,
 742             group: Optional[str],
 743             dest: str,
 744             value: Any,
 745     ) -> None:
 746         "Set namespace.<dest>[idx] to value, with idx derived from group"
 747
 748         if group not in namespace._groups:
 749             namespace._groups += [group]
 750         idx = namespace._groups.index(group)
 751
 752         old = getattr(namespace, dest, None)
 753         if old is None:
 754             old = []
 755         setattr(namespace, dest,
 756                 old + ([None] * (idx - len(old))) + [value])
 757
 758     @staticmethod
 759     def parse_boolean(s: str) -> bool:
 760         "Parse 1/true/yes/y/t/on as true and 0/false/no/n/f/off/None as false"
 761         s_l = s.lower()
 762         if s_l in {'1', 'true', 'yes', 'y', 't', 'on'}:
 763             return True
 764         if s_l in {'0', 'false', 'no', 'n', 'f', 'off'}:
 765             return False
 766         raise ValueError('f"Invalid boolean literal: {s!r}')
 767
 768     # arguments for argparse.ArgumentParser.add_argument()
 769     name: Union[str, tuple[str, str]]
 770     dest: Optional[str] = None
 771     metavar: Optional[str] = None
 772     type: Optional[Callable] = None
 773     nargs: Optional[str] = None
 774     action: Optional[Union[str, Callable]] = None
 775     default: Any = None
 776     version: Optional[str] = None
 777     choices: Optional[tuple[str, ...]] = None
 778     help: Optional[str] = None
 779
 780     # metadata for config file parsing
 781     config_key: Optional[str] = None
 782     config_push: Callable[[argparse.Namespace, Optional[str], str, Any], None] = \
 783                     config_set_if_unset
 784
 785     def _names(self) -> tuple[str, ...]:
 786         return self.name if isinstance(self.name, tuple) else (self.name,)
 787
 788     def argparse_dest(self) -> str:
 789         # It'd be nice if argparse exported this, but I don't see that in the API
 790         if self.dest:
 791             return self.dest
 792         return self._names()[0].lstrip('-').replace('-', '_')
 793
 794     def add_to(self, parser: argparse.ArgumentParser):
 795         kwargs = { key:val
 796                    for key in dataclasses.asdict(self)
 797                    if (key not in ('name', 'config_key', 'config_push') and
 798                        (val := getattr(self, key)) is not None) }
 799         args = self._names()
 800         parser.add_argument(*args, **kwargs)
 801
 802     def apply_config(self, namespace, section, group, key, value) -> None:
 803         assert f'{section}/{key}' == self.config_key
 804         dest = self.argparse_dest()
 805
 806         conv: Callable[[str], Any]
 807         if self.action == argparse.BooleanOptionalAction:
 808             # We need to handle this case separately: the options are called
 809             # --foo and --no-foo, and no argument is parsed. But in the config
 810             # file, we have Foo=yes or Foo=no.
 811             conv = self.parse_boolean
 812         elif self.type:
 813             conv = self.type
 814         else:
 815             conv = lambda s:s
 816
 817         if self.nargs == '*':
 818             value = [conv(v) for v in value.split()]
 819         else:
 820             value = conv(value)
 821
 822         self.config_push(namespace, group, dest, value)
 823
 824     def config_example(self) -> tuple[Optional[str], Optional[str], Optional[str]]:
 825         if not self.config_key:
 826             return None, None, None
 827         section_name, key = self.config_key.split('/', 1)
 828         if section_name.endswith(':'):
 829             section_name += 'NAME'
 830         if self.choices:
 831             value = '|'.join(self.choices)
 832         else:
 833             value = self.metavar or self.argparse_dest().upper()
 834         return (section_name, key, value)
 835
 836
 837 CONFIG_ITEMS = [
 838     ConfigItem(
 839         '--version',
 840         action = 'version',
 841         version = f'ukify {__version__}',
 842     ),
 843
 844     ConfigItem(
 845         '--summary',
 846         help = 'print parsed config and exit',
 847         action = 'store_true',
 848     ),
 849
 850     ConfigItem(
 851         'linux',
 852         metavar = 'LINUX',
 853         type = pathlib.Path,
 854         nargs = '?',
 855         help = 'vmlinuz file [.linux section]',
 856         config_key = 'UKI/Linux',
 857     ),
 858
 859     ConfigItem(
 860         'initrd',
 861         metavar = 'INITRD…',
 862         type = pathlib.Path,
 863         nargs = '*',
 864         help = 'initrd files [.initrd section]',
 865         config_key = 'UKI/Initrd',
 866         config_push = ConfigItem.config_list_prepend,
 867     ),
 868
 869     ConfigItem(
 870         ('--config', '-c'),
 871         metavar = 'PATH',
 872         help = 'configuration file',
 873     ),
 874
 875     ConfigItem(
 876         '--cmdline',
 877         metavar = 'TEXT|@PATH',
 878         help = 'kernel command line [.cmdline section]',
 879         config_key = 'UKI/Cmdline',
 880     ),
 881
 882     ConfigItem(
 883         '--os-release',
 884         metavar = 'TEXT|@PATH',
 885         help = 'path to os-release file [.osrel section]',
 886         config_key = 'UKI/OSRelease',
 887     ),
 888
 889     ConfigItem(
 890         '--devicetree',
 891         metavar = 'PATH',
 892         type = pathlib.Path,
 893         help = 'Device Tree file [.dtb section]',
 894         config_key = 'UKI/DeviceTree',
 895     ),
 896     ConfigItem(
 897         '--splash',
 898         metavar = 'BMP',
 899         type = pathlib.Path,
 900         help = 'splash image bitmap file [.splash section]',
 901         config_key = 'UKI/Splash',
 902     ),
 903     ConfigItem(
 904         '--pcrpkey',
 905         metavar = 'KEY',
 906         type = pathlib.Path,
 907         help = 'embedded public key to seal secrets to [.pcrpkey section]',
 908         config_key = 'UKI/PCRPKey',
 909     ),
 910     ConfigItem(
 911         '--uname',
 912         metavar='VERSION',
 913         help='"uname -r" information [.uname section]',
 914         config_key = 'UKI/Uname',
 915     ),
 916
 917     ConfigItem(
 918         '--efi-arch',
 919         metavar = 'ARCH',
 920         choices = ('ia32', 'x64', 'arm', 'aa64', 'riscv64'),
 921         help = 'target EFI architecture',
 922         config_key = 'UKI/EFIArch',
 923     ),
 924
 925     ConfigItem(
 926         '--stub',
 927         type = pathlib.Path,
 928         help = 'path to the sd-stub file [.text,.data,… sections]',
 929         config_key = 'UKI/Stub',
 930     ),
 931
 932     ConfigItem(
 933         '--sbat',
 934         metavar = 'TEXT|@PATH',
 935         help = 'SBAT policy [.sbat section] for addons',
 936         default = """sbat,1,SBAT Version,sbat,1,https://github.com/rhboot/shim/blob/main/SBAT.md
 937 uki.addon,1,UKI Addon,uki.addon,1,https://www.freedesktop.org/software/systemd/man/systemd-stub.html
 938 """,
 939         config_key = 'Addon/SBAT',
 940     ),
 941
 942     ConfigItem(
 943         '--section',
 944         dest = 'sections',
 945         metavar = 'NAME:TEXT|@PATH',
 946         type = Section.parse_arg,
 947         action = 'append',
 948         default = [],
 949         help = 'additional section as name and contents [NAME section]',
 950     ),
 951
 952     ConfigItem(
 953         '--pcr-banks',
 954         metavar = 'BANK…',
 955         type = parse_banks,
 956         config_key = 'UKI/PCRBanks',
 957     ),
 958
 959     ConfigItem(
 960         '--signing-engine',
 961         metavar = 'ENGINE',
 962         help = 'OpenSSL engine to use for signing',
 963         config_key = 'UKI/SigningEngine',
 964     ),
 965     ConfigItem(
 966         '--signtool',
 967         choices = ('sbsign', 'pesign'),
 968         dest = 'signtool',
 969         default = 'sbsign',
 970         help = 'whether to use sbsign or pesign. Default is sbsign.',
 971         config_key = 'UKI/SecureBootSigningTool',
 972     ),
 973     ConfigItem(
 974         '--secureboot-private-key',
 975         dest = 'sb_key',
 976         help = 'required by --signtool=sbsign. Path to key file or engine-specific designation for SB signing',
 977         config_key = 'UKI/SecureBootPrivateKey',
 978     ),
 979     ConfigItem(
 980         '--secureboot-certificate',
 981         dest = 'sb_cert',
 982         help = 'required by --signtool=sbsign. sbsign needs a path to certificate file or engine-specific designation for SB signing',
 983         config_key = 'UKI/SecureBootCertificate',
 984     ),
 985     ConfigItem(
 986         '--secureboot-certificate-dir',
 987         dest = 'sb_certdir',
 988         default = '/etc/pki/pesign',
 989         help = 'required by --signtool=pesign. Path to nss certificate database directory for PE signing. Default is /etc/pki/pesign',
 990         config_key = 'UKI/SecureBootCertificateDir',
 991     ),
 992     ConfigItem(
 993         '--secureboot-certificate-name',
 994         dest = 'sb_cert_name',
 995         help = 'required by --signtool=pesign. pesign needs a certificate nickname of nss certificate database entry to use for PE signing',
 996         config_key = 'UKI/SecureBootCertificateName',
 997     ),
 998
 999     ConfigItem(
1000         '--sign-kernel',
1001         action = argparse.BooleanOptionalAction,
1002         help = 'Sign the embedded kernel',
1003         config_key = 'UKI/SignKernel',
1004     ),
1005
1006     ConfigItem(
1007         '--pcr-private-key',
1008         dest = 'pcr_private_keys',
1009         metavar = 'PATH',
1010         type = pathlib.Path,
1011         action = 'append',
1012         help = 'private part of the keypair for signing PCR signatures',
1013         config_key = 'PCRSignature:/PCRPrivateKey',
1014         config_push = ConfigItem.config_set_group,
1015     ),
1016     ConfigItem(
1017         '--pcr-public-key',
1018         dest = 'pcr_public_keys',
1019         metavar = 'PATH',
1020         type = pathlib.Path,
1021         action = 'append',
1022         help = 'public part of the keypair for signing PCR signatures',
1023         config_key = 'PCRSignature:/PCRPublicKey',
1024         config_push = ConfigItem.config_set_group,
1025     ),
1026     ConfigItem(
1027         '--phases',
1028         dest = 'phase_path_groups',
1029         metavar = 'PHASE-PATH…',
1030         type = parse_phase_paths,
1031         action = 'append',
1032         help = 'phase-paths to create signatures for',
1033         config_key = 'PCRSignature:/Phases',
1034         config_push = ConfigItem.config_set_group,
1035     ),
1036
1037     ConfigItem(
1038         '--tools',
1039         type = pathlib.Path,
1040         action = 'append',
1041         help = 'Directories to search for tools (systemd-measure, …)',
1042     ),
1043
1044     ConfigItem(
1045         ('--output', '-o'),
1046         type = pathlib.Path,
1047         help = 'output file path',
1048     ),
1049
1050     ConfigItem(
1051         '--measure',
1052         action = argparse.BooleanOptionalAction,
1053         help = 'print systemd-measure output for the UKI',
1054     ),
1055 ]
1056
1057 CONFIGFILE_ITEMS = { item.config_key:item
1058                      for item in CONFIG_ITEMS
1059                      if item.config_key }
1060
1061
1062 def apply_config(namespace, filename=None):
1063     if filename is None:
1064         filename = namespace.config
1065     if filename is None:
1066         return
1067
1068     # Fill in ._groups based on --pcr-public-key=, --pcr-private-key=, and --phases=.
1069     assert '_groups' not in namespace
1070     n_pcr_priv = len(namespace.pcr_private_keys or ())
1071     namespace._groups = list(range(n_pcr_priv))
1072
1073     cp = configparser.ConfigParser(
1074         comment_prefixes='#',
1075         inline_comment_prefixes='#',
1076         delimiters='=',
1077         empty_lines_in_values=False,
1078         interpolation=None,
1079         strict=False)
1080     # Do not make keys lowercase
1081     cp.optionxform = lambda option: option
1082
1083     cp.read(filename)
1084
1085     for section_name, section in cp.items():
1086         idx = section_name.find(':')
1087         if idx >= 0:
1088             section_name, group = section_name[:idx+1], section_name[idx+1:]
1089             if not section_name or not group:
1090                 raise ValueError('Section name components cannot be empty')
1091             if ':' in group:
1092                 raise ValueError('Section name cannot contain more than one ":"')
1093         else:
1094             group = None
1095         for key, value in section.items():
1096             if item := CONFIGFILE_ITEMS.get(f'{section_name}/{key}'):
1097                 item.apply_config(namespace, section_name, group, key, value)
1098             else:
1099                 print(f'Unknown config setting [{section_name}] {key}=')
1100
1101
1102 def config_example():
1103     prev_section = None
1104     for item in CONFIG_ITEMS:
1105         section, key, value = item.config_example()
1106         if section:
1107             if prev_section != section:
1108                 if prev_section:
1109                     yield ''
1110                 yield f'[{section}]'
1111                 prev_section = section
1112             yield f'{key} = {value}'
1113
1114
1115 def create_parser():
1116     p = argparse.ArgumentParser(
1117         description='Build and sign Unified Kernel Images',
1118         allow_abbrev=False,
1119         usage='''\
1120 ukify [options…] [LINUX INITRD…]
1121 ''',
1122         epilog='\n  '.join(('config file:', *config_example())),
1123         formatter_class=argparse.RawDescriptionHelpFormatter,
1124     )
1125
1126     for item in CONFIG_ITEMS:
1127         item.add_to(p)
1128
1129     # Suppress printing of usage synopsis on errors
1130     p.error = lambda message: p.exit(2, f'{p.prog}: error: {message}\n')
1131
1132     return p
1133
1134
1135 def finalize_options(opts):
1136     if opts.cmdline and opts.cmdline.startswith('@'):
1137         opts.cmdline = pathlib.Path(opts.cmdline[1:])
1138     elif opts.cmdline:
1139         # Drop whitespace from the commandline. If we're reading from a file,
1140         # we copy the contents verbatim. But configuration specified on the commandline
1141         # or in the config file may contain additional whitespace that has no meaning.
1142         opts.cmdline = ' '.join(opts.cmdline.split())
1143
1144     if opts.os_release and opts.os_release.startswith('@'):
1145         opts.os_release = pathlib.Path(opts.os_release[1:])
1146     elif not opts.os_release and opts.linux:
1147         p = pathlib.Path('/etc/os-release')
1148         if not p.exists():
1149             p = pathlib.Path('/usr/lib/os-release')
1150         opts.os_release = p
1151
1152     if opts.efi_arch is None:
1153         opts.efi_arch = guess_efi_arch()
1154
1155     if opts.stub is None:
1156         if opts.linux is not None:
1157             opts.stub = pathlib.Path(f'/usr/lib/systemd/boot/efi/linux{opts.efi_arch}.efi.stub')
1158         else:
1159             opts.stub = pathlib.Path(f'/usr/lib/systemd/boot/efi/addon{opts.efi_arch}.efi.stub')
1160
1161     if opts.signing_engine is None:
1162         if opts.sb_key:
1163             opts.sb_key = pathlib.Path(opts.sb_key)
1164         if opts.sb_cert:
1165             opts.sb_cert = pathlib.Path(opts.sb_cert)
1166
1167     if opts.signtool == 'sbsign':
1168         if bool(opts.sb_key) ^ bool(opts.sb_cert):
1169             raise ValueError('--secureboot-private-key= and --secureboot-certificate= must be specified together when using --signtool=sbsign')
1170     else:
1171         if not bool(opts.sb_cert_name):
1172             raise ValueError('--certificate-name must be specified when using --signtool=pesign')
1173
1174     if opts.sign_kernel and not opts.sb_key and not opts.sb_cert_name:
1175         raise ValueError('--sign-kernel requires either --secureboot-private-key= and --secureboot-certificate= (for sbsign) or --secureboot-certificate-name= (for pesign) to be specified')
1176
1177     if opts.output is None:
1178         if opts.linux is None:
1179             raise ValueError('--output= must be specified when building a PE addon')
1180         suffix = '.efi' if opts.sb_key or opts.sb_cert_name else '.unsigned.efi'
1181         opts.output = opts.linux.name + suffix
1182
1183     for section in opts.sections:
1184         section.check_name()
1185
1186     if opts.summary:
1187         # TODO: replace pprint() with some fancy formatting.
1188         pprint.pprint(vars(opts))
1189         sys.exit()
1190
1191
1192 def parse_args(args=None):
1193     p = create_parser()
1194     opts = p.parse_args(args)
1195
1196     # Check that --pcr-public-key=, --pcr-private-key=, and --phases=
1197     # have either the same number of arguments are are not specified at all.
1198     n_pcr_pub = None if opts.pcr_public_keys is None else len(opts.pcr_public_keys)
1199     n_pcr_priv = None if opts.pcr_private_keys is None else len(opts.pcr_private_keys)
1200     n_phase_path_groups = None if opts.phase_path_groups is None else len(opts.phase_path_groups)
1201     if n_pcr_pub is not None and n_pcr_pub != n_pcr_priv:
1202         raise ValueError('--pcr-public-key= specifications must match --pcr-private-key=')
1203     if n_phase_path_groups is not None and n_phase_path_groups != n_pcr_priv:
1204         raise ValueError('--phases= specifications must match --pcr-private-key=')
1205
1206     apply_config(opts)
1207
1208     finalize_options(opts)
1209
1210     return opts
1211
1212
1213 def main():
1214     opts = parse_args()
1215     check_inputs(opts)
1216     make_uki(opts)
1217
1218
1219 if __name__ == '__main__':
1220     main()