Source code for homer

"""Homer package."""
import logging
import os
import pathlib
import sys

from collections import defaultdict
from importlib import import_module
from typing import Callable, DefaultDict, Dict, List, Mapping, Optional, Tuple

import pynetbox

from pkg_resources import DistributionNotFound, get_distribution

from homer.capirca import CapircaGenerate
from homer.config import HierarchicalConfig, load_yaml_config
from homer.devices import Device, Devices
from homer.exceptions import HomerAbortError, HomerConnectError, HomerError, HomerTimeoutError
from homer.netbox import NetboxData, NetboxDeviceData, NetboxInventory
from homer.templates import Renderer
from homer.transports import DEFAULT_TIMEOUT
from homer.transports.junos import connected_device


TIMEOUT_ATTEMPTS = 3
""":py:class:`int`: the number of attempts to try when there is a timeout."""
DIFF_EXIT_CODE = 99
""":py:class:`int`: the exit code used when the diff command is executed and there is a diff."""


try:
    __version__ = get_distribution('homer').version  # Must be the same used as 'name' in setup.py
    """:py:class:`str`: the version of the current Homer package."""
except DistributionNotFound:  # pragma: no cover - this should never happen during tests
    pass  # package is not installed

logger = logging.getLogger(__name__)


[docs]class Homer: """The instance to run Homer.""" OUT_EXTENSION = '.out' """:py:class:`str`: the extension for the generated output files.""" def __init__(self, main_config: Mapping): """Initialize the instance. Arguments: main_config (dict): the configuration dictionary. """ logger.debug('Initialized with configuration: %s', main_config) self._main_config = main_config self.private_base_path = self._main_config['base_paths'].get('private', '') self._config = HierarchicalConfig( self._main_config['base_paths']['public'], private_base_path=self.private_base_path) self._netbox_api = None self._device_plugin = None if self._main_config.get('netbox', {}): self._netbox_api = pynetbox.api( self._main_config['netbox']['url'], token=self._main_config['netbox']['token'], threading=True) if self._main_config['netbox'].get('plugin', ''): self._device_plugin = import_module( self._main_config['netbox']['plugin']).NetboxDeviceDataPlugin devices_all_config = load_yaml_config( os.path.join(self._main_config['base_paths']['public'], 'config', 'devices.yaml')) devices_config = {fqdn: data.get('config', {}) for fqdn, data in devices_all_config.items()} netbox_inventory = self._main_config.get('netbox', {}).get('inventory', {}) devices = devices_all_config.copy() for data in devices.values(): data.pop('config', None) if netbox_inventory: # Get the data from Netbox while keeping any existing metadata from the devices.yaml file. # The data from Netbox overrides the existing keys for each device, if both present. netbox_devices = NetboxInventory( self._netbox_api, netbox_inventory['device_roles'], netbox_inventory['device_statuses']).get_devices() for fqdn, data in netbox_devices.items(): if fqdn in devices: devices[fqdn].update(data) else: devices[fqdn] = data private_devices_config: Dict = {} if self.private_base_path: private_devices_config = load_yaml_config( os.path.join(self.private_base_path, 'config', 'devices.yaml')) self._ignore_warning = self._main_config.get('transports', {}).get('junos', {}).get('ignore_warning', False) self._transport_username = self._main_config.get('transports', {}).get('username', '') self._transport_timeout = self._main_config.get('transports', {}).get('timeout', DEFAULT_TIMEOUT) transport_ssh_config = self._main_config.get('transports', {}).get('ssh_config', None) if transport_ssh_config is not None: transport_ssh_config = str(pathlib.Path(transport_ssh_config).expanduser()) self._transport_ssh_config = transport_ssh_config self._devices = Devices(devices, devices_config, private_devices_config) self._renderer = Renderer(self._main_config['base_paths']['public'], self.private_base_path) self._output_base_path = pathlib.Path(self._main_config['base_paths']['output'])
[docs] def generate(self, query: str) -> int: """Generate the configuration only saving it locally, no remote action is performed. Arguments: query (str): the query to select the devices. Return: int: ``0`` on success, a small positive integer on failure. """ logger.info('Generating configuration for query %s', query) self._prepare_out_dir() successes, _ = self._execute(self._device_generate, query) return Homer._parse_results(successes)
[docs] def diff(self, query: str, *, omit_diff: bool = False) -> int: """Generate the configuration and check the diff with the current live one. Arguments: query (str): the query to select the devices. omit_diff (bool, optional): whether to not show the actual diff to avoid leak of private data. Return: int: ``0`` on success, a small positive integer on failure. """ logger.info('Generating diff for query %s', query) successes, diffs = self._execute(self._device_diff, query) has_diff = False for diff, diff_devices in diffs.items(): print(f'Changes for {len(diff_devices)} devices: {diff_devices}') if diff is None: print('# Failed') elif not diff: print('# No diff') else: has_diff = True if omit_diff: print('# Non-empty diff omitted, -o/--omit-diff set') else: print(diff) print('---------------') ret = Homer._parse_results(successes) if ret == 0 and has_diff: return DIFF_EXIT_CODE return ret
[docs] def commit(self, query: str, *, message: str = '-') -> int: """Commit the generated configuration asking for confirmation. Arguments: query (str): the query to select the devices. message (str): the commit message to use. Return: int: ``0`` on success, a small positive integer on failure. """ logger.info('Committing config for query %s with message: %s', query, message) successes, _ = self._execute(self._device_commit, query, message=message) return Homer._parse_results(successes)
def _device_generate(self, device: Device, device_config: str, _: int) -> Tuple[bool, Optional[str]]: """Save the generated configuration in a local file. Arguments: device (homer.devices.Device): the device instance. device_config (str): the generated configuration for the device. attempt (int, unused): the current attempt number. Returns: tuple: a two-element tuple with a boolean as first parameter that represent the success of the operation or not and a second element with a string or None that is not used but is required by the callback API. """ output_path = self._output_base_path / f'{device.fqdn}{Homer.OUT_EXTENSION}' with open(str(output_path), 'w', encoding='utf-8') as f: f.write(device_config) logger.info('Written configuration for %s in %s', device.fqdn, output_path) return True, None def _device_diff(self, device: Device, device_config: str, _: int) -> Tuple[bool, Optional[str]]: """Perform a configuration diff between the generated configuration and the live one. Arguments: device (homer.devices.Device): the device instance. device_config (str): the generated configuration for the device. attempt (int, unused): the current attempt number. Returns: tuple: a two-element tuple with a boolean as first parameter that represent the success of the operation or not and a second element with a string that contains the configuration differences or None if unable to load the new configuration in the device to generate the diff. """ timeout = device.metadata.get('timeout', self._transport_timeout) with connected_device(device.fqdn, username=self._transport_username, ssh_config=self._transport_ssh_config, timeout=timeout) as connection: return connection.commit_check(device_config, self._ignore_warning) def _device_commit(self, device: Device, device_config: str, # noqa: MC0001; pylint: disable=no-self-use attempt: int, *, message: str = '-') -> Tuple[bool, Optional[str]]: """Commit a new configuration to the device. Arguments: device (homer.devices.Device): the device instance. device_config (str): the generated configuration for the device. attempt (int): the current attempt number. message (str): the commit message to use. Raises: HomerTimeoutError: on timeout. Returns: tuple: a two-element tuple with a boolean as first parameter that represent the success of the operation or not and a second element with a string or None that is not used but is required by the callback API. """ def callback(fqdn: str, diff: str) -> None: """Callback as required by :py:class:`homer.transports.junos.ConnectedDevice.commit`.""" if not sys.stdout.isatty(): raise HomerError('Not in a TTY, unable to ask for confirmation') print(f'Configuration diff for {fqdn}:\n{diff}') print('Type "yes" to commit, "no" to abort.') for _ in range(2): resp = input('> ') if resp == 'yes': break if resp == 'no': raise HomerAbortError('Commit aborted') print(('Invalid response, please type "yes" to commit or "no" to abort. After 2 wrong answers the ' 'commit will be aborted.')) else: raise HomerAbortError('Too many invalid answers, commit aborted') is_retry = (attempt != 1) timeout = device.metadata.get('timeout', self._transport_timeout) with connected_device(device.fqdn, username=self._transport_username, ssh_config=self._transport_ssh_config, timeout=timeout) as connection: try: connection.commit(device_config, message, callback, ignore_warning=self._ignore_warning, is_retry=is_retry) return True, '' except HomerTimeoutError: raise # To be catched later for automatic retry except HomerAbortError as e: logger.warning('%s on %s', e, device.fqdn) except Exception as e: # pylint: disable=broad-except logger.error('Failed to commit on %s: %s', device.fqdn, e) logger.debug('Traceback:', exc_info=True) return False, '' def _prepare_out_dir(self) -> None: """Prepare the out directory creating the directory if doesn't exists and deleting any pre-generated file.""" self._output_base_path.mkdir(parents=True, exist_ok=True) for path in self._output_base_path.iterdir(): if path.is_file() and path.suffix == Homer.OUT_EXTENSION: path.unlink() def _execute(self, callback: Callable, query: str, **kwargs: str) -> Tuple[Dict, DefaultDict]: # noqa, mccabe: MC0001 too complex """Execute Homer based on the given action and query. Arguments: callback (Callable): the callback to call for each device. query (str): the query to filter the devices to act on. **kwargs (str): any additional keyword argument to pass to the callback Returns: tuple: a two-element tuple, with the first item as a dictionary that contains two keys (:py:data:`True` and :py:data:`False`) and as value a list of device FQDN that were successful (True) or failed (False) the operation and a second element a :py:class:`collections.defaultdict` that has as keys the configuration differences and as values the list of device FQDN that reported that diff. """ diffs: DefaultDict[str, list] = defaultdict(list) successes: Dict[bool, list] = {True: [], False: []} netbox_data = None if self._netbox_api is not None: logger.info('Gathering global Netbox data') netbox_data = NetboxData(self._netbox_api) for device in self._devices.query(query): logger.info('Generating configuration for %s', device.fqdn) try: device_config = [] device_data = self._config.get(device) # Render the ACLs using Capirca if 'capirca' in device_data and not self._main_config.get('capirca', {}).get('disabled', False): capirca = CapircaGenerate(self._main_config, device_data['capirca'], self._netbox_api) generated_acls = capirca.generate_acls() if generated_acls: device_config.extend(generated_acls) if netbox_data is not None: device_data['netbox'] = { 'global': netbox_data, 'device': NetboxDeviceData(self._netbox_api, device), } if self._device_plugin is not None: device_data['netbox']['device_plugin'] = self._device_plugin(self._netbox_api, device) # Render the Jinja templates based on yaml + netbox data device_config.append(self._renderer.render(device.metadata['role'], device_data)) except HomerError: logger.exception('Device %s failed to render the template, skipping.', device.fqdn) successes[False].append(device.fqdn) continue for attempt in range(1, TIMEOUT_ATTEMPTS + 1): try: device_success, device_diff = callback(device, '\n'.join(device_config), attempt, **kwargs) break except (HomerTimeoutError, HomerConnectError) as e: logger.error('Attempt %d/%d failed: %s', attempt, TIMEOUT_ATTEMPTS, e) else: device_success = False device_diff = '' successes[device_success].append(device.fqdn) diffs[device_diff].append(device.fqdn) return successes, diffs @staticmethod def _parse_results(successes: Mapping[bool, List[Device]]) -> int: """Parse the results dictionary, log and return the approriate exit status code. Arguments: successes (dict): a dictionary that contains two keys (:py:data:`True` and :py:data:`False`) and as value a list of device FQDN that were successful (True) or failed (False) the operation. Return: int: ``0`` on success, a small positive integer on failure. """ if successes[False]: logger.error('Homer run had issues on %d devices: %s', len(successes[False]), successes[False]) return 1 logger.info('Homer run completed successfully on %d devices: %s', len(successes[True]), successes[True]) return 0