TudbuT/conduit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(appservices): don't panic on empty registration url

Browse source 
perf(appservices): cache regex for namespaces
This commit is contained in:
Matthias Ahouansou 2023-12-27 13:22:21 +00:00
parent a095e02d04
commit fa930182ae
No known key found for this signature in database
15 changed files with 335 additions and 244 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

166
src/api/appservice_server.rs
View file

@ -1,105 +1,111 @@
use crate::{services, utils, Error, Result}; use crate::{services, utils, Error, Result};
use bytes::BytesMut; use bytes::BytesMut;
use ruma::api::{IncomingResponse, MatrixVersion, OutgoingRequest, SendAccessToken}; use ruma::api::{
appservice::Registration, IncomingResponse, MatrixVersion, OutgoingRequest, SendAccessToken,
};
use std::{fmt::Debug, mem, time::Duration}; use std::{fmt::Debug, mem, time::Duration};
use tracing::warn; use tracing::warn;
/// Sends a request to an appservice
///
/// Only returns None if there is no url specified in the appservice registration file
#[tracing::instrument(skip(request))] #[tracing::instrument(skip(request))]
pub(crate) async fn send_request<T: OutgoingRequest>( pub(crate) async fn send_request<T: OutgoingRequest>(
registration: serde_yaml::Value, registration: Registration,
request: T, request: T,
) -> Result<T::IncomingResponse> ) -> Option<Result<T::IncomingResponse>>
where where
T: Debug, T: Debug,
{ {
let destination = registration.get("url").unwrap().as_str().unwrap(); if let Some(destination) = registration.url {
let hs_token = registration.get("hs_token").unwrap().as_str().unwrap(); let hs_token = registration.hs_token.as_str();
let mut http_request = request let mut http_request = request
.try_into_http_request::<BytesMut>( .try_into_http_request::<BytesMut>(
destination, &destination,
SendAccessToken::IfRequired(hs_token), SendAccessToken::IfRequired(hs_token),
&[MatrixVersion::V1_0], &[MatrixVersion::V1_0],
) )
.unwrap() .unwrap()
.map(|body| body.freeze()); .map(|body| body.freeze());
let mut parts = http_request.uri().clone().into_parts(); let mut parts = http_request.uri().clone().into_parts();
let old_path_and_query = parts.path_and_query.unwrap().as_str().to_owned(); let old_path_and_query = parts.path_and_query.unwrap().as_str().to_owned();
let symbol = if old_path_and_query.contains('?') { let symbol = if old_path_and_query.contains('?') {
"&" "&"
} else { } else {
"?" "?"
}; };
parts.path_and_query = Some( parts.path_and_query = Some(
(old_path_and_query + symbol + "access_token=" + hs_token) (old_path_and_query + symbol + "access_token=" + hs_token)
.parse() .parse()
.unwrap(), .unwrap(),
); );
*http_request.uri_mut() = parts.try_into().expect("our manipulation is always valid"); *http_request.uri_mut() = parts.try_into().expect("our manipulation is always valid");
let mut reqwest_request = reqwest::Request::try_from(http_request) let mut reqwest_request = reqwest::Request::try_from(http_request)
.expect("all http requests are valid reqwest requests"); .expect("all http requests are valid reqwest requests");
*reqwest_request.timeout_mut() = Some(Duration::from_secs(30)); *reqwest_request.timeout_mut() = Some(Duration::from_secs(30));
let url = reqwest_request.url().clone(); let url = reqwest_request.url().clone();
let mut response = match services() let mut response = match services()
.globals .globals
.default_client() .default_client()
.execute(reqwest_request) .execute(reqwest_request)
.await .await
{ {
Ok(r) => r, Ok(r) => r,
Err(e) => { Err(e) => {
warn!(
"Could not send request to appservice {:?} at {}: {}",
registration.id, destination, e
);
return Some(Err(e.into()));
}
};
// reqwest::Response -> http::Response conversion
let status = response.status();
let mut http_response_builder = http::Response::builder()
.status(status)
.version(response.version());
mem::swap(
response.headers_mut(),
http_response_builder
.headers_mut()
.expect("http::response::Builder is usable"),
);
let body = response.bytes().await.unwrap_or_else(|e| {
warn!("server error: {}", e);
Vec::new().into()
}); // TODO: handle timeout
if status != 200 {
warn!( warn!(
"Could not send request to appservice {:?} at {}: {}", "Appservice returned bad response {} {}\n{}\n{:?}",
registration.get("id"),
destination, destination,
e status,
url,
utils::string_from_bytes(&body)
); );
return Err(e.into());
} }
};
// reqwest::Response -> http::Response conversion let response = T::IncomingResponse::try_from_http_response(
let status = response.status(); http_response_builder
let mut http_response_builder = http::Response::builder() .body(body)
.status(status) .expect("reqwest body is valid http body"),
.version(response.version());
mem::swap(
response.headers_mut(),
http_response_builder
.headers_mut()
.expect("http::response::Builder is usable"),
);
let body = response.bytes().await.unwrap_or_else(|e| {
warn!("server error: {}", e);
Vec::new().into()
}); // TODO: handle timeout
if status != 200 {
warn!(
"Appservice returned bad response {} {}\n{}\n{:?}",
destination,
status,
url,
utils::string_from_bytes(&body)
); );
Some(response.map_err(|_| {
warn!(
"Appservice returned invalid response bytes {}\n{}",
destination, url
);
Error::BadServerResponse("Server returned bad response.")
}))
} else {
None
} }
let response = T::IncomingResponse::try_from_http_response(
http_response_builder
.body(body)
.expect("reqwest body is valid http body"),
);
response.map_err(|_| {
warn!(
"Appservice returned invalid response bytes {}\n{}",
destination, url
);
Error::BadServerResponse("Server returned bad response.")
})
} }

34
src/api/client_server/alias.rs
View file

@ -1,6 +1,5 @@
use crate::{services, Error, Result, Ruma}; use crate::{services, Error, Result, Ruma};
use rand::seq::SliceRandom; use rand::seq::SliceRandom;
use regex::Regex;
use ruma::{ use ruma::{
api::{ api::{
appservice, appservice,
@ -101,31 +100,28 @@ pub(crate) async fn get_alias_helper(
match services().rooms.alias.resolve_local_alias(&room_alias)? { match services().rooms.alias.resolve_local_alias(&room_alias)? {
Some(r) => room_id = Some(r), Some(r) => room_id = Some(r),
None => { None => {
for (_id, registration) in services().appservice.all()? { for appservice in services()
let aliases = registration .appservice
.get("namespaces") .registration_info
.and_then(|ns| ns.get("aliases")) .read()
.and_then(|aliases| aliases.as_sequence()) .await
.map_or_else(Vec::new, |aliases| { .values()
aliases {
.iter() if appservice.aliases.is_match(room_alias.as_str())
.filter_map(|aliases| Regex::new(aliases.get("regex")?.as_str()?).ok()) && if let Some(opt_result) = services()
.collect::<Vec<_>>()
});
if aliases
.iter()
.any(|aliases| aliases.is_match(room_alias.as_str()))
&& services()
.sending .sending
.send_appservice_request( .send_appservice_request(
registration, appservice.registration.clone(),
appservice::query::query_room_alias::v1::Request { appservice::query::query_room_alias::v1::Request {
room_alias: room_alias.clone(), room_alias: room_alias.clone(),
}, },
) )
.await .await
.is_ok() {
opt_result.is_ok()
} else {
false
}
{ {
room_id = Some( room_id = Some(
services() services()

15
src/api/ruma_wrapper/axum.rs
View file

@ -81,12 +81,9 @@ where
let mut json_body = serde_json::from_slice::<CanonicalJsonValue>(&body).ok(); let mut json_body = serde_json::from_slice::<CanonicalJsonValue>(&body).ok();
let appservices = services().appservice.all().unwrap(); let appservices = services().appservice.all().unwrap();
let appservice_registration = appservices.iter().find(|(_id, registration)| { let appservice_registration = appservices
registration .iter()
.get("as_token") .find(|(_id, registration)| Some(registration.as_token.as_str()) == token);
.and_then(|as_token| as_token.as_str())
.map_or(false, |as_token| token == Some(as_token))
});
let (sender_user, sender_device, sender_servername, from_appservice) = let (sender_user, sender_device, sender_servername, from_appservice) =
if let Some((_id, registration)) = appservice_registration { if let Some((_id, registration)) = appservice_registration {
@ -95,11 +92,7 @@ where
let user_id = query_params.user_id.map_or_else( let user_id = query_params.user_id.map_or_else(
|| { || {
UserId::parse_with_server_name( UserId::parse_with_server_name(
registration registration.sender_localpart.as_str(),
.get("sender_localpart")
.unwrap()
.as_str()
.unwrap(),
services().globals.server_name(), services().globals.server_name(),
) )
.unwrap() .unwrap()

11
src/database/key_value/appservice.rs
View file

@ -1,10 +1,11 @@
use ruma::api::appservice::Registration;
use crate::{database::KeyValueDatabase, service, utils, Error, Result}; use crate::{database::KeyValueDatabase, service, utils, Error, Result};
impl service::appservice::Data for KeyValueDatabase { impl service::appservice::Data for KeyValueDatabase {
/// Registers an appservice and returns the ID to the caller /// Registers an appservice and returns the ID to the caller
fn register_appservice(&self, yaml: serde_yaml::Value) -> Result<String> { fn register_appservice(&self, yaml: Registration) -> Result<String> {
// TODO: Rumaify let id = yaml.id.as_str();
let id = yaml.get("id").unwrap().as_str().unwrap();
self.id_appserviceregistrations.insert( self.id_appserviceregistrations.insert(
id.as_bytes(), id.as_bytes(),
serde_yaml::to_string(&yaml).unwrap().as_bytes(), serde_yaml::to_string(&yaml).unwrap().as_bytes(),
@ -32,7 +33,7 @@ impl service::appservice::Data for KeyValueDatabase {
Ok(()) Ok(())
} }
fn get_registration(&self, id: &str) -> Result<Option<serde_yaml::Value>> { fn get_registration(&self, id: &str) -> Result<Option<Registration>> {
self.cached_registrations self.cached_registrations
.read() .read()
.unwrap() .unwrap()
@ -64,7 +65,7 @@ impl service::appservice::Data for KeyValueDatabase {
))) )))
} }
fn all(&self) -> Result<Vec<(String, serde_yaml::Value)>> { fn all(&self) -> Result<Vec<(String, Registration)>> {
self.iter_ids()? self.iter_ids()?
.filter_map(|id| id.ok()) .filter_map(|id| id.ok())
.map(move |id| { .map(move |id| {

47
src/database/key_value/rooms/state_cache.rs
View file

@ -1,13 +1,16 @@
use std::{collections::HashSet, sync::Arc}; use std::{collections::HashSet, sync::Arc};
use regex::Regex;
use ruma::{ use ruma::{
events::{AnyStrippedStateEvent, AnySyncStateEvent}, events::{AnyStrippedStateEvent, AnySyncStateEvent},
serde::Raw, serde::Raw,
OwnedRoomId, OwnedServerName, OwnedUserId, RoomId, ServerName, UserId, OwnedRoomId, OwnedServerName, OwnedUserId, RoomId, ServerName, UserId,
}; };
use crate::{database::KeyValueDatabase, service, services, utils, Error, Result}; use crate::{
database::KeyValueDatabase,
service::{self, appservice::RegistrationInfo},
services, utils, Error, Result,
};
impl service::rooms::state_cache::Data for KeyValueDatabase { impl service::rooms::state_cache::Data for KeyValueDatabase {
fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> { fn mark_as_once_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<()> {
@ -184,46 +187,28 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
} }
#[tracing::instrument(skip(self, room_id, appservice))] #[tracing::instrument(skip(self, room_id, appservice))]
fn appservice_in_room( fn appservice_in_room(&self, room_id: &RoomId, appservice: &RegistrationInfo) -> Result<bool> {
&self,
room_id: &RoomId,
appservice: &(String, serde_yaml::Value),
) -> Result<bool> {
let maybe = self let maybe = self
.appservice_in_room_cache .appservice_in_room_cache
.read() .read()
.unwrap() .unwrap()
.get(room_id) .get(room_id)
.and_then(|map| map.get(&appservice.0)) .and_then(|map| map.get(&appservice.registration.id))
.copied(); .copied();
if let Some(b) = maybe { if let Some(b) = maybe {
Ok(b) Ok(b)
} else if let Some(namespaces) = appservice.1.get("namespaces") { } else {
let users = namespaces let bridge_user_id = UserId::parse_with_server_name(
.get("users") appservice.registration.sender_localpart.as_str(),
.and_then(|users| users.as_sequence()) services().globals.server_name(),
.map_or_else(Vec::new, |users| { )
users .ok();
.iter()
.filter_map(|users| Regex::new(users.get("regex")?.as_str()?).ok())
.collect::<Vec<_>>()
});
let bridge_user_id = appservice
.1
.get("sender_localpart")
.and_then(|string| string.as_str())
.and_then(|string| {
UserId::parse_with_server_name(string, services().globals.server_name()).ok()
});
let in_room = bridge_user_id let in_room = bridge_user_id
.map_or(false, |id| self.is_joined(&id, room_id).unwrap_or(false)) .map_or(false, |id| self.is_joined(&id, room_id).unwrap_or(false))
|| self.room_members(room_id).any(|userid| { || self.room_members(room_id).any(|userid| {
userid.map_or(false, |userid| { userid.map_or(false, |userid| appservice.users.is_match(userid.as_str()))
users.iter().any(|r| r.is_match(userid.as_str()))
})
}); });
self.appservice_in_room_cache self.appservice_in_room_cache
@ -231,11 +216,9 @@ impl service::rooms::state_cache::Data for KeyValueDatabase {
.unwrap() .unwrap()
.entry(room_id.to_owned()) .entry(room_id.to_owned())
.or_default() .or_default()
.insert(appservice.0.clone(), in_room); .insert(appservice.registration.id.clone(), in_room);
Ok(in_room) Ok(in_room)
} else {
Ok(false)
} }
} }

20
src/database/mod.rs
View file

@ -8,7 +8,9 @@ use crate::{
use abstraction::{KeyValueDatabaseEngine, KvTree}; use abstraction::{KeyValueDatabaseEngine, KvTree};
use directories::ProjectDirs; use directories::ProjectDirs;
use lru_cache::LruCache; use lru_cache::LruCache;
use ruma::{ use ruma::{
api::appservice::Registration,
events::{ events::{
push_rules::{PushRulesEvent, PushRulesEventContent}, push_rules::{PushRulesEvent, PushRulesEventContent},
room::message::RoomMessageEventContent, room::message::RoomMessageEventContent,
@ -162,7 +164,7 @@ pub struct KeyValueDatabase {
//pub pusher: pusher::PushData, //pub pusher: pusher::PushData,
pub(super) senderkey_pusher: Arc<dyn KvTree>, pub(super) senderkey_pusher: Arc<dyn KvTree>,
pub(super) cached_registrations: Arc<RwLock<HashMap<String, serde_yaml::Value>>>, pub(super) cached_registrations: Arc<RwLock<HashMap<String, Registration>>>,
pub(super) pdu_cache: Mutex<LruCache<OwnedEventId, Arc<PduEvent>>>, pub(super) pdu_cache: Mutex<LruCache<OwnedEventId, Arc<PduEvent>>>,
pub(super) shorteventid_cache: Mutex<LruCache<u64, Arc<EventId>>>, pub(super) shorteventid_cache: Mutex<LruCache<u64, Arc<EventId>>>,
pub(super) auth_chain_cache: Mutex<LruCache<Vec<u64>, Arc<HashSet<u64>>>>, pub(super) auth_chain_cache: Mutex<LruCache<Vec<u64>, Arc<HashSet<u64>>>>,
@ -967,6 +969,22 @@ impl KeyValueDatabase {
); );
} }
// Inserting registrations into cache
for appservice in services().appservice.all()? {
services()
.appservice
.registration_info
.write()
.await
.insert(
appservice.0,
appservice
.1
.try_into()
.expect("Should be validated on registration"),
);
}
// This data is probably outdated // This data is probably outdated
db.presenceid_presence.clear()?; db.presenceid_presence.clear()?;

7
src/service/admin/mod.rs
View file

@ -8,6 +8,7 @@ use std::{
use clap::Parser; use clap::Parser;
use regex::Regex; use regex::Regex;
use ruma::{ use ruma::{
api::appservice::Registration,
events::{ events::{
room::{ room::{
canonical_alias::RoomCanonicalAliasEventContent, canonical_alias::RoomCanonicalAliasEventContent,
@ -335,10 +336,9 @@ impl Service {
if body.len() > 2 && body[0].trim() == "```" && body.last().unwrap().trim() == "```" if body.len() > 2 && body[0].trim() == "```" && body.last().unwrap().trim() == "```"
{ {
let appservice_config = body[1..body.len() - 1].join("\n"); let appservice_config = body[1..body.len() - 1].join("\n");
let parsed_config = let parsed_config = serde_yaml::from_str::<Registration>(&appservice_config);
serde_yaml::from_str::<serde_yaml::Value>(&appservice_config);
match parsed_config { match parsed_config {
Ok(yaml) => match services().appservice.register_appservice(yaml) { Ok(yaml) => match services().appservice.register_appservice(yaml).await {
Ok(id) => RoomMessageEventContent::text_plain(format!( Ok(id) => RoomMessageEventContent::text_plain(format!(
"Appservice registered with ID: {id}." "Appservice registered with ID: {id}."
)), )),
@ -361,6 +361,7 @@ impl Service {
} => match services() } => match services()
.appservice .appservice
.unregister_appservice(&appservice_identifier) .unregister_appservice(&appservice_identifier)
.await
{ {
Ok(()) => RoomMessageEventContent::text_plain("Appservice unregistered."), Ok(()) => RoomMessageEventContent::text_plain("Appservice unregistered."),
Err(e) => RoomMessageEventContent::text_plain(format!( Err(e) => RoomMessageEventContent::text_plain(format!(

8
src/service/appservice/data.rs
View file

@ -1,8 +1,10 @@
use ruma::api::appservice::Registration;
use crate::Result; use crate::Result;
pub trait Data: Send + Sync { pub trait Data: Send + Sync {
/// Registers an appservice and returns the ID to the caller /// Registers an appservice and returns the ID to the caller
fn register_appservice(&self, yaml: serde_yaml::Value) -> Result<String>; fn register_appservice(&self, yaml: Registration) -> Result<String>;
/// Remove an appservice registration /// Remove an appservice registration
/// ///
@ -11,9 +13,9 @@ pub trait Data: Send + Sync {
/// * `service_name` - the name you send to register the service previously /// * `service_name` - the name you send to register the service previously
fn unregister_appservice(&self, service_name: &str) -> Result<()>; fn unregister_appservice(&self, service_name: &str) -> Result<()>;
fn get_registration(&self, id: &str) -> Result<Option<serde_yaml::Value>>; fn get_registration(&self, id: &str) -> Result<Option<Registration>>;
fn iter_ids<'a>(&'a self) -> Result<Box<dyn Iterator<Item = Result<String>> + 'a>>; fn iter_ids<'a>(&'a self) -> Result<Box<dyn Iterator<Item = Result<String>> + 'a>>;
fn all(&self) -> Result<Vec<(String, serde_yaml::Value)>>; fn all(&self) -> Result<Vec<(String, Registration)>>;
} }

114
src/service/appservice/mod.rs
View file

@ -1,16 +1,113 @@
mod data; mod data;
use std::collections::HashMap;
pub use data::Data; pub use data::Data;
use crate::Result; use regex::RegexSet;
use ruma::api::appservice::{Namespace, Registration};
use tokio::sync::RwLock;
use crate::{services, Result};
/// Compiled regular expressions for a namespace
pub struct NamespaceRegex {
pub exclusive: Option<RegexSet>,
pub non_exclusive: Option<RegexSet>,
}
impl NamespaceRegex {
/// Checks if this namespace has rights to a namespace
pub fn is_match(&self, heystack: &str) -> bool {
if self.is_exclusive_match(heystack) {
return true;
}
if let Some(non_exclusive) = &self.non_exclusive {
if non_exclusive.is_match(heystack) {
return true;
}
}
false
}
/// Checks if this namespace has exlusive rights to a namespace
pub fn is_exclusive_match(&self, heystack: &str) -> bool {
if let Some(exclusive) = &self.exclusive {
if exclusive.is_match(heystack) {
return true;
}
}
false
}
}
impl TryFrom<Vec<Namespace>> for NamespaceRegex {
fn try_from(value: Vec<Namespace>) -> Result<Self, regex::Error> {
let mut exclusive = vec![];
let mut non_exclusive = vec![];
for namespace in value {
if namespace.exclusive {
exclusive.push(namespace.regex);
} else {
non_exclusive.push(namespace.regex);
}
}
Ok(NamespaceRegex {
exclusive: if exclusive.is_empty() {
None
} else {
Some(RegexSet::new(exclusive)?)
},
non_exclusive: if non_exclusive.is_empty() {
None
} else {
Some(RegexSet::new(non_exclusive)?)
},
})
}
type Error = regex::Error;
}
/// Compiled regular expressions for an appservice
pub struct RegistrationInfo {
pub registration: Registration,
pub users: NamespaceRegex,
pub aliases: NamespaceRegex,
pub rooms: NamespaceRegex,
}
impl TryFrom<Registration> for RegistrationInfo {
fn try_from(value: Registration) -> Result<RegistrationInfo, regex::Error> {
Ok(RegistrationInfo {
users: value.namespaces.users.clone().try_into()?,
aliases: value.namespaces.aliases.clone().try_into()?,
rooms: value.namespaces.rooms.clone().try_into()?,
registration: value,
})
}
type Error = regex::Error;
}
pub struct Service { pub struct Service {
pub db: &'static dyn Data, pub db: &'static dyn Data,
pub registration_info: RwLock<HashMap<String, RegistrationInfo>>,
} }
impl Service { impl Service {
/// Registers an appservice and returns the ID to the caller /// Registers an appservice and returns the ID to the caller
pub fn register_appservice(&self, yaml: serde_yaml::Value) -> Result<String> { pub async fn register_appservice(&self, yaml: Registration) -> Result<String> {
services()
.appservice
.registration_info
.write()
.await
.insert(yaml.id.clone(), yaml.clone().try_into()?);
self.db.register_appservice(yaml) self.db.register_appservice(yaml)
} }
@ -19,11 +116,18 @@ impl Service {
/// # Arguments /// # Arguments
/// ///
/// * `service_name` - the name you send to register the service previously /// * `service_name` - the name you send to register the service previously
pub fn unregister_appservice(&self, service_name: &str) -> Result<()> { pub async fn unregister_appservice(&self, service_name: &str) -> Result<()> {
services()
.appservice
.registration_info
.write()
.await
.remove(service_name);
self.db.unregister_appservice(service_name) self.db.unregister_appservice(service_name)
} }
pub fn get_registration(&self, id: &str) -> Result<Option<serde_yaml::Value>> { pub fn get_registration(&self, id: &str) -> Result<Option<Registration>> {
self.db.get_registration(id) self.db.get_registration(id)
} }
@ -31,7 +135,7 @@ impl Service {
self.db.iter_ids() self.db.iter_ids()
} }
pub fn all(&self) -> Result<Vec<(String, serde_yaml::Value)>> { pub fn all(&self) -> Result<Vec<(String, Registration)>> {
self.db.all() self.db.all()
} }
} }

7
src/service/mod.rs
View file

@ -4,7 +4,7 @@ use std::{
}; };
use lru_cache::LruCache; use lru_cache::LruCache;
use tokio::sync::Mutex; use tokio::sync::{Mutex, RwLock};
use crate::{Config, Result}; use crate::{Config, Result};
@ -56,7 +56,10 @@ impl Services {
config: Config, config: Config,
) -> Result<Self> { ) -> Result<Self> {
Ok(Self { Ok(Self {
appservice: appservice::Service { db }, appservice: appservice::Service {
db,
registration_info: RwLock::new(HashMap::new()),
},
pusher: pusher::Service { db }, pusher: pusher::Service { db },
rooms: rooms::Service { rooms: rooms::Service {
alias: rooms::alias::Service { db }, alias: rooms::alias::Service { db },

8
src/service/rooms/state_cache/data.rs
View file

@ -1,6 +1,6 @@
use std::{collections::HashSet, sync::Arc}; use std::{collections::HashSet, sync::Arc};
use crate::Result; use crate::{service::appservice::RegistrationInfo, Result};
use ruma::{ use ruma::{
events::{AnyStrippedStateEvent, AnySyncStateEvent}, events::{AnyStrippedStateEvent, AnySyncStateEvent},
serde::Raw, serde::Raw,
@ -22,11 +22,7 @@ pub trait Data: Send + Sync {
fn get_our_real_users(&self, room_id: &RoomId) -> Result<Arc<HashSet<OwnedUserId>>>; fn get_our_real_users(&self, room_id: &RoomId) -> Result<Arc<HashSet<OwnedUserId>>>;
fn appservice_in_room( fn appservice_in_room(&self, room_id: &RoomId, appservice: &RegistrationInfo) -> Result<bool>;
&self,
room_id: &RoomId,
appservice: &(String, serde_yaml::Value),
) -> Result<bool>;
/// Makes a user forget a room. /// Makes a user forget a room.
fn forget(&self, room_id: &RoomId, user_id: &UserId) -> Result<()>; fn forget(&self, room_id: &RoomId, user_id: &UserId) -> Result<()>;

4
src/service/rooms/state_cache/mod.rs
View file

@ -16,7 +16,7 @@ use ruma::{
}; };
use tracing::warn; use tracing::warn;
use crate::{services, Error, Result}; use crate::{service::appservice::RegistrationInfo, services, Error, Result};
pub struct Service { pub struct Service {
pub db: &'static dyn Data, pub db: &'static dyn Data,
@ -205,7 +205,7 @@ impl Service {
pub fn appservice_in_room( pub fn appservice_in_room(
&self, &self,
room_id: &RoomId, room_id: &RoomId,
appservice: &(String, serde_yaml::Value), appservice: &RegistrationInfo,
) -> Result<bool> { ) -> Result<bool> {
self.db.appservice_in_room(room_id, appservice) self.db.appservice_in_room(room_id, appservice)
} }

114
src/service/rooms/timeline/mod.rs
View file

@ -7,7 +7,7 @@ use std::{
}; };
pub use data::Data; pub use data::Data;
use regex::Regex;
use ruma::{ use ruma::{
api::{client::error::ErrorKind, federation}, api::{client::error::ErrorKind, federation},
canonical_json::to_canonical_value, canonical_json::to_canonical_value,
@ -21,8 +21,7 @@ use ruma::{
}, },
push::{Action, Ruleset, Tweak}, push::{Action, Ruleset, Tweak},
serde::Base64, serde::Base64,
state_res, state_res::{self, Event, RoomVersion},
state_res::{Event, RoomVersion},
uint, user_id, CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId, uint, user_id, CanonicalJsonObject, CanonicalJsonValue, EventId, OwnedEventId, OwnedRoomId,
OwnedServerName, RoomId, ServerName, UserId, OwnedServerName, RoomId, ServerName, UserId,
}; };
@ -33,7 +32,10 @@ use tracing::{error, info, warn};
use crate::{ use crate::{
api::server_server, api::server_server,
service::pdu::{EventHash, PduBuilder}, service::{
appservice::NamespaceRegex,
pdu::{EventHash, PduBuilder},
},
services, utils, Error, PduEvent, Result, services, utils, Error, PduEvent, Result,
}; };
@ -522,15 +524,21 @@ impl Service {
} }
} }
for appservice in services().appservice.all()? { for appservice in services()
.appservice
.registration_info
.read()
.await
.values()
{
if services() if services()
.rooms .rooms
.state_cache .state_cache
.appservice_in_room(&pdu.room_id, &appservice)? .appservice_in_room(&pdu.room_id, appservice)?
{ {
services() services()
.sending .sending
.send_pdu_appservice(appservice.0, pdu_id.clone())?; .send_pdu_appservice(appservice.registration.id.clone(), pdu_id.clone())?;
continue; continue;
} }
@ -542,73 +550,41 @@ impl Service {
.as_ref() .as_ref()
.and_then(|state_key| UserId::parse(state_key.as_str()).ok()) .and_then(|state_key| UserId::parse(state_key.as_str()).ok())
{ {
if let Some(appservice_uid) = appservice let appservice_uid = appservice.registration.sender_localpart.as_str();
.1 if state_key_uid == appservice_uid {
.get("sender_localpart") services().sending.send_pdu_appservice(
.and_then(|string| string.as_str()) appservice.registration.id.clone(),
.and_then(|string| { pdu_id.clone(),
UserId::parse_with_server_name(string, services().globals.server_name()) )?;
.ok() continue;
})
{
if state_key_uid == &appservice_uid {
services()
.sending
.send_pdu_appservice(appservice.0, pdu_id.clone())?;
continue;
}
} }
} }
} }
if let Some(namespaces) = appservice.1.get("namespaces") { let matching_users = |users: &NamespaceRegex| {
let users = namespaces appservice.users.is_match(pdu.sender.as_str())
.get("users") || pdu.kind == TimelineEventType::RoomMember
.and_then(|users| users.as_sequence()) && pdu
.map_or_else(Vec::new, |users| { .state_key
users .as_ref()
.iter() .map_or(false, |state_key| users.is_match(state_key))
.filter_map(|users| Regex::new(users.get("regex")?.as_str()?).ok()) };
.collect::<Vec<_>>() let matching_aliases = |aliases: &NamespaceRegex| {
}); services()
let aliases = namespaces .rooms
.get("aliases") .alias
.and_then(|aliases| aliases.as_sequence()) .local_aliases_for_room(&pdu.room_id)
.map_or_else(Vec::new, |aliases| { .filter_map(|r| r.ok())
aliases .any(|room_alias| aliases.is_match(room_alias.as_str()))
.iter() };
.filter_map(|aliases| Regex::new(aliases.get("regex")?.as_str()?).ok())
.collect::<Vec<_>>()
});
let rooms = namespaces
.get("rooms")
.and_then(|rooms| rooms.as_sequence());
let matching_users = |users: &Regex| { if matching_aliases(&appservice.aliases)
users.is_match(pdu.sender.as_str()) || appservice.rooms.is_match(pdu.room_id.as_str())
|| pdu.kind == TimelineEventType::RoomMember || matching_users(&appservice.users)
&& pdu {
.state_key services()
.as_ref() .sending
.map_or(false, |state_key| users.is_match(state_key)) .send_pdu_appservice(appservice.registration.id.clone(), pdu_id.clone())?;
};
let matching_aliases = |aliases: &Regex| {
services()
.rooms
.alias
.local_aliases_for_room(&pdu.room_id)
.filter_map(|r| r.ok())
.any(|room_alias| aliases.is_match(room_alias.as_str()))
};
if aliases.iter().any(matching_aliases)
|| rooms.map_or(false, |rooms| rooms.contains(&pdu.room_id.as_str().into()))
|| users.iter().any(matching_users)
{
services()
.sending
.send_pdu_appservice(appservice.0, pdu_id.clone())?;
}
} }
} }

19
src/service/sending/mod.rs
View file

@ -22,7 +22,7 @@ use base64::{engine::general_purpose, Engine as _};
use ruma::{ use ruma::{
api::{ api::{
appservice, appservice::{self, Registration},
federation::{ federation::{
self, self,
transactions::edu::{ transactions::edu::{
@ -484,7 +484,7 @@ impl Service {
let permit = services().sending.maximum_requests.acquire().await; let permit = services().sending.maximum_requests.acquire().await;
let response = appservice_server::send_request( let response = match appservice_server::send_request(
services() services()
.appservice .appservice
.get_registration(id) .get_registration(id)
@ -511,8 +511,12 @@ impl Service {
}, },
) )
.await .await
.map(|_response| kind.clone()) {
.map_err(|e| (kind, e)); None => Ok(kind.clone()),
Some(op_resp) => op_resp
.map(|_response| kind.clone())
.map_err(|e| (kind.clone(), e)),
};
drop(permit); drop(permit);
@ -698,12 +702,15 @@ impl Service {
response response
} }
/// Sends a request to an appservice
///
/// Only returns None if there is no url specified in the appservice registration file
#[tracing::instrument(skip(self, registration, request))] #[tracing::instrument(skip(self, registration, request))]
pub async fn send_appservice_request<T: OutgoingRequest>( pub async fn send_appservice_request<T: OutgoingRequest>(
&self, &self,
registration: serde_yaml::Value, registration: Registration,
request: T, request: T,
) -> Result<T::IncomingResponse> ) -> Option<Result<T::IncomingResponse>>
where where
T: Debug, T: Debug,
{ {

5
src/utils/error.rs
View file

@ -54,6 +54,11 @@ pub enum Error {
#[from] #[from]
source: reqwest::Error, source: reqwest::Error,
}, },
#[error("Could build regular expression: {source}")]
RegexError {
#[from]
source: regex::Error,
},
#[error("{0}")] #[error("{0}")]
FederationError(OwnedServerName, RumaError), FederationError(OwnedServerName, RumaError),
#[error("Could not do this io: {source}")] #[error("Could not do this io: {source}")]