Recevoir et analyser les messages Pub/Sub concernant les profils de données

Ce document fournit des exemples qui montrent comment recevoir et analyser les notifications concernant les modifications apportées à vos profils de données. La protection des données sensibles envoie ces mises à jour sous forme de messages Pub/Sub.

Présentation

Vous pouvez configurer Sensitive Data Protection pour qu'elle génère automatiquement des profils sur les données d'une organisation, d'un dossier ou d'un projet. Les profils de données contiennent des métriques et des métadonnées relatives à vos données, et vous permettent de déterminer l'emplacement des données sensibles et à haut risque. Sensitive Data Protection signale ces métriques à différents niveaux de détail. Pour en savoir plus sur les types de données que vous pouvez profiler, consultez Ressources compatibles.

Lorsque vous configurez le profileur de données, vous pouvez activer l'option permettant de publier des messages Pub/Sub chaque fois que des modifications importantes sont apportées à vos profils de données. Ces messages vous aident à prendre des mesures immédiates en réponse à ces changements. Voici les événements que vous pouvez écouter :

  • Un asset de données est profilé pour la première fois.
  • Un profil est mis à jour.
  • Le score de risque ou de sensibilité d'un profil augmente.
  • Une nouvelle erreur concernant vos profils de données s'est produite.

Les messages Pub/Sub publiés par le profileur de données contiennent un objet DataProfilePubSubMessage. Ces messages sont toujours envoyés au format binaire. Vous devez donc écrire du code qui les reçoit et les analyse.

Tarifs

Lorsque vous utilisez Pub/Sub, vous êtes facturé selon les tarifs de Pub/Sub.

Avant de commencer

Cette page suppose les éléments suivants :

Avant de commencer à travailler sur les exemples, procédez comme suit :

  1. Créez un sujet Pub/Sub et ajoutez-y un abonnement. N'attribuez pas de schéma au sujet.

    Pour des raisons de simplicité, les exemples figurant sur cette page n'écoutent qu'un seul abonnement. Toutefois, en pratique, vous pouvez créer un sujet et un abonnement pour chaque événement compatible avec Sensitive Data Protection.

  2. Si ce n'est pas déjà fait, configurez le profileur de données pour qu'il publie des messages Pub/Sub :

    1. Modifiez la configuration de votre analyse.

    2. Sur la page Modifier la configuration d'analyse, activez l'option Publier sur Pub/Sub et sélectionnez les événements que vous souhaitez écouter. Ensuite, configurez les paramètres de chaque événement.

    3. Enregistrez la configuration de l'analyse.

  3. Accordez à l'agent de service Sensitive Data Protection l'accès à la publication sur le sujet Pub/Sub. Le rôle Diffuseur Pub/Sub (roles/pubsub.publisher) est un exemple de rôle disposant d'un accès à la publication. L'agent de service Sensitive Data Protection est une adresse e-mail au format suivant :

    service-PROJECT_NUMBER@dlp-api.iam.gserviceaccount.com

    Si vous utilisez une configuration d'analyse au niveau de l'organisation ou du dossier, PROJECT_NUMBER correspond à l'identifiant numérique du conteneur de l'agent de service. Si vous utilisez une configuration d'analyse au niveau du projet, PROJECT_NUMBER correspond à l'identifiant numérique de votre projet.

  4. Installez et configurez la bibliothèque cliente Sensitive Data Protection pour Java ou Python.

Exemples

Les exemples suivants montrent comment recevoir et analyser les messages Pub/Sub publiés par le profileur de données. Vous pouvez réutiliser ces exemples et les déployer en tant que fonctions Cloud Run déclenchées par des événements Pub/Sub. Pour en savoir plus, consultez le tutoriel Pub/Sub (2e génération).

Dans les exemples suivants, remplacez les éléments suivants :

  • PROJECT_ID : ID du projet contenant l'abonnement Pub/Sub.
  • SUBSCRIPTION_ID : ID de l'abonnement Pub/Sub.

Java

import com.google.api.core.ApiService;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.privacy.dlp.v2.DataProfilePubSubMessage;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class DataProfilePubSubMessageParser {

  public static void main(String... args) throws Exception {
    String projectId = "PROJECT_ID";
    String subscriptionId = "SUBSCRIPTION_ID";
    int timeoutSeconds = 5;

    // The `ProjectSubscriptionName.of` method creates a fully qualified identifier
    // in the form `projects/{projectId}/subscriptions/{subscriptionId}`.
    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(projectId, subscriptionId);

    MessageReceiver receiver =
        (PubsubMessage pubsubMessage, AckReplyConsumer consumer) -> {
          try {
            DataProfilePubSubMessage message = DataProfilePubSubMessage.parseFrom(
                pubsubMessage.getData());
            System.out.println(
                "PubsubMessage with ID: " + pubsubMessage.getMessageId()
                    + "; message size: " + pubsubMessage.getData().size()
                    + "; event: " + message.getEvent()
                    + "; profile name: " + message.getProfile().getName()
                    + "; full resource: " + message.getProfile().getFullResource());
            consumer.ack();
          } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
          }
        };

    // Create subscriber client.
    Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();
    try {
      ApiService apiService = subscriber.startAsync();
      apiService.awaitRunning();
      System.out.printf("Listening for messages on %s for %d seconds.%n", subscriptionName,
          timeoutSeconds);
      subscriber.awaitTerminated(timeoutSeconds, TimeUnit.SECONDS);
    } catch (TimeoutException ignored) {
    } finally {
      subscriber.stopAsync();
    }
  }
}

Python

from google.cloud import pubsub_v1
from concurrent.futures import TimeoutError
from google.cloud import dlp_v2


project_id = "PROJECT_ID"
subscription_id = "SUBSCRIPTION_ID"
timeout = 5.0

subscriber = pubsub_v1.SubscriberClient()
# The `subscription_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/subscriptions/{subscription_id}`
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    print(f"Received {message.data}.")
    dlp_msg = dlp_v2.DataProfilePubSubMessage()
    dlp_msg._pb.ParseFromString(message.data)
    print("Parsed message: ", dlp_msg)
    print("--------")
    message.ack()

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path} for {timeout} seconds...")

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    try:
        # When `timeout` is not set, result() will block indefinitely,
        # unless an exception is encountered first.
        streaming_pull_future.result(timeout=timeout)
    except TimeoutError:
        streaming_pull_future.cancel()  # Trigger the shutdown.
        streaming_pull_future.result()  # Block until the shutdown is complete.
        print("Done waiting.")

Étapes suivantes