Repository: CTFd/CTFd
Pull request #2679 (2024-12-15 06:01:01)

Convert confirmation & reset password emails to single use
* Convert confirmation & reset password emails to single use by storing the email tokens in caching

br0xpl: @ColdHeat, one more thing: to be honest I don't know if we need to sign the tokens at all (hmac), two things:
1. If the token is random enough and is stored in redis and assuming noone has access to redis then random number without hmac is enough.
2. Still if we want to sign the token then it should be also verified during token verification.
ColdHeat: @br0xpl The reason the HMAC is there is to allow revocation of the tokens by rotating the secret key. 

Signing would really only be useful if we wanted to give the user a message that their token expired but I don't think that's needed anymore. With the HMAC there's no way to verify it without having the original random bytes. 
br0xpl: @ColdHeat  sorry I was not clear enough - I did not mean signing, what I meant is that we don't need hmac at all. 

Let me try again, we can have two options:
1. Use random number without hmac and store it in redis. You don't need to hmac it becuase you can revoke tokens by simply removing them from redis. 
2. The other option is to use hmac as you are now but right now you are not validating the hmac -- look at functions verify_reset_password_token and verify_email_confirm_token -- you are just checking if they exist in redis. This is equal to simply having a random number in redis without hmac at the moment. If you want to keep hmac, then you should also at the end of verify_...._token functions perform a hmac check on the email taken from redis and check if it is equal to the provided nonce.




ColdHeat: @br0xpl 

1. Revocation by redis key is more tedious in comparison to rotating the secret key and letting the redis keys expire. 

2. The hmac is built off of a random value, not the email. So there's no way to verify the hmac without also storing the random value (which seems unnecessary).
br0xpl: 2. Haha of course I forgot, you would have to store the random number as well ;)


1. So how do you want to revoke the token if there is no verification of hmac? You rotate the key but the token is still in Redis and still working.. this is exactly my original comment :-)

I don't see any point of using hmac here at all. Just random secure with enough entropy should be good to go. 
ColdHeat: @br0xpl Ah yes I understand now, thank you for clarifying. I will update the PR to use just the random bytes. 

Patch:
diff --git a/CTFd/auth.py b/CTFd/auth.py
index a645ad60..fbb27993 100644
--- a/CTFd/auth.py
+++ b/CTFd/auth.py
@@ -1,12 +1,13 @@
-import base64  # noqa: I001
-
 import requests
 from flask import Blueprint, abort
 from flask import current_app as app
 from flask import redirect, render_template, request, session, url_for
-from itsdangerous.exc import BadSignature, BadTimeSignature, SignatureExpired
 
 from CTFd.cache import clear_team_session, clear_user_session
+from CTFd.exceptions.email import (
+    UserConfirmTokenInvalidException,
+    UserResetPasswordTokenInvalidException,
+)
 from CTFd.models import Brackets, Teams, UserFieldEntries, UserFields, Users, db
 from CTFd.utils import config, email, get_app_config, get_config
 from CTFd.utils import user as current_user
@@ -21,7 +22,12 @@ from CTFd.utils.helpers import error_for, get_errors, markup
 from CTFd.utils.logging import log
 from CTFd.utils.modes import TEAMS_MODE
 from CTFd.utils.security.auth import login_user, logout_user
-from CTFd.utils.security.signing import unserialize
+from CTFd.utils.security.email import (
+    remove_email_confirm_token,
+    remove_reset_password_token,
+    verify_email_confirm_token,
+    verify_reset_password_token,
+)
 from CTFd.utils.validators import ValidationError
 
 auth = Blueprint("auth", __name__)
@@ -38,14 +44,11 @@ def confirm(data=None):
     # User is confirming email account
     if data and request.method == "GET":
         try:
-            user_email = unserialize(data, max_age=1800)
-        except (BadTimeSignature, SignatureExpired):
-            return render_template(
-                "confirm.html", errors=["Your confirmation link has expired"]
-            )
-        except (BadSignature, TypeError, base64.binascii.Error):
+            user_email = verify_email_confirm_token(data)
+        except (UserConfirmTokenInvalidException):
             return render_template(
-                "confirm.html", errors=["Your confirmation token is invalid"]
+                "confirm.html",
+                errors=["Your confirmation link is invalid, please generate a new one"],
             )
 
         user = Users.query.filter_by(email=user_email).first_or_404()
@@ -59,6 +62,7 @@ def confirm(data=None):
             name=user.name,
         )
         db.session.commit()
+        remove_email_confirm_token(data)
         clear_user_session(user_id=user.id)
         email.successful_registration_notification(user.email)
         db.session.close()
@@ -107,14 +111,11 @@ def reset_password(data=None):
 
     if data is not None:
         try:
-            email_address = unserialize(data, max_age=1800)
-        except (BadTimeSignature, SignatureExpired):
+            email_address = verify_reset_password_token(data)
+        except (UserResetPasswordTokenInvalidException):
             return render_template(
-                "reset_password.html", errors=["Your link has expired"]
-            )
-        except (BadSignature, TypeError, base64.binascii.Error):
-            return render_template(
-                "reset_password.html", errors=["Your reset token is invalid"]
+                "reset_password.html",
+                errors=["Your reset link is invalid, please generate a new one"],
             )
 
         if request.method == "GET":
@@ -138,6 +139,7 @@ def reset_password(data=None):
 
             user.password = password
             db.session.commit()
+            remove_reset_password_token(data)
             clear_user_session(user_id=user.id)
             log(
                 "logins",
diff --git a/CTFd/exceptions/email.py b/CTFd/exceptions/email.py
new file mode 100644
index 00000000..8e94d06f
--- /dev/null
+++ b/CTFd/exceptions/email.py
@@ -0,0 +1,6 @@
+class UserConfirmTokenInvalidException(Exception):
+    pass
+
+
+class UserResetPasswordTokenInvalidException(Exception):
+    pass
diff --git a/CTFd/utils/email/__init__.py b/CTFd/utils/email/__init__.py
index 08d7abf7..b4fac23d 100644
--- a/CTFd/utils/email/__init__.py
+++ b/CTFd/utils/email/__init__.py
@@ -5,7 +5,10 @@ from CTFd.utils.config import get_mail_provider
 from CTFd.utils.email.providers.mailgun import MailgunEmailProvider
 from CTFd.utils.email.providers.smtp import SMTPEmailProvider
 from CTFd.utils.formatters import safe_format
-from CTFd.utils.security.signing import serialize
+from CTFd.utils.security.email import (
+    generate_email_confirm_token,
+    generate_password_reset_token,
+)
 
 PROVIDERS = {"smtp": SMTPEmailProvider, "mailgun": MailgunEmailProvider}
 
@@ -72,7 +75,11 @@ def forgot_password(email):
         get_config("password_reset_body") or DEFAULT_PASSWORD_RESET_BODY,
         ctf_name=get_config("ctf_name"),
         ctf_description=get_config("ctf_description"),
-        url=url_for("auth.reset_password", data=serialize(email), _external=True),
+        url=url_for(
+            "auth.reset_password",
+            data=generate_password_reset_token(email),
+            _external=True,
+        ),
     )
 
     subject = safe_format(
@@ -88,7 +95,10 @@ def verify_email_address(addr):
         ctf_name=get_config("ctf_name"),
         ctf_description=get_config("ctf_description"),
         url=url_for(
-            "auth.confirm", data=serialize(addr), _external=True, _method="GET"
+            "auth.confirm",
+            data=generate_email_confirm_token(addr),
+            _external=True,
+            _method="GET",
         ),
     )
 
diff --git a/CTFd/utils/security/email.py b/CTFd/utils/security/email.py
new file mode 100644
index 00000000..2241f71e
--- /dev/null
+++ b/CTFd/utils/security/email.py
@@ -0,0 +1,42 @@
+import os
+
+from CTFd.cache import cache
+from CTFd.exceptions.email import (
+    UserConfirmTokenInvalidException,
+    UserResetPasswordTokenInvalidException,
+)
+from CTFd.utils.encoding import hexencode
+
+
+def generate_email_confirm_token(addr, timeout=1800):
+    nonce = hexencode(os.urandom(32))
+    cache.set(f"confirm_email_{nonce}", addr, timeout=timeout)
+    return nonce
+
+
+def verify_email_confirm_token(nonce):
+    addr = cache.get(f"confirm_email_{nonce}")
+    if addr is None:
+        raise UserConfirmTokenInvalidException
+    return addr
+
+
+def remove_email_confirm_token(nonce):
+    cache.delete(f"confirm_email_{nonce}")
+
+
+def generate_password_reset_token(addr, timeout=1800):
+    nonce = hexencode(os.urandom(32))
+    cache.set(f"reset_password_{nonce}", addr, timeout=timeout)
+    return nonce
+
+
+def verify_reset_password_token(nonce):
+    addr = cache.get(f"reset_password_{nonce}")
+    if addr is None:
+        raise UserResetPasswordTokenInvalidException
+    return addr
+
+
+def remove_reset_password_token(nonce):
+    cache.delete(f"reset_password_{nonce}")


Test patch:
diff --git a/tests/users/test_auth.py b/tests/users/test_auth.py
index 890666ac..42c52ee1 100644
--- a/tests/users/test_auth.py
+++ b/tests/users/test_auth.py
@@ -8,7 +8,6 @@ from freezegun import freeze_time
 from CTFd.models import Users, db
 from CTFd.utils import get_config, set_config
 from CTFd.utils.crypto import verify_password
-from CTFd.utils.security.signing import serialize
 from tests.helpers import create_ctfd, destroy_ctfd, login_as_user, register_user
 
 
@@ -190,17 +189,22 @@ def test_user_isnt_admin():
 def test_expired_confirmation_links():
     """Test that expired confirmation links are reported to the user"""
     app = create_ctfd()
-    with app.app_context(), freeze_time("2019-02-24 03:21:34"):
+    with app.app_context():
         set_config("verify_emails", True)
 
         register_user(app, email="user@user.com")
         client = login_as_user(app, name="user", password="password")
 
         # user@user.com "2012-01-14 03:21:34"
-        confirm_link = "http://localhost/confirm/InVzZXJAdXNlci5jb20i.TxD0vg.cAGwAy8cK1T0saEEbrDEBVF2plI"
+        confirm_link = (
+            "http://localhost/confirm/bb8a8526146e50778b211ae63074595880edbc0b"
+        )
         r = client.get(confirm_link)
 
-        assert "Your confirmation link has expired" in r.get_data(as_text=True)
+        assert (
+            "Your confirmation link is invalid, please generate a new one"
+            in r.get_data(as_text=True)
+        )
         user = Users.query.filter_by(email="user@user.com").first()
         assert user.verified is not True
     destroy_ctfd(app)
@@ -219,7 +223,10 @@ def test_invalid_confirmation_links():
         confirm_link = "http://localhost/confirm/a8375iyu<script>alert(1)<script>hn3048wueorighkgnsfg"
         r = client.get(confirm_link)
 
-        assert "Your confirmation token is invalid" in r.get_data(as_text=True)
+        assert (
+            "Your confirmation link is invalid, please generate a new one"
+            in r.get_data(as_text=True)
+        )
         user = Users.query.filter_by(email="user@user.com").first()
         assert user.verified is not True
     destroy_ctfd(app)
@@ -237,12 +244,14 @@ def test_expired_reset_password_link():
 
         register_user(app, name="user1", email="user@user.com")
 
-        with app.test_client() as client, freeze_time("2019-02-24 03:21:34"):
-            # user@user.com "2012-01-14 03:21:34"
-            forgot_link = "http://localhost/reset_password/InVzZXJAdXNlci5jb20i.TxD0vg.cAGwAy8cK1T0saEEbrDEBVF2plI"
+        with app.test_client() as client:
+            forgot_link = "http://localhost/reset_password/bb8a8526146e50778b211ae63074595880edbc0b"
             r = client.get(forgot_link)
 
-            assert "Your link has expired" in r.get_data(as_text=True)
+            assert (
+                "Your reset link is invalid, please generate a new one"
+                in r.get_data(as_text=True)
+            )
     destroy_ctfd(app)
 
 
@@ -263,7 +272,10 @@ def test_invalid_reset_password_link():
             forgot_link = "http://localhost/reset_password/5678ytfghjiu876tyfg<INVALID DATA>hvbnmkoi9u87y6trdf"
             r = client.get(forgot_link)
 
-            assert "Your reset token is invalid" in r.get_data(as_text=True)
+            assert (
+                "Your reset link is invalid, please generate a new one"
+                in r.get_data(as_text=True)
+            )
     destroy_ctfd(app)
 
 
@@ -309,8 +321,10 @@ def test_user_can_confirm_email(mock_smtp):
         mock_smtp.return_value.send_message.assert_called()
 
         with client.session_transaction() as sess:
-            data = {"nonce": sess.get("nonce")}
-            r = client.post("http://localhost/confirm", data=data)
+            urandom_value = b"\xff" * 32
+            with patch("os.urandom", return_value=urandom_value):
+                data = {"nonce": sess.get("nonce")}
+                r = client.post("http://localhost/confirm", data=data)
             assert "Confirmation email sent to" in r.get_data(as_text=True)
 
             r = client.get("/challenges")
@@ -318,7 +332,9 @@ def test_user_can_confirm_email(mock_smtp):
                 r.location == "http://localhost/confirm"
             )  # We got redirected to /confirm
 
-            r = client.get("http://localhost/confirm/" + serialize("user@user.com"))
+            r = client.get(
+                "http://localhost/confirm/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
+            )
             assert r.location == "http://localhost/challenges"
 
             # The team is now verified
@@ -336,7 +352,7 @@ def test_user_can_reset_password(mock_smtp):
     from email.message import EmailMessage
 
     app = create_ctfd()
-    with app.app_context(), freeze_time("2012-01-14 03:21:34"):
+    with app.app_context():
         # Set CTFd to send emails
         set_config("mail_server", "localhost")
         set_config("mail_port", 25)
@@ -355,7 +371,9 @@ def test_user_can_reset_password(mock_smtp):
                 data = {"nonce": sess.get("nonce"), "email": "user@user.com"}
 
             # Issue the password reset request
-            client.post("/reset_password", data=data)
+            urandom_value = b"\xff" * 32
+            with patch("os.urandom", return_value=urandom_value):
+                client.post("/reset_password", data=data)
 
             ctf_name = get_config("ctf_name")
             from_addr = get_config("mailfrom_addr") or app.config.get("MAILFROM_ADDR")
@@ -367,7 +385,7 @@ def test_user_can_reset_password(mock_smtp):
             msg = (
                 "Did you initiate a password reset on CTFd? If you didn't initiate this request you can ignore this email. "
                 "\n\nClick the following link to reset your password:\n"
-                "http://localhost/reset_password/InVzZXJAdXNlci5jb20i.TxD0vg.28dY_Gzqb1TH9nrcE_H7W8YFM-U\n\n"
+                "http://localhost/reset_password/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff\n\n"
                 "If the link is not clickable, try copying and pasting it into your browser."
             )
             ctf_name = get_config("ctf_name")
@@ -396,10 +414,10 @@ def test_user_can_reset_password(mock_smtp):
 
             # Do the password reset
             client.get(
-                "/reset_password/InVzZXJAdXNlci5jb20i.TxD0vg.28dY_Gzqb1TH9nrcE_H7W8YFM-U"
+                "/reset_password/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
             )
             client.post(
-                "/reset_password/InVzZXJAdXNlci5jb20i.TxD0vg.28dY_Gzqb1TH9nrcE_H7W8YFM-U",
+                "/reset_password/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
                 data=data,
             )
 
diff --git a/tests/utils/test_email.py b/tests/utils/test_email.py
index 4036da86..7944502d 100644
--- a/tests/utils/test_email.py
+++ b/tests/utils/test_email.py
@@ -2,16 +2,18 @@ from email.message import EmailMessage
 from unittest.mock import Mock, patch
 
 import requests
-from freezegun import freeze_time
 
+from CTFd.models import Users
 from CTFd.utils import get_config, set_config
+from CTFd.utils.crypto import verify_password
 from CTFd.utils.email import (
     check_email_is_whitelisted,
+    forgot_password,
     sendmail,
     successful_registration_notification,
     verify_email_address,
 )
-from tests.helpers import create_ctfd, destroy_ctfd
+from tests.helpers import create_ctfd, destroy_ctfd, login_as_user, register_user
 
 
 @patch("smtplib.SMTP")
@@ -178,14 +180,14 @@ def test_verify_email(mock_smtp):
 
         to_addr = "user@user.com"
 
-        with freeze_time("2012-01-14 03:21:34"):
+        urandom_value = b"\xff" * 32
+        with patch("os.urandom", return_value=urandom_value):
             verify_email_address(to_addr)
 
-        # This is currently not actually validated
         msg = (
             "Welcome to CTFd!\n\n"
             "Click the following link to confirm and activate your account:\n"
-            "http://localhost/confirm/InVzZXJAdXNlci5jb20i.TxD0vg.28dY_Gzqb1TH9nrcE_H7W8YFM-U\n\n"
+            "http://localhost/confirm/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff\n\n"
             "If the link is not clickable, try copying and pasting it into your browser."
         )
 
@@ -319,3 +321,156 @@ def test_email_whitelist():
         for case in test_cases_multiple_combined_domains:
             email, expected = case
             assert check_email_is_whitelisted(email) is expected
+    destroy_ctfd(app)
+
+
+def test_confirm_links_single_use():
+    """Test that confirm links are single use"""
+    app = create_ctfd()
+    with app.app_context():
+        set_config("verify_emails", True)
+        register_user(app)
+        client = login_as_user(app)
+
+        to_addr = "user@examplectf.com"
+
+        urandom_value = b"\xff" * 32
+        with patch("os.urandom", return_value=urandom_value):
+            verify_email_address(to_addr)
+
+        r = client.get(
+            "http://localhost/confirm/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
+        )
+        user = Users.query.filter_by(email=to_addr).first()
+        assert user.verified is True
+
+        r = client.get(
+            "http://localhost/confirm/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
+        )
+        assert (
+            "Your confirmation link is invalid, please generate a new one"
+            in r.get_data(as_text=True)
+        )
+    destroy_ctfd(app)
+
+
+@patch("smtplib.SMTP")
+def test_confirm_link_tokens_unique(mock_smtp):
+    app = create_ctfd()
+    with app.app_context():
+        set_config("mail_server", "localhost")
+        set_config("mail_port", 25)
+        set_config("mail_useauth", True)
+        set_config("mail_username", "username")
+        set_config("mail_password", "password")
+        set_config("verify_emails", True)
+        register_user(app, name="user1", email="user1@examplectf.com")
+        register_user(app, name="user2", email="user2@examplectf.com")
+
+        verify_email_address("user1@examplectf.com")
+        call1 = str(mock_smtp.return_value.send_message.call_args[0][0])
+
+        verify_email_address("user1@examplectf.com")
+        call2 = str(mock_smtp.return_value.send_message.call_args[0][0])
+
+        verify_email_address("user2@examplectf.com")
+        call3 = str(mock_smtp.return_value.send_message.call_args[0][0])
+        assert call1 != call2
+        assert call2 != call3
+        assert call1 != call3
+    destroy_ctfd(app)
+
+
+def test_reset_password_links_single_use():
+    """Test that reset password links are single use"""
+    app = create_ctfd()
+    with app.app_context():
+        register_user(app)
+
+        set_config("mail_server", "localhost")
+        set_config("mail_port", 25)
+        set_config("mail_useauth", True)
+        set_config("mail_username", "username")
+        set_config("mail_password", "password")
+
+        with app.test_client() as client:
+            client.get("/reset_password")
+
+            # Build reset password data
+            with client.session_transaction() as sess:
+                data = {"nonce": sess.get("nonce"), "email": "user@examplectf.com"}
+
+            # Issue the password reset request
+            urandom_value = b"\xff" * 32
+            with patch("os.urandom", return_value=urandom_value):
+                client.post("/reset_password", data=data)
+
+            # Get user's original password
+            user = Users.query.filter_by(email="user@examplectf.com").first()
+
+            # Build the POST data
+            with client.session_transaction() as sess:
+                data = {"nonce": sess.get("nonce"), "password": "passwordtwo"}
+
+            # Do the password reset
+            r = client.get(
+                "/reset_password/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
+            )
+            assert r.status_code == 200
+            r = client.post(
+                "/reset_password/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
+                data=data,
+            )
+            assert r.status_code == 302
+
+            # Make sure that the user's password changed
+            user = Users.query.filter_by(email="user@examplectf.com").first()
+            assert verify_password("passwordtwo", user.password)
+
+            # Attempt to re-use the password reset link
+            r = client.get(
+                "/reset_password/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
+            )
+            assert (
+                "Your reset link is invalid, please generate a new one"
+                in r.get_data(as_text=True)
+            )
+            assert r.status_code == 200
+            with client.session_transaction() as sess:
+                data = {"nonce": sess.get("nonce"), "password": "passwordthree"}
+            r = client.post(
+                "/reset_password/ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff",
+                data=data,
+            )
+            assert r.status_code == 200
+            # Password should not have been changed
+            user = Users.query.filter_by(email="user@examplectf.com").first()
+            assert verify_password("passwordthree", user.password) is False
+    destroy_ctfd(app)
+
+
+@patch("smtplib.SMTP")
+def test_reset_password_link_tokens_unique(mock_smtp):
+    app = create_ctfd()
+    with app.app_context():
+        set_config("mail_server", "localhost")
+        set_config("mail_port", 25)
+        set_config("mail_useauth", True)
+        set_config("mail_username", "username")
+        set_config("mail_password", "password")
+        set_config("verify_emails", True)
+        register_user(app, name="user1", email="user1@examplectf.com")
+        register_user(app, name="user2", email="user2@examplectf.com")
+
+        forgot_password("user1@examplectf.com")
+        call1 = str(mock_smtp.return_value.send_message.call_args[0][0])
+
+        forgot_password("user1@examplectf.com")
+        call2 = str(mock_smtp.return_value.send_message.call_args[0][0])
+
+        forgot_password("user2@examplectf.com")
+        call3 = str(mock_smtp.return_value.send_message.call_args[0][0])
+        assert call1 != call2
+        assert call2 != call3
+        assert call1 != call3
+    destroy_ctfd(app)