forms.py
480 lines
| 15.8 KiB
| text/x-python
|
PythonLexer
/ boards / forms.py
|
r1299 | import hashlib | ||
|
r69 | import re | ||
|
r527 | import time | ||
|
r1372 | import logging | ||
|
r1428 | |||
|
r1372 | import pytz | ||
|
r1296 | |||
|
r14 | from django import forms | ||
|
r954 | from django.core.files.uploadedfile import SimpleUploadedFile | ||
|
r1077 | from django.core.exceptions import ObjectDoesNotExist | ||
|
r1483 | from django.forms.utils import ErrorList | ||
|
r1333 | from django.utils.translation import ugettext_lazy as _, ungettext_lazy | ||
|
r1428 | from django.utils import timezone | ||
|
r527 | |||
|
r1494 | from boards.abstracts.settingsmanager import get_settings_manager | ||
|
r1500 | from boards.abstracts.attachment_alias import get_image_by_alias | ||
|
r438 | from boards.mdx_neboard import formatters | ||
|
r1511 | from boards.models.attachment.downloaders import download | ||
|
r386 | from boards.models.post import TITLE_MAX_LENGTH | ||
|
r1077 | from boards.models import Tag, Post | ||
|
r1371 | from boards.utils import validate_file_size, get_file_mimetype, \ | ||
FILE_EXTENSION_DELIMITER | ||||
|
r35 | from neboard import settings | ||
|
r333 | import boards.settings as board_settings | ||
|
r1299 | import neboard | ||
|
r76 | |||
|
r1428 | POW_HASH_LENGTH = 16 | ||
|
r1447 | POW_LIFE_MINUTES = 5 | ||
|
r1428 | |||
|
r937 | REGEX_TAGS = re.compile(r'^[\w\s\d]+$', re.UNICODE) | ||
|
r1429 | REGEX_USERNAMES = re.compile(r'^[\w\s\d,]+$', re.UNICODE) | ||
|
r937 | |||
|
r642 | VETERAN_POSTING_DELAY = 5 | ||
|
r517 | ATTRIBUTE_PLACEHOLDER = 'placeholder' | ||
|
r1006 | ATTRIBUTE_ROWS = 'rows' | ||
|
r517 | |||
LAST_POST_TIME = 'last_post_time' | ||||
LAST_LOGIN_TIME = 'last_login_time' | ||||
|
r937 | TEXT_PLACEHOLDER = _('Type message here. Use formatting panel for more advanced usage.') | ||
|
r1093 | TAGS_PLACEHOLDER = _('music images i_dont_like_tags') | ||
|
r211 | |||
|
r527 | LABEL_TITLE = _('Title') | ||
LABEL_TEXT = _('Text') | ||||
|
r566 | LABEL_TAG = _('Tag') | ||
|
r718 | LABEL_SEARCH = _('Search') | ||
|
r566 | |||
|
r1333 | ERROR_SPEED = 'Please wait %(delay)d second before sending message' | ||
ERROR_SPEED_PLURAL = 'Please wait %(delay)d seconds before sending message' | ||||
|
r1105 | |||
|
r566 | TAG_MAX_LENGTH = 20 | ||
|
r1006 | TEXTAREA_ROWS = 4 | ||
|
r1367 | TRIPCODE_DELIM = '#' | ||
|
r1372 | # TODO Maybe this may be converted into the database table? | ||
MIMETYPE_EXTENSIONS = { | ||||
'image/jpeg': 'jpeg', | ||||
'image/png': 'png', | ||||
'image/gif': 'gif', | ||||
'video/webm': 'webm', | ||||
'application/pdf': 'pdf', | ||||
'x-diff': 'diff', | ||||
|
r1373 | 'image/svg+xml': 'svg', | ||
|
r1382 | 'application/x-shockwave-flash': 'swf', | ||
|
r1454 | 'image/x-ms-bmp': 'bmp', | ||
'image/bmp': 'bmp', | ||||
|
r1372 | } | ||
|
r153 | |||
|
r1065 | def get_timezones(): | ||
timezones = [] | ||||
for tz in pytz.common_timezones: | ||||
timezones.append((tz, tz),) | ||||
return timezones | ||||
|
r438 | class FormatPanel(forms.Textarea): | ||
|
r937 | """ | ||
Panel for text formatting. Consists of buttons to add different tags to the | ||||
form text area. | ||||
""" | ||||
|
r438 | def render(self, name, value, attrs=None): | ||
output = '<div id="mark-panel">' | ||||
for formatter in formatters: | ||||
|
r769 | output += '<span class="mark_btn"' + \ | ||
' onClick="addMarkToMsg(\'' + formatter.format_left + \ | ||||
|
r438 | '\', \'' + formatter.format_right + '\')">' + \ | ||
formatter.preview_left + formatter.name + \ | ||||
|
r769 | formatter.preview_right + '</span>' | ||
|
r438 | |||
output += '</div>' | ||||
|
r1408 | output += super(FormatPanel, self).render(name, value, attrs=attrs) | ||
|
r438 | |||
return output | ||||
|
r76 | class PlainErrorList(ErrorList): | ||
def __unicode__(self): | ||||
return self.as_text() | ||||
def as_text(self): | ||||
|
r769 | return ''.join(['(!) %s ' % e for e in self]) | ||
|
r76 | |||
|
r16 | |||
|
r205 | class NeboardForm(forms.Form): | ||
|
r937 | """ | ||
Form with neboard-specific formatting. | ||||
""" | ||||
|
r1654 | required_css_class = 'required-field' | ||
|
r205 | |||
|
r426 | def as_div(self): | ||
|
r438 | """ | ||
Returns this form rendered as HTML <as_div>s. | ||||
""" | ||||
|
r205 | return self._html_output( | ||
|
r425 | # TODO Do not show hidden rows in the list here | ||
|
r1132 | normal_row='<div class="form-row">' | ||
'<div class="form-label">' | ||||
|
r205 | '%(label)s' | ||
|
r1132 | '</div>' | ||
'<div class="form-input">' | ||||
|
r205 | '%(field)s' | ||
|
r1132 | '</div>' | ||
'</div>' | ||||
|
r680 | '<div class="form-row">' | ||
|
r205 | '%(help_text)s' | ||
'</div>', | ||||
|
r426 | error_row='<div class="form-row">' | ||
'<div class="form-label"></div>' | ||||
'<div class="form-errors">%s</div>' | ||||
'</div>', | ||||
row_ender='</div>', | ||||
help_text_html='%s', | ||||
|
r205 | errors_on_separate_row=True) | ||
|
r533 | def as_json_errors(self): | ||
errors = [] | ||||
|
r771 | for name, field in list(self.fields.items()): | ||
|
r533 | if self[name].errors: | ||
errors.append({ | ||||
'field': name, | ||||
'errors': self[name].errors.as_text(), | ||||
}) | ||||
return errors | ||||
|
r438 | |||
|
r205 | class PostForm(NeboardForm): | ||
|
r76 | |||
|
r232 | title = forms.CharField(max_length=TITLE_MAX_LENGTH, required=False, | ||
|
r1299 | label=LABEL_TITLE, | ||
widget=forms.TextInput( | ||||
attrs={ATTRIBUTE_PLACEHOLDER: | ||||
'test#tripcode'})) | ||||
|
r517 | text = forms.CharField( | ||
|
r1006 | widget=FormatPanel(attrs={ | ||
ATTRIBUTE_PLACEHOLDER: TEXT_PLACEHOLDER, | ||||
ATTRIBUTE_ROWS: TEXTAREA_ROWS, | ||||
}), | ||||
|
r527 | required=False, label=LABEL_TEXT) | ||
|
r1273 | file = forms.FileField(required=False, label=_('File'), | ||
|
r721 | widget=forms.ClearableFileInput( | ||
|
r1273 | attrs={'accept': 'file/*'})) | ||
file_url = forms.CharField(required=False, label=_('File URL'), | ||||
|
r954 | widget=forms.TextInput( | ||
attrs={ATTRIBUTE_PLACEHOLDER: | ||||
'http://example.com/image.png'})) | ||||
|
r29 | |||
|
r207 | # This field is for spam prevention only | ||
|
r232 | email = forms.CharField(max_length=100, required=False, label=_('e-mail'), | ||
widget=forms.TextInput(attrs={ | ||||
'class': 'form-email'})) | ||||
|
r1077 | threads = forms.CharField(required=False, label=_('Additional threads'), | ||
widget=forms.TextInput(attrs={ATTRIBUTE_PLACEHOLDER: | ||||
'123 456 789'})) | ||||
|
r1625 | subscribe = forms.BooleanField(required=False, label=_('Subscribe to thread')) | ||
|
r207 | |||
|
r1428 | guess = forms.CharField(widget=forms.HiddenInput(), required=False) | ||
timestamp = forms.CharField(widget=forms.HiddenInput(), required=False) | ||||
iteration = forms.CharField(widget=forms.HiddenInput(), required=False) | ||||
|
r153 | session = None | ||
|
r271 | need_to_ban = False | ||
|
r1500 | image = None | ||
|
r153 | |||
|
r1371 | def _update_file_extension(self, file): | ||
if file: | ||||
|
r1372 | mimetype = get_file_mimetype(file) | ||
extension = MIMETYPE_EXTENSIONS.get(mimetype) | ||||
if extension: | ||||
filename = file.name.split(FILE_EXTENSION_DELIMITER, 1)[0] | ||||
new_filename = filename + FILE_EXTENSION_DELIMITER + extension | ||||
|
r1371 | |||
|
r1372 | file.name = new_filename | ||
else: | ||||
logger = logging.getLogger('boards.forms.extension') | ||||
logger.info('Unrecognized file mimetype: {}'.format(mimetype)) | ||||
|
r1371 | |||
|
r76 | def clean_title(self): | ||
title = self.cleaned_data['title'] | ||||
if title: | ||||
if len(title) > TITLE_MAX_LENGTH: | ||||
|
r211 | raise forms.ValidationError(_('Title must have less than %s ' | ||
'characters') % | ||||
str(TITLE_MAX_LENGTH)) | ||||
|
r76 | return title | ||
|
r29 | def clean_text(self): | ||
|
r678 | text = self.cleaned_data['text'].strip() | ||
|
r29 | if text: | ||
|
r1153 | max_length = board_settings.get_int('Forms', 'MaxTextLength') | ||
if len(text) > max_length: | ||||
|
r211 | raise forms.ValidationError(_('Text must have less than %s ' | ||
|
r1153 | 'characters') % str(max_length)) | ||
|
r29 | return text | ||
|
r1273 | def clean_file(self): | ||
file = self.cleaned_data['file'] | ||||
|
r954 | |||
|
r1273 | if file: | ||
|
r1328 | validate_file_size(file.size) | ||
|
r1371 | self._update_file_extension(file) | ||
|
r954 | |||
|
r1273 | return file | ||
|
r954 | |||
|
r1273 | def clean_file_url(self): | ||
url = self.cleaned_data['file_url'] | ||||
|
r954 | |||
|
r1273 | file = None | ||
|
r1500 | |||
|
r954 | if url: | ||
|
r1500 | file = get_image_by_alias(url, self.session) | ||
self.image = file | ||||
|
r954 | |||
|
r1500 | if file is not None: | ||
return | ||||
if file is None: | ||||
file = self._get_file_from_url(url) | ||||
if not file: | ||||
raise forms.ValidationError(_('Invalid URL')) | ||||
else: | ||||
validate_file_size(file.size) | ||||
|
r1371 | self._update_file_extension(file) | ||
|
r527 | |||
|
r1273 | return file | ||
|
r29 | |||
|
r1077 | def clean_threads(self): | ||
threads_str = self.cleaned_data['threads'] | ||||
if len(threads_str) > 0: | ||||
threads_id_list = threads_str.split(' ') | ||||
threads = list() | ||||
for thread_id in threads_id_list: | ||||
try: | ||||
thread = Post.objects.get(id=int(thread_id)) | ||||
|
r1414 | if not thread.is_opening() or thread.get_thread().is_archived(): | ||
|
r1077 | raise ObjectDoesNotExist() | ||
threads.append(thread) | ||||
except (ObjectDoesNotExist, ValueError): | ||||
raise forms.ValidationError(_('Invalid additional thread list')) | ||||
return threads | ||||
|
r29 | def clean(self): | ||
cleaned_data = super(PostForm, self).clean() | ||||
|
r211 | if cleaned_data['email']: | ||
|
r1632 | if board_settings.get_bool('Forms', 'Autoban'): | ||
self.need_to_ban = True | ||||
|
r211 | raise forms.ValidationError('A human cannot enter a hidden field') | ||
|
r207 | |||
|
r151 | if not self.errors: | ||
|
r1273 | self._clean_text_file() | ||
|
r77 | |||
|
r1428 | limit_speed = board_settings.get_bool('Forms', 'LimitPostingSpeed') | ||
|
r1607 | limit_first = board_settings.get_bool('Forms', 'LimitFirstPosting') | ||
|
r1494 | |||
settings_manager = get_settings_manager(self) | ||||
|
r1607 | if not self.errors and limit_speed or (limit_first and not settings_manager.get_setting('confirmed_user')): | ||
|
r1428 | pow_difficulty = board_settings.get_int('Forms', 'PowDifficulty') | ||
|
r1494 | if pow_difficulty > 0: | ||
|
r1607 | # PoW-based | ||
|
r1494 | 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']) | ||||
|
r1428 | else: | ||
|
r1607 | # Time-based | ||
|
r1428 | self._validate_posting_speed() | ||
|
r1494 | settings_manager.set_setting('confirmed_user', True) | ||
|
r153 | |||
|
r76 | return cleaned_data | ||
|
r1273 | def get_file(self): | ||
|
r954 | """ | ||
|
r1273 | Gets file from form or URL. | ||
|
r954 | """ | ||
|
r1273 | file = self.cleaned_data['file'] | ||
return file or self.cleaned_data['file_url'] | ||||
|
r954 | |||
|
r1293 | def get_tripcode(self): | ||
|
r1299 | title = self.cleaned_data['title'] | ||
|
r1367 | if title is not None and TRIPCODE_DELIM in title: | ||
code = title.split(TRIPCODE_DELIM, maxsplit=1)[1] + neboard.settings.SECRET_KEY | ||||
tripcode = hashlib.md5(code.encode()).hexdigest() | ||||
else: | ||||
tripcode = '' | ||||
return tripcode | ||||
|
r1299 | |||
def get_title(self): | ||||
title = self.cleaned_data['title'] | ||||
|
r1367 | if title is not None and TRIPCODE_DELIM in title: | ||
return title.split(TRIPCODE_DELIM, maxsplit=1)[0] | ||||
|
r1299 | else: | ||
return title | ||||
|
r1293 | |||
|
r1500 | def get_images(self): | ||
if self.image: | ||||
return [self.image] | ||||
else: | ||||
return [] | ||||
|
r1625 | def is_subscribe(self): | ||
return self.cleaned_data['subscribe'] | ||||
|
r1273 | def _clean_text_file(self): | ||
|
r76 | text = self.cleaned_data.get('text') | ||
|
r1273 | file = self.get_file() | ||
|
r1500 | images = self.get_images() | ||
|
r29 | |||
|
r1500 | if (not text) and (not file) and len(images) == 0: | ||
|
r1273 | error_message = _('Either text or file must be entered.') | ||
|
r77 | self._errors['text'] = self.error_class([error_message]) | ||
|
r153 | |||
def _validate_posting_speed(self): | ||||
can_post = True | ||||
|
r1610 | posting_delay = board_settings.get_int('Forms', 'PostingDelay') | ||
|
r642 | |||
|
r1153 | if board_settings.get_bool('Forms', 'LimitPostingSpeed'): | ||
|
r153 | now = time.time() | ||
|
r1105 | |||
current_delay = 0 | ||||
|
r1333 | if LAST_POST_TIME not in self.session: | ||
|
r1105 | self.session[LAST_POST_TIME] = now | ||
|
r153 | |||
|
r1105 | need_delay = True | ||
else: | ||||
last_post_time = self.session.get(LAST_POST_TIME) | ||||
current_delay = int(now - last_post_time) | ||||
|
r153 | |||
|
r1105 | need_delay = current_delay < posting_delay | ||
if need_delay: | ||||
|
r1333 | delay = posting_delay - current_delay | ||
error_message = ungettext_lazy(ERROR_SPEED, ERROR_SPEED_PLURAL, | ||||
delay) % {'delay': delay} | ||||
|
r153 | self._errors['text'] = self.error_class([error_message]) | ||
can_post = False | ||||
|
r1109 | if can_post: | ||
self.session[LAST_POST_TIME] = now | ||||
|
r29 | |||
|
r1273 | def _get_file_from_url(self, url: str) -> SimpleUploadedFile: | ||
|
r954 | """ | ||
|
r1273 | Gets an file file from URL. | ||
|
r954 | """ | ||
try: | ||||
|
r1553 | return download(url) | ||
|
r1328 | except forms.ValidationError as e: | ||
raise e | ||||
|
r1276 | except Exception as e: | ||
|
r1433 | raise forms.ValidationError(e) | ||
|
r954 | |||
|
r1428 | def _validate_hash(self, timestamp: str, iteration: str, guess: str, message: str): | ||
post_time = timezone.datetime.fromtimestamp( | ||||
int(timestamp[:-3]), tz=timezone.get_current_timezone()) | ||||
payload = timestamp + message.replace('\r\n', '\n') | ||||
difficulty = board_settings.get_int('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._errors['text'] = self.error_class( | ||||
[_('Invalid PoW.')]) | ||||
|
r29 | |||
|
r1494 | |||
|
r29 | class ThreadForm(PostForm): | ||
|
r232 | |||
|
r517 | tags = forms.CharField( | ||
widget=forms.TextInput(attrs={ATTRIBUTE_PLACEHOLDER: TAGS_PLACEHOLDER}), | ||||
|
r679 | max_length=100, label=_('Tags'), required=True) | ||
|
r1434 | monochrome = forms.BooleanField(label=_('Monochrome'), required=False) | ||
|
r31 | |||
def clean_tags(self): | ||||
|
r678 | tags = self.cleaned_data['tags'].strip() | ||
|
r69 | |||
|
r937 | if not tags or not REGEX_TAGS.match(tags): | ||
|
r679 | raise forms.ValidationError( | ||
_('Inappropriate characters in tags.')) | ||||
|
r31 | |||
|
r1658 | default_tag_name = board_settings.get('Forms', 'DefaultTag')\ | ||
.strip().lower() | ||||
|
r922 | required_tag_exists = False | ||
|
r1348 | tag_set = set() | ||
for tag_string in tags.split(): | ||||
|
r1658 | if tag_string.strip().lower() == default_tag_name: | ||
required = True | ||||
required_tag_exitst = True | ||||
else: | ||||
required = False | ||||
tag, created = Tag.objects.get_or_create( | ||||
name=tag_string.strip().lower(), required=required) | ||||
|
r1348 | tag_set.add(tag) | ||
# If this is a new tag, don't check for its parents because nobody | ||||
# added them yet | ||||
if not created: | ||||
|
r1361 | tag_set |= set(tag.get_all_parents()) | ||
|
r1348 | |||
for tag in tag_set: | ||||
if tag.required: | ||||
|
r1102 | required_tag_exists = True | ||
|
r937 | break | ||
|
r922 | |||
|
r1658 | # Use default tag if no section exists | ||
|
r922 | if not required_tag_exists: | ||
|
r1655 | default_tag, created = Tag.objects.get_or_create( | ||
name=default_tag_name, required=True) | ||||
tag_set.add(default_tag) | ||||
|
r922 | |||
|
r1348 | return tag_set | ||
|
r31 | |||
def clean(self): | ||||
cleaned_data = super(ThreadForm, self).clean() | ||||
|
r35 | return cleaned_data | ||
|
r1434 | def is_monochrome(self): | ||
return self.cleaned_data['monochrome'] | ||||
|
r35 | |||
|
r205 | class SettingsForm(NeboardForm): | ||
|
r1065 | theme = forms.ChoiceField(choices=settings.THEMES, label=_('Theme')) | ||
|
r1296 | image_viewer = forms.ChoiceField(choices=settings.IMAGE_VIEWERS, label=_('Image view mode')) | ||
|
r990 | username = forms.CharField(label=_('User name'), required=False) | ||
|
r1065 | timezone = forms.ChoiceField(choices=get_timezones(), label=_('Time zone')) | ||
|
r144 | |||
|
r995 | def clean_username(self): | ||
username = self.cleaned_data['username'] | ||||
|
r1429 | if username and not REGEX_USERNAMES.match(username): | ||
|
r995 | raise forms.ValidationError(_('Inappropriate characters.')) | ||
return username | ||||
|
r144 | |||
|
r718 | class SearchForm(NeboardForm): | ||
|
r729 | query = forms.CharField(max_length=500, label=LABEL_SEARCH, required=False) | ||