# Reticulum License # # Copyright (c) 2016-2025 Mark Qvist # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # - The Software shall not be used in any kind of system which includes amongst # its functions the ability to purposefully do harm to human beings. # # - The Software shall not be used, directly or indirectly, in the creation of # an artificial intelligence, machine learning or language model training # dataset, including but not limited to any use that contributes to the # training or development of such a model or algorithm. # # - The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. import os import time from RNS.Cryptography import HMAC from RNS.Cryptography import PKCS7 from RNS.Cryptography import AES from RNS.Cryptography.AES import AES_128_CBC from RNS.Cryptography.AES import AES_256_CBC # import RNS # TODO: Remove class Token(): """ This class provides a slightly modified implementation of the Fernet spec found at: https://github.com/fernet/spec/blob/master/Spec.md According to the spec, a Fernet token includes a one byte VERSION and eight byte TIMESTAMP field at the start of each token. These fields are not relevant to Reticulum. They are therefore stripped from this implementation, since they incur overhead and leak initiator metadata. """ TOKEN_OVERHEAD = 48 # Bytes @staticmethod def generate_key(mode=AES_128_CBC): if mode == AES_128_CBC: return os.urandom(32) elif mode == AES_256_CBC: return os.urandom(64) else: raise TypeError(f"Invalid token mode: {mode}") def __init__(self, key=None, mode=AES): if key == None: raise ValueError("Token key cannot be None") if mode == AES: if len(key) == 32: self.mode = AES_128_CBC self._signing_key = key[:16] self._encryption_key = key[16:] #################################################################### self.mode_legacy = AES_128_CBC # TODO: Remove after migration self._signing_key_128 = key[:16] # TODO: Remove after migration self._encryption_key_128 = key[16:] # TODO: Remove after migration #################################################################### elif len(key) == 64: self.mode = AES_256_CBC self._signing_key = key[:32] self._encryption_key = key[32:] #################################################################### self.mode_legacy = AES_128_CBC # TODO: Remove after migration self._signing_key_128 = key[:16] # TODO: Remove after migration self._encryption_key_128 = key[16:32] # TODO: Remove after migration #################################################################### else: raise ValueError("Token key must be 128 or 256 bits, not "+str(len(key)*8)) else: raise TypeError(f"Invalid token mode: {mode}") def verify_hmac(self, token): if len(token) <= 32: raise ValueError("Cannot verify HMAC on token of only "+str(len(token))+" bytes") else: received_hmac = token[-32:] expected_hmac = HMAC.new(self._signing_key, token[:-32]).digest() expected_hmac_128 = HMAC.new(self._signing_key_128, token[:-32]).digest() # TODO: Remove after migration # TODO: Reset after migration # if received_hmac == expected_hmac: return True if received_hmac == expected_hmac or received_hmac == expected_hmac_128: return True else: return False def encrypt(self, data = None): if not isinstance(data, bytes): raise TypeError("Token plaintext input must be bytes") iv = os.urandom(16) ciphertext = self.mode.encrypt( plaintext = PKCS7.pad(data), key = self._encryption_key, iv = iv) signed_parts = iv+ciphertext return signed_parts + HMAC.new(self._signing_key, signed_parts).digest() def decrypt(self, token = None): if not isinstance(token, bytes): raise TypeError("Token must be bytes") if not self.verify_hmac(token): raise ValueError("Token HMAC was invalid") iv = token[:16] ciphertext = token[16:-32] try: try: # RNS.log(f"Trying decryption with {self.mode}", RNS.LOG_DEBUG) # TODO: Remove plaintext = PKCS7.unpad( self.mode.decrypt( ciphertext = ciphertext, key = self._encryption_key, iv = iv)) # RNS.log(f"Decrypted packet with {self.mode}", RNS.LOG_DEBUG) # TODO: Remove return plaintext # TODO: Remove after migration ############################ except Exception as e: # RNS.log(f"{self.mode} decryption failed", RNS.LOG_DEBUG) # TODO: Remove # RNS.log(f"Trying decryption with {self.mode_legacy}", RNS.LOG_DEBUG) # TODO: Remove plaintext = PKCS7.unpad( self.mode_legacy.decrypt( ciphertext = ciphertext, key = self._encryption_key_128, iv = iv)) # RNS.log(f"Decrypted packet with {self.mode_legacy}", RNS.LOG_DEBUG) # TODO: Remove return plaintext ########################################################### except Exception as e: RNS.trace_exception(e) # TODO: Remove after migration raise ValueError("Could not decrypt token")