この記事は Akatsuki Advent Calendar の 16 日目の記事です。

アカツキでエンジニアをしているe__komaと申します。今年はAWS Summit TokyoやAmazon Game Developers Conference などを始め、色んなところでAWS運用を紹介させていただいております。

今回もそんなAWS運用話の1つです。 AWSアカウントが増えてくると、統制をとるのが大変ですよね。AWS Configを使えばポリシー違反のリソースを検知し、システム監査を自動化することができます。

この記事では、IAMユーザのIP制限を題材に、AWS Configのカスタムルールを作成する事例をご紹介いたします。

経緯

弊社ではEC2 IAMロールを利用したり、一時的な認証情報を利用することで、極力IAMユーザを作らない運用を目指しています。一方で、各種サードパーティツールの連携ではIAMユーザが必要になることもあります。 IAMユーザを作成した場合は必ずIP制限をかけることにしていますが、大量のAWSアカウントを運用していると、この統制を保つことが大変です。

一部のセキュリティポリシーに対してはAWS Configを使って自動監視していますが、AWSに最初から用意されているマネージドルールの中にはIAMユーザのIP制限をチェックするものはありません。そのため、カスタムルールを作ることにしました。

AWS Configとは？

AWSの設定がポリシーに準拠しているか自動で評価することができる、コンプライアンス監査のサービスです。例えば、IP制限がされていないセキュリティグループを検知したり、Publicになっているリソースを検知する…など、これらの監視を、最初から用意されているマネージドルールを有効にするだけで実施することができます。マネージドルールのリストはこちら。

マルチアカウントマルチリージョンのデータを集約したり、リソースの作成、変更、削除を検知したり、自動修復したりと、豊富な使い方が用意されています。

カスタムルールの作成

今回、やりたいことはIAMユーザにIP制限が設定されているかを監視することです。マネージドルールには用意されていませんが、AWS Configのルールは自分でカスタムルールを作ることができます。

監視したい内容をLambdaで実装して、AWS Configに評価結果を送信することで、ルールに準拠なのか非準拠なのかを判断することができます。 Lambdaで実装するのであれば、CloudWatch Eventsで定期チェックするのとあまり変わらないように思えるかもしれませんが、AWS Configのカスタムルールとして実装すると

マルチアカウントマルチリージョンの非準拠ルールを、1つのアグリゲータアカウントに集約し一元管理できる
つまり全アカウントを監視しつつ、その対応状況の管理がプロジェクト依存にならない
新規AWSアカウント作成時のフローに、ルールを組み込むことができる

といったようなメリットがあります。

実装

簡易ですが、以下のような感じになります。 IAMポリシーのStatement内にIPアドレスを指定するキーがあるか（ただしPublic IPはNG）をチェックしています。ここでは、Lambdaの実装のみで、Lambdaに必要なRole、デプロイ方法には言及しません。

import boto3
import logging
from datetime import datetime

session = boto3.Session()
iam_client = session.client('iam')
config_client = session.client('config')

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, _context):
    logger.info(f'event: {str(event)}')
    result_token = event['resultToken']

    iam_users = iam_client.list_users()
    iam_user_names = [i['UserName'] for i in iam_users['Users']]

    if not iam_user_names:
        logger.info('IAM User does not exist')
        return

    for user_name in iam_user_names:
        user_id = next(i['UserId'] for i in iam_users['Users'] if i['UserName'] == user_name)
        evaluate_compliance(user_name, user_id, result_token)


def evaluate_compliance(user_name, user_id, result_token):
    user_policies = iam_client.list_user_policies(UserName=user_name)

    compliance_type = 'NON_COMPLIANT'
    for policy_name in user_policies['PolicyNames']:
        user_policy = iam_client.get_user_policy(
            UserName=user_name,
            PolicyName=policy_name
        )
        if is_ip_restricted(user_policy):
            compliance_type = 'COMPLIANT'

    logger.info(f'{user_name} is {compliance_type}')
    config_client.put_evaluations(
        Evaluations=[
            {
                'ComplianceResourceType': 'AWS::IAM::User',
                'ComplianceResourceId': user_id,
                'ComplianceType': compliance_type,
                'OrderingTimestamp': datetime.today()
            }
        ],
        ResultToken=result_token
    )


def is_ip_restricted(user_policy):
    statements = user_policy['PolicyDocument']['Statement']

    ip_restriceted = False
    for statement in statements:
        try:
            if statement['Effect'] == 'Allow':
                allow_ips = statement['Condition']['IpAddress']['aws:SourceIp']
            else:
                allow_ips = statement['Condition']['NotIpAddress']['aws:SourceIp']
        except KeyError:
            allow_ips = []

        if not allow_ips:
            pass
        elif is_include_publice_ip(allow_ips):
            ip_restriceted = False
            break
        else:
            ip_restriceted = True

    return ip_restriceted


def is_include_publice_ip(ips):
    is_public = False
    for ip in ips:
        if '0.0.0.0' in ip or '::' in ip:
            is_public = True

    return is_public