Merge pull request #463 from hookdeck/feat/listen-binary-bodies

feat(listen): forward binary request bodies byte-exact

Author: leggetter

pkg/listen/proxy/binary_body_test.go

@@ -0,0 +1,133 @@
+package proxy
+
+import (
+	"bytes"
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/hookdeck/hookdeck-cli/pkg/websocket"
+)
+
+type receivedRequest struct {
+	body          []byte
+	contentType   string
+	contentLength int64
+}
+
+// forwardAttempt runs one attempt through processAttempt against a local
+// server and returns what that server received.
+func forwardAttempt(t *testing.T, request websocket.AttemptRequest) receivedRequest {
+	t.Helper()
+
+	received := make(chan receivedRequest, 1)
+	local := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, _ := io.ReadAll(r.Body)
+		received <- receivedRequest{body: body, contentType: r.Header.Get("Content-Type"), contentLength: r.ContentLength}
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(local.Close)
+
+	target, err := url.Parse(local.URL)
+	require.NoError(t, err)
+	p := New(&Config{URL: target, NoHealthcheck: true}, nil, newRecordingRenderer())
+
+	p.processAttempt(websocket.IncomingMessage{Attempt: &websocket.Attempt{
+		Body: websocket.AttemptBody{Path: "/webhooks", EventID: "evt_test", AttemptId: "evt_test", Request: request},
+	}})
+
+	select {
+	case r := <-received:
+		return r
+	case <-time.After(5 * time.Second):
+		t.Fatal("local server never received the request")
+		return receivedRequest{}
+	}
+}
+
+func headersJSON(t *testing.T, headers map[string]string) json.RawMessage {
+	t.Helper()
+	raw, err := json.Marshal(headers)
+	require.NoError(t, err)
+	return raw
+}
+
+// Every byte value, so any UTF-8 round trip along the way would show.
+func allBytes() []byte {
+	b := make([]byte, 256)
+	for i := range b {
+		b[i] = byte(i)
+	}
+	return b
+}
+
+func TestProcessAttemptForwardsBinaryBodyByteExact(t *testing.T) {
+	body := allBytes()
+
+	got := forwardAttempt(t, websocket.AttemptRequest{
+		Method:     http.MethodPost,
+		BodyFormat: websocket.BodyFormatBinary,
+		DataBase64: base64.StdEncoding.EncodeToString(body),
+		// A stale Content-Length must not win over the decoded body's length.
+		Headers: headersJSON(t, map[string]string{"content-type": "application/octet-stream", "content-length": "1"}),
+	})
+
+	assert.Equal(t, body, got.body)
+	assert.Equal(t, int64(len(body)), got.contentLength)
+	assert.Equal(t, "application/octet-stream", got.contentType)
+}
+
+func TestProcessAttemptForwardsBinaryMultipartUnparsed(t *testing.T) {
+	contentType := "multipart/form-data; boundary=hookdeck-boundary"
+	var body bytes.Buffer
+	body.WriteString("--hookdeck-boundary\r\nContent-Disposition: form-data; name=\"field\"\r\n\r\nvalue\r\n")
+	body.WriteString("--hookdeck-boundary\r\nContent-Disposition: form-data; name=\"file\"; filename=\"f.bin\"\r\nContent-Type: application/octet-stream\r\n\r\n")
+	body.Write(allBytes())
+	body.WriteString("\r\n--hookdeck-boundary--\r\n")
+
+	got := forwardAttempt(t, websocket.AttemptRequest{
+		Method:     http.MethodPost,
+		BodyFormat: websocket.BodyFormatBinary,
+		DataBase64: base64.StdEncoding.EncodeToString(body.Bytes()),
+		Headers:    headersJSON(t, map[string]string{"content-type": contentType}),
+	})
+
+	assert.Equal(t, body.Bytes(), got.body, "boundary and file part bytes must match the original")
+	assert.Equal(t, contentType, got.contentType, "the boundary travels in the original Content-Type")
+}
+
+func TestProcessAttemptForwardsTextBodyFromDataString(t *testing.T) {
+	// The shape every server sends today, and the only one older servers send.
+	got := forwardAttempt(t, websocket.AttemptRequest{
+		Method:     http.MethodPost,
+		DataString: `{"hello":"wörld"}`,
+		Headers:    headersJSON(t, map[string]string{"content-type": "application/json"}),
+	})
+
+	assert.Equal(t, `{"hello":"wörld"}`, string(got.body))
+	assert.Equal(t, int64(len(`{"hello":"wörld"}`)), got.contentLength)
+}
+
+func TestAttemptRequestDecodesFromTheServerWireFormat(t *testing.T) {
+	var msg websocket.IncomingMessage
+	require.NoError(t, json.Unmarshal([]byte(`{"event":"attempt","body":{"cli_path":"/","request":{"method":"POST","headers":{},"body_format":"binary","data_base64":"AP+A"}}}`), &msg))
+	require.NotNil(t, msg.Attempt)
+
+	body, err := msg.Attempt.Body.Request.Body()
+	require.NoError(t, err)
+	assert.Equal(t, []byte{0x00, 0xff, 0x80}, body)
+	assert.True(t, msg.Attempt.Body.Request.IsBinary())
+}
+
+func TestAttemptRequestRejectsInvalidBase64(t *testing.T) {
+	_, err := websocket.AttemptRequest{BodyFormat: websocket.BodyFormatBinary, DataBase64: "not base64!"}.Body()
+	assert.Error(t, err)
+}

pkg/listen/proxy/proxy.go

@@ -1,6 +1,7 @@
 package proxy
 
 import (
+	"bytes"
 	"context"
 	"crypto/tls"
 	"encoding/json"
@@ -13,7 +14,6 @@ import (
 	"os"
 	"os/signal"
 	"strconv"
-	"strings"
 	"sync"
 	"sync/atomic"
 	"syscall"
@@ -440,8 +440,27 @@ func (p *Proxy) processAttempt(msg websocket.IncomingMessage) {
 		req.Header.Set(key, unquoted_value)
 	}
 
-	req.Body = ioutil.NopCloser(strings.NewReader(webhookEvent.Body.Request.DataString))
-	req.ContentLength = int64(len(webhookEvent.Body.Request.DataString))
+	// Binary bodies (data_base64) are forwarded as the original bytes, with the
+	// original Content-Type (multipart boundary included) already set above.
+	// net/http ignores Content-Length in req.Header and uses ContentLength.
+	body, err := webhookEvent.Body.Request.Body()
+	if err != nil {
+		p.renderer.OnEventError(eventID, webhookEvent, fmt.Errorf("decoding binary request body: %w", err), time.Now())
+		// Fail the attempt now rather than leave Hookdeck waiting for its timeout.
+		if wsClient := p.currentWebSocketClient(); wsClient != nil {
+			wsClient.SendMessage(&websocket.OutgoingMessage{
+				ErrorAttemptResponse: &websocket.ErrorAttemptResponse{
+					Event: "attempt_response",
+					Body: websocket.ErrorAttemptBody{
+						AttemptId: webhookEvent.Body.AttemptId,
+						Error:     true,
+					},
+				}})
+		}
+		return
+	}
+	req.Body = ioutil.NopCloser(bytes.NewReader(body))
+	req.ContentLength = int64(len(body))
 
 	// For interactive mode: start 100ms timer and HTTP request concurrently
 	requestStartTime := time.Now()

pkg/listen/tui/model.go

@@ -366,8 +366,11 @@ func (m *Model) buildEventDetailsContent(event *EventInfo) (string, requestCopyC
 		requestCopy.headers = strings.TrimSuffix(requestCopy.headers, "\n")
 		content.WriteString("\n")
 
-		// Request body
-		if event.Data.Body.Request.DataString != "" {
+		// Request body. Binary bodies are summarised rather than printed: they
+		// are not text, and the copied request would not reproduce them.
+		if event.Data.Body.Request.IsBinary() {
+			content.WriteString(faintStyle.Render(binaryBodySummary(event.Data.Body.Request)) + "\n")
+		} else if event.Data.Body.Request.DataString != "" {
 			// Try to pretty print JSON
 			requestCopy.body = m.prettyPrintJSON(event.Data.Body.Request.DataString)
 			content.WriteString(requestCopy.body + "\n")
@@ -438,6 +441,16 @@ func (c *requestCopyContent) buildRequest() {
 	c.request = strings.Join(parts, "\n\n")
 }
 
+// binaryBodySummary describes a binary request body by size, since its bytes
+// cannot be shown as text.
+func binaryBodySummary(req websocket.AttemptRequest) string {
+	body, err := req.Body()
+	if err != nil {
+		return "(binary body could not be decoded)"
+	}
+	return fmt.Sprintf("(binary body, %d bytes)", len(body))
+}
+
 // prettyPrintJSON attempts to pretty print JSON, returns original if not valid JSON.
 // It uses json.Indent so object key order is preserved exactly as received rather
 // than being sorted (which json.Marshal would do).

pkg/websocket/attempt_messages.go

@@ -1,16 +1,47 @@
 package websocket
 
 import (
+	"encoding/base64"
 	"encoding/json"
 )
 
+// CapabilitiesHeader advertises, on every websocket connect, which attempt
+// formats this CLI understands. The server only sends a binary body
+// (data_base64) to a session that advertised CapabilityBinaryBody; without it,
+// binary events fail on the server instead of reaching an older CLI that would
+// forward an empty body.
+const CapabilitiesHeader = "X-Hookdeck-CLI-Capabilities"
+
+// CapabilityBinaryBody means the CLI forwards request.data_base64 as raw bytes.
+const CapabilityBinaryBody = "binary"
+
+// BodyFormatBinary marks an attempt whose body is carried in DataBase64.
+const BodyFormatBinary = "binary"
+
 type AttemptRequest struct {
 	Method     string          `json:"method"`
 	Timeout    int64           `json:"timeout"`
 	DataString string          `json:"data_string"`
+	BodyFormat string          `json:"body_format,omitempty"`
+	DataBase64 string          `json:"data_base64,omitempty"`
 	Headers    json.RawMessage `json:"headers"`
 }
 
+// IsBinary reports whether the body travels as base64 rather than as DataString.
+func (r AttemptRequest) IsBinary() bool {
+	return r.BodyFormat == BodyFormatBinary || r.DataBase64 != ""
+}
+
+// Body returns the exact bytes to forward to the local server. Binary bodies
+// are decoded from DataBase64; text bodies are DataString as sent, which keeps
+// working against servers that predate data_base64.
+func (r AttemptRequest) Body() ([]byte, error) {
+	if r.IsBinary() {
+		return base64.StdEncoding.DecodeString(r.DataBase64)
+	}
+	return []byte(r.DataString), nil
+}
+
 type AttemptBody struct {
 	Path         string         `json:"cli_path"`
 	EventID      string         `json:"event_id"`

pkg/websocket/client.go

@@ -303,6 +303,7 @@ func (c *Client) connect(ctx context.Context) error {
 	header.Set("Accept-Encoding", "identity")
 	header.Set("User-Agent", useragent.GetEncodedUserAgent())
 	header.Set("X-Hookdeck-Client-User-Agent", useragent.GetEncodedHookdeckUserAgent())
+	header.Set(CapabilitiesHeader, CapabilityBinaryBody)
 	header.Set("Websocket-Id", c.WebSocketID)
 	header.Set("X-Team-Id", c.TeamID)
 	header.Set("Authorization", "Basic "+basicAuth(c.CLIKey, ""))

pkg/websocket/client_test.go

@@ -72,6 +72,9 @@ func TestConnectSendsSessionRecreationHeaders(t *testing.T) {
 	if got := captured.Get("Websocket-Id"); got != "cses_test" {
 		t.Errorf("Websocket-Id = %q, want %q", got, "cses_test")
 	}
+	if got := captured.Get(CapabilitiesHeader); got != CapabilityBinaryBody {
+		t.Errorf("%s = %q, want %q", CapabilitiesHeader, got, CapabilityBinaryBody)
+	}
 	if got := captured.Get("X-Webhook-Ids"); got != "web_abc,web_def" {
 		t.Errorf("X-Webhook-Ids = %q, want %q", got, "web_abc,web_def")
 	}