libp2p.security.tls package

Submodules

libp2p.security.tls.certificate module

TLS certificate utilities for libp2p.

This module provides certificate generation and verification functions that embed libp2p peer identity information in X.509 extensions.

class libp2p.security.tls.certificate.SignedKey(public_key_bytes: bytes, signature: bytes)

Bases: object

Represents a signed public key embedded in certificate extension.

public_key_bytes: bytes

signature: bytes

libp2p.security.tls.certificate.add_libp2p_extension(cert_builder: CertificateBuilder, peer_public_key: PublicKey, signature: bytes) → CertificateBuilder

Add libp2p peer identity extension to certificate.

Args:: cert_builder: Certificate builder to modify peer_public_key: Peer’s public key to embed signature: Signature over the certificate’s public key
Returns:: Certificate builder with libp2p extension

libp2p.security.tls.certificate.create_cert_template() → CertificateBuilder

Create a certificate template for libp2p TLS certificates.

Uses secure defaults: - Random 64-bit serial number - 1 hour backdating for not_before to handle clock skew - 100 year validity (since these are ephemeral, self-signed certs) - Basic Constraints: not a CA - Key Usage: digital signature only

Returns:: Certificate builder template with secure defaults

libp2p.security.tls.certificate.decode_signed_key(der_bytes: bytes) → SignedKey

Parse DER-encoded SignedKey from the libp2p X.509 extension value.

Args:: der_bytes: DER bytes for SignedKey
Returns:: Parsed SignedKey instance

libp2p.security.tls.certificate.encode_signed_key(public_key_bytes: bytes, signature: bytes) → bytes

ASN.1-encode the SignedKey structure for inclusion in the libp2p X.509 extension.

Args:: public_key_bytes: libp2p protobuf-encoded public key bytes signature: signature over prefix+certificate public key
Returns:: DER-encoded bytes representing the SignedKey sequence
Raises:: ValueError: If inputs are empty or exceed maximum allowed sizes

libp2p.security.tls.certificate.generate_certificate(private_key: PrivateKey, cert_template: CertificateBuilder) → tuple[str, str]

Generate a self-signed certificate with libp2p extensions.

Args:: private_key: Private key for signing cert_template: Certificate template
Returns:: Tuple of (certificate PEM, private key PEM)

libp2p.security.tls.certificate.generate_self_signed_cert() → tuple[EllipticCurvePrivateKey, Certificate]

Generate a self-signed certificate for testing purposes.

This is a utility function based on the guide examples.

Returns:: Tuple of (private key, certificate)

libp2p.security.tls.certificate.pub_key_from_cert_chain(cert_chain: list[Certificate]) → PublicKey

Extract public key from certificate chain.

This is an alias for verify_certificate_chain for compatibility.

Args:: cert_chain: Certificate chain
Returns:: Public key

libp2p.security.tls.certificate.verify_certificate_chain(cert_chain: list[Certificate]) → PublicKey

Verify certificate chain and extract peer public key from libp2p extension.

Args:

cert_chain: List of certificates in the chain

Returns:

Public key from libp2p extension

Raises:

ValueError: If verification fails, such as expired certificate,: missing extension, invalid signature, or unsupported key type.

libp2p.security.tls.io module

TLS I/O utilities for libp2p.

This module provides TLS-specific stream reading and writing functionality. Unlike Noise (which is message-based), TLS is a stream protocol, so we use it directly as a stream without additional message framing.

class libp2p.security.tls.io.TLSReadWriter(conn: IRawConnection, ssl_context: SSLContext, local_prim_pk: bytes, server_side: bool = False, server_hostname: str | None = None)

Bases: EncryptedMsgReadWriter

TLS encrypted stream reader/writer.

This class handles TLS encryption/decryption over a raw connection. Unlike Noise (which is message-based), TLS is a stream protocol. We use it directly as a stream, allowing multistream’s varint framing to work over the TLS stream (like Go’s implementation).

The pattern matches Go’s TLS implementation: 1. Write plaintext directly to TLS stream (no additional framing) 2. Read chunks from TLS stream (multistream handles varint parsing)

async close() → None: Close the TLS connection.

decrypt(data: bytes) → bytes

Decrypt received data.

Args:: data: Encrypted data to decrypt
Returns:: Decrypted data

encrypt(data: bytes) → bytes

Encrypt data for transmission.

Args:: data: Data to encrypt
Returns:: Encrypted data

get_negotiated_protocol() → str | None: Return the ALPN-negotiated protocol (e.g., selected muxer) if any.

get_peer_certificate() → Certificate | None

Get the peer’s certificate after handshake.

Returns:: Peer certificate or None if not available

get_remote_address() → tuple[str, int] | None

Get remote address from underlying connection.

Returns:: Remote address tuple or None

async handshake(enable_autotls: bool = False) → None

Perform TLS handshake.

Raises:

HandshakeFailure: If handshake fails due to protocol errors RuntimeError: If handshake timeout or connection errors occur ssl.SSLError: For SSL-specific errors not related to want read/write

Notes:

Implements defense against slow handshakes that could tie up resources
Properly handles connection errors and cleanup
Verifies minimum TLS version (1.3)

async read_msg() → bytes

Read data from TLS stream (no message framing).

Returns a chunk of decrypted data from the TLS stream.

stream_writer: TLSStreamReadWriter

async write_msg(msg: bytes) → None

Write data to TLS stream (no additional framing).

Data is encrypted by TLS and sent over the raw connection.

class libp2p.security.tls.io.TLSStreamReadWriter(conn: IRawConnection, ssl_context: SSLContext, local_prim_pk: bytes, server_side: bool = False, server_hostname: str | None = None)

Bases: ReadWriteCloser

Low-level TLS stream reader/writer that handles SSL socket operations.

This class provides the raw TLS encryption/decryption as a stream. Used by TLSReadWriter which adapts it to EncryptedMsgReadWriter.

async close() → None

get_negotiated_protocol() → str | None

get_peer_certificate() → Certificate | None

get_remote_address() → tuple[str, int] | None

Return the remote address of the connected peer.

Returns:: A tuple of (host, port) or None if not available

async handshake(enable_autotls: bool = False) → None: Perform TLS handshake.

async read(n: int | None = None) → bytes: Read raw bytes from TLS stream.

should_do_primitive_key_exchang(enable_autotls: bool) → bool

Determine if the primitive key exchange can proceed.

Checks whether AutoTLS is enabled and whether any peers are in the middle of the AutoTLS procedure. Returns True only if key exchange is safe to run.

Parameters:: enable_autotls – Flag indicating whether AutoTLS is enabled.
Returns:: True if primitive key exchange should proceed, False otherwise.

async write(data: bytes) → None: Write raw bytes to TLS stream.

libp2p.security.tls.transport module

class libp2p.security.tls.transport.IdentityConfig(cert_template: CertificateBuilder | None = None, key_log_writer: Any | None = None)

Bases: object

Configuration for TLS identity.

cert_template: CertificateBuilder | None = None

key_log_writer: Any | None = None

class libp2p.security.tls.transport.TLSTransport(libp2p_keypair: KeyPair, early_data: bytes | None = None, muxers: list[str] | None = None, identity_config: IdentityConfig | None = None, enable_autotls: bool = False)

Bases: ISecureTransport

TLS transport implementation following the noise pattern.

Features: - TLS 1.3 support - Custom certificate generation with libp2p extensions - Peer ID verification - ALPN protocol negotiation

create_ssl_context(server_side: bool = False) → SSLContext

Create SSL context for TLS connections.

Args:: server_side: Whether this is for server-side connections
Returns:: Configured SSL context
Raises:: ValueError: If any trusted certificate contains dangerous content

early_data: bytes | None

get_certificate_pem() → str

get_negotiated_muxer() → str | None: Placeholder: return the muxer negotiated via ALPN, if any.

get_preferred_muxers() → list[str]: Return the list of preferred stream muxers for ALPN negotiation.

get_protocol_id() → TProtocol: Get the protocol ID for this transport.

libp2p_privkey: PrivateKey

local_peer: ID

async secure_inbound(conn: IRawConnection) → ISecureConn

Secure an inbound connection as server.

Args:: conn: Raw connection to secure
Returns:: Secured connection (SecureSession)

async secure_outbound(conn: IRawConnection, peer_id: ID) → ISecureConn

Secure an outbound connection as client.

Args:: conn: Raw connection to secure peer_id: Expected peer ID
Returns:: Secured connection (SecureSession)

trust_peer_cert_pem(pem: str) → None

Add a trusted peer certificate PEM.

Args:: pem: The PEM-encoded certificate to trust
Raises:: ValueError: If the certificate contains invalid characters or format

libp2p.security.tls.transport.create_tls_transport(libp2p_keypair: KeyPair, early_data: bytes | None = None, muxers: list[str] | None = None, identity_config: IdentityConfig | None = None) → TLSTransport

Create a new TLS transport.

Args:: libp2p_keypair: Key pair for the local peer early_data: Optional early data for TLS handshake muxers: Optional list of preferred stream muxer protocol IDs for ALPN identity_config: Optional TLS identity config (cert template, key log writer)
Returns:: TLS transport instance

Module contents

TLS security transport for libp2p.

This module provides a comprehensive TLS transport implementation that follows the Go libp2p TLS specification.

class libp2p.security.tls.IdentityConfig(cert_template: CertificateBuilder | None = None, key_log_writer: Any | None = None)

Bases: object

Configuration for TLS identity.

cert_template: CertificateBuilder | None = None

key_log_writer: Any | None = None

class libp2p.security.tls.SignedKey(public_key_bytes: bytes, signature: bytes)

Bases: object

Represents a signed public key embedded in certificate extension.

public_key_bytes: bytes

signature: bytes

class libp2p.security.tls.TLSReadWriter(conn: IRawConnection, ssl_context: SSLContext, local_prim_pk: bytes, server_side: bool = False, server_hostname: str | None = None)

Bases: EncryptedMsgReadWriter

TLS encrypted stream reader/writer.

This class handles TLS encryption/decryption over a raw connection. Unlike Noise (which is message-based), TLS is a stream protocol. We use it directly as a stream, allowing multistream’s varint framing to work over the TLS stream (like Go’s implementation).

The pattern matches Go’s TLS implementation: 1. Write plaintext directly to TLS stream (no additional framing) 2. Read chunks from TLS stream (multistream handles varint parsing)

async close() → None: Close the TLS connection.

conn: Any | None

decrypt(data: bytes) → bytes

Decrypt received data.

Args:: data: Encrypted data to decrypt
Returns:: Decrypted data

encrypt(data: bytes) → bytes

Encrypt data for transmission.

Args:: data: Data to encrypt
Returns:: Encrypted data

get_negotiated_protocol() → str | None: Return the ALPN-negotiated protocol (e.g., selected muxer) if any.

get_peer_certificate() → Certificate | None

Get the peer’s certificate after handshake.

Returns:: Peer certificate or None if not available

get_remote_address() → tuple[str, int] | None

Get remote address from underlying connection.

Returns:: Remote address tuple or None

async handshake(enable_autotls: bool = False) → None

Perform TLS handshake.

Raises:

HandshakeFailure: If handshake fails due to protocol errors RuntimeError: If handshake timeout or connection errors occur ssl.SSLError: For SSL-specific errors not related to want read/write

Notes:

Implements defense against slow handshakes that could tie up resources
Properly handles connection errors and cleanup
Verifies minimum TLS version (1.3)

async read_msg() → bytes

Read data from TLS stream (no message framing).

Returns a chunk of decrypted data from the TLS stream.

stream_writer: TLSStreamReadWriter

async write_msg(msg: bytes) → None

Write data to TLS stream (no additional framing).

Data is encrypted by TLS and sent over the raw connection.

class libp2p.security.tls.TLSTransport(libp2p_keypair: KeyPair, early_data: bytes | None = None, muxers: list[str] | None = None, identity_config: IdentityConfig | None = None, enable_autotls: bool = False)

Bases: ISecureTransport

TLS transport implementation following the noise pattern.

Features: - TLS 1.3 support - Custom certificate generation with libp2p extensions - Peer ID verification - ALPN protocol negotiation

create_ssl_context(server_side: bool = False) → SSLContext

Create SSL context for TLS connections.

Args:: server_side: Whether this is for server-side connections
Returns:: Configured SSL context
Raises:: ValueError: If any trusted certificate contains dangerous content

early_data: bytes | None

get_certificate_pem() → str

get_negotiated_muxer() → str | None: Placeholder: return the muxer negotiated via ALPN, if any.

get_preferred_muxers() → list[str]: Return the list of preferred stream muxers for ALPN negotiation.

get_protocol_id() → TProtocol: Get the protocol ID for this transport.

libp2p_privkey: PrivateKey

local_peer: ID

async secure_inbound(conn: IRawConnection) → ISecureConn

Secure an inbound connection as server.

Args:: conn: Raw connection to secure
Returns:: Secured connection (SecureSession)

async secure_outbound(conn: IRawConnection, peer_id: ID) → ISecureConn

Secure an outbound connection as client.

Args:: conn: Raw connection to secure peer_id: Expected peer ID
Returns:: Secured connection (SecureSession)

trust_peer_cert_pem(pem: str) → None

Add a trusted peer certificate PEM.

Args:: pem: The PEM-encoded certificate to trust
Raises:: ValueError: If the certificate contains invalid characters or format

libp2p.security.tls.create_cert_template() → CertificateBuilder

Create a certificate template for libp2p TLS certificates.

Uses secure defaults: - Random 64-bit serial number - 1 hour backdating for not_before to handle clock skew - 100 year validity (since these are ephemeral, self-signed certs) - Basic Constraints: not a CA - Key Usage: digital signature only

Returns:: Certificate builder template with secure defaults

libp2p.security.tls.create_tls_transport(libp2p_keypair: KeyPair, early_data: bytes | None = None, muxers: list[str] | None = None, identity_config: IdentityConfig | None = None) → TLSTransport

Create a new TLS transport.

Args:: libp2p_keypair: Key pair for the local peer early_data: Optional early data for TLS handshake muxers: Optional list of preferred stream muxer protocol IDs for ALPN identity_config: Optional TLS identity config (cert template, key log writer)
Returns:: TLS transport instance

libp2p.security.tls.generate_certificate(private_key: PrivateKey, cert_template: CertificateBuilder) → tuple[str, str]

Generate a self-signed certificate with libp2p extensions.

Args:: private_key: Private key for signing cert_template: Certificate template
Returns:: Tuple of (certificate PEM, private key PEM)

libp2p.security.tls.pub_key_from_cert_chain(cert_chain: list[Certificate]) → PublicKey

Extract public key from certificate chain.

This is an alias for verify_certificate_chain for compatibility.

Args:: cert_chain: Certificate chain
Returns:: Public key

libp2p.security.tls.verify_certificate_chain(cert_chain: list[Certificate]) → PublicKey

Verify certificate chain and extract peer public key from libp2p extension.

Args:

cert_chain: List of certificates in the chain

Returns:

Public key from libp2p extension

Raises:

ValueError: If verification fails, such as expired certificate,: missing extension, invalid signature, or unsupported key type.