From 76754e05d23245fce76a63c46319a09ae027afc7 Mon Sep 17 00:00:00 2001 From: HTHou Date: Wed, 15 Jul 2026 18:21:15 +0800 Subject: [PATCH] Add pluggable page-level AEAD encryption support --- .../apache/tsfile/i18n/messages.properties | 28 +++ .../apache/tsfile/i18n/messages_zh.properties | 28 +++ .../tsfile/encrypt/EncryptParameter.java | 213 +++++++++++++++- .../apache/tsfile/encrypt/EncryptUtils.java | 16 ++ .../encrypt/EncryptionProviderRegistry.java | 77 ++++++ .../org/apache/tsfile/encrypt/IDecryptor.java | 11 + .../org/apache/tsfile/encrypt/IEncrypt.java | 15 +- .../tsfile/encrypt/IEncryptProvider.java | 35 +++ .../org/apache/tsfile/encrypt/IEncryptor.java | 11 + .../tsfile/encrypt/PageCryptoContext.java | 171 +++++++++++++ .../org/apache/tsfile/file/MetaMarker.java | 3 + .../file/header/FileEncryptionHeader.java | 178 ++++++++++++++ .../tsfile/read/TsFileSequenceReader.java | 97 +++++++- .../tsfile/read/UnClosedTsFileReader.java | 3 +- .../org/apache/tsfile/read/common/Chunk.java | 4 + .../chunk/AbstractAlignedChunkReader.java | 30 ++- .../tsfile/read/reader/chunk/ChunkReader.java | 80 +++++- .../read/reader/page/LazyLoadPageData.java | 45 +++- .../apache/tsfile/utils/TsFileSketchTool.java | 3 +- .../org/apache/tsfile/write/TsFileWriter.java | 64 +++-- .../tsfile/write/chunk/ChunkWriterImpl.java | 7 +- .../tsfile/write/chunk/TimeChunkWriter.java | 7 +- .../tsfile/write/chunk/ValueChunkWriter.java | 7 +- .../tsfile/write/page/EncodedPageBody.java | 48 ++++ .../tsfile/write/page/PageBodyEncoder.java | 83 +++++++ .../apache/tsfile/write/page/PageWriter.java | 61 +---- .../tsfile/write/page/TimePageWriter.java | 60 +---- .../tsfile/write/page/ValuePageWriter.java | 60 +---- .../v4/AbstractTableModelTsFileWriter.java | 51 ++-- .../write/writer/ForceAppendTsFileWriter.java | 49 +++- .../writer/RestorableTsFileIOWriter.java | 67 +++++- .../tsfile/write/writer/TsFileIOWriter.java | 72 ++++++ .../tsfile/encrypt/PageCryptoContextTest.java | 89 +++++++ .../encrypt/TestAeadEncryptionProvider.java | 227 ++++++++++++++++++ .../file/header/FileEncryptionHeaderTest.java | 94 ++++++++ .../tsfile/read/UnClosedTsFileReaderTest.java | 58 +++++ .../tsfile/write/TDEPageAeadTsFileTest.java | 162 +++++++++++++ .../writer/ForceAppendTsFileWriterTest.java | 65 +++++ .../writer/RestorableTsFileIOWriterTest.java | 64 +++++ .../write/writer/TimePageWriterTest.java | 62 +++++ 40 files changed, 2280 insertions(+), 225 deletions(-) create mode 100644 java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptionProviderRegistry.java create mode 100644 java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptProvider.java create mode 100644 java/tsfile/src/main/java/org/apache/tsfile/encrypt/PageCryptoContext.java create mode 100644 java/tsfile/src/main/java/org/apache/tsfile/file/header/FileEncryptionHeader.java create mode 100644 java/tsfile/src/main/java/org/apache/tsfile/write/page/EncodedPageBody.java create mode 100644 java/tsfile/src/main/java/org/apache/tsfile/write/page/PageBodyEncoder.java create mode 100644 java/tsfile/src/test/java/org/apache/tsfile/encrypt/PageCryptoContextTest.java create mode 100644 java/tsfile/src/test/java/org/apache/tsfile/encrypt/TestAeadEncryptionProvider.java create mode 100644 java/tsfile/src/test/java/org/apache/tsfile/file/header/FileEncryptionHeaderTest.java create mode 100644 java/tsfile/src/test/java/org/apache/tsfile/write/TDEPageAeadTsFileTest.java diff --git a/java/common/src/main/resources/org/apache/tsfile/i18n/messages.properties b/java/common/src/main/resources/org/apache/tsfile/i18n/messages.properties index 98909f7a6..5c5ad94be 100644 --- a/java/common/src/main/resources/org/apache/tsfile/i18n/messages.properties +++ b/java/common/src/main/resources/org/apache/tsfile/i18n/messages.properties @@ -1111,6 +1111,34 @@ error.encrypt.encrypt_no_constructor = Get constructor for encryptor failed: %1$ # EncryptUtils.getEncrypt — instantiation failed (encryptType arg) error.encrypt.encrypt_instantiation_failed = New encryptor instance failed: %1$s +# EncryptionProviderRegistry — provider registration and lookup +error.encrypt.invalid_provider_id = Encryption provider id must not be null or empty +error.encrypt.duplicate_provider = Encryption provider id is already registered: %1$s +error.encrypt.provider_not_found = Encryption provider is not registered: %1$s +error.encrypt.provider_returned_null = Encryption provider returned null: %1$s + +# EncryptParameter — use after sensitive state was destroyed +error.encrypt.parameter_destroyed = Encryption parameters have been destroyed +error.encrypt.invalid_page_body_overhead = Invalid page body overhead %1$s for encryption profile %2$s +error.encrypt.page_body_too_short = Encrypted page body size %1$s is smaller than AEAD overhead %2$s +error.encrypt.page_context_invalid_parameter = Page crypto context requires page AEAD parameters +error.encrypt.page_context_invalid_file_id = Page crypto context requires a 16-byte file crypto id +error.encrypt.page_context_invalid_sizes = Invalid page crypto context values, uncompressed size: %1$s, compressed plaintext size: %2$s, page index: %3$s +error.encrypt.page_output_size_mismatch = AEAD page output size mismatch, expected %1$s bytes but got %2$s +error.encrypt.page_plaintext_size_mismatch = AEAD page plaintext size mismatch, expected %1$s bytes but got %2$s +error.chunk.merge_page_aead_unsupported = Page-AEAD encrypted chunks must be decrypted and rewritten before merging + +# FileEncryptionHeader — malformed or unsupported file encryption metadata +error.file.encryption_header_invalid_size = Invalid file encryption header size: %1$s +error.file.encryption_header_unsupported_version = Unsupported file encryption header version: %1$s +error.file.encryption_header_malformed = Malformed file encryption header +error.file.encryption_header_trailing_bytes = File encryption header contains trailing bytes +error.file.encryption_header_invalid_component_size = Invalid file encryption header component size: %1$s +error.file.encryption_header_truncated = File encryption header is truncated, expected %1$s bytes but read %2$s +error.file.encryption_header_invalid_field = Invalid file encryption header field: %1$s +error.write.encryption_header_must_precede_data = File encryption header must be written before TsFile data +error.write.encrypted_chunk_context_mismatch = Encrypted chunks can only be copied within the same file encryption context + # === compress === # ICompressor.getCompressor / IUnCompressor.getUnCompressor — null type diff --git a/java/common/src/main/resources/org/apache/tsfile/i18n/messages_zh.properties b/java/common/src/main/resources/org/apache/tsfile/i18n/messages_zh.properties index a1a437cfb..2298e9acc 100644 --- a/java/common/src/main/resources/org/apache/tsfile/i18n/messages_zh.properties +++ b/java/common/src/main/resources/org/apache/tsfile/i18n/messages_zh.properties @@ -1111,6 +1111,34 @@ error.encrypt.encrypt_no_constructor = 获取 encryptor 构造函数失败: %1$s # EncryptUtils.getEncrypt — instantiation failed (encryptType arg) error.encrypt.encrypt_instantiation_failed = 实例化 encryptor 失败: %1$s +# EncryptionProviderRegistry — provider registration and lookup +error.encrypt.invalid_provider_id = encryption provider id 不能为空 +error.encrypt.duplicate_provider = encryption provider id 已注册: %1$s +error.encrypt.provider_not_found = encryption provider 未注册: %1$s +error.encrypt.provider_returned_null = encryption provider 返回 null: %1$s + +# EncryptParameter — use after sensitive state was destroyed +error.encrypt.parameter_destroyed = encryption parameters 已销毁 +error.encrypt.invalid_page_body_overhead = encryption profile %2$s 的 page body 额外开销 %1$s 无效 +error.encrypt.page_body_too_short = encrypted page body 大小 %1$s 小于 AEAD 额外开销 %2$s +error.encrypt.page_context_invalid_parameter = page crypto context 需要 page AEAD 参数 +error.encrypt.page_context_invalid_file_id = page crypto context 需要 16 字节的 file crypto id +error.encrypt.page_context_invalid_sizes = page crypto context 值无效，uncompressed size: %1$s，compressed plaintext size: %2$s，page index: %3$s +error.encrypt.page_output_size_mismatch = AEAD page 输出大小不匹配，预期 %1$s 字节，实际 %2$s 字节 +error.encrypt.page_plaintext_size_mismatch = AEAD page plaintext 大小不匹配，预期 %1$s 字节，实际 %2$s 字节 +error.chunk.merge_page_aead_unsupported = Page-AEAD encrypted chunk 合并前必须先解密并重写 + +# FileEncryptionHeader — malformed or unsupported file encryption metadata +error.file.encryption_header_invalid_size = file encryption header 大小无效: %1$s +error.file.encryption_header_unsupported_version = 不支持的 file encryption header 版本: %1$s +error.file.encryption_header_malformed = file encryption header 格式错误 +error.file.encryption_header_trailing_bytes = file encryption header 包含多余字节 +error.file.encryption_header_invalid_component_size = file encryption header 组件大小无效: %1$s +error.file.encryption_header_truncated = file encryption header 已截断，预期 %1$s 字节，实际读取 %2$s 字节 +error.file.encryption_header_invalid_field = file encryption header 字段无效: %1$s +error.write.encryption_header_must_precede_data = file encryption header 必须写在 TsFile data 之前 +error.write.encrypted_chunk_context_mismatch = 仅允许在相同 file encryption context 中复制 encrypted chunk + # === compress === # ICompressor.getCompressor / IUnCompressor.getUnCompressor — null type diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptParameter.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptParameter.java index f7adb5221..e5a1c8197 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptParameter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptParameter.java @@ -18,14 +18,83 @@ */ package org.apache.tsfile.encrypt; -public class EncryptParameter { +import org.apache.tsfile.exception.encrypt.EncryptException; +import org.apache.tsfile.i18n.Messages; + +import javax.security.auth.Destroyable; + +import java.util.Arrays; + +/** + * Parameters required to encrypt or decrypt one TsFile. + * + *

The two-argument constructor preserves the legacy encryption API. New file-level encryption + * schemes should use {@link #pageAeadBuilder()} and identify an {@link IEncryptProvider} and one of + * its immutable encryption profiles by stable ids. Runtime implementation class names and algorithm + * implementation details are deliberately not part of the persisted parameters. + */ +public class EncryptParameter implements AutoCloseable, Destroyable { + + public static final int FILE_CRYPTO_ID_LENGTH = 16; private final String type; private final byte[] key; + private final boolean pageAead; + private final String providerId; + private final String profileId; + private final String keyId; + private final String keyVersion; + private final byte[] wrappedDataKey; + private final byte[] fileCryptoId; + + private transient volatile IEncrypt fileEncrypt; + private transient volatile boolean destroyed; public EncryptParameter(String type, byte[] key) { this.type = type; this.key = key; + this.pageAead = false; + this.providerId = null; + this.profileId = null; + this.keyId = null; + this.keyVersion = null; + this.wrappedDataKey = null; + this.fileCryptoId = null; + } + + private EncryptParameter(Builder builder) { + this.type = null; + this.key = copy(builder.key); + this.pageAead = true; + this.providerId = builder.providerId; + this.profileId = builder.profileId; + this.keyId = builder.keyId; + this.keyVersion = builder.keyVersion; + this.wrappedDataKey = copy(builder.wrappedDataKey); + this.fileCryptoId = copy(builder.fileCryptoId); + } + + public static Builder pageAeadBuilder() { + return new Builder(); + } + + /** Returns an independent parameter object with no initialized runtime provider state. */ + public EncryptParameter copy() { + if (destroyed) { + throw new EncryptException(Messages.get("error.encrypt.parameter_destroyed")); + } + if (!isTdePageAead()) { + return new EncryptParameter(type, key); + } + return pageAeadBuilder() + .key(key) + .providerId(providerId) + .profileId(profileId) + .keyId(keyId) + .keyVersion(keyVersion) + .wrappedDataKey(wrappedDataKey) + .fileCryptoId(fileCryptoId) + .build(); } public byte[] getKey() { @@ -35,4 +104,146 @@ public byte[] getKey() { public String getType() { return type; } + + public String getProviderId() { + return providerId; + } + + public String getProfileId() { + return profileId; + } + + public String getKeyId() { + return keyId; + } + + public String getKeyVersion() { + return keyVersion; + } + + public byte[] getWrappedDataKey() { + return copy(wrappedDataKey); + } + + public byte[] getFileCryptoId() { + return copy(fileCryptoId); + } + + public int getPageBodyOverhead() { + int overhead = getOrCreateFileEncrypt().getPageBodyOverhead(); + if (overhead <= 0) { + throw new EncryptException( + Messages.format("error.encrypt.invalid_page_body_overhead", overhead, profileId)); + } + return overhead; + } + + public boolean isTdePageAead() { + return pageAead; + } + + IEncrypt getOrCreateFileEncrypt() { + if (destroyed) { + throw new EncryptException(Messages.get("error.encrypt.parameter_destroyed")); + } + IEncrypt current = fileEncrypt; + if (current == null) { + synchronized (this) { + if (destroyed) { + throw new EncryptException(Messages.get("error.encrypt.parameter_destroyed")); + } + current = fileEncrypt; + if (current == null) { + current = EncryptionProviderRegistry.create(this); + fileEncrypt = current; + } + } + } + return current; + } + + @Override + public boolean isDestroyed() { + return destroyed; + } + + @Override + public void destroy() { + IEncrypt current; + synchronized (this) { + if (destroyed) { + return; + } + destroyed = true; + current = fileEncrypt; + fileEncrypt = null; + if (key != null && isTdePageAead()) { + Arrays.fill(key, (byte) 0); + } + } + if (current != null) { + current.close(); + } + } + + @Override + public void close() { + destroy(); + } + + private static byte[] copy(byte[] value) { + return value == null ? null : Arrays.copyOf(value, value.length); + } + + public static final class Builder { + + private byte[] key; + private String providerId; + private String profileId; + private String keyId; + private String keyVersion; + private byte[] wrappedDataKey; + private byte[] fileCryptoId; + + private Builder() {} + + public Builder key(byte[] key) { + this.key = copy(key); + return this; + } + + public Builder providerId(String providerId) { + this.providerId = providerId; + return this; + } + + public Builder profileId(String profileId) { + this.profileId = profileId; + return this; + } + + public Builder keyId(String keyId) { + this.keyId = keyId; + return this; + } + + public Builder keyVersion(String keyVersion) { + this.keyVersion = keyVersion; + return this; + } + + public Builder wrappedDataKey(byte[] wrappedDataKey) { + this.wrappedDataKey = copy(wrappedDataKey); + return this; + } + + public Builder fileCryptoId(byte[] fileCryptoId) { + this.fileCryptoId = copy(fileCryptoId); + return this; + } + + public EncryptParameter build() { + return new EncryptParameter(this); + } + } } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptUtils.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptUtils.java index 42ccec8f7..dcb712ef0 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptUtils.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptUtils.java @@ -248,6 +248,12 @@ public static EncryptParameter getEncryptParameter() { /** Get the second EncryptParameter object according to the given type and first key. */ public static EncryptParameter getEncryptParameter(EncryptParameter param) { + if (param == null) { + return new EncryptParameter("org.apache.tsfile.encrypt.UNENCRYPTED", null); + } + if (param.isTdePageAead()) { + return param; + } return encryptParamCache.computeIfAbsent(param, EncryptUtils::generateEncryptParameter); } @@ -285,6 +291,16 @@ public static IEncrypt getEncrypt() { return getEncrypt(TSFileDescriptor.getInstance().getConfig()); } + public static IEncrypt getEncrypt(EncryptParameter encryptParameter) { + if (encryptParameter == null) { + return getEncrypt("org.apache.tsfile.encrypt.UNENCRYPTED", null); + } + if (encryptParameter.isTdePageAead()) { + return encryptParameter.getOrCreateFileEncrypt(); + } + return getEncrypt(encryptParameter.getType(), encryptParameter.getKey()); + } + public static IEncrypt getEncrypt(String encryptType, byte[] dataEncryptKey) { try { String className = getEncryptClass(encryptType); diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptionProviderRegistry.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptionProviderRegistry.java new file mode 100644 index 000000000..76091a3c4 --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/EncryptionProviderRegistry.java @@ -0,0 +1,77 @@ +/* + * 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.tsfile.encrypt; + +import org.apache.tsfile.exception.encrypt.EncryptException; +import org.apache.tsfile.i18n.Messages; + +import java.util.ServiceLoader; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; + +/** Registry for file-level encryption providers identified by stable provider ids. */ +public final class EncryptionProviderRegistry { + + private static final ConcurrentHashMap PROVIDERS = + new ConcurrentHashMap<>(); + private static final AtomicBoolean SERVICES_LOADED = new AtomicBoolean(); + + private EncryptionProviderRegistry() {} + + public static void registerProvider(IEncryptProvider provider) { + String providerId = provider == null ? null : provider.getProviderId(); + if (providerId == null || providerId.trim().isEmpty()) { + throw new EncryptException(Messages.get("error.encrypt.invalid_provider_id")); + } + providerId = providerId.trim(); + IEncryptProvider existing = PROVIDERS.putIfAbsent(providerId, provider); + if (existing != null && !existing.getClass().equals(provider.getClass())) { + throw new EncryptException(Messages.format("error.encrypt.duplicate_provider", providerId)); + } + } + + public static void unregisterProvider(String providerId) { + if (providerId != null) { + PROVIDERS.remove(providerId); + } + } + + static IEncrypt create(EncryptParameter encryptParameter) { + loadServices(); + String providerId = encryptParameter.getProviderId(); + IEncryptProvider provider = providerId == null ? null : PROVIDERS.get(providerId); + if (provider == null) { + throw new EncryptException(Messages.format("error.encrypt.provider_not_found", providerId)); + } + IEncrypt encrypt = provider.create(encryptParameter); + if (encrypt == null) { + throw new EncryptException( + Messages.format("error.encrypt.provider_returned_null", providerId)); + } + return encrypt; + } + + private static void loadServices() { + if (SERVICES_LOADED.compareAndSet(false, true)) { + for (IEncryptProvider provider : ServiceLoader.load(IEncryptProvider.class)) { + registerProvider(provider); + } + } + } +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IDecryptor.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IDecryptor.java index b906e5b72..014c75501 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IDecryptor.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IDecryptor.java @@ -57,6 +57,12 @@ static IDecryptor getDecryptor(String type, byte[] key) { } static IDecryptor getDecryptor(EncryptParameter encryptParam) { + if (encryptParam == null) { + return getDecryptor("org.apache.tsfile.encrypt.UNENCRYPTED", null); + } + if (encryptParam.isTdePageAead()) { + return EncryptUtils.getEncrypt(encryptParam).getDecryptor(); + } String type = encryptParam.getType(); byte[] key = encryptParam.getKey(); return getDecryptor(type, key); @@ -66,5 +72,10 @@ static IDecryptor getDecryptor(EncryptParameter encryptParam) { byte[] decrypt(byte[] data, int offset, int size); + default byte[] decryptPage( + byte[] data, int offset, int size, PageCryptoContext pageCryptoContext) { + return decrypt(data, offset, size); + } + EncryptionType getEncryptionType(); } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncrypt.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncrypt.java index 0ea6bbef7..7f8b5e0a3 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncrypt.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncrypt.java @@ -20,7 +20,7 @@ import java.util.concurrent.ConcurrentHashMap; -public interface IEncrypt { +public interface IEncrypt extends AutoCloseable { static ConcurrentHashMap> encryptMap = new ConcurrentHashMap<>(); @@ -30,4 +30,17 @@ public interface IEncrypt { IDecryptor getDecryptor(); IEncryptor getEncryptor(); + + /** + * Returns the fixed number of bytes added to every encrypted page body by this profile. + * + *

Page AEAD providers must override this method. The value is part of the profile contract and + * must not change while files using that profile exist. + */ + default int getPageBodyOverhead() { + return -1; + } + + @Override + default void close() {} } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptProvider.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptProvider.java new file mode 100644 index 000000000..d10262f51 --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptProvider.java @@ -0,0 +1,35 @@ +/* + * 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.tsfile.encrypt; + +/** + * Service provider interface for file-level encryption implementations. + * + *

Provider ids are persisted in a TsFile and therefore must remain stable across releases. A + * provider implementation may be registered explicitly through {@link EncryptionProviderRegistry} + * or discovered through {@link java.util.ServiceLoader}. Providers own the encryption profiles + * referenced by {@link EncryptParameter#getProfileId()}; a profile's algorithms, page layout, key + * envelope format, and page body overhead must remain immutable while files using it exist. + */ +public interface IEncryptProvider { + + String getProviderId(); + + IEncrypt create(EncryptParameter encryptParameter); +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptor.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptor.java index 2c0b6cf2e..325b9131b 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptor.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/IEncryptor.java @@ -57,6 +57,12 @@ static IEncryptor getEncryptor(String type, byte[] key) { } static IEncryptor getEncryptor(EncryptParameter encryptParam) { + if (encryptParam == null) { + return getEncryptor("org.apache.tsfile.encrypt.UNENCRYPTED", null); + } + if (encryptParam.isTdePageAead()) { + return EncryptUtils.getEncrypt(encryptParam).getEncryptor(); + } String type = encryptParam.getType(); byte[] key = encryptParam.getKey(); return getEncryptor(type, key); @@ -66,5 +72,10 @@ static IEncryptor getEncryptor(EncryptParameter encryptParam) { byte[] encrypt(byte[] data, int offset, int size); + default byte[] encryptPage( + byte[] data, int offset, int size, PageCryptoContext pageCryptoContext) { + return encrypt(data, offset, size); + } + EncryptionType getEncryptionType(); } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/encrypt/PageCryptoContext.java b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/PageCryptoContext.java new file mode 100644 index 000000000..b1bfd45c3 --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/encrypt/PageCryptoContext.java @@ -0,0 +1,171 @@ +/* + * 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.tsfile.encrypt; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.exception.encrypt.EncryptException; +import org.apache.tsfile.i18n.Messages; + +import java.nio.ByteBuffer; +import java.util.Arrays; + +/** + * Canonical context for page-level authenticated encryption. + * + *

Encryption and decryption construct the same associated data from persisted file encryption + * metadata and stable page fields. Providers should pass {@link #getAssociatedData()} to their AEAD + * implementation unchanged. + */ +public final class PageCryptoContext { + + private static final int ASSOCIATED_DATA_VERSION = 1; + + private final int uncompressedSize; + private final int compressedPlaintextSize; + private final int encryptedPageBodySize; + private final int pageIndex; + private final byte[] associatedData; + + private PageCryptoContext( + EncryptParameter parameter, + int uncompressedSize, + int compressedPlaintextSize, + int encryptedPageBodySize, + int pageIndex) { + validate(parameter, uncompressedSize, compressedPlaintextSize, pageIndex); + this.uncompressedSize = uncompressedSize; + this.compressedPlaintextSize = compressedPlaintextSize; + this.encryptedPageBodySize = encryptedPageBodySize; + this.pageIndex = pageIndex; + this.associatedData = + buildAssociatedData(parameter, uncompressedSize, compressedPlaintextSize, pageIndex); + } + + public static PageCryptoContext forEncryption( + EncryptParameter parameter, + int uncompressedSize, + int compressedPlaintextSize, + int pageIndex) { + validate(parameter, uncompressedSize, compressedPlaintextSize, pageIndex); + int encryptedPageBodySize = + Math.addExact(compressedPlaintextSize, parameter.getPageBodyOverhead()); + return new PageCryptoContext( + parameter, uncompressedSize, compressedPlaintextSize, encryptedPageBodySize, pageIndex); + } + + public static PageCryptoContext forDecryption( + EncryptParameter parameter, int uncompressedSize, int encryptedPageBodySize, int pageIndex) { + validate(parameter, uncompressedSize, 0, pageIndex); + int compressedPlaintextSize = encryptedPageBodySize - parameter.getPageBodyOverhead(); + if (compressedPlaintextSize < 0) { + throw new EncryptException( + Messages.format( + "error.encrypt.page_body_too_short", + encryptedPageBodySize, + parameter.getPageBodyOverhead())); + } + return new PageCryptoContext( + parameter, uncompressedSize, compressedPlaintextSize, encryptedPageBodySize, pageIndex); + } + + public int getUncompressedSize() { + return uncompressedSize; + } + + /** Returns the compressed plaintext size on both encryption and decryption paths. */ + public int getInputSize() { + return compressedPlaintextSize; + } + + public int getCompressedPlaintextSize() { + return compressedPlaintextSize; + } + + public int getEncryptedPageBodySize() { + return encryptedPageBodySize; + } + + public int getPageIndex() { + return pageIndex; + } + + public boolean isFirstPage() { + return pageIndex == 0; + } + + public byte[] getAssociatedData() { + return Arrays.copyOf(associatedData, associatedData.length); + } + + private static void validate( + EncryptParameter parameter, + int uncompressedSize, + int compressedPlaintextSize, + int pageIndex) { + if (parameter == null || !parameter.isTdePageAead()) { + throw new EncryptException(Messages.get("error.encrypt.page_context_invalid_parameter")); + } + if (parameter.getFileCryptoId() == null + || parameter.getFileCryptoId().length != EncryptParameter.FILE_CRYPTO_ID_LENGTH) { + throw new EncryptException(Messages.get("error.encrypt.page_context_invalid_file_id")); + } + if (uncompressedSize < 0 || compressedPlaintextSize < 0 || pageIndex < 0) { + throw new EncryptException( + Messages.format( + "error.encrypt.page_context_invalid_sizes", + uncompressedSize, + compressedPlaintextSize, + pageIndex)); + } + } + + private static byte[] buildAssociatedData( + EncryptParameter parameter, + int uncompressedSize, + int compressedPlaintextSize, + int pageIndex) { + byte[][] components = + new byte[][] { + bytes(parameter.getProviderId()), + bytes(parameter.getProfileId()), + bytes(parameter.getKeyId()), + bytes(parameter.getKeyVersion()), + parameter.getFileCryptoId() + }; + int size = Integer.BYTES * (4 + components.length); + for (byte[] component : components) { + size = Math.addExact(size, component.length); + } + + ByteBuffer buffer = ByteBuffer.allocate(size); + buffer.putInt(ASSOCIATED_DATA_VERSION); + for (byte[] component : components) { + buffer.putInt(component.length); + buffer.put(component); + } + buffer.putInt(pageIndex); + buffer.putInt(uncompressedSize); + buffer.putInt(compressedPlaintextSize); + return buffer.array(); + } + + private static byte[] bytes(String value) { + return value == null ? new byte[0] : value.getBytes(TSFileConfig.STRING_CHARSET); + } +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/file/MetaMarker.java b/java/tsfile/src/main/java/org/apache/tsfile/file/MetaMarker.java index 5cca70594..ce8bf8b37 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/file/MetaMarker.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/file/MetaMarker.java @@ -51,6 +51,9 @@ public class MetaMarker { /** Chunk header marker and this chunk has only one page. */ public static final byte ONLY_ONE_PAGE_CHUNK_HEADER = 5; + /** Optional file encryption header immediately following the TsFile version byte. */ + public static final byte ENCRYPTION_HEADER = 6; + /** Time Chunk header marker and this chunk has more than one page. */ public static final byte TIME_CHUNK_HEADER = (byte) (CHUNK_HEADER | TsFileConstant.TIME_COLUMN_MASK); diff --git a/java/tsfile/src/main/java/org/apache/tsfile/file/header/FileEncryptionHeader.java b/java/tsfile/src/main/java/org/apache/tsfile/file/header/FileEncryptionHeader.java new file mode 100644 index 000000000..1e5590f2f --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/file/header/FileEncryptionHeader.java @@ -0,0 +1,178 @@ +/* + * 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.tsfile.file.header; + +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.exception.encrypt.EncryptException; +import org.apache.tsfile.file.MetaMarker; +import org.apache.tsfile.i18n.Messages; +import org.apache.tsfile.utils.PublicBAOS; +import org.apache.tsfile.utils.ReadWriteForEncodingUtils; +import org.apache.tsfile.utils.ReadWriteIOUtils; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; + +/** + * Versioned file encryption metadata stored before the first TsFile data marker. + * + *

The header contains only public identifiers and a wrapped data key. Plaintext key material and + * runtime implementation class names are never serialized. + */ +public final class FileEncryptionHeader { + + public static final byte VERSION = 1; + public static final int MAX_HEADER_SIZE = 1024 * 1024; + + private FileEncryptionHeader() {} + + public static int serialize(EncryptParameter parameter, OutputStream outputStream) + throws IOException { + validate(parameter); + + PublicBAOS payload = new PublicBAOS(); + int payloadSize = 0; + payloadSize += ReadWriteIOUtils.write(VERSION, payload); + payloadSize += ReadWriteIOUtils.writeVar(parameter.getProviderId(), payload); + payloadSize += ReadWriteIOUtils.writeVar(parameter.getProfileId(), payload); + payloadSize += ReadWriteIOUtils.writeVar(parameter.getKeyId(), payload); + payloadSize += ReadWriteIOUtils.writeVar(parameter.getKeyVersion(), payload); + payloadSize += writeBytes(parameter.getFileCryptoId(), payload); + payloadSize += writeBytes(parameter.getWrappedDataKey(), payload); + + if (payloadSize != payload.size() || payloadSize > MAX_HEADER_SIZE) { + throw new EncryptException( + Messages.format("error.file.encryption_header_invalid_size", payloadSize)); + } + + int size = ReadWriteIOUtils.write(MetaMarker.ENCRYPTION_HEADER, outputStream); + size += ReadWriteForEncodingUtils.writeUnsignedVarInt(payloadSize, outputStream); + payload.writeTo(outputStream); + return size + payloadSize; + } + + /** Deserializes a header after its {@link MetaMarker#ENCRYPTION_HEADER} marker was consumed. */ + public static EncryptParameter deserialize(InputStream inputStream) throws IOException { + int payloadSize = ReadWriteForEncodingUtils.readUnsignedVarInt(inputStream); + if (payloadSize <= 0 || payloadSize > MAX_HEADER_SIZE) { + throw new EncryptException( + Messages.format("error.file.encryption_header_invalid_size", payloadSize)); + } + + byte[] payload = readExactly(inputStream, payloadSize); + ByteBuffer buffer = ByteBuffer.wrap(payload); + byte version = buffer.get(); + if (version != VERSION) { + throw new EncryptException( + Messages.format("error.file.encryption_header_unsupported_version", version)); + } + + EncryptParameter parameter; + try { + parameter = + EncryptParameter.pageAeadBuilder() + .providerId(ReadWriteIOUtils.readVarIntString(buffer)) + .profileId(ReadWriteIOUtils.readVarIntString(buffer)) + .keyId(ReadWriteIOUtils.readVarIntString(buffer)) + .keyVersion(ReadWriteIOUtils.readVarIntString(buffer)) + .fileCryptoId(readBytes(buffer)) + .wrappedDataKey(readBytes(buffer)) + .build(); + } catch (RuntimeException e) { + throw new EncryptException(Messages.get("error.file.encryption_header_malformed"), e); + } + if (buffer.hasRemaining()) { + throw new EncryptException(Messages.get("error.file.encryption_header_trailing_bytes")); + } + validate(parameter); + return parameter; + } + + private static int writeBytes(byte[] bytes, OutputStream outputStream) throws IOException { + int size = ReadWriteForEncodingUtils.writeUnsignedVarInt(bytes.length, outputStream); + outputStream.write(bytes); + return size + bytes.length; + } + + private static byte[] readBytes(ByteBuffer buffer) { + int length = ReadWriteForEncodingUtils.readUnsignedVarInt(buffer); + if (length < 0 || length > MAX_HEADER_SIZE || length > buffer.remaining()) { + throw new EncryptException( + Messages.format("error.file.encryption_header_invalid_component_size", length)); + } + byte[] bytes = new byte[length]; + buffer.get(bytes); + return bytes; + } + + private static byte[] readExactly(InputStream inputStream, int size) throws IOException { + byte[] bytes = new byte[size]; + int offset = 0; + while (offset < size) { + int read = inputStream.read(bytes, offset, size - offset); + if (read < 0) { + throw new EncryptException( + Messages.format("error.file.encryption_header_truncated", size, offset)); + } + if (read == 0) { + int value = inputStream.read(); + if (value < 0) { + throw new EncryptException( + Messages.format("error.file.encryption_header_truncated", size, offset)); + } + bytes[offset++] = (byte) value; + } else { + offset += read; + } + } + return bytes; + } + + private static void validate(EncryptParameter parameter) { + if (parameter == null || !parameter.isTdePageAead()) { + invalidField("pageAead"); + } + requireText(parameter.getProviderId(), "providerId"); + requireText(parameter.getProfileId(), "profileId"); + requireText(parameter.getKeyId(), "keyId"); + requireText(parameter.getKeyVersion(), "keyVersion"); + + byte[] fileCryptoId = parameter.getFileCryptoId(); + if (fileCryptoId == null || fileCryptoId.length != EncryptParameter.FILE_CRYPTO_ID_LENGTH) { + invalidField("fileCryptoId"); + } + byte[] wrappedDataKey = parameter.getWrappedDataKey(); + if (wrappedDataKey == null || wrappedDataKey.length == 0) { + invalidField("wrappedDataKey"); + } + } + + private static void requireText(String value, String field) { + if (value == null || value.trim().isEmpty()) { + invalidField(field); + } + } + + private static void invalidField(String field) { + throw new EncryptException( + Messages.format("error.file.encryption_header_invalid_field", field)); + } +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java index 97dcddc0d..acc5f5c9f 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/TsFileSequenceReader.java @@ -30,6 +30,7 @@ import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.encrypt.IDecryptor; +import org.apache.tsfile.encrypt.PageCryptoContext; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.NotCompatibleTsFileException; import org.apache.tsfile.exception.StopReadTsFileByInterruptException; @@ -40,6 +41,7 @@ import org.apache.tsfile.file.MetaMarker; import org.apache.tsfile.file.header.ChunkGroupHeader; import org.apache.tsfile.file.header.ChunkHeader; +import org.apache.tsfile.file.header.FileEncryptionHeader; import org.apache.tsfile.file.header.PageHeader; import org.apache.tsfile.file.metadata.AbstractAlignedChunkMetadata; import org.apache.tsfile.file.metadata.AbstractAlignedTimeSeriesMetadata; @@ -113,6 +115,9 @@ public class TsFileSequenceReader implements AutoCloseable { + private static final int BASE_FILE_HEADER_SIZE = + TSFileConfig.MAGIC_STRING.getBytes(TSFileConfig.STRING_CHARSET).length + Byte.BYTES; + private static final Logger logger = LoggerFactory.getLogger(TsFileSequenceReader.class); private static final Logger resourceLogger = LoggerFactory.getLogger("FileMonitor"); protected static final TSFileConfig config = TSFileDescriptor.getInstance().getConfig(); @@ -146,6 +151,10 @@ public class TsFileSequenceReader implements AutoCloseable { private EncryptParameter dataEncryptParam = null; + private EncryptParameter fileEncryptionParam; + + private long dataStartOffset = BASE_FILE_HEADER_SIZE; + /** * Create a file reader of the given file. The reader will read the tail of the file to get the * file metadata size.Then the reader will skip the first @@ -167,7 +176,7 @@ public Map countChunksPerChunkGroup() throws IOException { return result; } - int headerLength = TSFileConfig.MAGIC_STRING.getBytes().length + Byte.BYTES; + long headerLength = getDataStartOffset(); if (checkFile.length() < headerLength) { return result; } @@ -314,6 +323,7 @@ public TsFileSequenceReader(String file, boolean loadMetadataSize, LongConsumer try { loadFileVersion(ioSizeRecorder); + loadFileEncryptionHeader(ioSizeRecorder); if (loadMetadataSize) { loadMetadataSize(ioSizeRecorder); } @@ -378,6 +388,7 @@ public TsFileSequenceReader(TsFileInput input, boolean loadMetadataSize) throws this.tsFileInput = input; this.file = input.getFilePath(); try { + loadFileEncryptionHeader(null); if (loadMetadataSize) { // NOTE no autoRepair here loadMetadataSize(); } @@ -440,6 +451,38 @@ private void loadFileVersion(LongConsumer ioSizeRecorder) throws IOException { } } + private void loadFileEncryptionHeader(LongConsumer ioSizeRecorder) throws IOException { + long savedPosition = tsFileInput.position(); + dataStartOffset = BASE_FILE_HEADER_SIZE; + fileEncryptionParam = null; + try { + if (tsFileInput.size() <= BASE_FILE_HEADER_SIZE) { + return; + } + ByteBuffer marker = ByteBuffer.allocate(Byte.BYTES); + if (tsFileInput.read(marker, BASE_FILE_HEADER_SIZE) != Byte.BYTES) { + return; + } + marker.flip(); + if (marker.get() != MetaMarker.ENCRYPTION_HEADER) { + return; + } + + tsFileInput.position(BASE_FILE_HEADER_SIZE + Byte.BYTES); + fileEncryptionParam = FileEncryptionHeader.deserialize(tsFileInput.wrapAsInputStream()); + dataStartOffset = tsFileInput.position(); + if (ioSizeRecorder != null) { + ioSizeRecorder.accept(dataStartOffset - BASE_FILE_HEADER_SIZE); + } + } finally { + tsFileInput.position(savedPosition); + } + } + + public long getDataStartOffset() { + return dataStartOffset; + } + private void configDeserializer() { if (fileVersion == TSFileConfig.VERSION_NUMBER_V3) { deserializeConfig = CompatibilityUtils.v3DeserializeConfig; @@ -642,6 +685,9 @@ public EncryptParameter getEncryptParam() throws IOException { public EncryptParameter getEncryptParam(LongConsumer ioSizeRecorder) throws IOException { if (dataEncryptParam != null) { return dataEncryptParam; + } else if (fileEncryptionParam != null) { + dataEncryptParam = fileEncryptionParam; + return dataEncryptParam; } else { if (fileMetadataSize != 0) { readFileMetadata(ioSizeRecorder); @@ -2205,17 +2251,48 @@ public ByteBuffer readCompressedPage(PageHeader header) throws IOException { } public ByteBuffer readPage(PageHeader header, CompressionType type) throws IOException { + return readPage(header, type, 0); + } + + public ByteBuffer readPage(PageHeader header, CompressionType type, int pageIndex) + throws IOException { ByteBuffer buffer = readData(-1, header.getCompressedSize()); - IDecryptor decryptor = IDecryptor.getDecryptor(getEncryptParam()); + EncryptParameter encryptParameter = getEncryptParam(); + IDecryptor decryptor = IDecryptor.getDecryptor(encryptParameter); if (header.getUncompressedSize() == 0) { return buffer; } - ByteBuffer finalBuffer = decrypt(decryptor, buffer); + ByteBuffer finalBuffer = decrypt(decryptor, encryptParameter, buffer, header, pageIndex); finalBuffer = uncompress(type, finalBuffer, header.getUncompressedSize()); return finalBuffer; } - private static ByteBuffer decrypt(IDecryptor decryptor, ByteBuffer buffer) { + private static ByteBuffer decrypt( + IDecryptor decryptor, + EncryptParameter encryptParameter, + ByteBuffer buffer, + PageHeader header, + int pageIndex) + throws IOException { + if (encryptParameter != null && encryptParameter.isTdePageAead()) { + PageCryptoContext pageCryptoContext = + PageCryptoContext.forDecryption( + encryptParameter, header.getUncompressedSize(), buffer.remaining(), pageIndex); + byte[] plaintext = + decryptor.decryptPage( + buffer.array(), + buffer.arrayOffset() + buffer.position(), + buffer.remaining(), + pageCryptoContext); + if (plaintext.length != pageCryptoContext.getCompressedPlaintextSize()) { + throw new IOException( + Messages.format( + "error.encrypt.page_plaintext_size_mismatch", + pageCryptoContext.getCompressedPlaintextSize(), + plaintext.length)); + } + return ByteBuffer.wrap(plaintext); + } if (decryptor == null || decryptor.getEncryptionType() == EncryptionType.UNENCRYPTED) { return buffer; } @@ -2266,7 +2343,13 @@ public void close() throws IOException { if (resourceLogger.isDebugEnabled()) { resourceLogger.debug("{} reader is closed.", file); } - this.tsFileInput.close(); + try { + this.tsFileInput.close(); + } finally { + if (dataEncryptParam != null && dataEncryptParam.isTdePageAead()) { + dataEncryptParam.close(); + } + } } public String getFileName() { @@ -2407,7 +2490,7 @@ public long selfCheck( // ChunkMetadata of current ChunkGroup List chunkMetadataList = new ArrayList<>(); - int headerLength = TSFileConfig.MAGIC_STRING.getBytes().length + Byte.BYTES; + long headerLength = getDataStartOffset(); if (fileSize < headerLength) { return TsFileCheckStatus.INCOMPATIBLE_FILE; } @@ -2722,7 +2805,7 @@ public long selfCheckWithInfo( long fileSize = checkFile.length(); logger.info(Messages.get("log.read.sequence_reader_file_length"), fileSize); - int headerLength = TSFileConfig.MAGIC_STRING.getBytes().length + Byte.BYTES; + long headerLength = getDataStartOffset(); if (fileSize < headerLength) { return TsFileCheckStatus.INCOMPATIBLE_FILE; } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/UnClosedTsFileReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/UnClosedTsFileReader.java index 230a2e164..fcf11c9d6 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/UnClosedTsFileReader.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/UnClosedTsFileReader.java @@ -21,7 +21,6 @@ import org.apache.tsfile.common.conf.TSFileDescriptor; import org.apache.tsfile.encrypt.EncryptParameter; -import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.exception.NotImplementedException; import org.apache.tsfile.file.metadata.TsFileMetadata; @@ -46,7 +45,7 @@ public UnClosedTsFileReader(String file, LongConsumer ioSizeRecorder) throws IOE public UnClosedTsFileReader( String file, EncryptParameter encryptParam, LongConsumer ioSizeRecorder) throws IOException { super(file, false, ioSizeRecorder, encryptParam); - this.dataEncryptParam = EncryptUtils.getEncryptParameter(encryptParam); + this.dataEncryptParam = super.getEncryptParam(); } /** unclosed file has no tail magic data. */ diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/common/Chunk.java b/java/tsfile/src/main/java/org/apache/tsfile/read/common/Chunk.java index bf83a4539..47f14bc8f 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/common/Chunk.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/common/Chunk.java @@ -124,6 +124,10 @@ public void setDeleteIntervalList(List list) { } public void mergeChunkByAppendPage(Chunk chunk) throws IOException { + if ((encryptParam != null && encryptParam.isTdePageAead()) + || (chunk.encryptParam != null && chunk.encryptParam.isTdePageAead())) { + throw new IOException(Messages.get("error.chunk.merge_page_aead_unsupported")); + } int dataSize = 0; // from where the page data of the merged chunk starts, if -1, it means the merged chunk has // more than one page diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/AbstractAlignedChunkReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/AbstractAlignedChunkReader.java index acc9789e4..0d4de4d34 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/AbstractAlignedChunkReader.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/AbstractAlignedChunkReader.java @@ -90,12 +90,15 @@ private void initAllPageReaders( List> valueChunkStatisticsList) throws IOException { // construct next satisfied page header + int pageIndex = 0; while (timeChunkDataBuffer.remaining() > 0) { // deserialize PageHeader from chunkDataBuffer AbstractAlignedPageReader alignedPageReader = isSinglePageChunk() - ? deserializeFromSinglePageChunk(timeChunkStatistics, valueChunkStatisticsList) - : deserializeFromMultiPageChunk(); + ? deserializeFromSinglePageChunk( + timeChunkStatistics, valueChunkStatisticsList, pageIndex) + : deserializeFromMultiPageChunk(pageIndex); + pageIndex++; if (alignedPageReader != null) { pageReaderList.add(alignedPageReader); } @@ -108,7 +111,8 @@ private boolean isSinglePageChunk() { private AbstractAlignedPageReader deserializeFromSinglePageChunk( Statistics timeChunkStatistics, - List> valueChunkStatisticsList) + List> valueChunkStatisticsList, + int pageIndex) throws IOException { PageHeader timePageHeader = PageHeader.deserializeFrom(timeChunkDataBuffer, timeChunkStatistics); @@ -132,12 +136,13 @@ private AbstractAlignedPageReader deserializeFromSinglePageChunk( skipCurrentPage(timePageHeader, valuePageHeaderList); return null; } - return constructAlignedPageReader(timePageHeader, valuePageHeaderList); + return constructAlignedPageReader(timePageHeader, valuePageHeaderList, pageIndex); } abstract boolean needSkipForSinglePageChunk(boolean isAllNull, PageHeader timePageHeader); - private AbstractAlignedPageReader deserializeFromMultiPageChunk() throws IOException { + private AbstractAlignedPageReader deserializeFromMultiPageChunk(int pageIndex) + throws IOException { PageHeader timePageHeader = PageHeader.deserializeFrom(timeChunkDataBuffer, timeChunkHeader.getDataType()); List valuePageHeaderList = new ArrayList<>(); @@ -158,7 +163,7 @@ private AbstractAlignedPageReader deserializeFromMultiPageChunk() throws IOExcep skipCurrentPage(timePageHeader, valuePageHeaderList); return null; } - return constructAlignedPageReader(timePageHeader, valuePageHeaderList); + return constructAlignedPageReader(timePageHeader, valuePageHeaderList, pageIndex); } abstract boolean needSkipForMultiPageChunk(boolean isAllNull, PageHeader timePageHeader); @@ -193,11 +198,17 @@ private void skipCurrentPage(PageHeader timePageHeader, List valuePa } private AbstractAlignedPageReader constructAlignedPageReader( - PageHeader timePageHeader, List rawValuePageHeaderList) throws IOException { + PageHeader timePageHeader, List rawValuePageHeaderList, int pageIndex) + throws IOException { IDecryptor decrytor = IDecryptor.getDecryptor(encryptParam); ByteBuffer timePageData = ChunkReader.deserializePageData( - timePageHeader, timeChunkDataBuffer, timeChunkHeader, decrytor); + timePageHeader, + timeChunkDataBuffer, + timeChunkHeader, + decrytor, + encryptParam, + pageIndex); List valuePageHeaderList = new ArrayList<>(); LazyLoadPageData[] lazyLoadPageDataArray = new LazyLoadPageData[rawValuePageHeaderList.size()]; @@ -238,7 +249,8 @@ private AbstractAlignedPageReader constructAlignedPageReader( valueChunkDataBufferList.get(i).array(), currentPagePosition, IUnCompressor.getUnCompressor(valueChunkHeader.getCompressionType()), - encryptParam); + encryptParam, + pageIndex); valueDataTypeList.add(valueChunkHeader.getDataType()); valueDecoderList.add(valueChunkHeader.calculateDecoderForNonTimeChunk()); isAllNull = false; diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/ChunkReader.java b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/ChunkReader.java index b555a25e1..cbb0892bc 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/ChunkReader.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/chunk/ChunkReader.java @@ -22,6 +22,7 @@ import org.apache.tsfile.compress.IUnCompressor; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.IDecryptor; +import org.apache.tsfile.encrypt.PageCryptoContext; import org.apache.tsfile.file.MetaMarker; import org.apache.tsfile.file.header.ChunkHeader; import org.apache.tsfile.file.header.PageHeader; @@ -81,7 +82,9 @@ public ChunkReader(Chunk chunk, long readStopTime) { private void initAllPageReaders(Statistics chunkStatistic) { // construct next satisfied page header + int pageIndex = 0; while (chunkDataBuffer.remaining() > 0) { + int currentPageIndex = pageIndex++; // deserialize a PageHeader from chunkDataBuffer PageHeader pageHeader; if (((byte) (chunkHeader.getChunkType() & 0x3F)) == MetaMarker.ONLY_ONE_PAGE_CHUNK_HEADER) { @@ -100,7 +103,7 @@ private void initAllPageReaders(Statistics chunkStatisti if (pageDeleted(pageHeader)) { skipCurrentPage(pageHeader); } else { - pageReaderList.add(constructPageReader(pageHeader)); + pageReaderList.add(constructPageReader(pageHeader, currentPageIndex)); } } } @@ -141,7 +144,7 @@ private void skipCurrentPage(PageHeader pageHeader) { chunkDataBuffer.position(chunkDataBuffer.position() + pageHeader.getCompressedSize()); } - private PageReader constructPageReader(PageHeader pageHeader) { + private PageReader constructPageReader(PageHeader pageHeader, int pageIndex) { IUnCompressor unCompressor = IUnCompressor.getUnCompressor(chunkHeader.getCompressionType()); // record the current position of chunkDataBuffer, use this to get the page data in PageReader // through directly accessing the buffer array @@ -151,7 +154,11 @@ private PageReader constructPageReader(PageHeader pageHeader) { new PageReader( pageHeader, new LazyLoadPageData( - chunkDataBuffer.array(), currentPagePosition, unCompressor, encryptParam), + chunkDataBuffer.array(), + currentPagePosition, + unCompressor, + encryptParam, + pageIndex), chunkHeader.getDataType(), chunkHeader.calculateDecoderForNonTimeChunk(), getConfiguredTimeDecoder(), @@ -221,7 +228,53 @@ public static ByteBuffer decryptAndUncompressPageData( compressedPageData.arrayOffset() + compressedPageData.position(), compressedPageBodyLength); unCompressor.uncompress( - decryptedPageData, 0, compressedPageBodyLength, uncompressedPageData, 0); + decryptedPageData, 0, decryptedPageData.length, uncompressedPageData, 0); + } catch (Exception e) { + throw new IOException( + Messages.format( + "error.read.uncompress_error_with_header", + pageHeader.getUncompressedSize(), + pageHeader.getCompressedSize(), + pageHeader, + e.getMessage()), + e); + } + compressedPageData.position(compressedPageData.position() + compressedPageBodyLength); + return ByteBuffer.wrap(uncompressedPageData); + } + + public static ByteBuffer decryptAndUncompressPageData( + PageHeader pageHeader, + IUnCompressor unCompressor, + ByteBuffer compressedPageData, + IDecryptor decryptor, + EncryptParameter encryptParameter, + int pageIndex) + throws IOException { + int compressedPageBodyLength = pageHeader.getCompressedSize(); + byte[] uncompressedPageData = new byte[pageHeader.getUncompressedSize()]; + try { + PageCryptoContext pageCryptoContext = + PageCryptoContext.forDecryption( + encryptParameter, + pageHeader.getUncompressedSize(), + compressedPageBodyLength, + pageIndex); + byte[] decryptedPageData = + decryptor.decryptPage( + compressedPageData.array(), + compressedPageData.arrayOffset() + compressedPageData.position(), + compressedPageBodyLength, + pageCryptoContext); + if (decryptedPageData.length != pageCryptoContext.getCompressedPlaintextSize()) { + throw new IOException( + Messages.format( + "error.encrypt.page_plaintext_size_mismatch", + pageCryptoContext.getCompressedPlaintextSize(), + decryptedPageData.length)); + } + unCompressor.uncompress( + decryptedPageData, 0, decryptedPageData.length, uncompressedPageData, 0); } catch (Exception e) { throw new IOException( Messages.format( @@ -243,6 +296,25 @@ public static ByteBuffer deserializePageData( ByteBuffer compressedPageBody = readCompressedPageData(pageHeader, chunkBuffer); if (decryptor == null || decryptor.getEncryptionType() == EncryptionType.UNENCRYPTED) { return uncompressPageData(pageHeader, unCompressor, compressedPageBody); + } + return decryptAndUncompressPageData(pageHeader, unCompressor, compressedPageBody, decryptor); + } + + public static ByteBuffer deserializePageData( + PageHeader pageHeader, + ByteBuffer chunkBuffer, + ChunkHeader chunkHeader, + IDecryptor decryptor, + EncryptParameter encryptParameter, + int pageIndex) + throws IOException { + IUnCompressor unCompressor = IUnCompressor.getUnCompressor(chunkHeader.getCompressionType()); + ByteBuffer compressedPageBody = readCompressedPageData(pageHeader, chunkBuffer); + if (encryptParameter != null && encryptParameter.isTdePageAead()) { + return decryptAndUncompressPageData( + pageHeader, unCompressor, compressedPageBody, decryptor, encryptParameter, pageIndex); + } else if (decryptor == null || decryptor.getEncryptionType() == EncryptionType.UNENCRYPTED) { + return uncompressPageData(pageHeader, unCompressor, compressedPageBody); } else { return decryptAndUncompressPageData(pageHeader, unCompressor, compressedPageBody, decryptor); } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/LazyLoadPageData.java b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/LazyLoadPageData.java index 7b01a67f5..a74c9bc8b 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/LazyLoadPageData.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/read/reader/page/LazyLoadPageData.java @@ -23,7 +23,9 @@ import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; import org.apache.tsfile.encrypt.IDecryptor; +import org.apache.tsfile.encrypt.PageCryptoContext; import org.apache.tsfile.file.header.PageHeader; +import org.apache.tsfile.file.metadata.enums.EncryptionType; import org.apache.tsfile.i18n.Messages; import java.io.IOException; @@ -39,11 +41,14 @@ public class LazyLoadPageData { private final EncryptParameter encryptParam; + private final int pageIndex; + public LazyLoadPageData(byte[] data, int offset, IUnCompressor unCompressor) { this.chunkData = data; this.pageDataOffset = offset; this.unCompressor = unCompressor; this.encryptParam = EncryptUtils.getEncryptParameter(); + this.pageIndex = -1; } public LazyLoadPageData( @@ -52,17 +57,51 @@ public LazyLoadPageData( this.pageDataOffset = offset; this.unCompressor = unCompressor; this.encryptParam = encryptParam; + this.pageIndex = -1; + } + + public LazyLoadPageData( + byte[] data, + int offset, + IUnCompressor unCompressor, + EncryptParameter encryptParam, + int pageIndex) { + this.chunkData = data; + this.pageDataOffset = offset; + this.unCompressor = unCompressor; + this.encryptParam = encryptParam; + this.pageIndex = pageIndex; } public ByteBuffer uncompressPageData(PageHeader pageHeader) throws IOException { int compressedPageBodyLength = pageHeader.getCompressedSize(); byte[] uncompressedPageData = new byte[pageHeader.getUncompressedSize()]; IDecryptor decryptor = IDecryptor.getDecryptor(encryptParam); - byte[] decryptedPageData = - decryptor.decrypt(chunkData, pageDataOffset, compressedPageBodyLength); + byte[] decryptedPageData; + if (encryptParam != null && encryptParam.isTdePageAead()) { + PageCryptoContext pageCryptoContext = + PageCryptoContext.forDecryption( + encryptParam, pageHeader.getUncompressedSize(), compressedPageBodyLength, pageIndex); + decryptedPageData = + decryptor.decryptPage( + chunkData, pageDataOffset, compressedPageBodyLength, pageCryptoContext); + if (decryptedPageData.length != pageCryptoContext.getCompressedPlaintextSize()) { + throw new IOException( + Messages.format( + "error.encrypt.page_plaintext_size_mismatch", + pageCryptoContext.getCompressedPlaintextSize(), + decryptedPageData.length)); + } + } else if (decryptor.getEncryptionType() == EncryptionType.UNENCRYPTED) { + decryptedPageData = + java.util.Arrays.copyOfRange( + chunkData, pageDataOffset, pageDataOffset + compressedPageBodyLength); + } else { + decryptedPageData = decryptor.decrypt(chunkData, pageDataOffset, compressedPageBodyLength); + } try { unCompressor.uncompress( - decryptedPageData, 0, compressedPageBodyLength, uncompressedPageData, 0); + decryptedPageData, 0, decryptedPageData.length, uncompressedPageData, 0); } catch (Exception e) { throw new IOException( Messages.format( diff --git a/java/tsfile/src/main/java/org/apache/tsfile/utils/TsFileSketchTool.java b/java/tsfile/src/main/java/org/apache/tsfile/utils/TsFileSketchTool.java index a20c0e888..a04c13aa5 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/utils/TsFileSketchTool.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/utils/TsFileSketchTool.java @@ -311,8 +311,7 @@ private void printFileInfo() { private void printChunk(List allChunkGroupMetadata) { try { - long nextChunkGroupHeaderPos = - (long) TSFileConfig.MAGIC_STRING.getBytes().length + Byte.BYTES; + long nextChunkGroupHeaderPos = reader.getDataStartOffset(); // ChunkGroup begins for (ChunkGroupMetadata chunkGroupMetadata : allChunkGroupMetadata) { printlnBoth( diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/TsFileWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/TsFileWriter.java index 5521a4f72..b931c3b73 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/TsFileWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/TsFileWriter.java @@ -118,7 +118,7 @@ public TsFileWriter(File file) throws IOException { public TsFileWriter(File file, EncryptParameter firstEncryptParam) throws IOException { this( - new TsFileIOWriter(file), + new TsFileIOWriter(file, copyEncryptParameter(firstEncryptParam)), new Schema(), TSFileDescriptor.getInstance().getConfig(), firstEncryptParam); @@ -151,7 +151,7 @@ public TsFileWriter(File file, Schema schema) throws IOException { public TsFileWriter(File file, Schema schema, EncryptParameter firstEncryptParam) throws IOException { this( - new TsFileIOWriter(file), + new TsFileIOWriter(file, copyEncryptParameter(firstEncryptParam)), schema, TSFileDescriptor.getInstance().getConfig(), firstEncryptParam); @@ -170,7 +170,7 @@ public TsFileWriter(TsFileOutput output, Schema schema) throws IOException { public TsFileWriter(TsFileOutput output, Schema schema, EncryptParameter firstEncryptParam) throws IOException { this( - new TsFileIOWriter(output), + new TsFileIOWriter(output, copyEncryptParameter(firstEncryptParam)), schema, TSFileDescriptor.getInstance().getConfig(), firstEncryptParam); @@ -190,7 +190,11 @@ public TsFileWriter(File file, Schema schema, TSFileConfig conf) throws IOExcept public TsFileWriter( File file, Schema schema, TSFileConfig conf, EncryptParameter firstEncryptParam) throws IOException { - this(new TsFileIOWriter(file), schema, conf, firstEncryptParam); + this( + new TsFileIOWriter(file, conf, copyEncryptParameter(firstEncryptParam)), + schema, + conf, + firstEncryptParam); } /** @@ -231,23 +235,39 @@ protected TsFileWriter( if (this.pageSize >= chunkGroupSizeThreshold) { LOG.warn(Messages.get("log.write.page_size_warn"), pageSize, chunkGroupSizeThreshold); } - this.secondEncryptParam = EncryptUtils.getEncryptParameter(firstEncryptParam); - String encryptLevel; - if (firstEncryptParam != null - && !Objects.equals(firstEncryptParam.getType(), "UNENCRYPTED") - && !Objects.equals(firstEncryptParam.getType(), "org.apache.tsfile.encrypt.UNENCRYPTED")) { - encryptLevel = "2"; - String str = - EncryptUtils.getKeyStr( - IEncryptor.getEncryptor(firstEncryptParam.getType(), firstEncryptParam.getKey()) - .encrypt(secondEncryptParam.getKey())); - fileWriter.setEncryptParam(encryptLevel, secondEncryptParam.getType(), str); + EncryptParameter fileEncryptParameter = fileWriter.getEncryptParameter(); + EncryptParameter effectiveEncryptParameter = + fileEncryptParameter != null && fileEncryptParameter.isTdePageAead() + ? fileEncryptParameter + : firstEncryptParam; + if (effectiveEncryptParameter != null && effectiveEncryptParameter.isTdePageAead()) { + this.secondEncryptParam = effectiveEncryptParameter; + fileWriter.setEncryptParam(effectiveEncryptParameter); + fileWriter.writeEncryptionHeaderIfNecessary(); } else { - encryptLevel = "0"; - fileWriter.setEncryptParam(encryptLevel, "org.apache.tsfile.encrypt.UNENCRYPTED", ""); + this.secondEncryptParam = EncryptUtils.getEncryptParameter(firstEncryptParam); + String encryptLevel; + if (firstEncryptParam != null + && !Objects.equals(firstEncryptParam.getType(), "UNENCRYPTED") + && !Objects.equals( + firstEncryptParam.getType(), "org.apache.tsfile.encrypt.UNENCRYPTED")) { + encryptLevel = "2"; + String str = + EncryptUtils.getKeyStr( + IEncryptor.getEncryptor(firstEncryptParam.getType(), firstEncryptParam.getKey()) + .encrypt(secondEncryptParam.getKey())); + fileWriter.setEncryptParam(encryptLevel, secondEncryptParam.getType(), str); + } else { + encryptLevel = "0"; + fileWriter.setEncryptParam(encryptLevel, "org.apache.tsfile.encrypt.UNENCRYPTED", ""); + } } } + private static EncryptParameter copyEncryptParameter(EncryptParameter parameter) { + return parameter == null ? null : parameter.copy(); + } + public void setChunkGroupSizeThreshold(long chunkGroupSizeThreshold) { this.chunkGroupSizeThreshold = chunkGroupSizeThreshold; } @@ -729,8 +749,14 @@ private void reset() { @TsFileApi public void close() throws IOException { LOG.info(Messages.get("log.write.close_file")); - flush(); - fileWriter.endFile(); + try { + flush(); + fileWriter.endFile(); + } finally { + if (secondEncryptParam != null && secondEncryptParam.isTdePageAead()) { + secondEncryptParam.close(); + } + } } /** diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java index 7d7912b14..1e56b5e68 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ChunkWriterImpl.java @@ -322,17 +322,18 @@ private void writePageToPageBuffer() { try { if (numOfPages == 0) { // record the firstPageStatistics this.firstPageStatistics = pageWriter.getStatistics(); - this.sizeWithoutStatistic = pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true); + this.sizeWithoutStatistic = + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true, numOfPages); } else if (numOfPages == 1) { // put the firstPageStatistics into pageBuffer byte[] b = pageBuffer.toByteArray(); pageBuffer.reset(); pageBuffer.write(b, 0, this.sizeWithoutStatistic); firstPageStatistics.serialize(pageBuffer); pageBuffer.write(b, this.sizeWithoutStatistic, b.length - this.sizeWithoutStatistic); - pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false); + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false, numOfPages); firstPageStatistics = null; } else { - pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false); + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false, numOfPages); } // update statistics of this chunk diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TimeChunkWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TimeChunkWriter.java index 392a0cf45..d19a8e566 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TimeChunkWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/TimeChunkWriter.java @@ -184,17 +184,18 @@ public void writePageToPageBuffer() { try { if (numOfPages == 0) { // record the firstPageStatistics this.firstPageStatistics = pageWriter.getStatistics(); - this.sizeWithoutStatistic = pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true); + this.sizeWithoutStatistic = + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true, numOfPages); } else if (numOfPages == 1) { // put the firstPageStatistics into pageBuffer byte[] b = pageBuffer.toByteArray(); pageBuffer.reset(); pageBuffer.write(b, 0, this.sizeWithoutStatistic); firstPageStatistics.serialize(pageBuffer); pageBuffer.write(b, this.sizeWithoutStatistic, b.length - this.sizeWithoutStatistic); - pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false); + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false, numOfPages); firstPageStatistics = null; } else { - pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false); + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false, numOfPages); } // update statistics of this chunk diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java index 9ade788c1..97c07441b 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/chunk/ValueChunkWriter.java @@ -209,7 +209,8 @@ public void writePageToPageBuffer() { // record the firstPageStatistics if it is not empty page this.firstPageStatistics = pageWriter.getStatistics(); } - this.sizeWithoutStatistic = pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true); + this.sizeWithoutStatistic = + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, true, numOfPages); } else if (numOfPages == 1) { // put the firstPageStatistics into pageBuffer if (firstPageStatistics != null) { // Consider previous page is an empty page byte[] b = pageBuffer.toByteArray(); @@ -218,10 +219,10 @@ public void writePageToPageBuffer() { firstPageStatistics.serialize(pageBuffer); pageBuffer.write(b, this.sizeWithoutStatistic, b.length - this.sizeWithoutStatistic); } - pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false); + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false, numOfPages); firstPageStatistics = null; } else { - pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false); + pageWriter.writePageHeaderAndDataIntoBuff(pageBuffer, false, numOfPages); } // update statistics of this chunk diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/EncodedPageBody.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/EncodedPageBody.java new file mode 100644 index 000000000..6089de625 --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/EncodedPageBody.java @@ -0,0 +1,48 @@ +/* + * 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.tsfile.write.page; + +class EncodedPageBody { + + private final byte[] data; + private final int offset; + private final int size; + + EncodedPageBody(byte[] data) { + this(data, 0, data.length); + } + + EncodedPageBody(byte[] data, int offset, int size) { + this.data = data; + this.offset = offset; + this.size = size; + } + + byte[] getData() { + return data; + } + + int getOffset() { + return offset; + } + + int size() { + return size; + } +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageBodyEncoder.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageBodyEncoder.java new file mode 100644 index 000000000..c464acae9 --- /dev/null +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageBodyEncoder.java @@ -0,0 +1,83 @@ +/* + * 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.tsfile.write.page; + +import org.apache.tsfile.compress.ICompressor; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.IEncryptor; +import org.apache.tsfile.encrypt.PageCryptoContext; +import org.apache.tsfile.exception.encrypt.EncryptException; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.EncryptionType; +import org.apache.tsfile.i18n.Messages; + +import java.io.IOException; +import java.nio.ByteBuffer; + +class PageBodyEncoder { + + private PageBodyEncoder() {} + + static EncodedPageBody encode( + ByteBuffer pageData, + int uncompressedSize, + ICompressor compressor, + EncryptParameter encryptParameter, + int pageIndex) + throws IOException { + byte[] plaintext; + int plaintextOffset = 0; + int compressedSize; + + if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) { + plaintext = pageData.array(); + plaintextOffset = pageData.position(); + compressedSize = uncompressedSize; + } else if (compressor.getType().equals(CompressionType.GZIP)) { + plaintext = compressor.compress(pageData.array(), pageData.position(), uncompressedSize); + compressedSize = plaintext.length; + } else { + plaintext = new byte[compressor.getMaxBytesForCompression(uncompressedSize)]; + compressedSize = + compressor.compress(pageData.array(), pageData.position(), uncompressedSize, plaintext); + } + + IEncryptor encryptor = IEncryptor.getEncryptor(encryptParameter); + byte[] encryptedPageBody; + if (encryptParameter != null && encryptParameter.isTdePageAead()) { + PageCryptoContext pageCryptoContext = + PageCryptoContext.forEncryption( + encryptParameter, uncompressedSize, compressedSize, pageIndex); + encryptedPageBody = + encryptor.encryptPage(plaintext, plaintextOffset, compressedSize, pageCryptoContext); + if (encryptedPageBody.length != pageCryptoContext.getEncryptedPageBodySize()) { + throw new EncryptException( + Messages.format( + "error.encrypt.page_output_size_mismatch", + pageCryptoContext.getEncryptedPageBodySize(), + encryptedPageBody.length)); + } + } else if (encryptor.getEncryptionType() == EncryptionType.UNENCRYPTED) { + return new EncodedPageBody(plaintext, plaintextOffset, compressedSize); + } else { + encryptedPageBody = encryptor.encrypt(plaintext, plaintextOffset, compressedSize); + } + return new EncodedPageBody(encryptedPageBody); + } +} diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java index d479ad8d4..d714284ff 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/PageWriter.java @@ -22,10 +22,7 @@ import org.apache.tsfile.encoding.encoder.Encoder; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; -import org.apache.tsfile.encrypt.IEncryptor; import org.apache.tsfile.enums.TSDataType; -import org.apache.tsfile.file.metadata.enums.CompressionType; -import org.apache.tsfile.file.metadata.enums.EncryptionType; import org.apache.tsfile.file.metadata.statistics.Statistics; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.PublicBAOS; @@ -38,8 +35,6 @@ import java.io.IOException; import java.io.Serializable; import java.nio.ByteBuffer; -import java.nio.channels.Channels; -import java.nio.channels.WritableByteChannel; /** * This writer is used to write time-value into a page. It consists of a time encoder, a value @@ -231,28 +226,21 @@ public ByteBuffer getUncompressedBytes() throws IOException { /** write the page header and data into the PageWriter's output stream. */ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) throws IOException { + return writePageHeaderAndDataIntoBuff(pageBuffer, first, first ? 0 : -1); + } + + /** write the page header and data into the PageWriter's output stream. */ + public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first, int pageIndex) + throws IOException { if (statistics.getCount() == 0) { return 0; } ByteBuffer pageData = getUncompressedBytes(); int uncompressedSize = pageData.remaining(); - int compressedSize; - byte[] compressedBytes = null; - - if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) { - compressedSize = uncompressedSize; - } else if (compressor.getType().equals(CompressionType.GZIP)) { - compressedBytes = - compressor.compress(pageData.array(), pageData.position(), uncompressedSize); - compressedSize = compressedBytes.length; - } else { - compressedBytes = new byte[compressor.getMaxBytesForCompression(uncompressedSize)]; - // data is never a directByteBuffer now, so we can use data.array() - compressedSize = - compressor.compress( - pageData.array(), pageData.position(), uncompressedSize, compressedBytes); - } + EncodedPageBody pageBody = + PageBodyEncoder.encode(pageData, uncompressedSize, compressor, encryptParam, pageIndex); + int pageBodySize = pageBody.size(); // write the page header to IOWriter int sizeWithoutStatistic = 0; @@ -260,41 +248,16 @@ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) sizeWithoutStatistic += ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer); sizeWithoutStatistic += - ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer); + ReadWriteForEncodingUtils.writeUnsignedVarInt(pageBodySize, pageBuffer); } else { ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer); - ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer); + ReadWriteForEncodingUtils.writeUnsignedVarInt(pageBodySize, pageBuffer); statistics.serialize(pageBuffer); } - IEncryptor encryptor = IEncryptor.getEncryptor(encryptParam); - // write page content to temp PBAOS logger.trace("start to flush a page data into buffer, buffer position {} ", pageBuffer.size()); - if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) { - if (encryptor.getEncryptionType().equals(EncryptionType.UNENCRYPTED)) { - try (WritableByteChannel channel = Channels.newChannel(pageBuffer)) { - channel.write(pageData); - } - } else { - byte[] encryptedBytes = null; - encryptedBytes = encryptor.encrypt(pageData.array(), pageData.position(), uncompressedSize); - // data is never a directByteBuffer now, so we can use data.array() - int encryptedSize = encryptedBytes.length; - pageBuffer.write(encryptedBytes, 0, encryptedSize); - } - - } else { - if (encryptor.getEncryptionType().equals(EncryptionType.UNENCRYPTED)) { - pageBuffer.write(compressedBytes, 0, compressedSize); - } else { - byte[] encryptedBytes = null; - encryptedBytes = encryptor.encrypt(compressedBytes, 0, compressedSize); - // data is never a directByteBuffer now, so we can use data.array() - int encryptedSize = encryptedBytes.length; - pageBuffer.write(encryptedBytes, 0, encryptedSize); - } - } + pageBuffer.write(pageBody.getData(), pageBody.getOffset(), pageBodySize); logger.trace("start to flush a page data into buffer, buffer position {} ", pageBuffer.size()); return sizeWithoutStatistic; } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java index e09e66cdf..ce8feb865 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/TimePageWriter.java @@ -22,9 +22,6 @@ import org.apache.tsfile.encoding.encoder.Encoder; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; -import org.apache.tsfile.encrypt.IEncryptor; -import org.apache.tsfile.file.metadata.enums.CompressionType; -import org.apache.tsfile.file.metadata.enums.EncryptionType; import org.apache.tsfile.file.metadata.statistics.TimeStatistics; import org.apache.tsfile.utils.PublicBAOS; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; @@ -34,8 +31,6 @@ import java.io.IOException; import java.nio.ByteBuffer; -import java.nio.channels.Channels; -import java.nio.channels.WritableByteChannel; /** * This writer is used to write time into a page. It consists of a time encoder and respective @@ -114,28 +109,21 @@ public ByteBuffer getUncompressedBytes() throws IOException { /** write the page header and data into the PageWriter's output stream. */ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) throws IOException { + return writePageHeaderAndDataIntoBuff(pageBuffer, first, first ? 0 : -1); + } + + /** write the page header and data into the PageWriter's output stream. */ + public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first, int pageIndex) + throws IOException { if (statistics.getCount() == 0) { return 0; } ByteBuffer pageData = getUncompressedBytes(); int uncompressedSize = pageData.remaining(); - int compressedSize; - byte[] compressedBytes = null; - - if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) { - compressedSize = uncompressedSize; - } else if (compressor.getType().equals(CompressionType.GZIP)) { - compressedBytes = - compressor.compress(pageData.array(), pageData.position(), uncompressedSize); - compressedSize = compressedBytes.length; - } else { - compressedBytes = new byte[compressor.getMaxBytesForCompression(uncompressedSize)]; - // data is never a directByteBuffer now, so we can use data.array() - compressedSize = - compressor.compress( - pageData.array(), pageData.position(), uncompressedSize, compressedBytes); - } + EncodedPageBody pageBody = + PageBodyEncoder.encode(pageData, uncompressedSize, compressor, encryptParam, pageIndex); + int pageBodySize = pageBody.size(); // write the page header to IOWriter int sizeWithoutStatistic = 0; @@ -143,41 +131,17 @@ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) sizeWithoutStatistic += ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer); sizeWithoutStatistic += - ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer); + ReadWriteForEncodingUtils.writeUnsignedVarInt(pageBodySize, pageBuffer); } else { ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer); - ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer); + ReadWriteForEncodingUtils.writeUnsignedVarInt(pageBodySize, pageBuffer); statistics.serialize(pageBuffer); } // write page content to temp PBAOS logger.trace( "start to flush a time page data into buffer, buffer position {} ", pageBuffer.size()); - IEncryptor encryptor = IEncryptor.getEncryptor(encryptParam); - if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) { - if (encryptor.getEncryptionType().equals(EncryptionType.UNENCRYPTED)) { - try (WritableByteChannel channel = Channels.newChannel(pageBuffer)) { - channel.write(pageData); - } - } else { - byte[] encryptedBytes = null; - encryptedBytes = encryptor.encrypt(pageData.array(), pageData.position(), uncompressedSize); - // data is never a directByteBuffer now, so we can use data.array() - int encryptedSize = encryptedBytes.length; - pageBuffer.write(encryptedBytes, 0, encryptedSize); - } - - } else { - if (encryptor.getEncryptionType().equals(EncryptionType.UNENCRYPTED)) { - pageBuffer.write(compressedBytes, 0, compressedSize); - } else { - byte[] encryptedBytes = null; - encryptedBytes = encryptor.encrypt(compressedBytes, 0, compressedSize); - // data is never a directByteBuffer now, so we can use data.array() - int encryptedSize = encryptedBytes.length; - pageBuffer.write(encryptedBytes, 0, encryptedSize); - } - } + pageBuffer.write(pageBody.getData(), pageBody.getOffset(), pageBodySize); logger.trace( "finish flushing a time page data into buffer, buffer position {} ", pageBuffer.size()); return sizeWithoutStatistic; diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java index 525c3199a..189804bcb 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/page/ValuePageWriter.java @@ -22,10 +22,7 @@ import org.apache.tsfile.encoding.encoder.Encoder; import org.apache.tsfile.encrypt.EncryptParameter; import org.apache.tsfile.encrypt.EncryptUtils; -import org.apache.tsfile.encrypt.IEncryptor; import org.apache.tsfile.enums.TSDataType; -import org.apache.tsfile.file.metadata.enums.CompressionType; -import org.apache.tsfile.file.metadata.enums.EncryptionType; import org.apache.tsfile.file.metadata.statistics.Statistics; import org.apache.tsfile.utils.Binary; import org.apache.tsfile.utils.PublicBAOS; @@ -37,8 +34,6 @@ import java.io.IOException; import java.io.Serializable; import java.nio.ByteBuffer; -import java.nio.channels.Channels; -import java.nio.channels.WritableByteChannel; /** * This writer is used to write value into a page. It consists of a value encoder and respective @@ -272,6 +267,12 @@ public int writeEmptyPageIntoBuff(PublicBAOS pageBuffer) { /** write the page header and data into the PageWriter's output stream. */ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) throws IOException { + return writePageHeaderAndDataIntoBuff(pageBuffer, first, first ? 0 : -1); + } + + /** write the page header and data into the PageWriter's output stream. */ + public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first, int pageIndex) + throws IOException { if (size == 0) { return 0; } else if (statistics.getCount() == 0) { @@ -281,22 +282,9 @@ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) ByteBuffer pageData = getUncompressedBytes(); int uncompressedSize = pageData.remaining(); - int compressedSize; - byte[] compressedBytes = null; - - if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) { - compressedSize = uncompressedSize; - } else if (compressor.getType().equals(CompressionType.GZIP)) { - compressedBytes = - compressor.compress(pageData.array(), pageData.position(), uncompressedSize); - compressedSize = compressedBytes.length; - } else { - compressedBytes = new byte[compressor.getMaxBytesForCompression(uncompressedSize)]; - // data is never a directByteBuffer now, so we can use data.array() - compressedSize = - compressor.compress( - pageData.array(), pageData.position(), uncompressedSize, compressedBytes); - } + EncodedPageBody pageBody = + PageBodyEncoder.encode(pageData, uncompressedSize, compressor, encryptParam, pageIndex); + int pageBodySize = pageBody.size(); // write the page header to IOWriter int sizeWithoutStatistic = 0; @@ -304,39 +292,15 @@ public int writePageHeaderAndDataIntoBuff(PublicBAOS pageBuffer, boolean first) sizeWithoutStatistic += ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer); sizeWithoutStatistic += - ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer); + ReadWriteForEncodingUtils.writeUnsignedVarInt(pageBodySize, pageBuffer); } else { ReadWriteForEncodingUtils.writeUnsignedVarInt(uncompressedSize, pageBuffer); - ReadWriteForEncodingUtils.writeUnsignedVarInt(compressedSize, pageBuffer); + ReadWriteForEncodingUtils.writeUnsignedVarInt(pageBodySize, pageBuffer); statistics.serialize(pageBuffer); } - IEncryptor encryptor = IEncryptor.getEncryptor(encryptParam); // write page content to temp PBAOS logger.trace("start to flush a page data into buffer, buffer position {} ", pageBuffer.size()); - if (compressor.getType().equals(CompressionType.UNCOMPRESSED)) { - if (encryptor.getEncryptionType().equals(EncryptionType.UNENCRYPTED)) { - try (WritableByteChannel channel = Channels.newChannel(pageBuffer)) { - channel.write(pageData); - } - } else { - byte[] encryptedBytes = null; - encryptedBytes = encryptor.encrypt(pageData.array(), pageData.position(), uncompressedSize); - // data is never a directByteBuffer now, so we can use data.array() - int encryptedSize = encryptedBytes.length; - pageBuffer.write(encryptedBytes, 0, encryptedSize); - } - - } else { - if (encryptor.getEncryptionType().equals(EncryptionType.UNENCRYPTED)) { - pageBuffer.write(compressedBytes, 0, compressedSize); - } else { - byte[] encryptedBytes = null; - encryptedBytes = encryptor.encrypt(compressedBytes, 0, compressedSize); - // data is never a directByteBuffer now, so we can use data.array() - int encryptedSize = encryptedBytes.length; - pageBuffer.write(encryptedBytes, 0, encryptedSize); - } - } + pageBuffer.write(pageBody.getData(), pageBody.getOffset(), pageBodySize); logger.trace("start to flush a page data into buffer, buffer position {} ", pageBuffer.size()); return sizeWithoutStatistic; } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/v4/AbstractTableModelTsFileWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/v4/AbstractTableModelTsFileWriter.java index 94be1fdc1..e2ce2ce03 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/v4/AbstractTableModelTsFileWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/v4/AbstractTableModelTsFileWriter.java @@ -94,7 +94,10 @@ protected AbstractTableModelTsFileWriter(File file, long chunkGroupSizeThreshold protected AbstractTableModelTsFileWriter( File file, long chunkGroupSizeThreshold, EncryptParameter firstEncryptParam) throws IOException { - this(new TsFileIOWriter(file), chunkGroupSizeThreshold, firstEncryptParam); + this( + new TsFileIOWriter(file, copyEncryptParameter(firstEncryptParam)), + chunkGroupSizeThreshold, + firstEncryptParam); } @TsFileApi @@ -123,23 +126,39 @@ protected AbstractTableModelTsFileWriter( LOG.warn(Messages.get("log.write.page_size_warn"), pageSize, chunkGroupSizeThreshold); } - this.secondEncryptParam = EncryptUtils.getEncryptParameter(firstEncryptParam); - String encryptLevel; - if (firstEncryptParam != null - && !Objects.equals(firstEncryptParam.getType(), "UNENCRYPTED") - && !Objects.equals(firstEncryptParam.getType(), "org.apache.tsfile.encrypt.UNENCRYPTED")) { - encryptLevel = "2"; - String str = - EncryptUtils.getKeyStr( - IEncryptor.getEncryptor(firstEncryptParam.getType(), firstEncryptParam.getKey()) - .encrypt(secondEncryptParam.getKey())); - fileWriter.setEncryptParam(encryptLevel, secondEncryptParam.getType(), str); + EncryptParameter fileEncryptParameter = fileWriter.getEncryptParameter(); + EncryptParameter effectiveEncryptParameter = + fileEncryptParameter != null && fileEncryptParameter.isTdePageAead() + ? fileEncryptParameter + : firstEncryptParam; + if (effectiveEncryptParameter != null && effectiveEncryptParameter.isTdePageAead()) { + this.secondEncryptParam = effectiveEncryptParameter; + fileWriter.setEncryptParam(effectiveEncryptParameter); + fileWriter.writeEncryptionHeaderIfNecessary(); } else { - encryptLevel = "0"; - fileWriter.setEncryptParam(encryptLevel, "org.apache.tsfile.encrypt.UNENCRYPTED", ""); + this.secondEncryptParam = EncryptUtils.getEncryptParameter(firstEncryptParam); + String encryptLevel; + if (firstEncryptParam != null + && !Objects.equals(firstEncryptParam.getType(), "UNENCRYPTED") + && !Objects.equals( + firstEncryptParam.getType(), "org.apache.tsfile.encrypt.UNENCRYPTED")) { + encryptLevel = "2"; + String str = + EncryptUtils.getKeyStr( + IEncryptor.getEncryptor(firstEncryptParam.getType(), firstEncryptParam.getKey()) + .encrypt(secondEncryptParam.getKey())); + fileWriter.setEncryptParam(encryptLevel, secondEncryptParam.getType(), str); + } else { + encryptLevel = "0"; + fileWriter.setEncryptParam(encryptLevel, "org.apache.tsfile.encrypt.UNENCRYPTED", ""); + } } } + private static EncryptParameter copyEncryptParameter(EncryptParameter parameter) { + return parameter == null ? null : parameter.copy(); + } + protected IChunkGroupWriter tryToInitialGroupWriter( IDeviceID deviceId, boolean isAligned, boolean isTableModel) throws IOException { IChunkGroupWriter groupWriter = groupWriters.get(deviceId); @@ -272,6 +291,10 @@ public void close() { fileWriter.endFile(); } catch (IOException e) { LOG.warn(Messages.get("log.write.close_file_exception"), e); + } finally { + if (secondEncryptParam != null && secondEncryptParam.isTdePageAead()) { + secondEncryptParam.close(); + } } } } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriter.java index 1f627d693..ff7789569 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriter.java @@ -45,6 +45,7 @@ public class ForceAppendTsFileWriter extends TsFileIOWriter { private long truncatePosition; private static Logger logger = LoggerFactory.getLogger(ForceAppendTsFileWriter.class); + private EncryptParameter ownedEncryptParameter; public ForceAppendTsFileWriter(File file) throws IOException { this( @@ -58,19 +59,26 @@ public ForceAppendTsFileWriter(File file, EncryptParameter param) throws IOExcep if (logger.isDebugEnabled()) { logger.debug(Messages.get("log.write.writer_opened"), file.getName()); } - this.out = FSFactoryProducer.getFileOutputFactory().getTsFileOutput(file.getPath(), true); this.file = file; - setEncryptParam(param); // file doesn't exist if (file.length() == 0 || !file.exists()) { throw new TsFileNotCompleteException( Messages.format("error.write.force_append_not_complete", file.getPath())); } + this.out = FSFactoryProducer.getFileOutputFactory().getTsFileOutput(file.getPath(), true); + setEncryptParam(param); try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath(), param, true)) { + EncryptParameter recoveredParameter = reader.getEncryptParam(); + if (recoveredParameter != null && recoveredParameter.isTdePageAead()) { + ownedEncryptParameter = recoveredParameter.copy(); + setEncryptParam(ownedEncryptParameter); + } + markExistingFileStarted(recoveredParameter != null && recoveredParameter.isTdePageAead()); + // this tsfile is not complete if (!reader.isComplete()) { throw new TsFileNotCompleteException( @@ -88,6 +96,9 @@ public ForceAppendTsFileWriter(File file, EncryptParameter param) throws IOExcep ChunkGroupMetadata chunkGroupMetadata = new ChunkGroupMetadata(device, chunkMetadataList); chunkGroupMetadataList.add(chunkGroupMetadata); } + } catch (IOException | RuntimeException e) { + closeAfterFailedInitialization(e); + throw e; } } @@ -98,4 +109,38 @@ public void doTruncate() throws IOException { public long getTruncatePosition() { return truncatePosition; } + + @Override + public void endFile() throws IOException { + try { + super.endFile(); + } finally { + closeOwnedEncryptParameter(); + } + } + + @Override + public void close() throws IOException { + try { + super.close(); + } finally { + closeOwnedEncryptParameter(); + } + } + + private void closeOwnedEncryptParameter() { + if (ownedEncryptParameter != null) { + ownedEncryptParameter.close(); + } + } + + private void closeAfterFailedInitialization(Exception exception) { + try { + out.close(); + } catch (IOException e) { + exception.addSuppressed(e); + } finally { + closeOwnedEncryptParameter(); + } + } } diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java index 09866b3eb..314a3c242 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriter.java @@ -126,21 +126,35 @@ public RestorableTsFileIOWriter(File file, boolean truncate, EncryptParameter pa } this.file = file; this.out = FSFactoryProducer.getFileOutputFactory().getTsFileOutput(file.getPath(), true); - this.param = param; - setEncryptParam(param); + this.param = param != null && param.isTdePageAead() ? param.copy() : param; + setEncryptParam(this.param); // file doesn't exist if (file.length() == 0) { - startFile(); - crashed = true; - canWrite = true; - return; + try { + startFile(); + crashed = true; + canWrite = true; + return; + } catch (IOException | RuntimeException e) { + closeAfterFailedInitialization(e); + throw e; + } } try { if (file.exists()) { try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getAbsolutePath(), param, false)) { + EncryptParameter recoveredParameter = reader.getEncryptParam(); + if (recoveredParameter != null && recoveredParameter.isTdePageAead()) { + if (this.param != null && this.param.isTdePageAead()) { + this.param.close(); + } + this.param = recoveredParameter.copy(); + setEncryptParam(this.param); + } + markExistingFileStarted(recoveredParameter != null && recoveredParameter.isTdePageAead()); schema.setEnabledUpdateSchema(false); truncatedSize = reader.selfCheck(schema, chunkGroupMetadataList, true); minPlanIndex = reader.getMinPlanIndex(); @@ -165,7 +179,7 @@ public RestorableTsFileIOWriter(File file, boolean truncate, EncryptParameter pa } } } catch (Exception e) { - out.close(); + closeAfterFailedInitialization(e); throw e; } } @@ -209,6 +223,45 @@ public Schema getKnownSchema() { return schema; } + @Override + public EncryptParameter getEncryptParameter() { + return param; + } + + @Override + public void endFile() throws IOException { + try { + super.endFile(); + } finally { + closeOwnedEncryptParameter(); + } + } + + @Override + public void close() throws IOException { + try { + super.close(); + } finally { + closeOwnedEncryptParameter(); + } + } + + private void closeOwnedEncryptParameter() { + if (param != null && param.isTdePageAead()) { + param.close(); + } + } + + private void closeAfterFailedInitialization(Exception exception) { + try { + out.close(); + } catch (IOException e) { + exception.addSuppressed(e); + } finally { + closeOwnedEncryptParameter(); + } + } + /** * For query. * diff --git a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java index df6d99b20..9f0db83f3 100644 --- a/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java +++ b/java/tsfile/src/main/java/org/apache/tsfile/write/writer/TsFileIOWriter.java @@ -29,6 +29,7 @@ import org.apache.tsfile.file.MetaMarker; import org.apache.tsfile.file.header.ChunkGroupHeader; import org.apache.tsfile.file.header.ChunkHeader; +import org.apache.tsfile.file.header.FileEncryptionHeader; import org.apache.tsfile.file.metadata.ChunkGroupMetadata; import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IChunkMetadata; @@ -62,6 +63,7 @@ import java.io.Serializable; import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.LinkedList; import java.util.List; @@ -135,6 +137,12 @@ public class TsFileIOWriter implements AutoCloseable { protected String encryptKey; + protected EncryptParameter encryptParameter; + + private boolean fileStarted; + + private boolean encryptionHeaderWritten; + private final List flushListeners = new ArrayList<>(); protected String currentTable; @@ -225,11 +233,19 @@ public void setEncryptParam(String encryptLevel, String encryptType, String encr this.encryptLevel = encryptLevel; this.encryptType = encryptType; this.encryptKey = encryptKey; + this.encryptParameter = new EncryptParameter(encryptType, null); } public void setEncryptParam(EncryptParameter param) { + this.encryptParameter = param; if (param == null) { setEncryptParam("0", "org.apache.tsfile.encrypt.UNENCRYPTED", null); + } else if (param.isTdePageAead()) { + // File-level encryption metadata is persisted in FileEncryptionHeader. Legacy footer fields + // stay explicitly unencrypted so wrapped key material is not duplicated. + encryptLevel = "0"; + encryptType = "org.apache.tsfile.encrypt.UNENCRYPTED"; + encryptKey = ""; } else { if (!Objects.equals(param.getType(), "UNENCRYPTED") && !Objects.equals(param.getType(), "org.apache.tsfile.encrypt.UNENCRYPTED")) { @@ -246,6 +262,29 @@ public void setEncryptParam(EncryptParameter param) { } } + public EncryptParameter getEncryptParameter() { + return encryptParameter; + } + + protected void markExistingFileStarted(boolean hasEncryptionHeader) { + fileStarted = true; + encryptionHeaderWritten = hasEncryptionHeader; + } + + public void writeEncryptionHeaderIfNecessary() throws IOException { + if (encryptParameter == null || !encryptParameter.isTdePageAead() || encryptionHeaderWritten) { + return; + } + long baseHeaderSize = MAGIC_STRING_BYTES.length + Byte.BYTES; + if (!fileStarted || out.getPosition() != baseHeaderSize) { + throw new IOException(Messages.get("error.write.encryption_header_must_precede_data")); + } + FileEncryptionHeader.serialize(encryptParameter, out.wrapAsStream()); + // A recoverable encrypted file must persist its wrapped data key before any encrypted page. + out.force(); + encryptionHeaderWritten = true; + } + public void addFlushListener(FlushChunkMetadataListener listener) { flushListeners.add(listener); } @@ -258,15 +297,19 @@ public void addFlushListener(FlushChunkMetadataListener listener) { * @throws IOException if an I/O error occurs. */ public void writeBytesToStream(PublicBAOS bytes) throws IOException { + writeEncryptionHeaderIfNecessary(); bytes.writeTo(out.wrapAsStream()); } protected void startFile() throws IOException { out.write(MAGIC_STRING_BYTES); out.write(VERSION_NUMBER_BYTE); + fileStarted = true; + writeEncryptionHeaderIfNecessary(); } public int startChunkGroup(IDeviceID deviceId) throws IOException { + writeEncryptionHeaderIfNecessary(); updateTableSize(deviceId); this.currentChunkGroupDeviceId = deviceId; if (logger.isDebugEnabled()) { @@ -328,6 +371,8 @@ public void startFlushChunk( int mask) throws IOException { + writeEncryptionHeaderIfNecessary(); + currentChunkMetadata = new ChunkMetadata( measurementId, @@ -352,6 +397,8 @@ public void startFlushChunk( /** Write a whole chunk in another file into this file. Providing fast merge for IoTDB. */ public void writeChunk(Chunk chunk, ChunkMetadata chunkMetadata) throws IOException { + writeEncryptionHeaderIfNecessary(); + validateChunkEncryptionContext(chunk); ChunkHeader chunkHeader = chunk.getHeader(); currentChunkMetadata = new ChunkMetadata( @@ -380,6 +427,7 @@ public void writeEmptyValueChunk( TSEncoding encodingType, Statistics statistics) throws IOException { + writeEncryptionHeaderIfNecessary(); currentChunkMetadata = new ChunkMetadata( measurementId, @@ -403,6 +451,8 @@ public void writeEmptyValueChunk( } public void writeChunk(Chunk chunk) throws IOException { + writeEncryptionHeaderIfNecessary(); + validateChunkEncryptionContext(chunk); ChunkHeader chunkHeader = chunk.getHeader(); currentChunkMetadata = new ChunkMetadata( @@ -435,6 +485,7 @@ public void endFile() throws IOException { if (!canWrite) { return; } + writeEncryptionHeaderIfNecessary(); updateTableSize(null); checkInMemoryPathCount(); @@ -670,12 +721,33 @@ public void setFile(File file) { } public void writePlanIndices() throws IOException { + writeEncryptionHeaderIfNecessary(); ReadWriteIOUtils.write(MetaMarker.OPERATION_INDEX_RANGE, out.wrapAsStream()); ReadWriteIOUtils.write(minPlanIndex, out.wrapAsStream()); ReadWriteIOUtils.write(maxPlanIndex, out.wrapAsStream()); out.flush(); } + private void validateChunkEncryptionContext(Chunk chunk) throws IOException { + EncryptParameter sourceParameter = chunk.getEncryptParam(); + boolean targetUsesPageAead = encryptParameter != null && encryptParameter.isTdePageAead(); + boolean sourceUsesPageAead = sourceParameter != null && sourceParameter.isTdePageAead(); + if (!targetUsesPageAead && !sourceUsesPageAead) { + return; + } + if (!targetUsesPageAead + || !sourceUsesPageAead + || !Objects.equals(encryptParameter.getProviderId(), sourceParameter.getProviderId()) + || !Objects.equals(encryptParameter.getProfileId(), sourceParameter.getProfileId()) + || !Objects.equals(encryptParameter.getKeyId(), sourceParameter.getKeyId()) + || !Objects.equals(encryptParameter.getKeyVersion(), sourceParameter.getKeyVersion()) + || !Arrays.equals(encryptParameter.getFileCryptoId(), sourceParameter.getFileCryptoId()) + || !Arrays.equals( + encryptParameter.getWrappedDataKey(), sourceParameter.getWrappedDataKey())) { + throw new IOException(Messages.get("error.write.encrypted_chunk_context_mismatch")); + } + } + public void truncate(long offset) throws IOException { out.truncate(offset); } diff --git a/java/tsfile/src/test/java/org/apache/tsfile/encrypt/PageCryptoContextTest.java b/java/tsfile/src/test/java/org/apache/tsfile/encrypt/PageCryptoContextTest.java new file mode 100644 index 000000000..0dcf2bcb1 --- /dev/null +++ b/java/tsfile/src/test/java/org/apache/tsfile/encrypt/PageCryptoContextTest.java @@ -0,0 +1,89 @@ +/* + * 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.tsfile.encrypt; + +import org.apache.tsfile.exception.encrypt.EncryptException; + +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.util.Arrays; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThrows; + +public class PageCryptoContextTest { + + @BeforeClass + public static void setUpProvider() { + EncryptionProviderRegistry.registerProvider(TestAeadEncryptionProvider.INSTANCE); + } + + @AfterClass + public static void tearDownProvider() { + EncryptionProviderRegistry.unregisterProvider(TestAeadEncryptionProvider.PROVIDER_ID); + } + + @Before + public void resetProviderCount() { + TestAeadEncryptionProvider.resetCreateCount(); + } + + @Test + public void testAeadRejectsTamperingAndPageSwap() { + byte[] dataKey = new byte[16]; + Arrays.fill(dataKey, (byte) 0x5A); + byte[] fileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + Arrays.fill(fileCryptoId, (byte) 0x2C); + EncryptParameter parameter = TestAeadEncryptionProvider.createParameter(dataKey, fileCryptoId); + byte[] plaintext = new byte[] {1, 3, 5, 7, 9, 11, 13, 15}; + + try { + IEncryptor encryptor = IEncryptor.getEncryptor(parameter); + IDecryptor decryptor = IDecryptor.getDecryptor(parameter); + PageCryptoContext page0 = + PageCryptoContext.forEncryption(parameter, plaintext.length, plaintext.length, 0); + PageCryptoContext page1 = + PageCryptoContext.forEncryption(parameter, plaintext.length, plaintext.length, 1); + + byte[] encryptedPage0 = encryptor.encryptPage(plaintext, 0, plaintext.length, page0); + byte[] encryptedPage1 = encryptor.encryptPage(plaintext, 0, plaintext.length, page1); + assertArrayEquals( + plaintext, decryptor.decryptPage(encryptedPage0, 0, encryptedPage0.length, page0)); + assertArrayEquals( + plaintext, decryptor.decryptPage(encryptedPage1, 0, encryptedPage1.length, page1)); + + assertThrows( + EncryptException.class, + () -> decryptor.decryptPage(encryptedPage0, 0, encryptedPage0.length, page1)); + + encryptedPage0[encryptedPage0.length - 1] ^= 1; + assertThrows( + EncryptException.class, + () -> decryptor.decryptPage(encryptedPage0, 0, encryptedPage0.length, page0)); + + assertEquals(1, TestAeadEncryptionProvider.getCreateCount()); + } finally { + parameter.close(); + } + } +} diff --git a/java/tsfile/src/test/java/org/apache/tsfile/encrypt/TestAeadEncryptionProvider.java b/java/tsfile/src/test/java/org/apache/tsfile/encrypt/TestAeadEncryptionProvider.java new file mode 100644 index 000000000..1e56df2d8 --- /dev/null +++ b/java/tsfile/src/test/java/org/apache/tsfile/encrypt/TestAeadEncryptionProvider.java @@ -0,0 +1,227 @@ +/* + * 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.tsfile.encrypt; + +import org.apache.tsfile.exception.encrypt.EncryptException; +import org.apache.tsfile.file.metadata.enums.EncryptionType; + +import javax.crypto.Cipher; +import javax.crypto.SecretKey; +import javax.crypto.spec.GCMParameterSpec; +import javax.crypto.spec.SecretKeySpec; + +import java.security.GeneralSecurityException; +import java.security.SecureRandom; +import java.util.Arrays; +import java.util.concurrent.atomic.AtomicInteger; + +/** JDK-only AEAD provider used to exercise the generic TsFile encryption SPI. */ +public final class TestAeadEncryptionProvider implements IEncryptProvider { + + public static final TestAeadEncryptionProvider INSTANCE = new TestAeadEncryptionProvider(); + public static final String PROVIDER_ID = "test-aead-provider"; + public static final String PROFILE_ID = "AES_GCM_128_AES_WRAP_128_V1"; + + private static final int IV_LENGTH = 12; + private static final int TAG_LENGTH = 16; + + private static final byte[] TEST_KEY_ENCRYPTION_KEY = + new byte[] { + 0x10, 0x32, 0x54, 0x76, 0x01, 0x23, 0x45, 0x67, 0x11, 0x22, 0x33, 0x44, 0x55, 0x66, 0x77, + 0x00 + }; + private static final SecureRandom SECURE_RANDOM = new SecureRandom(); + private static final AtomicInteger CREATE_COUNT = new AtomicInteger(); + + private TestAeadEncryptionProvider() {} + + @Override + public String getProviderId() { + return PROVIDER_ID; + } + + @Override + public IEncrypt create(EncryptParameter encryptParameter) { + if (!PROFILE_ID.equals(encryptParameter.getProfileId())) { + throw new EncryptException("unsupported test encryption profile"); + } + CREATE_COUNT.incrementAndGet(); + return new TestAeadEncrypt(resolveDataKey(encryptParameter)); + } + + public static EncryptParameter createParameter(byte[] dataKey, byte[] fileCryptoId) { + return EncryptParameter.pageAeadBuilder() + .key(dataKey) + .providerId(PROVIDER_ID) + .profileId(PROFILE_ID) + .keyId("test-key") + .keyVersion("1") + .wrappedDataKey(wrapDataKey(dataKey)) + .fileCryptoId(fileCryptoId) + .build(); + } + + public static int getCreateCount() { + return CREATE_COUNT.get(); + } + + public static void resetCreateCount() { + CREATE_COUNT.set(0); + } + + private static byte[] resolveDataKey(EncryptParameter parameter) { + byte[] dataKey = parameter.getKey(); + if (dataKey != null) { + return Arrays.copyOf(dataKey, dataKey.length); + } + try { + Cipher cipher = Cipher.getInstance("AESWrap"); + cipher.init(Cipher.UNWRAP_MODE, new SecretKeySpec(TEST_KEY_ENCRYPTION_KEY, "AES")); + SecretKey unwrapped = + (SecretKey) cipher.unwrap(parameter.getWrappedDataKey(), "AES", Cipher.SECRET_KEY); + return unwrapped.getEncoded(); + } catch (GeneralSecurityException e) { + throw new EncryptException("test data key unwrap failed", e); + } + } + + private static byte[] wrapDataKey(byte[] dataKey) { + try { + Cipher cipher = Cipher.getInstance("AESWrap"); + cipher.init(Cipher.WRAP_MODE, new SecretKeySpec(TEST_KEY_ENCRYPTION_KEY, "AES")); + return cipher.wrap(new SecretKeySpec(dataKey, "AES")); + } catch (GeneralSecurityException e) { + throw new EncryptException("test data key wrap failed", e); + } + } + + private static final class TestAeadEncrypt implements IEncrypt { + + private final byte[] dataKey; + + private TestAeadEncrypt(byte[] dataKey) { + this.dataKey = dataKey; + } + + @Override + public IDecryptor getDecryptor() { + return new AeadDecryptor(dataKey); + } + + @Override + public IEncryptor getEncryptor() { + return new AeadEncryptor(dataKey); + } + + @Override + public int getPageBodyOverhead() { + return IV_LENGTH + TAG_LENGTH; + } + + @Override + public void close() { + Arrays.fill(dataKey, (byte) 0); + } + } + + private static final class AeadEncryptor implements IEncryptor { + + private final byte[] dataKey; + + private AeadEncryptor(byte[] dataKey) { + this.dataKey = dataKey; + } + + @Override + public byte[] encrypt(byte[] data) { + return encrypt(data, 0, data.length); + } + + @Override + public byte[] encrypt(byte[] data, int offset, int size) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] encryptPage( + byte[] data, int offset, int size, PageCryptoContext pageCryptoContext) { + byte[] iv = new byte[IV_LENGTH]; + SECURE_RANDOM.nextBytes(iv); + try { + Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding"); + cipher.init( + Cipher.ENCRYPT_MODE, + new SecretKeySpec(dataKey, "AES"), + new GCMParameterSpec(TAG_LENGTH * Byte.SIZE, iv)); + cipher.updateAAD(pageCryptoContext.getAssociatedData()); + byte[] ciphertext = cipher.doFinal(data, offset, size); + byte[] pageBody = new byte[iv.length + ciphertext.length]; + System.arraycopy(iv, 0, pageBody, 0, iv.length); + System.arraycopy(ciphertext, 0, pageBody, iv.length, ciphertext.length); + return pageBody; + } catch (GeneralSecurityException e) { + throw new EncryptException("test page encryption failed", e); + } + } + + @Override + public EncryptionType getEncryptionType() { + return EncryptionType.NewWay; + } + } + + private static final class AeadDecryptor implements IDecryptor { + + private final byte[] dataKey; + + private AeadDecryptor(byte[] dataKey) { + this.dataKey = dataKey; + } + + @Override + public byte[] decrypt(byte[] data) { + return decrypt(data, 0, data.length); + } + + @Override + public byte[] decrypt(byte[] data, int offset, int size) { + throw new UnsupportedOperationException(); + } + + @Override + public byte[] decryptPage( + byte[] data, int offset, int size, PageCryptoContext pageCryptoContext) { + try { + GCMParameterSpec spec = + new GCMParameterSpec(TAG_LENGTH * Byte.SIZE, data, offset, IV_LENGTH); + Cipher cipher = Cipher.getInstance("AES/GCM/NoPadding"); + cipher.init(Cipher.DECRYPT_MODE, new SecretKeySpec(dataKey, "AES"), spec); + cipher.updateAAD(pageCryptoContext.getAssociatedData()); + return cipher.doFinal(data, offset + IV_LENGTH, size - IV_LENGTH); + } catch (GeneralSecurityException e) { + throw new EncryptException("test page decryption failed", e); + } + } + + @Override + public EncryptionType getEncryptionType() { + return EncryptionType.NewWay; + } + } +} diff --git a/java/tsfile/src/test/java/org/apache/tsfile/file/header/FileEncryptionHeaderTest.java b/java/tsfile/src/test/java/org/apache/tsfile/file/header/FileEncryptionHeaderTest.java new file mode 100644 index 000000000..82bd6a33c --- /dev/null +++ b/java/tsfile/src/test/java/org/apache/tsfile/file/header/FileEncryptionHeaderTest.java @@ -0,0 +1,94 @@ +/* + * 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.tsfile.file.header; + +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.TestAeadEncryptionProvider; +import org.apache.tsfile.exception.encrypt.EncryptException; +import org.apache.tsfile.file.MetaMarker; + +import org.junit.Test; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; + +public class FileEncryptionHeaderTest { + + @Test + public void testRoundTrip() throws IOException { + byte[] dataKey = new byte[16]; + byte[] fileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + for (int i = 0; i < dataKey.length; i++) { + dataKey[i] = (byte) (i + 1); + fileCryptoId[i] = (byte) (0x40 + i); + } + EncryptParameter parameter = TestAeadEncryptionProvider.createParameter(dataKey, fileCryptoId); + + ByteArrayOutputStream output = new ByteArrayOutputStream(); + FileEncryptionHeader.serialize(parameter, output); + byte[] serialized = output.toByteArray(); + assertEquals(MetaMarker.ENCRYPTION_HEADER, serialized[0]); + assertFalse(contains(serialized, dataKey)); + + ByteArrayInputStream input = new ByteArrayInputStream(serialized); + assertEquals(MetaMarker.ENCRYPTION_HEADER, input.read()); + EncryptParameter restored = FileEncryptionHeader.deserialize(input); + + assertNull(restored.getType()); + assertNull(restored.getKey()); + assertEquals(parameter.getProviderId(), restored.getProviderId()); + assertEquals(parameter.getProfileId(), restored.getProfileId()); + assertEquals(parameter.getKeyId(), restored.getKeyId()); + assertEquals(parameter.getKeyVersion(), restored.getKeyVersion()); + assertArrayEquals(parameter.getFileCryptoId(), restored.getFileCryptoId()); + assertArrayEquals(parameter.getWrappedDataKey(), restored.getWrappedDataKey()); + } + + @Test + public void testRejectInvalidFileCryptoId() { + EncryptParameter parameter = + TestAeadEncryptionProvider.createParameter(new byte[16], new byte[8]); + assertThrows( + EncryptException.class, + () -> FileEncryptionHeader.serialize(parameter, new ByteArrayOutputStream())); + } + + private static boolean contains(byte[] data, byte[] target) { + for (int i = 0; i <= data.length - target.length; i++) { + boolean matches = true; + for (int j = 0; j < target.length; j++) { + if (data[i + j] != target[j]) { + matches = false; + break; + } + } + if (matches) { + return true; + } + } + return false; + } +} diff --git a/java/tsfile/src/test/java/org/apache/tsfile/read/UnClosedTsFileReaderTest.java b/java/tsfile/src/test/java/org/apache/tsfile/read/UnClosedTsFileReaderTest.java index 8e865413a..5fadd06dc 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/read/UnClosedTsFileReaderTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/read/UnClosedTsFileReaderTest.java @@ -20,6 +20,9 @@ package org.apache.tsfile.read; import org.apache.tsfile.constant.TestConstant; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.EncryptionProviderRegistry; +import org.apache.tsfile.encrypt.TestAeadEncryptionProvider; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.ChunkMetadata; import org.apache.tsfile.file.metadata.IDeviceID.Factory; @@ -30,6 +33,8 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import org.apache.tsfile.write.writer.TsFileIOWriter; +import org.junit.AfterClass; +import org.junit.BeforeClass; import org.junit.Test; import java.io.File; @@ -41,6 +46,16 @@ public class UnClosedTsFileReaderTest { + @BeforeClass + public static void setUpEncryptionProvider() { + EncryptionProviderRegistry.registerProvider(TestAeadEncryptionProvider.INSTANCE); + } + + @AfterClass + public static void tearDownEncryptionProvider() { + EncryptionProviderRegistry.unregisterProvider(TestAeadEncryptionProvider.PROVIDER_ID); + } + @Test public void testRead() throws IOException { File file = new File(TestConstant.BASE_OUTPUT_PATH + File.separator + "test.tsfile"); @@ -71,4 +86,47 @@ public void testRead() throws IOException { file.delete(); } + + @Test + public void testReadEncryptedPagesUsingFileHeader() throws IOException { + File file = new File(TestConstant.BASE_OUTPUT_PATH + File.separator + "test-encrypted.tsfile"); + byte[] fileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + fileCryptoId[0] = 1; + EncryptParameter encryptParameter = + TestAeadEncryptionProvider.createParameter(new byte[16], fileCryptoId); + + try { + TsFileIOWriter writer = new TsFileIOWriter(file, encryptParameter); + writer.startChunkGroup(Factory.DEFAULT_FACTORY.create("root.sg1.d1")); + ChunkWriterImpl chunkWriter = + new ChunkWriterImpl(new MeasurementSchema("s1", TSDataType.INT64), encryptParameter); + chunkWriter.write(1, 1L); + chunkWriter.sealCurrentPage(); + chunkWriter.write(2, 2L); + chunkWriter.sealCurrentPage(); + chunkWriter.writeToFileWriter(writer); + writer.endChunkGroup(); + writer.close(); + ChunkMetadata chunkMetadata = + writer.getChunkGroupMetadataList().get(0).getChunkMetadataList().get(0); + + try (UnClosedTsFileReader reader = new UnClosedTsFileReader(file.getAbsolutePath(), null)) { + assertEquals("test-key", reader.getEncryptParam().getKeyId()); + ChunkReader chunkReader = new ChunkReader(reader.readMemChunk(chunkMetadata)); + + BatchData firstPage = chunkReader.nextPageData(); + assertTrue(firstPage.hasCurrent()); + assertEquals(1, firstPage.currentTime()); + assertEquals(1L, firstPage.currentValue()); + + BatchData secondPage = chunkReader.nextPageData(); + assertTrue(secondPage.hasCurrent()); + assertEquals(2, secondPage.currentTime()); + assertEquals(2L, secondPage.currentValue()); + } + } finally { + encryptParameter.close(); + file.delete(); + } + } } diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/TDEPageAeadTsFileTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/TDEPageAeadTsFileTest.java new file mode 100644 index 000000000..a5f32a95e --- /dev/null +++ b/java/tsfile/src/test/java/org/apache/tsfile/write/TDEPageAeadTsFileTest.java @@ -0,0 +1,162 @@ +/* + * 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.tsfile.write; + +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.common.conf.TSFileDescriptor; +import org.apache.tsfile.constant.TestConstant; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.EncryptionProviderRegistry; +import org.apache.tsfile.encrypt.TestAeadEncryptionProvider; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.header.ChunkHeader; +import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.read.TsFileReader; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.Chunk; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.expression.QueryExpression; +import org.apache.tsfile.read.query.dataset.QueryDataSet; +import org.apache.tsfile.write.record.TSRecord; +import org.apache.tsfile.write.record.datapoint.LongDataPoint; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.apache.tsfile.write.writer.TsFileIOWriter; + +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.List; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +public class TDEPageAeadTsFileTest { + + private final File file = + new File(TestConstant.BASE_OUTPUT_PATH + File.separator + "tde-page-aead.tsfile"); + + @BeforeClass + public static void setUpEncryptionProvider() { + EncryptionProviderRegistry.registerProvider(TestAeadEncryptionProvider.INSTANCE); + } + + @AfterClass + public static void tearDownEncryptionProvider() { + EncryptionProviderRegistry.unregisterProvider(TestAeadEncryptionProvider.PROVIDER_ID); + } + + @Before + public void setUp() { + if (!file.getParentFile().exists()) { + assertTrue(file.getParentFile().mkdirs()); + } + } + + @After + public void tearDown() { + if (file.exists()) { + assertTrue(file.delete()); + } + } + + @Test + public void testAlignedMultiPageReadWrite() throws Exception { + TSFileConfig config = TSFileDescriptor.getInstance().getConfig(); + int previousMaxPointsInPage = config.getMaxNumberOfPointsInPage(); + config.setMaxNumberOfPointsInPage(1); + byte[] fileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + fileCryptoId[0] = 1; + EncryptParameter encryptParameter = + TestAeadEncryptionProvider.createParameter(new byte[16], fileCryptoId); + + try { + List schemas = + Arrays.asList( + new MeasurementSchema("s1", TSDataType.INT64, TSEncoding.RLE), + new MeasurementSchema("s2", TSDataType.INT64, TSEncoding.RLE)); + try (TsFileWriter writer = new TsFileWriter(file, encryptParameter)) { + writer.registerAlignedTimeseries(new Path("d1"), schemas); + for (int i = 1; i <= 3; i++) { + writer.writeRecord( + new TSRecord("d1", i) + .addTuple(new LongDataPoint("s1", i * 10L)) + .addTuple(new LongDataPoint("s2", i * 100L))); + } + } + + try (TsFileReader reader = new TsFileReader(new TsFileSequenceReader(file.getPath()))) { + QueryDataSet dataSet = + reader.query( + QueryExpression.create( + Arrays.asList(new Path("d1", "s1", true), new Path("d1", "s2", true)), null)); + for (int i = 1; i <= 3; i++) { + RowRecord record = dataSet.next(); + assertEquals(i, record.getTimestamp()); + assertEquals(i * 10L, record.getFields().get(0).getLongV()); + assertEquals(i * 100L, record.getFields().get(1).getLongV()); + } + assertFalse(dataSet.hasNext()); + } + } finally { + encryptParameter.close(); + config.setMaxNumberOfPointsInPage(previousMaxPointsInPage); + } + } + + @Test + public void testRejectUnsafeEncryptedChunkReuse() throws Exception { + File targetFile = new File(file.getPath() + ".target"); + byte[] sourceFileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + byte[] targetFileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + sourceFileCryptoId[0] = 1; + targetFileCryptoId[0] = 2; + EncryptParameter sourceParameter = + TestAeadEncryptionProvider.createParameter(new byte[16], sourceFileCryptoId); + EncryptParameter targetParameter = + TestAeadEncryptionProvider.createParameter(new byte[16], targetFileCryptoId); + + try (TsFileIOWriter writer = new TsFileIOWriter(targetFile, targetParameter)) { + ChunkHeader header = + new ChunkHeader( + "s1", 0, TSDataType.INT64, CompressionType.UNCOMPRESSED, TSEncoding.PLAIN, 1); + Chunk sourceChunk = new Chunk(header, ByteBuffer.allocate(0), sourceParameter); + + assertThrows(IOException.class, () -> writer.writeChunk(sourceChunk)); + assertThrows(IOException.class, () -> sourceChunk.mergeChunkByAppendPage(sourceChunk)); + } finally { + sourceParameter.close(); + targetParameter.close(); + if (targetFile.exists()) { + assertTrue(targetFile.delete()); + } + } + } +} diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriterTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriterTest.java index 68d05cefd..311dd4e10 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriterTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/ForceAppendTsFileWriterTest.java @@ -18,6 +18,9 @@ */ package org.apache.tsfile.write.writer; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.EncryptionProviderRegistry; +import org.apache.tsfile.encrypt.TestAeadEncryptionProvider; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.fileSystem.FSFactoryProducer; @@ -34,11 +37,14 @@ import org.apache.tsfile.write.record.datapoint.FloatDataPoint; import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.AfterClass; import org.junit.Assert; +import org.junit.BeforeClass; import org.junit.Test; import java.io.File; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import static org.junit.Assert.assertEquals; @@ -51,6 +57,16 @@ public class ForceAppendTsFileWriterTest { TsFileGeneratorForTest.getTestTsFilePath("root.sg1", 0, 0, 1); private static FSFactory fsFactory = FSFactoryProducer.getFSFactory(); + @BeforeClass + public static void setUpEncryptionProvider() { + EncryptionProviderRegistry.registerProvider(TestAeadEncryptionProvider.INSTANCE); + } + + @AfterClass + public static void tearDownEncryptionProvider() { + EncryptionProviderRegistry.unregisterProvider(TestAeadEncryptionProvider.PROVIDER_ID); + } + @Test public void test() throws Exception { File file = fsFactory.getFile(FILE_NAME); @@ -120,4 +136,53 @@ record = dataSet.next(); assertTrue(file.delete()); } + + @Test + public void testEncryptedForceAppend() throws Exception { + File file = fsFactory.getFile(FILE_NAME + ".encrypted"); + if (!file.getParentFile().exists()) { + Assert.assertTrue(file.getParentFile().mkdirs()); + } + byte[] fileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + fileCryptoId[0] = 1; + EncryptParameter encryptParameter = + TestAeadEncryptionProvider.createParameter(new byte[16], fileCryptoId); + + try { + try (TsFileWriter writer = new TsFileWriter(file, encryptParameter)) { + writer.registerTimeseries( + new Path("d1"), new MeasurementSchema("s1", TSDataType.FLOAT, TSEncoding.RLE)); + writer.writeRecord(new TSRecord("d1", 1).addTuple(new FloatDataPoint("s1", 5))); + } + + ForceAppendTsFileWriter appendWriter = new ForceAppendTsFileWriter(file, encryptParameter); + EncryptParameter ownedParameter = appendWriter.getEncryptParameter(); + Assert.assertNotSame(encryptParameter, ownedParameter); + appendWriter.doTruncate(); + try (TsFileWriter writer = new TsFileWriter(appendWriter, encryptParameter)) { + writer.registerTimeseries( + new Path("d1"), new MeasurementSchema("s1", TSDataType.FLOAT, TSEncoding.RLE)); + writer.writeRecord(new TSRecord("d1", 2).addTuple(new FloatDataPoint("s1", 6))); + } + assertTrue(ownedParameter.isDestroyed()); + assertFalse(encryptParameter.isDestroyed()); + + try (TsFileReader reader = new TsFileReader(new TsFileSequenceReader(file.getPath()))) { + QueryDataSet dataSet = + reader.query( + QueryExpression.create( + Collections.singletonList(new Path("d1", "s1", true)), null)); + RowRecord first = dataSet.next(); + assertEquals(1, first.getTimestamp()); + assertEquals(5.0f, first.getFields().get(0).getFloatV(), 0.001); + RowRecord second = dataSet.next(); + assertEquals(2, second.getTimestamp()); + assertEquals(6.0f, second.getFields().get(0).getFloatV(), 0.001); + assertFalse(dataSet.hasNext()); + } + } finally { + encryptParameter.close(); + file.delete(); + } + } } diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriterTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriterTest.java index 0e1279fae..f0c770b0f 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriterTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/RestorableTsFileIOWriterTest.java @@ -20,6 +20,9 @@ package org.apache.tsfile.write.writer; import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.EncryptionProviderRegistry; +import org.apache.tsfile.encrypt.TestAeadEncryptionProvider; import org.apache.tsfile.enums.TSDataType; import org.apache.tsfile.exception.NotCompatibleTsFileException; import org.apache.tsfile.file.MetaMarker; @@ -44,8 +47,10 @@ import org.apache.tsfile.write.schema.MeasurementSchema; import org.junit.After; +import org.junit.AfterClass; import org.junit.Assert; import org.junit.Before; +import org.junit.BeforeClass; import org.junit.Test; import java.io.File; @@ -67,6 +72,16 @@ public class RestorableTsFileIOWriterTest { private static final FSFactory fsFactory = FSFactoryProducer.getFSFactory(); File file = fsFactory.getFile(FILE_NAME); + @BeforeClass + public static void setUpEncryptionProvider() { + EncryptionProviderRegistry.registerProvider(TestAeadEncryptionProvider.INSTANCE); + } + + @AfterClass + public static void tearDownEncryptionProvider() { + EncryptionProviderRegistry.unregisterProvider(TestAeadEncryptionProvider.PROVIDER_ID); + } + @Before public void setUp() throws IOException { if (!file.getParentFile().exists()) { @@ -105,6 +120,55 @@ public void testOnlyHeadMagic() throws Exception { rWriter.close(); } + @Test + public void testEncryptedHeaderSurvivesUnclosedFile() throws Exception { + byte[] fileCryptoId = new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]; + for (int i = 0; i < fileCryptoId.length; i++) { + fileCryptoId[i] = (byte) (i + 1); + } + EncryptParameter parameter = + TestAeadEncryptionProvider.createParameter(new byte[16], fileCryptoId); + TsFileWriter writer = new TsFileWriter(file, parameter); + long expectedDataStartOffset; + try (TsFileSequenceReader reader = new TsFileSequenceReader(file.getPath(), false)) { + expectedDataStartOffset = reader.getDataStartOffset(); + assertNotEquals( + TSFileConfig.MAGIC_STRING.getBytes().length + Byte.BYTES, expectedDataStartOffset); + assertEquals("test-key", reader.getEncryptParam().getKeyId()); + Assert.assertArrayEquals(fileCryptoId, reader.getEncryptParam().getFileCryptoId()); + } + writer.getIOWriter().close(); + writer.getIOWriter().getEncryptParameter().close(); + + RestorableTsFileIOWriter restorableWriter = new RestorableTsFileIOWriter(file, parameter); + assertEquals(expectedDataStartOffset, restorableWriter.getTruncatedSize()); + assertEquals("test-key", restorableWriter.getEncryptParameter().getKeyId()); + EncryptParameter ownedParameter = restorableWriter.getEncryptParameter(); + Assert.assertNotSame(parameter, ownedParameter); + try (TsFileWriter recoveredWriter = new TsFileWriter(restorableWriter, parameter)) { + // Closing writes a valid footer without replacing the file encryption header. + } + Assert.assertTrue(ownedParameter.isDestroyed()); + Assert.assertFalse(parameter.isDestroyed()); + parameter.close(); + } + + @Test + public void testCloseDestroysOwnedTdeParameter() throws Exception { + EncryptParameter parameter = + TestAeadEncryptionProvider.createParameter( + new byte[16], new byte[EncryptParameter.FILE_CRYPTO_ID_LENGTH]); + RestorableTsFileIOWriter writer = new RestorableTsFileIOWriter(file, parameter); + EncryptParameter ownedParameter = writer.getEncryptParameter(); + + writer.close(); + + Assert.assertNotSame(parameter, ownedParameter); + Assert.assertTrue(ownedParameter.isDestroyed()); + Assert.assertFalse(parameter.isDestroyed()); + parameter.close(); + } + @Test public void testOnlyFirstMask() throws Exception { TsFileWriter writer = new TsFileWriter(file); diff --git a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/TimePageWriterTest.java b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/TimePageWriterTest.java index 878b9d7a7..23100e114 100644 --- a/java/tsfile/src/test/java/org/apache/tsfile/write/writer/TimePageWriterTest.java +++ b/java/tsfile/src/test/java/org/apache/tsfile/write/writer/TimePageWriterTest.java @@ -23,14 +23,24 @@ import org.apache.tsfile.encoding.decoder.PlainDecoder; import org.apache.tsfile.encoding.encoder.Encoder; import org.apache.tsfile.encoding.encoder.PlainEncoder; +import org.apache.tsfile.encrypt.EncryptParameter; +import org.apache.tsfile.encrypt.EncryptionProviderRegistry; +import org.apache.tsfile.encrypt.IDecryptor; +import org.apache.tsfile.encrypt.TestAeadEncryptionProvider; import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.header.ChunkHeader; +import org.apache.tsfile.file.header.PageHeader; import org.apache.tsfile.file.metadata.enums.CompressionType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; import org.apache.tsfile.file.metadata.statistics.TimeStatistics; +import org.apache.tsfile.read.reader.chunk.ChunkReader; import org.apache.tsfile.utils.PublicBAOS; import org.apache.tsfile.utils.ReadWriteForEncodingUtils; import org.apache.tsfile.utils.ReadWriteIOUtils; import org.apache.tsfile.write.page.TimePageWriter; +import org.junit.AfterClass; +import org.junit.BeforeClass; import org.junit.Test; import java.io.IOException; @@ -41,6 +51,16 @@ public class TimePageWriterTest { + @BeforeClass + public static void setUpEncryptionProvider() { + EncryptionProviderRegistry.registerProvider(TestAeadEncryptionProvider.INSTANCE); + } + + @AfterClass + public static void tearDownEncryptionProvider() { + EncryptionProviderRegistry.unregisterProvider(TestAeadEncryptionProvider.PROVIDER_ID); + } + @Test public void testWrite() { Encoder timeEncoder = new PlainEncoder(TSDataType.INT64, 0); @@ -168,4 +188,46 @@ public void testWritePageHeaderAndDataIntoBuffWithSnappy() { fail(); } } + + @Test + public void testWriteAndReadPageWithAeadEncryption() throws IOException { + Encoder timeEncoder = new PlainEncoder(TSDataType.INT64, 0); + ICompressor compressor = ICompressor.getCompressor(CompressionType.UNCOMPRESSED); + EncryptParameter encryptParameter = + TestAeadEncryptionProvider.createParameter(new byte[16], new byte[16]); + TimePageWriter pageWriter = new TimePageWriter(timeEncoder, compressor, encryptParameter); + PublicBAOS publicBAOS = new PublicBAOS(); + + try { + pageWriter.write(1L); + pageWriter.write(2L); + pageWriter.write(3L); + pageWriter.writePageHeaderAndDataIntoBuff(publicBAOS, false, 0); + + ByteBuffer pageBuffer = ByteBuffer.wrap(publicBAOS.getBuf(), 0, publicBAOS.size()); + PageHeader pageHeader = PageHeader.deserializeFrom(pageBuffer, TSDataType.VECTOR); + assertEquals(24, pageHeader.getUncompressedSize()); + assertEquals(52, pageHeader.getCompressedSize()); + + ChunkHeader chunkHeader = + new ChunkHeader( + "time", + pageHeader.getCompressedSize(), + TSDataType.INT64, + CompressionType.UNCOMPRESSED, + TSEncoding.PLAIN, + 1); + IDecryptor decryptor = IDecryptor.getDecryptor(encryptParameter); + ByteBuffer uncompressedPageData = + ChunkReader.deserializePageData( + pageHeader, pageBuffer, chunkHeader, decryptor, encryptParameter, 0); + + assertEquals(1L, ReadWriteIOUtils.readLong(uncompressedPageData)); + assertEquals(2L, ReadWriteIOUtils.readLong(uncompressedPageData)); + assertEquals(3L, ReadWriteIOUtils.readLong(uncompressedPageData)); + assertEquals(0, uncompressedPageData.remaining()); + } finally { + encryptParameter.close(); + } + } }