diff --git a/CHANGELOG.md b/CHANGELOG.md index 761087490..b78592d05 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,7 @@ SPDX-License-Identifier: GPL-3.0-or-later --> ## Head +- Backup verification now decrypts backup contents and checks their integrity using the Backup Code - Added Coconut Wallet as a single-sig Connect Wallet option - Improved self-send transaction information formatting (PASS1-638) - Added the key manager extension, compatible with BIP85 and Nostr (PASS1-24) diff --git a/ports/stm32/boards/Passport/manifest.py b/ports/stm32/boards/Passport/manifest.py index d5f74bea8..afdbb908b 100644 --- a/ports/stm32/boards/Passport/manifest.py +++ b/ports/stm32/boards/Passport/manifest.py @@ -5,7 +5,8 @@ # Keep lists below sorted for easier reference freeze('$(MPY_DIR)/ports/stm32/boards/Passport/modules', - ('callgate.py', + ('backup_reader.py', + 'callgate.py', 'chains.py', 'common.py', 'compat7z.py', diff --git a/ports/stm32/boards/Passport/modules/backup_reader.py b/ports/stm32/boards/Passport/modules/backup_reader.py new file mode 100644 index 000000000..06a3619c1 --- /dev/null +++ b/ports/stm32/boards/Passport/modules/backup_reader.py @@ -0,0 +1,85 @@ +# SPDX-FileCopyrightText: © 2022 Foundation Devices, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later +# +# SPDX-FileCopyrightText: 2018 Coinkite, Inc. +# SPDX-License-Identifier: GPL-3.0-only +# +# Read and validate encrypted Passport backup files without changing device state. + +import compat7z + +from constants import MAX_BACKUP_FILE_SIZE +from errors import Error +from files import CardMissingError, CardSlot + + +def _clear_contents(contents): + if contents is not None: + for index in range(len(contents)): + contents[index] = 0 + + +def read_backup_file(decryption_password, backup_file_path, validate_only=False): + contents = None + + try: + with CardSlot(): + fd = open(backup_file_path, 'rb') + + try: + try: + compat7z.check_file_headers(fd) + except MemoryError: + return None, Error.OUT_OF_MEMORY_ERROR + except OSError: + return None, Error.FILE_READ_ERROR + except Exception: + return None, Error.INVALID_BACKUP_FILE_HEADER + + try: + zz = compat7z.Builder() + _fname, contents = zz.read_file( + fd, + decryption_password, + MAX_BACKUP_FILE_SIZE, + progress_fcn=None) + + # Match Restore's existing plaintext sanity check. + if contents[0:1] != b'#' or contents[-1:] != b'\n': + _clear_contents(contents) + return None, Error.INVALID_BACKUP_CODE + except MemoryError: + return None, Error.OUT_OF_MEMORY_ERROR + except OSError: + return None, Error.FILE_READ_ERROR + except Exception: + # The plaintext CRC deliberately does not distinguish a + # wrong code from damaged encrypted contents. + return None, Error.INVALID_BACKUP_CODE + finally: + fd.close() + except CardMissingError: + return None, Error.MICROSD_CARD_MISSING + except MemoryError: + _clear_contents(contents) + return None, Error.OUT_OF_MEMORY_ERROR + except OSError: + _clear_contents(contents) + return None, Error.FILE_READ_ERROR + except Exception: + _clear_contents(contents) + return None, Error.FILE_READ_ERROR + + if validate_only: + _clear_contents(contents) + return None, None + + return contents, None + + +def verify_backup_file(decryption_password, backup_file_path): + # Decryption necessarily materializes plaintext. Keep it inside this + # validation boundary and clear the mutable buffer before returning. + _contents, error = read_backup_file( + decryption_password, backup_file_path, validate_only=True) + return error diff --git a/ports/stm32/boards/Passport/modules/compat7z.py b/ports/stm32/boards/Passport/modules/compat7z.py index db7e8d3ee..35b820aad 100644 --- a/ports/stm32/boards/Passport/modules/compat7z.py +++ b/ports/stm32/boards/Passport/modules/compat7z.py @@ -117,7 +117,7 @@ def check_file_headers(f): # assume f is seekable fh = FileHeader.read(f) - if not fh.has_good_magic: + if not fh.has_good_magic(): raise ValueError("Bad magic bytes") # read only first header @@ -137,14 +137,17 @@ def check_file_headers(f): f.seek(sh.offset, 1) th = f.read(sh.size) if len(th) != sh.size: - raise IndexError("Truncated file? %s" % e.message) + raise IndexError( + "Truncated file: got %d of %d bytes" % (len(th), sh.size)) # Look for properties about compression. this could be # faked-out but good enough for now if b'\x24\x06\xf1\x07\x01' not in th: raise RuntimeError("Not marked as AES+SHA encrypted?") + except OSError: + raise except Exception as e: - raise ValueError("Confused file? %s" % e.message) + raise ValueError("Confused file? %s" % e) if masked_crc(th) != sh.crc: raise ValueError("Trailing header has wrong CRC") @@ -279,20 +282,28 @@ def read_file(self, fd, password, max_size, progress_fcn=None): # figure out key to be used key = self.calculate_key(password, progress_fcn) - out = b'' + out = bytearray(unpacked_size) # aes = tcc.AES(tcc.AES.CBC | tcc.AES.Decrypt, key, self.iv) aes = trezorcrypto.aes(trezorcrypto.aes.CBC, key, self.iv) - for blk in range(0, len(body), 16): - out += aes.decrypt(body[blk:blk + 16]) - - # trim padding, check CRC - out = out[0:unpacked_size] - if masked_crc(out) != expect_crc: - raise ValueError("Wrong password given, or damaged file.") - - # done. return contents - return fname, out + try: + for blk in range(0, len(body), 16): + decrypted = aes.decrypt(body[blk:blk + 16]) + end = min(blk + 16, unpacked_size) + if blk < end: + out[blk:end] = decrypted[0:end - blk] + + # Check the plaintext CRC after omitting block padding. + if masked_crc(out) != expect_crc: + raise ValueError("Wrong password given, or damaged file.") + + # Return a mutable buffer so callers that do not retain the + # plaintext can explicitly clear it. + return fname, out + except BaseException: + for i in range(len(out)): + out[i] = 0 + raise def verify_file_crc(self, fd, max_size, expected_sections=3): # Read each section, and check CRC of headers, return list of files & sizes. diff --git a/ports/stm32/boards/Passport/modules/flows/restore_backup_flow.py b/ports/stm32/boards/Passport/modules/flows/restore_backup_flow.py index 30dc626d7..b6b8c2a07 100644 --- a/ports/stm32/boards/Passport/modules/flows/restore_backup_flow.py +++ b/ports/stm32/boards/Passport/modules/flows/restore_backup_flow.py @@ -188,3 +188,9 @@ async def start_main_task(): elif error is Error.CORRUPT_BACKUP_FILE: await ErrorPage(text='Corrupt data in backup file. The backup may have been modified.').show() self.set_result(False) + elif error is Error.OUT_OF_MEMORY_ERROR: + await ErrorPage(text='Not enough memory to restore this backup.').show() + self.set_result(False) + else: + await ErrorPage(text='Unable to restore backup.').show() + self.set_result(False) diff --git a/ports/stm32/boards/Passport/modules/flows/verify_backup_flow.py b/ports/stm32/boards/Passport/modules/flows/verify_backup_flow.py index 13cb51c32..2a073559b 100644 --- a/ports/stm32/boards/Passport/modules/flows/verify_backup_flow.py +++ b/ports/stm32/boards/Passport/modules/flows/verify_backup_flow.py @@ -4,48 +4,92 @@ # verify_backup_flow.py - Verify a selected backup file. -from flows import Flow, FilePickerFlow -from pages import ErrorPage, SuccessPage, LongSuccessPage, InsertMicroSDPage -from utils import get_backups_folder_path, spinner_task +from constants import TOTAL_BACKUP_CODE_DIGITS +from flows import FilePickerFlow, Flow +from pages import BackupCodePage, ErrorPage, InsertMicroSDPage, LongSuccessPage, SuccessPage +from utils import get_backup_code_as_password, get_backups_folder_path, spinner_task from tasks import verify_backup_task from errors import Error +import microns import passport class VerifyBackupFlow(Flow): def __init__(self): super().__init__(initial_state=self.choose_file, name='VerifyBackupFlow') + self.backup_code = [None] * TOTAL_BACKUP_CODE_DIGITS + self.decryption_password = None async def choose_file(self): backups_path = get_backups_folder_path() result = await FilePickerFlow(initial_path=backups_path, suffix='.7z', show_folders=True).run() if result is None: # No file chosen, so go back to menu + self.clear_backup_code() self.set_result(False) return _filename, full_path, is_folder = result if not is_folder: self.backup_file_path = full_path - self.goto(self.do_verify) + self.goto(self.enter_backup_code) + + async def enter_backup_code(self): + result = await BackupCodePage( + digits=self.backup_code, + card_header={'title': 'Enter Backup Code'}).show() + if result is None: + self.back() + return + + self.backup_code = result + self.decryption_password = get_backup_code_as_password(self.backup_code) + self.goto(self.do_verify) async def do_verify(self): (error,) = await spinner_task( 'Verifying Backup', verify_backup_task, - args=[self.backup_file_path]) + args=[self.decryption_password, self.backup_file_path]) if error is None: + self.clear_backup_code() page_class = SuccessPage if passport.IS_COLOR else LongSuccessPage - await page_class(text='Backup file appears to be valid.\n\nPlease note this is only a check to ensure ' + - 'the file has not been modified or damaged.').show() + await page_class(text='Backup decrypted successfully and passed its integrity check.').show() self.set_result(True) elif error is Error.MICROSD_CARD_MISSING: result = await InsertMicroSDPage().show() if not result: + self.clear_backup_code() + self.set_result(False) + elif error is Error.INVALID_BACKUP_CODE: + result = await ErrorPage( + text='Unable to decrypt backup. The Backup Code may be incorrect, ' + 'or the backup may be damaged.', + left_micron=microns.Back, + right_micron=microns.Retry).show() + self.decryption_password = None + if result: + self.back() + else: + self.clear_backup_code() self.set_result(False) elif error is Error.FILE_READ_ERROR: - await ErrorPage(text='Unable to verify CRC of backup file. The backup may have been modified.').show() + self.clear_backup_code() + await ErrorPage(text='Unable to read backup file.').show() self.set_result(False) elif error is Error.INVALID_BACKUP_FILE_HEADER: + self.clear_backup_code() await ErrorPage(text='Unable to read backup file header. The backup may have been modified.').show() self.set_result(False) + elif error is Error.OUT_OF_MEMORY_ERROR: + self.clear_backup_code() + await ErrorPage(text='Not enough memory to verify this backup.').show() + self.set_result(False) + else: + self.clear_backup_code() + await ErrorPage(text='Unable to verify backup.').show() + self.set_result(False) + + def clear_backup_code(self): + self.backup_code = [None] * TOTAL_BACKUP_CODE_DIGITS + self.decryption_password = None diff --git a/ports/stm32/boards/Passport/modules/tasks/restore_backup_task.py b/ports/stm32/boards/Passport/modules/tasks/restore_backup_task.py index 5c170db47..d175af433 100644 --- a/ports/stm32/boards/Passport/modules/tasks/restore_backup_task.py +++ b/ports/stm32/boards/Passport/modules/tasks/restore_backup_task.py @@ -10,52 +10,21 @@ # restore_backup_task.py - Task for restoring Passport from a microSD backup file. import chains -import compat7z import stash import ujson -from files import CardSlot, CardMissingError +from backup_reader import read_backup_file from ubinascii import unhexlify as a2b_hex from errors import Error -from constants import MAX_BACKUP_FILE_SIZE from pincodes import SE_SECRET_LEN async def restore_backup_task(on_done, decryption_password, backup_file_path): from common import pa, settings - try: - with CardSlot() as card: - fd = open(backup_file_path, 'rb') - - try: - try: - compat7z.check_file_headers(fd) - except Exception as e: - await on_done(Error.INVALID_BACKUP_FILE_HEADER) - return - - try: - zz = compat7z.Builder() - fname, contents = zz.read_file(fd, decryption_password, MAX_BACKUP_FILE_SIZE, - progress_fcn=None) - - # Quick sanity check - assert contents[0:1] == b'#' and contents[-1:] == b'\n' - - except Exception as e: - # Assume all exceptions here are "incorrect password" errors - await on_done(Error.INVALID_BACKUP_CODE) - return - - finally: - fd.close() - - except CardMissingError: - await on_done(Error.MICROSD_CARD_MISSING) - return - except BaseException: - await on_done(Error.FILE_READ_ERROR) + contents, error = read_backup_file(decryption_password, backup_file_path) + if error is not None: + await on_done(error) return vals = {} diff --git a/ports/stm32/boards/Passport/modules/tasks/verify_backup_task.py b/ports/stm32/boards/Passport/modules/tasks/verify_backup_task.py index 15811f63f..4b99783d5 100644 --- a/ports/stm32/boards/Passport/modules/tasks/verify_backup_task.py +++ b/ports/stm32/boards/Passport/modules/tasks/verify_backup_task.py @@ -10,38 +10,9 @@ # verify_backup_task.py - Task for verifying a backup from microSD. -import compat7z +from backup_reader import verify_backup_file -from files import CardSlot, CardMissingError -from errors import Error -from constants import MAX_BACKUP_FILE_SIZE - -async def verify_backup_task(on_done, backup_file_path): - try: - with CardSlot() as card: - fd = open(backup_file_path, 'rb') - - try: - try: - compat7z.check_file_headers(fd) - except Exception as e: - await on_done(Error.INVALID_BACKUP_FILE_HEADER) - return - - zz = compat7z.Builder() - files = zz.verify_file_crc(fd, MAX_BACKUP_FILE_SIZE) - - assert len(files) == 1 - fname, fsize = files[0] - - finally: - fd.close() - except CardMissingError: - await on_done(Error.MICROSD_CARD_MISSING) - return - except Exception as e: - await on_done(Error.FILE_READ_ERROR) - return - - await on_done(None) +async def verify_backup_task(on_done, decryption_password, backup_file_path): + error = verify_backup_file(decryption_password, backup_file_path) + await on_done(error) diff --git a/ports/stm32/boards/Passport/modules/tests/test_backup_reader.py b/ports/stm32/boards/Passport/modules/tests/test_backup_reader.py new file mode 100644 index 000000000..b3f408ef8 --- /dev/null +++ b/ports/stm32/boards/Passport/modules/tests/test_backup_reader.py @@ -0,0 +1,206 @@ +# SPDX-FileCopyrightText: © 2026 Foundation Devices, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +import builtins +import importlib.util +import os +import sys +import types + +import pytest + + +MODULES_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +READER_PATH = os.path.join(MODULES_DIR, 'backup_reader.py') +sys.path.insert(1, MODULES_DIR) + +from errors import Error + + +class CardMissingError(Exception): + pass + + +class CardSlot: + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, traceback): + return False + + +class FakeFile: + def __init__(self): + self.closed = False + + def close(self): + self.closed = True + + +class ForbiddenDeviceState(types.ModuleType): + def __getattr__(self, name): + raise AssertionError('backup verification accessed common.{}'.format(name)) + + +def load_reader(monkeypatch, check_headers=None, read_file=None, card_slot=CardSlot): + calls = types.SimpleNamespace(header=0, reads=[]) + compat7z = types.ModuleType('compat7z') + + def default_check_headers(_fd): + calls.header += 1 + + compat7z.check_file_headers = check_headers or default_check_headers + + class Builder: + def read_file(self, fd, password, max_size, progress_fcn=None): + calls.reads.append((fd, password, max_size, progress_fcn)) + if read_file is not None: + return read_file(fd, password, max_size, progress_fcn) + return 'passport-backup.txt', bytearray(b'# Passport backup\n') + + compat7z.Builder = Builder + + files = types.ModuleType('files') + files.CardSlot = card_slot + files.CardMissingError = CardMissingError + + constants = types.ModuleType('constants') + constants.MAX_BACKUP_FILE_SIZE = 4096 + + monkeypatch.setitem(sys.modules, 'compat7z', compat7z) + monkeypatch.setitem(sys.modules, 'files', files) + monkeypatch.setitem(sys.modules, 'constants', constants) + + spec = importlib.util.spec_from_file_location('backup_reader_under_test', READER_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module, calls + + +def test_reads_and_validates_backup_without_accessing_device_state(monkeypatch): + fd = FakeFile() + monkeypatch.setitem(sys.modules, 'common', ForbiddenDeviceState('common')) + monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: fd) + module, calls = load_reader(monkeypatch) + + contents, error = module.read_backup_file('backup-code', 'backup.7z') + + assert contents == b'# Passport backup\n' + assert error is None + assert calls.header == 1 + assert calls.reads == [(fd, 'backup-code', 4096, None)] + assert fd.closed + + +def test_verify_clears_plaintext_before_returning(monkeypatch): + fd = FakeFile() + plaintext = bytearray(b'# Passport backup\n') + monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: fd) + + def return_plaintext(_fd, _password, _max_size, _progress_fcn): + return 'passport-backup.txt', plaintext + + module, _calls = load_reader(monkeypatch, read_file=return_plaintext) + + assert module.verify_backup_file('backup-code', 'backup.7z') is None + assert plaintext == bytearray(len(plaintext)) + + +@pytest.mark.parametrize( + ('failure', 'expected_error'), + ( + (MemoryError(), Error.OUT_OF_MEMORY_ERROR), + (OSError('read failed'), Error.FILE_READ_ERROR), + (ValueError('bad header'), Error.INVALID_BACKUP_FILE_HEADER), + )) +def test_classifies_header_failures(monkeypatch, failure, expected_error): + fd = FakeFile() + monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: fd) + + def reject_header(_fd): + raise failure + + module, _calls = load_reader(monkeypatch, check_headers=reject_header) + + assert module.read_backup_file('backup-code', 'backup.7z') == (None, expected_error) + assert fd.closed + + +@pytest.mark.parametrize( + ('failure', 'expected_error'), + ( + (MemoryError(), Error.OUT_OF_MEMORY_ERROR), + (OSError('read failed'), Error.FILE_READ_ERROR), + (ValueError('wrong code or damaged body'), Error.INVALID_BACKUP_CODE), + )) +def test_classifies_decryption_failures(monkeypatch, failure, expected_error): + fd = FakeFile() + monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: fd) + + def reject_body(_fd, _password, _max_size, _progress_fcn): + raise failure + + module, _calls = load_reader(monkeypatch, read_file=reject_body) + + assert module.read_backup_file('backup-code', 'backup.7z') == (None, expected_error) + assert fd.closed + + +def test_rejects_plaintext_that_is_not_a_passport_backup(monkeypatch): + fd = FakeFile() + plaintext = bytearray(b'not a Passport backup') + monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: fd) + + def invalid_plaintext(_fd, _password, _max_size, _progress_fcn): + return 'passport-backup.txt', plaintext + + module, _calls = load_reader(monkeypatch, read_file=invalid_plaintext) + + assert module.read_backup_file('backup-code', 'backup.7z') == (None, Error.INVALID_BACKUP_CODE) + assert plaintext == bytearray(len(plaintext)) + assert fd.closed + + +def test_classifies_card_removal_and_open_failures(monkeypatch): + class MissingCardSlot: + def __enter__(self): + raise CardMissingError + + def __exit__(self, exc_type, exc_value, traceback): + return False + + module, _calls = load_reader(monkeypatch, card_slot=MissingCardSlot) + assert module.read_backup_file('backup-code', 'backup.7z') == (None, Error.MICROSD_CARD_MISSING) + + def fail_open(*_args, **_kwargs): + raise OSError('open failed') + + monkeypatch.setattr(builtins, 'open', fail_open) + module, _calls = load_reader(monkeypatch) + assert module.read_backup_file('backup-code', 'backup.7z') == (None, Error.FILE_READ_ERROR) + + def fail_open_with_oom(*_args, **_kwargs): + raise MemoryError + + monkeypatch.setattr(builtins, 'open', fail_open_with_oom) + module, _calls = load_reader(monkeypatch) + assert module.read_backup_file('backup-code', 'backup.7z') == ( + None, Error.OUT_OF_MEMORY_ERROR) + + +def test_close_failure_clears_plaintext(monkeypatch): + plaintext = bytearray(b'# Passport backup\n') + + class CloseFailure(FakeFile): + def close(self): + raise OSError('close failed') + + def return_plaintext(_fd, _password, _max_size, _progress_fcn): + return 'passport-backup.txt', plaintext + + monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: CloseFailure()) + module, _calls = load_reader(monkeypatch, read_file=return_plaintext) + + assert module.read_backup_file('backup-code', 'backup.7z') == ( + None, Error.FILE_READ_ERROR) + assert plaintext == bytearray(len(plaintext)) diff --git a/ports/stm32/boards/Passport/modules/tests/test_unit.py b/ports/stm32/boards/Passport/modules/tests/test_unit.py index a088f75fb..474b9bb0c 100644 --- a/ports/stm32/boards/Passport/modules/tests/test_unit.py +++ b/ports/stm32/boards/Passport/modules/tests/test_unit.py @@ -20,6 +20,10 @@ def test_ext_settings(test): assert test('ext_settings.py') == b'OK' +def test_backup_verification(test): + assert test('backup_verification.py') == b'OK' + + def test_ui(test): assert test('ui.py') == b'OK' diff --git a/ports/stm32/boards/Passport/modules/tests/test_verify_backup_flow.py b/ports/stm32/boards/Passport/modules/tests/test_verify_backup_flow.py new file mode 100644 index 000000000..fbf26b5cc --- /dev/null +++ b/ports/stm32/boards/Passport/modules/tests/test_verify_backup_flow.py @@ -0,0 +1,248 @@ +# SPDX-FileCopyrightText: © 2026 Foundation Devices, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later + +import asyncio +import importlib.util +import os +import sys +import types + +import pytest + + +MODULES_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +FLOW_PATH = os.path.join(MODULES_DIR, 'flows', 'verify_backup_flow.py') +sys.path.insert(1, MODULES_DIR) + +from errors import Error + + +class Flow: + def __init__(self, initial_state, name): + self.state = initial_state + self.prev_states = [] + self.name = name + self.result = None + self.went_back = False + + def goto(self, state): + self.prev_states.append(self.state) + self.state = state + + def back(self): + self.went_back = True + self.state = self.prev_states.pop() + + def set_result(self, result): + self.result = result + + +class FilePickerFlow: + result = None + + def __init__(self, **_kwargs): + pass + + async def run(self): + return type(self).result + + +class FakePage: + shown = [] + result = None + + def __init__(self, **kwargs): + self.kwargs = kwargs + + async def show(self): + type(self).shown.append(self.kwargs) + return type(self).result + + +class BackupCodePage(FakePage): + pass + + +class ErrorPage(FakePage): + pass + + +class InsertMicroSDPage(FakePage): + pass + + +class LongSuccessPage(FakePage): + pass + + +class SuccessPage(FakePage): + pass + + +def load_flow(monkeypatch, is_color, spinner_error=None): + for page in (BackupCodePage, ErrorPage, InsertMicroSDPage, LongSuccessPage, SuccessPage): + page.shown = [] + page.result = None + FilePickerFlow.result = None + + constants = types.ModuleType('constants') + constants.TOTAL_BACKUP_CODE_DIGITS = 20 + + flows = types.ModuleType('flows') + flows.Flow = Flow + flows.FilePickerFlow = FilePickerFlow + + pages = types.ModuleType('pages') + pages.BackupCodePage = BackupCodePage + pages.ErrorPage = ErrorPage + pages.InsertMicroSDPage = InsertMicroSDPage + pages.LongSuccessPage = LongSuccessPage + pages.SuccessPage = SuccessPage + + spinner_calls = [] + utils = types.ModuleType('utils') + utils.get_backup_code_as_password = lambda digits: ''.join(str(digit) for digit in digits) + utils.get_backups_folder_path = lambda: '/backups' + + async def spinner_task(title, task, args): + spinner_calls.append((title, task, args)) + return (spinner_error,) + + utils.spinner_task = spinner_task + + tasks = types.ModuleType('tasks') + tasks.verify_backup_task = object() + + microns = types.ModuleType('microns') + microns.Back = object() + microns.Retry = object() + + passport = types.ModuleType('passport') + passport.IS_COLOR = is_color + + monkeypatch.setitem(sys.modules, 'constants', constants) + monkeypatch.setitem(sys.modules, 'flows', flows) + monkeypatch.setitem(sys.modules, 'pages', pages) + monkeypatch.setitem(sys.modules, 'utils', utils) + monkeypatch.setitem(sys.modules, 'tasks', tasks) + monkeypatch.setitem(sys.modules, 'microns', microns) + monkeypatch.setitem(sys.modules, 'passport', passport) + + spec = importlib.util.spec_from_file_location('verify_backup_flow_under_test', FLOW_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module.VerifyBackupFlow(), spinner_calls, tasks.verify_backup_task + + +def run(coroutine): + asyncio.run(coroutine) + + +def test_selected_file_routes_through_backup_code_before_verification(monkeypatch): + flow, _spinner_calls, _task = load_flow(monkeypatch, is_color=True) + digits = list(range(10)) * 2 + FilePickerFlow.result = ('backup.7z', '/backups/backup.7z', False) + BackupCodePage.result = digits + + run(flow.state()) + assert flow.state == flow.enter_backup_code + + run(flow.state()) + + assert flow.backup_code == digits + assert flow.decryption_password == ''.join(str(digit) for digit in digits) + assert flow.state == flow.do_verify + + +def test_selecting_folder_does_not_advance_to_code_entry(monkeypatch): + flow, _spinner_calls, _task = load_flow(monkeypatch, is_color=True) + FilePickerFlow.result = ('backups', '/backups', True) + + run(flow.state()) + + assert flow.state == flow.choose_file + assert flow.result is None + + +@pytest.mark.parametrize( + ('is_color', 'success_page'), + ((True, SuccessPage), (False, LongSuccessPage))) +def test_success_requires_decryption_on_both_screen_variants( + monkeypatch, is_color, success_page): + flow, spinner_calls, task = load_flow(monkeypatch, is_color=is_color) + flow.backup_file_path = '/backups/backup.7z' + flow.decryption_password = 'backup-code' + flow.backup_code = [1] * 20 + + run(flow.do_verify()) + + assert spinner_calls == [( + 'Verifying Backup', + task, + ['backup-code', '/backups/backup.7z'])] + assert success_page.shown == [{ + 'text': 'Backup decrypted successfully and passed its integrity check.'}] + assert flow.backup_code == [None] * 20 + assert flow.decryption_password is None + assert flow.result is True + + +def test_integrity_failure_returns_to_code_entry_without_leaking_detail(monkeypatch): + flow, _spinner_calls, _task = load_flow( + monkeypatch, is_color=True, spinner_error=Error.INVALID_BACKUP_CODE) + flow.backup_file_path = '/backups/backup.7z' + flow.decryption_password = 'backup-code' + flow.backup_code = [1] * 20 + flow.state = flow.do_verify + flow.prev_states = [flow.choose_file, flow.enter_backup_code] + ErrorPage.result = True + + run(flow.do_verify()) + + assert ErrorPage.shown == [{ + 'text': 'Unable to decrypt backup. The Backup Code may be incorrect, ' + 'or the backup may be damaged.', + 'left_micron': sys.modules['microns'].Back, + 'right_micron': sys.modules['microns'].Retry, + }] + assert flow.decryption_password is None + assert flow.backup_code == [1] * 20 + assert flow.went_back + assert flow.state == flow.enter_backup_code + assert flow.result is None + + +def test_card_cancel_clears_backup_code_and_exits(monkeypatch): + flow, _spinner_calls, _task = load_flow( + monkeypatch, is_color=True, spinner_error=Error.MICROSD_CARD_MISSING) + flow.backup_file_path = '/backups/backup.7z' + flow.decryption_password = 'backup-code' + flow.backup_code = [1] * 20 + InsertMicroSDPage.result = False + + run(flow.do_verify()) + + assert flow.backup_code == [None] * 20 + assert flow.decryption_password is None + assert flow.result is False + + +@pytest.mark.parametrize( + ('error', 'message'), + ( + (Error.OUT_OF_MEMORY_ERROR, 'Not enough memory to verify this backup.'), + (object(), 'Unable to verify backup.'), + )) +def test_unexpected_and_out_of_memory_errors_are_terminal(monkeypatch, error, message): + flow, _spinner_calls, _task = load_flow( + monkeypatch, is_color=True, spinner_error=error) + flow.backup_file_path = '/backups/backup.7z' + flow.decryption_password = 'backup-code' + flow.backup_code = [1] * 20 + + run(flow.do_verify()) + + assert ErrorPage.shown == [{'text': message}] + assert flow.backup_code == [None] * 20 + assert flow.decryption_password is None + assert flow.result is False diff --git a/ports/stm32/boards/Passport/modules/tests/test_verify_backup_task.py b/ports/stm32/boards/Passport/modules/tests/test_verify_backup_task.py index cf9ece130..eaa94de06 100644 --- a/ports/stm32/boards/Passport/modules/tests/test_verify_backup_task.py +++ b/ports/stm32/boards/Passport/modules/tests/test_verify_backup_task.py @@ -2,7 +2,6 @@ # SPDX-License-Identifier: GPL-3.0-or-later import asyncio -import builtins import importlib.util import os import sys @@ -16,43 +15,21 @@ from errors import Error -class CardMissingError(Exception): - pass +def load_task(monkeypatch, error=None): + calls = [] + backup_reader = types.ModuleType('backup_reader') + def verify_backup_file(password, path): + calls.append((password, path)) + return error -class CardSlot: - def __enter__(self): - return self - - def __exit__(self, exc_type, exc_value, traceback): - return False - - -def load_task(monkeypatch, card_slot=CardSlot, check_headers=None, files=None): - compat7z = types.ModuleType('compat7z') - compat7z.check_file_headers = check_headers or (lambda _fd: None) - - class Builder: - def verify_file_crc(self, _fd, _max_size): - return files or [('passport-backup.txt', 1)] - - compat7z.Builder = Builder - - files_module = types.ModuleType('files') - files_module.CardSlot = card_slot - files_module.CardMissingError = CardMissingError - - constants = types.ModuleType('constants') - constants.MAX_BACKUP_FILE_SIZE = 1024 - - monkeypatch.setitem(sys.modules, 'compat7z', compat7z) - monkeypatch.setitem(sys.modules, 'files', files_module) - monkeypatch.setitem(sys.modules, 'constants', constants) + backup_reader.verify_backup_file = verify_backup_file + monkeypatch.setitem(sys.modules, 'backup_reader', backup_reader) spec = importlib.util.spec_from_file_location('verify_backup_task_under_test', TASK_PATH) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - return module.verify_backup_task + return module.verify_backup_task, calls def run_task(task): @@ -61,45 +38,29 @@ def run_task(task): async def on_done(error): results.append(error) - asyncio.run(task(on_done, 'backup.7z')) + asyncio.run(task(on_done, '1111-2222-3333-4444-5555', 'backup.7z')) return results -def test_success_reports_once(monkeypatch): - fd = types.SimpleNamespace(close=lambda: None) - monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: fd) - - assert run_task(load_task(monkeypatch)) == [None] - - -def test_card_removal_reports_once(monkeypatch): - class MissingCardSlot: - def __enter__(self): - raise CardMissingError - - def __exit__(self, exc_type, exc_value, traceback): - return False - - assert run_task(load_task(monkeypatch, card_slot=MissingCardSlot)) == [Error.MICROSD_CARD_MISSING] - - -def test_file_read_failure_reports_once(monkeypatch): - def fail_open(*_args, **_kwargs): - raise OSError('read failed') - - monkeypatch.setattr(builtins, 'open', fail_open) - - assert run_task(load_task(monkeypatch)) == [Error.FILE_READ_ERROR] +def test_success_reports_once_and_passes_credentials(monkeypatch): + task, calls = load_task(monkeypatch) + assert run_task(task) == [None] + assert calls == [('1111-2222-3333-4444-5555', 'backup.7z')] -def test_invalid_header_reports_once(monkeypatch): - fd = types.SimpleNamespace(close=lambda: None) - monkeypatch.setattr(builtins, 'open', lambda *_args, **_kwargs: fd) - def reject_header(_fd): - raise ValueError('invalid header') +def test_each_reader_failure_reports_once(monkeypatch): + errors = ( + Error.MICROSD_CARD_MISSING, + Error.FILE_READ_ERROR, + Error.INVALID_BACKUP_FILE_HEADER, + Error.INVALID_BACKUP_CODE, + Error.OUT_OF_MEMORY_ERROR, + ) - assert run_task(load_task(monkeypatch, check_headers=reject_header)) == [Error.INVALID_BACKUP_FILE_HEADER] + for error in errors: + task, _calls = load_task(monkeypatch, error=error) + assert run_task(task) == [error] def test_related_error_members_are_available(): diff --git a/ports/stm32/boards/Passport/modules/tests/unit/backup_verification.py b/ports/stm32/boards/Passport/modules/tests/unit/backup_verification.py new file mode 100644 index 000000000..bba00a801 --- /dev/null +++ b/ports/stm32/boards/Passport/modules/tests/unit/backup_verification.py @@ -0,0 +1,89 @@ +# SPDX-FileCopyrightText: © 2026 Foundation Devices, Inc. +# SPDX-License-Identifier: GPL-3.0-or-later +# +# Exercise backup encryption and validation using the simulator's MicroPython +# crypto implementation. + +import compat7z + +from uio import BytesIO + + +PASSWORD = '1111-2222-3333-4444-5555' +CONTENTS = b'#' + (b'a' * 30) + b'\n' +MAX_SIZE = 1024 + + +def expect_failure(expected_type, operation, expected_message=None): + try: + operation() + except expected_type as error: + if expected_message is not None: + assert expected_message in str(error) + return + except Exception as error: + assert False, 'expected {}, got {}'.format(expected_type, type(error)) + assert False, 'operation unexpectedly succeeded' + + +def validate(archive, password=PASSWORD, max_size=MAX_SIZE): + fd = BytesIO(archive) + compat7z.check_file_headers(fd) + return compat7z.Builder().read_file(fd, password, max_size, progress_fcn=None) + + +# Avoid touching simulator device state while generating the fixture. +compat7z.urandom = lambda length: bytes(range(length)) + +builder = compat7z.Builder(password=PASSWORD) +builder.add_data(CONTENTS) +prefix, footer = builder.save('passport-backup.txt') +archive = prefix + builder.body + footer + +filename, plaintext = validate(archive) +assert filename == 'passport-backup.txt' +assert plaintext == CONTENTS +assert isinstance(plaintext, bytearray) + +expect_failure( + ValueError, + lambda: validate(archive, password='0000-0000-0000-0000-0000'), + 'Wrong password given, or damaged file.') + +damaged_body = bytearray(archive) +damaged_body[len(prefix) + 5] ^= 1 +expect_failure( + ValueError, + lambda: validate(damaged_body), + 'Wrong password given, or damaged file.') + +damaged_magic = bytearray(archive) +damaged_magic[0] ^= 1 +expect_failure(ValueError, lambda: validate(damaged_magic), 'Bad magic bytes') + +aes_marker = b'\x24\x06\xf1\x07\x01' +marker_offset = archive.find(aes_marker) +assert marker_offset > len(prefix) + len(builder.body) +damaged_metadata = bytearray(archive) +damaged_metadata[marker_offset] ^= 1 +expect_failure(ValueError, lambda: validate(damaged_metadata), 'Not marked as AES+SHA encrypted') + +for truncate_at in ( + 0, + 11, + len(prefix) - 1, + len(prefix) + len(builder.body) - 1, + len(archive) - 1): + expect_failure(ValueError, lambda: validate(archive[:truncate_at])) + +expect_failure( + ValueError, + lambda: validate(archive[:-1]), + 'Truncated file: got') + +expect_failure( + AssertionError, + lambda: validate(archive, max_size=len(CONTENTS) - 1), + 'too big') + +return_value.write(b'OK')