このシナリオのメイン アプリは、Azure App Service にデプロイされた単純な Flask アプリです。 このアプリは 、/api/v1/getcode という名前のパブリック API エンドポイントを提供します。これは、アプリ内の他の目的のためにコードを生成します (たとえば、人間のユーザーに対する 2 要素認証を使用)。 メイン アプリには、API エンドポイントへのリンクを表示する簡単なホーム ページも用意されています。
サンプルのプロビジョニング スクリプトでは、次の手順を実行します。
App Service ホストを作成し、
az webapp upAzure CLI コマンドを使用してコードをデプロイします。メイン アプリの Azure Storage アカウントを作成します (
az storage account createを使用)。(
az storage queue createを使用して) "code-requests" という名前のストレージ アカウントにキューを作成します。アプリがキューへの書き込みを許可されるようにするには、
az role assignment createを使用して"ストレージ キュー データ共同作成者" ロールをアプリに割り当てます。 ロールの詳細については、「 Azure CLI を使用してロールのアクセス許可を割り当てる方法」を参照してください。
主なアプリ コードは次のとおりです。重要な詳細については、このシリーズの次の部分で説明します。
from flask import Flask, request, jsonify
import requests, random, string, os
from datetime import datetime
from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
from azure.storage.queue import QueueClient
app = Flask(__name__)
app.config["DEBUG"] = True
number_url = os.environ["THIRD_PARTY_API_ENDPOINT"]
# Authenticate with Azure. First, obtain the DefaultAzureCredential
credential = DefaultAzureCredential()
# Next, get the client for the Key Vault. You must have first enabled managed identity
# on the App Service for the credential to authenticate with Key Vault.
key_vault_url = os.environ["KEY_VAULT_URL"]
keyvault_client = SecretClient(vault_url=key_vault_url, credential=credential)
# Obtain the secret: for this step to work you must add the app's service principal to
# the key vault's access policies for secret management.
api_secret_name = os.environ["THIRD_PARTY_API_SECRET_NAME"]
vault_secret = keyvault_client.get_secret(api_secret_name)
# The "secret" from Key Vault is an object with multiple properties. The key we
# want for the third-party API is in the value property.
access_key = vault_secret.value
# Set up the Storage queue client to which we write messages
queue_url = os.environ["STORAGE_QUEUE_URL"]
queue_client = QueueClient.from_queue_url(queue_url=queue_url, credential=credential)
@app.route('/', methods=['GET'])
def home():
return f'Home page of the main app. Make a request to <a href="./api/v1/getcode">/api/v1/getcode</a>.'
def random_char(num):
return ''.join(random.choice(string.ascii_letters) for x in range(num))
@app.route('/api/v1/getcode', methods=['GET'])
def get_code():
headers = {
'Content-Type': 'application/json',
'x-functions-key': access_key
}
r = requests.get(url = number_url, headers = headers)
if (r.status_code != 200):
return "Could not get you a code.", r.status_code
data = r.json()
chars1 = random_char(3)
chars2 = random_char(3)
code_value = f"{chars1}-{data['value']}-{chars2}"
code = { "code": code_value, "timestamp" : str(datetime.utcnow()) }
# Log a queue message with the code for, say, a process that invalidates
# the code after a certain period of time.
queue_client.send_message(code)
return jsonify(code)
if __name__ == '__main__':
app.run()