Azure AI Content Safety による有害なコンテンツの検知を Microsoft Sentinel でインシデント化してセキュリティ管理

本稿は、Microsoft Top Partner Engineer's Advent Calendar 2023 に参加しています。

本記事について

本記事では、アプリケーション内のコンテンツに有害な内容が含まれていないかを調べたいときに、Azure のサービスを使って検知し、それをセキュリティ管理していく具体的な方法をご紹介します。

特に登場人物として、Azure AI Content Safety と Microsoft Sentinel を扱います。

Azure AI Content Safety とは？

Azure AI Content Safety は、テキストや画像が有害なコンテンツかどうかを判定してくれる Azure の PaaS サービスです。2023年5月にプレビューリリースされ、同年10月に一般提供が開始された新しいサービスになります。Violence, Hate, Sexuality, Self-harm の4項目において、文章やイメージのリスクレベルを示してくれます。

特に、Azure AI サービスの一部として、生成 AI を活用したアプリケーションにおいて、ユーザーや AI モデルが不快、危険、または望ましくない可能性があるコンテンツを作成していないか、をチェックすることができます。

Microsoft Top Partner Engineer's Advent Calendar 2023 内でも、下記記事が Azure AI Content Safety について詳細にご解説されています。

一方で、基本的には API として、テキストや画像を渡し都度判定させるという使い方になります。そのため、アプリや生成 AI モデルが攻撃を受けていることを常時セキュリティ監視し、インシデント化して運用チームが管理したり、過去の統計情報をまとめてダッシュボード化したり、といった機能は提供されていません。

Microsoft Sentinel とは？

もし、そのあたりを実装したいときは、Azure のサービスでは、Microsoft Sentinel を使うことができます。Sentinel は、2019年にリリースされた、マイクロソフトの SIEM as a Service であり、セキュリティログ収集、インシデント管理と脅威分析、対処のワークフローによる自動化、などを行うことができます。

本記事では、ユーザーが生成するデータを Blob Storage に保存するアプリケーションにおいて、どう Azure AI Content Safety を使って有害なコンテンツを検知し、それを Microsoft Sentinel でセキュリティ管理できるかを、簡単なサンプルを使って見ていきます。

本記事のサンプルのアーキテクチャ

本記事では以下の構成で Blob Storage と Azure AI Content Safety, Microsoft Sentinel を組み合わせる方法を見ていきます。

アプリケーションやユーザーが、有害な内容を含む画像やファイルを Blob Storage にアップロードしてしまう可能性がある環境を想定します。そういった際に、各ファイルがアップロードされるたびに Azure Functions 上の関数をトリガーし、Azure AI Content Safety に内容の有害性を確認します。そのログを同関数から Log Analytics にアップロードし、Sentinel を使って脅威検知、インシデント化、ダッシュボード監視を行えるようにします。

なお、Azure AI Content Safety は他の Azure AI サービスと同じく、診断設定で Azure 側で定義されているログを Log Analytics ワークスペースに送信することができます。ただし、Content Safety 側での応答結果や対象の Blob Storage の URI を合わせてログ記録することはできないため、Functions からログを Log Analytics ワークスペースに送る必要があります。

サンプルの実装手順

ここからは、実際に上記の絵の内容を実現する方法を見ていきます。

1. Blob Storage の作成

まず、ファイルの格納用のストレージアカウントを用意します。今回はストレージアカウントで、user-photos というコンテナーを作り、user-photos/<ユーザーID>/<ファイル(.jpg)> という形で格納することにします。

また、Functions の関数で Blob Trigger を Event Grid を介して動かせるようにします。ストレージアカウントのイベントの設定で、Event Grid のイベントサブスクリプションとシステムトピックを作成します。

この手順については詳細を下記記事にまとめておりますのでご参照ください。

今回はフィルターを下記のように設定しました。

2. Azure AI Content Safety の準備

次に、Azure AI Content Safety のインスタンスをひとつ作成します。後ほどエンドポイントとキーの情報を使って、このインスタンスに判定の問い合わせを行います。

3. Log Analytics と Sentinel の準備

加えて、ログ格納用の Log Analytics ワークスペースを作成し、その上で Microsoft Sentinel を有効化します。後ほど Sentinel の設定を行い、インシデント管理ができるようにします。

ログの取り込みについては、Log Ingest API 経由でカスタムログを取りこむ方式を採用し、その Python SDK を使って実装します。下記の記事をベースに、コードを作成します。

Azure Monitor でデータ収集エンドポイントを作成し、Log Analytics ワークスペースでカスタムログ (DCR) のテーブルとそのためのデータ収集ルールを作成します。

テーブルは、下記のサンプルを読み込んで作成します。

サンプルログ

{
    "TimeGenerated": "2023-12-07T07:24:41.342470",
    "UserId": "100",
    "FileName": "image1.jpg",
    "Hate": 0,
    "Violence": 0,
    "Sexuality": 0,
    "SelfHarm": 0
}

4. Azure Functions 関数アプリと関数

最後に、今回のコードの実行環境となる Azure Functions の関数アプリをひとつ作成します。コードは Blob Trigger で VS Code にてひな型を用意することができます。

今回用意したサンプルのコードは下記になります。Blob Trigger でファイルがアップロードされたら関数を実行します。アップロードされたファイル (JPEG) を Azure AI Content Safety のインスタンスに送信して、判別結果を受け取ります。そして、ファイルの情報とその判別結果をログとして、Log Analytics ワークスペースに送ります。

実際にちゃんと実装しようとする場合は、Azure AI Content Safety に送るファイルのサイズを制限を超えないように調整したり、など様々な考慮点があるかと思いますので、下記はあくまで簡易的なデモ用コードとしてご参照ください。

function_app.py

import azure.functions as func
import logging
import datetime
from azure.ai.contentsafety import ContentSafetyClient
from azure.ai.contentsafety.models import ImageCategory
from azure.core.credentials import AzureKeyCredential
from azure.core.exceptions import HttpResponseError
from azure.ai.contentsafety.models import AnalyzeImageOptions, ImageData
from azure.identity import DefaultAzureCredential
from azure.monitor.ingestion import LogsIngestionClient

# ログの設定
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
stream_handler = logging.StreamHandler()
logger.addHandler(stream_handler)

# FunctionAppのインスタンスを作成
app = func.FunctionApp() 

# Blob Storage トリガーの関数の作成
@app.blob_trigger(arg_name="myblob", path="user-photos/{userId}/{fileName}",
                               connection="ConnectionString",
                               source="EventGrid") 
def yooContentSafety(myblob: func.InputStream):
    logger.info(f"Python blob trigger function processed blob"
                f", Name: {myblob.name}"
                f", Blob Size: {myblob.length} bytes")    

    # Blob Storage とファイルの情報を取得
    container_name = myblob.name.split('/')[0]
    userId= myblob.name.split('/')[1]
    fileName = myblob.name.split('/')[2]

    logger.info(f"container_name: {container_name}"
                f", userId: {userId}"
                f", fileName: {fileName}")
    
    # Azure AI Content Safety の設定

    # Content Safety のキーとエンドポイントを取得
    key = "<Content Safety のキー>"
    endpoint = "<Content Safety のエンドポイント>"

    # Content Safety のクライアントを作成
    client = ContentSafetyClient(endpoint, AzureKeyCredential(key))

    # Content Safety の分析を実行
    request = AnalyzeImageOptions(image=ImageData(content=myblob.read()))

    try:
        response = client.analyze_image(request)
    except HttpResponseError as e:
        logger.error("Analyze image failed.")
        if e.error:
            logger.error(f"Error code: {e.error.code}")
            logger.error(f"Error message: {e.error.message}")
            raise
        logger.error(e)
        raise

    logger.info(f"content safety results: {str(response.hate_result)}, {str(response.self_harm_result)}, {str(response.sexual_result)}, {str(response.violence_result)}")

    # Azure Log Analytics にログを送信

    # Azure Monitor の Data Collection Rule の設定
    data_collection_endpoint = "<Data Collection Rule のエンドポイント>"
    dcr_rule_id = "<Data Collection Rule の Immutable ID>"
    dcr_stream_name = "<DCR のストリーム名>"

    # Azure AD の認証情報 - DefaultAzureCredential クラスを利用する際に必要
    credential = DefaultAzureCredential() 

    # ログクライアントの作成
    logs_client = LogsIngestionClient(data_collection_endpoint, credential)

    # ログデータの作成
    # 現在時刻の取得
    dt_now = datetime.datetime.utcnow()
    logger.info(f"dt_now: {dt_now.isoformat(timespec='seconds')}")
   
    # ログ用のリストの生成
    logs = [{
        "TimeGenerated": dt_now.isoformat(timespec='milliseconds'),
        "UserId": userId,
        "FileName": fileName,
        "Hate": response.hate_result.severity,
        "Violence": response.violence_result.severity,
        "Sexuality": response.sexual_result.severity,
        "SelfHarm": response.self_harm_result.severity
    }]

    logger.info(f"logs: {logs}")

    # Azure Monitor にログデータを送信
    logs_client.upload(rule_id=dcr_rule_id, stream_name=dcr_stream_name, logs=logs)  
    logger.info("logs uploaded to Azure Monitor")

このコードが正しく実行されれば、今回やりたいことが実現できるはずです。

5. Sentinel でのインシデント化

Sentinel では Analytics でクエリルールをセットして、インシデント化することができます。今回のサンプルログに対しては、下記クエリを実行させるようにしてみます。どの Severity や対象をインシデントにするかや、アラートの拡張の設定は、個々の要件やログの出力の内容などによって変わってきます。

分析ルール

yooContentSafetyTest1_CL
| where Hate > 0 or Violence > 0 or Sexuality > 0 or SelfHarm > 0

これにて準備は完了です！

実際に動かしてみる！

ここからは、実際に不適切と判定されうる画像ファイルを用意して、それを Blob Storage のコンテナーにアップロードしてみます。

すると、Azure Functions がトリガーされ、Content Safety のインスタンスに問い合わせを行い、ログを記録してくれます。ちなみに、この画像は Violence が6という高い値が出ています。

次に Sentinel のインシデント管理画面に行ってみます。するとこのイベントがインシデントとして登録されていることがわかります。

インシデントの詳細に行くと、たとえば同じユーザーで別のインシデントが起こっていないか、なども確認できます。（分析ルールでエンティティの設定が必要です。）

Azure AI Content Safety を使うことで、簡単に画像が有害かを検知でき、かつ Sentinel とログ連携することで管理や検索、分析が容易に行えることを確認することができました。

最後に

本記事では、Azure AI Content Safety と Microsoft Sentinel を使って、有害なコンテンツのの検知とそのインシデント管理の簡単な構成例を見ていきました。実際に Azure AI Content Safety を利用したアプリケーションを作成するには、他にも多くの考慮点があります。ですが、この記事が Content Safety や Sentinel の利用を検討される際の参考になりましたら幸いです。

＊本稿は、個人の見解に基づいた内容であり、所属する会社の公式見解ではありません。また、いかなる保証を与えるものでもありません。正式な情報は、各製品の販売元にご確認ください。