Cloud Identity Federation: SSO and Identity Providers

requirements.txt:

boto3>=1.34.0
pyjwt>=2.8.0
cryptography>=41.0.0

Complete Identity Federation Manager:

#!/usr/bin/env python3
"""
Cloud Identity Federation - Identity Federation Manager
Production-ready identity federation with SAML, OIDC, and comprehensive error handling
"""

import boto3
from botocore.exceptions import ClientError, BotoCoreError
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import logging
import base64
import os
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class FederationError(Exception):
    """Base exception for federation errors."""
    pass


class ProviderNotFoundError(FederationError):
    """Raised when identity provider is not found."""
    pass


class InvalidMetadataError(FederationError):
    """Raised when SAML metadata is invalid."""
    pass


@dataclass
class SAMLProviderConfig:
    """SAML identity provider configuration."""
    name: str
    entity_id: str
    sso_url: str
    certificate: str
    metadata_document: Optional[str] = None
    attributes: Optional[Dict[str, str]] = None
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)


@dataclass
class OIDCProviderConfig:
    """OpenID Connect provider configuration."""
    name: str
    url: str
    client_ids: List[str]
    thumbprints: Optional[List[str]] = None
    groups_attribute: Optional[str] = None
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)


class CloudIdentityFederationManager:
    """Manages cloud identity federation with comprehensive error handling."""
    
    def __init__(
        self,
        region_name: str = 'us-east-1',
        aws_access_key_id: Optional[str] = None,
        aws_secret_access_key: Optional[str] = None
    ):
        """Initialize identity federation manager.
        
        Args:
            region_name: AWS region (default: us-east-1)
            aws_access_key_id: AWS access key (defaults to env/credentials)
            aws_secret_access_key: AWS secret key (defaults to env/credentials)
        """
        self.region_name = region_name
        
        try:
            session = boto3.Session(
                aws_access_key_id=aws_access_key_id or os.getenv('AWS_ACCESS_KEY_ID'),
                aws_secret_access_key=aws_secret_access_key or os.getenv('AWS_SECRET_ACCESS_KEY'),
                region_name=region_name
            )
            
            self.iam = session.client('iam')
            logger.info(f"Initialized CloudIdentityFederationManager for region: {region_name}")
            
        except (ClientError, BotoCoreError) as e:
            error_msg = f"Failed to initialize AWS clients: {e}"
            logger.error(error_msg)
            raise FederationError(error_msg) from e
    
    def create_saml_provider(
        self,
        config: SAMLProviderConfig,
        metadata_file_path: Optional[str] = None
    ) -> Dict:
        """Create SAML identity provider in AWS.
        
        Args:
            config: SAML provider configuration
            metadata_file_path: Optional path to SAML metadata XML file
            
        Returns:
            Created SAML provider ARN and metadata
            
        Raises:
            InvalidMetadataError: If SAML metadata is invalid
            FederationError: If provider creation fails
        """
        try:
            # Load metadata document if provided
            metadata_document = config.metadata_document
            if metadata_file_path and os.path.exists(metadata_file_path):
                with open(metadata_file_path, 'r') as f:
                    metadata_document = f.read()
            elif not metadata_document:
                # Generate basic metadata document from config
                metadata_document = self._generate_saml_metadata(config)
            
            # Validate metadata document
            if not self._validate_saml_metadata(metadata_document):
                raise InvalidMetadataError("Invalid SAML metadata document")
            
            logger.info(f"Creating SAML provider '{config.name}'")
            
            # Create SAML provider
            response = self.iam.create_saml_provider(
                SAMLMetadataDocument=metadata_document,
                Name=config.name
            )
            
            provider_arn = response['SAMLProviderArn']
            logger.info(f"Created SAML provider: {provider_arn}")
            
            return {
                'provider_arn': provider_arn,
                'metadata': {
                    'create_date': response['CreateDate'].isoformat(),
                    'valid_until': response.get('ValidUntil', 'N/A')
                }
            }
        
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'InvalidInput':
                raise InvalidMetadataError(f"Invalid SAML metadata: {e}") from e
            elif error_code == 'EntityAlreadyExists':
                raise FederationError(f"SAML provider '{config.name}' already exists") from e
            else:
                error_msg = f"Failed to create SAML provider: {e}"
                logger.error(error_msg)
                raise FederationError(error_msg) from e
        except Exception as e:
            error_msg = f"Unexpected error creating SAML provider: {e}"
            logger.error(error_msg, exc_info=True)
            raise FederationError(error_msg) from e
    
    def create_oidc_provider(self, config: OIDCProviderConfig) -> Dict:
        """Create OpenID Connect identity provider in AWS.
        
        Args:
            config: OIDC provider configuration
            
        Returns:
            Created OIDC provider ARN and metadata
            
        Raises:
            FederationError: If provider creation fails
        """
        try:
            logger.info(f"Creating OIDC provider '{config.name}' at {config.url}")
            
            # Prepare thumbprints (required for some providers)
            thumbprints = config.thumbprints or []
            
            # Create OIDC provider
            response = self.iam.create_open_id_connect_provider(
                Url=config.url,
                ClientIDList=config.client_ids,
                ThumbprintList=thumbprints
            )
            
            provider_arn = response['OpenIDConnectProviderArn']
            logger.info(f"Created OIDC provider: {provider_arn}")
            
            return {
                'provider_arn': provider_arn,
                'url': config.url,
                'client_ids': config.client_ids
            }
        
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'EntityAlreadyExists':
                raise FederationError(f"OIDC provider at '{config.url}' already exists") from e
            else:
                error_msg = f"Failed to create OIDC provider: {e}"
                logger.error(error_msg)
                raise FederationError(error_msg) from e
        except Exception as e:
            error_msg = f"Unexpected error creating OIDC provider: {e}"
            logger.error(error_msg, exc_info=True)
            raise FederationError(error_msg) from e
    
    def list_identity_providers(self) -> Dict:
        """List all identity providers.
        
        Returns:
            Dictionary with SAML and OIDC providers
        """
        providers = {
            'saml_providers': [],
            'oidc_providers': []
        }
        
        try:
            # List SAML providers
            saml_response = self.iam.list_saml_providers()
            for provider in saml_response.get('SAMLProviderList', []):
                providers['saml_providers'].append({
                    'arn': provider['Arn'],
                    'valid_until': provider.get('ValidUntil', 'N/A')
                })
            
            # List OIDC providers
            oidc_response = self.iam.list_open_id_connect_providers()
            for provider_arn in oidc_response.get('OpenIDConnectProviderArnList', []):
                provider_detail = self.iam.get_open_id_connect_provider(
                    OpenIDConnectProviderArn=provider_arn
                )
                providers['oidc_providers'].append({
                    'arn': provider_arn,
                    'url': provider_detail['Url'],
                    'client_ids': provider_detail['ClientIDList']
                })
        
        except ClientError as e:
            logger.error(f"Error listing identity providers: {e}")
            raise FederationError(f"Failed to list providers: {e}") from e
        
        return providers
    
    def get_saml_provider_metadata(self, provider_arn: str) -> Dict:
        """Get SAML provider metadata.
        
        Args:
            provider_arn: SAML provider ARN
            
        Returns:
            Provider metadata and metadata document
            
        Raises:
            ProviderNotFoundError: If provider not found
        """
        try:
            response = self.iam.get_saml_provider(SAMLProviderArn=provider_arn)
            
            return {
                'arn': provider_arn,
                'metadata_document': response['SAMLMetadataDocument'],
                'create_date': response['CreateDate'].isoformat(),
                'valid_until': response.get('ValidUntil', 'N/A')
            }
        
        except ClientError as e:
            if e.response['Error']['Code'] == 'NoSuchEntity':
                raise ProviderNotFoundError(f"SAML provider {provider_arn} not found") from e
            raise FederationError(f"Failed to get SAML provider: {e}") from e
    
    def delete_saml_provider(self, provider_arn: str) -> None:
        """Delete SAML identity provider.
        
        Args:
            provider_arn: SAML provider ARN
            
        Raises:
            ProviderNotFoundError: If provider not found
        """
        try:
            self.iam.delete_saml_provider(SAMLProviderArn=provider_arn)
            logger.info(f"Deleted SAML provider: {provider_arn}")
        
        except ClientError as e:
            if e.response['Error']['Code'] == 'NoSuchEntity':
                raise ProviderNotFoundError(f"SAML provider {provider_arn} not found") from e
            raise FederationError(f"Failed to delete SAML provider: {e}") from e
    
    def delete_oidc_provider(self, provider_arn: str) -> None:
        """Delete OIDC identity provider.
        
        Args:
            provider_arn: OIDC provider ARN
            
        Raises:
            ProviderNotFoundError: If provider not found
        """
        try:
            self.iam.delete_open_id_connect_provider(
                OpenIDConnectProviderArn=provider_arn
            )
            logger.info(f"Deleted OIDC provider: {provider_arn}")
        
        except ClientError as e:
            if e.response['Error']['Code'] == 'NoSuchEntity':
                raise ProviderNotFoundError(f"OIDC provider {provider_arn} not found") from e
            raise FederationError(f"Failed to delete OIDC provider: {e}") from e
    
    def _validate_saml_metadata(self, metadata: str) -> bool:
        """Validate SAML metadata document.
        
        Args:
            metadata: SAML metadata XML string
            
        Returns:
            True if valid, False otherwise
        """
        if not metadata or not isinstance(metadata, str):
            return False
        
        # Basic validation - check for required SAML elements
        required_elements = [
            'EntityDescriptor',
            'IDPSSODescriptor',
            'SingleSignOnService'
        ]
        
        for element in required_elements:
            if element not in metadata:
                logger.warning(f"Missing required SAML element: {element}")
                return False
        
        return True
    
    def _generate_saml_metadata(self, config: SAMLProviderConfig) -> str:
        """Generate basic SAML metadata document from configuration.
        
        Args:
            config: SAML provider configuration
            
        Returns:
            SAML metadata XML string
        """
        # Note: This is a simplified metadata generation
        # In production, use proper SAML metadata from your IdP
        metadata_template = f"""<?xml version="1.0"?>
<EntityDescriptor entityID="{config.entity_id}">
  <IDPSSODescriptor protocolSupportEnumeration="urn:oasis:names:tc:SAML:2.0:protocol">
    <KeyDescriptor use="signing">
      <KeyInfo>
        <X509Data>
          <X509Certificate>{config.certificate}</X509Certificate>
        </X509Data>
      </KeyInfo>
    </KeyDescriptor>
    <SingleSignOnService Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
                         Location="{config.sso_url}"/>
  </IDPSSODescriptor>
</EntityDescriptor>"""
        
        return metadata_template


# Example usage
if __name__ == "__main__":
    manager = CloudIdentityFederationManager(region_name='us-east-1')
    
    # Create SAML provider
    saml_config = SAMLProviderConfig(
        name="corporate-saml-provider",
        entity_id="https://idp.example.com",
        sso_url="https://idp.example.com/sso",
        certificate="MIICXjCCAcegAwIBAgIJAK...",  # Base64 encoded cert
        attributes={
            "email": "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress",
            "name": "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"
        }
    )
    
    try:
        result = manager.create_saml_provider(saml_config)
        print(f"Created SAML provider: {result['provider_arn']}")
    except FederationError as e:
        print(f"Error: {e}")
    
    # Create OIDC provider
    oidc_config = OIDCProviderConfig(
        name="google-oidc",
        url="https://accounts.google.com",
        client_ids=["123456789.apps.googleusercontent.com"],
        thumbprints=[]
    )
    
    try:
        result = manager.create_oidc_provider(oidc_config)
        print(f"Created OIDC provider: {result['provider_arn']}")
    except FederationError as e:
        print(f"Error: {e}")
    
    # List all providers
    providers = manager.list_identity_providers()
    print(f"SAML Providers: {len(providers['saml_providers'])}")
    print(f"OIDC Providers: {len(providers['oidc_providers'])}")

YAML Configuration Example:

# saml-provider-config.yaml
saml_provider:
  name: "cloud-federation"
  entity_id: "https://idp.example.com"
  sso_url: "https://idp.example.com/sso"
  certificate: "MIICXjCCAcegAwIBAgIJAK..."
  metadata_file: "saml-metadata.xml"  # Optional: path to full metadata
  attributes:
    email: "urn:oasis:names:tc:SAML:1.1:nameid-format:emailAddress"
    name: "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/name"
    groups: "http://schemas.xmlsoap.org/ws/2005/05/identity/claims/groups"

# oidc-provider-config.yaml
oidc_provider:
  name: "google-oidc"
  url: "https://accounts.google.com"
  client_ids:
    - "123456789.apps.googleusercontent.com"
  thumbprints: []  # Auto-discovered if empty
  groups_attribute: "groups"

See Step 1 for the complete Python implementation. The manager includes methods for creating and managing both SAML and OIDC providers in AWS.

Example AWS SAML Provider Creation:

manager = CloudIdentityFederationManager(region_name='us-east-1')

# Load SAML metadata from file
    with open('saml-metadata.xml', 'r') as f:
    metadata = f.read()

saml_config = SAMLProviderConfig(
    name="aws-federation-provider",
    entity_id="https://idp.example.com",
    sso_url="https://idp.example.com/sso",
    certificate="MIICXjCCAceg...",
    metadata_document=metadata
)

result = manager.create_saml_provider(saml_config)
print(f"Provider ARN: {result['provider_arn']}")

Example AWS OIDC Provider Creation:

oidc_config = OIDCProviderConfig(
    name="github-oidc",
    url="https://token.actions.githubusercontent.com",
    client_ids=["sts.amazonaws.com"]
)

result = manager.create_oidc_provider(oidc_config)
print(f"Provider ARN: {result['provider_arn']}")