NanoMQ 0.16.5 is vulnerable to heap-use-after-free in the nano_ctx_send function of nmq_mqtt.c.
**Describe the bug**
We found a heap-use-after-free  in `nano_ctx_send` function of `nmq_mqtt.c` when it processes malformed messages.

**Actual Behavior**
Heap-use-after-free

**To Reproduce**
Read the `pipe` in `nano_ctx_send`(nmq_mqtt.c)
<img width="666" alt="image" src="https://user-images.githubusercontent.com/131155516/232831609-cb006702-2f3e-4751-8dde-a91ef5087770.png">

Free the `pipe` in `reap_worker`(reap.c) -> `pipe_destroy`(pipe.c)
<img width="671" alt="image" src="https://user-images.githubusercontent.com/131155516/232833249-01f34876-46c1-4e1a-863e-85d6b3ea26a2.png">

`pipe_destroy`(pipe.c)
<img width="820" alt="image" src="https://user-images.githubusercontent.com/131155516/232833523-9f79a96f-3019-4576-b8d1-b073f1fcb5b3.png">


ASAN Log

> 
> =================================================================
> ==3144198==ERROR: AddressSanitizer: heap-use-after-free on address 0x61d00003ff98 at pc 0x0000006dce8c bp 0x7fe86eea87f0 sp 0x7fe86eea87e8
> READ of size 8 at 0x61d00003ff98 thread T11 (nng:task)
>     #0 0x6dce8b in nano_ctx_send /home/user/nanomq/nng/src/sp/protocol/mqtt/nmq_mqtt.c:405:9
>     #1 0x5a80b0 in nng_ctx_send /home/user/nanomq/nng/src/nng.c:401:2
>     #2 0x577770 in server_cb /home/user/nanomq/nanomq/apps/broker.c:696:3
>     #3 0x646595 in nni_taskq_thread /home/user/nanomq/nng/src/core/taskq.c:50:4
>     #4 0x64caf0 in nni_thr_wrap /home/user/nanomq/nng/src/core/thread.c:94:3
>     #5 0x66a073 in nni_plat_thr_main /home/user/nanomq/nng/src/platform/posix/posix_thread.c:266:2
>     #6 0x7fe876f21608 in start_thread /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
>     #7 0x7fe876cad132 in clone /build/glibc-SzIz7B/glibc-2.31/misc/../sysdeps/unix/sysv/linux/x86_64/clone.S:95
> 
> 0x61d00003ff98 is located 792 bytes inside of 2304-byte region [0x61d00003fc80,0x61d000040580)
> freed by thread T14 (nng:reap2) here:
>     #0 0x4acc8d in free (/home/user/nanomq/build/nanomq/nanomq+0x4acc8d)
>     #1 0x6172b6 in pipe_destroy /home/user/nanomq/nng/src/core/pipe.c:83:2
>     #2 0x618e7e in reap_worker /home/user/nanomq/nng/src/core/reap.c:58:5
>     #3 0x64caf0 in nni_thr_wrap /home/user/nanomq/nng/src/core/thread.c:94:3
>     #4 0x66a073 in nni_plat_thr_main /home/user/nanomq/nng/src/platform/posix/posix_thread.c:266:2
>     #5 0x7fe876f21608 in start_thread /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
> 
> previously allocated by thread T11 (nng:task) here:
>     #0 0x4ad082 in calloc (/home/user/nanomq/build/nanomq/nanomq+0x4ad082)
>     #1 0x6600a9 in nni_zalloc /home/user/nanomq/nng/src/platform/posix/posix_alloc.c:26:19
>     #2 0x61079e in pipe_create /home/user/nanomq/nng/src/core/pipe.c:252:11
>     #3 0x6133d8 in nni_pipe_create_listener /home/user/nanomq/nng/src/core/pipe.c:331:12
>     #4 0x6332c1 in nni_listener_add_pipe /home/user/nanomq/nng/src/core/socket.c:1578:6
>     #5 0x5f071c in listener_accept_cb /home/user/nanomq/nng/src/core/listener.c:357:3
>     #6 0x646595 in nni_taskq_thread /home/user/nanomq/nng/src/core/taskq.c:50:4
>     #7 0x64caf0 in nni_thr_wrap /home/user/nanomq/nng/src/core/thread.c:94:3
>     #8 0x66a073 in nni_plat_thr_main /home/user/nanomq/nng/src/platform/posix/posix_thread.c:266:2
>     #9 0x7fe876f21608 in start_thread /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
> 
> Thread T11 (nng:task) created by T0 here:
>     #0 0x497cba in pthread_create (/home/user/nanomq/build/nanomq/nanomq+0x497cba)
>     #1 0x669c07 in nni_plat_thr_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:279:7
>     #2 0x64bbbd in nni_thr_init /home/user/nanomq/nng/src/core/thread.c:121:12
>     #3 0x645507 in nni_taskq_init /home/user/nanomq/nng/src/core/taskq.c:95:8
>     #4 0x5e6479 in nni_init_helper /home/user/nanomq/nng/src/core/init.c:35:13
>     #5 0x66ad55 in nni_plat_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:422:12
>     #6 0x6fdfa8 in nni_proto_mqtt_open /home/user/nanomq/nng/src/sp/protocol.c:37:12
>     #7 0x592a92 in broker /home/user/nanomq/nanomq/apps/broker.c:872:25
>     #8 0x5a2d8d in broker_start /home/user/nanomq/nanomq/apps/broker.c:1603:7
>     #9 0x4e1842 in main /home/user/nanomq/nanomq/nanomq.c:142:10
>     #10 0x7fe876bb2082 in __libc_start_main /build/glibc-SzIz7B/glibc-2.31/csu/../csu/libc-start.c:308:16
> 
> Thread T14 (nng:reap2) created by T0 here:
>     #0 0x497cba in pthread_create (/home/user/nanomq/build/nanomq/nanomq+0x497cba)
>     #1 0x669c07 in nni_plat_thr_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:279:7
>     #2 0x64bbbd in nni_thr_init /home/user/nanomq/nng/src/core/thread.c:121:12
>     #3 0x6188e5 in nni_reap_sys_init /home/user/nanomq/nng/src/core/reap.c:110:12
>     #4 0x5e64b0 in nni_init_helper /home/user/nanomq/nng/src/core/init.c:36:13
>     #5 0x66ad55 in nni_plat_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:422:12
>     #6 0x6fdfa8 in nni_proto_mqtt_open /home/user/nanomq/nng/src/sp/protocol.c:37:12
>     #7 0x592a92 in broker /home/user/nanomq/nanomq/apps/broker.c:872:25
>     #8 0x5a2d8d in broker_start /home/user/nanomq/nanomq/apps/broker.c:1603:7
>     #9 0x4e1842 in main /home/user/nanomq/nanomq/nanomq.c:142:10
>     #10 0x7fe876bb2082 in __libc_start_main /build/glibc-SzIz7B/glibc-2.31/csu/../csu/libc-start.c:308:16
> 
> SUMMARY: AddressSanitizer: heap-use-after-free /home/user/nanomq/nng/src/sp/protocol/mqtt/nmq_mqtt.c:405:9 in nano_ctx_send
> Shadow bytes around the buggy address:
>   0x0c3a7fffffa0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffb0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffc0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffd0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffe0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
> =>0x0c3a7ffffff0: fd fd fd[fd]fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000000: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000010: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000020: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000030: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000040: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
> 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
>   Shadow gap:              cc
> ==3144198==ABORTING
> 

** Environment Details **
 - NanoMQ version: 0.16.5 (aa188a4d)
 - Operating system and version: Ubuntu 20.04
 - Compiler and language used: gcc 9.4.0 clang 10.0.0
 - testing scenario: Run the broker(build with ASAN and TSAN) with the ./nanomq start command

[Security]: data contention in pipe reaper
This is a GitHub Issue
repo:nanomq
owner:emqx
Title : [Security]: data contention in pipe reaper
Issue date:

--- start body ---
**Describe the bug**
We found a heap-use-after-free  in `nano_ctx_send` function of `nmq_mqtt.c` when it processes malformed messages.

**Actual Behavior**
Heap-use-after-free

**To Reproduce**
Read the `pipe` in `nano_ctx_send`(nmq_mqtt.c)
<img width="666" alt="image" src="https://user-images.githubusercontent.com/131155516/232831609-cb006702-2f3e-4751-8dde-a91ef5087770.png">

Free the `pipe` in `reap_worker`(reap.c) -> `pipe_destroy`(pipe.c)
<img width="671" alt="image" src="https://user-images.githubusercontent.com/131155516/232833249-01f34876-46c1-4e1a-863e-85d6b3ea26a2.png">

`pipe_destroy`(pipe.c)
<img width="820" alt="image" src="https://user-images.githubusercontent.com/131155516/232833523-9f79a96f-3019-4576-b8d1-b073f1fcb5b3.png">


ASAN Log

> 
> =================================================================
> ==3144198==ERROR: AddressSanitizer: heap-use-after-free on address 0x61d00003ff98 at pc 0x0000006dce8c bp 0x7fe86eea87f0 sp 0x7fe86eea87e8
> READ of size 8 at 0x61d00003ff98 thread T11 (nng:task)
>     #0 0x6dce8b in nano_ctx_send /home/user/nanomq/nng/src/sp/protocol/mqtt/nmq_mqtt.c:405:9
>     #1 0x5a80b0 in nng_ctx_send /home/user/nanomq/nng/src/nng.c:401:2
>     #2 0x577770 in server_cb /home/user/nanomq/nanomq/apps/broker.c:696:3
>     #3 0x646595 in nni_taskq_thread /home/user/nanomq/nng/src/core/taskq.c:50:4
>     #4 0x64caf0 in nni_thr_wrap /home/user/nanomq/nng/src/core/thread.c:94:3
>     #5 0x66a073 in nni_plat_thr_main /home/user/nanomq/nng/src/platform/posix/posix_thread.c:266:2
>     #6 0x7fe876f21608 in start_thread /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
>     #7 0x7fe876cad132 in clone /build/glibc-SzIz7B/glibc-2.31/misc/../sysdeps/unix/sysv/linux/x86_64/clone.S:95
> 
> 0x61d00003ff98 is located 792 bytes inside of 2304-byte region [0x61d00003fc80,0x61d000040580)
> freed by thread T14 (nng:reap2) here:
>     #0 0x4acc8d in free (/home/user/nanomq/build/nanomq/nanomq+0x4acc8d)
>     #1 0x6172b6 in pipe_destroy /home/user/nanomq/nng/src/core/pipe.c:83:2
>     #2 0x618e7e in reap_worker /home/user/nanomq/nng/src/core/reap.c:58:5
>     #3 0x64caf0 in nni_thr_wrap /home/user/nanomq/nng/src/core/thread.c:94:3
>     #4 0x66a073 in nni_plat_thr_main /home/user/nanomq/nng/src/platform/posix/posix_thread.c:266:2
>     #5 0x7fe876f21608 in start_thread /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
> 
> previously allocated by thread T11 (nng:task) here:
>     #0 0x4ad082 in calloc (/home/user/nanomq/build/nanomq/nanomq+0x4ad082)
>     #1 0x6600a9 in nni_zalloc /home/user/nanomq/nng/src/platform/posix/posix_alloc.c:26:19
>     #2 0x61079e in pipe_create /home/user/nanomq/nng/src/core/pipe.c:252:11
>     #3 0x6133d8 in nni_pipe_create_listener /home/user/nanomq/nng/src/core/pipe.c:331:12
>     #4 0x6332c1 in nni_listener_add_pipe /home/user/nanomq/nng/src/core/socket.c:1578:6
>     #5 0x5f071c in listener_accept_cb /home/user/nanomq/nng/src/core/listener.c:357:3
>     #6 0x646595 in nni_taskq_thread /home/user/nanomq/nng/src/core/taskq.c:50:4
>     #7 0x64caf0 in nni_thr_wrap /home/user/nanomq/nng/src/core/thread.c:94:3
>     #8 0x66a073 in nni_plat_thr_main /home/user/nanomq/nng/src/platform/posix/posix_thread.c:266:2
>     #9 0x7fe876f21608 in start_thread /build/glibc-SzIz7B/glibc-2.31/nptl/pthread_create.c:477:8
> 
> Thread T11 (nng:task) created by T0 here:
>     #0 0x497cba in pthread_create (/home/user/nanomq/build/nanomq/nanomq+0x497cba)
>     #1 0x669c07 in nni_plat_thr_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:279:7
>     #2 0x64bbbd in nni_thr_init /home/user/nanomq/nng/src/core/thread.c:121:12
>     #3 0x645507 in nni_taskq_init /home/user/nanomq/nng/src/core/taskq.c:95:8
>     #4 0x5e6479 in nni_init_helper /home/user/nanomq/nng/src/core/init.c:35:13
>     #5 0x66ad55 in nni_plat_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:422:12
>     #6 0x6fdfa8 in nni_proto_mqtt_open /home/user/nanomq/nng/src/sp/protocol.c:37:12
>     #7 0x592a92 in broker /home/user/nanomq/nanomq/apps/broker.c:872:25
>     #8 0x5a2d8d in broker_start /home/user/nanomq/nanomq/apps/broker.c:1603:7
>     #9 0x4e1842 in main /home/user/nanomq/nanomq/nanomq.c:142:10
>     #10 0x7fe876bb2082 in __libc_start_main /build/glibc-SzIz7B/glibc-2.31/csu/../csu/libc-start.c:308:16
> 
> Thread T14 (nng:reap2) created by T0 here:
>     #0 0x497cba in pthread_create (/home/user/nanomq/build/nanomq/nanomq+0x497cba)
>     #1 0x669c07 in nni_plat_thr_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:279:7
>     #2 0x64bbbd in nni_thr_init /home/user/nanomq/nng/src/core/thread.c:121:12
>     #3 0x6188e5 in nni_reap_sys_init /home/user/nanomq/nng/src/core/reap.c:110:12
>     #4 0x5e64b0 in nni_init_helper /home/user/nanomq/nng/src/core/init.c:36:13
>     #5 0x66ad55 in nni_plat_init /home/user/nanomq/nng/src/platform/posix/posix_thread.c:422:12
>     #6 0x6fdfa8 in nni_proto_mqtt_open /home/user/nanomq/nng/src/sp/protocol.c:37:12
>     #7 0x592a92 in broker /home/user/nanomq/nanomq/apps/broker.c:872:25
>     #8 0x5a2d8d in broker_start /home/user/nanomq/nanomq/apps/broker.c:1603:7
>     #9 0x4e1842 in main /home/user/nanomq/nanomq/nanomq.c:142:10
>     #10 0x7fe876bb2082 in __libc_start_main /build/glibc-SzIz7B/glibc-2.31/csu/../csu/libc-start.c:308:16
> 
> SUMMARY: AddressSanitizer: heap-use-after-free /home/user/nanomq/nng/src/sp/protocol/mqtt/nmq_mqtt.c:405:9 in nano_ctx_send
> Shadow bytes around the buggy address:
>   0x0c3a7fffffa0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffb0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffc0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffd0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a7fffffe0: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
> =>0x0c3a7ffffff0: fd fd fd[fd]fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000000: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000010: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000020: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000030: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
>   0x0c3a80000040: fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd fd
> 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
>   Shadow gap:              cc
> ==3144198==ABORTING
> 

** Environment Details **
 - NanoMQ version: 0.16.5 (aa188a4d)
 - Operating system and version: Ubuntu 20.04
 - Compiler and language used: gcc 9.4.0 clang 10.0.0
 - testing scenario: Run the broker(build with ASAN and TSAN) with the ./nanomq start command

--- end body ---
