latest.ipv8.messaging.anonymization.crypto

Module Contents

Classes

SessionKeys	Session keys to communicate between hops.
CryptoEndpoint	UDP endpoint capable of sending/relaying/exiting CellPayloads.
PythonCryptoEndpoint	UDP endpoint capable of sending/relaying/exiting CellPayloads.
TunnelCrypto	Add Diffie-Hellman key establishment logic to ECCrypto.

class latest.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 latest.ipv8.messaging.anonymization.crypto.CryptoException

Bases: Exception

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

class latest.ipv8.messaging.anonymization.crypto.CryptoEndpoint

UDP endpoint capable of sending/relaying/exiting CellPayloads.

abstract setup_tunnels(tunnel_community: latest.ipv8.messaging.anonymization.community.TunnelCommunity, settings: latest.ipv8.messaging.anonymization.community.TunnelSettings) → None

Set up the TunnelCommunity.

abstract send_cell(target_addr: ipv8.types.Address, cell: latest.ipv8.messaging.anonymization.payload.CellPayload) → None

Send the given payload directly to the given peer with the appropriate encryption rules.

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

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

UDP endpoint capable of sending/relaying/exiting CellPayloads.

property max_relay_early: int

Return the maximum number of relay_early cells that are allowed to pass a relay.

setup_tunnels(tunnel_community: latest.ipv8.messaging.anonymization.community.TunnelCommunity, settings: latest.ipv8.messaging.anonymization.community.TunnelSettings) → None

Set up the TunnelCommunity.

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: latest.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: latest.ipv8.messaging.anonymization.payload.CellPayload) → None

Forward the given cell, which contains the information needed for its own relaying.

outgoing_crypto(cell: latest.ipv8.messaging.anonymization.payload.CellPayload) → latest.ipv8.messaging.anonymization.payload.CellPayload | None

Encrypt a CellPayload using the SessionKeys currently available in the routing table.

incoming_crypto(cell: latest.ipv8.messaging.anonymization.payload.CellPayload) → latest.ipv8.messaging.anonymization.payload.CellPayload | None

Decrypt a CellPayload using the SessionKeys currently available in the routing table.

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

Encrypt a given cell.

Raises:

CryptoException – if encryption failed.

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

Decrypt a given cell.

Raises:

CryptoException – if decryption failed.

class latest.ipv8.messaging.anonymization.crypto.TunnelCrypto

Bases: latest.ipv8.keyvault.crypto.ECCrypto

Add Diffie-Hellman key establishment logic to ECCrypto.

initialize(key: latest.ipv8.keyvault.crypto.LibNaCLPK) → None

Make this ECCrypto fit for key establishment based on the given public key.

is_key_compatible(key: latest.ipv8.types.PublicKey) → bool

Whether the given key is a LibNaCLPK instance.

generate_diffie_secret() → tuple[latest.ipv8.keyvault.private.libnaclkey.LibNaCLSK, latest.ipv8.keyvault.crypto.LibNaCLPK]

Create a new private-public keypair.

generate_diffie_shared_secret(dh_received: bytes, key: latest.ipv8.keyvault.crypto.LibNaCLPK | None = None) → tuple[bytes, latest.ipv8.keyvault.crypto.LibNaCLPK, bytes]

Generate the shared secret from the received string and the given key.

verify_and_generate_shared_secret(dh_secret: latest.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.