Cloud DLP で BigQuery テーブルの特定のカラムの機密情報をマスキングしてみた〜カスタム infoType の指定方法を添えて〜

クラスメソッド Google Cloud Advent Calendar 2021

みかみ

2021.12.16

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは、みかみです。

本記事は、クラスメソッド Google Cloud Advent Calendar 2021 の 16 日目のエントリです。

Google Cloud 好きなクラスメソッド社員が紡ぐ今年のクラスメソッド Google Cloud Advent Calendar 2021、最終日の25日までお楽しみいただけたら幸いです！

やりたいこと

Cloud Data Loss Prevention（DLP）で、BigQueryのテーブルデータをマスキングしたい
BigQueryテーブルの特定のカラムのみをマスキングしたい
組み込み infoType ではマスキングされなかった項目を、カスタム infoType を使ってマスキングしたい

Google Cloud DLP については、下記エントリでもご紹介しているので、ご参照いただけますと幸いです。

Google Cloud の DLP（Data Loss Prevention）サービスである Cloud DLP によるマスキング処理は、以前も検証したことがありますが、

漢字でもOK？！ DLP API で個人情報をマスキングしてから BigQuery にロードしてみた。｜ DevelopersIO

BigQuery のデータを分析に利用する場合、特定のカラムの個人情報だけをマスキングしたいケースもあるかと思います。

なお Cloud DLP では、組み込み infoType を指定して氏名や電話番号、住所など、特定の種類の個人情報を検出＆匿名化することが可能ですが、組み込み infoType は国や地域に固有のデータタイプが定義されており、日本語の氏名や住所などが 100% 検出される保証はないので、本番環境での実運用に利用される場合は事前のデータ検証などを合わせてご検討ください。

前提

Google Cloud Python クライアントライブラリの実行環境は準備済みであるものとします。

本エントリでは、Cloud Shell を使用しました。

Cloud Shell の使用｜ Cloud Shell ドキュメント

サンプルデータを準備

とあるコールセンターのデータを想定します。

データには、担当者情報とお客様情報、問い合わせ内容や他の担当者への引継ぎ事項のフリーテキストが含まれるものとします。担当者の個人情報のマスキングは不要ですが、お客様の個人情報はマスキングが必要です。

BigQuery に、以下のテーブルを準備しました。

user_name と user_tel はお客様の個人情報なのでマスキングが必要で、message カラムにもお客様の個人情報が含まれる可能性があるため、該当箇所だけマスキングが必要です。

なお、テーブルに格納済みのデータは下記サイトで作成したサンプルデータを元にしており、実在の人物や電話番号、地名とは一切関連ありません。

検査・匿名化テンプレートを作成

Google Cloud 管理コンソール、ナビゲーションメニュー「セキュリティ」から「データ損失防止（DLP）」を選択します。

画面上部の「構成」タブから、検査用と匿名化用の2つのテンプレートを作成します。

まずは検査用のテンプレートから作成します。

「① テンプレートの定義」に任意のID、表示名、説明を入力し、リソースロケーションは「グローバル」のままで「続行」。

「② 検出の設定」の「INFOTYPE を管理」リンクから、検出対象とするデータの種別を組み込み infoType から指定して「完了」したら、画面一番下の「作成」ボタンをクリック。

検査テンプレートが作成できました。

同様に、匿名化テンプレートを作成します。「① テンプレートの定義」の「データ変換のタイプ」で「レコード」を指定すると、構造化（CSV or TSV）データの特定の項目に対してのみマスキングを実行することができます。

「② 匿名化の構成」「変換ルール」の「フィールド、列、条件」欄に、マスキング対象のカラム名を入力し、「変換を追加」リンクから「変換方法」プルダウンで「infoType 名での置換」を選択して「作成」します。

匿名化テンプレートも作成できました。

DLP API で特定の項目を匿名化

準備した BigQuery テーブルのサンプルデータを取得し、DLP API でマスキングしたデータを再度 BigQuery にロードする、以下の Python コードを準備しました。

from google.cloud import bigquery
import google.cloud.dlp
from datetime import datetime
import io
import csv

# BigQueryからテーブルヘッダ、データを取得
table_id = 'dataset_1.sample_callcenter'
query = """
    SELECT * FROM dataset_1.sample_callcenter
"""

client = bigquery.Client()
table = client.get_table(table_id)
headers = [col.name for col in table.schema]

query_job = client.query(query)
rows = [[col.strftime('%Y-%m-%d %H:%M:%S') if isinstance(col, datetime) else str(col) for col in row] for row in query_job]


# DLPでマスキング
project = 'cm-da-mikami-yuki-258308'
inspect_template = 'projects/cm-da-mikami-yuki-258308/locations/global/inspectTemplates/sample_callcenter'
deidentify_template = 'projects/cm-da-mikami-yuki-258308/locations/global/deidentifyTemplates/sample_callcenter'

def map_headers(header):
    return {"name": header}

def map_data(value):
    return {"string_value": value}

def map_rows(row):
    return {"values": map(map_data, row)}

csv_headers = map(map_headers, headers)
csv_rows = map(map_rows, rows)

table_item = {"table": {"headers": csv_headers, "rows": csv_rows}}

dlp = google.cloud.dlp_v2.DlpServiceClient()
parent = f"projects/{project}"
response = dlp.deidentify_content(
    request={
        "parent": parent,
        "inspect_template_name": inspect_template,
        "deidentify_template_name": deidentify_template,
        "item": table_item,
    }
)

def write_header(header):
    return header.name

def write_data(data):
    return data.string_value

with io.StringIO() as csvfile:
    write_file = csv.writer(csvfile, delimiter=",")
    write_file.writerow(map(write_header, response.item.table.headers))
    for row in response.item.table.rows:
        write_file.writerow(map(write_data, row.values))
    obj_csv = io.BytesIO(csvfile.getvalue().encode('utf-8'))


# マスキング済みデータをBigQueryにロード
table_id_dst = 'dataset_1.sample_callcenter_dlp'

client = bigquery.Client()
job_config = bigquery.LoadJobConfig(
    autodetect=True,
    write_disposition='WRITE_TRUNCATE',
    source_format='CSV',
    skip_leading_rows=1
)
load_job = client.load_table_from_file(
    obj_csv, table_id_dst, job_config=job_config
)
load_job.result()

はじめに、BigQuery のテーブル情報からカラム名を取得し、さらに SELECT クエリを実行して、dataset_1.sample_callcenter テーブルデータを取得します。

from google.cloud import bigquery
from datetime import datetime

# BigQueryからテーブルヘッダ、データを取得
table_id = 'dataset_1.sample_callcenter'
query = """
    SELECT * FROM dataset_1.sample_callcenter
"""

client = bigquery.Client()
table = client.get_table(table_id)
headers = [col.name for col in table.schema]

query_job = client.query(query)
rows = [[col.strftime('%Y-%m-%d %H:%M:%S') if isinstance(col, datetime) else str(col) for col in row] for row in query_job]

続いて、取得したテーブルカラムとデータを DLP API へのリクエストパラメータ用のデータフォーマットに変換し、あらかじめ作成しておいた検査・匿名化テンプレートを指定してマスキングを実行します。 API レスポンスのマスキング済みデータは、この後 BigQuery にロードするため、CSV ファイルオブジェクト形式に変換しておきます。

構造化データの匿名化処理部分は、以下のサンプルコードを参考にさせていただきました。

googleapis/python-dlp ｜ GitHub

import google.cloud.dlp
import io
import csv

# DLPでマスキング
project = 'cm-da-mikami-yuki-258308'
inspect_template = 'projects/cm-da-mikami-yuki-258308/locations/global/inspectTemplates/sample_callcenter'
deidentify_template = 'projects/cm-da-mikami-yuki-258308/locations/global/deidentifyTemplates/sample_callcenter'

def map_headers(header):
    return {"name": header}

def map_data(value):
    return {"string_value": value}

def map_rows(row):
    return {"values": map(map_data, row)}

csv_headers = map(map_headers, headers)
csv_rows = map(map_rows, rows)

table_item = {"table": {"headers": csv_headers, "rows": csv_rows}}

dlp = google.cloud.dlp_v2.DlpServiceClient()
parent = f"projects/{project}"
response = dlp.deidentify_content(
    request={
        "parent": parent,
        "inspect_template_name": inspect_template,
        "deidentify_template_name": deidentify_template,
        "item": table_item,
    }
)

def write_header(header):
    return header.name

def write_data(data):
    return data.string_value

with io.StringIO() as csvfile:
    write_file = csv.writer(csvfile, delimiter=",")
    write_file.writerow(map(write_header, response.item.table.headers))
    for row in response.item.table.rows:
        write_file.writerow(map(write_data, row.values))
    obj_csv = io.BytesIO(csvfile.getvalue().encode('utf-8'))

最後に、マスキング済みデータを BigQuery にロードします。マスキング前後のデータを比較するために、別名のテーブルを新規作成してロードしています。

from google.cloud import bigquery

# マスキング済みデータをBigQueryにロード
table_id_dst = 'dataset_1.sample_callcenter_dlp'

client = bigquery.Client()
job_config = bigquery.LoadJobConfig(
    autodetect=True,
    write_disposition='WRITE_TRUNCATE',
    source_format='CSV',
    skip_leading_rows=1
)
load_job = client.load_table_from_file(
    obj_csv, table_id_dst, job_config=job_config
)
load_job.result()

実行してみると、期待通りマスキングされたデータが新しいテーブルにロードできました。

マスキング前のテーブルデータがこちら。

今回指定した LOCATION の infoType はもともと一部のロケーションにしか対応していないため、また日本語では地名と同じ人名が多いためか、一部人名と地名が混同されている部分がありますが、ほとんどの個人情報は DLP API でマスキングされたことが確認できました。

カスタム infoType を使用

先ほどのサンプルデータでは、080 始まりのハイフンなしフリーテキスト内の携帯電話番号（と思われる）データがマスキングされませんでした。

この携帯電話番号を、カスタム infoType を登録してマスキングしてみます。

作成済みの検査用テンプレートの編集画面「INFOTYPE を管理」リンクから、正規表現のカスタム infoType を追加します。

「カスタム」タブの「カスタム INFOTYPE を追加」ボタンから、「種類」に「正規表現」を選択し、任意の「infoType」名を選択して、「正規表現のパターン」を入力して「完了」→「保存」します。

Cloud DLP 管理画面の匿名化テンプレート「テスト」タブでは、ブラウザ上でマスキング結果を確認できます。念のためカスタム infoType で先ほど検出できなかった携帯電話番号が検出できるようになったか確認してみます。

「入力のサンプル」欄に、CSV 形式の BigQuery テーブルデータを入力してみます。

ブラウザ上では、追加したカスタム infoType の [MOBILE_PHONE] で置換されるようになったことが確認できました。

先ほどと同じ Python コードを再実行してみます。

ブラウザ上でのテスト同様、カスタム infoType に指定した [MOBILE_PHONE] でマスキングされたことが確認できました。

DLP の制限事項

今回は動作確認目的だったので、テーブルデータはたった 10 レコードしかなく、マスキング対象外のカラム含めた全テーブルデータを送信してみましたが、DLP API リクエストのデフォルトのサイズ上限は 0.5MB なので、テーブルデータが多い場合はマスキング対象カラムデータのみに絞ったり、複数回に分けてリクエストするなど、処理内容を検討する必要があるかと思います。

データサイズ意外にも、リクエストのレートリミット（1分あたり600回）や infoType の上限などもあるので、要件に合わせてご確認、ご検討ください。

割り当てと上限｜ Cloud Data Loss Prevention ドキュメント

まとめ（所感）

ユーザーの行動分析などにデータを活用することでビジネスの加速が見込まれる一方、分析に利用するデータには個人情報が含まれるケースも多く、データ流出のリスクも気になるポイントではないかと思います。

例えば会員情報テーブルなど、あらかじめどのテーブルのどのカラムに個人情報が格納されるか分かっているのであれば、SQL 処理でハッシュ化や置換することも可能ですが、フリーテキスト項目を含むテーブルでは、個人情報が含まれるとは想定していなかったテーブルやカラムに個人情報が含まれている可能性も否めません。

Cloud DLP を利用すれば、BigQuery に格納済みのデータ全体に対する機密情報の検出や、特定のテーブル、カラムのフリーテキスト内の機密情報の秘匿化も可能です。

2021/12現在、日本語データではまだ組み込みの検出タイプ（infoType）のみの利用だと完全な秘匿化は期待できないのではないかと思いますが、検出タイプ（infoType）のカスタムや辞書登録もできるので、チューニングを視野に入れれば実運用にも十分使える便利なサービスだと思います！

Cloud DLP で BigQuery テーブルの特定のカラムの機密情報をマスキングしてみた〜カスタム infoType の指定方法を添えて〜

やりたいこと

前提

サンプルデータを準備

検査・匿名化テンプレートを作成

DLP API で特定の項目を匿名化

カスタム infoType を使用

DLP の制限事項

まとめ（所感）

参考

イベント

EVENT【4/23リモート】クラスメソッドのフリーランスエンジニア会社説明会〜AWS / システムアーキテクト案件特集〜を開催します

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

EVENT【4/30（火）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/9（木）】組織的なクラウド統制はじめの一歩

Cloud DLP で BigQuery テーブルの特定のカラムの機密情報をマスキングしてみた 〜カスタム infoType の指定方法を添えて〜

やりたいこと

前提

サンプルデータを準備

検査・匿名化テンプレートを作成

DLP API で特定の項目を匿名化

カスタム infoType を使用

DLP の制限事項

まとめ（所感）

参考

イベント

EVENT【4/23リモート】クラスメソッドのフリーランスエンジニア会社説明会 〜AWS / システムアーキテクト案件特集〜 を開催します

EVENT【5/28（火）】AWSを最大活用するための1dayカンファレンス

EVENT【5/17（金）】認証機能の開発工数削減をデモで体験！次世代認証基盤サービス『Auth0 by Okta』導入実践ウェビナー

EVENT【5/14（火）】アノテーションの中途向けオンライン会社説明会を開催します

EVENT【5/23（木）】ユースケースに学ぶAWS運用のノウハウ～可視化から統制まで～

EVENT【5/16（木）】Snowflakeを触ってみよう！初めての方向けハンズオンセミナー

EVENT【5/10（金） 名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【4/25（木）リモート】Google Cloudに携わるエンジニアのキャリア～クラスメソッド会社説明会～

EVENT【4/30（火）リモート】クラスメソッドの会社説明会を開催します

EVENT【5/9（木）】組織的なクラウド統制 はじめの一歩

関連記事

Cloud DLP APIで正規表現のカスタムinfoType検出器を使ってみる（Cloud Shell + curl編）

Cloud DLP で BigQuery テーブルの特定のカラムの機密情報をマスキングしてみた〜カスタム infoType の指定方法を添えて〜

EVENT【4/23リモート】クラスメソッドのフリーランスエンジニア会社説明会〜AWS / システムアーキテクト案件特集〜を開催します

EVENT【5/10（金）名古屋】クラスメソッドグループの会社説明会を開催します！

EVENT【5/9（木）】組織的なクラウド統制はじめの一歩