ACIL FM

Pilih semua

Nama	Ukuran	Aksi
arguments	-	Open
scripts	-	Open
__pycache__	-	Open
adhoc.py	8.04 MB	View DL Edit
config.py	21.63 MB	View DL Edit
console.py	21.43 MB	View DL Edit
doc.py	59.02 MB	View DL Edit
galaxy.py	90.78 MB	View DL Edit
inventory.py	16.03 MB	View DL Edit
playbook.py	10.25 MB	View DL Edit
pull.py	16.63 MB	View DL Edit
vault.py	21.88 MB	View DL Edit
__init__.py	27.3 MB	View DL Edit

Edit file: /usr/lib/python3.9/site-packages/ansible/cli/config.py

#!/usr/bin/env python
# Copyright: (c) 2017, Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
# PYTHON_ARGCOMPLETE_OK

from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

# ansible.cli needs to be imported first, to ensure the source bin/* scripts run that code first
from ansible.cli import CLI

import os
import yaml
import shlex
import subprocess

from collections.abc import Mapping

from ansible import context
import ansible.plugins.loader as plugin_loader

from ansible import constants as C
from ansible.cli.arguments import option_helpers as opt_help
from ansible.config.manager import ConfigManager, Setting
from ansible.errors import AnsibleError, AnsibleOptionsError
from ansible.module_utils._text import to_native, to_text, to_bytes
from ansible.module_utils.common.json import json_dump
from ansible.module_utils.six import string_types
from ansible.parsing.quoting import is_quoted
from ansible.parsing.yaml.dumper import AnsibleDumper
from ansible.utils.color import stringc
from ansible.utils.display import Display
from ansible.utils.path import unfrackpath

display = Display()

def yaml_dump(data, default_flow_style=False, default_style=None):
    return yaml.dump(data, Dumper=AnsibleDumper, default_flow_style=default_flow_style, default_style=default_style)

def yaml_short(data):
    return yaml_dump(data, default_flow_style=True, default_style="''")

def get_constants():
    ''' helper method to ensure we can template based on existing constants '''
    if not hasattr(get_constants, 'cvars'):
        get_constants.cvars = {k: getattr(C, k) for k in dir(C) if not k.startswith('__')}
    return get_constants.cvars

class ConfigCLI(CLI):
    """ Config command line class """

name = 'ansible-config'

def __init__(self, args, callback=None):

self.config_file = None
        self.config = None
        super(ConfigCLI, self).__init__(args, callback)

def init_parser(self):

super(ConfigCLI, self).init_parser(
            desc="View ansible configuration.",
        )

common = opt_help.argparse.ArgumentParser(add_help=False)
        opt_help.add_verbosity_options(common)
        common.add_argument('-c', '--config', dest='config_file',
                            help="path to configuration file, defaults to first file found in precedence.")
        common.add_argument("-t", "--type", action="store", default='base', dest='type', choices=['all', 'base'] + list(C.CONFIGURABLE_PLUGINS),
                            help="Filter down to a specific plugin type.")
        common.add_argument('args', help='Specific plugin to target, requires type of plugin to be set', nargs='*')

subparsers = self.parser.add_subparsers(dest='action')
        subparsers.required = True

list_parser = subparsers.add_parser('list', help='Print all config options', parents=[common])
        list_parser.set_defaults(func=self.execute_list)
        list_parser.add_argument('--format', '-f', dest='format', action='store', choices=['json', 'yaml'], default='yaml',
                                 help='Output format for list')

dump_parser = subparsers.add_parser('dump', help='Dump configuration', parents=[common])
        dump_parser.set_defaults(func=self.execute_dump)
        dump_parser.add_argument('--only-changed', '--changed-only', dest='only_changed', action='store_true',
                                 help="Only show configurations that have changed from the default")
        dump_parser.add_argument('--format', '-f', dest='format', action='store', choices=['json', 'yaml', 'display'], default='display',
                                 help='Output format for dump')

view_parser = subparsers.add_parser('view', help='View configuration file', parents=[common])
        view_parser.set_defaults(func=self.execute_view)

init_parser = subparsers.add_parser('init', help='Create initial configuration', parents=[common])
        init_parser.set_defaults(func=self.execute_init)
        init_parser.add_argument('--format', '-f', dest='format', action='store', choices=['ini', 'env', 'vars'], default='ini',
                                 help='Output format for init')
        init_parser.add_argument('--disabled', dest='commented', action='store_true', default=False,
                                 help='Prefixes all entries with a comment character to disable them')

# search_parser = subparsers.add_parser('find', help='Search configuration')
        # search_parser.set_defaults(func=self.execute_search)
        # search_parser.add_argument('args', help='Search term', metavar='<search term>')

def post_process_args(self, options):
        options = super(ConfigCLI, self).post_process_args(options)
        display.verbosity = options.verbosity

return options

def run(self):

super(ConfigCLI, self).run()

if context.CLIARGS['config_file']:
            self.config_file = unfrackpath(context.CLIARGS['config_file'], follow=False)
            b_config = to_bytes(self.config_file)
            if os.path.exists(b_config) and os.access(b_config, os.R_OK):
                self.config = ConfigManager(self.config_file)
            else:
                raise AnsibleOptionsError('The provided configuration file is missing or not accessible: %s' % to_native(self.config_file))
        else:
            self.config = C.config
            self.config_file = self.config._config_file

if self.config_file:
            try:
                if not os.path.exists(self.config_file):
                    raise AnsibleOptionsError("%s does not exist or is not accessible" % (self.config_file))
                elif not os.path.isfile(self.config_file):
                    raise AnsibleOptionsError("%s is not a valid file" % (self.config_file))

os.environ['ANSIBLE_CONFIG'] = to_native(self.config_file)
            except Exception:
                if context.CLIARGS['action'] in ['view']:
                    raise
                elif context.CLIARGS['action'] in ['edit', 'update']:
                    display.warning("File does not exist, used empty file: %s" % self.config_file)

elif context.CLIARGS['action'] == 'view':
            raise AnsibleError('Invalid or no config file was supplied')

# run the requested action
        context.CLIARGS['func']()

def execute_update(self):
        '''
        Updates a single setting in the specified ansible.cfg
        '''
        raise AnsibleError("Option not implemented yet")

# pylint: disable=unreachable
        if context.CLIARGS['setting'] is None:
            raise AnsibleOptionsError("update option requires a setting to update")

(entry, value) = context.CLIARGS['setting'].split('=')
        if '.' in entry:
            (section, option) = entry.split('.')
        else:
            section = 'defaults'
            option = entry
        subprocess.call([
            'ansible',
            '-m', 'ini_file',
            'localhost',
            '-c', 'local',
            '-a', '"dest=%s section=%s option=%s value=%s backup=yes"' % (self.config_file, section, option, value)
        ])

def execute_view(self):
        '''
        Displays the current config file
        '''
        try:
            with open(self.config_file, 'rb') as f:
                self.pager(to_text(f.read(), errors='surrogate_or_strict'))
        except Exception as e:
            raise AnsibleError("Failed to open config file: %s" % to_native(e))

def execute_edit(self):
        '''
        Opens ansible.cfg in the default EDITOR
        '''
        raise AnsibleError("Option not implemented yet")

# pylint: disable=unreachable
        try:
            editor = shlex.split(os.environ.get('EDITOR', 'vi'))
            editor.append(self.config_file)
            subprocess.call(editor)
        except Exception as e:
            raise AnsibleError("Failed to open editor: %s" % to_native(e))

def _list_plugin_settings(self, ptype, plugins=None):
        entries = {}
        loader = getattr(plugin_loader, '%s_loader' % ptype)

# build list
        if plugins:
            plugin_cs = []
            for plugin in plugins:
                p = loader.get(plugin, class_only=True)
                if p is None:
                    display.warning("Skipping %s as we could not find matching plugin" % plugin)
                else:
                    plugin_cs.append(p)
        else:
            plugin_cs = loader.all(class_only=True)

# iterate over class instances
        for plugin in plugin_cs:
            finalname = name = plugin._load_name
            if name.startswith('_'):
                # alias or deprecated
                if os.path.islink(plugin._original_path):
                    continue
                else:
                    finalname = name.replace('_', '', 1) + ' (DEPRECATED)'

entries[finalname] = self.config.get_configuration_definitions(ptype, name)

return entries

def _list_entries_from_args(self):
        '''
        build a dict with the list requested configs
        '''
        config_entries = {}
        if context.CLIARGS['type'] in ('base', 'all'):
            # this dumps main/common configs
            config_entries = self.config.get_configuration_definitions(ignore_private=True)

if context.CLIARGS['type'] != 'base':
            config_entries['PLUGINS'] = {}

if context.CLIARGS['type'] == 'all':
            # now each plugin type
            for ptype in C.CONFIGURABLE_PLUGINS:
                config_entries['PLUGINS'][ptype.upper()] = self._list_plugin_settings(ptype)
        elif context.CLIARGS['type'] != 'base':
            config_entries['PLUGINS'][context.CLIARGS['type']] = self._list_plugin_settings(context.CLIARGS['type'], context.CLIARGS['args'])

return config_entries

def execute_list(self):
        '''
        list and output available configs
        '''

config_entries = self._list_entries_from_args()
        if context.CLIARGS['format'] == 'yaml':
            output = yaml_dump(config_entries)
        elif context.CLIARGS['format'] == 'json':
            output = json_dump(config_entries)

self.pager(to_text(output, errors='surrogate_or_strict'))

def _get_settings_vars(self, settings, subkey):

data = []
        if context.CLIARGS['commented']:
            prefix = '#'
        else:
            prefix = ''

for setting in settings:

if not settings[setting].get('description'):
                continue

default = settings[setting].get('default', '')
            if subkey == 'env':
                stype = settings[setting].get('type', '')
                if stype == 'boolean':
                    if default:
                        default = '1'
                    else:
                        default = '0'
                elif default:
                    if stype == 'list':
                        if not isinstance(default, string_types):
                            # python lists are not valid env ones
                            try:
                                default = ', '.join(default)
                            except Exception as e:
                                # list of other stuff
                                default = '%s' % to_native(default)
                    if isinstance(default, string_types) and not is_quoted(default):
                        default = shlex.quote(default)
                elif default is None:
                    default = ''

if subkey in settings[setting] and settings[setting][subkey]:
                entry = settings[setting][subkey][-1]['name']
                if isinstance(settings[setting]['description'], string_types):
                    desc = settings[setting]['description']
                else:
                    desc = '\n#'.join(settings[setting]['description'])
                name = settings[setting].get('name', setting)
                data.append('# %s(%s): %s' % (name, settings[setting].get('type', 'string'), desc))

# TODO: might need quoting and value coercion depending on type
                if subkey == 'env':
                    if entry.startswith('_ANSIBLE_'):
                        continue
                    data.append('%s%s=%s' % (prefix, entry, default))
                elif subkey == 'vars':
                    if entry.startswith('_ansible_'):
                        continue
                    data.append(prefix + '%s: %s' % (entry, to_text(yaml_short(default), errors='surrogate_or_strict')))
                data.append('')

return data

def _get_settings_ini(self, settings):

sections = {}
        for o in sorted(settings.keys()):

opt = settings[o]

if not isinstance(opt, Mapping):
                # recursed into one of the few settings that is a mapping, now hitting it's strings
                continue

if not opt.get('description'):
                # its a plugin
                new_sections = self._get_settings_ini(opt)
                for s in new_sections:
                    if s in sections:
                        sections[s].extend(new_sections[s])
                    else:
                        sections[s] = new_sections[s]
                continue

if isinstance(opt['description'], string_types):
                desc = '# (%s) %s' % (opt.get('type', 'string'), opt['description'])
            else:
                desc = "# (%s) " % opt.get('type', 'string')
                desc += "\n# ".join(opt['description'])

if 'ini' in opt and opt['ini']:
                entry = opt['ini'][-1]
                if entry['section'] not in sections:
                    sections[entry['section']] = []

default = opt.get('default', '')
                if opt.get('type', '') == 'list' and not isinstance(default, string_types):
                    # python lists are not valid ini ones
                    default = ', '.join(default)
                elif default is None:
                    default = ''

if context.CLIARGS['commented']:
                    entry['key'] = ';%s' % entry['key']

key = desc + '\n%s=%s' % (entry['key'], default)
                sections[entry['section']].append(key)

return sections

def execute_init(self):
        """Create initial configuration"""

data = []
        config_entries = self._list_entries_from_args()
        plugin_types = config_entries.pop('PLUGINS', None)

if context.CLIARGS['format'] == 'ini':
            sections = self._get_settings_ini(config_entries)

if plugin_types:
                for ptype in plugin_types:
                    plugin_sections = self._get_settings_ini(plugin_types[ptype])
                    for s in plugin_sections:
                        if s in sections:
                            sections[s].extend(plugin_sections[s])
                        else:
                            sections[s] = plugin_sections[s]

if sections:
                for section in sections.keys():
                    data.append('[%s]' % section)
                    for key in sections[section]:
                        data.append(key)
                        data.append('')
                    data.append('')

elif context.CLIARGS['format'] in ('env', 'vars'):  # TODO: add yaml once that config option is added
            data = self._get_settings_vars(config_entries, context.CLIARGS['format'])
            if plugin_types:
                for ptype in plugin_types:
                    for plugin in plugin_types[ptype].keys():
                        data.extend(self._get_settings_vars(plugin_types[ptype][plugin], context.CLIARGS['format']))

self.pager(to_text('\n'.join(data), errors='surrogate_or_strict'))

def _render_settings(self, config):

entries = []
        for setting in sorted(config):
            changed = (config[setting].origin not in ('default', 'REQUIRED'))

if context.CLIARGS['format'] == 'display':
                if isinstance(config[setting], Setting):
                    # proceed normally
                    if config[setting].origin == 'default':
                        color = 'green'
                    elif config[setting].origin == 'REQUIRED':
                        # should include '_terms', '_input', etc
                        color = 'red'
                    else:
                        color = 'yellow'
                    msg = "%s(%s) = %s" % (setting, config[setting].origin, config[setting].value)
                else:
                    color = 'green'
                    msg = "%s(%s) = %s" % (setting, 'default', config[setting].get('default'))

entry = stringc(msg, color)
            else:
                entry = {}
                for key in config[setting]._fields:
                    entry[key] = getattr(config[setting], key)

if not context.CLIARGS['only_changed'] or changed:
                entries.append(entry)

return entries

def _get_global_configs(self):
        config = self.config.get_configuration_definitions(ignore_private=True).copy()
        for setting in config.keys():
            v, o = C.config.get_config_value_and_origin(setting, cfile=self.config_file, variables=get_constants())
            config[setting] = Setting(setting, v, o, None)

return self._render_settings(config)

def _get_plugin_configs(self, ptype, plugins):

# prep loading
        loader = getattr(plugin_loader, '%s_loader' % ptype)

# acumulators
        output = []
        config_entries = {}

# build list
        if plugins:
            plugin_cs = []
            for plugin in plugins:
                p = loader.get(plugin, class_only=True)
                if p is None:
                    display.warning("Skipping %s as we could not find matching plugin" % plugin)
                else:
                    plugin_cs.append(loader.get(plugin, class_only=True))
        else:
            plugin_cs = loader.all(class_only=True)

for plugin in plugin_cs:
            # in case of deprecastion they diverge
            finalname = name = plugin._load_name
            if name.startswith('_'):
                if os.path.islink(plugin._original_path):
                    # skip alias
                    continue
                # deprecated, but use 'nice name'
                finalname = name.replace('_', '', 1) + ' (DEPRECATED)'

# default entries per plugin
            config_entries[finalname] = self.config.get_configuration_definitions(ptype, name)

try:
                # populate config entries by loading plugin
                dump = loader.get(name, class_only=True)
            except Exception as e:
                display.warning('Skipping "%s" %s plugin, as we cannot load plugin to check config due to : %s' % (name, ptype, to_native(e)))
                continue

# actually get the values
            for setting in config_entries[finalname].keys():
                try:
                    v, o = C.config.get_config_value_and_origin(setting, cfile=self.config_file, plugin_type=ptype, plugin_name=name, variables=get_constants())
                except AnsibleError as e:
                    if to_text(e).startswith('No setting was provided for required configuration'):
                        v = None
                        o = 'REQUIRED'
                    else:
                        raise e

if v is None and o is None:
                    # not all cases will be error
                    o = 'REQUIRED'

config_entries[finalname][setting] = Setting(setting, v, o, None)

# pretty please!
            results = self._render_settings(config_entries[finalname])
            if results:
                if context.CLIARGS['format'] == 'display':
                    # avoid header for empty lists (only changed!)
                    output.append('\n%s:\n%s' % (finalname, '_' * len(finalname)))
                    output.extend(results)
                else:
                    output.append({finalname: results})

return output

def execute_dump(self):
        '''
        Shows the current settings, merges ansible.cfg if specified
        '''
        if context.CLIARGS['type'] == 'base':
            # deal with base
            output = self._get_global_configs()
        elif context.CLIARGS['type'] == 'all':
            # deal with base
            output = self._get_global_configs()
            # deal with plugins
            for ptype in C.CONFIGURABLE_PLUGINS:
                plugin_list = self._get_plugin_configs(ptype, context.CLIARGS['args'])
                if context.CLIARGS['format'] == 'display':
                    if not context.CLIARGS['only_changed'] or plugin_list:
                        output.append('\n%s:\n%s' % (ptype.upper(), '=' * len(ptype)))
                        output.extend(plugin_list)
                else:
                    if ptype in ('modules', 'doc_fragments'):
                        pname = ptype.upper()
                    else:
                        pname = '%s_PLUGINS' % ptype.upper()
                    output.append({pname: plugin_list})
        else:
            # deal with plugins
            output = self._get_plugin_configs(context.CLIARGS['type'], context.CLIARGS['args'])

if context.CLIARGS['format'] == 'display':
            text = '\n'.join(output)
        if context.CLIARGS['format'] == 'yaml':
            text = yaml_dump(output)
        elif context.CLIARGS['format'] == 'json':
            text = json_dump(output)

self.pager(to_text(text, errors='surrogate_or_strict'))

def main(args=None):
    ConfigCLI.cli_executor(args)

if __name__ == '__main__':
    main()

Batal

Isi Zip:

Create