runtime: Notify runtimes of its key manager policy updates

.changelog/2919.bugfix.md

@@ -0,0 +1,6 @@
+runtime: Notify runtimes of its key manager policy updates
+
+Before runtimes were unaware of any key-manager policy updates. The runtime
+only queried for the active key-manager policy at startup. This is now changed
+so that the host notifies runtimes of any key-manager policy changes and
+runtime updates the policies.

client/src/rpc/client.rs

@@ -1,7 +1,10 @@
 //! Enclave RPC client.
-use std::sync::{
-    atomic::{AtomicBool, Ordering},
-    Arc, Mutex,
+use std::{
+    collections::HashSet,
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc, Mutex,
+    },
 };
 
 use failure::{Fail, Fallible};
@@ -19,7 +22,7 @@ use tokio_executor::spawn;
 #[cfg(not(target_env = "sgx"))]
 use oasis_core_runtime::common::runtime::RuntimeId;
 use oasis_core_runtime::{
-    common::cbor,
+    common::{cbor, sgx::avr::EnclaveIdentity},
     protocol::Protocol,
     rpc::{
         session::{Builder, Session},
@@ -372,6 +375,15 @@ impl RpcClient {
                 }),
         )
     }
+
+    /// Update session enclaves if changed.
+    pub fn update_enclaves(&self, enclaves: Option<HashSet<EnclaveIdentity>>) {
+        let mut session = self.inner.session.lock().unwrap();
+        if session.builder.get_remote_enclaves() != &enclaves {
+            session.builder = session.builder.clone().remote_enclaves(enclaves);
+            session.reset();
+        }
+    }
 }
 
 #[cfg(test)]

go/oasis-test-runner/scenario/e2e/keymanager_upgrade.go

@@ -294,21 +294,16 @@ func (sc *kmUpgradeImpl) Run(childEnv *env.Env) error {
 		return fmt.Errorf("old keymanager node shutdown: %w", err)
 	}
 
-	// The last part of the test won't work until:
-	// https://github.com/oasislabs/oasis-core/issues/2919
-	/*
-		// Run test again.
-		sc.logger.Info("starting a second client to check if key manager works")
-		sc.runtimeImpl.clientArgs = []string{"--key", "key2"}
-		cmd, err = sc.startClient(childEnv)
-		if err != nil {
-			return err
-		}
-		client2ErrCh := make(chan error)
-		go func() {
-			client2ErrCh <- cmd.Wait()
-		}()
-		return sc.wait(childEnv, cmd, client2ErrCh)
-	*/
-	return nil
+	// Run client again.
+	sc.logger.Info("starting a second client to check if key manager works")
+	sc.runtimeImpl.clientArgs = []string{"--key", "key2"}
+	cmd, err = sc.startClient(childEnv)
+	if err != nil {
+		return err
+	}
+	client2ErrCh := make(chan error)
+	go func() {
+		client2ErrCh <- cmd.Wait()
+	}()
+	return sc.wait(childEnv, cmd, client2ErrCh)
 }

go/runtime/host/protocol/connection.go

@@ -340,10 +340,6 @@ func (c *connection) handleMessage(ctx context.Context, message *Message) {
 		var allowed bool
 		state := c.getState()
 		switch {
-		case state == stateInitializing:
-			// Only whitelisted methods are allowed.
-			body := message.Body
-			allowed = body.HostKeyManagerPolicyRequest != nil
 		case state == stateReady:
 			// All requests allowed.
 			allowed = true

go/runtime/host/protocol/types.go

@@ -51,6 +51,12 @@ type Message struct {
 	SpanContext []byte      `json:"span_context"`
 }
 
+// RuntimeKeyManagerPolicyUpdateRequest is a runtime key manager policy request
+// message body.
+type RuntimeKeyManagerPolicyUpdateRequest struct {
+	SignedPolicyRaw []byte `json:"signed_policy_raw"`
+}
+
 // Body is a protocol message body.
 type Body struct {
 	Empty *Empty `json:",omitempty"`
@@ -77,18 +83,18 @@ type Body struct {
 	RuntimeExecuteTxBatchResponse         *RuntimeExecuteTxBatchResponse         `json:",omitempty"`
 	RuntimeAbortRequest                   *Empty                                 `json:",omitempty"`
 	RuntimeAbortResponse                  *Empty                                 `json:",omitempty"`
+	RuntimeKeyManagerPolicyUpdateRequest  *RuntimeKeyManagerPolicyUpdateRequest  `json:",omitempty"`
+	RuntimeKeyManagerPolicyUpdateResponse *Empty                                 `json:",omitempty"`
 
 	// Host interface.
-	HostKeyManagerPolicyRequest  *HostKeyManagerPolicyRequest  `json:",omitempty"`
-	HostKeyManagerPolicyResponse *HostKeyManagerPolicyResponse `json:",omitempty"`
-	HostRPCCallRequest           *HostRPCCallRequest           `json:",omitempty"`
-	HostRPCCallResponse          *HostRPCCallResponse          `json:",omitempty"`
-	HostStorageSyncRequest       *HostStorageSyncRequest       `json:",omitempty"`
-	HostStorageSyncResponse      *HostStorageSyncResponse      `json:",omitempty"`
-	HostLocalStorageGetRequest   *HostLocalStorageGetRequest   `json:",omitempty"`
-	HostLocalStorageGetResponse  *HostLocalStorageGetResponse  `json:",omitempty"`
-	HostLocalStorageSetRequest   *HostLocalStorageSetRequest   `json:",omitempty"`
-	HostLocalStorageSetResponse  *Empty                        `json:",omitempty"`
+	HostRPCCallRequest          *HostRPCCallRequest          `json:",omitempty"`
+	HostRPCCallResponse         *HostRPCCallResponse         `json:",omitempty"`
+	HostStorageSyncRequest      *HostStorageSyncRequest      `json:",omitempty"`
+	HostStorageSyncResponse     *HostStorageSyncResponse     `json:",omitempty"`
+	HostLocalStorageGetRequest  *HostLocalStorageGetRequest  `json:",omitempty"`
+	HostLocalStorageGetResponse *HostLocalStorageGetResponse `json:",omitempty"`
+	HostLocalStorageSetRequest  *HostLocalStorageSetRequest  `json:",omitempty"`
+	HostLocalStorageSetResponse *Empty                       `json:",omitempty"`
 }
 
 // Type returns the message type by determining the name of the first non-nil member.
@@ -226,15 +232,6 @@ type RuntimeExecuteTxBatchResponse struct {
 	Batch ComputedBatch `json:"batch"`
 }
 
-// HostKeyManagerPolicyRequest is a host key manager policy request message body.
-type HostKeyManagerPolicyRequest struct {
-}
-
-// HostKeyManagerPolicyResponse is a host key manager policy response message body.
-type HostKeyManagerPolicyResponse struct {
-	SignedPolicyRaw []byte `json:"signed_policy_raw"`
-}
-
 // HostRPCCallRequest is a host RPC call request message body.
 type HostRPCCallRequest struct {
 	Endpoint string `json:"endpoint"`

go/runtime/host/tests/tester.go

@@ -28,30 +28,6 @@ type mockMessageHandler struct{}
 
 // Implements host.Handler.
 func (h *mockMessageHandler) Handle(ctx context.Context, body *protocol.Body) (*protocol.Body, error) {
-	// Key manager.
-	if body.HostKeyManagerPolicyRequest != nil {
-		// Generate a dummy key manager policy for tests.
-		policy := keymanager.PolicySGX{
-			Serial:   1,
-			Enclaves: map[sgx.EnclaveIdentity]*keymanager.EnclavePolicySGX{},
-		}
-		sigPolicy := keymanager.SignedPolicySGX{
-			Policy: policy,
-		}
-		for _, signer := range keymanager.TestSigners[1:] {
-			sig, err := signature.Sign(signer, keymanager.PolicySGXSignatureContext, cbor.Marshal(policy))
-			if err != nil {
-				return nil, fmt.Errorf("failed to sign mock policy: %w", err)
-			}
-
-			sigPolicy.Signatures = append(sigPolicy.Signatures, *sig)
-		}
-
-		return &protocol.Body{HostKeyManagerPolicyResponse: &protocol.HostKeyManagerPolicyResponse{
-			SignedPolicyRaw: cbor.Marshal(sigPolicy),
-		}}, nil
-	}
-
 	return nil, fmt.Errorf("method not supported")
 }
 
@@ -97,6 +73,29 @@ func TestProvisioner(
 	}
 }
 
+func mockKeyManagerPolicyRequest() (*protocol.Body, error) {
+	// Generate a dummy key manager policy for tests.
+	policy := keymanager.PolicySGX{
+		Serial:   1,
+		Enclaves: map[sgx.EnclaveIdentity]*keymanager.EnclavePolicySGX{},
+	}
+	sigPolicy := keymanager.SignedPolicySGX{
+		Policy: policy,
+	}
+	for _, signer := range keymanager.TestSigners[1:] {
+		sig, err := signature.Sign(signer, keymanager.PolicySGXSignatureContext, cbor.Marshal(policy))
+		if err != nil {
+			return nil, fmt.Errorf("failed to sign mock policy: %w", err)
+		}
+
+		sigPolicy.Signatures = append(sigPolicy.Signatures, *sig)
+	}
+
+	return &protocol.Body{RuntimeKeyManagerPolicyUpdateRequest: &protocol.RuntimeKeyManagerPolicyUpdateRequest{
+		SignedPolicyRaw: cbor.Marshal(sigPolicy),
+	}}, nil
+}
+
 func testBasic(t *testing.T, cfg host.Config, p host.Provisioner) {
 	require := require.New(t)
 
@@ -125,6 +124,13 @@ func testBasic(t *testing.T, cfg host.Config, p host.Provisioner) {
 	require.NoError(err, "Call")
 	require.NotNil(rsp.Empty, "runtime response to RuntimePingRequest should return an Empty body")
 
+	req, err := mockKeyManagerPolicyRequest()
+	require.NoError(err, "mockKeyManagerPolicyRequest")
+
+	rsp, err = r.Call(ctx, req)
+	require.NoError(err, "KeyManagerPolicyRequest Call")
+	require.NotNil(rsp.RuntimeKeyManagerPolicyUpdateResponse, "runtime response to KeyManagerPolicyRequest should return an RuntimeKeyManagerPolicyUpdateResponse body")
+
 	// Request the runtime to stop.
 	r.Stop()
 

go/worker/common/committee/runtime_host.go

@@ -3,21 +3,27 @@ package committee
 import (
 	"context"
 	"errors"
-	"fmt"
+	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/opentracing/opentracing-go"
 
 	"github.com/oasislabs/oasis-core/go/common/cbor"
-	consensus "github.com/oasislabs/oasis-core/go/consensus/api"
+	"github.com/oasislabs/oasis-core/go/common/logging"
 	keymanagerApi "github.com/oasislabs/oasis-core/go/keymanager/api"
 	keymanagerClient "github.com/oasislabs/oasis-core/go/keymanager/client"
-	registry "github.com/oasislabs/oasis-core/go/registry/api"
+	"github.com/oasislabs/oasis-core/go/runtime/host"
 	"github.com/oasislabs/oasis-core/go/runtime/host/protocol"
 	"github.com/oasislabs/oasis-core/go/runtime/localstorage"
 	runtimeRegistry "github.com/oasislabs/oasis-core/go/runtime/registry"
 	storage "github.com/oasislabs/oasis-core/go/storage/api"
 )
 
+const (
+	retryInterval = 1 * time.Second
+	maxRetries    = 15
+)
+
 var (
 	errMethodNotSupported   = errors.New("method not supported")
 	errEndpointNotSupported = errors.New("RPC endpoint not supported")
@@ -34,31 +40,6 @@ type computeRuntimeHostHandler struct {
 }
 
 func (h *computeRuntimeHostHandler) Handle(ctx context.Context, body *protocol.Body) (*protocol.Body, error) {
-	// Key manager.
-	if body.HostKeyManagerPolicyRequest != nil {
-		rt, err := h.runtime.RegistryDescriptor(ctx)
-		if err != nil {
-			return nil, fmt.Errorf("runtime host: failed to obtain runtime descriptor: %w", err)
-		}
-		if rt.KeyManager == nil {
-			return nil, errors.New("runtime has no key manager")
-		}
-		status, err := h.keyManager.GetStatus(ctx, &registry.NamespaceQuery{
-			ID:     *rt.KeyManager,
-			Height: consensus.HeightLatest,
-		})
-		if err != nil {
-			return nil, err
-		}
-
-		var policy keymanagerApi.SignedPolicySGX
-		if status != nil && status.Policy != nil {
-			policy = *status.Policy
-		}
-		return &protocol.Body{HostKeyManagerPolicyResponse: &protocol.HostKeyManagerPolicyResponse{
-			SignedPolicyRaw: cbor.Marshal(policy),
-		}}, nil
-	}
 	// RPC.
 	if body.HostRPCCallRequest != nil {
 		switch body.HostRPCCallRequest.Endpoint {
@@ -125,6 +106,92 @@ func (n *Node) GetRuntime() runtimeRegistry.Runtime {
 	return n.Runtime
 }
 
+// computeRuntimeHostNotifier is a runtime host notifier suitable for compute
+// runtimes.
+type computeRuntimeHostNotifier struct {
+	ctx context.Context
+
+	logger *logging.Logger
+
+	runtime    runtimeRegistry.Runtime
+	host       host.Runtime
+	keyManager keymanagerApi.Backend
+}
+
+func (h *computeRuntimeHostNotifier) watchPolicyUpdates() {
+	// Wait for the runtime.
+	rt, err := h.runtime.RegistryDescriptor(h.ctx)
+	if err != nil {
+		h.logger.Error("failed to wait for registry descriptor",
+			"err", err,
+		)
+		return
+	}
+	if rt.KeyManager == nil {
+		h.logger.Info("no keymanager needed, not watching for policy updates")
+		return
+	}
+
+	stCh, stSub := h.keyManager.WatchStatuses()
+	defer stSub.Close()
+	h.logger.Info("watching policy updates", "keymanager_runtime", rt.KeyManager)
+
+	for {
+		select {
+		case <-h.ctx.Done():
+			return
+		case st := <-stCh:
+			h.logger.Debug("got policy update", "status", st)
+
+			// Ignore status updates if key manager is not yet known (is nil)
+			// or if the status update is for a different key manager.
+			if !st.ID.Equal(rt.KeyManager) {
+				continue
+			}
+
+			raw := cbor.Marshal(st.Policy)
+			req := &protocol.Body{RuntimeKeyManagerPolicyUpdateRequest: &protocol.RuntimeKeyManagerPolicyUpdateRequest{
+				SignedPolicyRaw: raw,
+			}}
+
+			var response *protocol.Body
+			call := func() error {
+				resp, err := h.host.Call(h.ctx, req)
+				if err != nil {
+					h.logger.Error("failed to dispatch RPC call to runtime",
+						"err", err,
+					)
+					return err
+				}
+				response = resp
+				return nil
+			}
+
+			retry := backoff.WithMaxRetries(backoff.NewConstantBackOff(retryInterval), maxRetries)
+			err := backoff.Retry(call, backoff.WithContext(retry, h.ctx))
+			if err != nil {
+				h.logger.Error("failed dispatching key manager policy update to runtime",
+					"err", err,
+				)
+				continue
+			}
+			h.logger.Debug("key manager policy updated dispatched", "response", response)
+		}
+	}
+}
+
+// Implements RuntimeHostHandlerFactory.
+func (n *Node) NewNotifier(ctx context.Context, host host.Runtime) {
+	w := &computeRuntimeHostNotifier{
+		ctx,
+		logging.GetLogger("committee/runtime-host"),
+		n.Runtime,
+		host,
+		n.KeyManager,
+	}
+	go w.watchPolicyUpdates()
+}
+
 // Implements RuntimeHostHandlerFactory.
 func (n *Node) NewRuntimeHostHandler() protocol.Handler {
 	return &computeRuntimeHostHandler{