IAM USER로 로그인 하면 CloudWatch에 Log 기록하기

 

CloudTrail 만들어줍니다.

CloudTrail에서 Cloudwatch LogGroup을 만들어서 연결해주고

Cloudwatch LogGroup을 트리거로 한 Lambda함수를 생성해줍니다.

 

Lambda에서 LOG_GROUP_NAME, LOG_STREAM_NAME 변수에 IAM USER 로그인

로그를 저장할 로그 그룹 이름과 로그 스트림 이름을 지정해줍니다.

 

로그인 하기 전에 LOG_GROUP_NAME 의 이름을 가진 로그 그룹을 미리 만들어야 합니다.

 

Lambda Code    Python

import boto3
import base64
import zlib
import json
from datetime import datetime

def lambda_handler(event, context):
    # CloudWatch Logs 그룹 및 스트림 설정
    LOG_GROUP_NAME = '<로그 그룹 이름>'
    LOG_STREAM_NAME = '<로그 스트림 이름>'
    
    logs_client = boto3.client('logs')
    try:  # 로그 스트림이 없다면 만들기
        logs_client.create_log_stream(logGroupName=LOG_GROUP_NAME, logStreamName=LOG_STREAM_NAME)
    except logs_client.exceptions.ResourceAlreadyExistsException:
        pass
    
    event_data = event['awslogs']['data']  # 이벤트 데이터 추출
    decoded_data = base64.b64decode(event_data)  # base64 디코딩
    decompressed_data = zlib.decompress(decoded_data, 16 + zlib.MAX_WBITS)  # 압축 해제
    log_data = json.loads(decompressed_data)  # JSON 파싱
    log_events = log_data['logEvents']  # 로그 이벤트 추출
    
    LOG_EVENT = []
    for log_event in log_events:
        log_message = json.loads(log_event['message'])
        if log_message['eventName'] == 'ConsoleLogin' and log_message['userIdentity']['type'] == 'IAMUser':  # IAM USER 로그인 이벤트 확인 시
            # eventTime을 정수형 UNIX 타임스탬프로 변환
            timestamp = int(datetime.strptime(log_message['eventTime'], '%Y-%m-%dT%H:%M:%SZ').timestamp() * 1000)
            # 로그 스트림에 전송할 메시지
            message = json.dumps({
                'USER': f"{log_message['userIdentity']['userName']} has logged in!"
            })
            LOG_EVENT.append({'timestamp': timestamp, 'message': message})
    
    if LOG_EVENT:  # 로그 스트림에 메시지 보내기
        logs_client.put_log_events(
            logGroupName=LOG_GROUP_NAME,
            logStreamName=LOG_STREAM_NAME,
            logEvents=LOG_EVENT
        )

    return {'statusCode': 200, 'body': json.dumps('Processed successfully')}