diff --git a/Gemfile.lock b/Gemfile.lock
index 53f74b64..5a26a3ac 100644
--- a/Gemfile.lock
+++ b/Gemfile.lock
@@ -236,7 +236,7 @@ GEM
     test-unit (3.5.7)
       power_assert
     thor (1.2.1)
-    tilt (2.1.0)
+    tilt (2.3.0)
     timecop (0.9.1)
     timers (4.3.5)
     tomlrb (2.0.3)
diff --git a/spec/unit/queue_storage_spec.rb b/spec/unit/queue_storage_spec.rb
index 7d04b412..579ec718 100644
--- a/spec/unit/queue_storage_spec.rb
+++ b/spec/unit/queue_storage_spec.rb
@@ -45,7 +45,7 @@ module Backend
       def is_sentinel?(connection)
         if ThreeScale::Backend.configuration.redis.async
           connector = connection.instance_variable_get(:@redis_async)
-          connector.instance_of?(Async::Redis::SentinelsClient)
+          connector.instance_of?(ThreeScale::Backend::AsyncRedis::SentinelsClientACLTLS)
         else
           connector = connection.instance_variable_get(:@client)
                                 .instance_variable_get(:@config)
diff --git a/test/test_helpers/certificates.rb b/test/test_helpers/certificates.rb
new file mode 100644
index 00000000..a4b829bc
--- /dev/null
+++ b/test/test_helpers/certificates.rb
@@ -0,0 +1,67 @@
+# Copyright © 2020 Nicky Peeters
+# Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+# documentation files (the “Software”), to deal in the Software without restriction, including without limitation the
+# rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
+# permit persons to whom the Software is furnished to do so, subject to the following conditions:
+#
+# The above copyright notice and this permission notice shall be included in all copies or substantial portions
+# of the Software.
+# THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO
+# THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+# TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+# SOFTWARE.
+
+require 'rubygems'
+require 'openssl'
+
+module TestHelpers
+  module Certificates
+    def create_key(alg)
+      case alg
+      when :rsa
+        OpenSSL::PKey::RSA.new(2048)
+      when :dsa
+        OpenSSL::PKey::DSA.new(2048)
+      when :ec
+        OpenSSL::PKey::EC.generate("prime256v1")
+      end
+    end
+
+    def create_cert(key = create_key(:rsa))
+      public_key = get_public_key(key)
+
+      subject = "/C=BE/O=Test/OU=Test/CN=Test"
+
+      cert = OpenSSL::X509::Certificate.new
+      cert.subject = cert.issuer = OpenSSL::X509::Name.parse(subject)
+      cert.not_before = Time.now
+      cert.not_after = Time.now + 365 * 24 * 60 * 60
+      cert.public_key = public_key
+      cert.serial = 0x0
+      cert.version = 2
+
+      ef = OpenSSL::X509::ExtensionFactory.new
+      ef.subject_certificate = cert
+      ef.issuer_certificate = cert
+      cert.extensions = [
+        ef.create_extension("basicConstraints","CA:TRUE", true),
+        ef.create_extension("subjectKeyIdentifier", "hash"),
+      # ef.create_extension("keyUsage", "cRLSign,keyCertSign", true),
+      ]
+      cert.add_extension ef.create_extension("authorityKeyIdentifier",
+                                             "keyid:always,issuer:always")
+
+      cert.sign key, (key.is_a?(OpenSSL::PKey::DSA) ? OpenSSL::Digest::SHA1.new : OpenSSL::Digest::SHA512.new)
+      cert
+    end
+
+    private
+
+    def get_public_key(key)
+      return OpenSSL::PKey::EC.new key if key.is_a? OpenSSL::PKey::EC
+
+      key.public_key
+    end
+  end
+end
diff --git a/test/unit/storage_async_test.rb b/test/unit/storage_async_test.rb
index a043eab4..201fb6fa 100644
--- a/test/unit/storage_async_test.rb
+++ b/test/unit/storage_async_test.rb
@@ -1,7 +1,10 @@
+require 'tempfile'
 require File.expand_path(File.dirname(__FILE__) + '/../test_helper')
 require '3scale/backend/storage_async'
 
 class StorageAsyncTest < Test::Unit::TestCase
+  include TestHelpers::Certificates
+
   def test_basic_operations
     storage = StorageAsync::Client.instance(true)
     storage.del('foo')
@@ -34,6 +37,11 @@ def test_redis_malformed_url
     end
   end
 
+  def test_redis_no_scheme
+    storage = StorageAsync::Client.send :new, url('backend-redis')
+    assert_client_config({ url: URI('redis://backend-redis:6379') }, storage)
+  end
+
   def test_sentinels_connection_string
     config_obj = {
       url: 'redis://master-group-name',
@@ -182,30 +190,154 @@ def test_sentinels_empty
     end
   end
 
-  def test_redis_no_scheme
-    storage = StorageAsync::Client.send :new, url('backend-redis')
-    assert_client_config({ url: URI('redis://backend-redis:6379') }, storage)
+  def test_sentinels_acl
+    config_obj = {
+      url: 'redis://master-group-name',
+      sentinels: 'redis://127.0.0.1:26379, redis://127.0.0.1:36379, redis://127.0.0.1:46379',
+      username: 'apisonator-test',
+      password: 'p4ssW0rd',
+      sentinel_username: 'sentinel-test',
+      sentinel_password: 'p4ssW0rd#'
+    }
+
+    conn = StorageAsync::Client.send :new, Storage::Helpers.config_with(config_obj)
+    assert_sentinel_config({ **config_obj,
+                             sentinels: [{ host: '127.0.0.1', port: 26_379 },
+                                         { host: '127.0.0.1', port: 36_379 },
+                                         { host: '127.0.0.1', port: 46_379 }]},
+                           conn)
+  end
+
+  def test_sentinels_acl_tls
+    ca_file, cert, key = create_certs(:rsa).values_at(:ca_file, :cert, :key)
+
+    config_obj = {
+      url: 'rediss://master-group-name',
+      sentinels: 'rediss://127.0.0.1:26379, rediss://127.0.0.1:36379, rediss://127.0.0.1:46379',
+      username: 'apisonator-test',
+      password: 'p4ssW0rd',
+      sentinel_username: 'sentinel-test',
+      sentinel_password: 'p4ssW0rd#',
+      ssl_params: {
+        ca_file: ca_file.path,
+        cert: cert.path,
+        key: key.path
+      }
+    }
+
+    conn = StorageAsync::Client.send :new, Storage::Helpers.config_with(config_obj)
+    assert_sentinel_config({ **config_obj,
+                             sentinels: [{ host: '127.0.0.1', port: 26_379 },
+                                         { host: '127.0.0.1', port: 36_379 },
+                                         { host: '127.0.0.1', port: 46_379 }]},
+                           conn, :rsa)
+  ensure
+    [ca_file, cert, key].each(&:unlink)
+  end
+
+  def test_tls_no_client_certificate
+    config_obj = {
+      url: 'rediss://localhost:46379/0',
+      ssl_params: {
+        ca_file: File.expand_path(File.join(__FILE__, '..', '..', '..', 'script', 'config', 'ca-root-cert.pem'))
+      }
+    }
+    storage = StorageAsync::Client.send :new, Storage::Helpers.config_with(config_obj)
+    assert_client_config(config_obj, storage)
+  end
+
+  [:rsa, :dsa, :ec].each do |alg|
+    define_method "test_tls_client_cert_#{alg}" do
+      ca_file, cert, key = create_certs(alg).values_at(:ca_file, :cert, :key)
+
+      config_obj = {
+        url: 'rediss://localhost:46379/0',
+        ssl_params: {
+          ca_file: ca_file.path,
+          cert: cert.path,
+          key: key.path
+        }
+      }
+      storage = StorageAsync::Client.send :new, Storage::Helpers.config_with(config_obj)
+      assert_client_config(config_obj, storage, alg)
+    ensure
+      [ca_file, cert, key].each(&:unlink)
+    end
+  end
+
+  def test_acl
+    config_obj = {
+      url: 'redis://localhost:6379/0',
+      username: 'apisonator-test',
+      password: 'p4ssW0rd'
+    }
+    storage = StorageAsync::Client.send :new, Storage::Helpers.config_with(config_obj)
+    assert_client_config(config_obj, storage)
+  end
+
+  def test_acl_tls
+    ca_file, cert, key = create_certs(:rsa).values_at(:ca_file, :cert, :key)
+
+    config_obj = {
+      url: 'rediss://localhost:46379/0',
+      ssl_params: {
+        ca_file: ca_file.path,
+        cert: cert.path,
+        key: key.path
+      },
+      username: 'apisonator-test',
+      password: 'p4ssW0rd'
+    }
+    storage = StorageAsync::Client.send :new, Storage::Helpers.config_with(config_obj)
+    assert_client_config(config_obj, storage)
+  ensure
+    [ca_file, cert, key].each(&:unlink)
   end
 
   private
 
-  def assert_client_config(conf, conn)
+  def create_certs(alg)
+    ca_cert_file = Tempfile.new('ca-root-cert.pem')
+    ca_cert_file.write(create_cert.to_pem)
+    ca_cert_file.flush
+    ca_cert_file.close
+
+    key = create_key alg
+    key_file = Tempfile.new("redis-#{alg}.pem")
+    key_file.write(key.to_pem)
+    key_file.flush
+    key_file.close
+
+    cert_file = Tempfile.new("redis-#{alg}.crt")
+    cert_file.write(create_cert(key).to_pem)
+    cert_file.flush
+    cert_file.close
+
+    { ca_file: ca_cert_file, cert: cert_file, key: key_file }
+  end
+
+  def assert_client_config(conf, conn, test_cert_type = nil)
     client = conn.instance_variable_get(:@redis_async)
 
     url = URI(conf[:url])
     host, port = client.endpoint.address
     assert_equal url.host, host
     assert_equal url.port, port
+
+    assert_acl_credentials(conf, client)
+    assert_tls_certs(conf, client, test_cert_type)
   end
 
-  def assert_sentinel_config(conf, conn)
+  def assert_sentinel_config(conf, conn, test_cert_type = nil)
     client = conn.instance_variable_get(:@redis_async)
     uri = URI(conf[:url] || '')
     name = uri.host
     role = conf[:role] || :master
-    password = client.instance_variable_get(:@protocol).instance_variable_get(:@password)
 
-    assert_instance_of Async::Redis::SentinelsClient, client
+    assert_instance_of ThreeScale::Backend::AsyncRedis::SentinelsClientACLTLS, client
+
+    assert_acl_credentials(conf, client)
+    assert_tls_certs(conf, client, test_cert_type)
 
     assert_equal name, client.instance_variable_get(:@master_name)
     assert_equal role, client.instance_variable_get(:@role)
@@ -215,10 +347,48 @@ def assert_sentinel_config(conf, conn)
       host, port = endpoint.address
       assert_equal conf[:sentinels][i][:host], host
       assert_equal conf[:sentinels][i][:port], port
-      assert_equal(conf[:sentinels][i][:password], password) if conf[:sentinels][i].key? :password
     end unless conf[:sentinels].empty?
   end
 
+  def assert_acl_credentials(conf, client)
+    if conf[:username].to_s.strip.empty? && conf[:password].to_s.strip.empty?
+      assert_nil conf[:username]
+      assert_nil conf[:password]
+    else
+      assert_instance_of ThreeScale::Backend::AsyncRedis::Protocol::ExtendedRESP2, client.protocol
+      username, password = client.protocol.instance_variable_get(:@credentials)
+      assert_equal conf[:username], username
+      assert_equal conf[:password], password
+    end
+
+    if conf[:sentinel_username].to_s.strip.empty? && conf[:sentinel_password].to_s.strip.empty?
+      assert_nil conf[:sentinel_username]
+      assert_nil conf[:sentinel_password]
+    else
+      sentinel_username, sentinel_password = client.instance_variable_get(:@sentinel_credentials)
+      assert_equal conf[:sentinel_username], sentinel_username
+      assert_equal conf[:sentinel_password], sentinel_password
+    end
+  end
+
+  def assert_tls_certs(conf, client, test_cert_type)
+    unless conf[:ssl_params].to_s.strip.empty?
+      endpoint = client.endpoint || client.instance_variable_get(:@sentinel_endpoints).first
+      assert_instance_of Async::IO::SSLEndpoint, endpoint
+      assert_equal conf[:ssl_params][:ca_file], endpoint.options[:ssl_context].send(:ca_file)
+      assert_instance_of(OpenSSL::X509::Certificate, endpoint.options[:ssl_context].cert) unless conf[:ssl_params][:cert].to_s.strip.empty?
+
+      unless test_cert_type.to_s.strip.empty?
+        expected_classes = {
+          rsa: OpenSSL::PKey::RSA,
+          dsa: OpenSSL::PKey::DSA,
+          ec: OpenSSL::PKey::EC,
+        }
+        assert_instance_of(expected_classes[test_cert_type], endpoint.options[:ssl_context].key) unless conf[:ssl_params][:key].to_s.strip.empty?
+      end
+    end
+  end
+
   def url(url)
     Storage::Helpers.config_with({ url: url })
   end
diff --git a/test/unit/storage_sync_test.rb b/test/unit/storage_sync_test.rb
index e2110e22..f9ca69f4 100644
--- a/test/unit/storage_sync_test.rb
+++ b/test/unit/storage_sync_test.rb
@@ -244,6 +244,68 @@ def test_sentinels_empty
     end
   end
 
+  def test_ssl_from_url
+    cfg = Storage::Helpers.config_with({url: 'rediss://localhost:46379' })
+    assert cfg[:ssl]
+  end
+
+  def test_ssl_from_param
+    cfg = Storage::Helpers.config_with({url: 'redis://localhost:46379', ssl: true })
+    assert cfg[:ssl]
+  end
+
+  def test_ssl_url_precedence
+    cfg = Storage::Helpers.config_with({url: 'rediss://localhost:46379', ssl: false })
+    assert cfg[:ssl]
+  end
+
+  def test_tls_no_client_certificate
+    config_obj = {
+      url: 'rediss://localhost:46379/0',
+      ssl_params: {
+        ca_file: File.expand_path(File.join(__FILE__, '..', '..', '..', 'script', 'config', 'ca-root-cert.pem'))
+      }
+    }
+    storage = StorageSync.send :new, Storage::Helpers.config_with(config_obj)
+    assert_client_config(config_obj, storage)
+  end
+
+  def test_tls_client_cert
+    config_obj = {
+      url: 'rediss://localhost:46379/0',
+      ssl_params: {
+        ca_file: File.expand_path(File.join(__FILE__, '..', '..', '..', 'script', 'config', 'ca-root-cert.pem')),
+        cert: File.expand_path(File.join(__FILE__, '..', '..', '..', 'script', 'config', 'redis-client.crt')),
+        key: File.expand_path(File.join(__FILE__, '..', '..', '..', 'script', 'config', 'redis-client.key'))
+      }
+    }
+    storage = StorageSync.send :new, Storage::Helpers.config_with(config_obj)
+    assert_client_config(config_obj, storage)
+  end
+
+  def test_acl
+    config_obj = {
+      url: 'redis://localhost:6379/0',
+      username: 'apisonator-test',
+      password: 'p4ssW0rd'
+    }
+    storage = StorageSync.send :new, Storage::Helpers.config_with(config_obj)
+    assert_client_config(config_obj, storage)
+  end
+
+  def test_acl_tls
+    config_obj = {
+      url: 'rediss://localhost:46379/0',
+      ssl_params: {
+        ca_file: File.expand_path(File.join(__FILE__, '..', '..', '..', 'script', 'config', 'ca-root-cert.pem'))
+      },
+      username: 'apisonator-test',
+      password: 'p4ssW0rd'
+    }
+    storage = StorageSync.send :new, Storage::Helpers.config_with(config_obj)
+    assert_client_config(config_obj, storage)
+  end
+
   private
 
   def assert_client_config(conf, conn)
@@ -256,6 +318,15 @@ def assert_client_config(conf, conn)
       assert_equal url.host, config.host
       assert_equal url.port, config.port
     end
+
+    assert_equal conf[:username] || 'default', config.username
+    assert_equal conf[:password], config.password
+
+    unless conf[:ssl_params].to_s.strip.empty?
+      %i[ca_file cert key].each do |key|
+        assert_equal conf[:ssl_params][key], config.ssl_params[key]
+      end
+    end
   end
 
   def assert_sentinel_config(conf, client)