GCP 권한 변경시 실시간 알림 설정방안

개요
GCP(Google Cloud Platform) 환경에서 IAM(Identity and Access Management) 권한 변경은 보안 및 거버넌스 측면에서 가장 핵심적인 모니터링 대상입니다.

권한의 오남용이나 잘못된 설정은 데이터 유출 및 리소스 손상으로 직결될 수 있기 때문입니다.

본 가이드는 권한 부여 및 수정 시 발생하는 복잡한 JSON 형식의 감사 로그(Audit Log)를 실시간으로 탐지하고, 이를 운영자가 즉시 인지할 수 있도록 이메일 알림 체계를 구축하는 방안을 작성하였습니다.

 

시스템 구성도 및 워크플로우
본 시스템은 Google Cloud의 Serverless 아키텍처를 활용하여 비용 효율적이고 확장성 있는 실시간 알림 파이프라인을 구성 합니다.

1. 탐지 (Cloud Logging): 조직 또는 프로젝트 단위의 IAM 변경 이벤트(SetIamPolicy)를 실시간 감시합니다. 
2. 라우팅 (Log Sink): 필터링된 특정 로그를 분석을 위해 Pub/Sub 주제로 전송합니다. 
3. 가공 (Cloud Run): Python 기반의 함수를 실행하여 복잡한 JSON 데이터에서 핵심 변경 내역(Deltas)만 추출하고 한 줄 형태의 요약 로그(IAM_ALERT_SUMMARY)를 재발행합니다. 
4. 알림 (Cloud Monitoring): 요약 로그 발생 시 이를 트리거로 하여 관리자에게 즉시 이메일 알림을 발송합니다.

 

기본 로그 탐색기에서 IAM 권한에 대해 확인하는 방법

 

resource.type="organization"
protoPayload.methodName="SetIamPolicy"

 

resource.type을 "organization"가 아닌 "project"로 하면 프로젝트의 권한변경에 한해서만 볼 수 있습니다.

 

 

• 알람방안
  Logs Explorer 로 이동하여 아래의 로그로 검색합니다.

resource.type="project"
severity="NOTICE"
protoPayload.methodName="SetIamPolicy"

프로젝트의 로그 중 iamPolicy에 대한 알림만 조회하는 쿼리입니다.


작업 → 로그싱크 생성을 선택합니다.

- 로그 라우팅 싱크
싱크이름과 설명을 선택

Pub/sub을 만들어야 합니다. 

주제(topic) ID는 제목을 보고 알 수있도록 설정한 후 나머지 옵션은 기본으로 둔채로 넘어갑니다.

로그 라우팅 싱크가 Cloud Pub/Sub으로 보낼 설정이 완료되었는지 최종적으로 점검하고 생성합니다.

• Cloud Run 설정
  함수 작성 → python 선택

함수(function)을 선택하고, 서비스 이름과 리전을 선택합니다. 런타임은 pyton 3.14 이상이면 됩니다.
중요한 서비스와 리소스를 사용하는 서비스는 아니므로, 저는 수동 확장에 인스턴스는 1개만 선택했습니다.

해당 Cloud run이 동작되기 위해서는 트리거가 필요합니다.
트리거 추가를 선택하고, 아래와 같이 Cloud Pub/Sub을 연결해 줍니다.

생성 후 main.py와 requirements.txt를 아래와 같이 입력합니다.
함수 진입점은 "iam_notification_handler” 설정합니다.

• main.py

import base64
import json

def iam_notification_handler(request):
    try:
        request_json = request.get_json(silent=True)
        if not request_json: return "No Data", 200

        # 데이터 추출
        raw_data = request_json.get('message', {}).get('data') or request_json.get('data')
        if not raw_data: return "No Data", 200

        pubsub_message = base64.b64decode(raw_data).decode('utf-8')

        # 1. 실제 IAM 로그(JSON)인 경우 파싱
        try:
            log_data = json.loads(pubsub_message)
            proto = log_data.get('protoPayload', {})

            # 정보 추출
            actor = proto.get('authenticationInfo', {}).get('principalEmail', 'Unknown')
            project = log_data.get('resource', {}).get('labels', {}).get('project_id', 'N/A')

            # 변경 내역(Deltas) 추출
            deltas = (proto.get('serviceData', {}).get('policyDelta', {}).get('bindingDeltas') or 
                      proto.get('metadata', {}).get('request', {}).get('policy', {}).get('bindingDeltas') or [])

            details = []
            for d in deltas:
                action = "✅[추가]" if d.get('action') == 'ADD' else "❌[제거]"
                role = d.get('role', '').split('/')[-1] # roles/owner -> owner
                member = d.get('member', '').split(':')[-1] # user:test@test.com -> test@test.com
                details.append(f"{action} {role}({member})")

            change_str = ", ".join(details) if details else "상세내역없음"

            # 최종 한 줄 로그 출력
            print(f"IAM_ALERT_SUMMARY | 수정자: {actor} | 프로젝트: {project} | 내역: {change_str}")

        # 2. 단순 테스트 문자열인 경우
        except json.JSONDecodeError:
            print(f"IAM_ALERT_SUMMARY | 테스트 메시지 수신: {pubsub_message}")

        return "OK", 200

    except Exception as e:
        print(f"ERROR: {str(e)}")
        return "Internal Error", 500

requirements.txt

requests==2.28.1
functions-framework==3.*

 

• Log확인 로그탐색기에서 아래와 같이 검색 실행

textPayload: "IAM_ALERT_SUMMARY"

아래 화면과 같이 
수정자, 프로젝트, 권한 내역 (추가, 삭제, 수정)을 확인할 수 있습니다.

 

 

• E-mail로 알림보내기
  로그탐색기에서 textPayload: "IAM_ALERT_SUMMARY" 쿼리 실행 후 작업 → 로그 알람 만들기를 선택합니다.

알림 채널에 기존에 등록한 Email을 등록 한 후 저장 합니다.

변경사항 발생시 메일로 발송되며 “VIEW INCIDENT”를 선택합니다.

IAM 권한에 대한 수정 내역을 확인할 수 있습니다.
수정자 : 프로젝트 : 내역 으로 구분됩니다.