403WebShell

403Webshell
Server IP : 13.127.148.211 / Your IP : 216.73.216.13
Web Server : Apache/2.4.41 (Ubuntu)
System : Linux ip-172-31-43-195 5.15.0-1084-aws #91~20.04.1-Ubuntu SMP Fri May 2 06:59:36 UTC 2025 x86_64
User : www-data ( 33)
PHP Version : 7.4.3-4ubuntu2.29
Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON
Directory : /lib/python3/dist-packages/cloudinit/config/
Upload File :
[ Back ]
Current File : /lib/python3/dist-packages/cloudinit/config//modules.py
# Copyright (C) 2008-2022 Canonical Ltd.
# Copyright (C) 2012 Hewlett-Packard Development Company, L.P.
#
# Author: Chuck Short <chuck.short@canonical.com>
# Author: Juerg Haefliger <juerg.haefliger@hp.com>
#
# This file is part of cloud-init. See LICENSE file for license information.

import copy
from types import ModuleType
from typing import Dict, List, NamedTuple, Optional

from cloudinit import config, importer
from cloudinit import log as logging
from cloudinit import type_utils, util
from cloudinit.distros import ALL_DISTROS
from cloudinit.helpers import ConfigMerger
from cloudinit.reporting.events import ReportEventStack
from cloudinit.settings import FREQUENCIES
from cloudinit.stages import Init

LOG = logging.getLogger(__name__)

# This prefix is used to make it less
# of a chance that when importing
# we will not find something else with the same
# name in the lookup path...
MOD_PREFIX = "cc_"


class ModuleDetails(NamedTuple):
    module: ModuleType
    name: str
    frequency: str
    run_args: List[str]


def form_module_name(name):
    canon_name = name.replace("-", "_")
    if canon_name.lower().endswith(".py"):
        canon_name = canon_name[0 : (len(canon_name) - 3)]
    canon_name = canon_name.strip()
    if not canon_name:
        return None
    if not canon_name.startswith(MOD_PREFIX):
        canon_name = "%s%s" % (MOD_PREFIX, canon_name)
    return canon_name


def validate_module(mod, name):
    if (
        not hasattr(mod, "meta")
        or "frequency" not in mod.meta
        or "distros" not in mod.meta
    ):
        raise ValueError(
            f"Module '{mod}' with name '{name}' MUST have a 'meta' attribute "
            "of type 'MetaSchema'."
        )
    if mod.meta["frequency"] not in FREQUENCIES:
        raise ValueError(
            f"Module '{mod}' with name '{name}' has an invalid frequency "
            f"{mod.meta['frequency']}."
        )
    if hasattr(mod, "schema"):
        raise ValueError(
            f"Module '{mod}' with name '{name}' has a JSON 'schema' attribute "
            "defined. Please define schema in cloud-init-schema,json."
        )


def _is_active(module_details: ModuleDetails, cfg: dict) -> bool:
    activate_by_schema_keys_keys = frozenset(
        module_details.module.meta.get("activate_by_schema_keys", {})
    )
    if not activate_by_schema_keys_keys:
        return True
    if not activate_by_schema_keys_keys.intersection(cfg.keys()):
        return False
    return True


class Modules:
    def __init__(self, init: Init, cfg_files=None, reporter=None):
        self.init = init
        self.cfg_files = cfg_files
        # Created on first use
        self._cached_cfg: Optional[config.Config] = None
        if reporter is None:
            reporter = ReportEventStack(
                name="module-reporter",
                description="module-desc",
                reporting_enabled=False,
            )
        self.reporter = reporter

    @property
    def cfg(self) -> config.Config:
        # None check to avoid empty case causing re-reading
        if self._cached_cfg is None:
            merger = ConfigMerger(
                paths=self.init.paths,
                datasource=self.init.datasource,
                additional_fns=self.cfg_files,
                base_cfg=self.init.cfg,
            )
            self._cached_cfg = merger.cfg
        # Only give out a copy so that others can't modify this...
        return copy.deepcopy(self._cached_cfg)

    def _read_modules(self, name) -> List[Dict]:
        """Read the modules from the config file given the specified name.

        Returns a list of module definitions. E.g.,
        [
            {
                "mod": "bootcmd",
                "freq": "always",
                "args": "some_arg",
            }
        ]

        Note that in the default case, only "mod" will be set.
        """
        module_list: List[dict] = []
        if name not in self.cfg:
            return module_list
        cfg_mods = self.cfg.get(name)
        if not cfg_mods:
            return module_list
        for item in cfg_mods:
            if not item:
                continue
            if isinstance(item, str):
                module_list.append(
                    {
                        "mod": item.strip(),
                    }
                )
            elif isinstance(item, (list)):
                contents = {}
                # Meant to fall through...
                if len(item) >= 1:
                    contents["mod"] = item[0].strip()
                if len(item) >= 2:
                    contents["freq"] = item[1].strip()
                if len(item) >= 3:
                    contents["args"] = item[2:]
                if contents:
                    module_list.append(contents)
            elif isinstance(item, (dict)):
                contents = {}
                valid = False
                if "name" in item:
                    contents["mod"] = item["name"].strip()
                    valid = True
                if "frequency" in item:
                    contents["freq"] = item["frequency"].strip()
                if "args" in item:
                    contents["args"] = item["args"] or []
                if contents and valid:
                    module_list.append(contents)
            else:
                raise TypeError(
                    "Failed to read '%s' item in config, unknown type %s"
                    % (item, type_utils.obj_name(item))
                )
        return module_list

    def _fixup_modules(self, raw_mods) -> List[ModuleDetails]:
        """Convert list of returned from _read_modules() into new format.

        Invalid modules and arguments are ignored.
        Also ensures that the module has the required meta fields.
        """
        mostly_mods = []
        for raw_mod in raw_mods:
            raw_name = raw_mod["mod"]
            freq = raw_mod.get("freq")
            run_args = raw_mod.get("args") or []
            mod_name = form_module_name(raw_name)
            if not mod_name:
                continue
            if freq and freq not in FREQUENCIES:
                LOG.warning(
                    "Config specified module %s has an unknown frequency %s",
                    raw_name,
                    freq,
                )
                # Misconfigured in /etc/cloud/cloud.cfg. Reset so cc_* module
                # default meta attribute "frequency" value is used.
                freq = None
            mod_locs, looked_locs = importer.find_module(
                mod_name, ["", type_utils.obj_name(config)], ["handle"]
            )
            if not mod_locs:
                LOG.warning(
                    "Could not find module named %s (searched %s)",
                    mod_name,
                    looked_locs,
                )
                continue
            mod = importer.import_module(mod_locs[0])
            validate_module(mod, raw_name)
            if freq is None:
                # Use cc_* module default setting since no cloud.cfg overrides
                freq = mod.meta["frequency"]
            mostly_mods.append(
                ModuleDetails(
                    module=mod,
                    name=raw_name,
                    frequency=freq,
                    run_args=run_args,
                )
            )
        return mostly_mods

    def _run_modules(self, mostly_mods: List[ModuleDetails]):
        cc = self.init.cloudify()
        # Return which ones ran
        # and which ones failed + the exception of why it failed
        failures = []
        which_ran = []
        for (mod, name, freq, args) in mostly_mods:
            try:
                LOG.debug(
                    "Running module %s (%s) with frequency %s", name, mod, freq
                )

                # Use the configs logger and not our own
                # TODO(harlowja): possibly check the module
                # for having a LOG attr and just give it back
                # its own logger?
                func_args = [name, self.cfg, cc, LOG, args]
                # Mark it as having started running
                which_ran.append(name)
                # This name will affect the semaphore name created
                run_name = f"config-{name}"

                desc = "running %s with frequency %s" % (run_name, freq)
                myrep = ReportEventStack(
                    name=run_name, description=desc, parent=self.reporter
                )

                with myrep:
                    ran, _r = cc.run(
                        run_name, mod.handle, func_args, freq=freq
                    )
                    if ran:
                        myrep.message = "%s ran successfully" % run_name
                    else:
                        myrep.message = "%s previously ran" % run_name

            except Exception as e:
                util.logexc(LOG, "Running module %s (%s) failed", name, mod)
                failures.append((name, e))
        return (which_ran, failures)

    def run_single(self, mod_name, args=None, freq=None):
        # Form the users module 'specs'
        mod_to_be = {
            "mod": mod_name,
            "args": args,
            "freq": freq,
        }
        # Now resume doing the normal fixups and running
        raw_mods = [mod_to_be]
        mostly_mods = self._fixup_modules(raw_mods)
        return self._run_modules(mostly_mods)

    def run_section(self, section_name):
        """Runs all modules in the given section.

        section_name - One of the modules lists as defined in
          /etc/cloud/cloud.cfg. One of:
         - cloud_init_modules
         - cloud_config_modules
         - cloud_final_modules
        """
        raw_mods = self._read_modules(section_name)
        mostly_mods = self._fixup_modules(raw_mods)
        distro_name = self.init.distro.name

        skipped = []
        forced = []
        overridden = self.cfg.get("unverified_modules", [])
        inapplicable_mods = []
        active_mods = []
        for module_details in mostly_mods:
            (mod, name, _freq, _args) = module_details
            if mod is None:
                continue
            worked_distros = mod.meta["distros"]
            if not _is_active(module_details, self.cfg):
                inapplicable_mods.append(name)
                continue
            # Skip only when the following conditions are all met:
            #  - distros are defined in the module != ALL_DISTROS
            #  - the current d_name isn't in distros
            #  - and the module is unverified and not in the unverified_modules
            #    override list
            if worked_distros and worked_distros != [ALL_DISTROS]:
                if distro_name not in worked_distros:
                    if name not in overridden:
                        skipped.append(name)
                        continue
                    forced.append(name)
            active_mods.append([mod, name, _freq, _args])

        if inapplicable_mods:
            LOG.info(
                "Skipping modules '%s' because no applicable config "
                "is provided.",
                ",".join(inapplicable_mods),
            )
        if skipped:
            LOG.info(
                "Skipping modules '%s' because they are not verified "
                "on distro '%s'.  To run anyway, add them to "
                "'unverified_modules' in config.",
                ",".join(skipped),
                distro_name,
            )
        if forced:
            LOG.info("running unverified_modules: '%s'", ", ".join(forced))

        return self._run_modules(active_mods)
Youez - 2016 - github.com/yon3zu
LinuXploit