Rate limiting for Telegram notifications (10 notifications per chat per minute)

CHANGELOG.md

@@ -3,6 +3,9 @@ All notable changes to this project will be documented in this file.
 
 ## v1.15.0-dev - Unreleased
 
+### Improvements
+- Rate limiting for Telegram notifications (10 notifications per chat per minute)
+
 ### Bug Fixes
 - "Get a single check" API call now supports read-only API keys (#346)
 

hc/api/models.py

@@ -813,3 +813,10 @@ class TokenBucket(models.Model):
 
         # 20 password attempts per day
         return TokenBucket.authorize(value, 20, 3600 * 24)
+
+    @staticmethod
+    def authorize_telegram(telegram_id):
+        value = "tg-%s" % telegram_id
+
+        # 10 messages for a single chat per minute:
+        return TokenBucket.authorize(value, 10, 60)

hc/api/tests/test_notify.py

@@ -6,7 +6,7 @@ from unittest.mock import patch, Mock
 
 from django.core import mail
 from django.utils.timezone import now
-from hc.api.models import Channel, Check, Notification
+from hc.api.models import Channel, Check, Notification, TokenBucket
 from hc.test import BaseTestCase
 from requests.exceptions import ConnectionError, Timeout
 from django.test.utils import override_settings
@@ -602,6 +602,15 @@ class NotifyTestCase(BaseTestCase):
         n = Notification.objects.first()
         self.assertEqual(n.error, 'Received status code 400 with a message: "Hi"')
 
+    def test_telegram_obeys_rate_limit(self):
+        self._setup_data("telegram", json.dumps({"id": 123}))
+
+        TokenBucket.objects.create(value="tg-123", tokens=0)
+
+        self.channel.notify(self.check)
+        n = Notification.objects.first()
+        self.assertEqual(n.error, "Rate limit exceeded")
+
     @patch("hc.api.transports.requests.request")
     def test_sms(self, mock_post):
         self._setup_data("sms", "+1234567890")

hc/api/transports.py

@@ -452,11 +452,18 @@ class Telegram(HttpTransport):
 
     @classmethod
     def send(cls, chat_id, text):
+        # Telegram.send is a separate method because it is also used in
+        # hc.front.views.telegram_bot to send invite links.
         return cls.post(
             cls.SM, json={"chat_id": chat_id, "text": text, "parse_mode": "html"}
         )
 
     def notify(self, check):
+        from hc.api.models import TokenBucket
+
+        if not TokenBucket.authorize_telegram(self.channel.telegram_id):
+            return "Rate limit exceeded"
+
         text = tmpl("telegram_message.html", check=check)
         return self.send(self.channel.telegram_id, text)