go/worker/keymanager: Actually allow replication to maybe work

Access control forbidding replication may be more secure, but is not all
that useful.

go/keymanager/api/api.go

@@ -139,6 +139,9 @@ func VerifyExtraInfo(logger *logging.Logger, rt *registry.Runtime, nodeRt *node.
 	} else if err := registry.VerifyNodeRuntimeEnclaveIDs(logger, nodeRt, rt, ts); err != nil {
 		return nil, err
 	}
+	if nodeRt.ExtraInfo == nil {
+		return nil, fmt.Errorf("keymanager: missing ExtraInfo")
+	}
 
 	var untrustedSignedInitResponse SignedInitResponse
 	if err := cbor.Unmarshal(nodeRt.ExtraInfo, &untrustedSignedInitResponse); err != nil {

go/worker/keymanager/watcher.go

@@ -0,0 +1,115 @@
+package keymanager
+
+import (
+	"github.com/oasislabs/oasis-core/go/common/accessctl"
+	"github.com/oasislabs/oasis-core/go/common/crypto/signature"
+	"github.com/oasislabs/oasis-core/go/common/node"
+	registry "github.com/oasislabs/oasis-core/go/registry/api"
+	"github.com/oasislabs/oasis-core/go/runtime/committee"
+)
+
+type kmNodeWatcher struct {
+	w        *Worker
+	registry registry.Backend
+}
+
+func newKmNodeWatcher(w *Worker) *kmNodeWatcher {
+	return &kmNodeWatcher{
+		w:        w,
+		registry: w.commonWorker.Consensus.Registry(),
+	}
+}
+
+func (knw *kmNodeWatcher) watchNodes() {
+	nodesCh, nodesSub, err := knw.registry.WatchNodeList(knw.w.ctx)
+	if err != nil {
+		knw.w.logger.Error("worker/keymanager: failed to watch node list",
+			"err", err,
+		)
+		return
+	}
+	defer nodesSub.Close()
+
+	watcher, err := committee.NewNodeDescriptorWatcher(knw.w.ctx, knw.registry)
+	if err != nil {
+		knw.w.logger.Error("worker/keymanager: failed to create node desc watcher",
+			"err", err,
+		)
+		return
+	}
+	watcherCh, watcherSub, err := watcher.WatchNodeUpdates()
+	if err != nil {
+		knw.w.logger.Error("worker/keymanager: failed to watch node updates",
+			"err", err,
+		)
+		return
+	}
+	defer watcherSub.Close()
+
+	var activeNodes map[signature.PublicKey]bool
+	for {
+		select {
+		case nodeList := <-nodesCh:
+			watcher.Reset()
+			activeNodes = knw.rebuildActiveNodeIDs(nodeList.Nodes)
+			for id := range activeNodes {
+				if _, err := watcher.WatchNode(knw.w.ctx, id); err != nil {
+					knw.w.logger.Error("worker/keymanager: failed to watch node",
+						"err", err,
+						"id", id,
+					)
+				}
+			}
+		case watcherEv := <-watcherCh:
+			if watcherEv.Update == nil {
+				continue
+			}
+			if !activeNodes[watcherEv.Update.ID] {
+				continue
+			}
+		case <-knw.w.stopCh:
+			return
+		}
+
+		// Rebuild the access policy, something has changed.
+		policy := accessctl.NewPolicy()
+
+		sentryCerts := knw.w.commonWorker.GetConfig().SentryCertificates
+		for _, cert := range sentryCerts {
+			sentryNodesPolicy.AddCertPolicy(&policy, cert)
+		}
+
+		var nodes []*node.Node
+		for id := range activeNodes {
+			n := watcher.Lookup(id)
+			if n == nil {
+				continue
+			}
+			nodes = append(nodes, n)
+		}
+
+		kmNodesPolicy.AddRulesForNodeRoles(&policy, nodes, node.RoleKeyManager)
+		knw.w.grpcPolicy.SetAccessPolicy(policy, knw.w.runtime.ID())
+		knw.w.logger.Debug("worker/keymanager: new km runtime access policy in effect",
+			"policy", policy,
+		)
+	}
+}
+
+func (knw *kmNodeWatcher) rebuildActiveNodeIDs(nodeList []*node.Node) map[signature.PublicKey]bool {
+	m := make(map[signature.PublicKey]bool)
+	id := knw.w.runtime.ID()
+	for _, n := range nodeList {
+		if !n.HasRoles(node.RoleKeyManager) {
+			continue
+		}
+		for _, rt := range n.Runtimes {
+			if rt.ID.Equal(&id) {
+				m[n.ID] = true
+				break
+			}
+		}
+	}
+
+	return m
+}

go/worker/keymanager/worker.go

@@ -189,6 +189,22 @@ func (w *Worker) callLocal(ctx context.Context, data []byte) ([]byte, error) {
 }
 
 func (w *Worker) updateStatus(status *api.Status, startedEvent *host.StartedEvent) error {
+	var initOk bool
+	defer func() {
+		if !initOk {
+			// This is likely a new key manager that needs to replicate.
+			// Send a node registration anyway, so that other nodes know
+			// to update their access control.
+			w.roleProvider.SetAvailable(func(n *node.Node) error {
+				rt := n.AddOrUpdateRuntime(w.runtime.ID())
+				rt.Version = startedEvent.Version
+				rt.ExtraInfo = nil
+				rt.Capabilities.TEE = startedEvent.CapabilityTEE
+				return nil
+			})
+		}
+	}()
+
 	// Initialize the key manager.
 	type InitRequest struct {
 		Checksum    []byte `json:"checksum"`
@@ -287,6 +303,7 @@ func (w *Worker) updateStatus(status *api.Status, startedEvent *host.StartedEven
 	)
 
 	// Register as we are now ready to handle requests.
+	initOk = true
 	w.roleProvider.SetAvailable(func(n *node.Node) error {
 		rt := n.AddOrUpdateRuntime(w.runtime.ID())
 		rt.Version = startedEvent.Version
@@ -353,6 +370,11 @@ func (w *Worker) worker() { // nolint: gocyclo
 	case <-w.commonWorker.Consensus.Synced():
 	}
 
+	// Need to explicitly watch for updates related to the key manager runtime
+	// itself.
+	knw := newKmNodeWatcher(w)
+	go knw.watchNodes()
+
 	// Subscribe to key manager status updates.
 	statusCh, statusSub := w.backend.WatchStatuses()
 	defer statusSub.Close()
@@ -558,30 +580,8 @@ func (crw *clientRuntimeWatcher) updateExternalServicePolicyLocked(snapshot *com
 		sentryNodesPolicy.AddCertPolicy(&policy, cert)
 	}
 
-	// Fetch current KM node public keys, get their nodes, apply rules.
-	height := snapshot.GetGroupVersion()
-	status, err := crw.w.backend.GetStatus(crw.w.ctx, crw.w.runtime.ID(), height)
-	if err != nil {
-		crw.w.logger.Error("worker/keymanager: unable to get KM status",
-			"runtimeID", crw.w.runtime.ID(),
-			"err", err)
-	} else {
-		var kmNodes []*node.Node
-
-		for _, pk := range status.Nodes {
-			n, err := crw.node.Consensus.Registry().GetNode(crw.w.ctx, &registry.IDQuery{ID: pk, Height: height})
-			if err != nil {
-				crw.w.logger.Error("worker/keymanager: unable to get KM node info", "err", err)
-			} else {
-				kmNodes = append(kmNodes, n)
-			}
-		}
-
-		kmNodesPolicy.AddRulesForNodeRoles(&policy, kmNodes, node.RoleKeyManager)
-	}
-
 	crw.w.grpcPolicy.SetAccessPolicy(policy, crw.node.Runtime.ID())
-	crw.w.logger.Debug("worker/keymanager: new access policy in effect", "policy", policy)
+	crw.w.logger.Debug("worker/keymanager: new normal runtime access policy in effect", "policy", policy)
 }
 
 // Guarded by CrossNode.