Prevent registering recently-deleted usernames

Author: seanh

h/accounts/schemas.py

@@ -1,10 +1,12 @@
 import codecs
 import logging
+from datetime import datetime, timedelta
 from functools import lru_cache
 
 import colander
 import deform
 from markupsafe import Markup
+from sqlalchemy import select
 
 from h import i18n, models
 from h.models.user import (
@@ -16,6 +18,7 @@
 from h.schemas import validators
 from h.schemas.base import CSRFSchema
 from h.schemas.forms.accounts.util import PASSWORD_MIN_LENGTH
+from h.util.user import format_userid
 
 _ = i18n.TranslationString
 log = logging.getLogger(__name__)
@@ -46,11 +49,21 @@ def unique_email(node, value):
 
 def unique_username(node, value):
     """Colander validator that ensures the username does not exist."""
+    exc = colander.Invalid(node, _("This username is already taken."))
     request = node.bindings["request"]
     user = models.User.get_by_username(request.db, value, request.default_authority)
     if user:  # pragma: no cover
-        msg = _("This username is already taken.")
-        raise colander.Invalid(node, msg)
+        raise exc
+
+    if request.db.scalars(
+        select(models.UserDeletion.id)
+        .where(
+            models.UserDeletion.userid
+            == format_userid(value, request.default_authority)
+        )
+        .where(models.UserDeletion.requested_at > datetime.now() - timedelta(days=31))
+    ).first():
+        raise exc
 
 
 def email_node(**kwargs):

h/models/user.py

@@ -7,7 +7,7 @@
 
 from h.db import Base
 from h.exceptions import InvalidUserId
-from h.util.user import split_user
+from h.util.user import format_userid, split_user
 
 USERNAME_MIN_LENGTH = 3
 USERNAME_MAX_LENGTH = 30
@@ -231,7 +231,7 @@ def username(cls):  # pylint:disable=no-self-argument
 
     @hybrid_property
     def userid(self):
-        return f"acct:{self.username}@{self.authority}"
+        return format_userid(self.username, self.authority)
 
     @userid.comparator
     def userid(cls):  # pylint: disable=no-self-argument

h/util/user.py

@@ -19,3 +19,7 @@ def split_user(userid):
     if match:
         return {"username": match.groups()[0], "domain": match.groups()[1]}
     raise InvalidUserId(userid)
+
+
+def format_userid(username, authority):
+    return f"acct:{username}@{authority}"

tests/unit/h/accounts/schemas_test.py

@@ -1,3 +1,4 @@
+from datetime import datetime, timedelta
 from unittest.mock import Mock
 
 import colander
@@ -6,6 +7,7 @@
 
 from h.accounts import schemas
 from h.services.user_password import UserPasswordService
+from h.util.user import format_userid
 
 pytestmark = pytest.mark.usefixtures("pyramid_config")
 
@@ -72,7 +74,6 @@ def test_it_is_valid_when_authorized_users_email(
         schemas.unique_email(dummy_node, "[email protected]")
 
 
-@pytest.mark.usefixtures("user_model")
 class TestRegisterSchema:
     def test_it_is_invalid_when_password_too_short(self, pyramid_request):
         schema = schemas.RegisterSchema().bind(request=pyramid_request)
@@ -81,26 +82,21 @@ def test_it_is_invalid_when_password_too_short(self, pyramid_request):
             schema.deserialize({"password": "a"})
         assert exc.value.asdict()["password"] == ("Must be 8 characters or more.")
 
-    def test_it_is_invalid_when_username_too_short(self, pyramid_request, user_model):
+    def test_it_is_invalid_when_username_too_short(self, pyramid_request):
         schema = schemas.RegisterSchema().bind(request=pyramid_request)
-        user_model.get_by_username.return_value = None
 
         with pytest.raises(colander.Invalid) as exc:
             schema.deserialize({"username": "a"})
         assert exc.value.asdict()["username"] == ("Must be 3 characters or more.")
 
-    def test_it_is_invalid_when_username_too_long(self, pyramid_request, user_model):
+    def test_it_is_invalid_when_username_too_long(self, pyramid_request):
         schema = schemas.RegisterSchema().bind(request=pyramid_request)
-        user_model.get_by_username.return_value = None
 
         with pytest.raises(colander.Invalid) as exc:
             schema.deserialize({"username": "a" * 500})
         assert exc.value.asdict()["username"] == ("Must be 30 characters or less.")
 
-    def test_it_is_invalid_with_invalid_characters_in_username(
-        self, pyramid_request, user_model
-    ):
-        user_model.get_by_username.return_value = None
+    def test_it_is_invalid_with_invalid_characters_in_username(self, pyramid_request):
         schema = schemas.RegisterSchema().bind(request=pyramid_request)
 
         with pytest.raises(colander.Invalid) as exc:
@@ -128,24 +124,51 @@ def test_it_is_invalid_when_privacy_accepted_missing(self, pyramid_request):
 
         assert exc.value.asdict()["privacy_accepted"] == "Required"
 
-    def test_it_validates_with_valid_payload(self, pyramid_csrf_request, user_model):
-        user_model.get_by_username.return_value = None
-        user_model.get_by_email.return_value = None
+    def test_it_is_invalid_when_user_recently_deleted(
+        self, factories, pyramid_request, valid_params
+    ):
+        """If an account with the same username was recently deleted it should be invalid."""
+        schema = schemas.RegisterSchema().bind(request=pyramid_request)
+        factories.UserDeletion(
+            userid=format_userid(
+                username=valid_params["username"],
+                authority=pyramid_request.default_authority,
+            )
+        )
+
+        with pytest.raises(colander.Invalid) as exc:
+            schema.deserialize(valid_params)
 
+        assert exc.value.asdict() == {"username": "This username is already taken."}
+
+    def test_it_validates_with_valid_payload(
+        self, pyramid_csrf_request, valid_params, factories
+    ):
         schema = schemas.RegisterSchema().bind(request=pyramid_csrf_request)
-        params = {
+        # A user with the same username was deleted over a month ago.
+        # This should not prevent registration.
+        factories.UserDeletion(
+            userid=format_userid(
+                valid_params["username"], pyramid_csrf_request.default_authority
+            ),
+            requested_at=datetime.now() - timedelta(days=32),
+        )
+
+        result = schema.deserialize(valid_params)
+
+        assert result == dict(
+            valid_params, privacy_accepted=True, comms_opt_in=None, csrf_token=None
+        )
+
+    @pytest.fixture
+    def valid_params(self):
+        return {
             "username": "filbert",
             "email": "[email protected]",
             "password": "sdlkfjlk3j3iuei",
             "privacy_accepted": "true",
         }
 
-        result = schema.deserialize(params)
-
-        assert result == dict(
-            params, privacy_accepted=True, comms_opt_in=None, csrf_token=None
-        )
-
 
 @pytest.mark.usefixtures("models", "user_password_service")
 class TestEmailChangeSchema: