1306119416

Committed 26 May 2024 08:13PM UTC coverage: 77.573% (+0.1%) from 77.427%

Build # 1306119416

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

2 of 19 new or added lines in 1 file covered. (10.53%)

1 existing line in 1 file now uncovered.

2480 of 3197 relevant lines covered (77.57%)

9785.39 hits per line

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

55.14

/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
                }

                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)
                                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)
                                        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.Trace("dbusEventLoop: daemon needs to reconnect")
                                                needsReconnect = true
                                        }
                                }
                        }
                }

                if needsReconnect {
                        _, err := client.FetchJWTToken()
                        if err != nil {
                                log.Errorf("dbusEventLoop: error fetching JWT token")
                        }
                }
        }
        log.Trace("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 {	2,605,271✔
157	return d.stop	2,605,271✔
158	}	2,605,271✔
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 {	2✔
299	select {	2✔
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):	1✔
304	return false	1✔
305	}
306	}
307
308	func (d *MenderShellDaemon) dbusEventLoop(client mender.AuthClient) {	3✔
309	needsReconnect := false	3✔
310	for {	2,605,266✔
311	if d.shouldStop() {	2,605,264✔
312	break	1✔
313	}
314
315	p, err := client.WaitForJwtTokenStateChange()	2,605,262✔
316	log.Tracef("dbusEventLoop: WaitForJwtTokenStateChange %v, err %v", p, err)	2,605,262✔
317		2,605,262✔
318	if len(p) > 1 &&	2,605,262✔
319	p[0].ParamType == dbus.GDBusTypeString &&	2,605,262✔
320	p[1].ParamType == dbus.GDBusTypeString {	2,605,262✔
NEW 321		×
322	token := p[0].ParamData.(string)	×
323	serverURL := p[1].ParamData.(string)	×
NEW 324		×
325	d.processJwtTokenStateChange(token, serverURL)	×
326	if len(token) > 0 && len(serverURL) > 0 {	×
327	log.Tracef("dbusEventLoop: got a token len=%d, ServerURL=%s", len(token), serverURL)	×
328	if token != d.serverJwt \|\| serverURL != d.serverUrl {	×
NEW 329	e := MenderShellDaemonEvent{	×
NEW 330	event: EventReconnect,	×
NEW 331	data: token,	×
NEW 332	id: "(dbusEventLoop)",	×
NEW 333	}	×
NEW 334	d.serverJwt = token	×
NEW 335	d.serverUrl = serverURL	×
NEW 336	d.postEvent(e)	×
NEW 337	log.Tracef("(dbusEventLoop) posting Event: %s", e.event)	×
NEW 338	needsReconnect = false	×
339		×
340	// If the server (Mender client proxy) closed the connection, it is likely that	×
341	// both messageLoop is asking for a reconnection and we got JwtTokenStateChange	×
342	// signal. So drain here the reconnect channel and reconnect only once	×
343	if d.needsReconnect() {	×
NEW 344	log.Trace("dbusEventLoop: daemon needs to reconnect")	×
NEW 345	needsReconnect = true	×
UNCOV 346	}	×
347	}
348	}
349	}
350
351	if needsReconnect {	2,605,260✔
NEW 352	_, err := client.FetchJWTToken()	×
NEW 353	if err != nil {	×
NEW 354	log.Errorf("dbusEventLoop: error fetching JWT token")	×
355	}	×
356	}
357	}
358	log.Trace("dbusEventLoop: returning")	1✔
359	}
360
361	func (d *MenderShellDaemon) postEvent(event MenderShellDaemonEvent) {	3✔
362	d.eventChan <- event	3✔
363	}	3✔
364
365	func (d *MenderShellDaemon) readEvent() MenderShellDaemonEvent {	3✔
366	return <-d.eventChan	3✔
367	}	3✔
368
369	func (d *MenderShellDaemon) eventLoop() {	2✔
370	var err error	2✔
371	for {	5✔
372	if d.shouldStop() {	4✔
373	break	1✔
374	}
375
376	event := d.readEvent()	2✔
377	log.Tracef("eventLoop: got event: %s", event.event)	2✔
378	switch event.event {	2✔
379	case EventReconnect:	×
380	err = connectionmanager.Reconnect(	×
381	ws.ProtoTypeShell, d.serverUrl,	×
382	d.deviceConnectUrl, event.data,	×
383	d.ctx,	×
384	)	×
385	var event string	×
386	if err != nil {	×
387	log.Errorf("eventLoop: error reconnecting: %s", err.Error())	×
388	event = EventConnectionError	×
389	} else {	×
390	log.Infof("eventLoop: Connection established with %s", d.serverUrl)	×
391	event = EventConnectionEstablished	×
392	}	×
393	d.connectionEstChan <- MenderShellDaemonEvent{	×
394	event: event,	×
395	}	×
396	}
397	}
398
399	log.Trace("eventLoop: returning")	1✔
400	}
401
402	func (d *MenderShellDaemon) setupLogging() {	1✔
403	if d.trace {	1✔
404	log.SetLevel(log.TraceLevel)	×
405	} else if d.debug {	2✔
406	log.SetLevel(log.DebugLevel)	1✔
407	}	1✔
408	}
409
410	func (d *MenderShellDaemon) DecreaseSpawnedShellsCount(shellStoppedCount uint) {	7✔
411	d.spawnedShellsMutex.Lock()	7✔
412	defer d.spawnedShellsMutex.Unlock()	7✔
413	if d.shellsSpawned == 0 {	9✔
414	log.Warn("can't decrement shellsSpawned count: it is 0.")	2✔
415	} else {	7✔
416	if shellStoppedCount >= d.shellsSpawned {	8✔
417	d.shellsSpawned = 0	3✔
418	} else {	5✔
419	d.shellsSpawned -= shellStoppedCount	2✔
420	}	2✔
421	}
422	}
423
424	// starts all needed elements of the mender-connect daemon
425	// - executes given shell (shell.ExecuteShell)
426	// - get dbus API and starts the dbus main loop (dbus.GetDBusAPI(), go dbusAPI.MainLoopRun(loop))
427	// - creates a new dbus client and connects to dbus (mender.NewAuthClient(dbusAPI),
428	// client.Connect(...))
429	// - gets the JWT token from the mender-client via dbus (client.GetJWTToken())
430	// - connects to the backend and returns a new websocket (deviceconnect.Connect(...))
431	// - starts the message flow between the shell and websocket (shell.NewMenderShell(...))
432	func (d *MenderShellDaemon) Run() error {	1✔
433	d.setupLogging()	1✔
434		1✔
435	log.Trace("daemon Run starting")	1✔
436	u, err := user.Lookup(d.username)	1✔
437	if err == nil && u == nil {	1✔
438	return errors.New("unknown error while getting a user id")	×
439	}	×
440	if err != nil {	2✔
441	return err	1✔
442	}	1✔
443
444	d.homeDir = u.HomeDir	×
445		×
446	d.uid, err = strconv.ParseUint(u.Uid, 10, 32)	×
447	if err != nil {	×
448	return err	×
449	}	×
450
451	d.gid, err = strconv.ParseUint(u.Gid, 10, 32)	×
452	if err != nil {	×
453	return err	×
454	}	×
455
456	log.Trace("mender-connect connecting to dbus")	×
457		×
458	dbusAPI, err := dbus.GetDBusAPI()	×
459	if err != nil {	×
460	return err	×
461	}	×
462
463	//dbus main loop, required.
464	loop := dbusAPI.MainLoopNew()	×
465	go dbusAPI.MainLoopRun(loop)	×
466	defer dbusAPI.MainLoopQuit(loop)	×
467		×
468	//new dbus client	×
469	client, err := mender.NewAuthClient(dbusAPI)	×
470	if err != nil {	×
471	log.Errorf("mender-connect dbus failed to create client, error: %s", err.Error())	×
472	return err	×
473	}	×
474
475	//connection to dbus
476	err = client.Connect(mender.DBusObjectName, mender.DBusObjectPath, mender.DBusInterfaceName)	×
477	if err != nil {	×
478	log.Errorf("mender-connect dbus failed to connect, error: %s", err.Error())	×
479	return err	×
480	}	×
481
482	jwtToken, serverURL, err := client.GetJWTToken()	×
483	if err != nil {	×
484	log.Warnf("call to GetJWTToken on the Mender D-Bus API failed: %v", err)	×
485	}	×
486	d.serverJwt = jwtToken	×
487	d.serverUrl = serverURL	×
488	if len(d.serverJwt) > 0 && len(d.serverUrl) > 0 {	×
489	d.authorized = true	×
490	}	×
491
492	go func() {	×
493	_ = d.messageLoop()	×
494	}()	×
495	go d.dbusEventLoop(client)	×
496	go d.eventLoop()	×
497		×
498	log.Trace("mender-connect entering main loop.")	×
499	for {	×
500	if d.shouldStop() {	×
501	break	×
502	}
503
504	if d.shouldPrintStatus() {	×
505	d.outputStatus()	×
506	}	×
507
508	if d.timeToSweepSessions() {	×
509	shellStoppedCount, sessionStoppedCount, totalExpiredLeft, err :=	×
510	session.MenderSessionTerminateExpired()	×
511	if err != nil {	×
512	log.Errorf("main-loop: failed to terminate some expired sessions, left: %d",	×
513	totalExpiredLeft)	×
514	} else if sessionStoppedCount > 0 {	×
515	d.DecreaseSpawnedShellsCount(uint(sessionStoppedCount))	×
516	log.Infof("main-loop: stopped %d sessions, %d shells, expired sessions left: %d",	×
517	shellStoppedCount, sessionStoppedCount, totalExpiredLeft)	×
518	}	×
519	}
520
521	time.Sleep(time.Second)	×
522	}
523
524	log.Trace("mainLoop: returning")	×
525	return nil	×
526	}
527
528	func (d MenderShellDaemon) responseMessage(msg ws.ProtoMsg) (err error) {	14✔
529	log.Tracef("responseMessage: webSock.WriteMessage(%+v)", msg)	14✔
530	return connectionmanager.Write(ws.ProtoTypeShell, msg)	14✔
531	}	14✔
532
533	func (d MenderShellDaemon) routeMessage(msg ws.ProtoMsg) error {	18✔
534	var err error	18✔
535	// NOTE: the switch is required for backward compatibility, otherwise	18✔
536	// routing is performed and managed by the session.Router.	18✔
537	// Use the new API in sessions package (see filetransfer.go for an example)	18✔
538	switch msg.Header.Proto {	18✔
539	case ws.ProtoTypeShell:	16✔
540	if d.TerminalConfig.Disable {	16✔
541	break	×
542	}
543	switch msg.Header.MsgType {	16✔
544	case wsshell.MessageTypeSpawnShell:	8✔
545	return d.routeMessageSpawnShell(msg)	8✔
546	case wsshell.MessageTypeStopShell:	4✔
547	return d.routeMessageStopShell(msg)	4✔
548	case wsshell.MessageTypeShellCommand:	3✔
549	return d.routeMessageShellCommand(msg)	3✔
550	case wsshell.MessageTypeResizeShell:	×
551	return d.routeMessageShellResize(msg)	×
552	case wsshell.MessageTypePongShell:	×
553	return d.routeMessagePongShell(msg)	×
554	}
555	default:	2✔
556	return d.router.RouteMessage(	2✔
557	msg, session.ResponseWriterFunc(d.responseMessage),	2✔
558	)	2✔
559	}
560	err = errors.New(	1✔
561	fmt.Sprintf(	1✔
562	"unknown message protocol and type: %d/%s",	1✔
563	msg.Header.Proto,	1✔
564	msg.Header.MsgType,	1✔
565	),	1✔
566	)	1✔
567	response := &ws.ProtoMsg{	1✔
568	Header: ws.ProtoHdr{	1✔
569	Proto: msg.Header.Proto,	1✔
570	MsgType: msg.Header.MsgType,	1✔
571	SessionID: msg.Header.SessionID,	1✔
572	Properties: map[string]interface{}{	1✔
573	"status": wsshell.ErrorMessage,	1✔
574	},	1✔
575	},	1✔
576	Body: []byte(err.Error()),	1✔
577	}	1✔
578	if err := d.responseMessage(response); err != nil {	1✔
579	log.Errorf(errors.Wrap(err, "unable to send the response message").Error())	×
580	}	×
581	return err	1✔
582	}
583
584	func (d MenderShellDaemon) routeMessageResponse(response ws.ProtoMsg, err error) {	13✔
585	if err != nil {	18✔
586	log.Errorf(err.Error())	5✔
587	response.Header.Properties["status"] = wsshell.ErrorMessage	5✔
588	response.Body = []byte(err.Error())	5✔
589	} else if response == nil {	13✔
590	return	×
591	}	×
592	if err := d.responseMessage(response); err != nil {	13✔
593	log.Errorf(errors.Wrap(err, "unable to send the response message").Error())	×
594	}	×
595	}
596
597	func (d MenderShellDaemon) readMessage() (ws.ProtoMsg, error) {	18✔
598	msg, err := connectionmanager.Read(ws.ProtoTypeShell)	18✔
599	log.Tracef("webSock.ReadMessage()=%+v,%v", msg, err)	18✔
600	if err != nil {	19✔
601	return nil, err	1✔
602	}	1✔
603
604	return msg, nil	17✔
605	}

mendersoftware / mender-connect / 1306119416

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