[oidc] Handle verification of optional claims when not present (pypi#…

…13419)

* Handle verification of optional OIDC claims

* Rename to __required_verifiable_claims__

* Update wording of comment

tests/unit/oidc/test_models.py

@@ -45,6 +45,7 @@ def test_github_publisher_all_known_claims(self):
             "repository_owner",
             "repository_owner_id",
             "job_workflow_ref",
+            # optional verifiable claims
             "environment",
             # preverified claims
             "iss",
@@ -84,7 +85,7 @@ def test_github_publisher_computed_properties(self):
             environment="fakeenv",
         )
 
-        for claim_name in publisher.__verifiable_claims__.keys():
+        for claim_name in publisher.__required_verifiable_claims__.keys():
             assert getattr(publisher, claim_name) is not None
 
         assert str(publisher) == "fakeworkflow.yml @ fakeowner/fakerepo"
@@ -136,14 +137,40 @@ def test_github_publisher_missing_claims(self, monkeypatch):
         # Pop the first signed claim, so that it's the first one to fail.
         signed_claims.pop("sub")
         assert "sub" not in signed_claims
-        assert publisher.__verifiable_claims__
+        assert publisher.__required_verifiable_claims__
         assert not publisher.verify_claims(signed_claims=signed_claims)
         assert sentry_sdk.capture_message.calls == [
             pretend.call("JWT for GitHubPublisher is missing claim: sub")
         ]
 
+    def test_github_publisher_missing_optional_claims(self, monkeypatch):
+        publisher = models.GitHubPublisher(
+            repository_name="fakerepo",
+            repository_owner="fakeowner",
+            repository_owner_id="fakeid",
+            workflow_filename="fakeworkflow.yml",
+            environment="some-environment",  # The optional claim that should be present
+        )
+
+        sentry_sdk = pretend.stub(capture_message=pretend.call_recorder(lambda s: None))
+        monkeypatch.setattr(models, "sentry_sdk", sentry_sdk)
+
+        signed_claims = {
+            claim_name: getattr(publisher, claim_name)
+            for claim_name in models.GitHubPublisher.__required_verifiable_claims__
+        }
+        signed_claims["ref"] = "ref"
+        signed_claims["job_workflow_ref"] = publisher.job_workflow_ref + "@ref"
+        assert publisher.__required_verifiable_claims__
+        assert not publisher.verify_claims(signed_claims=signed_claims)
+        assert sentry_sdk.capture_message.calls == []
+
     @pytest.mark.parametrize("environment", [None, "some-environment"])
-    def test_github_publisher_verifies(self, monkeypatch, environment):
+    @pytest.mark.parametrize(
+        "missing_claims",
+        [set(), models.GitHubPublisher.__optional_verifiable_claims__.keys()],
+    )
+    def test_github_publisher_verifies(self, monkeypatch, environment, missing_claims):
         publisher = models.GitHubPublisher(
             repository_name="fakerepo",
             repository_owner="fakeowner",
@@ -154,16 +181,29 @@ def test_github_publisher_verifies(self, monkeypatch, environment):
 
         noop_check = pretend.call_recorder(lambda gt, sc, ac: True)
         verifiable_claims = {
-            claim_name: noop_check for claim_name in publisher.__verifiable_claims__
+            claim_name: noop_check
+            for claim_name in publisher.__required_verifiable_claims__
         }
-        monkeypatch.setattr(publisher, "__verifiable_claims__", verifiable_claims)
+        monkeypatch.setattr(
+            publisher, "__required_verifiable_claims__", verifiable_claims
+        )
+        optional_verifiable_claims = {
+            claim_name: noop_check
+            for claim_name in publisher.__optional_verifiable_claims__
+        }
+        monkeypatch.setattr(
+            publisher, "__optional_verifiable_claims__", optional_verifiable_claims
+        )
 
         signed_claims = {
             claim_name: "fake"
             for claim_name in models.GitHubPublisher.all_known_claims()
+            if claim_name not in missing_claims
         }
         assert publisher.verify_claims(signed_claims=signed_claims)
-        assert len(noop_check.calls) == len(verifiable_claims)
+        assert len(noop_check.calls) == len(verifiable_claims) + len(
+            optional_verifiable_claims
+        )
 
     @pytest.mark.parametrize(
         ("claim", "ref", "valid"),
@@ -221,7 +261,9 @@ def test_github_publisher_job_workflow_ref(self, claim, ref, valid):
             workflow_filename="baz.yml",
         )
 
-        check = models.GitHubPublisher.__verifiable_claims__["job_workflow_ref"]
+        check = models.GitHubPublisher.__required_verifiable_claims__[
+            "job_workflow_ref"
+        ]
         assert check(publisher.job_workflow_ref, claim, {"ref": ref}) is valid
 
     @pytest.mark.parametrize(
@@ -235,7 +277,7 @@ def test_github_publisher_job_workflow_ref(self, claim, ref, valid):
         ],
     )
     def test_github_publisher_sub_claim(self, truth, claim, valid):
-        check = models.GitHubPublisher.__verifiable_claims__["sub"]
+        check = models.GitHubPublisher.__required_verifiable_claims__["sub"]
         assert check(truth, claim, pretend.stub()) is valid
 
     @pytest.mark.parametrize(
@@ -250,7 +292,7 @@ def test_github_publisher_sub_claim(self, truth, claim, valid):
         ],
     )
     def test_github_publisher_environment_claim(self, truth, claim, valid):
-        check = models.GitHubPublisher.__verifiable_claims__["environment"]
+        check = models.GitHubPublisher.__optional_verifiable_claims__["environment"]
         assert check(truth, claim, pretend.stub()) is valid
 
 

warehouse/oidc/models.py

@@ -125,7 +125,12 @@ class OIDCPublisherMixin:
 
     # A map of claim names to "check" functions, each of which
     # has the signature `check(ground-truth, signed-claim, all-signed-claims) -> bool`.
-    __verifiable_claims__: dict[
+    __required_verifiable_claims__: dict[
+        str, Callable[[Any, Any, dict[str, Any]], bool]
+    ] = dict()
+
+    # Simlar to __verificable_claims__, but these claims are optional
+    __optional_verifiable_claims__: dict[
         str, Callable[[Any, Any, dict[str, Any]], bool]
     ] = dict()
 
@@ -150,7 +155,8 @@ def all_known_claims(cls):
         Returns all claims "known" to this publisher.
         """
         return (
-            cls.__verifiable_claims__.keys()
+            cls.__required_verifiable_claims__.keys()
+            | cls.__optional_verifiable_claims__.keys()
             | cls.__preverified_claims__
             | cls.__unchecked_claims__
         )
@@ -164,7 +170,7 @@ def verify_claims(self, signed_claims: SignedClaims):
 
         # Defensive programming: treat the absence of any claims to verify
         # as a failure rather than trivially valid.
-        if not self.__verifiable_claims__:
+        if not self.__required_verifiable_claims__:
             return False
 
         # All claims should be accounted for.
@@ -178,10 +184,10 @@ def verify_claims(self, signed_claims: SignedClaims):
             )
 
         # Finally, perform the actual claim verification.
-        for claim_name, check in self.__verifiable_claims__.items():
-            # All verifiable claims are mandatory. The absence of a missing
-            # claim *is* an error, since it indicates a breaking change in the
-            # JWT's payload.
+        for claim_name, check in self.__required_verifiable_claims__.items():
+            # All required claims are mandatory. The absence of a missing
+            # claim *is* an error with the JWT, since it indicates a breaking
+            # change in the JWT's payload.
             signed_claim = signed_claims.get(claim_name)
             if signed_claim is None:
                 sentry_sdk.capture_message(
@@ -192,6 +198,17 @@ def verify_claims(self, signed_claims: SignedClaims):
             if not check(getattr(self, claim_name), signed_claim, signed_claims):
                 return False
 
+        # Check optional verifiable claims
+        for claim_name, check in self.__optional_verifiable_claims__.items():
+            # All optional claims are optional. The absence of a missing
+            # claim is *NOT* an error with the JWT, however we should still
+            # verify this against the check, because the claim might be
+            # required for a given publisher.
+            signed_claim = signed_claims.get(claim_name)
+
+            if not check(getattr(self, claim_name), signed_claim, signed_claims):
+                return False
+
         return True
 
     @property
@@ -262,12 +279,15 @@ class GitHubPublisherMixin:
     workflow_filename = Column(String, nullable=False)
     environment = Column(String, nullable=True)
 
-    __verifiable_claims__ = {
+    __required_verifiable_claims__ = {
         "sub": _check_sub,
         "repository": _check_claim_binary(str.__eq__),
         "repository_owner": _check_claim_binary(str.__eq__),
         "repository_owner_id": _check_claim_binary(str.__eq__),
         "job_workflow_ref": _check_job_workflow_ref,
+    }
+
+    __optional_verifiable_claims__ = {
         "environment": _check_environment,
     }