AWS CloudTrail 로그 쿼리 - Amazon Athena
CloudTrail 로그 및 Athena 테이블 이해CloudTrail 콘솔을 사용하여 CloudTrail 로그용 Athena 테이블 생성 수동 분할을 사용하여 Athena에서 CloudTrail 로그용 테이블 생성수동 분할을 사용하여 조직 전체 추적을 위한 테이블 생성파티션 프로젝션을 사용하여 Athena에서 CloudTrail 로그용 테이블 생성중첩된 필드 쿼리쿼리 예CloudTrail 로그 쿼리 팁

AWS CloudTrail 로그 쿼리

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

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

참고

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

CloudTrail 로그와 함께 Athena를 사용하면 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에 저장하려면 AWS Management Console에서 CloudTrail을 사용하도록 설정합니다. 자세한 내용은 AWS CloudTrail 사용 설명서의 추적 생성을 참조하세요.

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

다음 예를 참조하세요.

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

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

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

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

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

참고

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

CloudTrail 콘솔을 사용하여 CloudTrail 추적용 Athena 테이블을 생성하려면

  1. https://console.aws.amazon.com/cloudwatch/에서 CloudTrail 콘솔을 엽니다.

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

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

    Athena 테이블 생성 선택

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

    참고

    추적과 연결된 버킷의 이름을 찾으려면 CloudTrail 탐색 창에서 추적을 선택하고 추적의 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: STRUCT<
               federatedProvider: STRING,
               attributes: 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://DOC-EXAMPLE-BUCKET/AWSLogs/Account_ID/CloudTrail/';
    참고

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

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

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

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

    1단계의 예제 CREATE TABLE 문은 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://DOC-EXAMPLE-BUCKET/AWSLogs/Account_ID/CloudTrail/us-east-1/2019/02/01/'

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

Athena에서 조직 전체 CloudTrail 로그 파일에 대한 테이블을 생성하려면 수동 분할을 사용하여 Athena에서 CloudTrail 로그용 테이블 생성의 단계를 따르되, 다음 절차에 설명된 대로 수정하세요.

조직 전체 CloudTrail 로그에 대한 Athena 테이블을 생성하려면

  1. CREATE TABLE 문에서 다음 예시와 같이 조직 ID를 포함하도록 LOCATION 절을 수정합니다.

    LOCATION 's3://DOC-EXAMPLE-BUCKET/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://DOC-EXAMPLE-BUCKET/AWSLogs/organization_id/Account_ID/CloudTrail/'

  3. ALTER TABLE 문 ADD PARTITION 절에서 다음 예시와 같이 계정 ID를 포함합니다.

    ALTER TABLE table_name ADD
PARTITION (account='111122223333',
region='us-east-1',
year='2022',
month='08',
day='08')

  4. ALTER TABLE 문 LOCATION 절에서 다음 예시와 같이 조직 ID, 계정 ID 및 추가할 파티션을 포함합니다.

    LOCATION 's3://DOC-EXAMPLE-BUCKET/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://DOC-EXAMPLE-BUCKET/AWSLogs/organization_id/111122223333/CloudTrail/us-east-1/2022/08/08/'

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

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

다음 CREATE TABLE 문 예제에서는 하나의 AWS 리전에 대해 지정된 날짜부터 현재까지의 CloudTrail 로그에 파티션 프로젝션을 자동으로 사용합니다. LOCATION 및 storage.location.template 절에서 bucket, account-id 및 aws-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: STRUCT<
                federatedProvider: STRING,
                attributes: 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://DOC-EXAMPLE-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://DOC-EXAMPLE-BUCKET/AWSLogs/account-id/CloudTrail/aws-region/${timestamp}')

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

중첩된 필드 쿼리

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

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 절을 포함시켜 지정한 행 하위 집합을 반환합니다.