diff --git a/README.md b/README.md index 7751782..efbf14c 100644 --- a/README.md +++ b/README.md @@ -113,6 +113,20 @@ Usage of ./dracula-server: ``` +### Protocol Version + +Dracula exposes its wire protocol version in-band with command `Z` over both UDP and TCP. +The version command is intentionally answered before auth validation so clients can safely +detect server capability before choosing a packet signing algorithm. + +Protocol version `2` signs packets with `xxhash64(secret || messageID || namespace || data)` +encoded as little-endian uint64. This is a breaking change from protocol version `1`, whose +auth bytes were just derived from the first 8 bytes of the hash input. + +The server rollout path is controlled by `--accept-legacy-auth` which defaults to `true`. +In that mode the server tries v2 auth first and then legacy v1 auth for normal commands. +Responses are signed using the same auth version as the request so old clients continue to work. + Then use the cli for testing. Put keys to the server: ``` @@ -265,8 +279,8 @@ The namespace can be 64 bytes and the data value can be 1419 bytes. The maximum entries in a key is the highest value of uint32. -Authentication is just strong enough to make sure you aren't sending messages to the wrong server. It is assumed dracula -is running in a trusted environment. +The auth/secret feature is intended only as a validity check. It is assumed dracula is running in +a trusted environment. ## Roadmap diff --git a/cmd/server/main.go b/cmd/server/main.go index 032d4a6..827a1ed 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -18,6 +18,7 @@ var ( secret = flag.String("s", "", "Optional pre-shared auth secret if not using env var DRACULA_SECRET") peerIPPort = flag.String("i", "", "Self peer IP and host like 192.168.0.1:3509 to identify self in the cluster") peers = flag.String("c", "", "Enable cluster replication. Peers must be comma-separated ip:port like `192.168.0.1:3509,192.168.0.2:3555`.") + acceptLegacy = flag.Bool("accept-legacy-auth", true, "Accept legacy v1 packet auth in addition to v2 during rollout") verbose = flag.Bool("v", false, "Verbose logging") printVersion = flag.Bool("version", false, "Print version") promHostPort = flag.String("prom", "", "Enable prometheus metrics. May cause pauses. Example: '0.0.0.0:9090'") @@ -58,6 +59,7 @@ func main() { } else { s = server.NewServer(*expireAfterSecs, preSharedSecret) } + s.SetAcceptLegacyAuth(*acceptLegacy) if *verbose { s.DebugEnable(fmt.Sprintf("udp:%d, tcp:%d, http:%s -", *port, *tcpPort, *restHostPort)) } diff --git a/protocol/protocol.go b/protocol/protocol.go index 7f91893..1a4eda5 100644 --- a/protocol/protocol.go +++ b/protocol/protocol.go @@ -17,11 +17,17 @@ const ( NamespaceSize int = 64 DataValueSize int = 1419 + AuthVersionLegacy = 1 + AuthVersionV2 = 2 + CmdCount byte = 'C' CmdPut byte = 'P' CmdPutReplicate byte = 'R' CmdCountNamespace byte = 'N' CmdCountServer byte = 'S' + CmdVersion byte = 'Z' + + ProtocolVersion = "2" CmdTCPOnlyKeys byte = 'K' CmdTCPOnlyValues byte = 'V' @@ -54,7 +60,7 @@ var StopSymbol = []byte("\n.\n") // IsRequestCmd indicates if the server should accept this as a command func IsRequestCmd(c byte) bool { - return c == CmdCount || c == CmdPut || c == CmdCountNamespace || c == CmdCountServer || c == CmdPutReplicate + return c == CmdCount || c == CmdPut || c == CmdCountNamespace || c == CmdCountServer || c == CmdPutReplicate || c == CmdVersion } func IsTcpOnlyCmd(c byte) bool { @@ -76,6 +82,7 @@ type Packet struct { DataValue []byte // fixed 1419 byte string RequestClient *net.TCPConn + AuthVersion int } // NewPacket is a friendlier way to construct a packet and will provide conversions inline @@ -94,6 +101,7 @@ func NewPacketFromParts(command byte, messageID, namespace, dataValue, preShared MessageIDBytes: messageID, Namespace: *PadRight(&namespace, NamespaceSize), DataValue: *PadRight(&dataValue, DataValueSize), + AuthVersion: AuthVersionV2, } p.SetHash(preSharedKey) return p @@ -142,7 +150,8 @@ func ParsePacket(buf []byte) (*Packet, error) { Namespace: nsBytes, // then a space - DataValue: rightSizeData, + DataValue: rightSizeData, + AuthVersion: AuthVersionV2, } commandIsValid := IsRequestCmd(p.Command) || IsResponseCmd(p.Command) || IsTcpOnlyCmd(p.Command) @@ -219,31 +228,55 @@ func (p *Packet) Bytes() ([]byte, error) { // HashPacket returns an 8 byte slice func HashPacket(p *Packet, preSharedKey []byte) []byte { - hasher := xxhash.New64() + return HashPacketVersion(p, preSharedKey, AuthVersionV2) +} + +// HashPacketVersion returns an 8 byte slice using the requested auth format. +func HashPacketVersion(p *Packet, preSharedKey []byte, authVersion int) []byte { + if authVersion == AuthVersionLegacy { + bytesToHash := make([]byte, 0, len(preSharedKey)+len(p.MessageIDBytes)+len(p.Namespace)+len(p.DataValue)) + bytesToHash = append(bytesToHash, preSharedKey...) + bytesToHash = append(bytesToHash, p.MessageIDBytes...) + bytesToHash = append(bytesToHash, p.Namespace...) + bytesToHash = append(bytesToHash, p.DataValue...) + padded := *PadRight(&bytesToHash, 8) + return padded[0:8] + } // omit the spaces and hash the Message ID, Namespace, and DataValue - bytesToHash := append(preSharedKey, p.MessageIDBytes...) + bytesToHash := make([]byte, 0, len(preSharedKey)+len(p.MessageIDBytes)+len(p.Namespace)+len(p.DataValue)) + bytesToHash = append(bytesToHash, preSharedKey...) + bytesToHash = append(bytesToHash, p.MessageIDBytes...) bytesToHash = append(bytesToHash, p.Namespace...) bytesToHash = append(bytesToHash, p.DataValue...) - return hasher.Sum(bytesToHash) + return Uint64ToBytes(xxhash.Checksum64(bytesToHash)) } // SetHash puts the hash on a packet func (p *Packet) SetHash(preSharedKey []byte) { - // omit the spaces and hash the Message ID, Namespace, and DataValue - bytesToHash := append(preSharedKey, p.MessageIDBytes...) - bytesToHash = append(bytesToHash, p.Namespace...) - bytesToHash = append(bytesToHash, p.DataValue...) - p.HashBytes = HashPacket(p, preSharedKey) + p.SetHashVersion(preSharedKey, AuthVersionV2) +} + +// SetHashVersion puts the requested auth format on a packet. +func (p *Packet) SetHashVersion(preSharedKey []byte, authVersion int) { + p.AuthVersion = authVersion + p.HashBytes = HashPacketVersion(p, preSharedKey, authVersion) p.Hash = Uint64FromBytes(p.HashBytes) } // Validate returns an error is the packet's hash does not authenticate against the preSharedKey. func (p *Packet) Validate(preSharedKey []byte) error { - expectedHash := Uint64FromBytes(HashPacket(p, preSharedKey)) + return p.ValidateVersion(preSharedKey, AuthVersionV2) +} + +// ValidateVersion returns an error if the packet hash does not authenticate +// against the preSharedKey under the requested auth format. +func (p *Packet) ValidateVersion(preSharedKey []byte, authVersion int) error { + expectedHash := Uint64FromBytes(HashPacketVersion(p, preSharedKey, authVersion)) if p.Hash != expectedHash { fmt.Printf("packet hash fail, packet: %d, server: %d \n", p.Hash, expectedHash) return ErrBadHash } + p.AuthVersion = authVersion return nil } diff --git a/protocol/protocol_test.go b/protocol/protocol_test.go index 9fbec2b..4a54144 100644 --- a/protocol/protocol_test.go +++ b/protocol/protocol_test.go @@ -1,6 +1,7 @@ package protocol import ( + "encoding/hex" "github.com/stretchr/testify/assert" "strings" "testing" @@ -64,6 +65,20 @@ func TestParseNewPacketFromPartsWishSecret(t *testing.T) { assert.Nil(t, parsed.Validate(secret)) } +func TestPacketHashUsesXXHash64LittleEndian(t *testing.T) { + packet := NewPacket('C', 1, "throttled", "user@mailsac.com", "eep") + + assert.Equal(t, "6a0eba16a460db3f", hex.EncodeToString(packet.HashBytes)) + assert.Nil(t, packet.Validate([]byte("eep"))) +} + +func TestPacketHashUsesEntireSecret(t *testing.T) { + packet := NewPacket('C', 1, "throttled", "user@mailsac.com", "12345678-good") + + assert.Nil(t, packet.Validate([]byte("12345678-good"))) + assert.Equal(t, ErrBadHash, packet.Validate([]byte("12345678-bad"))) +} + func TestParsePacketSizeTooLarge(t *testing.T) { packet := NewPacket('C', 3321, "somebody", "will replace", "") b, err := packet.Bytes() // will be 1500 diff --git a/server/server.go b/server/server.go index fce352d..3de0376 100644 --- a/server/server.go +++ b/server/server.go @@ -35,6 +35,7 @@ type Server struct { disposed bool preSharedKey []byte expireAfterSecs int64 + acceptLegacyAuth bool messageProcessing chan *rawmessage.RawMessage peers []net.UDPAddr log *log.Logger @@ -82,6 +83,7 @@ func NewServer(expireAfterSecs int64, preSharedKey string) *Server { StoreMetrics: st.LastMetrics, preSharedKey: psk, expireAfterSecs: expireAfterSecs, + acceptLegacyAuth: true, messageProcessing: make(chan *rawmessage.RawMessage, runtime.NumCPU()), log: log.New(os.Stdout, "", 0), } @@ -98,6 +100,10 @@ func (s *Server) DebugDisable() { s.log.SetOutput(ioutil.Discard) } +func (s *Server) SetAcceptLegacyAuth(v bool) { + s.acceptLegacyAuth = v +} + func (s *Server) Listen(udpPort, tcpPort int) error { if s.conn != nil { return ErrServerAlreadyInit @@ -242,7 +248,19 @@ func (s *Server) worker(messages <-chan *rawmessage.RawMessage) { respond() continue } - err = packet.Validate(s.preSharedKey) + if packet.Command == protocol.CmdVersion { + resPacket = protocol.NewPacketFromParts(protocol.CmdVersion, packet.MessageIDBytes, packet.Namespace, []byte(protocol.ProtocolVersion), s.preSharedKey) + respond() + continue + } + err = packet.ValidateVersion(s.preSharedKey, protocol.AuthVersionV2) + if err != nil && s.acceptLegacyAuth { + legacyErr := packet.ValidateVersion(s.preSharedKey, protocol.AuthVersionLegacy) + if legacyErr == nil { + s.log.Println("server accepted legacy auth packet:", remote, string(packet.Command), packet.MessageID, packet.NamespaceString()) + err = nil + } + } if err != nil { s.log.Println("server got bad hash:", remote, string(packet.Command), packet.MessageID, packet.NamespaceString(), packet.DataValueString()) resPacket = protocol.NewPacketFromParts(protocol.ResError, packet.MessageIDBytes, packet.Namespace, []byte(err.Error()), s.preSharedKey) @@ -260,6 +278,7 @@ func (s *Server) worker(messages <-chan *rawmessage.RawMessage) { case protocol.CmdPut: s.store.Put(packet.NamespaceString(), packet.DataValueString()) resPacket = protocol.NewPacketFromParts(protocol.CmdPut, packet.MessageIDBytes, packet.Namespace, []byte{}, s.preSharedKey) + resPacket.SetHashVersion(s.preSharedKey, packet.AuthVersion) respond() if len(s.peers) != 0 { // note that the packet is copied because it will be changed @@ -273,6 +292,7 @@ func (s *Server) worker(messages <-chan *rawmessage.RawMessage) { } c := uint32(countInt) resPacket = protocol.NewPacketFromParts(protocol.CmdCount, packet.MessageIDBytes, packet.Namespace, protocol.Uint32ToBytes(c), s.preSharedKey) + resPacket.SetHashVersion(s.preSharedKey, packet.AuthVersion) respond() break case protocol.CmdCountNamespace: @@ -282,6 +302,7 @@ func (s *Server) worker(messages <-chan *rawmessage.RawMessage) { } c := uint32(countInt) resPacket = protocol.NewPacketFromParts(protocol.CmdCountNamespace, packet.MessageIDBytes, packet.Namespace, protocol.Uint32ToBytes(c), s.preSharedKey) + resPacket.SetHashVersion(s.preSharedKey, packet.AuthVersion) respond() break case protocol.CmdCountServer: @@ -291,22 +312,26 @@ func (s *Server) worker(messages <-chan *rawmessage.RawMessage) { } c := uint32(countInt) resPacket = protocol.NewPacketFromParts(protocol.CmdCountServer, packet.MessageIDBytes, packet.Namespace, protocol.Uint32ToBytes(c), s.preSharedKey) + resPacket.SetHashVersion(s.preSharedKey, packet.AuthVersion) respond() break case protocol.CmdTCPOnlyKeys: matchedKeys := s.store.KeyMatch(packet.NamespaceString(), packet.DataValueString()) s.log.Println("KeyMatch", packet.NamespaceString(), packet.DataValueString(), matchedKeys) resPacket = protocol.NewPacketFromParts(protocol.CmdTCPOnlyKeys, packet.MessageIDBytes, packet.Namespace, []byte(strings.Join(matchedKeys, "\n")), s.preSharedKey) + resPacket.SetHashVersion(s.preSharedKey, packet.AuthVersion) respond() break case protocol.CmdTCPOnlyNamespaces: namespaces := s.store.Namespaces() s.log.Println("Namespaces", packet.NamespaceString(), packet.DataValueString(), namespaces) resPacket = protocol.NewPacketFromParts(protocol.CmdTCPOnlyNamespaces, packet.MessageIDBytes, packet.Namespace, []byte(strings.Join(namespaces, "\n")), s.preSharedKey) + resPacket.SetHashVersion(s.preSharedKey, packet.AuthVersion) respond() break default: resPacket = protocol.NewPacketFromParts(protocol.ResError, packet.MessageIDBytes, packet.Namespace, []byte("unknown_command_"+string(packet.Command)), s.preSharedKey) + resPacket.SetHashVersion(s.preSharedKey, packet.AuthVersion) respond() break } @@ -317,7 +342,7 @@ func (s *Server) worker(messages <-chan *rawmessage.RawMessage) { func (s *Server) republish(packet protocol.Packet) { // re-hash the packet packet.Command = protocol.CmdPutReplicate - packet.SetHash(s.preSharedKey) + packet.SetHashVersion(s.preSharedKey, packet.AuthVersion) b, err := packet.Bytes() if err != nil { diff --git a/server/server_test.go b/server/server_test.go index 3170eff..ec5287a 100644 --- a/server/server_test.go +++ b/server/server_test.go @@ -1,10 +1,13 @@ package server import ( + "bytes" "fmt" "github.com/mailsac/dracula/client" + "github.com/mailsac/dracula/protocol" "github.com/stretchr/testify/assert" "math/rand" + "net" "sync" "testing" "time" @@ -94,6 +97,181 @@ func TestServer_Roundtrip(t *testing.T) { } } +func TestServer_VersionCommandUDPPreAuth(t *testing.T) { + secret := "server-secret" + s := NewServer(60, secret) + if err := s.Listen(9040, 9040); err != nil { + t.Fatal(err) + } + defer s.Close() + + remote, err := net.ResolveUDPAddr("udp", "127.0.0.1:9040") + if err != nil { + t.Fatal(err) + } + conn, err := net.DialUDP("udp", nil, remote) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + req := protocol.NewPacket(protocol.CmdVersion, 1, "", "", "wrong-client-secret") + reqBytes, err := req.Bytes() + if err != nil { + t.Fatal(err) + } + if _, err = conn.Write(reqBytes); err != nil { + t.Fatal(err) + } + + assert.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second))) + resBytes := make([]byte, protocol.PacketSize) + n, err := conn.Read(resBytes) + if err != nil { + t.Fatal(err) + } + res, err := protocol.ParsePacket(resBytes[:n]) + if err != nil { + t.Fatal(err) + } + + assert.Equal(t, protocol.CmdVersion, res.Command) + assert.Equal(t, protocol.ProtocolVersion, res.DataValueString()) + assert.NoError(t, res.Validate([]byte(secret))) +} + +func TestServer_VersionCommandTCPPreAuth(t *testing.T) { + secret := "server-secret" + s := NewServer(60, secret) + if err := s.Listen(9041, 9041); err != nil { + t.Fatal(err) + } + defer s.Close() + + remote, err := net.ResolveTCPAddr("tcp", "127.0.0.1:9041") + if err != nil { + t.Fatal(err) + } + conn, err := net.DialTCP("tcp", nil, remote) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + req := protocol.NewPacket(protocol.CmdVersion, 1, "", "", "wrong-client-secret") + reqBytes, err := req.Bytes() + if err != nil { + t.Fatal(err) + } + if _, err = conn.Write(append(reqBytes, protocol.StopSymbol...)); err != nil { + t.Fatal(err) + } + + assert.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second))) + resBytes := make([]byte, protocol.PacketSize+len(protocol.StopSymbol)) + n, err := conn.Read(resBytes) + if err != nil { + t.Fatal(err) + } + resWire := bytes.TrimSuffix(resBytes[:n], protocol.StopSymbol) + res, err := protocol.ParsePacket(resWire) + if err != nil { + t.Fatal(err) + } + + assert.Equal(t, protocol.CmdVersion, res.Command) + assert.Equal(t, protocol.ProtocolVersion, res.DataValueString()) + assert.NoError(t, res.Validate([]byte(secret))) +} + +func TestServer_LegacyClientCompatibleByDefault(t *testing.T) { + secret := "server-secret" + s := NewServer(60, secret) + if err := s.Listen(9042, 9042); err != nil { + t.Fatal(err) + } + defer s.Close() + + remote, err := net.ResolveUDPAddr("udp", "127.0.0.1:9042") + if err != nil { + t.Fatal(err) + } + conn, err := net.DialUDP("udp", nil, remote) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + req := protocol.NewPacket(protocol.CmdCount, 7, "blocked", "legacy@example.com", secret) + req.SetHashVersion([]byte(secret), protocol.AuthVersionLegacy) + reqBytes, err := req.Bytes() + if err != nil { + t.Fatal(err) + } + if _, err = conn.Write(reqBytes); err != nil { + t.Fatal(err) + } + + assert.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second))) + resBytes := make([]byte, protocol.PacketSize) + n, err := conn.Read(resBytes) + if err != nil { + t.Fatal(err) + } + res, err := protocol.ParsePacket(resBytes[:n]) + if err != nil { + t.Fatal(err) + } + + assert.Equal(t, protocol.CmdCount, res.Command) + assert.NoError(t, res.ValidateVersion([]byte(secret), protocol.AuthVersionLegacy)) +} + +func TestServer_LegacyClientRejectedWhenDisabled(t *testing.T) { + secret := "server-secret" + s := NewServer(60, secret) + s.SetAcceptLegacyAuth(false) + if err := s.Listen(9043, 9043); err != nil { + t.Fatal(err) + } + defer s.Close() + + remote, err := net.ResolveUDPAddr("udp", "127.0.0.1:9043") + if err != nil { + t.Fatal(err) + } + conn, err := net.DialUDP("udp", nil, remote) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + + req := protocol.NewPacket(protocol.CmdCount, 7, "blocked", "legacy@example.com", secret) + req.SetHashVersion([]byte(secret), protocol.AuthVersionLegacy) + reqBytes, err := req.Bytes() + if err != nil { + t.Fatal(err) + } + if _, err = conn.Write(reqBytes); err != nil { + t.Fatal(err) + } + + assert.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second))) + resBytes := make([]byte, protocol.PacketSize) + n, err := conn.Read(resBytes) + if err != nil { + t.Fatal(err) + } + res, err := protocol.ParsePacket(resBytes[:n]) + if err != nil { + t.Fatal(err) + } + + assert.Equal(t, protocol.ResError, res.Command) + assert.NoError(t, res.Validate([]byte(secret))) + assert.Equal(t, protocol.ErrBadHash.Error(), res.DataValueString()) +} + func TestServer_Replication(t *testing.T) { peers := "127.0.0.1:9010,127.0.0.1:9020,127.0.0.1:9030" // setup 3 servers