m4a_matrix_core/
dag.rs

1//! The event DAG of one room, over any store that can answer [`DagRead`].
2//!
3//! Two entry points share one acceptance path:
4//! * [`local_event`] builds, signs and accepts an event this server creates;
5//! * [`accept_wire`] verifies and accepts an event received from a peer.
6//!
7//! Acceptance follows the spec order: state-independent auth rules, auth rules against the state
8//! named by the event's own `auth_events`, auth rules against the room state before the event
9//! (resolved from its `prev_events`), and finally against the current state, where a failure only
10//! soft-fails the event (stored, never made a forward extremity). Forked branches are merged with
11//! state resolution v2.
12
13use std::collections::BTreeSet;
14
15use ruma_common::{room_version_rules::RoomVersionRules, CanonicalJsonObject, EventId, OwnedEventId, OwnedRoomId, UserId};
16use ruma_events::{StateEventType, TimelineEventType};
17use ruma_signatures::{hash_and_sign_event, reference_hash, verify_event, PublicKeyMap, Verified};
18use ruma_state_res::{auth_types_for_event, check_state_dependent_auth_rules, check_state_independent_auth_rules, StateMap};
19use serde_json::{json, value::RawValue, Value};
20
21use crate::{resolve::resolve_state_with, Pdu, SignedEvent, Signer};
22
23/// What the DAG needs from storage.
24pub trait DagRead {
25    /// A stored event (accepted, soft-failed or imported outlier).
26    fn pdu(&self, id: &EventId) -> Option<Pdu>;
27    /// The room state after the event, for events that have one.
28    fn state_after(&self, id: &EventId) -> Option<StateMap<OwnedEventId>>;
29    /// The room's forward extremities.
30    fn extremities(&self) -> Vec<OwnedEventId>;
31}
32
33#[derive(Debug, thiserror::Error)]
34pub enum Reject {
35    #[error("malformed event: {0}")]
36    Malformed(String),
37    #[error("event already stored")]
38    Duplicate,
39    #[error("content hash does not match (content was altered or already redacted)")]
40    BadContentHash,
41    #[error("signature check failed: {0}")]
42    BadSignature(String),
43    #[error("unknown prev_event {0}")]
44    MissingPrev(String),
45    #[error("unknown auth_event {0}")]
46    MissingAuth(String),
47    #[error("auth rules refuse the event: {0}")]
48    Auth(String),
49    #[error("state resolution failed: {0}")]
50    Resolve(String),
51}
52
53/// An event that passed the checks, with everything a store must write for it.
54#[derive(Clone, Debug)]
55pub struct Accepted {
56    pub event_id: OwnedEventId,
57    pub pdu: Pdu,
58    pub json: CanonicalJsonObject,
59    /// Room state after this event (its `state_before` plus itself when it is a state event).
60    pub state_after: StateMap<OwnedEventId>,
61    /// Failed only the check against the current state: store it, do not extend the room with it.
62    pub soft_failed: bool,
63    /// Forward extremities after acceptance.
64    pub extremities: Vec<OwnedEventId>,
65    /// Resolved current room state after acceptance.
66    pub current_state: StateMap<OwnedEventId>,
67    /// The new extremity set has more than one member: branches are open and state was resolved.
68    pub forked: bool,
69}
70
71fn skey(kind: &TimelineEventType) -> StateEventType {
72    StateEventType::from(kind.to_string())
73}
74
75/// Resolved room state over a set of event ids' states.
76fn resolve_over(rules: &RoomVersionRules, db: &dyn DagRead, ids: &[OwnedEventId], extra: Option<(&OwnedEventId, &StateMap<OwnedEventId>)>) -> Result<StateMap<OwnedEventId>, Reject> {
77    let mut forks: Vec<StateMap<OwnedEventId>> = Vec::new();
78    for id in ids {
79        match extra {
80            Some((eid, st)) if eid == id => forks.push(st.clone()),
81            _ => forks.push(db.state_after(id).ok_or_else(|| Reject::MissingPrev(id.to_string()))?),
82        }
83    }
84    match forks.len() {
85        0 => Ok(StateMap::new()),
86        1 => Ok(forks.remove(0)),
87        _ => resolve_state_with(rules, &|id: &EventId| db.pdu(id), &forks).map_err(|e| Reject::Resolve(e.to_string())),
88    }
89}
90
91/// The store plus the event being accepted, which is not written yet.
92struct WithNew<'a> {
93    db: &'a dyn DagRead,
94    pdu: &'a Pdu,
95}
96
97impl DagRead for WithNew<'_> {
98    fn pdu(&self, id: &EventId) -> Option<Pdu> {
99        if id == self.pdu.event_id { Some(self.pdu.clone()) } else { self.db.pdu(id) }
100    }
101    fn state_after(&self, id: &EventId) -> Option<StateMap<OwnedEventId>> {
102        self.db.state_after(id)
103    }
104    fn extremities(&self) -> Vec<OwnedEventId> {
105        self.db.extremities()
106    }
107}
108
109/// The room's current state: the forward extremities' states, resolved.
110pub fn current_state(rules: &RoomVersionRules, db: &dyn DagRead) -> Result<StateMap<OwnedEventId>, Reject> {
111    resolve_over(rules, db, &db.extremities(), None)
112}
113
114/// Event id of a signed wire event: `$` plus the reference hash.
115pub fn compute_event_id(rules: &RoomVersionRules, json: &CanonicalJsonObject) -> Result<OwnedEventId, Reject> {
116    let h = reference_hash(json, rules).map_err(|e| Reject::Malformed(e.to_string()))?;
117    OwnedEventId::try_from(format!("${h}")).map_err(|e| Reject::Malformed(e.to_string()))
118}
119
120/// Checks the content hash and the origin signature (needs the signing server's public key).
121/// A skeleton (an event whose content was erased and which is already in its redacted form) is
122/// accepted: its signature verifies, its content hash cannot match any more.
123pub fn verify_wire(rules: &RoomVersionRules, json: &CanonicalJsonObject, keys: &PublicKeyMap) -> Result<(), Reject> {
124    match verify_event(keys, json, rules) {
125        Ok(Verified::All) => Ok(()),
126        Ok(Verified::Signatures) if is_skeleton(rules, json) => Ok(()),
127        Ok(Verified::Signatures) => Err(Reject::BadContentHash),
128        Err(e) => Err(Reject::BadSignature(e.to_string())),
129    }
130}
131
132/// The event is already in its redacted form (content erased down to what redaction keeps).
133pub fn is_skeleton(rules: &RoomVersionRules, json: &CanonicalJsonObject) -> bool {
134    skeleton(rules, json).map(|s| s == *json).unwrap_or(false)
135}
136
137/// The skeleton of an event: redacted form (content erased, ids, edges, hashes and signatures kept).
138/// Its reference hash, so its event id, is unchanged.
139pub fn skeleton(rules: &RoomVersionRules, json: &CanonicalJsonObject) -> Result<CanonicalJsonObject, Reject> {
140    ruma_common::canonical_json::redact(json.clone(), &rules.redaction, None).map_err(|e| Reject::Malformed(e.to_string()))
141}
142
143/// Accepts one event (already hashed, signed and id-checked by the caller).
144/// `state_before` overrides the state derived from `prev_events`; a joining server uses it to
145/// start a replica from a state snapshot whose history it does not hold.
146pub fn accept(rules: &RoomVersionRules, db: &dyn DagRead, event_id: OwnedEventId, json: CanonicalJsonObject, state_before: Option<StateMap<OwnedEventId>>) -> Result<Accepted, Reject> {
147    if db.pdu(&event_id).is_some() {
148        return Err(Reject::Duplicate);
149    }
150    let pdu = Pdu::from_wire(event_id.clone(), &json).map_err(Reject::Malformed)?;
151    let fetch = |id: &EventId| -> Option<Pdu> { if id == pdu.event_id { Some(pdu.clone()) } else { db.pdu(id) } };
152
153    // 1. State-independent rules (shape, auth_events present and of allowed types).
154    for a in &pdu.auth_events {
155        if db.pdu(a).is_none() {
156            return Err(Reject::MissingAuth(a.to_string()));
157        }
158    }
159    check_state_independent_auth_rules(&rules.authorization, &pdu, &fetch).map_err(Reject::Auth)?;
160
161    // 2. Against the state named by the event's own auth_events.
162    let mut auth_state: std::collections::HashMap<(StateEventType, String), Pdu> = Default::default();
163    for a in &pdu.auth_events {
164        if let Some(p) = db.pdu(a) {
165            if let Some(sk) = &p.state_key {
166                auth_state.insert((skey(&p.kind), sk.clone()), p);
167            }
168        }
169    }
170    check_state_dependent_auth_rules(&rules.authorization, &pdu, |t: &StateEventType, k: &str| auth_state.get(&(t.clone(), k.to_owned())).cloned()).map_err(Reject::Auth)?;
171
172    // 3. Against the room state before the event.
173    let before = match state_before {
174        Some(s) => s,
175        None => resolve_over(rules, db, &pdu.prev_events, None)?,
176    };
177    let by_state = |t: &StateEventType, k: &str| before.get(&(t.clone(), k.to_owned())).and_then(|id| db.pdu(id));
178    check_state_dependent_auth_rules(&rules.authorization, &pdu, by_state).map_err(Reject::Auth)?;
179
180    let mut state_after = before;
181    if let Some(sk) = &pdu.state_key {
182        state_after.insert((skey(&pdu.kind), sk.clone()), event_id.clone());
183    }
184
185    // 4. Against the current state: failing here only soft-fails.
186    let old_ext = db.extremities();
187    let current = resolve_over(rules, db, &old_ext, None)?;
188    let by_current = |t: &StateEventType, k: &str| current.get(&(t.clone(), k.to_owned())).and_then(|id| db.pdu(id));
189    let soft_failed = !old_ext.is_empty() && check_state_dependent_auth_rules(&rules.authorization, &pdu, by_current).is_err();
190
191    if soft_failed {
192        return Ok(Accepted { event_id, pdu, json, state_after, soft_failed: true, extremities: old_ext, current_state: current, forked: false });
193    }
194    let mut ext: Vec<OwnedEventId> = old_ext.into_iter().filter(|e| !pdu.prev_events.contains(e)).collect();
195    ext.push(event_id.clone());
196    ext.sort();
197    let forked = ext.len() > 1;
198    let current_state = resolve_over(rules, &WithNew { db, pdu: &pdu }, &ext, Some((&event_id, &state_after)))?;
199    Ok(Accepted { event_id, pdu, json, state_after, soft_failed: false, extremities: ext, current_state, forked })
200}
201
202/// An event of the past (fetched by get_missing_events or backfill, or part of an auth chain):
203/// checked against its own auth events and the state before it, never against the current state,
204/// and it does not touch the forward extremities. Its `prev_events` must already have a state.
205pub struct Historic {
206    pub event_id: OwnedEventId,
207    pub pdu: Pdu,
208    pub json: CanonicalJsonObject,
209    pub state_after: StateMap<OwnedEventId>,
210}
211
212pub fn accept_historic(rules: &RoomVersionRules, db: &dyn DagRead, event_id: OwnedEventId, json: CanonicalJsonObject) -> Result<Historic, Reject> {
213    let pdu = Pdu::from_wire(event_id.clone(), &json).map_err(Reject::Malformed)?;
214    let fetch = |id: &EventId| -> Option<Pdu> { if id == pdu.event_id { Some(pdu.clone()) } else { db.pdu(id) } };
215    for a in &pdu.auth_events {
216        if db.pdu(a).is_none() {
217            return Err(Reject::MissingAuth(a.to_string()));
218        }
219    }
220    check_state_independent_auth_rules(&rules.authorization, &pdu, &fetch).map_err(Reject::Auth)?;
221    let mut auth_state: std::collections::HashMap<(StateEventType, String), Pdu> = Default::default();
222    for a in &pdu.auth_events {
223        if let Some(p) = db.pdu(a) {
224            if let Some(sk) = &p.state_key {
225                auth_state.insert((skey(&p.kind), sk.clone()), p);
226            }
227        }
228    }
229    check_state_dependent_auth_rules(&rules.authorization, &pdu, |t: &StateEventType, k: &str| auth_state.get(&(t.clone(), k.to_owned())).cloned()).map_err(Reject::Auth)?;
230    let before = resolve_over(rules, db, &pdu.prev_events, None)?;
231    let by_state = |t: &StateEventType, k: &str| before.get(&(t.clone(), k.to_owned())).and_then(|id| db.pdu(id));
232    check_state_dependent_auth_rules(&rules.authorization, &pdu, by_state).map_err(Reject::Auth)?;
233    let mut state_after = before;
234    if let Some(sk) = &pdu.state_key {
235        state_after.insert((skey(&pdu.kind), sk.clone()), event_id.clone());
236    }
237    Ok(Historic { event_id, pdu, json, state_after })
238}
239
240/// Re-verifies an auth chain received from a peer: every event's signature and content hash (or
241/// skeleton form), its id, that every `auth_events` entry is in the set, and the auth rules of each
242/// event against the state its own auth events describe. Returns the ids in verified order.
243pub fn verify_auth_chain(rules: &RoomVersionRules, events: &[CanonicalJsonObject], keys: &PublicKeyMap) -> Result<Vec<OwnedEventId>, Reject> {
244    let mut pdus: std::collections::HashMap<OwnedEventId, Pdu> = Default::default();
245    for j in events {
246        verify_wire(rules, j, keys)?;
247        let id = compute_event_id(rules, j)?;
248        let p = Pdu::from_wire(id.clone(), j).map_err(Reject::Malformed)?;
249        pdus.insert(id, p);
250    }
251    let mut order: Vec<&Pdu> = pdus.values().collect();
252    order.sort_by_key(|p| (p.depth, p.event_id.clone()));
253    for p in &order {
254        for a in &p.auth_events {
255            if !pdus.contains_key(a) {
256                return Err(Reject::MissingAuth(a.to_string()));
257            }
258        }
259        let fetch = |id: &EventId| pdus.get(id).cloned();
260        check_state_independent_auth_rules(&rules.authorization, *p, &fetch).map_err(Reject::Auth)?;
261        let mut st: std::collections::HashMap<(StateEventType, String), Pdu> = Default::default();
262        for a in &p.auth_events {
263            let ap = &pdus[a];
264            if let Some(sk) = &ap.state_key {
265                st.insert((skey(&ap.kind), sk.clone()), ap.clone());
266            }
267        }
268        check_state_dependent_auth_rules(&rules.authorization, *p, |t: &StateEventType, k: &str| st.get(&(t.clone(), k.to_owned())).cloned()).map_err(Reject::Auth)?;
269    }
270    Ok(order.iter().map(|p| p.event_id.clone()).collect())
271}
272
273/// A peer's event: check the origin signature and content hash, derive the id, accept.
274pub fn accept_wire(rules: &RoomVersionRules, db: &dyn DagRead, json: CanonicalJsonObject, keys: &PublicKeyMap, state_before: Option<StateMap<OwnedEventId>>) -> Result<Accepted, Reject> {
275    verify_wire(rules, &json, keys)?;
276    let id = compute_event_id(rules, &json)?;
277    accept(rules, db, id, json, state_before)
278}
279
280/// An unsigned event for `sender`, ready to be filled in by the sender's server (make_join style):
281/// `prev_events`, `auth_events` and `depth` come from this room's DAG.
282pub fn template(rules: &RoomVersionRules, db: &dyn DagRead, room_id: &OwnedRoomId, ts: u64, sender: &str, kind: &str, state_key: Option<&str>, content: Value) -> Result<CanonicalJsonObject, Reject> {
283    let sender_id = UserId::parse(sender).map_err(|e| Reject::Malformed(e.to_string()))?;
284    let raw = RawValue::from_string(content.to_string()).map_err(|e| Reject::Malformed(e.to_string()))?;
285    let tl = TimelineEventType::from(kind.to_owned());
286    let ext = db.extremities();
287    let cur = resolve_over(rules, db, &ext, None)?;
288    let auth: Vec<OwnedEventId> = auth_types_for_event(&tl, &sender_id, state_key, &raw, &rules.authorization)
289        .map_err(Reject::Auth)?
290        .iter()
291        .filter_map(|k| cur.get(k).cloned())
292        .collect();
293    let depth = ext.iter().filter_map(|e| db.pdu(e)).map(|p| p.depth).max().unwrap_or(0) + 1;
294    let mut o = json!({
295        "room_id": room_id, "sender": sender, "type": kind, "content": content,
296        "origin_server_ts": ts, "prev_events": ext, "auth_events": auth, "depth": depth,
297    });
298    if let Some(sk) = state_key {
299        o["state_key"] = json!(sk);
300    }
301    serde_json::from_value(o).map_err(|e| Reject::Malformed(e.to_string()))
302}
303
304/// Hashes and signs a filled-in template; the event id is the reference hash.
305pub fn sign_template(rules: &RoomVersionRules, signer: &Signer, mut obj: CanonicalJsonObject) -> Result<SignedEvent, Reject> {
306    hash_and_sign_event(signer.server(), signer, &mut obj, &rules.redaction).map_err(|e| Reject::Malformed(e.to_string()))?;
307    let event_id = compute_event_id(rules, &obj)?;
308    Ok(SignedEvent { event_id, json: obj })
309}
310
311/// An event this server creates: template, sign, accept (same checks as a peer's event).
312pub fn local_event(rules: &RoomVersionRules, db: &dyn DagRead, signer: &Signer, room_id: &OwnedRoomId, ts: u64, sender: &str, kind: &str, state_key: Option<&str>, content: Value) -> Result<Accepted, Reject> {
313    let t = template(rules, db, room_id, ts, sender, kind, state_key, content)?;
314    let s = sign_template(rules, signer, t)?;
315    accept(rules, db, s.event_id, s.json, None)
316}
317
318/// Ids of the events in `ids` that `db` does not hold.
319pub fn missing(db: &dyn DagRead, ids: &[OwnedEventId]) -> BTreeSet<OwnedEventId> {
320    ids.iter().filter(|i| db.pdu(i).is_none()).cloned().collect()
321}
m4a_matrix_core/dag.rs

m4a_matrix_core/
dag.rs
