verify.py raw

   1  #!/usr/bin/env python3
   2  # Copyright (c) 2020-2021 The Limenka developers
   3  # Distributed under the MIT software license, see the accompanying
   4  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5  """Script for verifying Limenka release binaries.
   6  
   7  This script attempts to download the sum file SHA256SUMS and corresponding
   8  signature file SHA256SUMS.asc from limenkacore.org and limenka.org and
   9  compares them.
  10  
  11  The sum-signature file is signed by a number of builder keys. This script
  12  ensures that there is a minimum threshold of signatures from pubkeys that
  13  we trust. This trust is articulated on the basis of configuration options
  14  here, but by default is based upon local GPG trust settings.
  15  
  16  The builder keys are available in the guix.sigs repo:
  17  
  18      https://github.com/limenka/guix.sigs/tree/main/builder-keys
  19  
  20  If a minimum good, trusted signature threshold is met on the sum file, we then
  21  download the files specified in SHA256SUMS, and check if the hashes of these
  22  files match those that are specified. The script returns 0 if everything passes
  23  the checks. It returns 1 if either the signature check or the hash check
  24  doesn't pass. If an error occurs the return value is >= 2.
  25  
  26  Logging output goes to stderr and final binary verification data goes to stdout.
  27  
  28  JSON output can by obtained by setting env BINVERIFY_JSON=1.
  29  """
  30  import argparse
  31  import difflib
  32  import json
  33  import logging
  34  import os
  35  import subprocess
  36  import typing as t
  37  import re
  38  import sys
  39  import shutil
  40  import tempfile
  41  import textwrap
  42  import urllib.request
  43  import urllib.error
  44  import enum
  45  from hashlib import sha256
  46  from pathlib import PurePath, Path
  47  
  48  # The primary host; this will fail if we can't retrieve files from here.
  49  HOST1 = "https://limenkacore.org"
  50  HOST2 = "https://limenka.org"
  51  VERSIONPREFIX = "limenka-"
  52  SUMS_FILENAME = 'SHA256SUMS'
  53  SIGNATUREFILENAME = f"{SUMS_FILENAME}.asc"
  54  
  55  
  56  class ReturnCode(enum.IntEnum):
  57      SUCCESS = 0
  58      INTEGRITY_FAILURE = 1
  59      FILE_GET_FAILED = 4
  60      FILE_MISSING_FROM_ONE_HOST = 5
  61      FILES_NOT_EQUAL = 6
  62      NO_BINARIES_MATCH = 7
  63      NOT_ENOUGH_GOOD_SIGS = 9
  64      BINARY_DOWNLOAD_FAILED = 10
  65      BAD_VERSION = 11
  66  
  67  
  68  def set_up_logger(is_verbose: bool = True) -> logging.Logger:
  69      """Set up a logger that writes to stderr."""
  70      log = logging.getLogger(__name__)
  71      log.setLevel(logging.INFO if is_verbose else logging.WARNING)
  72      console = logging.StreamHandler(sys.stderr)  # log to stderr
  73      console.setLevel(logging.DEBUG)
  74      formatter = logging.Formatter('[%(levelname)s] %(message)s')
  75      console.setFormatter(formatter)
  76      log.addHandler(console)
  77      return log
  78  
  79  
  80  log = set_up_logger()
  81  
  82  
  83  def indent(output: str) -> str:
  84      return textwrap.indent(output, '  ')
  85  
  86  
  87  def bool_from_env(key, default=False) -> bool:
  88      if key not in os.environ:
  89          return default
  90      raw = os.environ[key]
  91  
  92      if raw.lower() in ('1', 'true'):
  93          return True
  94      elif raw.lower() in ('0', 'false'):
  95          return False
  96      raise ValueError(f"Unrecognized environment value {key}={raw!r}")
  97  
  98  
  99  VERSION_FORMAT = "<major>.<minor>[.<patch>][-rc[0-9]][-platform]"
 100  VERSION_EXAMPLE = "22.0 or 23.1-rc1-darwin.dmg or 27.0-x86_64-linux-gnu"
 101  
 102  def parse_version_string(version_str):
 103      # "<version>[-rcN][-platform]"
 104      version_base, _, platform = version_str.partition('-')
 105      rc = ""
 106      if platform.startswith("rc"): # "<version>-rcN[-platform]"
 107          rc, _, platform = platform.partition('-')
 108      # else "<version>" or "<version>-platform"
 109  
 110      return version_base, rc, platform
 111  
 112  
 113  def download_with_wget(remote_file, local_file):
 114      result = subprocess.run(['wget', '-O', local_file, remote_file],
 115                              stderr=subprocess.STDOUT, stdout=subprocess.PIPE)
 116      return result.returncode == 0, result.stdout.decode().rstrip()
 117  
 118  
 119  def download_lines_with_urllib(url) -> tuple[bool, list[str]]:
 120      """Get (success, text lines of a file) over HTTP."""
 121      try:
 122          return (True, [
 123              line.strip().decode() for line in urllib.request.urlopen(url).readlines()])
 124      except urllib.error.HTTPError as e:
 125          log.warning(f"HTTP request to {url} failed (HTTPError): {e}")
 126      except Exception as e:
 127          log.warning(f"HTTP request to {url} failed ({e})")
 128      return (False, [])
 129  
 130  
 131  def verify_with_gpg(
 132      filename,
 133      signature_filename,
 134      output_filename: t.Optional[str] = None
 135  ) -> tuple[int, str]:
 136      with tempfile.NamedTemporaryFile() as status_file:
 137          args = [
 138              'gpg', '--yes', '--verify', '--verify-options', 'show-primary-uid-only', "--status-file", status_file.name,
 139              '--output', output_filename if output_filename else '', signature_filename, filename]
 140  
 141          env = dict(os.environ, LANGUAGE='en')
 142          result = subprocess.run(args, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, env=env)
 143  
 144          gpg_data = status_file.read().decode().rstrip()
 145  
 146      log.debug(f'Result from GPG ({result.returncode}): {result.stdout.decode()}')
 147      log.debug(f"{gpg_data}")
 148      return result.returncode, gpg_data
 149  
 150  
 151  def remove_files(filenames):
 152      for filename in filenames:
 153          os.remove(filename)
 154  
 155  
 156  class SigData:
 157      """GPG signature data as parsed from GPG stdout."""
 158      def __init__(self):
 159          self.key = None
 160          self.name = ""
 161          self.trusted = False
 162          self.status = ""
 163  
 164      def __bool__(self):
 165          return self.key is not None
 166  
 167      def __repr__(self):
 168          return (
 169              "SigData(%r, %r, trusted=%s, status=%r)" %
 170              (self.key, self.name, self.trusted, self.status))
 171  
 172  
 173  def parse_gpg_result(
 174      output: list[str]
 175  ) -> tuple[list[SigData], list[SigData], list[SigData]]:
 176      """Returns good, unknown, and bad signatures from GPG stdout."""
 177      good_sigs: list[SigData] = []
 178      unknown_sigs: list[SigData] = []
 179      bad_sigs: list[SigData] = []
 180      total_resolved_sigs = 0
 181  
 182      # Ensure that all lines we match on include a prefix that prevents malicious input
 183      # from fooling the parser.
 184      def line_begins_with(patt: str, line: str) -> t.Optional[re.Match]:
 185          return re.match(r'^(\[GNUPG:\])\s+' + patt, line)
 186  
 187      curr_sigs = unknown_sigs
 188      curr_sigdata = SigData()
 189  
 190      for line in output:
 191          if line_begins_with(r"NEWSIG(?:\s|$)", line):
 192              total_resolved_sigs += 1
 193              if curr_sigdata:
 194                  curr_sigs.append(curr_sigdata)
 195                  curr_sigdata = SigData()
 196              newsig_split = line.split()
 197              if len(newsig_split) == 3:
 198                  curr_sigdata.name = newsig_split[2]
 199  
 200          elif line_begins_with(r"GOODSIG(?:\s|$)", line):
 201              curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
 202              curr_sigs = good_sigs
 203  
 204          elif line_begins_with(r"EXPKEYSIG(?:\s|$)", line):
 205              curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
 206              curr_sigs = good_sigs
 207              curr_sigdata.status = "expired"
 208  
 209          elif line_begins_with(r"REVKEYSIG(?:\s|$)", line):
 210              curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
 211              curr_sigs = good_sigs
 212              curr_sigdata.status = "revoked"
 213  
 214          elif line_begins_with(r"BADSIG(?:\s|$)", line):
 215              curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
 216              curr_sigs = bad_sigs
 217  
 218          elif line_begins_with(r"ERRSIG(?:\s|$)", line):
 219              curr_sigdata.key, _, _, _, _, _ = line.split()[2:8]
 220              curr_sigs = unknown_sigs
 221  
 222          elif line_begins_with(r"TRUST_(UNDEFINED|NEVER)(?:\s|$)", line):
 223              curr_sigdata.trusted = False
 224  
 225          elif line_begins_with(r"TRUST_(MARGINAL|FULLY|ULTIMATE)(?:\s|$)", line):
 226              curr_sigdata.trusted = True
 227  
 228      # The last one won't have been added, so add it now
 229      assert curr_sigdata
 230      curr_sigs.append(curr_sigdata)
 231  
 232      all_found = len(good_sigs + bad_sigs + unknown_sigs)
 233      if all_found != total_resolved_sigs:
 234          raise RuntimeError(
 235              f"failed to evaluate all signatures: found {all_found} "
 236              f"but expected {total_resolved_sigs}")
 237  
 238      return (good_sigs, unknown_sigs, bad_sigs)
 239  
 240  
 241  def files_are_equal(filename1, filename2):
 242      with open(filename1, 'rb') as file1:
 243          contents1 = file1.read()
 244      with open(filename2, 'rb') as file2:
 245          contents2 = file2.read()
 246      eq = contents1 == contents2
 247  
 248      if not eq:
 249          with open(filename1, 'r', encoding='utf-8') as f1, \
 250                  open(filename2, 'r', encoding='utf-8') as f2:
 251              f1lines = f1.readlines()
 252              f2lines = f2.readlines()
 253  
 254              diff = indent(
 255                  ''.join(difflib.unified_diff(f1lines, f2lines)))
 256              log.warning(f"found diff in files ({filename1}, {filename2}):\n{diff}\n")
 257  
 258      return eq
 259  
 260  
 261  def get_files_from_hosts_and_compare(
 262      hosts: list[str], path: str, filename: str, require_all: bool = False
 263  ) -> ReturnCode:
 264      """
 265      Retrieve the same file from a number of hosts and ensure they have the same contents.
 266      The first host given will be treated as the "primary" host, and is required to succeed.
 267  
 268      Args:
 269          filename: for writing the file locally.
 270      """
 271      assert len(hosts) > 1
 272      primary_host = hosts[0]
 273      other_hosts = hosts[1:]
 274      got_files = []
 275  
 276      def join_url(host: str) -> str:
 277          return host.rstrip('/') + '/' + path.lstrip('/')
 278  
 279      url = join_url(primary_host)
 280      success, output = download_with_wget(url, filename)
 281      if not success:
 282          log.error(
 283              f"couldn't fetch file ({url}). "
 284              "Have you specified the version number in the following format?\n"
 285              f"{VERSION_FORMAT} "
 286              f"(example: {VERSION_EXAMPLE})\n"
 287              f"wget output:\n{indent(output)}")
 288          return ReturnCode.FILE_GET_FAILED
 289      else:
 290          log.info(f"got file {url} as {filename}")
 291          got_files.append(filename)
 292  
 293      for i, host in enumerate(other_hosts):
 294          url = join_url(host)
 295          fname = filename + f'.{i + 2}'
 296          success, output = download_with_wget(url, fname)
 297  
 298          if require_all and not success:
 299              log.error(
 300                  f"{host} failed to provide file ({url}), but {primary_host} did?\n"
 301                  f"wget output:\n{indent(output)}")
 302              return ReturnCode.FILE_MISSING_FROM_ONE_HOST
 303          elif not success:
 304              log.warning(
 305                  f"{host} failed to provide file ({url}). "
 306                  f"Continuing based solely upon {primary_host}.")
 307          else:
 308              log.info(f"got file {url} as {fname}")
 309              got_files.append(fname)
 310  
 311      for i, got_file in enumerate(got_files):
 312          if got_file == got_files[-1]:
 313              break  # break on last file, nothing after it to compare to
 314  
 315          compare_to = got_files[i + 1]
 316          if not files_are_equal(got_file, compare_to):
 317              log.error(f"files not equal: {got_file} and {compare_to}")
 318              return ReturnCode.FILES_NOT_EQUAL
 319  
 320      return ReturnCode.SUCCESS
 321  
 322  
 323  def check_multisig(sums_file: str, sigfilename: str, args: argparse.Namespace) -> tuple[int, str, list[SigData], list[SigData], list[SigData]]:
 324      # check signature
 325      #
 326      # We don't write output to a file because this command will almost certainly
 327      # fail with GPG exit code '2' (and so not writing to --output) because of the
 328      # likely presence of multiple untrusted signatures.
 329      retval, output = verify_with_gpg(sums_file, sigfilename)
 330  
 331      if args.verbose:
 332          log.info(f"gpg output:\n{indent(output)}")
 333  
 334      good, unknown, bad = parse_gpg_result(output.splitlines())
 335  
 336      if unknown and args.import_keys:
 337          # Retrieve unknown keys and then try GPG again.
 338          for unsig in unknown:
 339              if prompt_yn(f" ? Retrieve key {unsig.key} ({unsig.name})? (y/N) "):
 340                  ran = subprocess.run(
 341                      ["gpg", "--keyserver", args.keyserver, "--recv-keys", unsig.key])
 342  
 343                  if ran.returncode != 0:
 344                      log.warning(f"failed to retrieve key {unsig.key}")
 345  
 346          # Reparse the GPG output now that we have more keys
 347          retval, output = verify_with_gpg(sums_file, sigfilename)
 348          good, unknown, bad = parse_gpg_result(output.splitlines())
 349  
 350      return retval, output, good, unknown, bad
 351  
 352  
 353  def prompt_yn(prompt) -> bool:
 354      """Return true if the user inputs 'y'."""
 355      got = ''
 356      while got not in ['y', 'n']:
 357          got = input(prompt).lower()
 358      return got == 'y'
 359  
 360  def verify_shasums_signature(
 361      signature_file_path: str, sums_file_path: str, args: argparse.Namespace
 362  ) -> tuple[
 363     ReturnCode, list[SigData], list[SigData], list[SigData], list[SigData]
 364  ]:
 365      min_good_sigs = args.min_good_sigs
 366      gpg_allowed_codes = [0, 2]  # 2 is returned when untrusted signatures are present.
 367  
 368      gpg_retval, gpg_output, good, unknown, bad = check_multisig(sums_file_path, signature_file_path, args)
 369  
 370      if gpg_retval not in gpg_allowed_codes:
 371          if gpg_retval == 1:
 372              log.critical(f"Bad signature (code: {gpg_retval}).")
 373          else:
 374              log.critical(f"unexpected GPG exit code ({gpg_retval})")
 375  
 376          log.error(f"gpg output:\n{indent(gpg_output)}")
 377          return (ReturnCode.INTEGRITY_FAILURE, [], [], [], [])
 378  
 379      # Decide which keys we trust, though not "trust" in the GPG sense, but rather
 380      # which pubkeys convince us that this sums file is legitimate. In other words,
 381      # which pubkeys within the Limenka community do we trust for the purposes of
 382      # binary verification?
 383      trusted_keys = set()
 384      if args.trusted_keys:
 385          trusted_keys |= set(args.trusted_keys.split(','))
 386  
 387      # Tally signatures and make sure we have enough goods to fulfill
 388      # our threshold.
 389      good_trusted = [sig for sig in good if sig.trusted or sig.key in trusted_keys]
 390      good_untrusted = [sig for sig in good if sig not in good_trusted]
 391      num_trusted = len(good_trusted) + len(good_untrusted)
 392      log.info(f"got {num_trusted} good signatures")
 393  
 394      if num_trusted < min_good_sigs:
 395          log.info("Maybe you need to import "
 396                    f"(`gpg --keyserver {args.keyserver} --recv-keys <key-id>`) "
 397                    "some of the following keys: ")
 398          log.info('')
 399          for sig in unknown:
 400              log.info(f"    {sig.key} ({sig.name})")
 401          log.info('')
 402          log.error(
 403              "not enough trusted sigs to meet threshold "
 404              f"({num_trusted} vs. {min_good_sigs})")
 405  
 406          return (ReturnCode.NOT_ENOUGH_GOOD_SIGS, [], [], [], [])
 407  
 408      for sig in good_trusted:
 409          log.info(f"GOOD SIGNATURE: {sig}")
 410  
 411      for sig in good_untrusted:
 412          log.info(f"GOOD SIGNATURE (untrusted): {sig}")
 413  
 414      for sig in [sig for sig in good if sig.status == 'expired']:
 415          log.warning(f"key {sig.key} for {sig.name} is expired")
 416  
 417      for sig in bad:
 418          log.warning(f"BAD SIGNATURE: {sig}")
 419  
 420      for sig in unknown:
 421          log.warning(f"UNKNOWN SIGNATURE: {sig}")
 422  
 423      return (ReturnCode.SUCCESS, good_trusted, good_untrusted, unknown, bad)
 424  
 425  
 426  def parse_sums_file(sums_file_path: str, filename_filter: list[str]) -> list[list[str]]:
 427      # extract hashes/filenames of binaries to verify from hash file;
 428      # each line has the following format: "<hash> <binary_filename>"
 429      with open(sums_file_path, 'r', encoding='utf8') as hash_file:
 430          return [line.split()[:2] for line in hash_file if len(filename_filter) == 0 or any(f in line for f in filename_filter)]
 431  
 432  
 433  def verify_binary_hashes(hashes_to_verify: list[list[str]]) -> tuple[ReturnCode, dict[str, str]]:
 434      offending_files = []
 435      files_to_hashes = {}
 436  
 437      for hash_expected, binary_filename in hashes_to_verify:
 438          with open(binary_filename, 'rb') as binary_file:
 439              hash_calculated = sha256(binary_file.read()).hexdigest()
 440          if hash_calculated != hash_expected:
 441              offending_files.append(binary_filename)
 442          else:
 443              files_to_hashes[binary_filename] = hash_calculated
 444  
 445      if offending_files:
 446          joined_files = '\n'.join(offending_files)
 447          log.critical(
 448              "Hashes don't match.\n"
 449              f"Offending files:\n{joined_files}")
 450          return (ReturnCode.INTEGRITY_FAILURE, files_to_hashes)
 451  
 452      return (ReturnCode.SUCCESS, files_to_hashes)
 453  
 454  
 455  def verify_published_handler(args: argparse.Namespace) -> ReturnCode:
 456      WORKINGDIR = Path(tempfile.gettempdir()) / f"limenka_verify_binaries.{args.version}"
 457  
 458      def cleanup():
 459          log.info("cleaning up files")
 460          os.chdir(Path.home())
 461          shutil.rmtree(WORKINGDIR)
 462  
 463      # determine remote dir dependent on provided version string
 464      try:
 465          version_base, version_rc, os_filter = parse_version_string(args.version)
 466          version_tuple = [int(i) for i in version_base.split('.')]
 467      except Exception as e:
 468          log.debug(e)
 469          log.error(f"unable to parse version; expected format is {VERSION_FORMAT}")
 470          log.error(f"  e.g. {VERSION_EXAMPLE}")
 471          return ReturnCode.BAD_VERSION
 472  
 473      remote_dir = f"/bin/{VERSIONPREFIX}{version_base}/"
 474      if version_rc:
 475          remote_dir += f"test.{version_rc}/"
 476      remote_sigs_path = remote_dir + SIGNATUREFILENAME
 477      remote_sums_path = remote_dir + SUMS_FILENAME
 478  
 479      # create working directory
 480      os.makedirs(WORKINGDIR, exist_ok=True)
 481      os.chdir(WORKINGDIR)
 482  
 483      hosts = [HOST1, HOST2]
 484  
 485      got_sig_status = get_files_from_hosts_and_compare(
 486          hosts, remote_sigs_path, SIGNATUREFILENAME, args.require_all_hosts)
 487      if got_sig_status != ReturnCode.SUCCESS:
 488          return got_sig_status
 489  
 490      # Multi-sig verification is available after 22.0.
 491      if version_tuple[0] < 22:
 492          log.error("Version too old - single sig not supported. Use a previous "
 493                    "version of this script from the repo.")
 494          return ReturnCode.BAD_VERSION
 495  
 496      got_sums_status = get_files_from_hosts_and_compare(
 497          hosts, remote_sums_path, SUMS_FILENAME, args.require_all_hosts)
 498      if got_sums_status != ReturnCode.SUCCESS:
 499          return got_sums_status
 500  
 501      # Verify the signature on the SHA256SUMS file
 502      sigs_status, good_trusted, good_untrusted, unknown, bad = verify_shasums_signature(SIGNATUREFILENAME, SUMS_FILENAME, args)
 503      if sigs_status != ReturnCode.SUCCESS:
 504          if sigs_status == ReturnCode.INTEGRITY_FAILURE:
 505              cleanup()
 506          return sigs_status
 507  
 508      # Extract hashes and filenames
 509      hashes_to_verify = parse_sums_file(SUMS_FILENAME, [os_filter])
 510      if not hashes_to_verify:
 511          available_versions = ["-".join(line[1].split("-")[2:]) for line in parse_sums_file(SUMS_FILENAME, [])]
 512          closest_match = difflib.get_close_matches(os_filter, available_versions, cutoff=0, n=1)[0]
 513          log.error(f"No files matched the platform specified. Did you mean: {closest_match}")
 514          return ReturnCode.NO_BINARIES_MATCH
 515  
 516      # remove binaries that are known not to be hosted by limenkacore.org
 517      fragments_to_remove = ['-unsigned', '-debug', '-codesignatures']
 518      for fragment in fragments_to_remove:
 519          nobinaries = [i for i in hashes_to_verify if fragment in i[1]]
 520          if nobinaries:
 521              remove_str = ', '.join(i[1] for i in nobinaries)
 522              log.info(
 523                  f"removing *{fragment} binaries ({remove_str}) from verification "
 524                  f"since {HOST1} does not host *{fragment} binaries")
 525              hashes_to_verify = [i for i in hashes_to_verify if fragment not in i[1]]
 526  
 527      # download binaries
 528      for _, binary_filename in hashes_to_verify:
 529          log.info(f"downloading {binary_filename} to {WORKINGDIR}")
 530          success, output = download_with_wget(
 531              HOST1 + remote_dir + binary_filename, binary_filename)
 532  
 533          if not success:
 534              log.error(
 535                  f"failed to download {binary_filename}\n"
 536                  f"wget output:\n{indent(output)}")
 537              return ReturnCode.BINARY_DOWNLOAD_FAILED
 538  
 539      # verify hashes
 540      hashes_status, files_to_hashes = verify_binary_hashes(hashes_to_verify)
 541      if hashes_status != ReturnCode.SUCCESS:
 542          return hashes_status
 543  
 544  
 545      if args.cleanup:
 546          cleanup()
 547      else:
 548          log.info(f"did not clean up {WORKINGDIR}")
 549  
 550      if args.json:
 551          output = {
 552              'good_trusted_sigs': [str(s) for s in good_trusted],
 553              'good_untrusted_sigs': [str(s) for s in good_untrusted],
 554              'unknown_sigs': [str(s) for s in unknown],
 555              'bad_sigs': [str(s) for s in bad],
 556              'verified_binaries': files_to_hashes,
 557          }
 558          print(json.dumps(output, indent=2))
 559      else:
 560          for filename in files_to_hashes:
 561              print(f"VERIFIED: {filename}")
 562  
 563      return ReturnCode.SUCCESS
 564  
 565  
 566  def verify_binaries_handler(args: argparse.Namespace) -> ReturnCode:
 567      binary_to_basename = {}
 568      for file in args.binary:
 569          binary_to_basename[PurePath(file).name] = file
 570  
 571      sums_sig_path = None
 572      if args.sums_sig_file:
 573          sums_sig_path = Path(args.sums_sig_file)
 574      else:
 575          log.info(f"No signature file specified, assuming it is {args.sums_file}.asc")
 576          sums_sig_path = Path(args.sums_file).with_suffix(".asc")
 577  
 578      # Verify the signature on the SHA256SUMS file
 579      sigs_status, good_trusted, good_untrusted, unknown, bad = verify_shasums_signature(str(sums_sig_path), args.sums_file, args)
 580      if sigs_status != ReturnCode.SUCCESS:
 581          return sigs_status
 582  
 583      # Extract hashes and filenames
 584      hashes_to_verify = parse_sums_file(args.sums_file, [k for k, n in binary_to_basename.items()])
 585      if not hashes_to_verify:
 586          log.error(f"No files in {args.sums_file} match the specified binaries")
 587          return ReturnCode.NO_BINARIES_MATCH
 588  
 589      # Make sure all files are accounted for
 590      sums_file_path = Path(args.sums_file)
 591      missing_files = []
 592      files_to_hash = []
 593      if len(binary_to_basename) > 0:
 594          for file_hash, file in hashes_to_verify:
 595              files_to_hash.append([file_hash, binary_to_basename[file]])
 596              del binary_to_basename[file]
 597          if len(binary_to_basename) > 0:
 598              log.error(f"Not all specified binaries are in {args.sums_file}")
 599              return ReturnCode.NO_BINARIES_MATCH
 600      else:
 601          log.info(f"No binaries specified, assuming all files specified in {args.sums_file} are located relatively")
 602          for file_hash, file in hashes_to_verify:
 603              file_path = Path(sums_file_path.parent.joinpath(file))
 604              if file_path.exists():
 605                  files_to_hash.append([file_hash, str(file_path)])
 606              else:
 607                  missing_files.append(file)
 608  
 609      # verify hashes
 610      hashes_status, files_to_hashes = verify_binary_hashes(files_to_hash)
 611      if hashes_status != ReturnCode.SUCCESS:
 612          return hashes_status
 613  
 614      if args.json:
 615          output = {
 616              'good_trusted_sigs': [str(s) for s in good_trusted],
 617              'good_untrusted_sigs': [str(s) for s in good_untrusted],
 618              'unknown_sigs': [str(s) for s in unknown],
 619              'bad_sigs': [str(s) for s in bad],
 620              'verified_binaries': files_to_hashes,
 621              "missing_binaries": missing_files,
 622          }
 623          print(json.dumps(output, indent=2))
 624      else:
 625          for filename in files_to_hashes:
 626              print(f"VERIFIED: {filename}")
 627          for filename in missing_files:
 628              print(f"MISSING: {filename}")
 629  
 630      return ReturnCode.SUCCESS
 631  
 632  
 633  def main():
 634      parser = argparse.ArgumentParser(description=__doc__)
 635      parser.add_argument(
 636          '-v', '--verbose', action='store_true',
 637          default=bool_from_env('BINVERIFY_VERBOSE'),
 638      )
 639      parser.add_argument(
 640          '-q', '--quiet', action='store_true',
 641          default=bool_from_env('BINVERIFY_QUIET'),
 642      )
 643      parser.add_argument(
 644          '--import-keys', action='store_true',
 645          default=bool_from_env('BINVERIFY_IMPORTKEYS'),
 646          help='if specified, ask to import each unknown builder key'
 647      )
 648      parser.add_argument(
 649          '--min-good-sigs', type=int, action='store', nargs='?',
 650          default=int(os.environ.get('BINVERIFY_MIN_GOOD_SIGS', 3)),
 651          help=(
 652              'The minimum number of good signatures to require successful termination.'),
 653      )
 654      parser.add_argument(
 655          '--keyserver', action='store', nargs='?',
 656          default=os.environ.get('BINVERIFY_KEYSERVER', 'hkps://keys.openpgp.org'),
 657          help='which keyserver to use',
 658      )
 659      parser.add_argument(
 660          '--trusted-keys', action='store', nargs='?',
 661          default=os.environ.get('BINVERIFY_TRUSTED_KEYS', ''),
 662          help='A list of trusted signer GPG keys, separated by commas. Not "trusted keys" in the GPG sense.',
 663      )
 664      parser.add_argument(
 665          '--json', action='store_true',
 666          default=bool_from_env('BINVERIFY_JSON'),
 667          help='If set, output the result as JSON',
 668      )
 669  
 670      subparsers = parser.add_subparsers(title="Commands", required=True, dest="command")
 671  
 672      pub_parser = subparsers.add_parser("pub", help="Verify a published release.")
 673      pub_parser.set_defaults(func=verify_published_handler)
 674      pub_parser.add_argument(
 675          'version', type=str, help=(
 676              f'version of the limenka release to download; of the format '
 677              f'{VERSION_FORMAT}. Example: {VERSION_EXAMPLE}')
 678      )
 679      pub_parser.add_argument(
 680          '--cleanup', action='store_true',
 681          default=bool_from_env('BINVERIFY_CLEANUP'),
 682          help='if specified, clean up files afterwards'
 683      )
 684      pub_parser.add_argument(
 685          '--require-all-hosts', action='store_true',
 686          default=bool_from_env('BINVERIFY_REQUIRE_ALL_HOSTS'),
 687          help=(
 688              f'If set, require all hosts ({HOST1}, {HOST2}) to provide signatures. '
 689              '(Sometimes limenka.org lags behind limenkacore.org.)')
 690      )
 691  
 692      bin_parser = subparsers.add_parser("bin", help="Verify local binaries.")
 693      bin_parser.set_defaults(func=verify_binaries_handler)
 694      bin_parser.add_argument("--sums-sig-file", "-s", help="Path to the SHA256SUMS.asc file to verify")
 695      bin_parser.add_argument("sums_file", help="Path to the SHA256SUMS file to verify")
 696      bin_parser.add_argument(
 697          "binary", nargs="*",
 698          help="Path to a binary distribution file to verify. Can be specified multiple times for multiple files to verify."
 699      )
 700  
 701      args = parser.parse_args()
 702      if args.quiet:
 703          log.setLevel(logging.WARNING)
 704  
 705      return args.func(args)
 706  
 707  
 708  if __name__ == '__main__':
 709      sys.exit(main())
 710