mail4agent-server 0.3.0

//! Federation F0 routes: this server's signing keys, a version probe, one
//! signed-request-protected query (profile existence), and the `X-Matrix`
//! authentication helper later federation routes call. Everything is off
//! (`M_UNRECOGNIZED`) unless federation is enabled in config.

use std::collections::HashMap;
use std::sync::Arc;

use axum::extract::{OriginalUri, Path, Query, State};
use axum::http::{header, HeaderMap, Uri};
use axum::body::Bytes;
use axum::routing::{get, post, put};
use axum::{Json, Router};
use serde_json::{json, Value};

use super::{with_conn_pub, Homeserver};
use crate::error::MatrixError;
use crate::fed_rooms as fr;
use crate::federation as fed;

pub(super) fn routes() -> Router<Arc<Homeserver>> {
    Router::new()
        .route("/key/v2/server", get(own_keys))
        .route("/key/v2/server/{key_id}", get(own_keys_by_id))
        .route("/federation/v1/version", get(version))
        .route("/federation/v1/query/profile", get(profile))
        .route("/federation/v1/send/{txn_id}", put(send_txn))
        .route("/federation/v1/make_join/{room_id}/{user_id}", get(make_join))
        .route("/federation/v2/send_join/{room_id}/{event_id}", put(send_join))
        .route("/federation/v2/invite/{room_id}/{event_id}", put(invite))
        .route("/federation/v1/backfill/{room_id}", get(backfill))
        .route("/federation/v1/user/keys/query", post(keys_query))
        .route("/federation/v1/user/keys/claim", post(keys_claim))
        .route("/federation/v1/user/devices/{user_id}", get(user_devices))
}

fn require_enabled(state: &Homeserver) -> Result<(), MatrixError> {
    state.federation_enabled.get().map(|_| ()).ok_or_else(MatrixError::unrecognized)
}

async fn own_keys(State(state): State<Arc<Homeserver>>) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let server = crate::store::matrix_server_name().to_string();
    let doc = with_conn_pub(&state, move |c| fed::server_keys_response(c, &server, fed::now_ms()).map_err(|_| MatrixError::internal())).await?;
    Ok(Json(doc))
}

async fn own_keys_by_id(State(state): State<Arc<Homeserver>>, Path(_key_id): Path<String>) -> Result<Json<Value>, MatrixError> {
    // The full document is returned whatever id is asked for (spec-permitted).
    own_keys(State(state)).await
}

async fn version(State(state): State<Arc<Homeserver>>) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    Ok(Json(json!({ "server": { "name": "mail4agent", "version": env!("CARGO_PKG_VERSION") } })))
}

async fn profile(
    State(state): State<Arc<Homeserver>>,
    OriginalUri(uri): OriginalUri,
    headers: HeaderMap,
    Query(q): Query<HashMap<String, String>>,
) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    authenticate(&state, "GET", &uri, &headers, &[]).await?;
    let user_id = q.get("user_id").cloned().ok_or_else(|| MatrixError::invalid_param("user_id required"))?;
    let known = with_conn_pub(&state, move |c| {
        let n: i64 = c
            .query_row("SELECT COUNT(*) FROM matrix_users WHERE mxid = ?1", [&user_id], |r| r.get(0))
            .map_err(|_| MatrixError::internal())?;
        Ok(n > 0)
    })
    .await?;
    if known {
        Ok(Json(json!({})))
    } else {
        Err(MatrixError::not_found("user not found"))
    }
}

fn internal<E>(_: E) -> MatrixError {
    MatrixError::internal()
}

fn rfc3339() -> String {
    chrono::Utc::now().to_rfc3339()
}

fn parse_body(body: &Bytes) -> Result<Value, MatrixError> {
    serde_json::from_slice(body).map_err(|_| MatrixError::bad_json("body is not JSON"))
}

/// Wake local users and poke the outbox (events ingested may need relaying).
fn after_ingest(state: &Arc<Homeserver>, wake: impl IntoIterator<Item = i64>) {
    super::wake_users(state, wake.into_iter().filter(|id| *id > 0));
    state.fed_notify.notify_one();
}

async fn send_txn(
    State(state): State<Arc<Homeserver>>,
    OriginalUri(uri): OriginalUri,
    headers: HeaderMap,
    Path(_txn): Path<String>,
    body: Bytes,
) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "PUT", &uri, &headers, &body).await?;
    let v = parse_body(&body)?;
    if v.get("origin").and_then(Value::as_str) != Some(origin.as_str()) {
        return Err(MatrixError::forbidden("origin in body does not match the signing server"));
    }
    let pdus = v.get("pdus").and_then(Value::as_array).cloned().unwrap_or_default();
    let edus = v.get("edus").and_then(Value::as_array).cloned().unwrap_or_default();
    if pdus.len() > 50 || edus.len() > 100 {
        return Err(MatrixError::invalid_param("too many pdus or edus"));
    }
    let mut results = serde_json::Map::new();
    let mut wake_all: Vec<i64> = Vec::new();
    for pdu in pdus {
        let id = pdu.get("event_id").and_then(Value::as_str).unwrap_or("?").to_string();
        let outcome = match super::fed_net::verify_pdu(&state, &pdu).await {
            Ok(content_ok) => {
                let p = pdu.clone();
                with_conn_pub(&state, move |c| fr::ingest_pdu(c, &p, content_ok, fr::Mode::Live, &rfc3339())).await
            }
            Err(e) => Err(e),
        };
        match outcome {
            Ok(i) => {
                wake_all.extend(i.wake);
                results.insert(id, json!({}));
            }
            Err(e) => {
                results.insert(id, json!({ "error": e.error }));
            }
        }
    }
    for edu in edus {
        if edu.get("edu_type").and_then(Value::as_str) == Some("m.direct_to_device") {
            let content = edu.get("content").cloned().unwrap_or(Value::Null);
            let origin2 = origin.clone();
            if let Ok(ids) = with_conn_pub(&state, move |c| apply_direct_to_device(c, &origin2, &content)).await {
                wake_all.extend(ids);
            }
        }
    }
    after_ingest(&state, wake_all);
    Ok(Json(json!({ "pdus": results })))
}

fn apply_direct_to_device(conn: &mut rusqlite::Connection, origin: &str, content: &Value) -> Result<Vec<i64>, MatrixError> {
    let sender = content.get("sender").and_then(Value::as_str).unwrap_or_default();
    if fr::domain_of(sender) != Some(origin) {
        return Err(MatrixError::forbidden("to-device sender is not on the origin server"));
    }
    let sender_uid = fr::ensure_remote_user(conn, sender, &rfc3339()).map_err(|_| MatrixError::bad_json("sender"))?;
    let ty = content.get("type").and_then(Value::as_str).unwrap_or_default().to_string();
    let message_id = content.get("message_id").and_then(Value::as_str).unwrap_or_default().to_string();
    let messages: std::collections::BTreeMap<String, std::collections::BTreeMap<String, Value>> =
        serde_json::from_value(content.get("messages").cloned().unwrap_or(Value::Null)).map_err(|_| MatrixError::bad_json("messages"))?;
    if ty.is_empty() || message_id.is_empty() {
        return Err(MatrixError::bad_json("type/message_id"));
    }
    match crate::key_ops::apply_send_to_device(conn, sender_uid, &format!("fed:{origin}"), &ty, &message_id, &messages, &rfc3339())? {
        crate::key_ops::SendToDeviceOutcome::New(ids) => Ok(ids.into_iter().collect()),
        crate::key_ops::SendToDeviceOutcome::AlreadySent => Ok(Vec::new()),
    }
}

async fn make_join(
    State(state): State<Arc<Homeserver>>,
    OriginalUri(uri): OriginalUri,
    headers: HeaderMap,
    Path((room_id, user_id)): Path<(String, String)>,
) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "GET", &uri, &headers, &[]).await?;
    if fr::domain_of(&user_id) != Some(origin.as_str()) {
        return Err(MatrixError::forbidden("user does not belong to the requesting server"));
    }
    with_conn_pub(&state, move |c| {
        let room = crate::store::get_room(c, &room_id)?.ok_or_else(|| MatrixError::not_found("unknown room"))?;
        let current = match crate::store::user_id_of(c, &user_id)? {
            Some(u) => crate::store::room_member(c, &room_id, u)?.map(|m| m.membership),
            None => None,
        };
        match current {
            Some(crate::store::Membership::Ban) => return Err(MatrixError::forbidden("banned")),
            Some(crate::store::Membership::Invite) | Some(crate::store::Membership::Join) => {}
            _ if room.join_rule == crate::store::JoinRule::Public => {}
            _ => return Err(MatrixError::forbidden("room is not open to this user")),
        }
        Ok(Json(json!({
            "room_version": room.room_version,
            "event": {
                "type": "m.room.member", "state_key": user_id, "sender": user_id, "room_id": room_id,
                "origin_server_ts": fed::now_ms(), "content": { "membership": "join" }
            }
        })))
    })
    .await
}

async fn send_join(
    State(state): State<Arc<Homeserver>>,
    OriginalUri(uri): OriginalUri,
    headers: HeaderMap,
    Path((room_id, event_id)): Path<(String, String)>,
    body: Bytes,
) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "PUT", &uri, &headers, &body).await?;
    let pdu = parse_body(&body)?;
    let sender = pdu.get("sender").and_then(Value::as_str).unwrap_or_default().to_string();
    let is_join = pdu.get("type").and_then(Value::as_str) == Some("m.room.member")
        && pdu.get("content").and_then(|c| c.get("membership")).and_then(Value::as_str) == Some("join")
        && pdu.get("state_key").and_then(Value::as_str) == Some(sender.as_str());
    if !is_join
        || fr::domain_of(&sender) != Some(origin.as_str())
        || pdu.get("event_id").and_then(Value::as_str) != Some(event_id.as_str())
        || pdu.get("room_id").and_then(Value::as_str) != Some(room_id.as_str())
    {
        return Err(MatrixError::bad_json("not a valid join event for this request"));
    }
    let content_ok = super::fed_net::verify_pdu(&state, &pdu).await?;
    let local = crate::store::matrix_server_name().to_string();
    let (p, origin2) = (pdu.clone(), origin.clone());
    let resp = with_conn_pub(&state, move |c| {
        let ing = fr::ingest_pdu(c, &p, content_ok, fr::Mode::Live, &rfc3339())?;
        let now = fed::now_ms();
        if !ing.duplicate {
            fr::relay_pdu(c, &room_id, &p, &[origin2.as_str()], now).map_err(internal)?;
        }
        let mut state_pdus = Vec::new();
        for ev in crate::store::current_state_all(c, &room_id)? {
            if let Ok(x) = fr::pdu_for_event(c, &ev, &local, now) {
                state_pdus.push(x);
            }
        }
        let mut timeline = Vec::new();
        for ev in crate::store::events_in_room_before(c, &room_id, i64::MAX, 50)? {
            if ev.state_key.is_none() {
                if let Ok(x) = fr::pdu_for_event(c, &ev, &local, now) {
                    timeline.push(x);
                }
            }
        }
        timeline.reverse();
        let info = fr::room_info(c, &room_id).map_err(internal)?;
        Ok((ing.wake, json!({ "origin": local, "state": state_pdus, "auth_chain": [], "members_omitted": false, "m4a_room": info, "m4a_timeline": timeline })))
    })
    .await?;
    after_ingest(&state, resp.0);
    Ok(Json(resp.1))
}

async fn invite(
    State(state): State<Arc<Homeserver>>,
    OriginalUri(uri): OriginalUri,
    headers: HeaderMap,
    Path((room_id, event_id)): Path<(String, String)>,
    body: Bytes,
) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "PUT", &uri, &headers, &body).await?;
    let v = parse_body(&body)?;
    let event = v.get("event").cloned().ok_or_else(|| MatrixError::bad_json("event"))?;
    let info = v.get("room_info").cloned().ok_or_else(|| MatrixError::bad_json("room_info"))?;
    let state_pdus = v.get("invite_room_state").and_then(Value::as_array).cloned().unwrap_or_default();
    let sender = event.get("sender").and_then(Value::as_str).unwrap_or_default();
    let invitee = event.get("state_key").and_then(Value::as_str).unwrap_or_default().to_string();
    let is_invite = event.get("type").and_then(Value::as_str) == Some("m.room.member")
        && event.get("content").and_then(|c| c.get("membership")).and_then(Value::as_str) == Some("invite");
    if !is_invite
        || fr::domain_of(sender) != Some(origin.as_str())
        || event.get("event_id").and_then(Value::as_str) != Some(event_id.as_str())
        || event.get("room_id").and_then(Value::as_str) != Some(room_id.as_str())
        || fr::is_remote_mxid(&invitee)
    {
        return Err(MatrixError::bad_json("not a valid invite for this request"));
    }
    let mut checked: Vec<(Value, bool)> = Vec::new();
    for p in &state_pdus {
        if p.get("room_id").and_then(Value::as_str) != Some(room_id.as_str()) {
            return Err(MatrixError::bad_json("state event of another room"));
        }
        checked.push((p.clone(), super::fed_net::verify_pdu(&state, p).await?));
    }
    let ev_ok = super::fed_net::verify_pdu(&state, &event).await?;
    let (ev2, invitee2) = (event.clone(), invitee.clone());
    let wake = with_conn_pub(&state, move |c| {
        if crate::store::user_id_of(c, &invitee2)?.is_none() {
            return Err(MatrixError::not_found("unknown local user"));
        }
        let now = rfc3339();
        fr::create_replica_room(c, &room_id, &info, &now)?;
        for (p, ok) in &checked {
            fr::ingest_pdu(c, p, *ok, fr::Mode::Trusted, &now)?;
        }
        Ok(fr::ingest_pdu(c, &ev2, ev_ok, fr::Mode::Live, &now)?.wake)
    })
    .await?;
    after_ingest(&state, wake);
    Ok(Json(json!({ "event": event })))
}

#[derive(serde::Deserialize)]
struct BackfillQuery {
    v: Option<String>,
    limit: Option<i64>,
}

async fn backfill(
    State(state): State<Arc<Homeserver>>,
    OriginalUri(uri): OriginalUri,
    headers: HeaderMap,
    Path(room_id): Path<String>,
    Query(q): Query<BackfillQuery>,
) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "GET", &uri, &headers, &[]).await?;
    let local = crate::store::matrix_server_name().to_string();
    let limit = q.limit.unwrap_or(50).clamp(1, 100);
    with_conn_pub(&state, move |c| {
        if !fr::remote_domains(c, &room_id)?.contains(&origin) {
            return Err(MatrixError::forbidden("server has no member in this room"));
        }
        let before = match q.v.as_deref() {
            Some(id) => crate::store::get_event(c, id)?.map(|e| e.stream_id).ok_or_else(|| MatrixError::not_found("unknown event"))?,
            None => i64::MAX,
        };
        let mut pdus = Vec::new();
        for ev in crate::store::events_in_room_before(c, &room_id, before, limit)? {
            if let Ok(p) = fr::pdu_for_event(c, &ev, &local, fed::now_ms()) {
                pdus.push(p);
            }
        }
        Ok(Json(json!({ "origin": local, "origin_server_ts": fed::now_ms(), "pdus": pdus })))
    })
    .await
}

async fn keys_query(State(state): State<Arc<Homeserver>>, OriginalUri(uri): OriginalUri, headers: HeaderMap, body: Bytes) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "POST", &uri, &headers, &body).await?;
    let req: crate::key_ops::KeysQueryRequest = serde_json::from_slice(&body).map_err(|_| MatrixError::bad_json("keys query"))?;
    with_conn_pub(&state, move |c| {
        let visible = fr::users_visible_to_origin(c, &origin)?;
        let local_only: std::collections::BTreeMap<String, Vec<String>> = req.device_keys.into_iter().filter(|(m, _)| !fr::is_remote_mxid(m)).collect();
        crate::key_ops::build_keys_query_visible(c, &visible, None, &local_only).map(Json)
    })
    .await
}

async fn keys_claim(State(state): State<Arc<Homeserver>>, OriginalUri(uri): OriginalUri, headers: HeaderMap, body: Bytes) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "POST", &uri, &headers, &body).await?;
    let req: crate::key_ops::KeysClaimRequest = serde_json::from_slice(&body).map_err(|_| MatrixError::bad_json("keys claim"))?;
    with_conn_pub(&state, move |c| {
        let visible = fr::users_visible_to_origin(c, &origin)?;
        let local_only: std::collections::BTreeMap<String, std::collections::BTreeMap<String, String>> =
            req.one_time_keys.into_iter().filter(|(m, _)| !fr::is_remote_mxid(m)).collect();
        crate::key_ops::build_keys_claim_visible(c, &visible, &local_only).map(Json)
    })
    .await
}

async fn user_devices(
    State(state): State<Arc<Homeserver>>,
    OriginalUri(uri): OriginalUri,
    headers: HeaderMap,
    Path(user_id): Path<String>,
) -> Result<Json<Value>, MatrixError> {
    require_enabled(&state)?;
    let origin = authenticate(&state, "GET", &uri, &headers, &[]).await?;
    with_conn_pub(&state, move |c| {
        let uid = crate::store::user_id_of(c, &user_id)?.filter(|u| *u > 0).ok_or_else(|| MatrixError::not_found("unknown user"))?;
        if !fr::users_visible_to_origin(c, &origin)?.contains(&uid) {
            return Err(MatrixError::not_found("unknown user"));
        }
        let one: std::collections::BTreeMap<String, Vec<String>> = [(user_id.clone(), Vec::new())].into();
        let visible: std::collections::HashSet<i64> = [uid].into();
        let q = crate::key_ops::build_keys_query_visible(c, &visible, None, &one)?;
        let devices: Vec<Value> = q["device_keys"][&user_id].as_object().map(|m| m.iter().map(|(id, keys)| json!({ "device_id": id, "keys": keys })).collect()).unwrap_or_default();
        Ok(Json(json!({
            "user_id": user_id, "stream_id": 0, "devices": devices,
            "master_key": q["master_keys"].get(&user_id).cloned().unwrap_or(Value::Null),
            "self_signing_key": q["self_signing_keys"].get(&user_id).cloned().unwrap_or(Value::Null),
        })))
    })
    .await
}

/// Path+query as the sender signed it: always with the `/_matrix` prefix,
/// which an edge in front of the core may have stripped.
fn signed_uri(uri: &Uri) -> String {
    let pq = uri.path_and_query().map(|p| p.as_str()).unwrap_or("/");
    if pq.starts_with("/_matrix/") {
        pq.to_string()
    } else {
        format!("/_matrix{pq}")
    }
}

/// Authenticate a federation request from its `X-Matrix` header and return
/// the verified origin server name. `body` is the raw request body (empty for GET).
pub(crate) async fn authenticate(
    state: &Arc<Homeserver>,
    method: &str,
    uri: &Uri,
    headers: &HeaderMap,
    body: &[u8],
) -> Result<String, MatrixError> {
    let value = headers
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .ok_or_else(|| MatrixError::unauthorized("missing X-Matrix authorization"))?;
    let xm = fed::parse_x_matrix(value).map_err(|e| MatrixError::unauthorized(e.to_string()))?;
    let local = crate::store::matrix_server_name();
    if let Some(dest) = &xm.destination {
        if !crate::store::is_local_server_name(dest) {
            return Err(MatrixError::unauthorized("request is addressed to another server"));
        }
    }
    if fed::parse_server_name(&xm.origin).is_none() || crate::store::is_local_server_name(&xm.origin) {
        return Err(MatrixError::unauthorized("invalid origin"));
    }
    if let Some(allow) = state.federation_allow.get() {
        if !allow.iter().any(|a| a.eq_ignore_ascii_case(&xm.origin)) {
            return Err(MatrixError::forbidden("origin is not on the federation allowlist"));
        }
    }
    let content: Option<Value> = if body.is_empty() {
        None
    } else {
        Some(serde_json::from_slice(body).map_err(|_| MatrixError::bad_json("body is not JSON"))?)
    };
    let public_key = remote_key(state, &xm.origin, &xm.key).await?;
    fed::verify_request_signature(&xm, method, &signed_uri(uri), local, content.as_ref(), &public_key)
        .map_err(|_| MatrixError::unauthorized("bad request signature"))?;
    Ok(xm.origin)
}

/// Public key of `server` for `key_id`: cache first, then one resolved fetch
/// (rate-limited per server), validated before it is stored.
pub(crate) async fn remote_key(state: &Arc<Homeserver>, server: &str, key_id: &str) -> Result<String, MatrixError> {
    let now = fed::now_ms();
    let (s, k) = (server.to_string(), key_id.to_string());
    let cached = with_conn_pub(state, move |c| fed::cached_remote_key(c, &s, &k, now).map_err(|_| MatrixError::internal())).await?;
    if let Some(pk) = cached {
        return Ok(pk);
    }
    let s = server.to_string();
    if !with_conn_pub(state, move |c| fed::may_refetch(c, &s, now).map_err(|_| MatrixError::internal())).await? {
        return Err(MatrixError::unauthorized("unknown signing key"));
    }
    let fetcher = state.key_fetcher.get().cloned().ok_or_else(|| MatrixError::unauthorized("remote key fetching is not configured"))?;
    let resp = fetcher.fetch_server_keys(server).await.map_err(|e| MatrixError::unauthorized(format!("cannot fetch keys: {e}")))?;
    let parsed = fed::parse_server_keys(&resp, server, now).map_err(|e| MatrixError::unauthorized(format!("invalid keys: {e}")))?;
    let found = parsed.keys.iter().find(|(id, _)| id == key_id).map(|(_, pk)| pk.clone());
    let (s, p) = (server.to_string(), parsed);
    with_conn_pub(state, move |c| fed::store_remote_keys(c, &s, &p, now).map_err(|_| MatrixError::internal())).await?;
    found.ok_or_else(|| MatrixError::unauthorized("unknown signing key"))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::federation::{FetchFuture, RemoteKeys};
    use axum::body::{to_bytes, Body};
    use axum::http::{Request, StatusCode};
    use rusqlite::Connection;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::Mutex;
    use tower::ServiceExt;

    struct Mock {
        docs: Mutex<HashMap<String, Value>>,
        calls: AtomicUsize,
    }
    impl RemoteKeys for Mock {
        fn fetch_server_keys<'a>(&'a self, server: &'a str) -> FetchFuture<'a> {
            Box::pin(async move {
                self.calls.fetch_add(1, Ordering::SeqCst);
                self.docs.lock().unwrap().get(server).cloned().ok_or(fed::FedError::Network("unreachable".into()))
            })
        }
    }

    fn mem() -> Connection {
        let c = Connection::open_in_memory().unwrap();
        crate::store::create_matrix_schema(&c).unwrap();
        crate::keys::create_matrix_keys_schema(&c).unwrap();
        c
    }

    /// A "receiving" homeserver plus a remote sender `a.example` whose key document the mock serves.
    fn setup() -> (Arc<Homeserver>, Arc<Mock>, rusqlite::Connection, String, ed25519_dalek::SigningKey) {
        let c = mem();
        crate::store::ensure_matrix_user(&c, 1, "alice000000000000000000000000a1", "2026-10-09T00:00:00+00:00").unwrap();
        let hs = Arc::new(Homeserver::new(c));
        hs.federation_enabled.set(()).unwrap();
        let sender = mem();
        let (kid, key) = fed::active_signing_key(&sender, 0).unwrap();
        let doc = fed::server_keys_response(&sender, "a.example", fed::now_ms()).unwrap();
        let mock = Arc::new(Mock { docs: Mutex::new(HashMap::from([("a.example".to_string(), doc)])), calls: AtomicUsize::new(0) });
        let _ = hs.key_fetcher.set(mock.clone());
        (hs, mock, sender, kid, key)
    }

    async fn call(hs: &Arc<Homeserver>, req: Request<Body>) -> (StatusCode, Value) {
        let resp = crate::http::router(Arc::clone(hs)).oneshot(req).await.unwrap();
        let st = resp.status();
        let body = to_bytes(resp.into_body(), usize::MAX).await.unwrap();
        (st, serde_json::from_slice(&body).unwrap_or(Value::Null))
    }

    fn signed_get(uri: &str, origin: &str, dest: &str, kid: &str, key: &ed25519_dalek::SigningKey, signed_uri: &str) -> Request<Body> {
        let h = fed::build_x_matrix_header(origin, dest, kid, key, "GET", signed_uri, None);
        Request::get(uri).header(header::AUTHORIZATION, h).body(Body::empty()).unwrap()
    }

    #[tokio::test]
    async fn federation_is_off_until_enabled() {
        let hs = Arc::new(Homeserver::new(mem()));
        assert_eq!(call(&hs, Request::get("/key/v2/server").body(Body::empty()).unwrap()).await.0, StatusCode::NOT_FOUND);
        assert_eq!(call(&hs, Request::get("/federation/v1/version").body(Body::empty()).unwrap()).await.0, StatusCode::NOT_FOUND);
    }

    #[tokio::test]
    async fn own_key_document_is_served_and_valid() {
        let (hs, _, _, _, _) = setup();
        let (st, doc) = call(&hs, Request::get("/key/v2/server").body(Body::empty()).unwrap()).await;
        assert_eq!(st, StatusCode::OK);
        let me = crate::store::matrix_server_name();
        assert_eq!(doc["server_name"], me);
        assert!(fed::parse_server_keys(&doc, me, fed::now_ms()).is_ok());
        let (st2, doc2) = call(&hs, Request::get("/key/v2/server/ed25519:whatever").body(Body::empty()).unwrap()).await;
        assert_eq!((st2, doc2["verify_keys"].clone()), (StatusCode::OK, doc["verify_keys"].clone()), "stable key");
        assert_eq!(call(&hs, Request::get("/federation/v1/version").body(Body::empty()).unwrap()).await.0, StatusCode::OK);
    }

    #[tokio::test]
    async fn signed_request_is_verified_with_fetched_then_cached_keys() {
        let (hs, mock, _, kid, key) = setup();
        let me = crate::store::matrix_server_name();
        let uri = "/federation/v1/query/profile?user_id=%40alice000000000000000000000000a1%3Aexample.org";
        let signed = format!("/_matrix{uri}");
        let (st, _) = call(&hs, signed_get(uri, "a.example", me, &kid, &key, &signed)).await;
        assert_eq!(st, StatusCode::OK);
        let (st, _) = call(&hs, signed_get(uri, "a.example", me, &kid, &key, &signed)).await;
        assert_eq!(st, StatusCode::OK);
        assert_eq!(mock.calls.load(Ordering::SeqCst), 1, "second request served from the key cache");
        // The prefix is added once, whether or not an edge stripped it.
        assert_eq!(signed_uri(&"/federation/v1/x?a=1".parse().unwrap()), "/_matrix/federation/v1/x?a=1");
        assert_eq!(signed_uri(&"/_matrix/federation/v1/x?a=1".parse().unwrap()), "/_matrix/federation/v1/x?a=1");
    }

    #[tokio::test]
    async fn bad_requests_are_rejected() {
        let (hs, mock, _, kid, key) = setup();
        let me = crate::store::matrix_server_name();
        let uri = "/federation/v1/query/profile?user_id=%40alice000000000000000000000000a1%3Aexample.org";
        let signed = format!("/_matrix{uri}");
        // no header
        assert_eq!(call(&hs, Request::get(uri).body(Body::empty()).unwrap()).await.0, StatusCode::UNAUTHORIZED);
        // signature over another uri
        assert_eq!(call(&hs, signed_get(uri, "a.example", me, &kid, &key, "/_matrix/federation/v1/query/profile")).await.0, StatusCode::UNAUTHORIZED);
        // wrong destination
        assert_eq!(call(&hs, signed_get(uri, "a.example", "other.example", &kid, &key, &signed)).await.0, StatusCode::UNAUTHORIZED);
        // signed by a different key than the origin published
        let evil = ed25519_dalek::SigningKey::from_bytes(&[9u8; 32]);
        assert_eq!(call(&hs, signed_get(uri, "a.example", me, &kid, &evil, &signed)).await.0, StatusCode::UNAUTHORIZED);
        // unknown key id
        assert_eq!(call(&hs, signed_get(uri, "a.example", me, "ed25519:nope", &key, &signed)).await.0, StatusCode::UNAUTHORIZED);
        // origin that cannot be reached
        assert_eq!(call(&hs, signed_get(uri, "ghost.example", me, &kid, &key, &signed)).await.0, StatusCode::UNAUTHORIZED);
        // claiming to be ourselves
        assert_eq!(call(&hs, signed_get(uri, me, me, &kid, &key, &signed)).await.0, StatusCode::UNAUTHORIZED);
        assert!(mock.calls.load(Ordering::SeqCst) >= 1);
    }

    #[tokio::test]
    async fn allowlist_and_unknown_user() {
        let (hs, _, _, kid, key) = setup();
        let me = crate::store::matrix_server_name();
        let uri = "/federation/v1/query/profile?user_id=%40nobody%3Aexample.org";
        let signed = format!("/_matrix{uri}");
        assert_eq!(call(&hs, signed_get(uri, "a.example", me, &kid, &key, &signed)).await.0, StatusCode::NOT_FOUND, "authenticated, user unknown");
        let (hs2, _, _, kid2, key2) = setup();
        hs2.federation_allow.set(vec!["b.example".into()]).unwrap();
        assert_eq!(call(&hs2, signed_get(uri, "a.example", me, &kid2, &key2, &signed)).await.0, StatusCode::FORBIDDEN);
    }
}