Files
hamlab/internal/api/server.go
T
Mitchell-JohnsonandClaude Fable 5.1 a9bcb88212 Files: play video and audio in the viewer, as the file is; reads answer Range requests
The viewer hands a video or audio file to the browser untouched — container, codec and
all — and the browser plays what it can; when it can't, the viewer says so and offers the
download. What the server adds is the part a player needs: /api/files/read now goes
through http.ServeContent for a real file of known size, so Range requests (seeking) and
HEAD work, and a download can resume. The root backend's helper takes an offset for that,
and a Seeker opens the backend once, at the range asked for, instead of streaming from
the top. Audio files are a preview kind too: their cover art is the thumbnail.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-11 16:40:42 -07:00

2668 lines
87 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
self string // this binary, for helpers run as tasks
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.self = self
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/move", s.auth(s.filesMove))
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
}
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 {
// A real file of known size goes through net/http, which speaks Range: a video player
// can seek and a download can resume. /proc files say size 0 and take the plain path.
rs := files.NewSeeker(r.Context(), fsys, p, in.Size)
defer rs.Close()
if _, local := fsys.(files.Local); local { // cheap here, and turns "can't open" into a status rather than a cut-off body
if err := rs.Check(); err != nil {
fsErr(w, err)
return
}
}
http.ServeContent(w, r, name, time.UnixMilli(in.Mtime), rs)
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)
}
_, _ = 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)
})
}
// filesMove: POST {path, to} — to is the full destination path. A rename when the two
// ends share a filesystem (204); otherwise the copy-and-remove runs as a task (202 with the
// task), since a folder between datasets can take as long as it takes.
func (s *Server) filesMove(w http.ResponseWriter, r *http.Request) {
fsys, ok := s.fsFor(w, r)
if !ok {
return
}
var body struct{ Path, To string }
if err := json.NewDecoder(io.LimitReader(r.Body, 64<<10)).Decode(&body); err != nil {
jsonErr(w, 400, "bad request")
return
}
from, err := files.Clean(body.Path)
if err != nil {
jsonErr(w, 400, err.Error())
return
}
to, err := files.Clean(body.To)
if err != nil {
jsonErr(w, 400, err.Error())
return
}
if from == to {
jsonErr(w, 400, "it is already there")
return
}
if files.Inside(from, to) {
jsonErr(w, 400, "a folder can't be moved into itself")
return
}
// The kernel answers EXDEV for two paths on different filesystems before it looks for
// the source, so a missing file would otherwise become a task that fails on arrival.
if _, err := fsys.Stat(r.Context(), from); err != nil {
fsErr(w, err)
return
}
err = fsys.Rename(r.Context(), from, to)
if err == nil {
log.Printf("files: move %s -> %s (from %s)", from, to, clientIP(r))
w.WriteHeader(204)
return
}
if !files.IsCrossDevice(err) {
fsErr(w, err)
return
}
cmd := files.MoveCmd(context.Background(), fsys, s.self, from, to)
t, err := s.tasks.Start("files:move", "move "+filepath.Base(from)+" to "+filepath.Dir(to), cmd)
if err != nil {
jsonErr(w, 500, err.Error())
return
}
log.Printf("files: move %s -> %s across filesystems as task %s (from %s)", from, to, t.ID, clientIP(r))
jsonOK(w, 202, t.Snapshot(false))
}
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")
}