internal/server/boot/boot.go
Ref: Size: 17.3 KiB History
// Package boot implements the eitri-server command line behind a tested RunCLI
// so cmd/eitri-server stays thin wiring (arch R14). It opens the store, handles
// the server certificate, starts the housekeeping goroutines, binds the QUIC
// and HTTP listeners, constructs the API, and wires the SSH jump gate — the
// single-node control plane's whole boot sequence.
//
// No TLS termination happens here — front eitri-server with a reverse proxy for
// TLS.
package boot
import (
"context"
"errors"
"flag"
"fmt"
"log/slog"
"net/http"
"os"
"os/signal"
"strings"
"syscall"
"time"
"github.com/a73x/eitri/internal/covsnap"
"github.com/a73x/eitri/internal/guest"
"github.com/a73x/eitri/internal/joinblob"
"github.com/a73x/eitri/internal/server/api"
serverconfig "github.com/a73x/eitri/internal/server/config"
"github.com/a73x/eitri/internal/server/delegation"
"github.com/a73x/eitri/internal/server/health"
"github.com/a73x/eitri/internal/server/hub"
"github.com/a73x/eitri/internal/server/mcphttp"
"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/server/syncsvc"
"github.com/a73x/eitri/internal/server/web"
"github.com/a73x/eitri/internal/transport"
"github.com/quic-go/quic-go"
"golang.org/x/crypto/ssh"
)
// httpServerReadHeaderTimeout bounds an unauthenticated client's request-header
// phase, the same defense the SSH gate's handshakeGrace and the sync service's
// helloGrace give their protocols: without it a client that opens a connection
// and dribbles (or never finishes) its request line + headers parks a goroutine
// and an fd indefinitely, and enough such connections — the Slowloris — starve
// the plane of accept slots. It governs only the header read, cleared once the
// headers are in, so it never touches the SSE event stream or the serial-console
// WebSocket that follow.
const httpServerReadHeaderTimeout = 20 * time.Second
// httpServerIdleTimeout reclaims a keep-alive connection that has gone quiet
// between requests. It applies only while a connection is idle — never during an
// in-flight request — so a live SSE stream or console session is untouched.
const httpServerIdleTimeout = 120 * time.Second
// httpServer builds the plane's HTTP server with the timeouts a public listener
// needs. Two of the four are deliberately left at zero: ReadTimeout and
// WriteTimeout each bound the WHOLE request, and this handler carries two
// long-lived-by-design responses — the SSE fleet-event stream (api/events.go,
// which writes for as long as a console tab is open) and the serial-console
// WebSocket (api/console.go). A WriteTimeout would sever a live console
// mid-session; a ReadTimeout cancels the request context at its deadline via
// net/http's client-disconnect background read, cutting the SSE stream the same
// way. The Slowloris is answered by ReadHeaderTimeout instead, which streaming
// does not feel. Front this listener with a reverse proxy for TLS and coarse
// body limits.
func httpServer(addr string, handler http.Handler) *http.Server {
return &http.Server{
Addr: addr,
Handler: handler,
ReadHeaderTimeout: httpServerReadHeaderTimeout,
IdleTimeout: httpServerIdleTimeout,
}
}
// RunCLI dispatches the eitri-server command line (everything after the binary
// name, --version excluded — that stays in cmd/eitri-server). It parses the
// -config flag and runs the control plane until SIGINT/SIGTERM.
func RunCLI(args []string) error {
fs := flag.NewFlagSet("eitri-server", flag.ContinueOnError)
cfgPath := fs.String("config", "/etc/eitri/server.json", "config file")
if err := fs.Parse(args); err != nil {
return err
}
return run(*cfgPath)
}
// run wires and serves the control plane, blocking until SIGINT/SIGTERM (a nil
// return) or an always-on server fails (a non-nil return). Every synchronous
// startup invariant surfaces as a returned error so cmd/eitri-server can exit
// non-zero — the process-fatal outcome each site previously reached directly.
func run(cfgPath string) error {
// Load enforces every startup invariant (required keys, OIDC block, URL
// shapes) — see internal/server/config, where the rules are tested.
cfg, err := serverconfig.Load(cfgPath)
if err != nil {
return fmt.Errorf("config %s: %w", cfgPath, err)
}
// Install the log handler before anything else logs, so the first line the
// server writes already obeys the configured floor.
if err := installLogger(cfg.LogLevel); err != nil {
return fmt.Errorf("config: %w", err)
}
// The key that seals the on-disk host CA and gate host key — the only key
// material this server encrypts at rest — decoded once and handed to each
// place that seals or opens: the gate's key files (setupSSHGate) and the host
// CA (internal/server/sshca). Tenant user CAs are BYO public keys and seal
// nothing. Load has already enforced it.
kek, err := cfg.KEKBytes()
if err != nil {
return fmt.Errorf("config: %w", err)
}
st, err := store.Open(cfg.DBPath, cfg.CIDRPool)
if err != nil {
return fmt.Errorf("open store: %w", err)
}
certPEM, certFP, err := st.ServerCert()
if err != nil {
return fmt.Errorf("server cert: %w", err)
}
keyPEM, err := st.ServerKeyPEM()
if err != nil {
return fmt.Errorf("server key: %w", err)
}
// Log the identity agents pin — the operator verifies this out-of-band
// during the rotation ceremony (docs/cert-rotation.md step 3).
slog.Info("server cert", "fingerprint", certFP)
// Rotation nudge: the agent pin ignores expiry so nothing breaks at
// NotAfter, but a long-lived key is a widening forgery window. Warn while
// inside the renewal window — at startup AND daily, because servers here
// are long-lived daemons that can cross into the window (or past expiry)
// without ever restarting. See docs/cert-rotation.md for the ceremony.
warnIfRenewalDue := func() {
notAfter, due := transport.CertRenewalDue(certPEM, time.Now())
if !due {
return
}
if notAfter.IsZero() {
slog.Warn("server cert unparseable — inspect server.crt (docs/cert-rotation.md)")
return
}
slog.Warn("server cert renewal due — rotate and re-enroll agents (docs/cert-rotation.md)",
"not_after", notAfter.Format(time.RFC3339))
}
warnIfRenewalDue()
// Audit retention: bound the append-only log (default 90 days, 0 disables).
// A negative value is almost certainly a typo — refuse rather than silently
// keeping the audit log forever ("0" is the explicit disable spelling).
auditRetention, err := serverconfig.ParseDuration("audit_retention", cfg.AuditRetention, 90*24*time.Hour,
func(d time.Duration) bool { return d >= 0 }, ">= 0")
if err != nil {
return fmt.Errorf("config: %w", err)
}
pruneAudit := func() {
if auditRetention <= 0 {
return
}
if n, err := st.PruneAudit(auditRetention); err != nil {
slog.Warn("audit prune failed", "err", err)
} else if n > 0 {
slog.Info("audit pruned", "rows", n, "retention", auditRetention)
}
}
pruneAudit()
// Daily housekeeping: cert-renewal nudge + audit retention.
go func() {
for range time.Tick(24 * time.Hour) {
warnIfRenewalDue()
pruneAudit()
}
}()
// Fail fast if the advertised addresses are non-empty but malformed (e.g. a
// URL with no scheme): otherwise every enroll-token mint would 500 at runtime.
if _, err := joinblob.Encode(cfg.AdvertiseHTTP, cfg.AdvertiseQUIC, "startup-probe", certFP); err != nil {
return fmt.Errorf("advertise_http/advertise_quic invalid: %w", err)
}
// SSH jump gate (§B): nil when ssh_listen is unset (gate OFF); see
// setupSSHGate. The listener itself is started below, once syncsvc.Service
// (the tunnel dialer) exists.
sshGate, err := setupSSHGate(cfg, kek)
if err != nil {
return err
}
reg := registry.New(time.Now)
h := hub.New()
a := api.New(api.Config{HostSecret: []byte(cfg.HostSecret),
DefaultImages: defaultImages(cfg.DefaultImages),
ServerCertSHA256: certFP,
AdvertiseHTTP: cfg.AdvertiseHTTP,
AdvertiseQUIC: cfg.AdvertiseQUIC,
OIDC: api.OIDCConfig{
Issuer: cfg.OIDC.Issuer,
ClientID: cfg.OIDC.ClientID,
ClientSecret: cfg.OIDC.ClientSecret,
PublicURL: cfg.OIDC.PublicURL,
AllowedDomains: cfg.OIDC.AllowedDomains,
AllowedIdentities: cfg.OIDC.AllowedIdentities,
}},
st, reg, h)
tlsConf, err := transport.ServerTLS(certPEM, keyPEM)
if err != nil {
return fmt.Errorf("server tls: %w", err)
}
lis, err := quic.ListenAddr(cfg.QUICListen, tlsConf,
// Shared with the agent dialer via transport so the two ends can't drift.
transport.SyncQUICConfig())
if err != nil {
return fmt.Errorf("quic listen: %w", err)
}
maxCredAge, err := serverconfig.ParseDuration("credential_max_age", cfg.CredentialMaxAge, 0, nil, "")
if err != nil {
return fmt.Errorf("config: %w", err)
}
// Publish the host CA: no-op when the gate is off, so the endpoint 404s.
sshGate.wireAPI(a)
// Say the policy out loud at boot: a max age is a deadline, not a rotation —
// nothing re-issues a credential, so every host reaching it needs enrolling
// again by hand. An operator who set it should see that stated somewhere.
if maxCredAge > 0 {
slog.Warn("host credentials expire by age; nothing renews them, so each host must be re-enrolled before it elapses",
"credential_max_age", maxCredAge.String())
}
svc := syncsvc.New(st, reg, h, []byte(cfg.HostSecret), maxCredAge)
// Certify the host keys guests generate for themselves. No-op when the gate
// is off, and then no guest waits for a certificate.
sshGate.wireSync(svc)
// The credentials tenants have lent eitri. In memory only, so a restart is a
// revocation; the sweeper keeps the keyring bounded by tenants that are
// actually using it. Wired to the API only when there is a CA to verify
// certificates against — with the gate off the routes answer 503, which is
// the same condition that makes remote exec refuse.
keyring := delegation.New(time.Now, guest.LoginUser)
if sshGate != nil {
a.SetDelegations(keyring)
}
go sweepDelegations(context.Background(), keyring)
// Console broker: the API bridges browser WebSockets to agent console
// streams over the live sync connections the service tracks.
a.SetConsoleDialer(svc)
// fatal collects the first failure from any always-on server goroutine (the
// SSH gate, QUIC, HTTP). Buffered so a failing server never blocks on send;
// run returns the first error and cmd/eitri-server exits non-zero — the same
// process-fatal outcome each site previously reached via os.Exit(1). A dead
// server must not run silently.
fatal := make(chan error, 3)
// SSH jump gate listener (no-op when off); a failed bind is fatal, like
// QUIC/HTTP below.
if err := sshGate.startListener(st, svc, fatal); err != nil {
return err
}
go func() {
slog.Info("quic listening", "addr", cfg.QUICListen)
// Serve loops until the listener fails, so it never returns nil: the
// only way out of the accept loop is the error being reported here.
err := svc.Serve(context.Background(), lis)
fatal <- fmt.Errorf("quic serve: %w", err)
}()
// Background: finalize drained decommissioning hosts.
go a.StartBackground(context.Background())
// Serve the REST API + SSE under /api/ and the embedded SPA everywhere else.
root := http.NewServeMux()
root.Handle("/api/", a.Handler())
// Sign-in endpoints live OUTSIDE /api/ and its auth middleware: they are how
// a browser establishes a session in the first place (spec §2).
root.Handle("/auth/", a.AuthHandler())
// The MCP endpoint: eitri's whole toolset, for a client anywhere on the
// internet holding nothing but a PAT. It speaks JSON-RPC
// rather than the REST contract, so like /auth/* it lives outside the route
// table — but wrapped in the API's own authentication, so a caller reaching
// it is the same authenticated principal /api/v1 would see. Both patterns are
// registered because a client may address it with or without a trailing path.
mcpHandler := a.UserAuth(mcphttp.New(mcphttp.Deps{
Handler: a.Handler(),
Creds: delegatedCreds{keyring: keyring, st: st},
TCP: svc,
Lookup: vmLookup(st),
HostCA: sshGate.hostCAPublicKey(),
Gate: cfg.SSHGateDomain,
VMUser: guest.LoginUser,
// The API's own origin, so a refusal names something dialable rather
// than a path on whichever host the caller happens to be talking to.
DelegationsURL: a.URL("/api/v1/delegations"),
}))
root.Handle("/mcp", mcpHandler)
root.Handle("/mcp/", mcpHandler)
// Unauthenticated probes (outside /api/, so a load balancer or the deploy
// script needs no token). /livez is process-up; /readyz gates on the
// dependencies the server needs to actually serve — the DB. The QUIC
// listener bind is a startup invariant: quic.ListenAddr above returns an
// error that stops the process before this HTTP server serves, so a response
// here already implies QUIC bound.
root.HandleFunc("/livez", health.Live)
root.Handle("/readyz", health.Ready(3*time.Second,
health.Check{Name: "db", Probe: st.Ping},
))
root.Handle("/", web.Handler())
// Graceful shutdown: SIGINT/SIGTERM stops accepting, drains in-flight
// requests, then closes the API — stopping the SSE snapshot-hub goroutine and
// releasing its notifier subscription. The QUIC listener and background worker
// die with the process.
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
// Release discovery: absent field ⇒ eitri.sh default; explicit "" disables.
manifestURL := "https://eitri.sh/dl/latest/manifest.json"
if cfg.ReleaseManifestURL != nil {
manifestURL = *cfg.ReleaseManifestURL
}
if manifestURL != "" {
rel := release.New(manifestURL)
go rel.Poll(ctx, 24*time.Hour, func(err error) {
slog.Warn("release manifest refresh failed", "err", err)
})
a.SetReleaseSource(rel)
}
a.SetAgentUpgrader(svc)
// Flush integration-coverage counters on SIGUSR1 (no-op unless built with
// -cover and GOCOVERDIR is set). Lets the deploy boot-gate snapshot coverage
// from the live server without bouncing the process.
covsnap.Install(ctx)
srv := httpServer(cfg.HTTPListen, securityHeaders(root))
go func() {
slog.Info("http listening", "addr", cfg.HTTPListen)
if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
fatal <- fmt.Errorf("http serve: %w", err)
}
}()
select {
case err := <-fatal:
return err
case <-ctx.Done():
}
slog.Info("shutting down")
shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(shutdownCtx); err != nil {
slog.Warn("http graceful shutdown", "err", err)
}
a.Close()
return nil
}
// delegatedCreds pairs the in-memory keyring with the store, which is the one
// place those two facts meet. It lives here rather than in vmssh so that
// package stays free of the store (R1).
type delegatedCreds struct {
keyring *delegation.Keyring
st *store.Store
}
func (d delegatedCreds) Delegated(tenant string) (ssh.Signer, bool) {
return d.keyring.Signer(tenant)
}
func (d delegatedCreds) TenantHasUserCA(tenant string) (bool, error) {
return d.st.TenantHasUserCA(tenant)
}
// delegationSweepInterval is how often the keyring drops entries nothing is
// using. Nothing depends on it being prompt — an expired delegation stops
// working the moment it expires, whether or not it has been swept — so it is
// slow on purpose.
const delegationSweepInterval = 10 * time.Minute
// sweepDelegations keeps the keyring bounded by ACTIVE tenants rather than by
// every tenant that ever started a delegation.
func sweepDelegations(ctx context.Context, k *delegation.Keyring) {
t := time.NewTicker(delegationSweepInterval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case <-t.C:
k.Sweep()
}
}
}
// defaultImages translates the config's per-architecture guest images into the
// API's own type, so the API package does not import the config schema (R1: the
// wiring converts, the leaves stay independent).
func defaultImages(in map[string]serverconfig.DefaultImage) map[string]api.DefaultImage {
out := make(map[string]api.DefaultImage, len(in))
for arch, img := range in {
out[arch] = api.DefaultImage{URL: img.URL, SHA256: img.SHA256}
}
return out
}
// installLogger sets the process-wide slog handler at the configured floor.
// Without this the default handler applies, whose floor is info — which is why
// every Debug line in the server was unreachable in a running plane no matter
// what the operator did.
//
// An empty level means info, which is the floor the default handler already
// used — so a config that says nothing keeps the same lines. It does not keep
// the same FORMATTING: this installs a TextHandler where the log package's own
// handler was, so an unset level still changes how a line looks, and anything
// parsing the journal should be checked once.
//
// An unrecognized level is a config error, not a silent fallback: an operator
// who wrote "verbose" and got info would conclude the logging is broken rather
// than the spelling.
func installLogger(level string) error {
lv := slog.LevelInfo
switch strings.ToLower(strings.TrimSpace(level)) {
case "", "info":
case "debug":
lv = slog.LevelDebug
case "warn", "warning":
lv = slog.LevelWarn
case "error":
lv = slog.LevelError
default:
return fmt.Errorf("log_level %q is not one of debug, info, warn, error", level)
}
slog.SetDefault(slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: lv})))
return nil
}