azext_iot/central/providers/enrollment_group_provider.py

# 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 typing import List
from knack.log import get_logger

from azext_iot.constants import CENTRAL_ENDPOINT
from azext_iot.central import services as central_services


logger = get_logger(__name__)


class CentralEnrollmentGroupProvider:
    def __init__(self, cmd, app_id: str, api_version: str, token=None):
        """
        Provider for enrollment group APIs

        Args:
            cmd: command passed into az
            app_id: name of app (used for forming request URL)
            api_version: API version (appendend to request URL)
            token: (OPTIONAL) authorization token to fetch API token details from IoTC.
                MUST INCLUDE type (e.g. 'SharedAccessToken ...', 'Bearer ...')
                Useful in scenarios where user doesn't own the app
                therefore AAD token won't work, but a SAS token generated by owner will
        """
        self._cmd = cmd
        self._app_id = app_id
        self._token = token
        self._api_version = api_version

    def get_enrollment_group(
        self,
        group_id: str,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> dict:
        return central_services.enrollment_group.get_enrollment_group(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            central_dns_suffix=central_dns_suffix,
        )

    def list_enrollment_groups(
        self,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> List[dict]:
        return central_services.enrollment_group.list_enrollment_groups(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            central_dns_suffix=central_dns_suffix,
        )

    def create_enrollment_group(
        self,
        group_id: str,
        attestation: str,
        display_name: str,
        type: str,
        primary_key: str = None,
        secondary_key: str = None,
        enabled: bool = True,
        etag : str = None,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ):
        return central_services.enrollment_group.create_enrollment_group(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            attestation=attestation,
            primary_key=primary_key,
            secondary_key=secondary_key,
            display_name=display_name,
            type=type,
            enabled=enabled,
            etag=etag,
            central_dns_suffix=central_dns_suffix,
        )

    def update_enrollment_group(
        self,
        group_id: str,
        display_name: str = None,
        type: str = None,
        enabled: bool = True,
        etag : str = None,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ):
        return central_services.enrollment_group.update_enrollment_group(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            display_name=display_name,
            type=type,
            enabled=enabled,
            etag=etag,
            central_dns_suffix=central_dns_suffix,
        )

    def delete_enrollment_group(
        self,
        group_id: str,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> dict:
        return central_services.enrollment_group.delete_enrollment_group(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            central_dns_suffix=central_dns_suffix,
        )

    def create_x509(
        self,
        group_id: str,
        primary_cert: str = None,
        secondary_cert: str = None,
        etag: str = None,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> dict:
        return central_services.enrollment_group.create_x509(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            primary_cert=primary_cert,
            secondary_cert=secondary_cert,
            etag=etag,
            central_dns_suffix=central_dns_suffix,
        )

    def get_x509(
        self,
        group_id: str,
        certificate_entry: str = None,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> dict:
        return central_services.enrollment_group.get_x509(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            certificate_entry=certificate_entry,
            central_dns_suffix=central_dns_suffix,
        )

    def delete_x509(
        self,
        group_id: str,
        certificate_entry: str = None,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> dict:
        return central_services.enrollment_group.delete_x509(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            certificate_entry=certificate_entry,
            central_dns_suffix=central_dns_suffix,
        )

    def verify_x509(
        self,
        group_id: str,
        primary_cert: str = None,
        secondary_cert: str = None,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> dict:
        return central_services.enrollment_group.verify_x509(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            primary_cert=primary_cert,
            secondary_cert=secondary_cert,
            central_dns_suffix=central_dns_suffix,
        )

    def generate_verification_code(
        self,
        group_id: str,
        certificate_entry: str = None,
        central_dns_suffix=CENTRAL_ENDPOINT,
    ) -> dict:
        return central_services.enrollment_group.generate_verification_code(
            cmd=self._cmd,
            app_id=self._app_id,
            token=self._token,
            api_version=self._api_version,
            group_id=group_id,
            certificate_entry=certificate_entry,
            central_dns_suffix=central_dns_suffix,
        )