mirror of
https://github.com/trezor/trezor-firmware.git
synced 2025-01-10 23:40:58 +00:00
trezorctl: updated firmware update flow
We can now locally verify firmware signatures and hashes. We also recognize min_firmware_version, so this resolves #308 This also helps with #273, as trezorlib is now mostly usable for signing firmware images.
This commit is contained in:
matejcik
parent
3e7b26b454
commit
e1efd493fd
224
trezorctl
224
trezorctl
@ -21,13 +21,12 @@
|
||||
# along with this library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
|
||||
import click
|
||||
import requests
|
||||
|
||||
from trezorlib import (
|
||||
btc,
|
||||
@ -37,6 +36,7 @@ from trezorlib import (
|
||||
debuglink,
|
||||
device,
|
||||
ethereum,
|
||||
exceptions,
|
||||
firmware,
|
||||
lisk,
|
||||
log,
|
||||
@ -499,14 +499,144 @@ def backup_device(connect):
|
||||
#
|
||||
|
||||
|
||||
@cli.command(help="Upload new firmware to device (must be in bootloader mode).")
|
||||
def validate_firmware_v1(fw, expected_fingerprint=None):
|
||||
click.echo("Trezor One firmware image.")
|
||||
distinct_sig_slots = set(i for i in fw.key_indexes if i != 0)
|
||||
if not distinct_sig_slots:
|
||||
if not click.confirm("No signatures found. Continue?", default=False):
|
||||
sys.exit(1)
|
||||
elif len(distinct_sig_slots) < 3:
|
||||
click.echo("Badly signed image (need 3 distinct signatures), aborting.")
|
||||
sys.exit(1)
|
||||
else:
|
||||
all_valid = True
|
||||
for i in range(len(fw.key_indexes)):
|
||||
if not firmware.check_sig_v1(fw, i):
|
||||
click.echo("INVALID signature in slot {}".format(i))
|
||||
all_valid = False
|
||||
|
||||
if all_valid:
|
||||
click.echo("Signatures are valid.")
|
||||
else:
|
||||
click.echo("Invalid signature detected, aborting.")
|
||||
sys.exit(4)
|
||||
|
||||
fingerprint = firmware.digest_v1(fw).hex()
|
||||
click.echo("Firmware fingerprint: {}".format(fingerprint))
|
||||
if expected_fingerprint and fingerprint != expected_fingerprint:
|
||||
click.echo("Expected fingerprint: {}".format(expected_fingerprint))
|
||||
click.echo("Fingerprints do not match, aborting.")
|
||||
sys.exit(5)
|
||||
|
||||
|
||||
def validate_firmware_v2(fw, expected_fingerprint=None, skip_vendor_header=False):
|
||||
click.echo("Trezor T firmware image.")
|
||||
vendor = fw.vendor_header.vendor_string
|
||||
vendor_version = "{major}.{minor}".format(**fw.vendor_header.version)
|
||||
version = fw.firmware_header.version
|
||||
click.echo("Vendor header from {}, version {}".format(vendor, vendor_version))
|
||||
click.echo(
|
||||
"Firmware version {major}.{minor}.{patch} build {build}".format(**version)
|
||||
)
|
||||
try:
|
||||
firmware.validate(fw, skip_vendor_header)
|
||||
click.echo("Signatures are valid.")
|
||||
except Exception as e:
|
||||
click.echo(e)
|
||||
sys.exit(4)
|
||||
|
||||
fingerprint = firmware.digest(fw).hex()
|
||||
click.echo("Firmware fingerprint: {}".format(fingerprint))
|
||||
if expected_fingerprint and fingerprint != expected_fingerprint:
|
||||
click.echo("Expected fingerprint: {}".format(expected_fingerprint))
|
||||
click.echo("Fingerprints do not match, aborting.")
|
||||
sys.exit(5)
|
||||
|
||||
|
||||
def find_best_firmware_version(bootloader_version, requested_version=None):
|
||||
url = "https://wallet.trezor.io/data/firmware/{}/releases.json"
|
||||
releases = requests.get(url.format(bootloader_version[0])).json()
|
||||
if not releases:
|
||||
raise click.ClickException("Failed to get list of releases")
|
||||
|
||||
releases.sort(key=lambda r: r["version"], reverse=True)
|
||||
|
||||
def version_str(version):
|
||||
return ".".join(map(str, version))
|
||||
|
||||
want_version = requested_version
|
||||
|
||||
if want_version is None:
|
||||
want_version = releases[0]["version"]
|
||||
click.echo("Best available version: {}".format(version_str(want_version)))
|
||||
|
||||
confirm_different_version = False
|
||||
while True:
|
||||
want_version_str = version_str(want_version)
|
||||
try:
|
||||
release = next(r for r in releases if r["version"] == want_version)
|
||||
except StopIteration:
|
||||
click.echo("Version {} not found.".format(want_version_str))
|
||||
sys.exit(1)
|
||||
|
||||
if (
|
||||
"min_bootloader_version" in release
|
||||
and release["min_bootloader_version"] > bootloader_version
|
||||
):
|
||||
need_version_str = version_str(release["min_firmware_version"])
|
||||
click.echo(
|
||||
"Version {} is required before upgrading to {}.".format(
|
||||
need_version_str, want_version_str
|
||||
)
|
||||
)
|
||||
want_version = release["min_firmware_version"]
|
||||
confirm_different_version = True
|
||||
else:
|
||||
break
|
||||
|
||||
if confirm_different_version:
|
||||
installing_different = "Installing version {} instead.".format(want_version_str)
|
||||
if requested_version is None:
|
||||
click.echo(installing_different)
|
||||
else:
|
||||
ok = click.confirm(installing_different + " Continue?", default=True)
|
||||
if not ok:
|
||||
sys.exit(1)
|
||||
|
||||
url = "https://wallet.trezor.io/" + release["url"]
|
||||
if url.endswith(".hex"):
|
||||
url = url[:-4]
|
||||
|
||||
return url, release["fingerprint"]
|
||||
|
||||
|
||||
@cli.command()
|
||||
@click.option("-f", "--filename")
|
||||
@click.option("-u", "--url")
|
||||
@click.option("-v", "--version")
|
||||
@click.option("-s", "--skip-check", is_flag=True)
|
||||
@click.option("--fingerprint", help="Expected firmware fingerprint in hex")
|
||||
@click.option("--skip-vendor-header", help="Skip vendor header validation on Trezor T")
|
||||
@click.pass_obj
|
||||
def firmware_update(connect, filename, url, version, skip_check, fingerprint):
|
||||
def firmware_update(
|
||||
connect, filename, url, version, skip_check, fingerprint, skip_vendor_header
|
||||
):
|
||||
"""Upload new firmware to device.
|
||||
|
||||
Device must be in bootloader mode.
|
||||
|
||||
You can specify a filename or URL from which the firmware can be downloaded.
|
||||
You can also explicitly specify a firmware version that you want.
|
||||
Otherwise, trezorctl will attempt to find latest available version
|
||||
from wallet.trezor.io.
|
||||
|
||||
If you provide a fingerprint via the --fingerprint option, it will be checked
|
||||
against downloaded firmware fingerprint. Otherwise fingerprint is checked
|
||||
against wallet.trezor.io information, if available.
|
||||
|
||||
If you are customizing Model T bootloader and providing your own vendor header,
|
||||
you can use --skip-vendor-header to ignore vendor header signatures.
|
||||
"""
|
||||
if sum(bool(x) for x in (filename, url, version)) > 1:
|
||||
click.echo("You can use only one of: filename, url, version.")
|
||||
sys.exit(1)
|
||||
@ -519,71 +649,47 @@ def firmware_update(connect, filename, url, version, skip_check, fingerprint):
|
||||
firmware_version = client.features.major_version
|
||||
|
||||
if filename:
|
||||
fp = open(filename, "rb").read()
|
||||
elif url:
|
||||
import requests
|
||||
data = open(filename, "rb").read()
|
||||
else:
|
||||
if not url:
|
||||
f = client.features
|
||||
bootloader_version = [f.major_version, f.minor_version, f.patch_version]
|
||||
version_list = [int(x) for x in version.split(".")] if version else None
|
||||
url, fp = find_best_firmware_version(bootloader_version, version_list)
|
||||
if not fingerprint:
|
||||
fingerprint = fp
|
||||
|
||||
click.echo("Downloading from {}".format(url))
|
||||
r = requests.get(url)
|
||||
fp = r.content
|
||||
else:
|
||||
import requests
|
||||
|
||||
url = "https://wallet.trezor.io/data/firmware/{}/releases.json"
|
||||
releases = requests.get(url.format(firmware_version)).json()
|
||||
|
||||
def version_func(r):
|
||||
return r["version"]
|
||||
|
||||
def version_string(r):
|
||||
return ".".join(map(str, version_func(r)))
|
||||
|
||||
if version:
|
||||
try:
|
||||
release = next(r for r in releases if version_string(r) == version)
|
||||
except StopIteration:
|
||||
click.echo("Version {} not found.".format(version))
|
||||
sys.exit(1)
|
||||
else:
|
||||
release = max(releases, key=version_func)
|
||||
click.echo("Fetching version: %s" % version_string(release))
|
||||
|
||||
if not fingerprint:
|
||||
fingerprint = release["fingerprint"]
|
||||
url = "https://wallet.trezor.io/" + release["url"]
|
||||
click.echo("Downloading from %s" % url)
|
||||
r = requests.get(url)
|
||||
fp = r.content
|
||||
data = r.content
|
||||
|
||||
if not skip_check:
|
||||
if fp[:8] == b"54525a52" or fp[:8] == b"54525a56":
|
||||
fp = bytes.fromhex(fp.decode())
|
||||
if fp[:4] != b"TRZR" and fp[:4] != b"TRZV":
|
||||
click.echo("Trezor firmware header expected.")
|
||||
try:
|
||||
version, fw = firmware.parse(data)
|
||||
except Exception as e:
|
||||
click.echo(e)
|
||||
sys.exit(2)
|
||||
|
||||
if fingerprint and firmware_version > 1:
|
||||
click.echo("Checking Trezor T fingerprint is not supported yet.")
|
||||
elif firmware_version == 1:
|
||||
calculated_fingerprint = hashlib.sha256(fp[256:]).hexdigest()
|
||||
click.echo("Firmware fingerprint: {}".format(calculated_fingerprint))
|
||||
if fingerprint and fingerprint != calculated_fingerprint:
|
||||
click.echo("Expected fingerprint: {}".format(fingerprint))
|
||||
click.echo("Fingerprints do not match, aborting.")
|
||||
sys.exit(5)
|
||||
if version == firmware.FirmwareFormat.TREZOR_ONE:
|
||||
validate_firmware_v1(fw, fingerprint)
|
||||
elif version == firmware.FirmwareFormat.TREZOR_T:
|
||||
validate_firmware_v2(fw, fingerprint)
|
||||
else:
|
||||
click.echo("Unrecognized firmware version.")
|
||||
|
||||
click.echo("If asked, please confirm the action on your device ...")
|
||||
if firmware_version != version.value:
|
||||
click.echo("Firmware does not match your device, aborting.")
|
||||
sys.exit(3)
|
||||
|
||||
try:
|
||||
return firmware.update(client, fp=io.BytesIO(fp))
|
||||
except tools.CallException as e:
|
||||
if e.args[0] in (
|
||||
proto.FailureType.FirmwareError,
|
||||
proto.FailureType.ActionCancelled,
|
||||
):
|
||||
if firmware_version == 1:
|
||||
# Trezor One does not send ButtonRequest
|
||||
click.echo("Please confirm action on your Trezor device")
|
||||
return firmware.update(client, data)
|
||||
except exceptions.Cancelled:
|
||||
click.echo("Update aborted on device.")
|
||||
else:
|
||||
click.echo("Update failed: {} {}".format(*e.args))
|
||||
except exceptions.TrezorException as e:
|
||||
click.echo("Update failed: {}".format(e))
|
||||
sys.exit(3)
|
||||
|
||||
|
||||
@ -1191,8 +1297,6 @@ def nem_sign_tx(connect, address, file, broadcast):
|
||||
payload = {"data": transaction.data.hex(), "signature": transaction.signature.hex()}
|
||||
|
||||
if broadcast:
|
||||
import requests
|
||||
|
||||
return requests.post(
|
||||
"{}/transaction/announce".format(broadcast), json=payload
|
||||
).json()
|
||||
|
||||
@ -1,10 +1,34 @@
|
||||
import hashlib
|
||||
from enum import Enum
|
||||
from typing import NewType, Tuple
|
||||
|
||||
import construct as c
|
||||
import ecdsa
|
||||
import pyblake2
|
||||
|
||||
from . import cosi, messages as proto, tools
|
||||
from . import cosi, messages, tools
|
||||
|
||||
V1_SIGNATURE_SLOTS = 3
|
||||
V1_BOOTLOADER_KEYS = {
|
||||
1: "04d571b7f148c5e4232c3814f777d8faeaf1a84216c78d569b71041ffc768a5b2d810fc3bb134dd026b57e65005275aedef43e155f48fc11a32ec790a93312bd58",
|
||||
2: "0463279c0c0866e50c05c799d32bd6bab0188b6de06536d1109d2ed9ce76cb335c490e55aee10cc901215132e853097d5432eda06b792073bd7740c94ce4516cb1",
|
||||
3: "0443aedbb6f7e71c563f8ed2ef64ec9981482519e7ef4f4aa98b27854e8c49126d4956d300ab45fdc34cd26bc8710de0a31dbdf6de7435fd0b492be70ac75fde58",
|
||||
4: "04877c39fd7c62237e038235e9c075dab261630f78eeb8edb92487159fffedfdf6046c6f8b881fa407c4a4ce6c28de0b19c1f4e29f1fcbc5a58ffd1432a3e0938a",
|
||||
5: "047384c51ae81add0a523adbb186c91b906ffb64c2c765802bf26dbd13bdf12c319e80c2213a136c8ee03d7874fd22b70d68e7dee469decfbbb510ee9a460cda45",
|
||||
}
|
||||
|
||||
V2_BOOTLOADER_KEYS = [
|
||||
bytes.fromhex("c2c87a49c5a3460977fbb2ec9dfe60f06bd694db8244bd4981fe3b7a26307f3f"),
|
||||
bytes.fromhex("80d036b08739b846f4cb77593078deb25dc9487aedcf52e30b4fb7cd7024178a"),
|
||||
bytes.fromhex("b8307a71f552c60a4cbb317ff48b82cdbf6b6bb5f04c920fec7badf017883751"),
|
||||
]
|
||||
V2_BOOTLOADER_M = 2
|
||||
V2_BOOTLOADER_N = 3
|
||||
|
||||
V2_CHUNK_SIZE = 1024 * 128
|
||||
|
||||
|
||||
def bytes_not(data):
|
||||
def bytes_not(data: bytes) -> bytes:
|
||||
return bytes(~b & 0xFF for b in data)
|
||||
|
||||
|
||||
@ -74,7 +98,8 @@ FirmwareHeader = c.Struct(
|
||||
c.Int32ul,
|
||||
lambda this:
|
||||
len(this._.code) if "code" in this._
|
||||
else (this.code_length or 0)),
|
||||
else (this.code_length or 0)
|
||||
),
|
||||
"version" / VersionLong,
|
||||
"fix_version" / VersionLong,
|
||||
"reserved" / c.Padding(8),
|
||||
@ -95,103 +120,176 @@ FirmwareHeader = c.Struct(
|
||||
Firmware = c.Struct(
|
||||
"vendor_header" / VendorHeader,
|
||||
"firmware_header" / FirmwareHeader,
|
||||
"_code_offset" / c.Tell,
|
||||
"code" / c.Bytes(c.this.firmware_header.code_length),
|
||||
c.Terminated,
|
||||
)
|
||||
|
||||
|
||||
FirmwareV1 = c.Struct(
|
||||
"magic" / c.Const(b"TRZR"),
|
||||
"code_length" / c.Rebuild(c.Int32ul, c.len_(c.this.code)),
|
||||
"key_indexes" / c.Int8ul[V1_SIGNATURE_SLOTS], # pylint: disable=E1136
|
||||
"flags" / c.BitStruct(
|
||||
c.Padding(7),
|
||||
"restore_storage" / c.Flag,
|
||||
),
|
||||
"reserved" / c.Padding(52),
|
||||
"signatures" / c.Bytes(64)[V1_SIGNATURE_SLOTS],
|
||||
"code" / c.Bytes(c.this.code_length),
|
||||
c.Terminated,
|
||||
)
|
||||
|
||||
# fmt: on
|
||||
|
||||
|
||||
def validate_firmware(filename):
|
||||
with open(filename, "rb") as f:
|
||||
data = f.read()
|
||||
if data[:6] == b"54525a":
|
||||
data = bytes.fromhex(data.decode())
|
||||
class FirmwareFormat(Enum):
|
||||
TREZOR_ONE = 1
|
||||
TREZOR_T = 2
|
||||
|
||||
|
||||
FirmwareType = NewType("FirmwareType", c.Container)
|
||||
ParsedFirmware = Tuple[FirmwareFormat, FirmwareType]
|
||||
|
||||
|
||||
def parse(data: bytes) -> ParsedFirmware:
|
||||
if data[:4] == b"TRZR":
|
||||
version = FirmwareFormat.TREZOR_ONE
|
||||
cls = FirmwareV1
|
||||
elif data[:4] == b"TRZV":
|
||||
version = FirmwareFormat.TREZOR_T
|
||||
cls = Firmware
|
||||
else:
|
||||
raise ValueError("Unrecognized firmware image type")
|
||||
|
||||
try:
|
||||
fw = Firmware.parse(data)
|
||||
fw = cls.parse(data)
|
||||
except Exception as e:
|
||||
raise ValueError("Invalid firmware image format") from e
|
||||
raise ValueError("Invalid firmware image") from e
|
||||
return version, FirmwareType(fw)
|
||||
|
||||
vendor = fw.vendor_header
|
||||
header = fw.firmware_header
|
||||
|
||||
print(
|
||||
"Vendor header from {}, version {}.{}".format(
|
||||
vendor.vendor_string, vendor.version.major, vendor.version.minor
|
||||
)
|
||||
)
|
||||
print(
|
||||
"Firmware version {v.major}.{v.minor}.{v.patch} build {v.build}".format(
|
||||
v=header.version
|
||||
)
|
||||
)
|
||||
def digest_v1(fw: FirmwareType) -> bytes:
|
||||
return hashlib.sha256(fw.code).digest()
|
||||
|
||||
# rebuild header without signatures
|
||||
|
||||
def check_sig_v1(fw: FirmwareType, idx: int) -> bool:
|
||||
key_idx = fw.key_indexes[idx]
|
||||
signature = fw.signatures[idx]
|
||||
|
||||
if key_idx == 0:
|
||||
# no signature = invalid signature
|
||||
return False
|
||||
|
||||
if key_idx not in V1_BOOTLOADER_KEYS:
|
||||
# unknown pubkey
|
||||
return False
|
||||
|
||||
pubkey = bytes.fromhex(V1_BOOTLOADER_KEYS[key_idx])[1:]
|
||||
verify = ecdsa.VerifyingKey.from_string(
|
||||
pubkey, curve=ecdsa.curves.SECP256k1, hashfunc=hashlib.sha256
|
||||
)
|
||||
try:
|
||||
verify.verify(signature, fw.code)
|
||||
return True
|
||||
except ecdsa.BadSignatureError:
|
||||
return False
|
||||
|
||||
|
||||
def _header_digest(header: c.Container, header_type: c.Construct) -> bytes:
|
||||
stripped_header = header.copy()
|
||||
stripped_header.sigmask = 0
|
||||
stripped_header.signature = b"\0" * 64
|
||||
header_bytes = FirmwareHeader.build(stripped_header)
|
||||
digest = pyblake2.blake2s(header_bytes).digest()
|
||||
header_bytes = header_type.build(stripped_header)
|
||||
return pyblake2.blake2s(header_bytes).digest()
|
||||
|
||||
print("Fingerprint: {}".format(digest.hex()))
|
||||
|
||||
global_pk = cosi.combine_keys(
|
||||
vendor.pubkeys[i] for i in range(8) if header.sigmask & (1 << i)
|
||||
def digest(fw: FirmwareType) -> bytes:
|
||||
return _header_digest(fw.firmware_header, FirmwareHeader)
|
||||
|
||||
|
||||
def validate(fw: FirmwareType, skip_vendor_header=False) -> bool:
|
||||
vendor_fingerprint = _header_digest(fw.vendor_header, VendorHeader)
|
||||
fingerprint = digest(fw)
|
||||
|
||||
if not skip_vendor_header:
|
||||
try:
|
||||
cosi.verify_m_of_n(
|
||||
fw.vendor_header.signature,
|
||||
vendor_fingerprint,
|
||||
V2_BOOTLOADER_M,
|
||||
V2_BOOTLOADER_N,
|
||||
fw.vendor_header.sigmask,
|
||||
V2_BOOTLOADER_KEYS,
|
||||
)
|
||||
# TODO support
|
||||
except Exception:
|
||||
raise ValueError("Invalid vendor header signature.")
|
||||
|
||||
# XXX expiry is not used now
|
||||
# now = time.gmtime()
|
||||
# if time.gmtime(fw.vendor_header.expiry) < now:
|
||||
# raise ValueError("Vendor header expired.")
|
||||
|
||||
try:
|
||||
cosi.verify(header.signature, digest, global_pk)
|
||||
print("Signature OK")
|
||||
cosi.verify_m_of_n(
|
||||
fw.firmware_header.signature,
|
||||
fingerprint,
|
||||
fw.vendor_header.vendor_sigs_required,
|
||||
fw.vendor_header.vendor_sigs_n,
|
||||
fw.firmware_header.sigmask,
|
||||
fw.vendor_header.pubkeys,
|
||||
)
|
||||
except Exception:
|
||||
print("Signature FAILED")
|
||||
raise
|
||||
raise ValueError("Invalid firmware signature.")
|
||||
|
||||
# XXX expiry is not used now
|
||||
# if time.gmtime(fw.firmware_header.expiry) < now:
|
||||
# raise ValueError("Firmware header expired.")
|
||||
|
||||
for i, expected_hash in enumerate(fw.firmware_header.hashes):
|
||||
if i == 0:
|
||||
# Because first chunk is sent along with headers, there is less code in it.
|
||||
chunk = fw.code[: V2_CHUNK_SIZE - fw._code_offset]
|
||||
else:
|
||||
# Subsequent chunks are shifted by the "missing header" size.
|
||||
ptr = i * V2_CHUNK_SIZE - fw._code_offset
|
||||
chunk = fw.code[ptr : ptr + V2_CHUNK_SIZE]
|
||||
|
||||
if not chunk and expected_hash == b"\0" * 32:
|
||||
continue
|
||||
chunk_hash = pyblake2.blake2s(chunk).digest()
|
||||
if chunk_hash != expected_hash:
|
||||
raise ValueError("Invalid firmware data.")
|
||||
|
||||
return True
|
||||
|
||||
|
||||
# ====== Client functions ====== #
|
||||
|
||||
|
||||
@tools.session
|
||||
def update(client, fp):
|
||||
def update(client, data):
|
||||
if client.features.bootloader_mode is False:
|
||||
raise RuntimeError("Device must be in bootloader mode")
|
||||
|
||||
data = fp.read()
|
||||
|
||||
resp = client.call(proto.FirmwareErase(length=len(data)))
|
||||
if isinstance(resp, proto.Failure) and resp.code == proto.FailureType.FirmwareError:
|
||||
return False
|
||||
resp = client.call(messages.FirmwareErase(length=len(data)))
|
||||
|
||||
# TREZORv1 method
|
||||
if isinstance(resp, proto.Success):
|
||||
# fingerprint = hashlib.sha256(data[256:]).hexdigest()
|
||||
# LOG.debug("Firmware fingerprint: " + fingerprint)
|
||||
resp = client.call(proto.FirmwareUpload(payload=data))
|
||||
if isinstance(resp, proto.Success):
|
||||
return True
|
||||
elif (
|
||||
isinstance(resp, proto.Failure)
|
||||
and resp.code == proto.FailureType.FirmwareError
|
||||
):
|
||||
return False
|
||||
if isinstance(resp, messages.Success):
|
||||
resp = client.call(messages.FirmwareUpload(payload=data))
|
||||
if isinstance(resp, messages.Success):
|
||||
return
|
||||
else:
|
||||
raise RuntimeError("Unexpected result %s" % resp)
|
||||
|
||||
# TREZORv2 method
|
||||
if isinstance(resp, proto.FirmwareRequest):
|
||||
import pyblake2
|
||||
|
||||
while True:
|
||||
while isinstance(resp, messages.FirmwareRequest):
|
||||
payload = data[resp.offset : resp.offset + resp.length]
|
||||
digest = pyblake2.blake2s(payload).digest()
|
||||
resp = client.call(proto.FirmwareUpload(payload=payload, hash=digest))
|
||||
if isinstance(resp, proto.FirmwareRequest):
|
||||
continue
|
||||
elif isinstance(resp, proto.Success):
|
||||
return True
|
||||
elif (
|
||||
isinstance(resp, proto.Failure)
|
||||
and resp.code == proto.FailureType.FirmwareError
|
||||
):
|
||||
return False
|
||||
raise RuntimeError("Unexpected result %s" % resp)
|
||||
resp = client.call(messages.FirmwareUpload(payload=payload, hash=digest))
|
||||
|
||||
if isinstance(resp, messages.Success):
|
||||
return
|
||||
else:
|
||||
raise RuntimeError("Unexpected message %s" % resp)
|
||||
|
||||
Loading…
Reference in New Issue
Block a user