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

간략한 설명

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

해결 방법

실행 역할 만들기

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

1. AWS Identity and Access Management(AWS IAM) 콘솔을 사용하여 실행 역할을 만듭니다.
2. 다음 권한을 IAM 역할의 정책에 추가합니다.

   {
     "Version": "2012-10-17",
     "Statement": [
       {
         "Effect": "Allow",
         "Action": [
           "cloudtrail:LookupEvents",
           "organizations:ListAccounts",
           "sso:ListAccountAssignments",
           "sso:ListPermissionSets",
           "sso:ListInstances",
           "sso:DeleteAccountAssignment",
           "logs:CreateLogGroup",
           "logs:CreateLogStream",
           "logs:PutLogEvents",
           "identitystore:ListUsers",
           "identitystore:DeleteUser"
         ],
         "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 boto3
import json
from datetime import datetime, timedelta
from botocore.exceptions import ClientError

# Initialize AWS clients
sso_admin_client = boto3.client('sso-admin')
identitystore_client = boto3.client('identitystore')
cloudtrail_client = boto3.client('cloudtrail')
org_client = boto3.client('organizations')

# Define the time threshold (90 days ago)
THRESHOLD_DATE = datetime.utcnow() - timedelta(days=90)

def get_identity_center_info():
    """Fetch the Identity Center directory ID and Instance ARN."""
    response = sso_admin_client.list_instances()
    if not response['Instances']:
        raise Exception("No Identity Center instance found.")
    instance = response['Instances'][0]
    return instance['InstanceArn'], instance['IdentityStoreId']

def list_users(identity_store_id):
    """List all users in AWS Identity Center."""
    users = []
    paginator = identitystore_client.get_paginator('list_users')
    for page in paginator.paginate(IdentityStoreId=identity_store_id):
        users.extend(page['Users'])
    return users

def check_user_authentication(username):
    """
    Check if a user has UserAuthentication logged in CloudTrail in the last 90 days.
    Returns True if the user has authenticated, otherwise False.
    """
    if not username:
        print("Warning: Username is empty. Skipping authentication check.")
        return False

    try:
        paginator = cloudtrail_client.get_paginator('lookup_events')
        for page in paginator.paginate(
            LookupAttributes=[
                {'AttributeKey': 'EventName', 'AttributeValue': 'UserAuthentication'}
            ],
            StartTime=THRESHOLD_DATE,
            EndTime=datetime.utcnow()
        ):
            for event in page['Events']:
                # Parse the CloudTrail event
                event_detail = json.loads(event['CloudTrailEvent'])
                username_in_event = event_detail.get('additionalEventData', {}).get('UserName')

                # Match the username to confirm the event belongs to the user
                if username_in_event == username:
                    return True
    except Exception as e:
        print(f"Error checking authentication for user '{username}': {e}")
    return False

def find_user_account_assignments(instance_arn, principal_id):
    """Find all account assignments for a specific user."""
    user_assignments = []

    # Get all accounts
    accounts = []
    try:
        paginator = org_client.get_paginator('list_accounts')
        for page in paginator.paginate():
            accounts.extend([account['Id'] for account in page['Accounts']])
    except ClientError as e:
        print(f"Error listing accounts: {e}")
        return user_assignments

    # Get all permission sets
    permission_sets = []
    try:
        paginator = sso_admin_client.get_paginator('list_permission_sets')
        for page in paginator.paginate(InstanceArn=instance_arn):
            permission_sets.extend(page['PermissionSets'])
    except ClientError as e:
        print(f"Error listing permission sets: {e}")
        return user_assignments

    # Check each account and permission set combination for the user
    for account_id in accounts:
        for permission_set_arn in permission_sets:
            try:
                paginator = sso_admin_client.get_paginator('list_account_assignments')
                for page in paginator.paginate(
                    InstanceArn=instance_arn,
                    AccountId=account_id,
                    PermissionSetArn=permission_set_arn
                ):
                    # Filter assignments for the specific user
                    for assignment in page['AccountAssignments']:
                        if assignment['PrincipalId'] == principal_id and assignment['PrincipalType'] == 'USER':
                            user_assignments.append({
                                'AccountId': account_id,
                                'PermissionSetArn': permission_set_arn,
                                'PrincipalId': principal_id
                            })
            except ClientError as e:
                print(f"Error listing assignments for account {account_id}, permission set {permission_set_arn}: {e}")
                continue

    return user_assignments

def remove_user_assignments(instance_arn, assignments):
    """Remove account assignments for a specific user."""
    success = True
    for assignment in assignments:
        try:
            print(f"Removing assignment: Account={assignment['AccountId']}, PermissionSet={assignment['PermissionSetArn']}")
            sso_admin_client.delete_account_assignment(
                InstanceArn=instance_arn,
                TargetId=assignment['AccountId'],
                TargetType='AWS_ACCOUNT',
                PermissionSetArn=assignment['PermissionSetArn'],
                PrincipalType='USER',
                PrincipalId=assignment['PrincipalId']
            )
        except ClientError as e:
            print(f"Error removing assignment: {e}")
            success = False

    return success

def delete_user(identity_store_id, user_id):
    """Delete a user from AWS Identity Center."""
    try:
        identitystore_client.delete_user(
            IdentityStoreId=identity_store_id,
            UserId=user_id
        )
        print(f"Deleted user with ID: {user_id}")
        return True
    except ClientError as e:
        print(f"Error deleting user with ID {user_id}: {e}")
        return False

def lambda_handler(event, context):
    """AWS Lambda entry point."""
    try:
        # Step 1: Get Identity Center directory ID and Instance ARN
        instance_arn, identity_store_id = get_identity_center_info()
        print(f"Found Identity Center instance: {instance_arn}")

        # Step 2: List all users
        users = list_users(identity_store_id)
        print(f"Found {len(users)} users in Identity Center")

        inactive_users = 0
        deleted_users = 0

        for user in users:
            user_name = user.get('UserName')
            user_id = user.get('UserId')

            # Step 3: Check if the user has authenticated recently
            if check_user_authentication(user_name):
                print(f"User '{user_name}' (ID: {user_id}) has authenticated recently. Skipping.")
                continue

            inactive_users += 1
            print(f"\nProcessing inactive user: '{user_name}' (ID: {user_id})")

            # Step 4: Find all account assignments for this user
            user_assignments = find_user_account_assignments(instance_arn, user_id)

            if user_assignments:
                print(f"Found {len(user_assignments)} assignments for user '{user_name}'")

                # Step 5: Remove the user's assignments
                if remove_user_assignments(instance_arn, user_assignments):
                    print(f"Successfully removed all assignments for user '{user_name}'")
                else:
                    print(f"Failed to remove some assignments for user '{user_name}'. Skipping deletion.")
                    continue
            else:
                print(f"No assignments found for user '{user_name}'")

            # Step 6: Delete the user
            if delete_user(identity_store_id, user_id):
                deleted_users += 1

        print(f"\nSummary: Found {inactive_users} inactive users, successfully deleted {deleted_users} users")

    except Exception as e:
        print(f"Error in main execution: {e}")

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

EventBridge 예약된 규칙 만들기

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

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

EventBridge 예약된 규칙 테스트

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

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

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

관련 정보

보안, ID 및 규정 준수를 위한 모범 사례

Security Hub CSPM에 대한 표준 참조

ID 소스 관리

사용하지 않는 사용자, 역할, 권한, 정책 및 자격 증명을 정기적으로 검토하고 제거