java-topology/defects/capistrano/unit/test_capistrano_cwe1333.rb

#!/usr/bin/env ruby
# frozen_string_literal: true
#
# CWE-1333 benchmark for Capistrano defects capistrano-0001 and capistrano-0002.
#
# UNDF: UNDF-2026-000001268 (capistrano-0001, timeout wrapper)
#       UNDF-2026-000001272 (capistrano-0002, RE2 correct fix)
# Patches:
#   capistrano-0001.patch
#   capistrano-0002-host-regex-redos-re2.patch
#
# Defect: ENV["HOSTS"] / ENV["ROLES"] values compiled with Regexp.new() without
#   timeout. An adversarial pattern like "(a+)+$" against a non-matching hostname
#   triggers catastrophic Oniguruma backtracking, hanging the deploy process.
#
# Fix capistrano-0001: Regexp.timeout= (Ruby 3.2+) with Timeout.timeout fallback.
# Fix capistrano-0002: RE2::Regexp.new() -- Thompson NFA, O(N), no backtracking.
#
# Complexity gate:
#   pattern "(a+)+$" against "a"*25+"b" must complete in <2s with fix
#   RE2 version must complete in <50ms

require "timeout"

ADVERSARIAL_PATTERN = "(a+)+$"
ADVERSARIAL_HOST_25  = "a" * 25 + "b"
ADVERSARIAL_HOST_20  = "a" * 20 + "b"
BENIGN_PATTERN       = "^web-\\d+$"
BENIGN_HOSTS         = %w[web-01 web-02 db-01 cache-01]
BENIGN_MATCHES       = %w[web-01 web-02]

REGEX_TIMEOUT = 1.0   # seconds


# ---------------------------------------------------------------------------
# Before: bare Regexp.new without timeout
# ---------------------------------------------------------------------------

def filter_hosts_before(hosts, pattern_str)
  re = Regexp.new(pattern_str)
  hosts.select { |h| re.match?(h) }
end


# ---------------------------------------------------------------------------
# After (capistrano-0001): Regexp.timeout= / Timeout.timeout wrapper
# ---------------------------------------------------------------------------

def safe_compile_regex(pattern)
  Regexp.new(pattern)
rescue RegexpError => e
  warn "[capistrano] Invalid filter regex #{pattern.inspect}: #{e}"
  nil
end

def safe_match?(regex, string)
  return false if regex.nil?

  if Regexp.respond_to?(:timeout=)
    old = Regexp.timeout
    Regexp.timeout = REGEX_TIMEOUT
    begin
      regex.match?(string)
    rescue Regexp::TimeoutError
      warn "[capistrano] Regex timeout matching #{string.inspect} -- excluding host"
      false
    ensure
      Regexp.timeout = old
    end
  else
    begin
      Timeout.timeout(REGEX_TIMEOUT) { regex.match?(string) }
    rescue Timeout::Error
      warn "[capistrano] Regex timeout matching #{string.inspect} -- excluding host"
      false
    end
  end
end

def filter_hosts_safe_wrapper(hosts, pattern_str)
  re = safe_compile_regex(pattern_str)
  hosts.select { |h| safe_match?(re, h) }
end


# ---------------------------------------------------------------------------
# After (capistrano-0002): RE2 correct fix
# ---------------------------------------------------------------------------

def filter_hosts_re2(hosts, pattern_str)
  begin
    require "re2"
    re = RE2::Regexp.new(pattern_str)
    hosts.select { |h| re.match?(h) }
  rescue LoadError
    raise "re2 gem not installed -- skipping RE2 test"
  end
end


# ---------------------------------------------------------------------------
# Test runner
# ---------------------------------------------------------------------------

PASS = [].freeze
FAIL = [].freeze

def assert_equal(expected, actual, msg)
  if expected == actual
    puts "PASS #{msg}"
  else
    puts "FAIL #{msg}: expected #{expected.inspect}, got #{actual.inspect}"
    exit 1
  end
end

def assert_lt(value, limit, msg)
  if value < limit
    puts "PASS #{msg} (#{(value * 1000).round(1)}ms < #{(limit * 1000).round}ms)"
  else
    puts "FAIL #{msg}: #{(value * 1000).round(1)}ms >= #{(limit * 1000).round}ms"
    exit 1
  end
end


# Test 1: benign pattern correctness
before_result = filter_hosts_before(BENIGN_HOSTS, BENIGN_PATTERN).sort
after_result  = filter_hosts_safe_wrapper(BENIGN_HOSTS, BENIGN_PATTERN).sort
assert_equal before_result, after_result, "capistrano-0001 benign correctness: both return #{after_result.inspect}"


# Test 2: invalid pattern returns empty (not exception)
begin
  result = filter_hosts_safe_wrapper(BENIGN_HOSTS, "[invalid")
  assert_equal [], result, "capistrano-0001 invalid pattern returns []"
rescue => e
  puts "FAIL capistrano-0001 invalid pattern raised: #{e}"
  exit 1
end


# Test 3: adversarial pattern must complete in <2s
t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
result = filter_hosts_safe_wrapper([ADVERSARIAL_HOST_25], ADVERSARIAL_PATTERN)
elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0

assert_equal [], result, "capistrano-0001 adversarial returns []"
assert_lt elapsed, 2.0, "capistrano-0001 complexity gate: adversarial N=25"


# Test 4: benign pattern still matches correct hosts after fix
result = filter_hosts_safe_wrapper(BENIGN_HOSTS, BENIGN_PATTERN).sort
assert_equal BENIGN_MATCHES.sort, result, "capistrano-0001 benign match after fix"


# Test 5: RE2 correct fix (if available)
begin
  t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  result = filter_hosts_re2([ADVERSARIAL_HOST_25], ADVERSARIAL_PATTERN)
  elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0
  assert_equal [], result, "capistrano-0002 RE2 adversarial returns []"
  assert_lt elapsed, 0.05, "capistrano-0002 complexity gate: RE2 N=25"
rescue RuntimeError => e
  puts "SKIP capistrano-0002 RE2: #{e}"
end


# Test 6: N=20 adversarial also completes fast
t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
result = filter_hosts_safe_wrapper([ADVERSARIAL_HOST_20], ADVERSARIAL_PATTERN)
elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0
assert_equal [], result, "capistrano-0001 N=20 adversarial returns []"
assert_lt elapsed, 2.0, "capistrano-0001 N=20 adversarial"


puts "ALL PASS"