Commit ca1c1825 authored by Oran Agra's avatar Oran Agra
Browse files

Sanitize dump payload: ziplist, listpack, zipmap, intset, stream

When loading an encoded payload we will at least do a shallow validation to
check that the size that's encoded in the payload matches the size of the
allocation.
This let's us later use this encoded size to make sure the various offsets
inside encoded payload don't reach outside the allocation, if they do, we'll
assert/panic, but at least we won't segfault or smear memory.

We can also do 'deep' validation which runs on all the records of the encoded
payload and validates that they don't contain invalid offsets. This lets us
detect corruptions early and reject a RESTORE command rather than accepting
it and asserting (crashing) later when accessing that payload via some command.

configuration:
- adding ACL flag skip-sanitize-payload
- adding config sanitize-dump-payload [yes/no/clients]

For now, we don't have a good way to ensure MIGRATE in cluster resharding isn't
being slowed down by these sanitation, so i'm setting the default value to `no`,
but later on it should be set to `clients` by default.

changes:
- changing rdbReportError not to `exit` in RESTORE command
- adding a new stat to be able to later check if cluster MIGRATE isn't being
  slowed down by sanitation.
parent c4fdf09c
......@@ -366,6 +366,21 @@ rdbcompression yes
# tell the loading code to skip the check.
rdbchecksum yes
# Enables or disables full sanitation checks for ziplist and listpack etc when
# loading an RDB or RESTORE payload. This reduces the chances of a assertion or
# crash later on while processing commands.
# Options:
# no - Never perform full sanitation
# yes - Always perform full sanitation
# clients - Perform full sanitation only for user connections.
# Excludes: RDB files, RESTORE commands received from the master
# connection, and client connections which have the
# skip-sanitize-payload ACL flag.
# The default should be 'clients' but since it currently affects cluster
# resharding via MIGRATE, it is temporarily set to 'no' by default.
#
# sanitize-dump-payload no
# The filename where to dump the DB
dbfilename dump.rdb
......@@ -725,6 +740,8 @@ replica-priority 100
# off Disable the user: it's no longer possible to authenticate
# with this user, however the already authenticated connections
# will still work.
# skip-sanitize-payload RESTORE dump-payload sanitation is skipped.
# sanitize-payload RESTORE dump-payload is sanitized (default).
# +<command> Allow the execution of that command
# -<command> Disallow the execution of that command
# +@<category> Allow the execution of all the commands in such category
......
......@@ -89,12 +89,15 @@ struct ACLUserFlag {
const char *name;
uint64_t flag;
} ACLUserFlags[] = {
/* Note: the order here dictates the emitted order at ACLDescribeUser */
{"on", USER_FLAG_ENABLED},
{"off", USER_FLAG_DISABLED},
{"allkeys", USER_FLAG_ALLKEYS},
{"allchannels", USER_FLAG_ALLCHANNELS},
{"allcommands", USER_FLAG_ALLCOMMANDS},
{"nopass", USER_FLAG_NOPASS},
{"skip-sanitize-payload", USER_FLAG_SANITIZE_PAYLOAD_SKIP},
{"sanitize-payload", USER_FLAG_SANITIZE_PAYLOAD},
{NULL,0} /* Terminator. */
};
......@@ -829,6 +832,12 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
} else if (!strcasecmp(op,"off")) {
u->flags |= USER_FLAG_DISABLED;
u->flags &= ~USER_FLAG_ENABLED;
} else if (!strcasecmp(op,"skip-sanitize-payload")) {
u->flags |= USER_FLAG_SANITIZE_PAYLOAD_SKIP;
u->flags &= ~USER_FLAG_SANITIZE_PAYLOAD;
} else if (!strcasecmp(op,"sanitize-payload")) {
u->flags &= ~USER_FLAG_SANITIZE_PAYLOAD_SKIP;
u->flags |= USER_FLAG_SANITIZE_PAYLOAD;
} else if (!strcasecmp(op,"allkeys") ||
!strcasecmp(op,"~*"))
{
......@@ -1004,6 +1013,7 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
serverAssert(ACLSetUser(u,"resetkeys",-1) == C_OK);
serverAssert(ACLSetUser(u,"resetchannels",-1) == C_OK);
serverAssert(ACLSetUser(u,"off",-1) == C_OK);
serverAssert(ACLSetUser(u,"sanitize-payload",-1) == C_OK);
serverAssert(ACLSetUser(u,"-@all",-1) == C_OK);
} else {
errno = EINVAL;
......
......@@ -119,6 +119,13 @@ configEnum acl_pubsub_default_enum[] = {
{NULL, 0}
};
configEnum sanitize_dump_payload_enum[] = {
{"no", SANITIZE_DUMP_NO},
{"yes", SANITIZE_DUMP_YES},
{"clients", SANITIZE_DUMP_CLIENTS},
{NULL, 0}
};
/* Output buffer limits presets. */
clientBufferLimitsConfig clientBufferLimitsDefaults[CLIENT_TYPE_OBUF_COUNT] = {
{0, 0, 0}, /* normal */
......@@ -2361,6 +2368,7 @@ standardConfig configs[] = {
createEnumConfig("appendfsync", NULL, MODIFIABLE_CONFIG, aof_fsync_enum, server.aof_fsync, AOF_FSYNC_EVERYSEC, NULL, NULL),
createEnumConfig("oom-score-adj", NULL, MODIFIABLE_CONFIG, oom_score_adj_enum, server.oom_score_adj, OOM_SCORE_ADJ_NO, NULL, updateOOMScoreAdj),
createEnumConfig("acl-pubsub-default", NULL, MODIFIABLE_CONFIG, acl_pubsub_default_enum, server.acl_pubusub_default, USER_FLAG_ALLCHANNELS, NULL, NULL),
createEnumConfig("sanitize-dump-payload", NULL, MODIFIABLE_CONFIG, sanitize_dump_payload_enum, server.sanitize_dump_payload, SANITIZE_DUMP_NO, NULL, NULL),
/* Integer configs */
createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.dbnum, 16, INTEGER_CONFIG, NULL, NULL),
......
......@@ -5,7 +5,6 @@
* We do that by scanning the keyspace and for each pointer we have, we can try to
* ask the allocator if moving it to a new address will help reduce fragmentation.
*
* Copyright (c) 2020, Oran Agra
* Copyright (c) 2020, Redis Labs, Inc
* All rights reserved.
*
......
......@@ -281,6 +281,33 @@ size_t intsetBlobLen(intset *is) {
return sizeof(intset)+intrev32ifbe(is->length)*intrev32ifbe(is->encoding);
}
int intsetValidateIntegrity(const unsigned char *p, size_t size) {
const intset *is = (const intset *)p;
/* check that we can actually read the header. */
if (size < sizeof(*is))
return 0;
uint32_t encoding = intrev32ifbe(is->encoding);
size_t record_size;
if (encoding == INTSET_ENC_INT64) {
record_size = INTSET_ENC_INT64;
} else if (encoding == INTSET_ENC_INT32) {
record_size = INTSET_ENC_INT32;
} else if (encoding == INTSET_ENC_INT16){
record_size = INTSET_ENC_INT16;
} else {
return 0;
}
/* check that the size matchies (all records are inside the buffer). */
uint32_t count = intrev32ifbe(is->length);
if (sizeof(*is) + count*record_size != size)
return 0;
return 1;
}
#ifdef REDIS_TEST
#include <sys/time.h>
#include <time.h>
......
......@@ -46,6 +46,7 @@ int64_t intsetRandom(intset *is);
uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value);
uint32_t intsetLen(const intset *is);
size_t intsetBlobLen(intset *is);
int intsetValidateIntegrity(const unsigned char *is, size_t size);
#ifdef REDIS_TEST
int intsetTest(int argc, char *argv[]);
......
......@@ -5,6 +5,7 @@
* https://github.com/antirez/listpack
*
* Copyright (c) 2017, Salvatore Sanfilippo <antirez at gmail dot com>
* Copyright (c) 2020, Redis Labs, Inc
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
......@@ -41,6 +42,7 @@
#include "listpack.h"
#include "listpack_malloc.h"
#include "redisassert.h"
#define LP_HDR_SIZE 6 /* 32 bit total len + 16 bit number of elements. */
#define LP_HDR_NUMELE_UNKNOWN UINT16_MAX
......@@ -114,6 +116,21 @@
(p)[5] = ((v)>>8)&0xff; \
} while(0)
/* Validates that 'p' is not ouside the listpack.
* All function that return a pointer to an element in the listpack will assert
* that this element is valid, so it can be freely used.
* Generally functions such lpNext and lpDelete assume the input pointer is
* already validated (since it's the return value of another function). */
#define ASSERT_INTEGRITY(lp, p) do { \
assert((p) >= (lp)+LP_HDR_SIZE && (p) < (lp)+lpGetTotalBytes((lp))); \
} while (0)
/* Similar to the above, but validates the entire element lenth rather than just
* it's pointer. */
#define ASSERT_INTEGRITY_LEN(lp, p, len) do { \
assert((p) >= (lp)+LP_HDR_SIZE && (p)+(len) < (lp)+lpGetTotalBytes((lp))); \
} while (0)
/* Convert a string into a signed 64 bit integer.
* The function returns 1 if the string could be parsed into a (non-overflowing)
* signed 64 bit int, 0 otherwise. The 'value' will be set to the parsed value
......@@ -367,9 +384,14 @@ void lpEncodeString(unsigned char *buf, unsigned char *s, uint32_t len) {
}
}
/* Return the encoded length of the listpack element pointed by 'p'. If the
* element encoding is wrong then 0 is returned. */
uint32_t lpCurrentEncodedSize(unsigned char *p) {
/* Return the encoded length of the listpack element pointed by 'p'.
* This includes the encoding byte, length bytes, and the element data itself.
* If the element encoding is wrong then 0 is returned.
* Note that this method may access additional bytes (in case of 12 and 32 bit
* str), so should only be called when we know 'p' was already validated by
* lpCurrentEncodedSizeBytes or ASSERT_INTEGRITY_LEN (possibly since 'p' is
* a return value of another function that validated its return. */
uint32_t lpCurrentEncodedSizeUnsafe(unsigned char *p) {
if (LP_ENCODING_IS_7BIT_UINT(p[0])) return 1;
if (LP_ENCODING_IS_6BIT_STR(p[0])) return 1+LP_ENCODING_6BIT_STR_LEN(p);
if (LP_ENCODING_IS_13BIT_INT(p[0])) return 2;
......@@ -383,12 +405,30 @@ uint32_t lpCurrentEncodedSize(unsigned char *p) {
return 0;
}
/* Return bytes needed to encode the length of the listpack element pointed by 'p'.
* This includes just the encodign byte, and the bytes needed to encode the length
* of the element (excluding the element data itself)
* If the element encoding is wrong then 0 is returned. */
uint32_t lpCurrentEncodedSizeBytes(unsigned char *p) {
if (LP_ENCODING_IS_7BIT_UINT(p[0])) return 1;
if (LP_ENCODING_IS_6BIT_STR(p[0])) return 1;
if (LP_ENCODING_IS_13BIT_INT(p[0])) return 1;
if (LP_ENCODING_IS_16BIT_INT(p[0])) return 1;
if (LP_ENCODING_IS_24BIT_INT(p[0])) return 1;
if (LP_ENCODING_IS_32BIT_INT(p[0])) return 1;
if (LP_ENCODING_IS_64BIT_INT(p[0])) return 1;
if (LP_ENCODING_IS_12BIT_STR(p[0])) return 2;
if (LP_ENCODING_IS_32BIT_STR(p[0])) return 5;
if (p[0] == LP_EOF) return 1;
return 0;
}
/* Skip the current entry returning the next. It is invalid to call this
* function if the current element is the EOF element at the end of the
* listpack, however, while this function is used to implement lpNext(),
* it does not return NULL when the EOF element is encountered. */
unsigned char *lpSkip(unsigned char *p) {
unsigned long entrylen = lpCurrentEncodedSize(p);
unsigned long entrylen = lpCurrentEncodedSizeUnsafe(p);
entrylen += lpEncodeBacklen(NULL,entrylen);
p += entrylen;
return p;
......@@ -398,8 +438,9 @@ unsigned char *lpSkip(unsigned char *p) {
* the pointer to the next element (the one on the right), or NULL if 'p'
* already pointed to the last element of the listpack. */
unsigned char *lpNext(unsigned char *lp, unsigned char *p) {
((void) lp); /* lp is not used for now. However lpPrev() uses it. */
assert(p);
p = lpSkip(p);
ASSERT_INTEGRITY(lp, p);
if (p[0] == LP_EOF) return NULL;
return p;
}
......@@ -408,11 +449,14 @@ unsigned char *lpNext(unsigned char *lp, unsigned char *p) {
* the pointer to the previous element (the one on the left), or NULL if 'p'
* already pointed to the first element of the listpack. */
unsigned char *lpPrev(unsigned char *lp, unsigned char *p) {
assert(p);
if (p-lp == LP_HDR_SIZE) return NULL;
p--; /* Seek the first backlen byte of the last element. */
uint64_t prevlen = lpDecodeBacklen(p);
prevlen += lpEncodeBacklen(NULL,prevlen);
return p-prevlen+1; /* Seek the first byte of the previous entry. */
p -= prevlen-1; /* Seek the first byte of the previous entry. */
ASSERT_INTEGRITY(lp, p);
return p;
}
/* Return a pointer to the first element of the listpack, or NULL if the
......@@ -570,8 +614,7 @@ unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf) {
/* Insert, delete or replace the specified element 'ele' of length 'len' at
* the specified position 'p', with 'p' being a listpack element pointer
* obtained with lpFirst(), lpLast(), lpIndex(), lpNext(), lpPrev() or
* lpSeek().
* obtained with lpFirst(), lpLast(), lpNext(), lpPrev() or lpSeek().
*
* The element is inserted before, after, or replaces the element pointed
* by 'p' depending on the 'where' argument, that can be LP_BEFORE, LP_AFTER
......@@ -610,6 +653,7 @@ unsigned char *lpInsert(unsigned char *lp, unsigned char *ele, uint32_t size, un
if (where == LP_AFTER) {
p = lpSkip(p);
where = LP_BEFORE;
ASSERT_INTEGRITY(lp, p);
}
/* Store the offset of the element 'p', so that we can obtain its
......@@ -639,8 +683,9 @@ unsigned char *lpInsert(unsigned char *lp, unsigned char *ele, uint32_t size, un
uint64_t old_listpack_bytes = lpGetTotalBytes(lp);
uint32_t replaced_len = 0;
if (where == LP_REPLACE) {
replaced_len = lpCurrentEncodedSize(p);
replaced_len = lpCurrentEncodedSizeUnsafe(p);
replaced_len += lpEncodeBacklen(NULL,replaced_len);
ASSERT_INTEGRITY_LEN(lp, p, replaced_len);
}
uint64_t new_listpack_bytes = old_listpack_bytes + enclen + backlen_size
......@@ -801,3 +846,86 @@ unsigned char *lpSeek(unsigned char *lp, long index) {
}
}
/* Validate the integrity of a single listpack entry and move to the next one.
* The input argument 'pp' is a reference to the current record and is advanced on exit.
* Returns 1 if valid, 0 if invalid. */
int lpValidateNext(unsigned char *lp, unsigned char **pp, size_t lpbytes) {
#define OUT_OF_RANGE(p) ( \
(p) < lp + LP_HDR_SIZE || \
(p) > lp + lpbytes - 1)
unsigned char *p = *pp;
if (!p)
return 0;
if (*p == LP_EOF) {
*pp = NULL;
return 1;
}
/* check that we can read the encoded size */
uint32_t lenbytes = lpCurrentEncodedSizeBytes(p);
if (!lenbytes)
return 0;
/* make sure the encoded entry length doesn't rech outside the edge of the listpack */
if (OUT_OF_RANGE(p + lenbytes))
return 0;
/* get the entry length and encoded backlen. */
unsigned long entrylen = lpCurrentEncodedSizeUnsafe(p);
unsigned long encodedBacklen = lpEncodeBacklen(NULL,entrylen);
entrylen += encodedBacklen;
/* make sure the entry doesn't rech outside the edge of the listpack */
if (OUT_OF_RANGE(p + entrylen))
return 0;
/* move to the next entry */
p += entrylen;
/* make sure the encoded length at the end patches the one at the beginning. */
uint64_t prevlen = lpDecodeBacklen(p-1);
if (prevlen + encodedBacklen != entrylen)
return 0;
*pp = p;
return 1;
#undef OUT_OF_RANGE
}
/* Validate the integrity of the data stracture.
* when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */
int lpValidateIntegrity(unsigned char *lp, size_t size, int deep){
/* Check that we can actually read the header. (and EOF) */
if (size < LP_HDR_SIZE + 1)
return 0;
/* Check that the encoded size in the header must match the allocated size. */
size_t bytes = lpGetTotalBytes(lp);
if (bytes != size)
return 0;
/* The last byte must be the terminator. */
if (lp[size-1] != LP_EOF)
return 0;
if (!deep)
return 1;
/* Validate the invividual entries. */
uint32_t count = 0;
unsigned char *p = lpFirst(lp);
while(p) {
if (!lpValidateNext(lp, &p, bytes))
return 0;
count++;
}
/* Check that the count in the header is correct */
uint32_t numele = lpGetNumElements(lp);
if (numele != LP_HDR_NUMELE_UNKNOWN && numele != count)
return 0;
return 1;
}
......@@ -57,5 +57,7 @@ unsigned char *lpNext(unsigned char *lp, unsigned char *p);
unsigned char *lpPrev(unsigned char *lp, unsigned char *p);
uint32_t lpBytes(unsigned char *lp);
unsigned char *lpSeek(unsigned char *lp, long index);
int lpValidateIntegrity(unsigned char *lp, size_t size, int deep);
int lpValidateNext(unsigned char *lp, unsigned char **pp, size_t lpbytes);
#endif
......@@ -68,17 +68,27 @@ void rdbReportError(int corruption_error, int linenum, char *reason, ...) {
vsnprintf(msg+len,sizeof(msg)-len,reason,ap);
va_end(ap);
if (!rdbCheckMode) {
if (rdbFileBeingLoaded || corruption_error) {
serverLog(LL_WARNING, "%s", msg);
char *argv[2] = {"",rdbFileBeingLoaded};
redis_check_rdb_main(2,argv,NULL);
} else {
serverLog(LL_WARNING, "%s. Failure loading rdb format from socket, assuming connection error, resuming operation.", msg);
return;
}
} else {
if (!server.loading) {
/* If we're in the context of a RESTORE command, just propagate the error. */
/* log in VERBOSE, and return (don't exit). */
serverLog(LL_VERBOSE, "%s", msg);
return;
} else if (rdbCheckMode) {
/* If we're inside the rdb checker, let it handle the error. */
rdbCheckError("%s",msg);
} else if (rdbFileBeingLoaded) {
/* If we're loading an rdb file form disk, run rdb check (and exit) */
serverLog(LL_WARNING, "%s", msg);
char *argv[2] = {"",rdbFileBeingLoaded};
redis_check_rdb_main(2,argv,NULL);
} else if (corruption_error) {
/* In diskless loading, in case of corrupt file, log and exit. */
serverLog(LL_WARNING, "%s. Failure loading rdb format", msg);
} else {
/* In diskless loading, in case of a short read (not a corrupt
* file), log and proceed (don't exit). */
serverLog(LL_WARNING, "%s. Failure loading rdb format from socket, assuming connection error, resuming operation.", msg);
return;
}
serverLog(LL_WARNING, "Terminating server after rdb file reading failure.");
exit(1);
......@@ -1489,6 +1499,17 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) {
uint64_t len;
unsigned int i;
int deep_integrity_validation = server.sanitize_dump_payload == SANITIZE_DUMP_YES;
if (server.sanitize_dump_payload == SANITIZE_DUMP_CLIENTS) {
/* Skip sanitization when loading (an RDB), or getting a RESTORE command
* from either the master or a client using an ACL user with the skip-sanitize-payload flag. */
int skip = server.loading ||
(server.current_client && (server.current_client->flags & CLIENT_MASTER));
if (!skip && server.current_client && server.current_client->user)
skip = !!(server.current_client->user->flags & USER_FLAG_SANITIZE_PAYLOAD_SKIP);
deep_integrity_validation = !skip;
}
if (rdbtype == RDB_TYPE_STRING) {
/* Read string value */
if ((o = rdbLoadEncodedStringObject(rdb)) == NULL) return NULL;
......@@ -1685,12 +1706,20 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) {
server.list_compress_depth);
while (len--) {
size_t encoded_len;
unsigned char *zl =
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL);
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,&encoded_len);
if (zl == NULL) {
decrRefCount(o);
return NULL;
}
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!ziplistValidateIntegrity(zl, encoded_len, deep_integrity_validation)) {
rdbExitReportCorruptRDB("Ziplist integrity check failed.");
decrRefCount(o);
zfree(zl);
return NULL;
}
quicklistAppendZiplist(o->ptr, zl);
}
} else if (rdbtype == RDB_TYPE_HASH_ZIPMAP ||
......@@ -1699,9 +1728,32 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) {
rdbtype == RDB_TYPE_ZSET_ZIPLIST ||
rdbtype == RDB_TYPE_HASH_ZIPLIST)
{
size_t encoded_len;
unsigned char *encoded =
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL);
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,&encoded_len);
if (encoded == NULL) return NULL;
if (rdbtype == RDB_TYPE_HASH_ZIPMAP) {
/* Since we don't keep zipmaps anymore, the rdb loading for these
* is O(n) anyway, use `deep` validation. */
if (!zipmapValidateIntegrity(encoded, encoded_len, 1)) {
rdbExitReportCorruptRDB("Zipmap integrity check failed.");
zfree(encoded);
return NULL;
}
} else if (rdbtype == RDB_TYPE_SET_INTSET) {
if (!intsetValidateIntegrity(encoded, encoded_len)) {
rdbExitReportCorruptRDB("Intset integrity check failed.");
zfree(encoded);
return NULL;
}
} else { /* ziplist */
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!ziplistValidateIntegrity(encoded, encoded_len, deep_integrity_validation)) {
rdbExitReportCorruptRDB("Ziplist integrity check failed.");
zfree(encoded);
return NULL;
}
}
o = createObject(OBJ_STRING,encoded); /* Obj type fixed below. */
/* Fix the object encoding, and make sure to convert the encoded
......@@ -1794,14 +1846,24 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) {
}
/* Load the listpack. */
size_t lp_size;
unsigned char *lp =
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL);
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,&lp_size);
if (lp == NULL) {
rdbReportReadError("Stream listpacks loading failed.");
sdsfree(nodekey);
decrRefCount(o);
return NULL;
}
if (deep_integrity_validation) server.stat_dump_payload_sanitizations++;
if (!streamValidateListpackIntegrity(lp, lp_size, deep_integrity_validation)) {
rdbExitReportCorruptRDB("Stream listpack integrity check failed.");
sdsfree(nodekey);
decrRefCount(o);
zfree(lp);
return NULL;
}
unsigned char *first = lpFirst(lp);
if (first == NULL) {
/* Serialized listpacks should never be empty, since on
......@@ -2389,6 +2451,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
(unsigned long long)expected,
(unsigned long long)cksum);
rdbExitReportCorruptRDB("RDB CRC error");
return C_ERR;
}
}
}
......
......@@ -372,6 +372,7 @@ int redis_check_rdb_main(int argc, char **argv, FILE *fp) {
if (shared.integers[0] == NULL)
createSharedObjects();
server.loading_process_events_interval_bytes = 0;
server.sanitize_dump_payload = SANITIZE_DUMP_YES;
rdbCheckMode = 1;
rdbCheckInfo("Checking RDB file %s", argv[1]);
rdbCheckSetupSignals();
......
......@@ -2934,6 +2934,7 @@ void resetServerStats(void) {
atomicSet(server.stat_net_input_bytes, 0);
atomicSet(server.stat_net_output_bytes, 0);
server.stat_unexpected_error_replies = 0;
server.stat_dump_payload_sanitizations = 0;
server.aof_delayed_fsync = 0;
server.blocked_last_cron = 0;
}
......@@ -4654,6 +4655,7 @@ sds genRedisInfoString(const char *section) {
"tracking_total_items:%lld\r\n"
"tracking_total_prefixes:%lld\r\n"
"unexpected_error_replies:%lld\r\n"
"dump_payload_sanitizations:%lld\r\n"
"total_reads_processed:%lld\r\n"
"total_writes_processed:%lld\r\n"
"io_threaded_reads_processed:%lld\r\n"
......@@ -4689,6 +4691,7 @@ sds genRedisInfoString(const char *section) {
(unsigned long long) trackingGetTotalItems(),
(unsigned long long) trackingGetTotalPrefixes(),
server.stat_unexpected_error_replies,
server.stat_dump_payload_sanitizations,
stat_total_reads_processed,
stat_total_writes_processed,
server.stat_io_reads_processed,
......
......@@ -378,6 +378,11 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
#define TLS_CLIENT_AUTH_YES 1
#define TLS_CLIENT_AUTH_OPTIONAL 2
/* Sanitize dump payload */
#define SANITIZE_DUMP_NO 0
#define SANITIZE_DUMP_YES 1
#define SANITIZE_DUMP_CLIENTS 2
/* Sets operations codes */
#define SET_OP_UNION 0
#define SET_OP_DIFF 1
......@@ -781,6 +786,12 @@ typedef struct readyList {
authenticated. */
#define USER_FLAG_ALLCHANNELS (1<<5) /* The user can mention any Pub/Sub
channel. */
#define USER_FLAG_SANITIZE_PAYLOAD (1<<6) /* The user require a deep RESTORE
* payload sanitization. */
#define USER_FLAG_SANITIZE_PAYLOAD_SKIP (1<<7) /* The user should skip the
* deep sanitization of RESTORE
* payload. */
typedef struct {
sds name; /* The username as an SDS string. */
uint64_t flags; /* See USER_FLAG_* */
......@@ -1213,6 +1224,7 @@ struct redisServer {
size_t stat_module_cow_bytes; /* Copy on write bytes during module fork. */
uint64_t stat_clients_type_memory[CLIENT_TYPE_COUNT];/* Mem usage by type */
long long stat_unexpected_error_replies; /* Number of unexpected (aof-loading, replica to master, etc.) error replies */
long long stat_dump_payload_sanitizations; /* Number deep dump payloads integrity validations. */
long long stat_io_reads_processed; /* Number of read events processed by IO / Main threads */
long long stat_io_writes_processed; /* Number of write events processed by IO / Main threads */
redisAtomic long long stat_total_reads_processed; /* Total number of read events processed */
......@@ -1232,6 +1244,7 @@ struct redisServer {
int active_expire_enabled; /* Can be disabled for testing purposes. */
int active_expire_effort; /* From 1 (default) to 10, active effort. */
int active_defrag_enabled;
int sanitize_dump_payload; /* Enables deep sanitization for ziplist and listpack in RDB and RESTORE. */
int jemalloc_bg_thread; /* Enable jemalloc background thread */
size_t active_defrag_ignore_bytes; /* minimum amount of fragmentation waste to start active defrag */
int active_defrag_threshold_lower; /* minimum percentage of fragmentation to start active defrag */
......
......@@ -120,5 +120,6 @@ int streamIncrID(streamID *id);
int streamDecrID(streamID *id);
void streamPropagateConsumerCreation(client *c, robj *key, robj *groupname, sds consumername);
robj *streamDup(robj *o);
int streamValidateListpackIntegrity(unsigned char *lp, size_t size, int deep);
#endif
......@@ -67,7 +67,7 @@ int hashTypeGetFromZiplist(robj *o, sds field,
zl = o->ptr;
fptr = ziplistIndex(zl, ZIPLIST_HEAD);
if (fptr != NULL) {
fptr = ziplistFind(fptr, (unsigned char*)field, sdslen(field), 1);
fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
if (fptr != NULL) {
/* Grab pointer to the value (fptr points to the field) */
vptr = ziplistNext(zl, fptr);
......@@ -208,7 +208,7 @@ int hashTypeSet(robj *o, sds field, sds value, int flags) {
zl = o->ptr;
fptr = ziplistIndex(zl, ZIPLIST_HEAD);
if (fptr != NULL) {
fptr = ziplistFind(fptr, (unsigned char*)field, sdslen(field), 1);
fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
if (fptr != NULL) {
/* Grab pointer to the value (fptr points to the field) */
vptr = ziplistNext(zl, fptr);
......@@ -285,7 +285,7 @@ int hashTypeDelete(robj *o, sds field) {
zl = o->ptr;
fptr = ziplistIndex(zl, ZIPLIST_HEAD);
if (fptr != NULL) {
fptr = ziplistFind(fptr, (unsigned char*)field, sdslen(field), 1);
fptr = ziplistFind(zl, fptr, (unsigned char*)field, sdslen(field), 1);
if (fptr != NULL) {
zl = ziplistDelete(zl,&fptr); /* Delete the key. */
zl = ziplistDelete(zl,&fptr); /* Delete the value. */
......
......@@ -255,21 +255,33 @@ unsigned char *lpReplaceInteger(unsigned char *lp, unsigned char **pos, int64_t
/* This is a wrapper function for lpGet() to directly get an integer value
* from the listpack (that may store numbers as a string), converting
* the string if needed. */
int64_t lpGetInteger(unsigned char *ele) {
* the string if needed.
* The 'valid" argument is an optional output parameter to get an indication
* if the record was valid, when this parameter is NULL, the function will
* fail with an assertion. */
static inline int64_t lpGetIntegerIfValid(unsigned char *ele, int *valid) {
int64_t v;
unsigned char *e = lpGet(ele,&v,NULL);
if (e == NULL) return v;
if (e == NULL) {
if (valid)
*valid = 1;
return v;
}
/* The following code path should never be used for how listpacks work:
* they should always be able to store an int64_t value in integer
* encoded form. However the implementation may change. */
long long ll;
int retval = string2ll((char*)e,v,&ll);
serverAssert(retval != 0);
int ret = string2ll((char*)e,v,&ll);
if (valid)
*valid = ret;
else
serverAssert(ret != 0);
v = ll;
return v;
}
#define lpGetInteger(ele) lpGetIntegerIfValid(ele, NULL)
/* Debugging function to log the full content of a listpack. Useful
* for development and debugging. */
void streamLogListpackContent(unsigned char *lp) {
......@@ -3037,3 +3049,90 @@ NULL
addReplySubcommandSyntaxError(c);
}
}
/* Validate the integrity stream listpack entries stracture. Both in term of a
* valid listpack, but also that the stracture of the entires matches a valid
* stream. return 1 if valid 0 if not valid. */
int streamValidateListpackIntegrity(unsigned char *lp, size_t size, int deep) {
int valid_record;
unsigned char *p, *next;
/* Since we don't want to run validation of all records twice, we'll
* run the listpack validation of just the header and do the rest here. */
if (!lpValidateIntegrity(lp, size, 0))
return 0;
/* In non-deep mode we just validated the listpack header (encoded size) */
if (!deep) return 1;
next = p = lpFirst(lp);
if (!lpValidateNext(lp, &next, size)) return 0;
if (!p) return 0;
/* entry count */
int64_t entry_count = lpGetIntegerIfValid(p, &valid_record);
if (!valid_record) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
/* deleted */
lpGetIntegerIfValid(p, &valid_record);
if (!valid_record) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
/* num-of-fields */
int64_t master_fields = lpGetIntegerIfValid(p, &valid_record);
if (!valid_record) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
/* the field names */
for (int64_t j = 0; j < master_fields; j++) {
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
}
/* the zero master entry terminator. */
int64_t zero = lpGetIntegerIfValid(p, &valid_record);
if (!valid_record || zero != 0) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
while (entry_count--) {
if (!p) return 0;
int64_t fields = master_fields, extra_fields = 3;
int64_t flags = lpGetIntegerIfValid(p, &valid_record);
if (!valid_record) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
/* entry id */
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
if (!(flags & STREAM_ITEM_FLAG_SAMEFIELDS)) {
/* num-of-fields */
fields = lpGetIntegerIfValid(p, &valid_record);
if (!valid_record) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
/* the field names */
for (int64_t j = 0; j < fields; j++) {
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
}
extra_fields += fields + 1;
}
/* the values */
for (int64_t j = 0; j < fields; j++) {
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
}
/* lp-count */
int64_t lp_count = lpGetIntegerIfValid(p, &valid_record);
if (!valid_record) return 0;
if (lp_count != fields + extra_fields) return 0;
p = next; if (!lpValidateNext(lp, &next, size)) return 0;
}
if (next)
return 0;
return 1;
}
This diff is collapsed.
......@@ -45,10 +45,11 @@ unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p);
unsigned char *ziplistDeleteRange(unsigned char *zl, int index, unsigned int num);
unsigned int ziplistCompare(unsigned char *p, unsigned char *s, unsigned int slen);
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip);
unsigned char *ziplistFind(unsigned char *zl, unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip);
unsigned int ziplistLen(unsigned char *zl);
size_t ziplistBlobLen(unsigned char *zl);
void ziplistRepr(unsigned char *zl);
int ziplistValidateIntegrity(unsigned char *zl, size_t size, int deep);
#ifdef REDIS_TEST
int ziplistTest(int argc, char *argv[]);
......
......@@ -111,6 +111,10 @@ static unsigned int zipmapDecodeLength(unsigned char *p) {
return len;
}
static unsigned int zipmapGetEncodedLengthSize(unsigned char *p) {
return (*p < ZIPMAP_BIGLEN) ? 1: 5;
}
/* Encode the length 'l' writing it in 'p'. If p is NULL it just returns
* the amount of bytes required to encode such a length. */
static unsigned int zipmapEncodeLength(unsigned char *p, unsigned int len) {
......@@ -370,6 +374,70 @@ size_t zipmapBlobLen(unsigned char *zm) {
return totlen;
}
/* Validate the integrity of the data stracture.
* when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */
int zipmapValidateIntegrity(unsigned char *zm, size_t size, int deep) {
#define OUT_OF_RANGE(p) ( \
(p) < zm + 2 || \
(p) > zm + size - 1)
unsigned int l, s, e;
/* check that we can actually read the header (or ZIPMAP_END). */
if (size < 2)
return 0;
/* the last byte must be the terminator. */
if (zm[size-1] != ZIPMAP_END)
return 0;
if (!deep)
return 1;
unsigned int count = 0;
unsigned char *p = zm + 1; /* skip the count */
while(*p != ZIPMAP_END) {
/* read the field name length encoding type */
s = zipmapGetEncodedLengthSize(p);
/* make sure the entry length doesn't rech outside the edge of the zipmap */
if (OUT_OF_RANGE(p+s))
return 0;
/* read the field name length */
l = zipmapDecodeLength(p);
p += s; /* skip the encoded field size */
p += l; /* skip the field */
/* make sure the entry doesn't rech outside the edge of the zipmap */
if (OUT_OF_RANGE(p))
return 0;
/* read the value length encoding type */
s = zipmapGetEncodedLengthSize(p);
/* make sure the entry length doesn't rech outside the edge of the zipmap */
if (OUT_OF_RANGE(p+s))
return 0;
/* read the value length */
l = zipmapDecodeLength(p);
p += s; /* skip the encoded value size*/
e = *p++; /* skip the encoded free space (always encoded in one byte) */
p += l+e; /* skip the value and free space */
count++;
/* make sure the entry doesn't rech outside the edge of the zipmap */
if (OUT_OF_RANGE(p))
return 0;
}
/* check that the count in the header is correct */
if (zm[0] != ZIPMAP_BIGLEN && zm[0] != count)
return 0;
return 1;
#undef OUT_OF_RANGE
}
#ifdef REDIS_TEST
static void zipmapRepr(unsigned char *p) {
unsigned int l;
......
......@@ -45,6 +45,7 @@ int zipmapExists(unsigned char *zm, unsigned char *key, unsigned int klen);
unsigned int zipmapLen(unsigned char *zm);
size_t zipmapBlobLen(unsigned char *zm);
void zipmapRepr(unsigned char *p);
int zipmapValidateIntegrity(unsigned char *zm, size_t size, int deep);
#ifdef REDIS_TEST
int zipmapTest(int argc, char *argv[]);
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment