一、先获取tooken
原链接：https://github.com/PiotrMachowski/Xiaomi-cloud-tokens-extractor
1 import base64
2 import hashlib
3 import hmac
4 import json
5 import os
6 import random
7 import time
8 from sys import platform
9 from Crypto.Cipher import ARC4
10
11 import requests
12
13 if platform != "win32":
14 import readline
15
16
17 class XiaomiCloudConnector:
18
19 def __init__(self, username, password):
20 self._username = username
21 self._password = password
22 self._agent = self.generate_agent()
23 self._device_id = self.generate_device_id()
24 self._session = requests.session()
25 self._sign = None
26 self._ssecurity = None
27 self._userId = None
28 self._cUserId = None
29 self._passToken = None
30 self._location = None
31 self._code = None
32 self._serviceToken = None
33
34 def login_step_1(self):
35 url = "https://account.xiaomi.com/pass/serviceLogin?sid=xiaomiio&_json=true"
36 headers = {
37 "User-Agent": self._agent,
38 "Content-Type": "application/x-www-form-urlencoded"
39 }
40 cookies = {
41 "userId": self._username
42 }
43 response = self._session.get(url, headers=headers, cookies=cookies)
44 valid = response.status_code == 200 and "_sign" in self.to_json(response.text)
45 if valid:
46 self._sign = self.to_json(response.text)["_sign"]
47 return valid
48
49 def login_step_2(self):
50 url = "https://account.xiaomi.com/pass/serviceLoginAuth2"
51 headers = {
52 "User-Agent": self._agent,
53 "Content-Type": "application/x-www-form-urlencoded"
54 }
55 fields = {
56 "sid": "xiaomiio",
57 "hash": hashlib.md5(str.encode(self._password)).hexdigest().upper(),
58 "callback": "https://sts.api.io.mi.com/sts",
59 "qs": "%3Fsid%3Dxiaomiio%26_json%3Dtrue",
60 "user": self._username,
61 "_sign": self._sign,
62 "_json": "true"
63 }
64 response = self._session.post(url, headers=headers, params=fields)
65 valid = response is not None and response.status_code == 200
66 if valid:
67 json_resp = self.to_json(response.text)
68 valid = "ssecurity" in json_resp and len(str(json_resp["ssecurity"])) > 4
69 if valid:
70 self._ssecurity = json_resp["ssecurity"]
71 self._userId = json_resp["userId"]
72 self._cUserId = json_resp["cUserId"]
73 self._passToken = json_resp["passToken"]
74 self._location = json_resp["location"]
75 self._code = json_resp["code"]
76 else:
77 if "notificationUrl" in json_resp:
78 print("Two factor authentication required, please use following url and restart extractor:")
79 print(json_resp["notificationUrl"])
80 print()
81 return valid
82
83 def login_step_3(self):
84 headers = {
85 "User-Agent": self._agent,
86 "Content-Type": "application/x-www-form-urlencoded"
87 }
88 response = self._session.get(self._location, headers=headers)
89 if response.status_code == 200:
90 self._serviceToken = response.cookies.get("serviceToken")
91 return response.status_code == 200
92
93 def login(self):
94 self._session.cookies.set("sdkVersion", "accountsdk-18.8.15", domain="mi.com")
95 self._session.cookies.set("sdkVersion", "accountsdk-18.8.15", domain="xiaomi.com")
96 self._session.cookies.set("deviceId", self._device_id, domain="mi.com")
97 self._session.cookies.set("deviceId", self._device_id, domain="xiaomi.com")
98 if self.login_step_1():
99 if self.login_step_2():
100 if self.login_step_3():
101 return True
102 else:
103 print("Unable to get service token.")
104 else:
105 print("Invalid login or password.")
106 else:
107 print("Invalid username.")
108 return False
109
110 def get_devices(self, country):
111 url = self.get_api_url(country) + "/home/device_list"
112 params = {
113 "data": '{"getVirtualModel":true,"getHuamiDevices":1,"get_split_device":false,"support_smart_home":true}'
114 }
115 return self.execute_api_call_encrypted(url, params)
116
117 def get_beaconkey(self, country, did):
118 url = self.get_api_url(country) + "/v2/device/blt_get_beaconkey"
119 params = {
120 "data": '{"did":"' + did + '","pdid":1}'
121 }
122 return self.execute_api_call_encrypted(url, params)
123
124 def execute_api_call_encrypted(self, url, params):
125 headers = {
126 "Accept-Encoding": "identity",
127 "User-Agent": self._agent,
128 "Content-Type": "application/x-www-form-urlencoded",
129 "x-xiaomi-protocal-flag-cli": "PROTOCAL-HTTP2",
130 "MIOT-ENCRYPT-ALGORITHM": "ENCRYPT-RC4",
131 }
132 cookies = {
133 "userId": str(self._userId),
134 "yetAnotherServiceToken": str(self._serviceToken),
135 "serviceToken": str(self._serviceToken),
136 "locale": "en_GB",
137 "timezone": "GMT+02:00",
138 "is_daylight": "1",
139 "dst_offset": "3600000",
140 "channel": "MI_APP_STORE"
141 }
142 millis = round(time.time() * 1000)
143 nonce = self.generate_nonce(millis)
144 signed_nonce = self.signed_nonce(nonce)
145 fields = self.generate_enc_params(url, "POST", signed_nonce, nonce, params, self._ssecurity)
146 response = self._session.post(url, headers=headers, cookies=cookies, params=fields)
147 if response.status_code == 200:
148 decoded = self.decrypt_rc4(self.signed_nonce(fields["_nonce"]), response.text)
149 return json.loads(decoded)
150 return None
151
152 def get_api_url(self, country):
153 return "https://" + ("" if country == "cn" else (country + ".")) + "api.io.mi.com/app"
154
155 def signed_nonce(self, nonce):
156 hash_object = hashlib.sha256(base64.b64decode(self._ssecurity) + base64.b64decode(nonce))
157 return base64.b64encode(hash_object.digest()).decode('utf-8')
158
159 @staticmethod
160 def generate_nonce(millis):
161 nonce_bytes = os.urandom(8) + (int(millis / 60000)).to_bytes(4, byteorder='big')
162 return base64.b64encode(nonce_bytes).decode()
163
164 @staticmethod
165 def generate_agent():
166 agent_id = "".join(map(lambda i: chr(i), [random.randint(65, 69) for _ in range(13)]))
167 return f"Android-7.1.1-1.0.0-ONEPLUS A3010-136-{agent_id} APP/xiaomi.smarthome APPV/62830"
168
169 @staticmethod
170 def generate_device_id():
171 return "".join(map(lambda i: chr(i), [random.randint(97, 122) for _ in range(6)]))
172
173 @staticmethod
174 def generate_signature(url, signed_nonce, nonce, params):
175 signature_params = [url.split("com")[1], signed_nonce, nonce]
176 for k, v in params.items():
177 signature_params.append(f"{k}={v}")
178 signature_string = "&".join(signature_params)
179 signature = hmac.new(base64.b64decode(signed_nonce), msg=signature_string.encode(), digestmod=hashlib.sha256)
180 return base64.b64encode(signature.digest()).decode()
181
182 @staticmethod
183 def generate_enc_signature(url, method, signed_nonce, params):
184 signature_params = [str(method).upper(), url.split("com")[1].replace("/app/", "/")]
185 for k, v in params.items():
186 signature_params.append(f"{k}={v}")
187 signature_params.append(signed_nonce)
188 signature_string = "&".join(signature_params)
189 return base64.b64encode(hashlib.sha1(signature_string.encode('utf-8')).digest()).decode()
190
191 @staticmethod
192 def generate_enc_params(url, method, signed_nonce, nonce, params, ssecurity):
193 params['rc4_hash__'] = XiaomiCloudConnector.generate_enc_signature(url, method, signed_nonce, params)
194 for k, v in params.items():
195 params[k] = XiaomiCloudConnector.encrypt_rc4(signed_nonce, v)
196 params.update({
197 'signature': XiaomiCloudConnector.generate_enc_signature(url, method, signed_nonce, params),
198 'ssecurity': ssecurity,
199 '_nonce': nonce,
200 })
201 return params
202
203 @staticmethod
204 def to_json(response_text):
205 return json.loads(response_text.replace("&&&START&&&", ""))
206
207 @staticmethod
208 def encrypt_rc4(password, payload):
209 r = ARC4.new(base64.b64decode(password))
210 r.encrypt(bytes(1024))
211 return base64.b64encode(r.encrypt(payload.encode())).decode()
212
213 @staticmethod
214 def decrypt_rc4(password, payload):
215 r = ARC4.new(base64.b64decode(password))
216 r.encrypt(bytes(1024))
217 return r.encrypt(base64.b64decode(payload))
218
219
220 def print_tabbed(value, tab):
221 print(" " * tab + value)
222
223
224 def print_entry(key, value, tab):
225 if value:
226 print_tabbed(f'{key + ":": <10}{value}', tab)
227
228
229 servers = ["cn", "de", "us", "ru", "tw", "sg", "in", "i2"]
230 servers_str = ", ".join(servers)
231 print("Username (email or user ID):")
232 username = input()
233 print("Password:")
234 password = input()
235 print(f"Server (one of: {servers_str}) Leave empty to check all available:")
236 server = input()
237 while server not in ["", *servers]:
238 print(f"Invalid server provided. Valid values: {servers_str}")
239 print("Server:")
240 server = input()
241
242 print()
243 if not server == "":
244 servers = [server]
245
246 connector = XiaomiCloudConnector(username, password)
247 print("Logging in...")
248 logged = connector.login()
249 if logged:
250 print("Logged in.")
251 print()
252 for current_server in servers:
253 devices = connector.get_devices(current_server)
254 if devices is not None:
255 if len(devices["result"]["list"]) == 0:
256 print(f"No devices found for server \"{current_server}\".")
257 continue
258 print(f"Devices found for server \"{current_server}\":")
259 for device in devices["result"]["list"]:
260 print_tabbed("---------", 3)
261 if "name" in device:
262 print_entry("NAME", device["name"], 3)
263 if "did" in device:
264 print_entry("ID", device["did"], 3)
265 if "blt" in device["did"]:
266 beaconkey = connector.get_beaconkey(current_server, device["did"])
267 if beaconkey and "result" in beaconkey and "beaconkey" in beaconkey["result"]:
268 print_entry("BLE KEY", beaconkey["result"]["beaconkey"], 3)
269 if "mac" in device:
270 print_entry("MAC", device["mac"], 3)
271 if "localip" in device:
272 print_entry("IP", device["localip"], 3)
273 if "token" in device:
274 print_entry("TOKEN", device["token"], 3)
275 if "model" in device:
276 print_entry("MODEL", device["model"], 3)
277 print_tabbed("---------", 3)
278 print()
279 else:
280 print(f"Unable to get devices from server {current_server}.")
281 else:
282 print("Unable to log in.")
283
284 print()
285 print("Press ENTER to finish")
286 input()
View Code
二、控制
1 from miio import FanP5 2 3 ip = '你的ip' 4 token = '你的tooken' 5 fan = FanP5(ip=ip, token=token) 6 fan.on() 7 #fan.off()
标签：nonce,miio,python,self,落地扇,params,url,._,def From： https://www.cnblogs.com/ff888/p/16726041.html