daemon: reliability + policy hardening (WSS half-open, watchdog restart-loop, port-policy fail-closed)

pkg/daemon/daemon.go

@@ -1825,10 +1825,42 @@ func (d *Daemon) NodeID() uint32 {
 // them for SYN-handler enforcement. Called at startup and after IPC joins.
 func (d *Daemon) loadNetworkPolicies() {
 	nets := d.nodeNetworks()
-	policies := make(map[uint16][]uint16, len(nets))
+
+	// Snapshot the current cache so a transient GetNetworkPolicy error
+	// carries the last-known policy FORWARD instead of dropping it.
+	// Dropping an entry makes isPortAllowed fall through to allow-all
+	// (empty list == no restriction), so a restricted network would go
+	// silently unrestricted on any registry hiccup — a fail-OPEN hole.
+	// A restriction is relaxed only by an explicit successful response
+	// that no longer lists it, never by a failed lookup.
+	d.netPolicyMu.RLock()
+	prior := make(map[uint16][]uint16, len(d.netPolicies))
+	for k, v := range d.netPolicies {
+		prior[k] = v
+	}
+	d.netPolicyMu.RUnlock()
+
+	results := make([]netPolicyResult, 0, len(nets))
 	for _, netID := range nets {
-		resp, err := d.regConn.GetNetworkPolicy(netID)
+		var resp map[string]interface{}
+		var err error
+		// Bounded retry so a transient blip on the very first load (when
+		// there is no prior to fall back to) still self-heals quickly.
+		for attempt := 0; attempt < 3; attempt++ {
+			if resp, err = d.regConn.GetNetworkPolicy(netID); err == nil {
+				break
+			}
+			time.Sleep(200 * time.Millisecond)
+		}
 		if err != nil {
+			_, hadPrior := prior[netID]
+			slog.Warn("network policy load failed — retaining last-known policy, not failing open",
+				"net_id", netID, "had_prior", hadPrior, "err", err)
+			d.publishEvent("network.policy_load_failed", map[string]any{
+				"net_id":    netID,
+				"had_prior": hadPrior,
+			})
+			results = append(results, netPolicyResult{netID: netID, err: err})
 			continue
 		}
 		portsRaw, _ := resp["allowed_ports"].([]interface{})
@@ -1838,15 +1870,45 @@ func (d *Daemon) loadNetworkPolicies() {
 				ports = append(ports, uint16(f))
 			}
 		}
-		if len(ports) > 0 {
-			policies[netID] = ports
-		}
+		results = append(results, netPolicyResult{netID: netID, ports: ports})
 	}
+
+	newPolicies := mergeNetworkPolicies(prior, results)
 	d.netPolicyMu.Lock()
-	d.netPolicies = policies
+	d.netPolicies = newPolicies
 	d.netPolicyMu.Unlock()
 }
 
+// netPolicyResult is one network's port-policy fetch outcome, fed to
+// mergeNetworkPolicies.
+type netPolicyResult struct {
+	netID uint16
+	ports []uint16
+	err   error
+}
+
+// mergeNetworkPolicies computes the new port-policy cache from the fetch
+// results and the prior cache. A FAILED fetch retains the prior policy
+// (fail-CLOSED — a transient registry error must not drop a restriction
+// and silently relax the network to allow-all); a SUCCESSFUL fetch is
+// stored verbatim, including an empty list, which explicitly means "no
+// restriction" and is the only thing that relaxes an existing policy.
+// Pure function so the fail-closed invariant is unit-testable without a
+// live registry.
+func mergeNetworkPolicies(prior map[uint16][]uint16, results []netPolicyResult) map[uint16][]uint16 {
+	policies := make(map[uint16][]uint16, len(results))
+	for _, r := range results {
+		if r.err != nil {
+			if p, ok := prior[r.netID]; ok {
+				policies[r.netID] = p
+			}
+			continue
+		}
+		policies[r.netID] = r.ports
+	}
+	return policies
+}
+
 // isPortAllowed checks whether dstPort is permitted by the network's AllowedPorts
 // policy. Returns true if no restriction is set (empty list = all ports allowed).
 func (d *Daemon) isPortAllowed(netID uint16, port uint16) bool {

pkg/daemon/rxwatchdog.go

@@ -6,6 +6,8 @@ import (
 	"log/slog"
 	"math/rand"
 	"os"
+	"strconv"
+	"strings"
 	"sync/atomic"
 	"time"
 )
@@ -82,6 +84,21 @@ const (
 	// Non-zero so launchd KeepAlive/SuccessfulExit=false respawns; a
 	// distinctive value so `pilotctl`/operators can attribute the restart.
 	rxWedgeExitCode = 86
+
+	// rxWedgeLoopWindow bounds how far back the restart-loop circuit
+	// breaker counts prior hard exits. Exits older than this don't count.
+	rxWedgeLoopWindow = 2 * time.Hour
+
+	// rxWedgeLoopMax is how many hard exits within rxWedgeLoopWindow are
+	// tolerated before the breaker OPENS and further exits are withheld.
+	// The hard exit assumes a restart CLEARS the wedge (stale NAT/relay
+	// mapping). When it doesn't — a NAT/firewall/topology change a
+	// restart can't fix — the daemon would otherwise boot-loop every
+	// ~30-40 min forever (wedge → 3 soft attempts → exit → respawn →
+	// wedge → …), each restart dropping every live tunnel for nothing.
+	// After this many restarts the breaker keeps the daemon UP and
+	// soft-recovering instead, and surfaces the condition for an operator.
+	rxWedgeLoopMax = 3
 )
 
 // rxWatchdogExit is swapped by tests to observe the hard escalation
@@ -107,6 +124,7 @@ const (
 	rxActionSoftRecover  rxWatchdogAction = "soft-recover"
 	rxActionExitWithheld rxWatchdogAction = "exit-withheld"
 	rxActionExit         rxWatchdogAction = "exit"
+	rxActionRestartLoop  rxWatchdogAction = "restart-loop-withheld"
 )
 
 func (d *Daemon) rxWatchdogLoop() {
@@ -261,6 +279,30 @@ func (d *Daemon) rxWatchdogTick(st *rxWatchdogState, now time.Time) (action rxWa
 			"silent_for", silence.Truncate(time.Second).String(),
 			"uptime", uptime.Truncate(time.Second).String())
 	default:
+		// Restart-loop circuit breaker. The hard exit is premised on a
+		// restart CLEARING the wedge; when it doesn't (a NAT/firewall/
+		// topology change no restart can fix), exiting just boot-loops
+		// the daemon and drops every live tunnel each cycle. If we have
+		// already respawned rxWedgeLoopMax times within rxWedgeLoopWindow
+		// (tracked in a sidecar file next to the identity so it survives
+		// the respawn), stop exiting: stay up, keep soft-recovering, and
+		// surface the condition for an operator.
+		exitLog := rxWedgeExitLogPath(d.config.IdentityPath)
+		if recent := recentRxWedgeExits(exitLog, now, rxWedgeLoopWindow); len(recent) >= rxWedgeLoopMax {
+			slog.Error("transport wedged but restart-loop detected — withholding exit, staying up to soft-recover",
+				"reason", wedgeReason,
+				"recent_exits", len(recent),
+				"window", rxWedgeLoopWindow.String(),
+				"silent_for", silence.Truncate(time.Second).String())
+			d.publishEvent("tunnel.rx_wedge_restart_loop", map[string]any{
+				"reason":             wedgeReason,
+				"recent_exits":       len(recent),
+				"window_seconds":     int64(rxWedgeLoopWindow.Seconds()),
+				"silent_for_seconds": int64(silence.Seconds()),
+			})
+			st.softAttempts = 0
+			return rxActionRestartLoop
+		}
 		slog.Error("transport wedged — exiting for supervisor respawn",
 			"reason", wedgeReason,
 			"silent_for", silence.Truncate(time.Second).String(),
@@ -277,6 +319,9 @@ func (d *Daemon) rxWatchdogTick(st *rxWatchdogState, now time.Time) (action rxWa
 			"soft_attempts":            st.softAttempts,
 			"exit_code":                rxWedgeExitCode,
 		})
+		// Record this exit BEFORE calling exit so the respawned process
+		// sees it in the recent-exits count.
+		recordRxWedgeExit(exitLog, now, rxWedgeLoopWindow)
 		rxWatchdogExit(rxWedgeExitCode)
 		return rxActionExit
 	}
@@ -285,3 +330,66 @@ func (d *Daemon) rxWatchdogTick(st *rxWatchdogState, now time.Time) (action rxWa
 	st.softAttempts = 0
 	return rxActionExitWithheld
 }
+
+// rxWedgeExitLogPath returns the sidecar file that records recent hard-
+// exit timestamps, stored next to the identity file so the count
+// survives the respawn. An empty identityPath (no persistence) returns
+// "", which disables the restart-loop breaker — the daemon behaves
+// exactly as before (always allowed to exit).
+func rxWedgeExitLogPath(identityPath string) string {
+	if identityPath == "" {
+		return ""
+	}
+	return identityPath + ".rxwedge"
+}
+
+// recentRxWedgeExits reads hard-exit timestamps from path and returns
+// those within window of now. A missing or unreadable file yields an
+// empty slice (breaker closed — exit allowed); malformed lines are
+// skipped. Never errors: a read problem must not itself block a
+// legitimate wedge exit.
+func recentRxWedgeExits(path string, now time.Time, window time.Duration) []int64 {
+	if path == "" {
+		return nil
+	}
+	// #nosec G304 -- path is derived from the daemon's own configured
+	// IdentityPath (rxWedgeExitLogPath), not from any peer/user input.
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil
+	}
+	cutoff := now.Add(-window).UnixNano()
+	var recent []int64
+	for _, line := range strings.Split(strings.TrimSpace(string(data)), "\n") {
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+		ts, err := strconv.ParseInt(line, 10, 64)
+		if err != nil {
+			continue
+		}
+		if ts >= cutoff {
+			recent = append(recent, ts)
+		}
+	}
+	return recent
+}
+
+// recordRxWedgeExit appends now to the exit log (pruned to window) so the
+// respawned process sees an accurate recent-exit count. Best-effort: a
+// failed write just under-counts, degrading to the prior always-restart
+// behavior — never worse.
+func recordRxWedgeExit(path string, now time.Time, window time.Duration) {
+	if path == "" {
+		return
+	}
+	recent := recentRxWedgeExits(path, now, window)
+	recent = append(recent, now.UnixNano())
+	var b strings.Builder
+	for _, ts := range recent {
+		b.WriteString(strconv.FormatInt(ts, 10))
+		b.WriteByte('\n')
+	}
+	_ = os.WriteFile(path, []byte(b.String()), 0o600)
+}

pkg/daemon/transport/wss/wss.go

@@ -55,6 +55,16 @@ import (
 // long enough not to flood the link.
 const DefaultIdlePingInterval = 30 * time.Second
 
+// DefaultIdlePingTimeout bounds how long a single idle keepalive ping
+// may wait for its pong before the connection is treated as half-open.
+// On a silently-dead conn (stale NAT/relay mapping, half-open TCP) the
+// pong never arrives; without this bound the ping blocks forever, the
+// keepalive loop wedges, and the daemon never notices the beacon is
+// gone. Kept well under DefaultIdlePingInterval so a slow-but-live link
+// is not falsely reconnected. This is the WSS-transport analogue of the
+// L4 rx-watchdog: detect a silently-dead inbound path and recover it.
+const DefaultIdlePingTimeout = 10 * time.Second
+
 // DefaultDialTimeout caps the time we spend on a single dial attempt
 // (DNS + TCP + TLS + WS upgrade + auth challenge). Beyond this we
 // fail fast and let the reconnect loop try again.
@@ -110,6 +120,10 @@ type Config struct {
 	// IdlePingInterval overrides DefaultIdlePingInterval.
 	IdlePingInterval time.Duration
 
+	// IdlePingTimeout overrides DefaultIdlePingTimeout — the per-ping
+	// deadline that turns a silently-dead conn into a reconnect.
+	IdlePingTimeout time.Duration
+
 	// DialTimeout overrides DefaultDialTimeout.
 	DialTimeout time.Duration
 
@@ -193,6 +207,9 @@ func Dial(ctx context.Context, cfg Config) (*Transport, error) {
 	if cfg.IdlePingInterval == 0 {
 		cfg.IdlePingInterval = DefaultIdlePingInterval
 	}
+	if cfg.IdlePingTimeout == 0 {
+		cfg.IdlePingTimeout = DefaultIdlePingTimeout
+	}
 	if cfg.DialTimeout == 0 {
 		cfg.DialTimeout = DefaultDialTimeout
 	}
@@ -552,6 +569,16 @@ func (t *Transport) drainReads(conn *websocket.Conn) {
 // to keep the WSS connection alive through proxy idle timeouts. Runs until
 // Close() fires via lifetimeCtx. Skips pings while the supervisor is
 // reconnecting (conn == nil).
+//
+// Each ping is bounded by IdlePingTimeout. A pong that never arrives —
+// the signature of a silently half-open conn (stale NAT/relay mapping,
+// half-open TCP the kernel hasn't torn down) — would otherwise block
+// this loop forever while it holds writeMu, freezing every Send behind
+// it and leaving the dead beacon conn undetected. On a ping error we
+// force the conn closed: that unblocks the supervisor's drainReads Read,
+// which drives the reconnect. Without this, a half-open beacon link is
+// invisible to the transport (Send only fails once there is traffic to
+// send, and drainReads sits blocked on a Read that never returns).
 func (t *Transport) idlePing() {
 	ticker := time.NewTicker(t.cfg.IdlePingInterval)
 	defer ticker.Stop()
@@ -569,11 +596,27 @@ func (t *Transport) idlePing() {
 			if conn == nil {
 				continue
 			}
+			pingCtx, cancel := context.WithTimeout(t.lifetimeCtx, t.cfg.IdlePingTimeout)
 			t.writeMu.Lock()
-			err := conn.Ping(t.lifetimeCtx)
+			err := conn.Ping(pingCtx)
 			t.writeMu.Unlock()
+			cancel()
 			if err != nil {
-				slog.Debug("wss transport: idle ping failed", "err", err)
+				if t.closed.Load() {
+					return
+				}
+				// Half-open conn: the pong timed out. Force the conn
+				// closed so the supervisor's blocked drainReads Read
+				// returns an error and the reconnect sequence fires.
+				// Also clear t.conn so Send fails fast meanwhile.
+				slog.Warn("wss transport: idle ping timed out — forcing reconnect",
+					"err", err, "timeout", t.cfg.IdlePingTimeout)
+				t.connMu.Lock()
+				if t.conn == conn {
+					t.conn = nil
+				}
+				t.connMu.Unlock()
+				_ = conn.Close(websocket.StatusPolicyViolation, "idle ping timeout")
 			}
 		}
 	}