Como faço para remover automaticamente usuários inativos do Centro de Identidade do IAM?

Breve descrição

Para remover automaticamente usuários inativos do Centro de identidade do IAM, crie um perfil de execução para que o AWS Lambda realize ações para você. Em seguida, crie uma função do Lambda e uma regra do Amazon EventBridge para que a função seja executada em um cronograma especificado.

Resolução

Crie um perfil de execução

Conclua as etapas a seguir:

1. Use o console do AWS Identity and Access Management (AWS IAM) para criar o perfil de execução.
2. Adicione as seguintes permissões à política do perfil do 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": "*"
           }
       ]
   }

As permissões na declaração de política anterior permitem que o perfil de execução realize as seguintes ações:

• Verificar o CloudTrail para ver os eventos UserAuthentication para identificar usuários que não fazem login há 90 dias.
• Marcar os usuários como inativos se eles não tiveram nenhum evento UserAuthentication nos últimos 90 dias.
• Adicionar a uma fila de exclusão usuários inativos e novos usuários que não tenham feito login no portal de acesso do Centro de Identidade do AWS IAM.
• Verificar as atribuições da conta da AWS ou aplicações para cada usuário inativo.
• Remover as atribuições e, em seguida, excluir o usuário.

Importante:

• se você usar um provedor de identidade externo como sua origem de identidade, deve excluir usuários não autorizados da aplicação SAML do Centro de Identidade do AWS no nível do provedor de identidade externo.
• Se você usar o Active Directory como sua origem de identidade, deve remover os usuários de destino e suas associações a grupos associados do escopo de sincronização.

Criar uma função do Lambda e anexe o perfil de execução

Use o console do Lambda para criar a função do Lambda. Em Runtime, selecione Python 3.13. No editor de código integrado, insira o seguinte código 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 você cria uma função do Lambda, o Lamba cria um perfil de execução com permissões mínimas. Atualize sua função para usar o perfil de execução que você criou.

Crie uma regra programada do EventBridge

Conclua as etapas a seguir:

1. Abra o console do Amazon EventBridge.
2. No painel de navegação, selecione Regras e, em seguida, clique em Criar regra.
3. Insira um nome e uma descrição.<br id=hardline_break/> Observação: uma regra não pode ter o mesmo nome de outra regra na mesma região da AWS e no mesmo barramento de eventos.
4. Em Barramento de eventos, selecione Barramento de eventos padrão da AWS.
5. Em Tipo de regra, selecione Cronograma.
6. Clique em Próximo.
7. Em Padrão de cronograma, selecione Cronograma recorrente.
8. Em Tipo de cronograma, clique em Cronograma baseado em Cron.
9. Em Expressão cron, especifique cron(0 0 1 * ? *) para ser executado mensalmente.<br id=hardline_break/> Observação: para obter mais informações sobre os valores cron, consulte Expressão cron.
10. Clique em Próximo.
11. Selecione AWS Lambda como destino.
12. Selecione sua função do Lambda.
13. Clique em Próximo.
14. Consulte e selecione Criar regra.

Teste a regra programada do EventBridge

Depois de criar sua regra programada, teste a regra para garantir que a automação funcione. Defina uma expressão cron para após apenas alguns minutos do horário atual.

Por exemplo, se você iniciar o teste às 11:57 AM, defina a expressão como cron(59 11 * * ? *) para iniciar às 11:59.

Depois de confirmar que sua automação funciona, modifique a expressão cron da sua regra para o cronograma de produção.

Informações relacionadas

Best practices for security, identity, and compliance (Práticas recomendadas de segurança, identidade e conformidade)

Referência de padrões para o CSPM do Security Hub

Gerencie sua origem de identidade

Revise e remova regularmente usuários, perfis, permissões, políticas e credenciais não utilizados