diff --git a/hc/api/models.py b/hc/api/models.py index 8bdd91ad..4561ce60 100644 --- a/hc/api/models.py +++ b/hc/api/models.py @@ -902,6 +902,14 @@ class TokenBucket(models.Model): # 6 messages for a single chat per minute: return TokenBucket.authorize(value, 6, 60) + @staticmethod + def authorize_signal(phone): + salted_encoded = (phone + settings.SECRET_KEY).encode() + value = "signal-%s" % hashlib.sha1(salted_encoded).hexdigest() + + # 6 messages for a single recipient per minute: + return TokenBucket.authorize(value, 6, 60) + @staticmethod def authorize_sudo_code(user): value = "sudo-%d" % user.id diff --git a/hc/api/tests/test_notify_signal.py b/hc/api/tests/test_notify_signal.py index 46f25f58..6b2d5eef 100644 --- a/hc/api/tests/test_notify_signal.py +++ b/hc/api/tests/test_notify_signal.py @@ -6,7 +6,7 @@ from unittest.mock import patch from django.utils.timezone import now from django.test.utils import override_settings -from hc.api.models import Channel, Check, Notification +from hc.api.models import Channel, Check, Notification, TokenBucket from hc.test import BaseTestCase @@ -80,3 +80,14 @@ class NotifySignalTestCase(BaseTestCase): cmd = " ".join(args[0]) self.assertIn("Foo & Bar", cmd) + + @override_settings(SECRET_KEY="test-secret") + def test_it_obeys_rate_limit(self): + # "2862..." is sha1("+123456789test-secret") + obj = TokenBucket(value="signal-2862991ccaa15c8856e7ee0abaf3448fb3c292e0") + obj.tokens = 0 + obj.save() + + self.channel.notify(self.check) + n = Notification.objects.first() + self.assertEqual(n.error, "Rate limit exceeded") diff --git a/hc/api/transports.py b/hc/api/transports.py index 5349d06a..c5786df0 100644 --- a/hc/api/transports.py +++ b/hc/api/transports.py @@ -673,6 +673,11 @@ class Signal(Transport): if not settings.SIGNAL_CLI_USERNAME: return "Signal notifications are not enabled" + from hc.api.models import TokenBucket + + if not TokenBucket.authorize_signal(self.channel.phone_number): + return "Rate limit exceeded" + text = tmpl("signal_message.html", check=check, site_name=settings.SITE_NAME) args = settings.SIGNAL_CLI_CMD.split()