feat(core): add support for quad-word only storage

[no changelog]
pull/3522/head
tychovrahe 9 months ago committed by TychoVrahe
parent f8eae3f023
commit 743ed413f6

@ -57,6 +57,7 @@ SOURCE_MOD += [
'embed/extmod/modtrezorconfig/modtrezorconfig.c',
'vendor/trezor-storage/norcow.c',
'vendor/trezor-storage/storage.c',
'vendor/trezor-storage/storage_utils.c',
'vendor/trezor-storage/flash_common.c',
]

@ -58,6 +58,7 @@ SOURCE_MOD += [
'embed/extmod/modtrezorconfig/modtrezorconfig.c',
'vendor/trezor-storage/norcow.c',
'vendor/trezor-storage/storage.c',
'vendor/trezor-storage/storage_utils.c',
'vendor/trezor-storage/flash_common.c',
]

@ -30,6 +30,6 @@
/*
* Current storage version.
*/
#define NORCOW_VERSION ((uint32_t)0x00000003)
#define NORCOW_VERSION ((uint32_t)0x00000004)
#endif

@ -126,6 +126,7 @@ OBJS += ../vendor/trezor-crypto/nem.o
OBJS += ../vendor/QR-Code-generator/c/qrcodegen.o
OBJS += ../vendor/trezor-storage/storage.o
OBJS += ../vendor/trezor-storage/storage_utils.o
OBJS += ../vendor/trezor-storage/norcow.o
OBJS += ../vendor/nanopb/pb_common.o

@ -38,6 +38,6 @@ extern const flash_area_t STORAGE_AREAS[NORCOW_SECTOR_COUNT];
/*
* Current storage version.
*/
#define NORCOW_VERSION ((uint32_t)0x00000003)
#define NORCOW_VERSION ((uint32_t)0x00000004)
#endif

@ -123,6 +123,9 @@ secbool flash_area_write_word(const flash_area_t *area, uint32_t offset,
secbool flash_area_write_quadword(const flash_area_t *area, uint32_t offset,
const uint32_t *data) {
if (offset % 16 != 0) {
return secfalse;
}
for (int i = 0; i < 4; i++) {
if (sectrue !=
flash_area_write_word(area, offset + i * sizeof(uint32_t), data[i])) {

@ -22,14 +22,13 @@
#include "common.h"
#include "flash.h"
#include "norcow.h"
#include "storage_utils.h"
// NRC2 = 4e524332
#define NORCOW_MAGIC ((uint32_t)0x3243524e)
// NRCW = 4e524357
#define NORCOW_MAGIC_V0 ((uint32_t)0x5743524e)
#define NORCOW_WORD_SIZE (sizeof(uint32_t))
#define NORCOW_PREFIX_LEN NORCOW_WORD_SIZE
#define NORCOW_MAGIC_LEN NORCOW_WORD_SIZE
#define NORCOW_VERSION_LEN NORCOW_WORD_SIZE
@ -39,10 +38,33 @@
// The key value which is used to indicate that the entry has been deleted.
#define NORCOW_KEY_DELETED (0x0000)
#define NORCOW_KEY_LEN 2
#define NORCOW_LEN_LEN 2
#ifdef FLASH_BYTE_ACCESS
#define COUNTER_TAIL_WORDS 2
#define NORCOW_WORD_SIZE (sizeof(uint32_t))
#define NORCOW_MAX_PREFIX_LEN (NORCOW_WORD_SIZE)
#define ALIGN(X) ((X) = ((X) + 3) & ~3)
// The offset from the beginning of the sector where stored items start.
#define NORCOW_STORAGE_START \
(NORCOW_HEADER_LEN + NORCOW_MAGIC_LEN + NORCOW_VERSION_LEN)
#else
#define COUNTER_TAIL_WORDS 0
#define NORCOW_WORD_SIZE (4 * sizeof(uint32_t))
// Small items are encoded more efficiently.
#define NORCOW_SMALL_ITEM_SIZE \
(NORCOW_WORD_SIZE - NORCOW_LEN_LEN - NORCOW_KEY_LEN)
// Larger items store first part of data with the same QW as key
#define NORCOW_DELETED_FLAG_LEN 1
#define NORCOW_DATA_OPT_SIZE (NORCOW_WORD_SIZE - NORCOW_DELETED_FLAG_LEN)
#define NORCOW_MAX_PREFIX_LEN (NORCOW_WORD_SIZE + NORCOW_DELETED_FLAG_LEN)
#define ALIGN(X) ((X) = ((X) + 0xF) & ~0xF)
// The offset from the beginning of the sector where stored items start.
#define NORCOW_STORAGE_START (NORCOW_HEADER_LEN + NORCOW_WORD_SIZE)
#endif
// The index of the active reading sector and writing sector. These should be
// equal except when storage version upgrade or compaction is in progress.
static uint8_t norcow_active_sector = 0;
@ -54,6 +76,14 @@ static uint32_t norcow_active_version = 0;
// The offset of the first free item in the writing sector.
static uint32_t norcow_free_offset = 0;
static uint16_t norcow_write_buffer_flashed = 0;
#ifndef FLASH_BYTE_ACCESS
static uint32_t norcow_write_buffer[4] = {0};
static uint16_t norcow_write_buffer_filled = 0;
static uint16_t norcow_write_buffer_filled_data = 0;
static int32_t norcow_write_buffer_key = -1;
#endif
/*
* Returns pointer to sector, starting with offset
* Fails when there is not enough space for data of given size
@ -66,21 +96,24 @@ static const void *norcow_ptr(uint8_t sector, uint32_t offset, uint32_t size) {
/*
* Writes data to given sector, starting from offset
*/
static secbool norcow_write(uint8_t sector, uint32_t offset, uint32_t prefix,
#ifdef FLASH_BYTE_ACCESS
static secbool norcow_write(uint8_t sector, uint32_t offset, uint16_t key,
const uint8_t *data, uint16_t len) {
if (sector >= NORCOW_SECTOR_COUNT) {
return secfalse;
}
if (offset + NORCOW_PREFIX_LEN + len > NORCOW_SECTOR_SIZE) {
if (offset + NORCOW_MAX_PREFIX_LEN + len > NORCOW_SECTOR_SIZE) {
return secfalse;
}
uint32_t prefix = ((uint32_t)len << 16) | key;
ensure(flash_unlock_write(), NULL);
// write prefix
ensure(flash_area_write_word(&STORAGE_AREAS[sector], offset, prefix), NULL);
offset += NORCOW_PREFIX_LEN;
offset += NORCOW_MAX_PREFIX_LEN;
if (data != NULL) {
// write data
@ -100,6 +133,94 @@ static secbool norcow_write(uint8_t sector, uint32_t offset, uint32_t prefix,
ensure(flash_lock_write(), NULL);
return sectrue;
}
#else
static secbool norcow_write(uint8_t sector, uint32_t offset, uint16_t key,
const uint8_t *data, uint16_t len) {
if (sector >= NORCOW_SECTOR_COUNT) {
return secfalse;
}
if (len <= NORCOW_SMALL_ITEM_SIZE) {
// the whole item fits into one quadword, let's not waste space
uint16_t len_adjusted = 0x10;
if (offset + len_adjusted > NORCOW_SECTOR_SIZE) {
return secfalse;
}
ensure(flash_unlock_write(), NULL);
uint32_t d[4];
d[0] = len | ((uint32_t)key << 16);
d[1] = 0;
d[2] = 0;
d[3] = 0;
if (len > 0) {
memcpy(&d[1], data, len); // write data
}
ensure(flash_area_write_quadword(&STORAGE_AREAS[sector], offset, d), NULL);
ensure(flash_lock_write(), NULL);
} else {
uint16_t len_adjusted = ((len) + 0xF) & ~0xF;
if (offset + NORCOW_MAX_PREFIX_LEN + len_adjusted > NORCOW_SECTOR_SIZE) {
return secfalse;
}
ensure(flash_unlock_write(), NULL);
uint32_t d[4];
d[0] = len | ((uint32_t)key << 16);
d[1] = 0xFFFFFFFF;
d[2] = 0xFFFFFFFF;
d[3] = 0xFFFFFFFF;
// write len
ensure(flash_area_write_quadword(&STORAGE_AREAS[sector], offset, d), NULL);
offset += NORCOW_WORD_SIZE;
if (data != NULL) {
memset(d, 0, sizeof(d));
d[0] = 0xFE;
if (len >= NORCOW_DATA_OPT_SIZE) {
// write key and first data part
memcpy(&(((uint8_t *)d)[NORCOW_DELETED_FLAG_LEN]), data,
NORCOW_DATA_OPT_SIZE);
ensure(flash_area_write_quadword(&STORAGE_AREAS[sector], offset, d),
NULL);
offset += NORCOW_WORD_SIZE;
data += NORCOW_DATA_OPT_SIZE;
len -= NORCOW_DATA_OPT_SIZE;
} else {
// write key and first data part
memcpy(&(((uint8_t *)d)[NORCOW_DELETED_FLAG_LEN]), data, len);
ensure(flash_area_write_quadword(&STORAGE_AREAS[sector], offset, d),
NULL);
offset += NORCOW_WORD_SIZE;
data += len;
len = 0;
}
while (len > 0) {
memset(d, 0, sizeof(d));
uint16_t data_to_write =
len > NORCOW_WORD_SIZE ? NORCOW_WORD_SIZE : len;
memcpy(d, data, data_to_write);
ensure(flash_area_write_quadword(&STORAGE_AREAS[sector], offset, d),
NULL);
offset += NORCOW_WORD_SIZE;
data += data_to_write;
len -= data_to_write;
}
memset(d, 0, sizeof(norcow_write_buffer));
}
ensure(flash_lock_write(), NULL);
}
return sectrue;
}
#endif
/*
* Erases sector (and sets a magic)
@ -126,16 +247,36 @@ static void erase_sector(uint8_t sector, secbool set_magic) {
#endif
if (sectrue == set_magic) {
ensure(norcow_write(sector, NORCOW_HEADER_LEN, NORCOW_MAGIC, NULL, 0),
"set magic failed");
ensure(norcow_write(sector, NORCOW_HEADER_LEN + NORCOW_MAGIC_LEN,
~NORCOW_VERSION, NULL, 0),
#ifdef FLASH_BYTE_ACCESS
ensure(flash_unlock_write(), NULL);
ensure(flash_area_write_word(&STORAGE_AREAS[sector], NORCOW_HEADER_LEN,
NORCOW_MAGIC),
NULL);
ensure(flash_area_write_word(&STORAGE_AREAS[sector],
NORCOW_HEADER_LEN + NORCOW_MAGIC_LEN,
~NORCOW_VERSION),
"set version failed");
ensure(flash_lock_write(), NULL);
#else
uint32_t d[4];
d[0] = NORCOW_MAGIC;
d[1] = ~NORCOW_VERSION;
d[2] = 0xFFFFFFFF;
d[3] = 0xFFFFFFFF;
ensure(flash_unlock_write(), NULL);
ensure(
flash_area_write_quadword(&STORAGE_AREAS[sector], NORCOW_HEADER_LEN, d),
"set magic and version failed");
ensure(flash_lock_write(), NULL);
#endif
}
}
#define ALIGN4(X) (X) = ((X) + 3) & ~3
/*
* Reads one item starting from offset
*/
@ -143,23 +284,61 @@ static secbool read_item(uint8_t sector, uint32_t offset, uint16_t *key,
const void **val, uint16_t *len, uint32_t *pos) {
*pos = offset;
const void *k = norcow_ptr(sector, *pos, 2);
#ifdef FLASH_BYTE_ACCESS
const void *k = norcow_ptr(sector, *pos, NORCOW_KEY_LEN);
if (k == NULL) return secfalse;
*pos += 2;
*pos += NORCOW_KEY_LEN;
memcpy(key, k, sizeof(uint16_t));
if (*key == NORCOW_KEY_FREE) {
return secfalse;
}
const void *l = norcow_ptr(sector, *pos, 2);
const void *l = norcow_ptr(sector, *pos, NORCOW_LEN_LEN);
if (l == NULL) return secfalse;
*pos += NORCOW_LEN_LEN;
memcpy(len, l, sizeof(uint16_t));
#else
const void *l = norcow_ptr(sector, *pos, NORCOW_LEN_LEN);
if (l == NULL) return secfalse;
*pos += 2;
memcpy(len, l, sizeof(uint16_t));
if (*len <= NORCOW_SMALL_ITEM_SIZE) {
*pos += NORCOW_LEN_LEN;
const void *k = norcow_ptr(sector, *pos, NORCOW_KEY_LEN);
if (k == NULL) return secfalse;
memcpy(key, k, sizeof(uint16_t));
if (*key == NORCOW_KEY_FREE) {
return secfalse;
}
*pos += NORCOW_KEY_LEN;
} else {
*pos += NORCOW_LEN_LEN;
const void *k = norcow_ptr(sector, *pos, NORCOW_KEY_LEN);
if (k == NULL) return secfalse;
*pos += (NORCOW_KEY_LEN + NORCOW_SMALL_ITEM_SIZE);
const void *flg = norcow_ptr(sector, *pos, NORCOW_DELETED_FLAG_LEN);
if (flg == NULL) return secfalse;
*pos += NORCOW_DELETED_FLAG_LEN;
if (*((const uint8_t *)flg) == 0) {
// Deleted item.
*key = NORCOW_KEY_DELETED;
} else {
memcpy(key, k, sizeof(uint16_t));
if (*key == NORCOW_KEY_FREE) {
return secfalse;
}
}
}
#endif
*val = norcow_ptr(sector, *pos, *len);
if (*val == NULL) return secfalse;
*pos += *len;
ALIGN4(*pos);
ALIGN(*pos);
return sectrue;
}
@ -168,10 +347,19 @@ static secbool read_item(uint8_t sector, uint32_t offset, uint16_t *key,
*/
static secbool write_item(uint8_t sector, uint32_t offset, uint16_t key,
const void *val, uint16_t len, uint32_t *pos) {
uint32_t prefix = ((uint32_t)len << 16) | key;
*pos = offset + NORCOW_PREFIX_LEN + len;
ALIGN4(*pos);
return norcow_write(sector, offset, prefix, val, len);
#ifndef FLASH_BYTE_ACCESS
if (len <= NORCOW_SMALL_ITEM_SIZE) {
*pos = offset + NORCOW_WORD_SIZE;
} else {
*pos = offset + NORCOW_MAX_PREFIX_LEN + len;
ALIGN(*pos);
}
#else
*pos = offset + NORCOW_MAX_PREFIX_LEN + len;
ALIGN(*pos);
#endif
return norcow_write(sector, offset, key, val, len);
}
/*
@ -417,6 +605,7 @@ secbool norcow_set_ex(uint16_t key, const void *val, uint16_t len,
return secfalse;
}
const flash_area_t *area = &STORAGE_AREAS[norcow_write_sector];
secbool ret = secfalse;
const void *ptr = NULL;
uint16_t len_old = 0;
@ -428,19 +617,100 @@ secbool norcow_set_ex(uint16_t key, const void *val, uint16_t len,
offset =
(const uint8_t *)ptr -
(const uint8_t *)norcow_ptr(norcow_write_sector, 0, NORCOW_SECTOR_SIZE);
#ifdef FLASH_BYTE_ACCESS
if (val != NULL && len_old == len) {
ret = sectrue;
ensure(flash_unlock_write(), NULL);
for (uint16_t i = 0; i < len; i++) {
if (sectrue !=
flash_area_write_byte(&STORAGE_AREAS[norcow_write_sector],
offset + i, ((const uint8_t *)val)[i])) {
if (sectrue != flash_area_write_byte(area, offset + i,
((const uint8_t *)val)[i])) {
ret = secfalse;
break;
}
}
ensure(flash_lock_write(), NULL);
}
#else
if (val != NULL && len_old == len) {
if (len <= NORCOW_SMALL_ITEM_SIZE) {
secbool updatable = sectrue;
for (uint16_t i = 0; i < len; i++) {
if (((const uint8_t *)val)[i] != ((const uint8_t *)ptr)[i]) {
updatable = secfalse;
break;
}
}
if (updatable == sectrue) {
ret = sectrue;
ensure(flash_unlock_write(), NULL);
uint32_t d[4];
d[0] = len | ((uint32_t)key << 16);
d[1] = 0;
d[2] = 0;
d[3] = 0;
if (len > 0) {
memcpy(&d[1], val, len); // write len
}
ensure(flash_area_write_quadword(area, offset - 4, d), NULL);
ensure(flash_lock_write(), NULL);
}
} else {
secbool updatable = sectrue;
for (uint16_t i = 0; i < len; i++) {
if (((const uint8_t *)val)[i] != ((const uint8_t *)ptr)[i]) {
updatable = secfalse;
break;
}
}
if (updatable == sectrue) {
ret = sectrue;
uint8_t *v = (uint8_t *)val;
int tmp_len = len;
ensure(flash_unlock_write(), NULL);
uint32_t d[4] = {0};
d[0] = 0xFF;
uint16_t data_to_copy =
((size_t)tmp_len > NORCOW_WORD_SIZE ? NORCOW_WORD_SIZE
: (size_t)tmp_len);
memcpy(&(((uint8_t *)norcow_write_buffer)[NORCOW_DELETED_FLAG_LEN]),
v, data_to_copy);
ensure(flash_area_write_quadword(area,
offset - NORCOW_DELETED_FLAG_LEN, d),
NULL);
offset += data_to_copy;
v += data_to_copy;
tmp_len -= data_to_copy;
while (tmp_len > 0) {
data_to_copy =
((size_t)tmp_len > NORCOW_WORD_SIZE ? NORCOW_WORD_SIZE
: (size_t)tmp_len);
memset(d, 0, sizeof(d));
memcpy(d, v, data_to_copy);
ensure(flash_area_write_quadword(area, offset, d), NULL);
offset += data_to_copy;
v += data_to_copy;
tmp_len -= data_to_copy;
}
ensure(flash_lock_write(), NULL);
}
}
}
#endif
}
// If the update was not possible then write the entry as a new item.
@ -449,25 +719,46 @@ secbool norcow_set_ex(uint16_t key, const void *val, uint16_t len,
if (sectrue == *found) {
ensure(flash_unlock_write(), NULL);
uint32_t end = offset + len_old;
#ifdef FLASH_BYTE_ACCESS
// Update the prefix to indicate that the old item has been deleted.
uint32_t prefix = (uint32_t)len_old << 16;
ensure(flash_area_write_word(&STORAGE_AREAS[norcow_write_sector],
offset - NORCOW_PREFIX_LEN, prefix),
NULL);
ensure(
flash_area_write_word(area, offset - NORCOW_MAX_PREFIX_LEN, prefix),
NULL);
#else
if (len_old <= NORCOW_SMALL_ITEM_SIZE) {
// Delete entire content of the quadword.
uint32_t d[4] = {0};
ensure(flash_area_write_quadword(area, offset - 4, d), NULL);
// Move to the next quadword.
offset += NORCOW_WORD_SIZE - NORCOW_LEN_LEN - NORCOW_KEY_LEN;
} else {
// Update the flag to indicate that the old item has been deleted.
// deletes a portion of data too
uint32_t d[4] = {0};
ensure(flash_area_write_quadword(area, offset - NORCOW_DELETED_FLAG_LEN,
d),
NULL);
offset += NORCOW_DATA_OPT_SIZE;
}
#endif
// Delete the old item data.
uint32_t end = offset + len_old;
while (offset < end) {
ensure(flash_area_write_word(&STORAGE_AREAS[norcow_write_sector],
offset, 0x00000000),
NULL);
#ifdef FLASH_BYTE_ACCESS
ensure(flash_area_write_word(area, offset, 0x00000000), NULL);
#else
uint32_t d[4] = {0};
ensure(flash_area_write_quadword(area, offset, d), NULL);
#endif
offset += NORCOW_WORD_SIZE;
}
ensure(flash_lock_write(), NULL);
}
// Check whether there is enough free space and compact if full.
if (norcow_free_offset + NORCOW_PREFIX_LEN + len > NORCOW_SECTOR_SIZE) {
if (norcow_free_offset + NORCOW_MAX_PREFIX_LEN + len > NORCOW_SECTOR_SIZE) {
compact();
}
// Write new item.
@ -490,6 +781,7 @@ secbool norcow_delete(uint16_t key) {
return secfalse;
}
const flash_area_t *area = &STORAGE_AREAS[norcow_write_sector];
const void *ptr = NULL;
uint16_t len = 0;
if (sectrue != find_item(norcow_write_sector, key, &ptr, &len)) {
@ -502,18 +794,37 @@ secbool norcow_delete(uint16_t key) {
ensure(flash_unlock_write(), NULL);
uint32_t end = offset + len;
#ifdef FLASH_BYTE_ACCESS
// Update the prefix to indicate that the item has been deleted.
uint32_t prefix = (uint32_t)len << 16;
ensure(flash_area_write_word(&STORAGE_AREAS[norcow_write_sector],
offset - NORCOW_PREFIX_LEN, prefix),
ensure(flash_area_write_word(area, offset - NORCOW_MAX_PREFIX_LEN, prefix),
NULL);
#else
uint32_t d[4] = {0};
if (len <= NORCOW_SMALL_ITEM_SIZE) {
// Delete entire content of the quadword, setting the length to 0.
ensure(flash_area_write_quadword(
area, offset - NORCOW_LEN_LEN - NORCOW_KEY_LEN, d),
NULL);
// Move to the next quadword.
offset += NORCOW_WORD_SIZE - NORCOW_LEN_LEN - NORCOW_KEY_LEN;
} else {
// Update the flag to indicate that the old item has been deleted.
// deletes a portion of data too
ensure(flash_area_write_quadword(area, offset - NORCOW_DELETED_FLAG_LEN, d),
NULL);
offset += NORCOW_DATA_OPT_SIZE;
}
#endif
// Delete the item data.
uint32_t end = offset + len;
while (offset < end) {
ensure(flash_area_write_word(&STORAGE_AREAS[norcow_write_sector], offset,
0x00000000),
NULL);
#ifdef FLASH_BYTE_ACCESS
ensure(flash_area_write_word(area, offset, 0x00000000), NULL);
#else
ensure(flash_area_write_quadword(area, offset, d), NULL);
#endif
offset += NORCOW_WORD_SIZE;
}
@ -522,6 +833,7 @@ secbool norcow_delete(uint16_t key) {
return sectrue;
}
#ifdef FLASH_BYTE_ACCESS
/*
* Update a word in flash at the given pointer. The pointer must point
* into the NORCOW area.
@ -546,30 +858,141 @@ secbool norcow_update_word(uint16_t key, uint16_t offset, uint32_t value) {
ensure(flash_lock_write(), NULL);
return sectrue;
}
#endif
secbool norcow_set_counter(uint16_t key, uint32_t count) {
// The count is stored as a 32-bit integer followed by a tail of "1" bits,
// which is used as a tally.
uint32_t value[1 + COUNTER_TAIL_WORDS] = {0};
value[0] = count;
memset(&value[1], 0xff, sizeof(value) - sizeof(value[0]));
return norcow_set(key, value, sizeof(value));
}
secbool norcow_next_counter(uint16_t key, uint32_t *count) {
uint16_t len = 0;
const uint32_t *val_stored = NULL;
if (sectrue != norcow_get(key, (const void **)&val_stored, &len)) {
*count = 0;
return norcow_set_counter(key, 0);
}
if (len < sizeof(uint32_t) || len % sizeof(uint32_t) != 0) {
return secfalse;
}
uint16_t len_words = len / sizeof(uint32_t);
uint16_t i = 1;
while (i < len_words && val_stored[i] == 0) {
++i;
}
*count = val_stored[0] + 1 + 32 * (i - 1);
if (*count < val_stored[0]) {
// Value overflow.
return secfalse;
}
if (i < len_words) {
*count += hamming_weight(~val_stored[i]);
if (*count < val_stored[0]) {
// Value overflow.
return secfalse;
}
#ifdef FLASH_BYTE_ACCESS
return norcow_update_word(key, sizeof(uint32_t) * i, val_stored[i] >> 1);
#else
return norcow_set_counter(key, *count);
#endif
} else {
return norcow_set_counter(key, *count);
}
}
/*
* Update the value of the given key starting at the given offset.
*/
secbool norcow_update_bytes(const uint16_t key, const uint16_t offset,
const uint8_t *data, const uint16_t len) {
secbool norcow_update_bytes(const uint16_t key, const uint8_t *data,
const uint16_t len) {
const void *ptr = NULL;
uint16_t allocated_len = 0;
if (sectrue != find_item(norcow_write_sector, key, &ptr, &allocated_len)) {
return secfalse;
}
if (offset + len > allocated_len) {
#ifndef FLASH_BYTE_ACCESS
if (allocated_len <= NORCOW_SMALL_ITEM_SIZE) {
// small items are not updated in place
return secfalse;
}
#endif
if (norcow_write_buffer_flashed + len > allocated_len) {
return secfalse;
}
uint32_t sector_offset =
(const uint8_t *)ptr -
(const uint8_t *)norcow_ptr(norcow_write_sector, 0, NORCOW_SECTOR_SIZE) +
offset;
(const uint8_t *)norcow_ptr(norcow_write_sector, 0, NORCOW_SECTOR_SIZE);
const flash_area_t *area = &STORAGE_AREAS[norcow_write_sector];
ensure(flash_unlock_write(), NULL);
#ifdef FLASH_BYTE_ACCESS
sector_offset += norcow_write_buffer_flashed;
for (uint16_t i = 0; i < len; i++, sector_offset++) {
ensure(flash_area_write_byte(&STORAGE_AREAS[norcow_write_sector],
sector_offset, data[i]),
NULL);
ensure(flash_area_write_byte(area, sector_offset, data[i]), NULL);
}
norcow_write_buffer_flashed += len;
if (norcow_write_buffer_flashed >= allocated_len) {
norcow_write_buffer_flashed = 0;
}
#else
if (norcow_write_buffer_key != key && norcow_write_buffer_key != -1) {
// some other update bytes is in process, abort
return secfalse;
}
if (norcow_write_buffer_key == -1) {
memset(norcow_write_buffer, 0, sizeof(norcow_write_buffer));
norcow_write_buffer_key = key;
norcow_write_buffer[0] = 0xFE;
norcow_write_buffer_filled = NORCOW_DELETED_FLAG_LEN;
norcow_write_buffer_filled_data = 0;
norcow_write_buffer_flashed = 0;
}
uint16_t tmp_len = len;
uint16_t flash_offset =
sector_offset - NORCOW_DELETED_FLAG_LEN + norcow_write_buffer_flashed;
while (tmp_len > 0) {
uint16_t buffer_space = NORCOW_WORD_SIZE - norcow_write_buffer_filled;
uint16_t data_to_copy = (tmp_len > buffer_space ? buffer_space : tmp_len);
memcpy(&((uint8_t *)norcow_write_buffer)[norcow_write_buffer_filled], data,
data_to_copy);
data += data_to_copy;
norcow_write_buffer_filled += data_to_copy;
norcow_write_buffer_filled_data += data_to_copy;
tmp_len -= data_to_copy;
if (norcow_write_buffer_filled == NORCOW_WORD_SIZE ||
(norcow_write_buffer_filled_data + norcow_write_buffer_flashed) ==
allocated_len + NORCOW_DELETED_FLAG_LEN) {
ensure(flash_area_write_quadword(area, flash_offset, norcow_write_buffer),
NULL);
ensure(flash_area_write_quadword(area, flash_offset, norcow_write_buffer),
NULL);
flash_offset += NORCOW_WORD_SIZE;
norcow_write_buffer_filled = 0;
norcow_write_buffer_flashed += NORCOW_WORD_SIZE;
memset(norcow_write_buffer, 0, sizeof(norcow_write_buffer));
if ((norcow_write_buffer_flashed) >=
allocated_len + NORCOW_DELETED_FLAG_LEN) {
norcow_write_buffer_key = -1;
norcow_write_buffer_flashed = 0;
}
norcow_write_buffer_filled_data = 0;
}
}
#endif
ensure(flash_lock_write(), NULL);
return sectrue;
}

@ -65,18 +65,17 @@ secbool norcow_set_ex(uint16_t key, const void *val, uint16_t len,
*/
secbool norcow_delete(uint16_t key);
/*
* Update a word in flash in the given key at the given offset.
* Note that you can only change bits from 1 to 0.
*/
secbool norcow_update_word(uint16_t key, uint16_t offset, uint32_t value);
secbool norcow_set_counter(uint16_t key, uint32_t count);
secbool norcow_next_counter(uint16_t key, uint32_t *count);
/*
* Update the value of the given key starting at the given offset.
* Update the value of the given key, data are written sequentially from start
* Data are guaranteed to be stored on flash once the total item len is reached.
* Note that you can only change bits from 1 to 0.
*/
secbool norcow_update_bytes(const uint16_t key, const uint16_t offset,
const uint8_t *data, const uint16_t len);
secbool norcow_update_bytes(const uint16_t key, const uint8_t *data,
const uint16_t len);
/*
* Complete storage version upgrade

@ -30,6 +30,7 @@
#include "random_delays.h"
#include "sha2.h"
#include "storage.h"
#include "storage_utils.h"
#if USE_OPTIGA
#include "optiga.h"
@ -147,9 +148,6 @@ const uint8_t WIPE_CODE_EMPTY[] = {0, 0, 0, 0};
// The uint32 representation of an empty wipe code used in storage version 2.
#define V2_WIPE_CODE_EMPTY 0
// The length of the counter tail in words.
#define COUNTER_TAIL_WORDS 2
// Values used in the guard key integrity check.
#define GUARD_KEY_MODULUS 6311
#define GUARD_KEY_REMAINDER 15
@ -211,8 +209,8 @@ static secbool secequal(const void *ptr1, const void *ptr2, size_t n) {
static secbool secequal32(const void *ptr1, const void *ptr2, size_t n) {
assert(n % sizeof(uint32_t) == 0);
assert((uintptr_t)ptr1 % sizeof(uint32_t) == 0);
assert((uintptr_t)ptr2 % sizeof(uint32_t) == 0);
// assert((uintptr_t)ptr1 % sizeof(uint32_t) == 0);
// assert((uintptr_t)ptr2 % sizeof(uint32_t) == 0);
size_t wn = n / sizeof(uint32_t);
const uint32_t *p1 = (const uint32_t *)ptr1;
@ -395,34 +393,21 @@ static secbool set_wipe_code(const uint8_t *wipe_code, size_t wipe_code_len) {
// The format of the WIPE_CODE_DATA_KEY entry is:
// wipe code (variable), random salt (16 bytes), authentication tag (16 bytes)
// NOTE: We allocate extra space for the HMAC result.
uint8_t salt_and_tag[WIPE_CODE_SALT_SIZE + SHA256_DIGEST_LENGTH] = {0};
uint8_t *salt = salt_and_tag;
uint8_t *tag = salt_and_tag + WIPE_CODE_SALT_SIZE;
uint8_t data[(MAX_WIPE_CODE_LEN + WIPE_CODE_SALT_SIZE +
SHA256_DIGEST_LENGTH)] = {0};
uint8_t *salt = data + wipe_code_len;
uint8_t *tag = salt + WIPE_CODE_SALT_SIZE;
memcpy(data, wipe_code, wipe_code_len);
random_buffer(salt, WIPE_CODE_SALT_SIZE);
hmac_sha256(salt, WIPE_CODE_SALT_SIZE, wipe_code, wipe_code_len, tag);
// Preallocate the entry in the flash storage.
if (sectrue !=
norcow_set(WIPE_CODE_DATA_KEY, NULL,
wipe_code_len + WIPE_CODE_SALT_SIZE + WIPE_CODE_TAG_SIZE)) {
return secfalse;
}
// Write wipe code into the preallocated entry.
if (sectrue !=
norcow_update_bytes(WIPE_CODE_DATA_KEY, 0, wipe_code, wipe_code_len)) {
return secfalse;
}
// Write salt and tag into the preallocated entry.
if (sectrue !=
norcow_update_bytes(WIPE_CODE_DATA_KEY, wipe_code_len, salt_and_tag,
WIPE_CODE_SALT_SIZE + WIPE_CODE_TAG_SIZE)) {
return secfalse;
}
secbool ret =
norcow_set(WIPE_CODE_DATA_KEY, data,
wipe_code_len + WIPE_CODE_SALT_SIZE + WIPE_CODE_TAG_SIZE);
return sectrue;
memzero(data, sizeof(data));
return ret;
}
static secbool is_not_wipe_code(const uint8_t *pin, size_t pin_len) {
@ -844,6 +829,10 @@ static secbool pin_fails_reset(void) {
return secfalse;
}
uint32_t new_logs[GUARD_KEY_WORDS + 2 * PIN_LOG_WORDS];
secbool edited = secfalse;
memcpy(new_logs, logs, len);
uint32_t guard_mask = 0;
uint32_t guard = 0;
wait_random();
@ -857,16 +846,23 @@ static secbool pin_fails_reset(void) {
const uint32_t *entry_log = success_log + PIN_LOG_WORDS;
for (size_t i = 0; i < PIN_LOG_WORDS; ++i) {
if (entry_log[i] == unused) {
if (edited == sectrue) {
return norcow_set(PIN_LOGS_KEY, new_logs, sizeof(new_logs));
}
return sectrue;
}
if (success_log[i] != guard) {
if (sectrue != norcow_update_word(
PIN_LOGS_KEY, sizeof(uint32_t) * (i + GUARD_KEY_WORDS),
entry_log[i])) {
return secfalse;
if (new_logs[(i + GUARD_KEY_WORDS)] != entry_log[i]) {
edited = sectrue;
new_logs[(i + GUARD_KEY_WORDS)] = entry_log[i];
}
}
}
if (edited == sectrue) {
if (sectrue != norcow_set(PIN_LOGS_KEY, new_logs, sizeof(new_logs))) {
return secfalse;
}
}
return pin_logs_init(0);
}
@ -885,6 +881,9 @@ secbool storage_pin_fails_increase(void) {
return secfalse;
}
uint32_t new_logs[GUARD_KEY_WORDS + 2 * PIN_LOG_WORDS];
memcpy(new_logs, logs, len);
uint32_t guard_mask = 0;
uint32_t guard = 0;
wait_random();
@ -909,11 +908,10 @@ secbool storage_pin_fails_increase(void) {
word = (word >> 2) | (word >> 1);
wait_random();
if (sectrue !=
norcow_update_word(
PIN_LOGS_KEY,
sizeof(uint32_t) * (i + GUARD_KEY_WORDS + PIN_LOG_WORDS),
(word & ~guard_mask) | guard)) {
new_logs[(i + GUARD_KEY_WORDS + PIN_LOG_WORDS)] =
(word & ~guard_mask) | guard;
if (sectrue != norcow_set(PIN_LOGS_KEY, new_logs, sizeof(new_logs))) {
handle_fault("PIN logs update");
return secfalse;
}
@ -924,15 +922,6 @@ secbool storage_pin_fails_increase(void) {
return secfalse;
}
static uint32_t hamming_weight(uint32_t value) {
value = value - ((value >> 1) & 0x55555555);
value = (value & 0x33333333) + ((value >> 2) & 0x33333333);
value = (value + (value >> 4)) & 0x0F0F0F0F;
value = value + (value >> 8);
value = value + (value >> 16);
return value & 0x3F;
}
static secbool pin_get_fails(uint32_t *ctr) {
*ctr = PIN_MAX_TRIES;
@ -1298,9 +1287,9 @@ static secbool storage_get_encrypted(const uint16_t key, void *val_dest,
}
const uint8_t *iv = (const uint8_t *)val_stored;
const uint8_t *tag_stored = (const uint8_t *)val_stored + CHACHA20_IV_SIZE;
const uint8_t *ciphertext =
(const uint8_t *)val_stored + CHACHA20_IV_SIZE + POLY1305_TAG_SIZE;
const uint8_t *tag_stored =
(const uint8_t *)val_stored + CHACHA20_IV_SIZE + *len;
const uint8_t *ciphertext = (const uint8_t *)val_stored + CHACHA20_IV_SIZE;
uint8_t tag_computed[POLY1305_TAG_SIZE] = {0};
chacha20poly1305_ctx ctx = {0};
rfc7539_init(&ctx, cached_dek, iv);
@ -1380,23 +1369,19 @@ static secbool storage_set_encrypted(const uint16_t key, const void *val,
// Write the IV to the flash.
uint8_t buffer[CHACHA20_BLOCK_SIZE] = {0};
random_buffer(buffer, CHACHA20_IV_SIZE);
uint16_t offset = 0;
if (sectrue != norcow_update_bytes(key, offset, buffer, CHACHA20_IV_SIZE)) {
if (sectrue != norcow_update_bytes(key, buffer, CHACHA20_IV_SIZE)) {
return secfalse;
}
offset += CHACHA20_IV_SIZE + POLY1305_TAG_SIZE;
// Encrypt all blocks except for the last one.
chacha20poly1305_ctx ctx = {0};
rfc7539_init(&ctx, cached_dek, buffer);
rfc7539_auth(&ctx, (const uint8_t *)&key, sizeof(key));
size_t i = 0;
for (i = 0; i + CHACHA20_BLOCK_SIZE < len;
i += CHACHA20_BLOCK_SIZE, offset += CHACHA20_BLOCK_SIZE) {
for (i = 0; i + CHACHA20_BLOCK_SIZE < len; i += CHACHA20_BLOCK_SIZE) {
chacha20poly1305_encrypt(&ctx, ((const uint8_t *)val) + i, buffer,
CHACHA20_BLOCK_SIZE);
if (sectrue !=
norcow_update_bytes(key, offset, buffer, CHACHA20_BLOCK_SIZE)) {
if (sectrue != norcow_update_bytes(key, buffer, CHACHA20_BLOCK_SIZE)) {
memzero(&ctx, sizeof(ctx));
memzero(buffer, sizeof(buffer));
return secfalse;
@ -1405,10 +1390,10 @@ static secbool storage_set_encrypted(const uint16_t key, const void *val,
// Encrypt final block and compute message authentication tag.
chacha20poly1305_encrypt(&ctx, ((const uint8_t *)val) + i, buffer, len - i);
secbool ret = norcow_update_bytes(key, offset, buffer, len - i);
secbool ret = norcow_update_bytes(key, buffer, len - i);
if (sectrue == ret) {
rfc7539_finish(&ctx, sizeof(key), len, buffer);
ret = norcow_update_bytes(key, CHACHA20_IV_SIZE, buffer, POLY1305_TAG_SIZE);
ret = norcow_update_bytes(key, buffer, POLY1305_TAG_SIZE);
}
memzero(&ctx, sizeof(ctx));
memzero(buffer, sizeof(buffer));
@ -1461,19 +1446,8 @@ secbool storage_set_counter(const uint16_t key, const uint32_t count) {
return secfalse;
}
// The count is stored as a 32-bit integer followed by a tail of "1" bits,
// which is used as a tally.
uint32_t value[1 + COUNTER_TAIL_WORDS] = {0};
memset(value, 0xff, sizeof(value));
value[0] = count;
return storage_set(key, value, sizeof(value));
}
secbool storage_next_counter(const uint16_t key, uint32_t *count) {
const uint8_t app = key >> 8;
// APP == 0 is reserved for PIN related values
if (sectrue != initialized || app == APP_STORAGE ||
(app & FLAG_PUBLIC) == 0) {
if (sectrue != initialized || app == APP_STORAGE) {
return secfalse;
}
@ -1481,39 +1455,27 @@ secbool storage_next_counter(const uint16_t key, uint32_t *count) {
return secfalse;
}
uint16_t len = 0;
const uint32_t *val_stored = NULL;
if (sectrue != norcow_get(key, (const void **)&val_stored, &len)) {
*count = 0;
return storage_set_counter(key, 0);
}
return norcow_set_counter(key, count);
}
secbool storage_next_counter(const uint16_t key, uint32_t *count) {
const uint8_t app = key >> 8;
if (len < sizeof(uint32_t) || len % sizeof(uint32_t) != 0) {
if ((app & FLAG_PUBLIC) == 0) {
return secfalse;
}
uint16_t len_words = len / sizeof(uint32_t);
uint16_t i = 1;
while (i < len_words && val_stored[i] == 0) {
++i;
// APP == 0 is reserved for PIN related values
if (sectrue != initialized || app == APP_STORAGE ||
(app & FLAG_PUBLIC) == 0) {
return secfalse;
}
*count = val_stored[0] + 1 + 32 * (i - 1);
if (*count < val_stored[0]) {
// Value overflow.
if (sectrue != unlocked && (app & FLAGS_WRITE) != FLAGS_WRITE) {
return secfalse;
}
if (i < len_words) {
*count += hamming_weight(~val_stored[i]);
if (*count < val_stored[0]) {
// Value overflow.
return secfalse;
}
return norcow_update_word(key, sizeof(uint32_t) * i, val_stored[i] >> 1);
} else {
return storage_set_counter(key, *count);
}
return norcow_next_counter(key, count);
}
secbool storage_has_pin(void) {
@ -1756,6 +1718,7 @@ static secbool storage_upgrade(void) {
// Storage version 1: encrypted norcow
// Storage version 2: adds 9 digit wipe code
// Storage version 3: adds variable length PIN and wipe code
// Storage version 4: changes data structure of encrypted data
const uint16_t V0_PIN_KEY = 0x0000;
const uint16_t V0_PIN_FAIL_KEY = 0x0001;
@ -1818,8 +1781,42 @@ static secbool storage_upgrade(void) {
// Copy all entries.
uint32_t offset = 0;
while (sectrue == norcow_get_next(&offset, &key, &val, &len)) {
if (sectrue != norcow_set(key, val, len)) {
return secfalse;
if (norcow_active_version < 4) {
// Change data structure for encrypted entries.
const uint8_t app = key >> 8;
if (((app & FLAG_PUBLIC) == 0) &&
!(key == PIN_LOGS_KEY || key == EDEK_PVC_KEY ||
key == PIN_NOT_SET_KEY || key == STORAGE_TAG_KEY ||
key == WIPE_CODE_DATA_KEY || key == STORAGE_UPGRADED_KEY ||
key == UNAUTH_VERSION_KEY)) {
if (sectrue != norcow_set(key, NULL, len)) {
return secfalse;
}
if (sectrue !=
norcow_update_bytes(key, ((uint8_t *)val), CHACHA20_IV_SIZE)) {
return secfalse;
}
if (sectrue !=
norcow_update_bytes(
key, ((uint8_t *)val) + CHACHA20_IV_SIZE + POLY1305_TAG_SIZE,
len - CHACHA20_IV_SIZE - POLY1305_TAG_SIZE)) {
return secfalse;
}
if (sectrue !=
norcow_update_bytes(key, ((uint8_t *)val) + CHACHA20_IV_SIZE,
POLY1305_TAG_SIZE)) {
return secfalse;
}
} else {
if (sectrue != norcow_set(key, val, len)) {
return secfalse;
}
}
} else {
if (sectrue != norcow_set(key, val, len)) {
return secfalse;
}
}
}
}
@ -1894,5 +1891,7 @@ static secbool storage_upgrade_unlocked(const uint8_t *pin, size_t pin_len,
memzero(wipe_code, wipe_code_len);
}
// nothing to do for upgrading to version 4
return ret;
}

@ -0,0 +1,11 @@
#include <stdint.h>
uint32_t hamming_weight(uint32_t value) {
value = value - ((value >> 1) & 0x55555555);
value = (value & 0x33333333) + ((value >> 2) & 0x33333333);
value = (value + (value >> 4)) & 0x0F0F0F0F;
value = value + (value >> 8);
value = value + (value >> 16);
return value & 0x3F;
}

@ -0,0 +1,4 @@
#include <stdint.h>
uint32_t hamming_weight(uint32_t value);

@ -4,6 +4,7 @@ export ASAN_OPTIONS=verify_asan_link_order=0
build:
$(MAKE) -C c
$(MAKE) -C c libtrezor-storage-qw.so
$(MAKE) -C c0
clean:

@ -16,6 +16,7 @@ SRC += storage/tests/c/random_delays.c
SRC += storage/tests/c/test_layout.c
SRC += storage/flash_common.c
SRC += storage/storage.c
SRC += storage/storage_utils.c
SRC += storage/norcow.c
SRC += crypto/pbkdf2.c
SRC += crypto/rand.c
@ -28,19 +29,32 @@ SRC += crypto/sha2.c
SRC += crypto/memzero.c
OBJ = $(SRC:%.c=build/%.o)
OBJ_QW = $(SRC:%.c=build_qw/%.o)
OUT = libtrezor-storage.so
OUT_QW = libtrezor-storage-qw.so
$(OUT): $(OBJ)
$(CC) $(CFLAGS) $(LIBS) $(OBJ) -shared -o $(OUT)
$(OUT_QW): $(OBJ_QW)
$(CC) $(CFLAGS) -DFLASH_QUADWORD $(LIBS) $(OBJ_QW) -shared -o $(OUT_QW)
build/crypto/chacha20poly1305/chacha_merged.o: $(BASE)crypto/chacha20poly1305/chacha_merged.c
mkdir -p $(@D)
$(CC) $(CFLAGS) $(INC) -c $< -o $@
build_qw/crypto/chacha20poly1305/chacha_merged.o: $(BASE)crypto/chacha20poly1305/chacha_merged.c
mkdir -p $(@D)
$(CC) $(CFLAGS) $(INC) -c $< -o $@
build/%.o: $(BASE)%.c $(BASE)%.h
mkdir -p $(@D)
$(CC) $(CFLAGS) $(INC) -c $< -o $@
build_qw/%.o: $(BASE)%.c $(BASE)%.h
mkdir -p $(@D)
$(CC) $(CFLAGS) -DFLASH_QUADWORD $(INC) -c $< -o $@
clean:
rm -f $(OUT) $(OBJ)

@ -24,7 +24,9 @@
#include <stdlib.h>
#include "secbool.h"
#ifndef FLASH_QUADWORD
#define FLASH_BYTE_ACCESS 1
#endif
#include "flash_common.h"
#include "test_layout.h"

@ -40,6 +40,6 @@
/*
* Current storage version.
*/
#define NORCOW_VERSION ((uint32_t)0x00000003)
#define NORCOW_VERSION ((uint32_t)0x00000004)
#endif

@ -4,11 +4,12 @@ import os
EXTERNAL_SALT_LEN = 32
sectrue = -1431655766 # 0xAAAAAAAAA
fname = os.path.join(os.path.dirname(__file__), "libtrezor-storage.so")
fname_qw = os.path.join(os.path.dirname(__file__), "libtrezor-storage-qw.so")
class Storage:
def __init__(self) -> None:
self.lib = c.cdll.LoadLibrary(fname)
def __init__(self, flash_byte_access=True) -> None:
self.lib = c.cdll.LoadLibrary(fname if flash_byte_access else fname_qw)
self.flash_size = c.cast(self.lib.FLASH_SIZE, c.POINTER(c.c_uint32))[0]
self.flash_buffer = c.create_string_buffer(self.flash_size)
c.cast(self.lib.FLASH_BUFFER, c.POINTER(c.c_void_p))[0] = c.addressof(

@ -120,6 +120,7 @@ POLY1305_MAC_SIZE = 16
# The length of the ChaCha20 IV (aka nonce) in bytes as per RFC 7539.
CHACHA_IV_SIZE = 12
CHACHA_IV_PADDING = 4
# ----- Norcow ----- #

@ -4,31 +4,43 @@ from struct import pack
from . import consts
def align4_int(i: int):
return (4 - i) % 4
def align_int(i: int, align: int):
return (align - i) % align
def align4_data(data):
return data + b"\x00" * align4_int(len(data))
def align_data(data, align: int):
return data + b"\x00" * align_int(len(data), align)
class Norcow:
def __init__(self):
def __init__(self, flash_byte_access=True):
self.sectors = None
self.active_sector = 0
self.flash_byte_access = flash_byte_access
if flash_byte_access:
self.word_size = consts.WORD_SIZE
self.magic = consts.NORCOW_MAGIC_AND_VERSION
self.item_prefix_len = 4
else:
self.word_size = 4 * consts.WORD_SIZE
self.magic = consts.NORCOW_MAGIC_AND_VERSION + bytes([0xFF] * 8)
self.item_prefix_len = 4 * consts.WORD_SIZE + 1
def init(self):
if self.sectors:
for sector in range(consts.NORCOW_SECTOR_COUNT):
if self.sectors[sector][:8] == consts.NORCOW_MAGIC_AND_VERSION:
if self.sectors[sector][: len(self.magic)] == self.magic:
self.active_sector = sector
self.active_offset = self.find_free_offset()
break
else:
self.wipe()
def is_byte_access(self):
return self.flash_byte_access
def find_free_offset(self):
offset = len(consts.NORCOW_MAGIC_AND_VERSION)
offset = len(self.magic)
while True:
try:
k, v = self._read_item(offset)
@ -45,9 +57,9 @@ class Norcow:
bytearray([0xFF] * consts.NORCOW_SECTOR_SIZE)
for _ in range(consts.NORCOW_SECTOR_COUNT)
]
self.sectors[sector][:8] = consts.NORCOW_MAGIC_AND_VERSION
self.sectors[sector][: len(self.magic)] = self.magic
self.active_sector = sector
self.active_offset = len(consts.NORCOW_MAGIC_AND_VERSION)
self.active_offset = len(self.magic)
def get(self, key: int) -> bytes:
value, _ = self._find_item(key)
@ -65,7 +77,10 @@ class Norcow:
else:
self._delete_old(pos, found_value)
if self.active_offset + 4 + len(val) > consts.NORCOW_SECTOR_SIZE:
if (
self.active_offset + self.item_prefix_len + len(val)
> consts.NORCOW_SECTOR_SIZE
):
self._compact()
self._append(key, val)
@ -95,15 +110,20 @@ class Norcow:
Item is updatable if the new value is the same or
it changes 1 to 0 only (the flash memory does not
allow to flip 0 to 1 unless you wipe it).
For flash with no byte access, item is updatable if the new value is the same
"""
if len(old) != len(new):
return False
if old == new:
return True
for a, b in zip(old, new):
if a & b != b:
return False
return True
if self.flash_byte_access:
for a, b in zip(old, new):
if a & b != b:
return False
return True
else:
return False
def _delete_old(self, pos: int, value: bytes):
wiped_data = b"\x00" * len(value)
@ -113,14 +133,48 @@ class Norcow:
self.active_offset += self._write(self.active_offset, key, value)
def _write(self, pos: int, key: int, new_value: bytes) -> int:
data = pack("<HH", key, len(new_value)) + align4_data(new_value)
if pos + len(data) > consts.NORCOW_SECTOR_SIZE:
raise RuntimeError("Norcow: item too big")
self.sectors[self.active_sector][pos : pos + len(data)] = data
return len(data)
if self.flash_byte_access:
data = pack("<HH", key, len(new_value)) + align_data(
new_value, self.word_size
)
if pos + len(data) > consts.NORCOW_SECTOR_SIZE:
raise RuntimeError("Norcow: item too big")
self.sectors[self.active_sector][pos : pos + len(data)] = data
return len(data)
else:
if len(new_value) <= 12:
if key == 0:
self.sectors[self.active_sector][pos : pos + self.word_size] = [
0
] * self.word_size
else:
if len(new_value) == 0:
data = pack("<HH", len(new_value), key) + bytes([0] * 12)
else:
data = pack("<HH", len(new_value), key) + align_data(
new_value, 12
)
if pos + len(data) > consts.NORCOW_SECTOR_SIZE:
raise RuntimeError("Norcow: item too big")
self.sectors[self.active_sector][pos : pos + self.word_size] = data
return len(data)
else:
if key == 0:
old_key = self.sectors[self.active_sector][pos + 2 : pos + 4]
old_key = int.from_bytes(old_key, sys.byteorder)
data = pack("<HH", len(new_value), old_key) + bytes([255] * 12)
data += align_data(bytes([0]) + new_value, self.word_size)
else:
data = pack("<HH", len(new_value), key) + bytes([255] * 12)
data += align_data(bytes([0xFE]) + new_value, self.word_size)
if pos + len(data) > consts.NORCOW_SECTOR_SIZE:
raise RuntimeError("Norcow: item too big")
self.sectors[self.active_sector][pos : pos + len(data)] = data
return len(data)
def _find_item(self, key: int) -> (bytes, int):
offset = len(consts.NORCOW_MAGIC_AND_VERSION)
offset = len(self.magic)
value = False
pos = offset
while True:
@ -135,7 +189,7 @@ class Norcow:
return value, pos
def _get_all_keys(self) -> (bytes, int):
offset = len(consts.NORCOW_MAGIC_AND_VERSION)
offset = len(self.magic)
keys = set()
while True:
try:
@ -147,21 +201,65 @@ class Norcow:
return keys
def _norcow_item_length(self, data: bytes) -> int:
# APP_ID, KEY_ID, LENGTH, DATA, ALIGNMENT
return 1 + 1 + 2 + len(data) + align4_int(len(data))
if self.flash_byte_access:
# APP_ID, KEY_ID, LENGTH, DATA, ALIGNMENT
return (
self.item_prefix_len + len(data) + align_int(len(data), self.word_size)
)
else:
if len(data) <= 12 and not self.flash_byte_access:
return self.word_size
else:
# APP_ID, KEY_ID, LENGTH, DATA, ALIGNMENT
return (
self.word_size
+ 1
+ len(data)
+ align_int(1 + len(data), self.word_size)
)
def _read_item(self, offset: int) -> (int, bytes):
key = self.sectors[self.active_sector][offset : offset + 2]
key = int.from_bytes(key, sys.byteorder)
if key == consts.NORCOW_KEY_FREE:
if offset >= consts.NORCOW_SECTOR_SIZE:
raise ValueError("Norcow: no data on this offset")
length = self.sectors[self.active_sector][offset + 2 : offset + 4]
length = int.from_bytes(length, sys.byteorder)
value = self.sectors[self.active_sector][offset + 4 : offset + 4 + length]
if self.flash_byte_access:
key = self.sectors[self.active_sector][offset : offset + 2]
key = int.from_bytes(key, sys.byteorder)
if key == consts.NORCOW_KEY_FREE:
raise ValueError("Norcow: no data on this offset")
length = self.sectors[self.active_sector][offset + 2 : offset + 4]
length = int.from_bytes(length, sys.byteorder)
value = self.sectors[self.active_sector][offset + 4 : offset + 4 + length]
else:
length = self.sectors[self.active_sector][offset : offset + 2]
length = int.from_bytes(length, sys.byteorder)
if length <= 12:
key = self.sectors[self.active_sector][offset + 2 : offset + 4]
key = int.from_bytes(key, sys.byteorder)
if key == consts.NORCOW_KEY_FREE:
raise ValueError("Norcow: no data on this offset")
value = self.sectors[self.active_sector][
offset + 4 : offset + 4 + length
]
else:
key = self.sectors[self.active_sector][offset + 2 : offset + 4]
key = int.from_bytes(key, sys.byteorder)
deleted = self.sectors[self.active_sector][offset + self.word_size]
value = self.sectors[self.active_sector][
offset + self.word_size + 1 : offset + self.word_size + 1 + length
]
if deleted == 0:
key = 0
else:
if key == consts.NORCOW_KEY_FREE:
raise ValueError("Norcow: no data on this offset")
return key, value
def _compact(self):
offset = len(consts.NORCOW_MAGIC_AND_VERSION)
offset = len(self.magic)
data = list()
while True:
try:

@ -117,7 +117,10 @@ class PinLog:
+ helpers.to_bytes_by_words(pin_success_log, consts.PIN_LOG_SIZE)
+ helpers.to_bytes_by_words(pin_entry_log, consts.PIN_LOG_SIZE)
)
try:
self.norcow.replace(consts.PIN_LOG_KEY, pin_log)
except RuntimeError:
if self.norcow.is_byte_access():
try:
self.norcow.replace(consts.PIN_LOG_KEY, pin_log)
except RuntimeError:
self.norcow.set(consts.PIN_LOG_KEY, pin_log)
else:
self.norcow.set(consts.PIN_LOG_KEY, pin_log)

@ -7,12 +7,12 @@ from .pin_log import PinLog
class Storage:
def __init__(self):
def __init__(self, flash_byte_access: bool = True):
self.initialized = False
self.unlocked = False
self.dek = None
self.sak = None
self.nc = Norcow()
self.nc = Norcow(flash_byte_access=flash_byte_access)
self.pin_log = PinLog(self.nc)
def init(self, hardware_salt: bytes = b""):
@ -153,9 +153,10 @@ class Storage:
if val > consts.UINT32_MAX:
raise RuntimeError("Failed to set value in storage.")
counter = val.to_bytes(4, sys.byteorder) + bytearray(
b"\xFF" * consts.COUNTER_TAIL_SIZE
)
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:
@ -168,21 +169,30 @@ class Storage:
return 0
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
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(
key,
current[:4]
+ helpers.to_bytes_by_words(tail >> 1, consts.COUNTER_TAIL_SIZE),
)
self.set_counter(key, increased_count)
return increased_count
def delete(self, key: int) -> bool:
@ -214,10 +224,9 @@ class Storage:
data = self.nc.get(key)
iv = data[: consts.CHACHA_IV_SIZE]
# cipher text with MAC
tag = data[
consts.CHACHA_IV_SIZE : consts.CHACHA_IV_SIZE + consts.POLY1305_MAC_SIZE
]
ciphertext = data[consts.CHACHA_IV_SIZE + consts.POLY1305_MAC_SIZE :]
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)
)
@ -237,7 +246,7 @@ class Storage:
cipher_text, tag = crypto.chacha_poly_encrypt(
self.dek, iv, val, key.to_bytes(2, sys.byteorder)
)
return self.nc.replace(key, iv + tag + cipher_text)
return self.nc.replace(key, iv + cipher_text + tag)
def _calculate_authentication_tag(self) -> bytes:
keys = []

@ -0,0 +1,373 @@
import pytest
from ..src import consts, norcow
from . import common
def test_norcow_set_qw():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0001, b"123")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:18] == b"\x03\x00" # length
assert data[18:20] == b"\x01\x00" # app + key
assert data[20:23] == b"123" # data
assert data[23:32] == bytes([0] * 9) # alignment
assert common.all_ff_bytes(data[32:])
n.wipe()
n.set(0x0901, b"hello")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:18] == b"\x05\x00" # length\x00
assert data[18:20] == b"\x01\x09" # app + key
assert data[20:25] == b"hello" # data
assert data[25:32] == bytes([0] * 7) # alignment
assert common.all_ff_bytes(data[32:])
offset = 32
n.set(0x0102, b"world!")
data = n._dump()[0][:256]
assert data[offset : offset + 2] == b"\x06\x00" # length
assert data[offset + 2 : offset + 4] == b"\x02\x01" # app + key
assert data[offset + 4 : offset + 4 + 6] == b"world!" # data
assert data[offset + 4 + 6 : offset + 16] == bytes([0] * 6) # alignment
assert common.all_ff_bytes(data[offset + 16 :])
def test_norcow_update():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0001, b"1234567890A")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:18] == b"\x0B\x00" # length
assert data[18:20] == b"\x01\x00" # app + key
assert data[20:31] == b"1234567890A" # data
assert data[31:32] == bytes([0] * 1) # alignment
assert common.all_ff_bytes(data[32:])
n.set(0x0001, b"A0987654321")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:32] == bytes([0] * 16) # empty data
assert data[32:34] == b"\x0B\x00" # length
assert data[34:36] == b"\x01\x00" # app + key
assert data[36:47] == b"A0987654321" # data
assert data[47:48] == bytes([0] * 1) # alignment
assert common.all_ff_bytes(data[48:])
n.wipe()
n.set(0x0001, b"1234567890AB")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:18] == b"\x0C\x00" # length
assert data[18:20] == b"\x01\x00" # app + key
assert data[20:32] == b"1234567890AB" # data
assert common.all_ff_bytes(data[32:])
n.set(0x0001, b"BA0987654321")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:32] == bytes([0] * 16) # empty data
assert data[32:34] == b"\x0C\x00" # length
assert data[34:36] == b"\x01\x00" # app + key
assert data[36:48] == b"BA0987654321" # data
assert common.all_ff_bytes(data[48:])
n.wipe()
offset = 16
n.set(0x0102, b"world!_world!")
data = n._dump()[0][:256]
assert data[offset : offset + 2] == b"\x0D\x00" # length
assert data[offset + 16 : offset + 18] == b"\x02\x01" # app + key
assert data[offset + 32 : offset + 32 + 13] == b"world!_world!" # data
assert data[offset + 32 + 13 : offset + 48] == b"\x00\x00\x00" # alignment
assert common.all_ff_bytes(data[offset + 48 :])
n.set(0x0102, b"hello!_hello!")
data = n._dump()[0][:256]
assert data[offset : offset + 2] == b"\x0D\x00" # length
assert data[offset + 16 : offset + 48] == bytes([0] * 32)
offset += 48
assert data[offset + 0 : offset + 2] == b"\x0D\x00" # length
assert data[offset + 16 : offset + 18] == b"\x02\x01" # app + key
assert data[offset + 32 : offset + 32 + 13] == b"hello!_hello!" # data
assert data[offset + 32 + 13 : offset + 48] == b"\x00\x00\x00" # alignment
assert common.all_ff_bytes(data[offset + 48 :])
def test_norcow_set_qw_long():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0001, b"1234567890abc")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:18] == b"\x0D\x00" # length
assert data[32:34] == b"\x01\x00" # app + key
assert data[48:61] == b"1234567890abc" # data
assert common.all_ff_bytes(data[64:])
n.wipe()
n.set(0x0901, b"hello_hello__")
data = n._dump()[0][:256]
assert data[:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert data[16:18] == b"\x0D\x00" # length\x00
assert data[32:34] == b"\x01\x09" # app + key
assert data[48:61] == b"hello_hello__" # data
assert data[61:64] == b"\x00\x00\x00" # alignment
assert common.all_ff_bytes(data[64:])
offset = 64
n.set(0x0102, b"world!_world!")
data = n._dump()[0][:256]
assert data[offset : offset + 2] == b"\x0D\x00" # length
assert data[offset + 16 : offset + 18] == b"\x02\x01" # app + key
assert data[offset + 32 : offset + 32 + 13] == b"world!_world!" # data
assert data[offset + 32 + 13 : offset + 48] == b"\x00\x00\x00" # alignment
assert common.all_ff_bytes(data[offset + 48 :])
def test_norcow_read_item_qw():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0001, b"123")
n.set(0x0002, b"456")
n.set(0x0101, b"789")
key, value = n._read_item(32)
assert key == 0x0002
assert value == b"456"
key, value = n._read_item(48)
assert key == 0x0101
assert value == b"789"
with pytest.raises(ValueError) as e:
key, value = n._read_item(204)
assert "no data" in str(e.value)
def test_norcow_get_item_qw():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0001, b"123")
n.set(0x0002, b"456")
n.set(0x0101, b"789")
value = n.get(0x0001)
assert value == b"123"
assert (
n._dump()[0][:80].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex()
+ "ffffffffffffffff"
+ "03000100313233000000000000000000"
+ "03000200343536000000000000000000"
+ "03000101373839000000000000000000"
+ "ffffffffffffffffffffffffffffffff"
)
# replacing item with the same value (update)
n.set(0x0101, b"789")
value = n.get(0x0101)
assert value == b"789"
assert (
n._dump()[0][:80].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex()
+ "ffffffffffffffff"
+ "03000100313233000000000000000000"
+ "03000200343536000000000000000000"
+ "03000101373839000000000000000000"
+ "ffffffffffffffffffffffffffffffff"
)
# replacing item with value with less 1 bits than before (wipe and new entry)
n.set(0x0101, b"788")
value = n.get(0x0101)
assert value == b"788"
assert (
n._dump()[0][:96].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex()
+ "ffffffffffffffff"
+ "03000100313233000000000000000000"
+ "03000200343536000000000000000000"
+ "00000000000000000000000000000000"
+ "03000101373838000000000000000000"
+ "ffffffffffffffffffffffffffffffff"
)
# replacing item with value with more 1 bits than before (wipe and new entry)
n.set(0x0101, b"787")
value = n.get(0x0101)
assert value == b"787"
assert (
n._dump()[0][:112].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex()
+ "ffffffffffffffff"
+ "03000100313233000000000000000000"
+ "03000200343536000000000000000000"
+ "00000000000000000000000000000000"
+ "00000000000000000000000000000000"
+ "03000101373837000000000000000000"
+ "ffffffffffffffffffffffffffffffff"
)
n.set(0x0002, b"world")
n.set(0x0002, b"earth")
value = n.get(0x0002)
assert value == b"earth"
def test_norcow_get_item_qw_long():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0001, b"1231231231231")
n.set(0x0002, b"4564564564564")
n.set(0x0101, b"7897897897897")
value = n.get(0x0001)
assert value == b"1231231231231"
assert (
n._dump()[0][:170].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex() + "ffffffffffffffff"
"0d000000ffffffffffffffffffffffff"
"01000000ffffffffffffffffffffffff"
"31323331323331323331323331000000"
"0d000000ffffffffffffffffffffffff"
"02000000ffffffffffffffffffffffff"
"34353634353634353634353634000000"
"0d000000ffffffffffffffffffffffff"
"01010000ffffffffffffffffffffffff"
"37383937383937383937383937000000"
"ffffffffffffffffffff"
)
# replacing item with the same value (update)
n.set(0x0101, b"7897897897897")
value = n.get(0x0101)
assert value == b"7897897897897"
assert (
n._dump()[0][:170].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex() + "ffffffffffffffff"
"0d000000ffffffffffffffffffffffff"
"01000000ffffffffffffffffffffffff"
"31323331323331323331323331000000"
"0d000000ffffffffffffffffffffffff"
"02000000ffffffffffffffffffffffff"
"34353634353634353634353634000000"
"0d000000ffffffffffffffffffffffff"
"01010000ffffffffffffffffffffffff"
"37383937383937383937383937000000"
"ffffffffffffffffffff"
)
# replacing item with value with less 1 bits than before (update)
n.set(0x0101, b"7887887887887")
value = n.get(0x0101)
assert value == b"7887887887887"
assert (
n._dump()[0][:218].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex() + "ffffffffffffffff"
"0d000000ffffffffffffffffffffffff"
"01000000ffffffffffffffffffffffff"
"31323331323331323331323331000000"
"0d000000ffffffffffffffffffffffff"
"02000000ffffffffffffffffffffffff"
"34353634353634353634353634000000"
"0d000000ffffffffffffffffffffffff"
"00000000000000000000000000000000"
"00000000000000000000000000000000"
"0d000000ffffffffffffffffffffffff"
"01010000ffffffffffffffffffffffff"
"37383837383837383837383837000000"
"ffffffffffffffffffff"
)
# replacing item with value with more 1 bits than before (wipe and new entry)
n.set(0x0101, b"7877877877877")
value = n.get(0x0101)
assert value == b"7877877877877"
assert (
n._dump()[0][:266].hex()
== consts.NORCOW_MAGIC_AND_VERSION.hex() + "ffffffffffffffff"
"0d000000ffffffffffffffffffffffff"
"01000000ffffffffffffffffffffffff"
"31323331323331323331323331000000"
"0d000000ffffffffffffffffffffffff"
"02000000ffffffffffffffffffffffff"
"34353634353634353634353634000000"
"0d000000ffffffffffffffffffffffff"
"00000000000000000000000000000000"
"00000000000000000000000000000000"
"0d000000ffffffffffffffffffffffff"
"00000000000000000000000000000000"
"00000000000000000000000000000000"
"0d000000ffffffffffffffffffffffff"
"01010000ffffffffffffffffffffffff"
"37383737383737383737383737000000"
"ffffffffffffffffffff"
)
n.set(0x0002, b"world")
n.set(0x0002, b"earth")
value = n.get(0x0002)
assert value == b"earth"
def test_norcow_replace_item_qw():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0001, b"123")
n.set(0x0002, b"456")
n.set(0x0101, b"789")
value = n.get(0x0002)
assert value == b"456"
n.replace(0x0001, b"000")
value = n.get(0x0001)
assert value == b"000"
n.replace(0x0002, b"111")
value = n.get(0x0002)
assert value == b"111"
value = n.get(0x0001)
assert value == b"000"
value = n.get(0x0101)
assert value == b"789"
with pytest.raises(RuntimeError) as e:
n.replace(0x0001, b"00000")
assert "same length" in str(e.value)
def test_norcow_compact_qw():
n = norcow.Norcow(flash_byte_access=False)
n.init()
n.set(0x0101, b"ahoj_ahoj_ahoj")
n.set(0x0101, b"a" * (consts.NORCOW_SECTOR_SIZE - 380))
n.set(0x0101, b"hello_hello__")
n.set(0x0103, b"123456789xxxx")
n.set(0x0104, b"123456789xxxx")
n.set(0x0105, b"123456789xxxx")
n.set(0x0106, b"123456789xxxx")
mem = n._dump()
assert mem[0][:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
assert mem[0][200:300] == b"\x00" * 100
# compact is triggered
n.set(0x0107, b"123456789xxxx")
mem = n._dump()
# assert the other sector is active
assert mem[1][:16] == consts.NORCOW_MAGIC_AND_VERSION + b"\xff" * 8
# assert the deleted item was not copied
assert mem[0][200:300] == b"\xff" * 100
n.set(0x0108, b"123456789x")
n.set(0x0109, b"123456789x")
assert n.get(0x0101) == b"hello_hello__"
assert n.get(0x0103) == b"123456789xxxx"

@ -7,10 +7,11 @@ test_uid = b"\x67\xce\x6a\xe8\xf7\x9b\x73\x96\x83\x88\x21\x5e"
def init(
unlock: bool = False, reseed: int = 0, uid: int = test_uid
unlock: bool = False, reseed: int = 0, uid: int = test_uid, flash_byte_access=True
) -> (StorageC, StoragePy):
sc = StorageC()
sp = StoragePy()
sc = StorageC(flash_byte_access)
sp = StoragePy(flash_byte_access)
sc.lib.random_reseed(reseed)
prng.random_reseed(reseed)
for s in (sc, sp):

@ -6,27 +6,31 @@ from . import common
def test_compact():
sc, sp = common.init(unlock=True)
for s in (sc, sp):
s.set(0xBEEF, b"hello")
s.set(0xBEEF, b"asdasdasdasd")
s.set(0xBEEF, b"fsdasdasdasdasdsadasdsadasdasd")
s.set(0x0101, b"a" * (consts.NORCOW_SECTOR_SIZE - 600))
s.set(0x03FE, b"world!")
s.set(0x04FE, b"world!xfffffffffffffffffffffffffffff")
s.set(0x05FE, b"world!affffffffffffffffffffffffffffff")
s.set(0x0101, b"s")
s.set(0x06FE, b"world!aaaaaaaaaaaaaaaaaaaaaaaaab")
s.set(0x07FE, b"worxxxxxxxxxxxxxxxxxx")
s.set(0x09EE, b"worxxxxxxxxxxxxxxxxxx")
assert common.memory_equals(sc, sp)
for byte_access in (
True,
False,
):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
s.set(0xBEEF, b"hello")
s.set(0xBEEF, b"asdasdasdasd")
s.set(0xBEEF, b"fsdasdasdasdasdsadasdsadasdasd")
s.set(0x0101, b"a" * (consts.NORCOW_SECTOR_SIZE - 1200))
s.set(0x03FE, b"world!")
s.set(0x04FE, b"world!xfffffffffffffffffffffffffffff")
s.set(0x05FE, b"world!affffffffffffffffffffffffffffff")
s.set(0x0101, b"s")
s.set(0x06FE, b"world!aaaaaaaaaaaaaaaaaaaaaaaaab")
s.set(0x07FE, b"worxxxxxxxxxxxxxxxxxx")
s.set(0x09EE, b"worxxxxxxxxxxxxxxxxxx")
assert common.memory_equals(sc, sp)
sc, sp = common.init(unlock=True)
for s in (sc, sp):
s.set(0xBEEF, b"asdasdasdasd")
s.set(0xBEEF, b"fsdasdasdasdasdsadasdsadasdasd")
s.set(0x8101, b"a" * (consts.NORCOW_SECTOR_SIZE - 1000))
with pytest.raises(RuntimeError):
s.set(0x0101, b"a" * (consts.NORCOW_SECTOR_SIZE - 100))
s.set(0x0101, b"hello")
assert common.memory_equals(sc, sp)
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
s.set(0xBEEF, b"asdasdasdasd")
s.set(0xBEEF, b"fsdasdasdasdasdsadasdsadasdasd")
s.set(0x8101, b"a" * (consts.NORCOW_SECTOR_SIZE - 1000))
with pytest.raises(RuntimeError):
s.set(0x0101, b"a" * (consts.NORCOW_SECTOR_SIZE - 100))
s.set(0x0101, b"hello")
assert common.memory_equals(sc, sp)

@ -6,61 +6,69 @@ from . import common
def test_init_pin():
sc, sp = common.init(uid=b"\x00\x00\x00\x00\x00\x00")
assert common.memory_equals(sc, sp)
for byte_access in (True, False):
sc, sp = common.init(
uid=b"\x00\x00\x00\x00\x00\x00", flash_byte_access=byte_access
)
assert common.memory_equals(sc, sp)
sc, sp = common.init(uid=b"\x22\x00\xDD\x00\x00\xBE")
assert common.memory_equals(sc, sp)
sc, sp = common.init(
uid=b"\x22\x00\xDD\x00\x00\xBE", flash_byte_access=byte_access
)
assert common.memory_equals(sc, sp)
def test_change_pin():
sc, sp = common.init(unlock=True)
for s in (sc, sp):
assert s.change_pin("", "222")
assert not s.change_pin("9999", "") # invalid PIN
assert s.unlock("222")
assert s.change_pin("222", "99999")
assert s.change_pin("99999", "Trezor")
assert s.unlock("Trezor")
assert not s.unlock("9999") # invalid PIN
assert not s.unlock("99999") # invalid old PIN
for byte_access in (True, False):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
assert s.change_pin("", "222")
assert not s.change_pin("9999", "") # invalid PIN
assert s.unlock("222")
assert s.change_pin("222", "99999")
assert s.change_pin("99999", "Trezor")
assert s.unlock("Trezor")
assert not s.unlock("9999") # invalid PIN
assert not s.unlock("99999") # invalid old PIN
assert common.memory_equals(sc, sp)
assert common.memory_equals(sc, sp)
def test_has_pin():
sc, sp = common.init()
for s in (sc, sp):
assert not s.has_pin()
assert s.unlock("")
assert not s.has_pin()
assert s.change_pin("", "22")
assert s.has_pin()
assert s.change_pin("22", "")
assert not s.has_pin()
for byte_access in (True, False):
sc, sp = common.init(flash_byte_access=byte_access)
for s in (sc, sp):
assert not s.has_pin()
assert s.unlock("")
assert not s.has_pin()
assert s.change_pin("", "22")
assert s.has_pin()
assert s.change_pin("22", "")
assert not s.has_pin()
def test_wipe_after_max_pin():
sc, sp = common.init(unlock=True)
for s in (sc, sp):
assert s.change_pin("", "222")
assert s.unlock("222")
s.set(0x0202, b"Hello")
for byte_access in (True, False):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
assert s.change_pin("", "222")
assert s.unlock("222")
s.set(0x0202, b"Hello")
# try an invalid PIN MAX - 1 times
for i in range(consts.PIN_MAX_TRIES - 1):
assert not s.unlock("9999")
# this should pass
assert s.unlock("222")
assert s.get(0x0202) == b"Hello"
# try an invalid PIN MAX times, the storage should get wiped
for i in range(consts.PIN_MAX_TRIES):
assert not s.unlock("9999")
assert i == consts.PIN_MAX_TRIES - 1
# this should return False and raise an exception, the storage is wiped
assert not s.unlock("222")
with pytest.raises(RuntimeError):
# try an invalid PIN MAX - 1 times
for i in range(consts.PIN_MAX_TRIES - 1):
assert not s.unlock("9999")
# this should pass
assert s.unlock("222")
assert s.get(0x0202) == b"Hello"
assert common.memory_equals(sc, sp)
# try an invalid PIN MAX times, the storage should get wiped
for i in range(consts.PIN_MAX_TRIES):
assert not s.unlock("9999")
assert i == consts.PIN_MAX_TRIES - 1
# this should return False and raise an exception, the storage is wiped
assert not s.unlock("222")
with pytest.raises(RuntimeError):
assert s.get(0x0202) == b"Hello"
assert common.memory_equals(sc, sp)

@ -0,0 +1,86 @@
import hypothesis.strategies as st
from hypothesis import assume, settings
from hypothesis.stateful import Bundle, RuleBasedStateMachine, invariant, rule
from . import common
from .storage_model import StorageModel
class StorageComparison(RuleBasedStateMachine):
def __init__(self):
super(StorageComparison, self).__init__()
self.sc, self.sp = common.init(unlock=True, flash_byte_access=False)
self.sm = StorageModel()
self.sm.init(b"")
self.sm.unlock("")
self.storages = (self.sc, self.sp, self.sm)
keys = Bundle("keys")
values = Bundle("values")
pins = Bundle("pins")
@rule(target=keys, app=st.integers(1, 0xFF), key=st.integers(0, 0xFF))
def k(self, app, key):
return (app << 8) | key
@rule(target=values, v=st.binary(min_size=0, max_size=10000))
def v(self, v):
return v
@rule(target=pins, p=st.integers(1, 3))
def p(self, p):
if p == 1:
return ""
else:
return str(p)
@rule(k=keys, v=values)
def set(self, k, v):
assume(k != 0xFFFF)
for s in self.storages:
s.set(k, v)
@rule(k=keys)
def delete(self, k):
assume(k != 0xFFFF)
assert len(set(s.delete(k) for s in self.storages)) == 1
@rule(p=pins)
def check_pin(self, p):
assert len(set(s.unlock(p) for s in self.storages)) == 1
self.ensure_unlocked()
@rule(oldpin=pins, newpin=pins)
def change_pin(self, oldpin, newpin):
assert len(set(s.change_pin(oldpin, newpin) for s in self.storages)) == 1
self.ensure_unlocked()
@rule()
def lock(self):
for s in self.storages:
s.lock()
self.ensure_unlocked()
@invariant()
def values_agree(self):
for k, v in self.sm:
assert self.sc.get(k) == v
@invariant()
def dumps_agree(self):
assert self.sc._dump() == self.sp._dump()
@invariant()
def pin_counters_agree(self):
assert len(set(s.get_pin_rem() for s in self.storages)) == 1
def ensure_unlocked(self):
if not self.sm.unlocked:
for s in self.storages:
assert s.unlock(self.sm.pin)
TestStorageComparison = StorageComparison.TestCase
TestStorageComparison.settings = settings(
deadline=None, max_examples=30, stateful_step_count=50
)

@ -13,189 +13,211 @@ chacha_strings = [
]
def test_set_delete():
for byte_access in (True, False):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
s.set(0xFF04, b"0123456789A")
s.delete(0xFF04)
s.set(0xFF04, b"0123456789AB")
s.delete(0xFF04)
s.set(0xFF04, b"0123456789ABC")
s.delete(0xFF04)
assert common.memory_equals(sc, sp)
def test_set_get():
sc, sp = common.init(unlock=True)
for s in (sc, sp):
s.set(0xBEEF, b"Hello")
s.set(0xCAFE, b"world! ")
s.set(0xDEAD, b"How\n")
s.set(0xAAAA, b"are")
s.set(0x0901, b"you?")
s.set(0x0902, b"Lorem")
s.set(0x0903, b"ipsum")
s.set(0xDEAD, b"A\n")
s.set(0xDEAD, b"AAAAAAAAAAA")
s.set(0x2200, b"BBBB")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.change_pin("", "222")
s.change_pin("222", "99")
s.set(0xAAAA, b"something else")
assert common.memory_equals(sc, sp)
# check data are not changed by gets
datasc = sc._dump()
datasp = sp._dump()
for s in (sc, sp):
assert s.get(0xAAAA) == b"something else"
assert s.get(0x0901) == b"you?"
assert s.get(0x0902) == b"Lorem"
assert s.get(0x0903) == b"ipsum"
assert s.get(0xDEAD) == b"AAAAAAAAAAA"
assert s.get(0x2200) == b"BBBB"
assert datasc == sc._dump()
assert datasp == sp._dump()
# test locked storage
for s in (sc, sp):
s.lock()
with pytest.raises(RuntimeError):
s.set(0xAAAA, b"test public")
with pytest.raises(RuntimeError):
s.set(0x0901, b"test protected")
with pytest.raises(RuntimeError):
s.get(0x0901)
assert s.get(0xAAAA) == b"something else"
# check that storage functions after unlock
for s in (sc, sp):
s.unlock("99")
s.set(0xAAAA, b"public")
s.set(0x0902, b"protected")
assert s.get(0xAAAA) == b"public"
assert s.get(0x0902) == b"protected"
# test delete
for s in (sc, sp):
assert s.delete(0x0902)
assert common.memory_equals(sc, sp)
for s in (sc, sp):
assert not s.delete(0x7777)
assert not s.delete(0x0902)
assert common.memory_equals(sc, sp)
for byte_access in (True, False):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
s.set(0xBEEF, b"Hello")
s.set(0xCAFE, b"world! ")
s.set(0xDEAD, b"How\n")
s.set(0xAAAA, b"are")
s.set(0x0901, b"you?")
s.set(0x0902, b"Lorem")
s.set(0x0903, b"ipsum")
s.set(0xDEAD, b"A\n")
s.set(0xDEAD, b"AAAAAAAAAAA")
s.set(0x2200, b"BBBB")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.change_pin("", "222")
s.change_pin("222", "99")
s.set(0xAAAA, b"something else")
assert common.memory_equals(sc, sp)
# check data are not changed by gets
datasc = sc._dump()
datasp = sp._dump()
for s in (sc, sp):
assert s.get(0xAAAA) == b"something else"
assert s.get(0x0901) == b"you?"
assert s.get(0x0902) == b"Lorem"
assert s.get(0x0903) == b"ipsum"
assert s.get(0xDEAD) == b"AAAAAAAAAAA"
assert s.get(0x2200) == b"BBBB"
assert datasc == sc._dump()
assert datasp == sp._dump()
# test locked storage
for s in (sc, sp):
s.lock()
with pytest.raises(RuntimeError):
s.set(0xAAAA, b"test public")
with pytest.raises(RuntimeError):
s.set(0x0901, b"test protected")
with pytest.raises(RuntimeError):
s.get(0x0901)
assert s.get(0xAAAA) == b"something else"
# check that storage functions after unlock
for s in (sc, sp):
s.unlock("99")
s.set(0xAAAA, b"public")
s.set(0x0902, b"protected")
assert s.get(0xAAAA) == b"public"
assert s.get(0x0902) == b"protected"
# test delete
for s in (sc, sp):
assert s.delete(0x0902)
assert common.memory_equals(sc, sp)
for s in (sc, sp):
assert not s.delete(0x7777)
assert not s.delete(0x0902)
assert common.memory_equals(sc, sp)
def test_invalid_key():
for s in common.init(unlock=True):
with pytest.raises(RuntimeError):
s.set(0xFFFF, b"Hello")
for byte_access in (True, False):
for s in common.init(unlock=True, flash_byte_access=byte_access):
with pytest.raises(RuntimeError):
s.set(0xFFFF, b"Hello")
def test_non_existing_key():
sc, sp = common.init()
for s in (sc, sp):
with pytest.raises(RuntimeError):
s.get(0xABCD)
for byte_access in (True, False):
sc, sp = common.init(flash_byte_access=byte_access)
for s in (sc, sp):
with pytest.raises(RuntimeError):
s.get(0xABCD)
def test_chacha_strings():
sc, sp = common.init(unlock=True)
for s in (sc, sp):
for i, string in enumerate(chacha_strings):
s.set(0x0301 + i, string)
assert common.memory_equals(sc, sp)
for byte_access in (True, False):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
for i, string in enumerate(chacha_strings):
s.set(0x0301 + i, string)
assert common.memory_equals(sc, sp)
for s in (sc, sp):
for i, string in enumerate(chacha_strings):
assert s.get(0x0301 + i) == string
for s in (sc, sp):
for i, string in enumerate(chacha_strings):
assert s.get(0x0301 + i) == string
def test_set_repeated():
test_strings = [[0x0501, b""], [0x0502, b"test"], [0x8501, b""], [0x8502, b"test"]]
sc, sp = common.init(unlock=True)
for s in (sc, sp):
for key, val in test_strings:
s.set(key, val)
s.set(key, val)
assert common.memory_equals(sc, sp)
for s in (sc, sp):
for key, val in test_strings:
s.set(key, val)
assert common.memory_equals(sc, sp)
for byte_access in (False,):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
for key, val in test_strings:
s.set(key, val)
s.set(key, val)
assert common.memory_equals(sc, sp)
for key, val in test_strings:
for s in (sc, sp):
assert s.delete(key)
for key, val in test_strings:
s.set(key, val)
assert common.memory_equals(sc, sp)
for key, val in test_strings:
for s in (sc, sp):
assert s.delete(key)
assert common.memory_equals(sc, sp)
def test_set_similar():
sc, sp = common.init(unlock=True)
for s in (sc, sp):
s.set(0xBEEF, b"Satoshi")
s.set(0xBEEF, b"satoshi")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.wipe()
s.unlock("")
s.set(0xBEEF, b"satoshi")
s.set(0xBEEF, b"Satoshi")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.wipe()
s.unlock("")
s.set(0xBEEF, b"satoshi")
s.set(0xBEEF, b"Satoshi")
s.set(0xBEEF, b"Satoshi")
s.set(0xBEEF, b"SatosHi")
s.set(0xBEEF, b"satoshi")
s.set(0xBEEF, b"satoshi\x00")
assert common.memory_equals(sc, sp)
for byte_access in (True, False):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for s in (sc, sp):
s.set(0xBEEF, b"Satoshi")
s.set(0xBEEF, b"satoshi")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.wipe()
s.unlock("")
s.set(0xBEEF, b"satoshi")
s.set(0xBEEF, b"Satoshi")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.wipe()
s.unlock("")
s.set(0xBEEF, b"satoshi")
s.set(0xBEEF, b"Satoshi")
s.set(0xBEEF, b"Satoshi")
s.set(0xBEEF, b"SatosHi")
s.set(0xBEEF, b"satoshi")
s.set(0xBEEF, b"satoshi\x00")
assert common.memory_equals(sc, sp)
def test_set_locked():
sc, sp = common.init()
for s in (sc, sp):
with pytest.raises(RuntimeError):
s.set(0x0303, b"test")
with pytest.raises(RuntimeError):
s.set(0x8003, b"test")
assert common.memory_equals(sc, sp)
for byte_access in (True, False):
sc, sp = common.init(flash_byte_access=byte_access)
for s in (sc, sp):
with pytest.raises(RuntimeError):
s.set(0x0303, b"test")
with pytest.raises(RuntimeError):
s.set(0x8003, b"test")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.set(0xC001, b"Ahoj")
s.set(0xC003, b"test")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.set(0xC001, b"Ahoj")
s.set(0xC003, b"test")
assert common.memory_equals(sc, sp)
for s in (sc, sp):
assert s.get(0xC001) == b"Ahoj"
assert s.get(0xC003) == b"test"
for s in (sc, sp):
assert s.get(0xC001) == b"Ahoj"
assert s.get(0xC003) == b"test"
def test_counter():
sc, sp = common.init(unlock=True)
for i in range(0, 200):
for byte_access in (True, False):
sc, sp = common.init(unlock=True, flash_byte_access=byte_access)
for i in range(0, 200):
for s in (sc, sp):
assert i == s.next_counter(0xC001)
assert common.memory_equals(sc, sp)
for s in (sc, sp):
assert i == s.next_counter(0xC001)
s.lock()
s.set_counter(0xC001, 500)
assert common.memory_equals(sc, sp)
for s in (sc, sp):
s.lock()
s.set_counter(0xC001, 500)
assert common.memory_equals(sc, sp)
for i in range(501, 700):
for s in (sc, sp):
assert i == s.next_counter(0xC001)
assert common.memory_equals(sc, sp)
for i in range(501, 700):
for s in (sc, sp):
assert i == s.next_counter(0xC001)
assert common.memory_equals(sc, sp)
for s in (sc, sp):
with pytest.raises(RuntimeError):
s.set_counter(0xC001, consts.UINT32_MAX + 1)
with pytest.raises(RuntimeError):
s.set_counter(0xC001, consts.UINT32_MAX + 1)
start = consts.UINT32_MAX - 100
s.set_counter(0xC001, start)
for i in range(start, consts.UINT32_MAX):
assert i + 1 == s.next_counter(0xC001)
start = consts.UINT32_MAX - 100
s.set_counter(0xC001, start)
for i in range(start, consts.UINT32_MAX):
assert i + 1 == s.next_counter(0xC001)
with pytest.raises(RuntimeError):
s.next_counter(0xC001)
with pytest.raises(RuntimeError):
s.next_counter(0xC001)
assert common.memory_equals(sc, sp)
assert common.memory_equals(sc, sp)

@ -59,18 +59,19 @@ def test_upgrade():
def test_python_set_sectors():
sp0 = StoragePy()
sp0.init(common.test_uid)
assert sp0.unlock("")
set_values(sp0)
for _ in range(10):
assert not sp0.unlock("3")
assert sp0.get_pin_rem() == 6
for byte_access in (True, False):
sp0 = StoragePy(byte_access)
sp0.init(common.test_uid)
assert sp0.unlock("")
set_values(sp0)
for _ in range(10):
assert not sp0.unlock("3")
assert sp0.get_pin_rem() == 6
sp1 = StoragePy()
sp1.nc._set_sectors(sp0._dump())
sp1.init(common.test_uid)
common.memory_equals(sp0, sp1)
sp1 = StoragePy(byte_access)
sp1.nc._set_sectors(sp0._dump())
sp1.init(common.test_uid)
common.memory_equals(sp0, sp1)
assert sp1.get_pin_rem() == 6
check_values(sp1)
assert sp1.get_pin_rem() == 6
check_values(sp1)

Loading…
Cancel
Save