Python tool ipcheckでNginxのアクセスログからIPアドレスを抽出し、アクセス元のIPアドレスをヒートマップで可視化する

はじめに

本記事はNginxのアクセスログからIPアドレスを抽出し、アクセス元のIPアドレスをヒートマップで可視化するツールについて紹介します。

クラウドにおけるWebサーバに対する偵察行為や、無料枠の利用を妨げるネットワーク下り（外向き）に関する通信のIPアドレスをファイアウォールに追加するための調査など、インテリジェンスに活用することができます。

必要なのはPython環境のみです。
以下Python toolのipcheckの使用方法について記載しています。

背景

Google Cloud Platform（以下、GCP）でGCEインタンスを2018年頃から4年以上利用し続けています。

ホームページを公開するためのWebサーバとして利用していますが、基本的に無量枠内で利用しています。
GCEインタンスを無料枠で利用するための方法は、以前書いたこれから始めるGCP（GCE）　安全に無料枠を使い倒せをご参照ください。

2023年2月2日、いつも通りにGoogle Cloud Platform & APIs の請求書を確認すると、¥1の請求が発生していました。

2021年に無料で使用できるインスタンスのマシンタイプがF1-microから、E2-microで変わった際に、アップグレードのタイミングで課金が発生してまった事象を除き、それ以外で課金が発生したことはありませんでした。（上記記事の追記2を参照。）

原因調査

課金が発生したということは、GCPのサービスに関する利用料が増えたということになりますが、身に覚えがありません。

課金が発生した原因を特定するために、調査を開始しました。

直接原因

請求書の詳細は、Cloud Billingの料金明細から確認できます。

対象のプロジェクトから、Compute Engineサービスを確認しましたが、何故課金が発生したのか分からなかったため、取り急ぎサポートに問い合わせすることで以下のことが判明しました。

• Compute engine 無料枠のネットワークへの適用は、「1 GB の北米から全リージョン宛ての下りネットワーク（1 か月あたり、中国とオーストラリアを除く）¹」となる。
• 先月の請求分について以下サービスの費用を合計すると、¥0.505885、四捨五入すると¥1になるため費用が発生。
  • Network Internet Egress from Americas to China
  • Network Inter Zone Egress
  • Network Internet Egress from Americas to Australia

以上のことから、費用が発生した原因はGCEインスタンスからの下りネットワークの通信であることが分かりました。
また、サポートからの回答を踏まえて中国とオーストラリア向けの送信料であることが確認できました。

根本原因

下りネットワークの通信について、例えばDNSクエリの通信などが考えられますが、GCEは内部DNSを利用しているため、料金が発生することはありません。

Webサーバとして公開しているGCEインスタンスは、80番ポートと、443番ポートが空いています。
考えられるとしたら、インバウンドのリクエストの通信は無料になるため、アウトバウンドとなる戻りの通信²しかありません。

Cloud Billingからレポートを選択し、対象の請求期間から該当するSKU IDを確認した結果は以下の通りです。

上記グラフの日付と、Nginxのアクセスログを突合した結果、グラフの伸びが高い日についてNginxのアクセスログのサイズが大きいことから、相関性が確認できました。

従って根本原因の仮説については、アクセス元が中国とオーストラリアでWebサーバに対するリクエストがあった場合、戻りの通信で課金が発生していたと考えることができます。

ソリューション

Webサーバとして公開する以上、色々な国々から偵察行為と思われるアクセスを受けるのは日常茶飯事なので、ファイアウォールで防ぐにしてもいたちごっこになるでしょう。

しかし、Cloud Billingを確認した結果を踏まえて、数ヶ月前から費用が発生したSKU IDの使用量が増加傾向（トラフィックが増えている）なため、このまま黙って見過ごしたくはありません。

まずはNginxのアクセスログからIPアドレスを抽出し、アクセス元のIPアドレスを可視化することを考えて、Pythonでツールを開発することにしました。

Pythonで開発したツールは、ツール実行時の引数に、解析したいアクセスログなどのテキストファイルを指定して実行します。

ツール内で実行する処理の流れは以下の通りです。

1. IPアドレスの抽出
2. CSVファイルの作成
3. IPアドレス調査
4. ヒートマップ出力

アクセスログなどのテキストファイルについて、第一フィールドにIPアドレスが確認できれば、どのような形式でもサポートしています。

例えば、第一フィールドにIPアドレスが記載されていないログファイルでも、以下に示すawkコマンドを用いてIPアドレスのみを抽出すれば、どのようなログファイルでも対応できます。

$ awk 'match($0, /[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+/) { ip=substr($0, RSTART, RLENGTH); if (ip ~ /^([0-9]{1,3}\.){3}[0-9]{1,3}$/) print ip }' logfile > logfile_ip

以降、ツール内で実装している具体的なメソッドについて記載しています。

IPアドレスの解析は、ipinfoのIPinfo Python Client Library、ヒートマップの作成は、Foliumを利用しています。
ipinfoについては以前書いたグローバルIPアドレスを最速で調査する方法をご参照ください。

IPアドレスの抽出

アクセス元のIPアドレスの抽出は、subprocessを用いてLinuxのコマンドを活用することで抽出しています。

ポイントは、uniqコマンドは隣接する重複行を1行に集約するために、ファイル内のすべての重複行を削除する場合は事前にsortコマンドでソートする必要があります。

@classmethod
def get_accesslog(cls, filename) -> List:
    process = subprocess.run(
        "cat %s | awk '{print $1}' | sort | uniq -c | sort -nr"
        % (filename),
        shell=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding='utf-8'
    )

    if process.stderr != '':
        raise BatchError(f'Error in command: {process.stderr}')

    accesslog = process.stdout
    accesslog = accesslog.split('\n')

    return accesslog

subprocess.run内で変数を利用するために、%sを用いて文字列フォーマットを行なっています。

CSVファイルの作成

上記get_accesslogで取得したアクセスログを引数に、CSVファイルを作成します。

@classmethod
def create_csvfile(cls, accesslog: list):
    if accesslog[0] == '':
        raise BatchError(f'No logs to parse')

    with open(Ipcheck.CSV, 'w', newline='') as csvfile:
        writer = csv.writer(csvfile)
        # Create CSV header
        writer.writerow(['No.',
                         'Count',
                         'IP address',
                         'Region',
                         'Country',
                         'Loc',
                         ])

        row_num = 1
        re_count = re.compile(r'\d{1,}')
        re_ip = re.compile(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}')
        for row in accesslog:
            if row == '':
                break
            count = re_count.search(row)
            ip = re_ip.search(row)
            writer.writerow([row_num,
                             count.group(),
                             ip.group(),
                             'unconfirmed',
                             'unconfirmed',
                             'unconfirmed',
                             ])
            row_num += 1

生成されるCSVファイルのイメージは以下の通りです。

IPアドレス調査

上記で書いた通りにIPアドレス調査は、ipinfoのIPinfo Python Client Libraryを利用しています。

開発者向けドキュメントより、Batching Requestsを活用することで、最大 1000個のIPinfo APIリクエストを1つのリクエストにグループ化できます。これにより、IPルックアップの処理が大幅に高速化されます。

なお、Batching RequestsはAPI認証が必要です。
無料枠は毎月、50,000リクエストが上限ですが、Batching Requestsを利用しなくても、実行できるように実装しています。（ipinfoのアカウント作成不要）

上記create_csvfileで作成したCSVファイルを引数に、IPアドレス調査を行なっています。

@classmethod
def fetch_ipinfo(cls, csvfile):
    csv_df = pd.read_csv(csvfile)
    iplist = [f'{ip}' for ip in csv_df["IP address"]]
    coords = []

    if Ipcheck.BATCH_MODE is True:
        logger.custom_logging('Batching Requests')
        if Ipcheck.TOKEN is None:
            raise BatchError(f'Token not set: {Ipcheck.TOKEN}')

        handler = ipinfo.getHandler(Ipcheck.TOKEN)
        response = list(handler.getBatchDetails(iplist).values())

        for row in response:
            csv_df.loc[csv_df['IP address'] == row["ip"], ['Region']] = row['region']
            csv_df.loc[csv_df['IP address'] == row["ip"], ['Country']] = row['country']
            csv_df.loc[csv_df['IP address'] == row["ip"], ['Loc']] = row['loc']

            coord = row['loc']
            lat, long = coord.split(',')
            lat, long = float(lat), float(long)
            coords.append([lat, long])

    else:
        for ip in iplist:
            response = requests.get(f'{config.common_setting["url"]}/{ip}')
            response = response.json()

            csv_df.loc[csv_df['IP address'] == ip, ['Region']] = response['region']
            csv_df.loc[csv_df['IP address'] == ip, ['Country']] = response['country']
            csv_df.loc[csv_df['IP address'] == ip, ['Loc']] = response['loc']

            coord = response['loc']
            lat, long = coord.split(',')
            lat, long = float(lat), float(long)
            coords.append([lat, long])

            time.sleep(1)

    # Update CSV file
    csv_df.to_csv(csvfile, index=False)

    return coords

CSVファイルから取り込んだデータは、処理しやすいようにpandasを用いてデータフレームに変換しています。

ヒートマップで可視化

最後は上記fetch_ipinfoメソッドで返す座標データを格納しているcoordsを引数に、ヒートマップを出力します。

@classmethod
    def output_heatmap(cls, loc):
        m = folium.Map(location=[36, 140], zoom_start=2)
        HeatMap(data=loc, radius=15,).add_to(m)
        m.save('heatmap.html')

        return

foliumのパラメーターについて、locationは地図の緯度と経度(北座標、東座標)、zoom_startについてはマップの初期ズームレベルを指定します。
本記事の例では日本の東京を中心にヒートマップ出力しています。

ヒートマップの解析

ルートディレクトリで以下のコマンドを実行します。

以下の場合は-bオプションを付与することでバッチモードで実行しています。

python3 src/main.py -f access.log -b

2023-02-16 01:41:33,288 [INFO] settings.custom_logger: main start
2023-02-16 01:41:33,289 [INFO] settings.custom_logger: Analysis target file: /Users/Your Name/Desktop/access.log
2023-02-16 01:41:33,289 [INFO] settings.custom_logger: Start extracting IP address
2023-02-16 01:41:33,311 [INFO] settings.custom_logger: Create CSV file
2023-02-16 01:41:33,313 [INFO] settings.custom_logger: IP address research
2023-02-16 01:41:33,336 [INFO] settings.custom_logger: Batching Requests
2023-02-16 01:41:34,507 [INFO] settings.custom_logger: Heatmap output
2023-02-16 01:41:34,517 [INFO] settings.custom_logger: main end

コマンド実行後、heatmap.htmlファイルが生成されるので、ブラウザなどから確認します。

以下のヒートマップは、実際にWebサーバで稼働しているNginxのアクセスログを基にツールで出力した結果です。

拡大してみると、中国からのアクセスが確認できます。

更に拡大することで、より具体的な位置情報を推測することができます。

仮説を立証することができました。

インテリジェンスで得た情報を活用するため、CSVファイルの出力結果からIPアドレスを確認して、GCPのVPC ファイアウォール ルールのdenyルールを更新します。

各国の国名コードはISO 3166-1から確認できます。

おわりに

「因果関係」とは、原因とそれによって生じる結果との関係です。

クラウド利用についてコストという結果が生じる場合は、何かしら原因があるということを改めて経験しました。

本記事で紹介しているPython toolはipcheckから取得できます。

以上です。

追記

作成したツールを活用し、無料枠を妨げる要因となる中国とオーストラリアのIPアドレスが確認できた場合、VPC ファイアウォール ルールのdenyルールに追加する運用を行ないました。

結果、トラフィックは落ち着きを見せて、2023年3月2日の請求は¥0でした。

1. 無料枠の使用量上限 ↩

2. ネットワーキングのすべての料金体系 ↩