비활성 IAM Identity Center 사용자를 자동으로 제거하려면 어떻게 해야 합니까?

7분 분량

90일 동안 로그인하지 않은 AWS IAM Identity Center 사용자를 자동으로 제거하고 싶습니다.

간략한 설명

비활성 IAM Identity Center 사용자를 자동으로 제거하려면 AWS Lambda가 자동으로 작업을 수행할 실행 역할을 만듭니다. 그런 다음 Lambda 함수를 만들고, 이 함수가 지정된 일정에 따라 실행되도록 Amazon EventBridge 규칙을 만듭니다.

해결 방법

실행 역할 만들기

다음 단계를 완료하십시오.

AWS Identity and Access Management(AWS IAM) 콘솔을 사용하여 실행 역할을 만듭니다.

다음 권한을 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": "*"
        }
    ]
}

위 정책 문의 권한은 실행 역할이 다음 작업을 수행할 수 있도록 허용합니다.

CloudTrail에서 UserAuthentication 이벤트를 확인하여 90일 동안 로그인하지 않은 사용자를 식별합니다.
지난 90일 동안 UserAuthentication 이벤트가 없었던 사용자를 비활성 상태로 표시합니다.
비활성 사용자와 AWS IAM Identity Center 액세스 포털에 로그인한 적 없는 신규 사용자를 삭제 대기열에 추가합니다.
각 비활성 사용자의 AWS 계정 또는 애플리케이션 할당을 확인합니다.
할당을 제거한 다음 사용자를 삭제합니다.

중요:

외부 ID 공급자를 ID 소스로 사용하는 경우 AWS Identity Center SAML 애플리케이션의 외부 ID 공급자 수준에서 승인되지 않은 사용자를 삭제해야 합니다.
Active Directory를 ID 소스로 사용하는 경우 동기화 범위에서 대상 사용자 및 관련 그룹 멤버십을 제거해야 합니다.

Lambda 함수를 만들고 실행 역할 연결

Lambda 콘솔을 사용하여 Lambda 함수를 만듭니다. 런타임에서 Python 3.13을 선택합니다. 내장된 코드 편집기에서 다음 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()

Lambda 함수를 만들면 Lamba에서 최소 권한을 가진 실행 역할을 만듭니다. 생성된 실행 역할을 사용하도록 함수를 업데이트합니다.

EventBridge 예약된 규칙 만들기

다음 단계를 완료하십시오.

Amazon EventBridge 콘솔을 엽니다.
탐색 창에서 규칙을 선택한 다음, 규칙 만들기를 선택합니다.
이름 및 설명을 입력합니다.<br id=hardline_break/> 참고: 규칙은 동일한 AWS 리전 및 동일한 이벤트 버스에 있는 다른 규칙과 동일한 이름을 가질 수 없습니다.
이벤트 버스에서 AWS 기본 이벤트 버스를 선택합니다.
규칙 유형에서 일정을 선택합니다.
다음을 선택합니다.
일정 패턴에서 반복 일정을 선택합니다.
일정 유형에서 CRON 기반 일정을 선택합니다.
cron 표현식에서 매달 실행되도록 **cron(0 0 1 * ? *)**을 지정합니다.<br id=hardline_break/> 참고: cron 값에 대한 자세한 내용은 cron 표현식을 참조하십시오.
다음을 선택합니다.
AWS Lambda를 대상으로 선택합니다.
Lambda 함수를 선택합니다.
다음을 선택합니다.
검토 후 규칙 만들기를 선택합니다.

EventBridge 예약된 규칙 테스트

예약된 규칙을 만든 후 테스트하여 자동화가 제대로 작동하는지 확인합니다. 현재 시간으로부터 몇 분 후에 시작되도록 cron 표현식을 설정합니다.

예를 들어 오전 11시 57분에 테스트를 시작하는 경우 오전 11시 59분에 시작되도록 표현식을 **cron(59 11 * * ? *)**으로 설정합니다.

자동화가 제대로 작동하는지 확인한 후 규칙의 cron 표현식을 프로덕션 일정에 맞게 수정합니다.