oidc: add missing claims check in publisher lookup (#16698)

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

Author: facutuesca

tests/unit/oidc/models/test_activestate.py

@@ -152,12 +152,6 @@ def test_activestate_publisher_all_known_claims(self):
         }
 
     def test_activestate_publisher_unaccounted_claims(self, monkeypatch):
-        publisher = ActiveStatePublisher(
-            organization=ORG_URL_NAME,
-            activestate_project_name=PROJECT_NAME,
-            actor_id=ACTOR_ID,
-        )
-
         scope = pretend.stub()
         sentry_sdk = pretend.stub(
             capture_message=pretend.call_recorder(lambda s: None),
@@ -173,9 +167,7 @@ def test_activestate_publisher_unaccounted_claims(self, monkeypatch):
         signed_claims["fake-claim"] = "fake"
         signed_claims["another-fake-claim"] = "also-fake"
 
-        assert publisher.verify_claims(
-            signed_claims=signed_claims, publisher_service=pretend.stub()
-        )
+        ActiveStatePublisher.check_claims_existence(signed_claims)
 
         assert sentry_sdk.capture_message.calls == [
             pretend.call(
@@ -225,17 +217,17 @@ def test_activestate_publisher_missing_claims(
 
         assert claim_to_drop not in signed_claims
         if valid:
+            ActiveStatePublisher.check_claims_existence(signed_claims)
             assert (
                 publisher.verify_claims(
-                    signed_claims=signed_claims, publisher_service=pretend.stub()
+                    signed_claims=signed_claims, publisher_service=pretend.stub
                 )
                 is valid
             )
         else:
             with pytest.raises(InvalidPublisherError) as e:
-                assert publisher.verify_claims(
-                    signed_claims=signed_claims, publisher_service=pretend.stub()
-                )
+                ActiveStatePublisher.check_claims_existence(signed_claims)
+
             assert str(e.value) == error_msg
             assert sentry_sdk.capture_message.calls == [
                 pretend.call(

tests/unit/oidc/models/test_core.py

@@ -51,7 +51,7 @@ def test_oidc_publisher_not_default_verifiable(self):
         publisher = _core.OIDCPublisher(projects=[])
 
         with pytest.raises(errors.InvalidPublisherError) as e:
-            publisher.verify_claims(signed_claims={}, publisher_service=pretend.stub())
+            publisher.check_claims_existence(signed_claims={})
         assert str(e.value) == "No required verifiable claims"
 
     def test_supports_attestations(self):

tests/unit/oidc/models/test_github.py

@@ -234,13 +234,6 @@ def test_github_publisher_computed_properties(self):
         }
 
     def test_github_publisher_unaccounted_claims(self, monkeypatch):
-        publisher = github.GitHubPublisher(
-            repository_name="fakerepo",
-            repository_owner="fakeowner",
-            repository_owner_id="fakeid",
-            workflow_filename="fakeworkflow.yml",
-        )
-
         scope = pretend.stub()
         sentry_sdk = pretend.stub(
             capture_message=pretend.call_recorder(lambda s: None),
@@ -259,11 +252,8 @@ def test_github_publisher_unaccounted_claims(self, monkeypatch):
         }
         signed_claims["fake-claim"] = "fake"
         signed_claims["another-fake-claim"] = "also-fake"
-        with pytest.raises(errors.InvalidPublisherError) as e:
-            publisher.verify_claims(
-                signed_claims=signed_claims, publisher_service=pretend.stub()
-            )
-        assert str(e.value) == "Check failed for required claim 'sub'"
+
+        github.GitHubPublisher.check_claims_existence(signed_claims)
         assert sentry_sdk.capture_message.calls == [
             pretend.call(
                 "JWT for GitHubPublisher has unaccounted claims: "
@@ -272,7 +262,11 @@ def test_github_publisher_unaccounted_claims(self, monkeypatch):
         ]
         assert scope.fingerprint == ["another-fake-claim", "fake-claim"]
 
-    @pytest.mark.parametrize("missing", ["sub", "ref"])
+    @pytest.mark.parametrize(
+        "missing",
+        github.GitHubPublisher.__required_verifiable_claims__.keys()
+        | github.GitHubPublisher.__required_unverifiable_claims__,
+    )
     def test_github_publisher_missing_claims(self, monkeypatch, missing):
         publisher = github.GitHubPublisher(
             repository_name="fakerepo",
@@ -301,9 +295,7 @@ def test_github_publisher_missing_claims(self, monkeypatch, missing):
         assert missing not in signed_claims
         assert publisher.__required_verifiable_claims__
         with pytest.raises(errors.InvalidPublisherError) as e:
-            publisher.verify_claims(
-                signed_claims=signed_claims, publisher_service=pretend.stub()
-            )
+            github.GitHubPublisher.check_claims_existence(signed_claims)
         assert str(e.value) == f"Missing claim {missing!r}"
         assert sentry_sdk.capture_message.calls == [
             pretend.call(f"JWT for GitHubPublisher is missing claim: {missing}")
@@ -376,6 +368,7 @@ def test_github_publisher_verifies(self, monkeypatch, environment, missing_claim
             for claim_name in github.GitHubPublisher.all_known_claims()
             if claim_name not in missing_claims
         }
+        github.GitHubPublisher.check_claims_existence(signed_claims)
         assert publisher.verify_claims(
             signed_claims=signed_claims, publisher_service=pretend.stub()
         )

tests/unit/oidc/models/test_gitlab.py

@@ -195,12 +195,6 @@ def test_gitlab_publisher_computed_properties(self):
         }
 
     def test_gitlab_publisher_unaccounted_claims(self, monkeypatch):
-        publisher = gitlab.GitLabPublisher(
-            project="fakerepo",
-            namespace="fakeowner",
-            workflow_filepath="subfolder/fakeworkflow.yml",
-        )
-
         scope = pretend.stub()
         sentry_sdk = pretend.stub(
             capture_message=pretend.call_recorder(lambda s: None),
@@ -219,11 +213,8 @@ def test_gitlab_publisher_unaccounted_claims(self, monkeypatch):
         }
         signed_claims["fake-claim"] = "fake"
         signed_claims["another-fake-claim"] = "also-fake"
-        with pytest.raises(errors.InvalidPublisherError) as e:
-            publisher.verify_claims(
-                signed_claims=signed_claims, publisher_service=pretend.stub()
-            )
-        assert str(e.value) == "Check failed for required claim 'sub'"
+
+        gitlab.GitLabPublisher.check_claims_existence(signed_claims)
         assert sentry_sdk.capture_message.calls == [
             pretend.call(
                 "JWT for GitLabPublisher has unaccounted claims: "
@@ -232,7 +223,11 @@ def test_gitlab_publisher_unaccounted_claims(self, monkeypatch):
         ]
         assert scope.fingerprint == ["another-fake-claim", "fake-claim"]
 
-    @pytest.mark.parametrize("missing", ["sub", "ref_path"])
+    @pytest.mark.parametrize(
+        "missing",
+        gitlab.GitLabPublisher.__required_verifiable_claims__.keys()
+        | gitlab.GitLabPublisher.__required_unverifiable_claims__,
+    )
     def test_gitlab_publisher_missing_claims(self, monkeypatch, missing):
         publisher = gitlab.GitLabPublisher(
             project="fakerepo",
@@ -260,9 +255,7 @@ def test_gitlab_publisher_missing_claims(self, monkeypatch, missing):
         assert missing not in signed_claims
         assert publisher.__required_verifiable_claims__
         with pytest.raises(errors.InvalidPublisherError) as e:
-            publisher.verify_claims(
-                signed_claims=signed_claims, publisher_service=pretend.stub()
-            )
+            gitlab.GitLabPublisher.check_claims_existence(signed_claims)
         assert str(e.value) == f"Missing claim {missing!r}"
         assert sentry_sdk.capture_message.calls == [
             pretend.call(f"JWT for GitLabPublisher is missing claim: {missing}")

tests/unit/oidc/models/test_google.py

@@ -72,11 +72,6 @@ def test_google_publisher_all_known_claims(self):
         }
 
     def test_google_publisher_unaccounted_claims(self, monkeypatch):
-        publisher = google.GooglePublisher(
-            sub="fakesubject",
-            email="[email protected]",
-        )
-
         scope = pretend.stub()
         sentry_sdk = pretend.stub(
             capture_message=pretend.call_recorder(lambda s: None),
@@ -95,11 +90,8 @@ def test_google_publisher_unaccounted_claims(self, monkeypatch):
         }
         signed_claims["fake-claim"] = "fake"
         signed_claims["another-fake-claim"] = "also-fake"
-        with pytest.raises(errors.InvalidPublisherError) as e:
-            publisher.verify_claims(
-                signed_claims=signed_claims, publisher_service=pretend.stub()
-            )
-        assert str(e.value) == "Check failed for required claim 'email'"
+
+        google.GooglePublisher.check_claims_existence(signed_claims)
         assert sentry_sdk.capture_message.calls == [
             pretend.call(
                 "JWT for GooglePublisher has unaccounted claims: "
@@ -108,12 +100,12 @@ def test_google_publisher_unaccounted_claims(self, monkeypatch):
         ]
         assert scope.fingerprint == ["another-fake-claim", "fake-claim"]
 
-    def test_google_publisher_missing_claims(self, monkeypatch):
-        publisher = google.GooglePublisher(
-            sub="fakesubject",
-            email="[email protected]",
-        )
-
+    @pytest.mark.parametrize(
+        "missing",
+        google.GooglePublisher.__required_verifiable_claims__.keys()
+        | google.GooglePublisher.__required_unverifiable_claims__,
+    )
+    def test_google_publisher_missing_claims(self, monkeypatch, missing):
         scope = pretend.stub()
         sentry_sdk = pretend.stub(
             capture_message=pretend.call_recorder(lambda s: None),
@@ -130,18 +122,17 @@ def test_google_publisher_missing_claims(self, monkeypatch):
             for claim_name in google.GooglePublisher.all_known_claims()
         }
         # Pop the first signed claim, so that it's the first one to fail.
-        signed_claims.pop("email")
-        assert "email" not in signed_claims
-        assert publisher.__required_verifiable_claims__
+        signed_claims.pop(missing)
+        assert missing not in signed_claims
+        assert google.GooglePublisher.__required_verifiable_claims__
         with pytest.raises(errors.InvalidPublisherError) as e:
-            publisher.verify_claims(
-                signed_claims=signed_claims, publisher_service=pretend.stub()
-            )
-        assert str(e.value) == "Missing claim 'email'"
+            google.GooglePublisher.check_claims_existence(signed_claims)
+
+        assert str(e.value) == f"Missing claim '{missing}'"
         assert sentry_sdk.capture_message.calls == [
-            pretend.call("JWT for GooglePublisher is missing claim: email")
+            pretend.call(f"JWT for GooglePublisher is missing claim: {missing}")
         ]
-        assert scope.fingerprint == ["email"]
+        assert scope.fingerprint == [missing]
 
     @pytest.mark.parametrize(
         ("email_verified", "valid"),

tests/unit/oidc/test_utils.py

@@ -24,6 +24,13 @@
     GooglePublisherFactory,
 )
 from warehouse.oidc import errors, utils
+from warehouse.oidc.models import (
+    ActiveStatePublisher,
+    GitHubPublisher,
+    GitLabPublisher,
+    GooglePublisher,
+)
+from warehouse.oidc.utils import OIDC_PUBLISHER_CLASSES
 from warehouse.utils.security_policy import principals_for
 
 
@@ -34,6 +41,35 @@ def test_find_publisher_by_issuer_bad_issuer_url():
         )
 
 
+@pytest.mark.parametrize(
+    ("issuer_url", "publisher_cls_dict"), OIDC_PUBLISHER_CLASSES.items()
+)
+def test_find_publisher_by_issuer_checks_claims_existence(
+    monkeypatch, issuer_url, publisher_cls_dict
+):
+    publisher_cls = pretend.stub(
+        check_claims_existence=pretend.call_recorder(lambda x: None),
+        lookup_by_claims=pretend.call_recorder(lambda x, y: None),
+    )
+    monkeypatch.setattr(
+        utils,
+        "OIDC_PUBLISHER_CLASSES",
+        {issuer_url: {False: publisher_cls, True: publisher_cls}},
+    )
+
+    signed_claims = {
+        claim_name: "fake"
+        for claim_name in publisher_cls_dict[False].all_known_claims()
+    }
+    session = pretend.stub()
+    utils.find_publisher_by_issuer(session, issuer_url, signed_claims)
+
+    assert publisher_cls.check_claims_existence.calls == [pretend.call(signed_claims)]
+    assert publisher_cls.lookup_by_claims.calls == [
+        pretend.call(session, signed_claims)
+    ]
+
+
 @pytest.mark.parametrize(
     ("environment", "expected_id"),
     [
@@ -62,10 +98,15 @@ def test_find_publisher_by_issuer_github(db_request, environment, expected_id):
     )
 
     signed_claims = {
-        "repository": "foo/bar",
-        "job_workflow_ref": "foo/bar/.github/workflows/ci.yml@refs/heads/main",
-        "repository_owner_id": "1234",
+        claim_name: "fake" for claim_name in GitHubPublisher.all_known_claims()
     }
+    signed_claims.update(
+        {
+            "repository": "foo/bar",
+            "job_workflow_ref": "foo/bar/.github/workflows/ci.yml@refs/heads/main",
+            "repository_owner_id": "1234",
+        }
+    )
     if environment:
         signed_claims["environment"] = environment
 
@@ -104,9 +145,15 @@ def test_find_publisher_by_issuer_gitlab(db_request, environment, expected_id):
     )
 
     signed_claims = {
-        "project_path": "foo/bar",
-        "ci_config_ref_uri": "gitlab.com/foo/bar//workflows/ci.yml@refs/heads/main",
+        claim_name: "fake" for claim_name in GitLabPublisher.all_known_claims()
     }
+
+    signed_claims.update(
+        {
+            "project_path": "foo/bar",
+            "ci_config_ref_uri": "gitlab.com/foo/bar//workflows/ci.yml@refs/heads/main",
+        }
+    )
     if environment:
         signed_claims["environment"] = environment
 
@@ -140,10 +187,16 @@ def test_find_publisher_by_issuer_google(db_request, sub, expected_id):
     )
 
     signed_claims = {
-        "email": "[email protected]",
-        "sub": sub,
+        claim_name: "fake" for claim_name in GooglePublisher.all_known_claims()
     }
 
+    signed_claims.update(
+        {
+            "email": "[email protected]",
+            "sub": sub,
+        }
+    )
+
     assert (
         utils.find_publisher_by_issuer(
             db_request.db,
@@ -227,13 +280,19 @@ def test_find_publisher_by_issuer_activestate(
     )
 
     signed_claims = {
-        "sub": sub,
-        "organization": organization,
-        "project": project,
-        "actor_id": actor_id,
-        "actor": actor,
+        claim_name: "fake" for claim_name in ActiveStatePublisher.all_known_claims()
     }
 
+    signed_claims.update(
+        {
+            "sub": sub,
+            "organization": organization,
+            "project": project,
+            "actor_id": actor_id,
+            "actor": actor,
+        }
+    )
+
     assert (
         utils.find_publisher_by_issuer(
             db_request.db,