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=import-outside-toplevel,consider-using-with,unused-argument
  20 # pylint: disable=unnecessary-lambda-assignment
  21
  22 import argparse
  23 import configparser
  24 import contextlib
  25 import collections
  26 import dataclasses
  27 import datetime
  28 import fnmatch
  29 import itertools
  30 import json
  31 import os
  32 import pathlib
  33 import pprint
  34 import pydoc
  35 import re
  36 import shlex
  37 import shutil
  38 import socket
  39 import subprocess
  40 import sys
  41 import tempfile
  42 import textwrap
  43 from hashlib import sha256
  44 from typing import (Any,
  45                     Callable,
  46                     IO,
  47                     Optional,
  48                     Sequence,
  49                     Union)
  50
  51 import pefile  # type: ignore
  52
  53 __version__ = '{{PROJECT_VERSION}} ({{GIT_VERSION}})'
  54
  55 EFI_ARCH_MAP = {
  56     # host_arch glob : [efi_arch, 32_bit_efi_arch if mixed mode is supported]
  57     'x86_64'       : ['x64', 'ia32'],
  58     'i[3456]86'    : ['ia32'],
  59     'aarch64'      : ['aa64'],
  60     'armv[45678]*l': ['arm'],
  61     'loongarch32'  : ['loongarch32'],
  62     'loongarch64'  : ['loongarch64'],
  63     'riscv32'      : ['riscv32'],
  64     'riscv64'      : ['riscv64'],
  65 }
  66 EFI_ARCHES: list[str] = sum(EFI_ARCH_MAP.values(), [])
  67
  68 # Default configuration directories and file name.
  69 # When the user does not specify one, the directories are searched in this order and the first file found is used.
  70 DEFAULT_CONFIG_DIRS = ['/run/systemd', '/etc/systemd', '/usr/local/lib/systemd', '/usr/lib/systemd']
  71 DEFAULT_CONFIG_FILE = 'ukify.conf'
  72
  73 class Style:
  74     bold = "\033[0;1;39m" if sys.stderr.isatty() else ""
  75     gray = "\033[0;38;5;245m" if sys.stderr.isatty() else ""
  76     red = "\033[31;1m" if sys.stderr.isatty() else ""
  77     yellow = "\033[33;1m" if sys.stderr.isatty() else ""
  78     reset = "\033[0m" if sys.stderr.isatty() else ""
  79
  80
  81 def guess_efi_arch():
  82     arch = os.uname().machine
  83
  84     for glob, mapping in EFI_ARCH_MAP.items():
  85         if fnmatch.fnmatch(arch, glob):
  86             efi_arch, *fallback = mapping
  87             break
  88     else:
  89         raise ValueError(f'Unsupported architecture {arch}')
  90
  91     # This makes sense only on some architectures, but it also probably doesn't
  92     # hurt on others, so let's just apply the check everywhere.
  93     if fallback:
  94         fw_platform_size = pathlib.Path('/sys/firmware/efi/fw_platform_size')
  95         try:
  96             size = fw_platform_size.read_text().strip()
  97         except FileNotFoundError:
  98             pass
  99         else:
 100             if int(size) == 32:
 101                 efi_arch = fallback[0]
 102
 103     # print(f'Host arch {arch!r}, EFI arch {efi_arch!r}')
 104     return efi_arch
 105
 106
 107 def page(text: str, enabled: Optional[bool]) -> None:
 108     if enabled:
 109         # Initialize less options from $SYSTEMD_LESS or provide a suitable fallback.
 110         os.environ['LESS'] = os.getenv('SYSTEMD_LESS', 'FRSXMK')
 111         pydoc.pager(text)
 112     else:
 113         print(text)
 114
 115
 116 def shell_join(cmd):
 117     # TODO: drop in favour of shlex.join once shlex.join supports pathlib.Path.
 118     return ' '.join(shlex.quote(str(x)) for x in cmd)
 119
 120
 121 def round_up(x, blocksize=4096):
 122     return (x + blocksize - 1) // blocksize * blocksize
 123
 124
 125 def try_import(modname, name=None):
 126     try:
 127         return __import__(modname)
 128     except ImportError as e:
 129         raise ValueError(f'Kernel is compressed with {name or modname}, but module unavailable') from e
 130
 131
 132 def maybe_decompress(filename):
 133     """Decompress file if compressed. Return contents."""
 134     f = open(filename, 'rb')
 135     start = f.read(4)
 136     f.seek(0)
 137
 138     if start.startswith(b'\x7fELF'):
 139         # not compressed
 140         return f.read()
 141
 142     if start.startswith(b'MZ'):
 143         # not compressed aarch64 and riscv64
 144         return f.read()
 145
 146     if start.startswith(b'\x1f\x8b'):
 147         gzip = try_import('gzip')
 148         return gzip.open(f).read()
 149
 150     if start.startswith(b'\x28\xb5\x2f\xfd'):
 151         zstd = try_import('zstd')
 152         return zstd.uncompress(f.read())
 153
 154     if start.startswith(b'\x02\x21\x4c\x18'):
 155         lz4 = try_import('lz4.frame', 'lz4')
 156         return lz4.frame.decompress(f.read())
 157
 158     if start.startswith(b'\x04\x22\x4d\x18'):
 159         print('Newer lz4 stream format detected! This may not boot!')
 160         lz4 = try_import('lz4.frame', 'lz4')
 161         return lz4.frame.decompress(f.read())
 162
 163     if start.startswith(b'\x89LZO'):
 164         # python3-lzo is not packaged for Fedora
 165         raise NotImplementedError('lzo decompression not implemented')
 166
 167     if start.startswith(b'BZh'):
 168         bz2 = try_import('bz2', 'bzip2')
 169         return bz2.open(f).read()
 170
 171     if start.startswith(b'\x5d\x00\x00'):
 172         lzma = try_import('lzma')
 173         return lzma.open(f).read()
 174
 175     raise NotImplementedError(f'unknown file format (starts with {start})')
 176
 177
 178 class Uname:
 179     # This class is here purely as a namespace for the functions
 180
 181     VERSION_PATTERN = r'(?P<version>[a-z0-9._-]+) \([^ )]+\) (?:#.*)'
 182
 183     NOTES_PATTERN = r'^\s+Linux\s+0x[0-9a-f]+\s+OPEN\n\s+description data: (?P<version>[0-9a-f ]+)\s*$'
 184
 185     # Linux version 6.0.8-300.fc37.ppc64le (mockbuild@buildvm-ppc64le-03.iad2.fedoraproject.org)
 186     # (gcc (GCC) 12.2.1 20220819 (Red Hat 12.2.1-2), GNU ld version 2.38-24.fc37)
 187     # #1 SMP Fri Nov 11 14:39:11 UTC 2022
 188     TEXT_PATTERN = rb'Linux version (?P<version>\d\.\S+) \('
 189
 190     @classmethod
 191     def scrape_x86(cls, filename, opts=None):
 192         # Based on https://gitlab.archlinux.org/archlinux/mkinitcpio/mkinitcpio/-/blob/master/functions#L136
 193         # and https://www.kernel.org/doc/html/latest/x86/boot.html#the-real-mode-kernel-header
 194         with open(filename, 'rb') as f:
 195             f.seek(0x202)
 196             magic = f.read(4)
 197             if magic != b'HdrS':
 198                 raise ValueError('Real-Mode Kernel Header magic not found')
 199             f.seek(0x20E)
 200             offset = f.read(1)[0] + f.read(1)[0]*256  # Pointer to kernel version string
 201             f.seek(0x200 + offset)
 202             text = f.read(128)
 203         text = text.split(b'\0', maxsplit=1)[0]
 204         text = text.decode()
 205
 206         if not (m := re.match(cls.VERSION_PATTERN, text)):
 207             raise ValueError(f'Cannot parse version-host-release uname string: {text!r}')
 208         return m.group('version')
 209
 210     @classmethod
 211     def scrape_elf(cls, filename, opts=None):
 212         readelf = find_tool('readelf', opts=opts)
 213
 214         cmd = [
 215             readelf,
 216             '--notes',
 217             filename,
 218         ]
 219
 220         print('+', shell_join(cmd))
 221         try:
 222             notes = subprocess.check_output(cmd, stderr=subprocess.PIPE, text=True)
 223         except subprocess.CalledProcessError as e:
 224             raise ValueError(e.stderr.strip()) from e
 225
 226         if not (m := re.search(cls.NOTES_PATTERN, notes, re.MULTILINE)):
 227             raise ValueError('Cannot find Linux version note')
 228
 229         text = ''.join(chr(int(c, 16)) for c in m.group('version').split())
 230         return text.rstrip('\0')
 231
 232     @classmethod
 233     def scrape_generic(cls, filename, opts=None):
 234         # import libarchive
 235         # libarchive-c fails with
 236         # ArchiveError: Unrecognized archive format (errno=84, retcode=-30, archive_p=94705420454656)
 237
 238         # Based on https://gitlab.archlinux.org/archlinux/mkinitcpio/mkinitcpio/-/blob/master/functions#L209
 239
 240         text = maybe_decompress(filename)
 241         if not (m := re.search(cls.TEXT_PATTERN, text)):
 242             raise ValueError(f'Cannot find {cls.TEXT_PATTERN!r} in {filename}')
 243
 244         return m.group('version').decode()
 245
 246     @classmethod
 247     def scrape(cls, filename, opts=None):
 248         for func in (cls.scrape_x86, cls.scrape_elf, cls.scrape_generic):
 249             try:
 250                 version = func(filename, opts=opts)
 251                 print(f'Found uname version: {version}')
 252                 return version
 253             except ValueError as e:
 254                 print(str(e))
 255         return None
 256
 257 DEFAULT_SECTIONS_TO_SHOW = {
 258         '.linux'    : 'binary',
 259         '.initrd'   : 'binary',
 260         '.splash'   : 'binary',
 261         '.dtb'      : 'binary',
 262         '.cmdline'  : 'text',
 263         '.osrel'    : 'text',
 264         '.uname'    : 'text',
 265         '.pcrpkey'  : 'text',
 266         '.pcrsig'   : 'text',
 267         '.sbat'     : 'text',
 268         '.sbom'     : 'binary',
 269 }
 270
 271 @dataclasses.dataclass
 272 class Section:
 273     name: str
 274     content: Optional[pathlib.Path]
 275     tmpfile: Optional[IO] = None
 276     measure: bool = False
 277     output_mode: Optional[str] = None
 278
 279     @classmethod
 280     def create(cls, name, contents, **kwargs):
 281         if isinstance(contents, (str, bytes)):
 282             mode = 'wt' if isinstance(contents, str) else 'wb'
 283             tmp = tempfile.NamedTemporaryFile(mode=mode, prefix=f'tmp{name}')
 284             tmp.write(contents)
 285             tmp.flush()
 286             contents = pathlib.Path(tmp.name)
 287         else:
 288             tmp = None
 289
 290         return cls(name, contents, tmpfile=tmp, **kwargs)
 291
 292     @classmethod
 293     def parse_input(cls, s):
 294         try:
 295             name, contents, *rest = s.split(':')
 296         except ValueError as e:
 297             raise ValueError(f'Cannot parse section spec (name or contents missing): {s!r}') from e
 298         if rest:
 299             raise ValueError(f'Cannot parse section spec (extraneous parameters): {s!r}')
 300
 301         if contents.startswith('@'):
 302             contents = pathlib.Path(contents[1:])
 303
 304         sec = cls.create(name, contents)
 305         sec.check_name()
 306         return sec
 307
 308     @classmethod
 309     def parse_output(cls, s):
 310         if not (m := re.match(r'([a-zA-Z0-9_.]+):(text|binary)(?:@(.+))?', s)):
 311             raise ValueError(f'Cannot parse section spec: {s!r}')
 312
 313         name, ttype, out = m.groups()
 314         out = pathlib.Path(out) if out else None
 315
 316         return cls.create(name, out, output_mode=ttype)
 317
 318     def size(self):
 319         return self.content.stat().st_size
 320
 321     def check_name(self):
 322         # PE section names with more than 8 characters are legal, but our stub does
 323         # not support them.
 324         if not self.name.isascii() or not self.name.isprintable():
 325             raise ValueError(f'Bad section name: {self.name!r}')
 326         if len(self.name) > 8:
 327             raise ValueError(f'Section name too long: {self.name!r}')
 328
 329
 330 @dataclasses.dataclass
 331 class UKI:
 332     executable: list[Union[pathlib.Path, str]]
 333     sections: list[Section] = dataclasses.field(default_factory=list, init=False)
 334
 335     def add_section(self, section):
 336         if section.name in [s.name for s in self.sections]:
 337             raise ValueError(f'Duplicate section {section.name}')
 338
 339         self.sections += [section]
 340
 341
 342 def parse_banks(s):
 343     banks = re.split(r',|\s+', s)
 344     # TODO: do some sanity checking here
 345     return banks
 346
 347
 348 KNOWN_PHASES = (
 349     'enter-initrd',
 350     'leave-initrd',
 351     'sysinit',
 352     'ready',
 353     'shutdown',
 354     'final',
 355 )
 356
 357 def parse_phase_paths(s):
 358     # Split on commas or whitespace here. Commas might be hard to parse visually.
 359     paths = re.split(r',|\s+', s)
 360
 361     for path in paths:
 362         for phase in path.split(':'):
 363             if phase not in KNOWN_PHASES:
 364                 raise argparse.ArgumentTypeError(f'Unknown boot phase {phase!r} ({path=})')
 365
 366     return paths
 367
 368
 369 def check_splash(filename):
 370     if filename is None:
 371         return
 372
 373     # import is delayed, to avoid import when the splash image is not used
 374     try:
 375         from PIL import Image
 376     except ImportError:
 377         return
 378
 379     img = Image.open(filename, formats=['BMP'])
 380     print(f'Splash image {filename} is {img.width}×{img.height} pixels')
 381
 382
 383 def check_inputs(opts):
 384     for name, value in vars(opts).items():
 385         if name in {'output', 'tools'}:
 386             continue
 387
 388         if isinstance(value, pathlib.Path):
 389             # Open file to check that we can read it, or generate an exception
 390             value.open().close()
 391         elif isinstance(value, list):
 392             for item in value:
 393                 if isinstance(item, pathlib.Path):
 394                     item.open().close()
 395
 396     check_splash(opts.splash)
 397
 398
 399 def check_cert_and_keys_nonexistent(opts):
 400     # Raise if any of the keys and certs are found on disk
 401     paths = itertools.chain(
 402         (opts.sb_key, opts.sb_cert),
 403         *((priv_key, pub_key)
 404           for priv_key, pub_key, _ in key_path_groups(opts)))
 405     for path in paths:
 406         if path and path.exists():
 407             raise ValueError(f'{path} is present')
 408
 409
 410 def find_tool(name, fallback=None, opts=None):
 411     if opts and opts.tools:
 412         for d in opts.tools:
 413             tool = d / name
 414             if tool.exists():
 415                 return tool
 416
 417     if shutil.which(name) is not None:
 418         return name
 419
 420     if fallback is None:
 421         print(f"Tool {name} not installed!")
 422
 423     return fallback
 424
 425 def combine_signatures(pcrsigs):
 426     combined = collections.defaultdict(list)
 427     for pcrsig in pcrsigs:
 428         for bank, sigs in pcrsig.items():
 429             for sig in sigs:
 430                 if sig not in combined[bank]:
 431                     combined[bank] += [sig]
 432     return json.dumps(combined)
 433
 434
 435 def key_path_groups(opts):
 436     if not opts.pcr_private_keys:
 437         return
 438
 439     n_priv = len(opts.pcr_private_keys)
 440     pub_keys = opts.pcr_public_keys or [None] * n_priv
 441     pp_groups = opts.phase_path_groups or [None] * n_priv
 442
 443     yield from zip(opts.pcr_private_keys,
 444                    pub_keys,
 445                    pp_groups)
 446
 447
 448 def call_systemd_measure(uki, linux, opts):
 449     measure_tool = find_tool('systemd-measure',
 450                              '/usr/lib/systemd/systemd-measure',
 451                              opts=opts)
 452
 453     banks = opts.pcr_banks or ()
 454
 455     # PCR measurement
 456
 457     if opts.measure:
 458         pp_groups = opts.phase_path_groups or []
 459
 460         cmd = [
 461             measure_tool,
 462             'calculate',
 463             f'--linux={linux}',
 464             *(f"--{s.name.removeprefix('.')}={s.content}"
 465               for s in uki.sections
 466               if s.measure),
 467             *(f'--bank={bank}'
 468               for bank in banks),
 469             # For measurement, the keys are not relevant, so we can lump all the phase paths
 470             # into one call to systemd-measure calculate.
 471             *(f'--phase={phase_path}'
 472               for phase_path in itertools.chain.from_iterable(pp_groups)),
 473         ]
 474
 475         print('+', shell_join(cmd))
 476         subprocess.check_call(cmd)
 477
 478     # PCR signing
 479
 480     if opts.pcr_private_keys:
 481         pcrsigs = []
 482
 483         cmd = [
 484             measure_tool,
 485             'sign',
 486             f'--linux={linux}',
 487             *(f"--{s.name.removeprefix('.')}={s.content}"
 488               for s in uki.sections
 489               if s.measure),
 490             *(f'--bank={bank}'
 491               for bank in banks),
 492         ]
 493
 494         for priv_key, pub_key, group in key_path_groups(opts):
 495             extra = [f'--private-key={priv_key}']
 496             if pub_key:
 497                 extra += [f'--public-key={pub_key}']
 498             extra += [f'--phase={phase_path}' for phase_path in group or ()]
 499
 500             print('+', shell_join(cmd + extra))
 501             pcrsig = subprocess.check_output(cmd + extra, text=True)
 502             pcrsig = json.loads(pcrsig)
 503             pcrsigs += [pcrsig]
 504
 505         combined = combine_signatures(pcrsigs)
 506         uki.add_section(Section.create('.pcrsig', combined))
 507
 508
 509 def join_initrds(initrds):
 510     if not initrds:
 511         return None
 512     if len(initrds) == 1:
 513         return initrds[0]
 514
 515     seq = []
 516     for file in initrds:
 517         initrd = file.read_bytes()
 518         n = len(initrd)
 519         padding = b'\0' * (round_up(n, 4) - n)  # pad to 32 bit alignment
 520         seq += [initrd, padding]
 521
 522     return b''.join(seq)
 523
 524
 525 def pairwise(iterable):
 526     a, b = itertools.tee(iterable)
 527     next(b, None)
 528     return zip(a, b)
 529
 530
 531 class PEError(Exception):
 532     pass
 533
 534
 535 def pe_add_sections(uki: UKI, output: str):
 536     pe = pefile.PE(uki.executable, fast_load=True)
 537
 538     # Old stubs do not have the symbol/string table stripped, even though image files should not have one.
 539     if symbol_table := pe.FILE_HEADER.PointerToSymbolTable:
 540         symbol_table_size = 18 * pe.FILE_HEADER.NumberOfSymbols
 541         if string_table_size := pe.get_dword_from_offset(symbol_table + symbol_table_size):
 542             symbol_table_size += string_table_size
 543
 544         # Let's be safe and only strip it if it's at the end of the file.
 545         if symbol_table + symbol_table_size == len(pe.__data__):
 546             pe.__data__ = pe.__data__[:symbol_table]
 547             pe.FILE_HEADER.PointerToSymbolTable = 0
 548             pe.FILE_HEADER.NumberOfSymbols = 0
 549             pe.FILE_HEADER.IMAGE_FILE_LOCAL_SYMS_STRIPPED = True
 550
 551     # Old stubs might have been stripped, leading to unaligned raw data values, so let's fix them up here.
 552     # pylint thinks that Structure doesn't have various members that it has…
 553     # pylint: disable=no-member
 554
 555     for i, section in enumerate(pe.sections):
 556         oldp = section.PointerToRawData
 557         oldsz = section.SizeOfRawData
 558         section.PointerToRawData = round_up(oldp, pe.OPTIONAL_HEADER.FileAlignment)
 559         section.SizeOfRawData = round_up(oldsz, pe.OPTIONAL_HEADER.FileAlignment)
 560         padp = section.PointerToRawData - oldp
 561         padsz = section.SizeOfRawData - oldsz
 562
 563         for later_section in pe.sections[i+1:]:
 564             later_section.PointerToRawData += padp + padsz
 565
 566         pe.__data__ = pe.__data__[:oldp] + bytes(padp) + pe.__data__[oldp:oldp+oldsz] + bytes(padsz) + pe.__data__[oldp+oldsz:]
 567
 568     # We might not have any space to add new sections. Let's try our best to make some space by padding the
 569     # SizeOfHeaders to a multiple of the file alignment. This is safe because the first section's data starts
 570     # at a multiple of the file alignment, so all space before that is unused.
 571     pe.OPTIONAL_HEADER.SizeOfHeaders = round_up(pe.OPTIONAL_HEADER.SizeOfHeaders, pe.OPTIONAL_HEADER.FileAlignment)
 572     pe = pefile.PE(data=pe.write(), fast_load=True)
 573
 574     warnings = pe.get_warnings()
 575     if warnings:
 576         raise PEError(f'pefile warnings treated as errors: {warnings}')
 577
 578     security = pe.OPTIONAL_HEADER.DATA_DIRECTORY[pefile.DIRECTORY_ENTRY['IMAGE_DIRECTORY_ENTRY_SECURITY']]
 579     if security.VirtualAddress != 0:
 580         # We could strip the signatures, but why would anyone sign the stub?
 581         raise PEError('Stub image is signed, refusing.')
 582
 583     for section in uki.sections:
 584         new_section = pefile.SectionStructure(pe.__IMAGE_SECTION_HEADER_format__, pe=pe)
 585         new_section.__unpack__(b'\0' * new_section.sizeof())
 586
 587         offset = pe.sections[-1].get_file_offset() + new_section.sizeof()
 588         if offset + new_section.sizeof() > pe.OPTIONAL_HEADER.SizeOfHeaders:
 589             raise PEError(f'Not enough header space to add section {section.name}.')
 590
 591         assert section.content
 592         data = section.content.read_bytes()
 593
 594         new_section.set_file_offset(offset)
 595         new_section.Name = section.name.encode()
 596         new_section.Misc_VirtualSize = len(data)
 597         # Non-stripped stubs might still have an unaligned symbol table at the end, making their size
 598         # unaligned, so we make sure to explicitly pad the pointer to new sections to an aligned offset.
 599         new_section.PointerToRawData = round_up(len(pe.__data__), pe.OPTIONAL_HEADER.FileAlignment)
 600         new_section.SizeOfRawData = round_up(len(data), pe.OPTIONAL_HEADER.FileAlignment)
 601         new_section.VirtualAddress = round_up(
 602             pe.sections[-1].VirtualAddress + pe.sections[-1].Misc_VirtualSize,
 603             pe.OPTIONAL_HEADER.SectionAlignment,
 604         )
 605
 606         new_section.IMAGE_SCN_MEM_READ = True
 607         if section.name == '.linux':
 608             # Old kernels that use EFI handover protocol will be executed inline.
 609             new_section.IMAGE_SCN_CNT_CODE = True
 610         else:
 611             new_section.IMAGE_SCN_CNT_INITIALIZED_DATA = True
 612
 613         # Special case, mostly for .sbat: the stub will already have a .sbat section, but we want to append
 614         # the one from the kernel to it. It should be small enough to fit in the existing section, so just
 615         # swap the data.
 616         for i, s in enumerate(pe.sections):
 617             if s.Name.rstrip(b"\x00").decode() == section.name:
 618                 if new_section.Misc_VirtualSize > s.SizeOfRawData:
 619                     raise PEError(f'Not enough space in existing section {section.name} to append new data.')
 620
 621                 padding = bytes(new_section.SizeOfRawData - new_section.Misc_VirtualSize)
 622                 pe.__data__ = pe.__data__[:s.PointerToRawData] + data + padding + pe.__data__[pe.sections[i+1].PointerToRawData:]
 623                 s.SizeOfRawData = new_section.SizeOfRawData
 624                 s.Misc_VirtualSize = new_section.Misc_VirtualSize
 625                 break
 626         else:
 627             pe.__data__ = pe.__data__[:] + bytes(new_section.PointerToRawData - len(pe.__data__)) + data + bytes(new_section.SizeOfRawData - len(data))
 628
 629             pe.FILE_HEADER.NumberOfSections += 1
 630             pe.OPTIONAL_HEADER.SizeOfInitializedData += new_section.Misc_VirtualSize
 631             pe.__structures__.append(new_section)
 632             pe.sections.append(new_section)
 633
 634     pe.OPTIONAL_HEADER.CheckSum = 0
 635     pe.OPTIONAL_HEADER.SizeOfImage = round_up(
 636         pe.sections[-1].VirtualAddress + pe.sections[-1].Misc_VirtualSize,
 637         pe.OPTIONAL_HEADER.SectionAlignment,
 638     )
 639
 640     pe.write(output)
 641
 642 def merge_sbat(input_pe: [pathlib.Path], input_text: [str]) -> str:
 643     sbat = []
 644
 645     for f in input_pe:
 646         try:
 647             pe = pefile.PE(f, fast_load=True)
 648         except pefile.PEFormatError:
 649             print(f"{f} is not a valid PE file, not extracting SBAT section.")
 650             continue
 651
 652         for section in pe.sections:
 653             if section.Name.rstrip(b"\x00").decode() == ".sbat":
 654                 split = section.get_data().rstrip(b"\x00").decode().splitlines()
 655                 if not split[0].startswith('sbat,'):
 656                     print(f"{f} does not contain a valid SBAT section, skipping.")
 657                     continue
 658                 # Filter out the sbat line, we'll add it back later, there needs to be only one and it
 659                 # needs to be first.
 660                 sbat += split[1:]
 661
 662     for t in input_text:
 663         if t.startswith('@'):
 664             t = pathlib.Path(t[1:]).read_text()
 665         split = t.splitlines()
 666         if not split[0].startswith('sbat,'):
 667             print(f"{t} does not contain a valid SBAT section, skipping.")
 668             continue
 669         sbat += split[1:]
 670
 671     return 'sbat,1,SBAT Version,sbat,1,https://github.com/rhboot/shim/blob/main/SBAT.md\n' + '\n'.join(sbat) + "\n\x00"
 672
 673 def signer_sign(cmd):
 674     print('+', shell_join(cmd))
 675     subprocess.check_call(cmd)
 676
 677 def find_sbsign(opts=None):
 678     return find_tool('sbsign', opts=opts)
 679
 680 def sbsign_sign(sbsign_tool, input_f, output_f, opts=None):
 681     sign_invocation = [
 682         sbsign_tool,
 683         '--key', opts.sb_key,
 684         '--cert', opts.sb_cert,
 685         input_f,
 686         '--output', output_f,
 687     ]
 688     if opts.signing_engine is not None:
 689         sign_invocation += ['--engine', opts.signing_engine]
 690     signer_sign(sign_invocation)
 691
 692 def find_pesign(opts=None):
 693     return find_tool('pesign', opts=opts)
 694
 695 def pesign_sign(pesign_tool, input_f, output_f, opts=None):
 696     sign_invocation = [
 697         pesign_tool, '-s', '--force',
 698         '-n', opts.sb_certdir,
 699         '-c', opts.sb_cert_name,
 700         '-i', input_f,
 701         '-o', output_f,
 702     ]
 703     signer_sign(sign_invocation)
 704
 705 SBVERIFY = {
 706     'name': 'sbverify',
 707     'option': '--list',
 708     'output': 'No signature table present',
 709 }
 710
 711 PESIGCHECK = {
 712     'name': 'pesign',
 713     'option': '-i',
 714     'output': 'No signatures found.',
 715     'flags': '-S'
 716 }
 717
 718 def verify(tool, opts):
 719     verify_tool = find_tool(tool['name'], opts=opts)
 720     cmd = [
 721         verify_tool,
 722         tool['option'],
 723         opts.linux,
 724     ]
 725     if 'flags' in tool:
 726         cmd.append(tool['flags'])
 727
 728     print('+', shell_join(cmd))
 729     info = subprocess.check_output(cmd, text=True)
 730
 731     return tool['output'] in info
 732
 733 def make_uki(opts):
 734     # kernel payload signing
 735
 736     sign_tool = None
 737     sign_args_present = opts.sb_key or opts.sb_cert_name
 738     sign_kernel = opts.sign_kernel
 739     sign = None
 740     linux = opts.linux
 741
 742     if sign_args_present:
 743         if opts.signtool == 'sbsign':
 744             sign_tool = find_sbsign(opts=opts)
 745             sign = sbsign_sign
 746             verify_tool = SBVERIFY
 747         else:
 748             sign_tool = find_pesign(opts=opts)
 749             sign = pesign_sign
 750             verify_tool = PESIGCHECK
 751
 752         if sign_tool is None:
 753             raise ValueError(f'{opts.signtool}, required for signing, is not installed')
 754
 755         if sign_kernel is None and opts.linux is not None:
 756             # figure out if we should sign the kernel
 757             sign_kernel = verify(verify_tool, opts)
 758
 759         if sign_kernel:
 760             linux_signed = tempfile.NamedTemporaryFile(prefix='linux-signed')
 761             linux = pathlib.Path(linux_signed.name)
 762             sign(sign_tool, opts.linux, linux, opts=opts)
 763
 764     if opts.uname is None and opts.linux is not None:
 765         print('Kernel version not specified, starting autodetection 😖.')
 766         opts.uname = Uname.scrape(opts.linux, opts=opts)
 767
 768     uki = UKI(opts.stub)
 769     initrd = join_initrds(opts.initrd)
 770
 771     pcrpkey = opts.pcrpkey
 772     if pcrpkey is None:
 773         if opts.pcr_public_keys and len(opts.pcr_public_keys) == 1:
 774             pcrpkey = opts.pcr_public_keys[0]
 775         elif opts.pcr_private_keys and len(opts.pcr_private_keys) == 1:
 776             from cryptography.hazmat.primitives import serialization
 777             privkey = serialization.load_pem_private_key(opts.pcr_private_keys[0].read_bytes(), password=None)
 778             pcrpkey = privkey.public_key().public_bytes(
 779                 encoding=serialization.Encoding.PEM,
 780                 format=serialization.PublicFormat.SubjectPublicKeyInfo,
 781             )
 782
 783     sections = [
 784         # name,      content,         measure?
 785         ('.osrel',   opts.os_release, True ),
 786         ('.cmdline', opts.cmdline,    True ),
 787         ('.dtb',     opts.devicetree, True ),
 788         ('.uname',   opts.uname,      True ),
 789         ('.splash',  opts.splash,     True ),
 790         ('.pcrpkey', pcrpkey,         True ),
 791         ('.initrd',  initrd,          True ),
 792
 793         # linux shall be last to leave breathing room for decompression.
 794         # We'll add it later.
 795     ]
 796
 797     for name, content, measure in sections:
 798         if content:
 799             uki.add_section(Section.create(name, content, measure=measure))
 800
 801     # systemd-measure doesn't know about those extra sections
 802     for section in opts.sections:
 803         uki.add_section(section)
 804
 805     if linux is not None:
 806         # Merge the .sbat sections from stub, kernel and parameter, so that revocation can be done on either.
 807         uki.add_section(Section.create('.sbat', merge_sbat([opts.stub, linux], opts.sbat), measure=True))
 808     else:
 809         # Addons don't use the stub so we add SBAT manually
 810         if not opts.sbat:
 811             opts.sbat = ["""sbat,1,SBAT Version,sbat,1,https://github.com/rhboot/shim/blob/main/SBAT.md
 812 uki,1,UKI,uki,1,https://www.freedesktop.org/software/systemd/man/systemd-stub.html
 813 """]
 814         uki.add_section(Section.create('.sbat', merge_sbat([], opts.sbat), measure=False))
 815
 816     # PCR measurement and signing
 817
 818     # We pass in the contents for .linux separately because we need them to do the measurement but can't add
 819     # the section yet because we want .linux to be the last section. Make sure any other sections are added
 820     # before this function is called.
 821     call_systemd_measure(uki, linux, opts=opts)
 822
 823     # UKI creation
 824
 825     if linux is not None:
 826         uki.add_section(Section.create('.linux', linux, measure=True))
 827
 828     if sign_args_present:
 829         unsigned = tempfile.NamedTemporaryFile(prefix='uki')
 830         unsigned_output = unsigned.name
 831     else:
 832         unsigned_output = opts.output
 833
 834     pe_add_sections(uki, unsigned_output)
 835
 836     # UKI signing
 837
 838     if sign_args_present:
 839         assert sign
 840         sign(sign_tool, unsigned_output, opts.output, opts=opts)
 841
 842         # We end up with no executable bits, let's reapply them
 843         os.umask(umask := os.umask(0))
 844         os.chmod(opts.output, 0o777 & ~umask)
 845
 846     print(f"Wrote {'signed' if sign_args_present else 'unsigned'} {opts.output}")
 847
 848
 849 @contextlib.contextmanager
 850 def temporary_umask(mask: int):
 851     # Drop <mask> bits from umask
 852     old = os.umask(0)
 853     os.umask(old | mask)
 854     try:
 855         yield
 856     finally:
 857         os.umask(old)
 858
 859
 860 def generate_key_cert_pair(
 861         common_name: str,
 862         valid_days: int,
 863         keylength: int = 2048,
 864 ) -> tuple[bytes]:
 865
 866     from cryptography import x509
 867     from cryptography.hazmat.primitives import serialization, hashes
 868     from cryptography.hazmat.primitives.asymmetric import rsa
 869
 870     # We use a keylength of 2048 bits. That is what Microsoft documents as
 871     # supported/expected:
 872     # https://learn.microsoft.com/en-us/windows-hardware/manufacture/desktop/windows-secure-boot-key-creation-and-management-guidance?view=windows-11#12-public-key-cryptography
 873
 874     now = datetime.datetime.now(datetime.UTC)
 875
 876     key = rsa.generate_private_key(
 877         public_exponent=65537,
 878         key_size=keylength,
 879     )
 880     cert = x509.CertificateBuilder(
 881     ).subject_name(
 882         x509.Name([x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, common_name)])
 883     ).issuer_name(
 884         x509.Name([x509.NameAttribute(x509.oid.NameOID.COMMON_NAME, common_name)])
 885     ).not_valid_before(
 886         now,
 887     ).not_valid_after(
 888         now + datetime.timedelta(days=valid_days)
 889     ).serial_number(
 890         x509.random_serial_number()
 891     ).public_key(
 892         key.public_key()
 893     ).add_extension(
 894         x509.BasicConstraints(ca=False, path_length=None),
 895         critical=True,
 896     ).sign(
 897         private_key=key,
 898         algorithm=hashes.SHA256(),
 899     )
 900
 901     cert_pem = cert.public_bytes(
 902         encoding=serialization.Encoding.PEM,
 903     )
 904     key_pem = key.private_bytes(
 905         encoding=serialization.Encoding.PEM,
 906         format=serialization.PrivateFormat.TraditionalOpenSSL,
 907         encryption_algorithm=serialization.NoEncryption(),
 908     )
 909
 910     return key_pem, cert_pem
 911
 912
 913 def generate_priv_pub_key_pair(keylength : int = 2048) -> tuple[bytes]:
 914     from cryptography.hazmat.primitives import serialization
 915     from cryptography.hazmat.primitives.asymmetric import rsa
 916
 917     key = rsa.generate_private_key(
 918         public_exponent=65537,
 919         key_size=keylength,
 920     )
 921     priv_key_pem = key.private_bytes(
 922         encoding=serialization.Encoding.PEM,
 923         format=serialization.PrivateFormat.TraditionalOpenSSL,
 924         encryption_algorithm=serialization.NoEncryption(),
 925     )
 926     pub_key_pem = key.public_key().public_bytes(
 927         encoding=serialization.Encoding.PEM,
 928         format=serialization.PublicFormat.SubjectPublicKeyInfo,
 929     )
 930
 931     return priv_key_pem, pub_key_pem
 932
 933
 934 def generate_keys(opts):
 935     work = False
 936
 937     # This will generate keys and certificates and write them to the paths that
 938     # are specified as input paths.
 939     if opts.sb_key or opts.sb_cert:
 940         fqdn = socket.getfqdn()
 941         cn = f'SecureBoot signing key on host {fqdn}'
 942         key_pem, cert_pem = generate_key_cert_pair(
 943             common_name=cn,
 944             valid_days=opts.sb_cert_validity,
 945         )
 946         print(f'Writing SecureBoot private key to {opts.sb_key}')
 947         with temporary_umask(0o077):
 948             opts.sb_key.write_bytes(key_pem)
 949         print(f'Writing SecureBoot certificate to {opts.sb_cert}')
 950         opts.sb_cert.write_bytes(cert_pem)
 951
 952         work = True
 953
 954     for priv_key, pub_key, _ in key_path_groups(opts):
 955         priv_key_pem, pub_key_pem = generate_priv_pub_key_pair()
 956
 957         print(f'Writing private key for PCR signing to {priv_key}')
 958         with temporary_umask(0o077):
 959             priv_key.write_bytes(priv_key_pem)
 960         if pub_key:
 961             print(f'Writing public key for PCR signing to {pub_key}')
 962             pub_key.write_bytes(pub_key_pem)
 963
 964         work = True
 965
 966     if not work:
 967         raise ValueError('genkey: --secureboot-private-key=/--secureboot-certificate= or --pcr-private-key/--pcr-public-key must be specified')
 968
 969
 970 def inspect_section(opts, section):
 971     name = section.Name.rstrip(b"\x00").decode()
 972
 973     # find the config for this section in opts and whether to show it
 974     config = opts.sections_by_name.get(name, None)
 975     show = (config or
 976             opts.all or
 977             (name in DEFAULT_SECTIONS_TO_SHOW and not opts.sections))
 978     if not show:
 979         return name, None
 980
 981     ttype = config.output_mode if config else DEFAULT_SECTIONS_TO_SHOW.get(name, 'binary')
 982
 983     size = section.Misc_VirtualSize
 984     # TODO: Use ignore_padding once we can depend on a newer version of pefile
 985     data = section.get_data(length=size)
 986     digest = sha256(data).hexdigest()
 987
 988     struct = {
 989         'size' : size,
 990         'sha256' : digest,
 991     }
 992
 993     if ttype == 'text':
 994         try:
 995             struct['text'] = data.decode()
 996         except UnicodeDecodeError as e:
 997             print(f"Section {name!r} is not valid text: {e}")
 998             struct['text'] = '(not valid UTF-8)'
 999
1000     if config and config.content:
1001         assert isinstance(config.content, pathlib.Path)
1002         config.content.write_bytes(data)
1003
1004     if opts.json == 'off':
1005         print(f"{name}:\n  size: {size} bytes\n  sha256: {digest}")
1006         if ttype == 'text':
1007             text = textwrap.indent(struct['text'].rstrip(), ' ' * 4)
1008             print(f"  text:\n{text}")
1009
1010     return name, struct
1011
1012
1013 def inspect_sections(opts):
1014     indent = 4 if opts.json == 'pretty' else None
1015
1016     for file in opts.files:
1017         pe = pefile.PE(file, fast_load=True)
1018         gen = (inspect_section(opts, section) for section in pe.sections)
1019         descs = {key:val for (key, val) in gen if val}
1020         if opts.json != 'off':
1021             json.dump(descs, sys.stdout, indent=indent)
1022
1023
1024 @dataclasses.dataclass(frozen=True)
1025 class ConfigItem:
1026     @staticmethod
1027     def config_list_prepend(
1028             namespace: argparse.Namespace,
1029             group: Optional[str],
1030             dest: str,
1031             value: Any,
1032     ) -> None:
1033         "Prepend value to namespace.<dest>"
1034
1035         assert not group
1036
1037         old = getattr(namespace, dest, [])
1038         if old is None:
1039             old = []
1040         setattr(namespace, dest, value + old)
1041
1042     @staticmethod
1043     def config_set_if_unset(
1044             namespace: argparse.Namespace,
1045             group: Optional[str],
1046             dest: str,
1047             value: Any,
1048     ) -> None:
1049         "Set namespace.<dest> to value only if it was None"
1050
1051         assert not group
1052
1053         if getattr(namespace, dest) is None:
1054             setattr(namespace, dest, value)
1055
1056     @staticmethod
1057     def config_set(
1058             namespace: argparse.Namespace,
1059             group: Optional[str],
1060             dest: str,
1061             value: Any,
1062     ) -> None:
1063         "Set namespace.<dest> to value only if it was None"
1064
1065         assert not group
1066
1067         setattr(namespace, dest, value)
1068
1069     @staticmethod
1070     def config_set_group(
1071             namespace: argparse.Namespace,
1072             group: Optional[str],
1073             dest: str,
1074             value: Any,
1075     ) -> None:
1076         "Set namespace.<dest>[idx] to value, with idx derived from group"
1077
1078         # pylint: disable=protected-access
1079         if group not in namespace._groups:
1080             namespace._groups += [group]
1081         idx = namespace._groups.index(group)
1082
1083         old = getattr(namespace, dest, None)
1084         if old is None:
1085             old = []
1086         setattr(namespace, dest,
1087                 old + ([None] * (idx - len(old))) + [value])
1088
1089     @staticmethod
1090     def parse_boolean(s: str) -> bool:
1091         "Parse 1/true/yes/y/t/on as true and 0/false/no/n/f/off/None as false"
1092         s_l = s.lower()
1093         if s_l in {'1', 'true', 'yes', 'y', 't', 'on'}:
1094             return True
1095         if s_l in {'0', 'false', 'no', 'n', 'f', 'off'}:
1096             return False
1097         raise ValueError('f"Invalid boolean literal: {s!r}')
1098
1099     # arguments for argparse.ArgumentParser.add_argument()
1100     name: Union[str, tuple[str, str]]
1101     dest: Optional[str] = None
1102     metavar: Optional[str] = None
1103     type: Optional[Callable] = None
1104     nargs: Optional[str] = None
1105     action: Optional[Union[str, Callable]] = None
1106     default: Any = None
1107     version: Optional[str] = None
1108     choices: Optional[tuple[str, ...]] = None
1109     const: Optional[Any] = None
1110     help: Optional[str] = None
1111
1112     # metadata for config file parsing
1113     config_key: Optional[str] = None
1114     config_push: Callable[[argparse.Namespace, Optional[str], str, Any], None] = \
1115                     config_set_if_unset
1116
1117     def _names(self) -> tuple[str, ...]:
1118         return self.name if isinstance(self.name, tuple) else (self.name,)
1119
1120     def argparse_dest(self) -> str:
1121         # It'd be nice if argparse exported this, but I don't see that in the API
1122         if self.dest:
1123             return self.dest
1124         return self._names()[0].lstrip('-').replace('-', '_')
1125
1126     def add_to(self, parser: argparse.ArgumentParser):
1127         kwargs = { key:val
1128                    for key in dataclasses.asdict(self)
1129                    if (key not in ('name', 'config_key', 'config_push') and
1130                        (val := getattr(self, key)) is not None) }
1131         args = self._names()
1132         parser.add_argument(*args, **kwargs)
1133
1134     def apply_config(self, namespace, section, group, key, value) -> None:
1135         assert f'{section}/{key}' == self.config_key
1136         dest = self.argparse_dest()
1137
1138         conv: Callable[[str], Any]
1139         if self.action == argparse.BooleanOptionalAction:
1140             # We need to handle this case separately: the options are called
1141             # --foo and --no-foo, and no argument is parsed. But in the config
1142             # file, we have Foo=yes or Foo=no.
1143             conv = self.parse_boolean
1144         elif self.type:
1145             conv = self.type
1146         else:
1147             conv = lambda s:s
1148
1149         # This is a bit ugly, but --initrd is the only option which is specified
1150         # with multiple args on the command line and a space-separated list in the
1151         # config file.
1152         if self.name == '--initrd':
1153             value = [conv(v) for v in value.split()]
1154         else:
1155             value = conv(value)
1156
1157         self.config_push(namespace, group, dest, value)
1158
1159     def config_example(self) -> tuple[Optional[str], Optional[str], Optional[str]]:
1160         if not self.config_key:
1161             return None, None, None
1162         section_name, key = self.config_key.split('/', 1)
1163         if section_name.endswith(':'):
1164             section_name += 'NAME'
1165         if self.choices:
1166             value = '|'.join(self.choices)
1167         else:
1168             value = self.metavar or self.argparse_dest().upper()
1169         return (section_name, key, value)
1170
1171
1172 VERBS = ('build', 'genkey', 'inspect')
1173
1174 CONFIG_ITEMS = [
1175     ConfigItem(
1176         'positional',
1177         metavar = 'VERB',
1178         nargs = '*',
1179         help = argparse.SUPPRESS,
1180     ),
1181
1182     ConfigItem(
1183         '--version',
1184         action = 'version',
1185         version = f'ukify {__version__}',
1186     ),
1187
1188     ConfigItem(
1189         '--summary',
1190         help = 'print parsed config and exit',
1191         action = 'store_true',
1192     ),
1193
1194     ConfigItem(
1195         '--linux',
1196         type = pathlib.Path,
1197         help = 'vmlinuz file [.linux section]',
1198         config_key = 'UKI/Linux',
1199     ),
1200
1201     ConfigItem(
1202         '--initrd',
1203         metavar = 'INITRD',
1204         type = pathlib.Path,
1205         action = 'append',
1206         help = 'initrd file [part of .initrd section]',
1207         config_key = 'UKI/Initrd',
1208         config_push = ConfigItem.config_list_prepend,
1209     ),
1210
1211     ConfigItem(
1212         ('--config', '-c'),
1213         metavar = 'PATH',
1214         type = pathlib.Path,
1215         help = 'configuration file',
1216     ),
1217
1218     ConfigItem(
1219         '--cmdline',
1220         metavar = 'TEXT|@PATH',
1221         help = 'kernel command line [.cmdline section]',
1222         config_key = 'UKI/Cmdline',
1223     ),
1224
1225     ConfigItem(
1226         '--os-release',
1227         metavar = 'TEXT|@PATH',
1228         help = 'path to os-release file [.osrel section]',
1229         config_key = 'UKI/OSRelease',
1230     ),
1231
1232     ConfigItem(
1233         '--devicetree',
1234         metavar = 'PATH',
1235         type = pathlib.Path,
1236         help = 'Device Tree file [.dtb section]',
1237         config_key = 'UKI/DeviceTree',
1238     ),
1239     ConfigItem(
1240         '--splash',
1241         metavar = 'BMP',
1242         type = pathlib.Path,
1243         help = 'splash image bitmap file [.splash section]',
1244         config_key = 'UKI/Splash',
1245     ),
1246     ConfigItem(
1247         '--pcrpkey',
1248         metavar = 'KEY',
1249         type = pathlib.Path,
1250         help = 'embedded public key to seal secrets to [.pcrpkey section]',
1251         config_key = 'UKI/PCRPKey',
1252     ),
1253     ConfigItem(
1254         '--uname',
1255         metavar='VERSION',
1256         help='"uname -r" information [.uname section]',
1257         config_key = 'UKI/Uname',
1258     ),
1259
1260     ConfigItem(
1261         '--efi-arch',
1262         metavar = 'ARCH',
1263         choices = ('ia32', 'x64', 'arm', 'aa64', 'riscv64'),
1264         help = 'target EFI architecture',
1265         config_key = 'UKI/EFIArch',
1266     ),
1267
1268     ConfigItem(
1269         '--stub',
1270         type = pathlib.Path,
1271         help = 'path to the sd-stub file [.text,.data,… sections]',
1272         config_key = 'UKI/Stub',
1273     ),
1274
1275     ConfigItem(
1276         '--sbat',
1277         metavar = 'TEXT|@PATH',
1278         help = 'SBAT policy [.sbat section]',
1279         default = [],
1280         action = 'append',
1281         config_key = 'UKI/SBAT',
1282     ),
1283
1284     ConfigItem(
1285         '--section',
1286         dest = 'sections',
1287         metavar = 'NAME:TEXT|@PATH',
1288         action = 'append',
1289         default = [],
1290         help = 'section as name and contents [NAME section] or section to print',
1291     ),
1292
1293     ConfigItem(
1294         '--pcr-banks',
1295         metavar = 'BANK…',
1296         type = parse_banks,
1297         config_key = 'UKI/PCRBanks',
1298     ),
1299
1300     ConfigItem(
1301         '--signing-engine',
1302         metavar = 'ENGINE',
1303         help = 'OpenSSL engine to use for signing',
1304         config_key = 'UKI/SigningEngine',
1305     ),
1306     ConfigItem(
1307         '--signtool',
1308         choices = ('sbsign', 'pesign'),
1309         dest = 'signtool',
1310         help = 'whether to use sbsign or pesign. It will also be inferred by the other \
1311         parameters given: when using --secureboot-{private-key/certificate}, sbsign \
1312         will be used, otherwise pesign will be used',
1313         config_key = 'UKI/SecureBootSigningTool',
1314     ),
1315     ConfigItem(
1316         '--secureboot-private-key',
1317         dest = 'sb_key',
1318         help = 'required by --signtool=sbsign. Path to key file or engine-specific designation for SB signing',
1319         config_key = 'UKI/SecureBootPrivateKey',
1320     ),
1321     ConfigItem(
1322         '--secureboot-certificate',
1323         dest = 'sb_cert',
1324         help = 'required by --signtool=sbsign. sbsign needs a path to certificate file or engine-specific designation for SB signing',
1325         config_key = 'UKI/SecureBootCertificate',
1326     ),
1327     ConfigItem(
1328         '--secureboot-certificate-dir',
1329         dest = 'sb_certdir',
1330         default = '/etc/pki/pesign',
1331         help = 'required by --signtool=pesign. Path to nss certificate database directory for PE signing. Default is /etc/pki/pesign',
1332         config_key = 'UKI/SecureBootCertificateDir',
1333         config_push = ConfigItem.config_set
1334     ),
1335     ConfigItem(
1336         '--secureboot-certificate-name',
1337         dest = 'sb_cert_name',
1338         help = 'required by --signtool=pesign. pesign needs a certificate nickname of nss certificate database entry to use for PE signing',
1339         config_key = 'UKI/SecureBootCertificateName',
1340     ),
1341     ConfigItem(
1342         '--secureboot-certificate-validity',
1343         metavar = 'DAYS',
1344         type = int,
1345         dest = 'sb_cert_validity',
1346         default = 365 * 10,
1347         help = "period of validity (in days) for a certificate created by 'genkey'",
1348         config_key = 'UKI/SecureBootCertificateValidity',
1349         config_push = ConfigItem.config_set
1350     ),
1351
1352     ConfigItem(
1353         '--sign-kernel',
1354         action = argparse.BooleanOptionalAction,
1355         help = 'Sign the embedded kernel',
1356         config_key = 'UKI/SignKernel',
1357     ),
1358
1359     ConfigItem(
1360         '--pcr-private-key',
1361         dest = 'pcr_private_keys',
1362         metavar = 'PATH',
1363         type = pathlib.Path,
1364         action = 'append',
1365         help = 'private part of the keypair for signing PCR signatures',
1366         config_key = 'PCRSignature:/PCRPrivateKey',
1367         config_push = ConfigItem.config_set_group,
1368     ),
1369     ConfigItem(
1370         '--pcr-public-key',
1371         dest = 'pcr_public_keys',
1372         metavar = 'PATH',
1373         type = pathlib.Path,
1374         action = 'append',
1375         help = 'public part of the keypair for signing PCR signatures',
1376         config_key = 'PCRSignature:/PCRPublicKey',
1377         config_push = ConfigItem.config_set_group,
1378     ),
1379     ConfigItem(
1380         '--phases',
1381         dest = 'phase_path_groups',
1382         metavar = 'PHASE-PATH…',
1383         type = parse_phase_paths,
1384         action = 'append',
1385         help = 'phase-paths to create signatures for',
1386         config_key = 'PCRSignature:/Phases',
1387         config_push = ConfigItem.config_set_group,
1388     ),
1389
1390     ConfigItem(
1391         '--tools',
1392         type = pathlib.Path,
1393         action = 'append',
1394         help = 'Directories to search for tools (systemd-measure, …)',
1395     ),
1396
1397     ConfigItem(
1398         ('--output', '-o'),
1399         type = pathlib.Path,
1400         help = 'output file path',
1401     ),
1402
1403     ConfigItem(
1404         '--measure',
1405         action = argparse.BooleanOptionalAction,
1406         help = 'print systemd-measure output for the UKI',
1407     ),
1408
1409     ConfigItem(
1410         '--json',
1411         choices = ('pretty', 'short', 'off'),
1412         default = 'off',
1413         help = 'generate JSON output',
1414     ),
1415     ConfigItem(
1416         '-j',
1417         dest='json',
1418         action='store_const',
1419         const='pretty',
1420         help='equivalent to --json=pretty',
1421     ),
1422
1423     ConfigItem(
1424         '--all',
1425         help = 'print all sections',
1426         action = 'store_true',
1427     ),
1428 ]
1429
1430 CONFIGFILE_ITEMS = { item.config_key:item
1431                      for item in CONFIG_ITEMS
1432                      if item.config_key }
1433
1434
1435 def apply_config(namespace, filename=None):
1436     if filename is None:
1437         if namespace.config:
1438             # Config set by the user, use that.
1439             filename = namespace.config
1440             print(f'Using config file: {filename}')
1441         else:
1442             # Try to look for a config file then use the first one found.
1443             for config_dir in DEFAULT_CONFIG_DIRS:
1444                 filename = pathlib.Path(config_dir) / DEFAULT_CONFIG_FILE
1445                 if filename.is_file():
1446                     # Found a config file, use it.
1447                     print(f'Using found config file: {filename}')
1448                     break
1449             else:
1450                 # No config file specified or found, nothing to do.
1451                 return
1452
1453     # Fill in ._groups based on --pcr-public-key=, --pcr-private-key=, and --phases=.
1454     assert '_groups' not in namespace
1455     n_pcr_priv = len(namespace.pcr_private_keys or ())
1456     namespace._groups = list(range(n_pcr_priv))  # pylint: disable=protected-access
1457
1458     cp = configparser.ConfigParser(
1459         comment_prefixes='#',
1460         inline_comment_prefixes='#',
1461         delimiters='=',
1462         empty_lines_in_values=False,
1463         interpolation=None,
1464         strict=False)
1465     # Do not make keys lowercase
1466     cp.optionxform = lambda option: option
1467
1468     # The API is not great.
1469     read = cp.read(filename)
1470     if not read:
1471         raise IOError(f'Failed to read {filename}')
1472
1473     for section_name, section in cp.items():
1474         idx = section_name.find(':')
1475         if idx >= 0:
1476             section_name, group = section_name[:idx+1], section_name[idx+1:]
1477             if not section_name or not group:
1478                 raise ValueError('Section name components cannot be empty')
1479             if ':' in group:
1480                 raise ValueError('Section name cannot contain more than one ":"')
1481         else:
1482             group = None
1483         for key, value in section.items():
1484             if item := CONFIGFILE_ITEMS.get(f'{section_name}/{key}'):
1485                 item.apply_config(namespace, section_name, group, key, value)
1486             else:
1487                 print(f'Unknown config setting [{section_name}] {key}=')
1488
1489
1490 def config_example():
1491     prev_section = None
1492     for item in CONFIG_ITEMS:
1493         section, key, value = item.config_example()
1494         if section:
1495             if prev_section != section:
1496                 if prev_section:
1497                     yield ''
1498                 yield f'[{section}]'
1499                 prev_section = section
1500             yield f'{key} = {value}'
1501
1502
1503 class PagerHelpAction(argparse._HelpAction):  # pylint: disable=protected-access
1504     def __call__(
1505         self,
1506         parser: argparse.ArgumentParser,
1507         namespace: argparse.Namespace,
1508         values: Union[str, Sequence[Any], None] = None,
1509         option_string: Optional[str] = None
1510     ) -> None:
1511         page(parser.format_help(), True)
1512         parser.exit()
1513
1514
1515 def create_parser():
1516     p = argparse.ArgumentParser(
1517         description='Build and sign Unified Kernel Images',
1518         usage='\n  ' + textwrap.dedent('''\
1519           ukify {b}build{e} [--linux=LINUX] [--initrd=INITRD] [options…]
1520             ukify {b}genkey{e} [options…]
1521             ukify {b}inspect{e} FILE… [options…]
1522         ''').format(b=Style.bold, e=Style.reset),
1523         allow_abbrev=False,
1524         add_help=False,
1525         epilog='\n  '.join(('config file:', *config_example())),
1526         formatter_class=argparse.RawDescriptionHelpFormatter,
1527     )
1528
1529     for item in CONFIG_ITEMS:
1530         item.add_to(p)
1531
1532     # Suppress printing of usage synopsis on errors
1533     p.error = lambda message: p.exit(2, f'{p.prog}: error: {message}\n')
1534
1535     # Make --help paged
1536     p.add_argument(
1537         '-h', '--help',
1538         action=PagerHelpAction,
1539         help='show this help message and exit',
1540     )
1541
1542     return p
1543
1544
1545 def finalize_options(opts):
1546     # Figure out which syntax is being used, one of:
1547     # ukify verb --arg --arg --arg
1548     # ukify linux initrd…
1549     if len(opts.positional) >= 1 and opts.positional[0] == 'inspect':
1550         opts.verb = opts.positional[0]
1551         opts.files = opts.positional[1:]
1552         if not opts.files:
1553             raise ValueError('file(s) to inspect must be specified')
1554         if len(opts.files) > 1 and opts.json != 'off':
1555             # We could allow this in the future, but we need to figure out the right structure
1556             raise ValueError('JSON output is not allowed with multiple files')
1557     elif len(opts.positional) == 1 and opts.positional[0] in VERBS:
1558         opts.verb = opts.positional[0]
1559     elif opts.linux or opts.initrd:
1560         raise ValueError('--linux/--initrd options cannot be used with positional arguments')
1561     else:
1562         print("Assuming obsolete command line syntax with no verb. Please use 'build'.")
1563         if opts.positional:
1564             opts.linux = pathlib.Path(opts.positional[0])
1565         # If we have initrds from parsing config files, append our positional args at the end
1566         opts.initrd = (opts.initrd or []) + [pathlib.Path(arg) for arg in opts.positional[1:]]
1567         opts.verb = 'build'
1568
1569     # Check that --pcr-public-key=, --pcr-private-key=, and --phases=
1570     # have either the same number of arguments are are not specified at all.
1571     n_pcr_pub = None if opts.pcr_public_keys is None else len(opts.pcr_public_keys)
1572     n_pcr_priv = None if opts.pcr_private_keys is None else len(opts.pcr_private_keys)
1573     n_phase_path_groups = None if opts.phase_path_groups is None else len(opts.phase_path_groups)
1574     if n_pcr_pub is not None and n_pcr_pub != n_pcr_priv:
1575         raise ValueError('--pcr-public-key= specifications must match --pcr-private-key=')
1576     if n_phase_path_groups is not None and n_phase_path_groups != n_pcr_priv:
1577         raise ValueError('--phases= specifications must match --pcr-private-key=')
1578
1579     if opts.cmdline and opts.cmdline.startswith('@'):
1580         opts.cmdline = pathlib.Path(opts.cmdline[1:])
1581     elif opts.cmdline:
1582         # Drop whitespace from the command line. If we're reading from a file,
1583         # we copy the contents verbatim. But configuration specified on the command line
1584         # or in the config file may contain additional whitespace that has no meaning.
1585         opts.cmdline = ' '.join(opts.cmdline.split())
1586
1587     if opts.os_release and opts.os_release.startswith('@'):
1588         opts.os_release = pathlib.Path(opts.os_release[1:])
1589     elif not opts.os_release and opts.linux:
1590         p = pathlib.Path('/etc/os-release')
1591         if not p.exists():
1592             p = pathlib.Path('/usr/lib/os-release')
1593         opts.os_release = p
1594
1595     if opts.efi_arch is None:
1596         opts.efi_arch = guess_efi_arch()
1597
1598     if opts.stub is None:
1599         if opts.linux is not None:
1600             opts.stub = pathlib.Path(f'/usr/lib/systemd/boot/efi/linux{opts.efi_arch}.efi.stub')
1601         else:
1602             opts.stub = pathlib.Path(f'/usr/lib/systemd/boot/efi/addon{opts.efi_arch}.efi.stub')
1603
1604     if opts.signing_engine is None:
1605         if opts.sb_key:
1606             opts.sb_key = pathlib.Path(opts.sb_key)
1607         if opts.sb_cert:
1608             opts.sb_cert = pathlib.Path(opts.sb_cert)
1609
1610     if bool(opts.sb_key) ^ bool(opts.sb_cert):
1611         # one param only given, sbsign needs both
1612         raise ValueError('--secureboot-private-key= and --secureboot-certificate= must be specified together')
1613     elif bool(opts.sb_key) and bool(opts.sb_cert):
1614         # both param given, infer sbsign and in case it was given, ensure signtool=sbsign
1615         if opts.signtool and opts.signtool != 'sbsign':
1616             raise ValueError(f'Cannot provide --signtool={opts.signtool} with --secureboot-private-key= and --secureboot-certificate=')
1617         opts.signtool = 'sbsign'
1618     elif bool(opts.sb_cert_name):
1619         # sb_cert_name given, infer pesign and in case it was given, ensure signtool=pesign
1620         if opts.signtool and opts.signtool != 'pesign':
1621             raise ValueError(f'Cannot provide --signtool={opts.signtool} with --secureboot-certificate-name=')
1622         opts.signtool = 'pesign'
1623
1624     if opts.sign_kernel and not opts.sb_key and not opts.sb_cert_name:
1625         raise ValueError('--sign-kernel requires either --secureboot-private-key= and --secureboot-certificate= (for sbsign) or --secureboot-certificate-name= (for pesign) to be specified')
1626
1627     if opts.verb == 'build' and opts.output is None:
1628         if opts.linux is None:
1629             raise ValueError('--output= must be specified when building a PE addon')
1630         suffix = '.efi' if opts.sb_key or opts.sb_cert_name else '.unsigned.efi'
1631         opts.output = opts.linux.name + suffix
1632
1633     # Now that we know if we're inputting or outputting, really parse section config
1634     f = Section.parse_output if opts.verb == 'inspect' else Section.parse_input
1635     opts.sections = [f(s) for s in opts.sections]
1636     # A convenience dictionary to make it easy to look up sections
1637     opts.sections_by_name = {s.name:s for s in opts.sections}
1638
1639     if opts.summary:
1640         # TODO: replace pprint() with some fancy formatting.
1641         pprint.pprint(vars(opts))
1642         sys.exit()
1643
1644
1645 def parse_args(args=None):
1646     opts = create_parser().parse_args(args)
1647     apply_config(opts)
1648     finalize_options(opts)
1649     return opts
1650
1651
1652 def main():
1653     opts = parse_args()
1654     if opts.verb == 'build':
1655         check_inputs(opts)
1656         make_uki(opts)
1657     elif opts.verb == 'genkey':
1658         check_cert_and_keys_nonexistent(opts)
1659         generate_keys(opts)
1660     elif opts.verb == 'inspect':
1661         inspect_sections(opts)
1662     else:
1663         assert False
1664
1665
1666 if __name__ == '__main__':
1667     main()