Add ability to only check CSRF on specified request types

Refs #28

Author: vimalloc

docs/options.rst

@@ -33,6 +33,8 @@ The available options are:
                                   will cause this access cookie to be sent in with every request. Should be modified
                                   for only the paths that need the refresh cookie
 ``JWT_COOKIE_CSRF_PROTECT``       Enable/disable CSRF protection. Only used when sending the JWT in via cookies
+``JWT_CSRF_METHODS``              The request types that will use CSRF protection. Defaults to
+                                  ```['POST', 'PUT', 'PATCH', 'DELETE']```
 ``JWT_ACCESS_CSRF_COOKIE_NAME``   Name of the CSRF access cookie. Defaults to ``'csrf_access_token'``. Only used
                                   if using cookies with CSRF protection enabled
 ``JWT_REFRESH_CSRF_COOKIE_NAME``  Name of the CSRF refresh cookie. Defaults to ``'csrf_refresh_token'``. Only used

flask_jwt_extended/config.py

@@ -18,6 +18,7 @@
 
 # Options for using double submit for verifying CSRF tokens
 COOKIE_CSRF_PROTECT = True
+CSRF_METHODS = ['POST', 'PUT', 'PATCH', 'DELETE']
 ACCESS_CSRF_COOKIE_NAME = 'csrf_access_token'
 REFRESH_CSRF_COOKIE_NAME = 'csrf_refresh_token'
 CSRF_HEADER_NAME = 'X-CSRF-TOKEN'
@@ -79,6 +80,10 @@ def get_cookie_csrf_protect():
     return current_app.config.get('JWT_COOKIE_CSRF_PROTECT', COOKIE_CSRF_PROTECT)
 
 
+def get_csrf_request_methods():
+    return current_app.config.get('JWT_CSRF_METHODS', CSRF_METHODS)
+
+
 def get_access_csrf_cookie_name():
     return current_app.config.get('JWT_ACCESS_CSRF_COOKIE_NAME', ACCESS_CSRF_COOKIE_NAME)
 

flask_jwt_extended/utils.py

@@ -18,7 +18,7 @@
     get_cookie_csrf_protect, get_access_csrf_cookie_name, \
     get_refresh_cookie_name, get_refresh_cookie_path, \
     get_refresh_csrf_cookie_name, get_token_location, \
-    get_csrf_header_name, get_jwt_header_name
+    get_csrf_header_name, get_jwt_header_name, get_csrf_request_methods
 from flask_jwt_extended.exceptions import JWTEncodeError, JWTDecodeError, \
     InvalidHeaderError, NoAuthorizationError, WrongTokenError, \
     FreshTokenRequired, CSRFError
@@ -195,7 +195,7 @@ def _decode_jwt_from_cookies(type):
     algorithm = get_algorithm()
     token = _decode_jwt(token, secret, algorithm)
 
-    if get_cookie_csrf_protect():
+    if get_cookie_csrf_protect() and request.method in get_csrf_request_methods():
         csrf_header_key = get_csrf_header_name()
         csrf_token_from_header = request.headers.get(csrf_header_key, None)
         csrf_token_from_cookie = token.get('csrf', None)

tests/test_protected_endpoints.py

@@ -390,25 +390,11 @@ def refresh():
             set_access_cookies(resp, access_token)
             return resp, 200
 
-        @self.app.route('/api/protected')
+        @self.app.route('/api/protected', methods=['POST'])
         @jwt_required
         def protected():
             return jsonify({'msg': "hello world"})
 
-    def _jwt_post(self, url, jwt):
-        response = self.client.post(url, content_type='application/json',
-                                    headers={'Authorization': 'Bearer {}'.format(jwt)})
-        status_code = response.status_code
-        data = json.loads(response.get_data(as_text=True))
-        return status_code, data
-
-    def _jwt_get(self, url, jwt, header_name='Authorization', header_type='Bearer'):
-        header_type = '{} {}'.format(header_type, jwt).strip()
-        response = self.client.get(url, headers={header_name: header_type})
-        status_code = response.status_code
-        data = json.loads(response.get_data(as_text=True))
-        return status_code, data
-
     def _login(self):
         resp = self.client.post('/auth/login')
         index = 1
@@ -491,7 +477,7 @@ def test_endpoints_with_cookies(self):
         self.app.config['JWT_COOKIE_CSRF_PROTECT'] = False
 
         # Try access without logging in
-        response = self.client.get('/api/protected')
+        response = self.client.post('/api/protected')
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 401)
@@ -506,7 +492,7 @@ def test_endpoints_with_cookies(self):
 
         # Try with logging in
         self._login()
-        response = self.client.get('/api/protected')
+        response = self.client.post('/api/protected')
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 200)
@@ -525,7 +511,7 @@ def test_endpoints_with_cookies(self):
         access_cookie_key = access_cookie_str.split('=')[0]
         access_cookie_value = "".join(access_cookie_str.split('=')[1:])
         self.client.set_cookie('localhost', access_cookie_key, access_cookie_value)
-        response = self.client.get('/api/protected')
+        response = self.client.post('/api/protected')
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 200)
@@ -535,7 +521,7 @@ def test_access_endpoints_with_cookies_and_csrf(self):
         self.app.config['JWT_COOKIE_CSRF_PROTECT'] = True
 
         # Try without logging in
-        response = self.client.get('/api/protected')
+        response = self.client.post('/api/protected')
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 401)
@@ -545,30 +531,30 @@ def test_access_endpoints_with_cookies_and_csrf(self):
         access_csrf, refresh_csrf = self._login()
 
         # Try with logging in but without double submit csrf protection
-        response = self.client.get('/api/protected')
+        response = self.client.post('/api/protected')
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 401)
         self.assertIn('msg', data)
 
         # Try with logged in and bad header name for double submit token
-        response = self.client.get('/api/protected',
+        response = self.client.post('/api/protected',
                                    headers={'bad-header-name': 'banana'})
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 401)
         self.assertIn('msg', data)
 
         # Try with logged in and bad header data for double submit token
-        response = self.client.get('/api/protected',
+        response = self.client.post('/api/protected',
                                    headers={'X-CSRF-TOKEN': 'banana'})
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 401)
         self.assertIn('msg', data)
 
         # Try with logged in and good double submit token
-        response = self.client.get('/api/protected',
+        response = self.client.post('/api/protected',
                                    headers={'X-CSRF-TOKEN': access_csrf})
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
@@ -582,7 +568,7 @@ def test_access_endpoints_with_cookie_missing_csrf_field(self):
         self._login()
         self.app.config['JWT_COOKIE_CSRF_PROTECT'] = True
 
-        response = self.client.get('/api/protected')
+        response = self.client.post('/api/protected')
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 422)
@@ -606,12 +592,58 @@ def test_access_endpoints_with_cookie_csrf_claim_not_string(self):
         self.client.set_cookie('localhost', access_cookie_key, encoded_token)
 
         self.app.config['JWT_COOKIE_CSRF_PROTECT'] = True
-        response = self.client.get('/api/protected')
+        response = self.client.post('/api/protected')
         status_code = response.status_code
         data = json.loads(response.get_data(as_text=True))
         self.assertEqual(status_code, 422)
         self.assertIn('msg', data)
 
+    def test_custom_csrf_methods(self):
+        @self.app.route('/protected-post', methods=['POST'])
+        @jwt_required
+        def protected_post():
+            return jsonify({'msg': "hello world"})
+
+        @self.app.route('/protected-get', methods=['GET'])
+        @jwt_required
+        def protected_get():
+            return jsonify({'msg': "hello world"})
+
+        # Login (saves jwts in the cookies for the test client
+        self.app.config['JWT_COOKIE_CSRF_PROTECT'] = True
+        self._login()
+
+        # Test being able to access GET without CSRF protection, and POST with
+        # CSRF protection
+        self.app.config['JWT_CSRF_METHODS'] = ['POST']
+
+        response = self.client.post('/protected-post')
+        status_code = response.status_code
+        data = json.loads(response.get_data(as_text=True))
+        self.assertEqual(status_code, 401)
+        self.assertIn('msg', data)
+
+        response = self.client.get('/protected-get')
+        status_code = response.status_code
+        data = json.loads(response.get_data(as_text=True))
+        self.assertEqual(status_code, 200)
+        self.assertEqual(data, {'msg': 'hello world'})
+
+        # Now swap it around, and verify the JWT_CRSF_METHODS are being honored
+        self.app.config['JWT_CSRF_METHODS'] = ['GET']
+
+        response = self.client.get('/protected-get')
+        status_code = response.status_code
+        data = json.loads(response.get_data(as_text=True))
+        self.assertEqual(status_code, 401)
+        self.assertIn('msg', data)
+
+        response = self.client.post('/protected-post')
+        status_code = response.status_code
+        data = json.loads(response.get_data(as_text=True))
+        self.assertEqual(status_code, 200)
+        self.assertEqual(data, {'msg': 'hello world'})
+
 
 class TestEndpointsWithHeadersAndCookies(unittest.TestCase):