Create and populate verified field for ReleaseUrl

tests/unit/forklift/test_legacy.py

@@ -48,6 +48,7 @@
     Project,
     ProjectMacaroonWarningAssociation,
     Release,
+    ReleaseURL,
     Role,
 )
 from warehouse.packaging.tasks import sync_file_to_cache, update_bigquery_release_files
@@ -3292,6 +3293,75 @@ def test_upload_succeeds_creates_release(
             ),
         ]
 
+    @pytest.mark.parametrize(
+        "url, expected",
+        [
+            ("https://xpto.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),  # Additonal parts
+            ("https://github.com/foo/bar", True),  # Missing trailing slash
+        ],
+    )
+    def test_release_url_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)"),
+                ("project_urls", f"Test, {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_url = (
+            db_request.db.query(ReleaseURL).filter(Release.project == project).one()
+        )
+        assert release_url is not None
+        assert release_url.verified == expected
+
     @pytest.mark.parametrize(
         "version, expected_version",
         [

tests/unit/oidc/models/test_activestate.py

@@ -84,6 +84,18 @@ def test_publisher_name(self):
 
         assert publisher.publisher_name == "ActiveState"
 
+    def test_publisher_base_url(self):
+        org_name = "fakeorg"
+        project_name = "fakeproject"
+        publisher = ActiveStatePublisher(
+            organization=org_name, activestate_project_name=project_name
+        )
+
+        assert (
+            publisher.publisher_base_url
+            == f"https://platform.activestate.com/{org_name}/{project_name}"
+        )
+
     def test_publisher_url(self):
         org_name = "fakeorg"
         project_name = "fakeproject"

tests/unit/oidc/models/test_github.py

@@ -96,6 +96,7 @@ def test_github_publisher_computed_properties(self):
             assert getattr(publisher, claim_name) is not None
 
         assert str(publisher) == "fakeworkflow.yml"
+        assert publisher.publisher_base_url == "https://github.com/fakeowner/fakerepo"
         assert publisher.publisher_url() == "https://github.com/fakeowner/fakerepo"
         assert (
             publisher.publisher_url({"sha": "somesha"})

tests/unit/oidc/models/test_gitlab.py

@@ -95,6 +95,7 @@ def test_gitlab_publisher_computed_properties(self):
             assert getattr(publisher, claim_name) is not None
 
         assert str(publisher) == "subfolder/fakeworkflow.yml"
+        assert publisher.publisher_base_url == "https://gitlab.com/fakeowner/fakerepo"
         assert publisher.publisher_url() == "https://gitlab.com/fakeowner/fakerepo"
         assert (
             publisher.publisher_url({"sha": "somesha"})

tests/unit/oidc/models/test_google.py

@@ -33,6 +33,11 @@ def test_publisher_name(self):
 
         assert publisher.publisher_name == "Google"
 
+    def test_publisher_base_url(self):
+        publisher = google.GooglePublisher(email="[email protected]")
+
+        assert publisher.publisher_base_url is None
+
     def test_publisher_url(self):
         publisher = google.GooglePublisher(email="[email protected]")
 

warehouse/forklift/legacy.py

@@ -676,6 +676,22 @@ def file_upload(request):
                 ),
             ) from None
 
+    # Verify any verifiable URLs
+    publisher_base_url = (
+        request.oidc_publisher.publisher_base_url if request.oidc_publisher else None
+    )
+    project_urls = (
+        {}
+        if not meta.project_urls
+        else {
+            name: {
+                "url": url,
+                "verified": publisher_base_url
+                and url.lower().startswith(publisher_base_url.lower()),
+            }
+            for name, url in meta.project_urls.items()
+        }
+    )
     try:
         canonical_version = packaging.utils.canonicalize_version(meta.version)
         release = (
@@ -730,7 +746,7 @@ def file_upload(request):
                 html=rendered or "",
                 rendered_by=readme.renderer_version(),
             ),
-            project_urls=meta.project_urls or {},
+            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
             #       packaging.metadata library handles correctly.

warehouse/migrations/versions/26455e3712a2_create_verified_field_for_releaseurl.py

@@ -0,0 +1,38 @@
+# 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.
+"""
+create verified field for ReleaseUrl
+
+Revision ID: 26455e3712a2
+Revises: b14df478c48f
+Create Date: 2024-04-30 18:40:17.149050
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "26455e3712a2"
+down_revision = "b14df478c48f"
+
+
+def upgrade():
+    op.add_column(
+        "release_urls",
+        sa.Column(
+            "verified", sa.Boolean(), server_default=sa.text("false"), nullable=False
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("release_urls", "verified")

warehouse/oidc/models/_core.py

@@ -235,6 +235,11 @@ def publisher_name(self) -> str:  # pragma: no cover
         # Only concrete subclasses are constructed.
         raise NotImplementedError
 
+    @property
+    def publisher_base_url(self) -> str | None:  # pragma: no cover
+        # Only concrete subclasses are constructed.
+        raise NotImplementedError
+
     def publisher_url(
         self, claims: SignedClaims | None = None
     ) -> str | None:  # pragma: no cover

warehouse/oidc/models/activestate.py

@@ -118,11 +118,15 @@ def publisher_name(self) -> str:
     def project(self) -> str:
         return self.activestate_project_name
 
-    def publisher_url(self, claims: SignedClaims | None = None) -> str:
+    @property
+    def publisher_base_url(self) -> str:
         return urllib.parse.urljoin(
             _ACTIVESTATE_URL, f"{self.organization}/{self.activestate_project_name}"
         )
 
+    def publisher_url(self, claims: SignedClaims | None = None) -> str:
+        return self.publisher_base_url
+
     def stored_claims(self, claims=None):
         return {}
 

warehouse/oidc/models/github.py

@@ -232,8 +232,12 @@ def job_workflow_ref(self):
     def sub(self):
         return f"repo:{self.repository}"
 
+    @property
+    def publisher_base_url(self):
+        return f"https://github.com/{self.repository}"
+
     def publisher_url(self, claims=None):
-        base = f"https://github.com/{self.repository}"
+        base = self.publisher_base_url
         sha = claims.get("sha") if claims else None
 
         if sha:

warehouse/oidc/models/gitlab.py

@@ -222,8 +222,12 @@ def ci_config_ref_uri(self):
     def publisher_name(self):
         return "GitLab"
 
+    @property
+    def publisher_base_url(self):
+        return f"https://gitlab.com/{self.project_path}"
+
     def publisher_url(self, claims=None):
-        base = f"https://gitlab.com/{self.project_path}"
+        base = self.publisher_base_url
         return f"{base}/commit/{claims['sha']}" if claims else base
 
     def stored_claims(self, claims=None):

warehouse/oidc/models/google.py

@@ -81,6 +81,10 @@ def __lookup_no_sub__(klass, signed_claims: SignedClaims) -> Query | None:
     def publisher_name(self):
         return "Google"
 
+    @property
+    def publisher_base_url(self):
+        return None
+
     def publisher_url(self, claims=None):
         return None
 

warehouse/packaging/models.py

@@ -476,6 +476,7 @@ class ReleaseURL(db.Model):
 
     name: Mapped[str] = mapped_column(String(32))
     url: Mapped[str]
+    verified: Mapped[bool] = mapped_column(default=False)
 
 
 DynamicFieldsEnum = ENUM(
@@ -578,7 +579,7 @@ def __table_args__(cls):  # noqa
     project_urls = association_proxy(
         "_project_urls",
         "url",
-        creator=lambda k, v: ReleaseURL(name=k, url=v),
+        creator=lambda k, v: ReleaseURL(name=k, url=v["url"], verified=v["verified"]),
     )
 
     files: Mapped[list[File]] = orm.relationship(