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 = 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         ('.splash',  opts.splash,     True ),
 662         ('.pcrpkey', pcrpkey,         True ),
 663         ('.initrd',  initrd,          True ),
 664         ('.uname',   opts.uname,      False),
 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 creation
 683
 684     if linux is not None:
 685         uki.add_section(Section.create('.linux', linux, measure=True))
 686
 687     if sign_args_present:
 688         unsigned = tempfile.NamedTemporaryFile(prefix='uki')
 689         output = unsigned.name
 690     else:
 691         output = opts.output
 692
 693     pe_add_sections(uki, output)
 694
 695     # UKI signing
 696
 697     if sign_args_present:
 698         sign(sign_tool, unsigned.name, opts.output, opts=opts)
 699
 700         # We end up with no executable bits, let's reapply them
 701         os.umask(umask := os.umask(0))
 702         os.chmod(opts.output, 0o777 & ~umask)
 703
 704     print(f"Wrote {'signed' if sign_args_present else 'unsigned'} {opts.output}")
 705
 706
 707 @dataclasses.dataclass(frozen=True)
 708 class ConfigItem:
 709     @staticmethod
 710     def config_list_prepend(
 711             namespace: argparse.Namespace,
 712             group: Optional[str],
 713             dest: str,
 714             value: Any,
 715     ) -> None:
 716         "Prepend value to namespace.<dest>"
 717
 718         assert not group
 719
 720         old = getattr(namespace, dest, [])
 721         setattr(namespace, dest, value + old)
 722
 723     @staticmethod
 724     def config_set_if_unset(
 725             namespace: argparse.Namespace,
 726             group: Optional[str],
 727             dest: str,
 728             value: Any,
 729     ) -> None:
 730         "Set namespace.<dest> to value only if it was None"
 731
 732         assert not group
 733
 734         if getattr(namespace, dest) is None:
 735             setattr(namespace, dest, value)
 736
 737     @staticmethod
 738     def config_set_group(
 739             namespace: argparse.Namespace,
 740             group: Optional[str],
 741             dest: str,
 742             value: Any,
 743     ) -> None:
 744         "Set namespace.<dest>[idx] to value, with idx derived from group"
 745
 746         if group not in namespace._groups:
 747             namespace._groups += [group]
 748         idx = namespace._groups.index(group)
 749
 750         old = getattr(namespace, dest, None)
 751         if old is None:
 752             old = []
 753         setattr(namespace, dest,
 754                 old + ([None] * (idx - len(old))) + [value])
 755
 756     @staticmethod
 757     def parse_boolean(s: str) -> bool:
 758         "Parse 1/true/yes/y/t/on as true and 0/false/no/n/f/off/None as false"
 759         s_l = s.lower()
 760         if s_l in {'1', 'true', 'yes', 'y', 't', 'on'}:
 761             return True
 762         if s_l in {'0', 'false', 'no', 'n', 'f', 'off'}:
 763             return False
 764         raise ValueError('f"Invalid boolean literal: {s!r}')
 765
 766     # arguments for argparse.ArgumentParser.add_argument()
 767     name: Union[str, tuple[str, str]]
 768     dest: Optional[str] = None
 769     metavar: Optional[str] = None
 770     type: Optional[Callable] = None
 771     nargs: Optional[str] = None
 772     action: Optional[Union[str, Callable]] = None
 773     default: Any = None
 774     version: Optional[str] = None
 775     choices: Optional[tuple[str, ...]] = None
 776     help: Optional[str] = None
 777
 778     # metadata for config file parsing
 779     config_key: Optional[str] = None
 780     config_push: Callable[[argparse.Namespace, Optional[str], str, Any], None] = \
 781                     config_set_if_unset
 782
 783     def _names(self) -> tuple[str, ...]:
 784         return self.name if isinstance(self.name, tuple) else (self.name,)
 785
 786     def argparse_dest(self) -> str:
 787         # It'd be nice if argparse exported this, but I don't see that in the API
 788         if self.dest:
 789             return self.dest
 790         return self._names()[0].lstrip('-').replace('-', '_')
 791
 792     def add_to(self, parser: argparse.ArgumentParser):
 793         kwargs = { key:val
 794                    for key in dataclasses.asdict(self)
 795                    if (key not in ('name', 'config_key', 'config_push') and
 796                        (val := getattr(self, key)) is not None) }
 797         args = self._names()
 798         parser.add_argument(*args, **kwargs)
 799
 800     def apply_config(self, namespace, section, group, key, value) -> None:
 801         assert f'{section}/{key}' == self.config_key
 802         dest = self.argparse_dest()
 803
 804         conv: Callable[[str], Any]
 805         if self.action == argparse.BooleanOptionalAction:
 806             # We need to handle this case separately: the options are called
 807             # --foo and --no-foo, and no argument is parsed. But in the config
 808             # file, we have Foo=yes or Foo=no.
 809             conv = self.parse_boolean
 810         elif self.type:
 811             conv = self.type
 812         else:
 813             conv = lambda s:s
 814
 815         if self.nargs == '*':
 816             value = [conv(v) for v in value.split()]
 817         else:
 818             value = conv(value)
 819
 820         self.config_push(namespace, group, dest, value)
 821
 822     def config_example(self) -> tuple[Optional[str], Optional[str], Optional[str]]:
 823         if not self.config_key:
 824             return None, None, None
 825         section_name, key = self.config_key.split('/', 1)
 826         if section_name.endswith(':'):
 827             section_name += 'NAME'
 828         if self.choices:
 829             value = '|'.join(self.choices)
 830         else:
 831             value = self.metavar or self.argparse_dest().upper()
 832         return (section_name, key, value)
 833
 834
 835 CONFIG_ITEMS = [
 836     ConfigItem(
 837         '--version',
 838         action = 'version',
 839         version = f'ukify {__version__}',
 840     ),
 841
 842     ConfigItem(
 843         '--summary',
 844         help = 'print parsed config and exit',
 845         action = 'store_true',
 846     ),
 847
 848     ConfigItem(
 849         'linux',
 850         metavar = 'LINUX',
 851         type = pathlib.Path,
 852         nargs = '?',
 853         help = 'vmlinuz file [.linux section]',
 854         config_key = 'UKI/Linux',
 855     ),
 856
 857     ConfigItem(
 858         'initrd',
 859         metavar = 'INITRD…',
 860         type = pathlib.Path,
 861         nargs = '*',
 862         help = 'initrd files [.initrd section]',
 863         config_key = 'UKI/Initrd',
 864         config_push = ConfigItem.config_list_prepend,
 865     ),
 866
 867     ConfigItem(
 868         ('--config', '-c'),
 869         metavar = 'PATH',
 870         help = 'configuration file',
 871     ),
 872
 873     ConfigItem(
 874         '--cmdline',
 875         metavar = 'TEXT|@PATH',
 876         help = 'kernel command line [.cmdline section]',
 877         config_key = 'UKI/Cmdline',
 878     ),
 879
 880     ConfigItem(
 881         '--os-release',
 882         metavar = 'TEXT|@PATH',
 883         help = 'path to os-release file [.osrel section]',
 884         config_key = 'UKI/OSRelease',
 885     ),
 886
 887     ConfigItem(
 888         '--devicetree',
 889         metavar = 'PATH',
 890         type = pathlib.Path,
 891         help = 'Device Tree file [.dtb section]',
 892         config_key = 'UKI/DeviceTree',
 893     ),
 894     ConfigItem(
 895         '--splash',
 896         metavar = 'BMP',
 897         type = pathlib.Path,
 898         help = 'splash image bitmap file [.splash section]',
 899         config_key = 'UKI/Splash',
 900     ),
 901     ConfigItem(
 902         '--pcrpkey',
 903         metavar = 'KEY',
 904         type = pathlib.Path,
 905         help = 'embedded public key to seal secrets to [.pcrpkey section]',
 906         config_key = 'UKI/PCRPKey',
 907     ),
 908     ConfigItem(
 909         '--uname',
 910         metavar='VERSION',
 911         help='"uname -r" information [.uname section]',
 912         config_key = 'UKI/Uname',
 913     ),
 914
 915     ConfigItem(
 916         '--efi-arch',
 917         metavar = 'ARCH',
 918         choices = ('ia32', 'x64', 'arm', 'aa64', 'riscv64'),
 919         help = 'target EFI architecture',
 920         config_key = 'UKI/EFIArch',
 921     ),
 922
 923     ConfigItem(
 924         '--stub',
 925         type = pathlib.Path,
 926         help = 'path to the sd-stub file [.text,.data,… sections]',
 927         config_key = 'UKI/Stub',
 928     ),
 929
 930     ConfigItem(
 931         '--section',
 932         dest = 'sections',
 933         metavar = 'NAME:TEXT|@PATH',
 934         type = Section.parse_arg,
 935         action = 'append',
 936         default = [],
 937         help = 'additional section as name and contents [NAME section]',
 938     ),
 939
 940     ConfigItem(
 941         '--pcr-banks',
 942         metavar = 'BANK…',
 943         type = parse_banks,
 944         config_key = 'UKI/PCRBanks',
 945     ),
 946
 947     ConfigItem(
 948         '--signing-engine',
 949         metavar = 'ENGINE',
 950         help = 'OpenSSL engine to use for signing',
 951         config_key = 'UKI/SigningEngine',
 952     ),
 953     ConfigItem(
 954         '--signtool',
 955         choices = ('sbsign', 'pesign'),
 956         dest = 'signtool',
 957         default = 'sbsign',
 958         help = 'whether to use sbsign or pesign. Default is sbsign.',
 959         config_key = 'UKI/SecureBootSigningTool',
 960     ),
 961     ConfigItem(
 962         '--secureboot-private-key',
 963         dest = 'sb_key',
 964         help = 'required by --signtool=sbsign. Path to key file or engine-specific designation for SB signing',
 965         config_key = 'UKI/SecureBootPrivateKey',
 966     ),
 967     ConfigItem(
 968         '--secureboot-certificate',
 969         dest = 'sb_cert',
 970         help = 'required by --signtool=sbsign. sbsign needs a path to certificate file or engine-specific designation for SB signing',
 971         config_key = 'UKI/SecureBootCertificate',
 972     ),
 973     ConfigItem(
 974         '--secureboot-certificate-dir',
 975         dest = 'sb_certdir',
 976         default = '/etc/pki/pesign',
 977         help = 'required by --signtool=pesign. Path to nss certificate database directory for PE signing. Default is /etc/pki/pesign',
 978         config_key = 'UKI/SecureBootCertificateDir',
 979     ),
 980     ConfigItem(
 981         '--secureboot-certificate-name',
 982         dest = 'sb_cert_name',
 983         help = 'required by --signtool=pesign. pesign needs a certificate nickname of nss certificate database entry to use for PE signing',
 984         config_key = 'UKI/SecureBootCertificateName',
 985     ),
 986
 987     ConfigItem(
 988         '--sign-kernel',
 989         action = argparse.BooleanOptionalAction,
 990         help = 'Sign the embedded kernel',
 991         config_key = 'UKI/SignKernel',
 992     ),
 993
 994     ConfigItem(
 995         '--pcr-private-key',
 996         dest = 'pcr_private_keys',
 997         metavar = 'PATH',
 998         type = pathlib.Path,
 999         action = 'append',
1000         help = 'private part of the keypair for signing PCR signatures',
1001         config_key = 'PCRSignature:/PCRPrivateKey',
1002         config_push = ConfigItem.config_set_group,
1003     ),
1004     ConfigItem(
1005         '--pcr-public-key',
1006         dest = 'pcr_public_keys',
1007         metavar = 'PATH',
1008         type = pathlib.Path,
1009         action = 'append',
1010         help = 'public part of the keypair for signing PCR signatures',
1011         config_key = 'PCRSignature:/PCRPublicKey',
1012         config_push = ConfigItem.config_set_group,
1013     ),
1014     ConfigItem(
1015         '--phases',
1016         dest = 'phase_path_groups',
1017         metavar = 'PHASE-PATH…',
1018         type = parse_phase_paths,
1019         action = 'append',
1020         help = 'phase-paths to create signatures for',
1021         config_key = 'PCRSignature:/Phases',
1022         config_push = ConfigItem.config_set_group,
1023     ),
1024
1025     ConfigItem(
1026         '--tools',
1027         type = pathlib.Path,
1028         action = 'append',
1029         help = 'Directories to search for tools (systemd-measure, …)',
1030     ),
1031
1032     ConfigItem(
1033         ('--output', '-o'),
1034         type = pathlib.Path,
1035         help = 'output file path',
1036     ),
1037
1038     ConfigItem(
1039         '--measure',
1040         action = argparse.BooleanOptionalAction,
1041         help = 'print systemd-measure output for the UKI',
1042     ),
1043 ]
1044
1045 CONFIGFILE_ITEMS = { item.config_key:item
1046                      for item in CONFIG_ITEMS
1047                      if item.config_key }
1048
1049
1050 def apply_config(namespace, filename=None):
1051     if filename is None:
1052         filename = namespace.config
1053     if filename is None:
1054         return
1055
1056     # Fill in ._groups based on --pcr-public-key=, --pcr-private-key=, and --phases=.
1057     assert '_groups' not in namespace
1058     n_pcr_priv = len(namespace.pcr_private_keys or ())
1059     namespace._groups = list(range(n_pcr_priv))
1060
1061     cp = configparser.ConfigParser(
1062         comment_prefixes='#',
1063         inline_comment_prefixes='#',
1064         delimiters='=',
1065         empty_lines_in_values=False,
1066         interpolation=None,
1067         strict=False)
1068     # Do not make keys lowercase
1069     cp.optionxform = lambda option: option
1070
1071     cp.read(filename)
1072
1073     for section_name, section in cp.items():
1074         idx = section_name.find(':')
1075         if idx >= 0:
1076             section_name, group = section_name[:idx+1], section_name[idx+1:]
1077             if not section_name or not group:
1078                 raise ValueError('Section name components cannot be empty')
1079             if ':' in group:
1080                 raise ValueError('Section name cannot contain more than one ":"')
1081         else:
1082             group = None
1083         for key, value in section.items():
1084             if item := CONFIGFILE_ITEMS.get(f'{section_name}/{key}'):
1085                 item.apply_config(namespace, section_name, group, key, value)
1086             else:
1087                 print(f'Unknown config setting [{section_name}] {key}=')
1088
1089
1090 def config_example():
1091     prev_section = None
1092     for item in CONFIG_ITEMS:
1093         section, key, value = item.config_example()
1094         if section:
1095             if prev_section != section:
1096                 if prev_section:
1097                     yield ''
1098                 yield f'[{section}]'
1099                 prev_section = section
1100             yield f'{key} = {value}'
1101
1102
1103 def create_parser():
1104     p = argparse.ArgumentParser(
1105         description='Build and sign Unified Kernel Images',
1106         allow_abbrev=False,
1107         usage='''\
1108 ukify [options…] [LINUX INITRD…]
1109 ''',
1110         epilog='\n  '.join(('config file:', *config_example())),
1111         formatter_class=argparse.RawDescriptionHelpFormatter,
1112     )
1113
1114     for item in CONFIG_ITEMS:
1115         item.add_to(p)
1116
1117     # Suppress printing of usage synopsis on errors
1118     p.error = lambda message: p.exit(2, f'{p.prog}: error: {message}\n')
1119
1120     return p
1121
1122
1123 def finalize_options(opts):
1124     if opts.cmdline and opts.cmdline.startswith('@'):
1125         opts.cmdline = pathlib.Path(opts.cmdline[1:])
1126     elif opts.cmdline:
1127         # Drop whitespace from the commandline. If we're reading from a file,
1128         # we copy the contents verbatim. But configuration specified on the commandline
1129         # or in the config file may contain additional whitespace that has no meaning.
1130         opts.cmdline = ' '.join(opts.cmdline.split())
1131
1132     if opts.os_release and opts.os_release.startswith('@'):
1133         opts.os_release = pathlib.Path(opts.os_release[1:])
1134     elif not opts.os_release and opts.linux:
1135         p = pathlib.Path('/etc/os-release')
1136         if not p.exists():
1137             p = pathlib.Path('/usr/lib/os-release')
1138         opts.os_release = p
1139
1140     if opts.efi_arch is None:
1141         opts.efi_arch = guess_efi_arch()
1142
1143     if opts.stub is None:
1144         opts.stub = pathlib.Path(f'/usr/lib/systemd/boot/efi/linux{opts.efi_arch}.efi.stub')
1145
1146     if opts.signing_engine is None:
1147         if opts.sb_key:
1148             opts.sb_key = pathlib.Path(opts.sb_key)
1149         if opts.sb_cert:
1150             opts.sb_cert = pathlib.Path(opts.sb_cert)
1151
1152     if opts.signtool == 'sbsign':
1153         if bool(opts.sb_key) ^ bool(opts.sb_cert):
1154             raise ValueError('--secureboot-private-key= and --secureboot-certificate= must be specified together when using --signtool=sbsign')
1155     else:
1156         if not bool(opts.sb_cert_name):
1157             raise ValueError('--certificate-name must be specified when using --signtool=pesign')
1158
1159     if opts.sign_kernel and not opts.sb_key and not opts.sb_cert_name:
1160         raise ValueError('--sign-kernel requires either --secureboot-private-key= and --secureboot-certificate= (for sbsign) or --secureboot-certificate-name= (for pesign) to be specified')
1161
1162     if opts.output is None:
1163         if opts.linux is None:
1164             raise ValueError('--output= must be specified when building a PE addon')
1165         suffix = '.efi' if opts.sb_key or opts.sb_cert_name else '.unsigned.efi'
1166         opts.output = opts.linux.name + suffix
1167
1168     for section in opts.sections:
1169         section.check_name()
1170
1171     if opts.summary:
1172         # TODO: replace pprint() with some fancy formatting.
1173         pprint.pprint(vars(opts))
1174         sys.exit()
1175
1176
1177 def parse_args(args=None):
1178     p = create_parser()
1179     opts = p.parse_args(args)
1180
1181     # Check that --pcr-public-key=, --pcr-private-key=, and --phases=
1182     # have either the same number of arguments are are not specified at all.
1183     n_pcr_pub = None if opts.pcr_public_keys is None else len(opts.pcr_public_keys)
1184     n_pcr_priv = None if opts.pcr_private_keys is None else len(opts.pcr_private_keys)
1185     n_phase_path_groups = None if opts.phase_path_groups is None else len(opts.phase_path_groups)
1186     if n_pcr_pub is not None and n_pcr_pub != n_pcr_priv:
1187         raise ValueError('--pcr-public-key= specifications must match --pcr-private-key=')
1188     if n_phase_path_groups is not None and n_phase_path_groups != n_pcr_priv:
1189         raise ValueError('--phases= specifications must match --pcr-private-key=')
1190
1191     apply_config(opts)
1192
1193     finalize_options(opts)
1194
1195     return opts
1196
1197
1198 def main():
1199     opts = parse_args()
1200     check_inputs(opts)
1201     make_uki(opts)
1202
1203
1204 if __name__ == '__main__':
1205     main()