-
-
Save shoeper/7a66513c31bde34a93ae2d4ffe512e5b to your computer and use it in GitHub Desktop.
Implementation of the Tuya API signing.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import requests | |
import hashlib | |
import time | |
import uuid | |
import os | |
import copy | |
import json | |
# This is based on my personal implementation but stripped down to only what is | |
# needed to verify it. | |
# This should correct anywhere I was unclear in my original post. | |
# I developed against a Mobile API key, which is why I used the mobile methods | |
# in my test cases. | |
# Signing works the same with the Cloud API. You get a "NO_PERMISSION" if your | |
# keys aren't valid for the request, but if you sign wrong, you'll get a error | |
# related to signing instead, which is checked earlier on their end. | |
# The signing process is also leaked in the mobile apps (which are all | |
# rebrandings, no joke) in the Android logs. | |
## For mobile app keys: | |
## Be careful about which host you use. If you are registered in the US, use the | |
## US host. | |
host = "https://a1.tuyaeu.com/api.json" | |
# Random and not really checked. Should keep persistent if you are using | |
# sessions. | |
device_id = os.urandom(32).encode('hex') | |
sid = "" # Used to mobile requests that require login. | |
# Needs to be set, but they don't care what it is. | |
os = ("TEST", "0.1.2", "TEST") | |
# Your API keys. | |
appKey = "<BLANKED>" | |
appSecret = "<BLANKED>" | |
# This is likely a cause of confusion. They decided for this and only this to | |
# rearrange the hash. Not their normal md5-mid which they use a lot in their | |
# MQTT client. | |
def post_data_hash_transform(post_data): | |
h = hashlib.md5() | |
h.update(post_data) | |
pre_hash = h.hexdigest() | |
return pre_hash[8:16] + pre_hash[0:8] + pre_hash[24:32] + pre_hash[16:24] | |
# This is the implementation of "sign". | |
def generate_request_sign(pairs): | |
# This are the values that get "signed" in a request, worth checking if I | |
# missed one in this list. | |
values_to_hash = ["a", "v", "lat", "lon", "lang", "deviceId", "imei", | |
"imsi", "appVersion", "ttid", "isH5", "h5Token", "os", | |
"clientId", "postData", "time", "n4h5", "sid", "sp"] | |
out = [] | |
sorted_pairs = sorted(pairs) | |
for item in sorted_pairs: | |
if item not in values_to_hash: | |
continue | |
if pairs[item] == "": | |
continue | |
if item == "postData": | |
out += [item + "=" + post_data_hash_transform(pairs["postData"])] | |
else: | |
out += [item + "=" + str(pairs[item])] | |
sign_request = "||".join(out) + "||" + appSecret | |
h = hashlib.md5() | |
h.update(sign_request) | |
return h.hexdigest() | |
# This will give you a dict with all the request parameters. | |
def url_generator(action, version, post_data=None, sid=None, time_param=None): | |
client_id = appKey | |
ttid = "sdk_google1@{}".format(appKey) | |
timezone_id = "America/New_York" | |
lang = "en" | |
if not time_param: | |
time_param = int(time.time()) | |
request_id = uuid.uuid4() | |
pairs = { | |
"sdkVersion": "1.11.11", | |
"platform": os[2], | |
"ttid": ttid, | |
"a": action, | |
"timeZoneId": timezone_id, | |
"deviceId": device_id, | |
"osSystem": os[1], | |
"os": os[0], | |
"v": version, | |
"appVersion": "2.9.1", | |
"clientId": client_id, | |
"lang": lang, | |
"requestId": request_id, | |
"time": time_param, | |
"appRnVersion": "2.9", | |
} | |
if sid: | |
pairs['sid'] = sid | |
to_sign = copy.deepcopy(pairs) | |
if post_data: | |
to_sign['postData'] = post_data | |
pairs['sign'] = generate_request_sign(to_sign) | |
return pairs | |
# Call the endpoint. | |
# * action is the name as defined in the tuya docs | |
# * version is the version they say to provide, normally "1.0" | |
# * data is the JSON data you want to pass to the action | |
# * requires_sid means it adds a session_id to the parameters. Used in mobile endpoints. | |
def preform_action(action, version, data=None, requires_sid=False): | |
if requires_sid is True and not sid: | |
return None | |
params = url_generator(action, version, data, sid) | |
print params | |
endpoint = host | |
headers = { | |
# Maybe set a user agent or something here. | |
} | |
if data: | |
r = requests.post(endpoint, params=params, data={"postData":data}, | |
headers=headers) | |
else: | |
r = requests.post(endpoint, params=params, headers=headers) | |
return r.status_code, r.json(), r.headers | |
if __name__ == "__main__": | |
# Should return a list of countries, doesn't use post data. | |
print preform_action("tuya.m.country.list", "1.0") | |
# Attempt to login with a fake email, takes POST data. | |
attempt = { | |
"email":"fake_email_attempt@fake_email.com", | |
"countryCode":1 | |
} | |
print preform_action("tuya.m.user.email.token.create", "1.0", | |
json.dumps(attempt)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment