ELB + VPC + Lambda を使って高頻度アクセスを自動遮断する

同一IPからの高頻度アクセスはサーバーの負荷になったりで遮断したい場合があります。
ただ、アプリケーションサーバーで検知や遮断をしてしまうと負荷になってしまいます。

そのため通常の処理とは別枠で解析と遮断を行い、ゲームサーバー本体への負荷をほぼ０で行いたいと思います。

ELB のアクセスログ出力機能

AWSのELBを使っているので、ELBのアクセスログをS3に出力する機能を使って作成したいと思います。

ELBのログ機能はコンソールからS3のバケット名を指定するだけで有効にできます。
5分おきと1時間おきが選べるので今回は5分を選びます。

S3にファイルが追加されたら Lambda を呼び出す

今度はこのS3バケットに追加されたログを元に高頻度でアクセスしているIPを遮断しましょう。
慣れているので Python を使います。

Lambda のサンプルで S3 Put というのがあるのでこれをベースにします。

対象の S3 バケットの設定などを行います。
今回の Lambda 用の IAM Role に対象の S3 バケットへの GET と NetworkACL への編集権限を与えておきます。

そしたら Python スクリプトとして以下を流し込んで設定完了です。

# -*- coding:utf-8 -*-
from __future__ import absolute_import
from __future__ import unicode_literals

import json
import urllib
from collections import defaultdict

import boto3

# ログの保存されるS3バケット名
BUCKET_NAME = '************'

# BANにつかう NetworkACL ID
ACL_ID = '************'

# ACLに使うルール番号の範囲
BANISH_ACL_RULE_NUM_MIN = 10
BANISH_ACL_RULE_NUM_MAX = 99

# 5分間に何回以上アクセスするとBANになるか
BANISH_THRESHOLD_VALUE = 20


def analysis_accesslog(key):
    """
    取得したログの中から接続元IPアドレスを集計する
    """
    counter = defaultdict(int)
    s3 = boto3.resource('s3')
    body = s3.ObjectSummary(BUCKET_NAME, key).get()['Body'].read().splitlines()
    for line in body:
        timestamp, _, client_ip_port = line.split(' ')[:3]
        client_ip = client_ip_port.split(':')[0]
        counter[client_ip] += 1

    target_ip_counts = sorted([(ip, count) for (ip, count) in counter.items() if count >= BANISH_THRESHOLD_VALUE],
                              key=lambda x: x[1])

    return target_ip_counts


def create_acl_entry(target_ips):
    """
    指定したIPのアクセス拒否ルールを NETWORK ACL に追加する
    :type target_ips: list of string
    :rtype: list of string
    """
    ec2 = boto3.resource('ec2')
    network_acl = ec2.NetworkAcl(ACL_ID)
    # 既存のBANルールを取得
    rules = [entry for entry in network_acl.entries
             if BANISH_ACL_RULE_NUM_MIN <= entry['RuleNumber'] <= BANISH_ACL_RULE_NUM_MAX]

    # 使用可能なルール番号リストを作成
    rule_nums = set(range(BANISH_ACL_RULE_NUM_MIN, BANISH_ACL_RULE_NUM_MAX + 1))
    used_rule_nums = [rule['RuleNumber'] for rule in rules]
    rule_nums -= set(used_rule_nums)
    if not rule_nums:
        # これ以上ルール番号の空きが無い
        raise

    # BAN対象からBAN済みIPを消す
    banished_ips = [rule['CidrBlock'].replace('/32', '') for rule in rules]
    target_ips = list(set(target_ips) - set(banished_ips))  # 返す時にlistにするのでここでlistにしてしまう
    # 対象IPを拒否するレコードを作成
    for target_ip in target_ips:
        next_rule_num = rule_nums.pop()
        print network_acl.create_entry(
            DryRun=False,  # 仮実行か
            RuleNumber=next_rule_num,  # ルール番号。このACL内でユニーク。若い番号が優先される
            Protocol='-1',  # プロトコル -1 で全部
            RuleAction='deny',  # deny or allow
            Egress=False,  # True=Outbound, False=Inbound
            CidrBlock='{}/32'.format(target_ip),  # 対象 ex. 172.16.0.0/24
            PortRange={'From': 80, 'To': 443}  # ポート範囲
        )
    return target_ips


def lambda_handler(event, context):
    print 'Received event: ' + json.dumps(event, indent=2)

    # Get the object from the event and show its content type
    s3_event = event['Records'][0]['s3']
    bucket_name = s3_event['bucket']['name']
    if bucket_name != BUCKET_NAME:
        # 指定バケットでなかった
        return

    object_key = urllib.unquote_plus(s3_event['object']['key']).decode('utf8')
    if not object_key.endswith('.log'):
        # ログ以外だった
        return

    # ログから対象IPを割り出し
    banish_ip_counts = analysis_accesslog(object_key)

    # ACLに追加
    print create_acl_entry([ip for (ip, count) in banish_ip_counts])

注意点

ELB のログは ELB の内部ノードごとに出力されるので必ずしも５分間のログがすべて１ファイルに入るわけではありません。
フロントは DNS ラウンドロビンのため、大抵のクライアントは 1ファイル内に集まっていることが多いですが保証されているわけではないです。

正しく行うのであれば、複数のファイルから集計を行い、それを元に BAN を行うほうが良いでしょう。

今回はシンプルにするために１ファイルごとの集計を元に BAN 対象を追加しています。

また WiFi 環境など、NATで同じ IP を複数人で使っている場合もあり、それは正規ユーザーですので慎重に扱う必要があります。

実際の運用ではかなり余裕を持って閾値を設定する必要があり、IP単位のアクセス数だけをみて自動で BAN するのであれば、時限式で解除されるのが望ましいでしょう。（1時間で解除されるなど）