Adds encryption by default.

This commit adds payload encryption to both types of RPC currently
supported by Pulse.

control/available_plugin.go

@@ -1,6 +1,7 @@
 package control
 
 import (
+	"crypto/rsa"
 	"errors"
 	"fmt"
 	"strconv"
@@ -67,7 +68,7 @@ type availablePlugin struct {
 
 // newAvailablePlugin returns an availablePlugin with information from a
 // plugin.Response
-func newAvailablePlugin(resp *plugin.Response, emitter gomit.Emitter, ep executablePlugin) (*availablePlugin, error) {
+func newAvailablePlugin(resp *plugin.Response, privKey *rsa.PrivateKey, emitter gomit.Emitter, ep executablePlugin) (*availablePlugin, error) {
 	if resp.Type != plugin.CollectorPluginType && resp.Type != plugin.ProcessorPluginType && resp.Type != plugin.PublisherPluginType {
 		return nil, ErrBadType
 	}
@@ -89,9 +90,13 @@ func newAvailablePlugin(resp *plugin.Response, emitter gomit.Emitter, ep executa
 	case plugin.CollectorPluginType:
 		switch resp.Meta.RPCType {
 		case plugin.JSONRPC:
-			ap.client = client.NewCollectorHttpJSONRPCClient(listenUrl, DefaultClientTimeout)
+			c, e := client.NewCollectorHttpJSONRPCClient(listenUrl, DefaultClientTimeout, resp.PublicKey, privKey)
+			if e != nil {
+				return nil, errors.New("error while creating client connection: " + e.Error())
+			}
+			ap.client = c
 		case plugin.NativeRPC:
-			c, e := client.NewCollectorNativeClient(resp.ListenAddress, DefaultClientTimeout)
+			c, e := client.NewCollectorNativeClient(resp.ListenAddress, DefaultClientTimeout, resp.PublicKey, privKey)
 			if e != nil {
 				return nil, errors.New("error while creating client connection: " + e.Error())
 			}
@@ -100,9 +105,13 @@ func newAvailablePlugin(resp *plugin.Response, emitter gomit.Emitter, ep executa
 	case plugin.PublisherPluginType:
 		switch resp.Meta.RPCType {
 		case plugin.JSONRPC:
-			ap.client = client.NewPublisherHttpJSONRPCClient(listenUrl, DefaultClientTimeout)
+			c, e := client.NewPublisherHttpJSONRPCClient(listenUrl, DefaultClientTimeout, resp.PublicKey, privKey)
+			if e != nil {
+				return nil, errors.New("error while creating client connection: " + e.Error())
+			}
+			ap.client = c
 		case plugin.NativeRPC:
-			c, e := client.NewPublisherNativeClient(resp.ListenAddress, DefaultClientTimeout)
+			c, e := client.NewPublisherNativeClient(resp.ListenAddress, DefaultClientTimeout, resp.PublicKey, privKey)
 			if e != nil {
 				return nil, errors.New("error while creating client connection: " + e.Error())
 			}
@@ -111,9 +120,13 @@ func newAvailablePlugin(resp *plugin.Response, emitter gomit.Emitter, ep executa
 	case plugin.ProcessorPluginType:
 		switch resp.Meta.RPCType {
 		case plugin.JSONRPC:
-			ap.client = client.NewProcessorHttpJSONRPCClient(listenUrl, DefaultClientTimeout)
+			c, e := client.NewProcessorHttpJSONRPCClient(listenUrl, DefaultClientTimeout, resp.PublicKey, privKey)
+			if e != nil {
+				return nil, errors.New("error while creating client connection: " + e.Error())
+			}
+			ap.client = c
 		case plugin.NativeRPC:
-			c, e := client.NewProcessorNativeClient(resp.ListenAddress, DefaultClientTimeout)
+			c, e := client.NewProcessorNativeClient(resp.ListenAddress, DefaultClientTimeout, resp.PublicKey, privKey)
 			if e != nil {
 				return nil, errors.New("error while creating client connection: " + e.Error())
 			}

control/available_plugin_test.go

@@ -1,6 +1,8 @@
 package control
 
 import (
+	"crypto/rand"
+	"crypto/rsa"
 	"errors"
 	"net"
 	"testing"
@@ -12,6 +14,8 @@ import (
 
 func TestAvailablePlugin(t *testing.T) {
 	Convey("newAvailablePlugin()", t, func() {
+		key, err := rsa.GenerateKey(rand.Reader, 2048)
+		So(err, ShouldBeNil)
 		Convey("returns an availablePlugin", func() {
 			ln, _ := net.Listen("tcp", ":4000")
 			defer ln.Close()
@@ -23,15 +27,17 @@ func TestAvailablePlugin(t *testing.T) {
 				Type:          plugin.CollectorPluginType,
 				ListenAddress: "127.0.0.1:4000",
 			}
-			ap, err := newAvailablePlugin(resp, nil, nil)
+			ap, err := newAvailablePlugin(resp, key, nil, nil)
 			So(ap, ShouldHaveSameTypeAs, new(availablePlugin))
 			So(err, ShouldBeNil)
 		})
 	})
 
 	Convey("Stop()", t, func() {
 		Convey("returns nil if plugin successfully stopped", func() {
-			r := newRunner(&routing.RoundRobinStrategy{})
+			key, err := rsa.GenerateKey(rand.Reader, 2048)
+			So(err, ShouldBeNil)
+			r := newRunner(&routing.RoundRobinStrategy{}, key)
 			a := plugin.Arg{
 				PluginLogPath: "/tmp/pulse-test-plugin-stop.log",
 			}
@@ -83,6 +89,8 @@ func TestAvailablePlugins(t *testing.T) {
 		})
 	})
 	Convey("it returns an error if client cannot be created", t, func() {
+		key, err := rsa.GenerateKey(rand.Reader, 2048)
+		So(err, ShouldBeNil)
 		resp := &plugin.Response{
 			Meta: plugin.PluginMeta{
 				Name:    "test",
@@ -91,7 +99,7 @@ func TestAvailablePlugins(t *testing.T) {
 			Type:          plugin.CollectorPluginType,
 			ListenAddress: "localhost:",
 		}
-		ap, err := newAvailablePlugin(resp, nil, nil)
+		ap, err := newAvailablePlugin(resp, key, nil, nil)
 		So(ap, ShouldBeNil)
 		So(err, ShouldNotBeNil)
 	})

control/control.go

@@ -1,6 +1,7 @@
 package control
 
 import (
+	"crypto/rand"
 	"crypto/rsa"
 	"errors"
 	"fmt"
@@ -46,8 +47,8 @@ type pluginControl struct {
 	Started        bool
 
 	autodiscoverPaths []string
-	controlPrivKey    *rsa.PrivateKey
-	controlPubKey     *rsa.PublicKey
+	privKey           *rsa.PrivateKey
+	pubKey            *rsa.PublicKey
 	eventManager      *gomit.EventController
 
 	pluginManager  managesPlugins
@@ -116,7 +117,14 @@ func CacheExpiration(t time.Duration) controlOpt {
 // New returns a new pluginControl instance
 func New(opts ...controlOpt) *pluginControl {
 
-	c := &pluginControl{}
+	key, err := rsa.GenerateKey(rand.Reader, 2048)
+	if err != nil {
+		panic(err)
+	}
+	c := &pluginControl{
+		pubKey:  &key.PublicKey,
+		privKey: key,
+	}
 	// Initialize components
 	//
 	// Event Manager
@@ -133,7 +141,7 @@ func New(opts ...controlOpt) *pluginControl {
 	}).Debug("metric catalog created")
 
 	// Plugin Manager
-	c.pluginManager = newPluginManager()
+	c.pluginManager = newPluginManager(c.pubKey, c.privKey)
 	controlLogger.WithFields(log.Fields{
 		"_block": "new",
 	}).Debug("plugin manager created")
@@ -148,7 +156,7 @@ func New(opts ...controlOpt) *pluginControl {
 
 	// Plugin Runner
 	// TODO (danielscottt): handle routing strat changes via events
-	c.pluginRunner = newRunner(&routing.RoundRobinStrategy{})
+	c.pluginRunner = newRunner(&routing.RoundRobinStrategy{}, c.privKey)
 	controlLogger.WithFields(log.Fields{
 		"_block": "new",
 	}).Debug("runner created")
@@ -161,7 +169,7 @@ func New(opts ...controlOpt) *pluginControl {
 	// Wire event manager
 
 	// Start stuff
-	err := c.pluginRunner.Start()
+	err = c.pluginRunner.Start()
 	if err != nil {
 		panic(err)
 	}

control/control_test.go

@@ -2,6 +2,8 @@ package control
 
 import (
 	"bytes"
+	"crypto/rand"
+	"crypto/rsa"
 	"encoding/gob"
 	"encoding/json"
 	"errors"
@@ -361,14 +363,15 @@ func TestStop(t *testing.T) {
 }
 
 func TestPluginCatalog(t *testing.T) {
+	key, _ := rsa.GenerateKey(rand.Reader, 2048)
 	ts := time.Now()
 
 	c := New()
 
 	// We need our own plugin manager to drop mock
 	// loaded plugins into.  Aribitrarily adding
 	// plugins from the pm is no longer supported.
-	tpm := newPluginManager()
+	tpm := newPluginManager(&key.PublicKey, key)
 	c.pluginManager = tpm
 
 	lp1 := new(loadedPlugin)

control/monitor_test.go

@@ -22,6 +22,10 @@ func (mp *mockPluginClient) Kill(r string) error {
 	return nil
 }
 
+func (mp *mockPluginClient) GetConfigPolicy() error {
+	return nil
+}
+
 func TestMonitor(t *testing.T) {
 	Convey("monitor", t, func() {
 		aps := newAvailablePlugins(&routing.RoundRobinStrategy{})

control/plugin/client/client.go

@@ -8,28 +8,27 @@ import (
 
 // PluginClient A client providing common plugin method calls.
 type PluginClient interface {
+	SetKey() error
 	Ping() error
 	Kill(string) error
+	GetConfigPolicy() (*cpolicy.ConfigPolicy, error)
 }
 
 // PluginCollectorClient A client providing collector specific plugin method calls.
 type PluginCollectorClient interface {
 	PluginClient
 	CollectMetrics([]core.Metric) ([]core.Metric, error)
 	GetMetricTypes() ([]core.Metric, error)
-	GetConfigPolicy() (cpolicy.ConfigPolicy, error)
 }
 
 // PluginProcessorClient A client providing processor specific plugin method calls.
 type PluginProcessorClient interface {
 	PluginClient
 	Process(contentType string, content []byte, config map[string]ctypes.ConfigValue) (string, []byte, error)
-	GetConfigPolicy() (cpolicy.ConfigPolicy, error)
 }
 
 // PluginPublisherClient A client providing publishing specific plugin method calls.
 type PluginPublisherClient interface {
 	PluginClient
 	Publish(contentType string, content []byte, config map[string]ctypes.ConfigValue) error
-	GetConfigPolicy() (cpolicy.ConfigPolicy, error)
 }