Authentication Pattern Mapping — SAS to Entra¶

# Device-side SAS authentication (BEFORE)
from azure.iot.device import IoTHubDeviceClient

# Connection string contains the device symmetric key
conn_str = (
    "HostName=hub-prod.azure-devices.net;"
    "DeviceId=sensor-floor3-unit47;"
    "SharedAccessKey=<device-symmetric-key>"
)

client = IoTHubDeviceClient.create_from_connection_string(conn_str)
client.connect()

# Device-side X.509 authentication (AFTER)
from azure.iot.device import IoTHubDeviceClient, X509

x509 = X509(
    cert_file="/certs/device-sensor-floor3-unit47.pem",
    key_file="/certs/device-sensor-floor3-unit47.key",
    pass_phrase="optional-passphrase",
)

client = IoTHubDeviceClient.create_from_x509_certificate(
    hostname="hub-prod.azure-devices.net",
    device_id="sensor-floor3-unit47",
    x509=x509,
)
client.connect()

# Service-side SAS authentication (BEFORE)
from azure.iot.hub import IoTHubRegistryManager

conn_str = os.environ["IOTHUB_CONNECTION_STRING"]
# "HostName=hub-prod.azure-devices.net;
#  SharedAccessKeyName=iothubowner;
#  SharedAccessKey=dGhpcyBpcyBhIGZha2Uga2V5..."

registry = IoTHubRegistryManager(conn_str)
device_twin = registry.get_twin("sensor-floor3-unit47")

# Service-side Managed Identity authentication (AFTER)
from azure.identity import DefaultAzureCredential
from azure.iot.hub import IoTHubRegistryManager

credential = DefaultAzureCredential()
registry = IoTHubRegistryManager.from_token_credential(
    url=f"https://{iot_hub_name}.azure-devices.net",
    token_credential=credential,
)
device_twin = registry.get_twin("sensor-floor3-unit47")

# DPS symmetric key provisioning (BEFORE)
from azure.iot.device import ProvisioningDeviceClient
import hmac, hashlib, base64

# Derive device key from group key
group_key = base64.b64decode(os.environ["DPS_GROUP_KEY"])
device_key = base64.b64encode(
    hmac.new(group_key, device_id.encode(), hashlib.sha256).digest()
).decode()

client = ProvisioningDeviceClient.create_from_symmetric_key(
    provisioning_host="global.azure-devices-provisioning.net",
    registration_id=device_id,
    id_scope=dps_id_scope,
    symmetric_key=device_key,
)
result = client.register()

# DPS X.509 provisioning (AFTER)
from azure.iot.device import ProvisioningDeviceClient, X509

x509 = X509(
    cert_file=f"/certs/{device_id}.pem",
    key_file=f"/certs/{device_id}.key",
)

client = ProvisioningDeviceClient.create_from_x509_certificate(
    provisioning_host="global.azure-devices-provisioning.net",
    registration_id=device_id,
    id_scope=dps_id_scope,
    x509=x509,
)
result = client.register()

# SAS token generation (SDK handles internally)
# Token format:
# SharedAccessSignature sig={signature}&se={expiry}&skn={policyName}&sr={resourceURI}

# SAS connection retry (BEFORE)
def connect_with_retry(conn_str, max_retries=5):
    for attempt in range(max_retries):
        try:
            client = IoTHubDeviceClient.create_from_connection_string(conn_str)
            client.connect()
            return client
        except Exception as e:
            if "401" in str(e):
                # Key may have been rotated -- cannot recover automatically
                log.error("Authentication failed. Key may be rotated.")
                raise  # Permanent failure
            time.sleep(min(2 ** attempt, 60))
    raise ConnectionError("Max retries exceeded")

# X.509 connection retry with certificate awareness (AFTER)
import datetime

def connect_with_cert_awareness(hostname, device_id, cert_path, key_path):
    # Check certificate expiry
    cert = load_certificate(cert_path)
    days_until_expiry = (cert.not_valid_after - datetime.datetime.utcnow()).days

    if days_until_expiry < 30:
        log.warning(f"Certificate expires in {days_until_expiry} days. "
                    "Triggering re-provisioning.")
        reprovision_via_dps(device_id, cert_path, key_path)

    x509 = X509(cert_file=cert_path, key_file=key_path)
    client = IoTHubDeviceClient.create_from_x509_certificate(
        hostname=hostname,
        device_id=device_id,
        x509=x509,
    )

    def on_connection_state_change():
        if not client.connected:
            log.info("Disconnected. Reconnecting with existing certificate.")
            client.connect()  # SDK handles TLS handshake with same cert

    client.on_connection_state_change = on_connection_state_change
    client.connect()
    return client

# Service reads ALL device twins with SAS (BEFORE)
registry = IoTHubRegistryManager(conn_str)
# This single connection string grants access to every device twin
twin_a = registry.get_twin("device-a")
twin_b = registry.get_twin("device-b")
twin_z = registry.get_twin("device-z")
# No differentiation in access scope

# Service reads device twins with Managed Identity (AFTER)
credential = DefaultAzureCredential()
registry = IoTHubRegistryManager.from_token_credential(
    url=f"https://{iot_hub_name}.azure-devices.net",
    token_credential=credential,
)

# Access is governed by the managed identity's RBAC role:
# - IoT Hub Data Reader: read twins only
# - IoT Hub Twin Contributor: read + write twins
# - IoT Hub Data Contributor: full data plane access
twin = registry.get_twin("sensor-floor3-unit47")