Render Terraform AWS security group diffs for humans

sg-diff

class SgDiff
  SGR_ANSI = "\e[%sm"

  SGR_CODES = {
    none:     0,
    bold:     1,
    black:   30,
    red:     31,
    green:   32,
    yellow:  33,
    blue:    34,
    magenta: 35,
    cyan:    36,
    white:   37,
  }

  BLANKS = [nil, '', '0', []]

  # Constructs an ANSI Select Graphic Rendition sequence.
  def sgr(codes)
    SGR_ANSI % codes.map{|c| SGR_CODES[c] || raise("Unknown SGR code key: #{c.inspect}") }.join(';')
  end

  # Apply color codes to a section of text.
  def color(text, *codes)
    [sgr(codes), text, sgr([:none])].join('')
  end

  # Parse an input line into an attribute structure.
  def parse_line(line)
    if /^ *(\S+): +"([^"]*)" => "([^"]*)"$/ === line.chomp
      {name: $1.split('.'), before: $2, after: $3}
    else
      STDERR.puts "Ignoring unknown line format: #{line.inspect}"
      nil
    end
  end

  def process(input, out)
    rule_map = {}

    # Process input.
    while input.gets do
      line = parse_line($_)
      name = line && line[:name]
      if name && name[0] == 'ingress'
        rule_id = name[1]
        next if rule_id == '#'
        rule_map[rule_id] ||= {}
        rule_attr = name[2]
        next if rule_attr == 'self'
        if name.count > 3
          next if name[3] == '#'
          rule_map[rule_id][rule_attr] ||= {before: [], after: []}
          rule_map[rule_id][rule_attr][:before] << line[:before] unless line[:before] == ''
          rule_map[rule_id][rule_attr][:after] << line[:after] unless line[:after] == ''
        else
          before = line[:before] != '' && line[:before] || nil
          after = line[:after] != '' && line[:after] || nil
          rule_map[rule_id][rule_attr] = {before: before, after: after} if before || after
        end
      end
    end

    rules = []

    def rule_unchanged?(rule)
      rule.values.all? {|v| v[:before] == v[:after] }
    end

    def rule_added?(rule)
      rule.all? {|k, v| k.is_a?(Symbol) || (BLANKS.include?(v[:before]) && v[:after]) }
    end

    def rule_removed?(rule)
      rule.all? {|k, v| k.is_a?(Symbol) || (v[:before] && BLANKS.include?(v[:after])) }
    end

    def find_match(rule, rside, candidates, cside, *fields)
      candidates.each do |candidate|
        matched = true
        fields.each do |field|
          rval = rule[field] && rule[field][rside]
          cval = candidate[field] && candidate[field][cside]
          matched = false unless rval && cval && rval == cval
        end
        return candidate if matched
      end
      nil
    end

    # If all the attributes are the same, no change. Remove these from dedupe
    # consideration.
    rule_map.keys.each do |rule_id|
      rule = rule_map[rule_id]
      next unless rule_unchanged?(rule)
      rule[:id] = {before: rule_id, after: rule_id}
      rule[:state] = :unchanged
      rules << rule
      rule_map.delete(rule_id)
    end

    # Set ids so we can access later.
    rule_map.each do |rule_id, rule|
      rule[:id] = rule_id
    end

    # Try to pair up additions and removals.
    rule_map.keys.each do |rule_id|
      rule = rule_map[rule_id]
      next unless rule
      if rule_added? rule
        candidates = rule_map.values.select {|r| rule_removed? r }
        match = find_match(rule, :after, candidates, :before, 'description') \
          || find_match(rule, :after, candidates, :before, 'protocol', 'from_port', 'to_port') \
          || find_match(rule, :after, candidates, :before, 'cidr_blocks') \
          || find_match(rule, :after, candidates, :before, 'ipv6_cidr_blocks') \
          || find_match(rule, :after, candidates, :before, 'security_groups')
        rule_map.delete(rule_id)
        if match
          rule_map.delete(match[:id])
          merged = {id: {before: match[:id], after: rule_id}}
          (rule.keys + match.keys).uniq.each do |k|
            next if k == :id
            merged[k] = {before: match[k][:before], after: rule[k][:after]}
          end
          merged[:state] = :diff
          rules << merged
        else
          rule[:id] = {after: rule_id}
          rule[:state] = :added
          rules << rule
        end
      elsif rule_removed? rule
        candidates = rule_map.values.select {|r| rule_added? r }
        match = find_match(rule, :before, candidates, :after, 'description') \
          || find_match(rule, :before, candidates, :after, 'protocol', 'from_port', 'to_port') \
          || find_match(rule, :before, candidates, :after, 'cidr_blocks') \
          || find_match(rule, :before, candidates, :after, 'ipv6_cidr_blocks') \
          || find_match(rule, :before, candidates, :after, 'security_groups')
        rule_map.delete(rule_id)
        if match
          rule_map.delete(match[:id])
          merged = {id: {before: rule_id, after: match[:id]}}
          (rule.keys + match.keys).uniq.each do |k|
            next if k == :id
            merged[k] = {before: rule[k][:before], after: match[k][:after]}
          end
          merged[:state] = :diff
          rules << merged
        else
          rule[:id] = {before: rule_id}
          rule[:state] = :removed
          rules << rule
        end
      else
        raise "Unknown rule changes: #{rule_id} => #{rule.inspect}"
      end
    end

    # Roughly sort rules by port order.
    rules.sort_by! do |rule|
      from_port = rule['from_port']
      to_port = rule['to_port']
      (from_port && (from_port[:before] || from_port[:after])) \
        || (to_port && (to_port[:before] || to_port[:after]))
    end

    # Return a human-readable string for the port range in a rule.
    def port_range(rule, side)
      protocol = rule['protocol'][side] || '???'
      from_port = rule['from_port'][side] || '?'
      to_port = rule['to_port'][side] || '?'
      if protocol == 'icmp' && from_port == '-1' && to_port == '-1'
        range = '(all)'
      elsif from_port == '0' && to_port == '65535'
        range = '(all)'
      elsif from_port == to_port
        range = from_port
      else
        range = "#{from_port}-#{to_port}"
      end
      "#{protocol.upcase} #{range}"
    end

    # Diff two lists and return colorized and indented text.
    def diff_lists(left, right)
      (left + right).uniq.sort.each do |val|
        if left.include?(val) && right.include?(val)
          out.puts "        #{color(val, :cyan)}"
        elsif left.include?(val)
          out.puts "        #{color(val, :red)}"
        else
          out.puts "        #{color(val, :green)}"
        end
      end
    end

    special_attrs = ['description', 'protocol', 'from_port', 'to_port']
    list_attrs = ['cidr_blocks', 'ipv6_cidr_blocks', 'security_groups']

    # Pretty print each rule change.
    rules.each do |rule|
      case rule[:state]
      when :unchanged
        out.puts color(rule[:id][:before], :bold, :cyan)
        out.puts "    #{rule['description'][:before]}" if rule['description']
        out.puts "    #{port_range(rule, :before)}"
      when :removed
        out.puts "#{color(rule[:id][:before], :bold, :red)}"
        out.puts "    #{rule['description'][:before]}" if rule['description']
        out.puts "    #{port_range(rule, :before)}"
      when :added
        out.puts "#{color(rule[:id][:after], :bold, :green)}"
        out.puts "    #{rule['description'][:after]}" if rule['description']
        out.puts "    #{port_range(rule, :after)}"
      else
        out.puts "#{color(rule[:id][:before], :bold, :yellow)} => #{color(rule[:id][:after], :bold, :yellow)}"
        description = rule['description']
        if description && description[:before] == description[:after]
          out.puts description[:before]
        elsif description
          out.puts "    #{description[:before].inspect} => #{description[:after].inspect}"
        end
        before_range = port_range(rule, :before)
        after_range = port_range(rule, :after)
        if before_range == after_range
          out.puts "    #{before_range}"
        else
          out.puts "    #{port_range(rule, :before)} => #{port_range(rule, :after)}"
        end
      end
      list_attrs.each do |k|
        next unless rule[k]
        before = rule[k][:before]
        after = rule[k][:after]
        out.puts "    #{k}"
        diff_lists before, after
      end
      rule.each do |k, v|
        next if special_attrs.include?(k) || list_attrs.include?(k) || k.is_a?(Symbol)
        before = v[:before]
        after = v[:after]
        if before == after
          out.puts "    #{k}: #{before}"
        else
          out.puts "    #{k}: #{before} => #{after}"
        end
      end
    end
  end
end

SgDiff.new