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}))

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

Here is the full script.
To use it you have to put values into global variables at the beginning.
I extracted most of these values from andoid emulator memory using memory monitor.:
Blue Stack to run android APK on windows: https://www.bluestacks.com/pl/index.html
Cheat Engine, to monitor bluestack memory, look for json field names like "ecode" or "sid": https://www.cheatengine.org/

I use this script as a part of Home Assistant automation.
You have to call API with sid periodically (for example once a day) because it expires otherwise.

import requests
import hashlib
import time
import uuid
import os
import copy
import json
import urllib3
import logging
import hmac
import base64
import string
import sys
from Crypto.Cipher import AES

tuya_appid =     "<digits and numbers>"
tuya_appsecret = "<secret1, digits and numbers>"
tuya_bmpkey =    "<secret2, digits and numbers, extracted from bitmap in android app resources, can be found in memory>"
tuya_endpoint =  "https://a1.tuyaeu.com/api.json"
tuya_sid =       "<session id extracted, after logging in, from andoid app memory using andoid emulator, digits and numbers. this is because i had problems with implementing authorization with login and password. so this is one session, but it stays active as long as in used by api request. after about a month of inactivity it expires>"
tuya_ecode =     "<ecode extracted from andoid app memory using andoid emulator, digits and numbers>"
tuya_gid =       "<digits, seems unimportant>"
tuya_certsign =  "<cert signature as hexadecimal bytes in form XX:XX:XX:XX...>"

urllib3.disable_warnings()

def addTuyaSignature(params):
    if not "sign" in params:
        values_to_hash = ["a", "v", "lat", "lon", "et", "lang", "deviceId", "imei",
                          "imsi", "appVersion", "ttid", "isH5", "h5Token", "os",
                          "clientId", "postData", "time", "n4h5", "sid", "sp", "requestId"]
        sorted_params = sorted(params)
        out = []
        for key in sorted_params:
            if key not in values_to_hash:
                continue
            if params[key] == "":
                continue    
            value = str(params[key])
            if key == "postData":
                h = hashlib.md5()
                h.update(value.encode('utf-8'))               
                value=h.hexdigest()
                value = value[8:16] + value[0:8] + value[24:32] + value[16:24]
            out += [key + "=" + value]

        feed = "||".join(out)
        hmac_key = tuya_certsign + "_"+tuya_bmpkey+"_"+tuya_appsecret
        signature = hmac.new(key=hmac_key.encode('utf-8'),msg=feed.encode('utf-8'),digestmod=hashlib.sha256).hexdigest()       
        params["sign"] = signature
    return params
    
def decryptResult(result, requestId):
    #create key from requestid and ecode
    keyparts = tuya_certsign + "_"+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])
    return decrypted.decode("utf-8")     

def encryptPostData(postData, requestId):
    #create key from requestid and ecode
    keyparts = tuya_certsign + "_"+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')
    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

def callTuyaApi(action, postData):
    time_param = int(time.time())
    request_id = str(uuid.uuid4())    
    params = {
               "a":action,
               "appRnVersion":"5.44",
               "appVersion":"3.33.5",
               "channel":"oem",
               "clientId":tuya_appid,
               "deviceCoreVersion":"3.29.5",
               "deviceId":"f1cf817055401e82b60fa5f74d8779e64133a59215b7",
               "et":"3",
               "lang":"pl_PL",
               "os":"Android",
               "osSystem":"7.1.1",
               "platform":"ONEPLUS A5000",
               "requestId":request_id,
               "sdkVersion":"3.29.5",
               "sid":tuya_sid,
               "time": str(time_param),
               "timeZoneId":"Europe/Warsaw",
               "ttid":"sdk_tuya_international",
               "v":"1.0" ,
               "gid": tuya_gid
            }

    #print(postData)
    params["postData"] = encryptPostData(postData, request_id)
    params = addTuyaSignature(params)    
    headers = { 'User-Agent' : 'Android/com.google.android.gms/203615023 (OnePlus5 NMF26X)' }   
    #print(params)
    response = requests.post(tuya_endpoint, params=params, headers=headers, verify=False)
    encryptedResult = response.json()["result"]       
    encryptedBytes = base64.b64decode(encryptedResult)
    jsonStr = decryptResult(encryptedResult, request_id)
    return json.loads(jsonStr)

def getDeviceDetails(devId):
    response = callTuyaApi("tuya.m.device.get", { "devId" :devId})
    #print(json.dumps(response, indent=4, sort_keys=True,ensure_ascii=False))
    dps = response["result"]["dps"]
    result = {}
    result["id"] = response["result"]["devId"]
    result["online"] = response["result"]["isOnline"]
    result["active"] = response["result"]["isActive"]
    result["localKey"] = response["result"]["localKey"]
    result["name"] = response["result"]["name"]
    for key in dps:
        result["dps_"+key] = dps[key]
    if "1" in dps and isinstance(dps["1"], bool):
        result["1"] = dps["1"]
    if "2" in dps and isinstance(dps["2"], bool):
        result["2"] = dps["2"]
    if "3" in dps and isinstance(dps["3"], bool):
        result["3"] = dps["3"]
    if "7" in dps and dps["7"]!=0:
        result["usb"] = dps["7"]
    if "20" in dps:
        result["voltage"] = float(dps["20"]) / 10.0
    if "18" in dps:
        result["current"] = float(dps["18"]) / 1000.0
    if "19" in dps:
        result["power"] = float(dps["19"]) / 10.0
    return result

if __name__ == "__main__":
    if len(sys.argv) <= 1:
        #print("usage: python tuyasmart.py <device-id> [1|2|3|usb] [on|off]")
        #print("devices:")
        devices = callTuyaApi("tuya.m.device.ext.prop.list", {})
        all = []
        for dev in devices["result"]:
            devId = dev["devId"]
            device = getDeviceDetails(devId)
            all.append(device)
        print(json.dumps(all, indent=4, sort_keys=True,ensure_ascii=False))
    elif len(sys.argv) == 2: 
        devId = sys.argv[1]
        device = getDeviceDetails(devId)
        print(json.dumps(device, indent=4, sort_keys=True,ensure_ascii=False))
    else: 
        devId = sys.argv[1]
        socket = sys.argv[2]
        if len(sys.argv) == 3:
            device = getDeviceDetails(devId)
            s = "1" if device[socket] else "0"
            print(s)
        else:   
            state = sys.argv[3]
            if state == "on" or state == "off":
                if socket == "usb":
                    socket = "7"
                postData = {}
                postData["devId"] = devId
                postData["dps"] = {}
                postData["dps"][str(socket)] = state == "on"
                response = callTuyaApi("tuya.m.device.dp.publish", postData)
                print(json.dumps(response, indent=4, sort_keys=True,ensure_ascii=False))
            elif state == "set":
                value = sys.argv[4]
                if value == "true":
                    value = True
                elif value == "false":
                    value = False
                elif value.isnumeric():
                    value = int(value)
                postData = {}
                postData["devId"] = devId
                postData["dps"] = {}
                postData["dps"][str(socket)] = value
                response = callTuyaApi("tuya.m.device.dp.publish", postData)
                print(json.dumps(response, indent=4, sort_keys=True,ensure_ascii=False))
                

thnx for this update!