Verify encrypted backup contents before reporting success

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)

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',

ports/stm32/boards/Passport/modules/backup_reader.py

@@ -0,0 +1,85 @@
+# SPDX-FileCopyrightText: © 2022 Foundation Devices, Inc. <hello@foundation.xyz>
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# SPDX-FileCopyrightText: 2018 Coinkite, Inc. <coldcardwallet.com>
+# 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

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.

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)

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

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 = {}

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)