diff --git a/docs/egress.md b/docs/egress.md index ad0d8ca0..265573a7 100644 --- a/docs/egress.md +++ b/docs/egress.md @@ -41,7 +41,10 @@ use the SOCKS5 listener silkd binds at `127.0.0.1:1080` and relays over second door: CONNECT only (BIND and UDP ASSOCIATE answer *command not supported*), no authentication (the socket path already carries the sandbox's identity), and a `DOMAINNAME` destination is resolved host-side, so the guest -still needs no resolver. +still needs no resolver. Each door serves at most 256 connections per sandbox +at a time; later ones sit unserved in the socket backlog until one closes, and +the proxy keeps at most 64 idle upstream connections per sandbox, so a guest +cannot turn its proxy into a host descriptor sink. A SOCKS5 tunnel takes the decision an HTTP `CONNECT` to the same host takes, through the same code: a rule with a nonempty `methods` list that omits diff --git a/sandboxd/egress/proxy.go b/sandboxd/egress/proxy.go index 5da3ea2e..1b2e5177 100644 --- a/sandboxd/egress/proxy.go +++ b/sandboxd/egress/proxy.go @@ -19,7 +19,11 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/utils" ) -const idleConnTimeout = 90 * time.Second +const ( + idleConnTimeout = 90 * time.Second + // maxIdleConns bounds the upstream pool per sandbox; a guest walking many hosts cannot park a descriptor per host. + maxIdleConns = 64 +) // hopHeaders are hop-by-hop and proxy-scoped headers this hop owns, never passed on. var hopHeaders = []string{ @@ -90,7 +94,7 @@ func New(sandbox, tenant string, policy Evaluator, secrets Secrets, ca *CA, dial dial: dial, holder: holder, // the stdlib default of 2 idle conns per host re-dials bursty same-host traffic. - tr: &http.Transport{DialContext: dial, MaxIdleConnsPerHost: 8, IdleConnTimeout: idleConnTimeout}, + tr: &http.Transport{DialContext: dial, MaxIdleConns: maxIdleConns, MaxIdleConnsPerHost: 8, IdleConnTimeout: idleConnTimeout}, conns: map[net.Conn]struct{}{}, } if ca != nil { diff --git a/sandboxd/pool/egress.go b/sandboxd/pool/egress.go index 7668416e..0c3daf8d 100644 --- a/sandboxd/pool/egress.go +++ b/sandboxd/pool/egress.go @@ -13,6 +13,7 @@ import ( "time" "github.com/projecteru2/core/log" + "golang.org/x/net/netutil" "github.com/cocoonstack/sandbox/sandboxd/egress" "github.com/cocoonstack/sandbox/sandboxd/engine" @@ -20,6 +21,9 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/types" ) +// egressDoorConns bounds one sandbox's open connections per door; the guest's next dial waits in the backlog. +const egressDoorConns = 256 + var ( nat64Range = netip.MustParsePrefix("64:ff9b::/96") // RFC 6052 NAT64; the embedded v4 is checked instead @@ -176,6 +180,10 @@ func (m *Manager) armEgressProxy(ctx context.Context, sb *types.Sandbox) error { } el.srv = &http.Server{Handler: proxy, ReadHeaderTimeout: 30 * time.Second} el.proxy = proxy + el.ln = netutil.LimitListener(el.ln, egressDoorConns) + if el.socks != nil { + el.socks = netutil.LimitListener(el.socks, egressDoorConns) + } m.mu.Lock() displaced := m.egressListeners[id] m.egressListeners[id] = el diff --git a/sandboxd/pool/egress_test.go b/sandboxd/pool/egress_test.go index 079a20ec..aefdb01c 100644 --- a/sandboxd/pool/egress_test.go +++ b/sandboxd/pool/egress_test.go @@ -12,6 +12,7 @@ import ( "path/filepath" "slices" "strings" + "syscall" "testing" "time" @@ -657,6 +658,58 @@ func TestTrimmedWarmVMClosesItsDoors(t *testing.T) { } } +func TestEgressDoorCapsConcurrentConnections(t *testing.T) { + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = io.WriteString(w, "ok") + })) + t.Cleanup(origin.Close) + pol := &egress.Policy{Allow: []egress.Rule{{Host: mustHostname(t, origin.URL)}}} + m := egressManager(t, newFakeEngine(), config.PoolSpec{PoolKey: testKey, Warm: 1, Egress: pol}) + m.dial = (&net.Dialer{}).DialContext + sb := vsockSandbox(t, "sb_cap") + if err := m.armEgress(t.Context(), sb); err != nil { + t.Fatalf("arm egress: %v", err) + } + path := engine.EgressSocketPath(sb.VsockSocket) + + idle := make([]net.Conn, 0, egressDoorConns) + for len(idle) < egressDoorConns { + conn, err := net.Dial("unix", path) + if errors.Is(err, syscall.ECONNREFUSED) { + time.Sleep(10 * time.Millisecond) + continue + } + if err != nil { + t.Fatalf("idle dial %d: %v", len(idle), err) + } + idle = append(idle, conn) + } + t.Cleanup(func() { + for _, conn := range idle { + _ = conn.Close() + } + }) + + client := egressClient(path) + client.Timeout = 500 * time.Millisecond + if resp, err := client.Get(origin.URL + "/"); err == nil { + resp.Body.Close() + t.Fatal("a request past the door cap was served") + } + + _ = idle[0].Close() + client.Timeout = 5 * time.Second + resp, err := client.Get(origin.URL + "/") + if err != nil { + t.Fatalf("request after a slot freed: %v", err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("status %d after a slot freed, want 200", resp.StatusCode) + } + m.disarmEgress(sb.ID, true) +} + func dialDoors(t *testing.T, sb *types.Sandbox) { t.Helper() for _, path := range []string{engine.EgressSocketPath(sb.VsockSocket), engine.SocksSocketPath(sb.VsockSocket)} { diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 41e3961f..941348c7 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -180,6 +180,8 @@ type pool struct { rate float64 lead time.Duration lastArrival time.Time + binStart time.Time + binCount int goldenDir string building bool diff --git a/sandboxd/pool/watermark.go b/sandboxd/pool/watermark.go index c953d028..13c40ecc 100644 --- a/sandboxd/pool/watermark.go +++ b/sandboxd/pool/watermark.go @@ -8,19 +8,24 @@ import ( const ( // ewmaAlpha weights the newest observation in the arrival-rate and provision-lead averages ewmaAlpha = 0.3 + rateBin = time.Second // rateDecayTau halves the observed arrival rate roughly every 40s of silence rateDecayTau = 60 * time.Second // leadSafety over-provisions against the measured lead: arrivals are bursty, not uniform leadSafety = 2.0 ) -// noteArrival folds one claim into the arrival-rate EWMA. Caller holds the manager mutex. +// noteArrival counts one claim; a bin that has spanned rateBin folds its arrivals per second into the rate EWMA. Caller holds the manager mutex. func (p *pool) noteArrival(now time.Time) { - if !p.lastArrival.IsZero() { - if dt := now.Sub(p.lastArrival).Seconds(); dt > 0 { - p.rate = ewmaAlpha*(1/dt) + (1-ewmaAlpha)*p.rate - } + switch elapsed := now.Sub(p.binStart); { + case p.binStart.IsZero(): + p.binStart = now + case elapsed >= rateBin: + decayed := p.rate * math.Exp(-elapsed.Seconds()/rateDecayTau.Seconds()) + p.rate = ewmaAlpha*(float64(p.binCount)/elapsed.Seconds()) + (1-ewmaAlpha)*decayed + p.binStart, p.binCount = now, 0 } + p.binCount++ p.lastArrival = now } diff --git a/sandboxd/pool/watermark_test.go b/sandboxd/pool/watermark_test.go index 2ba4201e..2dddf083 100644 --- a/sandboxd/pool/watermark_test.go +++ b/sandboxd/pool/watermark_test.go @@ -15,19 +15,53 @@ func TestEffectiveTargetTracksDemand(t *testing.T) { t.Fatalf("quiet pool target %d, want the floor", got) } - for i := range 20 { + for i := range 60 { p.noteArrival(now.Add(time.Duration(i) * 100 * time.Millisecond)) } - burstEnd := now.Add(2 * time.Second) - if got := p.effectiveTarget(burstEnd); got != 8 { - t.Errorf("burst target %d, want warmMax 8", got) + sustainedEnd := now.Add(6 * time.Second) + if got := p.effectiveTarget(sustainedEnd); got != 8 { + t.Errorf("sustained 10/s target %d, want warmMax 8", got) } - if got := p.effectiveTarget(burstEnd.Add(5 * time.Minute)); got != 2 { + if got := p.effectiveTarget(sustainedEnd.Add(5 * time.Minute)); got != 2 { t.Errorf("post-silence target %d, want the floor", got) } } +func TestEffectiveTargetIgnoresATwoClaimBurst(t *testing.T) { + now := time.Now() + p := &pool{key: types.PoolKey{}, floor: 2, warmMax: 20, lead: 500 * time.Millisecond} + + p.noteArrival(now) + p.noteArrival(now.Add(time.Millisecond)) + if got := p.effectiveTarget(now.Add(time.Second)); got != 2 { + t.Errorf("target %d after two claims 1 ms apart, want the floor", got) + } + + p.noteArrival(now.Add(2 * time.Second)) + if got := p.effectiveTarget(now.Add(2 * time.Second)); got != 2 { + t.Errorf("target %d after the burst's bin closed at 1/s, want the floor", got) + } +} + +func TestEffectiveTargetForgetsRateAcrossSilence(t *testing.T) { + now := time.Now() + p := &pool{key: types.PoolKey{}, floor: 2, warmMax: 8, lead: 500 * time.Millisecond} + for i := range 60 { + p.noteArrival(now.Add(time.Duration(i) * 100 * time.Millisecond)) + } + if got := p.effectiveTarget(now.Add(6 * time.Second)); got != 8 { + t.Fatalf("sustained target %d, want warmMax 8", got) + } + + later := now.Add(10 * time.Minute) + p.noteArrival(later) + p.noteArrival(later.Add(time.Second)) + if got := p.effectiveTarget(later.Add(time.Second)); got != 2 { + t.Errorf("target %d after two claims following ten minutes of silence, want the floor", got) + } +} + func TestEffectiveTargetOffWithoutWarmMax(t *testing.T) { now := time.Now() p := &pool{floor: 1, lead: time.Second} diff --git a/sdk/python/tests/test_frames.py b/sdk/python/tests/test_frames.py index ed40136a..9a995bc6 100644 --- a/sdk/python/tests/test_frames.py +++ b/sdk/python/tests/test_frames.py @@ -21,14 +21,35 @@ def test_nested_data_field_not_shadowed(): assert frame["meta"] == {"data": "WFhY"} +def test_tag_after_other_keys_takes_the_full_parse(): + frame = frames.decode_response(b'{"data":"aGk=","type":"stdout"}') + assert frame == {"data": b"hi", "type": "stdout"} + + @pytest.mark.parametrize( "raw", [ b'{"type":"stdout","data":"aGk="}garbage', b'{"type":"data","data":"QUJD\r\nREVG"}', b'{"type":"stdout","data":"aGk="', + b'{"type":"stdout"}garbage"data":"QQ=="}', + b'{"type":"stdout"garbage,"data":"QQ=="}', + b'{"type":"stdout":,"data":"QQ=="}', + b'{"type":"stdout",garbage"data":"QQ=="}', + b'{"type":"stdout","data":"QQ==" }x', + b'{"type":"stdout","data":"Q\nQ=="}', + ], + ids=[ + "trailing_bytes", + "control_bytes_in_base64", + "unterminated_frame", + "closed_before_data", + "garbage_after_tag", + "colon_after_tag", + "garbage_before_data", + "space_before_close", + "newline_in_base64", ], - ids=["trailing_bytes", "control_bytes_in_base64", "unterminated_frame"], ) def test_malformed_frame_rejected(raw): with pytest.raises(json.JSONDecodeError):