Move exists() check to Trusted Publisher classes

Signed-off-by: Facundo Tuesca <facundo.tuesca@trailofbits.com>

tests/unit/oidc/models/test_activestate.py

@@ -366,6 +366,21 @@ def test_activestate_publisher_verify_url(self, url, expected):
         )
         assert publisher.verify_url(url) == expected
 
+    @pytest.mark.parametrize("exists_in_db", [True, False])
+    def test_exists(self, db_request, exists_in_db):
+        publisher = ActiveStatePublisher(
+            organization="repository_name",
+            activestate_project_name="project_name",
+            actor_id=ACTOR_ID,
+            actor=ACTOR,
+        )
+
+        if exists_in_db:
+            db_request.db.add(publisher)
+            db_request.db.flush()
+
+        assert publisher.exists(db_request.db) == exists_in_db
+
 
 class TestPendingActiveStatePublisher:
     def test_reify_does_not_exist_yet(self, db_request):

tests/unit/oidc/models/test_github.py

@@ -685,6 +685,22 @@ def test_github_publisher_attestation_identity(self, environment):
         else:
             assert identity.environment == publisher.environment
 
+    @pytest.mark.parametrize("exists_in_db", [True, False])
+    def test_exists(self, db_request, exists_in_db):
+        publisher = github.GitHubPublisher(
+            repository_name="repository_name",
+            repository_owner="repository_owner",
+            repository_owner_id="666",
+            workflow_filename="workflow_filename.yml",
+            environment="",
+        )
+
+        if exists_in_db:
+            db_request.db.add(publisher)
+            db_request.db.flush()
+
+        assert publisher.exists(db_request.db) == exists_in_db
+
 
 class TestPendingGitHubPublisher:
     def test_reify_does_not_exist_yet(self, db_request):

tests/unit/oidc/models/test_gitlab.py

@@ -741,6 +741,21 @@ def test_gitlab_publisher_attestation_identity(self, environment):
         else:
             assert identity.environment == publisher.environment
 
+    @pytest.mark.parametrize("exists_in_db", [True, False])
+    def test_exists(self, db_request, exists_in_db):
+        publisher = gitlab.GitLabPublisher(
+            project="repository_name",
+            namespace="repository_owner",
+            workflow_filepath="subfolder/worflow_filename.yml",
+            environment="",
+        )
+
+        if exists_in_db:
+            db_request.db.add(publisher)
+            db_request.db.flush()
+
+        assert publisher.exists(db_request.db) == exists_in_db
+
 
 class TestPendingGitLabPublisher:
     def test_reify_does_not_exist_yet(self, db_request):

tests/unit/oidc/models/test_google.py

@@ -196,6 +196,19 @@ def test_google_publisher_sub_is_optional(self, expected_sub, actual_sub, valid)
                 )
             assert str(e.value) == "Check failed for optional claim 'sub'"
 
+    @pytest.mark.parametrize("exists_in_db", [True, False])
+    def test_exists(self, db_request, exists_in_db):
+        publisher = google.GooglePublisher(
+            email="fake@example.com",
+            sub="fakesubject",
+        )
+
+        if exists_in_db:
+            db_request.db.add(publisher)
+            db_request.db.flush()
+
+        assert publisher.exists(db_request.db) == exists_in_db
+
 
 class TestPendingGooglePublisher:
     @pytest.mark.parametrize("sub", ["fakesubject", None])

warehouse/locale/messages.pot

@@ -297,7 +297,7 @@ msgid "You are now ${role} of the '${project_name}' project."
 msgstr ""
 
 #: warehouse/accounts/views.py:1593 warehouse/accounts/views.py:1836
-#: warehouse/manage/views/__init__.py:1440
+#: warehouse/manage/views/__init__.py:1417
 msgid ""
 "Trusted publishing is temporarily disabled. See https://pypi.org/help"
 "#admin-intervention for details."
@@ -317,19 +317,19 @@ msgstr ""
 msgid "You can't register more than 3 pending trusted publishers at once."
 msgstr ""
 
-#: warehouse/accounts/views.py:1659 warehouse/manage/views/__init__.py:1621
-#: warehouse/manage/views/__init__.py:1736
-#: warehouse/manage/views/__init__.py:1850
-#: warehouse/manage/views/__init__.py:1962
+#: warehouse/accounts/views.py:1659 warehouse/manage/views/__init__.py:1598
+#: warehouse/manage/views/__init__.py:1713
+#: warehouse/manage/views/__init__.py:1827
+#: warehouse/manage/views/__init__.py:1939
 msgid ""
 "There have been too many attempted trusted publisher registrations. Try "
 "again later."
 msgstr ""
 
-#: warehouse/accounts/views.py:1669 warehouse/manage/views/__init__.py:1634
-#: warehouse/manage/views/__init__.py:1749
-#: warehouse/manage/views/__init__.py:1863
-#: warehouse/manage/views/__init__.py:1975
+#: warehouse/accounts/views.py:1669 warehouse/manage/views/__init__.py:1611
+#: warehouse/manage/views/__init__.py:1726
+#: warehouse/manage/views/__init__.py:1840
+#: warehouse/manage/views/__init__.py:1952
 msgid "The trusted publisher could not be registered"
 msgstr ""
 
@@ -496,10 +496,10 @@ msgid "Added alternate repository '${name}'"
 msgstr ""
 
 #: warehouse/manage/views/__init__.py:1260
-#: warehouse/manage/views/__init__.py:2310
-#: warehouse/manage/views/__init__.py:2395
-#: warehouse/manage/views/__init__.py:2496
-#: warehouse/manage/views/__init__.py:2596
+#: warehouse/manage/views/__init__.py:2287
+#: warehouse/manage/views/__init__.py:2372
+#: warehouse/manage/views/__init__.py:2473
+#: warehouse/manage/views/__init__.py:2573
 msgid "Confirm the request"
 msgstr ""
 
@@ -523,105 +523,105 @@ msgstr ""
 msgid "Deleted alternate repository '${name}'"
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:1489
+#: warehouse/manage/views/__init__.py:1466
 msgid "The trusted publisher could not be constrained"
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:1602
+#: warehouse/manage/views/__init__.py:1579
 msgid ""
 "GitHub-based trusted publishing is temporarily disabled. See "
 "https://pypi.org/help#admin-intervention for details."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:1717
+#: warehouse/manage/views/__init__.py:1694
 msgid ""
 "GitLab-based trusted publishing is temporarily disabled. See "
 "https://pypi.org/help#admin-intervention for details."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:1831
+#: warehouse/manage/views/__init__.py:1808
 msgid ""
 "Google-based trusted publishing is temporarily disabled. See "
 "https://pypi.org/help#admin-intervention for details."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:1942
+#: warehouse/manage/views/__init__.py:1919
 msgid ""
 "ActiveState-based trusted publishing is temporarily disabled. See "
 "https://pypi.org/help#admin-intervention for details."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2179
-#: warehouse/manage/views/__init__.py:2480
-#: warehouse/manage/views/__init__.py:2588
+#: warehouse/manage/views/__init__.py:2156
+#: warehouse/manage/views/__init__.py:2457
+#: warehouse/manage/views/__init__.py:2565
 msgid ""
 "Project deletion temporarily disabled. See https://pypi.org/help#admin-"
 "intervention for details."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2323
+#: warehouse/manage/views/__init__.py:2300
 msgid "Could not yank release - "
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2408
+#: warehouse/manage/views/__init__.py:2385
 msgid "Could not un-yank release - "
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2509
+#: warehouse/manage/views/__init__.py:2486
 msgid "Could not delete release - "
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2608
+#: warehouse/manage/views/__init__.py:2585
 msgid "Could not find file"
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2613
+#: warehouse/manage/views/__init__.py:2590
 msgid "Could not delete file - "
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2763
+#: warehouse/manage/views/__init__.py:2740
 #, python-brace-format
 msgid "Team '${team_name}' already has ${role_name} role for project"
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2870
+#: warehouse/manage/views/__init__.py:2847
 #, python-brace-format
 msgid "User '${username}' already has ${role_name} role for project"
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2937
+#: warehouse/manage/views/__init__.py:2914
 #, python-brace-format
 msgid "${username} is now ${role} of the '${project_name}' project."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2969
+#: warehouse/manage/views/__init__.py:2946
 #, python-brace-format
 msgid ""
 "User '${username}' does not have a verified primary email address and "
 "cannot be added as a ${role_name} for project"
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:2982
+#: warehouse/manage/views/__init__.py:2959
 #: warehouse/manage/views/organizations.py:890
 #, python-brace-format
 msgid "User '${username}' already has an active invite. Please try again later."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:3047
+#: warehouse/manage/views/__init__.py:3024
 #: warehouse/manage/views/organizations.py:955
 #, python-brace-format
 msgid "Invitation sent to '${username}'"
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:3079
+#: warehouse/manage/views/__init__.py:3056
 msgid "Could not find role invitation."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:3090
+#: warehouse/manage/views/__init__.py:3067
 msgid "Invitation already expired."
 msgstr ""
 
-#: warehouse/manage/views/__init__.py:3123
+#: warehouse/manage/views/__init__.py:3100
 #: warehouse/manage/views/organizations.py:1142
 #, python-brace-format
 msgid "Invitation revoked from '${username}'."

warehouse/manage/views/__init__.py

@@ -25,7 +25,7 @@
     HTTPTooManyRequests,
 )
 from pyramid.view import view_config, view_defaults
-from sqlalchemy import exists, func
+from sqlalchemy import func
 from sqlalchemy.exc import NoResultFound
 from sqlalchemy.orm import joinedload
 from venusian import lift
@@ -1384,29 +1384,6 @@ def _check_ratelimits(self):
                 )
             )
 
-    # Used in `constrain_environment` for GitHub/GitLab publishers
-    def _check_publisher_exists_in_db(
-        self, publisher: GitHubPublisher | GitLabPublisher
-    ) -> bool:
-        if isinstance(publisher, GitHubPublisher):
-            return self.request.db.query(
-                exists().where(
-                    GitHubPublisher.repository_name == publisher.repository_name,
-                    GitHubPublisher.repository_owner == publisher.repository_owner,
-                    GitHubPublisher.workflow_filename == publisher.workflow_filename,
-                    GitHubPublisher.environment == publisher.environment,
-                )
-            ).scalar()
-        else:
-            return self.request.db.query(
-                exists().where(
-                    GitLabPublisher.namespace == publisher.namespace,
-                    GitLabPublisher.project == publisher.project,
-                    GitLabPublisher.workflow_filepath == publisher.workflow_filepath,
-                    GitLabPublisher.environment == publisher.environment,
-                )
-            ).scalar()
-
     @property
     def default_response(self):
         return {
@@ -1527,7 +1504,7 @@ def constrain_environment(self):
 
         # The user might have already manually created the new constrained publisher
         # before clicking the magic link to constrain the existing publisher.
-        if self._check_publisher_exists_in_db(constrained_publisher):
+        if constrained_publisher.exists(self.request.db):
             self.request.session.flash(
                 self.request._(
                     f"{publisher} is already registered with {self.project.name}"

warehouse/oidc/models/_core.py

@@ -365,6 +365,13 @@ def verify_url(self, url: str) -> bool:
             url=url,
         )
 
+    def exists(self, session) -> bool:  # pragma: no cover
+        """
+        Check if the publisher exists in the database
+        """
+        # Only concrete subclasses are constructed.
+        raise NotImplementedError
+
 
 class OIDCPublisher(OIDCPublisherMixin, db.Model):
     __tablename__ = "oidc_publishers"

warehouse/oidc/models/activestate.py

@@ -15,7 +15,7 @@
 
 from typing import Any
 
-from sqlalchemy import ForeignKey, String, UniqueConstraint
+from sqlalchemy import ForeignKey, String, UniqueConstraint, and_, exists
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Query, mapped_column
 
@@ -135,6 +135,18 @@ def stored_claims(self, claims=None):
     def __str__(self) -> str:
         return self.publisher_url()
 
+    def exists(self, session) -> bool:
+        return session.query(
+            exists().where(
+                and_(
+                    self.__class__.organization == self.organization,
+                    self.__class__.activestate_project_name
+                    == self.activestate_project_name,
+                    self.__class__.actor_id == self.actor_id,
+                )
+            )
+        ).scalar()
+
 
 class ActiveStatePublisher(ActiveStatePublisherMixin, OIDCPublisher):
     __tablename__ = "activestate_oidc_publishers"

warehouse/oidc/models/github.py

@@ -15,7 +15,7 @@
 from typing import Any
 
 from pypi_attestations import GitHubPublisher as GitHubIdentity, Publisher
-from sqlalchemy import ForeignKey, String, UniqueConstraint
+from sqlalchemy import ForeignKey, String, UniqueConstraint, and_, exists
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Query, mapped_column
 
@@ -288,6 +288,18 @@ def stored_claims(self, claims=None):
     def __str__(self):
         return self.workflow_filename
 
+    def exists(self, session) -> bool:
+        return session.query(
+            exists().where(
+                and_(
+                    self.__class__.repository_name == self.repository_name,
+                    self.__class__.repository_owner == self.repository_owner,
+                    self.__class__.workflow_filename == self.workflow_filename,
+                    self.__class__.environment == self.environment,
+                )
+            )
+        ).scalar()
+
 
 class GitHubPublisher(GitHubPublisherMixin, OIDCPublisher):
     __tablename__ = "github_oidc_publishers"