Tighten Signal number verification rate limiting

hc/api/transports.py

@@ -884,7 +884,7 @@ class Signal(Transport):
         else:
             return not self.channel.signal_notify_up
 
-    def send(self, recipient, message):
+    def send(self, recipient: str, message: str) -> None:
         payload = {
             "jsonrpc": "2.0",
             "method": "send",
@@ -920,7 +920,7 @@ class Signal(Transport):
                 code = reply["error"].get("code")
                 raise TransportError("signal-cli call failed (%s)" % code)
 
-    def _read_replies(self, payload_bytes):
+    def _read_replies(self, payload_bytes: bytes):
         """Send a request to signal-cli over UNIX socket. Read and yield replies.
 
         This method:

hc/front/tests/test_verify_signal_number.py

@@ -69,9 +69,20 @@ class VerifySignalNumberTestCase(BaseTestCase):
         r = self.client.post(self.url, {"phone": "+1234567890"})
         self.assertRedirects(r, "/accounts/login/?next=" + self.url)
 
-    def test_it_obeys_verification_rate_limit(self):
+    def test_it_obeys_per_account_rate_limit(self):
         TokenBucket.objects.create(value=f"signal-verify-{self.alice.id}", tokens=0)
 
         self.client.login(username="alice@example.org", password="password")
         r = self.client.post(self.url, {"phone": "+1234567890"})
         self.assertContains(r, "Verification rate limit exceeded")
+
+    @override_settings(SECRET_KEY="test-secret")
+    def test_it_obeys_per_recipient_rate_limit(self):
+        # "2862..." is sha1("+123456789test-secret")
+        obj = TokenBucket(value="signal-2862991ccaa15c8856e7ee0abaf3448fb3c292e0")
+        obj.tokens = 0
+        obj.save()
+
+        self.client.login(username="alice@example.org", password="password")
+        r = self.client.post(self.url, {"phone": "+123456789"})
+        self.assertContains(r, "Verification rate limit exceeded")

hc/front/views.py

@@ -2410,10 +2410,11 @@ def signal_captcha(request: HttpRequest) -> HttpResponse:
 @require_setting("SIGNAL_CLI_SOCKET")
 @login_required
 @require_POST
-def verify_signal_number(request: HttpRequest):
-    def render_result(result):
+def verify_signal_number(request: HttpRequest) -> HttpResponse:
+    def render_result(result: str | None) -> HttpResponse:
         return render(request, "integrations/signal_result.html", {"result": result})
 
+    # Enforce per-account rate limit (50 verifications per day)
     if not TokenBucket.authorize_signal_verification(request.user):
         return render_result("Verification rate limit exceeded")
 
@@ -2421,8 +2422,12 @@ def verify_signal_number(request: HttpRequest):
     if not form.is_valid():
         return render_result("Invalid phone number")
 
+    phone = form.cleaned_data["phone"]
+    # Enforce per-recipient rate limit (6 messages per minute)
+    if not TokenBucket.authorize_signal(phone):
+        return render_result("Verification rate limit exceeded")
+
     try:
-        phone = form.cleaned_data["phone"]
         Signal(None).send(phone, f"Test message from {settings.SITE_NAME}")
     except TransportError as e:
         return render_result(e.message)