diff --git a/scripts/q7_iot_local_fields_reentry.sh b/scripts/q7_iot_local_fields_reentry.sh new file mode 100644 index 0000000..9c9c115 --- /dev/null +++ b/scripts/q7_iot_local_fields_reentry.sh @@ -0,0 +1,101 @@ +#!/bin/sh +# Offline research fixture only. No OTA packaging or device access. +# Arguments: existing JSON, API URL, MQTT URL, MQTT client ID, user, password. +set -eu + +if [ "$#" -ne 6 ]; then + printf 'usage: %s \n' "$0" >&2 + exit 2 +fi + +json_path=$1 +new_api=$2 +new_mqtt=$3 +new_clientid=$4 +new_user=$5 +new_password=$6 +bb=${Q7_BUSYBOX:-/bin/busybox} + +fail() { printf 'Q7 IoT local-field patch refused: %s\n' "$1" >&2; exit 1; } + +[ -f "$json_path" ] || fail 'input is not a regular file' +[ ! -L "$json_path" ] || fail 'input is a symbolic link' +[ -x "$bb" ] || fail 'BusyBox command is unavailable' +case "$json_path" in */iot.json) ;; *) fail 'input name must be iot.json' ;; esac + +# Q7's observed saved credential lengths are 16/16/32, and the recovered +# B01 derivation emits hexadecimal values of those lengths. This restriction +# also prevents JSON escaping and AWK -v escape interpretation. +for url in "$new_api" "$new_mqtt"; do + case "$url" in *://?*) ;; *) fail 'URL must contain a scheme and nonempty suffix' ;; esac + case "$url" in *[!A-Za-z0-9._:/+-]*) fail 'URL has unsupported characters' ;; esac + [ "${#url}" -le 240 ] || fail 'URL exceeds conservative 240-byte limit' +done +[ "${#new_clientid}" -eq 16 ] || fail 'MQTT client ID must have 16 characters' +[ "${#new_user}" -eq 16 ] || fail 'MQTT user must have 16 characters' +[ "${#new_password}" -eq 32 ] || fail 'MQTT password must have 32 characters' +for credential in "$new_clientid" "$new_user" "$new_password"; do + case "$credential" in *[!0-9a-fA-F]*) fail 'MQTT credentials must be hexadecimal' ;; esac +done + +backup="${json_path}.before-q7-local-edit" +reuse_backup=0 +if [ -e "$backup" ] || [ -L "$backup" ]; then + [ -f "$backup" ] && [ ! -L "$backup" ] || fail 'rollback copy is not a regular file' + "$bb" cmp -s "$json_path" "$backup" || fail 'rollback copy differs from restored source' + reuse_backup=1 +fi +tmp=$("$bb" mktemp "${json_path}.new.XXXXXX") || fail 'could not create same-directory temporary file' +cleanup() { "$bb" rm -f -- "$tmp"; } +trap cleanup EXIT +trap 'exit 1' HUP INT TERM + +"$bb" cp -p "$json_path" "$tmp" || fail 'could not stage file metadata' + +"$bb" awk -v api="$new_api" -v mqtt="$new_mqtt" \ + -v clientid="$new_clientid" -v username="$new_user" -v password="$new_password" ' +function patch(key, value, quoted, pattern, start, prefix, tail, endquote) { + quoted = "\"" key "\"" + if (index(line, quoted) == 0) return + pattern = "^[[:space:]]*" quoted "[[:space:]]*:[[:space:]]*\"[^\"]*\"[[:space:]]*,?[[:space:]]*$" + if (line !~ pattern) exit 10 + seen[key]++ + if (seen[key] != 1) exit 11 + start = match(line, quoted "[[:space:]]*:[[:space:]]*\"") + if (start == 0) exit 12 + prefix = substr(line, 1, RSTART + RLENGTH - 1) + tail = substr(line, RSTART + RLENGTH) + endquote = index(tail, "\"") + if (endquote == 0) exit 13 + line = prefix value "\"" substr(tail, endquote + 1) +} +{ + line = $0 + patch("api_url", api) + patch("mqtt_url", mqtt) + patch("mqtt_clientid", clientid) + patch("mqtt_usr", username) + patch("mqtt_passwd", password) + print line +} +END { + if (seen["api_url"] != 1 || seen["mqtt_url"] != 1 || + seen["mqtt_clientid"] != 1 || seen["mqtt_usr"] != 1 || + seen["mqtt_passwd"] != 1) exit 14 +} +' "$json_path" > "$tmp" || fail 'expected one standalone line for each of five fields' + +if "$bb" cmp -s "$json_path" "$tmp"; then + printf 'Fields already match; source unchanged\n' + exit 0 +fi + +if [ "$reuse_backup" -eq 0 ]; then + [ ! -e "$backup" ] && [ ! -L "$backup" ] || fail 'rollback copy appeared during edit' + "$bb" cp -p "$json_path" "$backup" || fail 'could not create rollback copy' + "$bb" cmp -s "$json_path" "$backup" || fail 'rollback copy does not match source' +fi +"$bb" sync || fail 'could not flush staged file and rollback copy' +"$bb" mv -f "$tmp" "$json_path" || fail 'atomic replacement failed' +"$bb" sync || fail 'replacement committed but sync failed; inspect source and backup' +printf 'Changed five IoT fields; rollback copy saved next to source\n' diff --git a/scripts/q7_migration_ota_builder.py b/scripts/q7_migration_ota_builder.py new file mode 100644 index 0000000..34f92f5 --- /dev/null +++ b/scripts/q7_migration_ota_builder.py @@ -0,0 +1,226 @@ +"""Build a Q7 custom-region OTA from a pinned firmware-wide profile. + +The profile is firmware-wide, not a dump of the next owner's vacuum. Its OTA +key and the output package are private. Cloud-identity fingerprints remain in +metadata and are not embedded in the OTA script. This command only builds +files; it does not host an update or contact a device. +""" + +from __future__ import annotations + +import argparse +import gzip +import hashlib +import json +import os +from pathlib import Path +import re +import struct +from urllib.parse import urlsplit + +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes + + +SCHEMA = "q7-sc05-03.01.74-migration-profile-v1" +REENTRY_SCHEMA = "q7-sc05-03.01.74-migration-profile-v2" +PROFILE_030180_SCHEMA = "q7-sc05-03.01.80-migration-profile-v1" +MODEL = "roborock.vacuum.sc05" +VERSION = "03.01.74" +PACKAGE_NAME = "q7-migration-v03.bin.gz.aes" +MAX_PACKAGE_BYTES = 4 * 1024 * 1024 +FIELDS = ("api_url", "mqtt_url", "mqtt_clientid", "mqtt_usr", "mqtt_passwd") +PROFILE_HASHES = { + "ota-key.bin": "a02bdcf7b3afdb5b0dce179326d81839c9d04d5bfb47fd318aba777633b01f5e", + "return.sh": "cffc933b62405211a18157c47c425344854607e7bdb830cac8fee527bc391a0c", + "editor.sh": "80ba06173a49e476d606d6f39610b2ed976e9f559961df80262f873e7623df01", +} +REENTRY_PROFILE_HASHES = { + **PROFILE_HASHES, + "editor.sh": "2fce38f7ba828b361fa644c5e8fea8c5858f9bc0eeaeab931610e18e6da3fb04", +} +PROFILE_030180_HASHES = { + "ota-key.bin": PROFILE_HASHES["ota-key.bin"], + "return.sh": "a5f599aaaa53c898090b81ff8e8a9a9312e48e75f18fb0a93f523492a6cacc26", + "editor.sh": REENTRY_PROFILE_HASHES["editor.sh"], +} +URL_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://[A-Za-z0-9._:/+\-]+$") +HEX_LENGTHS = {"mqtt_clientid": 16, "mqtt_usr": 16, "mqtt_passwd": 32} + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _package(begin: bytes, end: bytes, key: bytes) -> bytes: + """Pack the data-only SStarOta v0.3 format used by the inspected Q7.""" + tail = struct.pack("<6I", 0, 3, 0, 0, len(begin), len(end)) + container = struct.pack(" dict[str, str]: + if not isinstance(raw, dict) or set(raw) != set(FIELDS): + raise ValueError("Migration config must contain exactly the five IoT fields") + values = {} + for name in FIELDS: + value = raw[name] + if not isinstance(value, str): + raise ValueError(f"{name} must be a string") + values[name] = value + for name, scheme in (("api_url", "https://"), ("mqtt_url", "ssl://")): + value = values[name] + if not value.startswith(scheme) or len(value) > 240 or not URL_PATTERN.fullmatch(value): + raise ValueError(f"{name} must be a supported {scheme} URL up to 240 characters") + parsed = urlsplit(value) + if (not parsed.hostname or parsed.username or parsed.password or parsed.query + or parsed.fragment or parsed.path not in ("", "/")): + raise ValueError(f"{name} must be an origin without credentials, path, or query") + try: + _ = parsed.port + except ValueError as exc: + raise ValueError(f"{name} has an invalid port") from exc + for name, length in HEX_LENGTHS.items(): + if not re.fullmatch(rf"[0-9a-fA-F]{{{length}}}", values[name]): + raise ValueError(f"{name} must be {length} hexadecimal characters") + return values + + +def _profile(path: Path) -> tuple[bytes, bytes, bytes, str, str]: + manifest = json.loads((path / "manifest.json").read_text(encoding="utf-8")) + if not isinstance(manifest, dict): + raise ValueError("The firmware-wide profile manifest is malformed") + schema = manifest.get("schema") + spec = { + SCHEMA: (VERSION, PROFILE_HASHES), + REENTRY_SCHEMA: (VERSION, REENTRY_PROFILE_HASHES), + PROFILE_030180_SCHEMA: ("03.01.80", PROFILE_030180_HASHES), + }.get(schema) + if spec is None: + raise ValueError("The firmware-wide profile is not the inspected Q7 build") + version, expected = spec + if (manifest.get("model") != MODEL + or manifest.get("firmware_version") != version + or manifest.get("file_sha256") != expected): + raise ValueError("The firmware-wide profile is not the inspected Q7 build") + blobs = {} + for name, expected_hash in expected.items(): + item = path / name + if item.is_symlink() or not item.is_file(): + raise ValueError(f"Profile file {name} must be a regular file") + blob = item.read_bytes() + if _sha256(blob) != expected_hash: + raise ValueError(f"Profile file {name} failed its pinned hash") + blobs[name] = blob + key, end, editor = blobs["ota-key.bin"], blobs["return.sh"], blobs["editor.sh"] + if len(key) != 16 or not end.startswith(b"#!/bin/sh\n") or not editor.startswith(b"#!/bin/sh\n"): + raise ValueError("Profile key or recovery scripts have unexpected structure") + return key, end, editor, schema, version + + +def build(config: object, profile: Path, out: Path) -> dict[str, object]: + if not isinstance(config, dict): + raise ValueError("Migration config must be an object") + preflight = config.get("_preflight") + if preflight is not None: + if (not isinstance(preflight, dict) + or set(preflight) != {"duid_sha256", "local_key_sha256"} + or any(not isinstance(value, str) or not re.fullmatch(r"[0-9a-f]{64}", value) + for value in preflight.values())): + raise ValueError("Migration preflight fingerprints are malformed") + values = _fields({name: value for name, value in config.items() if name != "_preflight"}) + config_sha256 = _sha256(json.dumps(values, sort_keys=True, separators=(",", ":")).encode("ascii")) + if out.exists(): + raise FileExistsError(f"Refusing to overwrite {out}") + key, end, editor, schema, version = _profile(profile) + args = ['"${Q7_IOT_JSON_PATH:-/userdata/rriot/data_dir/iot.json}"'] + args.extend("'" + values[name] + "'" for name in FIELDS) + begin = ("#!/bin/sh\nset -- " + " ".join(args) + "\n").encode("ascii") + editor.split(b"\n", 1)[1] + encrypted = _package(begin, end, key) + + out.mkdir(parents=True) + try: + out.chmod(0o700) + except OSError: + pass + name = PACKAGE_NAME + artifact = out / name + descriptor = os.open(artifact, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(descriptor, "wb") as handle: + handle.write(encrypted) + metadata: dict[str, object] = { + "firmware": f"{MODEL} {version}, pinned inspected firmware profile", + "target_firmware": version, + "package": name, + "encrypted_size_bytes": len(encrypted), + "encrypted_md5": hashlib.md5(encrypted).hexdigest(), + "encrypted_sha256": _sha256(encrypted), + "begin_sha256": _sha256(begin), + "end_sha256": _sha256(end), + "build_mode": "portable_profile", + "profile_schema": schema, + "signed": False, + "secrets_in_package": True, + "config_field_names": list(FIELDS), + "config_sha256": config_sha256, + "preflight": preflight, + } + (out / "metadata.json").write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8") + return metadata + + +def inspect(artifact_dir: Path) -> tuple[bytes, dict[str, object]]: + """Verify the exact encrypted package that will be hosted for the Q7.""" + directory = artifact_dir.resolve() + metadata = json.loads((directory / "metadata.json").read_text(encoding="utf-8")) + if not isinstance(metadata, dict) or metadata.get("signed") is not False: + raise ValueError("Expected an unsigned Q7 migration metadata object") + version = metadata.get("target_firmware") + if (version not in ("03.01.74", "03.01.80") + or f"{MODEL} {version}" not in str(metadata.get("firmware", ""))): + raise ValueError("Package metadata does not name the inspected Q7 firmware") + if version == "03.01.80" and metadata.get("profile_schema") != PROFILE_030180_SCHEMA: + raise ValueError("03.01.80 requires its version-specific profile") + if metadata.get("package") != PACKAGE_NAME: + raise ValueError("Unexpected Q7 migration package name") + package = directory / PACKAGE_NAME + if package.is_symlink() or package.resolve().parent != directory: + raise ValueError("Package must be a regular file in the artifact directory") + payload = package.read_bytes() + if not payload or len(payload) > MAX_PACKAGE_BYTES or len(payload) % 16: + raise ValueError("Package length is outside the AES-aligned limit") + digest_sha = hashlib.sha256(payload).hexdigest() + digest_md5 = hashlib.md5(payload).hexdigest() + if (metadata.get("encrypted_size_bytes") != len(payload) + or metadata.get("encrypted_sha256") != digest_sha + or metadata.get("encrypted_md5") != digest_md5): + raise ValueError("Package bytes do not match metadata") + return payload, { + "package": PACKAGE_NAME, + "encrypted_size_bytes": len(payload), + "encrypted_sha256": digest_sha, + "encrypted_md5": digest_md5, + "target_firmware": version, + "preflight": metadata.get("preflight"), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--config", type=Path, required=True, help="private JSON manifest with five IoT fields and cloud-key preflight fingerprints") + parser.add_argument("--profile", type=Path, required=True, help="private firmware-wide OTA profile") + parser.add_argument("--out", type=Path, required=True, help="new private artifact directory") + args = parser.parse_args() + metadata = build(json.loads(args.config.read_text(encoding="utf-8")), args.profile, args.out) + summary = {key: metadata[key] for key in ( + "package", "encrypted_size_bytes", "encrypted_md5", "encrypted_sha256" + )} + print(json.dumps(summary, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/q7_owner_ota.py b/scripts/q7_owner_ota.py new file mode 100644 index 0000000..8b621f7 --- /dev/null +++ b/scripts/q7_owner_ota.py @@ -0,0 +1,219 @@ +"""Preflight or send one owner-authenticated Q7 migration OTA. + +The Q7 must still be online in the owner's Roborock account. This tool uses +normal account login and the current cloud DUID/local key; it needs no device +dump or factory HMAC secret. Without --live it sends read-only queries only. +--save-account optionally writes a private account export for later commands. +""" + +from __future__ import annotations + +import argparse +import asyncio +import hashlib +import hmac +import json +import os +from pathlib import Path +import re +import sys +from urllib.parse import urlsplit +from urllib.request import urlopen + +import aiohttp +from roborock.data import UserData +from roborock.devices.rpc.b01_q7_channel import send_decoded_command +from roborock.devices.transport.mqtt_channel import MqttChannel +from roborock.mqtt.roborock_session import create_mqtt_session +from roborock.protocol import create_mqtt_params +from roborock.protocols.b01_q7_protocol import Q7RequestMessage +from roborock.web_api import RoborockApiClient + +try: + from .q7_migration_ota_builder import inspect +except ImportError: # Direct ``python scripts/q7_owner_ota.py`` execution. + from q7_migration_ota_builder import inspect + + +def package_request(artifact_dir: Path, url: str) -> dict[str, object]: + payload, metadata = inspect(artifact_dir) + parsed = urlsplit(url) + if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password: + raise ValueError("Package URL must be HTTP(S) and reachable by the vacuum") + if parsed.hostname in ("localhost", "127.0.0.1", "::1") or parsed.fragment: + raise ValueError("Package URL cannot use a loopback host or fragment") + with urlopen(url, timeout=15) as response: + hosted = response.read(len(payload) + 1) + if hosted != payload: + raise ValueError("Hosted package differs from the pinned local artifact") + return { + "packageUrl": url, + "md5": metadata["encrypted_md5"], + "packageSize": str(len(payload)), + "signed": False, + "packageType": "robot", + } + + +def cloud_key_matches_server_import(preflight: object, *, duid: str, local_key: str) -> bool: + """Compare the current cloud identity with the server import pinned at build time.""" + if not isinstance(preflight, dict) or set(preflight) != {"duid_sha256", "local_key_sha256"}: + return False + try: + local_key_bytes = local_key.encode("ascii") + except UnicodeEncodeError: + return False + actual = { + "duid_sha256": hashlib.sha256(duid.encode("utf-8")).hexdigest(), + "local_key_sha256": hashlib.sha256(local_key_bytes).hexdigest(), + } + return all(isinstance(preflight[name], str) + and re.fullmatch(r"[0-9a-f]{64}", preflight[name]) + and hmac.compare_digest(actual[name], preflight[name]) for name in actual) + + +async def _account(args: argparse.Namespace, web_session: aiohttp.ClientSession) -> tuple[RoborockApiClient, UserData]: + if args.account: + saved = json.loads(args.account.read_text(encoding="utf-8")) + if not isinstance(saved, dict) or not isinstance(saved.get("username"), str): + raise ValueError("Account JSON must contain username and user_data") + owner = UserData.from_dict(saved["user_data"]) + api = RoborockApiClient(saved["username"], base_url=saved.get("base_url"), session=web_session) + return api, owner + if not args.email: + raise ValueError("Provide --email for code login or --account for an existing private account export") + api = RoborockApiClient(args.email, session=web_session) + await api.request_code_v4() + code = input("Roborock email login code: ").strip() + if not code: + raise ValueError("A login code is required") + return api, await api.code_login_v4(code) + + +def _save_account(path: Path, *, email: str, base_url: str, owner: UserData) -> None: + """Create an opt-in private export for subsequent owner commands.""" + payload = { + "username": email, + "base_url": base_url, + "user_data": owner.as_dict(), + } + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + descriptor = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) + with os.fdopen(descriptor, "w", encoding="utf-8") as output: + json.dump(payload, output, separators=(",", ":")) + output.write("\n") + + +async def run(args: argparse.Namespace) -> dict[str, object]: + if args.save_account and args.save_account.exists(): + raise FileExistsError("Private account export already exists") + async with aiohttp.ClientSession() as web_session: + api, owner = await _account(args, web_session) + home = await api.get_home_data_v3(owner) + if args.save_account: + _save_account(args.save_account, email=args.email, base_url=await api.base_url, owner=owner) + models = {product.id: product.model for product in home.products} + if args.list: + return { + "devices": [{ + "duid": device.duid, + "firmware": device.fv, + "cloud_online": device.online, + "local_key_length": len(device.local_key or ""), + } for device in home.devices if models.get(device.product_id) == "roborock.vacuum.sc05"], + "command_sent": False, + } + candidates = [device for device in home.devices if device.duid == args.duid] + if len(candidates) != 1: + raise ValueError("Exactly one device with the selected cloud DUID must belong to this account") + device = candidates[0] + if models.get(device.product_id) != "roborock.vacuum.sc05": + raise ValueError("The selected cloud device is not a Q7 sc05") + _payload, package = inspect(args.artifact_dir) + target_firmware = str(package["target_firmware"]) + report: dict[str, object] = { + "duid_hash": hashlib.sha256(device.duid.encode()).hexdigest()[:12], + "model": models[device.product_id], + "firmware": device.fv, + "cloud_online": device.online, + "package_sha256": package["encrypted_sha256"], + "target_firmware": target_firmware, + "command_sent": False, + } + if device.online is not True or device.fv != target_firmware or len(device.local_key or "") != 16: + report["aborted"] = f"Q7 must be cloud-online on {target_firmware} with a current 16-byte local key" + return report + if not cloud_key_matches_server_import( + package.get("preflight"), duid=device.duid, local_key=device.local_key + ): + report["aborted"] = "Current cloud identity differs from the local server import; refresh the server import and rebuild both packages" + return report + mqtt_params = create_mqtt_params(owner.rriot) + session = await create_mqtt_session(mqtt_params) + try: + channel = MqttChannel(session, device.duid, device.local_key, owner.rriot, mqtt_params) + + async def query(method: str, params: dict[str, object]) -> object: + return await asyncio.wait_for(send_decoded_command( + channel, Q7RequestMessage(dps=10000, command=method, params=params) + ), timeout=15) + + status = await query("prop.get", {"property": ["status"]}) + progress = await query("ota.progress.get", {}) + report["work_status"] = status.get("status") if isinstance(status, dict) else None + report["ota_state"] = progress.get("state") if isinstance(progress, dict) else None + if report["work_status"] != 4 or report["ota_state"] != "idle": + report["aborted"] = "Q7 must be charging with OTA idle" + return report + # Verify the temporary hosted URL immediately before it could be sent. + request = package_request(args.artifact_dir, args.url) + if not args.live: + return report + response = await query("ota.upgrade.set", request) + report["command_sent"] = True + report["set_result"] = response.get("result") if isinstance(response, dict) else None + observations: list[object] = [] + for _ in range(8): + await asyncio.sleep(2) + try: + state = await query("ota.progress.get", {}) + observations.append(state.get("state") if isinstance(state, dict) else type(state).__name__) + except Exception as exc: + observations.append(type(exc).__name__) + break + report["ota_states_after"] = observations + return report + finally: + await session.close() + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + identity = parser.add_mutually_exclusive_group(required=True) + identity.add_argument("--email", help="Roborock account email; requests a login code") + identity.add_argument("--account", type=Path, help="private JSON with username, base_url, user_data") + parser.add_argument("--save-account", type=Path, help="with --email, exclusively create a private account export for later runs") + parser.add_argument("--list", action="store_true", help="list account-owned Q7 DUIDs; no device connection") + parser.add_argument("--duid", help="current cloud DUID from the owner's account") + parser.add_argument("--artifact-dir", type=Path) + parser.add_argument("--url", help="HTTP(S) URL serving the exact encrypted artifact") + parser.add_argument("--live", action="store_true", help="send the single OTA command after preflight") + args = parser.parse_args() + if args.save_account and not args.email: + parser.error("--save-account requires --email") + if args.list: + if args.duid or args.artifact_dir or args.url or args.live: + parser.error("--list cannot be combined with OTA options") + elif not args.duid or not args.artifact_dir or not args.url: + parser.error("--duid, --artifact-dir and --url are required for an OTA preflight") + try: + print(json.dumps(asyncio.run(run(args)), sort_keys=True), flush=True) + except Exception as exc: + print(json.dumps({"status": "error", "error_type": type(exc).__name__}), flush=True) + raise SystemExit(1) from None + + +if __name__ == "__main__": + if sys.platform == "win32": + asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) + main() diff --git a/scripts/q7_prepare_migration.py b/scripts/q7_prepare_migration.py new file mode 100644 index 0000000..130bdde --- /dev/null +++ b/scripts/q7_prepare_migration.py @@ -0,0 +1,123 @@ +"""Reserve local MQTT credentials for a cloud-paired Q7 URL migration. + +Requires the local server's new admin endpoint and an existing cloud import +with the Q7 cloud DUID and true local key. A numeric DID is not required. +Writes a private JSON input with five IoT fields and cloud-identity fingerprints +for the offline OTA builder. This sends no command or package to the vacuum. +""" + +from __future__ import annotations + +import argparse +from getpass import getpass +import hashlib +import json +import os +from pathlib import Path +import re +from urllib.parse import urlsplit + +import httpx + + +URL_PATTERN = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://[A-Za-z0-9._:/+\-]+$") + + +def _https_origin(raw: str) -> str: + parsed = urlsplit(raw) + if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password: + raise ValueError("--server must be an HTTPS origin without embedded credentials") + if parsed.path not in ("", "/") or parsed.query or parsed.fragment: + raise ValueError("--server must not include a path, query, or fragment") + try: + parsed.port + except ValueError as exc: + raise ValueError("--server has an invalid port") from exc + return raw.rstrip("/") + + +def _target_url(raw: str, *, field: str, scheme: str) -> str: + if not raw.startswith(scheme + "://") or len(raw) > 240 or not URL_PATTERN.fullmatch(raw): + raise ValueError(f"{field} must be a supported {scheme} URL of at most 240 characters") + parsed = urlsplit(raw) + if not parsed.hostname or parsed.username or parsed.password or parsed.fragment: + raise ValueError(f"{field} must have a host and no embedded credentials or fragment") + try: + parsed.port + except ValueError as exc: + raise ValueError(f"{field} has an invalid port") from exc + return raw + + +def prepare(*, server: str, duid: str, api_url: str, + mqtt_url: str, out: Path, admin_password: str, + did: str = "") -> dict[str, str]: + origin = _https_origin(server) + api_url = _target_url(api_url, field="api_url", scheme="https") + mqtt_url = _target_url(mqtt_url, field="mqtt_url", scheme="ssl") + if out.exists(): + raise FileExistsError(f"Refusing to overwrite {out}") + if not duid.strip(): + raise ValueError("Cloud DUID is required") + with httpx.Client(base_url=origin, timeout=15.0, follow_redirects=False) as client: + login = client.post("/admin/api/login", json={"password": admin_password}) + login.raise_for_status() + response = client.post( + "/admin/api/q7/migration-credentials", + json={"did": did.strip(), "duid": duid.strip()}, + ) + response.raise_for_status() + payload = response.json() + if payload.get("duid") != duid.strip() or (did.strip() and payload.get("did") != did.strip()): + raise ValueError("Server response does not match the requested Q7 identity") + local_key_sha256 = payload.get("local_key_sha256") + if not isinstance(local_key_sha256, str) or not re.fullmatch(r"[0-9a-f]{64}", local_key_sha256): + raise ValueError("Server did not provide a valid cloud-imported local-key fingerprint") + fields = { + "api_url": api_url, + "mqtt_url": mqtt_url, + "mqtt_clientid": str(payload["mqtt_clientid"]), + "mqtt_usr": str(payload["mqtt_usr"]), + "mqtt_passwd": str(payload["mqtt_passwd"]), + "_preflight": { + "duid_sha256": hashlib.sha256(duid.strip().encode("utf-8")).hexdigest(), + "local_key_sha256": local_key_sha256, + }, + } + out.parent.mkdir(parents=True, exist_ok=True) + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + descriptor = os.open(out, flags, 0o600) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as handle: + json.dump(fields, handle, indent=2) + handle.write("\n") + except BaseException: + out.unlink(missing_ok=True) + raise + return {"did": str(payload.get("did") or ""), "duid": duid.strip(), "manifest": str(out.resolve())} + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--server", required=True, help="HTTPS admin origin, e.g. https://api.example.com:555") + parser.add_argument("--did", default="", help="optional known numeric device ID") + parser.add_argument("--duid", required=True) + parser.add_argument("--api-url", required=True) + parser.add_argument("--mqtt-url", required=True) + parser.add_argument("--out", type=Path, required=True, help="new private manifest path") + args = parser.parse_args() + admin_password = getpass("Local server admin password: ") + result = prepare( + server=args.server, + did=args.did, + duid=args.duid, + api_url=args.api_url, + mqtt_url=args.mqtt_url, + out=args.out, + admin_password=admin_password, + ) + print(json.dumps({**result, "device_command_sent": False}, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py index ee7e027..87c4217 100644 --- a/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py +++ b/src/roborock_local_server/bundled_backend/mqtt_broker_server/topic_bridge.py @@ -72,6 +72,7 @@ def __init__( fixed_device_duid: str = "", fixed_device_mqtt_usr: str = "", runtime_state: Any | None = None, + runtime_credentials: Any | None = None, inventory_path: Path | None = None, ) -> None: self._host = host @@ -91,6 +92,7 @@ def __init__( self._warned_unmapped_device_topics: set[DeviceTopicKey] = set() self._warned_multi_device = False self._runtime_state = runtime_state + self._runtime_credentials = runtime_credentials self._inventory_path = Path(inventory_path) if inventory_path is not None else None fixed_did = fixed_device_did.strip() or fixed_device_duid.strip() @@ -116,7 +118,10 @@ async def stop(self) -> None: self._task = None def _remember_device_seen(self, device_topic: DeviceTopicKey) -> None: + first_seen = device_topic not in self._seen_device_topics self._seen_device_topics[device_topic] = time.monotonic() + if first_seen: + self._last_duid_map_refresh_monotonic = 0.0 seen_device_count = self._seen_device_did_count() if seen_device_count <= 1: return @@ -173,7 +178,7 @@ def _load_inventory_devices(self) -> list[dict[str, Any]]: return devices def _refresh_duid_to_did_map(self) -> None: - if self._runtime_state is None or self._inventory_path is None: + if self._runtime_state is None and self._runtime_credentials is None: return now = time.monotonic() @@ -182,17 +187,13 @@ def _refresh_duid_to_did_map(self) -> None: self._last_duid_map_refresh_monotonic = now try: - key_models_by_did = self._runtime_state.key_models_by_did() + key_models_by_did = self._runtime_state.key_models_by_did() if self._runtime_state is not None else {} except Exception: key_models_by_did = {} if not isinstance(key_models_by_did, dict): key_models_by_did = {} inventory_devices = self._load_inventory_devices() - if not inventory_devices: - self._duid_to_did = {} - return - did_counts_by_model: dict[str, int] = {} unique_did_by_model: dict[str, str] = {} for did, model_value in key_models_by_did.items(): @@ -211,10 +212,16 @@ def _refresh_duid_to_did_map(self) -> None: inv_counts_by_model[model] = inv_counts_by_model.get(model, 0) + 1 fresh_map: dict[str, str] = {} + migration_duids = ( + self._runtime_credentials.q7_migration_duids() + if self._runtime_credentials is not None else set() + ) for raw in inventory_devices: duid = str(raw.get("duid") or raw.get("did") or raw.get("device_id") or "").strip() if not duid: continue + if duid in migration_duids: + continue explicit_did = str(raw.get("did") or raw.get("device_did") or "").strip() if explicit_did: fresh_map[duid] = explicit_did @@ -230,6 +237,10 @@ def _refresh_duid_to_did_map(self) -> None: if mapped_did: fresh_map[duid] = mapped_did + if self._runtime_credentials is not None: + for duid, did in self._runtime_credentials.verified_q7_migration_links().items(): + fresh_map[duid] = did + map_changed = fresh_map != self._duid_to_did if map_changed: self._duid_to_did = fresh_map diff --git a/src/roborock_local_server/bundled_backend/mqtt_tls_proxy_server/server.py b/src/roborock_local_server/bundled_backend/mqtt_tls_proxy_server/server.py index 851217c..5cace97 100644 --- a/src/roborock_local_server/bundled_backend/mqtt_tls_proxy_server/server.py +++ b/src/roborock_local_server/bundled_backend/mqtt_tls_proxy_server/server.py @@ -68,9 +68,10 @@ def __init__( self._counter = 0 self._lock = threading.Lock() self._conn_protocol_levels: dict[str, int] = {} + self._conn_auth: dict[str, tuple[str, bool]] = {} self._conn_endpoints: dict[str, tuple[socket.socket, socket.socket]] = {} self._pending_onboarding_auth: dict[str, dict[str, str]] = {} - self._trace_queue: queue.Queue[tuple[str, str, bytes] | None] = queue.Queue() + self._trace_queue: queue.Queue[tuple[str, str, bytes, str, bool] | None] = queue.Queue() self._trace_thread: threading.Thread | None = None self._protocol_auth = ( ProtocolAuthStore( @@ -513,7 +514,14 @@ def _parse_v1_rpc_payload(payload_utf8: str | None, protocol_value: int) -> dict parsed["error"] = rpc_obj.get("error") return parsed - def _trace_packet(self, conn_id: str, direction: str, packet: bytes) -> None: + def _trace_packet( + self, + conn_id: str, + direction: str, + packet: bytes, + authenticated_username: str = "", + device_credentials_verified: bool = False, + ) -> None: packet_type = packet[0] >> 4 if packet_type in (12, 13): # PINGREQ, PINGRESP return @@ -548,7 +556,12 @@ def _trace_packet(self, conn_id: str, direction: str, packet: bytes) -> None: payload_preview=payload_preview(payload), ) if self.runtime_credentials is not None: - self.runtime_credentials.record_mqtt_topic(topic=topic) + self.runtime_credentials.record_mqtt_topic( + topic=topic, + direction=direction, + authenticated_username=authenticated_username, + device_credentials_verified=device_credentials_verified, + ) messages, variant, decode_error, decode_key_source = self._decode_mqtt_payload(topic, payload) entry: dict[str, Any] = { @@ -677,9 +690,15 @@ def _run_trace_worker(self) -> None: item = self._trace_queue.get() if item is None: return - conn_id, direction, packet = item + conn_id, direction, packet, authenticated_username, device_credentials_verified = item try: - self._trace_packet(conn_id, direction, packet) + self._trace_packet( + conn_id, + direction, + packet, + authenticated_username, + device_credentials_verified, + ) except Exception: self.logger.exception( "[conn %s %s] tracing failed", @@ -700,7 +719,11 @@ def _ensure_trace_worker(self) -> None: def _queue_trace_packet(self, conn_id: str, direction: str, packet: bytes) -> None: self._ensure_trace_worker() - self._trace_queue.put((conn_id, direction, packet)) + with self._lock: + authenticated_username, device_credentials_verified = self._conn_auth.get(conn_id, ("", False)) + self._trace_queue.put(( + conn_id, direction, packet, authenticated_username, device_credentials_verified, + )) def _relay(self, src: socket.socket, dst: socket.socket, conn_id: str, direction: str, frame_buf: bytearray) -> None: try: @@ -770,6 +793,12 @@ def _handle_client(self, tls_conn: socket.socket | ssl.SSLSocket, addr: tuple[st self._queue_trace_packet(conn_id, "b2c", reject_packet) return + with self._lock: + self._conn_auth[conn_id] = ( + str((connect_info or {}).get("username") or "").strip(), + auth_reason == "device_mqtt_user", + ) + backend = socket.socket(socket.AF_INET, socket.SOCK_STREAM) backend.connect((self.backend_host, self.backend_port)) self._register_conn_endpoints(conn_id, tls_conn, backend) @@ -807,6 +836,7 @@ def _handle_client(self, tls_conn: socket.socket | ssl.SSLSocket, addr: tuple[st self.runtime_state.record_mqtt_disconnect(conn_id=conn_id) with self._lock: self._conn_protocol_levels.pop(conn_id, None) + self._conn_auth.pop(conn_id, None) self.logger.info("[conn %s] closed", conn_id) def start(self) -> threading.Thread: 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 bf44a3b..25724a3 100644 --- a/src/roborock_local_server/bundled_backend/shared/runtime_credentials.py +++ b/src/roborock_local_server/bundled_backend/shared/runtime_credentials.py @@ -4,8 +4,11 @@ from collections import Counter from datetime import datetime, timezone +import hashlib import json +import logging from pathlib import Path +import re import secrets import threading from typing import Any @@ -14,6 +17,9 @@ from .data_helpers import utcnow_iso +_LOGGER = logging.getLogger(__name__) + + def _clean_str(value: Any) -> str: return str(value or "").strip() @@ -252,6 +258,11 @@ def _normalize_device(raw: dict[str, Any]) -> dict[str, str]: "last_nc_at": _clean_str(raw.get("last_nc_at")), "last_mqtt_seen_at": _clean_str(raw.get("last_mqtt_seen_at")), } + migration_clientid = _clean_str(raw.get("migration_mqtt_clientid")) + if migration_clientid: + device["migration_mqtt_clientid"] = migration_clientid + if raw.get("migration_did_verified") is True or raw.get("migration_did_verified") == "1": + device["migration_did_verified"] = "1" return device def _save_locked(self) -> None: @@ -470,6 +481,15 @@ def ensure_device( if normalized_localkey and device.get("localkey") != normalized_localkey: device["localkey"] = normalized_localkey changed = True + # Inventory lookups and seeding are fallback provenance. A + # cloud-imported Q7 keeps its verified key origin and migration + # marker when the admin dashboard reads its inventory row. + if ( + normalized_source in {"inventory_seed", "inventory"} + and device.get("local_key_source") == "inventory_cloud" + and device.get("migration_mqtt_clientid") + ): + normalized_source = "" if normalized_source and device.get("local_key_source") != normalized_source: device["local_key_source"] = normalized_source changed = True @@ -485,6 +505,132 @@ def ensure_device( self._save_locked() return dict(device) + def prepare_q7_migration_credentials(self, *, duid: str, did: str = "") -> dict[str, str]: + """Reserve MQTT credentials for a cloud-imported sc05 without a flash dump.""" + normalized_duid = _clean_str(duid) + normalized_did = _clean_str(did) + if not normalized_duid: + raise ValueError("Cloud DUID is required") + + with self._lock: + matches = [ + item for item in self._devices + if item.get("duid") == normalized_duid + and item.get("model") == "roborock.vacuum.sc05" + and item.get("local_key_source") == "inventory_cloud" + ] + if not matches: + raise KeyError("Cloud-imported Q7 sc05 device was not found") + if len(matches) != 1: + raise ValueError("Cloud DUID is ambiguous") + device = matches[0] + localkey = _clean_str(device.get("localkey")) + try: + localkey_bytes = localkey.encode("ascii") + except UnicodeEncodeError: + localkey_bytes = b"" + if len(localkey_bytes) != 16: + raise ValueError("A 16-byte cloud-imported local key is required") + local_key_sha256 = hashlib.sha256(localkey_bytes).hexdigest() + stored_did = _clean_str(device.get("did")) + if normalized_did and stored_did and stored_did != normalized_did: + raise ValueError("DID conflicts with the cloud device") + if normalized_did and not stored_did: + linked = [ + item for item in self._devices + if item is not device + and item.get("did") == normalized_did + and item.get("localkey") == localkey + ] + if len(linked) != 1: + raise ValueError("DID is not linked to this cloud local key") + + username = _clean_str(device.get("device_mqtt_usr")) + password = _clean_str(device.get("device_mqtt_pass")) + clientid = _clean_str(device.get("migration_mqtt_clientid")) + if username or password or clientid: + if not ( + re.fullmatch(r"[0-9a-f]{16}", username) + and re.fullmatch(r"[0-9a-f]{32}", password) + and re.fullmatch(r"[0-9a-f]{16}", clientid) + ): + raise ValueError("Device already has non-migration MQTT credentials") + return { + "did": stored_did or normalized_did, + "duid": normalized_duid, + "local_key_sha256": local_key_sha256, + "mqtt_clientid": clientid, + "mqtt_usr": username, + "mqtt_passwd": password, + } + + existing_usernames = {_clean_str(item.get("device_mqtt_usr")) for item in self._devices} + while True: + username = secrets.token_hex(8) + if username not in existing_usernames: + break + password = secrets.token_hex(16) + clientid = secrets.token_hex(8) + device["device_mqtt_usr"] = username + device["device_mqtt_pass"] = password + device["migration_mqtt_clientid"] = clientid + device["updated_at"] = utcnow_iso() + self._save_locked() + return { + "did": stored_did or normalized_did, + "duid": normalized_duid, + "local_key_sha256": local_key_sha256, + "mqtt_clientid": clientid, + "mqtt_usr": username, + "mqtt_passwd": password, + } + + def migration_cloud_duid_for_mqtt_username(self, username: str) -> str: + """Resolve only a pre-registered Q7 migration username to its cloud DUID.""" + normalized_username = _clean_str(username) + if not normalized_username: + return "" + with self._lock: + matches = [ + item for item in self._devices + if item.get("model") == "roborock.vacuum.sc05" + and item.get("local_key_source") == "inventory_cloud" + and item.get("migration_mqtt_clientid") + and item.get("device_mqtt_usr") == normalized_username + and item.get("device_mqtt_pass") + ] + return _clean_str(matches[0].get("duid")) if len(matches) == 1 else "" + + def verified_q7_migration_links(self) -> dict[str, str]: + """Return DUID to DID links learned from authenticated device publishes.""" + with self._lock: + return { + _clean_str(item.get("duid")): _clean_str(item.get("did")) + for item in self._devices + if item.get("model") == "roborock.vacuum.sc05" + and item.get("local_key_source") == "inventory_cloud" + and item.get("migration_mqtt_clientid") + and item.get("device_mqtt_usr") + and item.get("device_mqtt_pass") + and item.get("migration_did_verified") == "1" + and item.get("duid") + and item.get("did") + } + + def q7_migration_duids(self) -> set[str]: + """DUIDs that must not use model-only bridge guesses during migration.""" + with self._lock: + return { + _clean_str(item.get("duid")) + for item in self._devices + if item.get("model") == "roborock.vacuum.sc05" + and item.get("local_key_source") == "inventory_cloud" + and item.get("migration_mqtt_clientid") + and item.get("device_mqtt_usr") + and item.get("device_mqtt_pass") + and item.get("duid") + } + def link_did_to_duid( self, *, @@ -605,15 +751,107 @@ def localkey_for_topic(self, topic: str) -> str: return _clean_str(device.get("localkey")) if device else "" return "" - def record_mqtt_topic(self, *, topic: str) -> None: + def record_mqtt_topic( + self, + *, + topic: str, + direction: str = "c2b", + authenticated_username: str = "", + device_credentials_verified: bool = False, + ) -> None: normalized_topic = _clean_str(topic) now = utcnow_iso() - if normalized_topic.startswith("rr/d/"): + if direction == "c2b" and normalized_topic.startswith("rr/d/i/"): parts = normalized_topic.split("/") if len(parts) >= 5: + did, username = parts[3], parts[4] + if authenticated_username and authenticated_username != username: + return + with self._lock: + migrated = [ + item for item in self._devices + if item.get("model") == "roborock.vacuum.sc05" + and item.get("local_key_source") in {"inventory_cloud", "inventory"} + and item.get("migration_mqtt_clientid") + and item.get("device_mqtt_usr") == username + and item.get("device_mqtt_pass") + ] + if not migrated and device_credentials_verified and authenticated_username == username: + _LOGGER.warning("Authenticated rr/d/i publish has no reserved Q7 migration record") + occupants = [ + item for item in self._devices + if item.get("did") == did + and all(item is not candidate for candidate in migrated) + ] + # Older startup code could observe the authenticated Q7 + # topic before recognizing the reserved migration record, + # leaving a topic-only row for the same MQTT credentials. + # Merge only that exact anonymous duplicate; a real device + # identity or different credentials remain a conflict. + stale_topic_row = ( + occupants[0] + if len(migrated) == 1 and len(occupants) == 1 + and not occupants[0].get("duid") + and not occupants[0].get("model") + and not occupants[0].get("localkey") + and not occupants[0].get("local_key_source") + and occupants[0].get("device_mqtt_usr") == username + and occupants[0].get("device_mqtt_pass") in ( + "", migrated[0].get("device_mqtt_pass"), + ) + else None + ) + did_taken = bool(occupants) and stale_topic_row is None + # A Q7 may publish its cloud DUID as the rr/d topic ID + # even when an older numeric DID was saved for the same + # reserved MQTT credentials. An authenticated publish on + # its exact cloud DUID is stronger evidence than that old + # hint; an arbitrary different topic ID remains refused. + prior_did = _clean_str(migrated[0].get("did")) if len(migrated) == 1 else "" + cloud_duid_topic = ( + len(migrated) == 1 and did == _clean_str(migrated[0].get("duid")) + ) + prior_did_conflicts = bool(prior_did and prior_did != did and not cloud_duid_topic) + if migrated and ( + len(migrated) != 1 + or not device_credentials_verified + or authenticated_username != username + or did_taken + or prior_did_conflicts + ): + _LOGGER.warning( + "Q7 migration topic link refused candidate_count=%d connect_verified=%s " + "connect_username_matches=%s topic_taken=%s prior_did_conflicts=%s", + len(migrated), device_credentials_verified, + authenticated_username == username, did_taken, + prior_did_conflicts, + ) + return + if len(migrated) == 1: + if stale_topic_row is not None: + self._devices.remove(stale_topic_row) + # Older dashboard reads could downgrade the saved + # provenance to plain "inventory". The migration + # marker plus an authenticated publish on the + # reserved credentials recovers that exact record. + migrated[0]["local_key_source"] = "inventory_cloud" + if prior_did and prior_did != did: + _LOGGER.info( + "Q7 migration topic ID rebound to its cloud DUID from authenticated publish" + ) + migrated[0]["did"] = did + migrated[0]["migration_did_verified"] = "1" + migrated[0]["last_mqtt_seen_at"] = now + migrated[0]["updated_at"] = now + self._save_locked() + _LOGGER.info( + "Q7 migration topic linked from authenticated device publish " + "(merged_stale_topic=%s)", stale_topic_row is not None, + ) + return self.ensure_device( - did=parts[3], - device_mqtt_usr=parts[4], + did=did, + device_mqtt_usr=username, last_mqtt_seen_at=now, assign_localkey=False, ) @@ -818,6 +1056,14 @@ def sync_inventory(self) -> None: normalized_model and inventory_model_counts.get(normalized_model, 0) == 1 and did_model_counts.get(normalized_model, 0) == 1 + # A cloud-imported Q7 being migrated must learn its DID + # from an authenticated device publish. Historical local + # key state can belong to an earlier account pairing. + and not ( + normalized_model == "roborock.vacuum.sc05" + and device.get("local_key_source") == "inventory_cloud" + and device.get("migration_mqtt_clientid") + ) ): did = next( ( @@ -827,7 +1073,10 @@ def sync_inventory(self) -> None: ), "", ) - if did and device.get("did") != did: + if ( + did and device.get("did") != did + and not (device.get("migration_mqtt_clientid") and device.get("did")) + ): device["did"] = did changed = True device_changed = True diff --git a/src/roborock_local_server/server.py b/src/roborock_local_server/server.py index 924b04e..062552f 100644 --- a/src/roborock_local_server/server.py +++ b/src/roborock_local_server/server.py @@ -1713,6 +1713,7 @@ async def start(self) -> None: port=self.config.broker.port, logger=self.loggers["mqtt"], runtime_state=self.runtime_state, + runtime_credentials=self.runtime_credentials, inventory_path=self.paths.inventory_path, ) await self._topic_bridge.start() diff --git a/src/roborock_local_server/standalone_admin.py b/src/roborock_local_server/standalone_admin.py index 8aac826..a9f2374 100644 --- a/src/roborock_local_server/standalone_admin.py +++ b/src/roborock_local_server/standalone_admin.py @@ -401,6 +401,29 @@ async def admin_onboarding_devices(request: Request) -> JSONResponse: supervisor._require_admin(request) return JSONResponse(supervisor._onboarding_devices_payload()) + @app.post("/admin/api/q7/migration-credentials") + async def admin_q7_migration_credentials(request: Request) -> JSONResponse: + """Reserve local MQTT credentials without requiring a Q7 flash dump.""" + supervisor._require_admin(request) + try: + body = await request.json() + except json.JSONDecodeError: + return JSONResponse({"error": "Invalid JSON body"}, status_code=400) + if not isinstance(body, dict): + return JSONResponse({"error": "JSON body must be an object"}, status_code=400) + did = str(body.get("did") or "").strip() + duid = str(body.get("duid") or "").strip() + try: + credentials = supervisor.runtime_credentials.prepare_q7_migration_credentials( + did=did, + duid=duid, + ) + except KeyError: + return JSONResponse({"error": "Q7 device was not found"}, status_code=404) + except ValueError as exc: + return JSONResponse({"error": str(exc)}, status_code=400) + return JSONResponse(credentials, headers={"Cache-Control": "no-store"}) + @app.post("/admin/api/onboarding/sessions") async def admin_onboarding_start(request: Request) -> JSONResponse: supervisor._require_admin(request) diff --git a/tests/test_mqtt_tls_proxy.py b/tests/test_mqtt_tls_proxy.py index 73bcc39..03bcd77 100644 --- a/tests/test_mqtt_tls_proxy.py +++ b/tests/test_mqtt_tls_proxy.py @@ -179,10 +179,18 @@ def fake_extract_packets(frame_buf: bytearray) -> list[bytes]: frame_buf.clear() return [data] - def slow_trace_packet(conn_id: str, direction: str, packet: bytes) -> None: + def slow_trace_packet( + conn_id: str, + direction: str, + packet: bytes, + authenticated_username: str = "", + device_credentials_verified: bool = False, + ) -> None: assert conn_id == "1" assert direction == "c2b" assert packet == b"packet-bytes" + assert authenticated_username == "" + assert device_credentials_verified is False trace_started.set() time.sleep(0.25) trace_finished.set() diff --git a/tests/test_q7_migration_credentials.py b/tests/test_q7_migration_credentials.py new file mode 100644 index 0000000..b651fcf --- /dev/null +++ b/tests/test_q7_migration_credentials.py @@ -0,0 +1,339 @@ +"""Admin-only, dump-free Q7 MQTT credential preparation.""" + +import json +import hashlib +import logging +import re +from pathlib import Path + +from fastapi.testclient import TestClient +import httpx + +from conftest import write_release_config +from scripts import q7_migration_ota_builder, q7_owner_ota, q7_prepare_migration +from scripts.q7_migration_ota_builder import inspect +from roborock_local_server.bundled_backend.mqtt_broker_server.topic_bridge import ( + CloudTopicKey, + DeviceTopicKey, + MqttTopicBridge, +) +from roborock_local_server.config import load_config, resolve_paths +from roborock_local_server.server import ReleaseSupervisor +from roborock_local_server.bundled_backend.mqtt_tls_proxy_server.server import MqttTlsProxy + + +def _client(tmp_path: Path, *, model: str = "roborock.vacuum.sc05", + did: str = "", + localkey: str = "0123456789abcdef", + local_key_source: str = "inventory_cloud", + mqtt_usr: str = "", + split_cloud_key: str | None = None) -> tuple[TestClient, ReleaseSupervisor]: + config_file = write_release_config(tmp_path) + config = load_config(config_file) + paths = resolve_paths(config_file, config) + paths.runtime_credentials_path.parent.mkdir(parents=True, exist_ok=True) + devices = [{ + "did": did, + "duid": "synthetic-q7-duid", + "model": model, + "localkey": localkey, + "local_key_source": local_key_source, + "device_mqtt_usr": mqtt_usr, + }] + if split_cloud_key is not None: + devices.append({ + "did": "", + "duid": "separate-cloud-duid", + "model": "roborock.vacuum.sc05", + "localkey": split_cloud_key, + "local_key_source": "inventory_cloud", + }) + paths.runtime_credentials_path.write_text( + json.dumps({"schema_version": 2, "devices": devices}) + "\n", + encoding="utf-8", + ) + supervisor = ReleaseSupervisor(config=config, paths=paths) + return TestClient(supervisor.app), supervisor + + +def _login(client: TestClient) -> None: + response = client.post("/admin/api/login", json={"password": "correct horse battery staple"}) + assert response.status_code == 200 + + +def _publish_packet(topic: str) -> bytes: + encoded = topic.encode("ascii") + body = len(encoded).to_bytes(2, "big") + encoded + b"{}" + assert len(body) < 128 + return bytes((0x30, len(body))) + body + + +def test_no_dump_owner_handoff_formats_agree_across_server_and_tools(tmp_path: Path, monkeypatch) -> None: + """One synthetic owner can prepare, build, and preflight a URL update.""" + app_client, _supervisor = _client(tmp_path) + + def handler(request: httpx.Request) -> httpx.Response: + response = app_client.request( + request.method, request.url.path, content=request.content, + headers={"content-type": "application/json"}, + ) + return httpx.Response(response.status_code, content=response.content, headers=response.headers) + + original_client = httpx.Client + monkeypatch.setattr( + q7_prepare_migration.httpx, "Client", + lambda **kwargs: original_client(transport=httpx.MockTransport(handler), **kwargs), + ) + config_path = tmp_path / "private" / "migration.json" + q7_prepare_migration.prepare( + server="https://local.test:555", duid="synthetic-q7-duid", + api_url="https://local.test:555", mqtt_url="ssl://local.test:8881", + out=config_path, admin_password="correct horse battery staple", + ) + config = json.loads(config_path.read_text(encoding="utf-8")) + profile = tmp_path / "profile" + profile.mkdir() + blobs = { + "ota-key.bin": b"synthetic-key-12", + "return.sh": b"#!/bin/sh\necho normal-boot\n", + "editor.sh": b"#!/bin/sh\necho edit-existing-iot\n", + } + hashes = {name: hashlib.sha256(blob).hexdigest() for name, blob in blobs.items()} + for name, blob in blobs.items(): + (profile / name).write_bytes(blob) + (profile / "manifest.json").write_text(json.dumps({ + "schema": q7_migration_ota_builder.PROFILE_030180_SCHEMA, + "model": "roborock.vacuum.sc05", + "firmware_version": "03.01.80", + "file_sha256": hashes, + })) + monkeypatch.setattr(q7_migration_ota_builder, "PROFILE_030180_HASHES", hashes) + artifact = tmp_path / "candidate" + metadata = q7_migration_ota_builder.build(config, profile, artifact) + migration = inspect(artifact)[1] + assert migration["target_firmware"] == "03.01.80" + assert metadata["preflight"] == migration["preflight"] + assert q7_owner_ota.cloud_key_matches_server_import( + migration["preflight"], duid="synthetic-q7-duid", local_key="0123456789abcdef" + ) + assert not q7_owner_ota.cloud_key_matches_server_import( + migration["preflight"], duid="synthetic-q7-duid", local_key="fedcba9876543210" + ) + + +def test_q7_migration_credentials_are_admin_only_idempotent_and_persisted(tmp_path: Path, monkeypatch) -> None: + client, supervisor = _client(tmp_path) + route = "/admin/api/q7/migration-credentials" + request = {"duid": "synthetic-q7-duid"} + assert client.post(route, json=request).status_code == 401 + _login(client) + + first = client.post(route, json=request) + assert first.status_code == 200 + body = first.json() + assert body["did"] == "" and body["duid"] == request["duid"] + assert "localkey" not in body + assert body["local_key_sha256"] == hashlib.sha256(b"0123456789abcdef").hexdigest() + for key, size in (("mqtt_clientid", 16), ("mqtt_usr", 16), ("mqtt_passwd", 32)): + assert re.fullmatch(rf"[0-9a-f]{{{size}}}", body[key]) + + second = client.post(route, json=request) + assert second.status_code == 200 + assert {key: second.json()[key] for key in ("mqtt_clientid", "mqtt_usr", "mqtt_passwd")} == { + key: body[key] for key in ("mqtt_clientid", "mqtt_usr", "mqtt_passwd") + } + stored = supervisor.runtime_credentials.resolve_device(duid=request["duid"]) + assert stored is not None + assert stored["migration_mqtt_clientid"] == body["mqtt_clientid"] + assert stored["device_mqtt_usr"] == body["mqtt_usr"] + assert stored["device_mqtt_pass"] == body["mqtt_passwd"] + authorized, reason, _device = supervisor.runtime_credentials.verify_device_mqtt_credentials( + username=body["mqtt_usr"], password=body["mqtt_passwd"] + ) + assert authorized and reason == "device_mqtt_user" + + bridge = MqttTopicBridge( + host="127.0.0.1", port=1883, + logger=logging.getLogger("test.q7_migration_bridge"), + runtime_credentials=supervisor.runtime_credentials, + runtime_state=supervisor.runtime_state, + inventory_path=supervisor.paths.inventory_path, + ) + supervisor.paths.inventory_path.write_text( + json.dumps({"devices": [{"duid": request["duid"], "model": "roborock.vacuum.sc05"}]}), + encoding="utf-8", + ) + # Merely opening the admin vacuum list must not erase the cloud key's + # provenance and silently disable migration routing. + assert client.get("/admin/api/vacuums").status_code == 200 + assert supervisor.runtime_credentials.resolve_device(duid=request["duid"])["local_key_source"] == "inventory_cloud" + monkeypatch.setattr(supervisor.runtime_state, "key_models_by_did", lambda: { + "incorrect-model-inferred-did": "roborock.vacuum.sc05" + }) + other_device = DeviceTopicKey(did="unrelated-did", mqtt_usr="unrelated-user") + migrated_device = DeviceTopicKey(did="1234567890123", mqtt_usr=body["mqtt_usr"]) + cloud_topic = CloudTopicKey(rriot_u="app-user", mqtt_username="app-mqtt-user", duid=request["duid"]) + bridge._remember_device_seen(other_device) + bridge._remember_device_seen(migrated_device) + assert bridge._resolve_device_for_cloud(cloud_topic) is None + assert request["duid"] not in bridge._duid_to_did + + supervisor.runtime_credentials.record_mqtt_topic(topic=migrated_device.topic_out, direction="b2c") + assert supervisor.runtime_credentials.resolve_device(duid=request["duid"])["did"] == "" + supervisor.runtime_credentials.record_mqtt_topic(topic=migrated_device.topic_in, direction="c2b") + assert supervisor.runtime_credentials.resolve_device(duid=request["duid"])["did"] == "" + supervisor.runtime_credentials.record_mqtt_topic( + topic=migrated_device.topic_in, + direction="c2b", + authenticated_username="another-device", + device_credentials_verified=True, + ) + assert supervisor.runtime_credentials.resolve_device(duid=request["duid"])["did"] == "" + supervisor.runtime_credentials.record_mqtt_topic( + topic=migrated_device.topic_in, + direction="c2b", + authenticated_username=body["mqtt_usr"], + device_credentials_verified=True, + ) + linked = supervisor.runtime_credentials.resolve_device(duid=request["duid"]) + assert linked is not None and linked["did"] == migrated_device.did + assert linked["migration_did_verified"] == "1" + bridge._last_duid_map_refresh_monotonic = 0.0 + assert bridge._resolve_device_for_cloud(cloud_topic) == migrated_device + assert len([item for item in supervisor.runtime_credentials.devices() if item["did"] == migrated_device.did]) == 1 + monkeypatch.setattr(supervisor.runtime_credentials, "_load_key_models_by_did", lambda: { + "incorrect-model-inferred-did": "roborock.vacuum.sc05" + }) + supervisor.runtime_credentials.sync_inventory() + assert supervisor.runtime_credentials.resolve_device(duid=request["duid"])["did"] == migrated_device.did + + +def test_q7_migration_credentials_reject_wrong_identity_or_existing_credentials(tmp_path: Path) -> None: + client, _supervisor = _client(tmp_path, mqtt_usr="existing-device-user") + _login(client) + route = "/admin/api/q7/migration-credentials" + assert client.post(route, json={"duid": "wrong-duid"}).status_code == 404 + assert client.post(route, json={"did": "unknown", "duid": "synthetic-q7-duid"}).status_code == 400 + blocked = client.post(route, json={"duid": "synthetic-q7-duid"}) + assert blocked.status_code == 400 + assert "non-migration" in blocked.json()["error"] + + +def test_q7_authenticated_duid_topic_replaces_stale_numeric_did(tmp_path: Path) -> None: + client, supervisor = _client(tmp_path, did="1234567890123") + _login(client) + response = client.post("/admin/api/q7/migration-credentials", json={"duid": "synthetic-q7-duid"}) + assert response.status_code == 200 + username = response.json()["mqtt_usr"] + store = supervisor.runtime_credentials + + def publish(did: str) -> None: + store.record_mqtt_topic( + topic=f"rr/d/i/{did}/{username}", direction="c2b", + authenticated_username=username, device_credentials_verified=True, + ) + + publish("1234567890123") + assert store.verified_q7_migration_links() == {"synthetic-q7-duid": "1234567890123"} + publish("unrelated-topic-id") + assert store.verified_q7_migration_links() == {"synthetic-q7-duid": "1234567890123"} + # Recover records written by releases where an admin inventory read + # downgraded source provenance after the migration was reserved. + store._devices[0]["local_key_source"] = "inventory" + store._save_locked() + publish("synthetic-q7-duid") + assert store.verified_q7_migration_links() == {"synthetic-q7-duid": "synthetic-q7-duid"} + assert store.resolve_device(duid="synthetic-q7-duid")["local_key_source"] == "inventory_cloud" + + bridge = MqttTopicBridge( + host="127.0.0.1", port=1883, logger=logging.getLogger("test.q7_duid_rebind"), + runtime_credentials=store, runtime_state=supervisor.runtime_state, + inventory_path=supervisor.paths.inventory_path, + ) + observed = DeviceTopicKey(did="synthetic-q7-duid", mqtt_usr=username) + cloud = CloudTopicKey(rriot_u="owner", mqtt_username="owner-mqtt", duid="synthetic-q7-duid") + bridge._remember_device_seen(observed) + assert bridge._resolve_device_for_cloud(cloud) == observed + + +def test_q7_migration_credentials_require_model_and_cloud_key(tmp_path: Path) -> None: + route = "/admin/api/q7/migration-credentials" + request = {"duid": "synthetic-q7-duid"} + for model, localkey, source in ( + ("roborock.vacuum.a15", "0123456789abcdef", "inventory_cloud"), + ("roborock.vacuum.sc05", "short", "inventory_cloud"), + ("roborock.vacuum.sc05", "0123456789abcdef", "server_assigned"), + ): + case = tmp_path / (model.rsplit(".", 1)[-1] + "-" + localkey + "-" + source) + case.mkdir() + client, _supervisor = _client(case, model=model, localkey=localkey, local_key_source=source) + _login(client) + expected = 400 if localkey == "short" else 404 + assert client.post(route, json=request).status_code == expected + + +def test_q7_migration_credentials_accept_unique_split_cloud_record(tmp_path: Path) -> None: + client, _supervisor = _client( + tmp_path, + model="", + did="1234567890123", + local_key_source="b01_nc", + split_cloud_key="0123456789abcdef", + ) + _login(client) + response = client.post( + "/admin/api/q7/migration-credentials", + json={"duid": "separate-cloud-duid"}, + ) + assert response.status_code == 200 + assert len(response.json()["mqtt_usr"]) == 16 + + +def test_q7_migration_credentials_reject_unmatched_split_cloud_key(tmp_path: Path) -> None: + client, _supervisor = _client( + tmp_path, + model="", + did="1234567890123", + local_key_source="b01_nc", + split_cloud_key="fedcba9876543210", + ) + _login(client) + response = client.post( + "/admin/api/q7/migration-credentials", + json={"did": "1234567890123", "duid": "separate-cloud-duid"}, + ) + assert response.status_code == 400 + assert "not linked" in response.json()["error"] + + +def test_q7_proxy_links_only_the_authenticated_device_topic(tmp_path: Path) -> None: + client, supervisor = _client(tmp_path) + _login(client) + response = client.post( + "/admin/api/q7/migration-credentials", + json={"duid": "synthetic-q7-duid"}, + ) + assert response.status_code == 200 + username = response.json()["mqtt_usr"] + proxy = MqttTlsProxy( + cert_file=tmp_path / "cert.pem", + key_file=tmp_path / "key.pem", + listen_host="127.0.0.1", + listen_port=8883, + backend_host="127.0.0.1", + backend_port=1883, + localkey="0123456789abcdef", + logger=logging.getLogger("test.q7_proxy_auth"), + decoded_jsonl=tmp_path / "mqtt.jsonl", + runtime_credentials=supervisor.runtime_credentials, + ) + topic = f"rr/d/i/1234567890123/{username}" + packet = _publish_packet(topic) + proxy._trace_packet("1", "b2c", packet, username, True) + proxy._trace_packet("1", "c2b", packet, "other-user", True) + proxy._trace_packet("1", "c2b", packet, username, False) + assert supervisor.runtime_credentials.verified_q7_migration_links() == {} + proxy._trace_packet("1", "c2b", packet, username, True) + assert supervisor.runtime_credentials.verified_q7_migration_links() == { + "synthetic-q7-duid": "1234567890123" + } diff --git a/tests/test_q7_migration_ota_builder.py b/tests/test_q7_migration_ota_builder.py new file mode 100644 index 0000000..dd14ecf --- /dev/null +++ b/tests/test_q7_migration_ota_builder.py @@ -0,0 +1,193 @@ +"""Portable Q7 profile builds a staged, script-only five-field package.""" + +import gzip +import json +import os +from pathlib import Path +import shutil +import struct +import subprocess +import sys + +from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes +import pytest + +from scripts import q7_migration_ota_builder as builder +from scripts.q7_migration_ota_builder import inspect + + +FIELDS = { + "api_url": "https://local.test:555", + "mqtt_url": "ssl://local.test:8881", + "mqtt_clientid": "0123456789abcdef", + "mqtt_usr": "abcdef0123456789", + "mqtt_passwd": "0123456789abcdef0123456789abcdef", +} + + +def _fixture_profile(path: Path, monkeypatch: pytest.MonkeyPatch) -> bytes: + key = b"synthetic-key-12" + files = { + "ota-key.bin": key, + "return.sh": b"#!/bin/sh\necho normal-boot\n", + "editor.sh": b"#!/bin/sh\necho edit-existing-iot\n", + } + path.mkdir() + hashes = {name: builder._sha256(blob) for name, blob in files.items()} + monkeypatch.setattr(builder, "PROFILE_HASHES", hashes) + for name, blob in files.items(): + (path / name).write_bytes(blob) + (path / "manifest.json").write_text(json.dumps({ + "schema": builder.SCHEMA, + "model": builder.MODEL, + "firmware_version": builder.VERSION, + "file_sha256": hashes, + })) + return key + + +def test_portable_package_is_stagable_and_contains_only_scripts( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile = tmp_path / "profile" + key = _fixture_profile(profile, monkeypatch) + out = tmp_path / "candidate" + metadata = builder.build(FIELDS, profile, out) + payload, details = inspect(out) + assert details["encrypted_sha256"] == metadata["encrypted_sha256"] + assert metadata["build_mode"] == "portable_profile" + assert metadata["signed"] is False + assert metadata["config_sha256"] == builder._sha256( + json.dumps(FIELDS, sort_keys=True, separators=(",", ":")).encode("ascii") + ) + + decryptor = Cipher(algorithms.AES(key), modes.ECB()).decryptor() + padded = decryptor.update(payload) + decryptor.finalize() + pad = padded[-1] + assert padded[-pad:] == bytes([pad]) * pad + container = gzip.decompress(padded[:-pad]) + checksum, major, minor, blocks, payload_size, begin_size, end_size = struct.unpack_from( + "<7I", container + ) + assert checksum == sum(container[4:]) + assert (major, minor, blocks, payload_size) == (0, 3, 0, 0) + assert len(container) == 28 + begin_size + end_size + begin = container[28 : 28 + begin_size] + end = container[28 + begin_size :] + assert begin.startswith(b"#!/bin/sh\nset -- ") + assert b"/userdata/rriot/data_dir/iot.json" in begin + assert all(value.encode("ascii") in begin for value in FIELDS.values()) + assert begin.endswith(b"echo edit-existing-iot\n") + assert end == b"#!/bin/sh\necho normal-boot\n" + +def test_portable_package_pins_cloud_key_without_embedding_it( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile = tmp_path / "profile" + key = _fixture_profile(profile, monkeypatch) + fingerprint = { + "duid_sha256": "1" * 64, + "local_key_sha256": "2" * 64, + } + out = tmp_path / "candidate" + metadata = builder.build({**FIELDS, "_preflight": fingerprint}, profile, out) + assert metadata["preflight"] == fingerprint + assert inspect(out)[1]["preflight"] == fingerprint + payload = inspect(out)[0] + legacy = tmp_path / "legacy" + builder.build(FIELDS, profile, legacy) + assert payload == inspect(legacy)[0] + padded = Cipher(algorithms.AES(key), modes.ECB()).decryptor().update(payload) + container = gzip.decompress(padded[:-padded[-1]]) + assert b"duid_sha256" not in container and b"local_key_sha256" not in container + with pytest.raises(ValueError, match="preflight"): + builder.build({**FIELDS, "_preflight": {"local_key_sha256": "2" * 64}}, profile, tmp_path / "bad") + + +def test_refuses_unsafe_values_and_tampered_profile( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile = tmp_path / "profile" + _fixture_profile(profile, monkeypatch) + out = tmp_path / "candidate" + with pytest.raises(ValueError, match="api_url"): + builder.build({**FIELDS, "api_url": "https://local.test';touch /tmp/pwn"}, profile, out) + assert not out.exists() + (profile / "return.sh").write_bytes(b"#!/bin/sh\necho changed\n") + with pytest.raises(ValueError, match="pinned hash"): + builder.build(FIELDS, profile, out) + assert not out.exists() + + +def test_accepts_pinned_reentry_profile(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + profile = tmp_path / "profile" + _fixture_profile(profile, monkeypatch) + manifest_path = profile / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["schema"] = builder.REENTRY_SCHEMA + manifest_path.write_text(json.dumps(manifest)) + monkeypatch.setattr(builder, "REENTRY_PROFILE_HASHES", manifest["file_sha256"]) + metadata = builder.build(FIELDS, profile, tmp_path / "candidate") + assert metadata["profile_schema"] == builder.REENTRY_SCHEMA + assert builder._sha256(Path(builder.__file__).with_name("q7_iot_local_fields_reentry.sh").read_bytes()) == ( + "2fce38f7ba828b361fa644c5e8fea8c5858f9bc0eeaeab931610e18e6da3fb04" + ) + + +def test_030180_profile_requires_its_own_return_script_and_version( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + profile = tmp_path / "profile" + _fixture_profile(profile, monkeypatch) + manifest_path = profile / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["schema"] = builder.PROFILE_030180_SCHEMA + manifest["firmware_version"] = "03.01.80" + monkeypatch.setattr(builder, "PROFILE_030180_HASHES", manifest["file_sha256"]) + manifest_path.write_text(json.dumps(manifest)) + candidate = tmp_path / "candidate" + metadata = builder.build(FIELDS, profile, candidate) + assert metadata["target_firmware"] == "03.01.80" + _, details = inspect(candidate) + assert details["target_firmware"] == "03.01.80" + manifest["firmware_version"] = "03.01.74" + manifest_path.write_text(json.dumps(manifest)) + with pytest.raises(ValueError, match="inspected Q7 build"): + builder.build(FIELDS, profile, tmp_path / "wrong-version") + + +@pytest.mark.skipif(sys.platform == "win32" or not shutil.which("sh"), reason="requires POSIX shell") +def test_reentry_editor_preserves_matching_backup_and_refuses_mismatch(tmp_path: Path) -> None: + script = Path(builder.__file__).with_name("q7_iot_local_fields_reentry.sh") + wrapper = tmp_path / "busybox" + wrapper.write_text('#!/bin/sh\nexec "$@"\n') + wrapper.chmod(0o755) + source = { + "api_url": "https://vendor.test", + "mqtt_url": "ssl://vendor.test:8883", + "mqtt_clientid": "fedcba9876543210", + "mqtt_usr": "1234567890abcdef", + "mqtt_passwd": "fedcba9876543210fedcba9876543210", + "duid": "leave-this-alone", + } + original = (json.dumps(source, indent=2) + "\n").encode() + iot = tmp_path / "iot.json" + backup = tmp_path / "iot.json.before-q7-local-edit" + env = {**os.environ, "Q7_BUSYBOX": str(wrapper)} + command = ["sh", str(script), str(iot), *FIELDS.values()] + + iot.write_bytes(original) + assert subprocess.run(command, env=env, capture_output=True).returncode == 0 + assert backup.read_bytes() == original + assert json.loads(iot.read_bytes())["duid"] == source["duid"] + + iot.write_bytes(original) + assert subprocess.run(command, env=env, capture_output=True).returncode == 0 + assert backup.read_bytes() == original + assert all(json.loads(iot.read_bytes())[name] == value for name, value in FIELDS.items()) + + iot.write_bytes(original) + backup.write_bytes(b"mismatch") + assert subprocess.run(command, env=env, capture_output=True).returncode != 0 + assert iot.read_bytes() == original + assert backup.read_bytes() == b"mismatch" diff --git a/tests/test_q7_owner_ota.py b/tests/test_q7_owner_ota.py new file mode 100644 index 0000000..63af6d1 --- /dev/null +++ b/tests/test_q7_owner_ota.py @@ -0,0 +1,156 @@ +"""Owner OTA sender must pin the hosted bytes before forming the command.""" + +import argparse +import asyncio +import hashlib +from io import BytesIO +import json +import os +from pathlib import Path +import stat +from types import SimpleNamespace + +import pytest +from roborock.data import UserData + +from scripts import q7_owner_ota + + +def _artifact(path: Path) -> bytes: + payload = bytes(range(16)) + path.mkdir() + (path / "q7-migration-v03.bin.gz.aes").write_bytes(payload) + (path / "metadata.json").write_text(json.dumps({ + "firmware": "roborock.vacuum.sc05 03.01.74, pinned inspected firmware profile", + "target_firmware": "03.01.74", + "package": "q7-migration-v03.bin.gz.aes", + "encrypted_size_bytes": len(payload), + "encrypted_sha256": hashlib.sha256(payload).hexdigest(), + "encrypted_md5": hashlib.md5(payload).hexdigest(), + "signed": False, + })) + return payload + + +def test_owner_request_uses_exact_hosted_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + artifact = tmp_path / "artifact" + payload = _artifact(artifact) + monkeypatch.setattr(q7_owner_ota, "urlopen", lambda _url, timeout: BytesIO(payload)) + request = q7_owner_ota.package_request(artifact, "http://192.0.2.1/update") + assert request == { + "packageUrl": "http://192.0.2.1/update", + "md5": hashlib.md5(payload).hexdigest(), + "packageSize": "16", + "signed": False, + "packageType": "robot", + } + + +def test_owner_request_rejects_wrong_hosted_bytes( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + artifact = tmp_path / "artifact" + _artifact(artifact) + monkeypatch.setattr(q7_owner_ota, "urlopen", lambda _url, timeout: BytesIO(b"wrong")) + with pytest.raises(ValueError, match="Host"): + q7_owner_ota.package_request(artifact, "http://192.0.2.1/update") + with pytest.raises(ValueError, match="loopback"): + q7_owner_ota.package_request(artifact, "http://127.0.0.1/update") + + +def test_cloud_key_preflight_rejects_stale_server_import() -> None: + duid = "current-cloud-duid" + key = "0123456789abcdef" + pinned = { + "duid_sha256": hashlib.sha256(duid.encode()).hexdigest(), + "local_key_sha256": hashlib.sha256(key.encode()).hexdigest(), + } + assert q7_owner_ota.cloud_key_matches_server_import(pinned, duid=duid, local_key=key) + assert not q7_owner_ota.cloud_key_matches_server_import(pinned, duid=duid, local_key="fedcba9876543210") + assert not q7_owner_ota.cloud_key_matches_server_import(pinned, duid="different-duid", local_key=key) + assert not q7_owner_ota.cloud_key_matches_server_import(None, duid=duid, local_key=key) + assert not q7_owner_ota.cloud_key_matches_server_import( + {**pinned, "local_key_sha256": "é" * 64}, duid=duid, local_key=key + ) + + +def test_email_code_list_creates_reusable_private_account_export( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + owner = UserData.from_dict({ + "rriot": {"u": "mqtt-user", "s": "mqtt-secret", "h": "mqtt-host", "k": "mqtt-key", "r": {}}, + "token": "private-token", + }) + home = SimpleNamespace( + products=[ + SimpleNamespace(id="q7-product", model="roborock.vacuum.sc05"), + SimpleNamespace(id="other-product", model="roborock.vacuum.a15"), + ], + devices=[ + SimpleNamespace(duid="q7-cloud-duid", product_id="q7-product", fv="03.01.80", online=True, + local_key="0123456789abcdef"), + SimpleNamespace(duid="other-duid", product_id="other-product", fv="01.00.00", online=True, + local_key="fedcba9876543210"), + ], + ) + events: list[str] = [] + + class FakeSession: + async def __aenter__(self): + return self + + async def __aexit__(self, *_args): + return None + + class FakeApi: + def __init__(self, email: str, *, base_url: str | None = None, session: object): + events.append(f"api:{email}:{base_url or 'discover'}") + assert isinstance(session, FakeSession) + + @property + async def base_url(self) -> str: + return "https://owner.example.test" + + async def request_code_v4(self) -> None: + events.append("request_code") + + async def code_login_v4(self, code: str) -> UserData: + events.append(f"code_login:{code}") + return owner + + async def get_home_data_v3(self, user_data: UserData): + assert user_data.rriot.u == "mqtt-user" + events.append("home_data") + return home + + monkeypatch.setattr(q7_owner_ota.aiohttp, "ClientSession", FakeSession) + monkeypatch.setattr(q7_owner_ota, "RoborockApiClient", FakeApi) + monkeypatch.setattr("builtins.input", lambda _prompt: " 123456 ") + monkeypatch.setattr(q7_owner_ota, "create_mqtt_session", lambda _params: pytest.fail("list contacted MQTT")) + + export = tmp_path / "private" / "owner-account.json" + args = argparse.Namespace(email="owner@example.test", account=None, save_account=export, list=True) + result = asyncio.run(q7_owner_ota.run(args)) + assert result == { + "devices": [{ + "duid": "q7-cloud-duid", "firmware": "03.01.80", "cloud_online": True, + "local_key_length": 16, + }], + "command_sent": False, + } + saved = json.loads(export.read_text(encoding="utf-8")) + assert saved["username"] == "owner@example.test" + assert saved["base_url"] == "https://owner.example.test" + assert UserData.from_dict(saved["user_data"]).token == "private-token" + if os.name != "nt": + assert stat.S_IMODE(export.stat().st_mode) == 0o600 + assert events == ["api:owner@example.test:discover", "request_code", "code_login:123456", "home_data"] + + args.email, args.account, args.save_account = None, export, None + assert asyncio.run(q7_owner_ota.run(args)) == result + assert events[-2:] == ["api:owner@example.test:https://owner.example.test", "home_data"] + + args.email, args.account, args.save_account = "owner@example.test", None, export + with pytest.raises(FileExistsError, match="already exists"): + asyncio.run(q7_owner_ota.run(args)) + assert events.count("request_code") == 1 diff --git a/tests/test_q7_prepare_migration.py b/tests/test_q7_prepare_migration.py new file mode 100644 index 0000000..863d629 --- /dev/null +++ b/tests/test_q7_prepare_migration.py @@ -0,0 +1,112 @@ +"""The Q7 migration manifest preparer never contacts a robot.""" + +import json +import hashlib +from pathlib import Path + +import httpx +import pytest + +from scripts import q7_prepare_migration as prep + + +def _prepare(tmp_path: Path, **overrides: object) -> dict[str, str]: + arguments = { + "server": "https://local.example:555", + "duid": "synthetic-duid", + "api_url": "https://local.example:555", + "mqtt_url": "ssl://local.example:8883", + "out": tmp_path / "private" / "migration.json", + "admin_password": "synthetic-admin-password", + } + arguments.update(overrides) + return prep.prepare(**arguments) + + +def test_bad_target_urls_are_rejected_before_server_contact(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + def unexpected_client(**_kwargs: object) -> None: + raise AssertionError("Invalid target URL must not reserve credentials") + + monkeypatch.setattr(prep.httpx, "Client", unexpected_client) + for field, value in ( + ("api_url", "http://local.example:555"), + ("api_url", "https://local.example:bad"), + ("mqtt_url", "tcp://local.example:8883"), + ("mqtt_url", "ssl://local.example:8883;touch-bad"), + ): + with pytest.raises(ValueError, match=field): + _prepare(tmp_path, **{field: value}) + with pytest.raises(ValueError, match="--server"): + _prepare(tmp_path, server="https://local.example:bad") + + +def test_preparer_writes_private_manifest_with_cloud_key_pin_without_printing_credentials( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + requests: list[tuple[str, object]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + data = json.loads(request.content) + requests.append((request.url.path, data)) + if request.url.path == "/admin/api/login": + return httpx.Response(200, json={"ok": True}) + assert request.url.path == "/admin/api/q7/migration-credentials" + return httpx.Response(200, json={ + "did": "", + "duid": "synthetic-duid", + "mqtt_clientid": "1" * 16, + "mqtt_usr": "2" * 16, + "mqtt_passwd": "3" * 32, + "local_key_sha256": hashlib.sha256(b"0123456789abcdef").hexdigest(), + }) + + original_client = httpx.Client + monkeypatch.setattr( + prep.httpx, "Client", + lambda **kwargs: original_client(transport=httpx.MockTransport(handler), **kwargs), + ) + result = _prepare(tmp_path) + manifest = tmp_path / "private" / "migration.json" + assert result["manifest"] == str(manifest.resolve()) + assert json.loads(manifest.read_text(encoding="utf-8")) == { + "api_url": "https://local.example:555", + "mqtt_url": "ssl://local.example:8883", + "mqtt_clientid": "1" * 16, + "mqtt_usr": "2" * 16, + "mqtt_passwd": "3" * 32, + "_preflight": { + "duid_sha256": hashlib.sha256(b"synthetic-duid").hexdigest(), + "local_key_sha256": hashlib.sha256(b"0123456789abcdef").hexdigest(), + }, + } + assert requests == [ + ("/admin/api/login", {"password": "synthetic-admin-password"}), + ("/admin/api/q7/migration-credentials", {"did": "", "duid": "synthetic-duid"}), + ] + assert capsys.readouterr().out == "" + with pytest.raises(FileExistsError): + _prepare(tmp_path) + assert len(requests) == 2 + + +def test_preparer_refuses_server_without_key_fingerprint( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path == "/admin/api/login": + return httpx.Response(200, json={"ok": True}) + return httpx.Response(200, json={ + "duid": "synthetic-duid", + "mqtt_clientid": "1" * 16, + "mqtt_usr": "2" * 16, + "mqtt_passwd": "3" * 32, + }) + + original_client = httpx.Client + monkeypatch.setattr( + prep.httpx, "Client", + lambda **kwargs: original_client(transport=httpx.MockTransport(handler), **kwargs), + ) + with pytest.raises(ValueError, match="fingerprint"): + _prepare(tmp_path) + assert not (tmp_path / "private" / "migration.json").exists() diff --git a/tests/test_runtime_credentials.py b/tests/test_runtime_credentials.py index 106f19e..16cbe09 100644 --- a/tests/test_runtime_credentials.py +++ b/tests/test_runtime_credentials.py @@ -4,6 +4,15 @@ from roborock_local_server.bundled_backend.shared.runtime_credentials import RuntimeCredentialsStore +def test_device_topic_tracking_keeps_existing_inbound_behavior(tmp_path: Path) -> None: + store = RuntimeCredentialsStore(tmp_path / "runtime_credentials.json") + store.record_mqtt_topic(topic="rr/d/i/123456789/device-user", direction="c2b") + device = store.resolve_device(did="123456789") + assert device is not None and device["device_mqtt_usr"] == "device-user" + store.record_mqtt_topic(topic="rr/d/o/other-id/other-user", direction="b2c") + assert len(store.devices()) == 1 + + def test_ensure_device_merges_split_did_and_duid_records(tmp_path: Path) -> None: credentials_path = tmp_path / "runtime_credentials.json" credentials_path.write_text(