use dashmap::{DashMap, DashSet};
use matrix_sdk_common::{api::r0::to_device::DeviceIdOrAllDevices, uuid::Uuid};
use std::{
cmp::min,
fmt,
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
Arc,
},
time::Duration,
};
use tracing::debug;
use matrix_sdk_common::{
events::{
room::{encrypted::EncryptedEventContent, encryption::EncryptionEventContent},
AnyMessageEventContent, EventContent,
},
identifiers::{DeviceId, DeviceIdBox, EventEncryptionAlgorithm, RoomId, UserId},
instant::Instant,
locks::Mutex,
};
use serde_json::{json, Value};
use olm_rs::outbound_group_session::OlmOutboundGroupSession;
pub use olm_rs::{
account::IdentityKeys,
session::{OlmMessage, PreKeyMessage},
utility::OlmUtility,
};
use crate::ToDeviceRequest;
use super::GroupSessionKey;
const ROTATION_PERIOD: Duration = Duration::from_millis(604800000);
const ROTATION_MESSAGES: u64 = 100;
#[derive(Debug)]
pub struct EncryptionSettings {
pub algorithm: EventEncryptionAlgorithm,
pub rotation_period: Duration,
pub rotation_period_msgs: u64,
}
impl Default for EncryptionSettings {
fn default() -> Self {
Self {
algorithm: EventEncryptionAlgorithm::MegolmV1AesSha2,
rotation_period: ROTATION_PERIOD,
rotation_period_msgs: ROTATION_MESSAGES,
}
}
}
impl From<&EncryptionEventContent> for EncryptionSettings {
fn from(content: &EncryptionEventContent) -> Self {
let rotation_period: Duration = content
.rotation_period_ms
.map_or(ROTATION_PERIOD, |r| Duration::from_millis(r.into()));
let rotation_period_msgs: u64 = content
.rotation_period_msgs
.map_or(ROTATION_MESSAGES, Into::into);
Self {
algorithm: content.algorithm.clone(),
rotation_period,
rotation_period_msgs,
}
}
}
#[derive(Clone)]
pub struct OutboundGroupSession {
inner: Arc<Mutex<OlmOutboundGroupSession>>,
device_id: Arc<DeviceIdBox>,
account_identity_keys: Arc<IdentityKeys>,
session_id: Arc<str>,
room_id: Arc<RoomId>,
pub(crate) creation_time: Arc<Instant>,
message_count: Arc<AtomicU64>,
shared: Arc<AtomicBool>,
invalidated: Arc<AtomicBool>,
settings: Arc<EncryptionSettings>,
shared_with_set: Arc<DashMap<UserId, DashSet<DeviceIdBox>>>,
to_share_with_set: Arc<DashMap<Uuid, Arc<ToDeviceRequest>>>,
}
impl OutboundGroupSession {
pub fn new(
device_id: Arc<DeviceIdBox>,
identity_keys: Arc<IdentityKeys>,
room_id: &RoomId,
settings: EncryptionSettings,
) -> Self {
let session = OlmOutboundGroupSession::new();
let session_id = session.session_id();
OutboundGroupSession {
inner: Arc::new(Mutex::new(session)),
room_id: Arc::new(room_id.to_owned()),
device_id,
account_identity_keys: identity_keys,
session_id: session_id.into(),
creation_time: Arc::new(Instant::now()),
message_count: Arc::new(AtomicU64::new(0)),
shared: Arc::new(AtomicBool::new(false)),
invalidated: Arc::new(AtomicBool::new(false)),
settings: Arc::new(settings),
shared_with_set: Arc::new(DashMap::new()),
to_share_with_set: Arc::new(DashMap::new()),
}
}
pub fn add_request(&self, request_id: Uuid, request: Arc<ToDeviceRequest>) {
self.to_share_with_set.insert(request_id, request);
}
pub fn add_recipient(&self, user_id: &UserId) {
self.shared_with_set
.entry(user_id.to_owned())
.or_insert_with(DashSet::new);
}
pub fn contains_recipient(&self, user_id: &UserId) -> bool {
self.shared_with_set.contains_key(user_id)
}
pub fn mark_request_as_sent(&self, request_id: &Uuid) {
if let Some((_, r)) = self.to_share_with_set.remove(request_id) {
let user_pairs = r.messages.iter().map(|(u, v)| {
(
u.clone(),
v.keys().filter_map(|d| {
if let DeviceIdOrAllDevices::DeviceId(d) = d {
Some(d.clone())
} else {
None
}
}),
)
});
user_pairs.for_each(|(u, d)| {
self.shared_with_set
.entry(u)
.or_insert_with(DashSet::new)
.extend(d);
});
if self.to_share_with_set.is_empty() {
debug!(
"Marking session {} for room {} as shared.",
self.session_id(),
self.room_id
);
self.mark_as_shared();
}
}
}
pub(crate) async fn encrypt_helper(&self, plaintext: String) -> String {
let session = self.inner.lock().await;
self.message_count.fetch_add(1, Ordering::SeqCst);
session.encrypt(plaintext)
}
pub async fn encrypt(&self, content: AnyMessageEventContent) -> EncryptedEventContent {
let json_content = json!({
"content": content,
"room_id": &*self.room_id,
"type": content.event_type(),
});
let plaintext = json_content.to_string();
let ciphertext = self.encrypt_helper(plaintext).await;
EncryptedEventContent::MegolmV1AesSha2(
matrix_sdk_common::events::room::encrypted::MegolmV1AesSha2ContentInit {
ciphertext,
sender_key: self.account_identity_keys.curve25519().to_owned(),
session_id: self.session_id().to_owned(),
device_id: (&*self.device_id).to_owned(),
}
.into(),
)
}
pub fn expired(&self) -> bool {
let count = self.message_count.load(Ordering::SeqCst);
count >= self.settings.rotation_period_msgs
|| self.creation_time.elapsed()
>= min(self.settings.rotation_period, Duration::from_secs(3600))
}
pub fn invalidated(&self) -> bool {
self.invalidated.load(Ordering::Relaxed)
}
pub fn mark_as_shared(&self) {
self.shared.store(true, Ordering::Relaxed);
}
pub fn shared(&self) -> bool {
self.shared.load(Ordering::Relaxed)
}
pub async fn session_key(&self) -> GroupSessionKey {
let session = self.inner.lock().await;
GroupSessionKey(session.session_key())
}
pub fn room_id(&self) -> &RoomId {
&self.room_id
}
pub fn session_id(&self) -> &str {
&self.session_id
}
pub async fn message_index(&self) -> u32 {
let session = self.inner.lock().await;
session.session_message_index()
}
pub async fn as_json(&self) -> Value {
json!({
"algorithm": EventEncryptionAlgorithm::MegolmV1AesSha2,
"room_id": &*self.room_id,
"session_id": &*self.session_id,
"session_key": self.session_key().await,
"chain_index": self.message_index().await,
})
}
pub fn invalidate_session(&self) {
self.invalidated.store(true, Ordering::Relaxed)
}
pub fn clear_requests(&self) -> Vec<Uuid> {
let request_ids = self
.to_share_with_set
.iter()
.map(|item| *item.key())
.collect();
self.to_share_with_set.clear();
request_ids
}
pub(crate) fn is_shared_with(&self, user_id: &UserId, device_id: &DeviceId) -> bool {
let shared_with = self
.shared_with_set
.get(user_id)
.map(|d| d.contains(device_id))
.unwrap_or(false);
let should_be_shared_with = if self.shared() {
false
} else {
let device_id = DeviceIdOrAllDevices::DeviceId(device_id.into());
self.to_share_with_set.iter().any(|item| {
if let Some(e) = item.value().messages.get(user_id) {
e.contains_key(&device_id)
} else {
false
}
})
};
shared_with || should_be_shared_with
}
#[cfg(test)]
pub fn mark_shared_with(&self, user_id: &UserId, device_id: &DeviceId) {
self.shared_with_set
.entry(user_id.to_owned())
.or_insert_with(DashSet::new)
.insert(device_id.to_owned());
}
}
#[cfg(not(tarpaulin_include))]
impl std::fmt::Debug for OutboundGroupSession {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("OutboundGroupSession")
.field("session_id", &self.session_id)
.field("room_id", &self.room_id)
.field("creation_time", &self.creation_time)
.field("message_count", &self.message_count)
.finish()
}
}
#[cfg(test)]
mod test {
use std::time::Duration;
use matrix_sdk_common::{
events::room::encryption::EncryptionEventContent, identifiers::EventEncryptionAlgorithm,
uint,
};
use super::{EncryptionSettings, ROTATION_MESSAGES, ROTATION_PERIOD};
#[test]
fn encryption_settings_conversion() {
let mut content = EncryptionEventContent::new(EventEncryptionAlgorithm::MegolmV1AesSha2);
let settings = EncryptionSettings::from(&content);
assert_eq!(settings.rotation_period, ROTATION_PERIOD);
assert_eq!(settings.rotation_period_msgs, ROTATION_MESSAGES);
content.rotation_period_ms = Some(uint!(3600));
content.rotation_period_msgs = Some(uint!(500));
let settings = EncryptionSettings::from(&content);
assert_eq!(settings.rotation_period, Duration::from_millis(3600));
assert_eq!(settings.rotation_period_msgs, 500);
}
}