Reticulum/RNS/Identity.py

# 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 math
import os
import RNS
import time
import atexit
import hashlib
import threading

from .vendor import umsgpack as umsgpack

from RNS.Cryptography import X25519PrivateKey, X25519PublicKey, Ed25519PrivateKey, Ed25519PublicKey
from RNS.Cryptography import Token


class Identity:
    """
    This class is used to manage identities in Reticulum. It provides methods
    for encryption, decryption, signatures and verification, and is the basis
    for all encrypted communication over Reticulum networks.

    :param create_keys: Specifies whether new encryption and signing keys should be generated.
    """

    CURVE = "Curve25519"
    """
    The curve used for Elliptic Curve DH key exchanges
    """

    KEYSIZE     = 256*2
    """
    X.25519 key size in bits. A complete key is the concatenation of a 256 bit encryption key, and a 256 bit signing key.
    """

    RATCHETSIZE = 256
    """
    X.25519 ratchet key size in bits.
    """

    RATCHET_EXPIRY = 60*60*24*30
    """
    The expiry time for received ratchets in seconds, defaults to 30 days. Reticulum will always use the most recently
    announced ratchet, and remember it for up to ``RATCHET_EXPIRY`` since receiving it, after which it will be discarded.
    If a newer ratchet is announced in the meantime, it will be replace the already known ratchet.
    """

    # Non-configurable constants
    TOKEN_OVERHEAD            = RNS.Cryptography.Token.TOKEN_OVERHEAD
    AES128_BLOCKSIZE          = 16          # In bytes
    HASHLENGTH                = 256         # In bits
    SIGLENGTH                 = KEYSIZE     # In bits

    NAME_HASH_LENGTH          = 80
    TRUNCATED_HASHLENGTH      = RNS.Reticulum.TRUNCATED_HASHLENGTH
    """
    Constant specifying the truncated hash length (in bits) used by Reticulum
    for addressable hashes and other purposes. Non-configurable.
    """

    DERIVED_KEY_LENGTH        = 512//8
    DERIVED_KEY_LENGTH_LEGACY = 256//8

    # Storage
    known_destinations = {}
    known_ratchets = {}

    ratchet_persist_lock = threading.Lock()
    known_destinations_lock = threading.Lock()

    @staticmethod
    def remember(packet_hash, destination_hash, public_key, app_data = None):
        if len(public_key) != Identity.KEYSIZE//8:
            raise TypeError("Can't remember "+RNS.prettyhexrep(destination_hash)+", the public key size of "+str(len(public_key))+" is not valid.", RNS.LOG_ERROR)
        else:
            with Identity.known_destinations_lock:
                if not destination_hash in Identity.known_destinations:
                    Identity.known_destinations[destination_hash] = [time.time(), packet_hash, public_key, app_data, 0]
                else:
                    entry = Identity.known_destinations[destination_hash]
                    entry[0] = time.time()
                    entry[1] = packet_hash
                    entry[2] = public_key
                    entry[3] = app_data

    @staticmethod
    def recall(target_hash, from_identity_hash=False, _no_use=False):
        """
        Recall identity for a destination or identity hash. By default, this function
        will return the identity associated with a given *destination* hash. As an
        example, if you know the ``lxmf.delivery`` destination hash of an endpoint,
        this function will return the associated underlying identity. You can also
        search for an identity from a known *identity hash*, by setting the
        ``from_identity_hash`` argument.

        :param target_hash: Destination or identity hash as *bytes*.
        :param from_identity_hash: Whether to search based on identity hash instead of destination hash as *bool*.
        :returns: An :ref:`RNS.Identity<api-identity>` instance that can be used to create an outgoing :ref:`RNS.Destination<api-destination>`, or *None* if the destination is unknown.
        """
        if from_identity_hash:
            for destination_hash in Identity.known_destinations:
                if target_hash == Identity.truncated_hash(Identity.known_destinations[destination_hash][2]):
                    if not _no_use: RNS.Reticulum.get_instance()._used_destination_data(destination_hash)
                    identity_data = Identity.known_destinations[destination_hash]
                    identity = Identity(create_keys=False)
                    identity.load_public_key(identity_data[2])
                    identity.app_data = identity_data[3]
                    return identity

            return None

        else:
            if target_hash in Identity.known_destinations:
                if not _no_use: RNS.Reticulum.get_instance()._used_destination_data(target_hash)
                identity_data = Identity.known_destinations[target_hash]
                identity = Identity(create_keys=False)
                identity.load_public_key(identity_data[2])
                identity.app_data = identity_data[3]
                return identity
            else:
                for registered_destination in RNS.Transport.destinations:
                    if target_hash == registered_destination.hash:
                        identity = Identity(create_keys=False)
                        identity.load_public_key(registered_destination.identity.get_public_key())
                        identity.app_data = None
                        return identity

                return None

    @staticmethod
    def recall_app_data(destination_hash, _no_use=False):
        """
        Recall last heard app_data for a destination hash.

        :param destination_hash: Destination hash as *bytes*.
        :returns: *Bytes* containing app_data, or *None* if the destination is unknown.
        """
        if destination_hash in Identity.known_destinations:
            if not _no_use: RNS.Reticulum.get_instance()._used_destination_data(destination_hash)
            app_data = Identity.known_destinations[destination_hash][3]
            return app_data
        
        else: return None

    @staticmethod
    def save_known_destinations(background=False, recombine=True):
        # TODO: Improve the storage method so we don't have to
        # deserialize and serialize the entire table on every
        # save, but the only changes. It might be possible to
        # simply overwrite on exit now that every local client
        # disconnect triggers a data persist.
        
        try:
            if hasattr(Identity, "saving_known_destinations"):
                wait_interval = 0.2
                wait_timeout = 5
                wait_start = time.time()
                while Identity.saving_known_destinations:
                    time.sleep(wait_interval)
                    if time.time() > wait_start+wait_timeout:
                        RNS.log("Could not save known destinations to storage, waiting for previous save operation timed out.", RNS.LOG_ERROR)
                        return False

            Identity.saving_known_destinations = True
            save_start = time.time()

            if recombine:
                storage_known_destinations = {}
                if os.path.isfile(RNS.Reticulum.storagepath+"/known_destinations"):
                    try:
                        with open(RNS.Reticulum.storagepath+"/known_destinations","rb") as file:
                            storage_known_destinations = umsgpack.load(file)
     
                    except: pass

                try:
                    for destination_hash in storage_known_destinations:
                        if not destination_hash in Identity.known_destinations:
                            with Identity.known_destinations_lock:
                                Identity.known_destinations[destination_hash] = storage_known_destinations[destination_hash]
                
                except Exception as e:
                    RNS.log("Skipped recombining known destinations from disk, since an error occurred: "+str(e), RNS.LOG_WARNING)

            RNS.log("Saving "+str(len(Identity.known_destinations))+" known destinations to storage...", RNS.LOG_VERBOSE)
            with open(RNS.Reticulum.storagepath+"/known_destinations","wb") as file:
                umsgpack.dump(Identity.known_destinations.copy(), file)

            save_time = time.time() - save_start
            if save_time < 1: time_str = str(round(save_time*1000,2))+"ms"
            else:             time_str = str(round(save_time,2))+"s"

            RNS.log("Saved known destinations to storage in "+time_str, RNS.LOG_VERBOSE)

        except Exception as e:
            RNS.log("Error while saving known destinations to disk, the contained exception was: "+str(e), RNS.LOG_ERROR)
            RNS.trace_exception(e)

        Identity.saving_known_destinations = False

    @staticmethod
    def load_known_destinations():
        if os.path.isfile(RNS.Reticulum.storagepath+"/known_destinations"):
            st = time.time()
            try:
                with open(RNS.Reticulum.storagepath+"/known_destinations","rb") as file:
                    loaded_known_destinations = umsgpack.load(file)

                Identity.known_destinations = {}
                for known_destination in loaded_known_destinations:
                    if len(known_destination) == RNS.Reticulum.TRUNCATED_HASHLENGTH//8:
                        if len(loaded_known_destinations[known_destination]) < 5:
                            e = loaded_known_destinations[known_destination]
                            loaded_known_destinations[known_destination] = [e[0], e[1], e[2], e[3], 0]

                        with Identity.known_destinations_lock:
                            Identity.known_destinations[known_destination] = loaded_known_destinations[known_destination]

                RNS.log(f"Loaded {len(Identity.known_destinations)} known destination from storage in {RNS.prettyshorttime(time.time()-st)}", RNS.LOG_VERBOSE)

            except Exception as e:
                RNS.log("Error loading known destinations from disk, file will be recreated on exit", RNS.LOG_ERROR)
                RNS.trace_exception(e)
        else:
            RNS.log("Destinations file does not exist, no known destinations loaded", RNS.LOG_VERBOSE)

    @staticmethod
    def _used_destination_data(destination_hash):
        with Identity.known_destinations_lock:
            if destination_hash in Identity.known_destinations:
                if not Identity.known_destinations[destination_hash][4] < 0:
                    Identity.known_destinations[destination_hash][4] = time.time()
                    return True

        return False

    @staticmethod
    def _retain_destination_data(destination_hash):
        with Identity.known_destinations_lock:
            if destination_hash in Identity.known_destinations:
                Identity.known_destinations[destination_hash][4] = -1
                return True

        return False

    @staticmethod
    def _unretain_destination_data(destination_hash):
        with Identity.known_destinations_lock:
            if destination_hash in Identity.known_destinations:
                Identity.known_destinations[destination_hash][4] = time.time()
                return True

        return False
    
    @staticmethod
    def clean_known_destinations():
        now        = time.time()
        st         = now
        total      = len(Identity.known_destinations)
        stale      = []
        no_path    = 0
        retained   = 0
        never_used = 0
        for destination_hash in Identity.known_destinations:
            try:
                if RNS.Transport.has_path(destination_hash): has_path = True
                else:
                    has_path = False
                    no_path += 1

                with Identity.known_destinations_lock:
                    if destination_hash in Identity.known_destinations:
                        last_announce =  Identity.known_destinations[destination_hash][0]
                        last_use = 0
                        was_used = False
                        is_retained = False

                        if Identity.known_destinations[destination_hash][4] > 0:
                            was_used = True
                            last_use = Identity.known_destinations[destination_hash][4]

                        elif Identity.known_destinations[destination_hash][4] == 0:
                            was_used = False
                            never_used += 1

                        elif Identity.known_destinations[destination_hash][4] == -1:
                            is_retained = True
                            retained += 1

                        unused_for = time.time() - Identity.known_destinations[destination_hash][4]

                        if not is_retained and not has_path:
                            if not was_used and now - last_announce > RNS.Transport.UNUSED_DESTINATION_LINGER: stale.append(destination_hash)
                            elif unused_for > RNS.Transport.DESTINATION_TIMEOUT*1.25:                          stale.append(destination_hash)

            except Exception as e: RNS.log(f"Faulty entry for {RNS.prettyhexrep(destination_hash)} while cleaning known destinations: {e}", RNS.LOG_DEBUG)

        removed = 0
        for destination_hash in stale:
            with Identity.known_destinations_lock:
                if destination_hash in Identity.known_destinations:
                    Identity.known_destinations.pop(destination_hash)
                    removed += 1

        # RNS.log(f"Total destinations: {total}, stale: {len(stale)}, removed: {removed}, no path: {no_path}, never used: {never_used}, with path: {total-no_path}, used: {total-never_used}, retained: {retained}. Completed in {RNS.prettyshorttime(time.time()-st)}", RNS.LOG_WARNING) # TODO: Remove
        if not RNS.Transport.owner.is_connected_to_shared_instance: Identity.save_known_destinations(recombine=False)

    @staticmethod
    def full_hash(data):
        """
        Get a SHA-256 hash of passed data.

        :param data: Data to be hashed as *bytes*.
        :returns: SHA-256 hash as *bytes*.
        """
        return RNS.Cryptography.sha256(data)

    @staticmethod
    def truncated_hash(data):
        """
        Get a truncated SHA-256 hash of passed data.

        :param data: Data to be hashed as *bytes*.
        :returns: Truncated SHA-256 hash as *bytes*.
        """
        return Identity.full_hash(data)[:(Identity.TRUNCATED_HASHLENGTH//8)]

    @staticmethod
    def get_random_hash():
        """
        Get a random SHA-256 hash.

        :param data: Data to be hashed as *bytes*.
        :returns: Truncated SHA-256 hash of random data as *bytes*.
        """
        return Identity.truncated_hash(os.urandom(Identity.TRUNCATED_HASHLENGTH//8))

    @staticmethod
    def current_ratchet_id(destination_hash):
        """
        Get the ID of the currently used ratchet key for a given destination hash

        :param destination_hash: A destination hash as *bytes*.
        :returns: A ratchet ID as *bytes* or *None*.
        """
        ratchet = Identity.get_ratchet(destination_hash)
        if ratchet == None:
            return None
        else:
            return Identity._get_ratchet_id(ratchet)

    @staticmethod
    def _get_ratchet_id(ratchet_pub_bytes):
        return Identity.full_hash(ratchet_pub_bytes)[:Identity.NAME_HASH_LENGTH//8]

    @staticmethod
    def _ratchet_public_bytes(ratchet):
        return X25519PrivateKey.from_private_bytes(ratchet).public_key().public_bytes()

    @staticmethod
    def _generate_ratchet():
        ratchet_prv = X25519PrivateKey.generate()
        ratchet_pub = ratchet_prv.public_key()
        return ratchet_prv.private_bytes()

    @staticmethod
    def _remember_ratchet(destination_hash, ratchet):
        try:
            if destination_hash in Identity.known_ratchets and Identity.known_ratchets[destination_hash] == ratchet:
                ratchet_exists = True
            else:
                ratchet_exists = False

            if not ratchet_exists:
                RNS.log(f"Remembering ratchet {RNS.prettyhexrep(Identity._get_ratchet_id(ratchet))} for {RNS.prettyhexrep(destination_hash)}", RNS.LOG_EXTREME)
                Identity.known_ratchets[destination_hash] = ratchet
                if not RNS.Transport.owner.is_connected_to_shared_instance:
                    def persist_job():
                        with Identity.ratchet_persist_lock:
                            hexhash = RNS.hexrep(destination_hash, delimit=False)
                            ratchet_data = {"ratchet": ratchet, "received": time.time()}

                            ratchetdir = RNS.Reticulum.storagepath+"/ratchets"
                            
                            if not os.path.isdir(ratchetdir):
                                os.makedirs(ratchetdir)

                            outpath   = f"{ratchetdir}/{hexhash}.out"
                            finalpath = f"{ratchetdir}/{hexhash}"
                            with open(outpath, "wb") as ratchet_file:
                                ratchet_file.write(umsgpack.packb(ratchet_data))
                            os.replace(outpath, finalpath)

                    
                    threading.Thread(target=persist_job, daemon=True).start()

        except Exception as e:
            RNS.log(f"Could not persist ratchet for {RNS.prettyhexrep(destination_hash)} to storage.", RNS.LOG_ERROR)
            RNS.log(f"The contained exception was: {e}")
            RNS.trace_exception(e)

    @staticmethod
    def _clean_ratchets():
        RNS.log("Cleaning ratchets...", RNS.LOG_DEBUG)
        try:
            count = 0
            removed = 0
            not_known = 0
            now = time.time()
            ratchetdir = RNS.Reticulum.storagepath+"/ratchets"
            if os.path.isdir(ratchetdir):
                for filename in os.listdir(ratchetdir):
                    count += 1
                    try:
                        expired = False
                        corrupted = False
                        with open(f"{ratchetdir}/{filename}", "rb") as rf:
                            try:
                                ratchet_data = umsgpack.unpackb(rf.read())
                                if now > ratchet_data["received"]+Identity.RATCHET_EXPIRY: expired = True

                            except Exception as e:
                                RNS.log(f"Corrupted ratchet data while reading {ratchetdir}/{filename}, removing file", RNS.LOG_ERROR)
                                corrupted = True

                        destination_hash = bytes.fromhex(filename)
                        if not destination_hash in RNS.Identity.known_destinations: unknown = True; not_known += 1
                        else:                                                       unknown = False

                        if expired or corrupted or unknown:
                            os.unlink(f"{ratchetdir}/{filename}")
                            removed += 1

                    except Exception as e:
                        RNS.log(f"An error occurred while cleaning ratchets, in the processing of {ratchetdir}/{filename}.", RNS.LOG_ERROR)
                        RNS.log(f"The contained exception was: {e}", RNS.LOG_ERROR)

        except Exception as e: RNS.log(f"An error occurred while cleaning ratchets. The contained exception was: {e}", RNS.LOG_ERROR)
        RNS.log(f"Processed {count} ratchets in {RNS.prettytime(time.time()-now)}, not in use {not_known}, removed {removed}", RNS.LOG_DEBUG)

    @staticmethod
    def get_ratchet(destination_hash):
        if not destination_hash in Identity.known_ratchets:
            ratchetdir = RNS.Reticulum.storagepath+"/ratchets"
            hexhash = RNS.hexrep(destination_hash, delimit=False)
            ratchet_path = f"{ratchetdir}/{hexhash}"
            if os.path.isfile(ratchet_path):
                try:
                    with open(ratchet_path, "rb") as ratchet_file:
                        ratchet_data = umsgpack.unpackb(ratchet_file.read())
                        if time.time() < ratchet_data["received"]+Identity.RATCHET_EXPIRY and len(ratchet_data["ratchet"]) == Identity.RATCHETSIZE//8:
                            Identity.known_ratchets[destination_hash] = ratchet_data["ratchet"]
                        else:
                            return None
                
                except Exception as e:
                    RNS.log(f"An error occurred while loading ratchet data for {RNS.prettyhexrep(destination_hash)} from storage.", RNS.LOG_ERROR)
                    RNS.log(f"The contained exception was: {e}", RNS.LOG_ERROR)
                    return None

        if destination_hash in Identity.known_ratchets:
            return Identity.known_ratchets[destination_hash]
        else:
            RNS.log(f"Could not load ratchet for {RNS.prettyhexrep(destination_hash)}", RNS.LOG_DEBUG)
            return None

    @staticmethod
    def validate_announce(packet, only_validate_signature=False):
        try:
            if packet.packet_type == RNS.Packet.ANNOUNCE:
                keysize       = Identity.KEYSIZE//8
                ratchetsize   = Identity.RATCHETSIZE//8
                name_hash_len = Identity.NAME_HASH_LENGTH//8
                sig_len       = Identity.SIGLENGTH//8
                destination_hash = packet.destination_hash

                # Get public key bytes from announce
                public_key = packet.data[:keysize]

                # If the packet context flag is set,
                # this announce contains a new ratchet
                if packet.context_flag == RNS.Packet.FLAG_SET:
                    name_hash   = packet.data[keysize:keysize+name_hash_len ]
                    random_hash = packet.data[keysize+name_hash_len:keysize+name_hash_len+10]
                    ratchet     = packet.data[keysize+name_hash_len+10:keysize+name_hash_len+10+ratchetsize]
                    signature   = packet.data[keysize+name_hash_len+10+ratchetsize:keysize+name_hash_len+10+ratchetsize+sig_len]
                    app_data    = b""
                    if len(packet.data) > keysize+name_hash_len+10+sig_len+ratchetsize:
                        app_data = packet.data[keysize+name_hash_len+10+sig_len+ratchetsize:]

                # If the packet context flag is not set,
                # this announce does not contain a ratchet
                else:
                    ratchet     = b""
                    name_hash   = packet.data[keysize:keysize+name_hash_len]
                    random_hash = packet.data[keysize+name_hash_len:keysize+name_hash_len+10]
                    signature   = packet.data[keysize+name_hash_len+10:keysize+name_hash_len+10+sig_len]
                    app_data    = b""
                    if len(packet.data) > keysize+name_hash_len+10+sig_len:
                        app_data = packet.data[keysize+name_hash_len+10+sig_len:]

                signed_data = destination_hash+public_key+name_hash+random_hash+ratchet+app_data

                if not len(packet.data) > Identity.KEYSIZE//8+Identity.NAME_HASH_LENGTH//8+10+Identity.SIGLENGTH//8:
                    app_data = None

                announced_identity = Identity(create_keys=False)
                announced_identity.load_public_key(public_key)

                if len(RNS.Transport.blackholed_identities) > 0:
                    if announced_identity.hash in RNS.Transport.blackholed_identities:
                        RNS.log(f"Invalidated and dropped announce from blackholed identity {RNS.prettyhexrep(announced_identity.hash)}", RNS.LOG_EXTREME)
                        return False

                if announced_identity.pub != None and announced_identity.validate(signature, signed_data):
                    if only_validate_signature:
                        del announced_identity
                        return True

                    hash_material = name_hash+announced_identity.hash
                    expected_hash = RNS.Identity.full_hash(hash_material)[:RNS.Reticulum.TRUNCATED_HASHLENGTH//8]

                    if destination_hash == expected_hash:
                        # Check if we already have a public key for this destination
                        # and make sure the public key is not different.
                        if destination_hash in Identity.known_destinations:
                            if public_key != Identity.known_destinations[destination_hash][2]:
                                # In reality, this should never occur, but in the odd case
                                # that someone manages a hash collision, we reject the announce.
                                RNS.log("Received announce with valid signature and destination hash, but announced public key does not match already known public key.", RNS.LOG_CRITICAL)
                                RNS.log("This may indicate an attempt to modify network paths, or a random hash collision. The announce was rejected.", RNS.LOG_CRITICAL)
                                return False

                        RNS.Identity.remember(packet.get_hash(), destination_hash, public_key, app_data)
                        del announced_identity

                        if packet.rssi != None or packet.snr != None:
                            signal_str = " ["
                            if packet.rssi != None:
                                signal_str += "RSSI "+str(packet.rssi)+"dBm"
                                if packet.snr != None:
                                    signal_str += ", "
                            if packet.snr != None:
                                signal_str += "SNR "+str(packet.snr)+"dB"
                            signal_str += "]"
                        else:
                            signal_str = ""

                        if hasattr(packet, "transport_id") and packet.transport_id != None:
                            RNS.log("Valid announce for "+RNS.prettyhexrep(destination_hash)+" "+str(packet.hops)+" hops away, received via "+RNS.prettyhexrep(packet.transport_id)+" on "+str(packet.receiving_interface)+signal_str, RNS.LOG_EXTREME)
                        else:
                            RNS.log("Valid announce for "+RNS.prettyhexrep(destination_hash)+" "+str(packet.hops)+" hops away, received on "+str(packet.receiving_interface)+signal_str, RNS.LOG_EXTREME)

                        if ratchet:
                            Identity._remember_ratchet(destination_hash, ratchet)

                        return True

                    else:
                        RNS.log("Received invalid announce for "+RNS.prettyhexrep(destination_hash)+": Destination mismatch.", RNS.LOG_DEBUG)
                        return False

                else:
                    RNS.log("Received invalid announce for "+RNS.prettyhexrep(destination_hash)+": Invalid signature.", RNS.LOG_DEBUG)
                    del announced_identity
                    return False
        
        except Exception as e:
            RNS.log("Error occurred while validating announce. The contained exception was: "+str(e), RNS.LOG_ERROR)
            return False

    @staticmethod
    def persist_data(background=False):
        if not RNS.Transport.owner.is_connected_to_shared_instance:
            Identity.save_known_destinations(background=background)

    @staticmethod
    def exit_handler():
        Identity.persist_data()


    @staticmethod
    def from_bytes(prv_bytes):
        """
        Create a new :ref:`RNS.Identity<api-identity>` instance from *bytes* of private key.
        Can be used to load previously created and saved identities into Reticulum.

        :param prv_bytes: The *bytes* of private a saved private key. **HAZARD!** Never use this to generate a new key by feeding random data in prv_bytes.
        :returns: A :ref:`RNS.Identity<api-identity>` instance, or *None* if the *bytes* data was invalid.
        """
        identity = Identity(create_keys=False)
        if identity.load_private_key(prv_bytes):
            return identity
        else:
            return None


    @staticmethod
    def from_file(path):
        """
        Create a new :ref:`RNS.Identity<api-identity>` instance from a file.
        Can be used to load previously created and saved identities into Reticulum.

        :param path: The full path to the saved :ref:`RNS.Identity<api-identity>` data
        :returns: A :ref:`RNS.Identity<api-identity>` instance, or *None* if the loaded data was invalid.
        """
        identity = Identity(create_keys=False)
        if identity.load(path):
            return identity
        else:
            return None

    def to_file(self, path):
        """
        Saves the identity to a file. This will write the private key to disk,
        and anyone with access to this file will be able to decrypt all
        communication for the identity. Be very careful with this method.

        :param path: The full path specifying where to save the identity.
        :returns: True if the file was saved, otherwise False.
        """
        try:
            with open(path, "wb") as key_file:
                key_file.write(self.get_private_key())
                return True
            return False
        except Exception as e:
            RNS.log("Error while saving identity to "+str(path), RNS.LOG_ERROR)
            RNS.log("The contained exception was: "+str(e))

    def __init__(self,create_keys=True):
        # Initialize keys to none
        self.prv           = None
        self.prv_bytes     = None
        self.sig_prv       = None
        self.sig_prv_bytes = None

        self.pub           = None
        self.pub_bytes     = None
        self.sig_pub       = None
        self.sig_pub_bytes = None

        self.hash          = None
        self.hexhash       = None

        if create_keys:
            self.create_keys()

    def create_keys(self):
        self.prv           = X25519PrivateKey.generate()
        self.prv_bytes     = self.prv.private_bytes()

        self.sig_prv       = Ed25519PrivateKey.generate()
        self.sig_prv_bytes = self.sig_prv.private_bytes()

        self.pub           = self.prv.public_key()
        self.pub_bytes     = self.pub.public_bytes()

        self.sig_pub       = self.sig_prv.public_key()
        self.sig_pub_bytes = self.sig_pub.public_bytes()

        self.update_hashes()

        RNS.log("Identity keys created for "+RNS.prettyhexrep(self.hash), RNS.LOG_VERBOSE)

    def get_private_key(self):
        """
        :returns: The private key as *bytes*
        """
        return self.prv_bytes+self.sig_prv_bytes

    def get_public_key(self):
        """
        :returns: The public key as *bytes*
        """
        return self.pub_bytes+self.sig_pub_bytes

    def load_private_key(self, prv_bytes):
        """
        Load a private key into the instance.

        :param prv_bytes: The private key as *bytes*.
        :returns: True if the key was loaded, otherwise False.
        """
        try:
            self.prv_bytes     = prv_bytes[:Identity.KEYSIZE//8//2]
            self.prv           = X25519PrivateKey.from_private_bytes(self.prv_bytes)
            self.sig_prv_bytes = prv_bytes[Identity.KEYSIZE//8//2:]
            self.sig_prv       = Ed25519PrivateKey.from_private_bytes(self.sig_prv_bytes)
            
            self.pub           = self.prv.public_key()
            self.pub_bytes     = self.pub.public_bytes()

            self.sig_pub       = self.sig_prv.public_key()
            self.sig_pub_bytes = self.sig_pub.public_bytes()

            self.update_hashes()

            return True

        except Exception as e:
            raise e
            RNS.log("Failed to load identity key", RNS.LOG_ERROR)
            RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
            return False

    def load_public_key(self, pub_bytes):
        """
        Load a public key into the instance.

        :param pub_bytes: The public key as *bytes*.
        :returns: True if the key was loaded, otherwise False.
        """
        try:
            self.pub_bytes     = pub_bytes[:Identity.KEYSIZE//8//2]
            self.sig_pub_bytes = pub_bytes[Identity.KEYSIZE//8//2:]

            self.pub           = X25519PublicKey.from_public_bytes(self.pub_bytes)
            self.sig_pub       = Ed25519PublicKey.from_public_bytes(self.sig_pub_bytes)

            self.update_hashes()
        except Exception as e:
            RNS.log("Error while loading public key, the contained exception was: "+str(e), RNS.LOG_ERROR)

    def update_hashes(self):
        self.hash = Identity.truncated_hash(self.get_public_key())
        self.hexhash = self.hash.hex()

    def load(self, path):
        try:
            with open(path, "rb") as key_file:
                prv_bytes = key_file.read()
                return self.load_private_key(prv_bytes)
            return False
        except Exception as e:
            RNS.log("Error while loading identity from "+str(path), RNS.LOG_ERROR)
            RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)

    def get_salt(self):
        return self.hash

    def get_context(self):
        return None

    def encrypt(self, plaintext, ratchet=None):
        """
        Encrypts information for the identity.

        :param plaintext: The plaintext to be encrypted as *bytes*.
        :returns: Ciphertext token as *bytes*.
        :raises: *KeyError* if the instance does not hold a public key.
        """
        if self.pub != None:
            ephemeral_key = X25519PrivateKey.generate()
            ephemeral_pub_bytes = ephemeral_key.public_key().public_bytes()

            if ratchet != None:
                target_public_key = X25519PublicKey.from_public_bytes(ratchet)
            else:
                target_public_key = self.pub

            shared_key = ephemeral_key.exchange(target_public_key)
            
            derived_key = RNS.Cryptography.hkdf(
                length=Identity.DERIVED_KEY_LENGTH,
                derive_from=shared_key,
                salt=self.get_salt(),
                context=self.get_context(),
            )

            token = Token(derived_key)
            ciphertext = token.encrypt(plaintext)
            token = ephemeral_pub_bytes+ciphertext

            return token
        else:
            raise KeyError("Encryption failed because identity does not hold a public key")

    def __decrypt(self, shared_key, ciphertext):
        derived_key = RNS.Cryptography.hkdf(
            length=Identity.DERIVED_KEY_LENGTH,
            derive_from=shared_key,
            salt=self.get_salt(),
            context=self.get_context())

        token = Token(derived_key)
        plaintext = token.decrypt(ciphertext)
        return plaintext

    def decrypt(self, ciphertext_token, ratchets=None, enforce_ratchets=False, ratchet_id_receiver=None):
        """
        Decrypts information for the identity.

        :param ciphertext: The ciphertext to be decrypted as *bytes*.
        :returns: Plaintext as *bytes*, or *None* if decryption fails.
        :raises: *KeyError* if the instance does not hold a private key.
        """

        if self.prv != None:
            if len(ciphertext_token) > Identity.KEYSIZE//8//2:
                plaintext = None
                try:
                    peer_pub_bytes = ciphertext_token[:Identity.KEYSIZE//8//2]
                    peer_pub = X25519PublicKey.from_public_bytes(peer_pub_bytes)
                    ciphertext = ciphertext_token[Identity.KEYSIZE//8//2:]

                    if ratchets:
                        for ratchet in ratchets:
                            try:
                                ratchet_prv = X25519PrivateKey.from_private_bytes(ratchet)
                                ratchet_id = Identity._get_ratchet_id(ratchet_prv.public_key().public_bytes())
                                shared_key = ratchet_prv.exchange(peer_pub)
                                plaintext = self.__decrypt(shared_key, ciphertext)
                                if ratchet_id_receiver:
                                    ratchet_id_receiver.latest_ratchet_id = ratchet_id
                                
                                break
                            
                            except Exception as e:
                                pass

                    if enforce_ratchets and plaintext == None:
                        RNS.log("Decryption with ratchet enforcement by "+RNS.prettyhexrep(self.hash)+" failed. Dropping packet.", RNS.LOG_DEBUG)
                        if ratchet_id_receiver:
                            ratchet_id_receiver.latest_ratchet_id = None
                        return None

                    if plaintext == None:
                        shared_key = self.prv.exchange(peer_pub)
                        plaintext = self.__decrypt(shared_key, ciphertext)

                        if ratchet_id_receiver:
                            ratchet_id_receiver.latest_ratchet_id = None

                except Exception as e:
                    RNS.log("Decryption by "+RNS.prettyhexrep(self.hash)+" failed: "+str(e), RNS.LOG_DEBUG)
                    if ratchet_id_receiver:
                        ratchet_id_receiver.latest_ratchet_id = None
                    
                return plaintext
            
            else:
                RNS.log("Decryption failed because the token size was invalid.", RNS.LOG_DEBUG)
                return None
        else:
            raise KeyError("Decryption failed because identity does not hold a private key")


    def sign(self, message):
        """
        Signs information by the identity.

        :param message: The message to be signed as *bytes*.
        :returns: Signature as *bytes*.
        :raises: *KeyError* if the instance does not hold a private key.
        """
        if self.sig_prv != None:
            try:
                return self.sig_prv.sign(message)    
            except Exception as e:
                RNS.log("The identity "+str(self)+" could not sign the requested message. The contained exception was: "+str(e), RNS.LOG_ERROR)
                raise e
        else:
            raise KeyError("Signing failed because identity does not hold a private key")

    def validate(self, signature, message):
        """
        Validates the signature of a signed message.

        :param signature: The signature to be validated as *bytes*.
        :param message: The message to be validated as *bytes*.
        :returns: True if the signature is valid, otherwise False.
        :raises: *KeyError* if the instance does not hold a public key.
        """
        if self.pub != None:
            try:
                self.sig_pub.verify(signature, message)
                return True
            except Exception as e:
                return False
        else:
            raise KeyError("Signature validation failed because identity does not hold a public key")

    def prove(self, packet, destination=None):
        signature = self.sign(packet.packet_hash)
        if RNS.Reticulum.should_use_implicit_proof():
            proof_data = signature
        else:
            proof_data = packet.packet_hash + signature
        
        if destination == None:
            destination = packet.generate_proof_destination()

        proof = RNS.Packet(destination, proof_data, RNS.Packet.PROOF, attached_interface = packet.receiving_interface)
        proof.send()

    def __str__(self):
        return RNS.prettyhexrep(self.hash)