Files
hamlab/internal/api/server.go
T
Mitchell-JohnsonandClaude Fable 5.1 f8c9479e93 Files: previews — a toggle that shows a thumbnail per image and video, made by ffmpeg on the machine's GPU where it has one, cached under the data dir
A previews button on the Files tab turns the icon column into a 256 px thumbnail for
every image and video row, loaded as rows scroll near. internal/thumbs probes what the
machine can do rather than assuming it: ffmpeg and ffprobe, every /dev/dri/renderD*
opened and exercised, vainfo's decode profiles per node where it exists. A video goes to
the first GPU that lists its codec, with ffmpeg's own software fallback behind it and a
CPU retry behind that; images decode on the CPU, JPEGs at a reduced DCT size. Previews
are kept under <data dir>/thumbs/ keyed by path, size and mtime, with undecodable files
remembered too. Root mode makes them through the same root helper the file reads use.
Where ffmpeg is missing the tab offers to install it; where a render node exists but is
closed to the daemon's user, it offers a drop-in adding the group and a restart.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-11 15:20:27 -07:00

2597 lines
85 KiB
Go

// Package api is the HTTP/WebSocket surface: auth, terminal CRUD, attach, static UI.
package api
import (
"bytes"
"context"
"crypto/subtle"
"encoding/json"
"errors"
"fmt"
"io"
"io/fs"
"log"
"mime"
"net"
"net/http"
"net/http/pprof"
"net/url"
"os"
"os/exec"
"os/user"
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"syscall"
"time"
"github.com/coder/websocket"
"hamlab/internal/auth"
"hamlab/internal/config"
"hamlab/internal/cron"
"hamlab/internal/files"
"hamlab/internal/hardware"
"hamlab/internal/incus"
"hamlab/internal/metrics"
"hamlab/internal/network"
"hamlab/internal/nodes"
"hamlab/internal/priv"
"hamlab/internal/sensors"
"hamlab/internal/services"
"hamlab/internal/setup"
"hamlab/internal/space"
"hamlab/internal/storage"
"hamlab/internal/store"
"hamlab/internal/tasks"
"hamlab/internal/term"
"hamlab/internal/updates"
)
const cookieName = "hamlab_session"
type Server struct {
// Restart replaces the running process with the binary on disk; it only returns on failure.
Restart func() error
cfg *config.Config
sessions *auth.Sessions
limiter auth.Limiter
mgr *term.Manager
metrics *metrics.Collector
sensors *sensors.Registry
streams streamGate
alerts *alertEngine
store *store.Store
svc *services.Manager
cron *cron.Manager
tasks *tasks.Manager
updates *updates.Manager
storage *storage.Manager
hardware *hardware.Manager
space *space.Manager
net *network.Watcher
update updateState
events eventLog
kvms kvmProxies
virt *incus.Manager
fsLocal files.FS
fsSudo files.FS // nil without passwordless sudo
thumbs *thumbSvc
nodes *nodes.Manager
fingerprint string
version string
started time.Time
user string
host string // the machine's hostname as last read; use hostName(), it can change under us
hostMu sync.Mutex
hostAt time.Time
ui fs.FS
mux *http.ServeMux
}
func New(cfg *config.Config, sessions *auth.Sessions, mgr *term.Manager, col *metrics.Collector, st *store.Store, svc *services.Manager, cr *cron.Manager, tm *tasks.Manager, up *updates.Manager, sto *storage.Manager, fsLocal, fsSudo files.FS, nm *nodes.Manager, fingerprint string, ui fs.FS, version string) *Server {
s := &Server{cfg: cfg, sessions: sessions, mgr: mgr, metrics: col, sensors: sensors.NewRegistry(col), store: st, svc: svc, cron: cr, tasks: tm, updates: up, storage: sto, fsLocal: fsLocal, fsSudo: fsSudo, nodes: nm, fingerprint: fingerprint, ui: ui, version: version,
hardware: hardware.New(), started: time.Now(), user: os.Getenv("USER")}
self, err := os.Executable()
if err != nil {
self = "hamlab"
}
s.space = space.New(self, filepath.Join(cfg.DataDir(), "space"))
s.thumbs = newThumbSvc(cfg.DataDir(), self)
s.hostName() // read it once now so the first request doesn't have to
if u, err := user.Current(); err == nil && u.Username != "" {
s.user = u.Username
}
if s.user == "" {
s.user = "?"
}
s.virt = incus.New(cfg.DataDir(), s.user, version)
s.migrateUsers()
// The speed test and the outside view. Its history outlives a restart, and its last result
// is published as sensors, so the line is graphable and alertable like anything else.
s.net = network.NewWatcher(s.keepResult)
s.net.Test.Load(s.loadNetwork())
s.sensors.Add(netSensors{s.net})
s.sensors.Add(newStatusSensors(s))
s.alerts = newAlertEngine(s)
// The saved dashboard layout carries the sampling interval; apply it on boot.
var histSec int
st.View(func(x *store.State) { histSec = x.HistorySec })
if histSec > 0 {
col.SetWindow(time.Duration(histSec) * time.Second)
}
// The sampler follows the console's refresh rate. A machine remembers what it was last told;
// the hub reads the rate from the saved UI prefs and tells everyone again, because a node
// that restarted, or was added since, would otherwise sample at its default until somebody
// happened to open its Dashboard.
var samplerMs int
var uiRaw json.RawMessage
st.View(func(x *store.State) {
samplerMs, uiRaw = x.SamplerMs, x.UI
if len(uiRaw) == 0 && len(x.Users) > 0 { // any account's saved rate, before the machine had one of its own
for _, u := range x.Users {
uiRaw = u.UI
break
}
}
})
if samplerMs > 0 {
col.SetInterval(time.Duration(samplerMs) * time.Millisecond)
}
if nm != nil {
var ui struct {
RefreshMs float64 `json:"refreshMs"`
}
if samplerMs > 0 {
go s.setSampler(samplerMs, false)
} else if json.Unmarshal(uiRaw, &ui) == nil && ui.RefreshMs >= 100 {
go s.setSampler(int(ui.RefreshMs), false)
}
go s.keepFleetInStep()
}
go s.watchGuests()
go s.watchForUpdates(context.Background())
go s.watchDockerEvents(context.Background())
s.mux = s.routes()
return s
}
// routes is the whole HTTP surface. A method of its own, rather than inline in New, so a
// test can build the real router without the machinery behind it — which is what the guest
// gate is checked against, since it decides a request by asking this router where it would go.
func (s *Server) routes() *http.ServeMux {
m := http.NewServeMux()
m.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(204) })
m.HandleFunc("POST /api/login", s.login)
m.HandleFunc("POST /api/tokens", s.issueToken)
m.HandleFunc("GET /api/tokens", s.auth(s.listTokens))
m.HandleFunc("DELETE /api/tokens/{name}", s.auth(s.revokeToken))
m.HandleFunc("GET /api/nodes", s.auth(s.listNodes))
m.HandleFunc("POST /api/nodes", s.auth(s.addNode))
m.HandleFunc("DELETE /api/nodes/{name}", s.auth(s.removeNode))
m.HandleFunc("/api/nodes/{name}/{rest...}", s.auth(s.proxyNode))
m.HandleFunc("POST /api/self-update", s.auth(s.selfUpdate))
m.HandleFunc("GET /api/update", s.auth(s.updateStatus))
m.HandleFunc("POST /api/update/check", s.auth(s.updateCheck))
m.HandleFunc("POST /api/update/install", s.auth(s.updateInstall))
m.HandleFunc("POST /api/machines/{name}/update", s.auth(s.updateMachine))
m.HandleFunc("POST /api/machines-update-all", s.auth(s.updateAllMachines))
m.HandleFunc("POST /api/logout", s.auth(s.logout))
m.HandleFunc("GET /api/me", s.auth(s.me))
m.HandleFunc("GET /api/terminals", s.auth(s.listTerminals))
m.HandleFunc("POST /api/terminals", s.auth(s.createTerminal))
m.HandleFunc("PATCH /api/terminals/{id}", s.auth(s.renameTerminal))
m.HandleFunc("DELETE /api/terminals/{id}", s.auth(s.deleteTerminal))
m.HandleFunc("GET /api/terminals/{id}/attach", s.auth(s.attach))
m.HandleFunc("GET /api/sensors", s.sensorAuth(s.sensorCatalog))
m.HandleFunc("GET /api/sensors/query", s.sensorAuth(s.sensorQuery))
m.HandleFunc("POST /api/sensors/query", s.sensorAuth(s.sensorQuery))
m.HandleFunc("GET /api/sensors/stream", s.sensorAuth(s.sensorStream))
m.HandleFunc("GET /api/sensors/history", s.sensorAuth(s.sensorHistories))
m.HandleFunc("GET /api/sensors/prometheus", s.sensorAuth(s.sensorProm))
m.HandleFunc("GET /metrics", s.sensorAuth(s.sensorProm))
m.HandleFunc("GET /api/sensors/{id}", s.sensorAuth(s.sensorOne))
m.HandleFunc("GET /api/sensors/{id}/history", s.sensorAuth(s.sensorHistory))
m.HandleFunc("GET /api/nodes/{name}/sensors", s.sensorAuth(s.proxyNodeSensors))
m.HandleFunc("GET /api/nodes/{name}/sensors/{rest...}", s.sensorAuth(s.proxyNodeSensors))
m.HandleFunc("POST /api/nodes/{name}/sensors/query", s.sensorAuth(s.proxyNodeSensors))
m.HandleFunc("POST /api/tokens/read", s.auth(s.issueReadToken))
m.HandleFunc("GET /api/docker", s.auth(s.dockerOverview))
m.HandleFunc("GET /api/docker/stacks", s.auth(s.stacksList))
m.HandleFunc("POST /api/docker/stacks", s.auth(s.stackCreate))
m.HandleFunc("GET /api/docker/stacks/homes", s.auth(s.stackHomes))
m.HandleFunc("POST /api/docker/stacks/{name}/delete", s.auth(s.stackDelete))
m.HandleFunc("POST /api/docker/images/pull", s.auth(s.imagePull))
m.HandleFunc("POST /api/docker/images/remove", s.auth(s.imageRemove))
m.HandleFunc("POST /api/docker/images/prune", s.auth(s.imagePrune))
m.HandleFunc("POST /api/docker/volumes/remove", s.auth(s.volumeRemove))
m.HandleFunc("POST /api/docker/volumes/prune", s.auth(s.volumePrune))
m.HandleFunc("POST /api/docker/networks/remove", s.auth(s.networkRemove))
m.HandleFunc("POST /api/docker/networks/prune", s.auth(s.networkPrune))
m.HandleFunc("GET /api/docker/stacks/{name}", s.auth(s.stackFile))
m.HandleFunc("PUT /api/docker/stacks/{name}", s.auth(s.stackSave))
m.HandleFunc("POST /api/docker/stacks/{name}/{action}", s.auth(s.stackAction))
m.HandleFunc("GET /api/docker/stacks/{name}/diff", s.auth(s.stackDiff))
m.HandleFunc("GET /api/docker/stacks/{name}/updates", s.auth(s.stackUpdates))
m.HandleFunc("GET /api/docker/updates", s.auth(s.machineUpdates))
m.HandleFunc("GET /api/docker/schedules", s.auth(s.dsList))
m.HandleFunc("PUT /api/docker/schedules", s.auth(s.dsPut))
m.HandleFunc("DELETE /api/docker/schedules/{id}", s.auth(s.dsDelete))
m.HandleFunc("POST /api/docker/schedules/{id}/run", s.auth(s.dsRun))
// Asked of a node through this hub, every one of these is the node's to answer: the
// generic /api/nodes/{name}/{rest...} proxy above carries them there unchanged.
m.HandleFunc("POST /api/docker/containers/{id}/{action}", s.auth(s.dockerAction))
m.HandleFunc("GET /api/docker/containers/{id}/logs", s.auth(s.dockerLogs))
m.HandleFunc("GET /api/docker/containers/{id}/logs/tail", s.auth(s.dockerTail))
m.HandleFunc("GET /api/docker/containers/{id}/inspect", s.auth(s.dockerInspect))
m.HandleFunc("GET /api/docker/containers/{id}/health", s.auth(s.dockerHealth))
m.HandleFunc("GET /api/docker/containers/{id}/events", s.auth(s.dockerEvents))
m.HandleFunc("POST /api/docker/containers/{id}/exec", s.auth(s.dockerExec))
m.HandleFunc("POST /api/docker/containers/{id}/limits", s.auth(s.dockerUpdate))
m.HandleFunc("GET /api/virt", s.auth(s.virtOverview))
m.HandleFunc("POST /api/virt/install", s.auth(s.virtInstall))
m.HandleFunc("GET /api/virt/images/remote", s.auth(s.virtRemoteImages))
m.HandleFunc("POST /api/virt/images/download", s.auth(s.virtImageDownload))
m.HandleFunc("POST /api/virt/images/import", s.auth(s.virtImageImport))
m.HandleFunc("POST /api/virt/images/publish", s.auth(s.virtImagePublish))
m.HandleFunc("POST /api/virt/images/delete", s.auth(s.virtImageDelete))
m.HandleFunc("POST /api/virt/instances", s.auth(s.virtCreate))
m.HandleFunc("GET /api/virt/instances/{name}", s.auth(s.virtInstance))
m.HandleFunc("POST /api/virt/instances/{name}/rename", s.auth(s.virtRename))
m.HandleFunc("PUT /api/virt/instances/{name}/config", s.auth(s.virtConfig))
m.HandleFunc("POST /api/virt/instances/{name}/snapshots", s.auth(s.virtSnapshotCreate))
m.HandleFunc("POST /api/virt/instances/{name}/snapshots/{snap}/{action}", s.auth(s.virtSnapshotAction))
m.HandleFunc("POST /api/virt/instances/from-iso", s.auth(s.virtFromISO))
m.HandleFunc("GET /api/virt/instances/{name}/vga", s.auth(s.virtVGA))
m.HandleFunc("POST /api/virt/instances/{name}/iso", s.auth(s.virtISOAttach))
m.HandleFunc("POST /api/virt/instances/{name}/iso/detach", s.auth(s.virtISODetach))
m.HandleFunc("POST /api/virt/instances/{name}/mounts", s.auth(s.virtMountAdd))
m.HandleFunc("POST /api/virt/instances/{name}/mounts/remove", s.auth(s.virtMountRemove))
m.HandleFunc("POST /api/virt/instances/{name}/{action}", s.auth(s.virtAction))
m.HandleFunc("POST /api/virt/isos/fetch", s.auth(s.virtISOFetch))
m.HandleFunc("POST /api/virt/isos/import", s.auth(s.virtISOImport))
m.HandleFunc("POST /api/virt/isos/delete", s.auth(s.virtISODelete))
m.HandleFunc("POST /api/virt/isos/rename", s.auth(s.virtISORename))
m.HandleFunc("GET /api/virt/exports", s.auth(s.virtExports))
m.HandleFunc("PUT /api/virt/exports", s.auth(s.virtExportsPut))
m.HandleFunc("POST /api/virt/exports/run", s.auth(s.virtExportsRun))
m.HandleFunc("POST /api/virt/exports/delete", s.auth(s.virtExportDelete))
m.HandleFunc("GET /api/virt/sshkeys", s.auth(s.virtSSHKeys))
m.HandleFunc("POST /api/virt/pools", s.auth(s.virtPoolCreate))
m.HandleFunc("POST /api/virt/pools/delete", s.auth(s.virtPoolDelete))
m.HandleFunc("GET /api/alerts", s.auth(s.alertsGet))
m.HandleFunc("PUT /api/alerts", s.auth(s.alertsPut))
m.HandleFunc("POST /api/alerts/test", s.auth(s.alertsTest))
m.HandleFunc("GET /api/fleet/sensors/query", s.sensorAuth(s.fleetQuery))
m.HandleFunc("POST /api/fleet/sensors/query", s.sensorAuth(s.fleetQuery))
m.HandleFunc("GET /api/services", s.auth(s.listServices))
m.HandleFunc("GET /api/services/{unit}", s.auth(s.showService))
m.HandleFunc("GET /api/services/{unit}/file", s.auth(s.serviceFile))
m.HandleFunc("GET /api/services/{unit}/logs", s.auth(s.serviceLogs))
m.HandleFunc("GET /api/services/{unit}/logs/tail", s.auth(s.serviceTail))
m.HandleFunc("POST /api/services/{unit}/{action}", s.auth(s.serviceAction))
m.HandleFunc("GET /api/cron", s.auth(s.cronOverview))
m.HandleFunc("GET /api/cron/source", s.auth(s.cronSource))
m.HandleFunc("PUT /api/cron/source", s.auth(s.cronWriteSource))
m.HandleFunc("GET /api/cron/jobs/{id}/logs", s.auth(s.cronJobLogs))
m.HandleFunc("GET /api/cron/jobs/{id}/output", s.auth(s.cronJobOutput))
m.HandleFunc("POST /api/cron/jobs/{id}/{action}", s.auth(s.cronJobAction))
m.HandleFunc("GET /api/cron/runs/{id}", s.auth(s.cronRun))
m.HandleFunc("POST /api/cron/timers/{unit}/{action}", s.auth(s.cronTimerAction))
m.HandleFunc("GET /api/tasks", s.auth(s.listTasks))
m.HandleFunc("GET /api/tasks/{id}", s.auth(s.getTask))
m.HandleFunc("GET /api/tasks/{id}/stream", s.auth(s.streamTask))
m.HandleFunc("POST /api/tasks/{id}/cancel", s.auth(s.cancelTask))
m.HandleFunc("GET /api/updates", s.auth(s.updatesStatus))
m.HandleFunc("POST /api/updates/refresh", s.auth(s.updatesRefresh))
m.HandleFunc("POST /api/updates/upgrade", s.auth(s.updatesUpgrade))
m.HandleFunc("GET /api/updates/changelog", s.auth(s.updatesChangelog))
m.HandleFunc("POST /api/power/{action}", s.auth(s.power))
m.HandleFunc("POST /api/updates/install", s.auth(s.updatesInstall))
m.HandleFunc("GET /api/files", s.auth(s.filesList))
m.HandleFunc("GET /api/files/stat", s.auth(s.filesStat))
m.HandleFunc("GET /api/files/read", s.auth(s.filesRead))
m.HandleFunc("GET /api/files/download", s.auth(s.filesRead))
m.HandleFunc("PUT /api/files/write", s.auth(s.filesWrite))
m.HandleFunc("PUT /api/files/upload", s.auth(s.filesUpload))
m.HandleFunc("POST /api/files/mkdir", s.auth(s.filesMkdir))
m.HandleFunc("POST /api/files/touch", s.auth(s.filesTouch))
m.HandleFunc("POST /api/files/rename", s.auth(s.filesRename))
m.HandleFunc("POST /api/files/delete", s.auth(s.filesDelete))
m.HandleFunc("POST /api/files/chmod", s.auth(s.filesChmod))
m.HandleFunc("GET /api/files/thumb", s.auth(s.filesThumb))
m.HandleFunc("GET /api/files/thumbs", s.auth(s.thumbsStatus))
m.HandleFunc("POST /api/files/thumbs/install", s.auth(s.thumbsInstall))
m.HandleFunc("POST /api/files/thumbs/gpu-access", s.auth(s.thumbsGPUAccess))
m.HandleFunc("GET /api/hardware", s.auth(s.hardwareReport))
m.HandleFunc("GET /api/network", s.auth(s.networkOverview))
m.HandleFunc("GET /api/vpn", s.auth(s.vpnReport))
m.HandleFunc("POST /api/vpn/{iface}/peers", s.auth(s.addVPNPeer))
m.HandleFunc("PATCH /api/vpn/{iface}/peers/{section}", s.auth(s.editVPNPeer))
m.HandleFunc("DELETE /api/vpn/{iface}/peers/{section}", s.auth(s.removeVPNPeer))
m.HandleFunc("GET /api/network/speedtest", s.auth(s.speedtestState))
m.HandleFunc("POST /api/network/speedtest", s.auth(s.speedtestStart))
m.HandleFunc("POST /api/network/speedtest/cancel", s.auth(s.speedtestCancel))
m.HandleFunc("GET /api/storage", s.auth(s.storageOverview))
m.HandleFunc("GET /api/storage/smart/{dev}", s.auth(s.storageSmart))
m.HandleFunc("POST /api/storage/smart/{dev}/test/{kind}", s.auth(s.storageSmartTest))
m.HandleFunc("GET /api/storage/zfs/snapshots", s.auth(s.zfsSnapshots))
m.HandleFunc("POST /api/storage/zfs/scrub/{pool}/{action}", s.auth(s.zfsScrub))
m.HandleFunc("POST /api/storage/zfs/snapshot", s.auth(s.zfsSnapshot))
m.HandleFunc("GET /api/zfs", s.auth(s.zfsOverview))
m.HandleFunc("GET /api/zfs/arc", s.auth(s.zfsARC))
m.HandleFunc("PUT /api/zfs/schedules", s.auth(s.zfsPolicyPut))
m.HandleFunc("DELETE /api/zfs/schedules/{id}", s.auth(s.zfsPolicyDelete))
m.HandleFunc("POST /api/zfs/schedules/{id}/run", s.auth(s.zfsRunNow))
m.HandleFunc("POST /api/zfs/snapshots/delete", s.auth(s.zfsSnapshotDelete))
m.HandleFunc("GET /api/space", s.auth(s.spaceOverview))
m.HandleFunc("GET /api/space/{disk}", s.auth(s.spaceStatus))
m.HandleFunc("POST /api/space/{disk}/scan", s.auth(s.spaceScan))
m.HandleFunc("POST /api/space/{disk}/cancel", s.auth(s.spaceCancel))
m.HandleFunc("DELETE /api/space/{disk}", s.auth(s.spaceForget))
m.HandleFunc("GET /api/space/{disk}/tree", s.auth(s.spaceTree))
m.HandleFunc("GET /api/space/{disk}/largest", s.auth(s.spaceLargest))
m.HandleFunc("GET /api/procs", s.auth(s.procList))
m.HandleFunc("POST /api/procs/{pid}/signal", s.auth(s.procSignal))
m.HandleFunc("POST /api/password", s.auth(s.changePassword))
m.HandleFunc("GET /api/users", s.auth(s.listUsers))
m.HandleFunc("POST /api/users", s.auth(s.addUser))
m.HandleFunc("DELETE /api/users/{name}", s.auth(s.removeUser))
m.HandleFunc("POST /api/users/{name}/password", s.auth(s.resetPassword))
m.HandleFunc("GET /api/guests", s.auth(s.listGuests))
m.HandleFunc("POST /api/guests", s.auth(s.addGuest))
m.HandleFunc("DELETE /api/guests/{name}", s.auth(s.revokeGuest))
m.HandleFunc("GET /api/sessions", s.auth(s.listSessions))
m.HandleFunc("DELETE /api/sessions/{id}", s.auth(s.revokeSession))
m.HandleFunc("POST /api/sessions/revoke-others", s.auth(s.revokeOtherSessions))
m.HandleFunc("POST /api/sessions/revoke-idle", s.auth(s.revokeIdleSessions))
m.HandleFunc("GET /api/prefs/ui", s.auth(s.getUIPrefs))
m.HandleFunc("PUT /api/prefs/ui", s.auth(s.putUIPrefs))
// Go runtime profiles, behind the login, for chasing memory/CPU (bug 14).
m.Handle("/api/debug/pprof/", s.auth(func(w http.ResponseWriter, r *http.Request) {
r.URL.Path = strings.TrimPrefix(r.URL.Path, "/api")
pprof.Index(w, r)
}))
m.HandleFunc("GET /api/prefs/sampler", s.auth(s.getSampler))
m.HandleFunc("PUT /api/prefs/sampler", s.auth(s.putSampler))
m.HandleFunc("GET /api/prefs/history", s.auth(s.getHistoryWindow))
m.HandleFunc("PUT /api/prefs/history", s.auth(s.putHistoryWindow))
m.HandleFunc("GET /api/kvms", s.auth(s.listKVMs))
m.HandleFunc("POST /api/kvms", s.auth(s.addKVM))
m.HandleFunc("DELETE /api/kvms/{id}", s.auth(s.removeKVM))
m.HandleFunc("/kvm/{id}/{rest...}", s.auth(s.serveKVM))
m.HandleFunc("/kvm/{id}", s.auth(s.serveKVM))
// A path Ham Lab does not serve may still belong to a KVM's page, which asks for its
// own assets by absolute path. Ham Lab's own routes matched first, so this can only
// catch what would otherwise have been a 404.
m.HandleFunc("/api/", s.auth(func(w http.ResponseWriter, r *http.Request) {
if s.kvmFallback(w, r) {
return
}
jsonErr(w, 404, "not found")
}))
m.HandleFunc("/", s.static)
return m
}
// sensorPath reports whether a path belongs to the read-only sensor API.
func sensorPath(p string) bool {
if p == "/metrics" || p == "/api/sensors" || strings.HasPrefix(p, "/api/sensors/") || strings.HasPrefix(p, "/api/fleet/sensors/") {
return true
}
if rest, ok := strings.CutPrefix(p, "/api/nodes/"); ok {
if _, after, ok := strings.Cut(rest, "/"); ok {
return after == "sensors" || strings.HasPrefix(after, "sensors/")
}
}
return false
}
// Handler wraps the mux with headers a root-capable console should always send.
func (s *Server) Handler() http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
h := w.Header()
// The console must never be framed. Two exceptions: an embedded widget exists to be
// (a page that can only read sensors, and only with a read-only token in its URL),
// and a proxied KVM is framed by the console itself — 'none' forbids that even from
// the same origin, so those paths say 'self' instead of opening up.
switch {
case r.URL.Path == "/embed/widget":
case strings.HasPrefix(r.URL.Path, kvmPrefix):
h.Set("X-Frame-Options", "SAMEORIGIN")
h.Set("Content-Security-Policy", "frame-ancestors 'self'")
default:
h.Set("X-Frame-Options", "DENY")
h.Set("Content-Security-Policy", "frame-ancestors 'none'")
}
h.Set("X-Content-Type-Options", "nosniff")
// The sensor API is for other people's pages too. Any origin may read it — with a
// token, never the session: "*" is the one value browsers refuse to pair with
// cookies, so a foreign page can't ride a logged-in console. Nothing else is opened.
if sensorPath(r.URL.Path) {
h.Set("Access-Control-Allow-Origin", "*")
if r.Method == http.MethodOptions {
h.Set("Access-Control-Allow-Methods", "GET, POST")
h.Set("Access-Control-Allow-Headers", "Authorization, Content-Type")
h.Set("Access-Control-Max-Age", "86400")
w.WriteHeader(http.StatusNoContent)
return
}
}
h.Set("Referrer-Policy", "no-referrer")
s.mux.ServeHTTP(w, r)
})
}
// ---- helpers ----
func jsonOK(w http.ResponseWriter, code int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
_ = json.NewEncoder(w).Encode(v)
}
func jsonErr(w http.ResponseWriter, code int, msg string) {
jsonOK(w, code, map[string]string{"error": msg})
}
func secure(r *http.Request) bool {
return r.TLS != nil || strings.EqualFold(r.Header.Get("X-Forwarded-Proto"), "https")
}
func (s *Server) setCookie(w http.ResponseWriter, r *http.Request, tok string, maxAge int) {
http.SetCookie(w, &http.Cookie{
Name: cookieName, Value: tok, Path: "/", MaxAge: maxAge,
HttpOnly: true, Secure: secure(r), SameSite: http.SameSiteStrictMode,
})
}
// signedIn reports whether a request carries a live browser session, without the
// side effects of the auth middleware.
func (s *Server) signedIn(r *http.Request) bool {
c, err := r.Cookie(cookieName)
if err != nil {
return false
}
_, ok := s.sessions.Lookup(c.Value, clientKey(r))
return ok
}
// cookieActor resolves the session cookie to who is calling. answered means the caller has
// already been sent an error and the handler must not run: the account behind a live session
// has gone away, or a guest login's time is up — so a console left open on a visitor's phone
// stops at the expiry, not at whatever they click next.
func (s *Server) cookieActor(w http.ResponseWriter, r *http.Request) (a actor, ok, answered bool) {
c, err := r.Cookie(cookieName)
if err != nil {
return actor{}, false, false
}
ses, live := s.sessions.Lookup(c.Value, clientKey(r))
if !live {
return actor{}, false, false
}
var u config.User
var found bool
s.cfg.View(func(cf *config.Config) { u, found = cf.User(ses.User) })
if !found || u.Expired(time.Now()) {
s.sessions.Delete(c.Value)
s.setCookie(w, r, "", -1)
if found {
jsonErr(w, 401, "this guest login has expired")
} else {
jsonErr(w, 401, "that account no longer exists")
}
return actor{}, false, true
}
return actor{User: ses.User, Session: ses.ID, token: c.Value, Guest: u.Guest, Expires: u.Expires}, true, false
}
// auth accepts the browser session cookie or a bearer API token issued to a hub.
func (s *Server) auth(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
a, ok, answered := s.cookieActor(w, r)
if answered {
return
}
if ok {
next(w, withActor(r, a))
return
}
if h := r.Header.Get("Authorization"); strings.HasPrefix(h, "Bearer ") {
var hashes, names []string
s.cfg.View(func(c *config.Config) {
for _, t := range c.Tokens {
if t.Scope == "" { // a scoped token is not a key to the machine
hashes = append(hashes, t.Hash)
names = append(names, t.Name)
}
}
})
if i := auth.TokenIndex(strings.TrimPrefix(h, "Bearer "), hashes); i >= 0 {
next(w, withActor(r, actor{Token: names[i]}))
return
}
}
jsonErr(w, 401, "unauthenticated")
}
}
// sensorAuth guards the read-only sensor endpoints: a session, any token in the
// Authorization header, or a sensors:read token as ?token= — an iframe or a browser
// EventSource can't send headers, and a token that can only read fan speeds is one we
// can afford to see in a URL. Full-control tokens are never accepted from the URL.
//
// A sensors:read token may be narrowed to ID patterns; they ride the request context
// and every handler reads sensors through s.visible. A hub forwarding such a caller to
// a node states the patterns in filterHeader, which a node believes only from a caller
// holding full control (the hub) — so the header can narrow a request, never widen one.
func (s *Server) sensorAuth(next http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {
ok, trusted := false, false
var patterns []string
if a, live, answered := s.cookieActor(w, r); answered {
return
} else if live {
ok, trusted = true, true
r = withActor(r, a)
}
if !ok {
presented, fromURL := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer "), false
if presented == "" {
presented, fromURL = r.URL.Query().Get("token"), true
}
if presented != "" {
h := []byte(auth.HashToken(presented))
s.cfg.View(func(c *config.Config) {
for _, t := range c.Tokens {
if subtle.ConstantTimeCompare(h, []byte(t.Hash)) != 1 {
continue
}
if t.Scope == config.ScopeSensorsRead {
ok, patterns = true, t.Sensors
} else if t.Scope == "" && !fromURL {
ok, trusted = true, true
}
}
})
}
}
if !ok {
jsonErr(w, 401, "unauthenticated")
return
}
if f := r.Header.Get(filterHeader); trusted && f != "" {
patterns = strings.Split(f, ",")
}
next(w, r.WithContext(context.WithValue(r.Context(), filterKey{}, patterns)))
}
}
// maxNameLen bounds the names people give things (terminals, tokens): they are shown in
// tabs and lists, and nothing good is 300 characters long.
const maxNameLen = 64
const filterHeader = "X-Hamlab-Sensor-Filter"
type filterKey struct{}
// allowed is the caller's ID filter: nil when it may read everything.
func allowed(r *http.Request) func(id string) bool {
pats, _ := r.Context().Value(filterKey{}).([]string)
if len(pats) == 0 {
return nil
}
match, err := sensors.Matcher(strings.Join(pats, ","))
if err != nil {
return func(string) bool { return false } // a broken restriction opens nothing
}
return match
}
// visible is every sensor this caller may read.
func (s *Server) visible(r *http.Request) []sensors.Sensor {
all, ok := s.sensors.All(), allowed(r)
if ok == nil {
return all
}
out := all[:0:0]
for _, x := range all {
if ok(x.ID) {
out = append(out, x)
}
}
return out
}
// proxyNodeSensors lets a sensors:read token reach a managed machine's sensors through
// this hub. The hub authenticates to the node as itself; the caller's token stops here.
func (s *Server) proxyNodeSensors(w http.ResponseWriter, r *http.Request) {
if q := r.URL.Query(); q.Has("token") {
q.Del("token")
r.URL.RawQuery = q.Encode()
}
r.Header.Del(filterHeader)
if pats, _ := r.Context().Value(filterKey{}).([]string); len(pats) > 0 {
r.Header.Set(filterHeader, strings.Join(pats, ","))
}
s.proxyNode(w, r)
}
// issueReadToken makes a sensors:read token for a dashboard. Shown once, stored hashed.
func (s *Server) issueReadToken(w http.ResponseWriter, r *http.Request) {
var body struct {
Name string `json:"name"`
Sensors []string `json:"sensors"` // optional ID patterns the token is limited to
}
if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil || strings.TrimSpace(body.Name) == "" {
jsonErr(w, 400, "name required")
return
}
var pats []string
for _, p := range body.Sensors {
if p = strings.TrimSpace(p); p != "" && !strings.Contains(p, ",") {
pats = append(pats, p)
}
}
if _, err := sensors.Matcher(strings.Join(pats, ",")); err != nil {
jsonErr(w, 400, err.Error())
return
}
name := strings.TrimSpace(body.Name)
if len(name) > maxNameLen || strings.ContainsAny(name, "/\\<>\"") {
jsonErr(w, 400, fmt.Sprintf("token names are at most %d characters, without / \\ < > or quotes", maxNameLen))
return
}
tok, hash, err := auth.NewToken()
if err != nil {
jsonErr(w, 500, "token error")
return
}
clash := false
if err := s.cfg.Update(func(c *config.Config) {
for _, t := range c.Tokens {
if t.Name == name {
clash = true
return
}
}
c.Tokens = append(c.Tokens, config.Token{Name: name, Hash: hash, Created: time.Now(), Scope: config.ScopeSensorsRead, Sensors: pats, IssuedBy: actorOf(r).User})
}); err != nil {
jsonErr(w, 500, err.Error())
return
}
if clash {
jsonErr(w, 409, "a token with that name exists; revoke it first")
return
}
log.Printf("read-only sensor token %q created (from %s)", name, clientIP(r))
jsonOK(w, 200, map[string]string{"token": tok, "name": name, "scope": config.ScopeSensorsRead})
}
// ---- API tokens: how a hub authenticates to this node ----
// issueToken trades the password for a long-lived token, named by the caller (a hub).
// Rate-limited like login; an existing token with the same name is replaced.
func (s *Server) issueToken(w http.ResponseWriter, r *http.Request) {
if err := s.limiter.Allow(clientKey(r)); err != nil {
jsonErr(w, 429, "too many attempts, try again later")
return
}
var body struct {
Username string `json:"username"`
Password string `json:"password"`
Name string `json:"name"`
}
if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil || strings.TrimSpace(body.Name) == "" {
jsonErr(w, 400, "password and name required")
return
}
user, ok := s.checkLogin(body.Username, body.Password)
if !ok {
s.limiter.Fail(clientKey(r))
log.Printf("token request failed for %q from %s", body.Username, clientIP(r))
if body.Username == "" {
jsonErr(w, 401, "wrong password, or this machine has several users: give a user name")
return
}
jsonErr(w, 401, "wrong user name or password")
return
}
s.limiter.Reset(clientKey(r))
s.touchUser(user)
tok, hash, err := auth.NewToken()
if err != nil {
jsonErr(w, 500, "token error")
return
}
name := strings.TrimSpace(body.Name)
if err := s.cfg.Update(func(c *config.Config) {
kept := c.Tokens[:0]
for _, t := range c.Tokens {
if t.Name != name {
kept = append(kept, t)
}
}
c.Tokens = append(kept, config.Token{Name: name, Hash: hash, Created: time.Now(), IssuedBy: user})
}); err != nil {
jsonErr(w, 500, err.Error())
return
}
log.Printf("API token %q issued to %s by %s", name, clientIP(r), user)
jsonOK(w, 200, map[string]string{"token": tok, "name": name})
}
func (s *Server) listTokens(w http.ResponseWriter, _ *http.Request) {
var out []map[string]any
s.cfg.View(func(c *config.Config) {
for _, t := range c.Tokens {
out = append(out, map[string]any{"name": t.Name, "created": t.Created, "scope": t.Scope, "sensors": t.Sensors, "issued_by": t.IssuedBy})
}
})
if out == nil {
out = []map[string]any{}
}
jsonOK(w, 200, out)
}
func (s *Server) revokeToken(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
found := false
_ = s.cfg.Update(func(c *config.Config) {
kept := c.Tokens[:0]
for _, t := range c.Tokens {
if t.Name == name {
found = true
continue
}
kept = append(kept, t)
}
c.Tokens = kept
})
if !found {
jsonErr(w, 404, "no such token")
return
}
log.Printf("API token %q revoked (from %s)", name, clientIP(r))
w.WriteHeader(204)
}
// ---- nodes: the machines this hub manages ----
func (s *Server) listNodes(w http.ResponseWriter, _ *http.Request) {
jsonOK(w, 200, map[string]any{"nodes": s.nodes.List(), "host": s.hostName(), "fingerprint": s.fingerprint})
}
func (s *Server) addNode(w http.ResponseWriter, r *http.Request) {
var body struct{ Name, URL, Username, Password string }
if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil {
jsonErr(w, 400, "bad request")
return
}
st, err := s.nodes.Add(r.Context(), strings.TrimSpace(body.Name), body.URL, strings.TrimSpace(body.Username), body.Password)
if err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("nodes: added %s (%s) from %s", st.Name, st.URL, clientIP(r))
jsonOK(w, 201, st)
}
func (s *Server) removeNode(w http.ResponseWriter, r *http.Request) {
name := r.PathValue("name")
if err := s.nodes.Remove(r.Context(), name); err != nil {
jsonErr(w, 404, err.Error())
return
}
log.Printf("nodes: removed %s (from %s)", name, clientIP(r))
w.WriteHeader(204)
}
// selfUpdate receives a new binary from a hub, checks it runs, installs it over the
// current executable (keeping a .prev copy) and restarts the service.
func (s *Server) selfUpdate(w http.ResponseWriter, r *http.Request) {
if err := s.updateAllowed(r.URL.Query().Get("force") == "1"); err != nil {
jsonErr(w, 409, err.Error())
return
}
tmp := filepath.Join(s.cfg.DataDir(), "hamlab.new")
f, err := os.OpenFile(tmp, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o755)
if err != nil {
jsonErr(w, 500, err.Error())
return
}
n, err := io.Copy(f, io.LimitReader(r.Body, 512<<20))
f.Close()
if err != nil || n < 1<<20 {
os.Remove(tmp)
jsonErr(w, 400, "upload failed or too small to be hamlab")
return
}
newVersion, err := s.installBinary(r.Context(), tmp)
if err != nil {
jsonErr(w, codeOf(err, 500), err.Error())
return
}
log.Printf("self-update: installed %s (was %s) from %s; restarting", newVersion, s.version, clientIP(r))
jsonOK(w, 200, map[string]string{"version": newVersion, "previous": s.version})
s.restartAfterUpdate()
}
// updateAllowed refuses to replace the binary under a package task or an attached terminal,
// since the restart drops both. force says to do it anyway.
func (s *Server) updateAllowed(force bool) error {
if force {
return nil
}
if t, ok := s.tasks.RunningOf("updates"); ok {
return errors.New("busy: a package task is running (" + t.Title + "); retry with force to interrupt it")
}
attached := 0
for _, t := range s.mgr.List() {
attached += t.Clients
}
if attached > 0 {
return fmt.Errorf("busy: %d terminal client(s) attached; retry with force to drop them", attached)
}
return nil
}
// httpErr carries the status a failure deserves, so one install path serves both the
// upload handler and the updater.
type httpErr struct {
code int
msg string
}
func (e httpErr) Error() string { return e.msg }
func codeOf(err error, dflt int) int {
var h httpErr
if errors.As(err, &h) {
return h.code
}
return dflt
}
// installBinary checks that the file at tmp is a working hamlab for this machine and puts
// it in place of the running one, keeping the old one as hamlab.prev. It returns the new
// version. The caller restarts.
func (s *Server) installBinary(ctx context.Context, tmp string) (string, error) {
defer os.Remove(tmp)
exe, err := os.Executable()
if err != nil {
return "", err
}
exe, _ = filepath.EvalSymlinks(exe)
ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
defer cancel()
// It must at least run and identify itself before it replaces us.
out, err := exec.CommandContext(ctx, tmp, "version").Output()
if err != nil || !strings.HasPrefix(string(out), "hamlab ") {
return "", httpErr{400, "that file is not a working hamlab binary for this machine (" + runtime.GOARCH + ")"}
}
newVersion := strings.TrimSpace(strings.TrimPrefix(string(out), "hamlab "))
// Builds that have `check` must also load this machine's config and state.
if msg, err := exec.CommandContext(ctx, tmp, "-config", s.cfg.Path(), "check").CombinedOutput(); err != nil && !bytes.Contains(msg, []byte("usage:")) {
return "", httpErr{400, "the new build can't start here: " + strings.TrimSpace(string(msg))}
}
prev := filepath.Join(s.cfg.DataDir(), "hamlab.prev")
_ = copyFile(exe, prev)
// Atomic replace, by the new binary itself: directly when the directory is ours
// (root, or a user-owned prefix), else as root.
var msg []byte
if err = setup.CopyAtomic(tmp, exe); err != nil && priv.Available() {
full := priv.Wrap(tmp, "replace", exe)
if msg, err = exec.CommandContext(ctx, full[0], full[1:]...).CombinedOutput(); err != nil { // a build from before `replace`
full = priv.Wrap("install", "-m", "0755", tmp, exe)
msg, err = exec.CommandContext(ctx, full[0], full[1:]...).CombinedOutput()
}
}
if err != nil {
return "", errors.New("install failed: " + strings.TrimSpace(string(msg)+" "+err.Error()))
}
return newVersion, nil
}
// restartAfterUpdate becomes the binary now on disk; if that fails the old one goes back
// and we exit, so the supervisor starts something that works.
func (s *Server) restartAfterUpdate() {
go func() {
time.Sleep(800 * time.Millisecond)
exe, _ := os.Executable()
exe, _ = filepath.EvalSymlinks(exe)
prev := filepath.Join(s.cfg.DataDir(), "hamlab.prev")
if s.Restart != nil {
log.Printf("self-update: exec failed (%v); putting %s back", s.Restart(), s.version)
full := priv.Wrap(prev, "replace", exe)
_ = exec.Command(full[0], full[1:]...).Run()
}
os.Exit(1) // the supervisor starts whatever is on disk
}()
}
func copyFile(src, dst string) error {
in, err := os.Open(src)
if err != nil {
return err
}
defer in.Close()
out, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0o755)
if err != nil {
return err
}
if _, err := io.Copy(out, in); err != nil {
out.Close()
return err
}
return out.Close()
}
func (s *Server) updateMachine(w http.ResponseWriter, r *http.Request) {
exe, _ := os.Executable()
res := s.nodes.Push(r.Context(), r.PathValue("name"), exe, runtime.GOARCH, r.URL.Query().Get("force") == "1")
log.Printf("machines: update %s -> ok=%v %s %s (from %s)", res.Name, res.OK, res.Version, res.Error, clientIP(r))
if !res.OK {
code := 400
if strings.HasPrefix(res.Error, "busy:") {
code = 409
}
jsonErr(w, code, res.Error)
return
}
jsonOK(w, 200, res)
}
func (s *Server) updateAllMachines(w http.ResponseWriter, r *http.Request) {
exe, _ := os.Executable()
res := s.nodes.PushAll(r.Context(), exe, runtime.GOARCH, r.URL.Query().Get("force") == "1")
for _, x := range res {
log.Printf("machines: update %s -> ok=%v %s %s (from %s)", x.Name, x.OK, x.Version, x.Error, clientIP(r))
}
jsonOK(w, 200, res)
}
func (s *Server) proxyNode(w http.ResponseWriter, r *http.Request) {
p, ok := s.nodes.Proxy(r.PathValue("name"))
if !ok {
jsonErr(w, 404, "no such machine")
return
}
p.ServeHTTP(w, r)
}
// ---- auth ----
func (s *Server) login(w http.ResponseWriter, r *http.Request) {
if err := s.limiter.Allow(clientKey(r)); err != nil {
jsonErr(w, 429, "too many attempts, try again later")
return
}
var body struct {
Username string `json:"username"`
Password string `json:"password"`
}
if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil {
jsonErr(w, 400, "bad request")
return
}
name, ok := s.checkLogin(body.Username, body.Password)
if !ok {
s.limiter.Fail(clientKey(r))
log.Printf("login failed for %q from %s", body.Username, clientIP(r))
jsonErr(w, 401, "wrong user name or password")
return
}
s.limiter.Reset(clientKey(r))
tok, err := s.sessions.Create(name, clientKey(r), r.UserAgent())
if err != nil {
jsonErr(w, 500, "session error")
return
}
s.setCookie(w, r, tok, int(auth.SessionTTL.Seconds()))
s.touchUser(name)
log.Printf("login ok: %s from %s", name, clientIP(r))
jsonOK(w, 200, map[string]string{"user": name})
}
// checkLogin verifies a user name and password, returning the account. An empty user
// name is allowed only while there is exactly one account — what a client from before
// user names sends, and what a hub pairing with a one-user node may leave out.
func (s *Server) checkLogin(username, password string) (string, bool) {
var users, real []config.User
s.cfg.View(func(c *config.Config) { users, real = append(users, c.Users...), c.Accounts() })
username = strings.TrimSpace(username)
if username == "" && len(real) == 1 { // guest logins are not the account an empty name means
username = real[0].Name
}
if username == "" {
auth.CheckUser(nil, "", password) // take as long as a real check
return "", false
}
return username, auth.CheckUser(users, username, password)
}
// touchUser notes that an account was just used.
func (s *Server) touchUser(name string) {
_ = s.cfg.Update(func(c *config.Config) {
for i := range c.Users {
if c.Users[i].Name == name {
c.Users[i].LastSeen = time.Now()
}
}
})
}
// migrateUsers is the one-time move from a single shared password to accounts: the old
// hash becomes the account named after the OS user the console runs as, the sessions that
// were signed in with it become that account's, and the machine-wide look becomes theirs.
func (s *Server) migrateUsers() {
first := ""
changed := false
if err := s.cfg.Update(func(c *config.Config) {
changed = c.MigrateUsers(s.user)
if len(c.Users) > 0 {
first = c.Users[0].Name
}
}); err != nil {
log.Printf("users: %v", err)
}
if changed {
log.Printf("users: the login password is now the account %q", first)
}
if first == "" {
return
}
_ = s.store.Update(func(st *store.State) {
for tok, ses := range st.Sessions {
if ses.User == "" {
ses.User = first
if ses.ID == "" {
ses.ID = tok[:12] // an old session has no ID of its own; a slice of the token is not usable as one
}
st.Sessions[tok] = ses
}
}
if len(st.UI) > 0 {
if u := st.UserState(first); len(u.UI) == 0 {
u.UI = st.UI
}
st.UI = nil
}
})
}
func (s *Server) logout(w http.ResponseWriter, r *http.Request) {
if c, err := r.Cookie(cookieName); err == nil {
s.sessions.Delete(c.Value)
}
s.setCookie(w, r, "", -1)
w.WriteHeader(204)
}
// clientKey is the address the limiter buckets by: the first X-Forwarded-For hop when a
// reverse proxy set it, else the socket peer.
func clientKey(r *http.Request) string {
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
return strings.TrimSpace(strings.Split(xff, ",")[0])
}
host, _, err := net.SplitHostPort(r.RemoteAddr)
if err != nil {
return r.RemoteAddr
}
return host
}
// clientIP names the caller for a log line: who (once signed in) and from where.
func clientIP(r *http.Request) string {
ip := r.RemoteAddr
if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
ip = strings.TrimSpace(strings.Split(xff, ",")[0]) + " via " + r.RemoteAddr
}
if who := actorOf(r).String(); who != "" {
return who + "@" + ip
}
return ip
}
// ---- me ----
// hostName is this machine's hostname now, not when the daemon started: rename a machine
// (hostnamectl, or a DHCP lease) and the console says so within a few seconds — in its header,
// in the machine menu, in alerts and in the fleet's sensor queries — without a restart. Read
// through a short cache because it is asked for on most requests and on every alert pass.
func (s *Server) hostName() string {
s.hostMu.Lock()
defer s.hostMu.Unlock()
if time.Since(s.hostAt) > 5*time.Second {
if h, err := os.Hostname(); err == nil && h != "" {
if s.host != "" && h != s.host {
log.Printf("hostname: %s is now %s", s.host, h)
}
s.host = h
}
s.hostAt = time.Now()
}
return s.host
}
func (s *Server) me(w http.ResponseWriter, r *http.Request) {
load := ""
if b, err := os.ReadFile("/proc/loadavg"); err == nil {
f := strings.Fields(string(b))
if len(f) >= 3 {
load = strings.Join(f[:3], " ")
}
}
uptime := 0.0
if b, err := os.ReadFile("/proc/uptime"); err == nil {
if f := strings.Fields(string(b)); len(f) > 0 {
uptime, _ = json.Number(f[0]).Float64()
}
}
_, rebootErr := os.Stat("/run/reboot-required")
updateVersion, _ := s.behind()
var addr string
s.cfg.View(func(c *config.Config) {
if h, _, err := net.SplitHostPort(c.Listen); err == nil && h != "" && h != "0.0.0.0" && h != "::" {
addr = h // the address this console listens on, for the machine menu
}
})
me := actorOf(r)
var guestUntil any
if me.Guest {
guestUntil = me.Expires
}
jsonOK(w, 200, map[string]any{
"host": s.hostName(), "user": s.user, "account": me.User, "version": s.version,
"guest": me.Guest, "guest_until": guestUntil,
"load": load, "uptime": uptime, "service_started": s.started,
"reboot_needed": rebootErr == nil, "fingerprint": s.fingerprint, "arch": runtime.GOARCH, "addr": addr,
"update_available": updateVersion,
})
}
// ---- sensors: the standard read interface (SENSORS.md) ----
// sensorCatalog lists every sensor with its current value; ?ids= narrows it with
// comma-separated IDs or globs.
func (s *Server) sensorCatalog(w http.ResponseWriter, r *http.Request) {
all := s.visible(r)
if ids := r.URL.Query().Get("ids"); ids != "" {
picked := []sensors.Sensor{}
for _, sel := range strings.Split(ids, ",") {
got, err := sensors.Select(all, strings.TrimSpace(sel))
if err != nil {
jsonErr(w, 400, err.Error())
return
}
picked = append(picked, got...)
}
all = picked
}
jsonOK(w, 200, map[string]any{"sensors": all})
}
// sensorOne is a single sensor; ?format=text is the bare value, for shell scripts and
// anything else that can only fetch a URL and read a number.
func (s *Server) sensorOne(w http.ResponseWriter, r *http.Request) {
got, _ := sensors.Select(s.visible(r), r.PathValue("id"))
if len(got) != 1 {
jsonErr(w, 404, "no such sensor")
return
}
if r.URL.Query().Get("format") == "text" {
w.Header().Set("Content-Type", "text/plain; charset=utf-8")
fmt.Fprintln(w, got[0].Value)
return
}
jsonOK(w, 200, got[0])
}
func (s *Server) sensorHistory(w http.ResponseWriter, r *http.Request) {
h, ok := s.sensors.History(r.PathValue("id"))
if may := allowed(r); may != nil && !may(r.PathValue("id")) {
ok = false
}
if !ok {
jsonErr(w, 404, "no history for that sensor")
return
}
jsonOK(w, 200, h)
}
// sensorHistories is many histories in one request: ?ids= takes IDs and globs.
func (s *Server) sensorHistories(w http.ResponseWriter, r *http.Request) {
match, err := sensors.Matcher(r.URL.Query().Get("ids"))
if err != nil {
jsonErr(w, 400, err.Error())
return
}
if may := allowed(r); may != nil {
asked := match
match = func(id string) bool { return asked(id) && may(id) }
}
jsonOK(w, 200, map[string]any{"series": s.sensors.Histories(match)})
}
// sensorProm is every numeric sensor in the Prometheus text format.
func (s *Server) sensorProm(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
sensors.WriteProm(w, s.visible(r))
}
// queryOf reads a shaped query from a request: the POST body, or on a GET either ?q= (the
// JSON shape, URL-encoded — a browser EventSource can only GET) or the flat form, where
// every other parameter is a key and its selector.
func queryOf(w http.ResponseWriter, r *http.Request) (q json.RawMessage, full bool, err error) {
params := r.URL.Query()
full = params.Get("full") == "1"
if r.Method == http.MethodPost {
b, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 64<<10))
if err != nil {
return nil, full, errors.New("query larger than 64 KiB")
}
return b, full, nil
}
if raw := params.Get("q"); raw != "" {
return json.RawMessage(raw), full, nil
}
flat := map[string]string{}
for k, v := range params {
if !reservedParam[k] && len(v) > 0 {
flat[k] = v[0]
}
}
q, _ = json.Marshal(flat)
return q, full, nil
}
var reservedParam = map[string]bool{"full": true, "token": true, "q": true, "every": true, "machines": true}
// A stream holds a goroutine and a connection for as long as the client likes, and a
// read-only token may be public. So streams are counted: per client address and in all.
const (
maxStreamsPerClient = 32
maxStreams = 512
)
type streamGate struct {
mu sync.Mutex
by map[string]int
total int
}
func (g *streamGate) enter(who string) bool {
g.mu.Lock()
defer g.mu.Unlock()
if g.by == nil {
g.by = map[string]int{}
}
if g.total >= maxStreams || g.by[who] >= maxStreamsPerClient {
return false
}
g.by[who]++
g.total++
return true
}
func (g *streamGate) leave(who string) {
g.mu.Lock()
defer g.mu.Unlock()
if g.by[who] <= 0 {
return
}
if g.by[who]--; g.by[who] == 0 {
delete(g.by, who)
}
g.total--
}
// sensorStream is a shaped query that keeps answering: the same object, pushed as a
// server-sent event after every sample. ?every=<ms> spaces the events out for a consumer
// that doesn't want them at the sampling rate.
func (s *Server) sensorStream(w http.ResponseWriter, r *http.Request) {
q, full, err := queryOf(w, r)
if err == nil {
_, err = sensors.Shape(s.visible(r), q, full) // refuse a bad query before the stream starts
}
if err != nil {
jsonErr(w, 400, err.Error())
return
}
who := clientKey(r)
if !s.streams.enter(who) {
jsonErr(w, 429, fmt.Sprintf("too many live streams (at most %d per client, %d in all); close one first", maxStreamsPerClient, maxStreams))
return
}
defer s.streams.leave(who)
tick, cancel, ok := s.sensors.Subscribe()
fl, canFlush := w.(http.Flusher)
if !ok || !canFlush {
jsonErr(w, 501, "streaming isn't available here")
return
}
defer cancel()
every := time.Duration(0)
if ms, err := strconv.Atoi(r.URL.Query().Get("every")); err == nil && ms > 0 {
every = time.Duration(ms) * time.Millisecond
}
// Asking for roughly the sampling rate means "every sample": don't let timer jitter
// drop alternate ticks.
if iv := s.metrics.Interval(); every <= iv+iv/2 {
every = 0
}
h := w.Header()
h.Set("Content-Type", "text/event-stream")
h.Set("Cache-Control", "no-store")
h.Set("X-Accel-Buffering", "no") // nginx (NPM) would otherwise hold events back
send := func() bool {
out, _ := sensors.Shape(s.visible(r), q, full)
b, err := json.Marshal(out)
if err != nil {
return false
}
if _, err := fmt.Fprintf(w, "data: %s\n\n", b); err != nil {
return false
}
fl.Flush()
return true
}
if !send() {
return
}
last := time.Now()
keepalive := time.NewTicker(25 * time.Second)
defer keepalive.Stop()
for {
select {
case <-r.Context().Done():
return
case <-keepalive.C:
if _, err := io.WriteString(w, ": keepalive\n\n"); err != nil {
return
}
fl.Flush()
case <-tick:
if every > 0 && time.Since(last) < every-5*time.Millisecond {
continue
}
last = time.Now()
if !send() {
return
}
}
}
}
// sensorQuery answers a shaped query. POST sends the shape as the body; GET builds a
// flat one from the query string (?cpu=cpu.usage&fans=fan.*). ?full=1 returns whole
// sensors instead of bare values.
func (s *Server) sensorQuery(w http.ResponseWriter, r *http.Request) {
q, full, err := queryOf(w, r)
if err != nil {
jsonErr(w, 413, err.Error())
return
}
out, err := sensors.Shape(s.visible(r), q, full)
if err != nil {
jsonErr(w, 400, err.Error())
return
}
jsonOK(w, 200, out)
}
// fleetQuery asks several machines one shaped query in one request and answers
// {machine: result}. POST {"machines": "*" | ["grid", …], "query": <shape>}, or GET with
// the flat form plus ?machines=grid,wyse (default: all). This machine answers under its
// own hostname. One that can't be reached answers {"error": …}; the rest still do.
func (s *Server) fleetQuery(w http.ResponseWriter, r *http.Request) {
var q json.RawMessage
var want []string
full := r.URL.Query().Get("full") == "1"
if r.Method == http.MethodPost {
var body struct {
Machines json.RawMessage `json:"machines"`
Query json.RawMessage `json:"query"`
}
if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)).Decode(&body); err != nil || len(body.Query) == 0 {
jsonErr(w, 400, `expected {"machines": "*" or [names], "query": <shape>}`)
return
}
q = body.Query
_ = json.Unmarshal(body.Machines, &want) // "*" or absent leaves it empty: everyone
} else {
q, full, _ = queryOf(w, r)
if m := r.URL.Query().Get("machines"); m != "" && m != "*" {
want = strings.Split(m, ",")
}
}
local, err := sensors.Shape(s.visible(r), q, full)
if err != nil {
jsonErr(w, 400, err.Error())
return
}
names := append([]string{s.hostName()}, s.nodes.Names()...)
if len(want) > 0 {
names = names[:0]
for _, n := range want {
names = append(names, strings.TrimSpace(n))
}
}
hdr := http.Header{"Content-Type": {"application/json"}}
if pats, _ := r.Context().Value(filterKey{}).([]string); len(pats) > 0 {
hdr.Set(filterHeader, strings.Join(pats, ","))
}
path := "/api/sensors/query"
if full {
path += "?full=1"
}
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
out := make(map[string]any, len(names))
var mu sync.Mutex
var wg sync.WaitGroup
for _, name := range names {
if name == s.hostName() {
out[name] = local
continue
}
wg.Add(1)
go func(name string) {
defer wg.Done()
var res any
code, b, err := s.nodes.Do(ctx, name, "POST", path, q, hdr)
switch {
case err != nil:
res = map[string]string{"error": err.Error()}
case code != 200 || json.Unmarshal(b, &res) != nil:
res = map[string]string{"error": fmt.Sprintf("HTTP %d", code)}
}
mu.Lock()
out[name] = res
mu.Unlock()
}(name)
}
wg.Wait()
jsonOK(w, 200, out)
}
// ---- services (systemd) ----
func (s *Server) listServices(w http.ResponseWriter, r *http.Request) {
list, err := s.svc.List(r.Context())
if err != nil {
jsonErr(w, 500, err.Error())
return
}
jsonOK(w, 200, map[string]any{"units": list, "sudo": s.svc.Sudo()})
}
func (s *Server) showService(w http.ResponseWriter, r *http.Request) {
d, err := s.svc.Show(r.Context(), r.PathValue("unit"))
if err != nil {
jsonErr(w, 404, err.Error())
return
}
jsonOK(w, 200, d)
}
func (s *Server) serviceFile(w http.ResponseWriter, r *http.Request) {
out, err := s.svc.Cat(r.Context(), r.PathValue("unit"))
if err != nil {
jsonErr(w, 404, err.Error())
return
}
jsonOK(w, 200, map[string]string{"content": out})
}
func logCount(r *http.Request) int {
n, _ := strconv.Atoi(r.URL.Query().Get("n"))
if n <= 0 {
n = 200
}
if n > 5000 {
n = 5000
}
return n
}
func (s *Server) serviceLogs(w http.ResponseWriter, r *http.Request) {
entries, err := s.svc.Journal(r.Context(), r.PathValue("unit"), logCount(r))
if err != nil {
jsonErr(w, 500, err.Error())
return
}
jsonOK(w, 200, entries)
}
// serviceTail streams journal entries as JSON text frames until the client goes away.
func (s *Server) serviceTail(w http.ResponseWriter, r *http.Request) {
unit := r.PathValue("unit")
if !services.Valid(unit) {
jsonErr(w, 400, "invalid unit name")
return
}
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{CompressionMode: websocket.CompressionDisabled})
if err != nil {
return
}
defer conn.Close(websocket.StatusNormalClosure, "")
ctx, cancel := context.WithCancel(r.Context())
defer cancel()
go func() { // reader: only to notice the close (and keep pings flowing)
defer cancel()
for {
if _, _, err := conn.Read(ctx); err != nil {
return
}
}
}()
go func() {
t := time.NewTicker(25 * time.Second)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
pctx, pc := context.WithTimeout(ctx, 10*time.Second)
err := conn.Ping(pctx)
pc()
if err != nil {
cancel()
return
}
}
}
}()
_ = s.svc.Tail(ctx, unit, logCount(r), func(e services.Entry) {
b, _ := json.Marshal(e)
if err := conn.Write(ctx, websocket.MessageText, b); err != nil {
cancel()
}
})
}
func (s *Server) serviceAction(w http.ResponseWriter, r *http.Request) {
unit, action := r.PathValue("unit"), r.PathValue("action")
if err := s.svc.Action(r.Context(), action, unit); err != nil {
log.Printf("systemctl %s %s: %v", action, unit, err)
jsonErr(w, 400, err.Error())
return
}
log.Printf("systemctl %s %s ok (from %s)", action, unit, clientIP(r))
w.WriteHeader(204)
}
// ---- cron (crontabs + systemd timers) ----
func (s *Server) cronOverview(w http.ResponseWriter, r *http.Request) {
jsonOK(w, 200, s.cron.Overview(r.Context()))
}
func (s *Server) cronSource(w http.ResponseWriter, r *http.Request) {
text, editable, err := s.cron.Source(r.Context(), r.URL.Query().Get("name"))
if err != nil {
jsonErr(w, 404, err.Error())
return
}
jsonOK(w, 200, map[string]any{"text": text, "editable": editable})
}
func (s *Server) cronWriteSource(w http.ResponseWriter, r *http.Request) {
var body struct {
Text string `json:"text"`
}
if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&body); err != nil {
jsonErr(w, 400, "bad request")
return
}
name := r.URL.Query().Get("name")
if err := s.cron.WriteSource(r.Context(), name, body.Text); err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("cron: wrote %s (from %s)", name, clientIP(r))
w.WriteHeader(204)
}
func (s *Server) cronJob(w http.ResponseWriter, r *http.Request) (cron.Job, bool) {
j, ok := s.cron.FindJob(r.Context(), r.PathValue("id"))
if !ok {
jsonErr(w, 404, "no such job (crontab may have changed; reload)")
}
return j, ok
}
func (s *Server) cronJobLogs(w http.ResponseWriter, r *http.Request) {
j, ok := s.cronJob(w, r)
if !ok {
return
}
entries, err := s.cron.JobLogs(r.Context(), j, logCount(r))
if err != nil {
jsonErr(w, 500, err.Error())
return
}
jsonOK(w, 200, entries)
}
func (s *Server) cronJobOutput(w http.ResponseWriter, r *http.Request) {
j, ok := s.cronJob(w, r)
if !ok {
return
}
out, err := s.cron.OutputTail(r.Context(), j, logCount(r))
if err != nil {
jsonErr(w, 400, err.Error())
return
}
jsonOK(w, 200, map[string]string{"file": j.OutFile, "text": out})
}
func (s *Server) cronJobAction(w http.ResponseWriter, r *http.Request) {
j, ok := s.cronJob(w, r)
if !ok {
return
}
action := r.PathValue("action")
var err error
switch action {
case "enable", "disable":
err = s.cron.Toggle(r.Context(), j, action == "enable")
case "run":
var run *cron.Run
if run, err = s.cron.StartRun(r.Context(), j); err == nil {
log.Printf("cron: run now %q as %s (from %s)", j.Command, j.User, clientIP(r))
jsonOK(w, 202, run.Snapshot())
return
}
default:
jsonErr(w, 400, "unknown action")
return
}
if err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("cron: %s %q in %s (from %s)", action, j.Command, j.Source, clientIP(r))
w.WriteHeader(204)
}
func (s *Server) cronRun(w http.ResponseWriter, r *http.Request) {
run, ok := s.cron.GetRun(r.PathValue("id"))
if !ok {
jsonErr(w, 404, "no such run")
return
}
jsonOK(w, 200, run.Snapshot())
}
func (s *Server) cronTimerAction(w http.ResponseWriter, r *http.Request) {
unit, action := r.PathValue("unit"), r.PathValue("action")
if err := s.cron.TimerAction(r.Context(), unit, action); err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("timer %s %s ok (from %s)", action, unit, clientIP(r))
w.WriteHeader(204)
}
// ---- tasks (background commands with followable output) ----
func (s *Server) listTasks(w http.ResponseWriter, _ *http.Request) { jsonOK(w, 200, s.tasks.List()) }
func (s *Server) getTask(w http.ResponseWriter, r *http.Request) {
t, ok := s.tasks.Get(r.PathValue("id"))
if !ok {
jsonErr(w, 404, "no such task")
return
}
jsonOK(w, 200, t.Snapshot(true))
}
func (s *Server) cancelTask(w http.ResponseWriter, r *http.Request) {
t, ok := s.tasks.Get(r.PathValue("id"))
if !ok {
jsonErr(w, 404, "no such task")
return
}
t.Cancel()
log.Printf("task %s cancelled (from %s)", t.ID, clientIP(r))
w.WriteHeader(204)
}
// streamTask sends the output so far, then live chunks as binary frames, then {"type":"exit"}.
func (s *Server) streamTask(w http.ResponseWriter, r *http.Request) {
t, ok := s.tasks.Get(r.PathValue("id"))
if !ok {
jsonErr(w, 404, "no such task")
return
}
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{CompressionMode: websocket.CompressionDisabled})
if err != nil {
return
}
defer conn.Close(websocket.StatusNormalClosure, "")
ctx, cancel := context.WithCancel(r.Context())
defer cancel()
go func() {
defer cancel()
for {
if _, _, err := conn.Read(ctx); err != nil {
return
}
}
}()
backlog, ch, unsub := t.Subscribe()
defer unsub()
if len(backlog) > 0 {
if err := conn.Write(ctx, websocket.MessageBinary, backlog); err != nil {
return
}
}
ping := time.NewTicker(25 * time.Second)
defer ping.Stop()
for {
select {
case <-ctx.Done():
return
case <-ping.C:
pctx, pc := context.WithTimeout(ctx, 10*time.Second)
err := conn.Ping(pctx)
pc()
if err != nil {
return
}
case chunk, ok := <-ch:
if !ok {
snap := t.Snapshot(false)
b, _ := json.Marshal(map[string]any{"type": "exit", "code": snap.ExitCode, "error": snap.Err})
_ = conn.Write(ctx, websocket.MessageText, b)
return
}
if err := conn.Write(ctx, websocket.MessageBinary, chunk); err != nil {
return
}
}
}
}
// ---- updates ----
func (s *Server) updatesStatus(w http.ResponseWriter, r *http.Request) {
jsonOK(w, 200, s.updates.Status(r.Context()))
}
func (s *Server) updatesRefresh(w http.ResponseWriter, r *http.Request) {
t, err := s.updates.Refresh()
if err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("updates: refresh started as task %s (from %s)", t.ID, clientIP(r))
jsonOK(w, 202, t.Snapshot(false))
}
func (s *Server) updatesUpgrade(w http.ResponseWriter, r *http.Request) {
var body struct {
Packages []string `json:"packages"`
}
_ = json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body)
t, err := s.updates.Upgrade(body.Packages)
if err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("updates: %s started as task %s (from %s)", t.Title, t.ID, clientIP(r))
jsonOK(w, 202, t.Snapshot(false))
}
func (s *Server) updatesChangelog(w http.ResponseWriter, r *http.Request) {
text, err := s.updates.Changelog(r.Context(), r.URL.Query().Get("pkg"))
if err != nil && text == "" {
jsonErr(w, 400, err.Error())
return
}
jsonOK(w, 200, map[string]string{"text": text})
}
// power: reboot/poweroff, gated on the hostname being typed back.
func (s *Server) power(w http.ResponseWriter, r *http.Request) {
var body struct {
Confirm string `json:"confirm"`
}
_ = json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body)
if body.Confirm != s.hostName() {
jsonErr(w, 400, "confirmation must be the hostname")
return
}
action := r.PathValue("action")
log.Printf("POWER: %s requested from %s", action, clientIP(r))
if err := s.updates.Power(r.Context(), action); err != nil {
jsonErr(w, 400, err.Error())
return
}
w.WriteHeader(204)
}
// ---- files ----
// fsFor picks the backend: ?sudo=1 asks for root, which needs passwordless sudo.
func (s *Server) fsFor(w http.ResponseWriter, r *http.Request) (files.FS, bool) {
if r.URL.Query().Get("sudo") == "1" {
if s.fsSudo == nil {
jsonErr(w, 403, "root access needs passwordless sudo on the host")
return nil, false
}
return s.fsSudo, true
}
return s.fsLocal, true
}
func fsErr(w http.ResponseWriter, err error) {
code := 400
switch {
case errors.Is(err, os.ErrNotExist):
code = 404
case errors.Is(err, os.ErrPermission), strings.Contains(err.Error(), "permission denied"):
code = 403
}
msg := err.Error()
var pe *os.PathError
var le *os.LinkError
switch {
case errors.As(err, &pe): // "open /x: permission denied" -> "permission denied"
msg = pe.Err.Error()
case errors.As(err, &le): // "rename /a /b: no such file" -> "no such file or directory"
msg = le.Err.Error()
}
jsonErr(w, code, msg)
}
func cleanQuery(w http.ResponseWriter, r *http.Request, key string) (string, bool) {
p, err := files.Clean(r.URL.Query().Get(key))
if err != nil {
jsonErr(w, 400, err.Error())
return "", false
}
return p, true
}
func (s *Server) filesList(w http.ResponseWriter, r *http.Request) {
fsys, ok := s.fsFor(w, r)
if !ok {
return
}
p, ok := cleanQuery(w, r, "path")
if !ok {
return
}
l, err := fsys.List(r.Context(), p)
if err != nil {
fsErr(w, err)
return
}
jsonOK(w, 200, l)
}
func (s *Server) filesStat(w http.ResponseWriter, r *http.Request) {
fsys, ok := s.fsFor(w, r)
if !ok {
return
}
p, ok := cleanQuery(w, r, "path")
if !ok {
return
}
in, err := fsys.Stat(r.Context(), p)
if err != nil {
fsErr(w, err)
return
}
jsonOK(w, 200, in)
}
// filesRead streams a file. /download adds an attachment header; ?max= caps the bytes.
func (s *Server) filesRead(w http.ResponseWriter, r *http.Request) {
fsys, ok := s.fsFor(w, r)
if !ok {
return
}
p, ok := cleanQuery(w, r, "path")
if !ok {
return
}
in, err := fsys.Stat(r.Context(), p)
if err != nil {
fsErr(w, err)
return
}
if in.Type == "dir" {
jsonErr(w, 400, "is a directory")
return
}
if in.Type == "other" {
jsonErr(w, 400, "not a regular file (device, pipe or socket)")
return
}
mx, _ := strconv.ParseInt(r.URL.Query().Get("max"), 10, 64)
if mx > 0 && in.Size > mx {
jsonErr(w, 413, "file too large")
return
}
f, err := fsys.Open(r.Context(), p)
if err != nil {
fsErr(w, err)
return
}
defer f.Close()
var body io.Reader = f
if mx > 0 { // stat can lie (size 0 on /proc); the cap must hold on the bytes actually sent
body = io.LimitReader(f, mx+1)
}
name := filepath.Base(p)
ct := mime.TypeByExtension(filepath.Ext(name))
if ct == "" {
if in.Text {
ct = "text/plain; charset=utf-8"
} else {
ct = "application/octet-stream"
}
}
w.Header().Set("Content-Type", ct)
w.Header().Set("Cache-Control", "no-store")
w.Header().Set("X-Content-Type-Options", "nosniff")
if strings.HasSuffix(r.URL.Path, "/download") {
w.Header().Set("Content-Disposition", "attachment; filename*=UTF-8''"+url.PathEscape(name))
} else {
w.Header().Set("Content-Disposition", "inline; filename*=UTF-8''"+url.PathEscape(name))
}
if in.Size > 0 && mx == 0 {
w.Header().Set("Content-Length", strconv.FormatInt(in.Size, 10))
}
_, _ = io.Copy(w, body)
}
func (s *Server) filesWrite(w http.ResponseWriter, r *http.Request) {
fsys, ok := s.fsFor(w, r)
if !ok {
return
}
p, ok := cleanQuery(w, r, "path")
if !ok {
return
}
if err := fsys.Write(r.Context(), p, io.LimitReader(r.Body, 8<<20)); err != nil {
fsErr(w, err)
return
}
log.Printf("files: wrote %s (from %s)", p, clientIP(r))
w.WriteHeader(204)
}
// filesUpload streams the body into <path>/<name>.
func (s *Server) filesUpload(w http.ResponseWriter, r *http.Request) {
fsys, ok := s.fsFor(w, r)
if !ok {
return
}
dir, ok := cleanQuery(w, r, "path")
if !ok {
return
}
name := r.URL.Query().Get("name")
if name == "" || name != filepath.Base(name) || name == "." || name == ".." {
jsonErr(w, 400, "bad file name")
return
}
dst := filepath.Join(dir, name)
if r.URL.Query().Get("overwrite") != "1" {
if _, err := fsys.Stat(r.Context(), dst); err == nil {
jsonErr(w, 409, "a file with that name already exists")
return
}
}
if err := fsys.Write(r.Context(), dst, r.Body); err != nil {
fsErr(w, err)
return
}
log.Printf("files: uploaded %s (from %s)", dst, clientIP(r))
w.WriteHeader(201)
}
func (s *Server) filesOp(w http.ResponseWriter, r *http.Request, fn func(fsys files.FS, body map[string]any, p string) error) {
fsys, ok := s.fsFor(w, r)
if !ok {
return
}
var body map[string]any
if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body); err != nil {
jsonErr(w, 400, "bad request")
return
}
p, err := files.Clean(str(body["path"]))
if err != nil {
jsonErr(w, 400, err.Error())
return
}
if err := fn(fsys, body, p); err != nil {
fsErr(w, err)
return
}
w.WriteHeader(204)
}
func str(v any) string { s, _ := v.(string); return s }
func (s *Server) filesMkdir(w http.ResponseWriter, r *http.Request) {
s.filesOp(w, r, func(fsys files.FS, _ map[string]any, p string) error {
log.Printf("files: mkdir %s (from %s)", p, clientIP(r))
return fsys.Mkdir(r.Context(), p)
})
}
func (s *Server) filesTouch(w http.ResponseWriter, r *http.Request) {
s.filesOp(w, r, func(fsys files.FS, _ map[string]any, p string) error {
log.Printf("files: create %s (from %s)", p, clientIP(r))
return fsys.Touch(r.Context(), p)
})
}
func (s *Server) filesRename(w http.ResponseWriter, r *http.Request) {
s.filesOp(w, r, func(fsys files.FS, body map[string]any, from string) error {
to, err := files.Clean(str(body["to"]))
if err != nil {
return err
}
log.Printf("files: rename %s -> %s (from %s)", from, to, clientIP(r))
return fsys.Rename(r.Context(), from, to)
})
}
func (s *Server) filesDelete(w http.ResponseWriter, r *http.Request) {
s.filesOp(w, r, func(fsys files.FS, body map[string]any, p string) error {
rec, _ := body["recursive"].(bool)
log.Printf("files: delete %s recursive=%v (from %s)", p, rec, clientIP(r))
return fsys.Delete(r.Context(), p, rec)
})
}
func (s *Server) filesChmod(w http.ResponseWriter, r *http.Request) {
s.filesOp(w, r, func(fsys files.FS, body map[string]any, p string) error {
perm, err := files.ParsePerm(str(body["mode"]))
if err != nil {
return err
}
log.Printf("files: chmod %04o %s (from %s)", perm, p, clientIP(r))
return fsys.Chmod(r.Context(), p, perm)
})
}
// ---- storage (Disks tab) ----
func (s *Server) updatesInstall(w http.ResponseWriter, r *http.Request) {
var body struct {
Packages []string `json:"packages"`
}
_ = json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body)
t, err := s.updates.Install(body.Packages)
if err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("updates: %s started as task %s (from %s)", t.Title, t.ID, clientIP(r))
jsonOK(w, 202, t.Snapshot(false))
}
func (s *Server) storageOverview(w http.ResponseWriter, r *http.Request) {
devs, err := storage.Devices(r.Context())
if err != nil {
jsonErr(w, 500, err.Error())
return
}
jsonOK(w, 200, map[string]any{"devices": devs, "zfs": s.storage.ZFS(r.Context()), "smartctl": s.storage.HasSmartctl()})
}
func (s *Server) storageSmart(w http.ResponseWriter, r *http.Request) {
sm := s.storage.Smart(r.Context(), r.PathValue("dev"), r.URL.Query().Get("force") == "1")
if sm.Error == "unknown device" {
jsonErr(w, 404, sm.Error)
return
}
jsonOK(w, 200, sm)
}
func (s *Server) storageSmartTest(w http.ResponseWriter, r *http.Request) {
if err := s.storage.SelfTest(r.Context(), r.PathValue("dev"), r.PathValue("kind")); err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("smart: %s test on %s (from %s)", r.PathValue("kind"), r.PathValue("dev"), clientIP(r))
w.WriteHeader(204)
}
func (s *Server) zfsSnapshots(w http.ResponseWriter, r *http.Request) {
snaps, err := s.storage.Snapshots(r.Context(), r.URL.Query().Get("dataset"))
if err != nil {
jsonErr(w, 400, err.Error())
return
}
jsonOK(w, 200, snaps)
}
func (s *Server) zfsScrub(w http.ResponseWriter, r *http.Request) {
if err := s.storage.Scrub(r.Context(), r.PathValue("pool"), r.PathValue("action")); err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("zfs: scrub %s %s (from %s)", r.PathValue("action"), r.PathValue("pool"), clientIP(r))
w.WriteHeader(204)
}
func (s *Server) zfsSnapshot(w http.ResponseWriter, r *http.Request) {
var body struct {
Dataset string `json:"dataset"`
Name string `json:"name"`
}
_ = json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body)
if err := s.storage.CreateSnapshot(r.Context(), body.Dataset, body.Name); err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("zfs: snapshot %s@%s (from %s)", body.Dataset, body.Name, clientIP(r))
w.WriteHeader(204)
}
// ---- processes ----
func (s *Server) procList(w http.ResponseWriter, _ *http.Request) {
jsonOK(w, 200, s.metrics.Processes())
}
func (s *Server) procSignal(w http.ResponseWriter, r *http.Request) {
pid, err := strconv.Atoi(r.PathValue("pid"))
if err != nil {
jsonErr(w, 400, "bad pid")
return
}
var body struct {
Signal string `json:"signal"`
}
_ = json.NewDecoder(io.LimitReader(r.Body, 1024)).Decode(&body)
sig := syscall.SIGTERM
if body.Signal == "KILL" {
sig = syscall.SIGKILL
}
if err := metrics.Signal(pid, sig); err != nil {
jsonErr(w, 400, err.Error())
return
}
log.Printf("procs: sent %s to pid %d (from %s)", sig, pid, clientIP(r))
w.WriteHeader(204)
}
// ---- account ----
func (s *Server) changePassword(w http.ResponseWriter, r *http.Request) {
var body struct{ Current, Password string }
if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil {
jsonErr(w, 400, "bad request")
return
}
if !s.confirmSelf(w, r, body.Current) {
return
}
if len(body.Password) < auth.MinPasswordLen {
jsonErr(w, 400, "password must be at least 12 characters")
return
}
if !s.setPassword(w, actorOf(r).User, body.Password) {
return
}
log.Printf("password changed (by %s)", clientIP(r))
w.WriteHeader(204)
}
// ---- ui prefs (theme) ----
// The look is the account's: theme and accents follow whoever is signed in. The refresh
// rate rides in the same object for the client's convenience but is the machine's — it is
// how often every machine samples — so it is kept once, machine-wide, and folded in here.
func (s *Server) getUIPrefs(w http.ResponseWriter, r *http.Request) {
prefs := s.userPrefs(r, func(u *store.UserState) json.RawMessage { return u.UI })
prefs["refreshMs"] = s.metrics.Interval().Milliseconds()
w.Header().Set("Cache-Control", "no-store")
jsonOK(w, 200, prefs)
}
// userPrefs reads one of the caller's pref blobs as a map (empty when none, or for a token).
func (s *Server) userPrefs(r *http.Request, pick func(*store.UserState) json.RawMessage) map[string]any {
out := map[string]any{}
user := actorOf(r).User
if user == "" {
return out
}
var raw json.RawMessage
s.store.View(func(st *store.State) {
if u := st.Users[user]; u != nil {
raw = pick(u)
}
})
if len(raw) > 0 {
_ = json.Unmarshal(raw, &out)
}
return out
}
func (s *Server) putUIPrefs(w http.ResponseWriter, r *http.Request) {
b, err := io.ReadAll(io.LimitReader(r.Body, 16<<10))
if err != nil || !json.Valid(b) || len(b) == 0 || b[0] != '{' {
jsonErr(w, 400, "expected a JSON object")
return
}
var p struct {
Theme *string `json:"theme"`
RefreshMs *float64 `json:"refreshMs"`
}
_ = json.Unmarshal(b, &p)
if p.RefreshMs != nil && (*p.RefreshMs < 100 || *p.RefreshMs > 60000) {
jsonErr(w, 400, "refreshMs must be between 100 and 60000")
return
}
if p.Theme != nil && (len(*p.Theme) > 32 || strings.ContainsAny(*p.Theme, "<>\"'")) {
jsonErr(w, 400, "theme must be a short identifier")
return
}
if user := actorOf(r).User; user != "" {
var m map[string]json.RawMessage
_ = json.Unmarshal(b, &m)
delete(m, "refreshMs")
kept, _ := json.Marshal(m)
if err := s.store.Update(func(st *store.State) { st.UserState(user).UI = kept }); err != nil {
jsonErr(w, 500, err.Error())
return
}
}
if p.RefreshMs != nil { // the refresh rate is also how often every machine samples
go s.setSampler(int(*p.RefreshMs), true)
}
w.WriteHeader(204)
}
// ---- how often the machine samples ----
//
// One rate for the fleet, the console's refresh rate: set in Settings (or the Top tab), saved
// with the UI prefs on the hub, and passed by the hub to every machine. Before this the
// sampler was told only by an open Dashboard page, so a machine whose Dashboard nobody had
// opened since the rate changed kept sampling at the old one, and every other page of it —
// Top, Processes, Disks, widgets — "ignored" the setting.
// setSampler applies a rate here and, from a hub, everywhere. The collector's own limits are
// 100 ms to 10 s; a slower console rate samples at 10 s.
func (s *Server) setSampler(ms int, fanOut bool) {
lo, hi := int(metrics.MinInterval.Milliseconds()), int(metrics.MaxInterval.Milliseconds())
if ms < lo {
ms = lo
}
if ms > hi {
ms = hi
}
s.metrics.SetInterval(time.Duration(ms) * time.Millisecond)
_ = s.store.Update(func(st *store.State) { st.SamplerMs = ms })
if !fanOut || s.nodes == nil {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
body, _ := json.Marshal(map[string]int{"ms": ms})
hdr := http.Header{"X-Hamlab-From-Hub": []string{"1"}, "Content-Type": []string{"application/json"}}
for _, name := range s.nodes.Names() {
if code, _, err := s.nodes.Do(ctx, name, "PUT", "/api/prefs/sampler", body, hdr); err != nil || code >= 400 {
log.Printf("sampler: %s did not take %d ms (%d %v)", name, ms, code, err)
}
}
}
// keepFleetInStep makes the two fleet-wide settings — the sampling rate and the history window —
// true of every machine, and keeps them true. Telling the machines once, when a setting
// changes, is not enough: a node that was off, restarting or mid-update at that moment never
// hears it (the first deploy of this very feature told three nodes that did not yet know the
// endpoint). So the hub asks each machine what it has, now and every two minutes, and corrects
// only what differs.
func (s *Server) keepFleetInStep() {
for first := true; ; first = false {
if first {
time.Sleep(15 * time.Second) // let a fleet that restarted together come up
} else {
time.Sleep(2 * time.Minute)
}
// What the hub itself is running is the fleet's answer — not what happens to be written
// in the UI prefs, which are only written when somebody changes a setting. A hub nobody
// has been into Settings on still has a sampling rate and a window, and its machines
// should be on them.
want := map[string][2]any{ // path → {field, value}
"/api/prefs/sampler": {"ms", int(s.metrics.Interval().Milliseconds())},
"/api/prefs/history": {"seconds", int(s.metrics.Window().Seconds())},
}
hdr := http.Header{"X-Hamlab-From-Hub": []string{"1"}, "Content-Type": []string{"application/json"}}
for _, name := range s.nodes.Names() {
for path, fv := range want {
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
code, b, err := s.nodes.Do(ctx, name, "GET", path, nil, hdr)
var have map[string]any
if err == nil && code == 200 && json.Unmarshal(b, &have) == nil {
if got, _ := have[fv[0].(string)].(float64); int(got) != fv[1].(int) {
body, _ := json.Marshal(map[string]any{fv[0].(string): fv[1]})
if c, _, e := s.nodes.Do(ctx, name, "PUT", path, body, hdr); e == nil && c < 400 {
log.Printf("fleet: %s had %s=%v, now %v", name, fv[0], got, fv[1])
}
}
}
cancel()
}
}
}
}
func (s *Server) getSampler(w http.ResponseWriter, _ *http.Request) {
jsonOK(w, 200, map[string]any{"ms": s.metrics.Interval().Milliseconds(), "min_ms": metrics.MinInterval.Milliseconds(), "max_ms": metrics.MaxInterval.Milliseconds()})
}
func (s *Server) putSampler(w http.ResponseWriter, r *http.Request) {
var body struct {
Ms int `json:"ms"`
}
if json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body) != nil || body.Ms < 100 || body.Ms > 60000 {
jsonErr(w, 400, "expected {\"ms\": 100…60000}")
return
}
s.setSampler(body.Ms, r.Header.Get("X-Hamlab-From-Hub") == "")
s.getSampler(w, r)
}
// ---- how far back the graphs reach ----
//
// One window for the whole fleet: the hub keeps it and pushes it to every machine, so a
// graph of `grid` opened from here reaches as far back as one of the hub. It is the
// machine's own setting either way — a node asked directly answers with what it was told.
func (s *Server) getHistoryWindow(w http.ResponseWriter, _ *http.Request) {
jsonOK(w, 200, map[string]any{
"seconds": int(s.metrics.Window().Seconds()),
"points": metrics.History,
"stride_sec": s.metrics.Stride().Seconds(),
"min_seconds": int(metrics.MinWindow.Seconds()),
"max_seconds": int(metrics.MaxWindow.Seconds()),
})
}
func (s *Server) putHistoryWindow(w http.ResponseWriter, r *http.Request) {
var body struct {
Seconds int `json:"seconds"`
}
if json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body) != nil {
jsonErr(w, 400, "expected {\"seconds\": <number>}")
return
}
lo, hi := int(metrics.MinWindow.Seconds()), int(metrics.MaxWindow.Seconds())
if body.Seconds < lo || body.Seconds > hi {
jsonErr(w, 400, fmt.Sprintf("a window is between %d and %d seconds", lo, hi))
return
}
s.metrics.SetWindow(time.Duration(body.Seconds) * time.Second)
if err := s.store.Update(func(st *store.State) { st.HistorySec = body.Seconds }); err != nil {
jsonErr(w, 500, err.Error())
return
}
if s.nodes != nil && len(s.nodes.Names()) > 0 && r.Header.Get("X-Hamlab-From-Hub") == "" {
go s.pushHistoryWindow(body.Seconds)
}
s.getHistoryWindow(w, r)
}
// pushHistoryWindow tells every managed machine the window. Best effort: a machine that is
// off or older simply keeps its own, and says so on its own pages.
func (s *Server) pushHistoryWindow(seconds int) {
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
defer cancel()
body, _ := json.Marshal(map[string]int{"seconds": seconds})
hdr := http.Header{"X-Hamlab-From-Hub": []string{"1"}, "Content-Type": []string{"application/json"}}
for _, name := range s.nodes.Names() {
code, _, err := s.nodes.Do(ctx, name, "PUT", "/api/prefs/history", body, hdr)
if err != nil || code >= 400 {
log.Printf("history window: %s did not take %ds (%d %v)", name, seconds, code, err)
}
}
}
// ---- terminals ----
func (s *Server) listTerminals(w http.ResponseWriter, _ *http.Request) {
jsonOK(w, 200, s.mgr.List())
}
func (s *Server) createTerminal(w http.ResponseWriter, r *http.Request) {
var body struct {
Name string `json:"name"`
}
_ = json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body)
name := strings.TrimSpace(body.Name)
if len(name) > maxNameLen {
jsonErr(w, 400, fmt.Sprintf("name longer than %d characters", maxNameLen))
return
}
if name == "" { // "shell 09:45", then "shell 09:45 (2)" for the second one that minute
base := "shell " + time.Now().Format("15:04")
taken := map[string]bool{}
for _, t := range s.mgr.List() {
taken[t.Name] = true
}
name = base
for n := 2; taken[name]; n++ {
name = fmt.Sprintf("%s (%d)", base, n)
}
}
sess, err := s.mgr.Create("", name)
if err != nil {
log.Printf("create terminal: %v", err)
jsonErr(w, 500, err.Error())
return
}
jsonOK(w, 201, map[string]string{"id": sess.ID, "name": sess.Name})
}
func (s *Server) renameTerminal(w http.ResponseWriter, r *http.Request) {
var body struct {
Name string `json:"name"`
}
if err := json.NewDecoder(io.LimitReader(r.Body, 4096)).Decode(&body); err != nil || strings.TrimSpace(body.Name) == "" {
jsonErr(w, 400, "name required")
return
}
if len(strings.TrimSpace(body.Name)) > maxNameLen {
jsonErr(w, 400, fmt.Sprintf("name longer than %d characters", maxNameLen))
return
}
if err := s.mgr.Rename(r.PathValue("id"), strings.TrimSpace(body.Name)); err != nil {
jsonErr(w, 404, err.Error())
return
}
w.WriteHeader(204)
}
func (s *Server) deleteTerminal(w http.ResponseWriter, r *http.Request) {
if err := s.mgr.Close(r.PathValue("id")); err != nil {
jsonErr(w, 404, err.Error())
return
}
w.WriteHeader(204)
}
type resizeMsg struct {
Type string `json:"type"`
Cols uint16 `json:"cols"`
Rows uint16 `json:"rows"`
}
func (s *Server) attach(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
client, snapshot, running, err := s.mgr.Attach(id)
if err != nil {
jsonErr(w, 404, err.Error())
return
}
// Origin must match Host (coder/websocket default) — blocks cross-site attaches.
conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{CompressionMode: websocket.CompressionDisabled})
if err != nil {
s.mgr.Detach(client)
return
}
defer s.mgr.Detach(client)
defer conn.Close(websocket.StatusNormalClosure, "")
conn.SetReadLimit(1 << 20)
ctx, cancel := context.WithCancel(r.Context())
defer cancel()
if len(snapshot) > 0 {
if err := conn.Write(ctx, websocket.MessageBinary, snapshot); err != nil {
return
}
}
if !running {
_ = conn.Write(ctx, websocket.MessageText, []byte(`{"type":"exit"}`))
}
// writer: PTY → socket, plus keepalive pings so idle proxies don't cut us.
go func() {
defer cancel()
ping := time.NewTicker(25 * time.Second)
defer ping.Stop()
for {
select {
case <-ctx.Done():
return
case <-ping.C:
pctx, pc := context.WithTimeout(ctx, 10*time.Second)
err := conn.Ping(pctx)
pc()
if err != nil {
return
}
case msg, ok := <-client.Ch:
if !ok {
_ = conn.Write(ctx, websocket.MessageText, []byte(`{"type":"closed"}`))
return
}
if msg.Data != nil {
if err := conn.Write(ctx, websocket.MessageBinary, msg.Data); err != nil {
return
}
continue
}
if msg.Event == "exit" {
b, _ := json.Marshal(map[string]any{"type": "exit", "code": msg.Code})
_ = conn.Write(ctx, websocket.MessageText, b)
}
}
}
}()
// reader: socket → PTY (binary) or control (text JSON).
for {
typ, data, err := conn.Read(ctx)
if err != nil {
return
}
switch typ {
case websocket.MessageBinary:
if err := s.mgr.Write(id, data); err != nil && !errors.Is(err, term.ErrNotFound) {
log.Printf("pty write %s: %v", id, err)
}
case websocket.MessageText:
var m resizeMsg
if json.Unmarshal(data, &m) == nil && m.Type == "resize" {
_ = s.mgr.Resize(id, m.Cols, m.Rows)
}
}
}
}
// ---- static UI with SPA fallback ----
func (s *Server) static(w http.ResponseWriter, r *http.Request) {
p := strings.TrimPrefix(r.URL.Path, "/")
if p == "" {
p = "index.html"
}
f, err := s.ui.Open(p)
if err == nil {
st, _ := f.Stat()
f.Close()
if st != nil && !st.IsDir() {
if strings.HasPrefix(p, "assets/") {
w.Header().Set("Cache-Control", "public, max-age=31536000, immutable")
} else {
w.Header().Set("Cache-Control", "no-cache")
}
http.ServeFileFS(w, r, s.ui, p)
return
}
}
// A path with no file of ours behind it may belong to a KVM's page, which asks for its
// own assets by absolute path. Checked here rather than in the router because this
// handler is the catch-all — and gated on a session, since this handler is public.
if s.signedIn(r) && s.kvmFallback(w, r) {
return
}
// SPA fallback
w.Header().Set("Cache-Control", "no-cache")
http.ServeFileFS(w, r, s.ui, "index.html")
}