6b7245e6 — eitri

docs/assumptions.md

+severe and quiet — refusing would read every live guest as lost and boot a
+second hypervisor onto its disk — so the first roll should watch for a VM going
+from running to creating rather than trusting the tests.
+### 30000–32767 belongs to eitri on every host
+Host ports for published guest ports are allocated from this range, and nothing
+else on the host is assumed to want it. Underpins allocating a port without
+asking the host what is free.
+**Proven** in code: a bind that loses the port is reported failed with the OS
+error and retried on every converge.
+### A published port may be bound on every interface
+The listener binds 0.0.0.0—every interface, IPv4 only. An exposure is
+reachable on any IPv4 address the host has, and on none of its IPv6 ones.
+Underpins there being one exposure rather than one per network.
+**Proven** in code; the posture is deliberate—see the auth answer in
+[faq.md](faq.md). A host whose networks are not equally trusted gets more
+reach than it asked for.

docs/faq.md

Old		New
@@ -12,6 +12,18 @@ install [Tailscale](https://tailscale.com) (or WireGuard) inside it—it is
12	a normal Linux machine. A public gateway is on the	12	a normal Linux machine. A public gateway is on the
13	[roadmap](../ROADMAP.md).	13	[roadmap](../ROADMAP.md).
14		14
		15	## Is an exposed port authenticated?
		16
		17	No. The SSH [jump gate](ssh-access.md) authenticates every connection against
		18	your tenant's CA. A published port has nothing in front of it: whoever can
		19	reach the host on that port reaches the service, exactly as if the service were
		20	running on the host itself.
		21
		22	That is the LAN trust posture, deliberately. The listener binds every interface
		23	on the host, IPv4 only. Publish what you are content to serve to everything
		24	that can reach it; put anything else behind the gate, or behind the service's
		25	own authentication.
		26
15	## VMs boot and SSH works, but have no outbound network—why?	27	## VMs boot and SSH works, but have no outbound network—why?
16		28
17	Docker. Installing (or starting) Docker on a host sets the kernel's iptables	29	Docker. Installing (or starting) Docker on a host sets the kernel's iptables

docs/openapi.json

@@ @@ -103,6 +103,17 @@ @@
+        ],
+        "type": "object"
+      },
+      "CreateExposureRequest": {
+        "properties": {
+          "guest_port": {
+            "type": "integer"
+          },
+          "host_port": {
+            "type": "integer"
+          }
+        },
+        "type": "object"
+      },
+      "CreateVMRequest": {
+        "properties": {
+          "cloud_init": {
@@ @@ -220,6 +231,58 @@ @@
+        ],
+        "type": "object"
+      },
+      "Exposure": {
+        "properties": {
+          "created_at": {
+            "format": "date-time",
+            "type": "string"
+          },
+          "guest_port": {
+            "type": "integer"
+          },
+          "host_addr": {
+            "type": "string"
+          },
+          "host_id": {
+            "type": "string"
+          },
+          "host_port": {
+            "type": "integer"
+          },
+          "id": {
+            "type": "string"
+          },
+          "protocol": {
+            "type": "string"
+          },
+          "reason": {
+            "type": "string"
+          },
+          "scope": {
+            "type": "string"
+          },
+          "state": {
+            "type": "string"
+          },
+          "vm_id": {
+            "type": "string"
+          }
+        },
+        "required": [
+          "created_at",
+          "guest_port",
+          "host_addr",
+          "host_id",
+          "host_port",
+          "id",
+          "protocol",
+          "reason",
+          "scope",
+          "state",
+          "vm_id"
+        ],
+        "type": "object"
+      },
+      "Host": {
+        "properties": {
+          "agent_update_available": {
@@ @@ -306,6 +369,9 @@ @@
+          "status": {
+            "type": "string"
+          },
+          "uplink_addr": {
+            "type": "string"
+          },
+          "virt": {
+            "type": "string"
@@ @@ -331,6 +397,7 @@ @@
+          "sessions",
+          "stale",
+          "status",
+          "uplink_addr",
+          "virt"
+        ],
+        "type": "object"
@@ @@ -802,6 +869,41 @@ @@
+        "summary": "Live fleet state stream (Server-Sent Events); each 'state' event carries a StateSnapshot."
+    },
+    "/api/v1/exposures/{id}": {
+      "delete": {
+        "parameters": [
+          {
+            "in": "path",
+            "name": "id",
+            "required": true,
+            "schema": {
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "204": {
+            "description": "success"
+          },
+          "default": {
+            "content": {
+              "text/plain": {
+                "schema": {
+                  "type": "string"
+                }
+              }
+            },
+            "description": "error (plain text)"
+          }
+        },
+        "security": [
+          {
+            "patToken": []
+          }
+        ],
+        "summary": "Revoke an exposure; its host closes the listener on the next converge."
+      }
+    },
+    "/api/v1/hosts": {
+      "get": {
+        "responses": {
@@ @@ -1681,6 +1783,101 @@ @@
+        "summary": "One VM's lifecycle timeline (audit rows carrying its vm_id), newest first; survives the VM row being reaped."
+    },
+    "/api/v1/vms/{id}/exposures": {
+      "get": {
+        "parameters": [
+          {
+            "in": "path",
+            "name": "id",
+            "required": true,
+            "schema": {
+              "type": "string"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "items": {
+                    "$ref": "#/components/schemas/Exposure"
+                  },
+                  "type": "array"
+                }
+              }
+            },
+            "description": "success"
+          },
+          "default": {
+            "content": {
+              "text/plain": {
+                "schema": {
+                  "type": "string"
+                }
+              }
+            },
+            "description": "error (plain text)"
+          }
+        },
+        "security": [
+          {
+            "patToken": []
+          }
+        ],
+        "summary": "List the VM's published ports, with the host address to dial and each listener's reported state."
+      },
+      "post": {
+        "parameters": [
+          {
+            "in": "path",
+            "name": "id",
+            "required": true,
+            "schema": {
+              "type": "string"
+            }
+          }
+        ],
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/CreateExposureRequest"
+              }
+            }
+          },
+          "required": true
+        },
+        "responses": {
+          "201": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/Exposure"
+                }
+              }
+            },
+            "description": "success"
+          },
+          "default": {
+            "content": {
+              "text/plain": {
+                "schema": {
+                  "type": "string"
+                }
+              }
+            },
+            "description": "error (plain text)"
+          }
+        },
+        "security": [
+          {
+            "patToken": []
+          }
+        ],
+        "summary": "Publish a guest TCP port on the VM's host. Omit host_port to allocate one from the reserved range 30000-32767; a named port must be \u003e= 1024 and is honored or refused."
+      }
+    },
+    "/api/v1/vms/{id}/restore": {
+      "post": {
+        "parameters": [

docs/quickstart.md

@@ @@ -135,6 +135,29 @@ eitri ssh <vm-name> @@
+eitri ssh <vm-name> uptime
+```
+### Expose a service
+The gate reaches a guest over SSH. Anything else a guest serves needs a
+published port: the fleet binds one on the VM's host and pipes it to the guest.
+Open the VM's page in the console, find **Exposures**, and enter the port your
+service listens on inside the guest—8080, say. Leave the host port blank and
+eitri allocates one from 30000–32767, the range it reserves on every host. The
+row then reads:
+    guest :8080  →  192.168.0.190:30080   ● active
+That address is the host's. Anything that can reach the host on that port
+reaches the service—there is no authentication in front of a published port, so
+publish what you are content to serve to everything on that network, and leave
+the rest to the gate.
+`active` means the host's listener is bound. Whether anything answers on the
+guest's 8080 is the guest's business.
+Remove the row to take the port down. Deleting the VM takes its exposures with
+it.
+## Self-hosting
+Run the whole control plane yourself—everything below is your own box.

docs/shape.html

Old		New
@@ -172,6 +172,14 @@
172	"imports": []	172	"imports": []
173	},	173	},
174	{	174	{
		175	"importPath": "internal/agent/exposeproxy",
		176	"plane": "data",
		177	"synopsis": "Package exposeproxy publishes guest ports on their host.",
		178	"imports": [
		179	"internal/pb"
		180	]
		181	},
		182	{
175	"importPath": "internal/agent/hostinfo",	183	"importPath": "internal/agent/hostinfo",
176	"plane": "data",	184	"plane": "data",
177	"synopsis": "Package hostinfo gathers best-effort facts and live metrics about the fleet host the agent runs on.",	185	"synopsis": "Package hostinfo gathers best-effort facts and live metrics about the fleet host the agent runs on.",
@@ -232,6 +240,7 @@
232	"internal/agent/bootstrap",	240	"internal/agent/bootstrap",
233	"internal/agent/cloudhv",	241	"internal/agent/cloudhv",
234	"internal/agent/enrollclient",	242	"internal/agent/enrollclient",
		243	"internal/agent/exposeproxy",
235	"internal/agent/hostinfo",	244	"internal/agent/hostinfo",
236	"internal/agent/imagecache",	245	"internal/agent/imagecache",
237	"internal/agent/netenv",	246	"internal/agent/netenv",

docs/shape.json

Old		New
@@ -121,6 +121,14 @@
121	"imports": []	121	"imports": []
122	},	122	},
123	{	123	{
		124	"importPath": "internal/agent/exposeproxy",
		125	"plane": "data",
		126	"synopsis": "Package exposeproxy publishes guest ports on their host.",
		127	"imports": [
		128	"internal/pb"
		129	]
		130	},
		131	{
124	"importPath": "internal/agent/hostinfo",	132	"importPath": "internal/agent/hostinfo",
125	"plane": "data",	133	"plane": "data",
126	"synopsis": "Package hostinfo gathers best-effort facts and live metrics about the fleet host the agent runs on.",	134	"synopsis": "Package hostinfo gathers best-effort facts and live metrics about the fleet host the agent runs on.",
@@ -181,6 +189,7 @@
181	"internal/agent/bootstrap",	189	"internal/agent/bootstrap",
182	"internal/agent/cloudhv",	190	"internal/agent/cloudhv",
183	"internal/agent/enrollclient",	191	"internal/agent/enrollclient",
		192	"internal/agent/exposeproxy",
184	"internal/agent/hostinfo",	193	"internal/agent/hostinfo",
185	"internal/agent/imagecache",	194	"internal/agent/imagecache",
186	"internal/agent/netenv",	195	"internal/agent/netenv",

internal/agent/exposeproxy/exposeproxy.go

@@ @@ -0,0 +1,205 @@ @@
+// Package exposeproxy publishes guest ports on their host. A Manager owns one
+// listener and one goroutine per exposure id — the same manager shape as the
+// serial pumps — and Converge drives that set toward the exposures the fleet
+// says this host should be serving: a new id binds, a vanished id closes, a
+// changed spec closes and rebinds.
+//
+// The listener binds 0.0.0.0. All interfaces is the point: the LAN address is
+// the feature, the tailnet address is how an operator reaches it from
+// elsewhere, and the guest-side bridge seeing it is the same trust domain the
+// LAN posture already accepts. There is no authentication in front of a
+// published port — reaching the host is reaching the service.
+//
+// Each accepted connection asks for the VM's address AT THAT MOMENT and dials
+// the guest fresh. That per-connection lookup is what makes an exposure created
+// before its guest has booted simply work: the listener binds now, and
+// connections start succeeding when the guest does.
+//
+// Nothing here is persisted. Listeners are rebuilt from the first snapshot
+// after the agent starts, the same way the consoles are; connections in flight
+// across an agent restart drop, and reconnecting works.
+package exposeproxy
+import (
+	"fmt"
+	"io"
+	"log/slog"
+	"net"
+	"strconv"
+	"sync"
+	"github.com/a73x/eitri/internal/pb"
+)
+// Manager runs one listener per active exposure.
+type Manager struct {
+	// addr answers a VM's current guest address, or "" when this host does not
+	// know one. Injected so the package stays a leaf: it never learns what a VM
+	// is, only where one is right now. It is called concurrently from connection
+	// goroutines; it must be safe for concurrent use.
+	addr func(vmID string) string
+	mu   sync.Mutex
+	live map[string]*exposure
+}
+// NewManager returns a Manager that resolves each connection's destination
+// through addr.
+func NewManager(addr func(vmID string) string) *Manager {
+	return &Manager{addr: addr, live: map[string]*exposure{}}
+}
+// exposure is one published port's running state: the spec key it was bound
+// for, its listener (nil when the bind failed), and what the OS said if it did.
+type exposure struct {
+	key    string
+	ln     net.Listener
+	reason string
+}
+// close tears down the listener, which is what ends its accept goroutine.
+func (e *exposure) close() {
+	if e.ln != nil {
+		e.ln.Close()
+		e.ln = nil
+	}
+}
+// specKey renders everything about a desired exposure that a bound listener
+// depends on. A change to any of it is a close and a rebind rather than an
+// edit — the listener's own address is part of what changed.
+func specKey(d *pb.ExposureDesired) string {
+	return fmt.Sprintf("%s/%d/%d/%s", d.GetVmId(), d.GetGuestPort(), d.GetHostPort(), d.GetProtocol())
+}
+// Converge drives the running listeners toward desired and reports what each
+// exposure is doing, in desired order. It is level-triggered: every call
+// re-examines the whole set, so an exposure whose listener is gone — a failed
+// bind, or an accept loop that died — is retried here and heals the moment the
+// port frees.
+//
+// "active" means the listener is bound. Eitri owns the host half of the pipe;
+// whether anything answers inside the guest is the guest's half, and this does
+// not pretend otherwise.
+func (m *Manager) Converge(desired []*pb.ExposureDesired) []*pb.ExposureActual {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	want := make(map[string]*pb.ExposureDesired, len(desired))
+	for _, d := range desired {
+		want[d.GetId()] = d
+	}
+	for id, ex := range m.live {
+		if d, ok := want[id]; ok && ex.key == specKey(d) {
+			continue
+		}
+		ex.close()
+		delete(m.live, id)
+	}
+	out := make([]*pb.ExposureActual, 0, len(desired))
+	for _, d := range desired {
+		ex, ok := m.live[d.GetId()]
+		if !ok {
+			ex = &exposure{key: specKey(d)}
+			m.live[d.GetId()] = ex
+		}
+		if ex.ln == nil {
+			ln, err := net.Listen("tcp", net.JoinHostPort("0.0.0.0", strconv.Itoa(int(d.GetHostPort()))))
+			if err != nil {
+				ex.reason = err.Error()
+				out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "failed", Reason: ex.reason})
+				continue
+			}
+			ex.ln, ex.reason = ln, ""
+			go m.accept(d.GetId(), ln, d.GetVmId(), d.GetGuestPort())
+		}
+		out = append(out, &pb.ExposureActual{Id: d.GetId(), State: "active"})
+	}
+	return out
+}
+// StopAll closes every listener (agent shutdown, tests).
+func (m *Manager) StopAll() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for id, ex := range m.live {
+		ex.close()
+		delete(m.live, id)
+	}
+}
+// accept serves one exposure's listener until it ends: a converge dropped it,
+// StopAll closed it, or the listener itself broke. That third exit is why this
+// gives the exposure back its listener-less state instead of just returning —
+// an accept loop that died leaves a port bound that nothing is serving, and a
+// caller would hang in the backlog while Converge went on reporting "active".
+// Releasing the listener and recording why makes the next converge rebind it.
+//
+// The pointer-identity guard is what keeps the normal exits quiet: on a
+// close-driven exit the entry is already gone, already nil, or already holds a
+// newer listener, so only the loop that owns the current listener writes here.
+func (m *Manager) accept(id string, ln net.Listener, vmID string, guestPort uint32) {
+	for {
+		conn, err := ln.Accept()
+		if err != nil {
+			m.mu.Lock()
+			if ex, ok := m.live[id]; ok && ex.ln == ln {
+				ex.close()
+				ex.reason = "accept: " + err.Error()
+				// The reason travels in the next report, which nobody watching
+				// the host sees; fd exhaustion and a broken listener belong in
+				// the agent's log too.
+				slog.Warn("exposure accept loop died", "exposure", id, "err", err)
+			}
+			m.mu.Unlock()
+			return
+		}
+		go m.pipe(conn, vmID, guestPort)
+	}
+}
+// pipe connects one accepted connection to the guest. No address (a guest still
+// leasing) or a guest that will not answer closes immediately: the host half of
+// the pipe exists, the guest half does not, and waiting would only hold the
+// caller open on a promise nothing is keeping.
+func (m *Manager) pipe(client net.Conn, vmID string, guestPort uint32) {
+	ip := m.addr(vmID)
+	if ip == "" {
+		client.Close()
+		return
+	}
+	guest, err := net.Dial("tcp", net.JoinHostPort(ip, strconv.Itoa(int(guestPort))))
+	if err != nil {
+		client.Close()
+		return
+	}
+	splice(client, guest)
+}
+// splice copies in both directions until each ends, closing both connections
+// once they have.
+func splice(a, b net.Conn) {
+	defer a.Close()
+	defer b.Close()
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() { defer wg.Done(); copyThenCloseWrite(b, a) }()
+	go func() { defer wg.Done(); copyThenCloseWrite(a, b) }()
+	wg.Wait()
+}
+// copyThenCloseWrite copies src into dst and then shuts down dst's WRITE half
+// only, so the peer sees a clean EOF while the opposite direction keeps
+// flowing. Half-close is what makes a protocol that signals end-of-request by
+// closing its write side work through the proxy instead of hanging. A
+// connection with no half-close is closed outright — consumer-side interface,
+// so nothing here depends on the concrete net type.
+func copyThenCloseWrite(dst, src net.Conn) {
+	_, _ = io.Copy(dst, src)
+	if cw, ok := dst.(interface{ CloseWrite() error }); ok {
+		_ = cw.CloseWrite()
+		return
+	}
+	_ = dst.Close()
+}

internal/agent/exposeproxy/exposeproxy_test.go

@@ @@ -0,0 +1,335 @@ @@
+package exposeproxy
+import (
+	"io"
+	"net"
+	"strconv"
+	"testing"
+	"time"
+	"github.com/a73x/eitri/internal/pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// fakeGuest is a loopback listener standing in for a service inside a guest.
+// It echoes everything it reads back, prefixed, so a test can prove bytes went
+// both ways through the proxy.
+type fakeGuest struct {
+	ln   net.Listener
+	addr string
+	port uint32
+}
+func newFakeGuest(t *testing.T) *fakeGuest {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { ln.Close() })
+	g := &fakeGuest{ln: ln, addr: "127.0.0.1", port: portOf(t, ln.Addr())}
+	go func() {
+		for {
+			c, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func() {
+				defer c.Close()
+				buf := make([]byte, 64)
+				n, err := c.Read(buf)
+				if n > 0 {
+					c.Write(append([]byte("echo:"), buf[:n]...)) //nolint:errcheck
+				}
+				if err != nil {
+					return
+				}
+			}()
+		}
+	}()
+	return g
+}
+func portOf(t *testing.T, a net.Addr) uint32 {
+	t.Helper()
+	ta, ok := a.(*net.TCPAddr)
+	require.True(t, ok)
+	return uint32(ta.Port)
+}
+// newTestManager builds a Manager whose address answer is a fixed map.
+func newTestManager(t *testing.T, addrs map[string]string) *Manager {
+	t.Helper()
+	m := NewManager(func(vmID string) string { return addrs[vmID] })
+	t.Cleanup(m.StopAll)
+	return m
+}
+// boundPort reads the port the manager's listener for id actually bound. Host
+// port 0 in a desired spec lets the OS pick, which is how these tests avoid
+// racing for a fixed number.
+func boundPort(t *testing.T, m *Manager, id string) string {
+	t.Helper()
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	ex, ok := m.live[id]
+	require.True(t, ok, "no listener for %q", id)
+	require.NotNil(t, ex.ln, "listener for %q is not bound", id)
+	return ex.ln.Addr().String()
+}
+func desired(id, vmID string, guestPort, hostPort uint32) *pb.ExposureDesired {
+	return &pb.ExposureDesired{Id: id, VmId: vmID, GuestPort: guestPort, HostPort: hostPort, Protocol: "tcp"}
+}
+// speak dials addr, sends msg, and returns what came back.
+func speak(t *testing.T, addr, msg string) string {
+	t.Helper()
+	c, err := net.Dial("tcp", addr)
+	require.NoError(t, err)
+	defer c.Close()
+	require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
+	_, err = c.Write([]byte(msg))
+	require.NoError(t, err)
+	out, err := io.ReadAll(c)
+	require.NoError(t, err)
+	return string(out)
+}
+func TestConvergeBindsAndPipesToTheGuest(t *testing.T) {
+	g := newFakeGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": g.addr})
+	got := m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
+	require.Len(t, got, 1)
+	assert.Equal(t, "e1", got[0].GetId())
+	assert.Equal(t, "active", got[0].GetState())
+	assert.Equal(t, "echo:hello", speak(t, boundPort(t, m, "e1"), "hello"))
+}
+func TestConvergeIsIdempotent(t *testing.T) {
+	g := newFakeGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": g.addr})
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
+	addr := boundPort(t, m, "e1")
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
+	assert.Equal(t, addr, boundPort(t, m, "e1"), "an unchanged spec keeps its listener")
+	assert.Equal(t, "echo:hi", speak(t, addr, "hi"))
+}
+func TestConvergeClosesAVanishedExposure(t *testing.T) {
+	g := newFakeGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": g.addr})
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
+	addr := boundPort(t, m, "e1")
+	got := m.Converge(nil)
+	assert.Empty(t, got)
+	_, err := net.Dial("tcp", addr)
+	assert.Error(t, err, "a revoked exposure stops answering")
+}
+func TestConvergeRebindsAChangedSpec(t *testing.T) {
+	first := newFakeGuest(t)
+	second := newFakeGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": first.addr})
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", first.port, 0)})
+	oldAddr := boundPort(t, m, "e1")
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", second.port, 0)})
+	newAddr := boundPort(t, m, "e1")
+	assert.NotEqual(t, oldAddr, newAddr, "a changed spec is a close and a rebind")
+	// The new listener reaches the new guest port.
+	assert.Equal(t, "echo:x", speak(t, newAddr, "x"))
+}
+func TestConvergeReportsABindFailureAndHealsWhenThePortFrees(t *testing.T) {
+	g := newFakeGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": g.addr})
+	// Squat on a host port, then ask for exactly it.
+	squatter, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	held := portOf(t, squatter.Addr())
+	got := m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, held)})
+	require.Len(t, got, 1)
+	assert.Equal(t, "failed", got[0].GetState())
+	assert.NotEmpty(t, got[0].GetReason(), "the report carries what the OS said")
+	// The squatter exits; the very next converge binds. A bind-time refusal is
+	// not an error path, it is a level to converge on.
+	require.NoError(t, squatter.Close())
+	var state string
+	for i := 0; i < 20; i++ {
+		got = m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, held)})
+		state = got[0].GetState()
+		if state == "active" {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	assert.Equal(t, "active", state)
+	assert.Equal(t, "echo:back", speak(t, net.JoinHostPort("127.0.0.1", strconv.Itoa(int(held))), "back"))
+}
+func TestConvergeRebindsAListenerWhoseAcceptLoopDied(t *testing.T) {
+	g := newFakeGuest(t)
+	m := newTestManager(t, map[string]string{"vm1": g.addr})
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
+	// Break the listener behind the manager's back, the way fd exhaustion would
+	// end an accept loop: the manager never asked for this and still believes
+	// the exposure is served.
+	m.mu.Lock()
+	old := m.live["e1"].ln
+	require.NotNil(t, old)
+	require.NoError(t, old.Close())
+	m.mu.Unlock()
+	// The loop hands the listener back and records why it left. Both are read
+	// under the lock the loop writes them under, which is what makes the reason
+	// deterministic rather than a sampled guess.
+	var reason string
+	for i := 0; i < 200; i++ {
+		m.mu.Lock()
+		ex := m.live["e1"]
+		gone, r := ex.ln == nil, ex.reason
+		m.mu.Unlock()
+		if gone {
+			reason = r
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	assert.Contains(t, reason, "accept:", "a dead loop must not leave a listener nothing is serving")
+	// The bind-failure level-trigger is the whole healing mechanism: the next
+	// converge sees no listener and binds one. This exposure asked for host port
+	// 0, so that rebind cannot fail and no converge ever reports "failed" — the
+	// interim report is only observable when the fresh bind is also refused,
+	// which is the case TestConvergeReportsABindFailure... already covers.
+	got := m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", g.port, 0)})
+	require.Len(t, got, 1)
+	assert.Equal(t, "active", got[0].GetState())
+	m.mu.Lock()
+	rebound := m.live["e1"].ln
+	m.mu.Unlock()
+	// Compared as pointers, not with NotEqual: a deep-equality walk would read
+	// the live listener's internals while its accept loop is using them.
+	assert.True(t, rebound != old, "a dead listener is rebound, not believed")
+	assert.Equal(t, "echo:again", speak(t, boundPort(t, m, "e1"), "again"))
+}
+func TestConnectionClosesWhenTheGuestHasNoAddress(t *testing.T) {
+	m := newTestManager(t, map[string]string{}) // the guest is still leasing
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", 8080, 0)})
+	c, err := net.Dial("tcp", boundPort(t, m, "e1"))
+	require.NoError(t, err)
+	defer c.Close()
+	require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
+	out, err := io.ReadAll(c)
+	require.NoError(t, err)
+	assert.Empty(t, out, "no address yet means close, not wait")
+}
+func TestSpliceIsHalfCloseAware(t *testing.T) {
+	// A guest that answers only after the client has finished speaking — the
+	// shape every request/response protocol over a half-closed write takes.
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { ln.Close() })
+	go func() {
+		c, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer c.Close()
+		body, _ := io.ReadAll(c) // blocks until the client's write half closes
+		c.Write(append([]byte("saw:"), body...))
+	}()
+	m := newTestManager(t, map[string]string{"vm1": "127.0.0.1"})
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", portOf(t, ln.Addr()), 0)})
+	c, err := net.Dial("tcp", boundPort(t, m, "e1"))
+	require.NoError(t, err)
+	defer c.Close()
+	require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
+	_, err = c.Write([]byte("request"))
+	require.NoError(t, err)
+	require.NoError(t, c.(*net.TCPConn).CloseWrite())
+	out, err := io.ReadAll(c)
+	require.NoError(t, err)
+	assert.Equal(t, "saw:request", string(out))
+}
+func TestRevokeDrainsRatherThanCuts(t *testing.T) {
+	// A guest that echoes for as long as the client keeps speaking, so one
+	// connection can carry two exchanges either side of the revocation.
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+	t.Cleanup(func() { ln.Close() })
+	go func() {
+		for {
+			c, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func() {
+				defer c.Close()
+				_, _ = io.Copy(c, c)
+			}()
+		}
+	}()
+	m := newTestManager(t, map[string]string{"vm1": "127.0.0.1"})
+	m.Converge([]*pb.ExposureDesired{desired("e1", "vm1", portOf(t, ln.Addr()), 0)})
+	addr := boundPort(t, m, "e1")
+	c, err := net.Dial("tcp", addr)
+	require.NoError(t, err)
+	defer c.Close()
+	require.NoError(t, c.SetDeadline(time.Now().Add(5*time.Second)))
+	exchange := func(msg string) string {
+		t.Helper()
+		_, err := c.Write([]byte(msg))
+		require.NoError(t, err)
+		buf := make([]byte, len(msg))
+		_, err = io.ReadFull(c, buf)
+		require.NoError(t, err)
+		return string(buf)
+	}
+	require.Equal(t, "one", exchange("one"), "the connection is spliced through before anything is revoked")
+	m.Converge(nil)
+	_, err = net.Dial("tcp", addr)
+	require.Error(t, err, "a revoked exposure stops answering new dials")
+	assert.Equal(t, "two", exchange("two"), "closing a listener drains: only new dials die")
+}
+func TestStopAllClosesEveryListener(t *testing.T) {
+	g := newFakeGuest(t)
+	m := NewManager(func(string) string { return g.addr })
+	m.Converge([]*pb.ExposureDesired{
+		desired("e1", "vm1", g.port, 0),
+		desired("e2", "vm1", g.port, 0),
+	})
+	a1, a2 := boundPort(t, m, "e1"), boundPort(t, m, "e2")
+	m.StopAll()
+	for _, addr := range []string{a1, a2} {
+		_, err := net.Dial("tcp", addr)
+		assert.Error(t, err, "listener %s survived StopAll", addr)
+	}
+}

internal/agent/hostinfo/hostinfo.go

Old		New
@@ -7,6 +7,8 @@ package hostinfo
7	import (	7	import (
8	"context"	8	"context"
9	"encoding/binary"	9	"encoding/binary"
		10	"net"
		11	"net/netip"
10	"os"	12	"os"
11	"runtime"	13	"runtime"
12	"strconv"	14	"strconv"
@@ -67,6 +69,45 @@ func Metrics(stateDir string) *pb.HostMetrics { return readMetrics(stateDir) }
67	// per-platform.	69	// per-platform.
68	func BootID() string { return readBootID() }	70	func BootID() string { return readBootID() }
69		71
		72	// UplinkAddr reports the address this host presents on the network it reaches
		73	// the control plane over — the address to dial for a port published on this
		74	// host. serverAddr is the control plane's own host:port.
		75	//
		76	// It asks the kernel's routing table rather than picking an interface off a
		77	// list: a UDP "connection" sends no packet, but it binds the socket to the
		78	// source address the route to serverAddr would actually use, which is the one
		79	// answer that stays right on a host with several addresses.
		80	//
		81	// Empty means "no answer": the route could not be resolved, or it resolves to
		82	// an address nothing off this host can dial. The fleet then keeps whatever it
		83	// already knows rather than recording somewhere unreachable.
		84	func UplinkAddr(serverAddr string) string {
		85	conn, err := net.Dial("udp", serverAddr)
		86	if err != nil {
		87	return ""
		88	}
		89	defer conn.Close()
		90	return usableSourceAddr(conn.LocalAddr())
		91	}
		92
		93	// usableSourceAddr reduces a socket's local address to the address an operator
		94	// could dial, or "" when it is one nothing off this host can reach.
		95	func usableSourceAddr(a net.Addr) string {
		96	ua, ok := a.(*net.UDPAddr)
		97	if !ok {
		98	return ""
		99	}
		100	addr, ok := netip.AddrFromSlice(ua.IP)
		101	if !ok {
		102	return ""
		103	}
		104	addr = addr.Unmap()
		105	if addr.IsUnspecified() \|\| addr.IsLoopback() \|\| addr.IsLinkLocalUnicast() {
		106	return ""
		107	}
		108	return addr.String()
		109	}
		110
70	// Capacity returns the host's TOTAL capacity: total disk at stateDir, total	111	// Capacity returns the host's TOTAL capacity: total disk at stateDir, total
71	// memory, and CPU count. The server computes allocated/available by subtracting	112	// memory, and CPU count. The server computes allocated/available by subtracting
72	// the sum of live VM specs, so these must be totals, not free space.	113	// the sum of live VM specs, so these must be totals, not free space.

internal/agent/hostinfo/hostinfo_test.go

Old		New
@@ -3,6 +3,7 @@ package hostinfo
3	import (	3	import (
4	"context"	4	"context"
5	"encoding/hex"	5	"encoding/hex"
		6	"net"
6	"os"	7	"os"
7	"path/filepath"	8	"path/filepath"
8	"slices"	9	"slices"
@@ -237,3 +238,38 @@ func TestBootIDMissingFile(t *testing.T) {
237	// TestMetricsComputes lives in hostinfo_linux_test.go: it stubs	238	// TestMetricsComputes lives in hostinfo_linux_test.go: it stubs
238	// sysinfoFn/statfsFn with syscall.Sysinfo_t/Statfs_t literals, which only	239	// sysinfoFn/statfsFn with syscall.Sysinfo_t/Statfs_t literals, which only
239	// exist on Linux.	240	// exist on Linux.
		241
		242	func TestUsableSourceAddr(t *testing.T) {
		243	for _, tc := range []struct {
		244	name string
		245	addr net.Addr
		246	want string
		247	}{
		248	{"routable v4", &net.UDPAddr{IP: net.ParseIP("192.168.0.190")}, "192.168.0.190"},
		249	{"routable v6", &net.UDPAddr{IP: net.ParseIP("2001:db8::1")}, "2001:db8::1"},
		250	{"v4-mapped v6 is unmapped", &net.UDPAddr{IP: net.ParseIP("::ffff:10.0.0.7")}, "10.0.0.7"},
		251	{"loopback reaches nobody", &net.UDPAddr{IP: net.ParseIP("127.0.0.1")}, ""},
		252	{"unspecified reaches nobody", &net.UDPAddr{IP: net.IPv4zero}, ""},
		253	{"link-local reaches nobody", &net.UDPAddr{IP: net.ParseIP("169.254.1.5")}, ""},
		254	{"not a udp address", &net.TCPAddr{IP: net.ParseIP("192.168.0.190")}, ""},
		255	{"no address at all", &net.UDPAddr{}, ""},
		256	} {
		257	t.Run(tc.name, func(t *testing.T) {
		258	if got := usableSourceAddr(tc.addr); got != tc.want {
		259	t.Errorf("usableSourceAddr(%v) = %q, want %q", tc.addr, got, tc.want)
		260	}
		261	})
		262	}
		263	}
		264
		265	func TestUplinkAddr(t *testing.T) {
		266	// Nowhere to route to: no answer, never a guess.
		267	if got := UplinkAddr("not-an-address"); got != "" {
		268	t.Errorf("UplinkAddr(garbage) = %q, want empty", got)
		269	}
		270	// A route that resolves to loopback names an address nothing else can
		271	// dial, so the agent says nothing rather than sending it.
		272	if got := UplinkAddr("127.0.0.1:9"); got != "" {
		273	t.Errorf("UplinkAddr(loopback) = %q, want empty", got)
		274	}
		275	}

internal/agent/run/cli.go

Old		New
@@ -14,6 +14,7 @@ import (
14	"flag"	14	"flag"
15	"fmt"	15	"fmt"
16	"log/slog"	16	"log/slog"
		17	"net"
17	"os"	18	"os"
18	"os/exec"	19	"os/exec"
19	"os/signal"	20	"os/signal"
@@ -22,6 +23,7 @@ import (
22	"time"	23	"time"
23		24
24	"github.com/a73x/eitri/internal/agent/enrollclient"	25	"github.com/a73x/eitri/internal/agent/enrollclient"
		26	"github.com/a73x/eitri/internal/agent/exposeproxy"
25	"github.com/a73x/eitri/internal/agent/hostinfo"	27	"github.com/a73x/eitri/internal/agent/hostinfo"
26	"github.com/a73x/eitri/internal/agent/imagecache"	28	"github.com/a73x/eitri/internal/agent/imagecache"
27	"github.com/a73x/eitri/internal/agent/reconcile"	29	"github.com/a73x/eitri/internal/agent/reconcile"
@@ -285,6 +287,22 @@ func serve(st *state.Store, cfg Config) error {
285	// is cancelled (see the note below where pumps are torn down). The process is	287	// is cancelled (see the note below where pumps are torn down). The process is
286	// exiting; the OS reclaims the goroutines.	288	// exiting; the OS reclaims the goroutines.
287		289
		290	// The proxy publishing this host's guest ports. It is built here, in the
		291	// composition root, because it needs one thing from the agent and nothing
		292	// from any backend: where a guest is right now.
		293	proxy := exposeproxy.NewManager(func(vmID string) string { return guestAddr(st, vmID) })
		294
		295	// The control plane's address, resolved once: the uplink address is asked on
		296	// every report, and a per-tick resolution would put DNS on the report path,
		297	// where a slow resolver stalls the session's writer. Only the route lookup
		298	// uses this — the sync connection dials the configured address itself.
		299	uplinkVia := id.ServerQUICAddr
		300	if ua, err := net.ResolveUDPAddr("udp", uplinkVia); err == nil {
		301	uplinkVia = ua.String()
		302	} else {
		303	slog.Warn("uplink target unresolved; per-report DNS lookups", "addr", uplinkVia, "err", err)
		304	}
		305
288	client := &syncclient.Client{	306	client := &syncclient.Client{
289	Engine: engine,	307	Engine: engine,
290	St: st,	308	St: st,
@@ -294,6 +312,8 @@ func serve(st *state.Store, cfg Config) error {
294	GuestCIDR: plat.GuestCIDR,	312	GuestCIDR: plat.GuestCIDR,
295	Runner: hostRunner,	313	Runner: hostRunner,
296	Console: pumps,	314	Console: pumps,
		315	Exposures: proxy,
		316	UplinkAddr: func() string { return hostinfo.UplinkAddr(uplinkVia) },
297	MaxVCPUs: cfg.MaxVCPUs,	317	MaxVCPUs: cfg.MaxVCPUs,
298	MaxMemMB: cfg.MaxMemMB,	318	MaxMemMB: cfg.MaxMemMB,
299	MaxDiskGB: cfg.MaxDiskGB,	319	MaxDiskGB: cfg.MaxDiskGB,
@@ -307,7 +327,9 @@ func serve(st *state.Store, cfg Config) error {
307	// Ensure/Stop calls are expected to reach pumps (an unjoined session	327	// Ensure/Stop calls are expected to reach pumps (an unjoined session
308	// worker could in principle still be mid-Engine.Step, but it has nothing	328	// worker could in principle still be mid-Engine.Step, but it has nothing
309	// left to drive once client.Run has returned). Tear down every serial	329	// left to drive once client.Run has returned). Tear down every serial
310	// console pump here, at the very end of agent shutdown.	330	// console pump and every published listener here, at the very end of agent
		331	// shutdown — a port this host no longer serves must not stay bound.
311	pumps.StopAll()	332	pumps.StopAll()
		333	proxy.StopAll()
312	return nil	334	return nil
313	}	335	}

internal/agent/run/exposures.go

@@ @@ -0,0 +1,21 @@ @@
+package run
+import "github.com/a73x/eitri/internal/agent/state"
+// guestAddr answers a VM's current address from this agent's own records —
+// exactly the answer the SSH tunnel dials, so a published port and a tunnelled
+// one can never disagree about where a guest is. Empty means "no answer": the
+// VM is not on this host, its record cannot be read, or its guest has not been
+// given an address yet.
+//
+// It is asked per connection rather than once at bind, which is what makes an
+// exposure created while its VM is still imaging simply start working when the
+// guest comes up. Safe to call concurrently: a *state.Store is immutable after
+// Open and Get only reads a record file that is written atomically.
+func guestAddr(st *state.Store, vmID string) string {
+	rec, ok, err := st.Get(vmID)
+	if err != nil || !ok {
+		return ""
+	}
+	return rec.IP
+}

internal/agent/run/exposures_test.go

@@ @@ -0,0 +1,32 @@ @@
+package run
+import (
+	"testing"
+	"github.com/a73x/eitri/internal/agent/state"
+	"github.com/stretchr/testify/require"
+)
+// TestGuestAddrAnswersFromTheAgentsOwnRecords pins that the proxy dials the
+// same address the tunnel does — the one reconcile keeps current on each VM's
+// record — rather than a second, independently drifting answer.
+func TestGuestAddrAnswersFromTheAgentsOwnRecords(t *testing.T) {
+	st, err := state.Open(t.TempDir())
+	require.NoError(t, err)
+	// A VM this host has never heard of has no address.
+	if got := guestAddr(st, "vm1"); got != "" {
+		t.Errorf("guestAddr(unknown) = %q, want empty", got)
+	}
+	require.NoError(t, st.SaveVM(state.Record{Spec: state.VMSpec{VMID: "vm1"}, IP: "10.77.1.5"}))
+	if got := guestAddr(st, "vm1"); got != "10.77.1.5" {
+		t.Errorf("guestAddr(vm1) = %q, want 10.77.1.5", got)
+	}
+	// A record with no address yet is "no answer", not an address of "".
+	require.NoError(t, st.SaveVM(state.Record{Spec: state.VMSpec{VMID: "vm2"}}))
+	if got := guestAddr(st, "vm2"); got != "" {
+		t.Errorf("guestAddr(vm2) = %q, want empty", got)
+	}
+}

internal/agent/syncclient/client.go

Old		New
@@ -42,6 +42,14 @@ type Console interface {
42	Attach(ctx context.Context, vmID string, rw io.ReadWriter, onReady func() error) error	42	Attach(ctx context.Context, vmID string, rw io.ReadWriter, onReady func() error) error
43	}	43	}
44		44
		45	// Exposures converges this host's published guest ports and reports what each
		46	// listener is doing (consumer-owned; the concrete implementation is
		47	// *exposeproxy.Manager). A nil Exposures converges nothing and reports nothing
		48	// — an agent with no proxy publishes no ports, and says so by saying nothing.
		49	type Exposures interface {
		50	Converge(desired []pb.ExposureDesired) []pb.ExposureActual
		51	}
		52
45	// DefaultTickInterval is the fallback report/reconcile cadence when	53	// DefaultTickInterval is the fallback report/reconcile cadence when
46	// Client.TickInterval is zero. It is COUPLED to the server's	54	// Client.TickInterval is zero. It is COUPLED to the server's
47	// registry.OnlineWindow (30s): the server marks a host offline after	55	// registry.OnlineWindow (30s): the server marks a host offline after
@@ -76,6 +84,13 @@ type Client struct {
76	// Console handles server-opened console streams (nil refuses them all).	84	// Console handles server-opened console streams (nil refuses them all).
77	Console Console	85	Console Console
78		86
		87	// Exposures publishes this host's guest ports (nil publishes none).
		88	Exposures Exposures
		89
		90	// UplinkAddr reports the address this host answers on, asked once per
		91	// report like GuestCIDR. Nil, or an empty return, means "no answer".
		92	UplinkAddr func() string
		93
79	// dialGuest connects to a VM's ssh port; overridable in tests. nil uses the	94	// dialGuest connects to a VM's ssh port; overridable in tests. nil uses the
80	// production dialer, which pins the guest port to 22 — the wire port is	95	// production dialer, which pins the guest port to 22 — the wire port is
81	// validated but never dialed, so a compromised control plane cannot redirect	96	// validated but never dialed, so a compromised control plane cannot redirect
@@ -195,6 +210,28 @@ func (c Client) maybeUpgrade(ctx context.Context, up pb.AgentUpgrade) {
195	}()	210	}()
196	}	211	}
197		212
		213	// convergeExposures drives this host's listeners toward the snapshot and
		214	// returns the rows the report carries. Level-triggered like everything else:
		215	// every snapshot re-converges, so a bind that lost its port to a squatting
		216	// process is retried on the next tick.
		217	func (c Client) convergeExposures(snap pb.DesiredStateSnapshot) []*pb.ExposureActual {
		218	if c.Exposures == nil {
		219	return nil
		220	}
		221	return c.Exposures.Converge(snap.GetExposures())
		222	}
		223
		224	// reportExposures converges only against a snapshot the engine ACCEPTED. A
		225	// fenced snapshot is one this host has already moved past, and driving
		226	// listeners from it would re-open a port the fleet has since revoked — the
		227	// same reason the fence path touches nothing else.
		228	func (c Client) reportExposures(snap pb.DesiredStateSnapshot, rep pb.ActualStateReport) []pb.ExposureActual {
		229	if rep.GetFenceViolation() {
		230	return nil
		231	}
		232	return c.convergeExposures(snap)
		233	}
		234
198	// errPermanentAuth marks a credential rejection so Run() backs off long instead	235	// errPermanentAuth marks a credential rejection so Run() backs off long instead
199	// of tight-looping a dead credential.	236	// of tight-looping a dead credential.
200	var errPermanentAuth = errors.New("auth rejected (permanent)")	237	var errPermanentAuth = errors.New("auth rejected (permanent)")
@@ -361,6 +398,13 @@ func (c *Client) session(ctx context.Context) error {
361	rep := c.Engine.Step(ctx, snap)	398	rep := c.Engine.Step(ctx, snap)
362	rep.Capacity = c.advertisedCapacity(stateDir)	399	rep.Capacity = c.advertisedCapacity(stateDir)
363	rep.Metrics = hostinfo.Metrics(stateDir)	400	rep.Metrics = hostinfo.Metrics(stateDir)
		401	rep.Exposures = c.reportExposures(snap, rep)
		402	// Polled, not resolved once: a host's address can change under a
		403	// live agent (a lease renews on a different address). Empty is "no
		404	// answer" and leaves the fleet's record alone.
		405	if c.UplinkAddr != nil {
		406	rep.HostUplinkAddr = c.UplinkAddr()
		407	}
364	// Polled, not resolved once: a host's guest subnet can change while the	408	// Polled, not resolved once: a host's guest subnet can change while the
365	// agent stays connected, and on a platform whose OS owns the network it	409	// agent stays connected, and on a platform whose OS owns the network it
366	// may not be knowable at connect time at all. Empty is "no answer" and	410	// may not be knowable at connect time at all. Empty is "no answer" and

internal/agent/syncclient/exposures_test.go

@@ @@ -0,0 +1,58 @@ @@
+package syncclient
+import (
+	"testing"
+	"github.com/a73x/eitri/internal/pb"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// fakeExposures records what it was asked to converge and answers with a fixed
+// report.
+type fakeExposures struct {
+	sawIDs []string
+	calls  int
+	out    []*pb.ExposureActual
+}
+func (f *fakeExposures) Converge(desired []*pb.ExposureDesired) []*pb.ExposureActual {
+	f.calls++
+	f.sawIDs = nil
+	for _, d := range desired {
+		f.sawIDs = append(f.sawIDs, d.GetId())
+	}
+	return f.out
+}
+func TestConvergeExposuresPassesTheSnapshotThrough(t *testing.T) {
+	fe := &fakeExposures{out: []*pb.ExposureActual{{Id: "e1", State: "active"}}}
+	c := &Client{Exposures: fe}
+	got := c.convergeExposures(&pb.DesiredStateSnapshot{
+		Exposures: []*pb.ExposureDesired{{Id: "e1", VmId: "vm1", GuestPort: 8080, HostPort: 30080}},
+	})
+	assert.Equal(t, []string{"e1"}, fe.sawIDs)
+	require.Len(t, got, 1)
+	assert.Equal(t, "active", got[0].GetState())
+}
+func TestConvergeExposuresWithoutAProxyConvergesNothing(t *testing.T) {
+	c := &Client{}
+	assert.Nil(t, c.convergeExposures(&pb.DesiredStateSnapshot{
+		Exposures: []*pb.ExposureDesired{{Id: "e1"}},
+	}), "an agent with no proxy publishes nothing, and says so by saying nothing")
+}
+func TestConvergeExposuresRefusesAFencedSnapshot(t *testing.T) {
+	fe := &fakeExposures{}
+	c := &Client{Exposures: fe}
+	got := c.reportExposures(&pb.DesiredStateSnapshot{
+		Exposures: []*pb.ExposureDesired{{Id: "e1"}},
+	}, &pb.ActualStateReport{FenceViolation: true})
+	assert.Zero(t, fe.calls, "a snapshot the engine refused must not drive the listeners")
+	assert.Nil(t, got)
+}

internal/pb/sync.pb.go

Old		New
@@ -761,9 +761,17 @@ type ActualStateReport struct {
761	// connected (an admin edits the host's network), and on some platforms is not	761	// connected (an admin edits the host's network), and on some platforms is not
762	// knowable at connect time at all. Capacity is in both messages for the same	762	// knowable at connect time at all. Capacity is in both messages for the same
763	// reason: an opening value, then the ongoing truth.	763	// reason: an opening value, then the ongoing truth.
764	GuestCidr string `protobuf:"bytes,8,opt,name=guest_cidr,json=guestCidr,proto3" json:"guest_cidr,omitempty"`	764	GuestCidr string `protobuf:"bytes,8,opt,name=guest_cidr,json=guestCidr,proto3" json:"guest_cidr,omitempty"`
765	unknownFields protoimpl.UnknownFields	765	Exposures []*ExposureActual `protobuf:"bytes,9,rep,name=exposures,proto3" json:"exposures,omitempty"`
766	sizeCache protoimpl.SizeCache	766	// The address this host presents on the network it reaches the control
		767	// plane over — the address an operator dials to reach a published guest
		768	// port. Empty means "not yet known", never "no address": a host that cannot
		769	// answer leaves the fleet's record alone rather than erasing it. In the
		770	// report rather than Hello for the same reason guest_cidr is: it can change
		771	// while an agent stays connected.
		772	HostUplinkAddr string `protobuf:"bytes,10,opt,name=host_uplink_addr,json=hostUplinkAddr,proto3" json:"host_uplink_addr,omitempty"`
		773	unknownFields protoimpl.UnknownFields
		774	sizeCache protoimpl.SizeCache
767	}	775	}
768		776
769	func (x *ActualStateReport) Reset() {	777	func (x *ActualStateReport) Reset() {
@@ -852,6 +860,20 @@ func (x *ActualStateReport) GetGuestCidr() string {
852	return ""	860	return ""
853	}	861	}
854		862
		863	func (x ActualStateReport) GetExposures() []ExposureActual {
		864	if x != nil {
		865	return x.Exposures
		866	}
		867	return nil
		868	}
		869
		870	func (x *ActualStateReport) GetHostUplinkAddr() string {
		871	if x != nil {
		872	return x.HostUplinkAddr
		873	}
		874	return ""
		875	}
		876
855	type VMDesired struct {	877	type VMDesired struct {
856	state protoimpl.MessageState `protogen:"open.v1"`	878	state protoimpl.MessageState `protogen:"open.v1"`
857	VmId string `protobuf:"bytes,1,opt,name=vm_id,json=vmId,proto3" json:"vm_id,omitempty"`	879	VmId string `protobuf:"bytes,1,opt,name=vm_id,json=vmId,proto3" json:"vm_id,omitempty"`
@@ -1013,6 +1035,7 @@ type DesiredStateSnapshot struct {
1013	Epoch uint64 `protobuf:"varint,1,opt,name=epoch,proto3" json:"epoch,omitempty"` // agents refuse epoch < highest seen	1035	Epoch uint64 `protobuf:"varint,1,opt,name=epoch,proto3" json:"epoch,omitempty"` // agents refuse epoch < highest seen
1014	Vms []*VMDesired `protobuf:"bytes,2,rep,name=vms,proto3" json:"vms,omitempty"` // FULL set for this host, including tombstoned	1036	Vms []*VMDesired `protobuf:"bytes,2,rep,name=vms,proto3" json:"vms,omitempty"` // FULL set for this host, including tombstoned
1015	AgentUpgrade *AgentUpgrade `protobuf:"bytes,3,opt,name=agent_upgrade,json=agentUpgrade,proto3" json:"agent_upgrade,omitempty"` // optional operator-initiated agent self-upgrade	1037	AgentUpgrade *AgentUpgrade `protobuf:"bytes,3,opt,name=agent_upgrade,json=agentUpgrade,proto3" json:"agent_upgrade,omitempty"` // optional operator-initiated agent self-upgrade
		1038	Exposures []*ExposureDesired `protobuf:"bytes,4,rep,name=exposures,proto3" json:"exposures,omitempty"` // FULL set for this host
1016	unknownFields protoimpl.UnknownFields	1039	unknownFields protoimpl.UnknownFields
1017	sizeCache protoimpl.SizeCache	1040	sizeCache protoimpl.SizeCache
1018	}	1041	}
@@ -1068,6 +1091,13 @@ func (x DesiredStateSnapshot) GetAgentUpgrade() AgentUpgrade {
1068	return nil	1091	return nil
1069	}	1092	}
1070		1093
		1094	func (x DesiredStateSnapshot) GetExposures() []ExposureDesired {
		1095	if x != nil {
		1096	return x.Exposures
		1097	}
		1098	return nil
		1099	}
		1100
1071	// AgentUpgrade asks the agent to replace its own binary: download url, verify	1101	// AgentUpgrade asks the agent to replace its own binary: download url, verify
1072	// sha256, swap atomically (keeping .prev), re-exec. Present only on hosts an	1102	// sha256, swap atomically (keeping .prev), re-exec. Present only on hosts an
1073	// operator explicitly clicked; absent otherwise. An agent already running	1103	// operator explicitly clicked; absent otherwise. An agent already running
@@ -1344,6 +1374,152 @@ func (x *TCPOpened) GetError() string {
1344	return ""	1374	return ""
1345	}	1375	}
1346		1376
		1377	// ExposureDesired is one published guest port a host should be serving: bind
		1378	// host_port on the host, pipe every accepted connection to guest_port inside
		1379	// the guest. It rides the snapshot at TOP LEVEL rather than nested in
		1380	// VMDesired, because exposures are their own objects converging on their own
		1381	// cadence — an exposure can be created while its VM is still imaging, and it
		1382	// binds immediately.
		1383	type ExposureDesired struct {
		1384	state protoimpl.MessageState `protogen:"open.v1"`
		1385	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
		1386	VmId string `protobuf:"bytes,2,opt,name=vm_id,json=vmId,proto3" json:"vm_id,omitempty"`
		1387	GuestPort uint32 `protobuf:"varint,3,opt,name=guest_port,json=guestPort,proto3" json:"guest_port,omitempty"`
		1388	HostPort uint32 `protobuf:"varint,4,opt,name=host_port,json=hostPort,proto3" json:"host_port,omitempty"`
		1389	Protocol string `protobuf:"bytes,5,opt,name=protocol,proto3" json:"protocol,omitempty"` // "tcp"
		1390	unknownFields protoimpl.UnknownFields
		1391	sizeCache protoimpl.SizeCache
		1392	}
		1393
		1394	func (x *ExposureDesired) Reset() {
		1395	*x = ExposureDesired{}
		1396	mi := &file_proto_eitri_v1_sync_proto_msgTypes[16]
		1397	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
		1398	ms.StoreMessageInfo(mi)
		1399	}
		1400
		1401	func (x *ExposureDesired) String() string {
		1402	return protoimpl.X.MessageStringOf(x)
		1403	}
		1404
		1405	func (*ExposureDesired) ProtoMessage() {}
		1406
		1407	func (x *ExposureDesired) ProtoReflect() protoreflect.Message {
		1408	mi := &file_proto_eitri_v1_sync_proto_msgTypes[16]
		1409	if x != nil {
		1410	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
		1411	if ms.LoadMessageInfo() == nil {
		1412	ms.StoreMessageInfo(mi)
		1413	}
		1414	return ms
		1415	}
		1416	return mi.MessageOf(x)
		1417	}
		1418
		1419	// Deprecated: Use ExposureDesired.ProtoReflect.Descriptor instead.
		1420	func (*ExposureDesired) Descriptor() ([]byte, []int) {
		1421	return file_proto_eitri_v1_sync_proto_rawDescGZIP(), []int{16}
		1422	}
		1423
		1424	func (x *ExposureDesired) GetId() string {
		1425	if x != nil {
		1426	return x.Id
		1427	}
		1428	return ""
		1429	}
		1430
		1431	func (x *ExposureDesired) GetVmId() string {
		1432	if x != nil {
		1433	return x.VmId
		1434	}
		1435	return ""
		1436	}
		1437
		1438	func (x *ExposureDesired) GetGuestPort() uint32 {
		1439	if x != nil {
		1440	return x.GuestPort
		1441	}
		1442	return 0
		1443	}
		1444
		1445	func (x *ExposureDesired) GetHostPort() uint32 {
		1446	if x != nil {
		1447	return x.HostPort
		1448	}
		1449	return 0
		1450	}
		1451
		1452	func (x *ExposureDesired) GetProtocol() string {
		1453	if x != nil {
		1454	return x.Protocol
		1455	}
		1456	return ""
		1457	}
		1458
		1459	// ExposureActual is one exposure's state as its host observes it: "active"
		1460	// once the host listener is bound, "failed" with the OS error otherwise.
		1461	// "active" means the HOST half of the pipe exists — whether anything answers
		1462	// inside the guest is the guest's half, and this does not pretend otherwise.
		1463	type ExposureActual struct {
		1464	state protoimpl.MessageState `protogen:"open.v1"`
		1465	Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"`
		1466	State string `protobuf:"bytes,2,opt,name=state,proto3" json:"state,omitempty"` // "active"\|"failed"
		1467	Reason string `protobuf:"bytes,3,opt,name=reason,proto3" json:"reason,omitempty"` // the OS error, when failed
		1468	unknownFields protoimpl.UnknownFields
		1469	sizeCache protoimpl.SizeCache
		1470	}
		1471
		1472	func (x *ExposureActual) Reset() {
		1473	*x = ExposureActual{}
		1474	mi := &file_proto_eitri_v1_sync_proto_msgTypes[17]
		1475	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
		1476	ms.StoreMessageInfo(mi)
		1477	}
		1478
		1479	func (x *ExposureActual) String() string {
		1480	return protoimpl.X.MessageStringOf(x)
		1481	}
		1482
		1483	func (*ExposureActual) ProtoMessage() {}
		1484
		1485	func (x *ExposureActual) ProtoReflect() protoreflect.Message {
		1486	mi := &file_proto_eitri_v1_sync_proto_msgTypes[17]
		1487	if x != nil {
		1488	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
		1489	if ms.LoadMessageInfo() == nil {
		1490	ms.StoreMessageInfo(mi)
		1491	}
		1492	return ms
		1493	}
		1494	return mi.MessageOf(x)
		1495	}
		1496
		1497	// Deprecated: Use ExposureActual.ProtoReflect.Descriptor instead.
		1498	func (*ExposureActual) Descriptor() ([]byte, []int) {
		1499	return file_proto_eitri_v1_sync_proto_rawDescGZIP(), []int{17}
		1500	}
		1501
		1502	func (x *ExposureActual) GetId() string {
		1503	if x != nil {
		1504	return x.Id
		1505	}
		1506	return ""
		1507	}
		1508
		1509	func (x *ExposureActual) GetState() string {
		1510	if x != nil {
		1511	return x.State
		1512	}
		1513	return ""
		1514	}
		1515
		1516	func (x *ExposureActual) GetReason() string {
		1517	if x != nil {
		1518	return x.Reason
		1519	}
		1520	return ""
		1521	}
		1522
1347	var File_proto_eitri_v1_sync_proto protoreflect.FileDescriptor	1523	var File_proto_eitri_v1_sync_proto protoreflect.FileDescriptor
1348		1524
1349	const file_proto_eitri_v1_sync_proto_rawDesc = "" +	1525	const file_proto_eitri_v1_sync_proto_rawDesc = "" +
@@ -1410,7 +1586,7 @@ const file_proto_eitri_v1_sync_proto_rawDesc = "" +
1410	"\x04name\x18\x02 \x01(\tR\x04name\x12\x1f\n" +	1586	"\x04name\x18\x02 \x01(\tR\x04name\x12\x1f\n" +
1411	"\vvmspec_json\x18\x03 \x01(\fR\n" +	1587	"\vvmspec_json\x18\x03 \x01(\fR\n" +
1412	"vmspecJson\x12&\n" +	1588	"vmspecJson\x12&\n" +
1413	"\x0fdestroy_at_unix\x18\x04 \x01(\x03R\rdestroyAtUnix\"\xe3\x02\n" +	1589	"\x0fdestroy_at_unix\x18\x04 \x01(\x03R\rdestroyAtUnix\"\xc5\x03\n" +
1414	"\x11ActualStateReport\x12$\n" +	1590	"\x11ActualStateReport\x12$\n" +
1415	"\x03vms\x18\x01 \x03(\v2\x12.eitri.v1.ActualVMR\x03vms\x12\x1c\n" +	1591	"\x03vms\x18\x01 \x03(\v2\x12.eitri.v1.ActualVMR\x03vms\x12\x1c\n" +
1416	"\tdestroyed\x18\x02 \x03(\tR\tdestroyed\x129\n" +	1592	"\tdestroyed\x18\x02 \x03(\tR\tdestroyed\x129\n" +
@@ -1420,7 +1596,10 @@ const file_proto_eitri_v1_sync_proto_rawDesc = "" +
1420	"\x0flast_seen_epoch\x18\x06 \x01(\x04R\rlastSeenEpoch\x12/\n" +	1596	"\x0flast_seen_epoch\x18\x06 \x01(\x04R\rlastSeenEpoch\x12/\n" +
1421	"\ametrics\x18\a \x01(\v2\x15.eitri.v1.HostMetricsR\ametrics\x12\x1d\n" +	1597	"\ametrics\x18\a \x01(\v2\x15.eitri.v1.HostMetricsR\ametrics\x12\x1d\n" +
1422	"\n" +	1598	"\n" +
1423	"guest_cidr\x18\b \x01(\tR\tguestCidr\"\x85\x04\n" +	1599	"guest_cidr\x18\b \x01(\tR\tguestCidr\x126\n" +
		1600	"\texposures\x18\t \x03(\v2\x18.eitri.v1.ExposureActualR\texposures\x12(\n" +
		1601	"\x10host_uplink_addr\x18\n" +
		1602	" \x01(\tR\x0ehostUplinkAddr\"\x85\x04\n" +
1424	"\tVMDesired\x12\x13\n" +	1603	"\tVMDesired\x12\x13\n" +
1425	"\x05vm_id\x18\x01 \x01(\tR\x04vmId\x12\x12\n" +	1604	"\x05vm_id\x18\x01 \x01(\tR\x04vmId\x12\x12\n" +
1426	"\x04name\x18\x02 \x01(\tR\x04name\x12\x1b\n" +	1605	"\x04name\x18\x02 \x01(\tR\x04name\x12\x1b\n" +
@@ -1443,11 +1622,12 @@ const file_proto_eitri_v1_sync_proto_rawDesc = "" +
1443	"\x12ssh_authorized_key\x18\f \x01(\tR\x10sshAuthorizedKey\x12'\n" +	1622	"\x12ssh_authorized_key\x18\f \x01(\tR\x10sshAuthorizedKey\x12'\n" +
1444	"\x10ssh_host_key_pem\x18\x10 \x01(\tR\rsshHostKeyPem\x12\"\n" +	1623	"\x10ssh_host_key_pem\x18\x10 \x01(\tR\rsshHostKeyPem\x12\"\n" +
1445	"\rssh_host_cert\x18\x11 \x01(\tR\vsshHostCert\x12<\n" +	1624	"\rssh_host_cert\x18\x11 \x01(\tR\vsshHostCert\x12<\n" +
1446	"\x1bssh_user_ca_authorized_keys\x18\x12 \x03(\tR\x17sshUserCaAuthorizedKeysJ\x04\b\r\x10\x0eJ\x04\b\x0e\x10\x0fJ\x04\b\x0f\x10\x10\"\x90\x01\n" +	1625	"\x1bssh_user_ca_authorized_keys\x18\x12 \x03(\tR\x17sshUserCaAuthorizedKeysJ\x04\b\r\x10\x0eJ\x04\b\x0e\x10\x0fJ\x04\b\x0f\x10\x10\"\xc9\x01\n" +
1447	"\x14DesiredStateSnapshot\x12\x14\n" +	1626	"\x14DesiredStateSnapshot\x12\x14\n" +
1448	"\x05epoch\x18\x01 \x01(\x04R\x05epoch\x12%\n" +	1627	"\x05epoch\x18\x01 \x01(\x04R\x05epoch\x12%\n" +
1449	"\x03vms\x18\x02 \x03(\v2\x13.eitri.v1.VMDesiredR\x03vms\x12;\n" +	1628	"\x03vms\x18\x02 \x03(\v2\x13.eitri.v1.VMDesiredR\x03vms\x12;\n" +
1450	"\ragent_upgrade\x18\x03 \x01(\v2\x16.eitri.v1.AgentUpgradeR\fagentUpgrade\"R\n" +	1629	"\ragent_upgrade\x18\x03 \x01(\v2\x16.eitri.v1.AgentUpgradeR\fagentUpgrade\x127\n" +
		1630	"\texposures\x18\x04 \x03(\v2\x19.eitri.v1.ExposureDesiredR\texposures\"R\n" +
1451	"\fAgentUpgrade\x12\x18\n" +	1631	"\fAgentUpgrade\x12\x18\n" +
1452	"\aversion\x18\x01 \x01(\tR\aversion\x12\x10\n" +	1632	"\aversion\x18\x01 \x01(\tR\aversion\x12\x10\n" +
1453	"\x03url\x18\x02 \x01(\tR\x03url\x12\x16\n" +	1633	"\x03url\x18\x02 \x01(\tR\x03url\x12\x16\n" +
@@ -1462,7 +1642,18 @@ const file_proto_eitri_v1_sync_proto_rawDesc = "" +
1462	"\x04port\x18\x02 \x01(\rR\x04port\"1\n" +	1642	"\x04port\x18\x02 \x01(\rR\x04port\"1\n" +
1463	"\tTCPOpened\x12\x0e\n" +	1643	"\tTCPOpened\x12\x0e\n" +
1464	"\x02ok\x18\x01 \x01(\bR\x02ok\x12\x14\n" +	1644	"\x02ok\x18\x01 \x01(\bR\x02ok\x12\x14\n" +
1465	"\x05error\x18\x02 \x01(\tR\x05errorB&Z$github.com/a73x/eitri/internal/pb;pbb\x06proto3"	1645	"\x05error\x18\x02 \x01(\tR\x05error\"\x8e\x01\n" +
		1646	"\x0fExposureDesired\x12\x0e\n" +
		1647	"\x02id\x18\x01 \x01(\tR\x02id\x12\x13\n" +
		1648	"\x05vm_id\x18\x02 \x01(\tR\x04vmId\x12\x1d\n" +
		1649	"\n" +
		1650	"guest_port\x18\x03 \x01(\rR\tguestPort\x12\x1b\n" +
		1651	"\thost_port\x18\x04 \x01(\rR\bhostPort\x12\x1a\n" +
		1652	"\bprotocol\x18\x05 \x01(\tR\bprotocol\"N\n" +
		1653	"\x0eExposureActual\x12\x0e\n" +
		1654	"\x02id\x18\x01 \x01(\tR\x02id\x12\x14\n" +
		1655	"\x05state\x18\x02 \x01(\tR\x05state\x12\x16\n" +
		1656	"\x06reason\x18\x03 \x01(\tR\x06reasonB&Z$github.com/a73x/eitri/internal/pb;pbb\x06proto3"
1466		1657
1467	var (	1658	var (
1468	file_proto_eitri_v1_sync_proto_rawDescOnce sync.Once	1659	file_proto_eitri_v1_sync_proto_rawDescOnce sync.Once
@@ -1476,7 +1667,7 @@ func file_proto_eitri_v1_sync_proto_rawDescGZIP() []byte {
1476	return file_proto_eitri_v1_sync_proto_rawDescData	1667	return file_proto_eitri_v1_sync_proto_rawDescData
1477	}	1668	}
1478		1669
1479	var file_proto_eitri_v1_sync_proto_msgTypes = make([]protoimpl.MessageInfo, 16)	1670	var file_proto_eitri_v1_sync_proto_msgTypes = make([]protoimpl.MessageInfo, 18)
1480	var file_proto_eitri_v1_sync_proto_goTypes = []any{	1671	var file_proto_eitri_v1_sync_proto_goTypes = []any{
1481	(*AgentMessage)(nil), // 0: eitri.v1.AgentMessage	1672	(*AgentMessage)(nil), // 0: eitri.v1.AgentMessage
1482	(*ServerMessage)(nil), // 1: eitri.v1.ServerMessage	1673	(*ServerMessage)(nil), // 1: eitri.v1.ServerMessage
@@ -1494,6 +1685,8 @@ var file_proto_eitri_v1_sync_proto_goTypes = []any{
1494	(*ConsoleOpened)(nil), // 13: eitri.v1.ConsoleOpened	1685	(*ConsoleOpened)(nil), // 13: eitri.v1.ConsoleOpened
1495	(*TCPOpen)(nil), // 14: eitri.v1.TCPOpen	1686	(*TCPOpen)(nil), // 14: eitri.v1.TCPOpen
1496	(*TCPOpened)(nil), // 15: eitri.v1.TCPOpened	1687	(*TCPOpened)(nil), // 15: eitri.v1.TCPOpened
		1688	(*ExposureDesired)(nil), // 16: eitri.v1.ExposureDesired
		1689	(*ExposureActual)(nil), // 17: eitri.v1.ExposureActual
1497	}	1690	}
1498	var file_proto_eitri_v1_sync_proto_depIdxs = []int32{	1691	var file_proto_eitri_v1_sync_proto_depIdxs = []int32{
1499	2, // 0: eitri.v1.AgentMessage.hello:type_name -> eitri.v1.Hello	1692	2, // 0: eitri.v1.AgentMessage.hello:type_name -> eitri.v1.Hello
@@ -1509,13 +1702,15 @@ var file_proto_eitri_v1_sync_proto_depIdxs = []int32{
1509	7, // 10: eitri.v1.ActualStateReport.quarantined:type_name -> eitri.v1.QuarantinedVM	1702	7, // 10: eitri.v1.ActualStateReport.quarantined:type_name -> eitri.v1.QuarantinedVM
1510	3, // 11: eitri.v1.ActualStateReport.capacity:type_name -> eitri.v1.Capacity	1703	3, // 11: eitri.v1.ActualStateReport.capacity:type_name -> eitri.v1.Capacity
1511	5, // 12: eitri.v1.ActualStateReport.metrics:type_name -> eitri.v1.HostMetrics	1704	5, // 12: eitri.v1.ActualStateReport.metrics:type_name -> eitri.v1.HostMetrics
1512	9, // 13: eitri.v1.DesiredStateSnapshot.vms:type_name -> eitri.v1.VMDesired	1705	17, // 13: eitri.v1.ActualStateReport.exposures:type_name -> eitri.v1.ExposureActual
1513	11, // 14: eitri.v1.DesiredStateSnapshot.agent_upgrade:type_name -> eitri.v1.AgentUpgrade	1706	9, // 14: eitri.v1.DesiredStateSnapshot.vms:type_name -> eitri.v1.VMDesired
1514	15, // [15:15] is the sub-list for method output_type	1707	11, // 15: eitri.v1.DesiredStateSnapshot.agent_upgrade:type_name -> eitri.v1.AgentUpgrade
1515	15, // [15:15] is the sub-list for method input_type	1708	16, // 16: eitri.v1.DesiredStateSnapshot.exposures:type_name -> eitri.v1.ExposureDesired
1516	15, // [15:15] is the sub-list for extension type_name	1709	17, // [17:17] is the sub-list for method output_type
1517	15, // [15:15] is the sub-list for extension extendee	1710	17, // [17:17] is the sub-list for method input_type
1518	0, // [0:15] is the sub-list for field type_name	1711	17, // [17:17] is the sub-list for extension type_name
		1712	17, // [17:17] is the sub-list for extension extendee
		1713	0, // [0:17] is the sub-list for field type_name
1519	}	1714	}
1520		1715
1521	func init() { file_proto_eitri_v1_sync_proto_init() }	1716	func init() { file_proto_eitri_v1_sync_proto_init() }
@@ -1540,7 +1735,7 @@ func file_proto_eitri_v1_sync_proto_init() {
1540	GoPackagePath: reflect.TypeOf(x{}).PkgPath(),	1735	GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
1541	RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_eitri_v1_sync_proto_rawDesc), len(file_proto_eitri_v1_sync_proto_rawDesc)),	1736	RawDescriptor: unsafe.Slice(unsafe.StringData(file_proto_eitri_v1_sync_proto_rawDesc), len(file_proto_eitri_v1_sync_proto_rawDesc)),
1542	NumEnums: 0,	1737	NumEnums: 0,
1543	NumMessages: 16,	1738	NumMessages: 18,
1544	NumExtensions: 0,	1739	NumExtensions: 0,
1545	NumServices: 0,	1740	NumServices: 0,
1546	},	1741	},

internal/server/api/api.go

Old		New
@@ -400,6 +400,7 @@ func toHostResponse(h store.Host, st registry.HostState, ok bool, alloc store.Al
400	Kernel: h.Kernel,	400	Kernel: h.Kernel,
401	CPUModel: h.CPUModel,	401	CPUModel: h.CPUModel,
402	Virt: h.Virt,	402	Virt: h.Virt,
		403	UplinkAddr: h.UplinkAddr,
403	}	404	}
404	if ok {	405	if ok {
405	hr.Online = st.Online	406	hr.Online = st.Online

internal/server/api/client/client.go

Old		New
@@ -35,6 +35,8 @@ type (
35	Me = types.Me	35	Me = types.Me
36	CreateAPITokenResponse = types.CreateAPITokenResponse	36	CreateAPITokenResponse = types.CreateAPITokenResponse
37	UserCA = types.UserCA	37	UserCA = types.UserCA
		38	Exposure = types.Exposure
		39	CreateExposureRequest = types.CreateExposureRequest
38	)	40	)
39		41
40	// Client calls the eitri API at BaseURL, authenticating with Token (sent as a	42	// Client calls the eitri API at BaseURL, authenticating with Token (sent as a
@@ -145,6 +147,28 @@ func (c *Client) PatchVM(ctx context.Context, id, powerState string) error {
145	return c.do(ctx, http.MethodPatch, "/api/v1/vms/"+url.PathEscape(id), PatchVMRequest{PowerState: powerState}, nil)	147	return c.do(ctx, http.MethodPatch, "/api/v1/vms/"+url.PathEscape(id), PatchVMRequest{PowerState: powerState}, nil)
146	}	148	}
147		149
		150	// CreateExposure publishes guestPort of a VM on its host. hostPort 0 asks the
		151	// control plane to allocate one from the reserved range; the returned exposure
		152	// carries whichever port it ended up with.
		153	func (c *Client) CreateExposure(ctx context.Context, vmID string, guestPort, hostPort int64) (Exposure, error) {
		154	req := CreateExposureRequest{GuestPort: guestPort, HostPort: hostPort}
		155	var out Exposure
		156	return out, c.do(ctx, http.MethodPost, "/api/v1/vms/"+url.PathEscape(vmID)+"/exposures", req, &out)
		157	}
		158
		159	// ListExposures returns a VM's published ports, with the host address to dial
		160	// and each listener's reported state.
		161	func (c *Client) ListExposures(ctx context.Context, vmID string) ([]Exposure, error) {
		162	var out []Exposure
		163	return out, c.do(ctx, http.MethodGet, "/api/v1/vms/"+url.PathEscape(vmID)+"/exposures", nil, &out)
		164	}
		165
		166	// DeleteExposure revokes an exposure by id; the host closes the listener on its
		167	// next converge.
		168	func (c *Client) DeleteExposure(ctx context.Context, id string) error {
		169	return c.do(ctx, http.MethodDelete, "/api/v1/exposures/"+url.PathEscape(id), nil, nil)
		170	}
		171
148	// Me returns the signed-in identity (tenant handle + bound email) for the	172	// Me returns the signed-in identity (tenant handle + bound email) for the
149	// credential this client carries. It takes no context — the consumers (smoke	173	// credential this client carries. It takes no context — the consumers (smoke
150	// gate, CLI) call it as a quick synchronous probe; the do timeout bounds it.	174	// gate, CLI) call it as a quick synchronous probe; the do timeout bounds it.

internal/server/api/client/client_test.go

@@ @@ -470,3 +470,60 @@ func TestUploadUserCAEscapesTenant(t *testing.T) { @@
+		t.Errorf("path = %q, want %q", cap.path, want)
+func TestCreateExposure(t *testing.T) {
+	var cap capture
+	srv := serve(t, &cap, http.StatusCreated,
+		`{"id":"x-1","vm_id":"v-1","host_id":"h-1","guest_port":8080,"host_port":30080,`+
+			`"host_addr":"192.168.0.190","protocol":"tcp","scope":"lan","state":"pending","reason":"",`+
+			`"created_at":"2026-08-05T12:00:00Z"}`)
+	c := &client.Client{BaseURL: srv.URL, Token: "eitri_pat_x"}
+	got, err := c.CreateExposure(context.Background(), "v-1", 8080, 0)
+	if err != nil {
+		t.Fatalf("CreateExposure: %v", err)
+	}
+	if cap.method != http.MethodPost || cap.path != "/api/v1/vms/v-1/exposures" {
+		t.Errorf("request = %s %s, want POST /api/v1/vms/v-1/exposures", cap.method, cap.path)
+	}
+	var sent types.CreateExposureRequest
+	if err := json.Unmarshal(cap.body, &sent); err != nil {
+		t.Fatalf("decode sent body: %v", err)
+	}
+	if sent.GuestPort != 8080 || sent.HostPort != 0 {
+		t.Errorf("sent = %+v, want guest 8080 / host 0", sent)
+	}
+	if got.ID != "x-1" || got.HostPort != 30080 {
+		t.Errorf("got = %+v, want the allocated exposure back", got)
+	}
+}
+func TestListExposures(t *testing.T) {
+	var cap capture
+	srv := serve(t, &cap, http.StatusOK, `[{"id":"x-1","host_port":30080,"state":"active"}]`)
+	c := &client.Client{BaseURL: srv.URL, Token: "eitri_pat_x"}
+	got, err := c.ListExposures(context.Background(), "v-1")
+	if err != nil {
+		t.Fatalf("ListExposures: %v", err)
+	}
+	if cap.method != http.MethodGet || cap.path != "/api/v1/vms/v-1/exposures" {
+		t.Errorf("request = %s %s, want GET /api/v1/vms/v-1/exposures", cap.method, cap.path)
+	}
+	if len(got) != 1 || got[0].State != "active" {
+		t.Errorf("got = %+v, want one active exposure", got)
+	}
+}
+func TestDeleteExposure(t *testing.T) {
+	var cap capture
+	srv := serve(t, &cap, http.StatusNoContent, "")
+	c := &client.Client{BaseURL: srv.URL, Token: "eitri_pat_x"}
+	if err := c.DeleteExposure(context.Background(), "x-1"); err != nil {
+		t.Fatalf("DeleteExposure: %v", err)
+	}
+	if cap.method != http.MethodDelete || cap.path != "/api/v1/exposures/x-1" {
+		t.Errorf("request = %s %s, want DELETE /api/v1/exposures/x-1", cap.method, cap.path)
+	}
+}

internal/server/api/exposures.go

@@ @@ -0,0 +1,185 @@ @@
+package api
+import (
+	"database/sql"
+	"errors"
+	"net/http"
+	"strconv"
+	"github.com/a73x/eitri/internal/server/api/types"
+	"github.com/a73x/eitri/internal/server/registry"
+	"github.com/a73x/eitri/internal/server/store"
+)
+// exposureVM resolves the {id} path segment to a VM the caller may act on. A
+// foreign-tenant VM answers exactly like a missing one — existence is not
+// leaked across tenants. On refusal it writes the response and returns
+// ok=false.
+func (a *API) exposureVM(w http.ResponseWriter, r *http.Request) (store.VM, bool) {
+	vm, err := a.st.GetVM(r.PathValue("id"))
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			http.Error(w, "not found", http.StatusNotFound)
+			return store.VM{}, false
+		}
+		http.Error(w, "internal error", http.StatusInternalServerError)
+		return store.VM{}, false
+	}
+	if !mayActAs(principalFromContext(r), vm.Tenant) {
+		http.Error(w, "not found", http.StatusNotFound)
+		return store.VM{}, false
+	}
+	return vm, true
+}
+// validateExposurePorts checks a create request, returning (msg, status) on
+// failure or ("", 0) when valid.
+//
+// guest_port is unrestricted within the port space: the guest owns the guest,
+// including which of its ports are worth publishing. host_port is where
+// privilege lives — the 1024 floor is uniformity as much as safety, because a
+// macOS agent is unprivileged and cannot bind lower, and low ports are the
+// future gateway's territory.
+func validateExposurePorts(req types.CreateExposureRequest) (string, int) {
+	if req.GuestPort < 1 || req.GuestPort > 65535 {
+		return "guest_port must be between 1 and 65535", http.StatusBadRequest
+	}
+	if req.HostPort != 0 && (req.HostPort < 1024 || req.HostPort > 65535) {
+		return "host_port must be between 1024 and 65535, or omitted to allocate one from " +
+				strconv.Itoa(store.MinAllocatedHostPort) + "-" + strconv.Itoa(store.MaxAllocatedHostPort),
+			http.StatusBadRequest
+	}
+	return "", 0
+}
+// exposureState folds a host's live report into one exposure's state. An
+// exposure the host has not reported on is "pending": the grant exists, and
+// nothing has said what the host made of it yet.
+func exposureState(statuses []registry.ExposureStatus, id string) (state, reason string) {
+	for _, s := range statuses {
+		if s.ID == id {
+			return s.State, s.Reason
+		}
+	}
+	return "pending", ""
+}
+// toExposureResponse merges a durable exposure row with its host's address and
+// live reported state into the wire shape.
+func toExposureResponse(e store.Exposure, hostAddr string, statuses []registry.ExposureStatus) types.Exposure {
+	state, reason := exposureState(statuses, e.ID)
+	return types.Exposure{
+		ID: e.ID, VMID: e.VMID, HostID: e.HostID,
+		GuestPort: e.GuestPort, HostPort: e.HostPort, HostAddr: hostAddr,
+		Protocol: e.Protocol, Scope: e.Scope,
+		State: state, Reason: reason, CreatedAt: e.CreatedAt,
+	}
+}
+// hostFacing returns the address to dial for exposures on hostID, and that
+// host's live exposure reports. Both are best-effort: a host that has not
+// reported yields an empty address and no statuses, which renders as a pending
+// exposure with nowhere named yet.
+func (a *API) hostFacing(hostID string) (string, []registry.ExposureStatus) {
+	var addr string
+	if h, err := a.st.GetHost(hostID); err == nil {
+		addr = h.UplinkAddr
+	}
+	if st, ok := a.reg.Get(hostID); ok {
+		return addr, st.Report.Exposures
+	}
+	return addr, nil
+}
+// handleCreateExposure publishes one guest port of one VM on that VM's host.
+func (a *API) handleCreateExposure(w http.ResponseWriter, r *http.Request) {
+	vm, ok := a.exposureVM(w, r)
+	if !ok {
+		return
+	}
+	var req types.CreateExposureRequest
+	if !decodeJSON(w, r, &req) {
+		return
+	}
+	if msg, code := validateExposurePorts(req); msg != "" {
+		http.Error(w, msg, code)
+		return
+	}
+	e, err := a.st.CreateExposure(vm.ID, req.GuestPort, req.HostPort)
+	if err != nil {
+		switch {
+		case errors.Is(err, store.ErrExposureVMNotFound):
+			http.Error(w, "not found", http.StatusNotFound)
+		case errors.Is(err, store.ErrHostPortTaken):
+			http.Error(w, "host port already in use on that host", http.StatusConflict)
+		case errors.Is(err, store.ErrNoFreeHostPort):
+			http.Error(w, "no free host port in the reserved range on that host", http.StatusConflict)
+		default:
+			http.Error(w, "internal error", http.StatusInternalServerError)
+		}
+		return
+	}
+	a.audit(vm.Tenant, "exposure.create", map[string]string{
+		"vm_id": vm.ID, "name": vm.Name, "exposure_id": e.ID,
+		"guest_port": strconv.FormatInt(e.GuestPort, 10),
+		"host_port":  strconv.FormatInt(e.HostPort, 10),
+	})
+	a.hub.Poke(vm.HostID)
+	a.notif.notify()
+	addr, statuses := a.hostFacing(vm.HostID)
+	writeJSON(w, http.StatusCreated, toExposureResponse(e, addr, statuses))
+}
+// handleListExposures lists one VM's published ports, lowest host port first.
+func (a *API) handleListExposures(w http.ResponseWriter, r *http.Request) {
+	vm, ok := a.exposureVM(w, r)
+	if !ok {
+		return
+	}
+	exps, err := a.st.ListExposuresForVM(vm.ID)
+	if err != nil {
+		http.Error(w, "internal error", http.StatusInternalServerError)
+		return
+	}
+	addr, statuses := a.hostFacing(vm.HostID)
+	out := make([]types.Exposure, 0, len(exps))
+	for _, e := range exps {
+		out = append(out, toExposureResponse(e, addr, statuses))
+	}
+	writeJSON(w, http.StatusOK, out)
+}
+// handleDeleteExposure revokes a grant. There is no update verb: wrong port,
+// delete it and issue another.
+func (a *API) handleDeleteExposure(w http.ResponseWriter, r *http.Request) {
+	id := r.PathValue("id")
+	e, err := a.st.GetExposure(id)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			http.Error(w, "not found", http.StatusNotFound)
+			return
+		}
+		http.Error(w, "internal error", http.StatusInternalServerError)
+		return
+	}
+	if !mayActAs(principalFromContext(r), e.Tenant) {
+		http.Error(w, "not found", http.StatusNotFound)
+		return
+	}
+	if err := a.st.DeleteExposure(id); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			http.Error(w, "not found", http.StatusNotFound)
+			return
+		}
+		http.Error(w, "internal error", http.StatusInternalServerError)
+		return
+	}
+	a.audit(e.Tenant, "exposure.delete", map[string]string{
+		"vm_id": e.VMID, "exposure_id": e.ID,
+		"guest_port": strconv.FormatInt(e.GuestPort, 10),
+		"host_port":  strconv.FormatInt(e.HostPort, 10),
+	})
+	a.hub.Poke(e.HostID)
+	a.notif.notify()
+	w.WriteHeader(http.StatusNoContent)
+}

internal/server/api/exposures_test.go

@@ @@ -0,0 +1,237 @@ @@
+package api
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"github.com/a73x/eitri/internal/server/registry"
+	"github.com/a73x/eitri/internal/server/store"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// createTestVM places one VM on the enrolled host and returns its id.
+func createTestVM(t *testing.T, ts *httptest.Server, hostID, name string) string {
+	t.Helper()
+	resp := do(t, "POST", ts.URL+"/api/v1/vms", testPAT,
+		map[string]any{"host_id": hostID, "name": name})
+	require.Equal(t, 201, resp.StatusCode)
+	var out map[string]string
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&out))
+	return out["id"]
+}
+// patForOtherTenant JIT-provisions a second tenant and mints a PAT for it, so
+// a test can act as somebody else entirely.
+func patForOtherTenant(t *testing.T, st *store.Store) string {
+	t.Helper()
+	tn, err := st.CreateTenantForIdentity("https://test-issuer", "other-subject", "other@test.local")
+	require.NoError(t, err)
+	secret, _, err := st.CreateAPIToken(tn.ID, "other", 0)
+	require.NoError(t, err)
+	return secret
+}
+func TestCreateExposureAllocatesAndLists(t *testing.T) {
+	ts, _, _ := testServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+		map[string]any{"guest_port": 8080})
+	require.Equal(t, 201, resp.StatusCode)
+	var e map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&e))
+	assert.Equal(t, float64(8080), e["guest_port"])
+	assert.GreaterOrEqual(t, e["host_port"], float64(30000))
+	assert.LessOrEqual(t, e["host_port"], float64(32767))
+	assert.Equal(t, "tcp", e["protocol"])
+	assert.Equal(t, "lan", e["scope"])
+	assert.Equal(t, "pending", e["state"], "no agent has reported on it yet")
+	assert.Equal(t, vmID, e["vm_id"])
+	resp = do(t, "GET", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT, nil)
+	require.Equal(t, 200, resp.StatusCode)
+	var list []map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&list))
+	require.Len(t, list, 1)
+	assert.Equal(t, e["id"], list[0]["id"])
+}
+func TestCreateExposureHonoursAndRefusesHostPorts(t *testing.T) {
+	ts, _, _ := testServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	otherID := createTestVM(t, ts, host["host_id"], "web-2")
+	resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+		map[string]any{"guest_port": 8080, "host_port": 8443})
+	require.Equal(t, 201, resp.StatusCode)
+	// The same port on the same host is taken.
+	resp = do(t, "POST", ts.URL+"/api/v1/vms/"+otherID+"/exposures", testPAT,
+		map[string]any{"guest_port": 9090, "host_port": 8443})
+	assert.Equal(t, 409, resp.StatusCode)
+}
+func TestCreateExposureValidatesPorts(t *testing.T) {
+	ts, _, _ := testServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	for _, tc := range []struct {
+		name string
+		body map[string]any
+	}{
+		{"guest port zero", map[string]any{"guest_port": 0}},
+		{"guest port too high", map[string]any{"guest_port": 65536}},
+		{"guest port negative", map[string]any{"guest_port": -1}},
+		{"privileged host port", map[string]any{"guest_port": 8080, "host_port": 443}},
+		{"host port too high", map[string]any{"guest_port": 8080, "host_port": 65536}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT, tc.body)
+			assert.Equal(t, 400, resp.StatusCode)
+		})
+	}
+}
+func TestDeleteExposure(t *testing.T) {
+	ts, _, _ := testServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+		map[string]any{"guest_port": 8080})
+	require.Equal(t, 201, resp.StatusCode)
+	var e map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&e))
+	resp = do(t, "DELETE", ts.URL+"/api/v1/exposures/"+e["id"].(string), testPAT, nil)
+	assert.Equal(t, http.StatusNoContent, resp.StatusCode)
+	resp = do(t, "GET", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT, nil)
+	var list []map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&list))
+	assert.Empty(t, list)
+	// A second delete is a 404 — an exposure is a grant, and it is gone.
+	resp = do(t, "DELETE", ts.URL+"/api/v1/exposures/"+e["id"].(string), testPAT, nil)
+	assert.Equal(t, 404, resp.StatusCode)
+}
+func TestExposureRoutesAreTenantScoped(t *testing.T) {
+	ts, st, _ := testServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+		map[string]any{"guest_port": 8080})
+	require.Equal(t, 201, resp.StatusCode)
+	var e map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&e))
+	otherPAT := patForOtherTenant(t, st)
+	// A foreign tenant answers exactly like a missing one — existence is not
+	// leaked across tenants.
+	assert.Equal(t, 404, do(t, "GET", ts.URL+"/api/v1/vms/"+vmID+"/exposures", otherPAT, nil).StatusCode)
+	assert.Equal(t, 404, do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", otherPAT,
+		map[string]any{"guest_port": 9090}).StatusCode)
+	assert.Equal(t, 404, do(t, "DELETE", ts.URL+"/api/v1/exposures/"+e["id"].(string), otherPAT, nil).StatusCode)
+}
+// TestExposureReportedStateAndAddressReachTheWire pins the fold from a host's
+// live report onto the exposures it was asked to run: the state and reason the
+// agent last said, and the address the host answers on. The second exposure —
+// present in the store, absent from the report — pins that the fold matches by
+// exposure id rather than by position, so a host reporting on one listener
+// cannot colour another one's state.
+func TestExposureReportedStateAndAddressReachTheWire(t *testing.T) {
+	ts, st, _, reg, _ := newServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	newExposure := func(guestPort, hostPort int64) string {
+		t.Helper()
+		resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+			map[string]any{"guest_port": guestPort, "host_port": hostPort})
+		require.Equal(t, 201, resp.StatusCode)
+		var e map[string]any
+		require.NoError(t, json.NewDecoder(resp.Body).Decode(&e))
+		return e["id"].(string)
+	}
+	// Named host ports, so the list order (lowest host port first) is fixed.
+	reported := newExposure(8080, 30080)
+	newExposure(9090, 30081)
+	require.NoError(t, st.RecordHostUplink(host["host_id"], "192.168.0.190"))
+	reg.UpdateReport(host["host_id"], registry.Report{
+		Exposures: []registry.ExposureStatus{{
+			ID:     reported,
+			State:  "failed",
+			Reason: "listen tcp 0.0.0.0:30080: bind: address already in use",
+		}},
+	})
+	resp := do(t, "GET", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT, nil)
+	require.Equal(t, 200, resp.StatusCode)
+	var list []map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&list))
+	require.Len(t, list, 2)
+	assert.Equal(t, reported, list[0]["id"])
+	assert.Equal(t, "failed", list[0]["state"])
+	assert.Contains(t, list[0]["reason"], "address already in use")
+	// The unreported one keeps its own state — nothing has said what the host
+	// made of it.
+	assert.Equal(t, "pending", list[1]["state"])
+	assert.Equal(t, "", list[1]["reason"])
+	// The address is the host's, so every exposure on it names the same place.
+	for _, e := range list {
+		assert.Equal(t, "192.168.0.190", e["host_addr"])
+	}
+}
+// TestCreateExposureOnTombstonedVMIsNotFound pins that a VM on its way out
+// takes no new grants: the store refuses, and the refusal reads as a missing
+// VM rather than a server error.
+func TestCreateExposureOnTombstonedVMIsNotFound(t *testing.T) {
+	ts, st, _ := testServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	require.NoError(t, st.TombstoneVM(vmID))
+	resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+		map[string]any{"guest_port": 8080})
+	assert.Equal(t, 404, resp.StatusCode)
+}
+func TestExposureCreateAndDeleteAreAudited(t *testing.T) {
+	ts, _, _ := testServer(t)
+	host := enroll(t, ts)
+	vmID := createTestVM(t, ts, host["host_id"], "web-1")
+	resp := do(t, "POST", ts.URL+"/api/v1/vms/"+vmID+"/exposures", testPAT,
+		map[string]any{"guest_port": 8080})
+	require.Equal(t, 201, resp.StatusCode)
+	var e map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&e))
+	require.Equal(t, 204, do(t, "DELETE", ts.URL+"/api/v1/exposures/"+e["id"].(string), testPAT, nil).StatusCode)
+	// Both rows key the VM as vm_id, so they land in that VM's own timeline.
+	resp = do(t, "GET", ts.URL+"/api/v1/vms/"+vmID+"/events", testPAT, nil)
+	require.Equal(t, 200, resp.StatusCode)
+	var events []map[string]any
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&events))
+	actions := map[string]bool{}
+	for _, ev := range events {
+		actions[ev["action"].(string)] = true
+	}
+	assert.True(t, actions["exposure.create"], "events: %v", actions)
+	assert.True(t, actions["exposure.delete"], "events: %v", actions)
+}

internal/server/api/routes.go

@@ @@ -233,6 +233,39 @@ var routeTable = []Route{ @@
+		Doc:      "One VM's lifecycle timeline (audit rows carrying its vm_id), newest first; survives the VM row being reaped.",
+		handler:  (*API).handleListVMEvents,
+	},
+	// Service exposure: a tenant publishes one TCP port of one VM, and the
+	// fleet binds it on that VM's host. There is no update verb — an exposure
+	// is a grant, not a document.
+	{
+		Method:   "POST",
+		Path:     "/api/v1/vms/{id}/exposures",
+		Auth:     AuthUser,
+		Kind:     KindJSON,
+		Request:  (*types.CreateExposureRequest)(nil),
+		Response: (*types.Exposure)(nil),
+		Success:  http.StatusCreated,
+		Doc:      "Publish a guest TCP port on the VM's host. Omit host_port to allocate one from the reserved range 30000-32767; a named port must be >= 1024 and is honored or refused.",
+		handler:  (*API).handleCreateExposure,
+	},
+	{
+		Method:   "GET",
+		Path:     "/api/v1/vms/{id}/exposures",
+		Auth:     AuthUser,
+		Kind:     KindJSON,
+		Response: []types.Exposure(nil),
+		Success:  http.StatusOK,
+		Doc:      "List the VM's published ports, with the host address to dial and each listener's reported state.",
+		handler:  (*API).handleListExposures,
+	},
+	{
+		Method:  "DELETE",
+		Path:    "/api/v1/exposures/{id}",
+		Auth:    AuthUser,
+		Kind:    KindJSON,
+		Success: http.StatusNoContent,
+		Doc:     "Revoke an exposure; its host closes the listener on the next converge.",
+		handler: (*API).handleDeleteExposure,
+	},
+	// BYO per-tenant SSH user CAs: eitri stores only the CA pubkey and never
+	// holds a user signing key. Tenant-scoped (caller must act for {tenant}).

internal/server/api/routes_test.go

Old		New
@@ -34,7 +34,7 @@ func exemplarElem(t *testing.T, route Route, role string, v any) reflect.Type {
34	// `required` array for request schemas, so a type serving both roles would	34	// `required` array for request schemas, so a type serving both roles would
35	// get the wrong treatment on one of them).	35	// get the wrong treatment on one of them).
36	func TestRouteTable(t *testing.T) {	36	func TestRouteTable(t *testing.T) {
37	const wantRoutes = 28	37	const wantRoutes = 31
38	if len(routeTable) != wantRoutes {	38	if len(routeTable) != wantRoutes {
39	t.Fatalf("route table has %d entries, want %d — new endpoint? update this pin and cmd/eitri-apispec coverage together", len(routeTable), wantRoutes)	39	t.Fatalf("route table has %d entries, want %d — new endpoint? update this pin and cmd/eitri-apispec coverage together", len(routeTable), wantRoutes)
40	}	40	}

internal/server/api/testdata/create-exposure-request.golden.json

@@ @@ -0,0 +1,4 @@ @@
+{
+  "guest_port": 8080,
+  "host_port": 30080
+}

internal/server/api/testdata/exposure.golden.json

@@ @@ -0,0 +1,15 @@ @@
+[
+  {
+    "id": "x-9012",
+    "vm_id": "v-5678",
+    "host_id": "h-1234",
+    "guest_port": 8080,
+    "host_port": 30080,
+    "host_addr": "192.168.0.190",
+    "protocol": "tcp",
+    "scope": "lan",
+    "state": "failed",
+    "reason": "listen tcp 0.0.0.0:30080: bind: address already in use",
+    "created_at": "2026-07-27T12:04:00Z"
+  }
+]

internal/server/api/testdata/host.golden.json

Old		New
@@ -30,6 +30,7 @@
30	"kernel": "6.15.4-arch1-1",	30	"kernel": "6.15.4-arch1-1",
31	"cpu_model": "AMD Ryzen 9 7950X",	31	"cpu_model": "AMD Ryzen 9 7950X",
32	"virt": "kvm",	32	"virt": "kvm",
		33	"uplink_addr": "192.168.0.190",
33	"metrics": {	34	"metrics": {
34	"uptime_s": 86400,	35	"uptime_s": 86400,
35	"mem_used_mb": 12000,	36	"mem_used_mb": 12000,

internal/server/api/testdata/snapshot.golden.json

Old		New
@@ -32,6 +32,7 @@
32	"kernel": "6.15.4-arch1-1",	32	"kernel": "6.15.4-arch1-1",
33	"cpu_model": "AMD Ryzen 9 7950X",	33	"cpu_model": "AMD Ryzen 9 7950X",
34	"virt": "kvm",	34	"virt": "kvm",
		35	"uplink_addr": "192.168.0.190",
35	"metrics": {	36	"metrics": {
36	"uptime_s": 86400,	37	"uptime_s": 86400,
37	"mem_used_mb": 12000,	38	"mem_used_mb": 12000,

internal/server/api/types/types.go

Old		New
@@ -75,6 +75,10 @@ type Host struct {
75	Kernel string `json:"kernel"`	75	Kernel string `json:"kernel"`
76	CPUModel string `json:"cpu_model"`	76	CPUModel string `json:"cpu_model"`
77	Virt string `json:"virt"`	77	Virt string `json:"virt"`
		78	// UplinkAddr is the address this host answers on, as the host reports it —
		79	// the address to dial for a published guest port. Empty until the host has
		80	// said.
		81	UplinkAddr string `json:"uplink_addr"`
78	// Metrics is live MEASURED utilization, present only when online. Distinct	82	// Metrics is live MEASURED utilization, present only when online. Distinct
79	// from Allocated (control-plane bookkeeping / VM quotas).	83	// from Allocated (control-plane bookkeeping / VM quotas).
80	Metrics *Metrics `json:"metrics"`	84	Metrics *Metrics `json:"metrics"`
@@ -222,6 +226,36 @@ type RevokeSSHCertRequest struct {
222	Reason string `json:"reason"`	226	Reason string `json:"reason"`
223	}	227	}
224		228
		229	// CreateExposureRequest is the POST /api/v1/vms/{id}/exposures body: publish
		230	// guest_port of that VM. host_port is optional — omitted (or 0) allocates one
		231	// from the reserved range 30000-32767; a named port must be >= 1024 and is
		232	// honored or refused. Request-side only.
		233	type CreateExposureRequest struct {
		234	GuestPort int64 `json:"guest_port"`
		235	HostPort int64 `json:"host_port"` // 0 = allocate
		236	}
		237
		238	// Exposure is one published guest port, served by
		239	// POST and GET /api/v1/vms/{id}/exposures. HostAddr and State are
		240	// server-derived: the address comes from the host's own report, and the state
		241	// from what its agent last said its listener is doing — "pending" until an
		242	// agent has reported at all, then "active" (the listener is bound) or "failed"
		243	// with the OS error in Reason. "active" describes the host half of the pipe;
		244	// whether anything answers inside the guest is the guest's business.
		245	type Exposure struct {
		246	ID string `json:"id"`
		247	VMID string `json:"vm_id"`
		248	HostID string `json:"host_id"`
		249	GuestPort int64 `json:"guest_port"`
		250	HostPort int64 `json:"host_port"`
		251	HostAddr string `json:"host_addr"`
		252	Protocol string `json:"protocol"`
		253	Scope string `json:"scope"`
		254	State string `json:"state"`
		255	Reason string `json:"reason"`
		256	CreatedAt time.Time `json:"created_at"`
		257	}
		258
225	// The response shapes below replace handlers' inline map[string]string	259	// The response shapes below replace handlers' inline map[string]string
226	// literals. Their fields are ordered ALPHABETICALLY BY JSON KEY on purpose:	260	// literals. Their fields are ordered ALPHABETICALLY BY JSON KEY on purpose:
227	// encoding/json marshals map keys sorted, so keeping struct fields in that	261	// encoding/json marshals map keys sorted, so keeping struct fields in that

internal/server/api/wire_golden_test.go

Old		New
@@ -75,6 +75,7 @@ func TestWireGolden(t *testing.T) {
75	Kernel: "6.15.4-arch1-1",	75	Kernel: "6.15.4-arch1-1",
76	CPUModel: "AMD Ryzen 9 7950X",	76	CPUModel: "AMD Ryzen 9 7950X",
77	Virt: "kvm",	77	Virt: "kvm",
		78	UplinkAddr: "192.168.0.190",
78	Metrics: &types.Metrics{	79	Metrics: &types.Metrics{
79	UptimeS: 86400,	80	UptimeS: 86400,
80	MemUsedMB: 12000,	81	MemUsedMB: 12000,
@@ -240,4 +241,23 @@ func TestWireGolden(t *testing.T) {
240	Name: "boot-gate",	241	Name: "boot-gate",
241	RevokedAt: "",	242	RevokedAt: "",
242	}})	243	}})
		244
		245	goldenCheck(t, "create-exposure-request", types.CreateExposureRequest{
		246	GuestPort: 8080,
		247	HostPort: 30080,
		248	})
		249
		250	goldenCheck(t, "exposure", []types.Exposure{{
		251	ID: "x-9012",
		252	VMID: "v-5678",
		253	HostID: "h-1234",
		254	GuestPort: 8080,
		255	HostPort: 30080,
		256	HostAddr: "192.168.0.190",
		257	Protocol: "tcp",
		258	Scope: "lan",
		259	State: "failed",
		260	Reason: "listen tcp 0.0.0.0:30080: bind: address already in use",
		261	CreatedAt: base.Add(4 * time.Minute),
		262	}})
243	}	263	}

internal/server/registry/registry.go

Old		New
@@ -46,9 +46,17 @@ type QuarantinedVM struct {
46	DestroyAtUnix int64	46	DestroyAtUnix int64
47	}	47	}
48		48
		49	// ExposureStatus is one exposure's live state as its host's agent reports it:
		50	// "active" once the host listener is bound, "failed" with the OS error
		51	// otherwise. Never persisted — like Metrics, it lives only while the host is
		52	// connected, and an exposure with no row here has simply not been reported on
		53	// yet.
		54	type ExposureStatus struct{ ID, State, Reason string }
		55
49	type Report struct {	56	type Report struct {
50	VMs []ActualVM	57	VMs []ActualVM
51	Quarantined []QuarantinedVM	58	Quarantined []QuarantinedVM
		59	Exposures []ExposureStatus
52	Capacity Capacity	60	Capacity Capacity
53	Metrics Metrics	61	Metrics Metrics
54	FenceViolation bool	62	FenceViolation bool
@@ -136,6 +144,7 @@ func (r *Registry) Get(hostID string) (HostState, bool) {
136	}	144	}
137	// Deep-copy slices so callers cannot corrupt registry state.	145	// Deep-copy slices so callers cannot corrupt registry state.
138	st.Report.VMs = slices.Clone(st.Report.VMs)	146	st.Report.VMs = slices.Clone(st.Report.VMs)
		147	st.Report.Exposures = slices.Clone(st.Report.Exposures)
139	quarantined := slices.Clone(st.Report.Quarantined)	148	quarantined := slices.Clone(st.Report.Quarantined)
140	for i := range quarantined {	149	for i := range quarantined {
141	quarantined[i].VMSpecJSON = bytes.Clone(quarantined[i].VMSpecJSON)	150	quarantined[i].VMSpecJSON = bytes.Clone(quarantined[i].VMSpecJSON)

internal/server/registry/registry_test.go

@@ @@ -131,3 +131,17 @@ func TestAgentVersionSurvivesReports(t *testing.T) { @@
+		t.Fatalf("AgentVersion = %q ok=%v, want v0.0.2 true", st.AgentVersion, ok)
+// TestGetClonesExposureStatuses pins that a caller cannot corrupt registry
+// state through the slice it is handed — the same rule the VM rows follow.
+func TestGetClonesExposureStatuses(t *testing.T) {
+	r := New(time.Now)
+	r.UpdateReport("h1", Report{Exposures: []ExposureStatus{{ID: "e1", State: "active"}}})
+	got, ok := r.Get("h1")
+	require.True(t, ok)
+	got.Report.Exposures[0].State = "clobbered"
+	again, _ := r.Get("h1")
+	assert.Equal(t, "active", again.Report.Exposures[0].State)
+}

internal/server/store/exposures.go

@@ @@ -0,0 +1,209 @@ @@
+package store
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"time"
+	"github.com/a73x/eitri/internal/random"
+	sqlite "modernc.org/sqlite"
+)
+// Exposure is one published guest port: the fleet binds HostPort on the VM's
+// host and pipes every accepted connection to GuestPort inside the guest.
+// Tenant and HostID are both derived from the VM at create — never accepted
+// from a caller — so an exposure can only ever name the partition and the
+// machine its VM already lives in.
+type Exposure struct {
+	ID, Tenant, VMID, HostID string
+	GuestPort, HostPort      int64
+	Protocol, Scope          string
+	CreatedAt                time.Time
+}
+// The reserved host-port range. It is the host contract: one documented span
+// that belongs to eitri on every host, so allocating a port needs no per-port
+// hygiene reasoning anywhere else. A caller that names its own port is not
+// held to it — only to the 1024 floor the API enforces.
+const (
+	MinAllocatedHostPort = 30000
+	MaxAllocatedHostPort = 32767
+)
+// ErrExposureVMNotFound reports that the VM an exposure would belong to does
+// not exist, or is tombstoned — a VM being torn down takes no new exposures.
+var ErrExposureVMNotFound = errors.New("vm not found")
+// ErrHostPortTaken reports that another exposure already holds the requested
+// host port on that host.
+var ErrHostPortTaken = errors.New("host port already exposed on this host")
+// ErrNoFreeHostPort reports that the reserved range is fully allocated on the
+// host, so there is nothing left to hand out.
+var ErrNoFreeHostPort = errors.New("no free host port in the reserved range")
+// exposureColumns is the positional column list every exposure SELECT uses, so
+// the order stays locked to scanExposure's positional Scan. Every column is
+// table-qualified because one of the reads joins vms — the alias `e` is part
+// of the contract each `from` clause below keeps.
+const exposureColumns = `e.id, e.tenant, e.vm_id, e.host_id, e.guest_port, e.host_port, e.protocol, e.scope, e.created_at`
+// scanExposure's column order must match exposureColumns exactly — it is
+// positional, not name-based.
+func scanExposure(rows *sql.Rows) (Exposure, error) {
+	var e Exposure
+	var createdAt string
+	if err := rows.Scan(&e.ID, &e.Tenant, &e.VMID, &e.HostID,
+		&e.GuestPort, &e.HostPort, &e.Protocol, &e.Scope, &createdAt); err != nil {
+		return Exposure{}, err
+	}
+	e.CreatedAt, _ = time.Parse(time.RFC3339, createdAt)
+	return e, nil
+}
+// queryExposures runs an exposureColumns-projected SELECT, with from appended
+// verbatim after the column list (it carries the FROM clause, any join, the
+// WHERE and the ORDER BY) and args bound in order.
+func queryExposures(q querier, from string, args ...any) ([]Exposure, error) {
+	rows, err := q.Query(`SELECT `+exposureColumns+` `+from, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+	var out []Exposure
+	for rows.Next() {
+		e, err := scanExposure(rows)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, e)
+	}
+	return out, rows.Err()
+}
+// CreateExposure publishes guestPort of vmID on that VM's host. hostPort 0 asks
+// for one from the reserved range; a named port is honored or refused. The
+// whole decision — which host, which tenant, which port — happens in one
+// transaction, so two concurrent creates cannot agree on the same port.
+func (s *Store) CreateExposure(vmID string, guestPort, hostPort int64) (Exposure, error) {
+	tx, err := s.db.Begin()
+	if err != nil {
+		return Exposure{}, err
+	}
+	defer tx.Rollback()
+	// The VM read is the derivation point: an exposure's tenant is ALWAYS its
+	// VM's, and its host is the host the VM is placed on. A tombstoned VM is
+	// treated as absent — its guest is on its way out.
+	var tenant, hostID string
+	switch err := tx.QueryRow(
+		`SELECT tenant, host_id FROM vms WHERE id=? AND deleted_at IS NULL`, vmID,
+	).Scan(&tenant, &hostID); {
+	case errors.Is(err, sql.ErrNoRows):
+		return Exposure{}, ErrExposureVMNotFound
+	case err != nil:
+		return Exposure{}, fmt.Errorf("lookup vm: %w", err)
+	}
+	if hostPort == 0 {
+		hostPort, err = allocateHostPort(tx, hostID)
+		if err != nil {
+			return Exposure{}, err
+		}
+	}
+	e := Exposure{
+		ID: random.Hex(16), Tenant: tenant, VMID: vmID, HostID: hostID,
+		GuestPort: guestPort, HostPort: hostPort,
+		Protocol: "tcp", Scope: "lan", CreatedAt: time.Now().UTC(),
+	}
+	if _, err := tx.Exec(
+		`INSERT INTO exposures(id, tenant, vm_id, host_id, guest_port, host_port, protocol, scope, created_at)
+		 VALUES (?,?,?,?,?,?,?,?,?)`,
+		e.ID, e.Tenant, e.VMID, e.HostID, e.GuestPort, e.HostPort, e.Protocol, e.Scope,
+		e.CreatedAt.Format(time.RFC3339),
+	); err != nil {
+		// SQLITE_CONSTRAINT_UNIQUE (2067): the only UNIQUE constraint this
+		// insert can trip besides the random-hex PK is exposures_host_port.
+		// Matched by errno, not message text, which embeds the index's column
+		// list and breaks on the next index change.
+		if serr, ok := errors.AsType[*sqlite.Error](err); ok && serr.Code() == 2067 {
+			return Exposure{}, ErrHostPortTaken
+		}
+		return Exposure{}, fmt.Errorf("insert exposure: %w", err)
+	}
+	if err := bumpEpoch(tx); err != nil {
+		return Exposure{}, fmt.Errorf("bump epoch: %w", err)
+	}
+	if err := tx.Commit(); err != nil {
+		return Exposure{}, err
+	}
+	return e, nil
+}
+// allocateHostPort returns the lowest free port of the reserved range on
+// hostID. It reads the range's held ports into memory before choosing, because
+// the store runs on a single connection: an open result set would block the
+// insert that follows in the same transaction.
+func allocateHostPort(tx *sql.Tx, hostID string) (int64, error) {
+	rows, err := tx.Query(
+		`SELECT host_port FROM exposures WHERE host_id=? AND host_port BETWEEN ? AND ?`,
+		hostID, MinAllocatedHostPort, MaxAllocatedHostPort)
+	if err != nil {
+		return 0, fmt.Errorf("read allocated host ports: %w", err)
+	}
+	defer rows.Close()
+	held := map[int64]bool{}
+	for rows.Next() {
+		var p int64
+		if err := rows.Scan(&p); err != nil {
+			return 0, err
+		}
+		held[p] = true
+	}
+	if err := rows.Err(); err != nil {
+		return 0, err
+	}
+	for p := int64(MinAllocatedHostPort); p <= MaxAllocatedHostPort; p++ {
+		if !held[p] {
+			return p, nil
+		}
+	}
+	return 0, ErrNoFreeHostPort
+}
+// DeleteExposure revokes an exposure. sql.ErrNoRows when there is none with
+// that id. Bumps the epoch so the host's agent closes the listener.
+func (s *Store) DeleteExposure(id string) error {
+	return s.mutate(`DELETE FROM exposures WHERE id=?`, id)
+}
+// GetExposure returns one exposure by id. sql.ErrNoRows when absent.
+func (s *Store) GetExposure(id string) (Exposure, error) {
+	out, err := queryExposures(s.db, `FROM exposures e WHERE e.id=?`, id)
+	if err != nil {
+		return Exposure{}, err
+	}
+	if len(out) == 0 {
+		return Exposure{}, sql.ErrNoRows
+	}
+	return out[0], nil
+}
+// ListExposuresForVM returns one VM's exposures, lowest host port first.
+func (s *Store) ListExposuresForVM(vmID string) ([]Exposure, error) {
+	return queryExposures(s.db, `FROM exposures e WHERE e.vm_id=? ORDER BY e.host_port`, vmID)
+}
+// ListExposuresForHost returns the exposures a host's agent should be running:
+// its own, for VMs that are not tombstoned. A VM on its way out has already
+// stopped being somewhere to send traffic, so its listener goes before its row
+// does.
+func (s *Store) ListExposuresForHost(hostID string) ([]Exposure, error) {
+	return queryExposures(s.db,
+		`FROM exposures e JOIN vms v ON v.id = e.vm_id
+		 WHERE e.host_id=? AND v.deleted_at IS NULL ORDER BY e.host_port`, hostID)
+}

internal/server/store/exposures_test.go

@@ @@ -0,0 +1,257 @@ @@
+package store
+import (
+	"database/sql"
+	"strconv"
+	"testing"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// makeExposureVM enrolls nothing new: it places one live VM on h so an exposure
+// has something to belong to.
+func makeExposureVM(t *testing.T, s *Store, h Host, name string) VM {
+	t.Helper()
+	return vmWithResources(t, s, h, name, 1, 512, 5)
+}
+func TestCreateExposureAllocatesFromTheReservedRange(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	e, err := s.CreateExposure(vm.ID, 8080, 0)
+	require.NoError(t, err)
+	assert.Equal(t, vm.ID, e.VMID)
+	assert.Equal(t, h.ID, e.HostID)
+	assert.Equal(t, h.Tenant, e.Tenant, "an exposure's tenant is always its VM's")
+	assert.Equal(t, int64(8080), e.GuestPort)
+	assert.Equal(t, int64(MinAllocatedHostPort), e.HostPort)
+	assert.Equal(t, "tcp", e.Protocol)
+	assert.Equal(t, "lan", e.Scope)
+	assert.NotEmpty(t, e.ID)
+}
+func TestCreateExposureAllocatesTheLowestFreePort(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	first, err := s.CreateExposure(vm.ID, 8080, 0)
+	require.NoError(t, err)
+	second, err := s.CreateExposure(vm.ID, 8081, 0)
+	require.NoError(t, err)
+	assert.Equal(t, int64(MinAllocatedHostPort), first.HostPort)
+	assert.Equal(t, int64(MinAllocatedHostPort+1), second.HostPort)
+	// A freed port is the lowest gap, so it is handed out again.
+	require.NoError(t, s.DeleteExposure(first.ID))
+	third, err := s.CreateExposure(vm.ID, 8082, 0)
+	require.NoError(t, err)
+	assert.Equal(t, int64(MinAllocatedHostPort), third.HostPort)
+}
+func TestCreateExposureHonoursARequestedPort(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	e, err := s.CreateExposure(vm.ID, 8080, 8443)
+	require.NoError(t, err)
+	assert.Equal(t, int64(8443), e.HostPort)
+}
+func TestCreateExposureRefusesAPortAnotherExposureHolds(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	other := makeExposureVM(t, s, h, "web-2")
+	_, err := s.CreateExposure(vm.ID, 8080, 8443)
+	require.NoError(t, err)
+	_, err = s.CreateExposure(other.ID, 9090, 8443)
+	assert.ErrorIs(t, err, ErrHostPortTaken, "the unique index IS the collision check")
+}
+func TestCreateExposureAllowsTheSamePortOnAnotherHost(t *testing.T) {
+	s := newStore(t)
+	h1 := enrollHost(t, s)
+	tok, err := s.CreateEnrollmentToken(testTenant)
+	require.NoError(t, err)
+	h2, err := s.RedeemEnrollmentToken(tok, EnrollFacts{Name: "h2", OS: "linux", Arch: "amd64", Provisioner: "cloudhv", Remote: ""})
+	require.NoError(t, err)
+	vm1 := makeExposureVM(t, s, h1, "web-1")
+	vm2 := makeExposureVM(t, s, h2, "web-2")
+	_, err = s.CreateExposure(vm1.ID, 8080, 8443)
+	require.NoError(t, err)
+	_, err = s.CreateExposure(vm2.ID, 8080, 8443)
+	assert.NoError(t, err, "the port is unique per host, not per fleet")
+}
+func TestCreateExposureRefusesAnUnknownOrTombstonedVM(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	_, err := s.CreateExposure("no-such-vm", 8080, 0)
+	assert.ErrorIs(t, err, ErrExposureVMNotFound)
+	require.NoError(t, s.TombstoneVM(vm.ID))
+	_, err = s.CreateExposure(vm.ID, 8080, 0)
+	assert.ErrorIs(t, err, ErrExposureVMNotFound, "a VM being torn down takes no new exposures")
+}
+func TestCreateExposureReportsRangeExhaustion(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	// Fill the whole reserved range by hand — CreateExposure would be 2768
+	// transactions.
+	for p := int64(MinAllocatedHostPort); p <= MaxAllocatedHostPort; p++ {
+		_, err := s.db.Exec(
+			`INSERT INTO exposures(id, tenant, vm_id, host_id, guest_port, host_port, protocol, scope, created_at)
+			 VALUES (?,?,?,?,?,?,'tcp','lan','2026-08-05T00:00:00Z')`,
+			"fill-"+strconv.FormatInt(p, 10), h.Tenant, vm.ID, h.ID, 8080, p)
+		require.NoError(t, err)
+	}
+	_, err := s.CreateExposure(vm.ID, 8080, 0)
+	assert.ErrorIs(t, err, ErrNoFreeHostPort)
+}
+func TestCreateExposureBumpsTheEpoch(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	before, err := s.Epoch()
+	require.NoError(t, err)
+	_, err = s.CreateExposure(vm.ID, 8080, 0)
+	require.NoError(t, err)
+	after, err := s.Epoch()
+	require.NoError(t, err)
+	assert.Greater(t, after, before, "exposures are desired state; agents must re-snapshot")
+}
+func TestCreateExposureCountsANamedPortInTheReservedRange(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	named, err := s.CreateExposure(vm.ID, 8080, MinAllocatedHostPort)
+	require.NoError(t, err)
+	assert.Equal(t, int64(MinAllocatedHostPort), named.HostPort)
+	auto, err := s.CreateExposure(vm.ID, 8081, 0)
+	require.NoError(t, err)
+	assert.Equal(t, int64(MinAllocatedHostPort+1), auto.HostPort, "a named port inside the reserved range is held against auto-allocation")
+}
+func TestListExposuresForVMIsPortOrdered(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	other := makeExposureVM(t, s, h, "web-2")
+	_, err := s.CreateExposure(vm.ID, 8081, 31000)
+	require.NoError(t, err)
+	_, err = s.CreateExposure(vm.ID, 8080, 30500)
+	require.NoError(t, err)
+	_, err = s.CreateExposure(other.ID, 9090, 30001)
+	require.NoError(t, err)
+	got, err := s.ListExposuresForVM(vm.ID)
+	require.NoError(t, err)
+	require.Len(t, got, 2, "another VM's exposures are not this VM's")
+	assert.Equal(t, int64(30500), got[0].HostPort)
+	assert.Equal(t, int64(31000), got[1].HostPort)
+}
+func TestListExposuresForHostSkipsTombstonedVMs(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	live := makeExposureVM(t, s, h, "web-1")
+	dying := makeExposureVM(t, s, h, "web-2")
+	_, err := s.CreateExposure(live.ID, 8080, 30001)
+	require.NoError(t, err)
+	_, err = s.CreateExposure(dying.ID, 8080, 30002)
+	require.NoError(t, err)
+	require.NoError(t, s.TombstoneVM(dying.ID))
+	got, err := s.ListExposuresForHost(h.ID)
+	require.NoError(t, err)
+	require.Len(t, got, 1, "no listener outlives the thing it points at")
+	assert.Equal(t, live.ID, got[0].VMID)
+}
+func TestGetExposureAndDelete(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	e, err := s.CreateExposure(vm.ID, 8080, 0)
+	require.NoError(t, err)
+	got, err := s.GetExposure(e.ID)
+	require.NoError(t, err)
+	assert.Equal(t, e.HostPort, got.HostPort)
+	assert.Equal(t, e.VMID, got.VMID)
+	require.NoError(t, s.DeleteExposure(e.ID))
+	_, err = s.GetExposure(e.ID)
+	assert.ErrorIs(t, err, sql.ErrNoRows)
+	// Deleting one that is already gone is not a silent success.
+	assert.ErrorIs(t, s.DeleteExposure(e.ID), sql.ErrNoRows)
+}
+func TestDeleteExposureBumpsTheEpoch(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	e, err := s.CreateExposure(vm.ID, 8080, 0)
+	require.NoError(t, err)
+	before, err := s.Epoch()
+	require.NoError(t, err)
+	require.NoError(t, s.DeleteExposure(e.ID))
+	after, err := s.Epoch()
+	require.NoError(t, err)
+	assert.Greater(t, after, before)
+}
+func TestReapingAVMDestroysItsExposures(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	e, err := s.CreateExposure(vm.ID, 8080, 0)
+	require.NoError(t, err)
+	require.NoError(t, s.TombstoneVM(vm.ID))
+	require.NoError(t, s.HardDeleteVM(vm.ID))
+	_, err = s.GetExposure(e.ID)
+	assert.ErrorIs(t, err, sql.ErrNoRows, "an exposure belongs to its VM and goes with it")
+}
+func TestForceRemovingAHostDestroysItsExposures(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	vm := makeExposureVM(t, s, h, "web-1")
+	e, err := s.CreateExposure(vm.ID, 8080, 0)
+	require.NoError(t, err)
+	purged, err := s.ForceRemoveHost(h.ID)
+	require.NoError(t, err)
+	assert.Equal(t, 1, purged)
+	_, err = s.GetExposure(e.ID)
+	assert.ErrorIs(t, err, sql.ErrNoRows)
+}

internal/server/store/store.go

Old		New
@@ -57,6 +57,11 @@ type Host struct {
57	Tenant string	57	Tenant string
58	// Host OS facts, best-effort, refreshed from the agent's Hello each connect.	58	// Host OS facts, best-effort, refreshed from the agent's Hello each connect.
59	OSID, OSPretty, OSVersion, Kernel, CPUModel, Virt string	59	OSID, OSPretty, OSVersion, Kernel, CPUModel, Virt string
		60	// UplinkAddr is the address this host presents on the network it reaches
		61	// the control plane over — the address an operator dials to reach a
		62	// published guest port. Empty means the host has not said, never that it
		63	// has none.
		64	UplinkAddr string
60	}	65	}
61		66
62	// HostFacts is the write-side shape for UpdateHostFacts (store stays pb-free;	67	// HostFacts is the write-side shape for UpdateHostFacts (store stays pb-free;
@@ -204,6 +209,32 @@ CREATE TABLE IF NOT EXISTS tenant_user_cas (
204	);	209	);
205	CREATE UNIQUE INDEX IF NOT EXISTS idx_tenant_user_cas_pubkey ON tenant_user_cas(ca_pubkey);	210	CREATE UNIQUE INDEX IF NOT EXISTS idx_tenant_user_cas_pubkey ON tenant_user_cas(ca_pubkey);
206		211
		212	-- exposures: one published guest port. The fleet binds host_port on the VM's
		213	-- host and pipes it to guest_port inside the guest. host_id is denormalized
		214	-- from the VM at create so allocation is a plain transaction against one
		215	-- table, and the unique index IS the collision check — a port is claimed by
		216	-- whoever inserts first, and the second insert fails at the storage layer
		217	-- rather than after a read-then-write race.
		218	--
		219	-- protocol is 'tcp' and scope is 'lan', both stored so 'udp' and 'public' are
		220	-- additive values later rather than a migration.
		221	--
		222	-- ON DELETE CASCADE against vms is what keeps the promise that no listener
		223	-- outlives the thing it pointed at: hard-deleting a reaped VM takes its
		224	-- exposures with it, with no sweep to write and none to forget.
		225	CREATE TABLE IF NOT EXISTS exposures (
		226	id TEXT PRIMARY KEY,
		227	tenant TEXT NOT NULL REFERENCES tenants(id),
		228	vm_id TEXT NOT NULL REFERENCES vms(id) ON DELETE CASCADE,
		229	host_id TEXT NOT NULL REFERENCES hosts(id),
		230	guest_port INTEGER NOT NULL,
		231	host_port INTEGER NOT NULL,
		232	protocol TEXT NOT NULL DEFAULT 'tcp',
		233	scope TEXT NOT NULL DEFAULT 'lan',
		234	created_at DATETIME NOT NULL
		235	);
		236	CREATE UNIQUE INDEX IF NOT EXISTS exposures_host_port ON exposures(host_id, host_port);
		237
207	-- Console sessions. Server-side so revocation works and restarts keep	238	-- Console sessions. Server-side so revocation works and restarts keep
208	-- users signed in. id is 256-bit random hex; expiry enforced on read.	239	-- users signed in. id is 256-bit random hex; expiry enforced on read.
209	CREATE TABLE IF NOT EXISTS sessions (	240	CREATE TABLE IF NOT EXISTS sessions (
@@ -282,6 +313,11 @@ func Open(path, cidrPool string) (*Store, error) {
282	{"vms", "injected_key_type", "TEXT NOT NULL DEFAULT ''"},	313	{"vms", "injected_key_type", "TEXT NOT NULL DEFAULT ''"},
283	{"vms", "injected_key_fp", "TEXT NOT NULL DEFAULT ''"},	314	{"vms", "injected_key_fp", "TEXT NOT NULL DEFAULT ''"},
284	{"vms", "injected_key_comment", "TEXT NOT NULL DEFAULT ''"},	315	{"vms", "injected_key_comment", "TEXT NOT NULL DEFAULT ''"},
		316	// The address a host presents on the network it reaches the fleet
		317	// over. Reported every tick like the guest subnet, and stored for the
		318	// same reason: the console renders `host:port` for every exposure,
		319	// including on a host that is momentarily offline.
		320	{"hosts", "uplink_addr", "TEXT NOT NULL DEFAULT ''"},
285	} {	321	} {
286	if err := ensureColumn(db, c.table, c.column, c.decl); err != nil {	322	if err := ensureColumn(db, c.table, c.column, c.decl); err != nil {
287	db.Close()	323	db.Close()
@@ -523,8 +559,8 @@ func (s *Store) GetHost(id string) (Host, error) {
523	var h Host	559	var h Host
524	var enrolledAt string	560	var enrolledAt string
525	err := s.db.QueryRow(	561	err := s.db.QueryRow(
526	`SELECT id, name, os, arch, provisioner, bridge_cidr, status, enrolled_at, cred_generation, tenant, os_id, os_pretty, os_version, kernel, cpu_model, virt FROM hosts WHERE id=?`, id,	562	`SELECT id, name, os, arch, provisioner, bridge_cidr, status, enrolled_at, cred_generation, tenant, os_id, os_pretty, os_version, kernel, cpu_model, virt, uplink_addr FROM hosts WHERE id=?`, id,
527	).Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration, &h.Tenant, &h.OSID, &h.OSPretty, &h.OSVersion, &h.Kernel, &h.CPUModel, &h.Virt)	563	).Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration, &h.Tenant, &h.OSID, &h.OSPretty, &h.OSVersion, &h.Kernel, &h.CPUModel, &h.Virt, &h.UplinkAddr)
528	if err != nil {	564	if err != nil {
529	return Host{}, err	565	return Host{}, err
530	}	566	}
@@ -572,6 +608,27 @@ func (s *Store) RecordHostNetwork(id, cidr string) error {
572	return err	608	return err
573	}	609	}
574		610
		611	// RecordHostUplink stores the address a host says it presents on the network it
		612	// reaches the fleet over. The value is free text supplied by whoever claims to
		613	// be that host, so it is checked the way every reported network fact is: it
		614	// must parse as an address. It asks whether the value IS an address, never
		615	// which address it ought to be — a host's own uplink is that host's business.
		616	//
		617	// An address that reaches nobody — loopback, unspecified, link-local,
		618	// multicast — is refused rather than stored: an operator dials this value, so a
		619	// row holding somewhere unreachable is worse than a row holding nothing. The
		620	// caller keeps what it already knows.
		621	//
		622	// The WHERE clause makes a repeat free at the storage layer; the caller still
		623	// guards the round trip, because reports arrive every tick forever.
		624	func (s *Store) RecordHostUplink(id, addr string) error {
		625	if !usableAddress(addr) {
		626	return fmt.Errorf("host uplink %q: not an address anyone can reach", addr)
		627	}
		628	_, err := s.db.Exec(`UPDATE hosts SET uplink_addr=? WHERE id=? AND uplink_addr<>?`, addr, id, addr)
		629	return err
		630	}
		631
575	// UpdateHostFacts refreshes what a host reports about itself. Every column	632	// UpdateHostFacts refreshes what a host reports about itself. Every column
576	// keeps its prior value when the reported one is EMPTY, because silence is not	633	// keeps its prior value when the reported one is EMPTY, because silence is not
577	// a statement: an agent too old to send a field, or a Hello carrying no facts	634	// a statement: an agent too old to send a field, or a Hello carrying no facts
@@ -632,7 +689,7 @@ type querier interface {
632	func (s *Store) ListHosts() ([]Host, error) { return listHosts(s.db) }	689	func (s *Store) ListHosts() ([]Host, error) { return listHosts(s.db) }
633		690
634	func listHosts(q querier) ([]Host, error) {	691	func listHosts(q querier) ([]Host, error) {
635	rows, err := q.Query(`SELECT id, name, os, arch, provisioner, bridge_cidr, status, enrolled_at, cred_generation, tenant, os_id, os_pretty, os_version, kernel, cpu_model, virt FROM hosts`)	692	rows, err := q.Query(`SELECT id, name, os, arch, provisioner, bridge_cidr, status, enrolled_at, cred_generation, tenant, os_id, os_pretty, os_version, kernel, cpu_model, virt, uplink_addr FROM hosts`)
636	if err != nil {	693	if err != nil {
637	return nil, err	694	return nil, err
638	}	695	}
@@ -641,7 +698,7 @@ func listHosts(q querier) ([]Host, error) {
641	for rows.Next() {	698	for rows.Next() {
642	var h Host	699	var h Host
643	var enrolledAt string	700	var enrolledAt string
644	if err := rows.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration, &h.Tenant, &h.OSID, &h.OSPretty, &h.OSVersion, &h.Kernel, &h.CPUModel, &h.Virt); err != nil {	701	if err := rows.Scan(&h.ID, &h.Name, &h.OS, &h.Arch, &h.Provisioner, &h.BridgeCIDR, &h.Status, &enrolledAt, &h.CredGeneration, &h.Tenant, &h.OSID, &h.OSPretty, &h.OSVersion, &h.Kernel, &h.CPUModel, &h.Virt, &h.UplinkAddr); err != nil {
645	return nil, err	702	return nil, err
646	}	703	}
647	h.EnrolledAt, _ = time.Parse(time.RFC3339, enrolledAt)	704	h.EnrolledAt, _ = time.Parse(time.RFC3339, enrolledAt)
@@ -1118,15 +1175,16 @@ func (s *Store) ForceRemoveHost(id string) (int, error) {
1118	return int(purged), nil	1175	return int(purged), nil
1119	}	1176	}
1120		1177
1121	// usableGuestAddress reports whether ip is an address a guest could actually be	1178	// usableAddress reports whether ip is an address something could actually be
1122	// reached on. It asks nothing about topology: a host's guests live wherever that	1179	// reached on — a guest on its host's network, or a host on the network it
1123	// host's network puts them — inside a Linux bridge the fleet allocated, or on	1180	// reaches the fleet over. It asks nothing about topology: a host's guests live
1124	// whatever subnet macOS's vmnet happens to run — and the fleet is told that	1181	// wherever that host's network puts them — inside a Linux bridge the fleet
1125	// subnet, it does not decide it. So the only thing worth rejecting here is a	1182	// allocated, or on whatever subnet macOS's vmnet happens to run — and the fleet
1126	// value that names no reachable guest under any topology: unparseable, the	1183	// is told that subnet, it does not decide it. So the only thing worth rejecting
		1184	// here is a value that names nobody under any topology: unparseable, the
1127	// unspecified address, loopback, link-local (the 169.254/16 a guest reports when	1185	// unspecified address, loopback, link-local (the 169.254/16 a guest reports when
1128	// DHCP never answered), or multicast.	1186	// DHCP never answered), or multicast.
1129	func usableGuestAddress(ip string) bool {	1187	func usableAddress(ip string) bool {
1130	addr, err := netip.ParseAddr(ip)	1188	addr, err := netip.ParseAddr(ip)
1131	if err != nil {	1189	if err != nil {
1132	return false	1190	return false
@@ -1149,7 +1207,7 @@ func usableGuestAddress(ip string) bool {
1149	// cache came to remember addresses that were never stored, and then suppress	1207	// cache came to remember addresses that were never stored, and then suppress
1150	// every later report that would have corrected them.	1208	// every later report that would have corrected them.
1151	func (s *Store) RecordVMStatus(id, status, lastErr, ip string) (string, error) {	1209	func (s *Store) RecordVMStatus(id, status, lastErr, ip string) (string, error) {
1152	if ip != "" && !usableGuestAddress(ip) {	1210	if ip != "" && !usableAddress(ip) {
1153	ip = ""	1211	ip = ""
1154	}	1212	}
1155		1213

internal/server/store/store_test.go

+	require.NoError(t, err)
+	assert.Equal(t, "192.168.64.0/24", got.BridgeCIDR)
+func TestRecordHostUplink(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	// A freshly enrolled host has said nothing about its uplink yet.
+	got, err := s.GetHost(h.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "", got.UplinkAddr)
+	require.NoError(t, s.RecordHostUplink(h.ID, "192.168.0.190"))
+	got, err = s.GetHost(h.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "192.168.0.190", got.UplinkAddr)
+	// The list read carries it too — the console renders it from there.
+	hosts, err := s.ListHosts()
+	require.NoError(t, err)
+	require.Len(t, hosts, 1)
+	assert.Equal(t, "192.168.0.190", hosts[0].UplinkAddr)
+	// A host that moves is recorded where it moved to.
+	require.NoError(t, s.RecordHostUplink(h.ID, "10.0.0.7"))
+	got, err = s.GetHost(h.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "10.0.0.7", got.UplinkAddr)
+}
+func TestRecordHostUplinkRefusesWhatIsNotAnAddress(t *testing.T) {
+	s := newStore(t)
+	h := enrollHost(t, s)
+	assert.Error(t, s.RecordHostUplink(h.ID, "not-an-address"))
+	assert.Error(t, s.RecordHostUplink(h.ID, "192.168.0.190/24"))
+	assert.Error(t, s.RecordHostUplink(h.ID, "127.0.0.1"), "an address that reaches nobody is not an uplink")
+	got, err := s.GetHost(h.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "", got.UplinkAddr, "a value that is not an address never reaches the row")
+}

internal/server/syncsvc/exposures_test.go

@@ @@ -0,0 +1,91 @@ @@
+package syncsvc
+import (
+	"testing"
+	"github.com/a73x/eitri/internal/pb"
+	"github.com/a73x/eitri/internal/server/registry"
+	"github.com/a73x/eitri/internal/server/store"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+// exposureVM places one live VM on the fixture's host.
+func exposureVM(t *testing.T, f *fixture, id, name string) {
+	t.Helper()
+	require.NoError(t, f.st.CreateVM(store.VM{ID: id, HostID: f.host.ID, Name: name,
+		ImageURL: "u", ImageSHA256: "s", VCPUs: 1, MemMB: 512, DiskGB: 5, PowerState: "running"}))
+}
+// TestSnapshotCarriesExposures pins that a host's desired state names every
+// exposure it should be running, at snapshot level rather than nested in a VM:
+// exposures are their own objects converging on their own cadence.
+func TestSnapshotCarriesExposures(t *testing.T) {
+	f := setup(t)
+	exposureVM(t, f, "vm1", "web-1")
+	e, err := f.st.CreateExposure("vm1", 8080, 30080)
+	require.NoError(t, err)
+	snap, err := f.svc.buildSnapshot(f.host.ID)
+	require.NoError(t, err)
+	require.Len(t, snap.GetExposures(), 1)
+	got := snap.GetExposures()[0]
+	assert.Equal(t, e.ID, got.GetId())
+	assert.Equal(t, "vm1", got.GetVmId())
+	assert.Equal(t, uint32(8080), got.GetGuestPort())
+	assert.Equal(t, uint32(30080), got.GetHostPort())
+	assert.Equal(t, "tcp", got.GetProtocol())
+}
+// TestSnapshotDropsExposuresOfATombstonedVM pins that a listener stops being
+// desired the moment its VM is on its way out, rather than when its row is
+// finally reaped.
+func TestSnapshotDropsExposuresOfATombstonedVM(t *testing.T) {
+	f := setup(t)
+	exposureVM(t, f, "vm1", "web-1")
+	_, err := f.st.CreateExposure("vm1", 8080, 30080)
+	require.NoError(t, err)
+	require.NoError(t, f.st.TombstoneVM("vm1"))
+	snap, err := f.svc.buildSnapshot(f.host.ID)
+	require.NoError(t, err)
+	assert.Empty(t, snap.GetExposures())
+}
+// TestApplyReportFoldsExposureStateIntoTheRegistry pins the return leg: what a
+// host says its listeners are doing is live state, held in memory beside the
+// VM rows it reports in the same message.
+func TestApplyReportFoldsExposureStateIntoTheRegistry(t *testing.T) {
+	f := setup(t)
+	f.svc.applyReport(f.host.ID, &pb.ActualStateReport{
+		Exposures: []*pb.ExposureActual{
+			{Id: "e1", State: "active"},
+			{Id: "e2", State: "failed", Reason: "listen tcp 0.0.0.0:30080: bind: address already in use"},
+		},
+	})
+	got, ok := f.reg.Get(f.host.ID)
+	require.True(t, ok)
+	require.Len(t, got.Report.Exposures, 2)
+	assert.Equal(t, registry.ExposureStatus{ID: "e1", State: "active"}, got.Report.Exposures[0])
+	assert.Equal(t, "failed", got.Report.Exposures[1].State)
+	assert.Contains(t, got.Report.Exposures[1].Reason, "address already in use")
+}
+// TestApplyReportRecordsTheHostUplinkAddress pins that the address an operator
+// dials is durable, and that a report which carries none leaves it alone.
+func TestApplyReportRecordsTheHostUplinkAddress(t *testing.T) {
+	f := setup(t)
+	f.svc.applyReport(f.host.ID, &pb.ActualStateReport{HostUplinkAddr: "192.168.0.190"})
+	h, err := f.st.GetHost(f.host.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "192.168.0.190", h.UplinkAddr)
+	f.svc.applyReport(f.host.ID, &pb.ActualStateReport{})
+	h, err = f.st.GetHost(f.host.ID)
+	require.NoError(t, err)
+	assert.Equal(t, "192.168.0.190", h.UplinkAddr, "silence is not a statement")
+}

internal/server/syncsvc/syncsvc.go

Old		New
@@ -48,6 +48,10 @@ type Service struct {
48	recorder vmStatusRecorder	48	recorder vmStatusRecorder
49	tracker *statusTracker	49	tracker *statusTracker
50	netTrack *netTracker	50	netTrack *netTracker
		51	// uplinkTrack remembers the uplink address each host last had WRITTEN, for
		52	// the same reason netTrack does: every host reports every tick forever, and
		53	// the store runs on a single connection.
		54	uplinkTrack *netTracker
51	// maxCredAge, when non-zero, rejects credentials whose issued-at is older.	55	// maxCredAge, when non-zero, rejects credentials whose issued-at is older.
52	// Zero disables the age check (default: expiry without an auto-renewal	56	// Zero disables the age check (default: expiry without an auto-renewal
53	// channel would force periodic re-enrolls; per-host generation revocation	57	// channel would force periodic re-enrolls; per-host generation revocation
@@ -83,7 +87,7 @@ func newWithWriteTimeout(st store.Store, reg registry.Registry, h *hub.Hub, se
83	}	87	}
84	return &Service{st: st, reg: reg, hub: h, secret: secret, maxCredAge: maxCredAge, writeTimeout: writeTimeout,	88	return &Service{st: st, reg: reg, hub: h, secret: secret, maxCredAge: maxCredAge, writeTimeout: writeTimeout,
85	conns: map[string]quic.Connection{}, recorder: st, tracker: newStatusTracker(), netTrack: newNetTracker(),	89	conns: map[string]quic.Connection{}, recorder: st, tracker: newStatusTracker(), netTrack: newNetTracker(),
86	offers: map[string]*pb.AgentUpgrade{}}	90	uplinkTrack: newNetTracker(), offers: map[string]*pb.AgentUpgrade{}}
87	}	91	}
88		92
89	// Serve accepts QUIC connections until ctx is cancelled.	93	// Serve accepts QUIC connections until ctx is cancelled.
@@ -202,11 +206,12 @@ func (s *Service) handleConn(ctx context.Context, conn quic.Connection) {
202	delete(s.conns, hostID)	206	delete(s.conns, hostID)
203	}	207	}
204	s.consoleMu.Unlock()	208	s.consoleMu.Unlock()
205	// Forget this host's cached guest subnet, so the next connection writes	209	// Forget this host's cached network facts, so the next connection writes
206	// it once rather than trusting a memory of a row that may have been	210	// them once rather than trusting a memory of a row that may have been
207	// removed, re-enrolled or edited while the host was away. Bounds the map	211	// removed, re-enrolled or edited while the host was away. Bounds the maps
208	// to connected hosts, and costs one write per reconnect.	212	// to connected hosts, and costs one write per reconnect.
209	s.netTrack.forget(hostID)	213	s.netTrack.forget(hostID)
		214	s.uplinkTrack.forget(hostID)
210	}()	215	}()
211		216
212	// Single writer for the down-stream: the poke goroutine.	217	// Single writer for the down-stream: the poke goroutine.
@@ -270,11 +275,12 @@ func (s *Service) handleConn(ctx context.Context, conn quic.Connection) {
270	}	275	}
271	}	276	}
272		277
273	// pushSnapshot reads the current desired state in a single transaction and sends it.	278	// buildSnapshot reads hostID's desired state — its VMs and the exposures it
274	func (s *Service) pushSnapshot(down quic.Stream, hostID string) error {	279	// should be serving — and renders it as one snapshot.
		280	func (s Service) buildSnapshot(hostID string) (pb.DesiredStateSnapshot, error) {
275	epoch, vms, err := s.st.DesiredForHost(hostID)	281	epoch, vms, err := s.st.DesiredForHost(hostID)
276	if err != nil {	282	if err != nil {
277	return fmt.Errorf("desired for host: %w", err)	283	return nil, fmt.Errorf("desired for host: %w", err)
278	}	284	}
279	snap := &pb.DesiredStateSnapshot{Epoch: epoch, Vms: make([]*pb.VMDesired, 0, len(vms))}	285	snap := &pb.DesiredStateSnapshot{Epoch: epoch, Vms: make([]*pb.VMDesired, 0, len(vms))}
280	snap.AgentUpgrade = s.offerFor(hostID)	286	snap.AgentUpgrade = s.offerFor(hostID)
@@ -284,7 +290,7 @@ func (s *Service) pushSnapshot(down quic.Stream, hostID string) error {
284	if !ok {	290	if !ok {
285	list, err := s.st.ListTenantUserCAs(v.Tenant)	291	list, err := s.st.ListTenantUserCAs(v.Tenant)
286	if err != nil {	292	if err != nil {
287	return fmt.Errorf("list tenant user cas: %w", err)	293	return nil, fmt.Errorf("list tenant user cas: %w", err)
288	}	294	}
289	cas = make([]string, 0, len(list))	295	cas = make([]string, 0, len(list))
290	for _, c := range list {	296	for _, c := range list {
@@ -302,6 +308,28 @@ func (s *Service) pushSnapshot(down quic.Stream, hostID string) error {
302	SshHostCert: v.SSHHostCert,	308	SshHostCert: v.SSHHostCert,
303	})	309	})
304	}	310	}
		311	exps, err := s.st.ListExposuresForHost(hostID)
		312	if err != nil {
		313	return nil, fmt.Errorf("list host exposures: %w", err)
		314	}
		315	for _, e := range exps {
		316	// In-range by construction: the API validates ports at create
		317	// (validateExposurePorts), and the store accepts rows only from the API.
		318	snap.Exposures = append(snap.Exposures, &pb.ExposureDesired{
		319	Id: e.ID, VmId: e.VMID,
		320	GuestPort: uint32(e.GuestPort), HostPort: uint32(e.HostPort),
		321	Protocol: e.Protocol,
		322	})
		323	}
		324	return snap, nil
		325	}
		326
		327	// pushSnapshot sends hostID's current desired state down the stream.
		328	func (s *Service) pushSnapshot(down quic.Stream, hostID string) error {
		329	snap, err := s.buildSnapshot(hostID)
		330	if err != nil {
		331	return err
		332	}
305	// Bound the write: if a stalled agent stops reading the down-stream but keeps	333	// Bound the write: if a stalled agent stops reading the down-stream but keeps
306	// the connection alive, the flow-control window fills and an unbounded Write	334	// the connection alive, the flow-control window fills and an unbounded Write
307	// would block forever, pinning this goroutine. The deadline turns that into a	335	// would block forever, pinning this goroutine. The deadline turns that into a
@@ -369,6 +397,7 @@ func (s Service) applyReport(hostID string, rep pb.ActualStateReport) {
369		397
370	r.VMs = toRegistryVMs(rep.GetVms())	398	r.VMs = toRegistryVMs(rep.GetVms())
371	r.Quarantined = toRegistryQuarantined(rep.GetQuarantined())	399	r.Quarantined = toRegistryQuarantined(rep.GetQuarantined())
		400	r.Exposures = toRegistryExposures(rep.GetExposures())
372	r.Capacity = toRegistryCapacity(rep.GetCapacity())	401	r.Capacity = toRegistryCapacity(rep.GetCapacity())
373	r.Metrics = toRegistryMetrics(rep.GetMetrics())	402	r.Metrics = toRegistryMetrics(rep.GetMetrics())
374		403
@@ -412,6 +441,16 @@ func (s Service) applyReport(hostID string, rep pb.ActualStateReport) {
412	}	441	}
413	}	442	}
414		443
		444	// The address this host says it answers on. Empty is "no answer", never
		445	// "no address" — the same rule guest_cidr follows, and for the same reason.
		446	if addr := rep.GetHostUplinkAddr(); addr != "" {
		447	if err := s.uplinkTrack.writeThrough(hostID, addr, func() error {
		448	return s.st.RecordHostUplink(hostID, addr)
		449	}); err != nil {
		450	slog.Warn("record host uplink", "host", hostID, "addr", addr, "err", err)
		451	}
		452	}
		453
415	// Fence violation: log ERROR and point at the restore runbook.	454	// Fence violation: log ERROR and point at the restore runbook.
416	if rep.GetFenceViolation() {	455	if rep.GetFenceViolation() {
417	slog.Error("agent refused snapshot: epoch fence violation — see restore runbook",	456	slog.Error("agent refused snapshot: epoch fence violation — see restore runbook",
@@ -498,6 +537,21 @@ func toRegistryQuarantined(in []*pb.QuarantinedVM) []registry.QuarantinedVM {
498	return out	537	return out
499	}	538	}
500		539
		540	// toRegistryExposures maps reported exposure state to registry rows. Returns
		541	// nil (not an empty slice) for empty input, matching append-into-nil behavior.
		542	func toRegistryExposures(in []*pb.ExposureActual) []registry.ExposureStatus {
		543	if len(in) == 0 {
		544	return nil
		545	}
		546	out := make([]registry.ExposureStatus, 0, len(in))
		547	for _, e := range in {
		548	out = append(out, registry.ExposureStatus{
		549	ID: e.GetId(), State: e.GetState(), Reason: e.GetReason(),
		550	})
		551	}
		552	return out
		553	}
		554
501	// toRegistryCapacity maps reported capacity (nil → zero value).	555	// toRegistryCapacity maps reported capacity (nil → zero value).
502	func toRegistryCapacity(c *pb.Capacity) registry.Capacity {	556	func toRegistryCapacity(c *pb.Capacity) registry.Capacity {
503	if c == nil {	557	if c == nil {

internal/server/syncsvc/tracker.go

Old		New
@@ -78,12 +78,14 @@ func (t *statusTracker) forget(vmID string) {
78	delete(t.last, vmID)	78	delete(t.last, vmID)
79	}	79	}
80		80
81	// netTracker remembers the guest subnet each host last had WRITTEN, so a report	81	// netTracker remembers the last value durably WRITTEN per host — one instance
82	// that repeats it costs nothing. The guard is not premature: every host reports	82	// per fact, currently the host's guest subnet and the address it answers on —
83	// every tick — ten seconds by default — forever, and the store runs on a single	83	// so a report that repeats it costs nothing. The guard is not premature: every
84	// connection, so an unguarded write spends a round trip on that connection for	84	// host reports every tick — ten seconds by default — forever, and the store runs
85	// every host for the life of the fleet. The SQL's own `WHERE bridge_cidr<>?`	85	// on a single connection, so an unguarded write spends a round trip on that
86	// makes a concurrent reconnect harmless; this makes the steady state free.	86	// connection for every host for the life of the fleet. Each write's own
		87	// `WHERE <column><>?` makes a concurrent reconnect harmless; this makes the
		88	// steady state free.
87	//	89	//
88	// Keyed by hostID. Only a SUCCESSFUL write is remembered, so a rejected one is	90	// Keyed by hostID. Only a SUCCESSFUL write is remembered, so a rejected one is
89	// retried by the next report rather than suppressed — the same rule the status	91	// retried by the next report rather than suppressed — the same rule the status
@@ -98,20 +100,20 @@ func newNetTracker() *netTracker { return &netTracker{last: map[string]string{}}
98	// writeThrough runs decide→write→commit for one host under a single lock, so	100	// writeThrough runs decide→write→commit for one host under a single lock, so
99	// two connections for the same host cannot interleave and leave the cache	101	// two connections for the same host cannot interleave and leave the cache
100	// disagreeing with the row.	102	// disagreeing with the row.
101	func (t *netTracker) writeThrough(hostID, cidr string, write func() error) error {	103	func (t *netTracker) writeThrough(hostID, value string, write func() error) error {
102	t.mu.Lock()	104	t.mu.Lock()
103	defer t.mu.Unlock()	105	defer t.mu.Unlock()
104	if t.last[hostID] == cidr {	106	if t.last[hostID] == value {
105	return nil	107	return nil
106	}	108	}
107	if err := write(); err != nil {	109	if err := write(); err != nil {
108	return err	110	return err
109	}	111	}
110	t.last[hostID] = cidr	112	t.last[hostID] = value
111	return nil	113	return nil
112	}	114	}
113		115
114	// forget drops a host's cached subnet, so a host that leaves and returns is	116	// forget drops a host's cached value, so a host that leaves and returns is
115	// written afresh rather than trusted to a memory of a row that may be gone.	117	// written afresh rather than trusted to a memory of a row that may be gone.
116	func (t *netTracker) forget(hostID string) {	118	func (t *netTracker) forget(hostID string) {
117	t.mu.Lock()	119	t.mu.Lock()

internal/smoke/exposure.go

@@ @@ -0,0 +1,110 @@ @@
+package smoke
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+// sshBannerPrefix is what an sshd answers a fresh connection with, before any
+// key exchange. Reading it back through a published port proves the whole path
+// — the host listener, the splice, and the guest dial — with no infrastructure
+// the fleet does not already have, and no privileged bind on either machine.
+const sshBannerPrefix = "SSH-2.0"
+// bannerFunc reads the first bytes a TCP peer sends after accepting.
+type bannerFunc func(ctx context.Context, addr string) (string, error)
+// proveExposure publishes the smoke VM's ssh port on its host, dials the host
+// address raw, and expects an SSH banner. hostAddr is the address the host
+// itself reported; the exposure is revoked before the leg returns, whatever
+// the outcome.
+func proveExposure(ctx context.Context, c vmAPI, hostAddr, vmID string, now func() time.Time, sleep func(time.Duration), dial bannerFunc) error {
+	if hostAddr == "" {
+		return errors.New("FAIL: the VM's host reported no uplink address; nothing to dial a published port on")
+	}
+	exp, err := c.CreateExposure(ctx, vmID, 22, 0)
+	if err != nil {
+		return fmt.Errorf("create exposure: %w", err)
+	}
+	defer func() {
+		// Revoke on the way out even when the leg failed: the smoke leaves no
+		// grant behind on the live fleet. A revoke that itself fails leaves a
+		// published port on a live host, so say so — the leg's own verdict
+		// stands either way.
+		if err := c.DeleteExposure(context.WithoutCancel(ctx), exp.ID); err != nil {
+			fmt.Fprintf(os.Stderr, "eitri-smoke: exposure revoke failed; grant %s left behind: %v\n", exp.ID, err)
+		}
+	}()
+	target := net.JoinHostPort(hostAddr, strconv.FormatInt(exp.HostPort, 10))
+	var lastErr error
+	err = pollLoop(ctx, now, sleep, 60*time.Second, 3*time.Second, func() (bool, error) {
+		banner, derr := dial(ctx, target)
+		if derr != nil {
+			// The listener binds on the host's next converge, a tick away — a
+			// refused dial in that window is expected, not fatal.
+			lastErr = derr
+			return false, nil
+		}
+		if !strings.HasPrefix(banner, sshBannerPrefix) {
+			return false, fmt.Errorf("FAIL: published port %s answered %q, want an %s banner",
+				target, truncateBanner(banner), sshBannerPrefix)
+		}
+		return true, nil
+	})
+	if errors.Is(err, errPollTimeout) {
+		return fmt.Errorf("FAIL: no %s banner from published port %s within 60s: %v", sshBannerPrefix, target, lastErr)
+	}
+	return err
+}
+// truncateBanner bounds what a wrong answer puts in the failure message.
+func truncateBanner(s string) string {
+	s = strings.TrimSpace(s)
+	if len(s) > 64 {
+		return s[:64]
+	}
+	return s
+}
+// readBanner is the real bannerFunc: dial addr and read the peer's first line,
+// bounded by a deadline so a listener that binds but never answers fails the
+// leg instead of hanging it. It reads until a newline (or 256 bytes) rather
+// than taking one Read's worth, because a spliced connection can deliver the
+// banner in fragments and a segment shorter than the prefix would fail a
+// perfectly good sshd. A read that returns nothing at all keeps its error, so
+// the caller can retry a listener that is not up yet.
+func readBanner(ctx context.Context, addr string) (string, error) {
+	var d net.Dialer
+	conn, err := d.DialContext(ctx, "tcp", addr)
+	if err != nil {
+		return "", err
+	}
+	defer conn.Close()
+	if err := conn.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil {
+		return "", err
+	}
+	buf := make([]byte, 256)
+	n := 0
+	for n < len(buf) {
+		m, err := conn.Read(buf[n:])
+		n += m
+		if err != nil {
+			if n == 0 {
+				return "", err
+			}
+			break
+		}
+		if bytes.IndexByte(buf[:n], '\n') >= 0 {
+			break
+		}
+	}
+	return string(buf[:n]), nil
+}

internal/smoke/exposure_test.go

@@ @@ -0,0 +1,191 @@ @@
+package smoke
+import (
+	"context"
+	"errors"
+	"net"
+	"strings"
+	"testing"
+	"time"
+	"github.com/a73x/eitri/internal/server/api/client"
+)
+func TestProveExposureReadsTheBannerBack(t *testing.T) {
+	clock := &fakeClock{}
+	deleted := ""
+	api := &testAPI{
+		createExposureFunc: func(ctx context.Context, vmID string, guest, host int64) (client.Exposure, error) {
+			if vmID != "vm-1" {
+				t.Errorf("CreateExposure vmID = %q, want vm-1", vmID)
+			}
+			if guest != 22 {
+				t.Errorf("CreateExposure guestPort = %d, want 22", guest)
+			}
+			if host != 0 {
+				t.Errorf("CreateExposure hostPort = %d, want 0 — the fleet allocates the host port", host)
+			}
+			return client.Exposure{ID: "x-1", HostPort: 30080}, nil
+		},
+		deleteExposureFunc: func(ctx context.Context, id string) error { deleted = id; return nil },
+	}
+	dialed := ""
+	dial := func(ctx context.Context, addr string) (string, error) {
+		dialed = addr
+		return "SSH-2.0-OpenSSH_9.6\r\n", nil
+	}
+	if err := proveExposure(context.Background(), api, "192.168.0.190", "vm-1", clock.now, clock.sleep, dial); err != nil {
+		t.Fatalf("proveExposure: %v", err)
+	}
+	if dialed != "192.168.0.190:30080" {
+		t.Errorf("dialed = %q, want 192.168.0.190:30080", dialed)
+	}
+	if deleted != "x-1" {
+		t.Errorf("revoked exposure = %q, want x-1 — the smoke leaves no exposure behind", deleted)
+	}
+}
+// TestProveExposureDialsAnIPv6Uplink pins the dial target's composition: a host
+// that answers on an IPv6 address only resolves when the address is bracketed,
+// which is what net.JoinHostPort does and plain concatenation does not.
+func TestProveExposureDialsAnIPv6Uplink(t *testing.T) {
+	clock := &fakeClock{}
+	api := &testAPI{
+		createExposureFunc: func(context.Context, string, int64, int64) (client.Exposure, error) {
+			return client.Exposure{ID: "x-1", HostPort: 30080}, nil
+		},
+		deleteExposureFunc: func(context.Context, string) error { return nil },
+	}
+	dialed := ""
+	dial := func(ctx context.Context, addr string) (string, error) {
+		dialed = addr
+		return "SSH-2.0-OpenSSH_9.6\r\n", nil
+	}
+	if err := proveExposure(context.Background(), api, "2001:db8::1", "vm-1", clock.now, clock.sleep, dial); err != nil {
+		t.Fatalf("proveExposure: %v", err)
+	}
+	if dialed != "[2001:db8::1]:30080" {
+		t.Errorf("dialed = %q, want [2001:db8::1]:30080", dialed)
+	}
+}
+func TestProveExposureRetriesUntilTheListenerConverges(t *testing.T) {
+	clock := &fakeClock{}
+	api := &testAPI{
+		createExposureFunc: func(context.Context, string, int64, int64) (client.Exposure, error) {
+			return client.Exposure{ID: "x-1", HostPort: 30080}, nil
+		},
+		deleteExposureFunc: func(context.Context, string) error { return nil },
+	}
+	calls := 0
+	dial := func(ctx context.Context, addr string) (string, error) {
+		calls++
+		if calls < 3 {
+			return "", errors.New("connection refused")
+		}
+		return "SSH-2.0-OpenSSH_9.6\r\n", nil
+	}
+	if err := proveExposure(context.Background(), api, "192.168.0.190", "vm-1", clock.now, clock.sleep, dial); err != nil {
+		t.Fatalf("proveExposure: %v", err)
+	}
+	if calls != 3 {
+		t.Errorf("dial calls = %d, want 3", calls)
+	}
+}
+func TestProveExposureFailsOnTheWrongBanner(t *testing.T) {
+	clock := &fakeClock{}
+	api := &testAPI{
+		createExposureFunc: func(context.Context, string, int64, int64) (client.Exposure, error) {
+			return client.Exposure{ID: "x-1", HostPort: 30080}, nil
+		},
+		deleteExposureFunc: func(context.Context, string) error { return nil },
+	}
+	dial := func(ctx context.Context, addr string) (string, error) { return "HTTP/1.1 400\r\n", nil }
+	err := proveExposure(context.Background(), api, "192.168.0.190", "vm-1", clock.now, clock.sleep, dial)
+	if err == nil {
+		t.Fatal("proveExposure: want error for a non-sshd answer, got nil")
+	}
+	if !strings.Contains(err.Error(), "SSH-2.0") {
+		t.Errorf("error = %q, want it to name the banner it wanted", err.Error())
+	}
+}
+func TestProveExposureFailsWhenTheHostNamesNoAddress(t *testing.T) {
+	clock := &fakeClock{}
+	api := &testAPI{}
+	err := proveExposure(context.Background(), api, "", "vm-1", clock.now, clock.sleep, nil)
+	if err == nil {
+		t.Fatal("proveExposure: want error when the host reported no address, got nil")
+	}
+	if !strings.Contains(err.Error(), "uplink") {
+		t.Errorf("error = %q, want it to mention the uplink address", err.Error())
+	}
+}
+// bannerListener serves one connection, writing each of writes in turn with a
+// pause between them, and returns the address to dial it on.
+func bannerListener(t *testing.T, writes ...string) string {
+	t.Helper()
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	t.Cleanup(func() { ln.Close() })
+	go func() {
+		c, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		defer c.Close()
+		for _, w := range writes {
+			if _, err := c.Write([]byte(w)); err != nil {
+				return
+			}
+			time.Sleep(20 * time.Millisecond)
+		}
+		time.Sleep(50 * time.Millisecond)
+	}()
+	return ln.Addr().String()
+}
+func TestReadBannerReadsWhatAListenerWrites(t *testing.T) {
+	got, err := readBanner(context.Background(), bannerListener(t, "SSH-2.0-Test\r\n"))
+	if err != nil {
+		t.Fatalf("readBanner: %v", err)
+	}
+	if !strings.HasPrefix(got, "SSH-2.0") {
+		t.Errorf("banner = %q, want an SSH-2.0 prefix", got)
+	}
+}
+// TestReadBannerReassemblesAFragmentedBanner pins the read-to-newline: a
+// spliced connection can deliver the banner in segments, and a first segment
+// shorter than the prefix must not fail a listener that is answering
+// correctly.
+func TestReadBannerReassemblesAFragmentedBanner(t *testing.T) {
+	got, err := readBanner(context.Background(), bannerListener(t, "SS", "H-2.0-Test\r\n"))
+	if err != nil {
+		t.Fatalf("readBanner: %v", err)
+	}
+	if !strings.HasPrefix(got, "SSH-2.0") {
+		t.Errorf("banner = %q, want the fragments reassembled into an SSH-2.0 prefix", got)
+	}
+}
+// TestReadBannerKeepsTheErrorWhenNothingArrives pins the retryable case: a
+// listener that accepts and hangs up without a byte returns its error, which
+// the leg treats as "not converged yet" rather than as a wrong answer.
+func TestReadBannerKeepsTheErrorWhenNothingArrives(t *testing.T) {
+	got, err := readBanner(context.Background(), bannerListener(t))
+	if err == nil {
+		t.Fatalf("readBanner = %q, want the read error from a silent listener", got)
+	}
+	if got != "" {
+		t.Errorf("banner = %q, want empty", got)
+	}
+}

internal/smoke/run.go

Old		New
@@ -97,7 +97,7 @@ func Run() error {
97	}	97	}
98		98
99	ctx := context.Background()	99	ctx := context.Background()
100	msg, err := runScenario(ctx, cfg, vmName, api, realRunSSH(cfg.AgentUserHost, cfg.AgentPort), gate, time.Now, time.Sleep, realReadPubKey)	100	msg, err := runScenario(ctx, cfg, vmName, api, realRunSSH(cfg.AgentUserHost, cfg.AgentPort), gate, time.Now, time.Sleep, realReadPubKey, readBanner)
101	if err != nil {	101	if err != nil {
102	return err	102	return err
103	}	103	}

internal/smoke/scenario.go

Old		New
@@ -72,6 +72,8 @@ type vmAPI interface {
72	ListVMs(ctx context.Context) ([]client.VM, error)	72	ListVMs(ctx context.Context) ([]client.VM, error)
73	DeleteVM(ctx context.Context, id string) error	73	DeleteVM(ctx context.Context, id string) error
74	PatchVM(ctx context.Context, id, powerState string) error	74	PatchVM(ctx context.Context, id, powerState string) error
		75	CreateExposure(ctx context.Context, vmID string, guestPort, hostPort int64) (client.Exposure, error)
		76	DeleteExposure(ctx context.Context, id string) error
75	}	77	}
76		78
77	// getVM finds the VM with the given id in the current listing. The bool	79	// getVM finds the VM with the given id in the current listing. The bool
@@ -90,6 +92,21 @@ func getVM(ctx context.Context, c vmAPI, id string) (client.VM, bool, error) {
90	return client.VM{}, false, nil	92	return client.VM{}, false, nil
91	}	93	}
92		94
		95	// uplinkFor re-reads the host's own reported address — the one an operator
		96	// pastes to reach a port published on it.
		97	func uplinkFor(ctx context.Context, c vmAPI, hostID string) (string, error) {
		98	hosts, err := c.ListHosts(ctx)
		99	if err != nil {
		100	return "", fmt.Errorf("list hosts for uplink address: %w", err)
		101	}
		102	for _, h := range hosts {
		103	if h.ID == hostID {
		104	return h.UplinkAddr, nil
		105	}
		106	}
		107	return "", fmt.Errorf("host %s vanished from the fleet mid-run", hostID)
		108	}
		109
93	// sshFunc runs remoteCmd on the agent host over SSH and returns its stdout.	110	// sshFunc runs remoteCmd on the agent host over SSH and returns its stdout.
94	type sshFunc func(ctx context.Context, remoteCmd string) (string, error)	111	type sshFunc func(ctx context.Context, remoteCmd string) (string, error)
95		112
@@ -124,8 +141,9 @@ func pollLoop(ctx context.Context, now func() time.Time, sleep func(time.Duratio
124	}	141	}
125		142
126	// runScenario drives the full register -> create -> ready -> boot-proof ->	143	// runScenario drives the full register -> create -> ready -> boot-proof ->
127	// gate-exec -> reap sequence against c (the API), runSSH (the boot-proof	144	// gate-exec -> published-port -> reap sequence against c (the API), runSSH
128	// transport), and readPubKey (the local SSH key source). vmName is the	145	// (the boot-proof transport), readPubKey (the local SSH key source), and
		146	// dialBanner (the published-port transport). vmName is the
129	// pre-generated name for the throwaway VM. gate, when non-nil, registers the	147	// pre-generated name for the throwaway VM. gate, when non-nil, registers the
130	// smoke's user CA with the tenant before create (the guest bakes its trusted	148	// smoke's user CA with the tenant before create (the guest bakes its trusted
131	// CAs at boot, so registration MUST happen first) and proves gate SSH access	149	// CAs at boot, so registration MUST happen first) and proves gate SSH access
@@ -133,7 +151,7 @@ func pollLoop(ctx context.Context, now func() time.Time, sleep func(time.Duratio
133	// now/sleep are the injected clock so the poll deadlines are unit-testable	151	// now/sleep are the injected clock so the poll deadlines are unit-testable
134	// without real waiting. On success it returns the human-readable COMPLETE	152	// without real waiting. On success it returns the human-readable COMPLETE
135	// line; on any failure it returns a descriptive error.	153	// line; on any failure it returns a descriptive error.
136	func runScenario(ctx context.Context, cfg Config, vmName string, c vmAPI, runSSH sshFunc, gate *gateHooks, now func() time.Time, sleep func(time.Duration), readPubKey func() string) (string, error) {	154	func runScenario(ctx context.Context, cfg Config, vmName string, c vmAPI, runSSH sshFunc, gate *gateHooks, now func() time.Time, sleep func(time.Duration), readPubKey func() string, dialBanner bannerFunc) (string, error) {
137	if gate != nil {	155	if gate != nil {
138	if err := gate.register(ctx); err != nil {	156	if err := gate.register(ctx); err != nil {
139	return "", fmt.Errorf("register smoke user CA: %w", err)	157	return "", fmt.Errorf("register smoke user CA: %w", err)
@@ -180,6 +198,7 @@ func runScenario(ctx context.Context, cfg Config, vmName string, c vmAPI, runSSH
180	}	198	}
181		199
182	gateOK := false	200	gateOK := false
		201	exposureOK := false
183	if gate != nil {	202	if gate != nil {
184	if err := gate.exec(ctx, vmName); err != nil {	203	if err := gate.exec(ctx, vmName); err != nil {
185	return "", err	204	return "", err
@@ -187,6 +206,19 @@ func runScenario(ctx context.Context, cfg Config, vmName string, c vmAPI, runSSH
187	gateOK = true	206	gateOK = true
188	}	207	}
189		208
		209	// Publish the guest's own sshd on its host and read the banner back
		210	// through the listener. The host's address comes from the host itself,
		211	// re-read here rather than reused from the placement lookup, so the leg
		212	// dials whatever the host is answering on right now.
		213	hostAddr, err := uplinkFor(ctx, c, hostID)
		214	if err != nil {
		215	return "", err
		216	}
		217	if err := proveExposure(ctx, c, hostAddr, vmID, now, sleep, dialBanner); err != nil {
		218	return "", err
		219	}
		220	exposureOK = true
		221
190	// Power cycle: prove the guest comes BACK. A first boot runs on the	222	// Power cycle: prove the guest comes BACK. A first boot runs on the
191	// kernel's in-memory partition table; only a stop→start proves the	223	// kernel's in-memory partition table; only a stop→start proves the
192	// on-disk GPT survived growpart. The sector-0 regression hid exactly	224	// on-disk GPT survived growpart. The sector-0 regression hid exactly
@@ -257,5 +289,8 @@ func runScenario(ctx context.Context, cfg Config, vmName string, c vmAPI, runSSH
257	if gateOK {	289	if gateOK {
258	msg += ", gate SSH: ok"	290	msg += ", gate SSH: ok"
259	}	291	}
		292	if exposureOK {
		293	msg += ", exposed port: ok"
		294	}
260	return msg, nil	295	return msg, nil
261	}	296	}

internal/smoke/scenario_test.go

Old		New
@@ -54,14 +54,17 @@ type fakeClock struct{ t time.Time }
54	func (c *fakeClock) now() time.Time { return c.t }	54	func (c *fakeClock) now() time.Time { return c.t }
55	func (c *fakeClock) sleep(d time.Duration) { c.t = c.t.Add(d) }	55	func (c *fakeClock) sleep(d time.Duration) { c.t = c.t.Add(d) }
56		56
57	// testAPI implements vmAPI by delegating to per-test closures. patchVMFunc	57	// testAPI implements vmAPI by delegating to per-test closures. patchVMFunc,
58	// defaults to accepting any power-state change when nil.	58	// createExposureFunc, and deleteExposureFunc default to a benign answer when
		59	// nil, so a test whose subject is elsewhere need not supply them.
59	type testAPI struct {	60	type testAPI struct {
60	listHostsFunc func(ctx context.Context) ([]client.Host, error)	61	listHostsFunc func(ctx context.Context) ([]client.Host, error)
61	createVMFunc func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error)	62	createVMFunc func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error)
62	listVMsFunc func(ctx context.Context) ([]client.VM, error)	63	listVMsFunc func(ctx context.Context) ([]client.VM, error)
63	deleteVMFunc func(ctx context.Context, id string) error	64	deleteVMFunc func(ctx context.Context, id string) error
64	patchVMFunc func(ctx context.Context, id, powerState string) error	65	patchVMFunc func(ctx context.Context, id, powerState string) error
		66	createExposureFunc func(ctx context.Context, vmID string, guestPort, hostPort int64) (client.Exposure, error)
		67	deleteExposureFunc func(ctx context.Context, id string) error
65	}	68	}
66		69
67	func (a *testAPI) ListHosts(ctx context.Context) ([]client.Host, error) {	70	func (a *testAPI) ListHosts(ctx context.Context) ([]client.Host, error) {
@@ -80,6 +83,19 @@ func (a *testAPI) PatchVM(ctx context.Context, id, powerState string) error {
80	}	83	}
81	return a.patchVMFunc(ctx, id, powerState)	84	return a.patchVMFunc(ctx, id, powerState)
82	}	85	}
		86	func (a *testAPI) CreateExposure(ctx context.Context, vmID string, guestPort, hostPort int64) (client.Exposure, error) {
		87	if a.createExposureFunc == nil {
		88	return client.Exposure{ID: "x-fake", HostPort: 30080}, nil
		89	}
		90	return a.createExposureFunc(ctx, vmID, guestPort, hostPort)
		91	}
		92
		93	func (a *testAPI) DeleteExposure(ctx context.Context, id string) error {
		94	if a.deleteExposureFunc == nil {
		95	return nil
		96	}
		97	return a.deleteExposureFunc(ctx, id)
		98	}
83		99
84	func baseCfg() Config {	100	func baseCfg() Config {
85	return Config{AgentStateDir: "/var/lib/eitri-agent", AgentUserHost: "ubuntu@10.0.0.5", AgentPort: 22}	101	return Config{AgentStateDir: "/var/lib/eitri-agent", AgentUserHost: "ubuntu@10.0.0.5", AgentPort: 22}
@@ -87,6 +103,10 @@ func baseCfg() Config {
87		103
88	func noopReadPubKey() string { return "ssh-ed25519 AAAAfake test@smoke" }	104	func noopReadPubKey() string { return "ssh-ed25519 AAAAfake test@smoke" }
89		105
		106	// okBanner is the banner a converged listener answers with, for tests whose
		107	// subject is not the exposure leg.
		108	func okBanner(ctx context.Context, addr string) (string, error) { return "SSH-2.0-Test\r\n", nil }
		109
90	// --- runScenario: success path ---------------------------------------------	110	// --- runScenario: success path ---------------------------------------------
91		111
92	func TestRunScenarioSuccess(t *testing.T) {	112	func TestRunScenarioSuccess(t *testing.T) {
@@ -95,7 +115,7 @@ func TestRunScenarioSuccess(t *testing.T) {
95	deleted := false	115	deleted := false
96	api := &testAPI{	116	api := &testAPI{
97	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {	117	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {
98	return []client.Host{{ID: "host-1"}}, nil	118	return []client.Host{{ID: "host-1", UplinkAddr: "192.168.0.190"}}, nil
99	},	119	},
100	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {	120	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {
101	if req.HostID != "host-1" {	121	if req.HostID != "host-1" {
@@ -149,7 +169,7 @@ func TestRunScenarioSuccess(t *testing.T) {
149	}	169	}
150		170
151	clock := &fakeClock{t: time.Unix(0, 0)}	171	clock := &fakeClock{t: time.Unix(0, 0)}
152	msg, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey)	172	msg, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey, okBanner)
153	if err != nil {	173	if err != nil {
154	t.Fatalf("runScenario: %v", err)	174	t.Fatalf("runScenario: %v", err)
155	}	175	}
@@ -162,6 +182,9 @@ func TestRunScenarioSuccess(t *testing.T) {
162	if !strings.Contains(msg, "reboot: ok") {	182	if !strings.Contains(msg, "reboot: ok") {
163	t.Errorf("message = %q, want reboot: ok", msg)	183	t.Errorf("message = %q, want reboot: ok", msg)
164	}	184	}
		185	if !strings.Contains(msg, "exposed port: ok") {
		186	t.Errorf("message = %q, want exposed port: ok", msg)
		187	}
165	if !sawTruncate {	188	if !sawTruncate {
166	t.Error("serial log was never truncated before the reboot proof — proof could pass on first-boot bytes")	189	t.Error("serial log was never truncated before the reboot proof — proof could pass on first-boot bytes")
167	}	190	}
@@ -175,7 +198,7 @@ func TestRunScenarioRebootDeathFails(t *testing.T) {
175	power := "running"	198	power := "running"
176	api := &testAPI{	199	api := &testAPI{
177	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {	200	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {
178	return []client.Host{{ID: "host-1"}}, nil	201	return []client.Host{{ID: "host-1", UplinkAddr: "192.168.0.190"}}, nil
179	},	202	},
180	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {	203	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {
181	return client.CreateVMResponse{ID: "vm-1"}, nil	204	return client.CreateVMResponse{ID: "vm-1"}, nil
@@ -207,7 +230,7 @@ func TestRunScenarioRebootDeathFails(t *testing.T) {
207	}	230	}
208		231
209	clock := &fakeClock{t: time.Unix(0, 0)}	232	clock := &fakeClock{t: time.Unix(0, 0)}
210	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey)	233	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey, okBanner)
211	if err == nil {	234	if err == nil {
212	t.Fatal("runScenario: want error for a VM that never came back, got nil")	235	t.Fatal("runScenario: want error for a VM that never came back, got nil")
213	}	236	}
@@ -216,12 +239,41 @@ func TestRunScenarioRebootDeathFails(t *testing.T) {
216	}	239	}
217	}	240	}
218		241
		242	// --- runScenario: published port ----------------------------------------
		243
		244	// TestRunScenarioExposureFailureFails pins the published-port leg as a hard
		245	// gate: a host whose listener answers with something that is not an sshd fails
		246	// the scenario, and the VM is never reaped over the failure.
		247	func TestRunScenarioExposureFailureFails(t *testing.T) {
		248	var calls []string
		249	api := happyPathAPI(t, &calls)
		250	api.deleteVMFunc = func(ctx context.Context, id string) error {
		251	t.Fatal("DeleteVM must not be called when the published port failed")
		252	return nil
		253	}
		254	revoked := ""
		255	api.deleteExposureFunc = func(ctx context.Context, id string) error { revoked = id; return nil }
		256	badBanner := func(ctx context.Context, addr string) (string, error) { return "HTTP/1.1 400\r\n", nil }
		257
		258	clock := &fakeClock{t: time.Unix(0, 0)}
		259	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, happyPathRunSSH(), nil, clock.now, clock.sleep, noopReadPubKey, badBanner)
		260	if err == nil {
		261	t.Fatal("runScenario: want error, got nil")
		262	}
		263	if !strings.Contains(err.Error(), "SSH-2.0") {
		264	t.Errorf("error = %q, want it to name the banner it wanted", err.Error())
		265	}
		266	if revoked != "x-fake" {
		267	t.Errorf("revoked exposure = %q, want the failed leg to revoke its grant", revoked)
		268	}
		269	}
		270
219	// --- runScenario: serial panic -----------------------------------------	271	// --- runScenario: serial panic -----------------------------------------
220		272
221	func TestRunScenarioSerialPanicFails(t *testing.T) {	273	func TestRunScenarioSerialPanicFails(t *testing.T) {
222	api := &testAPI{	274	api := &testAPI{
223	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {	275	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {
224	return []client.Host{{ID: "host-1"}}, nil	276	return []client.Host{{ID: "host-1", UplinkAddr: "192.168.0.190"}}, nil
225	},	277	},
226	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {	278	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {
227	return client.CreateVMResponse{ID: "vm-1"}, nil	279	return client.CreateVMResponse{ID: "vm-1"}, nil
@@ -239,7 +291,7 @@ func TestRunScenarioSerialPanicFails(t *testing.T) {
239	}	291	}
240		292
241	clock := &fakeClock{t: time.Unix(0, 0)}	293	clock := &fakeClock{t: time.Unix(0, 0)}
242	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey)	294	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey, okBanner)
243	if err == nil {	295	if err == nil {
244	t.Fatal("runScenario: want error, got nil")	296	t.Fatal("runScenario: want error, got nil")
245	}	297	}
@@ -253,7 +305,7 @@ func TestRunScenarioSerialPanicFails(t *testing.T) {
253	func TestRunScenarioNeverReadyTimesOut(t *testing.T) {	305	func TestRunScenarioNeverReadyTimesOut(t *testing.T) {
254	api := &testAPI{	306	api := &testAPI{
255	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {	307	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {
256	return []client.Host{{ID: "host-1"}}, nil	308	return []client.Host{{ID: "host-1", UplinkAddr: "192.168.0.190"}}, nil
257	},	309	},
258	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {	310	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {
259	return client.CreateVMResponse{ID: "vm-1"}, nil	311	return client.CreateVMResponse{ID: "vm-1"}, nil
@@ -272,7 +324,7 @@ func TestRunScenarioNeverReadyTimesOut(t *testing.T) {
272	}	324	}
273		325
274	clock := &fakeClock{t: time.Unix(0, 0)}	326	clock := &fakeClock{t: time.Unix(0, 0)}
275	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey)	327	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey, okBanner)
276	if err == nil {	328	if err == nil {
277	t.Fatal("runScenario: want error, got nil")	329	t.Fatal("runScenario: want error, got nil")
278	}	330	}
@@ -290,7 +342,7 @@ func TestRunScenarioNeverReapedTimesOut(t *testing.T) {
290	power := "running"	342	power := "running"
291	api := &testAPI{	343	api := &testAPI{
292	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {	344	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {
293	return []client.Host{{ID: "host-1"}}, nil	345	return []client.Host{{ID: "host-1", UplinkAddr: "192.168.0.190"}}, nil
294	},	346	},
295	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {	347	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {
296	return client.CreateVMResponse{ID: "vm-1"}, nil	348	return client.CreateVMResponse{ID: "vm-1"}, nil
@@ -310,7 +362,7 @@ func TestRunScenarioNeverReapedTimesOut(t *testing.T) {
310	}	362	}
311		363
312	clock := &fakeClock{t: time.Unix(0, 0)}	364	clock := &fakeClock{t: time.Unix(0, 0)}
313	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey)	365	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, runSSH, nil, clock.now, clock.sleep, noopReadPubKey, okBanner)
314	if err == nil {	366	if err == nil {
315	t.Fatal("runScenario: want error, got nil")	367	t.Fatal("runScenario: want error, got nil")
316	}	368	}
@@ -338,7 +390,7 @@ func TestRunScenarioNoHosts(t *testing.T) {
338	},	390	},
339	}	391	}
340	clock := &fakeClock{t: time.Unix(0, 0)}	392	clock := &fakeClock{t: time.Unix(0, 0)}
341	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, nil, nil, clock.now, clock.sleep, noopReadPubKey)	393	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, nil, nil, clock.now, clock.sleep, noopReadPubKey, okBanner)
342	if err == nil {	394	if err == nil {
343	t.Fatal("runScenario: want error, got nil")	395	t.Fatal("runScenario: want error, got nil")
344	}	396	}
@@ -355,7 +407,7 @@ func happyPathAPI(t testing.T, calls []string) *testAPI {
355	deleted := false	407	deleted := false
356	return &testAPI{	408	return &testAPI{
357	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {	409	listHostsFunc: func(ctx context.Context) ([]client.Host, error) {
358	return []client.Host{{ID: "host-1"}}, nil	410	return []client.Host{{ID: "host-1", UplinkAddr: "192.168.0.190"}}, nil
359	},	411	},
360	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {	412	createVMFunc: func(ctx context.Context, req client.CreateVMRequest) (client.CreateVMResponse, error) {
361	calls = append(calls, "createVM")	413	calls = append(calls, "createVM")
@@ -412,7 +464,7 @@ func TestRunScenarioGateRegistersBeforeCreateAndExecsAfterBoot(t *testing.T) {
412	}	464	}
413		465
414	clock := &fakeClock{t: time.Unix(0, 0)}	466	clock := &fakeClock{t: time.Unix(0, 0)}
415	msg, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, happyPathRunSSH(), gate, clock.now, clock.sleep, noopReadPubKey)	467	msg, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, happyPathRunSSH(), gate, clock.now, clock.sleep, noopReadPubKey, okBanner)
416	if err != nil {	468	if err != nil {
417	t.Fatalf("runScenario: %v", err)	469	t.Fatalf("runScenario: %v", err)
418	}	470	}
@@ -462,7 +514,7 @@ func TestRunScenarioGateRegisterErrorAbortsBeforeCreate(t *testing.T) {
462	}	514	}
463		515
464	clock := &fakeClock{t: time.Unix(0, 0)}	516	clock := &fakeClock{t: time.Unix(0, 0)}
465	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, nil, gate, clock.now, clock.sleep, noopReadPubKey)	517	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, nil, gate, clock.now, clock.sleep, noopReadPubKey, okBanner)
466	if err == nil {	518	if err == nil {
467	t.Fatal("runScenario: want error, got nil")	519	t.Fatal("runScenario: want error, got nil")
468	}	520	}
@@ -486,7 +538,7 @@ func TestRunScenarioGateExecErrorFails(t *testing.T) {
486	}	538	}
487		539
488	clock := &fakeClock{t: time.Unix(0, 0)}	540	clock := &fakeClock{t: time.Unix(0, 0)}
489	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, happyPathRunSSH(), gate, clock.now, clock.sleep, noopReadPubKey)	541	_, err := runScenario(context.Background(), baseCfg(), "smoke-test", api, happyPathRunSSH(), gate, clock.now, clock.sleep, noopReadPubKey, okBanner)
490	if err == nil {	542	if err == nil {
491	t.Fatal("runScenario: want error, got nil")	543	t.Fatal("runScenario: want error, got nil")
492	}	544	}

proto/eitri/v1/sync.proto

Old		New
@@ -99,6 +99,14 @@ message ActualStateReport {
99	// knowable at connect time at all. Capacity is in both messages for the same	99	// knowable at connect time at all. Capacity is in both messages for the same
100	// reason: an opening value, then the ongoing truth.	100	// reason: an opening value, then the ongoing truth.
101	string guest_cidr = 8;	101	string guest_cidr = 8;
		102	repeated ExposureActual exposures = 9;
		103	// The address this host presents on the network it reaches the control
		104	// plane over — the address an operator dials to reach a published guest
		105	// port. Empty means "not yet known", never "no address": a host that cannot
		106	// answer leaves the fleet's record alone rather than erasing it. In the
		107	// report rather than Hello for the same reason guest_cidr is: it can change
		108	// while an agent stays connected.
		109	string host_uplink_addr = 10;
102	}	110	}
103		111
104	message VMDesired {	112	message VMDesired {
@@ -125,6 +133,7 @@ message DesiredStateSnapshot {
125	uint64 epoch = 1; // agents refuse epoch < highest seen	133	uint64 epoch = 1; // agents refuse epoch < highest seen
126	repeated VMDesired vms = 2; // FULL set for this host, including tombstoned	134	repeated VMDesired vms = 2; // FULL set for this host, including tombstoned
127	AgentUpgrade agent_upgrade = 3; // optional operator-initiated agent self-upgrade	135	AgentUpgrade agent_upgrade = 3; // optional operator-initiated agent self-upgrade
		136	repeated ExposureDesired exposures = 4; // FULL set for this host
128	}	137	}
129		138
130	// AgentUpgrade asks the agent to replace its own binary: download url, verify	139	// AgentUpgrade asks the agent to replace its own binary: download url, verify
@@ -167,3 +176,27 @@ message TCPOpened {
167	bool ok = 1;	176	bool ok = 1;
168	string error = 2;	177	string error = 2;
169	}	178	}
		179
		180	// ExposureDesired is one published guest port a host should be serving: bind
		181	// host_port on the host, pipe every accepted connection to guest_port inside
		182	// the guest. It rides the snapshot at TOP LEVEL rather than nested in
		183	// VMDesired, because exposures are their own objects converging on their own
		184	// cadence — an exposure can be created while its VM is still imaging, and it
		185	// binds immediately.
		186	message ExposureDesired {
		187	string id = 1;
		188	string vm_id = 2;
		189	uint32 guest_port = 3;
		190	uint32 host_port = 4;
		191	string protocol = 5; // "tcp"
		192	}
		193
		194	// ExposureActual is one exposure's state as its host observes it: "active"
		195	// once the host listener is bound, "failed" with the OS error otherwise.
		196	// "active" means the HOST half of the pipe exists — whether anything answers
		197	// inside the guest is the guest's half, and this does not pretend otherwise.
		198	message ExposureActual {
		199	string id = 1;
		200	string state = 2; // "active"\|"failed"
		201	string reason = 3; // the OS error, when failed
		202	}

scripts/coverage.sh

Old		New
@@ -28,6 +28,7 @@ declare -A FLOOR=(
28	[internal/agent/pidfile]=95	28	[internal/agent/pidfile]=95
29	[internal/agent/vfkit]=85	29	[internal/agent/vfkit]=85
30	[internal/agent/syncclient]=74	30	[internal/agent/syncclient]=74
		31	[internal/agent/exposeproxy]=80
31	[internal/server/api]=76	32	[internal/server/api]=76
32	[internal/server/boot]=17	33	[internal/server/boot]=17
33	[internal/server/api/client]=89	34	[internal/server/api/client]=89

web/src/lib/api-types.ts

@@ @@ -194,6 +194,51 @@ export interface paths { @@
+        patch?: never;
+        trace?: never;
+    };
+    "/api/v1/exposures/{id}": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        get?: never;
+        put?: never;
+        post?: never;
+        /** Revoke an exposure; its host closes the listener on the next converge. */
+        delete: {
+            parameters: {
+                query?: never;
+                header?: never;
+                path: {
+                    id: string;
+                };
+                cookie?: never;
+            };
+            requestBody?: never;
+            responses: {
+                /** @description success */
+: {
+                    headers: {
+                        [name: string]: unknown;
+                    };
+                    content?: never;
+                };
+                /** @description error (plain text) */
+                default: {
+                    headers: {
+                        [name: string]: unknown;
+                    };
+                    content: {
+                        "text/plain": string;
+                    };
+                };
+            };
+        };
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
+    "/api/v1/hosts": {
+        parameters: {
+            query?: never;
@@ @@ -1174,6 +1219,88 @@ export interface paths { @@
+        patch?: never;
+        trace?: never;
+    };
+    "/api/v1/vms/{id}/exposures": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /** List the VM's published ports, with the host address to dial and each listener's reported state. */
+        get: {
+            parameters: {
+                query?: never;
+                header?: never;
+                path: {
+                    id: string;
+                };
+                cookie?: never;
+            };
+            requestBody?: never;
+            responses: {
+                /** @description success */
+: {
+                    headers: {
+                        [name: string]: unknown;
+                    };
+                    content: {
+                        "application/json": components["schemas"]["Exposure"][];
+                    };
+                };
+                /** @description error (plain text) */
+                default: {
+                    headers: {
+                        [name: string]: unknown;
+                    };
+                    content: {
+                        "text/plain": string;
+                    };
+                };
+            };
+        };
+        put?: never;
+        /** Publish a guest TCP port on the VM's host. Omit host_port to allocate one from the reserved range 30000-32767; a named port must be >= 1024 and is honored or refused. */
+        post: {
+            parameters: {
+                query?: never;
+                header?: never;
+                path: {
+                    id: string;
+                };
+                cookie?: never;
+            };
+            requestBody: {
+                content: {
+                    "application/json": components["schemas"]["CreateExposureRequest"];
+                };
+            };
+            responses: {
+                /** @description success */
+: {
+                    headers: {
+                        [name: string]: unknown;
+                    };
+                    content: {
+                        "application/json": components["schemas"]["Exposure"];
+                    };
+                };
+                /** @description error (plain text) */
+                default: {
+                    headers: {
+                        [name: string]: unknown;
+                    };
+                    content: {
+                        "text/plain": string;
+                    };
+                };
+            };
+        };
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
+    "/api/v1/vms/{id}/restore": {
+        parameters: {
+            query?: never;
@@ @@ -1252,6 +1379,10 @@ export interface components { @@
+            name: string;
+            token: string;
+        };
+        CreateExposureRequest: {
+            guest_port?: number;
+            host_port?: number;
+        };
+        CreateVMRequest: {
+            cloud_init?: string;
+            disk_gb?: number;
@@ @@ -1287,6 +1418,20 @@ export interface components { @@
+            join: string;
+            token: string;
+        };
+        Exposure: {
+            /** Format: date-time */
+            created_at: string;
+            guest_port: number;
+            host_addr: string;
+            host_id: string;
+            host_port: number;
+            id: string;
+            protocol: string;
+            reason: string;
+            scope: string;
+            state: string;
+            vm_id: string;
+        };
+        Host: {
+            agent_update_available: boolean;
+            agent_version: string;
@@ @@ -1313,6 +1458,7 @@ export interface components { @@
+            sessions: number;
+            stale: boolean;
+            status: string;
+            uplink_addr: string;
+            virt: string;
+        };
+        InjectedKey: {

web/src/lib/fleet.svelte.ts

Old		New
@@ -20,6 +20,11 @@ export type VMEvent = components['schemas']['AuditEvent'];
20	* the canonical pubkey line, an optional label, and the SHA256 fingerprint. */	20	* the canonical pubkey line, an optional label, and the SHA256 fingerprint. */
21	export type UserCA = components['schemas']['UserCA'];	21	export type UserCA = components['schemas']['UserCA'];
22		22
		23	/** Exposure is one published guest port: the fleet binds host_port on the VM's
		24	* host and pipes it to guest_port inside the guest. host_addr is the address
		25	* to dial; state is what the host says its listener is doing. */
		26	export type Exposure = components['schemas']['Exposure'];
		27
23	export type CreateVMRequest = components['schemas']['CreateVMRequest'];	28	export type CreateVMRequest = components['schemas']['CreateVMRequest'];
24		29
25	/** Me is the signed-in identity: the caller's tenant handle and bound email. */	30	/** Me is the signed-in identity: the caller's tenant handle and bound email. */
@@ -255,6 +260,31 @@ export async function vmEvents(id: string): Promise<VMEvent[]> {
255	return (await req('GET', `/api/v1/vms/${id}/events`)).json();	260	return (await req('GET', `/api/v1/vms/${id}/events`)).json();
256	}	261	}
257		262
		263	/** listExposures fetches the VM's published ports, lowest host port first. */
		264	export async function listExposures(id: string): Promise<Exposure[]> {
		265	return (await req('GET', `/api/v1/vms/${id}/exposures`)).json();
		266	}
		267
		268	/** createExposure publishes a guest port. hostPort 0 asks the control plane to
		269	* allocate one from the reserved range. */
		270	export async function createExposure(
		271	id: string,
		272	guestPort: number,
		273	hostPort: number
		274	): Promise<Exposure> {
		275	return (
		276	await req('POST', `/api/v1/vms/${id}/exposures`, {
		277	guest_port: guestPort,
		278	host_port: hostPort
		279	})
		280	).json();
		281	}
		282
		283	/** deleteExposure revokes an exposure; its host closes the listener. */
		284	export async function deleteExposure(id: string) {
		285	await req('DELETE', `/api/v1/exposures/${id}`);
		286	}
		287
258	/** listUserCAs fetches the tenant's registered SSH user CAs. */	288	/** listUserCAs fetches the tenant's registered SSH user CAs. */
259	export async function listUserCAs(): Promise<UserCA[]> {	289	export async function listUserCAs(): Promise<UserCA[]> {
260	return (await req('GET', `/api/v1/tenants/${tenant()}/user-cas`)).json();	290	return (await req('GET', `/api/v1/tenants/${tenant()}/user-cas`)).json();
@@ -371,6 +401,14 @@ export function teardownApprox(vm: VM, nowSec: number): string {
371	return `~${Math.ceil(left / 60)}m`;	401	return `~${Math.ceil(left / 60)}m`;
372	}	402	}
373		403
		404	/** formatHostPort renders a dial target from an address and a port. The API
		405	* ships the two apart precisely so the renderer owns this: an IPv6 address
		406	* (the only form carrying a colon) has to be bracketed, or the result reads
		407	* as one more hextet. */
		408	export function formatHostPort(addr: string, port: number): string {
		409	return addr.includes(':') ? `[${addr}]:${port}` : `${addr}:${port}`;
		410	}
		411
374	/** eventLabel maps a lifecycle VMEvent to a human label, folding in the one	412	/** eventLabel maps a lifecycle VMEvent to a human label, folding in the one
375	* detail that matters per action. detail arrives already decoded (raw JSON on	413	* detail that matters per action. detail arrives already decoded (raw JSON on
376	* the wire); guard the shape defensively rather than parsing. */	414	* the wire); guard the shape defensively rather than parsing. */
@@ -388,6 +426,10 @@ export function eventLabel(ev: VMEvent): string {
388	return 'Restored';	426	return 'Restored';
389	case 'vm.reap':	427	case 'vm.reap':
390	return `Destroyed (${detail.reason ?? '?'})`;	428	return `Destroyed (${detail.reason ?? '?'})`;
		429	case 'exposure.create':
		430	return `Exposed guest :${detail.guest_port ?? '?'} on host :${detail.host_port ?? '?'}`;
		431	case 'exposure.delete':
		432	return `Exposure removed (guest :${detail.guest_port ?? '?'})`;
391	default:	433	default:
392	return ev.action;	434	return ev.action;
393	}	435	}

web/src/routes/vms/[id]/+page.svelte

@@ @@ -14,8 +14,13 @@ @@
+		vmIsRunning,
+		vmEvents,
+		eventLabel,
+		listExposures,
+		createExposure,
+		deleteExposure,
+		formatHostPort,
+		clock,
+		type VMEvent,
+		type Exposure
+	} from '$lib/fleet.svelte';
+	const id = $derived(page.params.id);
@@ @@ -62,6 +67,71 @@ @@
+		loadEvents(id);
+	});
+	let exposures = $state<Exposure[]>([]);
+	let exposuresFailed = $state(false);
+	let guestPortInput = $state('');
+	let hostPortInput = $state('');
+	async function loadExposures(forId: string | undefined) {
+		if (!forId) return;
+		try {
+			const list = await listExposures(forId);
+			if (forId !== id) return; // navigated away mid-flight
+			exposures = list;
+			exposuresFailed = false;
+		} catch {
+			if (forId !== id) return;
+			exposuresFailed = true;
+		}
+	}
+	// Reload alongside the timeline: an exposure's state is what its host last
+	// reported, so it moves as the VM does.
+	$effect(() => {
+		void lifecycle;
+		exposures = [];
+		exposuresFailed = false;
+		loadExposures(id);
+	});
+	// exposureTarget is the address:port to dial, or '' while no address is
+	// known yet — an empty host_addr means the host hasn't reported one, never
+	// that the exposure has none, so the row says so rather than printing
+	// something undialable. The SSE-fed host is the second source, so a row
+	// heals within a tick of its host reporting instead of waiting for a refetch.
+	function exposureTarget(e: Exposure): string {
+		const addr = e.host_addr || host?.uplink_addr || '';
+		return addr ? formatHostPort(addr, e.host_port) : '';
+	}
+	// exposureState is the state a row shows. A VM being torn down dominates:
+	// its host stops reporting the listeners, so every row would otherwise read
+	// "pending" while the truer answer is that the VM is going away.
+	function exposureState(e: Exposure): string {
+		return tearingDown ? 'deleting' : e.state;
+	}
+	async function addExposure(e: SubmitEvent) {
+		e.preventDefault();
+		if (!vm) return;
+		const guest = Number(guestPortInput);
+		const host = hostPortInput === '' ? 0 : Number(hostPortInput);
+		const vmId = vm.id;
+		if (await action(() => createExposure(vmId, guest, host))) {
+			guestPortInput = '';
+			hostPortInput = '';
+			await loadExposures(vmId);
+		}
+	}
+	async function removeExposure(exposureId: string) {
+		if (!vm) return;
+		const vmId = vm.id;
+		if (await action(() => deleteExposure(exposureId))) {
+			await loadExposures(vmId);
+		}
+	}
+	async function power(p: 'running' | 'stopped') {
+		if (!vm) return;
+		const vmId = vm.id;
@@ @@ -167,6 +237,66 @@ @@
+		</div>
+	{/if}
+	<div class="exposures">
+		<h3>Exposures</h3>
+		{#if exposuresFailed}
+			<p class="hint">couldn't load exposures</p>
+		{:else if exposures.length === 0}
+			<p class="hint">
+				No published ports. The SSH gate reaches this guest; anything else it serves needs a port
+				published here.
+			</p>
+		{:else}
+			<table class="kv">
+				<tbody>
+					{#each exposures as e (e.id)}
+						<tr>
+							<th>guest :{e.guest_port}</th>
+							<td>
+								{#if exposureTarget(e)}
+									<code>{exposureTarget(e)}</code>
+								{:else}
+									<span class="hint">host address not yet known · :{e.host_port}</span>
+								{/if}
+								<span class="state {exposureState(e)}">● {exposureState(e)}</span>
+								{#if e.reason && !tearingDown}<span class="err">{e.reason}</span>{/if}
+							</td>
+							<td>
+								{#if !tearingDown}
+									<button class="ghost" onclick={() => removeExposure(e.id)}>Remove</button>
+								{/if}
+							</td>
+						</tr>
+					{/each}
+				</tbody>
+			</table>
+		{/if}
+		{#if !tearingDown}
+			<form class="expose-form" onsubmit={addExposure}>
+				<input
+					type="number"
+					min="1"
+					max="65535"
+					required
+					bind:value={guestPortInput}
+					placeholder="guest port"
+				/>
+				<input
+					type="number"
+					min="1024"
+					max="65535"
+					bind:value={hostPortInput}
+					placeholder="host port (optional)"
+				/>
+				<button type="submit">Expose</button>
+			</form>
+		{/if}
+		<p class="hint">
+			Anything that can reach the host can reach a published port—there is no authentication in
+			front of one.
+		</p>
+	</div>
+	{#if !tearingDown}
+		<Console vmId={vm.id} />
+	{/if}
@@ @@ -219,6 +349,41 @@ @@
+	:global(button.restore:hover) {
+		background: #37b24d;
+	.exposures {
+		margin-top: 1.2rem;
+	}
+	.exposures h3 {
+		margin: 0 0 0.2rem;
+		font-size: 13px;
+	}
+	.exposures .kv th {
+		color: #8b919c;
+		width: 140px;
+		font-weight: 400;
+		white-space: nowrap;
+	}
+	.state {
+		margin-left: 0.5rem;
+		font-size: 12px;
+	}
+	.state.active {
+		color: #51cf66;
+	}
+	.state.failed {
+		color: #ffb4b4;
+	}
+	.state.pending,
+	.state.deleting {
+		color: #8b919c;
+	}
+	.expose-form {
+		display: flex;
+		gap: 0.4rem;
+		margin-top: 0.5rem;
+	}
+	.expose-form input {
+		width: 10rem;
+	}
+	.timeline {
+		margin-top: 1.2rem;