Python での Relay ハイブリッド接続 WebSocket 要求の概要

[アーティクル]
06/05/2024

NET
Node.js

このクイックスタートでは、WebSocket プロトコルを使用してメッセージを送受信する Python のセンダーアプリケーションとレシーバーアプリケーションを作成します。これらのアプリケーションには、Azure Relay のハイブリッド接続機能が使用されます。 Azure Relay 全般については、Azure Relay に関するページを参照してください。

このクイックスタートでは、以下の手順を実行します。

Azure Portal を使用した Relay 名前空間の作成
Azure Portal を使用した、その名前空間内のハイブリッド接続の作成
接続の詳細を保存する config.json プロパティファイルを生成する
ヘルパー関数 (SAS トークン、URL) の relaylib.py を開発する
メッセージを受信するサーバー (リスナー) スクリプトを記述する。
メッセージを送信するクライアント (センダー) スクリプトを記述する。
サーバー (リスナー) スクリプトと、必要に応じてクライアント (センダー) スクリプトを実行する。

前提条件

Python。 Python 3.10 以降を実行していることを確認します
Azure サブスクリプション。お持ちでない場合は、開始する前に無料アカウントを作成してください。

Azure Portal を使用した名前空間の作成

Azure portal にサインインします。
左側のメニューから、 [すべてのサービス] を選択します。 [統合] を選択し、[リレー] を検索し、[リレー] の上にマウスポインターを移動して、[作成] を選択します。
[名前空間の作成] ページで、これらの手順を実行します。
1. 名前空間を作成する Azure サブスクリプションを選択します。
2. [リソースグループ] で、名前空間を追加する既存のリソースグループを選択するか、新しいリソースグループを作成します。
3. リレー名前空間の名前を入力します。
4. 名前空間をホストするリージョンを選択します。
5. ページ下部にある [確認と作成] を選択します。
6. [確認および作成] ページで、 [作成] を選択します。
7. 数分後に、名前空間の [リレー] ページを確認します。

管理資格情報を取得する

[リレー] ページで、左側のメニューの [共有アクセスポリシー] を選択します。
[共有アクセスポリシー] ページで、 [RootManageSharedAccessKey] を選択します。
[SAS ポリシー:RootManageSharedAccessKey] の下で、 [プライマリ接続文字列] の横の [コピー] ボタンを選択します。これで、後で使用できるように接続文字列がクリップボードにコピーされます。この値をメモ帳などに一時的に貼り付けます。
前の手順を繰り返し、 [主キー] の値をコピーして、後で使用するために一時的な場所に貼り付けます。

Azure Portal を使用したハイブリッド接続の作成

名前空間の [リレー] ページで、次の手順に従ってハイブリッド接続を作成します。

左側のメニューの [エンティティ] の下で、[ハイブリッド接続] を選択し、[+ ハイブリッド接続] を選択します。
[ハイブリッド接続の作成] ページで、ハイブリッド接続の名前を入力し、[作成] を選択します。

ヘルパー関数を開発する

Python スクリプトを作成する

このスクリプトは、Azure Relay ハイブリッド接続を利用するアプリケーション用のヘルパー関数を提供します。これらの関数は、SAS トークンの生成や、セキュリティで保護された通信のための WebSocket 接続の確立などのタスクに役立つ可能性があります。

依存関係

ヘルパー関数スクリプトを生成する前に、pip を使用して次の Python ライブラリをインストールします: base64、hashlib、hmac、math、time、urllib

これらのライブラリは、次のコマンドを使用してインストールできます。

pip install <package name>

ヘルパー関数スクリプトを記述する

relaylib.py ファイルは次のようになります。

import base64
import hashlib
import hmac
import math
import time
import urllib

# Function which generates the HMAC-SHA256 of a given message
def hmac_sha256(key, msg):
   hash_obj = hmac.new(key=key, msg=msg, digestmod=hashlib._hashlib.openssl_sha256)
   return hash_obj.digest()

# Function to create a WebSocket URL for listening for a server application
def createListenUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=listen&sb-hc-id=123456'
   if token is not None:
       url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the url for the client application
def createSendUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=connect&sb-hc-id=123456'
   if token is not None:
	url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the Service Bus SAS token. 
def createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey):
   uri = "http://" + serviceNamespace + "/" + entityPath
   encodedResourceUri = urllib.parse.quote(uri, safe = '')

   # Define the token validity period in seconds (48 hours in this case)   
   tokenValidTimeInSeconds = 60 * 60 * 48 
   unixSeconds = math.floor(time.time())
   expiryInSeconds = unixSeconds + tokenValidTimeInSeconds

   # Create the plain signature string by combining the encoded URI and the expiry time
   plainSignature = encodedResourceUri + "\n" + str(expiryInSeconds)

   # Encode the SAS key and the plain signature as bytes
   sasKeyBytes = sasKey.encode("utf-8")
   plainSignatureBytes = plainSignature.encode("utf-8")
   hashBytes = hmac_sha256(sasKeyBytes, plainSignatureBytes)
   base64HashValue = base64.b64encode(hashBytes)

    # Construct the SAS token string
   token = "SharedAccessSignature sr=" + encodedResourceUri + "&sig=" +  urllib.parse.quote(base64HashValue) + "&se=" + str(expiryInSeconds) + "&skn=" + sasKeyName
   return token

サーバーアプリケーション (リスナー) の作成

Relay からのメッセージをリッスンおよび受信するために、Python WebSocket サーバースクリプトを記述します。

Python スクリプトを作成する

リレーの作成時に [クライアント認証が必要] オプションを無効にした場合は、ブラウザーを問わずハイブリッド接続 URL に要求を送信できます。保護されているエンドポイントにアクセスするには、ここに示したように SAS トークンを作成して渡す必要があります。

WebSocket を利用する SAS トークンを使ってハイブリッド接続 URL に要求を送信する方法を示す簡単な Python スクリプトを次に示します。

依存関係

サーバーアプリケーションを実行する前に、pip を使用して次の Python ライブラリをインストールします

asyncio、json、logging, websockets

これらのライブラリは、次のコマンドを使用してインストールできます。
```
 pip install <package name>
```
接続の詳細を保存する config.json ファイルを生成します
```
 {
"namespace": "HYBRID_CONNECTION_NAMESPACE",
"path": "HYBRID_CONNECTION_ENTITY_NAME",
"keyrule": "SHARED_ACCESS_KEY_NAME",
"key": "SHARED_ACCESS_PRIMARY_KEY"
 }
```
中かっこ内のプレースホルダーを、ハイブリッド接続の作成時に取得した値に置き換えます。
- namespace - Relay 名前空間。必ず完全修飾名前空間名を使用してください (たとえば、{namespace}.servicebus.windows.net)。
- path - ハイブリッド接続の名前。
- keyrule - 共有アクセスポリシーキーの名前 (既定では RootManageSharedAccessKey)。
- key - 前に保存した名前空間の主キー。

ヘルパー関数のヘルパー関数ファイルを生成します

次のファイルは relaylib.py として使用され、WebSocket URL の生成と SAS トークン用のヘルパー関数を含みます

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを生成する前に、pip を使用して次の Python ライブラリをインストールします: base64、hashlib、hmac、math、time、urllib

これらのライブラリは、次のコマンドを使用してインストールできます。

pip install <package name>

ヘルパー関数スクリプトを記述する

relaylib.py ファイルは次のようになります。

import base64
import hashlib
import hmac
import math
import time
import urllib

# Function which generates the HMAC-SHA256 of a given message
def hmac_sha256(key, msg):
   hash_obj = hmac.new(key=key, msg=msg, digestmod=hashlib._hashlib.openssl_sha256)
   return hash_obj.digest()

# Function to create a WebSocket URL for listening for a server application
def createListenUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=listen&sb-hc-id=123456'
   if token is not None:
       url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the url for the client application
def createSendUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=connect&sb-hc-id=123456'
   if token is not None:
	url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the Service Bus SAS token. 
def createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey):
   uri = "http://" + serviceNamespace + "/" + entityPath
   encodedResourceUri = urllib.parse.quote(uri, safe = '')

   # Define the token validity period in seconds (48 hours in this case)   
   tokenValidTimeInSeconds = 60 * 60 * 48 
   unixSeconds = math.floor(time.time())
   expiryInSeconds = unixSeconds + tokenValidTimeInSeconds

   # Create the plain signature string by combining the encoded URI and the expiry time
   plainSignature = encodedResourceUri + "\n" + str(expiryInSeconds)

   # Encode the SAS key and the plain signature as bytes
   sasKeyBytes = sasKey.encode("utf-8")
   plainSignatureBytes = plainSignature.encode("utf-8")
   hashBytes = hmac_sha256(sasKeyBytes, plainSignatureBytes)
   base64HashValue = base64.b64encode(hashBytes)

    # Construct the SAS token string
   token = "SharedAccessSignature sr=" + encodedResourceUri + "&sig=" +  urllib.parse.quote(base64HashValue) + "&se=" + str(expiryInSeconds) + "&skn=" + sasKeyName
   return token

メッセージを送信するコードを記述する

依存関係 config.json と relaylib.py がパスにあることを確認します

listener.py ファイルは次のようになります。

 import asyncio
 import json
 import logging
 import relaylib
 import websockets

 async def run_application(config):
     serviceNamespace = config["namespace"]
     entityPath = config["path"]
     sasKeyName = config["keyrule"]
     sasKey = config["key"]
     serviceNamespace += ".servicebus.windows.net"
     # Configure logging
     logging.basicConfig(level=logging.INFO)  # Enable DEBUG/INFO logging as appropriate

     try:
         logging.debug("Generating SAS Token for: %s", serviceNamespace)
 	token = relaylib.createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey)
 	logging.debug("Generating WebSocket URI")
 	wssUri = relaylib.createListenUrl(serviceNamespace, entityPath, token)
 	async with websockets.connect(wssUri) as websocket:
 	    logging.info("Listening for messages on Azure Relay WebSocket...")
 	    while True:
 		message = await websocket.recv()
 		logging.info("Received message: %s", message)
 	    except KeyboardInterrupt:
 		logging.info("Exiting listener.")

 if __name__ == "__main__":
     # Load configuration from JSON file
     with open("config.json") as config_file:
        config = json.load(config_file)

     asyncio.run(run_application(config))

クライアントアプリケーション (センダー) の作成

Relay にメッセージを送信するには、任意の HTTP または WebSocket クライアントを使用できます。含まれているサンプルは Python の実装です。

Python スクリプトを作成する

WebSocket を利用する SAS トークンを使ってハイブリッド接続 URL に要求を送信する方法を示す簡単な Python スクリプトを次に示します。

依存関係

クライアントアプリケーションを実行する前に、pip を使用して次の Python ライブラリをインストールします

asyncio、json、logging, websockets

これらのライブラリは、次のコマンドを使用してインストールできます。
```
 pip install <package name>
```
接続の詳細を保存する config.json ファイルを生成します
```
 {
"namespace": "HYBRID_CONNECTION_NAMESPACE",
"path": "HYBRID_CONNECTION_ENTITY_NAME",
"keyrule": "SHARED_ACCESS_KEY_NAME",
"key": "SHARED_ACCESS_PRIMARY_KEY"
}
```
中かっこ内のプレースホルダーを、ハイブリッド接続の作成時に取得した値に置き換えます。
- namespace - Relay 名前空間。必ず完全修飾名前空間名を使用してください (たとえば、{namespace}.servicebus.windows.net)。
- path - ハイブリッド接続の名前。
- keyrule - 共有アクセスポリシーキーの名前 (既定では RootManageSharedAccessKey)。
- key - 前に保存した名前空間の主キー。

ヘルパー関数のヘルパー関数ファイルを生成します

次のファイルは relaylib.py として使用され、WebSocket URL の生成と SAS トークン用のヘルパー関数を含みます

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを生成する前に、pip を使用して次の Python ライブラリをインストールします: base64、hashlib、hmac、math、time、urllib

これらのライブラリは、次のコマンドを使用してインストールできます。

pip install <package name>

ヘルパー関数スクリプトを記述する

relaylib.py ファイルは次のようになります。

import base64
import hashlib
import hmac
import math
import time
import urllib

# Function which generates the HMAC-SHA256 of a given message
def hmac_sha256(key, msg):
   hash_obj = hmac.new(key=key, msg=msg, digestmod=hashlib._hashlib.openssl_sha256)
   return hash_obj.digest()

# Function to create a WebSocket URL for listening for a server application
def createListenUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=listen&sb-hc-id=123456'
   if token is not None:
       url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the url for the client application
def createSendUrl(serviceNamespace, entityPath, token = None):
   url = 'wss://' + serviceNamespace + '/$hc/' + entityPath + '?sb-hc-action=connect&sb-hc-id=123456'
   if token is not None:
	url = url + '&sb-hc-token=' + urllib.parse.quote(token)
   return url

# Function which creates the Service Bus SAS token. 
def createSasToken(serviceNamespace, entityPath, sasKeyName, sasKey):
   uri = "http://" + serviceNamespace + "/" + entityPath
   encodedResourceUri = urllib.parse.quote(uri, safe = '')

   # Define the token validity period in seconds (48 hours in this case)   
   tokenValidTimeInSeconds = 60 * 60 * 48 
   unixSeconds = math.floor(time.time())
   expiryInSeconds = unixSeconds + tokenValidTimeInSeconds

   # Create the plain signature string by combining the encoded URI and the expiry time
   plainSignature = encodedResourceUri + "\n" + str(expiryInSeconds)

   # Encode the SAS key and the plain signature as bytes
   sasKeyBytes = sasKey.encode("utf-8")
   plainSignatureBytes = plainSignature.encode("utf-8")
   hashBytes = hmac_sha256(sasKeyBytes, plainSignatureBytes)
   base64HashValue = base64.b64encode(hashBytes)

    # Construct the SAS token string
   token = "SharedAccessSignature sr=" + encodedResourceUri + "&sig=" +  urllib.parse.quote(base64HashValue) + "&se=" + str(expiryInSeconds) + "&skn=" + sasKeyName
   return token

メッセージを送信するコードを記述する

依存関係 config.json と relaylib.py がパスにあることを確認します

sender.py ファイルは次のようになります。

 import asyncio
 import json
 import logging
 import relaylib
 import websockets

 async def run_application(message, config):
 	service_namespace = config["namespace"]
 	entity_path = config["path"]
 	sas_key_name = config["keyrule"]
 	sas_key = config["key"]
 	service_namespace += ".servicebus.windows.net"

 	# Configure logging
 	logging.basicConfig(level=logging.DEBUG)  # Enable debug logging

 	token = relaylib.createSasToken(service_namespace, entity_path, sas_key_name, sas_key)
 	logging.debug("Token: %s", token)
 	wss_uri = relaylib.createListenUrl(service_namespace, entity_path, token)
 	logging.debug("WssURI: %s", wss_uri)

 	try:
 		async with websockets.connect(wss_uri) as websocket:
 			logging.info("Sending message to Azure Relay WebSocket...")
 			await websocket.send(json.dumps({"message": message}))
 			logging.info("Message sent: %s", message)
 	except Exception as e:
 		logging.error("An error occurred: %s", str(e))

 if __name__ == "__main__":
 	# Load configuration from JSON file
 	with open("config.json") as config_file:
 		config = json.load(config_file)

 	asyncio.run(run_application("This is a message to Azure Relay Hybrid Connections!", config))

Note

この記事のサンプルコードでは、接続文字列を使用して Azure Relay 名前空間に対する認証を行うことで、チュートリアルを簡素化します。運用環境では、接続文字列や共有アクセス署名ではなく、Microsoft Entra ID 認証を使用することをお勧めします。前者では、セキュリティ侵害の可能性が高まる場合があります。 Microsoft Entra ID 認証を使用するための詳細な情報とサンプルコードについては、「Azure Relay エンティティにアクセスするための Microsoft Entra ID を使用したアプリケーションの認証と承認」および、「Azure Relay リソースにアクセスするための Microsoft Entra ID でのマネージド ID の認証」を参照してください。

アプリケーションの実行

サーバーアプリケーションを実行: コマンドプロンプトで「python3 listener.py」と入力します。
クライアントアプリケーションを実行: コマンドプロンプトで「python3 sender.py」と入力します。

これで、Python を使ってエンドツーエンドのハイブリッド接続アプリケーションを作成できました。

次のステップ

このクイックスタートでは、WebSocket を使用してメッセージを送受信する Python のクライアントアプリケーションとサーバーアプリケーションを作成しました。 Azure Relay のハイブリッド接続機能は、HTTP を使用したメッセージの送受信もサポートしています。 Azure Relay のハイブリッド接続で HTTP を使用する方法については、HTTP のクイックスタートを参照してください。

このクイックスタートでは、Python を使ってクライアントとサーバーのアプリケーションを作成しました。 .NET Framework を使用してクライアントアプリケーションとサーバーアプリケーションを作成する方法については、.NET HTTP のクイックスタートまたは .NET HTTP のクイックスタートを参照してください。

次の方法で共有

Python での Relay ハイブリッド接続 WebSocket 要求の概要

前提条件

Azure Portal を使用した名前空間の作成

管理資格情報を取得する

Azure Portal を使用したハイブリッド接続の作成

ヘルパー関数を開発する

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを記述する

サーバーアプリケーション (リスナー) の作成

Python スクリプトを作成する

依存関係

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを記述する

メッセージを送信するコードを記述する

クライアントアプリケーション (センダー) の作成

Python スクリプトを作成する

依存関係

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを記述する

メッセージを送信するコードを記述する

アプリケーションの実行

次のステップ

フィードバック

その他のリソース

次の方法で共有

Python での Relay ハイブリッド接続 WebSocket 要求の概要

前提条件

Azure Portal を使用した名前空間の作成

管理資格情報を取得する

Azure Portal を使用したハイブリッド接続の作成

ヘルパー関数を開発する

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを記述する

サーバー アプリケーション (リスナー) の作成

Python スクリプトを作成する

依存関係

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを記述する

メッセージを送信するコードを記述する

クライアント アプリケーション (センダー) の作成

Python スクリプトを作成する

依存関係

Python スクリプトを作成する

依存関係

ヘルパー関数スクリプトを記述する

メッセージを送信するコードを記述する

アプリケーションの実行

次のステップ

フィードバック

その他のリソース

サーバーアプリケーション (リスナー) の作成

クライアントアプリケーション (センダー) の作成