From d72bd3a171d5624dacc32f3e83a47bc49fcbef99 Mon Sep 17 00:00:00 2001
From: Yannick Welsch <yannick@welsch.lu>
Date: Thu, 21 Nov 2019 10:03:11 +0100
Subject: [PATCH] Verify translog checksum before UUID check (#49394)

When opening a translog file, we check whether the UUID matches what we expect (the UUID
from the latest commit). The UUID check can in certain cases fail when the translog is
corrupted. This commit changes the ordering of the checks so that corruption is detected first.
---
 .../index/translog/TranslogHeader.java          | 17 ++++++++++-------
 .../index/translog/TranslogHeaderTests.java     |  7 +++++--
 2 files changed, 15 insertions(+), 9 deletions(-)

diff --git a/server/src/main/java/org/elasticsearch/index/translog/TranslogHeader.java b/server/src/main/java/org/elasticsearch/index/translog/TranslogHeader.java
index 9a9a9b1288d..df5f0c476ff 100644
--- a/server/src/main/java/org/elasticsearch/index/translog/TranslogHeader.java
+++ b/server/src/main/java/org/elasticsearch/index/translog/TranslogHeader.java
@@ -136,13 +136,6 @@ final class TranslogHeader {
             final BytesRef uuid = new BytesRef(uuidLen);
             uuid.length = uuidLen;
             in.read(uuid.bytes, uuid.offset, uuid.length);
-            final BytesRef expectedUUID = new BytesRef(translogUUID);
-            if (uuid.bytesEquals(expectedUUID) == false) {
-                throw new TranslogCorruptedException(
-                    path.toString(),
-                    "expected shard UUID " + expectedUUID + " but got: " + uuid +
-                        " this translog file belongs to a different translog");
-            }
             // Read the primary term
             final long primaryTerm;
             if (version == VERSION_PRIMARY_TERM) {
@@ -160,6 +153,16 @@ final class TranslogHeader {
             final int headerSizeInBytes = headerSizeInBytes(version, uuid.length);
             assert channel.position() == headerSizeInBytes :
                 "Header is not fully read; header size [" + headerSizeInBytes + "], position [" + channel.position() + "]";
+
+            // verify UUID only after checksum, to ensure that UUID is not corrupted
+            final BytesRef expectedUUID = new BytesRef(translogUUID);
+            if (uuid.bytesEquals(expectedUUID) == false) {
+                throw new TranslogCorruptedException(
+                    path.toString(),
+                    "expected shard UUID " + expectedUUID + " but got: " + uuid +
+                        " this translog file belongs to a different translog");
+            }
+
             return new TranslogHeader(translogUUID, primaryTerm, headerSizeInBytes);
         } catch (EOFException e) {
             throw new TranslogCorruptedException(path.toString(), "translog header truncated", e);
diff --git a/server/src/test/java/org/elasticsearch/index/translog/TranslogHeaderTests.java b/server/src/test/java/org/elasticsearch/index/translog/TranslogHeaderTests.java
index 00eea99b690..c98d84912ff 100644
--- a/server/src/test/java/org/elasticsearch/index/translog/TranslogHeaderTests.java
+++ b/server/src/test/java/org/elasticsearch/index/translog/TranslogHeaderTests.java
@@ -38,6 +38,7 @@ import static org.hamcrest.Matchers.anyOf;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.lessThan;
+import static org.hamcrest.Matchers.not;
 
 public class TranslogHeaderTests extends ESTestCase {
 
@@ -66,9 +67,10 @@ public class TranslogHeaderTests extends ESTestCase {
         for (int i = 0; i < corruptions && Files.size(translogFile) > 0; i++) {
             TestTranslog.corruptFile(logger, random(), translogFile, false);
         }
-        expectThrows(TranslogCorruptedException.class, () -> {
+        final TranslogCorruptedException corruption = expectThrows(TranslogCorruptedException.class, () -> {
             try (FileChannel channel = FileChannel.open(translogFile, StandardOpenOption.READ)) {
-                final TranslogHeader translogHeader = TranslogHeader.read(outHeader.getTranslogUUID(), translogFile, channel);
+                final TranslogHeader translogHeader = TranslogHeader.read(
+                    randomBoolean() ? outHeader.getTranslogUUID() : UUIDs.randomBase64UUID(), translogFile, channel);
                 // succeeds if the corruption corrupted the version byte making this look like a v2 translog, because we don't check the
                 // checksum on this version
                 assertThat("version " + TranslogHeader.VERSION_CHECKPOINTS + " translog",
@@ -82,6 +84,7 @@ public class TranslogHeaderTests extends ESTestCase {
                 throw new TranslogCorruptedException(translogFile.toString(), "adjusted translog version", e);
             }
         });
+        assertThat(corruption.getMessage(), not(containsString("this translog file belongs to a different translog")));
     }
 
     public void testHeaderWithoutPrimaryTerm() throws Exception {