# coding=utf-8 # ---------------------------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License file in the project root for license information. # ---------------------------------------------------------------------------------------------- from typing import Any, Dict, List from .base import ( CheckManager, add_display_and_eval, check_post_deployment, generate_target_resource_name, get_resources_by_name, process_list_resource, process_resource_properties, get_resources_grouped_by_namespace, ) from rich.padding import Padding from ...common import CheckTaskStatus from .common import ( ASSET_DATAPOINT_PROPERTIES, ASSET_EVENT_PROPERTIES, ASSET_PROPERTIES, MAX_ASSET_DATAPOINTS, MAX_ASSET_EVENTS, PADDING_SIZE, ResourceOutputDetailLevel, ) from ..edge_api import ( DEVICEREGISTRY_API_V1, DeviceRegistryResourceKinds, ) def check_deviceregistry_deployment( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_kinds: List[str] = None, resource_name: str = None, ) -> List[dict]: evaluate_funcs = { DeviceRegistryResourceKinds.ASSET: evaluate_assets, DeviceRegistryResourceKinds.ASSETENDPOINTPROFILE: evaluate_asset_endpoint_profiles, } return check_post_deployment( api_info=DEVICEREGISTRY_API_V1, check_name="enumerateDeviceRegistryApi", check_desc="Enumerate Device Registry API resources", resource_name=resource_name, evaluate_funcs=evaluate_funcs, as_list=as_list, detail_level=detail_level, resource_kinds=resource_kinds, ) def evaluate_assets( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_name: str = None, ) -> Dict[str, Any]: check_manager = CheckManager(check_name="evalAssets", check_desc="Evaluate Assets") asset_namespace_conditions = ["spec.assetEndpointProfileRef"] target_assets = generate_target_resource_name(api_info=DEVICEREGISTRY_API_V1, resource_kind=DeviceRegistryResourceKinds.ASSET.value) all_assets = get_resources_by_name( api_info=DEVICEREGISTRY_API_V1, kind=DeviceRegistryResourceKinds.ASSET, resource_name=resource_name ) if not all_assets: fetch_assets_warning_text = "Unable to fetch assets in any namespaces." check_manager.add_target(target_name=target_assets) check_manager.add_display(target_name=target_assets, display=Padding(fetch_assets_warning_text, (0, 0, 0, 8))) check_manager.add_target_eval( target_name=target_assets, status=CheckTaskStatus.skipped.value, value=fetch_assets_warning_text ) return check_manager.as_dict(as_list) for (namespace, assets) in get_resources_grouped_by_namespace(all_assets): check_manager.add_target(target_name=target_assets, namespace=namespace, conditions=asset_namespace_conditions) check_manager.add_display( target_name=target_assets, namespace=namespace, display=Padding( f"Device Registry assets in namespace {{[purple]{namespace}[/purple]}}", (0, 0, 0, 8) ) ) assets: List[dict] = list(assets) added_datapoint_conditions = False added_event_conditions = False added_status_conditions = False for asset in assets: padding = 10 asset_name = asset["metadata"]["name"] asset_status_text = ( f"- Asset {{[bright_blue]{asset_name}[/bright_blue]}} detected." ) check_manager.add_display( target_name=target_assets, namespace=namespace, display=Padding(asset_status_text, (0, 0, 0, padding)), ) asset_spec = asset["spec"] endpoint_profile_uri = asset_spec.get("assetEndpointProfileRef", "") endpoint_profile = get_resources_by_name( api_info=DEVICEREGISTRY_API_V1, kind=DeviceRegistryResourceKinds.ASSETENDPOINTPROFILE, resource_name=endpoint_profile_uri ) spec_padding = padding + PADDING_SIZE endpoint_profile_uri_value = {"spec.assetEndpointProfileRef": endpoint_profile_uri} endpoint_profile_uri_status = CheckTaskStatus.success.value if endpoint_profile: endpoint_profile_uri_text = ( f"Asset endpoint profile {{[bright_blue]{endpoint_profile_uri}[/bright_blue]}} property [green]detected[/green]." ) else: endpoint_profile_uri_text = ( f"Asset endpoint profile {{[bright_blue]{endpoint_profile_uri}[/bright_blue]}} [red]not detected[/red]." ) endpoint_profile_uri_status = CheckTaskStatus.error.value add_display_and_eval( check_manager=check_manager, target_name=target_assets, display_text=endpoint_profile_uri_text, eval_status=endpoint_profile_uri_status, eval_value=endpoint_profile_uri_value, resource_name=asset_name, namespace=namespace, padding=(0, 0, 0, spec_padding) ) # data points # all should be under one dataset dataset = asset_spec.get("datasets", []) data_points = [] if dataset: data_points = dataset[0].get("dataPoints", []) if data_points: if not added_datapoint_conditions: check_manager.add_target_conditions( target_name=target_assets, namespace=namespace, conditions=[ "len(spec.datasets[0].dataPoints)", "spec.datasets[0].dataPoints.dataSource" ] ) added_datapoint_conditions = True data_points_count = len(data_points) data_points_value = {"len(spec.datasets[0].dataPoints)": data_points_count} data_points_status = CheckTaskStatus.success.value if data_points_count > MAX_ASSET_DATAPOINTS: data_points_text = ( f"Data points [red]exceeding {MAX_ASSET_DATAPOINTS}[/red]. Detected {data_points_count}." ) else: data_points_text = ( f"[bright_blue]{data_points_count}[/bright_blue] data points detected." ) add_display_and_eval( check_manager=check_manager, target_name=target_assets, display_text=data_points_text, eval_status=data_points_status, eval_value=data_points_value, resource_name=asset_name, namespace=namespace, padding=(0, 0, 0, spec_padding) ) for index, data_point in enumerate(data_points): data_point_data_source = data_point.get("dataSource", "") datapoint_padding = spec_padding + PADDING_SIZE data_point_data_source_value = {f"spec.datasets[0].dataPoints.[{index}].dataSource": data_point_data_source} data_point_data_source_status = CheckTaskStatus.success.value if data_point_data_source: data_point_data_source_text = ( f"- Data source: {{[bright_blue]{data_point_data_source}[/bright_blue]}} [green]detected[/green]." ) else: data_point_data_source_text = ( "Data source [red]not detected[/red]." ) data_point_data_source_status = CheckTaskStatus.error.value add_display_and_eval( check_manager=check_manager, target_name=target_assets, display_text=data_point_data_source_text, eval_status=data_point_data_source_status, eval_value=data_point_data_source_value, resource_name=asset_name, namespace=namespace, padding=(0, 0, 0, datapoint_padding) ) if detail_level > ResourceOutputDetailLevel.summary.value: process_resource_properties( check_manager=check_manager, detail_level=detail_level, target_name=target_assets, prop_value=data_point, properties=ASSET_DATAPOINT_PROPERTIES, namespace=namespace, padding=(0, 0, 0, datapoint_padding + PADDING_SIZE) ) if detail_level > ResourceOutputDetailLevel.summary.value: process_resource_properties( check_manager=check_manager, detail_level=detail_level, target_name=target_assets, prop_value=asset_spec, properties=ASSET_PROPERTIES, namespace=namespace, padding=(0, 0, 0, spec_padding) ) # events events = asset_spec.get("events", []) if events: if not added_event_conditions: check_manager.add_target_conditions( target_name=target_assets, namespace=namespace, conditions=[ "len(spec.events)", "spec.events.eventNotifier" ] ) added_event_conditions = True events_count = len(events) events_count_value = {"len(spec.events)": events_count} events_count_status = CheckTaskStatus.success.value if events_count > MAX_ASSET_EVENTS: events_count_text = ( f"Events [red]exceeding {MAX_ASSET_EVENTS}[/red]. Detected {events_count}." ) events_count_status = CheckTaskStatus.error.value else: events_count_text = ( f"[bright_blue]{events_count}[/bright_blue] events detected." ) add_display_and_eval( check_manager=check_manager, target_name=target_assets, display_text=events_count_text, eval_status=events_count_status, eval_value=events_count_value, resource_name=asset_name, namespace=namespace, padding=(0, 0, 0, spec_padding) ) for index, event in enumerate(events): event_notifier = event.get("eventNotifier", "") event_padding = spec_padding + PADDING_SIZE event_notifier_value = {f"spec.events.[{index}].eventNotifier": event_notifier} event_notifier_status = CheckTaskStatus.success.value if event_notifier: event_notifier_text = ( f"- Event notifier: {{[bright_blue]{event_notifier}[/bright_blue]}} [green]detected[/green]." ) else: event_notifier_text = ( "Event notifier [red]not detected[/red]." ) event_notifier_status = CheckTaskStatus.error.value add_display_and_eval( check_manager=check_manager, target_name=target_assets, display_text=event_notifier_text, eval_status=event_notifier_status, eval_value=event_notifier_value, resource_name=asset_name, namespace=namespace, padding=(0, 0, 0, event_padding) ) if detail_level > ResourceOutputDetailLevel.summary.value: process_resource_properties( check_manager=check_manager, detail_level=detail_level, target_name=target_assets, prop_value=event, properties=ASSET_EVENT_PROPERTIES, namespace=namespace, padding=(0, 0, 0, event_padding + PADDING_SIZE) ) # status status = asset_spec.get("status", "") if status: _process_asset_status( check_manager=check_manager, added_status_conditions=added_status_conditions, target_assets=target_assets, asset_name=asset_name, status=status, padding=spec_padding, namespace=namespace ) return check_manager.as_dict(as_list) def evaluate_asset_endpoint_profiles( as_list: bool = False, detail_level: int = ResourceOutputDetailLevel.summary.value, resource_name: str = None, ) -> Dict[str, Any]: check_manager = CheckManager(check_name="evalAssetEndpointProfiles", check_desc="Evaluate Asset Endpoint Profiles") endpoint_namespace_conditions = ["spec.uuid"] all_asset_endpoint_profiles = get_resources_by_name( api_info=DEVICEREGISTRY_API_V1, kind=DeviceRegistryResourceKinds.ASSETENDPOINTPROFILE, resource_name=resource_name ) target_asset_endpoint_profiles = generate_target_resource_name(api_info=DEVICEREGISTRY_API_V1, resource_kind=DeviceRegistryResourceKinds.ASSETENDPOINTPROFILE.value) if not all_asset_endpoint_profiles: fetch_asset_endpoint_profiles_warning_text = "Unable to fetch asset endpoint profiles in any namespaces." check_manager.add_target(target_name=target_asset_endpoint_profiles) check_manager.add_display(target_name=target_asset_endpoint_profiles, display=Padding(fetch_asset_endpoint_profiles_warning_text, (0, 0, 0, 8))) check_manager.add_target_eval( target_name=target_asset_endpoint_profiles, status=CheckTaskStatus.skipped.value, value=fetch_asset_endpoint_profiles_warning_text ) return check_manager.as_dict(as_list) for (namespace, asset_endpoint_profiles) in get_resources_grouped_by_namespace(all_asset_endpoint_profiles): check_manager.add_target(target_name=target_asset_endpoint_profiles, namespace=namespace, conditions=endpoint_namespace_conditions) check_manager.add_display( target_name=target_asset_endpoint_profiles, namespace=namespace, display=Padding( f"Asset Endpoint Profiles in namespace {{[purple]{namespace}[/purple]}}", (0, 0, 0, 8) ) ) asset_endpoint_profiles: List[dict] = list(asset_endpoint_profiles) added_transport_authentication_conditions = False added_user_authentication_conditions = False for asset_endpoint_profile in asset_endpoint_profiles: asset_endpoint_profile_name = asset_endpoint_profile["metadata"]["name"] padding = 10 asset_endpoint_profile_status_text = ( f"- Profile {{[bright_blue]{asset_endpoint_profile_name}[/bright_blue]}} detected." ) check_manager.add_display( target_name=target_asset_endpoint_profiles, namespace=namespace, display=Padding(asset_endpoint_profile_status_text, (0, 0, 0, padding)), ) spec_padding = padding + PADDING_SIZE asset_endpoint_profile_spec = asset_endpoint_profile["spec"] endpoint_profile_uuid = asset_endpoint_profile_spec.get("uuid", "") endpoint_profile_uuid_value = {"spec.uuid": endpoint_profile_uuid} endpoint_profile_uuid_status = CheckTaskStatus.success.value if endpoint_profile_uuid: endpoint_profile_uuid_text = ( f"Uuid: {{[bright_blue]{endpoint_profile_uuid}[/bright_blue]}} [green]detected[/green]." ) else: endpoint_profile_uuid_text = ( "Uuid [red]not detected[/red]." ) endpoint_profile_uuid_status = CheckTaskStatus.error.value add_display_and_eval( check_manager=check_manager, target_name=target_asset_endpoint_profiles, display_text=endpoint_profile_uuid_text, eval_status=endpoint_profile_uuid_status, eval_value=endpoint_profile_uuid_value, resource_name=asset_endpoint_profile_name, namespace=namespace, padding=(0, 0, 0, spec_padding) ) # transportAuthentication transport_authentication = asset_endpoint_profile_spec.get("transportAuthentication", {}) if transport_authentication: if not added_transport_authentication_conditions: check_manager.add_target_conditions( target_name=target_asset_endpoint_profiles, namespace=namespace, conditions=[ "spec.transportAuthentication.ownCertificates" ] ) added_transport_authentication_conditions = True check_manager.add_display( target_name=target_asset_endpoint_profiles, namespace=namespace, display=Padding( "Transport authentication:", (0, 0, 0, spec_padding) ) ) transport_authentication_own_certificates = transport_authentication.get("ownCertificates", None) transport_authentication_own_certificates_value = {"spec.transportAuthentication.ownCertificates": transport_authentication_own_certificates} transport_authentication_own_certificates_status = CheckTaskStatus.success.value transport_authentication_padding = spec_padding + PADDING_SIZE if transport_authentication_own_certificates is None: transport_authentication_own_certificates_text = ( "Own certificates [red]not detected[/red]." ) transport_authentication_own_certificates_status = CheckTaskStatus.error.value else: transport_authentication_own_certificates_text = ( f"Own certificates: {len(transport_authentication_own_certificates)} detected." ) add_display_and_eval( check_manager=check_manager, target_name=target_asset_endpoint_profiles, display_text=transport_authentication_own_certificates_text, eval_status=transport_authentication_own_certificates_status, eval_value=transport_authentication_own_certificates_value, resource_name=asset_endpoint_profile_name, namespace=namespace, padding=(0, 0, 0, transport_authentication_padding) ) if detail_level > ResourceOutputDetailLevel.detail.value and transport_authentication_own_certificates: process_list_resource( check_manager=check_manager, target_name=target_asset_endpoint_profiles, resource=transport_authentication_own_certificates, namespace=namespace, padding=transport_authentication_padding + PADDING_SIZE ) # userAuthentication user_authentication = asset_endpoint_profile_spec.get("userAuthentication", {}) if user_authentication: if not added_user_authentication_conditions: check_manager.add_target_conditions( target_name=target_asset_endpoint_profiles, namespace=namespace, conditions=[ "spec.userAuthentication.mode", "spec.userAuthentication.x509Credentials.certificateReference", "spec.userAuthentication.usernamePasswordCredentials.usernameReference", "spec.userAuthentication.usernamePasswordCredentials.passwordReference" ] ) added_user_authentication_conditions = True check_manager.add_display( target_name=target_asset_endpoint_profiles, namespace=namespace, display=Padding( "User authentication:", (0, 0, 0, spec_padding) ) ) # check required mode user_authentication_mode = user_authentication.get("mode", "") user_authentication_mode_value = {"spec.userAuthentication.mode": user_authentication_mode} user_authentication_mode_status = CheckTaskStatus.success.value user_authentication_padding = spec_padding + PADDING_SIZE if user_authentication_mode: user_authentication_mode_text = ( f"User authentication mode: {{[bright_blue]{user_authentication_mode}[/bright_blue]}} [green]detected[/green]." ) else: user_authentication_mode_text = ( "User authentication mode [red]not detected[/red]." ) user_authentication_mode_status = CheckTaskStatus.error.value add_display_and_eval( check_manager=check_manager, target_name=target_asset_endpoint_profiles, display_text=user_authentication_mode_text, eval_status=user_authentication_mode_status, eval_value=user_authentication_mode_value, resource_name=asset_endpoint_profile_name, namespace=namespace, padding=(0, 0, 0, user_authentication_padding) ) if user_authentication_mode == "Certificate": # check x509Credentials user_authentication_x509_credentials = user_authentication.get("x509Credentials", {}) certificate_reference = user_authentication_x509_credentials.get("certificateReference", "") user_authentication_x509_credentials_value = {"spec.userAuthentication.x509Credentials.certificateReference": certificate_reference} user_authentication_x509_credentials_status = CheckTaskStatus.success.value if certificate_reference: user_authentication_x509_credentials_text = ( f"Certificate reference: {{[bright_blue]{certificate_reference}[/bright_blue]}} [green]detected[/green]." ) else: user_authentication_x509_credentials_text = ( "Certificate reference [red]not detected[/red]." ) user_authentication_x509_credentials_status = CheckTaskStatus.error.value add_display_and_eval( check_manager=check_manager, target_name=target_asset_endpoint_profiles, display_text=user_authentication_x509_credentials_text, eval_status=user_authentication_x509_credentials_status, eval_value=user_authentication_x509_credentials_value, resource_name=asset_endpoint_profile_name, namespace=namespace, padding=(0, 0, 0, user_authentication_padding + PADDING_SIZE) ) elif user_authentication_mode == "UsernamePassword": # check usernamePasswordCredentials user_authentication_username_password_credentials = user_authentication.get("usernamePasswordCredentials", {}) username_reference = user_authentication_username_password_credentials.get("usernameReference", "") password_reference = user_authentication_username_password_credentials.get("passwordReference", "") user_authentication_username_password_credentials_value = { "spec.userAuthentication.usernamePasswordCredentials.usernameReference": username_reference, "spec.userAuthentication.usernamePasswordCredentials.passwordReference": password_reference } user_authentication_username_password_credentials_status = CheckTaskStatus.success.value if username_reference and password_reference: user_authentication_username_password_credentials_text = ( f"Username reference: {{[bright_blue]{username_reference}[/bright_blue]}} [green]detected[/green].\n" f"Password reference: {{[bright_blue]{password_reference}[/bright_blue]}} [green]detected[/green]." ) else: user_authentication_username_password_credentials_text = ( "Username reference or password reference [red]not detected[/red]." ) user_authentication_username_password_credentials_status = CheckTaskStatus.error.value add_display_and_eval( check_manager=check_manager, target_name=target_asset_endpoint_profiles, display_text=user_authentication_username_password_credentials_text, eval_status=user_authentication_username_password_credentials_status, eval_value=user_authentication_username_password_credentials_value, resource_name=asset_endpoint_profile_name, namespace=namespace, padding=(0, 0, 0, user_authentication_padding + PADDING_SIZE) ) if detail_level > ResourceOutputDetailLevel.summary.value: process_resource_properties( check_manager=check_manager, detail_level=detail_level, target_name=target_asset_endpoint_profiles, prop_value=asset_endpoint_profile_spec, properties=[ ("additionalConfiguration", "Additional configuration", True), ("targetAddress", "Target address", False), ], namespace=namespace, padding=(0, 0, 0, spec_padding) ) return check_manager.as_dict(as_list) def _process_asset_status( check_manager: CheckManager, added_status_conditions: bool, target_assets: str, asset_name: str, status: dict, padding: int, namespace: str, ): if not added_status_conditions: check_manager.add_target_conditions( target_name=target_assets, namespace=namespace, conditions=["spec.status"] ) added_status_conditions = True status_value = {"spec.status": str(status)} status_status = CheckTaskStatus.success.value errors = status.get("errors", []) if errors: status_text = ( "Asset status [red]error[/red]." ) for error in errors: error_code = error.get("code", "") message = error.get("message", "") error_text = ( f"- Asset status error code: [red]{error_code}[/red]. Message: {message}" ) check_manager.add_display( target_name=target_assets, namespace=namespace, display=Padding(error_text, (0, 0, 0, padding + PADDING_SIZE)), ) status_status = CheckTaskStatus.error.value else: status_text = ( "Asset status [green]OK[/green]." ) add_display_and_eval( check_manager=check_manager, target_name=target_assets, display_text=status_text, eval_status=status_status, eval_value=status_value, resource_name=asset_name, namespace=namespace, padding=(0, 0, 0, padding) )