From bed02257272c8105a9d747a81974eab2cd46e282 Mon Sep 17 00:00:00 2001 From: DeVaris Brown Date: Sun, 26 Jul 2026 16:08:57 -0700 Subject: [PATCH 1/2] feat(pprocutils): add HTTPService egress ABI (HTTPRequest/HTTPResponse, error codes, converters) Shared host/guest types for the WASM host-mediated network-egress capability (host function http_request), per docs/design-documents/20260726-wasm-host-egress-capability.md in ConduitIO/conduit. Adds: - proto/procutils/v1/http.proto + generated http.pb.go (HTTPHeader/HTTPRequest/HTTPResponse) - pprocutils.HTTPService interface + HTTPRequest/HTTPResponse Go structs (AuthSecretRef reserved) - pprocutils ABI error-code band entries (ErrorCodeHTTP*) - fromproto/toproto converters + generated HTTPService mock Guest-side wasm stub + embedding-processor consumer are deferred to later slices. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_015GQFzakPShAYj8CcwajYDD --- pprocutils/errors.go | 33 +++ pprocutils/http.go | 54 +++++ pprocutils/mock/http.go | 81 +++++++ pprocutils/v1/fromproto/http.go | 53 +++++ pprocutils/v1/toproto/http.go | 59 ++++++ proto/procutils/v1/http.pb.go | 362 ++++++++++++++++++++++++++++++++ proto/procutils/v1/http.proto | 35 +++ 7 files changed, 677 insertions(+) create mode 100644 pprocutils/http.go create mode 100644 pprocutils/mock/http.go create mode 100644 pprocutils/v1/fromproto/http.go create mode 100644 pprocutils/v1/toproto/http.go create mode 100644 proto/procutils/v1/http.pb.go create mode 100644 proto/procutils/v1/http.proto diff --git a/pprocutils/errors.go b/pprocutils/errors.go index 8d1d2e8..1b2d307 100644 --- a/pprocutils/errors.go +++ b/pprocutils/errors.go @@ -31,6 +31,17 @@ const ( ErrorCodeSubjectNotFound ErrorCodeVersionNotFound ErrorCodeInvalidSchema + + // HTTP host-egress capability error codes (host function http_request). + // Appended to the same iota band so existing codes keep their stable + // numeric values — never renumber the codes above. + ErrorCodeHTTPEgressDisabled + ErrorCodeHTTPForbidden + ErrorCodeHTTPInvalidRequest + ErrorCodeHTTPDNS + ErrorCodeHTTPTimeout + ErrorCodeHTTPResponseTooLarge + ErrorCodeHTTPTransport ) var ( @@ -44,6 +55,14 @@ var ( ErrInvalidSchema = NewError(ErrorCodeInvalidSchema, "invalid schema") ErrInternal = NewError(ErrorCodeInternal, "internal error") + + ErrHTTPEgressDisabled = NewError(ErrorCodeHTTPEgressDisabled, "http egress is not enabled for this processor") + ErrHTTPForbidden = NewError(ErrorCodeHTTPForbidden, "http egress destination is forbidden by policy") + ErrHTTPInvalidRequest = NewError(ErrorCodeHTTPInvalidRequest, "invalid http egress request") + ErrHTTPDNS = NewError(ErrorCodeHTTPDNS, "http egress DNS resolution failed") + ErrHTTPTimeout = NewError(ErrorCodeHTTPTimeout, "http egress call timed out") + ErrHTTPResponseTooLarge = NewError(ErrorCodeHTTPResponseTooLarge, "http egress response exceeded the size cap") + ErrHTTPTransport = NewError(ErrorCodeHTTPTransport, "http egress transport error") ) type Error struct { @@ -88,6 +107,20 @@ func NewErrorFromCode(code uint32) *Error { return ErrInvalidSchema case ErrorCodeInternal: return ErrInternal + case ErrorCodeHTTPEgressDisabled: + return ErrHTTPEgressDisabled + case ErrorCodeHTTPForbidden: + return ErrHTTPForbidden + case ErrorCodeHTTPInvalidRequest: + return ErrHTTPInvalidRequest + case ErrorCodeHTTPDNS: + return ErrHTTPDNS + case ErrorCodeHTTPTimeout: + return ErrHTTPTimeout + case ErrorCodeHTTPResponseTooLarge: + return ErrHTTPResponseTooLarge + case ErrorCodeHTTPTransport: + return ErrHTTPTransport default: return NewError(code, "unknown error code") } diff --git a/pprocutils/http.go b/pprocutils/http.go new file mode 100644 index 0000000..4be1167 --- /dev/null +++ b/pprocutils/http.go @@ -0,0 +1,54 @@ +// 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. + +//go:generate mockgen -typed -destination=mock/http.go -package=mock -mock_names=HTTPService=HTTPService . HTTPService + +package pprocutils + +import "context" + +// HTTPRequest is a single-shot, buffered outbound HTTP request a standalone +// (WASM) processor asks the host to perform on its behalf. The guest never gets +// a socket: the host validates this request against a per-processor egress +// policy (allowlist + resolved-IP gate) and performs the I/O with full net/http. +type HTTPRequest struct { + Method string + URL string + Headers map[string][]string + Body []byte + // AuthSecretRef names a secret the HOST resolves and injects as the + // Authorization header, after all validation, immediately before dispatch. + // The guest never supplies the credential value, and a guest-supplied + // Authorization header is rejected — the key never enters guest memory. + AuthSecretRef string +} + +// HTTPResponse is the fully-buffered, host-size-capped response returned to the +// guest. There is no streaming contract: the whole (capped) body is +// materialized host-side before it crosses back to the guest. +type HTTPResponse struct { + StatusCode int + Headers map[string][]string + Body []byte +} + +// HTTPService is the host-side seam for the egress capability, mirroring +// SchemaService. It is implemented host-side, bound to a single processor +// instance's resolved egress policy; the implementation enforces the security +// boundary (allowlist, DNS-rebinding resolved-IP gate, no-proxy transport, +// redirect suppression, per-call timeout, response-size cap, host-injected +// credentials). DO NOT use this package directly. +type HTTPService interface { + Do(ctx context.Context, req HTTPRequest) (HTTPResponse, error) +} diff --git a/pprocutils/mock/http.go b/pprocutils/mock/http.go new file mode 100644 index 0000000..a454252 --- /dev/null +++ b/pprocutils/mock/http.go @@ -0,0 +1,81 @@ +// Code generated by MockGen. DO NOT EDIT. +// Source: github.com/conduitio/conduit-processor-sdk/pprocutils (interfaces: HTTPService) +// +// Generated by this command: +// +// mockgen -typed -destination=mock/http.go -package=mock -mock_names=HTTPService=HTTPService . HTTPService +// + +// Package mock is a generated GoMock package. +package mock + +import ( + context "context" + reflect "reflect" + + pprocutils "github.com/conduitio/conduit-processor-sdk/pprocutils" + gomock "go.uber.org/mock/gomock" +) + +// HTTPService is a mock of HTTPService interface. +type HTTPService struct { + ctrl *gomock.Controller + recorder *HTTPServiceMockRecorder + isgomock struct{} +} + +// HTTPServiceMockRecorder is the mock recorder for HTTPService. +type HTTPServiceMockRecorder struct { + mock *HTTPService +} + +// NewHTTPService creates a new mock instance. +func NewHTTPService(ctrl *gomock.Controller) *HTTPService { + mock := &HTTPService{ctrl: ctrl} + mock.recorder = &HTTPServiceMockRecorder{mock} + return mock +} + +// EXPECT returns an object that allows the caller to indicate expected use. +func (m *HTTPService) EXPECT() *HTTPServiceMockRecorder { + return m.recorder +} + +// Do mocks base method. +func (m *HTTPService) Do(ctx context.Context, req pprocutils.HTTPRequest) (pprocutils.HTTPResponse, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "Do", ctx, req) + ret0, _ := ret[0].(pprocutils.HTTPResponse) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// Do indicates an expected call of Do. +func (mr *HTTPServiceMockRecorder) Do(ctx, req any) *HTTPServiceDoCall { + mr.mock.ctrl.T.Helper() + call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Do", reflect.TypeOf((*HTTPService)(nil).Do), ctx, req) + return &HTTPServiceDoCall{Call: call} +} + +// HTTPServiceDoCall wrap *gomock.Call +type HTTPServiceDoCall struct { + *gomock.Call +} + +// Return rewrite *gomock.Call.Return +func (c *HTTPServiceDoCall) Return(arg0 pprocutils.HTTPResponse, arg1 error) *HTTPServiceDoCall { + c.Call = c.Call.Return(arg0, arg1) + return c +} + +// Do rewrite *gomock.Call.Do +func (c *HTTPServiceDoCall) Do(f func(context.Context, pprocutils.HTTPRequest) (pprocutils.HTTPResponse, error)) *HTTPServiceDoCall { + c.Call = c.Call.Do(f) + return c +} + +// DoAndReturn rewrite *gomock.Call.DoAndReturn +func (c *HTTPServiceDoCall) DoAndReturn(f func(context.Context, pprocutils.HTTPRequest) (pprocutils.HTTPResponse, error)) *HTTPServiceDoCall { + c.Call = c.Call.DoAndReturn(f) + return c +} diff --git a/pprocutils/v1/fromproto/http.go b/pprocutils/v1/fromproto/http.go new file mode 100644 index 0000000..29d2251 --- /dev/null +++ b/pprocutils/v1/fromproto/http.go @@ -0,0 +1,53 @@ +// 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 fromproto + +import ( + "github.com/conduitio/conduit-processor-sdk/pprocutils" + procutilsv1 "github.com/conduitio/conduit-processor-sdk/proto/procutils/v1" +) + +func httpHeaders(in []*procutilsv1.HTTPHeader) map[string][]string { + if len(in) == 0 { + return nil + } + out := make(map[string][]string, len(in)) + for _, h := range in { + if h == nil { + continue + } + // Preserve multi-value semantics; a repeated key merges its values. + out[h.Key] = append(out[h.Key], h.Values...) + } + return out +} + +func HTTPRequest(req *procutilsv1.HTTPRequest) pprocutils.HTTPRequest { + return pprocutils.HTTPRequest{ + Method: req.Method, + URL: req.Url, + Headers: httpHeaders(req.Headers), + Body: req.Body, + AuthSecretRef: req.AuthSecretRef, + } +} + +func HTTPResponse(resp *procutilsv1.HTTPResponse) pprocutils.HTTPResponse { + return pprocutils.HTTPResponse{ + StatusCode: int(resp.StatusCode), + Headers: httpHeaders(resp.Headers), + Body: resp.Body, + } +} diff --git a/pprocutils/v1/toproto/http.go b/pprocutils/v1/toproto/http.go new file mode 100644 index 0000000..385f533 --- /dev/null +++ b/pprocutils/v1/toproto/http.go @@ -0,0 +1,59 @@ +// 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 toproto + +import ( + "sort" + + "github.com/conduitio/conduit-processor-sdk/pprocutils" + procutilsv1 "github.com/conduitio/conduit-processor-sdk/proto/procutils/v1" +) + +func httpHeaders(in map[string][]string) []*procutilsv1.HTTPHeader { + if len(in) == 0 { + return nil + } + // Deterministic key order so the wire encoding is stable (round-trip tests, + // reproducible marshalling). + keys := make([]string, 0, len(in)) + for k := range in { + keys = append(keys, k) + } + sort.Strings(keys) + + out := make([]*procutilsv1.HTTPHeader, 0, len(keys)) + for _, k := range keys { + out = append(out, &procutilsv1.HTTPHeader{Key: k, Values: in[k]}) + } + return out +} + +func HTTPRequest(in pprocutils.HTTPRequest) *procutilsv1.HTTPRequest { + return &procutilsv1.HTTPRequest{ + Method: in.Method, + Url: in.URL, + Headers: httpHeaders(in.Headers), + Body: in.Body, + AuthSecretRef: in.AuthSecretRef, + } +} + +func HTTPResponse(in pprocutils.HTTPResponse) *procutilsv1.HTTPResponse { + return &procutilsv1.HTTPResponse{ + StatusCode: int32(in.StatusCode), //nolint:gosec // HTTP status codes are small + Headers: httpHeaders(in.Headers), + Body: in.Body, + } +} diff --git a/proto/procutils/v1/http.pb.go b/proto/procutils/v1/http.pb.go new file mode 100644 index 0000000..cd3c079 --- /dev/null +++ b/proto/procutils/v1/http.pb.go @@ -0,0 +1,362 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.32.0 +// protoc (unknown) +// source: procutils/v1/http.proto + +package procutilsv1 + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + reflect "reflect" + sync "sync" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// HTTPHeader is a single HTTP header with one or more values. Headers are +// modelled as a repeated message (not a proto map) so multi-valued headers +// round-trip exactly. +type HTTPHeader struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"` + Values []string `protobuf:"bytes,2,rep,name=values,proto3" json:"values,omitempty"` +} + +func (x *HTTPHeader) Reset() { + *x = HTTPHeader{} + if protoimpl.UnsafeEnabled { + mi := &file_procutils_v1_http_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HTTPHeader) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HTTPHeader) ProtoMessage() {} + +func (x *HTTPHeader) ProtoReflect() protoreflect.Message { + mi := &file_procutils_v1_http_proto_msgTypes[0] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HTTPHeader.ProtoReflect.Descriptor instead. +func (*HTTPHeader) Descriptor() ([]byte, []int) { + return file_procutils_v1_http_proto_rawDescGZIP(), []int{0} +} + +func (x *HTTPHeader) GetKey() string { + if x != nil { + return x.Key + } + return "" +} + +func (x *HTTPHeader) GetValues() []string { + if x != nil { + return x.Values + } + return nil +} + +// HTTPRequest is the guest-marshalled request for the host-mediated egress +// capability (host function `http_request`). The host validates and performs +// the call; the guest never gets a socket. See the WASM host-egress design doc. +type HTTPRequest struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Method string `protobuf:"bytes,1,opt,name=method,proto3" json:"method,omitempty"` + Url string `protobuf:"bytes,2,opt,name=url,proto3" json:"url,omitempty"` + Headers []*HTTPHeader `protobuf:"bytes,3,rep,name=headers,proto3" json:"headers,omitempty"` + Body []byte `protobuf:"bytes,4,opt,name=body,proto3" json:"body,omitempty"` + // auth_secret_ref names a secret the HOST resolves and injects as the + // Authorization header. The guest never supplies the credential value; a + // guest-set Authorization header is rejected host-side. + AuthSecretRef string `protobuf:"bytes,5,opt,name=auth_secret_ref,json=authSecretRef,proto3" json:"auth_secret_ref,omitempty"` +} + +func (x *HTTPRequest) Reset() { + *x = HTTPRequest{} + if protoimpl.UnsafeEnabled { + mi := &file_procutils_v1_http_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HTTPRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HTTPRequest) ProtoMessage() {} + +func (x *HTTPRequest) ProtoReflect() protoreflect.Message { + mi := &file_procutils_v1_http_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HTTPRequest.ProtoReflect.Descriptor instead. +func (*HTTPRequest) Descriptor() ([]byte, []int) { + return file_procutils_v1_http_proto_rawDescGZIP(), []int{1} +} + +func (x *HTTPRequest) GetMethod() string { + if x != nil { + return x.Method + } + return "" +} + +func (x *HTTPRequest) GetUrl() string { + if x != nil { + return x.Url + } + return "" +} + +func (x *HTTPRequest) GetHeaders() []*HTTPHeader { + if x != nil { + return x.Headers + } + return nil +} + +func (x *HTTPRequest) GetBody() []byte { + if x != nil { + return x.Body + } + return nil +} + +func (x *HTTPRequest) GetAuthSecretRef() string { + if x != nil { + return x.AuthSecretRef + } + return "" +} + +// HTTPResponse is the host-produced, fully-buffered, size-capped response +// written back into the guest-owned buffer. +type HTTPResponse struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + StatusCode int32 `protobuf:"varint,1,opt,name=status_code,json=statusCode,proto3" json:"status_code,omitempty"` + Headers []*HTTPHeader `protobuf:"bytes,2,rep,name=headers,proto3" json:"headers,omitempty"` + Body []byte `protobuf:"bytes,3,opt,name=body,proto3" json:"body,omitempty"` +} + +func (x *HTTPResponse) Reset() { + *x = HTTPResponse{} + if protoimpl.UnsafeEnabled { + mi := &file_procutils_v1_http_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *HTTPResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HTTPResponse) ProtoMessage() {} + +func (x *HTTPResponse) ProtoReflect() protoreflect.Message { + mi := &file_procutils_v1_http_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HTTPResponse.ProtoReflect.Descriptor instead. +func (*HTTPResponse) Descriptor() ([]byte, []int) { + return file_procutils_v1_http_proto_rawDescGZIP(), []int{2} +} + +func (x *HTTPResponse) GetStatusCode() int32 { + if x != nil { + return x.StatusCode + } + return 0 +} + +func (x *HTTPResponse) GetHeaders() []*HTTPHeader { + if x != nil { + return x.Headers + } + return nil +} + +func (x *HTTPResponse) GetBody() []byte { + if x != nil { + return x.Body + } + return nil +} + +var File_procutils_v1_http_proto protoreflect.FileDescriptor + +var file_procutils_v1_http_proto_rawDesc = []byte{ + 0x0a, 0x17, 0x70, 0x72, 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x2f, 0x76, 0x31, 0x2f, 0x68, + 0x74, 0x74, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0c, 0x70, 0x72, 0x6f, 0x63, 0x75, + 0x74, 0x69, 0x6c, 0x73, 0x2e, 0x76, 0x31, 0x22, 0x36, 0x0a, 0x0a, 0x48, 0x54, 0x54, 0x50, 0x48, + 0x65, 0x61, 0x64, 0x65, 0x72, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x16, 0x0a, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, + 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x73, 0x22, + 0xa7, 0x01, 0x0a, 0x0b, 0x48, 0x54, 0x54, 0x50, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, + 0x16, 0x0a, 0x06, 0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x06, 0x6d, 0x65, 0x74, 0x68, 0x6f, 0x64, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x72, 0x6c, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x12, 0x32, 0x0a, 0x07, 0x68, 0x65, 0x61, + 0x64, 0x65, 0x72, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x70, 0x72, 0x6f, + 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x54, 0x54, 0x50, 0x48, 0x65, + 0x61, 0x64, 0x65, 0x72, 0x52, 0x07, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x12, 0x12, 0x0a, + 0x04, 0x62, 0x6f, 0x64, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x62, 0x6f, 0x64, + 0x79, 0x12, 0x26, 0x0a, 0x0f, 0x61, 0x75, 0x74, 0x68, 0x5f, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, + 0x5f, 0x72, 0x65, 0x66, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x61, 0x75, 0x74, 0x68, + 0x53, 0x65, 0x63, 0x72, 0x65, 0x74, 0x52, 0x65, 0x66, 0x22, 0x77, 0x0a, 0x0c, 0x48, 0x54, 0x54, + 0x50, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1f, 0x0a, 0x0b, 0x73, 0x74, 0x61, + 0x74, 0x75, 0x73, 0x5f, 0x63, 0x6f, 0x64, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0a, + 0x73, 0x74, 0x61, 0x74, 0x75, 0x73, 0x43, 0x6f, 0x64, 0x65, 0x12, 0x32, 0x0a, 0x07, 0x68, 0x65, + 0x61, 0x64, 0x65, 0x72, 0x73, 0x18, 0x02, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x18, 0x2e, 0x70, 0x72, + 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x2e, 0x76, 0x31, 0x2e, 0x48, 0x54, 0x54, 0x50, 0x48, + 0x65, 0x61, 0x64, 0x65, 0x72, 0x52, 0x07, 0x68, 0x65, 0x61, 0x64, 0x65, 0x72, 0x73, 0x12, 0x12, + 0x0a, 0x04, 0x62, 0x6f, 0x64, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x62, 0x6f, + 0x64, 0x79, 0x42, 0xb9, 0x01, 0x0a, 0x10, 0x63, 0x6f, 0x6d, 0x2e, 0x70, 0x72, 0x6f, 0x63, 0x75, + 0x74, 0x69, 0x6c, 0x73, 0x2e, 0x76, 0x31, 0x42, 0x09, 0x48, 0x74, 0x74, 0x70, 0x50, 0x72, 0x6f, + 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x49, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, + 0x2f, 0x63, 0x6f, 0x6e, 0x64, 0x75, 0x69, 0x74, 0x69, 0x6f, 0x2f, 0x63, 0x6f, 0x6e, 0x64, 0x75, + 0x69, 0x74, 0x2d, 0x70, 0x72, 0x6f, 0x63, 0x65, 0x73, 0x73, 0x6f, 0x72, 0x2d, 0x73, 0x64, 0x6b, + 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x70, 0x72, 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, + 0x2f, 0x76, 0x31, 0x3b, 0x70, 0x72, 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x76, 0x31, 0xa2, + 0x02, 0x03, 0x50, 0x58, 0x58, 0xaa, 0x02, 0x0c, 0x50, 0x72, 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, + 0x73, 0x2e, 0x56, 0x31, 0xca, 0x02, 0x0c, 0x50, 0x72, 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, + 0x5c, 0x56, 0x31, 0xe2, 0x02, 0x18, 0x50, 0x72, 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x5c, + 0x56, 0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, + 0x0d, 0x50, 0x72, 0x6f, 0x63, 0x75, 0x74, 0x69, 0x6c, 0x73, 0x3a, 0x3a, 0x56, 0x31, 0x62, 0x06, + 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, +} + +var ( + file_procutils_v1_http_proto_rawDescOnce sync.Once + file_procutils_v1_http_proto_rawDescData = file_procutils_v1_http_proto_rawDesc +) + +func file_procutils_v1_http_proto_rawDescGZIP() []byte { + file_procutils_v1_http_proto_rawDescOnce.Do(func() { + file_procutils_v1_http_proto_rawDescData = protoimpl.X.CompressGZIP(file_procutils_v1_http_proto_rawDescData) + }) + return file_procutils_v1_http_proto_rawDescData +} + +var file_procutils_v1_http_proto_msgTypes = make([]protoimpl.MessageInfo, 3) +var file_procutils_v1_http_proto_goTypes = []interface{}{ + (*HTTPHeader)(nil), // 0: procutils.v1.HTTPHeader + (*HTTPRequest)(nil), // 1: procutils.v1.HTTPRequest + (*HTTPResponse)(nil), // 2: procutils.v1.HTTPResponse +} +var file_procutils_v1_http_proto_depIdxs = []int32{ + 0, // 0: procutils.v1.HTTPRequest.headers:type_name -> procutils.v1.HTTPHeader + 0, // 1: procutils.v1.HTTPResponse.headers:type_name -> procutils.v1.HTTPHeader + 2, // [2:2] is the sub-list for method output_type + 2, // [2:2] is the sub-list for method input_type + 2, // [2:2] is the sub-list for extension type_name + 2, // [2:2] is the sub-list for extension extendee + 0, // [0:2] is the sub-list for field type_name +} + +func init() { file_procutils_v1_http_proto_init() } +func file_procutils_v1_http_proto_init() { + if File_procutils_v1_http_proto != nil { + return + } + if !protoimpl.UnsafeEnabled { + file_procutils_v1_http_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HTTPHeader); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_procutils_v1_http_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HTTPRequest); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_procutils_v1_http_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*HTTPResponse); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: file_procutils_v1_http_proto_rawDesc, + NumEnums: 0, + NumMessages: 3, + NumExtensions: 0, + NumServices: 0, + }, + GoTypes: file_procutils_v1_http_proto_goTypes, + DependencyIndexes: file_procutils_v1_http_proto_depIdxs, + MessageInfos: file_procutils_v1_http_proto_msgTypes, + }.Build() + File_procutils_v1_http_proto = out.File + file_procutils_v1_http_proto_rawDesc = nil + file_procutils_v1_http_proto_goTypes = nil + file_procutils_v1_http_proto_depIdxs = nil +} diff --git a/proto/procutils/v1/http.proto b/proto/procutils/v1/http.proto new file mode 100644 index 0000000..f8a68f8 --- /dev/null +++ b/proto/procutils/v1/http.proto @@ -0,0 +1,35 @@ +syntax = "proto3"; + +package procutils.v1; + +option go_package = "github.com/conduitio/conduit-processor-sdk/proto/procutils/v1"; + +// HTTPHeader is a single HTTP header with one or more values. Headers are +// modelled as a repeated message (not a proto map) so multi-valued headers +// round-trip exactly. +message HTTPHeader { + string key = 1; + repeated string values = 2; +} + +// HTTPRequest is the guest-marshalled request for the host-mediated egress +// capability (host function `http_request`). The host validates and performs +// the call; the guest never gets a socket. See the WASM host-egress design doc. +message HTTPRequest { + string method = 1; + string url = 2; + repeated HTTPHeader headers = 3; + bytes body = 4; + // auth_secret_ref names a secret the HOST resolves and injects as the + // Authorization header. The guest never supplies the credential value; a + // guest-set Authorization header is rejected host-side. + string auth_secret_ref = 5; +} + +// HTTPResponse is the host-produced, fully-buffered, size-capped response +// written back into the guest-owned buffer. +message HTTPResponse { + int32 status_code = 1; + repeated HTTPHeader headers = 2; + bytes body = 3; +} From 0774e4794ad994d3807720f4b10762c83a7fc6e4 Mon Sep 17 00:00:00 2001 From: DeVaris Brown Date: Sun, 26 Jul 2026 18:20:08 -0700 Subject: [PATCH 2/2] feat(wasm): guest-side stub + ergonomic API for the http_request egress capability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deferred guest half of bed0225 (shared HTTPService ABI). A standalone (WASM) processor can now reach the host-mediated network-egress capability via the egress package: - wasm/imports.go: //go:wasmimport conduit http_request, mirroring create_schema/get_schema exactly (same park/resize protocol). - wasm/http.go: httpService, the guest-side pprocutils.HTTPService — marshals via toproto.HTTPRequest, calls hostCall, unmarshals via fromproto.HTTPResponse. Verbatim reuse of the schemaService pattern. - wasm/util.go: InitUtils wires egress.HTTPService = &httpService{} at startup, alongside the existing schema wiring. - egress package (new): the processor-facing API (Do, Request, Response, sentinel Err* vars). Defaults to a deny-all HTTPService outside a standalone processor, since there is no host boundary to broker the call through in that hosting mode. Godoc states the security contract plainly: deny-by-default/operator-gated, host-injected credentials only, no redirect-following, host-size-capped responses. Per docs/design-documents/20260726-wasm-host-egress-capability.md in ConduitIO/conduit (host side: feat/wasm-host-egress-impl, host_module.go httpRequest). ABI verified symmetric against that branch: import name, proto types, and the numeric ErrorCodeHTTP* band all match byte-for-byte (both repos share this exact checkout via the host's go.mod replace during bundled development). Tests: egress package (mocked HTTPService, sentinel-error decoding via errors.Is, deny-all default) and a wire-level round-trip test for toproto/fromproto HTTPRequest/HTTPResponse — a gap the ABI commit (bed0225) left uncovered. The guest stub itself (wasm/http.go) cannot be exercised without a running wasm host; that path belongs in the bundle's cross-repo e2e test, not here. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_015GQFzakPShAYj8CcwajYDD --- egress/doc.go | 61 ++++++++++++ egress/egress.go | 145 +++++++++++++++++++++++++++++ egress/egress_test.go | 115 +++++++++++++++++++++++ egress/example_test.go | 36 +++++++ pprocutils/v1/toproto/http_test.go | 124 ++++++++++++++++++++++++ wasm/http.go | 68 ++++++++++++++ wasm/imports.go | 22 +++++ wasm/util.go | 6 ++ 8 files changed, 577 insertions(+) create mode 100644 egress/doc.go create mode 100644 egress/egress.go create mode 100644 egress/egress_test.go create mode 100644 egress/example_test.go create mode 100644 pprocutils/v1/toproto/http_test.go create mode 100644 wasm/http.go diff --git a/egress/doc.go b/egress/doc.go new file mode 100644 index 0000000..ebdcad1 --- /dev/null +++ b/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 diff --git a/egress/egress.go b/egress/egress.go new file mode 100644 index 0000000..bf2b8ca --- /dev/null +++ b/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 +} diff --git a/egress/egress_test.go b/egress/egress_test.go new file mode 100644 index 0000000..6a07309 --- /dev/null +++ b/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)) +} diff --git a/egress/example_test.go b/egress/example_test.go new file mode 100644 index 0000000..4a4fce7 --- /dev/null +++ b/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 +} diff --git a/pprocutils/v1/toproto/http_test.go b/pprocutils/v1/toproto/http_test.go new file mode 100644 index 0000000..f9daaeb --- /dev/null +++ b/pprocutils/v1/toproto/http_test.go @@ -0,0 +1,124 @@ +// 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 toproto_test + +import ( + "testing" + + "github.com/conduitio/conduit-processor-sdk/pprocutils" + "github.com/conduitio/conduit-processor-sdk/pprocutils/v1/fromproto" + "github.com/conduitio/conduit-processor-sdk/pprocutils/v1/toproto" + procutilsv1 "github.com/conduitio/conduit-processor-sdk/proto/procutils/v1" + "github.com/matryer/is" + "google.golang.org/protobuf/proto" +) + +// TestHTTPRequest_RoundTrip exercises the exact path the guest stub in +// wasm/http.go relies on: a pprocutils.HTTPRequest converted to its proto +// form, marshalled to wire bytes, unmarshalled back, and converted back to a +// pprocutils.HTTPRequest — the same sequence hostCall's buffer protocol +// performs across the host boundary. +func TestHTTPRequest_RoundTrip(t *testing.T) { + tests := []struct { + name string + req pprocutils.HTTPRequest + }{ + { + name: "full request with multi-value headers", + req: pprocutils.HTTPRequest{ + Method: "POST", + URL: "https://api.example.com/v1/embeddings", + Headers: map[string][]string{ + "Content-Type": {"application/json"}, + "X-Trace-Id": {"abc", "def"}, + }, + Body: []byte(`{"input":"hello"}`), + AuthSecretRef: "openai_api_key", + }, + }, + { + name: "no headers, no body, no auth ref", + req: pprocutils.HTTPRequest{ + Method: "GET", + URL: "https://api.example.com/v1/models", + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + is := is.New(t) + + protoReq := toproto.HTTPRequest(tt.req) + + wire, err := proto.Marshal(protoReq) + is.NoErr(err) + + var decoded procutilsv1.HTTPRequest + err = proto.Unmarshal(wire, &decoded) + is.NoErr(err) + + got := fromproto.HTTPRequest(&decoded) + is.Equal(got, tt.req) + }) + } +} + +// TestHTTPResponse_RoundTrip is the response-side companion to +// TestHTTPRequest_RoundTrip, covering the direction the guest stub decodes: +// host-produced proto bytes back into a pprocutils.HTTPResponse. +func TestHTTPResponse_RoundTrip(t *testing.T) { + tests := []struct { + name string + resp pprocutils.HTTPResponse + }{ + { + name: "full response with multi-value headers", + resp: pprocutils.HTTPResponse{ + StatusCode: 200, + Headers: map[string][]string{ + "Content-Type": {"application/json"}, + "Set-Cookie": {"a=1", "b=2"}, + "X-Request-Id": {"req-123"}, + }, + Body: []byte(`{"result":"ok"}`), + }, + }, + { + name: "error status, no headers, no body", + resp: pprocutils.HTTPResponse{ + StatusCode: 403, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + is := is.New(t) + + protoResp := toproto.HTTPResponse(tt.resp) + + wire, err := proto.Marshal(protoResp) + is.NoErr(err) + + var decoded procutilsv1.HTTPResponse + err = proto.Unmarshal(wire, &decoded) + is.NoErr(err) + + got := fromproto.HTTPResponse(&decoded) + is.Equal(got, tt.resp) + }) + } +} diff --git a/wasm/http.go b/wasm/http.go new file mode 100644 index 0000000..36ab67d --- /dev/null +++ b/wasm/http.go @@ -0,0 +1,68 @@ +// 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. + +//go:build wasm + +package wasm + +import ( + "context" + "fmt" + + "github.com/conduitio/conduit-processor-sdk/pprocutils" + "github.com/conduitio/conduit-processor-sdk/pprocutils/v1/fromproto" + "github.com/conduitio/conduit-processor-sdk/pprocutils/v1/toproto" + procutilsv1 "github.com/conduitio/conduit-processor-sdk/proto/procutils/v1" + "google.golang.org/protobuf/proto" +) + +// httpService is the guest-side pprocutils.HTTPService implementation for +// standalone (WASM) processors. It is a verbatim reuse of the schemaService +// pattern: marshal the request into the shared buffer, invoke the host +// import via the existing park/resize protocol (hostCall), and unmarshal the +// buffered response the host wrote back. All I/O and policy enforcement +// (allowlist, resolved-IP dial-time gate, redirect suppression, size cap, +// host-injected credentials) happens host-side; this type only carries bytes +// across the host boundary. +type httpService struct{} + +// Do marshals req, calls the conduit.http_request host import, and +// unmarshals the response. A host-side rejection or failure surfaces as a +// *pprocutils.Error carrying one of the ErrorCodeHTTP* codes (see +// pprocutils/errors.go), decoded by hostCall from the numeric error-code band +// exactly as create_schema/get_schema already do. +func (*httpService) Do(_ context.Context, req pprocutils.HTTPRequest) (pprocutils.HTTPResponse, error) { + protoReq := toproto.HTTPRequest(req) + + buffer := bufferPool.Get().([]byte) + defer bufferPool.Put(buffer) + + buffer, err := proto.MarshalOptions{}.MarshalAppend(buffer[:0], protoReq) + if err != nil { + return pprocutils.HTTPResponse{}, fmt.Errorf("error marshalling request: %w", err) + } + + buffer, cmdSize, err := hostCall(_httpRequest, buffer) + if err != nil { + return pprocutils.HTTPResponse{}, fmt.Errorf("error calling httpRequest: %w", err) + } + + var resp procutilsv1.HTTPResponse + err = proto.Unmarshal(buffer[:cmdSize], &resp) + if err != nil { + return pprocutils.HTTPResponse{}, fmt.Errorf("failed unmarshalling %v bytes into proto type: %w", cmdSize, err) + } + + return fromproto.HTTPResponse(&resp), nil +} diff --git a/wasm/imports.go b/wasm/imports.go index 3f25b95..2f20865 100644 --- a/wasm/imports.go +++ b/wasm/imports.go @@ -67,3 +67,25 @@ func _createSchema(ptr unsafe.Pointer, size uint32) uint32 // //go:wasmimport conduit get_schema func _getSchema(ptr unsafe.Pointer, size uint32) uint32 + +// Imports `http_request` from the host, which performs a host-mediated, +// allowlisted outbound HTTP call on the guest's behalf. The guest never gets a +// socket: the host validates the request against this processor's resolved +// egress policy (allowlist, resolved-IP dial-time gate, no-proxy transport, +// redirect suppression, per-call timeout, response-size cap, host-injected +// credentials — see docs/design-documents/20260726-wasm-host-egress-capability.md +// in ConduitIO/conduit) and performs the I/O itself. +// +// The arguments are: +// (1) a pointer to the address where the marshalled HTTPRequest was written; +// the host writes the marshalled HTTPResponse back into the same address. +// (2) the size of allocated memory. +// +// The return value indicates the size of the written response in bytes, or a +// value >= pprocutils.ErrorCodeStart on failure (see pprocutils/errors.go for +// the ErrorCodeHTTP* band). If the response is larger than the allocated +// memory, the caller should reallocate the memory and call `http_request` +// again — identical to `create_schema`/`get_schema`. +// +//go:wasmimport conduit http_request +func _httpRequest(ptr unsafe.Pointer, size uint32) uint32 diff --git a/wasm/util.go b/wasm/util.go index f90f22b..165950a 100644 --- a/wasm/util.go +++ b/wasm/util.go @@ -19,6 +19,7 @@ package wasm import ( "os" + "github.com/conduitio/conduit-processor-sdk/egress" "github.com/conduitio/conduit-processor-sdk/pprocutils" "github.com/conduitio/conduit-processor-sdk/schema" "github.com/rs/zerolog" @@ -27,6 +28,7 @@ import ( func InitUtils(logLevel string) { initLogger(logLevel) initSchemaService() + initHTTPService() } func initLogger(logLevel string) { @@ -45,3 +47,7 @@ func initLogger(logLevel string) { func initSchemaService() { schema.SchemaService = &schemaService{} } + +func initHTTPService() { + egress.HTTPService = &httpService{} +}