MicroPythonBLEHID/hid_keystores.py at master · Heerkog/MicroPythonBLEHID · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
import esp32
import json
import binascii

# Class that represents a generic keystore
class KeyStore(object):

    def __init__(self):
        self.secrets = {}

    def add_secret(self, type, key, value):
        _key = (type, bytes(key))
        self.secrets[_key] = bytes(value)

    def get_secret(self, type, index, key):
        _key = (type, bytes(key) if key else None)
        value = None

        if key is None:
            i = 0
            for (t, _k), _val in self.secrets.items():
                if t == type:
                    if i == index:
                        value = _val
                    i += 1
        else:
            value = self.secrets.get(_key, None)

        return value

    def remove_secret(self, type, key):
        _key = (type, bytes(key))
        del self.secrets[_key]

    def has_secret(self, type, key):
        _key = (type, bytes(key))
        return _key in self.secrets

    def get_json_secrets(self):
        json_secrets = [
            (sec_type, binascii.b2a_base64(key, newline=False), binascii.b2a_base64(value, newline=False))
            for (sec_type, key), value in self.secrets.items()
        ]
        return json_secrets

    def add_json_secrets(self, entries):
        for sec_type, key, value in entries:
            self.secrets[sec_type, binascii.a2b_base64(key)] = binascii.a2b_base64(value)

    # Empty key store.
    def clear_secrets(self):
        self.secrets = {}

    def load_secrets(self):
        return

    def save_secrets(self):
        return


# Class that uses a JSON file to save the keystore
class JSONKeyStore(KeyStore):

    def __init__(self):
        super(JSONKeyStore, self).__init__()

    # Empty key store and save.
    def clear_secrets(self):
        super(JSONKeyStore, self).clear_secrets()
        self.save_secrets()

    # Load bonding keys from JSON file.
    def load_secrets(self):
        try:
            with open("keys.json", "r") as file:
                self.add_json_secrets(json.load(file))
        except:
            print("No secrets available")

    # Save bonding keys to JSON file.
    def save_secrets(self):
        try:
            with open("keys.json", "w") as file:
                json.dump(self.get_json_secrets(), file)
        except:
            print("Failed to save secrets")


# Class that uses non-volatile storage to save the keystore
class NVSKeyStore(KeyStore):

    def __init__(self):
        super(NVSKeyStore, self).__init__()
        self.nvsdata = esp32.NVS("BLE")

    # Empty key store and save.
    def clear_secrets(self):
        super(NVSKeyStore, self).clear_secrets()
        self.save_secrets()

    # Load bonding keys from non-volatile storage.
    def load_secrets(self):
        data = bytearray()
        num_bytes = 0

        try:
            num_bytes = self.nvsdata.get_blob("Keys", data)
        except:
            print("Failed to read NVS")

        if num_bytes > 0:
            try:
                data = bytearray(num_bytes)
                self.nvsdata.get_blob("Keys", data)
                entries = json.loads(str(data, 'utf-8'))
                self.add_json_secrets(entries)
            except:
                print("Failed to load secrets")
        else:
            print("No secrets available")

    # Save bonding keys to non-volatile storage.
    def save_secrets(self):
        try:
            self.nvsdata.set_blob("Keys", json.dumps(self.get_json_secrets()))
            self.nvsdata.commit()
        except:
            print("Failed to save secrets")