Module livekit.rtc.e2ee

Classes

class E2EEManager (room_handle: int, options: Optional[E2EEOptions])

Expand source code

class E2EEManager:
    def __init__(self, room_handle: int, options: Optional[E2EEOptions]):
        self.options = options
        self._room_handle = room_handle
        self._enabled = options is not None

        if options is not None:
            self._key_provider = KeyProvider(
                self._room_handle, options.key_provider_options
            )

    @property
    def key_provider(self) -> Optional[KeyProvider]:
        return self._key_provider

    @property
    def enabled(self) -> bool:
        return self._enabled

    def set_enabled(self, enabled: bool) -> None:
        """Enables or disables end-to-end encryption.

        Parameters:
            enabled (bool): True to enable, False to disable.

        Example:
            ```python
            e2ee_manager.set_enabled(True)
            ```
        """
        self._enabled = enabled
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.manager_set_enabled.enabled = enabled
        FfiClient.instance.request(req)

    def frame_cryptors(self) -> List[FrameCryptor]:
        """Retrieves the list of frame cryptors for participants.

        Returns:
            List[FrameCryptor]: A list of FrameCryptor instances.

        Example:
            ```python
            cryptors = e2ee_manager.frame_cryptors()
            for cryptor in cryptors:
                print(cryptor.participant_identity)
            ```
        """
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle

        resp = FfiClient.instance.request(req)
        frame_cryptors = []
        for frame_cryptor in resp.e2ee.manager_get_frame_cryptors.frame_cryptors:
            frame_cryptors.append(
                FrameCryptor(
                    self._room_handle,
                    frame_cryptor.participant_identity,
                    frame_cryptor.key_index,
                    frame_cryptor.enabled,
                )
            )
        return frame_cryptors

Instance variables

prop enabled : bool

Expand source code

@property
def enabled(self) -> bool:
    return self._enabled

prop key_provider : Optional[KeyProvider]

Expand source code

@property
def key_provider(self) -> Optional[KeyProvider]:
    return self._key_provider

Methods

def frame_cryptors(self) ‑> List[FrameCryptor]

Retrieves the list of frame cryptors for participants.

Returns

List[FrameCryptor]

A list of FrameCryptor instances.

Example

cryptors = e2ee_manager.frame_cryptors()
for cryptor in cryptors:
    print(cryptor.participant_identity)

def set_enabled(self, enabled: bool) ‑> None

Enables or disables end-to-end encryption.

Parameters

enabled (bool): True to enable, False to disable.

Example

e2ee_manager.set_enabled(True)

class E2EEOptions (key_provider_options: KeyProviderOptions = <factory>, encryption_type: int = 1)

E2EEOptions(key_provider_options: livekit.rtc.e2ee.KeyProviderOptions = , encryption_type: int = 1)

Expand source code

@dataclass
class E2EEOptions:
    key_provider_options: KeyProviderOptions = field(default_factory=KeyProviderOptions)
    encryption_type: proto_e2ee.EncryptionType.ValueType = proto_e2ee.EncryptionType.GCM

Class variables

var encryption_type : int

var key_provider_options : KeyProviderOptions

class FrameCryptor (room_handle: int, participant_identity: str, key_index: int, enabled: bool)

Expand source code

class FrameCryptor:
    def __init__(
        self, room_handle: int, participant_identity: str, key_index: int, enabled: bool
    ):
        self._room_handle = room_handle
        self._enabled = enabled
        self._participant_identity = participant_identity
        self._key_index = key_index

    @property
    def participant_identity(self) -> str:
        return self._participant_identity

    @property
    def key_index(self) -> int:
        return self._key_index

    @property
    def enabled(self) -> bool:
        return self._enabled

    def set_enabled(self, enabled: bool) -> None:
        """Enables or disables frame encryption.

        Parameters:
            enabled (bool): True to enable, False to disable.

        Example:
            ```python
            frame_cryptor.set_enabled(True)
            ```
        """
        self._enabled = enabled
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.cryptor_set_enabled.participant_identity = self._participant_identity
        req.e2ee.cryptor_set_enabled.enabled = enabled
        FfiClient.instance.request(req)

    def set_key_index(self, key_index: int) -> None:
        """Sets the key index for encryption/decryption.

        Parameters:
            key_index (int): The new key index.

        Example:
            ```python
            frame_cryptor.set_key_index(3)
            ```
        """
        self._key_index = key_index
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.cryptor_set_key_index.participant_identity = self._participant_identity
        req.e2ee.cryptor_set_key_index.key_index = key_index
        FfiClient.instance.request(req)

Instance variables

prop enabled : bool

Expand source code

@property
def enabled(self) -> bool:
    return self._enabled

prop key_index : int

Expand source code

@property
def key_index(self) -> int:
    return self._key_index

prop participant_identity : str

Expand source code

@property
def participant_identity(self) -> str:
    return self._participant_identity

Methods

def set_enabled(self, enabled: bool) ‑> None

Enables or disables frame encryption.

Parameters

enabled (bool): True to enable, False to disable.

Example

frame_cryptor.set_enabled(True)

def set_key_index(self, key_index: int) ‑> None

Sets the key index for encryption/decryption.

Parameters

key_index (int): The new key index.

Example

frame_cryptor.set_key_index(3)

class KeyProvider (room_handle: int, options: KeyProviderOptions)

Expand source code

class KeyProvider:
    def __init__(self, room_handle: int, options: KeyProviderOptions):
        self._options = options
        self._room_handle = room_handle

    @property
    def options(self) -> KeyProviderOptions:
        return self._options

    def set_shared_key(self, key: bytes, key_index: int) -> None:
        """Sets the shared encryption key.

        Parameters:
            key (bytes): The new shared key.
            key_index (int): The index of the key.

        Example:
            ```python
            key_provider.set_shared_key(b"my_shared_key", key_index=1)
            ```
        """
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.set_shared_key.key_index = key_index
        req.e2ee.set_shared_key.shared_key = key
        FfiClient.instance.request(req)

    def export_shared_key(self, key_index: int) -> bytes:
        """Exports the shared encryption key.

        Parameters:
            key_index (int): The index of the key to export.

        Returns:
            bytes: The exported shared key.

        Example:
            ```python
            key = key_provider.export_shared_key(key_index=1)
            ```
        """
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.get_shared_key.key_index = key_index
        resp = FfiClient.instance.request(req)
        key = resp.e2ee.get_shared_key.key
        return key

    def ratchet_shared_key(self, key_index: int) -> bytes:
        """Ratchets the shared encryption key to a new key.

        Parameters:
            key_index (int): The index of the key to ratchet.

        Returns:
            bytes: The new ratcheted shared key.

        Example:
            ```python
            new_key = key_provider.ratchet_shared_key(key_index=1)
            ```
        """
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.ratchet_shared_key.key_index = key_index

        resp = FfiClient.instance.request(req)

        new_key = resp.e2ee.ratchet_shared_key.new_key
        return new_key

    def set_key(self, participant_identity: str, key: bytes, key_index: int) -> None:
        """Sets the encryption key for a specific participant.

        Parameters:
            participant_identity (str): The identity of the participant.
            key (bytes): The encryption key to set.
            key_index (int): The index of the key.

        Example:
            ```python
            key_provider.set_key("participant123", b"participant_key", key_index=2)
            ```
        """
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.set_key.participant_identity = participant_identity
        req.e2ee.set_key.key_index = key_index
        req.e2ee.set_key.key = key

        self.key_index = key_index
        FfiClient.instance.request(req)

    def export_key(self, participant_identity: str, key_index: int) -> bytes:
        """Exports the encryption key for a specific participant.

        Parameters:
            participant_identity (str): The identity of the participant.
            key_index (int): The index of the key to export.

        Returns:
            bytes: The exported key.

        Example:
            ```python
            key = key_provider.export_key("participant123", key_index=2)
            ```
        """
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.get_key.participant_identity = participant_identity
        req.e2ee.get_key.key_index = key_index
        resp = FfiClient.instance.request(req)
        key = resp.e2ee.get_key.key
        return key

    def ratchet_key(self, participant_identity: str, key_index: int) -> bytes:
        """Ratchets the encryption key for a specific participant to a new key.

        Parameters:
            participant_identity (str): The identity of the participant.
            key_index (int): The index of the key to ratchet.

        Returns:
            bytes: The new ratcheted key.

        Example:
            ```python
            new_key = key_provider.ratchet_key("participant123", key_index=2)
            ```
        """
        req = proto_ffi.FfiRequest()
        req.e2ee.room_handle = self._room_handle
        req.e2ee.ratchet_key.participant_identity = participant_identity
        req.e2ee.ratchet_key.key_index = key_index

        resp = FfiClient.instance.request(req)
        new_key = resp.e2ee.ratchet_key.new_key
        return new_key

Instance variables

prop options : KeyProviderOptions

Expand source code

@property
def options(self) -> KeyProviderOptions:
    return self._options

Methods

def export_key(self, participant_identity: str, key_index: int) ‑> bytes

Exports the encryption key for a specific participant.

Parameters

participant_identity (str): The identity of the participant. key_index (int): The index of the key to export.

Returns

bytes

The exported key.

Example

key = key_provider.export_key("participant123", key_index=2)

def export_shared_key(self, key_index: int) ‑> bytes

Exports the shared encryption key.

Parameters

key_index (int): The index of the key to export.

Returns

bytes

The exported shared key.

Example

key = key_provider.export_shared_key(key_index=1)

def ratchet_key(self, participant_identity: str, key_index: int) ‑> bytes

Ratchets the encryption key for a specific participant to a new key.

Parameters

participant_identity (str): The identity of the participant. key_index (int): The index of the key to ratchet.

Returns

bytes

The new ratcheted key.

Example

new_key = key_provider.ratchet_key("participant123", key_index=2)

def ratchet_shared_key(self, key_index: int) ‑> bytes

Ratchets the shared encryption key to a new key.

Parameters

key_index (int): The index of the key to ratchet.

Returns

bytes

The new ratcheted shared key.

Example

new_key = key_provider.ratchet_shared_key(key_index=1)

def set_key(self, participant_identity: str, key: bytes, key_index: int) ‑> None

Sets the encryption key for a specific participant.

Parameters

participant_identity (str): The identity of the participant. key (bytes): The encryption key to set. key_index (int): The index of the key.

Example

key_provider.set_key("participant123", b"participant_key", key_index=2)

def set_shared_key(self, key: bytes, key_index: int) ‑> None

Sets the shared encryption key.

Parameters

key (bytes): The new shared key. key_index (int): The index of the key.

Example

key_provider.set_shared_key(b"my_shared_key", key_index=1)

class KeyProviderOptions (shared_key: Optional[bytes] = None, ratchet_salt: bytes = b'LKFrameEncryptionKey', ratchet_window_size: int = 16, failure_tolerance: int = -1)

KeyProviderOptions(shared_key: Optional[bytes] = None, ratchet_salt: bytes = b'LKFrameEncryptionKey', ratchet_window_size: int = 16, failure_tolerance: int = -1)

Expand source code

@dataclass
class KeyProviderOptions:
    shared_key: Optional[bytes] = None
    ratchet_salt: bytes = DEFAULT_RATCHET_SALT
    ratchet_window_size: int = DEFAULT_RATCHET_WINDOW_SIZE
    failure_tolerance: int = DEFAULT_FAILURE_TOLERANCE

Class variables

var failure_tolerance : int

var ratchet_salt : bytes

var ratchet_window_size : int

var shared_key : Optional[bytes]