857655394

Build Type

gitlab-ci

Committed by Krzysztof Jaskiewicz

Commit Message

chore: reduce cyclomatic complexity of VerifyToken method

Pull Request Pull Request #644: feat: handle device check-in time

Run Details

106 of 160 new or added lines in 3 files covered. (66.25%)

99 existing lines in 3 files now uncovered.

4627 of 5519 relevant lines covered (83.84%)

46.04 hits per line

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

78.01

/cmd/commands.go

// Copyright 2023 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 cmd

import (
        "context"
        "fmt"

        "github.com/mendersoftware/go-lib-micro/config"
        "github.com/mendersoftware/go-lib-micro/identity"
        "github.com/mendersoftware/go-lib-micro/log"
        "github.com/mendersoftware/go-lib-micro/mongo/migrate"
        mstore "github.com/mendersoftware/go-lib-micro/store"
        "github.com/pkg/errors"

        cinv "github.com/mendersoftware/deviceauth/client/inventory"
        "github.com/mendersoftware/deviceauth/client/orchestrator"
        "github.com/mendersoftware/deviceauth/client/tenant"
        dconfig "github.com/mendersoftware/deviceauth/config"
        "github.com/mendersoftware/deviceauth/model"
        "github.com/mendersoftware/deviceauth/store"
        "github.com/mendersoftware/deviceauth/store/mongo"
        "github.com/mendersoftware/deviceauth/utils"
)

var NowUnixMilis = utils.UnixMilis

func makeDataStoreConfig() mongo.DataStoreMongoConfig {
        return mongo.DataStoreMongoConfig{
                ConnectionString: config.Config.GetString(dconfig.SettingDb),

                SSL:           config.Config.GetBool(dconfig.SettingDbSSL),
                SSLSkipVerify: config.Config.GetBool(dconfig.SettingDbSSLSkipVerify),

                Username: config.Config.GetString(dconfig.SettingDbUsername),
                Password: config.Config.GetString(dconfig.SettingDbPassword),
        }

}

func Migrate(c config.Reader, tenant string, listTenantsFlag bool) error {
        db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())

        if err != nil {
                return errors.Wrap(err, "failed to connect to db")
        }

        // list tenants only
        if listTenantsFlag {
                return listTenants(db)
        }

        db = db.WithAutomigrate().(*mongo.DataStoreMongo)

        if config.Config.Get(dconfig.SettingTenantAdmAddr) != "" {
                db = db.WithMultitenant()
        }

        ctx := context.Background()
        if tenant == "" {
                err = db.Migrate(ctx, mongo.DbVersion)
        } else {
                tenantCtx := identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
                dbname := mstore.DbFromContext(tenantCtx, mongo.DbName)
                err = db.MigrateTenant(tenantCtx, dbname, mongo.DbVersion)
        }
        if err != nil {
                return errors.Wrap(err, "failed to run migrations")
        }

        return nil
}

func listTenants(db *mongo.DataStoreMongo) error {
        tdbs, err := db.GetTenantDbs()
        if err != nil {
                return errors.Wrap(err, "failed to retrieve tenant DBs")
        }

        for _, tenant := range tdbs {
                fmt.Println(mstore.TenantFromDbName(tenant, mongo.DbName))
        }

        return nil
}

func Maintenance(decommissioningCleanupFlag bool, tenant string, dryRunFlag bool) error {
        db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())
        if err != nil {
                return errors.Wrap(err, "failed to connect to db")
        }

        return maintenanceWithDataStore(decommissioningCleanupFlag, tenant, dryRunFlag, db)
}

func maintenanceWithDataStore(
        decommissioningCleanupFlag bool,
        tenant string,
        dryRunFlag bool,
        db *mongo.DataStoreMongo,
) error {
        // cleanup devauth database from leftovers after failed decommissioning
        if decommissioningCleanupFlag {
                return decommissioningCleanup(db, tenant, dryRunFlag)
        }

        return nil
}

func decommissioningCleanup(db *mongo.DataStoreMongo, tenant string, dryRunFlag bool) error {
        if tenant == "" {
                tdbs, err := db.GetTenantDbs()
                if err != nil {
                        return errors.Wrap(err, "failed to retrieve tenant DBs")
                }
                _ = decommissioningCleanupWithDbs(db, append(tdbs, mongo.DbName), dryRunFlag)
        } else {
                _ = decommissioningCleanupWithDbs(
                        db,
                        []string{mstore.DbNameForTenant(tenant, mongo.DbName)},
                        dryRunFlag,
                )
        }

        return nil
}

func decommissioningCleanupWithDbs(
        db *mongo.DataStoreMongo,
        tenantDbs []string,
        dryRunFlag bool,
) error {
        for _, dbName := range tenantDbs {
                println("database: ", dbName)
                if err := decommissioningCleanupWithDb(db, dbName, dryRunFlag); err != nil {
                        return err
                }
        }
        return nil
}

func decommissioningCleanupWithDb(db *mongo.DataStoreMongo, dbName string, dryRunFlag bool) error {
        if dryRunFlag {
                return decommissioningCleanupDryRun(db, dbName)
        } else {
                return decommissioningCleanupExecute(db, dbName)
        }
}

func decommissioningCleanupDryRun(db *mongo.DataStoreMongo, dbName string) error {
        //devices
        devices, err := db.GetDevicesBeingDecommissioned(dbName)
        if err != nil {
                return err
        }
        if len(devices) > 0 {
                fmt.Println("devices with decommissioning flag set:")
                for _, dev := range devices {
                        fmt.Println(dev.Id)
                }
        }

        //auth sets
        authSetIds, err := db.GetBrokenAuthSets(dbName)
        if err != nil {
                return err
        }
        if len(authSetIds) > 0 {
                fmt.Println("authentication sets to be removed:")
                for _, authSetId := range authSetIds {
                        fmt.Println(authSetId)
                }
        }

        return nil
}

func decommissioningCleanupExecute(db *mongo.DataStoreMongo, dbName string) error {
        if err := decommissioningCleanupDryRun(db, dbName); err != nil {
                return err
        }

        if err := db.DeleteDevicesBeingDecommissioned(dbName); err != nil {
                return err
        }

        if err := db.DeleteBrokenAuthSets(dbName); err != nil {
                return err
        }

        return nil
}

func PropagateStatusesInventory(
        db store.DataStore,
        c cinv.Client,
        tenant string,
        migrationVersion string,
        dryRun bool,
) error {
        l := log.NewEmpty()

        dbs, err := selectDbs(db, tenant)
        if err != nil {
                return errors.Wrap(err, "aborting")
        }

        var errReturned error
        for _, d := range dbs {
                err := tryPropagateStatusesInventoryForDb(db, c, d, migrationVersion, dryRun)
                if err != nil {
                        errReturned = err
                        l.Errorf("giving up on DB %s due to fatal error: %s", d, err.Error())
                        continue
                }
        }

        l.Info("all DBs processed, exiting.")
        return errReturned
}

func PropagateIdDataInventory(db store.DataStore, c cinv.Client, tenant string, dryRun bool) error {
        l := log.NewEmpty()

        dbs, err := selectDbs(db, tenant)
        if err != nil {
                return errors.Wrap(err, "aborting")
        }

        var errReturned error
        for _, d := range dbs {
                err := tryPropagateIdDataInventoryForDb(db, c, d, dryRun)
                if err != nil {
                        errReturned = err
                        l.Errorf("giving up on DB %s due to fatal error: %s", d, err.Error())
                        continue
                }
        }

        l.Info("all DBs processed, exiting.")
        return errReturned
}

func PropagateReporting(db store.DataStore, wflows orchestrator.ClientRunner, tenant string,
        dryRun bool) error {
        l := log.NewEmpty()

        dbs, err := selectDbs(db, tenant)
        if err != nil {
                return errors.Wrap(err, "aborting")
        }

        var errReturned error
        for _, d := range dbs {
                err := tryPropagateReportingForDb(db, wflows, d, dryRun)
                if err != nil {
                        errReturned = err
                        l.Errorf("giving up on DB %s due to fatal error: %s", d, err.Error())
                        continue
                }
        }

        l.Info("all DBs processed, exiting.")
        return errReturned
}

func selectDbs(db store.DataStore, tenant string) ([]string, error) {
        l := log.NewEmpty()

        var dbs []string

        if tenant != "" {
                l.Infof("propagating inventory for user-specified tenant %s", tenant)
                n := mstore.DbNameForTenant(tenant, mongo.DbName)
                dbs = []string{n}
        } else {
                l.Infof("propagating inventory for all tenants")

                // infer if we're in ST or MT
                tdbs, err := db.GetTenantDbs()
                if err != nil {
                        return nil, errors.Wrap(err, "failed to retrieve tenant DBs")
                }

                if len(tdbs) == 0 {
                        l.Infof("no tenant DBs found - will try the default database %s", mongo.DbName)
                        dbs = []string{mongo.DbName}
                } else {
                        dbs = tdbs
                }
        }

        return dbs, nil
}

const (
        devicesBatchSize = 512
)

func updateDevicesStatus(
        ctx context.Context,
        db store.DataStore,
        c cinv.Client,
        tenant string,
        status string,
        dryRun bool,
) error {
        var skip uint

        skip = 0
        for {
                devices, err := db.GetDevices(ctx,
                        skip,
                        devicesBatchSize,
                        model.DeviceFilter{Status: []string{status}},
                )
                if err != nil {
                        return errors.Wrap(err, "failed to get devices")
                }

                if len(devices) < 1 {
                        break
                }

                deviceUpdates := make([]model.DeviceInventoryUpdate, len(devices))

                for i, d := range devices {
                        deviceUpdates[i].Id = d.Id
                        deviceUpdates[i].Revision = d.Revision
                }

                if !dryRun {
                        err = c.SetDeviceStatus(ctx, tenant, deviceUpdates, status)
                        if err != nil {
                                return err
                        }
                }

                if len(devices) < devicesBatchSize {
                        break
                } else {
                        skip += devicesBatchSize
                }
        }
        return nil
}

func updateDevicesIdData(
        ctx context.Context,
        db store.DataStore,
        c cinv.Client,
        tenant string,
        dryRun bool,
) error {
        var skip uint

        skip = 0
        for {
                devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})
                if err != nil {
                        return errors.Wrap(err, "failed to get devices")
                }

                if len(devices) < 1 {
                        break
                }

                if !dryRun {
                        for _, d := range devices {
                                err := c.SetDeviceIdentity(ctx, tenant, d.Id, d.IdDataStruct)
                                if err != nil {
                                        return err
                                }
                        }
                }

                skip += devicesBatchSize
                if len(devices) < devicesBatchSize {
                        break
                }
        }
        return nil
}

func tryPropagateStatusesInventoryForDb(
        db store.DataStore,
        c cinv.Client,
        dbname string,
        migrationVersion string,
        dryRun bool,
) error {
        l := log.NewEmpty()

        l.Infof("propagating device statuses to inventory from DB: %s", dbname)

        tenant := mstore.TenantFromDbName(dbname, mongo.DbName)

        ctx := context.Background()
        if tenant != "" {
                ctx = identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
        }

        var err error
        var errReturned error
        for _, status := range model.DevStatuses {
                err = updateDevicesStatus(ctx, db, c, tenant, status, dryRun)
                if err != nil {
                        l.Infof(
                                "Done with DB %s status=%s, but there were errors: %s.",
                                dbname,
                                status,
                                err.Error(),
                        )
                        errReturned = err
                } else {
                        l.Infof("Done with DB %s status=%s", dbname, status)
                }
        }
        if migrationVersion != "" && !dryRun {
                if errReturned != nil {
                        l.Warnf(
                                "Will not store %s migration version in %s.migration_info due to errors.",
                                migrationVersion,
                                dbname,
                        )
                } else {
                        version, err := migrate.NewVersion(migrationVersion)
                        if version == nil || err != nil {
                                l.Warnf(
                                        "Will not store %s migration version in %s.migration_info due to bad version"+
                                                " provided.",
                                        migrationVersion,
                                        dbname,
                                )
                                errReturned = err
                        } else {
                                _ = db.StoreMigrationVersion(ctx, version)
                        }
                }
        }

        return errReturned
}

func tryPropagateIdDataInventoryForDb(
        db store.DataStore,
        c cinv.Client,
        dbname string,
        dryRun bool,
) error {
        l := log.NewEmpty()

        l.Infof("propagating device id_data to inventory from DB: %s", dbname)

        tenant := mstore.TenantFromDbName(dbname, mongo.DbName)

        ctx := context.Background()
        if tenant != "" {
                ctx = identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
        }

        err := updateDevicesIdData(ctx, db, c, tenant, dryRun)
        if err != nil {
                l.Infof("Done with DB %s, but there were errors: %s.", dbname, err.Error())
        } else {
                l.Infof("Done with DB %s", dbname)
        }

        return err
}

func tryPropagateReportingForDb(
        db store.DataStore,
        wflows orchestrator.ClientRunner,
        dbname string,
        dryRun bool,
) error {
        l := log.NewEmpty()

        l.Infof("propagating device data to reporting from DB: %s", dbname)

        tenant := mstore.TenantFromDbName(dbname, mongo.DbName)

        ctx := context.Background()
        if tenant != "" {
                ctx = identity.WithContext(ctx, &identity.Identity{
                        Tenant: tenant,
                })
        }

        err := reindexDevicesReporting(ctx, db, wflows, tenant, dryRun)
        if err != nil {
                l.Infof("Done with DB %s, but there were errors: %s.", dbname, err.Error())
        } else {
                l.Infof("Done with DB %s", dbname)
        }

        return err
}

func reindexDevicesReporting(
        ctx context.Context,
        db store.DataStore,
        wflows orchestrator.ClientRunner,
        tenant string,
        dryRun bool,
) error {
        var skip uint

        skip = 0
        for {
                devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})
                if err != nil {
                        return errors.Wrap(err, "failed to get devices")
                }

                if len(devices) < 1 {
                        break
                }

                if !dryRun {
                        deviceIDs := make([]string, len(devices))
                        for i, d := range devices {
                                deviceIDs[i] = d.Id
                        }
                        err := wflows.SubmitReindexReportingBatch(ctx, deviceIDs)
                        if err != nil {
                                return err
                        }
                }

                skip += devicesBatchSize
                if len(devices) < devicesBatchSize {
                        break
                }
        }
        return nil
}

const (
        WorkflowsDeviceLimitText    = "@/etc/workflows-enterprise/data/device_limit_email.txt"
        WorkflowsDeviceLimitHTML    = "@/etc/workflows-enterprise/data/device_limit_email.html"
        WorkflowsDeviceLimitSubject = "Device limit almost reached"
)

func warnTenantUsers(
        ctx context.Context,
        tenantID string,
        tadm tenant.ClientRunner,
        wflows orchestrator.ClientRunner,
        remainingDevices uint,
) error {
        users, err := tadm.GetTenantUsers(ctx, tenantID)
        if err != nil {
                // Log the event and continue with the other tenants
                return err
        }
        for i := range users {
                warnWFlow := orchestrator.DeviceLimitWarning{
                        RequestID:      "deviceAuthAdmin",
                        RecipientEmail: users[i].Email,

                        Subject:          WorkflowsDeviceLimitSubject,
                        Body:             WorkflowsDeviceLimitText,
                        BodyHTML:         WorkflowsDeviceLimitHTML,
                        RemainingDevices: &remainingDevices,
                }
                err = wflows.SubmitDeviceLimitWarning(ctx, warnWFlow)
                if err != nil {
                        return err
                }
        }
        return nil
}

// CheckDeviceLimits goes through all tenant databases and checks if the number
// of accepted devices is above a given threshold (in %) and sends an email
// to all registered users registered under the given tenant.
func CheckDeviceLimits(
        threshold float64,
        ds store.DataStore,
        tadm tenant.ClientRunner,
        wflows orchestrator.ClientRunner,
) error {
        // Sanitize threshold
        if threshold > 100.0 {
                threshold = 100.0
        } else if threshold < 0.0 {
                threshold = 0.0
        }
        threshProportion := threshold / 100.0

        // mapFunc is applied to all existing databases in datastore.
        mapFunc := func(ctx context.Context) error {
                id := identity.FromContext(ctx)
                if id == nil || id.Tenant == "" {
                        // Not a tenant db - skip!
                        return nil
                }
                tenantID := id.Tenant
                l := log.FromContext(ctx)

                lim, err := ds.GetLimit(ctx, model.LimitMaxDeviceCount)
                if err != nil {
                        return err
                }
                n, err := ds.GetDevCountByStatus(ctx, model.DevStatusAccepted)
                if err != nil {
                        return err
                }
                if float64(n) >= (float64(lim.Value) * threshProportion) {
                        // User is above limit

                        remainingUsers := uint(n) - uint(lim.Value)
                        err := warnTenantUsers(ctx, tenantID, tadm, wflows, remainingUsers)
                        if err != nil {
                                l.Warnf(`Failed to warn tenant "%s" `+
                                        `users nearing device limit: %s`,
                                        tenantID, err.Error(),
                                )
                        }
                }
                return nil
        }
        // Start looping through the databases.
        return ds.ForEachDatabase(
                context.Background(),
                mapFunc,
        )
}

1	// Copyright 2023 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 cmd
15
16	import (
17	"context"
18	"fmt"
19
20	"github.com/mendersoftware/go-lib-micro/config"
21	"github.com/mendersoftware/go-lib-micro/identity"
22	"github.com/mendersoftware/go-lib-micro/log"
23	"github.com/mendersoftware/go-lib-micro/mongo/migrate"
24	mstore "github.com/mendersoftware/go-lib-micro/store"
25	"github.com/pkg/errors"
26
27	cinv "github.com/mendersoftware/deviceauth/client/inventory"
28	"github.com/mendersoftware/deviceauth/client/orchestrator"
29	"github.com/mendersoftware/deviceauth/client/tenant"
30	dconfig "github.com/mendersoftware/deviceauth/config"
31	"github.com/mendersoftware/deviceauth/model"
32	"github.com/mendersoftware/deviceauth/store"
33	"github.com/mendersoftware/deviceauth/store/mongo"
34	"github.com/mendersoftware/deviceauth/utils"
35	)
36
37	var NowUnixMilis = utils.UnixMilis
38
39	func makeDataStoreConfig() mongo.DataStoreMongoConfig {	1✔
40	return mongo.DataStoreMongoConfig{	1✔
41	ConnectionString: config.Config.GetString(dconfig.SettingDb),	1✔
42		1✔
43	SSL: config.Config.GetBool(dconfig.SettingDbSSL),	1✔
44	SSLSkipVerify: config.Config.GetBool(dconfig.SettingDbSSLSkipVerify),	1✔
45		1✔
46	Username: config.Config.GetString(dconfig.SettingDbUsername),	1✔
47	Password: config.Config.GetString(dconfig.SettingDbPassword),	1✔
48	}	1✔
49		1✔
50	}	1✔
51
52	func Migrate(c config.Reader, tenant string, listTenantsFlag bool) error {	1✔
53	db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())	1✔
54		1✔
55	if err != nil {	1✔
UNCOV 56	return errors.Wrap(err, "failed to connect to db")	×
57	}	×
58
59	// list tenants only
60	if listTenantsFlag {	2✔
61	return listTenants(db)	1✔
62	}	1✔
63
64	db = db.WithAutomigrate().(*mongo.DataStoreMongo)	1✔
65		1✔
66	if config.Config.Get(dconfig.SettingTenantAdmAddr) != "" {	2✔
67	db = db.WithMultitenant()	1✔
68	}	1✔
69
70	ctx := context.Background()	1✔
71	if tenant == "" {	2✔
72	err = db.Migrate(ctx, mongo.DbVersion)	1✔
73	} else {	2✔
74	tenantCtx := identity.WithContext(ctx, &identity.Identity{	1✔
75	Tenant: tenant,	1✔
76	})	1✔
77	dbname := mstore.DbFromContext(tenantCtx, mongo.DbName)	1✔
78	err = db.MigrateTenant(tenantCtx, dbname, mongo.DbVersion)	1✔
79	}	1✔
80	if err != nil {	1✔
UNCOV 81	return errors.Wrap(err, "failed to run migrations")	×
82	}	×
83
84	return nil	1✔
85	}
86
87	func listTenants(db *mongo.DataStoreMongo) error {	1✔
88	tdbs, err := db.GetTenantDbs()	1✔
89	if err != nil {	1✔
UNCOV 90	return errors.Wrap(err, "failed to retrieve tenant DBs")	×
91	}	×
92
93	for _, tenant := range tdbs {	2✔
94	fmt.Println(mstore.TenantFromDbName(tenant, mongo.DbName))	1✔
95	}	1✔
96
97	return nil	1✔
98	}
99
UNCOV 100	func Maintenance(decommissioningCleanupFlag bool, tenant string, dryRunFlag bool) error {	×
101	db, err := mongo.NewDataStoreMongo(makeDataStoreConfig())	×
102	if err != nil {	×
103	return errors.Wrap(err, "failed to connect to db")	×
104	}	×
105
UNCOV 106	return maintenanceWithDataStore(decommissioningCleanupFlag, tenant, dryRunFlag, db)	×
107	}
108
109	func maintenanceWithDataStore(
110	decommissioningCleanupFlag bool,
111	tenant string,
112	dryRunFlag bool,
113	db *mongo.DataStoreMongo,
114	) error {	16✔
115	// cleanup devauth database from leftovers after failed decommissioning	16✔
116	if decommissioningCleanupFlag {	28✔
117	return decommissioningCleanup(db, tenant, dryRunFlag)	12✔
118	}	12✔
119
120	return nil	4✔
121	}
122
123	func decommissioningCleanup(db *mongo.DataStoreMongo, tenant string, dryRunFlag bool) error {	12✔
124	if tenant == "" {	20✔
125	tdbs, err := db.GetTenantDbs()	8✔
126	if err != nil {	8✔
UNCOV 127	return errors.Wrap(err, "failed to retrieve tenant DBs")	×
128	}	×
129	_ = decommissioningCleanupWithDbs(db, append(tdbs, mongo.DbName), dryRunFlag)	8✔
130	} else {	4✔
131	_ = decommissioningCleanupWithDbs(	4✔
132	db,	4✔
133	[]string{mstore.DbNameForTenant(tenant, mongo.DbName)},	4✔
134	dryRunFlag,	4✔
135	)	4✔
136	}	4✔
137
138	return nil	12✔
139	}
140
141	func decommissioningCleanupWithDbs(
142	db *mongo.DataStoreMongo,
143	tenantDbs []string,
144	dryRunFlag bool,
145	) error {	12✔
146	for _, dbName := range tenantDbs {	24✔
147	println("database: ", dbName)	12✔
148	if err := decommissioningCleanupWithDb(db, dbName, dryRunFlag); err != nil {	12✔
UNCOV 149	return err	×
150	}	×
151	}
152	return nil	12✔
153	}
154
155	func decommissioningCleanupWithDb(db *mongo.DataStoreMongo, dbName string, dryRunFlag bool) error {	12✔
156	if dryRunFlag {	18✔
157	return decommissioningCleanupDryRun(db, dbName)	6✔
158	} else {	12✔
159	return decommissioningCleanupExecute(db, dbName)	6✔
160	}	6✔
161	}
162
163	func decommissioningCleanupDryRun(db *mongo.DataStoreMongo, dbName string) error {	12✔
164	//devices	12✔
165	devices, err := db.GetDevicesBeingDecommissioned(dbName)	12✔
166	if err != nil {	12✔
UNCOV 167	return err	×
168	}	×
169	if len(devices) > 0 {	20✔
170	fmt.Println("devices with decommissioning flag set:")	8✔
171	for _, dev := range devices {	16✔
172	fmt.Println(dev.Id)	8✔
173	}	8✔
174	}
175
176	//auth sets
177	authSetIds, err := db.GetBrokenAuthSets(dbName)	12✔
178	if err != nil {	12✔
UNCOV 179	return err	×
180	}	×
181	if len(authSetIds) > 0 {	20✔
182	fmt.Println("authentication sets to be removed:")	8✔
183	for _, authSetId := range authSetIds {	16✔
184	fmt.Println(authSetId)	8✔
185	}	8✔
186	}
187
188	return nil	12✔
189	}
190
191	func decommissioningCleanupExecute(db *mongo.DataStoreMongo, dbName string) error {	6✔
192	if err := decommissioningCleanupDryRun(db, dbName); err != nil {	6✔
UNCOV 193	return err	×
194	}	×
195
196	if err := db.DeleteDevicesBeingDecommissioned(dbName); err != nil {	6✔
UNCOV 197	return err	×
198	}	×
199
200	if err := db.DeleteBrokenAuthSets(dbName); err != nil {	6✔
UNCOV 201	return err	×
202	}	×
203
204	return nil	6✔
205	}
206
207	func PropagateStatusesInventory(
208	db store.DataStore,
209	c cinv.Client,
210	tenant string,
211	migrationVersion string,
212	dryRun bool,
213	) error {	18✔
214	l := log.NewEmpty()	18✔
215		18✔
216	dbs, err := selectDbs(db, tenant)	18✔
217	if err != nil {	20✔
218	return errors.Wrap(err, "aborting")	2✔
219	}	2✔
220
221	var errReturned error	16✔
222	for _, d := range dbs {	38✔
223	err := tryPropagateStatusesInventoryForDb(db, c, d, migrationVersion, dryRun)	22✔
224	if err != nil {	32✔
225	errReturned = err	10✔
226	l.Errorf("giving up on DB %s due to fatal error: %s", d, err.Error())	10✔
227	continue	10✔
228	}
229	}
230
231	l.Info("all DBs processed, exiting.")	16✔
232	return errReturned	16✔
233	}
234
UNCOV 235	func PropagateIdDataInventory(db store.DataStore, c cinv.Client, tenant string, dryRun bool) error {	×
236	l := log.NewEmpty()	×
237		×
238	dbs, err := selectDbs(db, tenant)	×
239	if err != nil {	×
240	return errors.Wrap(err, "aborting")	×
241	}	×
242
UNCOV 243	var errReturned error	×
244	for _, d := range dbs {	×
245	err := tryPropagateIdDataInventoryForDb(db, c, d, dryRun)	×
246	if err != nil {	×
247	errReturned = err	×
248	l.Errorf("giving up on DB %s due to fatal error: %s", d, err.Error())	×
249	continue	×
250	}
251	}
252
UNCOV 253	l.Info("all DBs processed, exiting.")	×
254	return errReturned	×
255	}
256
257	func PropagateReporting(db store.DataStore, wflows orchestrator.ClientRunner, tenant string,
258	dryRun bool) error {	14✔
259	l := log.NewEmpty()	14✔
260		14✔
261	dbs, err := selectDbs(db, tenant)	14✔
262	if err != nil {	16✔
263	return errors.Wrap(err, "aborting")	2✔
264	}	2✔
265
266	var errReturned error	12✔
267	for _, d := range dbs {	30✔
268	err := tryPropagateReportingForDb(db, wflows, d, dryRun)	18✔
269	if err != nil {	26✔
270	errReturned = err	8✔
271	l.Errorf("giving up on DB %s due to fatal error: %s", d, err.Error())	8✔
272	continue	8✔
273	}
274	}
275
276	l.Info("all DBs processed, exiting.")	12✔
277	return errReturned	12✔
278	}
279
280	func selectDbs(db store.DataStore, tenant string) ([]string, error) {	32✔
281	l := log.NewEmpty()	32✔
282		32✔
283	var dbs []string	32✔
284		32✔
285	if tenant != "" {	40✔
286	l.Infof("propagating inventory for user-specified tenant %s", tenant)	8✔
287	n := mstore.DbNameForTenant(tenant, mongo.DbName)	8✔
288	dbs = []string{n}	8✔
289	} else {	32✔
290	l.Infof("propagating inventory for all tenants")	24✔
291		24✔
292	// infer if we're in ST or MT	24✔
293	tdbs, err := db.GetTenantDbs()	24✔
294	if err != nil {	28✔
295	return nil, errors.Wrap(err, "failed to retrieve tenant DBs")	4✔
296	}	4✔
297
298	if len(tdbs) == 0 {	28✔
299	l.Infof("no tenant DBs found - will try the default database %s", mongo.DbName)	8✔
300	dbs = []string{mongo.DbName}	8✔
301	} else {	20✔
302	dbs = tdbs	12✔
303	}	12✔
304	}
305
306	return dbs, nil	28✔
307	}
308
309	const (
310	devicesBatchSize = 512
311	)
312
313	func updateDevicesStatus(
314	ctx context.Context,
315	db store.DataStore,
316	c cinv.Client,
317	tenant string,
318	status string,
319	dryRun bool,
320	) error {	110✔
321	var skip uint	110✔
322		110✔
323	skip = 0	110✔
324	for {	220✔
325	devices, err := db.GetDevices(ctx,	110✔
326	skip,	110✔
327	devicesBatchSize,	110✔
328	model.DeviceFilter{Status: []string{status}},	110✔
329	)	110✔
330	if err != nil {	130✔
331	return errors.Wrap(err, "failed to get devices")	20✔
332	}	20✔
333
334	if len(devices) < 1 {	90✔
UNCOV 335	break	×
336	}
337
338	deviceUpdates := make([]model.DeviceInventoryUpdate, len(devices))	90✔
339		90✔
340	for i, d := range devices {	290✔
341	deviceUpdates[i].Id = d.Id	200✔
342	deviceUpdates[i].Revision = d.Revision	200✔
343	}	200✔
344
345	if !dryRun {	170✔
346	err = c.SetDeviceStatus(ctx, tenant, deviceUpdates, status)	80✔
347	if err != nil {	100✔
348	return err	20✔
349	}	20✔
350	}
351
352	if len(devices) < devicesBatchSize {	140✔
353	break	70✔
UNCOV 354	} else {	×
UNCOV 355	skip += devicesBatchSize	×
UNCOV 356	}	×
357	}
358	return nil	70✔
359	}
360
361	func updateDevicesIdData(
362	ctx context.Context,
363	db store.DataStore,
364	c cinv.Client,
365	tenant string,
366	dryRun bool,
UNCOV 367	) error {	×
UNCOV 368	var skip uint	×
UNCOV 369		×
UNCOV 370	skip = 0	×
UNCOV 371	for {	×
372	devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})	×
373	if err != nil {	×
374	return errors.Wrap(err, "failed to get devices")	×
375	}	×
376
377	if len(devices) < 1 {	×
378	break	×
379	}
380
UNCOV 381	if !dryRun {	×
382	for _, d := range devices {	×
383	err := c.SetDeviceIdentity(ctx, tenant, d.Id, d.IdDataStruct)	×
UNCOV 384	if err != nil {	×
UNCOV 385	return err	×
386	}	×
387	}
388	}
389
390	skip += devicesBatchSize	×
391	if len(devices) < devicesBatchSize {	×
UNCOV 392	break	×
393	}
394	}
395	return nil	×
396	}
397
398	func tryPropagateStatusesInventoryForDb(
399	db store.DataStore,
400	c cinv.Client,
401	dbname string,
402	migrationVersion string,
403	dryRun bool,
404	) error {	22✔
405	l := log.NewEmpty()	22✔
406		22✔
407	l.Infof("propagating device statuses to inventory from DB: %s", dbname)	22✔
408		22✔
409	tenant := mstore.TenantFromDbName(dbname, mongo.DbName)	22✔
410		22✔
411	ctx := context.Background()	22✔
412	if tenant != "" {	40✔
413	ctx = identity.WithContext(ctx, &identity.Identity{	18✔
414	Tenant: tenant,	18✔
415	})	18✔
416	}	18✔
417
418	var err error	22✔
419	var errReturned error	22✔
420	for _, status := range model.DevStatuses {	132✔
421	err = updateDevicesStatus(ctx, db, c, tenant, status, dryRun)	110✔
422	if err != nil {	150✔
423	l.Infof(	40✔
424	"Done with DB %s status=%s, but there were errors: %s.",	40✔
425	dbname,	40✔
426	status,	40✔
427	err.Error(),	40✔
428	)	40✔
429	errReturned = err	40✔
430	} else {	110✔
431	l.Infof("Done with DB %s status=%s", dbname, status)	70✔
432	}	70✔
433	}
434	if migrationVersion != "" && !dryRun {	26✔
435	if errReturned != nil {	4✔
UNCOV 436	l.Warnf(	×
UNCOV 437	"Will not store %s migration version in %s.migration_info due to errors.",	×
UNCOV 438	migrationVersion,	×
UNCOV 439	dbname,	×
UNCOV 440	)	×
441	} else {	4✔
442	version, err := migrate.NewVersion(migrationVersion)	4✔
443	if version == nil \|\| err != nil {	6✔
444	l.Warnf(	2✔
445	"Will not store %s migration version in %s.migration_info due to bad version"+	2✔
446	" provided.",	2✔
447	migrationVersion,	2✔
448	dbname,	2✔
449	)	2✔
450	errReturned = err	2✔
451	} else {	4✔
452	_ = db.StoreMigrationVersion(ctx, version)	2✔
453	}	2✔
454	}
455	}
456
457	return errReturned	22✔
458	}
459
460	func tryPropagateIdDataInventoryForDb(
461	db store.DataStore,
462	c cinv.Client,
463	dbname string,
464	dryRun bool,
UNCOV 465	) error {	×
UNCOV 466	l := log.NewEmpty()	×
UNCOV 467		×
UNCOV 468	l.Infof("propagating device id_data to inventory from DB: %s", dbname)	×
UNCOV 469		×
470	tenant := mstore.TenantFromDbName(dbname, mongo.DbName)	×
471		×
472	ctx := context.Background()	×
473	if tenant != "" {	×
474	ctx = identity.WithContext(ctx, &identity.Identity{	×
475	Tenant: tenant,	×
476	})	×
477	}	×
478
479	err := updateDevicesIdData(ctx, db, c, tenant, dryRun)	×
480	if err != nil {	×
481	l.Infof("Done with DB %s, but there were errors: %s.", dbname, err.Error())	×
482	} else {	×
UNCOV 483	l.Infof("Done with DB %s", dbname)	×
484	}	×
485
486	return err	×
487	}
488
489	func tryPropagateReportingForDb(
490	db store.DataStore,
491	wflows orchestrator.ClientRunner,
492	dbname string,
493	dryRun bool,
494	) error {	18✔
495	l := log.NewEmpty()	18✔
496		18✔
497	l.Infof("propagating device data to reporting from DB: %s", dbname)	18✔
498		18✔
499	tenant := mstore.TenantFromDbName(dbname, mongo.DbName)	18✔
500		18✔
501	ctx := context.Background()	18✔
502	if tenant != "" {	32✔
503	ctx = identity.WithContext(ctx, &identity.Identity{	14✔
504	Tenant: tenant,	14✔
505	})	14✔
506	}	14✔
507
508	err := reindexDevicesReporting(ctx, db, wflows, tenant, dryRun)	18✔
509	if err != nil {	26✔
510	l.Infof("Done with DB %s, but there were errors: %s.", dbname, err.Error())	8✔
511	} else {	18✔
512	l.Infof("Done with DB %s", dbname)	10✔
513	}	10✔
514
515	return err	18✔
516	}
517
518	func reindexDevicesReporting(
519	ctx context.Context,
520	db store.DataStore,
521	wflows orchestrator.ClientRunner,
522	tenant string,
523	dryRun bool,
524	) error {	18✔
525	var skip uint	18✔
526		18✔
527	skip = 0	18✔
528	for {	36✔
529	devices, err := db.GetDevices(ctx, skip, devicesBatchSize, model.DeviceFilter{})	18✔
530	if err != nil {	22✔
531	return errors.Wrap(err, "failed to get devices")	4✔
532	}	4✔
533
534	if len(devices) < 1 {	14✔
UNCOV 535	break	×
536	}
537
538	if !dryRun {	26✔
539	deviceIDs := make([]string, len(devices))	12✔
540	for i, d := range devices {	40✔
541	deviceIDs[i] = d.Id	28✔
542	}	28✔
543	err := wflows.SubmitReindexReportingBatch(ctx, deviceIDs)	12✔
544	if err != nil {	16✔
545	return err	4✔
546	}	4✔
547	}
548
549	skip += devicesBatchSize	10✔
550	if len(devices) < devicesBatchSize {	20✔
551	break	10✔
552	}
553	}
554	return nil	10✔
555	}
556
557	const (
558	WorkflowsDeviceLimitText = "@/etc/workflows-enterprise/data/device_limit_email.txt"
559	WorkflowsDeviceLimitHTML = "@/etc/workflows-enterprise/data/device_limit_email.html"
560	WorkflowsDeviceLimitSubject = "Device limit almost reached"
561	)
562
563	func warnTenantUsers(
564	ctx context.Context,
565	tenantID string,
566	tadm tenant.ClientRunner,
567	wflows orchestrator.ClientRunner,
568	remainingDevices uint,
569	) error {	11✔
570	users, err := tadm.GetTenantUsers(ctx, tenantID)	11✔
571	if err != nil {	13✔
572	// Log the event and continue with the other tenants	2✔
573	return err	2✔
574	}	2✔
575	for i := range users {	24✔
576	warnWFlow := orchestrator.DeviceLimitWarning{	15✔
577	RequestID: "deviceAuthAdmin",	15✔
578	RecipientEmail: users[i].Email,	15✔
579		15✔
580	Subject: WorkflowsDeviceLimitSubject,	15✔
581	Body: WorkflowsDeviceLimitText,	15✔
582	BodyHTML: WorkflowsDeviceLimitHTML,	15✔
583	RemainingDevices: &remainingDevices,	15✔
584	}	15✔
585	err = wflows.SubmitDeviceLimitWarning(ctx, warnWFlow)	15✔
586	if err != nil {	17✔
587	return err	2✔
588	}	2✔
589	}
590	return nil	7✔
591	}
592
593	// CheckDeviceLimits goes through all tenant databases and checks if the number
594	// of accepted devices is above a given threshold (in %) and sends an email
595	// to all registered users registered under the given tenant.
596	func CheckDeviceLimits(
597	threshold float64,
598	ds store.DataStore,
599	tadm tenant.ClientRunner,
600	wflows orchestrator.ClientRunner,
601	) error {	11✔
602	// Sanitize threshold	11✔
603	if threshold > 100.0 {	13✔
604	threshold = 100.0	2✔
605	} else if threshold < 0.0 {	13✔
606	threshold = 0.0	2✔
607	}	2✔
608	threshProportion := threshold / 100.0	11✔
609		11✔
610	// mapFunc is applied to all existing databases in datastore.	11✔
611	mapFunc := func(ctx context.Context) error {	36✔
612	id := identity.FromContext(ctx)	25✔
613	if id == nil \|\| id.Tenant == "" {	27✔
614	// Not a tenant db - skip!	2✔
615	return nil	2✔
616	}	2✔
617	tenantID := id.Tenant	23✔
618	l := log.FromContext(ctx)	23✔
619		23✔
620	lim, err := ds.GetLimit(ctx, model.LimitMaxDeviceCount)	23✔
621	if err != nil {	25✔
622	return err	2✔
623	}	2✔
624	n, err := ds.GetDevCountByStatus(ctx, model.DevStatusAccepted)	21✔
625	if err != nil {	23✔
626	return err	2✔
627	}	2✔
628	if float64(n) >= (float64(lim.Value) * threshProportion) {	30✔
629	// User is above limit	11✔
630		11✔
631	remainingUsers := uint(n) - uint(lim.Value)	11✔
632	err := warnTenantUsers(ctx, tenantID, tadm, wflows, remainingUsers)	11✔
633	if err != nil {	15✔
634	l.Warnf(`Failed to warn tenant "%s" `+	4✔
635	`users nearing device limit: %s`,	4✔
636	tenantID, err.Error(),	4✔
637	)	4✔
638	}	4✔
639	}
640	return nil	19✔
641	}
642	// Start looping through the databases.
643	return ds.ForEachDatabase(	11✔
644	context.Background(),	11✔
645	mapFunc,	11✔
646	)	11✔
647	}

mendersoftware / deviceauth / 857655394

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