fix(port): cap queue item size at the slot data bound

src/nxt_port_queue.h

@@ -53,6 +53,16 @@ nxt_port_queue_send(nxt_port_queue_t volatile *q, const void *p, uint8_t size,
     nxt_nncq_atomic_t      i;
     nxt_port_queue_item_t  *qi;
 
+    /*
+     * qi->data is NXT_PORT_QUEUE_MSG_SIZE bytes; refuse to enqueue an
+     * over-sized item rather than overwriting the adjacent queue slot.
+     * Callers must respect the queue's per-item budget.
+     */
+    if (nxt_slow_path(size > NXT_PORT_QUEUE_MSG_SIZE)) {
+        *notify = 0;
+        return NXT_ERROR;
+    }
+
     i = nxt_nncq_dequeue(&q->free_items);
     if (i == nxt_nncq_empty(&q->free_items)) {
         *notify = 0;
@@ -77,7 +87,7 @@ nxt_port_queue_send(nxt_port_queue_t volatile *q, const void *p, uint8_t size,
 nxt_inline ssize_t
 nxt_port_queue_recv(nxt_port_queue_t volatile *q, void *p)
 {
-    ssize_t                res;
+    size_t                 size;
     nxt_nncq_atomic_t      i;
     nxt_port_queue_item_t  *qi;
 
@@ -88,14 +98,27 @@ nxt_port_queue_recv(nxt_port_queue_t volatile *q, void *p)
 
     qi = (nxt_port_queue_item_t *) &q->items[i];
 
-    res = qi->size;
-    nxt_memcpy(p, qi->data, qi->size);
+    /*
+     * qi lives in shared memory that the peer can write.  Cap qi->size at
+     * the slot's data bound (NXT_PORT_QUEUE_MSG_SIZE) before the memcpy so
+     * a poisoned item cannot overflow the caller's receive buffer, which
+     * is sized for NXT_PORT_QUEUE_MSG_SIZE.  Read through a volatile cast
+     * so the compiler cannot fold the bound-check and the memcpy length
+     * into two separate reads of qi->size that a concurrent peer write
+     * could make disagree.
+     */
+    size = ((volatile nxt_port_queue_item_t *) qi)->size;
+    if (nxt_slow_path(size > NXT_PORT_QUEUE_MSG_SIZE)) {
+        size = NXT_PORT_QUEUE_MSG_SIZE;
+    }
+
+    nxt_memcpy(p, qi->data, size);
 
     nxt_nncq_enqueue(&q->free_items, i);
 
     nxt_atomic_fetch_add(&q->nitems, -1);
 
-    return res;
+    return size;
 }