Comment automatiser le chiffrement d'une instance Amazon RDS avec Python

Amazon RDS (Relational Database Service) est un service de base de données puissant et évolutif proposé par AWS, mais parfois, pour des raisons de conformité ou de sécurité, vous devez chiffrer une instance de base de données non chiffrée existante. Dans cet article, nous présenterons un script Python qui automatise le processus de migration d'une instance Amazon RDS non chiffrée vers une instance chiffrée.

Pourquoi chiffrer une instance RDS ?

Le chiffrement des instances RDS garantit que les données au repos sont sécurisées et répondent à diverses exigences de conformité, telles que PCI DSS, HIPAA, etc. Le chiffrement garantit que les sauvegardes, les instantanés et le stockage sous-jacent de votre base de données RDS sont automatiquement chiffrés.

Cependant, vous ne pouvez pas activer directement le chiffrement sur une instance RDS non chiffrée existante. Au lieu de cela, vous devez créer un instantané, copier cet instantané avec le chiffrement activé, puis restaurer une nouvelle instance RDS à partir de l'instantané chiffré.

C'est ce que nous allons automatiser dans ce tutoriel.

Conditions préalables

Pour suivre ce guide, vous aurez besoin de :

• Compte AWS : Accès à un compte AWS avec des autorisations pour gérer RDS et KMS (Key Management Service).
• Python 3.x : installé et configuré sur votre machine locale.
• Boto3 : Le SDK AWS pour Python, que vous pouvez installer à l'aide de pip :

  pip install boto3

Vous aurez également besoin des informations d'identification AWS suivantes :

1. AWS_ACCESS_KEY_ID
2. AWS_SECRET_ACCESS_KEY
3. AWS_DEFAULT_REGION

Le processus de migration du chiffrement

Ce script Python automatise les étapes suivantes :

1. Créer un instantané : prenez un instantané de l'instance RDS non chiffrée existante.
2. Copier l'instantané avec chiffrement : créez une copie chiffrée de l'instantané à l'aide d'AWS KMS (Key Management Service).
3. Restaurer la base de données : créez une nouvelle instance RDS à partir de l'instantané chiffré.

Script Python pour automatiser la migration

import boto3
import time
from botocore.exceptions import WaiterError

class RDSEncryptionMigrator:
    def __init__(self, source_db_identifier, target_db_identifier, kms_key_alias, region='us-east-1'):
        self.source_db_identifier = source_db_identifier
        self.target_db_identifier = target_db_identifier
        self.kms_key_alias = kms_key_alias if kms_key_alias.startswith('alias/') else f'alias/{kms_key_alias}'
        self.region = region

        self.rds_client = boto3.client('rds', region_name=region)
        self.kms_client = boto3.client('kms', region_name=region)

    def get_kms_key_id(self):
        """Get the KMS key ID from the alias"""
        try:
            response = self.kms_client.describe_key(
                KeyId=self.kms_key_alias
            )
            return response['KeyMetadata']['Arn']
        except Exception as e:
            print(f"Error getting KMS key ID from alias: {e}")
            raise

    def create_snapshot(self, snapshot_identifier):
        print(f"Creating snapshot of source database: {self.source_db_identifier}")
        response = self.rds_client.create_db_snapshot(
            DBSnapshotIdentifier=snapshot_identifier,
            DBInstanceIdentifier=self.source_db_identifier
        )

        # Wait for snapshot to be available
        waiter = self.rds_client.get_waiter('db_snapshot_available')
        try:
            waiter.wait(
                DBSnapshotIdentifier=snapshot_identifier,
                WaiterConfig={'Delay': 30, 'MaxAttempts': 60}
            )
        except WaiterError as e:
            print(f"Error waiting for snapshot: {e}")
            raise

        return response['DBSnapshot']['DBSnapshotArn']

    def create_encrypted_snapshot_copy(self, source_snapshot_id, encrypted_snapshot_id):
        print("Creating encrypted copy of snapshot")
        kms_key_id = self.get_kms_key_id()
        response = self.rds_client.copy_db_snapshot(
            SourceDBSnapshotIdentifier=source_snapshot_id,
            TargetDBSnapshotIdentifier=encrypted_snapshot_id,
            KmsKeyId=kms_key_id,
            CopyTags=True,
            SourceRegion=self.region
        )

        # Wait for encrypted snapshot to be available
        waiter = self.rds_client.get_waiter('db_snapshot_available')
        try:
            waiter.wait(
                DBSnapshotIdentifier=encrypted_snapshot_id,
                WaiterConfig={'Delay': 30, 'MaxAttempts': 60}
            )
        except WaiterError as e:
            print(f"Error waiting for encrypted snapshot: {e}")
            raise

        return response['DBSnapshot']['DBSnapshotArn']

    def restore_from_snapshot(self, snapshot_identifier):
        print(f"Restoring new encrypted database from snapshot")

        # Get source DB instance details
        source_db = self.rds_client.describe_db_instances(
            DBInstanceIdentifier=self.source_db_identifier
        )['DBInstances'][0]

        # Restore the encrypted instance
        response = self.rds_client.restore_db_instance_from_db_snapshot(
            DBInstanceIdentifier=self.target_db_identifier,
            DBSnapshotIdentifier=snapshot_identifier,
            DBInstanceClass=source_db['DBInstanceClass'],
            VpcSecurityGroupIds=self._get_security_group_ids(source_db),
            DBSubnetGroupName=source_db['DBSubnetGroup']['DBSubnetGroupName'],
            PubliclyAccessible=source_db['PubliclyAccessible'],
            MultiAZ=source_db['MultiAZ']
        )

        # Wait for the new instance to be available
        waiter = self.rds_client.get_waiter('db_instance_available')
        try:
            waiter.wait(
                DBInstanceIdentifier=self.target_db_identifier,
                WaiterConfig={'Delay': 30, 'MaxAttempts': 60}
            )
        except WaiterError as e:
            print(f"Error waiting for database restoration: {e}")
            raise

        return response['DBInstance']['DBInstanceArn']

    def _get_security_group_ids(self, db_instance):
        return [sg['VpcSecurityGroupId'] for sg in db_instance['VpcSecurityGroups']]

    def perform_encryption(self):
        try:
            # Create timestamp for unique identifiers
            timestamp = int(time.time())

            # Step 1: Create initial snapshot
            snapshot_id = f"{self.source_db_identifier}-snapshot-{timestamp}"
            self.create_snapshot(snapshot_id)

            # Step 2: Create encrypted copy of the snapshot
            encrypted_snapshot_id = f"{self.source_db_identifier}-encrypted-snapshot-{timestamp}"
            self.create_encrypted_snapshot_copy(snapshot_id, encrypted_snapshot_id)

            # Step 3: Restore from encrypted snapshot
            self.restore_from_snapshot(encrypted_snapshot_id)

            print(f"""
            Encryption process completed successfully!
            New encrypted database instance: {self.target_db_identifier}

            Next steps:
            1. Verify the new encrypted instance
            2. Update your application connection strings
            3. Once verified, you can delete the old unencrypted instance
            """)

        except Exception as e:
            print(f"Error during encryption process: {e}")
            raise

def main():
    # These values should ideally come from environment variables or command line arguments
    source_db_identifier = 'database-2'
    target_db_identifier = 'database-2-enc'
    kms_key_alias = 'aws/rds'
    region = 'us-east-1'

    migrator = RDSEncryptionMigrator(
        source_db_identifier=source_db_identifier,
        target_db_identifier=target_db_identifier,
        kms_key_alias=kms_key_alias,
        region=region
    )

    migrator.perform_encryption()

if __name__ == '__main__':
    main()

Comment fonctionne le script

Le script définit une classe RDSEncryptionMigrator qui gère :

1. Création d'un instantané : Un instantané de la base de données source est créé.
2. Copie d'instantané chiffré : l'instantané est copié et chiffré à l'aide de l'alias de clé KMS fourni.
3. Restauration de la base de données : l'instantané chiffré est utilisé pour restaurer une nouvelle instance RDS.

Conclusion

En utilisant le script fourni, vous pouvez automatiser le processus de cryptage de vos bases de données RDS et garantir la sécurité de vos données. Cette approche élimine le besoin d’intervention manuelle et réduit le risque d’erreur humaine dans le processus de migration. Assurez-vous de vérifier la nouvelle instance chiffrée, de mettre à jour les chaînes de connexion de votre application et de supprimer l'ancienne instance non chiffrée une fois que vous êtes prêt.

Si vous souhaitez étendre davantage cette fonctionnalité, vous pouvez intégrer ce script à AWS Lambda ou AWS Step Functions pour automatiser davantage le processus au sein de votre pipeline CI/CD.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!