m4a-product-example 0.4.11

//! Product server -> edge -> core, all real HTTP on loopback, in one process.

use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use m4a_product_example::{router, ProductApp, SqliteStore};
use m4a_product_kit::door::{DoorFuture, LoginDoor};
use m4a_product_kit::nick_rules::NickRules;
use m4a_product_kit::tiers::TierTable;
use m4a_product_kit::{EdgeLink, EventPublisher, UserService};
use mail4agent_server::http::{edge_auth::require_edge_secret, identity::Seam, Homeserver};
use tesserax_store::rusqlite::Connection;
use serde_json::{json, Value};

const EDGE_SECRET: &str = "0123456789abcdef0123456789abcdef-edge";
const SEAM_SECRET: &[u8] = b"seam-secret-0123456789abcdef";
const LINK: &str = "link-token-0123456789abcdef";
const ADMIN: &str = "admin-token-for-tests";

async fn serve(app: axum::Router) -> SocketAddr {
    let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let a = l.local_addr().unwrap();
    tokio::spawn(async move { axum::serve(l, app.into_make_service_with_connect_info::<SocketAddr>()).await.unwrap() });
    a
}

fn serve_unix(app: axum::Router, path: &std::path::Path) {
    let _ = std::fs::remove_file(path);
    let l = tokio::net::UnixListener::bind(path).unwrap();
    tokio::spawn(async move { axum::serve(l, app).await.unwrap() });
}

struct FakeDoor;
impl LoginDoor for FakeDoor {
    fn id(&self) -> &str {
        "fake"
    }
    fn verify<'a>(&'a self, proof: &'a Value) -> DoorFuture<'a> {
        Box::pin(async move { Ok(("fake".to_string(), proof["who"].as_str().unwrap_or("x").to_string())) })
    }
}

struct Stack {
    link: EdgeLink,
    edge: String,
    core: Arc<Homeserver>,
    product: String,
    c: reqwest::Client,
}

async fn stack() -> Stack {
    stack_with(None).await
}

/// With `unix_dir`, product -> edge and edge -> core run over unix sockets in that directory.
async fn stack_with(unix_dir: Option<&std::path::Path>) -> Stack {
    let conn = Connection::open_in_memory().unwrap();
    mail4agent_server::store::create_matrix_schema(&conn).unwrap();
    mail4agent_server::keys::create_matrix_keys_schema(&conn).unwrap();
    let core = Arc::new(Homeserver::new(conn));
    let _ = core.anon_read.set(());
    let _ = core.seam.set(Arc::new(Seam::new(vec![SEAM_SECRET.to_vec()], 30, None, None)));
    let core_app = require_edge_secret(mail4agent_server::http::router(core.clone()), EDGE_SECRET.to_string());
    let (core_url, edge_url) = match unix_dir {
        Some(d) => {
            serve_unix(core_app, &d.join("core.sock"));
            (format!("unix:{}", d.join("core.sock").display()), format!("unix:{}", d.join("edge.sock").display()))
        }
        None => (format!("http://{}", serve(core_app).await), String::new()),
    };
    let edge_app = m4a_edge::edge_router(m4a_edge::EdgeConfig { core_url, secret: EDGE_SECRET.into(), link_token: Some(LINK.into()) });
    let edge_url = match unix_dir {
        Some(d) => {
            serve_unix(edge_app, &d.join("edge.sock"));
            edge_url
        }
        None => format!("http://{}", serve(edge_app).await),
    };
    let link = EdgeLink::new(&edge_url, SEAM_SECRET.to_vec(), None).with_link_token(Some(LINK.into()));
    let app = Arc::new(ProductApp {
        users: UserService::new(Arc::new(SqliteStore::memory().unwrap()), NickRules::default(), TierTable::default()),
        events: EventPublisher::spawn_with(link.clone(), None, Duration::from_millis(50), Arc::new(m4a_product_kit::SqliteOutbox::memory().unwrap())),
        link: link.clone(),
        admin_token: ADMIN.into(),
        doors: vec![Arc::new(FakeDoor)],
        anon_read: true,
        tokens: Default::default(),
        access_ttl_ms: 3_600_000,
        challenges: m4a_product_kit::ChallengeBook::new("test-product"),
    });
    let product = format!("http://{}", serve(router(app)).await);
    Stack { link, edge: edge_url, core, product, c: reqwest::Client::new() }
}

impl Stack {
    async fn post(&self, path: &str, token: Option<&str>, body: Value) -> (u16, Value) {
        self.send("POST", path, token, Some(body)).await
    }
    async fn send(&self, method: &str, path: &str, token: Option<&str>, body: Option<Value>) -> (u16, Value) {
        let mut rb = self.c.request(method.parse().unwrap(), format!("{}{}", self.product, path));
        if let Some(t) = token {
            rb = rb.bearer_auth(t);
        }
        if let Some(b) = body {
            rb = rb.json(&b);
        }
        let r = rb.send().await.unwrap();
        (r.status().as_u16(), r.json().await.unwrap_or(Value::Null))
    }
    async fn register(&self) -> (String, String) {
        let (st, b) = self.post("/product/v1/register", None, json!({"password":"correct horse"})).await;
        assert_eq!(st, 200, "{b}");
        (b["nick"].as_str().unwrap().to_string(), b["token"].as_str().unwrap().to_string())
    }
    /// Poll the core until `f` holds (events are delivered asynchronously).
    async fn until(&self, what: &str, f: impl Fn(&Connection) -> bool + Send + Sync + 'static) {
        let f = Arc::new(f);
        for _ in 0..100 {
            let g = Arc::clone(&f);
            if self.core.conn_async(move |c| g(c)).await {
                return;
            }
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
        panic!("timed out waiting for: {what}");
    }
}

#[tokio::test]
async fn register_contact_room_message_rename_logout_and_delete_through_the_whole_stack() {
    let s = stack().await;
    let (a_nick, a_tok) = s.register().await;
    let (b_nick, b_tok) = s.register().await;
    assert_eq!((a_nick.len(), b_nick.len()), (8, 8), "placeholder nicks");

    // Registration alone does not touch the messenger.
    assert_eq!(s.core.conn_async(|c| c.query_row::<i64, _, _>("SELECT COUNT(*) FROM identities", [], |r| r.get(0)).unwrap()).await, 0);

    // Nick before first contact: free, and the messenger's localpart becomes the short nick.
    let (st, b) = s.send("PUT", "/product/v1/nick", Some(&a_tok), Some(json!({"nick":"alice"}))).await;
    assert_eq!((st, b["nick"].as_str()), (200, Some("alice")));
    let (st, b) = s.send("PUT", "/product/v1/nick", Some(&a_tok), Some(json!({"nick":"alice2"}))).await;
    assert_eq!((st, b["errcode"].as_str()), (429, Some("M4A_NICK_COOLDOWN")));
    assert!(b["retry_after_ms"].as_i64().unwrap() > 0);
    let (st, b) = s.send("PUT", "/product/v1/nick", Some(&b_tok), Some(json!({"nick":"Alice"}))).await;
    assert_eq!((st, b["errcode"].as_str()), (409, Some("M4A_NICK_TAKEN")));
    let (st, _) = s.send("PUT", "/product/v1/nick", Some(&b_tok), Some(json!({"nick":"a:b"}))).await;
    assert_eq!(st, 400);
    let (st, _) = s.send("PUT", "/product/v1/nick", Some(&b_tok), Some(json!({"nick":"bobby"}))).await;
    assert_eq!(st, 200);

    // First messenger requests are first contact.
    for t in [&a_tok, &b_tok] {
        let (st, b) = s.send("GET", "/_matrix/client/v3/capabilities", Some(t), None).await;
        assert_eq!(st, 200, "{b}");
    }
    let ids: Vec<String> = s.core.conn_async(|c| c.prepare("SELECT localpart FROM identities ORDER BY id").unwrap().query_map([], |r| r.get(0)).unwrap().map(Result::unwrap).collect()).await;
    assert_eq!(ids, vec!["alice", "bobby"]);

    // Room, invite, join, message, read: a normal Matrix flow through the proxy.
    let (st, b) = s.post("/_matrix/client/v3/createRoom", Some(&a_tok), json!({"name":"hello","preset":"private_chat"})).await;
    assert_eq!(st, 200, "{b}");
    let room = b["room_id"].as_str().unwrap().to_string();
    let (st, b) = s.post(&format!("/_matrix/client/v3/rooms/{room}/invite"), Some(&a_tok), json!({"user_id":"@bobby:example.org"})).await;
    assert_eq!(st, 200, "{b}");
    let (st, b) = s.post(&format!("/_matrix/client/v3/rooms/{room}/join"), Some(&b_tok), json!({})).await;
    assert_eq!(st, 200, "{b}");
    let (st, b) = s.send("PUT", &format!("/_matrix/client/v3/rooms/{room}/send/m.room.encrypted/t1"), Some(&a_tok), Some(json!({"algorithm":"m.megolm.v1.aes-sha2","ciphertext":"hi bob (opaque)","sender_key":"k","session_id":"s","device_id":"d"}))).await;
    assert_eq!(st, 200, "{b}");
    let (st, b) = s.send("GET", &format!("/_matrix/client/v3/rooms/{room}/messages?dir=b&limit=10"), Some(&b_tok), None).await;
    assert_eq!(st, 200, "{b}");
    assert!(b.to_string().contains("hi bob"), "{b}");

    // Without a token or with a forged assertion header nothing is asserted.
    let (st, _) = s.send("GET", "/_matrix/client/v3/capabilities", None, None).await;
    assert_eq!(st, 401);
    let r = s.c.get(format!("{}/_matrix/client/v3/capabilities", s.product)).header("x-m4a-assertion", "v1.e30.00").header("x-m4a-resolved", "1|X|1").send().await.unwrap();
    assert_eq!(r.status().as_u16(), 401);
    let (st, _) = s.send("GET", "/_matrix/client/v3/capabilities", Some("not-a-token"), None).await;
    assert_eq!(st, 401, "unknown token is refused by the product");

    // Rename after contact: the product publishes nick.changed; the localpart stays.
    let (c_nick, c_tok) = s.register().await;
    assert_eq!(s.send("GET", "/_matrix/client/v3/capabilities", Some(&c_tok), None).await.0, 200);
    let (st, b) = s.send("PUT", "/product/v1/nick", Some(&c_tok), Some(json!({"nick":"carla"}))).await;
    assert_eq!((st, b["nick"].as_str()), (200, Some("carla")));
    s.until("identity renamed", {let c_nick = c_nick.clone(); move |c: &Connection| c.query_row("SELECT nick FROM identities WHERE localpart = ?1", [&c_nick], |r| r.get::<_, String>(0)).map(|n| n == "carla").unwrap_or(false)}).await;
    let (st, b) = s.send("GET", &format!("/_matrix/client/v3/profile/@{c_nick}:example.org/displayname"), Some(&c_tok), None).await;
    assert_eq!((st, b["displayname"].as_str()), (200, Some("carla")));
    // Same credential keeps working under the new nick, same identity.
    assert_eq!(s.send("GET", "/_matrix/client/v3/capabilities", Some(&c_tok), None).await.0, 200);
    assert_eq!(s.core.conn_async(|c| c.query_row::<i64, _, _>("SELECT COUNT(*) FROM identities", [], |r| r.get(0)).unwrap()).await, 3);

    // Logout: product refuses the token at once; the core drops the device.
    let devs = |c: &Connection, n: &str| c.query_row::<i64, _, _>("SELECT COUNT(*) FROM devices d JOIN identities i ON i.id = d.user_id WHERE i.nick = ?1", [n], |r| r.get(0)).unwrap();
    assert_eq!(s.core.conn_async(move |c| devs(c, "bobby")).await, 1);
    assert_eq!(s.post("/product/v1/logout", Some(&b_tok), json!({})).await.0, 200);
    assert_eq!(s.send("GET", "/_matrix/client/v3/capabilities", Some(&b_tok), None).await.0, 401);
    s.until("device removed", move |c| devs(c, "bobby") == 0).await;

    // Admin API is closed without the admin token; delete retires the identity in the core.
    assert_eq!(s.post("/product/v1/admin/delete", Some("nope"), json!({"nick":"alice"})).await.0, 403);
    assert_eq!(s.post("/product/v1/admin/delete", Some(ADMIN), json!({"nick":"alice"})).await.0, 200);
    s.until("identity retired", |c| c.query_row::<i64, _, _>("SELECT COUNT(*) FROM identities WHERE nick = 'alice'", [], |r| r.get(0)).unwrap() == 0).await;
    assert_eq!(s.core.conn_async(|c| c.query_row::<i64, _, _>("SELECT COUNT(*) FROM reserved_localparts WHERE localpart = 'alice'", [], |r| r.get(0)).unwrap()).await, 1);
    assert_eq!(s.send("GET", "/_matrix/client/v3/capabilities", Some(&a_tok), None).await.0, 401);
}

#[tokio::test]
async fn login_password_doors_and_tier_flag() {
    let s = stack().await;
    let (nick, _) = s.register().await;
    let (st, b) = s.post("/product/v1/login", None, json!({"nick":nick,"password":"correct horse"})).await;
    assert_eq!(st, 200, "{b}");
    assert_eq!(s.post("/product/v1/login", None, json!({"nick":nick,"password":"wrong"})).await.0, 403);
    assert_eq!(s.post("/product/v1/login", None, json!({"nick":"nobody1","password":"x"})).await.0, 403);
    assert_eq!(s.post("/product/v1/register", None, json!({"password":"short"})).await.0, 400);

    let (_, d) = s.send("GET", "/product/v1/doors", None, None).await;
    assert_eq!(d["doors"], json!(["fake"]));
    let (st, one) = s.post("/product/v1/login/door/fake", None, json!({"who":"@x:other.example"})).await;
    assert_eq!(st, 200, "{one}");
    let (_, two) = s.post("/product/v1/login/door/fake", None, json!({"who":"@x:other.example"})).await;
    assert_eq!(one["nick"], two["nick"], "same foreign address, same user; nick is a placeholder, not the address");
    assert_eq!(one["nick"].as_str().unwrap().len(), 8);
    assert_eq!(s.post("/product/v1/login/door/none", None, json!({})).await.0, 404);

    let tok = one["token"].as_str().unwrap();
    assert_eq!(s.send("GET", "/product/v1/me", Some(tok), None).await.1["flag"], 0);
    let n = one["nick"].as_str().unwrap();
    assert_eq!(s.post("/product/v1/admin/tier", Some(ADMIN), json!({"nick":n,"tier":"paid"})).await.0, 200);
    assert_eq!(s.send("GET", "/product/v1/me", Some(tok), None).await.1["flag"], 1);
    assert_eq!(s.post("/product/v1/admin/tier", Some(ADMIN), json!({"nick":n,"tier":"gold"})).await.0, 400);

    // Roles: kind and role are independent; only the token grants; an admin account may tier but not grant.
    assert_eq!(s.send("GET", "/product/v1/me", Some(tok), None).await.1["role"], "user");
    assert_eq!(s.post("/product/v1/admin/tier", Some(tok), json!({"nick":n,"tier":"free"})).await.0, 403, "a plain user is not staff");
    assert_eq!(s.post("/product/v1/admin/role", Some(ADMIN), json!({"nick":n,"role":"moderator"})).await.0, 200);
    assert_eq!(s.post("/product/v1/admin/tier", Some(tok), json!({"nick":n,"tier":"free"})).await.0, 403, "a moderator cannot change tiers");
    let (st, r) = s.post("/product/v1/admin/role", Some(ADMIN), json!({"nick":n,"kind":"human","role":"admin"})).await;
    assert_eq!((st, r["kind"].as_str(), r["role"].as_str()), (200, Some("human"), Some("admin")));
    assert_eq!(s.post("/product/v1/admin/tier", Some(tok), json!({"nick":n,"tier":"free"})).await.0, 200, "an admin account may");
    assert_eq!(s.post("/product/v1/admin/role", Some(tok), json!({"nick":n,"role":"user"})).await.0, 403, "only the token grants roles");
    assert_eq!(s.post("/product/v1/admin/role", Some(ADMIN), json!({"nick":n,"role":"root"})).await.0, 400);
    let me = s.send("GET", "/product/v1/me", Some(tok), None).await.1;
    assert_eq!((me["kind"].as_str(), me["role"].as_str()), (Some("human"), Some("admin")));
    let (st, pr) = s.send("GET", &format!("/product/v1/profile/{n}"), None, None).await;
    assert_eq!((st, pr["kind"].as_str(), pr["role"].as_str()), (200, Some("human"), Some("admin")), "public profile");
    assert_eq!(s.send("GET", "/product/v1/profile/nobody-here", None, None).await.0, 404);
}

#[tokio::test]
async fn push_socket_is_signed_by_the_product_relayed_through_the_edge_and_registered_by_identity() {
    use futures_util::StreamExt;
    use tokio_tungstenite::tungstenite::{client::IntoClientRequest, Message};
    let s = stack().await;
    let (_, a_tok) = s.register().await;
    let (_, b_tok) = s.register().await;
    let ws = s.product.replacen("http", "ws", 1);
    // No token: refused before the upgrade. Wrong token: refused too.
    assert!(tokio_tungstenite::connect_async(format!("{ws}/client/v3/push")).await.is_err());
    let mut bad = format!("{ws}/_matrix/client/v3/push").into_client_request().unwrap();
    bad.headers_mut().insert("authorization", "Bearer nope".parse().unwrap());
    assert!(tokio_tungstenite::connect_async(bad).await.is_err());
    // Valid token: registered at once, no token frame needed.
    let mut req = format!("{ws}/client/v3/push").into_client_request().unwrap();
    req.headers_mut().insert("authorization", format!("Bearer {a_tok}").parse().unwrap());
    let (mut sock, _) = tokio_tungstenite::connect_async(req).await.expect("handshake");
    let first = tokio::time::timeout(Duration::from_secs(5), sock.next()).await.expect("registered in time").unwrap().unwrap();
    assert_eq!(first.into_text().unwrap().as_str(), r#"{"type":"registered"}"#);
    // The registration is by verified identity: an event for alice reaches the socket.
    s.send("GET", "/_matrix/client/v3/capabilities", Some(&b_tok), None).await;
    let a_nick = s.send("GET", "/product/v1/me", Some(&a_tok), None).await.1["nick"].as_str().unwrap().to_string();
    let (_, b) = s.post("/_matrix/client/v3/createRoom", Some(&b_tok), json!({"preset":"private_chat","name":"pub"})).await;
    let room = b["room_id"].as_str().unwrap().to_string();
    let (st, b) = s.post(&format!("/_matrix/client/v3/rooms/{room}/invite"), Some(&b_tok), json!({"user_id": format!("@{a_nick}:example.org")})).await;
    assert_eq!(st, 200, "{b}");
    let (st, _) = s.post(&format!("/_matrix/client/v3/rooms/{room}/join"), Some(&a_tok), json!({})).await;
    assert_eq!(st, 200);
    let (st, b) = s.send("PUT", &format!("/_matrix/client/v3/rooms/{room}/send/m.room.encrypted/p1"), Some(&b_tok), Some(json!({"algorithm":"m.megolm.v1.aes-sha2","ciphertext":"x","sender_key":"k","session_id":"s","device_id":"d"}))).await;
    assert_eq!(st, 200, "{b}");
    let got = tokio::time::timeout(Duration::from_secs(5), async {
        while let Some(Ok(Message::Text(t))) = sock.next().await {
            if t.contains("\"event\"") {
                return t.to_string();
            }
        }
        String::new()
    })
    .await
    .expect("push event in time");
    assert!(got.contains(&room), "{got}");
    let _ = sock.close(None).await;
}

#[tokio::test]
async fn barrier_token_gates_the_edge_but_not_the_public_protocol_surfaces() {
    let s = stack().await;
    let c = &s.c;
    // Without the token (or with a wrong one) the edge refuses client paths, even with a valid-looking header.
    let r = c.get(format!("{}/_matrix/client/v3/capabilities", s.edge)).send().await.unwrap();
    assert_eq!((r.status().as_u16(), r.json::<Value>().await.unwrap()["errcode"].as_str().map(str::to_string)), (401, Some("M4A_LINK_TOKEN_REQUIRED".into())));
    let r = c.get(format!("{}/_matrix/client/v3/capabilities", s.edge)).header("x-m4a-link-token", "wrong-token-0123456789").send().await.unwrap();
    assert_eq!(r.status().as_u16(), 401);
    // Public surfaces stay open (they are not product traffic).
    for p in ["/_matrix/client/versions", "/.well-known/matrix/client", "/edge/healthz"] {
        let r = c.get(format!("{}{p}", s.edge)).send().await.unwrap();
        assert_ne!(r.status().as_u16(), 401, "{p}");
    }
    // Through the product, with its token configured, everything works; a client-supplied token header is dropped.
    let (_, tok) = s.register().await;
    let r = c.get(format!("{}/_matrix/client/v3/capabilities", s.product)).bearer_auth(&tok).header("x-m4a-link-token", "client-forged-0123456789").send().await.unwrap();
    assert_eq!(r.status().as_u16(), 200);
}

#[tokio::test]
async fn unix_sockets_carry_product_to_edge_to_core_including_the_push_socket() {
    use futures_util::StreamExt;
    use tokio_tungstenite::tungstenite::client::IntoClientRequest;
    let dir = std::env::temp_dir().join(format!("m4a-unix-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let s = stack_with(Some(&dir)).await;
    let (_, tok) = s.register().await;
    // A normal request crosses both unix hops and the core answers for the asserted identity.
    let (st, who) = s.send("GET", "/_matrix/client/v3/account/whoami", Some(&tok), None).await;
    assert_eq!(st, 200, "{who}");
    assert!(who["user_id"].as_str().unwrap().ends_with(":example.org"));
    // The push socket is relayed over both hops as well.
    let mut req = format!("{}/client/v3/push", s.product.replacen("http", "ws", 1)).into_client_request().unwrap();
    req.headers_mut().insert("authorization", format!("Bearer {tok}").parse().unwrap());
    let (mut sock, _) = tokio_tungstenite::connect_async(req).await.expect("handshake");
    let first = tokio::time::timeout(Duration::from_secs(5), sock.next()).await.expect("registered in time").unwrap().unwrap();
    assert_eq!(first.into_text().unwrap().as_str(), r#"{"type":"registered"}"#);
    // The barrier still applies: a request that skips the product link token is refused by the edge.
    let c = reqwest::Client::builder().unix_socket(dir.join("edge.sock")).build().unwrap();
    let r = c.get("http://edge.local/_matrix/client/v3/capabilities").send().await.unwrap();
    assert_eq!(r.status().as_u16(), 401);
    let _ = std::fs::remove_dir_all(&dir);
}

#[tokio::test]
async fn anonymous_read_serves_public_rooms_only_and_never_writes() {
    let s = stack().await;
    let (_, tok) = s.register().await;
    // A public channel with one plaintext post, and a private room with one encrypted event.
    let (_, ch) = s.post("/_matrix/client/v3/createRoom", Some(&tok), json!({"visibility":"public","name":"open"})).await;
    let chan = ch["room_id"].as_str().unwrap().to_string();
    let (st, b) = s.send("PUT", &format!("/_matrix/client/v3/rooms/{chan}/send/m.room.message/a1"), Some(&tok), Some(json!({"msgtype":"m.text","body":"hello world"}))).await;
    assert_eq!(st, 200, "{b}");
    let (_, pr) = s.post("/_matrix/client/v3/createRoom", Some(&tok), json!({"preset":"private_chat","name":"closed"})).await;
    let closed = pr["room_id"].as_str().unwrap().to_string();
    // No token at all.
    let (st, b) = s.send("GET", &format!("/_matrix/client/v3/rooms/{chan}/messages?dir=b"), None, None).await;
    assert_eq!(st, 200, "{b}");
    assert!(b["chunk"].to_string().contains("hello world"), "{b}");
    let (st, _) = s.send("GET", "/_matrix/client/v3/publicRooms", None, None).await;
    assert_eq!(st, 200);
    // Closed rooms stay closed to anonymous readers.
    let (st, _) = s.send("GET", &format!("/_matrix/client/v3/rooms/{closed}/messages?dir=b"), None, None).await;
    assert_eq!(st, 403);
    // Anonymous writes and non-allowlisted reads never reach the core with an identity.
    let (st, _) = s.send("PUT", &format!("/_matrix/client/v3/rooms/{chan}/send/m.room.message/a2"), None, Some(json!({"msgtype":"m.text","body":"x"}))).await;
    assert_eq!(st, 401);
    let (st, _) = s.send("GET", "/_matrix/client/v3/sync", None, None).await;
    assert_eq!(st, 401);
    let (st, _) = s.send("GET", "/_matrix/client/v3/account/whoami", None, None).await;
    assert_eq!(st, 401);
    // The core refuses the marker itself outside the allowlist, even from a holder of the link token.
    let r = s.c.get(format!("{}/_matrix/client/v3/account/whoami", s.edge)).header("x-m4a-link-token", LINK).header("x-m4a-anon-read", "1").send().await.unwrap();
    assert_eq!(r.status().as_u16(), 401);
    // A user's token is not downgraded: an authenticated read still sees its own membership.
    let (st, b) = s.send("GET", &format!("/_matrix/client/v3/rooms/{closed}/messages?dir=b"), Some(&tok), None).await;
    assert_eq!(st, 200, "{b}");
}

#[tokio::test]
async fn startup_reconcile_applies_revocations_and_deletions_the_messenger_never_heard_of() {
    let s = stack().await;
    let (a_nick, a_tok) = s.register().await;
    let (b_nick, b_tok) = s.register().await;
    for t in [&a_tok, &b_tok] {
        assert_eq!(s.send("GET", "/_matrix/client/v3/account/whoami", Some(t), None).await.0, 200);
    }
    let count = |sql: &'static str| move |c: &Connection| c.query_row::<i64, _, _>(sql, [], |r| r.get(0)).unwrap();
    assert_eq!(s.core.conn_async(move |c| count("SELECT COUNT(*) FROM identities")(c)).await, 2);
    // The product lost track of b entirely and of a's only credential, without any event having been sent.
    let snap = m4a_seam::Reconcile { id: "r1".into(), complete: true, nicks: vec![m4a_seam::LiveNick { nick: a_nick.clone(), creds: vec![] }] };
    m4a_product_kit::send_reconcile(&s.link, None, &snap, Duration::from_millis(20), 3).await.unwrap();
    s.until("b retired and a's device gone", move |c| count("SELECT COUNT(*) FROM identities")(c) == 1 && count("SELECT COUNT(*) FROM devices")(c) == 0).await;
    let left: String = s.core.conn_async(|c| c.query_row("SELECT nick FROM identities", [], |r| r.get(0)).unwrap()).await;
    assert_eq!(left, a_nick);
    assert_ne!(left, b_nick);
    // A partial snapshot never retires anyone.
    let partial = m4a_seam::Reconcile { id: "r2".into(), complete: false, nicks: vec![] };
    m4a_product_kit::send_reconcile(&s.link, None, &partial, Duration::from_millis(20), 3).await.unwrap();
    assert_eq!(s.core.conn_async(move |c| count("SELECT COUNT(*) FROM identities")(c)).await, 1);
    // A forged snapshot is refused.
    let forged = s.c.post(format!("{}/_matrix/account-source/v1/reconcile", s.edge)).header("x-m4a-link-token", LINK).header(m4a_seam::DEFAULT_EVENT_SIG_HEADER, "00").body("{\"id\":\"x\",\"complete\":true,\"nicks\":[]}").send().await.unwrap();
    assert_eq!(forged.status().as_u16(), 401);
}

async fn open_push(s: &Stack, tok: &str) -> tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>> {
    use futures_util::StreamExt;
    use tokio_tungstenite::tungstenite::client::IntoClientRequest;
    let mut req = format!("{}/client/v3/push", s.product.replacen("http", "ws", 1)).into_client_request().unwrap();
    req.headers_mut().insert("authorization", format!("Bearer {tok}").parse().unwrap());
    let (mut sock, _) = tokio_tungstenite::connect_async(req).await.expect("handshake");
    let first = tokio::time::timeout(Duration::from_secs(5), sock.next()).await.expect("registered").unwrap().unwrap();
    assert!(first.into_text().unwrap().contains("registered"));
    sock
}

/// The socket must end (close frame, error or EOF) soon after the event is applied.
async fn assert_closes(sock: &mut tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>) {
    use futures_util::StreamExt;
    let ended = tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            match sock.next().await {
                None | Some(Err(_)) | Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => return,
                Some(Ok(_)) => {}
            }
        }
    })
    .await;
    assert!(ended.is_ok(), "push socket stayed open");
}

#[tokio::test]
async fn push_sockets_close_when_the_credential_is_revoked_or_the_account_deleted_or_reconcile_removes_it() {
    let s = stack().await;
    // Logout revokes the credential: that user's socket ends; another user's stays up.
    let (_, a_tok) = s.register().await;
    let (b_nick, b_tok) = s.register().await;
    let mut a = open_push(&s, &a_tok).await;
    let mut b = open_push(&s, &b_tok).await;
    assert_eq!(s.send("POST", "/product/v1/logout", Some(&a_tok), None).await.0, 200);
    assert_closes(&mut a).await;
    // Admin delete: the account's socket ends.
    let (st, _) = s.send("POST", "/product/v1/admin/delete", Some(ADMIN), Some(json!({"nick": b_nick}))).await;
    assert_eq!(st, 200);
    assert_closes(&mut b).await;
    // Reconcile removal: the product forgot the credential without any event.
    let (c_nick, c_tok) = s.register().await;
    let mut c = open_push(&s, &c_tok).await;
    let snap = m4a_seam::Reconcile { id: "rp".into(), complete: false, nicks: vec![m4a_seam::LiveNick { nick: c_nick, creds: vec![] }] };
    m4a_product_kit::send_reconcile(&s.link, None, &snap, Duration::from_millis(20), 3).await.unwrap();
    assert_closes(&mut c).await;
}

#[tokio::test]
async fn login_tokens_refresh_tokens_password_change_3pid_and_deactivation() {
    let s = stack().await;
    let (nick, _first) = s.register().await;
    let login = |extra: Value| {
        let mut b = json!({"type":"m.login.password","identifier":{"type":"m.id.user","user":nick},"password":"correct horse"});
        for (k, v) in extra.as_object().unwrap() {
            b[k] = v.clone();
        }
        b
    };

    let (_, flows) = s.send("GET", "/client/v3/login", None, None).await;
    assert!(flows["flows"].as_array().unwrap().iter().any(|f| f["type"] == "m.login.token" && f["get_login_token"] == true), "{flows}");

    // Login asking for a refresh token.
    let (st, a) = s.post("/client/v3/login", None, login(json!({"refresh_token": true}))).await;
    assert_eq!(st, 200, "{a}");
    let (tok, refresh) = (a["access_token"].as_str().unwrap().to_string(), a["refresh_token"].as_str().unwrap().to_string());
    assert_eq!(a["expires_in_ms"], 3_600_000);
    let device = a["device_id"].clone();
    let whoami = |t: String| {
        let s = &s;
        async move { s.send("GET", "/client/v3/account/whoami", Some(&t), None).await }
    };
    assert_eq!(whoami(tok.clone()).await.0, 200);

    // Refresh: new tokens, the same device; the old access token and the used refresh token are dead.
    let (st, r) = s.post("/client/v3/refresh", None, json!({"refresh_token": refresh})).await;
    assert_eq!(st, 200, "{r}");
    let (tok2, refresh2) = (r["access_token"].as_str().unwrap().to_string(), r["refresh_token"].as_str().unwrap().to_string());
    assert_ne!(tok2, tok);
    let (st, w) = whoami(tok2.clone()).await;
    assert_eq!((st, &w["device_id"]), (200, &device), "{w}");
    assert_eq!(whoami(tok.clone()).await.0, 401);
    assert_eq!(s.post("/client/v3/refresh", None, json!({"refresh_token": refresh})).await.0, 401);
    let (st, r3) = s.post("/client/v3/refresh", None, json!({"refresh_token": refresh2})).await;
    assert_eq!(st, 200, "{r3}");
    let tok3 = r3["access_token"].as_str().unwrap().to_string();

    // Capabilities say what the product provides.
    let (_, caps) = s.send("GET", "/client/v3/capabilities", Some(&tok3), None).await;
    assert_eq!(caps["capabilities"]["m.get_login_token"]["enabled"], true, "{caps}");

    // get_token needs the password again; the token logs another client in exactly once.
    assert_eq!(s.post("/client/v1/login/get_token", Some(&tok3), json!({})).await.0, 401);
    let bad = json!({"auth":{"type":"m.login.password","identifier":{"type":"m.id.user","user":nick},"password":"nope"}});
    assert_eq!(s.post("/client/v1/login/get_token", Some(&tok3), bad).await.0, 401);
    let good = json!({"auth":{"type":"m.login.password","identifier":{"type":"m.id.user","user":nick},"password":"correct horse"}});
    let (st, t) = s.post("/client/v1/login/get_token", Some(&tok3), good.clone()).await;
    assert_eq!(st, 200, "{t}");
    let lt = t["login_token"].as_str().unwrap().to_string();
    let (st, b) = s.post("/client/v3/login", None, json!({"type":"m.login.token","token":lt})).await;
    assert_eq!(st, 200, "{b}");
    let other = b["access_token"].as_str().unwrap().to_string();
    assert_ne!(b["device_id"], device, "a new session is a new device");
    assert_eq!(s.post("/client/v3/login", None, json!({"type":"m.login.token","token":lt})).await.0, 403, "one use");

    // Password change: wrong re-auth refused; a short password refused; success signs the others out.
    let change = |pw: &str, auth: &Value| json!({"new_password": pw, "auth": auth});
    assert_eq!(s.post("/client/v3/account/password", Some(&tok3), json!({"new_password":"a better one"})).await.0, 401);
    assert_eq!(s.post("/client/v3/account/password", Some(&tok3), change("short", &good["auth"])).await.0, 400);
    let (st, v) = s.post("/client/v3/account/password", Some(&tok3), change("a much better one", &good["auth"])).await;
    assert_eq!(st, 200, "{v}");
    assert_eq!(whoami(other.clone()).await.0, 401, "the other session is gone");
    assert_eq!(whoami(tok3.clone()).await.0, 200, "this one stays");
    assert_eq!(s.post("/client/v3/login", None, login(json!({}))).await.0, 403, "old password");
    let (st, n) = s.post("/client/v3/login", None, login(json!({"password":"a much better one"}))).await;
    assert_eq!(st, 200, "{n}");

    // Third-party identifiers: none, and none can be added.
    let (_, p) = s.send("GET", "/client/v3/account/3pid", Some(&tok3), None).await;
    assert_eq!(p["threepids"], json!([]));
    let (st, d) = s.post("/client/v3/account/3pid/add", Some(&tok3), json!({})).await;
    assert_eq!((st, d["errcode"].as_str()), (403, Some("M_THREEPID_DENIED")));

    // Deactivation re-authenticates, then the account and its sessions are gone.
    let new_auth = json!({"auth":{"type":"m.login.password","identifier":{"type":"m.id.user","user":nick},"password":"a much better one"}});
    assert_eq!(s.post("/client/v3/account/deactivate", Some(&tok3), json!({})).await.0, 401);
    let (st, v) = s.post("/client/v3/account/deactivate", Some(&tok3), new_auth).await;
    assert_eq!(st, 200, "{v}");
    assert_eq!(whoami(tok3).await.0, 401);
    assert_eq!(s.post("/client/v3/login", None, login(json!({"password":"a much better one"}))).await.0, 403);
}

// ---- Login by proof of possession of a key: the client resolves identity and credentials itself.

use m4a_agent::backend::{matrix::MatrixBackend, server::ServerBackend};
use m4a_agent::{AgentError, Backend, BackendKind, IdentityStore, KeyVault, MemoryVault};
use m4a_seam::keyproof;

impl Stack {
    async fn invite(&self, nick: Option<&str>) -> (String, String) {
        let body = nick.map(|n| json!({ "nick": n })).unwrap_or(json!({}));
        let (st, b) = self.post("/product/v1/admin/invite", Some(ADMIN), body).await;
        assert_eq!(st, 200, "{b}");
        (b["invite"].as_str().unwrap().to_string(), b["nick"].as_str().unwrap().to_string())
    }
}

/// The client half is blocking; run it off the async runtime.
async fn off<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> T {
    tokio::task::spawn_blocking(f).await.unwrap()
}

#[tokio::test]
async fn an_agent_client_enrolls_once_and_then_logs_in_by_signature_only() {
    let s = stack().await;
    assert_eq!(s.send("GET", "/client/v3/login", None, None).await.1["flows"].as_array().unwrap().iter().filter(|f| f["type"] == keyproof::LOGIN_TYPE).count(), 1);
    let (invite, nick) = s.invite(None).await;
    let vault: Arc<dyn KeyVault> = Arc::new(MemoryVault::new());
    let ids = Arc::new(IdentityStore::new(vault));
    let url = s.product.clone();

    let (ids2, url2, inv) = (ids.clone(), url.clone(), invite.clone());
    let (first, again, relog, key_id, pubkey) = off(move || {
        let b = ServerBackend::new(&url2).unwrap();
        let mut id = ids2.resolve("agent-1", BackendKind::Server, b.server_ref()).unwrap();
        // Not enrolled and no invite: the client says so, it does not invent anything.
        assert!(matches!(b.ensure_session(&ids2, &mut id, None), Err(AgentError::NeedsInvite)));
        let first = b.ensure_session(&ids2, &mut id, Some(&inv)).unwrap();
        assert!(id.enrolled);
        // Same token while the server accepts it; no new login.
        let again = b.ensure_session(&ids2, &mut id, None).unwrap();
        // Token lost: a signed login gives a fresh one for the SAME credential.
        ids2.vault().delete(&m4a_agent::identity::token_label("agent-1")).unwrap();
        let relog = b.ensure_session(&ids2, &mut id, None).unwrap();
        (first, again, relog, id.key_id.clone(), id.public_key.clone())
    })
    .await;
    assert_eq!(first.nick, nick, "the nick the operator assigned");
    assert_eq!(again.token.as_str(), first.token.as_str());
    assert_ne!(relog.token.as_str(), first.token.as_str());
    assert_eq!((relog.cred_ref.as_str(), first.cred_ref.as_str()), (key_id.as_str(), key_id.as_str()), "one credential set");
    assert_eq!(s.send("GET", "/product/v1/me", Some(&relog.token), None).await.1["nick"], json!(nick));
    assert_eq!(s.send("GET", "/product/v1/me", Some(&first.token), None).await.0, 401, "the old token is replaced");

    // The invite is spent.
    let other = ids.resolve("agent-2", BackendKind::Server, &url).unwrap();
    let (ids3, url3, inv) = (ids.clone(), url.clone(), invite.clone());
    let mut other = other;
    let spent = off(move || ServerBackend::new(&url3).unwrap().ensure_session(&ids3, &mut other, Some(&inv))).await;
    assert!(matches!(spent, Err(AgentError::Refused(_))), "{spent:?}");

    // Replay: a signed challenge works once; the same answer again does not.
    let (_, c) = s.post("/product/v1/login/key/challenge", None, json!({ "key_id": key_id })).await;
    let ch = keyproof::Challenge { challenge_id: c["challenge_id"].as_str().unwrap().into(), nonce: c["nonce"].as_str().unwrap().into(), expires_ms: c["expires_ms"].as_i64().unwrap(), audience: c["audience"].as_str().unwrap().into() };
    assert_eq!(ch.audience, "test-product");
    let id = ids.resolve("agent-1", BackendKind::Server, &url).unwrap();
    let sig = id.sign_login(ids.vault(), &ch).unwrap();
    let body = json!({ "key_id": key_id, "challenge_id": ch.challenge_id, "signature": sig });
    assert_eq!(s.post("/product/v1/login/key", None, body.clone()).await.0, 200);
    assert_eq!(s.post("/product/v1/login/key", None, body).await.0, 401, "replay");
    // A signature over a changed audience, a challenge for another key, a made-up key: all refused.
    let (_, c) = s.post("/product/v1/login/key/challenge", None, json!({ "key_id": key_id })).await;
    let ch = keyproof::Challenge { challenge_id: c["challenge_id"].as_str().unwrap().into(), nonce: c["nonce"].as_str().unwrap().into(), expires_ms: c["expires_ms"].as_i64().unwrap(), audience: "elsewhere".into() };
    let sig = id.sign_login(ids.vault(), &ch).unwrap();
    assert_eq!(s.post("/product/v1/login/key", None, json!({ "key_id": key_id, "challenge_id": ch.challenge_id, "signature": sig })).await.0, 401);
    let (_, c) = s.post("/product/v1/login/key/challenge", None, json!({ "key_id": "kunknown" })).await;
    let ch = keyproof::Challenge { challenge_id: c["challenge_id"].as_str().unwrap().into(), nonce: c["nonce"].as_str().unwrap().into(), expires_ms: c["expires_ms"].as_i64().unwrap(), audience: c["audience"].as_str().unwrap().into() };
    let sig = id.sign_login(ids.vault(), &ch).unwrap();
    assert_eq!(s.post("/product/v1/login/key", None, json!({ "key_id": key_id, "challenge_id": ch.challenge_id, "signature": sig })).await.0, 401, "challenge belongs to another key id");
    assert_eq!(s.post("/product/v1/login/key", None, json!({ "key_id": "kunknown", "challenge_id": "x", "signature": "y" })).await.0, 401);
    // Enrolling needs the admin's invite; a forged one is refused.
    assert_eq!(s.post("/product/v1/enroll", None, json!({ "invite": "nope", "public_key": pubkey, "signature": "AAAA" })).await.0, 401);
    assert_eq!(s.post("/product/v1/admin/invite", None, json!({})).await.0, 403, "only the operator invites");

    // Revoking the credential kills the key: the signature no longer logs in.
    let (st, _) = s.post("/product/v1/admin/revoke", Some(ADMIN), json!({ "cred_ref": key_id })).await;
    assert_eq!(st, 200);
    let (ids4, url4) = (ids.clone(), url.clone());
    let mut id = id;
    let r = off(move || ServerBackend::new(&url4).unwrap().ensure_session(&ids4, &mut id, None)).await;
    assert!(matches!(r, Err(AgentError::Refused(_))), "{r:?}");
}

#[tokio::test]
async fn an_agent_client_logs_in_over_the_matrix_api_with_the_custom_login_type() {
    let s = stack().await;
    let (invite, nick) = s.invite(None).await;
    let ids = Arc::new(IdentityStore::new(Arc::new(MemoryVault::new())));
    let url = s.product.clone();
    let (ids2, inv) = (ids.clone(), invite.clone());
    let (sess, again) = off(move || {
        let b = MatrixBackend::new(&url).unwrap();
        let mut id = ids2.resolve("agent-m", BackendKind::Matrix, b.server_ref()).unwrap();
        let sess = b.ensure_session(&ids2, &mut id, Some(&inv)).unwrap();
        let again = b.ensure_session(&ids2, &mut id, None).unwrap();
        (sess, again)
    })
    .await;
    assert_eq!(sess.nick, nick);
    assert!(sess.user_id.as_deref().unwrap().starts_with(&format!("@{nick}:")), "{sess:?}");
    assert!(sess.device_id.is_some());
    assert_eq!(again.token.as_str(), sess.token.as_str());
    // The access token is a working Matrix token.
    let (st, w) = s.send("GET", "/_matrix/client/v3/account/whoami", Some(&sess.token), None).await;
    assert_eq!(st, 200, "{w}");
    assert_eq!(w["user_id"], json!(sess.user_id));
    // Without the signature the Matrix login answers with the challenge, and a bad one is refused.
    let (st, v) = s.post("/_matrix/client/v3/login", None, json!({ "type": keyproof::LOGIN_TYPE, "key_id": "kx" })).await;
    assert_eq!((st, v["errcode"].as_str()), (401, Some("M_UNAUTHORIZED")));
    assert!(v[keyproof::LOGIN_TYPE]["nonce"].is_string());
    let (st, _) = s.post("/_matrix/client/v3/login", None, json!({ "type": keyproof::LOGIN_TYPE, "key_id": "kx", "challenge_id": "c", "signature": "s" })).await;
    assert_eq!(st, 401);
    // Password users are unaffected and cannot be reached by a key.
    let (st, _) = s.post("/_matrix/client/v3/login", None, json!({ "type": "m.login.password", "user": nick, "password": "whatever-long" })).await;
    assert_eq!(st, 403, "an invited identity has no password");
}

// ---- The client asks for the nick of its session name; the invite (operator) still decides.

async fn enroll_as(s: &Stack, ids: &Arc<IdentityStore>, session: &str, invite: &str, ask: Option<&str>) -> Result<m4a_agent::Session, AgentError> {
    let (ids, url, session, invite, ask) = (ids.clone(), s.product.clone(), session.to_string(), invite.to_string(), ask.map(str::to_string));
    off(move || {
        let b = ServerBackend::new(&url).unwrap();
        let mut id = ids.resolve(&session, BackendKind::Server, b.server_ref()).unwrap();
        id.requested_nick = ask;
        b.ensure_session(&ids, &mut id, Some(&invite))
    })
    .await
}

#[tokio::test]
async fn a_requested_nick_replaces_a_placeholder_but_never_a_reserved_nick_and_never_burns_the_invite() {
    let s = stack().await;
    let ids = Arc::new(IdentityStore::new(Arc::new(MemoryVault::new())));

    // Placeholder invite: the nick asked for is taken, the operator's approval is the invite itself.
    let (inv, placeholder) = s.invite(None).await;
    let got = enroll_as(&s, &ids, "scribe-1", &inv, Some("scribe")).await.unwrap();
    assert_eq!(got.nick, "scribe");
    assert_ne!(got.nick, placeholder);
    assert_eq!(s.send("GET", "/product/v1/me", Some(&got.token), None).await.1["nick"], json!("scribe"));

    // Taken, and against the grammar: refused with the invite left usable; without the request it works.
    let (inv, placeholder) = s.invite(None).await;
    let r = enroll_as(&s, &ids, "scribe-2", &inv, Some("Scribe")).await;
    assert!(matches!(r, Err(AgentError::Refused(ref m)) if m.contains("requested nick")), "taken: {r:?}");
    let r = enroll_as(&s, &ids, "scribe-2", &inv, Some("not a nick!")).await;
    assert!(matches!(r, Err(AgentError::Refused(_))), "grammar: {r:?}");
    assert_eq!(enroll_as(&s, &ids, "scribe-2", &inv, None).await.unwrap().nick, placeholder);

    // Reserved by the operator: a different request is refused and the invite stays good; the same
    // nick (any case) or no request is accepted.
    let (inv, reserved) = s.invite(Some("keeper")).await;
    assert_eq!(reserved, "keeper");
    let r = enroll_as(&s, &ids, "keeper-1", &inv, Some("other")).await;
    assert!(matches!(r, Err(AgentError::Refused(_))), "reserved: {r:?}");
    assert_eq!(enroll_as(&s, &ids, "keeper-1", &inv, Some("KEEPER")).await.unwrap().nick, "keeper");

    // No invite, no nick: the request alone enrolls nobody.
    let (_, b) = s.post("/product/v1/enroll", None, json!({ "invite": "forged", "public_key": "AAAA", "signature": "AAAA", "nick": "intruder" })).await;
    assert!(b.get("token").is_none(), "{b}");
}

async fn fresh_challenge(s: &Stack, kid: &str) -> keyproof::Challenge {
    let (_, c) = s.post("/product/v1/login/key/challenge", None, json!({ "key_id": kid })).await;
    keyproof::Challenge { challenge_id: c["challenge_id"].as_str().unwrap().into(), nonce: c["nonce"].as_str().unwrap().into(), expires_ms: c["expires_ms"].as_i64().unwrap(), audience: c["audience"].as_str().unwrap().into() }
}

// ---- A Matrix login token from a key signature (Element Web: #/login?loginToken=).

#[tokio::test]
async fn a_key_signature_gives_a_one_time_matrix_login_token_without_touching_the_keys_session() {
    let s = stack().await;
    let (inv, _) = s.invite(Some("opener")).await;
    let ids = Arc::new(IdentityStore::new(Arc::new(MemoryVault::new())));
    let sess = enroll_as(&s, &ids, "opener-1", &inv, None).await.unwrap();
    let id = ids.resolve("opener-1", BackendKind::Server, &s.product).unwrap();

    let ch = fresh_challenge(&s, &id.key_id).await;
    let sig = id.sign_login(ids.vault(), &ch).unwrap();
    let body = json!({ "key_id": id.key_id, "challenge_id": ch.challenge_id, "signature": sig });
    let (st, t) = s.post("/product/v1/login/key/token", None, body.clone()).await;
    assert_eq!(st, 200, "{t}");
    let token = t["login_token"].as_str().unwrap().to_string();
    assert!(t["expires_in_ms"].as_i64().unwrap() <= 300_000, "short lived: {t}");
    // The key's own session is untouched: the proof alone issued the token.
    assert_eq!(s.send("GET", "/product/v1/me", Some(&sess.token), None).await.0, 200);
    // The challenge is spent.
    assert_eq!(s.post("/product/v1/login/key/token", None, body).await.0, 401);

    // The token logs a Matrix client in, once, as the same user.
    let (st, l) = s.post("/_matrix/client/v3/login", None, json!({ "type": "m.login.token", "token": token })).await;
    assert_eq!(st, 200, "{l}");
    assert!(l["user_id"].as_str().unwrap().starts_with("@opener:"), "{l}");
    let access = l["access_token"].as_str().unwrap().to_string();
    assert_eq!(s.send("GET", "/_matrix/client/v3/account/whoami", Some(&access), None).await.0, 200);
    assert_eq!(s.post("/_matrix/client/v3/login", None, json!({ "type": "m.login.token", "token": token })).await.0, 403, "single use");

    // A wrong signature and an unknown key get no token.
    let ch = fresh_challenge(&s, &id.key_id).await;
    assert_eq!(s.post("/product/v1/login/key/token", None, json!({ "key_id": id.key_id, "challenge_id": ch.challenge_id, "signature": "AAAA" })).await.0, 401);
    assert_eq!(s.post("/product/v1/login/key/token", None, json!({ "key_id": "kunknown", "challenge_id": "x", "signature": "y" })).await.0, 401);
}

// ---- A new key for an account that exists (lost key, second device).

#[tokio::test]
async fn an_invite_for_an_existing_account_adds_a_key_to_it_and_nothing_else() {
    let s = stack().await;
    let ids = Arc::new(IdentityStore::new(Arc::new(MemoryVault::new())));
    let (inv, nick) = s.invite(Some("keeper")).await;
    let first = enroll_as(&s, &ids, "keeper-old", &inv, None).await.unwrap();
    assert_eq!(first.nick, nick);

    // The operator makes an invite for the account; the answer says so and names the nick.
    let (st, b) = s.post("/product/v1/admin/invite", Some(ADMIN), json!({ "nick": "keeper", "existing": true })).await;
    assert_eq!(st, 200, "{b}");
    assert_eq!((b["nick"].as_str(), b["existing"].as_bool()), (Some("keeper"), Some(true)));
    let again = b["invite"].as_str().unwrap().to_string();

    // A fresh identity enrolls with it and is the SAME account; the old key still works until revoked.
    let second = enroll_as(&s, &ids, "keeper-new", &again, Some("keeper")).await.unwrap();
    assert_eq!(second.nick, "keeper");
    assert_ne!(second.cred_ref, first.cred_ref, "its own credential");
    assert_eq!(s.send("GET", "/product/v1/me", Some(&second.token), None).await.1["nick"], json!("keeper"));
    let old = ids.resolve("keeper-old", BackendKind::Server, &s.product).unwrap();
    let relog = { let (ids2, url) = (ids.clone(), s.product.clone()); off(move || { let b = ServerBackend::new(&url).unwrap(); let mut id = ids2.resolve("keeper-old", BackendKind::Server, b.server_ref()).unwrap(); ids2.vault().delete(&m4a_agent::identity::token_label("keeper-old")).unwrap(); b.ensure_session(&ids2, &mut id, None) }).await };
    assert!(relog.is_ok(), "the old key still logs in: {relog:?} {}", old.key_id);

    // The invite is single use; another nick in the request is refused (it is reserved); an unknown
    // nick has no invite; and the invite needs the operator.
    let r = enroll_as(&s, &ids, "keeper-third", &again, None).await;
    assert!(matches!(r, Err(AgentError::Refused(_))), "spent: {r:?}");
    let (st, b) = s.post("/product/v1/admin/invite", Some(ADMIN), json!({ "nick": "keeper", "existing": true })).await;
    assert_eq!(st, 200, "{b}");
    let r = enroll_as(&s, &ids, "keeper-fourth", b["invite"].as_str().unwrap(), Some("someone-else")).await;
    assert!(matches!(r, Err(AgentError::Refused(_))), "reserved: {r:?}");
    assert_eq!(s.post("/product/v1/admin/invite", Some(ADMIN), json!({ "nick": "nobody-here", "existing": true })).await.0, 404);
    assert_eq!(s.post("/product/v1/admin/invite", Some(ADMIN), json!({ "existing": true })).await.0, 400);
    assert_eq!(s.post("/product/v1/admin/invite", None, json!({ "nick": "keeper", "existing": true })).await.0, 403);
}