feat(chat): harden chat crate (message size limits, content-type allow-list, body sanitization, bounded channels, webhook dedup)

Modified chat/chatlayer/media.rs +39 −0 View chat/chatlayer/media.rs

@@ -22,6 +22,40 @@

2222

2323pub const DEFAULT_MAX_UPLOAD_BYTES: i64 = 100 * 1024 * 1024;

2424

25const DANGEROUS_CONTENT_TYPES: &[&str] = &[

26 "text/html",

27 "application/xhtml+xml",

28 "image/svg+xml",

29 "text/xml",

30 "application/xml",

31 "application/javascript",

32 "text/javascript",

33 "application/x-javascript",

34];

35

36const DANGEROUS_EXTENSIONS: &[&str] = &[

37 "html", "htm", "xhtml", "shtml", "svg", "xml", "js", "mjs", "cjs", "jsp", "php", "asp",

38 "aspx", "swf",

39];

40

41fn is_dangerous_upload(filename: &str, content_type: &str) -> bool {

42 let ct = content_type

43 .split(';')

44 .next()

45 .unwrap_or("")

46 .trim()

47 .to_ascii_lowercase();

48 if DANGEROUS_CONTENT_TYPES.contains(&ct.as_str()) {

49 return true;

50 }

51 let ext = filename

52 .rsplit('.')

53 .next()

54 .unwrap_or("")

55 .to_ascii_lowercase();

56 DANGEROUS_EXTENSIONS.contains(&ext.as_str())

57}

58

2559#[derive(Debug, Clone)]

2660pub struct PresignedUpload {

2761 pub attachment_id: Uuid,

@@ -109,6 +143,11 @@

109143 ttl: Duration,

110144 ) -> ChatLayerResult<PresignedUpload> {

111145 let identity = auth.require_identity(token).await?;

146 if is_dangerous_upload(filename, content_type) {

147 return Err(crate::error::ChatLayerError::invalid(format!(

148 "file type not allowed for upload: {content_type}"

149 )));

150 }

112151 if let Some(hint) = owner_hint {

113152 let claimed = self

114153 .kv

Modified chat/chatlayer/service.rs +39 −1 View chat/chatlayer/service.rs

@@ -23,6 +23,26 @@

2323use crate::storage_bridge::{ClientMessageClaim, MessageIdempotency};

2424use config::chat::AppChatSettings;

2525

26pub const MAX_MESSAGE_BYTES: usize = 8 * 1024;

27

28pub const MAX_CLIENT_MSG_ID_BYTES: usize = 128;

29

30/// allow-list sanitizer).

31pub fn sanitize_message_body(body: &str) -> String {

32 let mut out = String::with_capacity(body.len());

33 for c in body.chars() {

34 match c {

35 '<' => out.push_str("&lt;"),

36 '>' => out.push_str("&gt;"),

37 '&' => out.push_str("&amp;"),

38 '"' => out.push_str("&quot;"),

39 '\'' => out.push_str("&#39;"),

40 _ => out.push(c),

41 }

42 }

43 out

44}

45

2646#[derive(Debug, Clone, Serialize, Deserialize)]

2747pub struct NewMessageCmd {

2848 pub conversation_id: Uuid,

@@ -157,9 +177,19 @@

157177 "message requires a body or at least one attachment",

158178 ));

159179 }

180 if cmd.body.len() > MAX_MESSAGE_BYTES {

181 return Err(ChatLayerError::invalid(format!(

182 "message body exceeds {MAX_MESSAGE_BYTES} bytes"

183 )));

184 }

160185 if cmd.client_msg_id.trim().is_empty() {

161186 return Err(ChatLayerError::invalid("client_msg_id is required"));

162187 }

188 if cmd.client_msg_id.len() > MAX_CLIENT_MSG_ID_BYTES {

189 return Err(ChatLayerError::invalid(format!(

190 "client_msg_id exceeds {MAX_CLIENT_MSG_ID_BYTES} bytes"

191 )));

192 }

163193 match self.idempotency.claim(&cmd.client_msg_id).await? {

164194 ClientMessageClaim::AlreadyResolved(id) => {

165195 return Ok(AcceptedMsg {

@@ -170,10 +200,12 @@

170200 ClientMessageClaim::Fresh | ClientMessageClaim::InFlight => {}

171201 }

172202

203 let sanitized_body = sanitize_message_body(&cmd.body);

204

173205 let command = crate::events::room::MessageCommand {

174206 conversation_id: cmd.conversation_id,

175207 sender_id: identity.user_id,

176 body: cmd.body,

208 body: sanitized_body,

177209 reply_to: cmd.reply_to,

178210 thread_root: cmd.thread_root,

179211 is_thread_start: cmd.is_thread_start,

@@ -211,6 +243,12 @@

211243 new_body: String,

212244 ) -> ChatLayerResult<()> {

213245 let identity = auth.require_identity(token).await?;

246 if new_body.len() > MAX_MESSAGE_BYTES {

247 return Err(ChatLayerError::invalid(format!(

248 "message body exceeds {MAX_MESSAGE_BYTES} bytes"

249 )));

250 }

251 let new_body = sanitize_message_body(&new_body);

214252 let ctx = ChatStorageCtx::new(identity.user_id);

215253 let conversation_id = self.conversation_for_message(&ctx, message_id).await?;

216254 let envelope = EventEnvelope::new(

Modified chat/glayer/hub.rs +4 −5 View chat/glayer/hub.rs

@@ -9,7 +9,7 @@

99pub struct Connection {

1010 pub id: Uuid,

1111 pub user_id: Uuid,

12 pub tx: mpsc::UnboundedSender<ServerFrame>,

12 pub tx: mpsc::Sender<ServerFrame>,

1313 pub established_at: OffsetDateTime,

1414}

1515

@@ -92,10 +92,9 @@

9292 return 0;

9393 };

9494 for (id, conn) in conns.iter() {

95 if conn.tx.send(frame.clone()).is_ok() {

96 delivered += 1;

97 } else {

98 dead.push(*id);

95 match conn.tx.try_send(frame.clone()) {

96 Ok(()) => delivered += 1,

97 Err(_) => dead.push(*id),

9998 }

10099 }

101100 for id in dead {

Modified chat/glayer/metrics.rs +4 −0 View chat/glayer/metrics.rs

@@ -31,3 +31,7 @@

3131pub fn hub_register_rejected() {

3232 counter!("chat.hub_register_rejected_total").increment(1);

3333}

34

35pub fn typing_rejected() {

36 counter!("chat.typing_rejected_total").increment(1);

37}

Modified chat/glayer/ws.rs +15 −3 View chat/glayer/ws.rs

@@ -6,6 +6,7 @@

66use axum::response::Response;

77use bytes::Bytes;

88use chatlayer::AuthIdentity;

9use storagelayer::ChatStorageCtx;

910use tokio::sync::mpsc;

1011use tokio::time::Sleep;

1112use uuid::Uuid;

@@ -20,6 +21,8 @@

2021const WS_CLOSE_TOO_MANY_CONNECTIONS: u16 = 4002;

2122const WS_CLOSE_AUTH_TIMEOUT: u16 = 4003;

2223

24const WS_SEND_QUEUE_CAPACITY: usize = 256;

25

2326pub async fn ws_handler(ws: WebSocketUpgrade, State(state): State<GatewayState>) -> Response {

2427 ws.max_message_size(64 * 1024)

2528 .max_write_buffer_size(1024 * 1024)

@@ -58,7 +61,7 @@

5861 }

5962 };

6063

61 let (tx, rx) = mpsc::unbounded_channel::<ServerFrame>();

64 let (tx, rx) = mpsc::channel::<ServerFrame>(WS_SEND_QUEUE_CAPACITY);

6265 let user_id = identity.user_id;

6366 let conn_id = Uuid::now_v7();

6467 let conn = Connection {

@@ -141,12 +144,21 @@

141144 if !state.typing.allow(conversation_id, user_id) {

142145 return;

143146 }

147 let store = state.chat.storage();

148 let member_ok = store

149 .conversations()

150 .list_member_user_ids(&ChatStorageCtx::new(user_id), conversation_id)

151 .await

152 .is_ok();

153 if !member_ok {

154 crate::metrics::typing_rejected();

155 return;

156 }

144157 let signal = TypingSignal {

145158 conversation_id,

146159 user_id,

147160 ts: time::OffsetDateTime::now_utc(),

148161 };

149 let store = state.chat.storage();

150162 broadcast_typing(&store, &state.hub, &signal).await;

151163 if let Err(e) = state.publish_typing(&signal).await {

152164 tracing::warn!(error = %e, "failed to publish typing signal");

@@ -155,7 +167,7 @@

155167

156168async fn pump(

157169 ws: &mut WebSocket,

158 rx: &mut mpsc::UnboundedReceiver<ServerFrame>,

170 rx: &mut mpsc::Receiver<ServerFrame>,

159171 state: &GatewayState,

160172 user_id: Uuid,

161173 _conn_id: Uuid,

Modified chat/live/handler/meet.rs +0 −4 View chat/live/handler/meet.rs

@@ -9,25 +9,21 @@

99use crate::service::LiveService;

1010use crate::types::{JoinInfo, MeetPresence, ParticipantRecord};

1111

12/// Request body for `POST /live/meet/join`.

1312#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

1413pub(crate) struct JoinRoomCmd {

1514 pub channel_id: Uuid,

1615}

1716

18/// Request body for `POST /live/meet/leave`.

1917#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

2018pub(crate) struct LeaveRoomCmd {

2119 pub meet_id: Uuid,

2220}

2321

24/// Request body for `POST /live/meet/end`.

2522#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

2623pub(crate) struct EndRoomCmd {

2724 pub channel_id: Uuid,

2825}

2926

30/// Query for `GET /live/meet/history`.

3127#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]

3228pub(crate) struct HistoryQuery {

3329 pub limit: u32,

Modified chat/live/handler/routes.rs +0 −5 View chat/live/handler/routes.rs

@@ -6,11 +6,6 @@

66use jwt::inject_jwt_ctx;

77use rate_limit::ratelimit;

88

9/// Builds the standalone live-service router.

10///

11/// - `POST /live/webhook` is public: LiveKit signs it with the webhook secret

12/// and no bearer token is involved.

13/// - All `/live/meet/*` routes require a valid access token (JWT middleware).

149pub fn live_router(svc: LiveService) -> Router {

1510 let public = Router::new().route(

1611 "/live/webhook",

Modified chat/live/metrics.rs +4 −0 View chat/live/metrics.rs

@@ -8,6 +8,10 @@

88 counter!("chat.live_webhooks_ignored_total", "event" => event.to_string()).increment(1);

99}

1010

11pub fn webhook_replay() {

12 counter!("chat.live_webhooks_replay_total").increment(1);

13}

14

1115pub fn token_minted() {

1216 counter!("chat.live_tokens_minted_total").increment(1);

1317}

Modified chat/live/service.rs +19 −1 View chat/live/service.rs

@@ -285,7 +285,25 @@

285285 }

286286

287287 pub async fn handle_webhook(&self, headers: &HeaderMap, body: &[u8]) -> LiveResult<()> {

288 verify_webhook(headers, body, &self.settings.webhook_secret)?;

288 let claims = verify_webhook(

289 headers,

290 body,

291 &self.settings.webhook_secret,

292 &self.settings.api_key,

293 )?;

294 if !self

295 .kv

296 .set_nx(

297 &format!("chat:live:webhook:{sub}", sub = claims.sub),

298 b"1".to_vec(),

299 300,

300 )

301 .await

302 .map_err(|e| LiveError::Internal(format!("webhook dedup failed: {e}")))?

303 {

304 crate::metrics::webhook_replay();

305 return Ok(());

306 }

289307 let webhook: LiveKitWebhook = serde_json::from_slice(body)?;

290308 crate::metrics::webhook_received(&webhook.event);

291309 let Some(envelope_payload) = event_from_webhook(&webhook)? else {

Modified chat/live/types.rs +0 −5 View chat/live/types.rs

@@ -4,7 +4,6 @@

44use utoipa::ToSchema;

55use uuid::Uuid;

66

7/// Public view of a meeting room attached to an object channel.

87#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]

98pub struct MeetInfo {

109 pub id: Uuid,

@@ -32,7 +31,6 @@

3231 }

3332}

3433

35/// A participant row with its leave outcome (audit trail).

3634#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]

3735pub struct ParticipantRecord {

3836 pub id: Uuid,

@@ -56,8 +54,6 @@

5654 }

5755}

5856

59/// Result of a `join_room` call: the meeting plus everything the client needs

60/// to connect straight to LiveKit.

6157#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]

6258pub struct JoinInfo {

6359 pub meet: MeetInfo,

@@ -66,7 +62,6 @@

6662 pub participants: Vec<ParticipantRecord>,

6763}

6864

69/// Active presence snapshot for a room.

7065#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ToSchema)]

7166pub struct MeetPresence {

7267 pub meet: MeetInfo,

Modified chat/live/webhook.rs +22 −5 View chat/live/webhook.rs

@@ -39,15 +39,28 @@

3939}

4040

4141#[derive(Debug, Clone, Deserialize)]

42struct WebhookClaims {

43 #[allow(dead_code)]

42pub struct WebhookClaims {

43 #[serde(rename = "sub")]

44 pub sub: String,

45 #[allow(dead_code)]

46 #[serde(rename = "iss", default)]

47 iss: Option<String>,

48 #[allow(dead_code)]

49 #[serde(rename = "nbf")]

50 nbf: i64,

51 #[allow(dead_code)]

4452 #[serde(rename = "exp")]

4553 exp: i64,

46 #[serde(rename = "nh")]

54 #[serde(rename = "nh")]

4755 nh: String,

4856}

4957

50pub fn verify_webhook(headers: &HeaderMap, body: &[u8], webhook_secret: &str) -> LiveResult<()> {

58pub fn verify_webhook(

59 headers: &HeaderMap,

60 body: &[u8],

61 webhook_secret: &str,

62 api_key: &str,

63) -> LiveResult<WebhookClaims> {

5164 if webhook_secret.is_empty() {

5265 return Err(LiveError::WebhookVerification(

5366 "webhook secret is empty".into(),

@@ -61,6 +74,10 @@

6174 let digest = hex::encode(Sha256::digest(body));

6275 let mut validation = Validation::new(Algorithm::HS256);

6376 validation.validate_exp = true;

77 validation.validate_nbf = true;

78 if !api_key.is_empty() {

79 validation.set_issuer(&[api_key]);

80 }

6481 let key = DecodingKey::from_secret(webhook_secret.as_bytes());

6582 let claims: WebhookClaims = decode(auth, &key, &validation)

6683 .map_err(|e| LiveError::WebhookVerification(format!("bad jwt: {e}")))?

@@ -70,7 +87,7 @@

7087 "body hash mismatch (nh claim)".into(),

7188 ));

7289 }

73 Ok(())

90 Ok(claims)

7491}

7592

7693pub async fn handle_webhook(

Modified chat/storagelayer/mapper/conversation.rs +1 −1 View chat/storagelayer/mapper/conversation.rs

@@ -131,7 +131,7 @@

131131 channel_id: Uuid,

132132 page: Page,

133133 ) -> ChatStorageResult<Vec<Conversation>> {

134 auth::require_authenticated(ctx).await?;

134 auth::require_channel_member(&self.db, ctx, channel_id).await?;

135135 let db = self.db.reader();

136136 let rows = chat_entity::Entity::find()

137137 .filter(chat_entity::COLUMN.channel_id.eq(channel_id))

Modified chat/storagelayer/mapper/meet.rs +4 −1 View chat/storagelayer/mapper/meet.rs

@@ -147,12 +147,15 @@

147147 }

148148

149149 async fn mark_ended(&self, ctx: &ChatStorageCtx, meet_id: Uuid) -> ChatStorageResult<Meet> {

150 auth::require_authenticated(ctx).await?;

150 let caller = auth::require_authenticated(ctx).await?;

151151 let db = self.db.writer();

152152 let model = meet_entity::Entity::find_by_id(meet_id)

153153 .one(&db)

154154 .await?

155155 .ok_or(ChatStorageError::MeetNotFound(meet_id))?;

156 if model.created_by != caller {

157 auth::require_channel_admin(&self.db, ctx, model.channel_id).await?;

158 }

156159 let mut ended: meet_entity::ActiveModel = model.into();

157160 ended.status = Set(meet_entity::MeetStatus::Ended);

158161 ended.ended_at = Set(Some(OffsetDateTime::now_utc()));

Modified chat/storagelayer/auth.rs +23 −5 View chat/storagelayer/auth.rs

@@ -120,11 +120,29 @@

120120 chat_id: Uuid,

121121) -> ChatStorageResult<Uuid> {

122122 let user_id = require_authenticated(ctx).await?;

123 let space_id = space_id_for_chat(&db.reader(), chat_id)

124 .await?

125 .ok_or_else(|| chat_not_in_space(chat_id))?;

126 require_active_member(&db.reader(), space_id, user_id).await?;

127 Ok(user_id)

123 let reader = db.reader();

124 let Some(chat) = chat_entity::Entity::find_by_id(chat_id).one(&reader).await? else {

125 return Err(chat_not_in_space(chat_id));

126 };

127 match chat.kind {

128 chat_entity::ChatKind::Private => {

129 let allowed = chat.creator_user_id == user_id

130 || chat.counterpart_user_id == Some(user_id);

131 if !allowed {

132 return Err(ChatStorageError::PermissionDenied(format!(

133 "user {user_id} is not a participant of private chat {chat_id}"

134 )));

135 }

136 Ok(user_id)

137 }

138 chat_entity::ChatKind::Group => {

139 let space_id = space_id_for_chat(&db.reader(), chat_id)

140 .await?

141 .ok_or_else(|| chat_not_in_space(chat_id))?;

142 require_active_member(&db.reader(), space_id, user_id).await?;

143 Ok(user_id)

144 }

145 }

128146}

129147

130148async fn require_admin_in_space<C: ConnectionTrait>(