IProjectService.check_project_name

Extract the distribution name validation logic into a separate function
for reuse elsewhere.

tests/unit/packaging/test_services.py

@@ -24,7 +24,7 @@
 
 import warehouse.packaging.services
 
-from warehouse.packaging.interfaces import IDocsStorage, IFileStorage, ISimpleStorage
+from warehouse.packaging.interfaces import IDocsStorage, IFileStorage, ISimpleStorage, ProjectNameUnavailableReason
 from warehouse.packaging.services import (
     B2FileStorage,
     GCSFileStorage,
@@ -35,6 +35,7 @@
     LocalFileStorage,
     LocalSimpleStorage,
     S3ArchiveFileStorage,
+    ProjectService,
     S3DocsStorage,
     S3FileStorage,
     project_service_factory,
@@ -979,6 +980,11 @@ def test_notimplementederror(self):
             GenericLocalBlobStorage.create_service(pretend.stub(), pretend.stub())
 
 
+class TestProjectService:
+    def test_verify_service(self):
+        assert verifyClass(IProjectService, ProjectService)
+
+
 def test_project_service_factory():
     db = pretend.stub()
     request = pretend.stub(

warehouse/packaging/interfaces.py

@@ -73,7 +73,20 @@ def remove_by_prefix(prefix):
         """
 
 
+class ProjectNameUnavailableReason(enum.Enum):
+    Invalid = "invalid"
+    Stdlib = "stdlib"
+    AlreadyExists = "already-exists"
+    Prohibited = "prohibited"
+    TooSimilar = "too-similar"
+
+
 class IProjectService(Interface):
+    def check_project_name(name, request):
+        """
+        Check if a project name is valid and available for use.
+        """
+
     def create_project(name, creator, request, *, creator_is_owner=True):
         """
         Creates a new project, recording a user as its creator.

warehouse/packaging/services.py

@@ -42,6 +42,7 @@
 from warehouse.packaging.interfaces import (
     IDocsStorage,
     IFileStorage,
+    ProjectNameUnavailableReason,
     IProjectService,
     ISimpleStorage,
     TooManyProjectsCreated,
@@ -442,63 +443,69 @@ def _hit_ratelimits(self, request, creator):
         self.ratelimiters["project.create.user"].hit(creator.id)
         self.ratelimiters["project.create.ip"].hit(request.remote_addr)
 
+    def check_project_name(self, name, request):
+        if not PROJECT_NAME_RE.match(name):
+            return ProjectNameUnavailableReason.Invalid
+
+        # Also check for collisions with Python Standard Library modules.
+        if canonicalize_name(name) in STDLIB_PROHIBITED:
+            return ProjectNameUnavailableReason.Stdlib
+
+        if request.db.query(
+            exists().where(Project.normalized_name == func.normalize_pep426_name(name))
+        ).scalar():
+            return ProjectNameUnavailableReason.AlreadyExists
+
+        if request.db.query(
+            exists().where(
+                ProhibitedProjectName.name == func.normalize_pep426_name(name)
+            )
+        ).scalar():
+            return ProjectNameUnavailableReason.Prohibited
+
+        if request.db.query(
+            exists().where(
+                func.ultranormalize_name(Project.name)
+                == func.ultranormalize_name(name)
+            )
+        ).scalar():
+            return ProjectNameUnavailableReason.TooSimilar
+
     def create_project(
         self, name, creator, request, *, creator_is_owner=True, ratelimited=True
     ):
         if ratelimited:
             self._check_ratelimits(request, creator)
 
-        # Sanity check that the project name is valid. This may have already
-        # happened via form validation prior to calling this function, but
-        # isn't guaranteed.
-        if not PROJECT_NAME_RE.match(name):
-            raise HTTPBadRequest(f"The name {name!r} is invalid.")
-
-        # Look up the project first before doing anything else, and fail if it
-        # already exists. If it does not exist, proceed with additional checks
-        # to ensure that the project has a valid name before creating it.
-        try:
-            # Find existing project or raise NoResultFound.
-            (
-                request.db.query(Project.id)
-                .filter(Project.normalized_name == func.normalize_pep426_name(name))
-                .one()
-            )
-
-            # Found existing project with conflicting name.
-            raise HTTPConflict(
+        # Check for AdminFlag set by a PyPI Administrator disabling new project
+        # registration, reasons for this include Spammers, security
+        # vulnerabilities, or just wanting to be lazy and not worry ;)
+        if request.flags.enabled(AdminFlagValue.DISALLOW_NEW_PROJECT_REGISTRATION):
+            raise HTTPForbidden(
                 (
-                    "The name {name!r} conflicts with an existing project. "
+                    "New project registration temporarily disabled. "
                     "See {projecthelp} for more information."
                 ).format(
-                    name=name,
-                    projecthelp=request.help_url(_anchor="project-name"),
+                    projecthelp=request.help_url(_anchor="admin-intervention")
                 ),
             ) from None
-        except NoResultFound:
-            # Check for AdminFlag set by a PyPI Administrator disabling new project
-            # registration, reasons for this include Spammers, security
-            # vulnerabilities, or just wanting to be lazy and not worry ;)
-            if request.flags.enabled(AdminFlagValue.DISALLOW_NEW_PROJECT_REGISTRATION):
-                raise HTTPForbidden(
+
+        # Verify that the project name is both valid and currently available.
+        match self.check_project_name(name, request):
+            case ProjectNameUnavailableReason.Invalid:
+                raise HTTPBadRequest(f"The name {name!r} is invalid.")
+            case ProjectNameUnavailableReason.AlreadyExists:
+                # Found existing project with conflicting name.
+                raise HTTPConflict(
                     (
-                        "New project registration temporarily disabled. "
+                        "The name {name!r} conflicts with an existing project. "
                         "See {projecthelp} for more information."
                     ).format(
-                        projecthelp=request.help_url(_anchor="admin-intervention")
+                        name=name,
+                        projecthelp=request.help_url(_anchor="project-name"),
                     ),
                 ) from None
-
-            # Before we create the project, we're going to check our prohibited
-            # names to see if this project name prohibited, or if the project name
-            # is a close approximation of an existing project name. If it is,
-            # then we're going to deny the request to create this project.
-            _prohibited_name = request.db.query(
-                exists().where(
-                    ProhibitedProjectName.name == func.normalize_pep426_name(name)
-                )
-            ).scalar()
-            if _prohibited_name:
+            case ProjectNameUnavailableReason.Prohibited:
                 raise HTTPBadRequest(
                     (
                         "The name {name!r} isn't allowed. "
@@ -508,14 +515,7 @@ def create_project(
                         projecthelp=request.help_url(_anchor="project-name"),
                     ),
                 ) from None
-
-            _ultranormalize_collision = request.db.query(
-                exists().where(
-                    func.ultranormalize_name(Project.name)
-                    == func.ultranormalize_name(name)
-                )
-            ).scalar()
-            if _ultranormalize_collision:
+            case ProjectNameUnavailableReason.TooSimilar:
                 raise HTTPBadRequest(
                     (
                         "The name {name!r} is too similar to an existing project. "
@@ -525,9 +525,7 @@ def create_project(
                         projecthelp=request.help_url(_anchor="project-name"),
                     ),
                 ) from None
-
-            # Also check for collisions with Python Standard Library modules.
-            if canonicalize_name(name) in STDLIB_PROHIBITED:
+            case ProjectNameUnavailableReason.Stdlib:
                 raise HTTPBadRequest(
                     (
                         "The name {name!r} isn't allowed (conflict with Python "
@@ -538,6 +536,8 @@ def create_project(
                         projecthelp=request.help_url(_anchor="project-name"),
                     ),
                 ) from None
+            case None:
+                break
 
         # The project name is valid: create it and add it
         project = Project(name=name)
@@ -596,6 +596,7 @@ def create_project(
             )
             .all()
         )
+        # TODO: Also dispose of pending publishers with ultranormalized collisions
         for stale_publisher in stale_pending_publishers:
             send_pending_trusted_publisher_invalidated_email(
                 request,