gatewayd: mm, Mattermost's events and REST calls, typed
Implemented-By: OpenCode session (model recorded in docs/implementer-log.md)
This commit is contained in:
@@ -3,6 +3,7 @@
|
||||
|
||||
pub mod config;
|
||||
pub mod http;
|
||||
pub mod mm;
|
||||
pub mod net;
|
||||
pub mod secrets;
|
||||
pub mod ws;
|
||||
|
||||
@@ -0,0 +1,208 @@
|
||||
//! Mattermost's JSON, typed: the posts and users we read, the events that arrive over the
|
||||
//! WebSocket, and the requests we send over it. Mattermost's JSON is not ours: unknown fields are
|
||||
//! ignored, but every id we keep must be a valid id.
|
||||
|
||||
pub mod rest;
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use serde::Deserialize;
|
||||
|
||||
use crate::config::valid_id;
|
||||
|
||||
/// At most this many changed posts come back from one `posts?since` call (Mattermost v11.11.0,
|
||||
/// `SqlPostStore::GetPostsSince`); a full answer may have left some out.
|
||||
pub const SINCE_LIMIT: usize = 1000;
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum MmError {
|
||||
/// No answer: connecting, TLS, or the HTTP exchange failed.
|
||||
Net(String),
|
||||
/// 401 or 403: the token is refused.
|
||||
Auth(u16),
|
||||
/// 429, still, after waiting as asked.
|
||||
RateLimited(Duration),
|
||||
/// Any other status that is not 2xx, with the start of the body.
|
||||
Status(u16, String),
|
||||
/// The answer is not the JSON we expect.
|
||||
Json(String),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for MmError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
MmError::Net(why) => write!(f, "{why}"),
|
||||
MmError::Auth(status) => write!(f, "Mattermost refused the token ({status})"),
|
||||
MmError::RateLimited(wait) => {
|
||||
write!(f, "rate limited; asked to wait {} s", wait.as_secs())
|
||||
}
|
||||
// Quoted: the body is the server's text, and must not forge a log line.
|
||||
MmError::Status(status, body) => write!(f, "status {status}: {body:?}"),
|
||||
MmError::Json(why) => write!(f, "unexpected JSON: {why}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for MmError {}
|
||||
|
||||
/// This bot, from `GET /users/me`.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
|
||||
pub struct Me {
|
||||
pub id: String,
|
||||
pub username: String,
|
||||
}
|
||||
|
||||
/// The fields of a post that `gatewayd` uses.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
|
||||
pub struct Post {
|
||||
pub id: String,
|
||||
pub user_id: String,
|
||||
pub channel_id: String,
|
||||
#[serde(default)]
|
||||
pub root_id: String,
|
||||
#[serde(default)]
|
||||
pub message: String,
|
||||
pub create_at: i64,
|
||||
#[serde(default)]
|
||||
pub delete_at: i64,
|
||||
/// Empty for a message a user wrote; anything else is a system message.
|
||||
#[serde(default, rename = "type")]
|
||||
pub kind: String,
|
||||
}
|
||||
|
||||
impl Post {
|
||||
/// Every id is a Mattermost id (the root may be empty): they end up in session ids and paths.
|
||||
pub(crate) fn check(self) -> Result<Post, MmError> {
|
||||
// id, user_id and channel_id must be `valid_id`, and root_id empty or `valid_id`; otherwise
|
||||
// Json("a post with an invalid id: <id quoted with {:?}>").
|
||||
if !valid_id(&self.id) || !valid_id(&self.user_id) || !valid_id(&self.channel_id) {
|
||||
return Err(MmError::Json(format!(
|
||||
"a post with an invalid id: {:?}",
|
||||
self.id
|
||||
)));
|
||||
}
|
||||
if !self.root_id.is_empty() && !valid_id(&self.root_id) {
|
||||
return Err(MmError::Json(format!(
|
||||
"a post with an invalid id: {:?}",
|
||||
self.root_id
|
||||
)));
|
||||
}
|
||||
Ok(self)
|
||||
}
|
||||
}
|
||||
|
||||
/// An event from the WebSocket.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum Event {
|
||||
Hello,
|
||||
/// A new post, and the type of its channel: `D` direct, `G` group, `O` open, `P` private.
|
||||
Posted {
|
||||
post: Post,
|
||||
channel_type: String,
|
||||
},
|
||||
/// Any other event, or a reply to one of our requests, by its name (empty for a reply).
|
||||
Other(String),
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RawEvent {
|
||||
#[serde(default)]
|
||||
event: String,
|
||||
#[serde(default)]
|
||||
data: serde_json::Map<String, serde_json::Value>,
|
||||
}
|
||||
|
||||
pub(crate) fn json<T: serde::de::DeserializeOwned>(bytes: &[u8]) -> Result<T, MmError> {
|
||||
// `serde_json::from_slice`; its error becomes Json(e.to_string()).
|
||||
serde_json::from_slice(bytes).map_err(|e| MmError::Json(e.to_string()))
|
||||
}
|
||||
|
||||
/// One WebSocket text message as an event. `posted` carries the post as a JSON **string**.
|
||||
pub fn parse_event(text: &str) -> Result<Event, MmError> {
|
||||
// Parse into RawEvent. "hello" -> Hello. "posted" -> data.post must be a JSON *string*; parse
|
||||
// that string as a Post and `check` it (a missing or non-string post is Json). channel_type is
|
||||
// data.channel_type when it is a string, else "". Any other event name -> Other(name) (a reply
|
||||
// has no event: Other("")).
|
||||
let raw: RawEvent = json(text.as_bytes())?;
|
||||
match raw.event.as_str() {
|
||||
"hello" => Ok(Event::Hello),
|
||||
"posted" => {
|
||||
let post = raw
|
||||
.data
|
||||
.get("post")
|
||||
.and_then(|p| p.as_str())
|
||||
.ok_or_else(|| {
|
||||
MmError::Json("a posted event carries no post string".to_string())
|
||||
})?;
|
||||
let post: Post = json(post.as_bytes())?;
|
||||
let channel_type = raw
|
||||
.data
|
||||
.get("channel_type")
|
||||
.and_then(|c| c.as_str())
|
||||
.unwrap_or("")
|
||||
.to_string();
|
||||
Ok(Event::Posted {
|
||||
post: post.check()?,
|
||||
channel_type,
|
||||
})
|
||||
}
|
||||
other => Ok(Event::Other(other.to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
/// The WebSocket request that shows this bot as typing in a thread (`parent` is the root).
|
||||
pub fn typing(seq: u64, channel: &str, parent: &str) -> String {
|
||||
// serde_json::json!({"action": "user_typing", "seq": seq, "data": {"channel_id": channel,
|
||||
// "parent_id": parent}}) as a string.
|
||||
serde_json::json!({
|
||||
"action": "user_typing",
|
||||
"seq": seq,
|
||||
"data": {"channel_id": channel, "parent_id": parent}
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
|
||||
/// What a `posts?since` call gave back.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct Since {
|
||||
/// Posts created after the time, not deleted, oldest first.
|
||||
pub posts: Vec<Post>,
|
||||
/// Mattermost's limit was reached: some posts may be missing.
|
||||
pub full: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct PostList {
|
||||
#[serde(default)]
|
||||
order: Vec<String>,
|
||||
#[serde(default)]
|
||||
posts: std::collections::HashMap<String, Post>,
|
||||
}
|
||||
|
||||
/// The body of `GET /channels/{id}/posts?since=<since>`. Only ids in `order` changed after
|
||||
/// `since`; `posts` also holds the roots of their threads, which may be older. Edited and deleted
|
||||
/// posts come back too: only new posts count.
|
||||
pub fn since_list(body: &[u8], since: i64) -> Result<Since, MmError> {
|
||||
// Parse a PostList. full = order.len() >= SINCE_LIMIT. For each id in `order` (never the keys
|
||||
// of `posts`), skip it if it is not in `posts`; keep the post if create_at > since and
|
||||
// delete_at == 0, after `check`. Sort by (create_at, id), remove repeated ids.
|
||||
let list: PostList = json(body)?;
|
||||
let full = list.order.len() >= SINCE_LIMIT;
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
let mut posts = Vec::new();
|
||||
for id in &list.order {
|
||||
let post = match list.posts.get(id) {
|
||||
Some(p) => p.clone(),
|
||||
None => continue,
|
||||
};
|
||||
let post = post.check()?;
|
||||
if post.create_at <= since || post.delete_at != 0 {
|
||||
continue;
|
||||
}
|
||||
if seen.insert(post.id.clone()) {
|
||||
posts.push(post);
|
||||
}
|
||||
}
|
||||
posts.sort_by(|a, b| a.create_at.cmp(&b.create_at).then_with(|| a.id.cmp(&b.id)));
|
||||
Ok(Since { posts, full })
|
||||
}
|
||||
@@ -0,0 +1,194 @@
|
||||
//! Mattermost's REST calls, one connection per request, with the token in the `Authorization`
|
||||
//! header. A 429 waits as asked; a 5xx is tried twice more; a 401 or 403 is `Auth` at once.
|
||||
|
||||
use std::time::{Duration, SystemTime};
|
||||
|
||||
use serde::Deserialize;
|
||||
|
||||
use crate::config::valid_id;
|
||||
use crate::http::{self, Response, rate_limit_wait};
|
||||
use crate::mm::{Me, MmError, Post, Since, json, since_list};
|
||||
use crate::net::Connector;
|
||||
use crate::secrets::Secret;
|
||||
use crate::ws::conn::host_header;
|
||||
|
||||
/// The pause before trying a call again after a 5xx.
|
||||
pub const RETRY_5XX: Duration = Duration::from_millis(500);
|
||||
/// How many times a call is tried again after a 5xx, and waits after a 429.
|
||||
pub const RETRIES: u32 = 2;
|
||||
/// How much of an error body is kept for the message.
|
||||
const BODY_KEPT: usize = 200;
|
||||
|
||||
pub struct Client {
|
||||
connector: Connector,
|
||||
token: Secret,
|
||||
timeout: Duration,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct Channel {
|
||||
id: String,
|
||||
}
|
||||
|
||||
impl Client {
|
||||
/// `timeout` bounds connecting and each read: no bytes for that long is an error.
|
||||
pub fn new(connector: Connector, token: Secret, timeout: Duration) -> Client {
|
||||
// Store the three.
|
||||
Client {
|
||||
connector,
|
||||
token,
|
||||
timeout,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn connector(&self) -> &Connector {
|
||||
// The Connector.
|
||||
&self.connector
|
||||
}
|
||||
|
||||
pub fn token(&self) -> &Secret {
|
||||
// The Secret.
|
||||
&self.token
|
||||
}
|
||||
|
||||
fn once(&self, method: &str, path: &str, body: Option<&[u8]>) -> Result<Response, MmError> {
|
||||
// 1. `self.connector.connect(self.timeout)`, then `set_read_timeout(Some(self.timeout))`.
|
||||
let mut stream = self
|
||||
.connector
|
||||
.connect(self.timeout)
|
||||
.map_err(|e| MmError::Net(format!("{method} {path}: {e}")))?;
|
||||
stream
|
||||
.set_read_timeout(Some(self.timeout))
|
||||
.map_err(|e| MmError::Net(format!("{method} {path}: {e}")))?;
|
||||
// 2. Headers: Authorization "Bearer <token>" (the only use of `expose`), Accept
|
||||
// "application/json", and Content-Type "application/json" when there is a body. Host is
|
||||
// `host_header(self.connector.server())`.
|
||||
let auth = format!("Bearer {}", self.token.expose());
|
||||
let mut headers: Vec<(&str, &str)> = vec![
|
||||
("Authorization", auth.as_str()),
|
||||
("Accept", "application/json"),
|
||||
];
|
||||
if body.is_some() {
|
||||
headers.push(("Content-Type", "application/json"));
|
||||
}
|
||||
// 3. `http::request`. Every error on the way is Net("<method> <path>: <error>").
|
||||
http::request(
|
||||
&mut stream,
|
||||
method,
|
||||
&host_header(self.connector.server()),
|
||||
path,
|
||||
&headers,
|
||||
body,
|
||||
)
|
||||
.map_err(|e| MmError::Net(format!("{method} {path}: {e}")))
|
||||
}
|
||||
|
||||
/// One call, tried again as the module comment says; the body of a 2xx answer.
|
||||
fn call(&self, method: &str, path: &str, body: Option<&[u8]>) -> Result<Vec<u8>, MmError> {
|
||||
// Loop over `once`: 2xx -> the body. 401 or 403 -> Auth(status) at once. 429 -> after
|
||||
// RETRIES waits already, RateLimited(wait); else sleep `rate_limit_wait(&head,
|
||||
// SystemTime::now())` and try again. 5xx -> up to RETRIES more tries, sleeping RETRY_5XX
|
||||
// before each. Anything else -> Status(status, the first BODY_KEPT characters of the body,
|
||||
// lossy UTF-8).
|
||||
let mut tries: u32 = 0;
|
||||
loop {
|
||||
let answer = self.once(method, path, body)?;
|
||||
let status = answer.head.status;
|
||||
if (200..300).contains(&status) {
|
||||
return Ok(answer.body);
|
||||
}
|
||||
match status {
|
||||
401 | 403 => return Err(MmError::Auth(status)),
|
||||
429 => {
|
||||
let wait = rate_limit_wait(&answer.head, SystemTime::now());
|
||||
if tries < RETRIES {
|
||||
tries += 1;
|
||||
std::thread::sleep(wait);
|
||||
continue;
|
||||
}
|
||||
return Err(MmError::RateLimited(wait));
|
||||
}
|
||||
_ if status >= 500 => {
|
||||
if tries < RETRIES {
|
||||
tries += 1;
|
||||
std::thread::sleep(RETRY_5XX);
|
||||
continue;
|
||||
}
|
||||
return Err(status_error(status, &answer.body));
|
||||
}
|
||||
_ => return Err(status_error(status, &answer.body)),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// `GET /api/v4/users/me`: who this token is.
|
||||
pub fn me(&self) -> Result<Me, MmError> {
|
||||
// GET /api/v4/users/me into Me; an id that is not `valid_id` or an empty username is Json.
|
||||
let body = self.call("GET", "/api/v4/users/me", None)?;
|
||||
let me: Me = json(&body)?;
|
||||
if !valid_id(&me.id) || me.username.is_empty() {
|
||||
return Err(MmError::Json(
|
||||
"the user me has no id or no username".to_string(),
|
||||
));
|
||||
}
|
||||
Ok(me)
|
||||
}
|
||||
|
||||
/// `POST /api/v4/posts`: a post in `channel`, in the thread of `root` (empty: top level).
|
||||
pub fn create_post(&self, channel: &str, root: &str, message: &str) -> Result<Post, MmError> {
|
||||
// POST /api/v4/posts with {"channel_id", "root_id", "message"}; the answer is a Post,
|
||||
// `check`ed.
|
||||
let payload = serde_json::json!({
|
||||
"channel_id": channel,
|
||||
"root_id": root,
|
||||
"message": message
|
||||
})
|
||||
.to_string();
|
||||
let body = self.call("POST", "/api/v4/posts", Some(payload.as_bytes()))?;
|
||||
let post: Post = json(&body)?;
|
||||
post.check()
|
||||
}
|
||||
|
||||
/// `GET /api/v4/channels/{channel}/posts?since=<ms>`: the posts created after `since`.
|
||||
pub fn posts_since(&self, channel: &str, since: i64) -> Result<Since, MmError> {
|
||||
// A channel that is not `valid_id` is Json, and nothing is sent. Otherwise GET
|
||||
// /api/v4/channels/<channel>/posts?since=<since>, through `since_list`.
|
||||
if !valid_id(channel) {
|
||||
return Err(MmError::Json(
|
||||
"a channel path must be a Mattermost id".to_string(),
|
||||
));
|
||||
}
|
||||
let body = self.call(
|
||||
"GET",
|
||||
&format!("/api/v4/channels/{channel}/posts?since={since}"),
|
||||
None,
|
||||
)?;
|
||||
since_list(&body, since)
|
||||
}
|
||||
|
||||
/// `POST /api/v4/channels/direct`: the id of the direct channel between two users.
|
||||
pub fn direct_channel(&self, a: &str, b: &str) -> Result<String, MmError> {
|
||||
// POST /api/v4/channels/direct with the JSON array [a, b]; the answer has an "id", which
|
||||
// must be `valid_id` (else Json).
|
||||
let payload = serde_json::to_vec(&serde_json::json!([a, b]))
|
||||
.map_err(|e| MmError::Json(e.to_string()))?;
|
||||
let body = self.call("POST", "/api/v4/channels/direct", Some(payload.as_slice()))?;
|
||||
let channel: Channel = json(&body)?;
|
||||
if !valid_id(&channel.id) {
|
||||
return Err(MmError::Json("the direct channel has no id".to_string()));
|
||||
}
|
||||
Ok(channel.id)
|
||||
}
|
||||
}
|
||||
|
||||
fn status_error(status: u16, body: &[u8]) -> MmError {
|
||||
// Any non-2xx, non-4xx-that-isn't-401/403/429/5xx: the status and the first `BODY_KEPT`
|
||||
// characters, lossy UTF-8.
|
||||
MmError::Status(
|
||||
status,
|
||||
String::from_utf8_lossy(body)
|
||||
.chars()
|
||||
.take(BODY_KEPT)
|
||||
.collect(),
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,173 @@
|
||||
//! Mattermost's JSON: events as the server sends them (v11.11.0 shapes), the typing request, and
|
||||
//! the `posts?since` list, whose edited, deleted and root-only posts must not be answered (M4a spec,
|
||||
//! sections 7 and 9). Do not edit.
|
||||
|
||||
use gatewayd::mm::{Event, MmError, SINCE_LIMIT, parse_event, since_list, typing};
|
||||
use serde_json::json;
|
||||
|
||||
const KYLE: &str = "k0000000000000000000000000";
|
||||
const DM: &str = "d0000000000000000000000000";
|
||||
|
||||
fn post(id: &str, create_at: i64) -> serde_json::Value {
|
||||
json!({
|
||||
"id": id, "create_at": create_at, "update_at": create_at, "edit_at": 0, "delete_at": 0,
|
||||
"is_pinned": false, "user_id": KYLE, "channel_id": DM, "root_id": "", "original_id": "",
|
||||
"message": "hello", "type": "", "props": {"from_bot": "true"}, "hashtags": "",
|
||||
"pending_post_id": "", "reply_count": 0, "metadata": {}
|
||||
})
|
||||
}
|
||||
|
||||
fn id(n: u32) -> String {
|
||||
format!("p{n:025}")
|
||||
}
|
||||
|
||||
fn posted(post: &serde_json::Value, channel_type: &str) -> String {
|
||||
json!({
|
||||
"event": "posted",
|
||||
"data": {
|
||||
"channel_display_name": "@kyle", "channel_name": "x__y", "channel_type": channel_type,
|
||||
"post": post.to_string(), "sender_name": "@kyle", "set_online": true, "team_id": ""
|
||||
},
|
||||
"broadcast": {"omit_users": null, "user_id": "", "channel_id": DM, "team_id": ""},
|
||||
"seq": 3
|
||||
})
|
||||
.to_string()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn hello_replies_and_other_events() {
|
||||
let hello =
|
||||
json!({"event": "hello", "data": {"server_version": "11.11.0"}, "broadcast": {}, "seq": 0});
|
||||
assert_eq!(parse_event(&hello.to_string()).unwrap(), Event::Hello);
|
||||
assert_eq!(
|
||||
parse_event(r#"{"status":"OK","seq_reply":1}"#).unwrap(),
|
||||
Event::Other(String::new())
|
||||
);
|
||||
let typing = json!({"event": "typing", "data": {"parent_id": ""}, "broadcast": {}, "seq": 4});
|
||||
assert_eq!(
|
||||
parse_event(&typing.to_string()).unwrap(),
|
||||
Event::Other("typing".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_posted_event_carries_the_post_as_a_string() {
|
||||
let mut p = post(&id(1), 1_758_650_000_123);
|
||||
p["root_id"] = json!(id(0));
|
||||
p["message"] = json!("line one\nline two");
|
||||
let Event::Posted { post, channel_type } = parse_event(&posted(&p, "D")).unwrap() else {
|
||||
panic!("not a post")
|
||||
};
|
||||
assert_eq!(channel_type, "D");
|
||||
assert_eq!(
|
||||
(post.id.as_str(), post.root_id.as_str()),
|
||||
(id(1).as_str(), id(0).as_str())
|
||||
);
|
||||
assert_eq!(
|
||||
(post.user_id.as_str(), post.channel_id.as_str()),
|
||||
(KYLE, DM)
|
||||
);
|
||||
assert_eq!(
|
||||
(post.message.as_str(), post.create_at, post.kind.as_str()),
|
||||
("line one\nline two", 1_758_650_000_123, "")
|
||||
);
|
||||
let mut s = self::post(&id(2), 5);
|
||||
s["type"] = json!("system_join_channel");
|
||||
let Event::Posted { post, .. } = parse_event(&posted(&s, "O")).unwrap() else {
|
||||
panic!()
|
||||
};
|
||||
assert_eq!(post.kind, "system_join_channel");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn bad_posts_are_errors_not_panics() {
|
||||
let p = post(&id(1), 5);
|
||||
let as_object =
|
||||
json!({"event": "posted", "data": {"post": p, "channel_type": "D"}}).to_string();
|
||||
let mut cases = vec![
|
||||
as_object,
|
||||
"not json".to_string(),
|
||||
"[1,2]".to_string(),
|
||||
"{\"event\":5}".to_string(),
|
||||
];
|
||||
for (field, value) in [
|
||||
("id", json!("../../etc")),
|
||||
("user_id", json!("")),
|
||||
("channel_id", json!("A0000000000000000000000000")),
|
||||
("root_id", json!("short")),
|
||||
("create_at", json!("soon")),
|
||||
] {
|
||||
let mut bad = post(&id(1), 5);
|
||||
bad[field] = value;
|
||||
cases.push(posted(&bad, "D"));
|
||||
}
|
||||
let mut missing = post(&id(1), 5);
|
||||
missing.as_object_mut().unwrap().remove("create_at");
|
||||
cases.push(posted(&missing, "D"));
|
||||
for case in cases {
|
||||
assert!(
|
||||
matches!(parse_event(&case), Err(MmError::Json(_))),
|
||||
"{case}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_typing_request() {
|
||||
let got: serde_json::Value = serde_json::from_str(&typing(7, DM, &id(0))).unwrap();
|
||||
assert_eq!(
|
||||
got,
|
||||
json!({"action": "user_typing", "seq": 7, "data": {"channel_id": DM, "parent_id": id(0)}})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn since_keeps_new_posts_in_order_and_nothing_else() {
|
||||
let since = 1000;
|
||||
let mut edited = post(&id(1), 900);
|
||||
edited["update_at"] = json!(1500);
|
||||
let mut deleted = post(&id(2), 1100);
|
||||
deleted["delete_at"] = json!(1200);
|
||||
let root_only = post(&id(3), 10);
|
||||
let later = post(&id(4), 1300);
|
||||
let earlier = post(&id(5), 1200);
|
||||
let at_since = post(&id(6), 1000);
|
||||
let mut posts = serde_json::Map::new();
|
||||
for p in [&edited, &deleted, &root_only, &later, &earlier, &at_since] {
|
||||
posts.insert(p["id"].as_str().unwrap().to_string(), p.clone());
|
||||
}
|
||||
let order = json!([id(4), id(2), id(1), id(5), id(9), id(6)]);
|
||||
let body = json!({"order": order, "posts": posts, "next_post_id": "", "prev_post_id": "", "has_next": false});
|
||||
let got = since_list(body.to_string().as_bytes(), since).unwrap();
|
||||
let ids: Vec<&str> = got.posts.iter().map(|p| p.id.as_str()).collect();
|
||||
assert_eq!(ids, [id(5), id(4)]);
|
||||
assert!(!got.full);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_full_since_answer_says_so() {
|
||||
let mut posts = serde_json::Map::new();
|
||||
let mut order = Vec::new();
|
||||
for n in 0..SINCE_LIMIT {
|
||||
let n = u32::try_from(n).unwrap();
|
||||
posts.insert(id(n), post(&id(n), 2000 + i64::from(n)));
|
||||
order.push(id(n));
|
||||
}
|
||||
let body = json!({"order": order, "posts": posts});
|
||||
let got = since_list(body.to_string().as_bytes(), 1000).unwrap();
|
||||
assert!(got.full);
|
||||
assert_eq!(got.posts.len(), SINCE_LIMIT);
|
||||
let empty = since_list(br#"{"order":[],"posts":{}}"#, 0).unwrap();
|
||||
assert!(empty.posts.is_empty() && !empty.full);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_server_error_body_is_quoted_in_messages() {
|
||||
let e = MmError::Status(500, "boom\ngatewayd: a forged line".to_string());
|
||||
let text = e.to_string();
|
||||
assert!(!text.contains('\n'), "{text}");
|
||||
assert_eq!(
|
||||
MmError::Auth(401).to_string(),
|
||||
"Mattermost refused the token (401)"
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,242 @@
|
||||
//! Mattermost's REST calls against a scripted server: the requests on the wire, and what a 401,
|
||||
//! 403, 429, 5xx or 404 does (M4a spec, sections 5 and 9). Do not edit.
|
||||
|
||||
#[path = "support/http_server.rs"]
|
||||
mod http_server;
|
||||
#[path = "support/tls_server.rs"]
|
||||
mod tls_server;
|
||||
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::mpsc::Receiver;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use gatewayd::config::ServerUrl;
|
||||
use gatewayd::mm::MmError;
|
||||
use gatewayd::mm::rest::Client;
|
||||
use gatewayd::net::Connector;
|
||||
use gatewayd::secrets::Secret;
|
||||
use http_server::{Request, reply, serve_http};
|
||||
use serde_json::json;
|
||||
use tls_server::{fixture, server_config};
|
||||
|
||||
const BOT: &str = "b0000000000000000000000000";
|
||||
const KYLE: &str = "k0000000000000000000000000";
|
||||
const DM: &str = "d0000000000000000000000000";
|
||||
const POST: &str = "p0000000000000000000000001";
|
||||
|
||||
fn client(addr: SocketAddr, tls: bool) -> Client {
|
||||
let url = ServerUrl {
|
||||
tls,
|
||||
host: "localhost".to_string(),
|
||||
port: addr.port(),
|
||||
};
|
||||
let ca = tls.then(|| fixture("test-ca.pem"));
|
||||
let connector = Connector::new(url, ca.as_deref()).unwrap();
|
||||
Client::new(
|
||||
connector,
|
||||
Secret::new("TOKEN".to_string()),
|
||||
Duration::from_secs(5),
|
||||
)
|
||||
}
|
||||
|
||||
fn me_body() -> String {
|
||||
json!({"id": BOT, "username": "boxmaker-straylight", "roles": "system_user", "is_bot": true})
|
||||
.to_string()
|
||||
}
|
||||
|
||||
fn requests(rx: &Receiver<Request>) -> Vec<Request> {
|
||||
rx.try_iter().collect()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn me_plain_and_over_tls() {
|
||||
for tls in [false, true] {
|
||||
let config = tls.then(|| server_config("server"));
|
||||
let (addr, rx) = serve_http(config, |_, _| reply(200, "", &me_body()));
|
||||
let me = client(addr, tls).me().unwrap();
|
||||
assert_eq!(
|
||||
(me.id.as_str(), me.username.as_str()),
|
||||
(BOT, "boxmaker-straylight")
|
||||
);
|
||||
let r = rx.recv_timeout(Duration::from_secs(5)).unwrap();
|
||||
assert_eq!(
|
||||
(r.method.as_str(), r.path.as_str()),
|
||||
("GET", "/api/v4/users/me")
|
||||
);
|
||||
assert!(
|
||||
r.head.contains("Authorization: Bearer TOKEN\r\n"),
|
||||
"{}",
|
||||
r.head
|
||||
);
|
||||
assert!(
|
||||
r.head
|
||||
.contains(&format!("Host: localhost:{}\r\n", addr.port())),
|
||||
"{}",
|
||||
r.head
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn create_post_sends_the_thread_and_reads_the_post_back() {
|
||||
let (addr, rx) = serve_http(None, |_, r| {
|
||||
let mut p = r.json();
|
||||
p["id"] = json!(POST);
|
||||
p["user_id"] = json!(BOT);
|
||||
p["create_at"] = json!(5);
|
||||
reply(201, "", &p.to_string())
|
||||
});
|
||||
let post = client(addr, false)
|
||||
.create_post(DM, KYLE, "an answer\nin two lines")
|
||||
.unwrap();
|
||||
assert_eq!((post.id.as_str(), post.root_id.as_str()), (POST, KYLE));
|
||||
let r = rx.recv_timeout(Duration::from_secs(5)).unwrap();
|
||||
assert_eq!(
|
||||
(r.method.as_str(), r.path.as_str()),
|
||||
("POST", "/api/v4/posts")
|
||||
);
|
||||
assert!(
|
||||
r.head.contains("Content-Type: application/json\r\n"),
|
||||
"{}",
|
||||
r.head
|
||||
);
|
||||
assert_eq!(
|
||||
r.json(),
|
||||
json!({"channel_id": DM, "root_id": KYLE, "message": "an answer\nin two lines"})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn posts_since_and_the_direct_channel() {
|
||||
let (addr, rx) = serve_http(None, |_, r| {
|
||||
if r.path.contains("/posts?since=") {
|
||||
reply(200, "", r#"{"order":[],"posts":{}}"#)
|
||||
} else {
|
||||
reply(201, "", &json!({"id": DM, "type": "D"}).to_string())
|
||||
}
|
||||
});
|
||||
let c = client(addr, false);
|
||||
assert!(
|
||||
c.posts_since(DM, 1_758_650_000_000)
|
||||
.unwrap()
|
||||
.posts
|
||||
.is_empty()
|
||||
);
|
||||
assert_eq!(c.direct_channel(BOT, KYLE).unwrap(), DM);
|
||||
let rs = requests(&rx);
|
||||
assert_eq!(
|
||||
rs[0].path,
|
||||
format!("/api/v4/channels/{DM}/posts?since=1758650000000")
|
||||
);
|
||||
assert_eq!(
|
||||
(rs[1].method.as_str(), rs[1].path.as_str()),
|
||||
("POST", "/api/v4/channels/direct")
|
||||
);
|
||||
assert_eq!(rs[1].json(), json!([BOT, KYLE]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_channel_that_is_not_an_id_is_never_sent() {
|
||||
let (addr, rx) = serve_http(None, |_, _| reply(200, "", r#"{"order":[],"posts":{}}"#));
|
||||
let err = client(addr, false)
|
||||
.posts_since("../users/me?x=", 0)
|
||||
.unwrap_err();
|
||||
assert!(matches!(err, MmError::Json(_)), "{err}");
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
assert!(requests(&rx).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_refused_token_stops_at_once() {
|
||||
for status in [401, 403] {
|
||||
let (addr, rx) = serve_http(None, move |_, _| {
|
||||
reply(
|
||||
status,
|
||||
"",
|
||||
r#"{"id":"api.context.session_expired.app_error"}"#,
|
||||
)
|
||||
});
|
||||
let err = client(addr, false).me().unwrap_err();
|
||||
assert!(matches!(err, MmError::Auth(s) if s == status), "{err}");
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
assert_eq!(requests(&rx).len(), 1, "no retry on {status}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_server_error_is_tried_twice_more() {
|
||||
let (addr, rx) = serve_http(None, |n, _| {
|
||||
if n < 2 {
|
||||
reply(503, "", "{}")
|
||||
} else {
|
||||
reply(200, "", &me_body())
|
||||
}
|
||||
});
|
||||
assert!(client(addr, false).me().is_ok());
|
||||
assert_eq!(requests(&rx).len(), 3);
|
||||
|
||||
let (addr, rx) = serve_http(None, |_, _| reply(502, "", "{}"));
|
||||
let err = client(addr, false).me().unwrap_err();
|
||||
assert!(matches!(err, MmError::Status(502, _)), "{err}");
|
||||
std::thread::sleep(Duration::from_millis(100));
|
||||
assert_eq!(requests(&rx).len(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_rate_limit_is_waited_out() {
|
||||
let (addr, rx) = serve_http(None, |n, _| {
|
||||
if n == 0 {
|
||||
reply(429, "X-Ratelimit-Reset: 1\r\n", "{}")
|
||||
} else {
|
||||
reply(200, "", &me_body())
|
||||
}
|
||||
});
|
||||
let started = Instant::now();
|
||||
assert!(client(addr, false).me().is_ok());
|
||||
assert!(
|
||||
started.elapsed() >= Duration::from_millis(900),
|
||||
"{:?}",
|
||||
started.elapsed()
|
||||
);
|
||||
assert_eq!(requests(&rx).len(), 2);
|
||||
|
||||
let (addr, _rx) = serve_http(None, |_, _| reply(429, "X-Ratelimit-Reset: 1\r\n", "{}"));
|
||||
assert!(matches!(
|
||||
client(addr, false).me(),
|
||||
Err(MmError::RateLimited(_))
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn other_statuses_keep_the_start_of_the_body() {
|
||||
let long = format!("{{\"message\":\"{}\"}}", "x".repeat(1000));
|
||||
let (addr, _rx) = serve_http(None, move |_, _| reply(404, "", &long));
|
||||
let Err(MmError::Status(404, body)) = client(addr, false).me() else {
|
||||
panic!("not a 404")
|
||||
};
|
||||
assert_eq!(body.chars().count(), 200);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn answers_that_are_not_what_we_expect() {
|
||||
for body in [
|
||||
"not json",
|
||||
r#"{"id":"short","username":"x"}"#,
|
||||
r#"{"id":"b0000000000000000000000000","username":""}"#,
|
||||
] {
|
||||
let (addr, _rx) = serve_http(None, move |_, _| reply(200, "", body));
|
||||
assert!(
|
||||
matches!(client(addr, false).me(), Err(MmError::Json(_))),
|
||||
"{body}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn nobody_listening_is_a_network_error() {
|
||||
let port = std::net::TcpListener::bind("127.0.0.1:0")
|
||||
.unwrap()
|
||||
.local_addr()
|
||||
.unwrap();
|
||||
assert!(matches!(client(port, false).me(), Err(MmError::Net(_))));
|
||||
}
|
||||
@@ -0,0 +1,93 @@
|
||||
//! A scripted HTTP server for tests: it records each request and answers with what the test's
|
||||
//! function returns for it. Built on `tls_server`. Do not edit.
|
||||
|
||||
#![allow(dead_code)] // each test file uses a different part of this module
|
||||
|
||||
use std::io::{Read, Write};
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex, mpsc};
|
||||
|
||||
use rustls::ServerConfig;
|
||||
|
||||
use crate::tls_server::serve;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Request {
|
||||
pub method: String,
|
||||
pub path: String,
|
||||
/// The whole head, for tests that look for a header.
|
||||
pub head: String,
|
||||
pub body: Vec<u8>,
|
||||
}
|
||||
|
||||
impl Request {
|
||||
pub fn json(&self) -> serde_json::Value {
|
||||
serde_json::from_slice(&self.body).unwrap()
|
||||
}
|
||||
}
|
||||
|
||||
/// A response with a JSON body.
|
||||
pub fn reply(status: u16, extra_headers: &str, body: &str) -> String {
|
||||
format!(
|
||||
"HTTP/1.1 {status} X\r\nContent-Type: application/json\r\n{extra_headers}Content-Length: {}\r\n\r\n{body}",
|
||||
body.len()
|
||||
)
|
||||
}
|
||||
|
||||
fn read_request(conn: &mut dyn Read) -> Option<Request> {
|
||||
let mut head = Vec::new();
|
||||
let mut byte = [0u8; 1];
|
||||
while !head.ends_with(b"\r\n\r\n") {
|
||||
if conn.read(&mut byte).ok()? == 0 {
|
||||
return None;
|
||||
}
|
||||
head.push(byte[0]);
|
||||
}
|
||||
let head = String::from_utf8(head).ok()?;
|
||||
let mut first = head.split_whitespace();
|
||||
let method = first.next()?.to_string();
|
||||
let path = first.next()?.to_string();
|
||||
let length = head
|
||||
.lines()
|
||||
.find_map(|l| {
|
||||
l.to_ascii_lowercase()
|
||||
.strip_prefix("content-length:")
|
||||
.map(|v| v.trim().parse().ok())
|
||||
})
|
||||
.flatten()
|
||||
.unwrap_or(0);
|
||||
let mut body = vec![0u8; length];
|
||||
conn.read_exact(&mut body).ok()?;
|
||||
Some(Request {
|
||||
method,
|
||||
path,
|
||||
head,
|
||||
body,
|
||||
})
|
||||
}
|
||||
|
||||
/// Serve requests: `answer(n, &request)` gives the response to the n-th request (from 0). Every
|
||||
/// request is sent on the returned channel.
|
||||
pub fn serve_http<F>(
|
||||
tls: Option<Arc<ServerConfig>>,
|
||||
answer: F,
|
||||
) -> (SocketAddr, mpsc::Receiver<Request>)
|
||||
where
|
||||
F: Fn(usize, &Request) -> String + Send + Sync + 'static,
|
||||
{
|
||||
let (tx, rx) = mpsc::channel();
|
||||
let tx = Mutex::new(tx);
|
||||
let count = AtomicUsize::new(0);
|
||||
let addr = serve(tls, move |mut conn| {
|
||||
let Some(request) = read_request(&mut conn) else {
|
||||
return;
|
||||
};
|
||||
let n = count.fetch_add(1, Ordering::SeqCst);
|
||||
let response = answer(n, &request);
|
||||
let _ = tx.lock().unwrap().send(request);
|
||||
let _ = conn.write_all(response.as_bytes());
|
||||
let _ = conn.flush();
|
||||
});
|
||||
(addr, rx)
|
||||
}
|
||||
@@ -6,6 +6,7 @@ reviewer adds findings under "Reviews" once per milestone.
|
||||
|
||||
| Task | Date | Status | Gate runs | First gate | Deviations | Notes | Model |
|
||||
|---|---|---|---|---|---|---|---|
|
||||
| M4a/10-gatewayd-mm | 2026-09-23 | done | 1 | pass | none | Copied `tests/mm_json.rs`, `tests/mm_rest.rs` and `tests/support/http_server.rs`, added the `src/mm/mod.rs` and `src/mm/rest.rs` skeletons to `crates/gatewayd/src/mm/` and `pub mod mm;` to `lib.rs` (before `secrets`, alphabetical). Filled `mod.rs`: `Post::check` requires `id`/`user_id`/`channel_id` `valid_id` and `root_id` empty-or-`valid_id`, else `Json` quoting the offending id with `{:?}`; `json` is `serde_json::from_slice` mapped to `Json(e.to_string())`; `parse_event` matches `hello`/`posted`/other — `posted` takes `data.post` as a JSON *string* (an object or missing is `Json`), parses it, `check`s it, and reads `data.channel_type` (else ""), any other name (or an empty-name reply) is `Other(name)`; `typing` is `serde_json::json!` compacted; `since_list` walks `order` only (skipping ids not in `posts`, keeping `create_at > since && delete_at == 0` after `check`, deduping, then sorting by `(create_at, id)`), `full` when `order.len() >= SINCE_LIMIT`. Filled `rest.rs`: `Client::new` stores the three fields; `once` connects within `timeout`, sets the read timeout, sends `Authorization: Bearer <token>` (the only `expose`), `Accept`/`Content-Type` headers and `host_header`, mapping every error to `Net("<method> <path>: <e>")`; `call` loops `once` — 2xx returns the body, 401/403 `Auth`, 429 waits `rate_limit_wait` up to `RETRIES` then `RateLimited`, 5xx retried up to `RETRIES` times sleeping `RETRY_5XX`, else `Status` with the first `BODY_KEPT` lossy-UTF-8 chars via a `status_error` helper; `me`/`create_post`/`posts_since`/`direct_channel` build the four calls, `posts_since` and `me`/`direct_channel` reject non-`valid_id` ids as `Json` before sending. All 7 `mm_json` and 10 `mm_rest` tests pass (the latter ~3 s on two deliberate rate-limit waits); `make gate` prints `gate: ok` first run. | ? |
|
||||
| M4a/09-gatewayd-ws-conn | 2026-09-23 | done | 1 | pass | none | Filled the eight functions in the copied `crates/gatewayd/src/ws/conn.rs` skeleton (the written `poll` was the glue). `open`: `connector.connect(dead_after)` mapped to `Handshake(e.to_string())`, then `handshake` with `host_header(connector.server())`, a Ws with a new `Decoder` and `last_heard`/`last_ping` both `now`. `send`: `read_exact` 4 mask bytes from `random`, then `encode(opcode, payload, mask)` written and flushed. `send_text`: `send(TEXT, text.as_bytes())`. `take_messages`: loop `next_message`, `Text` returns, `Ping` answered with `send(PONG, &payload)`, `Pong` ignored, `Close` replies the code as 2 big-endian bytes (empty when none) via a best-effort `send(CLOSE, ...)` (the peer may be gone) and returns `Closed`. `keep_alive`: `now.duration_since(last_heard) >= dead_after` is `Dead`, else `now.duration_since(last_ping) >= ping_every` pings and stamps `last_ping`. `read_timeout`: the least of next-ping, next-dead and until-left (each `saturating_duration_since`), then `.max(1ms)`. `read_some`: `set_read_timeout`, a 16 KiB buffer, `Ok(0)` -> `Closed`, `Ok(n)` feeds `buf.get(..n).unwrap_or_default()` and stamps `last_heard`, `WouldBlock`/`TimedOut`/`Interrupted` -> `Ok(())`, any other `Err` -> `Io`. `close`: best-effort `send(CLOSE, &1000u16.to_be_bytes())`. `host_header`: host alone when the port is the scheme default (443 for tls, 80 otherwise) else `host:port`. All 10 tests in `tests/ws_conn.rs` pass five runs under a second; `make gate` prints `gate: ok` first run. | ? |
|
||||
| M4a/08-gatewayd-ws-frames | 2026-09-23 | done | 1 | pass | none | Copied `tests/ws_frame.rs` and the `src/ws/frame.rs` skeleton, added `pub mod frame;` (before `handshake`, alphabetical). Filled the seven functions the comments specified verbatim: `check_first_bytes` (reserved bits `b0 & 0x70`, mask `b1 & 0x80`, opcode `matches!(b0 & 0x0F, CONTINUATION | TEXT | CLOSE | PING | PONG)`); `check_control` (not fin, then >125); `length` (match on `short`: 0..=125, 126 reading `buf.get(2..4)` into `u16::from_be_bytes` with `len < 126` refused, 127 reading `buf.get(2..10)` into `u64::from_be_bytes` with top-bit and `<= 0xFFFF` refused, `usize::try_from(len).unwrap_or(usize::MAX)`); `check_data` (TEXT while partial, CONTINUATION while none, `payload_len > MAX_MESSAGE.saturating_sub(so_far)`); `data_frame` (empty Vec for TEXT else `partial.take().unwrap_or_default()`, append, defer when not fin, `String::from_utf8` at fin); `close` (slice-pattern match, `u16::from_be_bytes` code, UTF-8 reason); `encode` (FIN+opcode, three length branches with mask bit, XOR with `mask.iter().cycle()`). All string literals got `.to_string()` for the `Protocol(String)` variant, matching http.rs. `length`'s match needed a defensive `_` arm (`128..=u8::MAX` unreachable since `short = b1 & 0x7F`) so the codec stays exhaustive without a panic. All 7 tests in `tests/ws_frame.rs` pass including the 300-seed property test against the naive decoder; `make gate` prints `gate: ok` first run. | ? |
|
||||
| M4a/07-gatewayd-ws-handshake | 2026-09-23 | done | 1 | pass | none | Filled the copied `crates/gatewayd/src/ws/handshake.rs` skeleton. `base64`: 3-byte chunks to 4 chars over ALPHABET with `=` padding, reading each byte via `first`/`get(..).copied().unwrap_or(0)` (no indexing) and masking to 0..=63 before the alphabet index; `accept_for`: `base64(sha1(key + GUID))` building `key+GUID` into one Vec; `new_key`: `read_exact` 16 bytes (too few is an io error) then base64; `check_response` in the task's order — status 101 (`"status <n>"`), `Upgrade` == `websocket` (ASCII case-insensitive), `Connection` with a comma-split token == `upgrade` (case-insensitive), then `Sec-WebSocket-Accept` exactly == `accept_for(key)`; `handshake`: `new_key`, write `request_text` + flush, `read_head` (mapped to `Handshake`), `check_response`, reading nothing past the head. All 7 tests in `tests/ws_handshake.rs` pass including the RFC 6455 accept vector and the first-frame-left-unread handshake; `make gate` prints `gate: ok` first run. | ? |
|
||||
|
||||
Reference in New Issue
Block a user