warehouse: OIDC beta flag/controls

tests/unit/accounts/test_models.py

@@ -162,3 +162,20 @@ def test_acl(self, db_session):
             ("Allow", "group:admins", "admin"),
             ("Allow", "group:moderators", "moderator"),
         ]
+
+    @pytest.mark.parametrize(
+        ("is_superuser", "has_oidc_beta_access", "in_oidc_beta"),
+        [
+            (False, False, False),
+            (True, False, True),
+            (True, True, True),
+            (False, True, True),
+        ],
+    )
+    def test_in_oidc_beta(
+        self, db_session, is_superuser, has_oidc_beta_access, in_oidc_beta
+    ):
+        user = DBUserFactory.create(
+            is_superuser=is_superuser, has_oidc_beta_access=has_oidc_beta_access
+        )
+        assert user.in_oidc_beta == in_oidc_beta

tests/unit/accounts/test_views.py

@@ -25,6 +25,7 @@
     HTTPSeeOther,
     HTTPTooManyRequests,
 )
+from pyramid.response import Response
 from sqlalchemy.exc import NoResultFound
 from webauthn.authentication.verify_authentication_response import (
     VerifiedAuthentication,
@@ -2981,6 +2982,9 @@ def find_service(iface, name=None, context=None):
     def test_manage_publishing(self, monkeypatch):
         metrics = pretend.stub()
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -3034,8 +3038,24 @@ def test_manage_publishing_oidc_disabled(self):
         with pytest.raises(HTTPNotFound):
             view.manage_publishing()
 
+    def test_manage_publishing_not_in_beta(self):
+        request = pretend.stub(
+            user=pretend.stub(in_oidc_beta=False),
+            registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
+            find_service=lambda *a, **kw: None,
+        )
+
+        view = views.ManageAccountPublishingViews(request)
+        resp = view.manage_publishing()
+
+        assert isinstance(resp, Response)
+        assert resp.status_code == 403
+
     def test_manage_publishing_admin_disabled(self, monkeypatch):
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -3098,8 +3118,24 @@ def test_add_pending_github_oidc_publisher_oidc_disabled(self):
         with pytest.raises(HTTPNotFound):
             view.add_pending_github_oidc_publisher()
 
+    def test_add_pending_github_oidc_publisher_not_in_beta(self):
+        request = pretend.stub(
+            user=pretend.stub(in_oidc_beta=False),
+            registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
+            find_service=lambda *a, **kw: None,
+        )
+
+        view = views.ManageAccountPublishingViews(request)
+        resp = view.add_pending_github_oidc_publisher()
+
+        assert isinstance(resp, Response)
+        assert resp.status_code == 403
+
     def test_add_pending_github_oidc_publisher_admin_disabled(self, monkeypatch):
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -3164,7 +3200,7 @@ def test_add_pending_github_oidc_publisher_user_cannot_register(self, monkeypatc
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             POST=pretend.stub(),
-            user=pretend.stub(has_primary_verified_email=False),
+            user=pretend.stub(has_primary_verified_email=False, in_oidc_beta=True),
             _=lambda s: s,
         )
 
@@ -3228,6 +3264,7 @@ def test_add_pending_github_oidc_publisher_too_many_already(self, monkeypatch):
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             POST=pretend.stub(),
             user=pretend.stub(
+                in_oidc_beta=True,
                 has_primary_verified_email=True,
                 pending_oidc_publishers=[
                     pretend.stub(),
@@ -3297,6 +3334,7 @@ def test_add_pending_github_oidc_publisher_ratelimited(self, monkeypatch):
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             POST=pretend.stub(),
             user=pretend.stub(
+                in_oidc_beta=True,
                 has_primary_verified_email=True,
                 pending_oidc_publishers=[],
             ),
@@ -3354,6 +3392,7 @@ def test_add_pending_github_oidc_publisher_invalid_form(self, monkeypatch):
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             POST=pretend.stub(),
             user=pretend.stub(
+                in_oidc_beta=True,
                 has_primary_verified_email=True,
                 pending_oidc_publishers=[],
             ),
@@ -3414,6 +3453,7 @@ def test_add_pending_github_oidc_publisher_already_exists(self, monkeypatch):
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             POST=pretend.stub(),
             user=pretend.stub(
+                in_oidc_beta=True,
                 has_primary_verified_email=True,
                 pending_oidc_publishers=[],
             ),
@@ -3490,6 +3530,7 @@ def test_add_pending_github_oidc_publisher(self, monkeypatch):
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
             POST=pretend.stub(),
             user=pretend.stub(
+                in_oidc_beta=True,
                 has_primary_verified_email=True,
                 pending_oidc_publishers=[],
                 record_event=pretend.call_recorder(lambda **kw: None),
@@ -3608,8 +3649,24 @@ def test_delete_pending_oidc_publisher_oidc_disabled(self):
         with pytest.raises(HTTPNotFound):
             view.delete_pending_oidc_publisher()
 
+    def test_delete_pending_oidc_publisher_not_in_beta(self):
+        request = pretend.stub(
+            user=pretend.stub(in_oidc_beta=False),
+            registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
+            find_service=lambda *a, **kw: None,
+        )
+
+        view = views.ManageAccountPublishingViews(request)
+        resp = view.delete_pending_oidc_publisher()
+
+        assert isinstance(resp, Response)
+        assert resp.status_code == 403
+
     def test_delete_pending_oidc_publisher_admin_disabled(self, monkeypatch):
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -3664,6 +3721,9 @@ def test_delete_pending_oidc_publisher_admin_disabled(self, monkeypatch):
     def test_delete_pending_oidc_publisher_invalid_form(self, monkeypatch):
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: metrics,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
@@ -3698,6 +3758,9 @@ def test_delete_pending_oidc_publisher_invalid_form(self, monkeypatch):
     def test_delete_pending_oidc_publisher_not_found(self, monkeypatch):
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: metrics,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
@@ -3758,7 +3821,9 @@ def test_delete_pending_oidc_publisher(self, monkeypatch):
                 delete=pretend.call_recorder(lambda m: None),
             ),
             session=pretend.stub(flash=pretend.call_recorder(lambda *a, **kw: None)),
-            user=pretend.stub(record_event=pretend.call_recorder(lambda **kw: None)),
+            user=pretend.stub(
+                in_oidc_beta=True, record_event=pretend.call_recorder(lambda **kw: None)
+            ),
             remote_addr="0.0.0.0",
             path="some-path",
         )

tests/unit/manage/test_views.py

@@ -9538,8 +9538,11 @@ def find_service(iface, name=None, context=None):
             view._check_ratelimits()
 
     def test_manage_project_oidc_publishers(self, monkeypatch):
-        project = pretend.stub()
+        project = pretend.stub(oidc_publishers=[])
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -9572,8 +9575,11 @@ def test_manage_project_oidc_publishers(self, monkeypatch):
         ]
 
     def test_manage_project_oidc_publishers_admin_disabled(self, monkeypatch):
-        project = pretend.stub()
+        project = pretend.stub(oidc_publishers=[])
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -9628,6 +9634,21 @@ def test_manage_project_oidc_publishers_oidc_not_enabled(self):
         with pytest.raises(HTTPNotFound):
             view.manage_project_oidc_publishers()
 
+    def test_manage_project_oidc_publishers_not_in_beta(self):
+        project = pretend.stub(oidc_publishers=[])
+        request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=False,
+            ),
+            registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
+            find_service=lambda *a, **kw: None,
+        )
+
+        view = views.ManageOIDCPublisherViews(project, request)
+        resp = view.manage_project_oidc_publishers()
+        assert isinstance(resp, Response)
+        assert resp.status_code == 403
+
     def test_add_github_oidc_publisher_preexisting(self, monkeypatch):
         publisher = pretend.stub(
             id="fakeid",
@@ -9650,6 +9671,9 @@ def test_add_github_oidc_publisher_preexisting(self, monkeypatch):
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
 
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -9731,6 +9755,9 @@ def test_add_github_oidc_publisher_created(self, monkeypatch):
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
 
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -9838,6 +9865,9 @@ def test_add_github_oidc_publisher_already_registered_with_project(
         )
 
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -9898,6 +9928,9 @@ def test_add_github_oidc_publisher_ratelimited(self, monkeypatch):
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
 
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(
                 settings={
                     "warehouse.oidc.enabled": True,
@@ -9943,9 +9976,27 @@ def test_add_github_oidc_publisher_oidc_not_enabled(self):
         with pytest.raises(HTTPNotFound):
             view.add_github_oidc_publisher()
 
+    def test_add_github_oidc_publisher_not_in_beta(self):
+        project = pretend.stub()
+        request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=False,
+            ),
+            registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
+            find_service=lambda *a, **kw: None,
+        )
+
+        view = views.ManageOIDCPublisherViews(project, request)
+        resp = view.add_github_oidc_publisher()
+        assert isinstance(resp, Response)
+        assert resp.status_code == 403
+
     def test_add_github_oidc_publisher_admin_disabled(self, monkeypatch):
         project = pretend.stub()
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: None,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: True)),
@@ -9974,6 +10025,9 @@ def test_add_github_oidc_publisher_invalid_form(self, monkeypatch):
         project = pretend.stub()
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: metrics,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
@@ -10030,6 +10084,9 @@ def test_delete_oidc_publisher_registered_to_multiple_projects(self, monkeypatch
         )
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: metrics,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
@@ -10128,6 +10185,9 @@ def test_delete_oidc_publisher_entirely(self, monkeypatch):
         )
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: metrics,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
@@ -10208,6 +10268,9 @@ def test_delete_oidc_publisher_invalid_form(self, monkeypatch):
         project = pretend.stub(oidc_publishers=[publisher])
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: metrics,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
@@ -10258,6 +10321,9 @@ def test_delete_oidc_publisher_not_found(self, monkeypatch, other_publisher):
         )
         metrics = pretend.stub(increment=pretend.call_recorder(lambda *a, **kw: None))
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: metrics,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: False)),
@@ -10314,9 +10380,27 @@ def test_delete_oidc_publisher_oidc_not_enabled(self):
         with pytest.raises(HTTPNotFound):
             view.delete_oidc_publisher()
 
+    def test_add_delete_oidc_publisher_not_in_beta(self):
+        project = pretend.stub()
+        request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=False,
+            ),
+            registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
+            find_service=lambda *a, **kw: None,
+        )
+
+        view = views.ManageOIDCPublisherViews(project, request)
+        resp = view.delete_oidc_publisher()
+        assert isinstance(resp, Response)
+        assert resp.status_code == 403
+
     def test_delete_oidc_publisher_admin_disabled(self, monkeypatch):
         project = pretend.stub()
         request = pretend.stub(
+            user=pretend.stub(
+                in_oidc_beta=True,
+            ),
             registry=pretend.stub(settings={"warehouse.oidc.enabled": True}),
             find_service=lambda *a, **kw: None,
             flags=pretend.stub(enabled=pretend.call_recorder(lambda f: True)),