This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12403-b0dd4945b1da8423a481b053082bd5fd11b9533f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit d051216fc2e253056c8871e7d89d4cfcda5ebb25 Author: chovygo <[email protected]> AuthorDate: Sun Oct 4 23:00:11 2026 +0000 [Feature][Transform-V2] Add AES_ENCRYPT/AES_DECRYPT SQL functions (#12403) Co-authored-by: zhaowei <[email protected]> --- .../introduction/concepts/incompatible-changes.md | 18 + docs/en/transforms/sql-functions.md | 44 ++ .../introduction/concepts/incompatible-changes.md | 8 + docs/zh/transforms/sql-functions.md | 44 ++ .../test/resources/sql_transform/func_string.conf | 16 +- .../transform/sql/zeta/ZetaSQLFunction.java | 9 + .../seatunnel/transform/sql/zeta/ZetaSQLType.java | 2 + .../sql/zeta/functions/CryptoFunction.java | 308 +++++++++++++ .../transform/sql/SQLCryptoFunctionsTest.java | 150 +++++++ .../sql/zeta/functions/CryptoFunctionTest.java | 479 +++++++++++++++++++++ 10 files changed, 1077 insertions(+), 1 deletion(-) diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 79d75c8cee..340ebb8492 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -53,6 +53,24 @@ You need to check this document before you upgrade to related version. - To keep using the default user, remove `user` and retain `auth` when a password is required. Named users require Redis 6 or later. Legacy configurations without a username remain unchanged. +### Zeta SQL Transform: built-in AES_ENCRYPT / AES_DECRYPT + +- **Behavior change: AES_ENCRYPT / AES_DECRYPT are now built-in functions** + - **Affected component**: `seatunnel-transforms-v2` (Zeta SQL transform). + - **Description**: `AES_ENCRYPT(value, key[, iv])` and `AES_DECRYPT(value, key[, iv])` are now + built-in Zeta SQL functions and are dispatched before user-registered `ZetaUDF`s. They use + `AES/CBC/PKCS5Padding` with Base64 output; without an explicit IV a random IV is generated and + prepended to the ciphertext so `AES_DECRYPT` can recover it without an explicit IV. + - **Impact**: A job that registered a custom `ZetaUDF` named `AES_ENCRYPT` or `AES_DECRYPT` (the + previous workaround for the missing built-in) will, after upgrading, silently start using this + built-in implementation instead of the UDF. If the UDF used a different key derivation, IV + handling or output encoding, ciphertext already written by the UDF may fail to decrypt (or, + roughly once in 256 for CBC padding, decrypt to garbage). + - **Migration Guide**: Rename the existing UDF, or switch to the built-in functions. To stay + wire-compatible with the `FieldEncrypt` `AesCbcEncryptor`, supply the key with the `base64:` + prefix (a bare key is derived as a passphrase via SHA-256 and is **not** interchangeable with + `FieldEncrypt`). See [SQL Functions](../../transforms/sql-functions.md) for the full contract. + ### RabbitMQ Connector - **Breaking Change: `amqps://` connections now verify broker certificates** diff --git a/docs/en/transforms/sql-functions.md b/docs/en/transforms/sql-functions.md index 0b629187bc..a244f3bad0 100644 --- a/docs/en/transforms/sql-functions.md +++ b/docs/en/transforms/sql-functions.md @@ -1110,6 +1110,50 @@ local_date_time AT TIME ZONE '+09:00' offset_date_time AT TIME ZONE 'Pacific/Honolulu' +## Crypto Functions + +### AES_ENCRYPT + +```AES_ENCRYPT(value, key[, iv]) -> STRING``` + +Encrypts `value` with AES/CBC/PKCS5Padding and returns a Base64-encoded ciphertext. Returns **NULL** if `value` is **NULL**. + +- `value`: the plaintext to encrypt. Any non-null scalar value is converted to a string; array and map inputs are rejected. +- `key`: the secret key. If it starts with `base64:`, the remainder is decoded as a raw AES key and must be 16, 24, or 32 bytes (AES-128/192/256); only this `base64:` form is wire-compatible with the `AesCbcEncryptor` used by the `FieldEncryptTransform`. Any other value is treated as a passphrase: it is hashed once with SHA-256 and the first 16 bytes are used as an AES-128 key, so arbitrary-length passphrases are supported. Passphrase mode is an unsalted fast KDF; for strong protection u [...] +- `iv`: optional initialization vector. If provided, its UTF-8 bytes are used as the IV and must be exactly 16 bytes; the returned ciphertext contains only the encrypted bytes and the caller is responsible for storing the IV. If omitted, a random 16-byte IV is generated per call and prepended to the ciphertext, so `AES_DECRYPT` can recover it without an explicit IV. An explicit `NULL` IV is rejected (omit the argument instead). + +Example: + +AES_ENCRYPT(name, 'mySecretPass') + +AES_ENCRYPT(name, 'mySecretPass', '1234567890123456') + +AES_ENCRYPT(name, 'base64:MTIzNDU2Nzg5MDEyMzQ1Ng==') + +SELECT AES_DECRYPT(AES_ENCRYPT(name, 'mySecretPass'), 'mySecretPass') AS name_enc FROM dual + +NOTE: +- CBC is an unauthenticated mode: a wrong key or a corrupted ciphertext may (about once in 256) decrypt to garbage instead of throwing. Do not rely on a decryption error to detect a wrong key. +- When `iv` is omitted the ciphertext is non-deterministic (a fresh random IV is generated for each call). To get a deterministic ciphertext, provide an explicit `iv`. + +### AES_DECRYPT + +```AES_DECRYPT(value, key[, iv]) -> STRING``` + +Decrypts a Base64 AES/CBC/PKCS5Padding ciphertext. Returns **NULL** if `value` is **NULL**. + +- `value`: the Base64-encoded ciphertext produced by `AES_ENCRYPT`. +- `key`: the secret key, with the same conventions as `AES_ENCRYPT`. It must match the key used for encryption. +- `iv`: optional initialization vector. When omitted, the first 16 bytes of the decoded payload are treated as the IV (the format produced by `AES_ENCRYPT` without an IV). When provided, its UTF-8 bytes are used as the IV (must be 16 bytes) and the whole decoded payload is treated as the ciphertext. An explicit `NULL` IV is rejected. + +Example: + +AES_DECRYPT(cipher, 'mySecretPass') + +AES_DECRYPT(AES_ENCRYPT(name, 'mySecretPass'), 'mySecretPass') + +AES_DECRYPT(cipher, 'mySecretPass', '1234567890123456') + ## System Functions ### CAST diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index bd67abb3c0..7facd5f3c2 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -25,6 +25,14 @@ - 如需继续使用默认用户，请移除 `user`，并在需要密码时保留 `auth`。 命名用户需要 Redis 6 或更新版本；未配置用户名的旧配置行为保持不变。 +### Zeta SQL Transform：内置 AES_ENCRYPT / AES_DECRYPT + +- **行为变更：AES_ENCRYPT / AES_DECRYPT 现为内置函数** + - **影响范围**：`seatunnel-transforms-v2`（Zeta SQL transform）。 + - **变更说明**：`AES_ENCRYPT(value, key[, iv])` 与 `AES_DECRYPT(value, key[, iv])` 现为内置 Zeta SQL 函数，且分发顺序在用户注册的 `ZetaUDF` 之前。使用 `AES/CBC/PKCS5Padding`，输出 Base64；未显式提供 IV 时生成随机 IV 并拼接到密文头部，故 `AES_DECRYPT` 无需显式 IV 即可恢复。 + - **影响**：若作业此前注册了名为 `AES_ENCRYPT` 或 `AES_DECRYPT` 的自定义 `ZetaUDF`（此前缺少内置函数时的变通做法），升级后将静默改用此内置实现而非 UDF。若该 UDF 使用了不同的密钥派生、IV 处理或输出编码，则已由 UDF 写入的密文可能无法解密（或在 CBC 填充校验以约 1/256 概率碰巧通过时解出垃圾）。 + - **迁移指南**：重命名已有 UDF，或迁移到内置函数。如需与 `FieldEncrypt` 的 `AesCbcEncryptor` 保持线兼容，请使用带 `base64:` 前缀的密钥（裸密钥会按口令经 SHA-256 派生，**不**与 `FieldEncrypt` 互通）。完整契约见 [SQL 函数](../../transforms/sql-functions.md)。 + ### RabbitMQ Connector - **破坏性变更：`amqps://` 连接现在会校验 Broker 证书** diff --git a/docs/zh/transforms/sql-functions.md b/docs/zh/transforms/sql-functions.md index 81b11c2b1d..b7eac38fd1 100644 --- a/docs/zh/transforms/sql-functions.md +++ b/docs/zh/transforms/sql-functions.md @@ -1115,6 +1115,50 @@ local_date_time AT TIME ZONE '+09:00' offset_date_time AT TIME ZONE 'Pacific/Honolulu' +## 加密函数 + +### AES_ENCRYPT + +```AES_ENCRYPT(value, key[, iv]) -> STRING``` + +使用 AES/CBC/PKCS5Padding 加密 `value`，返回 Base64 编码的密文。如果 `value` 为 **NULL**，返回 **NULL**。 + +- `value`：待加密的明文。任何非空标量值都会被转换为字符串；数组与 Map 输入会被拒绝。 +- `key`：密钥。如果以 `base64:` 开头，则剩余部分按 Base64 解码为原始 AES 密钥，长度必须为 16、24 或 32 字节（对应 AES-128/192/256）；只有 `base64:` 形式才与 `FieldEncryptTransform` 使用的 `AesCbcEncryptor` 线兼容。其他值按口令处理：对其 UTF-8 字节做一次 SHA-256，取前 16 字节作为 AES-128 密钥，因此支持任意长度的口令。口令模式是无盐的快速 KDF，安全性较低；生产环境请使用随机的 `base64:` 密钥并作为机密保管。`base64:` 密钥与口令不可互换。 +- `iv`：可选的初始化向量。提供时，其 UTF-8 字节作为 IV 使用，必须恰好为 16 字节；返回的密文仅包含加密后的字节，调用方需自行保存 IV。省略时，每次调用生成 16 字节随机 IV 并拼接到密文头部，`AES_DECRYPT` 无需显式 IV 即可恢复。显式传入 `NULL` 的 IV 会被拒绝（请改为省略该参数）。 + +示例: + +AES_ENCRYPT(name, 'mySecretPass') + +AES_ENCRYPT(name, 'mySecretPass', '1234567890123456') + +AES_ENCRYPT(name, 'base64:MTIzNDU2Nzg5MDEyMzQ1Ng==') + +SELECT AES_DECRYPT(AES_ENCRYPT(name, 'mySecretPass'), 'mySecretPass') AS name_enc FROM dual + +注意: +- CBC 是无认证模式：密钥错误或密文损坏时，约 1/256 的概率会解出垃圾串而非报错，切勿依赖解密报错来判断密钥是否正确。 +- 省略 `iv` 时密文是非确定性的（每次调用都会生成新的随机 IV）。如需确定性密文，请显式提供 `iv`。 + +### AES_DECRYPT + +```AES_DECRYPT(value, key[, iv]) -> STRING``` + +解密 Base64 编码的 AES/CBC/PKCS5Padding 密文。如果 `value` 为 **NULL**，返回 **NULL**。 + +- `value`：由 `AES_ENCRYPT` 生成的 Base64 密文。 +- `key`：密钥，约定与 `AES_ENCRYPT` 相同，必须与加密时使用的密钥一致。 +- `iv`：可选的初始化向量。省略时，取解码后前 16 字节作为 IV（即 `AES_ENCRYPT` 未提供 IV 时生成的格式）。提供时，其 UTF-8 字节作为 IV（必须为 16 字节），整个解码负载视为密文。显式传入 `NULL` 的 IV 会被拒绝。 + +示例: + +AES_DECRYPT(cipher, 'mySecretPass') + +AES_DECRYPT(AES_ENCRYPT(name, 'mySecretPass'), 'mySecretPass') + +AES_DECRYPT(cipher, 'mySecretPass', '1234567890123456') + ## System Functions ### CAST diff --git a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf index 192b6229e4..883a79a604 100644 --- a/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf +++ b/seatunnel-e2e/seatunnel-transforms-v2-e2e/seatunnel-transforms-v2-e2e-part-2/src/test/resources/sql_transform/func_string.conf @@ -54,7 +54,7 @@ transform { sql { plugin_input = "fake" plugin_output = "fake1" - query = "select ascii(c1) as c1_1, ascii(c2) as c2_1, bit_length(c4) as c4_1, length(c4) as c4_2, octet_length(c4) as c4_3, char(c5) as c5_1, concat(c1,id,'!') as c1_2, hextoraw(c6) as c6_1, rawtohex(c7) as c7_1, insert(name,2,2,'**') as name1, lower(name) as name2, upper(name) as name3, left(name, 3) as name4, right(name, 4) as name5, lpad(name, 10, '*') as name6, rpad(name, 10, '*') as name7, ltrim(c8, '*') as c8_1, rtrim(c8, '*') as c8_2, trim(c8, '*') as c8_3, regexp_replace(c9, [...] + query = "select ascii(c1) as c1_1, ascii(c2) as c2_1, bit_length(c4) as c4_1, length(c4) as c4_2, octet_length(c4) as c4_3, char(c5) as c5_1, concat(c1,id,'!') as c1_2, hextoraw(c6) as c6_1, rawtohex(c7) as c7_1, insert(name,2,2,'**') as name1, lower(name) as name2, upper(name) as name3, left(name, 3) as name4, right(name, 4) as name5, lpad(name, 10, '*') as name6, rpad(name, 10, '*') as name7, ltrim(c8, '*') as c8_1, rtrim(c8, '*') as c8_2, trim(c8, '*') as c8_3, regexp_replace(c9, [...] } } @@ -323,6 +323,20 @@ sink { field_value = [ {equals_to = "Joy Ding"} ] + }, + { + field_name = "name20" + field_type = "string" + field_value = [ + {equals_to = "Joy Ding"} + ] + }, + { + field_name = "name21" + field_type = "string" + field_value = [ + {equals_to = "Joy Ding"} + ] } ] } diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java index 10e927516a..76765466a3 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLFunction.java @@ -32,6 +32,7 @@ import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated; import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; import org.apache.seatunnel.transform.exception.TransformException; import org.apache.seatunnel.transform.sql.zeta.functions.ArrayFunction; +import org.apache.seatunnel.transform.sql.zeta.functions.CryptoFunction; import org.apache.seatunnel.transform.sql.zeta.functions.DateTimeFunction; import org.apache.seatunnel.transform.sql.zeta.functions.MapFunction; import org.apache.seatunnel.transform.sql.zeta.functions.NumericFunction; @@ -129,6 +130,10 @@ public class ZetaSQLFunction { public static final String SPLIT = "SPLIT"; public static final String MURMUR64 = "MURMUR64"; + // -------------------------crypto functions---------------------------- + public static final String AES_ENCRYPT = "AES_ENCRYPT"; + public static final String AES_DECRYPT = "AES_DECRYPT"; + // -------------------------numeric functions---------------------------- public static final String ABS = "ABS"; public static final String ACOS = "ACOS"; @@ -532,6 +537,10 @@ public class ZetaSQLFunction { return StringFunction.split(args); case MURMUR64: return StringFunction.murmur64(args); + case AES_ENCRYPT: + return CryptoFunction.aesEncrypt(args); + case AES_DECRYPT: + return CryptoFunction.aesDecrypt(args); case ABS: return NumericFunction.abs(args); case ACOS: diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java index 5fbf4c44d6..8082d3b303 100644 --- a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/ZetaSQLType.java @@ -357,6 +357,8 @@ public class ZetaSQLType { case ZetaSQLFunction.FROM_UNIXTIME: case ZetaSQLFunction.UUID: case ZetaSQLFunction.TRIM_SCALE: + case ZetaSQLFunction.AES_ENCRYPT: + case ZetaSQLFunction.AES_DECRYPT: return BasicType.STRING_TYPE; case ZetaSQLFunction.ASCII: case ZetaSQLFunction.LOCATE: diff --git a/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/CryptoFunction.java b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/CryptoFunction.java new file mode 100644 index 0000000000..10cc926c26 --- /dev/null +++ b/seatunnel-transforms-v2/src/main/java/org/apache/seatunnel/transform/sql/zeta/functions/CryptoFunction.java @@ -0,0 +1,308 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.transform.sql.zeta.functions; + +import org.apache.seatunnel.common.exception.CommonError; +import org.apache.seatunnel.transform.exception.TransformCommonError; +import org.apache.seatunnel.transform.sql.zeta.ZetaSQLFunction; + +import javax.crypto.Cipher; +import javax.crypto.spec.IvParameterSpec; +import javax.crypto.spec.SecretKeySpec; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.security.SecureRandom; +import java.util.Base64; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Built-in AES crypto functions for the Zeta SQL transform. + * + * <p>Both functions use {@code AES/CBC/PKCS5Padding} and return / accept Base64-encoded strings. + * The wire format when no explicit IV is supplied is {@code Base64(IV || ciphertext)} (a random + * 16-byte IV is generated per call and prepended), so {@link #aesDecrypt(List)} can recover it + * without an explicit IV. When an explicit 16-byte IV is supplied, the ciphertext carries only the + * encrypted bytes and the caller is responsible for the IV. + * + * <p>Key conventions: + * + * <ul> + * <li>a key prefixed with {@code base64:} is decoded as a raw AES key (16/24/32 bytes for + * AES-128/192/256); only this form is wire-compatible with the {@code AesCbcEncryptor} of the + * {@code FieldEncryptTransform}; + * <li>any other key is treated as a passphrase and derived via a single unsalted SHA-256 whose + * first 16 bytes are used as an AES-128 key. This is fast to brute-force for low-entropy + * passphrases; use a random {@code base64:} key for strong protection. + * </ul> + * + * <p>CBC is unauthenticated: a wrong key or corrupted ciphertext may (about once in 256) decrypt to + * garbage instead of throwing. + * + * <p>Error paths never embed the key, plaintext, ciphertext or IV into the exception message; only + * non-sensitive metadata (argument count, byte length, type name) is included, following the {@link + * CommonError#illegalArgument(String, String)} convention used across the codebase. + */ +public class CryptoFunction { + + private static final String TRANSFORMATION = "AES/CBC/PKCS5Padding"; + private static final String KEY_ALGORITHM = "AES"; + private static final int IV_SIZE = 16; + private static final String BASE64_PREFIX = "base64:"; + + // Bounded cache for derived keys, keyed by the key string. A constant key (the common case) + // is derived once instead of on every row. Cache hits are lock-free (ConcurrentHashMap.get); + // only a miss derives and inserts, and eviction removes a single entry when the cap is exceeded + // rather than clearing the whole map, so residency stays at most roughly KEY_CACHE_MAX_SIZE. + private static final int KEY_CACHE_MAX_SIZE = 64; + private static final ConcurrentHashMap<String, SecretKeySpec> KEY_CACHE = + new ConcurrentHashMap<>(KEY_CACHE_MAX_SIZE); + + private static final SecureRandom SECURE_RANDOM = new SecureRandom(); + + private CryptoFunction() {} + + /** + * Encrypts a value with AES/CBC/PKCS5Padding and returns a Base64 string. + * + * @param args [value, key[, iv]] + * @return Base64-encoded ciphertext, or null if the value is null + */ + public static String aesEncrypt(List<Object> args) { + String operation = ZetaSQLFunction.AES_ENCRYPT; + if (args.size() < 2 || args.size() > 3) { + throw CommonError.illegalArgument( + String.valueOf(args.size()), operation + " expects 2 or 3 arguments"); + } + Object value = args.get(0); + if (value == null) { + return null; + } + rejectNonScalar(value, "value", operation); + String plainText = value.toString(); + Object keyArg = args.get(1); + if (keyArg == null) { + throw CommonError.illegalArgument("key", operation + ": key must not be null"); + } + rejectNonScalar(keyArg, "key", operation); + SecretKeySpec keySpec = cachedKey(keyArg.toString(), operation); + + byte[] iv = resolveIv(args, operation, true); + boolean ivProvided = args.size() == 3; + + try { + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + cipher.init(Cipher.ENCRYPT_MODE, keySpec, new IvParameterSpec(iv)); + byte[] encrypted = cipher.doFinal(plainText.getBytes(StandardCharsets.UTF_8)); + if (ivProvided) { + return Base64.getEncoder().encodeToString(encrypted); + } + byte[] encryptedWithIv = new byte[IV_SIZE + encrypted.length]; + System.arraycopy(iv, 0, encryptedWithIv, 0, IV_SIZE); + System.arraycopy(encrypted, 0, encryptedWithIv, IV_SIZE, encrypted.length); + return Base64.getEncoder().encodeToString(encryptedWithIv); + } catch (Exception e) { + throw TransformCommonError.encryptionError(operation + ": encryption failed", e); + } + } + + /** + * Decrypts a Base64 AES/CBC/PKCS5Padding ciphertext. + * + * @param args [ciphertext, key[, iv]] + * @return decrypted UTF-8 plaintext, or null if the value is null + */ + public static String aesDecrypt(List<Object> args) { + String operation = ZetaSQLFunction.AES_DECRYPT; + if (args.size() < 2 || args.size() > 3) { + throw CommonError.illegalArgument( + String.valueOf(args.size()), operation + " expects 2 or 3 arguments"); + } + Object value = args.get(0); + if (value == null) { + return null; + } + rejectNonScalar(value, "value", operation); + String cipherText = value.toString(); + Object keyArg = args.get(1); + if (keyArg == null) { + throw CommonError.illegalArgument("key", operation + ": key must not be null"); + } + rejectNonScalar(keyArg, "key", operation); + SecretKeySpec keySpec = cachedKey(keyArg.toString(), operation); + + byte[] decoded; + try { + decoded = Base64.getDecoder().decode(cipherText); + } catch (IllegalArgumentException e) { + throw CommonError.illegalArgument("value", operation + ": value is not valid Base64"); + } + + boolean ivProvided = args.size() == 3; + byte[] iv; + byte[] encrypted; + if (ivProvided) { + iv = resolveIv(args, operation, false); + encrypted = decoded; + if (encrypted.length == 0) { + throw CommonError.illegalArgument( + String.valueOf(decoded.length), operation + ": ciphertext is empty"); + } + } else { + if (decoded.length < IV_SIZE) { + throw CommonError.illegalArgument( + String.valueOf(decoded.length), + operation + ": ciphertext too short to carry a 16-byte IV"); + } + iv = new byte[IV_SIZE]; + encrypted = new byte[decoded.length - IV_SIZE]; + System.arraycopy(decoded, 0, iv, 0, IV_SIZE); + System.arraycopy(decoded, IV_SIZE, encrypted, 0, encrypted.length); + } + + try { + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + cipher.init(Cipher.DECRYPT_MODE, keySpec, new IvParameterSpec(iv)); + byte[] original = cipher.doFinal(encrypted); + return new String(original, StandardCharsets.UTF_8); + } catch (Exception e) { + throw TransformCommonError.encryptionError( + operation + + ": decryption failed (wrong key, wrong IV, or corrupted ciphertext)", + e); + } + } + + /** + * Resolves the IV. When the IV argument is present it must be non-null and 16 bytes; an + * explicit null IV is rejected so a column of mixed null/non-null IVs cannot silently produce + * two different ciphertext layouts. When no IV is provided, a random 16-byte IV is generated. + */ + private static byte[] resolveIv(List<Object> args, String operation, boolean generateIfAbsent) { + if (args.size() != 3) { + if (!generateIfAbsent) { + throw new IllegalStateException("resolveIv called without an IV argument"); + } + byte[] iv = new byte[IV_SIZE]; + SECURE_RANDOM.nextBytes(iv); + return iv; + } + Object ivArg = args.get(2); + if (ivArg == null) { + throw CommonError.illegalArgument( + "iv", + operation + ": iv must not be null (omit the argument to use a random IV)"); + } + rejectNonScalar(ivArg, "iv", operation); + return buildIv(ivArg.toString(), operation); + } + + /** + * Builds an AES key from either a raw Base64 key or a passphrase. See the class Javadoc for the + * key conventions. + */ + private static SecretKeySpec buildKey(String key, String operation) { + if (key == null || key.trim().isEmpty()) { + throw CommonError.illegalArgument("key", operation + ": key must not be null or blank"); + } + byte[] keyBytes; + if (key.startsWith(BASE64_PREFIX)) { + String base64 = key.substring(BASE64_PREFIX.length()).trim(); + try { + keyBytes = Base64.getDecoder().decode(base64); + } catch (IllegalArgumentException e) { + throw CommonError.illegalArgument( + "key", operation + ": base64: key is not valid Base64"); + } + if (!(keyBytes.length == 16 || keyBytes.length == 24 || keyBytes.length == 32)) { + throw CommonError.illegalArgument( + String.valueOf(keyBytes.length), + operation + ": base64: key must be 16, 24 or 32 bytes"); + } + } else { + keyBytes = derivePassphraseKey(key, operation); + } + return new SecretKeySpec(keyBytes, KEY_ALGORITHM); + } + + private static byte[] derivePassphraseKey(String passphrase, String operation) { + try { + MessageDigest sha256 = MessageDigest.getInstance("SHA-256"); + byte[] digest = sha256.digest(passphrase.getBytes(StandardCharsets.UTF_8)); + byte[] keyBytes = new byte[16]; + System.arraycopy(digest, 0, keyBytes, 0, 16); + return keyBytes; + } catch (NoSuchAlgorithmException e) { + // SHA-256 is mandated by every JDK; if it is missing the runtime is unusable. + throw TransformCommonError.encryptionError(operation, e); + } + } + + private static byte[] buildIv(String iv, String operation) { + byte[] ivBytes = iv.getBytes(StandardCharsets.UTF_8); + if (ivBytes.length != IV_SIZE) { + throw CommonError.illegalArgument( + String.valueOf(ivBytes.length), operation + ": iv must be 16 bytes"); + } + return ivBytes; + } + + /** Rejects array / {@code byte[]} / map inputs that would otherwise be encrypted as garbage. */ + private static void rejectNonScalar(Object value, String argName, String operation) { + if (value.getClass().isArray() || value instanceof Map) { + throw CommonError.illegalArgument( + value.getClass().getName(), + operation + + ": unsupported input type, " + + argName + + " must be a scalar string"); + } + } + + private static SecretKeySpec cachedKey(String key, String operation) { + // Lock-free fast path on a hit; on a miss, putIfAbsent atomically inserts (per-key, not a + // global lock). A concurrent double-derivation is harmless since buildKey is deterministic. + SecretKeySpec cached = KEY_CACHE.get(key); + if (cached != null) { + return cached; + } + SecretKeySpec derived = buildKey(key, operation); + SecretKeySpec prior = KEY_CACHE.putIfAbsent(key, derived); + if (prior != null) { + return prior; + } + if (KEY_CACHE.size() > KEY_CACHE_MAX_SIZE) { + evictOne(); + } + return derived; + } + + private static void evictOne() { + // Weakly-consistent iterator; removing one entry keeps residency approximately bounded + // without a global lock and without strict LRU ordering. + Iterator<String> it = KEY_CACHE.keySet().iterator(); + if (it.hasNext()) { + it.next(); + it.remove(); + } + } +} diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLCryptoFunctionsTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLCryptoFunctionsTest.java new file mode 100644 index 0000000000..83377381cb --- /dev/null +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/SQLCryptoFunctionsTest.java @@ -0,0 +1,150 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.transform.sql; + +import org.apache.seatunnel.api.configuration.ReadonlyConfig; +import org.apache.seatunnel.api.table.catalog.CatalogTable; +import org.apache.seatunnel.api.table.catalog.CatalogTableUtil; +import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRow; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.List; + +public class SQLCryptoFunctionsTest { + + private SeaTunnelRow runSql(String query, SeaTunnelRowType rowType, Object... values) { + CatalogTable table = CatalogTableUtil.getCatalogTable("test", rowType); + ReadonlyConfig config = ReadonlyConfig.fromMap(Collections.singletonMap("query", query)); + SQLTransform transform = new SQLTransform(config, table); + List<SeaTunnelRow> out = transform.transformRow(new SeaTunnelRow(values)); + Assertions.assertNotNull(out); + Assertions.assertFalse(out.isEmpty()); + return out.get(0); + } + + @Test + public void testAesRoundTripWithPassphrase() { + // End-to-end: exercises ZetaSQLFunction dispatch + ZetaSQLType inference, not just the + // static CryptoFunction methods. + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"name"}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); + + SeaTunnelRow outRow = + runSql( + "select AES_DECRYPT(AES_ENCRYPT(name, 'mySecretPass'), 'mySecretPass') as r" + + " from dual", + rowType, + "Hello SeaTunnel"); + + Assertions.assertInstanceOf(String.class, outRow.getField(0)); + Assertions.assertEquals("Hello SeaTunnel", outRow.getField(0)); + } + + @Test + public void testAesRoundTripWithExplicitIv() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"name"}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); + + SeaTunnelRow outRow = + runSql( + "select AES_DECRYPT(" + + "AES_ENCRYPT(name, 'mySecretPass', '1234567890123456')," + + " 'mySecretPass', '1234567890123456') as r from dual", + rowType, + "Hello SeaTunnel"); + + Assertions.assertEquals("Hello SeaTunnel", outRow.getField(0)); + } + + @Test + public void testAesEncryptReturnsBase64String() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"name"}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); + + SeaTunnelRow outRow = + runSql("select AES_ENCRYPT(name, 'mySecretPass') as c from dual", rowType, "plain"); + + // Result type must be STRING (verifies ZetaSQLType.getFunctionType registration). + Assertions.assertInstanceOf(String.class, outRow.getField(0)); + // A random IV makes the ciphertext non-deterministic, but it must differ from the input + // and be non-empty. + Assertions.assertNotNull(outRow.getField(0)); + Assertions.assertNotEquals("plain", outRow.getField(0)); + } + + @Test + public void testAesExplicitIvIsDeterministic() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"name"}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); + + SeaTunnelRow out1 = + runSql( + "select AES_ENCRYPT(name, 'mySecretPass', '1234567890123456') as c" + + " from dual", + rowType, + "deterministic"); + SeaTunnelRow out2 = + runSql( + "select AES_ENCRYPT(name, 'mySecretPass', '1234567890123456') as c" + + " from dual", + rowType, + "deterministic"); + + Assertions.assertEquals(out1.getField(0), out2.getField(0)); + } + + @Test + public void testAesEncryptNullReturnsNull() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"name"}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); + + SeaTunnelRow outRow = + runSql( + "select AES_ENCRYPT(name, 'mySecretPass') as c from dual", + rowType, + (Object) null); + + Assertions.assertNull(outRow.getField(0)); + } + + @Test + public void testAesDecryptNullReturnsNull() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"name"}, new SeaTunnelDataType[] {BasicType.STRING_TYPE}); + + SeaTunnelRow outRow = + runSql( + "select AES_DECRYPT(name, 'mySecretPass') as r from dual", + rowType, + (Object) null); + + Assertions.assertNull(outRow.getField(0)); + } +} diff --git a/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/CryptoFunctionTest.java b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/CryptoFunctionTest.java new file mode 100644 index 0000000000..b15270dbf7 --- /dev/null +++ b/seatunnel-transforms-v2/src/test/java/org/apache/seatunnel/transform/sql/zeta/functions/CryptoFunctionTest.java @@ -0,0 +1,479 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.transform.sql.zeta.functions; + +import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException; +import org.apache.seatunnel.transform.encrypt.encryptor.AesCbcEncryptor; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import javax.crypto.Cipher; +import javax.crypto.spec.IvParameterSpec; +import javax.crypto.spec.SecretKeySpec; + +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Base64; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +public class CryptoFunctionTest { + + private static final String PASSPHRASE = "mySecretPass"; + private static final String EXPLICIT_IV = "1234567890123456"; // 16 bytes + private static final String PLAINTEXT = "Hello SeaTunnel"; + + private List<Object> args(Object... values) { + List<Object> list = new ArrayList<>(); + for (Object v : values) { + list.add(v); + } + return list; + } + + @Test + public void testRoundTripWithPassphraseAndRandomIv() { + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE)); + Assertions.assertNotNull(cipher); + Assertions.assertNotEquals(PLAINTEXT, cipher); + String decrypted = CryptoFunction.aesDecrypt(args(cipher, PASSPHRASE)); + Assertions.assertEquals(PLAINTEXT, decrypted); + } + + @Test + public void testRoundTripWithPassphraseAndExplicitIv() { + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV)); + Assertions.assertNotNull(cipher); + // With an explicit IV the ciphertext does not carry the IV prefix. + String decrypted = CryptoFunction.aesDecrypt(args(cipher, PASSPHRASE, EXPLICIT_IV)); + Assertions.assertEquals(PLAINTEXT, decrypted); + } + + @Test + public void testRoundTripWithBase64KeyAndRandomIv() { + byte[] rawKey = new byte[16]; + for (int i = 0; i < rawKey.length; i++) { + rawKey[i] = (byte) i; + } + String base64Key = "base64:" + Base64.getEncoder().encodeToString(rawKey); + + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, base64Key)); + Assertions.assertNotNull(cipher); + String decrypted = CryptoFunction.aesDecrypt(args(cipher, base64Key)); + Assertions.assertEquals(PLAINTEXT, decrypted); + } + + @Test + public void testRoundTripWithBase64KeyAndExplicitIv() { + byte[] rawKey = new byte[32]; + for (int i = 0; i < rawKey.length; i++) { + rawKey[i] = (byte) (i + 1); + } + String base64Key = "base64:" + Base64.getEncoder().encodeToString(rawKey); + + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, base64Key, EXPLICIT_IV)); + String decrypted = CryptoFunction.aesDecrypt(args(cipher, base64Key, EXPLICIT_IV)); + Assertions.assertEquals(PLAINTEXT, decrypted); + } + + @Test + public void testRandomIvProducesDifferentCiphertext() { + String cipher1 = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE)); + String cipher2 = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE)); + // A random IV makes the ciphertext non-deterministic. + Assertions.assertNotEquals(cipher1, cipher2); + Assertions.assertEquals(PLAINTEXT, CryptoFunction.aesDecrypt(args(cipher1, PASSPHRASE))); + Assertions.assertEquals(PLAINTEXT, CryptoFunction.aesDecrypt(args(cipher2, PASSPHRASE))); + } + + @Test + public void testExplicitIvIsDeterministic() { + String cipher1 = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV)); + String cipher2 = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV)); + Assertions.assertEquals(cipher1, cipher2); + } + + @Test + public void testEncryptNullValueReturnsNull() { + Assertions.assertNull(CryptoFunction.aesEncrypt(args(null, PASSPHRASE))); + } + + @Test + public void testDecryptNullValueReturnsNull() { + Assertions.assertNull(CryptoFunction.aesDecrypt(args(null, PASSPHRASE))); + } + + @Test + public void testEncryptTooFewArgsThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, () -> CryptoFunction.aesEncrypt(args(PLAINTEXT))); + } + + @Test + public void testEncryptTooManyArgsThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV, "extra"))); + } + + @Test + public void testDecryptTooFewArgsThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args("ciphertext"))); + } + + @Test + public void testDecryptTooManyArgsThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> + CryptoFunction.aesDecrypt( + args("ciphertext", PASSPHRASE, EXPLICIT_IV, "extra"))); + } + + @Test + public void testEncryptWithNullKeyThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, null))); + } + + @Test + public void testDecryptWithNullKeyThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args("ciphertext", null))); + } + + @Test + public void testEncryptWithEmptyKeyThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, ""))); + } + + @Test + public void testEncryptWithBlankKeyThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, " "))); + } + + @Test + public void testBase64KeyMalformedThrows() { + // The base64: prefix is present but the remainder is not valid Base64 content. + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, "base64:@@@@"))); + } + + @Test + public void testDecryptWithWrongKeyDoesNotLeakPlaintext() { + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE)); + // CBC/PKCS5Padding decryption with a wrong key usually fails the padding check, but a + // random block can pass it (~1/256). Both outcomes are acceptable: throwing, or returning + // garbage, as long as the real plaintext is never recovered and no secret is leaked. + try { + String result = CryptoFunction.aesDecrypt(args(cipher, "wrongPassphrase")); + Assertions.assertNotEquals(PLAINTEXT, result); + } catch (SeaTunnelRuntimeException e) { + assertMessageHasNoSecrets(e, PASSPHRASE, PLAINTEXT, cipher); + } + } + + @Test + public void testExplicitIvWrongLengthThrows() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, "short"))); + } + + @Test + public void testBase64KeyWrongLengthThrows() { + byte[] rawKey = new byte[10]; // invalid AES key length + String base64Key = "base64:" + Base64.getEncoder().encodeToString(rawKey); + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, base64Key))); + } + + @Test + public void testDecryptInvalidBase64Throws() { + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args("@@invalid@@", PASSPHRASE))); + } + + @Test + public void testDecryptTooShortPayloadThrows() { + // A valid Base64 string that decodes to fewer than 16 bytes cannot carry the IV. + String shortBase64 = Base64.getEncoder().encodeToString(new byte[8]); + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args(shortBase64, PASSPHRASE))); + } + + @Test + public void testRoundTripUnicodePlaintext() { + String unicode = "加密测试🔐数据"; + String cipher = CryptoFunction.aesEncrypt(args(unicode, PASSPHRASE)); + Assertions.assertEquals(unicode, CryptoFunction.aesDecrypt(args(cipher, PASSPHRASE))); + } + + @Test + public void testExplicitIvCiphertextCannotBeDecryptedWithoutIv() { + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV)); + // The explicit-IV ciphertext carries no IV prefix, so decrypting without an IV interprets + // the first 16 ciphertext bytes as the IV and decrypts the rest with it. This must not + // return the original plaintext (it throws on a padding failure, or yields garbage). + try { + String result = CryptoFunction.aesDecrypt(args(cipher, PASSPHRASE)); + Assertions.assertNotEquals(PLAINTEXT, result); + } catch (SeaTunnelRuntimeException e) { + assertMessageHasNoSecrets(e, PASSPHRASE, PLAINTEXT, cipher); + } + } + + @Test + public void testRandomIvCiphertextRoundTripsWithoutExplicitIv() { + // Ciphertext produced without an IV embeds the IV in its first 16 bytes; decrypting + // without an IV must recover the original plaintext. + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE)); + Assertions.assertEquals(PLAINTEXT, CryptoFunction.aesDecrypt(args(cipher, PASSPHRASE))); + } + + @Test + public void testExplicitIvEncryptionIsDeterministic() { + // Two encryptions with the same passphrase and explicit IV must produce the same + // ciphertext, proving the SHA-256 key derivation is deterministic. + String cipher1 = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV)); + String cipher2 = CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV)); + Assertions.assertEquals(cipher1, cipher2); + } + + @Test + public void testRoundTripWithAes192Base64Key() { + // Covers the 24-byte (AES-192) branch of the raw key length validation. + byte[] rawKey = new byte[24]; + for (int i = 0; i < rawKey.length; i++) { + rawKey[i] = (byte) (i + 5); + } + String base64Key = "base64:" + Base64.getEncoder().encodeToString(rawKey); + String cipher = CryptoFunction.aesEncrypt(args(PLAINTEXT, base64Key, EXPLICIT_IV)); + Assertions.assertEquals( + PLAINTEXT, CryptoFunction.aesDecrypt(args(cipher, base64Key, EXPLICIT_IV))); + } + + @Test + public void testDecryptExplicitIvEmptyPayloadThrows() { + // An empty ciphertext combined with an explicit IV has nothing to decrypt. + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args("", PASSPHRASE, EXPLICIT_IV))); + } + + @Test + public void testExplicitNullIvThrows() { + // An explicit null IV is rejected (consistent with a null key) so a column of mixed + // null/non-null IVs cannot silently produce two different ciphertext layouts. Omit the + // argument entirely to use a random IV. + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, null))); + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args("ciphertext", PASSPHRASE, null))); + } + + @Test + public void testErrorMessageDoesNotContainSecrets() { + // None of the error paths may echo the key, the plaintext, the ciphertext or the arg list + // into the exception message (which surfaces in engine logs). + String secretKey = "superSecretKeyValue"; + String sensitive = "sensitiveDataValue"; + + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> + CryptoFunction.aesEncrypt( + args(sensitive, secretKey, EXPLICIT_IV, "extra"))); + assertMessageHasNoSecrets(e, secretKey, sensitive, EXPLICIT_IV); + + e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(sensitive, null))); + assertMessageHasNoSecrets(e, secretKey, sensitive); + + e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(sensitive, ""))); + assertMessageHasNoSecrets(e, secretKey, sensitive); + + String badBase64Key = "base64:" + Base64.getEncoder().encodeToString(new byte[10]); + e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(sensitive, badBase64Key))); + assertMessageHasNoSecrets(e, secretKey, sensitive, badBase64Key); + + e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args("@@invalid@@", secretKey))); + assertMessageHasNoSecrets(e, secretKey, sensitive); + } + + @Test + public void testEncryptByteArrayInputThrows() { + // A byte[] column would otherwise be encrypted as "[B@...", which can never be recovered. + byte[] bytes = PLAINTEXT.getBytes(StandardCharsets.UTF_8); + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(bytes, PASSPHRASE))); + assertMessageHasNoSecrets(e, PASSPHRASE, PLAINTEXT); + } + + @Test + public void testEncryptArrayAndMapInputThrows() { + String[] array = new String[] {"a", "b"}; + Map<String, String> map = new HashMap<>(); + map.put("k", "v"); + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(array, PASSPHRASE))); + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(map, PASSPHRASE))); + } + + @Test + public void testEncryptByteArrayKeyThrows() { + // A byte[] key would be stringified as "[B@..." and silently derive a different AES key + // per row (each byte[] is a distinct object), so it must be rejected like a byte[] value. + byte[] bytes = PLAINTEXT.getBytes(StandardCharsets.UTF_8); + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, bytes))); + assertMessageHasNoSecrets(e, PASSPHRASE, PLAINTEXT); + } + + @Test + public void testDecryptByteArrayKeyThrows() { + byte[] bytes = PASSPHRASE.getBytes(StandardCharsets.UTF_8); + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args("ciphertext", bytes))); + assertMessageHasNoSecrets(e, PASSPHRASE, PLAINTEXT); + } + + @Test + public void testEncryptByteArrayIvThrows() { + byte[] bytes = new byte[16]; + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, bytes))); + assertMessageHasNoSecrets(e, PASSPHRASE, PLAINTEXT); + // Pins the iv guard in resolveIv: without it a byte[] iv would be stringified to "[B@..." + // and rejected only for the wrong length, with a message that omits this phrase. + Assertions.assertTrue( + e.getMessage().contains("unsupported input type"), + "byte[] iv must be rejected by the non-scalar guard"); + } + + @Test + public void testDecryptByteArrayIvThrows() { + // The value must be valid Base64 so the decode succeeds and execution reaches the IV check. + String validBase64 = Base64.getEncoder().encodeToString(new byte[32]); + byte[] bytes = new byte[16]; + SeaTunnelRuntimeException e = + Assertions.assertThrows( + SeaTunnelRuntimeException.class, + () -> CryptoFunction.aesDecrypt(args(validBase64, PASSPHRASE, bytes))); + assertMessageHasNoSecrets(e, PASSPHRASE, PLAINTEXT); + // Pins the iv guard in resolveIv: without it a byte[] iv would be stringified to "[B@..." + // and rejected only for the wrong length, with a message that omits this phrase. + Assertions.assertTrue( + e.getMessage().contains("unsupported input type"), + "byte[] iv must be rejected by the non-scalar guard"); + } + + @Test + public void testInteropWithAesCbcEncryptor() { + // The base64: key form must be wire-compatible with FieldEncrypt's AesCbcEncryptor: + // same algorithm, same Base64(IV || ciphertext) layout. + byte[] rawKey = new byte[16]; + for (int i = 0; i < rawKey.length; i++) { + rawKey[i] = (byte) (i + 1); + } + String base64Key = "base64:" + Base64.getEncoder().encodeToString(rawKey); + String plain = "interop data"; + + AesCbcEncryptor encryptor = new AesCbcEncryptor(); + encryptor.init(base64Key); + + // AesCbcEncryptor encrypts, CryptoFunction decrypts. + String cipher = encryptor.encrypt(plain); + Assertions.assertEquals(plain, CryptoFunction.aesDecrypt(args(cipher, base64Key))); + + // CryptoFunction encrypts, AesCbcEncryptor decrypts. + String cipher2 = CryptoFunction.aesEncrypt(args(plain, base64Key)); + Assertions.assertEquals(plain, encryptor.decrypt(cipher2)); + } + + @Test + public void testPassphraseKeyDerivationKnownAnswer() throws Exception { + // Known-answer: independently derive the passphrase key (SHA-256, first 16 bytes) and the + // ciphertext with raw javax.crypto, then assert CryptoFunction produces the exact same + // output. This pins both the key derivation and the explicit-IV wire format. + MessageDigest sha256 = MessageDigest.getInstance("SHA-256"); + byte[] digest = sha256.digest(PASSPHRASE.getBytes(StandardCharsets.UTF_8)); + byte[] keyBytes = Arrays.copyOf(digest, 16); + + Cipher cipher = Cipher.getInstance("AES/CBC/PKCS5Padding"); + cipher.init( + Cipher.ENCRYPT_MODE, + new SecretKeySpec(keyBytes, "AES"), + new IvParameterSpec(EXPLICIT_IV.getBytes(StandardCharsets.UTF_8))); + byte[] encrypted = cipher.doFinal(PLAINTEXT.getBytes(StandardCharsets.UTF_8)); + String expected = Base64.getEncoder().encodeToString(encrypted); + + Assertions.assertEquals( + expected, CryptoFunction.aesEncrypt(args(PLAINTEXT, PASSPHRASE, EXPLICIT_IV))); + } + + private static void assertMessageHasNoSecrets(SeaTunnelRuntimeException e, String... secrets) { + String message = e.getMessage(); + for (String secret : secrets) { + Assertions.assertFalse( + message.contains(secret), + "Exception message must not contain secret material: " + secret); + } + } +}
