fix: add tls-ca-cert-file and tls-ca-cert-dir flags to allow tls certificate validation (#1515)

1. add tls-ca-cert-file flag
2. add tls-ca-cert-dir flag
3. enables redis-cli to connect over tls without --insecure flag by properly validating certificate wtih CA

src/facade/dragonfly_listener.cc

@@ -25,6 +25,8 @@ ABSL_FLAG(bool, conn_use_incoming_cpu, false,
 
 ABSL_FLAG(string, tls_cert_file, "", "cert file for tls connections");
 ABSL_FLAG(string, tls_key_file, "", "key file for tls connections");
+ABSL_FLAG(string, tls_ca_cert_file, "", "ca signed certificate to validate tls connections");
+ABSL_FLAG(string, tls_ca_cert_dir, "", "ca signed certificates directory");
 
 #if 0
 enum TlsClientAuth {
@@ -58,6 +60,7 @@ namespace {
 static SSL_CTX* CreateSslCntx() {
   SSL_CTX* ctx = SSL_CTX_new(TLS_server_method());
   const auto& tls_key_file = GetFlag(FLAGS_tls_key_file);
+  unsigned mask = SSL_VERIFY_NONE;
   if (tls_key_file.empty()) {
     // To connect - use openssl s_client -cipher with either:
     // "AECDH:@SECLEVEL=0" or "ADH:@SECLEVEL=0" setting.
@@ -77,19 +80,24 @@ static SSL_CTX* CreateSslCntx() {
     if (!tls_cert_file.empty()) {
       // TO connect with redis-cli you need both tls-key-file and tls-cert-file
       // loaded. Use `redis-cli --tls -p 6380 --insecure  PING` to test
-
       CHECK_EQ(1, SSL_CTX_use_certificate_chain_file(ctx, tls_cert_file.c_str()));
     }
+
+    const auto tls_ca_cert_file = GetFlag(FLAGS_tls_ca_cert_file);
+    const auto tls_ca_cert_dir = GetFlag(FLAGS_tls_ca_cert_dir);
+    if (!tls_ca_cert_file.empty() || !tls_ca_cert_dir.empty()) {
+      const auto* file = tls_ca_cert_file.empty() ? nullptr : tls_ca_cert_file.data();
+      const auto* dir = tls_ca_cert_dir.empty() ? nullptr : tls_ca_cert_dir.data();
+      CHECK_EQ(1, SSL_CTX_load_verify_locations(ctx, file, dir));
+      mask = SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT;
+    }
+
     CHECK_EQ(1, SSL_CTX_set_cipher_list(ctx, "DEFAULT"));
   }
   SSL_CTX_set_min_proto_version(ctx, TLS1_2_VERSION);
 
   SSL_CTX_set_options(ctx, SSL_OP_DONT_INSERT_EMPTY_FRAGMENTS);
 
-  unsigned mask = SSL_VERIFY_NONE;
-
-  // if (tls_auth_clients_opt)
-  // mask |= SSL_VERIFY_FAIL_IF_NO_PEER_CERT;
   SSL_CTX_set_verify(ctx, mask, NULL);
 
   CHECK_EQ(1, SSL_CTX_set_dh_auto(ctx, 1));

tests/dragonfly/__init__.py

@@ -160,10 +160,6 @@ class DflyInstanceFactory:
     def __str__(self):
         return f"Factory({self.args})"
 
-    @property
-    def dfly_path(self):
-        return str(os.path.dirname(self.params.path))
-
 
 def dfly_args(*args):
     """ Used to define a singular set of arguments for dragonfly test """

tests/dragonfly/conftest.py

@@ -13,12 +13,13 @@ import redis
 import pymemcache
 import random
 import subprocess
+from copy import deepcopy
 
 from pathlib import Path
 from tempfile import TemporaryDirectory
 
 from . import DflyInstance, DflyInstanceFactory, DflyParams, PortPicker, dfly_args
-from .utility import DflySeederFactory
+from .utility import DflySeederFactory, gen_certificate
 
 logging.getLogger('asyncio').setLevel(logging.WARNING)
 
@@ -229,42 +230,58 @@ def memcached_connection(df_server: DflyInstance):
 
 
 @pytest.fixture(scope="session")
-def gen_tls_cert(df_factory: DflyInstanceFactory):
-    tls_server_key_file_name = "df-key.pem"
-    tls_server_cert_file_name = "df-cert.pem"
-    dfly_path = df_factory.dfly_path
+def gen_ca_cert(tmp_dir):
     # We first need to generate the tls certificates to be used by the server
 
-    # Step 1
     # Generate CA (certificate authority) key and self-signed certificate
     # In production, CA should be generated by a third party authority
     # Expires in one day and is not encrtypted (-nodes)
     # X.509 format for the key
-    ca_key = dfly_path + "ca-key.pem"
-    ca_cert = dfly_path + "ca-cert.pem"
-    step1 = rf'openssl req -x509 -newkey rsa:4096 -days 1 -nodes -keyout {ca_key} -out {ca_cert} -subj "/C=GR/ST=SKG/L=Thessaloniki/O=KK/OU=AcmeStudios/CN=Gr/emailAddress=acme@gmail.com"'
-    subprocess.run(step1, shell=True)
+    ca_key = os.path.join(tmp_dir, "ca-key.pem")
+    ca_cert = os.path.join(tmp_dir, "ca-cert.pem")
+    step = rf'openssl req -x509 -newkey rsa:4096 -days 1 -nodes -keyout {ca_key} -out {ca_cert} -subj "/C=GR/ST=SKG/L=Thessaloniki/O=KK/OU=AcmeStudios/CN=Gr/emailAddress=acme@gmail.com"'
+    subprocess.run(step, shell=True)
 
-    # Step 2
-    # Generate Dragonfly's private key and certificate signing request (CSR)
-    tls_server_key = dfly_path + tls_server_key_file_name
-    tls_server_req = dfly_path + "df-req.pem"
-    step2 = rf'openssl req -newkey rsa:4096 -nodes -keyout {tls_server_key} -out {tls_server_req} -subj "/C=GR/ST=SKG/L=Thessaloniki/O=KK/OU=Comp/CN=Gr/emailAddress=does_not_exist@gmail.com"'
-    subprocess.run(step2, shell=True)
-
-    # Step 3
-    # Use CA's private key to sign dragonfly's CSR and get back the signed certificate
-    tls_server_cert = dfly_path + tls_server_cert_file_name
-    step3 = fr'openssl x509 -req -in {tls_server_req} -days 1 -CA {ca_cert} -CAkey {ca_key} -CAcreateserial -out {tls_server_cert}'
-    subprocess.run(step3, shell=True)
-    return tls_server_key_file_name, tls_server_cert_file_name
+    return {"ca_key": ca_key, "ca_cert": ca_cert}
 
 
 @pytest.fixture(scope="session")
-def with_tls_args(df_factory: DflyInstanceFactory, gen_tls_cert):
-    tls_server_key_file_name, tls_server_cert_file_name = gen_tls_cert
+def with_tls_server_args(tmp_dir, gen_ca_cert):
+    tls_server_key = os.path.join(tmp_dir, "df-key.pem")
+    tls_server_req = os.path.join(tmp_dir, "df-req.pem")
+    tls_server_cert = os.path.join(tmp_dir, "df-cert.pem")
+
+    gen_certificate(gen_ca_cert["ca_key"], gen_ca_cert["ca_cert"], tls_server_req, tls_server_key, tls_server_cert)
+
     args = {"tls": "",
-            "tls_key_file": df_factory.dfly_path + tls_server_key_file_name,
-            "tls_cert_file": df_factory.dfly_path + tls_server_cert_file_name,
-            "no_tls_on_admin_port": "true"}
+            "tls_key_file": tls_server_key,
+            "tls_cert_file": tls_server_cert}
+    return args
+
+
+@pytest.fixture(scope="session")
+def with_ca_tls_server_args(with_tls_server_args, gen_ca_cert):
+    args = deepcopy(with_tls_server_args)
+    args["tls_ca_cert_file"] = gen_ca_cert["ca_cert"]
+    return args
+
+
+@pytest.fixture(scope="session")
+def with_tls_client_args(tmp_dir, gen_ca_cert):
+    tls_client_key = os.path.join(tmp_dir, "client-key.pem")
+    tls_client_req = os.path.join(tmp_dir, "client-req.pem")
+    tls_client_cert = os.path.join(tmp_dir, "client-cert.pem")
+
+    gen_certificate(gen_ca_cert["ca_key"], gen_ca_cert["ca_cert"], tls_client_req, tls_client_key, tls_client_cert)
+
+    args = {"ssl": True,
+            "ssl_keyfile": tls_client_key,
+            "ssl_certfile": tls_client_cert}
+    return args
+
+
+@pytest.fixture(scope="session")
+def with_ca_tls_client_args(with_tls_client_args, gen_ca_cert):
+    args = deepcopy(with_tls_client_args)
+    args["ssl_ca_certs"] = gen_ca_cert["ca_cert"]
     return args

tests/dragonfly/connection_test.py

@@ -7,6 +7,7 @@ import async_timeout
 
 from . import DflyInstance, dfly_args
 
+BASE_PORT = 1111
 
 async def run_monitor_eval(monitor, expected):
     async with monitor as mon:
@@ -419,18 +420,61 @@ async def test_large_cmd(async_client: aioredis.Redis):
 
 
 @pytest.mark.asyncio
-async def test_reject_non_tls_connections_on_tls_master(with_tls_args, df_local_factory):
-    master = df_local_factory.create(admin_port=1111, port=1211, **with_tls_args)
-    master.start()
+async def test_reject_non_tls_connections_on_tls(with_tls_server_args, df_local_factory):
+    server = df_local_factory.create(no_tls_on_admin_port="true", admin_port=1111, port=1211, **with_tls_server_args)
+    server.start()
 
-    # Try to connect on master without admin port. This should fail.
-    client = aioredis.Redis(port=master.port)
+    client = aioredis.Redis(port=server.port)
     try:
         await client.execute_command("DBSIZE")
-        raise "Non tls connection connected on master with tls. This should NOT happen"
     except redis_conn_error:
         pass
 
-    # Try to connect on master on admin port
-    client = aioredis.Redis(port=master.admin_port)
-    assert await client.ping()
+    client = aioredis.Redis(port=server.admin_port)
+    assert await client.dbsize() == 0
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_tls_insecure(with_ca_tls_server_args, with_tls_client_args, df_local_factory):
+    server = df_local_factory.create(port=BASE_PORT, **with_ca_tls_server_args)
+    server.start()
+
+    client = aioredis.Redis(port=server.port, **with_tls_client_args, ssl_cert_reqs=None)
+    assert await client.dbsize() == 0
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_tls_full_auth(with_ca_tls_server_args, with_ca_tls_client_args, df_local_factory):
+    server = df_local_factory.create(port=BASE_PORT, **with_ca_tls_server_args)
+    server.start()
+
+    client = aioredis.Redis(port=server.port, **with_ca_tls_client_args)
+    assert await client.dbsize() == 0
+    await client.close()
+
+
+@pytest.mark.asyncio
+async def test_tls_reject(with_ca_tls_server_args, with_tls_client_args, df_local_factory):
+    server = df_local_factory.create(port=BASE_PORT, **with_ca_tls_server_args)
+    server.start()
+
+    client = aioredis.Redis(port=server.port, **with_tls_client_args, ssl_cert_reqs=None)
+    try:
+        await client.ping()
+    except redis_conn_error:
+        pass
+
+    client = aioredis.Redis(port=server.port, **with_tls_client_args)
+    try:
+        assert await client.dbsize() != 0
+    except redis_conn_error:
+        pass
+
+    client = aioredis.Redis(port=server.port, ssl_cert_reqs=None)
+    try:
+        assert await client.dbsize() != 0
+    except redis_conn_error:
+        pass
+    await client.close()

tests/dragonfly/replication_test.py

@@ -1223,7 +1223,7 @@ async def test_take_over_timeout(df_local_factory, df_seeder_factory):
     assert await c_replica.execute_command("role") == [b'replica', b'localhost', bytes(str(master.port), 'ascii'), b'stable_sync']
 
     await disconnect_clients(c_master, c_replica)
-    
+
 
 # 1. Number of master threads
 # 2. Number of threads for each replica
@@ -1231,9 +1231,9 @@ replication_cases = [(8, 8)]
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize("t_master, t_replica", replication_cases)
-async def test_no_tls_on_admin_port(df_local_factory, df_seeder_factory, t_master, t_replica, with_tls_args):
+async def test_no_tls_on_admin_port(df_local_factory, df_seeder_factory, t_master, t_replica, with_tls_server_args):
     # 1. Spin up dragonfly without tls, debug populate
-    master = df_local_factory.create(admin_port=ADMIN_PORT, **with_tls_args, port=BASE_PORT, proactor_threads=t_master)
+    master = df_local_factory.create(no_tls_on_admin_port="true", admin_port=ADMIN_PORT, **with_tls_server_args, port=BASE_PORT, proactor_threads=t_master)
     master.start()
     c_master = aioredis.Redis(port=master.admin_port)
     await c_master.execute_command("DEBUG POPULATE 100")
@@ -1241,7 +1241,7 @@ async def test_no_tls_on_admin_port(df_local_factory, df_seeder_factory, t_maste
     assert 100 == db_size
 
     # 2. Spin up a replica and initiate a REPLICAOF
-    replica = df_local_factory.create(admin_port=ADMIN_PORT + 1, **with_tls_args, port=BASE_PORT + 1, proactor_threads=t_replica)
+    replica = df_local_factory.create(no_tls_on_admin_port="true", admin_port=ADMIN_PORT + 1, **with_tls_server_args, port=BASE_PORT + 1, proactor_threads=t_replica)
     replica.start()
     c_replica = aioredis.Redis(port=replica.admin_port)
     res = await c_replica.execute_command("REPLICAOF localhost " + str(master.admin_port))
@@ -1251,3 +1251,5 @@ async def test_no_tls_on_admin_port(df_local_factory, df_seeder_factory, t_maste
     # 3. Verify that replica dbsize == debug populate key size -- replication works
     db_size = await c_replica.execute_command("DBSIZE")
     assert 100 == db_size
+    await c_replica.close()
+    await c_master.close()

tests/dragonfly/utility.py

@@ -9,6 +9,7 @@ import itertools
 import time
 import difflib
 import json
+import subprocess
 from enum import Enum
 
 
@@ -560,3 +561,13 @@ class DflySeederFactory:
 
 async def disconnect_clients(*clients):
     await asyncio.gather(*(c.connection_pool.disconnect() for c in clients))
+
+
+def gen_certificate(ca_key_path, ca_certificate_path, certificate_request_path, private_key_path, certificate_path):
+    # Generate Dragonfly's private key and certificate signing request (CSR)
+    step1 = rf'openssl req -newkey rsa:4096 -nodes -keyout {private_key_path} -out {certificate_request_path} -subj "/C=GR/ST=SKG/L=Thessaloniki/O=KK/OU=Comp/CN=Gr/emailAddress=does_not_exist@gmail.com"'
+    subprocess.run(step1, shell=True)
+
+    # Use CA's private key to sign dragonfly's CSR and get back the signed certificate
+    step2 = fr'openssl x509 -req -in {certificate_request_path} -days 1 -CA {ca_certificate_path} -CAkey {ca_key_path} -CAcreateserial -out {certificate_path}'
+    subprocess.run(step2, shell=True)