fix: make HTTP body capture completion-aware

cmd/clawpatrol/credential_match_test.go

@@ -85,10 +85,25 @@ profile "default" {
 	// `http.method != 'GET'` condition, so it still falls through to the
 	// default deny. This guards against a "fix" that ignores the condition.
 	t.Run("condition still gates the pinned rule", func(t *testing.T) {
+		events, cancelEvents := h.gateway.sink.Subscribe()
+		defer cancelEvents()
 		resp := h.send(t, http.MethodGet, "")
 		if resp.status != http.StatusForbidden {
 			t.Fatalf("GET status = %d (body %q); want 403 from the default deny", resp.status, resp.body)
 		}
+
+		end := waitHTTPSAuditEnd(t, events, "deny")
+		if end.ReqBodyState != bodyCaptureComplete {
+			t.Fatalf("request body state = %q, want %q", end.ReqBodyState, bodyCaptureComplete)
+		}
+		if end.ReqBody != "" {
+			t.Fatalf("recorded request body = %q, want empty", end.ReqBody)
+		}
+		rw := httptest.NewRecorder()
+		(&webMux{g: h.gateway}).writeActionFixture(rw, &end)
+		if rw.Code != http.StatusOK {
+			t.Fatalf("fixture export status = %d, want 200; body=%s", rw.Code, rw.Body.String())
+		}
 	})
 }
 

cmd/clawpatrol/dev_seed.go

@@ -482,6 +482,8 @@ func devSeedAction(r *rand.Rand, devices []devSeedDevice, ts time.Time) Event {
 			devSeedClaudeModels[r.Intn(len(devSeedClaudeModels))],
 			devSeedSessionTitles[r.Intn(len(devSeedSessionTitles))])
 		ev.RespBody = `{"id":"msg_abc123","type":"message","role":"assistant","content":[{"type":"text","text":"Sure — let me think about that..."}],"stop_reason":"end_turn"}`
+		ev.ReqBodyState = bodyCaptureComplete
+		ev.RespBodyState = bodyCaptureComplete
 		ev.ReqHeaders = map[string]string{
 			"Content-Type":  "application/json",
 			"User-Agent":    "clawpatrol/0.1 anthropic-sdk/0.55",
@@ -561,15 +563,15 @@ func devSeedActions(g *Gateway, r *rand.Rand, devices []devSeedDevice, count int
 				  (action_id, ts_ns, mode, family, agent_ip, host,
 				   method, path, status, bytes_in, bytes_out,
 				   ms, action, reason, req_sha, resp_sha,
-				   req_body, resp_body,
+				   req_body, resp_body, req_body_state, resp_body_state,
 				   req_headers, resp_headers, extra,
 				   endpoint, rule)
-				VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
+				VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`,
 				ev.ID, ev.Ts.UnixNano(), ev.Mode, ev.Family, ev.AgentIP,
 				ev.Host, ev.Method, ev.Path, ev.Status,
 				ev.In, ev.Out, ev.Ms, ev.Action, ev.Reason,
 				ev.ReqSha, ev.RespSha,
-				ev.ReqBody, ev.RespBody,
+				ev.ReqBody, ev.RespBody, ev.ReqBodyState, ev.RespBodyState,
 				devSeedHeadersJSON(rqhJSON), devSeedHeadersJSON(rshJSON),
 				string(extraJSON),
 				ev.Endpoint, ev.Rule,

cmd/clawpatrol/hitl_body_sample_test.go

@@ -4,9 +4,11 @@ import (
 	"bufio"
 	"context"
 	"crypto/tls"
+	"encoding/json"
 	"io"
 	"net"
 	"net/http"
+	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
@@ -68,6 +70,8 @@ profile "default" { credentials = [bearer_token.pat] }
 		t.Fatalf("NewSink: %v", err)
 	}
 	defer close(sink.ch)
+	events, cancelEvents := sink.Subscribe()
+	defer cancelEvents()
 	certs, _ := inMemoryCertCache(t)
 	g := &Gateway{certs: certs, sink: sink}
 	g.cfg.Store(gw)
@@ -115,10 +119,49 @@ profile "default" { credentials = [bearer_token.pat] }
 		t.Fatal("timed out waiting for approver body sample")
 	}
 
+	end := waitHTTPSAuditEnd(t, events, "denied")
+	if end.ReqBodyState != bodyCaptureComplete {
+		t.Fatalf("request body state = %q, want %q", end.ReqBodyState, bodyCaptureComplete)
+	}
+	if end.ReqBody != requestBody {
+		t.Fatalf("recorded request body = %q, want %q", end.ReqBody, requestBody)
+	}
+	if end.ReqHeaders["Content-Type"] != "application/json" {
+		t.Fatalf("recorded Content-Type = %q, want application/json", end.ReqHeaders["Content-Type"])
+	}
+	rw := httptest.NewRecorder()
+	(&webMux{g: g}).writeActionFixture(rw, &end)
+	if rw.Code != http.StatusOK {
+		t.Fatalf("fixture export status = %d, want 200; body=%s", rw.Code, rw.Body.String())
+	}
+	var fixture Fixture
+	if err := json.Unmarshal(rw.Body.Bytes(), &fixture); err != nil {
+		t.Fatalf("decode fixture: %v", err)
+	}
+	if fixture.Action.HTTP == nil || fixture.Action.HTTP.Body != requestBody {
+		t.Fatalf("fixture HTTP body = %+v, want %q", fixture.Action.HTTP, requestBody)
+	}
+
 	_ = clientTLS.Close()
 	select {
 	case <-done:
 	case <-time.After(2 * time.Second):
 		t.Fatal("gateway did not exit after client close")
 	}
 }
+
+func waitHTTPSAuditEnd(t *testing.T, events <-chan eventPacket, action string) Event {
+	t.Helper()
+	deadline := time.NewTimer(2 * time.Second)
+	defer deadline.Stop()
+	for {
+		select {
+		case pkt := <-events:
+			if pkt.ev.Phase == "end" && pkt.ev.Action == action {
+				return pkt.ev
+			}
+		case <-deadline.C:
+			t.Fatalf("timed out waiting for terminal %q event", action)
+		}
+	}
+}

cmd/clawpatrol/http_body_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"errors"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -229,3 +230,47 @@ func TestBufferHTTPBodyForMatchHonorsCustomCap(t *testing.T) {
 		t.Fatalf("upstream got %d bytes, want %d (custom cap must not drop forwarded bytes)", upstreamLen, len(body))
 	}
 }
+
+func TestBufferHTTPBodyForMatchReadErrorPreservesPartialBodyAndFailsClosed(t *testing.T) {
+	const prefix = `{"partial":true}`
+	wantErr := errors.New("request body failed")
+	req := &http.Request{
+		Body:          &dataThenErrorReader{data: []byte(prefix), err: wantErr},
+		ContentLength: int64(len(prefix) + 10),
+		Header:        make(http.Header),
+	}
+
+	result := bufferHTTPBodyForMatchResult(req, maxHTTPMatchBody)
+	if string(result.body) != prefix {
+		t.Fatalf("match body = %q, want partial prefix %q", result.body, prefix)
+	}
+	if !result.truncated {
+		t.Fatal("truncated = false, want true so body-dependent policy evaluation fails closed")
+	}
+	if result.complete {
+		t.Fatal("complete = true after read error")
+	}
+	if !errors.Is(result.readErr, wantErr) {
+		t.Fatalf("read error = %v, want %v", result.readErr, wantErr)
+	}
+
+	forwarded, err := io.ReadAll(req.Body)
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("restored request body error = %v, want %v", err, wantErr)
+	}
+	if string(forwarded) != prefix {
+		t.Fatalf("restored request body = %q, want %q", forwarded, prefix)
+	}
+
+	ev := Event{}
+	applyTerminalRequestCapture(&ev, req, result, maxHTTPMatchBody)
+	if ev.ReqBodyState != bodyCaptureAborted {
+		t.Fatalf("capture state = %q, want %q", ev.ReqBodyState, bodyCaptureAborted)
+	}
+	if ev.ReqBody != prefix {
+		t.Fatalf("captured request body = %q, want partial prefix %q", ev.ReqBody, prefix)
+	}
+	if ev.ReqSha != "" {
+		t.Fatalf("aborted request SHA = %q, want empty", ev.ReqSha)
+	}
+}

cmd/clawpatrol/http_transform_fixture_test.go

@@ -0,0 +1,191 @@
+package main
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"crypto/tls"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/denoland/clawpatrol/internal/config"
+	"github.com/denoland/clawpatrol/internal/config/runtime"
+)
+
+type uppercaseRequestCredential struct{}
+
+func (uppercaseRequestCredential) InjectHTTP(_ context.Context, req *http.Request, _ runtime.Secret) error {
+	body, err := io.ReadAll(req.Body)
+	if err != nil {
+		return err
+	}
+	body = bytes.ToUpper(body)
+	req.Body = io.NopCloser(bytes.NewReader(body))
+	req.ContentLength = int64(len(body))
+	return nil
+}
+
+func (uppercaseRequestCredential) RewritesHTTPRequest() bool { return true }
+
+type transformFixtureSecretStore struct{}
+
+func (transformFixtureSecretStore) Get(string) (runtime.Secret, error) {
+	return runtime.Secret{Bytes: []byte("unused-test-secret")}, nil
+}
+
+func TestTransformedRequestBodyFixtureIsRejectedAfterPersistence(t *testing.T) {
+	gw, diags := config.LoadBytes([]byte(`
+gateway {
+  state_dir  = "/opt/clawpatrol"
+  public_url = "https://gw.example.test"
+  wireguard { subnet_cidr = "10.55.0.0/24" }
+}
+endpoint "https" "api" {
+  hosts = ["api.example.test"]
+}
+credential "bearer_token" "transform" { endpoint = https.api }
+profile "default" { credentials = [bearer_token.transform] }
+rule "allow-original-body" {
+  endpoint  = https.api
+  priority  = 100
+  condition = "http.body == 'hello world'"
+  verdict   = "allow"
+}
+rule "deny-other-body" {
+  endpoint = https.api
+  priority = -100
+  verdict  = "deny"
+}
+`), "transformed-fixture-test.hcl")
+	if diags.HasErrors() {
+		t.Fatalf("load: %v", diags)
+	}
+	policy, err := config.Compile(gw)
+	if err != nil {
+		t.Fatalf("compile: %v", err)
+	}
+	policy.Credentials["transform"].Body = uppercaseRequestCredential{}
+	ep := policy.Endpoints["api"]
+
+	upstreamBodies := make(chan string, 1)
+	upstream := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		body, err := io.ReadAll(r.Body)
+		if err != nil {
+			t.Errorf("read upstream body: %v", err)
+		}
+		upstreamBodies <- string(body)
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer upstream.Close()
+	upstreamAddr := upstream.Listener.Addr().String()
+	transport := &http.Transport{
+		DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
+			var d net.Dialer
+			return d.DialContext(ctx, network, upstreamAddr)
+		},
+		TLSClientConfig:   &tls.Config{InsecureSkipVerify: true},
+		ForceAttemptHTTP2: false,
+	}
+	defer transport.CloseIdleConnections()
+
+	db, err := OpenDB(filepath.Join(t.TempDir(), "clawpatrol.db"))
+	if err != nil {
+		t.Fatalf("OpenDB: %v", err)
+	}
+	defer func() { _ = db.Close() }()
+	sink, err := NewSink(db, 8)
+	if err != nil {
+		t.Fatalf("NewSink: %v", err)
+	}
+	defer func() {
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		_ = sink.Close(ctx)
+	}()
+	events, cancelEvents := sink.Subscribe()
+	defer cancelEvents()
+
+	certs, _ := inMemoryCertCache(t)
+	g := &Gateway{db: db, certs: certs, sink: sink, secrets: transformFixtureSecretStore{}}
+	g.cfg.Store(gw)
+	g.policy.Store(policy)
+	g.transports.Store(ep, transport)
+
+	serverConn, clientConn := net.Pipe()
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		g.mitmHTTPS(serverConn, "api.example.test", ep)
+	}()
+
+	clientTLS := tls.Client(clientConn, &tls.Config{InsecureSkipVerify: true, ServerName: "api.example.test"})
+	defer func() { _ = clientTLS.Close() }()
+	if err := clientTLS.Handshake(); err != nil {
+		t.Fatalf("client handshake: %v", err)
+	}
+	req, err := http.NewRequest(http.MethodPost, "https://api.example.test/transform", strings.NewReader("hello world"))
+	if err != nil {
+		t.Fatalf("new request: %v", err)
+	}
+	if err := req.Write(clientTLS); err != nil {
+		t.Fatalf("write request: %v", err)
+	}
+	resp, err := http.ReadResponse(bufio.NewReader(clientTLS), req)
+	if err != nil {
+		t.Fatalf("read response: %v", err)
+	}
+	_, _ = io.Copy(io.Discard, resp.Body)
+	_ = resp.Body.Close()
+	if resp.StatusCode != http.StatusNoContent {
+		t.Fatalf("status = %d, want %d; pre-transform body rule should allow", resp.StatusCode, http.StatusNoContent)
+	}
+
+	select {
+	case got := <-upstreamBodies:
+		if got != "HELLO WORLD" {
+			t.Fatalf("upstream body = %q, want transformed body", got)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for upstream body")
+	}
+
+	end := waitHTTPSAuditEnd(t, events, "allow")
+	if end.Rule != "allow-original-body" {
+		t.Fatalf("matched rule = %q, want pre-transform body rule", end.Rule)
+	}
+	if end.ReqBody != "HELLO WORLD" {
+		t.Fatalf("audit body = %q, want transformed body", end.ReqBody)
+	}
+	if !end.ReqTransformed {
+		t.Fatal("live event did not mark the successfully transformed request")
+	}
+
+	stored, err := (&webMux{g: g}).loadAction(end.ID)
+	if err != nil {
+		t.Fatalf("load persisted action: %v", err)
+	}
+	if !stored.ReqTransformed {
+		t.Fatal("persisted event lost the request-transformed flag")
+	}
+	rw := httptest.NewRecorder()
+	(&webMux{g: g}).writeActionFixture(rw, stored)
+	if rw.Code != http.StatusBadRequest {
+		t.Fatalf("fixture export status = %d, want 400; body=%s", rw.Code, rw.Body.String())
+	}
+	if !strings.Contains(rw.Body.String(), "transformed") {
+		t.Fatalf("fixture export error = %q, want transformed-request explanation", rw.Body.String())
+	}
+
+	_ = clientTLS.Close()
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("gateway did not exit after client close")
+	}
+}