/* Bounds the memory used by the throttling state */
#define AUTH_FAILURES_CACHE_SIZE 1024
+/*
+ * Admission control limits, both disabled by default so that this release
+ * behaves exactly as the previous ones unless an operator opts in.
+ */
+#define DEFAULT_MAX_CONNECTIONS 0
+#define DEFAULT_MAX_CONNECTIONS_PER_SOURCE 0
+
/* HTTP paths */
#define PATH_AUTH "/auth"
#define PATH_SYMBOLS "/symbols"
rspamd_lru_hash_t *auth_failures;
unsigned int max_auth_failures;
double auth_failure_window;
+ /*
+ * Admission control. Both limits are disabled (0) by default to preserve
+ * the historical behaviour; `conns_per_source` counts the connections that
+ * are currently being served, keyed by peer address.
+ */
+ unsigned int max_connections;
+ unsigned int max_connections_per_source;
+ GHashTable *conns_per_source;
+ /*
+ * Whether TCP clients of this worker may use the privileged File/Path/Shm
+ * message source inputs. Unix socket peers always may.
+ */
+ gboolean allow_file_and_shm_inputs;
/* HTTP server */
struct rspamd_http_context *http_ctx;
struct rspamd_http_connection_router *http;
return ret;
}
+/*
+ * Whether this connection may use the privileged File/Path/Shm message source
+ * inputs, which make rspamd open an object named by the client in the server's
+ * own filesystem or shared memory.
+ *
+ * This is a property of the transport the connection was accepted on and of
+ * nothing else. Authentication and this capability are deliberately separate:
+ * a correct `enable_password` proves who the client is, it does not make the
+ * client local, so it must not unlock filesystem access. The password-less
+ * shortcuts of `rspamd_controller_check_password` must not leak in here either
+ * — the AF_UNIX one happens to coincide with the rule below, but a `secure_ip`
+ * peer is still an ordinary TCP client and stays gated.
+ *
+ * Nothing that the client controls (headers, `Forwarded`, `User-Agent`, query
+ * arguments, request metadata) may be consulted here.
+ */
+static gboolean
+rspamd_controller_allow_file_shm(struct rspamd_controller_session *session)
+{
+ if (session->from_addr != NULL &&
+ rspamd_inet_address_get_af(session->from_addr) == AF_UNIX) {
+ /*
+ * A unix socket peer already needs filesystem access to reach us, so
+ * naming a file adds no privileges it does not have.
+ */
+ return TRUE;
+ }
+
+ return session->ctx->allow_file_and_shm_inputs;
+}
+
+/*
+ * Propagates the properties of the accepted transport onto a task created for
+ * this session. Must be called before anything can reach
+ * `rspamd_task_load_message` (or the v3 request parser), as those are the
+ * points where a privileged input would otherwise be opened.
+ */
+static void
+rspamd_controller_task_set_transport(struct rspamd_controller_session *session,
+ struct rspamd_task *task)
+{
+ if (!rspamd_controller_allow_file_shm(session)) {
+ task->protocol_flags &= ~RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT;
+ }
+
+ /* Locality is derived from the peer address only, never from a header */
+ if (session->from_addr != NULL &&
+ rspamd_worker_addr_is_loopback(session->from_addr)) {
+ task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT;
+ }
+ else {
+ task->protocol_flags &= ~RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT;
+ }
+
+ /*
+ * The session owns its address and outlives the task only in the sense that
+ * it frees it in the finish handler, so the task gets its own copy.
+ */
+ if (task->client_addr == NULL && session->from_addr != NULL) {
+ task->client_addr = rspamd_inet_address_copy(session->from_addr, NULL);
+ }
+}
+
/*
* A leaky bucket of failed authentication attempts for a single source address.
* `penalty` is decayed lazily on access, so no timer is needed to drain it.
NULL,
(event_finalizer_t) rspamd_task_free);
task->fin_arg = conn_ent;
+ rspamd_controller_task_set_transport(session, task);
ptask = lua_newuserdata(L, sizeof(*ptask));
*ptask = task;
task->http_conn = rspamd_http_connection_ref(conn_ent->conn);
task->sock = -1;
session->task = task;
+ rspamd_controller_task_set_transport(session, task);
if (msg->body_buf.len > 0) {
if (!rspamd_task_load_message(task, msg, msg->body_buf.begin, msg->body_buf.len)) {
task->http_conn = rspamd_http_connection_ref(conn_ent->conn);
task->sock = -1;
session->task = task;
+ rspamd_controller_task_set_transport(session, task);
cl_header = rspamd_http_message_find_header(msg, "classifier");
if (cl_header) {
task->http_conn = rspamd_http_connection_ref(conn_ent->conn);
task->sock = -1;
session->task = task;
+ rspamd_controller_task_set_transport(session, task);
cl_header = rspamd_http_message_find_header(msg, "classifier");
if (cl_header) {
task->sock = conn_ent->conn->fd;
task->flags |= RSPAMD_TASK_FLAG_MIME;
task->resolver = ctx->resolver;
+ /*
+ * Before the request is parsed: the v3 handler below can map a shm segment
+ * on its own, without going through rspamd_task_load_message
+ */
+ rspamd_controller_task_set_transport(session, task);
if (!rspamd_protocol_handle_request(task, msg)) {
task->flags |= RSPAMD_TASK_FLAG_SKIP;
cbdata);
task->fin_arg = cbdata;
task->http_conn = rspamd_http_connection_ref(conn_ent->conn);
- ;
task->sock = conn_ent->conn->fd;
+ rspamd_controller_task_set_transport(session, task);
ucl_object_insert_key(top, ucl_object_fromstring(RVERSION), "version", 0, false);
ucl_object_insert_key(top, ucl_object_fromstring(session->ctx->cfg->checksum), "config_id", 0, false);
task->fin_arg = cbdata;
task->http_conn = rspamd_http_connection_ref(conn_ent->conn);
task->sock = conn_ent->conn->fd;
+ rspamd_controller_task_set_transport(session, task);
if (stat_copy.messages_scanned > 0 && do_reset) {
for (int i = METRIC_ACTION_REJECT; i <= METRIC_ACTION_NOACTION; i++) {
(event_finalizer_t) rspamd_task_free);
task->fin_arg = conn_ent;
task->http_conn = rspamd_http_connection_ref(conn_ent->conn);
- ;
task->sock = -1;
session->task = task;
+ rspamd_controller_task_set_transport(session, task);
if (msg->body_buf.len > 0) {
if (!rspamd_task_load_message(task, msg, msg->body_buf.begin, msg->body_buf.len)) {
classifier_obj = ucl_object_typed_new(UCL_OBJECT);
ucl_object_insert_key(classifier_obj,
- ucl_object_fromstring(clc->name),
- "name", 0, false);
+ ucl_object_fromstring(clc->name),
+ "name", 0, false);
ucl_object_insert_key(classifier_obj,
- ucl_object_fromstring(rspamd_classifier_type(clc)),
- "type", 0, false);
+ ucl_object_fromstring(rspamd_classifier_type(clc)),
+ "type", 0, false);
ucl_object_insert_key(classifier_obj,
- ucl_object_frombool(rspamd_classifier_is_per_user(clc)),
- "per_user", 0, false);
+ ucl_object_frombool(rspamd_classifier_is_per_user(clc)),
+ "per_user", 0, false);
/* Collect unique class names from statfiles.
* Linear search is used since N < 10 in practice, avoiding hash table overhead. */
msg_err_session("http error occurred: %s", err->message);
}
+/*
+ * Pending connections accounting per source address. The global limit alone
+ * would let a single peer occupy the whole budget of the worker, so in-flight
+ * connections are also counted per source. Entries are dropped as soon as a
+ * source has no connection left, which bounds the table by the number of peers
+ * that are actually connected right now.
+ *
+ * Returns FALSE when the source is already at its limit; the caller must then
+ * not create a session (nothing has been accounted in that case).
+ */
+static gboolean
+rspamd_controller_source_conn_add(struct rspamd_controller_worker_ctx *ctx,
+ const rspamd_inet_addr_t *addr)
+{
+ unsigned int *nconns;
+
+ if (ctx->conns_per_source == NULL || ctx->max_connections_per_source == 0 ||
+ addr == NULL || rspamd_inet_address_get_af(addr) == AF_UNIX) {
+ /* Disabled, or a local peer that is not subject to the limit */
+ return TRUE;
+ }
+
+ nconns = g_hash_table_lookup(ctx->conns_per_source, addr);
+
+ if (nconns != NULL) {
+ if (*nconns >= ctx->max_connections_per_source) {
+ return FALSE;
+ }
+
+ (*nconns)++;
+ }
+ else {
+ nconns = g_malloc(sizeof(*nconns));
+ *nconns = 1;
+ g_hash_table_insert(ctx->conns_per_source,
+ rspamd_inet_address_copy(addr, NULL),
+ nconns);
+ }
+
+ return TRUE;
+}
+
+/*
+ * Releases one pending connection of a source. Mirrors
+ * `rspamd_controller_source_conn_add` and must be called exactly once for every
+ * successful call of it.
+ */
+static void
+rspamd_controller_source_conn_remove(struct rspamd_controller_worker_ctx *ctx,
+ const rspamd_inet_addr_t *addr)
+{
+ unsigned int *nconns;
+
+ if (ctx->conns_per_source == NULL || ctx->max_connections_per_source == 0 ||
+ addr == NULL || rspamd_inet_address_get_af(addr) == AF_UNIX) {
+ return;
+ }
+
+ nconns = g_hash_table_lookup(ctx->conns_per_source, addr);
+
+ if (nconns == NULL) {
+ return;
+ }
+
+ if (*nconns > 1) {
+ (*nconns)--;
+ }
+ else {
+ g_hash_table_remove(ctx->conns_per_source, addr);
+ }
+}
+
static void
rspamd_controller_finish_handler(struct rspamd_http_connection_entry *conn_ent)
{
}
session->wrk->nconns--;
+ rspamd_controller_source_conn_remove(session->ctx, session->from_addr);
rspamd_inet_address_free(session->from_addr);
CFG_REF_RELEASE(session->cfg);
return;
}
+ /*
+ * Admission control runs before anything is allocated for this connection:
+ * a session pool, a TLS handshake or a password KDF are all far more
+ * expensive than the accept itself, so the limits must be applied while the
+ * connection is still just a file descriptor. The connection is accepted
+ * and closed at once instead of being left in the listen queue, otherwise
+ * the level triggered accept watcher would spin on a readable listener.
+ */
+ if (ctx->max_connections > 0 && worker->nconns >= ctx->max_connections) {
+ msg_info_ctx("deny connection from %s: the worker already serves "
+ "%ud connections, the limit is %ud",
+ rspamd_inet_address_to_string_pretty(addr),
+ worker->nconns, ctx->max_connections);
+ rspamd_inet_address_free(addr);
+ close(nfd);
+
+ return;
+ }
+
+ if (!rspamd_controller_source_conn_add(ctx, addr)) {
+ msg_info_ctx("deny connection from %s: this source already has %ud "
+ "pending connections",
+ rspamd_inet_address_to_string_pretty(addr),
+ ctx->max_connections_per_source);
+ rspamd_inet_address_free(addr);
+ close(nfd);
+
+ return;
+ }
+
session = g_malloc0(sizeof(struct rspamd_controller_session));
session->pool = rspamd_mempool_new_short_lived("csession");
session->ctx = ctx;
session->from_addr = addr;
session->wrk = worker;
+ /* Decided once, from the transport, so that handlers in other modules
+ * (e.g. fuzzy_check) can honour it without seeing our private context */
+ session->allow_file_shm_input = rspamd_controller_allow_file_shm(session);
worker->nconns++;
rspamd_http_router_handle_socket_ssl(ctx->http, nfd, session,
ctx->task_timeout = NAN;
ctx->max_auth_failures = DEFAULT_MAX_AUTH_FAILURES;
ctx->auth_failure_window = DEFAULT_AUTH_FAILURE_WINDOW;
+ /* Permissive for this release, see the option description below */
+ ctx->allow_file_and_shm_inputs = TRUE;
+ ctx->max_connections = DEFAULT_MAX_CONNECTIONS;
+ ctx->max_connections_per_source = DEFAULT_MAX_CONNECTIONS_PER_SOURCE;
rspamd_rcl_register_worker_option(cfg,
type,
"Time for a source to regain all of its authentication "
"attempts, default: 60 seconds");
+ rspamd_rcl_register_worker_option(cfg,
+ type,
+ "allow_file_and_shm_inputs",
+ rspamd_rcl_parse_struct_boolean,
+ ctx,
+ G_STRUCT_OFFSET(struct rspamd_controller_worker_ctx,
+ allow_file_and_shm_inputs),
+ 0,
+ "Allow privileged File/Path/Shm message source inputs, which "
+ "make rspamd read a local file or shared memory segment named "
+ "by the client, for connections accepted over TCP; unix socket "
+ "clients are always allowed to use them. Default: true, this "
+ "will become false in the next major release");
+
+ rspamd_rcl_register_worker_option(cfg,
+ type,
+ "max_connections",
+ rspamd_rcl_parse_struct_integer,
+ ctx,
+ G_STRUCT_OFFSET(struct rspamd_controller_worker_ctx,
+ max_connections),
+ RSPAMD_CL_FLAG_UINT,
+ "Maximum number of connections served simultaneously by one "
+ "controller worker, further connections are closed right after "
+ "accept, 0 means unlimited, default: 0");
+
+ rspamd_rcl_register_worker_option(cfg,
+ type,
+ "max_connections_per_source",
+ rspamd_rcl_parse_struct_integer,
+ ctx,
+ G_STRUCT_OFFSET(struct rspamd_controller_worker_ctx,
+ max_connections_per_source),
+ RSPAMD_CL_FLAG_UINT,
+ "Maximum number of simultaneous connections from a single "
+ "source address, unix socket clients are not affected, "
+ "0 means unlimited, default: 0");
+
return ctx;
}
"controller",
rspamd_controller_accept_socket);
+ rspamd_worker_warn_file_shm_inputs(worker, "controller",
+ ctx->allow_file_and_shm_inputs);
+
ctx->worker = worker;
ctx->cfg = worker->srv->cfg;
CFG_REF_RETAIN(ctx->cfg);
msg_info_ctx("authentication failure throttling is disabled");
}
+ if (ctx->max_connections_per_source > 0) {
+ ctx->conns_per_source = g_hash_table_new_full(rspamd_inet_address_hash,
+ rspamd_inet_address_equal,
+ (GDestroyNotify) rspamd_inet_address_free,
+ g_free);
+ }
+
if (ctx->secure_ip != NULL) {
rspamd_config_radix_from_ucl(ctx->cfg, ctx->secure_ip,
"Allow unauthenticated requests from these addresses",
rspamd_http_switch_zc(struct _rspamd_http_privbuf *pbuf,
struct rspamd_http_message *msg)
{
+ if (msg->body_buf.begin == NULL ||
+ msg->body_buf.allocated_len <= msg->body_buf.len) {
+ /*
+ * There is no mapping to write into (e.g. growing it has just failed),
+ * so stay with the private buffer instead of pointing at nothing
+ */
+ pbuf->zc_buf = NULL;
+ pbuf->zc_remain = 0;
+
+ return;
+ }
+
pbuf->zc_buf = msg->body_buf.begin + msg->body_buf.len;
pbuf->zc_remain = msg->body_buf.allocated_len - msg->body_buf.len;
}
}
}
else {
+ if (msg->body_buf.begin == NULL ||
+ msg->body_buf.allocated_len < msg->body_buf.len ||
+ msg->body_buf.allocated_len - msg->body_buf.len < length) {
+ /*
+ * The parser cannot have read more than the mapping we handed to
+ * it, so this is not expected to happen; bail out rather than write
+ * past the end of the body storage
+ */
+ return -1;
+ }
+
if (msg->body_buf.begin + msg->body_buf.len != at) {
/* Likely chunked encoding */
memmove((char *) msg->body_buf.begin + msg->body_buf.len, at, length);
len = pbuf->zc_remain;
if (len == 0) {
- rspamd_http_message_grow_body(priv->msg, priv->buf->data->allocated);
- rspamd_http_switch_zc(pbuf, msg);
- data = (char *) pbuf->zc_buf;
- len = pbuf->zc_remain;
+ if (rspamd_http_message_grow_body(priv->msg,
+ priv->buf->data->allocated)) {
+ rspamd_http_switch_zc(pbuf, msg);
+ data = (char *) pbuf->zc_buf;
+ len = pbuf->zc_remain;
+ }
+
+ if (pbuf->zc_buf == NULL || len == 0) {
+ /*
+ * There is no room left in the body storage and no way to get
+ * any (growing it has just failed). Read into the private
+ * buffer instead: appending it to the body then fails in the
+ * parser callback and the error is reported through the usual
+ * path, whereas here we would have written past the mapping.
+ */
+ pbuf->zc_buf = NULL;
+ pbuf->zc_remain = 0;
+ data = priv->buf->data->str;
+ len = priv->buf->data->allocated;
+ }
}
}
REF_RETAIN(storage->shared.name);
}
+ /*
+ * The segment belongs to the original message and stays mutable:
+ * it can be both resized and rewritten whilst this copy keeps it
+ * mapped. `allocated_len` therefore records the length passed to
+ * mmap below (the only length that may ever be unmapped) and
+ * consumers of this body must snapshot the bytes they need rather
+ * than parsing them in place.
+ */
new_msg->body_buf.str = mmap(NULL, st.st_size,
PROT_READ, MAP_SHARED,
storage->shared.shm_fd, 0);
rspamd_http_detach_shared(msg);
}
+ /*
+ * `Shm`, `Shm-Offset` and `Shm-Length` describe our own segment and are
+ * generated internally, so they are a reserved triplet: any instance that
+ * came from a peer is dropped here. Otherwise the internally generated
+ * values would be appended behind the peer supplied ones and a receiver
+ * reading the first instance of each header would open a segment and an
+ * offset of somebody else's choice.
+ */
+ rspamd_http_message_remove_header(msg, "Shm");
+ rspamd_http_message_remove_header(msg, "Shm-Offset");
+ rspamd_http_message_remove_header(msg, "Shm-Length");
+
if (allow_shared) {
char tmpbuf[64];
shm_offset = msg->body_buf.begin - msg->body_buf.str;
}
- /* Insert new headers */
+ /* Insert exactly one internally generated triplet */
rspamd_http_message_add_header(msg, "Shm",
msg->body_buf.c.shared.name->shm_name);
rspamd_snprintf(tmpbuf, sizeof(tmpbuf), "%uz", shm_offset);
len = 0;
}
+ /*
+ * The segment created below is ours and is mapped writable, so the
+ * message is no longer immutable even if the previous body was
+ */
+ msg->flags &= ~RSPAMD_HTTP_FLAG_SHMEM_IMMUTABLE;
+
storage->shared.name = g_malloc(sizeof(*storage->shared.name));
REF_INIT_RETAIN(storage->shared.name, rspamd_http_shname_dtor);
#ifdef HAVE_SANE_SHMEM
#endif
if (storage->shared.shm_fd == -1) {
- return FALSE;
+ goto shm_fail;
}
if (len != 0) {
if (ftruncate(storage->shared.shm_fd, len) == -1) {
- return FALSE;
+ goto shm_fail;
}
msg->body_buf.str = mmap(NULL, len,
storage->shared.shm_fd, 0);
if (msg->body_buf.str == MAP_FAILED) {
- return FALSE;
+ goto shm_fail;
}
msg->body_buf.begin = msg->body_buf.str;
msg->flags |= RSPAMD_HTTP_FLAG_HAS_BODY;
return TRUE;
+
+shm_fail:
+ /*
+ * Release the descriptor and the segment we have just created and reset the
+ * body description: `begin` and `allocated_len` must never keep describing
+ * a mapping that we have failed to establish.
+ */
+ rspamd_http_message_storage_cleanup(msg);
+ msg->flags &= ~RSPAMD_HTTP_FLAG_HAS_BODY;
+
+ return FALSE;
}
void rspamd_http_message_set_method(struct rspamd_http_message *msg,
msg->body_buf.str = MAP_FAILED;
if (storage->shared.shm_fd == -1) {
- return FALSE;
+ goto fd_fail;
}
if (fstat(storage->shared.shm_fd, &st) == -1) {
- return FALSE;
+ goto fd_fail;
}
if (!S_ISREG(st.st_mode) || st.st_size <= 0) {
/* Nothing that can be mapped: not a regular file or an empty one */
- return FALSE;
+ goto fd_fail;
}
+ /*
+ * The object behind `fd` belongs to the caller and stays mutable: it can be
+ * resized and rewritten while we keep it mapped. Hence `allocated_len`
+ * records the length passed to mmap here (the only length that may ever be
+ * unmapped) and consumers of such a body must snapshot the bytes they need
+ * instead of parsing them in place.
+ */
msg->body_buf.str = mmap(NULL, st.st_size,
PROT_READ, MAP_SHARED,
storage->shared.shm_fd, 0);
if (msg->body_buf.str == MAP_FAILED) {
- return FALSE;
+ goto fd_fail;
}
msg->body_buf.begin = msg->body_buf.str;
msg->body_buf.allocated_len = st.st_size;
return TRUE;
+
+fd_fail:
+ /*
+ * Close the descriptor we have dup'ed and switch the message back to the
+ * ordinary heap storage, so that neither the flags nor the body pointers
+ * describe a shared segment that we do not have
+ */
+ rspamd_http_message_drop_shared_body(msg);
+ msg->flags &= ~RSPAMD_HTTP_FLAG_HAS_BODY;
+
+ return FALSE;
}
gboolean
gboolean
rspamd_http_message_grow_body(struct rspamd_http_message *msg, gsize len)
{
- struct stat st;
union _rspamd_storage_u *storage;
gsize newlen;
storage = &msg->body_buf.c;
if (msg->flags & RSPAMD_HTTP_FLAG_SHMEM) {
- if (storage->shared.shm_fd == -1) {
+ if (msg->flags & RSPAMD_HTTP_FLAG_SHMEM_IMMUTABLE) {
+ /*
+ * The segment is not ours: it is mapped read only and resizing it
+ * would affect every other holder of the very same object
+ */
return FALSE;
}
- if (len > G_MAXSIZE - msg->body_buf.len) {
- /* Integer overflow in the requested size */
+ if (storage->shared.shm_fd == -1) {
return FALSE;
}
- if (fstat(storage->shared.shm_fd, &st) == -1) {
+ if (len > G_MAXSIZE - msg->body_buf.len) {
+ /* Integer overflow in the requested size */
return FALSE;
}
- /* Check if we need to grow */
- if ((gsize) st.st_size < msg->body_buf.len + len) {
+ /*
+ * `allocated_len` is the length that was passed to the mmap which
+ * produced the current mapping. The underlying object can be resized by
+ * any other holder of it, so its current size is unrelated to the size
+ * of our mapping and must never be used to reason about it.
+ */
+ if (msg->body_buf.allocated_len < msg->body_buf.len + len) {
/* Need to grow */
- newlen = rspamd_fstring_suggest_size(msg->body_buf.len, st.st_size,
+ newlen = rspamd_fstring_suggest_size(msg->body_buf.len,
+ msg->body_buf.allocated_len,
len);
/* Unmap as we need another size of segment */
if (RSPAMD_HTTP_BODY_IS_MAPPED(msg)) {
- munmap(msg->body_buf.str, st.st_size);
- msg->body_buf.str = MAP_FAILED;
+ munmap(msg->body_buf.str, msg->body_buf.allocated_len);
}
+ /*
+ * From this point on there is no mapping: nothing may look at
+ * `begin` or `allocated_len` until a new one is established, hence
+ * they are reset before anything that can fail
+ */
+ msg->body_buf.str = MAP_FAILED;
+ msg->body_buf.begin = NULL;
+ msg->body_buf.allocated_len = 0;
+
if (ftruncate(storage->shared.shm_fd, newlen) == -1) {
+ msg->body_buf.len = 0;
+
return FALSE;
}
PROT_WRITE | PROT_READ, MAP_SHARED,
storage->shared.shm_fd, 0);
if (msg->body_buf.str == MAP_FAILED) {
+ msg->body_buf.len = 0;
+
return FALSE;
}
storage = &msg->body_buf.c;
if (msg->flags & RSPAMD_HTTP_FLAG_SHMEM) {
+ if (msg->flags & RSPAMD_HTTP_FLAG_SHMEM_IMMUTABLE) {
+ /* Read only mapping of a segment that we do not own */
+ return FALSE;
+ }
+
if (!rspamd_http_message_grow_body(msg, len)) {
return FALSE;
}
+ if (!RSPAMD_HTTP_BODY_IS_MAPPED(msg) ||
+ msg->body_buf.allocated_len < msg->body_buf.len ||
+ msg->body_buf.allocated_len - msg->body_buf.len < len) {
+ /* Should not happen after a successful grow, but never write out of the mapping */
+ return FALSE;
+ }
+
memcpy(msg->body_buf.str + msg->body_buf.len, data, len);
msg->body_buf.len += len;
}
void rspamd_http_message_storage_cleanup(struct rspamd_http_message *msg)
{
union _rspamd_storage_u *storage;
- struct stat st;
/* Free piecewise body iov if present */
if (msg->body_iov) {
if (msg->flags & RSPAMD_HTTP_FLAG_SHMEM) {
storage = &msg->body_buf.c;
- if (storage->shared.shm_fd >= 0) {
- if (RSPAMD_HTTP_BODY_IS_MAPPED(msg)) {
- /*
- * We map the whole segment, hence its current size is the
- * mapping length. If fstat fails somehow, we have no reliable
- * length to unmap, so we have to leak the mapping instead of
- * unmapping a wrong range.
- */
- if (fstat(storage->shared.shm_fd, &st) != -1) {
- munmap(msg->body_buf.str, st.st_size);
- }
- else {
- msg_err("cannot fstat shmem fd %d: %s; mapping is leaked",
- storage->shared.shm_fd, strerror(errno));
- }
- }
+ /*
+ * Unmap exactly what has been mapped: `allocated_len` is the length
+ * that was passed to mmap. The underlying object may have been resized
+ * by any other holder of it since then, so its current size is not the
+ * length of our mapping. The mapping is also independent of the
+ * descriptor, hence it is released even if the fd is gone already.
+ */
+ if (RSPAMD_HTTP_BODY_IS_MAPPED(msg)) {
+ munmap(msg->body_buf.str, msg->body_buf.allocated_len);
+ }
+ if (storage->shared.shm_fd >= 0) {
close(storage->shared.shm_fd);
}
}
msg->body_buf.c.normal = NULL;
+ msg->body_buf.str = NULL;
}
+ /*
+ * There is no storage anymore, so nothing may describe one: `begin` would
+ * otherwise dangle into an unmapped or freed range whilst `allocated_len`
+ * would still claim a size for it
+ */
+ msg->body_buf.begin = NULL;
msg->body_buf.len = 0;
+ msg->body_buf.allocated_len = 0;
}
void rspamd_http_message_drop_shared_body(struct rspamd_http_message *msg)
g_hash_table_replace(rcpt_args, key_tok, val_tok);
}
+/*
+ * TRUE if `hn_tok` names one of the privileged message source controls
+ * (File/Path/Shm/Shm-Offset/Shm-Length), which make rspamd read the message
+ * from a local object named by the client.
+ *
+ * `Filename` is intentionally not matched here: it is descriptive metadata and
+ * it never selects a message source.
+ */
+static gboolean
+rspamd_protocol_is_privileged_header(const rspamd_ftok_t *hn_tok)
+{
+ static const rspamd_ftok_t privileged_headers[] = {
+ {sizeof(FILE_HEADER) - 1, FILE_HEADER},
+ {sizeof(PATH_HEADER) - 1, PATH_HEADER},
+ {sizeof(SHM_HEADER) - 1, SHM_HEADER},
+ {sizeof(SHM_OFFSET_HEADER) - 1, SHM_OFFSET_HEADER},
+ {sizeof(SHM_LENGTH_HEADER) - 1, SHM_LENGTH_HEADER},
+ };
+ unsigned int i;
+
+ for (i = 0; i < G_N_ELEMENTS(privileged_headers); i++) {
+ if (rspamd_ftok_casecmp(hn_tok, &privileged_headers[i]) == 0) {
+ return TRUE;
+ }
+ }
+
+ return FALSE;
+}
+
+/*
+ * Vet a request header before it is recorded in task->request_headers.
+ *
+ * Returns TRUE when the header may be stored. Returns FALSE, with task->err set
+ * to a 400 protocol error, when it is a privileged message source control and
+ * this connection is not allowed to use one: such a header is never stored, so
+ * nothing downstream can act on it.
+ */
+static gboolean
+rspamd_protocol_check_privileged_header(struct rspamd_task *task,
+ const rspamd_ftok_t *hn_tok)
+{
+ if (!rspamd_protocol_is_privileged_header(hn_tok) ||
+ rspamd_task_allow_file_shm_input(task)) {
+ return TRUE;
+ }
+
+ msg_info_protocol("deny privileged message source header %T from %s: this "
+ "connection is not permitted to use file/shm inputs; set "
+ "`allow_file_and_shm_inputs = true` for this worker if all "
+ "of its clients are trusted",
+ hn_tok,
+ rspamd_inet_address_to_string_pretty(task->client_addr));
+ g_set_error(&task->err, rspamd_protocol_quark(), 400,
+ "file and shm message sources are not allowed on this connection");
+ /* This is a client error, so report it as a genuine 400 */
+ task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_VERBATIM_ERR_CODE;
+
+ return FALSE;
+}
+
#define IF_HEADER(name) \
srch.begin = (name); \
srch.len = sizeof(name) - 1; \
{
msg_debug_protocol("read user-agent header, value: %T", hv_tok);
- if (hv_tok->len == 6 &&
- rspamd_lc_cmp(hv_tok->begin, "rspamc", 6) == 0) {
- task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT;
- }
+ /*
+ * The value is recorded as a plain request header (below) and
+ * nothing else is derived from it on purpose.
+ *
+ * In particular RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT is
+ * transport derived: it is set by the accepting worker from the
+ * actual socket and peer identity and it MUST NEVER be inferred
+ * from `User-Agent` or from any other client controlled value,
+ * otherwise any remote client can claim to be a local rspamc.
+ */
}
break;
case 'l':
break;
}
+ /*
+ * Defence in depth: File/Path/Shm/Shm-Offset/Shm-Length have no
+ * case arm of their own, so this unconditional store is the only
+ * way they reach rspamd_task_load_message
+ */
+ if (!rspamd_protocol_check_privileged_header (task, hn_tok)) {
+ return FALSE;
+ }
+
rspamd_task_add_request_header (task, hn_tok, hv_tok);
}
}); /* End of kh_foreach_value */
struct rspamd_shmem_segment *seg;
GError *shm_err = NULL;
+ /*
+ * The named shared memory object is chosen by the client, so this
+ * transfer mode is only available on transports that the accepting
+ * worker has marked as privileged. Checked before the name is even
+ * looked at, so no shm_open/fstat/mmap happens otherwise.
+ */
+ if (!rspamd_task_allow_file_shm_input(task)) {
+ msg_info_protocol("deny v3 shm request body from %s: this "
+ "connection is not permitted to use file/shm "
+ "inputs; set `allow_file_and_shm_inputs = true` "
+ "for this worker if all of its clients are trusted",
+ rspamd_inet_address_to_string_pretty(task->client_addr));
+ g_set_error(&task->err, rspamd_protocol_quark(), 400,
+ "file and shm message sources are not allowed "
+ "on this connection");
+ /* This is a client error, so report it as a genuine 400 */
+ task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_VERBATIM_ERR_CODE;
+
+ return FALSE;
+ }
+
off_tok = rspamd_http_message_find_header(msg, "Shm-Offset");
len_tok = rspamd_http_message_find_header(msg, "Shm-Length");
const ucl_object_t *file_elt = ucl_object_lookup(metadata_obj, "file");
const ucl_object_t *shm_elt = ucl_object_lookup(metadata_obj, "shm");
+ /*
+ * These are the metadata equivalents of the File/Shm request headers: they
+ * make rspamd read the message from a local object named by the client, so
+ * they are only honoured on privileged transports. rspamd_task_load_message
+ * checks this again, but rejecting here guarantees that nothing is even
+ * inspected (no fpath is recorded, no header is synthesized) and gives the
+ * client a message that points at the actual metadata keys.
+ */
+ if (!rspamd_task_allow_file_shm_input(task) &&
+ (file_elt != NULL || shm_elt != NULL ||
+ ucl_object_lookup(metadata_obj, "shm_offset") != NULL ||
+ ucl_object_lookup(metadata_obj, "shm_length") != NULL)) {
+ msg_info_protocol("deny v3 metadata file/shm message source from %s: this "
+ "connection is not permitted to use file/shm inputs; set "
+ "`allow_file_and_shm_inputs = true` for this worker if all "
+ "of its clients are trusted",
+ rspamd_inet_address_to_string_pretty(task->client_addr));
+ g_set_error(&task->err, rspamd_protocol_quark(), 400,
+ "'file' and 'shm' metadata message sources are not allowed "
+ "on this connection");
+ /* This is a client error, so report it as a genuine 400 */
+ task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_VERBATIM_ERR_CODE;
+
+ return FALSE;
+ }
+
if (file_elt && ucl_object_type(file_elt) == UCL_STRING) {
/* Set file path and let rspamd_task_load_message handle it via task header */
gsize fplen;
ucl_object_t *top = NULL;
top = ucl_object_typed_new(UCL_OBJECT);
- msg->code = 500 + task->err->code % 100;
+
+ /*
+ * Historically every protocol error was folded into the 5xx range, and
+ * clients (and tests) rely on that for the pre-existing error paths.
+ * Sites that deliberately want to report a genuine client error set
+ * RSPAMD_TASK_PROTOCOL_FLAG_VERBATIM_ERR_CODE and their code is used
+ * as is; everything else keeps the legacy mapping.
+ */
+ if ((task->protocol_flags & RSPAMD_TASK_PROTOCOL_FLAG_VERBATIM_ERR_CODE) &&
+ task->err->code >= 100 && task->err->code < 600) {
+ msg->code = task->err->code;
+ }
+ else {
+ msg->code = 500 + task->err->code % 100;
+ }
+
msg->status = rspamd_fstring_new_init(task->err->message,
strlen(task->err->message));
ucl_object_insert_key(top, ucl_object_fromstring(task->err->message),
#define CONTENT_ENCODING_HEADER "Content-Encoding"
#define ACCEPT_ENCODING_HEADER "Accept-Enconding"
+/*
+ * Privileged message source controls.
+ *
+ * These headers make rspamd read the message from a local object that the
+ * client names (a file path or a POSIX shared memory object) instead of from
+ * the request body, so they are only honoured on transports that the accepting
+ * worker has explicitly marked as trusted via `allow_file_and_shm_inputs`.
+ *
+ * NB: FILENAME_HEADER ("Filename") is deliberately NOT part of this set: it is
+ * purely descriptive metadata about the message being sent inline and it never
+ * selects a message source.
+ */
+#define FILE_HEADER "File"
+#define PATH_HEADER "Path"
+#define SHM_HEADER "Shm"
+#define SHM_OFFSET_HEADER "Shm-Offset"
+#define SHM_LENGTH_HEADER "Shm-Length"
+
#ifdef __cplusplus
}
#endif
new_task->request_headers = kh_init(rspamd_req_headers_hash);
new_task->sock = -1;
new_task->flags |= (RSPAMD_TASK_FLAG_MIME);
+ /*
+ * Tasks that are not created by a network worker (rspamadm, Lua, embedded
+ * users) are local and trusted, so privileged message sources are allowed
+ * by default. The network workers clear this flag for every connection
+ * that is not permitted to use them.
+ */
+ new_task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT;
/* Default results chain */
rspamd_create_metric_result(new_task, NULL, -1);
}
}
-struct rspamd_task_map {
- gpointer begin;
- gulong len;
- int fd;
-};
-
-static void
-rspamd_task_unmapper(gpointer ud)
+/*
+ * Sanitise a client supplied path (or a POSIX shared memory object name) and
+ * copy it into `dst` which is `dstlen` bytes long.
+ *
+ * On success `dst` holds a NUL terminated, url decoded and unquoted string that
+ * is guaranteed to be non empty, to contain neither embedded NULs nor control
+ * characters, and to have been copied without truncation. Everything else is
+ * rejected here, before any syscall touches the name.
+ *
+ * `what` is used in the error messages only, e.g. "file path".
+ */
+static gboolean
+rspamd_task_sanitize_path(const rspamd_ftok_t *tok, char *dst, gsize dstlen,
+ const char *what, GError **err)
{
- struct rspamd_task_map *m = ud;
+ gsize len, i;
+
+ if (tok == NULL || tok->len == 0) {
+ g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
+ "empty %s", what);
+ return FALSE;
+ }
- munmap(m->begin, m->len);
- close(m->fd);
+ /*
+ * rspamd_strlcpy truncates silently, and operating on a truncated path is
+ * strictly worse than refusing it, so check the source length upfront
+ */
+ if ((gsize) tok->len >= dstlen) {
+ g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
+ "too long %s: %zu bytes, maximum is %zu", what,
+ (gsize) tok->len, (gsize) (dstlen - 1));
+ return FALSE;
+ }
+
+ /* An embedded NUL would silently cut the name short */
+ if (memchr(tok->begin, '\0', tok->len) != NULL) {
+ g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
+ "%s contains a NUL byte", what);
+ return FALSE;
+ }
+
+ rspamd_strlcpy(dst, tok->begin, tok->len + 1);
+ /* Decoding never expands the input, so it is safe to do it in place */
+ len = rspamd_url_decode(dst, dst, tok->len);
+ dst[len] = '\0';
+
+ if (len > 2 && dst[0] == '"' && dst[len - 1] == '"') {
+ /* Unquote in place, so that the caller always uses `dst` itself */
+ memmove(dst, dst + 1, len - 2);
+ len -= 2;
+ dst[len] = '\0';
+ }
+
+ if (len == 0) {
+ g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
+ "empty %s after decoding", what);
+ return FALSE;
+ }
+
+ /*
+ * A name is a single printable token, anything else is a mistake; this also
+ * catches a NUL that has been produced by the decoding step above
+ */
+ for (i = 0; i < len; i++) {
+ if ((unsigned char) dst[i] < ' ' || (unsigned char) dst[i] == 0x7f) {
+ g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
+ "invalid character at position %zu in %s", i, what);
+ return FALSE;
+ }
+ }
+
+ return TRUE;
}
-static void
-rspamd_shmem_segment_unmapper(gpointer ud)
+/*
+ * Read up to `len` bytes from `fd` into `buf`, dealing with short reads and
+ * EINTR. `*read_len` is set to the number of bytes that were actually available:
+ * an object that has shrunk under us yields less than `len` bytes instead of
+ * faulting, and an object that has grown is simply truncated to `len`.
+ */
+static gboolean
+rspamd_task_read_snapshot(int fd, char *buf, gsize len, gsize *read_len)
{
- struct rspamd_shmem_segment *seg = ud;
+ gsize total = 0;
+
+ while (total < len) {
+ ssize_t r = read(fd, buf + total, len - total);
- munmap(seg->map, seg->map_len);
- close(seg->fd);
+ if (r > 0) {
+ total += (gsize) r;
+ }
+ else if (r == 0) {
+ /* Truncated under us, whatever we have got is all there is */
+ break;
+ }
+ else if (errno == EINTR) {
+ continue;
+ }
+ else {
+ *read_len = total;
+
+ return FALSE;
+ }
+ }
+
+ *read_len = total;
+
+ return TRUE;
}
struct rspamd_shmem_segment *
gsize max_size,
GError **err)
{
- char namebuf[PATH_MAX], *name;
- gsize namelen, i;
+ char namebuf[PATH_MAX];
+ const char *name = namebuf;
gulong offset = 0, length = 0;
+ gsize page_size, aligned_offset, delta, map_len;
struct stat st;
int fd;
gpointer map;
+ char *data;
struct rspamd_shmem_segment *seg;
#ifdef HAVE_SANE_SHMEM
const char *ft = "shm";
+ const char *what = "shm segment name";
#else
const char *ft = "file";
+ const char *what = "file segment path";
#endif
- if (name_tok == NULL || name_tok->len == 0) {
- g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
- "empty %s segment name", ft);
- return NULL;
- }
-
- if (name_tok->len >= sizeof(namebuf)) {
- g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
- "too long %s segment name: %zu bytes", ft,
- (gsize) name_tok->len);
- return NULL;
- }
-
- rspamd_strlcpy(namebuf, name_tok->begin, name_tok->len + 1);
- /* Decoding never expands the input, so it is safe to do it in place */
- namelen = rspamd_url_decode(namebuf, namebuf, name_tok->len);
- namebuf[namelen] = '\0';
- name = namebuf;
-
- if (namelen > 2 && name[0] == '"' && name[namelen - 1] == '"') {
- /* Unquote the name */
- name[namelen - 1] = '\0';
- name++;
- namelen -= 2;
- }
-
- if (namelen == 0) {
- g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
- "empty %s segment name after decoding", ft);
+ if (!rspamd_task_sanitize_path(name_tok, namebuf, sizeof(namebuf), what,
+ err)) {
return NULL;
}
- /* A segment name is a single printable token, anything else is a mistake */
- for (i = 0; i < namelen; i++) {
- if ((unsigned char) name[i] < ' ' || (unsigned char) name[i] == 0x7f) {
- g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
- "invalid character at position %zu in %s segment name",
- i, ft);
- return NULL;
- }
- }
-
if (offset_tok != NULL &&
!rspamd_strtoul(offset_tok->begin, offset_tok->len, &offset)) {
g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
return NULL;
}
- map = mmap(NULL, st.st_size, PROT_READ, MAP_SHARED, fd, 0);
+ seg = rspamd_mempool_alloc0(pool, sizeof(*seg));
+ seg->name = rspamd_mempool_strdup(pool, name);
+ seg->offset = offset;
+ seg->data_len = length;
+ seg->map = NULL;
+ seg->map_len = 0;
+ seg->fd = -1;
+
+ if (length == 0) {
+ /* An empty window must never map the whole object */
+ close(fd);
+ seg->data = rspamd_mempool_strdup(pool, "");
+
+ return seg;
+ }
+
+ /*
+ * Map merely the window that is really needed: the offset is rounded down
+ * to a page boundary and the length is extended by the very same delta, so
+ * that a small payload inside a huge object never maps that whole object.
+ */
+ page_size = (gsize) sysconf(_SC_PAGESIZE);
+
+ if (page_size == 0 || page_size == (gsize) -1) {
+ page_size = 4096;
+ }
+
+ aligned_offset = ((gsize) offset / page_size) * page_size;
+ delta = (gsize) offset - aligned_offset;
+ map_len = (gsize) length + delta;
+
+ map = mmap(NULL, map_len, PROT_READ, MAP_SHARED, fd, (off_t) aligned_offset);
if (map == MAP_FAILED) {
g_set_error(err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
return NULL;
}
- seg = rspamd_mempool_alloc(pool, sizeof(*seg));
- seg->name = rspamd_mempool_strdup(pool, name);
- seg->map = map;
- seg->map_len = st.st_size;
- seg->offset = offset;
- seg->data = (const char *) map + offset;
- seg->data_len = length;
- seg->fd = fd;
+ /*
+ * The backing object belongs to the client and it can be truncated or
+ * rewritten at any moment, so a live mapping handed over to the parser
+ * could fault later on. Snapshot the window into the pool and drop both the
+ * mapping and the descriptor right away.
+ */
+ data = rspamd_mempool_alloc(pool, length);
+ memcpy(data, (const char *) map + delta, length);
+ munmap(map, map_len);
+ close(fd);
- rspamd_mempool_add_destructor(pool, rspamd_shmem_segment_unmapper, seg);
+ seg->data = data;
return seg;
}
+gboolean
+rspamd_task_allow_file_shm_input(struct rspamd_task *task)
+{
+ if (task == NULL) {
+ return FALSE;
+ }
+
+ return (task->protocol_flags &
+ RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT) != 0;
+}
+
+gboolean
+rspamd_task_has_file_shm_input(struct rspamd_task *task)
+{
+ /* The lookup hash is case insensitive, so lowercase names are enough */
+ static const char *privileged_headers[] = {
+ "file",
+ "path",
+ "shm",
+ "shm-offset",
+ "shm-length",
+ };
+ unsigned int i;
+
+ if (task == NULL || task->request_headers == NULL) {
+ return FALSE;
+ }
+
+ for (i = 0; i < G_N_ELEMENTS(privileged_headers); i++) {
+ if (rspamd_task_get_request_header(task, privileged_headers[i]) != NULL) {
+ return TRUE;
+ }
+ }
+
+ return FALSE;
+}
+
gboolean
rspamd_task_load_message(struct rspamd_task *task,
struct rspamd_http_message *msg, const char *start, gsize len)
{
- char filepath[PATH_MAX], *fp;
- int fd, flen;
+ char filepath[PATH_MAX];
+ int fd;
rspamd_ftok_t *tok;
- gpointer map;
+ gsize max_message;
struct stat st;
- struct rspamd_task_map *m;
if (msg && task->cmd != CMD_CHECK_V3) {
- rspamd_protocol_handle_headers(task, msg);
+ /*
+ * Header parsing rejects privileged message source controls that this
+ * connection may not use, so its verdict must be honoured here: it has
+ * already set task->err and the request must not be processed further.
+ */
+ if (!rspamd_protocol_handle_headers(task, msg)) {
+ return FALSE;
+ }
+ }
+
+ /*
+ * File and shm inputs make rspamd open an arbitrary local object of the
+ * client's choosing, so they are only honoured on the transports that the
+ * accepting worker has explicitly marked as privileged. This is checked
+ * before any of the values is even looked at, hence no stat/open/mmap and
+ * no shm_open can happen for a connection that is not allowed to use them.
+ */
+ if (rspamd_task_has_file_shm_input(task) &&
+ !rspamd_task_allow_file_shm_input(task)) {
+ msg_info_task("deny file/shm message source from %s: this connection is "
+ "not permitted to use privileged inputs; set "
+ "`allow_file_and_shm_inputs = true` for this worker if all "
+ "of its clients are trusted",
+ rspamd_inet_address_to_string_pretty(task->client_addr));
+ g_set_error(&task->err, rspamd_task_quark(), 400,
+ "file and shm message sources are not allowed "
+ "on this connection");
+ /* This is a client error, so report it as a genuine 400 */
+ task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_VERBATIM_ERR_CODE;
+
+ return FALSE;
}
+ max_message = (task->cfg != NULL) ? task->cfg->max_message : 0;
+
tok = rspamd_task_get_request_header(task, "shm");
if (tok) {
len_tok = rspamd_task_get_request_header(task, "shm-length");
seg = rspamd_shmem_segment_map(task->task_pool, tok, off_tok, len_tok,
- task->cfg ? task->cfg->max_message : 0,
- &task->err);
+ max_message, &task->err);
if (seg == NULL) {
return FALSE;
task->msg.len = seg->data_len;
msg_info_task("loaded message from shared memory %s "
- "(%uz size, %uz offset), fd=%d",
- seg->name, seg->data_len, seg->offset, seg->fd);
+ "(%uz size, %uz offset)",
+ seg->name, seg->data_len, seg->offset);
}
else {
/* Try file */
if (tok) {
debug_task("want to scan file %T", tok);
- size_t r = rspamd_strlcpy(filepath, tok->begin,
- MIN(sizeof(filepath), tok->len + 1));
+ if (!rspamd_task_sanitize_path(tok, filepath, sizeof(filepath),
+ "file path", &task->err)) {
+ return FALSE;
+ }
- rspamd_url_decode(filepath, filepath, r + 1);
- flen = strlen(filepath);
+ /*
+ * Open first and validate the descriptor afterwards: a path based
+ * stat(2) says nothing about the object that is actually opened.
+ * O_NONBLOCK: never block the worker on opening a special file.
+ */
+ fd = open(filepath, O_RDONLY | O_NONBLOCK);
- if (filepath[0] == '"' && flen > 2) {
- /* We need to unquote filepath */
- fp = &filepath[1];
- fp[flen - 2] = '\0';
- }
- else {
- fp = &filepath[0];
+ if (fd == -1) {
+ g_set_error(&task->err, rspamd_task_quark(),
+ RSPAMD_PROTOCOL_ERROR,
+ "Cannot open file (%s): %s", filepath,
+ strerror(errno));
+ return FALSE;
}
- if (stat(fp, &st) == -1) {
- g_set_error(&task->err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
- "Invalid file (%s): %s", fp, strerror(errno));
+ if (fstat(fd, &st) == -1) {
+ g_set_error(&task->err, rspamd_task_quark(),
+ RSPAMD_PROTOCOL_ERROR,
+ "Cannot stat file (%s): %s", filepath,
+ strerror(errno));
+ close(fd);
+
return FALSE;
}
if (!S_ISREG(st.st_mode)) {
- g_set_error(&task->err, rspamd_task_quark(), RSPAMD_PROTOCOL_ERROR,
- "Not a regular file (%s)", fp);
+ g_set_error(&task->err, rspamd_task_quark(),
+ RSPAMD_PROTOCOL_ERROR,
+ "Not a regular file (%s)", filepath);
+ close(fd);
+
+ return FALSE;
+ }
+
+ if (st.st_size < 0 ||
+ (uint64_t) st.st_size > (uint64_t) G_MAXSIZE) {
+ g_set_error(&task->err, rspamd_task_quark(),
+ RSPAMD_PROTOCOL_ERROR,
+ "Invalid size of file (%s): %lld", filepath,
+ (long long) st.st_size);
+ close(fd);
+
+ return FALSE;
+ }
+
+ /*
+ * A file input must obey the very same limit as an inline body,
+ * and it has to be enforced before anything is read at all
+ */
+ if (max_message > 0 && (gsize) st.st_size > max_message) {
+ g_set_error(&task->err, rspamd_task_quark(),
+ RSPAMD_PROTOCOL_ERROR,
+ "Too large file (%s): %zu, maximum is %zu",
+ filepath, (gsize) st.st_size, max_message);
+ close(fd);
+
return FALSE;
}
if (G_UNLIKELY(st.st_size == 0)) {
- /* Empty file */
+ /* Empty file, exactly as an empty inline message */
+ close(fd);
task->flags |= RSPAMD_TASK_FLAG_EMPTY;
task->msg.begin = rspamd_mempool_strdup(task->task_pool, "");
task->msg.len = 0;
}
else {
- /* O_NONBLOCK: never block the worker on opening a special file */
- fd = open(fp, O_RDONLY | O_NONBLOCK);
-
- if (fd == -1) {
+ /*
+ * The file belongs to the client and it can be truncated at any
+ * moment, which would turn a MAP_SHARED mapping into a fault in
+ * the middle of the parser. Take a bounded snapshot instead, so
+ * that the parser is never given a range that can go away.
+ */
+ gsize nread = 0;
+ char *buf = rspamd_mempool_alloc(task->task_pool,
+ (gsize) st.st_size);
+
+ if (!rspamd_task_read_snapshot(fd, buf, (gsize) st.st_size,
+ &nread)) {
g_set_error(&task->err, rspamd_task_quark(),
RSPAMD_PROTOCOL_ERROR,
- "Cannot open file (%s): %s", fp, strerror(errno));
- return FALSE;
- }
-
- /* Re-check after opening to close the stat/open race */
- if (fstat(fd, &st) == -1 || !S_ISREG(st.st_mode) ||
- st.st_size == 0) {
+ "Cannot read file (%s): %s", filepath,
+ strerror(errno));
close(fd);
- g_set_error(&task->err, rspamd_task_quark(),
- RSPAMD_PROTOCOL_ERROR,
- "Cannot use file (%s): not a regular non-empty file",
- fp);
- return FALSE;
- }
- map = mmap(NULL, st.st_size, PROT_READ, MAP_SHARED, fd, 0);
-
-
- if (map == MAP_FAILED) {
- close(fd);
- g_set_error(&task->err, rspamd_task_quark(),
- RSPAMD_PROTOCOL_ERROR,
- "Cannot mmap file (%s): %s", fp, strerror(errno));
return FALSE;
}
- task->msg.begin = map;
- task->msg.len = st.st_size;
- m = rspamd_mempool_alloc(task->task_pool, sizeof(*m));
- m->begin = map;
- m->len = st.st_size;
- m->fd = fd;
+ close(fd);
- rspamd_mempool_add_destructor(task->task_pool, rspamd_task_unmapper, m);
+ task->msg.begin = buf;
+ task->msg.len = nread;
}
- task->msg.fpath = rspamd_mempool_strdup(task->task_pool, fp);
+ task->msg.fpath = rspamd_mempool_strdup(task->task_pool, filepath);
task->flags |= RSPAMD_TASK_FLAG_FILE;
- msg_info_task("loaded message from file %s", fp);
+ msg_info_task("loaded message from file %s (%uz bytes)", filepath,
+ task->msg.len);
}
else {
/* Plain data */
#define RSPAMD_TASK_FLAG_MESSAGE_REWRITE (1u << 24u)
#define RSPAMD_TASK_FLAG_MAX_SHIFT (25u)
-/* Request has been done by a local client */
+/*
+ * Request has been done by a local client.
+ *
+ * This flag is purely informational and it MUST be derived from the actual
+ * accepted transport and the peer identity (e.g. an AF_UNIX socket or a
+ * loopback peer address), NEVER from `User-Agent` or any other value that the
+ * client itself controls: otherwise any remote client can simply claim to be
+ * local. Consequently it MUST NOT be used to grant file, shm, authentication
+ * or any write privileges; use
+ * RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT for privileged inputs.
+ */
#define RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT (1u << 1u)
/* Request has been sent via milter */
#define RSPAMD_TASK_PROTOCOL_FLAG_MILTER (1u << 2u)
#define RSPAMD_TASK_PROTOCOL_FLAG_MULTIPART_V3 (1u << 7u)
/* v3 request metadata part was msgpack-serialized (mirror it in the reply) */
#define RSPAMD_TASK_PROTOCOL_FLAG_V3_MSGPACK (1u << 8u)
-#define RSPAMD_TASK_PROTOCOL_FLAG_MAX_SHIFT (8u)
+/*
+ * Set when the transport this task was accepted on is permitted to carry
+ * privileged message source controls (File/Path/Shm*). Derived solely from
+ * the accepted socket and the worker's allow_file_and_shm_inputs option;
+ * NEVER from client supplied headers or request metadata.
+ */
+#define RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT (1u << 9u)
+/*
+ * Emit `task->err->code` verbatim as the HTTP status of the error reply.
+ *
+ * By default rspamd_protocol_write_reply() folds an error code into the 5xx
+ * range (`500 + code % 100`), which turns a client error such as 400 into a
+ * misleading `500 Internal Server Error`. Rejections that are genuinely the
+ * client's fault set this flag so that the real status reaches the client.
+ * Only codes in the valid HTTP range are honoured.
+ */
+#define RSPAMD_TASK_PROTOCOL_FLAG_VERBATIM_ERR_CODE (1u << 10u)
+#define RSPAMD_TASK_PROTOCOL_FLAG_MAX_SHIFT (10u)
#define RSPAMD_TASK_IS_SKIPPED(task) (G_UNLIKELY((task)->flags & RSPAMD_TASK_FLAG_SKIP))
#define RSPAMD_TASK_IS_SPAMC(task) (G_UNLIKELY((task)->cmd == CMD_CHECK_SPAMC))
struct rspamd_http_message *msg,
const char *start, gsize len);
+/**
+ * Returns TRUE if this task may use privileged message source controls
+ * (File/Path/Shm/Shm-Offset/Shm-Length and their V3 metadata equivalents).
+ *
+ * The answer depends only on RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT,
+ * which the accepting worker derives from the transport, never from anything
+ * that the client sends.
+ * @param task
+ * @return TRUE if privileged message sources are permitted for this task
+ */
+gboolean rspamd_task_allow_file_shm_input(struct rspamd_task *task);
+
+/**
+ * Returns TRUE if the task carries any privileged message source control
+ * in its request headers. Cheap header presence test, performs no I/O.
+ * @param task
+ * @return TRUE if any of `file`, `path`, `shm`, `shm-offset` or `shm-length`
+ * request headers is present
+ */
+gboolean rspamd_task_has_file_shm_input(struct rspamd_task *task);
+
/**
* A shared memory segment passed by a local client instead of the message body
+ *
+ * `data` always points to a private snapshot allocated from the pool, so it
+ * stays valid (and stable) even if the client mutates or truncates the backing
+ * object afterwards. The `map`, `map_len` and `fd` fields are kept for
+ * introspection only: the segment owns neither a mapping nor a descriptor once
+ * it has been returned, so they are always NULL, 0 and -1 respectively.
*/
struct rspamd_shmem_segment {
const char *name; /**< decoded segment name (allocated from the pool) */
- const char *data; /**< payload start, that is `map` + `offset` */
+ const char *data; /**< payload snapshot (allocated from the pool) */
gsize data_len; /**< payload length */
- gpointer map; /**< start of the mapping */
- gsize map_len; /**< length of the whole mapping */
- gsize offset; /**< payload offset within the mapping */
- int fd; /**< descriptor kept open while the mapping is alive */
+ gpointer map; /**< unused, always NULL */
+ gsize map_len; /**< unused, always 0 */
+ gsize offset; /**< payload offset within the backing object */
+ int fd; /**< unused, always -1 */
};
/**
- * Open and map a shared memory segment described by the `Shm`, `Shm-Offset` and
+ * Open and read a shared memory segment described by the `Shm`, `Shm-Offset` and
* `Shm-Length` values of a request.
*
* All inputs come from a (trusted, but not necessarily sane) client, so they are
* fully validated here: the resulting payload is always guaranteed to lie within
- * the mapping. The mapping is unmapped and the descriptor is closed when `pool`
- * is destroyed.
+ * the backing object. Only the requested window is mapped (rounded down to a
+ * page boundary), it is copied into `pool` and unmapped straight away, so no
+ * mapping and no descriptor outlive this call.
*
- * @param pool pool to allocate the result from and to attach the mapping to
+ * @param pool pool to allocate the result and the payload snapshot from
* @param name_tok value of the `Shm` header (url encoded, optionally quoted)
* @param offset_tok value of the `Shm-Offset` header or NULL
* @param length_tok value of the `Shm-Length` header or NULL
* @param max_size reject payloads larger than this value (0 disables the check)
* @param err error to set on failure
- * @return mapped segment or NULL on any error
+ * @return segment holding a snapshot of the payload or NULL on any error
*/
struct rspamd_shmem_segment *rspamd_shmem_segment_map(rspamd_mempool_t *pool,
const rspamd_ftok_t *name_tok,
return FALSE;
}
+
+gboolean
+rspamd_worker_is_unix_socket(struct rspamd_worker *worker, int fd)
+{
+ GList *cur;
+ struct rspamd_worker_listen_socket *ls;
+
+ if (worker == NULL || worker->cf == NULL) {
+ return FALSE;
+ }
+
+ cur = worker->cf->listen_socks;
+
+ while (cur) {
+ ls = (struct rspamd_worker_listen_socket *) cur->data;
+
+ if (ls->fd == fd) {
+ return ls->addr != NULL &&
+ rspamd_inet_address_get_af(ls->addr) == AF_UNIX;
+ }
+
+ cur = g_list_next(cur);
+ }
+
+ return FALSE;
+}
+
+gboolean
+rspamd_worker_addr_is_loopback(const rspamd_inet_addr_t *addr)
+{
+ int af;
+ socklen_t slen;
+ const struct sockaddr *sa;
+
+ if (addr == NULL) {
+ return FALSE;
+ }
+
+ af = rspamd_inet_address_get_af(addr);
+
+ if (af == AF_UNIX) {
+ /* Unix sockets are always local by definition */
+ return TRUE;
+ }
+
+ sa = rspamd_inet_address_get_sa(addr, &slen);
+
+ if (sa == NULL) {
+ return FALSE;
+ }
+
+ if (af == AF_INET) {
+ const struct sockaddr_in *sin = (const struct sockaddr_in *) sa;
+
+ return (ntohl(sin->sin_addr.s_addr) & 0xff000000U) == 0x7f000000U;
+ }
+ else if (af == AF_INET6) {
+ const struct sockaddr_in6 *sin6 = (const struct sockaddr_in6 *) sa;
+
+ if (IN6_IS_ADDR_LOOPBACK(&sin6->sin6_addr)) {
+ return TRUE;
+ }
+
+ /*
+ * The accept path normally unwraps v4-mapped addresses, but be
+ * defensive and treat ::ffff:127.0.0.0/104 as loopback as well.
+ */
+ if (IN6_IS_ADDR_V4MAPPED(&sin6->sin6_addr)) {
+ uint32_t v4;
+
+ memcpy(&v4, &sin6->sin6_addr.s6_addr[12], sizeof(v4));
+
+ return (ntohl(v4) & 0xff000000U) == 0x7f000000U;
+ }
+ }
+
+ /*
+ * Everything else, including link-local and site-local addresses, is
+ * reachable from other hosts and hence is NOT loopback.
+ */
+ return FALSE;
+}
+
+/*
+ * Try to find a human readable bind line that produced the given listen
+ * address, so that the operator can grep for it in the configuration.
+ */
+static const char *
+rspamd_worker_listen_bind_line(struct rspamd_worker *worker,
+ const rspamd_inet_addr_t *addr)
+{
+ struct rspamd_worker_bind_conf *bcf;
+ unsigned int i;
+
+ LL_FOREACH(worker->cf->bind_conf, bcf)
+ {
+ if (bcf->addrs == NULL) {
+ continue;
+ }
+
+ for (i = 0; i < bcf->addrs->len; i++) {
+ const rspamd_inet_addr_t *cur = g_ptr_array_index(bcf->addrs, i);
+
+ if (rspamd_inet_address_port_equal(cur, addr)) {
+ return bcf->bind_line ? bcf->bind_line : bcf->name;
+ }
+ }
+ }
+
+ return NULL;
+}
+
+void rspamd_worker_warn_file_shm_inputs(struct rspamd_worker *worker,
+ const char *worker_name,
+ gboolean enabled)
+{
+ GList *cur;
+ GHashTable *seen;
+ struct rspamd_worker_listen_socket *ls;
+ const char *bind_line;
+ char listener[512];
+
+ if (!enabled || worker == NULL || worker->cf == NULL) {
+ return;
+ }
+
+ if (worker_name == NULL) {
+ worker_name = "unknown";
+ }
+
+ /* A wildcard bind resolves to both address families, warn once per line */
+ seen = g_hash_table_new_full(g_str_hash, g_str_equal, g_free, NULL);
+
+ for (cur = worker->cf->listen_socks; cur != NULL; cur = g_list_next(cur)) {
+ ls = (struct rspamd_worker_listen_socket *) cur->data;
+
+ if (ls == NULL || ls->addr == NULL) {
+ continue;
+ }
+
+ if (rspamd_inet_address_get_af(ls->addr) == AF_UNIX) {
+ /* Access is limited by the filesystem permissions, that is fine */
+ continue;
+ }
+
+ bind_line = rspamd_worker_listen_bind_line(worker, ls->addr);
+
+ if (bind_line != NULL) {
+ rspamd_snprintf(listener, sizeof(listener), "%s (bind_socket = \"%s\")",
+ rspamd_inet_address_to_string_pretty(ls->addr),
+ bind_line);
+ }
+ else {
+ rspamd_snprintf(listener, sizeof(listener), "%s",
+ rspamd_inet_address_to_string_pretty(ls->addr));
+ }
+
+ if (g_hash_table_contains(seen, listener)) {
+ continue;
+ }
+
+ g_hash_table_insert(seen, g_strdup(listener), GINT_TO_POINTER(1));
+
+ if (rspamd_worker_addr_is_loopback(ls->addr)) {
+ msg_warn("SECURITY: worker %s accepts privileged File/Path/Shm "
+ "message source inputs on the TCP listener '%s'; any client "
+ "that can reach this port can make rspamd read arbitrary "
+ "files readable by the rspamd user. Prefer a unix socket "
+ "protected by filesystem permissions, or set "
+ "'allow_file_and_shm_inputs = false' in the worker section. "
+ "This option will default to false in the next major release",
+ worker_name, listener);
+ }
+ else {
+ msg_err("SECURITY: worker %s accepts privileged File/Path/Shm "
+ "message source inputs on the NON-LOOPBACK TCP listener '%s'; "
+ "any client that can reach this port can make rspamd read "
+ "arbitrary files readable by the rspamd user and disclose "
+ "their content. Bind to a unix socket protected by filesystem "
+ "permissions, or set 'allow_file_and_shm_inputs = false' in "
+ "the worker section. This option will default to false in the "
+ "next major release",
+ worker_name, listener);
+ }
+ }
+
+ g_hash_table_destroy(seen);
+}
struct rspamd_lang_detector *lang_det;
gboolean is_spam;
gboolean is_read_only;
+ /*
+ * Whether this connection may use the privileged File/Path/Shm message
+ * source inputs. Derived once, at accept time, from the transport the
+ * connection arrived on and the worker's `allow_file_and_shm_inputs`
+ * option; never from anything the client sends. Handlers outside
+ * controller.c (which cannot see the private worker context) must consult
+ * this before creating a task that can reach rspamd_task_load_message.
+ */
+ gboolean allow_file_shm_input;
};
/**
*/
gboolean rspamd_worker_has_ssl_socket(struct rspamd_worker *worker);
+/**
+ * Check if the given listen fd is a unix domain socket for this worker.
+ * The answer is derived from the worker listen sockets configuration and
+ * never from any client supplied data (e.g. headers), so it is safe to use
+ * it for privilege decisions.
+ * @param worker
+ * @param fd listen fd from accept event
+ * @return TRUE if the listening socket is AF_UNIX
+ */
+gboolean rspamd_worker_is_unix_socket(struct rspamd_worker *worker, int fd);
+
+/**
+ * Strict loopback test for an accepted peer address.
+ * Unlike `rspamd_inet_address_is_local` this function returns TRUE **only**
+ * for AF_UNIX, IPv4 127/8 and IPv6 ::1; link-local and site-local addresses
+ * are explicitly rejected as those are reachable from other hosts.
+ * @param addr peer address as returned by `rspamd_accept_from_socket`
+ * @return TRUE if the peer is strictly loopback
+ */
+gboolean rspamd_worker_addr_is_loopback(const rspamd_inet_addr_t *addr);
+
+/**
+ * Emit a startup warning for each TCP listener of `worker` when privileged
+ * File/Path/Shm message source inputs are enabled. The message is escalated
+ * to an error level for listeners that are not strictly loopback.
+ * @param worker worker being started
+ * @param worker_name name used in the log line ("normal", "controller", ...)
+ * @param enabled when FALSE the function is a no-op
+ */
+void rspamd_worker_warn_file_shm_inputs(struct rspamd_worker *worker,
+ const char *worker_name,
+ gboolean enabled);
+
#ifdef WITH_HYPERSCAN
struct rspamd_control_command;
task = rspamd_task_new(session->wrk, session->cfg, NULL,
session->lang_det, conn_ent->rt->event_loop, FALSE);
task->cfg = ctx->cfg;
+
+ /*
+ * This task loads its message straight from the request below, so it has to
+ * inherit the transport's privileged input capability just like the tasks
+ * that controller.c creates itself. Otherwise /fuzzyadd and /fuzzydel would
+ * keep honouring a client supplied File/Shm source on a connection where
+ * every other endpoint refuses it.
+ */
+ if (!session->allow_file_shm_input) {
+ task->protocol_flags &= ~RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT;
+ }
saved = rspamd_mempool_alloc0(session->pool, sizeof(int));
fuzzy_module_ctx = fuzzy_get_context(ctx->cfg);
GArray *cmp_refs;
/* Maximum count for retries */
unsigned int max_retries;
+ /* Maximum number of simultaneous connections, 0 means unlimited */
+ unsigned int max_connections;
+ /* Maximum number of simultaneous connections per source IP, 0 - unlimited */
+ unsigned int max_connections_per_source;
+ /* Pending connections per source address, created on demand */
+ GHashTable *conns_per_source;
gboolean encrypted_only;
+ /* Whether privileged File/Path/Shm message sources are honoured over TCP */
+ gboolean allow_file_and_shm_inputs;
/* If we have self_scanning backends, we need to work as a normal worker */
gboolean has_self_scan;
/* It is not HTTP but milter proxy */
int retries;
ref_entry_t ref;
enum rspamd_proxy_session_flags flags;
+ /*
+ * Both are derived from the accepted transport at accept time and never
+ * from anything that the client sends
+ */
+ gboolean allow_file_shm;
+ gboolean local_client;
+ /* Key in ctx->conns_per_source, owned by the session, NULL if not counted */
+ char *source_key;
/* ESMTP arguments from milter session */
GHashTable *mail_esmtp_args;
ctx->max_retries = DEFAULT_RETRIES;
ctx->spam_header = RSPAMD_MILTER_SPAM_HEADER;
ctx->log_tag_type = RSPAMD_PROXY_LOG_TAG_SESSION; /* Default to session tag */
+ /*
+ * Permissive code default for this release to keep the existing setups
+ * working; this is going to become FALSE in the next major release
+ */
+ ctx->allow_file_and_shm_inputs = TRUE;
+ ctx->max_connections = 0; /* Unlimited by default */
+ ctx->max_connections_per_source = 0; /* Unlimited by default */
rspamd_rcl_register_worker_option(cfg,
type,
G_STRUCT_OFFSET(struct rspamd_proxy_ctx, encrypted_only),
0,
"Allow only encrypted connections");
+ rspamd_rcl_register_worker_option(cfg,
+ type,
+ "allow_file_and_shm_inputs",
+ rspamd_rcl_parse_struct_boolean,
+ ctx,
+ G_STRUCT_OFFSET(struct rspamd_proxy_ctx, allow_file_and_shm_inputs),
+ 0,
+ "Honour privileged File/Path/Shm message source inputs on TCP "
+ "connections, and allow shared memory forwarding to loopback "
+ "upstreams; unix socket clients and upstreams always may use them. "
+ "Default: true, will become false in the next major release");
+ rspamd_rcl_register_worker_option(cfg,
+ type,
+ "max_connections",
+ rspamd_rcl_parse_struct_integer,
+ ctx,
+ G_STRUCT_OFFSET(struct rspamd_proxy_ctx, max_connections),
+ RSPAMD_CL_FLAG_UINT,
+ "Maximum number of simultaneous connections per worker "
+ "(default: 0, meaning unlimited)");
+ rspamd_rcl_register_worker_option(cfg,
+ type,
+ "max_connections_per_source",
+ rspamd_rcl_parse_struct_integer,
+ ctx,
+ G_STRUCT_OFFSET(struct rspamd_proxy_ctx, max_connections_per_source),
+ RSPAMD_CL_FLAG_UINT,
+ "Maximum number of simultaneous connections from a single source "
+ "IP per worker; unix socket clients are never counted "
+ "(default: 0, meaning unlimited)");
rspamd_rcl_register_worker_option(cfg,
type,
"ssl_cert",
lua_settop(L, err_idx - 1);
}
+/*
+ * Per source connection accounting.
+ *
+ * The key is the peer address with the port stripped, so that a single host
+ * cannot occupy all of the available slots. Unix socket peers are never
+ * counted: all of them would share the very same key and they are protected by
+ * the filesystem permissions anyway.
+ *
+ * The accounting follows worker->nconns exactly: a connection is reserved in
+ * proxy_accept_socket and released in proxy_conn_release, which is called from
+ * the session destructor for HTTP sessions and from the milter finish/error
+ * handlers for milter ones (a milter connection outlives the per message
+ * sessions created by proxy_session_refresh).
+ */
+static char *
+proxy_source_key_new(struct rspamd_proxy_ctx *ctx, rspamd_inet_addr_t *addr)
+{
+ if (ctx->max_connections_per_source == 0 || addr == NULL ||
+ rspamd_inet_address_get_af(addr) == AF_UNIX) {
+ return NULL;
+ }
+
+ return g_strdup(rspamd_inet_address_to_string(addr));
+}
+
+static gboolean
+proxy_source_conn_reserve(struct rspamd_proxy_ctx *ctx, const char *key)
+{
+ unsigned int cnt;
+
+ if (ctx->conns_per_source == NULL) {
+ ctx->conns_per_source = g_hash_table_new_full(g_str_hash, g_str_equal,
+ g_free, NULL);
+ }
+
+ cnt = GPOINTER_TO_UINT(g_hash_table_lookup(ctx->conns_per_source, key));
+
+ if (cnt >= ctx->max_connections_per_source) {
+ return FALSE;
+ }
+
+ g_hash_table_insert(ctx->conns_per_source, g_strdup(key),
+ GUINT_TO_POINTER(cnt + 1));
+
+ return TRUE;
+}
+
+/*
+ * Releases both the global and the per source connection slots. Idempotent:
+ * the source key is dropped once it has been accounted for.
+ */
+static void
+proxy_conn_release(struct rspamd_proxy_session *session)
+{
+ session->worker->nconns--;
+
+ if (session->source_key) {
+ if (session->ctx->conns_per_source) {
+ unsigned int cnt = GPOINTER_TO_UINT(
+ g_hash_table_lookup(session->ctx->conns_per_source,
+ session->source_key));
+
+ if (cnt <= 1) {
+ g_hash_table_remove(session->ctx->conns_per_source,
+ session->source_key);
+ }
+ else {
+ g_hash_table_insert(session->ctx->conns_per_source,
+ g_strdup(session->source_key),
+ GUINT_TO_POINTER(cnt - 1));
+ }
+ }
+
+ g_free(session->source_key);
+ session->source_key = NULL;
+ }
+}
+
static void
proxy_session_dtor(struct rspamd_proxy_session *session)
{
* because proxy_session_refresh creates a new session per message — each
* intermediate session would otherwise decrement nconns prematurely. */
if (!session->ctx->milter) {
- session->worker->nconns--;
+ proxy_conn_release(session);
}
+ g_free(session->source_key);
g_free(session);
}
nsession);
nsession->client_addr = session->client_addr;
session->client_addr = NULL;
+ /*
+ * The transport derived properties and the connection accounting belong to
+ * the MTA connection, not to a single message, so they are moved over
+ */
+ nsession->allow_file_shm = session->allow_file_shm;
+ nsession->local_client = session->local_client;
+ nsession->source_key = session->source_key;
+ session->source_key = NULL;
nsession->ctx = session->ctx;
nsession->worker = session->worker;
nsession->pool = rspamd_mempool_new_short_lived("proxy");
return nsession;
}
+/*
+ * Message source controls that make rspamd read an object of the client's
+ * choosing instead of the request body.
+ *
+ * `File` and `Path` are aliases of each other and they are consumed by the
+ * proxy itself; `Shm*` are reserved hop-by-hop headers that only ever have a
+ * meaning between the two ends of one connection and that are regenerated by
+ * the HTTP layer for the upstream connection.
+ *
+ * Note that a query argument is turned into a request header by
+ * rspamd_protocol_handle_url() at the *upstream*, so the URL has to be
+ * sanitised as thoroughly as the headers are.
+ */
+static const char *proxy_privileged_file_args[] = {"File", "Path"};
+static const char *proxy_privileged_shm_args[] = {"Shm", "Shm-Offset",
+ "Shm-Length"};
+
+static gboolean
+proxy_has_query_arg(GHashTable *query_args, const char *name)
+{
+ rspamd_ftok_t srch;
+
+ srch.begin = name;
+ srch.len = strlen(name);
+
+ return g_hash_table_lookup(query_args, &srch) != NULL;
+}
+
+/*
+ * Rebuilds msg->url without the listed query arguments. Does nothing at all
+ * unless at least one of them is really there.
+ */
+static void
+proxy_strip_query_args(struct rspamd_http_message *msg,
+ const char **names, unsigned int nnames)
+{
+ struct http_parser_url u;
+ GHashTable *query_args;
+ GHashTableIter it;
+ gpointer k, v;
+ rspamd_fstring_t *new_url;
+ unsigned int i;
+ gboolean found = FALSE;
+
+ if (msg->url == NULL || msg->url->len == 0) {
+ return;
+ }
+
+ if (http_parser_parse_url(RSPAMD_FSTRING_DATA(msg->url),
+ RSPAMD_FSTRING_LEN(msg->url), 0, &u) != 0) {
+ return;
+ }
+
+ if (!(u.field_set & (1 << UF_QUERY))) {
+ return;
+ }
+
+ query_args = rspamd_http_message_parse_query(msg);
+
+ for (i = 0; i < nnames; i++) {
+ if (proxy_has_query_arg(query_args, names[i])) {
+ found = TRUE;
+ break;
+ }
+ }
+
+ if (!found) {
+ g_hash_table_unref(query_args);
+
+ return;
+ }
+
+ new_url = rspamd_fstring_new_init(RSPAMD_FSTRING_DATA(msg->url),
+ u.field_data[UF_QUERY].off);
+ new_url = rspamd_fstring_append(new_url, "?", 1);
+
+ g_hash_table_iter_init(&it, query_args);
+
+ while (g_hash_table_iter_next(&it, &k, &v)) {
+ const rspamd_ftok_t *key_tok = k, *tok = v;
+ gboolean skip = FALSE;
+
+ for (i = 0; i < nnames; i++) {
+ rspamd_ftok_t srch;
+
+ srch.begin = names[i];
+ srch.len = strlen(names[i]);
+
+ if (rspamd_ftok_icase_equal(key_tok, &srch)) {
+ skip = TRUE;
+ break;
+ }
+ }
+
+ if (!skip) {
+ rspamd_printf_fstring(&new_url, "%T=%T&", key_tok, tok);
+ }
+ }
+
+ /* Erase last character (might be either & or ?) */
+ rspamd_fstring_erase(new_url, new_url->len - 1, 1);
+
+ rspamd_fstring_free(msg->url);
+ msg->url = new_url;
+
+ g_hash_table_unref(query_args);
+}
+
+/*
+ * Ingress sanitiser for the privileged message source controls.
+ *
+ * Called on the client message before anything is forwarded or scanned, and
+ * before proxy_check_file() may open anything at all. When this connection is
+ * not permitted to use privileged inputs:
+ *
+ * - a File/Path header or query argument is a hard error (400), the named
+ * object is never opened, mapped nor stat'ed;
+ * - the reserved Shm* headers and query arguments are removed, so that a
+ * client can never override the values that the proxy generates itself.
+ *
+ * The decision comes from the accepted transport only, never from the request.
+ */
+static gboolean
+proxy_sanitize_privileged_inputs(struct rspamd_http_message *msg,
+ struct rspamd_proxy_session *session,
+ int *err_code, const char **err_status)
+{
+ unsigned int i;
+ GHashTable *query_args = NULL;
+ struct http_parser_url u;
+ gboolean has_file_arg = FALSE;
+
+ if (session->allow_file_shm) {
+ return TRUE;
+ }
+
+ /*
+ * rspamd_http_message_remove_header() drops every instance of a header at
+ * once (all of the duplicates live in a single list), but loop anyway so
+ * that a client supplied duplicate can never survive here
+ */
+ for (i = 0; i < G_N_ELEMENTS(proxy_privileged_shm_args); i++) {
+ while (rspamd_http_message_remove_header(msg,
+ proxy_privileged_shm_args[i])) {
+ /* Keep removing */
+ }
+ }
+
+ proxy_strip_query_args(msg, proxy_privileged_shm_args,
+ G_N_ELEMENTS(proxy_privileged_shm_args));
+
+ for (i = 0; i < G_N_ELEMENTS(proxy_privileged_file_args); i++) {
+ if (rspamd_http_message_find_header(msg,
+ proxy_privileged_file_args[i])) {
+ has_file_arg = TRUE;
+ break;
+ }
+ }
+
+ if (!has_file_arg && msg->url != NULL && msg->url->len > 0 &&
+ http_parser_parse_url(RSPAMD_FSTRING_DATA(msg->url),
+ RSPAMD_FSTRING_LEN(msg->url), 0, &u) == 0 &&
+ (u.field_set & (1 << UF_QUERY))) {
+ query_args = rspamd_http_message_parse_query(msg);
+
+ for (i = 0; i < G_N_ELEMENTS(proxy_privileged_file_args); i++) {
+ if (proxy_has_query_arg(query_args, proxy_privileged_file_args[i])) {
+ has_file_arg = TRUE;
+ break;
+ }
+ }
+
+ g_hash_table_unref(query_args);
+ }
+
+ if (has_file_arg) {
+ msg_info_session("deny file message source from %s: this connection is "
+ "not permitted to use privileged inputs; set "
+ "`allow_file_and_shm_inputs = true` for this worker if "
+ "all of its clients are trusted",
+ rspamd_inet_address_to_string_pretty(session->client_addr));
+ *err_code = 400;
+ *err_status = "File and shm message sources are not allowed";
+
+ return FALSE;
+ }
+
+ return TRUE;
+}
+
static gboolean
proxy_check_file(struct rspamd_http_message *msg,
struct rspamd_proxy_session *session)
struct http_parser_url u;
rspamd_fstring_t *new_url;
+ if (!session->allow_file_shm) {
+ /*
+ * Privileged inputs have already been refused by
+ * proxy_sanitize_privileged_inputs(), so there is nothing left to map
+ * here. Guard it once more so that no path can ever reach
+ * rspamd_file_xmap() without the capability.
+ */
+ return TRUE;
+ }
+
tok = rspamd_http_message_find_header(msg, "File");
if (tok) {
return TRUE;
}
+/*
+ * Shared memory forwarding (and the `File` handover that goes with it) hands
+ * the upstream a name that it will open on its own, so it may only be used
+ * when the upstream transport itself is protected:
+ *
+ * - a unix socket, which is guarded by the filesystem permissions, as long as
+ * the backend is declared `local` or privileged inputs are enabled;
+ * - a strict loopback peer, as long as privileged inputs are enabled.
+ *
+ * Note that `local` alone is deliberately not sufficient for a TCP upstream any
+ * more: it used to let the proxy hand a shared memory name (or a file name of
+ * the client's choosing) to an arbitrary remote host, and
+ * rspamd_inet_address_is_local() was not a loopback test at all as it also
+ * accepted link local and site local addresses.
+ *
+ * The very same rule applies to encrypted and to unencrypted upstreams: an
+ * encrypted message never carries a shared body anyway (the HTTP layer detaches
+ * it), so this only ever removes the `File` header there.
+ */
+static gboolean
+rspamd_proxy_upstream_allows_shm(struct rspamd_proxy_session *session,
+ gboolean backend_local,
+ const rspamd_inet_addr_t *addr)
+{
+ if (addr == NULL) {
+ return FALSE;
+ }
+
+ if (rspamd_inet_address_get_af(addr) == AF_UNIX) {
+ return backend_local || session->ctx->allow_file_and_shm_inputs;
+ }
+
+ return rspamd_worker_addr_is_loopback(addr) &&
+ session->ctx->allow_file_and_shm_inputs;
+}
+
+/*
+ * An inline body is read into memory by the upstream, so it must obey the
+ * configured message size limit. A body that came from the request itself is
+ * already bounded by rspamd_http_connection_set_max_size(), a mapped file is
+ * not bounded by anything, hence this check.
+ */
+static gboolean
+proxy_inline_body_fits(struct rspamd_proxy_session *session, gsize len)
+{
+ gsize max_message = session->ctx->cfg ? session->ctx->cfg->max_message : 0;
+
+ if (max_message > 0 && len > max_message) {
+ return FALSE;
+ }
+
+ return TRUE;
+}
+
static void
proxy_backend_mirror_error_handler(struct rspamd_http_connection *conn, GError *err)
{
/* We found a keepalive connection, use it */
struct rspamd_http_connection *conn;
+ /*
+ * If this upstream may not receive a shared body then the
+ * mapped file has to be forwarded inline, which is only
+ * possible within the configured message size limit.
+ * Checked before anything is checked out of the keepalive
+ * pool, so that skipping the mirror is free.
+ */
+ if (session->fname &&
+ !rspamd_proxy_upstream_allows_shm(session, m->local,
+ keepalive_addr) &&
+ !proxy_inline_body_fits(session, session->map_len)) {
+ msg_warn_session("skip mirror %s: cannot forward a %uz "
+ "bytes file inline, the limit is %uz bytes",
+ m->name, session->map_len,
+ session->ctx->cfg->max_message);
+
+ continue;
+ }
+
conn = rspamd_http_context_check_keepalive(
session->ctx->http_ctx,
(rspamd_inet_addr_t *) keepalive_addr,
msg->peer_key = rspamd_pubkey_ref(m->key);
}
- if (m->local || rspamd_inet_address_is_local(keepalive_addr)) {
+ if (rspamd_proxy_upstream_allows_shm(session, m->local,
+ keepalive_addr)) {
if (session->fname) {
rspamd_http_message_add_header(msg, "File", session->fname);
}
continue;
}
- bk_conn->backend_sock = rspamd_inet_address_connect(
- rspamd_upstream_addr_next(bk_conn->up),
- SOCK_STREAM, TRUE);
+ const rspamd_inet_addr_t *up_addr = rspamd_upstream_addr_next(bk_conn->up);
+ gboolean mirror_allows_shm = rspamd_proxy_upstream_allows_shm(session,
+ m->local,
+ up_addr);
+
+ /*
+ * A mapped file that cannot be handed over as a shared body has to be
+ * forwarded inline, which is only possible within the configured
+ * message size limit. Checked before connecting, so that skipping the
+ * mirror costs nothing.
+ */
+ if (session->fname && !mirror_allows_shm &&
+ !proxy_inline_body_fits(session, session->map_len)) {
+ msg_warn_session("skip mirror %s: cannot forward a %uz bytes file "
+ "inline, the limit is %uz bytes",
+ m->name, session->map_len,
+ session->ctx->cfg->max_message);
+ rspamd_upstream_release(bk_conn->up);
+
+ continue;
+ }
+
+ bk_conn->backend_sock = rspamd_inet_address_connect(up_addr,
+ SOCK_STREAM, TRUE);
if (bk_conn->backend_sock == -1) {
msg_err_session("cannot connect upstream for %s", m->name);
msg->peer_key = rspamd_pubkey_ref(m->key);
}
- if (m->local ||
- rspamd_inet_address_is_local(rspamd_upstream_addr_cur(bk_conn->up))) {
+ if (mirror_allows_shm) {
if (session->fname) {
rspamd_http_message_add_header(msg, "File", session->fname);
task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_BODY_BLOCK;
}
+ /*
+ * Both properties come from the transport that has been accepted, never
+ * from anything that the client sends (a User-Agent, a forwarded header or
+ * a query argument are all trivially spoofable)
+ */
+ if (!session->allow_file_shm) {
+ task->protocol_flags &= ~RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT;
+ }
+
+ if (session->local_client) {
+ task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT;
+ }
+ else {
+ task->protocol_flags &= ~RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT;
+ }
+
task->sock = -1;
if (session->client_milter_conn) {
const rspamd_ftok_t *host;
GError *err = NULL;
char hostbuf[512];
+ int err_code = 404;
+ const char *err_status = "Backend not found";
host = rspamd_http_message_find_header(session->client_message, "Host");
goto err;
}
+ const rspamd_inet_addr_t *up_addr =
+ rspamd_upstream_addr_next(session->master_conn->up);
+ gboolean master_allows_shm = rspamd_proxy_upstream_allows_shm(
+ session, backend->local, up_addr);
+
+ /*
+ * When this upstream may not receive a shared body the mapped file has
+ * to be forwarded inline, and an inline body must obey the configured
+ * message size limit. Fail cleanly instead of truncating, and do it
+ * before connecting so that nothing has to be unwound.
+ */
+ if (session->fname && !master_allows_shm &&
+ !proxy_inline_body_fits(session, session->map_len)) {
+ msg_err_session("cannot forward a %uz bytes file inline to %s, the "
+ "limit is %uz bytes",
+ session->map_len, host ? hostbuf : "default",
+ session->ctx->cfg->max_message);
+ rspamd_upstream_release(session->master_conn->up);
+ err_code = 413;
+ err_status = "Message too large to forward";
+
+ goto err;
+ }
+
session->master_conn->backend_sock = rspamd_inet_address_connect(
- rspamd_upstream_addr_next(session->master_conn->up),
- SOCK_STREAM, TRUE);
+ up_addr, SOCK_STREAM, TRUE);
if (session->master_conn->backend_sock == -1) {
msg_err_session("cannot connect upstream: %s(%s)",
/* Add/overwrite IP header with the actual client IP */
proxy_add_client_ip_header(msg, session);
- if (backend->local ||
- rspamd_inet_address_is_local(
- rspamd_upstream_addr_cur(
- session->master_conn->up))) {
+ if (master_allows_shm) {
if (session->fname) {
rspamd_http_message_add_header(msg, "File", session->fname);
else {
rspamd_http_connection_steal_msg(session->client_conn);
rspamd_http_connection_reset(session->client_conn);
- proxy_client_write_error(session, 404, "Backend not found");
+ proxy_client_write_error(session, err_code, err_status);
}
return FALSE;
struct rspamd_http_message *msg)
{
struct rspamd_proxy_session *session = conn->ud;
+ int err_code = 404;
+ const char *err_status = "Backend not found";
if (!session->master_conn) {
session->master_conn = rspamd_mempool_alloc0(session->pool,
"/" MSG_CMD_CHECK_V2, strlen("/" MSG_CMD_CHECK_V2));
}
+ /*
+ * Ingress sanitising: this happens before proxy_check_file() may open
+ * anything and before the message is forwarded or self scanned, so a
+ * client supplied File/Path/Shm* can neither be acted upon nor be
+ * passed through to an upstream that trusts this proxy
+ */
+ if (!proxy_sanitize_privileged_inputs(msg, session, &err_code,
+ &err_status)) {
+ goto err;
+ }
+
if (!proxy_check_file(msg, session)) {
goto err;
}
rspamd_http_message_remove_header(msg, "Keep-Alive");
rspamd_http_message_remove_header(msg, "Connection");
rspamd_http_connection_reset(session->client_conn);
- proxy_client_write_error(session, 404, "Backend not found");
+ proxy_client_write_error(session, err_code, err_status);
return 0;
}
* decrement for nconns. proxy_session_dtor skips nconns-- for milter
* because proxy_session_refresh creates a new session per message and
* each intermediate session destruction must not touch the counter. */
- session->worker->nconns--;
+ proxy_conn_release(session);
REF_RELEASE(session);
}
else {
err);
/* Terminate session immediately */
proxy_backend_close_connection(session->master_conn);
- session->worker->nconns--;
+ proxy_conn_release(session);
REF_RELEASE(session);
}
else {
err);
/* Terminate session immediately */
proxy_backend_close_connection(session->master_conn);
- session->worker->nconns--;
+ proxy_conn_release(session);
REF_RELEASE(session);
}
}
struct rspamd_proxy_ctx *ctx;
rspamd_inet_addr_t *addr = NULL;
struct rspamd_proxy_session *session;
+ char *source_key;
int nfd;
ctx = worker->ctx;
return;
}
+ /*
+ * Admission control happens here, before a pool, a session, any keypair
+ * work, a file mapping or a shared body allocation is made for this
+ * connection. The connection is accepted and closed right away rather than
+ * left in the backlog, so that a level triggered listener does not spin.
+ */
+ if (ctx->max_connections > 0 && worker->nconns >= ctx->max_connections) {
+ msg_info("drop connection from %s: the limit of %ud simultaneous "
+ "connections has been reached",
+ rspamd_inet_address_to_string(addr), ctx->max_connections);
+ rspamd_inet_address_free(addr);
+ close(nfd);
+
+ return;
+ }
+
+ source_key = proxy_source_key_new(ctx, addr);
+
+ if (source_key != NULL && !proxy_source_conn_reserve(ctx, source_key)) {
+ msg_info("drop connection from %s: the limit of %ud simultaneous "
+ "connections per source has been reached",
+ source_key, ctx->max_connections_per_source);
+ g_free(source_key);
+ rspamd_inet_address_free(addr);
+ close(nfd);
+
+ return;
+ }
+
worker->nconns++;
session = g_malloc0(sizeof(*session));
REF_INIT_RETAIN(session, proxy_session_dtor);
session->client_sock = nfd;
session->client_addr = addr;
+ session->source_key = source_key;
session->mirror_conns = g_ptr_array_sized_new(ctx->mirrors->len);
session->pool = rspamd_mempool_new_short_lived("proxy");
session->ctx = ctx;
session->worker = worker;
+ /*
+ * Privileged message sources are decided from the accepted transport only:
+ * a unix socket is always trusted, everything else needs the option to be
+ * enabled explicitly
+ */
+ session->allow_file_shm = (rspamd_inet_address_get_af(addr) == AF_UNIX) ||
+ ctx->allow_file_and_shm_inputs;
+ session->local_client = (rspamd_inet_address_get_af(addr) == AF_UNIX) ||
+ rspamd_worker_addr_is_loopback(addr);
if (ctx->sessions_cache) {
rspamd_worker_session_cache_add(ctx->sessions_cache,
if (!ctx->milter) {
int http_opts = 0;
- if (ctx->encrypted_only && !rspamd_inet_address_is_local(addr)) {
+ /*
+ * Only a genuinely local peer is exempt from `encrypted_only`.
+ * rspamd_inet_address_is_local() used to be accepted here, but it also
+ * matches IPv6 link local and site local addresses, which are other
+ * hosts on the network and hence must not skip the encryption
+ * requirement.
+ */
+ if (ctx->encrypted_only && !session->local_client) {
http_opts |= RSPAMD_HTTP_REQUIRE_ENCRYPTION;
}
session->client_conn = rspamd_http_connection_new_server(
ctx->srv = worker->srv;
ctx->event_loop = rspamd_prepare_worker(worker, "rspamd_proxy",
proxy_accept_socket);
+ rspamd_worker_warn_file_shm_inputs(worker, "rspamd_proxy",
+ ctx->allow_file_and_shm_inputs);
+
+ if (ctx->max_connections_per_source > 0 && ctx->conns_per_source == NULL) {
+ ctx->conns_per_source = g_hash_table_new_full(g_str_hash, g_str_equal,
+ g_free, NULL);
+ rspamd_mempool_add_destructor(ctx->cfg->cfg_pool,
+ (rspamd_mempool_destruct_t) g_hash_table_unref,
+ ctx->conns_per_source);
+ }
ctx->resolver = rspamd_dns_resolver_init(worker->srv->logger,
ctx->event_loop,
struct rspamd_worker_ctx *ctx;
struct rspamd_http_connection *http_conn;
struct rspamd_worker *worker;
+ /*
+ * Whether this connection may use privileged File/Path/Shm message
+ * source inputs. Decided at accept time from the transport only.
+ */
+ gboolean allow_file_shm;
+ /* Whether the peer is strictly local (unix socket or loopback) */
+ gboolean is_local;
+ /* Whether this session is currently accounted in worker->nconns */
+ gboolean counted;
};
/*
* Reduce number of tasks proceeded
*/
static void
-reduce_tasks_count(gpointer arg)
+reduce_tasks_count(struct rspamd_worker *worker)
{
- struct rspamd_worker *worker = arg;
-
worker->nconns--;
if (worker->state == rspamd_worker_wait_connections && worker->nconns == 0) {
}
}
+/*
+ * Idempotent release of the connection slot acquired in `accept_socket`.
+ * It is safe to call it multiple times for the same session, so every
+ * teardown path can call it unconditionally.
+ */
+static void
+rspamd_worker_session_uncount(struct rspamd_worker_session *session)
+{
+ if (session->counted) {
+ session->counted = FALSE;
+ reduce_tasks_count(session->worker);
+ }
+}
+
+/*
+ * Session destructor installed on the task pool once the task takes the
+ * session ownership. It both releases the connection slot and frees the
+ * session itself, so the ordering of the pool destructors is irrelevant.
+ */
+static void
+rspamd_worker_session_task_dtor(gpointer arg)
+{
+ struct rspamd_worker_session *session = arg;
+
+ rspamd_worker_session_uncount(session);
+ g_free(session);
+}
+
static int
rspamd_worker_body_handler(struct rspamd_http_connection *conn,
struct rspamd_http_message *msg,
task->resolver = ctx->resolver;
- session->worker->nconns++;
- rspamd_mempool_add_destructor(task->task_pool,
- (rspamd_mempool_destruct_t) reduce_tasks_count,
- session->worker);
-
- /* Session memory is also now handled by task */
+ /*
+ * Both the connection accounting and the session memory are now handled
+ * by the task pool.
+ */
rspamd_mempool_add_destructor(task->task_pool,
- (rspamd_mempool_destruct_t) g_free,
+ (rspamd_mempool_destruct_t) rspamd_worker_session_task_dtor,
session);
+ /*
+ * Privileged message source inputs and the local client status are
+ * decided from the accepted transport only: never from the request
+ * headers or any other client controlled data.
+ */
+ if (!session->allow_file_shm) {
+ task->protocol_flags &= ~RSPAMD_TASK_PROTOCOL_FLAG_ALLOW_FILE_SHM_INPUT;
+ }
+
+ if (session->is_local) {
+ task->protocol_flags |= RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT;
+ }
+ else {
+ task->protocol_flags &= ~RSPAMD_TASK_PROTOCOL_FLAG_LOCAL_CLIENT;
+ }
+
/* Set up async session */
task->s = rspamd_task_create_session(task, task->task_pool, rspamd_task_fin,
NULL, (event_finalizer_t) rspamd_task_free);
rspamd_http_connection_unref(session->http_conn);
rspamd_inet_address_free(session->addr);
close(session->fd);
+ rspamd_worker_session_uncount(session);
g_free(session);
}
}
rspamd_http_connection_reset(session->http_conn);
rspamd_http_connection_unref(session->http_conn);
close(session->fd);
+ rspamd_worker_session_uncount(session);
g_free(session);
}
struct rspamd_worker_ctx *ctx;
struct rspamd_worker_session *session;
rspamd_inet_addr_t *addr = NULL;
+ gboolean over_limit, is_unix;
int nfd, http_opts = 0;
ctx = worker->ctx;
- if (ctx->max_tasks != 0 && worker->nconns > ctx->max_tasks) {
- msg_info_ctx("current tasks is now: %uD while maximum is: %uD",
- worker->nconns,
- ctx->max_tasks);
- return;
- }
+ /*
+ * Admission control must be evaluated before we accept anything and it
+ * must not depend on the protocol/body timeouts: a connection is counted
+ * from the accept and until the task pool (or the unmanaged session
+ * teardown) releases it.
+ */
+ over_limit = (ctx->max_tasks != 0 && worker->nconns >= ctx->max_tasks);
if ((nfd =
rspamd_accept_from_socket(w->fd, &addr,
return;
}
+ if (over_limit) {
+ /*
+ * We still have to accept and close the connection here: the listen
+ * event is level triggered, so merely returning would spin the worker
+ * on the listening socket.
+ */
+ ev_tstamp now = ev_now(EV_A);
+
+ if (now - ctx->last_overload_log >= 1.0) {
+ ctx->last_overload_log = now;
+ msg_info_ctx("dropping connection from %s: current connections "
+ "count is %ud while maximum is %uD",
+ rspamd_inet_address_to_string_pretty(addr),
+ worker->nconns,
+ ctx->max_tasks);
+ }
+
+ rspamd_inet_address_free(addr);
+ close(nfd);
+
+ return;
+ }
+
+ /* From this point the connection is accounted, see reduce_tasks_count */
+ worker->nconns++;
+
session = g_malloc0(sizeof(*session));
session->magic = G_MAXINT64;
session->addr = addr;
session->fd = nfd;
session->ctx = ctx;
session->worker = worker;
+ session->counted = TRUE;
- if (ctx->encrypted_only && !rspamd_inet_address_is_local(addr)) {
+ /*
+ * Privileged inputs are allowed on unix sockets unconditionally (they are
+ * protected by the filesystem permissions) and elsewhere only when the
+ * administrator has explicitly opted in. Never trust anything that comes
+ * from the client for this decision.
+ */
+ is_unix = (rspamd_inet_address_get_af(addr) == AF_UNIX);
+ session->allow_file_shm = is_unix || ctx->allow_file_and_shm_inputs;
+ session->is_local = is_unix || rspamd_worker_addr_is_loopback(addr);
+
+ /*
+ * Only a genuinely local peer is exempt from the encryption requirement.
+ * rspamd_inet_address_is_local() also matches IPv6 link-local and site-local
+ * addresses, which would let any host on the same segment skip encryption.
+ */
+ if (ctx->encrypted_only && !session->is_local) {
http_opts = RSPAMD_HTTP_REQUIRE_ENCRYPTION;
}
ctx->timeout = DEFAULT_WORKER_IO_TIMEOUT;
ctx->cfg = cfg;
ctx->task_timeout = NAN;
+ /*
+ * Permissive by default for this release; the default flips to FALSE in
+ * the next major release.
+ */
+ ctx->allow_file_and_shm_inputs = TRUE;
rspamd_rcl_register_worker_option(cfg,
type,
0,
"Allow only encrypted connections");
+ rspamd_rcl_register_worker_option(cfg,
+ type,
+ "allow_file_and_shm_inputs",
+ rspamd_rcl_parse_struct_boolean,
+ ctx,
+ G_STRUCT_OFFSET(struct rspamd_worker_ctx, allow_file_and_shm_inputs),
+ 0,
+ "Accept privileged File/Path/Shm message source inputs over TCP "
+ "(default: true; will become false in the next major release). "
+ "Unix sockets always allow these inputs");
+
rspamd_rcl_register_worker_option(cfg,
type,
ctx->cfg = worker->srv->cfg;
CFG_REF_RETAIN(ctx->cfg);
ctx->event_loop = rspamd_prepare_worker(worker, "normal", accept_socket);
+ rspamd_worker_warn_file_shm_inputs(worker, "normal",
+ ctx->allow_file_and_shm_inputs);
rspamd_symcache_start_refresh(worker->srv->cfg->cache, ctx->event_loop,
worker);
gboolean is_mime;
/* Allow encrypted requests only using network */
gboolean encrypted_only;
+ /* Allow privileged File/Path/Shm inputs over non unix sockets */
+ gboolean allow_file_and_shm_inputs;
/* Limit of tasks */
uint32_t max_tasks;
+ /* Rate limiting for the overload log messages */
+ ev_tstamp last_overload_log;
/* Maximum time for task processing */
ev_tstamp task_timeout;
/* Encryption key */
projects / thirdparty / rspamd.git / commitdiff
