mirror of
https://github.com/aljazceru/python-teos.git
synced 2025-12-17 22:24:23 +01:00
d7c89ddc91
The decryption for the `EncryptedBlob` using AES-GCM-128 (the only cipher available atm) raises an `InvalidTag` exception. This was not properly captured by the watcher making it crash. This behavior was already discovered during the `EncryptedBlob` unit testing and left to be fixed in the `Watcher` unit testing. However, making the EncryptedBlob raise such an exception may not be a good practice, since other ciphers may run into different exceptions. Therefore, the `EncryptedBlob` has been modified to return None upon facing a decryption issue, the `BlockProcessor` will detect that and return a None justice_txm and justice_txid. Upon receiving a None `justice_txid` the `Watcher` will delete the appointment without notifiying the `Responder`.
47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
from hashlib import sha256
|
|
from binascii import unhexlify, hexlify
|
|
from cryptography.exceptions import InvalidTag
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
from pisa.logger import Logger
|
|
|
|
logger = Logger("Watcher")
|
|
|
|
|
|
# FIXME: EncryptedBlob is assuming AES-128-GCM. A cipher field should be part of the object and the decryption should be
|
|
# performed depending on the cipher.
|
|
class EncryptedBlob:
|
|
def __init__(self, data):
|
|
self.data = data
|
|
|
|
def __eq__(self, other):
|
|
return isinstance(other, EncryptedBlob) and self.data == other.data
|
|
|
|
def decrypt(self, key):
|
|
# master_key = H(tx_id | tx_id)
|
|
key = unhexlify(key)
|
|
master_key = sha256(key + key).digest()
|
|
|
|
# The 16 MSB of the master key will serve as the AES GCM 128 secret key. The 16 LSB will serve as the IV.
|
|
sk = master_key[:16]
|
|
nonce = master_key[16:]
|
|
|
|
logger.info("Creating new blob.",
|
|
master_key=hexlify(master_key).decode(),
|
|
sk=hexlify(sk).decode(),
|
|
nonce=hexlify(sk).decode(),
|
|
encrypted_blob=self.data)
|
|
|
|
# Decrypt
|
|
aesgcm = AESGCM(sk)
|
|
data = unhexlify(self.data.encode())
|
|
|
|
try:
|
|
raw_tx = aesgcm.decrypt(nonce=nonce, data=data, associated_data=None)
|
|
hex_raw_tx = hexlify(raw_tx).decode('utf8')
|
|
|
|
except InvalidTag:
|
|
hex_raw_tx = None
|
|
|
|
return hex_raw_tx
|