1939385454

Committed 21 Jul 2025 12:22PM UTC coverage: 69.902% (-0.03%) from 69.932%

Build # 1939385454

Build Type

push

gitlab-ci

Committed by

web-flow

Commit Message

Merge pull request #160 from mendersoftware/cherry-2.3.x-preserve-groups

[Cherry 2.3.x]: fix: preserve group assignments

Run Details

3 of 14 new or added lines in 4 files covered. (21.43%)

4 existing lines in 2 files now uncovered.

2492 of 3565 relevant lines covered (69.9%)

6.09 hits per line

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

53.03

/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
        gids                    []uint32
        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)
                                if token != d.serverJwt || serverURL != d.serverUrl {
                                        log.Debugf(
                                                "dbusEventLoop: new token or ServerURL, reconnecting. len=%d, ServerURL=%s",
                                                len(token),
                                                serverURL,
                                        )
                                        needsReconnect = true

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

                                }
                                // TODO: moving these assignments one scope up would make d.authorized redundant...
                                d.serverJwt = token
                                d.serverUrl = serverURL
                        }
                }
                if needsReconnect && d.authorized {
                        jwtToken, serverURL, _ := client.GetJWTToken()
                        e := MenderShellDaemonEvent{
                                event: EventReconnect,
                                data:  jwtToken,
                                id:    "(dbusEventLoop)",
                        }
                        log.Tracef("(dbusEventLoop) posting Event: %s", e.event)
                        d.serverUrl = serverURL
                        d.serverJwt = jwtToken
                        d.postEvent(e)
                        needsReconnect = false
                }
        }

        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(...))
//
// nolint: gocyclo
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
        }

        gids, err := u.GroupIds()
        if err != nil {
                return errors.New("unknown error while getting group ids")
        }

        // Convert []string to []uint32
        d.gids = make([]uint32, len(gids))
        for i, g := range gids {
                gid, err := strconv.ParseUint(g, 10, 32)
                if err != nil {
                        return fmt.Errorf("failed to parse group id %q: %w", g, err)
                }
                d.gids[i] = uint32(gid)
        }

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

mendersoftware / mender-connect / 1939385454

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