validate_azure_dladmin_identity_14.py 1.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
  1. def _azure_dladmin_backup_storage_data_actions_check(
  2. config: Dict[str, Any],
  3. auth_client: AuthorizationManagementClient,
  4. resource_client: ResourceManagementClient,
  5. azure_data_required_data_actions: List[str],
  6. ) -> None: # pragma: no cover
  7. # noqa: D401,E501
  8. sub_id: str = get_config_value(config=config, key="infra:azure:subscription_id")
  9. rg_name: str = get_config_value(config=config, key="infra:azure:metagroup:name")
  10. storage_name: str = get_config_value(config=config, key="env:azure:storage:name")
  11. backup_path: str = get_config_value(
  12. config=config, key="env:azure:storage:path:backup"
  13. )
  14. datalake_admin: str = get_config_value(
  15. config=config, key="env:azure:role:name:datalake_admin"
  16. )
  17. parsed_logger_path = parse_adls_path(backup_path)
  18. container_name = parsed_logger_path[1]
  19. role_assignments = get_role_assignments(
  20. auth_client=auth_client,
  21. resource_client=resource_client,
  22. identity_name=datalake_admin,
  23. subscription_id=sub_id,
  24. resource_group=rg_name,
  25. )
  26. proper_scope = get_storage_container_scope(
  27. sub_id, rg_name, storage_name, container_name
  28. )
  29. _, missing_data_actions = check_for_actions(
  30. auth_client=auth_client,
  31. role_assigments=role_assignments,
  32. proper_scope=proper_scope,
  33. required_actions=[],
  34. required_data_actions=azure_data_required_data_actions,
  35. )
  36. if missing_data_actions:
  37. fail(
  38. AZURE_IDENTITY_MISSING_DATA_ACTIONS_FOR_LOCATION,
  39. subjects=[
  40. datalake_admin,
  41. f"storageAccounts/{storage_name}/blobServices/default/containers/{container_name}", # noqa: E501
  42. ],
  43. resources=missing_data_actions,
  44. )