如何在Snowflake中实现等效于SQL Server自定义加密函数的密码加密？

原SQL Server的C#函数采用DES-CBC模式加密，固定IV和密钥，将输入字符串转UTF-8后加密，最终输出Base64编码结果。以下是三种Snowflake兼容的实现方案：

方案1：Snowflake内置SQL函数组合（推荐，无需外部依赖） #

利用Snowflake原生函数实现，逻辑与原函数完全对齐：

CREATE OR REPLACE FUNCTION ENCRYPT_PASSWORD(string_to_encrypt VARCHAR)
RETURNS VARCHAR
AS
$$
    IF string_to_encrypt IS NOT NULL AND string_to_encrypt != '' THEN
        -- 原C#密钥与IV转十六进制二进制
        LET key_binary BINARY = TO_BINARY('2A105D9C4E04DA20', 'HEX');
        LET iv_binary BINARY = TO_BINARY('3767F64F2463A703', 'HEX');
        -- 输入字符串转UTF-8二进制
        LET input_binary BINARY = TO_BINARY(string_to_encrypt, 'UTF8');
        -- DES-CBC-PKCS5加密（PKCS5与PKCS7在8字节块大小下等价）
        LET encrypted_binary BINARY = ENCRYPT(input_binary, key_binary, iv_binary, 'DES_CBC_PKCS5');
        -- 加密结果转Base64
        RETURN TO_BASE64(encrypted_binary);
    ELSE
        RETURN '';
    END IF;
$$
LANGUAGE SQL;

方案2：Java UDF（完全复刻原C#逻辑） #

如果需要严格对齐原C#函数的加密细节，可使用Java UDF实现：

CREATE OR REPLACE FUNCTION ENCRYPT_PASSWORD(string_to_encrypt VARCHAR)
RETURNS VARCHAR
LANGUAGE JAVA
HANDLER = 'EncryptHandler.encrypt'
AS
$$
import javax.crypto.Cipher;
import javax.crypto.spec.IvParameterSpec;
import javax.crypto.spec.SecretKeySpec;
import java.util.Base64;

public class EncryptHandler {
    public static String encrypt(String stringToEncrypt) {
        if (stringToEncrypt == null || stringToEncrypt.isEmpty()) {
            return "";
        }
        try {
            // 复用原C#的密钥与IV
            byte[] iv = {55, 103, 246, 79, 36, 99, 167, 3};
            byte[] key = {42, 16, 93, 156, 78, 4, 218, 32};
            
            Cipher cipher = Cipher.getInstance("DES/CBC/PKCS5Padding");
            SecretKeySpec keySpec = new SecretKeySpec(key, "DES");
            IvParameterSpec ivSpec = new IvParameterSpec(iv);
            
            cipher.init(Cipher.ENCRYPT_MODE, keySpec, ivSpec);
            byte[] inputBytes = stringToEncrypt.getBytes("UTF-8");
            byte[] encryptedBytes = cipher.doFinal(inputBytes);
            
            return Base64.getEncoder().encodeToString(encryptedBytes);
        } catch (Exception e) {
            throw new RuntimeException("Encryption failed", e);
        }
    }
$$;

方案3：Python UDF（适合Python技术栈） #

使用cryptography库实现，手动处理PKCS7填充：

CREATE OR REPLACE FUNCTION ENCRYPT_PASSWORD(string_to_encrypt VARCHAR)
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = 3.8
HANDLER = 'encrypt'
PACKAGES = ('cryptography')
AS
$$
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import base64

def encrypt(string_to_encrypt):
    if not string_to_encrypt:
        return ""
    # 原C#密钥与IV
    key = bytes([42, 16, 93, 156, 78, 4, 218, 32])
    iv = bytes([55, 103, 246, 79, 36, 99, 167, 3])
    
    cipher = Cipher(algorithms.DES(key), modes.CBC(iv), backend=default_backend())
    encryptor = cipher.encryptor()
    
    # PKCS7填充处理
    input_bytes = string_to_encrypt.encode('utf-8')
    padding_length = 8 - (len(input_bytes) % 8)
    padded_input = input_bytes + bytes([padding_length]) * padding_length
    
    encrypted_bytes = encryptor.update(padded_input) + encryptor.finalize()
    return base64.b64encode(encrypted_bytes).decode('utf-8')
$$;

关键注意事项 #

• DES算法安全性较弱，建议后续逐步迁移至AES等现代加密算法
• 硬编码密钥/IV存在安全风险，建议使用Snowflake Secret Manager存储敏感信息，通过GET_SECRET函数动态获取
• 测试时需验证加密结果与原C#函数完全一致，避免编码或填充模式差异导致的不一致

内容的提问来源于stack exchange，提问作者Ajeet Verma