feat(core): add N1W1 backup/recovery demo flows

Can be tested by invoking:
```
$ core/emu.py -ea -c trezorctl device setup -b shamir   # will run multi-share backup
$ core/tools/n1w1-emu.py run 127.0.0.1:21325 /tmp/tagN  # simulate tag connection and I/O
```

[no changelog]

Author: romanz

.github/workflows/core.yml

@@ -152,7 +152,6 @@ jobs:
           retention-days: 7
 
   core_emu_arm:
-    if: github.event_name == 'schedule'
     name: Build emu arm
     runs-on: ubuntu-latest-arm64
     needs: param
@@ -781,8 +780,8 @@ jobs:
     if: github.event_name != 'schedule' && github.repository == 'trezor/trezor-firmware'
     runs-on: ubuntu-latest
     needs:
-      # Do not include ARM, they are only built on nightly
       - core_emu
+      - core_emu_arm
     steps:
       - uses: actions/download-artifact@70fc10c6e5e1ce46ad2ea6f2b72d43f7d47b13c3  # actions/download-artifact@v8.0.0
         with:

core/src/apps/management/recovery_device/layout.py

@@ -2,7 +2,7 @@
 
 import storage.recovery as storage_recovery
 import storage.recovery_shares as storage_recovery_shares
-from trezor import TR
+from trezor import TR, utils
 from trezor.ui.layouts.recovery import (  # noqa: F401
     request_word_count,
     show_already_added,
@@ -278,12 +278,117 @@ async def request_mnemonic(self) -> str | None:
             return None
 
 
-async def choose_handler(method: BackupMethod | None) -> type[RecoveryHandler]:
-    from trezor.enums import BackupMethod
+if not utils.USE_N4W1:
 
-    if method is not BackupMethod.Display and __debug__:
-        from trezor import log
+    async def choose_handler(method: BackupMethod | None) -> type[RecoveryHandler]:
+        from trezor.enums import BackupMethod
 
-        log.warning(__name__, "Unsupported backup method: %s", method)
+        if method is not BackupMethod.Display and __debug__:
+            from trezor import log
 
-    return _DisplayHandler
+            log.warning(__name__, "Unsupported backup method: %s", method)
+
+        return _DisplayHandler
+
+else:
+
+    if TYPE_CHECKING:
+        from trezor.messages import BackupMethod
+
+        from .recover import Slip39State
+
+    async def choose_handler(method: BackupMethod | None) -> type[RecoveryHandler]:
+        from trezor.enums import BackupMethod
+
+        if method is None:
+            method = await _choose_method()
+
+        if method is BackupMethod.N4W1:
+            return _N4W1Handler
+
+        if method not in (None, BackupMethod.Display):
+            from trezor import log
+
+            if __debug__:
+                log.warning(__name__, "Unsupported backup method: %s", method)
+
+        return _DisplayHandler
+
+    async def _choose_method() -> BackupMethod:
+        import trezorui_api
+        from trezor.enums import BackupMethod
+        from trezor.ui.layouts import interact
+
+        index = await interact(
+            trezorui_api.select_word(
+                title=TR.recovery__title,
+                description="Which type of wallet backup do you have?",
+                words=("N4W1 backup", "Wordlist backup", ""),
+            ),
+            br_name="backup_choose",
+        )
+        return (BackupMethod.N4W1, BackupMethod.Display)[index]
+
+    class _N4W1Handler:
+        def __init__(
+            self,
+            recovery_type: RecoveryType,
+            slip39_state: Slip39State | None,
+        ) -> None:
+            super().__init__()
+            self.recovery_type = recovery_type
+            # `slip39_state is None` indicates that we are (re)starting the first recovery step.
+            self.slip39_state = slip39_state
+
+        @classmethod
+        async def load(cls, recovery_type: RecoveryType) -> "RecoveryHandler":
+            return cls(recovery_type, load_slip39_state())
+
+        async def show_state(self, is_retry: bool) -> None:
+            if is_retry or self.slip39_state is None:
+                # don't show recovery state on retries and before the first share is entered
+                return
+            word_count = self.slip39_state[0]
+            await _request_share_first_screen(word_count, self.recovery_type)
+
+        async def request_mnemonic(self) -> str | None:
+            """Return the mnemonic or `None` on cancellation/validation error."""
+            from apps.debug import n4w1_mock
+
+            with n4w1_mock.ctx as ctx:
+                # returns `None` on cancellation or retriable error.
+                await ctx.confirm_connect(
+                    title=TR.recovery__title,
+                    description=TR.n4w1__hold_next,
+                    button=TR.n4w1__footer_next,
+                    br_name="backup_read",
+                )
+                # continue N4W1 communication (the tag is connected)
+
+                # TODO: animate during read?
+                # TODO: run mnemonic checks during animation
+                # TODO: show empty tag warning
+                if (blob := await ctx.read(key="mnemonic")) is None:
+                    return None
+
+            # TODO: use protobuf?
+            share = bytes(blob).decode()
+            return await self.check_words(share)
+
+        async def check_words(self, share: str) -> str | None:
+            # Can be `None` when checking the first share.
+            backup_type = self.slip39_state and self.slip39_state[1]
+            share_words = share.split(" ")
+
+            try:
+                from .word_validity import WordValidityResult, check
+
+                # Re-verify relevant prefixes
+                # TODO: can it be encapsulated too?
+                for prefix_len in range(1, 1 + len(share_words)):
+                    check(backup_type, partial_mnemonic=share_words[:prefix_len])
+                return share
+            except WordValidityResult as exc:
+                # if they were invalid or some checks failed we continue and request them again
+                await exc.show_error()
+                return None

core/src/apps/management/reset_device/layout.py

@@ -1,6 +1,7 @@
 from micropython import const
 from typing import TYPE_CHECKING, Iterable, Protocol, Sequence
 
+from trezor import utils
 from trezor.ui.layouts.reset import (  # noqa: F401
     show_share_words,
     slip39_advanced_prompt_group_threshold,
@@ -195,13 +196,123 @@ async def _backup_share(self, share: ShareInfo) -> None:
 
 
 async def choose_backup_handler(method: BackupMethod | None) -> BackupHandler:
-    # TODO: prompt the user if method is `None`.
-    if __debug__:
-        from trezor.enums import BackupMethod
+    from trezor.enums import BackupMethod
+
+    if utils.USE_N4W1:
+        if method is None:
+            method = await _choose_method()
+
+        if method is BackupMethod.N4W1:
+            return _N4W1Backup()
 
-        if method not in (None, BackupMethod.Display):
-            from trezor import log
+    if method not in (None, BackupMethod.Display):
+        from trezor import log
 
+        if __debug__:
             log.warning(__name__, "Unsupported backup method: %s", method)
 
     return _DisplayBackup()
+
+
+if utils.USE_N4W1:
+
+    from trezor import TR
+    from trezor.ui.layouts.common import interact
+
+    if TYPE_CHECKING:
+        from buffer_types import AnyBytes
+
+        from apps.debug.n4w1_mock import N4W1Context
+
+    class Retry(Exception):
+        def __init__(self, msg: str) -> None:
+            self.msg = msg
+
+    class _N4W1Backup:
+
+        async def intro(self, num_of_words: int | None = None) -> None:
+            # TODO: design/copy
+            pass
+
+        async def backup(self, iter_shares: Iterable[ShareInfo]) -> None:
+            # TODO: warn user about safety
+
+            # backup all shares
+            for share in iter_shares:
+                await self._backup_share(share)
+
+        async def _backup_share(self, share: ShareInfo) -> None:
+            import trezorui_api
+            from trezor import TR
+            from trezor.ui.layouts.common import raise_if_not_confirmed
+
+            from apps.debug import n4w1_mock
+
+            # TODO: use protobuf?
+            blob = " ".join(share.words).encode()
+
+            if share.index == 0 or share.num_of_shares is None:
+                description, button = TR.n4w1__hold_first, TR.n4w1__footer_first
+            elif share.index == share.num_of_shares - 1:
+                description, button = TR.n4w1__hold_last, TR.n4w1__footer_last
+            else:
+                description, button = TR.n4w1__hold_next, TR.n4w1__footer_next
+
+            while True:
+                try:
+                    with n4w1_mock.ctx as ctx:
+                        # may raise a retriable error
+                        return await _backup_share(ctx, description, button, blob)
+                except Retry as exc:
+                    await raise_if_not_confirmed(
+                        trezorui_api.show_warning(
+                            title=TR.words__important,
+                            button=TR.buttons__continue,
+                            description=exc.msg,
+                            danger=True,
+                        ),
+                        br_name="backup_retry",
+                    )
+                    # wait for a new N4W1 tag
+                    continue
+
+    async def _backup_share(
+        ctx: N4W1Context, description: str, button: str, blob: AnyBytes
+    ) -> None:
+        from trezor.ui.layouts.progress import progress
+
+        await ctx.confirm_connect(
+            title=TR.backup__title_create_wallet_backup,
+            description=description,
+            button=button,
+            br_name="backup_write",
+        )
+        # continue N4W1 communication (the tag is connected)
+        result = await ctx.read(key="mnemonic")
+        if result is not None:
+            raise Retry("Non-empty N4W1 tag.")
+
+        progress_obj = progress(description=TR.n4w1__writing)
+        progress_obj.start()
+        progress_obj.report(100)
+        try:
+            await ctx.write(key="mnemonic", value=blob)
+            # TODO: animate during I/O?
+            progress_obj.report(1000)
+        finally:
+            progress_obj.stop()
+            progress_obj = None
+
+    async def _choose_method() -> BackupMethod:
+        import trezorui_api
+        from trezor.enums import BackupMethod
+
+        index = await interact(
+            trezorui_api.select_word(
+                title=TR.backup__title_create_wallet_backup,
+                description="Select the type of wallet backup you want to create.",
+                words=("N4W1 backup", "Wordlist backup", ""),
+            ),
+            br_name="backup_choose",
+        )
+        return (BackupMethod.N4W1, BackupMethod.Display)[index]