Clean up CSRF implementation

We borrowed our CSRF implementation from Django (#88). This commit
refactors away some of that legacy. In particular:

  - We now call _get_new_token from only one place, instead of three.
  - We now raise 403 from only one place, instead of two.
  - We now work with unicode instead of casting the token to str early.

As a byproduct:

  - The set_cookie method now casts unicode key/values to bytestrings.
  - The csrf and utils modules are now __future__-proofed.

gratipay/main.py

@@ -117,7 +117,8 @@ def _erase_cookie(response, *args, **kw):
     i18n.set_up_i18n,
     authentication.start_user_as_anon,
     authentication.authenticate_user_if_possible,
-    csrf.get_csrf_token_from_request,
+    csrf.extract_token_from_cookie,
+    csrf.reject_forgeries,
 
     algorithm['dispatch_request_to_filesystem'],
 
@@ -133,7 +134,7 @@ def _erase_cookie(response, *args, **kw):
 
     gratipay.set_misc_headers,
     authentication.add_auth_to_response,
-    csrf.add_csrf_token_to_response,
+    csrf.add_token_to_response,
     cache_static.add_caching_to_response if website.cache_static else noop,
     x_frame_options,
 

gratipay/security/authentication.py

@@ -53,7 +53,7 @@ def _get_user_via_basic_auth(auth_header):
 def _turn_off_csrf(request):
     """Given a request, short-circuit CSRF.
     """
-    csrf_token = csrf._get_new_csrf_key()
+    csrf_token = csrf._get_new_token()
     request.headers.cookie['csrf_token'] = csrf_token
     request.headers['X-CSRF-TOKEN'] = csrf_token
 

gratipay/security/csrf.py

@@ -7,6 +7,7 @@
     https://github.com/gratipay/gratipay.com/issues/88
 
 """
+from __future__ import absolute_import, division, print_function, unicode_literals
 
 from datetime import timedelta
 import re
@@ -35,51 +36,37 @@ def patch_vary_headers(response, newheaders):
 
 
 from aspen import Response
-from crypto import constant_time_compare, get_random_string
-
-REASON_NO_CSRF_COOKIE = "CSRF cookie not set."
-REASON_BAD_TOKEN = "CSRF token missing or incorrect."
+from .crypto import constant_time_compare, get_random_string
 
 TOKEN_LENGTH = 32
 CSRF_TIMEOUT = timedelta(days=7)
 
-
-def _get_new_csrf_key():
-    return get_random_string(TOKEN_LENGTH)
-
-
-def _sanitize_token(token):
-    # Allow only alphanum, and ensure we return a 'str' for the sake
-    # of the post processing middleware.
-    if len(token) > TOKEN_LENGTH:
-        return _get_new_csrf_key()
-    token = re.sub('[^a-zA-Z0-9]+', '', str(token.decode('ascii', 'ignore')))
-    if token == "":
-        # In case the cookie has been truncated to nothing at some point.
-        return _get_new_csrf_key()
-    return token
+_get_new_token = lambda: get_random_string(TOKEN_LENGTH)
 
 
-
-def get_csrf_token_from_request(request, state):
-    """Given a Request object, reject it if it's a forgery.
+def extract_token_from_cookie(request):
+    """Given a Request object, return a csrf_token.
     """
     if request.line.uri.startswith('/assets/'): return
     if request.line.uri.startswith('/callbacks/'): return
 
     try:
-        cookie_token = _sanitize_token(request.headers.cookie['csrf_token'].value)
+        token = request.headers.cookie['csrf_token'].value
     except KeyError:
-        cookie_token = None
+        token = ""
+    else:
+        if len(token) > TOKEN_LENGTH:
+            token = ""
+        else:
+            token = re.sub('[^a-zA-Z0-9]+', '', token.decode('ascii', 'ignore'))
 
-    state['csrf_token'] = cookie_token or _get_new_csrf_key()
+    return {'csrf_token': token or _get_new_token()}
 
+
+def reject_forgeries(request, csrf_token):
     # Assume that anything not defined as 'safe' by RC2616 needs protection
     if request.line.method not in ('GET', 'HEAD', 'OPTIONS', 'TRACE'):
 
-        if cookie_token is None:
-            raise Response(403, REASON_NO_CSRF_COOKIE)
-
         # Check non-cookie token for match.
         second_token = ""
         if request.line.method == "POST":
@@ -91,11 +78,11 @@ def get_csrf_token_from_request(request, state):
             # and possible for PUT/DELETE.
             second_token = request.headers.get('X-CSRF-TOKEN', '')
 
-        if not constant_time_compare(second_token, cookie_token):
-            raise Response(403, REASON_BAD_TOKEN)
+        if not constant_time_compare(second_token, csrf_token):
+            raise Response(403, "Bad CSRF cookie")
 
 
-def add_csrf_token_to_response(response, csrf_token=None):
+def add_token_to_response(response, csrf_token=None):
     """Store the latest CSRF token as a cookie.
     """
     if csrf_token:

gratipay/utils/__init__.py

@@ -1,6 +1,6 @@
 # encoding: utf8
 
-from __future__ import division
+from __future__ import absolute_import, division, print_function, unicode_literals
 
 from datetime import datetime, timedelta
 
@@ -148,21 +148,23 @@ def is_card_expiring(expiration_year, expiration_month):
     return delta < EXPIRING_DELTA
 
 
-def set_cookie(cookies, key, value, expires=None, httponly=True, path='/'):
+def set_cookie(cookies, key, value, expires=None, httponly=True, path=b'/'):
+    key = key.encode('ascii')
+    value = value.encode('ascii')
     cookies[key] = value
     cookie = cookies[key]
     if expires:
         if isinstance(expires, timedelta):
             expires += utcnow()
         if isinstance(expires, datetime):
             expires = to_rfc822(expires).encode('ascii')
-        cookie['expires'] = expires
+        cookie[b'expires'] = expires
     if httponly:
-        cookie['httponly'] = True
+        cookie[b'httponly'] = True
     if path:
-        cookie['path'] = path
+        cookie[b'path'] = path
     if gratipay.canonical_scheme == 'https':
-        cookie['secure'] = True
+        cookie[b'secure'] = True
 
 
 def erase_cookie(cookies, key, **kw):

tests/py/test_hooks.py

@@ -7,7 +7,6 @@
 from environment import Environment
 
 from gratipay import wireup
-from gratipay.security.csrf import REASON_BAD_TOKEN, REASON_NO_CSRF_COOKIE
 from gratipay.security.user import SESSION
 from gratipay.testing import Harness
 
@@ -112,11 +111,11 @@ def test_error_spt_works(self):
     def test_no_csrf_cookie(self):
         r = self.client.POST('/', csrf_token=False, raise_immediately=False)
         assert r.code == 403
-        assert REASON_NO_CSRF_COOKIE in r.body
+        assert "Bad CSRF cookie" in r.body
         assert b'csrf_token' in r.headers.cookie
 
     def test_bad_csrf_cookie(self):
         r = self.client.POST('/', csrf_token=b'bad_token', raise_immediately=False)
         assert r.code == 403
-        assert REASON_BAD_TOKEN in r.body
+        assert "Bad CSRF cookie" in r.body
         assert r.headers.cookie[b'csrf_token'].value != 'bad_token'