Google Cloudの監査ログの長期保管とアクセス監査を考える

当記事はみずほリサーチ&テクノロジーズ × G-gen エンジニアコラボレーション企画で執筆されたものです。

みずほリサーチ&テクノロジーズ株式会社の舘山と申します。
本日は、Google Cloudの監査ログの長期保管とアクセス監査について、調査した内容を共有します。

参考資料として、Google Cloudのログ取得サービス、監査ログの全体像は下記のブログ、Google Cloudドキュメントにまとめられています。

G-genブログ
- Cloud Loggingの概念と仕組みをしっかり解説
- Cloud Audit Logsを解説。Google Cloud(GCP)の証跡管理
Google Cloudドキュメント

１．当記事が前提とする要件
２．デフォルトの監査ログ設定と要件のギャップ
３．要件を満たす対応の実施
４．アクセス監査資料の作成
５．まとめ

１．当記事が前提とする要件

当記事が前提とする要件を記載します。

組織全体の監査ログを本番/開発別の監査用プロジェクトに集約し2年間保管する。
組織全体で日次で開発者がアクセスした本番プロジェクト（組織、フォルダ）の監査を行う。
リソースの更新（ADMIN_WRITE）だけでなく、リソースの参照（ADMIN_READ）、データの参照、更新（DATA_READ、DATA_WRITE）の監査ログも保管する。
サービス制約がない限り、データは国内リージョンに保有する。
(組織的な統制には、組織ポリシーconstraints/gcp.resourceLocationsが利用可能です。)
サービス制約がない限り、暗号化には顧客管理の暗号鍵(CMEK)を利用する。
(組織的な統制には、組織ポリシーconstraints/gcp.restrictNonCmekServicesが利用可能です。)

実際のシステム構築では、以下のトピックについても考慮が必要ですが、当記事では基本的に扱いません。

VPCフローログ、ファイアウォールルールログ等のネットワークのログ
業務アプリケーションやミドルウェアのログ
GKEのKubernetes API（k8s.io）の監査ログ
Cloud Identity（ユーザーアカウント管理）、請求先アカウントの監査ログ
アクセスの透明性のログ（Googleのエンジニアによるアクセスの監査ログ）
リアルタイムの監査ログ監視
ログに対する細かいアクセス制御
Cloud Asset Inventoryによるリソースのスナップショット、変更履歴の集約
Security Command Centerの検出結果の集約

２．デフォルトの監査ログ設定と要件のギャップ

デフォルトの監査ログ設定での、今回の要件の順守状況を整理すると下の表のようになります。

要件	順守状況
全ての監査ログ取得	×：BigQuery以外のデータアクセス監査ログは取得対象外
国内リージョンの利用	×：Cloud Loggingバケットの保存先はグローバル(海外含む)
暗号化に顧客管理の暗号鍵(CMEK)の利用	×：Cloud LoggingバケットはGoogle管理の鍵で暗号化
組織全体の監査ログ集約、長期保管	×：各プロジェクトの監査ログは、各プロジェクトのCloud Loggingバケットにのみ保存　　Requiredバケットの保存期間は400日固定　　Defaultバケットの保存期間の初期設定は30日

３．要件を満たす対応の実施

（１）データアクセス監査ログの有効化

各プロジェクト単位ではなく、組織レベルでデータアクセス監査ログの有効化設定が可能です。

設定手順についてはCloud Loggingのガイドを参照してください。
データアクセス監査ログを構成する

各サービスのデータアクセス監査ログを有効にした場合、Cloud Loggingのデフォルト設定では、API実行対象の組織、フォルダ、プロジェクトの_Defaultバケットに全てのログが取り込まれ、追加の料金が発生します。
GoogleCloud のオペレーションスイートの料金

大量のログを取り込む場合のCloud Loggingの費用感としては、Cloud Bigtableの例が参考になります。（加えて、ログ集約先のストレージコスト、監査のためのクエリのコストにも影響があります。）

要件に対して不要なログを削減するため、特定のプリンシパルのロギングを制限することが可能です。

（２）Cloud Loggingバケットの保存先を東京リージョンに変更

こちらも組織レベルで保存先リージョンを設定することが可能です。

設定手順についてはCloud Loggingのガイドを参照してください。
Cloud Logging のデータリージョン

なお、Loggingバケットのリージョン指定は、既存のバケットには遡及しないため、フォルダ、プロジェクトを作成するより前に設定を行う必要があります。

組織作成時に作成された、組織自体に対する操作の監査ログが格納される _Requiredバケットについては、保存先をリージョンに変更することができません。

また、現状ではLoggingバケットの保存先をグローバルからリージョンに変更した場合、Error Reporting 機能が利用できない制約があります。

（３）Cloud Loggingバケットの暗号化に顧客管理の暗号鍵(CMEK)を利用

こちらも組織レベルでCMEKの利用を設定することが可能です。

設定手順についてはCloud Loggingのガイドを参照してください。
Logging のストレージデータを保護する鍵を管理する

ガイドの手順に従い、組織に関連付けられたサービスアカウント（cmek-o<組織ID>@gcp-sa-logging.iam.gserviceaccount.com）に暗号鍵の暗号化 / 復号の役割を割り当てる際に、追加で以下の対応が必要でした。

暗号鍵がVPC Service Controlsの境界内にある場合、組織のサービスアカウントをアクセスレベルに追加して、境界へのアクセス許可を設定する。
組織ポリシーconstraints/iam.allowedPolicyMemberDomainsで組織外部への権限付与を制限している場合、組織のサービスアカウントは外部IDと判定されポリシー違反となった。暗号鍵の利用権付与する際、鍵を保有するプロジェクトで一時的に組織ポリシーを上書きしてエラーを回避する。

保存先がグローバルのバケットの暗号化にはCMEKを利用できない制約があります。そのため、（保存先をグローバルから変更できない）組織自体の _RequiredバケットをCMEKで暗号化することはできません。

（４）組織の監査ログを集約して、長期保管

A.集約シンクの設定

Cloud Loggingの集約シンクを利用することで、組織の全プロジェクトの監査ログを監査プロジェクトへ配信することができます。

設定手順についてはCloud Loggingのガイドを参照してください。
集約シンクを構成する

全ての監査ログを転送する場合、包含フィルタ条件には logName:cloudaudit.googleapis.com を指定します。

本番/開発別の監査プロジェクトへ集約は、（本番と開発でプロジェクトを配置するフォルダを分けていることを前提に）フォルダに対して集約シンクを適用することで実現できます。

また組織自体に対する操作は、組織のCloud Loggingに記録されるため、下位のフォルダの集約シンクでは拾われません。組織自体にも（非集約）シンクを作成して、本番用の監査プロジェクトへログを転送します。

Cloud Identityの監査ログも集約対象とする場合、事前に監査ログの共有設定が必要です。

シンク設定前のログは転送されませんが、ログのコピー機能がプレビュー版で提供されています。

今回検証していませんが、別案として組織直下に集約シンクを２つ作成し、フィルタ条件で本番/開発プロジェクトのログを選別するという構成も可能だと思います。

また、後述のBigQueryによる監査クエリのコストを最適化する観点で、監査が人（Googleアカウント）による操作だけを対象とする場合には、人以外（サービスアカウント）のAPI実行の監査ログ保管先を分けるという構成も考えられます。

Ｂ.配信先ストレージの選択

シンクからログを配信する先のストレージと対応するユースケースは以下の表のようになります。

配信先	ユースケース
BigQuery	高度な分析
Cloud Storage	コンプライアンス、長期保管
Pub/Sub	Splunk等外部サービスとの統合
ログバケット	ログベースの指標、アラート

参考サイト： BigQuery の監査ログパイプラインを活用した使用状況分析

今回はコンプライアンスと長期保管を主目的として、Cloud Storageを選択します。

Cloud Storageバケットのロケーション、CMEKの利用についてはCloud Storageのガイドを参照してください。
バケットの保存場所
 顧客管理の暗号鍵の使用

Ｃ.Cloud Storageの保持ポリシーによるログの保護

Cloud Storageの保持ポリシー機能を利用することで、保存期間内の監査ログの更新、削除が不可になります。

設定手順についてはCloud Storageのガイドを参照してください。
保持ポリシーと保持ポリシーのロック

なお、保持ポリシーでバケット内のログは保護されますが、集約シンクが削除されると新規ログの配信がストップするため、集約シンク自体は別途保護（IAMポリシーを適切に設定する、変更を監視する等）する必要があります。

また、Cloud Storageのライフライクル管理機能により保存期間を超過したログを自動削除することも可能です。監査等によるアクセス頻度によっては、一定期間経過後のストレージクラスの変更もコスト削減に寄与します。

設定手順についてはCloud Storageのガイドを参照してください。
オブジェクトのライフサイクル管理

D.組織、フォルダーに紐づくリソースの設定変更監視について

リソースの設定変更監視は、Cloud Security Centerのモニタリング脆弱性チェックのように、監査ログに対してログベースの指標に基づくアラートを作成する方法があります。（現在はログベースのアラートもGAになっています。）

一方で、組織、フォルダ階層の監査ログをソースとする場合

ログベースの指標、アラートは組織、フォルダの階層では定義不可能。
組織、フォルダからプロジェクトのログバケットへ転送されたログに対する、ログベースの指標は、2023年1月時点でGA前。
組織、フォルダー階層の監視用ログシンクからPub/Subトピックへの配信を利用する場合、組み込みのメール送信が未サポート。

という制限事項があります。

また、監視にCloud Asset Inventoryの変更フィードを利用する場合、組織レベル、フォルダーレベルの変更フィードからVPC Service Controls境界内のPub/Subトピックへの配信不可など、VPC Service Controls関連の制限事項があります。

４．アクセス監査資料の作成

前掲の監査ログ保管ストレージの選択肢の表から、高度な分析にはBigQueryが適しています。

監査資料作成のためのクエリをBigQueryで実行するため、監査ログをBigQueryにも重複配信（BigQuery側での保存期間は監査要件に応じて短く設定）するという構成も可能ですが、今回は監査ログを保管しているCloud StorageバケットをBigQueryから外部テーブルとして参照する方法を紹介します。

（１）ログフォーマットの把握

Cloud LoggingのログレコードはLogEntryという構造になっています。

監査ログの場合、LogEntryのprotoPayload要素が、監査ログを表すAuditLogという構造になっています。

このレコード定義と実ログのサンプルから監査資料で必要そうな項目がどこにあるかを把握していきます。

下の画像はBigQueryテーブルのメタデータ更新が権限不足エラーとなった場合の監査ログを一部加工したものです。

日次のアクセス監査ではプロジェクト（組織、フォルダー）の粒度でアクセス先を識別できればよいものとして、必要な項目の格納場所を整理します。

項目	LogEntry内の格納場所
操作日時	timestampが使えそうです。注意点としてBigQueryのデータ読み取り量を最適化するため、WHERE句で日付を限定する条件はこの項目ではなく、_FILE_NAME疑似列に適用します。（詳細は後述）
操作ユーザー	protoPayload.authenticationInfo.principalEmail が該当します。システム自動処理や、Google Cloudの内部処理によるAPI実行も監査ログに記録されます。また、権限不足エラーとなった操作など操作ユーザーが匿名化されるケースがあります。
アクセスしたプロジェクト（組織、フォルダー）	logName が監査ログの出元です。 logNameの例： "projects/[PROJECT_ID]/logs/[LOG_ID]" "organizations/[ORGANIZATION_ID]/logs/[LOG_ID]" "billingAccounts/[BILLING_ACCOUNT_ID]/logs/[LOG_ID]" "folders/[FOLDER_ID]/logs/[LOG_ID]" logNameの二階層目の文字列までを抽出することで、アクセスしたプロジェクト（組織、フォルダー）が特定できそうです。
監査ログの種類（リソースの更新操作か）	logName末尾の[LOG_ID]部分を見ればわかりそうです。 cloudaudit.googleapis.com%2Factivity がリソース更新の監査ログです。注意点として操作日時同様、BiqQueryクエリのWHERE句では、_FILE_NAME疑似列に条件を適用します。データアクセス監査ログのサブタイプ(ADMIN_READ、DATA_READ、DATA_WRITE）については直接識別する属性がなく、protoPayload.serviceNameとprotoPayload.methodNameの組から個別に判断するしかなさそうです。同様にポリシー拒否監査ログの対象操作がADMIN_WRITE、ADMIN_READ、DATA_READ、DATA_WRITE、のどれに該当するかも直接識別できません。
権限不足エラーだったか	権限エラーとなった操作の監査ログを確認するとprotoPayload.status.code の値が 7 になっています。 Java Cloud Client LibrariesのEnumの定義を見ると、PERMISSION_DENIEDのエラーコードが7のようです。なお、VPC Service Controlsのポリシー違反も同じエラーコードになるようです。区別が必要な場合は、外部テーブルの_FILE_NAME疑似列の値にcloudaudit.googleapis.com/policyが含まれるものが、ポリシー違反のログになります。
接続元IPアドレス	protoPayload.requestMetadata.callerIpに格納されています。 Google Cloudサービス経由の呼び出しなど、IPアドレスが秘匿化されるケースも存在します。

（２）BigQuery外部テーブルの作成

監査ログを保管したCloud Storageバケットを参照する外部テーブルを作成します。

作業手順についてはBigQueryのガイドを参照してください。
Cloud Storage データのクエリ

シンクの転送先に直接BigQueryを指定した場合には特別な命名規則を使用するスキーマが自動生成されます。

一方で外部テーブルとして監査ログを保管したCloud Storageバケットを参照する場合、スキーマの自動認識に特別な配慮はないようです。@typeなどのフィールド名がBigQueryの命名規則に合致しないため、スキーマの自動認識はエラーとなりました。

インターネットで検索しましたが流用できるスキーマ定義が見つからなかったため、以下のスキーマ定義を手動で作成しました。

LogEntryの第一階層までを定義し、ネストされた構造化データのフィールドはRECORD型ではなく、JSON型としています。

（３）BigQueryから読み込むデータ量(=分析料金)の最適化

BigQueryではクエリで読み込んだデータ量に応じても従量課金されます。（オンデマンド料金モデルの場合）

そのためBigQueryのテーブルはクエリの条件となる項目（日付等）でパーテーション分割して、アクセスするデータ範囲を削減することが定石となっています。

Cloud Storageの場合、Hiveパーティション分割レイアウトに沿ったフォルダ構成

例：gs://[バケット名]/myTable/dt=2019-10-31/lang=en/foo

になっていれば、BigQueryからパーテーション分割された外部テーブルとして扱われます。
Cloud Storage の外部パーティション分割データに対してクエリを実行する

監査ログを集約シンクでCloud Storageへ配信した場合のファイルパスは下記の様になります。

gs://[バケット名]/cloudaudit.googleapis.com/[activity|data_access|policy]/2022/12/28/03:00:00_03:59:59_S0.json

これは2022/12/28 3時台のログファイルの例です。なお、タイムゾーンは日本時間ではなく、UTCです。

パスに日付と時間が含まれているものの、Hiveパーティション分割レイアウトに沿っていないため、BigQueryからはパーテーション分割されたデータとは認識されません。

一方で、ファイルのパスはクエリから _FILE_NAMEという疑似列で参照することができ、

BigQueryガイド Cloud Storage でパーティション分割されていないデータをクエリするには

クエリに _FILE_NAME 疑似列に対するフィルタ述語がある場合、BigQuery は、フィルタに一致しないファイルの読み取りをスキップしようとします。

と記載されています。

この特性から1日分のアクセス監査資料を作成したい場合、監査ログ内のタイムスタンプではなく、日時を含む_FILE_NAME疑似列のパスに対して条件を適用することで、アクセスするデータ範囲を削減することが可能になります。

_FILE_NAME LIKE '%2022/12/30%'

という条件では日本時間ではなく、UTCでの1日分の監査ログが対象になります。

日本時間での1日分の監査ログの抽出が必要な場合、日付を求める手順としては、
・正規表現でファイルパスから年/月/日/時間を抽出し
・タイムスタンプ型にパースした上で日本時間での日付に変換する
という方法が考えられます。

DATE(PARSE_TIMESTAMP('%Y/%m/%d/%H', REGEXP_EXTRACT(_FILE_NAME, '[0-9]+/[0-9]+/[0-9]+/[0-9]+')) ,'Asia/Tokyo') = '2022-12-30'

_FILE_NAME疑似列への条件の記載方法によっては、実データへのアクセスがスキップされないケースがあります。

実際に実データへのアクセスがスキップされるかは、先日付を指定してクエリ結果の「課金されるバイト数」が0 Bとなることで裏取りします。

なお、組織ポリシーconstraints/gcp.restrictNonCmekServicesでBigQueryのCMEK暗号化利用を必須としている場合、コンソールからクエリを実行する前に、クエリ結果を格納する一時テーブルの暗号化に利用する鍵を設定する必要があります。
（クエリ画面の「その他」→「クエリの設定」→「詳細オプション」→「顧客管理の暗号化」に鍵のリソースIDを入力します。）

集約シンクから直接エクスポートしたBigQueryテーブルの場合

ログシンクからBigQueryへ直接エクスポートする場合、自動生成されるテーブルはtimestampのUTC日付でパーテーション分割（クラスタリング列は未指定）されます。私が調べた限り、シンクからのエクスポート時にパーテーションのタイムゾーンを調整する設定は見つかりませんでした。

外部テーブルとの違いとして、日本時間での1日分のログの参照条件の記載は

DATE( timestamp ,"Asia/Tokyo" ) = "2022-12-30"

という記載では、全パーテーションが走査対象となってしまうため、パーテーション列側を加工するのを避け

 timestamp >= TIMESTAMP('2022-12-29 15:00:00') AND timestamp < TIMESTAMP('2022-12-30 15:00:00')

という記載にする必要がありました。

自然体だと日本時間での１日分のログを監査するには、UTC2日分のパーテーションの走査が必要となります。

クエリ最適化対策として、オンデマンド配車および配達ソリューションのログ解析のガイドでは、自動生成されたテーブルに後からtimestampをクラスタリング列に追加する方法が紹介されていました。

 for TABLE in `bq ls --format=csv -project_id=<プロジェクトID>  <ログエクスポート先のデータセットID> | cut -d, -f2 | tail +2` \
 ; do \
 bq update --clustering_fields=timestamp ${TABLE} \
 ; done

（４）アクセス監査のためのクエリを考える

一例として、本番プロジェクトへのアクセスには事前申請が必要であり、無申請のアクセスがないかを監査するシナリオを想定します。

ユーザーはコンソールからユーザーアカウント（Cloud Identityで管理するユーザー）でアクセスすることを前提として、各ユーザーが1日にアクセスしたプロジェクトの情報を出力します。

Cloud Identityへのログインは、特定のプロジェクトへのアクセスを示すものではないため、ログインの監査ログだけではアクセス先は不明です。

AWSでIAMユーザーでログイン後に各アカウントのIAMロールにスイッチロールする運用の場合には、スイッチロールを各アカウントへのログイン操作とみなす作戦が取れますが、 Google Cloudには当てはまりません。

代替手段として力技になりますが、なんらかの監査ログが記録されていれば当該プロジェクトへログインしたとみなし、各ユーザーアカウント/アクセス先プロジェクト毎に

ユーザーアカウント
アクセスしたプロジェクト（組織、フォルダ）
（権限エラーのぞく）最初の操作時間
（権限エラーのぞく）最後の操作時間
（権限エラーの）最初の操作時間
（権限エラーの）最後の操作時間
接続元IPアドレス（複数ある場合カンマで連結）
アクセス先横断で当該ユーザーの当日の全操作が権限エラーの場合〇を表示

を権限エラーの明細行を分けて出力します。改善の余地ありますが、以下のようなSQLが考えられます。

WITH audit_logs AS
(
 SELECT JSON_VALUE(protoPayload.authenticationInfo.principalEmail) AS principalEmail, 
        REGEXP_EXTRACT(logName,'[^/]+/[^/]+') AS accessedProject, /* 正規表現で[organizations/xxxx][folders/xxxxx][projects/xxxx]部分を抽出 */
        DATETIME(timestamp, 'Asia/Tokyo') AS operationTimestamp,/*パフォーマンス最重視する場合、tz補正は最後のSELECT句で行う*/
        JSON_VALUE(protoPayload.requestMetadata.callerIp) AS callerIp, 
        CASE WHEN JSON_VALUE(protoPayload.status.code) = '7' THEN true ELSE false END AS permissionDenied
 FROM `poc01-hn-audit-■■■■■■■■■■.audit.external_audit_log` 
 WHERE DATE(PARSE_TIMESTAMP('%Y/%m/%d/%H', REGEXP_EXTRACT(_FILE_NAME, '[0-9]+/[0-9]+/[0-9]+/[0-9]+')) ,'Asia/Tokyo') = '2022-12-29'/* 日付はパラメタ化*/
 /* 読み取りデータ量削減のため、実データではなく_FILE_NAME疑似列で日付、ログ種別を絞る。_FILE_NAMEはUTCの日時 */
 AND ( _FILE_NAME LIKE '%cloudaudit.googleapis.com/activity%' OR _FILE_NAME LIKE '%cloudaudit.googleapis.com/data_access%' OR   _FILE_NAME LIKE '%cloudaudit.googleapis.com/policy%')
 /* 監査対象はユーザーアカウントに絞る*/
 AND JSON_VALUE(protoPayload.authenticationInfo.principalEmail) LIKE '%@%'
 AND JSON_VALUE(protoPayload.authenticationInfo.principalEmail) NOT LIKE '%gserviceaccount.com'
)
SELECT principalEmail,
       accessedProject ,
       MIN( CASE WHEN permissionDenied THEN null ELSE operationTimestamp END) AS firstOperationTimestamp,
       MAX( CASE WHEN permissionDenied THEN null ELSE operationTimestamp END) AS lastOperationTimestamp,
       MIN( CASE WHEN permissionDenied THEN operationTimestamp ELSE null END)  AS firstAuthorizationErrorTimestamp,
       MAX( CASE WHEN permissionDenied THEN operationTimestamp ELSE null END)  AS lastAuthorizationErrorTimestamp,
       STRING_AGG(DISTINCT callerIp ORDER BY callerIp) AS callerIp ,
       CASE WHEN 
　     　COUNTIF( NOT permissionDenied ) OVER( PARTITION BY principalEmail ) > 0 
         THEN '' ELSE '〇' END AS permissionDeniedOnlyUser
FROM audit_logs
GROUP BY principalEmail,accessedProject,permissionDenied
order by 1,2,3,5