status-im
/
consul
mirror of https://github.com/status-im/consul.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

consul: refactor keyring, repeat RPC calls to all DC's

This commit is contained in:
Ryan Uber 2014-09-24 16:39:14 -07:00
parent cfbf2b4f94
commit 8dec2744da
7 changed files with 354 additions and 226 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

101
command/agent/keyring.go
View File

@ -7,6 +7,7 @@ import (
"io/ioutil"
"os"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/memberlist"
"github.com/hashicorp/serf/serf"
)
@ -61,78 +62,48 @@ func loadKeyringFile(c *serf.Config) error {
return nil
}
// ListKeysLAN returns the keys installed on the LAN gossip pool
func (a *Agent) ListKeysLAN() (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerLAN()
return km.ListKeys()
// keyringProcess is used to abstract away the semantic similarities in
// performing various operations on the encryption keyring.
func (a *Agent) keyringProcess(
method string,
args *structs.KeyringRequest) (*structs.KeyringResponse, error) {
var reply structs.KeyringResponse
if a.server == nil {
return nil, fmt.Errorf("keyring operations must run against a server node")
}
km := a.client.KeyManagerLAN()
return km.ListKeys()
if err := a.RPC(method, args, &reply); err != nil {
return &reply, err
}
return &reply, nil
}
// ListKeysWAN returns the keys installed on the WAN gossip pool
func (a *Agent) ListKeysWAN() (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerWAN()
return km.ListKeys()
}
return nil, fmt.Errorf("WAN keyring not available on client node")
// ListKeys lists out all keys installed on the collective Consul cluster. This
// includes both servers and clients in all DC's.
func (a *Agent) ListKeys() (*structs.KeyringResponse, error) {
args := structs.KeyringRequest{}
args.AllowStale = true
return a.keyringProcess("Internal.ListKeys", &args)
}
// InstallKeyWAN installs a new WAN gossip encryption key on server nodes
func (a *Agent) InstallKeyWAN(key string) (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerWAN()
return km.InstallKey(key)
}
return nil, fmt.Errorf("WAN keyring not available on client node")
// InstallKey installs a new gossip encryption key
func (a *Agent) InstallKey(key string) (*structs.KeyringResponse, error) {
args := structs.KeyringRequest{Key: key}
args.AllowStale = true
return a.keyringProcess("Internal.InstallKey", &args)
}
// InstallKeyLAN installs a new LAN gossip encryption key on all nodes
func (a *Agent) InstallKeyLAN(key string) (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerLAN()
return km.InstallKey(key)
}
km := a.client.KeyManagerLAN()
return km.InstallKey(key)
// UseKey changes the primary encryption key used to encrypt messages
func (a *Agent) UseKey(key string) (*structs.KeyringResponse, error) {
args := structs.KeyringRequest{Key: key}
args.AllowStale = true
return a.keyringProcess("Internal.UseKey", &args)
}
// UseKeyWAN changes the primary WAN gossip encryption key on server nodes
func (a *Agent) UseKeyWAN(key string) (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerWAN()
return km.UseKey(key)
}
return nil, fmt.Errorf("WAN keyring not available on client node")
}
// UseKeyLAN changes the primary LAN gossip encryption key on all nodes
func (a *Agent) UseKeyLAN(key string) (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerLAN()
return km.UseKey(key)
}
km := a.client.KeyManagerLAN()
return km.UseKey(key)
}
// RemoveKeyWAN removes a WAN gossip encryption key on server nodes
func (a *Agent) RemoveKeyWAN(key string) (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerWAN()
return km.RemoveKey(key)
}
return nil, fmt.Errorf("WAN keyring not available on client node")
}
// RemoveKeyLAN removes a LAN gossip encryption key on all nodes
func (a *Agent) RemoveKeyLAN(key string) (*serf.KeyResponse, error) {
if a.server != nil {
km := a.server.KeyManagerLAN()
return km.RemoveKey(key)
}
km := a.client.KeyManagerLAN()
return km.RemoveKey(key)
// RemoveKey will remove a gossip encryption key from the keyring
func (a *Agent) RemoveKey(key string) (*structs.KeyringResponse, error) {
args := structs.KeyringRequest{Key: key}
args.AllowStale = true
return a.keyringProcess("Internal.RemoveKey", &args)
}

94
command/agent/rpc.go
View File

@ -24,15 +24,17 @@ package agent
import (
"bufio"
"fmt"
"github.com/hashicorp/go-msgpack/codec"
"github.com/hashicorp/logutils"
"github.com/hashicorp/serf/serf"
"io"
"log"
"net"
"os"
"strings"
"sync"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/go-msgpack/codec"
"github.com/hashicorp/logutils"
"github.com/hashicorp/serf/serf"
)
const (
@ -51,14 +53,10 @@ const (
leaveCommand = "leave"
statsCommand = "stats"
reloadCommand = "reload"
listKeysLANCommand = "list-keys-lan"
listKeysWANCommand = "list-keys-wan"
installKeyLANCommand = "install-key-lan"
installKeyWANCommand = "install-key-wan"
useKeyLANCommand = "use-key-lan"
useKeyWANCommand = "use-key-wan"
removeKeyLANCommand = "remove-key-lan"
removeKeyWANCommand = "remove-key-wan"
installKeyCommand = "install-key"
useKeyCommand = "use-key"
removeKeyCommand = "remove-key"
listKeysCommand = "list-keys"
)
const (
@ -117,10 +115,10 @@ type keyRequest struct {
type keyResponse struct {
Messages map[string]string
Keys map[string]int
NumNodes int
NumResp int
NumErr int
Keys map[string]int
}
type membersResponse struct {
@ -393,17 +391,8 @@ func (i *AgentRPC) handleRequest(client *rpcClient, reqHeader *requestHeader) er
case reloadCommand:
return i.handleReload(client, seq)
case listKeysLANCommand, listKeysWANCommand:
return i.handleListKeys(client, seq, command)
case installKeyLANCommand, installKeyWANCommand:
return i.handleGossipKeyChange(client, seq, command)
case useKeyLANCommand, useKeyWANCommand:
return i.handleGossipKeyChange(client, seq, command)
case removeKeyLANCommand, removeKeyWANCommand:
return i.handleGossipKeyChange(client, seq, command)
case installKeyCommand, useKeyCommand, removeKeyCommand, listKeysCommand:
return i.handleKeyring(client, seq, command)
default:
respHeader := responseHeader{Seq: seq, Error: unsupportedCommand}
@ -615,56 +604,27 @@ func (i *AgentRPC) handleReload(client *rpcClient, seq uint64) error {
return client.Send(&resp, nil)
}
func (i *AgentRPC) handleListKeys(client *rpcClient, seq uint64, cmd string) error {
var queryResp *serf.KeyResponse
var err error
switch cmd {
case listKeysWANCommand:
queryResp, err = i.agent.ListKeysWAN()
default:
queryResp, err = i.agent.ListKeysLAN()
}
header := responseHeader{
Seq: seq,
Error: errToString(err),
}
resp := keyResponse{
Messages: queryResp.Messages,
Keys: queryResp.Keys,
NumResp: queryResp.NumResp,
NumErr: queryResp.NumErr,
NumNodes: queryResp.NumNodes,
}
return client.Send(&header, &resp)
}
func (i *AgentRPC) handleGossipKeyChange(client *rpcClient, seq uint64, cmd string) error {
func (i *AgentRPC) handleKeyring(client *rpcClient, seq uint64, cmd string) error {
var req keyRequest
var queryResp *structs.KeyringResponse
var resp keyResponse
var queryResp *serf.KeyResponse
var err error
if cmd != listKeysCommand {
if err = client.dec.Decode(&req); err != nil {
return fmt.Errorf("decode failed: %v", err)
}
}
switch cmd {
case installKeyWANCommand:
queryResp, err = i.agent.InstallKeyWAN(req.Key)
case installKeyLANCommand:
queryResp, err = i.agent.InstallKeyLAN(req.Key)
case useKeyWANCommand:
queryResp, err = i.agent.UseKeyWAN(req.Key)
case useKeyLANCommand:
queryResp, err = i.agent.UseKeyLAN(req.Key)
case removeKeyWANCommand:
queryResp, err = i.agent.RemoveKeyWAN(req.Key)
case removeKeyLANCommand:
queryResp, err = i.agent.RemoveKeyLAN(req.Key)
case listKeysCommand:
queryResp, err = i.agent.ListKeys()
case installKeyCommand:
queryResp, err = i.agent.InstallKey(req.Key)
case useKeyCommand:
queryResp, err = i.agent.UseKey(req.Key)
case removeKeyCommand:
queryResp, err = i.agent.RemoveKey(req.Key)
default:
respHeader := responseHeader{Seq: seq, Error: unsupportedCommand}
client.Send(&respHeader, nil)
@ -676,15 +636,17 @@ func (i *AgentRPC) handleGossipKeyChange(client *rpcClient, seq uint64, cmd stri
Error: errToString(err),
}
if queryResp != nil {
resp = keyResponse{
Messages: queryResp.Messages,
Keys: queryResp.Keys,
NumNodes: queryResp.NumNodes,
NumResp: queryResp.NumResp,
NumErr: queryResp.NumErr,
NumNodes: queryResp.NumNodes,
}
}
return client.Send(&header, &resp)
return client.Send(&header, resp)
}
// Used to convert an error to a string representation

66
command/agent/rpc_client.go
View File

@ -176,60 +176,44 @@ func (c *RPCClient) WANMembers() ([]Member, error) {
return resp.Members, err
}
func (c *RPCClient) ListKeysLAN() (map[string]int, int, map[string]string, error) {
func (c *RPCClient) ListKeys() (map[string]int, int, map[string]string, error) {
header := requestHeader{
Command: listKeysLANCommand,
Command: listKeysCommand,
Seq: c.getSeq(),
}
resp := new(keyResponse)
err := c.genericRPC(&header, nil, resp)
return resp.Keys, resp.NumNodes, resp.Messages, err
}
func (c *RPCClient) ListKeysWAN() (map[string]int, int, map[string]string, error) {
func (c *RPCClient) InstallKey(key string) (map[string]string, error) {
header := requestHeader{
Command: listKeysWANCommand,
Command: installKeyCommand,
Seq: c.getSeq(),
}
resp := new(keyResponse)
err := c.genericRPC(&header, nil, resp)
return resp.Keys, resp.NumNodes, resp.Messages, err
}
func (c *RPCClient) InstallKeyWAN(key string) (map[string]string, error) {
return c.changeGossipKey(key, installKeyWANCommand)
}
func (c *RPCClient) InstallKeyLAN(key string) (map[string]string, error) {
return c.changeGossipKey(key, installKeyLANCommand)
}
func (c *RPCClient) UseKeyWAN(key string) (map[string]string, error) {
return c.changeGossipKey(key, useKeyWANCommand)
}
func (c *RPCClient) UseKeyLAN(key string) (map[string]string, error) {
return c.changeGossipKey(key, useKeyLANCommand)
}
func (c *RPCClient) RemoveKeyWAN(key string) (map[string]string, error) {
return c.changeGossipKey(key, removeKeyWANCommand)
}
func (c *RPCClient) RemoveKeyLAN(key string) (map[string]string, error) {
return c.changeGossipKey(key, removeKeyLANCommand)
}
func (c *RPCClient) changeGossipKey(key, cmd string) (map[string]string, error) {
header := requestHeader{
Command: cmd,
Seq: c.getSeq(),
}
req := keyRequest{key}
resp := new(keyResponse)
err := c.genericRPC(&header, &req, resp)
return resp.Messages, err
}
func (c *RPCClient) UseKey(key string) (map[string]string, error) {
header := requestHeader{
Command: useKeyCommand,
Seq: c.getSeq(),
}
req := keyRequest{key}
resp := new(keyResponse)
err := c.genericRPC(&header, &req, resp)
return resp.Messages, err
}
func (c *RPCClient) RemoveKey(key string) (map[string]string, error) {
header := requestHeader{
Command: removeKeyCommand,
Seq: c.getSeq(),
}
req := keyRequest{key}
resp := new(keyResponse)
err := c.genericRPC(&header, &req, resp)
return resp.Messages, err

43
command/keyring.go
View File

@ -106,59 +106,34 @@ func (c *KeyringCommand) Run(args []string) int {
}
if listKeys {
if wan {
c.Ui.Info("Asking all WAN members for installed keys...")
return c.listKeysOperation(client.ListKeysWAN)
}
c.Ui.Info("Asking all LAN members for installed keys...")
return c.listKeysOperation(client.ListKeysLAN)
c.Ui.Info("Asking all members for installed keys...")
return c.listKeysOperation(client.ListKeys)
}
if installKey != "" {
if wan {
c.Ui.Info("Installing new WAN gossip encryption key...")
if rval := c.keyOperation(installKey, client.InstallKeyWAN); rval != 0 {
c.Ui.Info("Installing new gossip encryption key...")
if rval := c.keyOperation(installKey, client.InstallKey); rval != 0 {
return rval
}
} else {
c.Ui.Info("Installing new LAN gossip encryption key...")
if rval := c.keyOperation(installKey, client.InstallKeyLAN); rval != 0 {
return rval
}
}
c.Ui.Info("Successfully installed key!")
return 0
}
if useKey != "" {
if wan {
c.Ui.Info("Changing primary WAN gossip encryption key...")
if rval := c.keyOperation(useKey, client.UseKeyWAN); rval != 0 {
c.Ui.Info("Changing primary gossip encryption key...")
if rval := c.keyOperation(useKey, client.UseKey); rval != 0 {
return rval
}
} else {
c.Ui.Info("Changing primary LAN gossip encryption key...")
if rval := c.keyOperation(useKey, client.UseKeyLAN); rval != 0 {
return rval
}
}
c.Ui.Info("Successfully changed primary key!")
return 0
}
if removeKey != "" {
if wan {
c.Ui.Info("Removing WAN gossip encryption key...")
if rval := c.keyOperation(removeKey, client.RemoveKeyWAN); rval != 0 {
c.Ui.Info("Removing gossip encryption key...")
if rval := c.keyOperation(removeKey, client.RemoveKey); rval != 0 {
return rval
}
} else {
c.Ui.Info("Removing LAN gossip encryption key...")
if rval := c.keyOperation(removeKey, client.RemoveKeyLAN); rval != 0 {
return rval
}
}
c.Ui.Info("Successfully removed key!")
c.Ui.Info("Successfully removed gossip encryption key!")
return 0
}

205
consul/internal_endpoint.go
View File

@ -1,7 +1,10 @@
package consul
import (
"fmt"
"github.com/hashicorp/consul/consul/structs"
"github.com/hashicorp/serf/serf"
)
// Internal endpoint is used to query the miscellaneous info that
@ -62,3 +65,205 @@ func (m *Internal) EventFire(args *structs.EventFireRequest,
// Fire the event
return m.srv.UserEvent(args.Name, args.Payload)
}
// TODO(ryanuber): Clean up all of these methods
func (m *Internal) InstallKey(args *structs.KeyringRequest,
reply *structs.KeyringResponse) error {
var respLAN, respWAN *serf.KeyResponse
var err error
if reply.Messages == nil {
reply.Messages = make(map[string]string)
}
if reply.Keys == nil {
reply.Keys = make(map[string]int)
}
m.srv.setQueryMeta(&reply.QueryMeta)
// Do a LAN key install. This will be invoked in each DC once the RPC call
// is forwarded below.
respLAN, err = m.srv.KeyManagerLAN().InstallKey(args.Key)
for node, msg := range respLAN.Messages {
reply.Messages["client."+node+"."+m.srv.config.Datacenter] = msg
}
reply.NumResp += respLAN.NumResp
reply.NumErr += respLAN.NumErr
reply.NumNodes += respLAN.NumNodes
if err != nil {
return fmt.Errorf("failed rotating LAN keyring in %s: %s",
m.srv.config.Datacenter,
err)
}
if !args.Forwarded {
// Only perform WAN key rotation once.
respWAN, err = m.srv.KeyManagerWAN().InstallKey(args.Key)
if err != nil {
return err
}
for node, msg := range respWAN.Messages {
reply.Messages["server."+node] = msg
}
reply.NumResp += respWAN.NumResp
reply.NumErr += respWAN.NumErr
reply.NumNodes += respWAN.NumNodes
// Mark key rotation as being already forwarded, then forward.
args.Forwarded = true
return m.srv.forwardAll("Internal.InstallKey", args, reply)
}
return nil
}
func (m *Internal) UseKey(args *structs.KeyringRequest,
reply *structs.KeyringResponse) error {
var respLAN, respWAN *serf.KeyResponse
var err error
if reply.Messages == nil {
reply.Messages = make(map[string]string)
}
if reply.Keys == nil {
reply.Keys = make(map[string]int)
}
m.srv.setQueryMeta(&reply.QueryMeta)
// Do a LAN key install. This will be invoked in each DC once the RPC call
// is forwarded below.
respLAN, err = m.srv.KeyManagerLAN().UseKey(args.Key)
for node, msg := range respLAN.Messages {
reply.Messages["client."+node+"."+m.srv.config.Datacenter] = msg
}
reply.NumResp += respLAN.NumResp
reply.NumErr += respLAN.NumErr
reply.NumNodes += respLAN.NumNodes
if err != nil {
return fmt.Errorf("failed rotating LAN keyring in %s: %s",
m.srv.config.Datacenter,
err)
}
if !args.Forwarded {
// Only perform WAN key rotation once.
respWAN, err = m.srv.KeyManagerWAN().UseKey(args.Key)
if err != nil {
return err
}
for node, msg := range respWAN.Messages {
reply.Messages["server."+node] = msg
}
reply.NumResp += respWAN.NumResp
reply.NumErr += respWAN.NumErr
reply.NumNodes += respWAN.NumNodes
// Mark key rotation as being already forwarded, then forward.
args.Forwarded = true
return m.srv.forwardAll("Internal.UseKey", args, reply)
}
return nil
}
func (m *Internal) RemoveKey(args *structs.KeyringRequest,
reply *structs.KeyringResponse) error {
var respLAN, respWAN *serf.KeyResponse
var err error
if reply.Messages == nil {
reply.Messages = make(map[string]string)
}
if reply.Keys == nil {
reply.Keys = make(map[string]int)
}
m.srv.setQueryMeta(&reply.QueryMeta)
// Do a LAN key install. This will be invoked in each DC once the RPC call
// is forwarded below.
respLAN, err = m.srv.KeyManagerLAN().RemoveKey(args.Key)
for node, msg := range respLAN.Messages {
reply.Messages["client."+node+"."+m.srv.config.Datacenter] = msg
}
reply.NumResp += respLAN.NumResp
reply.NumErr += respLAN.NumErr
reply.NumNodes += respLAN.NumNodes
if err != nil {
return fmt.Errorf("failed rotating LAN keyring in %s: %s",
m.srv.config.Datacenter,
err)
}
if !args.Forwarded {
// Only perform WAN key rotation once.
respWAN, err = m.srv.KeyManagerWAN().RemoveKey(args.Key)
if err != nil {
return err
}
for node, msg := range respWAN.Messages {
reply.Messages["server."+node] = msg
}
reply.NumResp += respWAN.NumResp
reply.NumErr += respWAN.NumErr
reply.NumNodes += respWAN.NumNodes
// Mark key rotation as being already forwarded, then forward.
args.Forwarded = true
return m.srv.forwardAll("Internal.RemoveKey", args, reply)
}
return nil
}
func (m *Internal) ListKeys(args *structs.KeyringRequest,
reply *structs.KeyringResponse) error {
var respLAN, respWAN *serf.KeyResponse
var err error
if reply.Messages == nil {
reply.Messages = make(map[string]string)
}
if reply.Keys == nil {
reply.Keys = make(map[string]int)
}
m.srv.setQueryMeta(&reply.QueryMeta)
// Do a LAN key install. This will be invoked in each DC once the RPC call
// is forwarded below.
respLAN, err = m.srv.KeyManagerLAN().ListKeys()
for node, msg := range respLAN.Messages {
reply.Messages["client."+node+"."+m.srv.config.Datacenter] = msg
}
reply.NumResp += respLAN.NumResp
reply.NumErr += respLAN.NumErr
reply.NumNodes += respLAN.NumNodes
if err != nil {
return fmt.Errorf("failed rotating LAN keyring in %s: %s",
m.srv.config.Datacenter,
err)
}
if !args.Forwarded {
// Only perform WAN key rotation once.
respWAN, err = m.srv.KeyManagerWAN().ListKeys()
if err != nil {
return err
}
for node, msg := range respWAN.Messages {
reply.Messages["server."+node] = msg
}
reply.NumResp += respWAN.NumResp
reply.NumErr += respWAN.NumErr
reply.NumNodes += respWAN.NumNodes
// Mark key rotation as being already forwarded, then forward.
args.Forwarded = true
return m.srv.forwardAll("Internal.ListKeys", args, reply)
}
return nil
}

12
consul/rpc.go
View File

@ -223,6 +223,18 @@ func (s *Server) forwardDC(method, dc string, args interface{}, reply interface{
return s.connPool.RPC(server.Addr, server.Version, method, args, reply)
}
// forwardAll forwards a single RPC call to every known datacenter.
func (s *Server) forwardAll(method string, args, reply interface{}) error {
for dc, _ := range s.remoteConsuls {
if dc != s.config.Datacenter {
if err := s.forwardDC(method, dc, args, reply); err != nil {
return err
}
}
}
return nil
}
// raftApply is used to encode a message, run it through raft, and return
// the FSM response along with any errors
func (s *Server) raftApply(t structs.MessageType, msg interface{}) (interface{}, error) {

19
consul/structs/structs.go
View File

@ -531,3 +531,22 @@ func Encode(t MessageType, msg interface{}) ([]byte, error) {
err := codec.NewEncoder(&buf, msgpackHandle).Encode(msg)
return buf.Bytes(), err
}
// KeyringRequest encapsulates a request to modify an encryption keyring.
// It can be used for install, remove, or use key type operations.
type KeyringRequest struct {
Key string
Forwarded bool
QueryOptions
}
// KeyringResponse is a unified key response and can be used for install,
// remove, use, as well as listing key queries.
type KeyringResponse struct {
Messages map[string]string
Keys map[string]int
NumNodes int
NumResp int
NumErr int
QueryMeta
}