Monitor Amazon S3 activity using S3 server access logs and Pandas in Python

지뚜·2023년 9월 2일

aws

목록 보기

2/3

학습 재미 : 😀😀

Amazon S3 버킷의 Access Logs를 활성화 하고, python의 Pandas를 이용하여 집계 데이터를 분석해본다.
https://aws.amazon.com/ko/blogs/storage/monitor-amazon-s3-activity-using-s3-server-access-logs-and-pandas-in-python/

1. Amazon S3 Access Logs?

보유하고 있는 버킷의 request 요청에 대해서 로깅을 제공하는 기능으로 보안 감사 활동 등에서 활용도가 높을것이라 생각된다.

다만 위와같이 서비스 데이터가 보관된 버킷과 로깅되는 버킷을 구분하지 않으면 일명 Looping 구조가 발생할 수 있기에 주의해야 할것이다.

2. Summary

지난시간 스터디에서 배운 내역으로 다음과 같은 배경을 준비한다.

ahss-mybucket : 퍼블릭 접근이 가능한 버킷과, 랜덤한 파일 여러개

ahss-s3-access-logging : ahss-mybucket의 Access Log를 저장할 수 있는 버킷

그 외 다음과 같은 flow를 준비

python을 이용하여 random한 html 파일을 생성하고, 이를 boto3를 이용하여 ahss-mybucket에 업로드

python을 이용하여 내 PC에서 random한 User-Agent 및 다수 파일에 접근 시도

boto3 및 pandas를 이용하여 집계 데이터 생성

Amazon S3 수명주기 관리 모니터링

2-1. 퍼블릭 버킷 생성

버킷은 다음과 같이 퍼블릭 설정과 S3:GetObject 정책을 할당한다.

또한

2-2. Access Logging용 버킷 생성

Access Logging을 수행할 버킷은 퍼블릭 접근이 불필요하기에 일반적인 버킷 생성을 따른다.

2-3. 퍼블릭 버킷의 Access Logging 설정

서비스 데이터가 존재하는 퍼블릭 버킷의 설정 중 Access Logging을 설정하고, 대상을 2-2에서 생성한 버킷으로 지정한다.

2-4. 랜덤 html 파일 생성

python을 이용하여 랜덤한 html 파일을 생성하고, boto3 모듈을 이용하여 S3에 업로드를 수행한다.
이때 사전에 aws credential이 필요하다.

import boto3
import random
import string
import os

bucket_name = "ahss-mybucket"
num_files = 5
s3 = boto3.client('s3')

for i in range(1, num_files + 1):
    random_file_name = f"random_{i}.html"
    html_content = f"<html><head><title>Random HTML {i}</title></head><body><h1>This is random HTML file {i}</h1></body></html>"
    with open(random_file_name, 'w') as f:
        f.write(html_content)
    s3.upload_file(random_file_name, bucket_name, random_file_name)
    print(f"Uploaded {random_file_name} to S3 bucket {bucket_name}")
    os.remove(random_file_name)

실행 시 다음과 같은 결과를 얻을 수 있다.

Uploaded random_1.html to S3 bucket ahss-mybucket
Uploaded random_2.html to S3 bucket ahss-mybucket
Uploaded random_3.html to S3 bucket ahss-mybucket
Uploaded random_4.html to S3 bucket ahss-mybucket
Uploaded random_5.html to S3 bucket ahss-mybucket

2-5. PC에서 S3 URL에 대해 Request 테스트

다음과 같이 테스트 파일 1개에 대해 URL 주소 획득 후 PC에서 접근을 테스트 해본다.

정상적으로 접근이 가능함을 확인

2-6. PC에서 random한 User-Agent 및 다수 파일에 접근 시도

python에서 비동기 방식으로 미리 생성해둔 random 파일에 User-Agent를 임의로 바꿔가면서 퍼블릭으로 접근을 다수 시도한다.

import requests
import random

bucket_url = "https://ahss-mybucket.s3.ap-northeast-2.amazonaws.com/"

# 사용자 에이전트(User-Agent) 목록
user_agent_list = [
    # Chrome
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.0.0 Safari/537.36",
    
    # Firefox
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:100.0) Gecko/20100101 Firefox/100.0",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:101.0) Gecko/20100101 Firefox/101.0",

    # Safari
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/15.0 Safari/605.1.15",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.0 Safari/605.1.15",

    # Edge
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/100.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/101.0.0.0 Safari/537.36",
]

# 10000번의 요청 보내기
for _ in range(10000):
    user_agent = random.choice(user_agent_list)
    random_file_url = f"{bucket_url}random_{random.randint(1, 5)}.html"
    headers = {"User-Agent": user_agent}
    response = requests.get(random_file_url, headers=headers)
    if response.status_code == 200:
        print(f"Successfully fetched file with User-Agent: {user_agent}")
    else:
        print(f"Failed to fetch file with User-Agent: {user_agent}, Status code: {response.status_code}")

실행 결과
Successfully fetched file with User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.0.0 Safari/537.36
Successfully fetched file with User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.0.0 Safari/537.36
Successfully fetched file with User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/101.0.0.0 Safari/537.36
Successfully fetched file with User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/101.0.0.0 Safari/537.36

다음과 같이 ahss-s3-access-loggin 버킷에 접근 이력 관련된 로그가 잘 쌓인다.

3. boto3 및 pandas를 이용하여 집계 데이터 생성

https://aws.amazon.com/ko/blogs/storage/monitor-amazon-s3-activity-using-s3-server-access-logs-and-pandas-in-python/
이곳을 참조하여 학습 및 테스트를 수행해본다.

기본적인 python 외부 모듈 설치

pip3 install boto3, pandas

boto3 라이브러리 테스트

import os
import boto3
import pandas as pd

bucket = 'demo-access-logs-bucket'
s3_client = boto3.client('s3')
print (s3_client)

<botocore.client.S3 object at 0x7efecf37cfd0>

정상적으로 라이브러리 사용이 가능함을 확인하였으니 조금 더 확장해보자

import boto3
import pandas as pd
import matplotlib.pyplot as plt

class S3AccessLogAnalyzer:
    def __init__(self, bucket_name):
        self.bucket_name = bucket_name
        self.s3_client = boto3.client('s3')
        self.log_objects = []

    def list_log_objects(self):
        paginator = self.s3_client.get_paginator('list_objects_v2')
        result = paginator.paginate(Bucket=self.bucket_name, Prefix='')
        for each in result:
            key_list = each['Contents']
            for key in key_list:
                self.log_objects.append(key['Key'])

    def load_log_data(self):
        log_data = []
        for log_key in self.log_objects:
            log_data.append(pd.read_csv(f's3://{self.bucket_name}/{log_key}', sep=" ", names=['Bucket_Owner', 'Bucket', 'Time', 'Time_Offset', 'Remote_IP', 'Requester_ARN/Canonical_ID',
                                                                                             'Request_ID', 'Operation', 'Key', 'Request_URI', 'HTTP_status', 'Error_Code', 'Bytes_Sent', 'Object_Size',
                                                                                             'Total_Time', 'Turn_Around_Time', 'Referrer', 'User_Agent', 'Version_Id', 'Host_Id', 'Signature_Version',
                                                                                             'Cipher_Suite', 'Authentication_Type', 'Host_Header', 'TLS_version'],
                                       usecols=range(25)))

        return pd.concat(log_data)

    def analyze_and_visualize(self):
        df = self.load_log_data()
        
        # Pie 그래프: 가장 많이 요청된 파일 5개
        top_five_objects = df[(df['Operation'] == 'REST.GET.OBJECT')]['Key'].value_counts().nlargest(5)
        top_five_objects.plot.pie(label='')
        plt.savefig("/home/ec2-user/my_graph.png")
        plt.show()

        # Bar 그래프: 응답 상태 코드별 요청 수
        response_codes = df['HTTP_status'].value_counts()
        response_codes.plot.bar()
        plt.savefig("/home/ec2-user/my_graph2.png")

        # 접근 IP 목록
        access_ip_list = df[(df['Operation'] == 'REST.GET.OBJECT')]['Remote_IP'].value_counts()
        print("=== access ip list ===")
        print(access_ip_list)

        # 거부된 IP 목록
        deny_ip_list = df[(df['HTTP_status'] == 403)]['Remote_IP'].value_counts()
        print("=== deny ip list ===")
        print(deny_ip_list)

        # 거부된 파일 목록
        deny_key_list = df[(df['HTTP_status'] == 403)]['Key'].value_counts()
        print("=== deny key list ===")
        print(deny_key_list)

if __name__ == "__main__":
    bucket_name = 'ahss-s3-access-logging'
    analyzer = S3AccessLogAnalyzer(bucket_name)
    
    # S3 로그 객체 목록 가져오기
    analyzer.list_log_objects()
    
    # 데이터 분석 및 시각화
    analyzer.analyze_and_visualize()

Jupiter가 아닌, terminal을 이용하여 ssh 접근 후 테스트 하였기에 그래프 출력에 문제가 있어서 파일로 저장하고, 로컬 PC에서 다운로드하여 확인한다

가장 자주 액세스되는 객체 확인
❯ scp -i test.pem ec2-user@$EC2_Public_IP:/home/ec2-user/my_graph.png ./my_graph.png

버킷의 응답 코드 보기
❯ scp -i test.pem ec2-user@$EC2_Public_IP:/home/ec2-user/my_graph2.png ./my_graph2.png
특이 my_graph2.png는 아무래도 기존 plt를 초기화하지 않아서 발생한 오류로 보이고 추후 개선할 수 있도록 한다.