import argparse import base64 import getpass import hashlib import json import os import random import struct from datetime import datetime from urllib.parse import urlencode import httpx LOGIN_PREFIX = b"&&&START&&&" DEFAULT_KEYS = [ "weight", "steps", "step", "activity", "sleep", "heart_rate", "heartrate", "hr", "sport", "workout", "calories", ] DEFAULT_REGIONS = ["cn", "de", "i2", "ru", "sg", "us"] def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(prog="probe_mifitness") parser.add_argument("--username") parser.add_argument("--password") parser.add_argument("--token") parser.add_argument("--user-id") parser.add_argument("--pass-token") parser.add_argument("--start-date", default="2025-04-01") parser.add_argument("--end-date", default="2025-05-31") parser.add_argument("--regions", nargs="*", default=DEFAULT_REGIONS) parser.add_argument("--keys", nargs="*", default=DEFAULT_KEYS) return parser.parse_args() def md5_upper(text: str) -> str: return hashlib.md5(text.encode()).hexdigest().upper() def read_login_payload(text: str) -> dict: body = text.encode() if not body.startswith(LOGIN_PREFIX): raise RuntimeError("unexpected Xiaomi login response") return json.loads(body[len(LOGIN_PREFIX):].decode()) def rc4_crypt(key: bytes, payload: bytes) -> bytes: s = list(range(256)) j = 0 key_len = len(key) for i in range(256): j = (j + s[i] + key[i % key_len]) % 256 s[i], s[j] = s[j], s[i] i = 0 j = 0 def next_byte() -> int: nonlocal i, j i = (i + 1) % 256 j = (j + s[i]) % 256 s[i], s[j] = s[j], s[i] return s[(s[i] + s[j]) % 256] for _ in range(1024): next_byte() out = bytearray() for b in payload: out.append(b ^ next_byte()) return bytes(out) def gen_nonce() -> bytes: raw = bytearray(os.urandom(8)) raw.extend(struct.pack(">I", int(datetime.now().timestamp() // 60))) return bytes(raw) def gen_signed_nonce(ssecurity: bytes, nonce: bytes) -> bytes: return hashlib.sha256(ssecurity + nonce).digest() def gen_signature(method: str, path: str, values: dict[str, str], signed_nonce: bytes) -> str: base = method + "&" + path + "&data=" + values["data"] if "rc4_hash__" in values: base += "&rc4_hash__=" + values["rc4_hash__"] base += "&" + base64.b64encode(signed_nonce).decode() return base64.b64encode(hashlib.sha1(base.encode()).digest()).decode() class MiFitnessProbe: def __init__(self) -> None: self.client = httpx.Client(timeout=30.0, follow_redirects=False) self.sid = "miothealth" self.cookies = "" self.ssecurity = b"" self.pass_token = "" self.user_id = 0 def login(self, username: str, password: str) -> None: info = self.service_login() auth = self.service_login2(info, username, password) self.service_login3(auth["location"]) def login_with_token(self, token: str | None = None, user_id: str | None = None, pass_token: str | None = None) -> None: if token: user_id, pass_token = token.split(":", 1) if not user_id or not pass_token: raise RuntimeError("token login requires user_id and pass_token") response = self.client.get( f"https://account.xiaomi.com/pass/serviceLogin?_json=true&sid={self.sid}", headers={"Cookie": f"userId={user_id}; passToken={pass_token}"}, ) response.raise_for_status() payload = read_login_payload(response.text) if "passToken" not in payload or "ssecurity" not in payload or "userId" not in payload or "location" not in payload: details = { "keys": sorted(payload.keys()), "code": payload.get("code"), "description": payload.get("description"), "desc": payload.get("desc"), "location": payload.get("location"), } raise RuntimeError(f"xiaomi token login failed; details={json.dumps(details, ensure_ascii=False)}") self.pass_token = payload["passToken"] self.ssecurity = base64.b64decode(payload["ssecurity"]) self.user_id = int(payload["userId"]) self.service_login3(payload["location"]) def service_login(self) -> dict: response = self.client.get(f"https://account.xiaomi.com/pass/serviceLogin?_json=true&sid={self.sid}") response.raise_for_status() return read_login_payload(response.text) def service_login2(self, info: dict, username: str, password: str) -> dict: form = { "_json": "true", "hash": md5_upper(password), "sid": info["sid"], "callback": info["callback"], "_sign": info["_sign"], "qs": info["qs"], "user": username, } device_id = "".join(random.choice("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") for _ in range(16)) response = self.client.post( "https://account.xiaomi.com/pass/serviceLoginAuth2", headers={ "Cookie": f"deviceId={device_id}", "Content-Type": "application/x-www-form-urlencoded", }, content=urlencode(form), ) response.raise_for_status() payload = read_login_payload(response.text) if "passToken" not in payload or "ssecurity" not in payload or "userId" not in payload: message = payload.get("description") or payload.get("desc") or payload.get("message") details = { "keys": sorted(payload.keys()), "code": payload.get("code"), "description": payload.get("description"), "desc": payload.get("desc"), "captchaUrl": payload.get("captchaUrl"), "notificationUrl": payload.get("notificationUrl"), "location": payload.get("location"), } raise RuntimeError(f"xiaomi login failed: {message or 'missing auth tokens'}; details={json.dumps(details, ensure_ascii=False)}") self.pass_token = payload["passToken"] self.ssecurity = base64.b64decode(payload["ssecurity"]) self.user_id = int(payload["userId"]) return payload def service_login3(self, location: str) -> None: response = self.client.get(location) response.raise_for_status() cookie_parts = [] for value in response.headers.get_list("set-cookie"): cookie_parts.append(value.split(";", 1)[0]) self.cookies = "; ".join(cookie_parts) def request(self, base_url: str, api_path: str, params: dict) -> dict: form = {"data": json.dumps(params, separators=(",", ":"))} nonce = gen_nonce() signed_nonce = gen_signed_nonce(self.ssecurity, nonce) form["rc4_hash__"] = gen_signature("POST", api_path, form, signed_nonce) encrypted = {} for key, value in form.items(): encrypted[key] = base64.b64encode(rc4_crypt(signed_nonce, value.encode())).decode() encrypted["signature"] = gen_signature("POST", api_path, encrypted, signed_nonce) encrypted["_nonce"] = base64.b64encode(nonce).decode() response = self.client.post( base_url + api_path, headers={ "Cookie": self.cookies, "Content-Type": "application/x-www-form-urlencoded", }, content=urlencode(encrypted), ) response.raise_for_status() plaintext = rc4_crypt(signed_nonce, base64.b64decode(response.text)) payload = json.loads(plaintext) if payload.get("code") != 0: raise RuntimeError(payload.get("message", "unknown mi fitness error")) return payload.get("result", {}) def region_base(region: str) -> str: if region in ("", "cn"): return "https://hlth.io.mi.com" return f"https://{region}.hlth.io.mi.com" def sample_view(items: list[dict]) -> str: if not items: return "[]" return json.dumps(items[:2], ensure_ascii=False)[:700] def run_probe(args: argparse.Namespace) -> None: probe = MiFitnessProbe() if args.token or (args.user_id and args.pass_token): probe.login_with_token(token=args.token, user_id=args.user_id, pass_token=args.pass_token) else: username = args.username or input("Xiaomi login (email/phone): ").strip() password = args.password or getpass.getpass("Xiaomi password: ") probe.login(username, password) print(f"login ok: user_id={probe.user_id}") for region in args.regions: base = region_base(region) print(f"\n=== region={region} base={base} ===") for key in args.keys: try: result = probe.request( base, "/app/v1/data/get_fitness_data_by_time", { "start_time": int(datetime.fromisoformat(args.start_date).timestamp()), "end_time": int(datetime.fromisoformat(args.end_date + "T23:59:59").timestamp()), "key": key, }, ) data_list = result.get("data_list", []) print(f"key={key} count={len(data_list)} has_more={result.get('has_more')} next_key={bool(result.get('next_key'))}") if data_list: print(sample_view(data_list)) except Exception as exc: print(f"key={key} error={exc}") if __name__ == "__main__": run_probe(parse_args())