Copy from base64 import b64encode, urlsafe_b64decode
from hashlib import sha256
from time import time
from urllib import parse
from hmac import HMAC
import os
import requests
def generate_sas_token(signing_key, resource_uri, expiry_in_seconds, device_connector_id, device_id):
query = f"deviceConnectorIdAudience={device_connector_id}\n"
if device_id:
query += f"deviceIdAudience={device_id}\n"
query += f"expiry={expiry_in_seconds}"
query = parse.quote_plus(query).encode('utf-8')
padded_signing_key = signing_key
missing_padding = len(signing_key) % 4
if missing_padding:
padded_signing_key += '=' * (4 - missing_padding)
padded_signing_key = urlsafe_b64decode(padded_signing_key)
signature = b64encode(HMAC(padded_signing_key, query, sha256).digest()).decode("utf-8")
encoded_signature = parse.quote_plus(signature.encode('utf-8'))
token = f"sig={encoded_signature}&exp={expiry_in_seconds}&aud={parse.quote_plus(resource_uri)}"
return b64encode(token.encode('utf-8')).decode("utf-8")
if __name__ == "__main__":
base_url = "https://data-gateway.akenza.io"
signing_key = "<primaryOrSecondarySigningKey>"
device_id = "<yourPhysicalDEviceId>"
device_connector_id = "<yourDeviceConnectorId>"
expiry = round(time() + 60 * 60 * 24)
resource_uri = f"https://akenza.io/device-connectors/{device_connector_id}/devices/{device_id}"
token = generate_sas_token(signing_key, resource_uri, expiry, device_connector_id, device_id)
headers = {
'Content-Type': 'application/json',
"x-access-signature": token
}
body = {
"param1": "value"
}
response = requests.request("POST", f"{base_url}/v3/capture", headers=headers, json=body, timeout=10)
response.raise_for_status()
print("uplink sent")