Validate stream listpack first entry on RDB/RESTORE load (#15295)

**Summary**

The stream RDB/RESTORE loader read the first element of each stream
listpack node (the "valid entries" count) and immediately decoded it as
an integer, but under shallow validation (`sanitize-dump-payload no`)
only the listpack header had been checked — never the first entry
itself. A crafted payload whose first entry declares an oversized string
encoding (e.g. `LP_ENCODING_32BIT_STR` claiming `0x7FFFFFFF` bytes)
caused `lpGetIntegerValue()` to read encoding-dependent bytes past the
end of the listpack, triggering an out-of-bounds read / crash.

**Changes**

1. **First-entry validation before integer decode (`rdb.c`)**
In `rdbLoadObject`, the stream-loading path now obtains the first
element via `lpValidateFirst()` instead of `lpFirst()`, and then
validates that entry with `lpValidateNext()` before
`lpGetIntegerValue()` decodes it. If the entry is malformed, the load is
rejected cleanly via `rdbReportCorruptRDB("Stream listpack integrity
check failed.")` with proper cleanup (`sdsfree(nodekey)`,
`decrRefCount(o)`, `zfree(lp)`) and an early `NULL` return, matching how
other corrupt-payload cases are handled. This closes the gap where only
the listpack header — not its first entry — was checked under
`sanitize-dump-payload no`.

2. **Test (`corrupt-dump.tcl`)**
A new test (`corrupt payload: stream listpack entry with corrupt
encoding crashes lpFirst`) exercises the rejection path. It disables
payload sanitization and checksum validation, then issues a `RESTORE`
with a hand-built stream payload whose listpack has a valid header but a
first entry encoded as `LP_ENCODING_32BIT_STR` (`0xF0`) declaring a
`0x7FFFFFFF`-byte string that runs past the end of the listpack. The
test asserts the command fails with `*Bad data format*`, verifies the
`*Stream listpack integrity check failed*` warning is logged, and
confirms the server survives (`r ping`).

src/rdb.c

@@ -3328,12 +3328,14 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
                 return NULL;
             }
 
-            unsigned char *first = lpFirst(lp);
-            if (first == NULL) {
-                /* Serialized listpacks should never be empty, since on
-                 * deletion we should remove the radix tree key if the
-                 * resulting listpack is empty. */
-                rdbReportCorruptRDB("Empty listpack inside stream");
+            /* Under shallow validation (sanitize-dump-payload no) only the
+             * listpack header was checked, so validate the first entry here:
+             * reject an empty listpack, and ensure the entry is well-formed
+             * before lpGetIntegerValue() below decodes it unchecked. */
+            unsigned char *first, *next;
+            first = next = lpValidateFirst(lp);
+            if (first == NULL || !lpValidateNext(lp, &next, lp_size)) {
+                rdbReportCorruptRDB("Stream listpack integrity check failed.");
                 sdsfree(nodekey);
                 decrRefCount(o);
                 zfree(lp);

tests/integration/corrupt-dump.tcl

@@ -280,6 +280,20 @@ test {corrupt payload: listpack too long entry prev len} {
     }
 }
 
+test {corrupt payload: stream listpack entry with corrupt encoding crashes lpFirst} {
+    start_server [list overrides [list loglevel verbose use-exit-on-panic yes crash-memcheck-enabled no]] {
+        # Payload: stream listpack with a valid header but a first entry
+        # whose encoding byte is LP_ENCODING_32BIT_STR (0xF0) declaring a
+        # 0x7FFFFFFF-byte string running past the end of the listpack.
+        r config set sanitize-dump-payload no
+        r debug set-skip-checksum-validation 1
+        catch {r RESTORE mystream 0 "\x1B\x01\x10\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x00\x00\x1D\x1D\x00\x00\x00\x0A\x00\xF0\xFF\xFF\xFF\x7F\x01\x81\x6B\x02\x00\x01\x02\x01\x00\x01\x00\x01\x81\x76\x02\x04\x01\xFF\x01\x01\x00\x01\x00\x00\x00\x01\x00\x40\x64\x40\x64\x00\x00\x00\x0C\x00\x00\x00\x00\x00\x00\x00\x00\x00" REPLACE} err
+        assert_match "*Bad data format*" $err
+        verify_log_message 0 "*Stream listpack integrity check failed*" 0
+        r ping
+    }
+}
+
 test {corrupt payload: stream entry with invalid lp_count causing infinite loop in reverse iteration} {
     start_server [list overrides [list loglevel verbose use-exit-on-panic yes crash-memcheck-enabled no] ] {
         r config set sanitize-dump-payload no