diff --git a/src/service/server_keys/acquire.rs b/src/service/server_keys/acquire.rs index a14056257..4451f738d 100644 --- a/src/service/server_keys/acquire.rs +++ b/src/service/server_keys/acquire.rs @@ -4,9 +4,7 @@ use std::{ time::Duration, }; -use conduwuit::{ - debug, debug_error, debug_warn, error, implement, info, result::FlatOk, trace, warn, -}; +use conduwuit::{debug, debug_error, debug_warn, error, info, result::FlatOk, trace, warn}; use futures::{StreamExt, stream::FuturesUnordered}; use ruma::{ CanonicalJsonObject, OwnedServerName, OwnedServerSigningKeyId, ServerName, @@ -19,218 +17,228 @@ use super::key_exists; type Batch = BTreeMap>; -#[implement(super::Service)] -pub async fn acquire_events_pubkeys<'a, I>(&self, events: I) -where - I: Iterator> + Send, -{ - type Batch = BTreeMap>; - type Signatures = BTreeMap>; +impl super::Service { + /// Returns all verify keys required to verify the incoming events. + pub async fn acquire_events_pubkeys<'a, I>(&self, events: I) + where + I: Iterator> + Send, + { + type Batch = BTreeMap>; + type Signatures = BTreeMap>; - let mut batch = Batch::new(); - events - .cloned() - .map(Raw::::from_json) - .map(|event| event.get_field::("signatures")) - .filter_map(FlatOk::flat_ok) - .flat_map(IntoIterator::into_iter) - .for_each(|(server, sigs)| { - batch.entry(server).or_default().extend(sigs.into_keys()); - }); + let mut batch = Batch::new(); + events + .cloned() + .map(Raw::::from_json) + .map(|event| event.get_field::("signatures")) + .filter_map(FlatOk::flat_ok) + .flat_map(IntoIterator::into_iter) + .for_each(|(server, sigs)| { + batch.entry(server).or_default().extend(sigs.into_keys()); + }); - let batch = batch - .iter() - .map(|(server, keys)| (server.borrow(), keys.iter().map(Borrow::borrow))); - - self.acquire_pubkeys(batch).await; -} - -#[implement(super::Service)] -pub async fn acquire_pubkeys<'a, S, K>(&self, batch: S) -where - S: Iterator + Send + Clone, - K: Iterator + Send + Clone, -{ - let notary_only = self.services.server.config.only_query_trusted_key_servers; - let notary_first_always = self.services.server.config.query_trusted_key_servers_first; - let notary_first_on_join = self - .services - .server - .config - .query_trusted_key_servers_first_on_join; - - let requested_servers = batch.clone().count(); - let requested_keys = batch.clone().flat_map(|(_, key_ids)| key_ids).count(); - - debug!("acquire {requested_keys} keys from {requested_servers}"); - - let mut missing = self.acquire_locals(batch).await; - let mut missing_keys = keys_count(&missing); - let mut missing_servers = missing.len(); - if missing_servers == 0 { - return; - } - - info!("{missing_keys} keys for {missing_servers} servers will be acquired"); - - if notary_first_always || notary_first_on_join { - missing = self.acquire_notary(missing.into_iter()).await; - missing_keys = keys_count(&missing); - missing_servers = missing.len(); - if missing_keys == 0 { - return; - } - - warn!( - "missing {missing_keys} keys for {missing_servers} servers from all notaries first" - ); - } - - if !notary_only { - missing = self.acquire_origins(missing.into_iter()).await; - missing_keys = keys_count(&missing); - missing_servers = missing.len(); - if missing_keys == 0 { - return; - } - - debug_warn!("missing {missing_keys} keys for {missing_servers} servers unreachable"); - } - - if !notary_first_always && !notary_first_on_join { - missing = self.acquire_notary(missing.into_iter()).await; - missing_keys = keys_count(&missing); - missing_servers = missing.len(); - if missing_keys == 0 { - return; - } - - debug_warn!( - "still missing {missing_keys} keys for {missing_servers} servers from all notaries." - ); - } - - if missing_keys > 0 { - warn!( - "did not obtain {missing_keys} keys for {missing_servers} servers out of \ - {requested_keys} total keys for {requested_servers} total servers." - ); - } - - for (server, key_ids) in missing { - debug_warn!(%server, ?key_ids, "missing"); - } -} - -#[implement(super::Service)] -async fn acquire_locals<'a, S, K>(&self, batch: S) -> Batch -where - S: Iterator + Send, - K: Iterator + Send, -{ - let mut missing = Batch::new(); - for (server, key_ids) in batch { - for key_id in key_ids { - if !self.verify_key_exists(server, key_id).await { - missing - .entry(server.into()) - .or_default() - .push(key_id.into()); - } - } - } - - missing -} - -#[implement(super::Service)] -async fn acquire_origins(&self, batch: I) -> Batch -where - I: Iterator)> + Send, -{ - let timeout = Instant::now() - .checked_add(Duration::from_secs(45)) - .expect("timeout overflows"); - - let mut requests: FuturesUnordered<_> = batch - .map(|(origin, key_ids)| self.acquire_origin(origin, key_ids, timeout)) - .collect(); - - let mut missing = Batch::new(); - while let Some((origin, key_ids)) = requests.next().await { - if !key_ids.is_empty() { - missing.insert(origin, key_ids); - } - } - - missing -} - -#[implement(super::Service)] -async fn acquire_origin( - &self, - origin: OwnedServerName, - mut key_ids: Vec, - timeout: Instant, -) -> (OwnedServerName, Vec) { - match timeout_at(timeout, self.server_request(&origin)).await { - | Err(e) => debug_warn!(%origin, "timed out: {e}"), - | Ok(Err(e)) => debug_error!(%origin, "{e}"), - | Ok(Ok(server_keys)) => { - trace!( - %origin, - ?key_ids, - ?server_keys, - "received server_keys" - ); - - self.add_signing_keys(server_keys.clone()).await; - key_ids.retain(|key_id| !key_exists(&server_keys, key_id)); - }, - } - - (origin, key_ids) -} - -#[implement(super::Service)] -async fn acquire_notary(&self, batch: I) -> Batch -where - I: Iterator)> + Send, -{ - let mut missing: Batch = batch.collect(); - for notary in self.services.globals.trusted_servers() { - let missing_keys = keys_count(&missing); - let missing_servers = missing.len(); - debug!( - "Asking notary {notary} for {missing_keys} missing keys from {missing_servers} \ - servers" - ); - - let batch = missing + let batch = batch .iter() .map(|(server, keys)| (server.borrow(), keys.iter().map(Borrow::borrow))); - match self.batch_notary_request(notary, batch).await { - | Err(e) => error!("Failed to contact notary {notary:?}: {e}"), - | Ok(results) => - for server_keys in results { - self.acquire_notary_result(&mut missing, server_keys).await; - }, + self.acquire_pubkeys(batch).await; + } + + /// Acquires all the verify keys required in the provided batch, using + /// notaries and direct-fetch where applicable. + pub async fn acquire_pubkeys<'a, S, K>(&self, batch: S) + where + S: Iterator + Send + Clone, + K: Iterator + Send + Clone, + { + let notary_only = self.services.server.config.only_query_trusted_key_servers; + let notary_first_always = self.services.server.config.query_trusted_key_servers_first; + let notary_first_on_join = self + .services + .server + .config + .query_trusted_key_servers_first_on_join; + + let requested_servers = batch.clone().count(); + let requested_keys = batch.clone().flat_map(|(_, key_ids)| key_ids).count(); + + debug!("acquire {requested_keys} keys from {requested_servers}"); + + let mut missing = self.acquire_locals(batch).await; + let mut missing_keys = keys_count(&missing); + let mut missing_servers = missing.len(); + if missing_servers == 0 { + return; + } + + info!("{missing_keys} keys for {missing_servers} servers will be acquired"); + + if notary_first_always || notary_first_on_join { + missing = self.acquire_notary(missing.into_iter()).await; + missing_keys = keys_count(&missing); + missing_servers = missing.len(); + if missing_keys == 0 { + return; + } + + warn!( + "missing {missing_keys} keys for {missing_servers} servers from all notaries \ + first" + ); + } + + if !notary_only { + missing = self.acquire_origins(missing.into_iter()).await; + missing_keys = keys_count(&missing); + missing_servers = missing.len(); + if missing_keys == 0 { + return; + } + + debug_warn!("missing {missing_keys} keys for {missing_servers} servers unreachable"); + } + + if !notary_first_always && !notary_first_on_join { + missing = self.acquire_notary(missing.into_iter()).await; + missing_keys = keys_count(&missing); + missing_servers = missing.len(); + if missing_keys == 0 { + return; + } + + debug_warn!( + "still missing {missing_keys} keys for {missing_servers} servers from all \ + notaries." + ); + } + + if missing_keys > 0 { + warn!( + "did not obtain {missing_keys} keys for {missing_servers} servers out of \ + {requested_keys} total keys for {requested_servers} total servers." + ); + } + + for (server, key_ids) in missing { + debug_warn!(%server, ?key_ids, "missing"); } } - missing -} + /// Filters the batch to return only keys which we do not already have + /// locally. + async fn acquire_locals<'a, S, K>(&self, batch: S) -> Batch + where + S: Iterator + Send, + K: Iterator + Send, + { + let mut missing = Batch::new(); + for (server, key_ids) in batch { + for key_id in key_ids { + if !self.verify_key_exists(server, key_id).await { + missing + .entry(server.into()) + .or_default() + .push(key_id.into()); + } + } + } -#[implement(super::Service)] -async fn acquire_notary_result(&self, missing: &mut Batch, server_keys: ServerSigningKeys) { - let server = &server_keys.server_name; - self.add_signing_keys(server_keys.clone()).await; + missing + } - if let Some(key_ids) = missing.get_mut(server) { - key_ids.retain(|key_id| !key_exists(&server_keys, key_id)); - if key_ids.is_empty() { - missing.remove(server); + /// Fetches verification keys from the origin servers directly, returning + /// any keys that could not be fetched. + async fn acquire_origins(&self, batch: I) -> Batch + where + I: Iterator)> + Send, + { + let timeout = Instant::now() + .checked_add(Duration::from_secs(45)) + .expect("timeout overflows"); + + let mut requests: FuturesUnordered<_> = batch + .map(|(origin, key_ids)| self.acquire_origin(origin, key_ids, timeout)) + .collect(); + + let mut missing = Batch::new(); + while let Some((origin, key_ids)) = requests.next().await { + if !key_ids.is_empty() { + missing.insert(origin, key_ids); + } + } + + missing + } + + /// Fetches verification keys from an origin directly, returning any keys + /// that were fetched. + async fn acquire_origin( + &self, + origin: OwnedServerName, + mut key_ids: Vec, + timeout: Instant, + ) -> (OwnedServerName, Vec) { + match timeout_at(timeout, self.server_request(&origin)).await { + | Err(e) => debug_warn!(%origin, "timed out: {e}"), + | Ok(Err(e)) => debug_error!(%origin, "{e}"), + | Ok(Ok(server_keys)) => { + trace!( + %origin, + ?key_ids, + ?server_keys, + "received server_keys" + ); + + self.add_signing_keys(server_keys.clone()).await; + key_ids.retain(|key_id| !key_exists(&server_keys, key_id)); + }, + } + + (origin, key_ids) + } + + /// Fetches signing keys for the given batch from the configured notary + /// servers, returning any keys that could not be fetched. + async fn acquire_notary(&self, batch: I) -> Batch + where + I: Iterator)> + Send, + { + let mut missing: Batch = batch.collect(); + for notary in self.services.globals.trusted_servers() { + let missing_keys = keys_count(&missing); + let missing_servers = missing.len(); + debug!( + "Asking notary {notary} for {missing_keys} missing keys from {missing_servers} \ + servers" + ); + + let batch = missing + .iter() + .map(|(server, keys)| (server.borrow(), keys.iter().map(Borrow::borrow))); + + match self.batch_notary_request(notary, batch).await { + | Err(e) => error!("Failed to contact notary {notary:?}: {e}"), + | Ok(results) => + for server_keys in results { + self.acquire_notary_result(&mut missing, server_keys).await; + }, + } + } + + missing + } + + /// Contacts a notary and fetches signing keys for the given batch, removing + /// keys from the missing batch when they are found. + async fn acquire_notary_result(&self, missing: &mut Batch, server_keys: ServerSigningKeys) { + let server = &server_keys.server_name; + self.add_signing_keys(server_keys.clone()).await; + + if let Some(key_ids) = missing.get_mut(server) { + key_ids.retain(|key_id| !key_exists(&server_keys, key_id)); + if key_ids.is_empty() { + missing.remove(server); + } } } } diff --git a/src/service/server_keys/get.rs b/src/service/server_keys/get.rs index 7d334eb98..a05340704 100644 --- a/src/service/server_keys/get.rs +++ b/src/service/server_keys/get.rs @@ -1,6 +1,6 @@ use std::borrow::Borrow; -use conduwuit::{Err, Result, debug_error, implement, trace}; +use conduwuit::{Err, Result, trace}; use ruma::{ CanonicalJsonObject, ServerName, ServerSigningKeyId, api::federation::discovery::VerifyKey, room_version_rules::RoomVersionRules, @@ -9,133 +9,138 @@ use ruma::{ use super::{PubKeyMap, PubKeys, extract_key}; use crate::server_keys::util::required_keys; -#[implement(super::Service)] -pub async fn get_event_keys( - &self, - object: &CanonicalJsonObject, - room_version_rules: &RoomVersionRules, -) -> Result { - let required = match required_keys(object, &room_version_rules.signatures) { - | Ok(required) => required, - | Err(e) => { - debug_error!("Failed to determine keys required to verify: {e}"); - return Err!(BadServerResponse("Failed to determine keys required to verify: {e}")); - }, - }; - trace!(?required, "Keys required to verify event"); +impl super::Service { + /// Fetches the verify keys required to verify the incoming event. + pub async fn get_event_keys( + &self, + object: &CanonicalJsonObject, + room_version_rules: &RoomVersionRules, + ) -> Result { + let required = match required_keys(object, &room_version_rules.signatures) { + | Ok(required) => required, + | Err(e) => { + return Err!(BadServerResponse( + "Failed to determine keys required to verify event: {e}" + )); + }, + }; - let batch = required - .iter() - .map(|(s, ids)| (s.borrow(), ids.iter().map(Borrow::borrow))); + let batch = required + .iter() + .map(|(s, ids)| (s.borrow(), ids.iter().map(Borrow::borrow))); - Ok(self.get_pubkeys(batch).await) -} - -#[implement(super::Service)] -pub async fn get_pubkeys<'a, S, K>(&self, batch: S) -> PubKeyMap -where - S: Iterator + Send, - K: Iterator + Send, -{ - let mut keys = PubKeyMap::new(); - for (server, key_ids) in batch { - let pubkeys = self.get_pubkeys_for(server, key_ids).await; - keys.insert(server.into(), pubkeys); + Ok(self.get_pubkeys(batch).await) } - keys -} - -#[implement(super::Service)] -pub async fn get_pubkeys_for<'a, I>(&self, origin: &ServerName, key_ids: I) -> PubKeys -where - I: Iterator + Send, -{ - let mut keys = PubKeys::new(); - for key_id in key_ids { - if let Ok(verify_key) = self.get_verify_key(origin, key_id).await { - keys.insert(key_id.into(), verify_key.key); + /// Fetches all verify keys for the given batch. Any keys which could not be + /// fetched are elided from the resulting map. + pub async fn get_pubkeys<'a, S, K>(&self, batch: S) -> PubKeyMap + where + S: Iterator + Send, + K: Iterator + Send, + { + let mut keys = PubKeyMap::new(); + for (server, key_ids) in batch { + let pubkeys = self.get_pubkeys_for(server, key_ids).await; + keys.insert(server.into(), pubkeys); } + + keys } - keys -} - -#[implement(super::Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn get_verify_key( - &self, - origin: &ServerName, - key_id: &ServerSigningKeyId, -) -> Result { - let notary_first = self.services.server.config.query_trusted_key_servers_first; - let notary_only = self.services.server.config.only_query_trusted_key_servers; - - if let Some(result) = self.verify_keys_for(origin).await.remove(key_id) { - trace!("Found key in cache"); - return Ok(result); - } - - if notary_first { - if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await { - return Ok(result); - } - } - - if !notary_only { - if let Ok(result) = self.get_verify_key_from_origin(origin, key_id).await { - return Ok(result); - } - } - - if !notary_first { - if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await { - return Ok(result); - } - } - - Err!(BadServerResponse(debug_error!( - %key_id, - %origin, - "Failed to fetch federation signing-key" - ))) -} - -#[implement(super::Service)] -async fn get_verify_key_from_notaries( - &self, - origin: &ServerName, - key_id: &ServerSigningKeyId, -) -> Result { - for notary in self.services.globals.trusted_servers() { - if let Ok(server_keys) = self.notary_request(notary, origin).await { - for server_key in server_keys.clone() { - self.add_signing_keys(server_key).await; + /// Gets verify keys for the given server name and a set of key IDs. Any + /// keys that cannot be found are elided from the resulting set. + pub async fn get_pubkeys_for<'a, I>(&self, origin: &ServerName, key_ids: I) -> PubKeys + where + I: Iterator + Send, + { + let mut keys = PubKeys::new(); + for key_id in key_ids { + if let Ok(verify_key) = self.get_verify_key(origin, key_id).await { + keys.insert(key_id.into(), verify_key.key); } + } - for server_key in server_keys { - if let Some(result) = extract_key(server_key, key_id) { - return Ok(result); + keys + } + + /// Fetches a single verify key via origin and notaries. + pub async fn get_verify_key( + &self, + origin: &ServerName, + key_id: &ServerSigningKeyId, + ) -> Result { + let notary_first = self.services.server.config.query_trusted_key_servers_first; + let notary_only = self.services.server.config.only_query_trusted_key_servers; + + if let Some(result) = self.verify_keys_for(origin).await.remove(key_id) { + trace!("Found key in cache"); + return Ok(result); + } + + if notary_first { + if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await { + return Ok(result); + } + } + + if !notary_only { + if let Ok(result) = self.get_verify_key_from_origin(origin, key_id).await { + return Ok(result); + } + } + + if !notary_first { + if let Ok(result) = self.get_verify_key_from_notaries(origin, key_id).await { + return Ok(result); + } + } + + Err!(BadServerResponse(debug_error!( + %key_id, + %origin, + "Failed to fetch federation signing-key" + ))) + } + + /// Fetches verify keys from notary servers, adding them to the cache, and + /// returning the requested key if found. + async fn get_verify_key_from_notaries( + &self, + origin: &ServerName, + key_id: &ServerSigningKeyId, + ) -> Result { + for notary in self.services.globals.trusted_servers() { + if let Ok(server_keys) = self.notary_request(notary, origin).await { + for server_key in server_keys.clone() { + self.add_signing_keys(server_key).await; + } + + for server_key in server_keys { + if let Some(result) = extract_key(server_key, key_id) { + return Ok(result); + } } } } + + Err!(Request(NotFound("Failed to fetch signing-key from notaries"))) } - Err!(Request(NotFound("Failed to fetch signing-key from notaries"))) -} - -#[implement(super::Service)] -async fn get_verify_key_from_origin( - &self, - origin: &ServerName, - key_id: &ServerSigningKeyId, -) -> Result { - if let Ok(server_key) = self.server_request(origin).await { - self.add_signing_keys(server_key.clone()).await; - if let Some(result) = extract_key(server_key, key_id) { - return Ok(result); + /// Fetches verify keys from the origin server, adding them to the cache, + /// and returning the requested key if found. + async fn get_verify_key_from_origin( + &self, + origin: &ServerName, + key_id: &ServerSigningKeyId, + ) -> Result { + if let Ok(server_key) = self.server_request(origin).await { + self.add_signing_keys(server_key.clone()).await; + if let Some(result) = extract_key(server_key, key_id) { + return Ok(result); + } } - } - Err!(Request(NotFound("Failed to fetch signing-key from origin"))) + Err!(Request(NotFound("Failed to fetch signing-key from origin"))) + } } diff --git a/src/service/server_keys/mod.rs b/src/service/server_keys/mod.rs index 5df3e5193..b79b2b10d 100644 --- a/src/service/server_keys/mod.rs +++ b/src/service/server_keys/mod.rs @@ -9,7 +9,7 @@ mod verify; use std::{collections::BTreeMap, sync::Arc, time::Duration}; use conduwuit::{ - Result, Server, debug_error, debug_warn, implement, trace, + Result, Server, utils::{IterStream, timepoint_from_now}, }; use database::{Deserialized, Json, Map}; @@ -73,123 +73,124 @@ impl crate::Service for Service { fn name(&self) -> &str { crate::service::make_name(std::module_path!()) } } -#[implement(Service)] -#[inline] -pub fn keypair(&self) -> &Ed25519KeyPair { &self.keypair } +impl Service { + /// Fetches the server's active keypair. + pub fn keypair(&self) -> &Ed25519KeyPair { &self.keypair } -#[implement(Service)] -#[inline] -pub fn active_key_id(&self) -> &ServerSigningKeyId { self.active_verify_key().0 } + /// Fetches the server's active signing key ID. + pub fn active_key_id(&self) -> &ServerSigningKeyId { self.active_verify_key().0 } -#[implement(Service)] -#[inline] -pub fn active_verify_key(&self) -> (&ServerSigningKeyId, &VerifyKey) { - debug_assert!(self.verify_keys.len() <= 1, "more than one active verify_key"); - self.verify_keys - .iter() - .next() - .map(|(id, key)| (id.as_ref(), key)) - .expect("missing active verify_key") -} + /// Fetches the server's active signing key. Panics if there's more than 1 + /// active key. + pub fn active_verify_key(&self) -> (&ServerSigningKeyId, &VerifyKey) { + debug_assert!(self.verify_keys.len() <= 1, "more than one active verify_key"); + self.verify_keys + .iter() + .next() + .map(|(id, key)| (id.as_ref(), key)) + .expect("missing active verify_key") + } -#[implement(Service)] -async fn add_signing_keys(&self, new_keys: ServerSigningKeys) { - let origin = &new_keys.server_name; + /// Combines signing keys from the incoming response to the existing + /// database set. + async fn add_signing_keys(&self, new_keys: ServerSigningKeys) { + let origin = &new_keys.server_name; - // (timo) Not atomic, but this is not critical - let mut keys: ServerSigningKeys = self - .db - .server_signingkeys - .get(origin) - .await - .deserialized() - .unwrap_or_else(|_| { - // Just insert "now", it doesn't matter - ServerSigningKeys::new(origin.to_owned(), MilliSecondsSinceUnixEpoch::now()) - }); + // (timo) Not atomic, but this is not critical + let mut keys: ServerSigningKeys = self + .db + .server_signingkeys + .get(origin) + .await + .deserialized() + .unwrap_or_else(|_| { + // Just insert "now", it doesn't matter + ServerSigningKeys::new(origin.to_owned(), MilliSecondsSinceUnixEpoch::now()) + }); - keys.verify_keys.extend(new_keys.verify_keys); - keys.old_verify_keys.extend(new_keys.old_verify_keys); - self.db.server_signingkeys.raw_put(origin, Json(&keys)); -} + keys.verify_keys.extend(new_keys.verify_keys); + keys.old_verify_keys.extend(new_keys.old_verify_keys); + self.db.server_signingkeys.raw_put(origin, Json(&keys)); + } -#[implement(Service)] -#[tracing::instrument(skip(self, object), level = "debug")] -pub async fn required_keys_exist( - &self, - object: &CanonicalJsonObject, - room_version_rules: &RoomVersionRules, -) -> bool { - trace!(?object, "Checking required keys exist"); - let Ok(required_keys) = required_keys(object, &room_version_rules.signatures) else { - debug_error!("Failed to determine required keys"); - return false; - }; - trace!(?required_keys, "Required keys to verify event"); - required_keys - .iter() - .flat_map(|(server, key_ids)| key_ids.iter().map(move |key_id| (server, key_id))) - .stream() - .all(|(server, key_id)| self.verify_key_exists(server, key_id)) - .await -} + /// Checks if the keys required to verify an incoming PDU are already known + /// locally. + pub async fn required_keys_exist( + &self, + object: &CanonicalJsonObject, + room_version_rules: &RoomVersionRules, + ) -> bool { + let Ok(required_keys) = required_keys(object, &room_version_rules.signatures) else { + return false; + }; + required_keys + .iter() + .flat_map(|(server, key_ids)| key_ids.iter().map(move |key_id| (server, key_id))) + .stream() + .all(|(server, key_id)| self.verify_key_exists(server, key_id)) + .await + } -#[implement(Service)] -#[tracing::instrument(skip(self), level = "debug")] -pub async fn verify_key_exists(&self, origin: &ServerName, key_id: &ServerSigningKeyId) -> bool { - type KeysMap = BTreeMap>; + /// Checks if a single verify key belonging to the target server is already + /// known locally. + pub async fn verify_key_exists( + &self, + origin: &ServerName, + key_id: &ServerSigningKeyId, + ) -> bool { + type KeysMap = BTreeMap>; - let Ok(keys) = self - .db - .server_signingkeys - .get(origin) - .await - .deserialized::>() - else { - debug_warn!("No known signing keys found for {origin}"); - return false; - }; + let Ok(keys) = self + .db + .server_signingkeys + .get(origin) + .await + .deserialized::>() + else { + return false; + }; - if let Ok(Some(verify_keys)) = keys.get_field::("verify_keys") { - if verify_keys.contains_key(key_id) { - return true; + if let Ok(Some(verify_keys)) = keys.get_field::("verify_keys") { + if verify_keys.contains_key(key_id) { + return true; + } } - } - if let Ok(Some(old_verify_keys)) = keys.get_field::("old_verify_keys") { - if old_verify_keys.contains_key(key_id) { - return true; + if let Ok(Some(old_verify_keys)) = keys.get_field::("old_verify_keys") { + if old_verify_keys.contains_key(key_id) { + return true; + } } + + false } - debug_warn!("Key {key_id} not found for {origin}"); - false -} + /// Returns all verify keys for the origin. + pub async fn verify_keys_for(&self, origin: &ServerName) -> VerifyKeys { + let mut keys = self + .signing_keys_for(origin) + .await + .map_or(BTreeMap::new(), |keys| merge_old_keys(keys).verify_keys); -#[implement(Service)] -pub async fn verify_keys_for(&self, origin: &ServerName) -> VerifyKeys { - let mut keys = self - .signing_keys_for(origin) - .await - .map_or(BTreeMap::new(), |keys| merge_old_keys(keys).verify_keys); + if self.services.globals.server_is_ours(origin) { + keys.extend(self.verify_keys.clone()); + } - if self.services.globals.server_is_ours(origin) { - keys.extend(self.verify_keys.clone()); + keys } - keys -} + /// Returns the stored server signing keys response for the origin. + pub async fn signing_keys_for(&self, origin: &ServerName) -> Result { + self.db.server_signingkeys.get(origin).await.deserialized() + } -#[implement(Service)] -pub async fn signing_keys_for(&self, origin: &ServerName) -> Result { - self.db.server_signingkeys.get(origin).await.deserialized() -} - -#[implement(Service)] -fn minimum_valid_ts(&self) -> MilliSecondsSinceUnixEpoch { - let timepoint = - timepoint_from_now(self.minimum_valid).expect("SystemTime should not overflow"); - MilliSecondsSinceUnixEpoch::from_system_time(timepoint).expect("UInt should not overflow") + /// Returns the minimum_valid_ts for a key response from the current point + /// in time, plus the minimum valid time. + fn minimum_valid_ts(&self) -> MilliSecondsSinceUnixEpoch { + let timepoint = + timepoint_from_now(self.minimum_valid).expect("SystemTime should not overflow"); + MilliSecondsSinceUnixEpoch::from_system_time(timepoint).expect("UInt should not overflow") + } } fn merge_old_keys(mut keys: ServerSigningKeys) -> ServerSigningKeys { diff --git a/src/service/server_keys/request.rs b/src/service/server_keys/request.rs index c8823fdc7..043e894a9 100644 --- a/src/service/server_keys/request.rs +++ b/src/service/server_keys/request.rs @@ -1,6 +1,7 @@ use std::{collections::BTreeMap, fmt::Debug}; -use conduwuit::{Err, Result, debug, implement}; +use conduwuit::{Err, Result, debug}; +use get_remote_server_keys_batch::v2::Request; use ruma::{ OwnedServerName, OwnedServerSigningKeyId, ServerName, ServerSigningKeyId, api::federation::discovery::{ @@ -9,51 +10,79 @@ use ruma::{ get_server_keys, }, }; +type RumaBatch = BTreeMap>; -#[implement(super::Service)] -pub(super) async fn batch_notary_request<'a, S, K>( - &self, - notary: &ServerName, - batch: S, -) -> Result> -where - S: Iterator + Send, - K: Iterator + Send, -{ - use get_remote_server_keys_batch::v2::Request; - type RumaBatch = BTreeMap>; - - let mut criteria = QueryCriteria::new(); - criteria.minimum_valid_until_ts = Some(self.minimum_valid_ts()); - - let mut server_keys = batch.fold(RumaBatch::new(), |mut batch, (server, key_ids)| { - batch - .entry(server.into()) - .or_default() - .extend(key_ids.map(|key_id| (key_id.into(), criteria.clone()))); - - batch - }); - - debug_assert!(!server_keys.is_empty(), "empty batch request to notary"); - - let mut results = Vec::new(); - while let Some(batch) = server_keys - .keys() - .rev() - .take(self.services.server.config.trusted_server_batch_size) - .next_back() - .cloned() +impl super::Service { + /// Handles a batched notary request. + /// If the batch is empty, an empty vec is returned. + pub(super) async fn batch_notary_request<'a, S, K>( + &self, + notary: &ServerName, + batch: S, + ) -> Result> + where + S: Iterator + Send, + K: Iterator + Send, { - let request = Request::new(server_keys.split_off(&batch)); + let mut criteria = QueryCriteria::new(); + criteria.minimum_valid_until_ts = Some(self.minimum_valid_ts()); - debug!( - ?notary, - ?batch, - remaining = %server_keys.len(), - requesting = ?request.server_keys.keys(), - "notary request" - ); + let mut server_keys = batch.fold(RumaBatch::new(), |mut batch, (server, key_ids)| { + batch + .entry(server.into()) + .or_default() + .extend(key_ids.map(|key_id| (key_id.into(), criteria.clone()))); + + batch + }); + + if server_keys.is_empty() { + return Ok(vec![]); + } + + let mut results = Vec::new(); + while let Some(batch) = server_keys + .keys() + .rev() + .take(self.services.server.config.trusted_server_batch_size) + .next_back() + .cloned() + { + let request = Request::new(server_keys.split_off(&batch)); + + debug!( + ?notary, + ?batch, + remaining = %server_keys.len(), + requesting = ?request.server_keys.keys(), + "notary request" + ); + + let response = self + .services + .sending + .send_unauthenticated_request(notary, request) + .await? + .server_keys + .into_iter() + .map(|key| key.deserialize()) + .filter_map(Result::ok); + + results.extend(response); + } + + Ok(results) + } + + /// Makes a request to a notary server to fetch signing keys. + pub async fn notary_request( + &self, + notary: &ServerName, + target: &ServerName, + ) -> Result + Clone + Debug + Send + use<>> { + use get_remote_server_keys::v2::Request; + + let request = Request::new(target.into(), self.minimum_valid_ts()); let response = self .services @@ -65,54 +94,29 @@ where .map(|key| key.deserialize()) .filter_map(Result::ok); - results.extend(response); + Ok(response) } - Ok(results) -} + /// Makes a request to an origin server to fetch signing keys. + pub async fn server_request(&self, target: &ServerName) -> Result { + use get_server_keys::v2::Request; -#[implement(super::Service)] -pub async fn notary_request( - &self, - notary: &ServerName, - target: &ServerName, -) -> Result + Clone + Debug + Send + use<>> { - use get_remote_server_keys::v2::Request; + let server_signing_key = self + .services + .sending + .send_unauthenticated_request(target, Request::new()) + .await + .map(|response| response.server_key) + .and_then(|key| key.deserialize().map_err(Into::into))?; - let request = Request::new(target.into(), self.minimum_valid_ts()); + if server_signing_key.server_name != target { + return Err!(BadServerResponse(debug_warn!( + requested = ?target, + response = ?server_signing_key.server_name, + "Server responded with bogus server_name" + ))); + } - let response = self - .services - .sending - .send_unauthenticated_request(notary, request) - .await? - .server_keys - .into_iter() - .map(|key| key.deserialize()) - .filter_map(Result::ok); - - Ok(response) -} - -#[implement(super::Service)] -pub async fn server_request(&self, target: &ServerName) -> Result { - use get_server_keys::v2::Request; - - let server_signing_key = self - .services - .sending - .send_unauthenticated_request(target, Request::new()) - .await - .map(|response| response.server_key) - .and_then(|key| key.deserialize().map_err(Into::into))?; - - if server_signing_key.server_name != target { - return Err!(BadServerResponse(debug_warn!( - requested = ?target, - response = ?server_signing_key.server_name, - "Server responded with bogus server_name" - ))); + Ok(server_signing_key) } - - Ok(server_signing_key) } diff --git a/src/service/server_keys/sign.rs b/src/service/server_keys/sign.rs index c28fcd26b..405c955e2 100644 --- a/src/service/server_keys/sign.rs +++ b/src/service/server_keys/sign.rs @@ -1,23 +1,29 @@ -use conduwuit::{Result, implement}; -use ruma::{CanonicalJsonObject, room_version_rules::RoomVersionRules}; +use conduwuit::Result; +use ruma::{ + CanonicalJsonObject, room_version_rules::RoomVersionRules, signatures::hash_and_sign_event, +}; -#[implement(super::Service)] -pub fn sign_json(&self, object: &mut CanonicalJsonObject) -> Result { - use ruma::signatures::sign_json; +impl super::Service { + /// Signs an arbitrary JSON object. The object is mutated. + pub fn sign_json(&self, object: &mut CanonicalJsonObject) -> Result { + use ruma::signatures::sign_json; - let server_name = self.services.globals.server_name().as_str(); - sign_json(server_name, self.keypair(), object).map_err(Into::into) -} + let server_name = self.services.globals.server_name().as_str(); + sign_json(server_name, self.keypair(), object).map_err(Into::into) + } -#[implement(super::Service)] -pub fn hash_and_sign_event( - &self, - object: &mut CanonicalJsonObject, - room_version_rules: &RoomVersionRules, -) -> Result { - use ruma::signatures::hash_and_sign_event; - - let server_name = self.services.globals.server_name().as_str(); - hash_and_sign_event(server_name, self.keypair(), object, &room_version_rules.redaction) + /// Hashes and signs an event JSON object. The object is mutated. + pub fn hash_and_sign_event( + &self, + object: &mut CanonicalJsonObject, + room_version_rules: &RoomVersionRules, + ) -> Result { + hash_and_sign_event( + self.services.globals.server_name().as_str(), + self.keypair(), + object, + &room_version_rules.redaction, + ) .map_err(Into::into) + } } diff --git a/src/service/server_keys/verify.rs b/src/service/server_keys/verify.rs index a109b0ff5..8d28e10df 100644 --- a/src/service/server_keys/verify.rs +++ b/src/service/server_keys/verify.rs @@ -1,77 +1,73 @@ -use conduwuit::{ - Err, Result, debug_warn, implement, matrix::event::gen_event_id_canonical_json, trace, -}; +use conduwuit::{Err, Result, matrix::event::gen_event_id_canonical_json}; use ruma::{ CanonicalJsonObject, CanonicalJsonValue, OwnedEventId, room_version_rules::RoomVersionRules, signatures::Verified, }; use serde_json::value::RawValue as RawJsonValue; -#[implement(super::Service)] -pub async fn validate_and_add_event_id( - &self, - pdu: &RawJsonValue, - room_version_rules: &RoomVersionRules, -) -> Result<(OwnedEventId, CanonicalJsonObject)> { - let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?; - if let Err(e) = self.verify_event(&value, room_version_rules).await { - return Err!(BadServerResponse(debug_error!( - "Event {event_id} failed verification: {e:?}" - ))); +impl super::Service { + /// Validates the incoming event, and then inserts the calculated event ID + /// into the event_id field. + pub async fn validate_and_add_event_id( + &self, + pdu: &RawJsonValue, + room_version_rules: &RoomVersionRules, + ) -> Result<(OwnedEventId, CanonicalJsonObject)> { + let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?; + if let Err(e) = self.verify_event(&value, room_version_rules).await { + return Err!(BadServerResponse(debug_error!( + "Event {event_id} failed verification: {e:?}" + ))); + } + + value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into())); + + Ok((event_id, value)) } - value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into())); + /// Validates the incoming event only using locally known verification keys. + /// Inserts the calculated event ID into the event_id field upon success. + pub async fn validate_and_add_event_id_no_fetch( + &self, + pdu: &RawJsonValue, + room_version_rules: &RoomVersionRules, + ) -> Result<(OwnedEventId, CanonicalJsonObject)> { + let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?; + if !self.required_keys_exist(&value, room_version_rules).await { + return Err!(BadServerResponse(debug_warn!( + "Event {event_id} cannot be verified: missing keys." + ))); + } + if let Err(e) = self.verify_event(&value, room_version_rules).await { + return Err!(BadServerResponse(debug_error!( + "Event {event_id} failed verification: {e:?}" + ))); + } - Ok((event_id, value)) -} + value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into())); -#[implement(super::Service)] -pub async fn validate_and_add_event_id_no_fetch( - &self, - pdu: &RawJsonValue, - room_version_rules: &RoomVersionRules, -) -> Result<(OwnedEventId, CanonicalJsonObject)> { - trace!(?pdu, "Validating PDU without fetching keys"); - let (event_id, mut value) = gen_event_id_canonical_json(pdu, room_version_rules)?; - trace!(event_id = event_id.as_str(), "Generated event ID, checking required keys"); - if !self.required_keys_exist(&value, room_version_rules).await { - debug_warn!( - "Event {event_id} is missing required keys, cannot verify without fetching keys" - ); - return Err!(BadServerResponse(debug_warn!( - "Event {event_id} cannot be verified: missing keys." - ))); + Ok((event_id, value)) } - trace!("All required keys exist, verifying event"); - if let Err(e) = self.verify_event(&value, room_version_rules).await { - debug_warn!("Event verification failed"); - return Err!(BadServerResponse(debug_error!( - "Event {event_id} failed verification: {e:?}" - ))); + + /// Verifies the incoming event after fetching the keys required to do so. + pub async fn verify_event( + &self, + event: &CanonicalJsonObject, + room_version_rules: &RoomVersionRules, + ) -> Result { + let keys = self.get_event_keys(event, room_version_rules).await?; + ruma::signatures::verify_event(&keys, event, room_version_rules).map_err(Into::into) } - trace!("Event verified successfully"); - value.insert("event_id".into(), CanonicalJsonValue::String(event_id.as_str().into())); - - Ok((event_id, value)) -} - -#[implement(super::Service)] -pub async fn verify_event( - &self, - event: &CanonicalJsonObject, - room_version_rules: &RoomVersionRules, -) -> Result { - let keys = self.get_event_keys(event, room_version_rules).await?; - ruma::signatures::verify_event(&keys, event, room_version_rules).map_err(Into::into) -} - -#[implement(super::Service)] -pub async fn verify_json( - &self, - event: &CanonicalJsonObject, - room_version_rules: &RoomVersionRules, -) -> Result { - let keys = self.get_event_keys(event, room_version_rules).await?; - ruma::signatures::verify_json(&keys, event).map_err(Into::into) + /// Verifies an arbitrary JSON object after fetching the keys required to do + /// so. + pub async fn verify_json( + &self, + event: &CanonicalJsonObject, + room_version_rules: &RoomVersionRules, + ) -> Result { + // TODO: is this actually used anywhere? + let keys = self.get_event_keys(event, room_version_rules).await?; + ruma::signatures::verify_json(&keys, event).map_err(Into::into) + } }