mail4agent_server/http/
mod.rs

1//! Client-Server HTTP routes. Paths are relative; the process nests them
2//! under `/_matrix` if it wants that prefix.
3//!
4//! A bearer is `Authorization: Bearer <raw>` or the `access_token` query
5//! parameter. The stored credential is the SHA-256 hex of that raw token
6//! ([`hash_token`]). `POST /client/v3/register` returns the raw bearer once,
7//! in the response that creates the device. Nothing else returns it.
8
9use std::sync::Arc;
10
11use axum::extract::State;
12use axum::http::{header, HeaderMap};
13use axum::response::IntoResponse;
14use axum::routing::get;
15use axum::{Json, Router};
16use rusqlite::Connection;
17use sha2::{Digest, Sha256};
18
19use crate::error::MatrixError;
20#[cfg(test)]
21use crate::keys::CredentialKind;
22use crate::live::{ClaimRateLimiter, LiveRegistry};
23use crate::typing::TypingRegistry;
24
25mod account;
26pub(crate) mod extras;
27pub mod presence;
28mod media;
29pub(crate) mod spec_rest;
30mod fed_rest;
31mod compat;
32pub mod edge_auth;
33mod ephemeral;
34mod federation;
35pub mod fed_net;
36mod keys;
37mod messaging;
38mod rooms;
39mod push;
40mod sync;
41mod sliding;
42pub mod identity;
43
44pub struct Caller {
45    pub user_id: i64,
46    pub mxid: String,
47    pub device_id: String,
48    /// What the policy hook may know (`local` for device-bearer callers).
49    pub claims: crate::policy::Claims,
50}
51
52pub struct Homeserver {
53    /// The store: one writer, bounded parallel readers (tesserax-store). Reach it through
54    /// [`Homeserver::conn_scope`] (blocking threads) or [`with_conn_pub`] (async).
55    pub db: tesserax_store::Db,
56    /// Parallel read-only connections (file-backed stores only; `None` for in-memory stores).
57    pub readers: std::sync::OnceLock<tesserax_store::ReadPool>,
58    pub live: LiveRegistry,
59    pub typing: TypingRegistry,
60    pub claim_rate: ClaimRateLimiter,
61    pub push: crate::push::PushHub,
62    /// Public client base URL for `/.well-known/matrix/client` (config, never hard-coded).
63    pub public_base_url: std::sync::OnceLock<String>,
64    /// `m.server` value for `/.well-known/matrix/server`; unset = federation off.
65    pub federation_delegate: std::sync::OnceLock<String>,
66    /// Federation F0 switch: when set, the key and federation routes answer.
67    pub federation_enabled: std::sync::OnceLock<()>,
68    /// Origins allowed to call federation routes; unset = any origin that proves its keys.
69    pub federation_allow: std::sync::OnceLock<Vec<String>>,
70    /// Set (`M4A_ANON_READ=on`) to honour the product's marked, assertion-less read-only forwards.
71    pub anon_read: std::sync::OnceLock<()>,
72    /// How remote servers' signing keys are fetched; unset = no remote verification.
73    pub key_fetcher: std::sync::OnceLock<Arc<dyn crate::federation::RemoteKeys>>,
74    /// Outgoing federation transport; unset = no outgoing federation.
75    pub fed_transport: std::sync::OnceLock<Arc<dyn crate::federation::FedTransport>>,
76    /// Wakes the outbox worker when new federation work exists.
77    pub fed_notify: tokio::sync::Notify,
78    /// Accounts, doors and issuer assertions; unset = legacy device-bearer behaviour only.
79    pub seam: std::sync::OnceLock<Arc<identity::Seam>>,
80    /// Policy hook; unset = allow everything.
81    pub policy: std::sync::OnceLock<Arc<dyn crate::policy::PolicyHook>>,
82}
83
84impl Homeserver {
85    /// A store over an already-open in-memory connection (tests, the client's local bus).
86    /// File-backed stores come from [`Homeserver::from_db`].
87    pub fn new(conn: Connection) -> Self {
88        let db = tesserax_store::Db::open(&tesserax_store::DbConfig::in_memory()).expect("in-memory store");
89        // Nothing else holds the fresh writer yet, so this never contends (and never blocks a runtime thread).
90        db.blocking(|c| {
91            *c = conn;
92            Ok(())
93        })
94        .expect("install connection");
95        Self::from_db(db)
96    }
97
98    /// A store over an opened tesserax-store writer.
99    pub fn from_db(db: tesserax_store::Db) -> Self {
100        Self {
101            db,
102            readers: std::sync::OnceLock::new(),
103            live: LiveRegistry::new(),
104            typing: TypingRegistry::new(),
105            claim_rate: ClaimRateLimiter::new(),
106            push: crate::push::PushHub::new(),
107            public_base_url: std::sync::OnceLock::new(),
108            federation_delegate: std::sync::OnceLock::new(),
109            federation_enabled: std::sync::OnceLock::new(),
110            federation_allow: std::sync::OnceLock::new(),
111            anon_read: std::sync::OnceLock::new(),
112            key_fetcher: std::sync::OnceLock::new(),
113            fed_transport: std::sync::OnceLock::new(),
114            fed_notify: tokio::sync::Notify::new(),
115            seam: std::sync::OnceLock::new(),
116            policy: std::sync::OnceLock::new(),
117        }
118    }
119
120    /// Ask the policy hook (default: allow) about an action by `caller`.
121    pub fn check_policy(&self, caller: &Caller, action: crate::policy::Action, room_kind: Option<crate::store::RoomKind>) -> Result<(), MatrixError> {
122        match self.policy.get() {
123            Some(h) => crate::policy::enforce(h.as_ref(), &caller.claims, action, room_kind),
124            None => Ok(()),
125        }
126    }
127}
128
129pub fn hash_token(raw: &str) -> String {
130    let mut hasher = Sha256::new();
131    hasher.update(raw.as_bytes());
132    hex::encode(hasher.finalize())
133}
134
135pub fn raw_token(headers: &HeaderMap, query_token: Option<&str>) -> Option<String> {
136    if let Some(value) = headers.get(header::AUTHORIZATION).and_then(|v| v.to_str().ok()) {
137        if let Some(rest) = value.strip_prefix("Bearer ") {
138            let rest = rest.trim();
139            if !rest.is_empty() {
140                return Some(rest.to_string());
141            }
142        }
143    }
144    query_token.map(str::trim).filter(|s| !s.is_empty()).map(str::to_string)
145}
146
147impl Homeserver {
148    /// Runs `f` on the single writer connection, waiting for it. For blocking threads only
149    /// (`spawn_blocking`, plain threads); async code uses [`with_conn_pub`].
150    /// Async form of [`Homeserver::conn_scope`] (runs on the blocking pool).
151    pub async fn conn_async<T: Send + 'static>(&self, f: impl FnOnce(&mut Connection) -> T + Send + 'static) -> T {
152        self.db.write(move |c| Ok(f(c))).await.expect("store writer")
153    }
154
155    pub fn conn_scope<T>(&self, f: impl FnOnce(&mut Connection) -> T) -> T {
156        self.db.write_blocking(|c| Ok(f(c))).expect("store writer")
157    }
158}
159
160pub async fn resolve_caller(
161    state: &Arc<Homeserver>,
162    headers: &HeaderMap,
163    query_token: Option<&str>,
164) -> Result<Caller, MatrixError> {
165    if let Some((user_id, device_id, claims)) = identity::read_resolved(headers) {
166        if user_id == identity::ANON_USER {
167            // Anonymous read-only caller: no row, no device; only public data answers.
168            return Ok(Caller { user_id, mxid: String::new(), device_id: String::new(), claims });
169        }
170        let state = Arc::clone(state);
171        return tokio::task::spawn_blocking(move || -> Result<Caller, MatrixError> {
172            state.conn_scope(|conn: &mut rusqlite::Connection| {
173            let mxid = crate::store::mxid_of(&conn, user_id)?.ok_or_else(MatrixError::unknown_token)?;
174            Ok(Caller { user_id, mxid, device_id, claims })
175            })
176        })
177        .await
178        .map_err(|_| MatrixError::internal())?;
179    }
180    #[cfg(not(test))]
181    {
182        let _ = query_token;
183        return Err(MatrixError::missing_token());
184    }
185    #[cfg(test)]
186    resolve_bearer(state, headers, query_token).await
187}
188
189#[cfg(test)]
190async fn resolve_bearer(state: &Arc<Homeserver>, headers: &HeaderMap, query_token: Option<&str>) -> Result<Caller, MatrixError> {
191    let Some(raw) = raw_token(headers, query_token) else {
192        return Err(MatrixError::missing_token());
193    };
194    let hash = hash_token(&raw);
195    let state = Arc::clone(state);
196    tokio::task::spawn_blocking(move || -> Result<Caller, MatrixError> {
197        state.conn_scope(|conn: &mut rusqlite::Connection| {
198        let device = crate::keys::device_for_credential(&conn, CredentialKind::Bearer, &hash)?
199            .ok_or_else(MatrixError::unknown_token)?;
200        let mxid = crate::store::mxid_of(&conn, device.user_id)?.ok_or_else(MatrixError::unknown_token)?;
201        Ok(Caller { user_id: device.user_id, mxid, device_id: device.device_id, claims: crate::policy::Claims::new() })
202        })
203    })
204    .await
205    .map_err(|_| MatrixError::internal())?
206}
207
208/// Runs blocking DB work under the connection lock (shared by route modules).
209pub(crate) async fn with_conn_pub<T, F>(state: &Arc<Homeserver>, work: F) -> Result<T, MatrixError>
210where
211    T: Send + 'static,
212    F: FnOnce(&mut Connection) -> Result<T, MatrixError> + Send + 'static,
213{
214    let state = Arc::clone(state);
215    state.db.write(move |conn| Ok(work(conn))).await.map_err(|_| MatrixError::internal())?
216}
217
218/// Read-only work on a pooled reader connection (the store's parallel WAL readers); falls back to
219/// the writer when no read pool is attached (in-memory stores). `work` must not write.
220pub async fn with_read_pub<T, F>(state: &Arc<Homeserver>, work: F) -> Result<T, MatrixError>
221where
222    T: Send + 'static,
223    F: FnOnce(&Connection) -> Result<T, MatrixError> + Send + 'static,
224{
225    match state.readers.get() {
226        Some(pool) => pool.read(move |conn| Ok(work(conn))).await.map_err(|_| MatrixError::internal())?,
227        None => state.db.read(move |conn| Ok(work(conn))).await.map_err(|_| MatrixError::internal())?,
228    }
229}
230
231pub fn wake_users(state: &Homeserver, ids: impl IntoIterator<Item = i64>) {
232    state.live.wake_many(ids.into_iter().map(|id| format!("user:{id}")));
233    // Every write that wakes sync may also owe a federation delivery.
234    state.fed_notify.notify_one();
235}
236
237pub fn router(state: Arc<Homeserver>) -> Router {
238    Router::new()
239        .route("/client/versions", get(versions))
240        .route("/client/v3/capabilities", get(capabilities))
241        .merge(rooms::routes())
242        .merge(messaging::routes())
243        .merge(ephemeral::routes())
244        .merge(account::routes())
245        .merge(extras::routes())
246        .merge(presence::routes())
247        .merge(media::routes())
248        .merge(spec_rest::routes())
249        .merge(sliding::routes())
250                .merge(keys::routes())
251        .merge(sync::routes())
252        .merge(push::routes())
253        .merge(compat::routes())
254        .merge(federation::routes())
255        .merge(identity::routes())
256        .fallback(unrecognized)
257        .method_not_allowed_fallback(|| async { (axum::http::StatusCode::METHOD_NOT_ALLOWED, Json(serde_json::json!({ "errcode": "M_UNRECOGNIZED", "error": "Unrecognized request" }))) })
258        .layer(axum::middleware::from_fn_with_state(state.clone(), identity::assertion_layer))
259        .layer(axum::middleware::from_fn(cors))
260        .with_state(state)
261}
262
263/// Browser clients (Element, Cinny) call from another origin: answer preflights and allow it.
264pub async fn cors(req: axum::extract::Request, next: axum::middleware::Next) -> axum::response::Response {
265    use axum::http::{header, HeaderValue, Method, StatusCode};
266    let preflight = req.method() == Method::OPTIONS;
267    let mut resp = if preflight { axum::response::Response::builder().status(StatusCode::OK).body(axum::body::Body::empty()).unwrap() } else { next.run(req).await };
268    let h = resp.headers_mut();
269    h.insert(header::ACCESS_CONTROL_ALLOW_ORIGIN, HeaderValue::from_static("*"));
270    h.insert(header::ACCESS_CONTROL_ALLOW_METHODS, HeaderValue::from_static("GET, POST, PUT, DELETE, OPTIONS"));
271    h.insert(header::ACCESS_CONTROL_ALLOW_HEADERS, HeaderValue::from_static("X-Requested-With, Content-Type, Authorization, Date"));
272    resp
273}
274
275/// Every spec version up to the newest one supported (they are cumulative; clients test for the one a feature arrived in).
276const SPEC_VERSIONS: [&str; 19] = ["v1.1", "v1.2", "v1.3", "v1.4", "v1.5", "v1.6", "v1.7", "v1.8", "v1.9", "v1.10", "v1.11", "v1.12", "v1.13", "v1.14", "v1.15", "v1.16", "v1.17", "v1.18", "v1.19"];
277
278async fn versions() -> impl IntoResponse {
279    Json(serde_json::json!({
280        "versions": SPEC_VERSIONS,
281        "unstable_features": { "org.matrix.simplified_msc3575": true, "org.matrix.msc4186": true },
282    }))
283}
284
285async fn capabilities(
286    State(state): State<Arc<Homeserver>>,
287    headers: HeaderMap,
288) -> Result<impl IntoResponse, MatrixError> {
289    resolve_caller(&state, &headers, None).await?;
290    let room_version = crate::store::MATRIX_ROOM_VERSION;
291    Ok(Json(serde_json::json!({
292        "capabilities": {
293            "m.room_versions": {
294                "default": room_version,
295                "available": { room_version: "stable" },
296            },
297            "m.change_password": { "enabled": false },
298        }
299    })))
300}
301
302async fn unrecognized() -> MatrixError {
303    MatrixError::unrecognized()
304}
305
306#[cfg(test)]
307mod tests {
308    use super::*;
309    use axum::body::{to_bytes, Body};
310    use axum::http::{Request, StatusCode};
311    use tower::ServiceExt;
312
313    #[tokio::test]
314    async fn versions_reports_v1_19() {
315        let conn = Connection::open_in_memory().expect("memory");
316        let app = router(Arc::new(Homeserver::new(conn)));
317        let resp = app
318            .oneshot(Request::get("/client/versions").body(Body::empty()).expect("request"))
319            .await
320            .expect("response");
321        assert_eq!(resp.status(), StatusCode::OK);
322        let body = to_bytes(resp.into_body(), usize::MAX).await.expect("body");
323        let value: serde_json::Value = serde_json::from_slice(&body).expect("json");
324        assert_eq!(value["versions"].as_array().unwrap().last().unwrap(), "v1.19");
325        assert!(value["versions"].as_array().unwrap().iter().any(|v| v == "v1.1"));
326    }
327
328    #[tokio::test]
329    async fn unknown_path_is_unrecognized() {
330        let conn = Connection::open_in_memory().expect("memory");
331        let app = router(Arc::new(Homeserver::new(conn)));
332        let resp = app
333            .oneshot(Request::get("/client/v3/nope").body(Body::empty()).expect("request"))
334            .await
335            .expect("response");
336        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
337        let body = to_bytes(resp.into_body(), usize::MAX).await.expect("body");
338        let value: serde_json::Value = serde_json::from_slice(&body).expect("json");
339        assert_eq!(value["errcode"], "M_UNRECOGNIZED");
340    }
341}
342
mail4agent_server/http/mod.rs

mail4agent_server/http/
mod.rs
