plugins: Additive updates to services when discovery enabled

Earlier with discovery enabled, there was no protection against accidental
changes to the discovery service. This change prevents the discovery service
from being modified by checking it's config in the service bundle.

Fixes #2058

Signed-off-by: Ashutosh Narkar <anarkar4387@gmail.com>

docs/content/management.md

@@ -944,6 +944,8 @@ configuration labels or environment variables.
 
 ### Limitations
 
-The discovery feature cannot be used to dynamically modify `services`, `labels`
-and `discovery`. This means that these configuration settings should be included
-in the bootup configuration file provided to OPA.
+In practice, discovery services do not change frequently. These configuration sections are treated as
+immutable to avoid accidental configuration errors rendering OPA unable to discover a new configuration.
+If the discovered configuration changes the `discovery` or `labels` sections,
+those changes are ignored. If the discovered configuration changes the discovery service,
+an error will be logged.

internal/config/config.go

@@ -0,0 +1,48 @@
+// Copyright 2020 The OPA Authors.  All rights reserved.
+// Use of this source code is governed by an Apache2
+// license that can be found in the LICENSE file.
+
+// Package config implements helper functions to parse OPA's configuration.
+package config
+
+import (
+	"encoding/json"
+
+	"github.com/open-policy-agent/opa/plugins/rest"
+	"github.com/open-policy-agent/opa/util"
+)
+
+// ParseServicesConfig returns a set of named service clients. The service
+// clients can be specified either as an array or as a map. Some systems (e.g.,
+// Helm) do not have proper support for configuration values nested under
+// arrays, so just support both here.
+func ParseServicesConfig(raw json.RawMessage) (map[string]rest.Client, error) {
+
+	services := map[string]rest.Client{}
+
+	var arr []json.RawMessage
+	var obj map[string]json.RawMessage
+
+	if err := util.Unmarshal(raw, &arr); err == nil {
+		for _, s := range arr {
+			client, err := rest.New(s)
+			if err != nil {
+				return nil, err
+			}
+			services[client.Service()] = client
+		}
+	} else if util.Unmarshal(raw, &obj) == nil {
+		for k := range obj {
+			client, err := rest.New(obj[k], rest.Name(k))
+			if err != nil {
+				return nil, err
+			}
+			services[client.Service()] = client
+		}
+	} else {
+		// Return error from array decode as that is the default format.
+		return nil, err
+	}
+
+	return services, nil
+}

plugins/discovery/discovery.go

@@ -19,6 +19,7 @@ import (
 	bundleApi "github.com/open-policy-agent/opa/bundle"
 	"github.com/open-policy-agent/opa/config"
 	"github.com/open-policy-agent/opa/download"
+	cfg "github.com/open-policy-agent/opa/internal/config"
 	"github.com/open-policy-agent/opa/plugins"
 	"github.com/open-policy-agent/opa/plugins/bundle"
 	"github.com/open-policy-agent/opa/plugins/logs"
@@ -174,20 +175,11 @@ func (c *Discovery) processUpdate(ctx context.Context, u download.Update) {
 
 func (c *Discovery) reconfigure(ctx context.Context, u download.Update) error {
 
-	config, ps, err := processBundle(ctx, c.manager, c.factories, u.Bundle, c.config.query, c.metrics)
+	ps, err := processBundle(ctx, c.manager, c.factories, u.Bundle, c.config.query, c.config.service, c.metrics)
 	if err != nil {
 		return err
 	}
 
-	if err := c.manager.Reconfigure(config); err != nil {
-		return err
-	}
-
-	// TODO(tsandall): we don't currently support changes to discovery
-	// configuration. These changes are risky because errors would be
-	// unrecoverable (without keeping track of changes and rolling back...)
-
-	// TODO(tsandall): add protection against discovery -service- changing.
 	for _, p := range ps.Start {
 		if err := p.Start(ctx); err != nil {
 			return err
@@ -220,15 +212,35 @@ func (c *Discovery) logrusFields() logrus.Fields {
 	}
 }
 
-func processBundle(ctx context.Context, manager *plugins.Manager, factories map[string]plugins.Factory, b *bundleApi.Bundle, query string, m metrics.Metrics) (*config.Config, *pluginSet, error) {
+func processBundle(ctx context.Context, manager *plugins.Manager, factories map[string]plugins.Factory, b *bundleApi.Bundle, query, service string, m metrics.Metrics) (*pluginSet, error) {
 
 	config, err := evaluateBundle(ctx, manager.ID, manager.Info, b, query)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
+	}
+
+	// Note: We don't currently support changes to the discovery
+	// configuration. These changes are risky because errors would be
+	// unrecoverable (without keeping track of changes and rolling back...)
+
+	// check for updates to the discovery service
+	services, err := cfg.ParseServicesConfig(config.Services)
+	if err != nil {
+		return nil, err
+	}
+
+	if client, ok := services[service]; ok {
+		dClient := manager.Client(service)
+		if !client.Config().Equal(dClient.Config()) {
+			return nil, fmt.Errorf("updates to the discovery service are not allowed")
+		}
+	}
+
+	if err := manager.Reconfigure(config); err != nil {
+		return nil, err
 	}
 
-	ps, err := getPluginSet(factories, manager, config, m)
-	return config, ps, err
+	return getPluginSet(factories, manager, config, m)
 }
 
 func evaluateBundle(ctx context.Context, id string, info *ast.Term, b *bundleApi.Bundle, query string) (*config.Config, error) {

plugins/discovery/discovery_test.go

@@ -120,7 +120,7 @@ func TestProcessBundle(t *testing.T) {
 		}
 	`)
 
-	_, ps, err := processBundle(ctx, manager, nil, initialBundle, "data.config", nil)
+	ps, err := processBundle(ctx, manager, nil, initialBundle, "data.config", "default", nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -139,7 +139,7 @@ func TestProcessBundle(t *testing.T) {
 		}
 	`)
 
-	_, ps, err = processBundle(ctx, manager, nil, updatedBundle, "data.config", nil)
+	ps, err = processBundle(ctx, manager, nil, updatedBundle, "data.config", "default", nil)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -156,7 +156,7 @@ func TestProcessBundle(t *testing.T) {
 		}
 	`)
 
-	_, _, err = processBundle(ctx, manager, nil, updatedBundle, "data.config", nil)
+	_, err = processBundle(ctx, manager, nil, updatedBundle, "data.config", "default", nil)
 	if err == nil {
 		t.Fatal("Expected error but got success")
 	}
@@ -274,6 +274,227 @@ func TestReconfigure(t *testing.T) {
 
 }
 
+func TestReconfigureWithUpdates(t *testing.T) {
+
+	ctx := context.Background()
+
+	manager, err := plugins.New([]byte(`{
+		"labels": {"x": "y"},
+		"services": {
+			"localhost": {
+				"url": "http://localhost:9999"
+			}
+		},
+		"discovery": {"name": "config"},
+	}`), "test-id", inmem.New())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	disco, err := New(manager)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	initialBundle := makeDataBundle(1, `
+		{
+			"config": {
+				"bundle": {"name": "test1"},
+				"status": {},
+				"decision_logs": {}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: initialBundle})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	originalConfig := disco.config
+	// update the discovery configuration and check
+	// the boot configuration is not overwritten
+	updatedBundle := makeDataBundle(2, `
+		{
+			"config": {
+				"discovery": {
+					"name": "config",
+					"decision": "/foo/bar"
+				}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: updatedBundle})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	if !reflect.DeepEqual(originalConfig, disco.config) {
+		t.Fatal("Discovery configuration updated")
+	}
+
+	// no update to the discovery configuration and check no error generated
+	updatedBundle = makeDataBundle(3, `
+		{
+			"config": {
+				"discovery": {
+					"name": "config"
+				}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: updatedBundle})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	if !reflect.DeepEqual(originalConfig, disco.config) {
+		t.Fatal("Discovery configuration updated")
+	}
+
+	// update the discovery service and check that error generated
+	updatedBundle = makeDataBundle(4, `
+		{
+			"config": {
+				"services": {
+					"localhost": {
+						"url": "http://localhost:9999",
+						"credentials": {"bearer": {"token": "blah"}}
+					}
+				}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: updatedBundle})
+	if err == nil {
+		t.Fatal("Expected error but got nil")
+	}
+
+	expectedErrMsg := "updates to the discovery service are not allowed"
+	if err.Error() != expectedErrMsg {
+		t.Fatalf("Expected error message: %v but got: %v", expectedErrMsg, err.Error())
+	}
+
+	// no update to the discovery service and check no error generated
+	updatedBundle = makeDataBundle(5, `
+		{
+			"config": {
+				"services": {
+					"localhost": {
+						"url": "http://localhost:9999"
+					}
+				}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: updatedBundle})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	// add a new service and a new bundle
+	updatedBundle = makeDataBundle(6, `
+		{
+			"config": {
+				"services": {
+					"acmecorp": {
+						"url": "http://localhost:8181"
+					}
+				},
+				"bundles": {
+					"authz": {
+						"service": "acmecorp"
+					}
+				}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: updatedBundle})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	if len(disco.manager.Services()) != 2 {
+		t.Fatalf("Expected two services but got %v\n", len(disco.manager.Services()))
+	}
+
+	bPlugin := bundle.Lookup(disco.manager)
+	config := bPlugin.Config()
+	expected := "acmecorp"
+	if config.Bundles["authz"].Service != expected {
+		t.Fatalf("Expected service %v for bundle authz but got %v", expected, config.Bundles["authz"].Service)
+	}
+
+	// update existing bundle's config and add a new bundle
+	updatedBundle = makeDataBundle(7, `
+		{
+			"config": {
+				"bundles": {
+					"authz": {
+						"service": "localhost",
+						"resource": "foo/bar"
+					},
+					"main": {
+						"resource": "baz/bar"
+					}
+				}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: updatedBundle})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+
+	bPlugin = bundle.Lookup(disco.manager)
+	config = bPlugin.Config()
+	expectedSvc := "localhost"
+	if config.Bundles["authz"].Service != expectedSvc {
+		t.Fatalf("Expected service %v for bundle authz but got %v", expectedSvc, config.Bundles["authz"].Service)
+	}
+
+	expectedRes := "foo/bar"
+	if config.Bundles["authz"].Resource != expectedRes {
+		t.Fatalf("Expected resource %v for bundle authz but got %v", expectedRes, config.Bundles["authz"].Resource)
+	}
+
+	expectedSvcs := map[string]bool{"localhost": true, "acmecorp": true}
+	if _, ok := expectedSvcs[config.Bundles["main"].Service]; !ok {
+		t.Fatalf("Expected service for bundle main to be one of [%v, %v] but got %v", "localhost", "acmecorp", config.Bundles["main"].Service)
+	}
+
+	// update existing (non-discovery)service's config
+	updatedBundle = makeDataBundle(8, `
+		{
+			"config": {
+				"services": {
+					"acmecorp": {
+						"url": "http://localhost:8181",
+						"credentials": {"bearer": {"token": "blah"}}
+						}
+				},
+				"bundles": {
+					"authz": {
+						"service": "localhost",
+						"resource": "foo/bar"
+					}
+				}
+			}
+		}
+	`)
+
+	err = disco.reconfigure(ctx, download.Update{Bundle: updatedBundle})
+	if err != nil {
+		t.Fatalf("Unexpected error %v", err)
+	}
+}
+
 type testServer struct {
 	t       *testing.T
 	mtx     sync.Mutex