FM / uaclient

/usr/lib/python3/dist-packages/uaclient/

/usr/lib/python3/dist-packages/uaclient

Name	Size	Mode	Actions
api/	-	0755	rm
cli/	-	0755	rm
clouds/	-	0755	rm
daemon/	-	0755	rm
entitlements/	-	0755	rm
files/	-	0755	rm
http/	-	0755	rm
messages/	-	0755	rm
timer/	-	0755	rm
__pycache__/	-	0755	rm
actions.py	15187	0644	edit dl rm
apt.py	37126	0644	edit dl rm
apt_news.py	8587	0644	edit dl rm
config.py	19361	0644	edit dl rm
contract.py	36743	0644	edit dl rm
contract_data_types.py	10174	0644	edit dl rm
data_types.py	14329	0644	edit dl rm
defaults.py	3269	0644	edit dl rm
event_logger.py	8257	0644	edit dl rm
exceptions.py	18724	0644	edit dl rm
gpg.py	836	0644	edit dl rm
livepatch.py	13160	0644	edit dl rm
lock.py	4525	0644	edit dl rm
log.py	5017	0644	edit dl rm
secret_manager.py	648	0644	edit dl rm
security_status.py	26785	0644	edit dl rm
snap.py	6415	0644	edit dl rm
status.py	29212	0644	edit dl rm
system.py	28793	0644	edit dl rm
types.py	308	0644	edit dl rm
update_contract_info.py	1586	0644	edit dl rm
upgrade_lts_contract.py	3295	0644	edit dl rm
util.py	15820	0644	edit dl rm
version.py	2699	0644	edit dl rm
yaml.py	840	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /usr/lib/python3/dist-packages/uaclient/snap.py (6415B)

import logging
import re
from typing import List, NamedTuple, Optional

from uaclient import (
    api,
    apt,
    event_logger,
    exceptions,
    http,
    messages,
    system,
    util,
)

SNAP_CMD = "/usr/bin/snap"
SNAP_INSTALL_RETRIES = [0.5, 1.0, 5.0]
HTTP_PROXY_OPTION = "proxy.http"
HTTPS_PROXY_OPTION = "proxy.https"
SNAPD_SOCKET_PATH = "/run/snapd.socket"
SNAPD_SNAPS_API = "/v2/snaps/{}"

event = event_logger.get_event_logger()
LOG = logging.getLogger(util.replace_top_level_logger_name(__name__))

SnapPackage = NamedTuple(
    "SnapPackage",
    [
        ("name", str),
        ("version", str),
        ("revision", str),
        ("channel", str),
        ("publisher", str),
    ],
)

def is_snapd_installed() -> bool:
    """Returns whether or not snap is installed"""
    return "snapd" in apt.get_installed_packages_names()

def is_snapd_installed_as_a_snap() -> bool:
    """Returns whether or not snapd is installed as a snap"""
    return any((snap.name == "snapd" for snap in get_installed_snaps()))

def configure_snap_proxy(
    http_proxy: Optional[str] = None,
    https_proxy: Optional[str] = None,
    retry_sleeps: Optional[List[float]] = None,
) -> None:
    """
    Configure snap to use http and https proxies.

:param http_proxy: http proxy to be used by snap. If None, it will
                       not be configured
    :param https_proxy: https proxy to be used by snap. If None, it will
                        not be configured
    :param retry_sleeps: Optional list of sleep lengths to apply between
        retries. Specifying a list of [0.5, 1] tells subp to retry twice
        on failure; sleeping half a second before the first retry and 1 second
        before the second retry.
    """
    if not is_snapd_installed():
        LOG.debug("Skipping configure snap proxy. snapd is not installed.")
        return

if http_proxy or https_proxy:
        event.info(messages.SETTING_SERVICE_PROXY.format(service="snap"))

if http_proxy:
        system.subp(
            ["snap", "set", "system", "proxy.http={}".format(http_proxy)],
            retry_sleeps=retry_sleeps,
        )

if https_proxy:
        system.subp(
            ["snap", "set", "system", "proxy.https={}".format(https_proxy)],
            retry_sleeps=retry_sleeps,
        )

def unconfigure_snap_proxy(
    protocol_type: str, retry_sleeps: Optional[List[float]] = None
) -> None:
    """
    Unset snap configuration settings for http and https proxies.

:param protocol_type: String either http or https
    :param retry_sleeps: Optional list of sleep lengths to apply between
        retries. Specifying a list of [0.5, 1] tells subp to retry twice
        on failure; sleeping half a second before the first retry and 1 second
        before the second retry.
    """
    if not is_snapd_installed():
        LOG.debug("Skipping unconfigure snap proxy. snapd is not installed.")
        return
    system.subp(
        ["snap", "unset", "system", "proxy.{}".format(protocol_type)],
        retry_sleeps=retry_sleeps,
    )

def get_config_option_value(key: str) -> Optional[str]:
    """
    Gets the config value from snap.
    :param protocol: can be any valid snap config option
    :return: the value of the snap config option, or None if not set
    """
    try:
        out, _ = system.subp(["snap", "get", "system", key])
        return out.strip()
    except exceptions.ProcessExecutionError:
        return None

def get_installed_snaps() -> List[SnapPackage]:
    out, _ = system.subp(
        ["snap", "list", "--color", "never", "--unicode", "never"]
    )
    apps = out.splitlines()
    apps = apps[1:]
    snaps = []
    for line in apps:
        snap = line.split()[0]
        snaps.append(get_snap_info(snap))

return snaps

def install_snapd():
    event.info(messages.APT_UPDATING_LIST.format(name="standard Ubuntu"))
    try:
        apt.update_sources_list(apt.get_system_sources_file())
    except exceptions.UbuntuProError as e:
        LOG.debug(
            "Trying to install snapd. Ignoring apt-get update failure: %s",
            str(e),
        )
    try:
        system.subp(
            ["apt-get", "install", "--assume-yes", "snapd"],
            retry_sleeps=apt.APT_RETRIES,
        )
    except exceptions.ProcessExecutionError:
        raise exceptions.CannotInstallSnapdError()

def run_snapd_wait_cmd(progress: api.ProgressWrapper):
    try:
        system.subp([SNAP_CMD, "wait", "system", "seed.loaded"], capture=True)
    except exceptions.ProcessExecutionError as e:
        if re.search(r"unknown command .*wait", str(e).lower()):
            LOG.warning(
                "Detected version of snapd that does not have wait command"
            )
            progress.emit("info", messages.SNAPD_DOES_NOT_HAVE_WAIT_CMD)
        else:
            raise

def install_snap(
    snap: str,
    channel: Optional[str] = None,
    classic_confinement_support: bool = False,
):
    cmd = [SNAP_CMD, "install", snap]

if classic_confinement_support:
        cmd += ["--classic"]

if channel:
        cmd += ["--channel={}".format(channel)]

system.subp(
        cmd,
        capture=True,
        retry_sleeps=SNAP_INSTALL_RETRIES,
    )

def refresh_snap(snap: str):
    system.subp([SNAP_CMD, "refresh", snap], capture=True)

def get_snap_info(snap: str) -> SnapPackage:
    url = SNAPD_SNAPS_API.format(snap)
    try:
        resp = http.unix_socket_request(SNAPD_SOCKET_PATH, "GET", url)
        # This means that the snap doesn't exist or is not installed
        if resp.code != 200:
            if (
                resp.code == 404
                and resp.json_dict.get("result", {}).get("kind")
                == "snap-not-found"
            ):
                raise exceptions.SnapNotInstalledError(snap=snap)
            else:
                error_msg = resp.json_dict.get("result", {}).get("message")
                raise exceptions.UnexpectedSnapdAPIError(error=error_msg)

except ConnectionRefusedError:
        raise exceptions.SnapdAPIConnectionRefused()

snap_info = resp.json_dict.get("result", {})
    return SnapPackage(
        name=snap_info.get("name", ""),
        version=snap_info.get("version", ""),
        revision=snap_info.get("revision", ""),
        channel=snap_info.get("channel", ""),
        publisher=snap_info.get("publisher", {}).get("username", ""),
    )

cmd: