#include <worker_util.h>
-#ifdef SYS_ZSTD
-#include "zstd.h"
-#else
-#include "contrib/zstd/zstd.h"
-#endif
+#include "libutil/compression.h"
#undef MAP_DEBUG_REFS
#ifdef MAP_DEBUG_REFS
/* Prepare payload: decrypt (if needed) then optionally decompress */
unsigned char *payload = NULL;
gsize payload_len = 0;
- unsigned char *final_out = NULL;
if (cbd->bk->is_encrypted) {
if (!rspamd_map_secretbox_decrypt_buf(cbd->bk, in, dlen, &payload, &payload_len)) {
/* If compressed flag is set OR payload looks like zstd, decompress */
if (cbd->bk->is_compressed || rspamd_map_payload_is_zstd(payload, payload_len)) {
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
- gsize outlen, r;
gsize max_size = rspamd_map_effective_max_size(cbd);
+ GError *derr = NULL;
+ rspamd_fstring_t *decompressed;
- zin.pos = 0;
- zin.src = payload;
- zin.size = payload_len;
+ decompressed = rspamd_zstd_decompress_bounded(NULL, payload, payload_len,
+ max_size, &derr);
- if ((outlen = ZSTD_getDecompressedSize(zin.src, zin.size)) == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
- else if (max_size > 0 && outlen > max_size) {
- msg_err_map("%s(%s): declared decompressed size %z exceeds max_map_size %z",
+ if (decompressed == NULL) {
+ msg_err_map("%s(%s): cannot decompress data: %s",
cbd->bk->uri,
rspamd_inet_address_to_string_pretty(cbd->addr),
- outlen, max_size);
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
if (cbd->bk->is_encrypted && payload && payload != (unsigned char *) in) {
rspamd_explicit_memzero(payload, payload_len);
g_free(payload);
goto err;
}
- zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
-
- final_out = g_malloc(outlen);
-
- zout.dst = final_out;
- zout.pos = 0;
- zout.size = outlen;
-
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
-
- if (ZSTD_isError(r)) {
- msg_err_map("%s(%s): cannot decompress data: %s",
- cbd->bk->uri,
- rspamd_inet_address_to_string_pretty(cbd->addr),
- ZSTD_getErrorName(r));
- ZSTD_freeDStream(zstream);
- g_free(final_out);
- if (cbd->bk->is_encrypted && payload && payload != (unsigned char *) in) {
- rspamd_explicit_memzero(payload, payload_len);
- g_free(payload);
- }
- MAP_RELEASE(cbd->shmem_data, "shmem_data");
- goto err;
- }
-
- if (zout.pos == zout.size) {
- /* We need to extend output buffer */
- if (max_size > 0 && zout.size >= max_size) {
- msg_err_map("%s(%s): decompressed data exceeds max_map_size %z",
- cbd->bk->uri,
- rspamd_inet_address_to_string_pretty(cbd->addr),
- max_size);
- ZSTD_freeDStream(zstream);
- g_free(final_out);
- if (cbd->bk->is_encrypted && payload && payload != (unsigned char *) in) {
- rspamd_explicit_memzero(payload, payload_len);
- g_free(payload);
- }
- MAP_RELEASE(cbd->shmem_data, "shmem_data");
- goto err;
- }
-
- zout.size = zout.size * 2 + 1.0;
-
- if (max_size > 0 && zout.size > max_size) {
- zout.size = max_size;
- }
-
- final_out = g_realloc(zout.dst, zout.size);
- zout.dst = final_out;
- }
- }
-
- ZSTD_freeDStream(zstream);
msg_info_map("%s(%s): read map data %z bytes compressed, "
"%z uncompressed, next check at %s",
cbd->bk->uri,
rspamd_inet_address_to_string_pretty(cbd->addr),
- payload_len, zout.pos, next_check_date);
- if (!rspamd_map_save_http_cached_file(map, bk, cbd->data, final_out, zout.pos)) {
+ payload_len, decompressed->len, next_check_date);
+ if (!rspamd_map_save_http_cached_file(map, bk, cbd->data,
+ (const unsigned char *) decompressed->str,
+ decompressed->len)) {
msg_err_map("%s: failed to save cache file", bk->uri);
- g_free(final_out);
+ rspamd_fstring_free(decompressed);
MAP_RELEASE(cbd->shmem_data, "shmem_data");
goto err;
}
&cbd->periodic->cbdata, TRUE);
}
else {
- map->read_callback(final_out, zout.pos, &cbd->periodic->cbdata, TRUE);
+ map->read_callback(decompressed->str, decompressed->len,
+ &cbd->periodic->cbdata, TRUE);
}
- g_free(final_out);
+ rspamd_fstring_free(decompressed);
}
else {
msg_info_map("%s(%s): read map data %z bytes, next check at %s",
/* If compressed flag is set OR payload looks like zstd, decompress */
if (bk->is_compressed || rspamd_map_payload_is_zstd((const unsigned char *) payload, payload_len)) {
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
- unsigned char *out;
- gsize outlen, r;
-
- zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
-
- zin.pos = 0;
- zin.src = payload;
- zin.size = payload_len;
-
- if ((outlen = ZSTD_getDecompressedSize(zin.src, zin.size)) == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
-
- out = g_malloc(outlen);
-
- zout.dst = out;
- zout.pos = 0;
- zout.size = outlen;
-
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
-
- if (ZSTD_isError(r)) {
- msg_err_map("%s: cannot decompress data: %s",
- data->filename,
- ZSTD_getErrorName(r));
- ZSTD_freeDStream(zstream);
- g_free(out);
- if (dec) {
- rspamd_explicit_memzero(dec, declen);
- g_free(dec);
- }
- munmap(bytes, len);
- return FALSE;
- }
-
- if (zout.pos == zout.size) {
- /* We need to extend output buffer */
- zout.size = zout.size * 2 + 1;
- out = g_realloc(zout.dst, zout.size);
- zout.dst = out;
+ GError *derr = NULL;
+ rspamd_fstring_t *decompressed;
+
+ decompressed = rspamd_zstd_decompress_bounded(NULL, payload, payload_len,
+ map->cfg ? map->cfg->max_map_size : 0,
+ &derr);
+
+ if (decompressed == NULL) {
+ msg_err_map("%s: cannot decompress data: %s",
+ data->filename,
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
+ if (dec) {
+ rspamd_explicit_memzero(dec, declen);
+ g_free(dec);
}
+ munmap(bytes, len);
+ return FALSE;
}
- ZSTD_freeDStream(zstream);
msg_info_map("%s: read map data, %z bytes compressed, "
"%z uncompressed)",
data->filename,
- payload_len, zout.pos);
- map->read_callback(out, zout.pos, &periodic->cbdata, TRUE);
- g_free(out);
+ payload_len, decompressed->len);
+ map->read_callback(decompressed->str, decompressed->len,
+ &periodic->cbdata, TRUE);
+ rspamd_fstring_free(decompressed);
if (dec) {
rspamd_explicit_memzero(dec, declen);
if (len > 0) {
if (bk->is_compressed) {
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
- unsigned char *out;
- gsize outlen, r;
-
- zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
-
- zin.pos = 0;
- zin.src = bytes;
- zin.size = len;
-
- if ((outlen = ZSTD_getDecompressedSize(zin.src, zin.size)) == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
-
- out = g_malloc(outlen);
-
- zout.dst = out;
- zout.pos = 0;
- zout.size = outlen;
+ GError *derr = NULL;
+ rspamd_fstring_t *decompressed;
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
+ decompressed = rspamd_zstd_decompress_bounded(NULL, bytes, len,
+ map->cfg ? map->cfg->max_map_size : 0,
+ &derr);
- if (ZSTD_isError(r)) {
- msg_err_map("%s: cannot decompress data: %s",
- map->name,
- ZSTD_getErrorName(r));
- ZSTD_freeDStream(zstream);
- g_free(out);
+ if (decompressed == NULL) {
+ msg_err_map("%s: cannot decompress data: %s",
+ map->name,
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
- return FALSE;
- }
-
- if (zout.pos == zout.size) {
- /* We need to extend output buffer */
- zout.size = zout.size * 2 + 1;
- out = g_realloc(zout.dst, zout.size);
- zout.dst = out;
- }
+ return FALSE;
}
- ZSTD_freeDStream(zstream);
msg_info_map("%s: read map data, %z bytes compressed, "
"%z uncompressed)",
map->name,
- len, zout.pos);
- map->read_callback(out, zout.pos, &periodic->cbdata, TRUE);
- g_free(out);
+ len, decompressed->len);
+ map->read_callback(decompressed->str, decompressed->len,
+ &periodic->cbdata, TRUE);
+ rspamd_fstring_free(decompressed);
}
else {
msg_info_map("%s: read map data, %z bytes",
/* If compressed flag is set OR payload looks like zstd, decompress */
if (bk->is_compressed || rspamd_map_payload_is_zstd(payload, payload_len)) {
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
- unsigned char *out;
- gsize outlen, r;
-
- zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
-
- zin.pos = 0;
- zin.src = payload;
- zin.size = payload_len;
-
- if ((outlen = ZSTD_getDecompressedSize(zin.src, zin.size)) == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
-
- out = g_malloc(outlen);
-
- zout.dst = out;
- zout.pos = 0;
- zout.size = outlen;
-
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
-
- if (ZSTD_isError(r)) {
- msg_err_map("%s: cannot decompress data: %s",
- bk->uri,
- ZSTD_getErrorName(r));
- ZSTD_freeDStream(zstream);
- g_free(out);
- if (dec) {
- rspamd_explicit_memzero(dec, declen);
- g_free(dec);
- }
- munmap(in, mmap_len);
- return FALSE;
- }
-
- if (zout.pos == zout.size) {
- /* We need to extend output buffer */
- zout.size = zout.size * 2 + 1;
- out = g_realloc(zout.dst, zout.size);
- zout.dst = out;
+ GError *derr = NULL;
+ rspamd_fstring_t *decompressed;
+
+ decompressed = rspamd_zstd_decompress_bounded(NULL, payload, payload_len,
+ map->cfg ? map->cfg->max_map_size : 0,
+ &derr);
+
+ if (decompressed == NULL) {
+ msg_err_map("%s: cannot decompress data: %s",
+ bk->uri,
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
+ if (dec) {
+ rspamd_explicit_memzero(dec, declen);
+ g_free(dec);
}
+ munmap(in, mmap_len);
+ return FALSE;
}
- ZSTD_freeDStream(zstream);
msg_info_map("%s: read map data cached %z bytes compressed, "
"%z uncompressed",
bk->uri,
- payload_len, zout.pos);
- map->read_callback(out, zout.pos, &periodic->cbdata, TRUE);
- g_free(out);
+ payload_len, decompressed->len);
+ map->read_callback(decompressed->str, decompressed->len,
+ &periodic->cbdata, TRUE);
+ rspamd_fstring_free(decompressed);
if (dec) {
rspamd_explicit_memzero(dec, declen);
g_free(dec);
#include "libmime/content_type.h"
#include <math.h>
-#ifdef SYS_ZSTD
-#include "zstd.h"
-#else
-#include "contrib/zstd/zstd.h"
-#endif
+#include "libutil/compression.h"
INIT_LOG_MODULE(protocol)
msg_part->content_encoding_len,
"zstd", 4) != -1) {
/* Decompress message */
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
- gsize outlen, r;
+ GError *derr = NULL;
+ rspamd_fstring_t *decompressed;
if (!rspamd_libs_reset_decompression(task->cfg->libs_ctx)) {
g_set_error(&task->err, rspamd_protocol_quark(), 500,
return FALSE;
}
- zstream = task->cfg->libs_ctx->in_zstream;
- zin.src = msg_part->data;
- zin.size = msg_part->data_len;
- zin.pos = 0;
-
- outlen = ZSTD_getDecompressedSize(msg_part->data, msg_part->data_len);
- if (outlen == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
- else if (task->cfg->max_message > 0 && outlen > task->cfg->max_message) {
- g_set_error(&task->err, rspamd_protocol_quark(), 413,
- "decompressed message exceeds max_message limit: %lu > %lu",
- (unsigned long) outlen, (unsigned long) task->cfg->max_message);
- return FALSE;
- }
-
- unsigned char *out = (unsigned char *) g_malloc(outlen);
- zout.dst = out;
- zout.pos = 0;
- zout.size = outlen;
+ decompressed = rspamd_zstd_decompress_bounded(task->cfg->libs_ctx->in_zstream,
+ msg_part->data, msg_part->data_len,
+ task->cfg->max_message, &derr);
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
+ if (decompressed == NULL) {
+ int http_code = 400;
- if (ZSTD_isError(r)) {
- g_set_error(&task->err, rspamd_protocol_quark(), 400,
- "message decompression error: %s",
- ZSTD_getErrorName(r));
- g_free(out);
- return FALSE;
+ if (derr != NULL && derr->code == RSPAMD_DECOMPRESS_ERROR_TOO_LARGE) {
+ http_code = 413;
}
- if (zout.pos == zout.size) {
- if (task->cfg->max_message > 0 &&
- zout.size > task->cfg->max_message) {
- g_set_error(&task->err, rspamd_protocol_quark(), 413,
- "decompressed message exceeds max_message limit: %lu > %lu",
- (unsigned long) zout.size, (unsigned long) task->cfg->max_message);
- g_free(out);
- return FALSE;
- }
- zout.size = zout.size * 2 + 1;
- out = g_realloc(zout.dst, zout.size);
- zout.dst = out;
- }
+ g_set_error(&task->err, rspamd_protocol_quark(), http_code,
+ "message decompression error: %s",
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
+ return FALSE;
}
- rspamd_mempool_add_destructor(task->task_pool, g_free, zout.dst);
- task->msg.begin = (const char *) zout.dst;
- task->msg.len = zout.pos;
+ rspamd_mempool_add_destructor(task->task_pool,
+ (rspamd_mempool_destruct_t) rspamd_fstring_free,
+ decompressed);
+ task->msg.begin = decompressed->str;
+ task->msg.len = decompressed->len;
task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_COMPRESSED;
msg_info_protocol("v3: loaded message from zstd compressed part; "
"compressed: %ul; uncompressed: %ul",
- (gulong) zin.size, (gulong) zout.pos);
+ (gulong) msg_part->data_len, (gulong) decompressed->len);
}
else {
/* Zero-copy: point directly into the multipart buffer */
#include <math.h>
-#ifdef SYS_ZSTD
-#include "zstd.h"
-#else
-#include "contrib/zstd/zstd.h"
-#endif
+#include "libutil/compression.h"
__KHASH_IMPL(rspamd_req_headers_hash, static inline,
rspamd_ftok_t *, struct rspamd_request_header_chain *, 1,
t.len = 4;
if (rspamd_ftok_casecmp(tok, &t) == 0) {
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
- unsigned char *out;
- gsize outlen, r;
gulong dict_id;
if (!rspamd_libs_reset_decompression(task->cfg->libs_ctx)) {
}
}
- zstream = task->cfg->libs_ctx->in_zstream;
+ GError *derr = NULL;
+ gsize compressed_len = task->msg.len;
+ rspamd_fstring_t *decompressed;
- zin.pos = 0;
- zin.src = task->msg.begin;
- zin.size = task->msg.len;
+ decompressed = rspamd_zstd_decompress_bounded(task->cfg->libs_ctx->in_zstream,
+ task->msg.begin, task->msg.len,
+ task->cfg->max_message, &derr);
- if ((outlen = ZSTD_getDecompressedSize(task->msg.begin, task->msg.len)) == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
- else if (task->cfg->max_message > 0 && outlen > task->cfg->max_message) {
+ if (decompressed == NULL) {
g_set_error(&task->err, rspamd_task_quark(),
RSPAMD_PROTOCOL_ERROR,
- "decompressed message exceeds max_message limit: %lu > %lu",
- (unsigned long) outlen, (unsigned long) task->cfg->max_message);
+ "Decompression error: %s",
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
return FALSE;
}
- out = g_malloc(outlen);
- zout.dst = out;
- zout.pos = 0;
- zout.size = outlen;
-
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
-
- if (ZSTD_isError(r)) {
- g_set_error(&task->err, rspamd_task_quark(),
- RSPAMD_PROTOCOL_ERROR,
- "Decompression error: %s", ZSTD_getErrorName(r));
- g_free(zout.dst);
-
- return FALSE;
- }
-
- if (zout.pos == zout.size) {
- /* We need to extend output buffer */
- if (task->cfg->max_message > 0 &&
- zout.size > task->cfg->max_message) {
- g_set_error(&task->err, rspamd_task_quark(),
- RSPAMD_PROTOCOL_ERROR,
- "decompressed message exceeds max_message limit: %lu > %lu",
- (unsigned long) zout.size, (unsigned long) task->cfg->max_message);
- g_free(zout.dst);
-
- return FALSE;
- }
-
- zout.size = zout.size * 2 + 1;
- zout.dst = g_realloc(zout.dst, zout.size);
- }
- }
-
- rspamd_mempool_add_destructor(task->task_pool, g_free, zout.dst);
- task->msg.begin = zout.dst;
- task->msg.len = zout.pos;
+ rspamd_mempool_add_destructor(task->task_pool,
+ (rspamd_mempool_destruct_t) rspamd_fstring_free,
+ decompressed);
+ task->msg.begin = decompressed->str;
+ task->msg.len = decompressed->len;
task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_COMPRESSED;
msg_info_task("loaded message from zstd compressed stream; "
"compressed: %ul; uncompressed: %ul",
- (gulong) zin.size, (gulong) zout.pos);
+ (gulong) compressed_len, (gulong) decompressed->len);
}
else {
g_set_error(&task->err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
# Librspamd-util
SET(LIBRSPAMDUTILSRC
${CMAKE_CURRENT_SOURCE_DIR}/addr.c
+ ${CMAKE_CURRENT_SOURCE_DIR}/compression.c
${CMAKE_CURRENT_SOURCE_DIR}/libev_helper.c
${CMAKE_CURRENT_SOURCE_DIR}/expression.c
${CMAKE_CURRENT_SOURCE_DIR}/fstring.c
--- /dev/null
+/*
+ * Copyright 2026 Vsevolod Stakhov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "config.h"
+#include "compression.h"
+
+GQuark
+rspamd_decompress_quark(void)
+{
+ return g_quark_from_static_string("rspamd-decompress");
+}
+
+rspamd_fstring_t *
+rspamd_zstd_decompress_bounded(ZSTD_DStream *zstream,
+ const void *in, gsize inlen,
+ gsize max_out, GError **err)
+{
+ ZSTD_DStream *own_stream = NULL;
+ ZSTD_inBuffer zin;
+ ZSTD_outBuffer zout;
+ rspamd_fstring_t *body;
+ gsize outlen, r;
+
+ if (zstream == NULL) {
+ own_stream = ZSTD_createDStream();
+
+ if (own_stream == NULL || ZSTD_isError(ZSTD_initDStream(own_stream))) {
+ g_set_error(err, rspamd_decompress_quark(),
+ RSPAMD_DECOMPRESS_ERROR_INIT,
+ "cannot initialize zstd decompressor");
+ if (own_stream != NULL) {
+ ZSTD_freeDStream(own_stream);
+ }
+
+ return NULL;
+ }
+
+ zstream = own_stream;
+ }
+
+ zin.src = in;
+ zin.size = inlen;
+ zin.pos = 0;
+
+ outlen = ZSTD_getDecompressedSize(in, inlen);
+
+ if (outlen == 0) {
+ outlen = ZSTD_DStreamOutSize();
+
+ if (max_out > 0 && outlen > max_out) {
+ outlen = max_out;
+ }
+ }
+ else if (max_out > 0 && outlen > max_out) {
+ g_set_error(err, rspamd_decompress_quark(),
+ RSPAMD_DECOMPRESS_ERROR_TOO_LARGE,
+ "declared decompressed size %" G_GSIZE_FORMAT
+ " exceeds limit %" G_GSIZE_FORMAT,
+ outlen, max_out);
+ if (own_stream != NULL) {
+ ZSTD_freeDStream(own_stream);
+ }
+
+ return NULL;
+ }
+
+ body = rspamd_fstring_sized_new(outlen);
+ zout.dst = body->str;
+ zout.pos = 0;
+ zout.size = body->allocated;
+
+ if (max_out > 0 && zout.size > max_out) {
+ zout.size = max_out;
+ }
+
+ for (;;) {
+ r = ZSTD_decompressStream(zstream, &zout, &zin);
+
+ if (ZSTD_isError(r)) {
+ g_set_error(err, rspamd_decompress_quark(),
+ RSPAMD_DECOMPRESS_ERROR_DATA,
+ "decompression error: %s", ZSTD_getErrorName(r));
+ rspamd_fstring_free(body);
+ if (own_stream != NULL) {
+ ZSTD_freeDStream(own_stream);
+ }
+
+ return NULL;
+ }
+
+ if (zin.pos == zin.size && r == 0) {
+ /* All input consumed and the last frame is complete */
+ break;
+ }
+
+ if (zout.pos == zout.size) {
+ /* Need a larger output buffer */
+ if (max_out > 0 && zout.size >= max_out) {
+ g_set_error(err, rspamd_decompress_quark(),
+ RSPAMD_DECOMPRESS_ERROR_TOO_LARGE,
+ "decompressed size exceeds limit %" G_GSIZE_FORMAT,
+ max_out);
+ rspamd_fstring_free(body);
+ if (own_stream != NULL) {
+ ZSTD_freeDStream(own_stream);
+ }
+
+ return NULL;
+ }
+
+ body = rspamd_fstring_grow(body, zout.size + 1);
+ zout.dst = body->str;
+ zout.size = body->allocated;
+
+ if (max_out > 0 && zout.size > max_out) {
+ zout.size = max_out;
+ }
+ }
+ else if (zin.pos == zin.size) {
+ /*
+ * Input is exhausted mid-frame and the decoder has nothing more
+ * to flush: truncated input; keep the partial output as the
+ * legacy per-caller loops did
+ */
+ break;
+ }
+ }
+
+ body->len = zout.pos;
+
+ if (own_stream != NULL) {
+ ZSTD_freeDStream(own_stream);
+ }
+
+ return body;
+}
--- /dev/null
+/*
+ * Copyright 2026 Vsevolod Stakhov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef RSPAMD_COMPRESSION_H
+#define RSPAMD_COMPRESSION_H
+
+#include "config.h"
+#include "fstring.h"
+
+#ifdef SYS_ZSTD
+#include "zstd.h"
+#else
+#include "contrib/zstd/zstd.h"
+#endif
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+enum rspamd_decompress_error {
+ RSPAMD_DECOMPRESS_ERROR_INIT = 0,
+ RSPAMD_DECOMPRESS_ERROR_DATA,
+ RSPAMD_DECOMPRESS_ERROR_TOO_LARGE,
+};
+
+GQuark rspamd_decompress_quark(void);
+
+/**
+ * Decompress a zstd payload enforcing a hard ceiling on the output size.
+ *
+ * The frame-advertised decompressed size is validated before any allocation,
+ * the output buffer never grows beyond `max_out` and pending decoder output is
+ * drained after the input is consumed, so the returned length is always within
+ * the limit.
+ *
+ * @param zstream stream to use; its state is not reset by this function.
+ * Pass NULL to use a temporary stream created and freed internally
+ * @param in compressed input
+ * @param inlen input length
+ * @param max_out maximum allowed decompressed size, 0 means unlimited
+ * @param err error output (enum rspamd_decompress_error codes)
+ * @return decompressed payload to be freed with rspamd_fstring_free,
+ * or NULL on error
+ */
+rspamd_fstring_t *rspamd_zstd_decompress_bounded(ZSTD_DStream *zstream,
+ const void *in, gsize inlen,
+ gsize max_out,
+ GError **err)
+ G_GNUC_WARN_UNUSED_RESULT;
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /* RSPAMD_COMPRESSION_H */
#include <netinet/tcp.h> /* for TCP_NODELAY */
#endif
-#ifdef SYS_ZSTD
-#include "zstd.h"
-#else
-#include "contrib/zstd/zstd.h"
-#endif
+#include "libutil/compression.h"
/* Rotate keys each minute by default */
#define DEFAULT_ROTATION_TIME 60.0
const char *result_data = result_part->data;
gsize result_len = result_part->data_len;
- unsigned char *decompressed = NULL;
+ rspamd_fstring_t *decompressed = NULL;
/* Check for per-part zstd compression */
if (result_part->content_encoding &&
rspamd_substring_search_caseless(result_part->content_encoding,
result_part->content_encoding_len,
"zstd", 4) != -1) {
- ZSTD_DStream *zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
- ZSTD_inBuffer zin = {result_data, result_len, 0};
- gsize outlen = ZSTD_getDecompressedSize(result_data, result_len);
- if (outlen == 0) outlen = ZSTD_DStreamOutSize();
- decompressed = g_malloc(outlen);
- ZSTD_outBuffer zout = {decompressed, outlen, 0};
-
- while (zin.pos < zin.size) {
- gsize r = ZSTD_decompressStream(zstream, &zout, &zin);
- if (ZSTD_isError(r)) {
- g_free(decompressed);
- ZSTD_freeDStream(zstream);
- rspamd_multipart_form_free(form);
- msg_err_session("result decompression error: %s",
- ZSTD_getErrorName(r));
- return FALSE;
- }
- if (zout.pos == zout.size) {
- zout.size *= 2;
- decompressed = g_realloc(zout.dst, zout.size);
- zout.dst = decompressed;
- }
+ GError *derr = NULL;
+
+ decompressed = rspamd_zstd_decompress_bounded(NULL,
+ result_data, result_len,
+ session->ctx->cfg->max_message,
+ &derr);
+
+ if (decompressed == NULL) {
+ msg_err_session("result decompression error: %s",
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
+ rspamd_multipart_form_free(form);
+ return FALSE;
}
- ZSTD_freeDStream(zstream);
- result_data = (const char *) zout.dst;
- result_len = zout.pos;
+
+ result_data = decompressed->str;
+ result_len = decompressed->len;
}
/* Parse UCL from result part */
ucl_parser_add_chunk(parser, (const unsigned char *) result_data, result_len);
}
- g_free(decompressed);
+ rspamd_fstring_free(decompressed);
if (ucl_parser_get_error(parser)) {
msg_err_session("cannot parse result UCL: %s",
rspamd_substring_search_caseless(body_part_entry->content_encoding,
body_part_entry->content_encoding_len,
"zstd", 4) != -1) {
- ZSTD_DStream *zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
- ZSTD_inBuffer zin = {bp_data, bp_len, 0};
- gsize outlen = ZSTD_getDecompressedSize(bp_data, bp_len);
- if (outlen == 0) outlen = ZSTD_DStreamOutSize();
- unsigned char *bp_decompressed = g_malloc(outlen);
- ZSTD_outBuffer zout = {bp_decompressed, outlen, 0};
- gboolean decompress_ok = TRUE;
-
- while (zin.pos < zin.size) {
- gsize r = ZSTD_decompressStream(zstream, &zout, &zin);
- if (ZSTD_isError(r)) {
- msg_warn_session("body decompression error: %s",
- ZSTD_getErrorName(r));
- decompress_ok = FALSE;
- break;
- }
- if (zout.pos == zout.size) {
- zout.size *= 2;
- bp_decompressed = g_realloc(zout.dst, zout.size);
- zout.dst = bp_decompressed;
- }
+ GError *derr = NULL;
+ rspamd_fstring_t *bp_decompressed;
+
+ bp_decompressed = rspamd_zstd_decompress_bounded(NULL,
+ bp_data, bp_len,
+ session->ctx->cfg->max_message,
+ &derr);
+
+ if (bp_decompressed == NULL) {
+ msg_warn_session("body decompression error: %s",
+ derr ? derr->message : "unknown error");
+ g_clear_error(&derr);
}
- ZSTD_freeDStream(zstream);
-
- if (decompress_ok) {
+ else {
/* Copy to pool so it persists */
- conn->body_data = rspamd_mempool_alloc(session->pool, zout.pos);
- memcpy((void *) conn->body_data, zout.dst, zout.pos);
- conn->body_len = zout.pos;
+ conn->body_data = rspamd_mempool_alloc(session->pool,
+ bp_decompressed->len);
+ memcpy((void *) conn->body_data, bp_decompressed->str,
+ bp_decompressed->len);
+ conn->body_len = bp_decompressed->len;
+ rspamd_fstring_free(bp_decompressed);
}
- g_free(bp_decompressed);
}
else {
/* Uncompressed body — copy to pool */
}
static void
-proxy_request_decompress(struct rspamd_http_message *msg)
+proxy_request_decompress(struct rspamd_http_message *msg, gsize max_size)
{
rspamd_fstring_t *body;
const char *in;
- gsize inlen, outlen, r;
- ZSTD_DStream *zstream;
- ZSTD_inBuffer zin;
- ZSTD_outBuffer zout;
+ gsize inlen;
if (rspamd_http_message_find_header(msg, COMPRESSION_HEADER)) {
+ GError *err = NULL;
+
in = rspamd_http_message_get_body(msg, &inlen);
if (in == NULL || inlen == 0) {
return;
}
- zstream = ZSTD_createDStream();
- ZSTD_initDStream(zstream);
+ body = rspamd_zstd_decompress_bounded(NULL, in, inlen, max_size, &err);
- zin.pos = 0;
- zin.src = in;
- zin.size = inlen;
+ if (body == NULL) {
+ msg_err("cannot decompress body: %s",
+ err ? err->message : "unknown error");
+ g_clear_error(&err);
- if ((outlen = ZSTD_getDecompressedSize(zin.src, zin.size)) == 0) {
- outlen = ZSTD_DStreamOutSize();
- }
-
- body = rspamd_fstring_sized_new(outlen);
- zout.dst = body->str;
- zout.pos = 0;
- zout.size = outlen;
-
- while (zin.pos < zin.size) {
- r = ZSTD_decompressStream(zstream, &zout, &zin);
-
- if (ZSTD_isError(r)) {
- msg_err("Decompression error: %s", ZSTD_getErrorName(r));
- ZSTD_freeDStream(zstream);
- rspamd_fstring_free(body);
-
- return;
- }
-
- if (zout.pos == zout.size) {
- /* We need to extend output buffer */
- zout.size = zout.size * 2 + 1;
- body = rspamd_fstring_grow(body, zout.size);
- zout.size = body->allocated;
- zout.dst = body->str;
- }
+ return;
}
- body->len = zout.pos;
- ZSTD_freeDStream(zstream);
rspamd_http_message_set_body_from_fstring_steal(msg, body);
rspamd_http_message_remove_header(msg, COMPRESSION_HEADER);
rspamd_http_message_remove_header(msg, CONTENT_ENCODING_HEADER);
session = bk_conn->s;
- proxy_request_decompress(msg);
+ proxy_request_decompress(msg, session->ctx->cfg->max_message);
orig_ct = rspamd_http_message_find_header(msg, "Content-Type");
if (!proxy_backend_parse_results(session, bk_conn, session->ctx->lua_state,
session = bk_conn->s;
rspamd_http_connection_steal_msg(session->master_conn->backend_conn);
- proxy_request_decompress(msg);
+ proxy_request_decompress(msg, session->ctx->cfg->max_message);
/*
* These are likely set by an http library, so we will double these headers
#include "rspamd_cxx_unit_symcache_timeout.hxx"
#include "rspamd_cxx_unit_text_stats.hxx"
#include "rspamd_cxx_unit_multipattern.hxx"
+#include "rspamd_cxx_unit_compression.hxx"
static gboolean verbose = false;
static const GOptionEntry entries[] =
--- /dev/null
+/*
+ * Copyright 2026 Vsevolod Stakhov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#ifndef RSPAMD_RSPAMD_CXX_UNIT_COMPRESSION_HXX
+#define RSPAMD_RSPAMD_CXX_UNIT_COMPRESSION_HXX
+
+#define DOCTEST_CONFIG_IMPLEMENTATION_IN_DLL
+#include "doctest/doctest.h"
+
+#include "libutil/compression.h"
+
+#include <string>
+#include <vector>
+
+/*
+ * Tests for the bounded zstd decompression helper shared by the HTTP,
+ * proxy and map code paths. The limit must hold both when the frame header
+ * advertises the decompressed size (single-shot compression) and when it
+ * does not (streaming compression), and the returned length must never
+ * exceed it.
+ */
+
+static std::vector<char>
+unit_zstd_compress_oneshot(const std::string &in)
+{
+ std::vector<char> out(ZSTD_compressBound(in.size()));
+ auto r = ZSTD_compress(out.data(), out.size(), in.data(), in.size(), 1);
+ REQUIRE(!ZSTD_isError(r));
+ out.resize(r);
+
+ return out;
+}
+
+/* Streaming compression produces frames without a content-size header */
+static std::vector<char>
+unit_zstd_compress_stream(const std::string &in)
+{
+ auto *cs = ZSTD_createCStream();
+ REQUIRE(cs != nullptr);
+ ZSTD_initCStream(cs, 1);
+
+ std::vector<char> out(ZSTD_compressBound(in.size()) + 128);
+ ZSTD_outBuffer zout = {out.data(), out.size(), 0};
+ ZSTD_inBuffer zin = {in.data(), in.size(), 0};
+
+ while (zin.pos < zin.size) {
+ auto r = ZSTD_compressStream(cs, &zout, &zin);
+ REQUIRE(!ZSTD_isError(r));
+ }
+
+ auto r = ZSTD_endStream(cs, &zout);
+ REQUIRE(r == 0);
+ ZSTD_freeCStream(cs);
+ out.resize(zout.pos);
+
+ return out;
+}
+
+TEST_SUITE("bounded zstd decompression")
+{
+ TEST_CASE("roundtrip without limit")
+ {
+ std::string plain;
+ for (int i = 0; i < 1000; i++) {
+ plain += "some compressible payload ";
+ }
+
+ auto compressed = unit_zstd_compress_oneshot(plain);
+ GError *err = nullptr;
+ auto *body = rspamd_zstd_decompress_bounded(nullptr, compressed.data(),
+ compressed.size(), 0, &err);
+ REQUIRE(body != nullptr);
+ REQUIRE(err == nullptr);
+ CHECK(std::string{body->str, body->len} == plain);
+ rspamd_fstring_free(body);
+ }
+
+ TEST_CASE("limit exactly equal to decompressed size is allowed")
+ {
+ std::string plain(65536, 'x');
+ auto compressed = unit_zstd_compress_oneshot(plain);
+
+ GError *err = nullptr;
+ auto *body = rspamd_zstd_decompress_bounded(nullptr, compressed.data(),
+ compressed.size(),
+ plain.size(), &err);
+ REQUIRE(body != nullptr);
+ REQUIRE(err == nullptr);
+ CHECK(body->len == plain.size());
+ rspamd_fstring_free(body);
+ }
+
+ TEST_CASE("advertised size over the limit is rejected upfront")
+ {
+ std::string plain(4 * 1024 * 1024, '\0');
+ auto compressed = unit_zstd_compress_oneshot(plain);
+
+ GError *err = nullptr;
+ auto *body = rspamd_zstd_decompress_bounded(nullptr, compressed.data(),
+ compressed.size(),
+ 1024 * 1024, &err);
+ CHECK(body == nullptr);
+ REQUIRE(err != nullptr);
+ CHECK(err->code == RSPAMD_DECOMPRESS_ERROR_TOO_LARGE);
+ g_error_free(err);
+ }
+
+ TEST_CASE("unknown-size frame over the limit is rejected while streaming")
+ {
+ std::string plain(4 * 1024 * 1024, '\0');
+ auto compressed = unit_zstd_compress_stream(plain);
+ /* The frame must not advertise its decompressed size */
+ REQUIRE(ZSTD_getDecompressedSize(compressed.data(), compressed.size()) == 0);
+
+ GError *err = nullptr;
+ auto *body = rspamd_zstd_decompress_bounded(nullptr, compressed.data(),
+ compressed.size(),
+ 64 * 1024, &err);
+ CHECK(body == nullptr);
+ REQUIRE(err != nullptr);
+ CHECK(err->code == RSPAMD_DECOMPRESS_ERROR_TOO_LARGE);
+ g_error_free(err);
+ }
+
+ TEST_CASE("unknown-size frame within the limit roundtrips via buffer growth")
+ {
+ std::string plain;
+ for (int i = 0; i < 100000; i++) {
+ plain += "0123456789";
+ }
+ auto compressed = unit_zstd_compress_stream(plain);
+ REQUIRE(ZSTD_getDecompressedSize(compressed.data(), compressed.size()) == 0);
+
+ GError *err = nullptr;
+ auto *body = rspamd_zstd_decompress_bounded(nullptr, compressed.data(),
+ compressed.size(),
+ 2 * plain.size(), &err);
+ REQUIRE(body != nullptr);
+ REQUIRE(err == nullptr);
+ CHECK(std::string{body->str, body->len} == plain);
+ rspamd_fstring_free(body);
+ }
+
+ TEST_CASE("corrupt input is an error")
+ {
+ std::string garbage = "definitely not a zstd frame at all";
+
+ GError *err = nullptr;
+ auto *body = rspamd_zstd_decompress_bounded(nullptr, garbage.data(),
+ garbage.size(), 0, &err);
+ CHECK(body == nullptr);
+ REQUIRE(err != nullptr);
+ CHECK(err->code == RSPAMD_DECOMPRESS_ERROR_DATA);
+ g_error_free(err);
+ }
+
+ TEST_CASE("caller-provided stream is used and not freed")
+ {
+ std::string plain(1024, 'y');
+ auto compressed = unit_zstd_compress_oneshot(plain);
+
+ auto *ds = ZSTD_createDStream();
+ REQUIRE(ds != nullptr);
+ ZSTD_initDStream(ds);
+
+ GError *err = nullptr;
+ auto *body = rspamd_zstd_decompress_bounded(ds, compressed.data(),
+ compressed.size(), 0, &err);
+ REQUIRE(body != nullptr);
+ CHECK(std::string{body->str, body->len} == plain);
+ rspamd_fstring_free(body);
+
+ /* Stream must still be usable for another frame */
+ ZSTD_DCtx_reset(ds, ZSTD_reset_session_only);
+ body = rspamd_zstd_decompress_bounded(ds, compressed.data(),
+ compressed.size(), 0, &err);
+ REQUIRE(body != nullptr);
+ CHECK(body->len == plain.size());
+ rspamd_fstring_free(body);
+ ZSTD_freeDStream(ds);
+ }
+}
+
+#endif
projects / thirdparty / rspamd.git / commitdiff
