chore(core): divide aesgcm into `aesgcm_encrypt` and `aesgcm_decrypt`
What changed, and why it matters
This commit is a routine code cleanup that splits one combined AES-GCM encryption/decryption class into two separate classes: one for encryption and one for decryption. It does not fix a security bug or introduce a new vulnerability. The change makes the API clearer and prevents accidental misuse, such as trying to decrypt with an encryption-only object.
No security action required. Treat as a normal refactoring commit. Reviewers may verify that all former `aesgcm` consumers were updated and that the new decrypt-only class is used wherever ciphertext is verified.
Security signals we found
API hardening: encryption and decryption contexts are now separate types, reducing risk of calling the wrong operation on a shared context
Decrypt finish now requires a 16-byte expected tag and uses constant-time comparison via consteq
No change to cryptographic primitives, key handling, IV generation, or tag computation
Evidence from the diff
The patch refactors the MicroPython trezorcrypto.aesgcm binding. It replaces the single aesgcm class with aesgcm_encrypt and aesgcm_decrypt, each exposing only the relevant methods. State machine states are simplified, finish() is now type-specific (returns tag for encrypt, verifies tag for decrypt), and callers in THP crypto, benchmarks, and tests are updated. The underlying gcm_* primitives and tag verification logic remain unchanged.
Changed components
core/embed/upymod/modtrezorcrypto/modtrezorcrypto-aesgcm.hcore/embed/upymod/modtrezorcrypto/modtrezorcrypto.ccore/src/trezor/crypto/__init__.pycore/src/trezor/wire/thp/crypto.pycore/src/apps/benchmark/benchmarks.pycore/src/apps/benchmark/cipher_benchmark.pycore/tests/test_trezor.crypto.aesgcm.pycore/tests/test_trezor.wire.thp.crypto.pycore/mocks/generated/trezorcrypto/__init__.pyiInspect captured patch +230 / −155
diff --git a/core/embed/upymod/modtrezorcrypto/modtrezorcrypto-aesgcm.h b/core/embed/upymod/modtrezorcrypto/modtrezorcrypto-aesgcm.h
index ba76978a..1fc2472f 100644
--- a/core/embed/upymod/modtrezorcrypto/modtrezorcrypto-aesgcm.h
+++ b/core/embed/upymod/modtrezorcrypto/modtrezorcrypto-aesgcm.h
@@ -25,17 +25,16 @@
/// package: trezorcrypto.__init__
-/// class aesgcm:
+/// class aesgcm_encrypt:
/// """
-/// AES-GCM context.
+/// AES-GCM context for encryption.
/// """
typedef struct _mp_obj_AesGcm_t {
mp_obj_base_t base;
gcm_ctx ctx;
enum {
STATE_INIT,
- STATE_ENCRYPTING,
- STATE_DECRYPTING,
+ STATE_PROCESSING,
STATE_FINISHED,
STATE_FAILED,
} state;
@@ -43,7 +42,7 @@ typedef struct _mp_obj_AesGcm_t {
/// def __init__(self, key: AnyBytes, iv: AnyBytes) -> None:
/// """
-/// Initialize the AES-GCM context for encryption or decryption.
+/// Initialize the AES-GCM context for encryption.
/// """
STATIC mp_obj_t mod_trezorcrypto_AesGcm_make_new(const mp_obj_type_t *type,
size_t n_args, size_t n_kw,
@@ -68,9 +67,31 @@ STATIC mp_obj_t mod_trezorcrypto_AesGcm_make_new(const mp_obj_type_t *type,
return MP_OBJ_FROM_PTR(o);
}
+/// def auth(self, data: AnyBytes) -> None:
+/// """
+/// Include authenticated data chunk in the GCM authentication tag. This can
+/// be called repeatedly to add authenticated data at any point before
+/// finish().
+/// """
+STATIC mp_obj_t mod_trezorcrypto_AesGcm_auth(mp_obj_t self, mp_obj_t data) {
+ mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
+ if (o->state != STATE_INIT && o->state != STATE_PROCESSING) {
+ mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
+ }
+ mp_buffer_info_t in = {0};
+ mp_get_buffer_raise(data, &in, MP_BUFFER_READ);
+ if (gcm_auth_header(in.buf, in.len, &(o->ctx)) != RETURN_GOOD) {
+ o->state = STATE_FAILED;
+ mp_raise_type(&mp_type_RuntimeError);
+ }
+ return mp_const_none;
+}
+STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_auth_obj,
+ mod_trezorcrypto_AesGcm_auth);
+
/// def reset(self, iv: AnyBytes) -> None:
/// """
-/// Reset the IV for encryption or decryption.
+/// Reset the IV for encryption.
/// """
STATIC mp_obj_t mod_trezorcrypto_AesGcm_reset(mp_obj_t self, mp_obj_t iv) {
mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
@@ -92,10 +113,10 @@ STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_reset_obj,
/// """
STATIC mp_obj_t mod_trezorcrypto_AesGcm_encrypt(mp_obj_t self, mp_obj_t data) {
mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
- if (o->state != STATE_INIT && o->state != STATE_ENCRYPTING) {
+ if (o->state != STATE_INIT && o->state != STATE_PROCESSING) {
mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
}
- o->state = STATE_ENCRYPTING;
+ o->state = STATE_PROCESSING;
mp_buffer_info_t in = {0};
mp_get_buffer_raise(data, &in, MP_BUFFER_READ);
vstr_t vstr = {0};
@@ -118,10 +139,10 @@ STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_encrypt_obj,
STATIC mp_obj_t mod_trezorcrypto_AesGcm_encrypt_in_place(mp_obj_t self,
mp_obj_t data) {
mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
- if (o->state != STATE_INIT && o->state != STATE_ENCRYPTING) {
+ if (o->state != STATE_INIT && o->state != STATE_PROCESSING) {
mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
}
- o->state = STATE_ENCRYPTING;
+ o->state = STATE_PROCESSING;
mp_buffer_info_t in = {0};
mp_get_buffer_raise(data, &in, MP_BUFFER_READ | MP_BUFFER_WRITE);
if (gcm_encrypt((unsigned char *)in.buf, in.len, &(o->ctx)) != RETURN_GOOD) {
@@ -133,16 +154,61 @@ STATIC mp_obj_t mod_trezorcrypto_AesGcm_encrypt_in_place(mp_obj_t self,
STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_encrypt_in_place_obj,
mod_trezorcrypto_AesGcm_encrypt_in_place);
+/// def finish(self) -> bytes:
+/// """
+/// Compute the GCM authentication tag.
+/// """
+STATIC mp_obj_t mod_trezorcrypto_AesGcm_encrypt_finish(mp_obj_t self) {
+ mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
+ if (o->state != STATE_INIT && o->state != STATE_PROCESSING) {
+ mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
+ }
+
+ o->state = STATE_FINISHED;
+ vstr_t tag = {0};
+ vstr_init_len(&tag, 16);
+ if (gcm_compute_tag((unsigned char *)tag.buf, tag.len, &(o->ctx)) !=
+ RETURN_GOOD) {
+ o->state = STATE_FAILED;
+ mp_raise_type(&mp_type_RuntimeError);
+ }
+ return mp_obj_new_str_from_vstr(&mp_type_bytes, &tag);
+}
+STATIC MP_DEFINE_CONST_FUN_OBJ_1(mod_trezorcrypto_AesGcm_encrypt_finish_obj,
+ mod_trezorcrypto_AesGcm_encrypt_finish);
+
+/// class aesgcm_decrypt:
+/// """
+/// AES-GCM context for decryption.
+/// """
+
+/// def __init__(self, key: AnyBytes, iv: AnyBytes) -> None:
+/// """
+/// Initialize the AES-GCM context for decryption.
+/// """
+
+/// def auth(self, data: AnyBytes) -> None:
+/// """
+/// Include authenticated data chunk in the GCM authentication tag. This can
+/// be called repeatedly to add authenticated data at any point before
+/// finish().
+/// """
+
+/// def reset(self, iv: AnyBytes) -> None:
+/// """
+/// Reset the IV for decryption.
+/// """
+
/// def decrypt(self, data: AnyBytes) -> bytes:
/// """
/// Decrypt data chunk.
/// """
STATIC mp_obj_t mod_trezorcrypto_AesGcm_decrypt(mp_obj_t self, mp_obj_t data) {
mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
- if (o->state != STATE_INIT && o->state != STATE_DECRYPTING) {
+ if (o->state != STATE_INIT && o->state != STATE_PROCESSING) {
mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
}
- o->state = STATE_DECRYPTING;
+ o->state = STATE_PROCESSING;
mp_buffer_info_t in = {0};
mp_get_buffer_raise(data, &in, MP_BUFFER_READ);
vstr_t vstr = {0};
@@ -165,10 +231,10 @@ STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_decrypt_obj,
STATIC mp_obj_t mod_trezorcrypto_AesGcm_decrypt_in_place(mp_obj_t self,
mp_obj_t data) {
mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
- if (o->state != STATE_INIT && o->state != STATE_DECRYPTING) {
+ if (o->state != STATE_INIT && o->state != STATE_PROCESSING) {
mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
}
- o->state = STATE_DECRYPTING;
+ o->state = STATE_PROCESSING;
mp_buffer_info_t in = {0};
mp_get_buffer_raise(data, &in, MP_BUFFER_READ | MP_BUFFER_WRITE);
if (gcm_decrypt((unsigned char *)in.buf, in.len, &(o->ctx)) != RETURN_GOOD) {
@@ -180,48 +246,24 @@ STATIC mp_obj_t mod_trezorcrypto_AesGcm_decrypt_in_place(mp_obj_t self,
STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_decrypt_in_place_obj,
mod_trezorcrypto_AesGcm_decrypt_in_place);
-/// def auth(self, data: AnyBytes) -> None:
+/// def finish(self, expected_tag: AnyBytes) -> None:
/// """
-/// Include authenticated data chunk in the GCM authentication tag. This can
-/// be called repeatedly to add authenticated data at any point before
-/// finish().
+/// Verify the GCM authentication tag.
/// """
-STATIC mp_obj_t mod_trezorcrypto_AesGcm_auth(mp_obj_t self, mp_obj_t data) {
+STATIC mp_obj_t mod_trezorcrypto_AesGcm_decrypt_finish(mp_obj_t self,
+ mp_obj_t expected_tag) {
mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
- if (o->state != STATE_INIT && o->state != STATE_ENCRYPTING &&
- o->state != STATE_DECRYPTING) {
+ if (o->state != STATE_INIT && o->state != STATE_PROCESSING) {
mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
}
- mp_buffer_info_t in = {0};
- mp_get_buffer_raise(data, &in, MP_BUFFER_READ);
- if (gcm_auth_header(in.buf, in.len, &(o->ctx)) != RETURN_GOOD) {
- o->state = STATE_FAILED;
- mp_raise_type(&mp_type_RuntimeError);
- }
- return mp_const_none;
-}
-STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_auth_obj,
- mod_trezorcrypto_AesGcm_auth);
-
-/// def finish(self, expected_tag: AnyBytes | None = None) -> bytes:
-/// """
-/// Compute GCM authentication tag. The `expected_tag` is required when
-/// decrypting.
-/// """
-STATIC mp_obj_t mod_trezorcrypto_AesGcm_finish(size_t n_args,
- const mp_obj_t *args) {
- mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(args[0]);
- if (o->state != STATE_INIT && o->state != STATE_ENCRYPTING &&
- o->state != STATE_DECRYPTING) {
- mp_raise_msg(&mp_type_RuntimeError, MP_ERROR_TEXT("Invalid state."));
- }
- if (n_args == 1 && o->state == STATE_DECRYPTING) {
- mp_raise_msg(
- &mp_type_RuntimeError,
- MP_ERROR_TEXT("Argument `expected_tag` is required when decrypting."));
- }
o->state = STATE_FINISHED;
+ mp_buffer_info_t exp_tag = {0};
+ mp_get_buffer_raise(expected_tag, &exp_tag, MP_BUFFER_READ);
+ if (exp_tag.len != 16) {
+ mp_raise_ValueError(
+ MP_ERROR_TEXT("Invalid length of the tag. It has to be 16 bytes."));
+ }
vstr_t tag = {0};
vstr_init_len(&tag, 16);
if (gcm_compute_tag((unsigned char *)tag.buf, tag.len, &(o->ctx)) !=
@@ -229,24 +271,16 @@ STATIC mp_obj_t mod_trezorcrypto_AesGcm_finish(size_t n_args,
o->state = STATE_FAILED;
mp_raise_type(&mp_type_RuntimeError);
}
- if (n_args == 2) {
- mp_buffer_info_t expected_tag = {0};
- mp_get_buffer_raise(args[1], &expected_tag, MP_BUFFER_READ);
- if (expected_tag.len != 16) {
- mp_raise_ValueError(
- MP_ERROR_TEXT("Invalid length of the tag. It has to be 16 bytes."));
- }
- if (!consteq((uint8_t *)tag.buf, tag.len, (uint8_t *)expected_tag.buf,
- expected_tag.len)) {
- mp_raise_msg(&mp_type_RuntimeError,
- MP_ERROR_TEXT("Authentication failed."));
- }
+
+ if (!consteq(tag.buf, exp_tag.buf, exp_tag.len)) {
+ mp_raise_msg(&mp_type_RuntimeError,
+ MP_ERROR_TEXT("Authentication failed."));
}
- return mp_obj_new_str_from_vstr(&mp_type_bytes, &tag);
+
+ return mp_const_none;
}
-STATIC MP_DEFINE_CONST_FUN_OBJ_VAR_BETWEEN(mod_trezorcrypto_AesGcm_finish_obj,
- 1, 2,
- mod_trezorcrypto_AesGcm_finish);
+STATIC MP_DEFINE_CONST_FUN_OBJ_2(mod_trezorcrypto_AesGcm_decrypt_finish_obj,
+ mod_trezorcrypto_AesGcm_decrypt_finish);
STATIC mp_obj_t mod_trezorcrypto_AesGcm___del__(mp_obj_t self) {
mp_obj_AesGcm_t *o = MP_OBJ_TO_PTR(self);
@@ -256,29 +290,54 @@ STATIC mp_obj_t mod_trezorcrypto_AesGcm___del__(mp_obj_t self) {
STATIC MP_DEFINE_CONST_FUN_OBJ_1(mod_trezorcrypto_AesGcm___del___obj,
mod_trezorcrypto_AesGcm___del__);
-STATIC const mp_rom_map_elem_t mod_trezorcrypto_AesGcm_locals_dict_table[] = {
- {MP_ROM_QSTR(MP_QSTR_reset),
- MP_ROM_PTR(&mod_trezorcrypto_AesGcm_reset_obj)},
- {MP_ROM_QSTR(MP_QSTR_encrypt),
- MP_ROM_PTR(&mod_trezorcrypto_AesGcm_encrypt_obj)},
- {MP_ROM_QSTR(MP_QSTR_encrypt_in_place),
- MP_ROM_PTR(&mod_trezorcrypto_AesGcm_encrypt_in_place_obj)},
- {MP_ROM_QSTR(MP_QSTR_decrypt),
- MP_ROM_PTR(&mod_trezorcrypto_AesGcm_decrypt_obj)},
- {MP_ROM_QSTR(MP_QSTR_decrypt_in_place),
- MP_ROM_PTR(&mod_trezorcrypto_AesGcm_decrypt_in_place_obj)},
- {MP_ROM_QSTR(MP_QSTR_auth), MP_ROM_PTR(&mod_trezorcrypto_AesGcm_auth_obj)},
- {MP_ROM_QSTR(MP_QSTR_finish),
- MP_ROM_PTR(&mod_trezorcrypto_AesGcm_finish_obj)},
- {MP_ROM_QSTR(MP_QSTR___del__),
- MP_ROM_PTR(&mod_trezorcrypto_AesGcm___del___obj)},
+STATIC const mp_rom_map_elem_t
+ mod_trezorcrypto_AesGcmEncrypt_locals_dict_table[] = {
+ {MP_ROM_QSTR(MP_QSTR_auth),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_auth_obj)},
+ {MP_ROM_QSTR(MP_QSTR_reset),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_reset_obj)},
+ {MP_ROM_QSTR(MP_QSTR_encrypt),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_encrypt_obj)},
+ {MP_ROM_QSTR(MP_QSTR_encrypt_in_place),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_encrypt_in_place_obj)},
+ {MP_ROM_QSTR(MP_QSTR_finish),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_encrypt_finish_obj)},
+ {MP_ROM_QSTR(MP_QSTR___del__),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm___del___obj)},
+};
+
+STATIC MP_DEFINE_CONST_DICT(mod_trezorcrypto_AesGcmEncrypt_locals_dict,
+ mod_trezorcrypto_AesGcmEncrypt_locals_dict_table);
+
+STATIC const mp_rom_map_elem_t
+ mod_trezorcrypto_AesGcmDecrypt_locals_dict_table[] = {
+ {MP_ROM_QSTR(MP_QSTR_auth),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_auth_obj)},
+ {MP_ROM_QSTR(MP_QSTR_reset),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_reset_obj)},
+ {MP_ROM_QSTR(MP_QSTR_decrypt),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_decrypt_obj)},
+ {MP_ROM_QSTR(MP_QSTR_decrypt_in_place),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_decrypt_in_place_obj)},
+ {MP_ROM_QSTR(MP_QSTR_finish),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm_decrypt_finish_obj)},
+ {MP_ROM_QSTR(MP_QSTR___del__),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcm___del___obj)},
+};
+
+STATIC MP_DEFINE_CONST_DICT(mod_trezorcrypto_AesGcmDecrypt_locals_dict,
+ mod_trezorcrypto_AesGcmDecrypt_locals_dict_table);
+
+STATIC const mp_obj_type_t mod_trezorcrypto_AesGcmEncrypt_type = {
+ {&mp_type_type},
+ .name = MP_QSTR_aesgcm_encrypt,
+ .make_new = mod_trezorcrypto_AesGcm_make_new,
+ .locals_dict = (void *)&mod_trezorcrypto_AesGcmEncrypt_locals_dict,
};
-STATIC MP_DEFINE_CONST_DICT(mod_trezorcrypto_AesGcm_locals_dict,
- mod_trezorcrypto_AesGcm_locals_dict_table);
-STATIC const mp_obj_type_t mod_trezorcrypto_AesGcm_type = {
+STATIC const mp_obj_type_t mod_trezorcrypto_AesGcmDecrypt_type = {
{&mp_type_type},
- .name = MP_QSTR_AesGcm,
+ .name = MP_QSTR_aesgcm_decrypt,
.make_new = mod_trezorcrypto_AesGcm_make_new,
- .locals_dict = (void *)&mod_trezorcrypto_AesGcm_locals_dict,
+ .locals_dict = (void *)&mod_trezorcrypto_AesGcmDecrypt_locals_dict,
};
diff --git a/core/embed/upymod/modtrezorcrypto/modtrezorcrypto.c b/core/embed/upymod/modtrezorcrypto/modtrezorcrypto.c
index 542164f1..0c2d836e 100644
--- a/core/embed/upymod/modtrezorcrypto/modtrezorcrypto.c
+++ b/core/embed/upymod/modtrezorcrypto/modtrezorcrypto.c
@@ -85,7 +85,10 @@ STATIC const mp_rom_map_elem_t mp_module_trezorcrypto_globals_table[] = {
{MP_ROM_QSTR(MP_QSTR___name__), MP_ROM_QSTR(MP_QSTR_trezorcrypto)},
{MP_ROM_QSTR(MP_QSTR_aes), MP_ROM_PTR(&mod_trezorcrypto_AES_type)},
#if USE_AES_GCM
- {MP_ROM_QSTR(MP_QSTR_aesgcm), MP_ROM_PTR(&mod_trezorcrypto_AesGcm_type)},
+ {MP_ROM_QSTR(MP_QSTR_aesgcm_decrypt),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcmDecrypt_type)},
+ {MP_ROM_QSTR(MP_QSTR_aesgcm_encrypt),
+ MP_ROM_PTR(&mod_trezorcrypto_AesGcmEncrypt_type)},
#endif
{MP_ROM_QSTR(MP_QSTR_bech32), MP_ROM_PTR(&mod_trezorcrypto_bech32_module)},
{MP_ROM_QSTR(MP_QSTR_bip32), MP_ROM_PTR(&mod_trezorcrypto_bip32_module)},
diff --git a/core/mocks/generated/trezorcrypto/__init__.pyi b/core/mocks/generated/trezorcrypto/__init__.pyi
index 7afe91e1..b89216ec 100644
--- a/core/mocks/generated/trezorcrypto/__init__.pyi
+++ b/core/mocks/generated/trezorcrypto/__init__.pyi
@@ -35,19 +35,26 @@ class aes:
# upymod/modtrezorcrypto/modtrezorcrypto-aesgcm.h
-class aesgcm:
+class aesgcm_encrypt:
"""
- AES-GCM context.
+ AES-GCM context for encryption.
"""
def __init__(self, key: AnyBytes, iv: AnyBytes) -> None:
"""
- Initialize the AES-GCM context for encryption or decryption.
+ Initialize the AES-GCM context for encryption.
+ """
+
+ def auth(self, data: AnyBytes) -> None:
+ """
+ Include authenticated data chunk in the GCM authentication tag. This can
+ be called repeatedly to add authenticated data at any point before
+ finish().
"""
def reset(self, iv: AnyBytes) -> None:
"""
- Reset the IV for encryption or decryption.
+ Reset the IV for encryption.
"""
def encrypt(self, data: AnyBytes) -> bytes:
@@ -60,14 +67,21 @@ class aesgcm:
Encrypt data chunk in place. Returns the length of the encrypted data.
"""
- def decrypt(self, data: AnyBytes) -> bytes:
+ def finish(self) -> bytes:
"""
- Decrypt data chunk.
+ Compute the GCM authentication tag.
"""
- def decrypt_in_place(self, data: AnyBuffer) -> int:
+
+# upymod/modtrezorcrypto/modtrezorcrypto-aesgcm.h
+class aesgcm_decrypt:
+ """
+ AES-GCM context for decryption.
+ """
+
+ def __init__(self, key: AnyBytes, iv: AnyBytes) -> None:
"""
- Decrypt data chunk in place. Returns the length of the decrypted data.
+ Initialize the AES-GCM context for decryption.
"""
def auth(self, data: AnyBytes) -> None:
@@ -77,10 +91,24 @@ class aesgcm:
finish().
"""
- def finish(self, expected_tag: AnyBytes | None = None) -> bytes:
+ def reset(self, iv: AnyBytes) -> None:
"""
- Compute GCM authentication tag. The `expected_tag` is required when
- decrypting.
+ Reset the IV for decryption.
+ """
+
+ def decrypt(self, data: AnyBytes) -> bytes:
+ """
+ Decrypt data chunk.
+ """
+
+ def decrypt_in_place(self, data: AnyBuffer) -> int:
+ """
+ Decrypt data chunk in place. Returns the length of the decrypted data.
+ """
+
+ def finish(self, expected_tag: AnyBytes) -> None:
+ """
+ Verify the GCM authentication tag.
"""
diff --git a/core/src/apps/benchmark/benchmarks.py b/core/src/apps/benchmark/benchmarks.py
index cc6657fe..2b2c43af 100644
--- a/core/src/apps/benchmark/benchmarks.py
+++ b/core/src/apps/benchmark/benchmarks.py
@@ -1,4 +1,4 @@
-from trezor.crypto import aes, aesgcm, chacha20poly1305
+from trezor.crypto import aes, aesgcm_decrypt, aesgcm_encrypt, chacha20poly1305
from trezor.crypto.curve import curve25519, ed25519, nist256p1, secp256k1
from trezor.crypto.hashlib import (
blake2b,
@@ -63,16 +63,16 @@ benchmarks = {
lambda: aes(aes.ECB, random_bytes(16), random_bytes(16)), 16
),
"crypto/cipher/aesgcm128/encrypt": EncryptBenchmark(
- lambda: aesgcm(random_bytes(16), random_bytes(16)), 16
+ lambda: aesgcm_encrypt(random_bytes(16), random_bytes(16)), 16
),
"crypto/cipher/aesgcm128/decrypt": DecryptBenchmark(
- lambda: aesgcm(random_bytes(16), random_bytes(16)), 16
+ lambda: aesgcm_decrypt(random_bytes(16), random_bytes(16)), 16
),
"crypto/cipher/aesgcm256/encrypt": EncryptBenchmark(
- lambda: aesgcm(random_bytes(32), random_bytes(16)), 16
+ lambda: aesgcm_encrypt(random_bytes(32), random_bytes(16)), 16
),
"crypto/cipher/aesgcm256/decrypt": DecryptBenchmark(
- lambda: aesgcm(random_bytes(32), random_bytes(16)), 16
+ lambda: aesgcm_decrypt(random_bytes(32), random_bytes(16)), 16
),
"crypto/cipher/chacha20poly1305/encrypt": EncryptBenchmark(
lambda: chacha20poly1305(random_bytes(32), random_bytes(12)), 64
diff --git a/core/src/apps/benchmark/cipher_benchmark.py b/core/src/apps/benchmark/cipher_benchmark.py
index 6546ff93..6d4205cf 100644
--- a/core/src/apps/benchmark/cipher_benchmark.py
+++ b/core/src/apps/benchmark/cipher_benchmark.py
@@ -7,15 +7,16 @@ from .common import format_float, maximum_used_memory_in_bytes, random_bytes
if TYPE_CHECKING:
from typing import Protocol
- class CipherCtx(Protocol):
- def encrypt(self, data: bytes) -> bytes: ...
-
+ class CipherCtxDecrypt(Protocol):
def decrypt(self, data: bytes) -> bytes: ...
+ class CipherCtxEncrypt(Protocol):
+ def encrypt(self, data: bytes) -> bytes: ...
+
class EncryptBenchmark:
def __init__(
- self, cipher_ctx_constructor: Callable[[], CipherCtx], block_size: int
+ self, cipher_ctx_constructor: Callable[[], CipherCtxEncrypt], block_size: int
) -> None:
self.cipher_ctx_constructor = cipher_ctx_constructor
self.block_size = block_size
@@ -43,7 +44,7 @@ class EncryptBenchmark:
class DecryptBenchmark:
def __init__(
- self, cipher_ctx_constructor: Callable[[], CipherCtx], block_size: int
+ self, cipher_ctx_constructor: Callable[[], CipherCtxDecrypt], block_size: int
) -> None:
self.cipher_ctx_constructor = cipher_ctx_constructor
self.block_size = block_size
diff --git a/core/src/trezor/crypto/__init__.py b/core/src/trezor/crypto/__init__.py
index 2c9951f5..79e2bab0 100644
--- a/core/src/trezor/crypto/__init__.py
+++ b/core/src/trezor/crypto/__init__.py
@@ -10,7 +10,7 @@ from trezorcrypto import ( # noqa: F401
)
try:
- from trezorcrypto import aesgcm # noqa: F401
+ from trezorcrypto import aesgcm_decrypt, aesgcm_encrypt # noqa: F401
except Exception:
pass
diff --git a/core/src/trezor/wire/thp/crypto.py b/core/src/trezor/wire/thp/crypto.py
index 039ca594..532d5e77 100644
--- a/core/src/trezor/wire/thp/crypto.py
+++ b/core/src/trezor/wire/thp/crypto.py
@@ -1,6 +1,6 @@
import ustruct
from micropython import const
-from trezorcrypto import aesgcm, bip32, curve25519, hmac
+from trezorcrypto import aesgcm_decrypt, aesgcm_encrypt, bip32, curve25519, hmac
from typing import TYPE_CHECKING
from storage import device
@@ -30,7 +30,7 @@ def enc(buffer: AnyBuffer, key: bytes, nonce: int, auth_data: bytes = b"") -> by
if __debug__ and _TRACE:
log.debug(__name__, "enc (key: %s, nonce: %d)", hexlify_if_bytes(key), nonce)
iv = _get_iv_from_nonce(nonce)
- aes_ctx = aesgcm(key, iv)
+ aes_ctx = aesgcm_encrypt(key, iv)
aes_ctx.auth(auth_data)
aes_ctx.encrypt_in_place(buffer)
return aes_ctx.finish()
@@ -50,7 +50,7 @@ def dec(
iv = _get_iv_from_nonce(nonce)
if __debug__ and _TRACE:
log.debug(__name__, "dec (key: %s, nonce: %d)", hexlify_if_bytes(key), nonce)
- aes_ctx = aesgcm(key, iv)
+ aes_ctx = aesgcm_decrypt(key, iv)
aes_ctx.auth(auth_data)
aes_ctx.decrypt_in_place(buffer)
try:
@@ -106,7 +106,7 @@ class Handshake:
trezor_masked_static_public_key = curve25519.multiply(
mask, trezor_static_public_key
)
- aes_ctx = aesgcm(self.k, IV_1)
+ aes_ctx = aesgcm_encrypt(self.k, IV_1)
encrypted_trezor_static_public_key = aes_ctx.encrypt(
trezor_masked_static_public_key
)
@@ -129,7 +129,7 @@ class Handshake:
trezor_static_private_key, host_ephemeral_public_key
)
self.ck, self.k = _hkdf(self.ck, curve25519.multiply(mask, point))
- aes_ctx = aesgcm(self.k, IV_1)
+ aes_ctx = aesgcm_encrypt(self.k, IV_1)
aes_ctx.auth(self.h)
tag = aes_ctx.finish()
self.h = _hash_of_two(self.h, tag)
@@ -141,7 +141,7 @@ class Handshake:
encrypted_payload: AnyBuffer,
) -> None:
- aes_ctx = aesgcm(self.k, IV_2)
+ aes_ctx = aesgcm_decrypt(self.k, IV_2)
# The new value of hash `h` MUST be computed before the `encrypted_host_static_public_key` is decrypted.
# However, decryption of `encrypted_host_static_public_key` MUST use the previous value of `h` for
@@ -171,7 +171,7 @@ class Handshake:
self.trezor_ephemeral_private_key, host_static_public_key
),
)
- aes_ctx = aesgcm(self.k, IV_1)
+ aes_ctx = aesgcm_decrypt(self.k, IV_1)
aes_ctx.auth(self.h)
self.h = _hash_of_two(self.h, memoryview(encrypted_payload))
aes_ctx.decrypt_in_place(memoryview(encrypted_payload)[:-16])
@@ -194,7 +194,7 @@ class Handshake:
)
def get_handshake_completion_response(self, trezor_state: bytes) -> bytes:
- aes_ctx = aesgcm(self.key_send, IV_1)
+ aes_ctx = aesgcm_encrypt(self.key_send, IV_1)
encrypted_trezor_state = aes_ctx.encrypt(trezor_state)
tag = aes_ctx.finish()
return encrypted_trezor_state + tag
diff --git a/core/tests/test_trezor.crypto.aesgcm.py b/core/tests/test_trezor.crypto.aesgcm.py
index 44775e79..68eea549 100644
--- a/core/tests/test_trezor.crypto.aesgcm.py
+++ b/core/tests/test_trezor.crypto.aesgcm.py
@@ -1,7 +1,7 @@
# flake8: noqa: F403,F405
from common import * # isort:skip
-from trezor.crypto import aesgcm
+from trezor.crypto import aesgcm_decrypt, aesgcm_encrypt
class TestCryptoAes(unittest.TestCase):
@@ -49,18 +49,18 @@ class TestCryptoAes(unittest.TestCase):
key, iv, pt, aad, ct, tag = map(unhexlify, vector)
# Test encryption.
- ctx = aesgcm(key, iv)
+ ctx = aesgcm_encrypt(key, iv)
if aad:
ctx.auth(aad)
self.assertEqual(ctx.encrypt(pt), ct)
self.assertEqual(ctx.finish(), tag)
# Test decryption.
- ctx.reset(iv)
+ ctx = aesgcm_decrypt(key, iv)
if aad:
ctx.auth(aad)
self.assertEqual(ctx.decrypt(ct), pt)
- self.assertEqual(ctx.finish(tag), tag)
+ self.assertIsNone(ctx.finish(tag))
def test_gcm_in_place(self):
for vector in self.vectors:
@@ -68,7 +68,7 @@ class TestCryptoAes(unittest.TestCase):
buffer = bytearray(pt)
# Test encryption.
- ctx = aesgcm(key, iv)
+ ctx = aesgcm_encrypt(key, iv)
if aad:
ctx.auth(aad)
returned = ctx.encrypt_in_place(buffer)
@@ -77,13 +77,13 @@ class TestCryptoAes(unittest.TestCase):
self.assertEqual(ctx.finish(), tag)
# Test decryption.
- ctx.reset(iv)
+ ctx = aesgcm_decrypt(key, iv)
if aad:
ctx.auth(aad)
returned = ctx.decrypt_in_place(buffer)
self.assertEqual(buffer, pt)
self.assertEqual(returned, len(buffer))
- self.assertEqual(ctx.finish(tag), tag)
+ self.assertIsNone(ctx.finish(tag))
def test_gcm_chunks(self):
for vector in self.vectors:
@@ -92,20 +92,20 @@ class TestCryptoAes(unittest.TestCase):
chunk1 = len(pt) // 3
# Decrypt by chunks and add authenticated data by chunks.
- ctx = aesgcm(key, iv)
+ ctx = aesgcm_decrypt(key, iv)
self.assertEqual(ctx.decrypt(ct[:chunk1]), pt[:chunk1])
ctx.auth(aad[:17])
self.assertEqual(ctx.decrypt(ct[chunk1:]), pt[chunk1:])
ctx.auth(aad[17:])
- self.assertEqual(ctx.finish(tag), tag)
+ self.assertIsNone(ctx.finish(tag))
# Encrypt by chunks and add authenticated data by chunks.
- ctx.reset(iv)
+ ctx = aesgcm_encrypt(key, iv)
ctx.auth(aad[:7])
self.assertEqual(ctx.encrypt(pt[:chunk1]), ct[:chunk1])
ctx.auth(aad[7:])
self.assertEqual(ctx.encrypt(pt[chunk1:]), ct[chunk1:])
- self.assertEqual(ctx.finish(tag), tag)
+ self.assertEqual(ctx.finish(), tag)
def test_gcm_chunks_in_place(self):
for vector in self.vectors:
@@ -115,7 +115,7 @@ class TestCryptoAes(unittest.TestCase):
chunk2_length = len(pt) - chunk1_length
# Decrypt by chunks and add authenticated data by chunks.
- ctx = aesgcm(key, iv)
+ ctx = aesgcm_decrypt(key, iv)
returned = ctx.decrypt_in_place(memoryview(buffer)[:chunk1_length])
self.assertEqual(returned, chunk1_length)
ctx.auth(aad[:17])
@@ -123,10 +123,10 @@ class TestCryptoAes(unittest.TestCase):
ctx.auth(aad[17:])
self.assertEqual(returned, chunk2_length)
self.assertEqual(buffer, pt)
- self.assertEqual(ctx.finish(tag), tag)
+ self.assertIsNone(ctx.finish(tag))
# Encrypt by chunks and add authenticated data by chunks.
- ctx.reset(iv)
+ ctx = aesgcm_encrypt(key, iv)
ctx.auth(aad[:7])
returned = ctx.encrypt_in_place(memoryview(buffer)[:chunk1_length])
self.assertEqual(returned, chunk1_length)
@@ -134,23 +134,7 @@ class TestCryptoAes(unittest.TestCase):
returned = ctx.encrypt_in_place(memoryview(buffer)[chunk1_length:])
self.assertEqual(returned, chunk2_length)
self.assertEqual(buffer, ct)
- self.assertEqual(ctx.finish(tag), tag)
-
- def test_gcm_missing_expected_tag(self):
- for vector in self.vectors:
- key, iv, pt, aad, ct, _ = map(unhexlify, vector)
-
- ctx = aesgcm(key, iv)
- if aad:
- ctx.auth(aad)
- self.assertEqual(ctx.decrypt(ct), pt)
-
- # Try finishing the decryption with expected_tag missing
- with self.assertRaises(RuntimeError) as e:
- ctx.finish()
- self.assertEqual(
- e.value.value, "Argument `expected_tag` is required when decrypting."
- )
+ self.assertEqual(ctx.finish(), tag)
def test_gcm_invalid_tag_len(self):
for vector in self.vectors:
@@ -164,12 +148,12 @@ class TestCryptoAes(unittest.TestCase):
]
for tag in invalid_tags:
- ctx = aesgcm(key, iv)
+ ctx = aesgcm_decrypt(key, iv)
if aad:
ctx.auth(aad)
self.assertEqual(ctx.decrypt(ct), pt)
- # Try finishing the decryption with invalid-length tag
+ # Try finishing the decryption with an invalid-length tag
with self.assertRaises(ValueError) as e:
ctx.finish(tag)
self.assertEqual(
@@ -181,12 +165,12 @@ class TestCryptoAes(unittest.TestCase):
invalid_tag = b"\xab" * 16
for vector in self.vectors:
key, iv, pt, aad, ct, _ = map(unhexlify, vector)
- ctx = aesgcm(key, iv)
+ ctx = aesgcm_decrypt(key, iv)
if aad:
ctx.auth(aad)
self.assertEqual(ctx.decrypt(ct), pt)
- # Try finishing the decryption with invalid tag
+ # Try finishing the decryption with an invalid tag
with self.assertRaises(RuntimeError) as e:
ctx.finish(invalid_tag)
self.assertEqual(
diff --git a/core/tests/test_trezor.wire.thp.crypto.py b/core/tests/test_trezor.wire.thp.crypto.py
index a59e77ad..fc87fc81 100644
--- a/core/tests/test_trezor.wire.thp.crypto.py
+++ b/core/tests/test_trezor.wire.thp.crypto.py
@@ -1,6 +1,6 @@
# flake8: noqa: F403,F405
from common import * # isort:skip
-from trezorcrypto import aesgcm, curve25519
+from trezorcrypto import aesgcm_encrypt, curve25519
import storage
@@ -118,7 +118,7 @@ class TestTrezorHostProtocolCrypto(unittest.TestCase):
host_static_private_key = curve25519.generate_secret()
host_static_public_key = curve25519.publickey(host_static_private_key)
- aes_ctx = aesgcm(handshake.k, IV_2)
+ aes_ctx = aesgcm_encrypt(handshake.k, IV_2)
aes_ctx.auth(handshake.h)
encrypted_host_static_public_key = bytearray(
aes_ctx.encrypt(host_static_public_key) + aes_ctx.finish()
@@ -136,7 +136,7 @@ class TestTrezorHostProtocolCrypto(unittest.TestCase):
handshake.trezor_ephemeral_private_key, host_static_public_key
),
)
- aes_ctx = aesgcm(temp_k, IV_1)
+ aes_ctx = aesgcm_encrypt(temp_k, IV_1)
aes_ctx.encrypt_in_place(protomsg)
aes_ctx.auth(temp_h)
tag = aes_ctx.finish()
Why this scored 12/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.