기계 번역으로 제공되는 번역입니다. 제공된 번역과 원본 영어의 내용이 상충하는 경우에는 영어 버전이 우선합니다.

# Python을 사용하여 사용자 지정 CloudFormation 후크 모델링
<a name="hooks-model-python"></a>

사용자 지정 CloudFormation 후크 모델링에는 후크, 해당 속성 및 속성을 정의하는 스키마를 생성하는 작업이 포함됩니다. 이 자습서에서는 Python을 사용하여 사용자 지정 후크를 모델링하는 방법을 안내합니다.

## 1단계: 후크 프로젝트 패키지 생성
<a name="model-hook-project-package-python"></a>

후크 프로젝트 패키지를 생성합니다. 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` 리소스의 경우 후크는 다음과 같은 경우에만 전달합니다.
  +  AWS Key Management Service 키 ID가 설정됩니다.

### preUpdate 핸들러 구현
<a name="model-hook-project-code-handler-python-preupdate"></a>

`preUpdate` 핸들러에서 지정된 모든 대상에 대한 업데이트 작업 전에 시작하는 핸들러를 구현합니다. `preUpdate` 핸들러는 다음을 수행합니다.
+ `AWS::S3::Bucket` 리소스의 경우 후크는 다음과 같은 경우에만 전달합니다.
  + Amazon S3 버킷의 버킷 암호화 알고리즘이 수정되지 않았습니다.

### preDelete 핸들러 구현
<a name="model-hook-project-code-handler-python-predelete"></a>

`preDelete` 핸들러에서 지정된 모든 대상에 대한 삭제 작업 전에 시작하는 핸들러를 구현합니다. `preDelete` 핸들러는 다음을 수행합니다.
+ `AWS::S3::Bucket` 리소스의 경우 후크는 다음과 같은 경우에만 전달합니다.
  + 리소스를 삭제한 후 계정에 필요한 최소 규정 준수 리소스가 존재하는지 확인합니다.
  + 필요한 최소 규정 준수 리소스 양은 후크의 구성에서 설정됩니다.

### 후크 핸들러 구현
<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) 단원으로 이동합니다.