Comment supprimer automatiquement les utilisateurs inactifs d'IAM Identity Center ?

Brève description

Pour supprimer automatiquement les utilisateurs inactifs d'IAM Identity Center, créez un rôle d'exécution permettant à AWS Lambda d'effectuer des actions à votre place. Puis, créez une fonction Lambda et une règle Amazon EventBridge pour que la fonction s'exécute selon une planification spécifiée.

Résolution

Créer un rôle d’exécution

Procédez comme suit :

1. Utilisez la console Gestion des identités et des accès AWS (AWS IAM) pour créer le rôle d'exécution.
2. Ajoutez les autorisations suivantes à la politique du rôle IAM :

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "VisualEditor0",
               "Effect": "Allow",
               "Action": [
                   "cloudtrail:LookupEvents",
                   "sso:ListAccountAssignments",
                   "sso:ListPermissionSets",
                   "organizations:ListAccounts",
                   "sso:ListInstances",
                   "sso:DeleteAccountAssignment",
                   "identitystore:DeleteUser",
                   "sso-directory:ListUsers",
                   "sso-directory:DeleteUser",
                   "logs:CreateLogGroup",
                   "logs:CreateLogStream",
                   "logs:PutLogEvents",
                   "identitystore:ListUsers"
               ],
               "Resource": "*"
           }
       ]
   }

Les autorisations indiquées dans l’instruction de politique précédente permettent au rôle d'exécution d'effectuer les actions suivantes :

• Consultez CloudTrail pour les événements UserAuthentication afin d'identifier les utilisateurs qui ne se sont pas connectés depuis 90 jours.
• Marquez les utilisateurs comme inactifs s’ils n’ont subi aucun événement UserAuthentication au cours des 90 derniers jours.
• Ajoutez à une file d'attente de suppression les utilisateurs inactifs et les nouveaux utilisateurs qui ne se sont pas connectés au portail d'accès à AWS IAM Identity Center.
• Vérifiez les attributions de compte ou d'application AWS pour chaque utilisateur inactif.
• Supprimez les affectations, puis supprimez l'utilisateur.

Important :

• si vous utilisez un fournisseur d'identité externe comme source d'identité, vous devez supprimer les utilisateurs non autorisés de l'application SAML AWS Identity Center au niveau du fournisseur d'identité externe.
• Si vous utilisez Active Directory comme source d'identité, vous devez supprimer les utilisateurs cibles et les appartenances aux groupes qui leur sont associées de la zone de synchronisation.

Créer une fonction Lambda et associer le rôle d'exécution

Utilisez la console Lambda pour créer la fonction Lambda. Dans Exécution, sélectionnez Python 3.13. Dans l'éditeur de code intégré, saisissez le code Python suivant :

import json
from datetime import datetime, timedelta
import boto3
from botocore.exceptions import ClientError
# Create boto3 clients to call AWS services used by the script.
sso_admin = boto3.client("sso-admin")
identitystore = boto3.client("identitystore")
cloudtrail = boto3.client("cloudtrail")
org = boto3.client("organizations")
# Threshold to consider a user "active": any authentication after THRESHOLD_DATE counts.
THRESHOLD_DATE = datetime.utcnow() - timedelta(days=90)
def get_identity_center_info():
    """
    Retrieve the first Identity Center (SSO) instance's ARN and IdentityStoreId.
    """
    resp = sso_admin.list_instances()                         
    inst = resp.get("Instances") or []                     
    if not inst:
        raise RuntimeError("No Identity Center instance found")  
    # Return the InstanceArn and IdentityStoreId of the first instance found
    return inst[0]["InstanceArn"], inst[0]["IdentityStoreId"]
def list_users(identity_store_id):
    """
    List all users from the given Identity Store (IdentityCenter).
    Returns a list of users as returned by identitystore.list_users.
    """
    users = []
    # Use a paginator to handle large user directories
    for page in identitystore.get_paginator("list_users").paginate(IdentityStoreId=identity_store_id):
        users.extend(page.get("Users", []))                  
    return users
def build_active_sets():
    """
    Scan CloudTrail LookupEvents for UserAuthentication events in the time window
    and build two sets:
      - active_ids: set of userId values found at userIdentity.onBehalfOf.userId
      - active_names: set of usernames found at additionalEventData.UserName
    Scanning CloudTrail once is more efficient than querying per user.
    """
    active_ids = set()    # seen onBehalfOf.userId values
    active_names = set()  # seen additionalEventData.UserName values
    paginator = cloudtrail.get_paginator("lookup_events")
    try:
        # Iterate through pages of UserAuthentication events within the timeframe
        for page in paginator.paginate(
            LookupAttributes=[{"AttributeKey": "EventName", "AttributeValue": "UserAuthentication"}],
            StartTime=THRESHOLD_DATE,
            EndTime=datetime.utcnow(),
        ):
            for ev in page.get("Events", []):               
                cte = ev.get("CloudTrailEvent")            
                try:
                    # If CloudTrailEvent is a string, parse to dict; otherwise use as-is or empty dict
                    detail = json.loads(cte) if isinstance(cte, str) else (cte or {})
                except (ValueError, TypeError):
                    # Skip malformed or unexpected event content
                    continue
                # Extract the userId from userIdentity.onBehalfOf.userId if present (preferred)
                uid = detail.get("userIdentity", {}).get("onBehalfOf", {}).get("userId")
                # Extract username from additionalEventData.UserName as a fallback
                uname = detail.get("additionalEventData", {}).get("UserName")
                if uid:
                    active_ids.add(uid)                    
                if uname:
                    active_names.add(uname)                 
    except ClientError as e:
        # Bubble up failures in CloudTrail lookup as RuntimeError so caller can handle/log
        raise RuntimeError(f"CloudTrail lookup failed: {e}")
    return active_ids, active_names
def list_accounts():
    """
    Return a list of AWS Account IDs in the Organization.
    Uses organizations.list_accounts paginator to handle many accounts.
    """
    accounts = []
    for page in org.get_paginator("list_accounts").paginate():
        accounts.extend([a["Id"] for a in page.get("Accounts", [])])  
    return accounts
def list_permission_sets(instance_arn):
    """
    Return a list of PermissionSet ARNs for the Identity Center instance.
    Uses sso-admin.list_permission_sets paginator to support many permission sets.
    """
    perms = []
    for page in sso_admin.get_paginator("list_permission_sets").paginate(InstanceArn=instance_arn):
        perms.extend(page.get("PermissionSets", []))
    return perms
def remove_user_assignments(instance_arn, principal_id):
    """
    Remove all SSO account assignments for the given principal_id (user).
    Iterates every account and permission set, lists assignments, and deletes any
    assignment that matches the user PrincipalId and is of type USER.
    Returns True if no deletion errors occurred, False otherwise.
    """
    accounts = list_accounts()                         
    perms = list_permission_sets(instance_arn)           
    success = True
    for acct in accounts:
        for perm in perms:
            try:
                # List assignments for the (account, permission set) pair
                paginator = sso_admin.get_paginator("list_account_assignments")
                for page in paginator.paginate(InstanceArn=instance_arn, AccountId=acct, PermissionSetArn=perm):
                    for a in page.get("AccountAssignments", []):
                        # If this assignment is for the user, delete it
                        if a.get("PrincipalType") == "USER" and a.get("PrincipalId") == principal_id:
                            sso_admin.delete_account_assignment(
                                InstanceArn=instance_arn,
                                TargetId=acct,
                                TargetType="AWS_ACCOUNT",
                                PermissionSetArn=perm,
                                PrincipalType="USER",
                                PrincipalId=principal_id,
                            )
                            print(f"Removed assignment: user={principal_id} account={acct} permission_set={perm}")
            except ClientError as e:
                # Log the failure but continue attempting other assignments
                print(f"Warning: failed removing assignments for acct={acct} perm={perm}: {e}")
                success = False
    return success
def delete_user(identity_store_id, user_id):
    """
    Delete the user from the Identity Store using identitystore.delete_user.
    Returns True on success, False on failure.
    """
    try:
        identitystore.delete_user(IdentityStoreId=identity_store_id, UserId=user_id)
        print(f"Deleted user: {user_id}")
        return True
    except ClientError as e:
        print(f"Error deleting user {user_id}: {e}")
        return False
def lambda_handler(event=None, context=None):
    """
    Main entry point: discover Identity Center instance, fetch users,
    build active user sets from CloudTrail, then remove and delete inactive users.
    """
    # Get Identity Center instance ARN and the identity store ID
    instance_arn, identity_store_id = get_identity_center_info()
    # Retrieve all users from the identity store
    users = list_users(identity_store_id)
    print(f"Found {len(users)} users; scanning CloudTrail UserAuthentication events since {THRESHOLD_DATE.isoformat()}")
    # Build sets of active userIds and usernames by scanning CloudTrail once
    active_ids, active_names = build_active_sets()
    inactive_count = deleted_count = 0
    # Iterate through every user in the identity store
    for u in users:
        user_id = u.get("UserId")        # unique identifier for the user in the Identity Store
        user_name = u.get("UserName")    # may be None if not set or not logged in events
        # If user_id was seen in CloudTrail active_ids OR username was seen in active_names, skip deletion
        if (user_id and user_id in active_ids) or (user_name and user_name in active_names):
            # This user authenticated recently; treat as active
            continue
        # Mark user as inactive and process cleanup
        inactive_count += 1
        print(f"\nProcessing inactive user: id={user_id} username={user_name}")
        # Remove assignments across accounts and permission sets
        if remove_user_assignments(instance_arn, user_id):
            # If assignment removal succeeded, delete the user
            if delete_user(identity_store_id, user_id):
                deleted_count += 1
        else:
            # If we couldn't clean up assignments, avoid deleting to prevent orphaned assignments
            print(f"Skipping deletion for {user_id} due to assignment removal failures")
    print(f"\nSummary: inactive={inactive_count} deleted={deleted_count}")
if __name__ == "__main__":
    # Run the handler for local testing or invocation as a script
    lambda_handler()

Lorsque vous créez une fonction Lambda, Lamba crée un rôle d'exécution avec des autorisations minimales. Mettez à jour votre fonction pour utiliser le rôle d'exécution que vous avez créé.

Créer une règle planifiée EventBridge

Procédez comme suit :

1. Ouvrez la console Amazon EventBridge.
2. Dans le volet de navigation, choisissez Règles, puis Créer une règle.
3. Saisissez un nom et une description.<br id=hardline_break/> Remarque : une règle ne peut pas porter le même nom qu'une autre règle dans la même région AWS et sur le même bus d'événements.
4. Dans Bus d’événements, sélectionnez Bus d’événements par défaut AWS.
5. Dans Type de règle, choisissez Planification.
6. Sélectionnez Suivant.
7. Dans Modèle de planification, sélectionnez Planification récurrente.
8. Puis, dans Type de planification, choisissez Planification basée sur cron.
9. Dans Expression cron, spécifiez cron(0 0 1 * ? *) pour une exécution mensuelle.<br id=hardline_break/> Remarque : pour plus d'informations sur les valeurs cron, consultez la section Expressions cron.
10. Sélectionnez Suivant.
11. Sélectionnez AWS Lambda comme cible.
12. Sélectionnez votre fonction Lambda.
13. Sélectionnez Suivant.
14. Vérifiez et choisissez Créer une règle.

Tester la règle planifiée EventBridge

Après avoir créé votre règle planifiée, testez-la pour vous assurer que l'automatisation fonctionne. Définissez une expression cron pour une heure qui n'est qu'à quelques minutes de l'heure courante.

Par exemple, si vous commencez le test à 11h57, définissez l'expression sur cron(59 11 * * ? *) pour démarrer à 11h59.

Après avoir vérifié que votre automatisation fonctionne, modifiez l'expression cron de votre règle en fonction de votre planification de production.

Informations connexes

Bonnes pratiques en matière de sécurité, d'identité et de conformité

Référence des normes pour Security Hub CSPM

Gérer votre source d'identité

Vérifier et supprimer régulièrement les utilisateurs, rôles, autorisations, politiques et informations d'identification inutilisés