Federated Identity Credentials

def create_or_update(
    resource_group_name: str,
    resource_name: str,
    federated_identity_credential_resource_name: str,
    parameters: Union[FederatedIdentityCredential, IO[bytes]],
    **kwargs
) -> FederatedIdentityCredential:
    """
    Create or update a federated identity credential.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the user-assigned identity
    - federated_identity_credential_resource_name (str): Name of the federated credential
    - parameters (Union[FederatedIdentityCredential, IO[bytes]]): Credential configuration parameters or raw JSON bytes
    - **kwargs: Additional request options

    Returns:
    FederatedIdentityCredential: The created or updated credential

    Raises:
    ResourceNotFoundError: If the user-assigned identity does not exist
    HttpResponseError: For validation or other API errors
    """

from azure.mgmt.msi import ManagedServiceIdentityClient
from azure.identity import DefaultAzureCredential

client = ManagedServiceIdentityClient(
    credential=DefaultAzureCredential(),
    subscription_id="your-subscription-id"
)

# Create federated credential for Kubernetes workload identity
k8s_credential = client.federated_identity_credentials.create_or_update(
    resource_group_name="myResourceGroup",
    resource_name="myUserIdentity",
    federated_identity_credential_resource_name="k8s-default-workload",
    parameters={
        "properties": {
            "issuer": "https://oidc.prod-aks.azure.com/tenant-guid/issuer-guid",
            "subject": "system:serviceaccount:default:workload-service-account",
            "audiences": ["api://AzureADTokenExchange"]
        }
    }
)

print(f"Created Kubernetes federated credential: {k8s_credential.name}")
print(f"Issuer: {k8s_credential.issuer}")
print(f"Subject: {k8s_credential.subject}")

# Create federated credential for GitHub Actions
github_credential = client.federated_identity_credentials.create_or_update(
    resource_group_name="myResourceGroup", 
    resource_name="myUserIdentity",
    federated_identity_credential_resource_name="github-actions-main",
    parameters={
        "properties": {
            "issuer": "https://token.actions.githubusercontent.com",
            "subject": "repo:myorg/myrepo:ref:refs/heads/main",
            "audiences": ["api://AzureADTokenExchange"]
        }
    }
)

# Create federated credential for external OIDC provider
external_credential = client.federated_identity_credentials.create_or_update(
    resource_group_name="myResourceGroup",
    resource_name="myUserIdentity", 
    federated_identity_credential_resource_name="external-oidc",
    parameters={
        "properties": {
            "issuer": "https://auth.example.com",
            "subject": "external-workload-123",
            "audiences": ["api://AzureADTokenExchange", "custom-audience"]
        }
    }
)

def get(
    resource_group_name: str,
    resource_name: str,
    federated_identity_credential_resource_name: str,
    **kwargs
) -> FederatedIdentityCredential:
    """
    Get a federated identity credential.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the user-assigned identity
    - federated_identity_credential_resource_name (str): Name of the federated credential
    - **kwargs: Additional request options

    Returns:
    FederatedIdentityCredential: The requested credential

    Raises:
    ResourceNotFoundError: If credential does not exist
    HttpResponseError: For other API errors
    """

try:
    credential = client.federated_identity_credentials.get(
        resource_group_name="myResourceGroup",
        resource_name="myUserIdentity",
        federated_identity_credential_resource_name="k8s-default-workload"
    )
    
    print(f"Credential found: {credential.name}")
    print(f"Issuer: {credential.issuer}")
    print(f"Subject: {credential.subject}")
    print(f"Audiences: {credential.audiences}")
    
except ResourceNotFoundError:
    print("Federated credential not found")

def delete(
    resource_group_name: str,
    resource_name: str,
    federated_identity_credential_resource_name: str,
    **kwargs
) -> None:
    """
    Delete a federated identity credential.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the user-assigned identity
    - federated_identity_credential_resource_name (str): Name of the federated credential
    - **kwargs: Additional request options

    Returns:
    None

    Raises:
    ResourceNotFoundError: If credential does not exist
    HttpResponseError: For other API errors
    """

try:
    client.federated_identity_credentials.delete(
        resource_group_name="myResourceGroup",
        resource_name="myUserIdentity",
        federated_identity_credential_resource_name="old-k8s-credential"
    )
    print("Federated credential deleted successfully")
    
except ResourceNotFoundError:
    print("Credential not found")
except HttpResponseError as e:
    print(f"Failed to delete credential: {e.message}")

def list(
    resource_group_name: str,
    resource_name: str,
    top: Optional[int] = None,
    skiptoken: Optional[str] = None,
    **kwargs
) -> ItemPaged[FederatedIdentityCredential]:
    """
    List federated identity credentials for an identity.

    Parameters:
    - resource_group_name (str): Name of the resource group
    - resource_name (str): Name of the user-assigned identity
    - top (Optional[int]): Maximum number of results to return per page
    - skiptoken (Optional[str]): Token for retrieving the next page of results
    - **kwargs: Additional request options

    Returns:
    ItemPaged[FederatedIdentityCredential]: Paginated list of credentials

    Raises:
    ResourceNotFoundError: If identity does not exist
    HttpResponseError: For other API errors
    """

# List all federated credentials for an identity
print("Federated credentials for myUserIdentity:")
for credential in client.federated_identity_credentials.list(
    resource_group_name="myResourceGroup",
    resource_name="myUserIdentity"
):
    print(f"- {credential.name}")
    print(f"  Issuer: {credential.issuer}")
    print(f"  Subject: {credential.subject}")
    print(f"  Audiences: {', '.join(credential.audiences)}")

# Count credentials
credential_count = len(list(client.federated_identity_credentials.list(
    resource_group_name="myResourceGroup",
    resource_name="myUserIdentity"
)))
print(f"Total federated credentials: {credential_count}")

def setup_kubernetes_workload_identity(
    resource_group: str,
    identity_name: str, 
    oidc_issuer: str,
    namespace: str,
    service_account: str
):
    """
    Set up complete Kubernetes workload identity federation.
    
    Args:
        resource_group: Azure resource group name
        identity_name: User-assigned identity name
        oidc_issuer: AKS OIDC issuer URL
        namespace: Kubernetes namespace
        service_account: Kubernetes service account name
    
    Returns:
        tuple: (client_id, credential_name)
    """
    # First ensure the user-assigned identity exists
    try:
        identity = client.user_assigned_identities.get(
            resource_group_name=resource_group,
            resource_name=identity_name
        )
    except ResourceNotFoundError:
        print(f"Creating user-assigned identity: {identity_name}")
        identity = client.user_assigned_identities.create_or_update(
            resource_group_name=resource_group,
            resource_name=identity_name,
            parameters={
                "location": "eastus",
                "tags": {"workload-identity": "true", "namespace": namespace}
            }
        )
    
    # Create federated credential for the service account
    credential_name = f"{namespace}-{service_account}"
    credential = client.federated_identity_credentials.create_or_update(
        resource_group_name=resource_group,
        resource_name=identity_name,
        federated_identity_credential_resource_name=credential_name,
        parameters={
            "properties": {
                "issuer": oidc_issuer,
                "subject": f"system:serviceaccount:{namespace}:{service_account}",
                "audiences": ["api://AzureADTokenExchange"]
            }
        }
    )
    
    return identity.client_id, credential_name

# Set up workload identity for a Kubernetes service account
client_id, cred_name = setup_kubernetes_workload_identity(
    resource_group="myResourceGroup",
    identity_name="aks-workload-identity",
    oidc_issuer="https://oidc.prod-aks.azure.com/tenant-guid/issuer-guid",
    namespace="default",
    service_account="workload-service-account"
)

print(f"Workload identity configured:")
print(f"  Client ID: {client_id}")
print(f"  Credential: {cred_name}")
print(f"  Add this annotation to your service account:")
print(f"  azure.workload.identity/client-id: {client_id}")

def setup_github_actions_identity(
    resource_group: str,
    identity_name: str,
    github_repo: str,
    branch_or_environment: str = "main",
    environment_type: str = "branch"  # or "environment"
):
    """
    Set up federated identity for GitHub Actions.
    
    Args:
        resource_group: Azure resource group
        identity_name: User-assigned identity name
        github_repo: GitHub repository in format "owner/repo"
        branch_or_environment: Branch name or environment name
        environment_type: "branch" or "environment"
    
    Returns:
        dict: Configuration details
    """
    # Construct subject based on type
    if environment_type == "environment":
        subject = f"repo:{github_repo}:environment:{branch_or_environment}"
        credential_name = f"github-env-{branch_or_environment}"
    else:
        subject = f"repo:{github_repo}:ref:refs/heads/{branch_or_environment}"
        credential_name = f"github-branch-{branch_or_environment}"
    
    # Create or get identity
    try:
        identity = client.user_assigned_identities.get(
            resource_group_name=resource_group,
            resource_name=identity_name
        )
    except ResourceNotFoundError:
        identity = client.user_assigned_identities.create_or_update(
            resource_group_name=resource_group,
            resource_name=identity_name,
            parameters={
                "location": "eastus",
                "tags": {"github-actions": "true", "repo": github_repo}
            }
        )
    
    # Create federated credential
    credential = client.federated_identity_credentials.create_or_update(
        resource_group_name=resource_group,
        resource_name=identity_name,
        federated_identity_credential_resource_name=credential_name,
        parameters={
            "properties": {
                "issuer": "https://token.actions.githubusercontent.com",
                "subject": subject,
                "audiences": ["api://AzureADTokenExchange"]
            }
        }
    )
    
    return {
        "client_id": identity.client_id,
        "tenant_id": identity.tenant_id,
        "credential_name": credential_name,
        "subject": subject,
        "setup_instructions": {
            "AZURE_CLIENT_ID": identity.client_id,
            "AZURE_TENANT_ID": identity.tenant_id,
            "AZURE_SUBSCRIPTION_ID": client.subscription_id
        }
    }

# Set up GitHub Actions for main branch
github_config = setup_github_actions_identity(
    resource_group="myResourceGroup",
    identity_name="github-actions-identity",
    github_repo="myorg/myrepo",
    branch_or_environment="main"
)

print("GitHub Actions Configuration:")
print("Add these secrets to your GitHub repository:")
for key, value in github_config["setup_instructions"].items():
    print(f"  {key}: {value}")

def setup_multi_environment_federation(
    resource_group: str,
    base_identity_name: str,
    environments: dict
):
    """
    Set up federated credentials for multiple environments.
    
    Args:
        resource_group: Azure resource group
        base_identity_name: Base name for identities
        environments: Dict with env_name -> config mapping
    
    Example environments config:
    {
        "dev": {
            "issuer": "https://dev-oidc.example.com",
            "subject": "workload-dev-123"
        },
        "staging": {
            "issuer": "https://staging-oidc.example.com", 
            "subject": "workload-staging-456"
        },
        "prod": {
            "issuer": "https://prod-oidc.example.com",
            "subject": "workload-prod-789"
        }
    }
    """
    results = {}
    
    for env_name, config in environments.items():
        identity_name = f"{base_identity_name}-{env_name}"
        
        # Create identity for environment
        try:
            identity = client.user_assigned_identities.create_or_update(
                resource_group_name=resource_group,
                resource_name=identity_name,
                parameters={
                    "location": "eastus",
                    "tags": {"environment": env_name, "federation": "true"}
                }
            )
            
            # Create federated credential
            credential = client.federated_identity_credentials.create_or_update(
                resource_group_name=resource_group,
                resource_name=identity_name,
                federated_identity_credential_resource_name=f"{env_name}-federation",
                parameters={
                    "properties": {
                        "issuer": config["issuer"],
                        "subject": config["subject"],
                        "audiences": ["api://AzureADTokenExchange"]
                    }
                }
            )
            
            results[env_name] = {
                "identity_name": identity_name,
                "client_id": identity.client_id,
                "principal_id": identity.principal_id,
                "credential_name": credential.name,
                "status": "success"
            }
            
        except Exception as e:
            results[env_name] = {
                "identity_name": identity_name,
                "status": "failed",
                "error": str(e)
            }
    
    return results

# Configure multiple environments
environments = {
    "dev": {
        "issuer": "https://dev-k8s.example.com",
        "subject": "system:serviceaccount:dev:app-service-account"
    },
    "prod": {
        "issuer": "https://prod-k8s.example.com",
        "subject": "system:serviceaccount:prod:app-service-account"
    }
}

env_results = setup_multi_environment_federation(
    resource_group="myResourceGroup",
    base_identity_name="myapp-identity",
    environments=environments
)

for env, result in env_results.items():
    if result["status"] == "success":
        print(f"✓ {env}: {result['client_id']}")
    else:
        print(f"✗ {env}: {result['error']}")

def validate_federated_credential(
    resource_group: str,
    identity_name: str,
    credential_name: str
):
    """
    Validate federated credential configuration and provide troubleshooting info.
    """
    try:
        # Get the credential
        credential = client.federated_identity_credentials.get(
            resource_group_name=resource_group,
            resource_name=identity_name,
            federated_identity_credential_resource_name=credential_name
        )
        
        # Get the parent identity
        identity = client.user_assigned_identities.get(
            resource_group_name=resource_group,
            resource_name=identity_name
        )
        
        validation_result = {
            "valid": True,
            "credential_name": credential.name,
            "identity_client_id": identity.client_id,
            "issuer": credential.issuer,
            "subject": credential.subject,
            "audiences": credential.audiences,
            "validation_checks": []
        }
        
        # Validation checks
        if not credential.issuer.startswith("https://"):
            validation_result["validation_checks"].append(
                "WARNING: Issuer should use HTTPS"
            )
        
        if "api://AzureADTokenExchange" not in credential.audiences:
            validation_result["validation_checks"].append(
                "WARNING: Missing standard audience 'api://AzureADTokenExchange'"
            )
        
        if not credential.subject:
            validation_result["validation_checks"].append(
                "ERROR: Subject cannot be empty"
            )
            validation_result["valid"] = False
        
        # Add setup instructions based on issuer pattern
        if "token.actions.githubusercontent.com" in credential.issuer:
            validation_result["setup_type"] = "GitHub Actions"
            validation_result["env_vars"] = {
                "AZURE_CLIENT_ID": identity.client_id,
                "AZURE_TENANT_ID": identity.tenant_id,
                "AZURE_SUBSCRIPTION_ID": client.subscription_id
            }
        elif "oidc.prod-aks.azure.com" in credential.issuer:
            validation_result["setup_type"] = "AKS Workload Identity"
            validation_result["service_account_annotation"] = {
                "azure.workload.identity/client-id": identity.client_id
            }
        
        return validation_result
        
    except ResourceNotFoundError as e:
        return {
            "valid": False,
            "error": f"Resource not found: {e}",
            "troubleshooting": [
                "Check that the identity exists",
                "Check that the credential name is correct",
                "Verify resource group name"
            ]
        }

# Validate a federated credential
validation = validate_federated_credential(
    resource_group="myResourceGroup",
    identity_name="myUserIdentity",
    credential_name="k8s-default-workload"
)

if validation["valid"]:
    print("✓ Credential is valid")
    if "env_vars" in validation:
        print("Environment variables:")
        for key, value in validation["env_vars"].items():
            print(f"  {key}={value}")
else:
    print("✗ Credential validation failed")
    print(f"Error: {validation.get('error', 'Unknown error')}")

class FederatedIdentityCredential:
    """Federated identity credential for workload identity federation."""
    # Read-only properties
    id: str                    # Full Azure resource ID
    name: str                  # Credential resource name
    type: str                  # Resource type (Microsoft.ManagedIdentity/userAssignedIdentities/federatedIdentityCredentials)
    system_data: SystemData    # ARM metadata (creation/modification info)
    
    # Configurable properties
    issuer: str               # OIDC issuer URL (required) - must be HTTPS
    subject: str              # External identity identifier (required)
    audiences: List[str]      # Token audiences (required) - typically includes "api://AzureADTokenExchange"