Source code for sceptre.diffing.stack_differ

import difflib
import logging
from abc import abstractmethod
from typing import NamedTuple, Dict, List, Optional, Callable, Tuple, Generic, TypeVar, Union

import cfn_flip
import deepdiff
import yaml
from cfn_tools import ODict
from yaml import Dumper

from sceptre.plan.actions import StackActions
from sceptre.stack import Stack

DiffType = TypeVar('DiffType')

logger = logging.getLogger(__name__)


[docs]class StackConfiguration(NamedTuple): """A data container to represent the comparable parts of a Stack.""" stack_name: str parameters: Dict[str, Union[str, List[str]]] stack_tags: Dict[str, str] notifications: List[str] role_arn: Optional[str]
[docs]class StackDiff(NamedTuple): """A data container to represent the full difference between a deployed stack and the stack as it exists locally within Sceptre. """ stack_name: str template_diff: DiffType config_diff: DiffType is_deployed: bool generated_config: StackConfiguration generated_template: str
[docs]def repr_str(dumper: Dumper, data: str) -> str: """A YAML Representer that handles strings, breaking multi-line strings into something a lot more readable in the yaml output. This is useful for representing long, multiline strings in templates or in stack parameters. :param dumper: The Dumper that is being used to serialize this object :param data: The string to serialize :return: The represented string """ if '\n' in data: return dumper.represent_scalar(u'tag:yaml.org,2002:str', data, style='|') return dumper.represent_str(data)
[docs]def repr_odict(dumper: Dumper, data: ODict) -> str: """A YAML Representer for cfn-flip's ODict objects. ODicts are a variation on OrderedDicts for that library. Since the Diff command makes extensive use of ODicts, they can end up in diff output and the PyYaml library doesn't otherwise calls the dicts like !!ODict, which looks weird. We can just treat them like normal dicts when we serialize them, though. :param dumper: The Dumper that is being used to serialize this object :param data: The ODict object to serialize :return: The serialized ODict """ return dumper.represent_dict(data)
yaml.add_representer(str, repr_str) yaml.add_representer(ODict, repr_odict)
[docs]class StackDiffer(Generic[DiffType]): """A utility for producing a StackDiff that indicates the full difference between a given stack as it is currently DEPLOYED on CloudFormation and the stack as it exists in the local Sceptre configurations. This utility compares both the stack configuration (specifically those attributes that CAN be compared) as well as the stack template. As an abstract base class, the two comparison methods need to be implemented so that the StackDiff can be generated. """ STACK_STATUSES_INDICATING_NOT_DEPLOYED = [ 'CREATE_FAILED', 'ROLLBACK_COMPLETE', 'DELETE_COMPLETE', ] NO_ECHO_REPLACEMENT = "***HIDDEN***" def __init__(self, show_no_echo=False): """Initializes the StackDiffer. :param show_no_echo: If True, local parameters passed that the template says are NoEcho parameters will be displayed; Otherwise, they will be masked in the diff. """ self.show_no_echo = show_no_echo
[docs] def diff(self, stack_actions: StackActions) -> StackDiff: """Produces a StackDiff between the currently deployed stack (if it exists) and the stack as it exists locally in Sceptre. :param stack_actions: The StackActions object to use for generating and fetching templates as well as providing other details about the stack. :return: The StackDiff that expresses the difference. """ generated_config = self._create_generated_config(stack_actions.stack) deployed_config = self._create_deployed_stack_config(stack_actions) is_stack_deployed = bool(deployed_config) generated_template = self._generate_template(stack_actions) deployed_template = self._get_deployed_template(stack_actions, is_stack_deployed) self._handle_special_parameter_situations(stack_actions, generated_config, deployed_config) template_diff = self.compare_templates(deployed_template, generated_template) config_diff = self.compare_stack_configurations(deployed_config, generated_config) return StackDiff( stack_actions.stack.external_name, template_diff, config_diff, is_stack_deployed, generated_config, generated_template )
def _create_generated_config(self, stack: Stack) -> StackConfiguration: parameters = self._extract_parameters_from_generated_stack(stack) stack_configuration = StackConfiguration( stack_name=stack.external_name, parameters=parameters, stack_tags=stack.tags, notifications=stack.notifications, role_arn=stack.role_arn ) return stack_configuration def _extract_parameters_from_generated_stack(self, stack: Stack) -> dict: """Extracts a usable dict of parameters from the stack, performing some minor transformations to match the what CloudFormation does on its end. :param stack: The stack to extract the parameters from :return: A dictionary of stack parameters to be compared. """ formatted_parameters = {} for key, value in stack.parameters.items(): if isinstance(value, list): value = ','.join(item.rstrip('\n') for item in value) formatted_parameters[key] = value.rstrip('\n') return formatted_parameters def _create_deployed_stack_config(self, stack_actions: StackActions) -> Optional[StackConfiguration]: description = stack_actions.describe() if description is None: # This means the stack has not been deployed yet return None stacks = description['Stacks'] for stack in stacks: if stack['StackStatus'] in self.STACK_STATUSES_INDICATING_NOT_DEPLOYED: return None return StackConfiguration( parameters={ param['ParameterKey']: param['ParameterValue'] for param in stack.get('Parameters', []) }, stack_tags={ tag['Key']: tag['Value'] for tag in stack['Tags'] }, stack_name=stack['StackName'], notifications=stack['NotificationARNs'], role_arn=stack.get('RoleARN') ) def _handle_special_parameter_situations( self, stack_actions: StackActions, generated_config: StackConfiguration, deployed_config: StackConfiguration ): deployed_template_summary = stack_actions.fetch_remote_template_summary() generated_template_summary = stack_actions.fetch_local_template_summary() if deployed_config is not None: # Trailing linebreaks sometimes get removed by CloudFormation in certain circumstances # and can sometimes come from using the !file_contents resolver, but ultimately they # shouldn't affect the diff. We'll ignore all trailing linebreaks. self._remove_terminating_linebreaks_from_deployed_parameters( deployed_template_summary, deployed_config ) # If the parameter is not passed by Sceptre and the value on the deployed parameter is # the default value, we'll actually remove it from the deployed parameters list so it # doesn't show up as a false positive. self._remove_deployed_default_parameters_that_arent_passed( deployed_template_summary, generated_config, deployed_config ) if not self.show_no_echo: # We don't actually want to show parameters Sceptre is passing that the local template # marks as NoEcho parameters (unless show_no_echo is set to true). Therefore those # parameter values will be masked. self._mask_no_echo_parameters(generated_template_summary, generated_config) def _remove_terminating_linebreaks_from_deployed_parameters( self, template_summary: Optional[dict], deployed_config: StackConfiguration ): if template_summary is None: return parameter_types = { parameter['ParameterKey']: parameter['ParameterType'] for parameter in template_summary['Parameters'] } for key, value in deployed_config.parameters.items(): parameter_type = parameter_types[key] if parameter_type == 'CommaDelimitedList': # If it's a list of strings, remove trailing linebreaks for each item value = ','.join([item.rstrip('\n') for item in value.split(',')]) deployed_config.parameters[key] = value.rstrip('\n') def _remove_deployed_default_parameters_that_arent_passed( self, template_summary: dict, generated_config: StackConfiguration, deployed_config: StackConfiguration, ): deployed_config_default_map = self._get_parameter_default_map(template_summary) for parameter_key, default_value in deployed_config_default_map.items(): # If the generated config defines that parameter, leave that in the deployed config # so we can see the diff. if parameter_key in generated_config.parameters: continue # But if the stack config is relying on the template's default value for this parameter, # remove that parameter from the deployed config... but only if its value is set to the # default value. If we don't do this, it will show up as a difference when it isn't. if deployed_config.parameters[parameter_key] == default_value: del deployed_config.parameters[parameter_key] def _get_parameter_default_map(self, template_summary: dict) -> Dict[str, str]: if template_summary is None: return {} parameters = template_summary['Parameters'] default_map = {} for parameter in parameters: key = parameter['ParameterKey'] value = self._handle_default_value(parameter) if value is not None: default_map[key] = value return default_map def _handle_default_value(self, parameter): default_value = parameter.get('DefaultValue') param_type = parameter['ParameterType'] if default_value is None: return None if parameter.get('NoEcho'): default_value = '****' elif 'List' in param_type: # Eliminate whitespace around commas default_value = ','.join(value.strip() for value in default_value.split(',')) return default_value def _mask_no_echo_parameters(self, template_summary: dict, generated_config: StackConfiguration): parameters = template_summary['Parameters'] for parameter in parameters: key = parameter['ParameterKey'] if parameter.get('NoEcho') and key in generated_config.parameters: generated_config.parameters[key] = self.NO_ECHO_REPLACEMENT def _generate_template(self, stack_actions: StackActions) -> str: return stack_actions.generate() def _get_deployed_template(self, stack_actions: StackActions, is_deployed: bool) -> str: if is_deployed: return stack_actions.fetch_remote_template() or '{}' else: return '{}'
[docs] @abstractmethod def compare_templates( self, deployed: str, generated: str ) -> DiffType: """Implement this method to return the diff for the templates :param deployed: The stack template as it has been deployed :param generated: The stack template as it exists locally within Sceptre :return: The generated diff between the two """
[docs] @abstractmethod def compare_stack_configurations( self, deployed: Optional[StackConfiguration], generated: StackConfiguration ) -> DiffType: """Implement this method to return the diff for the stack configurations. :param deployed: The StackConfiguration as it has been deployed. This MIGHT be None, if the stack has not been deployed. :param generated: The StackConfiguration as it exists locally within Sceptre :return: The generated diff between the two """
[docs]class DeepDiffStackDiffer(StackDiffer[deepdiff.DeepDiff]): """A StackDiffer that relies upon the DeepDiff library to produce the difference between the stack as it has been deployed onto CloudFormation and as it exists locally within Sceptre. This differ relies upon a recursive key/value comparison of Python data structures, indicating specific keys or values that have been added, removed, or altered between the two. Templates are read in as dictionaries and compared this way, so json or yaml formatting changes will not be reflected, only changes in value. """ VERBOSITY_LEVEL_TO_INDICATE_CHANGED_VALUES = 2 def __init__( self, show_no_echo=False, *, universal_template_loader: Callable[[str], Tuple[dict, str]] = cfn_flip.load ): """Initializes a DeepDiffStackDiffer. :param show_no_echo: If True, local parameters passed that the template says are NoEcho parameters will be displayed; Otherwise, they will be masked in the diff. :param universal_template_loader: This should be a callable that can load either a json or yaml string and return a tuple where the first element is the loaded template and the second element is the template format (either "json" or "yaml") """ super().__init__(show_no_echo) self.load_template = universal_template_loader
[docs] def compare_stack_configurations( self, deployed: Optional[StackConfiguration], generated: StackConfiguration, ) -> deepdiff.DeepDiff: return deepdiff.DeepDiff( deployed, generated, verbose_level=self.VERBOSITY_LEVEL_TO_INDICATE_CHANGED_VALUES )
[docs] def compare_templates(self, deployed: str, generated: str) -> deepdiff.DeepDiff: # We don't actually care about the original formats here, since we only care about the # template VALUES. deployed_dict, _ = self.load_template(deployed) generated_dict, _ = self.load_template(generated) return deepdiff.DeepDiff( deployed_dict, generated_dict, verbose_level=self.VERBOSITY_LEVEL_TO_INDICATE_CHANGED_VALUES )
[docs]class DifflibStackDiffer(StackDiffer[List[str]]): """A StackDiffer that uses difflib to produce a diff between the stack as it exists on AWS and the stack as it exists locally within Sceptre. Because difflib generates diffs off of lists of strings, both StackConfigurations and """ def __init__( self, show_no_echo=False, *, universal_template_loader: Callable[[str], Tuple[dict, str]] = cfn_flip.load ): """Initializes a DifflibStackDiffer. :param show_no_echo: If True, local parameters passed that the template says are NoEcho parameters will be displayed; Otherwise, they will be masked in the diff. :param universal_template_loader: This should be a callable that can load either a json or yaml string and return a tuple where the first element is the loaded template and the second element is the template format (either "json" or "yaml") """ super().__init__(show_no_echo) self.load_template = universal_template_loader
[docs] def compare_stack_configurations( self, deployed: Optional[StackConfiguration], generated: StackConfiguration, ) -> List[str]: if deployed is None: comparable_deployed = None else: comparable_deployed = self._make_stack_configuration_comparable(deployed) comparable_generated = self._make_stack_configuration_comparable(generated) deployed_string = cfn_flip.dump_yaml(comparable_deployed) generated_string = cfn_flip.dump_yaml(comparable_generated) return self._make_string_diff( deployed_string, generated_string )
def _make_stack_configuration_comparable(self, config: Optional[StackConfiguration]): as_dict = dict(config._asdict()) return { key: value for key, value in as_dict.items() # stack_name isn't always going to be the same, otherwise we wouldn't be comparing them. # It's more confusing to have it in the diff output than to just remove it. if value not in (None, [], {}) and key != 'stack_name' }
[docs] def compare_templates( self, deployed: str, generated: str, ) -> List[str]: # Sometimes there might only be simple whitespace differences... which difflib will show but # are actually insignificant and "false positives". Also, it's POSSIBLE that the template # format might have changed, even if all the VALUES have stayed the same, so we'll read both # templates into dicts (regardless of their format) and we'll output both to the format of # the local template using identical serialization settings. This will truly enable comparison # of the actual values rather than other things that don't actually make a difference to # CloudFormation. If only the format/meaningless whitespace has changed, this will result in # there being no diff. deployed_dict, _ = self.load_template(deployed) generated_dict, generated_format = self.load_template(generated) dumpers = { 'json': cfn_flip.dump_json, 'yaml': cfn_flip.dump_yaml } deployed_reformatted = dumpers[generated_format](deployed_dict) generated_reformatted = dumpers[generated_format](generated_dict) return self._make_string_diff(deployed_reformatted, generated_reformatted)
def _make_string_diff(self, deployed: str, generated: str) -> List[str]: diff_lines = difflib.unified_diff( deployed.splitlines(), generated.splitlines(), fromfile="deployed", tofile="generated", lineterm="" ) return list(diff_lines)