Inside Bitcoin

1#!/usr/bin/env python3
2# Copyright (c) 2020-present The Bitcoin Core 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 Bitcoin Core release binaries.
6
7This script attempts to download the sum file SHA256SUMS and corresponding
8signature file SHA256SUMS.asc from bitcoincore.org and bitcoin.org and
9compares them.
10
11The sum-signature file is signed by a number of builder keys. This script
12ensures that there is a minimum threshold of signatures from pubkeys that
13we trust. This trust is articulated on the basis of configuration options
14here, but by default is based upon local GPG trust settings.
15
16The builder keys are available in the guix.sigs repo:
17
18    https://github.com/bitcoin-core/guix.sigs/tree/main/builder-keys
19
20If a minimum good, trusted signature threshold is met on the sum file, we then
21download the files specified in SHA256SUMS, and check if the hashes of these
22files match those that are specified. The script returns 0 if everything passes
23the checks. It returns 1 if either the signature check or the hash check
24doesn't pass. If an error occurs the return value is >= 2.
25
26Logging output goes to stderr and final binary verification data goes to stdout.
27
28JSON output can by obtained by setting env BINVERIFY_JSON=1.
29"""
30import argparse
31import difflib
32import json
33import logging
34import os
35import subprocess
36import typing as t
37import re
38import sys
39import shutil
40import tempfile
41import textwrap
42import enum
43from hashlib import sha256
44from pathlib import PurePath, Path
45
46# The primary host; this will fail if we can't retrieve files from here.
47HOST1 = "https://bitcoincore.org"
48HOST2 = "https://bitcoin.org"
49VERSIONPREFIX = "bitcoin-core-"
50SUMS_FILENAME = 'SHA256SUMS'
51SIGNATUREFILENAME = f"{SUMS_FILENAME}.asc"
52
53
54class ReturnCode(enum.IntEnum):
55    SUCCESS = 0
56    INTEGRITY_FAILURE = 1
57    FILE_GET_FAILED = 4
58    FILE_MISSING_FROM_ONE_HOST = 5
59    FILES_NOT_EQUAL = 6
60    NO_BINARIES_MATCH = 7
61    NOT_ENOUGH_GOOD_SIGS = 9
62    BINARY_DOWNLOAD_FAILED = 10
63    BAD_VERSION = 11
64
65
66def set_up_logger(is_verbose: bool = True) -> logging.Logger:
67    """Set up a logger that writes to stderr."""
68    log = logging.getLogger(__name__)
69    log.setLevel(logging.INFO if is_verbose else logging.WARNING)
70    console = logging.StreamHandler(sys.stderr)  # log to stderr
71    console.setLevel(logging.DEBUG)
72    formatter = logging.Formatter('[%(levelname)s] %(message)s')
73    console.setFormatter(formatter)
74    log.addHandler(console)
75    return log
76
77
78log = set_up_logger()
79
80
81def indent(output: str) -> str:
82    return textwrap.indent(output, '  ')
83
84
85def bool_from_env(key, default=False) -> bool:
86    if key not in os.environ:
87        return default
88    raw = os.environ[key]
89
90    if raw.lower() in ('1', 'true'):
91        return True
92    elif raw.lower() in ('0', 'false'):
93        return False
94    raise ValueError(f"Unrecognized environment value {key}={raw!r}")
95
96
97VERSION_FORMAT = "<major>.<minor>[.<patch>][-rc[0-9]][-platform]"
98VERSION_EXAMPLE = "22.0 or 23.1-rc1-darwin.dmg or 27.0-x86_64-linux-gnu"
99
100def parse_version_string(version_str):
101    # "<version>[-rcN][-platform]"
102    version_base, _, platform = version_str.partition('-')
103    rc = ""
104    if platform.startswith("rc"): # "<version>-rcN[-platform]"
105        rc, _, platform = platform.partition('-')
106    # else "<version>" or "<version>-platform"
107
108    return version_base, rc, platform
109
110
111def download_with_wget(remote_file, local_file):
112    result = subprocess.run(['wget', '-O', local_file, remote_file],
113                            stderr=subprocess.STDOUT, stdout=subprocess.PIPE)
114    return result.returncode == 0, result.stdout.decode().rstrip()
115
116
117def verify_with_gpg(
118    filename,
119    signature_filename,
120    output_filename: t.Optional[str] = None
121) -> tuple[int, str]:
122    with tempfile.NamedTemporaryFile() as status_file:
123        args = [
124            'gpg', '--yes', '--verify', '--verify-options', 'show-primary-uid-only', "--status-file", status_file.name,
125            '--output', output_filename if output_filename else '', signature_filename, filename]
126
127        env = dict(os.environ, LANGUAGE='en')
128        result = subprocess.run(args, stderr=subprocess.STDOUT, stdout=subprocess.PIPE, env=env)
129
130        gpg_data = status_file.read().decode().rstrip()
131
132    log.debug(f'Result from GPG ({result.returncode}): {result.stdout.decode()}')
133    log.debug(f"{gpg_data}")
134    return result.returncode, gpg_data
135
136
137class SigData:
138    """GPG signature data as parsed from GPG stdout."""
139    def __init__(self):
140        self.key = None
141        self.name = ""
142        self.trusted = False
143        self.status = ""
144
145    def __bool__(self):
146        return self.key is not None
147
148    def __repr__(self):
149        return (
150            "SigData(%r, %r, trusted=%s, status=%r)" %
151            (self.key, self.name, self.trusted, self.status))
152
153
154def parse_gpg_result(
155    output: list[str]
156) -> tuple[list[SigData], list[SigData], list[SigData]]:
157    """Returns good, unknown, and bad signatures from GPG stdout."""
158    good_sigs: list[SigData] = []
159    unknown_sigs: list[SigData] = []
160    bad_sigs: list[SigData] = []
161    total_resolved_sigs = 0
162
163    # Ensure that all lines we match on include a prefix that prevents malicious input
164    # from fooling the parser.
165    def line_begins_with(patt: str, line: str) -> t.Optional[re.Match]:
166        return re.match(r'^(\[GNUPG:\])\s+' + patt, line)
167
168    curr_sigs = unknown_sigs
169    curr_sigdata = SigData()
170
171    for line in output:
172        if line_begins_with(r"NEWSIG(?:\s|$)", line):
173            total_resolved_sigs += 1
174            if curr_sigdata:
175                curr_sigs.append(curr_sigdata)
176                curr_sigdata = SigData()
177            newsig_split = line.split()
178            if len(newsig_split) == 3:
179                curr_sigdata.name = newsig_split[2]
180
181        elif line_begins_with(r"GOODSIG(?:\s|$)", line):
182            curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
183            curr_sigs = good_sigs
184
185        elif line_begins_with(r"EXPKEYSIG(?:\s|$)", line):
186            curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
187            curr_sigs = good_sigs
188            curr_sigdata.status = "expired"
189
190        elif line_begins_with(r"REVKEYSIG(?:\s|$)", line):
191            curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
192            curr_sigs = good_sigs
193            curr_sigdata.status = "revoked"
194
195        elif line_begins_with(r"BADSIG(?:\s|$)", line):
196            curr_sigdata.key, curr_sigdata.name = line.split(maxsplit=3)[2:4]
197            curr_sigs = bad_sigs
198
199        elif line_begins_with(r"ERRSIG(?:\s|$)", line):
200            curr_sigdata.key, _, _, _, _, _ = line.split()[2:8]
201            curr_sigs = unknown_sigs
202
203        elif line_begins_with(r"TRUST_(UNDEFINED|NEVER)(?:\s|$)", line):
204            curr_sigdata.trusted = False
205
206        elif line_begins_with(r"TRUST_(MARGINAL|FULLY|ULTIMATE)(?:\s|$)", line):
207            curr_sigdata.trusted = True
208
209    # The last one won't have been added, so add it now
210    assert curr_sigdata
211    curr_sigs.append(curr_sigdata)
212
213    all_found = len(good_sigs + bad_sigs + unknown_sigs)
214    if all_found != total_resolved_sigs:
215        raise RuntimeError(
216            f"failed to evaluate all signatures: found {all_found} "
217            f"but expected {total_resolved_sigs}")
218
219    return (good_sigs, unknown_sigs, bad_sigs)
220
221
222def files_are_equal(filename1, filename2):
223    with open(filename1, 'rb') as file1:
224        contents1 = file1.read()
225    with open(filename2, 'rb') as file2:
226        contents2 = file2.read()
227    eq = contents1 == contents2
228
229    if not eq:
230        with open(filename1, 'r') as f1, \
231                open(filename2, 'r') as f2:
232            f1lines = f1.readlines()
233            f2lines = f2.readlines()
234
235            diff = indent(
236                ''.join(difflib.unified_diff(f1lines, f2lines)))
237            log.warning(f"found diff in files ({filename1}, {filename2}):\n{diff}\n")
238
239    return eq
240
241
242def get_files_from_hosts_and_compare(
243    hosts: list[str], path: str, filename: str, require_all: bool = False
244) -> ReturnCode:
245    """
246    Retrieve the same file from a number of hosts and ensure they have the same contents.
247    The first host given will be treated as the "primary" host, and is required to succeed.
248
249    Args:
250        filename: for writing the file locally.