internal/server/api/api.go
Ref: Size: 47.4 KiB History
// Package api implements the admin REST API and the unauthenticated enrollment endpoint.
package api
import (
"context"
"crypto/sha256"
"database/sql"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log/slog"
"net/http"
"slices"
"strings"
"time"
"github.com/a73x/eitri/internal/cloudinit"
"github.com/a73x/eitri/internal/joinblob"
"github.com/a73x/eitri/internal/names"
"github.com/a73x/eitri/internal/random"
"github.com/a73x/eitri/internal/server/api/types"
"github.com/a73x/eitri/internal/server/delegation"
"github.com/a73x/eitri/internal/server/hosttoken"
"github.com/a73x/eitri/internal/server/hub"
"github.com/a73x/eitri/internal/server/registry"
"github.com/a73x/eitri/internal/server/release"
"github.com/a73x/eitri/internal/server/store"
"github.com/a73x/eitri/internal/version"
)
// volumesFeature is the agent floor a volume-bearing create is judged against.
// It is a var, not the constant its siblings use, because the floor is a
// RELEASE TAG: until volumes ship there is no agent above it, so every test
// that exercises the admission has to name a floor its fixtures can clear.
// Nothing but a test writes it.
var volumesFeature = release.Volumes
// DefaultImage is the image applied to one-click VM creates.
type DefaultImage struct {
URL string
SHA256 string
}
// Config holds static configuration for the API server.
type Config struct {
HostSecret []byte
// DefaultImages is keyed by host ARCHITECTURE ("amd64", "arm64"). A
// one-click create takes the entry matching the host it is placed on; see
// applyVMDefaults.
DefaultImages map[string]DefaultImage
ServerCertSHA256 string
AdvertiseHTTP string // HTTP base URL agents use to reach this server (scheme+host+port)
AdvertiseQUIC string // QUIC host:port agents use to reach this server
OIDC OIDCConfig // console sign-in relying-party settings (see auth.go)
}
// ReleaseSource exposes the latest known release. *release.Client satisfies
// it; the API only ever reads the cached manifest — Refresh is the poller's
// job (main.go), not a request-path concern — so this interface stays a
// single method.
type ReleaseSource interface {
Latest() (release.Manifest, bool)
}
// AgentUpgrader records a pending per-host agent self-upgrade — and answers for
// the one it is holding, which is how the offer becomes visible to an operator
// instead of vanishing into a snapshot. *syncsvc.Service satisfies it.
type AgentUpgrader interface {
OfferAgentUpgrade(hostID, version, url, sha256 string)
// ClearAgentUpgrade drops any pending offer for hostID — called when the
// host leaves the fleet so a decommission cannot strand a stale offer.
ClearAgentUpgrade(hostID string)
// PendingAgentUpgrade reports the version offered to hostID and how long
// the offer has stood; ok is false when none is outstanding.
PendingAgentUpgrade(hostID string) (version string, age time.Duration, ok bool)
}
// API is the HTTP handler container.
type API struct {
cfg Config
st *store.Store
reg *registry.Registry
hub *hub.Hub
notif *notifier
enrolls *ipLimiter // per-client-bucket brake on the unauthenticated enroll endpoint (v4: address, v6: /64)
tickets *ticketStore // one-time SSE stream tickets
snap *snapshotHub // central SSE snapshot: one marshal fanned to all clients
console ConsoleDialer // nil until main wires syncsvc (SetConsoleDialer)
sshCAKey string // eitri CA public key (authorized_keys form) served by GET /api/v1/ssh-ca; empty ⇒ gate off
sshGate string // gate hop address (host:port) served by GET /api/v1/me; empty ⇒ gate off
release ReleaseSource // nil ⇒ release discovery disabled
upgrader AgentUpgrader // nil until main wires syncsvc (SetAgentUpgrader)
auth *authFlow // OIDC sign-in relying party (mounted via AuthHandler, outside /api/)
// delegations holds the credentials tenants have lent eitri. Nil until main
// wires it, which happens only when the jump gate is on; nil ⇒ the
// delegation routes answer 503.
delegations *delegation.Keyring
// volumeStuck remembers, per decommissioning host, the volume count the
// last "stuck on volumes" line reported. The sweep runs every two seconds
// and a volume never drains itself, so logging the refusal every pass would
// bury the log; logging only when the number CHANGES says it once when the
// decommission stalls and once more each time the operator deletes a claim.
// Touched only from StartBackground's goroutine.
volumeStuck map[string]int
}
// URL renders an API path as a full URL a caller can actually dial, using the
// plane's configured public_url. Endpoints named in errors and recipes go
// through here: the REST API and /mcp can live on different hostnames, so a
// bare path sends a caller to guess which one — a guess that costs real time
// when the answer is "not the host you are talking to".
//
// An unconfigured public_url falls back to the bare path, which is still true,
// just less useful.
func (a *API) URL(path string) string {
base := strings.TrimRight(a.cfg.OIDC.PublicURL, "/")
if base == "" {
return path
}
return base + path
}
// SetReleaseSource wires release discovery (nil leaves it disabled).
func (a *API) SetReleaseSource(rs ReleaseSource) { a.release = rs }
// SetAgentUpgrader wires the per-host upgrade offer sink.
func (a *API) SetAgentUpgrader(u AgentUpgrader) { a.upgrader = u }
// latestVersion returns the latest known release version ("" when discovery
// is disabled or the manifest hasn't been fetched).
func (a *API) latestVersion() string {
if a.release == nil {
return ""
}
if m, ok := a.release.Latest(); ok {
return m.Version
}
return ""
}
// New constructs an API. It starts the central SSE snapshot hub (one goroutine
// that marshals the fleet snapshot on a 1s tick / desired-state wake and fans
// the identical bytes to all connected clients). Call Close to stop it.
func New(cfg Config, st *store.Store, reg *registry.Registry, h *hub.Hub) *API {
a := &API{cfg: cfg, st: st, reg: reg, hub: h, notif: newNotifier(),
enrolls: newIPLimiter(time.Now), tickets: newTicketStore(time.Now)}
a.auth = &authFlow{st: st, cfg: cfg.OIDC}
a.snap = newSnapshotHub(a.marshalSnapshots, a.notif)
go a.snap.run()
return a
}
// Close stops the central snapshot hub goroutine. Idempotent. Tests and any
// future graceful-shutdown path should call it; the server binary itself blocks
// in ListenAndServe for the whole process lifetime and exits via os.Exit (which
// skips deferred cleanup), so the single hub goroutine simply lives until exit.
func (a *API) Close() { a.snap.Close() }
// Handler returns the ServeMux with all routes registered from the declared
// route table (routes.go) — the table is the single enumerable surface, shared
// with the OpenAPI generator.
func (a *API) Handler() http.Handler {
mux := http.NewServeMux()
admin := http.NewServeMux()
for _, rt := range routeTable {
h := rt.handler
hf := func(w http.ResponseWriter, r *http.Request) { h(a, w, r) }
pattern := rt.Method + " " + rt.Path
if rt.Auth == AuthUser {
admin.HandleFunc(pattern, hf)
} else {
mux.HandleFunc(pattern, hf)
}
}
mux.Handle("/api/v1/", a.userAuth(admin))
return mux
}
// AuthHandler returns the browser sign-in endpoints (/auth/login, /auth/callback,
// /auth/logout). main.go mounts it on the ROOT mux, OUTSIDE /api/ and its auth
// middleware — these establish the session the middleware later checks, so they
// cannot themselves require one. They are deliberately absent from the JSON
// route table (spec §3): they speak redirects and cookies, not the contract.
func (a *API) AuthHandler() http.Handler {
mux := http.NewServeMux()
mux.HandleFunc("GET /auth/login", a.auth.handleLogin)
mux.HandleFunc("GET /auth/callback", a.auth.handleCallback)
mux.HandleFunc("POST /auth/logout", a.auth.handleLogout)
return mux
}
// UserAuth wraps h in the same PAT/session resolution the /api/v1 subtree uses,
// for handlers mounted outside it. /mcp speaks JSON-RPC rather than the REST
// contract, so it lives outside the route table (like /auth/*) — but living
// outside the contract must not mean living outside its authentication.
func (a *API) UserAuth(h http.Handler) http.Handler { return a.userAuth(h) }
// StartBackground launches the periodic server-side sweeps: finalizing drained
// decommissioning hosts, and reaping tombstoned VMs whose host's agent never
// acked the destroy (abandoned on an offline host). It returns when ctx is
// cancelled.
func (a *API) StartBackground(ctx context.Context) {
t := time.NewTicker(2 * time.Second)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
changed := a.sweepDecommissioned()
if a.sweepAbandonedVMs(time.Now()) {
changed = true
}
if changed {
a.notif.notify()
}
}
}
}
// abandonedVMReapGrace is how long a tombstoned VM may sit unacked on an offline
// host before the server hard-deletes it itself. The normal delete path is the
// agent's quarantine→destroy→ack chain; this is the backstop for when no agent
// is there to run it (the host is gone, or crashed the way a failed-VM host
// tends to). The bound is deliberately generous: it exceeds the agent's own
// TombstoneGrace so a live agent always reaps first, and it is far longer than
// any routine agent restart or redeploy so a transient outage never trips a
// server-side force-delete. Correctness survives a host that returns AFTER a
// reap: the VM is then absent from desired state, and the agent's vanish-reap
// tears down any lingering guest. No config knob — a fixed policy bound.
const abandonedVMReapGrace = 15 * time.Minute
// An online host is left alone: its live agent owns the reap and will ack
// through the normal path, and racing it risks flipping the VM onto the longer
// vanished-grace clock. Returns true if it removed at least one row. now is
// threaded in so the age check and the loop share one clock.
func (a *API) sweepAbandonedVMs(now time.Time) bool {
vms, err := a.st.ListVMs()
if err != nil {
return false
}
reaped := false
for _, vm := range vms {
if vm.DeletedAt == nil {
continue // live VM — not a delete in progress
}
if now.Sub(*vm.DeletedAt) < abandonedVMReapGrace {
continue
}
if st, ok := a.reg.Get(vm.HostID); ok && st.Online {
continue
}
if err := a.st.ForceDeleteVM(vm.ID); err != nil {
slog.Warn("sweep abandoned VM failed", "vm", vm.ID, "host", vm.HostID, "err", err)
continue
}
reaped = true
a.audit(vm.Tenant, "vm.reap", map[string]string{
"vm_id": vm.ID, "host_id": vm.HostID,
"reason": "tombstoned VM reaped server-side: host offline past grace",
})
}
return reaped
}
// sweepDecommissioned finalizes any decommissioning host with no VM rows left
// (fully drained). Returns true if it removed at least one host.
//
// One refusal is not like the others. An undrained VM is a decommission still
// in progress and silence is correct — the next pass, two seconds later, is the
// retry. A held volume is a decommission that will NEVER finish on its own: a
// volume outlives its guests by design, so no amount of waiting releases it and
// the host sits in `decommissioning` until an operator deletes the claims. That
// one gets said out loud, and the map keeps it from being said 30 times a
// minute forever.
func (a *API) sweepDecommissioned() bool {
hosts, err := a.st.ListHosts()
if err != nil {
return false
}
removed := false
// Rebuilt every sweep, so a host that leaves (removed, or forced away)
// takes its entry with it instead of leaking one per decommission.
stuck := map[string]int{}
for _, h := range hosts {
if h.Status != "decommissioning" {
continue
}
// RemoveHost re-counts VM rows in-transaction and refuses while any
// remain, so a pre-check here would only save a wasted call on hosts
// that aren't yet drained — an undrained VM is handled by simply
// skipping to the next host.
err := a.st.RemoveHost(h.ID)
switch {
case err == nil:
removed = true
case errors.Is(err, store.ErrHostHoldsVolumes):
n, ids := a.volumesOnHost(h.ID)
prev, seen := a.volumeStuck[h.ID]
stuck[h.ID] = n
if !seen || prev != n {
slog.Warn("decommission is stuck: host still holds volumes",
"host", h.ID, "volumes", n, "volume_ids", ids,
"remedy", "delete the volume claims placed on this host")
}
}
}
a.volumeStuck = stuck
return removed
}
// volumesOnHost reports how many volume rows a host still holds and names them,
// for the stuck-decommission line. A read failure reports nothing rather than
// inventing a number: the count in the log would be the only lie in it.
func (a *API) volumesOnHost(hostID string) (int, string) {
vols, err := a.st.ListVolumesForHost(hostID)
if err != nil {
return 0, ""
}
ids := make([]string, 0, len(vols))
for _, v := range vols {
ids = append(ids, v.ID)
}
return len(vols), strings.Join(ids, ",")
}
// userAuth resolves a request to a tenant principal: PAT bearer first, then
// the eitri_session console cookie. There are no other credentials and no
// per-route exceptions (spec §3); a request with neither gets 401 and the SPA
// redirects to /auth/login. Unknown/expired/revoked PATs and sessions are all
// indistinguishable 401s.
func (a *API) userAuth(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
// PAT bearer first. The eitri_pat_ prefix disambiguates a PAT from any
// other Bearer value; a prefixed-but-invalid token is a hard 401 rather
// than falling through to the cookie path.
if tok, ok := strings.CutPrefix(r.Header.Get("Authorization"), "Bearer "); ok && strings.HasPrefix(tok, "eitri_pat_") {
if tenant, ok, err := a.st.TenantForAPIToken(tok); err == nil && ok {
next.ServeHTTP(w, r.WithContext(withPrincipal(r.Context(), Principal{Tenant: tenant})))
return
}
http.Error(w, "invalid token", http.StatusUnauthorized)
return
}
// Console session cookie next.
if c, err := r.Cookie("eitri_session"); err == nil {
if tenant, ok, err := a.st.SessionTenant(c.Value); err == nil && ok {
next.ServeHTTP(w, r.WithContext(withPrincipal(r.Context(), Principal{Tenant: tenant})))
return
}
}
http.Error(w, "sign in required", http.StatusUnauthorized)
})
}
// writeJSON encodes v as JSON with the correct Content-Type header and status.
func writeJSON(w http.ResponseWriter, status int, v any) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(status)
json.NewEncoder(w).Encode(v) //nolint:errcheck
}
// decodeJSON decodes the request body into v, reporting a 400 with the standard
// "bad request" body on failure. Returns false when it has already written the
// response (caller must return).
func decodeJSON(w http.ResponseWriter, r *http.Request, v any) bool {
if err := json.NewDecoder(r.Body).Decode(v); err != nil {
http.Error(w, "bad request", http.StatusBadRequest)
return false
}
return true
}
// --- enrollment ---
// audit appends an audit row, mirrored to the live log. Failures are logged,
// never fatal — the audit trail must not break the operation it records.
// BEST-EFFORT: rows written here can be lost on a crash between the audited
// operation and this insert. The one row that must be durable — host.enroll —
// is written inside the redeem transaction by the store, not here.
func (a *API) audit(tenant, action string, detail map[string]string) {
raw, _ := json.Marshal(detail)
if err := a.st.AppendAudit(tenant, action, string(raw)); err != nil {
slog.Warn("audit append failed", "action", action, "err", err)
}
}
// truncate bounds attacker-controlled strings before they reach the audit
// log (the name in a denied enroll is unauthenticated input).
func truncate(s string, n int) string {
if len(s) > n {
return s[:n]
}
return s
}
// tokenHashPrefix returns the first 8 hex chars of sha256(tok) — enough to
// correlate mint/redeem audit rows without ever storing the secret.
func tokenHashPrefix(tok string) string {
sum := sha256.Sum256([]byte(tok))
return hex.EncodeToString(sum[:])[:8]
}
func (a *API) handleEnroll(w http.ResponseWriter, r *http.Request) {
if !a.enrolls.allow(bucketKey(clientIP(r))) {
http.Error(w, "rate limited", http.StatusTooManyRequests)
return
}
// Unauthenticated endpoint: bound the body so junk can't bloat memory or
// the denied-audit trail. Legitimate enroll bodies are well under 1 KiB.
r.Body = http.MaxBytesReader(w, r.Body, 16<<10)
var req types.EnrollRequest
if !decodeJSON(w, r, &req) {
return
}
host, err := a.st.RedeemEnrollmentToken(req.Token, store.EnrollFacts{
Name: req.Name, OS: req.OS, Arch: req.Arch, Provisioner: req.Provisioner,
Remote: clientIP(r), BridgeCIDR: req.BridgeCIDR,
})
if err != nil {
// Unauthenticated enroll attempt: no tenant is resolvable (the token was
// rejected), so the denied row is filed under the system audit scope.
a.audit(store.SystemTenant, "host.enroll.denied", map[string]string{
"remote": clientIP(r), "name": truncate(req.Name, 64),
"token_hash_prefix": tokenHashPrefix(req.Token)})
http.Error(w, "forbidden", http.StatusForbidden)
return
}
// host.enroll is audited durably inside the redeem transaction.
cred := hosttoken.Mint(a.cfg.HostSecret, host.ID, host.CredGeneration, time.Now())
writeJSON(w, http.StatusCreated, types.EnrollResponse{
BridgeCIDR: host.BridgeCIDR,
Credential: cred,
HostID: host.ID,
ServerCertSHA256: a.cfg.ServerCertSHA256,
})
}
func (a *API) handleCreateEnrollToken(w http.ResponseWriter, r *http.Request) {
tok, err := a.st.CreateEnrollmentToken(principalFromContext(r).Tenant)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
a.audit(principalFromContext(r).Tenant, "enroll-token.mint", map[string]string{
"remote": clientIP(r), "token_hash_prefix": tokenHashPrefix(tok)})
join, err := joinblob.Encode(a.cfg.AdvertiseHTTP, a.cfg.AdvertiseQUIC, tok, a.cfg.ServerCertSHA256)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
writeJSON(w, http.StatusCreated, types.EnrollTokenResponse{Join: join, Token: tok})
}
// --- hosts ---
func toHostResponse(h store.Host, st registry.HostState, ok bool, alloc store.Alloc) types.Host {
hr := types.Host{
ID: h.ID,
Name: h.Name,
OS: h.OS,
Arch: h.Arch,
Provisioner: h.Provisioner,
BridgeCIDR: h.BridgeCIDR,
Status: h.Status,
EnrolledAt: h.EnrolledAt,
Allocated: types.Capacity{VCPUs: alloc.VCPUs, MemMB: alloc.MemMB, DiskGB: alloc.DiskGB},
OSID: h.OSID,
OSPretty: h.OSPretty,
OSVersion: h.OSVersion,
Kernel: h.Kernel,
CPUModel: h.CPUModel,
Virt: h.Virt,
UplinkAddr: h.UplinkAddr,
// A host that has advertised nothing serves an empty list, never JSON
// null: the field is declared non-nullable, and "no networks" is the
// same answer whether the host said so or has not spoken at all.
HostNetworks: []string{},
}
if ok {
hr.Online = st.Online
hr.Stale = st.Stale
hr.Sessions = st.Sessions
hr.AgentVersion = st.AgentVersion
if st.HostNetworks != nil {
hr.HostNetworks = st.HostNetworks
}
if !st.LastSeen.IsZero() {
seen := st.LastSeen
secs := int64(st.SinceLastSeen.Seconds())
hr.LastSeen = &seen
hr.SecondsSinceLastSeen = &secs
}
hr.Capacity = types.Capacity{
VCPUs: st.Capacity.VCPUs,
MemMB: st.Capacity.MemMB,
DiskGB: st.Capacity.DiskGB,
}
// Metrics are live/measured, so only emit them while the host is
// actually Online — a host with stale registry state (reported, then
// went silent) would otherwise serve stale utilization. Capacity above
// intentionally stays on the `ok` gate: totals are ~static, so a last
// known value is still useful; per-second metrics are not.
if st.Online {
hr.Metrics = &types.Metrics{
UptimeS: st.Metrics.UptimeS,
MemUsedMB: st.Metrics.MemUsedMB,
MemAvailableMB: st.Metrics.MemAvailableMB,
Load1: st.Metrics.Load1,
Load5: st.Metrics.Load5,
Load15: st.Metrics.Load15,
DiskUsedGB: st.Metrics.DiskUsedGB,
DiskFreeGB: st.Metrics.DiskFreeGB,
}
}
}
return hr
}
// regState is one host's live registry state, fetched ONCE per snapshot. It
// exists so buildVMResponses can index a host's state instead of calling
// reg.Get(vm.HostID) per VM — registry.Get deep-clones the whole host report on
// every call, so N VMs on a host used to re-clone that report N times.
type regState struct {
st registry.HostState
ok bool
}
// fetchStates fetches each distinct host's registry state exactly once. The
// returned map covers every id passed (a missing registry entry is stored with
// ok=false), so callers can index it without falling back to reg.Get.
func (a *API) fetchStates(ids ...[]string) map[string]regState {
states := map[string]regState{}
for _, group := range ids {
for _, id := range group {
if _, seen := states[id]; seen {
continue
}
st, ok := a.reg.Get(id)
states[id] = regState{st: st, ok: ok}
}
}
return states
}
// snapshotHosts builds the wire host list (durable host rows merged with live
// registry state and server-computed allocation) for GET /hosts.
func (a *API) snapshotHosts(p Principal) ([]types.Host, error) {
// Single-tx read: hosts and alloc must not mix two epochs (same property
// the SSE stream needs; the unused vms read is cheap on these small
// control-plane tables).
hosts, alloc, _, err := a.st.Snapshot()
if err != nil {
return nil, err
}
hosts = filterHosts(p, hosts)
return a.buildHostResponses(hosts, alloc, a.fetchStates(hostIDs(hosts))), nil
}
func hostIDs(hosts []store.Host) []string {
ids := make([]string, len(hosts))
for i, h := range hosts {
ids[i] = h.ID
}
return ids
}
func vmHostIDs(vms []store.VM) []string {
ids := make([]string, len(vms))
for i, vm := range vms {
ids[i] = vm.HostID
}
return ids
}
// buildHostResponses merges durable host rows with live registry state, indexing
// the pre-fetched states map (one reg.Get per host).
func (a *API) buildHostResponses(hosts []store.Host, alloc map[string]store.Alloc, states map[string]regState) []types.Host {
latest := a.latestVersion()
out := make([]types.Host, len(hosts))
for i, h := range hosts {
rs := states[h.ID]
out[i] = toHostResponse(h, rs.st, rs.ok, alloc[h.ID])
out[i].AgentUpdateAvailable = latest != "" && out[i].Online &&
out[i].AgentVersion != "" && version.Less(out[i].AgentVersion, latest)
// An offer the agent has not taken yet. The upgrader is the only place
// it exists — offers are held in memory beside the snapshot stream that
// carries them, never written down — so a restarted server reports none
// and the console offers the button again, which is exactly right: the
// restart dropped the offer too.
if a.upgrader != nil {
if v, age, ok := a.upgrader.PendingAgentUpgrade(h.ID); ok {
out[i].PendingUpgrade = &types.PendingUpgrade{Version: v, AgeS: int64(age.Seconds())}
}
}
}
return out
}
func (a *API) handleListHosts(w http.ResponseWriter, r *http.Request) {
out, err := a.snapshotHosts(principalFromContext(r))
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
writeJSON(w, http.StatusOK, out)
}
// --- VMs ---
func toVMResponse(vm store.VM, actualPower, phase, statusDetail string, destroyAt int64, hostOnline bool) types.VM {
return types.VM{
ID: vm.ID,
HostID: vm.HostID,
Name: vm.Name,
ImageURL: vm.ImageURL,
VCPUs: vm.VCPUs,
MemMB: vm.MemMB,
DiskGB: vm.DiskGB,
PowerState: vm.PowerState,
Status: vm.Status,
LastError: vm.LastError,
AssignedIP: vm.AssignedIP,
Network: vm.Network,
NetworkIP: vm.NetworkIP,
CreatedAt: vm.CreatedAt,
Deleted: vm.DeletedAt != nil,
ActualPower: actualPower,
Phase: phase,
StatusDetail: statusDetail,
DestroyAt: destroyAt,
Lifecycle: deriveLifecycle(vm, actualPower, phase, hostOnline),
InjectedKey: injectedKey(vm),
TrustedCAs: trustedCAs(vm),
}
}
// deriveLifecycle folds the orthogonal state axes into one coarse lifecycle
// word. The server owns this derivation exclusively — every types.VM ships
// the result as vm.lifecycle, and the client (vmStatus() in
// web/src/lib/fleet.svelte.ts) displays it directly, keeping only a small
// defensive fallback for malformed or missing snapshots rather than
// re-deriving the value itself.
//
// hostOnline is why "unreachable" exists. Every other word here is an
// observation, and the durable row holds the LAST one its host made: nothing
// rewrites status or power_state when an agent goes away, so those columns
// keep saying "ready"/"running" indefinitely after the machine under them is
// switched off. Reading them on a dark host would answer a question the plane
// cannot answer — and it is the answer an operator acts on, so it has to be
// the true one. See TestAVMOnADarkHostReadsAsUnreachable.
//
// This is a projection, not a transition: the VM row is untouched, nothing is
// reconciled, and the first report from a returning host restores every word
// (TestADarkHostsVMRecoversItsLifecycleWhenTheHostComesBack). A server that has
// just restarted holds an empty registry, so the whole fleet reads unreachable
// until each agent's first report lands — the same brief, self-healing silence
// the create-path preconditions already tolerate, and honest while it lasts.
func deriveLifecycle(vm store.VM, actualPower, phase string, hostOnline bool) string {
// The tombstone outranks unreachability because it is the plane's OWN
// intent rather than something a host told it: the delete is recorded here,
// it stays true while the host is away, and the console owes the operator
// the restore affordance for exactly as long as it does.
if vm.DeletedAt != nil {
return "deleting"
}
if !hostOnline {
return "unreachable"
}
if phase == "" {
phase = vm.Status
}
switch phase {
case "failed":
return "failed"
case "creating", "pending", "":
return "creating"
}
// phase is ready from here — reconcile power.
power := actualPower
if power == "" {
power = vm.PowerState
}
if power != "running" {
return "stopped"
}
return "ready"
}
// snapshotVMs builds the wire VM list (durable VM rows merged with live
// actual-state from the registry). Used by GET /vms; the SSE stream uses
// buildVMResponses over a single-tx store.Snapshot instead.
func (a *API) snapshotVMs(p Principal) ([]types.VM, error) {
vms, err := a.st.ListVMs()
if err != nil {
return nil, err
}
vms = filterVMs(p, vms)
return a.buildVMResponses(vms, a.fetchStates(vmHostIDs(vms))), nil
}
// buildVMResponses merges durable VM rows with live registry actual-state,
// indexing the pre-fetched states map (one reg.Get per host, not per VM).
func (a *API) buildVMResponses(vms []store.VM, states map[string]regState) []types.VM {
out := make([]types.VM, len(vms))
for i, vm := range vms {
var actualPower, phase, statusDetail string
var destroyAt int64
// Online, not merely present: a host whose registry entry has gone
// stale is as silent as one that never connected, and its last report
// is as much a memory. Both must read the same way — see
// deriveLifecycle.
//
// The gate covers the whole merge, not just the lifecycle word, so the
// response cannot contradict itself: everything below is an
// OBSERVATION (what power the host saw, what phase it was in, what it
// said it was doing, when it started the destroy clock), and a host
// that is not reporting is not observing. Serving `unreachable` beside
// a phase and a power would hand a client the same stale answer the
// status just refused to give, one field further down.
rs := states[vm.HostID]
hostOnline := rs.ok && rs.st.Online
if hostOnline {
for _, av := range rs.st.Report.VMs {
if av.VMID == vm.ID {
actualPower = av.PowerState
phase = av.Phase
statusDetail = av.StatusDetail
break
}
}
for _, qv := range rs.st.Report.Quarantined {
if qv.VMID == vm.ID {
destroyAt = qv.DestroyAtUnix
break
}
}
}
out[i] = toVMResponse(vm, actualPower, phase, statusDetail, destroyAt, hostOnline)
}
return out
}
func (a *API) handleListVMs(w http.ResponseWriter, r *http.Request) {
out, err := a.snapshotVMs(principalFromContext(r))
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
writeJSON(w, http.StatusOK, out)
}
// applyVMDefaults fills one-click defaults in place. hostArch is the
// architecture of the host the VM is being placed on, which selects the default
// image: a host can only execute a guest built for its own CPU, and nothing
// downstream checks — an image the host cannot run boots into nothing and is
// reported as a lost VM, with no clue as to why.
//
// An explicitly supplied image is NOT arch-checked. A URL does not say what its
// contents can execute, and guessing from the filename would reject legitimate
// custom images to catch a mistake the operator made deliberately. The guard is
// on the default, which eitri chooses, not on the choice the caller made.
//
// It returns an error message and HTTP status (msg=="" when ok) for the
// image-pairing rule, which is a validation, not a default.
func (a *API) applyVMDefaults(req *types.CreateVMRequest, hostArch string) (string, int) {
if req.Name == "" {
req.Name = "sandbox-" + random.Hex(3)
}
if req.ImageURL == "" && req.ImageSHA256 == "" {
img, ok := a.cfg.DefaultImages[hostArch]
if !ok {
return fmt.Sprintf("no default image configured for %s hosts; pass image_url and image_sha256, "+
"or add default_images[%q] to the server config", hostArch, hostArch), http.StatusBadRequest
}
req.ImageURL = img.URL
req.ImageSHA256 = img.SHA256
} else if req.ImageURL == "" || req.ImageSHA256 == "" {
return "image_url and image_sha256 must be provided together", http.StatusBadRequest
}
if req.VCPUs == 0 {
req.VCPUs = types.DefaultVCPUs
}
if req.MemMB == 0 {
req.MemMB = types.DefaultMemMB
}
if req.DiskGB == 0 {
req.DiskGB = types.DefaultDiskGB
}
if req.PowerState == "" {
req.PowerState = "running"
}
return "", 0
}
// validateCreateVM checks the post-defaults request, returning (msg, status) on
// failure or ("", 0) when valid. Messages are byte-identical to the prior inline
// checks so response bodies do not change.
func validateCreateVM(req *types.CreateVMRequest) (string, int) {
// Name becomes the guest hostname and is embedded into cloud-init YAML, so it
// must be a valid RFC-1123 DNS label.
if !names.IsRFC1123Label(req.Name) {
return "invalid name", http.StatusBadRequest
}
// SSH key must be single-line: a newline would allow YAML injection into the
// cloud-init user-data that embeds the key verbatim.
if strings.ContainsAny(req.SSHAuthorizedKey, "\n\r") {
return "ssh_authorized_key must be single-line", http.StatusBadRequest
}
// Exactly 64 lowercase hex digits — catches a misconfigured server default
// (e.g. "pinned") at create time with a clear error instead of a later mismatch.
if !names.IsSHA256Hex(req.ImageSHA256) {
return "invalid image_sha256", http.StatusBadRequest
}
// Negative values are never meaningful; a too-small disk_gb is also rejected
// by the agent's never-shrink guard, but nonsense should fail at create time.
if req.VCPUs < 1 || req.MemMB < 1 || req.DiskGB < 1 {
return "vcpus, mem_mb and disk_gb must each be >= 1", http.StatusBadRequest
}
// A network is optional (empty is the NAT underlay), but a name that cannot
// be a network name is malformed here rather than a refusal later — the
// grammar is names.IsNetworkName, the same one the agent's --host-network
// parser refuses with, so a name the API accepts is one a host could serve.
if req.Network != "" && !names.IsNetworkName(req.Network) {
return "invalid network", http.StatusBadRequest
}
return "", 0
}
// createVMBody is what a create decodes into: the contract request, plus the
// one field the contract dropped. decodeJSON is lenient everywhere — no
// endpoint sets DisallowUnknownFields, so an unrecognized key is ignored — and
// under that discipline a body carrying "persistent": false would be accepted
// and then contradicted, the VM created with the opposite policy to the one it
// asked for. Naming the field here costs one 400 that says what changed,
// without making every other endpoint strict about keys it has always taken.
type createVMBody struct {
types.CreateVMRequest
Persistent *bool `json:"persistent"`
}
func (a *API) handleCreateVM(w http.ResponseWriter, r *http.Request) {
var body createVMBody
if !decodeJSON(w, r, &body) {
return
}
if body.Persistent != nil {
http.Error(w, "every VM is persistent now; drop the persistent field", http.StatusBadRequest)
return
}
req := body.CreateVMRequest
if req.HostID == "" {
http.Error(w, "host_id required", http.StatusBadRequest)
return
}
// Tenant gate: you may only place VMs on hosts in your tenant. The host
// read also feeds the namespaced host-cert principal below, and the default
// image, which follows the host's architecture. CreateVM re-checks host
// status in-tx; this pre-read is the AUTHZ point.
//
// It runs BEFORE defaults and validation: a caller who may not see this host
// gets "unknown host_id" whatever else is wrong with the body, rather than a
// validation error that confirms the request got as far as the host.
host, err := a.st.GetHost(req.HostID)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
http.Error(w, "unknown host_id", http.StatusBadRequest)
return
}
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
if !mayActAs(principalFromContext(r), host.Tenant) {
http.Error(w, "unknown host_id", http.StatusBadRequest)
return
}
if msg, code := a.applyVMDefaults(&req, host.Arch); msg != "" {
http.Error(w, msg, code)
return
}
if msg, code := validateCreateVM(&req); msg != "" {
http.Error(w, msg, code)
return
}
// BYO CA precondition: a guest bakes its tenant's user-CA set into its sshd
// trust at create and nothing updates it afterwards, so a tenant with no
// registered CA creates guests that nothing can ever reach. The request is
// well-formed — the tenant is not ready to make a guest it can reach — so
// this is a 409, like the certified-host-key refusal below that it is the
// tenant-side half of. The CA is read against the HOST's tenant, which the
// authz gate above has already proven is the caller's own.
//
// This READS the set rather than counting it, and the set it read is what
// goes onto the row below. That is what makes the sentence above true: the
// trust is decided here, at create, by the same query that permits the
// create — not later, by whatever the tenant happens to have registered by
// the time a host gets around to building the guest's seed.
tenantCAs, err := a.st.ListTenantUserCAs(host.Tenant)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
if len(tenantCAs) == 0 {
http.Error(w, noUserCARefusal(host.Tenant, a.URL(userCAPath(host.Tenant))), http.StatusConflict)
return
}
// Certified-host-key precondition, the other half of the same story: a guest
// whose host key nothing signed is unreachable through the gate, and the
// signing depends on the agent that will run it. Refuse here, where the
// operator can still upgrade the host, rather than at connect time — by then
// the only remedy is to recreate the VM (see vmssh, which refuses the dial).
//
// A silent host is not judged for this floor — the row is desired state and
// a newer agent may materialize it, with vmssh refusing the dial if the
// guest it eventually boots carries no certificate. See refuseBelowFloor.
hostState, hostHasSpoken := a.reg.Get(req.HostID)
hostHasSpoken = hostHasSpoken && hostState.Online
if a.refuseBelowFloor(w, req.HostID, release.CertifiedHostKeys, false) {
return
}
// Volumes, the fifth refusal of this shape and the first that judges an
// OFFLINE host (see refuseBelowFloor): an old agent ignores volume_ids
// and boots the guest bare, so the data the tenant meant for the volume
// lands on the root disk — the one thing they did not ask for.
//
// It runs before the capacity block because it feeds it: a pending claim's
// bytes are disk this create commits, and the host has to have room for
// both. The claims are read against the HOST's tenant, which the authz
// gate above has already proven is the caller's own.
claimIDs, claimGB, msg, code, err := a.resolveClaims(host.Tenant, req.VolumeClaims)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
if msg != "" {
http.Error(w, msg, code)
return
}
if len(claimIDs) > 0 && a.refuseBelowFloor(w, req.HostID, volumesFeature, true) {
return
}
// Capacity precondition, the third refusal of this same shape: the request
// is fine, and the host has been told not to serve it. A host whose operator
// capped it is already holding as much as it may hold cannot boot one more
// guest — the agent refuses it at materialization ("host capacity limit
// reached", see reconcile.Engine) and the VM ends up failed, minutes later,
// with an answer that existed here. So the create answers 409 now, naming
// the dimension and the numbers.
//
// It answers 409 ONLY there. An uncapped host advertises the machine's
// totals and its agent admits past them on purpose (sparse disks, memory
// that is not preallocated), so treating those totals as a wall would refuse
// VMs the fleet is designed to run. overCapacityRefusal judges a dimension
// only when the report proves a cap, and says nothing otherwise.
//
// Judged only on a host that has spoken, exactly like the refusal above and
// for the same reason: reported capacity is registry state, so an offline or
// never-reporting host has zeroes that say nothing about the machine. Such a
// host takes the create as desired state, and the agent's own admission
// check — which is anyway the authority for the races this cannot close —
// remains the backstop.
if hostHasSpoken {
held, err := a.st.CommittedOnHost(req.HostID)
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
// claimGB is the PENDING claims only: a bound claim's volume already
// sits in held (CommittedOnHost counts live volumes), so adding it
// here would charge the host twice for disk this VM is re-using.
want := store.Alloc{VCPUs: req.VCPUs, MemMB: req.MemMB, DiskGB: req.DiskGB + claimGB}
if msg := overCapacityRefusal(host.Name, req.HostID, want, held, hostState.Report); msg != "" {
http.Error(w, msg, http.StatusConflict)
return
}
}
// Named-network precondition, the fourth refusal of this shape: the VM
// asked for a network, and only the agent's own configuration can say
// whether this host serves it. Advertisement arrives in the Hello, so —
// unlike the offline-tolerant refusals above — silence refuses: an agent
// that predates the field advertises nothing and would silently NAT the
// guest, which is the one outcome the spec forbids. A VM that asked for
// the LAN either gets it or does not exist.
//
// hostHasSpoken already folds in Online (see above): the server trusts no
// Hello-derived fact, including a network name the host may have just
// advertised, until the host is Online. A freshly-connected host can
// therefore see this refusal fire for one report cadence even though it
// did advertise the network — self-healing once the next report lands,
// and uniform with the sibling refusals this one sits beside.
if req.Network != "" {
if !hostHasSpoken || !slices.Contains(hostState.HostNetworks, req.Network) {
http.Error(w, noNetworkRefusal(host.Name, req.HostID, req.Network, host.OS, hostState.HostNetworks, hostHasSpoken), http.StatusConflict)
return
}
}
// Install the SSH key into user-supplied cloud-init. When only one of the
// two is set the seed builder handles it (verbatim user-data, or the
// generated default template); it's the BOTH case that used to silently
// drop the key. AddSSHKey is format-aware: it merges into a #cloud-config,
// or wraps other formats (shell script, etc.) in a MIME archive with a key
// part — eitri never edits the user's payload, only adds to it. An
// un-handleable format (jinja/gzip) is a clear 400, not a silent no-op.
//
// Two things worth knowing: (1) the stored user-data is the REWRITTEN form
// (a merged doc or a MIME archive), not the exact text submitted — write-only,
// so it is never echoed back. (2) unlike the key-only path (which generates a
// full users: block with sudo), here the key is added to the default user
// only; on a stock Ubuntu image that is `ubuntu`, but a custom base image
// with a different default user gets the key there. The key is embedded as a
// YAML scalar node, so it cannot inject structure (validateCreateVM's
// single-line check is belt-and-suspenders, not the load-bearing guard).
// Describe the key BEFORE the merge below clears the field. The record of
// what eitri installed must survive that clearing — it is the whole reason
// the console can answer "which key did you put in this VM".
keyType, keyFP, keyComment := describeKey(req.SSHAuthorizedKey)
if req.CloudInit != "" && req.SSHAuthorizedKey != "" {
merged, err := cloudinit.AddSSHKey(req.CloudInit, req.SSHAuthorizedKey)
if err != nil {
http.Error(w, "cannot add ssh_authorized_key to cloud_init: "+err.Error(), http.StatusBadRequest)
return
}
req.CloudInit = merged
req.SSHAuthorizedKey = "" // installed into cloud-init; don't also carry it separately
}
// Generate ID here so we can return it.
id := random.Hex(16)
vm := store.VM{
ID: id,
HostID: req.HostID,
Name: req.Name,
ImageURL: req.ImageURL,
ImageSHA256: req.ImageSHA256,
CloudInit: req.CloudInit,
SSHAuthorizedKey: req.SSHAuthorizedKey,
InjectedKeyType: keyType,
InjectedKeyFP: keyFP,
InjectedKeyComment: keyComment,
TrustedCAs: store.FreezeCAs(tenantCAs),
VCPUs: req.VCPUs,
MemMB: req.MemMB,
DiskGB: req.DiskGB,
PowerState: req.PowerState,
Network: req.Network,
// Resolved to ids, in the order the caller named them, which is the
// order the guest will see the devices in. CreateVM binds them inside
// its own transaction — placement and binding are one commit.
VolumeClaimIDs: claimIDs,
}
// The row carries no host key. A guest's host key is generated by the host
// that runs it and never leaves that machine; the host reports the public
// half on its next report and the control plane signs a certificate for it
// (see syncsvc.signAndRecordHostCert). Creating a VM therefore involves no
// key material at all, which is why there is nothing here to guard.
// The claim cases below are the same three the admission above already
// checks, re-answered from inside the transaction that is authoritative
// for them: the preflight read and this write are separate moments, and
// another create can attach a claim in between. That is the race the
// attachment table's unique index referees, and this is where its verdict
// is turned back into an answer.
var attached *store.ClaimAttachedError
var pinned *store.ClaimPinnedError
if err := a.st.CreateVM(vm); err != nil {
switch {
case errors.Is(err, store.ErrNameTaken):
http.Error(w, "name already in use", http.StatusConflict)
case errors.Is(err, store.ErrHostNotFound):
http.Error(w, "unknown host_id", http.StatusBadRequest)
case errors.Is(err, store.ErrHostNotEnrolled):
http.Error(w, "host is not accepting new VMs", http.StatusConflict)
case errors.Is(err, store.ErrClaimNotFound):
http.Error(w, "unknown volume claim", http.StatusNotFound)
case errors.As(err, &attached):
http.Error(w, "volume claim "+attached.ClaimID+" is attached to vm "+attached.VMID, http.StatusConflict)
case errors.As(err, &pinned):
http.Error(w, "volume claim "+pinned.ClaimID+" is bound to host "+pinned.HostID+
"; its data lives there, so a VM using it must be placed there", http.StatusConflict)
default:
http.Error(w, "internal error", http.StatusInternalServerError)
}
return
}
a.audit(host.Tenant, "vm.create", map[string]string{"vm_id": id, "name": req.Name, "host_id": req.HostID,
"volume_claims": strings.Join(claimIDs, ",")})
a.hub.Poke(req.HostID)
a.notif.notify()
writeJSON(w, http.StatusCreated, types.CreateVMResponse{ID: id, Name: req.Name})
}
// mutateVM implements the choreography shared by every VM mutation endpoint
// (patch/delete/restore): check ownership, then run mutate; a sql.ErrNoRows
// becomes the caller's not-found response, any other error a generic 500. On
// success an audit row is appended and the host's desired-state stream poked
// from the pre-read row (Name/HostID are immutable under every mutation, so
// no post-mutate re-fetch is needed — audit/poke fire even if the row
// vanishes post-mutate, since the mutation itself already succeeded).
// Finally SSE watchers are notified and the response is written as 204,
// unconditionally.
func (a *API) mutateVM(w http.ResponseWriter, r *http.Request, id string, mutate func(id string) error,
notFoundMsg string, notFoundStatus int,
auditAction string, auditDetail func(vm store.VM) map[string]string) {
vm, err := a.st.GetVM(id)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
http.Error(w, notFoundMsg, notFoundStatus)
return
}
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
if !mayActAs(principalFromContext(r), vm.Tenant) {
http.Error(w, notFoundMsg, notFoundStatus)
return
}
if err := mutate(id); err != nil {
if errors.Is(err, sql.ErrNoRows) {
http.Error(w, notFoundMsg, notFoundStatus)
return
}
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
a.audit(vm.Tenant, auditAction, auditDetail(vm))
a.hub.Poke(vm.HostID)
a.notif.notify()
w.WriteHeader(http.StatusNoContent)
}
func (a *API) handlePatchVM(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
var req types.PatchVMRequest
if !decodeJSON(w, r, &req) {
return
}
if req.PowerState != "running" && req.PowerState != "stopped" {
http.Error(w, "power_state must be running or stopped", http.StatusBadRequest)
return
}
a.mutateVM(w, r, id, func(id string) error { return a.st.SetVMPower(id, req.PowerState) },
"not found", http.StatusNotFound,
"vm.power", func(vm store.VM) map[string]string {
return map[string]string{"vm_id": id, "name": vm.Name, "power": req.PowerState}
})
}
func (a *API) handleDeleteVM(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
a.mutateVM(w, r, id, a.st.TombstoneVM,
"not found", http.StatusNotFound,
"vm.delete", func(vm store.VM) map[string]string {
return map[string]string{"vm_id": id, "name": vm.Name}
})
}
// handleRestoreVM un-tombstones a VM that is still within the teardown grace
// window (row present, not yet hard-deleted). The store reverses the tombstone
// and bumps the epoch; the agent's un-delete path re-adopts the guest and
// converges it back toward its power_state — so the server only pokes.
func (a *API) handleRestoreVM(w http.ResponseWriter, r *http.Request) {
id := r.PathValue("id")
a.mutateVM(w, r, id, a.st.RestoreVM,
"vm not restorable (already destroyed or not deleted)", http.StatusConflict,
"vm.restore", func(vm store.VM) map[string]string {
return map[string]string{"vm_id": id, "name": vm.Name}
})
}
// vmByID fetches the VM row with the given id via an indexed primary-key
// lookup (callers need name + host_id). Any error — including a missing row —
// reports ok=false, matching the callers' non-fatal not-found handling.
func (a *API) vmByID(vmID string) (store.VM, bool) {
vm, err := a.st.GetVM(vmID)
if err != nil {
return store.VM{}, false
}
return vm, true
}