diff --git a/foreign/go/client/connection_string.go b/foreign/go/client/connection_string.go new file mode 100644 index 0000000000..0d728e6bf2 --- /dev/null +++ b/foreign/go/client/connection_string.go @@ -0,0 +1,485 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +package client + +import ( + "fmt" + "math" + "math/big" + "net" + "strconv" + "strings" + "time" + + "github.com/apache/iggy/foreign/go/client/tcp" + ierror "github.com/apache/iggy/foreign/go/errors" +) + +const ( + defaultProtocol = "iggy" + protocolPrefix = "iggy+" + tcpProtocol = "tcp" +) + +// ParseConnectionString parses an Iggy TCP connection string into the TCP +// client options it describes, so the result can be passed to WithTcp: +// +// client.NewIggyClient(client.WithTcp(opts...)) +// +// with opts from client.ParseConnectionString("iggy://iggy:iggy@localhost:8090"). +// +// The accepted shape is: +// +// iggy://[username:password@ | personal_access_token@]host:port[?options] +// iggy+tcp://[username:password@ | personal_access_token@]host:port[?options] +// +// The scheme iggy:// means TCP, the same as the explicit iggy+tcp://. +// Other iggy+ schemes such as iggy+quic:// are rejected, +// because the Go SDK implements the TCP transport only, which covers both +// plain TCP and TLS. Credentials are username:password or a single +// personal access token before the @; either way auto-login is configured, +// matching what the Rust and Node SDKs do with a connection string. The +// host may be a bracketed IPv6 literal, as in [::1]:8090. +// +// Query options follow the grammar the other SDKs share: +// +// tls=true|false enable TLS +// tls_domain= TLS server name indication (SNI) +// tls_ca_file= CA certificate file for verification +// reconnection_retries= +// reconnection_interval= +// reestablish_after= +// heartbeat_interval= +// nodelay=true|false +// +// Durations take the units the other SDKs share, from nanos to years, +// including the hr, hrs, wk, wks, month and µs spellings, with optional +// fractions and parts run together or spaced, as in 1h30m, 1h 30m or 1 s. +// The reconnection_interval and heartbeat_interval spellings must be +// non-zero and no shorter than one microsecond, because the Rust SDK +// types them as non-zero durations whose wire format carries whole +// microseconds; reestablish_after takes the zero spellings, which Rust +// accepts there. +// The Go client keeps its own reconnect pacing, so the reconnection +// spellings are validated for compatibility and then ignored, and +// a value the other SDKs would reject is still rejected here, so one +// connection string is either valid everywhere or nowhere. The +// heartbeat_interval and nodelay spellings are applied, because the Go +// client carries matching options: heartbeat_interval defaults to the +// client five-second pacing, and nodelay defaults to true, the Go +// client default, so only nodelay=false ever turns Nagle back on. +func ParseConnectionString(connectionString string) ([]tcp.Option, error) { + protocolParts := strings.Split(connectionString, "://") + if len(protocolParts) != 2 { + return nil, ierror.InvalidConnectionString{} + } + + protocol, err := parseProtocol(protocolParts[0]) + if err != nil { + return nil, err + } + if protocol != tcpProtocol { + // Only a plain transport name is echoed, because a malformed + // scheme may carry credentials, as the C# SDK notes on the same + // check. + if !isASCIILetters(protocol) { + return nil, ierror.InvalidConnectionString{} + } + return nil, fmt.Errorf("%w: unsupported transport %q, Go SDK supports tcp only", + ierror.InvalidConnectionString{}, protocol) + } + + parts := strings.Split(protocolParts[1], "@") + if len(parts) != 2 { + return nil, ierror.InvalidConnectionString{} + } + + credentials, err := parseCredentials(parts[0]) + if err != nil { + return nil, err + } + + serverAndOptions := strings.Split(parts[1], "?") + if len(serverAndOptions) > 2 { + return nil, ierror.InvalidConnectionString{} + } + + // The no-delay default lives here for the no-options path and in + // parseConnectionOptions for queries, so a query that leaves the + // spelling out keeps Nagle off. + var tlsConfig parsedTLS + tlsConfig.noDelay = true + if len(serverAndOptions) == 2 { + tlsConfig, err = parseConnectionOptions(serverAndOptions[1]) + if err != nil { + return nil, err + } + } + + serverAddress := serverAndOptions[0] + if serverAddress == "" { + return nil, ierror.InvalidConnectionString{} + } + host, portStr, err := net.SplitHostPort(serverAddress) + if err != nil || host == "" { + return nil, ierror.InvalidConnectionString{} + } + // Node and Rust accept any u16 port, 0 included; the dial fails later + // instead of the parse. A bad port reports the same error as every other + // parse failure, which is what the Rust SDK returns here. + if _, err := strconv.ParseUint(portStr, 10, 16); err != nil { + return nil, ierror.InvalidConnectionString{} + } + + options := []tcp.Option{tcp.WithServerAddress(serverAddress)} + options = append(options, tcp.WithAutoLogin(credentials)) + if tlsConfig.enabled { + var tlsOpts []tcp.TLSOption + if tlsConfig.domain != "" { + tlsOpts = append(tlsOpts, tcp.WithTLSDomain(tlsConfig.domain)) + } + if tlsConfig.caFile != "" { + tlsOpts = append(tlsOpts, tcp.WithTLSCAFile(tlsConfig.caFile)) + } + options = append(options, tcp.WithTLS(tlsOpts...)) + } + options = append(options, tcp.WithNoDelay(tlsConfig.noDelay)) + if tlsConfig.heartbeatIntervalNanos != 0 { + // time.Duration counts nanoseconds in an int64, so an interval the + // grammar accepts but Go cannot represent fails the parse, like + // every other spelling the connection string cannot honor. + if tlsConfig.heartbeatIntervalNanos > math.MaxInt64 { + return nil, ierror.InvalidConnectionString{} + } + options = append(options, tcp.WithHeartbeatInterval(time.Duration(tlsConfig.heartbeatIntervalNanos))) + } + return options, nil +} + +// parseProtocol extracts the transport from the scheme, which is either the +// bare default protocol or an explicit iggy+ spelling. +func parseProtocol(scheme string) (string, error) { + if scheme == defaultProtocol { + return tcpProtocol, nil + } + if !strings.HasPrefix(scheme, protocolPrefix) { + return "", ierror.InvalidConnectionString{} + } + return strings.TrimPrefix(scheme, protocolPrefix), nil +} + +// parseCredentials turns the userinfo before the @ into auto-login +// credentials: username:password or a single personal access token. +func parseCredentials(userinfo string) (tcp.Credentials, error) { + username, password, hasSeparator := strings.Cut(userinfo, ":") + if !hasSeparator { + if username == "" { + return tcp.Credentials{}, ierror.InvalidConnectionString{} + } + return tcp.NewPersonalAccessTokenCredentials(username), nil + } + if username == "" || password == "" { + return tcp.Credentials{}, ierror.InvalidConnectionString{} + } + return tcp.NewUsernamePasswordCredentials(username, password), nil +} + +type parsedTLS struct { + enabled bool + domain string + caFile string + noDelay bool + // heartbeatIntervalNanos carries the parsed heartbeat_interval, zero + // when the spelling is absent so the client keeps its own pacing. + heartbeatIntervalNanos uint64 +} + +// parseConnectionOptions validates the shared connection-string grammar, +// keeping the tls and heartbeat options and ignoring the reconnect +// spellings the Go client does not take from a connection string. +func parseConnectionOptions(optionsString string) (parsedTLS, error) { + parsed := parsedTLS{noDelay: true} + if optionsString == "" { + return parsed, ierror.InvalidConnectionString{} + } + for _, option := range strings.Split(optionsString, "&") { + name, value, found := strings.Cut(option, "=") + if !found { + return parsed, ierror.InvalidConnectionString{} + } + switch name { + case "tls": + if value != "true" && value != "false" { + return parsed, ierror.InvalidConnectionString{} + } + parsed.enabled = value == "true" + case "tls_domain": + parsed.domain = value + case "tls_ca_file": + parsed.caFile = value + case "reconnection_retries": + if value != "unlimited" { + if _, err := strconv.ParseUint(value, 10, 32); err != nil { + return parsed, ierror.InvalidNumberValue{} + } + } + case "reconnection_interval", "heartbeat_interval": + // Rust types both as NonZeroIggyDuration, which refuses the + // zero spellings and anything shorter than the whole + // microseconds the wire format carries. + nanos, err := parseIggyDuration(value) + if err != nil || nanos < 1_000 { + return parsed, ierror.InvalidConnectionString{} + } + if name == "heartbeat_interval" { + parsed.heartbeatIntervalNanos = nanos + } + case "reestablish_after": + // Rust takes this one as a plain duration, where the zero + // spellings parse. + if _, err := parseIggyDuration(value); err != nil { + return parsed, ierror.InvalidConnectionString{} + } + case "nodelay": + if value != "true" && value != "false" { + return parsed, ierror.InvalidConnectionString{} + } + parsed.noDelay = value == "true" + default: + return parsed, ierror.InvalidConnectionString{} + } + } + return parsed, nil +} + +// isASCIILetters reports whether s is non-empty and ASCII letters only, so +// an unsupported-transport error can echo it without ever echoing the rest +// of a malformed scheme, which may carry credentials. +func isASCIILetters(s string) bool { + if s == "" { + return false + } + for i := 0; i < len(s); i++ { + if (s[i] < 'a' || s[i] > 'z') && (s[i] < 'A' || s[i] > 'Z') { + return false + } + } + return true +} + +// parseIggyDuration parses a duration the Rust SDK accepts, returning the +// total nanoseconds. It mirrors IggyDuration::from_str: the value is +// lowercased, the zero spellings become a zero duration, and everything +// else goes through a humantime-style grammar of amounts and units, with +// optional fractions and parts run together as in 1h30m or spaced as in +// 1h 30m. A space is allowed between an amount and its unit as in 1 s, +// and the µs spelling parses like us. Fractions must land on a whole +// nanosecond, as in the Rust and Node parsers, so 1.5us parses while +// 0.5ns does not. +func parseIggyDuration(value string) (uint64, error) { + lower := strings.ToLower(value) + switch lower { + case "0", "unlimited", "disabled", "none": + return 0, nil + } + remainder := strings.TrimLeft(lower, " ") + if remainder == "" { + return 0, errInvalidDuration(value) + } + + var seconds, nanos big.Int + for remainder != "" { + wholePart, fracPart, fracScale, rest, ok := cutLeadingAmount(remainder) + if !ok { + return 0, errInvalidDuration(value) + } + // humantime skips whitespace between an amount and its unit, as + // in 1 s, the way spaces between parts are skipped below. + remainder = strings.TrimLeft(rest, " ") + + index, rest, ok := cutDurationUnit(remainder) + if !ok { + return 0, errInvalidDuration(value) + } + remainder = strings.TrimLeft(rest, " ") + unit := durationUnits[index] + + whole := new(big.Int).SetUint64(wholePart) + seconds.Add(&seconds, whole.Mul(whole, new(big.Int).SetUint64(unit.seconds))) + whole = new(big.Int).SetUint64(wholePart) + nanos.Add(&nanos, whole.Mul(whole, new(big.Int).SetUint64(unit.nanos))) + + if fracScale != 0 { + // A fraction of a nanosecond always loses precision, as the + // Rust parser reports outright. + if unit.seconds == 0 && unit.nanos == 1 { + return 0, errInvalidDuration(value) + } + numerator := new(big.Int).SetUint64(fracPart) + denominator := new(big.Int).SetUint64(fracScale) + frac := new(big.Int) + switch { + case unit.nanos != 0: // usec and millis land in nanoseconds + frac.Mul(numerator, new(big.Int).SetUint64(unit.nanos)) + if new(big.Int).Mod(frac, denominator).Sign() != 0 { + return 0, errInvalidDuration(value) + } + nanos.Add(&nanos, frac.Div(frac, denominator)) + case unit.seconds == 1: // seconds land in nanoseconds + frac.Mul(numerator, big.NewInt(1_000_000_000)) + if new(big.Int).Mod(frac, denominator).Sign() != 0 { + return 0, errInvalidDuration(value) + } + nanos.Add(&nanos, frac.Div(frac, denominator)) + default: // minutes and above land in whole seconds + frac.Mul(numerator, new(big.Int).SetUint64(unit.seconds)) + if new(big.Int).Mod(frac, denominator).Sign() != 0 { + return 0, errInvalidDuration(value) + } + seconds.Add(&seconds, frac.Div(frac, denominator)) + } + } + } + + total := new(big.Int).Mul(&seconds, big.NewInt(1_000_000_000)) + total.Add(total, &nanos) + if total.BitLen() > 64 { + return 0, errInvalidDuration(value) + } + return total.Uint64(), nil +} + +func errInvalidDuration(value string) error { + return fmt.Errorf("%w: invalid duration %q", ierror.InvalidConnectionString{}, value) +} + +// cutLeadingAmount consumes the leading amount of s: decimal digits with an +// optional fractional part, as in 5, 0.5 or 1.005. A leading sign is +// rejected, because negative durations do not parse, and so is a fraction +// without a digit after the dot. The fraction comes back as a numerator and +// a power-of-ten scale, so 1.005 is 1005 over 1000 and the caller can +// divide exactly the way the Rust parser does. +func cutLeadingAmount(s string) (whole uint64, fraction uint64, scale uint64, rest string, ok bool) { + digits := 0 + for digits < len(s) && isDigitByte(s[digits]) { + digits++ + } + if digits == 0 { + return 0, 0, 0, "", false + } + whole, err := strconv.ParseUint(s[:digits], 10, 64) + if err != nil { + return 0, 0, 0, "", false + } + rest = s[digits:] + if len(rest) >= 2 && rest[0] == '.' && isDigitByte(rest[1]) { + fractionDigits := 1 + for fractionDigits < len(rest) && isDigitByte(rest[fractionDigits]) { + fractionDigits++ + } + text := rest[1:fractionDigits] + if len(text) > 19 { + // Beyond u64 precision, which a fraction could not carry + // exactly anyway. + return 0, 0, 0, "", false + } + fraction, err = strconv.ParseUint(text, 10, 64) + if err != nil { + return 0, 0, 0, "", false + } + scale = 1 + for range len(text) { + scale *= 10 + } + rest = rest[fractionDigits:] + } + return whole, fraction, scale, rest, true +} + +func isDigitByte(b byte) bool { + return b >= '0' && b <= '9' +} + +// durationUnit is one unit of the duration grammar, with its whole-second +// and sub-second nanosecond weights. A month is 30.44 days and a year is +// 365.25 days, the same weights the Rust and Node tables carry. +type durationUnit struct { + name string + seconds uint64 + nanos uint64 +} + +// durationUnits are the humantime units, ordered longest spelling first so +// a prefix match cannot cut a longer unit short, which is also why the +// single-letter minute spelling m comes after month and millis. The table +// matches the one the Node and C# SDKs parse: it carries the hr, hrs, wk, +// wks and month spellings, the µs spelling beside us, and takes millis +// but not milliseconds. +var durationUnits = []durationUnit{ + {"seconds", 1, 0}, + {"minutes", 60, 0}, + {"second", 1, 0}, + {"minute", 60, 0}, + {"millis", 0, 1_000_000}, + {"months", 2_630_016, 0}, + {"month", 2_630_016, 0}, + {"nanos", 0, 1}, + {"hours", 3600, 0}, + {"weeks", 604_800, 0}, + {"years", 31_557_600, 0}, + {"nsec", 0, 1}, + {"usec", 0, 1_000}, + {"µs", 0, 1_000}, + {"msec", 0, 1_000_000}, + {"secs", 1, 0}, + {"mins", 60, 0}, + {"hour", 3600, 0}, + {"week", 604_800, 0}, + {"year", 31_557_600, 0}, + {"days", 86_400, 0}, + {"sec", 1, 0}, + {"min", 60, 0}, + {"hrs", 3600, 0}, + {"day", 86_400, 0}, + {"wks", 604_800, 0}, + {"yrs", 31_557_600, 0}, + {"ns", 0, 1}, + {"us", 0, 1_000}, + {"ms", 0, 1_000_000}, + {"hr", 3600, 0}, + {"wk", 604_800, 0}, + {"yr", 31_557_600, 0}, + {"s", 1, 0}, + {"m", 60, 0}, + {"h", 3600, 0}, + {"d", 86_400, 0}, + {"w", 604_800, 0}, + {"y", 31_557_600, 0}, +} + +// cutDurationUnit consumes the leading duration unit of s. The unit may +// run straight into the next amount, as in 1h30m; a leftover with no +// leading unit, such as the x of 5x, still fails the caller. +func cutDurationUnit(s string) (int, string, bool) { + for i, unit := range durationUnits { + if strings.HasPrefix(s, unit.name) { + return i, s[len(unit.name):], true + } + } + return 0, s, false +} diff --git a/foreign/go/client/connection_string_test.go b/foreign/go/client/connection_string_test.go new file mode 100644 index 0000000000..7a7d6b52c7 --- /dev/null +++ b/foreign/go/client/connection_string_test.go @@ -0,0 +1,385 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +package client + +import ( + "context" + "crypto/tls" + "net" + "testing" + "time" + + "github.com/apache/iggy/foreign/go/client/tcp" + "github.com/apache/iggy/foreign/go/internal/command" + . "github.com/apache/iggy/foreign/go/internal/fakevsr" + "github.com/apache/iggy/foreign/go/internal/vsr" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// applyOptions runs options the way NewIggyClient does, and returns the +// resulting TCP client so a test can observe what the parser produced. +func applyOptions(t *testing.T, options []tcp.Option) *tcp.IggyTcpClient { + t.Helper() + return tcp.NewIggyTcpClient(nil, options...) +} + +func TestParseConnectionString_ParsesTheDefaultSchemeWithPasswordCredentials(t *testing.T) { + opts, err := ParseConnectionString("iggy://iggy:secret@127.0.0.1:8090") + require.NoError(t, err) + client := applyOptions(t, opts) + + assert.Equal(t, "127.0.0.1:8090", client.GetConnectionInfo().ServerAddress) +} + +func TestParseConnectionString_ParsesTheExplicitTcpSchemeWithAPersonalAccessToken(t *testing.T) { + opts, err := ParseConnectionString("iggy+tcp://iggypat-1234567890abcdef@localhost:8090") + require.NoError(t, err) + client := applyOptions(t, opts) + + assert.Equal(t, "localhost:8090", client.GetConnectionInfo().ServerAddress) +} + +func TestParseConnectionString_ParsesBracketedIPv6Host(t *testing.T) { + opts, err := ParseConnectionString("iggy://iggy:secret@[::1]:8090") + require.NoError(t, err) + client := applyOptions(t, opts) + + assert.Equal(t, "[::1]:8090", client.GetConnectionInfo().ServerAddress) +} + +// TestParseConnectionString_NoDelayOptionLandsOnTheClient pins the nodelay +// rule: the spelling validates and, unlike a dropped validation, the +// parsed value reaches the client, so ?nodelay=false really turns Nagle +// back on, while a query that leaves nodelay out keeps the client +// default and stays with Nagle off. +func TestParseConnectionString_NoDelayOptionLandsOnTheClient(t *testing.T) { + withoutOption, err := ParseConnectionString("iggy://iggy:secret@localhost:8090") + require.NoError(t, err) + assert.True(t, applyOptions(t, withoutOption).GetNoDelay(), + "the default must keep Nagle disabled") + + unspecified, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?tls=true") + require.NoError(t, err) + assert.True(t, applyOptions(t, unspecified).GetNoDelay(), + "a query without nodelay must keep Nagle disabled") + + enabled, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?nodelay=true") + require.NoError(t, err) + assert.True(t, applyOptions(t, enabled).GetNoDelay()) + + disabled, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?nodelay=false") + require.NoError(t, err) + assert.False(t, applyOptions(t, disabled).GetNoDelay(), + "?nodelay=false must reach the dial instead of being dropped") +} + +// TestParseConnectionString_HeartbeatIntervalLandsOnTheClient pins the +// parsed heartbeat_interval reaching the client instead of leaving the +// five-second default, and the representation check that refuses an +// interval wider than the client own duration type. +func TestParseConnectionString_HeartbeatIntervalLandsOnTheClient(t *testing.T) { + withoutOption, err := ParseConnectionString("iggy://iggy:secret@localhost:8090") + require.NoError(t, err) + assert.Equal(t, 5*time.Second, applyOptions(t, withoutOption).GetHeartbeatInterval(), + "a query without heartbeat_interval must keep the default pacing") + + oneSecond, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?heartbeat_interval=1s") + require.NoError(t, err) + assert.Equal(t, time.Second, applyOptions(t, oneSecond).GetHeartbeatInterval(), + "?heartbeat_interval=1s must reach the client instead of being dropped") + + fractions, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?heartbeat_interval=250ms") + require.NoError(t, err) + assert.Equal(t, 250*time.Millisecond, applyOptions(t, fractions).GetHeartbeatInterval()) + + tooLarge, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?heartbeat_interval=1000y") + require.Error(t, err, + "an interval wider than time.Duration must fail the parse instead of wrapping") + assert.Empty(t, tooLarge) +} + +// TestParseConnectionString_NonZeroDurationOptionsRejectZeroSpellings +// pins the typing the Rust SDK gives each option: +// reconnection_interval and heartbeat_interval are non-zero durations +// whose wire format carries whole microseconds, so the zero spellings, +// zero-valued durations and anything below one microsecond are refused, +// while reestablish_after is a plain duration where they parse. +func TestParseConnectionString_NonZeroDurationOptionsRejectZeroSpellings(t *testing.T) { + for _, option := range []string{"reconnection_interval", "heartbeat_interval"} { + for _, value := range []string{"0", "none", "disabled", "unlimited", "0s", "0ms", "1ns", "999ns"} { + invalid := "iggy://iggy:secret@localhost:8090?" + option + "=" + value + t.Run(invalid, func(t *testing.T) { + opts, err := ParseConnectionString(invalid) + require.Error(t, err, + "%s is a non-zero duration in the Rust SDK", option) + assert.Empty(t, opts) + }) + } + } + for _, value := range []string{"0", "none", "disabled", "unlimited", "0s", "0ms"} { + valid := "iggy://iggy:secret@localhost:8090?reestablish_after=" + value + t.Run(valid, func(t *testing.T) { + opts, err := ParseConnectionString(valid) + require.NoError(t, err, + "reestablish_after is a plain duration in the Rust SDK") + assert.NotEmpty(t, opts) + }) + } + for _, option := range []string{"reconnection_interval", "heartbeat_interval"} { + for _, value := range []string{"1000ns", "1µs", "1us"} { + valid := "iggy://iggy:secret@localhost:8090?" + option + "=" + value + t.Run(valid, func(t *testing.T) { + opts, err := ParseConnectionString(valid) + require.NoError(t, err, + "%s must parse at the one-microsecond floor", option) + assert.NotEmpty(t, opts) + }) + } + } +} + +// TestParseConnectionString_AcceptsTheExtendedDurationUnits pins the +// spellings the Node and C# SDKs parse and the shared grammar carries: +// hr, hrs, wk, wks and month, plus the millis spelling they use instead +// of milliseconds. +func TestParseConnectionString_AcceptsTheExtendedDurationUnits(t *testing.T) { + for _, options := range []string{ + "reconnection_interval=1hr", + "reconnection_interval=2hrs", + "reconnection_interval=1wk", + "reconnection_interval=2wks", + "reconnection_interval=1month", + "reconnection_interval=2months", + "reconnection_interval=1hr 30m", + "reconnection_interval=250millis", + "heartbeat_interval=1hr", + "reestablish_after=1month", + } { + t.Run(options, func(t *testing.T) { + opts, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?" + options) + require.NoError(t, err) + assert.NotEmpty(t, opts) + }) + } + // The long spelling stays outside the grammar, as it is in the Node + // and C# SDKs. + opts, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?reconnection_interval=1milliseconds") + require.Error(t, err) + assert.Empty(t, opts) +} + +// TestParseConnectionString_RejectsUnsupportedTransports checks the scheme +// rule and pins the credential rule together: unsupported transports are +// rejected with an error naming the transport, and a malformed scheme +// never leaks the credentials embedded in it. +func TestParseConnectionString_RejectsUnsupportedTransportsAndNeverEchoesCredentials(t *testing.T) { + for _, value := range []string{ + "iggy+quic://iggy:secret@localhost:8090", + "iggy+ws://iggy:secret@localhost:8090", + "http://iggy:secret@localhost:8090", + } { + t.Run(value, func(t *testing.T) { + opts, err := ParseConnectionString(value) + require.Error(t, err) + assert.Empty(t, opts) + assert.NotContains(t, err.Error(), "secret", + "the transport error must not echo the connection string") + }) + } + // A scheme that embeds credentials after iggy+ must not surface them + // in the error, only the ASCII-letter transport name is echoed. + leaking := "iggy+user:secret@host:8090://host:8090" + t.Run(leaking, func(t *testing.T) { + opts, err := ParseConnectionString(leaking) + require.Error(t, err) + assert.Empty(t, opts) + assert.NotContains(t, err.Error(), "secret") + assert.NotContains(t, err.Error(), "user:secret@host") + }) +} + +// TestParseConnectionString_ConnectsAndLogsInAgainstAServer drives the +// full connect-and-login flow from the issue acceptance criteria: opts +// from ParseConnectionString, wired into NewIggyClient via WithTcp, +// connecting and signing in against a running (fake) server. +func TestParseConnectionString_ConnectsAndLogsInAgainstAServer(t *testing.T) { + server := Listen(t, nil) + address := server.Listener().Addr().String() + + opts, err := ParseConnectionString("iggy://iggy:iggy@" + address) + require.NoError(t, err) + cli, err := NewIggyClient(WithTcp(opts...)) + require.NoError(t, err) + t.Cleanup(func() { _ = cli.Close() }) + + require.NoError(t, cli.Connect(context.Background())) + + recorded := server.Recorded() + require.Len(t, recorded, 2) + // The roster read is auth-gated, so the sign-in comes first. + assert.Equal(t, vsr.OperationRegister, recorded[0].Operation()) + assert.Contains(t, string(recorded[0].Payload), "iggy") + assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[1].Code()) +} + +// TestParseConnectionString_TLSEnabledByQueryString drives the same flow +// with tls=true and a TLS-only fake server: a client that ignored the +// option would send its register frame in plaintext and the TLS handshake +// would fail, so a successful connect-and-login proves the option landed. +func TestParseConnectionString_TLSEnabledByQueryString(t *testing.T) { + certificate, caPath := SelfSignedCert(t) + + server := Listen(t, func(conn net.Conn) net.Conn { + return tls.Server(conn, &tls.Config{Certificates: []tls.Certificate{certificate}}) + }) + address := server.Listener().Addr().String() + host, port, err := net.SplitHostPort(address) + require.NoError(t, err) + + connectionString := "iggy://iggy:iggy@" + host + ":" + port + + "?tls=true&tls_domain=localhost&tls_ca_file=" + caPath + opts, err := ParseConnectionString(connectionString) + require.NoError(t, err) + cli, err := NewIggyClient(WithTcp(opts...)) + require.NoError(t, err) + t.Cleanup(func() { _ = cli.Close() }) + + require.NoError(t, cli.Connect(context.Background())) + + recorded := server.Recorded() + require.Len(t, recorded, 2) + assert.Equal(t, vsr.OperationRegister, recorded[0].Operation()) + assert.Equal(t, uint32(command.GetClusterMetadataCode), recorded[1].Code()) +} + +// TestParseConnectionString_RejectsMalformedConnectionStrings pins every +// malformed shape the Node SDK rejects, copied from its test table. The Rust +// SDK is laxer about some of these shapes, so the Node table is the shared +// floor. +func TestParseConnectionString_RejectsMalformedConnectionStrings(t *testing.T) { + for _, value := range []string{ + "", + "iggy", + "iggy://", + "iggy://:secret@localhost:8090", + "iggy://iggy:@localhost:8090", + "iggy://iggy:secret@localhost", + "iggy://iggy:secret@:8090", + "iggy://iggy:secret@localhost:port", + "iggy://iggy:secret@localhost:70000", + "iggy://iggy:secret@localhost:8090?unknown=value", + "iggy://iggy:secret@localhost:8090?tls=maybe", + "iggy://iggy:secret@localhost:8090?reconnection_retries=three", + "iggy://iggy:secret@localhost:8090?reconnection_retries=4294967296", + "iggy://iggy:secret@localhost:8090?reconnection_interval=5", + "iggy://iggy:secret@localhost:8090?reconnection_interval=garbage", + "iggy://iggy:secret@localhost:8090?reconnection_interval=-1s", + "iggy://iggy:secret@localhost:8090?nodelay=yes", + "iggy://iggy:secret@[::1:8090", + "iggy://iggy:secret@[]:8090", + "iggy://iggy:secret@[::1]x:8090", + "iggy://iggy:secret@2001:db8::1:8090", + "iggy://iggy:secret@host:8090:9090", + "iggy://iggy:secret@localhost:8090?", + "iggy://iggy:secret@localhost:8090?&", + "iggy://iggy:secret@localhost:8090?reestablish_after=garbage", + } { + t.Run(value, func(t *testing.T) { + opts, err := ParseConnectionString(value) + require.Error(t, err) + assert.Empty(t, opts) + }) + } +} + +// TestParseConnectionString_AcceptsPortZero pins the u16 port rule the +// other SDKs share: port 0 parses, because the port is only a u16 on the +// wire and the dial fails later instead of the parse. +func TestParseConnectionString_AcceptsPortZero(t *testing.T) { + opts, err := ParseConnectionString("iggy://iggy:secret@localhost:0") + require.NoError(t, err) + client := applyOptions(t, opts) + assert.Equal(t, "localhost:0", client.GetConnectionInfo().ServerAddress) +} + +// TestParseConnectionString_AcceptsFractionalAndMultiUnitDurations pins +// the duration grammar the other SDKs share: fractional seconds, +// sequences of units, a space between an amount and its unit, and the +// µs spelling, all accepted in connection-string options. +func TestParseConnectionString_AcceptsFractionalAndMultiUnitDurations(t *testing.T) { + for _, options := range []string{ + "reconnection_interval=500ms", + "reconnection_interval=0.5s", + "reconnection_interval=1.005s", + "reconnection_interval=2m", + "reconnection_interval=1h 1m 1s", + "reconnection_interval=1h30m", + "reconnection_interval=1µs", + "reestablish_after=1 s", + "reestablish_after=1µs", + "reconnection_interval=5sec", + "reconnection_interval=5msec", + "heartbeat_interval=10s", + "reestablish_after=10s", + } { + t.Run(options, func(t *testing.T) { + opts, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?" + options) + require.NoError(t, err) + assert.NotEmpty(t, opts) + }) + } +} + +// TestParseConnectionString_AcceptsUnlimitedRetries pins the u32 ceiling +// rule: unlimited is the keyword, and any u32 value parses. +func TestParseConnectionString_AcceptsUnlimitedRetries(t *testing.T) { + for _, options := range []string{ + "reconnection_retries=unlimited", + "reconnection_retries=3", + "reconnection_retries=4294967295", + } { + t.Run(options, func(t *testing.T) { + opts, err := ParseConnectionString("iggy://iggy:secret@localhost:8090?" + options) + require.NoError(t, err) + assert.NotEmpty(t, opts) + }) + } +} + +// TestParseConnectionString_ErrorsNeverLeakTheSecret checks that no error +// message contains the password or the token from the rejected string. +func TestParseConnectionString_ErrorsNeverLeakTheSecret(t *testing.T) { + secrets := []string{"hunter2", "iggypat-1234567890abcdef"} + for _, value := range []string{ + "iggy://iggy:hunter2@localhost", + "iggy+tcp://iggypat-1234567890abcdef@localhost", + "iggy://iggy:hunter2@localhost:8090?unknown=value", + "iggy://iggy:hunter2@localhost:8090?tls=maybe", + "iggy://iggy:hunter2@localhost:8090?reconnection_retries=three", + "iggy://iggy:hunter2@localhost:70000", + } { + opts, err := ParseConnectionString(value) + require.Error(t, err) + assert.Empty(t, opts) + for _, secret := range secrets { + assert.NotContains(t, err.Error(), secret, + "error message leaked a secret") + } + } +} diff --git a/foreign/go/client/tcp/tcp_connect_test.go b/foreign/go/client/tcp/tcp_connect_test.go index 739fbea1d3..71848a11ad 100644 --- a/foreign/go/client/tcp/tcp_connect_test.go +++ b/foreign/go/client/tcp/tcp_connect_test.go @@ -19,19 +19,10 @@ package tcp import ( "context" - "crypto/ecdsa" - "crypto/elliptic" - "crypto/rand" "crypto/tls" - "crypto/x509" - "crypto/x509/pkix" - "encoding/pem" "fmt" "log/slog" - "math/big" "net" - "os" - "path/filepath" "strings" "sync" "sync/atomic" @@ -41,43 +32,18 @@ import ( ierror "github.com/apache/iggy/foreign/go/errors" "github.com/apache/iggy/foreign/go/internal/command" + "github.com/apache/iggy/foreign/go/internal/fakevsr" "github.com/apache/iggy/foreign/go/internal/vsr" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) -// selfSignedCert mints a throwaway server certificate for the loopback -// listener, so the test carries no expiry or checkout-layout dependency. +// selfSignedCert mints the throwaway loopback certificate through the +// shared fakevsr helper, so the transport suite and the connection-string +// suite cannot drift apart. func selfSignedCert(t *testing.T) (tls.Certificate, string) { t.Helper() - - key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) - require.NoError(t, err) - template := x509.Certificate{ - SerialNumber: big.NewInt(1), - Subject: pkix.Name{CommonName: "localhost"}, - NotBefore: time.Now().Add(-time.Hour), - NotAfter: time.Now().Add(time.Hour), - KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, - ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, - BasicConstraintsValid: true, - IsCA: true, - DNSNames: []string{"localhost"}, - IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1)}, - } - certDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key) - require.NoError(t, err) - keyDER, err := x509.MarshalECPrivateKey(key) - require.NoError(t, err) - - certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER}) - keyPEM := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}) - certificate, err := tls.X509KeyPair(certPEM, keyPEM) - require.NoError(t, err) - - caPath := filepath.Join(t.TempDir(), "ca.pem") - require.NoError(t, os.WriteFile(caPath, certPEM, 0o600)) - return certificate, caPath + return fakevsr.SelfSignedCert(t) } func TestConnectBoundsSilentSignInWithoutCallerDeadline(t *testing.T) { diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go index 6b4070a6d3..66cf87456e 100644 --- a/foreign/go/client/tcp/tcp_core.go +++ b/foreign/go/client/tcp/tcp_core.go @@ -509,6 +509,24 @@ func (c *IggyTcpClient) GetConnectionInfo() *iggcon.ConnectionInfo { return &iggcon.ConnectionInfo{Protocol: iggcon.Tcp, ServerAddress: c.links.address()} } +// GetNoDelay reports whether TCP_NODELAY is set on new connections, which +// tests and callers can use to observe the parsed nodelay option before a +// connection exists to read it off. +func (c *IggyTcpClient) GetNoDelay() bool { + // The client configuration does not change after construction, so this + // reads the field directly, like the dial path does. + return c.config.noDelay +} + +// GetHeartbeatInterval reports the pacing of the managed heartbeat task on +// ready replica links, which tests and callers can use to observe the +// parsed heartbeat_interval option before a connection exists. +func (c *IggyTcpClient) GetHeartbeatInterval() time.Duration { + c.mtx.Lock() + defer c.mtx.Unlock() + return c.config.heartbeatInterval +} + // ResolveOperation replays the retained bytes under their original identity. // A committed business rejection is returned unchanged and releases the lane. // For an operation that is no longer retained, it returns what a replay proved. diff --git a/foreign/go/client/tcp/tcp_testing_test.go b/foreign/go/client/tcp/tcp_testing_test.go index d327bea0d0..2f3d3d3ee0 100644 --- a/foreign/go/client/tcp/tcp_testing_test.go +++ b/foreign/go/client/tcp/tcp_testing_test.go @@ -20,7 +20,6 @@ package tcp import ( "context" "encoding/binary" - "io" "log/slog" "net" "sync" @@ -30,6 +29,7 @@ import ( iggcon "github.com/apache/iggy/foreign/go/contracts" ierror "github.com/apache/iggy/foreign/go/errors" "github.com/apache/iggy/foreign/go/internal/command" + "github.com/apache/iggy/foreign/go/internal/fakevsr" "github.com/apache/iggy/foreign/go/internal/vsr" "github.com/stretchr/testify/require" ) @@ -166,34 +166,9 @@ func bindReplyFrame(userID uint32, session uint64) []byte { } // clusterMetadataFrame builds a cluster roster where the node at leaderIndex -// is the healthy leader. +// is the healthy leader, through the shared fakevsr builder. func clusterMetadataFrame(t *testing.T, leaderIndex int, addresses ...string) []byte { - t.Helper() - - metadata := iggcon.ClusterMetadata{Name: "test-cluster"} - for index, address := range addresses { - host, port, err := net.SplitHostPort(address) - require.NoError(t, err) - tcpPort, err := net.LookupPort("tcp", port) - require.NoError(t, err) - - role := iggcon.RoleFollower - if index == leaderIndex { - role = iggcon.RoleLeader - } - metadata.Nodes = append(metadata.Nodes, iggcon.ClusterNode{ - ReplicaID: uint8(index), - Name: host, - IP: host, - Role: role, - Status: iggcon.Healthy, - Endpoints: iggcon.TransportEndpoints{Tcp: uint16(tcpPort)}, - }) - } - - body, err := metadata.MarshalBinary() - require.NoError(t, err) - return replyFrame(vsr.OperationNonReplicated, body) + return fakevsr.ClusterMetadataFrame(t, leaderIndex, addresses...) } // echoReplyRequest stamps the request id into a reply frame the way the @@ -241,35 +216,25 @@ func versionInfoEnd(t *testing.T, payload []byte) int { return offset } -// readRequest reads one complete frame from conn. +// readRequest reads one complete frame from conn through the shared +// fakevsr reader, adapting the result to the request type the assertions +// use. func readRequest(conn net.Conn) (request, error) { - var read request - if _, err := io.ReadFull(conn, read.header[:]); err != nil { - return read, err - } - size := binary.LittleEndian.Uint32(read.header[frameOffsetSize:]) - if size > vsr.HeaderSize { - read.payload = make([]byte, int(size)-vsr.HeaderSize) - if _, err := io.ReadFull(conn, read.payload); err != nil { - return read, err - } - } - return read, nil + read, err := fakevsr.ReadRequest(conn) + return request{header: read.Header, payload: read.Payload}, err } // fakeServer answers frames from a handler and records what it was asked. type fakeServer struct { mtx sync.Mutex requests []request - done chan struct{} } // serve reads frames off conn until it closes, answering each with the frame // the handler returns. A nil answer leaves the request unanswered. func serve(conn net.Conn, handler func(index int, read request) []byte) *fakeServer { - server := &fakeServer{done: make(chan struct{})} + server := &fakeServer{} go func() { - defer close(server.done) for index := 0; ; index++ { read, err := readRequest(conn) if err != nil { diff --git a/foreign/go/internal/fakevsr/fakevsr.go b/foreign/go/internal/fakevsr/fakevsr.go new file mode 100644 index 0000000000..fd51447410 --- /dev/null +++ b/foreign/go/internal/fakevsr/fakevsr.go @@ -0,0 +1,294 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +// Package fakevsr provides the fake VSR server the connection-string +// tests drive a full connect-and-login flow against, without a real +// server, and the shared certificate helper the transport tests mint +// their throwaway TLS certificates with. It is an internal package, so +// it never reaches SDK users. +package fakevsr + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/binary" + "encoding/pem" + "io" + "math/big" + "net" + "os" + "path/filepath" + "sync" + "testing" + "time" + + iggcon "github.com/apache/iggy/foreign/go/contracts" + "github.com/apache/iggy/foreign/go/internal/command" + "github.com/apache/iggy/foreign/go/internal/vsr" + "github.com/stretchr/testify/require" +) + +// Frame offsets the fake server reads and writes. They mirror the +// unexported tables of the codec package, which no test can reach. +const ( + FrameOffsetSize = 48 + FrameOffsetCommand = 60 + FrameOffsetRequest = 168 + FrameOffsetOperation = 176 + FrameOffsetSession = 184 + FrameOffsetReserved = 196 + + ReplyFrameOffsetRequest = 200 + ReplyFrameOffsetOperation = 208 + ReplyFrameOffsetStatus = 216 +) + +// Request is one frame the fake server read off the wire. +type Request struct { + Header [vsr.HeaderSize]byte + Payload []byte +} + +// Operation reads the request operation code. +func (r Request) Operation() vsr.Operation { + return vsr.Operation(r.Header[FrameOffsetOperation]) +} + +// Code reads the request command code. +func (r Request) Code() uint32 { + return binary.LittleEndian.Uint32(r.Header[FrameOffsetReserved:]) +} + +// SessionID reads the logical session the request carries. +func (r Request) SessionID() uint64 { + return binary.LittleEndian.Uint64(r.Header[FrameOffsetSession:]) +} + +// ReadRequest reads one complete frame from conn. +func ReadRequest(conn net.Conn) (Request, error) { + var read Request + if _, err := io.ReadFull(conn, read.Header[:]); err != nil { + return read, err + } + size := binary.LittleEndian.Uint32(read.Header[FrameOffsetSize:]) + if size > vsr.HeaderSize { + read.Payload = make([]byte, int(size)-vsr.HeaderSize) + if _, err := io.ReadFull(conn, read.Payload); err != nil { + return read, err + } + } + return read, nil +} + +// ReplyFrame builds a committed reply carrying body. +func ReplyFrame(operation vsr.Operation, body []byte) []byte { + return StatusReplyFrame(operation, 0, body) +} + +// StatusReplyFrame builds a reply whose header carries a pre-commit denial. +func StatusReplyFrame(operation vsr.Operation, status uint32, body []byte) []byte { + frame := make([]byte, vsr.HeaderSize+len(body)) + binary.LittleEndian.PutUint32(frame[FrameOffsetSize:], uint32(len(frame))) + frame[FrameOffsetCommand] = 8 + binary.LittleEndian.PutUint32(frame[ReplyFrameOffsetStatus:], status) + frame[ReplyFrameOffsetOperation] = byte(operation) + copy(frame[vsr.HeaderSize:], body) + return frame +} + +// ResultSection builds the committed result prefix of a metadata reply. +func ResultSection(codes ...uint32) []byte { + section := binary.LittleEndian.AppendUint32(nil, uint32(len(codes))) + for index, code := range codes { + section = binary.LittleEndian.AppendUint32(section, uint32(index)) + section = binary.LittleEndian.AppendUint32(section, code) + } + return section +} + +// RegisterReplyFrame builds the reply of a successful sign-in. +func RegisterReplyFrame(userID uint32, session uint64) []byte { + body := binary.LittleEndian.AppendUint32(ResultSection(), userID) + body = binary.LittleEndian.AppendUint64(body, session) + body = binary.LittleEndian.AppendUint32(body, vsr.ProtocolVersion) + body = append(body, byte(len("0.11.0-edge.1"))) + body = append(body, "0.11.0-edge.1"...) + return ReplyFrame(vsr.OperationRegister, body) +} + +// BindReplyFrame builds the reply of a successful session bind. +func BindReplyFrame(userID uint32, session uint64) []byte { + registered := RegisterReplyFrame(userID, session) + return ReplyFrame(vsr.OperationNonReplicated, registered[vsr.HeaderSize+len(ResultSection()):]) +} + +// ClusterMetadataFrame builds a cluster roster where the node at +// leaderIndex is the healthy leader. +func ClusterMetadataFrame(t *testing.T, leaderIndex int, addresses ...string) []byte { + t.Helper() + + metadata := iggcon.ClusterMetadata{Name: "test-cluster"} + for index, address := range addresses { + host, port, err := net.SplitHostPort(address) + require.NoError(t, err) + tcpPort, err := net.LookupPort("tcp", port) + require.NoError(t, err) + + role := iggcon.RoleFollower + if index == leaderIndex { + role = iggcon.RoleLeader + } + metadata.Nodes = append(metadata.Nodes, iggcon.ClusterNode{ + ReplicaID: uint8(index), + Name: host, + IP: host, + Role: role, + Status: iggcon.Healthy, + Endpoints: iggcon.TransportEndpoints{Tcp: uint16(tcpPort)}, + }) + } + + body, err := metadata.MarshalBinary() + require.NoError(t, err) + return ReplyFrame(vsr.OperationNonReplicated, body) +} + +// EchoReplyRequest stamps the request id into a reply frame the way the +// server echoes it, so handlers do not repeat the correlation plumbing. An +// eviction frame passes through untouched. +func EchoReplyRequest(answer []byte, read Request) { + if len(answer) < vsr.HeaderSize || answer[FrameOffsetCommand] != 8 { + return + } + copy(answer[ReplyFrameOffsetRequest:ReplyFrameOffsetRequest+8], + read.Header[FrameOffsetRequest:FrameOffsetRequest+8]) +} + +// Server answers frames from a handler and records what it was asked. +type Server struct { + mtx sync.Mutex + requests []Request + listener net.Listener +} + +// Listen answers frames the way a single-node Iggy server does, so a test +// can drive a full connect-and-login flow without a real server: the +// bind-session and cluster-metadata requests get their canned replies, a +// sign-in gets a successful register, and anything else gets an empty +// non-replicated reply. The wrap hook lets a test put TLS in front of the +// listener. +func Listen(t *testing.T, wrap func(net.Conn) net.Conn) *Server { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + + server := &Server{listener: listener} + go func() { + for { + conn, err := listener.Accept() + if err != nil { + return + } + if wrap != nil { + conn = wrap(conn) + } + go func(conn net.Conn) { + defer func() { _ = conn.Close() }() + for { + read, err := ReadRequest(conn) + if err != nil { + return + } + server.mtx.Lock() + server.requests = append(server.requests, read) + server.mtx.Unlock() + + var answer []byte + switch { + case read.Code() == uint32(command.BindSessionCode): + answer = BindReplyFrame(7, read.SessionID()) + case read.Code() == uint32(command.GetClusterMetadataCode): + answer = ClusterMetadataFrame(t, 0, server.listener.Addr().String()) + case read.Operation() == vsr.OperationRegister: + answer = RegisterReplyFrame(7, 128) + default: + answer = ReplyFrame(vsr.OperationNonReplicated, nil) + } + EchoReplyRequest(answer, read) + if _, err := conn.Write(answer); err != nil { + return + } + } + }(conn) + } + }() + return server +} + +// Recorded returns the requests the server read off the wire. +func (s *Server) Recorded() []Request { + s.mtx.Lock() + defer s.mtx.Unlock() + return append([]Request(nil), s.requests...) +} + +// Listener returns the loopback listener, so a test can read the address +// the fake server bound to before the client connects. +func (s *Server) Listener() net.Listener { + return s.listener +} + +// SelfSignedCert mints a throwaway server certificate for the loopback +// listener, so a TLS test carries no expiry or checkout-layout dependency. +// The connection-string and TCP suites share it. +func SelfSignedCert(t *testing.T) (tls.Certificate, string) { + t.Helper() + + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + template := x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "localhost"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + BasicConstraintsValid: true, + IsCA: true, + DNSNames: []string{"localhost"}, + IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1)}, + } + certDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key) + require.NoError(t, err) + keyDER, err := x509.MarshalECPrivateKey(key) + require.NoError(t, err) + + certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER}) + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}) + certificate, err := tls.X509KeyPair(certPEM, keyPEM) + require.NoError(t, err) + + caPath := filepath.Join(t.TempDir(), "ca.pem") + require.NoError(t, os.WriteFile(caPath, certPEM, 0o600)) + return certificate, caPath +}