AWS CloudTrail 로그 쿼리 - Amazon Athena

기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

AWS CloudTrail 로그 쿼리

AWS CloudTrail는 AWS API 호출 및 Amazon Web Services 계정 이벤트를 기록하는 서비스입니다.

CloudTrail 로그에는 콘솔을 포함하여 사용자에 대한 모든 API 호출에 대한 세부 정보가 포함됩니다. AWS 서비스 CloudTrail 암호화된 로그 파일을 생성하여 Amazon S3에 저장합니다. 자세한 정보는 AWS CloudTrail 사용 설명서를 참조하세요.

참고

계정, 지역 및 날짜의 CloudTrail 이벤트 정보에 대해 SQL 쿼리를 수행하려면 CloudTrail Lake를 사용하는 것이 좋습니다. CloudTrail Lake는 기업의 정보를 검색 가능한 단일 이벤트 데이터 저장소로 집계하는 트레일을 생성하는 AWS 대신 사용할 수 있습니다. 이 솔루션은 Amazon S3 버킷 스토리지를 사용하지 않고 대신 이벤트를 데이터 레이크에 저장하므로 더 풍부하고 빠른 쿼리가 가능합니다. 이를 사용하여 여러 조직, 지역 및 사용자 지정 시간 범위 내에서 이벤트를 검색하는 SQL 쿼리를 만들 수 있습니다. CloudTrail 콘솔 내에서 CloudTrail Lake 쿼리를 수행하므로 CloudTrail Lake를 사용할 때는 Athena가 필요하지 않습니다. 자세한 내용은 CloudTrail Lake 설명서를 참조하십시오.

Athena를 CloudTrail 로그와 함께 사용하면 활동 분석을 향상시킬 수 있는 강력한 방법입니다. AWS 서비스 예를 들어 쿼리를 사용하여 트렌드를 식별하고 소스 IP 주소나 사용자 등의 속성별로 활동을 추가로 격리할 수 있습니다.

일반적인 응용 분야는 CloudTrail 로그를 사용하여 보안 및 규정 준수를 위한 운영 활동을 분석하는 것입니다. 자세한 예제는 AWS Big Data Blog 게시물 Analyze Security, Compliance, and Operational Activity Using AWS CloudTrail and Amazon Athena를 참조하세요.

Athena를 사용하여 로그 파일의 LOCATION을 지정하면 Amazon S3에서 직접 이러한 로그 파일을 쿼리할 수 있습니다. 이 작업을 두 가지 방법으로 수행할 수 있습니다.

  • CloudTrail 콘솔에서 직접 CloudTrail 로그 파일용 테이블을 생성함으로써 가능합니다.

  • Athena 콘솔에서 CloudTrail 로그 파일에 대한 테이블을 수동으로 생성합니다.

CloudTrail 로그 및 Athena 테이블 이해

테이블 생성을 시작하기 전에 테이블 데이터 저장 CloudTrail 방법과 이에 대해 조금 더 이해해야 합니다. 이렇게 하면 테이블을 CloudTrail 콘솔에서 생성하든 Athena에서 생성하든 관계없이 필요한 테이블을 생성할 수 있습니다.

CloudTrail 로그를 압축된 gzip 형식 (*.json.gzip) 의 JSON 텍스트 파일로 저장합니다. 로그 파일의 위치는 추적 설정 방법, AWS 리전 또는 로깅 중인 리전 및 기타 요인에 따라 다릅니다.

로그가 저장되는 위치, JSON 구조 및 레코드 파일 내용에 대한 자세한 내용은 AWS CloudTrail 사용 설명서의 다음 주제를 참조하세요.

로그를 수집하여 Amazon S3에 저장하려면 CloudTrail 에서 활성화하십시오AWS Management Console. 자세한 내용은 AWS CloudTrail 사용 설명서추적 생성을 참조하세요.

로그가 저장되는 대상 Amazon S3 버킷을 기록해 둡니다. LOCATION절을 CloudTrail 로그 위치의 경로와 작업할 객체 세트로 대체하십시오. 이 예에서는 특정 계정에 대해 로그의 LOCATION 값을 사용하지만 애플리케이션에 적합한 정도의 특정성을 사용할 수 있습니다.

다음 예를 참조하세요.

  • 여러 계정의 데이터를 분석하려면 LOCATION 's3://MyLogFiles/AWSLogs/'를 이용해 LOCATION 범위 지정자를 롤백하여 모든 AWSLogs를 표시할 수 있습니다.

  • 특정 날짜, 계정 및 리전의 데이터를 분석하려면 LOCATION 's3://MyLogFiles/123456789012/CloudTrail/us-east-1/2016/03/14/'.를 사용합니다.

객체 계층 구조에서 최상위 수준을 사용하면 Athena로 쿼리할 때 가장 큰 유연성을 발휘할 수 있습니다.

CloudTrail 콘솔을 사용하여 로그용 Athena 테이블 생성 CloudTrail

콘솔에서 직접 로그를 CloudTrail 쿼리하기 위해 파티션을 나누지 않은 Athena 테이블을 생성할 수 있습니다. CloudTrail CloudTrail 콘솔에서 Athena 테이블을 생성하려면 Athena에서 테이블을 생성할 수 있는 충분한 권한이 있는 역할로 로그인해야 합니다.

참고

CloudTrail 콘솔을 사용하여 조직 트레일 로그에 대한 Athena 테이블을 만들 수는 없습니다. 대신 Athena 콘솔을 사용하여 테이블을 수동으로 만들고 올바른 저장 위치를 지정할 수 있습니다. 조직 추적에 대한 자세한 내용은 AWS CloudTrail 사용 설명서에서 조직에 대한 추적 생성을 참조하세요.

콘솔을 사용하여 CloudTrail 트레일에 대한 Athena 테이블을 만들려면 CloudTrail
  1. https://console.aws.amazon.com/cloudtrail/ 에서 CloudTrail 콘솔을 엽니다.

  2. 탐색 창에서 Event history(이벤트 기록)를 선택합니다.

  3. Athena 테이블 생성을 선택합니다.

    
                        Athena 테이블 생성 선택
  4. 스토리지 위치에서, 아래쪽 화살표를 사용하여 추적이 쿼리할 로그 파일이 저장되는 Amazon S3 버킷을 선택합니다.

    참고

    트레일과 연결된 버킷의 이름을 찾으려면 CloudTrail 탐색 창에서 Trails를 선택하고 트레일의 S3 버킷 열을 확인하십시오. 버킷의 Amazon S3 위치를 보려면 S3 버킷 열에서 버킷에 대한 링크를 선택합니다. 그러면 Amazon S3 콘솔이 열리고 CloudTrail 버킷 위치가 표시됩니다.

  5. 테이블 생성을 선택합니다. Amazon S3 버킷 이름을 포함하는 기본 이름을 가진 테이블이 생성됩니다.

수동 파티셔닝을 사용하여 Athena에서 CloudTrail 로그용 테이블 만들기

Athena 콘솔에서 수동으로 CloudTrail 로그 파일 테이블을 만든 다음 Athena에서 쿼리를 실행할 수 있습니다.

Athena 콘솔을 사용하여 CloudTrail 트레일에 대한 Athena 테이블을 만들려면
  1. 다음 DDL 문을 복사하여 Athena 콘솔 쿼리 편집기에 붙여 넣습니다.

    CREATE EXTERNAL TABLE cloudtrail_logs ( eventversion STRING, useridentity STRUCT< type:STRING, principalid:STRING, arn:STRING, accountid:STRING, invokedby:STRING, accesskeyid:STRING, userName:STRING, sessioncontext:STRUCT< attributes:STRUCT< mfaauthenticated:STRING, creationdate:STRING>, sessionissuer:STRUCT< type:STRING, principalId:STRING, arn:STRING, accountId:STRING, userName:STRING>, ec2RoleDelivery:string, webIdFederationData:map<string,string> > >, eventtime STRING, eventsource STRING, eventname STRING, awsregion STRING, sourceipaddress STRING, useragent STRING, errorcode STRING, errormessage STRING, requestparameters STRING, responseelements STRING, additionaleventdata STRING, requestid STRING, eventid STRING, resources ARRAY<STRUCT< arn:STRING, accountid:STRING, type:STRING>>, eventtype STRING, apiversion STRING, readonly STRING, recipientaccountid STRING, serviceeventdetails STRING, sharedeventid STRING, vpcendpointid STRING, eventCategory STRING, tlsDetails struct< tlsVersion:string, cipherSuite:string, clientProvidedHostHeader:string> ) PARTITIONED BY (region string, year string, month string, day string) ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe' STORED AS INPUTFORMAT 'com.amazon.emr.cloudtrail.CloudTrailInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' LOCATION 's3://CloudTrail_bucket_name/AWSLogs/Account_ID/CloudTrail/';
    참고

    예제에 org.apache.hive.hcatalog.data.JsonSerDe 표시된 것을 사용하는 것이 좋습니다. a가 com.amazon.emr.hive.serde.CloudTrailSerde 존재하긴 하지만 현재 일부 새 CloudTrail 필드를 처리하지 못합니다.

  2. (선택 사항) 테이블에 필요하지 않은 필드를 제거합니다. 특정 열 집합만 읽어야 하는 경우 테이블 정의에서 다른 열을 제외할 수 있습니다.

  3. 로그 데이터가 포함된 Amazon S3 버킷을 가리키도록 s3://CloudTrail_bucket_name/AWSLogs/Account_ID/CloudTrail/을 수정합니다.

  4. 필드가 올바르게 나열되는지 확인합니다. 레코드의 전체 필드 목록에 대한 자세한 내용은 CloudTrail CloudTrail 레코드 콘텐츠를 참조하십시오.

    다음 예제에서는 Hive JSON SerDe을(를) 사용합니다. 이 예에서 requestparameters, responseelements, additionaleventdata 필드는 쿼리에서 STRING 형식으로 나열되지만, JSON에서는 STRUCT 데이터 형식이 사용됩니다. 따라서 이러한 필드에서 데이터를 가져오려면 JSON_EXTRACT 함수를 사용합니다. 자세한 설명은 JSON에서 데이터 추출 섹션을 참조하세요. 성능 향상을 위해 예제에서는 데이터를 AWS 리전, 연도, 월 및 일별로 분할합니다.

  5. Athena 콘솔에서 CREATE TABLE 문을 실행합니다.

  6. 다음 예제와 같이 ALTER TABLE ADD PARTITION 명령을 사용하여 파티션을 쿼리할 수 있도록 파티션을 로드합니다.

    ALTER TABLE table_name ADD PARTITION (region='us-east-1', year='2019', month='02', day='01') LOCATION 's3://cloudtrail_bucket_name/AWSLogs/Account_ID/CloudTrail/us-east-1/2019/02/01/'

수동 분할을 사용하여 조직 전체 추적을 위한 테이블 생성

Athena에서 조직 전체 CloudTrail 로그 파일에 대한 테이블을 만들려면 의 단계를 따르되 다음 절차에 설명된 대로 수정하십시오. 수동 파티셔닝을 사용하여 Athena에서 CloudTrail 로그용 테이블 만들기

조직 전체 로그에 대한 Athena 테이블을 만들려면 CloudTrail
  1. CREATE TABLE 문에서 다음 예시와 같이 조직 ID를 포함하도록 LOCATION 절을 수정합니다.

    LOCATION 's3://cloudtrail_bucket_name/AWSLogs/organization_id/Account_ID/CloudTrail/'
  2. PARTITIONED BY 절에서 다음 예시와 같이 계정 ID에 대한 항목을 문자열로 추가합니다.

    PARTITIONED BY (account string, region string, year string, month string, day string)

    다음 예시는 결합된 결과를 보여줍니다.

    ... PARTITIONED BY (account string, region string, year string, month string, day string) ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe' STORED AS INPUTFORMAT 'com.amazon.emr.cloudtrail.CloudTrailInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' LOCATION 's3://cloudtrail_bucket_name/AWSLogs/organization_id/Account_ID/CloudTrail/'
  3. ALTER TABLEADD PARTITION 절에서 다음 예시와 같이 계정 ID를 포함합니다.

    ALTER TABLE table_name ADD PARTITION (account='111122223333', region='us-east-1', year='2022', month='08', day='08')
  4. ALTER TABLELOCATION 절에서 다음 예시와 같이 조직 ID, 계정 ID 및 추가할 파티션을 포함합니다.

    LOCATION 's3://cloudtrail_bucket_name/AWSLogs/organization_id/Account_ID/CloudTrail/us-east-1/2022/08/08/'

    다음 예시 ALTER TABLE 문은 결합된 결과를 보여줍니다.

    ALTER TABLE table_name ADD PARTITION (account='111122223333', region='us-east-1', year='2022', month='08', day='08') LOCATION 's3://cloudtrail_bucket_name/AWSLogs/organization_id/111122223333/CloudTrail/us-east-1/2022/08/08/'

파티션 프로젝션을 사용하여 Athena에서 CloudTrail 로그에 대한 테이블 생성

CloudTrail 로그에는 파티션 구성표를 미리 지정할 수 있는 알려진 구조가 있으므로 Athena 파티션 프로젝션 기능을 사용하여 쿼리 런타임을 줄이고 파티션 관리를 자동화할 수 있습니다. 새 데이터가 추가되면 파티션 프로젝션은 자동으로 새 파티션을 추가합니다. 따라서 ALTER TABLE ADD PARTITION을 사용해 파티션을 수동으로 추가할 필요가 없습니다.

다음 예제 CREATE TABLE 명령문에서는 로그에 대해 지정된 날짜부터 현재까지의 CloudTrail 로그에 파티션 프로젝션을 자동으로 사용합니다. AWS 리전 LOCATIONstorage.location.template 절에서 bucket, account-idaws-region 자리 표시자를 상응하는 동일한 값으로 바꿉니다. projection.timestamp.range에 대해 2020/01/01을 사용하려는 시작 날짜로 바꿉니다. 쿼리가 성공적으로 실행된면 테이블을 쿼리할 수 있습니다. 파티션을 로드하기 위해 ALTER TABLE ADD PARTITION을 실행하지 않아도 됩니다.

CREATE EXTERNAL TABLE cloudtrail_logs_pp( eventVersion STRING, userIdentity STRUCT< type: STRING, principalId: STRING, arn: STRING, accountId: STRING, invokedBy: STRING, accessKeyId: STRING, userName: STRING, sessionContext: STRUCT< attributes: STRUCT< mfaAuthenticated: STRING, creationDate: STRING>, sessionIssuer: STRUCT< type: STRING, principalId: STRING, arn: STRING, accountId: STRING, userName: STRING>, ec2RoleDelivery:string, webIdFederationData:map<string,string> > >, eventTime STRING, eventSource STRING, eventName STRING, awsRegion STRING, sourceIpAddress STRING, userAgent STRING, errorCode STRING, errorMessage STRING, requestparameters STRING, responseelements STRING, additionaleventdata STRING, requestId STRING, eventId STRING, readOnly STRING, resources ARRAY<STRUCT< arn: STRING, accountId: STRING, type: STRING>>, eventType STRING, apiVersion STRING, recipientAccountId STRING, serviceEventDetails STRING, sharedEventID STRING, vpcendpointid STRING, eventCategory STRING, tlsDetails struct< tlsVersion:string, cipherSuite:string, clientProvidedHostHeader:string> ) PARTITIONED BY ( `timestamp` string) ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe' STORED AS INPUTFORMAT 'com.amazon.emr.cloudtrail.CloudTrailInputFormat' OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat' LOCATION 's3://bucket/AWSLogs/account-id/CloudTrail/aws-region' TBLPROPERTIES ( 'projection.enabled'='true', 'projection.timestamp.format'='yyyy/MM/dd', 'projection.timestamp.interval'='1', 'projection.timestamp.interval.unit'='DAYS', 'projection.timestamp.range'='2020/01/01,NOW', 'projection.timestamp.type'='date', 'storage.location.template'='s3://bucket/AWSLogs/account-id/CloudTrail/aws-region/${timestamp}')

파티션 프로젝션에 대한 자세한 내용은 Amazon Athena를 사용한 파티션 프로젝션 단원을 참조하세요.

중첩된 필드 쿼리

userIdentityresources 필드는 중첩된 데이터 형식이기 때문에 이를 쿼리하려면 특수한 처리가 필요합니다.

userIdentity 객체는 중첩된 STRUCT 형식으로 구성됩니다. 다음 예에서처럼 점으로 필드를 구분하여 이 객체들을 쿼리할 수 있습니다.

SELECT eventsource, eventname, useridentity.sessioncontext.attributes.creationdate, useridentity.sessioncontext.sessionissuer.arn FROM cloudtrail_logs WHERE useridentity.sessioncontext.sessionissuer.arn IS NOT NULL ORDER BY eventsource, eventname LIMIT 10

resources 필드는 STRUCT 객체의 배열입니다. 이러한 배열의 경우 CROSS JOIN UNNEST를 사용해 배열의 중첩을 해제함으로써 해당 객체를 쿼리할 수 있습니다.

다음 예제는 리소스 ARN이 example/datafile.txt로 끝나는 모든 행을 반환합니다. replace 함수는 가독성을 위해 ARN에서 초기 arn:aws:s3::: 하위 문자열을 제거합니다.

SELECT awsregion, replace(unnested.resources_entry.ARN,'arn:aws:s3:::') as s3_resource, eventname, eventtime, useragent FROM cloudtrail_logs t CROSS JOIN UNNEST(t.resources) unnested (resources_entry) WHERE unnested.resources_entry.ARN LIKE '%example/datafile.txt' ORDER BY eventtime

다음 예제에서는 DeleteBucket 이벤트를 쿼리합니다. 이 쿼리는 resources 객체에서 버킷의 이름 및 버킷이 속한 계정 ID를 추출합니다.

SELECT awsregion, replace(unnested.resources_entry.ARN,'arn:aws:s3:::') as deleted_bucket, eventtime AS time_deleted, useridentity.username, unnested.resources_entry.accountid as bucket_acct_id FROM cloudtrail_logs t CROSS JOIN UNNEST(t.resources) unnested (resources_entry) WHERE eventname = 'DeleteBucket' ORDER BY eventtime

중첩 해제에 대한 자세한 내용은 배열 필터링 단원을 참조하세요.

쿼리 예

다음 예제는 CloudTrail 이벤트 로그용으로 만든 테이블에서 모든 익명 (서명되지 않은) 요청을 반환하는 쿼리의 일부를 보여줍니다. 이 쿼리는 useridentity.accountid가 익명이고 useridentity.arn이 지정되지 않은 요청을 선택합니다.

SELECT * FROM cloudtrail_logs WHERE eventsource = 's3.amazonaws.com' AND eventname in ('GetObject') AND useridentity.accountid = 'anonymous' AND useridentity.arn IS NULL AND requestparameters LIKE '%[your bucket name ]%';

자세한 내용은 AWS Big Data Blog 게시물 Analyze Security, Compliance, and Operational Activity Using AWS CloudTrail and Amazon Athena를 참조하세요.

로그 쿼리 CloudTrail 팁

CloudTrail 로그 데이터를 탐색하려면 다음 팁을 사용하세요.

  • 로그를 쿼리하기 전에 로그 테이블이 수동 파티셔닝을 사용하여 Athena에서 CloudTrail 로그용 테이블 만들기의 로그 테이블과 동일한 형태인지 확인합니다. 첫 번째 테이블이 아닌 경우 DROP TABLE cloudtrail_logs 명령을 사용하여 기존 테이블을 삭제합니다.

  • 기존 테이블을 삭제하고 다시 만드세요. 자세한 설명은 수동 파티셔닝을 사용하여 Athena에서 CloudTrail 로그용 테이블 만들기 섹션을 참조하세요.

    Athena 쿼리의 필드가 올바르게 나열되는지 확인합니다. 레코드의 전체 필드 목록에 대한 자세한 내용은 CloudTrail CloudTrail 레코드 콘텐츠를 참조하십시오.

    쿼리에 STRUCT 같은 JSON 형식의 필드가 포함된 경우, JSON에서 데이터를 추출합니다. 자세한 설명은 JSON에서 데이터 추출 섹션을 참조하세요.

    CloudTrail 테이블에 대해 쿼리를 실행하기 위한 몇 가지 제안:

  • 먼저 어떤 사용자가 어떤 API 작업을 호출했으며 어떤 소스 IP 주소에서 시작되었는지 확인합니다.

  • 다음 기본 SQL 쿼리를 템플릿으로 사용합니다. 쿼리를 Athena 콘솔에 붙여 넣고 실행합니다.

    SELECT useridentity.arn, eventname, sourceipaddress, eventtime FROM cloudtrail_logs LIMIT 100;
  • 데이터를 추가 탐색하려면 쿼리를 수정합니다.

  • 성능을 향상시키려면 LIMIT 절을 포함시켜 지정한 행 하위 집합을 반환합니다.