<filename>ubiquiti_config_generator/root_parser.py<gh_stars>1-10
"""
Contains the root configuration node
"""
import ipaddress
from os import path
from typing import List, Tuple

from ubiquiti_config_generator import file_paths, secondary_configs
from ubiquiti_config_generator.nodes import (
    GlobalSettings,
    PortGroup,
    ExternalAddresses,
    NAT,
    Network,
)


class RootNode:
    """
    Represents the root config node, from which everything else is based off of
    """

    # pylint: disable=too-many-arguments
    def __init__(
        self,
        global_settings: GlobalSettings,
        port_groups: List[PortGroup],
        external_addresses: ExternalAddresses,
        networks: List[Network],
        nat: NAT,
    ):
        self.global_settings = global_settings
        self.port_groups = port_groups
        self.external_addresses = external_addresses
        self.networks = networks
        self.nat = nat

    @classmethod
    def create_from_configs(cls, config_path: str):
        """
        Load configuration from files
        """
        nat = NAT(config_path)
        return cls(
            secondary_configs.get_global_configuration(config_path),
            secondary_configs.get_port_groups(config_path),
            secondary_configs.get_external_addresses(config_path),
            [
                Network(
                    network_folder.split(path.sep)[-2],
                    nat,
                    config_path,
                    **(file_paths.load_yaml_from_file(network_folder))
                )
                for network_folder in file_paths.get_folders_with_config(
                    [config_path, file_paths.NETWORK_FOLDER]
                )
            ],
            nat,
        )

    def is_valid(self) -> bool:
        """
        Are all fields in the configuration valid
        """
        settings_valid = self.global_settings.validate()
        addresses_valid = self.external_addresses.validate()
        ports_valid = all([port.validate() for port in self.port_groups])
        networks_valid = all([network.validate() for network in self.networks])
        nat_valid = self.nat.validate()

        return (
            settings_valid
            and addresses_valid
            and ports_valid
            and networks_valid
            and nat_valid
        )

    def is_consistent(self) -> bool:
        """
        Check configuration for consistency
        """
        addresses_consistent = self.external_addresses.is_consistent()
        globals_consistent = self.global_settings.is_consistent()
        port_groups_consistent = [group.is_consistent() for group in self.port_groups]
        networks_consistent = [network.is_consistent() for network in self.networks]
        nat_consistent = self.nat.is_consistent()

        networks_consistent = all(networks_consistent) and True

        network_count = len(self.networks)
        for network_index in range(network_count):
            network = self.networks[network_index]

            # Skip any DHCP networks, e.g. for WAN
            if network.cidr is None:
                continue

            ip_network = ipaddress.ip_network(network.cidr)
            for second_network_index in range(network_index + 1, network_count):
                second_network = self.networks[second_network_index]

                # Skip any DHCP networks, e.g. for WAN
                if second_network.cidr is None:
                    continue

                second_ip_network = ipaddress.ip_network(second_network.cidr)

                if ip_network.overlaps(second_ip_network):
                    network.add_validation_error(
                        "{0} overlaps with {1}".format(
                            str(network), str(second_network)
                        )
                    )
                    networks_consistent = False

        return (
            addresses_consistent
            and globals_consistent
            and all(port_groups_consistent)
            and networks_consistent
            and nat_consistent
        )

    def validate(self) -> bool:
        """
        Is the root node valid
        """
        # Ensure both are checked, for complete error messages
        valid = self.is_valid()
        consistent = self.is_consistent()
        return valid and consistent

    def validation_failures(self) -> List[str]:
        """
        Get all validation failures
        """
        failures = (
            self.global_settings.validation_errors()
            + self.external_addresses.validation_errors()
            + self.nat.validation_failures()
        )
        for port in self.port_groups:
            failures.extend(port.validation_errors())
        for network in self.networks:
            failures.extend(network.validation_failures())

        return failures

    def get_commands(self) -> Tuple[List[List[str]], List[str]]:
        """
        Returns the commands to generate this configuration

        First value is an ordered list of commands to run, segmented into
        distinct portions which need to be committed in-order
        Second value is a flat list of all commands, for comparison against
        the previous configuration, to check for needed deletions
        """
        # These 3 should just be a list of commands, since ordering won't matter
        external_addresses = self.external_addresses.commands()
        port_groups = []
        for group in self.port_groups:
            port_groups.extend(group.commands())
        global_settings = self.global_settings.commands()
        nat_commands = self.nat.commands()

        # Address groups are used in NAT and firewall rules, which are set prior to
        # the host being parsed. To avoid chicken and egg problems, just pull out
        # the address groups here instead, which is sub-optimal for efficiency but
        # necessary to avoid breaking at runtime
        address_groups = []
        for network in self.networks:
            for host in network.hosts:
                if hasattr(host, "address-groups"):
                    address_groups.extend(
                        [
                            "firewall group address-group {0} address {1}".format(
                                group, host.address
                            )
                            for group in getattr(host, "address-groups")
                        ]
                    )

        ordered_commands = [
            [*external_addresses, *port_groups, *address_groups],
            global_settings,
            nat_commands,
        ]

        all_commands = external_addresses + port_groups + global_settings + nat_commands

        # Group ordered network commands together, extending the list of ordered
        # commands by all of them after they've all been created
        network_ordered_commands = []
        for network in self.networks:
            net_ordered_commands, net_command_list = network.commands()
            all_commands.extend(net_command_list)

            for index, commands in enumerate(net_ordered_commands):
                while index >= len(network_ordered_commands):
                    network_ordered_commands.append([])

                network_ordered_commands[index].extend(commands)

        ordered_commands.extend(network_ordered_commands)

        return (ordered_commands, all_commands)

<filename>ubiquiti_config_generator/root_parser.py<gh_stars>1-10
"""
Contains the root configuration node
"""
import ipaddress
from os import path
from typing import List, Tuple

from ubiquiti_config_generator import file_paths, secondary_configs
from ubiquiti_config_generator.nodes import (
    GlobalSettings,
    PortGroup,
    ExternalAddresses,
    NAT,
    Network,
)


class RootNode:
    """
    Represents the root config node, from which everything else is based off of
    """

    # pylint: disable=too-many-arguments
    def __init__(
        self,
        global_settings: GlobalSettings,
        port_groups: List[PortGroup],
        external_addresses: ExternalAddresses,
        networks: List[Network],
        nat: NAT,
    ):
        self.global_settings = global_settings
        self.port_groups = port_groups
        self.external_addresses = external_addresses
        self.networks = networks
        self.nat = nat

    @classmethod
    def create_from_configs(cls, config_path: str):
        """
        Load configuration from files
        """
        nat = NAT(config_path)
        return cls(
            secondary_configs.get_global_configuration(config_path),
            secondary_configs.get_port_groups(config_path),
            secondary_configs.get_external_addresses(config_path),
            [
                Network(
                    network_folder.split(path.sep)[-2],
                    nat,
                    config_path,
                    **(file_paths.load_yaml_from_file(network_folder))
                )
                for network_folder in file_paths.get_folders_with_config(
                    [config_path, file_paths.NETWORK_FOLDER]
                )
            ],
            nat,
        )

    def is_valid(self) -> bool:
        """
        Are all fields in the configuration valid
        """
        settings_valid = self.global_settings.validate()
        addresses_valid = self.external_addresses.validate()
        ports_valid = all([port.validate() for port in self.port_groups])
        networks_valid = all([network.validate() for network in self.networks])
        nat_valid = self.nat.validate()

        return (
            settings_valid
            and addresses_valid
            and ports_valid
            and networks_valid
            and nat_valid
        )

    def is_consistent(self) -> bool:
        """
        Check configuration for consistency
        """
        addresses_consistent = self.external_addresses.is_consistent()
        globals_consistent = self.global_settings.is_consistent()
        port_groups_consistent = [group.is_consistent() for group in self.port_groups]
        networks_consistent = [network.is_consistent() for network in self.networks]
        nat_consistent = self.nat.is_consistent()

        networks_consistent = all(networks_consistent) and True

        network_count = len(self.networks)
        for network_index in range(network_count):
            network = self.networks[network_index]

            # Skip any DHCP networks, e.g. for WAN
            if network.cidr is None:
                continue

            ip_network = ipaddress.ip_network(network.cidr)
            for second_network_index in range(network_index + 1, network_count):
                second_network = self.networks[second_network_index]

                # Skip any DHCP networks, e.g. for WAN
                if second_network.cidr is None:
                    continue

                second_ip_network = ipaddress.ip_network(second_network.cidr)

                if ip_network.overlaps(second_ip_network):
                    network.add_validation_error(
                        "{0} overlaps with {1}".format(
                            str(network), str(second_network)
                        )
                    )
                    networks_consistent = False

        return (
            addresses_consistent
            and globals_consistent
            and all(port_groups_consistent)
            and networks_consistent
            and nat_consistent
        )

    def validate(self) -> bool:
        """
        Is the root node valid
        """
        # Ensure both are checked, for complete error messages
        valid = self.is_valid()
        consistent = self.is_consistent()
        return valid and consistent

    def validation_failures(self) -> List[str]:
        """
        Get all validation failures
        """
        failures = (
            self.global_settings.validation_errors()
            + self.external_addresses.validation_errors()
            + self.nat.validation_failures()
        )
        for port in self.port_groups:
            failures.extend(port.validation_errors())
        for network in self.networks:
            failures.extend(network.validation_failures())

        return failures

    def get_commands(self) -> Tuple[List[List[str]], List[str]]:
        """
        Returns the commands to generate this configuration

        First value is an ordered list of commands to run, segmented into
        distinct portions which need to be committed in-order
        Second value is a flat list of all commands, for comparison against
        the previous configuration, to check for needed deletions
        """
        # These 3 should just be a list of commands, since ordering won't matter
        external_addresses = self.external_addresses.commands()
        port_groups = []
        for group in self.port_groups:
            port_groups.extend(group.commands())
        global_settings = self.global_settings.commands()
        nat_commands = self.nat.commands()

        # Address groups are used in NAT and firewall rules, which are set prior to
        # the host being parsed. To avoid chicken and egg problems, just pull out
        # the address groups here instead, which is sub-optimal for efficiency but
        # necessary to avoid breaking at runtime
        address_groups = []
        for network in self.networks:
            for host in network.hosts:
                if hasattr(host, "address-groups"):
                    address_groups.extend(
                        [
                            "firewall group address-group {0} address {1}".format(
                                group, host.address
                            )
                            for group in getattr(host, "address-groups")
                        ]
                    )

        ordered_commands = [
            [*external_addresses, *port_groups, *address_groups],
            global_settings,
            nat_commands,
        ]

        all_commands = external_addresses + port_groups + global_settings + nat_commands

        # Group ordered network commands together, extending the list of ordered
        # commands by all of them after they've all been created
        network_ordered_commands = []
        for network in self.networks:
            net_ordered_commands, net_command_list = network.commands()
            all_commands.extend(net_command_list)

            for index, commands in enumerate(net_ordered_commands):
                while index >= len(network_ordered_commands):
                    network_ordered_commands.append([])

                network_ordered_commands[index].extend(commands)

        ordered_commands.extend(network_ordered_commands)

        return (ordered_commands, all_commands)

Contains the root configuration node Represents the root config node, from which everything else is based off of # pylint: disable=too-many-arguments Load configuration from files Are all fields in the configuration valid Check configuration for consistency # Skip any DHCP networks, e.g. for WAN # Skip any DHCP networks, e.g. for WAN Is the root node valid # Ensure both are checked, for complete error messages Get all validation failures Returns the commands to generate this configuration          First value is an ordered list of commands to run, segmented into         distinct portions which need to be committed in-order         Second value is a flat list of all commands, for comparison against         the previous configuration, to check for needed deletions # These 3 should just be a list of commands, since ordering won't matter # Address groups are used in NAT and firewall rules, which are set prior to # the host being parsed. To avoid chicken and egg problems, just pull out # the address groups here instead, which is sub-optimal for efficiency but # necessary to avoid breaking at runtime # Group ordered network commands together, extending the list of ordered # commands by all of them after they've all been created