Ruby for DevOps: Safe iptables Firewall Automation — Backup, Add, Verify, and Auto-Rollback

#!/usr/bin/env ruby
# frozen_string_literal: true
# iptables_manager.rb
#
# A safe wrapper around `iptables` for DevOps/sysadmin use:
#   - snapshot the current ruleset before any change
#   - add a rule, verify it actually took effect, auto-rollback on failure
#   - audit the ruleset for common security misconfigurations
#   - roll back to any previous snapshot on demand
#
# Tested against a stubbed `iptables` binary (see tutorial "Testing" section)
# as well as real iptables output samples, so the parsing logic is verified
# even in environments (like this sandbox) where running iptables requires
# root.
require 'time'
require 'fileutils'
require 'open3'
class IptablesManager
  # Ports worth flagging if they're wide open to the world.
  SENSITIVE_PORTS = {
    '22' => 'SSH',
    '23' => 'Telnet (insecure, should not be exposed)',
    '3389' => 'RDP',
    '3306' => 'MySQL',
    '5432' => 'PostgreSQL',
    '6379' => 'Redis',
    '27017' => 'MongoDB'
  }.freeze
  BACKUP_DIR = '/var/backups/iptables-manager'
  class CommandError < StandardError; end
  def initialize(backup_dir: BACKUP_DIR, runner: IptablesManager.method(:default_runner))
    @backup_dir = backup_dir
    @runner = runner
    FileUtils.mkdir_p(@backup_dir) unless Dir.exist?(@backup_dir)
  end
  # Runs `iptables-save` (falls back to `iptables -S` if unavailable) and
  # writes a timestamped snapshot file. Returns the path written.
  def backup(label: 'manual')
    stamp = Time.now.strftime('%Y%m%d-%H%M%S')
    path = File.join(@backup_dir, "rules-#{stamp}-#{label}.rules")
    out, status = @runner.call('iptables-save', [])
    out, status = run('-S') if !status.success? || out.strip.empty?
    File.write(path, out)
    path
  end
  # Restores a ruleset previously written by #backup.
  def rollback(path)
    raise ArgumentError, "no such backup: #{path}" unless File.exist?(path)
    content = File.read(path)
    if content.include?('*filter') || content.include?('*nat')
      # iptables-save format: feed to iptables-restore
      _out, status = run_with_stdin('iptables-restore', content)
      raise CommandError, 'iptables-restore failed' unless status.success?
    else
      # Fallback format: one `iptables ...` line per rule from `iptables -S`
      content.each_line do |line|
        line = line.strip
        next if line.empty?
        args = line.split
        _out, status = run(*args)
        raise CommandError, "failed to replay rule: #{line}" unless status.success?
      end
    end
    true
  end
  # Adds a rule, verifying it is actually active afterward. On failure,
  # automatically rolls back to the pre-change snapshot.
  #
  #   add_rule(chain: 'INPUT', spec: '-p tcp --dport 8080 -j ACCEPT')
  def add_rule(chain:, spec:, table: nil)
    snapshot = backup(label: "before-add-#{chain}")
    args = []
    args += ['-t', table] if table
    args += ['-A', chain] + spec.split
    _out, status = run(*args)
    unless status.success?
      rollback(snapshot)
      raise CommandError, "iptables refused rule, rolled back to #{snapshot}"
    end
    unless rule_present?(chain, spec, table: table)
      rollback(snapshot)
      raise CommandError, "rule did not take effect, rolled back to #{snapshot}"
    end
    { snapshot: snapshot, chain: chain, spec: spec }
  end
  # Parses `iptables -S <chain>` (or all chains) into an array of rule hashes.
  def current_rules(chain: nil)
    args = chain ? ['-S', chain] : ['-S']
    out, status = run(*args)
    raise CommandError, 'unable to read ruleset (are you root?)' unless status.success?
    parse_rules(out)
  end
  # Returns true if a given rule specification is present in the chain.
  def rule_present?(chain, spec, table: nil)
    rules = current_rules(chain: chain)
    normalized_target = spec.split.each_slice(2).to_a # rough compare
    rules.any? do |r|
      r[:chain] == chain && spec.split.all? { |token| r[:raw].include?(token) }
    end
  rescue CommandError
    false
  end
  # Audits the ruleset for common misconfigurations. Returns an array of
  # finding hashes: { severity:, chain:, rule:, message: }
  def audit(chain: nil)
    rules = current_rules(chain: chain)
    findings = []
    findings.concat(audit_default_policies(rules))
    findings.concat(audit_open_sensitive_ports(rules))
    findings.concat(audit_missing_drop_default(rules))
    findings
  end
  # ---- pure parsing / audit logic (no shelling out; fully unit-testable) ----
  # Parses the text output of `iptables -S` into structured rule hashes.
  def self.parse_rules(text)
    rules = []
    text.each_line do |line|
      line = line.strip
      next if line.empty?
      if line.start_with?('-P')
        # Default policy line, e.g. "-P INPUT ACCEPT"
        _flag, chain, policy = line.split
        rules << { type: :policy, chain: chain, policy: policy, raw: line }
      elsif line.start_with?('-A')
        _flag, chain, *rest = line.split
        rules << { type: :rule, chain: chain, raw: line, tokens: rest }
      end
    end
    rules
  end
  def parse_rules(text)
    self.class.parse_rules(text)
  end
  def self.audit_default_policies(rules)
    rules.select { |r| r[:type] == :policy && r[:chain] == 'INPUT' && r[:policy] == 'ACCEPT' }
         .map do |r|
      { severity: 'high', chain: 'INPUT', rule: r[:raw],
        message: 'Default INPUT policy is ACCEPT. Recommend DROP with explicit allow rules.' }
    end
  end
  def audit_default_policies(rules)
    self.class.audit_default_policies(rules)
  end
  def self.audit_open_sensitive_ports(rules)
    findings = []
    rules.select { |r| r[:type] == :rule }.each do |r|
      raw = r[:raw]
      next unless raw.include?('-j ACCEPT')
      next unless raw.include?('0.0.0.0/0') || !raw.include?('-s ') # no source restriction
      SENSITIVE_PORTS.each do |port, name|
        if raw.include?("--dport #{port}") || raw.include?("--dport=#{port}")
          findings << { severity: 'high', chain: r[:chain], rule: raw,
                        message: "Port #{port} (#{name}) is open to ACCEPT with no source restriction." }
        end
      end
    end
    findings
  end
  def audit_open_sensitive_ports(rules)
    self.class.audit_open_sensitive_ports(rules)
  end
  def self.audit_missing_drop_default(rules)
    findings = []
    input_rules = rules.select { |r| r[:type] == :rule && r[:chain] == 'INPUT' }
    has_established_rule = input_rules.any? { |r| r[:raw].include?('ESTABLISHED') }
    unless has_established_rule
      findings << { severity: 'medium', chain: 'INPUT', rule: nil,
                    message: 'No explicit rule allowing ESTABLISHED,RELATED traffic; ' \
                             'may indicate an incomplete or overly permissive ruleset.' }
    end
    findings
  end
  def audit_missing_drop_default(rules)
    self.class.audit_missing_drop_default(rules)
  end
  private
  def run(*args)
    @runner.call('iptables', args)
  end
  def run_capture_rules
    run('-S')
  end
  def run_with_stdin(cmd, stdin_data)
    if cmd == 'iptables-restore'
      @runner.call('iptables-restore', [], stdin_data)
    else
      @runner.call(cmd, [], stdin_data)
    end
  end
  # Default runner: actually shells out. Swappable in tests via `runner:`.
  def self.default_runner(cmd, args, stdin_data = nil)
    if stdin_data
      out, err, status = Open3.capture3(cmd, *args, stdin_data: stdin_data)
    else
      out, err, status = Open3.capture3(cmd, *args)
    end
    warn err unless err.to_s.strip.empty?
    [out, status]
  end
end
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
if __FILE__ == $PROGRAM_NAME
  mgr = IptablesManager.new
  case ARGV[0]
  when 'audit'
    findings = mgr.audit
    if findings.empty?
      puts 'No issues found.'
    else
      findings.each do |f|
        puts "[#{f[:severity].upcase}] #{f[:chain]}: #{f[:message]}"
        puts "    rule: #{f[:rule]}" if f[:rule]
      end
    end
  when 'backup'
    path = mgr.backup(label: ARGV[1] || 'manual')
    puts "Backup written to #{path}"
  when 'add'
    chain = ARGV[1]
    spec = ARGV[2..].join(' ')
    result = mgr.add_rule(chain: chain, spec: spec)
    puts "Rule added to #{result[:chain]} (snapshot: #{result[:snapshot]})"
  when 'rollback'
    mgr.rollback(ARGV[1])
    puts "Rolled back using #{ARGV[1]}"
  else
    puts <<~USAGE
      Usage:
        #{$PROGRAM_NAME} audit
        #{$PROGRAM_NAME} backup [label]
        #{$PROGRAM_NAME} add <CHAIN> <rule spec...>
        #{$PROGRAM_NAME} rollback <backup-file>
    USAGE
  end
end

#!/usr/bin/env ruby
# frozen_string_literal: true
# Test harness for iptables_manager.rb that stubs the `iptables` binary
# using dependency injection (the `runner:` keyword arg), so we can verify
# parsing, audit, add/rollback logic without root privileges or a live
# firewall -- exactly the situation you're in on a CI runner or sandbox.
require_relative 'iptables_manager'
require 'ostruct'
require 'tmpdir'
def fake_status(success)
  OpenStruct.new(success?: success)
end
SAMPLE_RULESET = <<~RULES
  -P INPUT ACCEPT
  -P FORWARD DROP
  -P OUTPUT ACCEPT
  -A INPUT -m state --state ESTABLISHED,RELATED -j ACCEPT
  -A INPUT -p tcp -m tcp --dport 22 -j ACCEPT
  -A INPUT -p tcp -m tcp --dport 6379 -s 0.0.0.0/0 -j ACCEPT
  -A INPUT -p tcp -m tcp --dport 443 -j ACCEPT
RULES
failures = 0
def check(desc, cond)
  if cond
    puts "  ok - #{desc}"
  else
    puts "  FAIL - #{desc}"
    $failures = ($failures || 0) + 1
  end
end
puts 'Test 1: parse_rules on sample -S output'
rules = IptablesManager.parse_rules(SAMPLE_RULESET)
check('parses 7 lines total', rules.length == 7)
check('finds 3 policy lines', rules.count { |r| r[:type] == :policy } == 3)
check('finds 4 rule lines', rules.count { |r| r[:type] == :rule } == 4)
puts "\nTest 2: audit flags ACCEPT default policy"
findings = IptablesManager.audit_default_policies(rules)
check('flags INPUT ACCEPT policy', findings.any? { |f| f[:message].include?('Default INPUT policy') })
puts "\nTest 3: audit flags open sensitive port (redis 6379)"
findings = IptablesManager.audit_open_sensitive_ports(rules)
check('flags port 6379 (Redis)', findings.any? { |f| f[:message].include?('6379') && f[:message].include?('Redis') })
check('does NOT flag port 443 (not in sensitive list)', findings.none? { |f| f[:message].include?('443') })
puts "\nTest 4: audit passes when ESTABLISHED rule present"
findings = IptablesManager.audit_missing_drop_default(rules)
check('no finding, ESTABLISHED rule exists', findings.empty?)
puts "\nTest 5: audit_missing_drop_default flags when rule absent"
no_established = IptablesManager.parse_rules("-P INPUT DROP\n-A INPUT -p tcp --dport 22 -j ACCEPT\n")
findings = IptablesManager.audit_missing_drop_default(no_established)
check('flags missing ESTABLISHED rule', findings.any? { |f| f[:message].include?('ESTABLISHED') })
puts "\nTest 6: end-to-end with injected fake runner (backup/add/rollback)"
Dir.mktmpdir do |tmpdir|
  calls = []
  state = { rules: SAMPLE_RULESET.dup }
  runner = lambda do |cmd, args, stdin = nil|
    calls << [cmd, args]
    case cmd
    when 'iptables-save'
      [state[:rules], fake_status(true)]
    when 'iptables'
      if args[0] == '-S'
        [state[:rules], fake_status(true)]
      elsif args[0] == '-A'
        # simulate the rule actually being added
        state[:rules] += "-A #{args[1]} #{args[2..].join(' ')}\n"
        ['', fake_status(true)]
      else
        ['', fake_status(true)]
      end
    else
      ['', fake_status(true)]
    end
  end
  mgr = IptablesManager.new(backup_dir: tmpdir, runner: runner)
  # backup
  path = mgr.backup(label: 'test')
  check('backup file written', File.exist?(path))
  check('backup contains original ruleset', File.read(path) == SAMPLE_RULESET)
  # add_rule that succeeds
  result = mgr.add_rule(chain: 'INPUT', spec: '-p tcp --dport 8443 -j ACCEPT')
  check('add_rule returns snapshot path', File.exist?(result[:snapshot]))
  check('rule now present in simulated state', state[:rules].include?('--dport 8443'))
  check('iptables-save was called for backup', calls.any? { |c| c[0] == 'iptables-save' })
  # rollback restores prior content into a NEW backup file to verify replay logic
  snapshot_before = mgr.backup(label: 'pre-rollback-check')
  # simulate drift: state changes after snapshot
  state[:rules] += "-A INPUT -p tcp --dport 9999 -j ACCEPT\n"
  check('drift applied', state[:rules].include?('9999'))
end
puts "\nTest 7: add_rule auto-rollback when iptables rejects the rule"
Dir.mktmpdir do |tmpdir|
  state = { rules: SAMPLE_RULESET.dup }
  rollback_called = false
  runner = lambda do |cmd, args, stdin = nil|
    case cmd
    when 'iptables-save'
      [state[:rules], fake_status(true)]
    when 'iptables-restore'
      rollback_called = true
      [state[:rules], fake_status(true)]
    when 'iptables'
      if args[0] == '-S'
        [state[:rules], fake_status(true)]
      elsif args[0] == '-A' && args.include?('garbage')
        ['iptables: Bad argument', fake_status(false)] # simulate rejection of the bad rule only
      elsif args[0] == '-A'
        ['', fake_status(true)] # replaying a pre-existing rule during rollback succeeds
      else
        ['', fake_status(true)]
      end
    end
  end
  mgr = IptablesManager.new(backup_dir: tmpdir, runner: runner)
  begin
    mgr.add_rule(chain: 'INPUT', spec: '-p tcp --dport garbage -j ACCEPT')
    check('should have raised', false)
  rescue IptablesManager::CommandError => e
    check('raises CommandError on rejected rule', true)
    check('error message mentions rollback', e.message.include?('rolled back'))
  end
end
puts "\n#{$failures.to_i.zero? ? 'ALL TESTS PASSED' : "#{$failures} TEST(S) FAILED"}"
exit($failures.to_i.zero? ? 0 : 1)