File manager - Edit - /home/newsbmcs.com/public_html/static/img/logo/clouds.tar
Back
gcp.py 0000644 00000010106 15030505054 0005662 0 ustar 00 import base64 import json import logging import os from typing import Any, Dict, List, Optional # noqa: F401 from uaclient import exceptions, http, secret_manager, system, util from uaclient.clouds import AutoAttachCloudInstance LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) TOKEN_URL = ( "http://metadata/computeMetadata/v1/instance/service-accounts/" "default/identity?audience=contracts.canonical.com&" "format=full&licenses=TRUE" ) LICENSES_URL = ( "http://metadata.google.internal/computeMetadata/v1/instance/licenses/" "?recursive=true" ) WAIT_FOR_CHANGE = "&wait_for_change=true" LAST_ETAG = "&last_etag={etag}" DMI_PRODUCT_NAME = "/sys/class/dmi/id/product_name" GCP_PRODUCT_NAME = "Google Compute Engine" GCP_LICENSES = { "xenial": "8045211386737108299", "bionic": "6022427724719891830", "focal": "599959289349842382", "jammy": "2592866803419978320", "noble": "2176054482269786025", } class UAAutoAttachGCPInstance(AutoAttachCloudInstance): def __init__(self): # store ETAG # https://cloud.google.com/compute/docs/metadata/querying-metadata#etags # noqa self.etag = None # type: Optional[str] # mypy does not handle @property around inner decorators # https://github.com/python/mypy/issues/1362 @property # type: ignore @util.retry(exceptions.CloudMetadataError, retry_sleeps=[0.5, 1, 1]) def identity_doc(self) -> Dict[str, Any]: response = http.readurl( TOKEN_URL, headers={"Metadata-Flavor": "Google"}, timeout=1 ) if response.code == 200: secret_manager.secrets.add_secret(response.body) return {"identityToken": response.body} error_desc = response.json_dict.get("error_description") if error_desc and "service account" in error_desc.lower(): raise exceptions.GCPServiceAccountError( status_code=response.code, error_msg=error_desc ) raise exceptions.CloudMetadataError( code=response.code, body=response.body ) @property def cloud_type(self) -> str: return "gcp" @property def is_viable(self) -> bool: """This machine is a viable GCPInstance""" if os.path.exists(DMI_PRODUCT_NAME): product_name = system.load_file(DMI_PRODUCT_NAME) if GCP_PRODUCT_NAME == product_name.strip(): return True return False def get_licenses_from_identity(self) -> List[str]: """Get a list of licenses from the GCP metadata. Instance identity token (jwt) carries a list of licenses associated with the instance itself. Returns an empty list if licenses are not present in the metadata. """ token = self.identity_doc["identityToken"] identity = base64.urlsafe_b64decode(token.split(".")[1] + "===") identity_dict = json.loads(identity.decode("utf-8")) return ( identity_dict.get("google", {}) .get("compute_engine", {}) .get("license_id", []) ) def should_poll_for_pro_license(self) -> bool: series = system.get_release_info().series if series not in GCP_LICENSES: LOG.info("This series isn't supported for GCP auto-attach.") return False return True def is_pro_license_present(self, *, wait_for_change: bool) -> bool: url = LICENSES_URL if wait_for_change: url += WAIT_FOR_CHANGE if self.etag: url += LAST_ETAG.format(etag=self.etag) response = http.readurl(url, headers={"Metadata-Flavor": "Google"}) if response.code == 200: license_ids = [license["id"] for license in response.json_list] self.etag = response.headers.get("etag") series = system.get_release_info().series return GCP_LICENSES.get(series) in license_ids LOG.error(response.body) if response.code == 400: raise exceptions.CancelProLicensePolling() else: raise exceptions.DelayProLicensePolling() __pycache__/gcp.cpython-310.pyc 0000644 00000010500 15030505054 0012217 0 ustar 00 o �ϴfF � @ s� d dl Z d dlZd dlZd dlZd dlmZmZmZmZ d dl m Z m Z m Z m Z mZ d dlmZ e�e�e��ZdZdZdZdZd Zd Zd d d ddd�ZG dd� de�Z dS )� N)�Any�Dict�List�Optional)� exceptions�http�secret_manager�system�util)�AutoAttachCloudInstancez�http://metadata/computeMetadata/v1/instance/service-accounts/default/identity?audience=contracts.canonical.com&format=full&licenses=TRUEzThttp://metadata.google.internal/computeMetadata/v1/instance/licenses/?recursive=truez&wait_for_change=truez&last_etag={etag}z /sys/class/dmi/id/product_namezGoogle Compute Engine�8045211386737108299�6022427724719891830�599959289349842382�2592866803419978320�2176054482269786025)�xenial�bionic�focal�jammy�noblec @ s� e Zd Zdd� Zeejejg d�d�de e e f fdd���Z ede fdd ��Z edefd d ��Zdee fd d �Zdefdd�Zdedefdd�ZdS )�UAAutoAttachGCPInstancec C s d | _ d S )N��etag��self� r �5/usr/lib/python3/dist-packages/uaclient/clouds/gcp.py�__init__% s z UAAutoAttachGCPInstance.__init__)g �?� r )� retry_sleeps�returnc C st t jtddidd�}|jdkrtj�|j� d|jiS |j� d�}|r1d|� � v r1t j |j|d ��t j |j|jd ��) N�Metadata-Flavor�Googler )�headers�timeout�� � identityToken�error_descriptionzservice account)� status_code� error_msg)�code�body)r �readurl� TOKEN_URLr* r �secrets� add_secretr+ � json_dict�get�lowerr �GCPServiceAccountError�CloudMetadataError)r �response� error_descr r r � identity_doc, s � ��z$UAAutoAttachGCPInstance.identity_docc C s dS )N�gcpr r r r r � cloud_type? s z"UAAutoAttachGCPInstance.cloud_typec C s* t j�t�rt�t�}t|�� krdS dS )z$This machine is a viable GCPInstanceTF)�os�path�exists�DMI_PRODUCT_NAMEr � load_file�GCP_PRODUCT_NAME�strip)r � product_namer r r � is_viableC s z!UAAutoAttachGCPInstance.is_viablec C sN | j d }t�|�d�d d �}t�|�d��}|�di ��di ��dg �S ) z�Get a list of licenses from the GCP metadata. Instance identity token (jwt) carries a list of licenses associated with the instance itself. Returns an empty list if licenses are not present in the metadata. r& �.r z===zutf-8�google�compute_engine� license_id)r7 �base64�urlsafe_b64decode�split�json�loads�decoder1 )r �token�identity� identity_dictr r r �get_licenses_from_identityM s �z2UAAutoAttachGCPInstance.get_licenses_from_identityc C s$ t �� j}|tvrt�d� dS dS )Nz0This series isn't supported for GCP auto-attach.FT)r �get_release_info�series� GCP_LICENSES�LOG�info)r rR r r r �should_poll_for_pro_license^ s z3UAAutoAttachGCPInstance.should_poll_for_pro_license�wait_for_changec C s� t }|r|t7 }| jr|tj| jd�7 }tj|ddid�}|jdkr=dd� |jD �}|j � d�| _t � � j }t� |�|v S t�|j� |jd krLt�� �t�� �) Nr r! r" )r# r% c S s g | ]}|d �qS )�idr )�.0�licenser r r � <listcomp>o s zBUAAutoAttachGCPInstance.is_pro_license_present.<locals>.<listcomp>r i� )� LICENSES_URL�WAIT_FOR_CHANGEr � LAST_ETAG�formatr r, r* � json_listr# r1 r rQ rR rS rT �errorr+ r �CancelProLicensePolling�DelayProLicensePolling)r rW �urlr5 � license_idsrR r r r �is_pro_license_presente s z.UAAutoAttachGCPInstance.is_pro_license_presentN)�__name__� __module__� __qualname__r �propertyr �retryr r4 r �strr r7 r9 �boolrB r rP rV rf r r r r r $ s r ) rG rJ �loggingr: �typingr r r r �uaclientr r r r r �uaclient.cloudsr � getLogger� replace_top_level_logger_namerg rT r- r\ r] r^ r= r? rS r r r r r �<module> s. ��� __pycache__/aws.cpython-310.pyc 0000644 00000007622 15030505054 0012253 0 ustar 00 o �ϴf� � @ s� d dl Z d dlmZmZ d dlmZmZmZmZm Z d dl m Z dZ dZ e e fZdZdZd Zd Zd Zd Zd Zed Ze �e �e��ZG dd� de �ZdS )� N)�Any�Dict)� exceptions�http�secret_manager�system�util)�AutoAttachCloudInstancez169.254.169.254z[fd00:ec2::254]zhttp://{}/latest/api/tokenz0http://{}/latest/dynamic/instance-identity/pkcs7z/sys/hypervisor/uuidz /sys/class/dmi/id/product_serialz /sys/class/dmi/id/product_uuid�21600zX-aws-ec2-metadata-tokenz -ttl-secondsc @ s� e Zd ZdZdZdd� Zeeje j g d�d�de e e f fdd���Zd d � Zeje j g d �d�d d � �Zede fdd��Zedefdd��Zdefdd�Zdedefdd�ZdS )�UAAutoAttachAWSInstanceNc C sB | � � }tjt�| j�|dd�}|jdkr|jS tj |j|jd��)N� )�headers�timeout�� ��code�body) � _request_imds_v2_token_headersr �readurl�IMDS_URL�format� _ip_addressr r r �CloudMetadataError)�selfr �response� r �5/usr/lib/python3/dist-packages/uaclient/clouds/aws.py�_get_imds_url_response s � �z.UAAutoAttachAWSInstance._get_imds_url_response)g �?r r )� retry_sleeps�returnc C s | � � }tj�|� d|iS )N�pkcs7)r r �secrets� add_secret)r �imds_url_responser r r � identity_doc+ s z$UAAutoAttachAWSInstance.identity_docc C s| t D ]+}z| j|d�}W n ty) } zt�d|t|dd�� W Y d }~qd }~ww || _ | jd u r<tjd� t �d��|S )N)� ip_addressz*Could not reach AWS IMDS at http://%s: %s �reason� z, )� addresses) �IMDS_IP_ADDRESS�_get_imds_v2_token_headers� Exception�LOG�warning�getattrr r �AWSNoValidIMDS�join)r �addressr �er r r r 2 s$ ��� �z6UAAutoAttachAWSInstance._request_imds_v2_token_headers)r � � c C s� | j dkrd S | j rt| j iS tjt�|�dttidd�}|jdkr2|j | _ t j � | j � t| j iS |jdkr<d| _ d S t j|j|j d��)N�IMDSv1�PUTr )�methodr r r i� r )� _api_token�AWS_TOKEN_PUT_HEADERr r �IMDS_V2_TOKEN_URLr �AWS_TOKEN_REQ_HEADER�AWS_TOKEN_TTL_SECONDSr r r r! r" r r )r r% r r r r r* E s( � �z2UAAutoAttachAWSInstance._get_imds_v2_token_headersc C � dS )N�awsr �r r r r � cloud_type^ � z"UAAutoAttachAWSInstance.cloud_typec C s� zt �t�}d|dd� krW dS W n ty Y nw t �t��� }t �t��� }d|dd� kr=|dd� kr@dS dS dS )z$This machine is a viable AWSInstance�ec2r � TF)r � load_file�SYS_HYPERVISOR_PRODUCT_UUID�FileNotFoundError�DMI_PRODUCT_UUID�lower�DMI_PRODUCT_SERIAL)r �hypervisor_uuid�dmi_uuid� dmi_serialr r r � is_viableb s � �"�z!UAAutoAttachAWSInstance.is_viablec C r= )� UnsupportedFr r? r r r �should_poll_for_pro_licenset rA z3UAAutoAttachAWSInstance.should_poll_for_pro_license�wait_for_changec C s t �� �)N)r � InPlaceUpgradeNotSupportedError)r rP r r r �is_pro_license_presentx s z.UAAutoAttachAWSInstance.is_pro_license_present)�__name__� __module__� __qualname__r8 r r �propertyr �retryr r r �strr r$ r r* r@ �boolrM rO rR r r r r r s r ) �logging�typingr r �uaclientr r r r r �uaclient.cloudsr �IMDS_IPV4_ADDRESS�IMDS_IPV6_ADDRESSr) r: r rE rI rG r< r9 r; � getLogger� replace_top_level_logger_namerS r, r r r r r �<module> s" __pycache__/__init__.cpython-310.pyc 0000644 00000003223 15030505054 0013211 0 ustar 00 o �ϴf� � @ s0 d dl Z d dlmZmZ G dd� de jd�ZdS )� N)�Any�Dictc @ s� e Zd Zeejdeeef fdd���Z eejdefdd���Z eejde fdd���Z ejde fdd ��Z ejd e de fd d ��Zd S )�AutoAttachCloudInstance�returnc C � dS )z=Return the identity document representing this cloud instanceN� ��selfr r �:/usr/lib/python3/dist-packages/uaclient/clouds/__init__.py� identity_doc � z$AutoAttachCloudInstance.identity_docc C r )z=Return a string of the cloud type on which this instance runsNr r r r r � cloud_type r z"AutoAttachCloudInstance.cloud_typec C r )z?Return True if the machine is a viable AutoAttachCloudInstance.Nr r r r r � is_viable r z!AutoAttachCloudInstance.is_viablec C r )zw Cloud-specific checks for whether the daemon should continously poll for Ubuntu Pro licenses. Nr r r r r �should_poll_for_pro_license s z3AutoAttachCloudInstance.should_poll_for_pro_license�wait_for_changec C r )z1 Check for an Ubuntu Pro license Nr )r r r r r �is_pro_license_present s z.AutoAttachCloudInstance.is_pro_license_presentN)�__name__� __module__� __qualname__�property�abc�abstractmethodr �strr r r �boolr r r r r r r r s r )� metaclass)r �typingr r �ABCMetar r r r r �<module> s __pycache__/identity.cpython-310.pyc 0000644 00000005763 15030505054 0013316 0 ustar 00 o �ϴfi � @ s d dl Z d dlmZ d dlmZ d dlmZmZmZm Z d dl m Z m Z m Z mZmZ d dlmZ e �e�e��Zddd d d d �Ze jje jje jjd �ZG dd� de�Z dee fdd�Z edd�ed�deee ee f fdd���Z ddee de j fdd�Z!dS )� N)�Enum)� lru_cache)�Dict�Optional�Tuple�Type)�clouds� exceptions�messages�system�util)� apply_config_settings_override�AWSz AWS ChinazAWS Gov�Azure�GCP��awsz aws-chinazaws-gov�azure�gce)r r r c @ s e Zd ZdZdZdS )�NoCloudTypeReasonr � N)�__name__� __module__� __qualname__�NO_CLOUD_DETECTED�CLOUD_ID_ERROR� r r �:/usr/lib/python3/dist-packages/uaclient/clouds/identity.pyr s r �returnc C s@ zt �g d��\} }| �� W S tjy Y nw t�d� dS )z%Query cloud instance-id from cmdline.)z cloud-init�query� instance_idz'Unable to determine current instance-idN)r �subp�stripr �ProcessExecutionError�LOG�warning)�out�_errr r r �get_instance_id s � r( )�maxsize� cloud_typec C st t �d�r5zt �dg�\} }| �� d fW S tjy4 } zt�dt|�� d t j fW Y d }~S d }~ww d t j fS )Nzcloud-idzerror running cloud-id: %s) r �whichr! r" r r# r$ �debug�strr r r )r& r' �excr r r �get_cloud_type+ s �� r/ �cloud_overridec C s� ddl m}m}m} |j|j|j|j|jd�}| dur | }nt� \}}|s)t� � �|� |�}|s6tj |d��|� }|j s@t� � �|S )a :raises CloudFactoryError: if no cloud instance object can be constructed :raises CloudFactoryNoCloudError: if no cloud instance object can be constructed because we are not on a cloud :raises CloudFactoryUnsupportedCloudError: if no cloud instance object can be constructed because we don't have a class for the cloud we're on :raises CloudFactoryNonViableCloudError: if no cloud instance object can be constructed because we explicitly do not support the cloud we're on r )r r �gcpr N)r* )�uaclient.cloudsr r r1 �UAAutoAttachAWSInstance�UAAutoAttachAzureInstance�UAAutoAttachGCPInstancer/ r �CloudFactoryNoCloudError�get�NonAutoAttachImageError� is_viable� CloudFactoryNonViableCloudError) r0 r r r1 �cloud_instance_mapr* �_�cls�instancer r r �cloud_instance_factory: s& � r? )N)"�logging�enumr � functoolsr �typingr r r r �uaclientr r r r r �uaclient.configr � getLogger� replace_top_level_logger_namer r$ �CLOUD_TYPE_TO_TITLE�urls�PRO_ON_AWS_HOME_PAGE�PRO_ON_AZURE_HOME_PAGE�PRO_ON_GCP_HOME_PAGE�PRO_CLOUD_URLSr r- r( r/ �AutoAttachCloudInstancer? r r r r �<module> s8 � � "��� __pycache__/azure.cpython-310.pyc 0000644 00000005516 15030505054 0012607 0 ustar 00 o �ϴfY � @ s� d dl Z d dlZd dlmZmZ d dlmZmZmZm Z m Z d dl m Z e � e �e��ZdZdZed e ed e d �Zd Zd Zd Zd ZG dd� de �ZdS )� N)�Any�Dict)� exceptions�http�secret_manager�system�util)�AutoAttachCloudInstancez http://169.254.169.254/metadata/z 2020-09-01z attested/document?api-version=z instance/compute?api-version=)�pkcs7�computez#/sys/class/dmi/id/chassis_asset_tagz%/var/lib/cloud/seed/azure/ovf-env.xmlz 7783-7084-3265-9085-8269-3286-77� UBUNTU_PROc @ s~ e Zd Zeejejg d�d�dee e f fdd���Z ede fdd��Z ede fdd ��Zde fd d �Zd e de fd d�ZdS )�UAAutoAttachAzureInstance)� r r )� retry_sleeps�returnc C s� i }t t�� �D ]5\}}tj|ddidd�}|jdkr$tj|j|jd��|dkr8|j d }|||< t j � |� q|j ||< q|S ) N�Metadata�truer )�headers�timeout�� )�code�bodyr � signature) �sorted� IMDS_URLS�itemsr �readurlr r �CloudMetadataErrorr � json_dictr �secrets� add_secret)�self� responses�key�url�responser � r& �7/usr/lib/python3/dist-packages/uaclient/clouds/azure.py� identity_doc s � � z&UAAutoAttachAzureInstance.identity_docc C � dS )N�azurer& �r! r& r& r' � cloud_type/ � z$UAAutoAttachAzureInstance.cloud_typec C s2 t j�t�rt�t�}t|�� krdS t j�t�S )z&This machine is a viable AzureInstanceT) �os�path�exists�DMI_CHASSIS_ASSET_TAGr � load_file�AZURE_CHASSIS_ASSET_TAG�strip�AZURE_OVF_ENV_FILE)r! �chassis_asset_tagr& r& r' � is_viable3 s z#UAAutoAttachAzureInstance.is_viablec C r) )NTr& r+ r& r& r' �should_poll_for_pro_license< r- z5UAAutoAttachAzureInstance.should_poll_for_pro_license�wait_for_changec C s� |rt �� �t�dd�}z tj|ddid�}W n ty, } z t�|� t �� �d }~ww |j dkr<t�|j � t �� �|j �d�t kS )Nr � r r )r r � licenseType) r �CancelProLicensePollingr �getr r �OSError�LOG�errorr r r �AZURE_PRO_LICENSE_TYPE)r! r9 r$ r% �er& r& r' �is_pro_license_present@ s �� z0UAAutoAttachAzureInstance.is_pro_license_presentN)�__name__� __module__� __qualname__�propertyr �retryr r r �strr r( r, �boolr7 r8 rC r& r& r& r' r s r )�loggingr. �typingr r �uaclientr r r r r �uaclient.cloudsr � getLogger� replace_top_level_logger_namerD r? � IMDS_BASE_URL� API_VERSIONr r1 r5 r3 rA r r& r& r& r' �<module> s � identity.py 0000644 00000006151 15030505054 0006747 0 ustar 00 import logging from enum import Enum from functools import lru_cache from typing import Dict, Optional, Tuple, Type # noqa: F401 from uaclient import clouds, exceptions, messages, system, util from uaclient.config import apply_config_settings_override LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) CLOUD_TYPE_TO_TITLE = { "aws": "AWS", "aws-china": "AWS China", "aws-gov": "AWS Gov", "azure": "Azure", "gce": "GCP", } PRO_CLOUD_URLS = { "aws": messages.urls.PRO_ON_AWS_HOME_PAGE, "azure": messages.urls.PRO_ON_AZURE_HOME_PAGE, "gce": messages.urls.PRO_ON_GCP_HOME_PAGE, } class NoCloudTypeReason(Enum): NO_CLOUD_DETECTED = 0 CLOUD_ID_ERROR = 1 def get_instance_id() -> Optional[str]: """Query cloud instance-id from cmdline.""" try: # Present in cloud-init on >= Xenial out, _err = system.subp(["cloud-init", "query", "instance_id"]) return out.strip() except exceptions.ProcessExecutionError: pass LOG.warning("Unable to determine current instance-id") return None @lru_cache(maxsize=None) @apply_config_settings_override("cloud_type") def get_cloud_type() -> Tuple[Optional[str], Optional[NoCloudTypeReason]]: if system.which("cloud-id"): # Present in cloud-init on >= Xenial try: out, _err = system.subp(["cloud-id"]) return (out.strip(), None) except exceptions.ProcessExecutionError as exc: LOG.debug("error running cloud-id: %s", str(exc)) return (None, NoCloudTypeReason.CLOUD_ID_ERROR) # If no cloud-id command, assume not on cloud return (None, NoCloudTypeReason.NO_CLOUD_DETECTED) def cloud_instance_factory( cloud_override: Optional[str] = None, ) -> clouds.AutoAttachCloudInstance: """ :raises CloudFactoryError: if no cloud instance object can be constructed :raises CloudFactoryNoCloudError: if no cloud instance object can be constructed because we are not on a cloud :raises CloudFactoryUnsupportedCloudError: if no cloud instance object can be constructed because we don't have a class for the cloud we're on :raises CloudFactoryNonViableCloudError: if no cloud instance object can be constructed because we explicitly do not support the cloud we're on """ from uaclient.clouds import aws, azure, gcp cloud_instance_map = { "aws": aws.UAAutoAttachAWSInstance, "aws-china": aws.UAAutoAttachAWSInstance, "aws-gov": aws.UAAutoAttachAWSInstance, "azure": azure.UAAutoAttachAzureInstance, "gce": gcp.UAAutoAttachGCPInstance, } # type: Dict[str, Type[clouds.AutoAttachCloudInstance]] if cloud_override is not None: cloud_type = cloud_override else: cloud_type, _ = get_cloud_type() if not cloud_type: raise exceptions.CloudFactoryNoCloudError() cls = cloud_instance_map.get(cloud_type) if not cls: raise exceptions.NonAutoAttachImageError(cloud_type=cloud_type) instance = cls() if not instance.is_viable: raise exceptions.CloudFactoryNonViableCloudError() return instance azure.py 0000644 00000005531 15030505054 0006245 0 ustar 00 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 __init__.py 0000644 00000001736 15030505054 0006661 0 ustar 00 import abc from typing import Any, Dict class AutoAttachCloudInstance(metaclass=abc.ABCMeta): @property @abc.abstractmethod def identity_doc(self) -> Dict[str, Any]: """Return the identity document representing this cloud instance""" pass @property @abc.abstractmethod def cloud_type(self) -> str: """Return a string of the cloud type on which this instance runs""" pass @property @abc.abstractmethod def is_viable(self) -> bool: """Return True if the machine is a viable AutoAttachCloudInstance.""" pass @abc.abstractmethod def should_poll_for_pro_license(self) -> bool: """ Cloud-specific checks for whether the daemon should continously poll for Ubuntu Pro licenses. """ pass @abc.abstractmethod def is_pro_license_present(self, *, wait_for_change: bool) -> bool: """ Check for an Ubuntu Pro license """ pass aws.py 0000644 00000010306 15030505054 0005705 0 ustar 00 import logging from typing import Any, Dict from uaclient import exceptions, http, secret_manager, system, util from uaclient.clouds import AutoAttachCloudInstance IMDS_IPV4_ADDRESS = "169.254.169.254" IMDS_IPV6_ADDRESS = "[fd00:ec2::254]" IMDS_IP_ADDRESS = (IMDS_IPV4_ADDRESS, IMDS_IPV6_ADDRESS) IMDS_V2_TOKEN_URL = "http://{}/latest/api/token" IMDS_URL = "http://{}/latest/dynamic/instance-identity/pkcs7" SYS_HYPERVISOR_PRODUCT_UUID = "/sys/hypervisor/uuid" DMI_PRODUCT_SERIAL = "/sys/class/dmi/id/product_serial" DMI_PRODUCT_UUID = "/sys/class/dmi/id/product_uuid" AWS_TOKEN_TTL_SECONDS = "21600" AWS_TOKEN_PUT_HEADER = "X-aws-ec2-metadata-token" AWS_TOKEN_REQ_HEADER = AWS_TOKEN_PUT_HEADER + "-ttl-seconds" LOG = logging.getLogger(util.replace_top_level_logger_name(__name__)) class UAAutoAttachAWSInstance(AutoAttachCloudInstance): _api_token = None _ip_address = None def _get_imds_url_response(self): headers = self._request_imds_v2_token_headers() response = http.readurl( IMDS_URL.format(self._ip_address), headers=headers, timeout=1 ) if response.code == 200: return response.body else: raise exceptions.CloudMetadataError( code=response.code, body=response.body ) # mypy does not handle @property around inner decorators # https://github.com/python/mypy/issues/1362 @property # type: ignore @util.retry(exceptions.CloudMetadataError, retry_sleeps=[0.5, 1, 1]) def identity_doc(self) -> Dict[str, Any]: imds_url_response = self._get_imds_url_response() secret_manager.secrets.add_secret(imds_url_response) return {"pkcs7": imds_url_response} def _request_imds_v2_token_headers(self): for address in IMDS_IP_ADDRESS: try: headers = self._get_imds_v2_token_headers(ip_address=address) except Exception as e: LOG.warning( "Could not reach AWS IMDS at http://%s: %s\n", address, getattr(e, "reason", ""), ) else: self._ip_address = address break if self._ip_address is None: raise exceptions.AWSNoValidIMDS( addresses=", ".join(IMDS_IP_ADDRESS) ) return headers @util.retry(exceptions.CloudMetadataError, retry_sleeps=[1, 2, 5]) def _get_imds_v2_token_headers(self, ip_address): if self._api_token == "IMDSv1": return None elif self._api_token: return {AWS_TOKEN_PUT_HEADER: self._api_token} response = http.readurl( IMDS_V2_TOKEN_URL.format(ip_address), method="PUT", headers={AWS_TOKEN_REQ_HEADER: AWS_TOKEN_TTL_SECONDS}, timeout=1, ) if response.code == 200: self._api_token = response.body secret_manager.secrets.add_secret(self._api_token) return {AWS_TOKEN_PUT_HEADER: self._api_token} if response.code == 404: self._api_token = "IMDSv1" return None raise exceptions.CloudMetadataError( code=response.code, body=response.body ) @property def cloud_type(self) -> str: return "aws" @property def is_viable(self) -> bool: """This machine is a viable AWSInstance""" try: hypervisor_uuid = system.load_file(SYS_HYPERVISOR_PRODUCT_UUID) if "ec2" == hypervisor_uuid[0:3]: return True except FileNotFoundError: # SYS_HYPERVISOR_PRODUCT_UUID isn't present on all EC2 instance # types, fall through pass # Both DMI product_uuid and product_serial start with 'ec2' dmi_uuid = system.load_file(DMI_PRODUCT_UUID).lower() dmi_serial = system.load_file(DMI_PRODUCT_SERIAL).lower() if "ec2" == dmi_uuid[0:3] == dmi_serial[0:3]: return True return False def should_poll_for_pro_license(self) -> bool: """Unsupported""" return False def is_pro_license_present(self, *, wait_for_change: bool) -> bool: raise exceptions.InPlaceUpgradeNotSupportedError()
| ver. 1.4 |
Github
|
.
| PHP 8.2.28 | Generation time: 0.02 |
proxy
|
phpinfo
|
Settings