Add V2 region public-key recovery from SHA-384 signatures

docs/onboarding.md

@@ -144,6 +144,7 @@ Everything in "What To Expect" above still applies. Some vacuums need 2-4 cycles
 
 ## Related Docs
 
+- [Experimental V2 public-key recovery](v2_onboarding.md)
 - [Installation](installation.md)
 - [Tested vacuums](tested_vacuums.md)
 - [Home Assistant](home_assistant.md)

docs/v2_onboarding.md

@@ -0,0 +1,76 @@
+# Experimental V2 public-key recovery
+
+This branch adds automatic RSA public-key recovery for the V2 `GET /region`
+request format observed on Saros model `roborock.vacuum.a279`. It does not yet
+establish complete V2 onboarding or MQTT connectivity.
+
+## Verified request format
+
+The captured requests carry `v: v2`, nonce `n`, timestamp `t`, and signature `s`
+headers. The Base64-decoded signature is 512 bytes. The signed message is:
+
+```text
+exact_query_without_&signature_suffix + ":" + header_n + ":" + header_t
+```
+
+Signatures use RSA-4096, public exponent 65537, and PKCS#1 v1.5 with SHA-384.
+Query order, escaping, percent-escape case, and literal plus signs must remain
+unchanged. A parsed query dictionary is not a substitute for the wire query.
+
+Three saved requests from one owner's Saros recovered its public modulus.
+All 14 available signatures verified against that modulus, including the 11
+not used for recovery. Replaying those requests through the persistent cache
+also recovered the same key after a cache reload. These were offline checks;
+no vacuum or vendor service was contacted. The private captures are not included
+in this branch.
+
+This recovers the public key needed to encrypt server replies. It does not
+recover the device's private key or a symmetric secret, and the demonstrated
+method does not require a firmware dump.
+
+## Server behavior
+
+- The server records the request version with each header-signature sample.
+- Complete V2 `GET /region` or `GET /.roborock.com/region` samples select SHA-384
+  recovery. The cache starts recovery after at least two distinct samples.
+- The worker uses up to three samples for modulus recovery, then verifies every
+  sample in its snapshot before accepting the key.
+- Pending recovery resumes when the persisted cache is loaded again.
+- Legacy query-signature recovery keeps SHA-256 as its default.
+- Unversioned header samples, POST requests, `/b/region`, and HMAC-sized tags
+  are excluded from this V2 contract.
+
+Old cache entries without version metadata cannot be automatically identified
+as V2. New pairing attempts can supply versioned samples. Recovery may require
+another pairing cycle once the public key is ready.
+
+## Remaining hardware validation
+
+The existing V2 unsupported onboarding status remains in place. In particular,
+the UI still reports V2 as unsupported even after public-key recovery. This
+branch is intended for protocol investigation, not confirmed V2 support.
+
+The server still uses its existing RSA-OAEP/SHA-1 bootstrap response encryption.
+SHA-384 request signatures alone do not establish the response encryption
+format. A real V2 device must still demonstrate:
+
+1. Acceptance of the encrypted `/region` reply and the supplied server URLs.
+2. Progression to NC registration and acceptance of its device ID and local key.
+3. TLS acceptance and authenticated MQTT connectivity.
+4. Status messages, a command response, and reconnection after reboot.
+
+The next useful hardware test is a pairing attempt after successful key
+recovery, checking whether the device advances from `/region` to NC and MQTT.
+Further protocol changes and onboarding-status updates depend on that result.
+
+## Regression tests
+
+```console
+uv sync --extra dev
+uv run pytest -q tests/test_rsa_sampling.py tests/test_v2_region_recovery.py tests/test_device_key_recovery.py tests/test_admin_api.py tests/test_runtime_state.py
+```
+
+The tests use synthetic signatures and temporary state. They cover RSA-2048 /
+SHA-256 compatibility, RSA-4096 / SHA-384 recovery, exact query preservation,
+protocol classification, version persistence, restart behavior, and rejection
+of a modified signature holdout. They do not prove physical onboarding.

src/roborock_local_server/bundled_backend/shared/device_key_recovery.py

@@ -56,11 +56,42 @@ def split_signed_query(query: str) -> tuple[str, str] | None:
     return canonical, signature_b64
 
 
-def _emsa_pkcs1_v1_5_sha256(msg: str, key_bytes: int) -> int:
-    digest = hashlib.sha256(msg.encode("utf-8")).digest()
-    digest_info = bytes.fromhex("3031300d060960864801650304020105000420") + digest
-    if key_bytes < len(digest_info) + 3:
-        raise ValueError("key too small for PKCS1v1.5 SHA-256 encoding")
+def split_v2_region_sample(sample: dict[str, str]) -> tuple[str, str] | None:
+    """Select the verified RSA-4096/SHA-384 GET /region header contract.
+
+    Preserve wire query bytes: decoding or re-encoding changes the signed message.
+    POST bodies and B01 /b/region are deliberately outside this verified contract.
+    """
+    if (
+        sample.get("version") != "v2"
+        or sample.get("method") != "GET"
+        or sample.get("path") not in ("/region", "/.roborock.com/region")
+    ):
+        return None
+    signature = sample.get("signature_b64", "")
+    try:
+        if len(base64.b64decode(signature, validate=True)) != 512:
+            return None
+    except (ValueError, TypeError):
+        return None
+    query = sample.get("query", "").split("&signature=", 1)[0]
+    nonce, timestamp = sample.get("nonce", ""), sample.get("ts", "")
+    if not query or not nonce or not timestamp:
+        return None
+    return f"{query}:{nonce}:{timestamp}", signature
+
+
+def _emsa_pkcs1_v1_5(msg: str, key_bytes: int, hash_name: str) -> int:
+    prefixes = {
+        "sha256": "3031300d060960864801650304020105000420",
+        "sha384": "3041300d060960864801650304020205000430",
+    }
+    if hash_name not in prefixes:
+        raise ValueError("RSA sample recovery supports sha256 or sha384")
+    digest = hashlib.new(hash_name, msg.encode("utf-8")).digest()
+    digest_info = bytes.fromhex(prefixes[hash_name]) + digest
+    if key_bytes < len(digest_info) + 11:
+        raise ValueError("key too small for PKCS1v1.5 encoding")
     ps = b"\xff" * (key_bytes - len(digest_info) - 3)
     em = b"\x00\x01" + ps + b"\x00" + digest_info
     return int.from_bytes(em, "big")
@@ -79,18 +110,25 @@ def recover_modulus_from_samples(
     samples: list[tuple[str, str]],
     *,
     e: int = DEFAULT_RSA_E,
+    hash_name: str = "sha256",
     diagnostics: dict[str, Any] | None = None,
 ) -> int | None:
-    """Recover an RSA modulus from canonical query/signature pairs.
+    """Recover an RSA public modulus from exact signed-message/signature pairs.
 
     If ``diagnostics`` is supplied, it is populated with information about each
     stage of recovery so callers can surface a precise failure reason.
+    SHA-256 remains the default for existing onboarding. Verified v2
+    RSA-4096 requests use SHA-384; callers must explicitly select it and
+    supply the exact canonical message. HMAC tags are not RSA.
     """
 
     def _diag(key: str, value: Any) -> None:
         if diagnostics is not None:
             diagnostics[key] = value
 
+    if hash_name not in ("sha256", "sha384"):
+        raise ValueError("RSA sample recovery supports sha256 or sha384")
+    _diag("hash_name", hash_name)
     _diag("input_samples", len(samples))
     dedup: dict[str, str] = {}
     for canonical, sig_b64 in samples:
@@ -106,14 +144,20 @@ def _diag(key: str, value: Any) -> None:
     _diag("sig_byte_lengths", sig_lengths)
     key_bytes = max(sig_lengths)
     _diag("key_bytes", key_bytes)
+    if key_bytes < MIN_RSA_SIGNATURE_BYTES:
+        _diag(
+            "reason",
+            "Signatures are too short for RSA sample recovery; HMAC tags cannot use this method.",
+        )
+        return None
     xs: list[Any] = []
     verifiers: list[tuple[int, int]] = []
     for canonical, sig_b64 in pairs:
         sig_bytes = base64.b64decode(sig_b64)
         if len(sig_bytes) != key_bytes:
             continue
         sig_int = int.from_bytes(sig_bytes, "big")
-        em_int = _emsa_pkcs1_v1_5_sha256(canonical, key_bytes)
+        em_int = _emsa_pkcs1_v1_5(canonical, key_bytes, hash_name)
         x = gmpy2.mpz(sig_int) ** e - gmpy2.mpz(em_int)
         xs.append(abs(x))
         verifiers.append((sig_int, em_int))
@@ -224,10 +268,27 @@ def _recover_modulus_subprocess(
     samples: list[tuple[str, str]],
     e: int,
     conn: Any,
+    hash_name: str = "sha256",
 ) -> None:
     diag: dict[str, Any] = {}
     try:
-        modulus = recover_modulus_from_samples(samples, e=e, diagnostics=diag)
+        # Three samples normally remove small common cofactors. Verify every
+        # remaining sample with modular exponentiation instead of huge powers.
+        recovery_samples = samples[:3] if hash_name == "sha384" else samples
+        modulus = recover_modulus_from_samples(
+            recovery_samples, e=e, hash_name=hash_name, diagnostics=diag
+        )
+        if modulus and hash_name == "sha384":
+            verified = sum(
+                len(base64.b64decode(signature, validate=True)) == 512
+                and pow(int.from_bytes(base64.b64decode(signature), "big"), e, modulus)
+                == _emsa_pkcs1_v1_5(canonical, 512, hash_name)
+                for canonical, signature in samples
+            )
+            diag["verified_samples"] = verified
+            if verified != len(samples):
+                modulus = None
+                diag["reason"] = "Recovered v2 modulus did not verify every captured sample."
         conn.send((int(modulus) if modulus else None, "", "", diag))
     except Exception as exc:  # noqa: BLE001
         tb = traceback.format_exc()
@@ -379,12 +440,13 @@ def _load(self) -> None:
                         {
                             "method": str(sample.get("method", "")).strip().upper(),
                             "path": str(sample.get("path", "")).strip(),
-                            "query": str(sample.get("query", "")).strip(),
+                            "query": str(sample.get("query", "")),
                             "nonce": str(sample.get("nonce", "")).strip(),
                             "ts": str(sample.get("ts", "")).strip(),
                             "signature_b64": signature_b64,
                             "body_sha256": str(sample.get("body_sha256", "")).strip(),
                             "signature_len": str(sig_len),
+                            "version": str(sample.get("version", "")).strip(),
                         }
                     )
                 if clean_headers:
@@ -409,9 +471,10 @@ def _load(self) -> None:
     def _resume_pending_recoveries(self) -> None:
         pending: list[str] = []
         with self._lock:
-            for did, samples in self._samples.items():
+            for did in set(self._samples) | set(self._header_samples):
                 if did in self._pubkeys:
                     continue
+                samples, _hash_name, _source = self._recovery_samples_locked(did)
                 if len(samples) < 2:
                     continue
                 pending.append(did)
@@ -516,6 +579,7 @@ def add_header_signature(
         ts: str,
         signature_b64: str,
         body_sha256: str = "",
+        version: str = "",
     ) -> bool:
         sign = (signature_b64 or "").strip()
         if not did or not sign:
@@ -527,12 +591,13 @@ def add_header_signature(
         entry = {
             "method": (method or "").strip().upper(),
             "path": (path or "").strip(),
-            "query": (query or "").strip(),
+            "query": query or "",
             "nonce": (nonce or "").strip(),
             "ts": (ts or "").strip(),
             "signature_b64": sign,
             "body_sha256": (body_sha256 or "").strip(),
             "signature_len": str(sig_len),
+            "version": (version or "").strip(),
         }
         with self._lock:
             arr = self._header_samples.setdefault(did, [])
@@ -547,16 +612,26 @@ def add_header_signature(
             self._save_safe_locked()
         return True
 
+    def _recovery_samples_locked(self, did: str) -> tuple[list[tuple[str, str]], str, str]:
+        headers = [
+            pair
+            for sample in self._header_samples.get(did, [])
+            if (pair := split_v2_region_sample(sample)) is not None
+        ]
+        if headers:
+            return list(dict.fromkeys(headers)), "sha384", "v2 region header"
+        return list(self._samples.get(did, [])), "sha256", "query"
+
     def maybe_recover_async(self, did: str) -> None:
         with self._lock:
             if did in self._pubkeys:
                 changed = self._set_recovery_meta_locked(did, state="recovered", note="Public key is available.")
                 if changed:
                     self._save_safe_locked()
                 return
-            samples = list(self._samples.get(did, []))
+            samples, hash_name, sample_source = self._recovery_samples_locked(did)
             if len(samples) < 2:
-                note = f"Need at least 2 query signature samples ({len(samples)} captured)."
+                note = f"Need at least 2 {sample_source} signature samples ({len(samples)} captured)."
                 changed = self._set_recovery_meta_locked(did, state="collecting", note=note)
                 if changed:
                     self._save_safe_locked()
@@ -598,7 +673,7 @@ def maybe_recover_async(self, did: str) -> None:
             self._set_recovery_meta_locked(
                 did,
                 state="recovering",
-                note="Recovering RSA modulus from query signatures.",
+                note=f"Recovering RSA modulus from {sample_source} signatures ({hash_name}).",
                 started_at=started_at,
             )
             self._save_safe_locked()
@@ -609,7 +684,7 @@ def _worker() -> None:
                 parent_conn, child_conn = _MP_CTX.Pipe(duplex=False)
                 process = _MP_CTX.Process(
                     target=_recover_modulus_subprocess,
-                    args=(samples, DEFAULT_RSA_E, child_conn),
+                    args=(samples, DEFAULT_RSA_E, child_conn, hash_name),
                     daemon=True,
                 )
                 process.start()

src/roborock_local_server/server.py

@@ -978,6 +978,7 @@ async def _handle_roborock_request(self, request: Request) -> Response:
                         ts=ts,
                         signature_b64=sign,
                         body_sha256=body_sha256,
+                        version=region_version,
                     )
                 except Exception as exc:  # noqa: BLE001
                     logger.warning("key_cache add_header_signature failed did=%s: %s", key_capture_did, exc)

tests/test_admin_api.py

@@ -683,6 +683,7 @@ def test_region_v2_request_surfaces_unsupported_onboarding_alert(tmp_path: Path)
             "signature_b64": "QUJD",
             "body_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
             "signature_len": "3",
+            "version": "v2",
         }
     ]
 

tests/test_rsa_sampling.py

@@ -0,0 +1,53 @@
+"""Recover public keys from independent cryptography-generated signatures."""
+
+import base64
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+import pytest
+
+from roborock_local_server.bundled_backend.shared.device_key_recovery import (
+    recover_modulus_from_samples,
+)
+
+
+@pytest.mark.parametrize(
+    "key_size,hash_name,algorithm",
+    [
+        (2048, "sha256", hashes.SHA256()),
+        (4096, "sha384", hashes.SHA384()),
+    ],
+)
+def test_recovers_exact_public_key_from_samples(key_size, hash_name, algorithm):
+    key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)
+    messages = [f"synthetic-exact-wire-message-{i}" for i in range(3)]
+    samples = [
+        (
+            message,
+            base64.b64encode(
+                key.sign(message.encode(), padding.PKCS1v15(), algorithm)
+            ).decode(),
+        )
+        for message in messages
+    ]
+    diagnostics = {}
+    # Exercise the original default as well as the newly explicit SHA-384 mode.
+    options = {} if hash_name == "sha256" else {"hash_name": hash_name}
+    modulus = recover_modulus_from_samples(samples, diagnostics=diagnostics, **options)
+    assert modulus == key.public_key().public_numbers().n
+    assert diagnostics["hash_name"] == hash_name
+
+
+def test_hmac_sized_samples_are_not_treated_as_rsa():
+    samples = [
+        ("one", base64.b64encode(bytes(32)).decode()),
+        ("two", base64.b64encode(bytes([1]) * 32).decode()),
+    ]
+    diagnostics = {}
+    assert recover_modulus_from_samples(samples, diagnostics=diagnostics) is None
+    assert "HMAC" in diagnostics["reason"]
+
+
+def test_unsupported_hash_is_explicit():
+    with pytest.raises(ValueError, match="sha256 or sha384"):
+        recover_modulus_from_samples([], hash_name="md5")