1#@!#!123s

: / lib / python3 / dist-packages / uaclient / entitlements /
Filename : anbox.py
back
from typing import Any, Dict, Optional, Tuple

from uaclient import api, contract, event_logger, messages, system
from uaclient.entitlements.entitlement_status import (
    CanEnableFailure,
    CanEnableFailureReason,
)
from uaclient.entitlements.repo import RepoEntitlement
from uaclient.files.state_files import (
    AnboxCloudData,
    anbox_cloud_credentials_file,
)
from uaclient.types import MessagingOperationsDict

event = event_logger.get_event_logger()


class AnboxEntitlement(RepoEntitlement):
    name = "anbox-cloud"
    title = messages.ANBOX_TITLE
    description = messages.ANBOX_DESCRIPTION
    help_doc_url = messages.urls.ANBOX_HOME_PAGE
    help_text = messages.ANBOX_HELP_TEXT
    repo_key_file = "ubuntu-pro-anbox-cloud.gpg"
    repo_url_tmpl = "{}"
    affordance_check_series = True
    supports_access_only = True
    origin = "Anbox"

    @property
    def messaging(self) -> MessagingOperationsDict:
        if not self.access_only:
            return {"post_enable": [messages.ANBOX_RUN_INIT_CMD]}
        else:
            return {}

    def can_enable(self) -> Tuple[bool, Optional[CanEnableFailure]]:
        ret, reason = super().can_enable()

        if not ret:
            return ret, reason

        if system.is_container() and not self.access_only:
            return (
                False,
                CanEnableFailure(
                    CanEnableFailureReason.ONLY_ACCESS_ONLY_SUPPORTED,
                    messages.ANBOX_FAIL_TO_ENABLE_ON_CONTAINER,
                ),
            )

        return True, None

    def _perform_enable(self, progress: api.ProgressWrapper) -> bool:
        ret = super()._perform_enable(progress)

        if not ret:
            return ret

        directives = self.entitlement_cfg.get("entitlement", {}).get(
            "directives", {}
        )
        machine_token = self.machine_token_file.machine_token["machineToken"]
        client = contract.UAContractClient(self.cfg)
        anbox_images_machine_access = client.get_resource_machine_access(
            machine_token, "anbox-images"
        )

        anbox_cloud_data = AnboxCloudData(
            anbox_images_url=anbox_images_machine_access.get("entitlement", {})
            .get("directives", {})
            .get("url", ""),
            anbox_images_resource_token=anbox_images_machine_access.get(
                "resourceToken", ""
            ),
            anbox_cloud_apt_url=directives.get("aptURL", ""),
            anbox_cloud_apt_token=directives.get("aptKey", ""),
        )

        anbox_cloud_credentials_file.write(anbox_cloud_data)

        return True

    def _perform_disable(self, progress: api.ProgressWrapper):
        super()._perform_disable(progress)
        anbox_cloud_credentials_file.delete()
        return True

    # TODO: remove this function
    # This is just a placeholder until we can deliver the Anbox
    # resourceToken from the contracts without relying on the
    # enableByDefault obligation
    def _should_enable_by_default(
        self, obligations: Dict[str, Any], resourceToken: Optional[str]
    ) -> bool:
        return False