Offer verified OTA source contact import with explicit approval

docs/lora_ota_automation.md

@@ -213,8 +213,10 @@ public key, or unique hexadecimal key prefix. The runner reads the controller's
 existing contact table and binds each selection to a full key before remote
 commands, so emoji names need not be typed and later name changes cannot
 redirect the run. Duplicate names, ambiguous prefixes, or the same radio under
-different participant aliases stop with an actionable error. Missing contacts
-must first be imported or discovered; a key alone does not create a contact.
+different participant aliases stop with an actionable error. Missing destination
+and relay contacts must first be imported or discovered. A missing connected
+source can be imported with the confirmation prompt below; a key alone does not
+create a contact.
 
 For a separate source with a managed USB/TCP console, automatic source selection
 reads the physical source's full public key and matches it in the controller's
@@ -236,9 +238,34 @@ the runner does not guess a key from the node name or skip the identity check.
 Source selection does not require the source's current name to match an
 old saved advert, and duplicate/emoji names cannot redirect this lookup. An
 explicit `--source-contact` must identify that same physical source. If its key
-really is missing, import the source contact or advertise/discover it on the
-normal channel first; the normal/TempRadio on-air ACK checks still require the
-contact. The script does not silently create contacts or skip those checks.
+is missing, an interactive run offers:
+
+```text
+Add the connected OTA source to this controller's contacts? [y/N]
+```
+
+This works through both the shell and PowerShell launchers. The runner reads the
+source's existing `card` command over USB/TCP, checks the full key in its contact
+card against the connected source, and shows its name/key before asking. On
+approval it rechecks the source/controller identities and live contact table,
+then uses MeshCore's existing `import_contact` API. Signature validation remains
+with the controller's normal advert importer. For manual-add controllers it can
+approve only that exact source through `add_pending`; it does not enable general
+auto-add. A fresh `reload_contacts` must confirm the full source key before OTA
+continues. No LoRa advertisement or firmware update is required for this import.
+
+The default answer is **No**; `--yes` does not approve adding a contact. A
+noninteractive run stops with instructions to rerun in a terminal. Unsupported
+`card` firmware, malformed/mismatched cards, and unverified imports stop with
+manual-import/discovery guidance. To protect existing contacts, the controller's
+overwrite-oldest policy must be verifiably disabled; otherwise the runner stops
+and shows how to disable that bit without changing other auto-add flags. It does
+not change this policy itself. A full table or rejected import cannot be worked
+around by deleting another contact; an uncertain import is not automatically
+retried. An approved contact remains on the controller for subsequent runs.
+
+The normal/TempRadio on-air ACK checks still run; adding a contact does not prove
+radio reachability or skip those checks.
 When the controller and source are the same TCP Full Companion, use the verified
 `--source-shares-controller` topology; it does not need a contact for itself.
 

tools/lora_ota/lora_ota.py

@@ -2520,6 +2520,146 @@ def contact_matches_selector(contact: dict, selector: str) -> bool:
     )
 
 
+def parse_source_contact_card(output: str, expected_key: str) -> tuple[str, dict]:
+    """Check the identity in a pathless self-card; firmware verifies its signature.
+
+    This is only an envelope/key check, not a replacement for MeshCore's advert
+    importer. Never construct a contact or route from a bare public key.
+    """
+    cards = re.findall(r"(?m)^[ \t]*(?:>[ \t]*)?(meshcore://[0-9A-Fa-f]+)[ \t]*$",
+                       output.replace("\r", ""))
+    if len(cards) != 1 or output.count("meshcore://") != 1:
+        raise OtaError("OTA source did not return one exact contact card")
+    try:
+        packet = bytes.fromhex(cards[0][11:])
+    except ValueError as exc:
+        raise OtaError("OTA source returned malformed contact-card hex") from exc
+    # Packet::writeTo(): v1 flood ADVERT, no path, key + timestamp + signature
+    # + app data. `card` creates this envelope without sending a radio packet.
+    if not 104 <= len(packet) <= 134 or packet[:2] != b"\x11\x00":
+        raise OtaError("OTA source card is not a supported pathless self-advert")
+    key = packet[2:34].hex()
+    if key != expected_key.lower():
+        raise OtaError(f"OTA source contact card key mismatch: expected {expected_key}, got {key}")
+    flags = packet[102]
+    name_offset = 103 + (8 if flags & 0x10 else 0) + (2 if flags & 0x20 else 0) + (2 if flags & 0x40 else 0)
+    name = packet[name_offset:].rstrip(b"\x00")
+    if (flags & 0x0f) not in (1, 2, 3, 4) or not flags & 0x80 or not name or b"\x00" in name:
+        raise OtaError("OTA source contact card has no supported named station")
+    return "meshcore://" + packet.hex(), {
+        "public_key": key, "adv_name": name.decode("utf-8", "replace"),
+    }
+
+
+def read_import_contact_table(controller: Controller, *, pending: bool = False) -> dict[str, dict]:
+    command = "pending_contacts" if pending else "reload_contacts"
+    objects = controller._run([command], f"verify OTA source {command}", timeout=15)
+    # reload_contacts explicitly refreshes the device table; `contacts` alone
+    # can return meshcli's cache in the persistent connection.
+    if len(objects) != 1 or not all(
+        isinstance(key, str) and re.fullmatch(r"[0-9A-Fa-f]{64}", key)
+        and isinstance(value, dict) and isinstance(value.get("public_key"), str)
+        and value["public_key"].lower() == key.lower()
+        for key, value in objects[0].items()
+    ):
+        raise OtaError(f"controller did not return a complete {command} table")
+    return {key.lower(): value for key, value in objects[0].items()}
+
+
+def require_contact_overwrite_disabled(controller: Controller) -> None:
+    # meshcli prints this getter as hex even with -j. Do not enable auto-add or
+    # change the user's policy. Refuse if insertion could evict another contact.
+    result = controller._execute(["get", "autoadd_config"], "check contact overwrite policy", timeout=15)
+    values = re.findall(r"(?m)^[ \t]*0x([0-9A-Fa-f]{2})[ \t]*$", result.stdout.replace("\r", ""))
+    if len(values) != 1 or re.search(r"(?im)^\s*(?:error|err|fail|can't)\b", result.stdout + "\n" + result.stderr):
+        raise OtaError("cannot verify that controller contact overwrite is disabled; import manually")
+    config = int(values[0], 16)
+    if config & 1:
+        raise OtaError(
+            "controller contact overwrite is enabled; automatic import refused to protect existing contacts. "
+            f"Disable overwrite first (`set autoadd_config 0x{config & ~1:02x}` in meshcli), "
+            "or import manually. No controller settings were changed"
+        )
+
+
+def execute_contact_import_command(controller: Controller, commands: list[str]) -> None:
+    # import_contact can succeed with no JSON output. Send each mutation once;
+    # never retry an uncertain acknowledgement or treat it as completed storage.
+    result = controller._execute(commands, f"OTA source {commands[0]}", timeout=15)
+    output = "\n".join((result.stdout, result.stderr))
+    if re.search(r"(?im)^\s*(?:error|err|fail|unknown command|can't)\b", output) or any(
+        "error" in obj or "error_code" in obj for obj in json_objects(output)
+    ):
+        raise OtaError(f"controller rejected OTA source {commands[0]}: {output.strip()}")
+
+
+def offer_source_contact_import(
+    controller: Controller, args: argparse.Namespace, source_key: str,
+    contacts: dict[str, dict],
+) -> dict[str, dict]:
+    missing = (
+        f"the connected OTA source public key {source_key} is not in the controller's contacts; "
+        "import its contact or advertise/discover it on the normal channel first. "
+        "If source and controller are the same TCP Full Companion, use the verified "
+        "--source-shares-controller topology instead"
+    )
+    if not sys.stdin.isatty():
+        raise OtaError(missing + ". Rerun interactively for the contact-import prompt; --yes does not approve it")
+    original_controller_key = controller.get_public_key().lower()
+    if original_controller_key == source_key:
+        raise OtaError("source is the controller itself; use --source-shares-controller instead of importing itself")
+    output = optional_source_cli_command(args, "card")
+    if output is None:
+        raise OtaError(missing + ". This source terminal does not support `card`; no import was attempted")
+    uri, card_contact = parse_source_contact_card(output, source_key)
+    selector = args.source_contact_value
+    if selector and not contact_matches_selector(card_contact, selector):
+        raise OtaError("--source-contact does not identify the connected source's contact card; no import attempted")
+    print(f"[contact] Missing OTA source: {card_contact['adv_name']!r} [{source_key}]")
+    print("[contact] Import its contact card over USB/TCP using MeshCore. No LoRa advertisement will be sent.")
+    print("[contact] This adds a saved contact; it does not change auto-add settings or remove other contacts.")
+    try:
+        approved = input("Add the connected OTA source to this controller's contacts? [y/N] ").strip().lower() in ("y", "yes")
+    except EOFError:
+        approved = False
+    if not approved:
+        raise OtaError(missing + ". Contact import declined; no contact changes were made")
+    # The user may have been at the prompt for a while. Recheck both identities
+    # and the live table before adding anything. Never overwrite a newly added
+    # source contact or import a card from a swapped source/controller.
+    controller_key = controller.get_public_key().lower()
+    if controller_key != original_controller_key or read_source_public_key_bounded(args).lower() != source_key:
+        raise OtaError("OTA source/controller identity changed before contact import; stopping")
+    current = read_import_contact_table(controller)
+    if not contacts.keys() <= current.keys():
+        raise OtaError("controller contacts changed during the import prompt; rerun before adding a contact")
+    if source_key in current:
+        return current
+    require_contact_overwrite_disabled(controller)
+    execute_contact_import_command(controller, ["import_contact", uri])
+    added_pending = False
+    for attempt in range(5):
+        current_after = read_import_contact_table(controller)
+        if not current.keys() <= current_after.keys():
+            raise OtaError("controller contact inventory lost an existing key during import; stopping for inspection")
+        if source_key in current_after:
+            print(f"[contact] Imported and verified OTA source [{source_key}]")
+            return current_after
+        # Manual-add controllers expose a signature-checked advert as pending.
+        # Add ONLY the exact approved key through the existing MeshCore API.
+        if not added_pending and source_key in read_import_contact_table(controller, pending=True):
+            require_contact_overwrite_disabled(controller)
+            execute_contact_import_command(controller, ["add_pending", source_key])
+            added_pending = True
+        if attempt < 4:
+            time.sleep(0.25)
+    raise OtaError(
+        "OTA source contact import was not confirmed in the controller's refreshed table. "
+        "Its table may be full, storage unavailable, or the card rejected. No contacts were deleted "
+        "by this script; check the controller and rerun. The import will not be retried automatically"
+    )
+
+
 def bind_contact_selectors(controller: Controller, args: argparse.Namespace) -> None:
     """Resolve the local contact table once, before any remote command is sent.
 
@@ -2569,23 +2709,24 @@ def bind_contact_selectors(controller: Controller, args: argparse.Namespace) ->
         # differ from the controller's saved advert (renames, truncation, emoji).
         # Never bind a same-named but different radio, even with --source-contact.
         source_key = read_source_public_key_bounded(args).lower()
+        selected = None
+        if source_selector and any(contact_matches_selector(value, source_selector) for value in contacts.values()):
+            selected = resolve(source_selector)
+            if selected != source_key:
+                raise OtaError(
+                    f"--source-contact identifies {selected}, but the connected OTA source "
+                    f"reports {source_key}; select the contact for the physical source"
+                )
+        if source_key not in contacts:
+            contacts = offer_source_contact_import(controller, args, source_key, contacts)
         if source_selector:
-            source = resolve(source_selector)
+            source = selected or resolve(source_selector)
             if source != source_key:
                 raise OtaError(
                     f"--source-contact identifies {source}, but the connected OTA source "
                     f"reports {source_key}; select the contact for the physical source"
                 )
         else:
-            if source_key not in contacts:
-                raise OtaError(
-                    f"the connected OTA source public key {source_key} is not in the "
-                    "controller's contacts; import its contact or advertise/discover it "
-                    "on the normal channel first. Renaming a different contact or passing "
-                    "--source-contact cannot replace this identity check. If source and "
-                    "controller are the same TCP Full Companion, use the verified "
-                    "--source-shares-controller topology instead"
-                )
             source = source_key
             print(f"[contact] OTA source -> {contacts[source].get('adv_name', '?')} [{source}]")
     else:
@@ -2928,7 +3069,8 @@ def source_cli_command(
                 raise TransmissionError("source TCP console returned an empty reply")
         else:
             wire_command = command_text
-            if full_companion_identity or getattr(args, "source_companion_terminal", False):
+            if (full_companion_identity or getattr(args, "source_full_companion", False)
+                    or getattr(args, "source_companion_terminal", False)):
                 # meshcli raw mode keeps one serial open while writing this
                 # compound command. STOP first makes this independent of the
                 # port's current state: ASCII consumes it and returns to
@@ -6970,7 +7112,8 @@ def build_parser() -> argparse.ArgumentParser:
         metavar="NAME_OR_KEY",
         help=(
             "controller contact name/key/unique key prefix for a separate OTA source, used for the "
-            "three-minute on-air proof; defaults to the source's local name "
+            "three-minute on-air proof; defaults to the connected source's full public key, "
+            "with a separately approved contact-card import if missing "
             "(no remote-admin password is required)"
         ),
     )
@@ -7100,7 +7243,7 @@ def build_parser() -> argparse.ArgumentParser:
             "used when a delta does not expose its reconstructed target hash"
         ),
     )
-    parser.add_argument("--yes", action="store_true", help="skip the destructive-action confirmation")
+    parser.add_argument("--yes", action="store_true", help="skip the OTA destructive-action confirmation (not contact import or host-tool installation)")
     parser.add_argument(
         "--require-system-watchdog-off",
         action="store_true",

tools/lora_ota/test_lora_ota_bootloader.py

@@ -462,6 +462,7 @@ class StationSelectionTests(unittest.TestCase):
         }
         with (
             mock.patch.object(ota, "read_source_public_key_bounded", return_value=source_key),
+            mock.patch.object(ota.sys.stdin, "isatty", return_value=False),
             contextlib.redirect_stdout(io.StringIO()),
             self.assertRaisesRegex(ota.OtaError, "connected OTA source public key.*not in") as error,
         ):

tools/lora_ota/test_lora_ota_contact_import.py

@@ -0,0 +1,295 @@
+"""Offline source-card import tests. No radios, flashes, or LoRa transmissions."""
+
+import argparse
+import contextlib
+import io
+import subprocess
+import unittest
+from unittest import mock
+
+import lora_ota as ota
+from test_lora_ota_source_identity import banner, connection
+
+
+SOURCE = "a5" * 32
+TARGET = "b6" * 32
+CONTROLLER = "c7" * 32
+NAME = "Source \U0001f4e1"
+
+
+def card(key=SOURCE, name=NAME, flags=0x81, extra=b""):
+    # Same Packet::writeTo envelope as firmware `card`. Signature bytes are
+    # dummy: production signature verification remains inside the controller.
+    packet = b"\x11\x00" + bytes.fromhex(key) + (123).to_bytes(4, "little") + b"s" * 64
+    return "meshcore://" + (packet + bytes([flags]) + extra + name.encode()).hex()
+
+
+def entry(key, name):
+    return {"public_key": key, "adv_name": name}
+
+
+def result(stdout="", stderr=""):
+    return subprocess.CompletedProcess([], 0, stdout, stderr)
+
+
+class SourceCardTests(unittest.TestCase):
+    def test_accepts_pathless_card_with_unicode_name_and_optional_fields(self):
+        for flags, extra in ((0x81, b""), (0xf2, b"x" * 12)):
+            uri = card(flags=flags, extra=extra)
+            output = banner() + "\r\n> " + uri + "\r\n> \r\nBinary mode\r\n"
+            self.assertEqual(ota.parse_source_contact_card(output, SOURCE), (uri, entry(SOURCE, NAME)))
+
+    def test_uppercase_hex_is_normalized(self):
+        uri = card()
+        self.assertEqual(ota.parse_source_contact_card("meshcore://" + uri[11:].upper(), SOURCE.upper())[0], uri)
+
+    def test_rejects_missing_duplicate_embedded_malformed_or_wrong_key(self):
+        uri = card()
+        values = ["OK", uri + "\n" + uri, "Received " + uri, uri + "z", uri[:-1],
+                  card(TARGET), "meshcore://", uri + "\nmeshcore://broken"]
+        for value in values:
+            with self.subTest(value=value), self.assertRaises(ota.OtaError):
+                ota.parse_source_contact_card(value, SOURCE)
+
+    def test_rejects_unsupported_envelopes_flags_and_name(self):
+        raw = bytes.fromhex(card()[11:])
+        values = [raw[:50], b"\x51" + raw[1:], b"\x12" + raw[1:],
+                  b"\x11\x01" + raw[2:], raw + b"x" * 40]
+        for packet in values:
+            with self.subTest(packet=packet), self.assertRaises(ota.OtaError):
+                ota.parse_source_contact_card("meshcore://" + packet.hex(), SOURCE)
+        for uri in (card(name=""), card(flags=0x01), card(flags=0x80), card(flags=0x91, name="x"), card(name="bad\x00name")):
+            with self.subTest(uri=uri), self.assertRaises(ota.OtaError):
+                ota.parse_source_contact_card(uri, SOURCE)
+
+    def test_full_tcp_card_uses_existing_read_only_terminal_command(self):
+        args = argparse.Namespace(source_cli_tcp="192.0.2.1:5002", source_cli_serial=None,
+                                  source_serial=None, source_full_companion=True)
+        conn = connection(reply=card())
+        with mock.patch.object(ota.socket, "create_connection", return_value=conn), contextlib.redirect_stdout(io.StringIO()):
+            output = ota.optional_source_cli_command(args, "card")
+        self.assertEqual(ota.parse_source_contact_card(output, SOURCE)[0], card())
+        conn.sendall.assert_called_once_with(b"card\r\n")
+
+    def test_full_usb_card_wraps_terminal_even_after_identity_left_binary_mode(self):
+        args = argparse.Namespace(source_cli_tcp=None, source_cli_serial="COM42", source_serial=None,
+                                  source_baud=115200, meshcli="meshcli", source_full_companion=True,
+                                  source_companion_terminal=False)
+        with mock.patch.object(ota, "run_checked", return_value=result(banner() + card())) as run, contextlib.redirect_stdout(io.StringIO()):
+            output = ota.optional_source_cli_command(args, "card")
+        self.assertEqual(ota.parse_source_contact_card(output, SOURCE)[0], card())
+        self.assertEqual(run.call_args.args[0][-1],
+                         f"{ota.COMPANION_TERMINAL_STOP}\r{ota.COMPANION_TERMINAL_START}\rcard\r{ota.COMPANION_TERMINAL_STOP}\r")
+
+
+class ContactImportTests(unittest.TestCase):
+    def setUp(self):
+        self.args = argparse.Namespace(target="Target", relay_values=[], source_contact_value=None,
+                                       source_shares_controller=False, source_cli_tcp="192.0.2.1:5002", yes=True)
+        self.before = {TARGET: entry(TARGET, "Target")}
+        self.after = {**self.before, SOURCE: entry(SOURCE, NAME)}
+        self.controller = mock.Mock()
+        self.controller.get_public_key.return_value = CONTROLLER
+        self.controller._run.side_effect = [[self.before], [self.before], [self.after]]
+        self.controller._execute.side_effect = [result("0x1e\n"), result()]
+        stack = contextlib.ExitStack()
+        self.addCleanup(stack.close)
+        stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
+        self.tty = stack.enter_context(mock.patch.object(ota.sys.stdin, "isatty", return_value=True))
+        self.answer = stack.enter_context(mock.patch("builtins.input", return_value="yes"))
+        self.read_key = stack.enter_context(mock.patch.object(ota, "read_source_public_key_bounded", return_value=SOURCE))
+        self.read_card = stack.enter_context(mock.patch.object(ota, "optional_source_cli_command", return_value=card()))
+        stack.enter_context(mock.patch.object(ota.time, "sleep"))
+
+    def bind(self):
+        ota.bind_contact_selectors(self.controller, self.args)
+
+    def test_approved_import_verified_then_continue_with_full_key(self):
+        self.bind()
+        self.assertEqual(self.args.source_contact_value, SOURCE)
+        self.assertEqual(self.args.target, TARGET)
+        self.answer.assert_called_once()
+        self.assertEqual(self.read_key.call_count, 2)
+        self.assertEqual([call.args[0] for call in self.controller._execute.call_args_list],
+                         [["get", "autoadd_config"], ["import_contact", card()]])
+        self.assertEqual([call.args[0] for call in self.controller._run.call_args_list],
+                         [["contacts"], ["reload_contacts"], ["reload_contacts"]])
+        self.controller.remote_command.assert_not_called()
+        self.controller.set_radio.assert_not_called()
+
+    def test_explicit_source_name_or_key_can_be_imported(self):
+        for selector in (NAME, SOURCE, SOURCE[:12]):
+            with self.subTest(selector=selector):
+                self.args.target = "Target"
+                self.controller._run.side_effect = [[self.before], [self.before], [self.after]]
+                self.controller._execute.side_effect = [result("0x1e\r\n"), result()]
+                self.args.source_contact_value = selector
+                self.bind()
+                self.assertEqual(self.args.source_contact_value, SOURCE)
+
+    def test_existing_contact_has_no_prompt_or_card_read(self):
+        self.controller._run.side_effect = [[self.after]]
+        self.bind()
+        self.answer.assert_not_called()
+        self.read_card.assert_not_called()
+        self.controller._execute.assert_not_called()
+
+    def test_declined_blank_eof_and_non_yes_do_not_mutate_even_with_yes_flag(self):
+        for answer in ("", "n", "no", "sure", EOFError()):
+            with self.subTest(answer=answer):
+                self.controller._run.side_effect = [[self.before]]
+                self.answer.side_effect = answer if isinstance(answer, Exception) else None
+                self.answer.return_value = answer
+                with self.assertRaisesRegex(ota.OtaError, "declined"):
+                    self.bind()
+                self.controller._execute.assert_not_called()
+
+    def test_noninteractive_gives_instructions_without_card_read_or_mutation(self):
+        self.tty.return_value = False
+        with self.assertRaisesRegex(ota.OtaError, "--yes does not approve"):
+            self.bind()
+        self.answer.assert_not_called()
+        self.read_card.assert_not_called()
+        self.controller._execute.assert_not_called()
+
+    def test_keyboard_interrupt_is_not_consent(self):
+        self.answer.side_effect = KeyboardInterrupt
+        with self.assertRaises(KeyboardInterrupt):
+            self.bind()
+        self.controller._execute.assert_not_called()
+
+    def test_unsupported_source_card_reports_manual_fallback(self):
+        self.read_card.return_value = None
+        with self.assertRaisesRegex(ota.OtaError, "does not support `card`"):
+            self.bind()
+        self.answer.assert_not_called()
+        self.controller._execute.assert_not_called()
+
+    def test_card_timeout_does_not_become_import(self):
+        self.read_card.side_effect = ota.TransmissionError("timeout")
+        with self.assertRaises(ota.TransmissionError):
+            self.bind()
+        self.controller._execute.assert_not_called()
+
+    def test_card_key_mismatch_is_refused_before_consent(self):
+        self.read_card.return_value = card(TARGET)
+        with self.assertRaisesRegex(ota.OtaError, "key mismatch"):
+            self.bind()
+        self.answer.assert_not_called()
+        self.controller._execute.assert_not_called()
+
+    def test_wrong_explicit_selector_does_not_prompt_or_import(self):
+        self.args.source_contact_value = "Target"
+        with self.assertRaisesRegex(ota.OtaError, "--source-contact identifies"):
+            self.bind()
+        self.answer.assert_not_called()
+        self.read_card.assert_not_called()
+
+    def test_unknown_explicit_selector_must_match_card_before_prompt(self):
+        self.args.source_contact_value = "Wrong name"
+        with self.assertRaisesRegex(ota.OtaError, "does not identify"):
+            self.bind()
+        self.answer.assert_not_called()
+
+    def test_shared_source_does_not_import_self(self):
+        self.args.source_shares_controller = True
+        self.controller._run.side_effect = [[self.before]]
+        self.bind()
+        self.read_key.assert_not_called()
+        self.answer.assert_not_called()
+        self.controller._execute.assert_not_called()
+
+    def test_misconfigured_same_controller_gives_topology_option(self):
+        self.controller.get_public_key.return_value = SOURCE
+        with self.assertRaisesRegex(ota.OtaError, "--source-shares-controller"):
+            self.bind()
+        self.read_card.assert_not_called()
+        self.answer.assert_not_called()
+
+    def test_source_swap_after_consent_stops_before_import(self):
+        self.read_key.side_effect = [SOURCE, TARGET]
+        with self.assertRaisesRegex(ota.OtaError, "identity changed"):
+            self.bind()
+        self.controller._execute.assert_not_called()
+
+    def test_controller_swap_after_consent_stops_before_import(self):
+        self.controller.get_public_key.side_effect = [CONTROLLER, "d8" * 32]
+        with self.assertRaisesRegex(ota.OtaError, "identity changed"):
+            self.bind()
+        self.controller._execute.assert_not_called()
+
+    def test_contact_added_while_prompt_open_is_not_reimported(self):
+        self.controller._run.side_effect = [[self.before], [self.after]]
+        self.bind()
+        self.controller._execute.assert_not_called()
+
+    def test_removed_existing_contact_stops_before_import(self):
+        self.controller._run.side_effect = [[self.before], [{}]]
+        with self.assertRaisesRegex(ota.OtaError, "contacts changed"):
+            self.bind()
+        self.controller._execute.assert_not_called()
+
+    def test_overwrite_enabled_is_refused_without_policy_change(self):
+        self.controller._execute.side_effect = [result("0x1f\n")]
+        with self.assertRaisesRegex(ota.OtaError, "set autoadd_config 0x1e"):
+            self.bind()
+        self.assertEqual(self.controller._execute.call_count, 1)
+
+    def test_unknown_overwrite_policy_is_refused(self):
+        self.controller._execute.side_effect = [result("Can't get autoadd_config")]
+        with self.assertRaisesRegex(ota.OtaError, "cannot verify"):
+            self.bind()
+        self.assertEqual(self.controller._execute.call_count, 1)
+
+    def test_import_acknowledgement_alone_is_not_success_and_is_not_retried(self):
+        self.controller._run.side_effect = [[self.before], [self.before]] + [[self.before], [{}]] * 5
+        with self.assertRaisesRegex(ota.OtaError, "table may be full"):
+            self.bind()
+        self.assertEqual(self.controller._execute.call_count, 2)
+
+    def test_delayed_device_import_is_polled_using_fresh_tables(self):
+        self.controller._run.side_effect = [[self.before], [self.before], [self.before], [{}], [self.after]]
+        self.bind()
+        self.assertEqual(self.controller._execute.call_count, 2)
+
+    def test_manual_add_uses_only_matching_pending_key_without_changing_settings(self):
+        pending = {SOURCE: entry(SOURCE, NAME), "d8" * 32: entry("d8" * 32, NAME)}
+        self.controller._run.side_effect = [[self.before], [self.before], [self.before], [pending], [self.after]]
+        self.controller._execute.side_effect = [result("0x00"), result(), result("0x00"), result("{}")]
+        self.bind()
+        self.assertEqual([call.args[0] for call in self.controller._execute.call_args_list],
+                         [["get", "autoadd_config"], ["import_contact", card()],
+                          ["get", "autoadd_config"], ["add_pending", SOURCE]])
+
+    def test_pending_add_once_when_table_full(self):
+        pending = {SOURCE: entry(SOURCE, NAME)}
+        self.controller._run.side_effect = [[self.before], [self.before], [self.before], [pending]] + [[self.before]] * 4
+        self.controller._execute.side_effect = [result("0x00"), result(), result("0x00"), result("{}")]
+        with self.assertRaisesRegex(ota.OtaError, "table may be full"):
+            self.bind()
+        self.assertEqual(self.controller._execute.call_count, 4)
+
+    def test_import_rejection_and_lost_ack_never_retry(self):
+        for response in (result("Error while importing contact: full"), result('{"error_code": 3}'), ota.OtaError("timeout")):
+            with self.subTest(response=response):
+                self.controller._run.side_effect = [[self.before], [self.before]]
+                self.controller._execute.reset_mock()
+                self.controller._execute.side_effect = [result("0x00"), response]
+                with self.assertRaises(ota.OtaError):
+                    self.bind()
+                self.assertEqual(self.controller._execute.call_count, 2)
+
+    def test_inventory_loss_after_import_stops(self):
+        self.controller._run.side_effect = [[self.before], [self.before], [{SOURCE: entry(SOURCE, NAME)}]]
+        with self.assertRaisesRegex(ota.OtaError, "lost an existing key"):
+            self.bind()
+
+    def test_malformed_refreshed_table_is_not_success(self):
+        self.controller._run.side_effect = [[self.before], [{"error": "storage error"}]]
+        with self.assertRaisesRegex(ota.OtaError, "complete reload_contacts"):
+            self.bind()
+        self.controller._execute.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()