Reject trailing NetIPC frame bytes

Author: ktsaou

.agents/sow/current/SOW-0015-20260605-codacy-scope-and-maintainability.md

@@ -738,6 +738,64 @@ Validation completed for this duplication-reduction increment:
   - `cppcheck_missingIncludeSystem` appears non-actionable in Codacy because it cannot see ordinary system headers in the analyzer environment.
   - `cppcheck_unusedStructMember` is noisy for NetIPC wire-layout structs and internal callback tables, but disabling it affects the whole Netdata repository unless a narrower suppression path is chosen.
 
+### 2026-06-07 - PR 22649 latest review-thread pass
+
+Netdata PR #22649 was rechecked after vendoring SDK commit `04f8b09` into Netdata commit `d60e6bc19a`.
+
+Facts from GitHub, SonarCloud, and CI:
+
+- GitHub review threads: ten total, eight resolved, two open.
+- SonarCloud PR issue API: zero unresolved issues.
+- SonarCloud PR hotspot API: zero unresolved hotspots.
+- Netdata CI: zero failed checks at the snapshot; 103 checks still running and one CLA check pending.
+
+Open GitHub review threads:
+
+- `src/go/pkg/netipc/transport/internal/framing/handshake.go`: cubic claimed the negotiated request payload should follow `min(max(client, server), 256MB)`.
+  - Current public spec evidence says otherwise: `docs/level1-wire-envelope.md` defines proposals above `1 MiB` as `LIMIT_EXCEEDED` and says accepted request payload limits are echoed unchanged.
+  - Same-language evidence: C POSIX, C Windows, Rust POSIX, Rust Windows, and Go all implement the same `1 MiB` reject plus echo behavior.
+  - Decision for this increment: do not change the handshake contract as a PR-comment side effect. Add source clarification if needed and report the thread as stale/contract-conflicting unless the user explicitly chooses to revise the protocol across docs, C, Rust, Go, and tests.
+- `src/go/pkg/netipc/transport/internal/framing/receive.go`: cubic claimed packets whose received length exceeds the declared frame length should be rejected instead of accepted.
+  - Local verification found the finding is real and same-pattern across all packet transports:
+    - Go shared framing accepted `n >= totalMsg`.
+    - C POSIX UDS accepted `(size_t)n >= total_msg`.
+    - C Windows Named Pipe accepted `n >= total_msg`.
+    - Rust POSIX UDS accepted `n >= total_msg`.
+    - Rust Windows Named Pipe accepted `n >= total_msg`.
+  - Decision for this increment: fix the whole class so exact-length packets are accepted, shorter packets enter the chunked path, and longer packets fail as framing/protocol errors.
+
+Implemented SDK follow-up:
+
+- Added a Go shared-framing comment clarifying the current Level 1 handshake contract: request payload limits are client-proposed, proposals above the wire cap are rejected, and accepted values are echoed unchanged.
+- Changed non-chunked receive completion in Go shared framing, C POSIX UDS, C Windows Named Pipe, Rust POSIX UDS, and Rust Windows Named Pipe from `received_len >= declared_total` to:
+  - reject `received_len > declared_total` as a protocol/framing error.
+  - accept only `received_len == declared_total` as a complete non-chunked message.
+  - keep `received_len < declared_total` as the chunked-message path.
+- Added regression tests for packets that declare one payload byte but contain two bytes in the same packet:
+  - Go shared framing unit test.
+  - POSIX C UDS integration test.
+  - Windows C Named Pipe integration test.
+  - POSIX Rust UDS socketpair test.
+  - Windows Rust Named Pipe test.
+
+Validation completed for this increment:
+
+- `cd src/go && go test -count=1 ./pkg/netipc/transport/internal/framing`: passed.
+- `cd src/go && go test -count=1 ./pkg/netipc/transport/internal/framing ./pkg/netipc/transport/posix`: passed.
+- `cd src/go && go test -count=1 ./pkg/netipc/...`: passed.
+- `cd src/go && GOOS=windows GOARCH=amd64 go test -c -o /tmp/netipc-transport-windows.test.exe ./pkg/netipc/transport/windows`: passed.
+- `cargo test --manifest-path src/crates/netipc/Cargo.toml test_receive_packet_longer_than_declared_payload -- --test-threads=1`: passed.
+- `cargo test --manifest-path src/crates/netipc/Cargo.toml transport::posix -- --test-threads=1`: 60 tests passed.
+- `cargo test --manifest-path src/crates/netipc/Cargo.toml -- --test-threads=1`: 333 tests passed.
+- `cmake --build build --target test_uds`: passed.
+- `/usr/bin/ctest --test-dir build --output-on-failure -R '^test_uds$'`: passed.
+- `cmake --build build`: passed.
+- `/usr/bin/ctest --test-dir build --output-on-failure`: 46/46 tests passed.
+- Win11 MSYS C validation: `test_named_pipe` built and passed.
+- Win11 Go validation: `cd src/go && MSYSTEM=MSYS CGO_ENABLED=0 go test -count=1 ./pkg/netipc/transport/internal/framing ./pkg/netipc/transport/windows`: passed.
+- Win11 Rust validation: `MSYSTEM=MSYS cargo test --manifest-path src/crates/netipc/Cargo.toml transport::windows -- --test-threads=1`: 26 Windows transport tests passed.
+- `git diff --check`: passed.
+
 ## Validation
 
 Acceptance criteria evidence:

src/crates/netipc/src/transport/posix.rs

@@ -448,8 +448,14 @@ impl UdsSession {
 
         let total_msg = HEADER_SIZE + hdr.payload_len as usize;
 
+        if n > total_msg {
+            return Err(UdsError::Protocol(
+                "packet exceeds declared payload_len".into(),
+            ));
+        }
+
         // Non-chunked: entire message in one packet
-        if n >= total_msg {
+        if n == total_msg {
             let payload = &buf[HEADER_SIZE..HEADER_SIZE + hdr.payload_len as usize];
 
             // Validate batch directory

src/crates/netipc/src/transport/posix_tests.rs

@@ -1461,6 +1461,40 @@ fn test_receive_packet_too_short_for_header() {
     unsafe { libc::close(fd1) };
 }
 
+#[test]
+fn test_receive_packet_longer_than_declared_payload() {
+    let (fd0, fd1) = socketpair_seqpacket();
+    let mut session = test_session(fd0, Role::Server, 4096);
+
+    let payload = [0xBE, 0xEF];
+    let mut pkt = [0u8; HEADER_SIZE + 2];
+    let hdr = Header {
+        magic: MAGIC_MSG,
+        version: VERSION,
+        header_len: protocol::HEADER_LEN,
+        kind: KIND_REQUEST,
+        code: 1,
+        flags: 0,
+        transport_status: protocol::STATUS_OK,
+        payload_len: 1,
+        item_count: 1,
+        message_id: 1,
+    };
+    hdr.encode(&mut pkt[..HEADER_SIZE]);
+    pkt[HEADER_SIZE..].copy_from_slice(&payload);
+
+    raw_send(fd1, &pkt).expect("send packet with trailing bytes");
+
+    let mut buf = [0u8; 128];
+    let err = session
+        .receive(&mut buf)
+        .expect_err("trailing bytes should be rejected");
+    assert!(matches!(err, UdsError::Protocol(ref msg)
+        if msg.contains("exceeds declared payload_len")));
+
+    unsafe { libc::close(fd1) };
+}
+
 #[test]
 fn test_receive_batch_directory_too_short_nonchunked() {
     let (fd0, fd1) = socketpair_seqpacket();

src/crates/netipc/src/transport/windows.rs

@@ -849,8 +849,14 @@ impl NpSession {
 
         let total_msg = HEADER_SIZE + hdr.payload_len as usize;
 
+        if n > total_msg {
+            return Err(NpError::Protocol(
+                "packet exceeds declared payload_len".into(),
+            ));
+        }
+
         // Non-chunked
-        if n >= total_msg {
+        if n == total_msg {
             let payload = &buf[HEADER_SIZE..HEADER_SIZE + hdr.payload_len as usize];
 
             // Validate batch directory
@@ -1794,6 +1800,60 @@ mod tests {
         server.join().expect("server join");
     }
 
+    #[cfg(windows)]
+    #[test]
+    fn test_receive_packet_longer_than_declared_payload() {
+        ensure_run_dir();
+        let svc = unique_service("rs_trailing");
+
+        let mut listener =
+            NpListener::bind(TEST_RUN_DIR, &svc, default_server_config()).expect("bind");
+        let server = thread::spawn(move || {
+            let mut session = listener.accept().expect("accept");
+            let mut buf = [0u8; 256];
+            let (req_hdr, _) = session.receive(&mut buf).expect("recv request");
+
+            let mut pkt = [0u8; HEADER_SIZE + 2];
+            let resp_hdr = Header {
+                magic: MAGIC_MSG,
+                version: VERSION,
+                header_len: protocol::HEADER_LEN,
+                kind: KIND_RESPONSE,
+                code: req_hdr.code,
+                flags: 0,
+                transport_status: protocol::STATUS_OK,
+                payload_len: 1,
+                item_count: 1,
+                message_id: req_hdr.message_id,
+            };
+            resp_hdr.encode(&mut pkt[..HEADER_SIZE]);
+            pkt[HEADER_SIZE..].copy_from_slice(&[0xBE, 0xEF]);
+            raw_write(session.handle, &pkt).expect("raw trailing packet write");
+            session.close();
+        });
+
+        let mut session =
+            NpSession::connect(TEST_RUN_DIR, &svc, &default_client_config()).expect("connect");
+        let mut hdr = Header {
+            kind: KIND_REQUEST,
+            code: protocol::METHOD_INCREMENT,
+            item_count: 1,
+            message_id: 43,
+            ..Header::default()
+        };
+        session.send(&mut hdr, &[0xAA]).expect("send request");
+
+        let mut rbuf = [0u8; 256];
+        let err = session
+            .receive(&mut rbuf)
+            .expect_err("trailing bytes should be rejected");
+        assert!(matches!(err, NpError::Protocol(ref msg)
+            if msg.contains("exceeds declared payload_len")));
+
+        session.close();
+        server.join().expect("server join");
+    }
+
     #[cfg(windows)]
     #[test]
     fn test_chunking_and_received_payload() {

src/go/pkg/netipc/transport/internal/framing/handshake.go

@@ -121,6 +121,8 @@ func NegotiateHello(hello protocol.Hello, config ServerHelloConfig) (protocol.He
 	if hello.AuthToken != config.AuthToken {
 		return protocol.HelloAck{}, protocol.StatusAuthFailed, false
 	}
+	// Level 1 keeps request payload limits client-proposed: the server rejects
+	// values over the wire cap and echoes accepted values unchanged.
 	if hello.MaxRequestPayloadBytes > protocol.MaxPayloadCap {
 		return protocol.HelloAck{}, protocol.StatusLimitExceeded, false
 	}

src/go/pkg/netipc/transport/internal/framing/receive.go

@@ -60,7 +60,10 @@ func (r Receiver) Receive(buf []byte) (protocol.Header, []byte, error) {
 		return protocol.Header{}, nil, err
 	}
 	payloadLen := int(hdr.PayloadLen)
-	if n >= totalMsg {
+	if n > totalMsg {
+		return protocol.Header{}, nil, r.ErrProtocol("packet exceeds declared payload_len")
+	}
+	if n == totalMsg {
 		payload := buf[protocol.HeaderSize : protocol.HeaderSize+payloadLen]
 		if err := r.validateBatchPayload(hdr, payload); err != nil {
 			return protocol.Header{}, nil, err

src/go/pkg/netipc/transport/internal/framing/receive_test.go

@@ -0,0 +1,73 @@
+package framing
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"github.com/netdata/plugin-ipc/go/pkg/netipc/protocol"
+)
+
+func TestReceiverRejectsPacketLongerThanDeclaredFrame(t *testing.T) {
+	packet := encodeReceiveTestPacket(1, []byte("ab"))
+	receiver := receiveTestReceiver(packet)
+
+	_, _, err := receiver.Receive(make([]byte, 64))
+	if err == nil {
+		t.Fatal("expected trailing packet bytes to be rejected")
+	}
+	if !strings.Contains(err.Error(), "exceeds declared payload_len") {
+		t.Fatalf("unexpected error: %v", err)
+	}
+}
+
+func TestReceiverAcceptsExactDeclaredFrame(t *testing.T) {
+	packet := encodeReceiveTestPacket(2, []byte("ab"))
+	receiver := receiveTestReceiver(packet)
+
+	hdr, payload, err := receiver.Receive(make([]byte, 64))
+	if err != nil {
+		t.Fatalf("Receive failed: %v", err)
+	}
+	if hdr.PayloadLen != 2 {
+		t.Fatalf("payload_len = %d, want 2", hdr.PayloadLen)
+	}
+	if string(payload) != "ab" {
+		t.Fatalf("payload = %q, want ab", payload)
+	}
+}
+
+func receiveTestReceiver(packet []byte) Receiver {
+	return Receiver{
+		PacketSize:    64,
+		MaxPayload:    64,
+		MaxBatchItems: 1,
+		Recv: func(buf []byte) (int, error) {
+			return copy(buf, packet), nil
+		},
+		ErrLimitExceeded: func(msg string) error { return errors.New(msg) },
+		ErrProtocol:      func(msg string) error { return errors.New(msg) },
+		ErrChunk:         func(msg string) error { return errors.New(msg) },
+		ErrUnknownMsgID:  func(msg string) error { return errors.New(msg) },
+		ErrRecv:          func(msg string) error { return errors.New(msg) },
+	}
+}
+
+func encodeReceiveTestPacket(declaredPayloadLen uint32, payload []byte) []byte {
+	hdr := protocol.Header{
+		Magic:           protocol.MagicMsg,
+		Version:         protocol.Version,
+		HeaderLen:       protocol.HeaderLen,
+		Kind:            protocol.KindRequest,
+		TransportStatus: protocol.StatusOK,
+		PayloadLen:      declaredPayloadLen,
+		ItemCount:       1,
+		MessageID:       1,
+	}
+	packet := make([]byte, protocol.HeaderSize+len(payload))
+	if hdr.Encode(packet) != protocol.HeaderSize {
+		panic("header encode failed")
+	}
+	copy(packet[protocol.HeaderSize:], payload)
+	return packet
+}

src/libnetdata/netipc/src/transport/posix/netipc_uds_receive.c

@@ -244,7 +244,10 @@ nipc_uds_error_t nipc_uds_receive(nipc_uds_session_t *session,
     if (!nipc_uds_header_payload_len(hdr_out->payload_len, &total_msg))
         return NIPC_UDS_ERR_LIMIT_EXCEEDED;
 
-    if ((size_t)n >= total_msg) {
+    if ((size_t)n > total_msg)
+        return NIPC_UDS_ERR_PROTOCOL;
+
+    if ((size_t)n == total_msg) {
         return return_complete_packet(buf, hdr_out, payload_out,
                                       payload_len_out);
     }

src/libnetdata/netipc/src/transport/windows/netipc_named_pipe.c

@@ -1137,8 +1137,11 @@ nipc_np_error_t nipc_np_receive(nipc_np_session_t *session,
     if (!header_payload_len(hdr_out->payload_len, &total_msg))
         return NIPC_NP_ERR_LIMIT_EXCEEDED;
 
+    if (n > total_msg)
+        return NIPC_NP_ERR_PROTOCOL;
+
     /* Non-chunked: entire message in one read */
-    if (n >= total_msg) {
+    if (n == total_msg) {
         *payload_out = (const uint8_t *)buf + NIPC_HEADER_LEN;
         *payload_len_out = hdr_out->payload_len;
 

tests/fixtures/c/test_named_pipe.c

@@ -95,6 +95,7 @@ typedef enum {
     FAKE_ACK_GOOD_THEN_SHORT_RESPONSE,
     FAKE_ACK_GOOD_THEN_BAD_RESPONSE_HEADER,
     FAKE_ACK_GOOD_THEN_ZERO_MESSAGE,
+    FAKE_ACK_GOOD_THEN_TRAILING_RESPONSE,
     FAKE_ACK_CLOSE_BEFORE_ACK,
 } fake_ack_mode_t;
 
@@ -422,7 +423,8 @@ static DWORD WINAPI fake_ack_server_thread(LPVOID arg)
         ctx->mode == FAKE_ACK_GOOD_THEN_BAD_BATCH_DIR ||
         ctx->mode == FAKE_ACK_GOOD_THEN_BAD_CHUNKED_BATCH_DIR ||
         ctx->mode == FAKE_ACK_GOOD_THEN_SHORT_RESPONSE ||
-        ctx->mode == FAKE_ACK_GOOD_THEN_BAD_RESPONSE_HEADER) {
+        ctx->mode == FAKE_ACK_GOOD_THEN_BAD_RESPONSE_HEADER ||
+        ctx->mode == FAKE_ACK_GOOD_THEN_TRAILING_RESPONSE) {
         if (!raw_pipe_read(pipe, buf, sizeof(buf), &bytes_read)) {
             CloseHandle(pipe);
             return 1;
@@ -768,6 +770,26 @@ static DWORD WINAPI fake_ack_server_thread(LPVOID arg)
                 CloseHandle(pipe);
                 return 1;
             }
+        } else if (ctx->mode == FAKE_ACK_GOOD_THEN_TRAILING_RESPONSE) {
+            uint8_t bad_pkt[NIPC_HEADER_LEN + 2] = { 0 };
+            nipc_header_t resp_hdr = {
+                .magic = NIPC_MAGIC_MSG,
+                .version = NIPC_VERSION,
+                .header_len = NIPC_HEADER_LEN,
+                .kind = NIPC_KIND_RESPONSE,
+                .code = req_hdr.code,
+                .item_count = 1,
+                .message_id = req_hdr.message_id,
+                .payload_len = 1,
+                .transport_status = NIPC_STATUS_OK,
+            };
+            nipc_header_encode(&resp_hdr, bad_pkt, NIPC_HEADER_LEN);
+            bad_pkt[NIPC_HEADER_LEN] = 0xBE;
+            bad_pkt[NIPC_HEADER_LEN + 1] = 0xEF;
+            if (!raw_pipe_write(pipe, bad_pkt, (DWORD)sizeof(bad_pkt))) {
+                CloseHandle(pipe);
+                return 1;
+            }
         }
     }
 
@@ -2248,6 +2270,7 @@ static void test_response_protocol_validation(void)
     } cases[] = {
         { "short response header rejected", FAKE_ACK_GOOD_THEN_SHORT_RESPONSE },
         { "bad decoded response header rejected", FAKE_ACK_GOOD_THEN_BAD_RESPONSE_HEADER },
+        { "trailing response packet rejected", FAKE_ACK_GOOD_THEN_TRAILING_RESPONSE },
     };
 
     for (size_t i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) {