feature(e2ee): add data channel encryption

examples/encrypted_text_stream/Cargo.toml

@@ -0,0 +1,14 @@
+[package]
+name = "encrypted_text_stream"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+tokio = { version = "1", features = ["full"] }
+futures-util = { version = "0.3", default-features = false, features = ["sink"] }
+livekit = { path = "../../livekit", features = ["native-tls"] }
+livekit-api = { path = "../../livekit-api" }
+log = "0.4.26"
+env_logger = "0.11.7"
+
+[workspace]

examples/encrypted_text_stream/src/main.rs

@@ -0,0 +1,172 @@
+use futures_util::TryStreamExt;
+use livekit::{
+    e2ee::{
+        key_provider::{KeyProvider, KeyProviderOptions},
+        E2eeOptions, EncryptionType,
+    },
+    prelude::*,
+    SimulateScenario, StreamReader, StreamTextOptions, TextStreamReader,
+};
+use livekit_api::access_token;
+use std::{env, error::Error, io::Write};
+use tokio::{
+    io::{self, AsyncBufReadExt, BufReader},
+    sync::mpsc::UnboundedReceiver,
+};
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    env_logger::init();
+
+    let url = env::var("LIVEKIT_URL").unwrap_or_else(|_| "ws://localhost:7880".to_string());
+    let api_key = env::var("LIVEKIT_API_KEY").unwrap_or_else(|_| "devkey".to_string());
+    let api_secret = env::var("LIVEKIT_API_SECRET").unwrap_or_else(|_| "secret".to_string());
+    let room_name = env::var("LIVEKIT_ROOM").unwrap_or_else(|_| "dev".to_string());
+    let identity = env::var("LIVEKIT_IDENTITY").unwrap_or_else(|_| "rust-participant".to_string());
+
+    // Prompt for encryption password
+    print!("Enter encryption password: ");
+    std::io::stdout().flush()?;
+    let mut password = String::new();
+    std::io::stdin().read_line(&mut password)?;
+    let password = password.trim().to_string();
+
+    // Create access token
+    let token = access_token::AccessToken::with_api_key(&api_key, &api_secret)
+        .with_identity(&identity)
+        .with_name(&format!("{} (Encrypted)", identity))
+        .with_grants(access_token::VideoGrants {
+            room_join: true,
+            room: room_name.clone(),
+            ..Default::default()
+        })
+        .to_jwt()?;
+
+    // Set up E2EE key provider
+    let key_provider =
+        KeyProvider::with_shared_key(KeyProviderOptions::default(), password.as_bytes().to_vec());
+
+    // Configure room options with encryption
+    let mut room_options = RoomOptions::default();
+    room_options.encryption =
+        Some(E2eeOptions { key_provider, encryption_type: livekit::e2ee::EncryptionType::Gcm });
+
+    // Connect to room
+    let (room, rx) = Room::connect(&url, &token, room_options).await?;
+    println!("Connected to encrypted room: {} - {}", room.name(), room.sid().await);
+
+    // Enable E2EE
+    room.e2ee_manager().set_enabled(true);
+    println!("End-to-end encryption enabled!");
+
+    // Run the interactive chat
+    run_interactive_chat(room, rx).await
+}
+
+async fn run_interactive_chat(
+    room: Room,
+    mut rx: UnboundedReceiver<RoomEvent>,
+) -> Result<(), Box<dyn Error>> {
+    println!("\n=== Encrypted Text Chat ===");
+    println!("Type messages to send (press Enter). Type 'quit' to exit.");
+    println!("Incoming messages will be displayed below:\n");
+
+    let stdin = io::stdin();
+    let mut stdin_reader = BufReader::new(stdin);
+
+    loop {
+        tokio::select! {
+            // Handle user input
+            input_result = read_user_input(&mut stdin_reader) => {
+                match input_result {
+                    Ok(Some(input)) => {
+                        if input == "quit" {
+                            println!("Goodbye!");
+                            break;
+                        }
+
+                        // Send the message
+                        let options = StreamTextOptions {
+                            topic: "lk.chat".to_string(),
+                            ..Default::default()
+                        };
+
+                        match room.local_participant().send_text(&input, options).await {
+                            Ok(_) => {
+                                println!("✓ Sent (encrypted): {}", input);
+                            }
+                            Err(e) => {
+                                println!("✗ Failed to send: {}", e);
+                            }
+                        }
+                    }
+                    Ok(None) => {
+                        // Empty input, continue
+                        continue;
+                    }
+                    Err(e) => {
+                        eprintln!("Error reading input: {}", e);
+                        break;
+                    }
+                }
+            }
+
+            // Handle incoming room events
+            event = rx.recv() => {
+                match event {
+                    Some(RoomEvent::TextStreamOpened { reader, topic, participant_identity }) => {
+                        if topic == "lk.chat" {
+                            if let Some(mut reader) = reader.take() {
+                                match reader.read_all().await {
+                                    Ok(message) => {
+                                        println!("📨 {} (decrypted): {}", participant_identity, message);
+                                    }
+                                    Err(e) => {
+                                        println!("✗ Failed to read message from {}: {}", participant_identity, e);
+                                    }
+                                }
+                            }
+                        }
+                    }
+                    Some(RoomEvent::ParticipantConnected( participant )) => {
+                        println!("👋 {} joined the room", participant.identity());
+                    }
+                    Some(RoomEvent::ParticipantDisconnected ( participant )) => {
+                        println!("👋 {} left the room", participant.identity());
+                    }
+                    Some(RoomEvent::Disconnected { reason }) => {
+                        println!("Disconnected from room: {:?}", reason);
+                        break;
+                    }
+                    Some(_) => {
+                        // Ignore other events
+                    }
+                    None => {
+                        println!("Room event stream ended");
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    Ok(())
+}
+
+async fn read_user_input(
+    stdin_reader: &mut BufReader<io::Stdin>,
+) -> Result<Option<String>, Box<dyn Error>> {
+    let mut input = String::new();
+    let bytes_read = stdin_reader.read_line(&mut input).await?;
+
+    if bytes_read == 0 {
+        return Ok(None); // EOF
+    }
+
+    let input = input.trim().to_string();
+    if input.is_empty() {
+        return Ok(None);
+    }
+
+    Ok(Some(input))
+}

libwebrtc/src/native/frame_cryptor.rs

@@ -36,6 +36,13 @@ pub enum EncryptionState {
     InternalError,
 }
 
+#[derive(Debug, Clone)]
+pub struct EncryptedPacket {
+    pub data: Vec<u8>,
+    pub iv: Vec<u8>,
+    pub key_index: u32,
+}
+
 #[derive(Clone)]
 pub struct KeyProvider {
     pub(crate) sys_handle: SharedPtr<sys_fc::ffi::KeyProvider>,
@@ -149,6 +156,49 @@ impl FrameCryptor {
     }
 }
 
+#[derive(Clone)]
+pub struct DataPacketCryptor {
+    pub(crate) sys_handle: SharedPtr<sys_fc::ffi::DataPacketCryptor>,
+}
+
+impl DataPacketCryptor {
+    pub fn new(algorithm: EncryptionAlgorithm, key_provider: KeyProvider) -> Self {
+        Self {
+            sys_handle: sys_fc::ffi::new_data_packet_cryptor(
+                algorithm.into(),
+                key_provider.sys_handle,
+            ),
+        }
+    }
+
+    pub fn encrypt(
+        &self,
+        participant_id: &str,
+        key_index: u32,
+        data: &[u8],
+    ) -> Result<EncryptedPacket, Box<dyn std::error::Error>> {
+        let data_vec: Vec<u8> = data.to_vec();
+        match self.sys_handle.encrypt_data_packet(participant_id.to_string(), key_index, data_vec) {
+            Ok(packet) => Ok(packet.into()),
+            Err(e) => Err(format!("Encryption failed: {}", e).into()),
+        }
+    }
+
+    pub fn decrypt(
+        &self,
+        participant_id: &str,
+        encrypted_packet: &EncryptedPacket,
+    ) -> Result<Vec<u8>, Box<dyn std::error::Error>> {
+        match self
+            .sys_handle
+            .decrypt_data_packet(participant_id.to_string(), &encrypted_packet.clone().into())
+        {
+            Ok(data) => Ok(data.into_iter().collect()),
+            Err(e) => Err(format!("Decryption failed: {}", e).into()),
+        }
+    }
+}
+
 #[derive(Default)]
 struct RtcFrameCryptorObserver {
     state_change_handler: Mutex<Option<OnStateChange>>,
@@ -211,3 +261,19 @@ impl From<KeyProviderOptions> for sys_fc::ffi::KeyProviderOptions {
         }
     }
 }
+
+impl From<sys_fc::ffi::EncryptedPacket> for EncryptedPacket {
+    fn from(value: sys_fc::ffi::EncryptedPacket) -> Self {
+        Self {
+            data: value.data.into_iter().collect(),
+            iv: value.iv.into_iter().collect(),
+            key_index: value.key_index,
+        }
+    }
+}
+
+impl From<EncryptedPacket> for sys_fc::ffi::EncryptedPacket {
+    fn from(value: EncryptedPacket) -> Self {
+        Self { data: value.data, iv: value.iv, key_index: value.key_index }
+    }
+}

livekit-ffi/protocol/data_stream.proto

@@ -18,6 +18,7 @@ package livekit.proto;
 option csharp_namespace = "LiveKit.Proto";
 
 import "handle.proto";
+import "e2ee.proto";
 
 // MARK: - Text stream reader
 
@@ -329,6 +330,7 @@ message TextStreamInfo {
     optional string reply_to_stream_id = 9;  // Optional: Reply to specific message
     repeated string attached_stream_ids = 10; // file attachments for text streams
     optional bool generated = 11; // true if the text has been generated by an agent from a participant's audio transcription
+    required EncryptionType encryption_type = 12;
 }
 message ByteStreamInfo {
     required string stream_id = 1; // unique identifier for this data stream
@@ -339,6 +341,7 @@ message ByteStreamInfo {
     map<string, string> attributes = 6;  // user defined attributes map that can carry additional info
 
     required string name = 7;
+    required EncryptionType encryption_type = 8;
 }
 
 message StreamTextOptions {

livekit-ffi/protocol/room.proto

@@ -291,9 +291,10 @@ message RoomOptions {
   optional bool auto_subscribe = 1;
   optional bool adaptive_stream = 2;
   optional bool dynacast = 3;
-  optional E2eeOptions e2ee = 4;
+  optional E2eeOptions e2ee = 4 [deprecated=true];
   optional RtcConfig rtc_config = 5; // allow to setup a custom RtcConfiguration
   optional uint32 join_retries = 6;
+  optional E2eeOptions encryption = 7;
 }
 
 //
@@ -358,6 +359,7 @@ message RoomEvent {
     ParticipantMetadataChanged participant_metadata_changed = 17;
     ParticipantNameChanged participant_name_changed = 18;
     ParticipantAttributesChanged participant_attributes_changed = 19;
+    ParticipantEncryptionStatusChanged participant_encryption_status_changed = 39;
     ConnectionQualityChanged connection_quality_changed = 20;
     ConnectionStateChanged connection_state_changed = 21;
     // Connected connected = 21;
@@ -496,6 +498,11 @@ message ParticipantAttributesChanged {
   repeated AttributesEntry changed_attributes = 3;
 }
 
+message ParticipantEncryptionStatusChanged {
+  required string participant_identity = 1;
+  required bool is_encrypted = 2;
+}
+
 message ParticipantNameChanged {
   required string participant_identity = 1;
   required string name = 2;
@@ -697,4 +704,4 @@ message ByteStreamOpened {
 message TextStreamOpened {
   required OwnedTextStreamReader reader = 1;
   required string participant_identity = 2;
-}
+}

livekit-ffi/src/conversion/data_stream.rs

@@ -35,6 +35,7 @@ impl From<TextStreamInfo> for proto::TextStreamInfo {
             reply_to_stream_id: info.reply_to_stream_id,
             attached_stream_ids: info.attached_stream_ids,
             generated: Some(info.generated),
+            encryption_type: info.encryption_type.into(),
         }
     }
 }
@@ -49,6 +50,7 @@ impl From<ByteStreamInfo> for proto::ByteStreamInfo {
             total_length: info.total_length,
             attributes: info.attributes,
             name: info.name,
+            encryption_type: info.encryption_type.into(),
         }
     }
 }

livekit-ffi/src/conversion/room.rs

@@ -179,6 +179,21 @@ impl From<proto::RoomOptions> for RoomOptions {
             })
         });
 
+        let encryption = value.encryption.and_then(|opts| {
+            let encryption_type = opts.encryption_type();
+            let provider_opts = opts.key_provider_options;
+
+            Some(E2eeOptions {
+                encryption_type: encryption_type.into(),
+                key_provider: if provider_opts.shared_key.is_some() {
+                    let shared_key = provider_opts.shared_key.clone().unwrap();
+                    KeyProvider::with_shared_key(provider_opts.into(), shared_key)
+                } else {
+                    KeyProvider::new(provider_opts.into())
+                },
+            })
+        });
+
         let rtc_config =
             value.rtc_config.map(Into::into).unwrap_or(RoomOptions::default().rtc_config);
 
@@ -189,6 +204,7 @@ impl From<proto::RoomOptions> for RoomOptions {
         options.rtc_config = rtc_config;
         options.join_retries = value.join_retries.unwrap_or(options.join_retries);
         options.e2ee = e2ee;
+        options.encryption = encryption;
         options
     }
 }