From 436b3c7d0c93bd6019a576425506c4c8f2963632 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Mon, 1 Sep 2014 17:54:54 +0100 Subject: [PATCH 1/7] Ratelimiter object --- synapse/api/ratelimiting.py | 36 ++++++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) create mode 100644 synapse/api/ratelimiting.py diff --git a/synapse/api/ratelimiting.py b/synapse/api/ratelimiting.py new file mode 100644 index 000000000..a730abd59 --- /dev/null +++ b/synapse/api/ratelimiting.py @@ -0,0 +1,36 @@ +import collections + + +class Ratelimiter(object): + + def __init__(self): + self.message_counts = collections.OrderedDict() + + def prune_message_counts(self, time_now): + for user_id in self.message_counts.keys(): + message_count, time_start, msg_rate_hz = ( + self.message_counts[user_id] + ) + time_delta = time_now - time_start + if message_count - time_delta * msg_rate_hz > 0: + break + else: + del self.message_counts[user_id] + + def send_message(self, user_id, time_now, msg_rate_hz, burst_count): + self.prune_message_counts(time_now) + message_count, time_start, _ignored = self.message_counts.pop( + user_id, (0., time_now, None), + ) + time_delta = time_now - time_start + if message_count - time_delta * msg_rate_hz < 0: + a + if message_count - (time_now - time_start) * msg_rate_hz > burst_count: + allowed = False + else: + allowed = True + message_count += 1 + self.message_counts[user_id] = ( + message_count, time_start, msg_rate_hz + ) + return allowed From dd2cd9312a9f9f8f782da9ed8f99544176ea9fc1 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 2 Sep 2014 15:06:20 +0100 Subject: [PATCH 2/7] Test ratelimiter --- synapse/api/ratelimiting.py | 67 ++++++++++++++++++++++++---------- tests/api/__init__.py | 0 tests/api/test_ratelimiting.py | 39 ++++++++++++++++++++ 3 files changed, 87 insertions(+), 19 deletions(-) create mode 100644 tests/api/__init__.py create mode 100644 tests/api/test_ratelimiting.py diff --git a/synapse/api/ratelimiting.py b/synapse/api/ratelimiting.py index a730abd59..ab26c2376 100644 --- a/synapse/api/ratelimiting.py +++ b/synapse/api/ratelimiting.py @@ -2,35 +2,64 @@ import collections class Ratelimiter(object): + """ + Ratelimit message sending by user. + """ def __init__(self): self.message_counts = collections.OrderedDict() - def prune_message_counts(self, time_now): - for user_id in self.message_counts.keys(): - message_count, time_start, msg_rate_hz = ( - self.message_counts[user_id] - ) - time_delta = time_now - time_start - if message_count - time_delta * msg_rate_hz > 0: - break - else: - del self.message_counts[user_id] - - def send_message(self, user_id, time_now, msg_rate_hz, burst_count): - self.prune_message_counts(time_now) + def send_message(self, user_id, time_now_s, msg_rate_hz, burst_count): + """Can the user send a message? + Args: + user_id: The user sending a message. + time_now_s: The time now. + msg_rate_hz: The long term number of messages a user can send in a + second. + burst_count: How many messages the user can send before being + limited. + Returns: + A pair of a bool indicating if they can send a message now and a + time in seconds of when they can next send a message. + """ + self.prune_message_counts(time_now_s) message_count, time_start, _ignored = self.message_counts.pop( - user_id, (0., time_now, None), + user_id, (0., time_now_s, None), ) - time_delta = time_now - time_start - if message_count - time_delta * msg_rate_hz < 0: - a - if message_count - (time_now - time_start) * msg_rate_hz > burst_count: + time_delta = time_now_s - time_start + sent_count = message_count - time_delta * msg_rate_hz + if sent_count < 0: + allowed = True + time_start = time_now_s + messagecount = 1. + elif sent_count > burst_count - 1.: allowed = False else: allowed = True message_count += 1 + self.message_counts[user_id] = ( message_count, time_start, msg_rate_hz ) - return allowed + + if msg_rate_hz > 0: + time_allowed = ( + time_start + (message_count - burst_count + 1) / msg_rate_hz + ) + if time_allowed < time_now_s: + time_allowed = time_now_s + else: + time_allowed = -1 + + return allowed, time_allowed + + def prune_message_counts(self, time_now_s): + for user_id in self.message_counts.keys(): + message_count, time_start, msg_rate_hz = ( + self.message_counts[user_id] + ) + time_delta = time_now_s - time_start + if message_count - time_delta * msg_rate_hz > 0: + break + else: + del self.message_counts[user_id] diff --git a/tests/api/__init__.py b/tests/api/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/api/test_ratelimiting.py b/tests/api/test_ratelimiting.py new file mode 100644 index 000000000..dc2f83c7e --- /dev/null +++ b/tests/api/test_ratelimiting.py @@ -0,0 +1,39 @@ +from synapse.api.ratelimiting import Ratelimiter + +import unittest + +class TestRatelimiter(unittest.TestCase): + + def test_allowed(self): + limiter = Ratelimiter() + allowed, time_allowed = limiter.send_message( + user_id="test_id", time_now_s=0, msg_rate_hz=0.1, burst_count=1, + ) + self.assertTrue(allowed) + self.assertEquals(10., time_allowed) + + allowed, time_allowed = limiter.send_message( + user_id="test_id", time_now_s=5, msg_rate_hz=0.1, burst_count=1, + ) + self.assertFalse(allowed) + self.assertEquals(10., time_allowed) + + allowed, time_allowed = limiter.send_message( + user_id="test_id", time_now_s=10, msg_rate_hz=0.1, burst_count=1 + ) + self.assertTrue(allowed) + self.assertEquals(20., time_allowed) + + def test_pruning(self): + limiter = Ratelimiter() + allowed, time_allowed = limiter.send_message( + user_id="test_id_1", time_now_s=0, msg_rate_hz=0.1, burst_count=1, + ) + + self.assertIn("test_id_1", limiter.message_counts) + + allowed, time_allowed = limiter.send_message( + user_id="test_id_2", time_now_s=10, msg_rate_hz=0.1, burst_count=1 + ) + + self.assertNotIn("test_id_1", limiter.message_counts) From c7a7cdf7346c9268b1d4f483b31e1fdc39b6d7e0 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 2 Sep 2014 17:57:04 +0100 Subject: [PATCH 3/7] Add ratelimiting function to basehandler --- synapse/api/errors.py | 1 + synapse/app/homeserver.py | 1 + synapse/config/homeserver.py | 4 +++- synapse/handlers/_base.py | 17 +++++++++++++++++ synapse/server.py | 5 +++++ 5 files changed, 27 insertions(+), 1 deletion(-) diff --git a/synapse/api/errors.py b/synapse/api/errors.py index 21ededc5a..3f33ca5b9 100644 --- a/synapse/api/errors.py +++ b/synapse/api/errors.py @@ -28,6 +28,7 @@ class Codes(object): UNKNOWN = "M_UNKNOWN" NOT_FOUND = "M_NOT_FOUND" UNKNOWN_TOKEN = "M_UNKNOWN_TOKEN" + LIMIT_EXCEEDED = "M_LIMIT_EXCEEDED" class CodeMessageException(Exception): diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index 606c9c650..8a7cd07fe 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -247,6 +247,7 @@ def setup(): upload_dir=os.path.abspath("uploads"), db_name=config.database_path, tls_context_factory=tls_context_factory, + config=config, ) hs.register_servlets() diff --git a/synapse/config/homeserver.py b/synapse/config/homeserver.py index 18072e319..a9aa4c735 100644 --- a/synapse/config/homeserver.py +++ b/synapse/config/homeserver.py @@ -17,8 +17,10 @@ from .tls import TlsConfig from .server import ServerConfig from .logger import LoggingConfig from .database import DatabaseConfig +from .ratelimiting import RatelimitConfig -class HomeServerConfig(TlsConfig, ServerConfig, DatabaseConfig, LoggingConfig): +class HomeServerConfig(TlsConfig, ServerConfig, DatabaseConfig, LoggingConfig, + RatelimitConfig): pass if __name__=='__main__': diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index b37c8be96..dc1298366 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -14,6 +14,7 @@ # limitations under the License. from twisted.internet import defer +from synapse.api.errors import cs_error, Codes class BaseHandler(object): @@ -25,8 +26,24 @@ class BaseHandler(object): self.room_lock = hs.get_room_lock_manager() self.state_handler = hs.get_state_handler() self.distributor = hs.get_distributor() + self.ratelimiter = hs.get_ratelimiter() + self.clock = hs.get_clock() self.hs = hs + def ratelimit(self, user_id): + time_now = self.clock.time() + allowed, time_allowed = self.ratelimiter.send_message( + user_id, time_now, + msg_rate_hz=self.hs.config.rc_messages_per_second, + burst_count=self.hs.config.rc_messsage_burst_count, + ) + if not allowed: + raise cs_error( + "Limit exceeded", + Codes.M_LIMIT_EXCEEDED, + retry_after_ms=1000*(time_allowed - time_now), + ) + class BaseRoomHandler(BaseHandler): diff --git a/synapse/server.py b/synapse/server.py index 3e72b2bcd..35e311a47 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -32,6 +32,7 @@ from synapse.util import Clock from synapse.util.distributor import Distributor from synapse.util.lockutils import LockManager from synapse.streams.events import EventSources +from synapse.api.ratelimiting import Ratelimiter class BaseHomeServer(object): @@ -73,6 +74,7 @@ class BaseHomeServer(object): 'resource_for_web_client', 'resource_for_content_repo', 'event_sources', + 'ratelimiter', ] def __init__(self, hostname, **kwargs): @@ -190,6 +192,9 @@ class HomeServer(BaseHomeServer): def build_event_sources(self): return EventSources(self) + def build_ratelimiter(self): + return Ratelimiter() + def register_servlets(self): """ Register all servlets associated with this HomeServer. """ From 0a1260b03a4522a41105478c9b123eed36f40a25 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 2 Sep 2014 18:00:15 +0100 Subject: [PATCH 4/7] Add ratelimiting config --- synapse/config/ratelimiting.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) create mode 100644 synapse/config/ratelimiting.py diff --git a/synapse/config/ratelimiting.py b/synapse/config/ratelimiting.py new file mode 100644 index 000000000..6c41d55d7 --- /dev/null +++ b/synapse/config/ratelimiting.py @@ -0,0 +1,21 @@ +from ._base import Config + +class RatelimitConfig(Config): + + def __init__(self, args): + super(RatelimitConfig, self).__init__(args) + self.rc_messages_per_second = args.rc_messages_per_second + self.rc_message_burst_count = args.rc_message_burst_count + + @classmethod + def add_arguments(cls, parser): + super(RatelimitConfig, cls).add_arguments(parser) + rc_group = parser.add_argument_group("ratelimiting") + rc_group.add_argument( + "--rc-messages-per-second", type=float, default=0.2 + help="number of messages a client can send per second" + ) + rc_group.add_argument( + "--rc_messsage_burst_count", type=float, default=10 + help="number of message a client can send before being throttled" + ) From 780548b577f5e59cd4f3aa03cf20dbf66a790a39 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Tue, 2 Sep 2014 18:22:15 +0100 Subject: [PATCH 5/7] rate limiting for message sending --- synapse/config/ratelimiting.py | 4 ++-- synapse/handlers/_base.py | 4 ++-- synapse/handlers/message.py | 2 ++ synapse/handlers/room.py | 1 + 4 files changed, 7 insertions(+), 4 deletions(-) diff --git a/synapse/config/ratelimiting.py b/synapse/config/ratelimiting.py index 6c41d55d7..a64aeeb6b 100644 --- a/synapse/config/ratelimiting.py +++ b/synapse/config/ratelimiting.py @@ -12,10 +12,10 @@ class RatelimitConfig(Config): super(RatelimitConfig, cls).add_arguments(parser) rc_group = parser.add_argument_group("ratelimiting") rc_group.add_argument( - "--rc-messages-per-second", type=float, default=0.2 + "--rc-messages-per-second", type=float, default=0.2, help="number of messages a client can send per second" ) rc_group.add_argument( - "--rc_messsage_burst_count", type=float, default=10 + "--rc-message-burst-count", type=float, default=10, help="number of message a client can send before being throttled" ) diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index dc1298366..c150b60e0 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -35,12 +35,12 @@ class BaseHandler(object): allowed, time_allowed = self.ratelimiter.send_message( user_id, time_now, msg_rate_hz=self.hs.config.rc_messages_per_second, - burst_count=self.hs.config.rc_messsage_burst_count, + burst_count=self.hs.config.rc_message_burst_count, ) if not allowed: raise cs_error( "Limit exceeded", - Codes.M_LIMIT_EXCEEDED, + Codes.LIMIT_EXCEEDED, retry_after_ms=1000*(time_allowed - time_now), ) diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 4aeb2089f..c9e3c4e45 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -76,6 +76,8 @@ class MessageHandler(BaseRoomHandler): Raises: SynapseError if something went wrong. """ + + self.ratelimit(event.user_id) # TODO(paul): Why does 'event' not have a 'user' object? user = self.hs.parse_userid(event.user_id) assert user.is_mine, "User must be our own: %s" % (user,) diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index 048b71930..b0c94d35a 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -49,6 +49,7 @@ class RoomCreationHandler(BaseRoomHandler): SynapseError if the room ID was taken, couldn't be stored, or something went horribly wrong. """ + self.ratelimit(user_id) if "room_alias_name" in config: room_alias = RoomAlias.create_local( From 683596f91e3b074601d607b9e5891ab85b5629ca Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 3 Sep 2014 08:58:48 +0100 Subject: [PATCH 6/7] Raise LimitExceedError when the ratelimiting is throttling requests --- synapse/api/errors.py | 34 +++++++++++++++++++++++++++------- synapse/handlers/_base.py | 6 ++---- 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/synapse/api/errors.py b/synapse/api/errors.py index 3f33ca5b9..23ce0af27 100644 --- a/synapse/api/errors.py +++ b/synapse/api/errors.py @@ -40,10 +40,13 @@ class CodeMessageException(Exception): self.code = code self.msg = msg + def error_dict(self): + return cs_error(self.msg) + class SynapseError(CodeMessageException): """A base error which can be caught for all synapse events.""" - def __init__(self, code, msg, errcode=""): + def __init__(self, code, msg, errcode=Codes.UNKNOWN): """Constructs a synapse error. Args: @@ -54,6 +57,11 @@ class SynapseError(CodeMessageException): super(SynapseError, self).__init__(code, msg) self.errcode = errcode + def error_dict(self): + return cs_error( + self.msg, + self.errcode, + ) class RoomError(SynapseError): """An error raised when a room event fails.""" @@ -92,13 +100,25 @@ class StoreError(SynapseError): pass -def cs_exception(exception): - if isinstance(exception, SynapseError): +class LimitExceededError(SynapseError): + """A client has sent too many requests and is being throttled. + """ + def __init__(self, code=429, msg="Too Many Requests", retry_after_ms=None, + errcode=Codes.LIMIT_EXCEEDED): + super(LimitExceededError, self).__init__(code, msg, errcode) + self.retry_after_ms = retry_after_ms + + def error_dict(self): return cs_error( - exception.msg, - Codes.UNKNOWN if not exception.errcode else exception.errcode) - elif isinstance(exception, CodeMessageException): - return cs_error(exception.msg) + self.msg, + self.errcode, + retry_after_ms=self.retry_after_ms, + ) + + +def cs_exception(exception): + if isinstance(exception, CodeMessageException): + return exception.error_dict() else: logging.error("Unknown exception type: %s", type(exception)) diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py index c150b60e0..935adea1a 100644 --- a/synapse/handlers/_base.py +++ b/synapse/handlers/_base.py @@ -14,7 +14,7 @@ # limitations under the License. from twisted.internet import defer -from synapse.api.errors import cs_error, Codes +from synapse.api.errors import LimitExceededError class BaseHandler(object): @@ -38,9 +38,7 @@ class BaseHandler(object): burst_count=self.hs.config.rc_message_burst_count, ) if not allowed: - raise cs_error( - "Limit exceeded", - Codes.LIMIT_EXCEEDED, + raise LimitExceededError( retry_after_ms=1000*(time_allowed - time_now), ) From cdd8602e746d37803ae734f999478fc84a19f43c Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 3 Sep 2014 09:15:22 +0100 Subject: [PATCH 7/7] Fix tests to support ratelimiting --- tests/handlers/test_room.py | 13 +++++++++++ tests/rest/test_events.py | 9 +++++++- tests/rest/test_rooms.py | 44 ++++++++++++++++++++++++++++++++++++- 3 files changed, 64 insertions(+), 2 deletions(-) diff --git a/tests/handlers/test_room.py b/tests/handlers/test_room.py index 219a53c42..4591a5ea5 100644 --- a/tests/handlers/test_room.py +++ b/tests/handlers/test_room.py @@ -39,6 +39,10 @@ class RoomMemberHandlerTestCase(unittest.TestCase): hs = HomeServer( self.hostname, db_pool=None, + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), datastore=NonCallableMock(spec_set=[ "persist_event", "get_joined_hosts_for_room", @@ -82,6 +86,8 @@ class RoomMemberHandlerTestCase(unittest.TestCase): self.snapshot = Mock() self.datastore.snapshot_room.return_value = self.snapshot + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) @defer.inlineCallbacks def test_invite(self): @@ -342,6 +348,10 @@ class RoomCreationTest(unittest.TestCase): ]), auth=NonCallableMock(spec_set=["check"]), state_handler=NonCallableMock(spec_set=["handle_new_event"]), + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) self.federation = NonCallableMock(spec_set=[ @@ -368,6 +378,9 @@ class RoomCreationTest(unittest.TestCase): return defer.succeed([]) self.datastore.get_joined_hosts_for_room.side_effect = hosts + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) + @defer.inlineCallbacks def test_room_creation(self): user_id = "@foo:red" diff --git a/tests/rest/test_events.py b/tests/rest/test_events.py index 1d1336d12..c8527f351 100644 --- a/tests/rest/test_events.py +++ b/tests/rest/test_events.py @@ -32,7 +32,7 @@ import logging from ..utils import MockHttpResource, MemoryDataStore from .utils import RestTestCase -from mock import Mock +from mock import Mock, NonCallableMock logging.getLogger().addHandler(logging.NullHandler()) @@ -136,8 +136,15 @@ class EventStreamPermissionsTestCase(RestTestCase): "call_later", "cancel_call_later", "time_msec", + "time" ]), + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) hs.get_handlers().federation_handler = Mock() diff --git a/tests/rest/test_rooms.py b/tests/rest/test_rooms.py index cdaf948a3..23c50824c 100644 --- a/tests/rest/test_rooms.py +++ b/tests/rest/test_rooms.py @@ -30,7 +30,7 @@ import urllib from ..utils import MockHttpResource, MemoryDataStore from .utils import RestTestCase -from mock import Mock +from mock import Mock, NonCallableMock PATH_PREFIX = "/_matrix/client/api/v1" @@ -58,7 +58,14 @@ class RoomPermissionsTestCase(RestTestCase): replication_layer=Mock(), state_handler=state_handler, persistence_service=persistence_service, + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) + hs.get_handlers().federation_handler = Mock() def _get_user_by_token(token=None): @@ -405,7 +412,14 @@ class RoomsMemberListTestCase(RestTestCase): replication_layer=Mock(), state_handler=state_handler, persistence_service=persistence_service, + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) + hs.get_handlers().federation_handler = Mock() self.auth_user_id = self.user_id @@ -483,7 +497,14 @@ class RoomsCreateTestCase(RestTestCase): replication_layer=Mock(), state_handler=state_handler, persistence_service=persistence_service, + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) + hs.get_handlers().federation_handler = Mock() def _get_user_by_token(token=None): @@ -573,7 +594,14 @@ class RoomTopicTestCase(RestTestCase): replication_layer=Mock(), state_handler=state_handler, persistence_service=persistence_service, + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) + hs.get_handlers().federation_handler = Mock() def _get_user_by_token(token=None): @@ -676,7 +704,14 @@ class RoomMemberStateTestCase(RestTestCase): replication_layer=Mock(), state_handler=state_handler, persistence_service=persistence_service, + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) + hs.get_handlers().federation_handler = Mock() def _get_user_by_token(token=None): @@ -801,7 +836,14 @@ class RoomMessagesTestCase(RestTestCase): replication_layer=Mock(), state_handler=state_handler, persistence_service=persistence_service, + ratelimiter=NonCallableMock(spec_set=[ + "send_message", + ]), + config=NonCallableMock(), ) + self.ratelimiter = hs.get_ratelimiter() + self.ratelimiter.send_message.return_value = (True, 0) + hs.get_handlers().federation_handler = Mock() def _get_user_by_token(token=None):