From: Sergey Matveev <stargrave@stargrave.org>
Date: Sun, 28 Apr 2024 11:35:46 +0000 (+0300)
Subject: Netstrings
X-Git-Tag: v3.0.0~7
X-Git-Url: http://www.git.stargrave.org/?a=commitdiff_plain;h=bbf26700c7d7b1ce9293c5a5060138e3ef22ad84e54ed96d46d2d3b02ecb4eaa;p=vors.git

Netstrings
---

diff --git a/cmd/client/main.go b/cmd/client/main.go
index 460e7b4..a07c75e 100644
--- a/cmd/client/main.go
+++ b/cmd/client/main.go
@@ -20,7 +20,6 @@ import (
 	"crypto/subtle"
 	"encoding/base64"
 	"encoding/binary"
-	"encoding/hex"
 	"flag"
 	"fmt"
 	"io"
@@ -35,6 +34,7 @@ import (
 	"github.com/dchest/siphash"
 	"github.com/flynn/noise"
 	"github.com/jroimartin/gocui"
+	"go.cypherpunks.ru/netstring/v2"
 	"go.stargrave.org/opus/v2"
 	vors "go.stargrave.org/vors/v3/internal"
 	"golang.org/x/crypto/blake2s"
@@ -57,17 +57,6 @@ var (
 	Muted    bool
 )
 
-func parseSID(s string) byte {
-	n, err := strconv.Atoi(s)
-	if err != nil {
-		log.Fatal(err)
-	}
-	if n > 255 {
-		log.Fatal("too big stream num")
-	}
-	return byte(n)
-}
-
 func incr(data []byte) {
 	for i := len(data) - 1; i >= 0; i-- {
 		data[i]++
@@ -103,6 +92,12 @@ func main() {
 		return
 	}
 
+	var passwdHsh []byte
+	if *passwd != "" {
+		hsh := blake2s.Sum256([]byte(*passwd))
+		passwdHsh = hsh[:]
+	}
+
 	srvPub, err := base64.RawURLEncoding.DecodeString(*srvPubB64)
 	if err != nil {
 		log.Fatal(err)
@@ -147,14 +142,15 @@ func main() {
 	}
 
 	vors.PreferIPv4 = *prefer4
-	ctrl, err := net.DialTCP("tcp", nil, vors.MustResolveTCP(*srvAddr))
+	ctrlConn, err := net.DialTCP("tcp", nil, vors.MustResolveTCP(*srvAddr))
 	if err != nil {
 		log.Fatalln("dial server:", err)
 	}
-	defer ctrl.Close()
-	if err = ctrl.SetNoDelay(true); err != nil {
+	defer ctrlConn.Close()
+	if err = ctrlConn.SetNoDelay(true); err != nil {
 		log.Fatalln("nodelay:", err)
 	}
+	ctrl := vors.NewNSConn(ctrlConn)
 
 	hs, err := noise.NewHandshakeState(noise.Config{
 		CipherSuite: vors.NoiseCipherSuite,
@@ -166,26 +162,26 @@ func main() {
 	if err != nil {
 		log.Fatalln("noise.NewHandshakeState:", err)
 	}
-	buf, _, _, err := hs.WriteMessage(nil, []byte(*Name+" "+*Room+" "+*passwd))
+	buf, _, _, err := hs.WriteMessage(nil, vors.ArgsEncode(
+		[]byte(*Name), []byte(*Room), passwdHsh,
+	))
 	if err != nil {
 		log.Fatalln("handshake encrypt:", err)
 	}
-	buf = append(
-		append(
-			[]byte(vors.NoisePrologue),
-			byte((len(buf)&0xFF00)>>8),
-			byte((len(buf)&0x00FF)>>0),
-		),
-		buf...,
-	)
-	_, err = io.Copy(ctrl, bytes.NewReader(buf))
+	{
+		var w bytes.Buffer
+		w.WriteString(vors.NoisePrologue)
+		netstring.NewWriter(&w).WriteChunk(buf)
+		buf = w.Bytes()
+	}
+	_, err = io.Copy(ctrlConn, bytes.NewReader(buf))
 	if err != nil {
 		log.Fatalln("write handshake:", err)
 		return
 	}
-	buf, err = vors.PktRead(ctrl)
-	if err != nil {
-		log.Fatalln("read handshake:", err)
+	buf = <-ctrl.Rx
+	if buf == nil {
+		log.Fatalln("read handshake:", ctrl.Err)
 	}
 	buf, txCS, rxCS, err := hs.ReadMessage(nil, buf)
 	if err != nil {
@@ -194,12 +190,7 @@ func main() {
 
 	rx := make(chan []byte)
 	go func() {
-		for {
-			buf, err := vors.PktRead(ctrl)
-			if err != nil {
-				log.Println("rx", err)
-				break
-			}
+		for buf := range ctrl.Rx {
 			buf, err = rxCS.Decrypt(buf[:0], nil, buf)
 			if err != nil {
 				log.Println("rx decrypt", err)
@@ -217,16 +208,22 @@ func main() {
 	}
 	var sid byte
 	{
-		cols := strings.Fields(string(buf))
-		if cols[0] != "OK" || len(cols) != 2 {
-			log.Fatalln("handshake failed:", cols)
+		args, err := vors.ArgsDecode(buf)
+		if err != nil {
+			log.Fatalln("args decode:", err)
+		}
+		if len(args) < 2 {
+			log.Fatalln("empty args")
 		}
 		var cookie vors.Cookie
-		cookieRaw, err := hex.DecodeString(cols[1])
-		if err != nil {
-			log.Fatal(err)
+		switch cmd := string(args[0]); cmd {
+		case vors.CmdErr:
+			log.Fatalln("handshake failed:", string(args[1]))
+		case vors.CmdCookie:
+			copy(cookie[:], args[1])
+		default:
+			log.Fatalln("unexpected post-handshake cmd:", cmd)
 		}
-		copy(cookie[:], cookieRaw)
 		timeout := time.NewTimer(vors.PingTime)
 		defer func() {
 			if !timeout.Stop() {
@@ -247,12 +244,22 @@ func main() {
 					log.Fatalln("write:", err)
 				}
 			case buf = <-rx:
-				cols = strings.Fields(string(buf))
-				if cols[0] != "SID" || len(cols) != 2 {
-					log.Fatalln("cookie acceptance failed:", string(buf))
+				args, err := vors.ArgsDecode(buf)
+				if err != nil {
+					log.Fatalln("args decode:", err)
+				}
+				if len(args) < 2 {
+					log.Fatalln("empty args")
+				}
+				switch cmd := string(args[0]); cmd {
+				case vors.CmdErr:
+					log.Fatalln("cookie acceptance failed:", string(args[1]))
+				case vors.CmdSID:
+					sid = args[1][0]
+					Streams[sid] = &Stream{name: *Name, stats: OurStats}
+				default:
+					log.Fatalln("unexpected post-cookie cmd:", cmd)
 				}
-				sid = parseSID(cols[1])
-				Streams[sid] = &Stream{name: *Name, stats: OurStats}
 				break WaitForCookieAcceptance
 			}
 		}
@@ -323,11 +330,13 @@ func main() {
 	go func() {
 		for {
 			time.Sleep(vors.PingTime)
-			buf, err := txCS.Encrypt(nil, nil, []byte(vors.CmdPing))
+			buf, err := txCS.Encrypt(nil, nil, vors.ArgsEncode(
+				[]byte(vors.CmdPing),
+			))
 			if err != nil {
 				log.Fatalln("tx encrypt:", err)
 			}
-			if err = vors.PktWrite(ctrl, buf); err != nil {
+			if err = ctrl.Tx(buf); err != nil {
 				log.Fatalln("tx:", err)
 			}
 		}
@@ -336,24 +345,24 @@ func main() {
 	go func(seen *time.Time) {
 		var now time.Time
 		for buf := range rx {
-			if string(buf) == vors.CmdPong {
+			args, err := vors.ArgsDecode(buf)
+			if err != nil {
+				log.Fatalln("args decode:", err)
+			}
+			if len(args) == 0 {
+				log.Fatalln("empty args")
+			}
+			switch cmd := string(args[0]); cmd {
+			case vors.CmdPong:
 				now = time.Now()
 				*seen = now
-				continue
-			}
-			cols := strings.Fields(string(buf))
-			switch cols[0] {
 			case vors.CmdAdd:
-				sidRaw, name, keyHex := cols[1], cols[2], cols[3]
-				log.Println("add", name, "sid:", sidRaw)
-				sid := parseSID(sidRaw)
-				key, err := hex.DecodeString(keyHex)
-				if err != nil {
-					log.Fatal(err)
-				}
+				sidRaw, name, key := args[1], args[2], args[3]
+				sid := sidRaw[0]
+				log.Println("add", string(name), "sid:", sid)
 				keyCiph, keyMAC := key[:chacha20.KeySize], key[chacha20.KeySize:]
 				stream := &Stream{
-					name:  name,
+					name:  string(name),
 					in:    make(chan []byte, 1<<10),
 					stats: &Stats{dead: make(chan struct{})},
 				}
@@ -483,18 +492,18 @@ func main() {
 				go statsDrawer(stream.stats, stream.name)
 				Streams[sid] = stream
 			case vors.CmdDel:
-				sid := parseSID(cols[1])
+				sid := args[1][0]
 				s := Streams[sid]
 				if s == nil {
 					log.Println("unknown sid:", sid)
 					continue
 				}
-				log.Println("del", s.name, "sid:", cols[1])
+				log.Println("del", s.name, "sid:", sid)
 				delete(Streams, sid)
 				close(s.in)
 				close(s.stats.dead)
 			default:
-				log.Fatal("unknown cmd:", cols[0])
+				log.Fatal("unexpected cmd:", cmd)
 			}
 		}
 	}(&seen)
diff --git a/cmd/server/main.go b/cmd/server/main.go
index 4ff5de2..7208476 100644
--- a/cmd/server/main.go
+++ b/cmd/server/main.go
@@ -19,7 +19,6 @@ import (
 	"crypto/rand"
 	"crypto/subtle"
 	"encoding/base64"
-	"encoding/hex"
 	"flag"
 	"fmt"
 	"io"
@@ -74,14 +73,15 @@ func newPeer(conn *net.TCPConn) {
 	if err != nil {
 		log.Fatalln("noise.NewHandshakeState:", err)
 	}
-	buf, err = vors.PktRead(conn)
-	if err != nil {
-		logger.Error("read handshake", "err", err)
+	nsConn := vors.NewNSConn(conn)
+	buf = <-nsConn.Rx
+	if buf == nil {
+		logger.Error("read handshake", "err", nsConn.Err)
 		return
 	}
 	peer := &Peer{
 		logger: logger,
-		conn:   conn,
+		conn:   nsConn,
 		stats:  &Stats{},
 		rx:     make(chan []byte),
 		tx:     make(chan []byte, 10),
@@ -89,21 +89,19 @@ func newPeer(conn *net.TCPConn) {
 	}
 	var room *Room
 	{
-		nameAndRoom, _, _, err := hs.ReadMessage(nil, buf)
+		argsRaw, _, _, err := hs.ReadMessage(nil, buf)
 		if err != nil {
 			logger.Error("handshake: decrypt", "err", err)
 			return
 		}
-		cols := strings.SplitN(string(nameAndRoom), " ", 3)
-		roomName := "/"
-		if len(cols) > 1 {
-			roomName = cols[1]
-		}
-		var key string
-		if len(cols) > 2 {
-			key = cols[2]
+		args, err := vors.ArgsDecode(argsRaw)
+		if err != nil {
+			logger.Error("handshake: decode args", "err", err)
+			return
 		}
-		peer.name = string(cols[0])
+		peer.name = string(args[0])
+		roomName := string(args[1])
+		key := string(args[2])
 		logger = logger.With("name", peer.name, "room", roomName)
 		RoomsM.Lock()
 		room = Rooms[roomName]
@@ -140,11 +138,13 @@ func newPeer(conn *net.TCPConn) {
 		RoomsM.Unlock()
 		if room.key != key {
 			logger.Error("wrong password")
-			buf, _, _, err = hs.WriteMessage(nil, []byte("wrong password"))
+			buf, _, _, err = hs.WriteMessage(nil, vors.ArgsEncode(
+				[]byte(vors.CmdErr), []byte("wrong password"),
+			))
 			if err != nil {
 				log.Fatal(err)
 			}
-			vors.PktWrite(conn, buf)
+			nsConn.Tx(buf)
 			return
 		}
 	}
@@ -155,11 +155,13 @@ func newPeer(conn *net.TCPConn) {
 			continue
 		}
 		logger.Error("name already taken")
-		buf, _, _, err = hs.WriteMessage(nil, []byte("name already taken"))
+		buf, _, _, err = hs.WriteMessage(nil, vors.ArgsEncode(
+			[]byte(vors.CmdErr), []byte("name already taken"),
+		))
 		if err != nil {
 			log.Fatal(err)
 		}
-		vors.PktWrite(conn, buf)
+		nsConn.Tx(buf)
 		return
 	}
 
@@ -181,11 +183,13 @@ func newPeer(conn *net.TCPConn) {
 		}
 		PeersM.Unlock()
 		if !found {
-			buf, _, _, err = hs.WriteMessage(nil, []byte("too many users"))
+			buf, _, _, err = hs.WriteMessage(nil, vors.ArgsEncode(
+				[]byte(vors.CmdErr), []byte("too many users"),
+			))
 			if err != nil {
 				log.Fatal(err)
 			}
-			vors.PktWrite(conn, buf)
+			nsConn.Tx(buf)
 			return
 		}
 	}
@@ -199,7 +203,7 @@ func newPeer(conn *net.TCPConn) {
 		delete(Peers, peer.sid)
 		delete(room.peers, peer.sid)
 		PeersM.Unlock()
-		s := []byte(fmt.Sprintf("%s %d", vors.CmdDel, peer.sid))
+		s := vors.ArgsEncode([]byte(vors.CmdDel), []byte{peer.sid})
 		for _, p := range room.peers {
 			go func(tx chan []byte) { tx <- s }(p.tx)
 		}
@@ -215,11 +219,11 @@ func newPeer(conn *net.TCPConn) {
 
 		var txCS, rxCS *noise.CipherState
 		buf, txCS, rxCS, err := hs.WriteMessage(nil,
-			[]byte(fmt.Sprintf("OK %s", hex.EncodeToString(cookie[:]))))
+			vors.ArgsEncode([]byte(vors.CmdCookie), cookie[:]))
 		if err != nil {
 			log.Fatalln("hs.WriteMessage:", err)
 		}
-		if err = vors.PktWrite(conn, buf); err != nil {
+		if err = nsConn.Tx(buf); err != nil {
 			logger.Error("handshake write", "err", err)
 			delete(Cookies, cookie)
 			return
@@ -235,20 +239,19 @@ func newPeer(conn *net.TCPConn) {
 			return
 		}
 		delete(Cookies, cookie)
-		logger.Info("got cookie", "addr", peer.addr)
 		if !timeout.Stop() {
 			<-timeout.C
 		}
 	}
 	go peer.Rx()
-	peer.tx <- []byte(fmt.Sprintf("SID %d", peer.sid))
+	peer.tx <- vors.ArgsEncode([]byte(vors.CmdSID), []byte{peer.sid})
 
 	for _, p := range room.peers {
 		if p.sid == peer.sid {
 			continue
 		}
-		peer.tx <- []byte(fmt.Sprintf("%s %d %s %s",
-			vors.CmdAdd, p.sid, p.name, hex.EncodeToString(p.key)))
+		peer.tx <- vors.ArgsEncode(
+			[]byte(vors.CmdAdd), []byte{p.sid}, []byte(p.name), p.key)
 	}
 
 	{
@@ -266,8 +269,8 @@ func newPeer(conn *net.TCPConn) {
 	}
 
 	{
-		s := []byte(fmt.Sprintf("%s %d %s %s",
-			vors.CmdAdd, peer.sid, peer.name, hex.EncodeToString(peer.key)))
+		s := vors.ArgsEncode(
+			[]byte(vors.CmdAdd), []byte{peer.sid}, []byte(peer.name), peer.key)
 		for _, p := range room.peers {
 			if p.sid != peer.sid {
 				p.tx <- s
@@ -294,9 +297,21 @@ func newPeer(conn *net.TCPConn) {
 	}(&seen)
 
 	for buf := range peer.rx {
-		if string(buf) == vors.CmdPing {
+		args, err := vors.ArgsDecode(buf)
+		if err != nil {
+			logger.Error("decode args", "err", err)
+			break
+		}
+		if len(args) == 0 {
+			logger.Error("empty args")
+			break
+		}
+		switch cmd := string(args[0]); cmd {
+		case vors.CmdPing:
 			seen = time.Now()
-			peer.tx <- []byte(vors.CmdPong)
+			peer.tx <- vors.ArgsEncode([]byte(vors.CmdPong))
+		default:
+			logger.Error("unknown", "cmd", cmd)
 		}
 	}
 }
diff --git a/cmd/server/peer.go b/cmd/server/peer.go
index 6edb9a7..2250d41 100644
--- a/cmd/server/peer.go
+++ b/cmd/server/peer.go
@@ -35,7 +35,7 @@ type Peer struct {
 	room  *Room
 
 	logger     *slog.Logger
-	conn       net.Conn
+	conn       *vors.NSConn
 	rx, tx     chan []byte
 	rxCS, txCS *noise.CipherState
 	alive      chan struct{}
@@ -47,17 +47,13 @@ func (peer *Peer) Close() {
 		close(peer.rx)
 		close(peer.tx)
 		close(peer.alive)
-		peer.conn.Close()
+		peer.conn.Conn.Close()
 	})
 }
 
 func (peer *Peer) Rx() {
-	for {
-		buf, err := vors.PktRead(peer.conn)
-		if err != nil {
-			peer.logger.Error("rx", "err", err)
-			break
-		}
+	var err error
+	for buf := range peer.conn.Rx {
 		buf, err = peer.rxCS.Decrypt(buf[:0], nil, buf)
 		if err != nil {
 			peer.logger.Error("rx decrypt", "err", err)
@@ -69,20 +65,17 @@ func (peer *Peer) Rx() {
 }
 
 func (peer *Peer) Tx() {
+	var err error
 	for buf := range peer.tx {
 		if peer.txCS == nil {
 			continue
 		}
-		buf, err := peer.txCS.Encrypt(buf[:0], nil, buf)
+		buf, err = peer.txCS.Encrypt(buf[:0], nil, buf)
 		if err != nil {
 			peer.logger.Error("tx encrypt", "err", err)
 			break
 		}
-		err = vors.PktWrite(peer.conn, buf)
-		if err != nil {
-			peer.logger.Error("tx", "err", err)
-			break
-		}
+		peer.conn.Tx(buf)
 	}
 	peer.Close()
 }
diff --git a/doc/proto.texi b/doc/proto.texi
index 1c7312b..97637a3 100644
--- a/doc/proto.texi
+++ b/doc/proto.texi
@@ -27,14 +27,22 @@ curve25519, ChaCha20-Poly1305 and BLAKE2s algorithms.
 
 @item Client sends @code{VoRS v3} to the socket. Just a magic number.
 
-@item All next messages are prepended with 16-bit big-endian length.
+@item All next messages are @url{http://cr.yp.to/proto/netstrings.txt,
+Netstring} encoded strings. Most of them contain netstring encoded
+sequence of netstrings if multiple values are expected:
 
-@item Client sends initial Noise handshake message with his username as
-a payload.
+@example
+NS(NS(arg0) || NS(arg1) || ...)
+@end example
+
+@item Client sends initial Noise handshake message with his username,
+room name and optional BLAKE2s-256 hash of the room's password (or an
+empty string) as a payload: @code{"USERNAME", "ROOM", BLAKE2s(PASSWORD)}.
 
-@item Server answers with final noise handshake message with the payload
-of @code{OK HEX(COOKIE)}, or any other failure message. It may reject a
-client if there are too many peers or its name is already taken.
+@item Server answers with final noise handshake message with the
+@code{"COOKIE", COOKIE}, or @code{"ERR", MSG} failure message.
+It may reject a client if there are too many peers, its name is
+already taken or it provided an invalid room's password.
 
 @item The 128-bit cookie is sent by client over UDP to the server every
 second. If UDP packets are lost, then no connection is possible and
@@ -48,28 +56,28 @@ after a timeout the server drops the TCP connection. That cookie means:
         may differ from known to client one)
     @end itemize
 
-@item Server replies with @code{SID XXX}, where XXX is ASCII decimal
+@item Server replies with @code{"SID", X}, where X is single byte
 stream number client must use.
 
-@item @code{PING} and @code{PONG} messages are then sent every ten
+@item @code{"PING"} and @code{"PONG"} messages are then sent every ten
 seconds as a heartbeat.
 
 @end itemize
 
 @example
-S <- C : e, es, "username"
-S -> C : e, ee, "OK COOKIE"
+S <- C : e, es, NS(NS("USERNAME") || NS("ROOM") || NS("PASSWORD"))
+S -> C : e, ee, NS(NS("COOKIE") || NS(COOKIE))
 S <- C : UDP(COOKIE)
-S -> C : "SID XXX"
+S -> C : NS(NS("SID") || NS(X))
 
-S <- C : "PING"
-S -> C : "PONG"
+S <- C : NS(NS("PING"))
+S -> C : NS(NS("PONG"))
 S <> C : ...
 
-S -> C : "ADD SID USERNAME HEX(KEY)"
+S -> C : NS(NS("ADD") || NS(SID) || NS(USERNAME) || NS(KEY))
 S -> C : ...
 
-S -> C : "DEL SID"
+S -> C : NS(NS("DEL") || NS(SID))
 S -> C : ...
 @end example
 
diff --git a/go.mod b/go.mod
index ebd7a2a..2f9037c 100644
--- a/go.mod
+++ b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/dustin/go-humanize v1.0.1
 	github.com/flynn/noise v1.1.0
 	github.com/jroimartin/gocui v0.5.0
+	go.cypherpunks.ru/netstring/v2 v2.5.0
 	go.stargrave.org/opus/v2 v2.1.0
 	golang.org/x/term v0.19.0
 )
diff --git a/go.sum b/go.sum
index 4646fd1..6fc7e69 100644
--- a/go.sum
+++ b/go.sum
@@ -15,6 +15,8 @@ github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/Qd
 github.com/mattn/go-runewidth v0.0.9/go.mod h1:H031xJmbD/WCDINGzjvQ9THkh0rPKHF+m2gUSrubnMI=
 github.com/nsf/termbox-go v1.1.1 h1:nksUPLCb73Q++DwbYUBEglYBRPZyoXJdrj5L+TkjyZY=
 github.com/nsf/termbox-go v1.1.1/go.mod h1:T0cTdVuOwf7pHQNtfhnEbzHbcNyCEcVU4YPpouCbVxo=
+go.cypherpunks.ru/netstring/v2 v2.5.0 h1:WGo1RhjLkhMcvse9zIi+1Gqd1m3ssU3joslvwHKd4po=
+go.cypherpunks.ru/netstring/v2 v2.5.0/go.mod h1:/b95GfHHgJYdLVJEgektI+hJGsONiB/eXTNOAkJLO6I=
 go.stargrave.org/opus/v2 v2.1.0 h1:WwyMf76wcIWEPIQlU2UI5V9YkqXRHQhq6wfZGslcMFc=
 go.stargrave.org/opus/v2 v2.1.0/go.mod h1:Y57qgcaXH7jBvKW89fscWOT/Wd3MYfhXUbYUcOMV0A8=
 golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
diff --git a/internal/noise.go b/internal/noise.go
index 3528b2c..bd073a1 100644
--- a/internal/noise.go
+++ b/internal/noise.go
@@ -1,10 +1,6 @@
 package internal
 
 import (
-	"bytes"
-	"io"
-	"net"
-
 	"github.com/flynn/noise"
 )
 
@@ -15,22 +11,3 @@ var NoiseCipherSuite = noise.NewCipherSuite(
 	noise.CipherChaChaPoly,
 	noise.HashBLAKE2s,
 )
-
-func PktRead(conn net.Conn) (buf []byte, err error) {
-	buf = make([]byte, 2)
-	_, err = io.ReadFull(conn, buf[:2])
-	if err != nil {
-		return
-	}
-	buf = make([]byte, int(buf[0])<<8|int(buf[1]))
-	_, err = io.ReadFull(conn, buf)
-	return
-}
-
-func PktWrite(conn net.Conn, buf []byte) (err error) {
-	_, err = io.Copy(conn, bytes.NewReader(append([]byte{
-		byte((len(buf) & 0xFF00) >> 8),
-		byte((len(buf) & 0x00FF) >> 0),
-	}, buf...)))
-	return
-}
diff --git a/internal/ns.go b/internal/ns.go
new file mode 100644
index 0000000..c65aecf
--- /dev/null
+++ b/internal/ns.go
@@ -0,0 +1,80 @@
+package internal
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"log"
+	"net"
+	"sync"
+
+	"go.cypherpunks.ru/netstring/v2"
+)
+
+func ArgsEncode(datum ...[]byte) []byte {
+	var buf bytes.Buffer
+	w := netstring.NewWriter(&buf)
+	for _, data := range datum {
+		if _, err := w.WriteChunk(data); err != nil {
+			log.Fatal(err)
+		}
+	}
+	return buf.Bytes()
+}
+
+func ArgsDecode(buf []byte) (args [][]byte, err error) {
+	r := netstring.NewReader(bytes.NewReader(buf))
+	var n uint64
+	for {
+		n, err = r.Next()
+		if err != nil {
+			if errors.Is(err, io.EOF) {
+				err = nil
+				break
+			}
+			return
+		}
+		arg := make([]byte, int(n))
+		_, err = io.ReadFull(r, arg)
+		if err != nil {
+			return
+		}
+		args = append(args, arg)
+	}
+	return
+}
+
+type NSConn struct {
+	Conn net.Conn
+	Rx   chan []byte
+	Err  error
+	sync.Mutex
+}
+
+func NewNSConn(conn net.Conn) *NSConn {
+	c := NSConn{Conn: conn, Rx: make(chan []byte)}
+	go func() {
+		r := netstring.NewReader(conn)
+		var n uint64
+		for {
+			n, c.Err = r.Next()
+			if c.Err != nil {
+				break
+			}
+			buf := make([]byte, int(n))
+			if _, c.Err = io.ReadFull(r, buf); c.Err != nil {
+				break
+			}
+			c.Rx <- buf
+		}
+		close(c.Rx)
+	}()
+	return &c
+}
+
+func (ns *NSConn) Tx(data []byte) (err error) {
+	ns.Lock()
+	_, err = netstring.NewWriter(ns.Conn).WriteChunk(data)
+	ns.Unlock()
+	return
+}
diff --git a/internal/var.go b/internal/var.go
index 0ef1d5a..ba13871 100644
--- a/internal/var.go
+++ b/internal/var.go
@@ -6,10 +6,13 @@ import (
 )
 
 const (
-	CmdPing = "PING"
-	CmdPong = "PONG"
-	CmdAdd  = "ADD"
-	CmdDel  = "DEL"
+	CmdErr    = "ERR"
+	CmdCookie = "COOKIE"
+	CmdSID    = "SID"
+	CmdPing   = "PING"
+	CmdPong   = "PONG"
+	CmdAdd    = "ADD"
+	CmdDel    = "DEL"
 )
 
 var (