| Server IP : 127.0.0.2 / Your IP : 18.116.239.11 Web Server : Apache/2.4.18 (Ubuntu) System : User : www-data ( ) PHP Version : 7.0.33-0ubuntu0.16.04.16 Disable Function : disk_free_space,disk_total_space,diskfreespace,dl,exec,fpaththru,getmyuid,getmypid,highlight_file,ignore_user_abord,leak,listen,link,opcache_get_configuration,opcache_get_status,passthru,pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,php_uname,phpinfo,posix_ctermid,posix_getcwd,posix_getegid,posix_geteuid,posix_getgid,posix_getgrgid,posix_getgrnam,posix_getgroups,posix_getlogin,posix_getpgid,posix_getpgrp,posix_getpid,posix,_getppid,posix_getpwnam,posix_getpwuid,posix_getrlimit,posix_getsid,posix_getuid,posix_isatty,posix_kill,posix_mkfifo,posix_setegid,posix_seteuid,posix_setgid,posix_setpgid,posix_setsid,posix_setuid,posix_times,posix_ttyname,posix_uname,pclose,popen,proc_open,proc_close,proc_get_status,proc_nice,proc_terminate,shell_exec,source,show_source,system,virtual MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : ON | Pkexec : ON Directory : /usr/lib/python3/dist-packages/uaclient/clouds/ |
Upload File : |
import logging
import os
from typing import Any, Dict
from uaclient import exceptions, http, secret_manager, system, util
from uaclient.clouds import AutoAttachCloudInstance
LOG = logging.getLogger(util.replace_top_level_logger_name(__name__))
IMDS_BASE_URL = "http://169.254.169.254/metadata/"
API_VERSION = "2020-09-01" # Needed to get subscription ID in attested data
IMDS_URLS = {
"pkcs7": IMDS_BASE_URL + "attested/document?api-version=" + API_VERSION,
"compute": IMDS_BASE_URL + "instance/compute?api-version=" + API_VERSION,
}
DMI_CHASSIS_ASSET_TAG = "/sys/class/dmi/id/chassis_asset_tag"
AZURE_OVF_ENV_FILE = "/var/lib/cloud/seed/azure/ovf-env.xml"
AZURE_CHASSIS_ASSET_TAG = "7783-7084-3265-9085-8269-3286-77"
AZURE_PRO_LICENSE_TYPE = "UBUNTU_PRO"
class UAAutoAttachAzureInstance(AutoAttachCloudInstance):
# mypy does not handle @property around inner decorators
# https://github.com/python/mypy/issues/1362
@property # type: ignore
@util.retry(exceptions.CloudMetadataError, retry_sleeps=[1, 1, 1])
def identity_doc(self) -> Dict[str, Any]:
responses = {}
for key, url in sorted(IMDS_URLS.items()):
response = http.readurl(
url, headers={"Metadata": "true"}, timeout=1
)
if response.code != 200:
raise exceptions.CloudMetadataError(
code=response.code, body=response.body
)
if key == "pkcs7":
signature = response.json_dict["signature"]
responses[key] = signature
secret_manager.secrets.add_secret(signature)
else:
responses[key] = response.json_dict
return responses
@property
def cloud_type(self) -> str:
return "azure"
@property
def is_viable(self) -> bool:
"""This machine is a viable AzureInstance"""
if os.path.exists(DMI_CHASSIS_ASSET_TAG):
chassis_asset_tag = system.load_file(DMI_CHASSIS_ASSET_TAG)
if AZURE_CHASSIS_ASSET_TAG == chassis_asset_tag.strip():
return True
return os.path.exists(AZURE_OVF_ENV_FILE)
def should_poll_for_pro_license(self) -> bool:
# Azure will make sure it is on all supported versions
return True
def is_pro_license_present(self, *, wait_for_change: bool) -> bool:
if wait_for_change:
raise exceptions.CancelProLicensePolling()
url = IMDS_URLS.get("compute", "")
try:
response = http.readurl(url, headers={"Metadata": "true"})
except OSError as e:
LOG.error(e)
raise exceptions.CancelProLicensePolling()
if response.code != 200:
LOG.error(response.body)
raise exceptions.CancelProLicensePolling()
return response.json_dict.get("licenseType") == AZURE_PRO_LICENSE_TYPE