diff --git a/synapse/api/errors.py b/synapse/api/errors.py
index 21ededc5ae..23ce0af277 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):
@@ -39,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:
@@ -53,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."""
@@ -91,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/api/ratelimiting.py b/synapse/api/ratelimiting.py
new file mode 100644
index 0000000000..ab26c2376a
--- /dev/null
+++ b/synapse/api/ratelimiting.py
@@ -0,0 +1,65 @@
+import collections
+
+
+class Ratelimiter(object):
+ """
+ Ratelimit message sending by user.
+ """
+
+ def __init__(self):
+ self.message_counts = collections.OrderedDict()
+
+ 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_s, None),
+ )
+ 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
+ )
+
+ 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/synapse/app/homeserver.py b/synapse/app/homeserver.py
index 606c9c650d..8a7cd07fec 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 18072e3196..a9aa4c735c 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/config/ratelimiting.py b/synapse/config/ratelimiting.py
new file mode 100644
index 0000000000..a64aeeb6b2
--- /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-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 b37c8be964..935adea1ac 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 LimitExceededError
class BaseHandler(object):
@@ -25,8 +26,22 @@ 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_message_burst_count,
+ )
+ if not allowed:
+ raise LimitExceededError(
+ retry_after_ms=1000*(time_allowed - time_now),
+ )
+
class BaseRoomHandler(BaseHandler):
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 4aeb2089f5..c9e3c4e451 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 53aa77405c..34a19bc64e 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(
diff --git a/synapse/server.py b/synapse/server.py
index 3e72b2bcd5..35e311a47d 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.
"""
diff --git a/tests/api/__init__.py b/tests/api/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/tests/api/__init__.py
diff --git a/tests/api/test_ratelimiting.py b/tests/api/test_ratelimiting.py
new file mode 100644
index 0000000000..dc2f83c7eb
--- /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)
diff --git a/tests/handlers/test_room.py b/tests/handlers/test_room.py
index 219a53c426..4591a5ea58 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 1d1336d12d..c8527f3517 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 cdaf948a3b..23c50824c7 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):
|
