Move posting validation to class-based validators for PoW and Time-based

boards/forms/validators.py

@@ -0,0 +1,131 b''
+import time
+import hashlib
+
+from django.core.cache import cache
+from django.utils.translation import ugettext_lazy as _, ungettext_lazy
+
+from boards import settings as board_settings
+from boards.settings import SECTION_FORMS
+
+
+LAST_POST_TIME = 'last_post_time'
+LAST_LOGIN_TIME = 'last_login_time'
+
+POW_HASH_LENGTH = 16
+POW_LIFE_MINUTES = 5
+
+ERROR_SPEED = 'Please wait %(delay)d second before sending message'
+ERROR_SPEED_PLURAL = 'Please wait %(delay)d seconds before sending message'
+ERROR_INVALID_POW = 'Invalid PoW.'
+
+
+class Validator:
+    def __init__(self, session):
+        self.errors = []
+        self.session = session
+
+    def validate(self) -> None:
+        """
+        Runs validation. If something fails, should add an error to the list.
+        """
+        pass
+
+    def get_errors(self):
+        return self.errors
+
+    def get_session(self):
+        return self.session
+
+    def add_error(self, error):
+        self.errors.append(error)
+
+
+class TimeValidator(Validator):
+    """
+    Validates the posting speed. New post must be sent no less than after some
+    specific amount of time defined in settings.
+    """
+    def validate(self):
+        can_post = True
+
+        posting_delay = board_settings.get_int(SECTION_FORMS, 'PostingDelay')
+
+        if board_settings.get_bool(SECTION_FORMS, 'LimitPostingSpeed'):
+            now = time.time()
+
+            current_delay = 0
+
+            if LAST_POST_TIME not in self.get_session():
+                self.get_session()[LAST_POST_TIME] = now
+
+                need_delay = True
+            else:
+                last_post_time = self._get_last_post_time()
+                current_delay = int(now - last_post_time)
+
+                need_delay = current_delay < posting_delay
+
+            self._set_session_cache(LAST_POST_TIME, now)
+
+            if need_delay:
+                delay = posting_delay - current_delay
+                error_message = ungettext_lazy(ERROR_SPEED, ERROR_SPEED_PLURAL,
+                                               delay) % {'delay': delay}
+                self.add_error(error_message)
+
+                can_post = False
+
+            if can_post:
+                self.get_session()[LAST_POST_TIME] = now
+            else:
+                # Reset the time since posting failed
+                self._set_session_cache(LAST_POST_TIME, self.get_session()[LAST_POST_TIME])
+
+    def _get_cache_key(self, key):
+        return '{}_{}'.format(self.get_session().session_key, key)
+
+    def _set_session_cache(self, key, value):
+        cache.set(self._get_cache_key(key), value)
+
+    def _get_session_cache(self, key):
+        return cache.get(self._get_cache_key(key))
+
+    def _get_last_post_time(self):
+        last = self._get_session_cache(LAST_POST_TIME)
+        if last is None:
+            last = self.get_session().get(LAST_POST_TIME)
+        return last
+
+
+class PowValidator(Validator):
+    """
+    Validates prof of work. A hash is computed on the client side and validated
+    here, all data must match. PoW difficulty is set in the settings.
+    """
+    def __init__(self, session, timestamp, iteration, guess, text):
+        super().__init__(session)
+
+        self.timestamp = timestamp
+        self.iteration = iteration
+        self.guess = guess
+
+        self.text = text
+
+    def validate(self):
+        if self.timestamp and self.iteration and self.guess:
+            self._validate_hash(self.timestamp, self.iteration,
+                                self.guess, self.text)
+        else:
+            self.add_error(_(ERROR_INVALID_POW))
+
+    def _validate_hash(self, timestamp: str, iteration: str, guess: str, message: str):
+        payload = timestamp + message.replace('\r\n', '\n')
+        difficulty = board_settings.get_int(SECTION_FORMS, 'PowDifficulty')
+        target = str(int(2 ** (POW_HASH_LENGTH * 3) / difficulty))
+        if len(target) < POW_HASH_LENGTH:
+            target = '0' * (POW_HASH_LENGTH - len(target)) + target
+
+        computed_guess = hashlib.sha256((payload + iteration).encode()) \
+                             .hexdigest()[0:POW_HASH_LENGTH]
+        if guess != computed_guess or guess > target:
+            self.add_error(_('Invalid PoW.'))

boards/abstracts/settingsmanager.py

@@ -19,6 +19,7 b" SETTING_IMAGE_VIEWER = 'image_viewer'"
 SETTING_IMAGES = 'images_aliases'
 SETTING_ONLY_FAVORITES = 'only_favorites'
 SETTING_LAST_POSTS = 'last_posts'
+SETTING_CONFIRMED_USER = 'confirmed_user'
 
 DEFAULT_THEME = 'md'
 

boards/forms/__init__.py

@@ -1,12 +1,9 b''
 import logging
-import time
 
-import hashlib
 import pytz
 import re
 from PIL import Image
 from django import forms
-from django.core.cache import cache
 from django.core.files.images import get_image_dimensions
 from django.core.files.uploadedfile import SimpleUploadedFile, UploadedFile
 from django.forms.utils import ErrorList
@@ -15,9 +12,11 b' from django.utils.translation import uge'
 import boards.settings as board_settings
 from boards import utils
 from boards.abstracts.constants import REGEX_TAGS
-from boards.abstracts.settingsmanager import get_settings_manager
+from boards.abstracts.settingsmanager import get_settings_manager, \
+    SETTING_CONFIRMED_USER
 from boards.abstracts.sticker_factory import get_attachment_by_alias
 from boards.forms.fields import UrlFileField
+from boards.forms.validators import TimeValidator, PowValidator
 from boards.mdx_neboard import formatters
 from boards.models import Attachment
 from boards.models import Tag
@@ -25,15 +24,13 b' from boards.models.attachment import Sti'
 from boards.models.attachment.downloaders import download, REGEX_MAGNET
 from boards.models.attachment.viewers import FILE_TYPES_IMAGE
 from boards.models.post import TITLE_MAX_LENGTH
+from boards.settings import SECTION_FORMS
 from boards.utils import validate_file_size, get_file_mimetype, \
     FILE_EXTENSION_DELIMITER, get_tripcode_from_text
-from boards.settings import SECTION_FORMS
 
 FORMAT_PANEL_BUTTON = '<span class="mark_btn" ' \
                       'onClick="addMarkToMsg(\'{}\',  \'{}\')">{}{}{}</span>'
 
-POW_HASH_LENGTH = 16
-POW_LIFE_MINUTES = 5
 
 REGEX_USERNAMES = re.compile(r'^[\w\s\d,]+$', re.UNICODE)
 REGEX_URL = re.compile(r'^(http|https|ftp):\/\/', re.UNICODE)
@@ -43,8 +40,6 b' VETERAN_POSTING_DELAY = 5'
 ATTRIBUTE_PLACEHOLDER = 'placeholder'
 ATTRIBUTE_ROWS = 'rows'
 
-LAST_POST_TIME = 'last_post_time'
-LAST_LOGIN_TIME = 'last_login_time'
 TEXT_PLACEHOLDER = _('Type message here. Use formatting panel for more advanced usage.')
 TAGS_PLACEHOLDER = _('music images i_dont_like_tags')
 
@@ -56,8 +51,6 b" LABEL_FILE = _('File')"
 LABEL_DUPLICATES = _('Check for duplicates')
 LABEL_URL = _('Do not download URLs')
 
-ERROR_SPEED = 'Please wait %(delay)d second before sending message'
-ERROR_SPEED_PLURAL = 'Please wait %(delay)d seconds before sending message'
 ERROR_MANY_FILES = 'You can post no more than %(files)d file.'
 ERROR_MANY_FILES_PLURAL = 'You can post no more than %(files)d files.'
 ERROR_DUPLICATES = 'Some files are already present on the board.'
@@ -249,18 +242,21 b' class PostForm(NeboardForm):'
         limit_first = board_settings.get_bool(SECTION_FORMS, 'LimitFirstPosting')
 
         settings_manager = get_settings_manager(self)
-        if not self.errors and limit_speed or (limit_first and not settings_manager.get_setting('confirmed_user')):
+        if not self.errors and limit_speed or (limit_first and not settings_manager.get_setting(SETTING_CONFIRMED_USER)):
             pow_difficulty = board_settings.get_int(SECTION_FORMS, 'PowDifficulty')
             if pow_difficulty > 0:
-                # PoW-based
-                if cleaned_data['timestamp'] \
-                        and cleaned_data['iteration'] and cleaned_data['guess'] \
-                        and not settings_manager.get_setting('confirmed_user'):
-                    self._validate_hash(cleaned_data['timestamp'], cleaned_data['iteration'], cleaned_data['guess'], cleaned_data['text'])
+                validator = PowValidator(
+                    self.session, cleaned_data['timestamp'],
+                    cleaned_data['iteration'], cleaned_data['guess'],
+                    cleaned_data['text'])
             else:
-                # Time-based
-                self._validate_posting_speed()
-            settings_manager.set_setting('confirmed_user', True)
+                validator = TimeValidator(self.session)
+
+            validator.validate()
+            for error in validator.get_errors():
+                self._add_general_error(error)
+
+            settings_manager.set_setting(SETTING_CONFIRMED_USER, True)
         if self.cleaned_data['download_mode'] == DOWN_MODE_DOWNLOAD_UNIQUE:
             self._check_file_duplicates(self.get_files())
 
@@ -399,57 +395,6 b' class PostForm(NeboardForm):'
             error_message = _('Either text or file must be entered.')
             self._add_general_error(error_message)
 
-    def _get_cache_key(self, key):
-        return '{}_{}'.format(self.session.session_key, key)
-
-    def _set_session_cache(self, key, value):
-        cache.set(self._get_cache_key(key), value)
-
-    def _get_session_cache(self, key):
-        return cache.get(self._get_cache_key(key))
-
-    def _get_last_post_time(self):
-        last = self._get_session_cache(LAST_POST_TIME)
-        if last is None:
-            last = self.session.get(LAST_POST_TIME)
-        return last
-
-    def _validate_posting_speed(self):
-        can_post = True
-
-        posting_delay = board_settings.get_int(SECTION_FORMS, 'PostingDelay')
-
-        if board_settings.get_bool(SECTION_FORMS, 'LimitPostingSpeed'):
-            now = time.time()
-
-            current_delay = 0
-
-            if LAST_POST_TIME not in self.session:
-                self.session[LAST_POST_TIME] = now
-
-                need_delay = True
-            else:
-                last_post_time = self._get_last_post_time()
-                current_delay = int(now - last_post_time)
-
-                need_delay = current_delay < posting_delay
-
-            self._set_session_cache(LAST_POST_TIME, now)
-
-            if need_delay:
-                delay = posting_delay - current_delay
-                error_message = ungettext_lazy(ERROR_SPEED, ERROR_SPEED_PLURAL,
-                                               delay) % {'delay': delay}
-                self._add_general_error(error_message)
-
-                can_post = False
-
-            if can_post:
-                self.session[LAST_POST_TIME] = now
-            else:
-                # Reset the time since posting failed
-                self._set_session_cache(LAST_POST_TIME, self.session[LAST_POST_TIME])
-
     def _get_file_from_url(self, url: str) -> SimpleUploadedFile:
         """
         Gets an file file from URL.
@@ -462,18 +407,6 b' class PostForm(NeboardForm):'
         except Exception as e:
             raise forms.ValidationError(e)
 
-    def _validate_hash(self, timestamp: str, iteration: str, guess: str, message: str):
-        payload = timestamp + message.replace('\r\n', '\n')
-        difficulty = board_settings.get_int(SECTION_FORMS, 'PowDifficulty')
-        target = str(int(2 ** (POW_HASH_LENGTH * 3) / difficulty))
-        if len(target) < POW_HASH_LENGTH:
-            target = '0' * (POW_HASH_LENGTH - len(target)) + target
-
-        computed_guess = hashlib.sha256((payload + iteration).encode())\
-                             .hexdigest()[0:POW_HASH_LENGTH]
-        if guess != computed_guess or guess > target:
-            self._add_general_error(_('Invalid PoW.'))
-
     def _check_file_duplicates(self, files):
         for file in files:
             file_hash = utils.get_file_hash(file)