Gecko [ tsuru-no-tsurugi.tokyo ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2026-07-18 06:45
__init__.py	1.76 KB	-rw-r--r--	2026-04-08 03:24
aws.py	5.67 KB	-rw-r--r--	2026-04-08 03:24
azure.py	2.85 KB	-rw-r--r--	2026-04-08 03:24
gcp.py	4.29 KB	-rw-r--r--	2026-04-08 03:24
identity.py	3.13 KB	-rw-r--r--	2026-04-08 03:24
lxd.py	2.54 KB	-rw-r--r--	2026-04-08 03:24

Rename

import logging

from uaclient import config, exceptions, http, log, secret_manager, util
from uaclient.clouds import AutoAttachInstance

LOG = logging.getLogger(util.replace_top_level_logger_name(__name__))

LXD_INSTANCE_API_SOCKET_PATH = "/dev/lxd/sock"
LXD_INSTANCE_API_ENDPOINT_UBUNTU_PRO = "/1.0/ubuntu-pro"
LXD_INSTANCE_API_ENDPOINT_UBUNTU_PRO_GUEST_TOKEN = "/1.0/ubuntu-pro/token"

class LXDAutoAttachInstance(AutoAttachInstance):
    @property
    def is_viable(self) -> bool:
        return True

def should_poll_for_pro_license(self) -> bool:
        """Yes, but only once - is_pro_license_present doesn't
        support wait_for_change"""
        return True

def is_pro_license_present(self, *, wait_for_change: bool) -> bool:
        if wait_for_change:
            # Unsupported
            raise exceptions.CancelProLicensePolling()

resp = http.unix_socket_request(
            LXD_INSTANCE_API_SOCKET_PATH,
            "GET",
            LXD_INSTANCE_API_ENDPOINT_UBUNTU_PRO,
        )
        if resp.code != 200:
            LOG.error(
                "LXD instance API returned error for ubuntu-pro query",
                extra=log.extra(code=resp.code, body=resp.body),
            )
            return False

# returning True will cause auto-attach on launch, so only "on" counts
        if resp.json_dict.get("guest_attach", "off") != "on":
            LOG.info(
                "guest_attach config is not turned on. The machine will not try to auto-attach."  # noqa
            )
            return False

return True

def acquire_pro_token(self, cfg: config.UAConfig) -> str:
        """
        Cloud-specific implementation of acquiring the pro token using whatever
        method suits the platform
        """
        resp = http.unix_socket_request(
            LXD_INSTANCE_API_SOCKET_PATH,
            "POST",
            LXD_INSTANCE_API_ENDPOINT_UBUNTU_PRO_GUEST_TOKEN,
        )
        if resp.code == 404:
            raise exceptions.LXDAutoAttachNotAvailable()
        elif resp.code == 403:
            raise exceptions.LXDAutoAttachNotAllowed()
        elif resp.code != 200:
            raise exceptions.ExternalAPIError(
                code=resp.code,
                url="unix://{}{}".format(
                    LXD_INSTANCE_API_SOCKET_PATH,
                    LXD_INSTANCE_API_ENDPOINT_UBUNTU_PRO_GUEST_TOKEN,
                ),
                body=resp.body,
            )
        guest_token = resp.json_dict.get("guest_token", "")
        secret_manager.secrets.add_secret(guest_token)
        return guest_token