Validate Alternate Repository Location URLS

warehouse/api/simple.py

@@ -18,6 +18,11 @@
 
 from warehouse.cache.http import add_vary, cache_control
 from warehouse.cache.origin import origin_cache
+from warehouse.constants import (
+    MIME_PYPI_SIMPLE_V1_HTML,
+    MIME_PYPI_SIMPLE_V1_JSON,
+    MIME_TEXT_HTML,
+)
 from warehouse.packaging.models import JournalEntry, Project
 from warehouse.packaging.utils import (
     _simple_detail,
@@ -26,10 +31,6 @@
 )
 from warehouse.utils.cors import _CORS_HEADERS
 
-MIME_TEXT_HTML = "text/html"
-MIME_PYPI_SIMPLE_V1_HTML = "application/vnd.pypi.simple.v1+html"
-MIME_PYPI_SIMPLE_V1_JSON = "application/vnd.pypi.simple.v1+json"
-
 
 def _select_content_type(request: Request) -> str:
     # The way this works, is this will return a list of

warehouse/constants.py

@@ -14,3 +14,12 @@
 ONE_GIB = 1 * 1024 * 1024 * 1024
 MAX_FILESIZE = 100 * ONE_MIB
 MAX_PROJECT_SIZE = 10 * ONE_GIB
+
+MIME_TEXT_HTML = "text/html"
+MIME_PYPI_SIMPLE_V1_HTML = "application/vnd.pypi.simple.v1+html"
+MIME_PYPI_SIMPLE_V1_JSON = "application/vnd.pypi.simple.v1+json"
+
+MIME_PYPI_SIMPLE_V1_ALL = [
+    MIME_PYPI_SIMPLE_V1_JSON,
+    MIME_PYPI_SIMPLE_V1_HTML,
+]

warehouse/manage/forms.py

@@ -14,6 +14,8 @@
 
 import wtforms
 
+from urllib3 import PoolManager, Timeout
+
 import warehouse.utils.otp as otp
 import warehouse.utils.webauthn as webauthn
 
@@ -25,6 +27,7 @@
     TOTPValueMixin,
     WebAuthnCredentialMixin,
 )
+from warehouse.constants import MIME_PYPI_SIMPLE_V1_ALL
 from warehouse.i18n import localize as _
 from warehouse.organizations.models import (
     OrganizationRoleType,
@@ -713,6 +716,23 @@ class CreateTeamForm(SaveTeamForm):
     __params__ = SaveTeamForm.__params__
 
 
+def validate_is_simple(form, field):
+    try:
+        timeout = Timeout(connect=1.0, read=1.0)
+        http = PoolManager(timeout=timeout)
+        response = http.request(
+            "HEAD", field.data, headers={"Accept": ", ".join(MIME_PYPI_SIMPLE_V1_ALL)}
+        )
+        if response.headers.get("Content-Type") not in MIME_PYPI_SIMPLE_V1_ALL:
+            raise wtforms.validators.ValidationError(
+                _("Unable to parse simple index at given url")
+            )
+    except Exception as exc:
+        raise wtforms.validators.ValidationError(
+            _(f"Unable to parse simple index at given url: {exc}")
+        )
+
+
 class AddAlternateRepositoryForm(forms.Form):
     """Form to add an Alternate Repository Location for a Project."""
 
@@ -744,6 +764,7 @@ class AddAlternateRepositoryForm(forms.Form):
                 ),
             ),
             forms.URIValidator(),
+            validate_is_simple,
         ]
     )
     description = wtforms.TextAreaField(

warehouse/manage/views/__init__.py

@@ -1129,7 +1129,7 @@ def __init__(self, project, request):
         self.add_alternate_repository_form_class = AddAlternateRepositoryForm
 
     @view_config(request_method="GET")
-    def manage_project_settings(self):
+    def manage_project_settings(self, add_alternate_repository_form=None):
         if not self.request.organization_access:
             # Disable transfer of project to any organization.
             organization_choices = set()
@@ -1153,7 +1153,11 @@ def manage_project_settings(self):
                 active_organizations_owned | active_organizations_managed
             ) - current_organization
 
-        add_alt_repo_form = self.add_alternate_repository_form_class()
+        add_alt_repo_form = (
+            self.add_alternate_repository_form_class()
+            if add_alternate_repository_form is None
+            else add_alternate_repository_form
+        )
 
         return {
             "project": self.project,
@@ -1182,12 +1186,7 @@ def add_project_alternate_repository(self):
                 self.request._("Invalid alternate repository location details"),
                 queue="error",
             )
-            return HTTPSeeOther(
-                self.request.route_path(
-                    "manage.project.settings",
-                    project_name=self.project.name,
-                )
-            )
+            return self.manage_project_settings(add_alternate_repository_form=form)
 
         # add the alternate repository location entry
         alt_repo = AlternateRepository(