Edge device creation, configuration, and device bundles

.gitignore

@@ -289,8 +289,8 @@ __pycache__/
 
 # Virtual environment
 env/
-env27/
-env36/
+env2*/
+env3*/
 .python-version
 
 # PTVS analysis
@@ -324,3 +324,4 @@ src/build
 *.sh
 
 /extensions
+device_bundles

azext_iot/common/certops.py

@@ -9,14 +9,16 @@
 """
 
 import datetime
-from os.path import exists, join
+from os.path import exists
 import base64
-from typing import Dict
+from typing import List, Optional
 from cryptography import x509
 from cryptography.x509.oid import NameOID
 from cryptography.hazmat.primitives import serialization, hashes
 from cryptography.hazmat.primitives.asymmetric import rsa
-from azext_iot.common.shared import SHAHashVersions
+from azext_iot.common.fileops import write_content_to_file
+from azext_iot.common.shared import SHAHashVersions, CertInfo
+from azure.cli.core.azclierror import FileOperationError
 
 
 def create_self_signed_certificate(
@@ -26,7 +28,7 @@ def create_self_signed_certificate(
     cert_only: bool = False,
     file_prefix: str = None,
     sha_version: int = SHAHashVersions.SHA1.value,
-) -> Dict[str, str]:
+) -> CertInfo:
     """
     Function used to create a basic self-signed certificate with no extensions.
 
@@ -85,19 +87,26 @@ def create_self_signed_certificate(
     if cert_output_dir and exists(cert_output_dir):
         cert_file = (file_prefix or subject) + "-cert.pem"
         key_file = (file_prefix or subject) + "-key.pem"
-
-        with open(join(cert_output_dir, cert_file), "wt", encoding="utf-8") as f:
-            f.write(cert_dump)
+        write_content_to_file(
+            content=cert_dump,
+            destination=cert_output_dir,
+            file_name=cert_file,
+            overwrite=True,
+        )
 
         if not cert_only:
-            with open(join(cert_output_dir, key_file), "wt", encoding="utf-8") as f:
-                f.write(key_dump)
+            write_content_to_file(
+                content=key_dump,
+                destination=cert_output_dir,
+                file_name=key_file,
+                overwrite=True,
+            )
 
-    result = {
-        "certificate": cert_dump,
-        "privateKey": key_dump,
-        "thumbprint": thumbprint,
-    }
+    result = CertInfo(
+        certificate=cert_dump,
+        privateKey=key_dump,
+        thumbprint=thumbprint,
+    )
 
     return result
 
@@ -125,3 +134,178 @@ def open_certificate(certificate_path: str) -> str:
         raise ValueError("Certificate file type must be either '.pem' or '.cer'.")
     # Remove trailing white space from the certificate content
     return certificate.rstrip()
+
+
+# TODO - Unit test, compare with test_utils::_generate_root_certificate
+def create_root_certificate(
+    subject: Optional[str] = "Azure_IoT_CLI_Extension_Cert",
+    valid_days: Optional[int] = 365,
+    key_size: Optional[int] = 4096
+) -> CertInfo:
+    key = rsa.generate_private_key(public_exponent=65537, key_size=key_size)
+
+    subject_name = x509.Name(
+        [
+            x509.NameAttribute(NameOID.COMMON_NAME, subject),
+        ]
+    )
+
+    # v3_ca extensions
+    subject_key_id = x509.SubjectKeyIdentifier.from_public_key(key.public_key())
+    authority_key_id = x509.AuthorityKeyIdentifier.from_issuer_subject_key_identifier(
+        subject_key_id
+    )
+    basic = x509.BasicConstraints(ca=True, path_length=None)
+    key_usage = x509.KeyUsage(
+        digital_signature=True,
+        crl_sign=True,
+        key_cert_sign=True,
+        content_commitment=False,
+        data_encipherment=False,
+        decipher_only=False,
+        encipher_only=False,
+        key_agreement=False,
+        key_encipherment=False,
+    )
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject_name)
+        .issuer_name(subject_name)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.datetime.utcnow())
+        .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=valid_days))
+        .add_extension(subject_key_id, critical=False)
+        .add_extension(authority_key_id, critical=False)
+        .add_extension(basic, critical=True)
+        .add_extension(key_usage, critical=True)
+        .sign(key, hashes.SHA256())
+    )
+    certificate = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8").rstrip()
+    thumbprint = cert.fingerprint(hashes.SHA256()).hex().upper()
+    private_key = key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode("utf-8").rstrip()
+
+    return CertInfo(
+        certificate=certificate,
+        privateKey=private_key,
+        thumbprint=thumbprint,
+    )
+
+
+# TODO - Unit test, compare with test_utils::_generate_device_certificate
+def create_signed_cert(
+    subject: str,
+    ca_public: str,
+    ca_private: str,
+    cert_output_dir: Optional[str] = None,
+    cert_file: Optional[str] = None,
+    valid_days: Optional[int] = 365,
+) -> CertInfo:
+
+    private_key = rsa.generate_private_key(public_exponent=65537, key_size=4096)
+    ca_public_key = ca_public.encode("utf-8")
+    ca_private_key = ca_private.encode("utf-8")
+    ca_key = serialization.load_pem_private_key(ca_private_key, password=None)
+    ca_cert = x509.load_pem_x509_certificate(ca_public_key)
+
+    # v3 certificate extensions
+    subject_key_id = x509.SubjectKeyIdentifier.from_public_key(private_key.public_key())
+    authority_key_id = x509.AuthorityKeyIdentifier.from_issuer_public_key(
+        ca_cert.public_key()
+    )
+    basic_constraints = x509.BasicConstraints(ca=True, path_length=None)
+    key_usage = x509.KeyUsage(
+        digital_signature=True,
+        crl_sign=True,
+        key_cert_sign=True,
+        content_commitment=False,
+        data_encipherment=False,
+        decipher_only=False,
+        encipher_only=False,
+        key_agreement=False,
+        key_encipherment=False,
+    )
+    csr = (
+        x509.CertificateSigningRequestBuilder()
+        .subject_name(
+            x509.Name(
+                [
+                    x509.NameAttribute(NameOID.COMMON_NAME, subject),
+                ]
+            )
+        )
+        .sign(private_key, hashes.SHA256())
+    )
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(csr.subject)
+        .issuer_name(ca_cert.subject)
+        .public_key(csr.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.datetime.utcnow())
+        .not_valid_after(datetime.datetime.utcnow() + datetime.timedelta(days=valid_days))
+        .add_extension(subject_key_id, False)
+        .add_extension(authority_key_id, False)
+        .add_extension(basic_constraints, True)
+        .add_extension(key_usage, True)
+        .sign(ca_key, hashes.SHA256())
+    )
+    certificate = cert.public_bytes(serialization.Encoding.PEM).decode("utf-8")
+    thumbprint = cert.fingerprint(hashes.SHA256()).hex().upper()
+    privateKey = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.TraditionalOpenSSL,
+        encryption_algorithm=serialization.NoEncryption(),
+    ).decode("utf-8")
+
+    if cert_output_dir and exists(cert_output_dir):
+        write_content_to_file(
+            content=certificate,
+            destination=cert_output_dir,
+            file_name=f"{cert_file or subject}.cert.pem",
+        )
+        write_content_to_file(
+            content=privateKey,
+            destination=cert_output_dir,
+            file_name=f"{cert_file or subject}.key.pem",
+        )
+    return CertInfo(
+        certificate=certificate, thumbprint=thumbprint, privateKey=privateKey
+    )
+
+
+def load_ca_cert_info(cert_path: str, key_path: str) -> CertInfo:
+    for path in [cert_path, key_path]:
+        if not exists(path):
+            raise FileOperationError(
+                "Error loading certificates. " f"No file found at path '{path}'"
+            )
+    key = open_certificate(key_path)
+    cert = open_certificate(cert_path).encode("utf-8")
+    certificate = x509.load_pem_x509_certificate(cert)
+    thumbprint = certificate.fingerprint(hashes.SHA256()).hex().upper()
+    return CertInfo(
+        certificate=certificate.public_bytes(serialization.Encoding.PEM).decode("utf-8").rstrip(),
+        thumbprint=thumbprint,
+        privateKey=key,
+    )
+
+
+def make_cert_chain(
+    certs: List[str],
+    output_dir: Optional[str] = None,
+    output_file: Optional[str] = "cert-chain.pem",
+):
+    cert_content = "".join(certs)
+    if output_dir and exists(output_dir) and len(certs):
+        write_content_to_file(
+            content=cert_content,
+            destination=output_dir,
+            file_name=output_file,
+            overwrite=True,
+        )
+    return cert_content

azext_iot/common/fileops.py

@@ -0,0 +1,47 @@
+from os import makedirs, remove, listdir
+from os.path import exists, join
+from pathlib import PurePath
+from typing import Optional, Union
+from azure.cli.core.azclierror import FileOperationError
+
+
+"""
+fileops: Functions for working with files.
+"""
+
+
+def write_content_to_file(
+    content: Union[str, bytes],
+    destination: str,
+    file_name: str,
+    overwrite: Optional[bool] = False,
+):
+    dest_path = PurePath(destination)
+    file_path = dest_path.joinpath(file_name)
+
+    if exists(file_path) and not overwrite:
+        raise FileOperationError(f"File already exists at path: {file_path}")
+    if overwrite and destination:
+        makedirs(destination, exist_ok=True)
+    write_content = bytes(content, "utf-8") if isinstance(content, str) else content
+    with open(file_path, "wb") as f:
+        f.write(write_content)
+
+
+def tar_directory(
+    target_directory: str,
+    tarfile_path: str,
+    tarfile_name: str,
+    overwrite: Optional[bool] = False,
+):
+    full_path = join(tarfile_path, f"{tarfile_name}.tgz")
+    if exists(full_path):
+        if not overwrite:
+            raise FileOperationError(f"File {full_path} already exists")
+        remove(full_path)
+    if not exists(tarfile_path):
+        makedirs(tarfile_path, exist_ok=overwrite)
+    import tarfile
+    with tarfile.open(full_path, "w:gz") as tar:
+        for file_name in listdir(target_directory):
+            tar.add(join(target_directory, file_name), file_name)

azext_iot/common/shared.py

@@ -10,6 +10,8 @@
 """
 
 from enum import Enum
+from typing import NamedTuple, Optional, Any, TypedDict, List
+from azext_iot.sdk.iothub.service.models import ConfigurationContent
 
 
 class SdkType(Enum):
@@ -263,6 +265,7 @@ class IoTDPSStateType(Enum):
     """
     IoT Hub Device Provisioning Service State Property
     """
+
     Activating = "Activating"
     ActivationFailed = "ActivationFailed"
     Active = "Active"
@@ -279,12 +282,13 @@ class IoTDPSStateType(Enum):
 
 class ConnectionStringParser(Enum):
     """
-        All connection string parser with respective functions
+    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
+        parse_iot_hub_connection_string,
     )
 
     Module = parse_iot_device_module_connection_string
@@ -296,6 +300,7 @@ class DiscoveryResourceType(Enum):
     """
     Resource types supported by discovery.
     """
+
     IoTHub = "IoT Hub"
     DPS = "IoT Hub Device Provisioning Service"
 
@@ -304,5 +309,43 @@ class SHAHashVersions(Enum):
     """
     Supported SHA types for generating the certificate thumbprint.
     """
+
     SHA1 = 1
     SHA256 = 256
+
+
+class CertInfo(TypedDict):
+    """
+    Typed Dict for certificate, thumbprint, and private key return type
+    """
+
+    certificate: str
+    privateKey: str
+    thumbprint: str
+
+
+# Utility classes for nested edge config file values and device arguments
+class EdgeContainerAuth(NamedTuple):
+    serveraddress: Optional[str] = None
+    username: Optional[str] = None
+    password: Optional[str] = None
+
+
+class NestedEdgeDeviceConfig(NamedTuple):
+    device_id: str
+    deployment: Optional[ConfigurationContent] = None
+    config: Optional[Any] = None
+    parent_id: Optional[str] = None
+    hostname: Optional[str] = None
+    parent_hostname: Optional[str] = None
+    edge_agent: Optional[str] = None
+    container_auth: Optional[EdgeContainerAuth] = None
+
+
+class NestedEdgeConfig(NamedTuple):
+    version: str
+    auth_method: DeviceAuthType
+    root_cert: CertInfo
+    devices: List[NestedEdgeDeviceConfig]
+    template_config_path: Optional[str] = None
+    default_edge_agent: Optional[str] = None