egress, server, pool: bound guest hosts and claim refs, sleeping relay, exec kill, renew idle clock

docs/egress.md

@@ -257,7 +257,10 @@ Each decision is metered as an `egress` usage event, which names the injected
 secrets in `secret` as the audit line does, and, when `audit_log` is
 on, written to `audit.jsonl` (`op:"egress"`, `dest`, `port`, `method`,
 `decision`, and the secret **name** in `secret`); an intercepted `CONNECT` is
-recorded as a decision of its own before its inner requests.
+recorded as a decision of its own before its inner requests. A host longer
+than 253 bytes (the DNS limit) is denied, and a sandbox records at most 10
+`deny` and `rate_limited` decisions per second after a burst of 100, dropping
+the rest, so a guest cannot flood the journals.
 
 ## Claim env and secrets
 

docs/sandboxd-api.md

@@ -42,7 +42,8 @@ Auth: `Authorization: Bearer <api_token>` (when configured).
 - `ttl_seconds` 0 means the server default (5 minutes); capped at 24h. The
   owning node reaps the sandbox after the TTL even if the client vanishes
 - `claim_ref` is an optional opaque caller reference echoed by the scoped
-  sandbox index; the aggregated apiserver uses `<namespace>/<name>`
+  sandbox index, at most 512 bytes of printable UTF-8 like a metadata value;
+  the aggregated apiserver uses `<namespace>/<name>`
 - `metadata` is an optional map of caller labels, echoed and filterable by
   the [sandbox index](#get-v1sandboxes): at most 16 pairs, each key 1 to 128
   bytes of printable ASCII without `=` or `&` (list filters encode pairs as
@@ -67,8 +68,8 @@ Auth: `Authorization: Bearer <api_token>` (when configured).
   `no_redirect` retry so a target whose gossip view is stale refuses instead of
   cold-booting an image named like the template
 - `env` is the claim's own environment, at most 64 entries: each name a shell
-  identifier of at most 128 bytes, each value at most 8 KiB with no control
-  characters other than tab, and names, values and inject fields at most 64 KiB in total. An entry holds only `value`, `guest` and `inject`; any other
+  identifier of at most 128 bytes, each value at most 8 KiB, a value and an
+  inject `query` with no control characters other than tab, and names, values and inject fields at most 64 KiB in total. An entry holds only `value`, `guest` and `inject`; any other
   member, or `"guest": null`, answers 400, so a mistyped flag never lands a
   host-only value in the guest. `guest` (default `true`) delivers the entry into the guest; a
   `guest: false` entry never enters it and only feeds the node's egress
@@ -554,6 +555,7 @@ claim. The sandbox's own token rides in the body as the ownership proof:
 `claim_ref_prefix`, when set, records each child under `prefix + child id` as
 its `claim_ref`, so a control plane that names claims `<namespace>/<name>`
 addresses the children by name; without it children carry no `claim_ref`.
+The prefix is at most 512 bytes of printable UTF-8, as `claim_ref` is.
 Children inherit the parent's `metadata`, `on_expire` and `"egress": false`; an
 `on_expire` in the body overrides the inherited action.
 Clones the sandbox into `count` fresh claims (1 up to the node's

sandboxd/egress/proxy.go

@@ -25,6 +25,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"golang.org/x/time/rate"
+
 	"github.com/cocoonstack/sandbox/sandboxd/utils"
 )
 
@@ -34,6 +36,10 @@ const (
 	maxIdleConns  = 64
 	maxInjectBody = 64 << 10
 	scrubChunk    = 16 << 10
+	maxHostLen    = 253
+
+	refusalRate  = 10
+	refusalBurst = 100
 )
 
 var (
@@ -143,6 +149,7 @@ type Proxy struct {
 	router   Router
 	holder   Holder
 	pools    atomic.Pointer[connPools]
+	refusals *rate.Limiter
 
 	// leaves caches this sandbox's interception leaves; per-proxy, never shared.
 	leafMu sync.Mutex
@@ -162,13 +169,14 @@ func New(policy Evaluator, secrets Secrets, ca *CA, router Router, audit AuditFu
 		return router.Dial(ctx, "", network, addr)
 	}
 	p := &Proxy{
-		policy:  policy,
-		secrets: secrets,
-		ca:      ca,
-		audit:   audit,
-		router:  router,
-		holder:  holder,
-		conns:   map[net.Conn]struct{}{},
+		policy:   policy,
+		secrets:  secrets,
+		ca:       ca,
+		audit:    audit,
+		router:   router,
+		holder:   holder,
+		refusals: rate.NewLimiter(refusalRate, refusalBurst),
+		conns:    map[net.Conn]struct{}{},
 	}
 	// The stdlib default of 2 idle conns per host re-dials bursty same-host traffic.
 	pools := &connPools{tr: &http.Transport{DialContext: direct, MaxIdleConns: maxIdleConns, MaxIdleConnsPerHost: 8, IdleConnTimeout: idleConnTimeout}, routed: map[routeKey]*http.Transport{}}
@@ -461,9 +469,11 @@ func (p *Proxy) reportTransfer(ev Event, sent, received int64) {
 }
 
 func (p *Proxy) record(ev Event) {
-	if p.audit != nil {
-		p.audit(ev)
+	if p.audit == nil || ev.Decision != DecisionAllow && !p.refusals.Allow() {
+		return
 	}
+	ev.Host, ev.Method = ev.Host[:min(len(ev.Host), maxHostLen)], ev.Method[:min(len(ev.Method), maxHostLen)]
+	p.audit(ev)
 }
 
 // countingBody counts the request body bytes the transport reads, possibly after RoundTrip returns.
@@ -721,20 +731,20 @@ func denied(w http.ResponseWriter, host string) {
 	http.Error(w, fmt.Sprintf("egress denied: %s", host), http.StatusForbidden)
 }
 
-// hostPort splits an authority; a bare host takes def, a port outside 1-65535 or a malformed authority is refused rather than defaulted.
+// hostPort splits an authority; a bare host takes def, a port outside 1-65535, a host past 253 bytes or a malformed authority is refused rather than defaulted.
 func hostPort(authority string, def uint16) (host string, port uint16, ok bool) {
 	host, text, err := net.SplitHostPort(authority)
 	if err != nil {
 		literal, bracketed := strings.CutPrefix(authority, "[")
 		if !bracketed {
-			return authority, def, authority != "" && !strings.ContainsAny(authority, ":[]")
+			return authority, def, authority != "" && len(authority) <= maxHostLen && !strings.ContainsAny(authority, ":[]")
 		}
 		literal, closed := strings.CutSuffix(literal, "]")
 		addr, parseErr := netip.ParseAddr(literal)
-		return literal, def, closed && parseErr == nil && addr.Is6()
+		return literal, def, closed && parseErr == nil && addr.Is6() && len(literal) <= maxHostLen
 	}
 	n, err := strconv.ParseUint(text, 10, 16)
-	return host, uint16(n), err == nil && n != 0
+	return host, uint16(n), err == nil && n != 0 && len(host) <= maxHostLen
 }
 
 // replaceParam fills the first name=placeholder pair of a query or form string, leaving every other byte as sent.

sandboxd/egress/proxy_test.go

@@ -10,8 +10,10 @@ import (
 	"net/http/httptest"
 	"net/netip"
 	"net/url"
+	"strings"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 )
 
@@ -570,6 +572,58 @@ func TestBracketedIPv6AuthorityIsUnwrapped(t *testing.T) {
 	}
 }
 
+func TestAHostPastTheDNSLimitIsRefusedAndRecordedClipped(t *testing.T) {
+	policy := Policy{Allow: []Rule{{Host: "*"}}}
+	events := make(chan Event, 4)
+	p := New(policy, nil, nil, fixedDial(echoServer(t)), func(ev Event) { events <- ev }, nil)
+	front := httptest.NewServer(p)
+	defer front.Close()
+
+	for _, tc := range []struct {
+		host   string
+		status string
+	}{
+		{strings.Repeat("a", maxHostLen+1) + ":443", "HTTP/1.1 403 Forbidden"},
+		{strings.Repeat("b", 64<<10), "HTTP/1.1 403 Forbidden"},
+		{strings.Repeat("c", maxHostLen) + ":443", "HTTP/1.1 200 Connection Established"},
+	} {
+		conn := dialConnect(t, front.Listener.Addr().String(), tc.host)
+		if status := readStatus(t, bufio.NewReader(conn)); status != tc.status {
+			t.Errorf("CONNECT to a %d-byte authority = %q, want %q", len(tc.host), status, tc.status)
+		}
+		_ = conn.Close()
+		if ev := recvEvent(t, events); len(ev.Host) > maxHostLen {
+			t.Errorf("audit event host is %d bytes, want at most %d", len(ev.Host), maxHostLen)
+		}
+	}
+}
+
+func TestRefusalRecordsPastTheBurstAreDropped(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		var allowed, refused int
+		p := New(Policy{}, nil, nil, fixedDial("127.0.0.1:1"), func(ev Event) {
+			if ev.Decision == DecisionAllow {
+				allowed++
+				return
+			}
+			refused++
+		}, nil)
+		for range 2 * refusalBurst {
+			p.record(Event{Decision: DecisionDeny})
+			p.record(Event{Decision: DecisionRateLimited})
+			p.record(Event{Decision: DecisionAllow})
+		}
+		if refused != refusalBurst || allowed != 2*refusalBurst {
+			t.Fatalf("recorded %d refusals and %d allows, want %d and %d", refused, allowed, refusalBurst, 2*refusalBurst)
+		}
+		time.Sleep(time.Second)
+		p.record(Event{Decision: DecisionDeny})
+		if refused != refusalBurst+1 {
+			t.Errorf("a refusal after a second recorded %d, want %d", refused, refusalBurst+1)
+		}
+	})
+}
+
 func TestAuditNamesTheConnectionEnds(t *testing.T) {
 	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
 		_, _ = io.WriteString(w, "ok")

sandboxd/engine/portconn.go

@@ -31,6 +31,9 @@ func newGuestPortConn(conn net.Conn, r *bufio.Reader) *guestPortConn {
 func (g *guestPortConn) Read(p []byte) (int, error) {
 	for len(g.pending) == 0 {
 		line, err := g.readFrame()
+		if errors.Is(err, io.EOF) {
+			return 0, io.ErrUnexpectedEOF
+		}
 		if err != nil {
 			return 0, err
 		}

sandboxd/engine/portconn_test.go

@@ -34,6 +34,28 @@ func TestGuestPortConnFramesBothWays(t *testing.T) {
 	}
 }
 
+func TestGuestPortConnCutBeforeDoneIsUnexpectedEOF(t *testing.T) {
+	for name, tail := range map[string]string{"between frames": "", "mid frame": `{"type":"da`} {
+		t.Run(name, func(t *testing.T) {
+			silk, gc := newTestGuestPortConn(t)
+
+			go func() {
+				frame, _ := json.Marshal(map[string]string{"type": "data", "data": base64.StdEncoding.EncodeToString([]byte("partial"))}, json.Deterministic(true))
+				_, _ = silk.Write(append(frame, '\n'))
+				_, _ = silk.Write([]byte(tail))
+				_ = silk.Close()
+			}()
+			got := make([]byte, 64)
+			if n, err := gc.Read(got); err != nil || string(got[:n]) != "partial" {
+				t.Fatalf("read %q, %v", got[:n], err)
+			}
+			if _, err := gc.Read(got); !errors.Is(err, io.ErrUnexpectedEOF) {
+				t.Errorf("after a cut stream: %v, want io.ErrUnexpectedEOF", err)
+			}
+		})
+	}
+}
+
 func TestGuestPortConnWriteWrapsInDataFrame(t *testing.T) {
 	silk, gc := newTestGuestPortConn(t)
 

sandboxd/pool/claim.go

@@ -197,7 +197,6 @@ func (m *Manager) Renew(ctx context.Context, id string, cred Cred, ttl time.Dura
 		m.store.recommit(ctx, rb)
 		return time.Time{}, fmt.Errorf("renew %s: persist claims: %w", id, err)
 	}
-	sb.Touch()
 	m.recordUsage(ctx, usage.Event{Event: "renew", ID: id, Tenant: sb.Tenant, Deadline: deadline})
 	if rowed {
 		if err := m.rewriteArchiveRow(ctx, sb, false); err != nil {

sandboxd/pool/idle_test.go

@@ -87,6 +87,23 @@ func TestActivityStampsBlockIdleSweep(t *testing.T) {
 	})
 }
 
+func TestRenewKeepsAnIdleClaimIdle(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		eng := newFakeEngine()
+		m := newTestManager(t, eng, config.PoolSpec{PoolKey: testKey, Warm: 1, IdleHibernateSeconds: 1})
+		sb := mustClaim(t, m, testKey)
+		backdate(m, sb, 2*time.Second)
+
+		if _, err := m.Renew(t.Context(), sb.ID, Cred{Token: sb.Token}, time.Hour, ""); err != nil {
+			t.Fatalf("Renew: %v", err)
+		}
+		sweepIdle(t, m)
+		if hibernated(m) != 1 {
+			t.Fatal("a renewed claim with no data-plane activity was not hibernated")
+		}
+	})
+}
+
 func TestHeldConnectionBlocksIdleSweep(t *testing.T) {
 	synctest.Test(t, func(t *testing.T) {
 		eng := newFakeEngine()

sandboxd/pool/journal.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json/v2"
 	"errors"
 	"os"
+	"slices"
 	"sync"
 )
 
@@ -17,6 +18,7 @@ type journal struct {
 	path string
 	f    *os.File
 	size int64
+	torn bool
 }
 
 func newJournal(path string) (*journal, error) {
@@ -49,8 +51,14 @@ func (j *journal) append(v any) error {
 	if j.size+int64(len(line)) > journalMaxBytes {
 		rotateErr = j.rotate()
 	}
+	if j.torn && j.size > 0 {
+		line = slices.Insert(line, 0, '\n')
+	}
 	n, err := j.f.Write(line)
 	j.size += int64(n)
+	if n > 0 {
+		j.torn = line[n-1] != '\n'
+	}
 	return errors.Join(rotateErr, err)
 }
 

sandboxd/pool/journal_test.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"syscall"
 	"testing"
 
 	"github.com/cocoonstack/sandbox/sandboxd/usage"
@@ -120,6 +121,48 @@ func TestJournalKeepsTheFirstAppendAfterATornLine(t *testing.T) {
 	}
 }
 
+func TestJournalEndsALineAShortWriteTore(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "usage.jsonl")
+	j, err := newJournal(path)
+	if err != nil {
+		t.Fatalf("new journal: %v", err)
+	}
+	if err = j.append(usage.Event{Event: "claim", ID: "sb_a"}); err != nil {
+		t.Fatalf("append: %v", err)
+	}
+	var limit syscall.Rlimit
+	if err = syscall.Getrlimit(syscall.RLIMIT_FSIZE, &limit); err != nil {
+		t.Fatalf("getrlimit: %v", err)
+	}
+	short := limit
+	short.Cur = uint64(j.size) + 5
+	if err = syscall.Setrlimit(syscall.RLIMIT_FSIZE, &short); err != nil {
+		t.Fatalf("setrlimit: %v", err)
+	}
+	tornErr := j.append(usage.Event{Event: "release", ID: "sb_a"})
+	if err = syscall.Setrlimit(syscall.RLIMIT_FSIZE, &limit); err != nil {
+		t.Fatalf("restore rlimit: %v", err)
+	}
+	if tornErr == nil {
+		t.Fatal("append past the file size limit succeeded, want a short write")
+	}
+	if err = j.append(usage.Event{Event: "claim", ID: "sb_b"}); err != nil {
+		t.Fatalf("append after the short write: %v", err)
+	}
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	lines := strings.Split(strings.TrimSuffix(string(raw), "\n"), "\n")
+	var ev usage.Event
+	if len(lines) != 3 || json.Unmarshal([]byte(lines[2]), &ev) != nil || ev.ID != "sb_b" {
+		t.Errorf("journal after a short write %q, want the next append on a line of its own", raw)
+	}
+	if st, _ := os.Stat(path); j.size != st.Size() {
+		t.Errorf("journal size %d, file %d", j.size, st.Size())
+	}
+}
+
 func TestJournalLeavesAWholeLastLine(t *testing.T) {
 	path := filepath.Join(t.TempDir(), "usage.jsonl")
 	if err := os.WriteFile(path, []byte(`{"ev":"claim"}`+"\n"), 0o600); err != nil {

sandboxd/server/exec.go

@@ -93,7 +93,7 @@ func (s *Server) handleExec(w http.ResponseWriter, r *http.Request) {
 		held += int64(n)
 		return true
 	})
-	if err != nil && pid != 0 {
+	if pid != 0 && (ctx.Err() != nil || errors.Is(err, errExecOutputCap) || errors.Is(err, errExecBudget)) {
 		// A dropped connection only reaches a child that writes; a silent one needs the kill
 		if killErr := s.killExec(ctx, id, token, pid); killErr != nil {
 			log.WithFunc("server.handleExec").Errorf(ctx, killErr, "kill exec pid %d in %s", pid, id)