feat: SSO identity provider config get/update + signing certificate rotation (REQ-1612)

pyatlan/client/aio/sso.py

@@ -3,7 +3,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, List
+from typing import List, cast
 
 from pydantic.v1 import validate_arguments
 
@@ -13,15 +13,15 @@
     SSOCreateGroupMapping,
     SSODeleteGroupMapping,
     SSOGetAllGroupMappings,
+    SSOGetAllIdentityProviders,
     SSOGetGroupMapping,
     SSOUpdateGroupMapping,
+    SSOUpdateIdentityProvider,
+    normalize_signing_certificate,
 )
 from pyatlan.errors import ErrorCode
 from pyatlan.model.group import AtlanGroup
-from pyatlan.model.sso import SSOMapper
-
-if TYPE_CHECKING:
-    pass
+from pyatlan.model.sso import SSOMapper, SSOProvider
 
 
 class AsyncSSOClient:
@@ -142,3 +142,81 @@ async def delete_group_mapping(self, sso_alias: str, group_map_id: str) -> None:
         )
         raw_json = await self._client._call_api(endpoint, request_obj=request_obj)
         return raw_json
+
+    async def get_all_identity_providers(self) -> List[SSOProvider]:
+        """
+        Retrieves all SSO identity providers configured on the tenant.
+
+        Requires an API token with the admin role or workspace-admin
+        subrole (READ_TENANT_IDP); other tokens receive a 403.
+
+        :raises AtlanError: on any error during API invocation.
+        :returns: list of the tenant's SSO identity providers
+        """
+        endpoint, request_obj = SSOGetAllIdentityProviders.prepare_request()
+        raw_json = await self._client._call_api(endpoint, request_obj=request_obj)
+        return SSOGetAllIdentityProviders.process_response(raw_json)
+
+    @validate_arguments
+    async def get_identity_provider(self, sso_alias: str) -> SSOProvider:
+        """
+        Retrieves the SSO identity provider with the given alias.
+
+        :param sso_alias: alias of the SSO provider (e.g. `azure`, `okta`)
+        :raises AtlanError: on any error during API invocation.
+        :raises NotFoundError: if no identity provider exists with the given alias.
+        :returns: the identity provider configuration
+        """
+        for provider in await self.get_all_identity_providers():
+            if provider.alias == sso_alias:
+                return provider
+        raise ErrorCode.IDP_NOT_FOUND_BY_ALIAS.exception_with_parameters(sso_alias)
+
+    @validate_arguments
+    async def update_identity_provider(self, provider: SSOProvider) -> SSOProvider:
+        """
+        Updates an SSO identity provider's configuration.
+
+        The backend treats this as a full replacement: always retrieve the
+        current configuration first (`get_identity_provider()`), modify it,
+        and pass the complete object here. Sending a partial object may
+        silently reset fields that were omitted.
+
+        Requires an API token with the admin role or workspace-admin
+        subrole (UPDATE_TENANT_IDP); other tokens receive a 403.
+
+        :param provider: the complete identity provider configuration to store
+        :raises AtlanError: on any error during API invocation.
+        :returns: the identity provider configuration, re-read after the update
+        """
+        endpoint, request_obj = SSOUpdateIdentityProvider.prepare_request(provider)
+        alias = cast(str, provider.alias)  # validated non-empty by prepare_request
+        await self._client._call_api(endpoint, request_obj=request_obj)
+        return await self.get_identity_provider(sso_alias=alias)
+
+    @validate_arguments
+    async def update_signing_certificate(
+        self, sso_alias: str, certificate: str
+    ) -> SSOProvider:
+        """
+        Replaces the signing certificate on the given SSO identity
+        provider, leaving the rest of the configuration untouched.
+
+        API tokens authenticate independently of SSO, so this works even
+        while SSO logins are failing (for example, after the certificate
+        expired) - as long as the token already exists.
+
+        :param sso_alias: alias of the SSO provider (e.g. `azure`, `okta`)
+        :param certificate: new X.509 certificate, as PEM or a single line
+            of base64; stored as one line with no BEGIN/END lines and no
+            line breaks
+        :raises AtlanError: on any error during API invocation.
+        :raises NotFoundError: if no identity provider exists with the given alias.
+        :returns: the identity provider configuration, re-read after the update
+        """
+        provider = await self.get_identity_provider(sso_alias=sso_alias)
+        provider.config = provider.config or {}
+        provider.config["signingCertificate"] = normalize_signing_certificate(
+            certificate
+        )
+        return await self.update_identity_provider(provider=provider)

pyatlan/client/common/__init__.py

@@ -128,8 +128,11 @@
     SSOCreateGroupMapping,
     SSODeleteGroupMapping,
     SSOGetAllGroupMappings,
+    SSOGetAllIdentityProviders,
     SSOGetGroupMapping,
     SSOUpdateGroupMapping,
+    SSOUpdateIdentityProvider,
+    normalize_signing_certificate,
 )
 
 # Task shared logic classes
@@ -284,6 +287,9 @@
     # SSO shared logic classes
     "SSOCheckExistingMappings",
     "SSOCreateGroupMapping",
+    "SSOGetAllIdentityProviders",
+    "SSOUpdateIdentityProvider",
+    "normalize_signing_certificate",
     "SSODeleteGroupMapping",
     "SSOGetAllGroupMappings",
     "SSOGetGroupMapping",

pyatlan/client/common/sso.py

@@ -1,20 +1,24 @@
 # SPDX-License-Identifier: Apache-2.0
 # Copyright 2025 Atlan Pte. Ltd.
 
+import base64
+import binascii
 from typing import List
 
 from pydantic.v1 import ValidationError, parse_obj_as
 
 from pyatlan.client.constants import (
     CREATE_SSO_GROUP_MAPPING,
     DELETE_SSO_GROUP_MAPPING,
+    GET_ALL_IDPS,
     GET_ALL_SSO_GROUP_MAPPING,
     GET_SSO_GROUP_MAPPING,
+    UPDATE_IDP,
     UPDATE_SSO_GROUP_MAPPING,
 )
 from pyatlan.errors import ErrorCode
 from pyatlan.model.group import AtlanGroup
-from pyatlan.model.sso import SSOMapper, SSOMapperConfig
+from pyatlan.model.sso import SSOMapper, SSOMapperConfig, SSOProvider
 from pyatlan.utils import get_epoch_timestamp
 
 GROUP_MAPPER_ATTRIBUTE = "memberOf"
@@ -272,3 +276,95 @@ def check_existing_group_mappings(
                 raise ErrorCode.SSO_GROUP_MAPPING_ALREADY_EXISTS.exception_with_parameters(
                     atlan_group.alias, group_map.config.attribute_value
                 )
+
+
+class SSOGetAllIdentityProviders:
+    """Shared logic for retrieving all SSO identity providers."""
+
+    @staticmethod
+    def prepare_request() -> tuple:
+        """
+        Prepare the request for retrieving all identity providers.
+
+        :returns: tuple of (endpoint, request_obj)
+        """
+        return GET_ALL_IDPS, None
+
+    @staticmethod
+    def process_response(raw_json) -> List[SSOProvider]:
+        """
+        Process the raw API response into a list of identity providers.
+
+        :param raw_json: raw API response
+        :returns: list of the tenant's SSO identity providers
+        """
+        if not raw_json:
+            return []
+        try:
+            return parse_obj_as(List[SSOProvider], raw_json)
+        except ValidationError as err:
+            raise ErrorCode.JSON_ERROR.exception_with_parameters(
+                raw_json, 200, str(err)
+            ) from err
+
+
+class SSOUpdateIdentityProvider:
+    """Shared logic for updating an SSO identity provider."""
+
+    @staticmethod
+    def prepare_request(provider: SSOProvider) -> tuple:
+        """
+        Prepare the request for updating an identity provider.
+
+        The backend treats this update as a full replacement of the
+        provider's configuration, so `provider` must be the complete
+        object (retrieve it first, modify it, then pass it here) -
+        never a partial one, or omitted fields may be reset.
+
+        :param provider: the complete identity provider configuration to store
+        :returns: tuple of (endpoint, request_obj)
+        """
+        if not provider.alias:
+            raise ErrorCode.MISSING_REQUIRED_QUERY_PARAM.exception_with_parameters(
+                "the identity provider", "alias"
+            )
+        endpoint = UPDATE_IDP.format_path({"sso_alias": provider.alias})
+        return endpoint, provider
+
+
+def normalize_signing_certificate(certificate: str) -> str:
+    """
+    Convert an X.509 certificate to the form the SSO configuration stores:
+    a single line of base64, with no BEGIN/END lines and no whitespace.
+
+    Accepts PEM or already-normalized input. Exactly one certificate must
+    be provided.
+
+    :param certificate: certificate as PEM or single-line base64
+    :raises InvalidRequestError: if the input contains more than one
+        certificate, no certificate, or is not base64
+    :returns: single-line base64 certificate value
+    """
+    cert_count = certificate.count("BEGIN CERTIFICATE")
+    if cert_count > 1:
+        raise ErrorCode.INVALID_CERTIFICATE.exception_with_parameters(
+            f"found {cert_count} certificates, expected 1 (IdP metadata files "
+            "often include both the old and new certificate - pass only the new one)"
+        )
+    lines = [
+        line.strip()
+        for line in certificate.strip().splitlines()
+        if line.strip() and "CERTIFICATE" not in line
+    ]
+    normalized = "".join("".join(line.split()) for line in lines)
+    if not normalized:
+        raise ErrorCode.INVALID_CERTIFICATE.exception_with_parameters(
+            "no certificate content between the BEGIN/END lines"
+        )
+    try:
+        base64.b64decode(normalized, validate=True)
+    except binascii.Error as err:
+        raise ErrorCode.INVALID_CERTIFICATE.exception_with_parameters(
+            f"not base64 ({err}) - check for truncation or non-certificate text"
+        ) from err
+    return normalized

pyatlan/client/constants.py

@@ -633,6 +633,14 @@
 SSO_API = "idp/"
 SSO_GROUP_MAPPER = SSO_API + "{sso_alias}/mappers"
 
+GET_ALL_IDPS = API("idp", HTTPMethod.GET, HTTPStatus.OK, endpoint=EndPoint.HERACLES)
+UPDATE_IDP = API(
+    SSO_API + "{sso_alias}",
+    HTTPMethod.POST,
+    HTTPStatus.OK,
+    endpoint=EndPoint.HERACLES,
+)
+
 GET_SSO_GROUP_MAPPING = API(
     SSO_GROUP_MAPPER + "/{group_map_id}",
     HTTPMethod.GET,

pyatlan/client/sso.py

@@ -1,4 +1,4 @@
-from typing import List
+from typing import List, cast
 
 from pydantic.v1 import validate_arguments
 
@@ -8,12 +8,15 @@
     SSOCreateGroupMapping,
     SSODeleteGroupMapping,
     SSOGetAllGroupMappings,
+    SSOGetAllIdentityProviders,
     SSOGetGroupMapping,
     SSOUpdateGroupMapping,
+    SSOUpdateIdentityProvider,
+    normalize_signing_certificate,
 )
 from pyatlan.errors import ErrorCode
 from pyatlan.model.group import AtlanGroup
-from pyatlan.model.sso import SSOMapper
+from pyatlan.model.sso import SSOMapper, SSOProvider
 
 
 class SSOClient:
@@ -134,3 +137,81 @@ def delete_group_mapping(self, sso_alias: str, group_map_id: str) -> None:
         )
         raw_json = self._client._call_api(endpoint, request_obj=request_obj)
         return raw_json
+
+    def get_all_identity_providers(self) -> List[SSOProvider]:
+        """
+        Retrieves all SSO identity providers configured on the tenant.
+
+        Requires an API token with the admin role or workspace-admin
+        subrole (READ_TENANT_IDP); other tokens receive a 403.
+
+        :raises AtlanError: on any error during API invocation.
+        :returns: list of the tenant's SSO identity providers
+        """
+        endpoint, request_obj = SSOGetAllIdentityProviders.prepare_request()
+        raw_json = self._client._call_api(endpoint, request_obj=request_obj)
+        return SSOGetAllIdentityProviders.process_response(raw_json)
+
+    @validate_arguments
+    def get_identity_provider(self, sso_alias: str) -> SSOProvider:
+        """
+        Retrieves the SSO identity provider with the given alias.
+
+        :param sso_alias: alias of the SSO provider (e.g. `azure`, `okta`)
+        :raises AtlanError: on any error during API invocation.
+        :raises NotFoundError: if no identity provider exists with the given alias.
+        :returns: the identity provider configuration
+        """
+        for provider in self.get_all_identity_providers():
+            if provider.alias == sso_alias:
+                return provider
+        raise ErrorCode.IDP_NOT_FOUND_BY_ALIAS.exception_with_parameters(sso_alias)
+
+    @validate_arguments
+    def update_identity_provider(self, provider: SSOProvider) -> SSOProvider:
+        """
+        Updates an SSO identity provider's configuration.
+
+        The backend treats this as a full replacement: always retrieve the
+        current configuration first (`get_identity_provider()`), modify it,
+        and pass the complete object here. Sending a partial object may
+        silently reset fields that were omitted.
+
+        Requires an API token with the admin role or workspace-admin
+        subrole (UPDATE_TENANT_IDP); other tokens receive a 403.
+
+        :param provider: the complete identity provider configuration to store
+        :raises AtlanError: on any error during API invocation.
+        :returns: the identity provider configuration, re-read after the update
+        """
+        endpoint, request_obj = SSOUpdateIdentityProvider.prepare_request(provider)
+        alias = cast(str, provider.alias)  # validated non-empty by prepare_request
+        self._client._call_api(endpoint, request_obj=request_obj)
+        return self.get_identity_provider(sso_alias=alias)
+
+    @validate_arguments
+    def update_signing_certificate(
+        self, sso_alias: str, certificate: str
+    ) -> SSOProvider:
+        """
+        Replaces the signing certificate on the given SSO identity
+        provider, leaving the rest of the configuration untouched.
+
+        API tokens authenticate independently of SSO, so this works even
+        while SSO logins are failing (for example, after the certificate
+        expired) - as long as the token already exists.
+
+        :param sso_alias: alias of the SSO provider (e.g. `azure`, `okta`)
+        :param certificate: new X.509 certificate, as PEM or a single line
+            of base64; stored as one line with no BEGIN/END lines and no
+            line breaks
+        :raises AtlanError: on any error during API invocation.
+        :raises NotFoundError: if no identity provider exists with the given alias.
+        :returns: the identity provider configuration, re-read after the update
+        """
+        provider = self.get_identity_provider(sso_alias=sso_alias)
+        provider.config = provider.config or {}
+        provider.config["signingCertificate"] = normalize_signing_certificate(
+            certificate
+        )
+        return self.update_identity_provider(provider=provider)