背景介绍

在原来的天翼云关系型数据库中，重分布时会在执行计划中生成RemoteSubplan算子。在执行时碰到 RemoteSubplan 算子的时候才会往下发整体的下一步查询计划，如果查询中重分布的层次比较多，每一层 DN 都会认为自己是一个发起者，会导致大量多层进程连接和网络连接消耗。为解决这个问题，重分布RDA模块应运而生。重构数据重分布实现，以Socket + 推送方式，实现数据重分布。主要实现方式是TeleDB使用Socket实现数据交换的RDA（Remote Data Access）算子替换现有的RemoteSubplan 算子和SharedQueue逻辑来做非分布键管理的数据重分布。

在做数据重分布时，牵涉到数据库数据在网络中进行交换。当牵涉到用户的敏感数据时，为了防止用户信息的泄露需要对网络通信进行加密处理。这里采用开源的OpenSSL加密算法融合RDA来实现RDA通信加密，OpenSSL 是一个用于实现安全通信的软件包，它由一组密码学函数库组成。它的主要目标是通过使用公开的密码学算法来保护数据的机密性、完整性和身份验证。它支持对称加密、非对称加密、数字签名、证书管理等功能。

模块设计

• 初始化
  • 服务器生成密钥参数；然后参与数据传输的设备与网络功能实体初始化，各自生成应用层密钥和对应的密钥标识信息。
  • 基于应用层密钥和对应的密钥标识信息实现设备与管理器之间的应用层密钥同步。服务器和CA为每个终端生成对应的身份标识，并基于各设备的身份标识、主密钥和公共参数生成对应设备基于身份标识的私钥
  • 在SSL建连中，数据库中的节点将身份标识、私钥、主密钥、系统公共参数等信息发送给各终端，各终端存储在本地。
• 发送端
  • 通过加密密钥用于加密将待传输信息生成第一密文
  • 根据终端私钥对预先生成的动态因子、第一密文和身份标识生成签名信息
  • 将签名信息、动态因子、第一密文和第一终端的身份标识发送至第二终端
• 接收端
  • 第二终端根据预设的对称密钥因子、动态因子以及第一终端的身份标识生成对称解密密钥以及与第一终端的私钥对应的公钥，以通过公钥对签名信息进行验证，并在验证通过后根据对称解密密钥对第一密文进行解密得到待传输信息

流程设计

系统总体的流程如下图所示，节点在本地生成密钥和数字证书，在数据库集群初始化完成之后通过TCP进行网络建连之后，再进SSL的一系列的握手操作，包括密钥交换和身份验证等，以建立安全的通信通道。之后基于该安全通道对要进行重分布数据进行加解密来实现TeleDB数据库中不同节点的加密通信。

• 生成密钥

  openssl genrsa -out ca.key
  

• 生成根证书

  openssl req -new -x509 -days 3650 -key ca.key -out ca.crt
  

• 生成客户端私钥

  openssl genrsa -out client_private.pem
  chmod 600 client_private.pem
  

• 生成客户端证书请求

  openssl req -new -key client_private.pem -out client.csr
  

• 使用本地证书、私钥生成客户端证书

  openssl x509 -req -days 3650 -in client.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out client.crt
  

postgresql.conf.user配置

fwd_ssl = on
fwd_ssl_cert_file = '/home/ctyun/openssl/client.crt'
fwd_ssl_key_file = '/home/ctyun/openssl/client_private.pem'
fwd_ssl_ca_file = '/home/ctyun/openssl/ca.crt'

伪代码实现

添加fwd_ssl开关

• 针对不同的用户使用场景，通过开关 fwd_ssl 决定是否加密

socket管理

• 由于加密通信通过SSL对象进行网络数据收发，ConnMgr中封装一个SSL对象

  typedef struct
  {
      int socket;
      SSL *ssl;
      ...
  } ConnMgr;
  

SSL初始化

• 加载证书和私钥

SSL_CTX *g_fwd_server_ssl_ctx = NULL;
SSL_CTX *g_fwd_client_ssl_ctx = NULL;

static void fwd_init_openssl();
static SSL_CTX* fwd_create_server_ssl_ctx();
static SSL_CTX* fwd_create_client_ssl_ctx();
static void fwd_load_ssl_file(SSL_CTX* ctx);

static void fwd_init_openssl()
{
    SSL_load_error_strings();  
    OpenSSL_add_all_algorithms();  
    SSL_library_init();
    
    g_fwd_server_ssl_ctx = fwd_create_server_ssl_ctx();
    fwd_load_ssl_file(g_fwd_server_ssl_ctx);
    g_fwd_client_ssl_ctx = fwd_create_client_ssl_ctx();
    fwd_load_ssl_file(g_fwd_client_ssl_ctx);
}

static SSL_CTX* fwd_create_server_ssl_ctx()
{
    SSL_CTX *ctx;  

    ctx = SSL_CTX_new(SSLv23_server_method());  
    if (!ctx)
    {  
        elog(ERROR, FORWARDER_PREFIX "fwd_create_server_ssl_ctx SSL_CTX_new failed.");
        return NULL;
    } 
    return ctx;
}

static SSL_CTX* fwd_create_client_ssl_ctx()
{
    SSL_CTX *ctx;  

    ctx = SSL_CTX_new(SSLv23_client_method());  
    if (!ctx)
    {  
        elog(ERROR, FORWARDER_PREFIX "fwd_create_client_ssl_ctx SSL_CTX_new failed.");
        return NULL;
    } 
    return ctx;
}

static void fwd_load_ssl_file(SSL_CTX* ctx)
{
    /* 载入用户的数字证书，证书里包含有公钥 */
    if (SSL_CTX_use_certificate_file(ctx, ssl_cert_file, SSL_FILETYPE_PEM) <= 0)
    {  
        elog(ERROR, FORWARDER_PREFIX "fwd_load_ssl_file load server cert failed."); 
    }
    /* 载入用户私钥 */
    if (SSL_CTX_use_PrivateKey_file(ctx, ssl_key_file, SSL_FILETYPE_PEM) <= 0)
    {  
        elog(ERROR, FORWARDER_PREFIX "fwd_load_ssl_file load server key failed.");
    }  
  
    /* 检查用户私钥是否正确 */
    if (!SSL_CTX_check_private_key(ctx))
    {
        elog(ERROR, FORWARDER_PREFIX "fwd_create_ssl_ctx cert and key mismatch.");
    }
    // 加载CA证书
    if (!SSL_CTX_load_verify_locations(ctx, ssl_ca_file, NULL))
    {  
        elog(ERROR, FORWARDER_PREFIX "fwd_create_ssl_ctx load ca cert mismatch.");
    }
    
    // 双向验证
    // SSL_VERIFY_PEER---要求对证书进行认证，没有证书也会放行
    // SSL_VERIFY_FAIL_IF_NO_PEER_CERT---要求客户端需要提供证书，但验证发现单独使用没有证书也会放行
    SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);  
}

客户端

建连函数新增openssl的建连（发送身份标识、密钥）：

static int fwd_connect_timeout(int socketfd, SSL *ssl, struct sockaddr *serv_addr, int addrlen, int timeo)
{
    ...
    
    if (g_fwd_ssl)
    {
        ssl = SSL_new(g_fwd_client_ssl_ctx);
        SSL_set_fd(ssl, socketfd);
        //ssl握手
        if (SSL_connect(ssl) == -1)
        {
            return EOF;
        }
     }
 }

发送数据（加密）

static int fwd_write(ConnMgr* mgr, char *data, int len, int* save_errno)
{
    int ret = 0;
    if (g_fwd_enable_network_encrypt)
    {
        ret = SSL_write(mgr->ssl, data, len);
        if (ret < 0)
        {
            *save_errno = SSL_get_error(mgr->ssl, ret);
        }
    }
    else
    {
        ret = write(mgr->socket, data, len);
        if (ret < 0)
        {
            *save_errno = errno;
        }
    }
    return ret;
}

服务端

接收连接（验证签名信息）

static int fwd_accept_connection(int server_fd, FWDServersPort *port)
{
    ...
    ssl = SSL_new(g_fwd_server_ssl_ctx);
    SSL_set_fd(ssl, con_fd);
    //等待ssl握手
    if (SSL_accept(ssl) == -1)
    {
        forwarder_thread_logger(LOG, FORWARDER_PREFIX "thread:%ld  SSL_accept failed: %s", pthread_self(), strerror(errno));
        return -1;
    } 
}

接收数据（解密）

static int fwd_read(ConnMgr* mgr, char *data, int len, int* save_errno)
{
    int ret = 0;
    if (g_fwd_enable_network_encrypt)
    {
        ret = SSL_read(mgr->ssl, data, len);
        if (ret < 0)
        {
            *save_errno = SSL_get_error(mgr->ssl, ret);
        }
    }
    else
    {
        ret = read(mgr->socket, data, len);
        if (ret < 0)
        {
            *save_errno = errno;
        }
    }
    return ret;
}

资源清理

void fwd_cleanup_socket_array()
{
    ...
    close(g_client_socket_array[i][j].socket);
    SSL_shutdown(ssl);
    SSL_free(ssl);
    SSL_CTX_free(g_fwd_client_ssl_ctx);
}

static void fwd_server_socket_status_destory(void)
{
    ...
    SSL_shutdown(ssl);
    SSL_free(ssl);
    SSL_CTX_free(g_fwd_server_ssl_ctx);
}