From f47d0627806abae8de1af545409a8057c8173fe1 Mon Sep 17 00:00:00 2001 From: Luke Date: Wed, 23 Sep 2026 13:45:49 -0400 Subject: [PATCH] feat(q7): local bootstrap for sc05 / B01 devices Add HMAC + AES bootstrap so Q7 (roborock.vacuum.sc05, B01) devices can onboard against the local server. The firmware signs /b/region and /b/nc with its per-device secret and expects an AES-encrypted response, so the server must hold that secret to reply in a form the firmware accepts. - b01_bootstrap: request-signature verification, AES region/NC responses, and MQTT credential derivation, from the firmware contract - b01_import: import a device's secret/DUID into state/b01_devices.json - server: route /b/region and /b/nc to the B01 handler - runtime_credentials: resolve rr/d/i/{duid}/... topics by DUID (not DID) so B01 MQTT frames decode with the correct localkey - tests + docs Validated end to end on a Q7 M5+: onboarding, region/NC, MQTT connect, and inbound B01 frame decode via python-roborock. Outbound command/map handling is the next step; the RSA-4096 (v2) request branch is out of scope and returns 501. Co-Authored-By: Claude Opus 4.8 --- docs/q7_b01.md | 64 ++++++ src/roborock_local_server/b01_bootstrap.py | 199 ++++++++++++++++++ src/roborock_local_server/b01_import.py | 60 ++++++ .../shared/runtime_credentials.py | 2 +- src/roborock_local_server/server.py | 47 +++-- tests/test_b01_bootstrap.py | 194 +++++++++++++++++ tests/test_b01_import.py | 64 ++++++ 7 files changed, 615 insertions(+), 15 deletions(-) create mode 100644 docs/q7_b01.md create mode 100644 src/roborock_local_server/b01_bootstrap.py create mode 100644 src/roborock_local_server/b01_import.py create mode 100644 tests/test_b01_bootstrap.py create mode 100644 tests/test_b01_import.py diff --git a/docs/q7_b01.md b/docs/q7_b01.md new file mode 100644 index 0000000..b356a18 --- /dev/null +++ b/docs/q7_b01.md @@ -0,0 +1,64 @@ +# Q7 (sc05 / B01) local bootstrap + +This adds local-server support for the Roborock Q7 family that speaks the **B01** +bootstrap protocol (firmware `roborock.vacuum.sc05`). Unlike the V1 flow, the Q7 +authenticates its bootstrap requests with an HMAC over the device's per-device +`secret` and expects an **AES-encrypted** response, so the server must know that +secret to answer in a form the firmware will accept. + +Validated end to end on real hardware: a Q7 M5+ paired through onboarding, +completed `/b/region` + `/b/nc`, connected over MQTT TLS, and its B01 telemetry +frames decode via python-roborock. + +## What it needs: the device bootstrap secret + +The one per-device value the server requires is the 32-character `secret` from the +robot's `device.json`. It is used both to verify the request HMAC and to encrypt +the region/NC responses (`AES-128-CBC`, key = `secret[8:24]`, IV = `md5(nonce)[12:28]`), +so it cannot be derived or substituted. + +Import it into the server's state before pairing: + +```bash +python -m roborock_local_server.b01_import \ + --device-config /path/to/rriot/config_dir/device.json \ + --iot-config /path/to/rriot/data_dir/iot.json \ + --output data/state/b01_devices.json +``` + +The importer refuses to overwrite a different secret for an existing DID and +preserves other entries. `data/` is git-ignored; secrets never enter source control. + +## The bootstrap flow + +| Stage | Request | Response | +| --- | --- | --- | +| Region | `GET /b/region` (HMAC headers `nonce`/`ts`/`sign`) | AES `{apiUrl, mqttUrl}` | +| Activation | `POST /b/nc`, `p=B01`, `scheme=1` | AES `{k: localkey, d: cloud DUID}` | + +`server.py` routes `/b/region` and `/b/nc` to `b01_bootstrap.build_response`, +which authenticates the request and returns the encrypted payload. MQTT +credentials are then derived (`b01_bootstrap.mqtt_credentials`) from the DUID, +localkey and activation `s`/`t`, and stored in the runtime credential store. + +## Onboarding + +Point the robot at the stack with the onboarding CLI; the firmware builds its +bootstrap host from the provisioning `token.r`: + +```bash +python start_onboarding.py --server api- +``` + +The robot's Wi-Fi network must resolve `api-` to the server, and the +served certificate must chain to a CA in the firmware's trust store (the sc05 +bundle includes ISRG Root X1, Actalis, USERTrust, DigiCert and GlobalSign roots, +so a normal publicly-trusted certificate for a domain you control works). + +## Status and scope + +- Bootstrap (region/NC), MQTT connect, and inbound B01 frame decode: **working**. +- Full B01 command/map/datapoint handling in Home Assistant: in progress; the + inbound decode path is wired, outbound command encoding is the next step. +- The alternate RSA-4096 (`v: v2`) request branch is out of scope here and is + explicitly rejected with HTTP 501. diff --git a/src/roborock_local_server/b01_bootstrap.py b/src/roborock_local_server/b01_bootstrap.py new file mode 100644 index 0000000..b0eef81 --- /dev/null +++ b/src/roborock_local_server/b01_bootstrap.py @@ -0,0 +1,199 @@ +"""Experimental Q7 sc05 bootstrap, recovered from firmware 03.01.74. + +This implements the HMAC request branch and AES response format. The alternate +4096-bit RSA request branch and live device compatibility are not yet verified. +Device secrets must be explicitly imported into state/b01_devices.json. +""" + +from __future__ import annotations + +import base64 +from datetime import datetime, timezone +import hashlib +import hmac +import json +from pathlib import Path +import re +from typing import Any, Mapping +from urllib.parse import parse_qs + +from cryptography.hazmat.primitives import padding +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + +def canonical_path(path: str) -> str | None: + """Called after stripping the injected /.roborock.com prefix.""" + return path if path in ("/b/region", "/b/nc") else None + + +def request_signature( + secret: str, path: str, params: bytes, nonce: str, timestamp: str +) -> str: + """Firmware 0x19e70..0x19f0e: MD5 path/params, then HMAC-SHA256.""" + material = ( + f"{nonce}:{timestamp}:{hashlib.md5(path.encode()).hexdigest()}:" + f"{hashlib.md5(params).hexdigest()}:" + ) + return base64.b64encode( + hmac.new( + secret.encode("ascii"), material.encode("ascii"), hashlib.sha256 + ).digest() + ).decode("ascii") + + +def encrypt_result(secret: str, nonce: str, payload: dict[str, Any]) -> dict[str, Any]: + """Firmware 0x1a00a/0x1a4d0: ASCII slices, not hex-decoded bytes.""" + secret_bytes = secret.encode("ascii") + if not 24 <= len(secret_bytes) <= 64: + raise ValueError("B01 device secret must contain 24 to 64 ASCII bytes") + iv = hashlib.md5(nonce.encode("ascii")).hexdigest()[12:28].encode("ascii") + plaintext = json.dumps(payload, ensure_ascii=True, separators=(",", ":")).encode( + "ascii" + ) + padder = padding.PKCS7(128).padder() + padded = padder.update(plaintext) + padder.finalize() + encryptor = Cipher(algorithms.AES(secret_bytes[8:24]), modes.CBC(iv)).encryptor() + ciphertext = encryptor.update(padded) + encryptor.finalize() + return {"code": 200, "result": base64.b64encode(ciphertext).decode("ascii")} + + +def mqtt_credentials( + duid: str, localkey: str, session: str, token: str +) -> dict[str, str]: + """Firmware 0x1a5d8..0x1a70a; distinct from the V1 MQTT derivation.""" + if not ( + 1 <= len(duid.encode("ascii")) <= 32 and len(localkey.encode("ascii")) == 16 + ): + raise ValueError( + "B01 requires a DUID of at most 32 bytes and a 16-byte local key" + ) + if not ( + 6 <= len(session.encode("ascii")) <= 64 + and 8 <= len(token.encode("ascii")) <= 64 + ): + raise ValueError("B01 activation session/token lengths are invalid") + digest = hashlib.sha256( + f"{duid}:{session[1:5]}:{localkey[3:11]}".encode("ascii") + ).hexdigest() + # The firmware copies six bytes at token+3 even for an eight-byte token. + token_slice = (token.encode("ascii") + b"\0")[3:9] + pass_digest = hashlib.sha256( + session[2:6].encode("ascii") + + b":" + + token_slice + + b":" + + localkey[9:15].encode("ascii") + ).hexdigest() + return { + "client_id": digest[11:27], + "username": digest[27:43], + "password": hashlib.md5(pass_digest[21:53].encode("ascii")).hexdigest(), + } + + +def build_response( + *, + ctx: Any, + state_file: Path, + path: str, + method: str, + query: str, + body: bytes, + headers: Mapping[str, str], +) -> tuple[str, int, dict[str, Any]]: + """Explicitly opted-in HMAC bootstrap; authenticate before storing credentials.""" + + def error(status: int, message: str) -> tuple[str, int, dict[str, Any]]: + return "b01_bootstrap_error", status, {"code": status, "msg": message} + + if path not in ("/b/region", "/b/nc"): + return error(404, "unknown_b01_bootstrap_path") + expected_method = "GET" if path == "/b/region" else "POST" + if method != expected_method: + return error(405, "b01_bootstrap_method_not_allowed") + # The firmware signs the exact query for region, and exact form body for NC. + try: + wire = query.encode("ascii") if method == "GET" else body + params = parse_qs( + wire.decode("ascii"), keep_blank_values=True, strict_parsing=True + ) + required = {"d", "m", "r", "s", "t"} + if path == "/b/nc": + required |= {"n", "p", "scheme"} + if not required.issubset(params) or any( + len(v) != 1 or not v[0] for v in params.values() + ): + return error(400, "invalid_b01_bootstrap_parameters") + did, model = params["d"][0], params["m"][0] + if model != "roborock.vacuum.sc05": + return error(400, "b01_model_not_validated") + if path == "/b/nc" and (params["p"][0] != "B01" or params["scheme"][0] != "1"): + return error(400, "unsupported_b01_nc_scheme") + state = json.loads(state_file.read_text(encoding="utf-8")) + device = state["devices"][did] + secret, duid = device["secret"], device["duid"] + if ( + device.get("model") != model + or not isinstance(secret, str) + or not isinstance(duid, str) + ): + return error(503, "invalid_b01_device_configuration") + if ( + not 24 <= len(secret.encode("ascii")) <= 64 + or not 1 <= len(duid.encode("ascii")) <= 32 + ): + return error(503, "invalid_b01_device_configuration") + except (OSError, KeyError): + return error(503, "b01_device_secret_required") + except (ValueError, TypeError, AttributeError): + return error(400, "invalid_b01_bootstrap_input") + + if headers.get("v", "").lower() == "v2": + return error(501, "b01_rsa_request_branch_not_implemented") + nonce, ts, sign = ( + headers.get("nonce", ""), + headers.get("ts", ""), + headers.get("sign", ""), + ) + if not re.fullmatch(r"[0-9a-fA-F]{16}", nonce) or not re.fullmatch( + r"[0-9]{1,10}", ts + ): + return error(401, "b01_nonce_or_timestamp_missing") + # Do not apply wall-clock freshness: a freshly reset robot may lack correct time. + expected_sign = request_signature(secret, path, wire, nonce, ts) + if not hmac.compare_digest(expected_sign.encode(), sign.encode()): + return error(401, "b01_signature_invalid") + + if path == "/b/region": + # Use configured endpoints; the unauthenticated Host header cannot redirect them. + return ( + "region", + 200, + encrypt_result( + secret, nonce, {"apiUrl": ctx.api_url(), "mqttUrl": ctx.mqtt_url()} + ), + ) + + session, token = params["s"][0], params["t"][0] + try: + # Validate tokens before resolving or mutating credentials. + mqtt_credentials(duid, "0123456789abcdef", session, token) + localkey = ctx.resolve_device_localkey( + did=did, duid=duid, model=model, source="b01_nc" + ) + mqtt = mqtt_credentials(duid, localkey, session, token) + except (ValueError, UnicodeError): + return error(400, "invalid_b01_nc_credentials") + result = encrypt_result(secret, nonce, {"k": localkey, "d": duid}) + if ctx.runtime_credentials is not None: + ctx.runtime_credentials.ensure_device( + did=did, + duid=duid, + model=model, + localkey=localkey, + local_key_source="b01_nc", + device_mqtt_usr=mqtt["username"], + device_mqtt_pass=mqtt["password"], + last_nc_at=datetime.now(timezone.utc).isoformat(), + ) + return "nc_prepare", 200, result diff --git a/src/roborock_local_server/b01_import.py b/src/roborock_local_server/b01_import.py new file mode 100644 index 0000000..23b5956 --- /dev/null +++ b/src/roborock_local_server/b01_import.py @@ -0,0 +1,60 @@ +"""Import the Q7 bootstrap secret from a local dump, without printing credentials.""" + +from __future__ import annotations + +import argparse +import json +import os +from pathlib import Path +import tempfile + + +def import_device(device_file: Path, iot_file: Path, output: Path) -> None: + device = json.loads(device_file.read_text(encoding="utf-8")) + iot = json.loads(iot_file.read_text(encoding="utf-8")) + did, model = str(device["did"]), device["model"] + secret, duid = device["secret"], iot["duid"] + if not did.isascii() or not did.isdecimal() or model != "roborock.vacuum.sc05": + raise ValueError("Expected a Q7 sc05 device with a numeric DID") + if not isinstance(secret, str) or not 24 <= len(secret.encode("ascii")) <= 64: + raise ValueError("Invalid Q7 device secret") + if not isinstance(duid, str) or not 1 <= len(duid.encode("ascii")) <= 32: + raise ValueError("Invalid Q7 cloud DUID") + state = ( + json.loads(output.read_text(encoding="utf-8")) + if output.exists() + else {"devices": {}} + ) + if not isinstance(state, dict) or not isinstance(state.get("devices"), dict): + raise ValueError("Invalid B01 device state file") + entry = {"model": model, "duid": duid, "secret": secret} + if did in state["devices"] and state["devices"][did] != entry: + raise ValueError( + "A different entry already exists for this DID; review it before replacing" + ) + state["devices"][did] = entry + output.parent.mkdir(parents=True, exist_ok=True) + fd, temp_path = tempfile.mkstemp(dir=output.parent, prefix=".b01-", suffix=".json") + try: + with os.fdopen(fd, "w", encoding="utf-8") as stream: + json.dump(state, stream, indent=2) + stream.write("\n") + os.replace(temp_path, output) + finally: + Path(temp_path).unlink(missing_ok=True) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--device-config", type=Path, required=True) + parser.add_argument("--iot-config", type=Path, required=True) + parser.add_argument("--output", type=Path, required=True) + args = parser.parse_args() + import_device(args.device_config, args.iot_config, args.output) + print( + f"Imported Q7 bootstrap configuration into {args.output}; credentials were not displayed." + ) + + +if __name__ == "__main__": + main() diff --git a/src/roborock_local_server/bundled_backend/shared/runtime_credentials.py b/src/roborock_local_server/bundled_backend/shared/runtime_credentials.py index 0977e49..bf44a3b 100644 --- a/src/roborock_local_server/bundled_backend/shared/runtime_credentials.py +++ b/src/roborock_local_server/bundled_backend/shared/runtime_credentials.py @@ -596,7 +596,7 @@ def localkey_for_topic(self, topic: str) -> str: if normalized_topic.startswith("rr/d/"): parts = normalized_topic.split("/") if len(parts) >= 5: - device = self.resolve_device(did=parts[3]) + device = self.resolve_device(duid=parts[3]) or self.resolve_device(did=parts[3]) return _clean_str(device.get("localkey")) if device else "" if normalized_topic.startswith("rr/m/"): parts = normalized_topic.split("/") diff --git a/src/roborock_local_server/server.py b/src/roborock_local_server/server.py index 5c959e3..924b04e 100644 --- a/src/roborock_local_server/server.py +++ b/src/roborock_local_server/server.py @@ -21,6 +21,7 @@ from python_multipart.exceptions import MultipartParseError import uvicorn +from .b01_bootstrap import build_response as build_b01_response, canonical_path as b01_path from .certs import CertificateManager from .bundled_backend.shared.data_helpers import utcnow_iso from .bundled_backend.shared.runtime_state import ONBOARDING_STEP_LABELS, REQUIRED_ONBOARDING_STEPS @@ -935,6 +936,7 @@ async def _handle_roborock_request(self, request: Request) -> Response: logger = self.context.loggers.get(group, self.context.loggers["unknown"]) raw_body = await request.body() clean_path = strip_roborock_prefix(request.url.path) + is_b01_bootstrap = b01_path(clean_path) is not None query_params = _request_query_params(request) body_text, body_params = await _request_body_params(request, raw_body) body_sha256 = hashlib.sha256(raw_body).hexdigest() @@ -960,7 +962,7 @@ async def _handle_roborock_request(self, request: Request) -> Response: query_sample_added = False header_sample_added = False - if key_cache is not None and key_capture_did: + if key_cache is not None and key_capture_did and not is_b01_bootstrap: if request.url.query: try: query_sample_added = key_cache.add_signed_query(key_capture_did, request.url.query) @@ -995,7 +997,7 @@ async def _handle_roborock_request(self, request: Request) -> Response: logger.warning("key_cache add_header_signature failed did=%s: %s", key_capture_did, exc) raw_path = request.url.path - if request.url.query: + if request.url.query and not is_b01_bootstrap: raw_path += f"?{request.url.query}" client_host = request.client.host if request.client else "-" client_port = request.client.port if request.client else 0 @@ -1016,11 +1018,16 @@ async def _handle_roborock_request(self, request: Request) -> Response: entry["did"] = explicit_did if explicit_pid: entry["pid"] = explicit_pid - if is_protocol_sync_request: + if is_b01_bootstrap: + # Activation tokens and nonces are unnecessary for the diagnostic log. + entry["raw_path"] = request.url.path + entry["query"] = {"d": [explicit_did], "m": [explicit_pid]} + entry["headers"] = {"host": host, "content-type": request.headers.get("content-type", "")} + if is_protocol_sync_request or is_b01_bootstrap: entry["body_redacted"] = True else: entry["body_b64"] = base64.b64encode(raw_body).decode("ascii") - if body_text and not is_protocol_sync_request: + if body_text and not is_protocol_sync_request and not is_b01_bootstrap: entry["body_text"] = body_text try: entry["body_json"] = json.loads(body_text) @@ -1292,14 +1299,26 @@ async def _handle_roborock_request(self, request: Request) -> Response: ) return response - route_name, response_payload = resolve_route( - rules=self.endpoint_rules, - context=self.context, - clean_path=clean_path, - query_params=query_params, - body_params=body_params, - method=request.method, - ) + status_code = 200 + if is_b01_bootstrap: + route_name, status_code, response_payload = build_b01_response( + ctx=self.context, + state_file=self.paths.state_dir / "b01_devices.json", + path=clean_path, + method=request.method, + query=request.url.query, + body=raw_body, + headers=request.headers, + ) + else: + route_name, response_payload = resolve_route( + rules=self.endpoint_rules, + context=self.context, + clean_path=clean_path, + query_params=query_params, + body_params=body_params, + method=request.method, + ) entry["route"] = route_name entry["response_json"] = response_payload try: @@ -1318,7 +1337,7 @@ async def _handle_roborock_request(self, request: Request) -> Response: except Exception as exc: # noqa: BLE001 logger.warning("runtime_state record_http_event failed: %s", exc) append_jsonl(self.context.http_jsonl, entry) - if key_cache is not None and key_capture_did: + if key_cache is not None and key_capture_did and not is_b01_bootstrap: try: key_cache.maybe_recover_async(key_capture_did) except Exception as exc: # noqa: BLE001 @@ -1332,7 +1351,7 @@ async def _handle_roborock_request(self, request: Request) -> Response: route_name, body_sha256[:16], ) - return JSONResponse(response_payload) + return JSONResponse(response_payload, status_code=status_code) def _status_payload(self) -> dict[str, Any]: health = self.runtime_state.health_snapshot() diff --git a/tests/test_b01_bootstrap.py b/tests/test_b01_bootstrap.py new file mode 100644 index 0000000..61c2835 --- /dev/null +++ b/tests/test_b01_bootstrap.py @@ -0,0 +1,194 @@ +"""Synthetic contract vectors checked against Q7 ARM code in Unicorn.""" + +import base64 +import json + +from Crypto.Cipher import AES +from Crypto.Util.Padding import unpad +from fastapi.testclient import TestClient +import pytest + +from conftest import write_release_config +from roborock_local_server.b01_bootstrap import ( + encrypt_result, + mqtt_credentials, + request_signature, +) +from roborock_local_server.config import load_config, resolve_paths +from roborock_local_server.server import ReleaseSupervisor + +SECRET = "0123456789abcdef0123456789abcdef" +NONCE = "0123456789abcdef" +DID = "123456789" +DUID = "synthetic-cloud-duid" +MODEL = "roborock.vacuum.sc05" +SESSION = "abcdefghijklmnop" +TOKEN = "fedcba9876543210" +REGION_QUERY = f"d={DID}&m={MODEL}&r=rr-lab.example/&s={SESSION}&t={TOKEN}" +NC_BODY = f"d={DID}&m={MODEL}&n=Synthetic&p=B01&r=rr-lab.example/&s={SESSION}&scheme=1&t={TOKEN}" + + +def headers(path, wire): + return { + "nonce": NONCE, + "ts": "123456", + "sign": request_signature(SECRET, path, wire.encode(), NONCE, "123456"), + } + + +def decrypt(response): + # Exact key and IV observed at firmware 0x1a010; independent crypto library. + cipher = AES.new(b"89abcdef01234567", AES.MODE_CBC, b"5123906e58e06714") + return json.loads( + unpad(cipher.decrypt(base64.b64decode(response.json()["result"])), 16) + ) + + +@pytest.fixture +def stack(tmp_path): + config_file = write_release_config(tmp_path, https_port=8443, mqtt_tls_port=9443) + config = load_config(config_file) + paths = resolve_paths(config_file, config) + paths.state_dir.mkdir(parents=True, exist_ok=True) + (paths.state_dir / "b01_devices.json").write_text( + json.dumps({"devices": {DID: {"secret": SECRET, "duid": DUID, "model": MODEL}}}) + ) + supervisor = ReleaseSupervisor(config=config, paths=paths) + supervisor.runtime_credentials.ensure_device( + did=DID, duid=DUID, model=MODEL, localkey="0123456789abcdef" + ) + return TestClient(supervisor.app), supervisor, paths + + +def test_signature_matches_firmware_trace(): + assert ( + request_signature(SECRET, "/b/region", REGION_QUERY.encode(), NONCE, "123456") + == "MWgBDF13JqtaXi8l024LeUjx21H6qOPJwGZ5dnxCB6M=" + ) + + +@pytest.mark.parametrize( + "token,password", + [ + (TOKEN, "01a3fc9993655075ecef7cabcf76af48"), + ("12345678", "cde0a3149b622a8c642008273eb40eed"), + ], +) +def test_mqtt_credentials_match_firmware_trace(token, password): + assert mqtt_credentials(DUID, "0123456789abcdef", SESSION, token) == { + "client_id": "4d12f0da94689dbb", + "username": "943561986905f676", + "password": password, + } + + +@pytest.mark.parametrize("prefix", ["", "/.roborock.com"]) +def test_region_uses_aes_and_configured_endpoints(stack, prefix): + client, _, _ = stack + response = client.get( + prefix + "/b/region?" + REGION_QUERY, + headers={ + **headers("/b/region", REGION_QUERY), + "host": "api-untrusted.example:1234", + }, + ) + assert response.status_code == 200 + assert decrypt(response) == { + "apiUrl": "https://api-roborock.example.com:8443", + "mqttUrl": "ssl://api-roborock.example.com:9443", + } + + +def test_nc_returns_cloud_duid_and_registers_matching_mqtt_credentials(stack): + client, supervisor, _ = stack + response = client.post( + "/.roborock.com/b/nc", + content=NC_BODY, + headers={ + **headers("/b/nc", NC_BODY), + "content-type": "application/x-www-form-urlencoded", + }, + ) + assert response.status_code == 200 + assert decrypt(response) == {"k": "0123456789abcdef", "d": DUID} + device = supervisor.runtime_credentials.resolve_device(did=DID) + assert device["device_mqtt_usr"] == "943561986905f676" + assert device["device_mqtt_pass"] == "01a3fc9993655075ecef7cabcf76af48" + + +def test_missing_secret_does_not_fall_back_to_rsa(stack): + client, _, paths = stack + (paths.state_dir / "b01_devices.json").unlink() + response = client.get( + "/b/region?" + REGION_QUERY, headers=headers("/b/region", REGION_QUERY) + ) + assert response.status_code == 503 + assert response.json()["msg"] == "b01_device_secret_required" + + +@pytest.mark.parametrize( + "change", [{"sign": "bad"}, {"nonce": "bad"}, {"ts": "bad"}, {"v": "v2"}] +) +def test_invalid_or_unimplemented_auth_is_explicit(stack, change): + client, supervisor, _ = stack + before = supervisor.runtime_credentials.devices() + response = client.post( + "/b/nc", content=NC_BODY, headers={**headers("/b/nc", NC_BODY), **change} + ) + assert response.status_code == (501 if "v" in change else 401) + assert supervisor.runtime_credentials.devices() == before + + +@pytest.mark.parametrize( + "body", + [ + NC_BODY.replace("p=B01", "p=1.0"), + NC_BODY.replace("scheme=1", "scheme=2"), + NC_BODY + "&d=other", + NC_BODY.replace("s=" + SESSION, "s=x"), + NC_BODY.replace(MODEL, "roborock.vacuum.sc01"), + ], +) +def test_invalid_nc_does_not_mutate_credentials(stack, body): + client, supervisor, _ = stack + before = supervisor.runtime_credentials.devices() + response = client.post("/b/nc", content=body, headers=headers("/b/nc", body)) + assert response.status_code == 400 + assert supervisor.runtime_credentials.devices() == before + + +def test_disabled_onboarding_still_blocks_b01(tmp_path): + config_file = write_release_config(tmp_path, new_connections_enabled=False) + config = load_config(config_file) + supervisor = ReleaseSupervisor( + config=config, paths=resolve_paths(config_file, config) + ) + response = TestClient(supervisor.app).get( + "/.roborock.com/b/region?" + REGION_QUERY, + headers=headers("/b/region", REGION_QUERY), + ) + assert response.status_code == 403 + logged = supervisor.paths.http_jsonl_path.read_text() + assert SESSION not in logged and TOKEN not in logged + assert SESSION not in json.dumps(supervisor.runtime_state.health_snapshot()) + + +def test_b01_logs_do_not_contain_activation_tokens(stack): + client, _, paths = stack + client.get("/b/region?" + REGION_QUERY, headers=headers("/b/region", REGION_QUERY)) + log = paths.http_jsonl_path.read_text() + assert SESSION not in log and TOKEN not in log and SECRET not in log + assert json.loads(log.splitlines()[-1])["body_redacted"] is True + + +def test_wrong_method_rejected(stack): + client, _, _ = stack + assert ( + client.get("/b/nc?" + NC_BODY, headers=headers("/b/nc", NC_BODY)).status_code + == 405 + ) + + +def test_encrypt_result_rejects_short_secrets(): + with pytest.raises(ValueError): + encrypt_result("short", NONCE, {}) diff --git a/tests/test_b01_import.py b/tests/test_b01_import.py new file mode 100644 index 0000000..62b8462 --- /dev/null +++ b/tests/test_b01_import.py @@ -0,0 +1,64 @@ +"""Device imports preserve unrelated entries and reject accidental replacement.""" + +import json + +import pytest + +from roborock_local_server.b01_import import import_device + + +@pytest.fixture +def files(tmp_path): + device = tmp_path / "device.json" + iot = tmp_path / "iot.json" + output = tmp_path / "state" / "b01_devices.json" + device.write_text( + json.dumps( + { + "did": 123456789, + "model": "roborock.vacuum.sc05", + "secret": "0123456789abcdef" * 2, + } + ) + ) + iot.write_text(json.dumps({"duid": "synthetic-cloud-duid"})) + return device, iot, output + + +def test_import_is_repeatable_and_preserves_other_devices(files): + device, iot, output = files + output.parent.mkdir() + other = {"model": "roborock.vacuum.sc05", "duid": "other", "secret": "a" * 32} + output.write_text(json.dumps({"devices": {"987654321": other}})) + import_device(*files) + first = output.read_bytes() + import_device(*files) + assert output.read_bytes() == first + entries = json.loads(first)["devices"] + assert entries["987654321"] == other + assert entries["123456789"]["duid"] == "synthetic-cloud-duid" + assert not list(output.parent.glob(".b01-*")) + + +def test_import_refuses_to_replace_existing_credentials(files): + device, iot, output = files + import_device(*files) + original = output.read_bytes() + iot.write_text(json.dumps({"duid": "different-cloud-duid"})) + with pytest.raises(ValueError, match="different entry"): + import_device(*files) + assert output.read_bytes() == original + + +@pytest.mark.parametrize( + "update", + [{"model": "roborock.vacuum.a15"}, {"did": "invalid"}, {"secret": "short"}], +) +def test_invalid_device_does_not_create_state(files, update): + device, _, output = files + value = json.loads(device.read_text()) + value.update(update) + device.write_text(json.dumps(value)) + with pytest.raises(ValueError): + import_device(*files) + assert not output.exists()