fix(kafka): fix use-after-free on task cancellation during kafka producer Send by disaykin · Pull Request #1299 · userver-framework/userver

disaykin · 2026-07-17T08:29:10Z

Fixes a Use-After-Free vulnerability in the asynchronous Kafka Producer API when a task is cancelled before librdkafka flushes its queue.

The kafka::Producer::Send API accepts parameters as views (string_view, zstring_view, HeaderViews). These views point to data owned by the caller. Inside Send, raw pointers from these views are passed down to librdkafka, which stores them in its internal asynchronous queues for background delivery.

If a client connection drops, a cascade cancellation of tasks occurs. This leads to a premature return from kafka::Producer::Send, causing the caller to destroy the original payload memory. However, librdkafka still retains these pointers and attempts to access them during subsequent network flushes to the broker, resulting in memory corruption.

To fix this, we ensure that Producer::Send is blocked from returning prematurely and cannot exit until the background SendImpl completes its interaction with librdkafka.

Note: by creating a PR or an issue you automatically agree to the CLA. See CONTRIBUTING.md. Feel free to remove this note, the agreement holds.

disaykin · 2026-07-20T04:50:14Z

@apolukhin прошу ревью. проблему с установкой кафки в тестах я починил (там протух урл), но вот что делать с флапающим тестом на постгрю - не знаю:

[ RUN      ] PoolTests/PostgrePool.PoolServerUnavailable/Sync
userver/postgresql/src/storages/postgres/tests/pool_pgtest.cpp:313: Failure
Expected equality of these values:
  0
  stats.connection.active
    Which is: 1

[  FAILED  ] PoolTests/PostgrePool.PoolServerUnavailable/Sync, where GetParam() = 4-byte object <00-00 00-00> (2013 ms)

Но это точно не относится к моему фиксу.

Также проблема компиляции с rabbitmq на debian12 привнесена не мной/

ArkadyRudenko · 2026-07-21T21:07:40Z

Tests in postgres is flap.

ArkadyRudenko · 2026-07-21T21:08:31Z

 set -o errexit -o nounset -o pipefail -o posix -x

-KAFKA_VERSION=4.0.1
+KAFKA_VERSION=4.3.1


Why did you increas it?

Старая ссылка для установки кафки больше не работает. Заменил на ссылку для скачивания с их официального сайта

Предлагаете оформить это отдельным запросом на влитие? Или претензия именно к изменению версии? Старая версия у меня вроде бы не захотела скачиваться по новой ссылке...

Если у них на сайте теперь написано, что надо так, то окей

тогда предлагаю считать, что все сделано правильно

ArkadyRudenko · 2026-07-22T11:27:21Z

 set -o errexit -o nounset -o pipefail -o posix -x

-KAFKA_VERSION=4.0.1
+KAFKA_VERSION=4.3.1


Если у них на сайте теперь написано, что надо так, то окей

ArkadyRudenko · 2026-07-22T11:28:06Z

 ) const {
+    engine::TaskCancellationBlocker blocker;
+
+    if (engine::current_task::IsCancelRequested())


Добавь тест, в котором таска отменяется и происходит use-after-free. Чтобы хотя бы локально убедиться, что бага есть, а после фикса её нет

Это будет флаки тест. Даже при наличии бага, он будет фейлиться не каждый раз. И фэйлиться будет только под адрес-санитайзером. Я пробовал написать, получается как-то не очень... Попробую откопать свои наработки, может очередная попытка родит что-то удачнее

О, придумал. Можно выставить большие значения либрдкафка накопления очереди перед сбросом (1000 сообщений, 1 секунда, что раньше наступит). Тогда тест будет не флаки под санитайзером

Вообще,я имел ввиду, чтобы ты показал нам тест и локально убедился, что он падает без твоего фикса. А с фиксом чинится (и потом его удалить и не мержить). Но если получится даже полноцнный сделать, то шикарно

написал тесты, если закомментировать блокер, то падает

SUMMARY: AddressSanitizer: heap-use-after-free (/home/disaikin/git/userver/build/kafka/userver-kafka-dbtest+0x48f00c5) (BuildId: df950da811b1ab7c) in memcpy Shadow bytes around the buggy address: 0x508000004b00: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fa 0x508000004b80: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fd 0x508000004c00: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fa 0x508000004c80: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fa 0x508000004d00: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fd =>0x508000004d80: fa fa fa fa fd fd[fd]fd fd fd fd fd fd fd fd fd 0x508000004e00: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fd 0x508000004e80: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fa 0x508000004f00: fa fa fa fa fd fd fd fd fd fd fd fd fd fd fd fa 0x508000004f80: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa 0x508000005000: fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa fa Shadow byte legend (one shadow byte represents 8 application bytes): Addressable: 00 Partially addressable: 01 02 03 04 05 06 07 Heap left redzone: fa Freed heap region: fd Stack left redzone: f1 Stack mid redzone: f2 Stack right redzone: f3 Stack after return: f5 Stack use after scope: f8 Global redzone: f9 Global init order: f6 Poisoned by user: f7 Container overflow: fc Array cookie: ac Intra object redzone: bb ASan internal: fe Left alloca redzone: ca Right alloca redzone: cb ==2408226==ABORTING

apolukhin · 2026-07-26T16:25:28Z

    std::optional<std::uint32_t> partition,
    HeaderViews headers
 ) const {
+    engine::TaskCancellationBlocker blocker;


Кажется что нужен просто CriticalAsync вместо TaskCancellationBlocker... Или надо перенести TaskCancellationBlocker внутрь Async

Могу ошибаться, но...

Если засунуть блокер внутрь асинка, то вложенная корутина продолжит выполняться, а внешняя вылетит по исключению из Get(), стек развернется, и память, на которую указывали вьюхи, будет освобождена. Librdkafka попытается отправить в сеть уже освобожденную память.

Если заменить Async на CriticalAsync, то вложенная корутина не отменится, а внешняя сразу же отменится, освободит память, и librdkafka снова попытается отправить в сеть уже освобожденную память.

Тут именно нужно не дать внешней корутине разрушиться, пока librdkafka не закончит трогать память вьюх.

выкинул лишний код с проверкой отмены задачи на входе в блок, так как вероятность отмены задачи очень низка, а код корутин должен обнаружить отмену задачи при запуске в другом таск-процессоре, если я правильно понимаю его логику.

@apolukhin Я же правильно понимаю, что utils::Async при входе проверит ShouldCancel(), получит false из-за блокера, создаст корутину и отдаст её в другой таск-процессор, который сразу же увидит, что ShouldCancel() равен true и завершит корутину с выбросом исключения?

был не прав, все совсем не так работает, как я думал. переделал на использование CancellationPoint(), чтобы отмененная задача отменялась на входе в Send() и перед самым выходом из Send(), так как эти места фактически являются местами, где можно безопасно отменить долгую операцию

…ucer Send Fixes a Use-After-Free vulnerability in the asynchronous Kafka Producer API when a task is cancelled before librdkafka flushes its queue. The `kafka::Producer::Send` API accepts parameters as views (`string_view`, `zstring_view`, `HeaderViews`). These views point to data owned by the caller. Inside `Send`, raw pointers from these views are passed down to `librdkafka`, which stores them in its internal asynchronous queues for background delivery. If a client connection drops, a cascade cancellation of tasks occurs. This leads to a premature return from `kafka::Producer::Send`, causing the caller to destroy the original payload memory. However, `librdkafka` still retains these pointers and attempts to access them during subsequent network flushes to the broker, resulting in memory corruption. To fix this, we ensure that `Producer::Send` is blocked from returning prematurely and cannot exit until the background `SendImpl` completes its interaction with `librdkafka`. Co-authored-by: Aleksander Rypalov <rypalov2002@gmail.com>

Old URL returns 404 now

disaykin force-pushed the kafka-producer-cancellation-fix branch from 30e88de to 82e223b Compare July 17, 2026 09:01

ArkadyRudenko reviewed Jul 21, 2026

View reviewed changes

ArkadyRudenko reviewed Jul 22, 2026

View reviewed changes

apolukhin reviewed Jul 26, 2026

View reviewed changes

disaykin force-pushed the kafka-producer-cancellation-fix branch from 3ea2658 to 7d693f1 Compare July 27, 2026 06:42

disaykin and others added 2 commits July 27, 2026 14:44

ubuntu_install_kafka.sh: update kafka download URL

003eae8

Old URL returns 404 now

disaykin force-pushed the kafka-producer-cancellation-fix branch from 7d693f1 to 003eae8 Compare July 27, 2026 11:45

disaykin requested a review from ArkadyRudenko July 27, 2026 11:48

Uh oh!

Conversation

disaykin commented Jul 17, 2026

Uh oh!

disaykin commented Jul 20, 2026

Uh oh!

ArkadyRudenko commented Jul 21, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ArkadyRudenko Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ArkadyRudenko Jul 22, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ArkadyRudenko Jul 22, 2026 •

edited

Loading

ArkadyRudenko Jul 22, 2026 •

edited

Loading