Change RateLimit to act as a middleware

pull/744/head
asonix 2020-04-19 22:59:07 -05:00
parent f300c67a4d
commit ac43b86b60
16 changed files with 962 additions and 881 deletions

2
server/Cargo.lock generated vendored
View File

@ -1410,6 +1410,7 @@ dependencies = [
"dotenv 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)", "dotenv 0.15.0 (registry+https://github.com/rust-lang/crates.io-index)",
"env_logger 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", "env_logger 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
"failure 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)", "failure 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
"futures 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)",
"hjson 0.8.2 (registry+https://github.com/rust-lang/crates.io-index)", "hjson 0.8.2 (registry+https://github.com/rust-lang/crates.io-index)",
"htmlescape 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)", "htmlescape 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
"isahc 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)", "isahc 0.9.1 (registry+https://github.com/rust-lang/crates.io-index)",
@ -1427,6 +1428,7 @@ dependencies = [
"sha2 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)", "sha2 0.8.1 (registry+https://github.com/rust-lang/crates.io-index)",
"strum 0.18.0 (registry+https://github.com/rust-lang/crates.io-index)", "strum 0.18.0 (registry+https://github.com/rust-lang/crates.io-index)",
"strum_macros 0.18.0 (registry+https://github.com/rust-lang/crates.io-index)", "strum_macros 0.18.0 (registry+https://github.com/rust-lang/crates.io-index)",
"tokio 0.2.18 (registry+https://github.com/rust-lang/crates.io-index)",
] ]
[[package]] [[package]]

2
server/Cargo.toml vendored
View File

@ -37,3 +37,5 @@ hjson = "0.8.2"
percent-encoding = "2.1.0" percent-encoding = "2.1.0"
isahc = "0.9" isahc = "0.9"
comrak = "0.7" comrak = "0.7"
tokio = "0.2.18"
futures = "0.3.4"

View File

@ -59,12 +59,13 @@ pub struct GetCommentsResponse {
comments: Vec<CommentView>, comments: Vec<CommentView>,
} }
impl Perform<CommentResponse> for Oper<CreateComment> { impl Perform for Oper<CreateComment> {
type Response = CommentResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> { ) -> Result<CommentResponse, Error> {
let data: &CreateComment = &self.data; let data: &CreateComment = &self.data;
@ -77,13 +78,6 @@ impl Perform<CommentResponse> for Oper<CreateComment> {
let hostname = &format!("https://{}", Settings::get().hostname); let hostname = &format!("https://{}", Settings::get().hostname);
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Check for a community ban // Check for a community ban
@ -253,12 +247,13 @@ impl Perform<CommentResponse> for Oper<CreateComment> {
} }
} }
impl Perform<CommentResponse> for Oper<EditComment> { impl Perform for Oper<EditComment> {
type Response = CommentResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> { ) -> Result<CommentResponse, Error> {
let data: &EditComment = &self.data; let data: &EditComment = &self.data;
@ -269,13 +264,6 @@ impl Perform<CommentResponse> for Oper<EditComment> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let orig_comment = CommentView::read(&conn, data.edit_id, None)?; let orig_comment = CommentView::read(&conn, data.edit_id, None)?;
@ -411,12 +399,13 @@ impl Perform<CommentResponse> for Oper<EditComment> {
} }
} }
impl Perform<CommentResponse> for Oper<SaveComment> { impl Perform for Oper<SaveComment> {
type Response = CommentResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> { ) -> Result<CommentResponse, Error> {
let data: &SaveComment = &self.data; let data: &SaveComment = &self.data;
@ -432,13 +421,6 @@ impl Perform<CommentResponse> for Oper<SaveComment> {
user_id, user_id,
}; };
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
if data.save { if data.save {
@ -462,12 +444,13 @@ impl Perform<CommentResponse> for Oper<SaveComment> {
} }
} }
impl Perform<CommentResponse> for Oper<CreateCommentLike> { impl Perform for Oper<CreateCommentLike> {
type Response = CommentResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> { ) -> Result<CommentResponse, Error> {
let data: &CreateCommentLike = &self.data; let data: &CreateCommentLike = &self.data;
@ -480,13 +463,6 @@ impl Perform<CommentResponse> for Oper<CreateCommentLike> {
let mut recipient_ids = Vec::new(); let mut recipient_ids = Vec::new();
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Don't do a downvote if site has downvotes disabled // Don't do a downvote if site has downvotes disabled
@ -567,12 +543,13 @@ impl Perform<CommentResponse> for Oper<CreateCommentLike> {
} }
} }
impl Perform<GetCommentsResponse> for Oper<GetComments> { impl Perform for Oper<GetComments> {
type Response = GetCommentsResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetCommentsResponse, Error> { ) -> Result<GetCommentsResponse, Error> {
let data: &GetComments = &self.data; let data: &GetComments = &self.data;
@ -592,13 +569,6 @@ impl Perform<GetCommentsResponse> for Oper<GetComments> {
let type_ = ListingType::from_str(&data.type_)?; let type_ = ListingType::from_str(&data.type_)?;
let sort = SortType::from_str(&data.sort)?; let sort = SortType::from_str(&data.sort)?;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let comments = match CommentQueryBuilder::create(&conn) let comments = match CommentQueryBuilder::create(&conn)

View File

@ -111,12 +111,13 @@ pub struct TransferCommunity {
auth: String, auth: String,
} }
impl Perform<GetCommunityResponse> for Oper<GetCommunity> { impl Perform for Oper<GetCommunity> {
type Response = GetCommunityResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetCommunityResponse, Error> { ) -> Result<GetCommunityResponse, Error> {
let data: &GetCommunity = &self.data; let data: &GetCommunity = &self.data;
@ -131,13 +132,6 @@ impl Perform<GetCommunityResponse> for Oper<GetCommunity> {
None => None, None => None,
}; };
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let community_id = match data.id { let community_id = match data.id {
@ -197,12 +191,13 @@ impl Perform<GetCommunityResponse> for Oper<GetCommunity> {
} }
} }
impl Perform<CommunityResponse> for Oper<CreateCommunity> { impl Perform for Oper<CreateCommunity> {
type Response = CommunityResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommunityResponse, Error> { ) -> Result<CommunityResponse, Error> {
let data: &CreateCommunity = &self.data; let data: &CreateCommunity = &self.data;
@ -227,13 +222,6 @@ impl Perform<CommunityResponse> for Oper<CreateCommunity> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = &rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_register(&rl.ip, true)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Check for a site ban // Check for a site ban
@ -283,25 +271,19 @@ impl Perform<CommunityResponse> for Oper<CreateCommunity> {
let community_view = CommunityView::read(&conn, inserted_community.id, Some(user_id))?; let community_view = CommunityView::read(&conn, inserted_community.id, Some(user_id))?;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_register(&rl.ip, false)?;
}
Ok(CommunityResponse { Ok(CommunityResponse {
community: community_view, community: community_view,
}) })
} }
} }
impl Perform<CommunityResponse> for Oper<EditCommunity> { impl Perform for Oper<EditCommunity> {
type Response = CommunityResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommunityResponse, Error> { ) -> Result<CommunityResponse, Error> {
let data: &EditCommunity = &self.data; let data: &EditCommunity = &self.data;
@ -326,13 +308,6 @@ impl Perform<CommunityResponse> for Oper<EditCommunity> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Check for a site ban // Check for a site ban
@ -410,12 +385,13 @@ impl Perform<CommunityResponse> for Oper<EditCommunity> {
} }
} }
impl Perform<ListCommunitiesResponse> for Oper<ListCommunities> { impl Perform for Oper<ListCommunities> {
type Response = ListCommunitiesResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<ListCommunitiesResponse, Error> { ) -> Result<ListCommunitiesResponse, Error> {
let data: &ListCommunities = &self.data; let data: &ListCommunities = &self.data;
@ -439,13 +415,6 @@ impl Perform<ListCommunitiesResponse> for Oper<ListCommunities> {
let sort = SortType::from_str(&data.sort)?; let sort = SortType::from_str(&data.sort)?;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let communities = CommunityQueryBuilder::create(&conn) let communities = CommunityQueryBuilder::create(&conn)
@ -461,12 +430,13 @@ impl Perform<ListCommunitiesResponse> for Oper<ListCommunities> {
} }
} }
impl Perform<CommunityResponse> for Oper<FollowCommunity> { impl Perform for Oper<FollowCommunity> {
type Response = CommunityResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommunityResponse, Error> { ) -> Result<CommunityResponse, Error> {
let data: &FollowCommunity = &self.data; let data: &FollowCommunity = &self.data;
@ -482,13 +452,6 @@ impl Perform<CommunityResponse> for Oper<FollowCommunity> {
user_id, user_id,
}; };
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
if data.follow { if data.follow {
@ -511,12 +474,13 @@ impl Perform<CommunityResponse> for Oper<FollowCommunity> {
} }
} }
impl Perform<GetFollowedCommunitiesResponse> for Oper<GetFollowedCommunities> { impl Perform for Oper<GetFollowedCommunities> {
type Response = GetFollowedCommunitiesResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetFollowedCommunitiesResponse, Error> { ) -> Result<GetFollowedCommunitiesResponse, Error> {
let data: &GetFollowedCommunities = &self.data; let data: &GetFollowedCommunities = &self.data;
@ -527,13 +491,6 @@ impl Perform<GetFollowedCommunitiesResponse> for Oper<GetFollowedCommunities> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let communities: Vec<CommunityFollowerView> = let communities: Vec<CommunityFollowerView> =
@ -547,12 +504,13 @@ impl Perform<GetFollowedCommunitiesResponse> for Oper<GetFollowedCommunities> {
} }
} }
impl Perform<BanFromCommunityResponse> for Oper<BanFromCommunity> { impl Perform for Oper<BanFromCommunity> {
type Response = BanFromCommunityResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<BanFromCommunityResponse, Error> { ) -> Result<BanFromCommunityResponse, Error> {
let data: &BanFromCommunity = &self.data; let data: &BanFromCommunity = &self.data;
@ -568,13 +526,6 @@ impl Perform<BanFromCommunityResponse> for Oper<BanFromCommunity> {
user_id: data.user_id, user_id: data.user_id,
}; };
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
if data.ban { if data.ban {
@ -625,12 +576,13 @@ impl Perform<BanFromCommunityResponse> for Oper<BanFromCommunity> {
} }
} }
impl Perform<AddModToCommunityResponse> for Oper<AddModToCommunity> { impl Perform for Oper<AddModToCommunity> {
type Response = AddModToCommunityResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<AddModToCommunityResponse, Error> { ) -> Result<AddModToCommunityResponse, Error> {
let data: &AddModToCommunity = &self.data; let data: &AddModToCommunity = &self.data;
@ -646,13 +598,6 @@ impl Perform<AddModToCommunityResponse> for Oper<AddModToCommunity> {
user_id: data.user_id, user_id: data.user_id,
}; };
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
if data.added { if data.added {
@ -693,12 +638,13 @@ impl Perform<AddModToCommunityResponse> for Oper<AddModToCommunity> {
} }
} }
impl Perform<GetCommunityResponse> for Oper<TransferCommunity> { impl Perform for Oper<TransferCommunity> {
type Response = GetCommunityResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetCommunityResponse, Error> { ) -> Result<GetCommunityResponse, Error> {
let data: &TransferCommunity = &self.data; let data: &TransferCommunity = &self.data;
@ -709,13 +655,6 @@ impl Perform<GetCommunityResponse> for Oper<TransferCommunity> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let read_community = Community::read(&conn, data.community_id)?; let read_community = Community::read(&conn, data.community_id)?;

View File

@ -22,7 +22,6 @@ use crate::{
naive_now, remove_slurs, send_email, slur_check, slurs_vec_to_str, naive_now, remove_slurs, send_email, slur_check, slurs_vec_to_str,
}; };
use crate::rate_limit::RateLimitInfo;
use crate::settings::Settings; use crate::settings::Settings;
use crate::websocket::UserOperation; use crate::websocket::UserOperation;
use crate::websocket::{ use crate::websocket::{
@ -69,13 +68,12 @@ impl<T> Oper<T> {
} }
} }
pub trait Perform<T> { pub trait Perform {
type Response: serde::ser::Serialize;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>, ) -> Result<Self::Response, Error>;
) -> Result<T, Error>
where
T: Sized;
} }

View File

@ -77,12 +77,13 @@ pub struct SavePost {
auth: String, auth: String,
} }
impl Perform<PostResponse> for Oper<CreatePost> { impl Perform for Oper<CreatePost> {
type Response = PostResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> { ) -> Result<PostResponse, Error> {
let data: &CreatePost = &self.data; let data: &CreatePost = &self.data;
@ -103,13 +104,6 @@ impl Perform<PostResponse> for Oper<CreatePost> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = &rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_post(&rl.ip, true)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Check for a community ban // Check for a community ban
@ -176,13 +170,6 @@ impl Perform<PostResponse> for Oper<CreatePost> {
Err(_e) => return Err(APIError::err("couldnt_find_post").into()), Err(_e) => return Err(APIError::err("couldnt_find_post").into()),
}; };
if let Some(rl) = &rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_post(&rl.ip, false)?;
}
let res = PostResponse { post: post_view }; let res = PostResponse { post: post_view };
if let Some(ws) = websocket_info { if let Some(ws) = websocket_info {
@ -197,12 +184,13 @@ impl Perform<PostResponse> for Oper<CreatePost> {
} }
} }
impl Perform<GetPostResponse> for Oper<GetPost> { impl Perform for Oper<GetPost> {
type Response = GetPostResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetPostResponse, Error> { ) -> Result<GetPostResponse, Error> {
let data: &GetPost = &self.data; let data: &GetPost = &self.data;
@ -217,13 +205,6 @@ impl Perform<GetPostResponse> for Oper<GetPost> {
None => None, None => None,
}; };
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let post_view = match PostView::read(&conn, data.id, user_id) { let post_view = match PostView::read(&conn, data.id, user_id) {
@ -277,12 +258,13 @@ impl Perform<GetPostResponse> for Oper<GetPost> {
} }
} }
impl Perform<GetPostsResponse> for Oper<GetPosts> { impl Perform for Oper<GetPosts> {
type Response = GetPostsResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetPostsResponse, Error> { ) -> Result<GetPostsResponse, Error> {
let data: &GetPosts = &self.data; let data: &GetPosts = &self.data;
@ -307,13 +289,6 @@ impl Perform<GetPostsResponse> for Oper<GetPosts> {
let type_ = ListingType::from_str(&data.type_)?; let type_ = ListingType::from_str(&data.type_)?;
let sort = SortType::from_str(&data.sort)?; let sort = SortType::from_str(&data.sort)?;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let posts = match PostQueryBuilder::create(&conn) let posts = match PostQueryBuilder::create(&conn)
@ -348,12 +323,13 @@ impl Perform<GetPostsResponse> for Oper<GetPosts> {
} }
} }
impl Perform<PostResponse> for Oper<CreatePostLike> { impl Perform for Oper<CreatePostLike> {
type Response = PostResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> { ) -> Result<PostResponse, Error> {
let data: &CreatePostLike = &self.data; let data: &CreatePostLike = &self.data;
@ -364,13 +340,6 @@ impl Perform<PostResponse> for Oper<CreatePostLike> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Don't do a downvote if site has downvotes disabled // Don't do a downvote if site has downvotes disabled
@ -429,12 +398,13 @@ impl Perform<PostResponse> for Oper<CreatePostLike> {
} }
} }
impl Perform<PostResponse> for Oper<EditPost> { impl Perform for Oper<EditPost> {
type Response = PostResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> { ) -> Result<PostResponse, Error> {
let data: &EditPost = &self.data; let data: &EditPost = &self.data;
@ -455,13 +425,6 @@ impl Perform<PostResponse> for Oper<EditPost> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Verify its the creator or a mod or admin // Verify its the creator or a mod or admin
@ -567,12 +530,13 @@ impl Perform<PostResponse> for Oper<EditPost> {
} }
} }
impl Perform<PostResponse> for Oper<SavePost> { impl Perform for Oper<SavePost> {
type Response = PostResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> { ) -> Result<PostResponse, Error> {
let data: &SavePost = &self.data; let data: &SavePost = &self.data;
@ -588,13 +552,6 @@ impl Perform<PostResponse> for Oper<SavePost> {
user_id, user_id,
}; };
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
if data.save { if data.save {

View File

@ -108,22 +108,16 @@ pub struct SaveSiteConfig {
auth: String, auth: String,
} }
impl Perform<ListCategoriesResponse> for Oper<ListCategories> { impl Perform for Oper<ListCategories> {
type Response = ListCategoriesResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<ListCategoriesResponse, Error> { ) -> Result<ListCategoriesResponse, Error> {
let _data: &ListCategories = &self.data; let _data: &ListCategories = &self.data;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let categories: Vec<Category> = Category::list_all(&conn)?; let categories: Vec<Category> = Category::list_all(&conn)?;
@ -133,22 +127,16 @@ impl Perform<ListCategoriesResponse> for Oper<ListCategories> {
} }
} }
impl Perform<GetModlogResponse> for Oper<GetModlog> { impl Perform for Oper<GetModlog> {
type Response = GetModlogResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetModlogResponse, Error> { ) -> Result<GetModlogResponse, Error> {
let data: &GetModlog = &self.data; let data: &GetModlog = &self.data;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let removed_posts = ModRemovePostView::list( let removed_posts = ModRemovePostView::list(
@ -220,12 +208,13 @@ impl Perform<GetModlogResponse> for Oper<GetModlog> {
} }
} }
impl Perform<SiteResponse> for Oper<CreateSite> { impl Perform for Oper<CreateSite> {
type Response = SiteResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<SiteResponse, Error> { ) -> Result<SiteResponse, Error> {
let data: &CreateSite = &self.data; let data: &CreateSite = &self.data;
@ -246,13 +235,6 @@ impl Perform<SiteResponse> for Oper<CreateSite> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Make sure user is an admin // Make sure user is an admin
@ -281,12 +263,12 @@ impl Perform<SiteResponse> for Oper<CreateSite> {
} }
} }
impl Perform<SiteResponse> for Oper<EditSite> { impl Perform for Oper<EditSite> {
type Response = SiteResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<SiteResponse, Error> { ) -> Result<SiteResponse, Error> {
let data: &EditSite = &self.data; let data: &EditSite = &self.data;
@ -307,13 +289,6 @@ impl Perform<SiteResponse> for Oper<EditSite> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Make sure user is an admin // Make sure user is an admin
@ -354,22 +329,16 @@ impl Perform<SiteResponse> for Oper<EditSite> {
} }
} }
impl Perform<GetSiteResponse> for Oper<GetSite> { impl Perform for Oper<GetSite> {
type Response = GetSiteResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteResponse, Error> { ) -> Result<GetSiteResponse, Error> {
let _data: &GetSite = &self.data; let _data: &GetSite = &self.data;
if let Some(rl) = &rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// TODO refactor this a little // TODO refactor this a little
@ -385,11 +354,7 @@ impl Perform<GetSiteResponse> for Oper<GetSite> {
admin: true, admin: true,
show_nsfw: true, show_nsfw: true,
}; };
let login_response = Oper::new(register).perform( let login_response = Oper::new(register).perform(pool.clone(), websocket_info.clone())?;
pool.clone(),
websocket_info.clone(),
rate_limit_info.clone(),
)?;
info!("Admin {} created", setup.admin_username); info!("Admin {} created", setup.admin_username);
let create_site = CreateSite { let create_site = CreateSite {
@ -400,7 +365,7 @@ impl Perform<GetSiteResponse> for Oper<GetSite> {
enable_nsfw: false, enable_nsfw: false,
auth: login_response.jwt, auth: login_response.jwt,
}; };
Oper::new(create_site).perform(pool, websocket_info.clone(), rate_limit_info)?; Oper::new(create_site).perform(pool, websocket_info.clone())?;
info!("Site {} created", setup.site_name); info!("Site {} created", setup.site_name);
Some(SiteView::read(&conn)?) Some(SiteView::read(&conn)?)
} else { } else {
@ -437,12 +402,13 @@ impl Perform<GetSiteResponse> for Oper<GetSite> {
} }
} }
impl Perform<SearchResponse> for Oper<Search> { impl Perform for Oper<Search> {
type Response = SearchResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<SearchResponse, Error> { ) -> Result<SearchResponse, Error> {
let data: &Search = &self.data; let data: &Search = &self.data;
@ -467,13 +433,6 @@ impl Perform<SearchResponse> for Oper<Search> {
// TODO no clean / non-nsfw searching rn // TODO no clean / non-nsfw searching rn
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
match type_ { match type_ {
@ -569,12 +528,13 @@ impl Perform<SearchResponse> for Oper<Search> {
} }
} }
impl Perform<GetSiteResponse> for Oper<TransferSite> { impl Perform for Oper<TransferSite> {
type Response = GetSiteResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteResponse, Error> { ) -> Result<GetSiteResponse, Error> {
let data: &TransferSite = &self.data; let data: &TransferSite = &self.data;
@ -585,13 +545,6 @@ impl Perform<GetSiteResponse> for Oper<TransferSite> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let read_site = Site::read(&conn, 1)?; let read_site = Site::read(&conn, 1)?;
@ -646,12 +599,13 @@ impl Perform<GetSiteResponse> for Oper<TransferSite> {
} }
} }
impl Perform<GetSiteConfigResponse> for Oper<GetSiteConfig> { impl Perform for Oper<GetSiteConfig> {
type Response = GetSiteConfigResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteConfigResponse, Error> { ) -> Result<GetSiteConfigResponse, Error> {
let data: &GetSiteConfig = &self.data; let data: &GetSiteConfig = &self.data;
@ -662,13 +616,6 @@ impl Perform<GetSiteConfigResponse> for Oper<GetSiteConfig> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Only let admins read this // Only let admins read this
@ -685,12 +632,13 @@ impl Perform<GetSiteConfigResponse> for Oper<GetSiteConfig> {
} }
} }
impl Perform<GetSiteConfigResponse> for Oper<SaveSiteConfig> { impl Perform for Oper<SaveSiteConfig> {
type Response = GetSiteConfigResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteConfigResponse, Error> { ) -> Result<GetSiteConfigResponse, Error> {
let data: &SaveSiteConfig = &self.data; let data: &SaveSiteConfig = &self.data;
@ -701,13 +649,6 @@ impl Perform<GetSiteConfigResponse> for Oper<SaveSiteConfig> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Only let admins read this // Only let admins read this

View File

@ -199,22 +199,16 @@ pub struct UserJoinResponse {
pub user_id: i32, pub user_id: i32,
} }
impl Perform<LoginResponse> for Oper<Login> { impl Perform for Oper<Login> {
type Response = LoginResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> { ) -> Result<LoginResponse, Error> {
let data: &Login = &self.data; let data: &Login = &self.data;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Fetch that username / email // Fetch that username / email
@ -234,22 +228,16 @@ impl Perform<LoginResponse> for Oper<Login> {
} }
} }
impl Perform<LoginResponse> for Oper<Register> { impl Perform for Oper<Register> {
type Response = LoginResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> { ) -> Result<LoginResponse, Error> {
let data: &Register = &self.data; let data: &Register = &self.data;
if let Some(rl) = &rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_register(&rl.ip, true)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Make sure site has open registration // Make sure site has open registration
@ -355,13 +343,6 @@ impl Perform<LoginResponse> for Oper<Register> {
}; };
} }
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_register(&rl.ip, false)?;
}
// Return the jwt // Return the jwt
Ok(LoginResponse { Ok(LoginResponse {
jwt: inserted_user.jwt(), jwt: inserted_user.jwt(),
@ -369,12 +350,13 @@ impl Perform<LoginResponse> for Oper<Register> {
} }
} }
impl Perform<LoginResponse> for Oper<SaveUserSettings> { impl Perform for Oper<SaveUserSettings> {
type Response = LoginResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> { ) -> Result<LoginResponse, Error> {
let data: &SaveUserSettings = &self.data; let data: &SaveUserSettings = &self.data;
@ -385,13 +367,6 @@ impl Perform<LoginResponse> for Oper<SaveUserSettings> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let read_user = User_::read(&conn, user_id)?; let read_user = User_::read(&conn, user_id)?;
@ -471,22 +446,16 @@ impl Perform<LoginResponse> for Oper<SaveUserSettings> {
} }
} }
impl Perform<GetUserDetailsResponse> for Oper<GetUserDetails> { impl Perform for Oper<GetUserDetails> {
type Response = GetUserDetailsResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetUserDetailsResponse, Error> { ) -> Result<GetUserDetailsResponse, Error> {
let data: &GetUserDetails = &self.data; let data: &GetUserDetails = &self.data;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let user_claims: Option<Claims> = match &data.auth { let user_claims: Option<Claims> = match &data.auth {
@ -582,12 +551,13 @@ impl Perform<GetUserDetailsResponse> for Oper<GetUserDetails> {
} }
} }
impl Perform<AddAdminResponse> for Oper<AddAdmin> { impl Perform for Oper<AddAdmin> {
type Response = AddAdminResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<AddAdminResponse, Error> { ) -> Result<AddAdminResponse, Error> {
let data: &AddAdmin = &self.data; let data: &AddAdmin = &self.data;
@ -598,13 +568,6 @@ impl Perform<AddAdminResponse> for Oper<AddAdmin> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Make sure user is an admin // Make sure user is an admin
@ -669,12 +632,13 @@ impl Perform<AddAdminResponse> for Oper<AddAdmin> {
} }
} }
impl Perform<BanUserResponse> for Oper<BanUser> { impl Perform for Oper<BanUser> {
type Response = BanUserResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<BanUserResponse, Error> { ) -> Result<BanUserResponse, Error> {
let data: &BanUser = &self.data; let data: &BanUser = &self.data;
@ -685,13 +649,6 @@ impl Perform<BanUserResponse> for Oper<BanUser> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Make sure user is an admin // Make sure user is an admin
@ -762,12 +719,13 @@ impl Perform<BanUserResponse> for Oper<BanUser> {
} }
} }
impl Perform<GetRepliesResponse> for Oper<GetReplies> { impl Perform for Oper<GetReplies> {
type Response = GetRepliesResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetRepliesResponse, Error> { ) -> Result<GetRepliesResponse, Error> {
let data: &GetReplies = &self.data; let data: &GetReplies = &self.data;
@ -780,13 +738,6 @@ impl Perform<GetRepliesResponse> for Oper<GetReplies> {
let sort = SortType::from_str(&data.sort)?; let sort = SortType::from_str(&data.sort)?;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let replies = ReplyQueryBuilder::create(&conn, user_id) let replies = ReplyQueryBuilder::create(&conn, user_id)
@ -800,12 +751,13 @@ impl Perform<GetRepliesResponse> for Oper<GetReplies> {
} }
} }
impl Perform<GetUserMentionsResponse> for Oper<GetUserMentions> { impl Perform for Oper<GetUserMentions> {
type Response = GetUserMentionsResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetUserMentionsResponse, Error> { ) -> Result<GetUserMentionsResponse, Error> {
let data: &GetUserMentions = &self.data; let data: &GetUserMentions = &self.data;
@ -818,13 +770,6 @@ impl Perform<GetUserMentionsResponse> for Oper<GetUserMentions> {
let sort = SortType::from_str(&data.sort)?; let sort = SortType::from_str(&data.sort)?;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let mentions = UserMentionQueryBuilder::create(&conn, user_id) let mentions = UserMentionQueryBuilder::create(&conn, user_id)
@ -838,12 +783,13 @@ impl Perform<GetUserMentionsResponse> for Oper<GetUserMentions> {
} }
} }
impl Perform<UserMentionResponse> for Oper<EditUserMention> { impl Perform for Oper<EditUserMention> {
type Response = UserMentionResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<UserMentionResponse, Error> { ) -> Result<UserMentionResponse, Error> {
let data: &EditUserMention = &self.data; let data: &EditUserMention = &self.data;
@ -854,13 +800,6 @@ impl Perform<UserMentionResponse> for Oper<EditUserMention> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let user_mention = UserMention::read(&conn, data.user_mention_id)?; let user_mention = UserMention::read(&conn, data.user_mention_id)?;
@ -885,12 +824,13 @@ impl Perform<UserMentionResponse> for Oper<EditUserMention> {
} }
} }
impl Perform<GetRepliesResponse> for Oper<MarkAllAsRead> { impl Perform for Oper<MarkAllAsRead> {
type Response = GetRepliesResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetRepliesResponse, Error> { ) -> Result<GetRepliesResponse, Error> {
let data: &MarkAllAsRead = &self.data; let data: &MarkAllAsRead = &self.data;
@ -901,13 +841,6 @@ impl Perform<GetRepliesResponse> for Oper<MarkAllAsRead> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let replies = ReplyQueryBuilder::create(&conn, user_id) let replies = ReplyQueryBuilder::create(&conn, user_id)
@ -983,12 +916,13 @@ impl Perform<GetRepliesResponse> for Oper<MarkAllAsRead> {
} }
} }
impl Perform<LoginResponse> for Oper<DeleteAccount> { impl Perform for Oper<DeleteAccount> {
type Response = LoginResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> { ) -> Result<LoginResponse, Error> {
let data: &DeleteAccount = &self.data; let data: &DeleteAccount = &self.data;
@ -999,13 +933,6 @@ impl Perform<LoginResponse> for Oper<DeleteAccount> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let user: User_ = User_::read(&conn, user_id)?; let user: User_ = User_::read(&conn, user_id)?;
@ -1078,22 +1005,16 @@ impl Perform<LoginResponse> for Oper<DeleteAccount> {
} }
} }
impl Perform<PasswordResetResponse> for Oper<PasswordReset> { impl Perform for Oper<PasswordReset> {
type Response = PasswordResetResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PasswordResetResponse, Error> { ) -> Result<PasswordResetResponse, Error> {
let data: &PasswordReset = &self.data; let data: &PasswordReset = &self.data;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Fetch that email // Fetch that email
@ -1123,22 +1044,16 @@ impl Perform<PasswordResetResponse> for Oper<PasswordReset> {
} }
} }
impl Perform<LoginResponse> for Oper<PasswordChange> { impl Perform for Oper<PasswordChange> {
type Response = LoginResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> { ) -> Result<LoginResponse, Error> {
let data: &PasswordChange = &self.data; let data: &PasswordChange = &self.data;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Fetch the user_id from the token // Fetch the user_id from the token
@ -1162,12 +1077,13 @@ impl Perform<LoginResponse> for Oper<PasswordChange> {
} }
} }
impl Perform<PrivateMessageResponse> for Oper<CreatePrivateMessage> { impl Perform for Oper<CreatePrivateMessage> {
type Response = PrivateMessageResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PrivateMessageResponse, Error> { ) -> Result<PrivateMessageResponse, Error> {
let data: &CreatePrivateMessage = &self.data; let data: &CreatePrivateMessage = &self.data;
@ -1180,13 +1096,6 @@ impl Perform<PrivateMessageResponse> for Oper<CreatePrivateMessage> {
let hostname = &format!("https://{}", Settings::get().hostname); let hostname = &format!("https://{}", Settings::get().hostname);
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
// Check for a site ban // Check for a site ban
@ -1249,12 +1158,13 @@ impl Perform<PrivateMessageResponse> for Oper<CreatePrivateMessage> {
} }
} }
impl Perform<PrivateMessageResponse> for Oper<EditPrivateMessage> { impl Perform for Oper<EditPrivateMessage> {
type Response = PrivateMessageResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PrivateMessageResponse, Error> { ) -> Result<PrivateMessageResponse, Error> {
let data: &EditPrivateMessage = &self.data; let data: &EditPrivateMessage = &self.data;
@ -1265,13 +1175,6 @@ impl Perform<PrivateMessageResponse> for Oper<EditPrivateMessage> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let orig_private_message = PrivateMessage::read(&conn, data.edit_id)?; let orig_private_message = PrivateMessage::read(&conn, data.edit_id)?;
@ -1318,12 +1221,13 @@ impl Perform<PrivateMessageResponse> for Oper<EditPrivateMessage> {
} }
} }
impl Perform<PrivateMessagesResponse> for Oper<GetPrivateMessages> { impl Perform for Oper<GetPrivateMessages> {
type Response = PrivateMessagesResponse;
fn perform( fn perform(
&self, &self,
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>, _websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PrivateMessagesResponse, Error> { ) -> Result<PrivateMessagesResponse, Error> {
let data: &GetPrivateMessages = &self.data; let data: &GetPrivateMessages = &self.data;
@ -1334,13 +1238,6 @@ impl Perform<PrivateMessagesResponse> for Oper<GetPrivateMessages> {
let user_id = claims.id; let user_id = claims.id;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_message(&rl.ip, false)?;
}
let conn = pool.get()?; let conn = pool.get()?;
let messages = PrivateMessageQueryBuilder::create(&conn, user_id) let messages = PrivateMessageQueryBuilder::create(&conn, user_id)
@ -1353,12 +1250,13 @@ impl Perform<PrivateMessagesResponse> for Oper<GetPrivateMessages> {
} }
} }
impl Perform<UserJoinResponse> for Oper<UserJoin> { impl Perform for Oper<UserJoin> {
type Response = UserJoinResponse;
fn perform( fn perform(
&self, &self,
_pool: Pool<ConnectionManager<PgConnection>>, _pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>, websocket_info: Option<WebsocketInfo>,
_rate_limit_info: Option<RateLimitInfo>,
) -> Result<UserJoinResponse, Error> { ) -> Result<UserJoinResponse, Error> {
let data: &UserJoin = &self.data; let data: &UserJoin = &self.data;

View File

@ -7,15 +7,13 @@ use actix_web::*;
use diesel::r2d2::{ConnectionManager, Pool}; use diesel::r2d2::{ConnectionManager, Pool};
use diesel::PgConnection; use diesel::PgConnection;
use lemmy_server::{ use lemmy_server::{
rate_limit::rate_limiter::RateLimiter, rate_limit::{rate_limiter::RateLimiter, RateLimit},
routes::{api, federation, feeds, index, nodeinfo, webfinger, websocket}, routes::{api, federation, feeds, index, nodeinfo, webfinger, websocket},
settings::Settings, settings::Settings,
websocket::server::*, websocket::server::*,
}; };
use std::{ use std::{io, sync::Arc};
io, use tokio::sync::Mutex;
sync::{Arc, Mutex},
};
embed_migrations!(); embed_migrations!();
@ -36,7 +34,7 @@ async fn main() -> io::Result<()> {
embedded_migrations::run(&conn).unwrap(); embedded_migrations::run(&conn).unwrap();
// Set up the rate limiter // Set up the rate limiter
let rate_limiter = Arc::new(Mutex::new(RateLimiter::default())); let rate_limiter = RateLimit(Arc::new(Mutex::new(RateLimiter::default())));
// Set up websocket server // Set up websocket server
let server = ChatServer::startup(pool.clone(), rate_limiter.clone()).start(); let server = ChatServer::startup(pool.clone(), rate_limiter.clone()).start();
@ -49,13 +47,13 @@ async fn main() -> io::Result<()> {
// Create Http server with websocket support // Create Http server with websocket support
HttpServer::new(move || { HttpServer::new(move || {
let settings = Settings::get(); let settings = Settings::get();
let rate_limiter = rate_limiter.clone();
App::new() App::new()
.wrap(middleware::Logger::default()) .wrap(middleware::Logger::default())
.data(pool.clone()) .data(pool.clone())
.data(server.clone()) .data(server.clone())
.data(rate_limiter.clone())
// The routes // The routes
.configure(api::config) .configure(move |cfg| api::config(cfg, &rate_limiter))
.configure(federation::config) .configure(federation::config)
.configure(feeds::config) .configure(feeds::config)
.configure(index::config) .configure(index::config)

View File

@ -2,17 +2,180 @@ pub mod rate_limiter;
use super::{IPAddr, Settings}; use super::{IPAddr, Settings};
use crate::api::APIError; use crate::api::APIError;
use crate::settings::RateLimitConfig;
use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform};
use failure::Error; use failure::Error;
use futures::future::{ok, Ready};
use log::warn; use log::warn;
use rate_limiter::RateLimiter; use rate_limiter::{RateLimitType, RateLimiter};
use std::collections::HashMap; use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc; use std::sync::Arc;
use std::sync::Mutex; use std::task::{Context, Poll};
use std::time::SystemTime; use std::time::SystemTime;
use strum::IntoEnumIterator; use strum::IntoEnumIterator;
use tokio::sync::Mutex;
#[derive(Debug, Clone)] #[derive(Debug, Clone)]
pub struct RateLimitInfo { pub struct RateLimit(pub Arc<Mutex<RateLimiter>>);
pub rate_limiter: Arc<Mutex<RateLimiter>>,
pub ip: IPAddr, #[derive(Debug, Clone)]
pub struct RateLimited(Arc<Mutex<RateLimiter>>, RateLimitType);
pub struct RateLimitedMiddleware<S>(RateLimited, S);
impl RateLimit {
pub fn message(&self) -> RateLimited {
self.kind(RateLimitType::Message)
}
pub fn post(&self) -> RateLimited {
self.kind(RateLimitType::Post)
}
pub fn register(&self) -> RateLimited {
self.kind(RateLimitType::Register)
}
fn kind(&self, type_: RateLimitType) -> RateLimited {
RateLimited(self.0.clone(), type_)
}
}
impl RateLimited {
pub async fn wrap<T, E>(
self,
ip_addr: String,
fut: impl Future<Output = Result<T, E>>,
) -> Result<T, E>
where
E: From<failure::Error>,
{
let rate_limit: RateLimitConfig = actix_web::web::block(move || {
// needs to be in a web::block because the RwLock in settings is from stdlib
Ok(Settings::get().rate_limit.clone()) as Result<_, failure::Error>
})
.await
.map_err(|e| match e {
actix_web::error::BlockingError::Error(e) => e,
_ => APIError::err("Operation canceled").into(),
})?;
// before
{
let mut limiter = self.0.lock().await;
match self.1 {
RateLimitType::Message => {
limiter.check_rate_limit_full(
self.1,
&ip_addr,
rate_limit.message,
rate_limit.message_per_second,
false,
)?;
return fut.await;
}
RateLimitType::Post => {
limiter.check_rate_limit_full(
self.1.clone(),
&ip_addr,
rate_limit.post,
rate_limit.post_per_second,
true,
)?;
}
RateLimitType::Register => {
limiter.check_rate_limit_full(
self.1,
&ip_addr,
rate_limit.register,
rate_limit.register_per_second,
true,
)?;
}
};
}
let res = fut.await;
// after
{
let mut limiter = self.0.lock().await;
if res.is_ok() {
match self.1 {
RateLimitType::Post => {
limiter.check_rate_limit_full(
self.1,
&ip_addr,
rate_limit.post,
rate_limit.post_per_second,
false,
)?;
}
RateLimitType::Register => {
limiter.check_rate_limit_full(
self.1,
&ip_addr,
rate_limit.register,
rate_limit.register_per_second,
false,
)?;
}
_ => (),
};
}
}
res
}
}
impl<S> Transform<S> for RateLimited
where
S: Service<Request = ServiceRequest, Response = ServiceResponse, Error = actix_web::Error>,
S::Future: 'static,
{
type Request = S::Request;
type Response = S::Response;
type Error = actix_web::Error;
type InitError = ();
type Transform = RateLimitedMiddleware<S>;
type Future = Ready<Result<Self::Transform, Self::InitError>>;
fn new_transform(&self, service: S) -> Self::Future {
ok(RateLimitedMiddleware(self.clone(), service))
}
}
impl<S> Service for RateLimitedMiddleware<S>
where
S: Service<Request = ServiceRequest, Response = ServiceResponse, Error = actix_web::Error>,
S::Future: 'static,
{
type Request = S::Request;
type Response = S::Response;
type Error = actix_web::Error;
type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.1.poll_ready(cx)
}
fn call(&mut self, req: S::Request) -> Self::Future {
let ip_addr = req
.connection_info()
.remote()
.unwrap_or("127.0.0.1:12345")
.split(':')
.next()
.unwrap_or("127.0.0.1")
.to_string();
let fut = self.0.clone().wrap(ip_addr, self.1.call(req));
Box::pin(async move { fut.await.map_err(actix_web::Error::from) })
}
} }

View File

@ -79,7 +79,7 @@ impl RateLimiter {
} }
#[allow(clippy::float_cmp)] #[allow(clippy::float_cmp)]
fn check_rate_limit_full( pub(super) fn check_rate_limit_full(
&mut self, &mut self,
type_: RateLimitType, type_: RateLimitType,
ip: &str, ip: &str,

View File

@ -4,119 +4,158 @@ use crate::api::community::*;
use crate::api::post::*; use crate::api::post::*;
use crate::api::site::*; use crate::api::site::*;
use crate::api::user::*; use crate::api::user::*;
use crate::rate_limit::RateLimit;
use actix_web::guard;
#[rustfmt::skip] #[rustfmt::skip]
pub fn config(cfg: &mut web::ServiceConfig) { pub fn config(cfg: &mut web::ServiceConfig, rate_limit: &RateLimit) {
cfg cfg.service(
// Site web::scope("/api/v1")
.route("/api/v1/site", web::get().to(route_get::<GetSite, GetSiteResponse>)) // Site
.route("/api/v1/categories", web::get().to(route_get::<ListCategories, ListCategoriesResponse>)) .service(
.route("/api/v1/modlog", web::get().to(route_get::<GetModlog, GetModlogResponse>)) web::scope("/site")
.route("/api/v1/search", web::get().to(route_get::<Search, SearchResponse>)) .wrap(rate_limit.message())
// Community .route("", web::get().to(route_get::<GetSite>))
.route("/api/v1/community", web::post().to(route_post::<CreateCommunity, CommunityResponse>)) // Admin Actions
.route("/api/v1/community", web::get().to(route_get::<GetCommunity, GetCommunityResponse>)) .route("", web::post().to(route_post::<CreateSite>))
.route("/api/v1/community", web::put().to(route_post::<EditCommunity, CommunityResponse>)) .route("", web::put().to(route_post::<EditSite>))
.route("/api/v1/community/list", web::get().to(route_get::<ListCommunities, ListCommunitiesResponse>)) .route("/transfer", web::post().to(route_post::<TransferSite>))
.route("/api/v1/community/follow", web::post().to(route_post::<FollowCommunity, CommunityResponse>)) .route("/config", web::get().to(route_get::<GetSiteConfig>))
// Post .route("/config", web::put().to(route_post::<SaveSiteConfig>)),
.route("/api/v1/post", web::post().to(route_post::<CreatePost, PostResponse>)) )
.route("/api/v1/post", web::put().to(route_post::<EditPost, PostResponse>)) .service(
.route("/api/v1/post", web::get().to(route_get::<GetPost, GetPostResponse>)) web::resource("/categories")
.route("/api/v1/post/list", web::get().to(route_get::<GetPosts, GetPostsResponse>)) .wrap(rate_limit.message())
.route("/api/v1/post/like", web::post().to(route_post::<CreatePostLike, PostResponse>)) .route(web::get().to(route_get::<ListCategories>)),
.route("/api/v1/post/save", web::put().to(route_post::<SavePost, PostResponse>)) )
// Comment .service(
.route("/api/v1/comment", web::post().to(route_post::<CreateComment, CommentResponse>)) web::resource("/modlog")
.route("/api/v1/comment", web::put().to(route_post::<EditComment, CommentResponse>)) .wrap(rate_limit.message())
.route("/api/v1/comment/like", web::post().to(route_post::<CreateCommentLike, CommentResponse>)) .route(web::get().to(route_get::<GetModlog>)),
.route("/api/v1/comment/save", web::put().to(route_post::<SaveComment, CommentResponse>)) )
// User .service(
.route("/api/v1/user", web::get().to(route_get::<GetUserDetails, GetUserDetailsResponse>)) web::resource("/search")
.route("/api/v1/user/mention", web::get().to(route_get::<GetUserMentions, GetUserMentionsResponse>)) .wrap(rate_limit.message())
.route("/api/v1/user/mention", web::put().to(route_post::<EditUserMention, UserMentionResponse>)) .route(web::get().to(route_get::<Search>)),
.route("/api/v1/user/replies", web::get().to(route_get::<GetReplies, GetRepliesResponse>)) )
.route("/api/v1/user/followed_communities", web::get().to(route_get::<GetFollowedCommunities, GetFollowedCommunitiesResponse>)) // Community
// Mod actions .service(
.route("/api/v1/community/transfer", web::post().to(route_post::<TransferCommunity, GetCommunityResponse>)) web::scope("/community")
.route("/api/v1/community/ban_user", web::post().to(route_post::<BanFromCommunity, BanFromCommunityResponse>)) .wrap(rate_limit.message())
.route("/api/v1/community/mod", web::post().to(route_post::<AddModToCommunity, AddModToCommunityResponse>)) .route("", web::post().to(route_post::<CreateCommunity>))
// Admin actions .route("", web::get().to(route_get::<GetCommunity>))
.route("/api/v1/site", web::post().to(route_post::<CreateSite, SiteResponse>)) .route("", web::put().to(route_post::<EditCommunity>))
.route("/api/v1/site", web::put().to(route_post::<EditSite, SiteResponse>)) .route("/list", web::get().to(route_get::<ListCommunities>))
.route("/api/v1/site/transfer", web::post().to(route_post::<TransferSite, GetSiteResponse>)) .route("/follow", web::post().to(route_post::<FollowCommunity>))
.route("/api/v1/site/config", web::get().to(route_get::<GetSiteConfig, GetSiteConfigResponse>)) // Mod Actions
.route("/api/v1/site/config", web::put().to(route_post::<SaveSiteConfig, GetSiteConfigResponse>)) .route("/transfer", web::post().to(route_post::<TransferCommunity>))
.route("/api/v1/admin/add", web::post().to(route_post::<AddAdmin, AddAdminResponse>)) .route("/ban_user", web::post().to(route_post::<BanFromCommunity>))
.route("/api/v1/user/ban", web::post().to(route_post::<BanUser, BanUserResponse>)) .route("/mod", web::post().to(route_post::<AddModToCommunity>)),
// User account actions )
.route("/api/v1/user/login", web::post().to(route_post::<Login, LoginResponse>)) // Post
.route("/api/v1/user/register", web::post().to(route_post::<Register, LoginResponse>)) .service(
.route("/api/v1/user/delete_account", web::post().to(route_post::<DeleteAccount, LoginResponse>)) // Handle POST to /post separately to add the post() rate limitter
.route("/api/v1/user/password_reset", web::post().to(route_post::<PasswordReset, PasswordResetResponse>)) web::resource("/post")
.route("/api/v1/user/password_change", web::post().to(route_post::<PasswordChange, LoginResponse>)) .guard(guard::Post())
.route("/api/v1/user/mark_all_as_read", web::post().to(route_post::<MarkAllAsRead, GetRepliesResponse>)) .wrap(rate_limit.post())
.route("/api/v1/user/save_user_settings", web::put().to(route_post::<SaveUserSettings, LoginResponse>)); .route(web::post().to(route_post::<CreatePost>)),
)
.service(
web::scope("/post")
.wrap(rate_limit.message())
.route("", web::get().to(route_get::<GetPost>))
.route("", web::put().to(route_post::<EditPost>))
.route("/list", web::get().to(route_get::<GetPosts>))
.route("/like", web::post().to(route_post::<CreatePostLike>))
.route("/save", web::put().to(route_post::<SavePost>)),
)
// Comment
.service(
web::scope("/comment")
.wrap(rate_limit.message())
.route("", web::post().to(route_post::<CreateComment>))
.route("", web::put().to(route_post::<EditComment>))
.route("/like", web::post().to(route_post::<CreateCommentLike>))
.route("/save", web::put().to(route_post::<SaveComment>)),
)
// User
.service(
// Account action, I don't like that it's in /user maybe /accounts
// Handle /user/register separately to add the register() rate limitter
web::resource("/user/register")
.guard(guard::Post())
.wrap(rate_limit.register())
.route(web::post().to(route_post::<Register>)),
)
// User actions
.service(
web::scope("/user")
.wrap(rate_limit.message())
.route("", web::get().to(route_get::<GetUserDetails>))
.route("/mention", web::get().to(route_get::<GetUserMentions>))
.route("/mention", web::put().to(route_post::<EditUserMention>))
.route("/replies", web::get().to(route_get::<GetReplies>))
.route("/followed_communities", web::get().to(route_get::<GetFollowedCommunities>))
// Admin action. I don't like that it's in /user
.route("/ban", web::post().to(route_post::<BanUser>))
// Account actions. I don't like that they're in /user maybe /accounts
.route("/login", web::post().to(route_post::<Login>))
.route("/delete_account", web::post().to(route_post::<DeleteAccount>))
.route("/password_reset", web::post().to(route_post::<PasswordReset>))
.route("/password_change", web::post().to(route_post::<PasswordChange>))
// mark_all_as_read feels off being in this section as well
.route("/mark_all_as_read", web::post().to(route_post::<MarkAllAsRead>))
.route("/save_user_settings", web::put().to(route_post::<SaveUserSettings>)),
)
// Admin Actions
.service(
web::resource("/admin/add")
.wrap(rate_limit.message())
.route(web::post().to(route_post::<AddAdmin>)),
),
);
} }
fn perform<Request, Response>( fn perform<Request>(
data: Request, data: Request,
db: DbPoolParam, db: DbPoolParam,
rate_limit_param: RateLimitParam,
chat_server: ChatServerParam, chat_server: ChatServerParam,
req: HttpRequest,
) -> Result<HttpResponse, Error> ) -> Result<HttpResponse, Error>
where where
Response: Serialize, Oper<Request>: Perform,
Oper<Request>: Perform<Response>,
{ {
let ws_info = WebsocketInfo { let ws_info = WebsocketInfo {
chatserver: chat_server.get_ref().to_owned(), chatserver: chat_server.get_ref().to_owned(),
id: None, id: None,
}; };
let rate_limit_info = RateLimitInfo {
rate_limiter: rate_limit_param.get_ref().to_owned(),
ip: get_ip(&req),
};
let oper: Oper<Request> = Oper::new(data); let oper: Oper<Request> = Oper::new(data);
let res = oper.perform( let res = oper.perform(db.get_ref().to_owned(), Some(ws_info));
db.get_ref().to_owned(),
Some(ws_info),
Some(rate_limit_info),
);
Ok(HttpResponse::Ok().json(res?)) Ok(HttpResponse::Ok().json(res?))
} }
async fn route_get<Data, Response>( async fn route_get<Data>(
data: web::Query<Data>, data: web::Query<Data>,
db: DbPoolParam, db: DbPoolParam,
rate_limit_param: RateLimitParam,
chat_server: ChatServerParam, chat_server: ChatServerParam,
req: HttpRequest,
) -> Result<HttpResponse, Error> ) -> Result<HttpResponse, Error>
where where
Data: Serialize, Data: Serialize,
Response: Serialize, Oper<Data>: Perform,
Oper<Data>: Perform<Response>,
{ {
perform::<Data, Response>(data.0, db, rate_limit_param, chat_server, req) perform::<Data>(data.0, db, chat_server)
} }
async fn route_post<Data, Response>( async fn route_post<Data>(
data: web::Json<Data>, data: web::Json<Data>,
db: DbPoolParam, db: DbPoolParam,
rate_limit_param: RateLimitParam,
chat_server: ChatServerParam, chat_server: ChatServerParam,
req: HttpRequest,
) -> Result<HttpResponse, Error> ) -> Result<HttpResponse, Error>
where where
Data: Serialize, Data: Serialize,
Response: Serialize, Oper<Data>: Perform,
Oper<Data>: Perform<Response>,
{ {
perform::<Data, Response>(data.0, db, rate_limit_param, chat_server, req) perform::<Data>(data.0, db, chat_server)
} }

View File

@ -1,6 +1,6 @@
use crate::api::{Oper, Perform}; use crate::api::{Oper, Perform};
use crate::db::site_view::SiteView; use crate::db::site_view::SiteView;
use crate::rate_limit::{rate_limiter::RateLimiter, RateLimitInfo}; use crate::rate_limit::rate_limiter::RateLimiter;
use crate::websocket::{server::ChatServer, WebsocketInfo}; use crate::websocket::{server::ChatServer, WebsocketInfo};
use crate::{get_ip, markdown_to_html, version, Settings}; use crate::{get_ip, markdown_to_html, version, Settings};
use actix::prelude::*; use actix::prelude::*;

View File

@ -123,10 +123,9 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WSSession {
.into_actor(self) .into_actor(self)
.then(|res, _, ctx| { .then(|res, _, ctx| {
match res { match res {
Ok(res) => ctx.text(res), Ok(Ok(res)) => ctx.text(res),
Err(e) => { Ok(Err(e)) => error!("{}", e),
error!("{}", &e); Err(e) => error!("{}", &e),
}
} }
actix::fut::ready(()) actix::fut::ready(())
}) })

View File

@ -12,8 +12,6 @@ use serde_json::Value;
use server::ChatServer; use server::ChatServer;
use std::collections::{HashMap, HashSet}; use std::collections::{HashMap, HashSet};
use std::str::FromStr; use std::str::FromStr;
use std::sync::Arc;
use std::sync::Mutex;
#[derive(EnumString, ToString, Debug, Clone)] #[derive(EnumString, ToString, Debug, Clone)]
pub enum UserOperation { pub enum UserOperation {

View File

@ -9,7 +9,7 @@ use crate::api::post::*;
use crate::api::site::*; use crate::api::site::*;
use crate::api::user::*; use crate::api::user::*;
use crate::api::*; use crate::api::*;
use crate::rate_limit::{rate_limiter::RateLimiter, RateLimitInfo}; use crate::rate_limit::RateLimit;
use crate::websocket::UserOperation; use crate::websocket::UserOperation;
use crate::{CommunityId, ConnectionId, IPAddr, PostId, UserId}; use crate::{CommunityId, ConnectionId, IPAddr, PostId, UserId};
@ -38,7 +38,7 @@ pub struct Disconnect {
/// The messages sent to websocket clients /// The messages sent to websocket clients
#[derive(Serialize, Deserialize, Message)] #[derive(Serialize, Deserialize, Message)]
#[rtype(String)] #[rtype(result = "Result<String, failure::Error>")]
pub struct StandardMessage { pub struct StandardMessage {
/// Id of the client session /// Id of the client session
pub id: ConnectionId, pub id: ConnectionId,
@ -152,13 +152,13 @@ pub struct ChatServer {
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
/// Rate limiting based on rate type and IP addr /// Rate limiting based on rate type and IP addr
rate_limiter: Arc<Mutex<RateLimiter>>, rate_limiter: RateLimit,
} }
impl ChatServer { impl ChatServer {
pub fn startup( pub fn startup(
pool: Pool<ConnectionManager<PgConnection>>, pool: Pool<ConnectionManager<PgConnection>>,
rate_limiter: Arc<Mutex<RateLimiter>>, rate_limiter: RateLimit,
) -> ChatServer { ) -> ChatServer {
ChatServer { ChatServer {
sessions: HashMap::new(), sessions: HashMap::new(),
@ -389,352 +389,526 @@ impl ChatServer {
} }
} }
fn do_user_operation<'a, Data, Response>(
&self,
id: ConnectionId,
ip: IPAddr,
op: UserOperation,
data: &str,
ctx: &mut Context<Self>,
) -> Result<String, Error>
where
for<'de> Data: Deserialize<'de> + 'a,
Response: Serialize,
Oper<Data>: Perform<Response>,
{
let parsed_data: Data = serde_json::from_str(data)?;
let ws_info = WebsocketInfo {
chatserver: ctx.address(),
id: Some(id),
};
let rate_limit_info = RateLimitInfo {
rate_limiter: self.rate_limiter.clone(),
ip,
};
let new_pool = self.pool.clone();
let res = Oper::new(parsed_data).perform(new_pool, Some(ws_info), Some(rate_limit_info))?;
to_json_string(&op, &res)
}
fn parse_json_message( fn parse_json_message(
&mut self, &mut self,
msg: StandardMessage, msg: StandardMessage,
ctx: &mut Context<Self>, ctx: &mut Context<Self>,
) -> Result<String, Error> { ) -> impl Future<Output = Result<String, Error>> {
let json: Value = serde_json::from_str(&msg.msg)?; let addr = ctx.address();
let data = &json["data"].to_string(); let pool = self.pool.clone();
let op = &json["op"].as_str().ok_or(APIError { let rate_limiter = self.rate_limiter.clone();
message: "Unknown op type".to_string(),
})?;
let user_operation: UserOperation = UserOperation::from_str(&op)?;
let ip: IPAddr = match self.sessions.get(&msg.id) { let ip: IPAddr = match self.sessions.get(&msg.id) {
Some(info) => info.ip.to_owned(), Some(info) => info.ip.to_owned(),
None => "blank_ip".to_string(), None => "blank_ip".to_string(),
}; };
match user_operation { async move {
// User ops let msg = msg;
UserOperation::Login => { let json: Value = serde_json::from_str(&msg.msg)?;
self.do_user_operation::<Login, LoginResponse>(msg.id, ip, user_operation, data, ctx) let data = &json["data"].to_string();
} let op = &json["op"].as_str().ok_or(APIError {
UserOperation::Register => { message: "Unknown op type".to_string(),
self.do_user_operation::<Register, LoginResponse>(msg.id, ip, user_operation, data, ctx) })?;
}
UserOperation::GetUserDetails => self
.do_user_operation::<GetUserDetails, GetUserDetailsResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::GetReplies => self.do_user_operation::<GetReplies, GetRepliesResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::AddAdmin => {
self.do_user_operation::<AddAdmin, AddAdminResponse>(msg.id, ip, user_operation, data, ctx)
}
UserOperation::BanUser => {
self.do_user_operation::<BanUser, BanUserResponse>(msg.id, ip, user_operation, data, ctx)
}
UserOperation::GetUserMentions => self
.do_user_operation::<GetUserMentions, GetUserMentionsResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::EditUserMention => self
.do_user_operation::<EditUserMention, UserMentionResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::MarkAllAsRead => self.do_user_operation::<MarkAllAsRead, GetRepliesResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::DeleteAccount => self.do_user_operation::<DeleteAccount, LoginResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::PasswordReset => self
.do_user_operation::<PasswordReset, PasswordResetResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::PasswordChange => self.do_user_operation::<PasswordChange, LoginResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::CreatePrivateMessage => self
.do_user_operation::<CreatePrivateMessage, PrivateMessageResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::EditPrivateMessage => self
.do_user_operation::<EditPrivateMessage, PrivateMessageResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::GetPrivateMessages => self
.do_user_operation::<GetPrivateMessages, PrivateMessagesResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::UserJoin => {
self.do_user_operation::<UserJoin, UserJoinResponse>(msg.id, ip, user_operation, data, ctx)
}
UserOperation::SaveUserSettings => self.do_user_operation::<SaveUserSettings, LoginResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
// Site ops let user_operation: UserOperation = UserOperation::from_str(&op)?;
UserOperation::GetModlog => self.do_user_operation::<GetModlog, GetModlogResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::CreateSite => {
self.do_user_operation::<CreateSite, SiteResponse>(msg.id, ip, user_operation, data, ctx)
}
UserOperation::EditSite => {
self.do_user_operation::<EditSite, SiteResponse>(msg.id, ip, user_operation, data, ctx)
}
UserOperation::GetSite => {
self.do_user_operation::<GetSite, GetSiteResponse>(msg.id, ip, user_operation, data, ctx)
}
UserOperation::GetSiteConfig => self
.do_user_operation::<GetSiteConfig, GetSiteConfigResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::SaveSiteConfig => self
.do_user_operation::<SaveSiteConfig, GetSiteConfigResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::Search => {
self.do_user_operation::<Search, SearchResponse>(msg.id, ip, user_operation, data, ctx)
}
UserOperation::TransferCommunity => self
.do_user_operation::<TransferCommunity, GetCommunityResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::TransferSite => self.do_user_operation::<TransferSite, GetSiteResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
UserOperation::ListCategories => self
.do_user_operation::<ListCategories, ListCategoriesResponse>(
msg.id,
ip,
user_operation,
data,
ctx,
),
// Community ops match user_operation {
UserOperation::GetCommunity => self.do_user_operation::<GetCommunity, GetCommunityResponse>( // User ops
msg.id, UserOperation::Login => {
ip, do_user_operation::<Login>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
user_operation, .await
data, }
ctx, UserOperation::Register => {
), do_user_operation::<Register>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
UserOperation::ListCommunities => self .await
.do_user_operation::<ListCommunities, ListCommunitiesResponse>( }
msg.id, UserOperation::GetUserDetails => {
ip, do_user_operation::<GetUserDetails>(
user_operation, pool,
data, rate_limiter,
ctx, addr,
), msg.id,
UserOperation::CreateCommunity => self ip,
.do_user_operation::<CreateCommunity, CommunityResponse>( user_operation,
msg.id, data,
ip, )
user_operation, .await
data, }
ctx, UserOperation::GetReplies => {
), do_user_operation::<GetReplies>(
UserOperation::EditCommunity => self.do_user_operation::<EditCommunity, CommunityResponse>( pool,
msg.id, rate_limiter,
ip, addr,
user_operation, msg.id,
data, ip,
ctx, user_operation,
), data,
UserOperation::FollowCommunity => self )
.do_user_operation::<FollowCommunity, CommunityResponse>( .await
msg.id, }
ip, UserOperation::AddAdmin => {
user_operation, do_user_operation::<AddAdmin>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
data, .await
ctx, }
), UserOperation::BanUser => {
UserOperation::GetFollowedCommunities => self do_user_operation::<BanUser>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.do_user_operation::<GetFollowedCommunities, GetFollowedCommunitiesResponse>( .await
msg.id, }
ip, UserOperation::GetUserMentions => {
user_operation, do_user_operation::<GetUserMentions>(
data, pool,
ctx, rate_limiter,
), addr,
UserOperation::BanFromCommunity => self msg.id,
.do_user_operation::<BanFromCommunity, BanFromCommunityResponse>( ip,
msg.id, user_operation,
ip, data,
user_operation, )
data, .await
ctx, }
), UserOperation::EditUserMention => {
UserOperation::AddModToCommunity => self do_user_operation::<EditUserMention>(
.do_user_operation::<AddModToCommunity, AddModToCommunityResponse>( pool,
msg.id, rate_limiter,
ip, addr,
user_operation, msg.id,
data, ip,
ctx, user_operation,
), data,
)
.await
}
UserOperation::MarkAllAsRead => {
do_user_operation::<MarkAllAsRead>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::DeleteAccount => {
do_user_operation::<DeleteAccount>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::PasswordReset => {
do_user_operation::<PasswordReset>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::PasswordChange => {
do_user_operation::<PasswordChange>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::CreatePrivateMessage => {
do_user_operation::<CreatePrivateMessage>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::EditPrivateMessage => {
do_user_operation::<EditPrivateMessage>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::GetPrivateMessages => {
do_user_operation::<GetPrivateMessages>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::UserJoin => {
do_user_operation::<UserJoin>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.await
}
UserOperation::SaveUserSettings => {
do_user_operation::<SaveUserSettings>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
// Post ops // Site ops
UserOperation::CreatePost => { UserOperation::GetModlog => {
self.do_user_operation::<CreatePost, PostResponse>(msg.id, ip, user_operation, data, ctx) do_user_operation::<GetModlog>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
} .await
UserOperation::GetPost => { }
self.do_user_operation::<GetPost, GetPostResponse>(msg.id, ip, user_operation, data, ctx) UserOperation::CreateSite => {
} do_user_operation::<CreateSite>(
UserOperation::GetPosts => { pool,
self.do_user_operation::<GetPosts, GetPostsResponse>(msg.id, ip, user_operation, data, ctx) rate_limiter,
} addr,
UserOperation::EditPost => { msg.id,
self.do_user_operation::<EditPost, PostResponse>(msg.id, ip, user_operation, data, ctx) ip,
} user_operation,
UserOperation::CreatePostLike => self.do_user_operation::<CreatePostLike, PostResponse>( data,
msg.id, )
ip, .await
user_operation, }
data, UserOperation::EditSite => {
ctx, do_user_operation::<EditSite>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
), .await
UserOperation::SavePost => { }
self.do_user_operation::<SavePost, PostResponse>(msg.id, ip, user_operation, data, ctx) UserOperation::GetSite => {
} do_user_operation::<GetSite>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.await
}
UserOperation::GetSiteConfig => {
do_user_operation::<GetSiteConfig>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::SaveSiteConfig => {
do_user_operation::<SaveSiteConfig>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::Search => {
do_user_operation::<Search>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.await
}
UserOperation::TransferCommunity => {
do_user_operation::<TransferCommunity>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::TransferSite => {
do_user_operation::<TransferSite>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::ListCategories => {
do_user_operation::<ListCategories>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
// Comment ops // Community ops
UserOperation::CreateComment => self.do_user_operation::<CreateComment, CommentResponse>( UserOperation::GetCommunity => {
msg.id, do_user_operation::<GetCommunity>(
ip, pool,
user_operation, rate_limiter,
data, addr,
ctx, msg.id,
), ip,
UserOperation::EditComment => self.do_user_operation::<EditComment, CommentResponse>( user_operation,
msg.id, data,
ip, )
user_operation, .await
data, }
ctx, UserOperation::ListCommunities => {
), do_user_operation::<ListCommunities>(
UserOperation::SaveComment => self.do_user_operation::<SaveComment, CommentResponse>( pool,
msg.id, rate_limiter,
ip, addr,
user_operation, msg.id,
data, ip,
ctx, user_operation,
), data,
UserOperation::GetComments => self.do_user_operation::<GetComments, GetCommentsResponse>( )
msg.id, .await
ip, }
user_operation, UserOperation::CreateCommunity => {
data, do_user_operation::<CreateCommunity>(
ctx, pool,
), rate_limiter,
UserOperation::CreateCommentLike => self addr,
.do_user_operation::<CreateCommentLike, CommentResponse>( msg.id,
msg.id, ip,
ip, user_operation,
user_operation, data,
data, )
ctx, .await
), }
UserOperation::EditCommunity => {
do_user_operation::<EditCommunity>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::FollowCommunity => {
do_user_operation::<FollowCommunity>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::GetFollowedCommunities => {
do_user_operation::<GetFollowedCommunities>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::BanFromCommunity => {
do_user_operation::<BanFromCommunity>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::AddModToCommunity => {
do_user_operation::<AddModToCommunity>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
// Post ops
UserOperation::CreatePost => {
do_user_operation::<CreatePost>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::GetPost => {
do_user_operation::<GetPost>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.await
}
UserOperation::GetPosts => {
do_user_operation::<GetPosts>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.await
}
UserOperation::EditPost => {
do_user_operation::<EditPost>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.await
}
UserOperation::CreatePostLike => {
do_user_operation::<CreatePostLike>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::SavePost => {
do_user_operation::<SavePost>(pool, rate_limiter, addr, msg.id, ip, user_operation, data)
.await
}
// Comment ops
UserOperation::CreateComment => {
do_user_operation::<CreateComment>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::EditComment => {
do_user_operation::<EditComment>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::SaveComment => {
do_user_operation::<SaveComment>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::GetComments => {
do_user_operation::<GetComments>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
UserOperation::CreateCommentLike => {
do_user_operation::<CreateCommentLike>(
pool,
rate_limiter,
addr,
msg.id,
ip,
user_operation,
data,
)
.await
}
}
} }
} }
} }
async fn do_user_operation<'a, Data>(
pool: Pool<ConnectionManager<PgConnection>>,
rate_limiter: RateLimit,
chatserver: Addr<ChatServer>,
id: ConnectionId,
ip: IPAddr,
op: UserOperation,
data: &str,
) -> Result<String, Error>
where
for<'de> Data: Deserialize<'de> + 'a,
Oper<Data>: Perform,
{
let ws_info = WebsocketInfo {
chatserver,
id: Some(id),
};
let data = data.to_string();
let op2 = op.clone();
let fut = async move {
let parsed_data: Data = serde_json::from_str(&data)?;
let res = Oper::new(parsed_data).perform(pool, Some(ws_info))?;
to_json_string(&op, &res)
};
match op2 {
UserOperation::Register => rate_limiter.register().wrap(ip, fut).await,
UserOperation::CreatePost => rate_limiter.post().wrap(ip, fut).await,
_ => rate_limiter.message().wrap(ip, fut).await,
}
}
/// Make actor from `ChatServer` /// Make actor from `ChatServer`
impl Actor for ChatServer { impl Actor for ChatServer {
/// We are going to use simple Context, we just need ability to communicate /// We are going to use simple Context, we just need ability to communicate
@ -789,19 +963,22 @@ impl Handler<Disconnect> for ChatServer {
/// Handler for Message message. /// Handler for Message message.
impl Handler<StandardMessage> for ChatServer { impl Handler<StandardMessage> for ChatServer {
type Result = MessageResult<StandardMessage>; type Result = ResponseFuture<Result<String, failure::Error>>;
fn handle(&mut self, msg: StandardMessage, ctx: &mut Context<Self>) -> Self::Result { fn handle(&mut self, msg: StandardMessage, ctx: &mut Context<Self>) -> Self::Result {
match self.parse_json_message(msg, ctx) { let fut = self.parse_json_message(msg, ctx);
Ok(m) => { Box::pin(async move {
info!("Message Sent: {}", m); match fut.await {
MessageResult(m) Ok(m) => {
info!("Message Sent: {}", m);
Ok(m)
}
Err(e) => {
error!("Error during message handling {}", e);
Ok(e.to_string())
}
} }
Err(e) => { })
error!("Error during message handling {}", e);
MessageResult(e.to_string())
}
}
} }
} }