Reusable callable type for claim checking code

Author: th3coop

tests/common/db/oidc.py

@@ -81,11 +81,8 @@ class Meta:
     organization_url_name = factory.Faker("pystr", max_chars=12)
     project_id = factory.Faker("uuid4")
     activestate_project_name = factory.Faker("pystr", max_chars=12)
-    project_path = f"{organization_url_name}/{activestate_project_name}"
     user_id = factory.Faker("uuid4")
     branch_id = factory.Faker("uuid4")
-    project_visibility = factory.Faker("random_element", elements=[True, False])
-    sub = factory.Faker("pystr", max_chars=12)
 
 
 class PendingActiveStatePublisherFactory(WarehouseFactory):
@@ -99,7 +96,5 @@ class Meta:
     organization_url_name = factory.Faker("pystr", max_chars=12)
     project_id = factory.Faker("uuid4")
     activestate_project_name = factory.Faker("pystr", max_chars=12)
-    project_path = f"{organization_url_name}/{activestate_project_name}"
     user_id = factory.Faker("uuid4")
-    project_visibility = factory.Faker("random_element", elements=[True, False])
     added_by = factory.SubFactory(UserFactory)

tests/conftest.py

@@ -351,18 +351,6 @@ def oidc_service(db_session):
     )
 
 
-# @pytest.fixture
-# def activestate_oidc_service(db_session):
-#     return oidc_services.NullOIDCPublisherService(
-#         db_session,
-#         pretend.stub(),
-#         ACTIVESTATE_OIDC_ISSUER_URL,
-#         pretend.stub(),
-#         pretend.stub(),
-#         pretend.stub(),
-#     )
-
-
 @pytest.fixture
 def macaroon_service(db_session):
     return macaroon_services.DatabaseMacaroonService(db_session)

tests/unit/oidc/models/test_activestate.py

@@ -29,6 +29,13 @@
 PROJECT_ID = "00000000-0000-1000-8000-000000000001"
 USER_ID = "00000000-0000-1000-8000-000000000002"
 BRANCH_ID = "00000000-0000-1000-8000-000000000003"
+# This follows the format of the subject that ActiveState sends us.  We don't
+# validate the format when verifying the JWT. That should happen when the
+# Publisher is configured.  We just need to make sure that the subject matches
+#
+# Technically, the branch should only be present if a branch was provided in the JWT
+# claims
+SUBJECT = "org:fake_org_id:project:fake_project_id:branch_id:fake_branch_id"
 
 
 def test_lookup_strategies():
@@ -40,65 +47,63 @@ def test_lookup_strategies():
 
 
 def new_signed_claims(
-    sub: str = "fake:fake:fake:fake",
+    sub: str = SUBJECT,
     organization_id: str = ORG_ID,
     org_url_name: str = "fakeorg",
     project_id: str = PROJECT_ID,
     project_name: str = "fakeproject",
     project_path: str = "fakeorg/fakeproject",
-    user_id: str = "00000000-0000-1000-8000-000000000002",
+    user_id: str = USER_ID,
     project_visibility: str = "public",
     branch_id: str | None = None,
 ) -> SignedClaims:
     project_name = "fakeproject"
     org_url_name = "fakeorg"
-    claims: SignedClaims = {
-        "sub": sub,
-        "organization_id": organization_id,
-        "organization_url_name": org_url_name,
-        "project_id": project_id,
-        "project_name": project_name,
-        "project_path": project_path,
-        "user_id": user_id,
-        "project_visibility": project_visibility,
-    }
+    claims = SignedClaims(
+        {
+            "sub": sub,
+            "organization_id": organization_id,
+            "organization_url_name": org_url_name,
+            "project_id": project_id,
+            "project_name": project_name,
+            "project_path": project_path,
+            "user_id": user_id,
+            "project_visibility": project_visibility,
+        }
+    )
     if branch_id:
         claims["branch_id"] = branch_id
     return claims
 
 
 class TestActiveStatePublisher:
     def test_publisher_name(self):
-        publisher = ActiveStatePublisher(
-            organization_id=ORG_ID,
-            project_id=PROJECT_ID,
-            user_id=USER_ID,
-            project_visibility="public",
-        )
+        publisher = ActiveStatePublisher()
 
         assert publisher.publisher_name == "ActiveState"
 
     def test_publisher_url(self):
+        org_name = "fakeorg"
+        project_name = "fakeproject"
         publisher = ActiveStatePublisher(
-            organization_id=ORG_ID,
-            project_id=PROJECT_ID,
-            user_id=USER_ID,
-            project_visibility="public",
+            organization_url_name=org_name, activestate_project_name=project_name
         )
 
-        assert publisher.publisher_url == "https://platform.activestate.com/oauth/oidc"
+        assert (
+            publisher.publisher_url()
+            == f"https://platform.activestate.com/{org_name}/{project_name}"
+        )
 
-    def test_stringifies_as_email(self):
+    def test_stringifies_as_project_url(self):
+        org_name = "fakeorg"
+        project_name = "fakeproject"
         publisher = ActiveStatePublisher(
-            organization_id=ORG_ID,
-            project_id=PROJECT_ID,
-            user_id=USER_ID,
-            project_visibility="public",
+            organization_url_name=org_name, activestate_project_name=project_name
         )
 
         assert (
             str(publisher)
-            == f"https://platform.activestate.com/{publisher.project_path}"
+            == f"https://platform.activestate.com/{org_name}/{project_name}"
         )
 
     def test_activestate_publisher_all_known_claims(self):
@@ -107,7 +112,6 @@ def test_activestate_publisher_all_known_claims(self):
             "organization_id",
             "project_id",
             "user_id",
-            "project_visibility",
             "sub",
             # optional verifiable claims
             "branch_id",
@@ -118,18 +122,15 @@ def test_activestate_publisher_all_known_claims(self):
             "exp",
             "aud",
             # unchecked claims
+            "project_visibility",
             "project_name",
             "project_path",
             "organization_url_name",
         }
 
     def test_activestate_publisher_unaccounted_claims(self, monkeypatch):
         publisher = ActiveStatePublisher(
-            sub="fake:fake:fake:fake:fake:fake",
-            organization_id=ORG_ID,
-            project_id=PROJECT_ID,
-            user_id=USER_ID,
-            project_visibility="public",
+            sub=SUBJECT,
         )
 
         scope = pretend.stub()
@@ -143,7 +144,6 @@ def test_activestate_publisher_unaccounted_claims(self, monkeypatch):
         )
         monkeypatch.setattr(_core, "sentry_sdk", sentry_sdk)
 
-        # We don't care if these actually verify, only that they're present.
         signed_claims = {
             claim_name: "fake" for claim_name in ActiveStatePublisher.all_known_claims()
         }
@@ -158,13 +158,27 @@ def test_activestate_publisher_unaccounted_claims(self, monkeypatch):
         ]
         assert scope.fingerprint == ["another-fake-claim", "fake-claim"]
 
-    def test_activestate_publisher_missing_claims(self, monkeypatch):
+    @pytest.mark.parametrize(
+        ("claim_to_drop", "valid"),
+        [
+            ("organization_id", False),
+            ("project_id", False),
+            ("user_id", False),
+            ("branch_id", True),
+            ("organization_url_name", True),
+            ("project_visibility", True),
+            ("project_name", True),
+            ("project_path", True),
+        ],
+    )
+    def test_activestate_publisher_missing_claims(
+        self, monkeypatch, claim_to_drop: str, valid: bool
+    ):
         publisher = ActiveStatePublisher(
-            sub="fake:fake:fake:fake:fake:fake",
-            organization_id="fake",
-            project_id="fake",
-            user_id="fake",
-            project_visibility="fake",
+            sub=SUBJECT,
+            organization_id=ORG_ID,
+            project_id=PROJECT_ID,
+            user_id=USER_ID,
         )
 
         scope = pretend.stub()
@@ -178,21 +192,21 @@ def test_activestate_publisher_missing_claims(self, monkeypatch):
         )
         monkeypatch.setattr(_core, "sentry_sdk", sentry_sdk)
 
-        signed_claims = {
-            claim_name: "fake" for claim_name in ActiveStatePublisher.all_known_claims()
-        }
-        signed_claims["sub"] = "fake:fake:fake:fake:fake:fake"
-        # Pop the first signed claim, so that it's the first one to fail.
-        signed_claims.pop("organization_id")
-        assert "organization_id" not in signed_claims
+        signed_claims = new_signed_claims()
+        # I don't set branch_id in the claims by default, so no need to remove it
+        if claim_to_drop != "branch_id":
+            # Remove a required claim to test that the code detects it's missing
+            signed_claims.pop(claim_to_drop)
+        assert claim_to_drop not in signed_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 ActiveStatePublisher is missing claim: organization_id"
-            )
-        ]
-        assert scope.fingerprint == ["organization_id"]
+        assert publisher.verify_claims(signed_claims=signed_claims) is valid
+        if not valid:
+            assert sentry_sdk.capture_message.calls == [
+                pretend.call(
+                    "JWT for ActiveStatePublisher is missing claim: " + claim_to_drop
+                )
+            ]
+            assert scope.fingerprint == [claim_to_drop]
 
     @pytest.mark.parametrize(
         ("expect", "actual", "valid"),
@@ -205,11 +219,10 @@ def test_activestate_publisher_org_id_verified(
         self, expect: str, actual: str, valid: bool
     ):
         publisher = ActiveStatePublisher(
-            sub="fake:fake:fake:fake",
+            sub=SUBJECT,
             organization_id=actual,
             project_id=PROJECT_ID,
             user_id=USER_ID,
-            project_visibility="public",
         )
 
         signed_claims = new_signed_claims(organization_id=expect)
@@ -218,66 +231,64 @@ def test_activestate_publisher_org_id_verified(
     @pytest.mark.parametrize(
         ("expect", "actual", "valid"),
         [
-            (PROJECT_ID, PROJECT_ID, True),
-            (PROJECT_ID, ORG_ID, False),
+            (BRANCH_ID, BRANCH_ID, True),
+            (BRANCH_ID, PROJECT_ID, False),
+            (None, None, True),
+            (BRANCH_ID, None, False),
         ],
     )
-    def test_activestate_publisher_project_id_verified(
+    def test_activestate_publisher_branch_id_verified(
         self, expect: str, actual: str, valid: bool
     ):
         publisher = ActiveStatePublisher(
-            sub="fake:fake:fake:fake",
+            sub=SUBJECT,
             organization_id=ORG_ID,
-            project_id=actual,
+            project_id=PROJECT_ID,
             user_id=USER_ID,
-            project_visibility="public",
+            branch_id=expect,
         )
 
-        signed_claims = new_signed_claims(project_id=expect)
+        signed_claims = new_signed_claims(branch_id=actual)
         assert publisher.verify_claims(signed_claims=signed_claims) is valid
 
     @pytest.mark.parametrize(
         ("expect", "actual", "valid"),
         [
-            (USER_ID, USER_ID, True),
-            (USER_ID, ORG_ID, False),
+            (PROJECT_ID, PROJECT_ID, True),
+            (PROJECT_ID, ORG_ID, False),
         ],
     )
-    def test_activestate_publisher_user_id_verified(
+    def test_activestate_publisher_project_id_verified(
         self, expect: str, actual: str, valid: bool
     ):
         publisher = ActiveStatePublisher(
-            sub="fake:fake:fake:fake",
+            sub=SUBJECT,
             organization_id=ORG_ID,
-            project_id=PROJECT_ID,
-            user_id=actual,
-            project_visibility="public",
+            project_id=actual,
+            user_id=USER_ID,
         )
 
-        signed_claims = new_signed_claims(user_id=expect)
+        signed_claims = new_signed_claims(project_id=expect)
         assert publisher.verify_claims(signed_claims=signed_claims) is valid
 
     @pytest.mark.parametrize(
         ("expect", "actual", "valid"),
         [
-            ("public", "public", True),
-            ("private", "private", True),
-            ("public", "private", False),
-            ("private", "public", False),
+            (USER_ID, USER_ID, True),
+            (USER_ID, ORG_ID, False),
         ],
     )
-    def test_activestate_publisher_project_vis_verified(
+    def test_activestate_publisher_user_id_verified(
         self, expect: str, actual: str, valid: bool
     ):
         publisher = ActiveStatePublisher(
-            sub="fake:fake:fake:fake",
+            sub=SUBJECT,
             organization_id=ORG_ID,
             project_id=PROJECT_ID,
-            user_id=USER_ID,
-            project_visibility=actual,
+            user_id=actual,
         )
 
-        signed_claims = new_signed_claims(project_visibility=expect)
+        signed_claims = new_signed_claims(user_id=expect)
         assert publisher.verify_claims(signed_claims=signed_claims) is valid
 
     @pytest.mark.parametrize(
@@ -317,7 +328,6 @@ def test_activestate_publisher_project_vis_verified(
     )
     def test_activestate_publisher_sub(self, expected: str, actual: str, valid: bool):
         check = ActiveStatePublisher.__required_verifiable_claims__["sub"]
-        # claims: SignedClaims = new_signed_claims(sub=actual)
         assert check(expected, actual, pretend.stub()) is valid
 
 
@@ -337,8 +347,6 @@ def test_reify_does_not_exist_yet(self, db_request):
         )
         publisher = pending_publisher.reify(db_request.db)
 
-        # If an OIDC publisher for this pending publisher does not already exist,
-        # a new one is created and the pending publisher is marked for deletion.
         assert isinstance(publisher, ActiveStatePublisher)
         assert pending_publisher in db_request.db.deleted
         assert publisher.organization_id == pending_publisher.organization_id
@@ -350,12 +358,9 @@ def test_reify_already_exists(self, db_request):
             organization_id=existing_publisher.organization_id,
             project_id=existing_publisher.project_id,
             branch_id=existing_publisher.branch_id,
-            project_visibility=existing_publisher.project_visibility,
             sub=existing_publisher.sub,
         )
         publisher = pending_publisher.reify(db_request.db)
 
-        # If an OIDC publisher for this pending publisher already exists,
-        # it is returned and the pending publisher is marked for deletion.
         assert existing_publisher == publisher
         assert pending_publisher in db_request.db.deleted