refactor(worker): Move tcp_proxy into its own package (#1458)

* refactor(worker): Move tcp_proxy into its own package

This refactors:

- the tcp_proxy.go into worker/proxy/tcp
- add a missing unit test for the HandleTcpProxyV1
- refactors session.go into worker/session, needed to avoid import cycle
- adds a baseline for a mock controller session client
- go lint fixes

* Add sleep between spinning listener goroutine and dialing

Co-authored-by: Jeff Mitchell <jeffrey.mitchell@gmail.com>

internal/gen/controller/servers/services/session_service_grpc.pb.go

@@ -25,7 +25,7 @@ type SessionServiceClient interface {
 	ActivateSession(ctx context.Context, in *ActivateSessionRequest, opts ...grpc.CallOption) (*ActivateSessionResponse, error)
 	// CancelSession allows a worker to request that the controller cancel a session.
 	CancelSession(ctx context.Context, in *CancelSessionRequest, opts ...grpc.CallOption) (*CancelSessionResponse, error)
-	// AuthorizeConnection allows a worker to activate a session on a controller.
+	// AuthorizeConnection allows a worker to authorize a connection on a controller.
 	AuthorizeConnection(ctx context.Context, in *AuthorizeConnectionRequest, opts ...grpc.CallOption) (*AuthorizeConnectionResponse, error)
 	// ConnectConnection updates a connection to set it to connected
 	ConnectConnection(ctx context.Context, in *ConnectConnectionRequest, opts ...grpc.CallOption) (*ConnectConnectionResponse, error)
@@ -106,7 +106,7 @@ type SessionServiceServer interface {
 	ActivateSession(context.Context, *ActivateSessionRequest) (*ActivateSessionResponse, error)
 	// CancelSession allows a worker to request that the controller cancel a session.
 	CancelSession(context.Context, *CancelSessionRequest) (*CancelSessionResponse, error)
-	// AuthorizeConnection allows a worker to activate a session on a controller.
+	// AuthorizeConnection allows a worker to authorize a connection on a controller.
 	AuthorizeConnection(context.Context, *AuthorizeConnectionRequest) (*AuthorizeConnectionResponse, error)
 	// ConnectConnection updates a connection to set it to connected
 	ConnectConnection(context.Context, *ConnectConnectionRequest) (*ConnectConnectionResponse, error)

internal/gen/controller/servers/services/testing.go

@@ -0,0 +1,41 @@
+package services
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+)
+
+// TODO (lcr 08/2021): implement mockSessionServiceClient outside of primitive support for
+// ConnectConnection
+type mockSessionServiceClient struct{}
+
+func NewMockSessionServiceClient() SessionServiceClient {
+	return &mockSessionServiceClient{}
+}
+
+func (c *mockSessionServiceClient) LookupSession(_ context.Context, _ *LookupSessionRequest, _ ...grpc.CallOption) (*LookupSessionResponse, error) {
+	panic("not implemented")
+}
+
+func (c *mockSessionServiceClient) ActivateSession(_ context.Context, _ *ActivateSessionRequest, _ ...grpc.CallOption) (*ActivateSessionResponse, error) {
+	panic("not implemented")
+}
+
+func (c *mockSessionServiceClient) CancelSession(_ context.Context, _ *CancelSessionRequest, _ ...grpc.CallOption) (*CancelSessionResponse, error) {
+	panic("not implemented")
+}
+
+func (c *mockSessionServiceClient) AuthorizeConnection(_ context.Context, _ *AuthorizeConnectionRequest, _ ...grpc.CallOption) (*AuthorizeConnectionResponse, error) {
+	panic("not implemented")
+}
+
+func (c *mockSessionServiceClient) ConnectConnection(_ context.Context, _ *ConnectConnectionRequest, _ ...grpc.CallOption) (*ConnectConnectionResponse, error) {
+	return &ConnectConnectionResponse{
+		Status: CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED,
+	}, nil
+}
+
+func (c *mockSessionServiceClient) CloseConnection(_ context.Context, _ *CloseConnectionRequest, _ ...grpc.CallOption) (*CloseConnectionResponse, error) {
+	panic("not implemented")
+}

internal/proto/local/controller/servers/services/v1/session_service.proto

@@ -19,7 +19,7 @@ service SessionService {
 	// CancelSession allows a worker to request that the controller cancel a session.
 	rpc CancelSession(CancelSessionRequest) returns (CancelSessionResponse) {}
 
-	// AuthorizeConnection allows a worker to activate a session on a controller.
+	// AuthorizeConnection allows a worker to authorize a connection on a controller.
 	rpc AuthorizeConnection(AuthorizeConnectionRequest) returns (AuthorizeConnectionResponse) {}
 
 	// ConnectConnection updates a connection to set it to connected

internal/servers/controller/handlers/targets/target_service.go

@@ -1017,9 +1017,9 @@ HostSourceIterationLoop:
 	for _, l := range libs {
 		reqs = append(reqs, credential.Request{
 			SourceId: l.Id(),
-			Purpose:  credential.Purpose(l.CredentialPurpose()),
+			Purpose:  l.CredentialPurpose(),
 		})
-		dynCreds = append(dynCreds, session.NewDynamicCredential(l.Id(), credential.Purpose(l.CredentialPurpose())))
+		dynCreds = append(dynCreds, session.NewDynamicCredential(l.Id(), l.CredentialPurpose()))
 	}
 
 	expTime := timestamppb.Now()

internal/servers/controller/handlers/workers/worker_service.go

@@ -367,8 +367,7 @@ func (ws *workerServiceServer) ActivateSession(ctx context.Context, req *pbs.Act
 }
 
 func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs.AuthorizeConnectionRequest) (*pbs.AuthorizeConnectionResponse, error) {
-	const op = "workers.(workerServiceServer"
-
+	const op = "workers.(workerServiceServer).AuthorizeConnection"
 	sessRepo, err := ws.sessionRepoFn()
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
@@ -399,7 +398,6 @@ func (ws *workerServiceServer) AuthorizeConnection(ctx context.Context, req *pbs
 
 func (ws *workerServiceServer) ConnectConnection(ctx context.Context, req *pbs.ConnectConnectionRequest) (*pbs.ConnectConnectionResponse, error) {
 	const op = "workers.(workerServiceServer).ConnectConnection"
-
 	sessRepo, err := ws.sessionRepoFn()
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, "error getting session repo: %v", err)
@@ -419,25 +417,9 @@ func (ws *workerServiceServer) ConnectConnection(ctx context.Context, req *pbs.C
 		return nil, status.Error(codes.Internal, "Invalid connect connection response.")
 	}
 
-	ret := &pbs.ConnectConnectionResponse{
+	return &pbs.ConnectConnectionResponse{
 		Status: connStates[0].Status.ProtoVal(),
-	}
-
-	loggerPairs := []interface{}{
-		"session_id", connectionInfo.SessionId,
-		"connection_id", req.ConnectionId,
-		"client_tcp_address", req.ClientTcpAddress,
-		"client_tcp_port", req.ClientTcpPort,
-	}
-	switch req.GetType() {
-	case "tcp":
-		loggerPairs = append(loggerPairs,
-			"endpoint_tcp_address", connectionInfo.EndpointTcpAddress,
-			"endpoint_tcp_port", connectionInfo.EndpointTcpPort,
-		)
-	}
-
-	return ret, nil
+	}, nil
 }
 
 func (ws *workerServiceServer) CloseConnection(ctx context.Context, req *pbs.CloseConnectionRequest) (*pbs.CloseConnectionResponse, error) {

internal/servers/worker/common/common.go

@@ -0,0 +1,17 @@
+package common
+
+import "time"
+
+// In the future we could make this configurable
+const (
+	// StatusInterval is the base duration used in the calculation of the random backoff
+	// during the worker status report
+	StatusInterval = 2 * time.Second
+
+	// StatusTimeout is the timeout duration on status calls to the controller from
+	// the worker
+	StatusTimeout = 5 * time.Second
+)
+
+// CredentialData represents the secret data of the credential
+type CredentialData interface{}

internal/servers/worker/common/doc.go

@@ -0,0 +1,3 @@
+// Package common contains types and helper functions that are used across the different packages under
+// internal/server/worker.
+package common

internal/servers/worker/config.go

@@ -10,6 +10,6 @@ type Config struct {
 	// Workers
 	*base.Server
 	// The underlying configuration, passed in here to avoid duplicating values
-	// everwyehere
+	// everywhere
 	RawConfig *config.Config
 }

internal/servers/worker/controller_connection.go

@@ -59,7 +59,7 @@ func (w *Worker) startControllerConnections() error {
 	return nil
 }
 
-func (w Worker) controllerDialerFunc() func(context.Context, string) (net.Conn, error) {
+func (w *Worker) controllerDialerFunc() func(context.Context, string) (net.Conn, error) {
 	const op = "worker.(Worker).controllerDialerFunc"
 	return func(ctx context.Context, addr string) (net.Conn, error) {
 		tlsConf, authInfo, err := w.workerAuthTLSConfig()
@@ -130,11 +130,11 @@ func (w *Worker) createClientConn(addr string) error {
 	w.controllerStatusConn.Store(pbs.NewServerCoordinationServiceClient(cc))
 	w.controllerSessionConn.Store(pbs.NewSessionServiceClient(cc))
 
-	event.WriteSysEvent(context.TODO(), op, "connected to controller", "address", addr)
+	event.WriteSysEvent(w.baseContext, op, "connected to controller", "address", addr)
 	return nil
 }
 
-func (w Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error) {
+func (w *Worker) workerAuthTLSConfig() (*tls.Config, *base.WorkerAuthInfo, error) {
 	var err error
 	info := &base.WorkerAuthInfo{
 		Name:        w.conf.RawConfig.Worker.Name,

internal/servers/worker/handler.go

@@ -11,6 +11,8 @@ import (
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/proxy"
+	proxyHandlers "github.com/hashicorp/boundary/internal/servers/worker/proxy"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
 	"github.com/hashicorp/go-secure-stdlib/listenerutil"
 	"nhooyr.io/websocket"
 	"nhooyr.io/websocket/wspb"
@@ -20,7 +22,7 @@ type HandlerProperties struct {
 	ListenerConfig *listenerutil.ListenerConfig
 }
 
-// Handler returns an http.Handler for the API. This can be used on
+// Handler returns a http.Handler for the API. This can be used on
 // its own to mount the Worker API within another web server.
 func (w *Worker) handler(props HandlerProperties) http.Handler {
 	// Create the muxer to handle the actual endpoints
@@ -35,7 +37,7 @@ func (w *Worker) handler(props HandlerProperties) http.Handler {
 
 func (w *Worker) handleProxy() http.HandlerFunc {
 	const op = "worker.(Worker).handleProxy"
-	return http.HandlerFunc(func(wr http.ResponseWriter, r *http.Request) {
+	return func(wr http.ResponseWriter, r *http.Request) {
 		ctx := r.Context()
 		if r.TLS == nil {
 			event.WriteError(ctx, op, errors.New("no request TLS information found"))
@@ -67,14 +69,14 @@ func (w *Worker) handleProxy() http.HandlerFunc {
 			wr.WriteHeader(http.StatusInternalServerError)
 			return
 		}
-		si := siRaw.(*sessionInfo)
+		si := siRaw.(*session.Info)
 		si.RLock()
-		expiration := si.lookupSessionResponse.GetExpiration()
-		tofuToken := si.lookupSessionResponse.GetTofuToken()
-		version := si.lookupSessionResponse.GetVersion()
-		endpoint := si.lookupSessionResponse.GetEndpoint()
-		// userId := si.lookupSessionResponse.GetAuthorization()
-		sessStatus := si.status
+		expiration := si.LookupSessionResponse.GetExpiration()
+		tofuToken := si.LookupSessionResponse.GetTofuToken()
+		version := si.LookupSessionResponse.GetVersion()
+		endpoint := si.LookupSessionResponse.GetEndpoint()
+		// userId := si.LookupSessionResponse.GetAuthorization()
+		sessStatus := si.Status
 		si.RUnlock()
 
 		opts := &websocket.AcceptOptions{
@@ -89,73 +91,106 @@ func (w *Worker) handleProxy() http.HandlerFunc {
 		// Later calls will cause this to noop if they return a different status
 		defer conn.Close(websocket.StatusNormalClosure, "done")
 
-		connCtx, connCancel := context.WithDeadline(r.Context(), expiration.AsTime())
+		connCtx, connCancel := context.WithDeadline(ctx, expiration.AsTime())
 		defer connCancel()
 
+		sessClient, err := w.ControllerSessionConn()
+		if err != nil {
+			event.WriteError(ctx, op, err)
+			if err = conn.Close(websocket.StatusInternalError, "unable to get controller session client"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
+			return
+		}
+		workerId := w.conf.RawConfig.Worker.Name
+
 		var handshake proxy.ClientHandshake
 		if err := wspb.Read(connCtx, conn, &handshake); err != nil {
 			event.WriteError(ctx, op, err, event.WithInfoMsg("error reading handshake from client"))
-			conn.Close(websocket.StatusPolicyViolation, "invalid handshake received")
+			if err = conn.Close(websocket.StatusPolicyViolation, "invalid handshake received"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
 			return
 		}
 		if len(handshake.GetTofuToken()) < 20 {
 			event.WriteError(ctx, op, errors.New("invalid tofu token"))
-			conn.Close(websocket.StatusUnsupportedData, "invalid tofu token")
+			if err = conn.Close(websocket.StatusUnsupportedData, "invalid tofu token"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
 			return
 		}
 
 		if tofuToken != "" {
 			if tofuToken != handshake.GetTofuToken() {
 				event.WriteError(ctx, op, errors.New("WARNING: mismatched tofu token"), event.WithInfo("session_id", sessionId))
-				conn.Close(websocket.StatusPolicyViolation, "tofu token not allowed")
+				if err = conn.Close(websocket.StatusPolicyViolation, "tofu token not allowed"); err != nil {
+					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+				}
 				return
 			}
 		} else {
 			if sessStatus != pbs.SESSIONSTATUS_SESSIONSTATUS_PENDING {
 				event.WriteError(ctx, op, err, event.WithInfoMsg("no tofu token but not in correct session state"))
-				conn.Close(websocket.StatusInternalError, "refusing to activate session")
+				if err = conn.Close(websocket.StatusInternalError, "refusing to activate session"); err != nil {
+					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+				}
 				return
 			}
 			if handshake.Command == proxy.HANDSHAKECOMMAND_HANDSHAKECOMMAND_UNSPECIFIED {
-				sessStatus, err = w.activateSession(r.Context(), sessionId, handshake.GetTofuToken(), version)
+				sessStatus, err = session.Activate(ctx, sessClient, workerId, sessionId, handshake.GetTofuToken(), version)
 				if err != nil {
 					event.WriteError(ctx, op, err, event.WithInfoMsg("unable to validate session"))
-					conn.Close(websocket.StatusInternalError, "unable to activate session")
+					if err = conn.Close(websocket.StatusInternalError, "unable to activate session"); err != nil {
+						event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+					}
 					return
 				}
 			}
 		}
 
 		if handshake.Command == proxy.HANDSHAKECOMMAND_HANDSHAKECOMMAND_SESSION_CANCEL {
-			_, err := w.cancelSession(r.Context(), sessionId)
+			_, err := session.Cancel(ctx, sessClient, sessionId)
 			if err != nil {
 				event.WriteError(ctx, op, err, event.WithInfoMsg("unable to cancel session"))
-				conn.Close(websocket.StatusInternalError, "unable to cancel session")
+				if err = conn.Close(websocket.StatusInternalError, "unable to cancel session"); err != nil {
+					event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+				}
 				return
 			}
-			conn.Close(websocket.StatusNormalClosure, "session canceled")
+			if err = conn.Close(websocket.StatusNormalClosure, "session canceled"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
 			return
 		}
 
-		var ci *connInfo
+		// Verify the subprotocol has a supported proxy before calling AuthorizeConnection
+		handleProxyFn, err := proxyHandlers.GetHandler(conn.Subprotocol())
+		if err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("worker received request for unsupported protocol %s", conn.Subprotocol()))
+			if err = conn.Close(websocket.StatusProtocolError, "unsupported-protocol"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
+			return
+		}
+
+		var ci *session.ConnInfo
 		var connsLeft int32
-		ci, connsLeft, err = w.authorizeConnection(r.Context(), sessionId)
+		ci, connsLeft, err = session.AuthorizeConnection(ctx, sessClient, workerId, sessionId)
 		if err != nil {
 			event.WriteError(ctx, op, err, event.WithInfoMsg("unable to authorize connection"))
-			conn.Close(websocket.StatusInternalError, "unable to authorize connection")
+			if err = conn.Close(websocket.StatusInternalError, "unable to authorize connection"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
 			return
 		}
-
-		defer func() {
-			w.closeConnections(r.Context(), map[string]string{ci.id: si.id})
-		}()
+		defer session.CloseConnections(ctx, sessClient, w.sessionInfoMap, map[string]string{ci.Id: si.Id})
 
 		si.Lock()
-		ci.connCtx = connCtx
-		ci.connCancel = connCancel
-		si.connInfoMap[ci.id] = ci
-		si.status = sessStatus
-		connectionLimit := si.lookupSessionResponse.GetConnectionLimit()
+		ci.ConnCtx = connCtx
+		ci.ConnCancel = connCancel
+		si.ConnInfoMap[ci.Id] = ci
+		si.Status = sessStatus
+		connectionLimit := si.LookupSessionResponse.GetConnectionLimit()
 		si.Unlock()
 
 		handshakeResult := &proxy.HandshakeResult{
@@ -165,114 +200,37 @@ func (w *Worker) handleProxy() http.HandlerFunc {
 		}
 		if err := wspb.Write(connCtx, conn, handshakeResult); err != nil {
 			event.WriteError(ctx, op, err, event.WithInfoMsg("error sending handshake result to client"))
-			conn.Close(websocket.StatusProtocolError, "unable to send handshake result")
+			if err = conn.Close(websocket.StatusProtocolError, "unable to send handshake result"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
 			return
 		}
 
-		switch conn.Subprotocol() {
-		case globals.TcpProxyV1:
-			w.handleTcpProxyV1(connCtx, clientAddr, conn, si, ci.id, endpoint)
-		default:
-			conn.Close(websocket.StatusProtocolError, "unsupported-protocol")
+		conf := proxyHandlers.Config{
+			ClientAddress:  clientAddr,
+			ClientConn:     conn,
+			RemoteEndpoint: endpoint,
+			SessionClient:  sessClient,
+			SessionInfo:    si,
+			ConnectionId:   ci.Id,
+		}
+
+		if err := conf.Validate(); err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error validating proxy config"))
+			if err = conn.Close(websocket.StatusInternalError, "unable to validate proxy parameters"); err != nil {
+				event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+			}
 			return
 		}
-	})
+
+		handleProxyFn(connCtx, conf)
+	}
 }
 
-func (w *Worker) wrapGenericHandler(h http.Handler, props HandlerProperties) http.Handler {
+func (w *Worker) wrapGenericHandler(h http.Handler, _ HandlerProperties) http.Handler {
 	return http.HandlerFunc(func(wr http.ResponseWriter, r *http.Request) {
 		// Set the Cache-Control header for all responses returned
 		wr.Header().Set("Cache-Control", "no-store")
 		h.ServeHTTP(wr, r)
 	})
 }
-
-/*
-func WrapForwardedForHandler(h http.Handler, authorizedAddrs []*sockaddr.SockAddrMarshaler, rejectNotPresent, rejectNonAuthz bool, hopSkips int) http.Handler {
-	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		headers, headersOK := r.Header[textproto.CanonicalMIMEHeaderKey("X-Forwarded-For")]
-		if !headersOK || len(headers) == 0 {
-			if !rejectNotPresent {
-				h.ServeHTTP(w, r)
-				return
-			}
-			respondError(w, http.StatusBadRequest, fmt.Errorf("missing x-forwarded-for header and configured to reject when not present"))
-			return
-		}
-
-		host, port, err := net.SplitHostPort(r.RemoteAddr)
-		if err != nil {
-			// If not rejecting treat it like we just don't have a valid
-			// header because we can't do a comparison against an address we
-			// can't understand
-			if !rejectNotPresent {
-				h.ServeHTTP(w, r)
-				return
-			}
-			respondError(w, http.StatusBadRequest, errwrap.Wrapf("error parsing client hostport: {{err}}", err))
-			return
-		}
-
-		addr, err := sockaddr.NewIPAddr(host)
-		if err != nil {
-			// We treat this the same as the case above
-			if !rejectNotPresent {
-				h.ServeHTTP(w, r)
-				return
-			}
-			respondError(w, http.StatusBadRequest, errwrap.Wrapf("error parsing client address: {{err}}", err))
-			return
-		}
-
-		var found bool
-		for _, authz := range authorizedAddrs {
-			if authz.Contains(addr) {
-				found = true
-				break
-			}
-		}
-		if !found {
-			// If we didn't find it and aren't configured to reject, simply
-			// don't trust it
-			if !rejectNonAuthz {
-				h.ServeHTTP(w, r)
-				return
-			}
-			respondError(w, http.StatusBadRequest, fmt.Errorf("client address not authorized for x-forwarded-for and configured to reject connection"))
-			return
-		}
-
-		// At this point we have at least one value and it's authorized
-
-		// Split comma separated ones, which are common. This brings it in line
-		// to the multiple-header case.
-		var acc []string
-		for _, header := range headers {
-			vals := strings.Split(header, ",")
-			for _, v := range vals {
-				acc = append(acc, strings.TrimSpace(v))
-			}
-		}
-
-		indexToUse := len(acc) - 1 - hopSkips
-		if indexToUse < 0 {
-			// This is likely an error in either configuration or other
-			// infrastructure. We could either deny the request, or we
-			// could simply not trust the value. Denying the request is
-			// "safer" since if this logic is configured at all there may
-			// be an assumption it can always be trusted. Given that we can
-			// deny accepting the request at all if it's not from an
-			// authorized address, if we're at this point the address is
-			// authorized (or we've turned off explicit rejection) and we
-			// should assume that what comes in should be properly
-			// formatted.
-			respondError(w, http.StatusBadRequest, fmt.Errorf("malformed x-forwarded-for configuration or request, hops to skip (%d) would skip before earliest chain link (chain length %d)", hopSkips, len(headers)))
-			return
-		}
-
-		r.RemoteAddr = net.JoinHostPort(acc[indexToUse], port)
-		h.ServeHTTP(w, r)
-		return
-	})
-}
-*/

internal/servers/worker/listeners.go

@@ -18,13 +18,12 @@ import (
 
 func (w *Worker) startListeners() error {
 	const op = "worker.(Worker).startListeners"
-	ctx := context.TODO()
 	servers := make([]func(), 0, len(w.conf.Listeners))
 	e := event.SysEventer()
 	if e == nil {
 		return fmt.Errorf("%s: sys eventer not initialized", op)
 	}
-	logger, err := e.StandardLogger(ctx, "listeners", event.ErrorType)
+	logger, err := e.StandardLogger(w.baseContext, "listeners", event.ErrorType)
 	if err != nil {
 		return fmt.Errorf("%s: unable to initialize std logger: %w", op, err)
 	}
@@ -110,7 +109,7 @@ func (w *Worker) stopListeners() error {
 			defer shutdownKillCancel()
 
 			if localLn.HTTPServer != nil {
-				localLn.HTTPServer.Shutdown(shutdownKill)
+				_ = localLn.HTTPServer.Shutdown(shutdownKill)
 			}
 		}()
 	}

internal/servers/worker/package_registration.go

@@ -0,0 +1,5 @@
+package worker
+
+import (
+	_ "github.com/hashicorp/boundary/internal/servers/worker/proxy/tcp"
+)

internal/servers/worker/proxy/doc.go

@@ -0,0 +1,4 @@
+// Package proxy contains a collection of proxy handlers for the worker to call once a
+// connection has been authorized. Each proxy handler should mark the connection as
+// connected once the proxy has successfully been established.
+package proxy

internal/servers/worker/proxy/options.go

@@ -0,0 +1,11 @@
+package proxy
+
+// Option - how Options are passed as arguments.
+type Option func(*options)
+
+// options = how options are represented
+type options struct{}
+
+func getDefaultOptions() options {
+	return options{}
+}

internal/servers/worker/proxy/proxy.go

@@ -0,0 +1,81 @@
+package proxy
+
+import (
+	"context"
+	"errors"
+	"net"
+	"sync"
+
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
+	"nhooyr.io/websocket"
+)
+
+// Config provides the core parameters needed for a worker to create a proxy between
+// a provided ClientConn and the RemoteEndpoint, as well as the parameters to update
+// the connection in the connection repository.
+type Config struct {
+	ClientAddress  *net.TCPAddr
+	ClientConn     *websocket.Conn
+	RemoteEndpoint string
+
+	SessionClient pbs.SessionServiceClient
+	SessionInfo   *session.Info
+	ConnectionId  string
+}
+
+// Validate checks that the provided config is valid. If invalid, an error is returned
+// specifying the error.
+func (c Config) Validate() error {
+	switch {
+	case c.ClientAddress == nil:
+		return errors.New("missing client address")
+	case c.ClientConn == nil:
+		return errors.New("missing client connection")
+	case c.RemoteEndpoint == "":
+		return errors.New("missing remote endpoint")
+	case c.SessionClient == nil:
+		return errors.New("missing session client")
+	case c.SessionInfo == nil:
+		return errors.New("missing session info")
+	case c.ConnectionId == "":
+		return errors.New("missing connection id")
+	default:
+		return nil
+	}
+}
+
+// Handler is the type that all proxies need to implement to be called by the worker
+// when a new client connection is created.
+type Handler func(ctx context.Context, config Config, opt ...Option)
+
+var (
+	// handlers is the map of registered handlers
+	handlers sync.Map
+
+	// ErrUnknownProtocol specifies the provided protocol has no registered handler
+	ErrUnknownProtocol = errors.New("proxy: handler not found for protocol")
+
+	// ErrProtocolAlreadyRegistered specifies the provided protocol has already been registered
+	ErrProtocolAlreadyRegistered = errors.New("proxy: protocol already registered")
+)
+
+// RegisterHandler registers the handler to call for the protocol. The protocol is
+// negotiated when the connection was established to the worker.
+func RegisterHandler(protocol string, handler Handler) error {
+	_, loaded := handlers.LoadOrStore(protocol, handler)
+	if loaded {
+		return ErrProtocolAlreadyRegistered
+	}
+	return nil
+}
+
+// GetHandler returns the handler registered for the provided protocol. If the protocol
+// is not registered nil and ErrUnknownProtocol is returned.
+func GetHandler(protocol string) (Handler, error) {
+	handler, ok := handlers.Load(protocol)
+	if !ok {
+		return nil, ErrUnknownProtocol
+	}
+	return handler.(Handler), nil
+}

internal/servers/worker/proxy/proxy_test.go

@@ -0,0 +1,170 @@
+package proxy
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"nhooyr.io/websocket"
+)
+
+func TestConfigValidate(t *testing.T) {
+	t.Parallel()
+
+	// Create mock data for session management section of HandleTcpProxyV1
+	clientAddr := &net.TCPAddr{
+		IP:   net.ParseIP("127.0.0.1"),
+		Port: 50000,
+	}
+	sessClient := pbs.NewMockSessionServiceClient()
+	si := &session.Info{
+		Id: "one",
+	}
+	conn := &websocket.Conn{}
+
+	tests := []struct {
+		name       string
+		conf       Config
+		wantErr    bool
+		wantErrMsg string
+	}{
+		{
+			name: "missing-client-address",
+			conf: Config{
+				ClientConn:     conn,
+				RemoteEndpoint: "tcp://remote",
+				SessionClient:  sessClient,
+				SessionInfo:    si,
+				ConnectionId:   "connection-id",
+			},
+			wantErr:    true,
+			wantErrMsg: "missing client address",
+		},
+		{
+			name: "missing-client-conn",
+			conf: Config{
+				ClientAddress:  clientAddr,
+				RemoteEndpoint: "tcp://remote",
+				SessionClient:  sessClient,
+				SessionInfo:    si,
+				ConnectionId:   "connection-id",
+			},
+			wantErr:    true,
+			wantErrMsg: "missing client connection",
+		},
+		{
+			name: "missing-remote-endpoint",
+			conf: Config{
+				ClientAddress: clientAddr,
+				ClientConn:    conn,
+				SessionClient: sessClient,
+				SessionInfo:   si,
+				ConnectionId:  "connection-id",
+			},
+			wantErr:    true,
+			wantErrMsg: "missing remote endpoint",
+		},
+		{
+			name: "missing-session-client",
+			conf: Config{
+				ClientAddress:  clientAddr,
+				ClientConn:     conn,
+				RemoteEndpoint: "tcp://remote",
+				SessionInfo:    si,
+				ConnectionId:   "connection-id",
+			},
+			wantErr:    true,
+			wantErrMsg: "missing session client",
+		},
+		{
+			name: "missing-session-info",
+			conf: Config{
+				ClientAddress:  clientAddr,
+				ClientConn:     conn,
+				RemoteEndpoint: "tcp://remote",
+				SessionClient:  sessClient,
+				ConnectionId:   "connection-id",
+			},
+			wantErr:    true,
+			wantErrMsg: "missing session info",
+		},
+		{
+			name: "missing-connection-id",
+			conf: Config{
+				ClientAddress:  clientAddr,
+				ClientConn:     conn,
+				RemoteEndpoint: "tcp://remote",
+				SessionClient:  sessClient,
+				SessionInfo:    si,
+			},
+			wantErr:    true,
+			wantErrMsg: "missing connection id",
+		},
+		{
+			name: "valid",
+			conf: Config{
+				ClientAddress:  clientAddr,
+				ClientConn:     conn,
+				RemoteEndpoint: "tcp://remote",
+				SessionClient:  sessClient,
+				SessionInfo:    si,
+				ConnectionId:   "connection-id",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		tt := tt
+		t.Run(tt.name, func(t *testing.T) {
+			assert, require := assert.New(t), require.New(t)
+			err := tt.conf.Validate()
+			if tt.wantErr {
+				require.Error(err)
+				assert.Equal(tt.wantErrMsg, err.Error())
+				return
+			}
+			require.NoError(err)
+		})
+	}
+}
+
+func TestRegisterHandler(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+
+	fn := func(context.Context, Config, ...Option) { return }
+
+	err := RegisterHandler("protocol", fn)
+	require.NoError(err)
+
+	// Register function with same protocol
+	err = RegisterHandler("protocol", fn)
+	require.Error(err)
+	assert.ErrorIs(err, ErrProtocolAlreadyRegistered)
+
+	err = RegisterHandler("new-protocol", fn)
+	require.NoError(err)
+}
+
+func TestGetHandler(t *testing.T) {
+	t.Parallel()
+	assert, require := assert.New(t), require.New(t)
+
+	fn := func(context.Context, Config, ...Option) {}
+
+	err := RegisterHandler("fn", fn)
+	require.NoError(err)
+
+	gotFn, err := GetHandler("fake")
+	require.Error(err)
+	assert.ErrorIs(err, ErrUnknownProtocol)
+	assert.Nil(gotFn)
+
+	gotFn, err = GetHandler("fn")
+	require.NoError(err)
+	assert.NotNil(gotFn)
+}

internal/servers/worker/proxy/tcp/tcp.go

@@ -0,0 +1,105 @@
+package tcp
+
+import (
+	"context"
+	"io"
+	"net"
+	"net/url"
+	"sync"
+
+	"github.com/hashicorp/boundary/globals"
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
+	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/servers/worker/proxy"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
+	"nhooyr.io/websocket"
+)
+
+func init() {
+	err := proxy.RegisterHandler(globals.TcpProxyV1, handleTcpProxyV1)
+	if err != nil {
+		panic(err)
+	}
+}
+
+// handleTcpProxyV1 creates a tcp proxy between the incoming websocket conn and the
+// connection it creates with the remote endpoint. handleTcpProxyV1 sets the connectionId
+// as connected in the repository.
+//
+// handleTcpProxyV1 blocks until an error (EOF on happy path) is received on either
+// connection.
+func handleTcpProxyV1(ctx context.Context, conf proxy.Config, _ ...proxy.Option) {
+	const op = "tcp.HandleTcpProxyV1"
+	si := conf.SessionInfo
+	si.RLock()
+	sessionId := si.LookupSessionResponse.GetAuthorization().GetSessionId()
+	si.RUnlock()
+
+	conn := conf.ClientConn
+	sessionUrl, err := url.Parse(conf.RemoteEndpoint)
+	if err != nil {
+		event.WriteError(ctx, op, err, event.WithInfoMsg("error parsing endpoint information", "session_id", sessionId, "endpoint", conf.RemoteEndpoint))
+		if err = conn.Close(websocket.StatusInternalError, "cannot parse endpoint url"); err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+		}
+		return
+	}
+	if sessionUrl.Scheme != "tcp" {
+		event.WriteError(ctx, op, err, event.WithInfo("session_id", sessionId, "endpoint", conf.RemoteEndpoint))
+		if err = conn.Close(websocket.StatusInternalError, "invalid scheme for type"); err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+		}
+		return
+	}
+	remoteConn, err := net.Dial("tcp", sessionUrl.Host)
+	if err != nil {
+		event.WriteError(ctx, op, err, event.WithInfoMsg("error dialing endpoint", "endpoint", conf.RemoteEndpoint))
+		if err = conn.Close(websocket.StatusInternalError, "endpoint dialing failed"); err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+		}
+		return
+	}
+	// Assert this for better Go 1.11 splice support
+	tcpRemoteConn := remoteConn.(*net.TCPConn)
+
+	endpointAddr := tcpRemoteConn.RemoteAddr().(*net.TCPAddr)
+	connectionInfo := &pbs.ConnectConnectionRequest{
+		ConnectionId:       conf.ConnectionId,
+		ClientTcpAddress:   conf.ClientAddress.IP.String(),
+		ClientTcpPort:      uint32(conf.ClientAddress.Port),
+		EndpointTcpAddress: endpointAddr.IP.String(),
+		EndpointTcpPort:    uint32(endpointAddr.Port),
+		Type:               "tcp",
+	}
+
+	connStatus, err := session.ConnectConnection(ctx, conf.SessionClient, connectionInfo)
+	if err != nil {
+		event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connection as connected"))
+		if err = conn.Close(websocket.StatusInternalError, "failed to mark connection as connected"); err != nil {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error closing client connection"))
+		}
+		return
+	}
+	si.Lock()
+	si.ConnInfoMap[conf.ConnectionId].Status = connStatus
+	si.Unlock()
+
+	// Get a wrapped net.Conn so we can use io.Copy
+	netConn := websocket.NetConn(ctx, conn, websocket.MessageBinary)
+
+	connWg := new(sync.WaitGroup)
+	connWg.Add(2)
+	go func() {
+		defer connWg.Done()
+		_, _ = io.Copy(netConn, tcpRemoteConn)
+		_ = netConn.Close()
+		_ = tcpRemoteConn.Close()
+	}()
+	go func() {
+		defer connWg.Done()
+		_, _ = io.Copy(tcpRemoteConn, netConn)
+		_ = tcpRemoteConn.Close()
+		_ = netConn.Close()
+	}()
+	connWg.Wait()
+}

internal/servers/worker/proxy/tcp/tcp_test.go

@@ -0,0 +1,100 @@
+package tcp
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"testing"
+
+	"github.com/hashicorp/boundary/internal/gen/controller/api/resources/targets"
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
+	"github.com/hashicorp/boundary/internal/servers/worker/proxy"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
+	"github.com/hashicorp/boundary/sdk/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"nhooyr.io/websocket"
+)
+
+func TestHandleTcpProxyV1(t *testing.T) {
+	t.Parallel()
+	require, assert := require.New(t), assert.New(t)
+
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	clientConn, proxyConn := proxy.TestWsConn(t, ctx)
+	require.NotNil(clientConn)
+	require.NotNil(proxyConn)
+
+	port := testutil.TestFreePort(t)
+	l, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	require.NoError(err)
+	defer l.Close()
+
+	var endpointConn net.Conn
+	ready := make(chan struct{})
+	go func() {
+		endpointConn, err = l.Accept()
+		require.NoError(err)
+
+		defer endpointConn.Close()
+		ready <- struct{}{}
+
+		// block waiting for test to complete
+		<-ctx.Done()
+	}()
+
+	// Create mock data for session management section of HandleTcpProxyV1
+	clientAddr := &net.TCPAddr{
+		IP:   net.ParseIP("127.0.0.1"),
+		Port: 50000,
+	}
+	sessClient := pbs.NewMockSessionServiceClient()
+	si := &session.Info{
+		Id: "one",
+		LookupSessionResponse: &pbs.LookupSessionResponse{
+			Authorization: &targets.SessionAuthorizationData{
+				SessionId: "mock-session",
+			},
+		},
+		ConnInfoMap: map[string]*session.ConnInfo{
+			"mock-connection": {},
+		},
+	}
+
+	conf := proxy.Config{
+		ClientAddress:  clientAddr,
+		ClientConn:     proxyConn,
+		RemoteEndpoint: fmt.Sprintf("tcp://localhost:%d", port),
+		SessionClient:  sessClient,
+		SessionInfo:    si,
+		ConnectionId:   "mock-connection",
+	}
+
+	go handleTcpProxyV1(ctx, conf)
+
+	// wait for HandleTcpProxyV1 to dial endpoint
+	<-ready
+	netConn := websocket.NetConn(ctx, clientConn, websocket.MessageBinary)
+
+	// Write from endpoint to client
+	writeLen, err := endpointConn.Write([]byte("endpoint write to client via proxy"))
+	require.NoError(err)
+
+	b := make([]byte, writeLen)
+	readLen, err := netConn.Read(b)
+	require.NoError(err)
+	assert.Equal(writeLen, readLen)
+	assert.Equal("endpoint write to client via proxy", string(b))
+
+	// Write from client to endpoint
+	writeLen, err = netConn.Write([]byte("client write to endpoint via proxy"))
+	require.NoError(err)
+
+	b1 := make([]byte, writeLen)
+	readLen, err = endpointConn.Read(b1)
+	require.NoError(err)
+	assert.Equal(writeLen, readLen)
+	assert.Equal("client write to endpoint via proxy", string(b1))
+
+	cancelCtx()
+}

internal/servers/worker/proxy/testing.go

@@ -0,0 +1,50 @@
+package proxy
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/boundary/sdk/testutil"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"nhooyr.io/websocket"
+)
+
+// TestWsConn creates a websocket client and handler simulating the local websocket proxy
+// created during Boundary connect. The proxyConn returned should be used as the connection
+// passed into the worker proxy handler, while the clientConn can be used to simulate
+// the local end user connection.
+func TestWsConn(t *testing.T, ctx context.Context) (clientConn, proxyConn *websocket.Conn) {
+	t.Helper()
+	require, assert := require.New(t), assert.New(t)
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	port := testutil.TestFreePort(t)
+	go func() {
+		err := http.ListenAndServe(fmt.Sprintf(":%d", port), http.HandlerFunc(
+			func(w http.ResponseWriter, r *http.Request) {
+				var err error
+
+				proxyConn, err = websocket.Accept(w, r, nil)
+				require.NoError(err)
+
+				wg.Done()
+				// block waiting for test to complete
+				select {
+				case <-ctx.Done():
+				}
+			}))
+		assert.NoError(err)
+	}()
+
+	time.Sleep(time.Duration(1.5 * float64(time.Second)))
+	clientConn, _, err := websocket.Dial(ctx, fmt.Sprintf("ws://localhost:%d", port), nil)
+	require.NoError(err)
+	wg.Wait()
+	return
+}

internal/servers/worker/proxy/testing_test.go

@@ -0,0 +1,47 @@
+package proxy
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"nhooyr.io/websocket"
+)
+
+func Test_TestWsConn(t *testing.T) {
+	t.Parallel()
+	require, assert := require.New(t), assert.New(t)
+
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	clientConn, proxyConn := TestWsConn(t, ctx)
+
+	successfulRead := make(chan struct{})
+	go func() {
+		_, msg, err := proxyConn.Read(ctx)
+		require.NoError(err)
+		assert.Equal("client to proxy", string(msg))
+		successfulRead <- struct{}{}
+	}()
+
+	err := clientConn.Write(ctx, websocket.MessageBinary, []byte("client to proxy"))
+	require.NoError(err)
+
+	// Wait for read to verify success
+	<-successfulRead
+
+	go func() {
+		_, msg, err := clientConn.Read(ctx)
+		require.NoError(err)
+		assert.Equal("proxy to client", string(msg))
+		successfulRead <- struct{}{}
+	}()
+
+	err = proxyConn.Write(ctx, websocket.MessageBinary, []byte("proxy to client"))
+	require.NoError(err)
+
+	// Wait for read to verify success
+	<-successfulRead
+
+	cancelCtx()
+}

internal/servers/worker/session.go

@@ -1,422 +0,0 @@
-package worker
-
-import (
-	"context"
-	"crypto/ed25519"
-	"crypto/tls"
-	"crypto/x509"
-	"errors"
-	"fmt"
-	"strings"
-	"sync"
-	"time"
-
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-	"github.com/hashicorp/boundary/internal/observability/event"
-	"github.com/hashicorp/boundary/internal/session"
-)
-
-const validateSessionTimeout = 90 * time.Second
-
-var errMakeSessionCloseInfoNilCloseInfo = errors.New("nil closeInfo supplied to makeSessionCloseInfo, this is a bug, please report it")
-
-type connInfo struct {
-	id         string
-	connCtx    context.Context
-	connCancel context.CancelFunc
-	status     pbs.CONNECTIONSTATUS
-	closeTime  time.Time
-}
-
-type sessionInfo struct {
-	sync.RWMutex
-	id                    string
-	sessionTls            *tls.Config
-	status                pbs.SESSIONSTATUS
-	lookupSessionResponse *pbs.LookupSessionResponse
-	connInfoMap           map[string]*connInfo
-}
-
-func (w *Worker) getSessionTls(hello *tls.ClientHelloInfo) (*tls.Config, error) {
-	const op = "worker.(Worker).getSessionTls"
-	ctx := context.TODO()
-	var sessionId string
-	switch {
-	case strings.HasPrefix(hello.ServerName, "s_"):
-		sessionId = hello.ServerName
-	default:
-		event.WriteSysEvent(ctx, op, "invalid session in SNI", "session_id", hello.ServerName)
-		return nil, fmt.Errorf("could not find session ID in SNI")
-	}
-
-	rawConn := w.controllerSessionConn.Load()
-	if rawConn == nil {
-		event.WriteSysEvent(ctx, op, "could not get a controller client", "session_id", sessionId)
-		return nil, errors.New("could not get a controller client")
-	}
-	conn, ok := rawConn.(pbs.SessionServiceClient)
-	if !ok {
-		event.WriteSysEvent(ctx, op, "could not cast controller client to the real thing", "session_id", sessionId)
-		return nil, errors.New("could not cast atomic controller client to the real thing")
-	}
-	if conn == nil {
-		event.WriteSysEvent(ctx, op, "controller client is nil", "session_id", sessionId)
-		return nil, errors.New("controller client is nil")
-	}
-
-	timeoutContext, cancel := context.WithTimeout(w.baseContext, validateSessionTimeout)
-	defer cancel()
-
-	resp, err := conn.LookupSession(timeoutContext, &pbs.LookupSessionRequest{
-		ServerId:  w.conf.RawConfig.Worker.Name,
-		SessionId: sessionId,
-	})
-	if err != nil {
-		return nil, fmt.Errorf("error validating session: %w", err)
-	}
-
-	if resp.GetExpiration().AsTime().Before(time.Now()) {
-		return nil, fmt.Errorf("session is expired")
-	}
-
-	parsedCert, err := x509.ParseCertificate(resp.GetAuthorization().Certificate)
-	if err != nil {
-		return nil, fmt.Errorf("error parsing session certificate: %w", err)
-	}
-
-	if len(parsedCert.DNSNames) != 1 {
-		return nil, fmt.Errorf("invalid length of DNS names (%d) in parsed certificate", len(parsedCert.DNSNames))
-	}
-
-	certPool := x509.NewCertPool()
-	certPool.AddCert(parsedCert)
-
-	tlsConf := &tls.Config{
-		Certificates: []tls.Certificate{
-			{
-				Certificate: [][]byte{resp.GetAuthorization().Certificate},
-				PrivateKey:  ed25519.PrivateKey(resp.GetAuthorization().PrivateKey),
-				Leaf:        parsedCert,
-			},
-		},
-		ServerName: parsedCert.DNSNames[0],
-		ClientAuth: tls.RequireAndVerifyClientCert,
-		ClientCAs:  certPool,
-		MinVersion: tls.VersionTLS13,
-	}
-
-	si := &sessionInfo{
-		id:                    resp.GetAuthorization().GetSessionId(),
-		sessionTls:            tlsConf,
-		lookupSessionResponse: resp,
-		status:                resp.GetStatus(),
-		connInfoMap:           make(map[string]*connInfo),
-	}
-	// TODO: Periodicially clean this up. We can't rely on things in here but
-	// not in cancellation because they could be on the way to being
-	// established. However, since cert lifetimes are short, we can simply range
-	// through and remove values that are expired.
-	actualSiRaw, loaded := w.sessionInfoMap.LoadOrStore(sessionId, si)
-	if loaded {
-		// Update the response to the latest
-		actualSi := actualSiRaw.(*sessionInfo)
-		actualSi.Lock()
-		actualSi.lookupSessionResponse = resp
-		actualSi.Unlock()
-	}
-
-	return tlsConf, nil
-}
-
-func (w *Worker) activateSession(ctx context.Context, sessionId, tofuToken string, version uint32) (pbs.SESSIONSTATUS, error) {
-	rawConn := w.controllerSessionConn.Load()
-	if rawConn == nil {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, errors.New("could not get a controller client")
-	}
-	conn, ok := rawConn.(pbs.SessionServiceClient)
-	if !ok {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, errors.New("could not cast atomic controller client to the real thing")
-	}
-	if conn == nil {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, errors.New("controller client is nil")
-	}
-
-	resp, err := conn.ActivateSession(ctx, &pbs.ActivateSessionRequest{
-		SessionId: sessionId,
-		TofuToken: tofuToken,
-		Version:   version,
-		WorkerId:  w.conf.RawConfig.Worker.Name,
-	})
-	if err != nil {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, fmt.Errorf("error activating session: %w", err)
-	}
-	return resp.GetStatus(), nil
-}
-
-func (w *Worker) cancelSession(ctx context.Context, sessionId string) (pbs.SESSIONSTATUS, error) {
-	rawConn := w.controllerSessionConn.Load()
-	if rawConn == nil {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, errors.New("could not get a controller client")
-	}
-	conn, ok := rawConn.(pbs.SessionServiceClient)
-	if !ok {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, errors.New("could not cast atomic controller client to the real thing")
-	}
-	if conn == nil {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, errors.New("controller client is nil")
-	}
-
-	resp, err := conn.CancelSession(ctx, &pbs.CancelSessionRequest{
-		SessionId: sessionId,
-	})
-	if err != nil {
-		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, fmt.Errorf("error canceling session: %w", err)
-	}
-	return resp.GetStatus(), nil
-}
-
-func (w *Worker) authorizeConnection(ctx context.Context, sessionId string) (*connInfo, int32, error) {
-	rawConn := w.controllerSessionConn.Load()
-	if rawConn == nil {
-		return nil, 0, errors.New("could not get a controller client")
-	}
-	conn, ok := rawConn.(pbs.SessionServiceClient)
-	if !ok {
-		return nil, 0, errors.New("could not cast atomic controller client to the real thing")
-	}
-	if conn == nil {
-		return nil, 0, errors.New("controller client is nil")
-	}
-
-	resp, err := conn.AuthorizeConnection(ctx, &pbs.AuthorizeConnectionRequest{
-		SessionId: sessionId,
-		WorkerId:  w.conf.RawConfig.Worker.Name,
-	})
-	if err != nil {
-		return nil, 0, fmt.Errorf("error authorizing connection: %w", err)
-	}
-
-	return &connInfo{
-		id:     resp.ConnectionId,
-		status: resp.GetStatus(),
-	}, resp.GetConnectionsLeft(), nil
-}
-
-func (w *Worker) connectConnection(ctx context.Context, req *pbs.ConnectConnectionRequest) (pbs.CONNECTIONSTATUS, error) {
-	rawConn := w.controllerSessionConn.Load()
-	if rawConn == nil {
-		return pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_UNSPECIFIED, errors.New("could not get a controller client")
-	}
-	conn, ok := rawConn.(pbs.SessionServiceClient)
-	if !ok {
-		return pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_UNSPECIFIED, errors.New("could not cast atomic controller client to the real thing")
-	}
-	if conn == nil {
-		return pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_UNSPECIFIED, errors.New("controller client is nil")
-	}
-
-	resp, err := conn.ConnectConnection(ctx, req)
-	if err != nil {
-		return pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_UNSPECIFIED, err
-	}
-
-	if resp.GetStatus() != pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED {
-		return pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_UNSPECIFIED, fmt.Errorf("unexpected state returned: %v", resp.GetStatus().String())
-	}
-
-	return resp.GetStatus(), nil
-}
-
-func (w *Worker) closeConnection(ctx context.Context, req *pbs.CloseConnectionRequest) (*pbs.CloseConnectionResponse, error) {
-	const op = "worker.(Worker).closeConnections"
-	rawConn := w.controllerSessionConn.Load()
-	if rawConn == nil {
-		return nil, errors.New("could not get a controller client")
-	}
-	conn, ok := rawConn.(pbs.SessionServiceClient)
-	if !ok {
-		return nil, errors.New("could not cast atomic controller client to the real thing")
-	}
-	if conn == nil {
-		return nil, errors.New("controller client is nil")
-	}
-
-	resp, err := conn.CloseConnection(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-	if len(resp.GetCloseResponseData()) != len(req.GetCloseRequestData()) {
-		event.WriteError(ctx, op, errors.New("mismatched number of states returned on connection closed"), event.WithInfo("expected", len(req.GetCloseRequestData()), "got", len(resp.GetCloseResponseData())))
-	}
-
-	return resp, nil
-}
-
-// closeConnections is a helper worker function that sends connection
-// close requests to the controller, and sets close times within the
-// worker. It is called during the worker status loop and on
-// connection exit on the proxy.
-//
-// closeInfo is a map of connections mapped to their individual
-// session.
-func (w *Worker) closeConnections(ctx context.Context, closeInfo map[string]string) {
-	const op = "worker.(Worker).closeConnections"
-	if closeInfo == nil {
-		// This should not happen, but it's a no-op if it does. Just
-		// return.
-		return
-	}
-
-	// How we handle close info depends on whether or not we succeeded with
-	// marking them closed on the controller.
-	var sessionCloseInfo map[string][]*pbs.CloseConnectionResponseData
-	var err error
-
-	// TODO: This, along with the status call to the controller, probably needs a
-	// bit of formalization in terms of how we handle timeouts. For now, this
-	// just ensures consistency with the same status call in that it times out
-	// within an adequate period of time.
-	closeConnCtx, closeConnCancel := context.WithTimeout(ctx, statusTimeout)
-	defer closeConnCancel()
-	response, err := w.closeConnection(closeConnCtx, w.makeCloseConnectionRequest(closeInfo))
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connections closed",
-			"warning", "error contacting controller, connections will be closed only on worker",
-			"session_and_connection_ids", fmt.Sprintf("%#v", closeInfo),
-		))
-
-		// Since we could not reach the controller, we have to make a "fake" response set.
-		sessionCloseInfo, err = w.makeFakeSessionCloseInfo(closeInfo)
-	} else {
-		// Connection succeeded, so we can proceed with making the sessionCloseInfo
-		// off of the response data.
-		sessionCloseInfo, err = w.makeSessionCloseInfo(closeInfo, response)
-	}
-
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("serious error in processing return data from controller, aborting additional session/connection state modification"))
-		return
-	}
-
-	// Mark connections as closed
-	_, errs := w.setCloseTimeForResponse(sessionCloseInfo)
-	if len(errs) > 0 {
-		for _, err := range errs {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connection closed in state"))
-		}
-	}
-
-}
-
-// makeCloseConnectionRequest creates a CloseConnectionRequest for
-// use with closing connections.
-//
-// closeInfo is a map, indexed by connection ID, to the individual
-// sessions IDs that those connections belong to. The values are
-// ignored; the parameter is expected as such just for convenience of
-// its caller.
-func (w *Worker) makeCloseConnectionRequest(closeInfo map[string]string) *pbs.CloseConnectionRequest {
-	closeData := make([]*pbs.CloseConnectionRequestData, 0, len(closeInfo))
-	for connId := range closeInfo {
-		closeData = append(closeData, &pbs.CloseConnectionRequestData{
-			ConnectionId: connId,
-			Reason:       session.UnknownReason.String(),
-		})
-	}
-
-	return &pbs.CloseConnectionRequest{
-		CloseRequestData: closeData,
-	}
-}
-
-// makeSessionCloseInfo takes the response from CloseConnections and
-// our original closeInfo map and makes a map of slices, indexed by
-// session ID, of all of the connection responses. This allows us to
-// easily lock on session once for all connections in
-// setCloseTimeForResponse.
-func (w *Worker) makeSessionCloseInfo(
-	closeInfo map[string]string,
-	response *pbs.CloseConnectionResponse,
-) (map[string][]*pbs.CloseConnectionResponseData, error) {
-	if closeInfo == nil {
-		return nil, errMakeSessionCloseInfoNilCloseInfo
-	}
-
-	result := make(map[string][]*pbs.CloseConnectionResponseData)
-	for _, v := range response.GetCloseResponseData() {
-		result[closeInfo[v.GetConnectionId()]] = append(result[closeInfo[v.GetConnectionId()]], v)
-	}
-
-	return result, nil
-}
-
-// makeFakeSessionCloseInfo makes a "fake" makeFakeSessionCloseInfo, intended
-// for use when we can't contact the controller.
-func (w *Worker) makeFakeSessionCloseInfo(
-	closeInfo map[string]string,
-) (map[string][]*pbs.CloseConnectionResponseData, error) {
-	if closeInfo == nil {
-		return nil, errMakeSessionCloseInfoNilCloseInfo
-	}
-
-	result := make(map[string][]*pbs.CloseConnectionResponseData)
-	for connectionId, sessionId := range closeInfo {
-		result[sessionId] = append(result[sessionId], &pbs.CloseConnectionResponseData{
-			ConnectionId: connectionId,
-			Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED,
-		})
-	}
-
-	return result, nil
-}
-
-// setCloseTimeForResponse iterates a CloseConnectionResponse and
-// sets the close time for any connection found to be closed to the
-// current time.
-//
-// sessionCloseInfo can be derived from the closeInfo supplied to
-// makeCloseConnectionRequest through reverseCloseInfo, which creates
-// a session ID to connection ID mapping.
-//
-// A non-zero error count does not necessarily mean the operation
-// failed, as some connections may have been marked as closed. The
-// actual list of connection IDs closed is returned as the first
-// return value.
-func (w *Worker) setCloseTimeForResponse(sessionCloseInfo map[string][]*pbs.CloseConnectionResponseData) ([]string, []error) {
-	closedIds := make([]string, 0)
-	var errors []error
-	for sessionId, responses := range sessionCloseInfo {
-		siRaw, ok := w.sessionInfoMap.Load(sessionId)
-		if !ok {
-			errors = append(errors, fmt.Errorf("could not find session ID %q in local state after closing connections", sessionId))
-			continue
-		}
-
-		si := siRaw.(*sessionInfo)
-		si.Lock()
-
-		for _, response := range responses {
-			ci, ok := si.connInfoMap[response.GetConnectionId()]
-			if !ok {
-				errors = append(errors,
-					fmt.Errorf(
-						"could not find connection ID %q for session ID %q in local state after closing connections",
-						response.GetConnectionId(),
-						sessionId,
-					),
-				)
-				continue
-			}
-
-			ci.status = response.GetStatus()
-			if ci.status == pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED {
-				ci.closeTime = time.Now()
-				closedIds = append(closedIds, ci.id)
-			}
-		}
-
-		si.Unlock()
-	}
-
-	return closedIds, errors
-}

internal/servers/worker/session/session.go

@@ -0,0 +1,282 @@
+package session
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
+	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/servers/worker/common"
+	"github.com/hashicorp/boundary/internal/session"
+)
+
+// ValidateSessionTimeout is the duration of the timeout when the worker queries the
+// controller for the sessionId for which the connection is being requested.
+const ValidateSessionTimeout = 90 * time.Second
+
+var errMakeSessionCloseInfoNilCloseInfo = errors.New("nil closeInfo supplied to makeSessionCloseInfo, this is a bug, please report it")
+
+// ConnInfo defines the information about a connection attached to a session
+type ConnInfo struct {
+	Id         string
+	ConnCtx    context.Context
+	ConnCancel context.CancelFunc
+	Status     pbs.CONNECTIONSTATUS
+	CloseTime  time.Time
+}
+
+// Info defines the information about a session
+type Info struct {
+	sync.RWMutex
+	Id                    string
+	SessionTls            *tls.Config
+	Status                pbs.SESSIONSTATUS
+	LookupSessionResponse *pbs.LookupSessionResponse
+	ConnInfoMap           map[string]*ConnInfo
+}
+
+// Activate is a helper worker function that sends session activation request to the
+// controller.
+func Activate(ctx context.Context, sessClient pbs.SessionServiceClient, workerId, sessionId, tofuToken string, version uint32) (pbs.SESSIONSTATUS, error) {
+	resp, err := sessClient.ActivateSession(ctx, &pbs.ActivateSessionRequest{
+		SessionId: sessionId,
+		TofuToken: tofuToken,
+		Version:   version,
+		WorkerId:  workerId,
+	})
+	if err != nil {
+		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, fmt.Errorf("error activating session: %w", err)
+	}
+	return resp.GetStatus(), nil
+}
+
+// Cancel is a helper worker function that sends session cancellation request to the
+// controller.
+func Cancel(ctx context.Context, sessClient pbs.SessionServiceClient, sessionId string) (pbs.SESSIONSTATUS, error) {
+	resp, err := sessClient.CancelSession(ctx, &pbs.CancelSessionRequest{
+		SessionId: sessionId,
+	})
+	if err != nil {
+		return pbs.SESSIONSTATUS_SESSIONSTATUS_UNSPECIFIED, fmt.Errorf("error canceling session: %w", err)
+	}
+	return resp.GetStatus(), nil
+}
+
+// AuthorizeConnection is a helper worker function that sends connection
+// authorization request to the controller. It is called by the worker handler after a
+// connection has been received by the worker, and the session has been validated.
+func AuthorizeConnection(ctx context.Context, sessClient pbs.SessionServiceClient, workerId, sessionId string) (*ConnInfo, int32, error) {
+	resp, err := sessClient.AuthorizeConnection(ctx, &pbs.AuthorizeConnectionRequest{
+		SessionId: sessionId,
+		WorkerId:  workerId,
+	})
+	if err != nil {
+		return nil, 0, fmt.Errorf("error authorizing connection: %w", err)
+	}
+
+	return &ConnInfo{
+		Id:     resp.ConnectionId,
+		Status: resp.GetStatus(),
+	}, resp.GetConnectionsLeft(), nil
+}
+
+// ConnectConnection is a helper worker function that sends connection
+// connect request to the controller. It is called by the worker handler after a
+// connection has been authorized.
+func ConnectConnection(ctx context.Context, sessClient pbs.SessionServiceClient, req *pbs.ConnectConnectionRequest) (pbs.CONNECTIONSTATUS, error) {
+	resp, err := sessClient.ConnectConnection(ctx, req)
+	if err != nil {
+		return pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_UNSPECIFIED, err
+	}
+
+	if resp.GetStatus() != pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CONNECTED {
+		return pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_UNSPECIFIED, fmt.Errorf("unexpected state returned: %v", resp.GetStatus().String())
+	}
+
+	return resp.GetStatus(), nil
+}
+
+func closeConnection(ctx context.Context, sessClient pbs.SessionServiceClient, req *pbs.CloseConnectionRequest) (*pbs.CloseConnectionResponse, error) {
+	const op = "session.closeConnection"
+	resp, err := sessClient.CloseConnection(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	if len(resp.GetCloseResponseData()) != len(req.GetCloseRequestData()) {
+		event.WriteError(ctx, op, errors.New("mismatched number of states returned on connection closed"), event.WithInfo("expected", len(req.GetCloseRequestData()), "got", len(resp.GetCloseResponseData())))
+	}
+
+	return resp, nil
+}
+
+// CloseConnections is a helper worker function that sends connection
+// close requests to the controller, and sets close times within the
+// worker. It is called during the worker status loop and on
+// connection exit on the proxy.
+//
+// closeInfo is a map of connections mapped to their individual
+// session.
+func CloseConnections(ctx context.Context, sessClient pbs.SessionServiceClient, sessionInfo *sync.Map, closeInfo map[string]string) {
+	const op = "session.CloseConnections"
+	if closeInfo == nil {
+		// This should not happen, but it's a no-op if it does. Just
+		// return.
+		return
+	}
+
+	// How we handle close info depends on whether or not we succeeded with
+	// marking them closed on the controller.
+	var sessionCloseInfo map[string][]*pbs.CloseConnectionResponseData
+	var err error
+
+	// TODO: This, along with the status call to the controller, probably needs a
+	// bit of formalization in terms of how we handle timeouts. For now, this
+	// just ensures consistency with the same status call in that it times out
+	// within an adequate period of time.
+	closeConnCtx, closeConnCancel := context.WithTimeout(ctx, common.StatusTimeout)
+	defer closeConnCancel()
+	response, err := closeConnection(closeConnCtx, sessClient, makeCloseConnectionRequest(closeInfo))
+	if err != nil {
+		event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connections closed",
+			"warning", "error contacting controller, connections will be closed only on worker",
+			"session_and_connection_ids", fmt.Sprintf("%#v", closeInfo),
+		))
+
+		// Since we could not reach the controller, we have to make a "fake" response set.
+		sessionCloseInfo, err = makeFakeSessionCloseInfo(closeInfo)
+	} else {
+		// Connection succeeded, so we can proceed with making the sessionCloseInfo
+		// off of the response data.
+		sessionCloseInfo, err = makeSessionCloseInfo(closeInfo, response)
+	}
+
+	if err != nil {
+		event.WriteError(ctx, op, err, event.WithInfoMsg("serious error in processing return data from controller, aborting additional session/connection state modification"))
+		return
+	}
+
+	// Mark connections as closed
+	_, errs := setCloseTimeForResponse(sessionInfo, sessionCloseInfo)
+	if len(errs) > 0 {
+		for _, err := range errs {
+			event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connection closed in state"))
+		}
+	}
+}
+
+// makeCloseConnectionRequest creates a CloseConnectionRequest for
+// use with closing connections.
+//
+// closeInfo is a map, indexed by connection ID, to the individual
+// sessions IDs that those connections belong to. The values are
+// ignored; the parameter is expected as such just for convenience of
+// its caller.
+func makeCloseConnectionRequest(closeInfo map[string]string) *pbs.CloseConnectionRequest {
+	closeData := make([]*pbs.CloseConnectionRequestData, 0, len(closeInfo))
+	for connId := range closeInfo {
+		closeData = append(closeData, &pbs.CloseConnectionRequestData{
+			ConnectionId: connId,
+			Reason:       session.UnknownReason.String(),
+		})
+	}
+
+	return &pbs.CloseConnectionRequest{
+		CloseRequestData: closeData,
+	}
+}
+
+// makeSessionCloseInfo takes the response from CloseConnections and
+// our original closeInfo map and makes a map of slices, indexed by
+// session ID, of all of the connection responses. This allows us to
+// easily lock on session once for all connections in
+// setCloseTimeForResponse.
+func makeSessionCloseInfo(
+	closeInfo map[string]string,
+	response *pbs.CloseConnectionResponse,
+) (map[string][]*pbs.CloseConnectionResponseData, error) {
+	if closeInfo == nil {
+		return nil, errMakeSessionCloseInfoNilCloseInfo
+	}
+
+	result := make(map[string][]*pbs.CloseConnectionResponseData)
+	for _, v := range response.GetCloseResponseData() {
+		result[closeInfo[v.GetConnectionId()]] = append(result[closeInfo[v.GetConnectionId()]], v)
+	}
+
+	return result, nil
+}
+
+// makeFakeSessionCloseInfo makes a "fake" makeFakeSessionCloseInfo, intended
+// for use when we can't contact the controller.
+func makeFakeSessionCloseInfo(
+	closeInfo map[string]string,
+) (map[string][]*pbs.CloseConnectionResponseData, error) {
+	if closeInfo == nil {
+		return nil, errMakeSessionCloseInfoNilCloseInfo
+	}
+
+	result := make(map[string][]*pbs.CloseConnectionResponseData)
+	for connectionId, sessionId := range closeInfo {
+		result[sessionId] = append(result[sessionId], &pbs.CloseConnectionResponseData{
+			ConnectionId: connectionId,
+			Status:       pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED,
+		})
+	}
+
+	return result, nil
+}
+
+// setCloseTimeForResponse iterates a CloseConnectionResponse and
+// sets the close time for any connection found to be closed to the
+// current time.
+//
+// sessionCloseInfo can be derived from the closeInfo supplied to
+// makeCloseConnectionRequest through reverseCloseInfo, which creates
+// a session ID to connection ID mapping.
+//
+// A non-zero error count does not necessarily mean the operation
+// failed, as some connections may have been marked as closed. The
+// actual list of connection IDs closed is returned as the first
+// return value.
+func setCloseTimeForResponse(sessionInfo *sync.Map, sessionCloseInfo map[string][]*pbs.CloseConnectionResponseData) ([]string, []error) {
+	closedIds := make([]string, 0)
+	var errors []error
+	for sessionId, responses := range sessionCloseInfo {
+		siRaw, ok := sessionInfo.Load(sessionId)
+		if !ok {
+			errors = append(errors, fmt.Errorf("could not find session ID %q in local state after closing connections", sessionId))
+			continue
+		}
+
+		si := siRaw.(*Info)
+		si.Lock()
+
+		for _, response := range responses {
+			ci, ok := si.ConnInfoMap[response.GetConnectionId()]
+			if !ok {
+				errors = append(errors,
+					fmt.Errorf(
+						"could not find connection ID %q for session ID %q in local state after closing connections",
+						response.GetConnectionId(),
+						sessionId,
+					),
+				)
+				continue
+			}
+
+			ci.Status = response.GetStatus()
+			if ci.Status == pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED {
+				ci.CloseTime = time.Now()
+				closedIds = append(closedIds, ci.Id)
+			}
+		}
+
+		si.Unlock()
+	}
+
+	return closedIds, errors
+}

internal/servers/worker/session/session_test.go

@@ -1,4 +1,4 @@
-package worker
+package session
 
 import (
 	"errors"
@@ -20,7 +20,7 @@ func TestWorkerMakeCloseConnectionRequest(t *testing.T) {
 			{ConnectionId: "bar", Reason: session.UnknownReason.String()},
 		},
 	}
-	actual := new(Worker).makeCloseConnectionRequest(in)
+	actual := makeCloseConnectionRequest(in)
 	require.ElementsMatch(expected.GetCloseRequestData(), actual.GetCloseRequestData())
 }
 
@@ -41,21 +41,21 @@ func TestMakeSessionCloseInfo(t *testing.T) {
 			{ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED},
 		},
 	}
-	actual, err := new(Worker).makeSessionCloseInfo(closeInfo, response)
+	actual, err := makeSessionCloseInfo(closeInfo, response)
 	require.NoError(err)
 	require.Equal(expected, actual)
 }
 
 func TestMakeSessionCloseInfoErrorIfCloseInfoNil(t *testing.T) {
 	require := require.New(t)
-	actual, err := new(Worker).makeSessionCloseInfo(nil, nil)
+	actual, err := makeSessionCloseInfo(nil, nil)
 	require.Nil(actual)
 	require.ErrorIs(err, errMakeSessionCloseInfoNilCloseInfo)
 }
 
 func TestMakeSessionCloseInfoEmpty(t *testing.T) {
 	require := require.New(t)
-	actual, err := new(Worker).makeSessionCloseInfo(make(map[string]string), nil)
+	actual, err := makeSessionCloseInfo(make(map[string]string), nil)
 	require.NoError(err)
 	require.Equal(
 		make(map[string][]*pbs.CloseConnectionResponseData),
@@ -74,21 +74,21 @@ func TestMakeFakeSessionCloseInfo(t *testing.T) {
 			{ConnectionId: "bar", Status: pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED},
 		},
 	}
-	actual, err := new(Worker).makeFakeSessionCloseInfo(closeInfo)
+	actual, err := makeFakeSessionCloseInfo(closeInfo)
 	require.NoError(err)
 	require.Equal(expected, actual)
 }
 
 func TestMakeFakeSessionCloseInfoErrorIfCloseInfoNil(t *testing.T) {
 	require := require.New(t)
-	actual, err := new(Worker).makeFakeSessionCloseInfo(nil)
+	actual, err := makeFakeSessionCloseInfo(nil)
 	require.Nil(actual)
 	require.ErrorIs(err, errMakeSessionCloseInfoNilCloseInfo)
 }
 
 func TestMakeFakeSessionCloseInfoEmpty(t *testing.T) {
 	require := require.New(t)
-	actual, err := new(Worker).makeFakeSessionCloseInfo(make(map[string]string))
+	actual, err := makeFakeSessionCloseInfo(make(map[string]string))
 	require.NoError(err)
 	require.Equal(
 		make(map[string][]*pbs.CloseConnectionResponseData),
@@ -117,22 +117,22 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 			},
 			sessionInfoMap: func() *sync.Map {
 				m := new(sync.Map)
-				m.Store("one", &sessionInfo{
-					id: "one",
-					connInfoMap: map[string]*connInfo{
-						"foo": {id: "foo"},
+				m.Store("one", &Info{
+					Id: "one",
+					ConnInfoMap: map[string]*ConnInfo{
+						"foo": {Id: "foo"},
 					},
 				})
-				m.Store("two", &sessionInfo{
-					id: "two",
-					connInfoMap: map[string]*connInfo{
-						"bar": {id: "bar"},
+				m.Store("two", &Info{
+					Id: "two",
+					ConnInfoMap: map[string]*ConnInfo{
+						"bar": {Id: "bar"},
 					},
 				})
-				m.Store("three", &sessionInfo{
-					id: "three",
-					connInfoMap: map[string]*connInfo{
-						"baz": {id: "baz"},
+				m.Store("three", &Info{
+					Id: "three",
+					ConnInfoMap: map[string]*ConnInfo{
+						"baz": {Id: "baz"},
 					},
 				})
 
@@ -156,16 +156,16 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 			},
 			sessionInfoMap: func() *sync.Map {
 				m := new(sync.Map)
-				m.Store("one", &sessionInfo{
-					id: "one",
-					connInfoMap: map[string]*connInfo{
-						"foo": {id: "foo"},
+				m.Store("one", &Info{
+					Id: "one",
+					ConnInfoMap: map[string]*ConnInfo{
+						"foo": {Id: "foo"},
 					},
 				})
-				m.Store("two", &sessionInfo{
-					id: "two",
-					connInfoMap: map[string]*connInfo{
-						"bar": {id: "bar"},
+				m.Store("two", &Info{
+					Id: "two",
+					ConnInfoMap: map[string]*ConnInfo{
+						"bar": {Id: "bar"},
 					},
 				})
 
@@ -188,10 +188,10 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 			},
 			sessionInfoMap: func() *sync.Map {
 				m := new(sync.Map)
-				m.Store("one", &sessionInfo{
-					id: "one",
-					connInfoMap: map[string]*connInfo{
-						"foo": {id: "foo"},
+				m.Store("one", &Info{
+					Id: "one",
+					ConnInfoMap: map[string]*ConnInfo{
+						"foo": {Id: "foo"},
 					},
 				})
 
@@ -217,13 +217,13 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 			},
 			sessionInfoMap: func() *sync.Map {
 				m := new(sync.Map)
-				m.Store("one", &sessionInfo{
-					id: "one",
-					connInfoMap: map[string]*connInfo{
-						"foo": {id: "foo"},
+				m.Store("one", &Info{
+					Id: "one",
+					ConnInfoMap: map[string]*ConnInfo{
+						"foo": {Id: "foo"},
 					},
 				})
-				m.Store("two", &sessionInfo{id: "two"})
+				m.Store("two", &Info{Id: "two"})
 
 				return m
 			},
@@ -240,10 +240,10 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 			sessionCloseInfo: make(map[string][]*pbs.CloseConnectionResponseData),
 			sessionInfoMap: func() *sync.Map {
 				m := new(sync.Map)
-				m.Store("one", &sessionInfo{
-					id: "one",
-					connInfoMap: map[string]*connInfo{
-						"foo": {id: "foo"},
+				m.Store("one", &Info{
+					Id: "one",
+					ConnInfoMap: map[string]*ConnInfo{
+						"foo": {Id: "foo"},
 					},
 				})
 
@@ -258,19 +258,17 @@ func TestWorkerSetCloseTimeForResponse(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			require := require.New(t)
-			w := &Worker{
-				sessionInfoMap: tc.sessionInfoMap(),
-			}
-			actual, actualErr := w.setCloseTimeForResponse(tc.sessionCloseInfo)
+			sessionInfoMap := tc.sessionInfoMap()
+			actual, actualErr := setCloseTimeForResponse(sessionInfoMap, tc.sessionCloseInfo)
 
 			// Assert all close times were set
-			w.sessionInfoMap.Range(func(key, value interface{}) bool {
+			sessionInfoMap.Range(func(key, value interface{}) bool {
 				t.Helper()
-				for _, ci := range value.(*sessionInfo).connInfoMap {
-					if _, ok := tc.expectedClosed[ci.id]; ok {
-						require.NotEqual(time.Time{}, ci.closeTime)
+				for _, ci := range value.(*Info).ConnInfoMap {
+					if _, ok := tc.expectedClosed[ci.Id]; ok {
+						require.NotEqual(time.Time{}, ci.CloseTime)
 					} else {
-						require.Equal(time.Time{}, ci.closeTime)
+						require.Equal(time.Time{}, ci.CloseTime)
 					}
 				}
 

internal/servers/worker/status.go

@@ -9,16 +9,12 @@ import (
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/servers"
+	"github.com/hashicorp/boundary/internal/servers/worker/common"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
 	"github.com/hashicorp/boundary/internal/types/resource"
 	"google.golang.org/grpc/resolver"
 )
 
-// In the future we could make this configurable
-const (
-	statusInterval = 2 * time.Second
-	statusTimeout  = 5 * time.Second
-)
-
 type LastStatusInformation struct {
 	*pbs.StatusResponse
 	StatusTime time.Time
@@ -28,7 +24,7 @@ func (w *Worker) startStatusTicking(cancelCtx context.Context) {
 	const op = "worker.(Worker).startStatusTicking"
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	// This function exists to desynchronize calls to controllers from
-	// workers so we aren't always getting status updates at the exact same
+	// workers, so we aren't always getting status updates at the exact same
 	// intervals, to ease the load on the DB.
 	getRandomInterval := func() time.Duration {
 		// 0 to 0.5 adjustment to the base
@@ -37,14 +33,14 @@ func (w *Worker) startStatusTicking(cancelCtx context.Context) {
 		if r.Float32() > 0.5 {
 			f = -1 * f
 		}
-		return statusInterval + time.Duration(f*float64(time.Second))
+		return common.StatusInterval + time.Duration(f*float64(time.Second))
 	}
 
 	timer := time.NewTimer(0)
 	for {
 		select {
 		case <-cancelCtx.Done():
-			event.WriteSysEvent(context.TODO(), op, "status ticking shutting down")
+			event.WriteSysEvent(w.baseContext, op, "status ticking shutting down")
 			return
 
 		case <-timer.C:
@@ -102,14 +98,14 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context) {
 	w.sessionInfoMap.Range(func(key, value interface{}) bool {
 		var jobInfo pbs.SessionJobInfo
 		sessionId := key.(string)
-		si := value.(*sessionInfo)
+		si := value.(*session.Info)
 		si.RLock()
-		status := si.status
-		connections := make([]*pbs.Connection, 0, len(si.connInfoMap))
-		for k, v := range si.connInfoMap {
+		status := si.Status
+		connections := make([]*pbs.Connection, 0, len(si.ConnInfoMap))
+		for k, v := range si.ConnInfoMap {
 			connections = append(connections, &pbs.Connection{
 				ConnectionId: k,
-				Status:       v.status,
+				Status:       v.Status,
 			})
 		}
 		si.RUnlock()
@@ -137,7 +133,7 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context) {
 	if w.updateTags.Load() {
 		tags = w.tags.Load().(map[string]*servers.TagValues)
 	}
-	statusCtx, statusCancel := context.WithTimeout(cancelCtx, statusTimeout)
+	statusCtx, statusCancel := context.WithTimeout(cancelCtx, common.StatusTimeout)
 	defer statusCancel()
 	result, err := client.Status(statusCtx, &pbs.StatusRequest{
 		Jobs: activeJobs,
@@ -199,20 +195,20 @@ func (w *Worker) sendWorkerStatus(cancelCtx context.Context) {
 						event.WriteError(statusCtx, op, errors.New("session change requested but could not find local information for it"), event.WithInfo("session_id", sessionId))
 						continue
 					}
-					si := siRaw.(*sessionInfo)
+					si := siRaw.(*session.Info)
 					si.Lock()
-					si.status = sessInfo.GetStatus()
+					si.Status = sessInfo.GetStatus()
 					// Update connection state if there are any connections in
 					// the request.
 					for _, conn := range sessInfo.GetConnections() {
 						connId := conn.GetConnectionId()
-						connInfo, ok := si.connInfoMap[connId]
+						connInfo, ok := si.ConnInfoMap[connId]
 						if !ok {
 							event.WriteError(statusCtx, op, errors.New("connection change requested but could not find local information for it"), event.WithInfo("connection_id", connId))
 							continue
 						}
 
-						connInfo.status = conn.GetStatus()
+						connInfo.Status = conn.GetStatus()
 					}
 
 					si.Unlock()
@@ -238,38 +234,38 @@ func (w *Worker) cleanupConnections(cancelCtx context.Context, ignoreSessionStat
 	closeInfo := make(map[string]string)
 	cleanSessionIds := make([]string, 0)
 	w.sessionInfoMap.Range(func(key, value interface{}) bool {
-		si := value.(*sessionInfo)
+		si := value.(*session.Info)
 		si.Lock()
 		defer si.Unlock()
 		switch {
 		case ignoreSessionState,
-			si.status == pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING,
-			si.status == pbs.SESSIONSTATUS_SESSIONSTATUS_TERMINATED,
-			time.Until(si.lookupSessionResponse.Expiration.AsTime()) < 0:
+			si.Status == pbs.SESSIONSTATUS_SESSIONSTATUS_CANCELING,
+			si.Status == pbs.SESSIONSTATUS_SESSIONSTATUS_TERMINATED,
+			time.Until(si.LookupSessionResponse.Expiration.AsTime()) < 0:
 			// Cancel connections without regard to individual connection
 			// state.
-			closedIds := w.cancelConnections(si.connInfoMap, true)
+			closedIds := w.cancelConnections(si.ConnInfoMap, true)
 			for _, connId := range closedIds {
-				closeInfo[connId] = si.id
-				event.WriteSysEvent(cancelCtx, op, "terminated connection due to cancellation or expiration", "session_id", si.id, "connection_id", connId)
+				closeInfo[connId] = si.Id
+				event.WriteSysEvent(cancelCtx, op, "terminated connection due to cancellation or expiration", "session_id", si.Id, "connection_id", connId)
 			}
 
-			// closeTime is marked by closeConnections iff the
+			// CloseTime is marked by CloseConnections iff the
 			// status is returned for that connection as closed. If
 			// the session is no longer valid and all connections
 			// are marked closed, clean up the session.
 			if len(closedIds) == 0 {
-				cleanSessionIds = append(cleanSessionIds, si.id)
+				cleanSessionIds = append(cleanSessionIds, si.Id)
 			}
 
 		default:
 			// Cancel connections *with* regard to individual connection
 			// state (ie: only ones that the controller has requested be
 			// terminated).
-			closedIds := w.cancelConnections(si.connInfoMap, false)
+			closedIds := w.cancelConnections(si.ConnInfoMap, false)
 			for _, connId := range closedIds {
-				closeInfo[connId] = si.id
-				event.WriteSysEvent(cancelCtx, op, "terminated connection due to cancellation or expiration", "session_id", si.id, "connection_id", connId)
+				closeInfo[connId] = si.Id
+				event.WriteSysEvent(cancelCtx, op, "terminated connection due to cancellation or expiration", "session_id", si.Id, "connection_id", connId)
 			}
 		}
 
@@ -282,7 +278,12 @@ func (w *Worker) cleanupConnections(cancelCtx context.Context, ignoreSessionStat
 		// Call out to a helper to send the connection close requests to the
 		// controller, and set the close time. This functionality is shared with
 		// post-close functionality in the proxy handler.
-		w.closeConnections(cancelCtx, closeInfo)
+		sessClient, err := w.ControllerSessionConn()
+		if err != nil {
+			event.WriteError(cancelCtx, op, err, event.WithInfo("failed to create controller session client, connections won't be cleaned up"))
+		} else {
+			session.CloseConnections(cancelCtx, sessClient, w.sessionInfoMap, closeInfo)
+		}
 	}
 
 	// Forget sessions where the session is expired/canceled and all
@@ -293,21 +294,21 @@ func (w *Worker) cleanupConnections(cancelCtx context.Context, ignoreSessionStat
 }
 
 // cancelConnections is run by cleanupConnections to iterate over a
-// session's connInfoMap and close connections based on the
+// session's ConnInfoMap and close connections based on the
 // connection's state (or regardless if ignoreConnectionState is
 // set).
 //
 // The returned map and slice are the maps of connection -> session,
 // and sessions to completely remove from local state, respectively.
-func (w *Worker) cancelConnections(connInfoMap map[string]*connInfo, ignoreConnectionState bool) []string {
+func (w *Worker) cancelConnections(connInfoMap map[string]*session.ConnInfo, ignoreConnectionState bool) []string {
 	var closedIds []string
 	for k, v := range connInfoMap {
-		if v.closeTime.IsZero() {
-			if !ignoreConnectionState && v.status != pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED {
+		if v.CloseTime.IsZero() {
+			if !ignoreConnectionState && v.Status != pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED {
 				continue
 			}
 
-			v.connCancel()
+			v.ConnCancel()
 			closedIds = append(closedIds, k)
 		}
 	}

internal/servers/worker/status_test.go

@@ -14,7 +14,7 @@ import (
 )
 
 func TestWorkerWaitForNextSuccessfulStatusUpdate(t *testing.T) {
-	// do not runn using t.Parallel() since it relies on the sys eventer
+	// do not run using t.Parallel() since it relies on the sys eventer
 	event.TestEnableEventing(t, true)
 	testConfig := event.DefaultEventerConfig()
 	testLock := &sync.Mutex{}

internal/servers/worker/tcp_proxy.go

@@ -1,81 +0,0 @@
-package worker
-
-import (
-	"context"
-	"io"
-	"net"
-	"net/url"
-	"sync"
-
-	"nhooyr.io/websocket"
-
-	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
-	"github.com/hashicorp/boundary/internal/observability/event"
-)
-
-func (w *Worker) handleTcpProxyV1(connCtx context.Context, clientAddr *net.TCPAddr, conn *websocket.Conn, si *sessionInfo, connectionId, endpoint string) {
-	const op = "worker.(Worker).handleTcpProxyV1"
-	ctx := context.TODO()
-	si.RLock()
-	sessionId := si.lookupSessionResponse.GetAuthorization().GetSessionId()
-	si.RUnlock()
-
-	sessionUrl, err := url.Parse(endpoint)
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error parsing endpoint information", "session_id", sessionId, "endpoint", endpoint))
-		conn.Close(websocket.StatusInternalError, "cannot parse endpoint url")
-		return
-	}
-	if sessionUrl.Scheme != "tcp" {
-		event.WriteError(ctx, op, err, event.WithInfo("session_id", sessionId, "endpoint", endpoint))
-		conn.Close(websocket.StatusInternalError, "invalid scheme for type")
-		return
-	}
-	remoteConn, err := net.Dial("tcp", sessionUrl.Host)
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error dialing endpoint", "endpoint", endpoint))
-		conn.Close(websocket.StatusInternalError, "endpoint dialing failed")
-		return
-	}
-	// Assert this for better Go 1.11 splice support
-	tcpRemoteConn := remoteConn.(*net.TCPConn)
-
-	endpointAddr := tcpRemoteConn.RemoteAddr().(*net.TCPAddr)
-	connectionInfo := &pbs.ConnectConnectionRequest{
-		ConnectionId:       connectionId,
-		ClientTcpAddress:   clientAddr.IP.String(),
-		ClientTcpPort:      uint32(clientAddr.Port),
-		EndpointTcpAddress: endpointAddr.IP.String(),
-		EndpointTcpPort:    uint32(endpointAddr.Port),
-		Type:               "tcp",
-	}
-
-	connStatus, err := w.connectConnection(connCtx, connectionInfo)
-	if err != nil {
-		event.WriteError(ctx, op, err, event.WithInfoMsg("error marking connection as connected"))
-		conn.Close(websocket.StatusInternalError, "failed to mark connection as connected")
-		return
-	}
-	si.Lock()
-	si.connInfoMap[connectionId].status = connStatus
-	si.Unlock()
-
-	// Get a wrapped net.Conn so we can use io.Copy
-	netConn := websocket.NetConn(connCtx, conn, websocket.MessageBinary)
-
-	connWg := new(sync.WaitGroup)
-	connWg.Add(2)
-	go func() {
-		defer connWg.Done()
-		_, _ = io.Copy(netConn, tcpRemoteConn)
-		netConn.Close()
-		tcpRemoteConn.Close()
-	}()
-	go func() {
-		defer connWg.Done()
-		_, _ = io.Copy(tcpRemoteConn, netConn)
-		tcpRemoteConn.Close()
-		netConn.Close()
-	}()
-	connWg.Wait()
-}

internal/servers/worker/testing.go

@@ -11,6 +11,7 @@ import (
 	"github.com/hashicorp/boundary/internal/cmd/config"
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
 	"github.com/hashicorp/go-hclog"
 	wrapping "github.com/hashicorp/go-kms-wrapping"
 	"github.com/hashicorp/go-secure-stdlib/base62"
@@ -83,7 +84,7 @@ func (tw *TestWorker) ProxyAddrs() []string {
 // the worker's local session state. This detail is a point-in-time
 // snapshot of what's in sessionInfoMap for a particular session, and
 // may not contain all of the information that is contained within
-// it, or the underlying connInfoMap. Only details that are really
+// it, or the underlying ConnInfoMap. Only details that are really
 // important to testing are passed along.
 type TestSessionInfo struct {
 	Id     string
@@ -117,21 +118,21 @@ func (tw *TestWorker) LookupSession(id string) (TestSessionInfo, bool) {
 		return result, false
 	}
 
-	sess := raw.(*sessionInfo)
+	sess := raw.(*session.Info)
 	sess.RLock()
 	defer sess.RUnlock()
 
 	conns := make(map[string]TestConnectionInfo)
-	for _, conn := range sess.connInfoMap {
-		conns[conn.id] = TestConnectionInfo{
-			Id:        conn.id,
-			Status:    conn.status,
-			CloseTime: conn.closeTime,
+	for _, conn := range sess.ConnInfoMap {
+		conns[conn.Id] = TestConnectionInfo{
+			Id:        conn.Id,
+			Status:    conn.Status,
+			CloseTime: conn.CloseTime,
 		}
 	}
 
-	result.Id = sess.id
-	result.Status = sess.status
+	result.Id = sess.Id
+	result.Status = sess.Status
 	result.Connections = conns
 
 	return result, true

internal/servers/worker/testing_test.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
 	"github.com/stretchr/testify/require"
 )
 
@@ -20,14 +21,14 @@ func TestTestWorkerLookupSession(t *testing.T) {
 	tw.w = &Worker{
 		sessionInfoMap: new(sync.Map),
 	}
-	tw.w.sessionInfoMap.Store("foo", &sessionInfo{
-		id:     "foo",
-		status: pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
-		connInfoMap: map[string]*connInfo{
+	tw.w.sessionInfoMap.Store("foo", &session.Info{
+		Id:     "foo",
+		Status: pbs.SESSIONSTATUS_SESSIONSTATUS_ACTIVE,
+		ConnInfoMap: map[string]*session.ConnInfo{
 			"one": {
-				id:        "one",
-				status:    pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED,
-				closeTime: refTime,
+				Id:        "one",
+				Status:    pbs.CONNECTIONSTATUS_CONNECTIONSTATUS_CLOSED,
+				CloseTime: refTime,
 			},
 		},
 	})

internal/servers/worker/worker.go

@@ -2,16 +2,23 @@ package worker
 
 import (
 	"context"
+	"crypto/ed25519"
 	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
+	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/hashicorp/boundary/internal/cmd/config"
+	pbs "github.com/hashicorp/boundary/internal/gen/controller/servers/services"
 	"github.com/hashicorp/boundary/internal/observability/event"
 	"github.com/hashicorp/boundary/internal/servers"
+	"github.com/hashicorp/boundary/internal/servers/worker/session"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-secure-stdlib/mlock"
 	ua "go.uber.org/atomic"
@@ -100,7 +107,7 @@ func New(conf *Config) (*Worker, error) {
 func (w *Worker) Start() error {
 	const op = "worker.(Worker).Start"
 	if w.started.Load() {
-		event.WriteSysEvent(context.TODO(), op, "already started, skipping")
+		event.WriteSysEvent(w.baseContext, op, "already started, skipping")
 		return nil
 	}
 
@@ -135,9 +142,8 @@ func (w *Worker) Start() error {
 // doable, but work for later.
 func (w *Worker) Shutdown(skipListeners bool) error {
 	const op = "worker.(Worker).Shutdown"
-	ctx := context.TODO()
 	if !w.started.Load() {
-		event.WriteSysEvent(context.TODO(), op, "already shut down, skipping")
+		event.WriteSysEvent(w.baseContext, op, "already shut down, skipping")
 		return nil
 	}
 
@@ -163,7 +169,7 @@ func (w *Worker) Shutdown(skipListeners bool) error {
 	defer nextStatusCancel()
 	for {
 		if err := nextStatusCtx.Err(); err != nil {
-			event.WriteError(ctx, op, err, event.WithInfoMsg("error waiting for next status report to controller"))
+			event.WriteError(w.baseContext, op, err, event.WithInfoMsg("error waiting for next status report to controller"))
 			break
 		}
 
@@ -211,3 +217,100 @@ func (w *Worker) ParseAndStoreTags(incoming map[string][]string) {
 	w.tags.Store(tags)
 	w.updateTags.Store(true)
 }
+
+func (w *Worker) ControllerSessionConn() (pbs.SessionServiceClient, error) {
+	rawConn := w.controllerSessionConn.Load()
+	if rawConn == nil {
+		return nil, errors.New("unable to load controller session service connection")
+	}
+	sessClient, ok := rawConn.(pbs.SessionServiceClient)
+	if !ok {
+		return nil, fmt.Errorf("invalid session client %T", rawConn)
+	}
+	if sessClient == nil {
+		return nil, fmt.Errorf("session client is nil")
+	}
+
+	return sessClient, nil
+}
+
+func (w *Worker) getSessionTls(hello *tls.ClientHelloInfo) (*tls.Config, error) {
+	const op = "worker.(Worker).getSessionTls"
+	ctx := w.baseContext
+	var sessionId string
+	switch {
+	case strings.HasPrefix(hello.ServerName, "s_"):
+		sessionId = hello.ServerName
+	default:
+		event.WriteSysEvent(ctx, op, "invalid session in SNI", "session_id", hello.ServerName)
+		return nil, fmt.Errorf("could not find session ID in SNI")
+	}
+
+	conn, err := w.ControllerSessionConn()
+	if err != nil {
+		event.WriteError(ctx, op, err, event.WithInfo("failed to create controller session client"))
+	}
+
+	timeoutContext, cancel := context.WithTimeout(w.baseContext, session.ValidateSessionTimeout)
+	defer cancel()
+
+	resp, err := conn.LookupSession(timeoutContext, &pbs.LookupSessionRequest{
+		ServerId:  w.conf.RawConfig.Worker.Name,
+		SessionId: sessionId,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("error validating session: %w", err)
+	}
+
+	if resp.GetExpiration().AsTime().Before(time.Now()) {
+		return nil, fmt.Errorf("session is expired")
+	}
+
+	parsedCert, err := x509.ParseCertificate(resp.GetAuthorization().Certificate)
+	if err != nil {
+		return nil, fmt.Errorf("error parsing session certificate: %w", err)
+	}
+
+	if len(parsedCert.DNSNames) != 1 {
+		return nil, fmt.Errorf("invalid length of DNS names (%d) in parsed certificate", len(parsedCert.DNSNames))
+	}
+
+	certPool := x509.NewCertPool()
+	certPool.AddCert(parsedCert)
+
+	tlsConf := &tls.Config{
+		Certificates: []tls.Certificate{
+			{
+				Certificate: [][]byte{resp.GetAuthorization().Certificate},
+				PrivateKey:  ed25519.PrivateKey(resp.GetAuthorization().PrivateKey),
+				Leaf:        parsedCert,
+			},
+		},
+		ServerName: parsedCert.DNSNames[0],
+		ClientAuth: tls.RequireAndVerifyClientCert,
+		ClientCAs:  certPool,
+		MinVersion: tls.VersionTLS13,
+	}
+
+	si := &session.Info{
+		Id:                    resp.GetAuthorization().GetSessionId(),
+		SessionTls:            tlsConf,
+		LookupSessionResponse: resp,
+		Status:                resp.GetStatus(),
+		ConnInfoMap:           make(map[string]*session.ConnInfo),
+	}
+	// TODO: Periodically clean this up. We can't rely on things in here but
+	// not in cancellation because they could be on the way to being
+	// established. However, since cert lifetimes are short, we can simply range
+	// through and remove values that are expired.
+	actualSiRaw, loaded := w.sessionInfoMap.LoadOrStore(sessionId, si)
+	if loaded {
+		// Update the response to the latest
+		actualSi := actualSiRaw.(*session.Info)
+		actualSi.Lock()
+		actualSi.LookupSessionResponse = resp
+		actualSi.Unlock()
+	}
+
+	return tlsConf, nil
+}