BLOG
지난 1월 5일 AWS는 CloudTrail Lake를 발표하였는데요, 이는 CloudTrail 로그를 저장 및 분석하기 위한 완전 관리형 데이터 레이크입니다.
오늘은 저희 OCTO팀이 이 CloudTrail Lake를 직접 테스트해보고 정리한 포스팅을 준비했습니다. AWS CloudTrail Lake의 A to Z를 샅샅이 파헤쳐 볼까요?😃
CloudTrail Lake 개요
AWS에서 클라우드 인프라를 관리하는 조직에는 보안 및 규정 준수를 위해 AWS 계정의 운영을 감사할 수 있는 효과적인 메커니즘이 필요합니다. AWS는 2013년 11월에 감사 플랫폼인 AWS CloudTrail을 발표했고, 이후 많은 사용자가 이 서비스를 도입했습니다.
하지만 CloudTrail은 아래와 같은 단점이 있었습니다.
✅AWS 계정에서 발생한 지난 90일간의 활동만 확인할 수 있음. (Event History에서 확인)
✅단일 AWS 계정으로 제한되고, 단일 Region의 이벤트만 반환하며 여러 속성을 쿼리할 수 없음.
✅S3 등의 데이터 이벤트는 표시되지 않음.
이러한 이슈로 인해 사용자는 Athena 테이블을 작성해 장기간의 이벤트에 대해서 SQL을 통한 분석 환경을 구현하기도 했습니다. 그러나 드디어 AWS가 2022년 1월 5일, 이벤트를 aggregate, immutably store 및 query 할 수 있는 Managed data lake인 AWS CloudTrail Lake의
GA(general availability)를 발표했습니다. CloudTrail Lake는 AWS의 여러 Region과 계정의 이벤트에 대해 수집, 저장, 최적화 및 쿼리 할 수 있도록 하나의 환경으로 통합함으로써 이벤트 분석을 단순화하였고, 별도의 데이터 처리 파이프라인도 필요하지 않습니다. 또한, 사용자에게 친숙한 SQL을 사용하여 조회할 수 있다는 점도 큰 장점입니다(Query 작성에 도움이 되는 Sample queries 포함). 이로써 사용자는 CloudTrail Lake에 의해 SQL을 통한 간편한 분석 환경을 가질 수 있게 되었습니다.
CloudTrail의 이벤트
CloudTrail 이벤트는 AWS 계정의 activity 기록입니다. 이 activity는 CloudTrail에서 모니터링할 수 있는 User, Role 또는 Service가 수행하는 Action입니다. CloudTrail 이벤트는 AWS Management Console, AWS SDK, 명령줄 도구 및 기타 AWS 서비스를 통해 수행된 API 계정 활동과 비API 계정 활동의 기록을 모두 제공합니다. 다만, 모든 AWS 서비스와 모든 이벤트를 기록하지는 않으니 자세한 내용은 설명서를 참조해주세요. CloudTrail에 기록할 수 있는 이벤트에는 관리 이벤트, 데이터 이벤트 등이 있는데요, 밑에서 더 자세히 살펴보겠습니다.
관리 이벤트
관리 이벤트는 AWS 계정의 Resource에 대해 수행되는 관리 작업에 대한 정보를 말하며, Control Plane Operations라고도 합니다.
- 보안 구성 (IAM의 AttachRolePolicy API 등)
- 디바이스 등록 (Amazon EC2 CreateDefualtVpc API 등)
- 데이터 라우팅 규칙 구성 (Amazon EC2 CreateSubnet API 등)
- 로깅 설정 (AWS CloudTrail API 등)
관리 이벤트에는 AWS 계정에서 발생한 비 API 이벤트도 포함될 수 있습니다.
(로그인 기록: ConsoleLogin )
데이터 이벤트
Resource 상에서 혹은 Resource 내에서 수행되는 Resource 작업에 대한 정보를 말하며, Data Plane Operations라고도 합니다.
- 버킷 및 버킷의 객체의 Amazon S3 객체 수준 API 활동(예: GetObject , DeleteObject , PutObject API 작업)
- AWS Lambda 함수 실행 활동( Invoke API)
- 테이블의 Amazon DynamoDB 객체 수준 API 활동(예: PutItem , DeleteItem , UpdateItem API 작업)
- Outposts의 Amazon S3 객체 수준 API 활동
- Ethereum 노드의 Amazon Managed Blockchain JSON-RPC 호출(예: eth_getBalance 또는 eth_getBlockByNumber )
- Amazon S3 객체 Lambda 액세스 포인트의 API 활동(예: CompleteMultipartUpload 및 GetObject 에 대한 호출)
- Amazon Elastic Block Store(EBS) 다이렉트 API(예: Amazon EBS 스냅샷의 PutSnapshotBlock , GetSnapshotBlock , ListChangedBlocks )
- 액세스 포인트에 대한 Amazon S3 API 활동
- 스트림의 Amazon DynamoDB API 활동
- 테이블에 대한 AWS Glue API 활동
CloudTrail Lake 특징
✅CloudTrail 이벤트에 대해 SQL 기반 Query실행
✅이벤트는 이벤트 데이터 스토어에 수집되며 최대 7년(2,555일) 보유
✅여러 Region, 계정 및 조직 내 이벤트 집계 가능
✅Query 결과는 최대 7일까지 표시 가능
비용
CloudTrail Lake는 사용한 만큼만 비용을 지불하며, 최소 요금이 없습니다.
- 수집 및 저장
- 처음 5TB까지 $2.5/GB
- 다음 20TB까지 $1/GB
- 25TB 이상 $0.5/GB
- 분석
- 스캔한 데이터에 대해 $0.005/GB
신규 고객은 추가 비용 없이 30일 동안 CloudTrail Lake를 사용해 볼 수 있으며, 이 기간 동안 전체 기능에 액세스할 수 있지만, 수집 5GB, 조회 5GB의 데이터로 제한됩니다.
Region
다음은 2022.01.05 기준, CloudTrail Lake를 사용할 수 있는 Region 입니다.
- US East (N. Virginia), US East (Ohio), US West (N. California), US West (Oregon)
- Canada (Central)
- Europe (Ireland), Europe (London), Europe (Paris), Europe (Frankfurt), Europe (Stockholm), Europe (Milan)
- Asia Pacific (Tokyo), Asia Pacific (Seoul), Asia Pacific (Osaka), Asia Pacific (Singapore), Asia Pacific (Sydney), Asia Pacific (Mumbai), Asia
Pacific (Hong Kong) - South America (Sao Paulo)
- Middle East (Bahrain), and South Africa (Cape Town)
사용 시 주의할 점
✅조회 시 보존 기간(최대 7년간)의 모든 데이터에 대해 실행되며, eventTime 등을 이용해 조회 범위를 지정해서 사용하면 조회 성능을 높이고 비용을 줄일 수 있습니다.
✅CloudTrail Lake SQL은 SELECT 문만 사용할 수 있습니다.
사용해보기
이벤트 데이터 저장소 생성
다음 단계를 사용하여 CloudTrail Lake를 활성화하고 이벤트 데이터 저장소를 생성하면 이벤트 발생 시 자동으로 해당 저장소에 저장됩니다. CloudTrail Lake를 사용하지 않으려면 이벤트 데이터 저장소를 삭제하기만 하면 됩니다. Event Data Store 탭에서 삭제하려는 저장소를 선택한 후 Termination protection 을 Disabled 로 변경하고 Delete 합니다. 이렇게 하면 데이터 저장소가 비활성화되고 7일 후에 영구적으로 삭제됩니다.
1. AWS 콘솔을 열고 AWS CloudTrail을 관리할 수 있는 관리 권한이 있는 계정으로 로그인.
2. CloudTrail 콘솔로 이동합니다. 왼쪽 탐색 메뉴에서 Lake를 선택.
3. Event Data Store 탭 선택.
4. Create event data store 버튼을 선택
- Step 1. Configure event data store
-
- Event data store name 입력 (예: “MyNewDataStore”)
- Retention period 입력 (7일 ~ 2,555일)
- 데이터 원본에 현재 Region만 포함할지 여부를 선택
- (Optional) 조직의 모든 계정을 포함하려면 선택. (AWS Organizations 환경에만 적용됨)
- (Optional) 태그 입력. (자세한 내용은 AWS 리소스에 태그 지정을 참조)
▶️Next 선택
- Step 2. Choose events
-
- Event type을 선택합니다.
- 관리 이벤트( Management events )를 선택한 경우
-
- 읽기( Read ) 및 쓰기( Write ) 이벤트 여부를 선택합니다.
- AWS Key Management Service(KMS) 및 AWS RDS 데이터 API 이벤트를 제외할지 여부를 선택합니다.
- 데이터 이벤트( Data events )를 선택한 경우
-
- Data event type을 선택합니다.
-
- Log selector template을 선택합니다.
(Optional) Selector name을 입력합니다.
Add data event type 을 선택하여 필요한 데이터 이벤트를 추가합니다. Next를 선택하세요.
- Step 3. Review and create
Review Page에서 구성한 옵션이 올바른지 확인 후 Create event data store 선택합니다.
조회
새로 생성한 이벤트 데이터 저장소는 Event data stores 목록에서 상태를 볼 수 있습니다. 몇 분 후에 데이터 저장소가 시작되고 Editor 탭에서 조회할 수 있습니다. FROM 절에는 이벤트 데이터 저장소 ID를 입력합니다. SELECT List 절에는 좌측의 Event properties 를 참고하여 필요한 Column 들을 콤마(,)로 구분하여 1개 이상 입력합니다.
- 전체 데이터 조회하기
- SELECT *
- FROM XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
- 전체 데이터 중 10건만 조회하기
- SELECT *
- FROM XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
- LIMIT 10
- 조회 기간을 지정하여 조회하기
- SELECT *
- FROM XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
- WHERE eventTime >= ‘2022-01-05 00:00:00’ AND eventTime <= ‘2022-01-07 00:00:00’
- LIMIT 10
- 특정 Region만 조회하기
- SELECT *
- FROM XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
- WHERE eventTime >= ‘2022-01-05 00:00:00’ AND eventTime <= ‘2022-01-07 00:00:00’
- AND awsRegion IN (‘ap-northeast-2’)
- LIMIT 10
- 저장된 마지막 이벤트의 시간과 현재 시간 비교하기
- SELECT MAX(eventTime )
- FROM XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
CloudTrail은 일반적으로 API 호출 후 평균 15분 이내에 로그를 전달합니다. 자세한 내용은 AWS CloudTrail 서비스 수준 계약에서 검토하세요.
모니터링 및 Alarm
현재 CloudTrail Lake는 CloudWatch Logs 등 다른 서비스를 통해 Alarm을 연동할 방법을 제공하고 있지 않습니다. CloudTrail Lake는 따로 사용자 영역에 관련 data 가 저장되지 않는 것으로 확인되며, SELECT Query 이상의 로직을 구현해야 한다면 AWS SDK, 명령줄 도구를 이용하면 가능합니다.
❗그러나 저희는 비용적인 측면에서 추천하지는 않는데요, 그 이유는 기존에 CloudTrail Logs 또는 AWS SDK 를 활용하고 있다면 중복해서 사용할 필요는 없기 때문입니다.
❗CloudTrail Lake의 경우 최근(2022.01.05)에 출시한 서비스 이므로 AWS SDK(boto3 등), 명령줄 도구(aws cli) 버전이 최신 버전인지 확인이 필요합니다.
boto3를 이용하여 CloudTrail Lake 활용
- 모니터링 주제
AWS 콘솔에 접속한 사용자가 허용된 ip ( ex. 211.60.xxx.xxx )에서 접속하지 않았을 경우 username과 region, 로그인 여부, 시도한 IP 정보를 추출하여 slack을 통해 전송합니다.
- boto3 를 활용한 CloudTrail Function 호출 순서는 아래와 같습니다.
list_event_data_stores() > start_query() > get_query_results() > Slack Alarm
-
- list_event_data_stores() : event data stores 정보 추출
- start_query : lake 조회 , select 문
return value 는 QueryId - get_query_results : QueryId 를 통해 결과를 가져옴,
❗주의할 점은 start_query 호출 후 쿼리가 실행 중 이더라도 바로 return 되므로 일정 시간 대기 후 get_query_results 호출이 필요합니다.
- CloudTrail Lake Query
SELECT
eventTime, eventName, userIdentity.arn, userIdentity.username, sourceIPAddress, responseElements, awsRegion
FROM
dd9b02cd-cf28-4ad1-xxxxxxxxxx << event data store ID
WHERE
eventTime > ‘2022-01-10 04:26:59.713433’
AND sourceIPAddress <> ‘211.60.50.190’
AND awsRegion in (‘us-east-1’, ‘us-west-2′,’ap-northeast-2’)
AND eventName = ‘ConsoleLogin’
- 스크립트 source
……
endtime = datetime.datetime.now()
interval = datetime.timedelta(hours=3)
starttime = endtime – interval
cloudtrail = boto3.client(‘cloudtrail’,’%s’ % ‘ap-northeast-2’)
response1 = cloudtrail.list_event_data_stores(
MaxResults=123
)
datastore_arn = response1[“EventDataStores”][0][“EventDataStoreArn”]
arn_id=response1[“EventDataStores”][0][“EventDataStoreArn”].split(‘/’)
Query = “SELECT eventTime, eventName, userIdentity.arn, userIdentity.username, sourceIPAddress, responseElements, awsRegion FROM %s WHERE eventTime > ‘%s’ AND sourceIPAddress <> ‘211.60.50.190’ AND awsRegion in (‘us-east-1’, ‘us-west-2′,’ap-northeast-2’) AND eventName = ‘ConsoleLogin'” % (arn_id[1],starttime)
response = cloudtrail.start_query(
QueryStatement = “%s” % str(Query)
)
### 쿼리 실행 결과 값은 실행이 완전히 끝난 후 리턴이 되므로 대기가 필요함
sleep(5)
response2 = cloudtrail.get_query_results(
EventDataStore=’%s’ % datastore_arn,
QueryId=’%s’ % response[“QueryId”] ,
MaxQueryResults=123
)
쿼리 실행 결과 리스트가 하나라도 있으면 값 추출 후 slack webhook 호출
if response2[“QueryStatistics”][“ResultsCount”] != 0:
for rows in response2[“QueryResultRows”]:
user = “”
for row in rows:
print(row)
if row.get(‘sourceIPAddress’):
alert_message = “Access from not permmitted IP [%s].\n” % str(row.get(‘sourceIPAddress’))
src_ip = str(row.get(‘sourceIPAddress’))
elif row.get(‘awsRegion’):
Region = row.get(‘awsRegion’)
elif row.get(‘username’):
user = row.get(‘username’)
elif row.get(‘arn’):
user = row.get(‘arn’)
elif row.get(‘eventTime’):
e_time = row.get(‘eventTime’)
elif row.get(‘eventName’):
e_name = row.get(‘eventName’)
elif row.get(‘responseElements’):
reponse_element = row.get(‘responseElements’)
### Slack webhook 호출
Slack_alert(alert_message,Region,user,e_time,e_name,src_ip,reponse_element)
……
- Result :
aws cli를 이용하여 CloudTrail Lake 활용
- 모니터링 주제
CloudTrail Lake의 가장 최신 이벤트의 시간과 현재 시간을 비교하고, 가장 최신 이벤트의 시간부터 최근 10분 동안에 5회 이상 ConsoleLogin Failure로 인해 ‘Failed authentication’ 메시지를 응답하는 이벤트가 있으면 slack을 통해 Source IP, Region, Event Time을 전송
합니다.
- 최신 버전의 aws cli 이용
- $ aws –version
- aws-cli/2.4.9 Python/3.8.8 Linux/5.4.0-1029-aws exe/x86_64.ubuntu.20 prompt/off
- 가장 최신 이벤트의 시간 조회하는 Query 시작하기
- $ aws cloudtrail start-query \
- –query-statement ‘SELECT MAX(eventTime) AS rslt FROM dd9b02cd-cf28-4ad1-xxxx-xxxxxxxxxx’
-
- output
-
- {
- “QueryId”: “28afed14-5651-420b-9b25-38e6638cbcf2”
- }
- QueryId를 이용하여 결과 조회하기
- $ aws cloudtrail get-query-results \
- –event-data-store dd9b02cd-cf28-4ad1-xxxx-xxxxxxxxxx \
- –query-id 28afed14-5651-420b-9b25-38e6638cbcf2
-
- output
1 {
2 “QueryStatus”: “FINISHED”,
3 “QueryStatistics”: {
4 “ResultsCount”: 1,
5 “TotalResultsCount”: 1
6 },
7 “QueryResultRows”: [
8 [
9 {
10 “rslt”: “2022-01-11 00:27:12.000”
11 }
12 ]
13 ]
14 }
- 가장 최신 이벤트로부터 10분 동안 ConsoleLogin Failure인 이벤트(responseElements: ‘{ConsoleLogin=Failure}’) 최근 5건만 조회하는 Query 시작하기
1 aws cloudtrail start-query \
2 –query-statement \
3 “SELECT eventTime,
4 awsRegion,
5 sourceIPAddress
6 FROM dd9b02cd-cf28-4ad1-xxxx-xxxxxxxxxx
7 WHERE eventType = ‘AwsConsoleSignIn’
8 AND element_at(responseElements,’ConsoleLogin’) = ‘Failure’
9 AND eventTime > ‘2022-01-11T00:17:12.000Z’ AND eventTime < ‘2022-01-11T00:27:12.000Z’
10 ORDER BY eventTime DESC
11 LIMIT 5”
-
- output
1 {
2 “QueryId”: “d6820b50-b5fb-4b6a-bf06-85e8a424d94d”
3 }
- QueryId를 이용하여 결과 조회하기
1 $ aws cloudtrail get-query-results \
2 –event-data-store dd9b02cd-cf28-4ad1-xxxx-xxxxxxxxxx \
3 –query-id d6820b50-b5fb-4b6a-bf06-85e8a424d94d
-
- output
1 {
2 “QueryStatus”: “FINISHED”,
3 “QueryStatistics”: {
4 “ResultsCount”: 5,
5 “TotalResultsCount”: 5
6 },
7 “QueryResultRows”: [……]
8 }
- 조회 결과(QueryResultRows)가 5건 이상인 경우 slack을 통해 Source IP, Region, Event Time을 전송하기
위 포스팅은 메가존클라우드 OCTO Team에서 직접 테스트한 내용을 바탕으로 하였습니다.