feat: Continuwuity notarizes the notary

changelog.d/2314.feature.md

@@ -0,0 +1 @@
+Added support for acting as a trusted server, meaning the `trusted_servers` list can consist of servers that are either Synapse OR continuwuity.

conduwuit-example.toml

@@ -1025,10 +1025,7 @@
 # (trusted key notary servers), as well as serve as trusted servers for
 # other operations like backfill and event fetching.
 #
-# Currently, continuwuity doesn't support inbound key requests, so
-# this list should only contain other Synapse servers.
-#
-# example: ["matrix.org", "starstruck.systems"]
+# example: ["continuwuity.org", "matrix.org"]
 #
 # It is possible to restrict which signing keys trusted servers are
 # allowed to sign responses with. Without configuring this, all responses
@@ -1066,7 +1063,7 @@
 # is discouraged for performance and reliability reasons.
 #
 # env: CONTINUWUITY_TRUSTED_SERVERS
-#trusted_servers = ["matrix.org"]
+#trusted_servers = ["continuwuity.org", "continuwuity.rocks", "matrix.org"]
 
 # Whether to prioritise trusted server lookups over origin server lookups.
 #

src/api/router.rs

@@ -207,7 +207,10 @@ pub fn build(router: Router<State>, state: State) -> Router<State> {
 	if config.allow_federation {
 		router = router
 			.ruma_route(&server::get_server_version_route)
-			.route("/_matrix/key/v2/server", get(server::get_server_keys_route))
+			.ruma_route(&server::get_server_keys_route)
+			.ruma_route(&server::get_remote_server_keys_batch_route)
+			.ruma_route(&server::get_remote_server_keys_batch_v3_unstable_route)
+			.ruma_route(&server::get_remote_server_keys_route)
 			.ruma_route(&server::get_public_rooms_route)
 			.ruma_route(&server::get_public_rooms_filtered_route)
 			.ruma_route(&server::send_transaction_message_route)

src/api/server/key.rs

@@ -1,89 +1,380 @@
-use std::{collections::BTreeMap, mem::take, time::Duration};
-
-use axum::{Json, extract::State, response::IntoResponse};
-use conduwuit::{Result, utils::timepoint_from_now};
-use futures::StreamExt;
-use ruma::{
-	MilliSecondsSinceUnixEpoch,
-	api::{
-		OutgoingResponseExt,
-		federation::discovery::{OldVerifyKey, ServerSigningKeys, get_server_keys},
-	},
-	assign,
-	serde::Raw,
+use std::{
+	collections::{BTreeMap, HashMap, HashSet},
+	ops::Index,
+	sync::Arc,
+	time::Instant,
 };
+
+use axum::extract::State;
+use conduwuit::{
+	Err, Result, debug, debug_info, error,
+	utils::{
+		ReadyExt,
+		stream::{BroadbandExt, automatic_width},
+		to_canonical_object,
+	},
+};
+use futures::{StreamExt, stream::FuturesUnordered};
+use ruma::{
+	MilliSecondsSinceUnixEpoch, OwnedServerName, OwnedServerSigningKeyId, ServerName,
+	api::federation::discovery::{
+		ServerSigningKeys, get_remote_server_keys, get_remote_server_keys_batch,
+		get_remote_server_keys_batch::v2::QueryCriteria, get_server_keys,
+	},
+	serde::Raw,
+	uint,
+};
+use serde_json::value::to_raw_value;
 use service::{server_keys, server_keys::in_one_week};
 
+use crate::router::Ruma;
+
 /// # `GET /_matrix/key/v2/server`
 ///
 /// Gets the public signing keys of this server.
 pub(crate) async fn get_server_keys_route(
 	State(services): State<crate::State>,
-) -> Result<impl IntoResponse> {
-	let server_name = services.globals.server_name();
-	let verify_keys = BTreeMap::from([services.server_keys.active_verify_key()]);
+	_body: Ruma<get_server_keys::v2::Request>,
+) -> Result<get_server_keys::v2::Response> {
+	services
+		.server_keys
+		.build_server_keys_response()
+		.await
+		.map(get_server_keys::v2::Response::new)
+}
 
-	let old_verify_keys = if let Some(k) = &services.config.old_verify_keys {
-		k.to_owned()
+const MAX_KEYS_PER_QUERY: usize = 16384;
+const MAX_SERVERS_PER_QUERY: usize = 4096;
+
+/// # `POST /_matrix/key/v2/query`
+///
+/// Fetches remote server signing keys in bulk, seeding from the local server's
+/// database.
+///
+/// This route applies a semaphore to slow down requests that query more signing
+/// keys than is allowed by [MSC4556].
+///
+/// [MSC4556]: https://github.com/matrix-org/matrix-spec-proposals/pull/4556
+pub(crate) async fn get_remote_server_keys_batch_route(
+	State(services): State<crate::State>,
+	body: Ruma<get_remote_server_keys_batch::v2::Request>,
+) -> Result<get_remote_server_keys_batch::v2::Response> {
+	query_inner(&services, &body.server_keys, false)
+		.await
+		.map(get_remote_server_keys_batch::v2::Response::new)
+}
+
+/// # `POST /_matrix/key/unstable/org.continuwuity.msc4556/query`
+/// ## `POST /_matrix/key/v3/query`
+///
+/// Fetches remote server signing keys in bulk, seeding from the local server's
+/// database.
+///
+/// MSC4556: https://github.com/matrix-org/matrix-spec-proposals/pull/4556
+pub(crate) async fn get_remote_server_keys_batch_v3_unstable_route(
+	State(services): State<crate::State>,
+	body: Ruma<ruminuwuity::federation::notary::unstable::Request>,
+) -> Result<ruminuwuity::federation::notary::unstable::Response> {
+	query_inner(&services, &body.server_keys, false)
+		.await
+		.map(ruminuwuity::federation::notary::unstable::Response::new)
+}
+
+/// Handles querying the server keys for `/_matrix/key/*/query`.
+///
+/// `v2` should be `true` if the request path is `/_matrix/key/v2/query` - this
+/// will potentially apply a rate-limit to the request.
+async fn query_inner(
+	services: &crate::State,
+	body: &BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, QueryCriteria>>,
+	v2: bool,
+) -> Result<Vec<Raw<ServerSigningKeys>>> {
+	let start = (Instant::now(), MilliSecondsSinceUnixEpoch::now());
+	// If the remote server isn't asking for any keys, we can return an empty
+	// array immediately.
+	if body.is_empty() {
+		return Ok(Vec::new());
+	}
+
+	// Count how many keys are being requested.
+	// Each server incurs a cost of at least one, and then the count of each key
+	// being queried per-server.
+	let total_queried_servers = body.len();
+	let total_queried_keys = body
+		.values()
+		.fold(body.len(), |acc, q| acc.saturating_add(q.len()));
+
+	if !v2 {
+		// If we're on the v3 endpoint, we can apply these limits explicitly.
+		if total_queried_servers > MAX_SERVERS_PER_QUERY {
+			return Err!(Request(TooLarge(
+				"Too many server keys requested ({total_queried_servers} > \
+				 {MAX_SERVERS_PER_QUERY}",
+			)));
+		}
+		if total_queried_keys > MAX_KEYS_PER_QUERY {
+			return Err!(Request(TooLarge(
+				"Too many keys requested ({total_queried_keys} > {MAX_KEYS_PER_QUERY})"
+			)));
+		}
+	}
+	let penalise =
+		total_queried_servers > MAX_SERVERS_PER_QUERY || total_queried_keys > MAX_KEYS_PER_QUERY;
+
+	debug_info!("Fetching {total_queried_keys} keys across {} servers", body.len());
+	let mut futs: FuturesUnordered<_> = FuturesUnordered::new();
+
+	// If we're on the v2 endpoint, we aren't allowed to reject the request.
+	// To avoid overloading the server, we'll apply a semaphore to limit how
+	// many futures can be active at a time.
+	let width = automatic_width();
+	let sem = if v2 && penalise {
+		Some(Arc::new(tokio::sync::Semaphore::new(width)))
 	} else {
-		// We need to figure out any old signing keys we've acquired from
-		// notaries.
-		let mut keys = services.server_keys.signing_keys_for(server_name);
-		let mut old_keys = BTreeMap::new();
-		while let Some(resp) = keys.next().await {
-			// We need to verify that this response is still trusted based on
-			// the current configuration. It's possible we got this response
-			// from a notary who we no longer trust.
-			if resp.valid_until_ts > in_one_week() {
-				continue;
-			}
-			let trusted = services.config.trusted_servers.iter().any(|notary| {
-				server_keys::Service::verify_server_keys_response(
-					&resp,
-					Some((notary.server_name(), notary.verify_keys())),
-				)
-				.is_ok()
-			});
-			if !trusted {
-				continue;
-			}
-			for (old_key_id, old_key) in resp.old_verify_keys {
-				if old_key.expired_ts > resp.valid_until_ts {
-					// This key expired in the future which is probably illegal
-					continue;
+		None
+	};
+	for (server_name, queries) in body.clone() {
+		futs.push(acquire_keys_as_notary(
+			services.server_keys.clone(),
+			services.globals.server_name().to_owned(),
+			server_name,
+			queries,
+			start.1,
+			sem.clone(),
+		));
+	}
+	let mut response = Vec::with_capacity(total_queried_keys);
+	while let Some(v) = futs.next().await {
+		response.extend(v);
+	}
+
+	debug_info!(
+		elapsed=?start.0.elapsed(),
+		%total_queried_servers,
+		%total_queried_keys,
+		"Fetched {} key responses",
+		response.len()
+	);
+	Ok(response)
+}
+
+async fn sign_ssk(
+	server_keys: &server_keys::Service,
+	ssk: ServerSigningKeys,
+	server_name: &ServerName,
+	our_name: &ServerName,
+) -> Result<Raw<ServerSigningKeys>> {
+	if server_name == our_name {
+		// Should already be signed by us if we got this far.
+		debug_info!(
+			object=?ssk,
+			%server_name,
+			%our_name,
+			"Refusing to sign our own signing keys response (should already be signed)"
+		);
+		return to_raw_value(&ssk)
+			.map(Raw::<ServerSigningKeys>::from_json)
+			.map_err(Into::into);
+	}
+	let mut canonical = to_canonical_object(&ssk)?;
+	server_keys.sign_json(&mut canonical)?;
+	server_keys::strip_extraneous_signatures(&mut canonical, server_name, &[our_name]);
+	to_raw_value(&canonical)
+		.map(Raw::<ServerSigningKeys>::from_json)
+		.map_err(Into::into)
+}
+
+#[tracing::instrument(skip(server_keys, my_name, queries, start, semaphore))]
+async fn acquire_keys_as_notary(
+	server_keys: Arc<server_keys::Service>,
+	my_name: OwnedServerName,
+	remote: OwnedServerName,
+	mut queries: BTreeMap<OwnedServerSigningKeyId, QueryCriteria>,
+	start: MilliSecondsSinceUnixEpoch,
+	semaphore: Option<Arc<tokio::sync::Semaphore>>,
+) -> Vec<Raw<ServerSigningKeys>> {
+	let cost = queries
+		.len()
+		.saturating_add(1)
+		.min(automatic_width())
+		.try_into()
+		.unwrap_or(u32::MAX);
+	let _permit = if let Some(sem) = semaphore {
+		sem.acquire_many_owned(cost).await.ok()
+	} else {
+		None
+	};
+	let fetch_all = queries.is_empty();
+	let mut results = Vec::with_capacity(queries.len().max(1));
+	let mut keymap = HashMap::with_capacity(queries.len().max(1));
+
+	// First contact the origin (if we're allowed to)
+	if server_keys.notary_may_contact_origin(&remote) {
+		debug_info!("Asking remote directly for verify keys");
+		if let Ok(res) = server_keys.origin_request(remote.clone(), start).await
+			&& let Ok(ssk) = sign_ssk(&server_keys, res.clone(), &remote, &my_name).await
+		{
+			let index = results.len();
+			results.push(ssk);
+			for key_id in res.verify_keys.keys().chain(res.old_verify_keys.keys()) {
+				if fetch_all
+					|| queries.get(key_id).is_some_and(|c| {
+						c.minimum_valid_until_ts
+							.is_none_or(|m| res.valid_until_ts >= m)
+					}) {
+					queries.remove(key_id);
+					keymap.insert(key_id.to_owned(), index);
 				}
-				old_keys
-					.entry(old_key_id)
-					.and_modify(|current_old_key: &mut OldVerifyKey| {
-						// Use the lowest expiry timestamp
-						if current_old_key.expired_ts > old_key.expired_ts {
-							*current_old_key = old_key.clone();
-						}
-					})
-					.or_insert(old_key);
 			}
 		}
-		old_keys
-	};
+	} else {
+		debug!("Not asking remote for keys (already asked recently)");
+	}
+	debug!(keys=?keymap.keys(), "Live verify keys");
+	if fetch_all && keymap.is_empty() {
+		// If the server asked for all keys, AND we didn't get anything from the
+		// origin, just fetch any fresh responses we have.
+		server_keys
+			.signing_keys_for(&remote)
+			.broad_filter_map(|ssk| {
+				let server_name = &remote;
+				let my_name = &my_name;
+				let server_keys = &server_keys;
+				async move {
+					if ssk.valid_until_ts > in_one_week() || ssk.valid_until_ts < start {
+						return None;
+					}
 
-	let server_key = assign!(ServerSigningKeys::new(server_name.to_owned(), valid_until_ts()), {
-		verify_keys,
-		old_verify_keys,
-	});
+					sign_ssk(server_keys, ssk.clone(), server_name, my_name)
+						.await
+						.map(|signed_ssk| (ssk, signed_ssk))
+						.ok()
+				}
+			})
+			.ready_for_each(|(ssk, signed_ssk)| {
+				let idx = results.len();
+				results.push(signed_ssk);
+				for key_id in ssk.verify_keys.keys().chain(ssk.old_verify_keys.keys()) {
+					keymap.insert(key_id.to_owned(), idx);
+				}
+			})
+			.await;
+	}
 
-	let server_key = Raw::new(&server_key)?;
-	let mut response = get_server_keys::v2::Response::new(server_key)
-		.try_into_http_response::<Vec<u8>>()
-		.map(|mut response| take(response.body_mut()))
-		.and_then(|body| serde_json::from_slice(&body).map_err(Into::into))?;
+	// If we're still missing some keys, fetch them from the local cache
+	for (key_id, criteria) in queries {
+		debug!(%key_id, "Fetching verify key from local repository");
 
-	services.server_keys.sign_json(&mut response)?;
+		if keymap.contains_key(&key_id) {
+			debug!(%key_id, "Already found key");
+			continue;
+		}
 
-	Ok(Json(response))
+		let minimum_valid_until_ts = criteria
+			.clone()
+			.minimum_valid_until_ts
+			.unwrap_or_else(|| MilliSecondsSinceUnixEpoch(uint!(0)));
+
+		let Some(ssk) = server_keys.get_signing_key(&remote, &key_id).await else {
+			debug!(%remote, %key_id, "Could not find a matching signing key locally.");
+			continue;
+		};
+
+		if ssk.valid_until_ts < minimum_valid_until_ts || ssk.valid_until_ts > in_one_week() {
+			debug!(
+				%key_id,
+				valid_until_ts=?ssk.valid_until_ts,
+				?minimum_valid_until_ts,
+				"Stored verify key does not satisfy query criteria"
+			);
+			continue;
+		}
+		if let Some(old_verify_key) = ssk.old_verify_keys.get(&key_id) {
+			if old_verify_key.expired_ts < minimum_valid_until_ts {
+				debug!(
+					%key_id,
+					expired_ts=?old_verify_key.expired_ts,
+					?minimum_valid_until_ts,
+					"Stored old verify key does not satisfy query criteria"
+				);
+				continue;
+			}
+		}
+
+		debug!(%key_id, ?ssk, "Found key locally");
+		let rep_key_ids = ssk.verify_keys.keys().cloned().collect::<Vec<_>>();
+		let Ok(signed_ssk) = sign_ssk(&server_keys, ssk.clone(), &remote, &my_name)
+			.await
+			.inspect_err(
+				|e| error!(%key_id, %remote, "Failed to sign signing keys chunk: {e:?}"),
+			)
+		else {
+			continue;
+		};
+
+		let idx = results.len();
+		results.push(signed_ssk);
+		keymap.insert(key_id, idx);
+		for key_id in rep_key_ids {
+			keymap.insert(key_id, idx);
+		}
+	}
+
+	keymap
+		.into_values()
+		.collect::<HashSet<_>>()
+		.into_iter()
+		.map(|idx| results.index(idx).to_owned())
+		.collect()
 }
 
-fn valid_until_ts() -> MilliSecondsSinceUnixEpoch {
-	let dur = Duration::from_hours(12);
-	let timepoint = timepoint_from_now(dur).expect("SystemTime should not overflow");
-	MilliSecondsSinceUnixEpoch::from_system_time(timepoint).expect("UInt should not overflow")
+/// `GET /_matrix/key/v2/query/{serverName}`
+///
+/// Fetches the latest keys for a specific server name.
+pub(crate) async fn get_remote_server_keys_route(
+	State(services): State<crate::State>,
+	body: Ruma<get_remote_server_keys::v2::Request>,
+) -> Result<get_remote_server_keys::v2::Response> {
+	let min_valid_ts = body.minimum_valid_until_ts;
+	if services
+		.server_keys
+		.notary_may_contact_origin(&body.server_name)
+		&& let Ok(response) = services
+			.server_keys
+			.origin_request(body.server_name.clone(), min_valid_ts)
+			.await
+	{
+		return sign_ssk(
+			&services.server_keys,
+			response,
+			&body.server_name,
+			services.globals.server_name(),
+		)
+		.await
+		.map(|r| Ok(get_remote_server_keys::v2::Response::new(vec![r])))?;
+	}
+
+	let response = services
+		.server_keys
+		.signing_keys_for(&body.server_name)
+		.broad_filter_map(|ssk| {
+			let server_name = body.server_name.clone();
+			async move {
+				if ssk.valid_until_ts > in_one_week() || ssk.valid_until_ts < min_valid_ts {
+					return None;
+				}
+
+				sign_ssk(
+					&services.server_keys,
+					ssk,
+					server_name.as_ref(),
+					services.globals.server_name(),
+				)
+				.await
+				.ok()
+			}
+		})
+		.collect::<Vec<_>>()
+		.await;
+	Ok(get_remote_server_keys::v2::Response::new(response))
 }

src/core/config/mod.rs

@@ -1132,10 +1132,7 @@ pub struct Config {
 	/// (trusted key notary servers), as well as serve as trusted servers for
 	/// other operations like backfill and event fetching.
 	///
-	/// Currently, continuwuity doesn't support inbound key requests, so
-	/// this list should only contain other Synapse servers.
-	///
-	/// example: ["matrix.org", "starstruck.systems"]
+	/// example: ["continuwuity.org", "matrix.org"]
 	///
 	/// It is possible to restrict which signing keys trusted servers are
 	/// allowed to sign responses with. Without configuring this, all responses
@@ -1172,7 +1169,7 @@ pub struct Config {
 	/// empty list (`[]`) to operate without trusted server assistance, but this
 	/// is discouraged for performance and reliability reasons.
 	///
-	/// default: ["matrix.org"]
+	/// default: ["continuwuity.org", "continuwuity.rocks", "matrix.org"]
 	#[serde(default = "default_trusted_servers")]
 	pub trusted_servers: Vec<TrustedServer>,
 
@@ -3262,12 +3259,11 @@ fn default_otlp_filter() -> String {
 fn default_tracing_flame_output_path() -> String { "./tracing.folded".to_owned() }
 
 fn default_trusted_servers() -> Vec<TrustedServer> {
-	// TODO(nex): Once we can be a notary, add maintainer(?) homeservers here.
-	// Rationale: Users are already running our code, arguably that's a higher
-	// level of trust than is assigned to notaries in the first place.
-	// We should still remind everyone that notaries are evil and out to get you
-	// and to replace this list with servers they actually trust.
-	let trusted_servers = vec![ruma::owned_server_name!("matrix.org")];
+	let trusted_servers = vec![
+		ruma::owned_server_name!("continuwuity.org"),
+		ruma::owned_server_name!("continuwuity.rocks"),
+		ruma::owned_server_name!("matrix.org"),
+	];
 	trusted_servers
 		.into_iter()
 		.map(TrustedServer::Name)

src/ruminuwuity/federation/mod.rs

@@ -0,0 +1 @@
+pub mod notary;

src/ruminuwuity/federation/notary.rs

@@ -0,0 +1,58 @@
+//! `POST /_matrix/key/*/query`
+//!
+//! Endpoint to query signing keys through other (notary) servers.
+
+pub mod unstable {
+	use std::collections::BTreeMap;
+
+	use ruma::{
+		OwnedServerName, OwnedServerSigningKeyId,
+		api::{
+			federation::{
+				authentication::ServerSignatures,
+				discovery::{ServerSigningKeys, get_remote_server_keys_batch::v2::QueryCriteria},
+			},
+			request, response,
+		},
+		metadata,
+		serde::Raw,
+	};
+
+	metadata! {
+		method: POST,
+		rate_limited: true,
+		authentication: ServerSignatures,
+		history: {
+			unstable("org.continuwuity.msc4553") => "/_matrix/key/unstable/org.continuwuity.msc4556/query",
+		}
+	}
+
+	#[request]
+	pub struct Request {
+		pub server_keys:
+			BTreeMap<OwnedServerName, BTreeMap<OwnedServerSigningKeyId, QueryCriteria>>,
+	}
+
+	#[response]
+	pub struct Response {
+		pub server_keys: Vec<Raw<ServerSigningKeys>>,
+	}
+
+	impl Request {
+		#[must_use]
+		pub fn new(
+			server_keys: BTreeMap<
+				OwnedServerName,
+				BTreeMap<OwnedServerSigningKeyId, QueryCriteria>,
+			>,
+		) -> Self {
+			Self { server_keys }
+		}
+	}
+
+	impl Response {
+		/// Creates a new `Response` with the given keys.
+		#[must_use]
+		pub fn new(server_keys: Vec<Raw<ServerSigningKeys>>) -> Self { Self { server_keys } }
+	}
+}

src/ruminuwuity/mod.rs

@@ -6,3 +6,5 @@ pub mod invite_permission_config;
 pub mod meowlnir_antispam;
 pub mod policy;
 pub mod pushers;
+
+pub mod federation;

src/service/server_keys/mod.rs

@@ -5,30 +5,42 @@ mod sign;
 mod util;
 mod verify;
 
-use std::{collections::BTreeMap, sync::Arc};
+use std::{
+	collections::{BTreeMap, HashMap},
+	sync::Arc,
+	time::{Duration, Instant},
+};
 
+use assign::assign;
+use async_trait::async_trait;
 use conduwuit::{
-	Result, Server,
-	utils::{IterStream, ReadyExt, stream::TryIgnore},
+	Result, Server, SyncRwLock,
+	utils::{IterStream, ReadyExt, stream::TryIgnore, timepoint_from_now, to_canonical_object},
 };
 use database::{Deserialized, Ignore, Interfix, Json, Map};
 use futures::{Stream, StreamExt};
 pub use request::in_one_week;
 use ruma::{
-	CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedServerSigningKeyId, ServerName,
-	ServerSigningKeyId,
-	api::federation::discovery::{ServerSigningKeys, VerifyKey},
+	CanonicalJsonObject, MilliSecondsSinceUnixEpoch, OwnedServerName, OwnedServerSigningKeyId,
+	ServerName, ServerSigningKeyId,
+	api::federation::discovery::{OldVerifyKey, ServerSigningKeys, VerifyKey},
 	room_version_rules::RoomVersionRules,
+	serde::Raw,
 	signatures::{Ed25519KeyPair, PublicKeyMap, PublicKeySet},
 };
+use serde_json::value::to_raw_value;
+pub use verify::strip_extraneous_signatures;
 
 use crate::{Dep, globals, sending, server_keys::util::required_keys};
 
+const VALID_UNTIL_TS_DURATION: Duration = Duration::from_hours(12);
+
 pub struct Service {
 	keypair: Box<Ed25519KeyPair>,
 	verify_keys: VerifyKeys,
 	services: Services,
 	db: Data,
+	last_lookup: SyncRwLock<HashMap<OwnedServerName, Instant>>,
 }
 
 struct Services {
@@ -45,6 +57,7 @@ pub type VerifyKeys = BTreeMap<OwnedServerSigningKeyId, VerifyKey>;
 pub type PubKeyMap = PublicKeyMap;
 pub type PubKeys = PublicKeySet;
 
+#[async_trait]
 impl crate::Service for Service {
 	fn build(args: crate::Args<'_>) -> Result<Arc<Self>> {
 		let (keypair, verify_keys) = keypair::init(args.db)?;
@@ -61,9 +74,12 @@ impl crate::Service for Service {
 			db: Data {
 				servernamekeyid_response: args.db["servernamekeyid_response"].clone(),
 			},
+			last_lookup: SyncRwLock::new(HashMap::new()),
 		}))
 	}
 
+	async fn clear_cache(&self) { self.last_lookup.write().clear(); }
+
 	fn name(&self) -> &str { crate::service::make_name(std::module_path!()) }
 }
 
@@ -183,6 +199,8 @@ impl Service {
 	}
 
 	/// Returns the stored server signing keys responses for the origin.
+	///
+	/// Does not imply that the stored responses are still in-date.
 	pub fn signing_keys_for<'a>(
 		&'a self,
 		origin: &'a ServerName,
@@ -193,4 +211,91 @@ impl Service {
 			.ignore_err()
 			.map(|(_, v): (Ignore, ServerSigningKeys)| v)
 	}
+
+	/// Determines if the server may contact the origin server to fetch keys
+	/// when acting as a notary server. This limits origin lookups to once per
+	/// minute, which prevents amplification attacks.
+	#[must_use]
+	pub fn notary_may_contact_origin(&self, server_name: &ServerName) -> bool {
+		if server_name == self.services.server.name {
+			// We always intercept origin requests to us with
+			// build_server_keys_response, so this is equivalent to
+			// calling `/_matrix/key/v2/server`.
+			return true;
+		}
+		self.last_lookup
+			.read()
+			.get(server_name)
+			.is_none_or(|last| last.elapsed() >= Duration::from_mins(1))
+	}
+
+	/// Generates a signing key response to serve to `/_matrix/key/v2/server`.
+	pub async fn build_server_keys_response(&self) -> Result<Raw<ServerSigningKeys>> {
+		let verify_keys = BTreeMap::from([self.active_verify_key()]);
+		let old_verify_keys = match &self.services.server.config.old_verify_keys {
+			| Some(old_verify_keys) => old_verify_keys.to_owned(),
+			| None => self.discover_old_signing_keys().await,
+		};
+		let valid_until_ts = timepoint_from_now(VALID_UNTIL_TS_DURATION)
+			.map(|tp| {
+				MilliSecondsSinceUnixEpoch::from_system_time(tp)
+					.expect("key validity period must be before the heat death of the universe")
+			})
+			.expect("key validity period must be before the heat death of the universe");
+		let server_keys = assign!(
+			ServerSigningKeys::new(self.services.server.name.clone(), valid_until_ts),
+			{verify_keys, old_verify_keys}
+		);
+		let mut canonical_obj = to_canonical_object(&server_keys)?;
+		self.sign_json(&mut canonical_obj)
+			.and_then(|()| to_raw_value(&canonical_obj).map_err(Into::into))
+			.map(Raw::from_json)
+	}
+
+	async fn discover_old_signing_keys(&self) -> BTreeMap<OwnedServerSigningKeyId, OldVerifyKey> {
+		let now = MilliSecondsSinceUnixEpoch::now();
+		let mut keys = self.signing_keys_for(&self.services.server.name);
+		let mut old_keys = BTreeMap::new();
+		while let Some(resp) = keys.next().await {
+			// We need to verify that this response is still trusted based on
+			// the current configuration. It's possible we got this response
+			// from a notary who we no longer trust.
+			if resp.valid_until_ts > in_one_week() || resp.valid_until_ts < now {
+				continue;
+			}
+			let trusted = self
+				.services
+				.server
+				.config
+				.trusted_servers
+				.iter()
+				.any(|notary| {
+					Self::verify_server_keys_response(
+						&resp,
+						Some((notary.server_name(), notary.verify_keys())),
+					)
+					.is_ok()
+				});
+			if !trusted {
+				continue;
+			}
+			for (old_key_id, old_key) in resp.old_verify_keys {
+				if old_key.expired_ts > resp.valid_until_ts {
+					// This key expired in the future which is probably illegal
+					continue;
+				}
+				old_keys
+					.entry(old_key_id)
+					.and_modify(|current_old_key: &mut OldVerifyKey| {
+						// Use the lowest expiry timestamp
+						if current_old_key.expired_ts < old_key.expired_ts {
+							*current_old_key = old_key.clone();
+						}
+					})
+					.or_insert(old_key);
+			}
+		}
+
+		old_keys
+	}
 }

src/service/server_keys/request.rs

@@ -1,4 +1,4 @@
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, time::Instant};
 
 use conduwuit::{Err, Result, trace, utils::millis_since_unix_epoch, warn};
 use ruma::{
@@ -30,6 +30,23 @@ impl super::Service {
 	) -> Result<ServerSigningKeys> {
 		use get_server_keys::v2::Request;
 
+		if target == self.services.server.name {
+			return self
+				.build_server_keys_response()
+				.await
+				.and_then(|r| r.deserialize().map_err(Into::into));
+		}
+
+		// N.B. The "last lookup" is written before the request is actually made
+		// to prevent concurrent notary requests from spawning... concurrent
+		// origin requests, especially if the origin is slow.
+		// This has the downside that, if the origin is temporarily unreachable
+		// (including if we're backing off from it), the notary might take an
+		// additional minute to recover compared to the rest of the server. This
+		// is deemed acceptable.
+		self.last_lookup
+			.write()
+			.insert(target.clone(), Instant::now());
 		let server_signing_key = self
 			.services
 			.sending

src/service/server_keys/verify.rs

@@ -78,7 +78,7 @@ impl super::Service {
 		pubkey_map.insert(server_keys.server_name.to_string(), set);
 
 		let mut canonical = to_canonical_object(server_keys)?;
-		Self::strip_extraneous_signatures(&mut canonical, &server_keys.server_name, &[]);
+		strip_extraneous_signatures(&mut canonical, &server_keys.server_name, &[]);
 		ruma::signatures::verify_json(&pubkey_map, &canonical).map_err(Into::into)
 	}
 
@@ -100,13 +100,12 @@ impl super::Service {
 			)));
 		};
 		let mut canonical_object = to_canonical_object(notary_signatures)?;
-		Self::strip_extraneous_signatures(&mut canonical_object, &server_keys.server_name, &[
+		strip_extraneous_signatures(&mut canonical_object, &server_keys.server_name, &[
 			notary_name,
 		]);
-		let for_verify = ruma::signatures::to_canonical_json_string_for_signing(
-			&to_canonical_object(server_keys).expect("server keys object must be canonical json"),
-		)
-		.expect("canonical JSON must be stringable");
+		let for_verify =
+			ruma::signatures::to_canonical_json_string_for_signing(&canonical_object)
+				.expect("canonical JSON must be stringable");
 
 		let raw_notary_keys = notary_keys.iter().map(Base64::as_bytes).collect::<Vec<_>>();
 		let raw_notary_signatures = notary_signatures
@@ -137,22 +136,22 @@ impl super::Service {
 			"No valid signature from {notary_name} present on signing keys response"
 		)))
 	}
+}
 
-	fn strip_extraneous_signatures(
-		canonical: &mut CanonicalJsonObject,
-		origin: &ServerName,
-		notaries: &[&ServerName],
-	) {
-		canonical.entry("signatures".to_owned()).and_modify(|sigs| {
-			sigs.as_object_mut().map(|s| {
-				s.retain(|server_name, _| {
-					let Ok(server_name) = ServerName::parse(server_name) else { return false };
-					origin == server_name || notaries.iter().any(|ns| *ns == server_name)
-				});
-				Some(s)
+pub fn strip_extraneous_signatures(
+	canonical: &mut CanonicalJsonObject,
+	origin: &ServerName,
+	notaries: &[&ServerName],
+) {
+	canonical.entry("signatures".to_owned()).and_modify(|sigs| {
+		sigs.as_object_mut().map(|s| {
+			s.retain(|server_name, _| {
+				let Ok(server_name) = ServerName::parse(server_name) else { return false };
+				origin == server_name || notaries.iter().any(|ns| *ns == server_name)
 			});
+			Some(s)
 		});
-	}
+	});
 }
 
 #[cfg(test)]