VSCodeでAzure Policyに準拠していないリソースをSlackに自動通知してみよう

はじめに
アプリの概要
- 構成図
- 通知イメージ
前提条件
作業前の準備
- Azure Policyのポリシー定義IDの取得
- SlackのWebhook URLの取得
VSCode上での開発環境の作成
コーディング
- function_app.py
  - ソースコード
  - 解説
- requirements.txt
  - ソースコード
Azure Functions環境作成
- リソース作成手順
注意点
- 環境変数に設定した設定値について
おわりに

はじめに

こんにちは、ACS事業部 CIチームの青木です。
この記事では、VSCodeでAzure Policy準拠状況をSlackに自動通知するアプリを開発する方法について紹介します。

以前、「Azure Policyで特定のタグが付与されていないリソースを特定する方法」についてのブログを書きました。
本稿はこのブログの続きの内容となっているため、まずはこちらをご覧ください。

techblog.ap-com.co.jp

上記のブログでは組織的にAzureのタグを利用する際にAzure Policyを利用することで、組織のルールにのっとって特定のタグの付与を行っているリソースとそうでないリソースを効果的に特定する方法について紹介させていただきました。

しかし、定期的に「Azure Policyのページを開いて、リソース状況を確認して、タグの付与を行っていないユーザーに対して連絡を入れ…」という行為を行うのは運用コストがまだまだ大きいです。
できればAzure Policyにて制定したルールに従っていないリソースを自動で列挙して、定期的にチャットなどでチームに連絡するところまでを自動化したいですよね？
ご安心ください。それ、Azure Functionsでできます。

アプリの概要

構成図

こちらが、今回作成するアプリの処理のイメージ図です。

実際にはAzure Functionsが動作するためのストレージアカウントなどの用意なども行いますが、ざっくりな処理の流れとしてご覧ください。

通知イメージ

実際にアプリを実行した際、Slackには以下のような通知が届きます。

今回の場合、ポリシーに非準拠のリソースがあった場合は、準拠していないリソース数と準拠していないリソース名を列挙し、Slackに自動通知します。

前提条件

「はじめに」に記載のブログ「Azure Policyで特定のタグが付与されていないリソースを特定する方法」に従ってAzure Policyを作成済みであること
WSLがインストールされていること(本ブログはver: 2.2.4.0を使用)
Microsoft Visual Studio Codeがインストールされていること(本ブログはver: 1.92.0を使用)
VSCodeの拡張機能Azure Functions - Visual Studio Marketplaceがインストールされていること
Pythonがインストールされていること(本ブログではver: 3.10.12を使用)
WSLにFunctions Project用の空ディレクトリを作成してあること(名前、場所は任意)

作業前の準備

開発環境の作成を始める前に、作業に必要な情報を用意しましょう。以下二点の準備作業を行います。

Azure Policyのポリシー定義IDの取得
SlackのWebhook URLの取得

Azure Policyのポリシー定義IDの取得

このアプリは特定のAzure Policyのポリシー定義を用いて、その定義に準拠していないリソース名のリストを通知する仕様となっていいます。
今回はブログ「Azure Policyで特定のタグが付与されていないリソースを特定する方法」で作成したポリシー定義を使用します。

Azure Policyのページにて、[作成]→[定義]→[リソースグループにownerタグが割り当てられていません]をクリックします。

表示されている定義 IDをクリップボードにコピーします。

この定義IDは「Azure Functions環境作成」の章で使用します。
"/subscription～Definitions/"までは不要のため、削除したうえで値を控えておいてください。

例

クリップボードにコピーした定義ID
"/subscriptions/<Subscription ID>/providers/Microsoft.Authorization/policyDefinitions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
↓
削除後
"xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"

SlackのWebhook URLの取得

Slack Webhookのアプリを用意して、Webhook URLを取得しましょう。
取得方法については以下のような詳しく書いてあるサイトがいくつかありますので、こちらでは割愛します。

blog.querier.io

VSCode上での開発環境の作成

本作業はVSCode上で開かれたWSLのターミナル上で作業を行います。

早速始めていきましょう。
VSCodeにて、Azureアイコンをクリックします。

Azure Functionsのマークをクリックし、[Create New Project]をクリックします。

新しいProjectを作成するにあたり、Functionのソースコードなどを配置するディレクトリを指定する必要があります。

あらかじめ作成しておいたディレクトリのフルパスを入力し、[OK]をクリックします。

次に、開発言語を選択します。
今回はPythonを使用して開発を行っていきますので、Pythonを選択しましょう。

続いて、プログラミングモデルの選択をします。
ここでは推奨されているV2を選択しましょう。

続いて、仮想環境を作成するインタプリタを選択します。
pythonがWSLにインストール済みであれば、インストールされているpythonのバージョンが表示されるはずです。そちらをそのまま選択します。

続いて、関数のテンプレートを選択します。
今回は特定の時間で作動する関数を作成したいので、[Time Trigger]を選択します。

続いて、関数名を入力します。
任意の関数名を入力してください。

最後に、関数が実行される日時を入力します。
NCRONTAB 式を参考に、定期実行したい日時を入力してください。
以下の例では、毎週月曜日のUTC時間の01:00(日本時間での10:00)に関数が実行されます。

ここまで完了すると、指定したディレクトリ配下に必要なファイルが作成され、サンプルコードが表示されます。

作業ディレクトリは以下のようなファイル群が配置されているはずです。

ディレクトリ
  ├.funcignore
  ├.gitignore
  ├.venv/
  ├__pycache__/
  ├function_app.py
  ├host.json
  ├local.settings.json
  └requirements.txt

このディレクトリ内にあるファイルに修正を加えていきましょう。

コーディング

では、いよいよコードを書いていきましょう。
主に、2つのファイルを更新します。

function_app.py
requirements.txt

function_app.py

このファイルに関数で実行したい処理を記載します。
書き方によって複数のファイルに分けることも可能ですが、今回はこのファイルにすべての処理を記載します。

ソースコード

import requests
import os
import azure.functions as func
from azure.identity import DefaultAzureCredential
from azure.mgmt.policyinsights import PolicyInsightsClient
from azure.mgmt.policyinsights.models import QueryOptions

# Azure Functionのインスタンス作成
app = func.FunctionApp()
# Time Triggerのスケジュールで毎週月曜日のUTC01:00(日本時間の10:00)にプログラムを実行
# ※Azure Functions上にデプロイした直後に一度スクリプトを実行したい場合は、run_on_startupをTrueにしてfunc azure functionapp publishコマンドを実行してください
@app.timer_trigger(schedule="0 0 1 * * 1", arg_name="myTimer", run_on_startup=False, use_monitor=False)

# 関数がトリガーされた時に実行されるコード
def timer_function(myTimer: func.TimerRequest):
    print("タイマートリガーによって関数が実行されました。")
    execution_main()

'''
Azure Functionで設定した時間に実行される関数により呼び出される関数です。

①環境変数からSlackのWebhook URL、サブスクリプションID、ポリシー定義IDを取得
②非準拠のリソースグループのリストを取得し、リストが空であれば準拠しているメッセージを、空でなければ準拠していないメッセージを作成
③作成したメッセージをSlackに送信
'''
# TimeTriggerで実行される関数
def execution_main():
    print("execution_sending_message")

    # SlackのWebhook URL
    slack_webhook_url = os.getenv('WEBHOOK_URL')
    subscription_id = os.getenv('SUBSCRIPTION_ID')
    policy_definition_id = os.getenv('POLICY_DEFINITION_ID')

    # ポリシー定義「リソースグループにownerタグが割り当てられていません」に準拠していないリソースグループ名のリストを取得
    non_compliant_resources = get_resources(policy_definition_id, subscription_id, compliance_status='false')   
   
    # 非準拠リソースがあるかないかで処理を分岐
    if not non_compliant_resources:
        print("make all compliant message")
        # 非準拠リソースがない場合のメッセージを作成する関数を実行し、メッセージを取得
        status_message = make_compliant_message()
    else:
        print("make uncompliant message")
        # 準拠、非準拠に関わらずポリシー定義「リソースグループにownerタグが割り当てられていません」に列挙されているリソースグループ名のリストを取得
        total_resources = get_resources(policy_definition_id, subscription_id)
        # 非準拠リソースがある場合のメッセージを作成する関数を実行し、メッセージを取得
        status_message = make_noncompliant_message(non_compliant_resources, total_resources)
    
    # 周知メッセージを作成

    ## ①周知メッセージの最初の部分部分を作成
    announce_message= "リソースグループのownerタグ設定状況についてお知らせします。"
    
    ## ②Azure Functionのリンクを作成
    azure_function_url = "https://learn.microsoft.com/ja-jp/azure/azure-functions/"
    azure_function_link_text = "Azure Function"   
    azure_function_message= "この通知は" + f"<{azure_function_url}|{azure_function_link_text}>" + "によって自動通知されています。"

    ## ①、リソース準拠状況のメッセージ、②のメッセージを連結して最終的な周知メッセージの作成
    final_message = announce_message + "\n\n" + status_message + "\n\n" + azure_function_message

    # Slackにメッセージを送信
    send_message_to_slack(final_message, slack_webhook_url)

'''
Azure Policyのポリシーからリソースを取得する関数です。
ポリシー定義ID、サブスクリプションID、ポリシー準拠状態を引数として受け取ります。
ポリシー準拠状態が指定されていない場合は、全てのリソースを取得し、falseが指定されている場合は、非準拠リソースのみを取得します。
'''
def get_resources(policy_definition_id, subscription_id, compliance_status=None):
    print("get policy resources")

    # 認証情報を取得
    credentials = DefaultAzureCredential()
   
    # ポリシーインサイトクライアントを初期化（サブスクリプションIDを追加）
    policy_insights_client = PolicyInsightsClient(credentials, subscription_id)

    if compliance_status == 'false':
        # ポリシー準拠状態を取得
        policy_states = policy_insights_client.policy_states.list_query_results_for_subscription(
            policy_states_resource='latest',
            subscription_id=subscription_id,
            query_options=QueryOptions(filter=f"policyDefinitionId eq '/subscriptions/{subscription_id}/providers/Microsoft.Authorization/policyDefinitions/{policy_definition_id}' and complianceState eq 'NonCompliant'")
        )
    else:
        policy_states = policy_insights_client.policy_states.list_query_results_for_subscription(
            policy_states_resource='latest',
            subscription_id=subscription_id,
            query_options=QueryOptions(filter=f"policyDefinitionId eq '/subscriptions/{subscription_id}/providers/Microsoft.Authorization/policyDefinitions/{policy_definition_id}'")
        )

    # resourcegroup_listを空のリストとして初期化
    resourcegroup_list = []

    # policy_statesを反復処理し、各stateのresource_idをresourcegroup_listに追加
    for state in policy_states:
        modified_resource_id = state.resource_id.replace(f"/subscriptions/{subscription_id}/resourcegroups/", "")
        resourcegroup_list.append(modified_resource_id)

    # resourcegroup_listを出力
    return resourcegroup_list

'''
非準拠リソースがない場合のメッセージを作成する関数です。
引数はなく、非準拠のリソースグループは存在しない旨のメッセージを返します。
'''
def make_compliant_message():
    compliant_message = ":raised_hands: ownerタグが付与されていないリソースグループはありません！ :raised_hands: "
    return compliant_message

'''
非準拠リソースがある場合のメッセージを作成する関数です。
非準拠リソースグループのリスト、全リソースグループのリストを引数として受け取ります。
この場合は、非準拠リソースグループの数と全リソースグループの数および非準拠リソースグループのリストをまとめたメッセージを返します。
'''
def make_noncompliant_message(non_compliant_resources_list, total_resources_list):
    
    # 非準拠の場合の頭のメッセージの作成
    warning_message = ":warning: 以下リソースグループに現在ownerタグが付与されていません :warning:"
    partition = "＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝＝"
    
    # <準拠していないリソースグループの数>/<すべてのリソースグループの数>の数の作成
    compliance_info = f"*準拠していないリソースグループの数：{len(non_compliant_resources_list)}/{len(total_resources_list)}*"

    # 非準拠リソースのメッセージ要素の連結
    noncompliant_message = compliance_info + "\n\n" + warning_message + "\n" + partition + "\n" + "\n".join(non_compliant_resources_list) + "\n" +  partition
    return noncompliant_message

'''
Slackにメッセージを送信する関数です。
送りたいメッセージとSlackのWebhook URLを引数として受け取ります。
Slackへの送信アクション後のステータスコードが200以外であれば、エラーを発生させます。
'''
def send_message_to_slack(message, slack_webhook_url):
    print("send slack message")
    payload = {'text': message}
    response = requests.post(slack_webhook_url, json=payload)
    if response.status_code != 200:
        raise ValueError(f"Slackへのメッセージ送信に失敗しました: {response.status_code}, {response.text}")

解説

では、内容について解説します。
細かいコードの仕様については、ソースコードのコメントをご覧ください。

この関数のコードは、主に以下の要素で構成されています。

指定時刻に関数を起動する(インスタンス作成とデコレーター、timer_function関数)
1によって呼び出されてメイン処理を行う(execution_main関数)
ポリシーの準拠状況を取得する(get_resources関数)
準拠状況に応じてメッセージを作成する(make_compliant_message関数とmake_noncompliant_message関数)
Slackにメッセージを送信する(send_message_to_slack関数)

1. 指定時刻に関数を起動する(インスタンス作成とデコレーター、timer_function関数)

具体的には以下の部分です。

# Azure Functionのインスタンス作成
app = func.FunctionApp()
# Time Triggerのスケジュールで毎週月曜日のUTC01:00(日本時間の10:00)にプログラムを実行
# ※Azure Functions上にデプロイした直後に一度スクリプトを実行したい場合は、run_on_startupをTrueにしてfunc azure functionapp publishコマンドを実行してください
@app.timer_trigger(schedule="0 0 1 * * 1", arg_name="myTimer", run_on_startup=False, use_monitor=False)

# 関数がトリガーされた時に実行されるコード
def timer_function(myTimer: func.TimerRequest):
    print("タイマートリガーによって関数が実行されました。")
    execution_main()

インスタンスを作成し、そのインスタンスのデコレーターの引数として時刻を設定しています。
デコレーターに設定された時刻になるとtimer_function関数が動作します。この関数はexecution_main関数を呼び出すように記載されているため、設定時刻になるとexecution_main関数が実行されるようになっています。

2. 1によって呼び出されてメイン処理を行う(execution_main関数)

具体的な処理はこの関数に記載されています。
3、4、5、の要素の関数はこの関数によって呼び出され実行されています。

3. ポリシーの準拠状況を取得する(get_resources関数)

サブスクリプショIDとポリシーの定義IDを引数にポリシーで管理しているリソース(今回はリソースグループのみ)のリストを取得する関数です。
必要に応じてポリシーに準拠しているかどうかのステータスを引数で渡すことで、準拠ステータスに合致するリソースのリストを渡すこともできます。

4. 準拠状況に応じてメッセージを作成する(make_compliant_message関数とmake_noncompliant_message関数)

execution_main関数内の以下の実行後、以下の変数に格納した非準拠リソースのリストが空であればmake_compliant_message関数を、リソースが１つでもあればmake_noncompliant_message関数を実行します。

    # ポリシー定義「リソースグループにownerタグが割り当てられていません」に準拠していないリソースグループ名のリストを取得
    non_compliant_resources = get_resources(policy_definition_id, subscription_id, compliance_status='false')

いずれも、Slackに送信するステータス状況のメッセージを作成するための関数です。

5. Slackにメッセージを送信する(send_message_to_slack関数)

execution_main関数の最後に呼び出される関数です。
それまでの処理で作成した送信メッセージとSlackのWebhook URLを引数に、メッセージ送信処理を実行します。

requirements.txt

function_app.pyに記載した処理の実行に必要なパッケージとそのバージョンを記載します。
Azure Functions上にデプロイするコマンドを実行する際、このファイルに記載されているパッケージとそのバージョンを見てスクリプト実行を行うインスタンス上にパッケージを自動インストールします。

ソースコード

# DO NOT include azure-functions-worker in this file
# The Python Worker is managed by Azure Functions platform
# Manually managing azure-functions-worker may cause unexpected issues
azure-functions==1.7.0
azure-identity==1.5.0
azure-mgmt-policyinsights==1.0.0
requests==2.25.1

Azure Functions環境作成

では、ここからはAzure Functions環境をAzure上に作成する手順を紹介します。

リソース作成手順

作業ディレクトリで以下のコマンドを実行しましょう。
"<>"のところは自分の環境に置き換えるか、または任意の文字を適宜入力してください。

# 変数の設定
SUBSCRIPTION="<Subcription ID>"
RESOURCE_GROUP="<リソースグループ名>"
LOCATION="<リージョン名>"
FUNCTION_APP="<関数アプリ名>"
STORAGE_ACCOUNT="<ストレージアカウント名>"

# 作業するサブスクリプションの指定
az account set --subscription ${SUBSCRIPTION}

# Resource Groupの作成
az group create --name ${RESOURCE_GROUP} --location ${LOCATION}

# ストレージアカウントの作成
az storage account create \
--name ${STORAGE_ACCOUNT} \
--resource-group ${RESOURCE_GROUP} \
--location ${LOCATION} \
--sku Standard_LRS

# Azure Functions Appの作成
az functionapp create \
--resource-group ${RESOURCE_GROUP} \
--consumption-plan-location ${LOCATION} \
--runtime python \
--runtime-version 3.10 \
--functions-version 4 \
--name ${FUNCTION_APP} \
--storage-account ${STORAGE_ACCOUNT} \
--os-type Linux

# 環境変数の設定
az functionapp config appsettings set --name ${FUNCTION_APP} --resource-group ${RESOURCE_GROUP} --settings WEBHOOK_URL="<Webhook URL>"
az functionapp config appsettings set --name ${FUNCTION_APP} --resource-group ${RESOURCE_GROUP} --settings SUBSCRIPTION_ID="${SUBSCRIPTION}"
az functionapp config appsettings set --name ${FUNCTION_APP} --resource-group ${RESOURCE_GROUP} --settings POLICY_DEFINITION_ID="<Policy定義ID>"

# Managed IDの付与
az functionapp identity assign \
  --resource-group ${RESOURCE_GROUP} \
  --name ${FUNCTION_APP}

# カスタムロールの作成
az role definition create --role-definition '{
  "Name": "Read Policy Status",
  "IsCustom": true,
  "Description": "Read Azure Policy Status",
  "Actions": [
    "Microsoft.Authorization/policyAssignments/read",
    "Microsoft.Authorization/policyDefinitions/read",
    "Microsoft.Authorization/policySetDefinitions/read",
    "Microsoft.PolicyInsights/policyStates/queryResults/read",
    "Microsoft.PolicyInsights/policyStates/summarize/read"
  ],
  "NotActions": [],
  "AssignableScopes": [
    "/subscriptions/<Subscription ID>"
  ]
}'

# カスタムロールをAzure Functions AppのマネージドIDにアサイン
az role assignment create \
--assignee $(az ad sp show --id $(az functionapp identity show --resource-group ${RESOURCE_GROUP} --name ${FUNCTION_APP} --query principalId --output tsv) \
--query appId --output tsv) \
--role "Read Policy Status" \
--scope "/subscriptions/${SUBSCRIPTION}"

コーディングが完了し、Azureリソースの作成が完了したら、作業ディレクトリ配下で以下コマンドを実行しましょう。
作業ディレクトリ内のコードに従ってAzure Function App上に必要なパッケージがインストールされ、アプリがデプロイされます。

func azure functionapp publish ${FUNCTION_APP}

デプロイ完了後、Azure Portalで同期されたことを確認してみましょう。

[関数アプリ]→[Azure Functionsのアプリ名]→[関数名]→[コードとテスト]タブとクリックすることで、Azure Functionsでデプロイされたコードの状態が確認できます。

あとは、設定した時間通りにスクリプトが起動しアプリが作動し、Slackのチャンネルに通知が来るかを確認しましょう。
すばやく確認したい場合はfunction_app.pyの"run_on_startup"の設定を"True"にしてアプリデプロイ時にスクリプトを作動するようにするか、"schedule"のスクリプト実行間隔を短くするようにしてみましょう。

注意点

環境変数に設定した設定値について

このようなアプリを利用して本格的な運用を検討する場合、Slack Webhook URLなどのパラメータは、環境変数ではなくAzure Key Vaultなどの機密情報を管理するサービスに格納して取得することが望ましいです。
今回はブログ作成のため、手順を簡略化するためにAzure Functionsアプリの環境変数に直接格納しています。
もし検証後余裕があれば、Azure Key Vaultをに値を格納したうえでアプリを実行するようにスクリプトを書き換えてみてください。

おわりに

いかがでしたか？
今回のポリシー準拠状況をSlackに定期的に通知する仕組みは、ポリシー定義の仕組みと組み合わせることで様々な用途に応用できると思います。
ぜひAzureのガバナンス管理に活用してみてください！

私達ACS事業部はAzure・AKSなどのクラウドネイティブ技術を活用した内製化のご支援をしております。ご相談等ありましたらぜひご連絡ください。

www.ap-com.co.jp

また、一緒に働いていただける仲間も募集中です！
今年もまだまだ組織規模拡大中なので、ご興味持っていただけましたらぜひお声がけください。

www.ap-com.co.jp

本記事の投稿者: 青木平
柔道初段のクラウドエンジニア

APC 技術ブログ

株式会社エーピーコミュニケーションズの技術ブログです。

VSCodeでAzure Policyに準拠していないリソースをSlackに自動通知してみよう

はじめに

アプリの概要

構成図

通知イメージ

前提条件

作業前の準備

Azure Policyのポリシー定義IDの取得

SlackのWebhook URLの取得

VSCode上での開発環境の作成

コーディング

function_app.py

ソースコード

解説

requirements.txt

ソースコード

Azure Functions環境作成

リソース作成手順

注意点

環境変数に設定した設定値について

おわりに