X.509 Certificate Authentication Migration¶

import datetime
from pathlib import Path

def check_and_renew_certificate(device_id, cert_path, key_path, dps_scope):
    """Check certificate expiry and renew if within 30 days."""
    cert = load_x509_certificate(cert_path)
    days_remaining = (cert.not_valid_after - datetime.datetime.utcnow()).days

    if days_remaining <= 30:
        log.info(f"Certificate expires in {days_remaining} days. Renewing.")

        # Request new certificate from CA/EST server
        new_cert, new_key = request_certificate_from_est(
            est_server="https://est.csa-iot.internal",
            device_id=device_id,
            existing_cert=cert_path,
            existing_key=key_path,
        )

        # Save new certificate (keep old as backup)
        backup_path = Path(cert_path).with_suffix(".old.pem")
        Path(cert_path).rename(backup_path)
        save_certificate(new_cert, cert_path)
        save_private_key(new_key, key_path)

        # Re-provision through DPS with new cert
        reprovision_device(device_id, cert_path, key_path, dps_scope)

        log.info("Certificate renewed and device re-provisioned.")
        return True

    log.info(f"Certificate valid for {days_remaining} more days.")
    return False

# Device provisioning with TPM-stored X.509 certificate
from azure.iot.device import ProvisioningDeviceClient, X509
from tpm2_pytss import ESAPI

def provision_with_tpm(device_id, dps_host, id_scope):
    """Provision device using X.509 certificate stored in TPM."""
    # Certificate is stored in TPM NV index
    # Private key operations happen inside TPM
    with ESAPI() as ectx:
        # Read certificate from TPM NV storage
        cert_pem = read_cert_from_tpm_nv(ectx, nv_index=0x01C00002)

        # Create X509 object pointing to TPM-backed key
        # The SDK will use the TPM for TLS handshake signing
        x509 = X509(
            cert_file=cert_pem,  # PEM certificate
            key_file=None,        # Key is in TPM, not filesystem
            pass_phrase=None,
        )

        # For TPM-backed keys, use the custom HSM interface
        client = ProvisioningDeviceClient.create_from_x509_certificate(
            provisioning_host=dps_host,
            registration_id=device_id,
            id_scope=id_scope,
            x509=x509,
        )

        result = client.register()
        return result.registration_state

# deploy-certs.py
# Deploy certificates to devices via secure channel (e.g., SSH, Azure IoT Edge)

import paramiko
import os
from concurrent.futures import ThreadPoolExecutor

CERT_DIR = "./fleet-certs"
DEVICES_FILE = "device-inventory.csv"  # device_id,ip_address,ssh_user

def deploy_cert_to_device(device_id, ip_address, ssh_user):
    """Deploy certificate and key to a single device."""
    cert_file = os.path.join(CERT_DIR, f"{device_id}.pem")
    key_file = os.path.join(CERT_DIR, f"{device_id}.key")

    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(ip_address, username=ssh_user, key_filename="~/.ssh/deploy_key")

    sftp = ssh.open_sftp()
    sftp.put(cert_file, f"/etc/iot-certs/{device_id}.pem")
    sftp.put(key_file, f"/etc/iot-certs/{device_id}.key")

    # Set restrictive permissions
    ssh.exec_command(f"chmod 644 /etc/iot-certs/{device_id}.pem")
    ssh.exec_command(f"chmod 600 /etc/iot-certs/{device_id}.key")

    # Restart IoT agent to use new certificate
    ssh.exec_command("systemctl restart iot-device-agent")

    sftp.close()
    ssh.close()
    print(f"Deployed certificate to {device_id} ({ip_address})")

# Deploy in parallel (max 50 concurrent connections)
with ThreadPoolExecutor(max_workers=50) as executor:
    with open(DEVICES_FILE) as f:
        for line in f:
            device_id, ip_address, ssh_user = line.strip().split(",")
            executor.submit(deploy_cert_to_device, device_id, ip_address, ssh_user)

"""
IoT device client with X.509 authentication.
Replaces SAS-based device client for CSA-0025 compliance.
"""

import asyncio
import datetime
import json
import logging
from pathlib import Path

from azure.iot.device.aio import IoTHubDeviceClient, ProvisioningDeviceClient
from azure.iot.device import X509, Message

log = logging.getLogger(__name__)

# Configuration
DPS_HOST = "global.azure-devices-provisioning.net"
DPS_ID_SCOPE = "0ne00XXXXXX"
DEVICE_ID = "sensor-floor3-unit47"
CERT_PATH = "/etc/iot-certs/sensor-floor3-unit47.pem"
KEY_PATH = "/etc/iot-certs/sensor-floor3-unit47.key"
CERT_RENEWAL_DAYS = 30


async def provision_device():
    """Provision device through DPS using X.509 certificate."""
    x509 = X509(cert_file=CERT_PATH, key_file=KEY_PATH)

    provisioning_client = ProvisioningDeviceClient.create_from_x509_certificate(
        provisioning_host=DPS_HOST,
        registration_id=DEVICE_ID,
        id_scope=DPS_ID_SCOPE,
        x509=x509,
    )

    result = await provisioning_client.register()
    log.info(f"Provisioned to hub: {result.registration_state.assigned_hub}")
    return result.registration_state


async def connect_and_send(hub_hostname):
    """Connect to IoT Hub and send telemetry using X.509 certificate."""
    x509 = X509(cert_file=CERT_PATH, key_file=KEY_PATH)

    client = IoTHubDeviceClient.create_from_x509_certificate(
        hostname=hub_hostname,
        device_id=DEVICE_ID,
        x509=x509,
    )

    await client.connect()
    log.info("Connected to IoT Hub with X.509 certificate.")

    try:
        while True:
            # Check certificate expiry
            cert_expiry = get_cert_expiry(CERT_PATH)
            days_remaining = (cert_expiry - datetime.datetime.utcnow()).days

            if days_remaining <= CERT_RENEWAL_DAYS:
                log.warning(f"Certificate expires in {days_remaining} days.")

            # Send telemetry
            telemetry = {
                "temperature": 22.5,
                "humidity": 45.2,
                "timestamp": datetime.datetime.utcnow().isoformat(),
                "certDaysRemaining": days_remaining,
            }
            message = Message(json.dumps(telemetry))
            message.content_type = "application/json"
            message.content_encoding = "utf-8"
            await client.send_message(message)

            await asyncio.sleep(60)
    finally:
        await client.disconnect()


def get_cert_expiry(cert_path):
    """Read certificate expiry date."""
    from cryptography import x509 as crypto_x509
    cert_data = Path(cert_path).read_bytes()
    cert = crypto_x509.load_pem_x509_certificate(cert_data)
    return cert.not_valid_after


async def main():
    registration = await provision_device()
    await connect_and_send(registration.assigned_hub)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())