feat(pprocutils,wasm): WASM host-egress ABI + guest stub (http_request)

egress/doc.go

@@ -0,0 +1,61 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package egress is the processor-facing API for the WASM host-mediated
+// network-egress capability. A standalone (WebAssembly) processor has no
+// socket API of its own — Conduit performs the outbound HTTP call on the
+// processor's behalf, through a security boundary the processor cannot see
+// or influence. See docs/design-documents/20260726-wasm-host-egress-capability.md
+// in ConduitIO/conduit for the full design.
+//
+// [Do] is the only entry point, backed by the package-level [HTTPService]:
+//
+//   - Standalone (WebAssembly): the engine replaces [HTTPService] at startup
+//     with a client that forwards calls through the host's http_request
+//     capability.
+//   - Built-in processors / tests: the default [HTTPService] always returns
+//     [ErrEgressDisabled]. A built-in processor already runs with a real
+//     socket and should call net/http directly instead of this package;
+//     replace [HTTPService] with a stub or mock in tests.
+//
+// # Egress is deny-by-default and operator-gated
+//
+// A processor gets zero egress unless its operator explicitly opts it in
+// with a destination allowlist, optionally further clamped by an
+// engine-level ceiling. A processor cannot widen its own policy: a [Do] call
+// outside the resolved allowlist fails with [ErrForbidden], and a call from a
+// processor that was never opted in fails with [ErrEgressDisabled] — neither
+// is ever a silent pass-through.
+//
+// # Credentials are host-injected, never guest-supplied
+//
+// [Request.AuthSecretRef] names a secret; Conduit resolves it and sets the
+// Authorization header itself, immediately before dispatch. There is no
+// guest-supplied-credential path — a [Request] with an explicit Authorization
+// header is rejected with [ErrInvalidRequest] — so a credential can never be
+// read out of a processor's own memory or leaked by a memory-disclosure bug.
+//
+// # Redirects are not followed
+//
+// A 3xx response is returned to the caller as an ordinary [Response], never
+// followed automatically. A redirect to a non-allowlisted or private
+// Location therefore surfaces as a normal response, not a policy bypass.
+//
+// # Responses are buffered and size-capped
+//
+// The call is single-shot request/response, not streaming: [Do] blocks until
+// the full response is available. A response larger than the host-enforced
+// size cap is not delivered — [Do] returns [ErrResponseTooLarge] instead,
+// with no partial body.
+package egress

egress/egress.go

@@ -0,0 +1,145 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package egress
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/conduitio/conduit-processor-sdk/pprocutils"
+)
+
+// HTTPService is the service backing [Do]. In a standalone (WebAssembly)
+// processor the engine overwrites this at startup with a client that
+// forwards calls through the host's http_request capability. In any other
+// hosting mode (built-in processors, tests) it defaults to a stub that
+// always returns [ErrEgressDisabled]: a built-in processor already has a
+// real socket and should call net/http directly instead of this package.
+// Replace it in tests to stub egress behavior.
+var HTTPService pprocutils.HTTPService = disabledHTTPService{}
+
+// Sentinel errors mirror the ABI's numeric error-code band
+// (pprocutils.ErrorCodeHTTP*, see pprocutils/errors.go) so a caller can
+// classify a [Do] failure with [errors.Is] without importing pprocutils
+// directly.
+var (
+	// ErrEgressDisabled is returned when the processor was not opted into
+	// egress by its operator. Egress is deny-all by default: a pipeline
+	// author must explicitly allowlist a destination before a processor can
+	// reach it at all.
+	ErrEgressDisabled = pprocutils.ErrHTTPEgressDisabled
+	// ErrForbidden is returned when the destination host, scheme, or
+	// resolved IP is not permitted by the host-enforced allowlist. This
+	// also covers the DNS-rebinding defense: a hostname that passed the
+	// allowlist can still be refused here if it resolves to a
+	// private/reserved address at dial time.
+	ErrForbidden = pprocutils.ErrHTTPForbidden
+	// ErrInvalidRequest is returned for a malformed request: an unparsable
+	// URL, a disallowed scheme, header injection (CRLF), or a header the
+	// host reserves for itself (Host, Authorization, Accept-Encoding).
+	ErrInvalidRequest = pprocutils.ErrHTTPInvalidRequest
+	// ErrDNS is returned when host-side name resolution fails. Unlike
+	// [ErrForbidden], this is a transient condition a processor may choose
+	// to retry.
+	ErrDNS = pprocutils.ErrHTTPDNS
+	// ErrTimeout is returned when the host-enforced per-call deadline is
+	// exceeded. The processor cannot extend this deadline.
+	ErrTimeout = pprocutils.ErrHTTPTimeout
+	// ErrResponseTooLarge is returned when the response body — after
+	// decompression, if any — exceeds the host-enforced size cap. No
+	// partial body is returned.
+	ErrResponseTooLarge = pprocutils.ErrHTTPResponseTooLarge
+	// ErrTransport is returned for a connection-level failure: reset, TLS
+	// handshake failure, and similar.
+	ErrTransport = pprocutils.ErrHTTPTransport
+)
+
+// Request is a single outbound HTTP call a standalone processor asks Conduit
+// to perform on its behalf. The guest never gets a socket: Conduit validates
+// the request against the processor's host-configured egress policy
+// (allowlist, resolved-IP dial-time gate) and performs the I/O itself with a
+// hardened net/http client. The call is buffered, not streaming: [Do] blocks
+// until the full (capped) response is available.
+type Request struct {
+	// Method is the HTTP method, e.g. "GET" or "POST".
+	Method string
+	// URL is the target URL, including scheme. Only https is permitted
+	// unless the exact (host, port) pair is explicitly allowlisted for http
+	// (the local-Ollama case).
+	URL string
+	// Headers are sent as-is, except for a small host-reserved set the
+	// processor cannot set: Host, Authorization, and Accept-Encoding.
+	// Setting any of them returns [ErrInvalidRequest].
+	Headers map[string][]string
+	// Body is the request body, if any.
+	Body []byte
+	// AuthSecretRef names a secret Conduit resolves and injects as the
+	// Authorization header, immediately before the request is dispatched.
+	// The credential value never enters the processor's memory — there is
+	// no guest-supplied-credential path, by design. Leave empty for an
+	// unauthenticated request.
+	AuthSecretRef string
+}
+
+// Response is the fully-buffered, host-size-capped response to a [Request].
+type Response struct {
+	StatusCode int
+	Headers    map[string][]string
+	// Body is the full response body, up to the host-enforced size cap. A
+	// response that exceeds the cap is never delivered: [Do] returns
+	// [ErrResponseTooLarge] instead, with no partial body.
+	Body []byte
+}
+
+// Do performs a host-mediated outbound HTTP call. It is the processor-facing
+// entry point for the WASM host-egress capability; [HTTPService] does the
+// actual work, so a processor should call Do rather than [HTTPService]
+// directly.
+//
+// Do returns a wrapped error for every host-side rejection or failure; test
+// against the sentinel errors in this package with [errors.Is] to classify
+// the outcome — for example, to distinguish a policy rejection
+// ([ErrForbidden], [ErrEgressDisabled]) from a transient one ([ErrTimeout],
+// [ErrDNS], [ErrTransport]) for a retry/DLQ decision.
+func Do(ctx context.Context, req Request) (Response, error) {
+	resp, err := HTTPService.Do(ctx, pprocutils.HTTPRequest{
+		Method:        req.Method,
+		URL:           req.URL,
+		Headers:       req.Headers,
+		Body:          req.Body,
+		AuthSecretRef: req.AuthSecretRef,
+	})
+	if err != nil {
+		return Response{}, fmt.Errorf("error performing http egress request: %w", err)
+	}
+
+	return Response{
+		StatusCode: resp.StatusCode,
+		Headers:    resp.Headers,
+		Body:       resp.Body,
+	}, nil
+}
+
+// disabledHTTPService is the default, non-WASM [HTTPService]. Outside a
+// standalone processor there is no host boundary to broker the call
+// through, so every request is refused with [ErrEgressDisabled] rather than
+// silently falling back to an unpolicied net/http call — the security
+// boundary this capability exists to enforce has no meaning without the host
+// mediating it.
+type disabledHTTPService struct{}
+
+func (disabledHTTPService) Do(context.Context, pprocutils.HTTPRequest) (pprocutils.HTTPResponse, error) {
+	return pprocutils.HTTPResponse{}, ErrEgressDisabled
+}

egress/egress_test.go

@@ -0,0 +1,115 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package egress
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/conduitio/conduit-processor-sdk/pprocutils"
+	"github.com/conduitio/conduit-processor-sdk/pprocutils/mock"
+	"github.com/matryer/is"
+	"go.uber.org/mock/gomock"
+)
+
+// withHTTPService swaps the package-level HTTPService for the duration of a
+// test and restores the deny-all default afterwards, so tests never leak
+// state into one another regardless of execution order.
+func withHTTPService(t *testing.T, svc pprocutils.HTTPService) {
+	t.Helper()
+	HTTPService = svc
+	t.Cleanup(func() { HTTPService = disabledHTTPService{} })
+}
+
+func TestDo_ConvertsRequestAndResponse(t *testing.T) {
+	is := is.New(t)
+	ctx := context.Background()
+	ctrl := gomock.NewController(t)
+	svc := mock.NewHTTPService(ctrl)
+	withHTTPService(t, svc)
+
+	req := Request{
+		Method:        "POST",
+		URL:           "https://api.example.com/v1/embeddings",
+		Headers:       map[string][]string{"Content-Type": {"application/json"}},
+		Body:          []byte(`{"input":"hello"}`),
+		AuthSecretRef: "openai_api_key",
+	}
+
+	svc.EXPECT().Do(ctx, pprocutils.HTTPRequest{
+		Method:        req.Method,
+		URL:           req.URL,
+		Headers:       req.Headers,
+		Body:          req.Body,
+		AuthSecretRef: req.AuthSecretRef,
+	}).Return(pprocutils.HTTPResponse{
+		StatusCode: 200,
+		Headers:    map[string][]string{"Content-Type": {"application/json"}},
+		Body:       []byte(`{"result":"ok"}`),
+	}, nil)
+
+	resp, err := Do(ctx, req)
+	is.NoErr(err)
+	is.Equal(resp, Response{
+		StatusCode: 200,
+		Headers:    map[string][]string{"Content-Type": {"application/json"}},
+		Body:       []byte(`{"result":"ok"}`),
+	})
+}
+
+func TestDo_WrapsSentinelErrors(t *testing.T) {
+	tests := []struct {
+		name string
+		code uint32
+		want error
+	}{
+		{"egress disabled", pprocutils.ErrorCodeHTTPEgressDisabled, ErrEgressDisabled},
+		{"forbidden", pprocutils.ErrorCodeHTTPForbidden, ErrForbidden},
+		{"invalid request", pprocutils.ErrorCodeHTTPInvalidRequest, ErrInvalidRequest},
+		{"dns", pprocutils.ErrorCodeHTTPDNS, ErrDNS},
+		{"timeout", pprocutils.ErrorCodeHTTPTimeout, ErrTimeout},
+		{"response too large", pprocutils.ErrorCodeHTTPResponseTooLarge, ErrResponseTooLarge},
+		{"transport", pprocutils.ErrorCodeHTTPTransport, ErrTransport},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			is := is.New(t)
+			ctx := context.Background()
+			ctrl := gomock.NewController(t)
+			svc := mock.NewHTTPService(ctrl)
+			withHTTPService(t, svc)
+
+			svc.EXPECT().Do(ctx, gomock.Any()).
+				Return(pprocutils.HTTPResponse{}, pprocutils.NewErrorFromCode(tt.code))
+
+			_, err := Do(ctx, Request{Method: "GET", URL: "https://example.com"})
+			is.True(errors.Is(err, tt.want))
+		})
+	}
+}
+
+func TestDo_DefaultServiceDeniesAll(t *testing.T) {
+	is := is.New(t)
+	ctx := context.Background()
+	// No withHTTPService call: exercises the package's actual zero-config
+	// default, which every hosting mode other than a standalone (WASM)
+	// processor gets unless it opts in with its own HTTPService.
+	withHTTPService(t, disabledHTTPService{})
+
+	_, err := Do(ctx, Request{Method: "GET", URL: "https://example.com"})
+	is.True(errors.Is(err, ErrEgressDisabled))
+}

egress/example_test.go

@@ -0,0 +1,36 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package egress_test
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/conduitio/conduit-processor-sdk/egress"
+)
+
+// ExampleDo demonstrates that, outside a standalone (WebAssembly) processor,
+// [egress.Do] refuses every call. There is no host boundary to broker the
+// request through in this hosting mode, so the package defaults to deny-all
+// rather than silently falling back to an unpolicied network call.
+func ExampleDo() {
+	_, err := egress.Do(context.Background(), egress.Request{
+		Method: "GET",
+		URL:    "https://api.example.com/v1/models",
+	})
+	fmt.Println(err)
+	// Output:
+	// error performing http egress request: http egress is not enabled for this processor
+}