3.0.ipv8.messaging.anonymization.crypto

Exceptions

Classes

Module Contents

class 3.0.ipv8.messaging.anonymization.crypto.SessionKeys

Session keys to communicate between hops.

key_forward: bytes

key_backward: bytes

salt_forward: bytes

salt_backward: bytes

salt_explicit_forward: int

salt_explicit_backward: int

exception 3.0.ipv8.messaging.anonymization.crypto.CryptoException

Bases: Exception

Exception for when anything goes wrong with sessions, encoding, and decoding.

class 3.0.ipv8.messaging.anonymization.crypto.CryptoEndpoint

UDP endpoint capable of sending/relaying/exiting CellPayloads.

settings: 3.0.ipv8.messaging.anonymization.community.TunnelSettings | None = None

prefix

circuits: dict[int, 3.0.ipv8.messaging.anonymization.tunnel.Circuit]

relays: dict[int, 3.0.ipv8.messaging.anonymization.tunnel.RelayRoute]

exit_sockets: dict[int, 3.0.ipv8.messaging.anonymization.exit_socket.TunnelExitSocket]

logger

abstract setup_tunnels(tunnel_community: 3.0.ipv8.messaging.anonymization.community.TunnelCommunity, settings: 3.0.ipv8.messaging.anonymization.community.TunnelSettings) → None: Set up the TunnelCommunity.

abstract send_cell(target_addr: ipv8.types.Address, cell: 3.0.ipv8.messaging.anonymization.payload.CellPayload) → None: Send the given payload directly to the given peer with the appropriate encryption rules.

class 3.0.ipv8.messaging.anonymization.crypto.PythonCryptoEndpoint(endpoint: ipv8.types.Endpoint)

Bases: CryptoEndpoint, 3.0.ipv8.messaging.interfaces.endpoint.EndpointListener

UDP endpoint capable of sending/relaying/exiting CellPayloads.

tunnel_community: 3.0.ipv8.messaging.anonymization.community.TunnelCommunity | None = None

setup_tunnels(tunnel_community: 3.0.ipv8.messaging.anonymization.community.TunnelCommunity, settings: 3.0.ipv8.messaging.anonymization.community.TunnelSettings) → None: Set up the TunnelCommunity.

property max_relay_early: int: Return the maximum number of relay_early cells that are allowed to pass a relay.

on_packet(packet: tuple[ipv8.types.Address, bytes], warn_unknown: bool = True) → None: Callback for when data is received on this endpoint.

send_cell(target_addr: ipv8.types.Address, cell: 3.0.ipv8.messaging.anonymization.payload.CellPayload) → None: Send the given payload directly to the given peer with the appropriate encryption rules.

process_cell(source_address: ipv8.types.Address, data: bytes) → None: Process incoming raw data, assumed to be a cell, originating from a given address.

relay_cell(cell: 3.0.ipv8.messaging.anonymization.payload.CellPayload) → None: Forward the given cell, which contains the information needed for its own relaying.

outgoing_crypto(cell: 3.0.ipv8.messaging.anonymization.payload.CellPayload) → 3.0.ipv8.messaging.anonymization.payload.CellPayload | None: Encrypt a CellPayload using the SessionKeys currently available in the routing table.

incoming_crypto(cell: 3.0.ipv8.messaging.anonymization.payload.CellPayload) → 3.0.ipv8.messaging.anonymization.payload.CellPayload | None: Decrypt a CellPayload using the SessionKeys currently available in the routing table.

encrypt_cell(cell: 3.0.ipv8.messaging.anonymization.payload.CellPayload, direction: int, *hops: 3.0.ipv8.messaging.anonymization.tunnel.Hop) → None

Encrypt a given cell.

Raises:: CryptoException – if encryption failed.

decrypt_cell(cell: 3.0.ipv8.messaging.anonymization.payload.CellPayload, direction: int, *hops: 3.0.ipv8.messaging.anonymization.tunnel.Hop) → None

Decrypt a given cell.

Raises:: CryptoException – if decryption failed.

class 3.0.ipv8.messaging.anonymization.crypto.TunnelCrypto

Bases: 3.0.ipv8.keyvault.crypto.ECCrypto

Add Diffie-Hellman key establishment logic to ECCrypto.

initialize(key: 3.0.ipv8.keyvault.crypto.LibNaCLPK) → None: Make this ECCrypto fit for key establishment based on the given public key.

is_key_compatible(key: 3.0.ipv8.types.PublicKey) → bool: Whether the given key is a LibNaCLPK instance.

generate_diffie_secret() → tuple[3.0.ipv8.keyvault.private.libnaclkey.LibNaCLSK, 3.0.ipv8.keyvault.crypto.LibNaCLPK]: Create a new private-public keypair.

generate_diffie_shared_secret(dh_received: bytes, key: 3.0.ipv8.keyvault.crypto.LibNaCLPK | None = None) → tuple[bytes, 3.0.ipv8.keyvault.crypto.LibNaCLPK, bytes]: Generate the shared secret from the received string and the given key.

verify_and_generate_shared_secret(dh_secret: 3.0.ipv8.keyvault.private.libnaclkey.LibNaCLSK, dh_received: bytes, auth: bytes, b: bytes) → bytes: Generate the shared secret based on the response to the shared string and our own key.

generate_session_keys(shared_secret: bytes) → SessionKeys: Generate new session keys based on the shared secret.

static encrypt_str(content: bytes, keys: SessionKeys, direction: int) → bytes: Encrypt content using the given key, salt, and incremental session salt.

static decrypt_str(content: bytes, keys: SessionKeys, direction: int) → bytes: Decrypt the given content using a key and salt.