i think this joins to this
You are the bomb! Have spent months trying to figure out a way to get notifications!
Thanks for the work done in node-red. Unfortunately it seems like my particular tuya doorbell is slightly different from yours and never ever presents a message with header “154”. Only a header with “115” ever shows up with information similar to what @lokster has managed to pull up. {"v":"3.0","bucket":"ty-eu-storage30-pic","files":[["/c06289-44351751-lkwl627faa9e420a6e16/detect/1707366649.jpeg","11b4aee9820c8060"]]}
If you manage to make any sense of this or could possibly assist to parse it into a URL would be great.
I worked out how to extract the image from the bucket/files message.
It was more annoying than I hoped it would be, but I got there in the end.
This link https://support.tuya.com/en/help/_detail/Kbfus79b0gcpi has some examples but the documentation it links to is missing, so i had to get the URL from squinting at the image, and it misses the decryption info.
You call a Tuya cloud API to get a URL for the file to download. That returns a file, but the contents isn’t just the actual file. It is a binary file that has a smaller header with a version and IV and then a block of AES encrypted data, for which the key is the bit after the .jpeg in the bucket/files message.
Here’s a python script that takes the base64 encoded raw DPS data and downloads and decrypts the file
from tuya_connector import TuyaOpenAPI
import base64
import sys
import json
from urllib.request import urlopen
from Crypto.Cipher import AES
import struct
import io
BLOCK_SIZE = 16
def pad(byte_array:bytearray):
pad_len = BLOCK_SIZE - len(byte_array) % BLOCK_SIZE
return byte_array + (bytes([pad_len]) * pad_len)
def unpad(s:bytearray):
return s[:-ord(s[len(s)-1:])]
ACCESS_ID = "********************"
ACCESS_KEY = "********************************"
API_ENDPOINT = "https://openapi.tuyaeu.com" # change for your own use case
DEVICE_ID = "**********************"
# Init OpenAPI and connect
openapi = TuyaOpenAPI(API_ENDPOINT, ACCESS_ID, ACCESS_KEY)
openapi.connect()
base64String = sys.argv[1]
decoded = json.loads(base64.b64decode(base64String))
bucket = decoded["bucket"]
file = decoded["files"][0][0]
key = decoded["files"][0][1].encode('utf-8')
fileURLFetch = openapi.get("/v1.0/devices/{0}/movement-configs?bucket={1}&file_path={2}".format(DEVICE_ID, bucket, file))
actualFileURL = fileURLFetch["result"]
fileContents = urlopen(actualFileURL).read()
with io.BytesIO(fileContents) as src_file:
# seems to be 1, which
version = struct.unpack('i', src_file.read(4))[0]
iv = src_file.read(16)
src_file.read(44)
file_contents = src_file.read()
cipher = AES.new(key, AES.MODE_CBC, iv)
result = cipher.decrypt(pad(file_contents))
with open("my_file.jpg", "wb") as binary_file:
binary_file.write(result)
From my investigation, based version it looks like there could be different cyphers used, but this seems to work for my camera at least.
I hope it helps!
Thank you, Paul. That seems like what I needed. However, it seems like this API is no longer working, right? I can’t find it in the API Explorer or the documentation.
Edit: I needed to authorize the Beta API