Ricevi e analizza i messaggi Pub/Sub relativi ai profili di dati

Questo documento fornisce esempi che mostrano come ricevere e analizzare le notifiche sulle modifiche ai profili di dati. Sensitive Data Protection invia questi aggiornamenti sotto forma di messaggi Pub/Sub.

Panoramica

Puoi configurare Sensitive Data Protection in modo da generare automaticamente profili per i dati in un'organizzazione, una cartella o un progetto. I profili dati contengono metriche e metadati sui tuoi dati e ti aiutano a determinare dove si trovano i dati sensibili e ad alto rischio. Sensitive Data Protection segnala queste metriche a vari livelli di dettaglio. Per informazioni sui tipi di dati che puoi profilare, consulta Risorse supportate.

Quando configuri il profiler dati, puoi attivare l'opzione che consente di pubblicare messaggi Pub/Sub ogni volta che si verificano modifiche significative ai profili dei dati. Questi messaggi consentono di intervenire immediatamente in risposta a queste modifiche. Di seguito sono riportati gli eventi che puoi ascoltare:

  • Viene profilata una tabella per la prima volta.
  • Un profilo viene aggiornato.
  • Il punteggio di rischio o sensibilità di un profilo aumenta.
  • Si è verificato un nuovo errore relativo ai profili di dati.

I messaggi Pub/Sub pubblicati dal profiler dati contengono un oggetto DataProfilePubSubMessage. Questi messaggi vengono sempre inviati in formato binario, quindi devi scrivere il codice che li riceve e li analizza.

Prezzi

Quando utilizzi Pub/Sub, i costi ti vengono addebitati in base ai prezzi di Pub/Sub.

Prima di iniziare

Questa pagina presuppone quanto segue:

Prima di iniziare a lavorare sugli esempi, segui questi passaggi:

  1. Crea un argomento Pub/Sub e aggiungi una sottoscrizione. Non assegnare uno schema all'argomento.

    Per semplicità, gli esempi in questa pagina ascoltano una sola sottoscrizione. Tuttavia, in pratica, puoi creare un argomento e un abbonamento per ogni evento supportato da Sensitive Data Protection.

  2. Se non lo hai già fatto, configura il profiler dati in modo da pubblicare i messaggi Pub/Sub:

    1. Modifica la configurazione della scansione.

    2. Nella pagina Modifica configurazione scansione, attiva l'opzione Pubblica in Pub/Sub e seleziona gli eventi che vuoi ascoltare. Poi, configura le impostazioni per ogni evento.

    3. Salva la configurazione della scansione.

  3. Concedi all'agente di servizio Sensitive Data Protection l'accesso per la pubblicazione nell'argomento Pub/Sub. Un esempio di ruolo con accesso in pubblicazione è il ruolo Publisher Pub/Sub (roles/pubsub.publisher). L'agente di servizio Sensitive Data Protection è un indirizzo email nel formato:

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

    Se utilizzi una configurazione di scansione a livello di organizzazione o cartella, PROJECT_NUMBER è l'identificatore numerico del container dell'agente di servizio. Se utilizzi una configurazione di scansione a livello di progetto, PROJECT_NUMBER è l'identificatore numerico del progetto.

  4. Installa e configura la libreria client di Sensitive Data Protection per Java o Python.

Esempi

I seguenti esempi mostrano come ricevere e analizzare i messaggi Pub/Sub pubblicati dal profiler dati. Puoi riadattare questi esempi ed eseguirne il deployment come funzioni Cloud Functions attivate da eventi Pub/Sub. Per ulteriori informazioni, consulta il tutorial su Pub/Sub (2ª generazione).

Negli esempi seguenti, sostituisci quanto segue:

  • PROJECT_ID: l'ID del progetto che contiene la sottoscrizione Pub/Sub.
  • SUBSCRIPTION_ID: l'ID della sottoscrizione 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.")

Passaggi successivi