Sensors: per-service and container sensors, pattern-limited tokens, fleet query, steady sampler

- service.<unit>.active|state for services that are enabled, up or failed - containers.* and container.<name>.running|state|health|mem|cpu (cgroup reads, no docker stats) - sensors:read tokens can be limited to ID patterns, enforced on every endpoint and, through the hub's filter header, on managed machines; the header can only narrow - /api/fleet/sensors/query: one shaped query, every machine, one request - the sampler waits what's left of the interval, so 250 ms means 4 a second - Monitors asks for the families it draws instead of * - unpowered hwmon sentinels (-93 °C) are not temperatures Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-19 08:08:44 -07:00
parent 4ddf10be51
commit 4c2eb57ea6
11 changed files with 383 additions and 51 deletions
@@ -4,6 +4,7 @@ package api
 import (
 	"bytes"
 	"context"
+	"crypto/subtle"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -22,6 +23,7 @@ import (
 	"runtime"
 	"strconv"
 	"strings"
+	"sync"
 	"syscall"
 	"time"

@@ -123,6 +125,8 @@ func New(cfg *config.Config, sessions *auth.Sessions, mgr *term.Manager, col *me
 	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/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))
@@ -250,34 +254,81 @@ func (s *Server) auth(next http.HandlerFunc) http.HandlerFunc {
 // 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 c, err := r.Cookie(cookieName); err == nil && s.sessions.Valid(c.Value) {
-			next(w, r)
-			return
-		}
-		presented, fromURL := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer "), false
-		if presented == "" {
-			presented, fromURL = r.URL.Query().Get("token"), true
-		}
-		if presented != "" {
-			var hashes []string
-			s.cfg.View(func(c *config.Config) {
-				for _, t := range c.Tokens {
-					if t.Scope == config.ScopeSensorsRead || (t.Scope == "" && !fromURL) {
-						hashes = append(hashes, t.Hash)
+			ok, trusted = true, true
+		} else {
+			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 auth.TokenMatches(presented, hashes) {
-				next(w, r)
-				return
+				})
 			}
 		}
-		jsonErr(w, 401, "unauthenticated")
+		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)))
 	}
 }

+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) {
@@ -285,18 +336,33 @@ func (s *Server) proxyNodeSensors(w http.ResponseWriter, r *http.Request) {
 		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"`
+		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)
 	tok, hash, err := auth.NewToken()
 	if err != nil {
@@ -311,7 +377,7 @@ func (s *Server) issueReadToken(w http.ResponseWriter, r *http.Request) {
 				return
 			}
 		}
-		c.Tokens = append(c.Tokens, config.Token{Name: name, Hash: hash, Created: time.Now(), Scope: config.ScopeSensorsRead})
+		c.Tokens = append(c.Tokens, config.Token{Name: name, Hash: hash, Created: time.Now(), Scope: config.ScopeSensorsRead, Sensors: pats})
 	}); err != nil {
 		jsonErr(w, 500, err.Error())
 		return
@@ -374,7 +440,7 @@ 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})
+			out = append(out, map[string]any{"name": t.Name, "created": t.Created, "scope": t.Scope, "sensors": t.Sensors})
 		}
 	})
 	if out == nil {
@@ -656,7 +722,7 @@ func (s *Server) me(w http.ResponseWriter, _ *http.Request) {
 // 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.sensors.All()
+	all := s.visible(r)
 	if ids := r.URL.Query().Get("ids"); ids != "" {
 		picked := []sensors.Sensor{}
 		for _, sel := range strings.Split(ids, ",") {
@@ -675,7 +741,7 @@ func (s *Server) sensorCatalog(w http.ResponseWriter, r *http.Request) {
 // 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.sensors.All(), r.PathValue("id"))
+	got, _ := sensors.Select(s.visible(r), r.PathValue("id"))
 	if len(got) != 1 {
 		jsonErr(w, 404, "no such sensor")
 		return
@@ -690,6 +756,9 @@ func (s *Server) sensorOne(w http.ResponseWriter, r *http.Request) {

 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
@@ -704,13 +773,17 @@ func (s *Server) sensorHistories(w http.ResponseWriter, r *http.Request) {
 		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, _ *http.Request) {
+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.sensors.All())
+	sensors.WriteProm(w, s.visible(r))
 }

 // queryOf reads a shaped query from a request: the POST body, or on a GET either ?q= (the
@@ -739,7 +812,7 @@ func queryOf(w http.ResponseWriter, r *http.Request) (q json.RawMessage, full bo
 	return q, full, nil
 }

-var reservedParam = map[string]bool{"full": true, "token": true, "q": true, "every": true}
+var reservedParam = map[string]bool{"full": true, "token": true, "q": true, "every": true, "machines": true}

 // 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
@@ -747,7 +820,7 @@ var reservedParam = map[string]bool{"full": true, "token": true, "q": true, "eve
 func (s *Server) sensorStream(w http.ResponseWriter, r *http.Request) {
 	q, full, err := queryOf(w, r)
 	if err == nil {
-		_, err = sensors.Shape(s.sensors.All(), q, full) // refuse a bad query before the stream starts
+		_, err = sensors.Shape(s.visible(r), q, full) // refuse a bad query before the stream starts
 	}
 	if err != nil {
 		jsonErr(w, 400, err.Error())
@@ -774,7 +847,7 @@ func (s *Server) sensorStream(w http.ResponseWriter, r *http.Request) {
 	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.sensors.All(), q, full)
+		out, _ := sensors.Shape(s.visible(r), q, full)
 		b, err := json.Marshal(out)
 		if err != nil {
 			return false
@@ -821,7 +894,7 @@ func (s *Server) sensorQuery(w http.ResponseWriter, r *http.Request) {
 		jsonErr(w, 413, err.Error())
 		return
 	}
-	out, err := sensors.Shape(s.sensors.All(), q, full)
+	out, err := sensors.Shape(s.visible(r), q, full)
 	if err != nil {
 		jsonErr(w, 400, err.Error())
 		return
@@ -829,6 +902,81 @@ func (s *Server) sensorQuery(w http.ResponseWriter, r *http.Request) {
 	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.host}, 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.host {
+			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) {