vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

refactor: extract admin_room_send_and_poll helper — deduplicate poll loop

Browse source
This commit is contained in:
atlas 2026-06-04 15:22:14 +02:00
commit e06d3e3a8b
1 changed files with 67 additions and 125 deletions
Download patch file Download diff file Expand all files Collapse all files

192
hive-c0re/src/matrix.rs
View file

@ -430,25 +430,25 @@ mod extract_new_password_tests {
} }
} }
/// Reset a user's password via the Matrix admin room as a fallback for /// Send a command to the Matrix admin room and poll for a bot response.
/// homeservers that do not implement the Synapse admin REST API (e.g.
/// tuwunel 1.6.x). Sends `reset-password @<localpart>:<server>` to
/// the `#admins:<server>` room as @hive, then polls for the bot's
/// response message containing the new password.
/// ///
/// Returns the new password on success; the caller is responsible for /// 1. Anchors at the current end-token (`dir=b&limit=1`).
/// persisting it to [`password_path`]. /// 2. Sends `command` as an `m.text` message from the @hive admin account.
async fn admin_room_reset_password( /// 3. Polls up to 5 × 1 s for a bot message that satisfies `check`.
/// 4. Calls `check(body)` on each non-self message; returns `Ok(T)` on the
/// first `Some(T)`.
///
/// Generic over `T` so both password-returning and `()` callers share the loop.
async fn admin_room_send_and_poll<T>(
client: &reqwest::Client, client: &reqwest::Client,
admin_token: &str, admin_token: &str,
server_name: &str, server_name: &str,
localpart: &str, room_url: &str,
) -> Result<String> { command: &str,
let room_id = discover_admin_room_id(client, admin_token, server_name).await?; check: impl Fn(&str) -> Option<T>,
let room_url = encode_room_id_for_url(&room_id); ) -> Result<T> {
// Pagination anchor — fetch the current tail so we only read events
// Get the current messages end-token so we can poll forward for the // that arrive *after* our command, not old history.
// bot's response without re-reading old messages.
let msgs_back_url = format!( let msgs_back_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/messages?dir=b&limit=1" "{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/messages?dir=b&limit=1"
); );
@ -470,12 +470,11 @@ async fn admin_room_reset_password(
})? })?
.to_owned(); .to_owned();
// Send the password-reset command. // Send the command.
let txn_id = random_hex(8)?; let txn_id = random_hex(8)?;
let send_url = format!( let send_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/send/m.room.message/{txn_id}" "{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/send/m.room.message/{txn_id}"
); );
let command = format!("reset-password @{localpart}:{server_name}");
let send_resp = client let send_resp = client
.put(&send_url) .put(&send_url)
.bearer_auth(admin_token) .bearer_auth(admin_token)
@ -484,17 +483,14 @@ async fn admin_room_reset_password(
.await .await
.context("matrix: PUT admin room message")?; .context("matrix: PUT admin room message")?;
if !send_resp.status().is_success() { if !send_resp.status().is_success() {
let body = send_resp let body = send_resp.json::<serde_json::Value>().await.unwrap_or_default();
.json::<serde_json::Value>()
.await
.unwrap_or_default();
anyhow::bail!("matrix: admin room send failed: {body}"); anyhow::bail!("matrix: admin room send failed: {body}");
} }
// Poll for a bot response containing the new password (up to 15 × 1 s). // Poll for bot response (up to 15 × 1 s). tuwunel can take several
// tuwunel can take several seconds to process the admin-room command // seconds to process admin-room commands during startup.
// during startup when the homeserver is under load. let own_user_id = format!("@{HIVE_ADMIN_LOCALPART}:{server_name}");
for attempt in 0..15_u8 { for _ in 0..15_u8 {
tokio::time::sleep(std::time::Duration::from_secs(1)).await; tokio::time::sleep(std::time::Duration::from_secs(1)).await;
let from_encoded = from_token.replace(':', "%3A").replace('+', "%2B"); let from_encoded = from_token.replace(':', "%3A").replace('+', "%2B");
let poll_url = format!( let poll_url = format!(
@ -511,39 +507,55 @@ async fn admin_room_reset_password(
.context("matrix: parse admin room poll response")?; .context("matrix: parse admin room poll response")?;
if let Some(events) = poll_json["chunk"].as_array() { if let Some(events) = poll_json["chunk"].as_array() {
let own_user_id = format!("@{HIVE_ADMIN_LOCALPART}:{server_name}");
for event in events { for event in events {
if event["type"].as_str() != Some("m.room.message") { if event["type"].as_str() != Some("m.room.message") {
continue; continue;
} }
// Skip our own command message.
if event["sender"].as_str() == Some(own_user_id.as_str()) { if event["sender"].as_str() == Some(own_user_id.as_str()) {
continue; continue;
} }
let body = event["content"]["body"].as_str().unwrap_or(""); let body = event["content"]["body"].as_str().unwrap_or("");
if let Some(pw) = extract_new_password(body) { if let Some(result) = check(body) {
tracing::info!( return Ok(result);
%localpart,
attempt,
"matrix: admin room reset-password response parsed"
);
return Ok(pw);
} }
} }
} }
// Advance the pagination token so we don't re-read the same events.
if let Some(end) = poll_json["end"].as_str() { if let Some(end) = poll_json["end"].as_str() {
from_token = end.to_owned(); from_token = end.to_owned();
} }
} }
anyhow::bail!( anyhow::bail!(
"matrix: admin room reset-password for @{localpart}:{server_name}: no password \ "matrix: admin room command timed out after 15 seconds. \
response received within 15 seconds. Command sent: '{command}'. \ Command: '{command}'. No matching bot response received."
Verify the admin room accepts 'reset-password @user:server' commands."
) )
} }
/// Reset a user's password via the Matrix admin room (`#admins:<server>`).
/// Sends `reset-password @<localpart>:<server>` as @hive, polls for the bot's
/// response containing the new password.
///
/// Returns the new password; caller is responsible for persisting it.
async fn admin_room_reset_password(
client: &reqwest::Client,
admin_token: &str,
server_name: &str,
localpart: &str,
) -> Result<String> {
let room_id = discover_admin_room_id(client, admin_token, server_name).await?;
let room_url = encode_room_id_for_url(&room_id);
let command = format!("reset-password @{localpart}:{server_name}");
admin_room_send_and_poll(client, admin_token, server_name, &room_url, &command, extract_new_password)
.await
.with_context(|| {
format!(
"matrix: admin room reset-password for @{localpart}:{server_name}: \
no password response received within 5 seconds. \
Verify the admin room accepts 'reset-password @user:server' commands."
)
})
}
/// Ensure `name` has a matrix user + token file on the local /// Ensure `name` has a matrix user + token file on the local
/// homeserver. Skips provisioning entirely if the token file already /// homeserver. Skips provisioning entirely if the token file already
/// exists (treating a present token as proof the account is good). /// exists (treating a present token as proof the account is good).
@ -795,8 +807,8 @@ pub async fn ensure_admin_user(client: &reqwest::Client, register_token: &str) -
} }
/// Promote a user to homeserver admin via the Matrix admin room /// Promote a user to homeserver admin via the Matrix admin room
/// (`#admins:<server>`). Sends `make-user-admin @<localpart>:<server>` to /// (`#admins:<server>`). Sends `make-user-admin @<localpart>:<server>` as
/// the room as @hive, then polls for the bot's success response. /// @hive, polls for the bot's "Done:" success response.
/// ///
/// tuwunel 1.6.x does not implement `/_synapse/admin/v2/users`; all admin /// tuwunel 1.6.x does not implement `/_synapse/admin/v2/users`; all admin
/// operations go through the admin room. /// operations go through the admin room.
@ -808,93 +820,23 @@ pub async fn promote_user_to_admin(
) -> Result<()> { ) -> Result<()> {
let room_id = discover_admin_room_id(client, admin_token, server_name).await?; let room_id = discover_admin_room_id(client, admin_token, server_name).await?;
let room_url = encode_room_id_for_url(&room_id); let room_url = encode_room_id_for_url(&room_id);
// Anchor for forward-poll — grab current end-token before sending.
let msgs_back_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/messages?dir=b&limit=1"
);
let msgs_back = client
.get(&msgs_back_url)
.bearer_auth(admin_token)
.send()
.await
.context("matrix: admin room messages (back, promote)")?
.json::<serde_json::Value>()
.await
.context("matrix: parse admin room messages response (promote)")?;
let mut from_token = msgs_back["end"]
.as_str()
.ok_or_else(|| {
anyhow::anyhow!(
"matrix: admin room messages response missing 'end' token: {msgs_back}"
)
})?
.to_owned();
// Send the promote command.
let txn_id = random_hex(8)?;
let send_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/send/m.room.message/{txn_id}"
);
let command = format!("make-user-admin @{localpart}:{server_name}"); let command = format!("make-user-admin @{localpart}:{server_name}");
let send_resp = client admin_room_send_and_poll(client, admin_token, server_name, &room_url, &command, |body| {
.put(&send_url) let lower = body.to_ascii_lowercase();
.bearer_auth(admin_token) if lower.starts_with("done") || lower.contains("made") && lower.contains("admin") {
.json(&serde_json::json!({"msgtype": "m.text", "body": command})) Some(())
.send() } else {
.await None
.context("matrix: PUT admin room message (promote)")?;
if !send_resp.status().is_success() {
let body = send_resp
.json::<serde_json::Value>()
.await
.unwrap_or_default();
anyhow::bail!("matrix: admin room send failed (promote): {body}");
}
// Poll for a bot "Done:" response (up to 5 × 1 s).
for attempt in 0..5_u8 {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
let from_encoded = from_token.replace(':', "%3A").replace('+', "%2B");
let poll_url = format!(
"{MATRIX_HTTP}/_matrix/client/v3/rooms/{room_url}/messages?dir=f&from={from_encoded}&limit=10"
);
let poll_json = client
.get(&poll_url)
.bearer_auth(admin_token)
.send()
.await
.context("matrix: admin room poll (promote)")?
.json::<serde_json::Value>()
.await
.context("matrix: parse admin room poll response (promote)")?;
if let Some(events) = poll_json["chunk"].as_array() {
let own_user_id = format!("@{HIVE_ADMIN_LOCALPART}:{server_name}");
for event in events {
if event["type"].as_str() != Some("m.room.message") {
continue;
}
if event["sender"].as_str() == Some(own_user_id.as_str()) {
continue;
}
let body = event["content"]["body"].as_str().unwrap_or("").to_ascii_lowercase();
if body.starts_with("done") || body.contains("made") && body.contains("admin") {
tracing::info!(%localpart, attempt, "matrix: admin room make-user-admin succeeded");
return Ok(());
}
}
} }
if let Some(end) = poll_json["end"].as_str() { })
from_token = end.to_owned(); .await
} .with_context(|| {
} format!(
"matrix: admin room make-user-admin for @{localpart}:{server_name}: \
anyhow::bail!( no success response within 5 seconds. \
"matrix: admin room make-user-admin for @{localpart}:{server_name}: no success \ Verify the admin room accepts 'make-user-admin @user:server' commands."
response within 5 seconds. Command sent: '{command}'. \ )
Verify the admin room accepts 'make-user-admin @user:server' commands." })
)
} }
/// Reset a user's password via the Matrix admin room (`#admins:<server>`). /// Reset a user's password via the Matrix admin room (`#admins:<server>`).