QPID-5767: [linearstore] broker segfaults when recovering journal file with damaged header

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1599243 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 27 insertions, 21 deletions

diff --git a/qpid/cpp/src/qpid/linearstore/ISSUES b/qpid/cpp/src/qpid/linearstore/ISSUES
index 45afb9f8e4..4da8232355 100644
--- a/qpid/cpp/src/qpid/linearstore/ISSUES
+++ b/qpid/cpp/src/qpid/linearstore/ISSUES
@@ -47,6 +47,7 @@ Current/pending:
    5464 -        [linearstore] Incompletely created journal files accumulate in EFP
    -    1088944  [Linearstore] store does not return all files to EFP after purging big queue
    -    1078937  [linearstore] Installation and tests for new store analysis tool qpid-qls-analyze
+                   svn r.1596633 2014-05-21: Modified to run from installed location
 
 Fixed/closed (in commit order):
 ===============================
@@ -157,6 +158,7 @@ no.   svn r  Q-JIRA     RHBZ       Date Alt Committer
 26. 1584379    5661        - 2014-04-03
 27. 1594215    5750  1078142 2014-05-13
 28. 1596509    5767  1098118 2014-05-21 (pmoravec)
+29. 1596633  NO-JIRA 1078937 2014-05-21
 
 See above sections for details on these checkins.
 
diff --git a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
index a8311ccbad..de91796971 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
+++ b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
@@ -399,10 +399,10 @@ void RecoveryManager::analyzeJournalFileHeaders(efpIdentity_t& efpIdentity) {
     stringList_t directoryList;
     jdir::read_dir(journalDirectory_, directoryList, false, true, false, true);
     for (stringListConstItr_t i = directoryList.begin(); i != directoryList.end(); ++i) {
-        readJournalFileHeader(*i, fileHeader, headerQueueName);
-        if (headerQueueName.empty()) {
+        bool hdrOk = readJournalFileHeader(*i, fileHeader, headerQueueName);
+        if (!hdrOk || headerQueueName.empty()) {
             std::ostringstream oss;
-            oss << "Journal file " << (*i) << " is uninitialized";
+            oss << "Journal file " << (*i) << " is uninitialized or corrupted";
             journalLogRef_.log(JournalLog::LOG_WARN, queueName_, oss.str());
             uninitFileList_.push_back(*i);
         } else if (headerQueueName.compare(queueName_) != 0) {
@@ -893,7 +893,7 @@ bool RecoveryManager::readFileHeader() {
     file_hdr_t fhdr;
     inFileStream_.read((char*)&fhdr, sizeof(fhdr));
     checkFileStreamOk(true);
-    if (::file_hdr_check(&fhdr, QLS_FILE_MAGIC, QLS_JRNL_VERSION, efpFileSize_kib_) != 0) {
+    if (::file_hdr_check(&fhdr, QLS_FILE_MAGIC, QLS_JRNL_VERSION, efpFileSize_kib_, QLS_MAX_QUEUE_NAME_LEN) != 0) {
         firstRecordOffset_ = fhdr._fro;
         currentSerial_ = fhdr._rhdr._serial;
     } else {
@@ -907,7 +907,7 @@ bool RecoveryManager::readFileHeader() {
 }
 
 // static private
-void RecoveryManager::readJournalFileHeader(const std::string& journalFileName,
+bool RecoveryManager::readJournalFileHeader(const std::string& journalFileName,
                                             ::file_hdr_t& fileHeaderRef,
                                             std::string& queueName) {
     const std::size_t headerBlockSize = QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_KIB * 1024;
@@ -928,8 +928,9 @@ void RecoveryManager::readJournalFileHeader(const std::string& journalFileName,
     }
     ifs.close();
     ::memcpy(&fileHeaderRef, buffer, sizeof(::file_hdr_t));
+    if (::file_hdr_check(&fileHeaderRef, QLS_FILE_MAGIC, QLS_JRNL_VERSION, 0, QLS_MAX_QUEUE_NAME_LEN)) return false;
     queueName.assign(buffer + sizeof(::file_hdr_t), fileHeaderRef._queue_name_len);
-
+    return true;
 }
 
 void RecoveryManager::removeEmptyFiles(EmptyFilePool* emptyFilePoolPtr) {
diff --git a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h
index e19f92e305..b668e1e2ea 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.h
@@ -147,7 +147,7 @@ protected:
     void readJournalData(char* target, const std::streamsize size);
     void removeEmptyFiles(EmptyFilePool* emptyFilePoolPtr);
 
-    static void readJournalFileHeader(const std::string& journalFileName,
+    static bool readJournalFileHeader(const std::string& journalFileName,
                                       ::file_hdr_t& fileHeaderRef,
                                       std::string& queueName);
 };
diff --git a/qpid/cpp/src/qpid/linearstore/journal/jcfg.h b/qpid/cpp/src/qpid/linearstore/journal/jcfg.h
index 1659fe37fa..b33a419a9d 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/jcfg.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/jcfg.h
@@ -54,6 +54,7 @@
 #define QLS_EMPTY_MAGIC                 0x78534c51  /**< ("QLSx" in little endian) Magic for empty dblk */
 #define QLS_JRNL_VERSION                2           /**< Version (of file layout) */
 #define QLS_JRNL_FHDR_RES_SIZE_SBLKS    1           /**< Journal file header reserved size in sblks (as defined by QLS_SBLK_SIZE_BYTES) */
+#define QLS_MAX_QUEUE_NAME_LEN          (QLS_JRNL_FHDR_RES_SIZE_SBLKS * QLS_SBLK_SIZE_BYTES) - sizeof(file_hdr_t)
 
 #define QLS_CLEAN                                   /**< If defined, writes QLS_CLEAN_CHAR to all filled areas on disk */
 #define QLS_CLEAN_CHAR                  0xff        /**< Char used to clear empty space on disk */
diff --git a/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c b/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c
index f504fb7943..4e6cf1b8fa 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c
+++ b/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.c
@@ -55,11 +55,11 @@ int file_hdr_init(void* dest, const uint64_t dest_len, const uint16_t uflag, con
     return set_time_now(dest);
 }
 
-int file_hdr_check(file_hdr_t* hdr, const uint32_t magic, const uint16_t version, const uint64_t data_size_kib) {
-    int res = rec_hdr_check_base(&hdr->_rhdr, magic, version);
-    if (res != 0) return res;
-    if (hdr->_data_size_kib != data_size_kib) return 3;
-    return 0;
+int file_hdr_check(file_hdr_t* hdr, const uint32_t magic, const uint16_t version, const uint64_t data_size_kib, const uint16_t max_queue_name_len) {
+    int err = rec_hdr_check_base(&hdr->_rhdr, magic, version);
+    if (data_size_kib && hdr->_data_size_kib != data_size_kib) err |= 0x1000;
+    if (hdr->_queue_name_len > max_queue_name_len) err |= 0x10000;
+    return err;
 }
 
 void file_hdr_copy(file_hdr_t* dest, const file_hdr_t* src) {
diff --git a/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.h b/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.h
index 2b2a3ffa3f..5987e1871e 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.h
+++ b/qpid/cpp/src/qpid/linearstore/journal/utils/file_hdr.h
@@ -92,8 +92,10 @@ typedef struct file_hdr_t {
 void file_hdr_create(file_hdr_t* dest, const uint32_t magic, const uint16_t version,
                      const uint16_t fhdr_size_sblks, const uint16_t efp_partition, const uint64_t file_size);
 int file_hdr_init(void* dest, const uint64_t dest_len, const uint16_t uflag, const uint64_t serial, const uint64_t rid,
-                  const uint64_t fro, const uint64_t file_number, const uint16_t queue_name_len, const char* queue_name);
-int file_hdr_check(file_hdr_t* hdr, const uint32_t magic, const uint16_t version, const uint64_t data_size_kib);
+                  const uint64_t fro, const uint64_t file_number, const uint16_t queue_name_len,
+                  const char* queue_name);
+int file_hdr_check(file_hdr_t* hdr, const uint32_t magic, const uint16_t version, const uint64_t data_size_kib,
+                   const uint16_t max_queue_name_len);
 void file_hdr_reset(file_hdr_t* target);
 int is_file_hdr_reset(file_hdr_t* target);
 void file_hdr_copy(file_hdr_t* dest, const file_hdr_t* src);
diff --git a/qpid/cpp/src/qpid/linearstore/journal/utils/rec_hdr.c b/qpid/cpp/src/qpid/linearstore/journal/utils/rec_hdr.c
index aa424abf37..32eda8de5a 100644
--- a/qpid/cpp/src/qpid/linearstore/journal/utils/rec_hdr.c
+++ b/qpid/cpp/src/qpid/linearstore/journal/utils/rec_hdr.c
@@ -38,14 +38,14 @@ void rec_hdr_copy(rec_hdr_t* dest, const rec_hdr_t* src) {
 }
 
 int rec_hdr_check_base(rec_hdr_t* header, const uint32_t magic, const uint16_t version) {
-    if (header->_magic != magic) return 1;
-    if (header->_version != version) return 2;
-    return 0;
+    int err = 0;
+    if (header->_magic != magic) err |= 0x1;
+    if (header->_version != version) err |= 0x10;
+    return err;
 }
 
 int rec_hdr_check(rec_hdr_t* header, const uint32_t magic, const uint16_t version, const uint64_t serial) {
-    int res = rec_hdr_check_base(header, magic, version);
-    if (res != 0) return res;
-    if (header->_serial != serial) return 3;
-    return 0;
+    int err = rec_hdr_check_base(header, magic, version);
+    if (header->_serial != serial) err |= 0x100;
+    return err;
 }