grishka/lemmy
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Change RateLimit to act as a middleware

Browse source
This commit is contained in:
asonix 2020-04-19 22:59:07 -05:00
parent 6f14f0756b
commit b55e9c08df
16 changed files with 962 additions and 881 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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)",
"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)",
"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)",
"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)",
@ -1427,6 +1428,7 @@ dependencies = [
"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_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]]

2
server/Cargo.toml vendored
View file

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

60
server/src/api/comment.rs
View file

@ -59,12 +59,13 @@ pub struct GetCommentsResponse {
comments: Vec<CommentView>,
}
impl Perform<CommentResponse> for Oper<CreateComment> {
impl Perform for Oper<CreateComment> {
type Response = CommentResponse;
fn perform(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> {
let data: &CreateComment = &self.data;
@ -77,13 +78,6 @@ impl Perform<CommentResponse> for Oper<CreateComment> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> {
let data: &EditComment = &self.data;
@ -269,13 +264,6 @@ impl Perform<CommentResponse> for Oper<EditComment> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> {
let data: &SaveComment = &self.data;
@ -432,13 +421,6 @@ impl Perform<CommentResponse> for Oper<SaveComment> {
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()?;
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommentResponse, Error> {
let data: &CreateCommentLike = &self.data;
@ -480,13 +463,6 @@ impl Perform<CommentResponse> for Oper<CreateCommentLike> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetCommentsResponse, Error> {
let data: &GetComments = &self.data;
@ -592,13 +569,6 @@ impl Perform<GetCommentsResponse> for Oper<GetComments> {
let type_ = ListingType::from_str(&data.type_)?;
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 comments = match CommentQueryBuilder::create(&conn)

115
server/src/api/community.rs
View file

@ -111,12 +111,13 @@ pub struct TransferCommunity {
auth: String,
}
impl Perform<GetCommunityResponse> for Oper<GetCommunity> {
impl Perform for Oper<GetCommunity> {
type Response = GetCommunityResponse;
fn perform(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetCommunityResponse, Error> {
let data: &GetCommunity = &self.data;
@ -131,13 +132,6 @@ impl Perform<GetCommunityResponse> for Oper<GetCommunity> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommunityResponse, Error> {
let data: &CreateCommunity = &self.data;
@ -227,13 +222,6 @@ impl Perform<CommunityResponse> for Oper<CreateCommunity> {
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()?;
// 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))?;
if let Some(rl) = rate_limit_info {
rl.rate_limiter
.lock()
.unwrap()
.check_rate_limit_register(&rl.ip, false)?;
}
Ok(CommunityResponse {
community: community_view,
})
}
}
impl Perform<CommunityResponse> for Oper<EditCommunity> {
impl Perform for Oper<EditCommunity> {
type Response = CommunityResponse;
fn perform(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommunityResponse, Error> {
let data: &EditCommunity = &self.data;
@ -326,13 +308,6 @@ impl Perform<CommunityResponse> for Oper<EditCommunity> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<ListCommunitiesResponse, Error> {
let data: &ListCommunities = &self.data;
@ -439,13 +415,6 @@ impl Perform<ListCommunitiesResponse> for Oper<ListCommunities> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<CommunityResponse, Error> {
let data: &FollowCommunity = &self.data;
@ -482,13 +452,6 @@ impl Perform<CommunityResponse> for Oper<FollowCommunity> {
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()?;
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetFollowedCommunitiesResponse, Error> {
let data: &GetFollowedCommunities = &self.data;
@ -527,13 +491,6 @@ impl Perform<GetFollowedCommunitiesResponse> for Oper<GetFollowedCommunities> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<BanFromCommunityResponse, Error> {
let data: &BanFromCommunity = &self.data;
@ -568,13 +526,6 @@ impl Perform<BanFromCommunityResponse> for Oper<BanFromCommunity> {
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()?;
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<AddModToCommunityResponse, Error> {
let data: &AddModToCommunity = &self.data;
@ -646,13 +598,6 @@ impl Perform<AddModToCommunityResponse> for Oper<AddModToCommunity> {
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()?;
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetCommunityResponse, Error> {
let data: &TransferCommunity = &self.data;
@ -709,13 +655,6 @@ impl Perform<GetCommunityResponse> for Oper<TransferCommunity> {
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 read_community = Community::read(&conn, data.community_id)?;

10
server/src/api/mod.rs
View file

@ -22,7 +22,6 @@ use crate::{
naive_now, remove_slurs, send_email, slur_check, slurs_vec_to_str,
};
use crate::rate_limit::RateLimitInfo;
use crate::settings::Settings;
use crate::websocket::UserOperation;
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<T, Error>
where
T: Sized;
) -> Result<Self::Response, Error>;
}

79
server/src/api/post.rs
View file

@ -77,12 +77,13 @@ pub struct SavePost {
auth: String,
}
impl Perform<PostResponse> for Oper<CreatePost> {
impl Perform for Oper<CreatePost> {
type Response = PostResponse;
fn perform(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> {
let data: &CreatePost = &self.data;
@ -103,13 +104,6 @@ impl Perform<PostResponse> for Oper<CreatePost> {
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()?;
// 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()),
};
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 };
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetPostResponse, Error> {
let data: &GetPost = &self.data;
@ -217,13 +205,6 @@ impl Perform<GetPostResponse> for Oper<GetPost> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetPostsResponse, Error> {
let data: &GetPosts = &self.data;
@ -307,13 +289,6 @@ impl Perform<GetPostsResponse> for Oper<GetPosts> {
let type_ = ListingType::from_str(&data.type_)?;
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> {
let data: &CreatePostLike = &self.data;
@ -364,13 +340,6 @@ impl Perform<PostResponse> for Oper<CreatePostLike> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> {
let data: &EditPost = &self.data;
@ -455,13 +425,6 @@ impl Perform<PostResponse> for Oper<EditPost> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PostResponse, Error> {
let data: &SavePost = &self.data;
@ -588,13 +552,6 @@ impl Perform<PostResponse> for Oper<SavePost> {
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()?;
if data.save {

115
server/src/api/site.rs
View file

@ -108,22 +108,16 @@ pub struct SaveSiteConfig {
auth: String,
}
impl Perform<ListCategoriesResponse> for Oper<ListCategories> {
impl Perform for Oper<ListCategories> {
type Response = ListCategoriesResponse;
fn perform(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<ListCategoriesResponse, Error> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetModlogResponse, Error> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<SiteResponse, Error> {
let data: &CreateSite = &self.data;
@ -246,13 +235,6 @@ impl Perform<SiteResponse> for Oper<CreateSite> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<SiteResponse, Error> {
let data: &EditSite = &self.data;
@ -307,13 +289,6 @@ impl Perform<SiteResponse> for Oper<EditSite> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteResponse, Error> {
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()?;
// TODO refactor this a little
@ -385,11 +354,7 @@ impl Perform<GetSiteResponse> for Oper<GetSite> {
admin: true,
show_nsfw: true,
};
let login_response = Oper::new(register).perform(
pool.clone(),
websocket_info.clone(),
rate_limit_info.clone(),
)?;
let login_response = Oper::new(register).perform(pool.clone(), websocket_info.clone())?;
info!("Admin {} created", setup.admin_username);
let create_site = CreateSite {
@ -400,7 +365,7 @@ impl Perform<GetSiteResponse> for Oper<GetSite> {
enable_nsfw: false,
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);
Some(SiteView::read(&conn)?)
} 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<SearchResponse, Error> {
let data: &Search = &self.data;
@ -467,13 +433,6 @@ impl Perform<SearchResponse> for Oper<Search> {
// 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()?;
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteResponse, Error> {
let data: &TransferSite = &self.data;
@ -585,13 +545,6 @@ impl Perform<GetSiteResponse> for Oper<TransferSite> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteConfigResponse, Error> {
let data: &GetSiteConfig = &self.data;
@ -662,13 +616,6 @@ impl Perform<GetSiteConfigResponse> for Oper<GetSiteConfig> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetSiteConfigResponse, Error> {
let data: &SaveSiteConfig = &self.data;
@ -701,13 +649,6 @@ impl Perform<GetSiteConfigResponse> for Oper<SaveSiteConfig> {
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()?;
// Only let admins read this

204
server/src/api/user.rs
View file

@ -199,22 +199,16 @@ pub struct UserJoinResponse {
pub user_id: i32,
}
impl Perform<LoginResponse> for Oper<Login> {
impl Perform for Oper<Login> {
type Response = LoginResponse;
fn perform(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> {
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()?;
// 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
Ok(LoginResponse {
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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> {
let data: &SaveUserSettings = &self.data;
@ -385,13 +367,6 @@ impl Perform<LoginResponse> for Oper<SaveUserSettings> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetUserDetailsResponse, Error> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<AddAdminResponse, Error> {
let data: &AddAdmin = &self.data;
@ -598,13 +568,6 @@ impl Perform<AddAdminResponse> for Oper<AddAdmin> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<BanUserResponse, Error> {
let data: &BanUser = &self.data;
@ -685,13 +649,6 @@ impl Perform<BanUserResponse> for Oper<BanUser> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetRepliesResponse, Error> {
let data: &GetReplies = &self.data;
@ -780,13 +738,6 @@ impl Perform<GetRepliesResponse> for Oper<GetReplies> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetUserMentionsResponse, Error> {
let data: &GetUserMentions = &self.data;
@ -818,13 +770,6 @@ impl Perform<GetUserMentionsResponse> for Oper<GetUserMentions> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<UserMentionResponse, Error> {
let data: &EditUserMention = &self.data;
@ -854,13 +800,6 @@ impl Perform<UserMentionResponse> for Oper<EditUserMention> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<GetRepliesResponse, Error> {
let data: &MarkAllAsRead = &self.data;
@ -901,13 +841,6 @@ impl Perform<GetRepliesResponse> for Oper<MarkAllAsRead> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> {
let data: &DeleteAccount = &self.data;
@ -999,13 +933,6 @@ impl Perform<LoginResponse> for Oper<DeleteAccount> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PasswordResetResponse, Error> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<LoginResponse, Error> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PrivateMessageResponse, Error> {
let data: &CreatePrivateMessage = &self.data;
@ -1180,13 +1096,6 @@ impl Perform<PrivateMessageResponse> for Oper<CreatePrivateMessage> {
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()?;
// 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PrivateMessageResponse, Error> {
let data: &EditPrivateMessage = &self.data;
@ -1265,13 +1175,6 @@ impl Perform<PrivateMessageResponse> for Oper<EditPrivateMessage> {
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 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(
&self,
pool: Pool<ConnectionManager<PgConnection>>,
_websocket_info: Option<WebsocketInfo>,
rate_limit_info: Option<RateLimitInfo>,
) -> Result<PrivateMessagesResponse, Error> {
let data: &GetPrivateMessages = &self.data;
@ -1334,13 +1238,6 @@ impl Perform<PrivateMessagesResponse> for Oper<GetPrivateMessages> {
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 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(
&self,
_pool: Pool<ConnectionManager<PgConnection>>,
websocket_info: Option<WebsocketInfo>,
_rate_limit_info: Option<RateLimitInfo>,
) -> Result<UserJoinResponse, Error> {
let data: &UserJoin = &self.data;

14
server/src/main.rs
View file

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

173
server/src/rate_limit/mod.rs
View file

@ -2,17 +2,180 @@ pub mod rate_limiter;
use super::{IPAddr, Settings};
use crate::api::APIError;
use crate::settings::RateLimitConfig;
use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform};
use failure::Error;
use futures::future::{ok, Ready};
use log::warn;
use rate_limiter::RateLimiter;
use rate_limiter::{RateLimitType, RateLimiter};
use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::Mutex;
use std::task::{Context, Poll};
use std::time::SystemTime;
use strum::IntoEnumIterator;
use tokio::sync::Mutex;
#[derive(Debug, Clone)]
pub struct RateLimitInfo {
pub rate_limiter: Arc<Mutex<RateLimiter>>,
pub ip: IPAddr,
pub struct RateLimit(pub Arc<Mutex<RateLimiter>>);
#[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) })
}
}

2
server/src/rate_limit/rate_limiter.rs
View file

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

185
server/src/routes/api.rs
View file

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

2
server/src/routes/mod.rs
View file

@ -1,6 +1,6 @@
use crate::api::{Oper, Perform};
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::{get_ip, markdown_to_html, version, Settings};
use actix::prelude::*;

7
server/src/routes/websocket.rs
View file

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

2
server/src/websocket/mod.rs
View file

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

569
server/src/websocket/server.rs
View file

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