tuwunel_service/appservice/
thirdparty.rs

1use std::collections::BTreeMap;
2
3use futures::StreamExt;
4use ruma::{
5	api::appservice::{
6		Registration,
7		thirdparty::{get_location_for_protocol, get_protocol, get_user_for_protocol},
8	},
9	serde::Raw,
10	thirdparty::{Location, Protocol, User},
11};
12use serde_json::{Value, value::to_raw_value};
13use tuwunel_core::{
14	implement,
15	utils::stream::{IterStream, ReadyExt, WidebandExt},
16};
17
18type Protocols = BTreeMap<String, Raw<Protocol>>;
19
20/// Fetches third-party protocol metadata from the registered appservices and
21/// keys each response by protocol id. `only` restricts the fan-out to a single
22/// protocol.
23///
24/// Metadata remains opaque JSON so fields unknown to this server survive the
25/// forwarding path. When several appservices advertise the same protocol, the
26/// first response supplies the metadata. Later responses append array-valued
27/// `instances` only when both documents expose that shape. Appservices without
28/// a usable destination and failed or undecodable responses contribute nothing
29/// rather than failing the client request.
30#[implement(super::Service)]
31#[tracing::instrument(level = "debug", skip(self))]
32pub async fn thirdparty_protocols(&self, only: Option<&str>) -> Protocols {
33	let jobs: Vec<(Registration, String)> = self
34		.read()
35		.await
36		.values()
37		.filter_map(|info| {
38			info.registration
39				.protocols
40				.as_ref()
41				.map(|protocols| (&info.registration, protocols))
42		})
43		.flat_map(|(registration, protocols)| {
44			protocols
45				.iter()
46				.filter(|protocol| only.is_none_or(|only| only == protocol.as_str()))
47				.map(move |protocol| (registration.clone(), protocol.clone()))
48		})
49		.collect();
50
51	jobs.into_iter()
52		.stream()
53		.wide_filter_map(async |(registration, protocol)| {
54			let request = get_protocol::v1::Request { protocol: protocol.clone() };
55
56			self.send_request(registration, request)
57				.await
58				.ok()
59				.flatten()
60				.map(|response| (protocol, Raw::from_json(response.protocol.into_json())))
61		})
62		.ready_fold(Protocols::new(), |mut protocols, (protocol, metadata)| {
63			protocols
64				.entry(protocol)
65				.and_modify(|existing| merge_instances(existing, &metadata))
66				.or_insert(metadata);
67
68			protocols
69		})
70		.await
71}
72
73/// Concatenates the `instances` of `addition` onto `base`, keeping every other
74/// field from `base`. Both bodies are opaque JSON, so a malformed side is left
75/// alone rather than dropped.
76fn merge_instances(base: &mut Raw<Protocol>, addition: &Raw<Protocol>) {
77	let (Ok(mut merged), Ok(extra)) =
78		(base.deserialize_as::<Value>(), addition.deserialize_as::<Value>())
79	else {
80		return;
81	};
82
83	let (Some(instances), Some(added)) = (
84		merged
85			.get_mut("instances")
86			.and_then(Value::as_array_mut),
87		extra.get("instances").and_then(Value::as_array),
88	) else {
89		return;
90	};
91
92	instances.extend(added.iter().cloned());
93
94	if let Ok(raw) = to_raw_value(&merged) {
95		*base = Raw::from_json(raw);
96	}
97}
98
99/// Looks up third-party users on `protocol` via the appservices declaring it,
100/// forwarding `fields` to each and concatenating their results.
101#[implement(super::Service)]
102#[tracing::instrument(level = "debug", skip(self, fields))]
103pub async fn thirdparty_users(
104	&self,
105	protocol: &str,
106	fields: &BTreeMap<String, String>,
107) -> Vec<User> {
108	self.declaring(protocol)
109		.await
110		.into_iter()
111		.stream()
112		.wide_filter_map(async |registration| {
113			let request = get_user_for_protocol::v1::Request {
114				protocol: protocol.to_owned(),
115				fields: forwarded_fields(fields).collect(),
116			};
117
118			self.send_request(registration, request)
119				.await
120				.ok()
121				.flatten()
122		})
123		.map(|response| response.users)
124		.concat()
125		.await
126}
127
128/// Looks up third-party locations on `protocol` via the appservices declaring
129/// it, forwarding `fields` to each and concatenating their results.
130#[implement(super::Service)]
131#[tracing::instrument(level = "debug", skip(self, fields))]
132pub async fn thirdparty_locations(
133	&self,
134	protocol: &str,
135	fields: &BTreeMap<String, String>,
136) -> Vec<Location> {
137	self.declaring(protocol)
138		.await
139		.into_iter()
140		.stream()
141		.wide_filter_map(async |registration| {
142			let request = get_location_for_protocol::v1::Request {
143				protocol: protocol.to_owned(),
144				fields: forwarded_fields(fields).collect(),
145			};
146
147			self.send_request(registration, request)
148				.await
149				.ok()
150				.flatten()
151		})
152		.map(|response| response.locations)
153		.concat()
154		.await
155}
156
157/// Snapshots the registrations declaring `protocol`. Cloning under the read
158/// lock lets the fan-out release its guard before the first network await.
159#[implement(super::Service)]
160async fn declaring(&self, protocol: &str) -> Vec<Registration> {
161	self.read()
162		.await
163		.values()
164		.filter(|info| declares(&info.registration, protocol))
165		.map(|info| info.registration.clone())
166		.collect()
167}
168
169/// Drops the client's `access_token` from the forwarded query so a legacy
170/// query-param credential never reaches the appservice.
171fn forwarded_fields(
172	fields: &BTreeMap<String, String>,
173) -> impl Iterator<Item = (String, String)> + '_ {
174	fields
175		.iter()
176		.filter(|(name, _)| name.as_str() != "access_token")
177		.map(|(name, value)| (name.clone(), value.clone()))
178}
179
180fn declares(registration: &Registration, protocol: &str) -> bool {
181	registration
182		.protocols
183		.as_ref()
184		.is_some_and(|protocols| {
185			protocols
186				.iter()
187				.any(|declared| declared == protocol)
188		})
189}
tuwunel_service/appservice/thirdparty.rs

tuwunel_service/appservice/
thirdparty.rs
