From: Ben Darnell <ben@bendarnell.com>
Date: Mon, 26 May 2014 17:33:21 +0000 (-0400)
Subject: Small security improvements to xsrf tokens; add tests.
X-Git-Tag: v3.2.2~4
X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=7279a303d1c366aabd4facfc6b29ed46c3422350;p=thirdparty%2Ftornado.git

Small security improvements to xsrf tokens; add tests.

Use os.urandom(16) instead of uuid.uuid4(), to reclaim a few bits of
entropy.  Use _time_independent_equals for comparison.
---

diff --git a/tornado/test/web_test.py b/tornado/test/web_test.py
index 44b323da5..7e4fa3133 100644
--- a/tornado/test/web_test.py
+++ b/tornado/test/web_test.py
@@ -1904,3 +1904,116 @@ class SignedValueTest(unittest.TestCase):
         decoded = decode_signed_value(SignedValueTest.SECRET, "key", signed,
                                       clock=self.present)
         self.assertEqual(value, decoded)
+
+
+@wsgi_safe
+class XSRFTest(SimpleHandlerTestCase):
+    class Handler(RequestHandler):
+        def get(self):
+            self.write(self.xsrf_token)
+
+        def post(self):
+            self.write("ok")
+
+    def get_app_kwargs(self):
+        return dict(xsrf_cookies=True)
+
+    def setUp(self):
+        super(XSRFTest, self).setUp()
+        self.xsrf_token = self.get_token()
+
+    def get_token(self, old_token=None):
+        if old_token is not None:
+            headers = self.cookie_headers(old_token)
+        else:
+            headers = None
+        response = self.fetch("/", headers=headers)
+        response.rethrow()
+        return native_str(response.body)
+
+    def cookie_headers(self, token=None):
+        if token is None:
+            token = self.xsrf_token
+        return {"Cookie": "_xsrf=" + token}
+
+    def test_xsrf_fail_no_token(self):
+        with ExpectLog(gen_log, ".*'_xsrf' argument missing"):
+            response = self.fetch("/", method="POST", body=b"")
+        self.assertEqual(response.code, 403)
+
+    def test_xsrf_fail_body_no_cookie(self):
+        with ExpectLog(gen_log, ".*XSRF cookie does not match POST"):
+            response = self.fetch(
+                "/", method="POST",
+                body=urllib_parse.urlencode(dict(_xsrf=self.xsrf_token)))
+        self.assertEqual(response.code, 403)
+
+    def test_xsrf_fail_cookie_no_body(self):
+        with ExpectLog(gen_log, ".*'_xsrf' argument missing"):
+            response = self.fetch(
+                "/", method="POST", body=b"",
+                headers=self.cookie_headers())
+        self.assertEqual(response.code, 403)
+
+    def test_xsrf_success_post_body(self):
+        response = self.fetch(
+            "/", method="POST",
+            body=urllib_parse.urlencode(dict(_xsrf=self.xsrf_token)),
+            headers=self.cookie_headers())
+        self.assertEqual(response.code, 200)
+
+    def test_xsrf_success_query_string(self):
+        response = self.fetch(
+            "/?" + urllib_parse.urlencode(dict(_xsrf=self.xsrf_token)),
+            method="POST", body=b"",
+            headers=self.cookie_headers())
+        self.assertEqual(response.code, 200)
+
+    def test_xsrf_success_header(self):
+        response = self.fetch("/", method="POST", body=b"",
+                              headers=dict({"X-Xsrftoken": self.xsrf_token},
+                                           **self.cookie_headers()))
+        self.assertEqual(response.code, 200)
+
+    def test_distinct_tokens(self):
+        # Every request gets a distinct token.
+        NUM_TOKENS = 10
+        tokens = set()
+        for i in range(NUM_TOKENS):
+            tokens.add(self.get_token())
+        self.assertEqual(len(tokens), NUM_TOKENS)
+
+    def test_cross_user(self):
+        token2 = self.get_token()
+        # Each token can be used to authenticate its own request.
+        for token in (self.xsrf_token, token2):
+            response  = self.fetch(
+                "/", method="POST",
+                body=urllib_parse.urlencode(dict(_xsrf=token)),
+                headers=self.cookie_headers(token))
+            self.assertEqual(response.code, 200)
+        # Sending one in the cookie and the other in the body is not allowed.
+        for cookie_token, body_token in ((self.xsrf_token, token2),
+                                         (token2, self.xsrf_token)):
+            with ExpectLog(gen_log, '.*XSRF cookie does not match POST'):
+                response = self.fetch(
+                    "/", method="POST",
+                    body=urllib_parse.urlencode(dict(_xsrf=body_token)),
+                    headers=self.cookie_headers(cookie_token))
+            self.assertEqual(response.code, 403)
+
+    def test_refresh_token(self):
+        token = self.xsrf_token
+        # A user's token is stable over time.  Refreshing the page in one tab
+        # might update the cookie while an older tab still has the old cookie
+        # in its DOM.  Simulate this scenario by passing a constant token
+        # in the body and re-querying for the token.
+        for i in range(5):
+            token = self.get_token(token)
+            # Implementation detail: the same token is returned each time
+            self.assertEqual(token, self.xsrf_token)
+            response = self.fetch(
+                "/", method="POST",
+                body=urllib_parse.urlencode(dict(_xsrf=self.xsrf_token)),
+                headers=self.cookie_headers(token))
+            self.assertEqual(response.code, 200)
diff --git a/tornado/web.py b/tornado/web.py
index d0fb16c14..1b618a70b 100644
--- a/tornado/web.py
+++ b/tornado/web.py
@@ -1073,7 +1073,7 @@ class RequestHandler(object):
         if not hasattr(self, "_xsrf_token"):
             token = self.get_cookie("_xsrf")
             if not token:
-                token = binascii.b2a_hex(uuid.uuid4().bytes)
+                token = binascii.b2a_hex(os.urandom(16))
                 expires_days = 30 if self.current_user else None
                 self.set_cookie("_xsrf", token, expires_days=expires_days)
             self._xsrf_token = token
@@ -1105,7 +1105,7 @@ class RequestHandler(object):
                  self.request.headers.get("X-Csrftoken"))
         if not token:
             raise HTTPError(403, "'_xsrf' argument missing from POST")
-        if self.xsrf_token != token:
+        if not _time_independent_equals(utf8(self.xsrf_token), utf8(token)):
             raise HTTPError(403, "XSRF cookie does not match POST argument")
 
     def xsrf_form_html(self):