Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- #!/usr/bin/python3.5
- from datetime import datetime
- from Crypto.Cipher import AES
- import time
- import random
- import socket
- def discover(timeout=None):
- s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- s.connect(('8.8.8.8', 53)) # connecting to a UDP address doesn't send packets
- local_ip_address = s.getsockname()[0]
- address = local_ip_address.split('.')
- cs = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- cs.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- cs.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
- cs.bind(('',0))
- port = cs.getsockname()[1]
- starttime = time.time()
- devices = []
- timezone = time.timezone/-3600
- packet = bytearray(0x30)
- year = datetime.now().year
- if timezone < 0:
- packet[0x08] = 0xff + timezone - 1
- packet[0x09] = 0xff
- packet[0x0a] = 0xff
- packet[0x0b] = 0xff
- else:
- packet[0x08] = int(timezone)
- packet[0x09] = 0
- packet[0x0a] = 0
- packet[0x0b] = 0
- packet[0x0c] = year & 0xff
- packet[0x0d] = year >> 8
- packet[0x0e] = datetime.now().minute
- packet[0x0f] = datetime.now().hour
- subyear = str(year)[2:]
- packet[0x10] = int(subyear)
- packet[0x11] = datetime.now().isoweekday()
- packet[0x12] = datetime.now().day
- packet[0x13] = datetime.now().month
- packet[0x18] = int(address[0])
- packet[0x19] = int(address[1])
- packet[0x1a] = int(address[2])
- packet[0x1b] = int(address[3])
- packet[0x1c] = port & 0xff
- packet[0x1d] = port >> 8
- packet[0x26] = 6
- checksum = 0xbeaf
- for i in range(len(packet)):
- checksum += packet[i]
- checksum = checksum & 0xffff
- packet[0x20] = checksum & 0xff
- packet[0x21] = checksum >> 8
- cs.sendto(packet, ('255.255.255.255', 80))
- if timeout is None:
- response = cs.recvfrom(1024)
- responsepacket = bytearray(response[0])
- host = response[1]
- mac = responsepacket[0x3a:0x40]
- return device(host=host, mac=mac)
- else:
- while (time.time() - starttime) < timeout:
- cs.settimeout(timeout - (time.time() - starttime))
- try:
- response = cs.recvfrom(1024)
- except socket.timeout:
- return devices
- responsepacket = bytearray(response[0])
- host = response[1]
- mac = responsepacket[0x3a:0x40]
- devices.append(device(host=host, mac=mac))
- class device:
- def __init__(self, host, mac):
- self.host = host
- self.mac = mac
- self.count = random.randrange(0xffff)
- self.key = bytearray([0x09, 0x76, 0x28, 0x34, 0x3f, 0xe9, 0x9e, 0x23, 0x76, 0x5c, 0x15, 0x13, 0xac, 0xcf, 0x8b, 0x02])
- self.iv = bytearray([0x56, 0x2e, 0x17, 0x99, 0x6d, 0x09, 0x3d, 0x28, 0xdd, 0xb3, 0xba, 0x69, 0x5a, 0x2e, 0x6f, 0x58])
- self.id = bytearray([0, 0, 0, 0])
- self.cs = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- self.cs.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- self.cs.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
- self.cs.bind(('',0))
- def auth(self):
- payload = bytearray(0x50)
- payload[0x04] = 0x31
- payload[0x05] = 0x31
- payload[0x06] = 0x31
- payload[0x07] = 0x31
- payload[0x08] = 0x31
- payload[0x09] = 0x31
- payload[0x0a] = 0x31
- payload[0x0b] = 0x31
- payload[0x0c] = 0x31
- payload[0x0d] = 0x31
- payload[0x0e] = 0x31
- payload[0x0f] = 0x31
- payload[0x10] = 0x31
- payload[0x11] = 0x31
- payload[0x12] = 0x31
- payload[0x1e] = 0x01
- payload[0x2d] = 0x01
- payload[0x30] = ord('T')
- payload[0x31] = ord('e')
- payload[0x32] = ord('s')
- payload[0x33] = ord('t')
- payload[0x34] = ord(' ')
- payload[0x35] = ord(' ')
- payload[0x36] = ord('1')
- response = self.send_packet(0x65, payload)
- enc_payload = response[0x38:]
- aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv))
- payload = aes.decrypt(bytes(enc_payload))
- self.id = payload[0x00:0x04]
- self.key = payload[0x04:0x14]
- def send_packet(self, command, payload):
- packet = bytearray(0x38)
- packet[0x00] = 0x5a
- packet[0x01] = 0xa5
- packet[0x02] = 0xaa
- packet[0x03] = 0x55
- packet[0x04] = 0x5a
- packet[0x05] = 0xa5
- packet[0x06] = 0xaa
- packet[0x07] = 0x55
- packet[0x24] = 0x2a
- packet[0x25] = 0x27
- packet[0x26] = command
- packet[0x28] = self.count & 0xff
- packet[0x29] = self.count >> 8
- packet[0x2a] = self.mac[0]
- packet[0x2b] = self.mac[1]
- packet[0x2c] = self.mac[2]
- packet[0x2d] = self.mac[3]
- packet[0x2e] = self.mac[4]
- packet[0x2f] = self.mac[5]
- packet[0x30] = self.id[0]
- packet[0x31] = self.id[1]
- packet[0x32] = self.id[2]
- packet[0x33] = self.id[3]
- checksum = 0xbeaf
- for i in range(len(payload)):
- checksum += payload[i]
- checksum = checksum & 0xffff
- aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv))
- payload = aes.encrypt(bytes(payload))
- packet[0x34] = checksum & 0xff
- packet[0x35] = checksum >> 8
- for i in range(len(payload)):
- packet.append(payload[i])
- checksum = 0xbeaf
- for i in range(len(packet)):
- checksum += packet[i]
- checksum = checksum & 0xffff
- packet[0x20] = checksum & 0xff
- packet[0x21] = checksum >> 8
- self.cs.sendto(packet, self.host)
- response = self.cs.recvfrom(1024)
- return response[0]
- def send_data(self, data):
- packet = bytearray([0x02, 0x00, 0x00, 0x00])
- packet += data
- self.send_packet(0x6a, packet)
- def enter_learning(self):
- packet = bytearray(16)
- packet[0] = 3
- self.send_packet(0x6a, packet)
- def check_sensors(self):
- packet = bytearray(16)
- packet[0] = 1
- response = self.send_packet(0x6a, packet)
- err = response[0x22] | (response[0x23] << 8)
- if err == 0:
- data = {}
- aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv))
- payload = aes.decrypt(bytes(response[0x38:]))
- data['temperature'] = (payload[0x4] * 10 + payload[0x5]) / 10.0
- data['humidity'] = (payload[0x6] * 10 + payload[0x7]) / 10.0
- light = payload[0x8]
- if light == 0:
- data['light'] = 'dark'
- elif light == 1:
- data['light'] = 'dim'
- elif light == 2:
- data['light'] = 'normal'
- elif light == 3:
- data['light'] = 'bright'
- else:
- data['light'] = 'unknown'
- air_quality = payload[0x0a]
- if air_quality == 0:
- data['air_quality'] = 'excellent'
- elif air_quality == 1:
- data['air_quality'] = 'good'
- elif air_quality == 2:
- data['air_quality'] = 'normal'
- elif air_quality == 3:
- data['air_quality'] = 'bad'
- else:
- data['air_quality'] = 'unknown'
- noise = payload[0xc]
- if noise == 0:
- data['noise'] = 'quiet'
- elif noise == 1:
- data['noise'] = 'normal'
- elif noise == 2:
- data['noise'] = 'noisy'
- else:
- data['noise'] = 'unknown'
- return data
- def check_temperature(self):
- packet = bytearray(16)
- packet[0] = 1
- response = self.send_packet(0x6a, packet)
- err = response[0x22] | (response[0x23] << 8)
- if err == 0:
- aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv))
- payload = aes.decrypt(bytes(response[0x38:]))
- temp = (payload[0x4] * 10 + payload[0x5]) / 10.0
- return temp
- def check_data(self):
- packet = bytearray(16)
- packet[0] = 4
- response = self.send_packet(0x6a, packet)
- err = response[0x22] | (response[0x23] << 8)
- if err == 0:
- aes = AES.new(bytes(self.key), AES.MODE_CBC, bytes(self.iv))
- payload = aes.decrypt(bytes(response[0x38:]))
- return payload[0x04:]
- class rm2(device):
- def __init__ (self):
- device.__init__(self, None, None)
- def discover(self):
- dev = discover()
- self.host = dev.host
- self.mac = dev.mac
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement