Canard64 / Domoticz-Tuya

Utilisation des API tuya pour intégration dans DOMOTICZ
1 stars 1 forks source link

Sign invalid #1

Open chirogocki opened 1 year ago

chirogocki commented 1 year ago

Bonjour,

Merci pour ce travail autour de l'intégration des devices tuya pour domoticz ! Je rencontre un problème pour l'utilisation des API. J'ai bien cloné le projet /home/pi/domoticz/scripts/python/ et j'ai modifié le fichier main.py pour que le path du fichier code.json soit correct (remplacement de self.full_path = "/home/pi/domoticz/scripts/python/domoticz-tuya/" par self.full_path = "/home/pi/domoticz/scripts/python/Domoticz-Tuya/") J'ai également renseigné le fichier code.json avec mes client_id et client_secret Lorceque je tente de demander un statut d'un device avec : ./main.py --status mondevice J'ai le retour suivent : Authentification Error: sign invalid None Est ce qu'il y aurait eu un changement dans l'API de tuya : https://developer.tuya.com/en/docs/iot/new-singnature?id=Kbw0q34cs2e5g ?

MigaFr commented 1 year ago

Bonjour,

la même pour moi...

ça crack au login .

Il semblerait qu'il y ai une variation sur la signature depuis fin 2022: https://developer.tuya.com/en/docs/iot/new-singnature?id=Kbw0q34cs2e5g

thibautautrive commented 1 year ago

Bonjour,

en se basant sur le script de ce repo (qui semble similaire) : https://github.com/BreizhCat/domoticz-tuya J'ai modifié le script pour prendre en compte la nouvelle authentification :

#!/opt/bin/python3

__author__ = "Yann LEROUX"
__version__ = "1.0.2"
__email__ = "[yleroux@gmail.com](mailto:yleroux@gmail.com)"

import requests
import hmac
import hashlib
import json
from time import time
import sys

class tuya_api:
    def __init__(self):
        self._isLogged  = False
        self._encode   = 'HMAC-SHA256'

        self.debug     = False
        self.url_api   = "https://openapi.tuyaeu.com/"
        self.full_path = "/usr/local/domoticz/var/scripts/domo-tuya/"

        self.stringToSign = ""

        with open(self.full_path + 'code.json') as param_data:
            data = json.load(param_data)
            self.client_id = data['client_id']
            self.secret    = data['app_id']

    def _debug(self, text):
        if self.debug:
            print(text)

    def _getInfo(self):
        print('Timestamp: '+str(self.timestamp))
        print('Signature:'+self.signature)
        print('Token:' + self.token)

    def _getStringToSign(self, httpMethod, url = "", body = ""):
        sha256 = hashlib.sha256(bytes(body).encode('UTF-8')).hexdigest()
        self.stringToSign = str(httpMethod) + "\n" + sha256 + "\n" + "" + "\n" + url

    def _getSignature(self, token = False):
        self._debug('Get sign...')
        self.timestamp = int(time()*1000)

        if token:
            message =   self.client_id + self.token + str(self.timestamp) + "" + self.stringToSign
        else:
            message = self.client_id + str(self.timestamp) + "" + self.stringToSign

        self.signature = hmac.new(bytes(self.secret).encode('latin-1'), msg = bytes(message).encode('latin-1'), digestmod = hashlib.sha256).hexdigest().upper()

    def login(self):
        self._debug('Login...')

        urlToCall = '/v1.0/token?grant_type=1'
        self._getStringToSign("GET", urlToCall, "")
        self._getSignature() 

        header = {
            'client_id'  : self.client_id,
            't'          : str(self.timestamp),
            'sign_method': 'HMAC-SHA256',
            'sign'       : self.signature
        }          

        res = requests.get(self.url_api + urlToCall, headers=header)

        if res.ok: 
            result = json.loads(res.content)
            if result['success']:
                self.token = result['result']['access_token']
                self._isLogged = True
            else:
                print('Authentification Error: ' + result['msg'])
        else:
            print("HTTP %i - %s, Message %s" % (res.status_code, res.reason, res.text))

    def switch(self, id, value):
        if not self._isLogged:
            return
        self._debug("Switch...")

        urlToCall = '/v1.0/devices/' + id + '/commands'
        data = '{\n\t\"commands\":[\n\t\t{\n\t\t\t\"code\": \"switch_1\",\n\t\t\t\"value\":'+value+'\n\t\t}\n\t]\n}' 

        self._getStringToSign("POST", urlToCall, data)
        self._getSignature(True)

        header = {
            'client_id'    : self.client_id,
            'access_token' : self.token,
            'sign'         : self.signature,
            't'            : str(self.timestamp),
            'sign_method'  : self._encode,
            'Content-Type' :'application/json'
        }

        res = [requests.post](http://requests.post/)(self.url_api + urlToCall, headers=header, data = data)
        if res.ok:
            result = json.loads(res.content)
            if result['success']:
                self._debug('Device ' + id + 'status set to ' + value)
            else:
                print('Execution Error: ' + result['msg'])   
        else:
            print("HTTP %i - %s, Message %s" % (res.status_code, res.reason, res.text))

    def getStatus(self, id):
        if not self._isLogged:
            return

        self._debug("Get Statuts...")

        urlToCall = '/v1.0/devices/'+ id+ '/status'

        self._getStringToSign("GET", urlToCall, "")
        self._getSignature(True)        

        header = {
            'client_id'    : self.client_id,
            'access_token' : self.token,
            'sign'         : self.signature,
            't'            : str(self.timestamp),
            'sign_method'  : self._encode,
        }

        res = requests.get(self.url_api + urlToCall, headers=header)

        if res.ok: 
            result = json.loads(res.content)
            if result['success']:
                return result['result'][0]['value']
            else:
                print('Authentification Error: ' + result['msg'])
        else:
            print("HTTP %i - %s, Message %s" % (res.status_code, res.reason, res.text))

def help():
    print('Options available')
    print('-----------------')
    print('main.py --switch <ID> <True|False>')
    print('main.py --status <ID>') 
    print('main.py --toggle <ID>')  

def main():
    if len(sys.argv) <= 1:
        help()
    else:
        if sys.argv[1] == '--switch':
            tuya = tuya_api()
            tuya.login()
            tuya.switch(sys.argv[2], sys.argv[3])
        elif sys.argv[1] == '--status':
            tuya = tuya_api()
            tuya.login()
            print(tuya.getStatus(sys.argv[2]))
        elif sys.argv[1] == '--toggle':
            tuya = tuya_api()
            tuya.login()
            if tuya.getStatus(sys.argv[2]):
                tuya.switch(sys.argv[2], "false")
            else:
                tuya.switch(sys.argv[2], "true")
        else: 
            help()

main()
tmgrisou commented 6 months ago

Bonjour, toujours des problêmes pour l'utilisation des API, en réutilisant le dernier code et le corrigeant avec l'aide de chatGPT, il fonctionne:

#!/opt/bin/python3

__author__ = "Yann LEROUX"
__version__ = "1.0.2"
__email__ = "[yleroux@gmail.com](mailto:yleroux@gmail.com)"

import requests
import hmac
import hashlib
import json
from time import time
import sys

class tuya_api:
    def __init__(self):
        self._isLogged  = False
        self._encode   = 'HMAC-SHA256'

        self.debug     = False
        self.url_api   = "https://openapi.tuyaeu.com/"
        self.full_path = "/home/pi/domoticz/scripts/python/Domoticz-Tuya/"

        self.stringToSign = ""

        with open(self.full_path + 'code.json') as param_data:
            data = json.load(param_data)
            self.client_id = data['client_id']
            self.secret    = data['app_id']

    def _debug(self, text):
        if self.debug:
            print(text)

    def _getInfo(self):
        print('Timestamp: '+str(self.timestamp))
        print('Signature:'+self.signature)
        print('Token:' + self.token)

    def _getStringToSign(self, httpMethod, url = "", body = ""):
       sha256 = hashlib.sha256(body.encode('UTF-8')).hexdigest()
       self.stringToSign = str(httpMethod) + "\n" + sha256 + "\n" + "" + "\n" + url

    def _getSignature(self, token = False):
        self._debug('Get sign...')
        self.timestamp = int(time()*1000)

        if token:
            message =   self.client_id + self.token + str(self.timestamp) + "" + self.stringToSign
        else:
            message = self.client_id + str(self.timestamp) + "" + self.stringToSign

        self.signature = hmac.new(bytes(self.secret, 'latin-1'), msg=bytes(message, 'latin-1'), digestmod=hashlib.sha256).hexdigest().upper()

    def login(self):
        self._debug('Login...')

        urlToCall = '/v1.0/token?grant_type=1'
        self._getStringToSign("GET", urlToCall, "")
        self._getSignature()

        header = {
            'client_id'  : self.client_id,
            't'          : str(self.timestamp),
            'sign_method': 'HMAC-SHA256',
            'sign'       : self.signature
        }

        res = requests.get(self.url_api + urlToCall, headers=header)

        if res.ok:
            result = json.loads(res.content)
            if result['success']:
                self.token = result['result']['access_token']
                self._isLogged = True
            else:
                print('Authentification Error: ' + result['msg'])
        else:
            print("HTTP %i - %s, Message %s" % (res.status_code, res.reason, res.text))

    def switch(self, id, value):
        if not self._isLogged:
            return
        self._debug("Switch...")

        urlToCall = '/v1.0/devices/' + id + '/commands'
        data = '{\n\t\"commands\":[\n\t\t{\n\t\t\t\"code\": \"switch_1\",\n\t\t\t\"value\":'+value+'\n\t\t}\n\t]\n}'

        self._getStringToSign("POST", urlToCall, data)
        self._getSignature(True)

        header = {
            'client_id'    : self.client_id,
            'access_token' : self.token,
            'sign'         : self.signature,
            't'            : str(self.timestamp),
            'sign_method'  : self._encode,
            'Content-Type' :'application/json'
        }

        res = requests.post(self.url_api + urlToCall, headers=header, data = data)
        if res.ok:
            result = json.loads(res.content)
            if result['success']:
                self._debug('Device ' + id + 'status set to ' + value)
            else:
                print('Execution Error: ' + result['msg'])
        else:
            print("HTTP %i - %s, Message %s" % (res.status_code, res.reason, res.text))

    def getStatus(self, id):
        if not self._isLogged:
            return

        self._debug("Get Statuts...")

        urlToCall = '/v1.0/devices/'+ id+ '/status'

        self._getStringToSign("GET", urlToCall, "")
        self._getSignature(True)

        header = {
            'client_id'    : self.client_id,
            'access_token' : self.token,
            'sign'         : self.signature,
            't'            : str(self.timestamp),
            'sign_method'  : self._encode,
        }

        res = requests.get(self.url_api + urlToCall, headers=header)

        if res.ok:
            result = json.loads(res.content)
            if result['success']:
                return result['result'][0]['value']
            else:
                print('Authentification Error: ' + result['msg'])
        else:
            print("HTTP %i - %s, Message %s" % (res.status_code, res.reason, res.text))

def help():
    print('Options available')
    print('-----------------')
    print('main.py --switch <ID> <True|False>')
    print('main.py --status <ID>')
    print('main.py --toggle <ID>')

def main():
    if len(sys.argv) <= 1:
        help()
    else:
        if sys.argv[1] == '--switch':
            tuya = tuya_api()
            tuya.login()
            tuya.switch(sys.argv[2], sys.argv[3])
        elif sys.argv[1] == '--status':
            tuya = tuya_api()
            tuya.login()
            print(tuya.getStatus(sys.argv[2]))
        elif sys.argv[1] == '--toggle':
            tuya = tuya_api()
            tuya.login()
            if tuya.getStatus(sys.argv[2]):
                tuya.switch(sys.argv[2], "false")
            else:
                tuya.switch(sys.argv[2], "true")
        else:
            help()

main()