mail4agent-server-bin 0.4.8

Loopback process for the mail4agent messenger homeserver. Listens on 127.0.0.1 only.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176

//! Two real server processes with different server names (`a.example`,
//! `b.example`) federate over loopback HTTP: public room join and posts,
//! cross-server DM invite, key query/claim, to-device, both directions.
//! The two servers find each other through the staging override map; no DNS.
//! Opaque blobs stand in for Olm ciphertext: clients' key exchange is
//! unchanged and runs on top of exactly these routes.

#[path = "support_product.rs"]
mod support_product;
#[path = "support/fed.rs"]
mod support_fed;

use std::time::Duration;

use mail4agent_server::federation::enc;
use reqwest::blocking::Client;
use serde_json::json;
use support_fed::{bodies, event_ids, free_port, is_hash_id, poll, send, Srv};

#[test]
fn two_servers_federate_public_room_dm_keys_and_to_device() {
    // Each server needs the other's port for the staging override map, so reserve both first.
    let (ra, rb) = (free_port(), free_port());
    let a = start_with_ports("a.example", "alice", ra, ("b.example", rb));
    let b = start_with_ports("b.example", "bob", rb, ("a.example", ra));
    let c = Client::builder().timeout(Duration::from_secs(10)).build().unwrap();

    // F0 over the wire.
    let (st, kd) = a.call(&c, "GET", "/_matrix/key/v2/server", None);
    assert_eq!((st, kd["server_name"].as_str()), (200, Some("a.example")));
    // Unsigned federation call is refused.
    let unsigned = c.put(format!("http://{}/_matrix/federation/v1/send/x", b.addr)).json(&json!({"origin":"a.example","pdus":[]})).send().unwrap();
    assert_eq!(unsigned.status().as_u16(), 401);

    // ---- public room: join, history, posts both ways
    let (st, v) = a.call(&c, "POST", "/client/v3/createRoom", Some(json!({"visibility":"public","name":"town"})));
    assert_eq!(st, 200, "{v}");
    let town = v["room_id"].as_str().unwrap().to_string();
    send(&a, &c, &town, "m.room.message", "e1", json!({"msgtype":"m.text","body":"early from a"}));
    let (st, v) = b.call(&c, "POST", &format!("/client/v3/rooms/{}/join", enc(&town)), Some(json!({})));
    assert_eq!(st, 200, "federated join: {v}");
    let (_, joined) = b.call(&c, "GET", "/client/v3/joined_rooms", None);
    assert!(joined["joined_rooms"].as_array().unwrap().iter().any(|r| r == &json!(town)), "bob joined on b");
    let hist = bodies(&b, &c, &town);
    assert!(hist.iter().any(|x| x == "early from a"), "history delivered with the join: {hist:?}");
    send(&a, &c, &town, "m.room.message", "e2", json!({"msgtype":"m.text","body":"hello from a"}));
    poll("b sees a's post", || bodies(&b, &c, &town).iter().any(|x| x == "hello from a").then_some(()));
    // Channels are read-only below level 50: alice promotes bob (a state event that must replicate to b).
    let pl_path = format!("/client/v3/rooms/{}/state/m.room.power_levels", enc(&town));
    let (_, mut pl) = a.call(&c, "GET", &pl_path, None);
    pl["users"][&b.user] = json!(50);
    assert_eq!(a.call(&c, "PUT", &pl_path, Some(pl)).0, 200);
    poll("b sees the new power levels", || (b.call(&c, "GET", &pl_path, None).1["users"][&b.user] == json!(50)).then_some(()));
    send(&b, &c, &town, "m.room.message", "e3", json!({"msgtype":"m.text","body":"hello from b"}));
    poll("a sees b's post", || bodies(&a, &c, &town).iter().any(|x| x == "hello from b").then_some(()));
    let (_, members) = a.call(&c, "GET", &format!("/client/v3/rooms/{}/joined_members", enc(&town)), None);
    assert!(members["joined"].get(&b.user).is_some(), "a lists bob as a member: {members}");

    // ---- DM across servers
    let (st, v) = a.call(&c, "POST", "/client/v3/createRoom", Some(json!({"is_direct":true,"invite":[b.user],"visibility":"private"})));
    assert_eq!(st, 200, "dm create: {v}");
    let dm = v["room_id"].as_str().unwrap().to_string();
    poll("bob has the invite", || {
        let (_, s) = b.call(&c, "GET", "/client/v3/sync?timeout=0", None);
        s["rooms"]["invite"].get(&dm).map(|_| ())
    });
    let (st, v) = b.call(&c, "POST", &format!("/client/v3/rooms/{}/join", enc(&dm)), Some(json!({})));
    assert_eq!(st, 200, "bob joins dm: {v}");
    poll("alice sees bob joined", || {
        let (_, m) = a.call(&c, "GET", &format!("/client/v3/rooms/{}/joined_members", enc(&dm)), None);
        m["joined"].get(&b.user).map(|_| ())
    });

    // ---- keys across servers (clients' key exchange is unchanged on top of this)
    let dev = |s: &Srv, curve: &str| {
        json!({"device_keys":{"user_id":s.user,"device_id":s.device,"algorithms":["m.olm.v1.curve25519-aes-sha2"],
            "keys":{format!("curve25519:{}",s.device):curve, format!("ed25519:{}",s.device):format!("ed-{curve}")},
            "signatures":{s.user.clone():{format!("ed25519:{}",s.device):"sig"}}},
          "one_time_keys":{"signed_curve25519:AAAAAA":{"key":format!("otk-{curve}"),"signatures":{}}}})
    };
    assert_eq!(a.call(&c, "POST", "/client/v3/keys/upload", Some(dev(&a, "curveA"))).0, 200);
    assert_eq!(b.call(&c, "POST", "/client/v3/keys/upload", Some(dev(&b, "curveB"))).0, 200);
    let (st, q) = b.call(&c, "POST", "/client/v3/keys/query", Some(json!({"device_keys":{a.user.clone():[]}})));
    assert_eq!(st, 200);
    assert_eq!(q["device_keys"][&a.user][&a.device]["keys"][format!("curve25519:{}", a.device)], "curveA", "bob queried alice's keys through federation: {q}");
    let (_, cl) = b.call(&c, "POST", "/client/v3/keys/claim", Some(json!({"one_time_keys":{a.user.clone():{a.device.clone():"signed_curve25519"}}})));
    assert_eq!(cl["one_time_keys"][&a.user][&a.device]["signed_curve25519:AAAAAA"]["key"], "otk-curveA", "claim across servers: {cl}");
    let (_, q2) = a.call(&c, "POST", "/client/v3/keys/query", Some(json!({"device_keys":{b.user.clone():[]}})));
    assert_eq!(q2["device_keys"][&b.user][&b.device]["keys"][format!("curve25519:{}", b.device)], "curveB", "reverse direction: {q2}");
    let (_, cl2) = a.call(&c, "POST", "/client/v3/keys/claim", Some(json!({"one_time_keys":{b.user.clone():{b.device.clone():"signed_curve25519"}}})));
    assert_eq!(cl2["one_time_keys"][&b.user][&b.device]["signed_curve25519:AAAAAA"]["key"], "otk-curveB");
    // a stranger who shares no room is not visible
    let (_, none) = b.call(&c, "POST", "/client/v3/keys/query", Some(json!({"device_keys":{"@ghost:a.example":[]}})));
    assert!(none["device_keys"].get("@ghost:a.example").is_none() || none["device_keys"]["@ghost:a.example"] == json!({}));

    // ---- E2E DM payloads (opaque ciphertext) both ways
    let ct = |tag: &str| json!({"algorithm":"m.olm.v1.curve25519-aes-sha2","sender_key":"k","ciphertext":format!("ct-{tag}")});
    send(&a, &c, &dm, "m.room.encrypted", "d1", ct("a-to-b"));
    poll("bob receives alice's ciphertext", || bodies(&b, &c, &dm).iter().any(|x| x == "ct-a-to-b").then_some(()));
    send(&b, &c, &dm, "m.room.encrypted", "d2", ct("b-to-a"));
    poll("alice receives bob's ciphertext", || bodies(&a, &c, &dm).iter().any(|x| x == "ct-b-to-a").then_some(()));

    // ---- to-device across servers
    let (st, _) = b.call(&c, "PUT", "/client/v3/sendToDevice/m.room.encrypted/t1", Some(json!({"messages":{a.user.clone():{a.device.clone():{"algorithm":"m.olm.v1.curve25519-aes-sha2","ciphertext":"td-b-to-a"}}}})));
    assert_eq!(st, 200);
    poll("alice gets the to-device message", || {
        let (_, s) = a.call(&c, "GET", "/client/v3/sync?timeout=0", None);
        s["to_device"]["events"].as_array().and_then(|e| e.iter().find(|x| x["content"]["ciphertext"] == "td-b-to-a").map(|_| ()))
    });
    let _ = (a.name, b.name);
}

/// Closed rooms are DAG rooms when the feature is on: every event of the room, on both servers,
/// carries the same reference-hash id; the public channel stays legacy; m4a-fed-1 carries both.
#[cfg(feature = "f3-hash-ids")]
#[test]
fn two_servers_federate_f3_closed_rooms_and_keep_public_rooms_legacy() {
    let (ra, rb) = (free_port(), free_port());
    let a = start_with_ports("a.example", "alice", ra, ("b.example", rb));
    let b = start_with_ports("b.example", "bob", rb, ("a.example", ra));
    let c = Client::builder().timeout(Duration::from_secs(10)).build().unwrap();
    let ct = |tag: &str| json!({"algorithm":"m.megolm.v1.aes-sha2","sender_key":"k","session_id":"s","device_id":"D","ciphertext":format!("ct-{tag}")});

    // A closed group: invite across servers, join, messages both ways.
    let (st, v) = a.call(&c, "POST", "/client/v3/createRoom", Some(json!({"visibility":"private","name":"closed","invite":[b.user]})));
    assert_eq!(st, 200, "create: {v}");
    let room = v["room_id"].as_str().unwrap().to_string();
    poll("bob has the invite", || b.call(&c, "GET", "/client/v3/sync?timeout=0", None).1["rooms"]["invite"].get(&room).map(|_| ()));
    let (st, v) = b.call(&c, "POST", &format!("/client/v3/rooms/{}/join", enc(&room)), Some(json!({})));
    assert_eq!(st, 200, "bob joins the f3 room: {v}");
    poll("alice sees bob joined", || a.call(&c, "GET", &format!("/client/v3/rooms/{}/joined_members", enc(&room)), None).1["joined"].get(&b.user).map(|_| ()));

    let m1 = send(&a, &c, &room, "m.room.encrypted", "m1", ct("a1"))["event_id"].as_str().unwrap().to_string();
    assert!(is_hash_id(&m1), "sender gets the computed id back: {m1}");
    poll("bob receives alice's message", || bodies(&b, &c, &room).iter().any(|x| x == "ct-a1").then_some(()));
    let m2 = send(&b, &c, &room, "m.room.encrypted", "m2", ct("b1"))["event_id"].as_str().unwrap().to_string();
    assert!(is_hash_id(&m2), "{m2}");
    poll("alice receives bob's message", || bodies(&a, &c, &room).iter().any(|x| x == "ct-b1").then_some(()));

    // Both servers hold the same events under the same ids.
    let (ea, eb) = (event_ids(&a, &c, &room), event_ids(&b, &c, &room));
    for id in [&m1, &m2] {
        assert!(ea.contains(id) && eb.contains(id), "{id} is on both servers: a={ea:?} b={eb:?}");
    }
    assert!(ea.iter().all(|i| is_hash_id(i)), "every event of the room on a is hash-id: {ea:?}");
    assert!(eb.iter().all(|i| is_hash_id(i)), "every event of the room on b is hash-id: {eb:?}");

    // State written on one side shows on the other: power levels (a), then a rename by bob (b), both checked by the DAG's auth rules.
    let pl_path = format!("/client/v3/rooms/{}/state/m.room.power_levels", enc(&room));
    let (_, mut pl) = a.call(&c, "GET", &pl_path, None);
    pl["users"][&b.user] = json!(50);
    assert_eq!(a.call(&c, "PUT", &pl_path, Some(pl)).0, 200);
    poll("b sees the new power levels", || (b.call(&c, "GET", &pl_path, None).1["users"][&b.user] == json!(50)).then_some(()));
    let name_path = format!("/client/v3/rooms/{}/state/m.room.name", enc(&room));
    let (st, _) = b.call(&c, "PUT", &name_path, Some(json!({"name":"renamed by bob"})));
    assert_eq!(st, 200);
    poll("a sees bob's rename", || (a.call(&c, "GET", &name_path, None).1["name"] == json!("renamed by bob")).then_some(()));

    // Bob leaves; alice sees it.
    assert_eq!(b.call(&c, "POST", &format!("/client/v3/rooms/{}/leave", enc(&room)), Some(json!({}))).0, 200);
    poll("alice sees bob left", || a.call(&c, "GET", &format!("/client/v3/rooms/{}/joined_members", enc(&room)), None).1["joined"].get(&b.user).is_none().then_some(()));

    // A public channel is not part of the layer: it still federates with legacy ids (the m4a-fed-1 test above covers its flow).
    let (_, v) = a.call(&c, "POST", "/client/v3/createRoom", Some(json!({"visibility":"public","name":"town"})));
    let town = v["room_id"].as_str().unwrap().to_string();
    let e = send(&a, &c, &town, "m.room.message", "p1", json!({"msgtype":"m.text","body":"public"}))["event_id"].as_str().unwrap().to_string();
    assert!(!is_hash_id(&e), "public channel keeps legacy ids: {e}");
    let (st, _) = b.call(&c, "POST", &format!("/client/v3/rooms/{}/join", enc(&town)), Some(json!({})));
    assert_eq!(st, 200);
    poll("bob sees the public post", || bodies(&b, &c, &town).iter().any(|x| x == "public").then_some(()));
}


fn start_with_ports(name: &'static str, localpart: &str, port: u16, peer: (&str, u16)) -> Srv {
    support_fed::start_node(name, localpart, port, &[peer])
}