Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

add new encryption method #165

Merged
merged 3 commits into from
May 27, 2024
Merged

add new encryption method #165

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
0c2e891
add new encryption method
tomaszduda23 May 26, 2024
814d3d4
fix decryption for version 1
tomaszduda23 May 26, 2024
694548a
remove duplicated code
tomaszduda23 May 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,8 @@ This component is added to HACS default repository list.
port: 7000
mac: '<mac address of your first AC. NOTE: Format can be XX:XX:XX:XX:XX:XX or XX-XX-XX-XX-XX-XX depending on your model>'
target_temp_step: 1
encryption_key: <OPTIONAL: custom encryption key>
encryption_key: <OPTIONAL: custom encryption key. Integration will try to get key from device if empty>
encryption_version: <OPTIONAL: should be set to 2 for V1.21>
uid: <some kind of device identifier. NOTE: for some devices this is optional>
temp_sensor: <entity id of the EXTERNAL temperature sensor. For example: sensor.bedroom_temperature>
lights: <OPTIONAL: input_boolean to switch AC lights mode on/off. For example: input_boolean.first_ac_lights>
Expand Down
80 changes: 70 additions & 10 deletions custom_components/gree/climate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
CONF_AUTO_XFAN = 'auto_xfan'
CONF_AUTO_LIGHT = 'auto_light'
CONF_TARGET_TEMP = 'target_temp'
CONF_ENCRYPTION_VERSION = 'encryption_version'

DEFAULT_PORT = 7000
DEFAULT_TIMEOUT = 10
Expand All @@ -83,6 +84,9 @@
SWING_MODES = ['Default', 'Swing in full range', 'Fixed in the upmost position', 'Fixed in the middle-up position', 'Fixed in the middle position', 'Fixed in the middle-low position', 'Fixed in the lowest position', 'Swing in the downmost region', 'Swing in the middle-low region', 'Swing in the middle region', 'Swing in the middle-up region', 'Swing in the upmost region']
PRESET_MODES = ['Default', 'Full swing', 'Fixed in the leftmost position', 'Fixed in the middle-left position', 'Fixed in the middle postion','Fixed in the middle-right position', 'Fixed in the rightmost position']

GCM_IV = b'\x54\x40\x78\x44\x49\x67\x5a\x51\x6c\x5e\x63\x13'
GCM_ADD = b'qualcomm-test'

PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Required(CONF_HOST): cv.string,
Expand All @@ -102,7 +106,8 @@
vol.Optional(CONF_UID): cv.positive_int,
vol.Optional(CONF_AUTO_XFAN): cv.boolean,
vol.Optional(CONF_AUTO_LIGHT): cv.boolean,
vol.Optional(CONF_TARGET_TEMP): cv.entity_id
vol.Optional(CONF_TARGET_TEMP): cv.entity_id,
vol.Optional(CONF_ENCRYPTION_VERSION, default=1): cv.positive_int,
})

async def async_setup_platform(hass, config, async_add_devices, discovery_info=None):
Expand Down Expand Up @@ -131,16 +136,17 @@ async def async_setup_platform(hass, config, async_add_devices, discovery_info=N
uid = config.get(CONF_UID)
auto_xfan = config.get(CONF_AUTO_XFAN)
auto_light = config.get(CONF_AUTO_LIGHT)
encryption_version = config.get(CONF_ENCRYPTION_VERSION)

_LOGGER.info('Adding Gree climate device to hass')

async_add_devices([
GreeClimate(hass, name, ip_addr, port, mac_addr, timeout, target_temp_step, temp_sensor_entity_id, lights_entity_id, xfan_entity_id, health_entity_id, powersave_entity_id, sleep_entity_id, eightdegheat_entity_id, air_entity_id, target_temp_entity_id, hvac_modes, fan_modes, swing_modes, preset_modes, auto_xfan, auto_light, encryption_key, uid)
GreeClimate(hass, name, ip_addr, port, mac_addr, timeout, target_temp_step, temp_sensor_entity_id, lights_entity_id, xfan_entity_id, health_entity_id, powersave_entity_id, sleep_entity_id, eightdegheat_entity_id, air_entity_id, target_temp_entity_id, hvac_modes, fan_modes, swing_modes, preset_modes, auto_xfan, auto_light, encryption_version, encryption_key, uid)
])

class GreeClimate(ClimateEntity):

def __init__(self, hass, name, ip_addr, port, mac_addr, timeout, target_temp_step, temp_sensor_entity_id, lights_entity_id, xfan_entity_id, health_entity_id, powersave_entity_id, sleep_entity_id, eightdegheat_entity_id, air_entity_id, target_temp_entity_id, hvac_modes, fan_modes, swing_modes, preset_modes, auto_xfan, auto_light,encryption_key=None, uid=None):
def __init__(self, hass, name, ip_addr, port, mac_addr, timeout, target_temp_step, temp_sensor_entity_id, lights_entity_id, xfan_entity_id, health_entity_id, powersave_entity_id, sleep_entity_id, eightdegheat_entity_id, air_entity_id, target_temp_entity_id, hvac_modes, fan_modes, swing_modes, preset_modes, auto_xfan, auto_light, encryption_version, encryption_key=None, uid=None):
_LOGGER.info('Initialize the GREE climate device')
self.hass = hass
self._name = name
Expand Down Expand Up @@ -182,12 +188,19 @@ def __init__(self, hass, name, ip_addr, port, mac_addr, timeout, target_temp_ste
self._preset_modes = preset_modes

self._enable_turn_on_off_backwards_compatibility = False

self.encryption_version = encryption_version

if encryption_key:
_LOGGER.info('Using configured encryption key: {}'.format(encryption_key))
self._encryption_key = encryption_key.encode("utf8")
else:
self._encryption_key = self.GetDeviceKey().encode("utf8")
if encryption_version == 1:
self._encryption_key = self.GetDeviceKey().encode("utf8")
elif encryption_version == 2:
self._encryption_key = self.GetDeviceKeyGCM().encode("utf8")
else:
_LOGGER.error('Encryption version %s is not implemented.' % encryption_version)
_LOGGER.info('Fetched device encrytion key: %s' % str(self._encryption_key))

self._auto_xfan = auto_xfan
Expand All @@ -202,8 +215,11 @@ def __init__(self, hass, name, ip_addr, port, mac_addr, timeout, target_temp_ste

self._firstTimeRun = True

# Cipher to use to encrypt/decrypt
self.CIPHER = AES.new(self._encryption_key, AES.MODE_ECB)
if encryption_version == 1:
# Cipher to use to encrypt/decrypt
self.CIPHER = AES.new(self._encryption_key, AES.MODE_ECB)
elif encryption_version != 2:
_LOGGER.error('Encryption version %s is not implemented.' % encryption_version)

if temp_sensor_entity_id:
_LOGGER.info('Setting up temperature sensor: ' + str(temp_sensor_entity_id))
Expand Down Expand Up @@ -269,6 +285,9 @@ def FetchResult(self, cipher, ip_addr, port, timeout, json):
pack = receivedJson['pack']
base64decodedPack = base64.b64decode(pack)
decryptedPack = cipher.decrypt(base64decodedPack)
if self.encryption_version == 2:
tag = receivedJson['tag']
cipher.verify(base64.b64decode(tag))
decodedPack = decryptedPack.decode("utf-8")
replacedPack = decodedPack.replace('\x0f', '').replace(decodedPack[decodedPack.rindex('}')+1:], '')
loadedJsonPack = simplejson.loads(replacedPack)
Expand All @@ -282,9 +301,36 @@ def GetDeviceKey(self):
jsonPayloadToSend = '{"cid": "app","i": 1,"pack": "' + pack + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid": 0}'
return self.FetchResult(cipher, self._ip_addr, self._port, self._timeout, jsonPayloadToSend)['key']

def GetDeviceKeyGCM(self):
_LOGGER.info('Retrieving HVAC encryption key')
GENERIC_GREE_DEVICE_KEY = b'{yxAHAY_Lm6pbC/<'
RobHofmann marked this conversation as resolved.
Show resolved Hide resolved
cipher = AES.new(GENERIC_GREE_DEVICE_KEY, AES.MODE_GCM, nonce=GCM_IV)
cipher.update(GCM_ADD)
plaintext = '{"cid":"' + str(self._mac_addr) + '", "mac":"' + str(self._mac_addr) + '","t":"bind","uid":0}'
encrypted_data, tag = cipher.encrypt_and_digest(plaintext.encode("utf8"))
pack = base64.b64encode(encrypted_data).decode('utf-8')
tag = base64.b64encode(tag).decode('utf-8')
jsonPayloadToSend = '{"cid": "app","i": 1,"pack": "' + pack + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid": 0, "tag" : "' + tag + '"}'
cipher = AES.new(GENERIC_GREE_DEVICE_KEY, AES.MODE_GCM, nonce=GCM_IV)
cipher.update(GCM_ADD)
return self.FetchResult(cipher, self._ip_addr, self._port, self._timeout, jsonPayloadToSend)['key']

def GreeGetValues(self, propertyNames):
jsonPayloadToSend = '{"cid":"app","i":0,"pack":"' + base64.b64encode(self.CIPHER.encrypt(self.Pad('{"cols":' + simplejson.dumps(propertyNames) + ',"mac":"' + str(self._mac_addr) + '","t":"status"}').encode("utf8"))).decode('utf-8') + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid":{}'.format(self._uid) + '}'
return self.FetchResult(self.CIPHER, self._ip_addr, self._port, self._timeout, jsonPayloadToSend)['dat']
plaintext = '{"cols":' + simplejson.dumps(propertyNames) + ',"mac":"' + str(self._mac_addr) + '","t":"status"}'
if self.encryption_version == 1:
cipher = self.CIPHER
RobHofmann marked this conversation as resolved.
Show resolved Hide resolved
jsonPayloadToSend = '{"cid":"app","i":0,"pack":"' + base64.b64encode(cipher.encrypt(self.Pad(plaintext).encode("utf8"))).decode('utf-8') + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid":{}'.format(self._uid) + '}'
elif self.encryption_version == 2:
cipher = AES.new(self._encryption_key, AES.MODE_GCM, nonce=GCM_IV)
cipher.update(GCM_ADD)
encrypted_data, tag = cipher.encrypt_and_digest(plaintext.encode("utf8"))
pack = base64.b64encode(encrypted_data).decode('utf-8')
tag = base64.b64encode(tag).decode('utf-8')
jsonPayloadToSend = '{"cid":"app","i":0,"pack":"' + pack + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid":{}'.format(self._uid) + ',"tag" : "' + tag + '"}'
cipher = AES.new(self._encryption_key, AES.MODE_GCM, nonce=GCM_IV)
cipher.update(GCM_ADD)

return self.FetchResult(cipher, self._ip_addr, self._port, self._timeout, jsonPayloadToSend)['dat']

def SetAcOptions(self, acOptions, newOptionsToOverride, optionValuesToOverride = None):
if not (optionValuesToOverride is None):
Expand All @@ -304,7 +350,18 @@ def SetAcOptions(self, acOptions, newOptionsToOverride, optionValuesToOverride =
def SendStateToAc(self, timeout):
_LOGGER.info('Start sending state to HVAC')
statePackJson = '{' + '"opt":["Pow","Mod","SetTem","WdSpd","Air","Blo","Health","SwhSlp","Lig","SwingLfRig","SwUpDn","Quiet","Tur","StHt","TemUn","HeatCoolType","TemRec","SvSt","SlpMod"],"p":[{Pow},{Mod},{SetTem},{WdSpd},{Air},{Blo},{Health},{SwhSlp},{Lig},{SwingLfRig},{SwUpDn},{Quiet},{Tur},{StHt},{TemUn},{HeatCoolType},{TemRec},{SvSt},{SlpMod}],"t":"cmd"'.format(**self._acOptions) + '}'
sentJsonPayload = '{"cid":"app","i":0,"pack":"' + base64.b64encode(self.CIPHER.encrypt(self.Pad(statePackJson).encode("utf8"))).decode('utf-8') + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid":{}'.format(self._uid) + '}'
if self.encryption_version == 1:
cipher = self.CIPHER
RobHofmann marked this conversation as resolved.
Show resolved Hide resolved
sentJsonPayload = '{"cid":"app","i":0,"pack":"' + base64.b64encode(cipher.encrypt(self.Pad(statePackJson).encode("utf8"))).decode('utf-8') + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid":{}'.format(self._uid) + '}'
elif self.encryption_version == 2:
cipher = AES.new(self._encryption_key, AES.MODE_GCM, nonce=GCM_IV)
cipher.update(GCM_ADD)
encrypted_data, tag = cipher.encrypt_and_digest(statePackJson.encode("utf8"))
pack = base64.b64encode(encrypted_data).decode('utf-8')
tag = base64.b64encode(tag).decode('utf-8')
sentJsonPayload = '{"cid":"app","i":0,"pack":"' + pack + '","t":"pack","tcid":"' + str(self._mac_addr) + '","uid":{}'.format(self._uid) + ',"tag":"' + tag +'"}'
cipher = AES.new(self._encryption_key, AES.MODE_GCM, nonce=GCM_IV)
cipher.update(GCM_ADD)
# Setup UDP Client & start transfering
clientSock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
clientSock.settimeout(timeout)
Expand All @@ -314,7 +371,10 @@ def SendStateToAc(self, timeout):
clientSock.close()
pack = receivedJson['pack']
base64decodedPack = base64.b64decode(pack)
decryptedPack = self.CIPHER.decrypt(base64decodedPack)
decryptedPack = cipher.decrypt(base64decodedPack)
if self.encryption_version == 2:
tag = receivedJson['tag']
cipher.verify(base64.b64decode(tag))
decodedPack = decryptedPack.decode("utf-8")
replacedPack = decodedPack.replace('\x0f', '').replace(decodedPack[decodedPack.rindex('}')+1:], '')
receivedJsonPayload = simplejson.loads(replacedPack)
Expand Down