# Copyright (c) 2017 Ansible Project
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import annotations

DOCUMENTATION = """
    name: ini
    version_added: "2.4"
    short_description: Uses an Ansible INI file as inventory source.
    description:
        - INI file based inventory, sections are groups or group related with special C(:modifiers).
        - Entries in sections C([group_1]) are hosts, members of the group.
        - Hosts can have variables defined inline as key/value pairs separated by C(=).
        - The C(children) modifier indicates that the section contains groups.
        - The C(vars) modifier indicates that the section contains variables assigned to members of the group.
        - Anything found outside a section is considered an 'ungrouped' host.
        - Values passed in the INI format using the C(key=value) syntax are interpreted differently depending on where they are declared within your inventory.
        - When declared inline with the host, INI values are processed by Python's ast.literal_eval function
          (U(https://docs.python.org/3/library/ast.html#ast.literal_eval)) and interpreted as Python literal structures
         (strings, numbers, tuples, lists, dicts, booleans, None). If you want a number to be treated as a string, you must quote it.
          Host lines accept multiple C(key=value) parameters per line.
          Therefore they need a way to indicate that a space is part of a value rather than a separator.
        - When declared in a C(:vars) section, INI values are interpreted as strings. For example C(var=FALSE) would create a string equal to C(FALSE).
          Unlike host lines, C(:vars) sections accept only a single entry per line, so everything after the C(=) must be the value for the entry.
        - Do not rely on types set during definition, always make sure you specify type with a filter when needed when consuming the variable.
        - See the Examples for proper quoting to prevent changes to variable type.
    notes:
        - Enabled in configuration by default.
        - Consider switching to YAML format for inventory sources to avoid confusion on the actual type of a variable.
          The YAML inventory plugin processes variable values consistently and correctly.
"""

EXAMPLES = """# fmt: ini
# Example 1
[web]
host1
host2 ansible_port=222 # defined inline, interpreted as an integer

[web:vars]
http_port=8080 # all members of 'web' will inherit these
myvar=23 # defined in a :vars section, interpreted as a string

[web:children] # child groups will automatically add their hosts to parent group
apache
nginx

[apache]
tomcat1
tomcat2 myvar=34 # host specific vars override group vars
tomcat3 mysecret="'03#pa33w0rd'" # proper quoting to prevent value changes

[nginx]
jenkins1

[nginx:vars]
has_java = True # vars in child groups override same in parent

[all:vars]
has_java = False # 'all' is 'top' parent

# Example 2
host1 # this is 'ungrouped'

# both hosts have same IP but diff ports, also 'ungrouped'
host2 ansible_host=127.0.0.1 ansible_port=44
host3 ansible_host=127.0.0.1 ansible_port=45

[g1]
host4

[g2]
host4 # same host as above, but member of 2 groups, will inherit vars from both
      # inventory hostnames are unique
"""

import ast
import os
import re
import typing as t
import warnings

from ansible.inventory.group import to_safe_group_name
from ansible.plugins.inventory import BaseFileInventoryPlugin

from ansible.errors import AnsibleError, AnsibleParserError
from ansible.module_utils.common.text.converters import to_bytes, to_text
from ansible._internal._datatag._tags import Origin, TrustedAsTemplate
from ansible.utils.shlex import shlex_split


class InventoryModule(BaseFileInventoryPlugin):
    """
    Takes an INI-format inventory file and builds a list of groups and subgroups
    with their associated hosts and variable settings.
    """
    NAME = 'ini'
    _COMMENT_MARKERS = frozenset((u';', u'#'))
    b_COMMENT_MARKERS = frozenset((b';', b'#'))

    # template trust is applied internally to strings

    def __init__(self) -> None:

        super(InventoryModule, self).__init__()

        self.patterns: dict[str, re.Pattern] = {}
        self._origin: Origin | None = None

    def verify_file(self, path):
        # hardcode exclusion for TOML to prevent partial parsing of things we know we don't want
        return super().verify_file(path) and os.path.splitext(path)[1] != '.toml'

    def parse(self, inventory, loader, path: str, cache=True):

        super(InventoryModule, self).parse(inventory, loader, path)

        try:
            # Read in the hosts, groups, and variables defined in the inventory file.
            if self.loader:
                (b_data, private) = self.loader._get_file_contents(path)
            else:
                b_path = to_bytes(path, errors='surrogate_or_strict')
                with open(b_path, 'rb') as fh:
                    b_data = fh.read()

            try:
                # Faster to do to_text once on a long string than many
                # times on smaller strings
                data = to_text(b_data, errors='surrogate_or_strict').splitlines()
            except UnicodeError:
                # Handle non-utf8 in comment lines: https://github.com/ansible/ansible/issues/17593
                data = []
                for line in b_data.splitlines():
                    if line and line[0] in self.b_COMMENT_MARKERS:
                        # Replace is okay for comment lines
                        # data.append(to_text(line, errors='surrogate_then_replace'))
                        # Currently we only need these lines for accurate lineno in errors
                        data.append(u'')
                    else:
                        # Non-comment lines still have to be valid uf-8
                        data.append(to_text(line, errors='surrogate_or_strict'))

            self._origin = Origin(path=path, line_num=0)

            try:
                self._parse(data)
            finally:
                self._origin = self._origin.replace(line_num=None)

        except Exception as ex:
            raise AnsibleParserError('Failed to parse inventory.', obj=self._origin) from ex

    def _raise_error(self, message):
        raise AnsibleError(message)

    def _parse(self, lines):
        """
        Populates self.groups from the given array of lines. Raises an error on
        any parse failure.
        """

        self._compile_patterns()

        # We behave as though the first line of the inventory is '[ungrouped]',
        # and begin to look for host definitions. We make a single pass through
        # each line of the inventory, building up self.groups and adding hosts,
        # subgroups, and setting variables as we go.

        pending_declarations = {}
        groupname = 'ungrouped'
        state = 'hosts'
        for line in lines:
            self._origin = self._origin.replace(line_num=self._origin.line_num + 1)

            line = line.strip()
            # Skip empty lines and comments
            if not line or line[0] in self._COMMENT_MARKERS:
                continue

            # Is this a [section] header? That tells us what group we're parsing
            # definitions for, and what kind of definitions to expect.

            m = self.patterns['section'].match(line)
            if m:
                (groupname, state) = m.groups()

                groupname = to_safe_group_name(groupname)

                state = state or 'hosts'
                if state not in ['hosts', 'children', 'vars']:
                    title = ":".join(m.groups())
                    self._raise_error("Section [%s] has unknown type: %s" % (title, state))

                # If we haven't seen this group before, we add a new Group.
                if groupname not in self.inventory.groups:
                    # Either [groupname] or [groupname:children] is sufficient to declare a group,
                    # but [groupname:vars] is allowed only if the # group is declared elsewhere.
                    # We add the group anyway, but make a note in pending_declarations to check at the end.
                    #
                    # It's possible that a group is previously pending due to being defined as a child
                    # group, in that case we simply pass so that the logic below to process pending
                    # declarations will take the appropriate action for a pending child group instead of
                    # incorrectly handling it as a var state pending declaration
                    if state == 'vars' and groupname not in pending_declarations:
                        pending_declarations[groupname] = dict(line=self._origin.line_num, state=state, name=groupname)

                    self.inventory.add_group(groupname)

                # When we see a declaration that we've been waiting for, we process and delete.
                if groupname in pending_declarations and state != 'vars':
                    if pending_declarations[groupname]['state'] == 'children':
                        self._add_pending_children(groupname, pending_declarations)
                    elif pending_declarations[groupname]['state'] == 'vars':
                        del pending_declarations[groupname]

                continue
            elif line.startswith('[') and line.endswith(']'):
                self._raise_error("Invalid section entry: '%s'. Please make sure that there are no spaces" % line + " " +
                                  "in the section entry, and that there are no other invalid characters")

            # It's not a section, so the current state tells us what kind of
            # definition it must be. The individual parsers will raise an
            # error if we feed them something they can't digest.

            # [groupname] contains host definitions that must be added to
            # the current group.
            if state == 'hosts':
                hosts, port, variables = self._parse_host_definition(line)
                self._populate_host_vars(hosts, variables, groupname, port)

            # [groupname:vars] contains variable definitions that must be
            # applied to the current group.
            elif state == 'vars':
                (k, v) = self._parse_variable_definition(line)
                self.inventory.set_variable(groupname, k, v)

            # [groupname:children] contains subgroup names that must be
            # added as children of the current group. The subgroup names
            # must themselves be declared as groups, but as before, they
            # may only be declared later.
            elif state == 'children':
                child = self._parse_group_name(line)
                if child not in self.inventory.groups:
                    if child not in pending_declarations:
                        pending_declarations[child] = dict(line=self._origin.line_num, state=state, name=child, parents=[groupname])
                    else:
                        pending_declarations[child]['parents'].append(groupname)
                else:
                    self.inventory.add_child(groupname, child)
            else:
                # This can happen only if the state checker accepts a state that isn't handled above.
                self._raise_error("Entered unhandled state: %s" % (state))

        # Any entries in pending_declarations not removed by a group declaration above mean that there was an unresolved reference.
        # We report only the first such error here.
        for g in pending_declarations:
            decl = pending_declarations[g]
            self._origin = self._origin.replace(line_num=decl['line'])
            if decl['state'] == 'vars':
                raise ValueError(f"Section [{decl['name']}:vars] not valid for undefined group {decl['name']!r}.")
            elif decl['state'] == 'children':
                raise ValueError(f"Section [{decl['parents'][-1]}:children] includes undefined group {decl['name']!r}.")

    def _add_pending_children(self, group, pending):
        for parent in pending[group]['parents']:
            self.inventory.add_child(parent, group)
            if parent in pending and pending[parent]['state'] == 'children':
                self._add_pending_children(parent, pending)
        del pending[group]

    def _parse_group_name(self, line):
        """
        Takes a single line and tries to parse it as a group name. Returns the
        group name if successful, or raises an error.
        """

        m = self.patterns['groupname'].match(line)
        if m:
            return m.group(1)

        self._raise_error("Expected group name, got: %s" % (line))

    def _parse_variable_definition(self, line):
        """
        Takes a string and tries to parse it as a variable definition. Returns
        the key and value if successful, or raises an error.
        """

        # TODO: We parse variable assignments as a key (anything to the left of
        # an '='"), an '=', and a value (anything left) and leave the value to
        # _parse_value to sort out. We should be more systematic here about
        # defining what is acceptable, how quotes work, and so on.

        if '=' in line:
            (k, v) = [e.strip() for e in line.split("=", 1)]
            return (self._origin.tag(k), self._parse_value(v))

        self._raise_error("Expected key=value, got: %s" % (line))

    def _parse_host_definition(self, line):
        """
        Takes a single line and tries to parse it as a host definition. Returns
        a list of Hosts if successful, or raises an error.
        """

        # A host definition comprises (1) a non-whitespace hostname or range,
        # optionally followed by (2) a series of key="some value" assignments.
        # We ignore any trailing whitespace and/or comments. For example, here
        # are a series of host definitions in a group:
        #
        # [groupname]
        # alpha
        # beta:2345 user=admin      # we'll tell shlex
        # gamma sudo=True user=root # to ignore comments

        try:
            tokens = shlex_split(line, comments=True)
        except ValueError as e:
            self._raise_error("Error parsing host definition '%s': %s" % (line, e))

        (hostnames, port) = self._expand_hostpattern(tokens[0])

        # Try to process anything remaining as a series of key=value pairs.
        variables = {}
        for t in tokens[1:]:
            if '=' not in t:
                self._raise_error("Expected key=value host variable assignment, got: %s" % (t))
            (k, v) = t.split('=', 1)
            variables[self._origin.tag(k)] = self._parse_value(v)

        return hostnames, port, variables

    def _expand_hostpattern(self, hostpattern):
        """
        do some extra checks over normal processing
        """
        # specification?

        hostnames, port = super(InventoryModule, self)._expand_hostpattern(hostpattern)

        if hostpattern.strip().endswith(':') and port is None:
            raise AnsibleParserError("Invalid host pattern '%s' supplied, ending in ':' is not allowed, this character is reserved to provide a port." %
                                     hostpattern)
        for pattern in hostnames:
            # some YAML parsing prevention checks
            if pattern.strip() == '---':
                raise AnsibleParserError("Invalid host pattern '%s' supplied, '---' is normally a sign this is a YAML file." % hostpattern)

        return (hostnames, port)

    def _parse_recursive_coerce_types_and_tag(self, value: t.Any) -> t.Any:
        if isinstance(value, str):
            return TrustedAsTemplate().tag(self._origin.tag(value))
        if isinstance(value, (list, tuple, set)):
            # NB: intentional coercion of tuple/set to list, deal with it
            return self._origin.tag([self._parse_recursive_coerce_types_and_tag(v) for v in value])
        if isinstance(value, dict):
            # FIXME: enforce keys are strings
            return self._origin.tag({self._origin.tag(k): self._parse_recursive_coerce_types_and_tag(v) for k, v in value.items()})

        if value is ...:  # literal_eval parses ellipsis, but it's not a supported variable type
            value = TrustedAsTemplate().tag("...")

        if isinstance(value, complex):  # convert unsupported variable types recognized by literal_eval back to str
            value = TrustedAsTemplate().tag(str(value))

        value = to_text(value, nonstring='passthru', errors='surrogate_or_strict')

        return self._origin.tag(value)

    def _parse_value(self, v: str) -> t.Any:
        """
        Attempt to transform the string value from an ini file into a basic python object
        (int, dict, list, unicode string, etc).
        """
        try:
            with warnings.catch_warnings():
                warnings.simplefilter("ignore", SyntaxWarning)
                v = ast.literal_eval(v)
        # Using explicit exceptions.
        # Likely a string that literal_eval does not like. We will then just set it.
        except ValueError:
            # For some reason this was thought to be malformed.
            pass
        except SyntaxError:
            # Is this a hash with an equals at the end?
            pass

        # this is mostly unnecessary, but prevents the (possible) case of bytes literals showing up in inventory
        return self._parse_recursive_coerce_types_and_tag(v)

    def _compile_patterns(self):
        """
        Compiles the regular expressions required to parse the inventory and
        stores them in self.patterns.
        """

        # Section names are square-bracketed expressions at the beginning of a
        # line, comprising (1) a group name optionally followed by (2) a tag
        # that specifies the contents of the section. We ignore any trailing
        # whitespace and/or comments. For example:
        #
        # [groupname]
        # [somegroup:vars]
        # [naughty:children] # only get coal in their stockings

        self.patterns['section'] = re.compile(
            to_text(r"""^\[
                    ([^:\]\s]+)             # group name (see groupname below)
                    (?::(\w+))?             # optional : and tag name
                \]
                \s*                         # ignore trailing whitespace
                (?:\#.*)?                   # and/or a comment till the
                $                           # end of the line
            """, errors='surrogate_or_strict'), re.X
        )

        # FIXME: What are the real restrictions on group names, or rather, what
        # should they be? At the moment, they must be non-empty sequences of non
        # whitespace characters excluding ':' and ']', but we should define more
        # precise rules in order to support better diagnostics.

        self.patterns['groupname'] = re.compile(
            to_text(r"""^
                ([^:\]\s]+)
                \s*                         # ignore trailing whitespace
                (?:\#.*)?                   # and/or a comment till the
                $                           # end of the line
            """, errors='surrogate_or_strict'), re.X
        )