diff --git a/server/channels/api4/job_test.go b/server/channels/api4/job_test.go index 7f060e96126..5c9f4984397 100644 --- a/server/channels/api4/job_test.go +++ b/server/channels/api4/job_test.go @@ -51,6 +51,31 @@ func TestCreateJob(t *testing.T) { }) } +func TestCreatePatExpiryNotifyJob(t *testing.T) { + mainHelper.Parallel(t) + th := Setup(t) + + th.LoginSystemManager(t) + + job := &model.Job{Type: model.JobTypePatExpiryNotify} + + t.Run("forbidden without manage_jobs permission", func(t *testing.T) { + _, resp, err := th.SystemManagerClient.CreateJob(context.Background(), job) + require.Error(t, err) + CheckForbiddenStatus(t, resp) + }) + + t.Run("allowed as system admin", func(t *testing.T) { + received, _, err := th.SystemAdminClient.CreateJob(context.Background(), job) + require.NoError(t, err) + defer func() { + result, appErr := th.App.Srv().Store().Job().Delete(received.Id) + require.NoErrorf(t, appErr, "Failed to delete job (result: %v): %v", result, appErr) + }() + require.Equal(t, model.JobTypePatExpiryNotify, received.Type) + }) +} + func TestGetJob(t *testing.T) { mainHelper.Parallel(t) th := Setup(t) diff --git a/server/channels/app/job.go b/server/channels/app/job.go index 63c7e2cf311..c7401acee53 100644 --- a/server/channels/app/job.go +++ b/server/channels/app/job.go @@ -224,7 +224,8 @@ func (a *App) SessionHasPermissionToCreateJob(session model.Session, job *model. model.JobTypeExportDelete, model.JobTypeCloud, model.JobTypeExtractContent, - model.JobTypeCleanupExpiredAccessTokens: + model.JobTypeCleanupExpiredAccessTokens, + model.JobTypePatExpiryNotify: return a.SessionHasPermissionTo(session, model.PermissionManageJobs), model.PermissionManageJobs case model.JobTypeAccessControlSync: // Allow system admins to create access control sync jobs diff --git a/server/channels/app/patexpirynotify.go b/server/channels/app/patexpirynotify.go new file mode 100644 index 00000000000..c6649b5fca2 --- /dev/null +++ b/server/channels/app/patexpirynotify.go @@ -0,0 +1,158 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "net/http" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/i18n" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/public/shared/request" +) + +// patExpiryNotifyBatchLimit bounds the number of tokens processed per run. +// GetExpiringTokens returns only actionable rows (most urgent first), so this +// caps work per run while a backlog larger than the limit drains across +// subsequent hourly runs without starving the least-urgent tokens. +const patExpiryNotifyBatchLimit = 1000 + +// patExpiryThresholds is the pre-expiry warning cascade, in days, ordered from +// most to least urgent. A token owner is warned once as the token crosses into +// each bucket. +var patExpiryThresholds = []int{1, 3, 7} + +// NotifyPersonalAccessTokensExpiring warns the owners of personal access tokens +// that are approaching expiry, on a fixed 7 / 3 / 1 day cascade. It is invoked +// hourly by the pat_expiry_notify job. +// +// Bot tokens and tokens owned by deactivated users are excluded by the store +// query. For each remaining token this computes the current warning bucket, +// sends a single system-bot DM, and records the per-token LastNotifiedAt marker +// so the same (or a less urgent) warning is never re-sent. Because only +// the most urgent applicable bucket is ever sent, a token that first becomes +// visible already inside the window (e.g. created with a short lifetime, or owned +// by a just-reactivated user) gets a single warning rather than a catch-up burst +// of every threshold it has already passed. +func (a *App) NotifyPersonalAccessTokensExpiring() error { + if !*a.Config().ServiceSettings.EnableUserAccessTokens { + return nil + } + + rctx := request.EmptyContext(a.Log().With(mlog.String("component", "pat_expiry_notify"))) + + now := model.GetMillis() + + tokens, err := a.Srv().Store().UserAccessToken().GetExpiringTokens(now, patExpiryThresholds, patExpiryNotifyBatchLimit) + if err != nil { + return model.NewAppError("NotifyPersonalAccessTokensExpiring", "app.user_access_token.get_expiring.app_error", nil, "", http.StatusInternalServerError).Wrap(err) + } + if len(tokens) == 0 { + return nil + } + + systemBot, appErr := a.GetSystemBot(rctx) + if appErr != nil { + return appErr + } + + for _, token := range tokens { + bucket := patExpiryBucket(token.ExpiresAt, now) + if bucket == 0 { + // Already expired or outside the cascade; the store window should + // prevent this, but guard against clock skew between runs. + continue + } + + // Skip if the owner has already been warned at this bucket or a more + // urgent (smaller) one. A prior warning covers the current bucket when it + // was sent while no more time remained than the bucket spans, i.e. + // LastNotifiedAt >= ExpiresAt - bucket. nil means no warning sent yet. + if token.LastNotifiedAt != nil && *token.LastNotifiedAt >= token.ExpiresAt-int64(bucket)*model.DayInMilliseconds { + continue + } + + if appErr := a.sendPATExpiryNotification(rctx, systemBot, token, bucket); appErr != nil { + rctx.Logger().Error("Failed to send personal access token expiry notification", + mlog.String("token_id", token.Id), + mlog.String("user_id", token.UserId), + mlog.Err(appErr), + ) + continue + } + + if err := a.Srv().Store().UserAccessToken().UpdateLastNotifiedAt(token.Id, now); err != nil { + rctx.Logger().Error("Failed to update LastNotifiedAt for personal access token", + mlog.String("token_id", token.Id), + mlog.Err(err), + ) + } + } + + return nil +} + +// patExpiryBucket returns the most urgent warning threshold (in days) that the +// token currently qualifies for, or 0 if it is already expired or further out +// than the largest threshold. A token with N days remaining maps to the smallest +// threshold T such that N <= T (e.g. 2 days remaining -> the 3-day bucket). +func patExpiryBucket(expiresAt int64, now int64) int { + remaining := expiresAt - now + if remaining <= 0 { + return 0 + } + for _, threshold := range patExpiryThresholds { + if remaining <= int64(threshold)*model.DayInMilliseconds { + return threshold + } + } + return 0 +} + +func (a *App) sendPATExpiryNotification(rctx request.CTX, systemBot *model.Bot, token *model.UserAccessToken, bucket int) *model.AppError { + channel, appErr := a.GetOrCreateDirectChannel(rctx, token.UserId, systemBot.UserId) + if appErr != nil { + return appErr + } + + user, appErr := a.GetUser(token.UserId) + if appErr != nil { + return appErr + } + + T := i18n.GetUserTranslations(user.Locale) + + description := token.Description + if description == "" { + description = T("app.pat_expiry_notify.unnamed_token") + } + + // The cascade is 7/3/1; the multi-day message is only ever rendered with a + // plural day count (7 or 3), and the 1-day bucket uses a dedicated "24 hours" + // message, so neither string needs an awkward "day(s)" plural. + var message string + if bucket <= 1 { + message = T("app.pat_expiry_notify.dm_final", map[string]any{ + "Description": description, + }) + } else { + message = T("app.pat_expiry_notify.dm", map[string]any{ + "Description": description, + "Days": bucket, + }) + } + + post := &model.Post{ + ChannelId: channel.Id, + Message: message, + Type: model.PostTypeDefault, + UserId: systemBot.UserId, + } + + if _, _, err := a.CreatePost(rctx, post, channel, model.CreatePostFlags{SetOnline: false}); err != nil { + return err + } + + return nil +} diff --git a/server/channels/app/patexpirynotify_test.go b/server/channels/app/patexpirynotify_test.go new file mode 100644 index 00000000000..4d118e3af1f --- /dev/null +++ b/server/channels/app/patexpirynotify_test.go @@ -0,0 +1,150 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package app + +import ( + "testing" + + "github.com/stretchr/testify/require" + + "github.com/mattermost/mattermost/server/public/model" +) + +func TestPatExpiryBucket(t *testing.T) { + now := int64(1_000_000_000_000) + + testCases := []struct { + name string + expiresAt int64 + expected int + }{ + {"already expired", now - model.DayInMilliseconds, 0}, + {"expires now", now, 0}, + {"12 hours left -> 1-day bucket", now + 12*60*60*1000, 1}, + {"exactly 1 day left -> 1-day bucket", now + 1*model.DayInMilliseconds, 1}, + {"2 days left -> 3-day bucket", now + 2*model.DayInMilliseconds, 3}, + {"exactly 3 days left -> 3-day bucket", now + 3*model.DayInMilliseconds, 3}, + {"5 days left -> 7-day bucket", now + 5*model.DayInMilliseconds, 7}, + {"exactly 7 days left -> 7-day bucket", now + 7*model.DayInMilliseconds, 7}, + {"8 days left -> outside cascade", now + 8*model.DayInMilliseconds, 0}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.expected, patExpiryBucket(tc.expiresAt, now)) + }) + } +} + +func TestNotifyPersonalAccessTokensExpiring(t *testing.T) { + setup := func(t *testing.T) *TestHelper { + th := Setup(t).InitBasic(t) + th.App.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.EnableUserAccessTokens = true }) + return th + } + + systemBotDMMessages := func(t *testing.T, th *TestHelper, userID string) []string { + systemBot, appErr := th.App.GetSystemBot(th.Context) + require.Nil(t, appErr) + channel, appErr := th.App.GetOrCreateDirectChannel(th.Context, userID, systemBot.UserId) + require.Nil(t, appErr) + posts, appErr := th.App.GetPosts(th.Context, channel.Id, 0, 50) + require.Nil(t, appErr) + var messages []string + for _, p := range posts.Posts { + if p.UserId == systemBot.UserId { + messages = append(messages, p.Message) + } + } + return messages + } + + t.Run("disabled when EnableUserAccessTokens is off", func(t *testing.T) { + th := setup(t) + th.App.UpdateConfig(func(cfg *model.Config) { *cfg.ServiceSettings.EnableUserAccessTokens = false }) + + token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "t", ExpiresAt: model.GetMillis() + 5*model.DayInMilliseconds} + _, err := th.App.Srv().Store().UserAccessToken().Save(token) + require.NoError(t, err) + + require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring()) + require.Empty(t, systemBotDMMessages(t, th, th.BasicUser.Id)) + }) + + t.Run("warns once and dedups on re-run", func(t *testing.T) { + th := setup(t) + + token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "deploy bot", ExpiresAt: model.GetMillis() + 5*model.DayInMilliseconds} + _, err := th.App.Srv().Store().UserAccessToken().Save(token) + require.NoError(t, err) + + require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring()) + messages := systemBotDMMessages(t, th, th.BasicUser.Id) + require.Len(t, messages, 1, "owner should get exactly one warning") + require.Contains(t, messages[0], "deploy bot", "warning should name the token") + require.Contains(t, messages[0], "7 days or less", "5-day token is in the 7-day bucket") + require.Contains(t, messages[0], "regenerate", "warning should tell the owner to regenerate the token") + require.NotContains(t, messages[0], "day(s)", "message must not use the awkward day(s) plural") + + stored, err := th.App.Srv().Store().UserAccessToken().Get(token.Id) + require.NoError(t, err) + require.NotNil(t, stored.LastNotifiedAt, "last-notified time should be recorded") + + // A second run in the same bucket must not send another DM. + require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring()) + require.Len(t, systemBotDMMessages(t, th, th.BasicUser.Id), 1, "re-run in the same bucket must not duplicate") + }) + + t.Run("token already inside window sends only the current bucket, not a catch-up burst", func(t *testing.T) { + th := setup(t) + + // Token surfaces with only 2 days left (e.g. created short-lived, or owner + // just reactivated). It must get a single warning at the 3-day bucket, never + // the already-passed 7-day one. + token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "short lived", ExpiresAt: model.GetMillis() + 2*model.DayInMilliseconds} + _, err := th.App.Srv().Store().UserAccessToken().Save(token) + require.NoError(t, err) + + require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring()) + messages := systemBotDMMessages(t, th, th.BasicUser.Id) + require.Len(t, messages, 1, "exactly one warning, no 7+3 burst") + require.Contains(t, messages[0], "3 days or less", "2-day token is in the 3-day bucket") + + stored, err := th.App.Srv().Store().UserAccessToken().Get(token.Id) + require.NoError(t, err) + require.NotNil(t, stored.LastNotifiedAt, "last-notified time should be recorded") + }) + + t.Run("sub-day token uses the 24-hour wording", func(t *testing.T) { + th := setup(t) + + token := &model.UserAccessToken{Token: model.NewId(), UserId: th.BasicUser.Id, Description: "almost gone", ExpiresAt: model.GetMillis() + 12*60*60*1000} + _, err := th.App.Srv().Store().UserAccessToken().Save(token) + require.NoError(t, err) + + require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring()) + messages := systemBotDMMessages(t, th, th.BasicUser.Id) + require.Len(t, messages, 1) + require.Contains(t, messages[0], "within 24 hours", "sub-day token uses the dedicated 24-hour message") + require.NotContains(t, messages[0], "day", "the 24-hour message must not mention days") + }) + + t.Run("bot-owned token is not notified", func(t *testing.T) { + th := setup(t) + + bot, appErr := th.App.CreateBot(th.Context, &model.Bot{Username: "expiring_bot", OwnerId: th.BasicUser.Id}) + require.Nil(t, appErr) + + token := &model.UserAccessToken{Token: model.NewId(), UserId: bot.UserId, Description: "bot token", ExpiresAt: model.GetMillis() + 5*model.DayInMilliseconds} + _, err := th.App.Srv().Store().UserAccessToken().Save(token) + require.NoError(t, err) + + require.NoError(t, th.App.NotifyPersonalAccessTokensExpiring()) + require.Empty(t, systemBotDMMessages(t, th, bot.UserId), "bot accounts must not be warned") + + stored, err := th.App.Srv().Store().UserAccessToken().Get(token.Id) + require.NoError(t, err) + require.Nil(t, stored.LastNotifiedAt, "bot token marker must stay unset") + }) +} diff --git a/server/channels/app/server.go b/server/channels/app/server.go index 400296d5d73..29a038b05d5 100644 --- a/server/channels/app/server.go +++ b/server/channels/app/server.go @@ -59,6 +59,7 @@ import ( "github.com/mattermost/mattermost/server/v8/channels/jobs/migrations" "github.com/mattermost/mattermost/server/v8/channels/jobs/mobile_session_metadata" "github.com/mattermost/mattermost/server/v8/channels/jobs/notify_admin" + "github.com/mattermost/mattermost/server/v8/channels/jobs/pat_expiry_notify" "github.com/mattermost/mattermost/server/v8/channels/jobs/plugins" "github.com/mattermost/mattermost/server/v8/channels/jobs/post_persistent_notifications" "github.com/mattermost/mattermost/server/v8/channels/jobs/product_notices" @@ -1747,6 +1748,12 @@ func (s *Server) initJobs() { cleanup_expired_access_tokens.MakeScheduler(s.Jobs), ) + s.Jobs.RegisterJobType( + model.JobTypePatExpiryNotify, + pat_expiry_notify.MakeWorker(s.Jobs, New(ServerConnector(s.Channels())).NotifyPersonalAccessTokensExpiring), + pat_expiry_notify.MakeScheduler(s.Jobs), + ) + s.Jobs.RegisterJobType( model.JobTypeRefreshMaterializedViews, refresh_materialized_views.MakeWorker(s.Jobs, *s.platform.Config().SqlSettings.DriverName), diff --git a/server/channels/db/migrations/migrations.list b/server/channels/db/migrations/migrations.list index b5094138724..c9980270a1c 100644 --- a/server/channels/db/migrations/migrations.list +++ b/server/channels/db/migrations/migrations.list @@ -399,3 +399,5 @@ channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_i channels/db/migrations/postgres/000201_create_property_fields_groupid_updateat_id_index.up.sql channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.down.sql channels/db/migrations/postgres/000202_create_property_values_groupid_updateat_id_index.up.sql +channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.down.sql +channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.up.sql diff --git a/server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.down.sql b/server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.down.sql new file mode 100644 index 00000000000..028d3c1c617 --- /dev/null +++ b/server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.down.sql @@ -0,0 +1 @@ +ALTER TABLE useraccesstokens DROP COLUMN IF EXISTS lastnotifiedat; diff --git a/server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.up.sql b/server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.up.sql new file mode 100644 index 00000000000..c6e80cb3e82 --- /dev/null +++ b/server/channels/db/migrations/postgres/000203_add_lastnotifiedat_to_user_access_tokens.up.sql @@ -0,0 +1 @@ +ALTER TABLE useraccesstokens ADD COLUMN IF NOT EXISTS lastnotifiedat bigint; diff --git a/server/channels/jobs/pat_expiry_notify/scheduler.go b/server/channels/jobs/pat_expiry_notify/scheduler.go new file mode 100644 index 00000000000..bf988383d4c --- /dev/null +++ b/server/channels/jobs/pat_expiry_notify/scheduler.go @@ -0,0 +1,20 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package pat_expiry_notify + +import ( + "time" + + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/v8/channels/jobs" +) + +const schedFreq = 1 * time.Hour + +func MakeScheduler(jobServer *jobs.JobServer) *jobs.PeriodicScheduler { + isEnabled := func(cfg *model.Config) bool { + return *cfg.ServiceSettings.EnableUserAccessTokens + } + return jobs.NewPeriodicScheduler(jobServer, model.JobTypePatExpiryNotify, schedFreq, isEnabled) +} diff --git a/server/channels/jobs/pat_expiry_notify/worker.go b/server/channels/jobs/pat_expiry_notify/worker.go new file mode 100644 index 00000000000..a48b665bf20 --- /dev/null +++ b/server/channels/jobs/pat_expiry_notify/worker.go @@ -0,0 +1,27 @@ +// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved. +// See LICENSE.txt for license information. + +package pat_expiry_notify + +import ( + "github.com/mattermost/mattermost/server/public/model" + "github.com/mattermost/mattermost/server/public/shared/mlog" + "github.com/mattermost/mattermost/server/v8/channels/jobs" +) + +// MakeWorker creates a worker that warns the owners of personal access tokens +// that are approaching expiry. The actual notification logic lives in the app +// layer and is passed in as notifyExpiring, mirroring the expirynotify job. +func MakeWorker(jobServer *jobs.JobServer, notifyExpiring func() error) *jobs.SimpleWorker { + const workerName = "PatExpiryNotify" + + isEnabled := func(cfg *model.Config) bool { + return *cfg.ServiceSettings.EnableUserAccessTokens + } + execute := func(logger mlog.LoggerIFace, job *model.Job) error { + defer jobServer.HandleJobPanic(logger, job) + + return notifyExpiring() + } + return jobs.NewSimpleWorker(workerName, jobServer, execute, isEnabled) +} diff --git a/server/channels/store/retrylayer/retrylayer.go b/server/channels/store/retrylayer/retrylayer.go index 5000a37b283..f83818969fa 100644 --- a/server/channels/store/retrylayer/retrylayer.go +++ b/server/channels/store/retrylayer/retrylayer.go @@ -18252,6 +18252,27 @@ func (s *RetryLayerUserAccessTokenStore) GetExpiredBefore(cutoff int64, limit in } +func (s *RetryLayerUserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) { + + tries := 0 + for { + result, err := s.UserAccessTokenStore.GetExpiringTokens(now, thresholds, limit) + if err == nil { + return result, nil + } + if !isRepeatableError(err) { + return result, err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return result, err + } + timepkg.Sleep(100 * timepkg.Millisecond) + } + +} + func (s *RetryLayerUserAccessTokenStore) Save(token *model.UserAccessToken) (*model.UserAccessToken, error) { tries := 0 @@ -18294,6 +18315,27 @@ func (s *RetryLayerUserAccessTokenStore) Search(term string) ([]*model.UserAcces } +func (s *RetryLayerUserAccessTokenStore) UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error { + + tries := 0 + for { + err := s.UserAccessTokenStore.UpdateLastNotifiedAt(tokenID, notifiedAt) + if err == nil { + return nil + } + if !isRepeatableError(err) { + return err + } + tries++ + if tries >= 3 { + err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures") + return err + } + timepkg.Sleep(100 * timepkg.Millisecond) + } + +} + func (s *RetryLayerUserAccessTokenStore) UpdateTokenDisable(tokenID string) error { tries := 0 diff --git a/server/channels/store/sqlstore/user_access_token_store.go b/server/channels/store/sqlstore/user_access_token_store.go index e35a11ffa63..d311df1610c 100644 --- a/server/channels/store/sqlstore/user_access_token_store.go +++ b/server/channels/store/sqlstore/user_access_token_store.go @@ -33,6 +33,7 @@ func newSqlUserAccessTokenStore(sqlStore *SqlStore) store.UserAccessTokenStore { "UserAccessTokens.Description", "UserAccessTokens.IsActive", "UserAccessTokens.ExpiresAt", + "UserAccessTokens.LastNotifiedAt", ). From("UserAccessTokens") @@ -270,6 +271,89 @@ func (s SqlUserAccessTokenStore) GetExpiredBefore(cutoff int64, limit int) ([]*m return tokens, nil } +// GetExpiringTokens returns active, non-bot tokens belonging to non-deactivated +// users that need a pre-expiry warning for one of the given day thresholds +// (e.g. 7/3/1), ordered most-urgent first, up to the given limit. Bot tokens are +// excluded because bot accounts are exempt from the expiry policy and have no +// human inbox to notify. +// +// Only *actionable* rows are returned: for each threshold T a token qualifies +// when it has entered the T-day bucket (ExpiresAt <= now + T days) and has not +// yet been notified at T or a more urgent bucket. "Not yet notified at T" means +// the last warning (if any) was sent while more than T days remained, i.e. +// LastNotifiedAt < ExpiresAt - T days (or LastNotifiedAt IS NULL). OR-ing this +// across every threshold yields exactly the tokens whose current (most urgent) +// bucket is still un-notified, so tokens already warned at their current bucket +// never consume a result slot — without this, a backlog of already-warned tokens +// ordered ahead of a less-urgent unnotified token could starve it past the limit +// on every run. The worker still recomputes each token's bucket and re-checks the +// last-notified time as a guard against races. The secret Token column is +// intentionally NOT selected — callers use the returned rows for metadata +// (notification, marker update) only. +// +// A non-positive limit or empty thresholds returns an empty slice without +// hitting the DB rather than relying on the int -> uint64 cast (which would +// otherwise wrap a negative value into an enormous unsigned limit and +// effectively disable the bound). +func (s SqlUserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) { + tokens := []*model.UserAccessToken{} + + if limit <= 0 || len(thresholds) == 0 { + return tokens, nil + } + + actionable := sq.Or{} + for _, t := range thresholds { + bucketMillis := int64(t) * model.DayInMilliseconds + actionable = append(actionable, sq.And{ + sq.LtOrEq{"UserAccessTokens.ExpiresAt": now + bucketMillis}, + sq.Or{ + sq.Eq{"UserAccessTokens.LastNotifiedAt": nil}, + sq.Expr("UserAccessTokens.LastNotifiedAt < UserAccessTokens.ExpiresAt - ?", bucketMillis), + }, + }) + } + + query := s.getQueryBuilder(). + Select( + "UserAccessTokens.Id", + "UserAccessTokens.UserId", + "UserAccessTokens.Description", + "UserAccessTokens.IsActive", + "UserAccessTokens.ExpiresAt", + "UserAccessTokens.LastNotifiedAt", + ). + From("UserAccessTokens"). + InnerJoin("Users ON Users.Id = UserAccessTokens.UserId"). + LeftJoin("Bots ON Bots.UserId = UserAccessTokens.UserId"). + Where(sq.Eq{"UserAccessTokens.IsActive": true}). + Where(sq.Gt{"UserAccessTokens.ExpiresAt": now}). + Where(sq.Eq{"Users.DeleteAt": 0}). + Where(sq.Eq{"Bots.UserId": nil}). + Where(actionable). + OrderBy("UserAccessTokens.ExpiresAt ASC"). + Limit(uint64(limit)) + + // Read from master: this dedups against LastNotifiedAt (written to master), + // so a lagging replica could re-surface an already-warned token and send a + // duplicate. It runs at most once per job run, so master costs nothing here. + if err := s.GetMaster().SelectBuilder(&tokens, query); err != nil { + return nil, errors.Wrap(err, "failed to find expiring UserAccessTokens") + } + + return tokens, nil +} + +// UpdateLastNotifiedAt records the time (Unix milliseconds) at which the token +// owner was most recently warned about the token's upcoming expiry, so the hourly +// pat_expiry_notify job does not re-send the same warning on subsequent runs. +func (s SqlUserAccessTokenStore) UpdateLastNotifiedAt(tokenId string, notifiedAt int64) error { + if _, err := s.GetMaster().Exec("UPDATE UserAccessTokens SET LastNotifiedAt = ? WHERE Id = ?", notifiedAt, tokenId); err != nil { + return errors.Wrapf(err, "failed to update LastNotifiedAt for UserAccessToken with id=%s", tokenId) + } + return nil +} + // CountNonCompliantExpiry returns the number of active, non-bot tokens that // violate the maximum lifetime policy implied by maxExpiresAt. It is used to // preview the blast radius before revoking. diff --git a/server/channels/store/store.go b/server/channels/store/store.go index c3af7b10498..f7749cc98a2 100644 --- a/server/channels/store/store.go +++ b/server/channels/store/store.go @@ -853,11 +853,13 @@ type UserAccessTokenStore interface { GetByToken(tokenString string) (*model.UserAccessToken, error) GetByUser(userID string, page, perPage int) ([]*model.UserAccessToken, error) GetExpiredBefore(cutoff int64, limit int) ([]*model.UserAccessToken, error) + GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) CountNonCompliantExpiry(maxExpiresAt int64) (int64, error) DeleteNonCompliantExpiry(maxExpiresAt int64, limit int) ([]string, error) Search(term string) ([]*model.UserAccessToken, error) UpdateTokenEnable(tokenID string) error UpdateTokenDisable(tokenID string) error + UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error } type PluginStore interface { diff --git a/server/channels/store/storetest/mocks/UserAccessTokenStore.go b/server/channels/store/storetest/mocks/UserAccessTokenStore.go index 0546df2c54a..8b31ea81394 100644 --- a/server/channels/store/storetest/mocks/UserAccessTokenStore.go +++ b/server/channels/store/storetest/mocks/UserAccessTokenStore.go @@ -286,6 +286,36 @@ func (_m *UserAccessTokenStore) GetExpiredBefore(cutoff int64, limit int) ([]*mo return r0, r1 } +// GetExpiringTokens provides a mock function with given fields: now, thresholds, limit +func (_m *UserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) { + ret := _m.Called(now, thresholds, limit) + + if len(ret) == 0 { + panic("no return value specified for GetExpiringTokens") + } + + var r0 []*model.UserAccessToken + var r1 error + if rf, ok := ret.Get(0).(func(int64, []int, int) ([]*model.UserAccessToken, error)); ok { + return rf(now, thresholds, limit) + } + if rf, ok := ret.Get(0).(func(int64, []int, int) []*model.UserAccessToken); ok { + r0 = rf(now, thresholds, limit) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*model.UserAccessToken) + } + } + + if rf, ok := ret.Get(1).(func(int64, []int, int) error); ok { + r1 = rf(now, thresholds, limit) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // Save provides a mock function with given fields: token func (_m *UserAccessTokenStore) Save(token *model.UserAccessToken) (*model.UserAccessToken, error) { ret := _m.Called(token) @@ -346,6 +376,24 @@ func (_m *UserAccessTokenStore) Search(term string) ([]*model.UserAccessToken, e return r0, r1 } +// UpdateLastNotifiedAt provides a mock function with given fields: tokenID, notifiedAt +func (_m *UserAccessTokenStore) UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error { + ret := _m.Called(tokenID, notifiedAt) + + if len(ret) == 0 { + panic("no return value specified for UpdateLastNotifiedAt") + } + + var r0 error + if rf, ok := ret.Get(0).(func(string, int64) error); ok { + r0 = rf(tokenID, notifiedAt) + } else { + r0 = ret.Error(0) + } + + return r0 +} + // UpdateTokenDisable provides a mock function with given fields: tokenID func (_m *UserAccessTokenStore) UpdateTokenDisable(tokenID string) error { ret := _m.Called(tokenID) diff --git a/server/channels/store/storetest/user_access_token_store.go b/server/channels/store/storetest/user_access_token_store.go index bea2ab93614..2e835e3a005 100644 --- a/server/channels/store/storetest/user_access_token_store.go +++ b/server/channels/store/storetest/user_access_token_store.go @@ -19,6 +19,8 @@ func TestUserAccessTokenStore(t *testing.T, rctx request.CTX, ss store.Store) { t.Run("UserAccessTokenSearch", func(t *testing.T) { testUserAccessTokenSearch(t, rctx, ss) }) t.Run("UserAccessTokenPagination", func(t *testing.T) { testUserAccessTokenPagination(t, rctx, ss) }) t.Run("UserAccessTokenExpiry", func(t *testing.T) { testUserAccessTokenExpiry(t, rctx, ss) }) + t.Run("UserAccessTokenGetExpiring", func(t *testing.T) { testUserAccessTokenGetExpiring(t, rctx, ss) }) + t.Run("UserAccessTokenUpdateLastNotifiedAt", func(t *testing.T) { testUserAccessTokenUpdateLastNotifiedAt(t, rctx, ss) }) t.Run("UserAccessTokenNonCompliant", func(t *testing.T) { testUserAccessTokenNonCompliant(t, rctx, ss) }) } @@ -358,6 +360,167 @@ func testUserAccessTokenExpiry(t *testing.T, rctx request.CTX, ss store.Store) { require.Equal(t, int64(0), deleted) } +func testUserAccessTokenGetExpiring(t *testing.T, rctx request.CTX, ss store.Store) { + now := model.GetMillis() + thresholds := []int{1, 3, 7} + + activeUser, err := ss.User().Save(rctx, &model.User{Email: MakeEmail(), Username: model.NewUsername()}) + require.NoError(t, err) + + // Token expiring inside the 7-day window, never notified. + inWindow := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "in-window", ExpiresAt: now + 5*dayMillisTest} + _, err = ss.UserAccessToken().Save(inWindow) + require.NoError(t, err) + + // Token expiring beyond the horizon. + beyond := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "beyond", ExpiresAt: now + 30*dayMillisTest} + _, err = ss.UserAccessToken().Save(beyond) + require.NoError(t, err) + + // Already-expired token. + past := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "past", ExpiresAt: now - 60*1000} + _, err = ss.UserAccessToken().Save(past) + require.NoError(t, err) + + // Non-expiring token. + nonExpiring := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "non-expiring"} + _, err = ss.UserAccessToken().Save(nonExpiring) + require.NoError(t, err) + + // Token with 12h left that was already warned within the last day (i.e. at the + // terminal 1-day bucket): excluded. + terminal := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "terminal", ExpiresAt: now + 12*60*60*1000} + _, err = ss.UserAccessToken().Save(terminal) + require.NoError(t, err) + require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(terminal.Id, now)) + + // Token warned at the 7-day mark (when 7 days remained) but now inside the + // 3-day window: still returned, because the 3-day bucket hasn't been covered. + stillNotify := &model.UserAccessToken{Token: model.NewId(), UserId: activeUser.Id, Description: "still-notify", ExpiresAt: now + 2*dayMillisTest} + _, err = ss.UserAccessToken().Save(stillNotify) + require.NoError(t, err) + stillNotifyAt := stillNotify.ExpiresAt - 7*dayMillisTest + require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(stillNotify.Id, stillNotifyAt)) + + // Deactivated user with an in-window token: excluded. + deletedUser := &model.User{Email: MakeEmail(), Username: model.NewUsername()} + _, err = ss.User().Save(rctx, deletedUser) + require.NoError(t, err) + deletedUser.DeleteAt = model.GetMillis() + _, err = ss.User().Update(rctx, deletedUser, true) + require.NoError(t, err) + deactivatedToken := &model.UserAccessToken{Token: model.NewId(), UserId: deletedUser.Id, Description: "deactivated", ExpiresAt: now + 5*dayMillisTest} + _, err = ss.UserAccessToken().Save(deactivatedToken) + require.NoError(t, err) + + // Bot with an in-window token: excluded. + botUser, err := ss.User().Save(rctx, model.UserFromBot(&model.Bot{Username: model.NewUsername(), OwnerId: activeUser.Id})) + require.NoError(t, err) + _, nErr := ss.Bot().Save(&model.Bot{UserId: botUser.Id, Username: botUser.Username, OwnerId: activeUser.Id}) + require.NoError(t, nErr) + botToken := &model.UserAccessToken{Token: model.NewId(), UserId: botUser.Id, Description: "bot", ExpiresAt: now + 5*dayMillisTest} + _, err = ss.UserAccessToken().Save(botToken) + require.NoError(t, err) + + t.Cleanup(func() { + for _, id := range []string{inWindow.Id, beyond.Id, past.Id, nonExpiring.Id, terminal.Id, stillNotify.Id, deactivatedToken.Id, botToken.Id} { + _ = ss.UserAccessToken().Delete(id) + } + _ = ss.Bot().PermanentDelete(botUser.Id) + _ = ss.User().PermanentDelete(rctx, botUser.Id) + _ = ss.User().PermanentDelete(rctx, activeUser.Id) + _ = ss.User().PermanentDelete(rctx, deletedUser.Id) + }) + + rows, err := ss.UserAccessToken().GetExpiringTokens(now, thresholds, 100) + require.NoError(t, err) + + got := make(map[string]*model.UserAccessToken) + for _, row := range rows { + got[row.Id] = row + require.Empty(t, row.Token, "GetExpiringTokens must never return the secret Token value") + } + + require.Contains(t, got, inWindow.Id, "in-window token should be returned") + require.Contains(t, got, stillNotify.Id, "token warned at 7 days but now inside the 3-day window should still be returned") + require.NotContains(t, got, beyond.Id, "token beyond the horizon must not be returned") + require.NotContains(t, got, past.Id, "already-expired token must not be returned") + require.NotContains(t, got, nonExpiring.Id, "non-expiring token must not be returned") + require.NotContains(t, got, terminal.Id, "token already warned within the terminal bucket must not be returned") + require.NotContains(t, got, deactivatedToken.Id, "token owned by a deactivated user must not be returned") + require.NotContains(t, got, botToken.Id, "bot token must not be returned") + + require.Nil(t, got[inWindow.Id].LastNotifiedAt) + require.NotNil(t, got[stillNotify.Id].LastNotifiedAt) + require.Equal(t, stillNotifyAt, *got[stillNotify.Id].LastNotifiedAt) + + // Results must be ordered by ExpiresAt ascending (most urgent first); the + // worker relies on this to drain a batch in expiry order. stillNotify (2d) + // must come before inWindow (5d), and the slice must be non-decreasing. + var prev int64 + for i, row := range rows { + if i > 0 { + require.GreaterOrEqual(t, row.ExpiresAt, prev, "GetExpiringTokens must return rows in ExpiresAt ascending order") + } + prev = row.ExpiresAt + } + require.Less(t, indexOfToken(rows, stillNotify.Id), indexOfToken(rows, inWindow.Id), "sooner-expiring token must be returned first") + + // Non-positive limit and empty thresholds short-circuit. + zero, err := ss.UserAccessToken().GetExpiringTokens(now, thresholds, 0) + require.NoError(t, err) + require.Empty(t, zero) + neg, err := ss.UserAccessToken().GetExpiringTokens(now, thresholds, -3) + require.NoError(t, err) + require.Empty(t, neg) + none, err := ss.UserAccessToken().GetExpiringTokens(now, nil, 100) + require.NoError(t, err) + require.Empty(t, none) +} + +func indexOfToken(tokens []*model.UserAccessToken, id string) int { + for i, token := range tokens { + if token.Id == id { + return i + } + } + return -1 +} + +func testUserAccessTokenUpdateLastNotifiedAt(t *testing.T, rctx request.CTX, ss store.Store) { + token := &model.UserAccessToken{ + Token: model.NewId(), + UserId: model.NewId(), + Description: "notified-at", + ExpiresAt: model.GetMillis() + dayMillisTest, + } + _, err := ss.UserAccessToken().Save(token) + require.NoError(t, err) + t.Cleanup(func() { _ = ss.UserAccessToken().Delete(token.Id) }) + + // A freshly saved token has no last-notified time set (NULL -> nil). + stored, err := ss.UserAccessToken().Get(token.Id) + require.NoError(t, err) + require.Nil(t, stored.LastNotifiedAt) + + first := model.GetMillis() + require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(token.Id, first)) + stored, err = ss.UserAccessToken().Get(token.Id) + require.NoError(t, err) + require.NotNil(t, stored.LastNotifiedAt) + require.Equal(t, first, *stored.LastNotifiedAt) + + // A later warning advances the timestamp. + second := first + dayMillisTest + require.NoError(t, ss.UserAccessToken().UpdateLastNotifiedAt(token.Id, second)) + stored, err = ss.UserAccessToken().Get(token.Id) + require.NoError(t, err) + require.NotNil(t, stored.LastNotifiedAt) + require.Equal(t, second, *stored.LastNotifiedAt) +} + +const dayMillisTest = int64(24 * 60 * 60 * 1000) + func testUserAccessTokenNonCompliant(t *testing.T, rctx request.CTX, ss store.Store) { now := model.GetMillis() day := int64(24 * 60 * 60 * 1000) diff --git a/server/channels/store/timerlayer/timerlayer.go b/server/channels/store/timerlayer/timerlayer.go index 58c916b5c36..5c6f0f9d2b2 100644 --- a/server/channels/store/timerlayer/timerlayer.go +++ b/server/channels/store/timerlayer/timerlayer.go @@ -14417,6 +14417,22 @@ func (s *TimerLayerUserAccessTokenStore) GetExpiredBefore(cutoff int64, limit in return result, err } +func (s *TimerLayerUserAccessTokenStore) GetExpiringTokens(now int64, thresholds []int, limit int) ([]*model.UserAccessToken, error) { + start := time.Now() + + result, err := s.UserAccessTokenStore.GetExpiringTokens(now, thresholds, limit) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("UserAccessTokenStore.GetExpiringTokens", success, elapsed) + } + return result, err +} + func (s *TimerLayerUserAccessTokenStore) Save(token *model.UserAccessToken) (*model.UserAccessToken, error) { start := time.Now() @@ -14449,6 +14465,22 @@ func (s *TimerLayerUserAccessTokenStore) Search(term string) ([]*model.UserAcces return result, err } +func (s *TimerLayerUserAccessTokenStore) UpdateLastNotifiedAt(tokenID string, notifiedAt int64) error { + start := time.Now() + + err := s.UserAccessTokenStore.UpdateLastNotifiedAt(tokenID, notifiedAt) + + elapsed := float64(time.Since(start)) / float64(time.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("UserAccessTokenStore.UpdateLastNotifiedAt", success, elapsed) + } + return err +} + func (s *TimerLayerUserAccessTokenStore) UpdateTokenDisable(tokenID string) error { start := time.Now() diff --git a/server/i18n/en.json b/server/i18n/en.json index f5393787345..fbd98166373 100644 --- a/server/i18n/en.json +++ b/server/i18n/en.json @@ -7912,6 +7912,18 @@ "id": "app.pap.update_access_control_policies_active.app_error", "translation": "Could not update active status of access control policies." }, + { + "id": "app.pat_expiry_notify.dm", + "translation": "Your personal access token \"{{.Description}}\" will expire in {{.Days}} days or less. Open **Profile > Security** to regenerate it before it stops working." + }, + { + "id": "app.pat_expiry_notify.dm_final", + "translation": "Your personal access token \"{{.Description}}\" will expire within 24 hours. Open **Profile > Security** to regenerate it before it stops working." + }, + { + "id": "app.pat_expiry_notify.unnamed_token", + "translation": "(no description)" + }, { "id": "app.pdp.access_evaluation.app_error", "translation": "Failed evaluate access control policy." @@ -9698,6 +9710,10 @@ "id": "app.user_access_token.get_by_user.app_error", "translation": "Unable to get the personal access tokens by user." }, + { + "id": "app.user_access_token.get_expiring.app_error", + "translation": "Unable to retrieve expiring personal access tokens." + }, { "id": "app.user_access_token.invalid_or_missing", "translation": "Invalid or missing token." diff --git a/server/public/model/job.go b/server/public/model/job.go index 77bfd7bfb1a..dc847c580a0 100644 --- a/server/public/model/job.go +++ b/server/public/model/job.go @@ -49,6 +49,7 @@ const ( JobTypeDeleteExpiredPosts = "delete_expired_posts" JobTypeAutoTranslationRecovery = "autotranslation_recovery" JobTypeCleanupExpiredAccessTokens = "cleanup_expired_access_tokens" + JobTypePatExpiryNotify = "pat_expiry_notify" JobStatusPending = "pending" JobStatusInProgress = "in_progress" @@ -80,6 +81,7 @@ var AllJobTypes = [...]string{ JobTypeLastAccessibleFile, JobTypeCleanupDesktopTokens, JobTypeCleanupExpiredAccessTokens, + JobTypePatExpiryNotify, JobTypeRefreshMaterializedViews, JobTypeMobileSessionMetadata, } diff --git a/server/public/model/user_access_token.go b/server/public/model/user_access_token.go index 964470ecd51..0765c06db5c 100644 --- a/server/public/model/user_access_token.go +++ b/server/public/model/user_access_token.go @@ -25,6 +25,15 @@ type UserAccessToken struct { // ExpiresAt is non-zero and in the past are considered expired and // MUST be rejected at validation time. ExpiresAt int64 `json:"expires_at"` + // LastNotifiedAt is the Unix timestamp in milliseconds at which the token + // owner was last warned that the token is approaching expiry. It is nil until + // the first warning is sent. The pat_expiry_notify job uses it to dedup: the + // warning bucket already covered is recovered from (ExpiresAt - LastNotifiedAt), + // so the same (or a less urgent) warning is not re-sent every run. Storing the + // moment rather than the bucket keeps the column independent of the day cascade + // and correct if the buckets are later changed. It is internal server + // bookkeeping that is never exposed over the API. + LastNotifiedAt *int64 `json:"-"` } func (t *UserAccessToken) IsValid() *AppError {