Come posso rimuovere automaticamente gli utenti inattivi del Centro identità IAM?

Breve descrizione

Per rimuovere automaticamente gli utenti inattivi del Centro identità IAM, crea un ruolo di esecuzione per consentire ad AWS Lambda di eseguire azioni per te. Quindi crea una funzione Lambda e una regola in Amazon EventBridge per l'esecuzione della funzione in base a una pianificazione specificata.

Risoluzione

Crea un ruolo di esecuzione

Completa i seguenti passaggi:

1. Utilizza la console AWS Identity and Access Management (AWS IAM) per creare il ruolo di esecuzione.
2. Aggiungi le seguenti autorizzazioni alla policy del ruolo 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": "*"
           }
       ]
   }

Le autorizzazioni nella precedente istruzione di policy consentono al ruolo di esecuzione di effettuare le seguenti azioni:

• Controlla gli eventi UserAuthentication in CloudTrail per identificare gli utenti che non hanno effettuato l'accesso per 90 giorni.
• Contrassegna gli utenti come inattivi se non hanno avuto alcun evento UserAuthentication negli ultimi 90 giorni.
• Aggiungi a una coda di eliminazione gli utenti inattivi e i nuovi utenti che non hanno effettuato l'accesso al portale di accesso del Centro identità AWS IAM.
• Verifica le assegnazioni di account o applicazione AWS per ogni utente inattivo.
• Rimuovi le assegnazioni, quindi elimina l'utente.

Importante:

• se utilizzi un gestore dell'identità digitale esterno come origine dell'identità, devi eliminare gli utenti non autorizzati dall'applicazione SAML del Centro identità AWS a livello di gestore dell'identità digitale esterno.
• Se utilizzi Active Directory come origine dell'identità, devi rimuovere gli utenti di destinazione e le loro appartenenze ai gruppi associati dall'ambito di sincronizzazione.

Dopo aver rimosso gli utenti di destinazione e non autorizzati, non puoi rieseguire il loro provisioning.

Crea una funzione Lambda e collega il ruolo di esecuzione

Utilizza la console Lambda per creare la funzione Lambda. Per Runtime, scegli Python 3.13. Nell'editor di codice integrato, inserisci il seguente codice Python:

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()

Quando crei una funzione Lambda, Lamba crea un ruolo di esecuzione con autorizzazioni minime. Aggiorna la funzione per utilizzare il ruolo di esecuzione che hai creato.

Crea una regola pianificata in EventBridge

Completa i seguenti passaggi:

1. Apri la console Amazon EventBridge.
2. Nel pannello di navigazione, scegli Regole, quindi seleziona Crea regola.
3. Inserisci un nome e una descrizione.<br id=hardline_break/> Nota: una regola non può avere lo stesso nome di un'altra regola nella stessa Regione AWS e sullo stesso bus di eventi.
4. Per Bus di eventi, seleziona Bus di eventi predefiniti di AWS.
5. Per Tipo di regola, scegli Pianificazione.
6. Scegli Avanti.
7. Per Modello di pianificazione,scegli Pianificazione ricorrente.
8. Quindi, per Tipo di pianificazione, scegli Pianificazione basata su Cron.
9. Per Espressione Cron, specifica cron(0 0 1 * ? *) per l'esecuzione mensile.<br id=hardline_break/> Nota: per informazioni sui valori Cron, consulta Espressioni Cron.
10. Scegli Avanti.
11. Seleziona AWS Lambda come destinazione.
12. Seleziona la funzione Lambda.
13. Scegli Avanti.
14. Rivedi e scegli Crea regola.

Esegui un test per verificare la regola pianificata in EventBridge

Dopo aver creato la regola pianificata, esegui un test per assicurarti che l'automazione funzioni. Imposta un'espressione Cron per un periodo di pochi minuti dall'ora corrente.

Ad esempio, se inizi il test alle 11:57, imposta l'espressione su cron(59 11 * * ? *) per iniziare alle 11:59.

Dopo aver verificato il funzionamento dell'automazione, modifica l'espressione Cron della regola in base alla pianificazione di produzione.

Informazioni correlate

Best Practices for Security, Identity, & Compliance (Best practice per la sicurezza, l'identità e la conformità)

Riferimento agli standard per Security Hub CSPM

Manage your identity source (Gestione dell'origine dell'identità)

Esaminare e rimuovere regolarmente utenti, ruoli, autorizzazioni, policy e credenziali inutilizzati