Wie entferne ich automatisch inaktive IAM-Identity-Center-Benutzer?

Kurzbeschreibung

Um inaktive IAM-Identity-Center-Benutzer automatisch zu entfernen, erstelle eine Ausführungsrolle, damit AWS Lambda Aktionen für dich ausführt. Erstelle dann eine Lambda-Funktion und eine Amazon-EventBridge-Regel, damit die Funktion nach einem bestimmten Zeitplan ausgeführt wird.

Lösung

Eine Ausführungsrolle erstellen

Gehe wie folgt vor:

1. Verwende die AWS Identity and Access Management (IAM)-Konsole, um die Ausführungsrolle zu erstellen.
2. Füge der Richtlinie der IAM-Rolle die folgenden Berechtigungen hinzu:

   {
       "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": "*"
           }
       ]
   }

Die Berechtigungen in der vorherigen Richtlinienanweisung ermöglichen es der Ausführungsrolle, die folgenden Aktionen durchzuführen:

• Suche in CloudTrail nach UserAuthentication-Ereignissen, um Benutzer zu identifizieren, die sich seit 90 Tagen nicht angemeldet haben.
• Markiere Benutzer als inaktiv, wenn sie in den letzten 90 Tagen kein UserAuthentication-Ereignis hatten.
• Füge einer Löschwarteschlange inaktive Benutzer und neue Benutzer hinzu, die sich nicht beim Zugriffsportal des AWS IAM Identity Centers angemeldet haben.
• Suche nach AWS-Konto- oder Anwendungszuweisungen für alle inaktiven Benutzer.
• Entferne die Zuweisungen und lösche dann den/die Benutzer:in.

Wichtig:

• Wenn du einen externen Identitätsanbieter als Identitätsquelle verwendest, musst du nicht autorisierte Benutzer aus der SAML-Anwendung des AWS Identity Centers auf Ebene des externen Identitätsanbieters löschen. 
• Wenn du Active Directory als Identitätsquelle verwendest, musst du die Zielbenutzer und ihre zugehörigen Gruppenmitgliedschaften aus dem Synchronisierungsbereich entfernen. 

Eine Lambda-Funktion erstellen und die Ausführungsrolle anfügen

Verwende die Lambda-Konsole, um die Lambda-Funktion zu erstellen. Wähle für Laufzeit die Option Python 3.13 aus. Gib im integrierten Code Editor den folgenden Python-Code ein:

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

Wenn du eine Lambda-Funktion erstellst, erstellt Lamba eine Ausführungsrolle mit Mindestberechtigungen. Aktualisiere deine Funktion, um die von dir erstellte Ausführungsrolle zu verwenden.

Eine geplante EventBridge-Regel erstellen

Gehe wie folgt vor:

1. Öffne die Amazon-EventBridge-Konsole.
2. Wähle im Navigationsbereich Regeln und dann Regel erstellen aus.
3. Gib einen Namen und eine Beschreibung ein.<br id=hardline_break/> Hinweis: Eine Regel kann nicht denselben Namen wie eine andere Regel in derselben AWS-Region und auf demselben Event Bus haben.
4. Wähle für Event Bus die Option AWS Standard-Event-Bus aus.
5. Wähle als Typ der Regel die Option Zeitplan aus.
6. Wähle Nächster Schritt aus.
7. Wähle für Zeitplanmuster die Option Wiederkehrender Zeitplan aus.
8. Wähle als Zeitplantyp die Option Cron-basierter Zeitplan aus.
9. Gib als Cron-Ausdruck cron(0 0 1 * ? *) ein, damit die Ausführung monatlich erfolgt.<br id=hardline_break/> Hinweis: Informationen zu Cron-Werten findest du unter Cron-Ausdrücke.
10. Wähle Nächster Schritt aus.
11. Wähle AWS Lambda als Ziel aus.
12. Wähle deine Lambda-Funktion aus.
13. Wähle Nächster Schritt aus.
14. Überprüfe und wähle Regel erstellen aus.

Geplante EventBridge-Regel testen

Nachdem du deine geplante Regel erstellt hast, teste die Regel, um sicherzustellen, dass die Automatisierung funktioniert. Stelle einen Cron-Ausdruck für eine Zeit ein, die nur wenige Minuten von der aktuellen Uhrzeit entfernt ist.

Wenn du den Test beispielsweise um 11.57 Uhr startest, setze den Ausdruck auf cron(59 11 * * ? *), um den Test um 11.59 Uhr zu beginnen. 

Nachdem du dich vergewissert hast, dass deine Automatisierung funktioniert, passe den Cron-Ausdruck deiner Regel an deinen Produktionsplan an.

Ähnliche Informationen

Bewährte Methoden für Sicherheit, Identität und Compliance

Standardreferenz für Security Hub CSPM

Identitätsquelle verwalten

Regelmäßig ungenutzte Benutzer, Rollen, Berechtigungen, Richtlinien und Anmeldeinformationen überprüfen und entfernen