使用python SAS令牌连接到服务总线

Question

我需要使用SAS令牌连接到Azure服务总线（生成并连接）。

我没有看到任何关于python实现的东西。

此链接提供Eventhubs的实现-

https://docs.microsoft.com/en-us/rest/api/eventhub/generate-sas-token#python

不确定在哪里可以找到servicebus的python实现。

Answer 1

使用Azure门户创建服务总线后，ServiceBusService对象可用于处理队列。

有关创建队列、向队列发送消息、使用python从队列接收消息以编程方式访问服务总线的详细信息，请参阅此文档。

Answer 2

我已经找到了一种方法，您可以在ServiceBusService类中执行此操作。运行"pip安装azure.servicebus"后，我将其导入为：

from azure.servicebus.control_client import ServiceBusService

ServiceBusService构造函数接受一个名为“authentication”的参数，默认情况下未指定该参数。

如果进入ServiceBusService init文件，您可以更详细地了解身份验证是如何处理的。

if authentication:
        self.authentication = authentication
    else:
        if not account_key:
            account_key = os.environ.get(AZURE_SERVICEBUS_ACCESS_KEY)
        if not issuer:
            issuer = os.environ.get(AZURE_SERVICEBUS_ISSUER)

        if shared_access_key_name and shared_access_key_value:
            self.authentication = ServiceBusSASAuthentication(
                shared_access_key_name,
                shared_access_key_value)
        elif account_key and issuer:
            self.authentication = ServiceBusWrapTokenAuthentication(
                account_key,
                issuer)

如果未传递自定义身份验证对象，则它将尝试使用ServiceBusSAS身份验证类，如果填充shared_access_key_name和shared_access_key_value，则默认使用该类。

因此，如果跳入ServiceBusSASAuthentication类，您会注意到一些有用的东西。

class ServiceBusSASAuthentication:
    def __init__(self, key_name, key_value):
        self.key_name = key_name
        self.key_value = key_value
        self.account_key = None
        self.issuer = None

    def sign_request(self, request, httpclient):
        request.headers.append(
            ('Authorization', self._get_authorization(request, httpclient)))

    def _get_authorization(self, request, httpclient):
        uri = httpclient.get_uri(request)
        uri = url_quote(uri, '').lower()
        expiry = str(self._get_expiry())

        to_sign = uri + '\n' + expiry
        signature = url_quote(_sign_string(self.key_value, to_sign, False), '')

        auth_format = 'SharedAccessSignature sig={0}&se={1}&skn={2}&sr={3}'  # <----awww, yeah
        auth = auth_format.format(signature, expiry, self.key_name, uri)

        return auth  # <--after inserting values into string, the SAS Token is just returned.

    def _get_expiry(self):  # pylint: disable=no-self-use
        '''Returns the UTC datetime, in seconds since Epoch, when this signed
        request expires (5 minutes from now).'''
        return int(round(time.time() + 300))

sign_request函数是ServiceBusService类在进行身份验证时直接引用的唯一函数，但是您会注意到它所做的只是向请求添加身份验证标头...是SAS代号的形式。

因此，在这一点上，我拥有了创建自己的身份验证类所需的所有信息。我做了一个和这个一模一样的。

class ServiceBusSASTokenAuthentication:
    def __init__(self, sas_token):
        self.sas_token = sas_token

    # this method is the one used by ServiceBusService for authentication, need to leave signature as is
    # even though we don't use httpClient like the original.
    def sign_request(self, request, httpclient):
        request.headers.append(
            ('Authorization', self._get_authorization())
        )

    def _get_authorization(self):
        return self.sas_token

我可能可以一起去掉_get_auth函数，但我还没有完善所有的东西。

现在，如果您使用有效的SAS令牌在ServiceBusService构造函数中像这样调用这个类，它应该可以工作。

            subscription_client = ServiceBusService(
                authentication=ServiceBusSASTokenAuthentication(sas_token=sas_token),
                service_namespace=service_namespace
            )