trezor-firmware/storage/tests/python/src/storage.py

import hashlib
import sys

from . import consts, crypto, helpers, prng
from .norcow import Norcow
from .pin_log import PinLog


class Storage:
    def __init__(self, flash_byte_access: bool = True):
        self.initialized = False
        self.unlocked = False
        self.dek = None
        self.sak = None
        self.nc = Norcow(flash_byte_access=flash_byte_access)
        self.pin_log = PinLog(self.nc)

    def init(self, hardware_salt: bytes = b""):
        """
        Initializes storage. Normally we would check if EDEK is already present,
        but we simplify things in the python version and suppose we are starting
        a new storage each time.
        """
        self.nc.init()
        self.initialized = True
        self.hw_salt_hash = hashlib.sha256(hardware_salt).digest()

        edek_esak_pvc = self.nc.get(consts.EDEK_ESEK_PVC_KEY)
        if not edek_esak_pvc:
            self._init_pin()

    def _init_pin(self):
        """
        Initalizes PIN counters, generates random
        Data Encryption Key and Storage Authentication Key
        """
        self.dek = prng.random_buffer(consts.DEK_SIZE)
        self.sak = prng.random_buffer(consts.SAK_SIZE)

        self.nc.set(consts.SAT_KEY, crypto.init_hmacs(self.sak))
        self._set_encrypt(consts.VERSION_KEY, consts.NORCOW_VERSION)
        self.nc.set(consts.UNAUTH_VERSION_KEY, consts.NORCOW_VERSION)
        self.nc.set(consts.STORAGE_UPGRADED_KEY, consts.FALSE_WORD)
        self.pin_log.init()
        self._set_wipe_code(consts.WIPE_CODE_EMPTY)
        self._set_pin(consts.PIN_EMPTY)
        self.unlocked = False

    def _set_pin(self, pin: str):
        random_salt = prng.random_buffer(consts.PIN_SALT_SIZE)
        salt = self.hw_salt_hash + random_salt
        kek, keiv = crypto.derive_kek_keiv(salt, pin)

        # Encrypted Data Encryption Key and Encrypted Storage Authentication Key
        edek_esak, tag = crypto.chacha_poly_encrypt(kek, keiv, self.dek + self.sak)
        # Pin Verification Code
        pvc = tag[: consts.PVC_SIZE]

        self.nc.set(consts.EDEK_ESEK_PVC_KEY, random_salt + edek_esak + pvc)
        if pin == consts.PIN_EMPTY:
            self._set_bool(consts.PIN_NOT_SET_KEY, True)
        else:
            self._set_bool(consts.PIN_NOT_SET_KEY, False)

    def _set_wipe_code(self, wipe_code: str):
        if wipe_code == consts.PIN_EMPTY:
            wipe_code = consts.WIPE_CODE_EMPTY
        wipe_code_bytes = wipe_code.encode()
        salt = prng.random_buffer(consts.WIPE_CODE_SALT_SIZE)
        tag = crypto._hmac(salt, wipe_code_bytes)[: consts.WIPE_CODE_TAG_SIZE]
        self.nc.set(consts.WIPE_CODE_DATA_KEY, wipe_code_bytes + salt + tag)

    def wipe(self):
        self.nc.wipe()
        self._init_pin()

    def check_pin(self, pin: str) -> bool:
        self.pin_log.write_attempt()

        data = self.nc.get(consts.EDEK_ESEK_PVC_KEY)
        salt = self.hw_salt_hash + data[: consts.PIN_SALT_SIZE]
        edek_esak = data[consts.PIN_SALT_SIZE : -consts.PVC_SIZE]
        pvc = data[-consts.PVC_SIZE :]

        try:
            dek, sak = crypto.decrypt_edek_esak(pin, salt, edek_esak, pvc)
            self.pin_log.write_success()
            self.dek = dek
            self.sak = sak
            return True
        except crypto.InvalidPinError:
            fails = self.pin_log.get_failures_count()
            if fails >= consts.PIN_MAX_TRIES:
                self.wipe()
            return False

    def lock(self) -> None:
        self.unlocked = False

    def unlock(self, pin: str) -> bool:
        if not self.initialized or not self.check_pin(pin):
            return False

        version = self._decrypt(consts.VERSION_KEY)
        if version != consts.NORCOW_VERSION:
            return False

        self.unlocked = True
        return True

    def has_pin(self) -> bool:
        val = self.nc.get(consts.PIN_NOT_SET_KEY)
        return val != consts.TRUE_BYTE

    def get_pin_rem(self) -> int:
        return consts.PIN_MAX_TRIES - self.pin_log.get_failures_count()

    def change_pin(self, oldpin: str, newpin: str) -> bool:
        if not self.initialized or not self.unlocked:
            return False
        if not self.check_pin(oldpin):
            return False
        self._set_pin(newpin)
        return True

    def get(self, key: int) -> bytes:
        app = key >> 8
        if not self.initialized or consts.is_app_private(app):
            raise RuntimeError("Storage not initialized or app is private")
        if not self.unlocked and not consts.is_app_public(app):
            # public fields can be read from an unlocked device
            raise RuntimeError("Storage locked")
        if consts.is_app_public(app):
            value = self.nc.get(key)
        else:
            value = self._get_encrypted(key)
        if value is False:
            raise RuntimeError("Failed to find key in storage.")
        return value

    def set(self, key: int, val: bytes) -> bool:
        app = key >> 8
        self._check_lock(app)
        if consts.is_app_public(app):
            return self.nc.set(key, val)
        return self._set_encrypt(key, val)

    def set_counter(self, key: int, val: int):
        app = key >> 8
        if not consts.is_app_public(app):
            raise RuntimeError("Counter can be set only for public items")

        if val > consts.UINT32_MAX:
            raise RuntimeError("Failed to set value in storage.")

        counter = val.to_bytes(4, sys.byteorder)

        if self.nc.is_byte_access():
            counter += bytearray(b"\xFF" * consts.COUNTER_TAIL_SIZE)
        self.set(key, counter)

    def next_counter(self, key: int) -> int:
        app = key >> 8
        self._check_lock(app)

        current = self.nc.get(key)
        if current is False:
            self.set_counter(key, 0)
            return 0

        base = int.from_bytes(current[:4], sys.byteorder)

        if self.nc.is_byte_access():
            base = int.from_bytes(current[:4], sys.byteorder)
            tail = helpers.to_int_by_words(current[4:])
            tail_count = f"{tail:064b}".count("0")
            increased_count = base + tail_count + 1
            if increased_count > consts.UINT32_MAX:
                raise RuntimeError("Failed to set value in storage.")

            if tail_count == consts.COUNTER_MAX_TAIL:
                self.set_counter(key, increased_count)
                return increased_count

            self.set(
                key,
                current[:4]
                + helpers.to_bytes_by_words(tail >> 1, consts.COUNTER_TAIL_SIZE),
            )
        else:
            increased_count = base + 1
            if increased_count > consts.UINT32_MAX:
                raise RuntimeError("Failed to set value in storage.")

            self.set_counter(key, increased_count)
        return increased_count

    def delete(self, key: int) -> bool:
        app = key >> 8
        self._check_lock(app)
        ret = self.nc.delete(key)
        if consts.is_app_protected(app):
            sat = self._calculate_authentication_tag()
            self.nc.set(consts.SAT_KEY, sat)
        return ret

    def _check_lock(self, app: int):
        if not self.initialized or consts.is_app_private(app):
            raise RuntimeError("Storage not initialized or app is private")
        if not self.unlocked and not consts.is_app_lock_writable(app):
            raise RuntimeError("Storage locked and app is not public-writable")

    def _get_encrypted(self, key: int) -> bytes:
        if not consts.is_app_protected(key):
            raise RuntimeError("Only protected values are encrypted")
        sat = self.nc.get(consts.SAT_KEY)
        if not sat:
            raise RuntimeError("SAT not found")
        if sat != self._calculate_authentication_tag():
            raise RuntimeError("Storage authentication tag mismatch")
        return self._decrypt(key)

    def _decrypt(self, key: int) -> bytes:
        data = self.nc.get(key)
        iv = data[: consts.CHACHA_IV_SIZE]
        # cipher text with MAC

        tag = data[len(data) - consts.POLY1305_MAC_SIZE :]
        ciphertext = data[consts.CHACHA_IV_SIZE : len(data) - consts.POLY1305_MAC_SIZE]
        return crypto.chacha_poly_decrypt(
            self.dek, key, iv, ciphertext + tag, key.to_bytes(2, sys.byteorder)
        )

    def _set_encrypt(self, key: int, val: bytes):
        # In C, data are preallocated beforehand for encrypted values,
        # to match the behaviour we do the same.
        preallocate = b"\xFF" * (
            consts.CHACHA_IV_SIZE + len(val) + consts.POLY1305_MAC_SIZE
        )
        self.nc.set(key, preallocate)
        if consts.is_app_protected(key >> 8):
            sat = self._calculate_authentication_tag()
            self.nc.set(consts.SAT_KEY, sat)

        iv = prng.random_buffer(consts.CHACHA_IV_SIZE)
        cipher_text, tag = crypto.chacha_poly_encrypt(
            self.dek, iv, val, key.to_bytes(2, sys.byteorder)
        )
        return self.nc.replace(key, iv + cipher_text + tag)

    def _calculate_authentication_tag(self) -> bytes:
        keys = []
        for key in self.nc._get_all_keys():
            if consts.is_app_protected(key >> 8):
                keys.append(key.to_bytes(2, sys.byteorder))
        if not keys:
            return crypto.init_hmacs(self.sak)
        return crypto.calculate_hmacs(self.sak, keys)

    def _set_bool(self, key: int, val: bool) -> bool:
        if val:
            return self.nc.set(key, consts.TRUE_BYTE)
        # False is stored as an empty value
        return self.nc.set(key, consts.FALSE_BYTE)

    def _dump(self) -> bytes:
        return self.nc._dump()