翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

# Python を使用したカスタム CloudFormation フックのモデリング
<a name="hooks-model-python"></a>

カスタム CloudFormation フックのモデリングには、フック、そのプロパティ、および属性を定義するスキーマの作成が含まれます。このチュートリアルでは、Python を使用してカスタムフックをモデリングする方法について説明します。

## ステップ 1: Hook プロジェクトパッケージを生成する
<a name="model-hook-project-package-python"></a>

Hook プロジェクトパッケージを生成します。CloudFormation CLI は、フック仕様で定義されているように、ターゲットライフサイクル内の特定のフックアクションに対応する空のハンドラー関数を作成します。

```
cfn generate
```

このコマンドは、以下の出力を返します。

```
Generated files for MyCompany::Testing::MyTestHook
```

**注記**  
廃止されたバージョンを使用しないように、Lambda ランタイムがup-to-dateであることを確認します。詳細については、[「リソースタイプとフックの Lambda ランタイムの更新](https://docs.aws.amazon.com/cloudformation-cli/latest/userguide/runtime-update.html)」を参照してください。

## ステップ 2: フックハンドラーを追加する
<a name="model-hook-project-add-handler"></a>

実装するハンドラーに独自のフックハンドラーランタイムコードを追加します。たとえば、ログ記録に次のコードを追加できます。

```
LOG.setLevel(logging.INFO)
LOG.info("Internal testing Hook triggered for target: " + request.hookContext.targetName);
```

CloudFormation CLI は、 から `src/models.py` ファイルを生成します[フック設定スキーマ構文リファレンス](hook-configuration-schema.md)。

**Example models.py**  

```
import sys
from dataclasses import dataclass
from inspect import getmembers, isclass
from typing import (
    AbstractSet,
    Any,
    Generic,
    Mapping,
    MutableMapping,
    Optional,
    Sequence,
    Type,
    TypeVar,
)

from cloudformation_cli_python_lib.interface import (
    BaseModel,
    BaseHookHandlerRequest,
)
from cloudformation_cli_python_lib.recast import recast_object
from cloudformation_cli_python_lib.utils import deserialize_list

T = TypeVar("T")


def set_or_none(value: Optional[Sequence[T]]) -> Optional[AbstractSet[T]]:
    if value:
        return set(value)
    return None


@dataclass
class HookHandlerRequest(BaseHookHandlerRequest):
    pass


@dataclass
class TypeConfigurationModel(BaseModel):
    limitSize: Optional[str]
    cidr: Optional[str]
    encryptionAlgorithm: Optional[str]

    @classmethod
    def _deserialize(
        cls: Type["_TypeConfigurationModel"],
        json_data: Optional[Mapping[str, Any]],
    ) -> Optional["_TypeConfigurationModel"]:
        if not json_data:
            return None
        return cls(
            limitSize=json_data.get("limitSize"),
            cidr=json_data.get("cidr"),
            encryptionAlgorithm=json_data.get("encryptionAlgorithm"),
        )


_TypeConfigurationModel = TypeConfigurationModel
```

## ステップ 3: フックハンドラーを実装する
<a name="model-hook-project-code-handler-python"></a>

Python データクラスを生成すると、実際にフックの機能を実装するハンドラーを記述できます。この例では、ハンドラーの `preCreate`、`preUpdate`、および `preDelete`呼び出しポイントを実装します。

**Topics**
+ [preCreate ハンドラーを実装する](#model-hook-project-code-handler-python-precreate)
+ [preUpdate ハンドラーを実装する](#model-hook-project-code-handler-python-preupdate)
+ [preDelete ハンドラーを実装する](#model-hook-project-code-handler-python-predelete)
+ [フックハンドラーの実装](#model-hook-project-code-handler-python-hook-handler)

### preCreate ハンドラーを実装する
<a name="model-hook-project-code-handler-python-precreate"></a>

`preCreate` ハンドラーは、 `AWS::S3::Bucket `または `AWS::SQS::Queue`リソースのサーバー側の暗号化設定を検証します。
+ `AWS::S3::Bucket` リソースの場合、フックは以下が true の場合にのみ渡されます。
  + Amazon S3 バケット暗号化が設定されています。
  + Amazon S3 バケットキーはバケットに対して有効になっています。
  + Amazon S3 バケットに設定された暗号化アルゴリズムは、必要な正しいアルゴリズムです。
  +  AWS Key Management Service キー ID が設定されます。
+ `AWS::SQS::Queue` リソースの場合、フックは以下が true の場合にのみ渡されます。
  +  AWS Key Management Service キー ID が設定されます。

### preUpdate ハンドラーを実装する
<a name="model-hook-project-code-handler-python-preupdate"></a>

`preUpdate` ハンドラーを実装します。ハンドラーは、ハンドラー内のすべての指定されたターゲットの更新オペレーションの前に開始されます。`preUpdate` ハンドラーは以下を実行します。
+ `AWS::S3::Bucket` リソースの場合、フックは以下が true の場合にのみ渡されます。
  + Amazon S3 バケットのバケット暗号化アルゴリズムは変更されていません。

### preDelete ハンドラーを実装する
<a name="model-hook-project-code-handler-python-predelete"></a>

`preDelete` ハンドラーを実装します。ハンドラーは、ハンドラー内のすべての指定されたターゲットの削除オペレーションの前に開始されます。`preDelete` ハンドラーは以下を実行します。
+ `AWS::S3::Bucket` リソースの場合、フックは以下が true の場合にのみ渡されます。
  + リソースを削除した後、最低限必要な準拠リソースがアカウントに存在することを確認します。
  + 最低限必要な準拠リソースの量は、フックの設定で設定されます。

### フックハンドラーの実装
<a name="model-hook-project-code-handler-python-hook-handler"></a>

1. IDE で、 `src`フォルダにある `handlers.py` ファイルを開きます。

1. `handlers.py` ファイルの内容全体を次のコードに置き換えます。  
**Example handlers.py**  

   ```
   import logging
   from typing import Any, MutableMapping, Optional
   import botocore
   
   from cloudformation_cli_python_lib import (
       BaseHookHandlerRequest,
       HandlerErrorCode,
       Hook,
       HookInvocationPoint,
       OperationStatus,
       ProgressEvent,
       SessionProxy,
       exceptions,
   )
   
   from .models import HookHandlerRequest, TypeConfigurationModel
   
   # Use this logger to forward log messages to CloudWatch Logs.
   LOG = logging.getLogger(__name__)
   TYPE_NAME = "MyCompany::Testing::MyTestHook"
   
   LOG.setLevel(logging.INFO)
   
   hook = Hook(TYPE_NAME, TypeConfigurationModel)
   test_entrypoint = hook.test_entrypoint
   
   
   def _validate_s3_bucket_encryption(
       bucket: MutableMapping[str, Any], required_encryption_algorithm: str
   ) -> ProgressEvent:
       status = None
       message = ""
       error_code = None
   
       if bucket:
           bucket_name = bucket.get("BucketName")
   
           bucket_encryption = bucket.get("BucketEncryption")
           if bucket_encryption:
               server_side_encryption_rules = bucket_encryption.get(
                   "ServerSideEncryptionConfiguration"
               )
               if server_side_encryption_rules:
                   for rule in server_side_encryption_rules:
                       bucket_key_enabled = rule.get("BucketKeyEnabled")
                       if bucket_key_enabled:
                           server_side_encryption_by_default = rule.get(
                               "ServerSideEncryptionByDefault"
                           )
   
                           encryption_algorithm = server_side_encryption_by_default.get(
                               "SSEAlgorithm"
                           )
                           kms_key_id = server_side_encryption_by_default.get(
                               "KMSMasterKeyID"
                           )  # "KMSMasterKeyID" is name of the property for an AWS::S3::Bucket
   
                           if encryption_algorithm == required_encryption_algorithm:
                               if encryption_algorithm == "aws:kms" and not kms_key_id:
                                   status = OperationStatus.FAILED
                                   message = f"KMS Key ID not set for bucket with name: f{bucket_name}"
                               else:
                                   status = OperationStatus.SUCCESS
                                   message = f"Successfully invoked PreCreateHookHandler for AWS::S3::Bucket with name: {bucket_name}"
                           else:
                               status = OperationStatus.FAILED
                               message = f"SSE Encryption Algorithm is incorrect for bucket with name: {bucket_name}"
                       else:
                           status = OperationStatus.FAILED
                           message = f"Bucket key not enabled for bucket with name: {bucket_name}"
   
                       if status == OperationStatus.FAILED:
                           break
               else:
                   status = OperationStatus.FAILED
                   message = f"No SSE Encryption configurations for bucket with name: {bucket_name}"
           else:
               status = OperationStatus.FAILED
               message = (
                   f"Bucket Encryption not enabled for bucket with name: {bucket_name}"
               )
       else:
           status = OperationStatus.FAILED
           message = "Resource properties for S3 Bucket target model are empty"
   
       if status == OperationStatus.FAILED:
           error_code = HandlerErrorCode.NonCompliant
   
       return ProgressEvent(status=status, message=message, errorCode=error_code)
   
   
   def _validate_sqs_queue_encryption(queue: MutableMapping[str, Any]) -> ProgressEvent:
       if not queue:
           return ProgressEvent(
               status=OperationStatus.FAILED,
               message="Resource properties for SQS Queue target model are empty",
               errorCode=HandlerErrorCode.NonCompliant,
           )
       queue_name = queue.get("QueueName")
   
       kms_key_id = queue.get(
           "KmsMasterKeyId"
       )  # "KmsMasterKeyId" is name of the property for an AWS::SQS::Queue
       if not kms_key_id:
           return ProgressEvent(
               status=OperationStatus.FAILED,
               message=f"Server side encryption turned off for queue with name: {queue_name}",
               errorCode=HandlerErrorCode.NonCompliant,
           )
   
       return ProgressEvent(
           status=OperationStatus.SUCCESS,
           message=f"Successfully invoked PreCreateHookHandler for targetAWS::SQS::Queue with name: {queue_name}",
       )
   
   
   @hook.handler(HookInvocationPoint.CREATE_PRE_PROVISION)
   def pre_create_handler(
       session: Optional[SessionProxy],
       request: HookHandlerRequest,
       callback_context: MutableMapping[str, Any],
       type_configuration: TypeConfigurationModel,
   ) -> ProgressEvent:
       target_name = request.hookContext.targetName
       if "AWS::S3::Bucket" == target_name:
           return _validate_s3_bucket_encryption(
               request.hookContext.targetModel.get("resourceProperties"),
               type_configuration.encryptionAlgorithm,
           )
       elif "AWS::SQS::Queue" == target_name:
           return _validate_sqs_queue_encryption(
               request.hookContext.targetModel.get("resourceProperties")
           )
       else:
           raise exceptions.InvalidRequest(f"Unknown target type: {target_name}")
   
   
   def _validate_bucket_encryption_rules_not_updated(
       resource_properties, previous_resource_properties
   ) -> ProgressEvent:
       bucket_encryption_configs = resource_properties.get("BucketEncryption", {}).get(
           "ServerSideEncryptionConfiguration", []
       )
       previous_bucket_encryption_configs = previous_resource_properties.get(
           "BucketEncryption", {}
       ).get("ServerSideEncryptionConfiguration", [])
   
       if len(bucket_encryption_configs) != len(previous_bucket_encryption_configs):
           return ProgressEvent(
               status=OperationStatus.FAILED,
               message=f"Current number of bucket encryption configs does not match previous. Current has {str(len(bucket_encryption_configs))} configs while previously there were {str(len(previous_bucket_encryption_configs))} configs",
               errorCode=HandlerErrorCode.NonCompliant,
           )
   
       for i in range(len(bucket_encryption_configs)):
           current_encryption_algorithm = (
               bucket_encryption_configs[i]
               .get("ServerSideEncryptionByDefault", {})
               .get("SSEAlgorithm")
           )
           previous_encryption_algorithm = (
               previous_bucket_encryption_configs[i]
               .get("ServerSideEncryptionByDefault", {})
               .get("SSEAlgorithm")
           )
   
           if current_encryption_algorithm != previous_encryption_algorithm:
               return ProgressEvent(
                   status=OperationStatus.FAILED,
                   message=f"Bucket Encryption algorithm can not be changed once set. The encryption algorithm was changed to {current_encryption_algorithm} from {previous_encryption_algorithm}.",
                   errorCode=HandlerErrorCode.NonCompliant,
               )
   
       return ProgressEvent(
           status=OperationStatus.SUCCESS,
           message="Successfully invoked PreUpdateHookHandler for target: AWS::SQS::Queue",
       )
   
   
   def _validate_queue_encryption_not_disabled(
       resource_properties, previous_resource_properties
   ) -> ProgressEvent:
       if previous_resource_properties.get(
           "KmsMasterKeyId"
       ) and not resource_properties.get("KmsMasterKeyId"):
           return ProgressEvent(
               status=OperationStatus.FAILED,
               errorCode=HandlerErrorCode.NonCompliant,
               message="Queue encryption can not be disable",
           )
       else:
           return ProgressEvent(status=OperationStatus.SUCCESS)
   
   
   @hook.handler(HookInvocationPoint.UPDATE_PRE_PROVISION)
   def pre_update_handler(
       session: Optional[SessionProxy],
       request: BaseHookHandlerRequest,
       callback_context: MutableMapping[str, Any],
       type_configuration: MutableMapping[str, Any],
   ) -> ProgressEvent:
       target_name = request.hookContext.targetName
       if "AWS::S3::Bucket" == target_name:
           resource_properties = request.hookContext.targetModel.get("resourceProperties")
           previous_resource_properties = request.hookContext.targetModel.get(
               "previousResourceProperties"
           )
   
           return _validate_bucket_encryption_rules_not_updated(
               resource_properties, previous_resource_properties
           )
       elif "AWS::SQS::Queue" == target_name:
           resource_properties = request.hookContext.targetModel.get("resourceProperties")
           previous_resource_properties = request.hookContext.targetModel.get(
               "previousResourceProperties"
           )
   
           return _validate_queue_encryption_not_disabled(
               resource_properties, previous_resource_properties
           )
       else:
           raise exceptions.InvalidRequest(f"Unknown target type: {target_name}")
   ```

次のトピック「[カスタムフックを に登録する CloudFormation](registering-hooks.md)」に進みます。