diff --git a/serving/fleet-manager/README.md b/serving/fleet-manager/README.md index c0c0db3..ddb4d07 100644 --- a/serving/fleet-manager/README.md +++ b/serving/fleet-manager/README.md @@ -66,7 +66,12 @@ inspectable, reproducible; a DB swap is a drop-in behind the `Store` interface. | `GET /api/enrol/pending`, `POST /api/enrol/{id}/approve\|deny` | ⏳ ckpt 3 (drives the broker) | | `GET /api/hosts/{serial}/metrics` | ⏳ ckpt 4 (InfluxDB/Flux) | -Agent plane (gRPC `FleetAgent.Session` bidi) — ⏳ ckpt 2. +**Agent plane** — gRPC `FleetAgent.Session` bidi over mTLS (fleet CA): ✅ ckpt 2. +`Hello`/`HealthSnapshot`/`CommandResult` up; `DesiredState`/`Command` down. +Identity is the peer cert serial (never self-asserted); an unenrolled serial is +refused. Desired-state is pushed live on connect + on change, and is durable in +inventory for offline delivery. Metrics writing is behind a `MetricsSink` +(no-op now; InfluxDB in ckpt 4). ## Config (env) @@ -76,11 +81,14 @@ Agent plane (gRPC `FleetAgent.Session` bidi) — ⏳ ckpt 2. | `FLEET_MGR_ADMIN_CERT` / `_KEY` | — | admin server cert/key (mTLS) | | `FLEET_MGR_ADMIN_CLIENT_CA` | — | **dedicated admin CA** (NOT the fleet CA) | | `FLEET_MGR_ADMIN_INSECURE` | `false` | plaintext, no client-cert — **dev only** | +| `FLEET_MGR_AGENT_ADDR` | `127.0.0.1:9600` | gRPC agent listener | +| `FLEET_MGR_AGENT_CERT` / `_KEY` | — | agent-plane server cert/key | +| `FLEET_MGR_AGENT_CLIENT_CA` | — | **fleet CA** (verifies host TPM certs). Agent plane starts only when all three are set. | | `FLEET_MGR_INVENTORY` | `/var/lib/fleet-manager/inventory.json` | inventory file (mount a volume) | | `FLEET_MGR_SEED` | `false` | inject demo hosts so the SPA has data — dev only | | `HOSTNQN_BASE` | `nqn.2026-06.org.example.fleet:host-` | hostnqn prefix (match the broker) | -Agent-plane + InfluxDB env land with their checkpoints. +InfluxDB env lands in checkpoint 4. ## Dev run @@ -95,5 +103,8 @@ curl -s localhost:9500/api/hosts | jq '.hosts[] | {name:.identity.starship_name, - **Checkpoint 1 (done):** Go skeleton, JSON-file inventory `Store`, admin JSON API (monitor + lifecycle), admin mTLS wiring, dev seed. Stdlib-only. -- Next: gRPC agent plane (ckpt 2) → authorise/broker fold-in (ckpt 3) → InfluxDB - read API (ckpt 4) → themed SPA (ckpt 5) → Docker/compose + PKI + docs (ckpt 6). +- **Checkpoint 2 (done):** gRPC agent plane — proto contract, mTLS (fleet CA) + server, stream registry, identity-from-cert, snapshot→inventory, live + desired-state push, `MetricsSink` seam. Full mTLS integration test. +- Next: authorise/broker fold-in (ckpt 3) → InfluxDB read/write (ckpt 4) → themed + SPA (ckpt 5) → Docker/compose + PKI + docs (ckpt 6). diff --git a/serving/fleet-manager/admin.go b/serving/fleet-manager/admin.go index 423c9e8..5056859 100644 --- a/serving/fleet-manager/admin.go +++ b/serving/fleet-manager/admin.go @@ -65,9 +65,14 @@ func (s *server) handleSetDesired(w http.ResponseWriter, r *http.Request) { writeErr(w, http.StatusNotFound, "no such host") return } - // checkpoint 2: after persisting, push the new desired-state down the host's - // live agent stream here (if connected); offline hosts get it on reconnect. - writeJSON(w, http.StatusOK, h) + // Push the new desired-state down the host's live agent stream if connected; + // an offline host (or one we can't reach right now) picks it up from the + // persisted inventory record on its next connect — inventory is authoritative. + pushed := false + if s.hub != nil { + pushed = s.hub.pushDesired(serial, h.Desired) + } + writeJSON(w, http.StatusOK, map[string]any{"host": h, "pushed_live": pushed}) } // adminCN returns the CN of the admin's mTLS client cert, for audit attribution. diff --git a/serving/fleet-manager/agent.go b/serving/fleet-manager/agent.go new file mode 100644 index 0000000..b4975b4 --- /dev/null +++ b/serving/fleet-manager/agent.go @@ -0,0 +1,316 @@ +// agent.go — the gRPC agent plane: the long-lived bidi stream each host holds. +// +// mTLS is verified against the FLEET CA (the host's reused TPM identity cert), +// distinct from the admin plane's dedicated admin CA. The host's cert SERIAL is +// the inventory key, taken from the peer cert — never self-asserted — so a host +// can only report/receive its own state. +// +// Concurrency model per Session: one receive loop (host->manager) in the handler +// goroutine + one send loop draining a per-connection channel (manager->host). +// The channel decouples admin-initiated pushes (SetDesired, Command) from the +// stream's Send, which must not be called concurrently. Inventory is the source +// of truth; the live stream is only the low-latency delivery path — a host that +// is offline or whose send buffer is full simply gets the current desired-state +// on its next (re)connect. +package main + +import ( + "context" + "crypto/tls" + "crypto/x509" + "fmt" + "log" + "net" + "os" + "sync" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/keepalive" + "google.golang.org/grpc/peer" + "google.golang.org/grpc/status" + + fleetpb "fleet-manager/proto" +) + +// MetricsSink receives each health snapshot for time-series storage. The no-op +// sink is the default; the InfluxDB sink lands in checkpoint 4. +type MetricsSink interface { + WriteSnapshot(serial, hostnqn, starship string, s *fleetpb.HealthSnapshot, at time.Time) +} + +type noopSink struct{} + +func (noopSink) WriteSnapshot(string, string, string, *fleetpb.HealthSnapshot, time.Time) {} + +// agentConn is one live host stream's downstream (manager->host) side. +type agentConn struct { + serial string + send chan *fleetpb.ManagerDown + done chan struct{} + once sync.Once +} + +func (c *agentConn) close() { c.once.Do(func() { close(c.done) }) } + +// agentHub implements the FleetAgent gRPC service and holds the live-stream +// registry that admin actions push through. +type agentHub struct { + fleetpb.UnimplementedFleetAgentServer + store Store + sink MetricsSink + + mu sync.Mutex + conns map[string]*agentConn // serial -> live connection +} + +func newAgentHub(store Store, sink MetricsSink) *agentHub { + if sink == nil { + sink = noopSink{} + } + return &agentHub{store: store, sink: sink, conns: map[string]*agentConn{}} +} + +// register installs a connection, evicting any prior stream for the same serial +// (a reconnect supersedes a lingering old one). +func (h *agentHub) register(c *agentConn) { + h.mu.Lock() + if old := h.conns[c.serial]; old != nil { + old.close() + } + h.conns[c.serial] = c + h.mu.Unlock() +} + +func (h *agentHub) deregister(c *agentConn) { + h.mu.Lock() + if h.conns[c.serial] == c { // don't clobber a newer reconnect + delete(h.conns, c.serial) + } + h.mu.Unlock() + c.close() +} + +// pushDown delivers a message to a host's live stream if connected. Returns false +// if the host is offline or its buffer is backed up (it will reconcile on its +// next connect from persisted desired-state — no message is "lost" that matters). +func (h *agentHub) pushDown(serial string, m *fleetpb.ManagerDown) bool { + h.mu.Lock() + c := h.conns[serial] + h.mu.Unlock() + if c == nil { + return false + } + select { + case c.send <- m: + return true + case <-c.done: + return false + default: + return false // slow consumer; drop — desired-state is durable in inventory + } +} + +// pushDesired sends a host its current desired-state over the live stream. +func (h *agentHub) pushDesired(serial string, d Desired) bool { + return h.pushDown(serial, desiredMsg(d)) +} + +// Session is the bidirectional stream RPC. +func (h *agentHub) Session(stream fleetpb.FleetAgent_SessionServer) error { + serial, err := peerSerial(stream.Context()) + if err != nil { + return status.Error(codes.Unauthenticated, err.Error()) + } + // A host must be enrolled (inventory record exists) before it can stream. + host, ok := h.store.Get(serial) + if !ok { + return status.Errorf(codes.PermissionDenied, "host %s is not enrolled", serial) + } + + c := &agentConn{serial: serial, send: make(chan *fleetpb.ManagerDown, 16), done: make(chan struct{})} + h.register(c) + h.store.SetOnline(serial, true, time.Now().UTC()) + log.Printf("agent connected: %s (%s)", serial, host.Identity.StarshipName) + defer func() { + h.deregister(c) + h.store.SetOnline(serial, false, time.Now().UTC()) + log.Printf("agent disconnected: %s", serial) + }() + + // Deliver current desired-state immediately on connect. + c.send <- desiredMsg(host.Desired) + + // Sender goroutine: the ONLY caller of stream.Send. + sendErr := make(chan error, 1) + go func() { + for { + select { + case m := <-c.send: + if err := stream.Send(m); err != nil { + sendErr <- err + return + } + case <-c.done: + return + } + } + }() + + // Receive loop (this goroutine). + for { + select { + case err := <-sendErr: + return err + default: + } + up, err := stream.Recv() + if err != nil { + return err // EOF / transport error -> disconnect (defer cleans up) + } + h.handleUp(serial, up) + } +} + +func (h *agentHub) handleUp(serial string, up *fleetpb.AgentUp) { + switch m := up.GetMsg().(type) { + case *fleetpb.AgentUp_Hello: + hello := m.Hello + host, ok := h.store.Get(serial) + if !ok { + return + } + _, _, _ = h.store.ReportState(serial, + Version{ImageVersion: hello.GetImageVersion(), RootfsDigest: hello.GetRootfsDigest(), Kernel: hello.GetKernel()}, + host.Health, // unchanged until first snapshot + Usage{KioskURL: hello.GetKioskUrl(), Display: host.Usage.Display}, + time.Now().UTC()) + case *fleetpb.AgentUp_Snapshot: + snap := m.Snapshot + at := time.Now().UTC() + if snap.GetAtUnix() > 0 { + at = time.Unix(snap.GetAtUnix(), 0).UTC() + } + host, ok := h.store.Get(serial) + if !ok { + return + } + h.store.ReportState(serial, host.Version, snapshotToHealth(snap), snapshotToUsage(snap, host.Usage), at) + h.sink.WriteSnapshot(serial, host.Identity.Hostnqn, host.Identity.StarshipName, snap, at) + case *fleetpb.AgentUp_Result: + res := m.Result + log.Printf("agent %s command %s result: ok=%v %s", serial, res.GetCommandId(), res.GetOk(), res.GetDetail()) + } +} + +// ---- conversions ---------------------------------------------------------- + +func desiredMsg(d Desired) *fleetpb.ManagerDown { + var rebootAt int64 + if d.RebootAt != nil { + rebootAt = d.RebootAt.Unix() + } + return &fleetpb.ManagerDown{Msg: &fleetpb.ManagerDown_Desired{Desired: &fleetpb.DesiredState{ + TargetVersion: d.TargetVersion, + KioskUrl: d.KioskURL, + RebootAtUnix: rebootAt, + Drain: d.Drain, + }}} +} + +func snapshotToHealth(s *fleetpb.HealthSnapshot) Health { + return Health{ + LoadAvg: s.GetLoadAvg(), + CPUPct: s.GetCpuPct(), + TempC: s.GetTempC(), + MemUsedPct: s.GetMemUsedPct(), + TmpfsUsedPct: s.GetTmpfsUsedPct(), + NVMeRoot: s.GetNvmeRoot(), + Auth: s.GetAuth(), + SELinux: s.GetSelinux(), + Chrony: s.GetChrony(), + KioskActive: s.GetKioskActive(), + } +} + +func snapshotToUsage(s *fleetpb.HealthSnapshot, prev Usage) Usage { + u := prev + u.SessionUptime = s.GetSessionUptime() + u.ChromiumCrashes = int(s.GetChromiumCrashes()) + if s.GetDisplay() != "" { + u.Display = s.GetDisplay() + } + return u +} + +// ---- mTLS identity -------------------------------------------------------- + +// peerSerial extracts the lowercase-hex serial of the client's mTLS leaf cert — +// the inventory key. It is trusted because RequireAndVerifyClientCert already +// verified the chain to the fleet CA. +func peerSerial(ctx context.Context) (string, error) { + p, ok := peer.FromContext(ctx) + if !ok { + return "", fmt.Errorf("no peer in context") + } + tlsInfo, ok := p.AuthInfo.(credentials.TLSInfo) + if !ok || len(tlsInfo.State.PeerCertificates) == 0 { + return "", fmt.Errorf("no client certificate") + } + return serialHex(tlsInfo.State.PeerCertificates[0]), nil +} + +// serialHex renders a cert serial as lowercase hex — the same form the broker +// stores (store.go) so the value matches the inventory key. +func serialHex(cert *x509.Certificate) string { + return fmt.Sprintf("%x", cert.SerialNumber) +} + +// ---- gRPC server wiring --------------------------------------------------- + +// agentServer builds the configured gRPC server (mTLS against the fleet CA + +// keepalive) with the hub registered. Split from the listen/serve so tests can +// drive it on an ephemeral listener. +func agentServer(cfg config, hub *agentHub) (*grpc.Server, error) { + cert, err := tls.LoadX509KeyPair(cfg.agentCert, cfg.agentKey) + if err != nil { + return nil, fmt.Errorf("agent server cert: %w", err) + } + caPEM, err := os.ReadFile(cfg.agentClientCA) + if err != nil { + return nil, fmt.Errorf("fleet CA: %w", err) + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(caPEM) { + return nil, fmt.Errorf("no certificates parsed from FLEET_MGR_AGENT_CLIENT_CA") + } + creds := credentials.NewTLS(&tls.Config{ + MinVersion: tls.VersionTLS12, + Certificates: []tls.Certificate{cert}, + ClientAuth: tls.RequireAndVerifyClientCert, + ClientCAs: pool, + }) + gs := grpc.NewServer( + grpc.Creds(creds), + grpc.KeepaliveParams(keepalive.ServerParameters{Time: 30 * time.Second, Timeout: 10 * time.Second}), + grpc.KeepaliveEnforcementPolicy(keepalive.EnforcementPolicy{MinTime: 10 * time.Second, PermitWithoutStream: true}), + ) + fleetpb.RegisterFleetAgentServer(gs, hub) + return gs, nil +} + +// serveAgent starts the gRPC agent listener with mTLS against the fleet CA. +func serveAgent(cfg config, hub *agentHub) error { + gs, err := agentServer(cfg, hub) + if err != nil { + return err + } + lis, err := net.Listen("tcp", cfg.agentAddr) + if err != nil { + return err + } + log.Printf("fleet-manager agent (gRPC/mTLS) listening on %s", cfg.agentAddr) + return gs.Serve(lis) +} diff --git a/serving/fleet-manager/agent_test.go b/serving/fleet-manager/agent_test.go new file mode 100644 index 0000000..e3b928c --- /dev/null +++ b/serving/fleet-manager/agent_test.go @@ -0,0 +1,243 @@ +package main + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "net" + "os" + "path/filepath" + "testing" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials" + + fleetpb "fleet-manager/proto" +) + +// TestAgentSession drives the full agent plane over real mTLS: a host dials in +// with a fleet-CA client cert, the manager derives its serial from that cert, +// pushes the current desired-state on connect, ingests a Hello + HealthSnapshot +// into inventory, and delivers a live desired-state change over the same stream. +func TestAgentSession(t *testing.T) { + dir := t.TempDir() + caCert, caKey := makeCA(t) + caPath := writeFile(t, dir, "ca.crt", pemCert(caCert.Raw)) + + srvCertPEM, srvKeyPEM := makeLeaf(t, caCert, caKey, "fleet-manager", big.NewInt(1), true) + srvCrt := writeFile(t, dir, "srv.crt", srvCertPEM) + srvKey := writeFile(t, dir, "srv.key", srvKeyPEM) + + const serialHexStr = "1234abcd" + cliCertPEM, cliKeyPEM := makeLeaf(t, caCert, caKey, "fleet-host", big.NewInt(0x1234abcd), false) + cliCert, err := tls.X509KeyPair(cliCertPEM, cliKeyPEM) + if err != nil { + t.Fatalf("client keypair: %v", err) + } + + store, err := newJSONStore("") // in-memory + if err != nil { + t.Fatalf("store: %v", err) + } + if err := store.Upsert(&Host{ + Identity: Identity{CertSerial: serialHexStr, Hostnqn: "nqn.test:host-" + serialHexStr, StarshipName: "test-ship"}, + Desired: Desired{TargetVersion: "0.5.0", KioskURL: "http://k/"}, + }); err != nil { + t.Fatalf("seed: %v", err) + } + + hub := newAgentHub(store, nil) + gs, err := agentServer(config{agentCert: srvCrt, agentKey: srvKey, agentClientCA: caPath}, hub) + if err != nil { + t.Fatalf("agentServer: %v", err) + } + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + go gs.Serve(lis) + defer gs.Stop() + + // ---- host dials in ---- + caPool := x509.NewCertPool() + caPool.AddCert(caCert) + creds := credentials.NewTLS(&tls.Config{ + Certificates: []tls.Certificate{cliCert}, + RootCAs: caPool, + ServerName: "localhost", + }) + conn, err := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(creds)) + if err != nil { + t.Fatalf("dial: %v", err) + } + defer conn.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream, err := fleetpb.NewFleetAgentClient(conn).Session(ctx) + if err != nil { + t.Fatalf("session: %v", err) + } + + // 1. current desired-state arrives on connect. + down, err := stream.Recv() + if err != nil { + t.Fatalf("recv desired: %v", err) + } + if got := down.GetDesired().GetTargetVersion(); got != "0.5.0" { + t.Fatalf("initial desired target_version = %q, want 0.5.0", got) + } + + // 2. host reports Hello (running an older version) + a health snapshot. + if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Hello{Hello: &fleetpb.Hello{ImageVersion: "0.4.2", Kernel: "5.14.0"}}}); err != nil { + t.Fatalf("send hello: %v", err) + } + if err := stream.Send(&fleetpb.AgentUp{Msg: &fleetpb.AgentUp_Snapshot{Snapshot: &fleetpb.HealthSnapshot{ + TmpfsUsedPct: 63, NvmeRoot: "ok", Auth: "up", Selinux: "enforcing", KioskActive: true, + }}}); err != nil { + t.Fatalf("send snapshot: %v", err) + } + + // 3. inventory reflects the report + online presence. + waitFor(t, 3*time.Second, func() bool { + h, ok := store.Get(serialHexStr) + return ok && h.Online && h.Version.ImageVersion == "0.4.2" && h.Health.TmpfsUsedPct == 63 && h.Health.NVMeRoot == "ok" + }, "inventory did not reflect the agent report") + + // 4. an admin changes desired-state -> live push over the same stream. + if _, _, err := store.SetDesired(serialHexStr, Desired{TargetVersion: "0.6.0"}); err != nil { + t.Fatalf("set desired: %v", err) + } + if !hub.pushDesired(serialHexStr, Desired{TargetVersion: "0.6.0"}) { + t.Fatal("pushDesired reported the host as not live, but it is connected") + } + down2, err := stream.Recv() + if err != nil { + t.Fatalf("recv desired push: %v", err) + } + if got := down2.GetDesired().GetTargetVersion(); got != "0.6.0" { + t.Fatalf("pushed desired target_version = %q, want 0.6.0", got) + } +} + +// TestAgentRejectsUnenrolled: a valid fleet-CA cert whose serial has no inventory +// record must be refused (enrolment gates the stream). +func TestAgentRejectsUnenrolled(t *testing.T) { + dir := t.TempDir() + caCert, caKey := makeCA(t) + caPath := writeFile(t, dir, "ca.crt", pemCert(caCert.Raw)) + srvCertPEM, srvKeyPEM := makeLeaf(t, caCert, caKey, "fleet-manager", big.NewInt(1), true) + srvCrt := writeFile(t, dir, "srv.crt", srvCertPEM) + srvKey := writeFile(t, dir, "srv.key", srvKeyPEM) + cliCertPEM, cliKeyPEM := makeLeaf(t, caCert, caKey, "fleet-host", big.NewInt(0xdead), false) + cliCert, _ := tls.X509KeyPair(cliCertPEM, cliKeyPEM) + + store, _ := newJSONStore("") // empty — the host is NOT enrolled + hub := newAgentHub(store, nil) + gs, err := agentServer(config{agentCert: srvCrt, agentKey: srvKey, agentClientCA: caPath}, hub) + if err != nil { + t.Fatalf("agentServer: %v", err) + } + lis, _ := net.Listen("tcp", "127.0.0.1:0") + go gs.Serve(lis) + defer gs.Stop() + + caPool := x509.NewCertPool() + caPool.AddCert(caCert) + creds := credentials.NewTLS(&tls.Config{Certificates: []tls.Certificate{cliCert}, RootCAs: caPool, ServerName: "localhost"}) + conn, _ := grpc.NewClient(lis.Addr().String(), grpc.WithTransportCredentials(creds)) + defer conn.Close() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream, err := fleetpb.NewFleetAgentClient(conn).Session(ctx) + if err != nil { + t.Fatalf("session open: %v", err) + } + if _, err := stream.Recv(); err == nil { + t.Fatal("expected the stream to be rejected for an unenrolled host, got nil error") + } +} + +// ---- test helpers --------------------------------------------------------- + +func makeCA(t *testing.T) (*x509.Certificate, *ecdsa.PrivateKey) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("ca key: %v", err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "test-fleet-ca"}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + KeyUsage: x509.KeyUsageCertSign | x509.KeyUsageDigitalSignature, + BasicConstraintsValid: true, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatalf("ca cert: %v", err) + } + cert, _ := x509.ParseCertificate(der) + return cert, key +} + +func makeLeaf(t *testing.T, ca *x509.Certificate, caKey *ecdsa.PrivateKey, cn string, serial *big.Int, server bool) (certPEM, keyPEM []byte) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatalf("leaf key: %v", err) + } + tmpl := &x509.Certificate{ + SerialNumber: serial, + Subject: pkix.Name{CommonName: cn}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth}, + } + if server { + tmpl.ExtKeyUsage = append(tmpl.ExtKeyUsage, x509.ExtKeyUsageServerAuth) + tmpl.DNSNames = []string{"localhost"} + tmpl.IPAddresses = []net.IP{net.ParseIP("127.0.0.1")} + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, ca, &key.PublicKey, caKey) + if err != nil { + t.Fatalf("leaf cert: %v", err) + } + kb, _ := x509.MarshalECPrivateKey(key) + return pemCert(der), pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: kb}) +} + +func pemCert(der []byte) []byte { + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) +} + +func writeFile(t *testing.T, dir, name string, b []byte) string { + t.Helper() + p := filepath.Join(dir, name) + if err := os.WriteFile(p, b, 0o600); err != nil { + t.Fatalf("write %s: %v", name, err) + } + return p +} + +func waitFor(t *testing.T, d time.Duration, cond func() bool, msg string) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal(msg) +} diff --git a/serving/fleet-manager/go.mod b/serving/fleet-manager/go.mod index 96b3464..c50807c 100644 --- a/serving/fleet-manager/go.mod +++ b/serving/fleet-manager/go.mod @@ -7,9 +7,20 @@ // so stdlib-only), this is a SERVER-SIDE service that ships as its own container // on the mgmt network — external deps are acceptable here. They are added at the // build checkpoint that needs them: -// - google.golang.org/grpc + protobuf : the agent bidi stream (checkpoint 2) -// - github.com/influxdata/influxdb-client-go/v2 : metrics read/write (ckpt 2/4) -// Checkpoint 1 (this commit) is stdlib-only. +// - google.golang.org/grpc + protobuf : the agent bidi stream (added ckpt 2) +// - github.com/influxdata/influxdb-client-go/v2 : metrics read/write (ckpt 4) module fleet-manager -go 1.22 +go 1.25.0 + +require ( + google.golang.org/grpc v1.82.0 + google.golang.org/protobuf v1.36.11 +) + +require ( + golang.org/x/net v0.53.0 // indirect + golang.org/x/sys v0.43.0 // indirect + golang.org/x/text v0.36.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 // indirect +) diff --git a/serving/fleet-manager/go.sum b/serving/fleet-manager/go.sum new file mode 100644 index 0000000..0815dbd --- /dev/null +++ b/serving/fleet-manager/go.sum @@ -0,0 +1,38 @@ +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +golang.org/x/net v0.53.0 h1:d+qAbo5L0orcWAr0a9JweQpjXF19LMXJE8Ey7hwOdUA= +golang.org/x/net v0.53.0/go.mod h1:JvMuJH7rrdiCfbeHoo3fCQU24Lf5JJwT9W3sJFulfgs= +golang.org/x/sys v0.43.0 h1:Rlag2XtaFTxp19wS8MXlJwTvoh8ArU6ezoyFsMyCTNI= +golang.org/x/sys v0.43.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.36.0 h1:JfKh3XmcRPqZPKevfXVpI1wXPTqbkE5f7JA92a55Yxg= +golang.org/x/text v0.36.0/go.mod h1:NIdBknypM8iqVmPiuco0Dh6P5Jcdk8lJL0CUebqK164= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478 h1:RmoJA1ujG+/lRGNfUnOMfhCy5EipVMyvUE+KNbPbTlw= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260414002931-afd174a4e478/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.82.0 h1:vguDnZUPjE26w09A63VoxZPnvPjB5Riyc0mkXPFmAIU= +google.golang.org/grpc v1.82.0/go.mod h1:yzTZ1TB1Z3SG+LIYaI+WiE8D5+PZ3ArnrSp8zF3+/ZA= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= diff --git a/serving/fleet-manager/inventory.go b/serving/fleet-manager/inventory.go index baa0d97..c39021c 100644 --- a/serving/fleet-manager/inventory.go +++ b/serving/fleet-manager/inventory.go @@ -34,7 +34,7 @@ type Binding struct { // Enrolled — provenance of the approval that birthed this record. type Enrolled struct { Approver string `json:"approver,omitempty"` // admin CN from the approve mTLS - At time.Time `json:"at,omitempty"` // approval timestamp + At time.Time `json:"at,omitzero"` // approval timestamp Fingerprint string `json:"fingerprint,omitempty"` // the human-verified anchor } @@ -79,7 +79,7 @@ type Desired struct { RebootAt *time.Time `json:"reboot_at,omitempty"` Drain bool `json:"drain,omitempty"` UpdatedBy string `json:"updated_by,omitempty"` // admin CN - UpdatedAt time.Time `json:"updated_at,omitempty"` + UpdatedAt time.Time `json:"updated_at,omitzero"` } // Host is the full inventory record. diff --git a/serving/fleet-manager/main.go b/serving/fleet-manager/main.go index 3f5396f..5882c60 100644 --- a/serving/fleet-manager/main.go +++ b/serving/fleet-manager/main.go @@ -30,6 +30,12 @@ type config struct { adminCert string adminKey string adminClientCA string // dedicated admin CA (verifies admin browser certs) + + agentAddr string + agentCert string + agentKey string + agentClientCA string // FLEET CA (verifies host TPM identity certs) + inventoryPath string seed bool hostnqnBase string @@ -42,6 +48,12 @@ func loadConfig() config { adminCert: env("FLEET_MGR_ADMIN_CERT", ""), adminKey: env("FLEET_MGR_ADMIN_KEY", ""), adminClientCA: env("FLEET_MGR_ADMIN_CLIENT_CA", ""), + + agentAddr: env("FLEET_MGR_AGENT_ADDR", "127.0.0.1:9600"), + agentCert: env("FLEET_MGR_AGENT_CERT", ""), + agentKey: env("FLEET_MGR_AGENT_KEY", ""), + agentClientCA: env("FLEET_MGR_AGENT_CLIENT_CA", ""), + inventoryPath: env("FLEET_MGR_INVENTORY", "/var/lib/fleet-manager/inventory.json"), seed: env("FLEET_MGR_SEED", "") == "true", hostnqnBase: env("HOSTNQN_BASE", "nqn.2026-06.org.example.fleet:host-"), @@ -52,6 +64,7 @@ func loadConfig() config { type server struct { cfg config store Store + hub *agentHub // live agent streams; nil if the agent plane isn't configured } func main() { @@ -70,6 +83,20 @@ func main() { s := &server{cfg: cfg, store: store} + // Agent plane (gRPC/mTLS, fleet CA). Started only when configured; the admin + // plane can run standalone (e.g. before any host is enrolled). The hub is the + // live-stream registry the lifecycle API pushes desired-state through. + if cfg.agentCert != "" && cfg.agentKey != "" && cfg.agentClientCA != "" { + s.hub = newAgentHub(store, nil) // InfluxDB sink wired in checkpoint 4 + go func() { + if err := serveAgent(cfg, s.hub); err != nil { + log.Fatalf("agent plane: %v", err) + } + }() + } else { + log.Printf("agent plane DISABLED (set FLEET_MGR_AGENT_CERT/_KEY/_CLIENT_CA to enable)") + } + mux := http.NewServeMux() s.routes(mux) diff --git a/serving/fleet-manager/proto/fleet_agent.pb.go b/serving/fleet-manager/proto/fleet_agent.pb.go new file mode 100644 index 0000000..2da81fb --- /dev/null +++ b/serving/fleet-manager/proto/fleet_agent.pb.go @@ -0,0 +1,810 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v3.21.12 +// source: proto/fleet_agent.proto + +// The steady-state contract between the on-host fleet agent and the +// fleet-manager. ONE long-lived, host-dialed, mTLS bidi stream (health/metrics +// up, desired-state down) — the design's resolved transport fork (bespoke gRPC, +// not OTLP+Wazuh). See docs/fleet-management-design.md + tpm-enrolment-design.md. +// +// Identity is the mTLS peer cert (the host's reused TPM identity cert, fleet CA); +// the manager derives the cert serial (= inventory key) from it, so AgentUp +// messages carry NO self-asserted identity — a host can only ever be itself. + +package fleetpb + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type Command_Kind int32 + +const ( + Command_KIND_UNSPECIFIED Command_Kind = 0 + Command_REBOOT Command_Kind = 1 // reboot now (into desired target_version) + Command_DRAIN Command_Kind = 2 // stop taking work / prepare for maintenance + Command_RESYNC Command_Kind = 3 // re-fetch + re-apply desired-state now +) + +// Enum value maps for Command_Kind. +var ( + Command_Kind_name = map[int32]string{ + 0: "KIND_UNSPECIFIED", + 1: "REBOOT", + 2: "DRAIN", + 3: "RESYNC", + } + Command_Kind_value = map[string]int32{ + "KIND_UNSPECIFIED": 0, + "REBOOT": 1, + "DRAIN": 2, + "RESYNC": 3, + } +) + +func (x Command_Kind) Enum() *Command_Kind { + p := new(Command_Kind) + *p = x + return p +} + +func (x Command_Kind) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (Command_Kind) Descriptor() protoreflect.EnumDescriptor { + return file_proto_fleet_agent_proto_enumTypes[0].Descriptor() +} + +func (Command_Kind) Type() protoreflect.EnumType { + return &file_proto_fleet_agent_proto_enumTypes[0] +} + +func (x Command_Kind) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use Command_Kind.Descriptor instead. +func (Command_Kind) EnumDescriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{6, 0} +} + +type AgentUp struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Msg: + // + // *AgentUp_Hello + // *AgentUp_Snapshot + // *AgentUp_Result + Msg isAgentUp_Msg `protobuf_oneof:"msg"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *AgentUp) Reset() { + *x = AgentUp{} + mi := &file_proto_fleet_agent_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *AgentUp) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AgentUp) ProtoMessage() {} + +func (x *AgentUp) ProtoReflect() protoreflect.Message { + mi := &file_proto_fleet_agent_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AgentUp.ProtoReflect.Descriptor instead. +func (*AgentUp) Descriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{0} +} + +func (x *AgentUp) GetMsg() isAgentUp_Msg { + if x != nil { + return x.Msg + } + return nil +} + +func (x *AgentUp) GetHello() *Hello { + if x != nil { + if x, ok := x.Msg.(*AgentUp_Hello); ok { + return x.Hello + } + } + return nil +} + +func (x *AgentUp) GetSnapshot() *HealthSnapshot { + if x != nil { + if x, ok := x.Msg.(*AgentUp_Snapshot); ok { + return x.Snapshot + } + } + return nil +} + +func (x *AgentUp) GetResult() *CommandResult { + if x != nil { + if x, ok := x.Msg.(*AgentUp_Result); ok { + return x.Result + } + } + return nil +} + +type isAgentUp_Msg interface { + isAgentUp_Msg() +} + +type AgentUp_Hello struct { + Hello *Hello `protobuf:"bytes,1,opt,name=hello,proto3,oneof"` // first frame after connect +} + +type AgentUp_Snapshot struct { + Snapshot *HealthSnapshot `protobuf:"bytes,2,opt,name=snapshot,proto3,oneof"` // periodic telemetry (also written to InfluxDB) +} + +type AgentUp_Result struct { + Result *CommandResult `protobuf:"bytes,3,opt,name=result,proto3,oneof"` // outcome of a ManagerDown.Command +} + +func (*AgentUp_Hello) isAgentUp_Msg() {} + +func (*AgentUp_Snapshot) isAgentUp_Msg() {} + +func (*AgentUp_Result) isAgentUp_Msg() {} + +// Hello — what the host is running, sent once on connect so the manager can +// reconcile version drift immediately. +type Hello struct { + state protoimpl.MessageState `protogen:"open.v1"` + ImageVersion string `protobuf:"bytes,1,opt,name=image_version,json=imageVersion,proto3" json:"image_version,omitempty"` + RootfsDigest string `protobuf:"bytes,2,opt,name=rootfs_digest,json=rootfsDigest,proto3" json:"rootfs_digest,omitempty"` + Kernel string `protobuf:"bytes,3,opt,name=kernel,proto3" json:"kernel,omitempty"` + KioskUrl string `protobuf:"bytes,4,opt,name=kiosk_url,json=kioskUrl,proto3" json:"kiosk_url,omitempty"` + StarshipName string `protobuf:"bytes,5,opt,name=starship_name,json=starshipName,proto3" json:"starship_name,omitempty"` // advisory; manager already knows it from inventory + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Hello) Reset() { + *x = Hello{} + mi := &file_proto_fleet_agent_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Hello) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Hello) ProtoMessage() {} + +func (x *Hello) ProtoReflect() protoreflect.Message { + mi := &file_proto_fleet_agent_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Hello.ProtoReflect.Descriptor instead. +func (*Hello) Descriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{1} +} + +func (x *Hello) GetImageVersion() string { + if x != nil { + return x.ImageVersion + } + return "" +} + +func (x *Hello) GetRootfsDigest() string { + if x != nil { + return x.RootfsDigest + } + return "" +} + +func (x *Hello) GetKernel() string { + if x != nil { + return x.Kernel + } + return "" +} + +func (x *Hello) GetKioskUrl() string { + if x != nil { + return x.KioskUrl + } + return "" +} + +func (x *Hello) GetStarshipName() string { + if x != nil { + return x.StarshipName + } + return "" +} + +// HealthSnapshot — one telemetry sample. Cached as the latest value in inventory +// AND written as a time-series point to InfluxDB (tagged with the authenticated +// cert serial / hostnqn). +type HealthSnapshot struct { + state protoimpl.MessageState `protogen:"open.v1"` + AtUnix int64 `protobuf:"varint,1,opt,name=at_unix,json=atUnix,proto3" json:"at_unix,omitempty"` // sample time, unix seconds (0 = manager stamps now) + LoadAvg float64 `protobuf:"fixed64,2,opt,name=load_avg,json=loadAvg,proto3" json:"load_avg,omitempty"` + CpuPct float64 `protobuf:"fixed64,3,opt,name=cpu_pct,json=cpuPct,proto3" json:"cpu_pct,omitempty"` + TempC float64 `protobuf:"fixed64,4,opt,name=temp_c,json=tempC,proto3" json:"temp_c,omitempty"` + MemUsedPct float64 `protobuf:"fixed64,5,opt,name=mem_used_pct,json=memUsedPct,proto3" json:"mem_used_pct,omitempty"` + TmpfsUsedPct float64 `protobuf:"fixed64,6,opt,name=tmpfs_used_pct,json=tmpfsUsedPct,proto3" json:"tmpfs_used_pct,omitempty"` // overlay RAM pressure + NvmeRoot string `protobuf:"bytes,7,opt,name=nvme_root,json=nvmeRoot,proto3" json:"nvme_root,omitempty"` // ok|degraded|down + Auth string `protobuf:"bytes,8,opt,name=auth,proto3" json:"auth,omitempty"` // 802.1x/macsec: up|down + Selinux string `protobuf:"bytes,9,opt,name=selinux,proto3" json:"selinux,omitempty"` // enforcing|permissive + Chrony string `protobuf:"bytes,10,opt,name=chrony,proto3" json:"chrony,omitempty"` // synced|unsynced + KioskActive bool `protobuf:"varint,11,opt,name=kiosk_active,json=kioskActive,proto3" json:"kiosk_active,omitempty"` + SessionUptime string `protobuf:"bytes,12,opt,name=session_uptime,json=sessionUptime,proto3" json:"session_uptime,omitempty"` + ChromiumCrashes int32 `protobuf:"varint,13,opt,name=chromium_crashes,json=chromiumCrashes,proto3" json:"chromium_crashes,omitempty"` + Display string `protobuf:"bytes,14,opt,name=display,proto3" json:"display,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HealthSnapshot) Reset() { + *x = HealthSnapshot{} + mi := &file_proto_fleet_agent_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HealthSnapshot) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HealthSnapshot) ProtoMessage() {} + +func (x *HealthSnapshot) ProtoReflect() protoreflect.Message { + mi := &file_proto_fleet_agent_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HealthSnapshot.ProtoReflect.Descriptor instead. +func (*HealthSnapshot) Descriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{2} +} + +func (x *HealthSnapshot) GetAtUnix() int64 { + if x != nil { + return x.AtUnix + } + return 0 +} + +func (x *HealthSnapshot) GetLoadAvg() float64 { + if x != nil { + return x.LoadAvg + } + return 0 +} + +func (x *HealthSnapshot) GetCpuPct() float64 { + if x != nil { + return x.CpuPct + } + return 0 +} + +func (x *HealthSnapshot) GetTempC() float64 { + if x != nil { + return x.TempC + } + return 0 +} + +func (x *HealthSnapshot) GetMemUsedPct() float64 { + if x != nil { + return x.MemUsedPct + } + return 0 +} + +func (x *HealthSnapshot) GetTmpfsUsedPct() float64 { + if x != nil { + return x.TmpfsUsedPct + } + return 0 +} + +func (x *HealthSnapshot) GetNvmeRoot() string { + if x != nil { + return x.NvmeRoot + } + return "" +} + +func (x *HealthSnapshot) GetAuth() string { + if x != nil { + return x.Auth + } + return "" +} + +func (x *HealthSnapshot) GetSelinux() string { + if x != nil { + return x.Selinux + } + return "" +} + +func (x *HealthSnapshot) GetChrony() string { + if x != nil { + return x.Chrony + } + return "" +} + +func (x *HealthSnapshot) GetKioskActive() bool { + if x != nil { + return x.KioskActive + } + return false +} + +func (x *HealthSnapshot) GetSessionUptime() string { + if x != nil { + return x.SessionUptime + } + return "" +} + +func (x *HealthSnapshot) GetChromiumCrashes() int32 { + if x != nil { + return x.ChromiumCrashes + } + return 0 +} + +func (x *HealthSnapshot) GetDisplay() string { + if x != nil { + return x.Display + } + return "" +} + +type CommandResult struct { + state protoimpl.MessageState `protogen:"open.v1"` + CommandId string `protobuf:"bytes,1,opt,name=command_id,json=commandId,proto3" json:"command_id,omitempty"` + Ok bool `protobuf:"varint,2,opt,name=ok,proto3" json:"ok,omitempty"` + Detail string `protobuf:"bytes,3,opt,name=detail,proto3" json:"detail,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *CommandResult) Reset() { + *x = CommandResult{} + mi := &file_proto_fleet_agent_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *CommandResult) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*CommandResult) ProtoMessage() {} + +func (x *CommandResult) ProtoReflect() protoreflect.Message { + mi := &file_proto_fleet_agent_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use CommandResult.ProtoReflect.Descriptor instead. +func (*CommandResult) Descriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{3} +} + +func (x *CommandResult) GetCommandId() string { + if x != nil { + return x.CommandId + } + return "" +} + +func (x *CommandResult) GetOk() bool { + if x != nil { + return x.Ok + } + return false +} + +func (x *CommandResult) GetDetail() string { + if x != nil { + return x.Detail + } + return "" +} + +type ManagerDown struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Msg: + // + // *ManagerDown_Desired + // *ManagerDown_Command + Msg isManagerDown_Msg `protobuf_oneof:"msg"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ManagerDown) Reset() { + *x = ManagerDown{} + mi := &file_proto_fleet_agent_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ManagerDown) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ManagerDown) ProtoMessage() {} + +func (x *ManagerDown) ProtoReflect() protoreflect.Message { + mi := &file_proto_fleet_agent_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ManagerDown.ProtoReflect.Descriptor instead. +func (*ManagerDown) Descriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{4} +} + +func (x *ManagerDown) GetMsg() isManagerDown_Msg { + if x != nil { + return x.Msg + } + return nil +} + +func (x *ManagerDown) GetDesired() *DesiredState { + if x != nil { + if x, ok := x.Msg.(*ManagerDown_Desired); ok { + return x.Desired + } + } + return nil +} + +func (x *ManagerDown) GetCommand() *Command { + if x != nil { + if x, ok := x.Msg.(*ManagerDown_Command); ok { + return x.Command + } + } + return nil +} + +type isManagerDown_Msg interface { + isManagerDown_Msg() +} + +type ManagerDown_Desired struct { + Desired *DesiredState `protobuf:"bytes,1,opt,name=desired,proto3,oneof"` // full desired-state: sent on connect + on change +} + +type ManagerDown_Command struct { + Command *Command `protobuf:"bytes,2,opt,name=command,proto3,oneof"` // imperative action +} + +func (*ManagerDown_Desired) isManagerDown_Msg() {} + +func (*ManagerDown_Command) isManagerDown_Msg() {} + +// DesiredState — server-authoritative target the agent reconciles toward. +type DesiredState struct { + state protoimpl.MessageState `protogen:"open.v1"` + TargetVersion string `protobuf:"bytes,1,opt,name=target_version,json=targetVersion,proto3" json:"target_version,omitempty"` + KioskUrl string `protobuf:"bytes,2,opt,name=kiosk_url,json=kioskUrl,proto3" json:"kiosk_url,omitempty"` + RebootAtUnix int64 `protobuf:"varint,3,opt,name=reboot_at_unix,json=rebootAtUnix,proto3" json:"reboot_at_unix,omitempty"` // 0 = no scheduled reboot + Drain bool `protobuf:"varint,4,opt,name=drain,proto3" json:"drain,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *DesiredState) Reset() { + *x = DesiredState{} + mi := &file_proto_fleet_agent_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *DesiredState) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DesiredState) ProtoMessage() {} + +func (x *DesiredState) ProtoReflect() protoreflect.Message { + mi := &file_proto_fleet_agent_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DesiredState.ProtoReflect.Descriptor instead. +func (*DesiredState) Descriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{5} +} + +func (x *DesiredState) GetTargetVersion() string { + if x != nil { + return x.TargetVersion + } + return "" +} + +func (x *DesiredState) GetKioskUrl() string { + if x != nil { + return x.KioskUrl + } + return "" +} + +func (x *DesiredState) GetRebootAtUnix() int64 { + if x != nil { + return x.RebootAtUnix + } + return 0 +} + +func (x *DesiredState) GetDrain() bool { + if x != nil { + return x.Drain + } + return false +} + +// Command — an imperative action pushed live (admin clicked a button in the SPA). +type Command struct { + state protoimpl.MessageState `protogen:"open.v1"` + CommandId string `protobuf:"bytes,1,opt,name=command_id,json=commandId,proto3" json:"command_id,omitempty"` + Kind Command_Kind `protobuf:"varint,2,opt,name=kind,proto3,enum=fleet.agent.v1.Command_Kind" json:"kind,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Command) Reset() { + *x = Command{} + mi := &file_proto_fleet_agent_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Command) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Command) ProtoMessage() {} + +func (x *Command) ProtoReflect() protoreflect.Message { + mi := &file_proto_fleet_agent_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Command.ProtoReflect.Descriptor instead. +func (*Command) Descriptor() ([]byte, []int) { + return file_proto_fleet_agent_proto_rawDescGZIP(), []int{6} +} + +func (x *Command) GetCommandId() string { + if x != nil { + return x.CommandId + } + return "" +} + +func (x *Command) GetKind() Command_Kind { + if x != nil { + return x.Kind + } + return Command_KIND_UNSPECIFIED +} + +var File_proto_fleet_agent_proto protoreflect.FileDescriptor + +const file_proto_fleet_agent_proto_rawDesc = "" + + "\n" + + "\x17proto/fleet_agent.proto\x12\x0efleet.agent.v1\"\xb6\x01\n" + + "\aAgentUp\x12-\n" + + "\x05hello\x18\x01 \x01(\v2\x15.fleet.agent.v1.HelloH\x00R\x05hello\x12<\n" + + "\bsnapshot\x18\x02 \x01(\v2\x1e.fleet.agent.v1.HealthSnapshotH\x00R\bsnapshot\x127\n" + + "\x06result\x18\x03 \x01(\v2\x1d.fleet.agent.v1.CommandResultH\x00R\x06resultB\x05\n" + + "\x03msg\"\xab\x01\n" + + "\x05Hello\x12#\n" + + "\rimage_version\x18\x01 \x01(\tR\fimageVersion\x12#\n" + + "\rrootfs_digest\x18\x02 \x01(\tR\frootfsDigest\x12\x16\n" + + "\x06kernel\x18\x03 \x01(\tR\x06kernel\x12\x1b\n" + + "\tkiosk_url\x18\x04 \x01(\tR\bkioskUrl\x12#\n" + + "\rstarship_name\x18\x05 \x01(\tR\fstarshipName\"\xae\x03\n" + + "\x0eHealthSnapshot\x12\x17\n" + + "\aat_unix\x18\x01 \x01(\x03R\x06atUnix\x12\x19\n" + + "\bload_avg\x18\x02 \x01(\x01R\aloadAvg\x12\x17\n" + + "\acpu_pct\x18\x03 \x01(\x01R\x06cpuPct\x12\x15\n" + + "\x06temp_c\x18\x04 \x01(\x01R\x05tempC\x12 \n" + + "\fmem_used_pct\x18\x05 \x01(\x01R\n" + + "memUsedPct\x12$\n" + + "\x0etmpfs_used_pct\x18\x06 \x01(\x01R\ftmpfsUsedPct\x12\x1b\n" + + "\tnvme_root\x18\a \x01(\tR\bnvmeRoot\x12\x12\n" + + "\x04auth\x18\b \x01(\tR\x04auth\x12\x18\n" + + "\aselinux\x18\t \x01(\tR\aselinux\x12\x16\n" + + "\x06chrony\x18\n" + + " \x01(\tR\x06chrony\x12!\n" + + "\fkiosk_active\x18\v \x01(\bR\vkioskActive\x12%\n" + + "\x0esession_uptime\x18\f \x01(\tR\rsessionUptime\x12)\n" + + "\x10chromium_crashes\x18\r \x01(\x05R\x0fchromiumCrashes\x12\x18\n" + + "\adisplay\x18\x0e \x01(\tR\adisplay\"V\n" + + "\rCommandResult\x12\x1d\n" + + "\n" + + "command_id\x18\x01 \x01(\tR\tcommandId\x12\x0e\n" + + "\x02ok\x18\x02 \x01(\bR\x02ok\x12\x16\n" + + "\x06detail\x18\x03 \x01(\tR\x06detail\"\x83\x01\n" + + "\vManagerDown\x128\n" + + "\adesired\x18\x01 \x01(\v2\x1c.fleet.agent.v1.DesiredStateH\x00R\adesired\x123\n" + + "\acommand\x18\x02 \x01(\v2\x17.fleet.agent.v1.CommandH\x00R\acommandB\x05\n" + + "\x03msg\"\x8e\x01\n" + + "\fDesiredState\x12%\n" + + "\x0etarget_version\x18\x01 \x01(\tR\rtargetVersion\x12\x1b\n" + + "\tkiosk_url\x18\x02 \x01(\tR\bkioskUrl\x12$\n" + + "\x0ereboot_at_unix\x18\x03 \x01(\x03R\frebootAtUnix\x12\x14\n" + + "\x05drain\x18\x04 \x01(\bR\x05drain\"\x9b\x01\n" + + "\aCommand\x12\x1d\n" + + "\n" + + "command_id\x18\x01 \x01(\tR\tcommandId\x120\n" + + "\x04kind\x18\x02 \x01(\x0e2\x1c.fleet.agent.v1.Command.KindR\x04kind\"?\n" + + "\x04Kind\x12\x14\n" + + "\x10KIND_UNSPECIFIED\x10\x00\x12\n" + + "\n" + + "\x06REBOOT\x10\x01\x12\t\n" + + "\x05DRAIN\x10\x02\x12\n" + + "\n" + + "\x06RESYNC\x10\x032Q\n" + + "\n" + + "FleetAgent\x12C\n" + + "\aSession\x12\x17.fleet.agent.v1.AgentUp\x1a\x1b.fleet.agent.v1.ManagerDown(\x010\x01B\x1dZ\x1bfleet-manager/proto;fleetpbb\x06proto3" + +var ( + file_proto_fleet_agent_proto_rawDescOnce sync.Once + file_proto_fleet_agent_proto_rawDescData []byte +) + +func file_proto_fleet_agent_proto_rawDescGZIP() []byte { + file_proto_fleet_agent_proto_rawDescOnce.Do(func() { + file_proto_fleet_agent_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_proto_fleet_agent_proto_rawDesc), len(file_proto_fleet_agent_proto_rawDesc))) + }) + return file_proto_fleet_agent_proto_rawDescData +} + +var file_proto_fleet_agent_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_proto_fleet_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 7) +var file_proto_fleet_agent_proto_goTypes = []any{ + (Command_Kind)(0), // 0: fleet.agent.v1.Command.Kind + (*AgentUp)(nil), // 1: fleet.agent.v1.AgentUp + (*Hello)(nil), // 2: fleet.agent.v1.Hello + (*HealthSnapshot)(nil), // 3: fleet.agent.v1.HealthSnapshot + (*CommandResult)(nil), // 4: fleet.agent.v1.CommandResult + (*ManagerDown)(nil), // 5: fleet.agent.v1.ManagerDown + (*DesiredState)(nil), // 6: fleet.agent.v1.DesiredState + (*Command)(nil), // 7: fleet.agent.v1.Command +} +var file_proto_fleet_agent_proto_depIdxs = []int32{ + 2, // 0: fleet.agent.v1.AgentUp.hello:type_name -> fleet.agent.v1.Hello + 3, // 1: fleet.agent.v1.AgentUp.snapshot:type_name -> fleet.agent.v1.HealthSnapshot + 4, // 2: fleet.agent.v1.AgentUp.result:type_name -> fleet.agent.v1.CommandResult + 6, // 3: fleet.agent.v1.ManagerDown.desired:type_name -> fleet.agent.v1.DesiredState + 7, // 4: fleet.agent.v1.ManagerDown.command:type_name -> fleet.agent.v1.Command + 0, // 5: fleet.agent.v1.Command.kind:type_name -> fleet.agent.v1.Command.Kind + 1, // 6: fleet.agent.v1.FleetAgent.Session:input_type -> fleet.agent.v1.AgentUp + 5, // 7: fleet.agent.v1.FleetAgent.Session:output_type -> fleet.agent.v1.ManagerDown + 7, // [7:8] is the sub-list for method output_type + 6, // [6:7] is the sub-list for method input_type + 6, // [6:6] is the sub-list for extension type_name + 6, // [6:6] is the sub-list for extension extendee + 0, // [0:6] is the sub-list for field type_name +} + +func init() { file_proto_fleet_agent_proto_init() } +func file_proto_fleet_agent_proto_init() { + if File_proto_fleet_agent_proto != nil { + return + } + file_proto_fleet_agent_proto_msgTypes[0].OneofWrappers = []any{ + (*AgentUp_Hello)(nil), + (*AgentUp_Snapshot)(nil), + (*AgentUp_Result)(nil), + } + file_proto_fleet_agent_proto_msgTypes[4].OneofWrappers = []any{ + (*ManagerDown_Desired)(nil), + (*ManagerDown_Command)(nil), + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_fleet_agent_proto_rawDesc), len(file_proto_fleet_agent_proto_rawDesc)), + NumEnums: 1, + NumMessages: 7, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_proto_fleet_agent_proto_goTypes, + DependencyIndexes: file_proto_fleet_agent_proto_depIdxs, + EnumInfos: file_proto_fleet_agent_proto_enumTypes, + MessageInfos: file_proto_fleet_agent_proto_msgTypes, + }.Build() + File_proto_fleet_agent_proto = out.File + file_proto_fleet_agent_proto_goTypes = nil + file_proto_fleet_agent_proto_depIdxs = nil +} diff --git a/serving/fleet-manager/proto/fleet_agent.proto b/serving/fleet-manager/proto/fleet_agent.proto new file mode 100644 index 0000000..80f230b --- /dev/null +++ b/serving/fleet-manager/proto/fleet_agent.proto @@ -0,0 +1,96 @@ +syntax = "proto3"; + +// The steady-state contract between the on-host fleet agent and the +// fleet-manager. ONE long-lived, host-dialed, mTLS bidi stream (health/metrics +// up, desired-state down) — the design's resolved transport fork (bespoke gRPC, +// not OTLP+Wazuh). See docs/fleet-management-design.md + tpm-enrolment-design.md. +// +// Identity is the mTLS peer cert (the host's reused TPM identity cert, fleet CA); +// the manager derives the cert serial (= inventory key) from it, so AgentUp +// messages carry NO self-asserted identity — a host can only ever be itself. +package fleet.agent.v1; + +option go_package = "fleet-manager/proto;fleetpb"; + +service FleetAgent { + // Session: the single bidirectional stream. The host opens it (dials out, no + // inbound host port); the manager pushes desired-state/commands back over the + // same connection. Closing/keepalive-timeout = the host's offline signal. + rpc Session(stream AgentUp) returns (stream ManagerDown); +} + +// ---- host -> manager ------------------------------------------------------ + +message AgentUp { + oneof msg { + Hello hello = 1; // first frame after connect + HealthSnapshot snapshot = 2; // periodic telemetry (also written to InfluxDB) + CommandResult result = 3; // outcome of a ManagerDown.Command + } +} + +// Hello — what the host is running, sent once on connect so the manager can +// reconcile version drift immediately. +message Hello { + string image_version = 1; + string rootfs_digest = 2; + string kernel = 3; + string kiosk_url = 4; + string starship_name = 5; // advisory; manager already knows it from inventory +} + +// HealthSnapshot — one telemetry sample. Cached as the latest value in inventory +// AND written as a time-series point to InfluxDB (tagged with the authenticated +// cert serial / hostnqn). +message HealthSnapshot { + int64 at_unix = 1; // sample time, unix seconds (0 = manager stamps now) + double load_avg = 2; + double cpu_pct = 3; + double temp_c = 4; + double mem_used_pct = 5; + double tmpfs_used_pct = 6; // overlay RAM pressure + string nvme_root = 7; // ok|degraded|down + string auth = 8; // 802.1x/macsec: up|down + string selinux = 9; // enforcing|permissive + string chrony = 10; // synced|unsynced + bool kiosk_active = 11; + string session_uptime = 12; + int32 chromium_crashes = 13; + string display = 14; +} + +message CommandResult { + string command_id = 1; + bool ok = 2; + string detail = 3; +} + +// ---- manager -> host ------------------------------------------------------ + +message ManagerDown { + oneof msg { + DesiredState desired = 1; // full desired-state: sent on connect + on change + Command command = 2; // imperative action + } +} + +// DesiredState — server-authoritative target the agent reconciles toward. +message DesiredState { + string target_version = 1; + string kiosk_url = 2; + int64 reboot_at_unix = 3; // 0 = no scheduled reboot + bool drain = 4; +} + +// Command — an imperative action pushed live (admin clicked a button in the SPA). +message Command { + string command_id = 1; + Kind kind = 2; + + enum Kind { + KIND_UNSPECIFIED = 0; + REBOOT = 1; // reboot now (into desired target_version) + DRAIN = 2; // stop taking work / prepare for maintenance + RESYNC = 3; // re-fetch + re-apply desired-state now + } +} diff --git a/serving/fleet-manager/proto/fleet_agent_grpc.pb.go b/serving/fleet-manager/proto/fleet_agent_grpc.pb.go new file mode 100644 index 0000000..7e42882 --- /dev/null +++ b/serving/fleet-manager/proto/fleet_agent_grpc.pb.go @@ -0,0 +1,130 @@ +// Code generated by protoc-gen-go-grpc. DO NOT EDIT. +// versions: +// - protoc-gen-go-grpc v1.6.2 +// - protoc v3.21.12 +// source: proto/fleet_agent.proto + +// The steady-state contract between the on-host fleet agent and the +// fleet-manager. ONE long-lived, host-dialed, mTLS bidi stream (health/metrics +// up, desired-state down) — the design's resolved transport fork (bespoke gRPC, +// not OTLP+Wazuh). See docs/fleet-management-design.md + tpm-enrolment-design.md. +// +// Identity is the mTLS peer cert (the host's reused TPM identity cert, fleet CA); +// the manager derives the cert serial (= inventory key) from it, so AgentUp +// messages carry NO self-asserted identity — a host can only ever be itself. + +package fleetpb + +import ( + context "context" + grpc "google.golang.org/grpc" + codes "google.golang.org/grpc/codes" + status "google.golang.org/grpc/status" +) + +// This is a compile-time assertion to ensure that this generated file +// is compatible with the grpc package it is being compiled against. +// Requires gRPC-Go v1.64.0 or later. +const _ = grpc.SupportPackageIsVersion9 + +const ( + FleetAgent_Session_FullMethodName = "/fleet.agent.v1.FleetAgent/Session" +) + +// FleetAgentClient is the client API for FleetAgent service. +// +// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream. +type FleetAgentClient interface { + // Session: the single bidirectional stream. The host opens it (dials out, no + // inbound host port); the manager pushes desired-state/commands back over the + // same connection. Closing/keepalive-timeout = the host's offline signal. + Session(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[AgentUp, ManagerDown], error) +} + +type fleetAgentClient struct { + cc grpc.ClientConnInterface +} + +func NewFleetAgentClient(cc grpc.ClientConnInterface) FleetAgentClient { + return &fleetAgentClient{cc} +} + +func (c *fleetAgentClient) Session(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[AgentUp, ManagerDown], error) { + cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...) + stream, err := c.cc.NewStream(ctx, &FleetAgent_ServiceDesc.Streams[0], FleetAgent_Session_FullMethodName, cOpts...) + if err != nil { + return nil, err + } + x := &grpc.GenericClientStream[AgentUp, ManagerDown]{ClientStream: stream} + return x, nil +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type FleetAgent_SessionClient = grpc.BidiStreamingClient[AgentUp, ManagerDown] + +// FleetAgentServer is the server API for FleetAgent service. +// All implementations must embed UnimplementedFleetAgentServer +// for forward compatibility. +type FleetAgentServer interface { + // Session: the single bidirectional stream. The host opens it (dials out, no + // inbound host port); the manager pushes desired-state/commands back over the + // same connection. Closing/keepalive-timeout = the host's offline signal. + Session(grpc.BidiStreamingServer[AgentUp, ManagerDown]) error + mustEmbedUnimplementedFleetAgentServer() +} + +// UnimplementedFleetAgentServer must be embedded to have +// forward compatible implementations. +// +// NOTE: this should be embedded by value instead of pointer to avoid a nil +// pointer dereference when methods are called. +type UnimplementedFleetAgentServer struct{} + +func (UnimplementedFleetAgentServer) Session(grpc.BidiStreamingServer[AgentUp, ManagerDown]) error { + return status.Error(codes.Unimplemented, "method Session not implemented") +} +func (UnimplementedFleetAgentServer) mustEmbedUnimplementedFleetAgentServer() {} +func (UnimplementedFleetAgentServer) testEmbeddedByValue() {} + +// UnsafeFleetAgentServer may be embedded to opt out of forward compatibility for this service. +// Use of this interface is not recommended, as added methods to FleetAgentServer will +// result in compilation errors. +type UnsafeFleetAgentServer interface { + mustEmbedUnimplementedFleetAgentServer() +} + +func RegisterFleetAgentServer(s grpc.ServiceRegistrar, srv FleetAgentServer) { + // If the following call panics, it indicates UnimplementedFleetAgentServer was + // embedded by pointer and is nil. This will cause panics if an + // unimplemented method is ever invoked, so we test this at initialization + // time to prevent it from happening at runtime later due to I/O. + if t, ok := srv.(interface{ testEmbeddedByValue() }); ok { + t.testEmbeddedByValue() + } + s.RegisterService(&FleetAgent_ServiceDesc, srv) +} + +func _FleetAgent_Session_Handler(srv interface{}, stream grpc.ServerStream) error { + return srv.(FleetAgentServer).Session(&grpc.GenericServerStream[AgentUp, ManagerDown]{ServerStream: stream}) +} + +// This type alias is provided for backwards compatibility with existing code that references the prior non-generic stream type by name. +type FleetAgent_SessionServer = grpc.BidiStreamingServer[AgentUp, ManagerDown] + +// FleetAgent_ServiceDesc is the grpc.ServiceDesc for FleetAgent service. +// It's only intended for direct use with grpc.RegisterService, +// and not to be introspected or modified (even as a copy) +var FleetAgent_ServiceDesc = grpc.ServiceDesc{ + ServiceName: "fleet.agent.v1.FleetAgent", + HandlerType: (*FleetAgentServer)(nil), + Methods: []grpc.MethodDesc{}, + Streams: []grpc.StreamDesc{ + { + StreamName: "Session", + Handler: _FleetAgent_Session_Handler, + ServerStreams: true, + ClientStreams: true, + }, + }, + Metadata: "proto/fleet_agent.proto", +}