fix(readers): reject truncated gzip streams

.github/workflows/ci.yml

@@ -182,6 +182,12 @@ jobs:
         run: cmake --build build-otlp --parallel 2
       - name: Run OTLP tests
         run: ctest --test-dir build-otlp --output-on-failure -R '^otlp_'
+      - name: Run compressed file reader tests
+        run: >-
+          ctest --test-dir build-otlp
+          --output-on-failure
+          --timeout 10
+          -R '^file_logger_(gzip_compression|truncated_gzip|zstd_compression)_test$'
       - name: Upload OTLP logs
         if: failure()
         uses: actions/upload-artifact@v4

bench/adapters/LogItAdapter.cpp

@@ -1,6 +1,7 @@
 #include "LogItAdapter.hpp"
 
 #include <atomic>
+#include <condition_variable>
 #include <filesystem>
 #include <fstream>
 #include <limits>
@@ -89,9 +90,19 @@ namespace logit_bench {
 
         void wait() override {
             if (m_async) {
-                logit::detail::TaskExecutor::get_instance().wait();
+                const std::uint64_t generation = flush_generation();
+                logit::detail::TaskExecutor::get_instance().add_task([this]() {
+                    std::lock_guard<std::mutex> lock(m_flush_mutex);
+                    ++m_flush_generation;
+                    m_flush_cv.notify_all();
+                });
+
+                std::unique_lock<std::mutex> lock(m_flush_mutex);
+                m_flush_cv.wait(lock, [this, generation]() {
+                    return m_flush_generation > generation;
+                });
             }
-    
+
             std::lock_guard<std::mutex> lock(m_file_mutex);
             if (m_file.is_open()) {
                 m_file.flush();
@@ -118,14 +129,23 @@ namespace logit_bench {
                 }
             }
         }
+
+        std::uint64_t flush_generation() const {
+            std::lock_guard<std::mutex> lock(m_flush_mutex);
+            return m_flush_generation;
+        }
 
         bool m_async = false;
         SinkKind m_sink = SinkKind::Null;
         LatencyRecorder* m_recorder = nullptr;
 
         std::ofstream m_file;
         mutable std::mutex m_file_mutex;
-
+
+        mutable std::mutex m_flush_mutex;
+        std::condition_variable m_flush_cv;
+        std::uint64_t m_flush_generation = 0;
+
         std::atomic<int> m_level{static_cast<int>(logit::LogLevel::LOG_LVL_TRACE)};
     };
 

include/logit_cpp/logit/detail/CompressionUtils.hpp

@@ -103,14 +103,28 @@ inline bool decompress_string_gzip(const std::string& input, std::string& output
         zs.next_out = reinterpret_cast<Bytef*>(&output[offset]);
         zs.avail_out = static_cast<uInt>(buf_size);
 
+        const uInt previous_avail_in = zs.avail_in;
+        const uLong previous_total_out = zs.total_out;
         ret = inflate(&zs, Z_NO_FLUSH);
-        if (ret == Z_STREAM_ERROR || ret == Z_DATA_ERROR || ret == Z_MEM_ERROR) {
+
+        // Only these two results are valid for the loop below.  In particular,
+        // Z_BUF_ERROR means that inflate made no progress (typically because a
+        // compressed stream was truncated after consuming all input).
+        if (ret != Z_OK && ret != Z_STREAM_END) {
             inflateEnd(&zs);
             output.clear();
             return false;
         }
 
         offset = zs.total_out;
+
+        if (ret == Z_OK &&
+            zs.avail_in == previous_avail_in &&
+            zs.total_out == previous_total_out) {
+            inflateEnd(&zs);
+            output.clear();
+            return false;
+        }
     } while (ret != Z_STREAM_END);
 
     output.resize(zs.total_out);

tests/CMakeLists.txt

@@ -25,6 +25,7 @@ else()
         file_logger_external_cmd_compression_test.cpp
         file_logger_file_api_test.cpp
         file_logger_gzip_compression_test.cpp
+        file_logger_truncated_gzip_test.cpp
         file_logger_remove_old_logs_suffixes_test.cpp
         file_logger_rotation_naming_sequence_test.cpp
         file_logger_rotation_naming_timestamp_ms_test.cpp
@@ -89,6 +90,7 @@ else()
     )
     if(NOT LOGIT_WITH_GZIP)
         list(REMOVE_ITEM TEST_SOURCES file_logger_gzip_compression_test.cpp)
+        list(REMOVE_ITEM TEST_SOURCES file_logger_truncated_gzip_test.cpp)
         list(REMOVE_ITEM TEST_SOURCES file_logger_external_cmd_compression_test.cpp)
         list(REMOVE_ITEM TEST_SOURCES otlp_http_logger_gzip_test.cpp)
     endif()

tests/file_logger_truncated_gzip_test.cpp

@@ -0,0 +1,109 @@
+#include <logit.hpp>
+
+#if defined(LOGIT_HAS_ZLIB)
+
+#include <zlib.h>
+
+#include <chrono>
+#include <cstdio>
+#include <fstream>
+#include <sstream>
+#include <string>
+
+namespace {
+
+std::string make_unique_directory_name() {
+    const long long stamp = static_cast<long long>(
+        std::chrono::steady_clock::now().time_since_epoch().count());
+    return "truncated_gzip_logs_" + std::to_string(stamp);
+}
+
+std::string make_rotated_path(const std::string& current_path) {
+    std::string rotated = current_path;
+    const size_t pos = rotated.rfind(".log");
+    if (pos != std::string::npos) {
+        rotated.insert(pos, ".001");
+    }
+    return rotated + ".gz";
+}
+
+bool write_gzip_file(const std::string& path, const std::string& content) {
+    gzFile file = gzopen(path.c_str(), "wb");
+    if (!file) return false;
+
+    const int written = gzwrite(
+        file,
+        content.data(),
+        static_cast<unsigned int>(content.size()));
+    const int close_result = gzclose(file);
+    return written == static_cast<int>(content.size()) && close_result == Z_OK;
+}
+
+bool read_binary_file(const std::string& path, std::string& content) {
+    std::ifstream file(path.c_str(), std::ios_base::binary);
+    if (!file.is_open()) return false;
+
+    std::ostringstream stream;
+    stream << file.rdbuf();
+    content = stream.str();
+    return !file.bad();
+}
+
+bool write_binary_file(const std::string& path, const std::string& content) {
+    std::ofstream file(path.c_str(), std::ios_base::binary | std::ios_base::trunc);
+    if (!file.is_open()) return false;
+
+    file.write(content.data(), static_cast<std::streamsize>(content.size()));
+    return file.good();
+}
+
+} // namespace
+
+int main() {
+    const std::string directory = make_unique_directory_name();
+    const std::string payload = "truncated gzip payload\n";
+    std::string current_path;
+    std::string compressed_path;
+    int result_code = 1;
+
+    {
+        logit::FileLogger::Config config;
+        config.directory = directory;
+        config.async = false;
+
+        logit::FileLogger logger(config);
+        current_path = logger.get_string_param(logit::LoggerParam::LastFilePath);
+        compressed_path = make_rotated_path(current_path);
+
+        if (!write_gzip_file(compressed_path, payload)) {
+            logger.shutdown();
+        } else {
+            const logit::LogFileReadResult valid =
+                logger.read_log_file(compressed_path);
+
+            std::string compressed_bytes;
+            if (valid.ok && valid.content == payload &&
+                read_binary_file(compressed_path, compressed_bytes) &&
+                compressed_bytes.size() > 8) {
+                compressed_bytes.resize(compressed_bytes.size() - 8);
+                if (write_binary_file(compressed_path, compressed_bytes)) {
+                    const logit::LogFileReadResult truncated =
+                        logger.read_log_file(compressed_path);
+                    result_code =
+                        !truncated.ok && truncated.content.empty() ? 0 : 1;
+                }
+            }
+            logger.shutdown();
+        }
+    }
+
+    std::remove(compressed_path.c_str());
+    std::remove(current_path.c_str());
+    return result_code;
+}
+
+#else
+
+int main() { return 0; }
+
+#endif