Implement the defense system against spammers in suggestions

Introduce a simple system, based on timeouts, to restrict spammers and
flooders sending hundreds of messages within a short period of time.

Author: kozalosev

app/bot.py

@@ -2,9 +2,11 @@
 
 import os
 import asyncio
+import logging
 from aiohttp import web
 from aiotg import Bot, Chat, InlineQuery, CallbackQuery
 from klocmod import LocalizationsContainer
+from typing import *
 
 import msgdb
 import strconv
@@ -13,14 +15,21 @@
 from data.config import *
 from queryutil import *
 from userutil import *
+from timeoututil import *
 
 
 ISSUES_LINK = "https://{}/issues/".format(REPO_URL)
 DECRYPT_BUTTON_CACHE_TIME = 3600    # in seconds
+MAX_SUGGESTIONS = 5
+MAX_SUGGESTIONS_TIMEOUT = 720       # in minutes
 
 bot = Bot(api_token=TOKEN, default_in_groups=True)
 localizations = LocalizationsContainer.from_file("app/localizations.ini")
 text_processors = TextProcessorsLoader(strconv)
+logger = logging.getLogger(__name__)
+
+UserId = str
+suggestion_counters = {}    # type: Dict[UserId, SuggestionsCounter]
 
 
 @bot.command("/start")
@@ -37,6 +46,20 @@ async def suggest(chat: Chat, match) -> None:
     username = escape_html(get_username_or_fullname(user))
     lang = localizations.get_lang(user['language_code'])
 
+    if user['id'] not in suggestion_counters:
+        suggestion_counters[user['id']] = SuggestionsCounter()
+    counter = suggestion_counters[user['id']]
+    counter.increment()
+    timeout_in_minutes = compute_timeout_in_minutes(counter, MAX_SUGGESTIONS, MAX_SUGGESTIONS_TIMEOUT)
+    logger.debug(counter)
+    logger.debug("User ID: %d, username: %s, timeout: %d minutes", user['id'], username, timeout_in_minutes)
+    if counter.minutes_elapsed >= timeout_in_minutes:
+        counter.reset()
+    counter.update_timestamp()
+    if counter.value > MAX_SUGGESTIONS:
+        chat.send_text(lang['suggestion_timeout'].format(timeout_in_minutes))
+        return
+
     first_line = match.group(1)
     rest_lines = chat.message['text'].split('\n')[1:]
     suggestion = escape_html(first_line + '\n' + '\n'.join(rest_lines))

app/localizations.ini

@@ -10,6 +10,7 @@ suggestion_report_template = User {} <b>suggests</b>:
 
                              {}
 suggestion_sent = You suggestion was sent to the developer!
+suggestion_timeout = Oops! The defense system against spammers and flooders has been activated! You will be able to send a new suggestion in {:d} minutes.
 suggest_create_issue = To keep an eye on your suggestion, you can additionally [create an issue on GitHub]({}).
 decrypt = Decrypt
 missing_original_text = Oops! I lost the original text :(
@@ -35,6 +36,7 @@ suggestion_report_template = Пользователь {} <b>предлагает
 
                              {}
 suggestion_sent = Предложение отправлено разработчику!
+suggestion_timeout = Упс! Сработала защита от спамеров и флудеров! Новое предложение можно будет отправить через {:d} минут.
 suggest_create_issue = Чтобы иметь возможность следить за судьбой предложения, можно дополнительно [создать тикет на GitHub]({}).
 decrypt = Расшифровать
 missing_original_text = Расшифровка потерялась :(

app/timeoututil.py

@@ -0,0 +1,79 @@
+"""Utility functions and classes for implementations of different timeouts."""
+
+import math
+import time
+
+
+__all__ = ['SuggestionsCounter', 'compute_timeout_in_minutes']
+
+
+class SuggestionsCounter:
+    """A simple counter that remembers the last time it was adjusted."""
+
+    _value = 0                      # type: int
+    _last_change_timestamp = 0.0    # type: float
+
+    def __init__(self):
+        self.update_timestamp()
+
+    def increment(self) -> int:
+        self._value += 1
+        return self._value
+
+    def reset(self) -> None:
+        self._value = 0
+
+    @property
+    def value(self) -> int:
+        return self._value
+
+    @property
+    def last_change_timestamp(self) -> float:
+        return self._last_change_timestamp
+
+    @property
+    def seconds_elapsed(self) -> int:
+        """:return: the number of seconds elapsed since the last change."""
+        return math.floor(time.time() - self._last_change_timestamp)
+
+    @property
+    def minutes_elapsed(self) -> int:
+        """:return: the number of minutes elapsed since the last change."""
+        return math.floor(self.seconds_elapsed / 60)
+
+    def update_timestamp(self) -> None:
+        self._last_change_timestamp = time.time()
+
+    def __str__(self) -> str:
+        return "{class_name}(value={value}, last_change_timestamp={last_change_timestamp}, " \
+               "seconds_elapsed={seconds_elapsed}, minutes_elapsed={minutes_elapsed})".format(
+                    class_name=self.__class__.__name__,
+                    value=self.value,
+                    last_change_timestamp=self.last_change_timestamp,
+                    seconds_elapsed=self.seconds_elapsed,
+                    minutes_elapsed=self.minutes_elapsed
+               )
+
+
+def compute_timeout_in_minutes(counter: SuggestionsCounter, max_suggestions: int, max_suggestions_timeout: int) -> int:
+    """
+    Compute timeout applying a multiplier taking into account the value of the counter.
+
+    Example of calculations:
+
+    MAX_SUGGESTIONS = 3
+    MAX_SUGGESTIONS_TIMEOUT = 720
+
+    counter    multiplier    timeout_in_minutes
+    -------------------------------------------
+    1          1             720
+    2          1             720
+    3          1             720
+    4          1             720
+    5          2             1440
+    6          3             2160
+    7          4             2880
+    ..         ..            ..
+    """
+    multiplier = counter.value - max_suggestions if counter.value > max_suggestions else 1
+    return max_suggestions_timeout * multiplier

tests/test_timeoututil.py

@@ -0,0 +1,45 @@
+import pytest
+from unittest import mock
+from timeoututil import SuggestionsCounter, compute_timeout_in_minutes
+
+
+@pytest.fixture
+def counter() -> SuggestionsCounter:
+    return SuggestionsCounter()
+
+
+class TestSuggestionsCounter:
+    @staticmethod
+    def test_increment(counter):
+        assert counter.value == 0
+        for i in range(1, 10):
+            assert counter.increment() == counter.value == i
+        counter.reset()
+        assert counter.value == 0
+
+    @staticmethod
+    def test_elapsed_time_properties():
+        seconds = [15, 55, 60, 72, 125, 250]
+        minutes = [0, 0, 1, 1, 2, 4]
+
+        def seconds_generator():
+            yield 0
+            for n in seconds:
+                yield n
+                yield n
+
+        with mock.patch("time.time", side_effect=seconds_generator()):
+            counter = SuggestionsCounter()
+            for expected_seconds, expected_minutes in zip(seconds, minutes):
+                assert counter.seconds_elapsed == expected_seconds
+                assert counter.minutes_elapsed == expected_minutes
+
+
+def test_compute_timeout_in_minutes(counter):
+    max_suggestions = 3
+    max_suggestions_timeout = 720
+
+    results = [720, 720, 720, 720, 1440, 2160, 2880]
+    for i, expected in zip(range(1, 10), results):
+        assert i == counter.increment()
+        assert compute_timeout_in_minutes(counter, max_suggestions, max_suggestions_timeout) == expected