Cloud endpoint

tuya_cloud.py

import requests
import hashlib
import time
import uuid
import os
import copy
import json

# Fixed up version of my previous code to work with the Cloud endpoints.
# Hopefully this works.
# Had a quick look at their public cloud API implementation at:
# https://github.com/TuyaInc/TuyaDemo/
# to fix the issue.

## Use the region your device is registered in.

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 = ""
# Needs to be set, but they don't care what it is.
os = ("Linux", "0.1.2", "TEST")
# Your API keys.
appKey = "<BLANKED>"
appSecret = "<BLANKED>"

# 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
        out += [item + "=" + str(pairs[item])]
    sign_request = appSecret+"|".join(out)
    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
    lang = "zh-Hans"
    if not time_param:
        time_param = int(time.time())
    request_id = uuid.uuid4()
    pairs = {
        "a": action,
        "deviceId": device_id,
        "os": os[0],
        "v": version,
        "clientId": client_id,
        "lang": lang,
        #"requestId": request_id,
        "time": time_param,
    }

    if sid:
        pairs['sid'] = sid
    if post_data:
        pairs['postData'] = post_data
    pairs['sign'] = generate_request_sign(pairs)
    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,
                          headers=headers)
    else:
        r = requests.post(endpoint, params=params, headers=headers)
    return r.status_code, r.json(), r.headers


if __name__ == "__main__":
    deviceID = "<DEVICE_ID_HERE>"
    print preform_action("tuya.p.weather.city.info.list", "1.0",
                         json.dumps({"countryCode":"CN"}))
    print preform_action("tuya.cloud.device.get", "1.0",
                         json.dumps({"devId": deviceID}))

I've done similar integration with https://a1.tuyaeu.com/api.json endpoint, but had to used different signatures ans encrypt postData:

1. Instead of md5 to sign the request I had to use HMAC-SHA256:

sign_request = "||".join(out)  #no appsecret here
hmac_key = "A_"+tuya_bmpkey+"_"+tuya_appsecret     # here you have to use secret2 (encoded in the image file) and standard secret
signature = hmac.new(key=hmac_key.encode('utf-8'),msg=feed.encode('utf-8'),digestmod=hashlib.sha256).hexdigest() 

encrypted postData as base64 still has to be put to above sign_request using peculiar "post_data_hash_transform" explained here: https://gist.github.com/bahorn/9bebbbf37c2167f7057aea0244ff2d92

2. the postData field has to be encrypted with AES in MODE_GMC with 12 bytes of random nonce as prefix and 16 bytes of validation MAC as suffix. The key is derived from request_id using HMAC-SHA256 with key obtained by contatenation of various secret values:

def encryptPostData(postData, requestId):
    #create key from requestid and ecode. ecode is created together with session id upon login, as far as i can see it is valid undefinietly,
    #so it's easier to sniff it than to request it 
    keyparts = "A_"+tuya_bmpkey+"_"+tuya_appsecret+"_"+tuya_ecode     # secret1, secret2 and ecode used here
    #generate key from request_id and secrets
    keyHex = hmac.new(key=requestId.encode('utf-8'),msg=keyparts.encode('utf-8'),digestmod=hashlib.sha256).hexdigest()     
    shortKey = keyHex[0:16].encode('utf-8')   #yes! you use only the first 16 characters of hexadecimal form as AES key
    postDataStr = json.dumps(postData)   
    nonce = os.urandom(12)
    plainBytes = postDataStr.encode('utf-8')   
    encryptedPostData, mac = AES.new(shortKey, AES.MODE_GCM, nonce).encrypt_and_digest(plainBytes)
    encryptedPostDataWithNonce = nonce+encryptedPostData+mac 
    encryptedPostDataBase64 = base64.b64encode(encryptedPostDataWithNonce).decode("utf-8")
    return encryptedPostDataBase64

3. The same method is used to decrypt the response. In json response, the "result" field is encrypted:

def decryptResult(result, requestId):
    #create key from requestid and ecode
    keyparts = "A_"+tuya_bmpkey+"_"+tuya_appsecret+"_"+tuya_ecode
    #generate key from request_id and ecode
    keyHex = hmac.new(key=requestId.encode('utf-8'),msg=keyparts.encode('utf-8'),digestmod=hashlib.sha256).hexdigest()     
    shortKey = keyHex[0:16].encode('utf-8')
    encryptedBytes = base64.b64decode(result)
    nonce = encryptedBytes[0:12]
    encryptedPayload = encryptedBytes[12:]
    decrypted = AES.new(shortKey, AES.MODE_GCM, nonce).decrypt(encryptedPayload[:-16]) #drop last 16 bytes, it's MAC signature
    return decrypted.decode("utf-8")  

hope it'll help someone!

hi @panjanek , thnx for the update, gonna try this script, seems this cloud api from app, gives more info about missing DP's then the Cloud API itself

Do you have a full script somewhere thats working?

thnx