From 4d16c53e3c9c6542b21ca1b84caf7d0ab56dc680 Mon Sep 17 00:00:00 2001 From: Pierre-Yves Chibon Date: Nov 08 2016 11:49:37 +0000 Subject: [PATCH 1/2] Create a PagureForm all forms inherit from and which sets the TIME_LIMIT This will allow us to properly implement the support for the configuration key `WTF_CSRF_TIME_LIMIT` which isn't implemented as expected in the versions of flask_wtf we are using. --- diff --git a/pagure/forms.py b/pagure/forms.py index a2d7316..bbdde4b 100644 --- a/pagure/forms.py +++ b/pagure/forms.py @@ -12,6 +12,7 @@ # pylint: disable=no-init # pylint: disable=super-on-old-class +import datetime import re import tempfile @@ -35,6 +36,20 @@ PROJECT_NAME_REGEX = \ '^[a-zA-z0-9_][a-zA-Z0-9-_]*$' +class PagureForm(FlaskForm): + """ Local form allowing us to form set the time limit. """ + + def __init__(self, *args, **kwargs): + delta = pagure.APP.config.get('WTF_CSRF_TIME_LIMIT', 3600) + if delta \ + and (not hasattr(wtf, '__version__') \ + or tuple(wtf.__version__.split('.')) <= (0,10,0)): + self.TIME_LIMIT = datetime.timedelta(seconds=delta) + else: + self.TIME_LIMIT = delta + super(PagureForm, self).__init__(*args, **kwargs) + + def convert_value(val): """ Convert the provided values to strings when possible. """ if val: @@ -89,7 +104,7 @@ def ssh_key_validator(form, field): raise wtforms.ValidationError('Invalid SSH keys') -class ProjectFormSimplified(FlaskForm): +class ProjectFormSimplified(PagureForm): ''' Form to edit the description of a project. ''' description = wtforms.TextField( 'Description *', @@ -145,7 +160,7 @@ class ProjectForm(ProjectFormSimplified): self.namespace.choices.insert(0, ('', '')) -class IssueFormSimplied(FlaskForm): +class IssueFormSimplied(PagureForm): ''' Form to create or edit an issue. ''' title = wtforms.TextField( 'Title*', @@ -181,7 +196,7 @@ class IssueForm(IssueFormSimplied): ] -class RequestPullForm(FlaskForm): +class RequestPullForm(PagureForm): ''' Form to create a request pull. ''' title = wtforms.TextField( 'Title*', @@ -207,7 +222,7 @@ class RemoteRequestPullForm(RequestPullForm): ) -class AddIssueTagForm(FlaskForm): +class AddIssueTagForm(PagureForm): ''' Form to add a comment to an issue. ''' tag = wtforms.TextField( 'tag', @@ -219,7 +234,7 @@ class AddIssueTagForm(FlaskForm): ) -class StatusForm(FlaskForm): +class StatusForm(PagureForm): ''' Form to add/change the status of an issue. ''' status = wtforms.SelectField( 'Status', @@ -249,7 +264,7 @@ class StatusForm(FlaskForm): self.close_status.choices.insert(0, ('', '')) -class NewTokenForm(FlaskForm): +class NewTokenForm(PagureForm): ''' Form to add/change the status of an issue. ''' acls = wtforms.SelectMultipleField( 'ACLs', @@ -270,7 +285,7 @@ class NewTokenForm(FlaskForm): -class UpdateIssueForm(FlaskForm): +class UpdateIssueForm(PagureForm): ''' Form to add a comment to an issue. ''' tag = wtforms.TextField( 'tag', @@ -350,7 +365,7 @@ class UpdateIssueForm(FlaskForm): self.close_status.choices.insert(0, ('', '')) -class AddPullRequestCommentForm(FlaskForm): +class AddPullRequestCommentForm(PagureForm): ''' Form to add a comment to a pull-request. ''' commit = wtforms.HiddenField('commit identifier') filename = wtforms.HiddenField('file changed') @@ -363,7 +378,7 @@ class AddPullRequestCommentForm(FlaskForm): ) -class AddPullRequestFlagForm(FlaskForm): +class AddPullRequestFlagForm(PagureForm): ''' Form to add a flag to a pull-request. ''' username = wtforms.TextField( 'Username', [wtforms.validators.Required()]) @@ -377,7 +392,7 @@ class AddPullRequestFlagForm(FlaskForm): 'UID', [wtforms.validators.optional()]) -class UserSettingsForm(FlaskForm): +class UserSettingsForm(PagureForm): ''' Form to create or edit project. ''' ssh_key = wtforms.TextAreaField( 'Public SSH key *', @@ -386,7 +401,7 @@ class UserSettingsForm(FlaskForm): ) -class AddUserForm(FlaskForm): +class AddUserForm(PagureForm): ''' Form to add a user to a project. ''' user = wtforms.TextField( 'Username *', @@ -394,7 +409,7 @@ class AddUserForm(FlaskForm): ) -class AssignIssueForm(FlaskForm): +class AssignIssueForm(PagureForm): ''' Form to assign an user to an issue. ''' assignee = wtforms.TextField( 'Assignee *', @@ -402,7 +417,7 @@ class AssignIssueForm(FlaskForm): ) -class AddGroupForm(FlaskForm): +class AddGroupForm(PagureForm): ''' Form to add a group to a project. ''' group = wtforms.TextField( 'Group *', @@ -413,19 +428,19 @@ class AddGroupForm(FlaskForm): ) -class ConfirmationForm(FlaskForm): +class ConfirmationForm(PagureForm): ''' Simple form used just for CSRF protection. ''' pass -class UploadFileForm(FlaskForm): +class UploadFileForm(PagureForm): ''' Form to upload a file. ''' filestream = wtforms.FileField( 'File', [wtforms.validators.Required(), file_virus_validator]) -class UserEmailForm(FlaskForm): +class UserEmailForm(PagureForm): ''' Form to edit the description of a project. ''' email = wtforms.TextField( 'email', [wtforms.validators.Required()] @@ -442,7 +457,7 @@ class UserEmailForm(FlaskForm): self.email.validators = [wtforms.validators.Required()] -class ProjectCommentForm(FlaskForm): +class ProjectCommentForm(PagureForm): ''' Form to represent project. ''' objid = wtforms.TextField( 'Ticket/Request id', @@ -454,14 +469,14 @@ class ProjectCommentForm(FlaskForm): ) -class CommentForm(FlaskForm): +class CommentForm(PagureForm): ''' Form to upload a file. ''' comment = wtforms.FileField( 'Comment', [wtforms.validators.Required(), file_virus_validator]) -class EditGroupForm(FlaskForm): +class EditGroupForm(PagureForm): """ Form to ask for a password change. """ display_name = wtforms.TextField( 'Group name to display', @@ -507,7 +522,7 @@ class NewGroupForm(EditGroupForm): ] -class EditFileForm(FlaskForm): +class EditFileForm(PagureForm): """ Form used to edit a file. """ content = wtforms.TextAreaField( 'content', [wtforms.validators.Required()]) @@ -534,7 +549,7 @@ class EditFileForm(FlaskForm): ] -class DefaultBranchForm(FlaskForm): +class DefaultBranchForm(PagureForm): """Form to change the default branh for a repository""" branches = wtforms.SelectField( 'default_branch', @@ -554,7 +569,7 @@ class DefaultBranchForm(FlaskForm): ] -class EditCommentForm(FlaskForm): +class EditCommentForm(PagureForm): """ Form to verify that comment is not empty """ update_comment = wtforms.TextAreaField( @@ -563,7 +578,7 @@ class EditCommentForm(FlaskForm): ) -class ForkRepoForm(FlaskForm): +class ForkRepoForm(PagureForm): ''' Form to fork a project in the API. ''' repo = wtforms.TextField( 'The project name', @@ -578,7 +593,7 @@ class ForkRepoForm(FlaskForm): ) -class AddReportForm(FlaskForm): +class AddReportForm(PagureForm): """ Form to verify that comment is not empty """ report_name = wtforms.TextAreaField( @@ -587,7 +602,7 @@ class AddReportForm(FlaskForm): ) -class PublicNotificationForm(FlaskForm): +class PublicNotificationForm(PagureForm): """ Form to verify that comment is not empty """ issue_notifs = wtforms.TextAreaField( @@ -601,9 +616,10 @@ class PublicNotificationForm(FlaskForm): ) -class SubscribtionForm(FlaskForm): +class SubscribtionForm(PagureForm): ''' Form to subscribe or unsubscribe to an issue or a PR. ''' status = wtforms.BooleanField( 'Subscription status', [wtforms.validators.optional()], ) + From 05aa810eb1a8030de01b64a9225b45e23c986f8e Mon Sep 17 00:00:00 2001 From: Pierre-Yves Chibon Date: Nov 08 2016 11:50:56 +0000 Subject: [PATCH 2/2] Add unit-tests to check the behavior of our CSRF tokens --- diff --git a/tests/test_pagure_flask_form.py b/tests/test_pagure_flask_form.py new file mode 100644 index 0000000..46b492d --- /dev/null +++ b/tests/test_pagure_flask_form.py @@ -0,0 +1,94 @@ +# -*- coding: utf-8 -*- + +""" + (c) 2016 - Copyright Red Hat Inc + + Authors: + Pierre-Yves Chibon + +""" + +__requires__ = ['SQLAlchemy >= 0.8'] +import pkg_resources + +import datetime +import unittest +import sys +import time +import os + +import flask +import flask_wtf +from mock import patch + +sys.path.insert(0, os.path.join(os.path.dirname( + os.path.abspath(__file__)), '..')) + +import pagure.forms +import tests + + +class PagureFlaskFormTests(tests.Modeltests): + """ Tests for forms of the flask application """ + + def setUpt(self): + pagure.APP.config['TESTING'] = True + pagure.APP.config['SERVER_NAME'] = 'pagure.org' + pagure.SESSION = self.session + self.app = pagure.APP.test_client() + + def test_csrf_form_no_input(self): + """ Test the CSRF validation if not CSRF is specified. """ + with pagure.APP.test_request_context(method='POST'): + form = pagure.forms.ConfirmationForm() + self.assertFalse(form.validate_on_submit()) + + def test_csrf_form_w_invalid_input(self): + """ Test the CSRF validation with an invalid CSRF specified. """ + with pagure.APP.test_request_context(method='POST'): + form = pagure.forms.ConfirmationForm() + form.csrf_token.data = 'foobar' + self.assertFalse(form.validate_on_submit()) + + def test_csrf_form_w_input(self): + """ Test the CSRF validation with a valid CSRF specified. """ + with pagure.APP.test_request_context(method='POST'): + form = pagure.forms.ConfirmationForm() + form.csrf_token.data = form.csrf_token.current_token + self.assertTrue(form.validate_on_submit()) + + def test_csrf_form_w_expired_input(self): + """ Test the CSRF validation with an expired CSRF specified. """ + with pagure.APP.test_request_context(method='POST'): + form = pagure.forms.ConfirmationForm() + data = form.csrf_token.current_token + _, hmac_csrf = data.split('##', 1) + + # CSRF token expired + if hasattr(flask_wtf, '__version__') and \ + tuple(flask_wtf.__version__.split('.')) >= (0,10,0): + expires = time.time() - 1 + else: + expires = ( + datetime.datetime.now() - datetime.timedelta(minutes=1) + ).strftime('%Y%m%d%H%M%S') + form.csrf_token.data = '%s##%s' % (expires, hmac_csrf) + self.assertFalse(form.validate_on_submit()) + + def test_csrf_form_w_unexpiring_input(self): + """ Test the CSRF validation with a CSRF not expiring. """ + pagure.APP.config['WTF_CSRF_TIME_LIMIT'] = None + with pagure.APP.test_request_context(method='POST'): + form = pagure.forms.ConfirmationForm() + data = form.csrf_token.current_token + _, hmac_csrf = data.split('##', 1) + # CSRF can no longer expire, they have no expiration info + form.csrf_token.data = '##%s' % hmac_csrf + self.assertTrue(form.validate_on_submit()) + + +if __name__ == '__main__': + SUITE = unittest.TestLoader().loadTestsFromTestCase( + PagureFlaskFormTests) + unittest.TextTestRunner(verbosity=2).run(SUITE) +