feat(fw): answer /f/c rrcheck with a locally-forged access grant (#120)

src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py

@@ -67,6 +67,8 @@
 from .routes.api.v5.product import match as _match_get_products_v5
 from .routes.bootstrap.catchall import build as _build_catchall
 from .routes.bootstrap.catchall import match as _match_catchall
+from .routes.fw.check import build as _build_fw_check
+from .routes.fw.check import match as _match_fw_check
 from .routes.bootstrap.location import build as _build_location
 from .routes.bootstrap.location import match as _match_location
 from .routes.bootstrap.nc_prepare import build as _build_nc_prepare
@@ -598,6 +600,7 @@ def default_endpoint_rules() -> Sequence[EndpointRule]:
         EndpointRule("nc_prepare", _match_nc_prepare, _build_nc_prepare),
         EndpointRule("time", _match_time, _build_time),
         EndpointRule("location", _match_location, _build_location),
+        EndpointRule("fw_check", _match_fw_check, _build_fw_check),
         EndpointRule("catchall", _match_catchall, _build_catchall),
     )
 

src/roborock_local_server/bundled_backend/https_server/routes/fw/check.py

@@ -0,0 +1,111 @@
+"""Local answer for the firmware ``POST /f/c`` "rrcheck" access renewal.
+
+Background (issue #120): the vacuum periodically calls ``/f/c`` to renew a local
+access grant. When the server had no route for it, the request fell through to the
+generic catchall, which returns ``{"ok": true, ...}`` -- not a valid grant. The
+firmware scores every check as failed, tolerates the failures for a ~48h grace
+period, then flips its RPC ``AccessControl`` state to denied. From that point every
+command comes back as ``-10002 / rrcheck access denied`` while MQTT stays connected.
+Re-onboarding only resets the grace window, so control returns for another ~48h.
+
+This route forges an *accepted* grant locally, so the robot never has to reach the
+Roborock cloud. The response contract is reconstructed from the S7-family firmware:
+
+    { "k": base64(RSA-OAEP-SHA1(device_pubkey, aes128_key)),
+      "d": base64(AES-128-ECB(pkcs7({"f": "t", "c": <country>, "t": <unix>}))) }
+
+``f == "t"`` is the accepted flag; ``t`` must be within +/-300s of the device clock
+(the robot syncs its clock from our ``/time/now``). The device unwraps ``k`` with the
+same RSA private key it uses for ``/region``, so the public key the server already
+recovered during onboarding is exactly what we need -- no Roborock key is involved.
+
+Note: the ``{k, d}`` / ``{f, c, t}`` shape and the absence of a response signature
+are verified only on S7 firmware. If a newer model (e.g. a298) expects a different
+shape or checks a signature, this response is simply treated as another failed check
+-- identical to the previous catchall behaviour -- so enabling it cannot regress a
+working device. See the issue for the one-capture validation that would confirm the
+a298 contract. If we have no recovered key for the device we fall back to the
+catchall so behaviour is unchanged.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import secrets
+import time
+from typing import Any
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives.padding import PKCS7
+
+from shared.context import ServerContext
+
+from ..bootstrap.catchall import build as _build_catchall
+
+# Hosts differ per region/model but the firmware always posts to this path.
+_MATCH_PATHS = {"/f/c", "/api/f/c", "/b/f/c", "/api/b/f/c"}
+
+# Inner flag the firmware requires to mark the grant accepted.
+_ACCEPT_FLAG = "t"
+
+# The firmware decrypts the inner JSON into a 128-byte stack buffer; keep the
+# plaintext well under that. "c" is copied but does not affect acceptance.
+_MAX_INNER_BYTES = 120
+
+
+def match(path: str, _method: str = "POST") -> bool:
+    return path.rstrip("/") in _MATCH_PATHS
+
+
+def build_fw_check_payload(pubkey: Any, country: str, now: int) -> dict[str, str]:
+    """Build the ``{k, d}`` acceptance envelope for one device public key.
+
+    Raises ``ValueError`` if the inner JSON would exceed the firmware buffer.
+    """
+    inner = json.dumps(
+        {"f": _ACCEPT_FLAG, "c": country, "t": int(now)},
+        ensure_ascii=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+    if len(inner) >= _MAX_INNER_BYTES:
+        raise ValueError("rrcheck inner payload too large for firmware buffer")
+
+    aes_key = secrets.token_bytes(16)
+    padder = PKCS7(algorithms.AES.block_size).padder()
+    padded = padder.update(inner) + padder.finalize()
+    encryptor = Cipher(algorithms.AES(aes_key), modes.ECB()).encryptor()
+    ciphertext = encryptor.update(padded) + encryptor.finalize()
+
+    wrapped_key = pubkey.encrypt(
+        aes_key,
+        padding.OAEP(
+            mgf=padding.MGF1(algorithm=hashes.SHA1()),
+            algorithm=hashes.SHA1(),
+            label=None,
+        ),
+    )
+    return {
+        "k": base64.b64encode(wrapped_key).decode("ascii"),
+        "d": base64.b64encode(ciphertext).decode("ascii"),
+    }
+
+
+def build(
+    ctx: ServerContext,
+    query_params: dict[str, list[str]],
+    body_params: dict[str, list[str]],
+    clean_path: str,
+) -> dict[str, Any]:
+    did = ctx.extract_did(query_params, body_params)
+    pubkey = ctx.device_public_key(did) if did else None
+    if pubkey is None:
+        # No recovered key -> we cannot forge a valid grant; keep prior behaviour.
+        return _build_catchall(ctx, query_params, body_params, clean_path)
+    country = (getattr(ctx, "region", "") or "us").upper()
+    try:
+        return build_fw_check_payload(pubkey, country, int(time.time()))
+    except Exception:
+        return _build_catchall(ctx, query_params, body_params, clean_path)

src/roborock_local_server/bundled_backend/shared/bootstrap_crypto.py

@@ -82,6 +82,9 @@ def _load(self) -> None:
     def known_dids(self) -> list[str]:
         return sorted(self._pubkeys.keys())
 
+    def get_pubkey(self, did: str) -> Any | None:
+        return self._pubkeys.get(did)
+
     def encrypt_for_did(self, did: str, payload: dict[str, Any] | list[Any] | str) -> str | None:
         pub = self._pubkeys.get(did)
         if pub is None:

src/roborock_local_server/bundled_backend/shared/context.py

@@ -168,6 +168,22 @@ def known_bootstrap_dids(self) -> list[str]:
     def device_key_cache(self) -> DeviceKeyCache | None:
         return self._device_key_cache
 
+    def device_public_key(self, did: str) -> Any | None:
+        """Return the device's recovered RSA public key, or None if unknown.
+
+        This is the same key used to encrypt the /region bootstrap response, so a
+        device that onboarded through this server always has one available.
+        """
+        if not did:
+            return None
+        if self._device_key_cache is not None:
+            pub = self._device_key_cache.get_pubkey(did)
+            if pub is not None:
+                return pub
+        if self._bootstrap_encryptor is not None:
+            return self._bootstrap_encryptor.get_pubkey(did)
+        return None
+
     def resolve_device_localkey(
         self,
         *,

tests/test_fw_check.py

@@ -0,0 +1,128 @@
+"""Tests for the local ``/f/c`` (rrcheck) access-grant route -- issue #120.
+
+The route forges an accepted grant that the vacuum decrypts with the same RSA
+private key it uses for ``/region``. Here we play the device: build a keypair,
+register the public modulus the way onboarding recovery does, drive the request
+through ``resolve_route``, then decrypt the response and assert it is accepted.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.primitives.padding import PKCS7
+
+# Importing the backend bridge puts the bundled_backend dir on sys.path so the
+# bare ``shared`` / ``https_server`` imports inside the route modules resolve.
+import roborock_local_server.backend as backend  # noqa: E402
+from shared.context import ServerContext  # noqa: E402
+
+
+def _make_context(tmp_path, *, did: str, region: str = "eu", with_key: bool = True):
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    key_state = tmp_path / "device_keys.json"
+    if with_key:
+        modulus = key.public_key().public_numbers().n
+        key_state.write_text(
+            json.dumps({"devices": {did: {"modulus_hex": format(modulus, "x")}}}),
+            encoding="utf-8",
+        )
+    else:
+        key_state.write_text(json.dumps({"devices": {}}), encoding="utf-8")
+
+    ctx = ServerContext(
+        api_host="api.example.com",
+        mqtt_host="mqtt.example.com",
+        wood_host="wood.example.com",
+        region=region,
+        protocol_login_email="user@example.com",
+        localkey="0123456789abcdef",
+        duid=did,
+        mqtt_usr="mqtt_usr",
+        mqtt_passwd="mqtt_passwd",
+        mqtt_clientid="mqtt_clientid",
+        https_port=443,
+        mqtt_tls_port=8883,
+        http_jsonl=tmp_path / "http.jsonl",
+        mqtt_jsonl=tmp_path / "mqtt.jsonl",
+        loggers={"real_stack": logging.getLogger("test.real_stack")},
+        key_state_file=key_state,
+    )
+    return ctx, key
+
+
+def _decrypt_grant(private_key, payload: dict) -> dict:
+    aes_key = private_key.decrypt(
+        base64.b64decode(payload["k"]),
+        padding.OAEP(
+            mgf=padding.MGF1(algorithm=hashes.SHA1()),
+            algorithm=hashes.SHA1(),
+            label=None,
+        ),
+    )
+    decryptor = Cipher(algorithms.AES(aes_key), modes.ECB()).decryptor()
+    padded = decryptor.update(base64.b64decode(payload["d"])) + decryptor.finalize()
+    unpadder = PKCS7(algorithms.AES.block_size).unpadder()
+    plaintext = unpadder.update(padded) + unpadder.finalize()
+    return json.loads(plaintext)
+
+
+def test_fc_returns_decryptable_accepted_grant(tmp_path):
+    did = "1234567890"
+    ctx, key = _make_context(tmp_path, did=did, region="eu")
+
+    route_name, payload = backend.resolve_route(
+        rules=backend.default_endpoint_rules(),
+        context=ctx,
+        clean_path="/f/c",
+        query_params={},
+        body_params={"did": [did]},
+        method="POST",
+    )
+
+    assert route_name == "fw_check"
+    assert set(payload) >= {"k", "d"}
+    inner = _decrypt_grant(key, payload)
+    assert inner["f"] == "t"  # accepted flag
+    assert inner["c"] == "EU"
+    assert isinstance(inner["t"], int)
+
+
+def test_fc_matches_common_path_variants(tmp_path):
+    did = "42"
+    ctx, key = _make_context(tmp_path, did=did)
+    for path in ("/f/c", "/api/f/c", "/b/f/c", "/f/c/"):
+        route_name, payload = backend.resolve_route(
+            rules=backend.default_endpoint_rules(),
+            context=ctx,
+            clean_path=path,
+            query_params={},
+            body_params={"did": [did]},
+            method="POST",
+        )
+        assert route_name == "fw_check", path
+        assert _decrypt_grant(key, payload)["f"] == "t"
+
+
+def test_fc_falls_back_to_catchall_without_key(tmp_path):
+    did = "99"
+    ctx, _key = _make_context(tmp_path, did=did, with_key=False)
+
+    route_name, payload = backend.resolve_route(
+        rules=backend.default_endpoint_rules(),
+        context=ctx,
+        clean_path="/f/c",
+        query_params={},
+        body_params={"did": [did]},
+        method="POST",
+    )
+
+    # Route still matches, but with no device key it must not emit a bogus grant;
+    # it falls back to the previous catchall response shape instead.
+    assert route_name == "fw_check"
+    assert "k" not in payload and "d" not in payload