Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

# Modellieren von benutzerdefinierten CloudFormation Hooks mit Python
<a name="hooks-model-python"></a>

Die Modellierung von benutzerdefinierten CloudFormation Hooks beinhaltet die Erstellung eines Schemas, das den Hook, seine Eigenschaften und seine Attribute definiert. Dieses Tutorial führt Sie durch die Modellierung benutzerdefinierter Hooks mit Python.

## Schritt 1: Generieren Sie das Hook-Projektpaket
<a name="model-hook-project-package-python"></a>

Generieren Sie Ihr Hook-Projektpaket. Die CloudFormation CLI erstellt leere Handler-Funktionen, die bestimmten Hook-Aktionen im Ziellebenszyklus entsprechen, wie in der Hook-Spezifikation definiert.

```
cfn generate
```

Der Befehl gibt die folgende Ausgabe zurück.

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

**Anmerkung**  
Stellen Sie sicher, dass Ihre Lambda-Laufzeiten up-to-date die Verwendung einer veralteten Version vermeiden. Weitere Informationen finden Sie unter [Lambda-Laufzeiten für Ressourcentypen und Hooks aktualisieren](https://docs.aws.amazon.com/cloudformation-cli/latest/userguide/runtime-update.html).

## Schritt 2: Hook-Handler hinzufügen
<a name="model-hook-project-add-handler"></a>

Fügen Sie Ihren eigenen Hook-Handler-Laufzeitcode zu den Handlern hinzu, die Sie implementieren möchten. Sie können beispielsweise den folgenden Code für die Protokollierung hinzufügen.

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

Die CloudFormation CLI generiert die `src/models.py` Datei aus dem[Referenz der Syntax des Hook-Konfigurationsschemas](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
```

## Schritt 3: Implementieren Sie Hook-Handler
<a name="model-hook-project-code-handler-python"></a>

Mit den generierten Python-Datenklassen können Sie die Handler schreiben, die die Funktionalität des Hooks tatsächlich implementieren. In diesem Beispiel implementieren Sie die `preDelete` Aufrufpunkte `preCreate``preUpdate`, und für die Handler.

**Topics**
+ [Implementieren Sie den PreCreate-Handler](#model-hook-project-code-handler-python-precreate)
+ [Implementieren Sie den PreUpdate-Handler](#model-hook-project-code-handler-python-preupdate)
+ [Implementieren Sie den PreDelete-Handler](#model-hook-project-code-handler-python-predelete)
+ [Implementieren Sie einen Hook-Handler](#model-hook-project-code-handler-python-hook-handler)

### Implementieren Sie den PreCreate-Handler
<a name="model-hook-project-code-handler-python-precreate"></a>

Der `preCreate` Handler überprüft die serverseitigen Verschlüsselungseinstellungen für eine `AWS::S3::Bucket ` Oder-Ressource. `AWS::SQS::Queue`
+ Für eine `AWS::S3::Bucket` Ressource ist der Hook nur erfolgreich, wenn Folgendes zutrifft.
  + Die Amazon S3 S3-Bucket-Verschlüsselung ist eingestellt.
  + Der Amazon S3 S3-Bucket-Schlüssel ist für den Bucket aktiviert.
  + Der für den Amazon S3 S3-Bucket festgelegte Verschlüsselungsalgorithmus ist der richtige erforderliche Algorithmus.
  + Die AWS Key Management Service Schlüssel-ID ist festgelegt.
+ Für eine `AWS::SQS::Queue` Ressource wird der Hook nur erfolgreich sein, wenn Folgendes zutrifft.
  + Die AWS Key Management Service Schlüssel-ID ist gesetzt.

### Implementieren Sie den PreUpdate-Handler
<a name="model-hook-project-code-handler-python-preupdate"></a>

Implementieren Sie einen `preUpdate` Handler, der vor den Aktualisierungsvorgängen für alle angegebenen Ziele im Handler initiiert. Der `preUpdate` Handler erreicht Folgendes:
+ Für eine `AWS::S3::Bucket` Ressource wird der Hook nur erfolgreich sein, wenn Folgendes zutrifft:
  + Der Bucket-Verschlüsselungsalgorithmus für einen Amazon S3 S3-Bucket wurde nicht geändert.

### Implementieren Sie den PreDelete-Handler
<a name="model-hook-project-code-handler-python-predelete"></a>

Implementieren Sie einen `preDelete` Handler, der vor den Löschvorgängen für alle angegebenen Ziele im Handler initiiert. Der `preDelete` Handler erreicht Folgendes:
+ Für eine `AWS::S3::Bucket` Ressource wird der Hook nur erfolgreich sein, wenn Folgendes zutrifft:
  + Überprüft, ob nach dem Löschen der Ressource die mindestens erforderlichen konformen Ressourcen im Konto vorhanden sind.
  + Die erforderliche Mindestmenge an konformen Ressourcen ist in der Konfiguration des Hooks festgelegt.

### Implementieren Sie einen Hook-Handler
<a name="model-hook-project-code-handler-python-hook-handler"></a>

1. Öffnen Sie in Ihrer IDE die `handlers.py` Datei, die sich im `src` Ordner befindet.

1. Ersetzen Sie den gesamten Inhalt der `handlers.py` Datei durch den folgenden Code.  
**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}")
   ```

Fahren Sie mit dem nächsten Thema, [Einen benutzerdefinierten Hook registrieren mit CloudFormation](registering-hooks.md), fort.