feat: Add custom OIDC issuers to orgs (#18841)

* feat(orgs): add 1-many relationship to OIDC issuers Allows storing a specific `issuer_url` associated to the org, for later lookup. Signed-off-by: Mike Fiedler <miketheman@gmail.com> * feat(admin/orgs): set/remove org-based OIDC issuers Signed-off-by: Mike Fiedler <miketheman@gmail.com> --------- Signed-off-by: Mike Fiedler <miketheman@gmail.com>

Author: miketheman

tests/common/db/organizations.py

@@ -7,12 +7,14 @@
 
 from warehouse.observations.models import ObservationKind
 from warehouse.organizations.models import (
+ OIDCIssuerType,
  Organization,
  OrganizationApplication,
  OrganizationApplicationStatus,
  OrganizationInvitation,
  OrganizationManualActivation,
  OrganizationNameCatalog,
+ OrganizationOIDCIssuer,
  OrganizationProject,
  OrganizationRole,
  OrganizationRoleType,
@@ -210,3 +212,17 @@ class Meta:
  role_name = TeamProjectRoleType.Owner
  project = factory.SubFactory(ProjectFactory)
  team = factory.SubFactory(TeamFactory)
+
+
+class OrganizationOIDCIssuerFactory(WarehouseFactory):
+ class Meta:
+ model = OrganizationOIDCIssuer
+
+ organization_id = factory.SelfAttribute("organization.id")
+ organization = factory.SubFactory(OrganizationFactory)
+ issuer_type = factory.LazyFunction(
+ lambda: fake.random_element(elements=[e.value for e in OIDCIssuerType])
+ )
+ issuer_url = factory.Faker("url", schemes=["https"])
+ created_by_id = factory.SelfAttribute("created_by.id")
+ created_by = factory.SubFactory(UserFactory)

tests/unit/admin/test_routes.py

@@ -69,6 +69,16 @@ def test_includeme():
  "/admin/organizations/{organization_id}/set_total_size_limit/",
  domain=warehouse,
  ),
+ pretend.call(
+ "admin.organization.add_oidc_issuer",
+ "/admin/organizations/{organization_id}/oidc-issuers/add/",
+ domain=warehouse,
+ ),
+ pretend.call(
+ "admin.organization.delete_oidc_issuer",
+ "/admin/organizations/{organization_id}/oidc-issuers/{issuer_id}/delete/",
+ domain=warehouse,
+ ),
  pretend.call(
  "admin.organization_application.list",
  "/admin/organization_applications/",

tests/unit/admin/views/test_organizations.py

@@ -13,13 +13,16 @@
 from tests.common.db.organizations import (
  OrganizationFactory,
  OrganizationManualActivationFactory,
+ OrganizationOIDCIssuerFactory,
  OrganizationRoleFactory,
  OrganizationStripeCustomerFactory,
 )
 from tests.common.db.subscriptions import StripeCustomerFactory
 from warehouse.admin.views import organizations as views
 from warehouse.organizations.models import (
+ OIDCIssuerType,
  OrganizationManualActivation,
+ OrganizationOIDCIssuer,
  OrganizationRole,
  OrganizationRoleType,
  OrganizationType,
@@ -1736,3 +1739,317 @@ def test_set_total_size_limit_below_default(self, db_request):
  )
  ]
  assert result.status_code == 303
+
+
+class TestAddOIDCIssuer:
+ def test_add_oidc_issuer_success(self, db_request, monkeypatch):
+ organization = OrganizationFactory.create()
+ admin_user = UserFactory.create(username="admin")
+
+ # Mock record_event
+ record_event = pretend.call_recorder(lambda **kwargs: None)
+ monkeypatch.setattr(organization, "record_event", record_event)
+
+ db_request.matchdict = {"organization_id": str(organization.id)}
+ db_request.user = admin_user
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ db_request.POST = MultiDict(
+ {
+ "issuer_type": "gitlab",
+ "issuer_url": "https://gitlab.company.com",
+ }
+ )
+
+ result = views.add_oidc_issuer(db_request)
+
+ assert isinstance(result, HTTPSeeOther)
+ assert result.location == "/admin/organizations/"
+
+ assert db_request.session.flash.calls == [
+ pretend.call(
+ "OIDC issuer 'https://gitlab.company.com' (gitlab) added to "
+ f"'{organization.name}'",
+ queue="success",
+ )
+ ]
+
+ issuer = db_request.db.query(OrganizationOIDCIssuer).one()
+ assert issuer.issuer_type == OIDCIssuerType.GitLab
+ assert issuer.issuer_url == "https://gitlab.company.com"
+ assert issuer.organization == organization
+ assert issuer.created_by == admin_user
+
+ # Check event was recorded
+ assert record_event.calls == [
+ pretend.call(
+ request=db_request,
+ tag="admin:organization:oidc_issuer:add",
+ additional={
+ "issuer_type": "gitlab",
+ "issuer_url": "https://gitlab.company.com",
+ },
+ )
+ ]
+
+ def test_add_oidc_issuer_invalid_form(self, db_request):
+ organization = OrganizationFactory.create()
+ admin_user = UserFactory.create(username="admin")
+
+ db_request.matchdict = {"organization_id": str(organization.id)}
+ db_request.user = admin_user
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ # Missing issuer_url
+ db_request.POST = MultiDict({"issuer_type": "gitlab"})
+
+ result = views.add_oidc_issuer(db_request)
+ assert isinstance(result, HTTPSeeOther)
+
+ # Should flash form validation errors
+ assert len(db_request.session.flash.calls) > 0
+ assert "error" in str(db_request.session.flash.calls[0])
+
+ def test_add_oidc_issuer_invalid_url(self, db_request):
+ organization = OrganizationFactory.create()
+ admin_user = UserFactory.create(username="admin")
+
+ db_request.matchdict = {"organization_id": str(organization.id)}
+ db_request.user = admin_user
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ # Invalid URL (not https)
+ db_request.POST = MultiDict(
+ {
+ "issuer_type": "gitlab",
+ "issuer_url": "http://gitlab.company.com",
+ }
+ )
+
+ result = views.add_oidc_issuer(db_request)
+ assert isinstance(result, HTTPSeeOther)
+
+ # Should flash form validation errors
+ flash_messages = [call.args[0] for call in db_request.session.flash.calls]
+ assert any("https://" in msg for msg in flash_messages)
+
+ def test_add_oidc_issuer_duplicate(self, db_request, monkeypatch):
+ organization = OrganizationFactory.create()
+ admin_user = UserFactory.create(username="admin")
+
+ # Create existing issuer
+ OrganizationOIDCIssuerFactory.create(
+ organization=organization,
+ issuer_type=OIDCIssuerType.GitLab,
+ issuer_url="https://gitlab.company.com",
+ created_by=admin_user,
+ )
+
+ # Mock record_event (should not be called on duplicate)
+ record_event = pretend.call_recorder(lambda **kwargs: None)
+ monkeypatch.setattr(organization, "record_event", record_event)
+
+ db_request.matchdict = {"organization_id": str(organization.id)}
+ db_request.user = admin_user
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ db_request.POST = MultiDict(
+ {
+ "issuer_type": "gitlab",
+ "issuer_url": "https://gitlab.company.com",
+ }
+ )
+
+ result = views.add_oidc_issuer(db_request)
+ assert isinstance(result, HTTPSeeOther)
+
+ assert db_request.session.flash.calls == [
+ pretend.call(
+ "Issuer 'https://gitlab.company.com' already exists "
+ f"for organization '{organization.name}'",
+ queue="error",
+ )
+ ]
+
+ # No new event recorded
+ assert record_event.calls == []
+
+ def test_add_oidc_issuer_organization_not_found(self, db_request):
+ admin_user = UserFactory.create(username="admin")
+
+ db_request.matchdict = {
+ "organization_id": "00000000-0000-0000-0000-000000000000"
+ }
+ db_request.user = admin_user
+
+ with pytest.raises(HTTPNotFound):
+ views.add_oidc_issuer(db_request)
+
+
+class TestDeleteOIDCIssuer:
+ def test_delete_oidc_issuer_success(self, db_request, monkeypatch):
+ organization = OrganizationFactory.create()
+ admin_user = UserFactory.create(username="admin")
+
+ issuer = OrganizationOIDCIssuerFactory.create(
+ organization=organization,
+ issuer_type=OIDCIssuerType.GitLab,
+ issuer_url="https://gitlab.company.com",
+ created_by=admin_user,
+ )
+
+ # Mock record_event
+ record_event = pretend.call_recorder(lambda **kwargs: None)
+ monkeypatch.setattr(organization, "record_event", record_event)
+
+ db_request.matchdict = {
+ "organization_id": str(organization.id),
+ "issuer_id": str(issuer.id),
+ }
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ db_request.POST = MultiDict({"confirm": "https://gitlab.company.com"})
+
+ result = views.delete_oidc_issuer(db_request)
+
+ assert isinstance(result, HTTPSeeOther)
+ assert result.location == "/admin/organizations/"
+
+ assert db_request.session.flash.calls == [
+ pretend.call(
+ "OIDC issuer 'https://gitlab.company.com' removed "
+ f"from '{organization.name}'",
+ queue="success",
+ )
+ ]
+
+ assert db_request.db.query(OrganizationOIDCIssuer).count() == 0
+
+ # Check event was recorded
+ assert record_event.calls == [
+ pretend.call(
+ request=db_request,
+ tag="admin:organization:oidc_issuer:delete",
+ additional={
+ "issuer_type": "gitlab",
+ "issuer_url": "https://gitlab.company.com",
+ },
+ )
+ ]
+
+ def test_delete_oidc_issuer_not_found(self, db_request):
+ organization = OrganizationFactory.create()
+
+ db_request.matchdict = {
+ "organization_id": str(organization.id),
+ "issuer_id": "00000000-0000-0000-0000-000000000000",
+ }
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ db_request.POST = MultiDict({"confirm": "https://gitlab.company.com"})
+
+ result = views.delete_oidc_issuer(db_request)
+ assert isinstance(result, HTTPSeeOther)
+
+ assert db_request.session.flash.calls == [
+ pretend.call("This issuer does not exist", queue="error")
+ ]
+
+ def test_delete_oidc_issuer_wrong_confirmation(self, db_request):
+ organization = OrganizationFactory.create()
+ admin_user = UserFactory.create(username="admin")
+
+ issuer = OrganizationOIDCIssuerFactory.create(
+ organization=organization,
+ issuer_type=OIDCIssuerType.GitLab,
+ issuer_url="https://gitlab.company.com",
+ created_by=admin_user,
+ )
+
+ db_request.matchdict = {
+ "organization_id": str(organization.id),
+ "issuer_id": str(issuer.id),
+ }
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ db_request.POST = MultiDict({"confirm": "https://wrong-url.com"})
+
+ result = views.delete_oidc_issuer(db_request)
+ assert isinstance(result, HTTPSeeOther)
+
+ assert db_request.session.flash.calls == [
+ pretend.call("Confirm the request", queue="error")
+ ]
+
+ # Issuer should still exist
+ assert db_request.db.query(OrganizationOIDCIssuer).count() == 1
+
+ def test_delete_oidc_issuer_no_confirmation(self, db_request):
+ organization = OrganizationFactory.create()
+ admin_user = UserFactory.create(username="admin")
+
+ issuer = OrganizationOIDCIssuerFactory.create(
+ organization=organization,
+ issuer_type=OIDCIssuerType.GitLab,
+ issuer_url="https://gitlab.company.com",
+ created_by=admin_user,
+ )
+
+ db_request.matchdict = {
+ "organization_id": str(organization.id),
+ "issuer_id": str(issuer.id),
+ }
+ db_request.route_path = pretend.call_recorder(
+ lambda *a, **kw: "/admin/organizations/"
+ )
+ db_request.session = pretend.stub(
+ flash=pretend.call_recorder(lambda *a, **kw: None)
+ )
+ db_request.POST = MultiDict({})
+
+ result = views.delete_oidc_issuer(db_request)
+ assert isinstance(result, HTTPSeeOther)
+
+ assert db_request.session.flash.calls == [
+ pretend.call("Confirm the request", queue="error")
+ ]
+
+ # Issuer should still exist
+ assert db_request.db.query(OrganizationOIDCIssuer).count() == 1
+
+ def test_delete_oidc_issuer_organization_not_found(self, db_request):
+ db_request.matchdict = {
+ "organization_id": "00000000-0000-0000-0000-000000000000",
+ "issuer_id": "00000000-0000-0000-0000-000000000001",
+ }
+
+ with pytest.raises(HTTPNotFound):
+ views.delete_oidc_issuer(db_request)