Merged PR 12878148: bridge: Force sequential message handling for confidential containers

[cherry-picked from f81b450894206a79fff4d63182ff034ba503ebdb]

This PR contains 2 commits. The first one is the fix:

**bridge: Force sequential message handling for confidential containers**

This fixes a vulnerability (and reduces the surface for other similar potential
vulnerabilities) in confidential containers where if the host sends a
mount/unmount modification request concurrently with an ongoing CreateContainer
request, the host could re-order or skip image layers for the container to be
started.

While this could be fixed by adding mutex lock/unlock around the individual
modifyMappedVirtualDisk/modifyCombinedLayers/CreateContainer functions, we
decided that in order to prevent any more of this class of issues, the UVM, when
running in confidential mode, should just not allow concurrent requests (with
exception for any actually long-running requests, which for now is just
waitProcess).

The second one adds a log entry for when the processing thread blocks.  This
will make it easier to debug should the gcs "hung" on a request.

This PR is created on ADO targeting the conf branch as this security
vulnerability is not public yet. This fix should be backported to main once
deployed.

Related work items: #33357501, #34327300
Signed-off-by: Tingmao Wang <tingmaowang@microsoft.com>

Author: micromaomao

cmd/gcs/main.go

@@ -31,6 +31,7 @@ import (
 	"github.com/Microsoft/hcsshim/internal/log"
 	"github.com/Microsoft/hcsshim/internal/oc"
 	"github.com/Microsoft/hcsshim/internal/version"
+	"github.com/Microsoft/hcsshim/pkg/amdsevsnp"
 	"github.com/Microsoft/hcsshim/pkg/securitypolicy"
 )
 
@@ -362,6 +363,10 @@ func main() {
 	b := bridge.Bridge{
 		Handler:  mux,
 		EnableV4: *v4,
+
+		// For confidential containers, we protect ourselves against attacks caused
+		// by concurrent modifications, by processing one request at a time.
+		ForceSequential: amdsevsnp.IsSNP(),
 	}
 	h := hcsv2.NewHost(rtime, tport, initialEnforcer, logWriter)
 	// Initialize virtual pod support in the host

internal/guest/bridge/bridge.go

@@ -177,6 +177,10 @@ type Bridge struct {
 	Handler Handler
 	// EnableV4 enables the v4+ bridge and the schema v2+ interfaces.
 	EnableV4 bool
+	// Setting ForceSequential to true will force the bridge to only process one
+	// request at a time, except for certain long-running operations (as defined
+	// in asyncMessages).
+	ForceSequential bool
 
 	// responseChan is the response channel used for both request/response
 	// and publish notification workflows.
@@ -191,6 +195,14 @@ type Bridge struct {
 	protVer prot.ProtocolVersion
 }
 
+// Messages that will be processed asynchronously even in sequential mode.  Note
+// that in sequential mode, these messages will still wait for any in-progress
+// non-async messages to be handled before they are processed, but once they are
+// "acknowledged", the rest will be done asynchronously.
+var alwaysAsyncMessages map[prot.MessageIdentifier]bool = map[prot.MessageIdentifier]bool{
+	prot.ComputeSystemWaitForProcessV1: true,
+}
+
 // AssignHandlers creates and assigns the appropriate bridge
 // events to be listen for and intercepted on `mux` before forwarding
 // to `gcs` for handling.
@@ -238,6 +250,10 @@ func (b *Bridge) ListenAndServe(bridgeIn io.ReadCloser, bridgeOut io.WriteCloser
 	defer close(requestErrChan)
 	defer bridgeIn.Close()
 
+	if b.ForceSequential {
+		log.G(context.Background()).Info("bridge: ForceSequential enabled")
+	}
+
 	// Receive bridge requests and schedule them to be processed.
 	go func() {
 		var recverr error
@@ -340,30 +356,36 @@ func (b *Bridge) ListenAndServe(bridgeIn io.ReadCloser, bridgeOut io.WriteCloser
 	}()
 	// Process each bridge request async and create the response writer.
 	go func() {
-		for req := range requestChan {
-			go func(r *Request) {
-				br := bridgeResponse{
-					ctx: r.Context,
-					header: &prot.MessageHeader{
-						Type: prot.GetResponseIdentifier(r.Header.Type),
-						ID:   r.Header.ID,
-					},
-				}
-				resp, err := b.Handler.ServeMsg(r)
-				if resp == nil {
-					resp = &prot.MessageResponseBase{}
-				}
-				resp.Base().ActivityID = r.ActivityID
-				if err != nil {
-					span := trace.FromContext(r.Context)
-					if span != nil {
-						oc.SetSpanStatus(span, err)
-					}
-					setErrorForResponseBase(resp.Base(), err, "gcs" /* moduleName */)
+		doOneRequest := func(r *Request) {
+			br := bridgeResponse{
+				ctx: r.Context,
+				header: &prot.MessageHeader{
+					Type: prot.GetResponseIdentifier(r.Header.Type),
+					ID:   r.Header.ID,
+				},
+			}
+			resp, err := b.Handler.ServeMsg(r)
+			if resp == nil {
+				resp = &prot.MessageResponseBase{}
+			}
+			resp.Base().ActivityID = r.ActivityID
+			if err != nil {
+				span := trace.FromContext(r.Context)
+				if span != nil {
+					oc.SetSpanStatus(span, err)
 				}
-				br.response = resp
-				b.responseChan <- br
-			}(req)
+				setErrorForResponseBase(resp.Base(), err, "gcs" /* moduleName */)
+			}
+			br.response = resp
+			b.responseChan <- br
+		}
+
+		for req := range requestChan {
+			if b.ForceSequential && !alwaysAsyncMessages[req.Header.Type] {
+				runSequentialRequestHandler(req, doOneRequest)
+			} else {
+				go doOneRequest(req)
+			}
 		}
 	}()
 	// Process each bridge response sync. This channel is for request/response and publish workflows.
@@ -423,6 +445,32 @@ func (b *Bridge) ListenAndServe(bridgeIn io.ReadCloser, bridgeOut io.WriteCloser
 	}
 }
 
+// Do handleFn(r), but prints a warning if handleFn does not, or takes too long
+// to return.
+func runSequentialRequestHandler(r *Request, handleFn func(*Request)) {
+	// Note that this is only a context used for triggering the blockage
+	// warning, the request processing still uses r.Context.  We don't want to
+	// cancel the request handling itself when we reach the 5s timeout.
+	timeoutCtx, cancel := context.WithTimeout(r.Context, 5*time.Second)
+	go func() {
+		<-timeoutCtx.Done()
+		if errors.Is(timeoutCtx.Err(), context.DeadlineExceeded) {
+			log.G(timeoutCtx).WithFields(logrus.Fields{
+				// We want to log those even though we're providing r.Context, since if
+				// the request never finishes the span end log will never get written,
+				// and we may therefore not be able to find out about the following info
+				// otherwise:
+				"message-type": r.Header.Type.String(),
+				"message-id":   r.Header.ID,
+				"activity-id":  r.ActivityID,
+				"container-id": r.ContainerID,
+			}).Warnf("bridge: request processing thread in sequential mode blocked on the current request for more than 5 seconds")
+		}
+	}()
+	defer cancel()
+	handleFn(r)
+}
+
 // PublishNotification writes a specific notification to the bridge.
 func (b *Bridge) PublishNotification(n *prot.ContainerNotification) {
 	ctx, span := oc.StartSpan(context.Background(),