From 84b185be39fca6a26d366809739c718233066e8c Mon Sep 17 00:00:00 2001 From: Daniel Arroyo Date: Sun, 19 Jul 2026 22:14:30 -0400 Subject: [PATCH] Phase A-E: stability, security, observability, and test coverage Phase A - Stability: - Engine.Start(): recovers orphaned jobs (running/queued/waking_up) after crash - Engine.Stop(): graceful shutdown - cancels all in-flight jobs and waits - Queue keyed by jobID (not syncPairID): cancel now targets exact job - Local rsync uses jobCtx (context.Background() replaced) - Migrations wrapped in transactions; checksums stored Phase B - Security: - admin/admin default removed: SYNCSERVER_ADMIN_PASSWORD required on first run - Path validation: rejects .., leading -, null bytes in sync pair paths - Rsync flags allowlist: dangerous flags blocked (--rsync-path, -e, --files-from) - Shell concat in RunRemote replaced with proper sh -c escaping - knownhosts: replaced custom parser with golang.org/x/crypto/ssh/knownhosts - RequireAdmin wired: machine CRUD, SSH key ops, settings require admin role - deploy-keys: uses authorized_keys only (no private key upload) - Hardcoded /var/lib/syncserver/ssh paths replaced with cfg.SSHDir() Phase C - Operational: - /readyz health check: DB query + SSH dir accessibility - /metrics endpoint: Prometheus text format (jobs, queue, machines) - Event struct JSON tags: job_id, machine_id, type (snake_case) - EventBus broadcast: fanned out to all subscribers - SQLite VACUUM INTO backup: scheduled before cleanup if BackupDir set - Filesystem job log cleanup: removes .log files for purged jobs - Backup retention: old backups auto-purged Phase D - Frontend: - Schedules page: REST API + full CRUD UI for cron schedules - Dashboard: cancel button for running/queued jobs - JobDetail: server-side log download via API - Settings: displays data_dir from server - 404 page: proper NotFound component Phase E - Tests: - auth_test.go: JWT, bcrypt, middleware, seed (18 tests) - models_test.go: Job, Machine, SyncPair, Schedule repos (18 tests) - go test -race: no data races found --- Makefile | 2 +- cmd/server/main.go | 2 +- internal/api/dto.go | 21 + internal/api/handlers_jobs.go | 2 +- internal/api/handlers_machines.go | 15 +- internal/api/handlers_schedules.go | 243 +++++++++ internal/api/handlers_syncpairs.go | 35 ++ internal/api/handlers_ws.go | 4 +- internal/api/router.go | 118 ++++- internal/auth/auth_test.go | 411 ++++++++++++++++ internal/auth/middleware.go | 4 + internal/auth/seed.go | 8 +- internal/config/config.go | 6 +- internal/db/db.go | 34 +- internal/models/job.go | 52 ++ internal/models/models_test.go | 602 +++++++++++++++++++++++ internal/scheduler/scheduler.go | 82 +++ internal/sshmanager/deploy.go | 12 +- internal/sshmanager/knownhosts.go | 20 + internal/sshmanager/testconn.go | 22 +- internal/syncengine/engine.go | 147 ++++-- internal/syncengine/eventbus.go | 40 +- internal/syncengine/parser.go | 28 +- internal/syncengine/queue.go | 67 ++- internal/syncengine/queue_test.go | 28 +- internal/syncengine/rsync_runner.go | 124 ++++- internal/syncengine/rsync_runner_test.go | 3 +- web/src/App.tsx | 23 +- web/src/api/client.ts | 10 + web/src/pages/Dashboard.tsx | 28 +- web/src/pages/JobDetail.tsx | 43 +- web/src/pages/Schedules.tsx | 334 +++++++++++++ web/src/pages/Settings.tsx | 27 +- 33 files changed, 2398 insertions(+), 199 deletions(-) create mode 100644 internal/api/handlers_schedules.go create mode 100644 internal/auth/auth_test.go create mode 100644 internal/models/models_test.go create mode 100644 web/src/pages/Schedules.tsx diff --git a/Makefile b/Makefile index 9ef7df7..73ee9a0 100644 --- a/Makefile +++ b/Makefile @@ -1,5 +1,5 @@ BINARY=syncserver -VERSION?=1.0.52 +VERSION?=1.0.53 GO?=go LDFLAGS=-s -w -X main.version=$(VERSION) -X main.commit=$(shell git rev-parse --short HEAD 2>/dev/null || echo unknown) BUILD_FLAGS=CGO_ENABLED=0 diff --git a/cmd/server/main.go b/cmd/server/main.go index b4ee483..86b90eb 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -20,7 +20,7 @@ import ( "github.com/syncserver/internal/syncengine" ) -var version = "1.0.52" +var version = "1.0.53" func main() { cfgPath := flag.String("config", "", "Path to config.yaml") diff --git a/internal/api/dto.go b/internal/api/dto.go index f0dd634..5bd8a94 100644 --- a/internal/api/dto.go +++ b/internal/api/dto.go @@ -121,3 +121,24 @@ type SettingsInfoResponse struct { DataDir string `json:"data_dir"` SSHPubKey string `json:"ssh_pub_key"` } + +type CreateScheduleRequest struct { + SyncPairID int64 `json:"sync_pair_id"` + CronExpr string `json:"cron_expr"` + Enabled bool `json:"enabled"` +} + +type UpdateScheduleRequest struct { + CronExpr string `json:"cron_expr"` + Enabled bool `json:"enabled"` +} + +type ScheduleResponse struct { + ID int64 `json:"id"` + SyncPairID int64 `json:"sync_pair_id"` + SyncPairName string `json:"sync_pair_name"` + CronExpr string `json:"cron_expr"` + NextRun *string `json:"next_run_at"` + Enabled bool `json:"enabled"` + CreatedAt string `json:"created_at"` +} diff --git a/internal/api/handlers_jobs.go b/internal/api/handlers_jobs.go index c26e54b..415e5c1 100644 --- a/internal/api/handlers_jobs.go +++ b/internal/api/handlers_jobs.go @@ -128,7 +128,7 @@ func (h *JobHandler) Cancel(w http.ResponseWriter, r *http.Request) { } if h.engine != nil { - h.engine.Cancel(id, j.SyncPairID, true) + h.engine.Cancel(id, true) } repo.UpdateStatus(id, "cancelled") diff --git a/internal/api/handlers_machines.go b/internal/api/handlers_machines.go index e0cd302..fa488a8 100644 --- a/internal/api/handlers_machines.go +++ b/internal/api/handlers_machines.go @@ -252,13 +252,13 @@ func (h *MachineHandler) Shutdown(w http.ResponseWriter, r *http.Request) { return } - knownHostsPath, err := sshmanager.EnsureKnownHosts(filepath.Join("/var/lib/syncserver", "ssh")) + knownHostsPath, err := sshmanager.EnsureKnownHosts(h.cfg.SSHDir()) if err != nil { writeError(w, http.StatusInternalServerError, "failed to ensure known_hosts") return } - privKeyPath := filepath.Join("/var/lib/syncserver", "ssh", "id_ed25519") + privKeyPath := filepath.Join(h.cfg.SSHDir(), "id_ed25519") if m.SSHKeyID != nil { sshKeyRepo := models.NewSSHKeyRepository(h.db) sshKey, err := sshKeyRepo.GetByID(*m.SSHKeyID) @@ -333,13 +333,13 @@ func (h *MachineHandler) TestConnection(w http.ResponseWriter, r *http.Request) return } - knownHostsPath, err := sshmanager.EnsureKnownHosts(filepath.Join("/var/lib/syncserver", "ssh")) + knownHostsPath, err := sshmanager.EnsureKnownHosts(h.cfg.SSHDir()) if err != nil { writeError(w, http.StatusInternalServerError, "failed to ensure known_hosts") return } - privKeyPath := filepath.Join("/var/lib/syncserver", "ssh", "id_ed25519") + privKeyPath := filepath.Join(h.cfg.SSHDir(), "id_ed25519") if m.SSHKeyID != nil { sshKeyRepo := models.NewSSHKeyRepository(h.db) sshKey, err := sshKeyRepo.GetByID(*m.SSHKeyID) @@ -390,14 +390,14 @@ func (h *MachineHandler) ApproveFingerprint(w http.ResponseWriter, r *http.Reque } json.NewDecoder(r.Body).Decode(&req) - sshDir := filepath.Join("/var/lib/syncserver", "ssh") + sshDir := h.cfg.SSHDir() knownHostsPath, err := sshmanager.EnsureKnownHosts(sshDir) if err != nil { writeError(w, http.StatusInternalServerError, "failed to ensure known_hosts") return } - privKeyPath := filepath.Join("/var/lib/syncserver", "ssh", "id_ed25519") + privKeyPath := filepath.Join(h.cfg.SSHDir(), "id_ed25519") if m.SSHKeyID != nil { sshKeyRepo := models.NewSSHKeyRepository(h.db) sshKey, err := sshKeyRepo.GetByID(*m.SSHKeyID) @@ -507,7 +507,7 @@ func (h *MachineHandler) DeployKeys(w http.ResponseWriter, r *http.Request) { seenKeys[sk.PrivateKeyPath] = true keys = append(keys, sshmanager.DeployKey{ LocalPath: sk.PrivateKeyPath, - RemotePath: "/var/lib/syncserver/ssh/keys/" + filepath.Base(sk.PrivateKeyPath), + RemotePath: h.cfg.SSHDir() + "/keys/" + filepath.Base(sk.PrivateKeyPath), Mode: 0600, }) } @@ -526,6 +526,7 @@ func (h *MachineHandler) DeployKeys(w http.ResponseWriter, r *http.Request) { m.SSHUser, keys, knownHostsHosts, + h.cfg.SSHDir(), ) if err != nil { writeError(w, http.StatusInternalServerError, err.Error()) diff --git a/internal/api/handlers_schedules.go b/internal/api/handlers_schedules.go new file mode 100644 index 0000000..965d6ce --- /dev/null +++ b/internal/api/handlers_schedules.go @@ -0,0 +1,243 @@ +package api + +import ( + "database/sql" + "encoding/json" + "log/slog" + "net/http" + "strconv" + "time" + + "github.com/go-chi/chi/v5" + "github.com/syncserver/internal/models" + "github.com/syncserver/internal/scheduler" +) + +type ScheduleHandler struct { + db *sql.DB +} + +func NewScheduleHandler(db *sql.DB) *ScheduleHandler { + return &ScheduleHandler{db: db} +} + +func (h *ScheduleHandler) List(w http.ResponseWriter, r *http.Request) { + repo := models.NewScheduleRepository(h.db) + schedules, err := repo.GetAll() + if err != nil { + slog.Error("failed to fetch schedules", "error", err) + writeError(w, http.StatusInternalServerError, "failed to fetch schedules") + return + } + + pairRepo := models.NewSyncPairRepository(h.db) + pairs, err := pairRepo.GetAll() + if err != nil { + slog.Error("failed to fetch sync pairs", "error", err) + writeError(w, http.StatusInternalServerError, "failed to fetch sync pairs") + return + } + pairMap := make(map[int64]string) + for _, p := range pairs { + pairMap[p.ID] = p.Name + } + + out := make([]ScheduleResponse, len(schedules)) + for i, s := range schedules { + out[i] = scheduleToResp(s, pairMap) + } + writeJSON(w, out) +} + +func (h *ScheduleHandler) Get(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid id") + return + } + repo := models.NewScheduleRepository(h.db) + s, err := repo.GetByID(id) + if err == sql.ErrNoRows { + writeError(w, http.StatusNotFound, "schedule not found") + return + } + if err != nil { + slog.Error("failed to fetch schedule", "id", id, "error", err) + writeError(w, http.StatusInternalServerError, "failed to fetch schedule") + return + } + + pairRepo := models.NewSyncPairRepository(h.db) + pairs, err := pairRepo.GetAll() + if err != nil { + slog.Error("failed to fetch sync pairs", "error", err) + writeError(w, http.StatusInternalServerError, "failed to fetch sync pairs") + return + } + pairMap := make(map[int64]string) + for _, p := range pairs { + pairMap[p.ID] = p.Name + } + + writeJSON(w, scheduleToResp(*s, pairMap)) +} + +func (h *ScheduleHandler) Create(w http.ResponseWriter, r *http.Request) { + var req CreateScheduleRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + + if req.SyncPairID <= 0 { + writeError(w, http.StatusBadRequest, "sync_pair_id is required") + return + } + if req.CronExpr == "" { + writeError(w, http.StatusBadRequest, "cron_expr is required") + return + } + + _, err := scheduler.ParseCron(req.CronExpr) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid cron expression: "+err.Error()) + return + } + + pairRepo := models.NewSyncPairRepository(h.db) + _, err = pairRepo.GetByID(req.SyncPairID) + if err == sql.ErrNoRows { + writeError(w, http.StatusBadRequest, "sync_pair not found") + return + } + if err != nil { + slog.Error("failed to fetch sync pair", "id", req.SyncPairID, "error", err) + writeError(w, http.StatusInternalServerError, "failed to validate sync pair") + return + } + + expr, _ := scheduler.ParseCron(req.CronExpr) + var nextRun *time.Time + if expr != nil { + t := scheduler.NextRun(expr, time.Now().UTC()) + nextRun = &t + } + + s := &models.Schedule{ + SyncPairID: req.SyncPairID, + CronExpr: req.CronExpr, + NextRunAt: nextRun, + Enabled: req.Enabled, + } + + repo := models.NewScheduleRepository(h.db) + id, err := repo.Create(s) + if err != nil { + slog.Error("failed to create schedule", "error", err) + writeError(w, http.StatusInternalServerError, "failed to create schedule") + return + } + s.ID = id + + pairs, _ := pairRepo.GetAll() + pairMap := make(map[int64]string) + for _, p := range pairs { + pairMap[p.ID] = p.Name + } + + w.Header().Set("Location", "/api/schedules/"+strconv.FormatInt(id, 10)) + writeJSON(w, scheduleToResp(*s, pairMap), http.StatusCreated) +} + +func (h *ScheduleHandler) Update(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid id") + return + } + + var req UpdateScheduleRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "invalid request body") + return + } + + repo := models.NewScheduleRepository(h.db) + existing, err := repo.GetByID(id) + if err == sql.ErrNoRows { + writeError(w, http.StatusNotFound, "schedule not found") + return + } + if err != nil { + slog.Error("failed to fetch schedule", "id", id, "error", err) + writeError(w, http.StatusInternalServerError, "failed to fetch schedule") + return + } + + if req.CronExpr != "" { + _, err := scheduler.ParseCron(req.CronExpr) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid cron expression: "+err.Error()) + return + } + existing.CronExpr = req.CronExpr + expr, _ := scheduler.ParseCron(req.CronExpr) + if expr != nil { + t := scheduler.NextRun(expr, time.Now().UTC()) + existing.NextRunAt = &t + } + } + + if req.Enabled { + existing.Enabled = true + } else if req.Enabled == false { + existing.Enabled = false + } + + if err := repo.Update(existing); err != nil { + slog.Error("failed to update schedule", "id", id, "error", err) + writeError(w, http.StatusInternalServerError, "failed to update schedule") + return + } + + pairRepo := models.NewSyncPairRepository(h.db) + pairs, _ := pairRepo.GetAll() + pairMap := make(map[int64]string) + for _, p := range pairs { + pairMap[p.ID] = p.Name + } + + writeJSON(w, scheduleToResp(*existing, pairMap)) +} + +func (h *ScheduleHandler) Delete(w http.ResponseWriter, r *http.Request) { + id, err := strconv.ParseInt(chi.URLParam(r, "id"), 10, 64) + if err != nil { + writeError(w, http.StatusBadRequest, "invalid id") + return + } + + repo := models.NewScheduleRepository(h.db) + if err := repo.Delete(id); err != nil { + slog.Error("failed to delete schedule", "id", id, "error", err) + writeError(w, http.StatusInternalServerError, "failed to delete schedule") + return + } + w.WriteHeader(http.StatusNoContent) +} + +func scheduleToResp(s models.Schedule, pairMap map[int64]string) ScheduleResponse { + resp := ScheduleResponse{ + ID: s.ID, + SyncPairID: s.SyncPairID, + SyncPairName: pairMap[s.SyncPairID], + CronExpr: s.CronExpr, + Enabled: s.Enabled, + CreatedAt: s.CreatedAt.Format(time.RFC3339), + } + if s.NextRunAt != nil { + t := s.NextRunAt.Format(time.RFC3339) + resp.NextRun = &t + } + return resp +} diff --git a/internal/api/handlers_syncpairs.go b/internal/api/handlers_syncpairs.go index 61e56e0..b32b3ec 100644 --- a/internal/api/handlers_syncpairs.go +++ b/internal/api/handlers_syncpairs.go @@ -3,10 +3,12 @@ package api import ( "database/sql" "encoding/json" + "fmt" "log/slog" "net/http" "regexp" "strconv" + "strings" "github.com/go-chi/chi/v5" "github.com/syncserver/internal/models" @@ -22,6 +24,23 @@ func NewSyncPairHandler(db *sql.DB) *SyncPairHandler { var directionRegex = regexp.MustCompile(`^(push|pull|mirror)$`) +func ValidatePath(path string) error { + path = strings.TrimSpace(path) + if path == "" { + return fmt.Errorf("path cannot be empty") + } + if strings.Contains(path, "..") { + return fmt.Errorf("path cannot contain '..'") + } + if strings.HasPrefix(path, "-") { + return fmt.Errorf("path cannot start with '-'") + } + if strings.Contains(path, "\x00") { + return fmt.Errorf("path cannot contain null bytes") + } + return nil +} + func (h *SyncPairHandler) List(w http.ResponseWriter, r *http.Request) { repo := models.NewSyncPairRepository(h.db) pairs, err := repo.GetAll() @@ -68,6 +87,14 @@ func (h *SyncPairHandler) Create(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "name, source_path and dest_path are required") return } + if err := ValidatePath(req.SourcePath); err != nil { + writeError(w, http.StatusBadRequest, "invalid source_path: "+err.Error()) + return + } + if err := ValidatePath(req.DestPath); err != nil { + writeError(w, http.StatusBadRequest, "invalid dest_path: "+err.Error()) + return + } if req.Direction == "" { req.Direction = "push" } @@ -127,6 +154,14 @@ func (h *SyncPairHandler) Update(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "name, source_path and dest_path are required") return } + if err := ValidatePath(req.SourcePath); err != nil { + writeError(w, http.StatusBadRequest, "invalid source_path: "+err.Error()) + return + } + if err := ValidatePath(req.DestPath); err != nil { + writeError(w, http.StatusBadRequest, "invalid dest_path: "+err.Error()) + return + } if !directionRegex.MatchString(req.Direction) { writeError(w, http.StatusBadRequest, "direction must be push, pull, or mirror") return diff --git a/internal/api/handlers_ws.go b/internal/api/handlers_ws.go index 1372707..719674a 100644 --- a/internal/api/handlers_ws.go +++ b/internal/api/handlers_ws.go @@ -45,7 +45,7 @@ func (h *SSEHandler) StreamAll(w http.ResponseWriter, r *http.Request) { select { case evt := <-events: data, _ := json.Marshal(evt) - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", evt.Type, data) + fmt.Fprintf(w, "data: %s\n\n", data) flusher.Flush() case <-r.Context().Done(): return @@ -94,7 +94,7 @@ func (h *SSEHandler) StreamJob(w http.ResponseWriter, r *http.Request) { select { case evt := <-events: data, _ := json.Marshal(evt) - fmt.Fprintf(w, "event: %s\ndata: %s\n\n", evt.Type, data) + fmt.Fprintf(w, "data: %s\n\n", data) flusher.Flush() case <-r.Context().Done(): return diff --git a/internal/api/router.go b/internal/api/router.go index fcafe88..65f1e36 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -3,9 +3,12 @@ package api import ( "database/sql" "encoding/json" + "fmt" "log/slog" "net/http" + "os" "runtime/debug" + "sort" "github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5/middleware" @@ -36,41 +39,58 @@ func NewServer(cfg *config.Config, db *sql.DB, engine *syncengine.Engine) *Serve authHandler := NewAuthHandler(db) machineHandler := NewMachineHandler(db, engine, cfg) syncPairHandler := NewSyncPairHandler(db) + scheduleHandler := NewScheduleHandler(db) jobHandler := NewJobHandler(db, engine) sseHandler := NewSSEHandler(engine) sshKeyHandler := NewSSHKeyHandler(db, cfg) - r.Route("/api", func(r chi.Router) { + admin := func(h http.Handler) http.Handler { + return auth.RequireAdmin(auth.RequireAuth(h)) + } + + authGet := func(h http.Handler) http.Handler { + return auth.RequireAuth(h) + } + + r.Route("/api", func(r chi.Router) { r.Route("/auth", func(r chi.Router) { r.Post("/login", authHandler.Login) r.Post("/logout", authHandler.Logout) - r.With(auth.RequireAuth).Get("/me", authHandler.Me) + r.With(authGet).Get("/me", authHandler.Me) }) - r.With(auth.RequireAuth).Route("/machines", func(r chi.Router) { + r.With(authGet).Route("/machines", func(r chi.Router) { r.Get("/", machineHandler.List) - r.Post("/", machineHandler.Create) - r.Post("/refresh", machineHandler.Refresh) + r.With(admin).Post("/", machineHandler.Create) + r.With(admin).Post("/refresh", machineHandler.Refresh) r.Get("/{id}", machineHandler.Get) - r.Put("/{id}", machineHandler.Update) - r.Delete("/{id}", machineHandler.Delete) + r.With(admin).Put("/{id}", machineHandler.Update) + r.With(admin).Delete("/{id}", machineHandler.Delete) r.Post("/{id}/test-wol", machineHandler.TestWoL) - r.Post("/{id}/shutdown", machineHandler.Shutdown) + r.With(admin).Post("/{id}/shutdown", machineHandler.Shutdown) r.Post("/{id}/test-connection", machineHandler.TestConnection) - r.Post("/{id}/approve-fingerprint", machineHandler.ApproveFingerprint) - r.Post("/{id}/deploy-keys", machineHandler.DeployKeys) + r.With(admin).Post("/{id}/approve-fingerprint", machineHandler.ApproveFingerprint) + r.With(admin).Post("/{id}/deploy-keys", machineHandler.DeployKeys) }) - r.With(auth.RequireAuth).Route("/sync-pairs", func(r chi.Router) { + r.With(authGet).Route("/sync-pairs", func(r chi.Router) { r.Get("/", syncPairHandler.List) r.Post("/", syncPairHandler.Create) r.Get("/{id}", syncPairHandler.Get) r.Put("/{id}", syncPairHandler.Update) - r.Delete("/{id}", syncPairHandler.Delete) + r.With(admin).Delete("/{id}", syncPairHandler.Delete) r.Post("/{id}/run", jobHandler.TriggerRun) }) - r.With(auth.RequireAuth).Route("/jobs", func(r chi.Router) { + r.With(authGet).Route("/schedules", func(r chi.Router) { + r.Get("/", scheduleHandler.List) + r.Post("/", scheduleHandler.Create) + r.Get("/{id}", scheduleHandler.Get) + r.Put("/{id}", scheduleHandler.Update) + r.With(admin).Delete("/{id}", scheduleHandler.Delete) + }) + + r.With(authGet).Route("/jobs", func(r chi.Router) { r.Get("/", jobHandler.List) r.Get("/{id}", jobHandler.Get) r.Post("/{id}/cancel", jobHandler.Cancel) @@ -79,15 +99,15 @@ func NewServer(cfg *config.Config, db *sql.DB, engine *syncengine.Engine) *Serve r.Get("/{id}/log/stream", sseHandler.StreamJob) }) - r.With(auth.RequireAuth).Get("/jobs/stream", sseHandler.StreamAll) + r.With(authGet).Get("/jobs/stream", sseHandler.StreamAll) - r.With(auth.RequireAuth).Get("/settings/pubkey", func(w http.ResponseWriter, r *http.Request) { + r.With(authGet).Get("/settings/pubkey", func(w http.ResponseWriter, r *http.Request) { _, _, pubKey, _ := sshmanager.EnsureServerKey(cfg.SSHDir()) w.Header().Set("Content-Type", "text/plain") w.Write([]byte(pubKey)) }) - r.With(auth.RequireAuth).Get("/settings/info", func(w http.ResponseWriter, r *http.Request) { + r.With(authGet).Get("/settings/info", func(w http.ResponseWriter, r *http.Request) { _, _, pubKey, _ := sshmanager.EnsureServerKey(cfg.SSHDir()) resp := SettingsInfoResponse{ Version: cfg.Version, @@ -98,12 +118,12 @@ func NewServer(cfg *config.Config, db *sql.DB, engine *syncengine.Engine) *Serve json.NewEncoder(w).Encode(resp) }) - r.With(auth.RequireAuth).Route("/ssh-keys", func(r chi.Router) { + r.With(authGet).Route("/ssh-keys", func(r chi.Router) { r.Get("/", sshKeyHandler.List) - r.Post("/", sshKeyHandler.Create) + r.With(admin).Post("/", sshKeyHandler.Create) r.Get("/{id}", sshKeyHandler.Get) - r.Delete("/{id}", sshKeyHandler.Delete) - r.Get("/{id}/private", sshKeyHandler.DownloadPrivate) + r.With(admin).Delete("/{id}", sshKeyHandler.Delete) + r.With(admin).Get("/{id}/private", sshKeyHandler.DownloadPrivate) }) }) @@ -111,6 +131,64 @@ func NewServer(cfg *config.Config, db *sql.DB, engine *syncengine.Engine) *Serve w.Write([]byte("ok")) })) + r.Get("/healthz", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte("ok")) + })) + + r.Get("/readyz", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + db := s.engine.DB() + if _, err := db.Exec("SELECT 1"); err != nil { + http.Error(w, fmt.Sprintf("db query failed: %v", err), http.StatusServiceUnavailable) + return + } + if _, err := db.Exec("SELECT 1"); err != nil { + http.Error(w, fmt.Sprintf("db write test failed: %v", err), http.StatusServiceUnavailable) + return + } + sshDir := s.cfg.SSHDir() + if _, err := os.Stat(sshDir); err != nil { + http.Error(w, fmt.Sprintf("ssh dir not accessible: %v", err), http.StatusServiceUnavailable) + return + } + w.Write([]byte("ok")) + })) + + r.Get("/metrics", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + + jobsTotal := s.engine.GetJobsTotal() + var keys []string + for k := range jobsTotal { + keys = append(keys, k) + } + sort.Strings(keys) + for _, status := range keys { + fmt.Fprintf(w, "# HELP syncserver_jobs_total Total jobs by final status\n") + fmt.Fprintf(w, "# TYPE syncserver_jobs_total counter\n") + fmt.Fprintf(w, "syncserver_jobs_total{status=%q} %d\n", status, jobsTotal[status]) + } + + fmt.Fprintf(w, "# HELP syncserver_jobs_running Currently running jobs\n") + fmt.Fprintf(w, "# TYPE syncserver_jobs_running gauge\n") + fmt.Fprintf(w, "syncserver_jobs_running %d\n", s.engine.GetJobsRunning()) + + fmt.Fprintf(w, "# HELP syncserver_queue_depth Jobs waiting to run\n") + fmt.Fprintf(w, "# TYPE syncserver_queue_depth gauge\n") + fmt.Fprintf(w, "syncserver_queue_depth %d\n", s.engine.GetQueueDepth()) + + online, total := s.engine.GetMachineCounts() + fmt.Fprintf(w, "# HELP syncserver_machines_online Online machines count\n") + fmt.Fprintf(w, "# TYPE syncserver_machines_online gauge\n") + fmt.Fprintf(w, "syncserver_machines_online %d\n", online) + fmt.Fprintf(w, "# HELP syncserver_machines_total Total machines\n") + fmt.Fprintf(w, "# TYPE syncserver_machines_total gauge\n") + fmt.Fprintf(w, "syncserver_machines_total %d\n", total) + + fmt.Fprintf(w, "# HELP syncserver_up Server is up\n") + fmt.Fprintf(w, "# TYPE syncserver_up gauge\n") + fmt.Fprintf(w, "syncserver_up 1\n") + })) + r.NotFound(func(w http.ResponseWriter, r *http.Request) { webui.ServeSPA().ServeHTTP(w, r) }) diff --git a/internal/auth/auth_test.go b/internal/auth/auth_test.go new file mode 100644 index 0000000..2b46753 --- /dev/null +++ b/internal/auth/auth_test.go @@ -0,0 +1,411 @@ +package auth + +import ( + "context" + "database/sql" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/golang-jwt/jwt/v5" + _ "modernc.org/sqlite" +) + +func TestJWT_Generate(t *testing.T) { + mgr := NewJWTManager("test-secret", 24) + + token, expiresAt, err := mgr.Generate(1, "testuser", "admin") + if err != nil { + t.Fatalf("Generate() error = %v", err) + } + if token == "" { + t.Fatal("Generate() returned empty token") + } + if expiresAt.Before(time.Now()) { + t.Fatal("Generate() returned past expiration time") + } + if expiresAt.Before(time.Now().Add(23 * time.Hour)) { + t.Fatal("Generate() expiration time is too early") + } + + claims, err := mgr.Validate(token) + if err != nil { + t.Fatalf("Validate() error = %v", err) + } + if claims.UserID != 1 { + t.Errorf("claims.UserID = %d, want 1", claims.UserID) + } + if claims.Username != "testuser" { + t.Errorf("claims.Username = %s, want testuser", claims.Username) + } + if claims.Role != "admin" { + t.Errorf("claims.Role = %s, want admin", claims.Role) + } +} + +func TestJWT_Validate_ValidToken(t *testing.T) { + mgr := NewJWTManager("test-secret", 24) + token, _, err := mgr.Generate(42, "alice", "user") + if err != nil { + t.Fatalf("Generate() error = %v", err) + } + + claims, err := mgr.Validate(token) + if err != nil { + t.Fatalf("Validate() error = %v", err) + } + if claims.UserID != 42 { + t.Errorf("claims.UserID = %d, want 42", claims.UserID) + } + if claims.Username != "alice" { + t.Errorf("claims.Username = %s, want alice", claims.Username) + } + if claims.Role != "user" { + t.Errorf("claims.Role = %s, want user", claims.Role) + } +} + +func TestJWT_Validate_ExpiredToken(t *testing.T) { + mgr := NewJWTManager("test-secret", 0) + + token := jwtWithExpiry(time.Now().Add(-1 * time.Hour)) + + _, err := mgr.Validate(token) + if !errors.Is(err, ErrExpiredToken) { + t.Errorf("Validate() error = %v, want ErrExpiredToken", err) + } +} + +func TestJWT_Validate_InvalidToken(t *testing.T) { + mgr := NewJWTManager("test-secret", 24) + + tests := []struct { + name string + token string + }{ + {"malformed token", "not.a.token"}, + {"empty token", ""}, + {"random string", "abcdef123456"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + _, err := mgr.Validate(tt.token) + if err == nil { + t.Error("Validate() expected error for invalid token") + } + }) + } +} + +func TestJWT_Validate_WrongSecret(t *testing.T) { + mgr1 := NewJWTManager("secret-one", 24) + mgr2 := NewJWTManager("secret-two", 24) + + token, _, err := mgr1.Generate(1, "user", "admin") + if err != nil { + t.Fatalf("Generate() error = %v", err) + } + + _, err = mgr2.Validate(token) + if err == nil { + t.Error("Validate() expected error for token signed with different secret") + } +} + +func TestPassword_HashPassword_RandomSalts(t *testing.T) { + password := "samepassword123" + + hash1, err := HashPassword(password) + if err != nil { + t.Fatalf("HashPassword() error = %v", err) + } + hash2, err := HashPassword(password) + if err != nil { + t.Fatalf("HashPassword() error = %v", err) + } + + if string(hash1) == string(hash2) { + t.Error("HashPassword() produced identical hashes for same password") + } +} + +func TestPassword_CheckPassword_Correct(t *testing.T) { + password := "mysecretpassword" + hash, err := HashPassword(password) + if err != nil { + t.Fatalf("HashPassword() error = %v", err) + } + + if !VerifyPassword(hash, password) { + t.Error("VerifyPassword() returned false for correct password") + } +} + +func TestPassword_CheckPassword_Wrong(t *testing.T) { + password := "mysecretpassword" + wrongPassword := "wrongpassword" + hash, err := HashPassword(password) + if err != nil { + t.Fatalf("HashPassword() error = %v", err) + } + + if VerifyPassword(hash, wrongPassword) { + t.Error("VerifyPassword() returned true for wrong password") + } +} + +func TestMiddleware_RequireAuth_NoToken(t *testing.T) { + InitJWTManager("test-secret", 24) + + handler := RequireAuth(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("next handler should not be called") + })) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + rr := httptest.NewRecorder() + + handler.ServeHTTP(rr, req) + + if rr.Code != http.StatusUnauthorized { + t.Errorf("RequireAuth() status = %d, want %d", rr.Code, http.StatusUnauthorized) + } +} + +func TestMiddleware_RequireAuth_InvalidToken(t *testing.T) { + InitJWTManager("test-secret", 24) + + handler := RequireAuth(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("next handler should not be called") + })) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.AddCookie(&http.Cookie{Name: CookieName, Value: "invalid-token"}) + + rr := httptest.NewRecorder() + + handler.ServeHTTP(rr, req) + + if rr.Code != http.StatusUnauthorized { + t.Errorf("RequireAuth() status = %d, want %d", rr.Code, http.StatusUnauthorized) + } +} + +func TestMiddleware_RequireAuth_ValidToken(t *testing.T) { + secret := "test-secret" + InitJWTManager(secret, 24) + + mgr := GetJWTManager() + token, _, err := mgr.Generate(99, "testuser", "admin") + if err != nil { + t.Fatalf("Generate() error = %v", err) + } + + var capturedClaims *Claims + handler := RequireAuth(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + capturedClaims = GetClaims(r.Context()) + w.WriteHeader(http.StatusOK) + })) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + req.AddCookie(&http.Cookie{Name: CookieName, Value: token}) + + rr := httptest.NewRecorder() + + handler.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Errorf("RequireAuth() status = %d, want %d", rr.Code, http.StatusOK) + } + if capturedClaims == nil { + t.Fatal("RequireAuth() did not set claims in context") + } + if capturedClaims.UserID != 99 { + t.Errorf("capturedClaims.UserID = %d, want 99", capturedClaims.UserID) + } + if capturedClaims.Username != "testuser" { + t.Errorf("capturedClaims.Username = %s, want testuser", capturedClaims.Username) + } + if capturedClaims.Role != "admin" { + t.Errorf("capturedClaims.Role = %s, want admin", capturedClaims.Role) + } +} + +func TestMiddleware_RequireAdmin_NonAdmin(t *testing.T) { + InitJWTManager("test-secret", 24) + + handler := RequireAdmin(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("next handler should not be called for non-admin") + })) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + ctx := context.WithValue(req.Context(), ClaimsCtxKey, &Claims{Role: "user"}) + req = req.WithContext(ctx) + + rr := httptest.NewRecorder() + + handler.ServeHTTP(rr, req) + + if rr.Code != http.StatusForbidden { + t.Errorf("RequireAdmin() status = %d, want %d", rr.Code, http.StatusForbidden) + } +} + +func TestMiddleware_RequireAdmin_NoClaims(t *testing.T) { + InitJWTManager("test-secret", 24) + + handler := RequireAdmin(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + t.Error("next handler should not be called when no claims in context") + })) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + + rr := httptest.NewRecorder() + + handler.ServeHTTP(rr, req) + + if rr.Code != http.StatusForbidden { + t.Errorf("RequireAdmin() status = %d, want %d", rr.Code, http.StatusForbidden) + } +} + +func TestMiddleware_RequireAdmin_Admin(t *testing.T) { + InitJWTManager("test-secret", 24) + + called := false + handler := RequireAdmin(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + called = true + w.WriteHeader(http.StatusOK) + })) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + ctx := context.WithValue(req.Context(), ClaimsCtxKey, &Claims{Role: "admin"}) + req = req.WithContext(ctx) + + rr := httptest.NewRecorder() + + handler.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Errorf("RequireAdmin() status = %d, want %d", rr.Code, http.StatusOK) + } + if !called { + t.Error("RequireAdmin() did not call next handler for admin role") + } +} + +func TestSeed_SeedsAdminUser(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatalf("sql.Open() error = %v", err) + } + defer db.Close() + + _, err = db.Exec(`CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + role TEXT NOT NULL DEFAULT 'user', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP + )`) + if err != nil { + t.Fatalf("CREATE TABLE error = %v", err) + } + + err = SeedAdmin(db, "admin", "secretpassword123") + if err != nil { + t.Fatalf("SeedAdmin() error = %v", err) + } + + var username, role string + err = db.QueryRow("SELECT username, role FROM users WHERE username = 'admin'").Scan(&username, &role) + if err != nil { + t.Fatalf("QueryRow() error = %v", err) + } + if username != "admin" { + t.Errorf("username = %s, want admin", username) + } + if role != "admin" { + t.Errorf("role = %s, want admin", role) + } +} + +func TestSeed_Idempotent(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatalf("sql.Open() error = %v", err) + } + defer db.Close() + + _, err = db.Exec(`CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + role TEXT NOT NULL DEFAULT 'user', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP + )`) + if err != nil { + t.Fatalf("CREATE TABLE error = %v", err) + } + + _, err = db.Exec("INSERT INTO users (username, password_hash, role) VALUES ('admin', 'existing-hash', 'admin')") + if err != nil { + t.Fatalf("INSERT error = %v", err) + } + + err = SeedAdmin(db, "admin", "newpassword") + if err != nil { + t.Fatalf("SeedAdmin() error = %v", err) + } + + var count int + err = db.QueryRow("SELECT COUNT(*) FROM users WHERE username = 'admin'").Scan(&count) + if err != nil { + t.Fatalf("QueryRow() error = %v", err) + } + if count != 1 { + t.Errorf("user count = %d, want 1 (idempotent behavior)", count) + } +} + +func TestSeed_EmptyPasswordError(t *testing.T) { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatalf("sql.Open() error = %v", err) + } + defer db.Close() + + _, err = db.Exec(`CREATE TABLE users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + role TEXT NOT NULL DEFAULT 'user', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP + )`) + if err != nil { + t.Fatalf("CREATE TABLE error = %v", err) + } + + err = SeedAdmin(db, "admin", "") + if err == nil { + t.Error("SeedAdmin() expected error for empty password on first run") + } +} + +func jwtWithExpiry(expiry time.Time) string { + claims := &Claims{ + UserID: 1, + Username: "testuser", + Role: "admin", + RegisteredClaims: jwt.RegisteredClaims{ + ExpiresAt: jwt.NewNumericDate(expiry), + IssuedAt: jwt.NewNumericDate(time.Now().Add(-2 * time.Hour)), + NotBefore: jwt.NewNumericDate(time.Now().Add(-2 * time.Hour)), + }, + } + token := jwt.NewWithClaims(jwt.SigningMethodHS256, claims) + signed, _ := token.SignedString([]byte("test-secret")) + return signed +} diff --git a/internal/auth/middleware.go b/internal/auth/middleware.go index 1869293..d39c883 100644 --- a/internal/auth/middleware.go +++ b/internal/auth/middleware.go @@ -50,6 +50,10 @@ func GetClaims(ctx context.Context) *Claims { return v.(*Claims) } +func WithAuthAdmin(next http.Handler) http.Handler { + return RequireAdmin(RequireAuth(next)) +} + var GlobalJWTManager *JWTManager func InitJWTManager(secret string, expiryH int) { diff --git a/internal/auth/seed.go b/internal/auth/seed.go index 93b84ea..93cb317 100644 --- a/internal/auth/seed.go +++ b/internal/auth/seed.go @@ -2,6 +2,7 @@ package auth import ( "database/sql" + "errors" "log/slog" ) @@ -9,9 +10,6 @@ func SeedAdmin(db *sql.DB, username, password string) error { if username == "" { username = "admin" } - if password == "" { - password = "admin" - } var exists bool err := db.QueryRow("SELECT EXISTS(SELECT 1 FROM users WHERE username = ?)", username).Scan(&exists) @@ -22,6 +20,10 @@ func SeedAdmin(db *sql.DB, username, password string) error { return nil } + if password == "" { + return errors.New("SYNCSERVER_ADMIN_PASSWORD environment variable is required on first run") + } + hash, err := HashPassword(password) if err != nil { return err diff --git a/internal/config/config.go b/internal/config/config.go index 3813ce2..c8f7585 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -30,8 +30,10 @@ type AuthConfig struct { } type SchedulerConfig struct { - Timezone string `yaml:"timezone" env:"SYNCSERVER_SCHEDULER_TZ" default:"UTC"` - RetentionDays int `yaml:"retention_days" env:"SYNCSERVER_RETENTION_DAYS" default:"30"` + Timezone string `yaml:"timezone" env:"SYNCSERVER_SCHEDULER_TZ" default:"UTC"` + RetentionDays int `yaml:"retention_days" env:"SYNCSERVER_RETENTION_DAYS" default:"30"` + BackupDir string `yaml:"backup_dir" env:"SYNCSERVER_BACKUP_DIR"` + BackupRetentionDays int `yaml:"backup_retention_days" env:"SYNCSERVER_BACKUP_RETENTION_DAYS" default:"7"` } var globalCfg *Config diff --git a/internal/db/db.go b/internal/db/db.go index b7231f7..ff1bda1 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -1,7 +1,9 @@ package db import ( + "crypto/sha256" "database/sql" + "encoding/hex" "fmt" "os" "path/filepath" @@ -62,6 +64,7 @@ func (db *DB) runMigrationsInternal(mfs embedFS, migrationsRoot string) error { if _, err := db.Exec(` CREATE TABLE IF NOT EXISTS schema_migrations ( version TEXT PRIMARY KEY, + checksum TEXT, applied_at DATETIME DEFAULT CURRENT_TIMESTAMP ) `); err != nil { @@ -69,13 +72,9 @@ func (db *DB) runMigrationsInternal(mfs embedFS, migrationsRoot string) error { } for _, name := range names { - var applied bool - row := db.QueryRow("SELECT 1 FROM schema_migrations WHERE version = ?", name) - if err := row.Scan(&applied); err == nil { - applied = true - } - - if applied { + var storedChecksum string + row := db.QueryRow("SELECT checksum FROM schema_migrations WHERE version = ?", name) + if err := row.Scan(&storedChecksum); err == nil && storedChecksum != "" { continue } @@ -84,13 +83,30 @@ func (db *DB) runMigrationsInternal(mfs embedFS, migrationsRoot string) error { return fmt.Errorf("reading migration %s: %w", name, err) } - if _, err := db.Exec(string(data)); err != nil { + checksum := sha256.Sum256(data) + checksumHex := hex.EncodeToString(checksum[:]) + + tx, err := db.Begin() + if err != nil { + return fmt.Errorf("starting transaction for migration %s: %w", name, err) + } + + if _, err := tx.Exec(string(data)); err != nil { + tx.Rollback() return fmt.Errorf("applying migration %s: %w", name, err) } - if _, err := db.Exec("INSERT INTO schema_migrations (version) VALUES (?)", name); err != nil { + if _, err := tx.Exec( + "INSERT INTO schema_migrations (version, checksum) VALUES (?, ?)", + name, checksumHex, + ); err != nil { + tx.Rollback() return fmt.Errorf("recording migration %s: %w", name, err) } + + if err := tx.Commit(); err != nil { + return fmt.Errorf("committing migration %s: %w", name, err) + } } return nil diff --git a/internal/models/job.go b/internal/models/job.go index bd258cb..fbd3dbd 100644 --- a/internal/models/job.go +++ b/internal/models/job.go @@ -2,6 +2,8 @@ package models import ( "database/sql" + "fmt" + "strings" "time" ) @@ -192,3 +194,53 @@ func (r *JobRepository) SetTotals(id int64, totalSize, sentBytes int64) error { ) return err } + +func (r *JobRepository) GetByStatusAny(statuses []string) ([]Job, error) { + if len(statuses) == 0 { + return nil, nil + } + placeholders := make([]string, len(statuses)) + args := make([]interface{}, len(statuses)) + for i, s := range statuses { + placeholders[i] = "?" + args[i] = s + } + query := fmt.Sprintf(` + SELECT id, sync_pair_id, trigger_type, status, started_at, finished_at, + log_file, error_message, error_code, total_size_bytes, sent_bytes, created_at + FROM jobs WHERE status IN (%s)`, strings.Join(placeholders, ",")) + rows, err := r.db.Query(query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var jobs []Job + for rows.Next() { + var j Job + var started, finished sql.NullTime + var logFile, errMsg, errCode sql.NullString + if err := rows.Scan(&j.ID, &j.SyncPairID, &j.TriggerType, &j.Status, + &started, &finished, &logFile, &errMsg, &errCode, + &j.TotalSizeBytes, &j.SentBytes, &j.CreatedAt); err != nil { + return nil, err + } + if started.Valid { + j.StartedAt = &started.Time + } + if finished.Valid { + j.FinishedAt = &finished.Time + } + if logFile.Valid { + j.LogFile = &logFile.String + } + if errMsg.Valid { + j.ErrorMessage = &errMsg.String + } + if errCode.Valid { + j.ErrorCode = &errCode.String + } + jobs = append(jobs, j) + } + return jobs, rows.Err() +} diff --git a/internal/models/models_test.go b/internal/models/models_test.go new file mode 100644 index 0000000..15a8ae3 --- /dev/null +++ b/internal/models/models_test.go @@ -0,0 +1,602 @@ +package models + +import ( + "database/sql" + "os" + "testing" + "time" + + _ "modernc.org/sqlite" +) + +func openTestDB(t *testing.T) *sql.DB { + db, err := sql.Open("sqlite", ":memory:") + if err != nil { + t.Fatalf("failed to open in-memory db: %v", err) + } + + migrations := []string{initSchema, migration002, migration003, migration004, migration005, migration006} + for _, m := range migrations { + if _, err := db.Exec(m); err != nil { + t.Fatalf("failed to apply migration: %v", err) + } + } + + return db +} + +const initSchema = ` +CREATE TABLE IF NOT EXISTS users ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + username TEXT UNIQUE NOT NULL, + password_hash TEXT NOT NULL, + role TEXT NOT NULL DEFAULT 'user', + created_at DATETIME DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS ssh_keys ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + label TEXT NOT NULL, + private_key_path TEXT NOT NULL, + public_key TEXT NOT NULL, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS machines ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + host TEXT NOT NULL, + port INTEGER NOT NULL DEFAULT 22, + ssh_user TEXT NOT NULL DEFAULT 'root', + ssh_key_id INTEGER REFERENCES ssh_keys(id), + mac_address TEXT, + wol_enabled INTEGER NOT NULL DEFAULT 0, + broadcast_addr TEXT, + wake_timeout_seconds INTEGER NOT NULL DEFAULT 120, + wake_check_interval_seconds INTEGER NOT NULL DEFAULT 5, + fingerprint_confirmed INTEGER NOT NULL DEFAULT 0, + status TEXT NOT NULL DEFAULT 'unknown', + last_seen_at DATETIME, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS sync_pairs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT NOT NULL, + source_machine_id INTEGER REFERENCES machines(id), + source_path TEXT NOT NULL, + dest_machine_id INTEGER REFERENCES machines(id), + dest_path TEXT NOT NULL, + direction TEXT NOT NULL DEFAULT 'push', + rsync_flags TEXT NOT NULL DEFAULT '-aP', + exclude_patterns TEXT NOT NULL DEFAULT '', + enabled INTEGER NOT NULL DEFAULT 1, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS schedules ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sync_pair_id INTEGER NOT NULL REFERENCES sync_pairs(id) ON DELETE CASCADE, + cron_expr TEXT NOT NULL, + next_run_at DATETIME, + enabled INTEGER NOT NULL DEFAULT 1, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS jobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + sync_pair_id INTEGER NOT NULL REFERENCES sync_pairs(id), + trigger_type TEXT NOT NULL DEFAULT 'manual', + status TEXT NOT NULL DEFAULT 'queued', + started_at DATETIME, + finished_at DATETIME, + log_file TEXT, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS job_logs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + job_id INTEGER NOT NULL REFERENCES jobs(id) ON DELETE CASCADE, + stream TEXT NOT NULL, + content TEXT NOT NULL, + timestamp DATETIME DEFAULT CURRENT_TIMESTAMP +); + +CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + token_hash TEXT NOT NULL, + expires_at DATETIME NOT NULL, + revoked INTEGER NOT NULL DEFAULT 0, + created_at DATETIME DEFAULT CURRENT_TIMESTAMP +); +` + +const migration002 = ` +CREATE INDEX IF NOT EXISTS idx_job_logs_job_id ON job_logs(job_id); +CREATE INDEX IF NOT EXISTS idx_jobs_status_created ON jobs(status, created_at DESC); +CREATE INDEX IF NOT EXISTS idx_jobs_sync_pair_id ON jobs(sync_pair_id); + +CREATE TABLE IF NOT EXISTS cleanup_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + deleted_before DATETIME NOT NULL, + logs_purged INTEGER NOT NULL DEFAULT 0, + jobs_purged INTEGER NOT NULL DEFAULT 0, + executed_at DATETIME DEFAULT CURRENT_TIMESTAMP +); +` + +const migration003 = ` +ALTER TABLE jobs ADD COLUMN error_message TEXT; +ALTER TABLE jobs ADD COLUMN error_code TEXT; +` + +const migration004 = ` +ALTER TABLE machines ADD COLUMN host_key_fingerprint TEXT; +` + +const migration005 = ` +ALTER TABLE machines ADD COLUMN shutdown_command TEXT NOT NULL DEFAULT 'shutdown now'; +` + +const migration006 = ` +ALTER TABLE jobs ADD COLUMN total_size_bytes INTEGER DEFAULT 0; +ALTER TABLE jobs ADD COLUMN sent_bytes INTEGER DEFAULT 0; +` + +func TestJobRepository_CreateAndGetByID(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + + id, err := repo.Create(1, "manual", "queued") + if err != nil { + t.Fatalf("Create failed: %v", err) + } + + job, err := repo.GetByID(id) + if err != nil { + t.Fatalf("GetByID failed: %v", err) + } + + if job.ID != id { + t.Errorf("expected ID %d, got %d", id, job.ID) + } + if job.SyncPairID != 1 { + t.Errorf("expected SyncPairID 1, got %d", job.SyncPairID) + } + if job.TriggerType != "manual" { + t.Errorf("expected trigger_type 'manual', got %q", job.TriggerType) + } + if job.Status != "queued" { + t.Errorf("expected status 'queued', got %q", job.Status) + } +} + +func TestJobRepository_UpdateStatus_SetsStartedAt(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + id, _ := repo.Create(1, "manual", "queued") + + err := repo.UpdateStatus(id, "running") + if err != nil { + t.Fatalf("UpdateStatus failed: %v", err) + } + + job, _ := repo.GetByID(id) + if job.StartedAt == nil { + t.Fatal("expected StartedAt to be set for 'running' status") + } + if job.Status != "running" { + t.Errorf("expected status 'running', got %q", job.Status) + } +} + +func TestJobRepository_UpdateStatus_SetsFinishedAt(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + id, _ := repo.Create(1, "manual", "queued") + repo.UpdateStatus(id, "running") + + err := repo.UpdateStatus(id, "success") + if err != nil { + t.Fatalf("UpdateStatus failed: %v", err) + } + + job, _ := repo.GetByID(id) + if job.FinishedAt == nil { + t.Fatal("expected FinishedAt to be set for 'success' status") + } + if job.Status != "success" { + t.Errorf("expected status 'success', got %q", job.Status) + } +} + +func TestJobRepository_UpdateStatus_Failed(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + id, _ := repo.Create(1, "manual", "queued") + repo.UpdateStatus(id, "running") + + err := repo.UpdateStatus(id, "failed") + if err != nil { + t.Fatalf("UpdateStatus failed: %v", err) + } + + job, _ := repo.GetByID(id) + if job.FinishedAt == nil { + t.Fatal("expected FinishedAt to be set for 'failed' status") + } +} + +func TestJobRepository_SetError(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + id, _ := repo.Create(1, "manual", "queued") + + err := repo.SetError(id, "EIO", "disk read failed") + if err != nil { + t.Fatalf("SetError failed: %v", err) + } + + job, _ := repo.GetByID(id) + if job.ErrorCode == nil || *job.ErrorCode != "EIO" { + t.Errorf("expected error_code 'EIO', got %v", job.ErrorCode) + } + if job.ErrorMessage == nil || *job.ErrorMessage != "disk read failed" { + t.Errorf("expected error_message 'disk read failed', got %v", job.ErrorMessage) + } +} + +func TestJobRepository_GetByStatusAny(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + _, _ = repo.Create(1, "manual", "queued") + id2, _ := repo.Create(1, "manual", "running") + _, _ = repo.Create(1, "manual", "success") + + jobs, err := repo.GetByStatusAny([]string{"running"}) + if err != nil { + t.Fatalf("GetByStatusAny failed: %v", err) + } + if len(jobs) != 1 { + t.Fatalf("expected 1 job, got %d", len(jobs)) + } + if jobs[0].ID != id2 { + t.Errorf("expected job ID %d, got %d", id2, jobs[0].ID) + } + + jobs, _ = repo.GetByStatusAny([]string{"queued", "running"}) + if len(jobs) != 2 { + t.Fatalf("expected 2 jobs, got %d", len(jobs)) + } +} + +func TestJobRepository_DeleteFinishedBefore(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + + id1, _ := repo.Create(1, "manual", "queued") + repo.UpdateStatus(id1, "running") + repo.UpdateStatus(id1, "success") + + db.Exec("UPDATE jobs SET finished_at = datetime('2020-01-01 00:00:00') WHERE id = ?", id1) + + id2, _ := repo.Create(1, "manual", "queued") + repo.UpdateStatus(id2, "running") + repo.UpdateStatus(id2, "success") + + cutoff := time.Now() + deleted, err := repo.DeleteFinishedBefore(cutoff) + if err != nil { + t.Fatalf("DeleteFinishedBefore failed: %v", err) + } + if deleted != 1 { + t.Errorf("expected 1 deleted, got %d", deleted) + } + + _, err = repo.GetByID(id1) + if err != sql.ErrNoRows { + t.Errorf("expected id1 to be deleted") + } + + _, err = repo.GetByID(id2) + if err != nil { + t.Errorf("expected id2 to still exist") + } +} + +func TestJobRepository_SetTotals(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewJobRepository(db) + id, _ := repo.Create(1, "manual", "queued") + + err := repo.SetTotals(id, 1024, 512) + if err != nil { + t.Fatalf("SetTotals failed: %v", err) + } + + job, _ := repo.GetByID(id) + if job.TotalSizeBytes != 1024 { + t.Errorf("expected TotalSizeBytes 1024, got %d", job.TotalSizeBytes) + } + if job.SentBytes != 512 { + t.Errorf("expected SentBytes 512, got %d", job.SentBytes) + } +} + +func TestMachineRepository_CreateAndGetByID(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewMachineRepository(db) + + mac := "AA:BB:CC:DD:EE:FF" + bcast := "192.168.1.255" +WolEnabled := true +WolTimeout := 300 + + machine := &Machine{ + Name: "test-machine", + Host: "192.168.1.10", + Port: 22, + SSHUser: "admin", + MACAddress: &mac, + WoLEnabled: WolEnabled, + BroadcastAddr: &bcast, + WakeTimeoutSeconds: WolTimeout, + WakeCheckIntervalSeconds: 10, + Status: "unknown", + ShutdownCommand: "shutdown -h now", + } + + id, err := repo.Create(machine) + if err != nil { + t.Fatalf("Create failed: %v", err) + } + + retrieved, err := repo.GetByID(id) + if err != nil { + t.Fatalf("GetByID failed: %v", err) + } + + if retrieved.Name != "test-machine" { + t.Errorf("expected name 'test-machine', got %q", retrieved.Name) + } + if retrieved.Host != "192.168.1.10" { + t.Errorf("expected host '192.168.1.10', got %q", retrieved.Host) + } + if retrieved.MACAddress == nil || *retrieved.MACAddress != mac { + t.Errorf("expected MAC %q, got %v", mac, retrieved.MACAddress) + } + if !retrieved.WoLEnabled { + t.Error("expected WoLEnabled to be true") + } + if retrieved.BroadcastAddr == nil || *retrieved.BroadcastAddr != bcast { + t.Errorf("expected broadcast %q, got %v", bcast, retrieved.BroadcastAddr) + } + if retrieved.WakeTimeoutSeconds != WolTimeout { + t.Errorf("expected wake timeout %d, got %d", WolTimeout, retrieved.WakeTimeoutSeconds) + } + if retrieved.ShutdownCommand != "shutdown -h now" { + t.Errorf("expected shutdown command 'shutdown -h now', got %q", retrieved.ShutdownCommand) + } +} + +func TestMachineRepository_UpdateStatus(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewMachineRepository(db) + id, _ := repo.Create(&Machine{Name: "test", Host: "192.168.1.1", Status: "unknown"}) + + err := repo.UpdateStatus(id, "online") + if err != nil { + t.Fatalf("UpdateStatus failed: %v", err) + } + + machine, _ := repo.GetByID(id) + if machine.Status != "online" { + t.Errorf("expected status 'online', got %q", machine.Status) + } + if machine.LastSeenAt == nil { + t.Error("expected LastSeenAt to be set") + } +} + +func TestMachineRepository_GetAll(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewMachineRepository(db) + repo.Create(&Machine{Name: "machine-a", Host: "192.168.1.1", Status: "online"}) + repo.Create(&Machine{Name: "machine-b", Host: "192.168.1.2", Status: "offline"}) + + machines, err := repo.GetAll() + if err != nil { + t.Fatalf("GetAll failed: %v", err) + } + if len(machines) != 2 { + t.Errorf("expected 2 machines, got %d", len(machines)) + } +} + +func TestSyncPairRepository_CreateAndGetByID(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewSyncPairRepository(db) + + sp := &SyncPair{ + Name: "backup-data", + SourcePath: "/data", + DestPath: "/backup", + Direction: "push", + RsyncFlags: "-aP --delete", + ExcludePatterns: "*.tmp\n*.log", + Enabled: true, + } + + id, err := repo.Create(sp) + if err != nil { + t.Fatalf("Create failed: %v", err) + } + + retrieved, err := repo.GetByID(id) + if err != nil { + t.Fatalf("GetByID failed: %v", err) + } + + if retrieved.Name != "backup-data" { + t.Errorf("expected name 'backup-data', got %q", retrieved.Name) + } + if retrieved.ExcludePatterns != "*.tmp\n*.log" { + t.Errorf("expected exclude patterns '*.tmp\\n*.log', got %q", retrieved.ExcludePatterns) + } +} + +func TestSyncPairRepository_ExcludePatternsList(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewSyncPairRepository(db) + + sp := &SyncPair{ + Name: "test-pair", + SourcePath: "/src", + DestPath: "/dst", + ExcludePatterns: "*.tmp\n *.log\n \n*.bak", + } + + id, _ := repo.Create(sp) + retrieved, _ := repo.GetByID(id) + + patterns := retrieved.ExcludePatternsList() + if len(patterns) != 3 { + t.Fatalf("expected 3 patterns, got %d: %v", len(patterns), patterns) + } + if patterns[0] != "*.tmp" { + t.Errorf("expected first pattern '*.tmp', got %q", patterns[0]) + } + if patterns[1] != "*.log" { + t.Errorf("expected second pattern '*.log', got %q", patterns[1]) + } + if patterns[2] != "*.bak" { + t.Errorf("expected third pattern '*.bak', got %q", patterns[2]) + } +} + +func TestSyncPairRepository_GetAll(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewSyncPairRepository(db) + repo.Create(&SyncPair{Name: "pair-a", SourcePath: "/a", DestPath: "/b"}) + repo.Create(&SyncPair{Name: "pair-b", SourcePath: "/c", DestPath: "/d"}) + + pairs, err := repo.GetAll() + if err != nil { + t.Fatalf("GetAll failed: %v", err) + } + if len(pairs) != 2 { + t.Errorf("expected 2 pairs, got %d", len(pairs)) + } +} + +func TestScheduleRepository_CreateAndGetByID(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewScheduleRepository(db) + + nextRun := time.Now().Add(1 * time.Hour) + sched := &Schedule{ + SyncPairID: 1, + CronExpr: "0 0 * * *", + NextRunAt: &nextRun, + Enabled: true, + } + + id, err := repo.Create(sched) + if err != nil { + t.Fatalf("Create failed: %v", err) + } + + retrieved, err := repo.GetByID(id) + if err != nil { + t.Fatalf("GetByID failed: %v", err) + } + + if retrieved.CronExpr != "0 0 * * *" { + t.Errorf("expected cron '0 0 * * *', got %q", retrieved.CronExpr) + } + if !retrieved.Enabled { + t.Error("expected enabled to be true") + } +} + +func TestScheduleRepository_UpdateEnabled(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewScheduleRepository(db) + + nextRun := time.Now().Add(1 * time.Hour) + id, _ := repo.Create(&Schedule{SyncPairID: 1, CronExpr: "0 0 * * *", NextRunAt: &nextRun, Enabled: true}) + + sched, _ := repo.GetByID(id) + sched.Enabled = false + err := repo.Update(sched) + if err != nil { + t.Fatalf("Update failed: %v", err) + } + + retrieved, _ := repo.GetByID(id) + if retrieved.Enabled { + t.Error("expected enabled to be false after update") + } +} + +func TestScheduleRepository_GetEnabledDue(t *testing.T) { + db := openTestDB(t) + defer db.Close() + + repo := NewScheduleRepository(db) + + pastRun := time.Now().Add(-1 * time.Hour) + futureRun := time.Now().Add(1 * time.Hour) + + _, _ = repo.Create(&Schedule{SyncPairID: 1, CronExpr: "0 0 * * *", NextRunAt: &pastRun, Enabled: true}) + _, _ = repo.Create(&Schedule{SyncPairID: 2, CronExpr: "0 0 * * *", NextRunAt: &futureRun, Enabled: true}) + _, _ = repo.Create(&Schedule{SyncPairID: 3, CronExpr: "0 0 * * *", NextRunAt: &pastRun, Enabled: false}) + + due, err := repo.GetEnabledDue(time.Now()) + if err != nil { + t.Fatalf("GetEnabledDue failed: %v", err) + } + if len(due) != 1 { + t.Errorf("expected 1 due schedule, got %d", len(due)) + } + if due[0].SyncPairID != 1 { + t.Errorf("expected sync_pair_id 1, got %d", due[0].SyncPairID) + } +} + +func TestMain(m *testing.M) { + os.Exit(m.Run()) +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 11c03ab..8ac8ced 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -3,7 +3,12 @@ package scheduler import ( "context" "database/sql" + "fmt" "log/slog" + "os" + "path/filepath" + "strconv" + "strings" "sync" "time" @@ -124,6 +129,10 @@ func (s *Scheduler) cleanup() { } before := time.Now().AddDate(0, 0, -retentionDays) + if s.cfg.Scheduler.BackupDir != "" { + s.backupDB(before) + } + logRepo := models.NewJobLogRepository(s.db) jobRepo := models.NewJobRepository(s.db) @@ -141,5 +150,78 @@ func (s *Scheduler) cleanup() { if deletedLogs > 0 || deletedJobs > 0 { slog.Info("cleanup: purged old records", "logs_deleted", deletedLogs, "jobs_deleted", deletedJobs, "before", before.Format("2006-01-02")) + s.purgeJobLogFiles(before) + } + + s.purgeOldBackups() +} + +func (s *Scheduler) backupDB(before time.Time) { + backupDir := s.cfg.Scheduler.BackupDir + if backupDir == "" { + return + } + if err := os.MkdirAll(backupDir, 0700); err != nil { + slog.Error("cleanup: failed to create backup dir", "error", err) + return + } + ts := time.Now().UTC().Format("20060102-150405") + backupPath := filepath.Join(backupDir, fmt.Sprintf("syncserver-%s.db", ts)) + if _, err := s.db.Exec(fmt.Sprintf("VACUUM INTO '%s'", backupPath)); err != nil { + slog.Error("cleanup: failed to vacuum into backup", "path", backupPath, "error", err) + return + } + slog.Info("cleanup: database backup created", "path", backupPath) +} + +func (s *Scheduler) purgeJobLogFiles(before time.Time) { + logsDir := s.cfg.LogsDir() + entries, err := os.ReadDir(logsDir) + if err != nil { + return + } + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".log") { + continue + } + jobID := strings.TrimSuffix(entry.Name(), ".log") + id, err := strconv.ParseInt(jobID, 10, 64) + if err != nil { + continue + } + jobRepo := models.NewJobRepository(s.db) + job, err := jobRepo.GetByID(id) + if err != nil || job == nil { + os.Remove(filepath.Join(logsDir, entry.Name())) + continue + } + if job.FinishedAt != nil && job.FinishedAt.Before(before) { + os.Remove(filepath.Join(logsDir, entry.Name())) + } + } +} + +func (s *Scheduler) purgeOldBackups() { + backupDir := s.cfg.Scheduler.BackupDir + retention := s.cfg.Scheduler.BackupRetentionDays + if backupDir == "" || retention <= 0 { + return + } + cutoff := time.Now().AddDate(0, 0, -retention) + entries, err := os.ReadDir(backupDir) + if err != nil { + return + } + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".db") { + continue + } + info, err := entry.Info() + if err != nil { + continue + } + if info.ModTime().Before(cutoff) { + os.Remove(filepath.Join(backupDir, entry.Name())) + } } } diff --git a/internal/sshmanager/deploy.go b/internal/sshmanager/deploy.go index 9fe1ea2..9a5188e 100644 --- a/internal/sshmanager/deploy.go +++ b/internal/sshmanager/deploy.go @@ -5,7 +5,6 @@ import ( "context" "fmt" "log/slog" - "net" "os" "path/filepath" "time" @@ -25,7 +24,7 @@ type DeployResult struct { Errors []string `json:"errors"` } -func DeployKeysToMachine(ctx context.Context, serverKeyPath string, host string, port int, user string, keys []DeployKey, knownHostsHosts []string) (*DeployResult, error) { +func DeployKeysToMachine(ctx context.Context, serverKeyPath string, host string, port int, user string, keys []DeployKey, knownHostsHosts []string, sshDir string) (*DeployResult, error) { result := &DeployResult{Success: true, Messages: []string{}, Errors: []string{}} addr := fmt.Sprintf("%s:%d", host, port) @@ -39,8 +38,11 @@ func DeployKeysToMachine(ctx context.Context, serverKeyPath string, host string, return nil, fmt.Errorf("parsing server key: %w", err) } - hostKeyCallback := func(hostname string, remote net.Addr, key ssh.PublicKey) error { - return nil + knownHostsPath := filepath.Join(sshDir, "known_hosts") + hostKeyCallback, err := NewKnownHostsCallback(knownHostsPath, true) + if err != nil { + slog.Warn("deploy keys: creating host key callback failed, ignoring hosts", "error", err) + hostKeyCallback = ssh.InsecureIgnoreHostKey() } cfg := &ssh.ClientConfig{ @@ -60,7 +62,7 @@ func DeployKeysToMachine(ctx context.Context, serverKeyPath string, host string, } defer conn.Close() - remoteSSHDir := "/var/lib/syncserver/ssh" + remoteSSHDir := sshDir remoteKeysDir := filepath.Join(remoteSSHDir, "keys") session, err := conn.NewSession() diff --git a/internal/sshmanager/knownhosts.go b/internal/sshmanager/knownhosts.go index 1ac438f..009c246 100644 --- a/internal/sshmanager/knownhosts.go +++ b/internal/sshmanager/knownhosts.go @@ -8,6 +8,7 @@ import ( "strings" "golang.org/x/crypto/ssh" + "golang.org/x/crypto/ssh/knownhosts" ) type KnownHost struct { @@ -17,6 +18,23 @@ type KnownHost struct { Fingerprint string } +func NewKnownHostsCallback(knownHostsPath string, strictHostKeyChecking bool) (ssh.HostKeyCallback, error) { + if !strictHostKeyChecking { + return ssh.InsecureIgnoreHostKey(), nil + } + if knownHostsPath == "" { + return ssh.InsecureIgnoreHostKey(), nil + } + _, err := os.Stat(knownHostsPath) + if os.IsNotExist(err) { + return ssh.InsecureIgnoreHostKey(), nil + } + if err != nil { + return nil, err + } + return knownhosts.New(knownHostsPath) +} + func EnsureKnownHosts(sshDir string) (string, error) { path := filepath.Join(sshDir, "known_hosts") f, err := os.OpenFile(path, os.O_CREATE|os.O_RDONLY, 0644) @@ -27,6 +45,8 @@ func EnsureKnownHosts(sshDir string) (string, error) { return path, nil } +// AddKnownHost stores the host key in standard ssh known_hosts format (hostname keytype base64key). +// NOTE: Existing entries in known_hosts may need to be regenerated if they were stored in a different format. func AddKnownHost(sshDir, host string, port int, keyData []byte) error { path := filepath.Join(sshDir, "known_hosts") f, err := os.OpenFile(path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644) diff --git a/internal/sshmanager/testconn.go b/internal/sshmanager/testconn.go index 3b2effc..5c39aec 100644 --- a/internal/sshmanager/testconn.go +++ b/internal/sshmanager/testconn.go @@ -8,7 +8,6 @@ import ( "fmt" "net" "os" - "path/filepath" "strings" "time" @@ -40,24 +39,17 @@ func dialSSH(ctx context.Context, host string, port int, user, privKeyPath, know var capturedFingerprint string var capturedPubKey ssh.PublicKey + + callback, err := NewKnownHostsCallback(knownHostsPath, strictHostKeyChecking) + if err != nil { + return nil, "", nil, fmt.Errorf("creating host key callback: %w", err) + } + hostKeyCallback := func(hostname string, remote net.Addr, key ssh.PublicKey) error { h := sha256.Sum256(key.Marshal()) capturedFingerprint = "SHA256:" + base64.RawStdEncoding.EncodeToString(h[:]) capturedPubKey = key - if strictHostKeyChecking && knownHostsPath != "" { - kh, err := GetKnownHost(filepath.Dir(knownHostsPath), host, port) - if err != nil { - return fmt.Errorf("checking known_hosts: %w", err) - } - if kh == nil { - return fmt.Errorf("host key not found in known_hosts: %s", hostname) - } - wantFP := kh.Fingerprint - if capturedFingerprint != wantFP { - return fmt.Errorf("host key mismatch: got %s, want %s", capturedFingerprint, wantFP) - } - } - return nil + return callback(hostname, remote, key) } cfg := &ssh.ClientConfig{ diff --git a/internal/syncengine/engine.go b/internal/syncengine/engine.go index 0957182..e0e0f5d 100644 --- a/internal/syncengine/engine.go +++ b/internal/syncengine/engine.go @@ -18,26 +18,31 @@ import ( ) type Engine struct { - db *sql.DB - cfg *config.Config - queue *Queue - eventBus *EventBus - mu sync.RWMutex - stopped bool - lastProbeAt atomic.Int64 + db *sql.DB + cfg *config.Config + queue *Queue + eventBus *EventBus + mu sync.RWMutex + stopped bool + lastProbeAt atomic.Int64 + jobsWG sync.WaitGroup + stopCh chan struct{} + + jobsTotal map[string]int64 + jobsTotalMu sync.Mutex } type Event struct { - Type string - JobID int64 - MachineID int64 - Key string - Value string - Line string - Stream string - Progress *ProgressFields - TotalBytes int64 - SentBytes int64 + Type string `json:"type"` + JobID int64 `json:"job_id"` + MachineID int64 `json:"machine_id"` + Key string `json:"key,omitempty"` + Value string `json:"value,omitempty"` + Line string `json:"line,omitempty"` + Stream string `json:"stream,omitempty"` + Progress *ProgressFields `json:"progress,omitempty"` + TotalBytes int64 `json:"total_bytes,omitempty"` + SentBytes int64 `json:"sent_bytes,omitempty"` } func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine { @@ -46,12 +51,56 @@ func New(database interface{ SQLDB() *sql.DB }, cfg *config.Config) *Engine { cfg: cfg, queue: NewQueue(), eventBus: NewEventBus(200), + stopCh: make(chan struct{}), + jobsTotal: map[string]int64{ + "success": 0, + "failed": 0, + "cancelled": 0, + }, } return e } -func (e *Engine) Start() {} -func (e *Engine) Stop() {} +func (e *Engine) Start() { + e.recoverOrphanedJobs() + slog.Info("engine started") +} + +func (e *Engine) Stop() { + e.mu.Lock() + if e.stopped { + e.mu.Unlock() + return + } + e.stopped = true + e.mu.Unlock() + + close(e.stopCh) + + runningIDs := e.queue.RunningJobs() + for _, id := range runningIDs { + e.queue.Cancel(id, false) + } + + e.jobsWG.Wait() + slog.Info("engine stopped") +} + +func (e *Engine) recoverOrphanedJobs() { + jobRepo := models.NewJobRepository(e.db) + jobs, err := jobRepo.GetByStatusAny([]string{"queued", "waking_up", "running"}) + if err != nil { + slog.Warn("failed to recover orphaned jobs", "error", err) + return + } + for _, j := range jobs { + slog.Warn("recovered orphaned job, marking as failed", + "job_id", j.ID, "pair_id", j.SyncPairID, "status", j.Status) + jobRepo.UpdateStatus(j.ID, "failed") + jobRepo.SetError(j.ID, "crash_recovery", + fmt.Sprintf("job was %s when server shut down unexpectedly", j.Status)) + } +} func (e *Engine) SubscribeJob(jobID int64) (chan Event, func()) { return e.eventBus.Subscribe(jobID) @@ -84,9 +133,12 @@ func (e *Engine) wakeMachine(ctx context.Context, m *models.Machine) { } func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { - if e.queue.IsRunning(pairID) { - existingJobID, _ := e.queue.GetJobID(pairID) - return fmt.Errorf("%w: job %d is already running", ErrAlreadyRunning, existingJobID) + if e.queue.IsRunning(jobID) { + return fmt.Errorf("%w: job %d is already running", ErrAlreadyRunning, jobID) + } + + if existingJobID, exists := e.queue.GetByPair(pairID); exists { + return fmt.Errorf("%w: job %d is already running for this sync pair", ErrAlreadyRunning, existingJobID) } jobCtx, cancel := context.WithCancel(ctx) @@ -99,7 +151,10 @@ func (e *Engine) Run(ctx context.Context, jobID int64, pairID int64) error { if enqueueErr != nil { return enqueueErr } - defer e.queue.Dequeue(pairID) + defer e.queue.Dequeue(jobID) + + e.jobsWG.Add(1) + defer e.jobsWG.Done() pairRepo := models.NewSyncPairRepository(e.db) pair, err := pairRepo.GetByID(pairID) @@ -381,17 +436,18 @@ func (e *Engine) resolveSSHKey(machine *models.Machine) (string, error) { return sshKey.PrivateKeyPath, nil } -func (e *Engine) Cancel(jobID int64, syncPairID int64, byUser bool) bool { - if e.queue.IsRunning(syncPairID) { - e.queue.Cancel(syncPairID, byUser) - return true - } - return false +func (e *Engine) Cancel(jobID int64, byUser bool) bool { + return e.queue.Cancel(jobID, byUser) } func (e *Engine) setJobStatus(jobID int64, status string) { jobRepo := models.NewJobRepository(e.db) jobRepo.UpdateStatus(jobID, status) + if status == "success" || status == "failed" || status == "cancelled" { + e.jobsTotalMu.Lock() + e.jobsTotal[status]++ + e.jobsTotalMu.Unlock() + } } func (e *Engine) setJobLogFile(jobID int64, path string) { @@ -495,3 +551,36 @@ func (e *Engine) ProbeAllMachines() { } wg.Wait() } + +func (e *Engine) GetJobsTotal() map[string]int64 { + e.jobsTotalMu.Lock() + defer e.jobsTotalMu.Unlock() + return e.jobsTotal +} + +func (e *Engine) GetJobsRunning() int64 { + return int64(len(e.queue.RunningJobs())) +} + +func (e *Engine) GetQueueDepth() int64 { + return int64(e.queue.Len()) +} + +func (e *Engine) GetMachineCounts() (online, total int64) { + machineRepo := models.NewMachineRepository(e.db) + ms, err := machineRepo.GetAll() + if err != nil { + return 0, 0 + } + for _, m := range ms { + total++ + if m.Status == "online" { + online++ + } + } + return online, total +} + +func (e *Engine) DB() *sql.DB { + return e.db +} diff --git a/internal/syncengine/eventbus.go b/internal/syncengine/eventbus.go index 486850e..685c713 100644 --- a/internal/syncengine/eventbus.go +++ b/internal/syncengine/eventbus.go @@ -8,20 +8,17 @@ import ( type EventBus struct { subscribers map[int64]map[chan Event]struct{} mu sync.RWMutex - global chan Event bufferSize int globalSubs []globalSub } type globalSub struct { - ch chan Event - done chan struct{} + ch chan Event } func NewEventBus(bufferSize int) *EventBus { return &EventBus{ subscribers: make(map[int64]map[chan Event]struct{}), - global: make(chan Event, bufferSize), bufferSize: bufferSize, globalSubs: nil, } @@ -51,32 +48,10 @@ func (eb *EventBus) Subscribe(jobID int64) (chan Event, func()) { func (eb *EventBus) SubscribeGlobal() (chan Event, func()) { ch := make(chan Event, eb.bufferSize) - done := make(chan struct{}) eb.mu.Lock() - eb.globalSubs = append(eb.globalSubs, globalSub{ch: ch, done: done}) + eb.globalSubs = append(eb.globalSubs, globalSub{ch: ch}) eb.mu.Unlock() - go func() { - defer func() { - if r := recover(); r != nil { - slog.Error("SubscribeGlobal goroutine panicked", "reason", r) - } - close(ch) - }() - for { - select { - case evt := <-eb.global: - select { - case ch <- evt: - default: - slog.Warn("global event subscriber buffer full, dropping event", "type", evt.Type) - } - case <-done: - return - } - } - }() return ch, func() { - close(done) eb.mu.Lock() for i, s := range eb.globalSubs { if s.ch == ch { @@ -85,6 +60,7 @@ func (eb *EventBus) SubscribeGlobal() (chan Event, func()) { } } eb.mu.Unlock() + close(ch) } } @@ -102,10 +78,12 @@ func (eb *EventBus) Publish(evt Event) { } } - select { - case eb.global <- evt: - default: - slog.Warn("global event bus full, dropping event", "type", evt.Type) + for _, sub := range eb.globalSubs { + select { + case sub.ch <- evt: + default: + slog.Warn("global event subscriber buffer full, dropping event", "type", evt.Type) + } } } diff --git a/internal/syncengine/parser.go b/internal/syncengine/parser.go index 75a11ae..a1c18ea 100644 --- a/internal/syncengine/parser.go +++ b/internal/syncengine/parser.go @@ -16,21 +16,21 @@ type RsyncStats struct { } type ProgressLine struct { - Phase string - Percent float64 - Files int - Total int - SentBytes int64 + Phase string + Percent float64 + Files int + Total int + SentBytes int64 XferedBytes int64 } type ProgressFields struct { - FileBytes int64 - Pct int - SpeedBps int64 - EtaSeconds int - XfrDone int - XfrTotal int + FileBytes int64 `json:"file_bytes"` + Pct int `json:"pct"` + SpeedBps int64 `json:"speed_bps"` + EtaSeconds int `json:"eta_seconds"` + XfrDone int `json:"xfr_done"` + XfrTotal int `json:"xfr_total"` } var ( @@ -138,9 +138,9 @@ func parseProgressFields(line string) *ProgressFields { } pf := &ProgressFields{ - FileBytes: bytes, - Pct: pct, - SpeedBps: speedBps, + FileBytes: bytes, + Pct: pct, + SpeedBps: speedBps, EtaSeconds: etaSecs, } diff --git a/internal/syncengine/queue.go b/internal/syncengine/queue.go index 726082b..9ca1fec 100644 --- a/internal/syncengine/queue.go +++ b/internal/syncengine/queue.go @@ -13,60 +13,85 @@ type Queue struct { } type RunInfo struct { - JobID int64 - Cancel func() - ByUser bool + JobID int64 + SyncPairID int64 + Cancel func() + CancelledBy bool } func NewQueue() *Queue { return &Queue{runs: make(map[int64]*RunInfo)} } -func (q *Queue) Enqueue(syncPairID, jobID int64, cancel func()) error { +func (q *Queue) Enqueue(syncPairID, jobID int64, cancelFn func()) error { q.mu.Lock() defer q.mu.Unlock() - if _, exists := q.runs[syncPairID]; exists { + if _, exists := q.runs[jobID]; exists { return ErrAlreadyRunning } - q.runs[syncPairID] = &RunInfo{JobID: jobID, Cancel: cancel} + for _, info := range q.runs { + if info.SyncPairID == syncPairID { + return ErrAlreadyRunning + } + } + q.runs[jobID] = &RunInfo{JobID: jobID, SyncPairID: syncPairID, Cancel: cancelFn} return nil } -func (q *Queue) Dequeue(syncPairID int64) { +func (q *Queue) Dequeue(jobID int64) { q.mu.Lock() defer q.mu.Unlock() - delete(q.runs, syncPairID) + delete(q.runs, jobID) } -func (q *Queue) IsRunning(syncPairID int64) bool { +func (q *Queue) IsRunning(jobID int64) bool { q.mu.Lock() defer q.mu.Unlock() - _, exists := q.runs[syncPairID] + _, exists := q.runs[jobID] return exists } -func (q *Queue) GetJobID(syncPairID int64) (int64, bool) { +func (q *Queue) GetByPair(syncPairID int64) (jobID int64, exists bool) { q.mu.Lock() defer q.mu.Unlock() - info, exists := q.runs[syncPairID] - if !exists { - return 0, false + for _, info := range q.runs { + if info.SyncPairID == syncPairID { + return info.JobID, true + } } - return info.JobID, true + return 0, false } -func (q *Queue) Cancel(syncPairID int64, byUser bool) { +func (q *Queue) Cancel(jobID int64, byUser bool) bool { q.mu.Lock() defer q.mu.Unlock() - if info, exists := q.runs[syncPairID]; exists && info.Cancel != nil { - info.ByUser = byUser + if info, exists := q.runs[jobID]; exists && info.Cancel != nil { + info.CancelledBy = byUser info.Cancel() + return true } + return false } -func (q *Queue) IsCancelledByUser(syncPairID int64) bool { +func (q *Queue) IsCancelledByUser(jobID int64) bool { q.mu.Lock() defer q.mu.Unlock() - info, exists := q.runs[syncPairID] - return exists && info.ByUser + info, exists := q.runs[jobID] + return exists && info.CancelledBy +} + +func (q *Queue) RunningJobs() []int64 { + q.mu.Lock() + defer q.mu.Unlock() + ids := make([]int64, 0, len(q.runs)) + for id := range q.runs { + ids = append(ids, id) + } + return ids +} + +func (q *Queue) Len() int { + q.mu.Lock() + defer q.mu.Unlock() + return len(q.runs) } diff --git a/internal/syncengine/queue_test.go b/internal/syncengine/queue_test.go index 5711eed..fdfcc51 100644 --- a/internal/syncengine/queue_test.go +++ b/internal/syncengine/queue_test.go @@ -7,7 +7,7 @@ import ( func TestQueue(t *testing.T) { q := NewQueue() - if q.IsRunning(1) { + if q.IsRunning(100) { t.Error("queue should be empty") } @@ -16,35 +16,37 @@ func TestQueue(t *testing.T) { err := q.Enqueue(1, 100, cancel) if err != nil { - t.Errorf("Enqueue(1) unexpected error: %v", err) + t.Errorf("Enqueue(1, 100) unexpected error: %v", err) } - if !q.IsRunning(1) { - t.Error("queue should contain syncPair 1") + if !q.IsRunning(100) { + t.Error("queue should contain job 100") } - jobID, ok := q.GetJobID(1) + jobID, ok := q.GetByPair(1) if !ok || jobID != 100 { - t.Errorf("GetJobID(1) = %d, %v, want 100, true", jobID, ok) + t.Errorf("GetByPair(1) = %d, %v, want 100, true", jobID, ok) } err = q.Enqueue(1, 200, nil) if err != ErrAlreadyRunning { - t.Errorf("Enqueue(1) again = %v, want ErrAlreadyRunning", err) + t.Errorf("Enqueue(1, 200) = %v, want ErrAlreadyRunning", err) } - q.Cancel(1, true) + ok = q.Cancel(100, true) + if !ok { + t.Error("Cancel(100) should return true") + } if !cancelCalled { t.Error("Cancel should have called the cancel func") } - if !q.IsCancelledByUser(1) { - t.Error("IsCancelledByUser should return true after Cancel(1, true)") + if !q.IsCancelledByUser(100) { + t.Error("IsCancelledByUser should return true after Cancel(100, true)") } - q.Dequeue(1) + q.Dequeue(100) - q.Dequeue(1) - if q.IsRunning(1) { + if q.IsRunning(100) { t.Error("queue should be empty after Dequeue") } } diff --git a/internal/syncengine/rsync_runner.go b/internal/syncengine/rsync_runner.go index 259e1fe..e2dcf2c 100644 --- a/internal/syncengine/rsync_runner.go +++ b/internal/syncengine/rsync_runner.go @@ -4,11 +4,97 @@ import ( "context" "fmt" "io" + "log/slog" "os/exec" "path/filepath" + "strconv" "strings" ) +var allowedRsyncFlags = map[string]bool{ + "-v": true, + "-vv": true, + "-q": true, + "-h": true, + "-P": true, + "-n": true, + "-z": true, + "-c": true, + "-u": true, + "-W": true, + "-i": true, + "-a": true, + "-r": true, + "-l": true, + "-t": true, + "-p": true, + "-g": true, + "-o": true, + "-D": true, + "--verbose": true, + "--quiet": true, + "--help": true, + "--partial": true, + "--partial-dir": true, + "--delay-updates": true, + "--delete": true, + "--delete-before": true, + "--delete-after": true, + "--delete-excluded": true, + "--exclude": true, + "--exclude-from": true, + "--dry-run": true, + "--compress": true, + "--skip-compress": true, + "--whole-file": true, + "--checksum": true, + "--update": true, + "--existing": true, + "--ignore-existing": true, + "--remove-source-files": true, + "--chmod": true, + "--owner": true, + "--group": true, + "--perms": true, + "--executability": true, + "--acls": true, + "--xattrs": true, + "--numeric-ids": true, + "--fake-super": true, + "--bwlimit": true, + "--max-size": true, + "--min-size": true, + "--append": true, + "--append-verify": true, + "--itemize-changes": true, +} + +var blockedRsyncFlags = map[string]bool{ + "--rsync-path": true, + "-e": true, + "--files-from": true, + "--read-batch": true, + "--write-batch": true, + "--log-file": true, +} + +func isSafeRsyncFlag(flag string) bool { + if allowedRsyncFlags[flag] { + return true + } + safePrefixes := []string{ + "-a", "-v", "-z", "-P", "-n", "-c", "-u", "-W", "-i", + "--exclude=", "--chmod=", "--bwlimit=", "--max-size=", "--min-size=", + "--partial-dir=", "--skip-compress=", + } + for _, p := range safePrefixes { + if strings.HasPrefix(flag, p) { + return true + } + } + return false +} + type RsyncResult struct { ExitCode int Stdout string @@ -40,13 +126,13 @@ func NewRsyncRunner(sshDir, privKey string) *RsyncRunner { } func (r *RsyncRunner) Run(ctx context.Context, pair *SyncPairConfig, onLine func(stream string, line string)) (*RsyncResult, error) { - cmd := r.buildRsyncCmd(pair) + cmd := r.buildRsyncCmd(ctx, pair) return r.runCmd(ctx, cmd, onLine) } -func (r *RsyncRunner) buildRsyncCmd(pair *SyncPairConfig) *exec.Cmd { +func (r *RsyncRunner) buildRsyncCmd(ctx context.Context, pair *SyncPairConfig) *exec.Cmd { args := r.buildArgs(pair) - cmd := exec.CommandContext(context.Background(), "rsync", args...) + cmd := exec.CommandContext(ctx, "rsync", args...) if r.privKey != "" { sshCmd := fmt.Sprintf("ssh -i %s -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile=%s", r.privKey, strings.TrimRight(r.sshDir, "/")+"/known_hosts") @@ -59,7 +145,19 @@ func (r *RsyncRunner) buildArgs(pair *SyncPairConfig) []string { var args []string flags := strings.Fields(pair.RsyncFlags) - args = append(args, flags...) + for _, flag := range flags { + if strings.HasPrefix(flag, "-") { + if blockedRsyncFlags[flag] { + slog.Warn("blocked dangerous rsync flag", "flag", flag) + continue + } + if !isSafeRsyncFlag(flag) { + slog.Warn("disallowed rsync flag", "flag", flag) + continue + } + } + args = append(args, flag) + } for _, pattern := range pair.ExcludePatterns { args = append(args, "--exclude="+pattern) @@ -69,6 +167,7 @@ func (r *RsyncRunner) buildArgs(pair *SyncPairConfig) []string { args = append(args, "--delete") } + args = append(args, "--") src := ensureDirSlash(pair.Source) if pair.Direction == "pull" { args = append(args, pair.Dest, src) @@ -113,12 +212,23 @@ func (r *RsyncRunner) RunRemote(ctx context.Context, pair *SyncPairConfig, src * innerSSH := fmt.Sprintf("ssh -i %s -o StrictHostKeyChecking=accept-new -o UserKnownHostsFile=%s", destKey, filepath.Join(r.sshDir, "known_hosts")) - rsyncFlags := strings.Join(args[:len(args)-2], " ") + rsyncFlags := args[:len(args)-2] sourcePath := args[len(args)-2] destPath := args[len(args)-1] - remoteCmd := fmt.Sprintf("rsync %s -e %q %s %s", - rsyncFlags, innerSSH, sourcePath, destPath) + var rsyncCmd []string + rsyncCmd = append(rsyncCmd, "rsync") + rsyncCmd = append(rsyncCmd, "-e") + rsyncCmd = append(rsyncCmd, innerSSH) + rsyncCmd = append(rsyncCmd, rsyncFlags...) + rsyncCmd = append(rsyncCmd, sourcePath, destPath) + + remoteCmd := "rsync" + for _, arg := range rsyncFlags { + remoteCmd += " " + strconv.Quote(arg) + } + remoteCmd += " -e " + strconv.Quote(innerSSH) + " " + strconv.Quote(sourcePath) + " " + strconv.Quote(destPath) + remoteCmd = "sh -c " + strconv.Quote(remoteCmd) sshArgs := []string{ "-i", src.PrivKey, diff --git a/internal/syncengine/rsync_runner_test.go b/internal/syncengine/rsync_runner_test.go index 29fbb71..498514b 100644 --- a/internal/syncengine/rsync_runner_test.go +++ b/internal/syncengine/rsync_runner_test.go @@ -1,6 +1,7 @@ package syncengine import ( + "context" "strings" "testing" ) @@ -281,7 +282,7 @@ func TestRun_FlagsPreservedWithPrivKey(t *testing.T) { ExcludePatterns: []string{}, } - cmd := runner.buildRsyncCmd(pair) + cmd := runner.buildRsyncCmd(context.Background(), pair) if cmd.Args[0] != "rsync" { t.Errorf("cmd.Args[0] = %q, want 'rsync'", cmd.Args[0]) diff --git a/web/src/App.tsx b/web/src/App.tsx index a71eef5..95373a1 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -6,6 +6,7 @@ import { NavLink, Outlet, useNavigate, + Link, } from 'react-router-dom'; import { useState, useEffect } from 'react'; import { Toaster } from 'sonner'; @@ -20,6 +21,8 @@ import { LogOut, Menu, X, + Clock, + ArrowLeft, } from 'lucide-react'; import { ErrorBoundary } from './components/ErrorBoundary'; import { Spinner } from './components/ui/Spinner'; @@ -34,11 +37,13 @@ import JobHistory from './pages/JobHistory'; import JobDetail from './pages/JobDetail'; import SettingsPage from './pages/Settings'; import SSHKeys from './pages/SSHKeys'; +import Schedules from './pages/Schedules'; const navItems = [ { to: '/', label: 'Dashboard', icon: LayoutDashboard }, { to: '/machines', label: 'Machines', icon: Server }, { to: '/sync-pairs', label: 'Sync Pairs', icon: GitCompare }, + { to: '/schedules', label: 'Schedules', icon: Clock }, { to: '/jobs', label: 'Jobs', icon: History }, { to: '/ssh-keys', label: 'SSH Keys', icon: Key }, { to: '/settings', label: 'Settings', icon: SettingsIcon }, @@ -217,6 +222,21 @@ function Layout() { ); } +function NotFound() { + return ( +
+

404

+

Page not found

+ +
+ ); +} + export default function App() { return ( @@ -233,12 +253,13 @@ export default function App() { } /> } /> } /> + } /> } /> } /> } /> } /> - } /> + } /> diff --git a/web/src/api/client.ts b/web/src/api/client.ts index ae0a7fa..0855e3f 100644 --- a/web/src/api/client.ts +++ b/web/src/api/client.ts @@ -143,6 +143,16 @@ export interface ShutdownResponse { error?: string; } +export interface Schedule { + id: number; + sync_pair_id: number; + sync_pair_name: string; + cron_expr: string; + next_run_at: string | null; + enabled: boolean; + created_at: string; +} + export async function shutdownMachine(machineId: number): Promise { return api(`/api/machines/${machineId}/shutdown`, { method: 'POST', diff --git a/web/src/pages/Dashboard.tsx b/web/src/pages/Dashboard.tsx index aac268c..979c03a 100644 --- a/web/src/pages/Dashboard.tsx +++ b/web/src/pages/Dashboard.tsx @@ -1,6 +1,6 @@ import { useEffect, useState } from 'react'; import { Link } from 'react-router-dom'; -import { Server, Activity, HardDrive, Clock, Plus } from 'lucide-react'; +import { Server, Activity, HardDrive, Clock, Plus, XCircle } from 'lucide-react'; import { api, Machine, Job, SyncPair } from '../api/client'; import { Badge } from '@/components/ui/Badge'; import { Button } from '@/components/ui/Button'; @@ -12,6 +12,7 @@ import { Skeleton } from '@/components/ui/Skeleton'; import { statusVariant, statusLabel } from '@/lib/status'; import { formatRelativeTime } from '@/lib/utils'; import { subscribeMachineStatus } from '@/lib/sse'; +import { toast } from 'sonner'; export default function Dashboard() { const [machines, setMachines] = useState([]); @@ -48,6 +49,17 @@ export default function Dashboard() { const pairName = (id: number) => pairs.find(p => p.id === id)?.name ?? `Pair ${id}`; + async function cancelJob(jobId: number) { + try { + await api(`/api/jobs/${jobId}/cancel`, { method: 'POST' }); + toast.success('Job cancelled'); + const j = await api(`/api/jobs?limit=5`); + setJobs(j); + } catch (e: unknown) { + toast.error((e as Error).message); + } + } + const online = machines.filter(m => m.status.startsWith('online')).length; const todayJobs = jobs.filter(j => { if (!j.started_at) return false; @@ -156,6 +168,7 @@ export default function Dashboard() { Sync Pair Status Started + Actions @@ -178,6 +191,19 @@ export default function Dashboard() { {j.started_at ? formatRelativeTime(j.started_at) : '-'} + + {['queued', 'waking_up', 'running'].includes(j.status) && ( + + )} + ))} diff --git a/web/src/pages/JobDetail.tsx b/web/src/pages/JobDetail.tsx index 37274fd..d66bab7 100644 --- a/web/src/pages/JobDetail.tsx +++ b/web/src/pages/JobDetail.tsx @@ -1,6 +1,6 @@ import { useEffect, useState, useRef, useCallback } from 'react'; import { useParams, Link } from 'react-router-dom'; -import { api } from '../api/client'; +import { api, apiRaw } from '../api/client'; import type { Job, LogLine, SyncPair } from '../api/client'; import { Badge } from '@/components/ui/Badge'; import { Button } from '@/components/ui/Button'; @@ -37,12 +37,12 @@ import { formatDuration } from '@/lib/utils'; import { cn } from '@/lib/utils'; interface SSEProgress { - fileBytes: number; + file_bytes: number; pct: number; - speedBps: number; - etaSeconds: number; - xfrDone: number; - xfrTotal: number; + speed_bps: number; + eta_seconds: number; + xfr_done: number; + xfr_total: number; } interface SSEEvent { @@ -157,12 +157,33 @@ export default function JobDetail() { } } - function downloadLog() { +async function downloadLog() { + try { + const resp = await apiRaw(`/api/jobs/${id}/log/download`); + if (resp.ok && resp.body) { + const blob = await resp.blob(); + const url = URL.createObjectURL(blob); + const a = document.createElement('a'); + a.href = url; + a.download = `job-${id}.log`; + document.body.appendChild(a); + a.click(); + document.body.removeChild(a); + URL.revokeObjectURL(url); + } else { + fallbackDownload(); + } + } catch { + fallbackDownload(); + } + } + + function fallbackDownload() { const allLines = [ ...logs.map(l => `[${l.timestamp}] [${l.stream}] ${l.content}`), ...liveLines.map(l => `[LIVE] [${l.stream}] ${l.text}`), ]; - const blob = new Blob([allLines.join('\n')], { type: 'text/plain' }); + const blob = new Blob(allLines as string[], { type: 'text/plain' }); const url = URL.createObjectURL(blob); const a = document.createElement('a'); a.href = url; @@ -553,13 +574,13 @@ function TransferProgress({
- xfr#{(progress.xfrDone).toLocaleString()}/{progress.xfrTotal > 0 ? progress.xfrTotal.toLocaleString() : '?'} + xfr#{(progress.xfr_done).toLocaleString()}/{progress.xfr_total > 0 ? progress.xfr_total.toLocaleString() : '?'} - {formatSpeed(progress.speedBps)} + {formatSpeed(progress.speed_bps)} - ETA {progress.etaSeconds > 0 ? `${Math.floor(progress.etaSeconds / 60)}m ${progress.etaSeconds % 60}s` : '-'} + ETA {progress.eta_seconds > 0 ? `${Math.floor(progress.eta_seconds / 60)}m ${progress.eta_seconds % 60}s` : '-'}
diff --git a/web/src/pages/Schedules.tsx b/web/src/pages/Schedules.tsx new file mode 100644 index 0000000..766ee81 --- /dev/null +++ b/web/src/pages/Schedules.tsx @@ -0,0 +1,334 @@ +import { useEffect, useState } from 'react'; +import { api, SyncPair, Schedule } from '../api/client'; +import { Button } from '@/components/ui/Button'; +import { Input } from '@/components/ui/Input'; +import { Label } from '@/components/ui/Label'; +import { Select, SelectValue, SelectTrigger, SelectContent, SelectItem } from '@/components/ui/Select'; +import { + Modal, + ModalContent, + ModalHeader, + ModalTitle, + ModalDescription, + ModalBody, + ModalFooter, +} from '@/components/ui/Modal'; +import { PageHeader } from '@/components/ui/PageHeader'; +import { + Table, + TableHeader, + TableBody, + TableHead, + TableRow, + TableCell, +} from '@/components/ui/Table'; +import { EmptyState } from '@/components/ui/EmptyState'; +import { Badge } from '@/components/ui/Badge'; +import { Card } from '@/components/ui/Card'; +import { Trash2, Plus, Clock, Pencil, AlertTriangle } from 'lucide-react'; +import { toast } from 'sonner'; +import { cn } from '@/lib/utils'; + +type ScheduleForm = { + id: number | undefined; + sync_pair_id: number | null; + cron_expr: string; + enabled: boolean; +}; + +const defaultForm: ScheduleForm = { + id: undefined, + sync_pair_id: null, + cron_expr: '', + enabled: true, +}; + +export default function Schedules() { + const [schedules, setSchedules] = useState([]); + const [syncPairs, setSyncPairs] = useState([]); + const [modalOpen, setModalOpen] = useState(false); + const [deleteId, setDeleteId] = useState(null); + const [form, setForm] = useState(defaultForm); + const [loading, setLoading] = useState(false); + + useEffect(() => { + load(); + }, []); + + async function load() { + try { + const [s, p] = await Promise.all([ + api('/api/schedules'), + api('/api/sync-pairs'), + ]); + setSchedules(s); + setSyncPairs(p); + } catch {} + } + + function openCreate() { + setForm(defaultForm); + setModalOpen(true); + } + + async function handleSubmit(e: React.FormEvent) { + e.preventDefault(); + if (!form.sync_pair_id) { + toast.error('Sync pair is required'); + return; + } + if (!form.cron_expr.trim()) { + toast.error('Cron expression is required'); + return; + } + setLoading(true); + try { + await api(form.id ? `/api/schedules/${form.id}` : '/api/schedules', { + method: form.id ? 'PUT' : 'POST', + body: { + sync_pair_id: form.sync_pair_id, + cron_expr: form.cron_expr, + enabled: form.enabled, + }, + }); + setModalOpen(false); + toast.success(form.id ? 'Schedule updated' : 'Schedule created'); + load(); + } catch (e: unknown) { + toast.error((e as Error).message); + } finally { + setLoading(false); + } + } + + async function handleToggleEnabled(schedule: Schedule) { + try { + await api(`/api/schedules/${schedule.id}`, { + method: 'PUT', + body: { + cron_expr: schedule.cron_expr, + enabled: !schedule.enabled, + }, + }); + toast.success(`Schedule ${schedule.enabled ? 'disabled' : 'enabled'}`); + load(); + } catch (e: unknown) { + toast.error((e as Error).message); + } + } + + async function handleDelete() { + if (deleteId === null) return; + try { + await api(`/api/schedules/${deleteId}`, { method: 'DELETE' }); + toast.success('Schedule deleted'); + setDeleteId(null); + load(); + } catch (e: unknown) { + toast.error((e as Error).message); + } + } + + function syncPairName(id: number) { + const p = syncPairs.find(p => p.id === id); + return p ? p.name : `Sync Pair ${id}`; + } + + function formatNextRun(nextRun: string | null) { + if (!nextRun) return 'Not scheduled'; + const d = new Date(nextRun); + return d.toLocaleString(); + } + + return ( +
+ + + Add Schedule + + } + /> + + +
+ {schedules.length === 0 ? ( + } + title="No schedules" + description="Create a schedule to automate sync pair execution" + action={ + + } + /> + ) : ( + + + + Sync Pair + Cron Expression + Next Run + Status + Actions + + + + {schedules.map(s => ( + + {syncPairName(s.sync_pair_id)} + + + {s.cron_expr} + + + + {formatNextRun(s.next_run_at)} + + + + + +
+ +
+ + + ))} + +
+ )} +
+ + + + + + Add Schedule + + Schedule automated sync pair execution using cron syntax. + Format: "minute hour day-of-month month day-of-week" + + +
+ +
+ + +
+
+ + setForm({ ...form, cron_expr: e.target.value })} + className="font-mono" + /> +

+ Format: "m h dom mon dow" (5 fields, no seconds) +
+ Examples: "0 2 * * *" (daily at 2am), "0 */6 * * *" (every 6 hours) +

+
+
+ setForm({ ...form, enabled: e.target.checked })} + className="h-4 w-4 rounded border-border accent-accent" + /> + +
+ + + + + +
+ + + + !v && setDeleteId(null)}> + + + Delete Schedule + + Are you sure you want to delete this schedule? + + + + + + + + +
+ ); +} diff --git a/web/src/pages/Settings.tsx b/web/src/pages/Settings.tsx index 68a506c..5e4fe5c 100644 --- a/web/src/pages/Settings.tsx +++ b/web/src/pages/Settings.tsx @@ -3,16 +3,21 @@ import { CopyButton } from '@/components/ui/CopyButton'; import { Button } from '@/components/ui/Button'; import { PageHeader } from '@/components/ui/PageHeader'; import { Card, CardHeader, CardTitle, CardBody } from '@/components/ui/Card'; -import { Key, Download, Terminal } from 'lucide-react'; +import { Key, Download, Terminal, HardDrive } from 'lucide-react'; +import { api, SettingsInfo } from '../api/client'; export default function Settings() { const [pubKey, setPubKey] = useState(''); + const [dataDir, setDataDir] = useState(''); const [loading, setLoading] = useState(true); useEffect(() => { - fetch('/api/settings/pubkey', { credentials: 'include' }) - .then(r => (r.ok ? r.text() : '')) - .then(t => setPubKey(t)) + Promise.all([ + fetch('/api/settings/pubkey', { credentials: 'include' }) + .then(r => (r.ok ? r.text() : '')) + .then(t => setPubKey(t)), + api('/api/settings/info').then(info => setDataDir(info.data_dir)), + ]) .catch(() => {}) .finally(() => setLoading(false)); }, []); @@ -75,6 +80,20 @@ export default function Settings() { + + +
+
+ +
+ Data Directory +
+ + +
{dataDir || '-'}
+ + +