package api import ( "context" "encoding/json" "errors" "fmt" "net/http" "os" "path/filepath" "sort" "strings" "time" "github.com/arescom/dockmv/internal/dkr" "github.com/arescom/dockmv/internal/job" "github.com/arescom/dockmv/internal/migrate" "github.com/arescom/dockmv/internal/spec" "github.com/arescom/dockmv/internal/sshx" "github.com/arescom/dockmv/internal/store" ) func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second) defer cancel() body := map[string]any{ "ok": true, "dockerHost": s.docker.Endpoint, "packageDir": s.cfg.PackageDir, "dataDir": s.cfg.DataDir, "knownHosts": s.hosts.Path(), "authRequired": s.cfg.Token != "", } if v, err := s.docker.Ping(ctx); err != nil { body["ok"] = false body["dockerError"] = err.Error() } else { body["dockerVersion"] = v } writeJSON(w, http.StatusOK, body) } // handleSource returns the full inventory of the source daemon, plus the // default selection for every container so the UI can render immediately. func (s *Server) handleSource(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute) defer cancel() inv, err := s.docker.Inventory(ctx) if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } defaults := make(map[string]spec.ItemSelection, len(inv.Containers)) for i := range inv.Containers { defaults[inv.Containers[i].ID] = spec.DefaultSelection(&inv.Containers[i]) } writeJSON(w, http.StatusOK, map[string]any{ "inventory": inv, "defaults": defaults, "options": spec.DefaultOptions(), }) } // handleSourceSizes measures volume sizes, which is slow enough that the UI // asks for it separately. func (s *Server) handleSourceSizes(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithTimeout(r.Context(), 5*time.Minute) defer cancel() sizes, err := s.docker.VolumeSizes(ctx) if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } writeJSON(w, http.StatusOK, map[string]any{"volumes": sizes}) } func (s *Server) handleListConnections(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, s.conns.List()) } func (s *Server) handleSaveConnection(w http.ResponseWriter, r *http.Request) { var cfg sshx.Config if err := decode(r, &cfg); err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } saved, err := s.conns.Save(cfg) if err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } writeJSON(w, http.StatusOK, saved) } func (s *Server) handleDeleteConnection(w http.ResponseWriter, r *http.Request) { if err := s.conns.Delete(r.PathValue("id")); err != nil { writeError(w, http.StatusNotFound, "%v", err) return } w.WriteHeader(http.StatusNoContent) } // handleProbe reads the target's host key so the operator can compare the // fingerprint before trusting it. func (s *Server) handleProbe(w http.ResponseWriter, r *http.Request) { cfg, err := s.conns.Get(r.PathValue("id")) if err != nil { writeError(w, http.StatusNotFound, "%v", err) return } ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) defer cancel() info, err := sshx.Probe(ctx, cfg, s.hosts) if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } writeJSON(w, http.StatusOK, info) } // handleTrust records the host key. The fingerprint the operator approved is // echoed back and re-checked, so approving one key cannot trust another. func (s *Server) handleTrust(w http.ResponseWriter, r *http.Request) { cfg, err := s.conns.Get(r.PathValue("id")) if err != nil { writeError(w, http.StatusNotFound, "%v", err) return } var body struct { Fingerprint string `json:"fingerprint"` } if err := decode(r, &body); err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) defer cancel() if err := sshx.TrustFromProbe(ctx, cfg, s.hosts, body.Fingerprint); err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } writeJSON(w, http.StatusOK, map[string]any{"trusted": true}) } func (s *Server) handleTestConnection(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) defer cancel() rd, closeFn, err := s.dialTarget(ctx, r.PathValue("id")) if err != nil { s.writeDialError(w, err) return } defer closeFn() pre, err := rd.Preflight(ctx) if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } writeJSON(w, http.StatusOK, pre) } func (s *Server) handleTargetInventory(w http.ResponseWriter, r *http.Request) { ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute) defer cancel() rd, closeFn, err := s.dialTarget(ctx, r.PathValue("id")) if err != nil { s.writeDialError(w, err) return } defer closeFn() inv, err := rd.Inventory(ctx) if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } writeJSON(w, http.StatusOK, inv) } // PreviewItem is what the UI shows when the operator asks "what will this do?". type PreviewItem struct { ContainerID string `json:"containerId"` Name string `json:"name"` TargetName string `json:"targetName"` Image string `json:"image"` Commands []string `json:"commands"` Transfers []string `json:"transfers"` Notes []string `json:"notes"` Warnings []string `json:"warnings"` TotalBytes int64 `json:"totalBytes"` } // handlePreview renders the exact docker commands a plan would run, without // touching either host. func (s *Server) handlePreview(w http.ResponseWriter, r *http.Request) { var plan spec.Plan if err := decode(r, &plan); err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } ctx, cancel := context.WithTimeout(r.Context(), 2*time.Minute) defer cancel() inv, err := s.docker.Inventory(ctx) if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } byID := map[string]*spec.Container{} for i := range inv.Containers { byID[inv.Containers[i].ID] = &inv.Containers[i] } out := []PreviewItem{} netSeen := map[string]bool{} var netCmds []string for _, sel := range plan.Items { if !sel.Include { continue } p, err := migrate.Prepare(byID[sel.ContainerID], sel, inv.Volumes, inv.Networks) if err != nil { writeError(w, http.StatusBadRequest, "container %s: %v", sel.ContainerID, err) return } item := PreviewItem{ ContainerID: p.Source.ID, Name: p.Source.Name, TargetName: p.ContainerName(), Image: p.Target.Image, Notes: p.Notes, Warnings: p.Source.Warnings, } for _, n := range p.Networks { if !netSeen[n.Name] { netSeen[n.Name] = true netCmds = append(netCmds, "docker "+spec.ShellQuoteAll(n.CreateArgs())) } } for _, v := range p.Volumes { item.Commands = append(item.Commands, "docker "+spec.ShellQuoteAll(v.CreateArgs())) } item.Commands = append(item.Commands, "docker "+spec.ShellQuoteAll(p.Target.CreateArgs(p.Render))) for _, args := range p.Target.NetworkConnectArgs(p.Render) { item.Commands = append(item.Commands, "docker "+spec.ShellQuoteAll(args)) } for _, t := range p.Transfers { item.Transfers = append(item.Transfers, t.Label) item.Commands = append(item.Commands, fmt.Sprintf("docker cp -a - %s:%s # contents of %s", p.ContainerName(), t.RestoreInto, t.SourcePath)) if t.SizeBytes > 0 { item.TotalBytes += t.SizeBytes } } if p.Selection.StartAfter { item.Commands = append(item.Commands, "docker start "+spec.ShellQuote(p.ContainerName())) } out = append(out, item) } writeJSON(w, http.StatusOK, map[string]any{ "networkCommands": netCmds, "items": out, }) } type sshMigrateRequest struct { ConnectionID string `json:"connectionId"` Plan spec.Plan `json:"plan"` } func (s *Server) handleMigrateSSH(w http.ResponseWriter, r *http.Request) { var req sshMigrateRequest if err := decode(r, &req); err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } if req.ConnectionID == "" { writeError(w, http.StatusBadRequest, "connectionId is required") return } if countIncluded(req.Plan) == 0 { writeError(w, http.StatusBadRequest, "no containers selected") return } cfg, err := s.conns.Get(req.ConnectionID) if err != nil { writeError(w, http.StatusNotFound, "%v", err) return } // The inventory is re-read now so the plan is applied to current state // rather than to whatever the browser last loaded. invCtx, cancel := context.WithTimeout(r.Context(), 2*time.Minute) inv, err := s.docker.Inventory(invCtx) cancel() if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } // Dial before the job starts so a bad credential is an immediate error in // the UI instead of a failed job. dialCtx, dialCancel := context.WithTimeout(r.Context(), 45*time.Second) client, err := sshx.Dial(dialCtx, cfg, s.hosts) dialCancel() if err != nil { s.writeDialError(w, err) return } title := fmt.Sprintf("%d container(s) to %s", countIncluded(req.Plan), cfg.Name) runner := &migrate.SSHRunner{ Src: s.docker, Dst: sshx.NewRemoteDocker(client), Containers: inv.Containers, Volumes: inv.Volumes, Networks: inv.Networks, Plan: req.Plan, } j := s.jobs.Run(context.Background(), job.KindSSH, title, req.Plan.Options.DryRun, func(ctx context.Context, j *job.Job) error { defer client.Close() j.Logf(job.LevelInfo, "", "migrating to %s@%s over ssh", cfg.User, cfg.Host) return runner.Run(ctx, j) }) writeJSON(w, http.StatusAccepted, j.Snapshot()) } type packageRequest struct { Plan spec.Plan `json:"plan"` Format migrate.PackageFormat `json:"format"` } func (s *Server) handleBuildPackage(w http.ResponseWriter, r *http.Request) { var req packageRequest if err := decode(r, &req); err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } if countIncluded(req.Plan) == 0 { writeError(w, http.StatusBadRequest, "no containers selected") return } if req.Format == "" { req.Format = migrate.FormatTar } if req.Format != migrate.FormatTar && req.Format != migrate.FormatDir { writeError(w, http.StatusBadRequest, "format must be %q or %q", migrate.FormatTar, migrate.FormatDir) return } invCtx, cancel := context.WithTimeout(r.Context(), 2*time.Minute) inv, err := s.docker.Inventory(invCtx) cancel() if err != nil { writeError(w, http.StatusBadGateway, "%v", err) return } packager := &migrate.Packager{ Src: s.docker, Containers: inv.Containers, Volumes: inv.Volumes, Networks: inv.Networks, Plan: req.Plan, OutputDir: s.cfg.PackageDir, Format: req.Format, SourceHost: inv.Host, } title := fmt.Sprintf("package of %d container(s)", countIncluded(req.Plan)) j := s.jobs.Run(context.Background(), job.KindPackage, title, req.Plan.Options.DryRun, func(ctx context.Context, j *job.Job) error { res, err := packager.Run(ctx, j) if err != nil { return err } j.Logf(job.LevelInfo, "", "package ready: %s (%s)", res.Name, humanBytes(res.Bytes)) return nil }) writeJSON(w, http.StatusAccepted, j.Snapshot()) } func (s *Server) handleListJobs(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, s.jobs.List()) } func (s *Server) handleGetJob(w http.ResponseWriter, r *http.Request) { j, err := s.jobs.Get(r.PathValue("id")) if err != nil { writeError(w, http.StatusNotFound, "%v", err) return } writeJSON(w, http.StatusOK, j.Snapshot()) } // handleJobEvents streams job snapshots over server-sent events, coalescing // rapid updates so a fast transfer does not saturate the browser. func (s *Server) handleJobEvents(w http.ResponseWriter, r *http.Request) { j, err := s.jobs.Get(r.PathValue("id")) if err != nil { writeError(w, http.StatusNotFound, "%v", err) return } flusher, ok := w.(http.Flusher) if !ok { writeError(w, http.StatusInternalServerError, "streaming unsupported") return } w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") w.Header().Set("X-Accel-Buffering", "no") w.WriteHeader(http.StatusOK) changes, unsubscribe := j.Subscribe() defer unsubscribe() send := func() bool { b, err := json.Marshal(j.Snapshot()) if err != nil { return false } if _, err := fmt.Fprintf(w, "data: %s\n\n", b); err != nil { return false } flusher.Flush() return true } if !send() { return } // Updates are batched: at most one frame every 250ms while work is busy. ticker := time.NewTicker(250 * time.Millisecond) defer ticker.Stop() keepalive := time.NewTicker(20 * time.Second) defer keepalive.Stop() dirty := false for { select { case <-r.Context().Done(): return case _, open := <-changes: if !open { return } dirty = true case <-ticker.C: if !dirty { continue } dirty = false if !send() { return } if j.Snapshot().State.Terminal() { fmt.Fprint(w, "event: done\ndata: {}\n\n") flusher.Flush() return } case <-keepalive.C: fmt.Fprint(w, ": keepalive\n\n") flusher.Flush() case <-j.Done(): // Drain one last snapshot so the client sees the final state. send() fmt.Fprint(w, "event: done\ndata: {}\n\n") flusher.Flush() return } } } func (s *Server) handleCancelJob(w http.ResponseWriter, r *http.Request) { j, err := s.jobs.Get(r.PathValue("id")) if err != nil { writeError(w, http.StatusNotFound, "%v", err) return } j.Cancel() writeJSON(w, http.StatusOK, map[string]any{"canceled": true}) } func (s *Server) handleDeleteJob(w http.ResponseWriter, r *http.Request) { if err := s.jobs.Delete(r.PathValue("id")); err != nil { writeError(w, http.StatusNotFound, "%v", err) return } w.WriteHeader(http.StatusNoContent) } // PackageInfo describes one built package on disk. type PackageInfo struct { Name string `json:"name"` Path string `json:"path"` Bytes int64 `json:"bytes"` IsDir bool `json:"isDir"` CreatedAt time.Time `json:"createdAt"` } func (s *Server) handleListPackages(w http.ResponseWriter, r *http.Request) { entries, err := os.ReadDir(s.cfg.PackageDir) if err != nil { writeError(w, http.StatusInternalServerError, "read package directory: %v", err) return } out := []PackageInfo{} for _, e := range entries { info, err := e.Info() if err != nil { continue } p := PackageInfo{ Name: e.Name(), Path: filepath.Join(s.cfg.PackageDir, e.Name()), IsDir: e.IsDir(), CreatedAt: info.ModTime(), } if e.IsDir() { p.Bytes, _ = dirBytes(p.Path) } else { p.Bytes = info.Size() } out = append(out, p) } sort.Slice(out, func(i, j int) bool { return out[i].CreatedAt.After(out[j].CreatedAt) }) writeJSON(w, http.StatusOK, out) } // handleDownloadPackage streams a package file. Directory packages are not // downloadable as-is; the operator picks the tar format for that. func (s *Server) handleDownloadPackage(w http.ResponseWriter, r *http.Request) { path, err := s.packagePath(r.PathValue("name")) if err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } info, err := os.Stat(path) if err != nil { writeError(w, http.StatusNotFound, "package not found") return } if info.IsDir() { writeError(w, http.StatusBadRequest, "this package is a directory; copy it from %s, or rebuild with the tar format", path) return } f, err := os.Open(path) if err != nil { writeError(w, http.StatusInternalServerError, "%v", err) return } defer f.Close() w.Header().Set("Content-Type", "application/x-tar") w.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=%q", filepath.Base(path))) http.ServeContent(w, r, filepath.Base(path), info.ModTime(), f) } func (s *Server) handleDeletePackage(w http.ResponseWriter, r *http.Request) { path, err := s.packagePath(r.PathValue("name")) if err != nil { writeError(w, http.StatusBadRequest, "%v", err) return } if err := os.RemoveAll(path); err != nil { writeError(w, http.StatusInternalServerError, "%v", err) return } w.WriteHeader(http.StatusNoContent) } // packagePath resolves a package name against the package directory, refusing // anything that would escape it. func (s *Server) packagePath(name string) (string, error) { if name == "" || strings.ContainsAny(name, `/\`) || name == "." || name == ".." { return "", errors.New("invalid package name") } base, err := filepath.Abs(s.cfg.PackageDir) if err != nil { return "", err } full := filepath.Join(base, name) if !strings.HasPrefix(full, base+string(os.PathSeparator)) { return "", errors.New("invalid package name") } return full, nil } // dialTarget opens a short-lived connection for an interactive request. func (s *Server) dialTarget(ctx context.Context, id string) (*sshx.RemoteDocker, func(), error) { cfg, err := s.conns.Get(id) if err != nil { return nil, nil, err } client, err := sshx.Dial(ctx, cfg, s.hosts) if err != nil { return nil, nil, err } return sshx.NewRemoteDocker(client), func() { client.Close() }, nil } // writeDialError turns an untrusted host key into a structured response the UI // can turn into a trust prompt. func (s *Server) writeDialError(w http.ResponseWriter, err error) { var hk *sshx.HostKeyError if errors.As(err, &hk) { writeJSON(w, http.StatusPreconditionRequired, map[string]any{ "error": err.Error(), "hostKey": hk.Fingerprint, "keyType": hk.KeyType, "changed": hk.Changed, "needsTrust": true, "host": hk.Host, }) return } if errors.Is(err, store.ErrNotFound) { writeError(w, http.StatusNotFound, "%v", err) return } writeError(w, http.StatusBadGateway, "%v", err) } func countIncluded(p spec.Plan) int { n := 0 for _, i := range p.Items { if i.Include { n++ } } return n } func dirBytes(dir string) (int64, error) { var total int64 err := filepath.Walk(dir, func(_ string, info os.FileInfo, err error) error { if err != nil { return err } if !info.IsDir() { total += info.Size() } return nil }) return total, err } func humanBytes(n int64) string { const unit = 1024 if n < unit { return fmt.Sprintf("%d B", n) } div, exp := int64(unit), 0 for v := n / unit; v >= unit; v /= unit { div *= unit exp++ } return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp]) } var _ = dkr.RestorePath