WIP: random message to generate test data, looks good

go.mod

@@ -3,9 +3,11 @@ module gitea.suyono.dev/suyono/netbounce
 go 1.24
 
 require (
+	gitea.suyono.dev/suyono/go-sizes v0.1.2
 	github.com/rs/zerolog v1.34.0
 	github.com/spf13/pflag v1.0.6
 	github.com/spf13/viper v1.20.1
+	golang.org/x/crypto v0.38.0
 	golang.org/x/sys v0.33.0
 )
 
@@ -22,7 +24,6 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
-	golang.org/x/crypto v0.38.0 // indirect
 	golang.org/x/text v0.25.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )

go.sum

@@ -1,3 +1,5 @@
+gitea.suyono.dev/suyono/go-sizes v0.1.2 h1:uXViVtTh8/mrJrrv7ApBMIOD4/0dz+Z1Y2R2anqklhU=
+gitea.suyono.dev/suyono/go-sizes v0.1.2/go.mod h1:v9jk4b+wlIUkYoT6KwoLHfE/EpFefey6c/WR51liI98=
 github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -59,12 +61,8 @@ golang.org/x/crypto v0.38.0/go.mod h1:MvrbAqul58NNYPKnOra203SB9vpuZW0e+RRZV+Ggqj
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.32.0 h1:s77OFDvIQeibCmezSnk/q6iAfkdiQaJi4VzroCFrN20=
-golang.org/x/sys v0.32.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
 golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
 golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
-golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
-golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=
 golang.org/x/text v0.25.0 h1:qVyWApTSYLk/drJRO5mDlNYskwQznZmkpV2c8q9zls4=
 golang.org/x/text v0.25.0/go.mod h1:WEdwpYrmk1qmdHvhkSTNPm3app7v4rsT8F2UD6+VHIA=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

testlib/randommessage/chunkedmessage.go

@@ -0,0 +1,70 @@
+package randommessage
+
+/*
+   Copyright 2025 Suyono <suyono3484@gmail.com>
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+import (
+	"encoding/binary"
+	"errors"
+	"hash/crc32"
+	"io"
+)
+
+type ChunkedMessage struct {
+	seq uint32
+	sr  *StreamReader
+}
+
+func NewChunkedReader(size int64) (*ChunkedMessage, error) {
+	sr, err := NewStreamReader(size)
+	if err != nil {
+		return nil, err
+	}
+
+	return &ChunkedMessage{
+		seq: 0,
+		sr:  sr,
+	}, nil
+}
+
+func (c *ChunkedMessage) Read(buf []byte) (int, error) {
+	if len(buf) < 8 {
+		return 0, ErrInvalidBufSize
+	}
+
+	var (
+		n   int
+		err error
+	)
+	limit := len(buf) - 4
+
+	binary.BigEndian.PutUint32(buf, c.seq)
+	if n, err = c.sr.Read(buf[4:limit]); err != nil && !errors.Is(err, io.EOF) {
+		return n, err
+	}
+
+	if n == 0 {
+		return n, err
+	}
+
+	if n < limit-4 {
+		limit = n + 4
+	}
+
+	binary.BigEndian.PutUint32(buf[limit:], crc32.Checksum(buf[:limit], crc32.MakeTable(crc32.Castagnoli)))
+	c.seq++
+	return n + 8, err
+}

testlib/randommessage/randommessage.go

@@ -17,6 +17,7 @@ package randommessage
 */
 
 import (
+	"bytes"
 	"crypto/rand"
 	"encoding/binary"
 	"errors"
@@ -30,10 +31,14 @@ type StreamReader struct {
 	pos        int64
 	size       int64
 	hashEngine hash.Hash
+	sum        []byte
+	hashPos    int
 }
 
 var (
 	ErrInvalidBufSize        = errors.New("invalid buffer size")
+	ErrInvalidTrailingLength = errors.New("invalid trailing length")
+	ErrMessageVerification   = errors.New("message verification failed")
 )
 
 func NewStreamReader(size int64) (*StreamReader, error) {
@@ -49,6 +54,8 @@ func NewStreamReader(size int64) (*StreamReader, error) {
 		pos:        0,
 		size:       size,
 		hashEngine: h,
+		sum:        nil,
+		hashPos:    0,
 	}, nil
 }
 
@@ -70,6 +77,25 @@ func (s *StreamReader) Read(buf []byte) (int, error) {
 		n, err = s.read(buf)
 	}
 
+	if err != nil && errors.Is(err, io.EOF) {
+		if s.hashPos > 0 && s.hashPos < blake2s.Size {
+			n += copy(buf[n:], s.sum[s.hashPos:])
+			s.hashPos += n
+			if s.hashPos < blake2s.Size {
+				err = nil
+			}
+		} else if s.hashPos == 0 {
+			s.sum = s.hashEngine.Sum(nil)
+			if n < len(buf) {
+				s.hashPos = copy(buf[n:], s.sum[s.hashPos:])
+				if s.hashPos < blake2s.Size {
+					err = nil
+				}
+				n += s.hashPos
+			}
+		}
+	}
+
 	return n, err
 }
 
@@ -98,3 +124,74 @@ func (s *StreamReader) read(buf []byte) (int, error) {
 
 	return n, err
 }
+
+type MessageChecker struct {
+	expectedLen int64
+	hashEngine  hash.Hash
+	pos         int64
+	buffer      *bytes.Buffer
+}
+
+func NewMessageChecker() (*MessageChecker, error) {
+	var (
+		h   hash.Hash
+		err error
+	)
+
+	if h, err = blake2s.New256(nil); err != nil {
+		return nil, err
+	}
+
+	return &MessageChecker{
+		hashEngine:  h,
+		expectedLen: -1,
+		pos:         0,
+		buffer:      bytes.NewBuffer(nil),
+	}, nil
+}
+
+func (m *MessageChecker) Write(buf []byte) (int, error) {
+	var (
+		n, limit int
+	)
+
+	n = 0
+	if m.expectedLen == -1 {
+		if len(buf) < 4 {
+			return 0, ErrInvalidBufSize
+		}
+		m.expectedLen = int64(binary.BigEndian.Uint32(buf))
+		n = 4
+	}
+
+	if m.expectedLen > 0 {
+		limit = len(buf)
+		if m.pos < m.expectedLen {
+			if m.pos+int64(limit-n) > m.expectedLen {
+				limit = int(int64(limit) - (m.pos + int64(limit-n) - m.expectedLen))
+				m.buffer.Write(buf[limit:])
+			}
+			m.hashEngine.Write(buf[n:limit])
+			m.pos += int64(limit - n)
+		} else {
+			m.buffer.Write(buf[n:limit])
+			m.pos += int64(limit - n)
+		}
+	}
+
+	return len(buf), nil
+}
+
+func (m *MessageChecker) Close() error {
+	buf := m.buffer.Bytes()
+	if len(buf) != blake2s.Size {
+		return ErrInvalidTrailingLength
+	}
+
+	sum := m.hashEngine.Sum(nil)
+	if !bytes.Equal(sum, buf) {
+		return ErrMessageVerification
+	}
+
+	return nil
+}

testlib/randommessage/randommessage_test.go

@@ -0,0 +1,183 @@
+package randommessage
+
+/*
+   Copyright 2025 Suyono <suyono3484@gmail.com>
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"hash"
+	"hash/crc32"
+	"io"
+	"testing"
+
+	"gitea.suyono.dev/suyono/go-sizes/sizes"
+	"golang.org/x/crypto/blake2s"
+)
+
+func TestStreamReader(t *testing.T) {
+	var (
+		n        int
+		h        hash.Hash
+		testSize int64 = int64(16 * sizes.KibiByte)
+		mc       *MessageChecker
+	)
+
+	sr, err := NewStreamReader(testSize)
+	if err != nil {
+		t.Fatalf("failed to initialize stream: %v", err)
+	}
+
+	buf := make([]byte, (10 * sizes.KibiByte).MustInt())
+
+	if n, err = sr.Read(buf); err != nil {
+		t.Fatalf("failed to read from stream: %v", err)
+	}
+
+	if int64(binary.BigEndian.Uint32(buf)) != testSize {
+		t.Fatal("test size failed")
+	}
+
+	if mc, err = NewMessageChecker(); err != nil {
+		t.Fatalf("failed to instantiate Message Checker: %v", err)
+	}
+
+	h, err = blake2s.New256(nil)
+	if err != nil {
+		t.Fatalf("failed to instanciate blake2s: %v", err)
+	}
+	h.Write(buf[4:n])
+	if _, err = mc.Write(buf[:n]); err != nil {
+		t.Fatalf("write to Message Checker failed: %v", err)
+	}
+
+	if n, err = sr.Read(buf); err != nil {
+		if !errors.Is(err, io.EOF) {
+			t.Fatalf("failed to read from stream: %v", err)
+		}
+	}
+	h.Write(buf[:n-32])
+	sum := h.Sum(nil)
+	if !bytes.Equal(buf[n-32:n], sum) {
+		t.Fatal("mismatch hash")
+	}
+	if _, err = mc.Write(buf[:n]); err != nil {
+		t.Fatalf("write to Message Checker failed: %v", err)
+	}
+	if err = mc.Close(); err != nil {
+		t.Fatalf("message checker: %v", err)
+	}
+}
+
+func TestStreamReaderCasePartialHash(t *testing.T) {
+	var (
+		n        int
+		testSize int64 = int64(16 * sizes.KibiByte)
+		mc       *MessageChecker
+	)
+
+	sr, err := NewStreamReader(testSize)
+	if err != nil {
+		t.Fatalf("failed to initialize stream: %v", err)
+	}
+
+	buf := make([]byte, (10 * sizes.KibiByte).MustInt())
+
+	if n, err = sr.Read(buf); err != nil {
+		t.Fatalf("failed to read from stream: %v", err)
+	}
+
+	if int64(binary.BigEndian.Uint32(buf)) != testSize {
+		t.Fatal("test size failed")
+	}
+
+	if mc, err = NewMessageChecker(); err != nil {
+		t.Fatalf("failed to instantiate Message Checker: %v", err)
+	}
+
+	if _, err = mc.Write(buf[:n]); err != nil {
+		t.Fatalf("write to Message Checker failed: %v", err)
+	}
+
+	if n, err = sr.Read(buf[:6170]); err != nil {
+		t.Fatalf("failed to read from stream: %v", err)
+	}
+
+	var m int
+	if m, err = sr.Read(buf[n:]); err != nil {
+		if !errors.Is(err, io.EOF) {
+			t.Fatalf("failed to read from stream: %v", err)
+		}
+	}
+	n += m
+
+	if _, err = mc.Write(buf[:n]); err != nil {
+		t.Fatalf("write to Message Checker failed: %v", err)
+	}
+	if err = mc.Close(); err != nil {
+		t.Fatalf("message checker: %v", err)
+	}
+}
+
+func TestChunkedMessage(t *testing.T) {
+	var (
+		cm  *ChunkedMessage
+		err error
+		n   int
+		crc uint32
+		h   hash.Hash
+	)
+
+	if cm, err = NewChunkedReader(int64(16 * sizes.KibiByte)); err != nil {
+		t.Fatalf("failed to instantiate chunked reader: %v", err)
+	}
+
+	if h, err = blake2s.New256(nil); err != nil {
+		t.Fatalf("failed to instantiate blake2s: %v", err)
+	}
+	buf := make([]byte, sizes.KibiByte.MustInt())
+	for i := range 16 {
+		if n, err = cm.Read(buf); err != nil {
+			t.Fatalf("failed to read: %v", err)
+		}
+		crc = binary.BigEndian.Uint32(buf[n-4:])
+		if crc != crc32.Checksum(buf[:n-4], crc32.MakeTable(crc32.Castagnoli)) {
+			t.Fatalf("mismatch checksum")
+		}
+
+		if i == 0 {
+			h.Write(buf[8 : n-4])
+		} else {
+			h.Write(buf[4 : n-4])
+		}
+	}
+	if n, err = cm.Read(buf); err != nil && !errors.Is(err, io.EOF) {
+		t.Fatalf("failed to read: %v", err)
+	}
+	if err == nil {
+		t.Fatal("unexpected nil error")
+	}
+	crc = binary.BigEndian.Uint32(buf[n-4:])
+	if crc != crc32.Checksum(buf[:n-4], crc32.MakeTable(crc32.Castagnoli)) {
+		t.Fatalf("mismatch checksum")
+	}
+
+	h.Write(buf[4 : n-36])
+	if !bytes.Equal(h.Sum(nil), buf[n-36:n-4]) {
+		t.Fatal("mismatch hash")
+	}
+}