bahorn/tuya_cloud.py

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

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!