diff --git a/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py b/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py index c081bdf..119d3fa 100644 --- a/src/roborock_local_server/bundled_backend/https_server/endpoint_rules.py +++ b/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), ) diff --git a/src/roborock_local_server/bundled_backend/https_server/routes/fw/__init__.py b/src/roborock_local_server/bundled_backend/https_server/routes/fw/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/roborock_local_server/bundled_backend/https_server/routes/fw/check.py b/src/roborock_local_server/bundled_backend/https_server/routes/fw/check.py new file mode 100644 index 0000000..5e1a000 --- /dev/null +++ b/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": , "t": }))) } + +``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) diff --git a/src/roborock_local_server/bundled_backend/shared/bootstrap_crypto.py b/src/roborock_local_server/bundled_backend/shared/bootstrap_crypto.py index 887561c..c0847d4 100644 --- a/src/roborock_local_server/bundled_backend/shared/bootstrap_crypto.py +++ b/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: diff --git a/src/roborock_local_server/bundled_backend/shared/context.py b/src/roborock_local_server/bundled_backend/shared/context.py index d990ff5..01e9906 100644 --- a/src/roborock_local_server/bundled_backend/shared/context.py +++ b/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, *, diff --git a/tests/test_fw_check.py b/tests/test_fw_check.py new file mode 100644 index 0000000..0618303 --- /dev/null +++ b/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