You are viewing a plain text version of this content. The canonical link for it is here.
Posted to oak-commits@jackrabbit.apache.org by fo...@apache.org on 2023/04/04 12:26:27 UTC
[jackrabbit-oak] branch trunk updated: OAK-10171: datastore-copy cmd: add checksum validation (#886)
This is an automated email from the ASF dual-hosted git repository.
fortino pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
The following commit(s) were added to refs/heads/trunk by this push:
new 8ef786b89d OAK-10171: datastore-copy cmd: add checksum validation (#886)
8ef786b89d is described below
commit 8ef786b89d37617ff2b2f6d2926ad8ce02ff4e05
Author: Fabrizio Fortino <fa...@gmail.com>
AuthorDate: Tue Apr 4 14:26:20 2023 +0200
OAK-10171: datastore-copy cmd: add checksum validation (#886)
* OAK-10171: datastore-copy cmd: add checksum validation
* OAK-10171: remove unused import
* OAK-10171: use channels only instead of streams
* OAK-10171: configurable buffer size
* OAK-10171: set default buffer size to 16k
---
.../jackrabbit/oak/run/DataStoreCopyCommand.java | 12 ++-
.../org/apache/jackrabbit/oak/run/Downloader.java | 74 ++++++++++++++++---
.../oak/run/DataStoreCopyCommandTest.java | 86 ++++++++++++++++++----
3 files changed, 148 insertions(+), 24 deletions(-)
diff --git a/oak-run/src/main/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommand.java b/oak-run/src/main/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommand.java
index d71f9051db..3b6ccb7fa4 100644
--- a/oak-run/src/main/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommand.java
+++ b/oak-run/src/main/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommand.java
@@ -59,6 +59,8 @@ public class DataStoreCopyCommand implements Command {
private long retryInitialInterval;
private boolean failOnError;
private int slowLogThreshold;
+ private String checksumAlgorithm;
+ private int bufferSize;
@Override
public void execute(String... args) throws Exception {
@@ -67,7 +69,7 @@ public class DataStoreCopyCommand implements Command {
Stream<String> ids = null;
try (Downloader downloader = new Downloader(concurrency, connectTimeout, readTimeout, maxRetries,
- retryInitialInterval, failOnError, slowLogThreshold)) {
+ retryInitialInterval, failOnError, slowLogThreshold, checksumAlgorithm, bufferSize)) {
if (fileIncludePath != null) {
ids = Files.lines(fileIncludePath);
} else {
@@ -83,6 +85,7 @@ public class DataStoreCopyCommand implements Command {
item.source += "?" + sasToken;
}
item.destination = getDestinationFromId(id);
+ item.checksum = id.replaceAll("-", "");
downloader.offer(item);
});
@@ -164,6 +167,11 @@ public class DataStoreCopyCommand implements Command {
OptionSpec<Boolean> failOnErrorOpt = parser.accepts("fail-on-error",
"If true fails the execution immediately after the first error, otherwise it continues processing all the blobs (default false)")
.withRequiredArg().ofType(Boolean.class).defaultsTo(false);
+ OptionSpec<String> checksumOpt = parser.accepts("checksum", "The algorithm to compute the checksum (examples: SHA-256, MD5) or empty (the default) to skip checksum validation")
+ .withOptionalArg().ofType(String.class);
+ OptionSpec<Integer> bufferSizeOpt = parser.accepts("buffer-size",
+ "The buffer size for downloading and checksumming blobs (default 16384[16KB])")
+ .withRequiredArg().ofType(Integer.class).defaultsTo(16384);
OptionSet optionSet = parser.parse(args);
@@ -180,6 +188,8 @@ public class DataStoreCopyCommand implements Command {
this.retryInitialInterval = optionSet.valueOf(retryInitialIntervalOpt);
this.fileIncludePath = optionSet.valueOf(fileIncludePathOpt);
this.failOnError = optionSet.valueOf(failOnErrorOpt);
+ this.checksumAlgorithm = optionSet.valueOf(checksumOpt);
+ this.bufferSize = optionSet.valueOf(bufferSizeOpt);
}
protected static void setupLogging() throws IOException {
diff --git a/oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java b/oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java
index c22ee99169..be8e4bebc2 100644
--- a/oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java
+++ b/oak-run/src/main/java/org/apache/jackrabbit/oak/run/Downloader.java
@@ -24,15 +24,19 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.Closeable;
-import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URL;
import java.net.URLConnection;
+import java.nio.ByteBuffer;
import java.nio.channels.Channels;
+import java.nio.channels.FileChannel;
import java.nio.channels.ReadableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
+import java.nio.file.StandardOpenOption;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -60,14 +64,16 @@ public class Downloader implements Closeable {
private final int maxRetries;
private final long retryInitialInterval;
private final boolean failOnError;
+ private final String checksumAlgorithm;
+ private final int bufferSize;
private final List<Future<ItemResponse>> responses;
public Downloader(int concurrency, int connectTimeoutMs, int readTimeoutMs) {
- this(concurrency, connectTimeoutMs, readTimeoutMs, 3, 100L, false, 10_000);
+ this(concurrency, connectTimeoutMs, readTimeoutMs, 3, 100L, false, 10_000, null, 16384);
}
public Downloader(int concurrency, int connectTimeoutMs, int readTimeoutMs, int maxRetries, long retryInitialInterval,
- boolean failOnError, int slowLogThreshold) {
+ boolean failOnError, int slowLogThreshold, String checksumAlgorithm, int bufferSize) {
if (concurrency <= 0 || concurrency > 1000) {
throw new IllegalArgumentException("concurrency range must be between 1 and 1000");
}
@@ -84,6 +90,19 @@ public class Downloader implements Closeable {
this.maxRetries = maxRetries;
this.retryInitialInterval = retryInitialInterval;
this.failOnError = failOnError;
+ // fail fast in case the algorithm is not supported
+ if (checksumAlgorithm != null && checksumAlgorithm.trim().length() > 0) {
+ this.checksumAlgorithm = checksumAlgorithm.trim();
+ try {
+ MessageDigest.getInstance(checksumAlgorithm);
+ } catch (NoSuchAlgorithmException e) {
+ throw new RuntimeException(e);
+ }
+ } else {
+ this.checksumAlgorithm = null;
+ }
+ this.bufferSize = bufferSize;
+
this.executorService = new ThreadPoolExecutor(
(int) Math.ceil(concurrency * .1), concurrency, 60L, TimeUnit.SECONDS,
new LinkedBlockingQueue<>(),
@@ -144,17 +163,52 @@ public class Downloader implements Closeable {
long t0 = System.nanoTime();
URLConnection sourceUrl = new URL(item.source).openConnection();
- sourceUrl.setConnectTimeout(Downloader.this.connectTimeoutMs);
- sourceUrl.setReadTimeout(Downloader.this.readTimeoutMs);
+ sourceUrl.setConnectTimeout(connectTimeoutMs);
+ sourceUrl.setReadTimeout(readTimeoutMs);
+
+ // Updating a MessageDigest from multiple threads is not thread safe, so we cannot reuse a single instance.
+ // Creating a new instance is a lightweight operation, no need to increase complexity by creating a pool.
+ MessageDigest md = null;
+ if (checksumAlgorithm != null && item.checksum != null) {
+ md = MessageDigest.getInstance(checksumAlgorithm);
+ }
Path destinationPath = Paths.get(item.destination);
Files.createDirectories(destinationPath.getParent());
- long size;
+ long size = 0;
try (ReadableByteChannel byteChannel = Channels.newChannel(sourceUrl.getInputStream());
- FileOutputStream outputStream = new FileOutputStream(destinationPath.toFile())) {
- size = outputStream.getChannel()
- .transferFrom(byteChannel, 0, Long.MAX_VALUE);
+ FileChannel outputChannel = FileChannel.open(destinationPath, StandardOpenOption.CREATE, StandardOpenOption.WRITE)) {
+ ByteBuffer buffer = ByteBuffer.allocate(bufferSize);
+ int bytesRead;
+ while ((bytesRead = byteChannel.read(buffer)) != -1) {
+ buffer.flip();
+ if (md != null) {
+ md.update(buffer);
+ }
+ size += bytesRead;
+ outputChannel.write(buffer);
+ buffer.clear();
+ }
}
+
+ if (md != null) {
+ byte[] checksumBytes = md.digest();
+
+ // Convert the checksum bytes to a hexadecimal string
+ StringBuilder sb = new StringBuilder();
+ for (byte b : checksumBytes) {
+ sb.append(String.format("%02x", b));
+ }
+ String checksum = sb.toString();
+ // Warning: most modern checksum algorithms used for cryptographic purposes are designed to be case-insensitive,
+ // to ensure that the same checksum value is produced regardless of the input's case. There may be some
+ // legacy algorithms that are case-sensitive. Using equalsIgnoreCase can be considered safe here.
+ if (!checksum.equalsIgnoreCase(item.checksum)) {
+ Files.deleteIfExists(destinationPath);
+ throw new IOException("Checksum does not match! Expected: " + item.checksum + ", Actual: " + checksum);
+ }
+ }
+
long elapsed = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - t0);
if (slowLogThreshold > 0 && elapsed >= slowLogThreshold) {
LOG.warn("{} [{}] downloaded in {} ms", item.source, IOUtils.humanReadableByteCount(size), elapsed);
@@ -237,12 +291,14 @@ public class Downloader implements Closeable {
public static class Item {
public String source;
public String destination;
+ public String checksum;
@Override
public String toString() {
return "Item{" +
"source='" + source + '\'' +
", destination='" + destination + '\'' +
+ (checksum != null ? ", checksum='" + checksum + '\'' : "") +
'}';
}
}
diff --git a/oak-run/src/test/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommandTest.java b/oak-run/src/test/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommandTest.java
index b8e551e034..72efac276a 100644
--- a/oak-run/src/test/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommandTest.java
+++ b/oak-run/src/test/java/org/apache/jackrabbit/oak/run/DataStoreCopyCommandTest.java
@@ -17,7 +17,6 @@
package org.apache.jackrabbit.oak.run;
import com.google.common.base.Joiner;
-import com.google.common.collect.ImmutableSet;
import com.microsoft.azure.storage.blob.CloudBlobContainer;
import com.microsoft.azure.storage.blob.SharedAccessBlobPermissions;
import com.microsoft.azure.storage.blob.SharedAccessBlobPolicy;
@@ -39,6 +38,9 @@ import java.time.Duration;
import java.time.Instant;
import java.util.Date;
import java.util.EnumSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
import java.util.stream.Stream;
import static com.microsoft.azure.storage.blob.SharedAccessBlobPermissions.LIST;
@@ -52,9 +54,15 @@ public class DataStoreCopyCommandTest {
@ClassRule
public static AzuriteDockerRule AZURITE = new AzuriteDockerRule();
- private static final String BLOB1 = "8897-b9025dc534d4a9fa5920569b373cd714c8cfe5d030ca3f5edb25004894a5";
- private static final String BLOB2 = "c1f0-8893512e1e00910a9caff05487805632031f15a0bd8ee869c9205da59cb8";
- private static final ImmutableSet<String> BLOBS = ImmutableSet.of(BLOB1, BLOB2);
+ private static final String BLOB1 = "290F-493C44F5D63D06B374D0A5ABD292FAE38B92CAB2FAE5EFEFE1B0E9347F56";
+ private static final String BLOB2 = "F73F-16EDE021D01EFECF627B5E658BE52293F167CFE06C6B8D0E591CB25B68C9";
+ private static final String BLOB3 = "A096-9499131919BEED06F508FF848DE3D49C227374702466E22D944EAD6ADB8E";
+
+ private static final Map<String, String> BLOBS_WITH_CONTENT = Map.of(
+ BLOB1, "some content",
+ BLOB2, "some other content",
+ BLOB3, "content with different checksum"
+ );
@Rule
public TemporaryFolder outDir = new TemporaryFolder(new File("target"));
@@ -120,13 +128,14 @@ public class DataStoreCopyCommandTest {
assertTrue(thirdNode.exists() && thirdNode.isDirectory());
File blob = new File(thirdNode, blobName);
assertTrue(blob.exists() && blob.isFile());
- assertEquals(BLOB1, IOUtils.toString(blob.toURI(), StandardCharsets.UTF_8));
+ assertEquals(BLOBS_WITH_CONTENT.get(BLOB1), IOUtils.toString(blob.toURI(), StandardCharsets.UTF_8));
}
@Test
public void allBlobsWithFileIncludePath() throws Exception {
Path blobs = Files.createTempFile("blobs", "txt");
- IOUtils.write(BLOB1 + "\n" + BLOB2, Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
+ IOUtils.write(String.join("\n", BLOBS_WITH_CONTENT.keySet()),
+ Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
DataStoreCopyCommand cmd = new DataStoreCopyCommand();
cmd.execute(
"--source-repo",
@@ -140,14 +149,15 @@ public class DataStoreCopyCommandTest {
);
try (Stream<Path> files = Files.walk(outDir.getRoot().toPath()).filter(p -> p.toFile().isFile())) {
- assertEquals(2, files.count());
+ assertEquals(3, files.count());
}
}
@Test
public void allBlobsPlusMissingOne() throws Exception {
Path blobs = Files.createTempFile("blobs", "txt");
- IOUtils.write(BLOB1 + "\n" + BLOB2 + "\n" + "foo", Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
+ IOUtils.write(String.join("\n", BLOBS_WITH_CONTENT.keySet()) + "\n" + "foo",
+ Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
DataStoreCopyCommand cmd = new DataStoreCopyCommand();
cmd.execute(
"--source-repo",
@@ -161,7 +171,7 @@ public class DataStoreCopyCommandTest {
);
try (Stream<Path> files = Files.walk(outDir.getRoot().toPath()).filter(p -> p.toFile().isFile())) {
- assertEquals(2, files.count());
+ assertEquals(3, files.count());
}
}
@@ -185,7 +195,8 @@ public class DataStoreCopyCommandTest {
@Test
public void allBlobsPlusMissingOneWithFailOnError() throws Exception {
Path blobs = Files.createTempFile("blobs", "txt");
- IOUtils.write(BLOB1 + "\n" + BLOB2 + "\n" + "foo", Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
+ IOUtils.write(String.join("\n", BLOBS_WITH_CONTENT.keySet()) + "\n" + "foo",
+ Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
DataStoreCopyCommand cmd = new DataStoreCopyCommand();
assertThrows(RuntimeException.class, () -> cmd.execute(
"--source-repo",
@@ -212,16 +223,63 @@ public class DataStoreCopyCommandTest {
"--out-dir",
outDir.getRoot().getAbsolutePath()
);
- assertEquals(Joiner.on(File.separator).join(outDir.getRoot().getAbsolutePath(), "88",
- "97", "b9","8897b9025dc534d4a9fa5920569b373cd714c8cfe5d030ca3f5edb25004894a5"),
+ assertEquals(Joiner.on(File.separator).join(outDir.getRoot().getAbsolutePath(), "29",
+ "0F", "49","290F493C44F5D63D06B374D0A5ABD292FAE38B92CAB2FAE5EFEFE1B0E9347F56"),
cmd.getDestinationFromId(BLOB1)
);
}
+ @Test
+ public void allBlobsWithBogusChecksumAlgorithm() throws Exception {
+ Path blobs = Files.createTempFile("blobs", "txt");
+ IOUtils.write(String.join("\n", BLOBS_WITH_CONTENT.keySet()),
+ Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
+ DataStoreCopyCommand cmd = new DataStoreCopyCommand();
+ assertThrows(RuntimeException.class, () -> cmd.execute(
+ "--source-repo",
+ container.getUri().toURL().toString(),
+ "--file-include-path",
+ blobs.toString(),
+ "--sas-token",
+ container.generateSharedAccessSignature(policy(EnumSet.of(READ, LIST)), null),
+ "--out-dir",
+ outDir.getRoot().getAbsolutePath(),
+ "--checksum",
+ "SHA-foo"
+ ));
+ }
+
+ @Test
+ public void allBlobsWithChecksum() throws Exception {
+ Path blobs = Files.createTempFile("blobs", "txt");
+ IOUtils.write(String.join("\n", BLOBS_WITH_CONTENT.keySet()),
+ Files.newOutputStream(blobs.toFile().toPath()), StandardCharsets.UTF_8);
+ DataStoreCopyCommand cmd = new DataStoreCopyCommand();
+ cmd.execute(
+ "--source-repo",
+ container.getUri().toURL().toString(),
+ "--file-include-path",
+ blobs.toString(),
+ "--sas-token",
+ container.generateSharedAccessSignature(policy(EnumSet.of(READ, LIST)), null),
+ "--out-dir",
+ outDir.getRoot().getAbsolutePath(),
+ "--checksum",
+ "SHA-256"
+ );
+
+ try (Stream<Path> files = Files.walk(outDir.getRoot().toPath()).filter(p -> p.toFile().isFile())) {
+ assertEquals(
+ Set.of(Path.of(cmd.getDestinationFromId(BLOB1)).getFileName().toString(),
+ Path.of(cmd.getDestinationFromId(BLOB2)).getFileName().toString()),
+ files.map(f -> f.getFileName().toString()).collect(Collectors.toSet()));
+ }
+ }
+
private CloudBlobContainer createBlobContainer() throws Exception {
container = AZURITE.getContainer("blobstore");
- for (String blob : BLOBS) {
- container.getBlockBlobReference(blob).uploadText(blob);
+ for (Map.Entry<String, String> blob : BLOBS_WITH_CONTENT.entrySet()) {
+ container.getBlockBlobReference(blob.getKey()).uploadText(blob.getValue());
}
return container;
}