diff --git a/cmd/urbmd/main.go b/cmd/urbmd/main.go index 05ab503..79113e1 100644 --- a/cmd/urbmd/main.go +++ b/cmd/urbmd/main.go @@ -87,5 +87,8 @@ func main() { shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 15*time.Second) defer shutdownCancel() _ = server.Shutdown(shutdownCtx) + if err := svc.Wait(shutdownCtx); err != nil { + log.Warn("queue cleanup did not finish before shutdown timeout", "error", err) + } log.Info("urbm daemon stopped") } diff --git a/dist/urbm-2026.07.10.r007-x86_64-1.txz.sha256 b/dist/urbm-2026.07.10.r007-x86_64-1.txz.sha256 deleted file mode 100644 index 1761141..0000000 --- a/dist/urbm-2026.07.10.r007-x86_64-1.txz.sha256 +++ /dev/null @@ -1 +0,0 @@ -3e3f832d98656655d591c05ad4454d8480d5cfcdf60c03a2a498a6498e206beb urbm-2026.07.10.r007-x86_64-1.txz diff --git a/dist/urbm-2026.07.10.r007-x86_64-1.txz b/dist/urbm-2026.07.13.r001-x86_64-1.txz similarity index 56% rename from dist/urbm-2026.07.10.r007-x86_64-1.txz rename to dist/urbm-2026.07.13.r001-x86_64-1.txz index 2a7338d..ce4c67a 100644 Binary files a/dist/urbm-2026.07.10.r007-x86_64-1.txz and b/dist/urbm-2026.07.13.r001-x86_64-1.txz differ diff --git a/dist/urbm-2026.07.13.r001-x86_64-1.txz.sha256 b/dist/urbm-2026.07.13.r001-x86_64-1.txz.sha256 new file mode 100644 index 0000000..8944814 --- /dev/null +++ b/dist/urbm-2026.07.13.r001-x86_64-1.txz.sha256 @@ -0,0 +1 @@ +d2d572254dc7dd9d8a7489cbd7908fc80869efb506d4ecc01845f513c4cfb08b urbm-2026.07.13.r001-x86_64-1.txz diff --git a/dist/urbm.plg b/dist/urbm.plg index fabbe86..b2a3151 100644 --- a/dist/urbm.plg +++ b/dist/urbm.plg @@ -2,13 +2,19 @@ - + - + ]> +### 2026.07.13.r001 +- Harden restore task validation and make confirmed in-place restores target original paths safely. +- Serialize repository and managed-mount access, block unsafe force-unlock operations, and wait for task cleanup during daemon shutdown. +- Bound snapshot browser memory, stream Restic diffs, continuously drain command output, and reject symlink-overlapping Rsync paths. +- Activate persistent run logs and align package, WebGUI, and manifest versions during builds. + ### 2026.07.10.r007 - Serve snapshot browser folders as compact cached tree pages instead of sending the complete Restic file list to the browser. - Prevent large snapshots with hundreds of thousands of entries from failing with generic 500 responses during snapshot browsing. diff --git a/internal/api/server.go b/internal/api/server.go index 369276b..f749735 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -301,6 +301,10 @@ func decode(r *http.Request, target any) error { if err := decoder.Decode(target); err != nil { return fmt.Errorf("validation: invalid JSON: %w", err) } + var trailing any + if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) { + return errors.New("validation: JSON body must contain exactly one value and be at most 2 MiB") + } return nil } diff --git a/internal/model/model.go b/internal/model/model.go index 063fe8e..d05d645 100644 --- a/internal/model/model.go +++ b/internal/model/model.go @@ -138,38 +138,39 @@ type NotificationTarget struct { } type Run struct { - SchemaVersion int `json:"schemaVersion"` - ID string `json:"id"` - JobID string `json:"jobId,omitempty"` - TaskType string `json:"taskType"` - Status string `json:"status"` - Priority int `json:"priority"` - CreatedAt time.Time `json:"createdAt"` - StartedAt *time.Time `json:"startedAt,omitempty"` - FinishedAt *time.Time `json:"finishedAt,omitempty"` - SnapshotID string `json:"snapshotId,omitempty"` - Message string `json:"message,omitempty"` - ErrorCode string `json:"errorCode,omitempty"` - BytesAdded int64 `json:"bytesAdded,omitempty"` - FilesNew int64 `json:"filesNew,omitempty"` - FilesChanged int64 `json:"filesChanged,omitempty"` - BytesProcessed int64 `json:"bytesProcessed,omitempty"` - FilesProcessed int64 `json:"filesProcessed,omitempty"` - RetentionBefore int `json:"retentionBefore,omitempty"` - RetentionAfter int `json:"retentionAfter,omitempty"` - RetentionRemoved int `json:"retentionRemoved,omitempty"` - RepositoryBefore int64 `json:"repositoryBefore,omitempty"` - RepositoryAfter int64 `json:"repositoryAfter,omitempty"` - RepositoryFreed int64 `json:"repositoryFreed,omitempty"` - ProgressPercent float64 `json:"progressPercent,omitempty"` - ProgressBytes int64 `json:"progressBytes,omitempty"` - ProgressTotal int64 `json:"progressTotal,omitempty"` - ProgressFiles int64 `json:"progressFiles,omitempty"` - ProgressFileTotal int64 `json:"progressFileTotal,omitempty"` - BytesPerSecond float64 `json:"bytesPerSecond,omitempty"` - SecondsRemaining int64 `json:"secondsRemaining,omitempty"` - CurrentFile string `json:"currentFile,omitempty"` - LiveLog []string `json:"liveLog,omitempty"` + SchemaVersion int `json:"schemaVersion"` + ID string `json:"id"` + JobID string `json:"jobId,omitempty"` + TaskType string `json:"taskType"` + Status string `json:"status"` + Priority int `json:"priority"` + CreatedAt time.Time `json:"createdAt"` + StartedAt *time.Time `json:"startedAt,omitempty"` + FinishedAt *time.Time `json:"finishedAt,omitempty"` + SnapshotID string `json:"snapshotId,omitempty"` + Message string `json:"message,omitempty"` + ErrorCode string `json:"errorCode,omitempty"` + BytesAdded int64 `json:"bytesAdded,omitempty"` + FilesNew int64 `json:"filesNew,omitempty"` + FilesChanged int64 `json:"filesChanged,omitempty"` + BytesProcessed int64 `json:"bytesProcessed,omitempty"` + FilesProcessed int64 `json:"filesProcessed,omitempty"` + RetentionBefore int `json:"retentionBefore,omitempty"` + RetentionAfter int `json:"retentionAfter,omitempty"` + RetentionRemoved int `json:"retentionRemoved,omitempty"` + RepositoryBefore int64 `json:"repositoryBefore,omitempty"` + RepositoryAfter int64 `json:"repositoryAfter,omitempty"` + RepositoryFreed int64 `json:"repositoryFreed,omitempty"` + ProgressPercent float64 `json:"progressPercent,omitempty"` + ProgressBytes int64 `json:"progressBytes,omitempty"` + ProgressTotal int64 `json:"progressTotal,omitempty"` + ProgressFiles int64 `json:"progressFiles,omitempty"` + ProgressFileTotal int64 `json:"progressFileTotal,omitempty"` + BytesPerSecond float64 `json:"bytesPerSecond,omitempty"` + SecondsRemaining int64 `json:"secondsRemaining,omitempty"` + CurrentFile string `json:"currentFile,omitempty"` + LiveLog []string `json:"liveLog,omitempty"` + Restore *RestoreTask `json:"restore,omitempty"` } type RestoreTask struct { diff --git a/internal/model/validate.go b/internal/model/validate.go index 0cca932..4da3e09 100644 --- a/internal/model/validate.go +++ b/internal/model/validate.go @@ -10,12 +10,15 @@ import ( ) var idPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$`) +var snapshotPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9._:-]{0,255}$`) +var safePathPattern = regexp.MustCompile(`^/[a-zA-Z0-9_./-]+$`) func ValidateConfig(c Config) error { if c.SchemaVersion != SchemaVersion { return fmt.Errorf("unsupported schemaVersion %d", c.SchemaVersion) } repos := make(map[string]Repository, len(c.Repositories)) + mountPoints := make(map[string]string) for _, repo := range c.Repositories { if err := ValidateRepository(repo); err != nil { return fmt.Errorf("repository %q: %w", repo.ID, err) @@ -24,6 +27,13 @@ func ValidateConfig(c Config) error { return fmt.Errorf("duplicate repository id %q", repo.ID) } repos[repo.ID] = repo + if repo.Mount != nil && repo.Mount.Managed { + point := filepath.Clean(repo.Mount.MountPoint) + if previous, exists := mountPoints[point]; exists { + return fmt.Errorf("repositories %q and %q use the same managed mountPoint %q", previous, repo.ID, point) + } + mountPoints[point] = repo.ID + } } jobs := make(map[string]struct{}, len(c.Jobs)) for _, job := range c.Jobs { @@ -50,8 +60,52 @@ func ValidateConfig(c Config) error { } notifications[target.ID] = struct{}{} } - if !filepath.IsAbs(c.Settings.RestoreRoot) { - return errors.New("restoreRoot must be absolute") + restoreRoot := filepath.Clean(c.Settings.RestoreRoot) + if !filepath.IsAbs(restoreRoot) || !belowStorageRoot(restoreRoot) { + return errors.New("restoreRoot must be below /mnt/user, /mnt/disks, or /mnt/remotes") + } + if strings.TrimSpace(c.Settings.ResticPath) == "" || !filepath.IsAbs(c.Settings.ResticPath) { + return errors.New("resticPath must be an absolute path") + } + if strings.TrimSpace(c.Settings.RsyncPath) == "" || !filepath.IsAbs(c.Settings.RsyncPath) { + return errors.New("rsyncPath must be an absolute path") + } + if c.Settings.CatchUpWindowHrs < 0 || c.Settings.CatchUpWindowHrs > 24*31 { + return errors.New("catchUpWindowHours must be between 0 and 744") + } + if c.Settings.PersistentLogDir != "" { + logDir := filepath.Clean(c.Settings.PersistentLogDir) + if !filepath.IsAbs(logDir) || (!pathWithin(logDir, "/mnt/user") && !pathWithin(logDir, "/mnt/disks") && !pathWithin(logDir, "/mnt/remotes")) { + return errors.New("persistentLogDir must be below /mnt/user, /mnt/disks, or /mnt/remotes") + } + } + return nil +} + +func ValidateRestoreTask(task RestoreTask) error { + if !idPattern.MatchString(task.RepositoryID) { + return errors.New("restore repositoryId is invalid") + } + if err := ValidateSnapshotID(task.SnapshotID); err != nil { + return err + } + if strings.ContainsRune(task.Target, '\x00') { + return errors.New("restore target contains a forbidden control character") + } + if task.InPlace && len(task.Includes) == 0 { + return errors.New("in-place restore requires at least one included path") + } + for _, include := range task.Includes { + if strings.ContainsAny(include, "\x00\r\n") || !filepath.IsAbs(include) || filepath.Clean(include) == "/" { + return fmt.Errorf("invalid restore include %q", include) + } + } + return nil +} + +func ValidateSnapshotID(value string) error { + if !snapshotPattern.MatchString(value) { + return errors.New("restore snapshotId is invalid") } return nil } @@ -178,6 +232,15 @@ func pathWithin(path, root string) bool { return err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) } +func belowStorageRoot(path string) bool { + for _, root := range []string{"/mnt/user", "/mnt/disks", "/mnt/remotes"} { + if filepath.Clean(path) != root && pathWithin(path, root) { + return true + } + } + return false +} + func ValidateRepository(r Repository) error { if r.SchemaVersion != SchemaVersion || !idPattern.MatchString(r.ID) { return errors.New("invalid schemaVersion or id") @@ -200,11 +263,14 @@ func ValidateRepository(r Repository) error { if r.Mount.Remote == "" || !filepath.IsAbs(r.Mount.MountPoint) { return errors.New("managed mount requires remote and absolute mountPoint") } + if strings.HasPrefix(r.Mount.Remote, "-") || strings.ContainsAny(r.Mount.Remote, "\x00\r\n") { + return errors.New("managed mount remote contains forbidden characters") + } } if r.Type == RepositorySFTP && r.CredentialRef != "" { knownHosts := r.Options["knownHostsPath"] - if !filepath.IsAbs(knownHosts) || strings.ContainsAny(knownHosts, " \t\r\n") { - return errors.New("SFTP key authentication requires an absolute knownHostsPath without whitespace") + if !safePathPattern.MatchString(knownHosts) || filepath.Clean(knownHosts) != knownHosts { + return errors.New("SFTP key authentication requires a clean absolute knownHostsPath containing only letters, digits, dot, underscore, slash, and hyphen") } } return nil diff --git a/internal/model/validate_test.go b/internal/model/validate_test.go index 89e6643..afcbb51 100644 --- a/internal/model/validate_test.go +++ b/internal/model/validate_test.go @@ -118,3 +118,39 @@ func TestRsyncJobRejectsOverlappingPaths(t *testing.T) { t.Fatal("rsync target inside source accepted") } } + +func TestValidateRestoreTaskRejectsDelimiterInjection(t *testing.T) { + task := RestoreTask{RepositoryID: "repo", SnapshotID: "abc123\x00/", Target: "/mnt/user/restore", Includes: []string{"/mnt/user/data"}} + if err := ValidateRestoreTask(task); err == nil { + t.Fatal("NUL-delimited snapshot injection accepted") + } + task.SnapshotID = "abc123" + task.InPlace = true + task.Includes = nil + if err := ValidateRestoreTask(task); err == nil { + t.Fatal("in-place restore without an include accepted") + } +} + +func TestValidateConfigRejectsDuplicateManagedMountPoint(t *testing.T) { + c := DefaultConfig() + c.Repositories = []Repository{ + {SchemaVersion: 1, ID: "one", Name: "One", Type: RepositorySMB, Location: "/mnt/remotes/shared", PasswordRef: "one-password", Mount: &MountConfig{Managed: true, Remote: "//one/share", MountPoint: "/mnt/remotes/shared"}}, + {SchemaVersion: 1, ID: "two", Name: "Two", Type: RepositoryNFS, Location: "/mnt/remotes/shared", PasswordRef: "two-password", Mount: &MountConfig{Managed: true, Remote: "two:/share", MountPoint: "/mnt/remotes/shared"}}, + } + if err := ValidateConfig(c); err == nil { + t.Fatal("duplicate managed mount point accepted") + } +} + +func TestValidateConfigRejectsUnsafeRestoreRoot(t *testing.T) { + c := DefaultConfig() + c.Settings.RestoreRoot = "/etc/urbm-restores" + if err := ValidateConfig(c); err == nil { + t.Fatal("restoreRoot outside Unraid storage roots accepted") + } + c.Settings.RestoreRoot = "/mnt/user/urbm-restores" + if err := ValidateConfig(c); err != nil { + t.Fatalf("safe restoreRoot rejected: %v", err) + } +} diff --git a/internal/platform/mounts.go b/internal/platform/mounts.go index 356b5b6..099e9e7 100644 --- a/internal/platform/mounts.go +++ b/internal/platform/mounts.go @@ -8,6 +8,7 @@ import ( "os/exec" "path/filepath" "strings" + "time" "git.casaderoll.de/michael/urbm/internal/model" ) @@ -66,13 +67,16 @@ func (m *MountManager) Prepare(ctx context.Context, repo model.Repository) (Moun result.CredentialFile = f.Name() if err := f.Chmod(0600); err != nil { f.Close() + os.Remove(result.CredentialFile) return result, err } if _, err := f.WriteString(credential); err != nil { f.Close() + os.Remove(result.CredentialFile) return result, err } if err := f.Close(); err != nil { + os.Remove(result.CredentialFile) return result, err } options = append(options, "credentials="+result.CredentialFile) @@ -97,5 +101,10 @@ func (m *MountManager) Cleanup(ctx context.Context, mounted MountedRepository) e if !mounted.Mounted { return nil } + if _, hasDeadline := ctx.Deadline(); !hasDeadline { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, 5*time.Second) + defer cancel() + } return exec.CommandContext(ctx, "umount", mounted.Repository.Location).Run() } diff --git a/internal/platform/paths.go b/internal/platform/paths.go index 5ca1fcc..d66a79f 100644 --- a/internal/platform/paths.go +++ b/internal/platform/paths.go @@ -5,9 +5,42 @@ import ( "os" "path/filepath" "strings" + + "git.casaderoll.de/michael/urbm/internal/model" ) +func ValidateRsyncPaths(job model.Job) error { + target, err := filepath.EvalSymlinks(job.Rsync.Target) + if err != nil { + return errors.New("validation: resolve rsync target: " + err.Error()) + } + for _, source := range job.Sources { + resolved, err := filepath.EvalSymlinks(source.Path) + if err != nil { + return errors.New("validation: resolve rsync source: " + err.Error()) + } + if pathsContainEachOther(resolved, target) { + return errors.New("validation: resolved rsync source and target must not contain each other") + } + } + return nil +} + +func pathsContainEachOther(first, second string) bool { + within := func(path, root string) bool { + rel, err := filepath.Rel(filepath.Clean(root), filepath.Clean(path)) + return err == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) + } + return within(first, second) || within(second, first) +} + func ValidateRestoreTarget(target, restoreRoot string, inPlace, confirmed bool) (string, error) { + if inPlace { + if !confirmed { + return "", errors.New("validation: in-place restore requires explicit confirmation") + } + return string(os.PathSeparator), nil + } if !filepath.IsAbs(target) { return "", errors.New("validation: restore target must be absolute") } @@ -19,20 +52,13 @@ func ValidateRestoreTarget(target, restoreRoot string, inPlace, confirmed bool) if err := validateStagingRestoreTarget(clean, restoreRoot); err != nil { return "", err } - } else if !confirmed { - return "", errors.New("validation: in-place restore requires explicit confirmation") - } - if inPlace { - anchor := string(os.PathSeparator) + strings.Split(strings.TrimPrefix(clean, string(os.PathSeparator)), string(os.PathSeparator))[0] - if err := rejectSymlinks(anchor, clean); err != nil { - return "", err - } } return clean, nil } func validateStagingRestoreTarget(target, restoreRoot string) error { - allowedRoots := []string{filepath.Clean(restoreRoot), "/mnt/user", "/mnt/disks", "/mnt/remotes"} + _ = restoreRoot // Configuration validation keeps this below one of the storage roots. + allowedRoots := []string{"/mnt/user", "/mnt/disks", "/mnt/remotes"} for _, root := range allowedRoots { if root == "." || root == "" { continue @@ -45,6 +71,13 @@ func validateStagingRestoreTarget(target, restoreRoot string) error { } func rejectSymlinks(root, target string) error { + if info, err := os.Lstat(root); err == nil { + if info.Mode()&os.ModeSymlink != 0 { + return errors.New("validation: restore target traverses a symlink") + } + } else if !errors.Is(err, os.ErrNotExist) { + return err + } relative, err := filepath.Rel(root, target) if err != nil { return err diff --git a/internal/platform/paths_test.go b/internal/platform/paths_test.go index 152fd4a..eb88d1c 100644 --- a/internal/platform/paths_test.go +++ b/internal/platform/paths_test.go @@ -4,10 +4,12 @@ import ( "os" "path/filepath" "testing" + + "git.casaderoll.de/michael/urbm/internal/model" ) func TestValidateRestoreTarget(t *testing.T) { - root := t.TempDir() + root := "/mnt/user/urbm-restores" target := filepath.Join(root, "task") if got, err := ValidateRestoreTarget(target, root, false, false); err != nil || got != target { t.Fatalf("staging target rejected: %q %v", got, err) @@ -18,14 +20,30 @@ func TestValidateRestoreTarget(t *testing.T) { if _, err := ValidateRestoreTarget("/home/root/restore", root, false, false); err == nil { t.Fatal("staging target outside allowed restore roots accepted") } - if _, err := ValidateRestoreTarget("/etc", root, true, true); err == nil { - t.Fatal("protected target accepted") + if got, err := ValidateRestoreTarget("/mnt/user/data", root, true, true); err != nil || got != "/" { + t.Fatalf("confirmed in-place target = %q, %v", got, err) } if _, err := ValidateRestoreTarget("/mnt/user/data", root, true, false); err == nil { t.Fatal("unconfirmed in-place restore accepted") } } +func TestValidateRsyncPathsRejectsSymlinkOverlap(t *testing.T) { + root := t.TempDir() + source := filepath.Join(root, "source") + if err := os.Mkdir(source, 0755); err != nil { + t.Fatal(err) + } + target := filepath.Join(root, "target-link") + if err := os.Symlink(source, target); err != nil { + t.Fatal(err) + } + job := model.Job{Sources: []model.Source{{Path: source}}, Rsync: model.RsyncOptions{Target: target}} + if err := ValidateRsyncPaths(job); err == nil { + t.Fatal("symlinked rsync target overlapping the source was accepted") + } +} + func TestValidateRestoreTargetRejectsSymlink(t *testing.T) { root := t.TempDir() real := filepath.Join(root, "real") @@ -36,7 +54,22 @@ func TestValidateRestoreTargetRejectsSymlink(t *testing.T) { if err := os.Symlink(real, link); err != nil { t.Fatal(err) } - if _, err := ValidateRestoreTarget(filepath.Join(link, "task"), root, false, false); err == nil { + if err := rejectSymlinks(root, filepath.Join(link, "task")); err == nil { t.Fatal("symlink traversal accepted") } } + +func TestRejectSymlinksChecksRootItself(t *testing.T) { + base := t.TempDir() + real := filepath.Join(base, "real") + link := filepath.Join(base, "root-link") + if err := os.Mkdir(real, 0755); err != nil { + t.Fatal(err) + } + if err := os.Symlink(real, link); err != nil { + t.Fatal(err) + } + if err := rejectSymlinks(link, filepath.Join(link, "task")); err == nil { + t.Fatal("symlinked restore root accepted") + } +} diff --git a/internal/platform/workloads.go b/internal/platform/workloads.go index 908ef0a..40064fb 100644 --- a/internal/platform/workloads.go +++ b/internal/platform/workloads.go @@ -203,6 +203,11 @@ func (w *WorkloadManager) FlashDevice(ctx context.Context) (string, error) { } func (w *WorkloadManager) Cleanup(ctx context.Context, prepared Prepared) error { + if _, hasDeadline := ctx.Deadline(); !hasDeadline { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, 5*time.Second) + defer cancel() + } var first error for i := len(prepared.Stopped) - 1; i >= 0; i-- { source := prepared.Stopped[i] diff --git a/internal/queue/queue.go b/internal/queue/queue.go index e6bce4c..492cac1 100644 --- a/internal/queue/queue.go +++ b/internal/queue/queue.go @@ -25,10 +25,12 @@ type Queue struct { stopping bool handler Handler onChange OnChange + done chan struct{} + doneOnce sync.Once } func New(handler Handler, onChange OnChange) *Queue { - q := &Queue{handler: handler, onChange: onChange} + q := &Queue{handler: handler, onChange: onChange, done: make(chan struct{})} q.cond = sync.NewCond(&q.mu) return q } @@ -71,6 +73,7 @@ func (q *Queue) Enqueue(run model.Run) error { } func (q *Queue) Run(ctx context.Context) { + defer q.doneOnce.Do(func() { close(q.done) }) for { q.mu.Lock() for len(q.pending) == 0 && !q.stopping { @@ -104,6 +107,15 @@ func (q *Queue) Run(ctx context.Context) { } } +func (q *Queue) Wait(ctx context.Context) error { + select { + case <-q.done: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + func (q *Queue) Cancel(id string) bool { q.mu.Lock() defer q.mu.Unlock() diff --git a/internal/queue/queue_test.go b/internal/queue/queue_test.go index 505538d..5c3de88 100644 --- a/internal/queue/queue_test.go +++ b/internal/queue/queue_test.go @@ -90,3 +90,33 @@ func TestPauseAndResumeActiveRun(t *testing.T) { } close(release) } + +func TestWaitDoesNotReturnBeforeCancelledHandlerCleanup(t *testing.T) { + started := make(chan struct{}) + cleanup := make(chan struct{}) + q := New(func(ctx context.Context, run model.Run) model.Run { + close(started) + <-ctx.Done() + close(cleanup) + run.Status = "cancelled" + return run + }, nil) + ctx, cancel := context.WithCancel(context.Background()) + go q.Run(ctx) + if err := q.Enqueue(model.Run{ID: "run", JobID: "job"}); err != nil { + t.Fatal(err) + } + <-started + q.Stop() + waitCtx, waitCancel := context.WithTimeout(context.Background(), time.Second) + defer waitCancel() + if err := q.Wait(waitCtx); err != nil { + t.Fatal(err) + } + select { + case <-cleanup: + default: + t.Fatal("queue wait returned before handler cleanup") + } + cancel() +} diff --git a/internal/restic/restic.go b/internal/restic/restic.go index 2f59c05..2d32163 100644 --- a/internal/restic/restic.go +++ b/internal/restic/restic.go @@ -30,6 +30,18 @@ type Runner struct { processes map[string]*os.Process } +func (r *Runner) SetBinary(path string) { + r.mu.Lock() + r.Binary = path + r.mu.Unlock() +} + +func (r *Runner) binary() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.Binary +} + type runIDKey struct{} func WithRunID(ctx context.Context, runID string) context.Context { @@ -270,28 +282,32 @@ func decodeStats(output []byte) (Stats, error) { } func (r *Runner) List(ctx context.Context, repo model.Repository, snapshot, path string) ([]map[string]any, error) { + items := []map[string]any{} + err := r.WalkList(ctx, repo, snapshot, path, func(item map[string]any) { items = append(items, item) }) + return items, err +} + +func (r *Runner) WalkList(ctx context.Context, repo model.Repository, snapshot, path string, onItem func(map[string]any)) error { args := []string{"ls", snapshot, "--json"} if path != "" { args = append(args, path) } - items := []map[string]any{} - if err := r.run(ctx, repo, args, func(line []byte) { + return r.run(ctx, repo, args, func(line []byte) { var item map[string]any if json.Unmarshal(line, &item) == nil && item["struct_type"] == "node" { - items = append(items, item) + onItem(item) } - }, nil); err != nil { - return nil, err - } - return items, nil + }, nil) } func (r *Runner) Diff(ctx context.Context, repo model.Repository, before, after string, limit int) (DiffResult, error) { - var output []byte - if err := r.run(ctx, repo, []string{"diff", "--metadata", before, after}, nil, &output); err != nil { + result := DiffResult{} + if err := r.run(ctx, repo, []string{"diff", "--metadata", before, after}, func(line []byte) { + appendDiffLine(&result, string(line), limit) + }, nil); err != nil { return DiffResult{}, err } - return parseDiffOutput(output, limit), nil + return result, nil } func parseDiffOutput(output []byte, limit int) DiffResult { @@ -302,29 +318,35 @@ func parseDiffOutput(output []byte, limit int) DiffResult { scanner := bufio.NewScanner(bytes.NewReader(output)) scanner.Buffer(make([]byte, 0, 64*1024), 2*1024*1024) for scanner.Scan() { - line := scanner.Text() - if len(line) < 2 { - continue - } - kind := diffKind(line[0]) - if kind == "" { - continue - } - path := strings.TrimSpace(line[1:]) - if path == "" || !strings.HasPrefix(path, "/") { - continue - } - result.Total++ - if len(result.Entries) == limit { - copy(result.Entries, result.Entries[1:]) - result.Entries[len(result.Entries)-1] = DiffEntry{Kind: kind, Path: path} - continue - } - result.Entries = append(result.Entries, DiffEntry{Kind: kind, Path: path}) + appendDiffLine(&result, scanner.Text(), limit) } return result } +func appendDiffLine(result *DiffResult, line string, limit int) { + if limit <= 0 { + limit = 50 + } + if len(line) < 2 { + return + } + kind := diffKind(line[0]) + if kind == "" { + return + } + path := strings.TrimSpace(line[1:]) + if path == "" || !strings.HasPrefix(path, "/") { + return + } + result.Total++ + if len(result.Entries) == limit { + copy(result.Entries, result.Entries[1:]) + result.Entries[len(result.Entries)-1] = DiffEntry{Kind: kind, Path: path} + return + } + result.Entries = append(result.Entries, DiffEntry{Kind: kind, Path: path}) +} + func diffKind(marker byte) string { switch marker { case '+': @@ -404,7 +426,7 @@ func (r *Runner) runWithInputEnvPassword(ctx context.Context, repo model.Reposit globalArgs = append(globalArgs, "-o", "sftp.command="+command) } fullArgs := append(globalArgs, args...) - cmd := exec.CommandContext(ctx, r.Binary, fullArgs...) + cmd := exec.CommandContext(ctx, r.binary(), fullArgs...) cmd.Stdin = input cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} cmd.Cancel = func() error { @@ -451,9 +473,11 @@ func (r *Runner) runWithInputEnvPassword(ctx context.Context, repo model.Reposit } close(done) }() - errBytes, _ := io.ReadAll(io.LimitReader(stderr, 1024*1024)) + stderrDone := make(chan []byte, 1) + go func() { stderrDone <- drainLimited(stderr, 1024*1024) }() err = cmd.Wait() <-done + errBytes := <-stderrDone if capture != nil { *capture = captured } @@ -470,6 +494,24 @@ func (r *Runner) runWithInputEnvPassword(ctx context.Context, repo model.Reposit return nil } +func drainLimited(reader io.Reader, limit int) []byte { + result := make([]byte, 0, limit) + buffer := make([]byte, 32*1024) + for { + count, err := reader.Read(buffer) + if count > 0 && len(result) < limit { + keep := count + if keep > limit-len(result) { + keep = limit - len(result) + } + result = append(result, buffer[:keep]...) + } + if err != nil { + return result + } + } +} + func (r *Runner) writePasswordFile(pattern, password string) (string, func(), error) { secretDir := filepath.Join(r.RuntimeDir, "secrets") if err := os.MkdirAll(secretDir, 0700); err != nil { diff --git a/internal/rsync/rsync.go b/internal/rsync/rsync.go index 3bc0a1a..df39017 100644 --- a/internal/rsync/rsync.go +++ b/internal/rsync/rsync.go @@ -23,6 +23,18 @@ type Runner struct { processes map[string]*os.Process } +func (r *Runner) SetBinary(path string) { + r.mu.Lock() + r.Binary = path + r.mu.Unlock() +} + +func (r *Runner) binary() string { + r.mu.Lock() + defer r.mu.Unlock() + return r.Binary +} + type Progress struct { Percent float64 Bytes int64 @@ -43,7 +55,7 @@ var ( func (r *Runner) Run(ctx context.Context, runID string, job model.Job, callback func(Progress)) (Summary, error) { args := Arguments(job) - cmd := exec.CommandContext(ctx, r.Binary, args...) + cmd := exec.CommandContext(ctx, r.binary(), args...) cmd.Env = append(os.Environ(), "LC_ALL=C", "LANG=C") cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} cmd.Cancel = func() error { @@ -92,9 +104,11 @@ func (r *Runner) Run(ctx context.Context, runID string, job model.Job, callback }) close(done) }() - errBytes, _ := io.ReadAll(io.LimitReader(stderr, 1024*1024)) + stderrDone := make(chan []byte, 1) + go func() { stderrDone <- drainLimited(stderr, 1024*1024) }() err = cmd.Wait() <-done + errBytes := <-stderrDone if err != nil { if errors.Is(ctx.Err(), context.Canceled) { return Summary{}, fmt.Errorf("cancelled: %w", ctx.Err()) @@ -117,7 +131,7 @@ func (r *Runner) Estimate(ctx context.Context, runID string, job model.Job) (Sum args[index] = "--out-format=" } } - cmd := exec.CommandContext(ctx, r.Binary, args...) + cmd := exec.CommandContext(ctx, r.binary(), args...) cmd.Env = append(os.Environ(), "LC_ALL=C", "LANG=C") cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} cmd.Cancel = func() error { @@ -158,9 +172,11 @@ func (r *Runner) Estimate(ctx context.Context, runID string, job model.Job) (Sum }) close(done) }() - errBytes, _ := io.ReadAll(io.LimitReader(stderr, 1024*1024)) + stderrDone := make(chan []byte, 1) + go func() { stderrDone <- drainLimited(stderr, 1024*1024) }() err = cmd.Wait() <-done + errBytes := <-stderrDone if err != nil { if errors.Is(ctx.Err(), context.Canceled) { return Summary{}, fmt.Errorf("cancelled: %w", ctx.Err()) @@ -258,6 +274,24 @@ func scanOutput(reader io.Reader, callback func(string)) { } } +func drainLimited(reader io.Reader, limit int) []byte { + result := make([]byte, 0, limit) + buffer := make([]byte, 32*1024) + for { + count, err := reader.Read(buffer) + if count > 0 && len(result) < limit { + keep := count + if keep > limit-len(result) { + keep = limit - len(result) + } + result = append(result, buffer[:keep]...) + } + if err != nil { + return result + } + } +} + func splitLinesAndCarriageReturns(data []byte, atEOF bool) (advance int, token []byte, err error) { for i, value := range data { if value == '\n' || value == '\r' { diff --git a/internal/scheduler/cron.go b/internal/scheduler/cron.go index 0a27024..65da394 100644 --- a/internal/scheduler/cron.go +++ b/internal/scheduler/cron.go @@ -7,7 +7,10 @@ import ( "time" ) -type Expression struct{ fields [5]field } +type Expression struct { + fields [5]field + restricted [5]bool +} type field map[int]bool var bounds = [5][2]int{{0, 59}, {0, 23}, {1, 31}, {1, 12}, {0, 6}} @@ -24,6 +27,7 @@ func Parse(spec string) (Expression, error) { return Expression{}, fmt.Errorf("cron field %d: %w", i+1, err) } expression.fields[i] = parsed + expression.restricted[i] = part != "*" } return expression, nil } @@ -76,11 +80,19 @@ func parseField(value string, min, max int) (field, error) { func (e Expression) Matches(t time.Time) bool { values := [5]int{t.Minute(), t.Hour(), t.Day(), int(t.Month()), int(t.Weekday())} for i, value := range values { + if i == 2 || i == 4 { + continue + } if !e.fields[i][value] { return false } } - return true + dayOfMonth := e.fields[2][values[2]] + dayOfWeek := e.fields[4][values[4]] + if e.restricted[2] && e.restricted[4] { + return dayOfMonth || dayOfWeek + } + return dayOfMonth && dayOfWeek } func (e Expression) Next(after time.Time) time.Time { diff --git a/internal/scheduler/cron_test.go b/internal/scheduler/cron_test.go index a2ade8b..21c6d87 100644 --- a/internal/scheduler/cron_test.go +++ b/internal/scheduler/cron_test.go @@ -43,3 +43,13 @@ func TestPreviousWithinCatchUpWindow(t *testing.T) { t.Fatalf("unexpected catch-up match %v", got) } } + +func TestCronUsesOrForRestrictedMonthDayAndWeekday(t *testing.T) { + expr, err := Parse("0 2 1 * 1") + if err != nil { + t.Fatal(err) + } + if !expr.Matches(time.Date(2026, 6, 8, 2, 0, 0, 0, time.UTC)) { + t.Fatal("restricted day-of-month and weekday were treated as AND") + } +} diff --git a/internal/service/safety_test.go b/internal/service/safety_test.go new file mode 100644 index 0000000..dbae671 --- /dev/null +++ b/internal/service/safety_test.go @@ -0,0 +1,67 @@ +package service + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "git.casaderoll.de/michael/urbm/internal/model" +) + +func TestRepositoryLockHonorsContextCancellation(t *testing.T) { + s := &Service{repositoryLocks: map[string]chan struct{}{}} + repo := model.Repository{Type: model.RepositoryLocal, Location: "/repo"} + unlock, err := s.lockRepository(context.Background(), repo) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := s.lockRepository(ctx, repo); err == nil { + t.Fatal("concurrent repository access ignored the request deadline") + } + unlock() +} + +func TestPersistentRunLogsAreWrittenAndPruned(t *testing.T) { + dir := t.TempDir() + config := model.DefaultConfig() + config.Settings.PersistentLogDir = dir + s := &Service{config: config} + finished := time.Now() + run := model.Run{ID: "run-1", FinishedAt: &finished, LiveLog: []string{"first", "second"}} + if err := s.persistRunLogs([]model.Run{run}); err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, "urbm-run-1.log") + if contents, err := os.ReadFile(path); err != nil || string(contents) != "first\nsecond\n" { + t.Fatalf("persistent log = %q, %v", contents, err) + } + if err := s.persistRunLogs(nil); err != nil { + t.Fatal(err) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("cleared run log still exists: %v", err) + } +} + +func TestQueuePersistenceDoesNotBlockWhenWriterIsBusy(t *testing.T) { + s := &Service{persistRuns: make(chan []model.Run, 1)} + s.persistRuns <- []model.Run{{ID: "old"}} + done := make(chan struct{}) + go func() { + s.queuePersistence([]model.Run{{ID: "latest"}}) + close(done) + }() + select { + case <-done: + case <-time.After(100 * time.Millisecond): + t.Fatal("queue persistence blocked the queue callback") + } + runs := <-s.persistRuns + if len(runs) != 1 || runs[0].ID != "latest" { + t.Fatalf("queued persistence state = %#v", runs) + } +} diff --git a/internal/service/service.go b/internal/service/service.go index 5039e8a..dc24d7b 100644 --- a/internal/service/service.go +++ b/internal/service/service.go @@ -33,19 +33,25 @@ type SecretStore interface { } type Service struct { - store *store.Store - secrets SecretStore - restic *restic.Runner - rsync *rsync.Runner - mounts *platform.MountManager - workloads *platform.WorkloadManager - notifier *notify.Sender - log *slog.Logger - queue *queue.Queue - mu sync.RWMutex - config model.Config - snapshotMu sync.Mutex - snapshotTree map[string]snapshotTreeCache + store *store.Store + secrets SecretStore + restic *restic.Runner + rsync *rsync.Runner + mounts *platform.MountManager + workloads *platform.WorkloadManager + notifier *notify.Sender + log *slog.Logger + queue *queue.Queue + mu sync.RWMutex + config model.Config + snapshotMu sync.Mutex + snapshotTree map[string]snapshotTreeCache + repositoryMu sync.Mutex + repositoryLocks map[string]chan struct{} + persistRuns chan []model.Run + persistStop chan struct{} + persistDone chan struct{} + persistStopOnce sync.Once } type SnapshotBrowserNode struct { @@ -78,27 +84,46 @@ type snapshotTreeNode struct { children map[string]*snapshotTreeNode } +type snapshotTreeBuilder struct { + root *snapshotTreeNode + nodes map[string]*snapshotTreeNode + total int +} + func New(st *store.Store, secrets SecretStore, rr *restic.Runner, rs *rsync.Runner, mounts *platform.MountManager, workloads *platform.WorkloadManager, notifier *notify.Sender, log *slog.Logger) (*Service, error) { config, err := st.LoadConfig() if err != nil { return nil, err } - s := &Service{store: st, secrets: secrets, restic: rr, rsync: rs, mounts: mounts, workloads: workloads, notifier: notifier, log: log, config: config} - s.queue = queue.New(s.execute, func(runs []model.Run) { - if err := st.SaveRuns(runs); err != nil { - log.Error("save run history", "error", err) - } - }) + s := &Service{ + store: st, secrets: secrets, restic: rr, rsync: rs, mounts: mounts, workloads: workloads, notifier: notifier, log: log, config: config, + repositoryLocks: map[string]chan struct{}{}, persistRuns: make(chan []model.Run, 1), persistStop: make(chan struct{}), persistDone: make(chan struct{}), + } + go s.runPersistence() + s.queue = queue.New(s.execute, s.queuePersistence) if runs, err := st.LoadRuns(); err == nil { s.queue.RestoreHistory(runs) } + s.queuePersistence(s.queue.Snapshot()) return s, nil } func (s *Service) RunQueue(ctx context.Context) { s.queue.Run(ctx) } func (s *Service) Stop() { s.queue.Stop() } -func (s *Service) Config() model.Config { s.mu.RLock(); defer s.mu.RUnlock(); return s.config } -func (s *Service) Runs() []model.Run { return s.queue.Snapshot() } +func (s *Service) Wait(ctx context.Context) error { + if err := s.queue.Wait(ctx); err != nil { + return err + } + s.persistStopOnce.Do(func() { close(s.persistStop) }) + select { + case <-s.persistDone: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} +func (s *Service) Config() model.Config { s.mu.RLock(); defer s.mu.RUnlock(); return s.config } +func (s *Service) Runs() []model.Run { return s.queue.Snapshot() } func (s *Service) Logs() map[string]any { return map[string]any{ "daemon": tailFile("/var/log/urbm.log", 1024*1024, 1000), @@ -117,6 +142,123 @@ func runsWithLogs(runs []model.Run, limit int) []model.Run { return result } +func (s *Service) queuePersistence(runs []model.Run) { + select { + case s.persistRuns <- runs: + return + default: + } + select { + case <-s.persistRuns: + default: + } + select { + case s.persistRuns <- runs: + default: + } +} + +func (s *Service) runPersistence() { + defer close(s.persistDone) + for { + select { + case runs := <-s.persistRuns: + s.saveRunState(runs) + case <-s.persistStop: + var latest []model.Run + for { + select { + case latest = <-s.persistRuns: + default: + if latest != nil { + s.saveRunState(latest) + } + return + } + } + } + } +} + +func (s *Service) saveRunState(runs []model.Run) { + if err := s.store.SaveRuns(runs); err != nil { + s.log.Error("save run history", "error", err) + } + if err := s.persistRunLogs(runs); err != nil { + s.log.Error("save persistent run logs", "error", err) + } +} + +func (s *Service) persistRunLogs(runs []model.Run) error { + dir := strings.TrimSpace(s.Config().Settings.PersistentLogDir) + if dir == "" { + return nil + } + if err := os.MkdirAll(dir, 0700); err != nil { + return err + } + wanted := make(map[string]struct{}) + for _, run := range runs { + if run.FinishedAt == nil || len(run.LiveLog) == 0 { + continue + } + name := "urbm-" + safeLogName(run.ID) + ".log" + wanted[name] = struct{}{} + path := filepath.Join(dir, name) + if _, err := os.Stat(path); err == nil { + continue + } + if err := writeTextAtomic(path, strings.Join(run.LiveLog, "\n")+"\n"); err != nil { + return err + } + } + entries, err := os.ReadDir(dir) + if err != nil { + return err + } + for _, entry := range entries { + if entry.IsDir() || !strings.HasPrefix(entry.Name(), "urbm-") || !strings.HasSuffix(entry.Name(), ".log") { + continue + } + if _, ok := wanted[entry.Name()]; !ok { + if err := os.Remove(filepath.Join(dir, entry.Name())); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + } + } + return nil +} + +func safeLogName(value string) string { + return strings.Map(func(char rune) rune { + if char >= 'a' && char <= 'z' || char >= 'A' && char <= 'Z' || char >= '0' && char <= '9' || char == '-' || char == '_' { + return char + } + return '_' + }, value) +} + +func writeTextAtomic(path, value string) error { + tmp, err := os.CreateTemp(filepath.Dir(path), ".urbm-log-*") + if err != nil { + return err + } + tmpPath := tmp.Name() + defer os.Remove(tmpPath) + if err := tmp.Chmod(0600); err != nil { + tmp.Close() + return err + } + if _, err := tmp.WriteString(value); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmpPath, path) +} + func tailFile(path string, maxBytes int64, maxLines int) map[string]any { data, err := os.ReadFile(path) if err != nil { @@ -193,8 +335,8 @@ func (s *Service) SaveConfig(config model.Config) error { s.mu.Lock() s.config = config s.mu.Unlock() - s.restic.Binary = config.Settings.ResticPath - s.rsync.Binary = config.Settings.RsyncPath + s.restic.SetBinary(config.Settings.ResticPath) + s.rsync.SetBinary(config.Settings.RsyncPath) return nil } @@ -215,6 +357,11 @@ func (s *Service) ChangeRepositoryPassword(ctx context.Context, repoID, newPassw if !ok { return errors.New("validation: unknown repository") } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return err + } + defer unlock() oldPassword, err := s.secrets.Get(repo.PasswordRef) if err != nil { return fmt.Errorf("authentication: load current repository password: %w", err) @@ -260,6 +407,11 @@ func (s *Service) AdoptRepositoryPassword(ctx context.Context, repoID, password if !ok { return errors.New("validation: unknown repository") } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return err + } + defer unlock() mounted, err := s.mounts.Prepare(ctx, repo) if err != nil { return err @@ -313,18 +465,21 @@ func (s *Service) EnqueueMaintenance(repoID, action string) (model.Run, error) { } func (s *Service) EnqueueRestore(task model.RestoreTask) (model.Run, error) { + if err := model.ValidateRestoreTask(task); err != nil { + return model.Run{}, fmt.Errorf("validation: %w", err) + } + if _, ok := s.repository(task.RepositoryID); !ok { + return model.Run{}, errors.New("validation: unknown repository") + } config := s.Config() target, err := platform.ValidateRestoreTarget(task.Target, config.Settings.RestoreRoot, task.InPlace, task.Confirmed) if err != nil { return model.Run{}, err } task.Target = target - if task.ID == "" { - task.ID = newID() - } + task.ID = newID() task.SchemaVersion = model.SchemaVersion - encoded := strings.Join(append([]string{task.RepositoryID, task.SnapshotID, task.Target, fmt.Sprint(task.InPlace)}, task.Includes...), "\x00") - run := model.Run{SchemaVersion: model.SchemaVersion, ID: task.ID, JobID: encoded, TaskType: "restore", Priority: 10, CreatedAt: time.Now().UTC()} + run := model.Run{SchemaVersion: model.SchemaVersion, ID: task.ID, JobID: task.RepositoryID, TaskType: "restore", Priority: 10, CreatedAt: time.Now().UTC(), Restore: &task} return run, s.queue.Enqueue(run) } @@ -375,6 +530,11 @@ func (s *Service) Snapshots(ctx context.Context, repoID string) ([]model.Snapsho if !ok { return nil, errors.New("validation: unknown repository") } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return nil, err + } + defer unlock() mounted, err := s.mounts.Prepare(ctx, repo) if err != nil { return nil, err @@ -388,6 +548,12 @@ func (s *Service) RepositoryStats(ctx context.Context) []model.RepositoryStats { result := make([]model.RepositoryStats, 0, len(config.Repositories)) for _, repo := range config.Repositories { item := model.RepositoryStats{RepositoryID: repo.ID, RepositoryName: repo.Name} + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + item.Error = err.Error() + result = append(result, item) + break + } mounted, err := s.mounts.Prepare(ctx, repo) if err == nil { var stored, files restic.Stats @@ -406,6 +572,7 @@ func (s *Service) RepositoryStats(ctx context.Context) []model.RepositoryStats { err = cleanupErr } } + unlock() if err != nil { item.Error = err.Error() } @@ -420,6 +587,9 @@ func (s *Service) RepositoryStats(ctx context.Context) []model.RepositoryStats { func (s *Service) SnapshotFiles(ctx context.Context, repoID, snapshot, path string) (SnapshotBrowserPage, error) { started := time.Now() s.log.Info("snapshot file listing started", "repoID", repoID, "snapshot", snapshot, "path", path) + if err := model.ValidateSnapshotID(snapshot); err != nil { + return SnapshotBrowserPage{}, fmt.Errorf("validation: %w", err) + } repo, ok := s.repository(repoID) if !ok { return SnapshotBrowserPage{}, errors.New("validation: unknown repository") @@ -431,6 +601,14 @@ func (s *Service) SnapshotFiles(ctx context.Context, repoID, snapshot, path stri s.log.Info("snapshot file listing loaded from cache", "repoID", repoID, "snapshot", snapshot, "path", normalizedPath, "items", len(page.Items), "total", page.Total, "durationMs", time.Since(started).Milliseconds()) return page, nil } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return SnapshotBrowserPage{}, err + } + defer unlock() + if page, ok := s.snapshotBrowserPage(key, normalizedPath); ok { + return page, nil + } mounted, err := s.mounts.Prepare(ctx, repo) if err != nil { @@ -439,12 +617,24 @@ func (s *Service) SnapshotFiles(ctx context.Context, repoID, snapshot, path stri } defer s.mounts.Cleanup(context.Background(), mounted) - items, err := s.restic.List(ctx, mounted.Repository, snapshot, "") + const maxSnapshotEntries = 1_000_000 + builder := newSnapshotTreeBuilder() + tooLarge := false + err = s.restic.WalkList(ctx, mounted.Repository, snapshot, "", func(item map[string]any) { + if builder.total >= maxSnapshotEntries { + tooLarge = true + return + } + builder.add(item) + }) if err != nil { s.log.Warn("snapshot file listing failed", "repoID", repoID, "snapshot", snapshot, "path", normalizedPath, "durationMs", time.Since(started).Milliseconds(), "error", err) return SnapshotBrowserPage{}, err } - cache := buildSnapshotTreeCache(items) + if tooLarge { + return SnapshotBrowserPage{}, fmt.Errorf("environment: snapshot contains more than %d entries; narrow the backup or use restic directly", maxSnapshotEntries) + } + cache := builder.cache() s.storeSnapshotTree(key, cache) page, _ := s.snapshotBrowserPage(key, normalizedPath) s.log.Info("snapshot file listing loaded", "repoID", repoID, "snapshot", snapshot, "path", normalizedPath, "items", len(page.Items), "total", cache.total, "durationMs", time.Since(started).Milliseconds()) @@ -480,53 +670,67 @@ func (s *Service) storeSnapshotTree(key string, cache snapshotTreeCache) { delete(s.snapshotTree, cacheKey) } } + for len(s.snapshotTree) >= 3 { + oldestKey := "" + var oldest time.Time + for cacheKey, value := range s.snapshotTree { + if oldestKey == "" || value.generated.Before(oldest) { + oldestKey, oldest = cacheKey, value.generated + } + } + delete(s.snapshotTree, oldestKey) + } s.snapshotTree[key] = cache } func buildSnapshotTreeCache(items []map[string]any) snapshotTreeCache { - root := &snapshotTreeNode{path: "/", name: "/", nodeType: "dir", children: map[string]*snapshotTreeNode{}} - nodes := map[string]*snapshotTreeNode{"/": root} - ensure := func(path string, nodeType string, size int64) *snapshotTreeNode { - return root + builder := newSnapshotTreeBuilder() + for _, item := range items { + builder.add(item) } - ensure = func(path string, nodeType string, size int64) *snapshotTreeNode { - path = normalizeSnapshotPath(path) - if node, ok := nodes[path]; ok { - if nodeType != "" && nodeType != "dir" { - node.nodeType = nodeType - node.size = size - } - return node + return builder.cache() +} + +func newSnapshotTreeBuilder() *snapshotTreeBuilder { + root := &snapshotTreeNode{path: "/", name: "/", nodeType: "dir", children: map[string]*snapshotTreeNode{}} + return &snapshotTreeBuilder{root: root, nodes: map[string]*snapshotTreeNode{"/": root}} +} + +func (b *snapshotTreeBuilder) ensure(path string, nodeType string, size int64) *snapshotTreeNode { + path = normalizeSnapshotPath(path) + if node, ok := b.nodes[path]; ok { + if nodeType != "" && nodeType != "dir" { + node.nodeType, node.size = nodeType, size } - parentPath := "/" - if path != "/" { - parentPath = filepath.Dir(path) - if parentPath == "." { - parentPath = "/" - } - } - parent := ensure(parentPath, "dir", 0) - node := &snapshotTreeNode{path: path, name: filepath.Base(path), nodeType: nodeType, size: size, children: map[string]*snapshotTreeNode{}} - if path == "/" { - node.name = "/" - } - nodes[path] = node - parent.children[path] = node return node } - for _, item := range items { - path := snapshotItemPath(item) - if path == "/" { - continue - } - nodeType := "file" - if value, _ := item["type"].(string); value == "dir" { - nodeType = "dir" - } - ensure(path, nodeType, snapshotItemSize(item)) + parentPath := filepath.Dir(path) + if parentPath == "." { + parentPath = "/" } - children := make(map[string][]SnapshotBrowserNode, len(nodes)) - for _, node := range nodes { + parent := b.ensure(parentPath, "dir", 0) + node := &snapshotTreeNode{path: path, name: filepath.Base(path), nodeType: nodeType, size: size, children: map[string]*snapshotTreeNode{}} + b.nodes[path] = node + parent.children[path] = node + return node +} + +func (b *snapshotTreeBuilder) add(item map[string]any) { + path := snapshotItemPath(item) + if path == "/" { + return + } + nodeType := "file" + if value, _ := item["type"].(string); value == "dir" { + nodeType = "dir" + } + b.ensure(path, nodeType, snapshotItemSize(item)) + b.total++ +} + +func (b *snapshotTreeBuilder) cache() snapshotTreeCache { + children := make(map[string][]SnapshotBrowserNode, len(b.nodes)) + for _, node := range b.nodes { if len(node.children) == 0 { continue } @@ -542,7 +746,7 @@ func buildSnapshotTreeCache(items []map[string]any) snapshotTreeCache { }) children[node.path] = list } - return snapshotTreeCache{generated: time.Now().UTC(), total: len(items), children: children} + return snapshotTreeCache{generated: time.Now().UTC(), total: b.total, children: children} } func snapshotItemPath(item map[string]any) string { @@ -580,6 +784,11 @@ func (s *Service) TestRepository(ctx context.Context, repoID string, initialize if !ok { return errors.New("validation: unknown repository") } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return err + } + defer unlock() mounted, err := s.mounts.Prepare(ctx, repo) if err != nil { return err @@ -630,6 +839,11 @@ func (s *Service) UnlockRepository(ctx context.Context, repoID string) error { if !ok { return errors.New("validation: unknown repository") } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return err + } + defer unlock() mounted, err := s.mounts.Prepare(ctx, repo) if err != nil { return err @@ -643,6 +857,14 @@ func (s *Service) ForceUnlockRepository(ctx context.Context, repoID string) erro if !ok { return errors.New("validation: unknown repository") } + if s.repositoryBusy(repoID) { + return errors.New("validation: repository has a queued, running, or paused task") + } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return err + } + defer unlock() mounted, err := s.mounts.Prepare(ctx, repo) if err != nil { return err @@ -672,6 +894,11 @@ func (s *Service) execute(ctx context.Context, run model.Run) (result model.Run) if !ok { return failed(run, "validation", "repository no longer exists") } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return failedError(run, err) + } + defer unlock() maintenanceRepoName = repo.Name var pruneStatsBefore restic.Stats pruneStatsBeforeOK := false @@ -730,6 +957,9 @@ func (s *Service) executeRsync(ctx context.Context, run model.Run) (result model return failed(run, "validation", "job no longer exists") } defer func() { result = s.finishJob(run, job, result) }() + if err := platform.ValidateRsyncPaths(job); err != nil { + return failedError(run, err) + } live := run appendLiveLog(&live, fmt.Sprintf("Rsync-Kopie wird vorbereitet: %d Quelle(n) nach %s", len(job.Sources), job.Rsync.Target)) s.updateActive(live) @@ -819,6 +1049,11 @@ func (s *Service) executeBackup(ctx context.Context, run model.Run) (result mode if !ok { return failed(run, "validation", "repository no longer exists") } + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return failedError(run, err) + } + defer unlock() live := run appendLiveLog(&live, "Backup wird vorbereitet") s.updateActive(live) @@ -1053,15 +1288,28 @@ func (s *Service) executeRestore(ctx context.Context, run model.Run) model.Run { live := run appendLiveLog(&live, "Restore wird vorbereitet") s.updateActive(live) - parts := strings.Split(run.JobID, "\x00") - if len(parts) < 4 { + if run.Restore == nil { return failed(run, "internal", "invalid restore payload") } - repo, ok := s.repository(parts[0]) + task := *run.Restore + if err := model.ValidateRestoreTask(task); err != nil { + return failed(run, "validation", err.Error()) + } + config := s.Config() + target, err := platform.ValidateRestoreTarget(task.Target, config.Settings.RestoreRoot, task.InPlace, task.Confirmed) + if err != nil { + return failedError(run, err) + } + task.Target = target + repo, ok := s.repository(task.RepositoryID) if !ok { return failed(run, "validation", "repository no longer exists") } - task := model.RestoreTask{SchemaVersion: model.SchemaVersion, ID: run.ID, RepositoryID: parts[0], SnapshotID: parts[1], Target: parts[2], InPlace: parts[3] == "true", Confirmed: true, Includes: parts[4:]} + unlock, err := s.lockRepository(ctx, repo) + if err != nil { + return failedError(run, err) + } + defer unlock() if err := os.MkdirAll(task.Target, 0750); err != nil { failedRun := failedError(run, err) appendLiveLog(&live, "Restore-Ziel konnte nicht vorbereitet werden: "+err.Error()) @@ -1302,6 +1550,47 @@ func (s *Service) repository(id string) (model.Repository, bool) { return model.Repository{}, false } +func (s *Service) lockRepository(ctx context.Context, repo model.Repository) (func(), error) { + key := string(repo.Type) + ":" + filepath.Clean(repo.Location) + if repo.Mount != nil && repo.Mount.Managed { + key = "mount:" + filepath.Clean(repo.Mount.MountPoint) + } + s.repositoryMu.Lock() + lock := s.repositoryLocks[key] + if lock == nil { + lock = make(chan struct{}, 1) + s.repositoryLocks[key] = lock + } + s.repositoryMu.Unlock() + select { + case lock <- struct{}{}: + return func() { <-lock }, nil + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +func (s *Service) repositoryBusy(repoID string) bool { + for _, run := range s.queue.Snapshot() { + if run.Status != "queued" && run.Status != "running" && run.Status != "paused" { + continue + } + if run.TaskType == "restore" && run.Restore != nil && run.Restore.RepositoryID == repoID { + return true + } + if run.TaskType == "check" || run.TaskType == "prune" { + if run.JobID == repoID { + return true + } + continue + } + if job, ok := s.job(run.JobID); ok && job.RepositoryID == repoID { + return true + } + } + return false +} + func failed(run model.Run, code, message string) model.Run { run.Status, run.ErrorCode, run.Message = "failed", code, message return run diff --git a/packaging/build-package.sh b/packaging/build-package.sh index cc8bcdf..2f40fea 100755 --- a/packaging/build-package.sh +++ b/packaging/build-package.sh @@ -2,7 +2,9 @@ set -euo pipefail ROOT=$(cd "$(dirname "$0")/.." && pwd) -VERSION=${VERSION:-0.1.0} +PLUGIN_VERSION=$(sed -n 's/^/\1/p' "$ROOT/plugin/urbm.plg") +VERSION=${VERSION:-$PLUGIN_VERSION} +[ -n "$VERSION" ] || { echo "Could not determine plugin version" >&2; exit 1; } RESTIC_VERSION=0.19.0 RESTIC_SHA256=13176fe6d89d4357947a2cd107218ab2873a5f9d8e1ac2d4cd1c8e07e6839c21 BUILD="$ROOT/build" @@ -28,7 +30,7 @@ fi bzip2 -dc "$RESTIC_ARCHIVE" > "$STAGE/usr/local/libexec/urbm/restic" chmod 0755 "$STAGE/usr/local/libexec/urbm/restic" "$STAGE/usr/local/sbin/urbmd" -cp "$ROOT/webgui/URBM.page" "$STAGE/usr/local/emhttp/plugins/urbm/" +sed "s/$PLUGIN_VERSION/$VERSION/g" "$ROOT/webgui/URBM.page" > "$STAGE/usr/local/emhttp/plugins/urbm/URBM.page" cp "$ROOT/webgui/api.php" "$STAGE/usr/local/emhttp/plugins/urbm/" cp "$ROOT/webgui/assets/"* "$STAGE/usr/local/emhttp/plugins/urbm/assets/" cp "$ROOT/webgui/images/"* "$STAGE/usr/local/emhttp/plugins/urbm/images/" @@ -40,5 +42,5 @@ chmod 0755 "$STAGE/etc/rc.d/rc.urbm" "$STAGE/install/doinst.sh" (cd "$STAGE" && tar --owner=0 --group=0 -cJf "$DIST/urbm-${VERSION}-x86_64-1.txz" .) (cd "$DIST" && sha256sum "urbm-${VERSION}-x86_64-1.txz" > "urbm-${VERSION}-x86_64-1.txz.sha256") PACKAGE_SHA256=$(sha256sum "$DIST/urbm-${VERSION}-x86_64-1.txz" | awk '{print $1}') -sed "s/REPLACE_DURING_RELEASE/$PACKAGE_SHA256/" "$ROOT/plugin/urbm.plg" > "$DIST/urbm.plg" +sed -e "s///" -e "s/REPLACE_DURING_RELEASE/$PACKAGE_SHA256/" "$ROOT/plugin/urbm.plg" > "$DIST/urbm.plg" echo "Built $DIST/urbm-${VERSION}-x86_64-1.txz" diff --git a/plugin/rc.urbm b/plugin/rc.urbm index 6c28db7..6b8d549 100755 --- a/plugin/rc.urbm +++ b/plugin/rc.urbm @@ -4,30 +4,45 @@ DAEMON=/usr/local/sbin/urbmd PIDFILE=/run/urbm/urbm.pid LOGFILE=/var/log/urbm.log +is_daemon_pid() { + PID=${1:-} + [ -n "$PID" ] && [ "$(readlink -f "/proc/$PID/exe" 2>/dev/null)" = "$DAEMON" ] +} + start() { mkdir -p /run/urbm /var/lib/urbm /boot/config/plugins/urbm chmod 0750 /run/urbm /var/lib/urbm chmod 0700 /boot/config/plugins/urbm - if [ -s "$PIDFILE" ] && kill -0 "$(cat "$PIDFILE")" 2>/dev/null; then + if [ -s "$PIDFILE" ] && is_daemon_pid "$(cat "$PIDFILE")"; then return 0 fi rm -f "$PIDFILE" /run/urbm/urbm.sock nohup "$DAEMON" >>"$LOGFILE" 2>&1 & echo $! >"$PIDFILE" + for _ in $(seq 1 50); do + [ -S /run/urbm/urbm.sock ] && return 0 + is_daemon_pid "$(cat "$PIDFILE" 2>/dev/null)" || break + sleep 0.1 + done + rm -f "$PIDFILE" + echo "urbmd failed to create its API socket" >&2 + return 1 } stop() { STOPPED=0 if [ -s "$PIDFILE" ]; then PID=$(cat "$PIDFILE") - kill "$PID" 2>/dev/null || true - for _ in $(seq 1 30); do - kill -0 "$PID" 2>/dev/null || break - sleep 1 - done - kill -9 "$PID" 2>/dev/null || true + if is_daemon_pid "$PID"; then + kill "$PID" 2>/dev/null || true + for _ in $(seq 1 30); do + is_daemon_pid "$PID" || break + sleep 1 + done + is_daemon_pid "$PID" && kill -9 "$PID" 2>/dev/null || true + STOPPED=1 + fi rm -f "$PIDFILE" - STOPPED=1 fi if [ "$STOPPED" -eq 0 ]; then PIDS=$(pidof urbmd 2>/dev/null || true) @@ -50,7 +65,7 @@ case "${1:-}" in stop) stop ;; restart) stop; start ;; status) - if [ -s "$PIDFILE" ] && kill -0 "$(cat "$PIDFILE")" 2>/dev/null; then echo "urbmd is running"; else echo "urbmd is stopped"; exit 1; fi + if [ -s "$PIDFILE" ] && is_daemon_pid "$(cat "$PIDFILE")"; then echo "urbmd is running"; else echo "urbmd is stopped"; exit 1; fi ;; *) echo "Usage: $0 {start|stop|restart|status}"; exit 2 ;; esac diff --git a/plugin/urbm.plg b/plugin/urbm.plg index 076ef67..153d598 100644 --- a/plugin/urbm.plg +++ b/plugin/urbm.plg @@ -2,13 +2,19 @@ - + ]> +### 2026.07.13.r001 +- Harden restore task validation and make confirmed in-place restores target original paths safely. +- Serialize repository and managed-mount access, block unsafe force-unlock operations, and wait for task cleanup during daemon shutdown. +- Bound snapshot browser memory, stream Restic diffs, continuously drain command output, and reject symlink-overlapping Rsync paths. +- Activate persistent run logs and align package, WebGUI, and manifest versions during builds. + ### 2026.07.10.r007 - Serve snapshot browser folders as compact cached tree pages instead of sending the complete Restic file list to the browser. - Prevent large snapshots with hundreds of thousands of entries from failing with generic 500 responses during snapshot browsing. diff --git a/webgui/URBM.page b/webgui/URBM.page index 13516d9..aa66c91 100644 --- a/webgui/URBM.page +++ b/webgui/URBM.page @@ -8,13 +8,15 @@ Tag="URBM Unraid Restic Backup Manager backup snapshots restore" --- - +
- -

URBM 2026.07.10.r007

Restic Backup Manager für Unraid

+ +

URBM

Restic Backup Manager für Unraid

Verbindung wird hergestellt...
@@ -33,4 +35,4 @@ $pluginRoot = '/plugins/urbm';
- + diff --git a/webgui/assets/urbm.js b/webgui/assets/urbm.js index 058b5b6..0fa54af 100644 --- a/webgui/assets/urbm.js +++ b/webgui/assets/urbm.js @@ -181,8 +181,12 @@ const runTarget = run => { if(run.taskType==='backup'||run.taskType==='rsync') return state.config.jobs.find(job=>job.id===run.jobId)?.name || 'Unbekannter Job'; if(['check','prune'].includes(run.taskType)) return state.config.repositories.find(repo=>repo.id===run.jobId)?.name || 'Unbekanntes Repository'; - if(run.taskType==='restore') { - const parts=String(run.jobId||'').split('\u0000'); + if(run.taskType==='restore') { + if(run.restore) { + const repo=state.config.repositories.find(candidate=>candidate.id===run.restore.repositoryId)?.name || 'Repository'; + return `${repo} · ${String(run.restore.snapshotId||'').slice(0,12)} → ${run.restore.target||''}`; + } + const parts=String(run.jobId||'').split('\u0000'); if(parts.length>=4) { const repo=state.config.repositories.find(candidate=>candidate.id===parts[0])?.name || 'Repository'; return `${repo} · ${parts[1].slice(0,12)} → ${parts[2]}`; @@ -915,10 +919,19 @@ const manual=e.target.closest('.bu-schedule')?.querySelector('.bu-schedule-manual'); if(manual) manual.hidden=e.target.value!=='manual'; } - if (e.target.name === 'type' && e.target.closest('#bu-job-form')) { + if (e.target.name === 'type' && e.target.closest('#bu-job-form')) { state.backupSelections=[]; state.workloadSelections=[]; renderJobSpecific(e.target.value); - } + } + if (e.target.name === 'inPlace' && e.target.closest('#bu-restore-form')) { + const target=e.target.closest('form').querySelector('[name="target"]'); + const tree=e.target.closest('form').querySelector('#bu-dir-tree-restoreTarget'); + if(target) { + if(e.target.checked) { target.dataset.stagingTarget=target.value; target.value='/'; target.readOnly=true; } + else { target.value=target.dataset.stagingTarget||state.config.settings.restoreRoot; target.readOnly=false; } + } + if(tree) tree.hidden=e.target.checked; + } if (e.target.classList.contains('bu-source-select')) { const path=e.target.value; const prefix=path.replace(/\/$/,'')+'/'; if(e.target.checked) state.backupSelections=normalizeSourceSelections([...state.backupSelections.filter(selected=>!selected.startsWith(prefix)),path]);