use std::str::FromStr;
use diesel::PgConnection;
use failure::Error;
+use std::time::{SystemTime};
use {Crud, Joinable, Likeable, Followable, Bannable, Saveable, establish_connection, naive_now, naive_from_unix, SortType, SearchType, has_slurs, remove_slurs};
use actions::community::*;
use actions::moderator_views::*;
use actions::moderator::*;
+const RATE_LIMIT_MESSAGES: i32 = 30;
+const RATE_LIMIT_PER_SECOND: i32 = 60;
+const RATE_LIMIT_REGISTER_MESSAGES: i32 = 1;
+const RATE_LIMIT_REGISTER_PER_SECOND: i32 = 60;
+
#[derive(EnumString,ToString,Debug)]
pub enum UserOperation {
Login, Register, CreateCommunity, CreatePost, ListCommunities, ListCategories, GetPost, GetCommunity, CreateComment, EditComment, SaveComment, CreateCommentLike, GetPosts, CreatePostLike, EditPost, SavePost, EditCommunity, FollowCommunity, GetFollowedCommunities, GetUserDetails, GetReplies, GetModlog, BanFromCommunity, AddModToCommunity, CreateSite, EditSite, GetSite, AddAdmin, BanUser, Search, MarkAllAsRead
#[rtype(usize)]
pub struct Connect {
pub addr: Recipient<WSMessage>,
+ pub ip: String,
}
/// Session is disconnected
#[derive(Message)]
pub struct Disconnect {
pub id: usize,
+ pub ip: String,
}
/// Send message to specific room
auth: String
}
+#[derive(Debug)]
+pub struct RateLimitBucket {
+ last_checked: SystemTime,
+ allowance: f64
+}
+
+pub struct SessionInfo {
+ pub addr: Recipient<WSMessage>,
+ pub ip: String,
+}
+
/// `ChatServer` manages chat rooms and responsible for coordinating chat
/// session. implementation is super primitive
pub struct ChatServer {
- sessions: HashMap<usize, Recipient<WSMessage>>, // A map from generated random ID to session addr
+ sessions: HashMap<usize, SessionInfo>, // A map from generated random ID to session addr
+ rate_limits: HashMap<String, RateLimitBucket>,
rooms: HashMap<i32, HashSet<usize>>, // A map from room / post name to set of connectionIDs
rng: ThreadRng,
}
ChatServer {
sessions: HashMap::new(),
+ rate_limits: HashMap::new(),
rooms: rooms,
rng: rand::thread_rng(),
}
if let Some(sessions) = self.rooms.get(&room) {
for id in sessions {
if *id != skip_id {
- if let Some(addr) = self.sessions.get(id) {
- let _ = addr.do_send(WSMessage(message.to_owned()));
+ if let Some(info) = self.sessions.get(id) {
+ let _ = info.addr.do_send(WSMessage(message.to_owned()));
}
}
}
Ok(())
}
+
+ fn check_rate_limit_register(&mut self, addr: usize) -> Result<(), Error> {
+ self.check_rate_limit_full(addr, RATE_LIMIT_REGISTER_MESSAGES, RATE_LIMIT_REGISTER_PER_SECOND)
+ }
+
+ fn check_rate_limit(&mut self, addr: usize) -> Result<(), Error> {
+ self.check_rate_limit_full(addr, RATE_LIMIT_MESSAGES, RATE_LIMIT_PER_SECOND)
+ }
+
+ fn check_rate_limit_full(&mut self, addr: usize, rate: i32, per: i32) -> Result<(), Error> {
+ if let Some(info) = self.sessions.get(&addr) {
+ if let Some(rate_limit) = self.rate_limits.get_mut(&info.ip) {
+ if rate_limit.allowance == -2f64 {
+ rate_limit.allowance = rate as f64;
+ };
+
+ let current = SystemTime::now();
+ let time_passed = current.duration_since(rate_limit.last_checked)?.as_secs() as f64;
+ rate_limit.last_checked = current;
+ rate_limit.allowance += time_passed * (rate as f64 / per as f64);
+ if rate_limit.allowance > rate as f64 {
+ rate_limit.allowance = rate as f64;
+ }
+
+ if rate_limit.allowance < 1.0 {
+ println!("Rate limited IP: {}, time_passed: {}, allowance: {}", &info.ip, time_passed, rate_limit.allowance);
+ Err(ErrorMessage {
+ op: "Rate Limit".to_string(),
+ message: format!("Too many requests. {} per {} seconds", rate, per),
+ })?
+ } else {
+ rate_limit.allowance -= 1.0;
+ Ok(())
+ }
+ } else {
+ Ok(())
+ }
+ } else {
+ Ok(())
+ }
+ }
}
+
/// Make actor from `ChatServer`
impl Actor for ChatServer {
/// We are going to use simple Context, we just need ability to communicate
impl Handler<Connect> for ChatServer {
type Result = usize;
- fn handle(&mut self, msg: Connect, _: &mut Context<Self>) -> Self::Result {
+ fn handle(&mut self, msg: Connect, _ctx: &mut Context<Self>) -> Self::Result {
// notify all users in same room
// self.send_room_message(&"Main".to_owned(), "Someone joined", 0);
// register session with random id
let id = self.rng.gen::<usize>();
- self.sessions.insert(id, msg.addr);
+ println!("{} Joined", &msg.ip);
+
+ self.sessions.insert(id, SessionInfo {
+ addr: msg.addr,
+ ip: msg.ip.to_owned(),
+ });
+
+ if self.rate_limits.get(&msg.ip).is_none() {
+ self.rate_limits.insert(msg.ip, RateLimitBucket {
+ last_checked: SystemTime::now(),
+ allowance: -2f64,
+ });
+ }
+
+ // for (k,v) in &self.rate_limits {
+ // println!("{}: {:?}", k,v);
+ // }
// auto join session to Main room
// self.rooms.get_mut(&"Main".to_owned()).unwrap().insert(id);
}
}
+
/// Handler for Disconnect message.
impl Handler<Disconnect> for ChatServer {
type Result = ();
fn op_type(&self) -> UserOperation {
UserOperation::Register
}
- fn perform(&self, _chat: &mut ChatServer, _addr: usize) -> Result<String, Error> {
+ fn perform(&self, chat: &mut ChatServer, addr: usize) -> Result<String, Error> {
let conn = establish_connection();
+ chat.check_rate_limit_register(addr)?;
+
// Make sure passwords match
if &self.password != &self.password_verify {
return Err(self.error("Passwords do not match."))?
UserOperation::CreateCommunity
}
- fn perform(&self, _chat: &mut ChatServer, _addr: usize) -> Result<String, Error> {
+ fn perform(&self, chat: &mut ChatServer, addr: usize) -> Result<String, Error> {
let conn = establish_connection();
+ chat.check_rate_limit_register(addr)?;
+
let claims = match Claims::decode(&self.auth) {
Ok(claims) => claims.claims,
Err(_e) => {
UserOperation::CreatePost
}
- fn perform(&self, _chat: &mut ChatServer, _addr: usize) -> Result<String, Error> {
+ fn perform(&self, chat: &mut ChatServer, addr: usize) -> Result<String, Error> {
let conn = establish_connection();
+ chat.check_rate_limit_register(addr)?;
+
let claims = match Claims::decode(&self.auth) {
Ok(claims) => claims.claims,
Err(_e) => {
let conn = establish_connection();
+ chat.check_rate_limit(addr)?;
+
let claims = match Claims::decode(&self.auth) {
Ok(claims) => claims.claims,
Err(_e) => {
deleted: self.deleted.to_owned(),
read: self.read.to_owned(),
updated: if self.read.is_some() { orig_comment.updated } else {Some(naive_now())}
- };
+ };
let _updated_comment = match Comment::update(&conn, self.edit_id, &comment_form) {
Ok(comment) => comment,
let conn = establish_connection();
+ chat.check_rate_limit(addr)?;
+
let claims = match Claims::decode(&self.auth) {
Ok(claims) => claims.claims,
Err(_e) => {
UserOperation::CreatePostLike
}
- fn perform(&self, _chat: &mut ChatServer, _addr: usize) -> Result<String, Error> {
+ fn perform(&self, chat: &mut ChatServer, addr: usize) -> Result<String, Error> {
let conn = establish_connection();
+ chat.check_rate_limit(addr)?;
+
let claims = match Claims::decode(&self.auth) {
Ok(claims) => claims.claims,
Err(_e) => {
},
SearchType::Comments => {
comments = CommentView::list(&conn,
- &sort,
+ &sort,
None,
None,
Some(self.q.to_owned()),
self.page,
self.limit)?;
comments = CommentView::list(&conn,
- &sort,
+ &sort,
None,
None,
Some(self.q.to_owned()),
projects / lemmy.git / commitdiff