Verify Home-Page and Download-URL metadata URLs (#16568)

* Verify Home-Page and Download-URL metadata URLs

* Update warehouse/forklift/legacy.py

Co-authored-by: Dustin Ingram <[email protected]>

---------

Co-authored-by: William Woodruff <[email protected]>
Co-authored-by: Dustin Ingram <[email protected]>

Author: 3 people

tests/unit/forklift/test_legacy.py

@@ -3898,26 +3898,110 @@ def test_new_release_url_verified(
         assert release_db.urls_by_verify_status(verified=expected) == {"Test": url}
         assert not release_db.urls_by_verify_status(verified=not expected)
 
+    @pytest.mark.parametrize(
+        ("url", "expected"),
+        [
+            ("https://google.com", False),  # Totally different
+            ("https://github.com/foo", False),  # Missing parts
+            ("https://github.com/foo/bar/", True),  # Exactly the same
+            ("https://github.com/foo/bar/readme.md", True),  # Additional parts
+            ("https://github.com/foo/bar", True),  # Missing trailing slash
+        ],
+    )
+    def test_new_release_homepage_download_urls_verified(
+        self, monkeypatch, pyramid_config, db_request, metrics, url, expected
+    ):
+        project = ProjectFactory.create()
+        publisher = GitHubPublisherFactory.create(projects=[project])
+        publisher.repository_owner = "foo"
+        publisher.repository_name = "bar"
+        claims = {"sha": "somesha"}
+        identity = PublisherTokenContext(publisher, SignedClaims(claims))
+        db_request.oidc_publisher = identity.publisher
+        db_request.oidc_claims = identity.claims
+
+        db_request.db.add(Classifier(classifier="Environment :: Other Environment"))
+        db_request.db.add(Classifier(classifier="Programming Language :: Python"))
+
+        filename = "{}-{}.tar.gz".format(project.name, "1.0")
+
+        pyramid_config.testing_securitypolicy(identity=identity)
+        db_request.user_agent = "warehouse-tests/6.6.6"
+        db_request.POST = MultiDict(
+            {
+                "metadata_version": "1.2",
+                "name": project.name,
+                "version": "1.0",
+                "summary": "This is my summary!",
+                "filetype": "sdist",
+                "md5_digest": _TAR_GZ_PKG_MD5,
+                "content": pretend.stub(
+                    filename=filename,
+                    file=io.BytesIO(_TAR_GZ_PKG_TESTDATA),
+                    type="application/tar",
+                ),
+            }
+        )
+        db_request.POST.extend(
+            [
+                ("classifiers", "Environment :: Other Environment"),
+                ("classifiers", "Programming Language :: Python"),
+                ("requires_dist", "foo"),
+                ("requires_dist", "bar (>1.0)"),
+                ("home_page", url),
+                ("download_url", url),
+                ("requires_external", "Cheese (>1.0)"),
+                ("provides", "testing"),
+            ]
+        )
+
+        storage_service = pretend.stub(store=lambda path, filepath, meta: None)
+        db_request.find_service = lambda svc, name=None, context=None: {
+            IFileStorage: storage_service,
+            IMetricsService: metrics,
+        }.get(svc)
+
+        legacy.file_upload(db_request)
+        release_db = (
+            db_request.db.query(Release).filter(Release.project == project).one()
+        )
+        assert release_db.urls_by_verify_status(verified=expected) == {
+            "Homepage": url,
+            "Download": url,
+        }
+        assert not release_db.urls_by_verify_status(verified=not expected)
+
+    @pytest.mark.parametrize(
+        ("home_page_verified", "download_url_verified"),
+        [(False, False), (False, True), (True, False), (True, True)],
+    )
     def test_new_publisher_verifies_existing_release_url(
         self,
         monkeypatch,
         pyramid_config,
         db_request,
         metrics,
+        home_page_verified,
+        download_url_verified,
     ):
         repo_name = "my_new_repo"
         verified_url = "https://github.com/foo/bar"
         unverified_url = f"https://github.com/foo/{repo_name}"
 
         project = ProjectFactory.create()
         release = ReleaseFactory.create(project=project, version="1.0")
-        # We start with an existing release, with one verified URL and one unverified
-        # URL. Uploading a new file with a Trusted Publisher that matches the unverified
-        # URL should mark it as verified.
+        # We start with an existing release, with one verified URL and some unverified
+        # URLs. Uploading a new file with a Trusted Publisher that matches the
+        # unverified URLs should mark them as verified.
         release.project_urls = {
             "verified_url": {"url": verified_url, "verified": True},
             "unverified_url": {"url": unverified_url, "verified": False},
         }
+        release.home_page = verified_url if home_page_verified else unverified_url
+        release.home_page_verified = home_page_verified
+        release.download_url = verified_url if download_url_verified else unverified_url
+        release.download_url_verified = download_url_verified
+
         publisher = GitHubPublisherFactory.create(projects=[project])
         publisher.repository_owner = "foo"
         publisher.repository_name = repo_name
@@ -3956,6 +4040,11 @@ def test_new_publisher_verifies_existing_release_url(
                 ("requires_dist", "bar (>1.0)"),
                 ("requires_external", "Cheese (>1.0)"),
                 ("provides", "testing"),
+                ("home_page", verified_url if home_page_verified else unverified_url),
+                (
+                    "download_url",
+                    verified_url if download_url_verified else unverified_url,
+                ),
             ]
         )
         db_request.POST.add("project_urls", f"verified_url, {verified_url}")
@@ -3969,13 +4058,16 @@ def test_new_publisher_verifies_existing_release_url(
 
         legacy.file_upload(db_request)
 
-        # After successful upload, the Release should have now both URLs verified
+        # After successful upload, the Release should have now all URLs verified
         release_db = (
             db_request.db.query(Release).filter(Release.project == project).one()
         )
+
         assert release_db.urls_by_verify_status(verified=True) == {
             "unverified_url": unverified_url,
             "verified_url": verified_url,
+            "Homepage": release.home_page,
+            "Download": release.download_url,
         }
         assert not release_db.urls_by_verify_status(verified=False)
 

warehouse/forklift/legacy.py

@@ -879,6 +879,30 @@ def file_upload(request):
             for name, url in meta.project_urls.items()
         }
     )
+    home_page = meta.home_page
+    home_page_verified = (
+        False
+        if home_page is None
+        else _verify_url(
+            url=home_page,
+            publisher=request.oidc_publisher,
+            project_name=project.name,
+            project_normalized_name=project.normalized_name,
+        )
+    )
+
+    download_url = meta.download_url
+    download_url_verified = (
+        False
+        if download_url is None
+        else _verify_url(
+            url=download_url,
+            publisher=request.oidc_publisher,
+            project_name=project.name,
+            project_normalized_name=project.normalized_name,
+        )
+    )
+
     try:
         is_new_release = False
         canonical_version = packaging.utils.canonicalize_version(meta.version)
@@ -934,6 +958,10 @@ def file_upload(request):
                 html=rendered or "",
                 rendered_by=readme.renderer_version(),
             ),
+            home_page=home_page,
+            home_page_verified=home_page_verified,
+            download_url=download_url,
+            download_url_verified=download_url_verified,
             project_urls=project_urls,
             # TODO: Fix this, we currently treat platform as if it is a single
             #       use field, but in reality it is a multi-use field, which the
@@ -964,8 +992,6 @@ def file_upload(request):
                     "author_email",
                     "maintainer",
                     "maintainer_email",
-                    "home_page",
-                    "download_url",
                     "provides_extra",
                 }
             },
@@ -1384,6 +1410,11 @@ def file_upload(request):
             ):
                 release_url.verified = True
 
+        if home_page_verified and not release.home_page_verified:
+            release.home_page_verified = True
+        if download_url_verified and not release.download_url_verified:
+            release.download_url_verified = True
+
     request.db.flush()  # flush db now so server default values are populated for celery
 
     # Push updates to BigQuery

warehouse/migrations/versions/606abd3b8e7f_add_columns_to_release_for_verified_url.py

@@ -0,0 +1,51 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""
+Add columns to Release for verified URL status
+
+Revision ID: 606abd3b8e7f
+Revises: 7ca0f1f5e7b3
+Create Date: 2024-08-26 08:31:32.734838
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "606abd3b8e7f"
+down_revision = "7ca0f1f5e7b3"
+
+
+def upgrade():
+    op.add_column(
+        "releases",
+        sa.Column(
+            "home_page_verified",
+            sa.Boolean(),
+            server_default=sa.text("false"),
+            nullable=False,
+        ),
+    )
+    op.add_column(
+        "releases",
+        sa.Column(
+            "download_url_verified",
+            sa.Boolean(),
+            server_default=sa.text("false"),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("releases", "download_url_verified")
+    op.drop_column("releases", "home_page_verified")

warehouse/packaging/models.py

@@ -559,6 +559,7 @@ def __table_args__(cls):  # noqa
     maintainer: Mapped[str | None]
     maintainer_email: Mapped[str | None]
     home_page: Mapped[str | None]
+    home_page_verified: Mapped[bool_false]
     license: Mapped[str | None]
     summary: Mapped[str | None]
     keywords: Mapped[str | None]
@@ -571,6 +572,7 @@ def __table_args__(cls):  # noqa
     )
     platform: Mapped[str | None]
     download_url: Mapped[str | None]
+    download_url_verified: Mapped[bool_false]
     _pypi_ordering: Mapped[int | None]
     requires_python: Mapped[str | None] = mapped_column(Text)
     created: Mapped[datetime_now] = mapped_column()
@@ -690,30 +692,15 @@ def urls(self):
         return _urls
 
     def urls_by_verify_status(self, *, verified: bool):
-        verified_urls = {
+        matching_urls = {
             release_url.url
-            for release_url in self._project_urls.values()  # type: ignore[attr-defined]
-            if release_url.verified
+            for release_url in self._project_urls.values()  # type: ignore[attr-defined] # noqa: E501
+            if release_url.verified == verified
         }
-
-        if verified:
-            matching_urls = verified_urls
-        else:
-            matching_urls = {
-                release_url.url
-                for release_url in self._project_urls.values()  # type: ignore[attr-defined] # noqa: E501
-                if not release_url.verified
-            }
-            # The Homepage and Download URLs in the release metadata are currently not
-            # verified, so we return them if the user requests non-verified URLs *and*
-            # they are not verified in `project_urls`.
-            matching_urls.update(
-                {
-                    url
-                    for url in (self.home_page, self.download_url)
-                    if url is not None and url not in verified_urls
-                }
-            )
+        if self.home_page and self.home_page_verified == verified:
+            matching_urls.add(self.home_page)
+        if self.download_url and self.download_url_verified == verified:
+            matching_urls.add(self.download_url)
 
         # Filter the output of `Release.urls`, since it has custom logic to de-duplicate
         # release URLs