CDK Aspectsでリソースのコンプライアンスチェックをしてみる（python）

Posted at 2024-07-21

CDK Aspectsとは

CDK Aspectsとは、特定のスコープ内のconstructに対して、共通の操作を適用する方法です。

CDK Aspectsを利用することで、簡単にコンプライアンスチェックを行うことができます。本記事では、下記のチェックを試してみます。

EC2のインスタンスタイプがt2.micro以外になっていないか
セキュリティグループでsshのポートが公開されていないか

CDKのライフサイクル

実際にCDK Aspectsを使ってみる前に、CDKのライフサイクルについて少し触れておきます。ライフサイクルについては、公式ドキュメントの図がわかりやすいのですが、CDKではデプロイ時に次のようなフェーズを実行します。

※公式ドキュメントより抜粋

詳細は省きますが（というか私も細かいところは勉強中ですが）、CDK AspectsはPreparationフェーズで実行されるということを覚えておいてください。

やってみる

今回は、下記のサンプルコードをベースにチェックを行っていきます。このサンプルコードは簡単なWebサイトを構築するものです。インスタンスタイプとしてt3.smallが利用されており、セキュリティグループでsshのポートが公開されています。

このコードをそのままデプロイすると、sshポートが公開されたサーバーが立ち上がりますので、注意して利用してください。

demo_cdk_aspects_stack.py

from aws_cdk import Stack, CfnOutput, aws_ec2 as ec2, aws_s3_assets as s3_assets
from constructs import Construct


class DemoCdkAspectsStack(Stack):

    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        my_vpc = ec2.Vpc(self, "MyVPC", nat_gateways=0, max_azs=2)

        web_server = ec2.Instance(
            self,
            "WebServer",
            instance_type=ec2.InstanceType("t3.small"),
            machine_image=ec2.MachineImage.latest_amazon_linux2(),
            vpc=my_vpc,
            vpc_subnets=ec2.SubnetSelection(subnet_type=ec2.SubnetType.PUBLIC),
        )

        # Allow All 22, 80 Access
        web_server.connections.allow_from_any_ipv4(ec2.Port.tcp(80))
        web_server.connections.allow_from_any_ipv4(ec2.Port.tcp(22))

        # Attaching an Elastic IP to the Instance
        ec2.CfnEIP(self, "WebServerEIP", instance_id=web_server.instance_id)
        # Installing packages at instance launch
        web_server.add_user_data(
            "yum update -y",
            "amazon-linux-extras install nginx1",
            "rm -rf /usr/share/nginx/html/*",
        )

        # Upload index.html to s3
        web_page_asset = s3_assets.Asset(
            self, "WebPageAsset", path="web_pages/index.html"
        )

        # Install Index.html from S3
        web_server.user_data.add_s3_download_command(
            bucket=web_page_asset.bucket,
            bucket_key=web_page_asset.s3_object_key,
            local_file="/usr/share/nginx/html/index.html",
        )
        web_page_asset.grant_read(web_server.role)

        web_server.add_user_data("service nginx start")

        # Output the DNS of the web server
        CfnOutput(self, "WebServerDNS", value=web_server.instance_public_dns_name)

EC2のインスタンスタイプをチェックしてみる

CDK Aspects用のコードを記述するファイルを新規に作成します。今回は、stackを定義しているファイルと同じフォルダにaspects.pyを作成します。

作成したaspects.pyファイルを次のように記述します。

aspects.py

import jsii
from aws_cdk import IAspect, Annotations, aws_ec2 as ec2


@jsii.implements(IAspect)
class EC2InstanceTypeChecker:

    def visit(self, node):

        if isinstance(node, ec2.CfnInstance):
            if node.instance_type != "t2.micro":
                Annotations.of(node).add_warning(
                    f"{node.instance_type} is not approved. Use t2.micro"
                )

CDK Aspectsでは、IAspectインターフェースを利用します。IAspectsでは、Visitorパターンを採用しています。Visitorパターンは、データの構造と処理を分離する手法です。また、CDKの元のコードはTypeScriptで記述されているため、pythonからも処理ができるようJSIIを利用します。

次に、app.pyファイルを編集します。

app.py

import aws_cdk as cdk
from demo_cdk_aspects.demo_cdk_aspects_stack import DemoCdkAspectsStack
from demo_cdk_aspects.aspects import EC2InstanceTypeChecker


app = cdk.App()
DemoCdkAspectsStack(
    app,
    "DemoCdkAspectsStack",
)

# Aspect attachment
cdk.Aspects.of(app).add(EC2InstanceTypeChecker())

app.synth()

この状態で、cdk synthコマンドを実行してみましょう。-qオプションをつけることで、余計な出力を省くことができます。

cdk synth -q

次のような警告が表示されました。なお、警告は表示されますがデプロイすることは可能です。

デプロイ自体を制限したい場合は、aspects.pyファイルのadd_warning部分をadd_errorに変更します。

aspects.py

import jsii
from aws_cdk import IAspect, Annotations, aws_ec2 as ec2


@jsii.implements(IAspect)
class EC2InstanceTypeChecker:

    def visit(self, node):

        if isinstance(node, ec2.CfnInstance):
            if node.instance_type != "t2.micro":
                # edit
                Annotations.of(node).add_error(
                    f"{node.instance_type} is not approved. Use t2.micro"
                )

この状態で、デプロイコマンドを実行すると、下記のエラーが出力されます。

aspects.pyの記載によって、結果が変わることがわかりましたね。今回は、さらに自動でインスタンスタイプを変更してみたいと思います。aspects.pyを下記のように書き換えます。

aspects.py

import jsii
from aws_cdk import IAspect, Annotations, aws_ec2 as ec2


@jsii.implements(IAspect)
class EC2InstanceTypeChecker:

    def visit(self, node):

        if isinstance(node, ec2.CfnInstance):
            if node.instance_type not in ["t2.micro", "t3.micro"]:
                Annotations.of(node).add_warning(
                    f"{node.instance_type} is not approved. Automatically change InstanceType"
                )
                # change instance type
                node.instance_type = "t2.micro"

このように記載することで、デプロイコマンド時に自動でインスタンスタイプを書き換えることができます。実際にデプロイコマンドを実行すると次のような警告が出力され、

t2.microのインスタンスが立ち上がります。

CDK Aspectsを利用してリソースの値を変更すると、意図しない変更が起きる可能性があります。そのため、基本的にはコンプライアンスチェックとして使う方がオススメです。

セキュリティグループをチェックしてみる

続いて、セキュリティグループをチェックしてみます。今回は、sshのポートが公開されてないかをチェックします。aspects.pyを次のように書き換えます。

aspects.py

import jsii
from aws_cdk import IAspect, Stack, Annotations, aws_ec2 as ec2


@jsii.implements(IAspect)
class SSHAnywhereChecker:

    def visit(self, node):

        if isinstance(node, ec2.CfnSecurityGroup):

            rules = Stack.of(node).resolve(node.security_group_ingress)
            for rule in rules:
                if (
                    rule["ipProtocol"] == "tcp"
                    and rule["fromPort"] <= 22
                    and rule["toPort"] >= 22
                ):
                    if rule["cidrIp"] == "0.0.0.0/0":
                        Annotations.of(node).add_warning(
                            "SSH access is open to the world."
                        )

先ほどとは少し違う点がありますね。resolve()というメソッドが使われています。これはなんでしょう？これを理解するために、冒頭で触れたライフサイクルを思い出してみましょう。

※公式ドキュメントより抜粋

AWS CDKでは、トークンという値が存在します。AWS CDKでは、よく他のリソースを参照させるかと思います（Cloud FormationでいうRef）。セキュリティグループの設定も、EC2インスタンスを参照しているため、トークンとして扱われています。

基本的に、トークンはSynthesisフェーズでresolve()メソッドにより解決されます。しかし、CDK AspectsはPreparationフェーズで実行されます。そのため、aspect.py側でresolve()メソッドを利用してトークンを解決してあげる必要があります。

それでは、app.pyファイルを編集して出力結果を確認してみましょう。

app.py

import aws_cdk as cdk
from demo_cdk_aspects.demo_cdk_aspects_stack import DemoCdkAspectsStack
from demo_cdk_aspects.aspects import SSHAnywhereChecker

app = cdk.App()
DemoCdkAspectsStack(
    app,
    "DemoCdkAspectsStack",
)

# Aspect attachment
cdk.Aspects.of(app).add(SSHAnywhereChecker())

app.synth()