Le traduzioni sono generate tramite traduzione automatica. In caso di conflitto tra il contenuto di una traduzione e la versione originale in Inglese, quest'ultima prevarrà.

# Modellazione di CloudFormation hook personalizzati usando Python
<a name="hooks-model-python"></a>

La modellazione di CloudFormation Hooks personalizzati implica la creazione di uno schema che definisce l'Hook, le sue proprietà e i relativi attributi. Questo tutorial ti guida attraverso la modellazione di Hooks personalizzati usando Python.

## Passaggio 1: generare il pacchetto del progetto Hook
<a name="model-hook-project-package-python"></a>

Genera il tuo pacchetto di progetto Hook. La CloudFormation CLI crea funzioni di gestione vuote che corrispondono a specifiche azioni Hook nel ciclo di vita di destinazione, come definito nella specifica Hook.

```
cfn generate
```

Questo comando restituisce il seguente output.

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

**Nota**  
Assicurati che i runtime Lambda evitino l'uso up-to-date di una versione obsoleta. Per ulteriori informazioni, consulta [Aggiornamento dei runtime Lambda per i tipi di risorse](https://docs.aws.amazon.com/cloudformation-cli/latest/userguide/runtime-update.html) e gli hook.

## Passaggio 2: aggiungere i gestori Hook
<a name="model-hook-project-add-handler"></a>

Aggiungi il tuo codice di runtime del gestore Hook ai gestori che scegli di implementare. Ad esempio, puoi aggiungere il seguente codice per la registrazione.

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

La CloudFormation CLI genera il `src/models.py` file da. [Riferimento alla sintassi dello schema di configurazione dell’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
```

## Fase 3: Implementazione dei gestori Hook
<a name="model-hook-project-code-handler-python"></a>

Con le classi di dati Python generate, puoi scrivere i gestori che implementano effettivamente la funzionalità dell'Hook. In questo esempio, implementerete i punti `preCreate``preUpdate`, e di `preDelete` invocazione per i gestori.

**Topics**
+ [Implementa il gestore preCreate](#model-hook-project-code-handler-python-precreate)
+ [Implementa il gestore PreUpdate](#model-hook-project-code-handler-python-preupdate)
+ [Implementa il gestore PreDelete](#model-hook-project-code-handler-python-predelete)
+ [Implementa un gestore Hook](#model-hook-project-code-handler-python-hook-handler)

### Implementa il gestore preCreate
<a name="model-hook-project-code-handler-python-precreate"></a>

Il `preCreate` gestore verifica le impostazioni di crittografia sul lato server per una risorsa or. `AWS::S3::Bucket ` `AWS::SQS::Queue`
+ Per una `AWS::S3::Bucket` risorsa, l'Hook passerà solo se è vero quanto segue.
  + La crittografia del bucket Amazon S3 è impostata.
  + La chiave del bucket Amazon S3 è abilitata per il bucket.
  + L'algoritmo di crittografia impostato per il bucket Amazon S3 è l'algoritmo corretto richiesto.
  + L'ID della AWS Key Management Service chiave è impostato.
+ Per una `AWS::SQS::Queue` risorsa, l'Hook passerà solo se è vero quanto segue.
  + L'ID della AWS Key Management Service chiave è impostato.

### Implementa il gestore PreUpdate
<a name="model-hook-project-code-handler-python-preupdate"></a>

`preUpdate`Implementa un gestore, che si avvia prima delle operazioni di aggiornamento per tutte le destinazioni specificate nel gestore. Il `preUpdate` gestore esegue le seguenti operazioni:
+ Per una `AWS::S3::Bucket` risorsa, l'Hook passerà solo se è vero quanto segue:
  + L'algoritmo di crittografia dei bucket per un bucket Amazon S3 non è stato modificato.

### Implementa il gestore PreDelete
<a name="model-hook-project-code-handler-python-predelete"></a>

`preDelete`Implementa un gestore, che si avvia prima delle operazioni di eliminazione per tutte le destinazioni specificate nel gestore. Il `preDelete` gestore esegue le seguenti operazioni:
+ Per una `AWS::S3::Bucket` risorsa, l'Hook passerà solo se è vero quanto segue:
  + Verifica che le risorse conformi minime richieste esistano nell'account dopo l'eliminazione della risorsa.
  + La quantità minima di risorse conformi richiesta è impostata nella configurazione di Hook.

### Implementa un gestore Hook
<a name="model-hook-project-code-handler-python-hook-handler"></a>

1. Nel tuo IDE, apri il `handlers.py` file, che si trova nella `src` cartella.

1. Sostituisci l'intero contenuto del `handlers.py` file con il codice seguente.  
**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}")
   ```

Continua fino all'argomento successivo [Registrazione di un Hook personalizzato con CloudFormation](registering-hooks.md).