1 use activitypub_federation::config::Data as ContextData;
14 use actix_web::{web, Error, HttpRequest, HttpResponse};
15 use actix_web_actors::ws;
16 use lemmy_api::Perform;
17 use lemmy_api_common::{
45 context::LemmyContext,
46 custom_emoji::{CreateCustomEmoji, DeleteCustomEmoji, EditCustomEmoji},
62 MarkCommentReplyAsRead,
63 MarkPersonMentionAsRead,
64 PasswordChangeAfterReset,
89 CreatePrivateMessageReport,
93 ListPrivateMessageReports,
94 MarkPrivateMessageAsRead,
95 ResolvePrivateMessageReport,
98 ApproveRegistrationApplication,
103 GetUnreadRegistrationApplicationCount,
105 ListRegistrationApplications,
115 connect::{Connect, Disconnect},
118 serialize_websocket_message,
119 structs::{CommunityJoin, ModJoin, PostJoin, UserJoin},
125 use lemmy_api_crud::PerformCrud;
126 use lemmy_apub::{api::PerformApub, SendActivity};
127 use lemmy_utils::{error::LemmyError, rate_limit::RateLimitCell, ConnectionId, IpAddr};
128 use serde::Deserialize;
129 use serde_json::Value;
134 time::{Duration, Instant},
136 use tracing::{debug, error};
138 /// How often heartbeat pings are sent
139 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(25);
141 /// How long before lack of client response causes a timeout
142 const CLIENT_TIMEOUT: Duration = Duration::from_secs(60);
144 pub struct WsChatSession {
145 /// unique session id
146 pub id: ConnectionId,
150 /// Client must send ping at least once per 10 seconds (CLIENT_TIMEOUT),
151 /// otherwise we drop connection.
155 apub_data: ContextData<LemmyContext>,
158 pub async fn websocket(
161 rate_limiter: web::Data<RateLimitCell>,
162 apub_data: ContextData<LemmyContext>,
163 ) -> Result<HttpResponse, Error> {
164 let client_ip = IpAddr(
167 .realip_remote_addr()
168 .unwrap_or("blank_ip")
172 let check = rate_limiter.message().check(client_ip.clone());
175 "Websocket join with IP: {} has been rate limited.",
178 return Ok(HttpResponse::TooManyRequests().finish());
193 /// helper method that sends ping to client every few seconds (HEARTBEAT_INTERVAL).
195 /// also this method checks heartbeats from client
196 fn hb(ctx: &mut ws::WebsocketContext<WsChatSession>) {
197 ctx.run_interval(HEARTBEAT_INTERVAL, |act, ctx| {
198 // check client heartbeats
199 if Instant::now().duration_since(act.hb) > CLIENT_TIMEOUT {
200 // heartbeat timed out
202 // notify chat server
206 .do_send(Disconnect { id: act.id });
211 // don't try to send a ping
219 impl Actor for WsChatSession {
220 type Context = ws::WebsocketContext<Self>;
222 /// Method is called on actor start.
223 /// We register ws session with ChatServer
224 fn started(&mut self, ctx: &mut Self::Context) {
225 // we'll start heartbeat process on session start.
228 // register self in chat server. `AsyncContext::wait` register
229 // future within context, but context waits until this future resolves
230 // before processing any other events.
231 // HttpContext::state() is instance of WsChatSessionState, state is shared
232 // across all routes within application
233 let addr = ctx.address();
238 addr: addr.recipient(),
241 .then(|res, act, ctx| {
243 Ok(res) => act.id = res,
244 // something is wrong with chat server
251 fn stopping(&mut self, _: &mut Self::Context) -> Running {
252 // notify chat server
256 .do_send(Disconnect { id: self.id });
261 /// Handle messages from chat server, we simply send it to peer websocket
262 impl Handler<WsMessage> for WsChatSession {
265 fn handle(&mut self, msg: WsMessage, ctx: &mut Self::Context) {
270 /// WebSocket message handler
271 impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WsChatSession {
272 fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
273 let msg = match msg {
282 ws::Message::Ping(msg) => {
283 self.hb = Instant::now();
286 ws::Message::Pong(_) => {
287 self.hb = Instant::now();
289 ws::Message::Text(text) => {
290 let ip_clone = self.ip.clone();
291 let id_clone = self.id.to_owned();
292 let context_clone = self.apub_data.reset_request_count();
294 let fut = Box::pin(async move {
295 let msg = text.trim().to_string();
296 parse_json_message(msg, ip_clone, id_clone, context_clone).await
300 .then(|res, _, ctx| {
302 Ok(res) => ctx.text(res),
303 Err(e) => error!("{}", &e),
305 actix::fut::ready(())
309 ws::Message::Binary(_) => println!("Unexpected binary"),
310 ws::Message::Close(reason) => {
314 ws::Message::Continuation(_) => {
317 ws::Message::Nop => (),
322 /// Entry point for our websocket route
323 async fn parse_json_message(
326 connection_id: ConnectionId,
327 context: ContextData<LemmyContext>,
328 ) -> Result<String, LemmyError> {
329 let rate_limiter = context.settings_updated_channel();
330 let json: Value = serde_json::from_str(&msg)?;
334 .ok_or_else(|| LemmyError::from_message("missing data"))?;
336 let missing_op_err = || LemmyError::from_message("missing op");
340 .ok_or_else(missing_op_err)?
342 .ok_or_else(missing_op_err)?;
344 // check if api call passes the rate limit, and generate future for later execution
345 if let Ok(user_operation_crud) = UserOperationCrud::from_str(op) {
346 let passed = match user_operation_crud {
347 UserOperationCrud::Register => rate_limiter.register().check(ip),
348 UserOperationCrud::CreatePost => rate_limiter.post().check(ip),
349 UserOperationCrud::CreateCommunity => rate_limiter.register().check(ip),
350 UserOperationCrud::CreateComment => rate_limiter.comment().check(ip),
351 _ => rate_limiter.message().check(ip),
353 check_rate_limit_passed(passed)?;
354 match_websocket_operation_crud(context, connection_id, user_operation_crud, data).await
355 } else if let Ok(user_operation) = UserOperation::from_str(op) {
356 let passed = match user_operation {
357 UserOperation::GetCaptcha => rate_limiter.post().check(ip),
358 _ => rate_limiter.message().check(ip),
360 check_rate_limit_passed(passed)?;
361 match_websocket_operation(context, connection_id, user_operation, data).await
363 let user_operation = UserOperationApub::from_str(op)?;
364 let passed = match user_operation {
365 UserOperationApub::Search => rate_limiter.search().check(ip),
366 _ => rate_limiter.message().check(ip),
368 check_rate_limit_passed(passed)?;
369 match_websocket_operation_apub(context, connection_id, user_operation, data).await
373 fn check_rate_limit_passed(passed: bool) -> Result<(), LemmyError> {
377 // if rate limit was hit, respond with message
378 Err(LemmyError::from_message("rate_limit_error"))
382 pub async fn match_websocket_operation_crud(
383 context: ContextData<LemmyContext>,
385 op: UserOperationCrud,
387 ) -> result::Result<String, LemmyError> {
390 UserOperationCrud::Register => {
391 do_websocket_operation_crud::<Register>(context, id, op, data).await
393 UserOperationCrud::DeleteAccount => {
394 do_websocket_operation_crud::<DeleteAccount>(context, id, op, data).await
397 // Private Message ops
398 UserOperationCrud::CreatePrivateMessage => {
399 do_websocket_operation_crud::<CreatePrivateMessage>(context, id, op, data).await
401 UserOperationCrud::EditPrivateMessage => {
402 do_websocket_operation_crud::<EditPrivateMessage>(context, id, op, data).await
404 UserOperationCrud::DeletePrivateMessage => {
405 do_websocket_operation_crud::<DeletePrivateMessage>(context, id, op, data).await
407 UserOperationCrud::GetPrivateMessages => {
408 do_websocket_operation_crud::<GetPrivateMessages>(context, id, op, data).await
412 UserOperationCrud::CreateSite => {
413 do_websocket_operation_crud::<CreateSite>(context, id, op, data).await
415 UserOperationCrud::EditSite => {
416 do_websocket_operation_crud::<EditSite>(context, id, op, data).await
418 UserOperationCrud::GetSite => {
419 do_websocket_operation_crud::<GetSite>(context, id, op, data).await
423 UserOperationCrud::ListCommunities => {
424 do_websocket_operation_crud::<ListCommunities>(context, id, op, data).await
426 UserOperationCrud::CreateCommunity => {
427 do_websocket_operation_crud::<CreateCommunity>(context, id, op, data).await
429 UserOperationCrud::EditCommunity => {
430 do_websocket_operation_crud::<EditCommunity>(context, id, op, data).await
432 UserOperationCrud::DeleteCommunity => {
433 do_websocket_operation_crud::<DeleteCommunity>(context, id, op, data).await
435 UserOperationCrud::RemoveCommunity => {
436 do_websocket_operation_crud::<RemoveCommunity>(context, id, op, data).await
440 UserOperationCrud::CreatePost => {
441 do_websocket_operation_crud::<CreatePost>(context, id, op, data).await
443 UserOperationCrud::GetPost => {
444 do_websocket_operation_crud::<GetPost>(context, id, op, data).await
446 UserOperationCrud::EditPost => {
447 do_websocket_operation_crud::<EditPost>(context, id, op, data).await
449 UserOperationCrud::DeletePost => {
450 do_websocket_operation_crud::<DeletePost>(context, id, op, data).await
452 UserOperationCrud::RemovePost => {
453 do_websocket_operation_crud::<RemovePost>(context, id, op, data).await
457 UserOperationCrud::CreateComment => {
458 do_websocket_operation_crud::<CreateComment>(context, id, op, data).await
460 UserOperationCrud::EditComment => {
461 do_websocket_operation_crud::<EditComment>(context, id, op, data).await
463 UserOperationCrud::DeleteComment => {
464 do_websocket_operation_crud::<DeleteComment>(context, id, op, data).await
466 UserOperationCrud::RemoveComment => {
467 do_websocket_operation_crud::<RemoveComment>(context, id, op, data).await
469 UserOperationCrud::GetComment => {
470 do_websocket_operation_crud::<GetComment>(context, id, op, data).await
473 UserOperationCrud::CreateCustomEmoji => {
474 do_websocket_operation_crud::<CreateCustomEmoji>(context, id, op, data).await
476 UserOperationCrud::EditCustomEmoji => {
477 do_websocket_operation_crud::<EditCustomEmoji>(context, id, op, data).await
479 UserOperationCrud::DeleteCustomEmoji => {
480 do_websocket_operation_crud::<DeleteCustomEmoji>(context, id, op, data).await
485 async fn do_websocket_operation_crud<'a, 'b, Data>(
486 context: ContextData<LemmyContext>,
488 op: UserOperationCrud,
490 ) -> result::Result<String, LemmyError>
492 Data: PerformCrud + SendActivity<Response = <Data as PerformCrud>::Response> + Send,
493 for<'de> Data: Deserialize<'de>,
495 let parsed_data: Data = serde_json::from_value(data)?;
496 let res = parsed_data
497 .perform(&web::Data::new(context.deref().clone()), Some(id))
499 SendActivity::send_activity(&parsed_data, &res, &context).await?;
500 serialize_websocket_message(&op, &res)
503 pub async fn match_websocket_operation_apub(
504 context: ContextData<LemmyContext>,
506 op: UserOperationApub,
508 ) -> result::Result<String, LemmyError> {
510 UserOperationApub::GetPersonDetails => {
511 do_websocket_operation_apub::<GetPersonDetails>(context, id, op, data).await
513 UserOperationApub::GetCommunity => {
514 do_websocket_operation_apub::<GetCommunity>(context, id, op, data).await
516 UserOperationApub::GetComments => {
517 do_websocket_operation_apub::<GetComments>(context, id, op, data).await
519 UserOperationApub::GetPosts => {
520 do_websocket_operation_apub::<GetPosts>(context, id, op, data).await
522 UserOperationApub::ResolveObject => {
523 do_websocket_operation_apub::<ResolveObject>(context, id, op, data).await
525 UserOperationApub::Search => do_websocket_operation_apub::<Search>(context, id, op, data).await,
529 async fn do_websocket_operation_apub<'a, 'b, Data>(
530 context: ContextData<LemmyContext>,
532 op: UserOperationApub,
534 ) -> result::Result<String, LemmyError>
536 Data: PerformApub + SendActivity<Response = <Data as PerformApub>::Response> + Send,
537 for<'de> Data: Deserialize<'de>,
539 let parsed_data: Data = serde_json::from_value(data)?;
540 let res = parsed_data.perform(&context, Some(id)).await?;
541 SendActivity::send_activity(&parsed_data, &res, &context).await?;
542 serialize_websocket_message(&op, &res)
545 pub async fn match_websocket_operation(
546 context: ContextData<LemmyContext>,
550 ) -> result::Result<String, LemmyError> {
553 UserOperation::Login => do_websocket_operation::<Login>(context, id, op, data).await,
554 UserOperation::GetCaptcha => do_websocket_operation::<GetCaptcha>(context, id, op, data).await,
555 UserOperation::GetReplies => do_websocket_operation::<GetReplies>(context, id, op, data).await,
556 UserOperation::AddAdmin => do_websocket_operation::<AddAdmin>(context, id, op, data).await,
557 UserOperation::GetUnreadRegistrationApplicationCount => {
558 do_websocket_operation::<GetUnreadRegistrationApplicationCount>(context, id, op, data).await
560 UserOperation::ListRegistrationApplications => {
561 do_websocket_operation::<ListRegistrationApplications>(context, id, op, data).await
563 UserOperation::ApproveRegistrationApplication => {
564 do_websocket_operation::<ApproveRegistrationApplication>(context, id, op, data).await
566 UserOperation::BanPerson => do_websocket_operation::<BanPerson>(context, id, op, data).await,
567 UserOperation::GetBannedPersons => {
568 do_websocket_operation::<GetBannedPersons>(context, id, op, data).await
570 UserOperation::BlockPerson => {
571 do_websocket_operation::<BlockPerson>(context, id, op, data).await
573 UserOperation::GetPersonMentions => {
574 do_websocket_operation::<GetPersonMentions>(context, id, op, data).await
576 UserOperation::MarkPersonMentionAsRead => {
577 do_websocket_operation::<MarkPersonMentionAsRead>(context, id, op, data).await
579 UserOperation::MarkCommentReplyAsRead => {
580 do_websocket_operation::<MarkCommentReplyAsRead>(context, id, op, data).await
582 UserOperation::MarkAllAsRead => {
583 do_websocket_operation::<MarkAllAsRead>(context, id, op, data).await
585 UserOperation::PasswordReset => {
586 do_websocket_operation::<PasswordReset>(context, id, op, data).await
588 UserOperation::PasswordChange => {
589 do_websocket_operation::<PasswordChangeAfterReset>(context, id, op, data).await
591 UserOperation::UserJoin => do_websocket_operation::<UserJoin>(context, id, op, data).await,
592 UserOperation::PostJoin => do_websocket_operation::<PostJoin>(context, id, op, data).await,
593 UserOperation::CommunityJoin => {
594 do_websocket_operation::<CommunityJoin>(context, id, op, data).await
596 UserOperation::ModJoin => do_websocket_operation::<ModJoin>(context, id, op, data).await,
597 UserOperation::SaveUserSettings => {
598 do_websocket_operation::<SaveUserSettings>(context, id, op, data).await
600 UserOperation::ChangePassword => {
601 do_websocket_operation::<ChangePassword>(context, id, op, data).await
603 UserOperation::GetReportCount => {
604 do_websocket_operation::<GetReportCount>(context, id, op, data).await
606 UserOperation::GetUnreadCount => {
607 do_websocket_operation::<GetUnreadCount>(context, id, op, data).await
609 UserOperation::VerifyEmail => {
610 do_websocket_operation::<VerifyEmail>(context, id, op, data).await
613 // Private Message ops
614 UserOperation::MarkPrivateMessageAsRead => {
615 do_websocket_operation::<MarkPrivateMessageAsRead>(context, id, op, data).await
617 UserOperation::CreatePrivateMessageReport => {
618 do_websocket_operation::<CreatePrivateMessageReport>(context, id, op, data).await
620 UserOperation::ResolvePrivateMessageReport => {
621 do_websocket_operation::<ResolvePrivateMessageReport>(context, id, op, data).await
623 UserOperation::ListPrivateMessageReports => {
624 do_websocket_operation::<ListPrivateMessageReports>(context, id, op, data).await
628 UserOperation::GetModlog => do_websocket_operation::<GetModlog>(context, id, op, data).await,
629 UserOperation::PurgePerson => {
630 do_websocket_operation::<PurgePerson>(context, id, op, data).await
632 UserOperation::PurgeCommunity => {
633 do_websocket_operation::<PurgeCommunity>(context, id, op, data).await
635 UserOperation::PurgePost => do_websocket_operation::<PurgePost>(context, id, op, data).await,
636 UserOperation::PurgeComment => {
637 do_websocket_operation::<PurgeComment>(context, id, op, data).await
639 UserOperation::TransferCommunity => {
640 do_websocket_operation::<TransferCommunity>(context, id, op, data).await
642 UserOperation::LeaveAdmin => do_websocket_operation::<LeaveAdmin>(context, id, op, data).await,
645 UserOperation::FollowCommunity => {
646 do_websocket_operation::<FollowCommunity>(context, id, op, data).await
648 UserOperation::BlockCommunity => {
649 do_websocket_operation::<BlockCommunity>(context, id, op, data).await
651 UserOperation::BanFromCommunity => {
652 do_websocket_operation::<BanFromCommunity>(context, id, op, data).await
654 UserOperation::AddModToCommunity => {
655 do_websocket_operation::<AddModToCommunity>(context, id, op, data).await
659 UserOperation::LockPost => do_websocket_operation::<LockPost>(context, id, op, data).await,
660 UserOperation::FeaturePost => {
661 do_websocket_operation::<FeaturePost>(context, id, op, data).await
663 UserOperation::CreatePostLike => {
664 do_websocket_operation::<CreatePostLike>(context, id, op, data).await
666 UserOperation::MarkPostAsRead => {
667 do_websocket_operation::<MarkPostAsRead>(context, id, op, data).await
669 UserOperation::SavePost => do_websocket_operation::<SavePost>(context, id, op, data).await,
670 UserOperation::CreatePostReport => {
671 do_websocket_operation::<CreatePostReport>(context, id, op, data).await
673 UserOperation::ListPostReports => {
674 do_websocket_operation::<ListPostReports>(context, id, op, data).await
676 UserOperation::ResolvePostReport => {
677 do_websocket_operation::<ResolvePostReport>(context, id, op, data).await
679 UserOperation::GetSiteMetadata => {
680 do_websocket_operation::<GetSiteMetadata>(context, id, op, data).await
684 UserOperation::SaveComment => {
685 do_websocket_operation::<SaveComment>(context, id, op, data).await
687 UserOperation::CreateCommentLike => {
688 do_websocket_operation::<CreateCommentLike>(context, id, op, data).await
690 UserOperation::DistinguishComment => {
691 do_websocket_operation::<DistinguishComment>(context, id, op, data).await
693 UserOperation::CreateCommentReport => {
694 do_websocket_operation::<CreateCommentReport>(context, id, op, data).await
696 UserOperation::ListCommentReports => {
697 do_websocket_operation::<ListCommentReports>(context, id, op, data).await
699 UserOperation::ResolveCommentReport => {
700 do_websocket_operation::<ResolveCommentReport>(context, id, op, data).await
705 async fn do_websocket_operation<'a, 'b, Data>(
706 context: ContextData<LemmyContext>,
710 ) -> result::Result<String, LemmyError>
712 Data: Perform + SendActivity<Response = <Data as Perform>::Response> + Send,
713 for<'de> Data: Deserialize<'de>,
715 let parsed_data: Data = serde_json::from_value(data)?;
716 let res = parsed_data
717 .perform(&web::Data::new(context.deref().clone()), Some(id))
719 SendActivity::send_activity(&parsed_data, &res, &context).await?;
720 serialize_websocket_message(&op, &res)