1208696808

Committed 11 Mar 2024 11:36AM UTC coverage: 77.341% (+0.3%) from 77.049%

Build # 1208696808

Build Type

Pull #134

gitlab-ci

Committed by

SebOpsahl

Commit Message

fix: make in `mender-connect` to request a new JWT token on its own

Fix to make `mender-connect` request a new JWT token on its own, so its not dependent on `mender-update` to do so. Previously, `mender-connect` relied the Mender client to request new JWT tokens. Since the split of `mender` into `mender-update` and `mender-auth`, it is theoretically possible to run `mender-connect` without `mender-update`, in which case it wouldn't request JWT tokens without this fix.

Changelog: None

Ticket: MEN-6877
Signed-off-by: Sebastian Opsahl <sebastian.opsahl@northern.tech>

Pull Request Pull Request #134: fix: Fix to make mender-connect request a new JWT token on its own, so its not dependent on mender-update to do so

Run Details

3 of 21 new or added lines in 1 file covered. (14.29%)

4 existing lines in 1 file now uncovered.

2478 of 3204 relevant lines covered (77.34%)

7.08 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

54.71

/app/daemon.go

// Copyright 2022 Northern.tech AS
//
//        Licensed under the Apache License, Version 2.0 (the "License");
//        you may not use this file except in compliance with the License.
//        You may obtain a copy of the License at
//
//            http://www.apache.org/licenses/LICENSE-2.0
//
//        Unless required by applicable law or agreed to in writing, software
//        distributed under the License is distributed on an "AS IS" BASIS,
//        WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//        See the License for the specific language governing permissions and
//        limitations under the License.
package app

import (
        "context"
        "fmt"
        "os/user"
        "strconv"
        "strings"
        "sync"
        "time"

        "github.com/pkg/errors"
        log "github.com/sirupsen/logrus"

        "github.com/mendersoftware/go-lib-micro/ws"
        wsshell "github.com/mendersoftware/go-lib-micro/ws/shell"

        "github.com/mendersoftware/mender-connect/client/dbus"
        "github.com/mendersoftware/mender-connect/client/mender"
        "github.com/mendersoftware/mender-connect/config"
        "github.com/mendersoftware/mender-connect/connectionmanager"
        "github.com/mendersoftware/mender-connect/limits/filetransfer"
        "github.com/mendersoftware/mender-connect/session"
)

var lastExpiredSessionSweep = time.Now()
var expiredSessionsSweepFrequency = time.Second * 32

const (
        EventReconnect             = "reconnect"
        EventReconnectRequest      = "reconnect-req"
        EventConnectionEstablished = "connected"
        EventConnectionError       = "connected-error"
)

type MenderShellDaemonEvent struct {
        event string
        data  string
        id    string
}

type MenderShellDaemon struct {
        ctx                     context.Context
        ctxCancel               context.CancelFunc
        writeMutex              *sync.Mutex
        spawnedShellsMutex      *sync.Mutex
        eventChan               chan MenderShellDaemonEvent
        connectionEstChan       chan MenderShellDaemonEvent
        reconnectChan           chan MenderShellDaemonEvent
        stop                    bool
        authorized              bool
        printStatus             bool
        username                string
        shell                   string
        shellArguments          []string
        serverJwt               string
        serverUrl               string
        deviceConnectUrl        string
        expireSessionsAfter     time.Duration
        expireSessionsAfterIdle time.Duration
        terminalString          string
        uid                     uint64
        gid                     uint64
        homeDir                 string
        shellsSpawned           uint
        debug                   bool
        trace                   bool
        router                  session.Router
        config.TerminalConfig
        config.FileTransferConfig
        config.PortForwardConfig
        config.MenderClientConfig
}

func NewDaemon(conf *config.MenderShellConfig) *MenderShellDaemon {
        ctx, ctxCancel := context.WithCancel(context.Background())

        // Setup ProtoMsg routes.
        routes := make(session.ProtoRoutes)
        if !conf.Terminal.Disable {
                // Shell message is not handled by the Session, but the map
                // entry must be set to give the correct 'accept' response.
                routes[ws.ProtoTypeShell] = nil
        }
        if !conf.FileTransfer.Disable {
                routes[ws.ProtoTypeFileTransfer] = session.FileTransfer(conf.Limits)
        }
        if !conf.PortForward.Disable {
                routes[ws.ProtoTypePortForward] = session.PortForward()
        }
        if !conf.MenderClient.Disable {
                routes[ws.ProtoTypeMenderClient] = session.MenderClient()
        }
        router := session.NewRouter(
                routes, session.Config{
                        IdleTimeout: connectionmanager.DefaultPingWait,
                },
        )

        daemon := MenderShellDaemon{
                ctx:                     ctx,
                ctxCancel:               ctxCancel,
                writeMutex:              &sync.Mutex{},
                spawnedShellsMutex:      &sync.Mutex{},
                eventChan:               make(chan MenderShellDaemonEvent),
                connectionEstChan:       make(chan MenderShellDaemonEvent),
                reconnectChan:           make(chan MenderShellDaemonEvent),
                stop:                    false,
                authorized:              false,
                username:                conf.User,
                shell:                   conf.ShellCommand,
                shellArguments:          conf.ShellArguments,
                serverUrl:               "",
                expireSessionsAfter:     time.Second * time.Duration(conf.Sessions.ExpireAfter),
                expireSessionsAfterIdle: time.Second * time.Duration(conf.Sessions.ExpireAfterIdle),
                deviceConnectUrl:        config.DefaultDeviceConnectPath,
                terminalString:          config.DefaultTerminalString,
                TerminalConfig:          conf.Terminal,
                FileTransferConfig:      conf.FileTransfer,
                PortForwardConfig:       conf.PortForward,
                MenderClientConfig:      conf.MenderClient,
                shellsSpawned:           0,
                debug:                   conf.Debug,
                trace:                   conf.Trace,
                router:                  router,
        }

        if conf.Sessions.MaxPerUser > 0 {
                session.MaxUserSessions = int(conf.Sessions.MaxPerUser)
        }
        return &daemon
}

func (d *MenderShellDaemon) StopDaemon() {
        d.stop = true
        d.ctxCancel()
}

func (d *MenderShellDaemon) PrintStatus() {
        d.printStatus = true
}

func (d *MenderShellDaemon) shouldStop() bool {
        return d.stop
}

func (d *MenderShellDaemon) shouldPrintStatus() bool {
        return d.printStatus
}

func (d *MenderShellDaemon) timeToSweepSessions() bool {
        if d.expireSessionsAfter == time.Duration(0) && d.expireSessionsAfterIdle == time.Duration(0) {
                return false
        }

        now := time.Now()
        nextSweepAt := lastExpiredSessionSweep.Add(expiredSessionsSweepFrequency)
        if now.After(nextSweepAt) {
                lastExpiredSessionSweep = now
                return true
        } else {
                return false
        }
}

func (d *MenderShellDaemon) outputStatus() {
        log.Infof("mender-connect daemon v%s", config.VersionString())
        log.Info(" status: ")
        d.spawnedShellsMutex.Lock()
        log.Infof("  shells: %d/%d", d.shellsSpawned, config.MaxShellsSpawned)
        d.spawnedShellsMutex.Unlock()
        log.Infof("  sessions: %d", session.MenderShellSessionGetCount())
        sessionIds := session.MenderShellSessionGetSessionIds()
        for _, id := range sessionIds {
                s := session.MenderShellSessionGetById(id)
                log.Infof("   id:%s status:%d started:%s", id, s.GetStatus(), s.GetStartedAtFmt())
                log.Infof("   expires:%s active:%s", s.GetExpiresAtFmt(), s.GetActiveAtFmt())
                log.Infof("   shell:%s", s.GetShellCommandPath())
        }
        log.Info("  file-transfer:")
        tx, rx, tx1m, rx1m := filetransfer.GetCounters()
        log.Infof("   total: tx/rx %d/%d", tx, rx)
        log.Infof("   1m: tx rx %.2f %.2f (w)", tx1m, rx1m)
        d.printStatus = false
}

func (d *MenderShellDaemon) messageLoop() (err error) {
        log.Trace("messageLoop: starting")
        sendConnectReq := true
        waitConnectResp := true
        for {
                if d.shouldStop() {
                        log.Trace("messageLoop: returning")
                        break
                }

                if sendConnectReq {
                        e := MenderShellDaemonEvent{
                                event: EventReconnectRequest,
                        }
                        log.Tracef("messageLoop: posting event: %s; waiting for response", e.event)
                        d.reconnectChan <- e
                        sendConnectReq = false
                }

                if waitConnectResp {
                        response := <-d.connectionEstChan
                        log.Tracef("messageLoop: got response: %+v", response)
                        if response.event == EventConnectionEstablished {
                                waitConnectResp = false
                        } else {
                                // The re-connection failed, retry
                                sendConnectReq = true
                                waitConnectResp = true
                                continue
                        }
                }

                log.Trace("messageLoop: calling readMessage")
                message, err := d.readMessage()
                log.Tracef("messageLoop: called readMessage: %v,%v", message, err)
                if err != nil {
                        log.Errorf(
                                "messageLoop: error on readMessage: %v; disconnecting, waiting for reconnect.",
                                err,
                        )
                        // nolint:lll
                        // If we used a closed connection means that it has been closed from other goroutine
                        // and a reconnection is ongoing (or done). Just wait for the event and continue
                        // This can happen when dbusEventLoop detects a change in ServerURL and/or JWT token.
                        // It should be safe to use this string, see:
                        // https://github.com/golang/go/blob/529939072eef730c82333344f321972874758be8/src/net/error_test.go#L502-L507 or about
                        if !strings.Contains(err.Error(), "use of closed network connection") {
                                connectionmanager.Close(ws.ProtoTypeShell)
                                sendConnectReq = true
                        }
                        waitConnectResp = true
                        continue
                }

                log.Tracef("got message: type:%s data length:%d", message.Header.MsgType, len(message.Body))
                err = d.routeMessage(message)
                if err != nil {
                        log.Tracef("error routing message: %s", err.Error())
                }
        }

        log.Trace("messageLoop: returning")
        return err
}

func (d *MenderShellDaemon) processJwtTokenStateChange(jwtToken, serverUrl string) {
        jwtTokenLength := len(jwtToken)
        if jwtTokenLength > 0 && len(serverUrl) > 0 {
                if !d.authorized {
                        log.Tracef("dbusEventLoop: StateChanged from unauthorized"+
                                " to authorized, len(token)=%d, ServerURL=%q", jwtTokenLength, serverUrl)
                        //in here it is technically possible that we close a closed connection
                        //but it is not a critical error, the important thing is not to leave
                        //messageLoop waiting forever on readMessage
                        connectionmanager.Close(ws.ProtoTypeShell)
                }
                d.authorized = true
        } else {
                if d.authorized {
                        log.Tracef("dbusEventLoop: StateChanged from authorized to unauthorized." +
                                "terminating all sessions and disconnecting.")
                        shellsCount, sessionsCount, err := session.MenderSessionTerminateAll()
                        if err == nil {
                                log.Infof("dbusEventLoop terminated %d sessions, %d shells",
                                        sessionsCount, shellsCount)
                        } else {
                                log.Errorf("dbusEventLoop error terminating all sessions: %s",
                                        err.Error())
                        }
                        if shellsCount > 0 {
                                d.DecreaseSpawnedShellsCount(uint(shellsCount))
                        }
                }
                connectionmanager.Close(ws.ProtoTypeShell)
                d.authorized = false
        }
}

func (d *MenderShellDaemon) needsReconnect() bool {
        select {
        case e := <-d.reconnectChan:
                log.Tracef("needsReconnect: got event: %s", e.event)
                return true
        case <-time.After(time.Second):
                return false
        }
}

func (d *MenderShellDaemon) dbusEventLoop(client mender.AuthClient) {
        needsReconnect := false
        for {
                if d.shouldStop() {
                        break
                }

                if d.needsReconnect() {
                        log.Trace("dbusEventLoop: daemon needs to reconnect")
                        needsReconnect = true
                }

                p, err := client.WaitForJwtTokenStateChange()
                log.Tracef("dbusEventLoop: WaitForJwtTokenStateChange %v, err %v", p, err)

                if len(p) > 1 &&
                        p[0].ParamType == dbus.GDBusTypeString &&
                        p[1].ParamType == dbus.GDBusTypeString {

                        token := p[0].ParamData.(string)
                        serverURL := p[1].ParamData.(string)

                        d.processJwtTokenStateChange(token, serverURL)
                        if len(token) > 0 && len(serverURL) > 0 {
                                log.Tracef("dbusEventLoop: got a token len=%d, ServerURL=%s", len(token), serverURL)
                                log.Infof("dbusEventLoop: got a token len=%d, ServerURL=%s", len(token), serverURL)
                                if token != d.serverJwt || serverURL != d.serverUrl {
                                        e := MenderShellDaemonEvent{
                                                event: EventReconnect,
                                                data:  token,
                                                id:    "(dbusEventLoop)",
                                        }
                                        d.serverJwt = token
                                        d.serverUrl = serverURL
                                        d.postEvent(e)
                                        log.Tracef("(dbusEventLoop) posting Event: %s", e.event)
                                        log.Infof("(dbusEventLoop) posting Event: %s", e.event)
                                        needsReconnect = false

                                        // If the server (Mender client proxy) closed the connection, it is likely that
                                        // both messageLoop is asking for a reconnection and we got JwtTokenStateChange
                                        // signal. So drain here the reconnect channel and reconnect only once
                                        if d.needsReconnect() {
                                                log.Debug("dbusEventLoop: drained reconnect req channel")
                                        }
                                }
                        }
                }

                if needsReconnect {
                        log.Infof("Within needsReconnect ------")
                        _, err := client.FetchJWTToken()
                        if err != nil {
                                log.Errorf("dbusEventLoop: error fetching JWT token")
                        }
                }
        }
        log.Trace("dbusEventLoop: returning")
        log.Infof("dbusEventLoop: returning")
}

func (d *MenderShellDaemon) postEvent(event MenderShellDaemonEvent) {
        d.eventChan <- event
}

func (d *MenderShellDaemon) readEvent() MenderShellDaemonEvent {
        return <-d.eventChan
}

func (d *MenderShellDaemon) eventLoop() {
        var err error
        for {
                if d.shouldStop() {
                        break
                }

                event := d.readEvent()
                log.Tracef("eventLoop: got event: %s", event.event)
                switch event.event {
                case EventReconnect:
                        err = connectionmanager.Reconnect(
                                ws.ProtoTypeShell, d.serverUrl,
                                d.deviceConnectUrl, event.data,
                                d.ctx,
                        )
                        var event string
                        if err != nil {
                                log.Errorf("eventLoop: error reconnecting: %s", err.Error())
                                event = EventConnectionError
                        } else {
                                log.Infof("eventLoop: Connection established with %s", d.serverUrl)
                                event = EventConnectionEstablished
                        }
                        d.connectionEstChan <- MenderShellDaemonEvent{
                                event: event,
                        }
                }
        }

        log.Trace("eventLoop: returning")
}

func (d *MenderShellDaemon) setupLogging() {
        if d.trace {
                log.SetLevel(log.TraceLevel)
        } else if d.debug {
                log.SetLevel(log.DebugLevel)
        }
}

func (d *MenderShellDaemon) DecreaseSpawnedShellsCount(shellStoppedCount uint) {
        d.spawnedShellsMutex.Lock()
        defer d.spawnedShellsMutex.Unlock()
        if d.shellsSpawned == 0 {
                log.Warn("can't decrement shellsSpawned count: it is 0.")
        } else {
                if shellStoppedCount >= d.shellsSpawned {
                        d.shellsSpawned = 0
                } else {
                        d.shellsSpawned -= shellStoppedCount
                }
        }
}

// starts all needed elements of the mender-connect daemon
//   - executes given shell (shell.ExecuteShell)
//   - get dbus API and starts the dbus main loop (dbus.GetDBusAPI(), go dbusAPI.MainLoopRun(loop))
//   - creates a new dbus client and connects to dbus (mender.NewAuthClient(dbusAPI),
//     client.Connect(...))
//   - gets the JWT token from the mender-client via dbus (client.GetJWTToken())
//   - connects to the backend and returns a new websocket (deviceconnect.Connect(...))
//   - starts the message flow between the shell and websocket (shell.NewMenderShell(...))
func (d *MenderShellDaemon) Run() error {
        d.setupLogging()

        log.Trace("daemon Run starting")
        u, err := user.Lookup(d.username)
        if err == nil && u == nil {
                return errors.New("unknown error while getting a user id")
        }
        if err != nil {
                return err
        }

        d.homeDir = u.HomeDir

        d.uid, err = strconv.ParseUint(u.Uid, 10, 32)
        if err != nil {
                return err
        }

        d.gid, err = strconv.ParseUint(u.Gid, 10, 32)
        if err != nil {
                return err
        }

        log.Trace("mender-connect connecting to dbus")

        dbusAPI, err := dbus.GetDBusAPI()
        if err != nil {
                return err
        }

        //dbus main loop, required.
        loop := dbusAPI.MainLoopNew()
        go dbusAPI.MainLoopRun(loop)
        defer dbusAPI.MainLoopQuit(loop)

        //new dbus client
        client, err := mender.NewAuthClient(dbusAPI)
        if err != nil {
                log.Errorf("mender-connect dbus failed to create client, error: %s", err.Error())
                return err
        }

        //connection to dbus
        err = client.Connect(mender.DBusObjectName, mender.DBusObjectPath, mender.DBusInterfaceName)
        if err != nil {
                log.Errorf("mender-connect dbus failed to connect, error: %s", err.Error())
                return err
        }

        jwtToken, serverURL, err := client.GetJWTToken()
        if err != nil {
                log.Warnf("call to GetJWTToken on the Mender D-Bus API failed: %v", err)
        }
        d.serverJwt = jwtToken
        d.serverUrl = serverURL
        if len(d.serverJwt) > 0 && len(d.serverUrl) > 0 {
                d.authorized = true
        }

        go func() {
                _ = d.messageLoop()
        }()
        go d.dbusEventLoop(client)
        go d.eventLoop()

        log.Trace("mender-connect entering main loop.")
        for {
                if d.shouldStop() {
                        break
                }

                if d.shouldPrintStatus() {
                        d.outputStatus()
                }

                if d.timeToSweepSessions() {
                        shellStoppedCount, sessionStoppedCount, totalExpiredLeft, err :=
                                session.MenderSessionTerminateExpired()
                        if err != nil {
                                log.Errorf("main-loop: failed to terminate some expired sessions, left: %d",
                                        totalExpiredLeft)
                        } else if sessionStoppedCount > 0 {
                                d.DecreaseSpawnedShellsCount(uint(sessionStoppedCount))
                                log.Infof("main-loop: stopped %d sessions, %d shells, expired sessions left: %d",
                                        shellStoppedCount, sessionStoppedCount, totalExpiredLeft)
                        }
                }

                time.Sleep(time.Second)
        }

        log.Trace("mainLoop: returning")
        return nil
}

func (d *MenderShellDaemon) responseMessage(msg *ws.ProtoMsg) (err error) {
        log.Tracef("responseMessage: webSock.WriteMessage(%+v)", msg)
        return connectionmanager.Write(ws.ProtoTypeShell, msg)
}

func (d *MenderShellDaemon) routeMessage(msg *ws.ProtoMsg) error {
        var err error
        // NOTE: the switch is required for backward compatibility, otherwise
        //       routing is performed and managed by the session.Router.
        //       Use the new API in sessions package (see filetransfer.go for an example)
        switch msg.Header.Proto {
        case ws.ProtoTypeShell:
                if d.TerminalConfig.Disable {
                        break
                }
                switch msg.Header.MsgType {
                case wsshell.MessageTypeSpawnShell:
                        return d.routeMessageSpawnShell(msg)
                case wsshell.MessageTypeStopShell:
                        return d.routeMessageStopShell(msg)
                case wsshell.MessageTypeShellCommand:
                        return d.routeMessageShellCommand(msg)
                case wsshell.MessageTypeResizeShell:
                        return d.routeMessageShellResize(msg)
                case wsshell.MessageTypePongShell:
                        return d.routeMessagePongShell(msg)
                }
        default:
                return d.router.RouteMessage(
                        msg, session.ResponseWriterFunc(d.responseMessage),
                )
        }
        err = errors.New(
                fmt.Sprintf(
                        "unknown message protocol and type: %d/%s",
                        msg.Header.Proto,
                        msg.Header.MsgType,
                ),
        )
        response := &ws.ProtoMsg{
                Header: ws.ProtoHdr{
                        Proto:     msg.Header.Proto,
                        MsgType:   msg.Header.MsgType,
                        SessionID: msg.Header.SessionID,
                        Properties: map[string]interface{}{
                                "status": wsshell.ErrorMessage,
                        },
                },
                Body: []byte(err.Error()),
        }
        if err := d.responseMessage(response); err != nil {
                log.Errorf(errors.Wrap(err, "unable to send the response message").Error())
        }
        return err
}

func (d *MenderShellDaemon) routeMessageResponse(response *ws.ProtoMsg, err error) {
        if err != nil {
                log.Errorf(err.Error())
                response.Header.Properties["status"] = wsshell.ErrorMessage
                response.Body = []byte(err.Error())
        } else if response == nil {
                return
        }
        if err := d.responseMessage(response); err != nil {
                log.Errorf(errors.Wrap(err, "unable to send the response message").Error())
        }
}

func (d *MenderShellDaemon) readMessage() (*ws.ProtoMsg, error) {
        msg, err := connectionmanager.Read(ws.ProtoTypeShell)
        log.Tracef("webSock.ReadMessage()=%+v,%v", msg, err)
        if err != nil {
                return nil, err
        }

        return msg, nil
}

1	// Copyright 2022 Northern.tech AS
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14	package app
15
16	import (
17	"context"
18	"fmt"
19	"os/user"
20	"strconv"
21	"strings"
22	"sync"
23	"time"
24
25	"github.com/pkg/errors"
26	log "github.com/sirupsen/logrus"
27
28	"github.com/mendersoftware/go-lib-micro/ws"
29	wsshell "github.com/mendersoftware/go-lib-micro/ws/shell"
30
31	"github.com/mendersoftware/mender-connect/client/dbus"
32	"github.com/mendersoftware/mender-connect/client/mender"
33	"github.com/mendersoftware/mender-connect/config"
34	"github.com/mendersoftware/mender-connect/connectionmanager"
35	"github.com/mendersoftware/mender-connect/limits/filetransfer"
36	"github.com/mendersoftware/mender-connect/session"
37	)
38
39	var lastExpiredSessionSweep = time.Now()
40	var expiredSessionsSweepFrequency = time.Second * 32
41
42	const (
43	EventReconnect = "reconnect"
44	EventReconnectRequest = "reconnect-req"
45	EventConnectionEstablished = "connected"
46	EventConnectionError = "connected-error"
47	)
48
49	type MenderShellDaemonEvent struct {
50	event string
51	data string
52	id string
53	}
54
55	type MenderShellDaemon struct {
56	ctx context.Context
57	ctxCancel context.CancelFunc
58	writeMutex *sync.Mutex
59	spawnedShellsMutex *sync.Mutex
60	eventChan chan MenderShellDaemonEvent
61	connectionEstChan chan MenderShellDaemonEvent
62	reconnectChan chan MenderShellDaemonEvent
63	stop bool
64	authorized bool
65	printStatus bool
66	username string
67	shell string
68	shellArguments []string
69	serverJwt string
70	serverUrl string
71	deviceConnectUrl string
72	expireSessionsAfter time.Duration
73	expireSessionsAfterIdle time.Duration
74	terminalString string
75	uid uint64
76	gid uint64
77	homeDir string
78	shellsSpawned uint
79	debug bool
80	trace bool
81	router session.Router
82	config.TerminalConfig
83	config.FileTransferConfig
84	config.PortForwardConfig
85	config.MenderClientConfig
86	}
87
88	func NewDaemon(conf config.MenderShellConfig) MenderShellDaemon {	26✔
89	ctx, ctxCancel := context.WithCancel(context.Background())	26✔
90		26✔
91	// Setup ProtoMsg routes.	26✔
92	routes := make(session.ProtoRoutes)	26✔
93	if !conf.Terminal.Disable {	52✔
94	// Shell message is not handled by the Session, but the map	26✔
95	// entry must be set to give the correct 'accept' response.	26✔
96	routes[ws.ProtoTypeShell] = nil	26✔
97	}	26✔
98	if !conf.FileTransfer.Disable {	52✔
99	routes[ws.ProtoTypeFileTransfer] = session.FileTransfer(conf.Limits)	26✔
100	}	26✔
101	if !conf.PortForward.Disable {	52✔
102	routes[ws.ProtoTypePortForward] = session.PortForward()	26✔
103	}	26✔
104	if !conf.MenderClient.Disable {	52✔
105	routes[ws.ProtoTypeMenderClient] = session.MenderClient()	26✔
106	}	26✔
107	router := session.NewRouter(	26✔
108	routes, session.Config{	26✔
109	IdleTimeout: connectionmanager.DefaultPingWait,	26✔
110	},	26✔
111	)	26✔
112		26✔
113	daemon := MenderShellDaemon{	26✔
114	ctx: ctx,	26✔
115	ctxCancel: ctxCancel,	26✔
116	writeMutex: &sync.Mutex{},	26✔
117	spawnedShellsMutex: &sync.Mutex{},	26✔
118	eventChan: make(chan MenderShellDaemonEvent),	26✔
119	connectionEstChan: make(chan MenderShellDaemonEvent),	26✔
120	reconnectChan: make(chan MenderShellDaemonEvent),	26✔
121	stop: false,	26✔
122	authorized: false,	26✔
123	username: conf.User,	26✔
124	shell: conf.ShellCommand,	26✔
125	shellArguments: conf.ShellArguments,	26✔
126	serverUrl: "",	26✔
127	expireSessionsAfter: time.Second * time.Duration(conf.Sessions.ExpireAfter),	26✔
128	expireSessionsAfterIdle: time.Second * time.Duration(conf.Sessions.ExpireAfterIdle),	26✔
129	deviceConnectUrl: config.DefaultDeviceConnectPath,	26✔
130	terminalString: config.DefaultTerminalString,	26✔
131	TerminalConfig: conf.Terminal,	26✔
132	FileTransferConfig: conf.FileTransfer,	26✔
133	PortForwardConfig: conf.PortForward,	26✔
134	MenderClientConfig: conf.MenderClient,	26✔
135	shellsSpawned: 0,	26✔
136	debug: conf.Debug,	26✔
137	trace: conf.Trace,	26✔
138	router: router,	26✔
139	}	26✔
140		26✔
141	if conf.Sessions.MaxPerUser > 0 {	27✔
142	session.MaxUserSessions = int(conf.Sessions.MaxPerUser)	1✔
143	}	1✔
144	return &daemon	26✔
145	}
146
147	func (d *MenderShellDaemon) StopDaemon() {	1✔
148	d.stop = true	1✔
149	d.ctxCancel()	1✔
150	}	1✔
151
152	func (d *MenderShellDaemon) PrintStatus() {	2✔
153	d.printStatus = true	2✔
154	}	2✔
155
156	func (d *MenderShellDaemon) shouldStop() bool {	124✔
157	return d.stop	124✔
158	}	124✔
159
160	func (d *MenderShellDaemon) shouldPrintStatus() bool {	1✔
161	return d.printStatus	1✔
162	}	1✔
163
164	func (d *MenderShellDaemon) timeToSweepSessions() bool {	4✔
165	if d.expireSessionsAfter == time.Duration(0) && d.expireSessionsAfterIdle == time.Duration(0) {	6✔
166	return false	2✔
167	}	2✔
168
169	now := time.Now()	2✔
170	nextSweepAt := lastExpiredSessionSweep.Add(expiredSessionsSweepFrequency)	2✔
171	if now.After(nextSweepAt) {	3✔
172	lastExpiredSessionSweep = now	1✔
173	return true	1✔
174	} else {	2✔
175	return false	1✔
176	}	1✔
177	}
178
179	func (d *MenderShellDaemon) outputStatus() {	1✔
180	log.Infof("mender-connect daemon v%s", config.VersionString())	1✔
181	log.Info(" status: ")	1✔
182	d.spawnedShellsMutex.Lock()	1✔
183	log.Infof(" shells: %d/%d", d.shellsSpawned, config.MaxShellsSpawned)	1✔
184	d.spawnedShellsMutex.Unlock()	1✔
185	log.Infof(" sessions: %d", session.MenderShellSessionGetCount())	1✔
186	sessionIds := session.MenderShellSessionGetSessionIds()	1✔
187	for _, id := range sessionIds {	5✔
188	s := session.MenderShellSessionGetById(id)	4✔
189	log.Infof(" id:%s status:%d started:%s", id, s.GetStatus(), s.GetStartedAtFmt())	4✔
190	log.Infof(" expires:%s active:%s", s.GetExpiresAtFmt(), s.GetActiveAtFmt())	4✔
191	log.Infof(" shell:%s", s.GetShellCommandPath())	4✔
192	}	4✔
193	log.Info(" file-transfer:")	1✔
194	tx, rx, tx1m, rx1m := filetransfer.GetCounters()	1✔
195	log.Infof(" total: tx/rx %d/%d", tx, rx)	1✔
196	log.Infof(" 1m: tx rx %.2f %.2f (w)", tx1m, rx1m)	1✔
197	d.printStatus = false	1✔
198	}
199
200	func (d *MenderShellDaemon) messageLoop() (err error) {	3✔
201	log.Trace("messageLoop: starting")	3✔
202	sendConnectReq := true	3✔
203	waitConnectResp := true	3✔
204	for {	6✔
205	if d.shouldStop() {	4✔
206	log.Trace("messageLoop: returning")	1✔
207	break	1✔
208	}
209
210	if sendConnectReq {	4✔
211	e := MenderShellDaemonEvent{	2✔
212	event: EventReconnectRequest,	2✔
213	}	2✔
214	log.Tracef("messageLoop: posting event: %s; waiting for response", e.event)	2✔
215	d.reconnectChan <- e	2✔
216	sendConnectReq = false	2✔
217	}	2✔
218
219	if waitConnectResp {	×
220	response := <-d.connectionEstChan	×
221	log.Tracef("messageLoop: got response: %+v", response)	×
222	if response.event == EventConnectionEstablished {	×
223	waitConnectResp = false	×
224	} else {	×
225	// The re-connection failed, retry	×
226	sendConnectReq = true	×
227	waitConnectResp = true	×
228	continue	×
229	}
230	}
231
232	log.Trace("messageLoop: calling readMessage")	×
233	message, err := d.readMessage()	×
234	log.Tracef("messageLoop: called readMessage: %v,%v", message, err)	×
235	if err != nil {	×
236	log.Errorf(	×
237	"messageLoop: error on readMessage: %v; disconnecting, waiting for reconnect.",	×
238	err,	×
239	)	×
240	// nolint:lll	×
241	// If we used a closed connection means that it has been closed from other goroutine	×
242	// and a reconnection is ongoing (or done). Just wait for the event and continue	×
243	// This can happen when dbusEventLoop detects a change in ServerURL and/or JWT token.	×
244	// It should be safe to use this string, see:	×
245	// https://github.com/golang/go/blob/529939072eef730c82333344f321972874758be8/src/net/error_test.go#L502-L507 or about	×
246	if !strings.Contains(err.Error(), "use of closed network connection") {	×
247	connectionmanager.Close(ws.ProtoTypeShell)	×
248	sendConnectReq = true	×
249	}	×
250	waitConnectResp = true	×
251	continue	×
252	}
253
254	log.Tracef("got message: type:%s data length:%d", message.Header.MsgType, len(message.Body))	×
255	err = d.routeMessage(message)	×
256	if err != nil {	×
257	log.Tracef("error routing message: %s", err.Error())	×
258	}	×
259	}
260
261	log.Trace("messageLoop: returning")	1✔
262	return err	1✔
263	}
264
265	func (d *MenderShellDaemon) processJwtTokenStateChange(jwtToken, serverUrl string) {	×
266	jwtTokenLength := len(jwtToken)	×
267	if jwtTokenLength > 0 && len(serverUrl) > 0 {	×
268	if !d.authorized {	×
269	log.Tracef("dbusEventLoop: StateChanged from unauthorized"+	×
270	" to authorized, len(token)=%d, ServerURL=%q", jwtTokenLength, serverUrl)	×
271	//in here it is technically possible that we close a closed connection	×
272	//but it is not a critical error, the important thing is not to leave	×
273	//messageLoop waiting forever on readMessage	×
274	connectionmanager.Close(ws.ProtoTypeShell)	×
275	}	×
276	d.authorized = true	×
277	} else {	×
278	if d.authorized {	×
279	log.Tracef("dbusEventLoop: StateChanged from authorized to unauthorized." +	×
280	"terminating all sessions and disconnecting.")	×
281	shellsCount, sessionsCount, err := session.MenderSessionTerminateAll()	×
282	if err == nil {	×
283	log.Infof("dbusEventLoop terminated %d sessions, %d shells",	×
284	sessionsCount, shellsCount)	×
285	} else {	×
286	log.Errorf("dbusEventLoop error terminating all sessions: %s",	×
287	err.Error())	×
288	}	×
289	if shellsCount > 0 {	×
290	d.DecreaseSpawnedShellsCount(uint(shellsCount))	×
291	}	×
292	}
293	connectionmanager.Close(ws.ProtoTypeShell)	×
294	d.authorized = false	×
295	}
296	}
297
298	func (d *MenderShellDaemon) needsReconnect() bool {	117✔
299	select {	117✔
300	case e := <-d.reconnectChan:	1✔
301	log.Tracef("needsReconnect: got event: %s", e.event)	1✔
302	return true	1✔
303	case <-time.After(time.Second):	114✔
304	return false	114✔
305	}
306	}
307
308	func (d *MenderShellDaemon) dbusEventLoop(client mender.AuthClient) {	3✔
309	needsReconnect := false	3✔
310	for {	119✔
311	if d.shouldStop() {	117✔
312	break	1✔
313	}
314
315	if d.needsReconnect() {	115✔
316	log.Trace("dbusEventLoop: daemon needs to reconnect")	×
317	needsReconnect = true	×
318	}	×
319
320	p, err := client.WaitForJwtTokenStateChange()	113✔
321	log.Tracef("dbusEventLoop: WaitForJwtTokenStateChange %v, err %v", p, err)	113✔
322		113✔
323	if len(p) > 1 &&	113✔
324	p[0].ParamType == dbus.GDBusTypeString &&	113✔
325	p[1].ParamType == dbus.GDBusTypeString {	113✔
NEW 326		×
327	token := p[0].ParamData.(string)	×
328	serverURL := p[1].ParamData.(string)	×
NEW 329		×
330	d.processJwtTokenStateChange(token, serverURL)	×
331	if len(token) > 0 && len(serverURL) > 0 {	×
332	log.Tracef("dbusEventLoop: got a token len=%d, ServerURL=%s", len(token), serverURL)	×
NEW 333	log.Infof("dbusEventLoop: got a token len=%d, ServerURL=%s", len(token), serverURL)	×
334	if token != d.serverJwt \|\| serverURL != d.serverUrl {	×
NEW 335	e := MenderShellDaemonEvent{	×
NEW 336	event: EventReconnect,	×
NEW 337	data: token,	×
NEW 338	id: "(dbusEventLoop)",	×
NEW 339	}	×
NEW 340	d.serverJwt = token	×
NEW 341	d.serverUrl = serverURL	×
NEW 342	d.postEvent(e)	×
NEW 343	log.Tracef("(dbusEventLoop) posting Event: %s", e.event)	×
NEW 344	log.Infof("(dbusEventLoop) posting Event: %s", e.event)	×
NEW 345	needsReconnect = false	×
346		×
347	// If the server (Mender client proxy) closed the connection, it is likely that	×
348	// both messageLoop is asking for a reconnection and we got JwtTokenStateChange	×
349	// signal. So drain here the reconnect channel and reconnect only once	×
350	if d.needsReconnect() {	×
351	log.Debug("dbusEventLoop: drained reconnect req channel")	×
352	}	×
353	}
354	}
355	}
356
357	if needsReconnect {	113✔
NEW 358	log.Infof("Within needsReconnect ------")	×
NEW 359	_, err := client.FetchJWTToken()	×
NEW 360	if err != nil {	×
NEW 361	log.Errorf("dbusEventLoop: error fetching JWT token")	×
362	}	×
363	}
364	}
365	log.Trace("dbusEventLoop: returning")	1✔
366	log.Infof("dbusEventLoop: returning")	1✔
367	}
368
369	func (d *MenderShellDaemon) postEvent(event MenderShellDaemonEvent) {	3✔
370	d.eventChan <- event	3✔
371	}	3✔
372
373	func (d *MenderShellDaemon) readEvent() MenderShellDaemonEvent {	3✔
374	return <-d.eventChan	3✔
375	}	3✔
376
377	func (d *MenderShellDaemon) eventLoop() {	2✔
378	var err error	2✔
379	for {	5✔
380	if d.shouldStop() {	4✔
381	break	1✔
382	}
383
384	event := d.readEvent()	2✔
385	log.Tracef("eventLoop: got event: %s", event.event)	2✔
386	switch event.event {	2✔
387	case EventReconnect:	×
388	err = connectionmanager.Reconnect(	×
389	ws.ProtoTypeShell, d.serverUrl,	×
390	d.deviceConnectUrl, event.data,	×
391	d.ctx,	×
392	)	×
393	var event string	×
394	if err != nil {	×
395	log.Errorf("eventLoop: error reconnecting: %s", err.Error())	×
396	event = EventConnectionError	×
397	} else {	×
398	log.Infof("eventLoop: Connection established with %s", d.serverUrl)	×
399	event = EventConnectionEstablished	×
400	}	×
401	d.connectionEstChan <- MenderShellDaemonEvent{	×
402	event: event,	×
403	}	×
404	}
405	}
406
407	log.Trace("eventLoop: returning")	1✔
408	}
409
410	func (d *MenderShellDaemon) setupLogging() {	1✔
411	if d.trace {	1✔
412	log.SetLevel(log.TraceLevel)	×
413	} else if d.debug {	2✔
414	log.SetLevel(log.DebugLevel)	1✔
415	}	1✔
416	}
417
418	func (d *MenderShellDaemon) DecreaseSpawnedShellsCount(shellStoppedCount uint) {	7✔
419	d.spawnedShellsMutex.Lock()	7✔
420	defer d.spawnedShellsMutex.Unlock()	7✔
421	if d.shellsSpawned == 0 {	9✔
422	log.Warn("can't decrement shellsSpawned count: it is 0.")	2✔
423	} else {	7✔
424	if shellStoppedCount >= d.shellsSpawned {	8✔
425	d.shellsSpawned = 0	3✔
426	} else {	5✔
427	d.shellsSpawned -= shellStoppedCount	2✔
428	}	2✔
429	}
430	}
431
432	// starts all needed elements of the mender-connect daemon
433	// - executes given shell (shell.ExecuteShell)
434	// - get dbus API and starts the dbus main loop (dbus.GetDBusAPI(), go dbusAPI.MainLoopRun(loop))
435	// - creates a new dbus client and connects to dbus (mender.NewAuthClient(dbusAPI),
436	// client.Connect(...))
437	// - gets the JWT token from the mender-client via dbus (client.GetJWTToken())
438	// - connects to the backend and returns a new websocket (deviceconnect.Connect(...))
439	// - starts the message flow between the shell and websocket (shell.NewMenderShell(...))
440	func (d *MenderShellDaemon) Run() error {	1✔
441	d.setupLogging()	1✔
442		1✔
443	log.Trace("daemon Run starting")	1✔
444	u, err := user.Lookup(d.username)	1✔
445	if err == nil && u == nil {	1✔
446	return errors.New("unknown error while getting a user id")	×
447	}	×
448	if err != nil {	2✔
449	return err	1✔
450	}	1✔
451
452	d.homeDir = u.HomeDir	×
453		×
454	d.uid, err = strconv.ParseUint(u.Uid, 10, 32)	×
455	if err != nil {	×
456	return err	×
457	}	×
458
459	d.gid, err = strconv.ParseUint(u.Gid, 10, 32)	×
460	if err != nil {	×
461	return err	×
462	}	×
463
464	log.Trace("mender-connect connecting to dbus")	×
465		×
466	dbusAPI, err := dbus.GetDBusAPI()	×
467	if err != nil {	×
468	return err	×
469	}	×
470
471	//dbus main loop, required.
472	loop := dbusAPI.MainLoopNew()	×
473	go dbusAPI.MainLoopRun(loop)	×
474	defer dbusAPI.MainLoopQuit(loop)	×
475		×
476	//new dbus client	×
477	client, err := mender.NewAuthClient(dbusAPI)	×
478	if err != nil {	×
479	log.Errorf("mender-connect dbus failed to create client, error: %s", err.Error())	×
480	return err	×
481	}	×
482
483	//connection to dbus
484	err = client.Connect(mender.DBusObjectName, mender.DBusObjectPath, mender.DBusInterfaceName)	×
485	if err != nil {	×
486	log.Errorf("mender-connect dbus failed to connect, error: %s", err.Error())	×
487	return err	×
488	}	×
489
490	jwtToken, serverURL, err := client.GetJWTToken()	×
491	if err != nil {	×
492	log.Warnf("call to GetJWTToken on the Mender D-Bus API failed: %v", err)	×
493	}	×
494	d.serverJwt = jwtToken	×
495	d.serverUrl = serverURL	×
496	if len(d.serverJwt) > 0 && len(d.serverUrl) > 0 {	×
497	d.authorized = true	×
498	}	×
499
500	go func() {	×
501	_ = d.messageLoop()	×
502	}()	×
503	go d.dbusEventLoop(client)	×
504	go d.eventLoop()	×
505		×
506	log.Trace("mender-connect entering main loop.")	×
507	for {	×
508	if d.shouldStop() {	×
509	break	×
510	}
511
512	if d.shouldPrintStatus() {	×
513	d.outputStatus()	×
514	}	×
515
516	if d.timeToSweepSessions() {	×
517	shellStoppedCount, sessionStoppedCount, totalExpiredLeft, err :=	×
518	session.MenderSessionTerminateExpired()	×
519	if err != nil {	×
520	log.Errorf("main-loop: failed to terminate some expired sessions, left: %d",	×
521	totalExpiredLeft)	×
522	} else if sessionStoppedCount > 0 {	×
523	d.DecreaseSpawnedShellsCount(uint(sessionStoppedCount))	×
524	log.Infof("main-loop: stopped %d sessions, %d shells, expired sessions left: %d",	×
525	shellStoppedCount, sessionStoppedCount, totalExpiredLeft)	×
526	}	×
527	}
528
529	time.Sleep(time.Second)	×
530	}
531
532	log.Trace("mainLoop: returning")	×
533	return nil	×
534	}
535
536	func (d MenderShellDaemon) responseMessage(msg ws.ProtoMsg) (err error) {	14✔
537	log.Tracef("responseMessage: webSock.WriteMessage(%+v)", msg)	14✔
538	return connectionmanager.Write(ws.ProtoTypeShell, msg)	14✔
539	}	14✔
540
541	func (d MenderShellDaemon) routeMessage(msg ws.ProtoMsg) error {	18✔
542	var err error	18✔
543	// NOTE: the switch is required for backward compatibility, otherwise	18✔
544	// routing is performed and managed by the session.Router.	18✔
545	// Use the new API in sessions package (see filetransfer.go for an example)	18✔
546	switch msg.Header.Proto {	18✔
547	case ws.ProtoTypeShell:	16✔
548	if d.TerminalConfig.Disable {	16✔
549	break	×
550	}
551	switch msg.Header.MsgType {	16✔
552	case wsshell.MessageTypeSpawnShell:	8✔
553	return d.routeMessageSpawnShell(msg)	8✔
554	case wsshell.MessageTypeStopShell:	4✔
555	return d.routeMessageStopShell(msg)	4✔
556	case wsshell.MessageTypeShellCommand:	3✔
557	return d.routeMessageShellCommand(msg)	3✔
558	case wsshell.MessageTypeResizeShell:	×
559	return d.routeMessageShellResize(msg)	×
560	case wsshell.MessageTypePongShell:	×
561	return d.routeMessagePongShell(msg)	×
562	}
563	default:	2✔
564	return d.router.RouteMessage(	2✔
565	msg, session.ResponseWriterFunc(d.responseMessage),	2✔
566	)	2✔
567	}
568	err = errors.New(	1✔
569	fmt.Sprintf(	1✔
570	"unknown message protocol and type: %d/%s",	1✔
571	msg.Header.Proto,	1✔
572	msg.Header.MsgType,	1✔
573	),	1✔
574	)	1✔
575	response := &ws.ProtoMsg{	1✔
576	Header: ws.ProtoHdr{	1✔
577	Proto: msg.Header.Proto,	1✔
578	MsgType: msg.Header.MsgType,	1✔
579	SessionID: msg.Header.SessionID,	1✔
580	Properties: map[string]interface{}{	1✔
581	"status": wsshell.ErrorMessage,	1✔
582	},	1✔
583	},	1✔
584	Body: []byte(err.Error()),	1✔
585	}	1✔
586	if err := d.responseMessage(response); err != nil {	1✔
587	log.Errorf(errors.Wrap(err, "unable to send the response message").Error())	×
588	}	×
589	return err	1✔
590	}
591
592	func (d MenderShellDaemon) routeMessageResponse(response ws.ProtoMsg, err error) {	13✔
593	if err != nil {	18✔
594	log.Errorf(err.Error())	5✔
595	response.Header.Properties["status"] = wsshell.ErrorMessage	5✔
596	response.Body = []byte(err.Error())	5✔
597	} else if response == nil {	13✔
598	return	×
599	}	×
600	if err := d.responseMessage(response); err != nil {	13✔
601	log.Errorf(errors.Wrap(err, "unable to send the response message").Error())	×
602	}	×
603	}
604
605	func (d MenderShellDaemon) readMessage() (ws.ProtoMsg, error) {	18✔
606	msg, err := connectionmanager.Read(ws.ProtoTypeShell)	18✔
607	log.Tracef("webSock.ReadMessage()=%+v,%v", msg, err)	18✔
608	if err != nil {	19✔
609	return nil, err	1✔
610	}	1✔
611
612	return msg, nil	17✔
613	}

mendersoftware / mender-connect / 1208696808

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous