Na}{

File Content: `debian.py`

# Copyright (C) 2012 Canonical Ltd.
# Copyright (C) 2012 Hewlett-Packard Development Company, L.P.
# Copyright (C) 2012 Yahoo! Inc.
#
# Author: Scott Moser <scott.moser@canonical.com>
# Author: Juerg Haefliger <juerg.haefliger@hp.com>
# Author: Joshua Harlow <harlowja@yahoo-inc.com>
#
# This file is part of cloud-init. See LICENSE file for license information.
import logging
import os
from typing import List

from cloudinit import distros, subp, util
from cloudinit.distros.package_management.apt import Apt
from cloudinit.distros.package_management.package_manager import PackageManager
from cloudinit.distros.parsers.hostname import HostnameConf
from cloudinit.net.netplan import CLOUDINIT_NETPLAN_FILE

LOG = logging.getLogger(__name__)

NETWORK_FILE_HEADER = """\
# This file is generated from information provided by the datasource.  Changes
# to it will not persist across an instance reboot.  To disable cloud-init's
# network configuration capabilities, write a file
# /etc/cloud/cloud.cfg.d/99-disable-network-config.cfg with the following:
# network: {config: disabled}
"""

LOCALE_CONF_FN = "/etc/default/locale"


class Distro(distros.Distro):
    hostname_conf_fn = "/etc/hostname"
    network_conf_fn = {
        "eni": "/etc/network/interfaces.d/50-cloud-init",
        "netplan": CLOUDINIT_NETPLAN_FILE,
    }
    renderer_configs = {
        "eni": {
            "eni_path": network_conf_fn["eni"],
            "eni_header": NETWORK_FILE_HEADER,
        },
        "netplan": {
            "netplan_path": network_conf_fn["netplan"],
            "netplan_header": NETWORK_FILE_HEADER,
            "postcmds": True,
        },
    }
    # Debian stores dhclient leases at following location:
    # /var/lib/dhcp/dhclient.<iface_name>.leases
    dhclient_lease_directory = "/var/lib/dhcp"
    dhclient_lease_file_regex = r"dhclient\.\w+\.leases"

    def __init__(self, name, cfg, paths):
        super().__init__(name, cfg, paths)
        # This will be used to restrict certain
        # calls from repeatedly happening (when they
        # should only happen say once per instance...)
        self.osfamily = "debian"
        self.default_locale = "C.UTF-8"
        self.system_locale = None
        self.apt = Apt.from_config(self._runner, cfg)
        self.package_managers: List[PackageManager] = [self.apt]

    def get_locale(self):
        """Return the default locale if set, else use default locale"""
        # read system locale value
        if not self.system_locale:
            self.system_locale = read_system_locale()

        # Return system_locale setting if valid, else use default locale
        return (
            self.system_locale if self.system_locale else self.default_locale
        )

    def apply_locale(self, locale, out_fn=None, keyname="LANG"):
        """Apply specified locale to system, regenerate if specified locale
        differs from system default."""
        if not out_fn:
            out_fn = LOCALE_CONF_FN

        if not locale:
            raise ValueError("Failed to provide locale value.")

        # Only call locale regeneration if needed
        # Update system locale config with specified locale if needed
        distro_locale = self.get_locale()
        conf_fn_exists = os.path.exists(out_fn)
        sys_locale_unset = not self.system_locale
        if sys_locale_unset:
            LOG.debug(
                "System locale not found in %s. "
                "Assuming system locale is %s based on hardcoded default",
                LOCALE_CONF_FN,
                self.default_locale,
            )
        else:
            LOG.debug(
                "System locale set to %s via %s",
                self.system_locale,
                LOCALE_CONF_FN,
            )
        need_regen = (
            locale.lower() != distro_locale.lower()
            or not conf_fn_exists
            or sys_locale_unset
        )
        need_conf = not conf_fn_exists or need_regen or sys_locale_unset

        if need_regen:
            regenerate_locale(
                locale,
                out_fn,
                keyname=keyname,
                install_function=self.install_packages,
            )
        else:
            LOG.debug(
                "System has '%s=%s' requested '%s', skipping regeneration.",
                keyname,
                self.system_locale,
                locale,
            )

        if need_conf:
            update_locale_conf(
                locale,
                out_fn,
                keyname=keyname,
                install_function=self.install_packages,
            )
            # once we've updated the system config, invalidate cache
            self.system_locale = None

    def _write_network_state(self, *args, **kwargs):
        _maybe_remove_legacy_eth0()
        return super()._write_network_state(*args, **kwargs)

    def _write_hostname(self, hostname, filename):
        conf = None
        try:
            # Try to update the previous one
            # so lets see if we can read it first.
            conf = self._read_hostname_conf(filename)
        except IOError:
            create_hostname_file = util.get_cfg_option_bool(
                self._cfg, "create_hostname_file", True
            )
            if create_hostname_file:
                pass
            else:
                LOG.info(
                    "create_hostname_file is False; hostname file not created"
                )
                return
        if not conf:
            conf = HostnameConf("")
        conf.set_hostname(hostname)
        util.write_file(filename, str(conf), 0o644)

    def _read_system_hostname(self):
        sys_hostname = self._read_hostname(self.hostname_conf_fn)
        return (self.hostname_conf_fn, sys_hostname)

    def _read_hostname_conf(self, filename):
        conf = HostnameConf(util.load_text_file(filename))
        conf.parse()
        return conf

    def _read_hostname(self, filename, default=None):
        hostname = None
        try:
            conf = self._read_hostname_conf(filename)
            hostname = conf.hostname
        except IOError:
            pass
        if not hostname:
            return default
        return hostname

    def _get_localhost_ip(self):
        # Note: http://www.leonardoborda.com/blog/127-0-1-1-ubuntu-debian/
        return "127.0.1.1"

    def set_timezone(self, tz):
        distros.set_etc_timezone(tz=tz, tz_file=self._find_tz_file(tz))

    def package_command(self, command, args=None, pkgs=None):
        # As of this writing, the only use of `package_command` outside of
        # distros calling it within their own classes is calling "upgrade"
        if command != "upgrade":
            raise RuntimeError(f"Unable to handle {command} command")
        self.apt.run_package_command("upgrade")

    def get_primary_arch(self):
        return util.get_dpkg_architecture()

    def set_keymap(self, layout: str, model: str, variant: str, options: str):
        # localectl is broken on some versions of Debian. See
        # https://bugs.launchpad.net/ubuntu/+source/systemd/+bug/2030788 and
        # https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=1038762
        #
        # Instead, write the file directly. According to the keyboard(5) man
        # page, this file is shared between both X and the console.

        contents = "\n".join(
            [
                "# This file was generated by cloud-init",
                "",
                f'XKBMODEL="{model}"',
                f'XKBLAYOUT="{layout}"',
                f'XKBVARIANT="{variant}"',
                f'XKBOPTIONS="{options}"',
                "",
                'BACKSPACE="guess"',  # This is provided on default installs
                "",
            ]
        )
        util.write_file(
            filename="/etc/default/keyboard",
            content=contents,
            mode=0o644,
            omode="w",
        )

        # Due to
        # https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=926037
        # if localectl can be used in the future, this line may still
        # be needed
        self.manage_service("restart", "console-setup")


def _maybe_remove_legacy_eth0(path="/etc/network/interfaces.d/eth0.cfg"):
    """Ubuntu cloud images previously included a 'eth0.cfg' that had
    hard coded content.  That file would interfere with the rendered
    configuration if it was present.

    if the file does not exist do nothing.
    If the file exists:
      - with known content, remove it and warn
      - with unknown content, leave it and warn
    """

    if not os.path.exists(path):
        return

    bmsg = "Dynamic networking config may not apply."
    try:
        contents = util.load_text_file(path)
        known_contents = ["auto eth0", "iface eth0 inet dhcp"]
        lines = [
            f.strip() for f in contents.splitlines() if not f.startswith("#")
        ]
        if lines == known_contents:
            util.del_file(path)
            msg = "removed %s with known contents" % path
        else:
            msg = bmsg + " '%s' exists with user configured content." % path
    except Exception:
        msg = bmsg + " %s exists, but could not be read." % path

    LOG.warning(msg)


def read_system_locale(sys_path=LOCALE_CONF_FN, keyname="LANG"):
    """Read system default locale setting, if present"""
    sys_val = ""
    if not sys_path:
        raise ValueError("Invalid path: %s" % sys_path)

    if os.path.exists(sys_path):
        locale_content = util.load_text_file(sys_path)
        sys_defaults = util.load_shell_content(locale_content)
        sys_val = sys_defaults.get(keyname, "")

    return sys_val


def update_locale_conf(
    locale, sys_path, keyname="LANG", install_function=None
):
    """Update system locale config"""
    LOG.debug(
        "Updating %s with locale setting %s=%s", sys_path, keyname, locale
    )
    if not subp.which("update-locale"):
        install_function(["locales"])
    subp.subp(
        [
            "update-locale",
            "--locale-file=" + sys_path,
            "%s=%s" % (keyname, locale),
        ],
        capture=False,
    )


def regenerate_locale(locale, sys_path, keyname="LANG", install_function=None):
    """
    Run locale-gen for the provided locale and set the default
    system variable `keyname` appropriately in the provided `sys_path`.

    """
    # special case for locales which do not require regen
    # % locale -a
    # C
    # C.UTF-8
    # POSIX
    if locale.lower() in ["c", "c.utf-8", "posix"]:
        LOG.debug("%s=%s does not require rengeneration", keyname, locale)
        return

    # finally, trigger regeneration
    if not subp.which("locale-gen"):
        install_function(["locales"])
    LOG.debug("Generating locales for %s", locale)
    subp.subp(["locale-gen", locale], capture=False)
Name	Size	Permission
__pycache__	---	0755
package_management	---	0755
parsers	---	0755
OpenCloudOS.py	277 bytes	0644
TencentOS.py	277 bytes	0644
__init__.py	67245 bytes	0644
almalinux.py	151 bytes	0644
alpine.py	24932 bytes	0644
amazon.py	907 bytes	0644
aosc.py	4482 bytes	0644
arch.py	5037 bytes	0644
azurelinux.py	2405 bytes	0644
bsd.py	6490 bytes	0644
bsd_utils.py	1440 bytes	0644
centos.py	151 bytes	0644
cloudlinux.py	151 bytes	0644
cos.py	247 bytes	0644
debian.py	10857 bytes	0644
dragonflybsd.py	230 bytes	0644
eurolinux.py	151 bytes	0644
fedora.py	437 bytes	0644
freebsd.py	8735 bytes	0644
gentoo.py	4990 bytes	0644
mariner.py	1740 bytes	0644
miraclelinux.py	151 bytes	0644
netbsd.py	5647 bytes	0644
networking.py	11070 bytes	0644
openbsd.py	2423 bytes	0644
openeuler.py	275 bytes	0644
openmandriva.py	237 bytes	0644
opensuse-leap.py	247 bytes	0644
opensuse-microos.py	247 bytes	0644
opensuse-tumbleweed.py	247 bytes	0644
opensuse.py	10201 bytes	0644
photon.py	5486 bytes	0644
rhel.py	8172 bytes	0644
rhel_util.py	1430 bytes	0644
rocky.py	151 bytes	0644
sle-micro.py	247 bytes	0644
sle_hpc.py	247 bytes	0644
sles.py	247 bytes	0644
suse.py	81 bytes	0644
ubuntu.py	2374 bytes	0644
ug_util.py	10001 bytes	0644
virtuozzo.py	151 bytes	0644
File Content: debian.py

File Content: `debian.py`
