Merge pull request #172 from colevscode/anti-formspree-spoof

adding check to ensure referrer hostname doesn't match SERVICE_URL

formspree/forms/views.py

@@ -13,7 +13,8 @@
 
 from formspree import settings
 from formspree.app import DB
-from formspree.utils import request_wants_json, jsonerror, IS_VALID_EMAIL
+from formspree.utils import request_wants_json, jsonerror, IS_VALID_EMAIL, \
+                            url_domain
 from helpers import http_form_to_dict, ordered_storage, referrer_to_path, \
                     remove_www, referrer_to_baseurl, sitewide_file_check, \
                     verify_captcha, temp_store_hostname, get_temp_hostname, \
@@ -57,6 +58,8 @@ def send(email_or_string):
 
     sorted_keys = [k for k in sorted_keys if k not in EXCLUDE_KEYS]
 
+    # NOTE: host in this function generally refers to the referrer hostname.
+
     try:
         # Get stored hostname from redis (from captcha)
         host, referrer = get_temp_hostname(received_data['_host_nonce'])
@@ -72,7 +75,7 @@ def send(email_or_string):
             )
         ), 500
 
-    if not host or host == 'www.google.com':
+    if not host:
         if request_wants_json():
             return jsonerror(400, {'error': "Invalid \"Referrer\" header"})
         else:
@@ -150,8 +153,23 @@ def send(email_or_string):
         email = email_or_string.lower()
 
         # get the form for this request
-        form = Form.query.filter_by(hash=HASH(email, host)).first() \
-               or Form(email, host) # or create it if it doesn't exists
+        form = Form.query.filter_by(hash=HASH(email, host)).first()
+
+        # or create it if it doesn't exist
+        if not form:
+            if not url_domain(settings.SERVICE_URL) in host:
+                form = Form(email, host)
+            else:
+                # Bad user is trying to submit a form spoofing formspree.io
+                # Error out silently
+                if request_wants_json():
+                    return jsonerror(400, {'error': "Unable to submit form"})
+                else:
+                    return render_template(
+                        'error.html',
+                        title='Unable to submit form',
+                        text='Sorry'), 400
+
 
         # Check if it has been assigned about using AJAX or not
         assign_ajax(form, request_wants_json())

formspree/utils.py

@@ -70,6 +70,11 @@ def get_url(endpoint, secure=False, **values):
     return path
 
 
+def url_domain(url):
+    parsed = urlparse.urlparse(url)
+    return '.'.join(parsed.netloc.split('.')[-2:])
+
+
 def unix_time_for_12_months_from_now(now=None):
     now = now or datetime.date.today()
     month = now.month - 1 + 12

manage.py

@@ -1,15 +1,10 @@
 import os
-import dotenv
-
-# Must come first, even before some imports. It reads the .env file and put the content as environment variables.
-dotenv.load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))
-
 import datetime
 
 from flask.ext.script import Manager, prompt_bool
 from flask.ext.migrate import Migrate, MigrateCommand
 
-from formspree import create_app, app
+from formspree import create_app, app, settings
 from formspree.app import redis_store
 from formspree.forms.helpers import REDIS_COUNTER_KEY
 from formspree.forms.models import Form
@@ -86,12 +81,15 @@ def monthly_counters(email=None, host=None, id=None, month=datetime.date.today()
         print '%s submissions for %s' % (nsubmissions, form)
 
 
-@manager.command
-def test():
+@manager.option('-t', '--testname', dest='testname', default=None, help='name of test')
+def test(testname=None):
     import unittest
 
     test_loader = unittest.defaultTestLoader
-    test_suite = test_loader.discover('.')
+    if testname:
+        test_suite = test_loader.loadTestsFromName(testname)
+    else:
+        test_suite = test_loader.discover('.')
 
     test_runner = unittest.TextTestRunner()
     test_runner.run(test_suite)

tests/formspree_test_case.py

@@ -26,7 +26,6 @@ def create_app(self):
         settings.SQLALCHEMY_DATABASE_URI = os.getenv('TEST_DATABASE_URL')
         settings.STRIPE_PUBLISHABLE_KEY = settings.STRIPE_TEST_PUBLISHABLE_KEY
         settings.STRIPE_SECRET_KEY = settings.STRIPE_TEST_SECRET_KEY
-        settings.SERVICE_URL = os.getenv('SERVICE_URL')
         settings.PRESERVE_CONTEXT_ON_EXCEPTION = False
         settings.TESTING = True
         return create_app()

tests/test_content_types.py

@@ -12,8 +12,8 @@ class ContentTypeTestCase(FormspreeTestCase):
     @httpretty.activate
     def test_various_content_types(self):
         httpretty.register_uri(httpretty.POST, 'https://api.sendgrid.com/api/mail.send.json')
-        r = self.client.post('/bob@example.com',
-            headers = {'Referer': 'http://example.com'},
+        r = self.client.post('/bob@testwebsite.com',
+            headers = {'Referer': 'http://testwebsite.com'},
             data={'name': 'bob'}
         )
         f = Form.query.first()
@@ -63,7 +63,7 @@ def ishtml(res):
         settings.MONTHLY_SUBMISSIONS_LIMIT = len(types)
 
         for ct, acc, check in types:
-            headers = {'Referer': 'http://example.com'}
+            headers = {'Referer': 'http://testwebsite.com'}
             if ct:
                 headers['Content-Type'] = ct
             if acc:
@@ -72,7 +72,7 @@ def ishtml(res):
             data = {'name': 'bob'}
             data = json.dumps(data) if ct and 'json' in ct else data
 
-            res = self.client.post('/bob@example.com',
+            res = self.client.post('/bob@testwebsite.com',
                 headers=headers,
                 data=data
             )
@@ -82,7 +82,7 @@ def ishtml(res):
             # and expect json in all of them
             headers['X-Requested-With'] = 'XMLHttpRequest'
 
-            res = self.client.post('/bob@example.com',
+            res = self.client.post('/bob@testwebsite.com',
                 headers=headers,
                 data=data
             )

tests/test_form_creation.py

@@ -57,7 +57,7 @@ def test_form_creation(self):
 
         # post to form
         r = self.client.post('/' + form_endpoint,
-            headers={'Referer': 'http://formspree.io'},
+            headers={'Referer': 'http://testsite.com'},
             data={'name': 'bruce'}
         )
         self.assertIn("sent an email confirmation", r.data)
@@ -86,7 +86,7 @@ def test_form_creation(self):
         self.assertEqual(settings.MONTHLY_SUBMISSIONS_LIMIT, 2)
         for i in range(5):
             r = self.client.post('/' + form_endpoint,
-                headers={'Referer': 'formspree.io'},
+                headers={'Referer': 'testsite.com'},
                 data={'name': 'ana',
                       'submission': '__%s__' % i}
             )
@@ -112,11 +112,11 @@ def test_form_creation_with_a_registered_email(self):
 
         # register user
         r = self.client.post('/register',
-            data={'email': 'user@formspree.io',
+            data={'email': 'user@testsite.com',
                   'password': 'banana'}
         )
         # upgrade user manually
-        user = User.query.filter_by(email='user@formspree.io').first()
+        user = User.query.filter_by(email='user@testsite.com').first()
         user.upgraded = True
         DB.session.add(user)
         DB.session.commit()
@@ -127,14 +127,14 @@ def test_form_creation_with_a_registered_email(self):
         # create form without providing an url should not send verification email
         r = self.client.post('/forms',
             headers={'Accept': 'application/json', 'Content-type': 'application/json'},
-            data=json.dumps({'email': 'email@formspree.io'})
+            data=json.dumps({'email': 'email@testsite.com'})
         )
         self.assertEqual(httpretty.has_request(), False)
 
         # create form without a confirmed email should send a verification email
         r = self.client.post('/forms',
             headers={'Accept': 'application/json', 'Content-type': 'application/json'},
-            data=json.dumps({'email': 'email@formspree.io',
+            data=json.dumps({'email': 'email@testsite.com',
                              'url': 'https://www.testsite.com/contact.html'})
         )
         resp = json.loads(r.data)
@@ -145,15 +145,15 @@ def test_form_creation_with_a_registered_email(self):
 
         # manually verify an email
         email = Email()
-        email.address = 'owned-by@formspree.io'
+        email.address = 'owned-by@testsite.com'
         email.owner_id = user.id
         DB.session.add(email)
         DB.session.commit()
 
         # create a form with the verified email address
         r = self.client.post('/forms',
             headers={'Accept': 'application/json', 'Content-type': 'application/json'},
-            data=json.dumps({'email': 'owned-by@formspree.io',
+            data=json.dumps({'email': 'owned-by@testsite.com',
                              'url': 'https://www.testsite.com/about.html'})
         )
         resp = json.loads(r.data)
@@ -176,11 +176,11 @@ def test_sitewide_forms(self):
 
         # register user
         r = self.client.post('/register',
-            data={'email': 'user@formspree.io',
+            data={'email': 'user@testsite.com',
                   'password': 'banana'}
         )
         # upgrade user manually
-        user = User.query.filter_by(email='user@formspree.io').first()
+        user = User.query.filter_by(email='user@testsite.com').first()
         user.upgraded = True
         DB.session.add(user)
         DB.session.commit()