Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

# Modélisation de CloudFormation Hooks personnalisés à l'aide de Python
<a name="hooks-model-python"></a>

La modélisation de CloudFormation Hooks personnalisés implique la création d'un schéma qui définit le Hook, ses propriétés et ses attributs. Ce didacticiel vous explique comment modéliser des Hooks personnalisés à l'aide de Python.

## Étape 1 : Générer le package du projet Hook
<a name="model-hook-project-package-python"></a>

Générez votre package de projet Hook. La CloudFormation CLI crée des fonctions de gestion vides qui correspondent à des actions Hook spécifiques dans le cycle de vie cible, telles que définies dans la spécification Hook.

```
cfn generate
```

La commande renvoie le résultat suivant.

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

**Note**  
Assurez-vous que vos environnements d'exécution Lambda doivent éviter up-to-date d'utiliser une version obsolète. Pour plus d'informations, consultez la section [Mise à jour des environnements d'exécution Lambda pour les types de ressources](https://docs.aws.amazon.com/cloudformation-cli/latest/userguide/runtime-update.html) et les Hooks.

## Étape 2 : Ajouter des gestionnaires Hook
<a name="model-hook-project-add-handler"></a>

Ajoutez votre propre code d'exécution du gestionnaire Hook aux gestionnaires que vous choisissez d'implémenter. Par exemple, vous pouvez ajouter le code suivant pour la journalisation.

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

La CloudFormation CLI génère le `src/models.py` fichier à partir du[Référence syntaxique du schéma de configuration Hook](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
```

## Étape 3 : Implémenter les gestionnaires Hook
<a name="model-hook-project-code-handler-python"></a>

Avec les classes de données Python générées, vous pouvez écrire les gestionnaires qui implémentent réellement les fonctionnalités du Hook. Dans cet exemple, vous allez implémenter les points `preCreate``preUpdate`, et `preDelete` d'invocation pour les gestionnaires.

**Topics**
+ [Implémenter le gestionnaire PreCreate](#model-hook-project-code-handler-python-precreate)
+ [Implémenter le gestionnaire PreUpdate](#model-hook-project-code-handler-python-preupdate)
+ [Implémenter le gestionnaire PreDelete](#model-hook-project-code-handler-python-predelete)
+ [Implémenter un gestionnaire Hook](#model-hook-project-code-handler-python-hook-handler)

### Implémenter le gestionnaire PreCreate
<a name="model-hook-project-code-handler-python-precreate"></a>

Le `preCreate` gestionnaire vérifie les paramètres de chiffrement côté serveur pour une ressource ou. `AWS::S3::Bucket ` `AWS::SQS::Queue`
+ Pour une `AWS::S3::Bucket` ressource, le Hook ne sera accepté que si les conditions suivantes sont vraies.
  + Le chiffrement du compartiment Amazon S3 est défini.
  + La clé du compartiment Amazon S3 est activée pour le compartiment.
  + L'algorithme de chiffrement défini pour le compartiment Amazon S3 est le bon algorithme requis.
  + L'identifiant de la AWS Key Management Service clé est défini.
+ Pour une `AWS::SQS::Queue` ressource, le Hook ne sera accepté que si les conditions suivantes sont vraies.
  + L'identifiant de la AWS Key Management Service clé est défini.

### Implémenter le gestionnaire PreUpdate
<a name="model-hook-project-code-handler-python-preupdate"></a>

Implémentez un `preUpdate` gestionnaire, qui démarre avant les opérations de mise à jour pour toutes les cibles spécifiées dans le gestionnaire. Le `preUpdate` gestionnaire effectue les opérations suivantes :
+ Pour une `AWS::S3::Bucket` ressource, le Hook ne sera accepté que si les conditions suivantes sont vraies :
  + L'algorithme de chiffrement des compartiments pour un compartiment Amazon S3 n'a pas été modifié.

### Implémenter le gestionnaire PreDelete
<a name="model-hook-project-code-handler-python-predelete"></a>

Implémentez un `preDelete` gestionnaire, qui démarre avant les opérations de suppression pour toutes les cibles spécifiées dans le gestionnaire. Le `preDelete` gestionnaire effectue les opérations suivantes :
+ Pour une `AWS::S3::Bucket` ressource, le Hook ne sera accepté que si les conditions suivantes sont vraies :
  + Vérifie que les ressources conformes minimales requises existeront dans le compte après la suppression de la ressource.
  + Le montant minimum de ressources conformes requis est défini dans la configuration du Hook.

### Implémenter un gestionnaire Hook
<a name="model-hook-project-code-handler-python-hook-handler"></a>

1. Dans votre IDE, ouvrez le `handlers.py` fichier situé dans le `src` dossier.

1. Remplacez l'intégralité du contenu du `handlers.py` fichier par le code suivant.  
**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}")
   ```

Passez à la rubrique suivante [Enregistrer un Hook personnalisé avec CloudFormation](registering-hooks.md).