こんにちは、アプリケーションサービス部ディベロップメントサービス1課の滝澤です。
本記事をご覧いただきありがとうございます。
今回、AWS Config Rule Development Kit(以下、RDK)を使用してAWS ConfigカスタムLambdaルールを作成してみたので、その方法を複数本に分けてお届けしたいと思います。
<RDKシリーズの記事>
本記事の目的
本記事ではRDKの概要を理解した方向けに、カスタムLambdaルールのロジックの書き方、テスト、デプロイを実際の簡単な例をもとにご説明させていただきます。
併せてRDKを使用せずにLambda関数を0から作成するときと比べたメリットデメリットもお伝えできればと思います。
このブログは第2弾実践編となります。もし概要を知りたい方がいらっしゃいましたら第1弾の概要編をぜひご覧ください。
前提
今回作成するルールはEC2インスタンスのインスタンスタイプがt2.micro
の場合にCOMPLIANT
、それ以外はNON_COMPLIANT
と評価するものです。
AWS公式のYouTube動画で作成しているものを少し噛み砕いてお伝えするものになります。
RDKのインストール、セットアップが完了していることを前提にcreate
コマンド実行の説明から始めます。
ルールの作成
createコマンドの実行
ルールの作成にはcreate
コマンドを実行します。
$ rdk create ec2_desired_instance_type --runtime python3.7 --resource-types AWS::EC2::Instance --input-parameters '{"desiredInstanceType":"t2.micro"}'
--runtime
は前提の部分でお話ししたYouTube動画でpython3.7
を指定しており、わかりやすくするめpython3.7
統一しています。
今回の評価対象はEC2インスタンスなので--resource-types
にはAWS::EC2::Instance
を設定します。
また評価条件であるインスタンスタイプを--input-parameters
パラメータを使用して、'{"desiredInstanceType":"t2.micro"}'
と指定しています。
このパラメータの値のt2.micro
を他の任意のインスタンスタイプに変更することもできます。
コマンドを実行するとカレントディレクトリにec2_desired_instance_type
というディレクトリが作成され、その中に以下の3つのファイルが作成されています。
- ec2_desired_instance_type.py
- ec2_desired_instance_type_test.py
- parameters.json
ルールのロジックの実装
上記のec2_desired_instance_type.py
に変更を加えていきます。
############# # Main Code # #############
の中の
############################### # Add your custom logic here. # ###############################
にロジックを追加していきましょう。
今回実装するコード
今回実装するコードはこちらです。
(略) def evaluate_compliance(event, configuration_item, valid_rule_parameters): """Form the evaluation(s) to be return to Config Rules Return either: None -- when no result needs to be displayed a string -- either COMPLIANT, NON_COMPLIANT or NOT_APPLICABLE a dictionary -- the evaluation dictionary, usually built by build_evaluation_from_config_item() a list of dictionary -- a list of evaluation dictionary , usually built by build_evaluation() Keyword arguments: event -- the event variable given in the lambda handler configuration_item -- the configurationItem dictionary in the invokingEvent valid_rule_parameters -- the output of the evaluate_parameters() representing validated parameters of the Config Rule Advanced Notes: 1 -- if a resource is deleted and generate a configuration change with ResourceDeleted status, the Boilerplate code will put a NOT_APPLICABLE on this resource automatically. 2 -- if a None or a list of dictionary is returned, the old evaluation(s) which are not returned in the new evaluation list are returned as NOT_APPLICABLE by the Boilerplate code 3 -- if None or an empty string, list or dict is returned, the Boilerplate code will put a "shadow" evaluation to feedback that the evaluation took place properly """ ############################### # Add your custom logic here. # ############################### if configuration_item['resourceType'] != 'AWS::EC2::Instance': return 'NOT_APPLICABLE' if configuration_item['configuration']['instanceType'] == valid_rule_parameters['desiredInstanceType']: return 'COMPLIANT' return 'NON_COMPLIANT' (略)
シンプルなものなので実装自体もわかりやすいものになっているかと思います。
リソースの設定項目(configration_item)の確認方法
コードの中身を説明していきます。
まずconfigration_item
ですが、ここにリソースの設定項目が入ってきます。
リソースの設定項目のサンプルについては、以下の3つで確認することができます。
- AWSマネジメントコンソールから確認する
- AWS公式のGitHub(https://github.com/awslabs/aws-config-resource-schema/tree/master/config/properties/resource-types)を参照する
- rdkの
sample-ci
コマンドを使用する
1. AWSマネジメントコンソールから確認する
以下の手順で自分のAWSアカウントにあるリソースの設定項目を見ることができます。
AWSマネジメントコンソール > AWS Config > リソース > リソースタイプで任意のリソースタイプ > 任意のリソース > 設定項目(JSON)の表示
実際に設定項目にどのような値が入るのか確認したいときに有効です。
2. AWS公式のGitHubを参照する
公式が出しているリソースの設定項目一覧です。ほぼサポートされていますが、Configでサポートされていても表記がないリソースもあるので注意が必要です。
3. rdkのsample-ci
コマンドを使用する
$ rdk sample-ci <Resource Type>
を実行するとそのリソースタイプの設定項目のサンプルを出力してくれます。
あくまでサンプルレスポンスなので、紐つけているAWSアカウントの情報ではない点に注意が必要です。
[ec2-user@ip-10-0-0-41 ~]$ rdk sample-ci AWS::EC2::Instance { "version": "1.2", "accountId": "681361479661", "configurationItemCaptureTime": "2017-04-05T22:23:11.677Z", "configurationItemStatus": "ResourceDiscovered", "configurationStateId": "1491430991677", "configurationItemMD5Hash": "7d83283adb8b966945d43cee39c7419c", "arn": "arn:aws:ec2:us-east-1:681361479661:instance/i-03402838daac1d611", "resourceType": "AWS::EC2::Instance", "resourceId": "i-03402838daac1d611", "awsRegion": "us-east-1", "availabilityZone": "us-east-1b", "resourceCreationTime": "2017-04-05T22:15:53.000Z", "tags": {}, "relatedEvents": [ "d3d87c29-bde3-4380-a7de-810f379246cc" ], "relationships": [ { "resourceType": "AWS::EC2::NetworkInterface", "resourceId": "eni-d055cfc4", "relationshipName": "Contains NetworkInterface" }, { "resourceType": "AWS::EC2::SecurityGroup", "resourceId": "sg-fd215482", "relationshipName": "Is associated with SecurityGroup" }, { "resourceType": "AWS::EC2::Subnet", "resourceId": "subnet-1aaccc7f", "relationshipName": "Is contained in Subnet" }, { "resourceType": "AWS::EC2::Volume", "resourceId": "vol-0c24aa343c564eda8", "relationshipName": "Is attached to Volume" }, { "resourceType": "AWS::EC2::VPC", "resourceId": "vpc-79b3ea1e", "relationshipName": "Is contained in Vpc" } ], "configuration": { "instanceId": "i-03402838daac1d611", "imageId": "ami-22ce4934", "state": { "code": 16, "name": "running" }, "privateDnsName": "ip-172-31-74-239.ec2.internal", "publicDnsName": "ec2-34-205-29-138.compute-1.amazonaws.com", "stateTransitionReason": "", "keyName": "ssm-key", "amiLaunchIndex": 0, "productCodes": [], "instanceType": "t2.micro", "launchTime": "2017-04-05T22:15:53.000Z", "placement": { "availabilityZone": "us-east-1b", "groupName": "", "tenancy": "default" }, "monitoring": { "state": "disabled" }, "subnetId": "subnet-1aaccc7f", "vpcId": "vpc-79b3ea1e", "privateIpAddress": "172.31.74.239", "publicIpAddress": "34.205.29.138", "architecture": "x86_64", "rootDeviceType": "ebs", "rootDeviceName": "/dev/xvda", "blockDeviceMappings": [ { "deviceName": "/dev/xvda", "ebs": { "volumeId": "vol-0c24aa343c564eda8", "status": "attached", "attachTime": "2017-04-05T22:15:54.000Z", "deleteOnTermination": true } } ], "virtualizationType": "hvm", "clientToken": "UuPNx1491430552432", "tags": [], "securityGroups": [ { "groupName": "launch-wizard-2", "groupId": "sg-fd215482" } ], "sourceDestCheck": true, "hypervisor": "xen", "networkInterfaces": [ { "networkInterfaceId": "eni-d055cfc4", "subnetId": "subnet-1aaccc7f", "vpcId": "vpc-79b3ea1e", "description": "", "ownerId": "681361479661", "status": "in-use", "macAddress": "02:3e:c3:cb:e9:da", "privateIpAddress": "172.31.74.239", "privateDnsName": "ip-172-31-74-239.ec2.internal", "sourceDestCheck": true, "groups": [ { "groupName": "launch-wizard-2", "groupId": "sg-fd215482" } ], "attachment": { "attachmentId": "eni-attach-e8d5c971", "deviceIndex": 0, "status": "attached", "attachTime": "2017-04-05T22:15:53.000Z", "deleteOnTermination": true }, "association": { "publicIp": "34.205.29.138", "publicDnsName": "ec2-34-205-29-138.compute-1.amazonaws.com", "ipOwnerId": "amazon" }, "privateIpAddresses": [ { "privateIpAddress": "172.31.74.239", "privateDnsName": "ip-172-31-74-239.ec2.internal", "primary": true, "association": { "publicIp": "34.205.29.138", "publicDnsName": "ec2-34-205-29-138.compute-1.amazonaws.com", "ipOwnerId": "amazon" } } ], "ipv6Addresses": [] } ], "ebsOptimized": true, "enaSupport": true }, "supplementaryConfiguration": {} } For more info, try checking: https://github.com/awslabs/aws-config-resource-schema/blob/master/config/properties/resource-types/
コードの詳細
再掲になりますが、今回実装したコードです。
(略) if configuration_item['resourceType'] != 'AWS::EC2::Instance': return 'NOT_APPLICABLE' if configuration_item['configuration']['instanceType'] == valid_rule_parameters['desiredInstanceType']: return 'COMPLIANT' return 'NON_COMPLIANT' (略)
configration_item
に入る設定項目が分かったところでこのルールの詳細について説明します。
まず、この関数の戻り値としては文字列でCOMPLIANT, NON_COMPLIANT or NOT_APPLICABLE
を選択することができます。
今回のルールでは
- 評価しているリソースのリソースタイプが
AWS::EC2::Instance
ではなかった場合、NOT_APPLICABLE
つまりコンプライアンスステータスを返さない。 - インスタンスタイプがパラメータの値と等しかったときに
COMPLIANT
つまり準拠と評価する。 - それ以外の場合には
NON_COMPLIANT
つまり非準拠として評価する。
という仕組みになっています。
さらに複雑なロジックを組みたい場合には、この関数の中、もしくは新たに関数を追加するなどして実装していきましょう。
ルールのテスト
概要編でもお伝えしましたが、
テストは、
$ rdk test-local <rulename>
で実行が可能で、特にテストファイル(今回だとec2_desired_instance_type_test.py
)に変更を加えていなければ以下の3つが実行されます。
- test_sample
- 基本的にここにテストコードを記述します
- test_sts_access_denied
- "Access Denied"エラーに対する動作をテストします。
- test_sts_unknown_error
- "Unknown Error"エラーに対する動作をテストします。
今回はこのec2_desired_instance_type_test.py
のtest_sample
にあたる部分に準拠/非準拠のリソースの設定項目を渡して想定通りの動作をするかのテストを作成していきます。
テストコードの実装
<変更前>
class ComplianceTest(unittest.TestCase): rule_parameters = '{"SomeParameterKey":"SomeParameterValue","SomeParameterKey2":"SomeParameterValue2"}' invoking_event_iam_role_sample = '{"configurationItem":{"relatedEvents":[],"relationships":[],"configuration":{},"tags":{},"configurationItemCaptureTime":"2018-07-02T03:37:52.418Z","awsAccountId":"123456789012","configurationItemStatus":"ResourceDiscovered","resourceType":"AWS::IAM::Role","resourceId":"some-resource-id","resourceName":"some-resource-name","ARN":"some-arn"},"notificationCreationTime":"2018-07-02T23:05:34.445Z","messageType":"ConfigurationItemChangeNotification"}' def setUp(self): pass def test_sample(self): self.assertTrue(True) # def test_sample_2(self): # RULE.ASSUME_ROLE_MODE = False # response = RULE.lambda_handler(build_lambda_configurationchange_event(self.invoking_event_iam_role_sample, self.rule_parameters), {}) # resp_expected = [] # resp_expected.append(build_expected_response('NOT_APPLICABLE', 'some-resource-id', 'AWS::IAM::Role')) # assert_successful_evaluation(self, response, resp_expected)
<実装したコード>
class ComplianceTest(unittest.TestCase): rule_parameters = '{"desiredInstanceType": "t2.micro"}' invoking_event_expected_compliant_sample = '{"configurationItem":{"relatedEvents":[],"relationships":[],"configuration":{"instanceType": "t2.micro"},"tags":{},"configurationItemCaptureTime":"2018-07-02T03:37:52.418Z","awsAccountId":"123456789012","configurationItemStatus":"ResourceDiscovered","resourceType":"AWS::EC2::Instance","resourceId":"some-resource-id","resourceName":"some-resource-name","ARN":"some-arn"},"notificationCreationTime":"2018-07-02T23:05:34.445Z","messageType":"ConfigurationItemChangeNotification"}' invoking_event_expected_non_compliant_sample = '{"configurationItem":{"relatedEvents":[],"relationships":[],"configuration":{"instanceType": "t3.micro"},"tags":{},"configurationItemCaptureTime":"2018-07-02T03:37:52.418Z","awsAccountId":"123456789012","configurationItemStatus":"ResourceDiscovered","resourceType":"AWS::EC2::Instance","resourceId":"some-resource-id","resourceName":"some-resource-name","ARN":"some-arn"},"notificationCreationTime":"2018-07-02T23:05:34.445Z","messageType":"ConfigurationItemChangeNotification"}' def test_expected_compliant(self): RULE.ASSUME_ROLE_MODE = False response = RULE.lambda_handler(build_lambda_configurationchange_event(self.invoking_event_expected_compliant_sample, self.rule_parameters), {}) resp_expected = [] resp_expected.append(build_expected_response('COMPLIANT', 'some-resource-id', 'AWS::EC2::Instance')) assert_successful_evaluation(self, response, resp_expected) def test_expected_non_compliant(self): RULE.ASSUME_ROLE_MODE = False response = RULE.lambda_handler(build_lambda_configurationchange_event(self.invoking_event_expected_non_compliant_sample, self.rule_parameters), {}) resp_expected = [] resp_expected.append(build_expected_response('NON_COMPLIANT', 'some-resource-id', 'AWS::EC2::Instance')) assert_successful_evaluation(self, response, resp_expected)
今回はデフォルトの形にある程度則ってテストコードを作成しています。
変更が必要な箇所を説明していきます。
1. [変数]rule_parameters
ルール作成のときに指定したパラメータである'{"desiredInstanceType": "t2.micro"}'
を指定します。
他にもcreate
コマンド実行の際に--input-parameters
パラメータで指定したパラメータがある場合にはここで指定しましょう。
2. [変数]invoking_event_expected_***
ここにはそれぞれCOMPLIANT、NON_COMPLIANTになると期待される設定項目を記述します。
- 変更点
- configuration
- resourceType
本記事のルールの作成 > ルールのロジックの実装 > リソースの設定項目(configration_item)の確認方法
で評価に必要な設定項目だけ抜き出してきます。
今回の場合だと{"configuration":{"instanceType": "t2.micro"}}
が評価に関わってきますので、その項目をこの変数のconfiguration
の箇所に追記しましょう。
また、resourceType
がデフォルトだとAWS::IAM::Role
になっているので適宜評価対象のリソースタイプに変更します(今回だとAWS::EC2::Instance
)。
3. [関数]test_expected_***()
関数内2行目のinvoking_event_iam_role_sample
を参照していた部分を先ほど作成した変更した変数名に置き換えます(invoking_event_expected_***
)。
関数内4行目の'NOT_APPLICABLE'
をそれぞれ期待される戻り値に変更します('COMPLIANT','NON_COMPLIANT'
)。
テストの実行
この状態で以下のコマンドを実行してみます。
$ rdk test-local ec2_desired_instance_type Running local test! Testing ec2_desired_instance_type Looking for tests in /home/ec2-user/rdk-test/ec2_desired_instance_type ec2_desired_instance_type_test.py Debug! <unittest.suite.TestSuite tests=[<unittest.suite.TestSuite tests=[<ec2_desired_instance_type_test.ComplianceTest testMethod=test_expected_compliant>, <ec2_desired_instance_type_test.ComplianceTest testMethod=test_expected_non_compliant>]>, <unittest.suite.TestSuite tests=[<ec2_desired_instance_type_test.TestStsErrors testMethod=test_sts_access_denied>, <ec2_desired_instance_type_test.TestStsErrors testMethod=test_sts_unknown_error>]>]> test_expected_compliant (ec2_desired_instance_type_test.ComplianceTest) ... ok test_expected_non_compliant (ec2_desired_instance_type_test.ComplianceTest) ... ok test_sts_access_denied (ec2_desired_instance_type_test.TestStsErrors) ... ok test_sts_unknown_error (ec2_desired_instance_type_test.TestStsErrors) ... ok ---------------------------------------------------------------------- Ran 4 tests in 0.004s OK <unittest.runner.TextTestResult run=4 errors=0 failures=0>
無事に4つのテストをクリアしました。
ここで例えば、test_expected_compliant()
の期待される戻り値を以下のように'NON_COMPLIANT'
に変更して実行してみます(関数内4行目)。
def test_expected_compliant(self): RULE.ASSUME_ROLE_MODE = False response = RULE.lambda_handler(build_lambda_configurationchange_event(self.invoking_event_expected_compliant_sample, self.rule_parameters), {}) resp_expected = [] resp_expected.append(build_expected_response('NON_COMPLIANT', 'some-resource-id', 'AWS::EC2::Instance')) assert_successful_evaluation(self, response, resp_expected)
$ rdk test-local ec2_desired_instance_type Running local test! Testing ec2_desired_instance_type Looking for tests in /home/ec2-user/rdk-test/ec2_desired_instance_type ec2_desired_instance_type_test.py Debug! <unittest.suite.TestSuite tests=[<unittest.suite.TestSuite tests=[<ec2_desired_instance_type_test.ComplianceTest testMethod=test_expected_compliant>, <ec2_desired_instance_type_test.ComplianceTest testMethod=test_expected_non_compliant>]>, <unittest.suite.TestSuite tests=[<ec2_desired_instance_type_test.TestStsErrors testMethod=test_sts_access_denied>, <ec2_desired_instance_type_test.TestStsErrors testMethod=test_sts_unknown_error>]>]> test_expected_compliant (ec2_desired_instance_type_test.ComplianceTest) ... FAIL test_expected_non_compliant (ec2_desired_instance_type_test.ComplianceTest) ... ok test_sts_access_denied (ec2_desired_instance_type_test.TestStsErrors) ... ok test_sts_unknown_error (ec2_desired_instance_type_test.TestStsErrors) ... ok ====================================================================== FAIL: test_expected_compliant (ec2_desired_instance_type_test.ComplianceTest) ---------------------------------------------------------------------- Traceback (most recent call last): File "/home/ec2-user/rdk-test/ec2_desired_instance_type/ec2_desired_instance_type_test.py", line 49, in test_expected_compliant assert_successful_evaluation(self, response, resp_expected) File "/home/ec2-user/rdk-test/ec2_desired_instance_type/ec2_desired_instance_type_test.py", line 125, in assert_successful_evaluation test_class.assertEquals(response_expected["ComplianceType"], response[i]["ComplianceType"]) AssertionError: 'NON_COMPLIANT' != 'COMPLIANT' - NON_COMPLIANT ? ---- + COMPLIANT ---------------------------------------------------------------------- Ran 4 tests in 0.005s FAILED (failures=1) <unittest.runner.TextTestResult run=4 errors=0 failures=1>
test_expected_compliant
がFAILになっていることがわかると思います。
このように自分が期待している結果が返ってこなかった場合に、ルールのロジックに誤りがある可能性をデプロイ前に発見することができます。
ルールのデプロイ
デプロイにはdeploy
コマンドを使用します。
$ rdk deploy ec2_desired_instance_type [ap-northeast-1]: Running deploy! [ap-northeast-1]: Found Custom Rule. [ap-northeast-1]: Zipping ec2_desired_instance_type [ap-northeast-1]: Uploading ec2_desired_instance_type [ap-northeast-1]: Upload complete. [ap-northeast-1]: Creating CloudFormation Stack for ec2_desired_instance_type [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: Waiting for CloudFormation stack operation to complete... [ap-northeast-1]: CloudFormation stack operation complete. [ap-northeast-1]: Config deploy complete.
コマンドを実行するとCloudFormationスタックが作成され、写真のようにリソースが作成されていることが分かります。
実際にAWSアカウント上にルールに準拠/非準拠となるEC2インスタンスを作成してAWSマネジメントコンソール上で評価結果を確認してみます。
期待した通りに動作していることが確認できました。
Lambdaを0から作成するケースとの比較
メリット
1. コアなロジックだけ考えれば良い
今回、RDKで作成したpythonファイルのMainCodeの部分にしか変更を加えておりません。
Lambdaを0から作成する場合にはコアの部分以外も記述する必要があるので、時間的なメリットがかなり大きいです。
2. 準拠/非準拠のリソースを想定したテストの実行が可能
こちらの7-dにあるとおり関数が正常に実行されるのは、LambdaがAWS Configからresult tokenを受け取った場合に限られます。 そのため、Lambdaを0から作成した場合、準拠/非準拠のリソースの設定項目を用意しても今回行ったようなテストを実施することはできません。
実際に準拠/非準拠のリソースをAWSアカウントに用意しなくてもコードでテストができるのはメリットと言えます。
デメリット
1. Lambdaのランタイムが限られている。
こちらは概要編でお伝えした内容になりますが、RDKのサポートしているLambdaのランタイムは現時点で下記のものになります。
- "java8",
- "python3.7"
- "python3.7-lib"
- "python3.8"
- "python3.8-lib"
- "python3.9"
- "python3.9-lib"
- "python3.10"
- "python3.10-lib"
もし他のランタイムを使用したい場合はRDKを使用せずにLambda関数を作成する必要があります。
まとめ
今回の記事では、RDKを使用した場合のルールのロジックとテストコードの書き方を中心に紹介させていただきました。
RDKについての記事は他にもありますのでよければ併せてご覧ください。
本記事が少しでもお役に立てれば幸いです。