技術メモ/
Lambdaで定期的にCloudWatch Logs Insightsを実行して不自然なリクエストを監視する
Webサービスを公開していると、たくさんボットがやってきます。Googleのクローラーくらいならまだよいのですが、いわゆるスキャナーやプローブと呼ばれるような、悪質なボットも多く、かなりの割合のトラフィックをボットに取られてしまうことになります。中には大量のリクエストを連続して送り続けるようなボットもあり、そんなものがサーバーに高い負荷を与えるせいで処理が遅くなり、さらに高い請求書が送られてきたりすると、はらわたが煮えくり返って病気になってしまいます。
なので一般的には、WAFを使って特定のユーザーエージェントをブロックしたり、DoS攻撃や不正なリクエストを検知してブロックしたりするわけで、弊社ももちろんWAFを使っていますが、おまかせで設定してしまうと正常なリクエストまでブロックしてしまう恐れもあります。しかもWAFは単一のリクエストに対しての判定になるので、例えば一定期間不自然なリクエストを送ってきたクライアントをブロックするみたいな機能は備えていないことが多いです。
AWS WAFセキュリティオートメーションについて
AWSには「WAFセキュリティオートメーション」というものが公開されていまして、これをデプロイするだけで各種WAFルールやAthenaによるログ解析ツール、IPレピュテーションリストやハニーポッドまで駆使してサービスのセキュリティを高めてくれるというシロモノです。がぜんぶ導入するにはそれなりにコストもかかりますし、Too muchな印象もあります。ハニーポッドも誤作動とか怖いですよね。
LambdaからCloudWatch Logs Insightsを実行する
そこで、監視をもう少し簡易的に行うために、Lambdaから一定間隔でNginxのアクセスログを監視して、同一IPアドレスからしきい値を超えたエラーリクエストを抽出してみることにします。
detect_scanning.py
client = boto3.client('logs')
def query_log(logGroupName, startTime, endTime, queryString, log):
log.info("[detect_scaning: query_log] Start")
start_query_res = client.start_query(
logGroupName=logGroupName,
startTime=int(startTime.timestamp()),
endTime=int(endTime.timestamp()),
queryString=queryString
)
queryId = start_query_res['queryId']
get_results_res = client.get_query_results(
queryId=queryId
)
while get_results_res['status'] == 'Running' or get_results_res['status'] == 'Scheduled':
time.sleep(5)
get_results_res = client.get_query_results(
queryId=queryId
)
log.info("[detect_scaning: query_log] query_log results \n{}.".format(get_results_res))
return get_results_res['results']
def extract_ip(result):
return result[0]['value']
def lambda_handler(event, context):
log = logging.getLogger()
log.info('[lambda_handler] Start')
log_level = str(os.getenv('LOG_LEVEL').upper())
if log_level not in ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']:
log_level = 'ERROR'
log.setLevel(log_level)
log.info('[detect_scaning: lambda_handler] Start')
logGroupName = os.getenv('LOG_GROUP_NAME')
threshold = int(os.getenv('ERROR_THRESHOLD'))
endTime = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(seconds = 90)
startTime = endTime - datetime.timedelta(minutes = 1)
queryString = 'fields @message'\
' | parse @message \'* - * [*] "* * *" * * * "*" "*" "*"\' as remote_addr, remote_user, timestamp, request_type, location, protocol, status, body_bytes_sent, request_time, host, referer, forwarded_for'\
' | filter status >= 400'\
' | stats COUNT() as count BY remote_addr'\
' | filter count >= %d'\
' | sort count desc'\
' | limit 100' % threshold
log.info("query access logs from %s to %s", str(startTime), str(endTime))
try:
results = query_log(logGroupName, startTime, endTime, queryString, log)
outstanding_requesters = list(map(extract_ip, results))
if len(outstanding_requesters) > 0:
log.info("Adding blocked ips %s", str(outstanding_requesters))
invoke_result = boto3.client('lambda').invoke(
FunctionName=os.getenv('ADD_BLOCKED_IP_FUNCTION_NAME'),
InvocationType='RequestResponse',
Payload=json.dumps({ 'sourceIps': outstanding_requesters, 'eventType': 'scaning', 'description': 'Detect scannin bot.' })
)
output = invoke_result['Payload'].read().decode('utf-8')
log.debug("[detect_scaning: lambda_handler] lambda invokation result: \n{}.".format(output))
except Exception as e:
log.error(e)
raise
finally:
response = {
'statusCode': 200,
'headers': {'Content-Type': 'application/json'},
'body': 'ok'
}
log.info('[lambda_handler] End')
return response
作成したLambda関数をEventBridgeで５分毎に実行します。CloudWatch Logs Insightsで直近の１分間のログの中から、HTTPステータス400以上のレスポンスをIPアドレスでグループ化して、しきい値を超えたものだけを抽出します。endTimeが90秒前になっているのは、あまり早すぎるとログが取得できないことがあったためです。
抽出した不審なIPアドレスはSlackなどに通知するなりよしなに処理してください。うちでは、別のLambdaにIPアドレスを渡して、WAFのブロックリストに登録しています。
add_blocked_ip.py
def lambda_handler(event, context):
log = logging.getLogger()
log.info('[lambda_handler] Start')
log_level = str(os.getenv('LOG_LEVEL').upper())
if log_level not in ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']:
log_level = 'ERROR'
log.setLevel(log_level)
log.info('[add_blocked_ip: lambda_handler] Start')
# ----------------------------------------------------------
# Read inputs parameters
# ----------------------------------------------------------
try:
scope = os.getenv('SCOPE')
ipset_name_v4 = os.getenv('IP_SET_NAME_BAD_BOTV4')
ipset_name_v6 = os.getenv('IP_SET_NAME_BAD_BOTV6')
ipset_arn_v4 = os.getenv('IP_SET_ID_BAD_BOTV4')
ipset_arn_v6 = os.getenv('IP_SET_ID_BAD_BOTV6')
pending_event_types = os.getenv('PENDING_EVENT_TYPES').split(',')
# Fixed as old line had security exposure based on user supplied IP address
log.info("Event->%s<-", str(event))
source_ips = event.get('sourceIps', [])
event_type = event.get('eventType', 'default')
description = event.get('description', '')
# log.info("scope = %s", scope)
# log.info("ipset_name_v4 = %s", ipset_name_v4)
# log.info("ipset_name_v6 = %s", ipset_name_v6)
# log.info("IPARNV4 = %s", ipset_arn_v4)
# log.info("IPARNV6 = %s", ipset_arn_v6)
log.info("source_ips: %s, event_typ: %s, description: %s", str(source_ips), str(event_type), str(description))
if event_type in pending_event_types:
log.info("Event type %s is pending. Skip. End.", str(event_type))
return {}
except Exception as e:
log.error(e)
raise
new_address_v4 = []
new_address_v6 = []
output = None
try:
for source_ip in source_ips:
ip_type = "IPV%s" % ip_address(source_ip).version
if ip_type == "IPV4":
new_address_v4.append(IPv4Network(source_ip).with_prefixlen)
elif ip_type == "IPV6":
new_address_v6.append(IPv6Network(source_ip).with_prefixlen)
if len(new_address_v4) > 0:
ipset = waflib.get_ip_set(log, scope, ipset_name_v4, ipset_arn_v4)
# merge old addresses with this one
log.info(ipset)
current_list = ipset["IPSet"]["Addresses"]
log.info(current_list)
new_list = list(set(current_list) | set(new_address_v4))
log.info(new_list)
if len(list(set(new_list) - set(current_list))) > 0:
output = waflib.update_ip_set(log, scope, ipset_name_v4, ipset_arn_v4, new_list)
log.info("Output->%s<-", output)
else:
log.info('These v4 addresses are already registered.')
if len(new_address_v6) > 0:
ipset = waflib.get_ip_set(log, scope, ipset_name_v6, ipset_arn_v6)
# merge old addresses with this one
log.info(ipset)
current_list = ipset["IPSet"]["Addresses"]
log.info(current_list)
new_list = list(set(current_list) | set(new_address_v6))
log.info(new_list)
if len(list(set(new_list) - set(current_list))) > 0:
output = waflib.update_ip_set(log, scope, ipset_name_v6, ipset_arn_v6, new_list)
log.info("Output->%s<-", output)
else:
log.info('These v6 addresses are already registered.')
except Exception as e:
log.error(e)
raise
finally:
response = {
'statusCode': 200,
'headers': {'Content-Type': 'application/json'},
'body': 'ok'
}
log.info('[lambda_handler] End')
return responseあと、AWSのオートメーションでは、WAFのブロックリストにIPアドレスが追加されたイベントで、自動的にDynamoDBにTTLを設定したレコードを登録し、DynamoDBストリームの削除トリガーで、ブロックリストから削除する処理を行っています。興味がある人はソースコードを見てみてください。
https://github.com/aws-solutions/aws-waf-security-automations/tree/main/source/ip_retention_handler
最後に、ブロックリストにIPアドレスが追加されたときにLambda関数を実行するための、EventBridgeのイベントパターンを載せておきます。
{
"source": ["aws.wafv2"],
"detail-type": ["AWS API Call via CloudTrail"],
"detail": {
"eventSource": ["wafv2.amazonaws.com"],
"eventName": ["UpdateIPSet"],
"requestParameters": {
"name": ["badbot-v4", "badbot-v6"]
}
}
}それではみなさまもよい監視ライフをお送りください。
最終更新:
