Verify TLS certificates for rediss:// queue URLs by default

ruby/README.md

@@ -260,3 +260,17 @@ After merging changes to `main`, follow these steps to release and propagate the
 `ci-queue` expects the Redis server to have an [eviction policy](https://redis.io/docs/manual/eviction/#eviction-policies) of `allkeys-lru`.
 
 You can also use `--redis-ttl` to set a custom expiration time for all CI Queue keys, this defaults to 8 hours (28,800 seconds)
+
+## Redis over TLS
+
+Use a `rediss://` queue URL to connect to Redis over TLS. The server certificate and hostname are verified using the system's trusted CAs.
+
+> **Upgrading from 0.99.0 or earlier:** those versions did not verify certificates for `rediss://` URLs. If your Redis presents a self-signed or otherwise untrusted certificate, connections now fail with `certificate verify failed`. Workers treat this like an unreachable Redis: they report `Ran 0 tests` and exit successfully, and the error only surfaces when `report` fails with `Redis::CannotConnectError`. Trust the issuing CA or opt out as described below.
+
+If your Redis server presents a certificate signed by a private CA, point OpenSSL at that CA (for example with `SSL_CERT_FILE=/path/to/ca.pem`).
+
+If your hosted Redis only offers a self-signed certificate that can't be verified, you can opt out of verification with `CI_QUEUE_REDIS_SSL_VERIFY=0` (or `CI::Queue::Configuration#redis_ssl_verify = false`). The connection stays encrypted, but the server isn't authenticated, so anyone who can intercept traffic between your CI workers and Redis can impersonate the server and read the credentials in the queue URL. Only use this when you accept that risk.
+
+| Environment variable | Description |
+|---|---|
+| `CI_QUEUE_REDIS_SSL_VERIFY=0` | Disable TLS certificate verification for `rediss://` connections. Defaults to enabled. No CLI equivalent. |

ruby/lib/ci/queue/configuration.rb

@@ -11,6 +11,7 @@ class Configuration
       attr_writer :lazy_load_streaming_timeout
       attr_accessor :lazy_load_test_helpers
       attr_accessor :skip_stale_tests
+      attr_accessor :redis_ssl_verify
       attr_reader :circuit_breakers
       attr_writer :seed, :build_id
       attr_writer :queue_init_timeout, :report_timeout, :inactive_workers_timeout
@@ -34,6 +35,7 @@ def from_env(env)
             lazy_load_streaming_timeout: (env['CI_QUEUE_LAZY_LOAD_STREAM_TIMEOUT'] || env['CI_QUEUE_STREAM_TIMEOUT'])&.to_i,
             lazy_load_test_helpers: env['CI_QUEUE_LAZY_LOAD_TEST_HELPERS'] || env['CI_QUEUE_TEST_HELPERS'],
             skip_stale_tests: %w(1 true).include?(env['CI_QUEUE_SKIP_STALE_TESTS']&.strip&.downcase),
+            redis_ssl_verify: !%w(0 false).include?(env['CI_QUEUE_REDIS_SSL_VERIFY']&.strip&.downcase),
           )
         end
 
@@ -60,7 +62,7 @@ def initialize(
         queue_init_timeout: nil, redis_ttl: 8 * 60 * 60, report_timeout: nil, inactive_workers_timeout: nil,
         export_flaky_tests_file: nil, warnings_file: nil, debug_log: nil, max_missed_heartbeat_seconds: nil, heartbeat_max_test_duration: nil,
         lazy_load: false, lazy_load_stream_batch_size: nil, lazy_load_streaming_timeout: nil, lazy_load_test_helpers: nil,
-        skip_stale_tests: false)
+        skip_stale_tests: false, redis_ssl_verify: true)
         @build_id = build_id
         @circuit_breakers = [CircuitBreaker::Disabled]
         @failure_file = failure_file
@@ -94,6 +96,7 @@ def initialize(
         @lazy_load_streaming_timeout = lazy_load_streaming_timeout
         @lazy_load_test_helpers = lazy_load_test_helpers
         @skip_stale_tests = skip_stale_tests
+        @redis_ssl_verify = redis_ssl_verify
       end
 
       def lazy_load_test_helper_paths

ruby/lib/ci/queue/redis/base.rb

@@ -41,19 +41,33 @@ def initialize(redis_url, config)
               # it makes sense to retry for a while before giving up.
               reconnect_attempts: reconnect_attempts,
               middlewares: custom_middlewares,
-              # Hosted Redis servers use self signed certificates
-              # (because they do not own the domain they're running on such as compute-1.amazonaws.com)
-              # therefore a full SSL connection verification will fail.
-              # ci-queue should not contain any sensitive data, so we can just disable the verification.
-              ssl_params: { verify_mode: OpenSSL::SSL::VERIFY_NONE },
+              ssl_params: self.class.redis_ssl_params(ssl_verify: redis_ssl_verify?),
               custom: custom_config,
               timeout: DEFAULT_TIMEOUT,
             )
           else
-            @redis = ::Redis.new(url: redis_url, timeout: DEFAULT_TIMEOUT)
+            @redis = ::Redis.new(
+              url: redis_url,
+              timeout: DEFAULT_TIMEOUT,
+              ssl_params: self.class.redis_ssl_params(ssl_verify: redis_ssl_verify?),
+            )
           end
         end
 
+        # TLS (`rediss://`) connections verify the server certificate and hostname
+        # by default. Some hosted Redis providers present self-signed certificates;
+        # those deployments can either trust the issuing CA (e.g. via `SSL_CERT_FILE`)
+        # or explicitly opt out of verification with `CI_QUEUE_REDIS_SSL_VERIFY=0`.
+        def self.redis_ssl_params(ssl_verify:)
+          ssl_verify ? {} : { verify_mode: OpenSSL::SSL::VERIFY_NONE }
+        end
+
+        def redis_ssl_verify?
+          return true unless @config.respond_to?(:redis_ssl_verify)
+
+          @config.redis_ssl_verify != false
+        end
+
         def reconnect_attempts
           return [] if ENV["CI_QUEUE_DISABLE_RECONNECT_ATTEMPTS"]
 
@@ -271,8 +285,9 @@ class HeartbeatProcess
           # on every heartbeat, which fires once per running test per worker.
           TICK_COMMAND = 'tick!'.freeze
 
-          def initialize(redis_url, zset_key, owners_key, leases_key)
+          def initialize(redis_url, zset_key, owners_key, leases_key, ssl_verify: true)
             @redis_url = redis_url
+            @ssl_verify = ssl_verify
             @zset_key = zset_key
             @owners_key = owners_key
             @leases_key = leases_key
@@ -283,6 +298,7 @@ def boot!
             ready_pipe, child_write = IO.pipe
             @pipe.binmode
             @pid = Process.spawn(
+              { 'CI_QUEUE_REDIS_SSL_VERIFY' => @ssl_verify ? '1' : '0' },
               RbConfig.ruby,
               ::File.join(__dir__, "monitor.rb"),
               @redis_url,
@@ -380,6 +396,7 @@ def heartbeat_process
             key('running'),
             key('owners'),
             key('leases'),
+            ssl_verify: redis_ssl_verify?,
           )
         end
 

ruby/lib/ci/queue/redis/monitor.rb

@@ -3,6 +3,7 @@
 # frozen_string_literal: true
 
 require 'logger'
+require 'openssl'
 require 'redis'
 require 'json'
 
@@ -18,7 +19,11 @@ def initialize(pipe, logger, redis_url, zset_key, owners_key, leases_key)
           @owners_key = owners_key
           @leases_key = leases_key
           @logger = logger
-          @redis = ::Redis.new(url: redis_url, reconnect_attempts: [0, 0, 0.1, 0.5, 1, 3, 5])
+          @redis = ::Redis.new(
+            url: redis_url,
+            reconnect_attempts: [0, 0, 0.1, 0.5, 1, 3, 5],
+            ssl_params: ssl_params,
+          )
           @shutdown = false
           @pipe = pipe
           @self_pipe_reader, @self_pipe_writer = IO.pipe
@@ -30,6 +35,17 @@ def initialize(pipe, logger, redis_url, zset_key, owners_key, leases_key)
           end
         end
 
+        # Mirrors CI::Queue::Redis::Base.redis_ssl_params. This script runs as a
+        # standalone subprocess that doesn't load ci-queue or its Configuration,
+        # so the parent passes the setting through the environment.
+        def ssl_params
+          if %w(0 false).include?(ENV['CI_QUEUE_REDIS_SSL_VERIFY']&.strip&.downcase)
+            { verify_mode: OpenSSL::SSL::VERIFY_NONE }
+          else
+            {}
+          end
+        end
+
         def soft_signal(sig)
           @queue << sig
           @self_pipe_writer << '.'

ruby/test/ci/queue/configuration_test.rb

@@ -77,6 +77,20 @@ def test_redis_ttl_from_env
       assert_equal(14_400, config.redis_ttl)
     end
 
+    def test_redis_ssl_verify_defaults_to_true
+      assert Configuration.new.redis_ssl_verify
+      assert Configuration.from_env({}).redis_ssl_verify
+    end
+
+    def test_redis_ssl_verify_from_env
+      %w(0 false FALSE).each do |value|
+        refute Configuration.from_env("CI_QUEUE_REDIS_SSL_VERIFY" => value).redis_ssl_verify, value
+      end
+      %w(1 true).each do |value|
+        assert Configuration.from_env("CI_QUEUE_REDIS_SSL_VERIFY" => value).redis_ssl_verify, value
+      end
+    end
+
     def test_parses_file_correctly
       Tempfile.open('flaky_test_file') do |file|
         file.write(SharedTestCases::TEST_NAMES.join("\n") + "\n")

ruby/test/ci/queue/redis_test.rb

@@ -624,6 +624,65 @@ def test_initialise_from_rediss_uri
     assert_instance_of CI::Queue::Redis::Worker, queue
   end
 
+  def test_rediss_uri_rejects_untrusted_server_certificate_by_default
+    stub = TLSRedisStub.new
+    queue = without_reconnect_attempts { CI::Queue.from_uri(stub.url, config) }
+
+    error = assert_raises(Redis::CannotConnectError) { queue.send(:redis).ping }
+    assert_match(/certificate verify failed/, error.message)
+    assert_equal :handshake_failed, stub.next_event
+  ensure
+    stub&.close
+  end
+
+  def test_rediss_uri_verifies_server_hostname_by_default
+    queue = CI::Queue.from_uri('rediss://localhost:6379/0', config)
+    context = queue.send(:redis)._client.config.ssl_context
+    assert_equal OpenSSL::SSL::VERIFY_PEER, context.verify_mode
+    assert context.verify_hostname
+  end
+
+  def test_rediss_uri_can_explicitly_opt_out_of_certificate_verification
+    stub = TLSRedisStub.new
+    config.redis_ssl_verify = false
+    queue = CI::Queue.from_uri(stub.url, config)
+
+    queue.send(:redis).ping
+    assert stub.wait_for_command('PING')
+  ensure
+    stub&.close
+  end
+
+  def test_heartbeat_monitor_rejects_untrusted_server_certificate_by_default
+    stub = TLSRedisStub.new
+    config.max_missed_heartbeat_seconds = 1
+    queue = CI::Queue.from_uri(stub.url, config)
+    queue.boot_heartbeat_process!
+
+    queue.send(:heartbeat_process).tick!('entry', 'lease')
+    assert_equal :handshake_failed, stub.next_event
+  ensure
+    kill_heartbeat_process(queue)
+    stub&.close
+  end
+
+  def test_heartbeat_monitor_follows_certificate_verification_opt_out
+    stub = TLSRedisStub.new
+    config.max_missed_heartbeat_seconds = 1
+    config.redis_ssl_verify = false
+    queue = CI::Queue.from_uri(stub.url, config)
+    queue.boot_heartbeat_process!
+
+    queue.send(:heartbeat_process).tick!('entry', 'lease')
+    event = stub.next_event
+    refute_equal :handshake_failed, event, 'monitor did not honor the verification opt-out'
+    assert_equal %w(script load), event.first(2).map(&:downcase)
+    assert_predicate queue.stop_heartbeat!, :success?
+  ensure
+    kill_heartbeat_process(queue)
+    stub&.close
+  end
+
   def test_first_reserve_at_is_set_on_first_reserve
     queue = worker(1)
     assert_nil queue.first_reserve_at
@@ -1051,4 +1110,29 @@ def worker(id, **args)
       populate(queue, tests: tests)
     end
   end
+
+  # A failed TLS handshake is otherwise retried for ~10 seconds.
+  def without_reconnect_attempts
+    original = ENV['CI_QUEUE_DISABLE_RECONNECT_ATTEMPTS']
+    ENV['CI_QUEUE_DISABLE_RECONNECT_ATTEMPTS'] = '1'
+    yield
+  ensure
+    if original.nil?
+      ENV.delete('CI_QUEUE_DISABLE_RECONNECT_ATTEMPTS')
+    else
+      ENV['CI_QUEUE_DISABLE_RECONNECT_ATTEMPTS'] = original
+    end
+  end
+
+  # A monitor that can't connect keeps retrying for ~10 seconds, so don't wait
+  # for a clean exit.
+  def kill_heartbeat_process(queue)
+    pid = queue&.send(:heartbeat_process)&.instance_variable_get(:@pid)
+    return unless pid
+
+    Process.kill(:KILL, pid)
+    Process.wait(pid)
+  rescue Errno::ESRCH, Errno::ECHILD
+    nil
+  end
 end

ruby/test/support/tls_redis_stub.rb

@@ -0,0 +1,90 @@
+# frozen_string_literal: true
+require 'openssl'
+require 'socket'
+require 'timeout'
+
+# Minimal TLS endpoint that speaks just enough RESP to observe whether a
+# client completed the handshake and what commands it sent. It presents a
+# self-signed certificate, so a verifying client must reject it.
+class TLSRedisStub
+  def self.ssl_context
+    @ssl_context ||= begin
+      key = OpenSSL::PKey::RSA.new(2048)
+      cert = OpenSSL::X509::Certificate.new
+      cert.version = 2
+      cert.serial = 1
+      cert.subject = cert.issuer = OpenSSL::X509::Name.parse('/CN=127.0.0.1')
+      cert.public_key = key.public_key
+      cert.not_before = Time.now - 60
+      cert.not_after = Time.now + 3600
+      cert.sign(key, OpenSSL::Digest::SHA256.new)
+
+      context = OpenSSL::SSL::SSLContext.new
+      context.cert = cert
+      context.key = key
+      context
+    end
+  end
+
+  def initialize
+    @tcp_server = TCPServer.new('127.0.0.1', 0)
+    @server = OpenSSL::SSL::SSLServer.new(@tcp_server, self.class.ssl_context)
+    @server.start_immediately = false
+    @events = Thread::Queue.new
+    @thread = Thread.new { accept_loop }
+  end
+
+  def url
+    "rediss://127.0.0.1:#{@tcp_server.addr[1]}/0"
+  end
+
+  # Returns the next event: :handshake_failed, or an Array with a command's arguments.
+  def next_event(timeout: 10)
+    Timeout.timeout(timeout) { @events.pop }
+  end
+
+  def wait_for_command(name, timeout: 10)
+    Timeout.timeout(timeout) do
+      loop do
+        event = @events.pop
+        return event if event.is_a?(Array) && event.first.casecmp?(name)
+      end
+    end
+  end
+
+  def close
+    @thread.kill
+    @server.close
+  end
+
+  private
+
+  def accept_loop
+    loop do
+      socket = @server.accept
+      Thread.new { serve(socket) }
+    end
+  end
+
+  def serve(socket)
+    socket.accept
+    while (command = read_command(socket))
+      @events << command
+      socket.write("+OK\r\n")
+    end
+  rescue OpenSSL::SSL::SSLError
+    @events << :handshake_failed
+  rescue IOError, SystemCallError
+    nil
+  ensure
+    socket.close rescue nil
+  end
+
+  def read_command(socket)
+    header = socket.gets("\r\n") or return
+    Array.new(header[1..].to_i) do
+      length = socket.gets("\r\n")[1..].to_i
+      socket.read(length + 2)[0, length]
+    end
+  end
+end