fix(relay): give cic-flow actuate the egress limits of the Rust http-executor

cmd/relay/cic_flow_host_funcs.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/base64"
+	"errors"
 	"io"
 	"net/http"
 	"net/url"
@@ -178,13 +179,24 @@ func makeActuateFn(deps cicFlowDeps) func(context.Context, api.Module, uint32, u
 		}
 		resp, err := deps.doer.Do(httpReq)
 		if err != nil {
+			if errors.Is(err, errEgressBlockedAddr) {
+				return egressWriteError(mod, outPtr, outLen, "blocked_address", err.Error())
+			}
 			return gitWriteError(mod, outPtr, outLen, err.Error())
 		}
 		defer func() { _ = resp.Body.Close() }()
-		respBody, err := io.ReadAll(resp.Body)
+		// A response over the cap is a typed error and is never held in memory: the declared
+		// length is checked first, and the read stops one byte past the cap.
+		if resp.ContentLength > egressMaxResponseBytes {
+			return egressWriteError(mod, outPtr, outLen, "response_too_large", egressTooLargeMsg)
+		}
+		respBody, err := io.ReadAll(io.LimitReader(resp.Body, egressMaxResponseBytes+1))
 		if err != nil {
 			return gitWriteError(mod, outPtr, outLen, err.Error())
 		}
+		if len(respBody) > egressMaxResponseBytes {
+			return egressWriteError(mod, outPtr, outLen, "response_too_large", egressTooLargeMsg)
+		}
 		respHeaders := make(map[string]string, len(resp.Header))
 		for k := range resp.Header {
 			respHeaders[strings.ToLower(k)] = resp.Header.Get(k)
@@ -197,6 +209,15 @@ func makeActuateFn(deps cicFlowDeps) func(context.Context, api.Module, uint32, u
 	}
 }
 
+// egressTooLargeMsg is the message of the response_too_large error.
+var egressTooLargeMsg = "egress: response exceeds the " + strconv.Itoa(egressMaxResponseBytes) + "-byte limit"
+
+// egressWriteError writes a typed egress error: {"error": <msg>, "code": <code>}. The "error"
+// key is the one every guest already reads; "code" is additive.
+func egressWriteError(mod api.Module, outPtr, outLen uint32, code, msg string) int32 {
+	return gitWriteJSON(mod, outPtr, outLen, map[string]string{"error": msg, "code": code})
+}
+
 // portOf returns the request's TCP port: the explicit URL port, else the scheme
 // default (443 for https, 80 for http).
 func portOf(u *url.URL) int {

cmd/relay/cic_flow_registry.go

@@ -1,9 +1,6 @@
 package main
 
-import (
-	"net/http"
-	"sync"
-)
+import "sync"
 
 // R3a live wiring (feature-011, r3-plan.md). The `cic-flow` host module is a
 // single shared instance on the wazero runtime, but capabilities are per-module:
@@ -30,12 +27,13 @@ type cicFlowRegistry struct {
 // module at load (fail-closed). It is set once at startup before any load.
 var registerModuleCapability func(moduleID string, projectYAML []byte) error
 
-// newCICFlowRegistry builds an empty registry. A nil doer defaults to
-// http.DefaultClient; a nil resolve leaves every module's signer nil (sign inert)
-// until the OCI-key provisioning path wires a real resolver.
+// newCICFlowRegistry builds an empty registry. A nil doer defaults to the egress client
+// (newEgressClient: no redirects, no proxy, timeouts, blocked address ranges); a nil
+// resolve leaves every module's signer nil (sign inert) until the OCI-key provisioning
+// path wires a real resolver.
 func newCICFlowRegistry(resolve keyResolver, doer httpDoer) *cicFlowRegistry {
 	if doer == nil {
-		doer = http.DefaultClient
+		doer = newEgressClient(false)
 	}
 	return &cicFlowRegistry{resolve: resolve, doer: doer, deps: map[string]cicFlowDeps{}}
 }

cmd/relay/cic_flow_registry_test.go

@@ -11,9 +11,16 @@ import (
 
 func TestNewCICFlowRegistry_DefaultsDoer(t *testing.T) {
 	reg := newCICFlowRegistry(nil, nil)
-	assert.Equal(t, http.DefaultClient, reg.doer)
-	_, ok := reg.lookup("anything")
-	assert.False(t, ok) // empty → deny
+	// The default doer is the egress client, not http.DefaultClient (which follows
+	// redirects, honours the proxy environment and has no timeout).
+	assert.NotSame(t, http.DefaultClient, reg.doer)
+	c, ok := reg.doer.(*http.Client)
+	require.True(t, ok, "the default doer must be the egress *http.Client")
+	assert.Equal(t, egressRequestTimeout, c.Timeout)
+	require.NotNil(t, c.CheckRedirect)
+	assert.ErrorIs(t, c.CheckRedirect(nil, nil), http.ErrUseLastResponse)
+	_, found := reg.lookup("anything")
+	assert.False(t, found) // empty → deny
 }
 
 func TestCICFlowRegistry_RegisterVerified_Binds(t *testing.T) {

cmd/relay/egress_client.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"errors"
+	"net"
+	"net/http"
+	"syscall"
+	"time"
+)
+
+// The egress limits of the `actuate` host function. They mirror the Rust
+// http-executor (http-executor/src/lib.rs, Limits::default and EgressPolicy); the shared
+// fixture testdata/egress-parity/policy.txt is asserted on both sides, so changing one
+// without the other fails a test.
+const (
+	egressRequestTimeout   = 30 * time.Second
+	egressConnectTimeout   = 10 * time.Second
+	egressMaxResponseBytes = 1 << 20 // 1 MiB
+)
+
+// errEgressBlockedAddr is returned when a connection would reach an address an actuation
+// must never reach.
+var errEgressBlockedAddr = errors.New("egress: destination address is blocked (private, loopback, link-local, unspecified or broadcast)")
+
+// isBlockedIP reports whether ip is in a private, loopback, link-local, unspecified or
+// broadcast range (SSRF defense). IPv4-mapped IPv6 addresses are judged as IPv4. The
+// ranges are exactly those of the Rust http-executor's is_blocked_ip.
+func isBlockedIP(ip net.IP) bool {
+	if v4 := ip.To4(); v4 != nil {
+		return v4.IsLoopback() || v4.IsPrivate() || v4.IsLinkLocalUnicast() ||
+			v4.IsUnspecified() || v4.Equal(net.IPv4bcast)
+	}
+	return ip.IsLoopback() || ip.IsUnspecified() || ip.IsPrivate() || ip.IsLinkLocalUnicast()
+}
+
+// newEgressClient builds the HTTP client of the `actuate` host function:
+//   - a redirect is never followed (the 3xx is returned to the caller), so the egress
+//     allow-list, which only sees the first URL, cannot be bypassed by a redirect;
+//   - ambient proxy environment is ignored;
+//   - connect and end-to-end timeouts are set;
+//   - every connection is checked at dial time, after DNS resolution, against the blocked
+//     address ranges (this also covers a DNS answer that points at a private address).
+//
+// allowPrivate lifts the address check; it exists only for tests against a local server.
+func newEgressClient(allowPrivate bool) *http.Client {
+	dialer := &net.Dialer{
+		Timeout: egressConnectTimeout,
+		Control: func(_, address string, _ syscall.RawConn) error {
+			host, _, err := net.SplitHostPort(address)
+			if err != nil {
+				return errEgressBlockedAddr // an address we cannot read is not dialed
+			}
+			if ip := net.ParseIP(host); ip == nil || (!allowPrivate && isBlockedIP(ip)) {
+				return errEgressBlockedAddr
+			}
+			return nil
+		},
+	}
+	return &http.Client{
+		Timeout: egressRequestTimeout,
+		Transport: &http.Transport{
+			Proxy:               nil, // never ambient HTTP(S)_PROXY
+			DialContext:         dialer.DialContext,
+			TLSHandshakeTimeout: egressConnectTimeout,
+			ForceAttemptHTTP2:   true,
+		},
+		CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse },
+	}
+}

cmd/relay/egress_client_test.go

@@ -0,0 +1,217 @@
+package main
+
+import (
+	"bufio"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"os"
+	"strconv"
+	"strings"
+	"sync/atomic"
+	"testing"
+
+	"context"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestEgressParity_Fixture asserts the shared policy fixture, which the Rust http-executor
+// asserts too (http-executor/src/lib.rs, egress_parity_fixture_matches_policy).
+func TestEgressParity_Fixture(t *testing.T) {
+	f, err := os.Open("../../testdata/egress-parity/policy.txt")
+	require.NoError(t, err)
+	defer f.Close()
+
+	consts := map[string]int64{
+		"request_timeout_secs": int64(egressRequestTimeout.Seconds()),
+		"connect_timeout_secs": int64(egressConnectTimeout.Seconds()),
+		"max_response_bytes":   egressMaxResponseBytes,
+	}
+	seen := map[string]bool{}
+	ips := 0
+	sc := bufio.NewScanner(f)
+	for line := 1; sc.Scan(); line++ {
+		text := strings.TrimSpace(sc.Text())
+		if text == "" || strings.HasPrefix(text, "#") {
+			continue
+		}
+		fields := strings.Fields(text)
+		switch {
+		case len(fields) == 3 && fields[0] == "const":
+			want, perr := strconv.ParseInt(fields[2], 10, 64)
+			require.NoError(t, perr, "line %d", line)
+			got, known := consts[fields[1]]
+			require.True(t, known, "line %d: unknown const %q", line, fields[1])
+			assert.Equal(t, want, got, "const %s", fields[1])
+			seen[fields[1]] = true
+		case len(fields) == 3 && fields[0] == "ip":
+			ip := net.ParseIP(fields[2])
+			require.NotNil(t, ip, "line %d: bad address %q", line, fields[2])
+			assert.Equal(t, fields[1] == "blocked", isBlockedIP(ip), "address %s", fields[2])
+			ips++
+		default:
+			t.Fatalf("line %d: cannot parse %q", line, text)
+		}
+	}
+	require.NoError(t, sc.Err())
+	for name := range consts {
+		assert.True(t, seen[name], "fixture lacks const %s", name)
+	}
+	assert.Greater(t, ips, 20, "the fixture must keep its address cases")
+}
+
+// callActuate runs makeActuateFn against policy/doer and returns the decoded output.
+func callActuate(t *testing.T, doer httpDoer, policy egressPolicy, target string) map[string]interface{} {
+	t.Helper()
+	mod := newMod(16384)
+	_, out := invoke(makeActuateFn(cicFlowDeps{doer: doer, egress: policy}), mod, map[string]interface{}{
+		"method": "GET", "url": target,
+	})
+	return out
+}
+
+func allowLocal(t *testing.T, srv *httptest.Server) egressPolicy {
+	t.Helper()
+	u, err := url.Parse(srv.URL)
+	require.NoError(t, err)
+	port, err := strconv.Atoi(u.Port())
+	require.NoError(t, err)
+	return egressPolicy{{Host: u.Hostname(), Ports: []int{port}}}
+}
+
+// Acceptance 1: an allowed URL that answers 302 to a non-allowed host (loopback and
+// link-local included) must not be followed.
+func TestActuate_RedirectToNonAllowedHost_NotFollowed(t *testing.T) {
+	var hits atomic.Int32
+	other := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		hits.Add(1)
+		_, _ = w.Write([]byte("must not be reached"))
+	}))
+	defer other.Close()
+
+	for name, location := range map[string]string{
+		"other loopback port": other.URL + "/secret",
+		"link-local metadata": "http://169.254.169.254/latest/meta-data/",
+		"other host":          "https://evil.example/",
+	} {
+		t.Run(name, func(t *testing.T) {
+			allowed := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				http.Redirect(w, r, location, http.StatusFound)
+			}))
+			defer allowed.Close()
+
+			// allowPrivate: the test servers are on loopback; redirect handling is what is tested.
+			out := callActuate(t, newEgressClient(true), allowLocal(t, allowed), allowed.URL+"/start")
+
+			assert.EqualValues(t, 0, hits.Load(), "the redirect target must never be contacted")
+			require.Equal(t, float64(http.StatusFound), out["status"], "the 302 is returned to the caller, not followed (output: %v)", out)
+			hdrs, ok := out["headers"].(map[string]interface{})
+			require.True(t, ok, "the response headers must be returned: %v", out)
+			assert.Equal(t, location, hdrs["location"])
+		})
+	}
+}
+
+// Acceptance 2: a response over the cap is a typed error and is never held in memory.
+func TestActuate_ResponseOverCap_TypedError(t *testing.T) {
+	t.Run("chunked, no declared length", func(t *testing.T) {
+		var written atomic.Int64
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			chunk := make([]byte, 64<<10)
+			for i := 0; i < 1024; i++ { // up to 64 MiB offered
+				n, err := w.Write(chunk)
+				written.Add(int64(n))
+				if err != nil {
+					return
+				}
+				w.(http.Flusher).Flush()
+			}
+		}))
+		defer srv.Close()
+
+		out := callActuate(t, newEgressClient(true), allowLocal(t, srv), srv.URL+"/big")
+		assert.Equal(t, "response_too_large", out["code"])
+		assert.Contains(t, out["error"], "byte limit")
+		assert.Less(t, written.Load(), int64(16<<20), "the reader must stop near the cap, not drain the body")
+	})
+	t.Run("declared length over the cap", func(t *testing.T) {
+		srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Length", strconv.Itoa(egressMaxResponseBytes+1))
+			w.WriteHeader(http.StatusOK)
+		}))
+		defer srv.Close()
+		out := callActuate(t, newEgressClient(true), allowLocal(t, srv), srv.URL+"/declared")
+		assert.Equal(t, "response_too_large", out["code"])
+	})
+	t.Run("one byte over, injected doer", func(t *testing.T) {
+		doer := &stubDoer{resp: okResponse(200, strings.Repeat("x", egressMaxResponseBytes+1), nil)}
+		out := callActuate(t, doer, egressPolicy{{Host: "api.example.com"}}, "https://api.example.com/x")
+		assert.Equal(t, "response_too_large", out["code"], "the cap applies whichever doer is injected")
+	})
+}
+
+// A body exactly at the cap is accepted: the limit is on bytes over, not at.
+func TestActuate_ResponseAtCap_NotRefused(t *testing.T) {
+	body := strings.Repeat("x", egressMaxResponseBytes)
+	doer := &stubDoer{resp: okResponse(200, body, nil)}
+	mod := newMod(2 << 20)
+	reqLen := putJSON(mod, 0, map[string]interface{}{"method": "GET", "url": "https://api.example.com/x"})
+	n := makeActuateFn(cicFlowDeps{doer: doer, egress: egressPolicy{{Host: "api.example.com"}}})(
+		context.Background(), mod, 0, reqLen, 4096, uint32(len(mod.mem.buf))-4096)
+	require.Positive(t, n)
+	out := map[string]interface{}{}
+	require.NoError(t, json.Unmarshal(mod.mem.buf[4096:4096+uint32(n)], &out))
+	require.Nil(t, out["error"], "a body of exactly the cap must pass: %v", out["error"])
+	encoded, ok := out["body_base64"].(string)
+	require.True(t, ok, "the body must be returned: %v", out)
+	raw, err := base64.StdEncoding.DecodeString(encoded)
+	require.NoError(t, err)
+	assert.Len(t, raw, egressMaxResponseBytes)
+}
+
+// The default client must refuse to dial blocked addresses (SSRF guard, as in the Rust executor).
+func TestEgressClient_DialGuard_BlocksLoopback(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte("reached"))
+	}))
+	defer srv.Close()
+
+	_, err := newEgressClient(false).Get(srv.URL)
+	require.Error(t, err)
+	assert.True(t, errors.Is(err, errEgressBlockedAddr), "got %v", err)
+
+	reg := newCICFlowRegistry(nil, nil)
+	out := callActuate(t, reg.doer, allowLocal(t, srv), srv.URL)
+	assert.Equal(t, "blocked_address", out["code"])
+}
+
+// Ambient proxy environment must be ignored. (Go never proxies loopback destinations, so a
+// request-level check against a local server could not fail; the Transport itself is asserted.)
+func TestEgressClient_IgnoresProxyEnvironment(t *testing.T) {
+	t.Setenv("HTTP_PROXY", "http://proxy.invalid:3128")
+	t.Setenv("HTTPS_PROXY", "http://proxy.invalid:3128")
+
+	for _, allowPrivate := range []bool{false, true} {
+		tr, ok := newEgressClient(allowPrivate).Transport.(*http.Transport)
+		require.True(t, ok)
+		assert.Nil(t, tr.Proxy, "no proxy function: the environment is never consulted (allowPrivate=%v)", allowPrivate)
+	}
+	// The contrast: the default transport does consult the environment.
+	assert.NotNil(t, http.DefaultTransport.(*http.Transport).Proxy)
+}
+
+// The dial guard refuses an address it cannot parse instead of letting it through.
+func TestEgressClient_DialGuard_MalformedAddressNotDialed(t *testing.T) {
+	tr, ok := newEgressClient(false).Transport.(*http.Transport)
+	require.True(t, ok)
+	_, err := tr.DialContext(context.Background(), "tcp", "not-an-ip-literal:80")
+	require.Error(t, err)
+	assert.NotContains(t, fmt.Sprint(err), "reached")
+}