84b185be39
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
221 lines
7.1 KiB
Go
221 lines
7.1 KiB
Go
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"
|
|
"github.com/syncserver/internal/auth"
|
|
"github.com/syncserver/internal/config"
|
|
"github.com/syncserver/internal/sshmanager"
|
|
"github.com/syncserver/internal/syncengine"
|
|
"github.com/syncserver/internal/webui"
|
|
)
|
|
|
|
type Server struct {
|
|
router *chi.Mux
|
|
cfg *config.Config
|
|
engine *syncengine.Engine
|
|
}
|
|
|
|
func NewServer(cfg *config.Config, db *sql.DB, engine *syncengine.Engine) *Server {
|
|
auth.InitJWTManager(cfg.Auth.JWTSecret, cfg.Auth.JWTExpiryH)
|
|
|
|
r := chi.NewRouter()
|
|
r.Use(middleware.RequestID)
|
|
r.Use(middleware.RealIP)
|
|
r.Use(middleware.Logger)
|
|
r.Use(recoverer)
|
|
|
|
s := &Server{router: r, cfg: cfg, engine: engine}
|
|
|
|
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)
|
|
|
|
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(authGet).Get("/me", authHandler.Me)
|
|
})
|
|
|
|
r.With(authGet).Route("/machines", func(r chi.Router) {
|
|
r.Get("/", machineHandler.List)
|
|
r.With(admin).Post("/", machineHandler.Create)
|
|
r.With(admin).Post("/refresh", machineHandler.Refresh)
|
|
r.Get("/{id}", machineHandler.Get)
|
|
r.With(admin).Put("/{id}", machineHandler.Update)
|
|
r.With(admin).Delete("/{id}", machineHandler.Delete)
|
|
r.Post("/{id}/test-wol", machineHandler.TestWoL)
|
|
r.With(admin).Post("/{id}/shutdown", machineHandler.Shutdown)
|
|
r.Post("/{id}/test-connection", machineHandler.TestConnection)
|
|
r.With(admin).Post("/{id}/approve-fingerprint", machineHandler.ApproveFingerprint)
|
|
r.With(admin).Post("/{id}/deploy-keys", machineHandler.DeployKeys)
|
|
})
|
|
|
|
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.With(admin).Delete("/{id}", syncPairHandler.Delete)
|
|
r.Post("/{id}/run", jobHandler.TriggerRun)
|
|
})
|
|
|
|
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)
|
|
r.Get("/{id}/log", jobHandler.GetLog)
|
|
r.Get("/{id}/log/download", jobHandler.DownloadLog)
|
|
r.Get("/{id}/log/stream", sseHandler.StreamJob)
|
|
})
|
|
|
|
r.With(authGet).Get("/jobs/stream", sseHandler.StreamAll)
|
|
|
|
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(authGet).Get("/settings/info", func(w http.ResponseWriter, r *http.Request) {
|
|
_, _, pubKey, _ := sshmanager.EnsureServerKey(cfg.SSHDir())
|
|
resp := SettingsInfoResponse{
|
|
Version: cfg.Version,
|
|
DataDir: cfg.DataDir,
|
|
SSHPubKey: pubKey,
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
json.NewEncoder(w).Encode(resp)
|
|
})
|
|
|
|
r.With(authGet).Route("/ssh-keys", func(r chi.Router) {
|
|
r.Get("/", sshKeyHandler.List)
|
|
r.With(admin).Post("/", sshKeyHandler.Create)
|
|
r.Get("/{id}", sshKeyHandler.Get)
|
|
r.With(admin).Delete("/{id}", sshKeyHandler.Delete)
|
|
r.With(admin).Get("/{id}/private", sshKeyHandler.DownloadPrivate)
|
|
})
|
|
})
|
|
|
|
r.Get("/health", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
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)
|
|
})
|
|
|
|
return s
|
|
}
|
|
|
|
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
s.router.ServeHTTP(w, r)
|
|
}
|
|
|
|
func recoverer(next http.Handler) http.Handler {
|
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
defer func() {
|
|
if err := recover(); err != nil {
|
|
slog.Error("panic recovered",
|
|
"error", err,
|
|
"stack", string(debug.Stack()),
|
|
"method", r.Method,
|
|
"path", r.URL.Path,
|
|
)
|
|
w.Header().Set("Content-Type", "application/json")
|
|
w.WriteHeader(http.StatusInternalServerError)
|
|
http.Error(w, `{"error":"internal server error"}`, http.StatusInternalServerError)
|
|
}
|
|
}()
|
|
next.ServeHTTP(w, r)
|
|
})
|
|
}
|