404

[ Avaa Bypassed ]




Upload:

Command:

botdev@3.131.37.237: ~ $
from typing import Any, Dict, Optional, Tuple

from uaclient import 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, silent: bool = False) -> bool:
        ret = super()._perform_enable(silent=silent)

        if not ret:
            return ret

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

        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, silent=False):
        super()._perform_disable(silent=silent)
        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

Filemanager

Name Type Size Permission Actions
__pycache__ Folder 0755
__init__.py File 6.76 KB 0644
anbox.py File 3.04 KB 0644
base.py File 44.05 KB 0644
cc.py File 1013 B 0644
cis.py File 1.14 KB 0644
entitlement_status.py File 3.12 KB 0644
esm.py File 3.85 KB 0644
fips.py File 19.99 KB 0644
landscape.py File 4.84 KB 0644
livepatch.py File 12.63 KB 0644
realtime.py File 5.46 KB 0644
repo.py File 21.85 KB 0644
ros.py File 1.47 KB 0644