fleet-agent: on-host TPM-mTLS telemetry client (v1)
Build the on-host agent that pairs with fleet-hostapi and feeds the fleet-manager control plane (TODO #14 "STILL TO BUILD (a)"). Nothing on the workstation talked to the manager before this; Monitor + InfluxDB stayed empty. Scope: telemetry only. The agent reuses the host TPM identity for mTLS (read-only tpmkey.LoadSigner at 0x81000100 + cert from NV 0x01800100; the private key never leaves the TPM), opens the long-lived FleetAgent.Session bidi stream, sends Hello + periodic HealthSnapshot (load/cpu/temp/mem/tmpfs/nvme/auth/selinux/chrony/kiosk, probed read-only from /proc+/sys plus hostapi facts), and LOGS but does not act on the DesiredState/Command it receives. Reconcile actuators (kiosk URL, reboot, drain) are deferred until the telemetry half is bench-verified. - files/agent/: new module (go-tpm + grpc), static CGO_ENABLED=0 like enrol/hostapi. Copies the enrol tpmkey (+ read-only LoadSigner) and the generated fleet.agent.v1 proto so it stays self-contained without importing the initramfs enrol module. - files/agent/simtest/: proves the TPM-signed cert completes a real mTLS client-auth handshake (RequireAndVerifyClientCert) against the go-tpm software simulator. - files/etc/fleet-agent/agent.env: fleet-generic config (manager DNS name, fleet CA reused from radius-ca.pem, TPM handle/nvindex, interval). No per-host state. - fleet-agent.service: hardened unit (tss group for /dev/tpmrm0, INET egress). - CI builds fleet-agent out-of-VM + runs vet/test + the simulator handshake test. - Packer: enable_agent flag (default on) + 2b-agent provisioner + fapolicyd trust. Post-switch-root only, so no dracut/initramfs changes. - docs: TODO #14 + fleet-management-design as-built updated. Requires the host already enrolled (broker approve births the inventory row) — else the manager returns PermissionDenied, the correct "not enrolled yet" signal.
This commit is contained in:
Audiobuzz
@@ -0,0 +1,245 @@
|
||||
// fleet-agent — the on-host steady-state control-plane client (telemetry v1).
|
||||
//
|
||||
// Runs in the booted system as a systemd service. It dials the fleet-manager over
|
||||
// mTLS using the host's reused TPM identity (private key stays in the TPM), opens
|
||||
// the long-lived FleetAgent.Session bidi stream, sends a Hello then periodic
|
||||
// HealthSnapshots, and — in this telemetry-only build — LOGS but does NOT act on
|
||||
// the DesiredState/Command messages the manager pushes back. The receive path is
|
||||
// wired now so the reconcile actuators (kiosk URL, reboot, drain) can be added in a
|
||||
// follow-up without a protocol change.
|
||||
//
|
||||
// No inbound host port: the host always dials out (works behind the OPNsense L3
|
||||
// hop from the authenticated host VLAN to the mgmt plane). A dropped stream is
|
||||
// retried with jittered backoff; the manager treats stream loss as "offline".
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"math/rand"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
fleetpb "fleet-agent/proto"
|
||||
|
||||
"github.com/google/go-tpm/tpm2"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/credentials"
|
||||
"google.golang.org/grpc/keepalive"
|
||||
)
|
||||
|
||||
type config struct {
|
||||
manager string // manager host:port (DNS name — fleet-generic, invariant #1 safe)
|
||||
serverName string // TLS SNI / hostname to verify (defaults to manager host)
|
||||
caFile string // fleet CA bundle: verifies the manager's server cert
|
||||
tpmPath string
|
||||
handle tpm2.TPMHandle
|
||||
nvIndex tpm2.TPMHandle
|
||||
interval time.Duration
|
||||
hostapiURL string
|
||||
kioskEnv string // path to /etc/kiosk/kiosk.env (source of KIOSK_URL)
|
||||
}
|
||||
|
||||
func main() {
|
||||
log.SetFlags(0)
|
||||
log.SetPrefix("fleet-agent: ")
|
||||
cfg := loadConfig()
|
||||
|
||||
id, err := loadTPMIdentity(cfg.tpmPath, cfg.handle, cfg.nvIndex)
|
||||
if err != nil {
|
||||
log.Fatalf("TPM identity: %v", err)
|
||||
}
|
||||
defer id.Close()
|
||||
log.Printf("identity: cert serial %x (CN=%q) — dialing %s",
|
||||
id.cert.Leaf.SerialNumber, id.cert.Leaf.Subject.CommonName, cfg.manager)
|
||||
|
||||
tlsCfg, err := clientTLS(id, cfg.caFile, cfg.serverName)
|
||||
if err != nil {
|
||||
log.Fatalf("client TLS: %v", err)
|
||||
}
|
||||
cc, err := grpc.NewClient(cfg.manager,
|
||||
grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)),
|
||||
grpc.WithKeepaliveParams(keepalive.ClientParameters{
|
||||
Time: 20 * time.Second, // >= server MinTime (10s)
|
||||
Timeout: 10 * time.Second,
|
||||
PermitWithoutStream: true,
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatalf("grpc client: %v", err)
|
||||
}
|
||||
defer cc.Close()
|
||||
client := fleetpb.NewFleetAgentClient(cc)
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
p := &prober{hostapiURL: cfg.hostapiURL}
|
||||
backoff := time.Second
|
||||
for ctx.Err() == nil {
|
||||
start := time.Now()
|
||||
if err := runSession(ctx, client, cfg, p); err != nil && ctx.Err() == nil {
|
||||
log.Printf("session ended: %v", err)
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
break
|
||||
}
|
||||
// Reset backoff after a session that stayed up a while; otherwise grow it
|
||||
// (capped) with jitter to avoid reconnect storms across the fleet.
|
||||
if time.Since(start) > 30*time.Second {
|
||||
backoff = time.Second
|
||||
}
|
||||
sleep := min(backoff+time.Duration(rand.Int63n(int64(backoff+1))), 30*time.Second)
|
||||
log.Printf("reconnecting in %s", sleep.Round(time.Millisecond))
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(sleep):
|
||||
}
|
||||
if backoff < 15*time.Second {
|
||||
backoff *= 2
|
||||
}
|
||||
}
|
||||
log.Printf("shutting down")
|
||||
}
|
||||
|
||||
// runSession opens one Session stream, sends Hello + periodic snapshots, and logs
|
||||
// any ManagerDown. It returns when the stream fails or ctx is cancelled.
|
||||
func runSession(ctx context.Context, client fleetpb.FleetAgentClient, cfg config, p *prober) error {
|
||||
sctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
stream, err := client.Session(sctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Hello first — what we're running, so the manager reconciles version drift.
|
||||
facts := fetchFacts(sctx, cfg.hostapiURL)
|
||||
if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Hello{Hello: buildHello(facts, readKioskURL(cfg.kioskEnv))}}); err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("connected to %s — streaming health every %s", cfg.manager, cfg.interval)
|
||||
|
||||
// Receive loop: v1 LOGS desired-state/commands, does not act on them.
|
||||
go func() {
|
||||
for {
|
||||
down, err := stream.Recv()
|
||||
if err != nil {
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
logDown(down)
|
||||
}
|
||||
}()
|
||||
|
||||
// Send one snapshot immediately, then on the interval.
|
||||
t := time.NewTicker(cfg.interval)
|
||||
defer t.Stop()
|
||||
if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Snapshot{Snapshot: p.snapshot(sctx)}}); err != nil {
|
||||
return err
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-sctx.Done():
|
||||
return sctx.Err()
|
||||
case <-t.C:
|
||||
if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Snapshot{Snapshot: p.snapshot(sctx)}}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// logDown records a manager->host message. TELEMETRY-ONLY v1: received, logged,
|
||||
// NOT applied. The reconcile actuators land in a follow-up (see docs/TODO.md #14).
|
||||
func logDown(down *fleetpb.ManagerDown) {
|
||||
switch m := down.GetMsg().(type) {
|
||||
case *fleetpb.ManagerDown_Desired:
|
||||
d := m.Desired
|
||||
log.Printf("desired-state (NOT applied — telemetry-only): target_version=%q kiosk_url=%q reboot_at=%d drain=%v",
|
||||
d.GetTargetVersion(), d.GetKioskUrl(), d.GetRebootAtUnix(), d.GetDrain())
|
||||
case *fleetpb.ManagerDown_Command:
|
||||
log.Printf("command (NOT applied — telemetry-only): id=%s kind=%s", m.Command.GetCommandId(), m.Command.GetKind())
|
||||
}
|
||||
}
|
||||
|
||||
// ---- config ----------------------------------------------------------------
|
||||
|
||||
func loadConfig() config {
|
||||
manager := env("FLEET_AGENT_MANAGER", "diskless-server.bitasylum.org:9600")
|
||||
cfg := config{
|
||||
manager: manager,
|
||||
serverName: env("FLEET_AGENT_SERVER_NAME", hostOf(manager)),
|
||||
caFile: env("FLEET_AGENT_CA", "/etc/dot1x/radius-ca.pem"),
|
||||
tpmPath: env("FLEET_AGENT_TPM", "/dev/tpmrm0"),
|
||||
handle: mustHandle(env("FLEET_AGENT_HANDLE", "0x81000100")),
|
||||
nvIndex: mustHandle(env("FLEET_AGENT_NVINDEX", "0x01800100")),
|
||||
interval: envDuration("FLEET_AGENT_INTERVAL", 30*time.Second),
|
||||
hostapiURL: env("FLEET_AGENT_HOSTAPI", "http://127.0.0.1:9099/v1/host"),
|
||||
kioskEnv: env("FLEET_AGENT_KIOSK_ENV", "/etc/kiosk/kiosk.env"),
|
||||
}
|
||||
return cfg
|
||||
}
|
||||
|
||||
func env(k, def string) string {
|
||||
if v := os.Getenv(k); v != "" {
|
||||
return v
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
func envDuration(k string, def time.Duration) time.Duration {
|
||||
if v := os.Getenv(k); v != "" {
|
||||
if d, err := time.ParseDuration(v); err == nil {
|
||||
return d
|
||||
}
|
||||
log.Printf("bad %s=%q — using %s", k, v, def)
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// hostOf strips the :port from a host:port, leaving the DNS name for TLS
|
||||
// verification. Handles bare hosts and IPv6 in brackets conservatively.
|
||||
func hostOf(hostport string) string {
|
||||
if i := strings.LastIndex(hostport, ":"); i > strings.LastIndex(hostport, "]") {
|
||||
return strings.Trim(hostport[:i], "[]")
|
||||
}
|
||||
return strings.Trim(hostport, "[]")
|
||||
}
|
||||
|
||||
func mustHandle(s string) tpm2.TPMHandle {
|
||||
h, err := parseHandle(s)
|
||||
if err != nil {
|
||||
log.Fatalf("bad TPM handle/index %q: %v", s, err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// parseHandle parses a 0x-prefixed (or bare) hex TPM handle/NV index.
|
||||
func parseHandle(s string) (tpm2.TPMHandle, error) {
|
||||
v, err := strconv.ParseUint(strings.TrimPrefix(strings.ToLower(s), "0x"), 16, 32)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return tpm2.TPMHandle(v), nil
|
||||
}
|
||||
|
||||
// readKioskURL extracts KIOSK_URL from the kiosk env file (KEY=VALUE lines).
|
||||
// Best-effort: "" if the file is absent (the manager still gets a Hello).
|
||||
func readKioskURL(path string) string {
|
||||
b, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
for line := range strings.SplitSeq(string(b), "\n") {
|
||||
line = strings.TrimSpace(line)
|
||||
if v, ok := strings.CutPrefix(line, "KIOSK_URL="); ok {
|
||||
return strings.Trim(strings.TrimSpace(v), `"'`)
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
Reference in New Issue
Block a user