fix(backup): never restore a toggle whose credential can't come with it (#2656)

A settings restore refuses to write anything credential-shaped, but wrote the
switches that depend on those credentials like any other key. Restoring the two
halves apart is not a partial restore, it is a downgrade.

The sharp case is Prometheus. /api/v1/metrics is on PUBLIC_API_ROUTES and its
only gate is `if token:`, so an empty or absent token means no authentication at
all. prometheus_token matches the `token` hint and is refused; prometheus_enabled
is an ordinary key and was written. On an instance that never enabled Prometheus
there is no local token row, so overwrite-*off* alone was enough to publish the
whole metrics body to anyone who could reach the port. The new integration test
shows exactly that: 200 with a full unauthenticated body before, 404 after.

Four more pairs are the same shape and break an integration rather than open one:
ldap_enabled/ldap_bind_password, mqtt_enabled/mqtt_password, ha_enabled/ha_token
(with an HA_TOKEN env arm, since get_homeassistant_settings prefers the
environment over the row), and virtual_printer_enabled/virtual_printer_access_code
— the last largely vestigial post-migration, included for consistency.

A toggle is refused only when all five hold: the payload value is truthy, the
backup carried a non-empty companion credential, that credential is denylisted,
this instance has no usable value for it, and the toggle is not already on
locally. The second condition is what keeps the rule honest — an anonymous MQTT
broker and an anonymous LDAP bind are legitimate configs that pass empty
credentials straight through, and without it both would be false positives. With
it, the rule fires only when the restore would produce a config weaker than both
the backup and the local instance. A present-but-blank prometheus_token row
counts as unusable, since that is precisely the `if token:` hole.

The rule needs the payload *and* local database state, which the old static
_count_items could not see, so preview and restore now share one classifier:
_plan_settings() runs a single SELECT over both halves of every candidate pair
before anything enters the session, and returns the three refusal buckets.
preview() takes the session the route already has. _is_skipped_setting_key is
gone rather than having its docstring corrected as asked: a name is no longer
enough to decide, so the union predicate had no caller left.

Also implements the review's third ruling — the tally counts what the preview
counted, and refusals live in the notes. Two `skipped += 1` increments are
dropped (blocked, protected) and the companion refusal adds none; the value-is-
None and overwrite-off skips stay, because they depend on the run's flags, which
the preview cannot see. restored + skipped + failed now equals the item count the
user was shown — off by three before.

Behaviour change called out for review: test_credential_keys_are_never_restored
and test_auth_settings_are_never_restored asserted skipped == 2 and 4; both are
now 0, which is the point of the ruling.

16 new unit tests plus 2 integration tests. Nine of them are controls, because
over-refusal is the real risk of this change — the anonymous-broker and
anonymous-bind guards are load-bearing, not decoration.

backend/app/api/routes/github_backup.py

@@ -432,7 +432,7 @@ async def preview_restore(
     if not config:
         raise HTTPException(status_code=404, detail="No configuration found. Configure backup first.")
 
-    preview = await github_restore_service.preview(config, ref=ref)
+    preview = await github_restore_service.preview(db, config, ref=ref)
     return GitHubRestorePreview(**preview)
 
 

backend/app/services/github_restore.py

@@ -31,7 +31,9 @@ Design notes worth knowing before editing:
 import asyncio
 import json
 import logging
+import os
 import re
+from dataclasses import dataclass
 from datetime import datetime, timezone
 
 import httpx
@@ -135,9 +137,93 @@ def _is_protected_setting_key(key: str) -> bool:
     return key in _PROTECTED_SETTING_KEYS
 
 
-def _is_skipped_setting_key(key: str) -> bool:
-    """True for any key ``_restore_settings`` refuses to write, for either reason."""
-    return _is_blocked_setting_key(key) or _is_protected_setting_key(key)
+# There used to be an ``_is_skipped_setting_key`` here, the union of the two
+# predicates above, shared by the preview and the restore so neither could drift
+# from the other. It is gone because a name is no longer enough to decide: the
+# third refusal below depends on the payload's *other* values and on local
+# database state. ``_plan_settings`` is the shared classifier now, and it covers
+# all three reasons.
+
+
+# Toggles whose *safety* depends on a companion credential that the blocklist
+# above refuses to restore. Writing the toggle alone is not a partial restore,
+# it is a downgrade:
+#
+#   * prometheus_enabled with no token opens /api/v1/metrics. The route is on
+#     PUBLIC_API_ROUTES and its own gate is ``if token:`` (api/routes/metrics.py),
+#     so an empty or absent token means no authentication at all — a full,
+#     unauthenticated dump of the instance to anyone who can reach the port. On
+#     an instance that never enabled Prometheus there is no token row, so
+#     overwrite-off alone is enough to do it.
+#   * the other four switch an integration on with no way to authenticate to it,
+#     which breaks the login path (LDAP) or the connection (MQTT, HA).
+#
+# virtual_printer_enabled is largely vestigial post-migration — core/database.py
+# copies the rows into the virtual_printers table — but it is the same shape, and
+# refusing a vestigial toggle is a harmless no-op.
+_COMPANION_CREDENTIALS = {
+    "prometheus_enabled": "prometheus_token",
+    "ldap_enabled": "ldap_bind_password",
+    "mqtt_enabled": "mqtt_password",
+    "ha_enabled": "ha_token",
+    "virtual_printer_enabled": "virtual_printer_access_code",
+}
+
+# Companion credentials a reader takes from the environment rather than from a
+# Settings row. ha_token is the only one: get_homeassistant_settings prefers
+# HA_TOKEN over the row, and auto-enables ha_enabled when HA_URL and HA_TOKEN are
+# both set, so an env-configured instance has a usable credential and no row.
+_COMPANION_CREDENTIAL_ENV = {"ha_token": "HA_TOKEN"}
+
+
+def _setting_value_is_true(value: object) -> bool:
+    """True if a settings *payload* value would be stored as "on".
+
+    Deliberately as narrow as ``api.routes.settings.setting_is_true``: a restore
+    writes ``str(value)`` verbatim and no reader in the codebase treats "1",
+    "on" or "yes" as on, so restoring one of those cannot switch anything on.
+    Bool-tolerant because a backup's JSON can carry a real boolean.
+    """
+    if isinstance(value, bool):
+        return value
+    if value is None:
+        return False
+    return str(value).strip().lower() == "true"
+
+
+def _is_usable_credential(value: object) -> bool:
+    """True if a credential value is present and not blank.
+
+    A present-but-*blank* ``prometheus_token`` row counts as unusable, because an
+    empty token is exactly the ``if token:`` hole the companion rule exists to
+    stop a restore from opening.
+    """
+    return value is not None and bool(str(value).strip())
+
+
+@dataclass(frozen=True)
+class _SettingsPlan:
+    """Which keys of a settings payload will not be written, and why.
+
+    Built once, before anything is added to the session, and shared by the
+    preview and the restore so the two cannot disagree about what a commit will
+    change. The companion bucket is why this needs a session at all: unlike the
+    two name-based buckets it depends on local database state.
+
+    The three buckets are disjoint — a key is classified once, in order.
+    """
+
+    blocked: tuple[str, ...] = ()
+    protected: tuple[str, ...] = ()
+    companion: tuple[str, ...] = ()
+
+    @property
+    def refused(self) -> frozenset[str]:
+        return frozenset(self.blocked) | frozenset(self.protected) | frozenset(self.companion)
+
+    @property
+    def refused_count(self) -> int:
+        return len(self.blocked) + len(self.protected) + len(self.companion)
 
 
 class _CategoryTally:
@@ -239,8 +325,88 @@ class GitHubRestoreService:
                 bad.append(path)
         return parsed, bad
 
-    async def preview(self, config: GitHubBackupConfig, ref: str = "HEAD") -> dict:
-        """Report which categories a commit contains, and how much is in each."""
+    @staticmethod
+    async def _plan_settings(db: AsyncSession, values: dict) -> _SettingsPlan:
+        """Classify every key of a settings payload into its refusal bucket.
+
+        Keys with an unusable name land in no bucket: they are the restore's
+        ``failed``, not a refusal, and the preview counts them because the run
+        will still report on them.
+
+        Reads local state, so it must run before anything is added to the
+        session — otherwise "does this instance already have a credential" would
+        see the restore's own writes.
+        """
+        blocked: list[str] = []
+        protected: list[str] = []
+        # Toggle -> credential for the pairs that survived the payload-only
+        # conditions and still need local state to judge.
+        candidates: dict[str, str] = {}
+
+        for key, value in values.items():
+            if not isinstance(key, str) or not key:
+                continue
+            if _is_blocked_setting_key(key):
+                blocked.append(key)
+                continue
+            if _is_protected_setting_key(key):
+                protected.append(key)
+                continue
+
+            credential = _COMPANION_CREDENTIALS.get(key)
+            if credential is None:
+                continue
+            # Turning something *off* is always safe to write.
+            if not _setting_value_is_true(value):
+                continue
+            # Expressed as the predicate rather than assumed, so the map cannot
+            # go quietly inert if _SECRET_KEY_HINTS is ever edited: a credential
+            # the restore is willing to write travels with its toggle.
+            if not _is_blocked_setting_key(credential):
+                continue
+            # The backup itself carried no credential here. An anonymous MQTT
+            # broker and an anonymous LDAP bind are both legitimate configs
+            # (mqtt_relay.py and ldap_service.py pass empty credentials straight
+            # through), so refusing this toggle would be a false positive — the
+            # restore is not producing anything weaker than the backup.
+            if not _is_usable_credential(values.get(credential)):
+                continue
+            candidates[key] = credential
+
+        if not candidates:
+            return _SettingsPlan(blocked=tuple(blocked), protected=tuple(protected))
+
+        # One SELECT covering both halves of every candidate pair.
+        wanted = set(candidates) | set(candidates.values())
+        rows = await db.execute(select(Settings).where(Settings.key.in_(wanted)))
+        local = {row.key: row.value for row in rows.scalars().all()}
+
+        companion: list[str] = []
+        for toggle, credential in candidates.items():
+            if _is_usable_credential(local.get(credential)):
+                continue
+            env_name = _COMPANION_CREDENTIAL_ENV.get(credential)
+            if env_name and _is_usable_credential(os.environ.get(env_name)):
+                continue
+            # Already on locally with no credential: the exposure pre-dates this
+            # restore, so refusing changes nothing and "left switched off" would
+            # be a lie.
+            if _setting_value_is_true(local.get(toggle)):
+                continue
+            companion.append(toggle)
+
+        return _SettingsPlan(
+            blocked=tuple(blocked),
+            protected=tuple(protected),
+            companion=tuple(companion),
+        )
+
+    async def preview(self, db: AsyncSession, config: GitHubBackupConfig, ref: str = "HEAD") -> dict:
+        """Report which categories a commit contains, and how much is in each.
+
+        Takes a session because the settings count depends on local state — see
+        ``_plan_settings``. ``ref`` stays keyword-friendly for callers.
+        """
         resolved, error = await self._resolve_ref(config, ref)
         if resolved is None:
             return {"success": False, "message": error, "ref": ref, "categories": []}
@@ -298,7 +464,7 @@ class GitHubRestoreService:
                     }
                 )
                 continue
-            count, detail = self._count_items(category, parsed)
+            count, detail = await self._count_items(db, category, parsed)
             categories.append({"category": category, "available": True, "item_count": count, "detail": detail})
 
         commit_info = None
@@ -317,22 +483,28 @@ class GitHubRestoreService:
             "categories": categories,
         }
 
-    @staticmethod
-    def _count_items(category: RestoreCategory, parsed: dict) -> tuple[int, str | None]:
+    async def _count_items(self, db: AsyncSession, category: RestoreCategory, parsed: dict) -> tuple[int, str | None]:
         """Count restorable items for ``category`` and describe any caveat."""
         if category == RestoreCategory.SETTINGS:
             payload = parsed.get(SETTINGS_PATH)
             values = payload.get("settings") if isinstance(payload, dict) else None
             if not isinstance(values, dict):
                 return 0, "No settings in payload"
-            # Both refusals are counted the same way here so the preview's item
-            # count matches what the restore actually writes; the wording only
-            # calls out the credential ones, which are what a user might expect
-            # to come back.
-            blocked = sum(1 for key in values if _is_blocked_setting_key(key))
-            skipped = sum(1 for key in values if _is_skipped_setting_key(key))
-            detail = f"{blocked} credential-like keys will be skipped" if blocked else None
-            return len(values) - skipped, detail
+            # Every refusal is subtracted so the count matches what the restore
+            # actually writes. The wording calls out the credential ones (what a
+            # user might expect to come back) and the companion ones (a
+            # behaviour change worth explaining before it happens); the auth
+            # policy keys stay unmentioned on purpose.
+            plan = await self._plan_settings(db, values)
+            detail = None
+            if plan.companion:
+                detail = (
+                    f"{len(plan.blocked)} credential-like key(s) will be skipped, and "
+                    f"{len(plan.companion)} switch(es) that depend on them will be left off"
+                )
+            elif plan.blocked:
+                detail = f"{len(plan.blocked)} credential-like keys will be skipped"
+            return len(values) - plan.refused_count, detail
 
         if category == RestoreCategory.SPOOLS:
             payload = parsed.get(SPOOLS_PATH)
@@ -938,19 +1110,23 @@ class GitHubRestoreService:
             tally.note("No settings data in this backup")
             return
 
-        blocked = 0
-        protected = 0
+        # Planned before the first write, so the companion rule reads genuinely
+        # pre-restore local state, and so the preview and this run classify the
+        # payload identically.
+        plan = await self._plan_settings(db, values)
+        refused = plan.refused
+
         for key, value in values.items():
             if not isinstance(key, str) or not key:
                 tally.failed += 1
                 continue
-            if _is_blocked_setting_key(key):
-                blocked += 1
-                tally.skipped += 1
-                continue
-            if _is_protected_setting_key(key):
-                protected += 1
-                tally.skipped += 1
+            if key in refused:
+                # Refusals are reported in the notes and nowhere else. They are
+                # already outside the preview's item count, and the preview is
+                # the number the user was shown, so counting them here would
+                # make restored + skipped + failed exceed it. The two skips
+                # below stay counted because they depend on this run's flags,
+                # which the preview cannot see.
                 continue
             if value is None:
                 tally.skipped += 1
@@ -973,13 +1149,19 @@ class GitHubRestoreService:
             if keys_written is not None:
                 keys_written.add(key)
 
-        if blocked:
-            tally.note(f"{blocked} credential-like key(s) skipped — re-enter secrets manually")
-        if protected:
+        if plan.blocked:
+            tally.note(f"{len(plan.blocked)} credential-like key(s) skipped — re-enter secrets manually")
+        if plan.protected:
             tally.note(
-                f"{protected} authentication setting(s) skipped — change those in Settings > "
+                f"{len(plan.protected)} authentication setting(s) skipped — change those in Settings > "
                 "Authentication so the lockout checks still run"
             )
+        if plan.companion:
+            tally.note(
+                f"{', '.join(sorted(plan.companion))} left switched off — the credential each one needs "
+                "cannot be restored from a backup and this instance has none stored, so switching them "
+                "on would leave the integration unauthenticated"
+            )
 
     async def _reconfigure_mqtt_relay(self, db: AsyncSession, keys_written: set[str], tally: _CategoryTally) -> None:
         """Push restored mqtt_* settings into the live relay.

backend/tests/integration/test_github_restore_api.py

@@ -274,3 +274,66 @@ class TestStatusExposesRestoreState:
         response = await async_client.get("/api/v1/github-backup/status")
         assert response.status_code == 200
         assert response.json()["restore_running"] is False
+
+
+class TestRestoreDoesNotOpenTheMetricsEndpoint:
+    """The companion-credential rule, proved against the endpoint it protects.
+
+    ``/api/v1/metrics`` is on ``PUBLIC_API_ROUTES`` and its only gate is
+    ``if token:``, so writing ``prometheus_enabled`` onto an instance with no
+    ``prometheus_token`` row hands the entire metrics body to anyone who can
+    reach the port. The restore refuses that token as credential-shaped, so
+    before this change the pair came apart and the endpoint opened — with
+    overwrite *off*, since the local row is missing rather than present.
+
+    Driven through the real service and the real endpoint against one database:
+    the unit tests can show the toggle is not written, only this can show what
+    that means.
+    """
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_restoring_prometheus_enabled_leaves_the_endpoint_shut(self, async_client: AsyncClient, db_session):
+        from backend.app.services.github_restore import _CategoryTally, github_restore_service
+
+        # An instance that never enabled Prometheus: no toggle row, no token row.
+        assert (await async_client.get("/api/v1/metrics")).status_code == 404
+
+        tally = _CategoryTally()
+        await github_restore_service._restore_settings(
+            db_session,
+            {"settings": {"prometheus_enabled": "true", "prometheus_token": "s3cret", "currency": "EUR"}},
+            overwrite=False,
+            tally=tally,
+        )
+        await db_session.commit()
+
+        response = await async_client.get("/api/v1/metrics")
+        assert response.status_code == 404, "a settings restore opened the metrics endpoint"
+        assert "bambuddy_build_info" not in response.text
+        assert any("switched off" in note for note in tally.notes)
+
+    @pytest.mark.asyncio
+    @pytest.mark.integration
+    async def test_an_instance_with_its_own_token_still_gets_the_toggle_back(
+        self, async_client: AsyncClient, db_session
+    ):
+        """Control. The rule must not break a legitimate Prometheus restore."""
+        from backend.app.services.github_restore import _CategoryTally, github_restore_service
+
+        await async_client.put(
+            "/api/v1/settings/", json={"prometheus_enabled": False, "prometheus_token": "local-token"}
+        )
+
+        await github_restore_service._restore_settings(
+            db_session,
+            {"settings": {"prometheus_enabled": "true", "prometheus_token": "s3cret"}},
+            overwrite=True,
+            tally=_CategoryTally(),
+        )
+        await db_session.commit()
+
+        assert (await async_client.get("/api/v1/metrics")).status_code == 401
+        authorised = await async_client.get("/api/v1/metrics", headers={"Authorization": "Bearer local-token"})
+        assert authorised.status_code == 200
+        assert "bambuddy_build_info" in authorised.text

backend/tests/unit/test_github_restore.py

@@ -19,6 +19,8 @@ from backend.app.models.spool import Spool
 from backend.app.models.spool_usage_history import SpoolUsageHistory
 from backend.app.schemas.github_backup import GitHubRestoreRequest, RestoreCategory
 from backend.app.services.github_restore import (
+    _COMPANION_CREDENTIAL_ENV,
+    _COMPANION_CREDENTIALS,
     ARCHIVES_PATH,
     SETTINGS_PATH,
     SPOOL_USAGE_PATH,
@@ -27,8 +29,10 @@ from backend.app.services.github_restore import (
     _CategoryTally,
     _is_blocked_setting_key,
     _is_protected_setting_key,
-    _is_skipped_setting_key,
+    _is_usable_credential,
     _parse_dt,
+    _setting_value_is_true,
+    _SettingsPlan,
 )
 
 
@@ -81,7 +85,6 @@ class TestSettingKeyBlocklist:
         # Not credential-shaped, so the secret hints never catch them.
         assert _is_blocked_setting_key(key) is False
         assert _is_protected_setting_key(key) is True
-        assert _is_skipped_setting_key(key) is True
 
     @pytest.mark.parametrize("key", ["currency", "ldap_enabled", "auth_secret_key"])
     def test_protected_set_is_only_the_auth_policy_keys(self, key):
@@ -177,7 +180,10 @@ class TestRestoreSettings:
 
         keys = {s.key for s in (await db_session.execute(select(Settings))).scalars().all()}
         assert keys == {"currency"}
-        assert tally.skipped == 2
+        # Refusals are notes, not tally rows: the preview never counted these
+        # keys, so counting them here would put the total above what the user
+        # was shown before they pressed Restore.
+        assert tally.skipped == 0
         assert any("credential-like" in note for note in tally.notes)
 
     @pytest.mark.asyncio
@@ -207,7 +213,9 @@ class TestRestoreSettings:
         assert "setup_completed" not in rows
         assert rows["currency"] == "EUR"
         assert tally.restored == 1
-        assert tally.skipped == 4
+        # As above: refused keys are outside the preview's count, so outside the
+        # tally too.
+        assert tally.skipped == 0
         assert any("authentication setting" in note for note in tally.notes)
 
     @pytest.mark.asyncio
@@ -218,6 +226,229 @@ class TestRestoreSettings:
         assert tally.notes
 
 
+class TestSettingValueIsTrue:
+    """Only the spellings a reader actually treats as "on" count as on."""
+
+    @pytest.mark.parametrize("value", ["true", "TRUE", " True ", True])
+    def test_on(self, value):
+        assert _setting_value_is_true(value) is True
+
+    @pytest.mark.parametrize("value", ["false", "1", "on", "yes", "", None, False, 0])
+    def test_off(self, value):
+        # "1"/"on"/"yes" are deliberately off: no reader in the codebase treats
+        # them as on, so restoring one cannot switch anything on either.
+        assert _setting_value_is_true(value) is False
+
+
+class TestUsableCredential:
+    @pytest.mark.parametrize("value", ["s3cret", " x "])
+    def test_present_values_are_usable(self, value):
+        assert _is_usable_credential(value) is True
+
+    @pytest.mark.parametrize("value", [None, "", "   "])
+    def test_absent_or_blank_is_not(self, value):
+        # A present-but-blank prometheus_token row is exactly the `if token:`
+        # hole in the metrics route, so it must not count as protection.
+        assert _is_usable_credential(value) is False
+
+
+class TestCompanionCredentials:
+    """Toggles whose safety depends on a credential the restore refuses to write.
+
+    ``prometheus_enabled`` is the sharp one. ``/api/v1/metrics`` is a public
+    route whose only gate is a non-empty ``prometheus_token``, so restoring the
+    toggle onto an instance that has no token row publishes the entire metrics
+    body to anyone who can reach the port — and with overwrite *off*, since the
+    row is missing rather than present. The other four break an integration
+    rather than open one, but they are the same shape.
+    """
+
+    async def _restore(self, db, tally=None, overwrite=False, **settings) -> _CategoryTally:
+        tally = tally or _CategoryTally()
+        await _service()._restore_settings(db, {"settings": settings}, overwrite=overwrite, tally=tally)
+        await db.commit()
+        return tally
+
+    async def _rows(self, db) -> dict:
+        return {s.key: s.value for s in (await db.execute(select(Settings))).scalars().all()}
+
+    # --- The refusal itself ------------------------------------------------
+
+    @pytest.mark.asyncio
+    async def test_prometheus_toggle_is_refused_when_its_token_was_skipped(self, db_session):
+        """The headline case: overwrite off, empty database, endpoint stays shut."""
+        tally = await self._restore(db_session, currency="EUR", prometheus_enabled="true", prometheus_token="s3cret")
+
+        rows = await self._rows(db_session)
+        assert rows == {"currency": "EUR"}
+        assert any("prometheus_enabled" in note and "switched off" in note for note in tally.notes)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("toggle,credential", sorted(_COMPANION_CREDENTIALS.items()))
+    async def test_every_pair_refuses_its_toggle(self, db_session, toggle, credential, monkeypatch):
+        monkeypatch.delenv("HA_TOKEN", raising=False)
+        await self._restore(db_session, **{toggle: "true", credential: "s3cret"})
+        assert toggle not in await self._rows(db_session)
+
+    @pytest.mark.asyncio
+    async def test_ha_toggle_is_refused_when_the_environment_has_no_token(self, db_session, monkeypatch):
+        monkeypatch.delenv("HA_TOKEN", raising=False)
+        await self._restore(db_session, ha_enabled="true", ha_token="s3cret", ha_url="http://ha.local")
+
+        rows = await self._rows(db_session)
+        assert "ha_enabled" not in rows
+        assert rows["ha_url"] == "http://ha.local"
+
+    @pytest.mark.asyncio
+    async def test_a_blank_local_credential_row_is_not_usable(self, db_session):
+        db_session.add(Settings(key="prometheus_token", value=""))
+        await db_session.commit()
+
+        await self._restore(db_session, prometheus_enabled="true", prometheus_token="s3cret")
+
+        assert "prometheus_enabled" not in await self._rows(db_session)
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("value", ["TRUE", " True ", True])
+    async def test_true_is_refused_however_it_is_spelled(self, db_session, value):
+        await self._restore(db_session, prometheus_enabled=value, prometheus_token="s3cret")
+        assert "prometheus_enabled" not in await self._rows(db_session)
+
+    # --- Ruling 3: the tally counts what the preview counted ---------------
+
+    @pytest.mark.asyncio
+    async def test_refusals_are_not_counted_in_the_tally(self, db_session):
+        tally = await self._restore(db_session, currency="EUR", prometheus_enabled="true", prometheus_token="s3cret")
+        assert (tally.restored, tally.skipped, tally.failed) == (1, 0, 0)
+
+    @pytest.mark.asyncio
+    async def test_tally_total_equals_the_preview_item_count(self, db_session):
+        """The ruling, encoded: the user is shown a number, and it has to hold.
+
+        Off by three before this change — the two name-based refusals and the
+        companion one were all counted as ``skipped`` despite never being in the
+        preview's count.
+        """
+        db_session.add(Settings(key="theme", value="light"))
+        await db_session.commit()
+
+        values = {
+            "currency": "EUR",  # inserted    -> restored
+            "theme": "dark",  # exists, overwrite off -> skipped
+            "low_stock_threshold": None,  # no value    -> skipped
+            "": "junk",  # unusable key -> failed
+            "bambu_cloud_token": "x",  # blocked     -> refused
+            "auth_enabled": "false",  # protected   -> refused
+            "prometheus_enabled": "true",  # companion   -> refused
+            "prometheus_token": "s3cret",  # blocked     -> refused
+        }
+        item_count, _ = await _service()._count_items(
+            db_session, RestoreCategory.SETTINGS, {SETTINGS_PATH: {"settings": values}}
+        )
+
+        tally = _CategoryTally()
+        await _service()._restore_settings(db_session, {"settings": values}, overwrite=False, tally=tally)
+        await db_session.commit()
+
+        assert tally.restored + tally.skipped + tally.failed == item_count
+        assert (tally.restored, tally.skipped, tally.failed) == (1, 2, 1)
+
+    @pytest.mark.asyncio
+    async def test_preview_count_drops_by_one_when_the_local_credential_is_missing(self, db_session):
+        parsed = {
+            SETTINGS_PATH: {"settings": {"currency": "EUR", "prometheus_enabled": "true", "prometheus_token": "s3cret"}}
+        }
+
+        refused_count, refused_detail = await _service()._count_items(db_session, RestoreCategory.SETTINGS, parsed)
+
+        db_session.add(Settings(key="prometheus_token", value="already-set"))
+        await db_session.commit()
+        allowed_count, allowed_detail = await _service()._count_items(db_session, RestoreCategory.SETTINGS, parsed)
+
+        assert refused_count == allowed_count - 1
+        assert "switch(es)" in refused_detail
+        assert "switch(es)" not in allowed_detail
+
+    # --- Controls: over-refusal is the real risk here ----------------------
+
+    @pytest.mark.asyncio
+    async def test_a_usable_local_credential_lets_the_toggle_through(self, db_session):
+        db_session.add(Settings(key="prometheus_token", value="already-set"))
+        await db_session.commit()
+
+        tally = await self._restore(db_session, prometheus_enabled="true", prometheus_token="s3cret")
+
+        assert (await self._rows(db_session))["prometheus_enabled"] == "true"
+        assert not any("switched off" in note for note in tally.notes)
+
+    @pytest.mark.asyncio
+    async def test_an_anonymous_broker_is_not_a_false_positive(self, db_session):
+        """mqtt_relay passes an empty password straight through — a real config."""
+        tally = await self._restore(db_session, mqtt_enabled="true", mqtt_broker="10.0.0.5")
+
+        assert (await self._rows(db_session))["mqtt_enabled"] == "true"
+        assert not any("switched off" in note for note in tally.notes)
+
+    @pytest.mark.asyncio
+    async def test_an_anonymous_ldap_bind_is_not_a_false_positive(self, db_session):
+        """Same for a backup that carries the key with a blank value."""
+        tally = await self._restore(db_session, ldap_enabled="true", ldap_bind_password="   ")
+
+        assert (await self._rows(db_session))["ldap_enabled"] == "true"
+        assert not any("switched off" in note for note in tally.notes)
+
+    @pytest.mark.asyncio
+    async def test_turning_a_toggle_off_is_always_written(self, db_session):
+        await self._restore(db_session, prometheus_enabled="false", prometheus_token="s3cret")
+        assert (await self._rows(db_session))["prometheus_enabled"] == "false"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("value", ["1", "on", "yes"])
+    async def test_spellings_no_reader_treats_as_on_are_written(self, db_session, value):
+        await self._restore(db_session, prometheus_enabled=value, prometheus_token="s3cret")
+        assert (await self._rows(db_session))["prometheus_enabled"] == value
+
+    @pytest.mark.asyncio
+    async def test_ha_token_in_the_environment_counts_as_usable(self, db_session, monkeypatch):
+        monkeypatch.setenv("HA_TOKEN", "from-env")
+        await self._restore(db_session, ha_enabled="true", ha_token="s3cret")
+        assert (await self._rows(db_session))["ha_enabled"] == "true"
+
+    @pytest.mark.asyncio
+    async def test_a_toggle_already_on_locally_is_written(self, db_session):
+        """The exposure pre-dates the restore, so "left switched off" would be a lie."""
+        db_session.add(Settings(key="prometheus_enabled", value="true"))
+        await db_session.commit()
+
+        tally = await self._restore(db_session, overwrite=True, prometheus_enabled="true", prometheus_token="s3cret")
+
+        assert (await self._rows(db_session))["prometheus_enabled"] == "true"
+        assert not any("switched off" in note for note in tally.notes)
+
+    # --- The map itself ----------------------------------------------------
+
+    def test_every_companion_credential_is_blocked_and_no_toggle_is(self):
+        """Guards the rule against a future edit to _SECRET_KEY_HINTS.
+
+        If a credential stopped being blocked, its toggle would travel with it
+        and the refusal would be pointless; if a toggle started being blocked,
+        the pair would never be reached at all.
+        """
+        for toggle, credential in _COMPANION_CREDENTIALS.items():
+            assert _is_blocked_setting_key(credential) is True, credential
+            assert _is_blocked_setting_key(toggle) is False, toggle
+            assert _is_protected_setting_key(toggle) is False, toggle
+
+    def test_every_environment_override_names_a_companion_credential(self):
+        assert set(_COMPANION_CREDENTIAL_ENV) <= set(_COMPANION_CREDENTIALS.values())
+
+    @pytest.mark.asyncio
+    async def test_plan_leaves_unusable_key_names_in_no_bucket(self, db_session):
+        """They are the restore's ``failed``, not a refusal."""
+        plan = await _service()._plan_settings(db_session, {"": "x", 7: "y", "currency": "EUR"})
+        assert plan == _SettingsPlan()
+
+
 class TestRestoreSpools:
     def _spool_entry(self, **overrides):
         entry = {
@@ -1226,6 +1457,27 @@ class TestMqttRelayReconfigure:
 
         assert written == set()
 
+    @pytest.mark.asyncio
+    async def test_a_refused_mqtt_enabled_is_not_reported_as_written(self, db_session):
+        """So the relay reconfigures from the *local* mqtt_enabled, not the backup's.
+
+        The companion rule refuses ``mqtt_enabled`` when the backup's password
+        cannot come across and there is none stored locally. It must not then
+        appear in ``keys_written``, or _reconfigure_mqtt_relay would be asked to
+        bring up a broker connection the restore deliberately declined to enable.
+        """
+        written: set[str] = set()
+
+        await _service()._restore_settings(
+            db_session,
+            {"settings": {"mqtt_enabled": "true", "mqtt_password": "refused", "mqtt_broker": "new.local"}},
+            overwrite=True,
+            tally=_CategoryTally(),
+            keys_written=written,
+        )
+
+        assert written == {"mqtt_broker"}
+
 
 class TestApplyOrdering:
     """_apply must not hold SQLite's write transaction across the MQTT phase."""