test: invalidate used password reset tokens

Also pep8 compliance and suggested changes.

flask_security/recoverable.py

@@ -11,6 +11,7 @@
 
 from flask import current_app as app
 from werkzeug.local import LocalProxy
+from werkzeug.security import safe_str_cmp
 
 from .signals import password_reset, reset_password_instructions_sent
 from .utils import send_mail, md5, encrypt_password, url_for_security, \
@@ -66,11 +67,13 @@ def reset_password_token_status(token):
 
     :param token: The password reset token
     """
-    expired, invalid, user, data = get_token_status(token, 'reset', 'RESET_PASSWORD', return_data=True)
+    expired, invalid, user, data = get_token_status(token, 'reset', 'RESET_PASSWORD',
+                                                    return_data=True)
     if not invalid:
-        password_hash = md5(user.password) if user.password else None
-        if password_hash != data[1]:
-            invalid = True
+        if user.password:
+            password_hash = md5(user.password)
+            if not safe_str_cmp(password_hash, data[1]):
+                invalid = True
 
     return expired, invalid, user
 

tests/test_recoverable.py

@@ -122,6 +122,32 @@ def test_expired_reset_token(client, get_message):
     assert msg in response.data
 
 
+def test_used_reset_token(client, get_message):
+    with capture_reset_password_requests() as requests:
+        client.post('/reset', data=dict(email='joe@lp.com'), follow_redirects=True)
+
+    token = requests[0]['token']
+
+    # use the token
+    response = client.post('/reset/' + token, data={
+        'password': 'newpassword',
+        'password_confirm': 'newpassword'
+    }, follow_redirects=True)
+
+    assert get_message('PASSWORD_RESET') in response.data
+
+    logout(client)
+
+    # attempt to use it a second time
+    response2 = client.post('/reset/' + token, data={
+        'password': 'otherpassword',
+        'password_confirm': 'otherpassword'
+    }, follow_redirects=True)
+
+    msg = get_message('INVALID_RESET_PASSWORD_TOKEN')
+    assert msg in response2.data
+
+
 @pytest.mark.settings(reset_url='/custom_reset')
 def test_custom_reset_url(client):
     response = client.get('/custom_reset')