bastion 0.61.2

CHANGELOG.md

@@ -22,6 +22,18 @@ From 0.39.1 on, a heading marks a version *not published* or *held* once
 that is settled. The newest entries may not be published yet; the manifest
 says which version is. Older entries are not marked.
 
+## 0.61.2
+
+- **A machine a Host holds stops dialling it.** When a Host's owner
+  quarantines this machine there, the Host refuses each of the bastion's lane
+  connections, and three of the five lanes dialled again at once, the other
+  two a second later: tens of connections a second, each a TLS handshake.
+  Now a refusal holds all of that pairing's lanes for 5 seconds, twice as long
+  after each further refusal up to 5 minutes; and while the Host's heartbeat
+  reply says the machine is quarantined, the lanes do not dial at all. The
+  first reply that no longer says so brings them back. Other Hosts the
+  machine is paired with are not slowed.
+
 ## 0.61.1
 
 - **An op that creates a folder runs on every Linux kernel.** Where the

README.md

@@ -3,7 +3,7 @@
 [![License: GPL v3+](https://img.shields.io/badge/license-GPL--3.0--or--later-blue.svg)](LICENSE)
 [![Go 1.26+](https://img.shields.io/badge/go-1.26%2B-00ADD8.svg)](client-go/go.mod)
 [![Platforms](https://img.shields.io/badge/platforms-linux%20%7C%20macOS%20%7C%20windows-lightgrey.svg)](#supported-platforms)
-[![ci](https://git.privatae.ai/privatae/privatae-bastion/actions/workflows/ci.yml/badge.svg?branch=main)](https://git.privatae.ai/privatae/privatae-bastion/actions)
+[![ci](https://github.com/Privatae/bastion/actions/workflows/ci.yml/badge.svg?branch=main)](https://github.com/Privatae/bastion/actions/workflows/ci.yml)
 
 **See every AI agent running on your computer, what each one is doing and what it can reach, and stop any of them. One program, no account, no AI of its own.**
 

client-go/egress_relay.go

@@ -124,7 +124,11 @@ func runEgressWorkers(pctx context.Context, self string, wsURL string, tlsCfg *t
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
+			hold := laneHoldFor(self)
 			for pctx.Err() == nil {
+				if !hold.waitClear(pctx) {
+					return
+				}
 				ctx, cancel := context.WithTimeout(pctx, 25*time.Second)
 				ws, _, err := dialer.DialContext(ctx, wsURL, http.Header{})
 				cancel()
@@ -137,12 +141,17 @@ func runEgressWorkers(pctx context.Context, self string, wsURL string, tlsCfg *t
 					sleepCtx(pctx, time.Second)
 					continue
 				}
+				held := watchHeldClose(ws)
 				stopKA := armWSKeepalive(ws) // ping/read-deadline: a dead tunnel unblocks the park
 				unpark := context.AfterFunc(pctx, func() { _ = ws.Close() })
 				took := handleEgressConn(&wsConn{ws: ws}, dialOut, self) // blocks until a job, serves it, then returns
 				unpark()
 				stopKA()
 				_ = ws.Close()
+				if held() {
+					hold.refused(self) // the Host holds this machine (lanehold.go)
+					continue
+				}
 				if !took {
 					sleepCtx(pctx, egressRefusedWait)
 				}

client-go/harness.go

@@ -114,7 +114,11 @@ func runHarnessWorkers(pctx context.Context, self, dcnDir string, wsURL string,
 	}
 	for i := 0; i < n; i++ {
 		go func() {
+			hold := laneHoldFor(self)
 			for pctx.Err() == nil {
+				if !hold.waitClear(pctx) {
+					return
+				}
 				ctx, cancel := context.WithTimeout(pctx, 25*time.Second)
 				ws, _, err := dialer.DialContext(ctx, wsURL, http.Header{})
 				cancel()
@@ -127,12 +131,16 @@ func runHarnessWorkers(pctx context.Context, self, dcnDir string, wsURL string,
 					sleepCtx(pctx, time.Second)
 					continue
 				}
+				held := watchHeldClose(ws)
 				stopKA := armWSKeepalive(ws) // ping/read-deadline: a dead tunnel unblocks the park
 				unpark := context.AfterFunc(pctx, func() { _ = ws.Close() })
 				handleHarnessConn(ws, self, dcnDir) // blocks until a job, serves it, then returns
 				unpark()
 				stopKA()
 				_ = ws.Close()
+				if held() {
+					hold.refused(self) // the Host holds this machine (lanehold.go)
+				}
 			}
 		}()
 	}

client-go/lanehold.go

@@ -0,0 +1,157 @@
+package main
+
+// When a Host holds this machine (its owner quarantined it there), the Host
+// refuses every worker socket of every lane with close code 4403, and says
+// "quarantined": true on each beat it answers. The lane workers park a pool of
+// sockets per lane and used to dial again at once (mgmt, harness, session) or
+// a second after a close (egress, serve), so one held machine whose tunnel
+// stayed up dialled the Host 30 to 80 times a second, each one a TLS
+// handshake. A pairing's workers now share one hold:
+//
+//   - a socket the Host closes with 4403 holds every worker of that pairing for
+//     laneHoldFirst, doubling with each hold up to laneHoldMax;
+//   - a beat that says quarantined stops them dialling until a beat no longer
+//     does, which is how the owner's release reaches this machine.
+//
+// The hold is the pairing's, so another Host this machine is paired with is
+// never slowed by it.
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"math/rand/v2"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+// laneHeldCloseCode is the close code a Host gives a worker socket of a
+// machine it holds (the Host's quarantine gate).
+const laneHeldCloseCode = 4403
+
+// The first hold, and the longest. Variables so a test can shorten them.
+var (
+	laneHoldFirst = 5 * time.Second
+	laneHoldMax   = 5 * time.Minute
+)
+
+type laneHold struct {
+	mu          sync.Mutex
+	wait        time.Duration // the last hold's length; the next one doubles it
+	until       time.Time     // no worker of the pairing dials before this
+	last        time.Time     // when the last hold began
+	quarantined bool          // what the last beat said
+	wake        chan struct{} // closed when the beat lifts a quarantine
+}
+
+var laneHolds sync.Map // host name -> *laneHold
+
+func laneHoldFor(host string) *laneHold {
+	if h, ok := laneHolds.Load(host); ok {
+		return h.(*laneHold)
+	}
+	h, _ := laneHolds.LoadOrStore(host, &laneHold{wake: make(chan struct{})})
+	return h.(*laneHold)
+}
+
+// refused is a worker socket the Host closed with 4403. The workers of a pool
+// are refused together, so a refusal while a hold is on is that hold's and
+// does not lengthen it. A refusal long after the last hold starts again from
+// laneHoldFirst.
+func (h *laneHold) refused(host string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	now := time.Now()
+	if now.Before(h.until) {
+		return
+	}
+	switch {
+	case h.wait == 0 || now.Sub(h.last) > 2*laneHoldMax:
+		h.wait = laneHoldFirst
+	default:
+		h.wait = min(2*h.wait, laneHoldMax)
+	}
+	h.last = now
+	h.until = now.Add(h.wait)
+	fmt.Printf("[%s] the Host refused a lane: it holds this machine; its lanes dial again in %s\n", host, h.wait)
+}
+
+// fromBeat is what a beat's reply said. A release ends the hold at once and
+// starts the next one from laneHoldFirst.
+func (h *laneHold) fromBeat(host string, quarantined bool) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	switch {
+	case quarantined && !h.quarantined:
+		h.quarantined = true
+		fmt.Printf("[%s] the Host holds this machine (quarantined there): its lanes wait until its owner releases it\n", host)
+	case !quarantined && h.quarantined:
+		h.quarantined = false
+		h.wait, h.until = 0, time.Time{}
+		close(h.wake)
+		h.wake = make(chan struct{})
+		fmt.Printf("[%s] the Host released this machine: its lanes are back\n", host)
+	}
+}
+
+// waitClear blocks a worker until it may dial: the hold is over and the last
+// beat did not say quarantined. A worker waits a little past the hold's end,
+// a different little for each, so a pool does not dial in one burst. False
+// when ctx ends first.
+func (h *laneHold) waitClear(ctx context.Context) bool {
+	for {
+		h.mu.Lock()
+		quarantined, until, wake := h.quarantined, h.until, h.wake
+		h.mu.Unlock()
+		if ctx.Err() != nil {
+			return false
+		}
+		var t *time.Timer
+		var timer <-chan time.Time
+		switch {
+		case quarantined:
+			// Only a beat lifts it.
+		case time.Now().Before(until):
+			d := time.Until(until)
+			t = time.NewTimer(d + rand.N(d/10+time.Millisecond))
+			timer = t.C
+		default:
+			return true
+		}
+		select {
+		case <-ctx.Done():
+		case <-wake:
+		case <-timer:
+		}
+		if t != nil {
+			t.Stop()
+		}
+	}
+}
+
+// watchHeldClose reports, once the socket has been read to its end, whether
+// the Host closed it with laneHeldCloseCode.
+func watchHeldClose(ws *websocket.Conn) (held func() bool) {
+	var got atomic.Bool
+	def := ws.CloseHandler()
+	ws.SetCloseHandler(func(code int, text string) error {
+		if code == laneHeldCloseCode {
+			got.Store(true)
+		}
+		return def(code, text)
+	})
+	return got.Load
+}
+
+// beatSaysQuarantined reads the "quarantined" a Host puts on the beat's reply
+// for a machine it holds. A reply without it, or one that cannot be read, is
+// not a hold.
+func beatSaysQuarantined(body []byte) bool {
+	var r struct {
+		Quarantined bool `json:"quarantined"`
+	}
+	return json.Unmarshal(body, &r) == nil && r.Quarantined
+}

client-go/lanehold_test.go

@@ -0,0 +1,223 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+// heldHost stands in for a Host's lane port: it counts each worker's hello,
+// then either closes the socket with the code a Host gives a machine it holds,
+// or parks it.
+type heldHost struct {
+	hellos atomic.Int32
+	hold   atomic.Bool
+	tls    *tls.Config
+	dial   func(ctx context.Context) (net.Conn, error)
+	url    string
+}
+
+func startHeldHost(t *testing.T) *heldHost {
+	t.Helper()
+	h := &heldHost{}
+	h.hold.Store(true)
+	up := websocket.Upgrader{}
+	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		c, err := up.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer c.Close()
+		if _, _, err := c.ReadMessage(); err != nil {
+			return
+		}
+		h.hellos.Add(1)
+		if h.hold.Load() {
+			_ = c.WriteControl(websocket.CloseMessage,
+				websocket.FormatCloseMessage(laneHeldCloseCode, "this bastion is quarantined"), time.Now().Add(time.Second))
+			return
+		}
+		for {
+			if _, _, err := c.ReadMessage(); err != nil {
+				return
+			}
+		}
+	}))
+	srv.StartTLS()
+	t.Cleanup(srv.Close)
+	pool := x509.NewCertPool()
+	pool.AddCert(srv.Certificate())
+	h.tls = &tls.Config{RootCAs: pool, ServerName: "example.com"}
+	addr := srv.Listener.Addr().String()
+	h.dial = func(ctx context.Context) (net.Conn, error) {
+		var d net.Dialer
+		return d.DialContext(ctx, "tcp", addr)
+	}
+	h.url = "wss://example.com/lane"
+	return h
+}
+
+// shortHolds makes the holds short enough for a test. Set before any worker
+// runs, and never from a parallel subtest.
+func shortHolds(t *testing.T, first, max time.Duration) {
+	t.Helper()
+	f, m := laneHoldFirst, laneHoldMax
+	laneHoldFirst, laneHoldMax = first, max
+	t.Cleanup(func() { laneHoldFirst, laneHoldMax = f, m })
+}
+
+// ownHold gives a test its own pairing's hold, and forgets it after.
+func ownHold(t *testing.T, host string) {
+	t.Helper()
+	laneHolds.Delete(host)
+	t.Cleanup(func() { laneHolds.Delete(host) })
+}
+
+// Every lane's pool, so a lane added later without the hold is caught.
+func lanePools(dir string) map[string]func(ctx context.Context, host string, h *heldHost, n int) {
+	return map[string]func(ctx context.Context, host string, h *heldHost, n int){
+		"egress": func(ctx context.Context, host string, h *heldHost, n int) {
+			var d net.Dialer
+			runEgressWorkers(ctx, host, h.url, h.tls, h.dial, "b", d.DialContext, n)
+		},
+		"serve": func(ctx context.Context, host string, h *heldHost, n int) {
+			runServeWorkers(ctx, host, dir, h.url, h.tls, h.dial, "b", n)
+		},
+		"mgmt": func(ctx context.Context, host string, h *heldHost, n int) {
+			runMgmtWorkers(ctx, host, h.url, h.tls, h.dial, "b", dir, dir, dir, n)
+		},
+		"harness": func(ctx context.Context, host string, h *heldHost, n int) {
+			runHarnessWorkers(ctx, host, dir, h.url, h.tls, h.dial, "b", n)
+		},
+		"session": func(ctx context.Context, host string, h *heldHost, n int) {
+			runSessionWorkers(ctx, host, h.url, h.tls, h.dial, "b", n,
+				sessionEnv{CtxRoot: dir, DcnDir: dir, HostName: host, Server: "test"})
+		},
+	}
+}
+
+// A Host that holds this machine closes every worker socket with 4403. No
+// worker of any lane dials it again before the hold is over: one hello per
+// worker, where a pool used to dial again at once or a second later, 30 to 80
+// times a second from one held machine.
+func TestEveryLanePoolWaitsOutAHostThatHoldsThisMachine(t *testing.T) {
+	shortHolds(t, 3*time.Second, time.Minute)
+	for lane, run := range lanePools(t.TempDir()) {
+		t.Run(lane, func(t *testing.T) {
+			t.Parallel()
+			host := "held-" + lane + ".example.test"
+			ownHold(t, host)
+			h := startHeldHost(t)
+			ctx, cancel := context.WithCancel(context.Background())
+			t.Cleanup(cancel)
+			const n = 2
+			go run(ctx, host, h, n)
+			time.Sleep(2 * time.Second)
+			if got := h.hellos.Load(); got != n {
+				t.Fatalf("%d dials within the hold from %d workers; want one each", got, n)
+			}
+		})
+	}
+}
+
+// A beat whose reply says quarantined stops the lanes before any of them
+// dials, and the first beat that no longer says it brings them back at once.
+func TestABeatThatSaysQuarantinedStopsTheLanesUntilOneDoesNot(t *testing.T) {
+	host := "quarantined.example.test"
+	shortHolds(t, time.Hour, time.Hour)
+	ownHold(t, host)
+	h := startHeldHost(t)
+	h.hold.Store(false)
+	laneHoldFor(host).fromBeat(host, beatSaysQuarantined([]byte(`{"ok":true,"quarantined":true}`)))
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+	const n = 2
+	go lanePools(t.TempDir())["mgmt"](ctx, host, h, n)
+	time.Sleep(700 * time.Millisecond)
+	if got := h.hellos.Load(); got != 0 {
+		t.Fatalf("%d dials while the Host says it holds this machine", got)
+	}
+	laneHoldFor(host).fromBeat(host, beatSaysQuarantined([]byte(`{"ok":true}`)))
+	deadline := time.Now().Add(3 * time.Second)
+	for h.hellos.Load() < n && time.Now().Before(deadline) {
+		time.Sleep(20 * time.Millisecond)
+	}
+	if got := h.hellos.Load(); got != n {
+		t.Fatalf("after the release, %d of %d workers parked again", got, n)
+	}
+}
+
+// A pool's workers are refused together, so a refusal inside a hold is that
+// hold's; each hold after one runs out is twice as long, up to the most; a
+// refusal long after the last starts again from the first; a release ends a
+// hold at once.
+func TestAHoldDoublesOncePerHoldUpToItsMost(t *testing.T) {
+	host := "doubling.example.test"
+	shortHolds(t, time.Second, 4*time.Second)
+	ownHold(t, host)
+	h := laneHoldFor(host)
+	expire := func() {
+		h.mu.Lock()
+		h.until = time.Now().Add(-time.Millisecond)
+		h.mu.Unlock()
+	}
+	wait := func() time.Duration {
+		h.mu.Lock()
+		defer h.mu.Unlock()
+		return h.wait
+	}
+	h.refused(host)
+	h.refused(host)
+	h.refused(host)
+	if got := wait(); got != time.Second {
+		t.Fatalf("three refusals in one hold: %s, want the first hold", got)
+	}
+	var got []string
+	for range 3 {
+		expire()
+		h.refused(host)
+		got = append(got, wait().String())
+	}
+	if strings.Join(got, " ") != "2s 4s 4s" {
+		t.Fatalf("holds after the first: %v, want 2s 4s 4s", got)
+	}
+	expire()
+	h.mu.Lock()
+	h.last = time.Now().Add(-9 * time.Second)
+	h.mu.Unlock()
+	h.refused(host)
+	if got := wait(); got != time.Second {
+		t.Fatalf("a refusal after a long quiet: %s, want the first hold again", got)
+	}
+	h.fromBeat(host, true)
+	h.fromBeat(host, false)
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	if !h.waitClear(ctx) {
+		t.Fatal("a release must end the hold at once")
+	}
+}
+
+func TestBeatSaysQuarantinedReadsOnlyTheHostsTrue(t *testing.T) {
+	for body, want := range map[string]bool{
+		`{"ok":true,"quarantined":true}`:  true,
+		`{"ok":true,"quarantined":false}`: false,
+		`{"ok":true}`:                     false,
+		`{"quarantined":"yes"}`:           false,
+		`not json`:                        false,
+		``:                                false,
+	} {
+		if got := beatSaysQuarantined([]byte(body)); got != want {
+			t.Errorf("%q: %v, want %v", body, got, want)
+		}
+	}
+}

client-go/mgmt.go

@@ -347,7 +347,11 @@ func runMgmtWorkers(pctx context.Context, self string, wsURL string, tlsCfg *tls
 	}
 	for i := 0; i < n; i++ {
 		go func() {
+			hold := laneHoldFor(self)
 			for pctx.Err() == nil {
+				if !hold.waitClear(pctx) {
+					return
+				}
 				ctx, cancel := context.WithTimeout(pctx, 25*time.Second)
 				ws, _, err := dialer.DialContext(ctx, wsURL, http.Header{})
 				cancel()
@@ -360,6 +364,7 @@ func runMgmtWorkers(pctx context.Context, self string, wsURL string, tlsCfg *tls
 					sleepCtx(pctx, time.Second)
 					continue
 				}
+				held := watchHeldClose(ws)
 				stopKA := armWSKeepalive(ws) // ping/read-deadline: a dead tunnel unblocks the park
 				// A revoke while parked: closing the socket is what unblocks the read.
 				unpark := context.AfterFunc(pctx, func() { _ = ws.Close() })
@@ -367,6 +372,9 @@ func runMgmtWorkers(pctx context.Context, self string, wsURL string, tlsCfg *tls
 				unpark()
 				stopKA()
 				_ = ws.Close()
+				if held() {
+					hold.refused(self) // the Host holds this machine (lanehold.go)
+				}
 			}
 		}()
 	}

client-go/run.go

@@ -646,6 +646,8 @@ func runPairing(ctx context.Context, cfg *Config) error {
 					// silence path reads this — 14 days with no 2xx from the
 					// anchor host lets root re-anchor without its consent.
 					recordPairingOK(cfg.HostName)
+					// A Host that holds this machine says so; its lanes wait (lanehold.go).
+					laneHoldFor(cfg.HostName).fromBeat(cfg.HostName, beatSaysQuarantined(body))
 				}
 				// Box-wide directives (update, contain, mirror) are anchor-only
 				// on an anchored box — each takes this pairing's identity and
@@ -891,6 +893,7 @@ func runUnixPairing(ctx context.Context, cfg *Config) error {
 			}
 			if resp.StatusCode < 300 {
 				recordPairingOK(cfg.HostName)
+				laneHoldFor(cfg.HostName).fromBeat(cfg.HostName, beatSaysQuarantined(body))
 			}
 			maybeSelfUpdate(cfg.HostName, cfg.dcnDir(), body)
 			maybeContain(cfg.HostName, cfg.dcnDir(), body)

client-go/servelane.go

@@ -616,7 +616,11 @@ func runServeWorkers(pctx context.Context, self, dcnDir, wsURL string, tlsCfg *t
 	spawn = func() {
 		go func() {
 			wait := 3 * time.Second
+			hold := laneHoldFor(self)
 			for pctx.Err() == nil {
+				if !hold.waitClear(pctx) {
+					return
+				}
 				ctx, cancel := context.WithTimeout(pctx, 25*time.Second)
 				ws, resp, err := dialer.DialContext(ctx, wsURL, http.Header{})
 				cancel()
@@ -635,6 +639,7 @@ func runServeWorkers(pctx context.Context, self, dcnDir, wsURL string, tlsCfg *t
 					sleepCtx(pctx, time.Second)
 					continue
 				}
+				held := watchHeldClose(ws)
 				stopKA := armWSKeepalive(ws)
 				unpark := context.AfterFunc(pctx, func() { _ = ws.Close() })
 				took := serveJob(ws, self, dcnDir, spawn)
@@ -644,6 +649,10 @@ func runServeWorkers(pctx context.Context, self, dcnDir, wsURL string, tlsCfg *t
 				if took {
 					return
 				}
+				if held() {
+					hold.refused(self) // the Host holds this machine (lanehold.go)
+					continue
+				}
 				// Closed while parked (a Host restart), or the job refused: not
 				// straight back, so a Host that closes or misuses every worker is
 				// not dialled in a loop.

client-go/session.go

@@ -83,7 +83,11 @@ func runSessionWorkers(pctx context.Context, self string, wsURL string, tlsCfg *
 	}
 	for i := 0; i < n; i++ {
 		go func() {
+			hold := laneHoldFor(self)
 			for pctx.Err() == nil {
+				if !hold.waitClear(pctx) {
+					return
+				}
 				ctx, cancel := context.WithTimeout(pctx, 25*time.Second)
 				ws, _, err := dialer.DialContext(ctx, wsURL, http.Header{})
 				cancel()
@@ -96,6 +100,7 @@ func runSessionWorkers(pctx context.Context, self string, wsURL string, tlsCfg *
 					sleepCtx(pctx, time.Second)
 					continue
 				}
+				held := watchHeldClose(ws)
 				stopKA := armWSKeepalive(ws)
 				// A revoked pairing closes the socket under the session, which ends the
 				// shell through the same path a host disconnect takes.
@@ -104,6 +109,9 @@ func runSessionWorkers(pctx context.Context, self string, wsURL string, tlsCfg *
 				unpark()
 				stopKA()
 				_ = ws.Close()
+				if held() {
+					hold.refused(self) // the Host holds this machine (lanehold.go)
+				}
 			}
 		}()
 	}

deploy/VERSION

@@ -1 +1 @@
-0.61.1
+0.61.2