eculocate

Unnamed repository; edit this file 'description' to name the repository.
Log | Files | Refs | README

ble.py (3255B)

      1 import asyncio
      2 from bleak import BleakClient,BleakScanner
      3 import sys
      4 import os
      5 
      6 import nacl
      7 from pathlib import Path
      8 from nacl.public import PrivateKey, Box, PublicKey
      9 from nacl.encoding import RawEncoder
     10 import sys
     11 
     12 class Network():
     13     def __init__(self, bytestr):
     14         self.rssi = bytestr[0]
     15         self.ssid = bytestr[1:7]
     16         self.name = bytestr[8:]
     17 
     18     def __str__(self):
     19         return f"<Network {self.name} ssid:{self.ssid} rssi:{self.rssi}>"
     20 
     21 UUIDS = {
     22     'connecting':       "76be25c2-0801-44fc-a97c-bcb56c69c926",
     23     'max_index':        "76be25c2-0802-44fc-a97c-bcb56c69c926",
     24     'current_index':    "76be25c2-0803-44fc-a97c-bcb56c69c926",
     25     'current_network':  "76be25c2-0804-44fc-a97c-bcb56c69c926",
     26     'secret':           "76be25c2-0805-44fc-a97c-bcb56c69c926",
     27 }
     28 
     29 
     30 class Encrypter():
     31     def __init__(self, private_key, public_key):
     32         self.sbox = Box(private_key, public_key)
     33 
     34     def encrypt(self, plaintext):
     35         mybytes = self.sbox.encrypt(plaintext.encode('UTF-8'))
     36         r = bytearray(mybytes)
     37         r.insert(0, len(mybytes))
     38         return r
     39 
     40 class Device():
     41     def __init__(self, ble_client, encrypter):
     42         self.client = ble_client
     43         self._networks = None
     44         self.max_index = 0
     45         self.encrypter = encrypter
     46 
     47     async def networks(self):
     48         if self._networks != None:
     49             return self._networks
     50 
     51         self.max_index = ord(await self.client.read_gatt_char(UUIDS['max_index']))
     52         self._networks = []
     53 
     54         for i in range(1, self.max_index + 1):
     55             await self.client.write_gatt_char(UUIDS['current_index'], bytes([i]))
     56             current_network = await self.client.read_gatt_char(UUIDS['current_network'])
     57             self._networks.append(Network(current_network))
     58         return self._networks
     59 
     60     async def choose_network(self, name, password):
     61         client = self.client
     62         secret = self.encrypter.encrypt(password)
     63         buf = bytearray(name)
     64         buf.insert(0, len(name))
     65         print(f"writing {len(buf)}, {len(secret)}")
     66         await client.write_gatt_char(UUIDS['secret'], buf + secret)
     67         await client.write_gatt_char(UUIDS['connecting'], bytes([1]))
     68 
     69 
     70 async def main(address, preferred):
     71     with open("keys/client.key", "rb") as f:
     72         secret = PrivateKey(f.read(), encoder=RawEncoder)
     73 
     74     with open("keys/device.pub", "rb") as f:
     75         rcpt = PublicKey(f.read(), encoder=RawEncoder)
     76 
     77     encrypter = Encrypter(secret, rcpt)
     78 
     79     devices = await BleakScanner.discover()
     80     for d in devices:
     81         if d.name == address or d.address == address:
     82             address = d.address
     83         print(d.name, d.address)
     84     print(f"connecting to {address}")
     85 
     86     async with BleakClient(address, timeout=60) as client:
     87         device = Device(client, encrypter)
     88         networks = await device.networks()
     89         for n in networks:
     90             print(f"network {n}")
     91         if preferred != None:
     92             name_bytes = preferred.encode(encoding="utf-8")
     93             await device.choose_network(name_bytes, os.getenv('PASSWORD'))
     94 
     95 
     96 if __name__ == '__main__':
     97     rcpt = sys.argv[1] if len(sys.argv) > 1 else "Eculocate"
     98     preferred = sys.argv[2] if len(sys.argv) > 2 else None
     99     asyncio.run(main(rcpt, preferred))