Extend generate token

HISTORY.rst

@@ -17,18 +17,19 @@ Release History
 * Fix for "az iot hub c2d-message receive" - the command will use the "ContentEncoding" header value (which indicates the message body encoding)
   or fallback to utf-8 to decode the received message body.
 
+* Addition for "az iot hub generate-sas-token" - the command will allow offline generation of a SAS Token using a connection string.
+
 * Changes to Edge validation for set-modules and edge deployment creation:
 
   By default only properties of system modules $edgeAgent and $edgeHub are validated against schemas installed with the IoT extension.
   This can be disabled by using the --no-validation switch.
 
-
 **Azure Digital Twins updates**
 
 * Addition of the following commands
 
   * az dt reset - Preview command which deletes all data entities from the target instance (models, twins, twin relationships).
-  
+
 
 0.10.13
 +++++++++++++++
@@ -46,7 +47,7 @@ Release History
 * Public API GA update
 
   * Remove preview tag for  api-token, device, device-template, user routes. Default routes use central GA API's.
-  * Add support for preview and 1.0 routes. 
+  * Add support for preview and 1.0 routes.
   * Addition of the optional '--av' argument to specify the version of API for the requested operation.
 
 **IoT Hub updates**
@@ -114,7 +115,7 @@ For more information about IoT Hub support for AAD visit: https://docs.microsoft
 
 * Addition of the following commands
 
-  * az iot central device manual-failover - Execute a manual failover of device across multiple IoT Hubs 
+  * az iot central device manual-failover - Execute a manual failover of device across multiple IoT Hubs
   * az iot central device manual-failback - Reverts the previously executed failover command by moving the device back to it's original IoT Hub
 
 For more information about device high availability visit https://github.com/iot-for-all/iot-central-high-availability-clients#readme
@@ -150,7 +151,7 @@ For more information about device high availability visit https://github.com/iot
 **IoT Hub updates**
 
 * Improve http debug logging.
-* Fix bug related to issue #296. Adds a clause to device-identity update that allows user to update primary-key / secondary-key 
+* Fix bug related to issue #296. Adds a clause to device-identity update that allows user to update primary-key / secondary-key
   and primary-thumbprint / secondary-thumbprint values (respectively, per auth method) without needing to specify the auth_method in the update command.
 
 

azext_iot/_help.py

@@ -577,6 +577,18 @@
       text: >
         az iot hub generate-sas-token -d {device_id}
         --login 'HostName=myhub.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=12345'
+    - name: Generate an Iot Hub SAS token using an IoT Hub connection string
+      text: >
+        az iot hub generate-sas-token
+        --connection-string 'HostName=myhub.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=12345'
+    - name: Generate a Device SAS token using a Device connection string
+      text: >
+        az iot hub generate-sas-token --connection-string
+        'HostName=myhub.azure-devices.net;DeviceId=mydevice;SharedAccessKeyName=iothubowner;SharedAccessKey=12345'
+    - name: Generate a Module SAS token using a Module connection string
+      text: >
+        az iot hub generate-sas-token --connection-string
+        'HostName=myhub.azure-devices.net;DeviceId=mydevice;ModuleId=mymodule;SharedAccessKeyName=iothubowner;SharedAccessKey=12345'
 """
 
 helps[

azext_iot/_params.py

@@ -232,6 +232,13 @@ def load_arguments(self, _):
             arg_type=get_three_state_flag(),
             help="Flag indicating edge enablement.",
         )
+        context.argument(
+            "connection_string",
+            options_list=["--connection-string", "--cs"],
+            help="Target connection string. This bypasses the IoT Hub registry and generates the SAS token directly"
+            " from the supplied symmetric key without further validation. All other command parameters aside from"
+            " duration will be ignored. Supported connection string types: Iot Hub, Device, Module."
+        )
 
     with self.argument_context("iot hub") as context:
         context.argument(

azext_iot/_validators.py

@@ -16,6 +16,7 @@ def mode2_iot_login_handler(cmd, namespace):
             login_value = args['login']
             iot_cmd_type = None
             entity_value = None
+            offline = None
 
             if 'hub_name' in args:
                 iot_cmd_type = 'IoT Hub'
@@ -24,5 +25,8 @@ def mode2_iot_login_handler(cmd, namespace):
                 iot_cmd_type = 'DPS'
                 entity_value = args['dps_name']
 
-            if not any([login_value, entity_value]):
+            if 'connection_string' in args:
+                offline = args['connection_string']
+
+            if not any([login_value, entity_value, offline]):
                 raise CLIError(error_no_hub_or_login_on_input(iot_cmd_type))

azext_iot/common/_azure.py

@@ -23,11 +23,6 @@ def _parse_connection_string(cs, validate=None, cstring_type="entity"):
     return decomposed
 
 
-def parse_pnp_connection_string(cs):
-    validate = ["HostName", "RepositoryId", "SharedAccessKeyName", "SharedAccessKey"]
-    return _parse_connection_string(cs, validate, "PnP Model Repository")
-
-
 def parse_iot_hub_connection_string(cs):
     validate = ["HostName", "SharedAccessKeyName", "SharedAccessKey"]
     return _parse_connection_string(cs, validate, "IoT Hub")

azext_iot/common/shared.py

@@ -267,3 +267,18 @@ class IoTHubStateType(Enum):
     KeyEncryptionKeyRevoking = "KeyEncryptionKeyRevoking"
     KeyEncryptionKeyRevoked = "KeyEncryptionKeyRevoked"
     ReActivating = "ReActivating"
+
+
+class ConnectionStringParser(Enum):
+    """
+        All connection string parser with respective functions
+    """
+    from azext_iot.common._azure import (
+        parse_iot_device_connection_string,
+        parse_iot_device_module_connection_string,
+        parse_iot_hub_connection_string
+    )
+
+    Module = parse_iot_device_module_connection_string
+    Device = parse_iot_device_connection_string
+    IotHub = parse_iot_hub_connection_string

azext_iot/operations/hub.py

@@ -29,6 +29,7 @@
     RenewKeyType,
     IoTHubStateType,
     DeviceAuthApiType,
+    ConnectionStringParser,
 )
 from azext_iot.iothub.providers.discovery import IotHubDiscovery
 from azext_iot.common.utility import (
@@ -1958,6 +1959,7 @@ def iot_get_sas_token(
     login=None,
     module_id=None,
     auth_type_dataplane=None,
+    connection_string=None,
 ):
     key_type = key_type.lower()
     policy_name = policy_name.lower()
@@ -1975,6 +1977,14 @@ def iot_get_sas_token(
             "You are unable to get sas token for module without device information."
         )
 
+    if connection_string:
+        return {
+            DeviceAuthApiType.sas.value: _iot_build_sas_token_from_cs(
+                connection_string,
+                duration,
+            ).generate_sas_token()
+        }
+
     return {
         DeviceAuthApiType.sas.value: _iot_build_sas_token(
             cmd,
@@ -1991,6 +2001,44 @@ def iot_get_sas_token(
     }
 
 
+def _iot_build_sas_token_from_cs(connection_string, duration=3600):
+    uri = None
+    policy = None
+    key = None
+
+    parsed_cs = None
+    all_parsers = [
+        ConnectionStringParser.Module,
+        ConnectionStringParser.Device,
+        ConnectionStringParser.IotHub,
+    ]
+
+    for parser in all_parsers:
+        try:
+            parsed_cs = parser(connection_string)
+
+            if "SharedAccessKeyName" in parsed_cs:
+                policy = parsed_cs["SharedAccessKeyName"]
+            key = parsed_cs["SharedAccessKey"]
+
+            if parser == ConnectionStringParser.IotHub:
+                uri = parsed_cs["HostName"]
+            elif parser == ConnectionStringParser.Module:
+                uri = "{}/devices/{}/modules/{}".format(
+                    parsed_cs["HostName"], parsed_cs["DeviceId"], parsed_cs["ModuleId"]
+                )
+            elif parser == ConnectionStringParser.Device:
+                uri = "{}/devices/{}".format(parsed_cs["HostName"], parsed_cs["DeviceId"])
+            else:
+                raise CLIError("Given Connection String was not in a supported format.")
+
+            return SasTokenAuthentication(uri, policy, key, duration)
+        except ValueError:
+            continue
+
+    raise CLIError("Connection String did not match any presets")
+
+
 def _iot_build_sas_token(
     cmd,
     hub_name=None,
@@ -2023,6 +2071,7 @@ def _iot_build_sas_token(
     uri = None
     policy = None
     key = None
+
     if device_id:
         logger.info(
             'Obtaining device "%s" details from registry, using IoT Hub policy "%s"',

azext_iot/tests/iothub/test_iothub_utilities_int.py

@@ -62,6 +62,17 @@ def test_iothub_generate_sas_token(self):
             expect_failure=True,
         )
 
+        # Offline SAS token generation
+        self.cmd(
+            f"iot hub generate-sas-token --connection-string {self.connection_string}",
+            checks=[self.exists("sas")],
+        )
+
+        self.cmd(
+            f"iot hub generate-sas-token --connection-string {self.connection_string} --du 1000",
+            checks=[self.exists("sas")],
+        )
+
     def test_iothub_connection_string_show(self):
         conn_str_pattern = r"^HostName={0}.azure-devices.net;SharedAccessKeyName=iothubowner;SharedAccessKey=".format(
             LIVE_HUB

azext_iot/tests/iothub/test_iothub_utilities_unit.py

@@ -0,0 +1,85 @@
+# coding=utf-8
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+from azext_iot.common.sas_token_auth import SasTokenAuthentication
+import pytest
+from knack.cli import CLIError
+from azext_iot.operations import hub as subject
+from azext_iot.tests.generators import generate_generic_id
+
+
+def generate_valid_cs(validate_pairs=[]):
+    host_name = generate_generic_id()
+    shared_access_key = generate_generic_id()
+    cs = f"HostName={host_name};"
+    input_pairs = dict((k, generate_generic_id()) for k in validate_pairs)
+    policy = input_pairs["SharedAccessKeyName"] if "SharedAccessKeyName" in input_pairs else None
+
+    for key, value in input_pairs.items():
+        cs += "{}={};".format(
+            key, value
+        )
+
+    cs = f"{cs}SharedAccessKey={shared_access_key}"
+    uri = host_name
+    if "DeviceId" in input_pairs:
+        uri = f"{uri}/devices/{input_pairs['DeviceId']}"
+    if "ModuleId" in input_pairs:
+        uri = f"{uri}/modules/{input_pairs['ModuleId']}"
+
+    return {
+        "connection_string": cs,
+        "uri": uri,
+        "policy": policy,
+        "key": shared_access_key
+    }
+
+
+class TestGenerateSasToken:
+    @pytest.mark.parametrize(
+        "duration, req",
+        [
+            (3600, generate_valid_cs(["DeviceId"])),
+            (30, generate_valid_cs(["DeviceId"])),
+            (60000, generate_valid_cs(["DeviceId"])),
+            (3600, generate_valid_cs(["SharedAccessKeyName"])),
+            (3600, generate_valid_cs(["DeviceId"])),
+            (3600, generate_valid_cs(["DeviceId", "ModuleId"])),
+            (3600, generate_valid_cs(["Test", "DeviceId", "ModuleId"])),
+            (3600, generate_valid_cs(["RepositoryId", "DeviceId", "ModuleId"])),
+        ],
+    )
+    def test_generate_sas_token_from_cs(self, mocker, fixture_cmd, duration, req):
+        patched_time = mocker.patch(
+            "azext_iot.common.sas_token_auth.time"
+        )
+        patched_time.return_value = 0
+        result = subject.iot_get_sas_token(
+            cmd=fixture_cmd,
+            connection_string=req["connection_string"],
+            duration=duration
+        )
+
+        duration = duration if duration else 3600
+        expected_sas = SasTokenAuthentication(
+            req["uri"], req["policy"], req["key"], duration
+        ).generate_sas_token()
+        assert result["sas"] == expected_sas
+
+    @pytest.mark.parametrize(
+        "req",
+        [
+            (generate_valid_cs()),
+            (generate_valid_cs(["ModuleId"])),
+            (generate_valid_cs(["Test"]))
+        ],
+    )
+    def test_generate_sas_token_from_cs_error(self, mocker, fixture_cmd, req):
+        with pytest.raises(CLIError):
+            subject.iot_get_sas_token(
+                cmd=fixture_cmd,
+                connection_string=req["connection_string"],
+            )