feat: Reject unknown fields in YAML config, allow fluent output without secret (#58)

* accessing internal yaml v3 methods, see golang/go#67361
* added tests for client worker's connection opening

since the YAML change, all practically optional fields in config can be turned optional by removing the value checks.

Author: jiping-s

.gitignore

@@ -1,3 +1,4 @@
+.DS_Store
 *.test
 /.vscode
 /BUILD/*

base/bconfig/configholder.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/relex/gotils/logger"
 	"github.com/relex/slog-agent/util"
+	"github.com/relex/slog-agent/util/yamlinternal"
 	"gopkg.in/yaml.v3"
 )
 
@@ -45,7 +46,7 @@ func (holder *ConfigHolder[C]) UnmarshalYAML(value *yaml.Node) error {
 	}
 	holder.Value = createFunc()
 
-	if err := value.Decode(holder.Value); err != nil {
+	if err := yamlinternal.NodeDecodeKnownFields(value, holder.Value); err != nil {
 		return util.NewYamlError(value, err.Error())
 	}
 	holder.Location = util.GetYamlLocation(value)

output/fluentdforward/clientworker.go

@@ -50,18 +50,25 @@ func openForwardConnection(parentLogger logger.Logger, config UpstreamConfig) (b
 	}
 	connLogger.Info("connected to ", sock.RemoteAddr())
 
-	success, reason, herr := forwardprotocol.DoClientHandshake(sock, config.Secret, defs.ForwarderHandshakeTimeout)
-	if herr != nil {
-		if err := sock.Close(); err != nil && !util.IsNetworkClosed(err) {
-			connLogger.Warn("error closing connection: ", err)
+	if len(config.Secret) > 0 {
+		success, reason, protocolErr := forwardprotocol.DoClientHandshake(sock, config.Secret, defs.ForwarderHandshakeTimeout)
+
+		var authErr error
+		switch {
+		case protocolErr != nil:
+			authErr = fmt.Errorf("failed to handshake due to unexpected error: %w", protocolErr)
+		case !success:
+			authErr = fmt.Errorf("login rejected: %s", reason)
+		default:
+			authErr = nil
 		}
-		return nil, fmt.Errorf("failed to handshake due to error: %w", herr)
-	}
-	if !success {
-		if err := sock.Close(); err != nil && !util.IsNetworkClosed(err) {
-			connLogger.Warn("error closing connection: ", err)
+
+		if authErr != nil {
+			if err := sock.Close(); err != nil && !util.IsNetworkClosed(err) {
+				connLogger.Warn("error closing connection: ", err)
+			}
+			return nil, authErr
 		}
-		return nil, fmt.Errorf("failed to handshake due to authentication: %s", reason)
 	}
 
 	return &forwardConnection{

output/fluentdforward/clientworker_test.go

@@ -0,0 +1,47 @@
+package fluentdforward
+
+import (
+	"os"
+	"testing"
+
+	"github.com/relex/fluentlib/server"
+	"github.com/relex/fluentlib/server/receivers"
+	"github.com/relex/gotils/logger"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOpenClientConnection(t *testing.T) {
+	recv := receivers.NewMessageWriter(os.Stdout)
+
+	t.Run("connect fails when protocols are different", func(t *testing.T) {
+		srvCfg := server.Config{
+			Address: "localhost:0",
+			TLS:     false,
+		}
+		srv, srvAddr := server.LaunchServer(logger.WithField("test", t.Name()), srvCfg, recv)
+		defer srv.Shutdown()
+
+		_, err := openForwardConnection(logger.Root(), UpstreamConfig{
+			Address: srvAddr.String(),
+			TLS:     true, // attempt to request TLS handshake
+		})
+		assert.ErrorContains(t, err, "failed to connect: tls:")
+	})
+
+	t.Run("login fails when secrets are different", func(t *testing.T) {
+		srvCfg := server.Config{
+			Address: "localhost:0",
+			Secret:  "real pass",
+			TLS:     false,
+		}
+		srv, srvAddr := server.LaunchServer(logger.WithField("test", t.Name()), srvCfg, recv)
+		defer srv.Shutdown()
+
+		_, err := openForwardConnection(logger.Root(), UpstreamConfig{
+			Address: srvAddr.String(),
+			TLS:     false,
+			Secret:  "wrong pass",
+		})
+		assert.ErrorContains(t, err, "login rejected:")
+	})
+}

output/fluentdforward/config.go

@@ -129,10 +129,6 @@ func (cfg *Config) VerifyConfig(schema base.LogSchema) error {
 		return fmt.Errorf(".upstream.address is invalid: %w", err)
 	}
 
-	if cfg.Upstream.TLS && len(cfg.Upstream.Secret) == 0 {
-		return fmt.Errorf(".upstream.secret is unspecified when tls=true")
-	}
-
 	if cfg.Upstream.MaxDuration == 0 {
 		return fmt.Errorf(".upstream.maxDuration is unspecified")
 	}

run/loader_test.go

@@ -77,6 +77,7 @@ outputBufferPairs:
         upstream:
             address: %s
             tls: false
+            secret: Hi
             maxDuration: 500ms
 `
 
@@ -88,6 +89,11 @@ var sampleConf = assembleConfig(
 	sampleOutputConf,
 )
 
+var serverConfig = server.Config{
+	Address: "localhost:0",
+	Secret:  "Hi",
+}
+
 func TestLoader(t *testing.T) {
 	logRecv, outBatchCh := receivers.NewMessageCollector(5 * time.Second)
 
@@ -147,9 +153,7 @@ func runTestEnv(t *testing.T, logReceiver receivers.Receiver, confYML string,
 	assert.NoError(t, confFileErr)
 	defer os.Remove(confFile.Name())
 
-	srvConf := server.Config{}
-	srvConf.Address = "localhost:0"
-	srv, srvAddr := server.LaunchServer(logger.WithField("test", t.Name()), srvConf, logReceiver)
+	srv, srvAddr := server.LaunchServer(logger.WithField("test", t.Name()), serverConfig, logReceiver)
 
 	_, writeErr := confFile.WriteString(fmt.Sprintf(confYML, bufDir, srvAddr.String()))
 	assert.NoError(t, writeErr)

util/yaml.go

@@ -56,7 +56,7 @@ func UnmarshalYamlFile(path string, output interface{}) error {
 // UnmarshalYamlReader loads and unmarshals YAML from IO reader to interface or pointer to struct
 func UnmarshalYamlReader(reader io.Reader, output interface{}) error {
 	decoder := yaml.NewDecoder(reader)
-	decoder.KnownFields(true)
+	decoder.KnownFields(true) // only works outside of custom unmarshalers
 	return decoder.Decode(output)
 }
 

util/yamlinternal/decode.go

@@ -0,0 +1,77 @@
+// Copyright (c) 2024 RELEX Oy
+// Copyright (c) 2011-2019 Canonical Ltd
+//
+// 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.
+
+package yamlinternal
+
+import (
+	"fmt"
+	"reflect"
+
+	_ "unsafe"
+
+	"gopkg.in/yaml.v3"
+)
+
+// decoder is a copy from gopkg.in/yaml.v3/decode.go
+//
+// FIXME: Can we get the actual type definition of the returned valued from newDecoder()? Attempts resulted in "<invalid reflect.Value>"
+//
+//lint:ignore U1000 all fields until the last used field must be kept for proper offsets
+type decoder struct {
+	doc     *yaml.Node
+	aliases map[*yaml.Node]bool
+	terrors []string
+
+	stringMapType  reflect.Type
+	generalMapType reflect.Type
+
+	knownFields bool
+	uniqueKeys  bool
+	decodeCount int
+	aliasCount  int
+	aliasDepth  int
+
+	mergedFields map[interface{}]bool
+}
+
+//go:linkname handleErr gopkg.in/yaml%2ev3.handleErr
+func handleErr(err *error)
+
+//go:linkname newDecoder gopkg.in/yaml%2ev3.newDecoder
+func newDecoder() *decoder
+
+//go:linkname unmarshal gopkg.in/yaml%2ev3.(*decoder).unmarshal
+func unmarshal(d *decoder, n *yaml.Node, out reflect.Value) (good bool)
+
+// NodeDecodeKnownFields is yaml.Node.Decode with KnownFields=true, to disallow unknown fields in YAML source.
+func NodeDecodeKnownFields(node *yaml.Node, v interface{}) (err error) {
+	d := newDecoder()
+	d.knownFields = true
+	defer handleErr(&err)
+	out := reflect.ValueOf(v)
+	if out.Kind() == reflect.Ptr && !out.IsNil() {
+		out = out.Elem()
+	}
+	good := unmarshal(d, node, out)
+
+	if len(d.terrors) > 0 {
+		return &yaml.TypeError{Errors: d.terrors}
+	}
+
+	if !good {
+		return fmt.Errorf("not good")
+	}
+	return nil
+}

util/yamlinternal/decode_test.go

@@ -0,0 +1,62 @@
+package yamlinternal
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"gopkg.in/yaml.v3"
+)
+
+func TestNodeDecodeKnownFields(t *testing.T) {
+	var correctNode yaml.Node
+	correctYaml := `
+value: ok
+
+list:
+  - "Hi here"
+  - "Hey correct"
+`
+
+	var incorrectNode yaml.Node
+	incorrectYaml := `
+value: ok
+
+list:
+  - "Hi here"
+  - "Hey incorrect"
+
+unknown: 9
+`
+	assert.NoError(t, yaml.Unmarshal([]byte(correctYaml), &correctNode))
+	assert.NoError(t, yaml.Unmarshal([]byte(incorrectYaml), &incorrectNode))
+
+	type ResultType struct {
+		Value string   `yaml:"value"`
+		List  []string `yaml:"list"`
+	}
+
+	t.Run("decode with builtin method to accept unknown fields", func(t *testing.T) {
+		var result ResultType
+		assert.NoError(t, incorrectNode.Decode(&result))
+		assert.Equal(t, "ok", result.Value)
+		if assert.Equal(t, 2, len(result.List)) {
+			assert.Equal(t, "Hi here", result.List[0])
+			assert.Equal(t, "Hey incorrect", result.List[1])
+		}
+	})
+
+	t.Run("decode with new method to reject unknown fields", func(t *testing.T) {
+		var result ResultType
+		assert.ErrorContains(t, NodeDecodeKnownFields(&incorrectNode, &result), "line 8: field unknown not found in type yamlinternal.ResultType")
+	})
+
+	t.Run("decode with new method to accept known fields", func(t *testing.T) {
+		var result ResultType
+		assert.NoError(t, NodeDecodeKnownFields(&correctNode, &result))
+		assert.Equal(t, "ok", result.Value)
+		if assert.Equal(t, 2, len(result.List)) {
+			assert.Equal(t, "Hi here", result.List[0])
+			assert.Equal(t, "Hey correct", result.List[1])
+		}
+	})
+}