1# This file describes eFuses for ESP32-P4 chip
2#
3# SPDX-FileCopyrightText: 2023 Espressif Systems (Shanghai) CO LTD
4#
5# SPDX-License-Identifier: GPL-2.0-or-later
6
7import binascii
8import struct
9import sys
10import time
11
12from bitstring import BitArray
13
14import esptool
15
16import reedsolo
17
18from .mem_definition import EfuseDefineBlocks, EfuseDefineFields, EfuseDefineRegisters
19from .. import base_fields
20from .. import util
21
22
23class EfuseBlock(base_fields.EfuseBlockBase):
24    def len_of_burn_unit(self):
25        # The writing register window is 8 registers for any blocks.
26        # len in bytes
27        return 8 * 4
28
29    def __init__(self, parent, param, skip_read=False):
30        parent.read_coding_scheme()
31        super(EfuseBlock, self).__init__(parent, param, skip_read=skip_read)
32
33    def apply_coding_scheme(self):
34        data = self.get_raw(from_read=False)[::-1]
35        if len(data) < self.len_of_burn_unit():
36            add_empty_bytes = self.len_of_burn_unit() - len(data)
37            data = data + (b"\x00" * add_empty_bytes)
38        if self.get_coding_scheme() == self.parent.REGS.CODING_SCHEME_RS:
39            # takes 32 bytes
40            # apply RS encoding
41            rs = reedsolo.RSCodec(12)
42            # 32 byte of data + 12 bytes RS
43            encoded_data = rs.encode([x for x in data])
44            words = struct.unpack("<" + "I" * 11, encoded_data)
45            # returns 11 words (8 words of data + 3 words of RS coding)
46        else:
47            # takes 32 bytes
48            words = struct.unpack("<" + ("I" * (len(data) // 4)), data)
49            # returns 8 words
50        return words
51
52
53class EspEfuses(base_fields.EspEfusesBase):
54    """
55    Wrapper object to manage the efuse fields in a connected ESP bootloader
56    """
57
58    debug = False
59    do_not_confirm = False
60
61    def __init__(self, esp, skip_connect=False, debug=False, do_not_confirm=False):
62        self.Blocks = EfuseDefineBlocks()
63        self.Fields = EfuseDefineFields()
64        self.REGS = EfuseDefineRegisters
65        self.BURN_BLOCK_DATA_NAMES = self.Blocks.get_burn_block_data_names()
66        self.BLOCKS_FOR_KEYS = self.Blocks.get_blocks_for_keys()
67        self._esp = esp
68        self.debug = debug
69        self.do_not_confirm = do_not_confirm
70        if esp.CHIP_NAME != "ESP32-P4":
71            raise esptool.FatalError(
72                "Expected the 'esp' param for ESP32-P4 chip but got for '%s'."
73                % (esp.CHIP_NAME)
74            )
75        if not skip_connect:
76            flags = self._esp.get_security_info()["flags"]
77            GET_SECURITY_INFO_FLAG_SECURE_DOWNLOAD_ENABLE = 1 << 2
78            if flags & GET_SECURITY_INFO_FLAG_SECURE_DOWNLOAD_ENABLE:
79                raise esptool.FatalError(
80                    "Secure Download Mode is enabled. The tool can not read eFuses."
81                )
82        self.blocks = [
83            EfuseBlock(self, self.Blocks.get(block), skip_read=skip_connect)
84            for block in self.Blocks.BLOCKS
85        ]
86        if not skip_connect:
87            self.get_coding_scheme_warnings()
88        self.efuses = [EfuseField.convert(self, efuse) for efuse in self.Fields.EFUSES]
89        self.efuses += [
90            EfuseField.convert(self, efuse) for efuse in self.Fields.KEYBLOCKS
91        ]
92        if skip_connect:
93            self.efuses += [
94                EfuseField.convert(self, efuse)
95                for efuse in self.Fields.BLOCK2_CALIBRATION_EFUSES
96            ]
97        else:
98            # TODO add processing of self.Fields.BLOCK2_CALIBRATION_EFUSES
99            # if self["BLK_VERSION_MINOR"].get() == 1:
100            #     self.efuses += [
101            #         EfuseField.convert(self, efuse)
102            #         for efuse in self.Fields.BLOCK2_CALIBRATION_EFUSES
103            #     ]
104            self.efuses += [
105                EfuseField.convert(self, efuse) for efuse in self.Fields.CALC
106            ]
107
108    def __getitem__(self, efuse_name):
109        """Return the efuse field with the given name"""
110        for e in self.efuses:
111            if efuse_name == e.name or any(x == efuse_name for x in e.alt_names):
112                return e
113        new_fields = False
114        for efuse in self.Fields.BLOCK2_CALIBRATION_EFUSES:
115            if efuse.name == efuse_name or any(
116                x == efuse_name for x in efuse.alt_names
117            ):
118                self.efuses += [
119                    EfuseField.convert(self, efuse)
120                    for efuse in self.Fields.BLOCK2_CALIBRATION_EFUSES
121                ]
122                new_fields = True
123        if new_fields:
124            for e in self.efuses:
125                if efuse_name == e.name or any(x == efuse_name for x in e.alt_names):
126                    return e
127        raise KeyError
128
129    def read_coding_scheme(self):
130        self.coding_scheme = self.REGS.CODING_SCHEME_RS
131
132    def print_status_regs(self):
133        print("")
134        self.blocks[0].print_block(self.blocks[0].err_bitarray, "err__regs", debug=True)
135        print(
136            "{:27} 0x{:08x}".format(
137                "EFUSE_RD_RS_ERR0_REG", self.read_reg(self.REGS.EFUSE_RD_RS_ERR0_REG)
138            )
139        )
140        print(
141            "{:27} 0x{:08x}".format(
142                "EFUSE_RD_RS_ERR1_REG", self.read_reg(self.REGS.EFUSE_RD_RS_ERR1_REG)
143            )
144        )
145
146    def efuse_controller_setup(self):
147        self.set_efuse_timing()
148        self.clear_pgm_registers()
149        self.wait_efuse_idle()
150
151    def write_efuses(self, block):
152        self.efuse_program(block)
153        return self.get_coding_scheme_warnings(silent=True)
154
155    def clear_pgm_registers(self):
156        self.wait_efuse_idle()
157        for r in range(
158            self.REGS.EFUSE_PGM_DATA0_REG, self.REGS.EFUSE_PGM_DATA0_REG + 32, 4
159        ):
160            self.write_reg(r, 0)
161
162    def wait_efuse_idle(self):
163        deadline = time.time() + self.REGS.EFUSE_BURN_TIMEOUT
164        while time.time() < deadline:
165            cmds = self.REGS.EFUSE_PGM_CMD | self.REGS.EFUSE_READ_CMD
166            if self.read_reg(self.REGS.EFUSE_CMD_REG) & cmds == 0:
167                if self.read_reg(self.REGS.EFUSE_CMD_REG) & cmds == 0:
168                    # Due to a hardware error, we have to read READ_CMD again
169                    # to make sure the efuse clock is normal.
170                    # For PGM_CMD it is not necessary.
171                    return
172        raise esptool.FatalError(
173            "Timed out waiting for Efuse controller command to complete"
174        )
175
176    def efuse_program(self, block):
177        self.wait_efuse_idle()
178        self.write_reg(self.REGS.EFUSE_CONF_REG, self.REGS.EFUSE_WRITE_OP_CODE)
179        self.write_reg(self.REGS.EFUSE_CMD_REG, self.REGS.EFUSE_PGM_CMD | (block << 2))
180        self.wait_efuse_idle()
181        self.clear_pgm_registers()
182        self.efuse_read()
183
184    def efuse_read(self):
185        self.wait_efuse_idle()
186        self.write_reg(self.REGS.EFUSE_CONF_REG, self.REGS.EFUSE_READ_OP_CODE)
187        # need to add a delay after triggering EFUSE_READ_CMD, as ROM loader checks some
188        # efuse registers after each command is completed
189        # if ENABLE_SECURITY_DOWNLOAD or DIS_DOWNLOAD_MODE is enabled by the current cmd, then we need to try to reconnect to the chip.
190        try:
191            self.write_reg(
192                self.REGS.EFUSE_CMD_REG, self.REGS.EFUSE_READ_CMD, delay_after_us=1000
193            )
194            self.wait_efuse_idle()
195        except esptool.FatalError:
196            secure_download_mode_before = self._esp.secure_download_mode
197
198            try:
199                self._esp = self.reconnect_chip(self._esp)
200            except esptool.FatalError:
201                print("Can not re-connect to the chip")
202                if not self["DIS_DOWNLOAD_MODE"].get() and self[
203                    "DIS_DOWNLOAD_MODE"
204                ].get(from_read=False):
205                    print(
206                        "This is the correct behavior as we are actually burning "
207                        "DIS_DOWNLOAD_MODE which disables the connection to the chip"
208                    )
209                    print("DIS_DOWNLOAD_MODE is enabled")
210                    print("Successful")
211                    sys.exit(0)  # finish without errors
212                raise
213
214            print("Established a connection with the chip")
215            if self._esp.secure_download_mode and not secure_download_mode_before:
216                print("Secure download mode is enabled")
217                if not self["ENABLE_SECURITY_DOWNLOAD"].get() and self[
218                    "ENABLE_SECURITY_DOWNLOAD"
219                ].get(from_read=False):
220                    print(
221                        "espefuse tool can not continue to work in Secure download mode"
222                    )
223                    print("ENABLE_SECURITY_DOWNLOAD is enabled")
224                    print("Successful")
225                    sys.exit(0)  # finish without errors
226            raise
227
228    def set_efuse_timing(self):
229        """Set timing registers for burning efuses"""
230        # Configure clock
231        apb_freq = self.get_crystal_freq()
232        if apb_freq != 40:
233            raise esptool.FatalError(
234                "The eFuse supports only xtal=40M (xtal was %d)" % apb_freq
235            )
236        # keep default timing settings
237
238    def get_coding_scheme_warnings(self, silent=False):
239        """Check if the coding scheme has detected any errors."""
240        old_addr_reg = 0
241        reg_value = 0
242        ret_fail = False
243        for block in self.blocks:
244            if block.id == 0:
245                words = [
246                    self.read_reg(self.REGS.EFUSE_RD_REPEAT_ERR0_REG + offs * 4)
247                    for offs in range(5)
248                ]
249                block.err_bitarray.pos = 0
250                for word in reversed(words):
251                    block.err_bitarray.overwrite(BitArray("uint:32=%d" % word))
252                block.num_errors = block.err_bitarray.count(True)
253                block.fail = block.num_errors != 0
254            else:
255                addr_reg, err_num_mask, err_num_offs, fail_bit = self.REGS.BLOCK_ERRORS[
256                    block.id
257                ]
258                if err_num_mask is None or err_num_offs is None or fail_bit is None:
259                    continue
260                if addr_reg != old_addr_reg:
261                    old_addr_reg = addr_reg
262                    reg_value = self.read_reg(addr_reg)
263                block.fail = reg_value & (1 << fail_bit) != 0
264                block.num_errors = (reg_value >> err_num_offs) & err_num_mask
265            ret_fail |= block.fail
266            if not silent and (block.fail or block.num_errors):
267                print(
268                    "Error(s) in BLOCK%d [ERRORS:%d FAIL:%d]"
269                    % (block.id, block.num_errors, block.fail)
270                )
271        if (self.debug or ret_fail) and not silent:
272            self.print_status_regs()
273        return ret_fail
274
275    def summary(self):
276        # TODO add support set_flash_voltage - "Flash voltage (VDD_SPI)"
277        return ""
278
279
280class EfuseField(base_fields.EfuseFieldBase):
281    @staticmethod
282    def convert(parent, efuse):
283        return {
284            "mac": EfuseMacField,
285            "keypurpose": EfuseKeyPurposeField,
286            "t_sensor": EfuseTempSensor,
287            "adc_tp": EfuseAdcPointCalibration,
288        }.get(efuse.class_type, EfuseField)(parent, efuse)
289
290
291class EfuseTempSensor(EfuseField):
292    def get(self, from_read=True):
293        value = self.get_bitstring(from_read)
294        sig = -1 if value[0] else 1
295        return sig * value[1:].uint * 0.1
296
297
298class EfuseAdcPointCalibration(EfuseField):
299    def get(self, from_read=True):
300        STEP_SIZE = 4
301        value = self.get_bitstring(from_read)
302        sig = -1 if value[0] else 1
303        return sig * value[1:].uint * STEP_SIZE
304
305
306class EfuseMacField(EfuseField):
307    def check_format(self, new_value_str):
308        if new_value_str is None:
309            raise esptool.FatalError(
310                "Required MAC Address in AA:CD:EF:01:02:03 format!"
311            )
312        num_bytes = 8 if self.name == "MAC_EUI64" else 6
313        if new_value_str.count(":") != num_bytes - 1:
314            raise esptool.FatalError(
315                f"MAC Address needs to be a {num_bytes}-byte hexadecimal format "
316                "separated by colons (:)!"
317            )
318        hexad = new_value_str.replace(":", "").split(" ", 1)[0]
319        hexad = hexad.split(" ", 1)[0] if self.is_field_calculated() else hexad
320        if len(hexad) != num_bytes * 2:
321            raise esptool.FatalError(
322                f"MAC Address needs to be a {num_bytes}-byte hexadecimal number "
323                f"({num_bytes * 2} hexadecimal characters)!"
324            )
325        # order of bytearray = b'\xaa\xcd\xef\x01\x02\x03',
326        bindata = binascii.unhexlify(hexad)
327
328        if not self.is_field_calculated():
329            # unicast address check according to
330            # https://tools.ietf.org/html/rfc7042#section-2.1
331            if esptool.util.byte(bindata, 0) & 0x01:
332                raise esptool.FatalError("Custom MAC must be a unicast MAC!")
333        return bindata
334
335    def check(self):
336        errs, fail = self.parent.get_block_errors(self.block)
337        if errs != 0 or fail:
338            output = "Block%d has ERRORS:%d FAIL:%d" % (self.block, errs, fail)
339        else:
340            output = "OK"
341        return "(" + output + ")"
342
343    def get(self, from_read=True):
344        if self.name == "CUSTOM_MAC":
345            mac = self.get_raw(from_read)[::-1]
346        elif self.name == "MAC":
347            mac = self.get_raw(from_read)
348        elif self.name == "MAC_EUI64":
349            mac = self.parent["MAC"].get_bitstring(from_read).copy()
350            mac_ext = self.parent["MAC_EXT"].get_bitstring(from_read)
351            mac.insert(mac_ext, 24)
352            mac = mac.bytes
353        else:
354            mac = self.get_raw(from_read)
355        return "%s %s" % (util.hexify(mac, ":"), self.check())
356
357    def save(self, new_value):
358        def print_field(e, new_value):
359            print(
360                "    - '{}' ({}) {} -> {}".format(
361                    e.name, e.description, e.get_bitstring(), new_value
362                )
363            )
364
365        if self.name == "CUSTOM_MAC":
366            bitarray_mac = self.convert_to_bitstring(new_value)
367            print_field(self, bitarray_mac)
368            super(EfuseMacField, self).save(new_value)
369        else:
370            # Writing the BLOCK1 (MAC_SPI_8M_0) default MAC is not possible,
371            # as it's written in the factory.
372            raise esptool.FatalError(f"Burning {self.name} is not supported")
373
374
375# fmt: off
376class EfuseKeyPurposeField(EfuseField):
377    KEY_PURPOSES = [
378        ("USER",                         0,  None,       None,      "no_need_rd_protect"),   # User purposes (software-only use)
379        ("ECDSA_KEY",                    1,  None,       "Reverse", "need_rd_protect"),      # ECDSA key
380        ("XTS_AES_256_KEY_1",            2,  None,       "Reverse", "need_rd_protect"),      # XTS_AES_256_KEY_1 (flash/PSRAM encryption)
381        ("XTS_AES_256_KEY_2",            3,  None,       "Reverse", "need_rd_protect"),      # XTS_AES_256_KEY_2 (flash/PSRAM encryption)
382        ("XTS_AES_128_KEY",              4,  None,       "Reverse", "need_rd_protect"),      # XTS_AES_128_KEY (flash/PSRAM encryption)
383        ("HMAC_DOWN_ALL",                5,  None,       None,      "need_rd_protect"),      # HMAC Downstream mode
384        ("HMAC_DOWN_JTAG",               6,  None,       None,      "need_rd_protect"),      # JTAG soft enable key (uses HMAC Downstream mode)
385        ("HMAC_DOWN_DIGITAL_SIGNATURE",  7,  None,       None,      "need_rd_protect"),      # Digital Signature peripheral key (uses HMAC Downstream mode)
386        ("HMAC_UP",                      8,  None,       None,      "need_rd_protect"),      # HMAC Upstream mode
387        ("SECURE_BOOT_DIGEST0",          9,  "DIGEST",   None,      "no_need_rd_protect"),   # SECURE_BOOT_DIGEST0 (Secure Boot key digest)
388        ("SECURE_BOOT_DIGEST1",          10, "DIGEST",   None,      "no_need_rd_protect"),   # SECURE_BOOT_DIGEST1 (Secure Boot key digest)
389        ("SECURE_BOOT_DIGEST2",          11, "DIGEST",   None,      "no_need_rd_protect"),   # SECURE_BOOT_DIGEST2 (Secure Boot key digest)
390        ("KM_INIT_KEY",                  12, None,       None,      "need_rd_protect"),      # init key that is used for the generation of AES/ECDSA key
391        ("XTS_AES_256_KEY",              -1, "VIRTUAL",  None,      "no_need_rd_protect"),   # Virtual purpose splits to XTS_AES_256_KEY_1 and XTS_AES_256_KEY_2
392    ]
393# fmt: on
394    KEY_PURPOSES_NAME = [name[0] for name in KEY_PURPOSES]
395    DIGEST_KEY_PURPOSES = [name[0] for name in KEY_PURPOSES if name[2] == "DIGEST"]
396
397    def check_format(self, new_value_str):
398        # str convert to int: "XTS_AES_128_KEY" - > str(4)
399        # if int: 4 -> str(4)
400        raw_val = new_value_str
401        for purpose_name in self.KEY_PURPOSES:
402            if purpose_name[0] == new_value_str:
403                raw_val = str(purpose_name[1])
404                break
405        if raw_val.isdigit():
406            if int(raw_val) not in [p[1] for p in self.KEY_PURPOSES if p[1] > 0]:
407                raise esptool.FatalError("'%s' can not be set (value out of range)" % raw_val)
408        else:
409            raise esptool.FatalError("'%s' unknown name" % raw_val)
410        return raw_val
411
412    def need_reverse(self, new_key_purpose):
413        for key in self.KEY_PURPOSES:
414            if key[0] == new_key_purpose:
415                return key[3] == "Reverse"
416
417    def need_rd_protect(self, new_key_purpose):
418        for key in self.KEY_PURPOSES:
419            if key[0] == new_key_purpose:
420                return key[4] == "need_rd_protect"
421
422    def get(self, from_read=True):
423        for p in self.KEY_PURPOSES:
424            if p[1] == self.get_raw(from_read):
425                return p[0]
426        return "FORBIDDEN_STATE"
427
428    def get_name(self, raw_val):
429        for key in self.KEY_PURPOSES:
430            if key[1] == raw_val:
431                return key[0]
432
433    def save(self, new_value):
434        raw_val = int(self.check_format(str(new_value)))
435        return super(EfuseKeyPurposeField, self).save(raw_val)
436