Verify release URLs using Trusted Publisher information

tests/unit/forklift/test_legacy.py

@@ -58,6 +58,7 @@
     Project,
     ProjectMacaroonWarningAssociation,
     Release,
+    ReleaseURL,
     Role,
 )
 from warehouse.packaging.tasks import sync_file_to_cache, update_bigquery_release_files
@@ -3808,6 +3809,155 @@ def failing_verify(_self, _verifier, _policy, _dist):
         assert resp.status_code == 400
         assert resp.status.startswith(expected_msg)
 
+    @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_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
+
+    def test_new_publisher_verifies_existing_release_url(
+        self,
+        monkeypatch,
+        pyramid_config,
+        db_request,
+        metrics,
+    ):
+        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.
+        release.project_urls = {
+            "verified_url": {"url": verified_url, "verified": True},
+            "unverified_url": {"url": unverified_url, "verified": False},
+        }
+        publisher = GitHubPublisherFactory.create(projects=[project])
+        publisher.repository_owner = "foo"
+        publisher.repository_name = repo_name
+        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)"),
+                ("requires_external", "Cheese (>1.0)"),
+                ("provides", "testing"),
+            ]
+        )
+        db_request.POST.add("project_urls", f"verified_url, {verified_url}")
+        db_request.POST.add("project_urls", f"unverified_url, {unverified_url}")
+
+        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)
+
+        # After successful upload, the Release should have now both URLs verified
+        release_urls = (
+            db_request.db.query(ReleaseURL).filter(Release.project == project).all()
+        )
+        release_urls = {r.name: r.verified for r in release_urls}
+        assert "verified_url" in release_urls and "unverified_url" in release_urls
+        assert release_urls["verified_url"]
+        assert release_urls["unverified_url"]
+
     @pytest.mark.parametrize(
         "version, expected_version",
         [
@@ -4614,3 +4764,86 @@ def test_missing_trailing_slash_redirect(pyramid_request):
         "/legacy/ (with a trailing slash)"
     )
     assert resp.headers["Location"] == "/legacy/"
+
+
+@pytest.mark.parametrize(
+    ("url", "publisher_url", "expected"),
+    [
+        (  # GitHub trivial case
+            "https://github.com/owner/project",
+            "https://github.com/owner/project",
+            True,
+        ),
+        (  # ActiveState trivial case
+            "https://platform.activestate.com/owner/project",
+            "https://platform.activestate.com/owner/project",
+            True,
+        ),
+        (  # GitLab trivial case
+            "https://gitlab.com/owner/project",
+            "https://gitlab.com/owner/project",
+            True,
+        ),
+        (  # URL is a sub-path of the TP URL
+            "https://github.com/owner/project/issues",
+            "https://github.com/owner/project",
+            True,
+        ),
+        (  # Normalization
+            "https://github.com/owner/project/",
+            "https://github.com/owner/project",
+            True,
+        ),
+        (  # TP URL is a prefix, but not a parent of the URL
+            "https://github.com/owner/project22",
+            "https://github.com/owner/project",
+            False,
+        ),
+        (  # URL is a parent of the TP URL
+            "https://github.com/owner",
+            "https://github.com/owner/project",
+            False,
+        ),
+        (  # Scheme component does not match
+            "http://github.com/owner/project",
+            "https://github.com/owner/project",
+            False,
+        ),
+        (  # Host component does not match
+            "https://gitlab.com/owner/project",
+            "https://github.com/owner/project",
+            False,
+        ),
+        (  # Host component matches, but contains user and port info
+            "https://user@github.com:443/owner/project",
+            "https://github.com/owner/project",
+            False,
+        ),
+        (  # URL path component is empty
+            "https://github.com",
+            "https://github.com/owner/project",
+            False,
+        ),
+        (  # TP URL path component is empty
+            # (currently no TPs have an empty path, so even if the given URL is a
+            # sub-path of the TP URL, we fail the verification)
+            "https://github.com/owner/project",
+            "https://github.com",
+            False,
+        ),
+        (  # Both path components are empty
+            # (currently no TPs have an empty path, so even if the given URL is the
+            # same as the TP URL, we fail the verification)
+            "https://github.com",
+            "https://github.com",
+            False,
+        ),
+        (  # Publisher URL is None
+            "https://github.com/owner/project",
+            None,
+            False,
+        ),
+    ],
+)
+def test_verify_url(url, publisher_url, expected):
+    assert legacy._verify_url(url, publisher_url) == expected

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

@@ -97,6 +97,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="fake@example.com")
+
+        assert publisher.publisher_base_url is None
+
     def test_publisher_url(self):
         publisher = google.GooglePublisher(email="fake@example.com")