openGauss数据库源码解析系列文章—— 密态等值查询技术详解（下）

上期我们介绍了密态等值查询技术及如何创建客户端密钥CMK，本期继续讲解如何创建列加密密钥CEK和创建加密表。

目录

创建列加密密钥CEK

创建加密表

---

客户端主密钥创建语法本质上是将CMK的元信息解析并保存在CreateClientLogicGlobal结构体中。其中global_key_name是密钥名称，global_setting_params是一个List结构，每个节点是一个ClientLogicGlobalParam结构，以键值的形式保存着密钥的信息。客户端先通过解析器“fe_raw_parser()”解析为CreateClientLogicGlobal结构体，对其参数进行校验并发送查询语句到服务端；服务端解析为CreateClientLogicGlobal结构体并检查用户namespace等权限，CMK元信息保存在系统表中。创建CMK的总体流程如图所示。

创建列加密密钥CEK

 有了主密钥CMK，可以基于此创建CEK，下面将对CREATE COLUMN ENCRYPTION KEY语句所涉及的语法结构定义进行逐一介绍。
CREATE COLUMN ENCRYPTION KEY语法相关数据结构：

/*  保存创建列加密密钥的语法信息  */
typedef struct CreateClientLogicColumn {
    NodeTag type;
    List *column_key_name;        /*  列加密密钥名称  */
    List *column_setting_params; /*  列加密密钥参数  */
} CreateClientLogicColumn;

/*  保存列加密密钥参数，保存在CreateClientLogicColumn的column_setting_params中  */
typedef struct ClientLogicColumnParam {
    NodeTag type;
    ClientLogicColumnProperty key;
    char *value;
    unsigned int len;
    List *qualname;
    int location;
} ClientLogicColumnParam;

/*  保存列加密密钥参数的key的枚举类型  */
typedef enum class ClientLogicColumnProperty {
    CLIENT_GLOBAL_SETTING,    /*  加密CEK的CMK  */
    CEK_ALGORITHM,             /*  加密用户数据的算法  */
    CEK_EXPECTED_VALUE,       /*  CEK密钥明文，可选参数  */
    COLUMN_COLUMN_FUNCTION,  /*  默认为encryption  */
} ClientLogicColumnProperty;

CREATE COLUMN ENCRYPTION KEY cek_1 WITH VALUES (CLIENT_MASTER_KEY = cmk_1, ALGORITHM = AEAD_AES_256_CBC_HMAC_SHA256);

上面命令的参数说明为：
（1） CLIENT_MASTER_KEY：指定用于加密CEK的CMK对象。
（2） ALGORITHM：CEK被用于加密用户数据，该参数指定加密用户数据的算法，即指定CEK的密钥类型。
（3） ENCRYPTED_VALUE：列加密密钥的明文，默认随机生成，也可由用户指定，用户指定时密钥长度范围为28～256位。
列加密密钥创建语法是通过前端解析器将参数解析成CreateClientLogicColumn结构体后，通过校验指定用于加密CEK的CMK对象是否存在后加载CMK缓存，然后通过“HooksManager::ColumnSettings::pre_create”语句调用加密函数“EncryptionColumnHookExecutor::pre_create”来校验各参数并生成或加密ENCRYPTED_VALUE值，最后在“EncryptionPreProcess::set_new_query”函数中替换ENCRYPTED_VALUE参数为CEK密文，重构SQL查询语句。重构后的SQL语句发送给服务端后服务端解析为CreateClientLogicColumn结构体并检查用户namespace等权限，将CEK的信息保存在系统表中。创建CEK的总体流程如图9-40所示，组织结构如图9-41所示。

 图9-40  列加密密钥CEK创建流程

图9-41  客户端主密钥CMK的组织结构 

在对CEK参数进行解析后，使用CMK对ENCRYPTED_VALUE参数进行加密，加密完成后使用加密后的ENCRYPTED_VALUE参数和其他参数对创建CEK的语法进行重构。将输入的查询语句转换成加密查询语句的主要函数入口代码如下：

void EncryptionPreProcess::set_new_query(char **query, size_t query_size, StringArgs string_args, int location,
    int encrypted_value_location, size_t encrypted_value_size, size_t quote_num)
{
for (size_t i = 0; i < string_args.Size(); i++) {
    /*  从string_args中读取键值存到变量中  */
        char string_to_add[MAX_KEY_ADD_LEN];
        errno_t rc = memset_s(string_to_add, MAX_KEY_ADD_LEN, 0, MAX_KEY_ADD_LEN);
        securec_check_c(rc, "\0", "\0");
        size_t total_in = 0;
        if (string_args.at(i) == NULL) {
            continue;
        }
        const char *key = string_args.at(i)->key;
        const char *value = string_args.at(i)->value;
        const size_t vallen = string_args.at(i)->valsize;
        if (!key || !value) {
            Assert(false);
            continue;
        }
        Assert(vallen < MAX_KEY_ADD_LEN);
        /*  将key和value构造成encrypted_value = '密文值'的形式  */
        check_strncat_s(strncat_s(string_to_add, MAX_KEY_ADD_LEN, key, strlen(key)));
        total_in += strlen(key);
        check_strncat_s(strncat_s(string_to_add, MAX_KEY_ADD_LEN, "=\'", strlen("=\'")));
        total_in += strlen("=\'");
        check_strncat_s(strncat_s(string_to_add, MAX_KEY_ADD_LEN, value, vallen));
        total_in += vallen;
        check_strncat_s(strncat_s(string_to_add, MAX_KEY_ADD_LEN, "\'", strlen("\'")));
        total_in += strlen("\'");
        Assert(total_in < MAX_KEY_ADD_LEN);
        /*  encrypted_value_location不为空，则说明用户提供EXPECTED_VALUE，将明文值替换成密文值   */
        if (encrypted_value_location && encrypted_value_size) {
            *query = (char *)libpq_realloc(*query, query_size, query_size + vallen + 1);
            if (*query == NULL) {
                return;
            }
            check_memset_s(memset_s(*query + query_size, vallen + 1, 0, vallen + 1));
            char *replace_dest = *query + encrypted_value_location + strlen("\'");
            char *move_src =
                *query + encrypted_value_location + encrypted_value_size + quote_num + strlen("\'");
            char *move_dest = *query + encrypted_value_location + vallen + strlen("\'");
            check_memmove_s(memmove_s(move_dest,
                query_size - encrypted_value_location - encrypted_value_size - strlen("\'") + 1,
                move_src,
                query_size - encrypted_value_location - encrypted_value_size - strlen("\'")));
            query_size = query_size + vallen - encrypted_value_size;
            check_memcpy_s(memcpy_s(replace_dest, query_size - encrypted_value_location, value, vallen));
        } else { 
/*  EXPECTED_VALUE是随机生成的，则直接插入原先的语句中  */
            check_strcat_s(strcat_s(string_to_add, MAX_KEY_ADD_LEN, ","));
            size_t string_to_add_size = strlen(string_to_add);
            *query = (char *)libpq_realloc(*query, query_size, query_size + string_to_add_size + 1);
            if (*query == NULL) {
                return;
            }
            check_memmove_s(memmove_s(*query + location + string_to_add_size, query_size - location, *query + location,
                query_size - location));
            query_size += string_to_add_size;
            check_memcpy_s(memcpy_s(*query + location, query_size - location, string_to_add, string_to_add_size));
        }
        query[0][query_size] = '\0';
    }
    return;
}

创建加密表

接下来创建加密表。

CREATE TABLE creditcard_info (id_number int, 
name text encrypted with (column_encryption_key = cek_1, encryption_type = DETERMINISTIC), gender varchar(10) encrypted with (column_encryption_key = cek_1, encryption_type = DETERMINISTIC), 
salary float4 encrypted with (column_encryption_key = cek_1, encryption_type = DETERMINISTIC),
credit_card varchar(19) encrypted with (column_encryption_key = cek_1, encryption_type = DETERMINISTIC));

创建加密表的SQL语句在语法解析后进入CreateStmt函数处理逻辑，在run_pre_create_statement函数中，对CreateStmt->tableElts中每个ListCell进行判断，当前加密表仍存在一定的约束，加密表列定义及约束处理函数段代码如下：

bool createStmtProcessor::run_pre_create_statement(const CreateStmt * const stmt, StatementData *statement_data)
{
    …
  /*  加密表列定义及约束处理  */
    foreach (elements, stmt->tableElts) {
        Node *element = (Node *)lfirst(elements);
        switch (nodeTag(element)) {
            case T_ColumnDef: {
                …
               /*  校验distribute by是否符合规格  */
                if (column->colname != NULL &&
                    !check_distributeby(stmt->distributeby, column->colname)) {
                    return false;
                }
               /*  列定义处理，存储加密类型，加密密钥等信息  */
                if (!process_column_defintion(column, element, &expr_vec, &cached_columns, 
                    &cached_columns_for_defaults, statement_data)) {
                    return false;
                }
                break;
            }
            /*  处理check, unique 或其他约束  */
            case T_Constraint: {
                Constraint *constraint = (Constraint*)element;
                if (constraint->keys != NULL) {
                    ListCell *ixcell = NULL;
                    foreach (ixcell, constraint->keys) {
                        char *ikname = strVal(lfirst(ixcell));
                        for (size_t i = 0; i < cached_columns.size(); i++) {
                            if (strcmp((cached_columns.at(i))->get_col_name(), ikname) == 0 && !check_constraint(
                                constraint, cached_columns.at(i)->get_data_type(), ikname, &cached_columns)) {
                                return false;
                            }
                        }
                    }
                } else if (constraint->raw_expr != NULL) {
                    if (!transform_expr(constraint->raw_expr, "", &cached_columns)) {
                        return false;
                    }
                }
                break;
            }
            default:
                break;
        }
    }
     …
    /*  加密约束中需要加密的明文数据  */
    if (!RawValues::get_raw_values_from_consts_vec(&expr_vec, statement_data, 0, &raw_values_list)) {
        return false;
    }
    return ValuesProcessor::process_values(statement_data, &cached_columns_for_defaults, 1,
        &raw_values_list);
}

在将创建加密表的查询语句发送给服务端后，服务端创建成功并返回执行成功的消息。数据加密驱动程序能够实现在数据发送到数据库之前透明地加密数据，数据在整个语句的处理过程中以密文形式存在，在返回结果时，解密返回的数据集，从而保证整个过程对用户是透明、无感知的。
定义了完整的加密表后，用户就可以用正常的方式将数据插入到表中。完整的加密过程见encrypt_data函数，其核心逻辑代码如下所示：

int encrypt_data(const unsigned char *plain_text, int plain_text_length, const AeadAesHamcEncKey &column_encryption_key,
    EncryptionType encryption_type, unsigned char *result, ColumnEncryptionAlgorithm column_encryption_algorithm)
{
    ……
    /*  得到16位的iv值  */
    unsigned char _iv [g_key_size + 1] = {0};
unsigned char iv_truncated[g_iv_size + 1] = {0};
/*  确定性加密，则通过hmac_sha256生成iv  */
    if (encryption_type == EncryptionType::DETERMINISTIC_TYPE) {
           hmac_sha256(column_encryption_key.get_iv_key(), g_auth_tag_size, plain_text, plain_text_length, _iv);
       ……
} else {
/*  随机加密，则随机生成iv  */
        if (encryption_type != EncryptionType::RANDOMIZED_TYPE) {
            return 0; 
        }
        int res = RAND_priv_bytes(iv_truncated, g_block_size);
        if (res != 1) {
            return 0;
        }
    }
    int cipherStart = g_algo_version_size + g_auth_tag_size + g_iv_size;
    /*  调用encrypt计算密文  */
int cipherTextSize = encrypt(plain_text, plain_text_length, column_encryption_key.get_encyption_key(), iv_truncated,
        result + cipherStart, column_encryption_algorithm);
   ……
    int ivStartIndex = g_auth_tag_size + g_algo_version_size;
    res = memcpy_s(result + ivStartIndex, g_iv_size, iv_truncated, g_iv_size);
    securec_check_c(res, "\0", "\0");
    /*  计算 HMAC  */
    int hmacDataSize = g_algo_version_size + g_iv_size + cipherTextSize;
    hmac_sha256(column_encryption_key.get_mac_key(), g_auth_tag_size,
                result + g_auth_tag_size, hmacDataSize, result); 
    return (g_auth_tag_size + hmacDataSize);
}

openGauss密态数据库在进行等值查询的时候，整个查询过程对用户是无感知的，虽然存储在数据库中的数据是密文形式，但在展示数据给用户的时候会将密文数据进行解密处理。以从加密表中进行等值查询语句为例，整个语句处理过程如图9-42所示。客户端解析SELECT查询语句中的列属性信息，如果缓存已有则从缓存中提取列属性信息；如果缓存中找不到，需要从服务端查询该信息，并缓存。列加密密钥CEK是以密文形式存储在服务端，客户端需要解密CEK。然后用其加密SELECT查询语句中条件参数。加密后的SELECT查询语句发送给数据库服务端执行完成后，返回加密的查询结果集。客户端用解密后的列加密密钥CEK解密SELECT查询结果集，并返回解密后的明文结果集给应用端。

 图9-42  SELECT语句时序图

等值查询处理run_pre_insert_statement函数，其核心逻辑代码如下所示：

bool Processor::run_pre_select_statement(const SelectStmt * const select_stmt, const SetOperation &parent_set_operation,
    const bool &parent_all, StatementData *statement_data, ICachedColumns *cached_columns, ICachedColumns *cached_columns_parents)
{
bool select_res = false;
/*  处理SELECT语句中的集合操作  */
    if (select_stmt->op != SETOP_NONE) {
        select_res = process_select_set_operation(select_stmt, statement_data, cached_columns);
        RETURN_IF(!select_res);
    }
    /*  处理WHERE子句  */
    ExprPartsList where_expr_parts_list;
    select_res = exprProcessor::expand_expr(select_stmt->whereClause, statement_data, &where_expr_parts_list);
    RETURN_IF(!select_res);
……
    /*  从FROM子句中获取缓存加密列  */
CachedColumns cached_columns_from(false, true);
    select_res = run_pre_from_list_statement(select_stmt->fromClause, statement_data, &cached_columns_from,
        cached_columns_parents);
……
/*  将查询的加密列放在cached_columns结构中  */
    for (size_t i = 0; i < cached_columns_from.size(); i++) {
        if (find_in_name_map(target_list, cached_columns_from.at(i)->get_col_name())) {
            CachedColumn *target = new (std::nothrow) CachedColumn(cached_columns_from.at(i));
            if (target == NULL) {
                fprintf(stderr, "failed to new CachedColumn object\n");
                return false;
            }
            cached_columns->push(target);
        }
    }
    if (cached_columns_from.is_empty()) {
        return true; 
    }
    /*  加密列不支持ORDER BY（排序）操作  */
    if (!deal_order_by_statement(select_stmt, cached_columns)) {
        return false;
    }
   
/*  将WHERE子句中加密的值进行加密处理  */
    if (!WhereClauseProcessor::process(&cached_columns_from, &where_expr_parts_list, statement_data)) {
        return false;
}
……
    return true;
}

完整的客户端密文解密函数代码如下所示： 

int decrypt_data(const unsigned char *cipher_text, int cipher_text_length,
    const AeadAesHamcEncKey &column_encryption_key, unsigned char *decryptedtext,
    ColumnEncryptionAlgorithm column_encryption_algorithm)
{
    if (cipher_text == NULL || cipher_text_length <= 0 || decryptedtext == NULL) {
        return 0;
    }
    /*  校验密文长度  */
    if (cipher_text_length < min_ciph_len_in_bytes_with_authen_tag) {
        printf("ERROR(CLIENT): The length of cipher_text is invalid, cannot decrypt.\n");
        return 0;
    }
    /*  校验密文中的版本号  */
    if (cipher_text[g_auth_tag_size] != '1') {
        printf("ERROR(CLIENT): Version byte of cipher_text is invalid, cannot decrypt.\n");
        return 0;
    }
    ……
    /*  计算MAC标签  */
    unsigned char authenticationTag [g_auth_tag_size] = {0};
    int HMAC_length = cipher_text_length - g_auth_tag_size;
    int res = hmac_sha256(column_encryption_key.get_mac_key(), g_auth_tag_size,
        cipher_text + g_auth_tag_size, HMAC_length, authenticationTag);
    if (res != 1) {
        printf("ERROR(CLIENT): Fail to compute a keyed hash of a given text.\n");
        return 0;
    }
    /*  校验密文是否被篡改  */
    int cmp_result = my_memcmp(authenticationTag, cipher_text, g_auth_tag_size);
    /*  解密数据  */
    int decryptedtext_len = decrypt(cipher_text + cipher_start_index, cipher_value_length,
        column_encryption_key.get_encyption_key(), iv, decryptedtext, column_encryption_algorithm);
    if (decryptedtext_len < 0) {
        return 0;
    }
    decryptedtext[decryptedtext_len] = '\0';
    return decryptedtext_len;
}