Files
rollcall/cmd/rollcall/rackd.go
T
Ilker ManapandClaude Opus 5.5 2f708e8f69
CI / agent (push) Successful in 3s
CI / site (push) Successful in 3s
CI / go (push) Successful in 47s
Root rotation without re-joining: prepare, switch, retire
The CA directory holds the active root and, during a rotation, the next
or the previous one; everything that trusts the region gets the trust
bundle (all of them). Rack controllers receive it in their state, signed
by the active root; the root update unit installs it only when the
signature verifies against the roots it already trusts, and binary
signatures verify against any of them. After switch racks renew their
client and server certificates from the new root by themselves and
report what they run on; switch and retire wait for every rack (-force).

/ca.pem stays one certificate, the root of that server's certificate, and
the PXE command line names that root, so agents in older images keep
working; /ca-bundle.pem is the bundle. Deployed machines get one file per
root and a daily timer that refreshes them from their rack over HTTPS.

Refs #12

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-31 15:37:44 +01:00

884 lines
28 KiB
Go

package main
// `rollcall rack`: a rack controller next to the machines. It registers with
// the region, long-polls its desired state and writes it (PXE files, dnsmasq
// configuration, host files), runs BMC commands the region queues for it,
// serves boot and OS images from a local cache filled from the region on
// first use, and proxies the node API (commissioning, deploy jobs, overlays,
// phone-home) to the region, marking requests with its name.
import (
"bytes"
"context"
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"net"
"net/http"
"net/http/httputil"
"net/url"
"os"
"os/user"
"path"
"path/filepath"
"runtime"
"strconv"
"strings"
"sync"
"time"
)
type RackConfig struct {
Name string `json:"name,omitempty"` // default: hostname
Region string `json:"region,omitempty"` // the region's URL, e.g. http://10.0.0.234
Secret string `json:"secret,omitempty"` // the region's rack_secret
URL string `json:"url,omitempty"` // how machines reach this rack, e.g. http://10.213.78.1
Listen string `json:"listen,omitempty"` // default ":80"
TFTPRoot string `json:"tftp_root,omitempty"` // default /srv/tftp
DnsmasqDir string `json:"dnsmasq_dir,omitempty"` // default /var/lib/rollcall/dnsmasq
CacheDir string `json:"cache_dir,omitempty"` // default /var/cache/rollcall
IPMITool string `json:"ipmitool,omitempty"` // default ipmitool
DHCPInterfaces []string `json:"dhcp_interfaces,omitempty"` // default: the interfaces with an address in a served subnet
CAFile string `json:"ca_file,omitempty"` // set by rollcall rack join: the region's CA,
CertFile string `json:"cert_file,omitempty"` // this rack controller's certificate
KeyFile string `json:"key_file,omitempty"` // and key; with them the secret is not needed
AutoUpdate *bool `json:"auto_update,omitempty"` // follow the region's version (default true)
UpdateDir string `json:"update_dir,omitempty"` // default /var/lib/rollcall/update
// CertStateDir: where a renewed certificate goes (cert and key in one file,
// rack.pem). The files from rack join are root's; this directory is the
// service user's. Default /var/lib/rollcall/rack-cert.
CertStateDir string `json:"cert_state_dir,omitempty"`
// TLSListen: HTTPS for the agents (node API, images), with a server
// certificate from the region for the URL's host. Default <url host>:443;
// "off" disables it.
TLSListen string `json:"tls_listen,omitempty"`
}
func (c RackConfig) renewedPath() string {
return filepath.Join(orDefault(c.CertStateDir, "/var/lib/rollcall/rack-cert"), "rack.pem")
}
type rackd struct {
srvMu sync.Mutex
srvCert *tls.Certificate // server certificate for agents (HTTPS)
tlsURL string
cfg RackConfig
region *url.URL
hc *http.Client // long polls
fast *http.Client
big *http.Client // image downloads, no timeout
gen int64
tlsc *tls.Config
certMu sync.Mutex
upd rackUpdater
cert *tls.Certificate
headMu sync.Mutex
heads map[string]headInfo
fetch sync.Mutex // one boot-file prefetch at a time
}
type headInfo struct {
size int64
mod time.Time
at time.Time
}
func loadRackConfig(p string) (RackConfig, error) {
var c RackConfig
b, err := os.ReadFile(p)
if err != nil {
return c, err
}
if err := json.Unmarshal(b, &c); err != nil {
return c, fmt.Errorf("%s: %w", p, err)
}
if c.Name == "" {
c.Name, _ = os.Hostname()
}
if c.Listen == "" {
c.Listen = ":80"
}
if c.TFTPRoot == "" {
c.TFTPRoot = "/srv/tftp"
}
if c.DnsmasqDir == "" {
c.DnsmasqDir = "/var/lib/rollcall/dnsmasq"
}
if c.CacheDir == "" {
c.CacheDir = "/var/cache/rollcall"
}
if c.IPMITool == "" {
c.IPMITool = "ipmitool"
}
c.Region, c.URL = strings.TrimRight(c.Region, "/"), strings.TrimRight(c.URL, "/")
for _, k := range []struct{ v, n string }{{c.Region, "region"}, {c.URL, "url"}} {
if k.v == "" {
return c, fmt.Errorf("%s: %s is required", p, k.n)
}
}
if c.Secret == "" && c.CertFile == "" {
return c, fmt.Errorf("%s: not enrolled: run rollcall rack join CODE (or set the legacy secret)", p)
}
return c, nil
}
// clients builds the HTTP clients; with a certificate they speak mutual TLS
// and pick up a renewed certificate without being rebuilt.
func (r *rackd) clients() error {
r.tlsc = &tls.Config{MinVersion: tls.VersionTLS12}
if r.cfg.CertFile != "" {
cab, err := os.ReadFile(r.cfg.CAFile)
if err != nil {
return err
}
pool := x509.NewCertPool()
if !pool.AppendCertsFromPEM(cab) {
return fmt.Errorf("%s: no certificate", r.cfg.CAFile)
}
pair, err := tls.LoadX509KeyPair(r.cfg.CertFile, r.cfg.KeyFile)
if err != nil {
return err
}
pair.Leaf, _ = x509.ParseCertificate(pair.Certificate[0])
r.cert = &pair
if renewed := loadRenewed(r.cfg.renewedPath(), pool, r.cfg.Name, pair.Leaf); renewed != nil {
r.cert = renewed
}
r.tlsc.RootCAs = pool
r.tlsc.VerifyConnection = regionLeaf
r.tlsc.GetClientCertificate = func(*tls.CertificateRequestInfo) (*tls.Certificate, error) {
r.certMu.Lock()
defer r.certMu.Unlock()
return r.cert, nil
}
}
tr := func() *http.Transport {
return &http.Transport{TLSClientConfig: r.tlsc, Proxy: http.ProxyFromEnvironment, MaxIdleConnsPerHost: 8,
DialContext: (&net.Dialer{Timeout: 10 * time.Second}).DialContext}
}
r.hc = &http.Client{Timeout: 45 * time.Second, Transport: tr()}
r.fast = &http.Client{Timeout: 30 * time.Second, Transport: tr()}
r.big = &http.Client{Transport: tr()}
return nil
}
// renewLoop asks for a new certificate a month before the current one ends.
func (r *rackd) renewLoop() {
for {
r.certMu.Lock()
leaf := r.cert.Leaf
r.certMu.Unlock()
if leaf != nil && time.Until(leaf.NotAfter) < renewBefore {
if err := r.renew(); err != nil {
log.Printf("certificate renewal: %v", err)
}
}
time.Sleep(12 * time.Hour)
}
}
func (r *rackd) renew() error {
key, csr, err := newCSR("rack:" + r.cfg.Name)
if err != nil {
return err
}
var resp struct {
Cert string `json:"cert"`
}
if _, err := r.call(r.fast, "POST", "/api/v1/rack/renew", enrollRequest{CSR: csr}, &resp); err != nil {
return err
}
pair, err := tls.X509KeyPair([]byte(resp.Cert), key)
if err != nil {
return err
}
pair.Leaf, _ = x509.ParseCertificate(pair.Certificate[0])
// one file, renamed into place: a crash never leaves a key without its certificate
if err := writeAtomic(r.cfg.renewedPath(), append([]byte(resp.Cert), key...), 0o600); err != nil {
return fmt.Errorf("keep the renewed certificate (the region already switched to it; until this works the previous one is accepted for 7 days): %w; create the directory: install -d -m 0700 -o rollcall -g rollcall %s", err, filepath.Dir(r.cfg.renewedPath()))
}
r.certMu.Lock()
r.cert = &pair
r.certMu.Unlock()
log.Printf("renewed the rack certificate, valid until %s", pair.Leaf.NotAfter.Format("2006-01-02"))
return nil
}
// loadRenewed is the certificate renew() kept, when it belongs to this rack,
// comes from the CA and outlives the one from rack join.
func loadRenewed(p string, pool *x509.CertPool, name string, current *x509.Certificate) *tls.Certificate {
b, err := os.ReadFile(p)
if err != nil {
return nil
}
pair, err := tls.X509KeyPair(b, b)
if err != nil {
log.Printf("%s: %v (ignored)", p, err)
return nil
}
leaf, err := x509.ParseCertificate(pair.Certificate[0])
if err != nil || leaf.Subject.CommonName != "rack:"+name {
return nil
}
if _, err := leaf.Verify(x509.VerifyOptions{Roots: pool, KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}}); err != nil {
log.Printf("%s: %v (ignored)", p, err)
return nil
}
if current != nil && !leaf.NotAfter.After(current.NotAfter) {
return nil
}
pair.Leaf = leaf
return &pair
}
func writeAtomic(p string, data []byte, mode os.FileMode) error {
f, err := os.CreateTemp(filepath.Dir(p), ".tmp-*")
if err != nil {
return err
}
defer os.Remove(f.Name())
if _, err := f.Write(data); err != nil {
f.Close()
return err
}
if err := f.Chmod(mode); err != nil {
f.Close()
return err
}
if err := f.Sync(); err != nil {
f.Close()
return err
}
if err := f.Close(); err != nil {
return err
}
return os.Rename(f.Name(), p)
}
// rackRenew renews the certificate now (rollcall rack renew), as the service user.
func rackRenew(args []string) error {
fs := newFlags("rack renew")
cfgPath := fs.String("config", "/etc/rollcall/rack.json", "rack controller configuration")
_ = fs.Parse(args)
cfg, err := loadRackConfig(*cfgPath)
if err != nil {
return err
}
ru, err := url.Parse(cfg.Region)
if err != nil {
return err
}
r := &rackd{cfg: cfg, region: ru, heads: map[string]headInfo{}}
if err := r.clients(); err != nil {
return err
}
if r.cert == nil {
return errors.New("this rack controller has no certificate (rollcall rack join)")
}
if err := r.renew(); err != nil {
return err
}
fmt.Printf("renewed: %s, valid until %s; restart rollcall-rack to use it\n", cfg.renewedPath(), r.cert.Leaf.NotAfter.Format("2006-01-02"))
return nil
}
func newCSR(cn string) (keyPEM []byte, csrPEM string, err error) {
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
if err != nil {
return nil, "", err
}
der, err := x509.CreateCertificateRequest(rand.Reader, &x509.CertificateRequest{Subject: pkix.Name{CommonName: cn}}, key)
if err != nil {
return nil, "", err
}
kder, _ := x509.MarshalECPrivateKey(key)
return pemEncode("EC PRIVATE KEY", kder), string(pemEncode("CERTIFICATE REQUEST", der)), nil
}
// rackJoin enrolls this host as a rack controller with a code from
// `rollcall controller join-code` and writes rack.json and the certificate.
func rackJoin(args []string) error {
if len(args) < 1 || strings.HasPrefix(args[0], "-") {
return errors.New("usage: rollcall rack join CODE -url http://ADDR [-name N] [-listen ADDR:80] [-config /etc/rollcall/rack.json]")
}
code, err := parseJoinCode(args[0])
if err != nil {
return err
}
fs := newFlags("rack join")
host, _ := os.Hostname()
name := fs.String("name", host, "rack controller name")
u := fs.String("url", "", "how machines reach this rack controller, e.g. http://10.213.78.1 (required)")
listen := fs.String("listen", "", "listen address (default: the url's host, port 80)")
cfgPath := fs.String("config", "/etc/rollcall/rack.json", "configuration to write")
dir := fs.String("dir", "/etc/rollcall/rack", "where the certificate goes")
_ = fs.Parse(args[1:])
if *u == "" {
return errors.New("-url is required: the address machines on this segment reach this host at")
}
cfg := RackConfig{}
if b, err := os.ReadFile(*cfgPath); err == nil {
_ = json.Unmarshal(b, &cfg)
}
cfg.Name, cfg.URL, cfg.Region, cfg.Secret = *name, strings.TrimRight(*u, "/"), "https://"+code.Addr, ""
if *listen != "" {
cfg.Listen = *listen
} else if cfg.Listen == "" {
if pu, err := url.Parse(cfg.URL); err == nil {
cfg.Listen = net.JoinHostPort(pu.Hostname(), "80")
}
}
cfg.CAFile, cfg.CertFile, cfg.KeyFile = filepath.Join(*dir, "ca.pem"), filepath.Join(*dir, "rack.pem"), filepath.Join(*dir, "rack-key.pem")
if cfg.TFTPRoot == "" {
cfg.TFTPRoot = "/srv/tftp"
}
key, csr, err := newCSR("rack:" + cfg.Name)
if err != nil {
return err
}
req := enrollRequest{Secret: base64.RawURLEncoding.EncodeToString(code.Secret[:]), Name: cfg.Name, CSR: csr,
Rack: RackController{URL: cfg.URL, Version: version, Interfaces: localInterfaces(), DHCPInterfaces: cfg.DHCPInterfaces, TFTPRoot: cfg.TFTPRoot}}
body, _ := json.Marshal(req)
hc := &http.Client{Timeout: 30 * time.Second, Transport: &http.Transport{TLSClientConfig: pinnedTLS(code.Fingerprint)}}
resp, err := hc.Post(cfg.Region+"/api/v1/rack/enroll", "application/json", bytes.NewReader(body))
if err != nil {
return fmt.Errorf("cannot reach the region at %s: %w", cfg.Region, err)
}
defer resp.Body.Close()
data, _ := io.ReadAll(resp.Body)
if resp.StatusCode != 200 {
var e apiError
_ = json.Unmarshal(data, &e)
return fmt.Errorf("the region refused: %s %s", resp.Status, e.Error)
}
var out struct {
Name, Cert, CA string
}
if err := json.Unmarshal(data, &out); err != nil {
return err
}
// what is written here is trusted for good: the CA must be the one the code
// names, and the certificate ours from it
caCert, err := parseCertPEM([]byte(out.CA))
if err != nil {
return fmt.Errorf("the region sent no CA: %w", err)
}
if fp := sha256.Sum256(caCert.Raw); string(fp[:16]) != string(code.Fingerprint[:]) || !caCert.IsCA {
return errors.New("the region sent a CA that is not the one in the join code; not trusting it")
}
pair, err := tls.X509KeyPair([]byte(out.Cert), key)
if err != nil {
return fmt.Errorf("the certificate does not match our key: %w", err)
}
leaf, err := x509.ParseCertificate(pair.Certificate[0])
if err != nil {
return err
}
pool := x509.NewCertPool()
pool.AddCert(caCert)
if _, err := leaf.Verify(x509.VerifyOptions{Roots: pool, KeyUsages: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}}); err != nil {
return fmt.Errorf("the certificate is not from the code's CA: %w", err)
}
cfg.Name = out.Name
if err := os.MkdirAll(*dir, 0o750); err != nil {
return err
}
for _, f := range []struct {
p string
b []byte
mode os.FileMode
}{{cfg.CAFile, []byte(out.CA), 0o644}, {cfg.CertFile, []byte(out.Cert), 0o644}, {cfg.KeyFile, key, 0o640}} {
if err := os.WriteFile(f.p, f.b, f.mode); err != nil {
return err
}
}
cb, _ := json.MarshalIndent(cfg, "", " ")
if err := os.MkdirAll(filepath.Dir(*cfgPath), 0o755); err != nil {
return err
}
if err := os.WriteFile(*cfgPath, append(cb, '\n'), 0o640); err != nil {
return err
}
// the service runs as user rollcall: give its group the key and the configuration
if g, err := user.LookupGroup("rollcall"); err == nil {
if gid, err := strconv.Atoi(g.Gid); err == nil {
for _, p := range []string{*dir, cfg.CAFile, cfg.CertFile, cfg.KeyFile, *cfgPath} {
_ = os.Chown(p, 0, gid)
}
}
}
ca, _ := parseCertPEM([]byte(out.CA))
fmt.Printf("joined the region at %s as rack controller %s\n CA %s\n wrote %s and %s\nnext: sh install-rack.sh (first time) or systemctl restart rollcall-rack\n",
cfg.Region, cfg.Name, fpString(certFingerprint(ca)), *cfgPath, *dir)
return nil
}
func runRack(args []string) error {
fs := newFlags("rack")
cfgPath := fs.String("config", "/etc/rollcall/rack.json", "rack controller configuration")
_ = fs.Parse(args)
cfg, err := loadRackConfig(*cfgPath)
if err != nil {
return err
}
ru, err := url.Parse(cfg.Region)
if err != nil {
return fmt.Errorf("region: %w", err)
}
r := &rackd{cfg: cfg, region: ru, heads: map[string]headInfo{}, upd: rackUpdater{tried: map[string]time.Time{}}}
if err := r.clients(); err != nil {
return fmt.Errorf("tls: %w", err)
}
if r.cert != nil {
go r.renewLoop()
} else {
log.Printf("warning: using the legacy shared secret; enroll with rollcall rack join for mutual TLS")
}
for _, d := range []string{filepath.Join(cfg.TFTPRoot, "pxelinux.cfg"), cfg.CacheDir, filepath.Join(cfg.DnsmasqDir, "conf.d")} {
if err := os.MkdirAll(d, 0o755); err != nil {
return err
}
}
go r.loop()
mux := http.NewServeMux()
for _, p := range []string{"/rollcall/", "/images/", "/os/", "/firmware/"} { // firmware: BMCs pull packages from here
mux.HandleFunc("GET "+p, r.serveCached)
mux.HandleFunc("HEAD "+p, r.serveCached)
}
proxy := &httputil.ReverseProxy{
Rewrite: func(pr *httputil.ProxyRequest) {
pr.SetURL(ru)
pr.SetXForwarded()
pr.Out.Header.Del("X-Rollcall-Rack-Secret")
pr.Out.Header.Set("X-Rollcall-Rack", cfg.Name)
if cfg.Secret != "" {
pr.Out.Header.Set("X-Rollcall-Rack-Secret", cfg.Secret)
}
},
Transport: r.big.Transport,
FlushInterval: 100 * time.Millisecond,
}
for _, p := range []string{"/api/v1/commission", "/api/v1/bmc-hint", "/api/v1/jobs/self", "/api/v1/jobs/self/progress",
"/api/v1/nodes/self/overlay", "/api/v1/phone-home/"} {
mux.Handle(p, proxy)
}
mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, 200, obj{"rack": cfg.Name, "gen": r.gen, "version": version})
})
mux.HandleFunc("GET /ca.pem", r.serveCA(false)) // agents check it by its fingerprint
mux.HandleFunc("GET /ca-bundle.pem", r.serveCA(true))
if r.cert != nil && cfg.TLSListen != "off" {
go r.serveTLS(logging(mux))
}
srv := &http.Server{Addr: cfg.Listen, Handler: logging(mux), ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 2 * time.Minute, IdleTimeout: 2 * time.Minute}
log.Printf("rack controller %s: region %s, serving %s on %s (tftp %s, cache %s)", cfg.Name, cfg.Region, cfg.URL, cfg.Listen, cfg.TFTPRoot, cfg.CacheDir)
return srv.ListenAndServe()
}
func (r *rackd) req(ctx context.Context, method, p string, body any) (*http.Request, error) {
var rd io.Reader
if body != nil {
b, _ := json.Marshal(body)
rd = bytes.NewReader(b)
}
req, err := http.NewRequestWithContext(ctx, method, r.cfg.Region+p, rd)
if err != nil {
return nil, err
}
req.Header.Set("X-Rollcall-Rack", r.cfg.Name)
if r.cfg.Secret != "" {
req.Header.Set("X-Rollcall-Rack-Secret", r.cfg.Secret)
}
req.Header.Set("Content-Type", "application/json")
return req, nil
}
func (r *rackd) call(c *http.Client, method, p string, body, out any) (int, error) {
req, err := r.req(context.Background(), method, p, body)
if err != nil {
return 0, err
}
resp, err := c.Do(req)
if err != nil {
return 0, err
}
defer resp.Body.Close()
data, _ := io.ReadAll(resp.Body)
if resp.StatusCode >= 300 {
var e apiError
_ = json.Unmarshal(data, &e)
if e.Error == "" {
e.Error = strings.TrimSpace(string(data))
}
return resp.StatusCode, fmt.Errorf("%s %s: %s", method, p, e.Error)
}
if out != nil {
return resp.StatusCode, json.Unmarshal(data, out)
}
return resp.StatusCode, nil
}
func (r *rackd) register() error {
r.srvMu.Lock()
tlsURL := r.tlsURL
r.srvMu.Unlock()
bundle, serverCA := r.trustReport()
_, err := r.call(r.fast, "POST", "/api/v1/rack/register", RackController{URL: r.cfg.URL, TLSURL: tlsURL, Version: version, Arch: runtime.GOARCH,
Interfaces: localInterfaces(), DHCPInterfaces: r.cfg.DHCPInterfaces, TFTPRoot: r.cfg.TFTPRoot,
CABundleSHA256: bundle, ServerCA: serverCA}, nil)
return err
}
func (r *rackd) loop() {
backoff := time.Second
for {
if err := r.register(); err != nil {
log.Printf("register: %v (retry in %s)", err, backoff)
time.Sleep(backoff)
backoff = min(backoff*2, 30*time.Second)
continue
}
break
}
log.Printf("registered with %s as %s", r.cfg.Region, r.cfg.Name)
go r.workLoop()
go func() { // interfaces change; registration also refreshes last_seen
for range time.Tick(5 * time.Minute) {
if err := r.register(); err != nil {
log.Printf("re-register: %v", err)
}
}
}()
for {
var st rackState
code, err := r.call(r.hc, "GET", "/api/v1/rack/state?since="+strconv.FormatInt(r.gen, 10), nil, &st)
if err != nil {
log.Printf("state: %v", err)
if code == 409 || code == 401 {
_ = r.register()
}
time.Sleep(3 * time.Second)
continue
}
go r.maybeUpdate(st.Update)
go r.maybeTrust(st.Trust)
if st.Gen == r.gen {
continue
}
ch, err := applyRackState(&st, r.cfg.TFTPRoot, r.cfg.DnsmasqDir)
if err != nil {
log.Printf("apply: %v", err)
} else if len(ch) > 0 {
log.Printf("state %d: wrote %s", st.Gen, strings.Join(ch, " "))
}
r.gen = st.Gen
go r.prefetch(st.BootFiles)
}
}
func (r *rackd) workLoop() {
for {
var jobs []rackJob
if _, err := r.call(r.hc, "GET", "/api/v1/rack/work", nil, &jobs); err != nil {
log.Printf("work: %v", err)
time.Sleep(3 * time.Second)
continue
}
for _, j := range jobs {
go func(j rackJob) {
res := rackResult{ID: j.ID}
switch j.Kind {
case "power":
out, err := powerRun(context.Background(), r.cfg.IPMITool, &bmcCreds{Driver: j.Driver, Address: j.Address, Port: j.Port,
User: j.User, Password: j.Password, Extra: j.Extra}, j.Op, j.Arg)
res.Output = out
if err != nil {
res.Error = err.Error()
if out != "" {
res.Error = out + " (" + err.Error() + ")"
}
}
default:
res.Error = "unknown job kind " + j.Kind + " (update this rack controller)"
}
log.Printf("job %s %s %s %s %s: %s", j.ID, j.Kind, j.Driver, j.Op, j.Arg, oneLine(res.Output+" "+res.Error))
if _, err := r.call(r.fast, "POST", "/api/v1/rack/result", res, nil); err != nil {
log.Printf("result %s: %v", j.ID, err)
}
}(j)
}
}
}
// ---------------------------------------------------------------- image cache
// head asks the region for a file's size and modification time (cached briefly).
func (r *rackd) head(rel string) (headInfo, error) {
r.headMu.Lock()
if h, ok := r.heads[rel]; ok && time.Since(h.at) < 30*time.Second {
r.headMu.Unlock()
return h, nil
}
r.headMu.Unlock()
req, _ := http.NewRequest("HEAD", r.cfg.Region+rel, nil)
resp, err := r.fast.Do(req)
if err != nil {
return headInfo{}, err
}
resp.Body.Close()
if resp.StatusCode != 200 {
return headInfo{}, fmt.Errorf("%s: %s", rel, resp.Status)
}
mod, _ := http.ParseTime(resp.Header.Get("Last-Modified"))
h := headInfo{size: resp.ContentLength, mod: mod, at: time.Now()}
r.headMu.Lock()
r.heads[rel] = h
r.headMu.Unlock()
return h, nil
}
// fresh says whether the cached copy matches the region's file. OS images live
// under versioned paths and never change; the rest is checked by size and time.
func (r *rackd) fresh(rel string, fi os.FileInfo) bool {
if (strings.HasPrefix(rel, "/os/") && !strings.Contains(rel, "/current/")) || strings.HasPrefix(rel, "/firmware/") {
return true // packages live under their id and never change
}
h, err := r.head(rel)
if err != nil {
return true // region unreachable: serve what we have
}
return h.size == fi.Size() && h.mod.Unix() == fi.ModTime().Unix()
}
func (r *rackd) localPath(rel string) (string, bool) {
clean := path.Clean("/" + rel)
if clean == "/" || strings.Contains(clean, "..") {
return "", false
}
return filepath.Join(r.cfg.CacheDir, filepath.FromSlash(clean)), true
}
// fill downloads rel from the region into the cache, copying to w on the way when given.
func (r *rackd) fill(rel string, w http.ResponseWriter) error {
local, ok := r.localPath(rel)
if !ok {
return errors.New("bad path")
}
resp, err := r.big.Get(r.cfg.Region + rel) // no timeout: images are large
if err != nil {
return err
}
defer resp.Body.Close()
if resp.StatusCode != 200 {
if w != nil {
w.WriteHeader(resp.StatusCode)
}
return fmt.Errorf("%s: %s", rel, resp.Status)
}
if err := os.MkdirAll(filepath.Dir(local), 0o755); err != nil {
return err
}
tmp, err := os.CreateTemp(filepath.Dir(local), ".fill-*")
if err != nil {
return err
}
defer os.Remove(tmp.Name())
var dst io.Writer = tmp
if w != nil {
for _, h := range []string{"Content-Length", "Content-Type", "Last-Modified"} {
if v := resp.Header.Get(h); v != "" {
w.Header().Set(h, v)
}
}
w.WriteHeader(200)
dst = io.MultiWriter(tmp, w)
}
n, err := io.Copy(dst, resp.Body)
if cerr := tmp.Close(); err == nil {
err = cerr
}
if err != nil {
return err
}
if resp.ContentLength >= 0 && n != resp.ContentLength {
return fmt.Errorf("%s: short read %d/%d", rel, n, resp.ContentLength)
}
if mod, err := http.ParseTime(resp.Header.Get("Last-Modified")); err == nil {
_ = os.Chtimes(tmp.Name(), mod, mod)
}
_ = os.Chmod(tmp.Name(), 0o644)
if err := os.Rename(tmp.Name(), local); err != nil {
return err
}
log.Printf("cached %s (%s)", rel, human(float64(n)))
return nil
}
func (r *rackd) serveCached(w http.ResponseWriter, req *http.Request) {
rel := path.Clean(req.URL.Path)
local, ok := r.localPath(rel)
if !ok {
http.NotFound(w, req)
return
}
if fi, err := os.Stat(local); err == nil && !fi.IsDir() && r.fresh(rel, fi) {
http.ServeFile(w, req, local)
return
}
if req.Method == http.MethodHead {
h, err := r.head(rel)
if err != nil {
http.NotFound(w, req)
return
}
w.Header().Set("Content-Length", strconv.FormatInt(h.size, 10))
w.Header().Set("Last-Modified", h.mod.UTC().Format(http.TimeFormat))
return
}
if err := r.fill(rel, w); err != nil {
log.Printf("cache %s: %v", rel, err)
}
}
// prefetch keeps kernels and initrds cached: TFTP cannot fetch on demand.
func (r *rackd) prefetch(files []string) {
r.fetch.Lock()
defer r.fetch.Unlock()
for _, f := range files {
rel := "/" + strings.TrimPrefix(f, "/")
local, ok := r.localPath(rel)
if !ok {
continue
}
if fi, err := os.Stat(local); err == nil && r.fresh(rel, fi) {
continue
}
if err := r.fill(rel, nil); err != nil {
log.Printf("prefetch %s: %v", rel, err)
}
}
}
// ---------------------------------------------------------------- HTTPS for agents
func (c RackConfig) serverCertPath() string {
return filepath.Join(orDefault(c.CertStateDir, "/var/lib/rollcall/rack-cert"), "server.pem")
}
// serverCert keeps a server certificate for the URL's host, from the region.
func (r *rackd) serverCert() (*tls.Certificate, error) {
host, err := rackHost(r.cfg.URL)
if err != nil {
return nil, err
}
p := r.cfg.serverCertPath()
if b, err := os.ReadFile(p); err == nil {
if pair, err := tls.X509KeyPair(b, b); err == nil {
leaf, _ := x509.ParseCertificate(pair.Certificate[0])
if leaf != nil && time.Until(leaf.NotAfter) > renewBefore && leaf.VerifyHostname(host) == nil {
pair.Leaf = leaf
return &pair, nil
}
}
}
return r.newServerCert()
}
// newServerCert asks the region for a server certificate and keeps it.
func (r *rackd) newServerCert() (*tls.Certificate, error) {
p := r.cfg.serverCertPath()
key, csr, err := newCSR(rackServerCN + r.cfg.Name)
if err != nil {
return nil, err
}
var resp struct {
Cert string `json:"cert"`
}
if _, err := r.call(r.fast, "POST", "/api/v1/rack/server-cert", enrollRequest{CSR: csr}, &resp); err != nil {
return nil, err
}
pair, err := tls.X509KeyPair([]byte(resp.Cert), key)
if err != nil {
return nil, err
}
pair.Leaf, _ = x509.ParseCertificate(pair.Certificate[0])
if err := writeAtomic(p, append([]byte(resp.Cert), key...), 0o600); err != nil {
log.Printf("keep the server certificate: %v (it is used, but asked for again at the next start)", err)
}
return &pair, nil
}
// serveTLS serves the agents' endpoints over HTTPS once the region issued a
// certificate, re-registers with the HTTPS URL and renews it when it ages.
func (r *rackd) serveTLS(h http.Handler) {
host, err := rackHost(r.cfg.URL)
if err != nil {
log.Printf("https for agents: %v", err)
return
}
addr := r.cfg.TLSListen
if addr == "" {
addr = net.JoinHostPort(host, "443")
}
for {
c, err := r.serverCert()
if err == nil {
r.srvMu.Lock()
r.srvCert = c
r.srvMu.Unlock()
break
}
log.Printf("https for agents: no server certificate yet: %v", err)
time.Sleep(30 * time.Second)
}
srv := &http.Server{Addr: addr, Handler: h, ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 2 * time.Minute, IdleTimeout: 2 * time.Minute,
TLSConfig: &tls.Config{MinVersion: tls.VersionTLS12, GetCertificate: func(*tls.ClientHelloInfo) (*tls.Certificate, error) {
r.srvMu.Lock()
defer r.srvMu.Unlock()
return r.srvCert, nil
}}}
ln, err := net.Listen("tcp", addr)
if err != nil {
log.Printf("https for agents on %s: %v", addr, err)
return
}
_, port, _ := net.SplitHostPort(addr)
u := "https://" + hostPort(host, "")
if port != "443" {
u = "https://" + hostPort(host, port)
}
r.srvMu.Lock()
r.tlsURL = u
r.srvMu.Unlock()
if err := r.register(); err != nil {
log.Printf("register with %s: %v", u, err)
}
log.Printf("serving agents over HTTPS on %s (%s)", addr, u)
go func() {
for range time.Tick(12 * time.Hour) {
r.srvMu.Lock()
leaf := r.srvCert.Leaf
r.srvMu.Unlock()
if leaf != nil && time.Until(leaf.NotAfter) < renewBefore {
if c, err := r.serverCert(); err == nil {
r.srvMu.Lock()
r.srvCert = c
r.srvMu.Unlock()
}
}
}
}()
log.Printf("https for agents: %v", srv.ServeTLS(ln, "", ""))
}