Enterprise Phase 3: security hardening, RBAC, policies, key lifecycle, observability

- Secure channel authentication: Ed25519 identity verification inside ECDH tunnel
- Registry hardening: per-operation rate limits, connection limits, 64KB message cap, snapshot checksums
- Peer resilience: endpoint caching with TTL, automatic reconnection on tunnel failure
- Observability: /healthz endpoint on beacon, pilotctl health command, daemon uptime/stats
- RBAC: owner/admin/member roles per network, role-gated mutations, legacy snapshot backfill
- Protocol versioning: version field in handshake, compatibility negotiation
- Network policies: max_members, allowed_ports, description with merge-on-update semantics
- Key lifecycle: CreatedAt/RotatedAt/RotateCount/ExpiresAt metadata, set_key_expiry command
- Invite consent fix: PollInvites non-destructive, RespondInvite verifies + consumes invite
- Policy merge fix: partial updates preserve unset fields
- Key age guard: zero-time check prevents bogus age for legacy nodes
- 8 new test files, 43 new tests passing

Author: teovl

cmd/pilotctl/main.go

@@ -395,6 +395,7 @@ Mailbox:
 
 Diagnostic commands:
   pilotctl info
+  pilotctl health
   pilotctl peers [--search <query>]
   pilotctl ping <address|hostname> [--count <n>] [--timeout <dur>]
   pilotctl traceroute <address> [--timeout <dur>]
@@ -623,6 +624,8 @@ func main() {
 	// Diagnostics
 	case "info":
 		cmdInfo()
+	case "health":
+		cmdHealth()
 	case "peers":
 		cmdPeers(cmdArgs)
 	case "ping":
@@ -3536,6 +3539,37 @@ func cmdInfo() {
 	}
 }
 
+func cmdHealth() {
+	d := connectDriver()
+	defer d.Close()
+
+	health, err := d.Health()
+	if err != nil {
+		fatalCode("connection_failed", "health: %v", err)
+	}
+
+	if jsonOutput {
+		output(health)
+		return
+	}
+
+	uptime := int64(0)
+	if v, ok := health["uptime_seconds"].(float64); ok {
+		uptime = int64(v)
+	}
+	hours := uptime / 3600
+	mins := (uptime % 3600) / 60
+	secs := uptime % 60
+
+	fmt.Printf("Daemon Health\n")
+	fmt.Printf("  Status:      %s\n", health["status"])
+	fmt.Printf("  Uptime:      %02d:%02d:%02d\n", hours, mins, secs)
+	fmt.Printf("  Connections: %d\n", int(health["connections"].(float64)))
+	fmt.Printf("  Peers:       %d\n", int(health["peers"].(float64)))
+	fmt.Printf("  Bytes Sent:  %s\n", formatBytes(uint64(health["bytes_sent"].(float64))))
+	fmt.Printf("  Bytes Recv:  %s\n", formatBytes(uint64(health["bytes_recv"].(float64))))
+}
+
 func cmdPeers(args []string) {
 	flags, _ := parseFlags(args)
 	search := flagString(flags, "search", "")

pkg/daemon/daemon.go

@@ -90,6 +90,16 @@ const (
 	ZeroWinProbeMax     = 30 * time.Second       // max zero-window probe backoff
 )
 
+// EndpointCacheTTL is how long a cached endpoint is considered fresh.
+// After this, the entry is stale but still usable as a fallback.
+const EndpointCacheTTL = 5 * time.Minute
+
+// endpointEntry caches a resolved endpoint for a peer node.
+type endpointEntry struct {
+	addr      string    // "host:port"
+	cachedAt  time.Time // when the entry was stored
+}
+
 type Daemon struct {
 	config     Config
 	addrMu     sync.RWMutex // protects nodeID and addr (H6 fix)
@@ -107,6 +117,10 @@ type Daemon struct {
 	stopCh     chan struct{} // closed on Stop() to signal goroutines
 	lanAddrs   []string     // LAN addresses for same-network peer detection
 
+	// Endpoint cache: nodeID -> last-known endpoint (peer resilience)
+	epCacheMu sync.RWMutex
+	epCache   map[uint32]*endpointEntry
+
 	// SYN rate limiter (token bucket)
 	synMu       sync.Mutex
 	synTokens   int
@@ -177,6 +191,7 @@ func New(cfg Config) *Daemon {
 		synTokens:   cfg.synRateLimit(),
 		synLastFill: time.Now(),
 		perSrcSYN:   make(map[uint32]*srcSYNBucket),
+		epCache:     make(map[uint32]*endpointEntry),
 	}
 	d.ipc = NewIPCServer(cfg.SocketPath, d)
 	d.handshakes = NewHandshakeManager(d)
@@ -744,12 +759,15 @@ type DaemonInfo struct {
 	Identity           bool   // true if identity is persisted
 	PublicKey          string // base64 Ed25519 public key (empty if no identity)
 	Email              string // email address for account identification and key recovery
-	BytesSent          uint64
-	BytesRecv          uint64
-	PktsSent           uint64
-	PktsRecv           uint64
-	PeerList           []PeerInfo
-	ConnList           []ConnectionInfo
+	BytesSent             uint64
+	BytesRecv             uint64
+	PktsSent              uint64
+	PktsRecv              uint64
+	EncryptOK             uint64
+	EncryptFail           uint64
+	HandshakePendingCount int
+	PeerList              []PeerInfo
+	ConnList              []ConnectionInfo
 }
 
 // Info returns current daemon status.
@@ -805,12 +823,15 @@ func (d *Daemon) Info() *DaemonInfo {
 		Identity:           hasIdentity,
 		PublicKey:          pubKeyStr,
 		Email:              d.config.Email,
-		BytesSent:          atomic.LoadUint64(&d.tunnels.BytesSent),
-		BytesRecv:          atomic.LoadUint64(&d.tunnels.BytesRecv),
-		PktsSent:           atomic.LoadUint64(&d.tunnels.PktsSent),
-		PktsRecv:           atomic.LoadUint64(&d.tunnels.PktsRecv),
-		PeerList:           peerList,
-		ConnList:           d.ports.ConnectionList(),
+		BytesSent:             atomic.LoadUint64(&d.tunnels.BytesSent),
+		BytesRecv:             atomic.LoadUint64(&d.tunnels.BytesRecv),
+		PktsSent:              atomic.LoadUint64(&d.tunnels.PktsSent),
+		PktsRecv:              atomic.LoadUint64(&d.tunnels.PktsRecv),
+		EncryptOK:             atomic.LoadUint64(&d.tunnels.EncryptOK),
+		EncryptFail:           atomic.LoadUint64(&d.tunnels.EncryptFail),
+		HandshakePendingCount: d.handshakes.PendingCount(),
+		PeerList:              peerList,
+		ConnList:              d.ports.ConnectionList(),
 	}
 }
 
@@ -1822,8 +1843,45 @@ func (d *Daemon) broadcastDatagram(netID uint16, srcPort, dstPort uint16, data [
 	return nil
 }
 
+// cacheEndpoint stores a resolved endpoint in the cache.
+func (d *Daemon) cacheEndpoint(nodeID uint32, addr string) {
+	d.epCacheMu.Lock()
+	d.epCache[nodeID] = &endpointEntry{addr: addr, cachedAt: time.Now()}
+	d.epCacheMu.Unlock()
+}
+
+// cachedEndpoint returns a previously cached endpoint, or empty string if none.
+// The second return value is true if the entry exists (even if stale).
+func (d *Daemon) cachedEndpoint(nodeID uint32) (string, bool) {
+	d.epCacheMu.RLock()
+	e, ok := d.epCache[nodeID]
+	d.epCacheMu.RUnlock()
+	if !ok {
+		return "", false
+	}
+	return e.addr, true
+}
+
+// isEndpointStale returns true if the cached entry is older than EndpointCacheTTL.
+// Stale entries are still usable as fallback, but a fresh resolve is preferred.
+func (d *Daemon) isEndpointStale(nodeID uint32) bool {
+	d.epCacheMu.RLock()
+	e, ok := d.epCache[nodeID]
+	d.epCacheMu.RUnlock()
+	if !ok {
+		return true
+	}
+	return time.Since(e.cachedAt) > EndpointCacheTTL
+}
+
+// CachedEndpoint returns a previously cached endpoint for a peer (exported for testing).
+func (d *Daemon) CachedEndpoint(nodeID uint32) (string, bool) {
+	return d.cachedEndpoint(nodeID)
+}
+
 // ensureTunnel makes sure we have a route to the given node.
 // Requests beacon hole-punching for NAT traversal when beacon is configured.
+// Uses an endpoint cache as fallback when the registry is unreachable.
 func (d *Daemon) ensureTunnel(nodeID uint32) error {
 	if d.tunnels.HasPeer(nodeID) {
 		return nil
@@ -1832,6 +1890,18 @@ func (d *Daemon) ensureTunnel(nodeID uint32) error {
 	// Resolve the node's real address from registry (requires our node ID)
 	resp, err := d.regConn.Resolve(nodeID, d.NodeID())
 	if err != nil {
+		// Registry unreachable — fall back to cached endpoint
+		if cached, ok := d.cachedEndpoint(nodeID); ok {
+			stale := d.isEndpointStale(nodeID)
+			slog.Warn("registry resolve failed, using cached endpoint",
+				"node_id", nodeID, "cached_addr", cached, "stale", stale, "error", err)
+			udpAddr, udpErr := net.ResolveUDPAddr("udp", cached)
+			if udpErr != nil {
+				return fmt.Errorf("resolve cached %s: %w", cached, udpErr)
+			}
+			d.tunnels.AddPeer(nodeID, udpAddr)
+			return nil
+		}
 		return fmt.Errorf("resolve node %d: %w", nodeID, err)
 	}
 
@@ -1849,6 +1919,9 @@ func (d *Daemon) ensureTunnel(nodeID uint32) error {
 		}
 	}
 
+	// Cache the resolved endpoint for fallback
+	d.cacheEndpoint(nodeID, targetAddr)
+
 	udpAddr, err := net.ResolveUDPAddr("udp", targetAddr)
 	if err != nil {
 		return fmt.Errorf("resolve %s: %w", targetAddr, err)

pkg/daemon/handshake.go

@@ -812,6 +812,13 @@ func (hm *HandshakeManager) PendingRequests() []PendingHandshake {
 	return list
 }
 
+// PendingCount returns the number of pending handshake requests.
+func (hm *HandshakeManager) PendingCount() int {
+	hm.mu.RLock()
+	defer hm.mu.RUnlock()
+	return len(hm.pending)
+}
+
 // sendAcceptLocked sends an accept message (caller must hold hm.mu).
 func (hm *HandshakeManager) sendAcceptLocked(peerNodeID uint32) {
 	hm.goRPC(func() {

pkg/daemon/ipc.go

@@ -48,6 +48,8 @@ const (
 	CmdSetTaskExecOK     byte = 0x1E
 	CmdNetwork           byte = 0x1F
 	CmdNetworkOK         byte = 0x20
+	CmdHealth            byte = 0x21
+	CmdHealthOK          byte = 0x22
 )
 
 // Network sub-commands (second byte of CmdNetwork payload)
@@ -223,6 +225,8 @@ func (s *IPCServer) handleClient(conn *ipcConn) {
 			s.handleSetTaskExec(conn, payload)
 		case CmdNetwork:
 			s.handleNetwork(conn, payload)
+		case CmdHealth:
+			s.handleHealth(conn)
 		default:
 			s.sendError(conn, fmt.Sprintf("unknown command: 0x%02X", cmd))
 		}
@@ -418,10 +422,13 @@ func (s *IPCServer) handleInfo(conn *ipcConn) {
 		"email":               info.Email,
 		"bytes_sent":          info.BytesSent,
 		"bytes_recv":          info.BytesRecv,
-		"pkts_sent":           info.PktsSent,
-		"pkts_recv":           info.PktsRecv,
-		"peer_list":           peers,
-		"conn_list":           conns,
+		"pkts_sent":                    info.PktsSent,
+		"pkts_recv":                    info.PktsRecv,
+		"tunnel_encryption_success":    info.EncryptOK,
+		"tunnel_encryption_failure":    info.EncryptFail,
+		"handshake_pending_count":      info.HandshakePendingCount,
+		"peer_list":                    peers,
+		"conn_list":                    conns,
 	})
 	if err != nil {
 		s.sendError(conn, fmt.Sprintf("info marshal: %v", err))
@@ -435,6 +442,28 @@ func (s *IPCServer) handleInfo(conn *ipcConn) {
 	}
 }
 
+func (s *IPCServer) handleHealth(conn *ipcConn) {
+	info := s.daemon.Info()
+	data, err := json.Marshal(map[string]interface{}{
+		"status":         "ok",
+		"uptime_seconds": int64(info.Uptime.Seconds()),
+		"connections":    info.Connections,
+		"peers":          info.Peers,
+		"bytes_sent":     info.BytesSent,
+		"bytes_recv":     info.BytesRecv,
+	})
+	if err != nil {
+		s.sendError(conn, fmt.Sprintf("health marshal: %v", err))
+		return
+	}
+	resp := make([]byte, 1+len(data))
+	resp[0] = CmdHealthOK
+	copy(resp[1:], data)
+	if err := conn.ipcWrite(resp); err != nil {
+		slog.Debug("IPC health reply failed", "err", err)
+	}
+}
+
 func (s *IPCServer) handleResolveHostname(conn *ipcConn, payload []byte) {
 	hostname := string(payload)
 	if hostname == "" {

pkg/daemon/tunnel.go

@@ -120,10 +120,12 @@ type TunnelManager struct {
 	webhook *WebhookClient
 
 	// Metrics
-	BytesSent uint64
-	BytesRecv uint64
-	PktsSent  uint64
-	PktsRecv  uint64
+	BytesSent         uint64
+	BytesRecv         uint64
+	PktsSent          uint64
+	PktsRecv          uint64
+	EncryptOK         uint64
+	EncryptFail       uint64
 }
 
 type IncomingPacket struct {
@@ -633,6 +635,7 @@ func (tm *TunnelManager) handleEncrypted(data []byte, from *net.UDPAddr) {
 
 	plaintext, err := pc.aead.Open(nil, nonce, ciphertext, nil)
 	if err != nil {
+		atomic.AddUint64(&tm.EncryptFail, 1)
 		slog.Error("tunnel decrypt error", "peer_node_id", peerNodeID, "error", err)
 		// Undo the nonce record on decrypt failure — it was not a valid packet
 		pc.replayMu.Lock()
@@ -800,6 +803,7 @@ func (tm *TunnelManager) encryptFrame(pc *peerCrypto, plaintext []byte) []byte {
 	binary.BigEndian.PutUint64(nonce[pc.aead.NonceSize()-8:], counter)
 
 	ciphertext := pc.aead.Seal(nil, nonce, plaintext, nil)
+	atomic.AddUint64(&tm.EncryptOK, 1)
 
 	frame := make([]byte, 4+4+len(nonce)+len(ciphertext))
 	copy(frame[0:4], protocol.TunnelMagicSecure[:])

pkg/driver/driver.go

@@ -142,6 +142,11 @@ func (d *Driver) Info() (map[string]interface{}, error) {
 	return d.jsonRPC([]byte{cmdInfo}, cmdInfoOK, "info")
 }
 
+// Health returns a lightweight health check from the daemon.
+func (d *Driver) Health() (map[string]interface{}, error) {
+	return d.jsonRPC([]byte{cmdHealth}, cmdHealthOK, "health")
+}
+
 // Handshake sends a trust handshake request to a remote node.
 func (d *Driver) Handshake(nodeID uint32, justification string) (map[string]interface{}, error) {
 	msg := make([]byte, 1+1+4+len(justification))

pkg/driver/ipc.go

@@ -44,6 +44,8 @@ const (
 	cmdSetTaskExecOK     byte = 0x1E
 	cmdNetwork           byte = 0x1F
 	cmdNetworkOK         byte = 0x20
+	cmdHealth            byte = 0x21
+	cmdHealthOK          byte = 0x22
 )
 
 // Network sub-commands (must match daemon SubNetwork* constants)

pkg/protocol/packet.go

@@ -95,6 +95,12 @@ func Unmarshal(data []byte) (*Packet, error) {
 		return nil, ErrChecksumMismatch
 	}
 
+	// Validate protocol version.
+	wireVersion := (data[0] >> 4) & 0x0F
+	if wireVersion != Version {
+		return nil, fmt.Errorf("unsupported protocol version %d (expected %d)", wireVersion, Version)
+	}
+
 	p := &Packet{
 		Version:  (data[0] >> 4) & 0x0F,
 		Flags:    data[0] & 0x0F,