Menu Close

How to use python to send and receive an Azure Service Bus notification without SDK

Azure Service Bus(ASB) uses AAD or SAS token to authenticate. Here we assume you are the owner and is able to generate a Send/Listen SAS key from Portal. Here is how to create a valid SAS token

def get_auth_token(sb_name, eh_name, sas_name, sas_value):
    # generate SAS token
    uri = "https://{}.servicebus.windows.net/{}".format(sb_name, eh_name)
    sas = sas_value.encode('utf-8')
    expiry = str(int(time.time() + 10000))
    string_to_sign = (urllib.parse.quote_plus(uri) + '\n' + expiry).encode('utf-8')
    signed_hmac_sha256 = hmac.HMAC(sas, string_to_sign, hashlib.sha256)
    signature = urllib.parse.quote(base64.b64encode(signed_hmac_sha256.digest()))
    return  {"uri": uri,
             "token":'SharedAccessSignature sr={}&sig={}&se={}&skn={}' \
                     .format(uri, signature, expiry, sas_name)
            }

Once token is generated. Sending and receiving becomes trivial. Below is the full code snippet I put together to generate SAS token and message its ip address via Azure Service Bus.

import time
import urllib
import hmac
import hashlib
import base64
import requests
import socket

h_name = socket.gethostname()
IP_address = socket.gethostbyname(h_name)

def get_auth_token(sb_name, eh_name, sas_name, sas_value):
    # generate SAS token
    uri = "https://{}.servicebus.windows.net/{}".format(sb_name, eh_name)
    sas = sas_value.encode('utf-8')
    expiry = str(int(time.time() + 10000))
    string_to_sign = (urllib.parse.quote_plus(uri) + '\n' + expiry).encode('utf-8')
    signed_hmac_sha256 = hmac.HMAC(sas, string_to_sign, hashlib.sha256)
    signature = urllib.parse.quote(base64.b64encode(signed_hmac_sha256.digest()))
    return  {"uri": uri,
             "token":'SharedAccessSignature sr={}&sig={}&se={}&skn={}' \
                     .format(uri, signature, expiry, sas_name)
            }

def send_message(token, message):
    # POST http{s}://{serviceNamespace}.servicebus.windows.net/{queuePath}/messages
    r=requests.post(token['uri']+"/messages",
    headers={
        "Authorization":token['token'],
        "Content-Type":"application/json"
    },
    json=message)

def recieve_message(token): 
    # DELETE	http{s}://{serviceNamespace}.servicebus.windows.net/{queuePath}/messages/head
    # 204 if no message
    r=requests.delete(token['uri']+"/messages/head",
    headers={
        "Authorization":token['token'],
    })
    return r.text

sb_name ="<service bus name>"
q_name = "<service bus queue name>"

skn = "<key name for that access key>"
key = "<access key created in portal>"

token=get_auth_token(sb_name, q_name, skn, key)
print (token['token'])

uri = "https://"+sb_name+".servicebus.windows.net/"+q_name+"/messages"

send_message(token, {'ip':IP_address})
recieve_message(token)

Leave a Reply

Your email address will not be published.