diff --git a/internal/gen/controller/servers/services/session_service_grpc.pb.go b/internal/gen/controller/servers/services/session_service_grpc.pb.go index 3a0ac7997..ab0a6d5de 100644 --- a/internal/gen/controller/servers/services/session_service_grpc.pb.go +++ b/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) diff --git a/internal/gen/controller/servers/services/testing.go b/internal/gen/controller/servers/services/testing.go new file mode 100644 index 000000000..8ce5e9586 --- /dev/null +++ b/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") +} diff --git a/internal/proto/local/controller/servers/services/v1/session_service.proto b/internal/proto/local/controller/servers/services/v1/session_service.proto index e1919aabf..01a594344 100644 --- a/internal/proto/local/controller/servers/services/v1/session_service.proto +++ b/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 diff --git a/internal/servers/controller/handlers/targets/target_service.go b/internal/servers/controller/handlers/targets/target_service.go index 75e39189e..649ee59b5 100644 --- a/internal/servers/controller/handlers/targets/target_service.go +++ b/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() diff --git a/internal/servers/controller/handlers/workers/worker_service.go b/internal/servers/controller/handlers/workers/worker_service.go index cb6a3f2f5..2fec629aa 100644 --- a/internal/servers/controller/handlers/workers/worker_service.go +++ b/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) { diff --git a/internal/servers/worker/common/common.go b/internal/servers/worker/common/common.go new file mode 100644 index 000000000..15d38e055 --- /dev/null +++ b/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{} diff --git a/internal/servers/worker/common/doc.go b/internal/servers/worker/common/doc.go new file mode 100644 index 000000000..ef082cb74 --- /dev/null +++ b/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 diff --git a/internal/servers/worker/config.go b/internal/servers/worker/config.go index 61fd575ce..0c0ecd97a 100644 --- a/internal/servers/worker/config.go +++ b/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 } diff --git a/internal/servers/worker/controller_connection.go b/internal/servers/worker/controller_connection.go index 789e11c44..e2d8b5a81 100644 --- a/internal/servers/worker/controller_connection.go +++ b/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, diff --git a/internal/servers/worker/handler.go b/internal/servers/worker/handler.go index 2c22faf35..47af24e9d 100644 --- a/internal/servers/worker/handler.go +++ b/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 - }) -} -*/ diff --git a/internal/servers/worker/listeners.go b/internal/servers/worker/listeners.go index 57983119e..adac049b0 100644 --- a/internal/servers/worker/listeners.go +++ b/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) } }() } diff --git a/internal/servers/worker/package_registration.go b/internal/servers/worker/package_registration.go new file mode 100644 index 000000000..ebbbd56d3 --- /dev/null +++ b/internal/servers/worker/package_registration.go @@ -0,0 +1,5 @@ +package worker + +import ( + _ "github.com/hashicorp/boundary/internal/servers/worker/proxy/tcp" +) diff --git a/internal/servers/worker/proxy/doc.go b/internal/servers/worker/proxy/doc.go new file mode 100644 index 000000000..330974b11 --- /dev/null +++ b/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 diff --git a/internal/servers/worker/proxy/options.go b/internal/servers/worker/proxy/options.go new file mode 100644 index 000000000..648e8e1b0 --- /dev/null +++ b/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{} +} diff --git a/internal/servers/worker/proxy/proxy.go b/internal/servers/worker/proxy/proxy.go new file mode 100644 index 000000000..4fff22c82 --- /dev/null +++ b/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 +} diff --git a/internal/servers/worker/proxy/proxy_test.go b/internal/servers/worker/proxy/proxy_test.go new file mode 100644 index 000000000..100e3fcbc --- /dev/null +++ b/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) +} diff --git a/internal/servers/worker/proxy/tcp/tcp.go b/internal/servers/worker/proxy/tcp/tcp.go new file mode 100644 index 000000000..f2ff1e718 --- /dev/null +++ b/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() +} diff --git a/internal/servers/worker/proxy/tcp/tcp_test.go b/internal/servers/worker/proxy/tcp/tcp_test.go new file mode 100644 index 000000000..c9178aa7b --- /dev/null +++ b/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() +} diff --git a/internal/servers/worker/proxy/testing.go b/internal/servers/worker/proxy/testing.go new file mode 100644 index 000000000..ec062e47b --- /dev/null +++ b/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 +} diff --git a/internal/servers/worker/proxy/testing_test.go b/internal/servers/worker/proxy/testing_test.go new file mode 100644 index 000000000..232627c9b --- /dev/null +++ b/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() +} diff --git a/internal/servers/worker/session.go b/internal/servers/worker/session.go deleted file mode 100644 index 9c5da6bc3..000000000 --- a/internal/servers/worker/session.go +++ /dev/null @@ -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 -} diff --git a/internal/servers/worker/session/session.go b/internal/servers/worker/session/session.go new file mode 100644 index 000000000..3fc3092d8 --- /dev/null +++ b/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 +} diff --git a/internal/servers/worker/session_test.go b/internal/servers/worker/session/session_test.go similarity index 76% rename from internal/servers/worker/session_test.go rename to internal/servers/worker/session/session_test.go index 750eecfdb..b836b5083 100644 --- a/internal/servers/worker/session_test.go +++ b/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) } } diff --git a/internal/servers/worker/status.go b/internal/servers/worker/status.go index d96b44cfd..2ae48a3ae 100644 --- a/internal/servers/worker/status.go +++ b/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) } } diff --git a/internal/servers/worker/status_test.go b/internal/servers/worker/status_test.go index b49d156c9..718546dbb 100644 --- a/internal/servers/worker/status_test.go +++ b/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{} diff --git a/internal/servers/worker/tcp_proxy.go b/internal/servers/worker/tcp_proxy.go deleted file mode 100644 index 328ef944a..000000000 --- a/internal/servers/worker/tcp_proxy.go +++ /dev/null @@ -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() -} diff --git a/internal/servers/worker/testing.go b/internal/servers/worker/testing.go index 66aa29155..a9a06906a 100644 --- a/internal/servers/worker/testing.go +++ b/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 diff --git a/internal/servers/worker/testing_test.go b/internal/servers/worker/testing_test.go index 1e86a53cc..b1ce3108e 100644 --- a/internal/servers/worker/testing_test.go +++ b/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, }, }, }) diff --git a/internal/servers/worker/worker.go b/internal/servers/worker/worker.go index 79680d5ba..fe89bdd88 100644 --- a/internal/servers/worker/worker.go +++ b/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 +}