Onboarding Report: Saros 20 Sonic Complete

[Flash1232]

First of all: Thank you so much for drawing inspiration from Valetudo and having enough drive to provide a privacy-focused solution for vacuums from Roborock :)!

For my new vacuum (updated to FW v02.51.82 using original App with Internet connection), I tried the following:

I am using Traefik as a reverse proxy (handles Let's Encrypt certificate renewal) for both :555 (https -> http) and :8881 (tcp) to forward to the HA addon. This unfortunately leads to the source IP to be improperly attributed (see below, "<TRAEFIK_IP>" instead of "<VACUUM_IP>"), but I don't think this matters for onboarding (correct me if I'm wrong). HA addon gets fake certs so it doesn't complain about missing certs.

The admin server is running and GUI onboarding seems to be starting as it should. The robot says "connecting to Wifi" after I send the onboarding packet, but then it will just keep blinking and never properly say it's connected as it did before with the app pairing (maybe that's alright?).

The problem now is that it never captures any samples, baseline also stays at "0", pubkey shows "missing" and connected "no". The only thing that even tells me I perhaps did something right is that the live vacuums json object will (after like 3-4 mins into onboarding) go from

//...
"missing_steps": [
  "region",
  "nc_prepare"
],
//...

to

//...
"missing_steps": [
  "nc_prepare"
],
//...

and I can see the "/region" request in the logs. But apart from that, nothing will happen after that.

Also, to even get to that point I needed to change the "DUID" from the cloud import from some cryptic ID (alphanumeric "random" string) to the DID (so DID and DUID are equal). If I don't do that I get a "new" vacuum entry - during onboarding - in the admin overview that is vastly incomplete, with just the DID reported as the DUID, empty DID and empty almost everything:

{
	"duid": "110xxxxxxxxxx",
	"did": "",
	"id_kind": "",
	"name": "110xxxxxxxxxx",
	"local_key": "",
	"source": "",
	"key_model": "",
	"ips": [
	  "<TRAEFIK_IP>"
	],
	"connected": false,
	"last_ip": "<TRAEFIK_IP>",
	"last_http_at": "2026-06-04T17:35:34.006021+00:00",
	"last_http_route": "region",
	"last_http_path": "/region",
	"last_http_remote": "<TRAEFIK_IP>:40316",
	"last_http_host": "api-bla.example.com:555",
	"last_mqtt_at": "",
	"last_mqtt_topic": "",
	"last_mqtt_direction": "",
	"last_mqtt_payload_preview": "",
	"last_disconnect_at": "",
	"last_message_at": "2026-06-04T17:35:34.006021+00:00",
	"last_message_source": "http",
	"onboarding_steps": {
	  "region": "2026-06-04T17:35:34.006021+00:00"
	},
	"onboarding": {
	  "required_steps": [
	    "region",
	    "nc_prepare"
	  ],
	  "step_labels": {
	  "region": "Region",
	  "nc_prepare": "NC Prepare",
	  "login_key_sign": "Key Sign"
	},
	"missing_steps": [
  	  "nc_prepare"
	],
	"has_required_messages": false,
	"has_public_key": false,
	"public_key_ready": false,
	"status": "collecting_messages",
	"guidance": "Still waiting for onboarding messages: NC Prepare. The vacuum reached /region, but some models delay NC Prepare for a few minutes.",
	"key_state": {
	  "query_samples": 0,
      "header_samples": 0,
	  "max_signature_len": 0,
	  "has_modulus": false,
	  "recovery_state": "collecting",
	  "recovery_note": "Need at least 2 query signature samples (0 captured).",
	  "recovery_error": "",
	  "recovery_started_at": "",
	  "recovery_finished_at": ""
	  }
	},
	"model": "",
	"product_id": "",
	"inventory_source": ""
}

(it's also missing the "runtime_did" field which the imported complete object has.)

Of course, this lead to not even attributing any traffic to the imported vacuum during onboarding and what I did was to then just override the two inventory json files' DUID with the DID. Also, I had to remove the phantom vacuum from the runtime_credentials.json as the onboarding GUI would otherwise show "Runtime DID 110xxx... is already linked to DUID yyyyyy...; not reassigning it automatically."

As explained above, after rectifying all of that, I ended up with the Saros' entry being updated to the state of "only missing the nc_prepare step".

Now, this begs the question: is this already an indication of "Robot might not be supported"/ "new Firmware might have different behavior" or do you think I should provide some logs so this can be sorted out? I am unsure if the reverse proxy might also be culprit to some extent?

I am asking because it means I will have to manually censor vast amounts of configs/ output (which I am very happy to do!) but maybe this already rings a bell or the lack of "nc_prepare" call is already quite bad?

Thank you in advance!

[Lash-L]

Also, to even get to that point I needed to change the "DUID" from the cloud import from some cryptic ID (alphanumeric "random" string) to the DID (so DID and DUID are equal). If I don't do that I get a "new" vacuum entry - during onboarding - in the admin overview that is vastly incomplete, with just the DID reported as the DUID, empty DID and empty almost everything:

My guess is that this may be the problem. I'm not 100% sure but i just pushed a commit that I think might help.

It hitting region is a good sign. That means it trusts your cert which I think is the easiest fail condition. I'd have to see the response the server is sending to region and if there are any other calls or not.

Can you please give the latest main a try? If this does not work, we can dive deeper into logs

[Flash1232]

Thanks for your blazing fast response and fix for the did matching! That solves the necessity to correct the jsons for me.

Unfortunately, even after this fix, the problem of only receiving the "/region" call persists.

Here's the (slightly censored) excerpt from the decompiled_http.jsonl:

{
  "time": "2026-06-05T18:59:56.332220+00:00",
  "server": "api",
  "host": "api-rr.example.com:555",
  "method": "GET",
  "raw_path": "/.roborock.com/region?did=110xxxxxxxxxx&pid=roborock.vacuum.a279&region=rr.example.com:555/&session=S_TOKEN_ee978b89fc2720xxxxxxxxxxxxc19e9c&token=T_TOKEN_e39ecdcc8d6087xxxxxxxxxxxx00102f",
  "clean_path": "/region",
  "query": {
    "did": [
      "110xxxxxxxxxx"
    ],
    "pid": [
      "roborock.vacuum.a279"
    ],
    "region": [
      "rr.example.com:555/"
    ],
    "session": [
      "S_TOKEN_ee978b89fc2720xxxxxxxxxxxxc19e9c"
    ],
    "token": [
      "T_TOKEN_e39ecdcc8d6087xxxxxxxxxxxx00102f"
    ],
    "__host": [
      "api-rr.example.com:555"
    ]
  },
  "headers": {
    "host": "api-rr.example.com:555",
    "accept": "*/*",
    "n": "9277c0",
    "s": "UgmlT8Jp9QZOiywMaPiVenGrD1gBN0/NhD9...f4eW/Dn8=",
    "t": "1780685994",
    "v": "v2",
    "x-forwarded-for": "10.0.0.2",
    "x-forwarded-host": "api-rr.example.com:555",
    "x-forwarded-port": "555",
    "x-forwarded-proto": "https",
    "x-forwarded-server": "b35b769bcf1b",
    "x-real-ip": "10.0.0.2",
    "accept-encoding": "gzip"
  },
  "body_len": 0,
  "body_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
  "remote": "<TRAEFIK_IP>:57558",
  "did": "110xxxxxxxxxx",
  "pid": "roborock.vacuum.a279",
  "body_b64": "",
  "route": "region",
  "response_json": {
    "success": true,
    "code": 200,
    "msg": "success",
    "data": {
      "apiUrl": "https://api-rr.example.com:555",
      "mqttUrl": "ssl://api-rr.example.com:8881",
      "api_url": "https://api-rr.example.com:555",
      "mqtt_url": "ssl://api-rr.example.com:8881"
    },
    "result": {
      "apiUrl": "https://api-rr.example.com:555",
      "mqttUrl": "ssl://api-rr.example.com:8881",
      "api_url": "https://api-rr.example.com:555",
      "mqtt_url": "ssl://api-rr.example.com:8881"
    }
  }
}

There are no other calls made by the Vacuum. Most notably, there is no MQTT traffic, just this one HTTP call.

[Lash-L]

@Flash1232 just to make sure before i dive into this. All of the .example.coms are you redacting, correct?

[Lash-L]

Ah. there is no signature in the URL.

There is a "s" in the headers. I wonder if that is the signature. Is that field modified at all? Can you tell me the unredacted length of that key?

[Flash1232]

The whole "s" has a length of 684 (and it's always that length), I truncated it. So that's normally concatenated in the URL params?

[Lash-L]

Yes it is normally in the url as a query parameter.

Would you mind getting 3-4 full payloads and sharing with me privately? Understand if you would like to avoid doing that but I want to see if I can get the public key from the samples

[Flash1232]

If you don't mind: is there any way I could test this myself? I am currently trying to adjust server.py (extending the _pick_first_header with "s" (signature), "t" (ts) and "n" (nonce)). I will let you know of the results and of course will try to make it work with you :)

Perhaps I can just "drop" the samples somewhere (haven't really looked at where they're stored yet) so that would tell us if it's working with the existing logic?

Preliminary question: I can now see it picking up "header samples" (after adding the new header keys):

"key_state": {
        "query_samples": 0,
        "header_samples": 2,
        "max_signature_len": 512, // that's the 684 b64 decoded
        // ...
}

But it still says "requires at least 2 query samples", implying they're different things.

(If that's not working, I may share the samples with you privately)

[Lash-L]

So the tricky thing is that we need to figure out how the signature is made. Here is a script that may help. Point it at the decompiled_html.

#!/usr/bin/env python3
"""Recover a header-signed Roborock device's RSA public key from captured signatures.

Newer firmwares (e.g. roborock.vacuum.a279) put the request signature in the ``s``
header (with ``n`` nonce, ``t`` ts, ``v`` version) instead of a ``signature=`` URL
param. To answer ``/region`` with an *encrypted* bootstrap result the local server
needs the device's RSA public key. We can recover it without the private key:

    if M is the exact string the firmware signed and EM its PKCS#1 v1.5 encoding,
    then n divides (sig**e - EM) for every request, so
        n = gcd(sig0**e - EM0, sig1**e - EM1).

We don't know M (which fields, order, separators, hash...), so we try many guesses
and let the math reject the wrong ones -- only the right guess yields a clean,
self-verifying modulus.

    python recover_header_key.py data/runtime/decompiled_http.jsonl --did <did>

Capture 3-4 onboarding attempts first. Small exponents (3, 17) are instant; pass
--slow to also try 65537 (one GCD there is ~1-2 minutes, so it's off by default).
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import urllib.parse
from pathlib import Path

import gmpy2

# PKCS#1 v1.5 DigestInfo prefixes.
OID = {
    "sha1": bytes.fromhex("3021300906052b0e03021a05000414"),
    "sha256": bytes.fromhex("3031300d060960864801650304020105000420"),
    "sha512": bytes.fromhex("3051300d060960864801650304020305000440"),
}


def emsa(msg: str, key_bytes: int, alg: str, with_oid: bool) -> int:
    """PKCS#1 v1.5 encode msg -> integer (with_oid=False = raw padded hash)."""
    h = hashlib.new(alg, msg.encode()).digest()
    t = (OID[alg] + h) if with_oid else h
    em = b"\x00\x01" + b"\xff" * (key_bytes - len(t) - 3) + b"\x00" + t
    return int.from_bytes(em, "big")


def fields(rec: dict) -> dict:
    """Pull the signable pieces out of one decompiled_http.jsonl record."""
    raw_path = str(rec.get("raw_path") or "")
    path, _, query = raw_path.partition("?")
    pairs = urllib.parse.parse_qsl(query, keep_blank_values=True)
    h = rec.get("headers") or {}
    return {
        "method": str(rec.get("method") or "").upper(),
        "host": str(rec.get("host") or h.get("host") or ""),
        "path": path or str(rec.get("clean_path") or ""),
        "query": query,
        "query_sorted": "&".join(f"{k}={v}" for k, v in sorted(pairs)),
        "query_decoded": urllib.parse.unquote(query),
        "n": str(h.get("n") or ""),
        "t": str(h.get("t") or ""),
        "v": str(h.get("v") or ""),
        "body_sha256": str(rec.get("body_sha256") or hashlib.sha256(b"").hexdigest()),
    }


def candidates() -> list[tuple[str, "callable"]]:
    """(label, f(fields)->str) guesses for what the firmware signs. Add your own here."""
    out = []
    queries = ("query", "query_sorted", "query_decoded")
    for q in queries:
        out.append((q, lambda f, q=q: f[q]))
        out.append((f"path?{q}", lambda f, q=q: f"{f['path']}?{f[q]}"))
        for sep in ("\n", "&", ":"):
            name = sep.encode("unicode_escape").decode()
            out.append((f"M{name}path{name}{q}", lambda f, q=q, s=sep: s.join([f["method"], f["path"], f[q]])))
            out.append((f"{q}{name}n{name}t", lambda f, q=q, s=sep: s.join([f[q], f["n"], f["t"]])))
            out.append((f"{q}{name}n{name}t{name}v", lambda f, q=q, s=sep: s.join([f[q], f["n"], f["t"], f["v"]])))
        out.append((f"host+path?{q}", lambda f, q=q: f"{f['host']}{f['path']}?{f[q]}"))
        out.append((f"{q}+bodysha", lambda f, q=q: f"{f[q]}:{f['body_sha256']}"))
    return out


def recover(sigs: list[int], fld: list[dict], e: int, key_bytes: int, min_bits: int):
    """Try every format; return (n, label, alg, with_oid) on the first verified hit."""
    pe0, pe1 = gmpy2.mpz(sigs[0]) ** e, gmpy2.mpz(sigs[1]) ** e
    seen = set()
    for label, fn in candidates():
        for alg in ("sha256", "sha1", "sha512"):
            for with_oid in (True, False):
                try:
                    m0, m1 = fn(fld[0]), fn(fld[1])
                except Exception:
                    continue
                key = (alg, with_oid, m0, m1)
                if key in seen:
                    continue
                seen.add(key)
                em0 = emsa(m0, key_bytes, alg, with_oid)
                em1 = emsa(m1, key_bytes, alg, with_oid)
                g = gmpy2.gcd(pe0 - em0, pe1 - em1)
                if g.bit_length() < min_bits:
                    continue
                n = int(g)
                ems = [emsa(fn(f), key_bytes, alg, with_oid) for f in fld]
                for p in (2, 3, 5, 7, 11, 13):  # strip tiny gcd cofactor
                    while n % p == 0 and all(pow(s, e, n // p) == em for s, em in zip(sigs, ems)):
                        n //= p
                if all(pow(s, e, n) == em for s, em in zip(sigs, ems)):
                    return n, label, alg, with_oid
    return None


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("jsonl", nargs="+", help="decompiled_http.jsonl file(s)")
    ap.add_argument("--did", default="", help="only use samples for this device id")
    ap.add_argument("--slow", action="store_true", help="also try e=65537 (~1-2 min/format)")
    ap.add_argument("--min-bits", type=int, default=1024)
    args = ap.parse_args()

    sigs, fld = [], []
    seen_sig = set()
    for fp in args.jsonl:
        for line in Path(fp).read_text(encoding="utf-8").splitlines():
            try:
                rec = json.loads(line)
            except Exception:
                continue
            s = (rec.get("headers") or {}).get("s")
            if not s or (args.did and str(rec.get("did") or "") != args.did):
                continue
            raw = base64.b64decode(s)
            if len(raw) >= 64 and raw not in seen_sig:
                seen_sig.add(raw)
                sigs.append(int.from_bytes(raw, "big"))
                fld.append(fields(rec))

    if len(sigs) < 2:
        print(f"Need >=2 distinct header-signed samples, found {len(sigs)}. "
              "Trigger onboarding a few more times.")
        return 1
    key_bytes = (max(sigs).bit_length() + 7) // 8
    print(f"{len(sigs)} signatures, ~{key_bytes * 8}-bit key.")

    for e in ([3, 17, 65537] if args.slow else [3, 17]):
        print(f"trying e={e}{' (slow)' if e > 100 else ''}...")
        hit = recover(sigs, fld, e, key_bytes, args.min_bits)
        if hit:
            return report(hit, e, fld[0], key_bytes)
    print("\nNo match. Add guesses to candidates(), capture more samples, try --slow, "
          "or the scheme may use RSA-PSS (GCD can't recover that).")
    return 2


def report(hit, e, f0, key_bytes) -> int:
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric import rsa

    n, label, alg, with_oid = hit
    print("\n=== RECOVERED PUBLIC KEY (verified against all samples) ===")
    print(f"  format = {label}   hash = {alg}   oid = {with_oid}   e = {e}   {n.bit_length()} bits")
    print(f"  signed string (sample 0): {dict(candidates())[label](f0)!r}")
    pem = rsa.RSAPublicNumbers(e, n).public_key().public_bytes(
        serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo
    ).decode()
    Path("device_pubkey.pem").write_text(pem)
    print('\n  paste into data/state/device_key_state.json:')
    print(f'    {{"devices": {{"<did>": {{"modulus_hex": "{format(n, "x")}"}}}}}}')
    print("  Report the format/hash line in the GitHub issue.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

[Flash1232]

Well...it's not looking great (the additional messages about key length are just for debugging):

14 signatures, ~4096-bit key.
trying e=3...
4095 bits (512 bytes)
4091 bits (512 bytes)
4096 bits (512 bytes)
4094 bits (512 bytes)
4093 bits (512 bytes)
4096 bits (512 bytes)
4092 bits (512 bytes)
4096 bits (512 bytes)
4092 bits (512 bytes)
4095 bits (512 bytes)
4094 bits (512 bytes)
4095 bits (512 bytes)
4095 bits (512 bytes)
4088 bits (511 bytes)
trying e=17...
4095 bits (512 bytes)
4091 bits (512 bytes)
4096 bits (512 bytes)
4094 bits (512 bytes)
4093 bits (512 bytes)
4096 bits (512 bytes)
4092 bits (512 bytes)
4096 bits (512 bytes)
4092 bits (512 bytes)
4095 bits (512 bytes)
4094 bits (512 bytes)
4095 bits (512 bytes)
4095 bits (512 bytes)
4088 bits (511 bytes)

No match. Add guesses to candidates(), capture more samples, try --slow, or the scheme may use RSA-PSS (GCD can't recover that).

Currently running with "--slow" and waiting for the result..

EDIT: e=65537 failed as well....

This means probably that the newer firmware has to be reversed again. Question is how to dump it as it's fetched by the vacuum from the roborock servers directly, no?

If it still helps you, I could provide you with 4 samples privately (where to?)

EDIT 2: I've tried many more candidates for all kinds of potential new signature formats - to no avail. Maybe it's protobuf now, maybe the "v" string is used somewhere, maybe not?

[Lash-L]

If it still helps you, I could provide you with 4 samples privately (where to?)

You can email me at conway220@gmail.com

I don't think I'll have much luck, but I'll throw a bunch of compute at the problem and do a very long running task and see if I can get anything.

It's very likely they have a secret that is being used. i.e. concat some string in during the signing process. If that's the case, firmware dumps may be our only hope.

For getting the firmware, I've been exploring with seeing if I can add a 'fake' vacuum to the servers and have it request the firmware. But there is a lot that has to be 'right' for that to work. It has to be a real vacuum id seemingly, the vacuum has to be pending a firmware update, etc. I have not had a ton of luck.

The other approach is disassembly and see if it can be done there. Less of my wheel house and can't be done without the vacuum. (i.e. I can't just push out code and have someone run it)

[Kombustor]

I'm having the same issue. Just got a Curv 2 Pro, updated the firmware unknowingly to V02.44.34 and also just receive one /region call, no "Connected to WIFI" message or anything else, and nothing is happening.

If I can help with anything I'd be glad. Really sad about this right now. :/

[Lash-L]

Hey @Kombustor thanks for letting me know. Helps us build a timeline of what vacs aren't supported. It might not be the firmware version but rather just the vac itself. The only way to figure it out is a firmware dump. I have tried a decent bit through code, but I don't think that will be a valid option. The device has it's own private mqtt topic that it communicates to the cloud with. We can't access it and that is where firmware urls are sent. At least not in anyway I have been able to figure out.

the only solution is a firmware dump using the board. I don't think there is any easy way to do it. If anyone want to either a) try to dump the firmware themselves. or b) ship their board (I have someone who can do Xrays and dump the flash, but wont say his name w/o permission) Obviously, you'd be without a vac for probably 1.5 weeks. We can get firmware and go from there. To be honest, it is not a guarantee that when we get the firmware we will be able to do the exploit and you will lose warranty

If anyone is interested, please let me know and we can go from there.

Otherwise, I'm saving up from donations + affiliate links and will attempt to save up enough to buy one that we can work with.

[Kombustor]

Thanks for the quick reply. I'm not able to send my vac in, but I just donated :)

Just confirming that my decompiled_http.jsonl looks the same as above, with the s key and that the recover_header_key.py is unable to do anything with it:

2 signatures, ~4096-bit key.
trying e=3...
trying e=17...
trying e=65537 (slow)...

No match. Add guesses to candidates(), capture more samples, try --slow, or the scheme may use RSA-PSS (GCD can't recover that).

Just some random ideas/questions:

• Would it help if we had a vacuum that is not on the latest firmware? Maybe someone else who just ordered one or ordering one and returning it, or waiting for the next update?
• We can't sniff on any network traffic from the robo, right, because it's all encrypted?