KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _generate_sas_token method in lisa

Best Python code snippet using lisa_python

azure_credentials.py

Source:azure_credentials.py

...49    @return: (Tuple[str, str]) The username and password, respectively50    """51    hostname = _retrieve_hostname(scope_id, device_id, device_key)52    username = "{}/{}/?api-version=2018-06-30".format(hostname, device_id)53    password = _generate_sas_token(hostname, device_key)54    return username, password55def _retrieve_hostname(scope_id, device_id, device_key):56    """Retrieve the IoT Central hostname associated to the device57    @param scope_id:   (str) The device's Scope ID58    @param device_id:  (str) The device's ID59    @param device_key: (str) The device's Shared Access Key60    @return:           (str) The IoT Central hostname61    """62    # Get the authentication token for the requests63    expiration = int(time() + 30)64    resource = "{}%2Fregistrations%2F{}".format(scope_id, device_id)65    auth_token = _generate_sas_token(resource, device_key, expiration)66    auth_token += "&skn=registration"67    # Set up the initial HTTP request68    endpoint = "{}/{}/registrations/{}/".format(AZURE_DPS_ENDPOINT, scope_id, device_id)69    registration = "register?api-version=2018-11-01"70    headers = {71        "Accept": "application/json",72        "Content-Type": "application/json; charset=utf-8",73        "Connection": "keep-alive",74        "UserAgent": "prov_device_client/1.0",75        "Authorization": auth_token76    }77    data = {78        "registrationId": device_id79    }80    # Place a registration request for the device (it should already be registered)81    result = requests.put(endpoint + registration, data=json.dumps(data), headers=headers)82    data = json.loads(result.text)83    # Continue checking device's registration status until it resolves84    while data.get("status") == "assigning" and result.ok:85        operation_id = data.get("operationId")86        operation = "operations/{}?api-version=2018-11-01".format(operation_id)87        result = requests.get(endpoint + operation, headers=headers)88        data = json.loads(result.text)89        sleep(1)  # Pause for a bit90    # Get the device's assigned hub91    if not result.ok:92        logger.error("Ran into an error: %s %s", result.status_code, result.text)93    else:94        hub = data.get("registrationState").get("assignedHub")95        logger.debug("Retrieved hostname: %s", hub)96        return hub97def _generate_sas_token(resource, device_key, expiration=None):98    """Create a SAS token for authentication. More information:99    https://docs.microsoft.com/en-us/azure/iot-hub/iot-hub-devguide-security100    @param resource:   (str) The IoT Central resource for which the key is created101    @param device_key: (str) The device's Shared Access Key102    @param expiration: (int) The time at which the token expires103    @return:           (str) The SAS token104    """105    if not expiration:106        expiration = int(time() + AZURE_TOKEN_EXPIRATION)107    sign_key = "{}\n{}".format(resource, expiration)108    signature = b64encode((HMAC(b64decode(device_key.encode("utf-8")), sign_key.encode("utf-8"), sha256).digest()))109    signature = quote(signature)110    return "SharedAccessSignature sr={}&sig={}&se={}".format(111        resource,...

azure_client.py

Source:azure_client.py

...28            hostname=self.endpoint,29            device_id=self.device_id30        )31        # generate a SAS token from URI, shared access key and policy name32        sas_token = self._generate_sas_token(self.uri, self.key, self.policy)33        # check if device already exists in registry and register if not34        register_url = "https://{uri}?api-version={version}".format(35            uri=self.uri,36            version=self.API_VERSION37        )38        r = requests.get(register_url, headers={'Content-Type': 'application/json', 'Authorization': sas_token})39        if r.status_code == 404:  # device is not registered yet40            r.close()41            self.device_key = self._register_device(register_url, sas_token)42        elif r.status_code == 200:43            print("Device already registered: {}".format(r.text))44            self.device_key = _get_device_key(r)45        else:46            raise Exception(47                "!! GET request failed with status code {}: {}".format(r.status_code, r.text))48        # generate URL for device to cloud messaging49        self.message_url = "https://{uri}/messages/events?api-version={version}".format(50            uri=self.uri,51            version=self.API_VERSION52        )53    def _register_device(self, url, sas_token):54        """55        create new device identity in Azure IoT hub device registry56        """57        body = '{deviceId: "%s"}' % self.device_id58        r = requests.put(url, headers={'Content-Type': 'application/json', 'Authorization': sas_token}, data=body)59        if r.status_code == 200:60            print("Registered device at hub: {}".format(r.text))61            return _get_device_key(r)62        else:63            raise Exception(64                "!! Device not registered. Request failed with status code {}: {}".format(r.status_code, r.text))65    def _generate_sas_token(self, uri, key, policy_name, valid_secs=10):66        # uri = quote(uri, safe='').lower()67        encoded_uri = quote(uri, safe='')68        expiry = time() + valid_secs69        ttl = int(expiry)70        sign_key = '%s\n%d' % (encoded_uri, ttl)71        signature = b2a_base64(HMAC(a2b_base64(key), sign_key.encode('utf-8'), sha256).digest())72        token = 'SharedAccessSignature ' + urlencode({73            'sr': uri,74            'sig': signature[:-1],75            'se': str(ttl),76            'skn': policy_name77        })78        return token79    def send(self, msg: str):80        # generate a new SAS token for every message to make sure it's still valid81        sas_token = self._generate_sas_token(self.uri, self.key, self.policy)82        r = requests.post(self.message_url, headers={'Authorization': sas_token}, data=msg)83        if r.status_code == 204:84            r.close()85        else:86            raise Exception(...

iothub.py

Source:iothub.py

...20        self._module_id = parsed_connection.get("ModuleId")21        self._username = self._hostname + '/' + self._device_id22        self._c2d_cb = None23        24        self._sas = self._generate_sas_token()25        26        c = MQTTClient(client_id=self._device_id, server=self._hostname, port=8883, user=self._username, password=self._sas, keepalive=120, ssl=True)27        c.DEBUG = True28        c.set_callback(self._callback_handler)29        30        self._mqtt_client = c31        32    def connect(self):33        try:34            self._mqtt_client.reconnect()35            print("Connected to IoT Hub")36        except:37            print("Could not connect")38            39    def send_telemetry(self, msg):  40        try:41            topic=self._get_topic_base() + "/messages/events/"42            self._mqtt_client.publish(topic=topic, msg=msg)43        except:44            print("Could not send telemetry")45    46    def set_c2d_cb(self, cb):47        try:48            self._mqtt_client.subscribe(topic=self._get_topic_base() + "/messages/devicebound/#")49            self._c2d_cb = cb50        except:51            print("Could not set cloud2device msg callback")52            53    def check_msg(self):54        self._mqtt_client.check_msg()55    def _get_topic_base(self):56        if self._module_id:57            base_str = "devices/" + self._device_id + "/modules/" + self._module_id58        else:59            base_str = "devices/" + self._device_id60        return base_str61    def _parse_connection(self):62        cs_args = self._connection_string.split(";")63        dictionary = dict(arg.split("=", 1) for arg in cs_args)64        65        return dictionary66    67    def _generate_sas_token(self, expiry = 86400):  # default to one day expiry period68        print("Retrieving NTP time for token expiration")69        urlencoder=urlencode.Urlencode()70        now=071        while now == 0:72            try:73                now=ntptime.time() + 946684800 # offset for embedded vs POSIX epoch.74            except:75                time.sleep(1)76                print("Failed retrieving NTP time, retrying.")77                78        print("Generating SAS token from key")79        ttl=now + expiry80        urlToSign=urlencoder.quote(self._hostname + '/devices/' + self._device_id)81        msg="{0}\n{1}".format(urlToSign, ttl).encode('utf-8')82        key=base64.b64decode(self._shared_access_key)83        h=hmac.HMAC(key, msg = msg, digestmod = sha256)84        decodedDigest=base64.b64encode(h.digest()).decode()85        signature=urlencoder.quote(decodedDigest)86        sas="SharedAccessSignature sr={0}&sig={1}&se={2}".format(87            urlToSign, signature, ttl)88        return sas89    90    def _renew_sas_token(self):     91        if time.ticks_diff(time.time(), self._lastUpdated) > 60 * 15:92            self._lastUpdated = time.time()93            self._updateSas = True94        if self._updateSas:95            self._sas = self._generate_sas_token()96            print('Updating Sas')97            self._updateSas = False98            99    def _callback_handler(self, topic, msg):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.