// Copyright (C) 2019-2023 Nicola Murino // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published // by the Free Software Foundation, version 3. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . package common import ( "bytes" "context" "encoding/csv" "encoding/json" "errors" "fmt" "io" "mime" "mime/multipart" "net/http" "net/textproto" "net/url" "os" "os/exec" "path" "path/filepath" "strconv" "strings" "sync" "sync/atomic" "time" "github.com/bmatcuk/doublestar/v4" "github.com/klauspost/compress/zip" "github.com/robfig/cron/v3" "github.com/rs/xid" "github.com/sftpgo/sdk" "github.com/wneessen/go-mail" "github.com/drakkan/sftpgo/v2/internal/dataprovider" "github.com/drakkan/sftpgo/v2/internal/logger" "github.com/drakkan/sftpgo/v2/internal/plugin" "github.com/drakkan/sftpgo/v2/internal/smtp" "github.com/drakkan/sftpgo/v2/internal/util" "github.com/drakkan/sftpgo/v2/internal/vfs" ) const ( ipBlockedEventName = "IP Blocked" maxAttachmentsSize = int64(10 * 1024 * 1024) objDataPlaceholder = "{{ObjectData}}" ) // Supported IDP login events const ( IDPLoginUser = "IDP login user" IDPLoginAdmin = "IDP login admin" ) var ( // eventManager handle the supported event rules actions eventManager eventRulesContainer multipartQuoteEscaper = strings.NewReplacer("\\", "\\\\", `"`, "\\\"") ) func init() { eventManager = eventRulesContainer{ schedulesMapping: make(map[string][]cron.EntryID), // arbitrary maximum number of concurrent asynchronous tasks, // each task could execute multiple actions concurrencyGuard: make(chan struct{}, 200), } dataprovider.SetEventRulesCallbacks(eventManager.loadRules, eventManager.RemoveRule, func(operation, executor, ip, objectType, objectName, role string, object plugin.Renderer) { p := EventParams{ Name: executor, ObjectName: objectName, Event: operation, Status: 1, ObjectType: objectType, IP: ip, Role: role, Timestamp: time.Now().UnixNano(), Object: object, } if u, ok := object.(*dataprovider.User); ok { p.Email = u.Email } else if a, ok := object.(*dataprovider.Admin); ok { p.Email = a.Email } eventManager.handleProviderEvent(p) }) } // HandleCertificateEvent checks and executes action rules for certificate events func HandleCertificateEvent(params EventParams) { eventManager.handleCertificateEvent(params) } // HandleIDPLoginEvent executes actions defined for a successful login from an Identity Provider func HandleIDPLoginEvent(params EventParams, customFields *map[string]any) (*dataprovider.User, *dataprovider.Admin, error) { return eventManager.handleIDPLoginEvent(params, customFields) } // eventRulesContainer stores event rules by trigger type eventRulesContainer struct { sync.RWMutex lastLoad atomic.Int64 FsEvents []dataprovider.EventRule ProviderEvents []dataprovider.EventRule Schedules []dataprovider.EventRule IPBlockedEvents []dataprovider.EventRule CertificateEvents []dataprovider.EventRule IPDLoginEvents []dataprovider.EventRule schedulesMapping map[string][]cron.EntryID concurrencyGuard chan struct{} } func (r *eventRulesContainer) addAsyncTask() { activeHooks.Add(1) r.concurrencyGuard <- struct{}{} } func (r *eventRulesContainer) removeAsyncTask() { activeHooks.Add(-1) <-r.concurrencyGuard } func (r *eventRulesContainer) getLastLoadTime() int64 { return r.lastLoad.Load() } func (r *eventRulesContainer) setLastLoadTime(modTime int64) { r.lastLoad.Store(modTime) } // RemoveRule deletes the rule with the specified name func (r *eventRulesContainer) RemoveRule(name string) { r.Lock() defer r.Unlock() r.removeRuleInternal(name) eventManagerLog(logger.LevelDebug, "event rules updated after delete, fs events: %d, provider events: %d, schedules: %d", len(r.FsEvents), len(r.ProviderEvents), len(r.Schedules)) } func (r *eventRulesContainer) removeRuleInternal(name string) { for idx := range r.FsEvents { if r.FsEvents[idx].Name == name { lastIdx := len(r.FsEvents) - 1 r.FsEvents[idx] = r.FsEvents[lastIdx] r.FsEvents = r.FsEvents[:lastIdx] eventManagerLog(logger.LevelDebug, "removed rule %q from fs events", name) return } } for idx := range r.ProviderEvents { if r.ProviderEvents[idx].Name == name { lastIdx := len(r.ProviderEvents) - 1 r.ProviderEvents[idx] = r.ProviderEvents[lastIdx] r.ProviderEvents = r.ProviderEvents[:lastIdx] eventManagerLog(logger.LevelDebug, "removed rule %q from provider events", name) return } } for idx := range r.IPBlockedEvents { if r.IPBlockedEvents[idx].Name == name { lastIdx := len(r.IPBlockedEvents) - 1 r.IPBlockedEvents[idx] = r.IPBlockedEvents[lastIdx] r.IPBlockedEvents = r.IPBlockedEvents[:lastIdx] eventManagerLog(logger.LevelDebug, "removed rule %q from IP blocked events", name) return } } for idx := range r.CertificateEvents { if r.CertificateEvents[idx].Name == name { lastIdx := len(r.CertificateEvents) - 1 r.CertificateEvents[idx] = r.CertificateEvents[lastIdx] r.CertificateEvents = r.CertificateEvents[:lastIdx] eventManagerLog(logger.LevelDebug, "removed rule %q from certificate events", name) return } } for idx := range r.IPDLoginEvents { if r.IPDLoginEvents[idx].Name == name { lastIdx := len(r.IPDLoginEvents) - 1 r.IPDLoginEvents[idx] = r.IPDLoginEvents[lastIdx] r.IPDLoginEvents = r.IPDLoginEvents[:lastIdx] eventManagerLog(logger.LevelDebug, "removed rule %q from IDP login events", name) return } } for idx := range r.Schedules { if r.Schedules[idx].Name == name { if schedules, ok := r.schedulesMapping[name]; ok { for _, entryID := range schedules { eventManagerLog(logger.LevelDebug, "removing scheduled entry id %d for rule %q", entryID, name) eventScheduler.Remove(entryID) } delete(r.schedulesMapping, name) } lastIdx := len(r.Schedules) - 1 r.Schedules[idx] = r.Schedules[lastIdx] r.Schedules = r.Schedules[:lastIdx] eventManagerLog(logger.LevelDebug, "removed rule %q from scheduled events", name) return } } } func (r *eventRulesContainer) addUpdateRuleInternal(rule dataprovider.EventRule) { r.removeRuleInternal(rule.Name) if rule.DeletedAt > 0 { deletedAt := util.GetTimeFromMsecSinceEpoch(rule.DeletedAt) if deletedAt.Add(30 * time.Minute).Before(time.Now()) { eventManagerLog(logger.LevelDebug, "removing rule %q deleted at %s", rule.Name, deletedAt) go dataprovider.RemoveEventRule(rule) //nolint:errcheck } return } if rule.Status != 1 || rule.Trigger == dataprovider.EventTriggerOnDemand { return } switch rule.Trigger { case dataprovider.EventTriggerFsEvent: r.FsEvents = append(r.FsEvents, rule) eventManagerLog(logger.LevelDebug, "added rule %q to fs events", rule.Name) case dataprovider.EventTriggerProviderEvent: r.ProviderEvents = append(r.ProviderEvents, rule) eventManagerLog(logger.LevelDebug, "added rule %q to provider events", rule.Name) case dataprovider.EventTriggerIPBlocked: r.IPBlockedEvents = append(r.IPBlockedEvents, rule) eventManagerLog(logger.LevelDebug, "added rule %q to IP blocked events", rule.Name) case dataprovider.EventTriggerCertificate: r.CertificateEvents = append(r.CertificateEvents, rule) eventManagerLog(logger.LevelDebug, "added rule %q to certificate events", rule.Name) case dataprovider.EventTriggerIDPLogin: r.IPDLoginEvents = append(r.IPDLoginEvents, rule) eventManagerLog(logger.LevelDebug, "added rule %q to IDP login events", rule.Name) case dataprovider.EventTriggerSchedule: for _, schedule := range rule.Conditions.Schedules { cronSpec := schedule.GetCronSpec() job := &eventCronJob{ ruleName: dataprovider.ConvertName(rule.Name), } entryID, err := eventScheduler.AddJob(cronSpec, job) if err != nil { eventManagerLog(logger.LevelError, "unable to add scheduled rule %q, cron string %q: %v", rule.Name, cronSpec, err) return } r.schedulesMapping[rule.Name] = append(r.schedulesMapping[rule.Name], entryID) eventManagerLog(logger.LevelDebug, "schedule for rule %q added, id: %d, cron string %q, active scheduling rules: %d", rule.Name, entryID, cronSpec, len(r.schedulesMapping)) } r.Schedules = append(r.Schedules, rule) eventManagerLog(logger.LevelDebug, "added rule %q to scheduled events", rule.Name) default: eventManagerLog(logger.LevelError, "unsupported trigger: %d", rule.Trigger) } } func (r *eventRulesContainer) loadRules() { eventManagerLog(logger.LevelDebug, "loading updated rules") modTime := util.GetTimeAsMsSinceEpoch(time.Now()) rules, err := dataprovider.GetRecentlyUpdatedRules(r.getLastLoadTime()) if err != nil { eventManagerLog(logger.LevelError, "unable to load event rules: %v", err) return } eventManagerLog(logger.LevelDebug, "recently updated event rules loaded: %d", len(rules)) if len(rules) > 0 { r.Lock() defer r.Unlock() for _, rule := range rules { r.addUpdateRuleInternal(rule) } } eventManagerLog(logger.LevelDebug, "event rules updated, fs events: %d, provider events: %d, schedules: %d, ip blocked events: %d, certificate events: %d, IDP login events: %d", len(r.FsEvents), len(r.ProviderEvents), len(r.Schedules), len(r.IPBlockedEvents), len(r.CertificateEvents), len(r.IPDLoginEvents)) r.setLastLoadTime(modTime) } func (*eventRulesContainer) checkIPDLoginEventMatch(conditions *dataprovider.EventConditions, params *EventParams) bool { switch conditions.IDPLoginEvent { case dataprovider.IDPLoginUser: if params.Event != IDPLoginUser { return false } case dataprovider.IDPLoginAdmin: if params.Event != IDPLoginAdmin { return false } } return checkEventConditionPatterns(params.Name, conditions.Options.Names) } func (*eventRulesContainer) checkProviderEventMatch(conditions *dataprovider.EventConditions, params *EventParams) bool { if !util.Contains(conditions.ProviderEvents, params.Event) { return false } if !checkEventConditionPatterns(params.Name, conditions.Options.Names) { return false } if !checkEventConditionPatterns(params.Role, conditions.Options.RoleNames) { return false } if len(conditions.Options.ProviderObjects) > 0 && !util.Contains(conditions.Options.ProviderObjects, params.ObjectType) { return false } return true } func (*eventRulesContainer) checkFsEventMatch(conditions *dataprovider.EventConditions, params *EventParams) bool { if !util.Contains(conditions.FsEvents, params.Event) { return false } if !checkEventConditionPatterns(params.Name, conditions.Options.Names) { return false } if !checkEventConditionPatterns(params.Role, conditions.Options.RoleNames) { return false } if !checkEventGroupConditionPatterns(params.Groups, conditions.Options.GroupNames) { return false } if !checkEventConditionPatterns(params.VirtualPath, conditions.Options.FsPaths) { return false } if len(conditions.Options.Protocols) > 0 && !util.Contains(conditions.Options.Protocols, params.Protocol) { return false } if params.Event == operationUpload || params.Event == operationDownload { if conditions.Options.MinFileSize > 0 { if params.FileSize < conditions.Options.MinFileSize { return false } } if conditions.Options.MaxFileSize > 0 { if params.FileSize > conditions.Options.MaxFileSize { return false } } } return true } // hasFsRules returns true if there are any rules for filesystem event triggers func (r *eventRulesContainer) hasFsRules() bool { r.RLock() defer r.RUnlock() return len(r.FsEvents) > 0 } // handleFsEvent executes the rules actions defined for the specified event. // The boolean parameter indicates whether a sync action was executed func (r *eventRulesContainer) handleFsEvent(params EventParams) (bool, error) { if params.Protocol == protocolEventAction { return false, nil } r.RLock() var rulesWithSyncActions, rulesAsync []dataprovider.EventRule for _, rule := range r.FsEvents { if r.checkFsEventMatch(&rule.Conditions, ¶ms) { if err := rule.CheckActionsConsistency(""); err != nil { eventManagerLog(logger.LevelWarn, "rule %q skipped: %v, event %q", rule.Name, err, params.Event) continue } hasSyncActions := false for _, action := range rule.Actions { if action.Options.ExecuteSync { hasSyncActions = true break } } if hasSyncActions { rulesWithSyncActions = append(rulesWithSyncActions, rule) } else { rulesAsync = append(rulesAsync, rule) } } } r.RUnlock() params.sender = params.Name if len(rulesAsync) > 0 { go executeAsyncRulesActions(rulesAsync, params) } if len(rulesWithSyncActions) > 0 { return true, executeSyncRulesActions(rulesWithSyncActions, params) } return false, nil } func (r *eventRulesContainer) handleIDPLoginEvent(params EventParams, customFields *map[string]any) (*dataprovider.User, *dataprovider.Admin, error, ) { r.RLock() var rulesWithSyncActions, rulesAsync []dataprovider.EventRule for _, rule := range r.IPDLoginEvents { if r.checkIPDLoginEventMatch(&rule.Conditions, ¶ms) { if err := rule.CheckActionsConsistency(""); err != nil { eventManagerLog(logger.LevelWarn, "rule %q skipped: %v, event %q", rule.Name, err, params.Event) continue } hasSyncActions := false for _, action := range rule.Actions { if action.Options.ExecuteSync { hasSyncActions = true break } } if hasSyncActions { rulesWithSyncActions = append(rulesWithSyncActions, rule) } else { rulesAsync = append(rulesAsync, rule) } } } r.RUnlock() if len(rulesAsync) == 0 && len(rulesWithSyncActions) == 0 { return nil, nil, nil } params.addIDPCustomFields(customFields) if len(rulesWithSyncActions) > 1 { var ruleNames []string for _, r := range rulesWithSyncActions { ruleNames = append(ruleNames, r.Name) } return nil, nil, fmt.Errorf("more than one account check action rules matches: %q", strings.Join(ruleNames, ",")) } if len(rulesAsync) > 0 { go executeAsyncRulesActions(rulesAsync, params) } if len(rulesWithSyncActions) > 0 { return executeIDPAccountCheckRule(rulesWithSyncActions[0], params) } return nil, nil, nil } // username is populated for user objects func (r *eventRulesContainer) handleProviderEvent(params EventParams) { r.RLock() defer r.RUnlock() var rules []dataprovider.EventRule for _, rule := range r.ProviderEvents { if r.checkProviderEventMatch(&rule.Conditions, ¶ms) { if err := rule.CheckActionsConsistency(params.ObjectType); err == nil { rules = append(rules, rule) } else { eventManagerLog(logger.LevelWarn, "rule %q skipped: %v, event %q object type %q", rule.Name, err, params.Event, params.ObjectType) } } } if len(rules) > 0 { params.sender = params.ObjectName go executeAsyncRulesActions(rules, params) } } func (r *eventRulesContainer) handleIPBlockedEvent(params EventParams) { r.RLock() defer r.RUnlock() if len(r.IPBlockedEvents) == 0 { return } var rules []dataprovider.EventRule for _, rule := range r.IPBlockedEvents { if err := rule.CheckActionsConsistency(""); err == nil { rules = append(rules, rule) } else { eventManagerLog(logger.LevelWarn, "rule %q skipped: %v, event %q", rule.Name, err, params.Event) } } if len(rules) > 0 { go executeAsyncRulesActions(rules, params) } } func (r *eventRulesContainer) handleCertificateEvent(params EventParams) { r.RLock() defer r.RUnlock() if len(r.CertificateEvents) == 0 { return } var rules []dataprovider.EventRule for _, rule := range r.CertificateEvents { if err := rule.CheckActionsConsistency(""); err == nil { rules = append(rules, rule) } else { eventManagerLog(logger.LevelWarn, "rule %q skipped: %v, event %q", rule.Name, err, params.Event) } } if len(rules) > 0 { go executeAsyncRulesActions(rules, params) } } type executedRetentionCheck struct { Username string ActionName string Results []folderRetentionCheckResult } // EventParams defines the supported event parameters type EventParams struct { Name string Groups []sdk.GroupMapping Event string Status int VirtualPath string FsPath string VirtualTargetPath string FsTargetPath string ObjectName string ObjectType string FileSize int64 Elapsed int64 Protocol string IP string Role string Email string Timestamp int64 IDPCustomFields *map[string]string Object plugin.Renderer sender string updateStatusFromError bool errors []string retentionChecks []executedRetentionCheck } func (p *EventParams) getACopy() *EventParams { params := *p params.errors = make([]string, len(p.errors)) copy(params.errors, p.errors) retentionChecks := make([]executedRetentionCheck, 0, len(p.retentionChecks)) for _, c := range p.retentionChecks { executedCheck := executedRetentionCheck{ Username: c.Username, ActionName: c.ActionName, } executedCheck.Results = make([]folderRetentionCheckResult, len(c.Results)) copy(executedCheck.Results, c.Results) retentionChecks = append(retentionChecks, executedCheck) } params.retentionChecks = retentionChecks if p.IDPCustomFields != nil { fields := make(map[string]string) for k, v := range *p.IDPCustomFields { fields[k] = v } params.IDPCustomFields = &fields } return ¶ms } func (p *EventParams) addIDPCustomFields(customFields *map[string]any) { if customFields == nil { return } fields := make(map[string]string) for k, v := range *customFields { switch val := v.(type) { case string: fields[k] = val } } p.IDPCustomFields = &fields } // AddError adds a new error to the event params and update the status if needed func (p *EventParams) AddError(err error) { if err == nil { return } if p.updateStatusFromError && p.Status == 1 { p.Status = 2 } p.errors = append(p.errors, err.Error()) } func (p *EventParams) setBackupParams(backupPath string) { if p.sender != "" { return } p.sender = dataprovider.ActionExecutorSystem p.FsPath = backupPath p.ObjectName = filepath.Base(backupPath) p.VirtualPath = "/" + p.ObjectName p.Timestamp = time.Now().UnixNano() info, err := os.Stat(backupPath) if err == nil { p.FileSize = info.Size() } } func (p *EventParams) getStatusString() string { switch p.Status { case 1: return "OK" default: return "KO" } } // getUsers returns users with group settings not applied func (p *EventParams) getUsers() ([]dataprovider.User, error) { if p.sender == "" { dump, err := dataprovider.DumpData([]string{dataprovider.DumpScopeUsers}) if err != nil { eventManagerLog(logger.LevelError, "unable to get users: %+v", err) return nil, errors.New("unable to get users") } return dump.Users, nil } user, err := p.getUserFromSender() if err != nil { return nil, err } return []dataprovider.User{user}, nil } func (p *EventParams) getUserFromSender() (dataprovider.User, error) { if p.sender == dataprovider.ActionExecutorSystem { return dataprovider.User{ BaseUser: sdk.BaseUser{ Status: 1, Username: p.sender, HomeDir: dataprovider.GetBackupsPath(), Permissions: map[string][]string{ "/": {dataprovider.PermAny}, }, }, }, nil } user, err := dataprovider.UserExists(p.sender, "") if err != nil { eventManagerLog(logger.LevelError, "unable to get user %q: %+v", p.sender, err) return user, fmt.Errorf("error getting user %q", p.sender) } return user, nil } func (p *EventParams) getFolders() ([]vfs.BaseVirtualFolder, error) { if p.sender == "" { dump, err := dataprovider.DumpData([]string{dataprovider.DumpScopeFolders}) return dump.Folders, err } folder, err := dataprovider.GetFolderByName(p.sender) if err != nil { return nil, fmt.Errorf("error getting folder %q: %w", p.sender, err) } return []vfs.BaseVirtualFolder{folder}, nil } func (p *EventParams) getCompressedDataRetentionReport() ([]byte, error) { if len(p.retentionChecks) == 0 { return nil, errors.New("no data retention report available") } var b bytes.Buffer if _, err := p.writeCompressedDataRetentionReports(&b); err != nil { return nil, err } return b.Bytes(), nil } func (p *EventParams) writeCompressedDataRetentionReports(w io.Writer) (int64, error) { var n int64 wr := zip.NewWriter(w) for _, check := range p.retentionChecks { data, err := getCSVRetentionReport(check.Results) if err != nil { return n, fmt.Errorf("unable to get CSV report: %w", err) } dataSize := int64(len(data)) n += dataSize // we suppose a 3:1 compression ratio if n > (maxAttachmentsSize * 3) { eventManagerLog(logger.LevelError, "unable to get retention report, size too large: %s", util.ByteCountIEC(n)) return n, fmt.Errorf("unable to get retention report, size too large: %s", util.ByteCountIEC(n)) } fh := &zip.FileHeader{ Name: fmt.Sprintf("%s-%s.csv", check.ActionName, check.Username), Method: zip.Deflate, Modified: time.Now().UTC(), } f, err := wr.CreateHeader(fh) if err != nil { return n, fmt.Errorf("unable to create zip header for file %q: %w", fh.Name, err) } _, err = io.CopyN(f, bytes.NewBuffer(data), dataSize) if err != nil { return n, fmt.Errorf("unable to write content to zip file %q: %w", fh.Name, err) } } if err := wr.Close(); err != nil { return n, fmt.Errorf("unable to close zip writer: %w", err) } return n, nil } func (p *EventParams) getRetentionReportsAsMailAttachment() (*mail.File, error) { if len(p.retentionChecks) == 0 { return nil, errors.New("no data retention report available") } return &mail.File{ Name: "retention-reports.zip", Header: make(map[string][]string), Writer: p.writeCompressedDataRetentionReports, }, nil } func (*EventParams) getStringReplacement(val string, jsonEscaped bool) string { if jsonEscaped { return util.JSONEscape(val) } return val } func (p *EventParams) getStringReplacements(addObjectData, jsonEscaped bool) []string { replacements := []string{ "{{Name}}", p.getStringReplacement(p.Name, jsonEscaped), "{{Event}}", p.Event, "{{Status}}", fmt.Sprintf("%d", p.Status), "{{VirtualPath}}", p.getStringReplacement(p.VirtualPath, jsonEscaped), "{{FsPath}}", p.getStringReplacement(p.FsPath, jsonEscaped), "{{VirtualTargetPath}}", p.getStringReplacement(p.VirtualTargetPath, jsonEscaped), "{{FsTargetPath}}", p.getStringReplacement(p.FsTargetPath, jsonEscaped), "{{ObjectName}}", p.getStringReplacement(p.ObjectName, jsonEscaped), "{{ObjectType}}", p.ObjectType, "{{FileSize}}", fmt.Sprintf("%d", p.FileSize), "{{Elapsed}}", fmt.Sprintf("%d", p.Elapsed), "{{Protocol}}", p.Protocol, "{{IP}}", p.IP, "{{Role}}", p.getStringReplacement(p.Role, jsonEscaped), "{{Email}}", p.getStringReplacement(p.Email, jsonEscaped), "{{Timestamp}}", fmt.Sprintf("%d", p.Timestamp), "{{StatusString}}", p.getStatusString(), } if p.VirtualPath != "" { replacements = append(replacements, "{{VirtualDirPath}}", p.getStringReplacement(path.Dir(p.VirtualPath), jsonEscaped)) } if p.VirtualTargetPath != "" { replacements = append(replacements, "{{VirtualTargetDirPath}}", p.getStringReplacement(path.Dir(p.VirtualTargetPath), jsonEscaped)) replacements = append(replacements, "{{TargetName}}", p.getStringReplacement(path.Base(p.VirtualTargetPath), jsonEscaped)) } if len(p.errors) > 0 { replacements = append(replacements, "{{ErrorString}}", p.getStringReplacement(strings.Join(p.errors, ", "), jsonEscaped)) } else { replacements = append(replacements, "{{ErrorString}}", "") } replacements = append(replacements, objDataPlaceholder, "") if addObjectData { data, err := p.Object.RenderAsJSON(p.Event != operationDelete) if err == nil { replacements[len(replacements)-1] = p.getStringReplacement(string(data), jsonEscaped) } } if p.IDPCustomFields != nil { for k, v := range *p.IDPCustomFields { replacements = append(replacements, fmt.Sprintf("{{IDPField%s}}", k), p.getStringReplacement(v, jsonEscaped)) } } return replacements } func getCSVRetentionReport(results []folderRetentionCheckResult) ([]byte, error) { var b bytes.Buffer csvWriter := csv.NewWriter(&b) err := csvWriter.Write([]string{"path", "retention (hours)", "deleted files", "deleted size (bytes)", "elapsed (ms)", "info", "error"}) if err != nil { return nil, err } for _, result := range results { err = csvWriter.Write([]string{result.Path, strconv.Itoa(result.Retention), strconv.Itoa(result.DeletedFiles), strconv.FormatInt(result.DeletedSize, 10), strconv.FormatInt(result.Elapsed.Milliseconds(), 10), result.Info, result.Error}) if err != nil { return nil, err } } csvWriter.Flush() err = csvWriter.Error() return b.Bytes(), err } func closeWriterAndUpdateQuota(w io.WriteCloser, conn *BaseConnection, virtualSourcePath, virtualTargetPath string, numFiles int, truncatedSize int64, errTransfer error, operation string, startTime time.Time, ) error { var fsDstPath string var errDstFs error errWrite := w.Close() targetPath := virtualSourcePath if virtualTargetPath != "" { targetPath = virtualTargetPath var fsDst vfs.Fs fsDst, fsDstPath, errDstFs = conn.GetFsAndResolvedPath(virtualTargetPath) if errTransfer != nil && errDstFs == nil { // try to remove a partial file on error. If this fails, we can't do anything errRemove := fsDst.Remove(fsDstPath, false) conn.Log(logger.LevelDebug, "removing partial file %q after write error, result: %v", virtualTargetPath, errRemove) } } info, err := conn.doStatInternal(targetPath, 0, false, false) if err == nil { updateUserQuotaAfterFileWrite(conn, targetPath, numFiles, info.Size()-truncatedSize) var fsSrcPath string var errSrcFs error if virtualSourcePath != "" { _, fsSrcPath, errSrcFs = conn.GetFsAndResolvedPath(virtualSourcePath) } if errSrcFs == nil && errDstFs == nil { elapsed := time.Since(startTime).Nanoseconds() / 1000000 if errTransfer == nil { errTransfer = errWrite } if operation == operationCopy { logger.CommandLog(copyLogSender, fsSrcPath, fsDstPath, conn.User.Username, "", conn.ID, conn.protocol, -1, -1, "", "", "", info.Size(), conn.localAddr, conn.remoteAddr, elapsed) } ExecuteActionNotification(conn, operation, fsSrcPath, virtualSourcePath, fsDstPath, virtualTargetPath, "", info.Size(), errTransfer, elapsed) //nolint:errcheck } } else { eventManagerLog(logger.LevelWarn, "unable to update quota after writing %q: %v", targetPath, err) } if errTransfer != nil { return errTransfer } return errWrite } func updateUserQuotaAfterFileWrite(conn *BaseConnection, virtualPath string, numFiles int, fileSize int64) { vfolder, err := conn.User.GetVirtualFolderForPath(path.Dir(virtualPath)) if err != nil { dataprovider.UpdateUserQuota(&conn.User, numFiles, fileSize, false) //nolint:errcheck return } dataprovider.UpdateVirtualFolderQuota(&vfolder.BaseVirtualFolder, numFiles, fileSize, false) //nolint:errcheck if vfolder.IsIncludedInUserQuota() { dataprovider.UpdateUserQuota(&conn.User, numFiles, fileSize, false) //nolint:errcheck } } func checkWriterPermsAndQuota(conn *BaseConnection, virtualPath string, numFiles int, expectedSize, truncatedSize int64) error { if numFiles == 0 { if !conn.User.HasPerm(dataprovider.PermOverwrite, path.Dir(virtualPath)) { return conn.GetPermissionDeniedError() } } else { if !conn.User.HasPerm(dataprovider.PermUpload, path.Dir(virtualPath)) { return conn.GetPermissionDeniedError() } } q, _ := conn.HasSpace(numFiles > 0, false, virtualPath) if !q.HasSpace { return conn.GetQuotaExceededError() } if expectedSize != -1 { sizeDiff := expectedSize - truncatedSize if sizeDiff > 0 { remainingSize := q.GetRemainingSize() if remainingSize > 0 && remainingSize < sizeDiff { return conn.GetQuotaExceededError() } } } return nil } func getFileWriter(conn *BaseConnection, virtualPath string, expectedSize int64) (io.WriteCloser, int, int64, func(), error) { fs, fsPath, err := conn.GetFsAndResolvedPath(virtualPath) if err != nil { return nil, 0, 0, nil, err } var truncatedSize, fileSize int64 numFiles := 1 isFileOverwrite := false info, err := fs.Lstat(fsPath) if err == nil { fileSize = info.Size() if info.IsDir() { return nil, numFiles, truncatedSize, nil, fmt.Errorf("cannot write to a directory: %q", virtualPath) } if info.Mode().IsRegular() { isFileOverwrite = true truncatedSize = fileSize } numFiles = 0 } if err != nil && !fs.IsNotExist(err) { return nil, numFiles, truncatedSize, nil, conn.GetFsError(fs, err) } if err := checkWriterPermsAndQuota(conn, virtualPath, numFiles, expectedSize, truncatedSize); err != nil { return nil, numFiles, truncatedSize, nil, err } f, w, cancelFn, err := fs.Create(fsPath, 0, conn.GetCreateChecks(virtualPath, numFiles == 1)) if err != nil { return nil, numFiles, truncatedSize, nil, conn.GetFsError(fs, err) } vfs.SetPathPermissions(fs, fsPath, conn.User.GetUID(), conn.User.GetGID()) if isFileOverwrite { if vfs.HasTruncateSupport(fs) || vfs.IsCryptOsFs(fs) { updateUserQuotaAfterFileWrite(conn, virtualPath, numFiles, -fileSize) truncatedSize = 0 } } if cancelFn == nil { cancelFn = func() {} } if f != nil { return f, numFiles, truncatedSize, cancelFn, nil } return w, numFiles, truncatedSize, cancelFn, nil } func addZipEntry(wr *zipWriterWrapper, conn *BaseConnection, entryPath, baseDir string) error { if entryPath == wr.Name { // skip the archive itself return nil } info, err := conn.DoStat(entryPath, 1, false) if err != nil { eventManagerLog(logger.LevelError, "unable to add zip entry %q, stat error: %v", entryPath, err) return err } entryName, err := getZipEntryName(entryPath, baseDir) if err != nil { eventManagerLog(logger.LevelError, "unable to get zip entry name: %v", err) return err } if _, ok := wr.Entries[entryName]; ok { eventManagerLog(logger.LevelInfo, "skipping duplicate zip entry %q, is dir %t", entryPath, info.IsDir()) return nil } wr.Entries[entryName] = true if info.IsDir() { _, err = wr.Writer.CreateHeader(&zip.FileHeader{ Name: entryName + "/", Method: zip.Deflate, Modified: info.ModTime(), }) if err != nil { eventManagerLog(logger.LevelError, "unable to create zip entry %q: %v", entryPath, err) return fmt.Errorf("unable to create zip entry %q: %w", entryPath, err) } contents, err := conn.ListDir(entryPath) if err != nil { eventManagerLog(logger.LevelError, "unable to add zip entry %q, read dir error: %v", entryPath, err) return fmt.Errorf("unable to add zip entry %q: %w", entryPath, err) } for _, info := range contents { fullPath := util.CleanPath(path.Join(entryPath, info.Name())) if err := addZipEntry(wr, conn, fullPath, baseDir); err != nil { eventManagerLog(logger.LevelError, "unable to add zip entry: %v", err) return err } } return nil } if !info.Mode().IsRegular() { // we only allow regular files eventManagerLog(logger.LevelInfo, "skipping zip entry for non regular file %q", entryPath) return nil } reader, cancelFn, err := getFileReader(conn, entryPath) if err != nil { eventManagerLog(logger.LevelError, "unable to add zip entry %q, cannot open file: %v", entryPath, err) return fmt.Errorf("unable to open %q: %w", entryPath, err) } defer cancelFn() defer reader.Close() f, err := wr.Writer.CreateHeader(&zip.FileHeader{ Name: entryName, Method: zip.Deflate, Modified: info.ModTime(), }) if err != nil { eventManagerLog(logger.LevelError, "unable to create zip entry %q: %v", entryPath, err) return fmt.Errorf("unable to create zip entry %q: %w", entryPath, err) } _, err = io.Copy(f, reader) return err } func getZipEntryName(entryPath, baseDir string) (string, error) { if !strings.HasPrefix(entryPath, baseDir) { return "", fmt.Errorf("entry path %q is outside base dir %q", entryPath, baseDir) } entryPath = strings.TrimPrefix(entryPath, baseDir) return strings.TrimPrefix(entryPath, "/"), nil } func getFileReader(conn *BaseConnection, virtualPath string) (io.ReadCloser, func(), error) { if !conn.User.HasPerm(dataprovider.PermDownload, path.Dir(virtualPath)) { return nil, nil, conn.GetPermissionDeniedError() } fs, fsPath, err := conn.GetFsAndResolvedPath(virtualPath) if err != nil { return nil, nil, err } f, r, cancelFn, err := fs.Open(fsPath, 0) if err != nil { return nil, nil, conn.GetFsError(fs, err) } if cancelFn == nil { cancelFn = func() {} } if f != nil { return f, cancelFn, nil } return r, cancelFn, nil } func writeFileContent(conn *BaseConnection, virtualPath string, w io.Writer) error { reader, cancelFn, err := getFileReader(conn, virtualPath) if err != nil { return err } defer cancelFn() defer reader.Close() _, err = io.Copy(w, reader) return err } func getFileContentFn(conn *BaseConnection, virtualPath string, size int64) func(w io.Writer) (int64, error) { return func(w io.Writer) (int64, error) { reader, cancelFn, err := getFileReader(conn, virtualPath) if err != nil { return 0, err } defer cancelFn() defer reader.Close() return io.CopyN(w, reader, size) } } func getMailAttachments(conn *BaseConnection, attachments []string, replacer *strings.Replacer) ([]*mail.File, error) { var files []*mail.File totalSize := int64(0) for _, virtualPath := range replacePathsPlaceholders(attachments, replacer) { info, err := conn.DoStat(virtualPath, 0, false) if err != nil { return nil, fmt.Errorf("unable to get info for file %q, user %q: %w", virtualPath, conn.User.Username, err) } if !info.Mode().IsRegular() { return nil, fmt.Errorf("cannot attach non regular file %q", virtualPath) } totalSize += info.Size() if totalSize > maxAttachmentsSize { return nil, fmt.Errorf("unable to send files as attachment, size too large: %s", util.ByteCountIEC(totalSize)) } files = append(files, &mail.File{ Name: path.Base(virtualPath), Header: make(map[string][]string), Writer: getFileContentFn(conn, virtualPath, info.Size()), }) } return files, nil } func replaceWithReplacer(input string, replacer *strings.Replacer) string { if !strings.Contains(input, "{{") { return input } return replacer.Replace(input) } func checkEventConditionPattern(p dataprovider.ConditionPattern, name string) bool { var matched bool var err error if strings.Contains(p.Pattern, "**") { matched, err = doublestar.Match(p.Pattern, name) } else { matched, err = path.Match(p.Pattern, name) } if err != nil { eventManagerLog(logger.LevelError, "pattern matching error %q, err: %v", p.Pattern, err) return false } if p.InverseMatch { return !matched } return matched } func checkUserConditionOptions(user *dataprovider.User, conditions *dataprovider.ConditionOptions) bool { if !checkEventConditionPatterns(user.Username, conditions.Names) { return false } if !checkEventConditionPatterns(user.Role, conditions.RoleNames) { return false } if !checkEventGroupConditionPatterns(user.Groups, conditions.GroupNames) { return false } return true } // checkConditionPatterns returns false if patterns are defined and no match is found func checkEventConditionPatterns(name string, patterns []dataprovider.ConditionPattern) bool { if len(patterns) == 0 { return true } matches := false for _, p := range patterns { // assume, that multiple InverseMatches are set if p.InverseMatch { if checkEventConditionPattern(p, name) { matches = true } else { return false } } else if checkEventConditionPattern(p, name) { return true } } return matches } func checkEventGroupConditionPatterns(groups []sdk.GroupMapping, patterns []dataprovider.ConditionPattern) bool { if len(patterns) == 0 { return true } matches := false for _, group := range groups { for _, p := range patterns { // assume, that multiple InverseMatches are set if p.InverseMatch { if checkEventConditionPattern(p, group.Name) { matches = true } else { return false } } else { if checkEventConditionPattern(p, group.Name) { return true } } } } return matches } func getHTTPRuleActionEndpoint(c *dataprovider.EventActionHTTPConfig, replacer *strings.Replacer) (string, error) { u, err := url.Parse(c.Endpoint) if err != nil { return "", fmt.Errorf("invalid endpoint: %w", err) } if strings.Contains(u.Path, "{{") { pathComponents := strings.Split(u.Path, "/") for idx := range pathComponents { part := replaceWithReplacer(pathComponents[idx], replacer) if part != pathComponents[idx] { pathComponents[idx] = url.PathEscape(part) } } u.Path = "" u = u.JoinPath(pathComponents...) } if len(c.QueryParameters) > 0 { q := u.Query() for _, keyVal := range c.QueryParameters { q.Add(keyVal.Key, replaceWithReplacer(keyVal.Value, replacer)) } u.RawQuery = q.Encode() } return u.String(), nil } func writeHTTPPart(m *multipart.Writer, part dataprovider.HTTPPart, h textproto.MIMEHeader, conn *BaseConnection, replacer *strings.Replacer, params *EventParams, addObjectData bool, ) error { partWriter, err := m.CreatePart(h) if err != nil { eventManagerLog(logger.LevelError, "unable to create part %q, err: %v", part.Name, err) return err } if part.Body != "" { cType := h.Get("Content-Type") if part.Body != objDataPlaceholder && strings.Contains(strings.ToLower(cType), "application/json") { replacements := params.getStringReplacements(addObjectData, true) jsonReplacer := strings.NewReplacer(replacements...) _, err = partWriter.Write([]byte(replaceWithReplacer(part.Body, jsonReplacer))) } else { _, err = partWriter.Write([]byte(replaceWithReplacer(part.Body, replacer))) } if err != nil { eventManagerLog(logger.LevelError, "unable to write part %q, err: %v", part.Name, err) return err } return nil } if part.Filepath == dataprovider.RetentionReportPlaceHolder { data, err := params.getCompressedDataRetentionReport() if err != nil { return err } _, err = partWriter.Write(data) if err != nil { eventManagerLog(logger.LevelError, "unable to write part %q, err: %v", part.Name, err) return err } return nil } err = writeFileContent(conn, util.CleanPath(replacer.Replace(part.Filepath)), partWriter) if err != nil { eventManagerLog(logger.LevelError, "unable to write file part %q, err: %v", part.Name, err) return err } return nil } func jsonEscapeRuleActionBody(c *dataprovider.EventActionHTTPConfig) bool { return c.Body != objDataPlaceholder && c.HasJSONBody() } func getHTTPRuleActionBody(c *dataprovider.EventActionHTTPConfig, replacer *strings.Replacer, cancel context.CancelFunc, user dataprovider.User, params *EventParams, addObjectData bool, ) (io.Reader, string, error) { var body io.Reader if c.Method == http.MethodGet { return body, "", nil } if c.Body != "" { if c.Body == dataprovider.RetentionReportPlaceHolder { data, err := params.getCompressedDataRetentionReport() if err != nil { return body, "", err } return bytes.NewBuffer(data), "", nil } if jsonEscapeRuleActionBody(c) { replacements := params.getStringReplacements(addObjectData, true) jsonReplacer := strings.NewReplacer(replacements...) return bytes.NewBufferString(replaceWithReplacer(c.Body, jsonReplacer)), "", nil } return bytes.NewBufferString(replaceWithReplacer(c.Body, replacer)), "", nil } if len(c.Parts) > 0 { r, w := io.Pipe() m := multipart.NewWriter(w) var conn *BaseConnection if user.Username != "" { var err error user, err = getUserForEventAction(user) if err != nil { return body, "", err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) if err != nil { user.CloseFs() //nolint:errcheck return body, "", fmt.Errorf("error getting multipart file/s, unable to check root fs for user %q: %w", user.Username, err) } conn = NewBaseConnection(connectionID, protocolEventAction, "", "", user) } go func() { defer w.Close() defer user.CloseFs() //nolint:errcheck for _, part := range c.Parts { h := make(textproto.MIMEHeader) if part.Body != "" { h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="%s"`, multipartQuoteEscaper.Replace(part.Name))) } else { h.Set("Content-Disposition", fmt.Sprintf(`form-data; name="%s"; filename="%s"`, multipartQuoteEscaper.Replace(part.Name), multipartQuoteEscaper.Replace(path.Base(part.Filepath)))) contentType := mime.TypeByExtension(path.Ext(part.Filepath)) if contentType == "" { contentType = "application/octet-stream" } h.Set("Content-Type", contentType) } for _, keyVal := range part.Headers { h.Set(keyVal.Key, replaceWithReplacer(keyVal.Value, replacer)) } if err := writeHTTPPart(m, part, h, conn, replacer, params, addObjectData); err != nil { cancel() return } } m.Close() }() return r, m.FormDataContentType(), nil } return body, "", nil } func setHTTPReqHeaders(req *http.Request, c *dataprovider.EventActionHTTPConfig, replacer *strings.Replacer, contentType string, ) { if contentType != "" { req.Header.Set("Content-Type", contentType) } if c.Username != "" || c.Password.GetPayload() != "" { req.SetBasicAuth(replaceWithReplacer(c.Username, replacer), c.Password.GetPayload()) } for _, keyVal := range c.Headers { req.Header.Set(keyVal.Key, replaceWithReplacer(keyVal.Value, replacer)) } } func executeHTTPRuleAction(c dataprovider.EventActionHTTPConfig, params *EventParams) error { if err := c.TryDecryptPassword(); err != nil { return err } addObjectData := false if params.Object != nil { addObjectData = c.HasObjectData() } replacements := params.getStringReplacements(addObjectData, false) replacer := strings.NewReplacer(replacements...) endpoint, err := getHTTPRuleActionEndpoint(&c, replacer) if err != nil { return err } ctx, cancel := c.GetContext() defer cancel() var user dataprovider.User if c.HasMultipartFiles() { user, err = params.getUserFromSender() if err != nil { return err } } body, contentType, err := getHTTPRuleActionBody(&c, replacer, cancel, user, params, addObjectData) if err != nil { return err } if body != nil { rc, ok := body.(io.ReadCloser) if ok { defer rc.Close() } } req, err := http.NewRequestWithContext(ctx, c.Method, endpoint, body) if err != nil { return err } setHTTPReqHeaders(req, &c, replacer, contentType) client := c.GetHTTPClient() defer client.CloseIdleConnections() startTime := time.Now() resp, err := client.Do(req) if err != nil { eventManagerLog(logger.LevelDebug, "unable to send http notification, endpoint: %s, elapsed: %s, err: %v", endpoint, time.Since(startTime), err) return fmt.Errorf("error sending HTTP request: %w", err) } defer resp.Body.Close() eventManagerLog(logger.LevelDebug, "http notification sent, endpoint: %s, elapsed: %s, status code: %d", endpoint, time.Since(startTime), resp.StatusCode) if resp.StatusCode < http.StatusOK || resp.StatusCode > http.StatusNoContent { if rb, err := io.ReadAll(io.LimitReader(resp.Body, 2048)); err == nil { eventManagerLog(logger.LevelDebug, "error notification response from endpoint %q: %s", endpoint, string(rb)) } return fmt.Errorf("unexpected status code: %d", resp.StatusCode) } return nil } func executeCommandRuleAction(c dataprovider.EventActionCommandConfig, params *EventParams) error { addObjectData := false if params.Object != nil { for _, k := range c.EnvVars { if strings.Contains(k.Value, objDataPlaceholder) { addObjectData = true break } } } replacements := params.getStringReplacements(addObjectData, false) replacer := strings.NewReplacer(replacements...) args := make([]string, 0, len(c.Args)) for _, arg := range c.Args { args = append(args, replaceWithReplacer(arg, replacer)) } ctx, cancel := context.WithTimeout(context.Background(), time.Duration(c.Timeout)*time.Second) defer cancel() cmd := exec.CommandContext(ctx, c.Cmd, args...) cmd.Env = []string{} for _, keyVal := range c.EnvVars { cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", keyVal.Key, replaceWithReplacer(keyVal.Value, replacer))) } startTime := time.Now() err := cmd.Run() eventManagerLog(logger.LevelDebug, "executed command %q, elapsed: %s, error: %v", c.Cmd, time.Since(startTime), err) return err } func getEmailAddressesWithReplacer(addrs []string, replacer *strings.Replacer) []string { if len(addrs) == 0 { return nil } recipients := make([]string, 0, len(addrs)) for _, recipient := range addrs { rcpt := replaceWithReplacer(recipient, replacer) if rcpt != "" { recipients = append(recipients, rcpt) } } return recipients } func executeEmailRuleAction(c dataprovider.EventActionEmailConfig, params *EventParams) error { addObjectData := false if params.Object != nil { if strings.Contains(c.Body, objDataPlaceholder) { addObjectData = true } } replacements := params.getStringReplacements(addObjectData, false) replacer := strings.NewReplacer(replacements...) body := replaceWithReplacer(c.Body, replacer) subject := replaceWithReplacer(c.Subject, replacer) recipients := getEmailAddressesWithReplacer(c.Recipients, replacer) bcc := getEmailAddressesWithReplacer(c.Bcc, replacer) startTime := time.Now() var files []*mail.File fileAttachments := make([]string, 0, len(c.Attachments)) for _, attachment := range c.Attachments { if attachment == dataprovider.RetentionReportPlaceHolder { f, err := params.getRetentionReportsAsMailAttachment() if err != nil { return err } files = append(files, f) continue } fileAttachments = append(fileAttachments, attachment) } if len(fileAttachments) > 0 { user, err := params.getUserFromSender() if err != nil { return err } user, err = getUserForEventAction(user) if err != nil { return err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) defer user.CloseFs() //nolint:errcheck if err != nil { return fmt.Errorf("error getting email attachments, unable to check root fs for user %q: %w", user.Username, err) } conn := NewBaseConnection(connectionID, protocolEventAction, "", "", user) res, err := getMailAttachments(conn, fileAttachments, replacer) if err != nil { return err } files = append(files, res...) } err := smtp.SendEmail(recipients, bcc, subject, body, smtp.EmailContentType(c.ContentType), files...) eventManagerLog(logger.LevelDebug, "executed email notification action, elapsed: %s, error: %v", time.Since(startTime), err) if err != nil { return fmt.Errorf("unable to send email: %w", err) } return nil } func getUserForEventAction(user dataprovider.User) (dataprovider.User, error) { err := user.LoadAndApplyGroupSettings() if err != nil { eventManagerLog(logger.LevelError, "unable to get group for user %q: %+v", user.Username, err) return dataprovider.User{}, fmt.Errorf("unable to get groups for user %q", user.Username) } user.UploadDataTransfer = 0 user.UploadBandwidth = 0 user.DownloadBandwidth = 0 user.Filters.DisableFsChecks = false user.Filters.FilePatterns = nil user.Filters.BandwidthLimits = nil for k := range user.Permissions { user.Permissions[k] = []string{dataprovider.PermAny} } return user, nil } func replacePathsPlaceholders(paths []string, replacer *strings.Replacer) []string { results := make([]string, 0, len(paths)) for _, p := range paths { results = append(results, util.CleanPath(replaceWithReplacer(p, replacer))) } return util.RemoveDuplicates(results, false) } func executeDeleteFileFsAction(conn *BaseConnection, item string, info os.FileInfo) error { fs, fsPath, err := conn.GetFsAndResolvedPath(item) if err != nil { return err } return conn.RemoveFile(fs, fsPath, item, info) } func executeDeleteFsActionForUser(deletes []string, replacer *strings.Replacer, user dataprovider.User) error { user, err := getUserForEventAction(user) if err != nil { return err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) defer user.CloseFs() //nolint:errcheck if err != nil { return fmt.Errorf("delete error, unable to check root fs for user %q: %w", user.Username, err) } conn := NewBaseConnection(connectionID, protocolEventAction, "", "", user) for _, item := range replacePathsPlaceholders(deletes, replacer) { info, err := conn.DoStat(item, 0, false) if err != nil { if conn.IsNotExistError(err) { continue } return fmt.Errorf("unable to check item to delete %q, user %q: %w", item, user.Username, err) } if info.IsDir() { if err = conn.RemoveDir(item); err != nil { return fmt.Errorf("unable to remove dir %q, user %q: %w", item, user.Username, err) } } else { if err = executeDeleteFileFsAction(conn, item, info); err != nil { return fmt.Errorf("unable to remove file %q, user %q: %w", item, user.Username, err) } } eventManagerLog(logger.LevelDebug, "item %q removed for user %q", item, user.Username) } return nil } func executeDeleteFsRuleAction(deletes []string, replacer *strings.Replacer, conditions dataprovider.ConditionOptions, params *EventParams, ) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping fs delete for user %s, condition options don't match", user.Username) continue } } executed++ if err = executeDeleteFsActionForUser(deletes, replacer, user); err != nil { params.AddError(err) failures = append(failures, user.Username) } } if len(failures) > 0 { return fmt.Errorf("fs delete failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no delete executed") return errors.New("no delete executed") } return nil } func executeMkDirsFsActionForUser(dirs []string, replacer *strings.Replacer, user dataprovider.User) error { user, err := getUserForEventAction(user) if err != nil { return err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) defer user.CloseFs() //nolint:errcheck if err != nil { return fmt.Errorf("mkdir error, unable to check root fs for user %q: %w", user.Username, err) } conn := NewBaseConnection(connectionID, protocolEventAction, "", "", user) for _, item := range replacePathsPlaceholders(dirs, replacer) { if err = conn.CheckParentDirs(path.Dir(item)); err != nil { return fmt.Errorf("unable to check parent dirs for %q, user %q: %w", item, user.Username, err) } if err = conn.createDirIfMissing(item); err != nil { return fmt.Errorf("unable to create dir %q, user %q: %w", item, user.Username, err) } eventManagerLog(logger.LevelDebug, "directory %q created for user %q", item, user.Username) } return nil } func executeMkdirFsRuleAction(dirs []string, replacer *strings.Replacer, conditions dataprovider.ConditionOptions, params *EventParams, ) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping fs mkdir for user %s, condition options don't match", user.Username) continue } } executed++ if err = executeMkDirsFsActionForUser(dirs, replacer, user); err != nil { failures = append(failures, user.Username) } } if len(failures) > 0 { return fmt.Errorf("fs mkdir failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no mkdir executed") return errors.New("no mkdir executed") } return nil } func executeRenameFsActionForUser(renames []dataprovider.KeyValue, replacer *strings.Replacer, user dataprovider.User, ) error { user, err := getUserForEventAction(user) if err != nil { return err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) defer user.CloseFs() //nolint:errcheck if err != nil { return fmt.Errorf("rename error, unable to check root fs for user %q: %w", user.Username, err) } conn := NewBaseConnection(connectionID, protocolEventAction, "", "", user) for _, item := range renames { source := util.CleanPath(replaceWithReplacer(item.Key, replacer)) target := util.CleanPath(replaceWithReplacer(item.Value, replacer)) if err = conn.renameInternal(source, target, true); err != nil { return fmt.Errorf("unable to rename %q->%q, user %q: %w", source, target, user.Username, err) } eventManagerLog(logger.LevelDebug, "rename %q->%q ok, user %q", source, target, user.Username) } return nil } func executeCopyFsActionForUser(copy []dataprovider.KeyValue, replacer *strings.Replacer, user dataprovider.User, ) error { user, err := getUserForEventAction(user) if err != nil { return err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) defer user.CloseFs() //nolint:errcheck if err != nil { return fmt.Errorf("copy error, unable to check root fs for user %q: %w", user.Username, err) } conn := NewBaseConnection(connectionID, protocolEventAction, "", "", user) for _, item := range copy { source := util.CleanPath(replaceWithReplacer(item.Key, replacer)) target := util.CleanPath(replaceWithReplacer(item.Value, replacer)) if strings.HasSuffix(item.Key, "/") { source += "/" } if strings.HasSuffix(item.Value, "/") { target += "/" } if err = conn.Copy(source, target); err != nil { return fmt.Errorf("unable to copy %q->%q, user %q: %w", source, target, user.Username, err) } eventManagerLog(logger.LevelDebug, "copy %q->%q ok, user %q", source, target, user.Username) } return nil } func executeExistFsActionForUser(exist []string, replacer *strings.Replacer, user dataprovider.User, ) error { user, err := getUserForEventAction(user) if err != nil { return err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) defer user.CloseFs() //nolint:errcheck if err != nil { return fmt.Errorf("existence check error, unable to check root fs for user %q: %w", user.Username, err) } conn := NewBaseConnection(connectionID, protocolEventAction, "", "", user) for _, item := range replacePathsPlaceholders(exist, replacer) { if _, err = conn.DoStat(item, 0, false); err != nil { return fmt.Errorf("error checking existence for path %q, user %q: %w", item, user.Username, err) } eventManagerLog(logger.LevelDebug, "path %q exists for user %q", item, user.Username) } return nil } func executeRenameFsRuleAction(renames []dataprovider.KeyValue, replacer *strings.Replacer, conditions dataprovider.ConditionOptions, params *EventParams, ) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping fs rename for user %s, condition options don't match", user.Username) continue } } executed++ if err = executeRenameFsActionForUser(renames, replacer, user); err != nil { failures = append(failures, user.Username) params.AddError(err) } } if len(failures) > 0 { return fmt.Errorf("fs rename failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no rename executed") return errors.New("no rename executed") } return nil } func executeCopyFsRuleAction(copy []dataprovider.KeyValue, replacer *strings.Replacer, conditions dataprovider.ConditionOptions, params *EventParams, ) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string var executed int for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping fs copy for user %s, condition options don't match", user.Username) continue } } executed++ if err = executeCopyFsActionForUser(copy, replacer, user); err != nil { failures = append(failures, user.Username) params.AddError(err) } } if len(failures) > 0 { return fmt.Errorf("fs copy failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no copy executed") return errors.New("no copy executed") } return nil } func getArchiveBaseDir(paths []string) string { var parentDirs []string for _, p := range paths { parentDirs = append(parentDirs, path.Dir(p)) } parentDirs = util.RemoveDuplicates(parentDirs, false) baseDir := "/" if len(parentDirs) == 1 { baseDir = parentDirs[0] } return baseDir } func getSizeForPath(conn *BaseConnection, p string, info os.FileInfo) (int64, error) { if info.IsDir() { var dirSize int64 entries, err := conn.ListDir(p) if err != nil { return 0, err } for _, entry := range entries { size, err := getSizeForPath(conn, path.Join(p, entry.Name()), entry) if err != nil { return 0, err } dirSize += size } return dirSize, nil } if info.Mode().IsRegular() { return info.Size(), nil } return 0, nil } func estimateZipSize(conn *BaseConnection, zipPath string, paths []string) (int64, error) { q, _ := conn.HasSpace(false, false, zipPath) if q.HasSpace && q.GetRemainingSize() > 0 { var size int64 for _, item := range paths { info, err := conn.DoStat(item, 1, false) if err != nil { return size, err } itemSize, err := getSizeForPath(conn, item, info) if err != nil { return size, err } size += itemSize } eventManagerLog(logger.LevelDebug, "archive paths %v, archive name %q, size: %d", paths, zipPath, size) // we assume the zip size will be half of the real size return size / 2, nil } return -1, nil } func executeCompressFsActionForUser(c dataprovider.EventActionFsCompress, replacer *strings.Replacer, user dataprovider.User, ) error { user, err := getUserForEventAction(user) if err != nil { return err } connectionID := fmt.Sprintf("%s_%s", protocolEventAction, xid.New().String()) err = user.CheckFsRoot(connectionID) defer user.CloseFs() //nolint:errcheck if err != nil { return fmt.Errorf("compress error, unable to check root fs for user %q: %w", user.Username, err) } conn := NewBaseConnection(connectionID, protocolEventAction, "", "", user) name := util.CleanPath(replaceWithReplacer(c.Name, replacer)) conn.CheckParentDirs(path.Dir(name)) //nolint:errcheck paths := make([]string, 0, len(c.Paths)) for idx := range c.Paths { p := util.CleanPath(replaceWithReplacer(c.Paths[idx], replacer)) if p == name { return fmt.Errorf("cannot compress the archive to create: %q", name) } paths = append(paths, p) } paths = util.RemoveDuplicates(paths, false) estimatedSize, err := estimateZipSize(conn, name, paths) if err != nil { eventManagerLog(logger.LevelError, "unable to estimate size for archive %q: %v", name, err) return fmt.Errorf("unable to estimate archive size: %w", err) } writer, numFiles, truncatedSize, cancelFn, err := getFileWriter(conn, name, estimatedSize) if err != nil { eventManagerLog(logger.LevelError, "unable to create archive %q: %v", name, err) return fmt.Errorf("unable to create archive: %w", err) } defer cancelFn() baseDir := getArchiveBaseDir(paths) eventManagerLog(logger.LevelDebug, "creating archive %q for paths %+v", name, paths) zipWriter := &zipWriterWrapper{ Name: name, Writer: zip.NewWriter(writer), Entries: make(map[string]bool), } startTime := time.Now() for _, item := range paths { if err := addZipEntry(zipWriter, conn, item, baseDir); err != nil { closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload, startTime) //nolint:errcheck return err } } if err := zipWriter.Writer.Close(); err != nil { eventManagerLog(logger.LevelError, "unable to close zip file %q: %v", name, err) closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload, startTime) //nolint:errcheck return fmt.Errorf("unable to close zip file %q: %w", name, err) } return closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload, startTime) } func executeExistFsRuleAction(exist []string, replacer *strings.Replacer, conditions dataprovider.ConditionOptions, params *EventParams, ) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping fs exist for user %s, condition options don't match", user.Username) continue } } executed++ if err = executeExistFsActionForUser(exist, replacer, user); err != nil { failures = append(failures, user.Username) params.AddError(err) } } if len(failures) > 0 { return fmt.Errorf("fs existence check failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no existence check executed") return errors.New("no existence check executed") } return nil } func executeCompressFsRuleAction(c dataprovider.EventActionFsCompress, replacer *strings.Replacer, conditions dataprovider.ConditionOptions, params *EventParams, ) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping fs compress for user %s, condition options don't match", user.Username) continue } } executed++ if err = executeCompressFsActionForUser(c, replacer, user); err != nil { failures = append(failures, user.Username) params.AddError(err) } } if len(failures) > 0 { return fmt.Errorf("fs compress failed for users: %s", strings.Join(failures, ",")) } if executed == 0 { eventManagerLog(logger.LevelError, "no file/folder compressed") return errors.New("no file/folder compressed") } return nil } func executeFsRuleAction(c dataprovider.EventActionFilesystemConfig, conditions dataprovider.ConditionOptions, params *EventParams, ) error { addObjectData := false replacements := params.getStringReplacements(addObjectData, false) replacer := strings.NewReplacer(replacements...) switch c.Type { case dataprovider.FilesystemActionRename: return executeRenameFsRuleAction(c.Renames, replacer, conditions, params) case dataprovider.FilesystemActionDelete: return executeDeleteFsRuleAction(c.Deletes, replacer, conditions, params) case dataprovider.FilesystemActionMkdirs: return executeMkdirFsRuleAction(c.MkDirs, replacer, conditions, params) case dataprovider.FilesystemActionExist: return executeExistFsRuleAction(c.Exist, replacer, conditions, params) case dataprovider.FilesystemActionCompress: return executeCompressFsRuleAction(c.Compress, replacer, conditions, params) case dataprovider.FilesystemActionCopy: return executeCopyFsRuleAction(c.Copy, replacer, conditions, params) default: return fmt.Errorf("unsupported filesystem action %d", c.Type) } } func executeQuotaResetForUser(user *dataprovider.User) error { if err := user.LoadAndApplyGroupSettings(); err != nil { eventManagerLog(logger.LevelError, "skipping scheduled quota reset for user %s, cannot apply group settings: %v", user.Username, err) return err } if !QuotaScans.AddUserQuotaScan(user.Username, user.Role) { eventManagerLog(logger.LevelError, "another quota scan is already in progress for user %q", user.Username) return fmt.Errorf("another quota scan is in progress for user %q", user.Username) } defer QuotaScans.RemoveUserQuotaScan(user.Username) numFiles, size, err := user.ScanQuota() if err != nil { eventManagerLog(logger.LevelError, "error scanning quota for user %q: %v", user.Username, err) return fmt.Errorf("error scanning quota for user %q: %w", user.Username, err) } err = dataprovider.UpdateUserQuota(user, numFiles, size, true) if err != nil { eventManagerLog(logger.LevelError, "error updating quota for user %q: %v", user.Username, err) return fmt.Errorf("error updating quota for user %q: %w", user.Username, err) } return nil } func executeUsersQuotaResetRuleAction(conditions dataprovider.ConditionOptions, params *EventParams) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping quota reset for user %q, condition options don't match", user.Username) continue } } executed++ if err = executeQuotaResetForUser(&user); err != nil { params.AddError(err) failures = append(failures, user.Username) } } if len(failures) > 0 { return fmt.Errorf("quota reset failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no user quota reset executed") return errors.New("no user quota reset executed") } return nil } func executeFoldersQuotaResetRuleAction(conditions dataprovider.ConditionOptions, params *EventParams) error { folders, err := params.getFolders() if err != nil { return fmt.Errorf("unable to get folders: %w", err) } var failures []string executed := 0 for _, folder := range folders { // if sender is set, the conditions have already been evaluated if params.sender == "" && !checkEventConditionPatterns(folder.Name, conditions.Names) { eventManagerLog(logger.LevelDebug, "skipping scheduled quota reset for folder %s, name conditions don't match", folder.Name) continue } if !QuotaScans.AddVFolderQuotaScan(folder.Name) { eventManagerLog(logger.LevelError, "another quota scan is already in progress for folder %q", folder.Name) params.AddError(fmt.Errorf("another quota scan is already in progress for folder %q", folder.Name)) failures = append(failures, folder.Name) continue } executed++ f := vfs.VirtualFolder{ BaseVirtualFolder: folder, VirtualPath: "/", } numFiles, size, err := f.ScanQuota() QuotaScans.RemoveVFolderQuotaScan(folder.Name) if err != nil { eventManagerLog(logger.LevelError, "error scanning quota for folder %q: %v", folder.Name, err) params.AddError(fmt.Errorf("error scanning quota for folder %q: %w", folder.Name, err)) failures = append(failures, folder.Name) continue } err = dataprovider.UpdateVirtualFolderQuota(&folder, numFiles, size, true) if err != nil { eventManagerLog(logger.LevelError, "error updating quota for folder %q: %v", folder.Name, err) params.AddError(fmt.Errorf("error updating quota for folder %q: %w", folder.Name, err)) failures = append(failures, folder.Name) } } if len(failures) > 0 { return fmt.Errorf("quota reset failed for folders: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no folder quota reset executed") return errors.New("no folder quota reset executed") } return nil } func executeTransferQuotaResetRuleAction(conditions dataprovider.ConditionOptions, params *EventParams) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping scheduled transfer quota reset for user %s, condition options don't match", user.Username) continue } } executed++ err = dataprovider.UpdateUserTransferQuota(&user, 0, 0, true) if err != nil { eventManagerLog(logger.LevelError, "error updating transfer quota for user %q: %v", user.Username, err) params.AddError(fmt.Errorf("error updating transfer quota for user %q: %w", user.Username, err)) failures = append(failures, user.Username) } } if len(failures) > 0 { return fmt.Errorf("transfer quota reset failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no transfer quota reset executed") return errors.New("no transfer quota reset executed") } return nil } func executeDataRetentionCheckForUser(user dataprovider.User, folders []dataprovider.FolderRetention, params *EventParams, actionName string, ) error { if err := user.LoadAndApplyGroupSettings(); err != nil { eventManagerLog(logger.LevelError, "skipping scheduled retention check for user %s, cannot apply group settings: %v", user.Username, err) return err } check := RetentionCheck{ Folders: folders, } c := RetentionChecks.Add(check, &user) if c == nil { eventManagerLog(logger.LevelError, "another retention check is already in progress for user %q", user.Username) return fmt.Errorf("another retention check is in progress for user %q", user.Username) } defer func() { params.retentionChecks = append(params.retentionChecks, executedRetentionCheck{ Username: user.Username, ActionName: actionName, Results: c.results, }) }() if err := c.Start(); err != nil { eventManagerLog(logger.LevelError, "error checking retention for user %q: %v", user.Username, err) return fmt.Errorf("error checking retention for user %q: %w", user.Username, err) } return nil } func executeDataRetentionCheckRuleAction(config dataprovider.EventActionDataRetentionConfig, conditions dataprovider.ConditionOptions, params *EventParams, actionName string, ) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string executed := 0 for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping scheduled retention check for user %s, condition options don't match", user.Username) continue } } executed++ if err = executeDataRetentionCheckForUser(user, config.Folders, params, actionName); err != nil { failures = append(failures, user.Username) params.AddError(err) } } if len(failures) > 0 { return fmt.Errorf("retention check failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no retention check executed") return errors.New("no retention check executed") } return nil } func executeUserExpirationCheckRuleAction(conditions dataprovider.ConditionOptions, params *EventParams) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string var executed int for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping expiration check for user %q, condition options don't match", user.Username) continue } } executed++ if user.ExpirationDate > 0 { expDate := util.GetTimeFromMsecSinceEpoch(user.ExpirationDate) if expDate.Before(time.Now()) { failures = append(failures, user.Username) } } } if len(failures) > 0 { return fmt.Errorf("expired users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no user expiration check executed") return errors.New("no user expiration check executed") } return nil } func executeMetadataCheckForUser(user *dataprovider.User) error { if err := user.LoadAndApplyGroupSettings(); err != nil { eventManagerLog(logger.LevelError, "skipping scheduled quota reset for user %s, cannot apply group settings: %v", user.Username, err) return err } if !ActiveMetadataChecks.Add(user.Username, user.Role) { eventManagerLog(logger.LevelError, "another metadata check is already in progress for user %q", user.Username) return fmt.Errorf("another metadata check is in progress for user %q", user.Username) } defer ActiveMetadataChecks.Remove(user.Username) if err := user.CheckMetadataConsistency(); err != nil { eventManagerLog(logger.LevelError, "error checking metadata consistence for user %q: %v", user.Username, err) return fmt.Errorf("error checking metadata consistence for user %q: %w", user.Username, err) } return nil } func executeMetadataCheckRuleAction(conditions dataprovider.ConditionOptions, params *EventParams) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string var executed int for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping metadata check for user %q, condition options don't match", user.Username) continue } } executed++ if err = executeMetadataCheckForUser(&user); err != nil { params.AddError(err) failures = append(failures, user.Username) } } if len(failures) > 0 { return fmt.Errorf("metadata check failed for users: %s", strings.Join(failures, ", ")) } if executed == 0 { eventManagerLog(logger.LevelError, "no metadata check executed") return errors.New("no metadata check executed") } return nil } func executePwdExpirationCheckForUser(user *dataprovider.User, config dataprovider.EventActionPasswordExpiration) error { if err := user.LoadAndApplyGroupSettings(); err != nil { eventManagerLog(logger.LevelError, "skipping password expiration check for user %q, cannot apply group settings: %v", user.Username, err) return err } if user.ExpirationDate > 0 { if expDate := util.GetTimeFromMsecSinceEpoch(user.ExpirationDate); expDate.Before(time.Now()) { eventManagerLog(logger.LevelDebug, "skipping password expiration check for expired user %q, expiration date: %s", user.Username, expDate) return nil } } if user.Filters.PasswordExpiration == 0 { eventManagerLog(logger.LevelDebug, "password expiration not set for user %q skipping check", user.Username) return nil } days := user.PasswordExpiresIn() if days > config.Threshold { eventManagerLog(logger.LevelDebug, "password for user %q expires in %d days, threshold %d, no need to notify", user.Username, days, config.Threshold) return nil } body := new(bytes.Buffer) data := make(map[string]any) data["Username"] = user.Username data["Days"] = days if err := smtp.RenderPasswordExpirationTemplate(body, data); err != nil { eventManagerLog(logger.LevelError, "unable to notify password expiration for user %s: %v", user.Username, err) return err } subject := "SFTPGo password expiration notification" startTime := time.Now() if err := smtp.SendEmail([]string{user.Email}, nil, subject, body.String(), smtp.EmailContentTypeTextHTML); err != nil { eventManagerLog(logger.LevelError, "unable to notify password expiration for user %s: %v, elapsed: %s", user.Username, err, time.Since(startTime)) return err } eventManagerLog(logger.LevelDebug, "password expiration email sent to user %s, days: %d, elapsed: %s", user.Username, days, time.Since(startTime)) return nil } func executePwdExpirationCheckRuleAction(config dataprovider.EventActionPasswordExpiration, conditions dataprovider.ConditionOptions, params *EventParams) error { users, err := params.getUsers() if err != nil { return fmt.Errorf("unable to get users: %w", err) } var failures []string for _, user := range users { // if sender is set, the conditions have already been evaluated if params.sender == "" { if !checkUserConditionOptions(&user, &conditions) { eventManagerLog(logger.LevelDebug, "skipping password check for user %q, condition options don't match", user.Username) continue } } if err = executePwdExpirationCheckForUser(&user, config); err != nil { params.AddError(err) failures = append(failures, user.Username) } } if len(failures) > 0 { return fmt.Errorf("password expiration check failed for users: %s", strings.Join(failures, ", ")) } return nil } func executeAdminCheckAction(c *dataprovider.EventActionIDPAccountCheck, params *EventParams) (*dataprovider.Admin, error) { admin, err := dataprovider.AdminExists(params.Name) exists := err == nil if exists && c.Mode == 1 { return &admin, nil } if err != nil && !errors.Is(err, util.ErrNotFound) { return nil, err } replacements := params.getStringReplacements(false, true) replacer := strings.NewReplacer(replacements...) data := replaceWithReplacer(c.TemplateAdmin, replacer) var newAdmin dataprovider.Admin err = json.Unmarshal([]byte(data), &newAdmin) if err != nil { return nil, err } if newAdmin.Password == "" { newAdmin.Password = util.GenerateUniqueID() } if exists { eventManagerLog(logger.LevelDebug, "updating admin %q after IDP login", params.Name) err = dataprovider.UpdateAdmin(&newAdmin, dataprovider.ActionExecutorSystem, "", "") } else { eventManagerLog(logger.LevelDebug, "creating admin %q after IDP login", params.Name) err = dataprovider.AddAdmin(&newAdmin, dataprovider.ActionExecutorSystem, "", "") } return &newAdmin, err } func executeUserCheckAction(c *dataprovider.EventActionIDPAccountCheck, params *EventParams) (*dataprovider.User, error) { user, err := dataprovider.UserExists(params.Name, "") exists := err == nil if exists && c.Mode == 1 { err = user.LoadAndApplyGroupSettings() return &user, err } if err != nil && !errors.Is(err, util.ErrNotFound) { return nil, err } replacements := params.getStringReplacements(false, true) replacer := strings.NewReplacer(replacements...) data := replaceWithReplacer(c.TemplateUser, replacer) var newUser dataprovider.User err = json.Unmarshal([]byte(data), &newUser) if err != nil { return nil, err } if exists { eventManagerLog(logger.LevelDebug, "updating user %q after IDP login", params.Name) err = dataprovider.UpdateUser(&newUser, dataprovider.ActionExecutorSystem, "", "") } else { eventManagerLog(logger.LevelDebug, "creating user %q after IDP login", params.Name) err = dataprovider.AddUser(&newUser, dataprovider.ActionExecutorSystem, "", "") } if err != nil { return nil, err } u, err := dataprovider.GetUserWithGroupSettings(params.Name, "") return &u, err } func executeRuleAction(action dataprovider.BaseEventAction, params *EventParams, conditions dataprovider.ConditionOptions, ) error { var err error switch action.Type { case dataprovider.ActionTypeHTTP: err = executeHTTPRuleAction(action.Options.HTTPConfig, params) case dataprovider.ActionTypeCommand: err = executeCommandRuleAction(action.Options.CmdConfig, params) case dataprovider.ActionTypeEmail: err = executeEmailRuleAction(action.Options.EmailConfig, params) case dataprovider.ActionTypeBackup: var backupPath string backupPath, err = dataprovider.ExecuteBackup() if err == nil { params.setBackupParams(backupPath) } case dataprovider.ActionTypeUserQuotaReset: err = executeUsersQuotaResetRuleAction(conditions, params) case dataprovider.ActionTypeFolderQuotaReset: err = executeFoldersQuotaResetRuleAction(conditions, params) case dataprovider.ActionTypeTransferQuotaReset: err = executeTransferQuotaResetRuleAction(conditions, params) case dataprovider.ActionTypeDataRetentionCheck: err = executeDataRetentionCheckRuleAction(action.Options.RetentionConfig, conditions, params, action.Name) case dataprovider.ActionTypeMetadataCheck: err = executeMetadataCheckRuleAction(conditions, params) case dataprovider.ActionTypeFilesystem: err = executeFsRuleAction(action.Options.FsConfig, conditions, params) case dataprovider.ActionTypePasswordExpirationCheck: err = executePwdExpirationCheckRuleAction(action.Options.PwdExpirationConfig, conditions, params) case dataprovider.ActionTypeUserExpirationCheck: err = executeUserExpirationCheckRuleAction(conditions, params) default: err = fmt.Errorf("unsupported action type: %d", action.Type) } if err != nil { err = fmt.Errorf("action %q failed: %w", action.Name, err) } params.AddError(err) return err } func executeIDPAccountCheckRule(rule dataprovider.EventRule, params EventParams) (*dataprovider.User, *dataprovider.Admin, error, ) { for _, action := range rule.Actions { if action.Type == dataprovider.ActionTypeIDPAccountCheck { startTime := time.Now() var user *dataprovider.User var admin *dataprovider.Admin var err error var failedActions []string paramsCopy := params.getACopy() switch params.Event { case IDPLoginAdmin: admin, err = executeAdminCheckAction(&action.BaseEventAction.Options.IDPConfig, paramsCopy) case IDPLoginUser: user, err = executeUserCheckAction(&action.BaseEventAction.Options.IDPConfig, paramsCopy) default: err = fmt.Errorf("unsupported IDP login event: %q", params.Event) } if err != nil { paramsCopy.AddError(fmt.Errorf("unable to handle %q: %w", params.Event, err)) eventManagerLog(logger.LevelError, "unable to handle IDP login event %q, err: %v", params.Event, err) failedActions = append(failedActions, action.Name) } else { eventManagerLog(logger.LevelDebug, "executed action %q for rule %q, elapsed %s", action.Name, rule.Name, time.Since(startTime)) } // execute async actions if any, including failure actions go executeRuleAsyncActions(rule, paramsCopy, failedActions) return user, admin, err } } eventManagerLog(logger.LevelError, "no action executed for IDP login event %q, event rule: %q", params.Event, rule.Name) return nil, nil, errors.New("no action executed") } func executeSyncRulesActions(rules []dataprovider.EventRule, params EventParams) error { var errRes error for _, rule := range rules { var failedActions []string paramsCopy := params.getACopy() for _, action := range rule.Actions { if !action.Options.IsFailureAction && action.Options.ExecuteSync { startTime := time.Now() if err := executeRuleAction(action.BaseEventAction, paramsCopy, rule.Conditions.Options); err != nil { eventManagerLog(logger.LevelError, "unable to execute sync action %q for rule %q, elapsed %s, err: %v", action.Name, rule.Name, time.Since(startTime), err) failedActions = append(failedActions, action.Name) // we return the last error, it is ok for now errRes = err if action.Options.StopOnFailure { break } } else { eventManagerLog(logger.LevelDebug, "executed sync action %q for rule %q, elapsed: %s", action.Name, rule.Name, time.Since(startTime)) } } } // execute async actions if any, including failure actions go executeRuleAsyncActions(rule, paramsCopy, failedActions) } return errRes } func executeAsyncRulesActions(rules []dataprovider.EventRule, params EventParams) { eventManager.addAsyncTask() defer eventManager.removeAsyncTask() for _, rule := range rules { executeRuleAsyncActions(rule, params.getACopy(), nil) } } func executeRuleAsyncActions(rule dataprovider.EventRule, params *EventParams, failedActions []string) { for _, action := range rule.Actions { if !action.Options.IsFailureAction && !action.Options.ExecuteSync { startTime := time.Now() if err := executeRuleAction(action.BaseEventAction, params, rule.Conditions.Options); err != nil { eventManagerLog(logger.LevelError, "unable to execute action %q for rule %q, elapsed %s, err: %v", action.Name, rule.Name, time.Since(startTime), err) failedActions = append(failedActions, action.Name) if action.Options.StopOnFailure { break } } else { eventManagerLog(logger.LevelDebug, "executed action %q for rule %q, elapsed %s", action.Name, rule.Name, time.Since(startTime)) } } } if len(failedActions) > 0 { params.updateStatusFromError = false // execute failure actions for _, action := range rule.Actions { if action.Options.IsFailureAction { startTime := time.Now() if err := executeRuleAction(action.BaseEventAction, params, rule.Conditions.Options); err != nil { eventManagerLog(logger.LevelError, "unable to execute failure action %q for rule %q, elapsed %s, err: %v", action.Name, rule.Name, time.Since(startTime), err) if action.Options.StopOnFailure { break } } else { eventManagerLog(logger.LevelDebug, "executed failure action %q for rule %q, elapsed: %s", action.Name, rule.Name, time.Since(startTime)) } } } } } type eventCronJob struct { ruleName string } func (j *eventCronJob) getTask(rule *dataprovider.EventRule) (dataprovider.Task, error) { if rule.GuardFromConcurrentExecution() { task, err := dataprovider.GetTaskByName(rule.Name) if err != nil { if errors.Is(err, util.ErrNotFound) { eventManagerLog(logger.LevelDebug, "adding task for rule %q", rule.Name) task = dataprovider.Task{ Name: rule.Name, UpdateAt: 0, Version: 0, } err = dataprovider.AddTask(rule.Name) if err != nil { eventManagerLog(logger.LevelWarn, "unable to add task for rule %q: %v", rule.Name, err) return task, err } } else { eventManagerLog(logger.LevelWarn, "unable to get task for rule %q: %v", rule.Name, err) } } return task, err } return dataprovider.Task{}, nil } func (j *eventCronJob) Run() { eventManagerLog(logger.LevelDebug, "executing scheduled rule %q", j.ruleName) rule, err := dataprovider.EventRuleExists(j.ruleName) if err != nil { eventManagerLog(logger.LevelError, "unable to load rule with name %q", j.ruleName) return } if err := rule.CheckActionsConsistency(""); err != nil { eventManagerLog(logger.LevelWarn, "scheduled rule %q skipped: %v", rule.Name, err) return } task, err := j.getTask(&rule) if err != nil { return } if task.Name != "" { updateInterval := 5 * time.Minute updatedAt := util.GetTimeFromMsecSinceEpoch(task.UpdateAt) if updatedAt.Add(updateInterval*2 + 1).After(time.Now()) { eventManagerLog(logger.LevelDebug, "task for rule %q too recent: %s, skip execution", rule.Name, updatedAt) return } err = dataprovider.UpdateTask(rule.Name, task.Version) if err != nil { eventManagerLog(logger.LevelInfo, "unable to update task timestamp for rule %q, skip execution, err: %v", rule.Name, err) return } ticker := time.NewTicker(updateInterval) done := make(chan bool) defer func() { done <- true ticker.Stop() }() go func(taskName string) { eventManagerLog(logger.LevelDebug, "update task %q timestamp worker started", taskName) for { select { case <-done: eventManagerLog(logger.LevelDebug, "update task %q timestamp worker finished", taskName) return case <-ticker.C: err := dataprovider.UpdateTaskTimestamp(taskName) eventManagerLog(logger.LevelInfo, "updated timestamp for task %q, err: %v", taskName, err) } } }(task.Name) executeAsyncRulesActions([]dataprovider.EventRule{rule}, EventParams{Status: 1, updateStatusFromError: true}) } else { executeAsyncRulesActions([]dataprovider.EventRule{rule}, EventParams{Status: 1, updateStatusFromError: true}) } eventManagerLog(logger.LevelDebug, "execution for scheduled rule %q finished", j.ruleName) } // RunOnDemandRule executes actions for a rule with on-demand trigger func RunOnDemandRule(name string) error { eventManagerLog(logger.LevelDebug, "executing on demand rule %q", name) rule, err := dataprovider.EventRuleExists(name) if err != nil { eventManagerLog(logger.LevelDebug, "unable to load rule with name %q", name) return util.NewRecordNotFoundError(fmt.Sprintf("rule %q does not exist", name)) } if rule.Trigger != dataprovider.EventTriggerOnDemand { eventManagerLog(logger.LevelDebug, "cannot run rule %q as on demand, trigger: %d", name, rule.Trigger) return util.NewValidationError(fmt.Sprintf("rule %q is not defined as on-demand", name)) } if rule.Status != 1 { eventManagerLog(logger.LevelDebug, "on-demand rule %q is inactive", name) return util.NewValidationError(fmt.Sprintf("rule %q is inactive", name)) } if err := rule.CheckActionsConsistency(""); err != nil { eventManagerLog(logger.LevelError, "on-demand rule %q has incompatible actions: %v", name, err) return util.NewValidationError(fmt.Sprintf("rule %q has incosistent actions", name)) } eventManagerLog(logger.LevelDebug, "on-demand rule %q started", name) go executeAsyncRulesActions([]dataprovider.EventRule{rule}, EventParams{Status: 1, updateStatusFromError: true}) return nil } type zipWriterWrapper struct { Name string Entries map[string]bool Writer *zip.Writer } func eventManagerLog(level logger.LogLevel, format string, v ...any) { logger.Log(level, "eventmanager", "", format, v...) }