From e6bbf8d1fddb7f63f5fc261f6c999f55e5b22f1e Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 15:41:33 +0900 Subject: [PATCH 01/11] fix: make body sampling completion-aware --- cmd/clawpatrol/main.go | 27 +-- cmd/clawpatrol/sampler_test.go | 369 ++++++++++++++++++++++++++++++++- cmd/clawpatrol/web.go | 184 +++++++++++++--- 3 files changed, 533 insertions(+), 47 deletions(-) diff --git a/cmd/clawpatrol/main.go b/cmd/clawpatrol/main.go index 8131e673..7654bcfa 100644 --- a/cmd/clawpatrol/main.go +++ b/cmd/clawpatrol/main.go @@ -2783,7 +2783,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c if trackKind != "" && len(trackedReqBody) > 0 && g.agents != nil { g.preCreateLLMSession(c, trackKind, req.URL.Path, trackedReqBody, sessionHint) } - reqS := newSampler(g.cfg.Load().BodyStorageLimit()) + reqS := newSampler(g.cfg.Load().BodyStorageLimit(), req.ContentLength) if req.Body != nil { req.Body = wrapBodySampler(req.Body, reqS) } @@ -2808,9 +2808,10 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c ev.Action = "error" ev.Reason = err.Error() ev.Ms = time.Since(start).Milliseconds() - ev.ReqSha = reqS.sha() - ev.ReqBody = redactCredentialSample(reqS.sample(req.Header.Get("Content-Encoding")), reqBodySecretRedactions) - ev.In = reqS.n + reqSnapshot := reqS.snapshot(req.Header.Get("Content-Encoding")) + ev.ReqSha = reqSnapshot.sha + ev.ReqBody = redactCredentialSample(reqSnapshot.auditSample(), reqBodySecretRedactions) + ev.In = reqSnapshot.n g.emitEnd(ev) return } @@ -2830,7 +2831,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c resp.Body = io.NopCloser(io.TeeReader(resp.Body, trackBuf)) } } - respS := newSampler(g.cfg.Load().BodyStorageLimit()) + respS := newSampler(g.cfg.Load().BodyStorageLimit(), resp.ContentLength) resp.Body = wrapBodySampler(resp.Body, respS) // Close-delimited responses (no Content-Length, no Transfer- // Encoding) come from h2 upstreams that we forced to http/1.1 @@ -2899,16 +2900,18 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c } ev.Status = strconv.Itoa(resp.StatusCode) ev.ReqHeaders = flatHeadersRedacted(req.Header, reqBodySecretRedactions) - ev.In = reqS.n - ev.Out = respS.n - ev.ReqSha = reqS.sha() - ev.ReqBody = redactCredentialSample(reqS.sample(req.Header.Get("Content-Encoding")), reqBodySecretRedactions) - ev.RespSha = respS.sha() - ev.RespBody = respS.sample(resp.Header.Get("Content-Encoding")) + reqSnapshot := reqS.snapshot(req.Header.Get("Content-Encoding")) + respSnapshot := respS.snapshot(resp.Header.Get("Content-Encoding")) + ev.In = reqSnapshot.n + ev.Out = respSnapshot.n + ev.ReqSha = reqSnapshot.sha + ev.ReqBody = redactCredentialSample(reqSnapshot.auditSample(), reqBodySecretRedactions) + ev.RespSha = respSnapshot.sha + ev.RespBody = respSnapshot.auditSample() ev.Ms = time.Since(start).Milliseconds() g.emitEnd(ev) if g.agents != nil && agentAddr != "" { - g.agents.trackUA(agentAddr, host, req.UserAgent(), reqS.n, respS.n) + g.agents.trackUA(agentAddr, host, req.UserAgent(), reqSnapshot.n, respSnapshot.n) } if writeErr != nil { diff --git a/cmd/clawpatrol/sampler_test.go b/cmd/clawpatrol/sampler_test.go index e9af4538..145e53dc 100644 --- a/cmd/clawpatrol/sampler_test.go +++ b/cmd/clawpatrol/sampler_test.go @@ -5,6 +5,11 @@ import ( "compress/flate" "compress/gzip" "compress/zlib" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "net/http" "strings" "testing" @@ -12,6 +17,71 @@ import ( "github.com/klauspost/compress/zstd" ) +type gatedRequestBody struct { + first []byte + rest []byte + read int + waitingForRest chan struct{} + releaseRest chan struct{} +} + +func (b *gatedRequestBody) Read(p []byte) (int, error) { + switch b.read { + case 0: + b.read++ + return copy(p, b.first), nil + case 1: + b.read++ + close(b.waitingForRest) + <-b.releaseRest + return copy(p, b.rest), nil + default: + return 0, io.EOF + } +} + +func (*gatedRequestBody) Close() error { return nil } + +type earlyResponseRoundTripper struct { + waitingForBody <-chan struct{} + bodyReadDone chan error +} + +type dataThenErrorReader struct { + data []byte + err error +} + +func (r *dataThenErrorReader) Read(p []byte) (int, error) { + if len(r.data) == 0 { + return 0, r.err + } + n := copy(p, r.data) + r.data = r.data[n:] + return n, nil +} + +func (*dataThenErrorReader) Close() error { return nil } + +func (rt earlyResponseRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + go func() { + _, err := io.Copy(io.Discard, req.Body) + rt.bodyReadDone <- err + }() + <-rt.waitingForBody + return &http.Response{ + StatusCode: http.StatusUnauthorized, + Header: make(http.Header), + Body: http.NoBody, + Request: req, + }, nil +} + +func fullBodySHA(body []byte) string { + sum := sha256.Sum256(body) + return hex.EncodeToString(sum[:]) +} + func gzipped(t *testing.T, s string) []byte { t.Helper() var buf bytes.Buffer @@ -85,7 +155,7 @@ func zstdded(t *testing.T, s string) []byte { func TestSamplerSampleGzip(t *testing.T) { want := `{"hello":"world","arr":[1,2,3]}` - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write(gzipped(t, want)) got := s.sample("gzip") if got != want { @@ -95,7 +165,7 @@ func TestSamplerSampleGzip(t *testing.T) { func TestSamplerSampleBrotli(t *testing.T) { want := `{"hello":"world","arr":[1,2,3]}` - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write(brotlied(t, want)) got := s.sample("br") if got != want { @@ -105,7 +175,7 @@ func TestSamplerSampleBrotli(t *testing.T) { func TestSamplerSampleDeflateZlib(t *testing.T) { want := `{"hello":"world"}` - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write(zlibbed(t, want)) got := s.sample("deflate") if got != want { @@ -117,7 +187,7 @@ func TestSamplerSampleDeflateRaw(t *testing.T) { // Some servers send raw deflate under "Content-Encoding: deflate" // despite the RFC requiring zlib framing. want := `{"hello":"world"}` - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write(rawDeflated(t, want)) got := s.sample("deflate") if got != want { @@ -127,7 +197,7 @@ func TestSamplerSampleDeflateRaw(t *testing.T) { func TestSamplerSampleZstd(t *testing.T) { want := `{"hello":"world","arr":[1,2,3]}` - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write(zstdded(t, want)) got := s.sample("zstd") if got != want { @@ -158,7 +228,7 @@ func TestMaybeDecodeCapsExpandedGzip(t *testing.T) { } func TestSamplerSamplePlaintext(t *testing.T) { - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write([]byte(`{"hello":"world"}`)) if got := s.sample(""); got != `{"hello":"world"}` { t.Fatalf("plaintext sample: %q", got) @@ -167,7 +237,7 @@ func TestSamplerSamplePlaintext(t *testing.T) { func TestSamplerSampleBinaryFallback(t *testing.T) { // Raw binary bytes with no encoding header — should hex-prefix. - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write([]byte{0x00, 0xff, 0x01, 0xfe}) got := s.sample("") if !strings.HasPrefix(got, "binary:") { @@ -177,7 +247,7 @@ func TestSamplerSampleBinaryFallback(t *testing.T) { func TestSamplerSampleUnknownEncodingIgnored(t *testing.T) { // Unknown encoding falls through to the printable check on raw bytes. - s := newSampler(4096) + s := newSampler(4096, -1) _, _ = s.Write([]byte{0x1f, 0x8b, 0x08, 0x00}) got := s.sample("compress") if !strings.HasPrefix(got, "binary:") { @@ -187,7 +257,7 @@ func TestSamplerSampleUnknownEncodingIgnored(t *testing.T) { func TestSamplerTruncationMarker(t *testing.T) { // Body fits within the cap: no marker, sample is the full body. - s := newSampler(32) + s := newSampler(32, -1) body := `{"k":"v"}` _, _ = s.Write([]byte(body)) if got := s.sample(""); got != body { @@ -199,7 +269,7 @@ func TestSamplerTruncationMarker(t *testing.T) { // Body exceeds the cap: sample keeps only the prefix and ends with // the truncation marker so the dashboard can flag it. - s = newSampler(8) + s = newSampler(8, -1) big := strings.Repeat("a", 100) _, _ = s.Write([]byte(big)) if !s.truncated() { @@ -214,3 +284,282 @@ func TestSamplerTruncationMarker(t *testing.T) { t.Fatalf("over-cap prefix = %q, want %q", prefix, strings.Repeat("a", 8)) } } + +func TestSamplerCompletesAtDeclaredContentLengthWithoutEOF(t *testing.T) { + body := []byte("complete body") + s := newSampler(4096, int64(len(body))) + rc := wrapBodySampler(io.NopCloser(bytes.NewReader(body)), s) + + buf := make([]byte, len(body)) + n, err := rc.Read(buf) + if err != nil { + t.Fatalf("read body: %v", err) + } + if n != len(body) { + t.Fatalf("read = %d bytes, want %d", n, len(body)) + } + + snap := s.snapshot("") + if snap.state != samplerStateComplete { + t.Fatalf("state = %q, want %q", snap.state, samplerStateComplete) + } + if snap.sha != fullBodySHA(body) { + t.Fatalf("sha = %q, want full-body SHA %q", snap.sha, fullBodySHA(body)) + } + if err := rc.Close(); err != nil { + t.Fatalf("close complete body: %v", err) + } + if state := s.snapshot("").state; state != samplerStateComplete { + t.Fatalf("state after close = %q, want %q", state, samplerStateComplete) + } +} + +func TestSamplerEmptyBodyIsComplete(t *testing.T) { + s := newSampler(4096, 0) + snap := s.snapshot("") + if snap.state != samplerStateComplete { + t.Fatalf("state = %q, want %q", snap.state, samplerStateComplete) + } + if snap.sha != "" { + t.Fatalf("empty body SHA = %q, want empty", snap.sha) + } + if got := snap.auditSample(); got != "" { + t.Fatalf("empty body sample = %q, want empty", got) + } +} + +func TestSamplerCompletesUnknownLengthAtEOF(t *testing.T) { + body := []byte("chunked body") + s := newSampler(4096, -1) + rc := wrapBodySampler(io.NopCloser(bytes.NewReader(body)), s) + + if _, err := io.ReadAll(rc); err != nil { + t.Fatalf("read body: %v", err) + } + + snap := s.snapshot("") + if snap.state != samplerStateComplete { + t.Fatalf("state = %q, want %q", snap.state, samplerStateComplete) + } + if snap.sha != fullBodySHA(body) { + t.Fatalf("sha = %q, want full-body SHA %q", snap.sha, fullBodySHA(body)) + } +} + +func TestSamplerShortEOFAbortsDeclaredBody(t *testing.T) { + body := []byte("short") + s := newSampler(4096, int64(len(body)+10)) + rc := wrapBodySampler(io.NopCloser(bytes.NewReader(body)), s) + + if _, err := io.ReadAll(rc); err != nil { + t.Fatalf("read body: %v", err) + } + + snap := s.snapshot("") + if snap.state != samplerStateAborted { + t.Fatalf("state = %q, want %q after short EOF", snap.state, samplerStateAborted) + } + if snap.sha != "" { + t.Fatalf("short body SHA = %q, want empty", snap.sha) + } +} + +func TestSamplerCloseBeforeCompletionIsAborted(t *testing.T) { + body := []byte("partial") + s := newSampler(4096, int64(len(body)+10)) + rc := wrapBodySampler(io.NopCloser(bytes.NewReader(body)), s) + + buf := make([]byte, len(body)) + if _, err := rc.Read(buf); err != nil { + t.Fatalf("read body: %v", err) + } + if err := rc.Close(); err != nil { + t.Fatalf("close body: %v", err) + } + + snap := s.snapshot("") + if snap.state != samplerStateAborted { + t.Fatalf("state = %q, want %q", snap.state, samplerStateAborted) + } + if snap.sha != "" { + t.Fatalf("aborted body SHA = %q, want empty", snap.sha) + } + if got := snap.auditSample(); !strings.HasSuffix(got, bodyAbortedMarker) { + t.Fatalf("aborted sample = %q, want suffix %q", got, bodyAbortedMarker) + } +} + +func TestSamplerReadErrorBeforeCompletionIsAborted(t *testing.T) { + wantErr := errors.New("request body failed") + s := newSampler(4096, -1) + rc := wrapBodySampler(&dataThenErrorReader{data: []byte("partial"), err: wantErr}, s) + + _, err := io.ReadAll(rc) + if !errors.Is(err, wantErr) { + t.Fatalf("read error = %v, want %v", err, wantErr) + } + + snap := s.snapshot("") + if snap.state != samplerStateAborted { + t.Fatalf("state = %q, want %q", snap.state, samplerStateAborted) + } + if snap.sha != "" { + t.Fatalf("failed body SHA = %q, want empty", snap.sha) + } + if got := snap.auditSample(); !strings.HasSuffix(got, bodyAbortedMarker) { + t.Fatalf("failed sample = %q, want suffix %q", got, bodyAbortedMarker) + } +} + +func TestSamplerAbortedBodyCannotBecomeComplete(t *testing.T) { + s := newSampler(4096, 5) + if _, err := s.Write([]byte("ab")); err != nil { + t.Fatalf("write prefix: %v", err) + } + s.abort() + if _, err := s.Write([]byte("cde")); err != nil { + t.Fatalf("write after abort: %v", err) + } + + snap := s.snapshot("") + if snap.state != samplerStateAborted { + t.Fatalf("state = %q, want terminal %q", snap.state, samplerStateAborted) + } + if snap.sha != "" { + t.Fatalf("aborted body SHA = %q, want empty", snap.sha) + } +} + +func TestSamplerContentLengthOverrunIsAborted(t *testing.T) { + body := []byte("too long") + s := newSampler(4096, int64(len(body)-1)) + if _, err := s.Write(body); err != nil { + t.Fatalf("write body: %v", err) + } + + snap := s.snapshot("") + if snap.state != samplerStateAborted { + t.Fatalf("state = %q, want %q", snap.state, samplerStateAborted) + } + if snap.sha != "" { + t.Fatalf("overrun body SHA = %q, want empty", snap.sha) + } +} + +func TestSamplerWriteAfterCompleteInvalidatesDigest(t *testing.T) { + s := newSampler(4096, 3) + if _, err := s.Write([]byte("one")); err != nil { + t.Fatalf("write declared body: %v", err) + } + if state := s.snapshot("").state; state != samplerStateComplete { + t.Fatalf("state = %q, want %q", state, samplerStateComplete) + } + if _, err := s.Write([]byte("extra")); err != nil { + t.Fatalf("write overrun: %v", err) + } + + snap := s.snapshot("") + if snap.state != samplerStateAborted { + t.Fatalf("state after overrun = %q, want %q", snap.state, samplerStateAborted) + } + if snap.sha != "" { + t.Fatalf("overrun body SHA = %q, want empty", snap.sha) + } +} + +func TestSamplerIncompleteCappedBodyKeepsBothMarkers(t *testing.T) { + s := newSampler(2, 10) + if _, err := s.Write([]byte("partial")); err != nil { + t.Fatalf("write partial body: %v", err) + } + + snap := s.snapshot("") + want := "pa" + bodyIncompleteMarker + bodyTruncatedMarker + if got := snap.auditSample(); got != want { + t.Fatalf("sample = %q, want %q", got, want) + } + if snap.sha != "" { + t.Fatalf("incomplete body SHA = %q, want empty", snap.sha) + } +} + +func TestSamplerCappedCompleteBodyKeepsFullSHA(t *testing.T) { + body := []byte("whole body larger than cap") + s := newSampler(5, int64(len(body))) + rc := wrapBodySampler(io.NopCloser(bytes.NewReader(body)), s) + + if _, err := io.ReadAll(rc); err != nil { + t.Fatalf("read body: %v", err) + } + + snap := s.snapshot("") + if snap.state != samplerStateComplete { + t.Fatalf("state = %q, want %q", snap.state, samplerStateComplete) + } + if snap.sha != fullBodySHA(body) { + t.Fatalf("sha = %q, want full-body SHA %q", snap.sha, fullBodySHA(body)) + } + if got := snap.auditSample(); got != string(body[:5])+bodyTruncatedMarker { + t.Fatalf("sample = %q, want capped prefix plus marker", got) + } +} + +func TestSamplerEarlyResponseDoesNotEmitPartialSHA(t *testing.T) { + first := []byte(`{"prompt":"first half`) + rest := []byte(` and second half"}`) + whole := append(append([]byte(nil), first...), rest...) + body := &gatedRequestBody{ + first: first, + rest: rest, + waitingForRest: make(chan struct{}), + releaseRest: make(chan struct{}), + } + + req, err := http.NewRequest(http.MethodPost, "https://api.example/v1/messages", body) + if err != nil { + t.Fatalf("new request: %v", err) + } + req.ContentLength = int64(len(whole)) + s := newSampler(4096, req.ContentLength) + req.Body = wrapBodySampler(req.Body, s) + bodyReadDone := make(chan error, 1) + rt := earlyResponseRoundTripper{ + waitingForBody: body.waitingForRest, + bodyReadDone: bodyReadDone, + } + + resp, err := rt.RoundTrip(req) + if err != nil { + t.Fatalf("round trip: %v", err) + } + if resp.StatusCode != http.StatusUnauthorized { + t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusUnauthorized) + } + + partial := s.snapshot("") + if partial.state != samplerStatePending { + t.Fatalf("early-response state = %q, want %q", partial.state, samplerStatePending) + } + if partial.sha != "" { + t.Fatalf("early-response SHA = %q, want empty (partial body)", partial.sha) + } + if got := partial.auditSample(); !strings.HasSuffix(got, bodyIncompleteMarker) { + t.Fatalf("early-response sample = %q, want suffix %q", got, bodyIncompleteMarker) + } + + close(body.releaseRest) + if err := <-bodyReadDone; err != nil { + t.Fatalf("finish reading body: %v", err) + } + + complete := s.snapshot("") + if complete.state != samplerStateComplete { + t.Fatalf("finished state = %q, want %q", complete.state, samplerStateComplete) + } + if complete.sha != fullBodySHA(whole) { + t.Fatalf("finished SHA = %q, want full-body SHA %q", complete.sha, fullBodySHA(whole)) + } + if got := complete.auditSample(); got != string(whole) { + t.Fatalf("finished sample = %q, want %q", got, whole) + } +} diff --git a/cmd/clawpatrol/web.go b/cmd/clawpatrol/web.go index 6e5eb92e..b984365e 100644 --- a/cmd/clawpatrol/web.go +++ b/cmd/clawpatrol/web.go @@ -2814,10 +2814,29 @@ func (s *Sink) Subscribe() (<-chan eventPacket, func()) { } type sampler struct { - hash hash.Hash - cap int - buf bytes.Buffer - n int64 + mu sync.Mutex + hash hash.Hash + cap int + contentLength int64 + buf bytes.Buffer + n int64 + state samplerState +} + +type samplerState string + +const ( + samplerStatePending samplerState = "pending" + samplerStateComplete samplerState = "complete" + samplerStateAborted samplerState = "aborted" +) + +type samplerSnapshot struct { + sha string + sample string + n int64 + truncated bool + state samplerState } func unmarshalHeaders(s string, dst *map[string]string) { @@ -2846,12 +2865,25 @@ func flatHeadersRedacted(h http.Header, redactions []string) map[string]string { return out } -func newSampler(capBytes int) *sampler { - return &sampler{hash: sha256.New(), cap: capBytes} +func newSampler(capBytes int, contentLength int64) *sampler { + state := samplerStatePending + if contentLength == 0 { + state = samplerStateComplete + } + return &sampler{ + hash: sha256.New(), + cap: capBytes, + contentLength: contentLength, + state: state, + } } func (s *sampler) Write(p []byte) (int, error) { - s.hash.Write(p) + s.mu.Lock() + defer s.mu.Unlock() + + wasComplete := s.state == samplerStateComplete + _, _ = s.hash.Write(p) s.n += int64(len(p)) if remain := s.cap - s.buf.Len(); remain > 0 { take := len(p) @@ -2860,14 +2892,17 @@ func (s *sampler) Write(p []byte) (int, error) { } s.buf.Write(p[:take]) } - return len(p), nil -} - -func (s *sampler) sha() string { - if s.n == 0 { - return "" + if wasComplete && len(p) > 0 { + s.state = samplerStateAborted + } else if s.contentLength >= 0 { + switch { + case s.n > s.contentLength: + s.state = samplerStateAborted + case s.n == s.contentLength && s.state == samplerStatePending: + s.state = samplerStateComplete + } } - return hex.EncodeToString(s.hash.Sum(nil)) + return len(p), nil } // bodyTruncatedMarker is appended to a persisted body sample when the @@ -2877,9 +2912,22 @@ func (s *sampler) sha() string { // parsing/rendering; see HttpBody in dashboard RequestDetailPage.tsx. const bodyTruncatedMarker = "\n[clawpatrol:body-truncated]" +// These markers make a partial request-body capture explicit in the +// persisted sample. Incomplete means the body was still being read when the +// event was emitted; aborted means it was closed or failed before EOF/full +// Content-Length. A partial body never receives a whole-body SHA. +const ( + bodyIncompleteMarker = "\n[clawpatrol:body-incomplete]" + bodyAbortedMarker = "\n[clawpatrol:body-aborted]" +) + // truncated reports whether the sampler saw more bytes than it kept, // i.e. the persisted sample is a prefix of the real body. -func (s *sampler) truncated() bool { return s.n > int64(s.cap) } +func (s *sampler) truncated() bool { + s.mu.Lock() + defer s.mu.Unlock() + return s.n > int64(s.cap) +} // sample returns the audit-log preview of the captured body. When // encoding names a compression we know how to decode (gzip, br, @@ -2887,16 +2935,44 @@ func (s *sampler) truncated() bool { return s.n > int64(s.cap) } // JSON response doesn't get rendered as "binary:" just because // it's still on the wire compressed. func (s *sampler) sample(encoding string) string { - if s.buf.Len() == 0 { + return s.snapshot(encoding).sample +} + +// snapshot atomically copies every mutable sampler field. Decoding happens +// after the lock is released against the copied prefix. The SHA is populated +// only after EOF or the full declared Content-Length has been observed. +func (s *sampler) snapshot(encoding string) samplerSnapshot { + s.mu.Lock() + n := s.n + state := s.state + capBytes := s.cap + raw := bytes.Clone(s.buf.Bytes()) + var sha string + if state == samplerStateComplete && n > 0 { + sha = hex.EncodeToString(s.hash.Sum(nil)) + } + s.mu.Unlock() + + truncated := n > int64(capBytes) + return samplerSnapshot{ + sha: sha, + sample: sampledBody(raw, truncated, encoding), + n: n, + truncated: truncated, + state: state, + } +} + +func sampledBody(raw []byte, truncated bool, encoding string) string { + if len(raw) == 0 { // An empty buffer with bytes counted means the cap was 0 (or the // body never reached the buffer); still flag truncation so the // dashboard doesn't render a capped body as the full thing. - if s.truncated() { + if truncated { return bodyTruncatedMarker } return "" } - raw := s.buf.Bytes() body := maybeDecode(raw, encoding) var out string if isPrintable(body) { @@ -2904,12 +2980,59 @@ func (s *sampler) sample(encoding string) string { } else { out = "binary:" + hex.EncodeToString(raw[:min(64, len(raw))]) } - if s.truncated() { + if truncated { out += bodyTruncatedMarker } return out } +// auditSample appends the lifecycle state to partial captures while keeping +// the existing cap-truncation marker last for backward-compatible rendering. +func (s samplerSnapshot) auditSample() string { + var marker string + switch s.state { + case samplerStatePending: + marker = bodyIncompleteMarker + case samplerStateAborted: + marker = bodyAbortedMarker + default: + return s.sample + } + if s.truncated && strings.HasSuffix(s.sample, bodyTruncatedMarker) { + body := strings.TrimSuffix(s.sample, bodyTruncatedMarker) + return body + marker + bodyTruncatedMarker + } + return s.sample + marker +} + +func (s *sampler) finishRead(err error) { + if err == nil { + return + } + s.mu.Lock() + defer s.mu.Unlock() + if s.state != samplerStatePending { + return + } + if errors.Is(err, io.EOF) { + if s.contentLength < 0 || s.n == s.contentLength { + s.state = samplerStateComplete + } else { + s.state = samplerStateAborted + } + return + } + s.state = samplerStateAborted +} + +func (s *sampler) abort() { + s.mu.Lock() + defer s.mu.Unlock() + if s.state == samplerStatePending { + s.state = samplerStateAborted + } +} + const ( decodedSampleCap = 4096 decodedSampleTruncatedMarker = "\n[decoded response sample truncated]" @@ -2974,19 +3097,30 @@ func isPrintable(b []byte) bool { return true } -type teeReadCloser struct { - r io.Reader - c io.Closer +type sampledReadCloser struct { + rc io.ReadCloser + s *sampler +} + +func (r *sampledReadCloser) Read(p []byte) (int, error) { + n, err := r.rc.Read(p) + if n > 0 { + _, _ = r.s.Write(p[:n]) + } + r.s.finishRead(err) + return n, err } -func (t teeReadCloser) Read(p []byte) (int, error) { return t.r.Read(p) } -func (t teeReadCloser) Close() error { return t.c.Close() } +func (r *sampledReadCloser) Close() error { + r.s.abort() + return r.rc.Close() +} func wrapBodySampler(rc io.ReadCloser, s *sampler) io.ReadCloser { if rc == nil { return nil } - return teeReadCloser{r: io.TeeReader(rc, s), c: rc} + return &sampledReadCloser{rc: rc, s: s} } // HITL — human-in-the-loop request approval. Rules with `approve = [...]` From 98cc13909ac387f947280c21b1836635617a8e85 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 15:43:12 +0900 Subject: [PATCH 02/11] ui: show partial body captures --- .../src/components/RequestDetailPage.tsx | 41 ++++++++++++----- dashboard/src/lib/bodyCapture.test.ts | 44 +++++++++++++++++++ dashboard/src/lib/bodyCapture.ts | 35 +++++++++++++++ 3 files changed, 108 insertions(+), 12 deletions(-) create mode 100644 dashboard/src/lib/bodyCapture.test.ts create mode 100644 dashboard/src/lib/bodyCapture.ts diff --git a/dashboard/src/components/RequestDetailPage.tsx b/dashboard/src/components/RequestDetailPage.tsx index 7414994a..81773c1d 100644 --- a/dashboard/src/components/RequestDetailPage.tsx +++ b/dashboard/src/components/RequestDetailPage.tsx @@ -9,6 +9,7 @@ import { type FacetSchema, type RulePreview, } from "../lib/api"; +import { splitBodyCapture, type BodyCaptureState } from "../lib/bodyCapture"; import { copyText, headersToJSON } from "../lib/clipboard"; import { formatFacetValue, useFacets } from "../lib/facets"; import { fmtDateTime, statusColorClass } from "../lib/format"; @@ -792,12 +793,6 @@ function parseSSE(text: string): SseEvent[] | null { return events.length > 0 ? events : null; } -// BODY_TRUNCATED_MARKER mirrors bodyTruncatedMarker in cmd/clawpatrol/web.go. -// The gateway appends it to a persisted body sample when the body exceeded -// the actions-table cap. We strip it before parsing/rendering and surface a -// badge so an operator knows they are looking at a prefix, not the whole body. -const BODY_TRUNCATED_MARKER = "\n[clawpatrol:body-truncated]"; - function CapBadge() { return (
@@ -806,14 +801,36 @@ function CapBadge() { ); } +function CaptureStateBadge({ state }: { state: BodyCaptureState }) { + if (state === "complete") return null; + const aborted = state === "aborted"; + return ( +
+ {aborted ? "aborted — body did not finish" : "incomplete — body was still streaming"} +
+ ); +} + +function CaptureBadges({ capped, state }: { capped: boolean; state: BodyCaptureState }) { + return ( +
+ + {capped && } +
+ ); +} + function HttpBody({ text: rawText }: { text: string }) { if (!rawText) return
(empty)
; - const capped = rawText.endsWith(BODY_TRUNCATED_MARKER); - const text = capped ? rawText.slice(0, -BODY_TRUNCATED_MARKER.length) : rawText; + const { text, capped, state } = splitBodyCapture(rawText); if (!text) { return (
- +
); } @@ -821,7 +838,7 @@ function HttpBody({ text: rawText }: { text: string }) { if (result) { return (
- {capped && } + {result.truncated &&
(truncated)
}
@@ -831,7 +848,7 @@ function HttpBody({ text: rawText }: { text: string }) { if (sse) { return (
- {capped && } + {sse.map((e, i) => { const dataJson = tryParseJSON(e.data); return ( @@ -860,7 +877,7 @@ function HttpBody({ text: rawText }: { text: string }) { } return (
- {capped && } + 
{text}
); diff --git a/dashboard/src/lib/bodyCapture.test.ts b/dashboard/src/lib/bodyCapture.test.ts new file mode 100644 index 00000000..b9154d80 --- /dev/null +++ b/dashboard/src/lib/bodyCapture.test.ts @@ -0,0 +1,44 @@ +import { + BODY_ABORTED_MARKER, + BODY_INCOMPLETE_MARKER, + BODY_TRUNCATED_MARKER, + splitBodyCapture, +} from "./bodyCapture.ts"; + +function assertEquals(actual: T, expected: T): void { + if (JSON.stringify(actual) !== JSON.stringify(expected)) { + throw new Error(`actual ${JSON.stringify(actual)} != expected ${JSON.stringify(expected)}`); + } +} + +Deno.test("splitBodyCapture leaves complete bodies unchanged", () => { + assertEquals(splitBodyCapture(`{"ok":true}`), { + text: `{"ok":true}`, + capped: false, + state: "complete", + }); +}); + +Deno.test("splitBodyCapture removes an incomplete marker before parsing", () => { + assertEquals(splitBodyCapture(`{"partial":true}${BODY_INCOMPLETE_MARKER}`), { + text: `{"partial":true}`, + capped: false, + state: "incomplete", + }); +}); + +Deno.test("splitBodyCapture removes an aborted marker before parsing", () => { + assertEquals(splitBodyCapture(`partial${BODY_ABORTED_MARKER}`), { + text: "partial", + capped: false, + state: "aborted", + }); +}); + +Deno.test("splitBodyCapture keeps lifecycle and cap truncation orthogonal", () => { + assertEquals(splitBodyCapture(`partial${BODY_INCOMPLETE_MARKER}${BODY_TRUNCATED_MARKER}`), { + text: "partial", + capped: true, + state: "incomplete", + }); +}); diff --git a/dashboard/src/lib/bodyCapture.ts b/dashboard/src/lib/bodyCapture.ts new file mode 100644 index 00000000..2087d5a3 --- /dev/null +++ b/dashboard/src/lib/bodyCapture.ts @@ -0,0 +1,35 @@ +// These sentinels mirror the persisted body markers in cmd/clawpatrol/web.go. +export const BODY_TRUNCATED_MARKER = "\n[clawpatrol:body-truncated]"; +export const BODY_INCOMPLETE_MARKER = "\n[clawpatrol:body-incomplete]"; +export const BODY_ABORTED_MARKER = "\n[clawpatrol:body-aborted]"; + +export type BodyCaptureState = "complete" | "incomplete" | "aborted"; + +export type BodyCapture = { + text: string; + capped: boolean; + state: BodyCaptureState; +}; + +// splitBodyCapture removes audit metadata before JSON/SSE parsing. The +// lifecycle marker is independent of the storage-cap marker, so both can be +// surfaced when a partial capture also exceeded the configured cap. +export function splitBodyCapture(rawText: string): BodyCapture { + let text = rawText; + let capped = false; + let state: BodyCaptureState = "complete"; + + if (text.endsWith(BODY_TRUNCATED_MARKER)) { + capped = true; + text = text.slice(0, -BODY_TRUNCATED_MARKER.length); + } + if (text.endsWith(BODY_INCOMPLETE_MARKER)) { + state = "incomplete"; + text = text.slice(0, -BODY_INCOMPLETE_MARKER.length); + } else if (text.endsWith(BODY_ABORTED_MARKER)) { + state = "aborted"; + text = text.slice(0, -BODY_ABORTED_MARKER.length); + } + + return { text, capped, state }; +} From 7e3a6dc408c59f57c0670e1f680294032620047b Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 15:59:18 +0900 Subject: [PATCH 03/11] fix: persist body capture state --- cmd/clawpatrol/dev_seed.go | 8 +- cmd/clawpatrol/main.go | 14 +- .../sqlite/0020_action_body_state.sql | 2 + cmd/clawpatrol/sampler_test.go | 241 +++++++++++------- cmd/clawpatrol/web.go | 217 ++++++++++------ cmd/clawpatrol/web_fixture_export_test.go | 33 +++ cmd/clawpatrol/web_sink_test.go | 40 +++ 7 files changed, 367 insertions(+), 188 deletions(-) create mode 100644 cmd/clawpatrol/migrations/sqlite/0020_action_body_state.sql diff --git a/cmd/clawpatrol/dev_seed.go b/cmd/clawpatrol/dev_seed.go index e071caf1..956d28d3 100644 --- a/cmd/clawpatrol/dev_seed.go +++ b/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, diff --git a/cmd/clawpatrol/main.go b/cmd/clawpatrol/main.go index 7654bcfa..899d1d5b 100644 --- a/cmd/clawpatrol/main.go +++ b/cmd/clawpatrol/main.go @@ -2809,9 +2809,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c ev.Reason = err.Error() ev.Ms = time.Since(start).Milliseconds() reqSnapshot := reqS.snapshot(req.Header.Get("Content-Encoding")) - ev.ReqSha = reqSnapshot.sha - ev.ReqBody = redactCredentialSample(reqSnapshot.auditSample(), reqBodySecretRedactions) - ev.In = reqSnapshot.n + applyRequestBodySnapshot(&ev, reqSnapshot, reqBodySecretRedactions) g.emitEnd(ev) return } @@ -2831,7 +2829,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c resp.Body = io.NopCloser(io.TeeReader(resp.Body, trackBuf)) } } - respS := newSampler(g.cfg.Load().BodyStorageLimit(), resp.ContentLength) + respS := newSampler(g.cfg.Load().BodyStorageLimit(), responseBodyContentLength(req.Method, resp)) resp.Body = wrapBodySampler(resp.Body, respS) // Close-delimited responses (no Content-Length, no Transfer- // Encoding) come from h2 upstreams that we forced to http/1.1 @@ -2902,12 +2900,8 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c ev.ReqHeaders = flatHeadersRedacted(req.Header, reqBodySecretRedactions) reqSnapshot := reqS.snapshot(req.Header.Get("Content-Encoding")) respSnapshot := respS.snapshot(resp.Header.Get("Content-Encoding")) - ev.In = reqSnapshot.n - ev.Out = respSnapshot.n - ev.ReqSha = reqSnapshot.sha - ev.ReqBody = redactCredentialSample(reqSnapshot.auditSample(), reqBodySecretRedactions) - ev.RespSha = respSnapshot.sha - ev.RespBody = respSnapshot.auditSample() + applyRequestBodySnapshot(&ev, reqSnapshot, reqBodySecretRedactions) + applyResponseBodySnapshot(&ev, respSnapshot) ev.Ms = time.Since(start).Milliseconds() g.emitEnd(ev) if g.agents != nil && agentAddr != "" { diff --git a/cmd/clawpatrol/migrations/sqlite/0020_action_body_state.sql b/cmd/clawpatrol/migrations/sqlite/0020_action_body_state.sql new file mode 100644 index 00000000..db9bf718 --- /dev/null +++ b/cmd/clawpatrol/migrations/sqlite/0020_action_body_state.sql @@ -0,0 +1,2 @@ +ALTER TABLE actions ADD COLUMN req_body_state TEXT; +ALTER TABLE actions ADD COLUMN resp_body_state TEXT; diff --git a/cmd/clawpatrol/sampler_test.go b/cmd/clawpatrol/sampler_test.go index 145e53dc..e8601993 100644 --- a/cmd/clawpatrol/sampler_test.go +++ b/cmd/clawpatrol/sampler_test.go @@ -8,45 +8,18 @@ import ( "crypto/sha256" "encoding/hex" "errors" + "hash" "io" "net/http" "strings" + "sync" "testing" + "time" "github.com/andybalholm/brotli" "github.com/klauspost/compress/zstd" ) -type gatedRequestBody struct { - first []byte - rest []byte - read int - waitingForRest chan struct{} - releaseRest chan struct{} -} - -func (b *gatedRequestBody) Read(p []byte) (int, error) { - switch b.read { - case 0: - b.read++ - return copy(p, b.first), nil - case 1: - b.read++ - close(b.waitingForRest) - <-b.releaseRest - return copy(p, b.rest), nil - default: - return 0, io.EOF - } -} - -func (*gatedRequestBody) Close() error { return nil } - -type earlyResponseRoundTripper struct { - waitingForBody <-chan struct{} - bodyReadDone chan error -} - type dataThenErrorReader struct { data []byte err error @@ -63,18 +36,28 @@ func (r *dataThenErrorReader) Read(p []byte) (int, error) { func (*dataThenErrorReader) Close() error { return nil } -func (rt earlyResponseRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { - go func() { - _, err := io.Copy(io.Discard, req.Body) - rt.bodyReadDone <- err - }() - <-rt.waitingForBody - return &http.Response{ - StatusCode: http.StatusUnauthorized, - Header: make(http.Header), - Body: http.NoBody, - Request: req, - }, nil +type blockingWriteHash struct { + hash.Hash + entered chan<- struct{} + release <-chan struct{} +} + +func (h *blockingWriteHash) Write(p []byte) (int, error) { + close(h.entered) + <-h.release + return h.Hash.Write(p) +} + +type blockingSumHash struct { + hash.Hash + entered chan<- struct{} + release <-chan struct{} +} + +func (h *blockingSumHash) Sum(b []byte) []byte { + close(h.entered) + <-h.release + return h.Hash.Sum(b) } func fullBodySHA(body []byte) string { @@ -323,9 +306,12 @@ func TestSamplerEmptyBodyIsComplete(t *testing.T) { if snap.sha != "" { t.Fatalf("empty body SHA = %q, want empty", snap.sha) } - if got := snap.auditSample(); got != "" { + if got := snap.sample; got != "" { t.Fatalf("empty body sample = %q, want empty", got) } + if got := snap.captureState(); got != bodyCaptureComplete { + t.Fatalf("capture state = %q, want %q", got, bodyCaptureComplete) + } } func TestSamplerCompletesUnknownLengthAtEOF(t *testing.T) { @@ -384,8 +370,8 @@ func TestSamplerCloseBeforeCompletionIsAborted(t *testing.T) { if snap.sha != "" { t.Fatalf("aborted body SHA = %q, want empty", snap.sha) } - if got := snap.auditSample(); !strings.HasSuffix(got, bodyAbortedMarker) { - t.Fatalf("aborted sample = %q, want suffix %q", got, bodyAbortedMarker) + if got := snap.captureState(); got != bodyCaptureAborted { + t.Fatalf("capture state = %q, want %q", got, bodyCaptureAborted) } } @@ -406,8 +392,8 @@ func TestSamplerReadErrorBeforeCompletionIsAborted(t *testing.T) { if snap.sha != "" { t.Fatalf("failed body SHA = %q, want empty", snap.sha) } - if got := snap.auditSample(); !strings.HasSuffix(got, bodyAbortedMarker) { - t.Fatalf("failed sample = %q, want suffix %q", got, bodyAbortedMarker) + if got := snap.captureState(); got != bodyCaptureAborted { + t.Fatalf("capture state = %q, want %q", got, bodyCaptureAborted) } } @@ -467,17 +453,20 @@ func TestSamplerWriteAfterCompleteInvalidatesDigest(t *testing.T) { } } -func TestSamplerIncompleteCappedBodyKeepsBothMarkers(t *testing.T) { +func TestSamplerIncompleteCappedBodyKeepsStateAndCapMarker(t *testing.T) { s := newSampler(2, 10) if _, err := s.Write([]byte("partial")); err != nil { t.Fatalf("write partial body: %v", err) } snap := s.snapshot("") - want := "pa" + bodyIncompleteMarker + bodyTruncatedMarker - if got := snap.auditSample(); got != want { + want := "pa" + bodyTruncatedMarker + if got := snap.sample; got != want { t.Fatalf("sample = %q, want %q", got, want) } + if got := snap.captureState(); got != bodyCaptureIncomplete { + t.Fatalf("capture state = %q, want %q", got, bodyCaptureIncomplete) + } if snap.sha != "" { t.Fatalf("incomplete body SHA = %q, want empty", snap.sha) } @@ -499,67 +488,133 @@ func TestSamplerCappedCompleteBodyKeepsFullSHA(t *testing.T) { if snap.sha != fullBodySHA(body) { t.Fatalf("sha = %q, want full-body SHA %q", snap.sha, fullBodySHA(body)) } - if got := snap.auditSample(); got != string(body[:5])+bodyTruncatedMarker { + if got := snap.sample; got != string(body[:5])+bodyTruncatedMarker { t.Fatalf("sample = %q, want capped prefix plus marker", got) } } -func TestSamplerEarlyResponseDoesNotEmitPartialSHA(t *testing.T) { - first := []byte(`{"prompt":"first half`) - rest := []byte(` and second half"}`) - whole := append(append([]byte(nil), first...), rest...) - body := &gatedRequestBody{ - first: first, - rest: rest, - waitingForRest: make(chan struct{}), - releaseRest: make(chan struct{}), - } +func TestSamplerEarlyResponseSerializesEventSnapshotWithInFlightWrite(t *testing.T) { + prefix := []byte(`{"prompt":"first half`) + s := newSampler(4096, int64(len(prefix)+20)) + writeEntered := make(chan struct{}) + releaseWrite := make(chan struct{}) + var releaseWriteOnce sync.Once + releaseWriter := func() { releaseWriteOnce.Do(func() { close(releaseWrite) }) } + defer releaseWriter() + s.hash = &blockingWriteHash{Hash: s.hash, entered: writeEntered, release: releaseWrite} - req, err := http.NewRequest(http.MethodPost, "https://api.example/v1/messages", body) - if err != nil { - t.Fatalf("new request: %v", err) + snapshotAttempted := make(chan struct{}) + var snapshotAttemptOnce sync.Once + s.snapshotStartForTest = func() { + snapshotAttemptOnce.Do(func() { close(snapshotAttempted) }) } - req.ContentLength = int64(len(whole)) - s := newSampler(4096, req.ContentLength) - req.Body = wrapBodySampler(req.Body, s) - bodyReadDone := make(chan error, 1) - rt := earlyResponseRoundTripper{ - waitingForBody: body.waitingForRest, - bodyReadDone: bodyReadDone, + + writeDone := make(chan error, 1) + go func() { + _, err := s.Write(prefix) + writeDone <- err + }() + + select { + case <-writeEntered: + case <-time.After(time.Second): + t.Fatal("sampler write did not reach the deterministic gate") } - resp, err := rt.RoundTrip(req) - if err != nil { - t.Fatalf("round trip: %v", err) + eventDone := make(chan Event, 1) + go func() { + snap := s.snapshot("") + ev := Event{} + applyRequestBodySnapshot(&ev, snap, nil) + eventDone <- ev + }() + + select { + case <-snapshotAttempted: + case <-time.After(time.Second): + t.Fatal("event snapshot did not reach the sampler lock") } - if resp.StatusCode != http.StatusUnauthorized { - t.Fatalf("status = %d, want %d", resp.StatusCode, http.StatusUnauthorized) + select { + case <-eventDone: + t.Fatal("event snapshot completed while sampler Write held the lock") + default: } - partial := s.snapshot("") - if partial.state != samplerStatePending { - t.Fatalf("early-response state = %q, want %q", partial.state, samplerStatePending) + releaseWriter() + if err := <-writeDone; err != nil { + t.Fatalf("write prefix: %v", err) } - if partial.sha != "" { - t.Fatalf("early-response SHA = %q, want empty (partial body)", partial.sha) + ev := <-eventDone + if ev.ReqBody != string(prefix) { + t.Fatalf("request sample = %q, want %q", ev.ReqBody, prefix) } - if got := partial.auditSample(); !strings.HasSuffix(got, bodyIncompleteMarker) { - t.Fatalf("early-response sample = %q, want suffix %q", got, bodyIncompleteMarker) + if ev.ReqBodyState != bodyCaptureIncomplete { + t.Fatalf("request capture state = %q, want %q", ev.ReqBodyState, bodyCaptureIncomplete) } + if ev.ReqSha != "" { + t.Fatalf("early-response SHA = %q, want empty", ev.ReqSha) + } +} - close(body.releaseRest) - if err := <-bodyReadDone; err != nil { - t.Fatalf("finish reading body: %v", err) +func TestSamplerSnapshotHoldsLockWhileHashing(t *testing.T) { + body := []byte("complete body") + s := newSampler(4096, int64(len(body))) + if _, err := s.Write(body); err != nil { + t.Fatalf("write body: %v", err) } - complete := s.snapshot("") - if complete.state != samplerStateComplete { - t.Fatalf("finished state = %q, want %q", complete.state, samplerStateComplete) + sumEntered := make(chan struct{}) + releaseSum := make(chan struct{}) + var releaseSumOnce sync.Once + releaseHasher := func() { releaseSumOnce.Do(func() { close(releaseSum) }) } + defer releaseHasher() + s.hash = &blockingSumHash{Hash: s.hash, entered: sumEntered, release: releaseSum} + + snapshotDone := make(chan samplerSnapshot, 1) + go func() { snapshotDone <- s.snapshot("") }() + select { + case <-sumEntered: + case <-time.After(time.Second): + t.Fatal("snapshot did not reach the deterministic hash gate") } - if complete.sha != fullBodySHA(whole) { - t.Fatalf("finished SHA = %q, want full-body SHA %q", complete.sha, fullBodySHA(whole)) + if s.mu.TryLock() { + s.mu.Unlock() + t.Fatal("snapshot hashed without holding the sampler mutex") } - if got := complete.auditSample(); got != string(whole) { - t.Fatalf("finished sample = %q, want %q", got, whole) + + releaseHasher() + snap := <-snapshotDone + if snap.sha != fullBodySHA(body) { + t.Fatalf("SHA = %q, want %q", snap.sha, fullBodySHA(body)) + } +} + +func TestResponseBodyContentLengthUsesHTTPBodySemantics(t *testing.T) { + tests := []struct { + name string + method string + status int + contentLength int64 + body io.ReadCloser + want int64 + }{ + {name: "ordinary", method: http.MethodGet, status: http.StatusOK, contentLength: 12, body: http.NoBody, want: 12}, + {name: "head", method: http.MethodHead, status: http.StatusOK, contentLength: 12, body: http.NoBody, want: 0}, + {name: "informational", method: http.MethodGet, status: http.StatusEarlyHints, contentLength: -1, body: http.NoBody, want: 0}, + {name: "no content", method: http.MethodGet, status: http.StatusNoContent, contentLength: -1, body: http.NoBody, want: 0}, + {name: "not modified", method: http.MethodGet, status: http.StatusNotModified, contentLength: 55, body: http.NoBody, want: 0}, + {name: "explicit no body", method: http.MethodGet, status: http.StatusOK, contentLength: -1, body: http.NoBody, want: 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resp := &http.Response{ + StatusCode: tt.status, + ContentLength: tt.contentLength, + Body: tt.body, + } + if got := responseBodyContentLength(tt.method, resp); got != tt.want { + t.Fatalf("response body content length = %d, want %d", got, tt.want) + } + }) } } diff --git a/cmd/clawpatrol/web.go b/cmd/clawpatrol/web.go index b984365e..eb9848a0 100644 --- a/cmd/clawpatrol/web.go +++ b/cmd/clawpatrol/web.go @@ -1752,36 +1752,38 @@ func (w *webMux) loadAction(actionID string) (*Event, error) { return nil, fmt.Errorf("missing id") } var ( - e Event - tsNs int64 - mode sql.NullString - family sql.NullString - agentIP sql.NullString - method sql.NullString - path sql.NullString - status sql.NullString - in, ot sql.NullInt64 - ms sql.NullInt64 - action sql.NullString - reason sql.NullString - reqSha sql.NullString - respSha sql.NullString - reqBody sql.NullString - respBody sql.NullString - reqHeaders sql.NullString - respHeaders sql.NullString - extra sql.NullString - endpoint sql.NullString - rule sql.NullString - approver sql.NullString - approverType sql.NullString - approverBy sql.NullString + e Event + tsNs int64 + mode sql.NullString + family sql.NullString + agentIP sql.NullString + method sql.NullString + path sql.NullString + status sql.NullString + in, ot sql.NullInt64 + ms sql.NullInt64 + action sql.NullString + reason sql.NullString + reqSha sql.NullString + respSha sql.NullString + reqBody sql.NullString + respBody sql.NullString + reqBodyState sql.NullString + respBodyState sql.NullString + reqHeaders sql.NullString + respHeaders sql.NullString + extra sql.NullString + endpoint sql.NullString + rule sql.NullString + approver sql.NullString + approverType sql.NullString + approverBy sql.NullString ) err := w.g.db.QueryRow(` SELECT 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, approver, approver_type, approver_by @@ -1790,7 +1792,7 @@ func (w *webMux) loadAction(actionID string) (*Event, error) { &tsNs, &mode, &family, &agentIP, &e.Host, &method, &path, &status, &in, &ot, &ms, &action, &reason, &reqSha, &respSha, - &reqBody, &respBody, + &reqBody, &respBody, &reqBodyState, &respBodyState, &reqHeaders, &respHeaders, &extra, &endpoint, &rule, &approver, &approverType, &approverBy, @@ -1815,6 +1817,8 @@ func (w *webMux) loadAction(actionID string) (*Event, error) { e.RespSha = respSha.String e.ReqBody = reqBody.String e.RespBody = respBody.String + e.ReqBodyState = reqBodyState.String + e.RespBodyState = respBodyState.String unmarshalHeaders(reqHeaders.String, &e.ReqHeaders) unmarshalHeaders(respHeaders.String, &e.RespHeaders) if extra.String != "" { @@ -1884,6 +1888,10 @@ func (w *webMux) writeActionFixture(rw http.ResponseWriter, ev *Event) { fx := &Fixture{Match: m, Action: Action{PeerIP: ev.AgentIP}} switch ep.Family { case "http": + if err := validateHTTPFixtureBodyCapture(ev); err != nil { + http.Error(rw, err.Error(), http.StatusBadRequest) + return + } fx.Action.Host = ev.Host fx.Action.HTTP = exportHTTP(ev) case "k8s": @@ -1930,6 +1938,20 @@ func (w *webMux) writeActionFixture(rw http.ResponseWriter, ev *Event) { _ = enc.Encode(fx) } +func validateHTTPFixtureBodyCapture(ev *Event) error { + switch ev.ReqBodyState { + case "", bodyCaptureComplete: + case bodyCaptureIncomplete, bodyCaptureAborted: + return fmt.Errorf("request body capture is %s; cannot export as fixture", ev.ReqBodyState) + default: + return fmt.Errorf("request body capture state %q is not complete; cannot export as fixture", ev.ReqBodyState) + } + if strings.HasSuffix(ev.ReqBody, bodyTruncatedMarker) { + return fmt.Errorf("request body capture is truncated; cannot export as fixture") + } + return nil +} + // matchFromEvent maps post-chain Event.Action onto the fixture's // terminal verdict vocabulary. hitl_* collapses to "approve". // Empty Event.Action maps to "allow" — that's the legacy default @@ -2441,15 +2463,17 @@ type Event struct { // / llm_approver / dashboard), and the approver-specific "By" // string (Slack handle, llm:, ...). All empty for rule- // driven verdicts. - Approver string `json:"approver,omitempty"` - ApproverType string `json:"approver_type,omitempty"` - ApproverBy string `json:"approver_by,omitempty"` - ReqSha string `json:"req_sha,omitempty"` - ReqBody string `json:"req_body,omitempty"` - RespSha string `json:"resp_sha,omitempty"` - RespBody string `json:"resp_body,omitempty"` - ReqHeaders map[string]string `json:"req_headers,omitempty"` - RespHeaders map[string]string `json:"resp_headers,omitempty"` + Approver string `json:"approver,omitempty"` + ApproverType string `json:"approver_type,omitempty"` + ApproverBy string `json:"approver_by,omitempty"` + ReqSha string `json:"req_sha,omitempty"` + ReqBody string `json:"req_body,omitempty"` + ReqBodyState string `json:"req_body_state,omitempty"` + RespSha string `json:"resp_sha,omitempty"` + RespBody string `json:"resp_body,omitempty"` + RespBodyState string `json:"resp_body_state,omitempty"` + ReqHeaders map[string]string `json:"req_headers,omitempty"` + RespHeaders map[string]string `json:"resp_headers,omitempty"` // Frame is set for Phase="frame" only — a single WS frame's text // payload (truncated at sampleCap). Direction is "c→s" or "s→c" // to disambiguate masked client frames from unmasked server frames. @@ -2540,7 +2564,8 @@ func readTailEvents(db *sql.DB, n int) ([]Event, error) { rows, err := db.Query(` SELECT action_id, ts_ns, mode, family, agent_ip, host, method, path, status, bytes_in, bytes_out, - ms, action, reason, req_sha, resp_sha, extra, + ms, action, reason, req_sha, resp_sha, + req_body_state, resp_body_state, extra, endpoint, rule, approver, approver_type, approver_by FROM actions ORDER BY id DESC LIMIT ?`, n) @@ -2551,32 +2576,35 @@ func readTailEvents(db *sql.DB, n int) ([]Event, error) { out := make([]Event, 0, n) for rows.Next() { var ( - e Event - actionID sql.NullString - tsNs int64 - mode sql.NullString - family sql.NullString - agentIP sql.NullString - method sql.NullString - path sql.NullString - status sql.NullString - in, ot sql.NullInt64 - ms sql.NullInt64 - action sql.NullString - reason sql.NullString - reqSha sql.NullString - respSha sql.NullString - extra sql.NullString - endpoint sql.NullString - rule sql.NullString - approver sql.NullString - approverType sql.NullString - approverBy sql.NullString + e Event + actionID sql.NullString + tsNs int64 + mode sql.NullString + family sql.NullString + agentIP sql.NullString + method sql.NullString + path sql.NullString + status sql.NullString + in, ot sql.NullInt64 + ms sql.NullInt64 + action sql.NullString + reason sql.NullString + reqSha sql.NullString + respSha sql.NullString + reqBodyState sql.NullString + respBodyState sql.NullString + extra sql.NullString + endpoint sql.NullString + rule sql.NullString + approver sql.NullString + approverType sql.NullString + approverBy sql.NullString ) if err := rows.Scan( &actionID, &tsNs, &mode, &family, &agentIP, &e.Host, &method, &path, &status, &in, &ot, &ms, - &action, &reason, &reqSha, &respSha, &extra, + &action, &reason, &reqSha, &respSha, + &reqBodyState, &respBodyState, &extra, &endpoint, &rule, &approver, &approverType, &approverBy, ); err != nil { @@ -2599,6 +2627,8 @@ func readTailEvents(db *sql.DB, n int) ([]Event, error) { e.Reason = reason.String e.ReqSha = reqSha.String e.RespSha = respSha.String + e.ReqBodyState = reqBodyState.String + e.RespBodyState = respBodyState.String if extra.String != "" { _ = json.Unmarshal([]byte(extra.String), &e.Facets) } @@ -2702,16 +2732,16 @@ func (s *Sink) drain() { (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, approver, approver_type, approver_by) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) `, e.ID, e.Ts.UnixNano(), e.Mode, e.Family, e.AgentIP, e.Host, e.Method, e.Path, e.Status, e.In, e.Out, e.Ms, e.Action, e.Reason, e.ReqSha, e.RespSha, - e.ReqBody, e.RespBody, + e.ReqBody, e.RespBody, e.ReqBodyState, e.RespBodyState, string(rqhJSON), string(rshJSON), string(extraJSON), e.Endpoint, e.Rule, @@ -2821,6 +2851,9 @@ type sampler struct { buf bytes.Buffer n int64 state samplerState + // snapshotStartForTest is a deterministic test seam used to prove that + // an event snapshot reaches the sampler while Write holds mu. + snapshotStartForTest func() } type samplerState string @@ -2829,6 +2862,10 @@ const ( samplerStatePending samplerState = "pending" samplerStateComplete samplerState = "complete" samplerStateAborted samplerState = "aborted" + + bodyCaptureComplete = "complete" + bodyCaptureIncomplete = "incomplete" + bodyCaptureAborted = "aborted" ) type samplerSnapshot struct { @@ -2878,6 +2915,22 @@ func newSampler(capBytes int, contentLength int64) *sampler { } } +func responseBodyContentLength(method string, resp *http.Response) int64 { + if resp == nil { + return 0 + } + if method == http.MethodHead || + (resp.StatusCode >= 100 && resp.StatusCode <= 199) || + resp.StatusCode == http.StatusNoContent || + resp.StatusCode == http.StatusNotModified { + return 0 + } + if (resp.Body == nil || resp.Body == http.NoBody) && resp.ContentLength <= 0 { + return 0 + } + return resp.ContentLength +} + func (s *sampler) Write(p []byte) (int, error) { s.mu.Lock() defer s.mu.Unlock() @@ -2912,15 +2965,6 @@ func (s *sampler) Write(p []byte) (int, error) { // parsing/rendering; see HttpBody in dashboard RequestDetailPage.tsx. const bodyTruncatedMarker = "\n[clawpatrol:body-truncated]" -// These markers make a partial request-body capture explicit in the -// persisted sample. Incomplete means the body was still being read when the -// event was emitted; aborted means it was closed or failed before EOF/full -// Content-Length. A partial body never receives a whole-body SHA. -const ( - bodyIncompleteMarker = "\n[clawpatrol:body-incomplete]" - bodyAbortedMarker = "\n[clawpatrol:body-aborted]" -) - // truncated reports whether the sampler saw more bytes than it kept, // i.e. the persisted sample is a prefix of the real body. func (s *sampler) truncated() bool { @@ -2942,6 +2986,9 @@ func (s *sampler) sample(encoding string) string { // after the lock is released against the copied prefix. The SHA is populated // only after EOF or the full declared Content-Length has been observed. func (s *sampler) snapshot(encoding string) samplerSnapshot { + if s.snapshotStartForTest != nil { + s.snapshotStartForTest() + } s.mu.Lock() n := s.n state := s.state @@ -2986,23 +3033,29 @@ func sampledBody(raw []byte, truncated bool, encoding string) string { return out } -// auditSample appends the lifecycle state to partial captures while keeping -// the existing cap-truncation marker last for backward-compatible rendering. -func (s samplerSnapshot) auditSample() string { - var marker string +func (s samplerSnapshot) captureState() string { switch s.state { case samplerStatePending: - marker = bodyIncompleteMarker + return bodyCaptureIncomplete case samplerStateAborted: - marker = bodyAbortedMarker + return bodyCaptureAborted default: - return s.sample + return bodyCaptureComplete } - if s.truncated && strings.HasSuffix(s.sample, bodyTruncatedMarker) { - body := strings.TrimSuffix(s.sample, bodyTruncatedMarker) - return body + marker + bodyTruncatedMarker - } - return s.sample + marker +} + +func applyRequestBodySnapshot(ev *Event, snapshot samplerSnapshot, redactions []string) { + ev.In = snapshot.n + ev.ReqSha = snapshot.sha + ev.ReqBody = redactCredentialSample(snapshot.sample, redactions) + ev.ReqBodyState = snapshot.captureState() +} + +func applyResponseBodySnapshot(ev *Event, snapshot samplerSnapshot) { + ev.Out = snapshot.n + ev.RespSha = snapshot.sha + ev.RespBody = snapshot.sample + ev.RespBodyState = snapshot.captureState() } func (s *sampler) finishRead(err error) { diff --git a/cmd/clawpatrol/web_fixture_export_test.go b/cmd/clawpatrol/web_fixture_export_test.go index 54cfe22b..0dffde6b 100644 --- a/cmd/clawpatrol/web_fixture_export_test.go +++ b/cmd/clawpatrol/web_fixture_export_test.go @@ -100,6 +100,39 @@ func TestExporterHTTPSHappyPath(t *testing.T) { } } +func TestExporterRejectsPartialRequestBodyCapture(t *testing.T) { + w := &webMux{g: gatewayWithPolicy(t, fixtureHCL)} + tests := []struct { + name string + state string + body string + wantError string + }{ + {name: "incomplete", state: bodyCaptureIncomplete, body: `{"partial":true}`, wantError: "incomplete"}, + {name: "aborted", state: bodyCaptureAborted, body: `{"partial":true}`, wantError: "aborted"}, + {name: "storage capped", state: bodyCaptureComplete, body: `{"partial":true` + bodyTruncatedMarker, wantError: "truncated"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ev := &Event{ + ID: "partial", Mode: "mitm", Family: "https", + Host: "api.github.com", Method: "POST", Path: "/user", + Action: "allow", Endpoint: "github", + ReqBody: tt.body, ReqBodyState: tt.state, + } + rw := httptest.NewRecorder() + w.writeActionFixture(rw, ev) + + if rw.Code != 400 { + t.Fatalf("status=%d want 400; body=%s", rw.Code, rw.Body.String()) + } + if !strings.Contains(strings.ToLower(rw.Body.String()), tt.wantError) { + t.Fatalf("body=%q want error containing %q", rw.Body.String(), tt.wantError) + } + }) + } +} + // Events recorded before the Endpoint column was populated have // no endpoint to find; the exporter must 400 with a clear reason. func TestExporterRejectsEmptyEndpoint(t *testing.T) { diff --git a/cmd/clawpatrol/web_sink_test.go b/cmd/clawpatrol/web_sink_test.go index 59e19ccc..28a7e175 100644 --- a/cmd/clawpatrol/web_sink_test.go +++ b/cmd/clawpatrol/web_sink_test.go @@ -77,6 +77,46 @@ func TestSinkPersistsGeneratedID(t *testing.T) { } } +func TestSinkPersistsBodyCaptureState(t *testing.T) { + db, err := OpenDB(filepath.Join(t.TempDir(), "clawpatrol.db")) + if err != nil { + t.Fatalf("OpenDB: %v", err) + } + defer func() { _ = db.Close() }() + + s, err := NewSink(db, 4) + if err != nil { + t.Fatalf("NewSink: %v", err) + } + defer close(s.ch) + + ch, cancel := s.Subscribe() + defer cancel() + s.Emit(Event{ + ID: "body-state", Phase: "end", Mode: "mitm", Host: "api.example.com", + ReqBody: "partial request", ReqBodyState: bodyCaptureIncomplete, + RespBody: "complete response", RespBodyState: bodyCaptureComplete, + }) + + select { + case <-ch: + case <-time.After(time.Second): + t.Fatal("timed out waiting for persisted event") + } + + w := &webMux{g: &Gateway{db: db}} + got, err := w.loadAction("body-state") + if err != nil { + t.Fatalf("loadAction: %v", err) + } + if got.ReqBodyState != bodyCaptureIncomplete { + t.Fatalf("request body state = %q, want %q", got.ReqBodyState, bodyCaptureIncomplete) + } + if got.RespBodyState != bodyCaptureComplete { + t.Fatalf("response body state = %q, want %q", got.RespBodyState, bodyCaptureComplete) + } +} + func TestSinkPreservesExistingPersistentEventID(t *testing.T) { s, err := NewSink(nil, 4) if err != nil { From 43e841872a2a32b280bdf272c388a70dad5b9582 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 15:59:25 +0900 Subject: [PATCH 04/11] ui: read structured body state --- .../src/components/RequestDetailPage.tsx | 43 ++++++++++++++----- dashboard/src/lib/api.ts | 2 + dashboard/src/lib/bodyCapture.test.ts | 39 ++++++++++++++--- dashboard/src/lib/bodyCapture.ts | 37 ++++++++++------ 4 files changed, 91 insertions(+), 30 deletions(-) diff --git a/dashboard/src/components/RequestDetailPage.tsx b/dashboard/src/components/RequestDetailPage.tsx index 81773c1d..35a55634 100644 --- a/dashboard/src/components/RequestDetailPage.tsx +++ b/dashboard/src/components/RequestDetailPage.tsx @@ -92,8 +92,10 @@ export function RequestDetailPage({ id, agents }: { id: string; agents: Agent[] const fullUrl = ev.host + (body && !body.startsWith("/") ? " " : "") + body; const facetFields = facetDetailRows(ev, schema); const resultFields = resultDetailRows(ev, schema); - const hasReq = !!ev.req_body; - const hasResp = !!ev.resp_body; + const hasReq = + !!ev.req_body || ev.req_body_state === "incomplete" || ev.req_body_state === "aborted"; + const hasResp = + !!ev.resp_body || ev.resp_body_state === "incomplete" || ev.resp_body_state === "aborted"; const hasReqH = ev.req_headers && Object.keys(ev.req_headers).length > 0; const hasRespH = ev.resp_headers && Object.keys(ev.resp_headers).length > 0; const hasFacets = !isSQL && facetFields.length > 0; @@ -181,9 +183,14 @@ export function RequestDetailPage({ id, agents }: { id: string; agents: Agent[] {hasReq && (
ev.req_body!} />} + action={ + splitBodyCapture(ev.req_body ?? "", ev.req_body_state).text} + /> + } > - +
)} {hasRespH && ( @@ -199,9 +206,14 @@ export function RequestDetailPage({ id, agents }: { id: string; agents: Agent[] {hasResp && (
ev.resp_body!} />} + action={ + splitBodyCapture(ev.resp_body ?? "", ev.resp_body_state).text} + /> + } > - +
)}
@@ -804,13 +816,22 @@ function CapBadge() { function CaptureStateBadge({ state }: { state: BodyCaptureState }) { if (state === "complete") return null; const aborted = state === "aborted"; + const unknown = state === "unknown"; return (
- {aborted ? "aborted — body did not finish" : "incomplete — body was still streaming"} + {aborted + ? "aborted — body did not finish" + : unknown + ? "completion unknown — legacy capture" + : "incomplete — body was still streaming"}
); } @@ -824,13 +845,13 @@ function CaptureBadges({ capped, state }: { capped: boolean; state: BodyCaptureS ); } -function HttpBody({ text: rawText }: { text: string }) { - if (!rawText) return
(empty)
; - const { text, capped, state } = splitBodyCapture(rawText); +function HttpBody({ text: rawText, state: storedState }: { text: string; state?: string }) { + const { text, capped, state } = splitBodyCapture(rawText, storedState); if (!text) { return (
+ {!capped && state === "complete" && "(empty)"}
); } diff --git a/dashboard/src/lib/api.ts b/dashboard/src/lib/api.ts index 7d5eb680..5b565570 100644 --- a/dashboard/src/lib/api.ts +++ b/dashboard/src/lib/api.ts @@ -541,6 +541,8 @@ export type EventRecord = { resp_sha?: string; req_body?: string; resp_body?: string; + req_body_state?: "complete" | "incomplete" | "aborted"; + resp_body_state?: "complete" | "incomplete" | "aborted"; req_headers?: Record; resp_headers?: Record; // family identifies which facet plugin emitted this event; facets diff --git a/dashboard/src/lib/bodyCapture.test.ts b/dashboard/src/lib/bodyCapture.test.ts index b9154d80..b4e15dbb 100644 --- a/dashboard/src/lib/bodyCapture.test.ts +++ b/dashboard/src/lib/bodyCapture.test.ts @@ -12,33 +12,58 @@ function assertEquals(actual: T, expected: T): void { } Deno.test("splitBodyCapture leaves complete bodies unchanged", () => { - assertEquals(splitBodyCapture(`{"ok":true}`), { + assertEquals(splitBodyCapture(`{"ok":true}`, "complete"), { text: `{"ok":true}`, capped: false, state: "complete", }); }); -Deno.test("splitBodyCapture removes an incomplete marker before parsing", () => { - assertEquals(splitBodyCapture(`{"partial":true}${BODY_INCOMPLETE_MARKER}`), { +Deno.test("splitBodyCapture uses structured incomplete state", () => { + assertEquals(splitBodyCapture(`{"partial":true}`, "incomplete"), { text: `{"partial":true}`, capped: false, state: "incomplete", }); }); -Deno.test("splitBodyCapture removes an aborted marker before parsing", () => { - assertEquals(splitBodyCapture(`partial${BODY_ABORTED_MARKER}`), { +Deno.test("splitBodyCapture uses structured aborted state", () => { + assertEquals(splitBodyCapture("partial", "aborted"), { text: "partial", capped: false, state: "aborted", }); }); -Deno.test("splitBodyCapture keeps lifecycle and cap truncation orthogonal", () => { - assertEquals(splitBodyCapture(`partial${BODY_INCOMPLETE_MARKER}${BODY_TRUNCATED_MARKER}`), { +Deno.test("splitBodyCapture keeps structured state and cap truncation orthogonal", () => { + assertEquals(splitBodyCapture(`partial${BODY_TRUNCATED_MARKER}`, "incomplete"), { text: "partial", capped: true, state: "incomplete", }); }); + +Deno.test("splitBodyCapture marks legacy rows with no state as unknown", () => { + assertEquals(splitBodyCapture("legacy body"), { + text: "legacy body", + capped: false, + state: "unknown", + }); +}); + +Deno.test("splitBodyCapture understands legacy lifecycle markers", () => { + assertEquals(splitBodyCapture(`partial${BODY_ABORTED_MARKER}${BODY_TRUNCATED_MARKER}`), { + text: "partial", + capped: true, + state: "aborted", + }); +}); + +Deno.test("structured state prevents lifecycle-marker suffix ambiguity", () => { + const text = `literal${BODY_INCOMPLETE_MARKER}`; + assertEquals(splitBodyCapture(text, "complete"), { + text, + capped: false, + state: "complete", + }); +}); diff --git a/dashboard/src/lib/bodyCapture.ts b/dashboard/src/lib/bodyCapture.ts index 2087d5a3..44f6b3b3 100644 --- a/dashboard/src/lib/bodyCapture.ts +++ b/dashboard/src/lib/bodyCapture.ts @@ -3,7 +3,7 @@ export const BODY_TRUNCATED_MARKER = "\n[clawpatrol:body-truncated]"; export const BODY_INCOMPLETE_MARKER = "\n[clawpatrol:body-incomplete]"; export const BODY_ABORTED_MARKER = "\n[clawpatrol:body-aborted]"; -export type BodyCaptureState = "complete" | "incomplete" | "aborted"; +export type BodyCaptureState = "complete" | "incomplete" | "aborted" | "unknown"; export type BodyCapture = { text: string; @@ -11,25 +11,38 @@ export type BodyCapture = { state: BodyCaptureState; }; -// splitBodyCapture removes audit metadata before JSON/SSE parsing. The -// lifecycle marker is independent of the storage-cap marker, so both can be -// surfaced when a partial capture also exceeded the configured cap. -export function splitBodyCapture(rawText: string): BodyCapture { +// splitBodyCapture removes audit metadata before JSON/SSE parsing. New rows +// carry lifecycle state in dedicated columns; marker parsing remains only for +// rows written by versions that embedded lifecycle state in the body text. +export function splitBodyCapture(rawText: string, storedState?: string): BodyCapture { let text = rawText; let capped = false; - let state: BodyCaptureState = "complete"; + let state: BodyCaptureState = normalizeBodyCaptureState(storedState); if (text.endsWith(BODY_TRUNCATED_MARKER)) { capped = true; text = text.slice(0, -BODY_TRUNCATED_MARKER.length); } - if (text.endsWith(BODY_INCOMPLETE_MARKER)) { - state = "incomplete"; - text = text.slice(0, -BODY_INCOMPLETE_MARKER.length); - } else if (text.endsWith(BODY_ABORTED_MARKER)) { - state = "aborted"; - text = text.slice(0, -BODY_ABORTED_MARKER.length); + if (!storedState) { + if (text.endsWith(BODY_INCOMPLETE_MARKER)) { + state = "incomplete"; + text = text.slice(0, -BODY_INCOMPLETE_MARKER.length); + } else if (text.endsWith(BODY_ABORTED_MARKER)) { + state = "aborted"; + text = text.slice(0, -BODY_ABORTED_MARKER.length); + } } return { text, capped, state }; } + +function normalizeBodyCaptureState(state?: string): BodyCaptureState { + switch (state) { + case "complete": + case "incomplete": + case "aborted": + return state; + default: + return "unknown"; + } +} From 9f56d81938073dd28d748f021343feed6ab6cbf0 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 16:04:18 +0900 Subject: [PATCH 05/11] fix: reject unknown body fixtures --- cmd/clawpatrol/web.go | 16 ++++++++++++- cmd/clawpatrol/web_fixture_export_test.go | 28 +++++++++++++---------- 2 files changed, 31 insertions(+), 13 deletions(-) diff --git a/cmd/clawpatrol/web.go b/cmd/clawpatrol/web.go index eb9848a0..2f864dbc 100644 --- a/cmd/clawpatrol/web.go +++ b/cmd/clawpatrol/web.go @@ -1940,9 +1940,18 @@ func (w *webMux) writeActionFixture(rw http.ResponseWriter, ev *Event) { func validateHTTPFixtureBodyCapture(ev *Event) error { switch ev.ReqBodyState { - case "", bodyCaptureComplete: + case bodyCaptureComplete: case bodyCaptureIncomplete, bodyCaptureAborted: return fmt.Errorf("request body capture is %s; cannot export as fixture", ev.ReqBodyState) + case "": + switch { + case strings.HasSuffix(ev.ReqBody, legacyBodyIncompleteMarker): + return fmt.Errorf("request body capture is incomplete; cannot export as fixture") + case strings.HasSuffix(ev.ReqBody, legacyBodyAbortedMarker): + return fmt.Errorf("request body capture is aborted; cannot export as fixture") + default: + return fmt.Errorf("request body capture completion is unknown; cannot export as fixture") + } default: return fmt.Errorf("request body capture state %q is not complete; cannot export as fixture", ev.ReqBodyState) } @@ -2965,6 +2974,11 @@ func (s *sampler) Write(p []byte) (int, error) { // parsing/rendering; see HttpBody in dashboard RequestDetailPage.tsx. const bodyTruncatedMarker = "\n[clawpatrol:body-truncated]" +const ( + legacyBodyIncompleteMarker = "\n[clawpatrol:body-incomplete]" + legacyBodyAbortedMarker = "\n[clawpatrol:body-aborted]" +) + // truncated reports whether the sampler saw more bytes than it kept, // i.e. the persisted sample is a prefix of the real body. func (s *sampler) truncated() bool { diff --git a/cmd/clawpatrol/web_fixture_export_test.go b/cmd/clawpatrol/web_fixture_export_test.go index 0dffde6b..6f8b26a5 100644 --- a/cmd/clawpatrol/web_fixture_export_test.go +++ b/cmd/clawpatrol/web_fixture_export_test.go @@ -53,16 +53,17 @@ profile "default" { credentials = [bearer_token.tok] } func TestExporterHTTPSHappyPath(t *testing.T) { w := &webMux{g: gatewayWithPolicy(t, fixtureHCL)} ev := &Event{ - ID: "evt-1", - Mode: "mitm", - Family: "https", - Host: "api.github.com", - Method: "GET", - Path: "/user", - AgentIP: "100.64.0.7", - Action: "allow", - Endpoint: "github", - Rule: "github-reads", + ID: "evt-1", + Mode: "mitm", + Family: "https", + Host: "api.github.com", + Method: "GET", + Path: "/user", + AgentIP: "100.64.0.7", + Action: "allow", + Endpoint: "github", + Rule: "github-reads", + ReqBodyState: bodyCaptureComplete, ReqHeaders: map[string]string{ "Authorization": "***", "User-Agent": "clawpatrol-test", @@ -111,6 +112,9 @@ func TestExporterRejectsPartialRequestBodyCapture(t *testing.T) { {name: "incomplete", state: bodyCaptureIncomplete, body: `{"partial":true}`, wantError: "incomplete"}, {name: "aborted", state: bodyCaptureAborted, body: `{"partial":true}`, wantError: "aborted"}, {name: "storage capped", state: bodyCaptureComplete, body: `{"partial":true` + bodyTruncatedMarker, wantError: "truncated"}, + {name: "unknown", body: `{"legacy":true}`, wantError: "unknown"}, + {name: "legacy incomplete marker", body: `{"partial":true}` + legacyBodyIncompleteMarker, wantError: "incomplete"}, + {name: "legacy aborted marker", body: `{"partial":true}` + legacyBodyAbortedMarker, wantError: "aborted"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -168,7 +172,7 @@ profile "default" { credentials = [bearer_token.a, bearer_token.b] } ev := &Event{ ID: "evt-3", Mode: "mitm", Family: "https", Host: "api.example.com", Method: "GET", Path: "/x", - Action: "allow", Endpoint: "beta", Rule: "", + Action: "allow", Endpoint: "beta", Rule: "", ReqBodyState: bodyCaptureComplete, } rw := httptest.NewRecorder() w.writeActionFixture(rw, ev) @@ -444,7 +448,7 @@ profile "default" { credentials = [bearer_token.tok] } ID: "evt-rt", Mode: "mitm", Family: "https", Host: "api.github.com", Method: "GET", Path: "/user", AgentIP: "100.64.0.7", Action: "allow", - Endpoint: "github", Rule: "reads", + Endpoint: "github", Rule: "reads", ReqBodyState: bodyCaptureComplete, } rw := httptest.NewRecorder() w.writeActionFixture(rw, ev) From c9f4789f43d3c51add0ce467b4e91e6e20311ac6 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 17:13:44 +0900 Subject: [PATCH 06/11] test: make sampler lock check deterministic --- cmd/clawpatrol/sampler_test.go | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/cmd/clawpatrol/sampler_test.go b/cmd/clawpatrol/sampler_test.go index e8601993..5f283993 100644 --- a/cmd/clawpatrol/sampler_test.go +++ b/cmd/clawpatrol/sampler_test.go @@ -520,6 +520,10 @@ func TestSamplerEarlyResponseSerializesEventSnapshotWithInFlightWrite(t *testing case <-time.After(time.Second): t.Fatal("sampler write did not reach the deterministic gate") } + if s.mu.TryLock() { + s.mu.Unlock() + t.Fatal("sampler Write reached hash.Write without holding the sampler mutex") + } eventDone := make(chan Event, 1) go func() { @@ -534,12 +538,6 @@ func TestSamplerEarlyResponseSerializesEventSnapshotWithInFlightWrite(t *testing case <-time.After(time.Second): t.Fatal("event snapshot did not reach the sampler lock") } - select { - case <-eventDone: - t.Fatal("event snapshot completed while sampler Write held the lock") - default: - } - releaseWriter() if err := <-writeDone; err != nil { t.Fatalf("write prefix: %v", err) From cfdddd21c58ba9ffde19774acbbdf1db4d8064a8 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 17:13:49 +0900 Subject: [PATCH 07/11] fix: capture denied request bodies --- cmd/clawpatrol/credential_match_test.go | 15 ++++++++ cmd/clawpatrol/hitl_body_sample_test.go | 43 ++++++++++++++++++++++ cmd/clawpatrol/main.go | 49 +++++++++++++++++++++---- 3 files changed, 100 insertions(+), 7 deletions(-) diff --git a/cmd/clawpatrol/credential_match_test.go b/cmd/clawpatrol/credential_match_test.go index d436f81c..c0fb565e 100644 --- a/cmd/clawpatrol/credential_match_test.go +++ b/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()) + } }) } diff --git a/cmd/clawpatrol/hitl_body_sample_test.go b/cmd/clawpatrol/hitl_body_sample_test.go index 227abe28..bac314b8 100644 --- a/cmd/clawpatrol/hitl_body_sample_test.go +++ b/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,6 +119,29 @@ 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: @@ -122,3 +149,19 @@ profile "default" { credentials = [bearer_token.pat] } 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) + } + } +} diff --git a/cmd/clawpatrol/main.go b/cmd/clawpatrol/main.go index 899d1d5b..f7d63151 100644 --- a/cmd/clawpatrol/main.go +++ b/cmd/clawpatrol/main.go @@ -2250,8 +2250,7 @@ func (w *countWriter) Write(p []byte) (int, error) { const maxHTTPMatchBody = int(config.DefaultBodyBufferLimit) func bufferHTTPBodyForMatch(req *http.Request, capBytes int) []byte { - b, _ := bufferHTTPBodyForMatchTruncated(req, capBytes) - return b + return bufferHTTPBodyForMatchResult(req, capBytes).body } // bufferHTTPBodyForMatchTruncated is bufferHTTPBodyForMatch with the @@ -2263,12 +2262,24 @@ func bufferHTTPBodyForMatch(req *http.Request, capBytes int) []byte { // http.body / http.body_json become CEL unknowns and rules whose // outcome depends on them fail-close. func bufferHTTPBodyForMatchTruncated(req *http.Request, capBytes int) (body []byte, truncated bool) { + result := bufferHTTPBodyForMatchResult(req, capBytes) + return result.body, result.truncated +} + +type bufferedHTTPBodyResult struct { + body []byte + truncated bool + complete bool + readErr error +} + +func bufferHTTPBodyForMatchResult(req *http.Request, capBytes int) bufferedHTTPBodyResult { if req.Body == nil { - return nil, false + return bufferedHTTPBodyResult{complete: true} } b, err := io.ReadAll(io.LimitReader(req.Body, int64(capBytes)+1)) if err != nil { - return nil, false + return bufferedHTTPBodyResult{readErr: err} } if len(b) > capBytes { // Pulled one byte past the cap — body is over-sized. Keep @@ -2276,14 +2287,33 @@ func bufferHTTPBodyForMatchTruncated(req *http.Request, capBytes int) (body []by // full read (including the probe byte) in front of the // remaining stream so the upstream forward stays byte-exact. req.Body = io.NopCloser(io.MultiReader(bytes.NewReader(b), req.Body)) - return b[:capBytes], true + return bufferedHTTPBodyResult{body: b[:capBytes], truncated: true} } // Body fit inside the cap (or was exactly cap bytes). Re-attach // what we read — req.Body may still hold bytes past it on a // chunked / unknown-length stream that just hadn't surfaced // before the ReadAll returned. req.Body = io.NopCloser(io.MultiReader(bytes.NewReader(b), req.Body)) - return b, false + return bufferedHTTPBodyResult{body: b, complete: true} +} + +func applyTerminalRequestCapture(ev *Event, req *http.Request, buffered bufferedHTTPBodyResult, capBytes int) { + contentLength := req.ContentLength + if buffered.truncated { + // The matcher retained only a prefix, so equality with a declared + // length cannot prove that this audit sample is the whole body. + contentLength = -1 + } + s := newSampler(capBytes, contentLength) + _, _ = s.Write(buffered.body) + switch { + case buffered.readErr != nil: + s.finishRead(buffered.readErr) + case buffered.complete: + s.finishRead(io.EOF) + } + applyRequestBodySnapshot(ev, s.snapshot(req.Header.Get("Content-Encoding")), nil) + ev.ReqHeaders = flatHeaders(req.Header) } // mitmHTTPS handles an SNI-matched TLS connection for an HTTPS-family @@ -2352,11 +2382,14 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c // unbuffered (rare for agent traffic) but surface as // Truncated=true so the dispatcher/retry relay can fail-close // any path that needed the complete body. + var bufferedBody bufferedHTTPBodyResult var matchBody []byte var truncated bool retryOperationID := strings.TrimSpace(req.Header.Get(hitlRetryOperationHeader)) if req.Method == "POST" || req.Method == "PUT" || req.Method == "PATCH" || retryOperationID != "" { - matchBody, truncated = bufferHTTPBodyForMatchTruncated(req, g.cfg.Load().BodyBufferLimit()) + bufferedBody = bufferHTTPBodyForMatchResult(req, g.cfg.Load().BodyBufferLimit()) + matchBody = bufferedBody.body + truncated = bufferedBody.truncated } mreq := &match.Request{ @@ -2533,6 +2566,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c ev.ApproverBy = v.By ev.Reason = reason ev.Ms = time.Since(start).Milliseconds() + applyTerminalRequestCapture(&ev, req, bufferedBody, g.cfg.Load().BodyStorageLimit()) g.emitEnd(ev) return } @@ -2568,6 +2602,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c ev.Action = "deny" ev.Reason = reason ev.Ms = time.Since(start).Milliseconds() + applyTerminalRequestCapture(&ev, req, bufferedBody, g.cfg.Load().BodyStorageLimit()) g.emitEnd(ev) return } From a4b773013f091ab8a20184855af3a6cf1d64b647 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 17:13:53 +0900 Subject: [PATCH 08/11] fix: reject lossy body fixtures --- cmd/clawpatrol/sampler_test.go | 9 ++++++++ cmd/clawpatrol/web.go | 25 +++++++++++++++++++++++ cmd/clawpatrol/web_fixture_export_test.go | 6 +++++- 3 files changed, 39 insertions(+), 1 deletion(-) diff --git a/cmd/clawpatrol/sampler_test.go b/cmd/clawpatrol/sampler_test.go index 5f283993..cd089936 100644 --- a/cmd/clawpatrol/sampler_test.go +++ b/cmd/clawpatrol/sampler_test.go @@ -228,6 +228,15 @@ func TestSamplerSampleBinaryFallback(t *testing.T) { } } +func TestSamplerSampleInvalidUTF8UsesBinaryFallback(t *testing.T) { + s := newSampler(4096, 2) + _, _ = s.Write([]byte{0xff, 0xfe}) + got := s.sample("") + if !strings.HasPrefix(got, "binary:") { + t.Fatalf("expected binary: prefix for invalid UTF-8, got %q", got) + } +} + func TestSamplerSampleUnknownEncodingIgnored(t *testing.T) { // Unknown encoding falls through to the printable check on raw bytes. s := newSampler(4096, -1) diff --git a/cmd/clawpatrol/web.go b/cmd/clawpatrol/web.go index 2f864dbc..10429209 100644 --- a/cmd/clawpatrol/web.go +++ b/cmd/clawpatrol/web.go @@ -28,6 +28,7 @@ import ( "sync" "sync/atomic" "time" + "unicode/utf8" "github.com/andybalholm/brotli" "github.com/klauspost/compress/zstd" @@ -1958,9 +1959,30 @@ func validateHTTPFixtureBodyCapture(ev *Event) error { if strings.HasSuffix(ev.ReqBody, bodyTruncatedMarker) { return fmt.Errorf("request body capture is truncated; cannot export as fixture") } + if encoding := eventHeaderValue(ev.ReqHeaders, "Content-Encoding"); encoding != "" && !strings.EqualFold(strings.TrimSpace(encoding), "identity") { + return fmt.Errorf("request body capture is content-encoded; cannot export as fixture") + } + if strings.HasPrefix(ev.ReqBody, "binary:") { + return fmt.Errorf("request body capture is a binary preview; cannot export as fixture") + } + if strings.HasSuffix(ev.ReqBody, decodedSampleTruncatedMarker) { + return fmt.Errorf("request body decoded preview is truncated; cannot export as fixture") + } + if !utf8.ValidString(ev.ReqBody) { + return fmt.Errorf("request body capture is not valid UTF-8; cannot export as fixture") + } return nil } +func eventHeaderValue(headers map[string]string, name string) string { + for key, value := range headers { + if strings.EqualFold(key, name) { + return value + } + } + return "" +} + // matchFromEvent maps post-chain Event.Action onto the fixture's // terminal verdict vocabulary. hitl_* collapses to "approve". // Empty Event.Action maps to "allow" — that's the legacy default @@ -3156,6 +3178,9 @@ func maybeDecode(buf []byte, encoding string) []byte { } func isPrintable(b []byte) bool { + if !utf8.Valid(b) { + return false + } for _, x := range b { if x == 0 || (x < 0x20 && x != '\n' && x != '\r' && x != '\t') { return false diff --git a/cmd/clawpatrol/web_fixture_export_test.go b/cmd/clawpatrol/web_fixture_export_test.go index 6f8b26a5..36c1bd13 100644 --- a/cmd/clawpatrol/web_fixture_export_test.go +++ b/cmd/clawpatrol/web_fixture_export_test.go @@ -107,6 +107,7 @@ func TestExporterRejectsPartialRequestBodyCapture(t *testing.T) { name string state string body string + headers map[string]string wantError string }{ {name: "incomplete", state: bodyCaptureIncomplete, body: `{"partial":true}`, wantError: "incomplete"}, @@ -115,6 +116,9 @@ func TestExporterRejectsPartialRequestBodyCapture(t *testing.T) { {name: "unknown", body: `{"legacy":true}`, wantError: "unknown"}, {name: "legacy incomplete marker", body: `{"partial":true}` + legacyBodyIncompleteMarker, wantError: "incomplete"}, {name: "legacy aborted marker", body: `{"partial":true}` + legacyBodyAbortedMarker, wantError: "aborted"}, + {name: "content encoded", state: bodyCaptureComplete, body: `{"decoded":true}`, headers: map[string]string{"content-encoding": "gzip"}, wantError: "content-encoded"}, + {name: "binary preview", state: bodyCaptureComplete, body: "binary:00ff", wantError: "binary"}, + {name: "decoded preview capped", state: bodyCaptureComplete, body: "expanded" + decodedSampleTruncatedMarker, wantError: "decoded preview is truncated"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -122,7 +126,7 @@ func TestExporterRejectsPartialRequestBodyCapture(t *testing.T) { ID: "partial", Mode: "mitm", Family: "https", Host: "api.github.com", Method: "POST", Path: "/user", Action: "allow", Endpoint: "github", - ReqBody: tt.body, ReqBodyState: tt.state, + ReqBody: tt.body, ReqBodyState: tt.state, ReqHeaders: tt.headers, } rw := httptest.NewRecorder() w.writeActionFixture(rw, ev) From eef82de54b75473e9b5093de43ffdab34300b0d3 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 17:22:34 +0900 Subject: [PATCH 09/11] fix: fail closed on request body read errors --- cmd/clawpatrol/http_body_test.go | 45 ++++++++++++++++++++++++++++++++ cmd/clawpatrol/main.go | 24 +++++++++++------ 2 files changed, 61 insertions(+), 8 deletions(-) diff --git a/cmd/clawpatrol/http_body_test.go b/cmd/clawpatrol/http_body_test.go index 195582f2..9d9fb063 100644 --- a/cmd/clawpatrol/http_body_test.go +++ b/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) + } +} diff --git a/cmd/clawpatrol/main.go b/cmd/clawpatrol/main.go index f7d63151..17ac0455 100644 --- a/cmd/clawpatrol/main.go +++ b/cmd/clawpatrol/main.go @@ -2254,13 +2254,12 @@ func bufferHTTPBodyForMatch(req *http.Request, capBytes int) []byte { } // bufferHTTPBodyForMatchTruncated is bufferHTTPBodyForMatch with the -// overflow signal exposed: it reads one byte past the cap to detect -// truncation, then re-attaches whatever it pulled (cap + 1 byte) in -// front of the original stream so upstream still receives the body -// byte-for-byte. truncated is true iff the body extended beyond -// maxHTTPMatchBody; callers stash this on match.Request.Truncated so -// http.body / http.body_json become CEL unknowns and rules whose -// outcome depends on them fail-close. +// incomplete-body signal exposed. It reads one byte past the cap and +// re-attaches whatever it pulled in front of the original stream so upstream +// still receives those bytes. truncated is true when the body exceeds the cap +// or cannot be read to EOF; callers stash this on match.Request.Truncated so +// http.body / http.body_json become CEL unknowns and rules whose outcome +// depends on them fail-close. func bufferHTTPBodyForMatchTruncated(req *http.Request, capBytes int) (body []byte, truncated bool) { result := bufferHTTPBodyForMatchResult(req, capBytes) return result.body, result.truncated @@ -2279,7 +2278,16 @@ func bufferHTTPBodyForMatchResult(req *http.Request, capBytes int) bufferedHTTPB } b, err := io.ReadAll(io.LimitReader(req.Body, int64(capBytes)+1)) if err != nil { - return bufferedHTTPBodyResult{readErr: err} + // Preserve bytes returned alongside the error for both the audit trail + // and any later attempt to forward the request. The matcher must still + // treat the body as incomplete rather than a known prefix (or empty + // body), so surface the same fail-closed signal used for capped input. + req.Body = io.NopCloser(io.MultiReader(bytes.NewReader(b), req.Body)) + body := b + if len(body) > capBytes { + body = body[:capBytes] + } + return bufferedHTTPBodyResult{body: body, truncated: true, readErr: err} } if len(b) > capBytes { // Pulled one byte past the cap — body is over-sized. Keep From ad2cd62dd8ee636cf4f7c0fbb27fd9cde38142da Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 17:22:38 +0900 Subject: [PATCH 10/11] fix: reject redacted body fixtures --- cmd/clawpatrol/telegram_request_body_redaction_test.go | 8 ++++++++ cmd/clawpatrol/web.go | 3 +++ 2 files changed, 11 insertions(+) diff --git a/cmd/clawpatrol/telegram_request_body_redaction_test.go b/cmd/clawpatrol/telegram_request_body_redaction_test.go index f14c5f22..85b635f4 100644 --- a/cmd/clawpatrol/telegram_request_body_redaction_test.go +++ b/cmd/clawpatrol/telegram_request_body_redaction_test.go @@ -145,6 +145,14 @@ rule "allow-telegram" { if !strings.Contains(end.ReqBody, telegramTestPlaceholder) && !strings.Contains(strings.ToLower(end.ReqBody), "redact") { t.Fatalf("request body audit sample = %q, want placeholder or redaction marker", end.ReqBody) } + rw := httptest.NewRecorder() + (&webMux{g: g}).writeActionFixture(rw, &end) + if rw.Code != http.StatusBadRequest { + t.Fatalf("fixture export status = %d, want 400 for redacted request body; body=%s", rw.Code, rw.Body.String()) + } + if !strings.Contains(rw.Body.String(), "redacted") { + t.Fatalf("fixture export error = %q, want redacted-body explanation", rw.Body.String()) + } _ = clientTLS.Close() select { diff --git a/cmd/clawpatrol/web.go b/cmd/clawpatrol/web.go index 10429209..9c186f0d 100644 --- a/cmd/clawpatrol/web.go +++ b/cmd/clawpatrol/web.go @@ -1968,6 +1968,9 @@ func validateHTTPFixtureBodyCapture(ev *Event) error { if strings.HasSuffix(ev.ReqBody, decodedSampleTruncatedMarker) { return fmt.Errorf("request body decoded preview is truncated; cannot export as fixture") } + if strings.Contains(ev.ReqBody, credentialSampleRedaction) { + return fmt.Errorf("request body capture was redacted; cannot export as fixture") + } if !utf8.ValidString(ev.ReqBody) { return fmt.Errorf("request body capture is not valid UTF-8; cannot export as fixture") } From fcf9cf50156f188f6fe58e14697522a8806b08d8 Mon Sep 17 00:00:00 2001 From: Yusuke Tanaka Date: Mon, 17 Aug 2026 17:31:43 +0900 Subject: [PATCH 11/11] fix: reject transformed request fixtures --- cmd/clawpatrol/http_transform_fixture_test.go | 191 ++++++++++++++++++ cmd/clawpatrol/main.go | 6 +- .../0021_action_request_transformed.sql | 3 + cmd/clawpatrol/web.go | 145 +++++++------ dashboard/src/lib/api.ts | 1 + 5 files changed, 278 insertions(+), 68 deletions(-) create mode 100644 cmd/clawpatrol/http_transform_fixture_test.go create mode 100644 cmd/clawpatrol/migrations/sqlite/0021_action_request_transformed.sql diff --git a/cmd/clawpatrol/http_transform_fixture_test.go b/cmd/clawpatrol/http_transform_fixture_test.go new file mode 100644 index 00000000..130873b7 --- /dev/null +++ b/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") + } +} diff --git a/cmd/clawpatrol/main.go b/cmd/clawpatrol/main.go index 17ac0455..6f8c9eeb 100644 --- a/cmd/clawpatrol/main.go +++ b/cmd/clawpatrol/main.go @@ -2729,6 +2729,8 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c } case wantsHTTP: reqBodySecretRedactions = appendCredentialSecretRedactions(reqBodySecretRedactions, sec) + rewriter, isRewriter := injector.(runtime.HTTPRequestRewriter) + rewritesRequest := isRewriter && rewriter.RewritesHTTPRequest() // Match existing request-signing behavior: an injection failure is logged, // then the request continues with the agent's placeholder. The upstream // service should reject that placeholder without exposing gateway secrets. @@ -2738,7 +2740,7 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c // is corrupted, not merely un-injected — fail closed instead of // forwarding a half-transformed request. if err := injector.InjectHTTP(req.Context(), req, sec); err != nil { - if rw, ok := injector.(runtime.HTTPRequestRewriter); ok && rw.RewritesHTTPRequest() { + if rewritesRequest { log.Printf("transform %s: %v; failing closed", cc.Credential.Symbol.Name, err) _, _ = fmt.Fprintf(tc, "HTTP/1.1 502 Bad Gateway\r\nContent-Length: 0\r\nConnection: close\r\n\r\n") ev.Status = "502" @@ -2749,6 +2751,8 @@ func (g *Gateway) mitmHTTPSWithCertHost(c net.Conn, host, certHost string, ep *c return } log.Printf("inject %s: %v; forwarding without injection", cc.Credential.Symbol.Name, err) + } else if rewritesRequest { + ev.ReqTransformed = true } if rp, ok := injector.(runtime.HTTPCredentialRedactionProvider); ok { for _, secret := range rp.ConsumeHTTPRedactions(req) { diff --git a/cmd/clawpatrol/migrations/sqlite/0021_action_request_transformed.sql b/cmd/clawpatrol/migrations/sqlite/0021_action_request_transformed.sql new file mode 100644 index 00000000..a0515480 --- /dev/null +++ b/cmd/clawpatrol/migrations/sqlite/0021_action_request_transformed.sql @@ -0,0 +1,3 @@ +ALTER TABLE actions ADD COLUMN req_transformed INTEGER NOT NULL DEFAULT 0; + +INSERT INTO _schema (version) VALUES (21); diff --git a/cmd/clawpatrol/web.go b/cmd/clawpatrol/web.go index 9c186f0d..074fa5fc 100644 --- a/cmd/clawpatrol/web.go +++ b/cmd/clawpatrol/web.go @@ -1753,38 +1753,39 @@ func (w *webMux) loadAction(actionID string) (*Event, error) { return nil, fmt.Errorf("missing id") } var ( - e Event - tsNs int64 - mode sql.NullString - family sql.NullString - agentIP sql.NullString - method sql.NullString - path sql.NullString - status sql.NullString - in, ot sql.NullInt64 - ms sql.NullInt64 - action sql.NullString - reason sql.NullString - reqSha sql.NullString - respSha sql.NullString - reqBody sql.NullString - respBody sql.NullString - reqBodyState sql.NullString - respBodyState sql.NullString - reqHeaders sql.NullString - respHeaders sql.NullString - extra sql.NullString - endpoint sql.NullString - rule sql.NullString - approver sql.NullString - approverType sql.NullString - approverBy sql.NullString + e Event + tsNs int64 + mode sql.NullString + family sql.NullString + agentIP sql.NullString + method sql.NullString + path sql.NullString + status sql.NullString + in, ot sql.NullInt64 + ms sql.NullInt64 + action sql.NullString + reason sql.NullString + reqSha sql.NullString + respSha sql.NullString + reqBody sql.NullString + respBody sql.NullString + reqBodyState sql.NullString + respBodyState sql.NullString + reqTransformed sql.NullBool + reqHeaders sql.NullString + respHeaders sql.NullString + extra sql.NullString + endpoint sql.NullString + rule sql.NullString + approver sql.NullString + approverType sql.NullString + approverBy sql.NullString ) err := w.g.db.QueryRow(` SELECT 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_state, resp_body_state, + req_body, resp_body, req_body_state, resp_body_state, req_transformed, req_headers, resp_headers, extra, endpoint, rule, approver, approver_type, approver_by @@ -1793,7 +1794,7 @@ func (w *webMux) loadAction(actionID string) (*Event, error) { &tsNs, &mode, &family, &agentIP, &e.Host, &method, &path, &status, &in, &ot, &ms, &action, &reason, &reqSha, &respSha, - &reqBody, &respBody, &reqBodyState, &respBodyState, + &reqBody, &respBody, &reqBodyState, &respBodyState, &reqTransformed, &reqHeaders, &respHeaders, &extra, &endpoint, &rule, &approver, &approverType, &approverBy, @@ -1820,6 +1821,7 @@ func (w *webMux) loadAction(actionID string) (*Event, error) { e.RespBody = respBody.String e.ReqBodyState = reqBodyState.String e.RespBodyState = respBodyState.String + e.ReqTransformed = reqTransformed.Bool unmarshalHeaders(reqHeaders.String, &e.ReqHeaders) unmarshalHeaders(respHeaders.String, &e.RespHeaders) if extra.String != "" { @@ -1940,6 +1942,9 @@ func (w *webMux) writeActionFixture(rw http.ResponseWriter, ev *Event) { } func validateHTTPFixtureBodyCapture(ev *Event) error { + if ev.ReqTransformed { + return fmt.Errorf("request was transformed by a credential; cannot export as fixture") + } switch ev.ReqBodyState { case bodyCaptureComplete: case bodyCaptureIncomplete, bodyCaptureAborted: @@ -2497,17 +2502,21 @@ type Event struct { // / llm_approver / dashboard), and the approver-specific "By" // string (Slack handle, llm:, ...). All empty for rule- // driven verdicts. - Approver string `json:"approver,omitempty"` - ApproverType string `json:"approver_type,omitempty"` - ApproverBy string `json:"approver_by,omitempty"` - ReqSha string `json:"req_sha,omitempty"` - ReqBody string `json:"req_body,omitempty"` - ReqBodyState string `json:"req_body_state,omitempty"` - RespSha string `json:"resp_sha,omitempty"` - RespBody string `json:"resp_body,omitempty"` - RespBodyState string `json:"resp_body_state,omitempty"` - ReqHeaders map[string]string `json:"req_headers,omitempty"` - RespHeaders map[string]string `json:"resp_headers,omitempty"` + Approver string `json:"approver,omitempty"` + ApproverType string `json:"approver_type,omitempty"` + ApproverBy string `json:"approver_by,omitempty"` + ReqSha string `json:"req_sha,omitempty"` + ReqBody string `json:"req_body,omitempty"` + ReqBodyState string `json:"req_body_state,omitempty"` + // ReqTransformed records that credential injection successfully rewrote + // the request after policy matching. Such an audit body is not a faithful + // fixture input and must not be exported as one. + ReqTransformed bool `json:"req_transformed,omitempty"` + RespSha string `json:"resp_sha,omitempty"` + RespBody string `json:"resp_body,omitempty"` + RespBodyState string `json:"resp_body_state,omitempty"` + ReqHeaders map[string]string `json:"req_headers,omitempty"` + RespHeaders map[string]string `json:"resp_headers,omitempty"` // Frame is set for Phase="frame" only — a single WS frame's text // payload (truncated at sampleCap). Direction is "c→s" or "s→c" // to disambiguate masked client frames from unmasked server frames. @@ -2599,7 +2608,7 @@ func readTailEvents(db *sql.DB, n int) ([]Event, error) { SELECT 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_state, resp_body_state, extra, + req_body_state, resp_body_state, req_transformed, extra, endpoint, rule, approver, approver_type, approver_by FROM actions ORDER BY id DESC LIMIT ?`, n) @@ -2610,35 +2619,36 @@ func readTailEvents(db *sql.DB, n int) ([]Event, error) { out := make([]Event, 0, n) for rows.Next() { var ( - e Event - actionID sql.NullString - tsNs int64 - mode sql.NullString - family sql.NullString - agentIP sql.NullString - method sql.NullString - path sql.NullString - status sql.NullString - in, ot sql.NullInt64 - ms sql.NullInt64 - action sql.NullString - reason sql.NullString - reqSha sql.NullString - respSha sql.NullString - reqBodyState sql.NullString - respBodyState sql.NullString - extra sql.NullString - endpoint sql.NullString - rule sql.NullString - approver sql.NullString - approverType sql.NullString - approverBy sql.NullString + e Event + actionID sql.NullString + tsNs int64 + mode sql.NullString + family sql.NullString + agentIP sql.NullString + method sql.NullString + path sql.NullString + status sql.NullString + in, ot sql.NullInt64 + ms sql.NullInt64 + action sql.NullString + reason sql.NullString + reqSha sql.NullString + respSha sql.NullString + reqBodyState sql.NullString + respBodyState sql.NullString + reqTransformed sql.NullBool + extra sql.NullString + endpoint sql.NullString + rule sql.NullString + approver sql.NullString + approverType sql.NullString + approverBy sql.NullString ) if err := rows.Scan( &actionID, &tsNs, &mode, &family, &agentIP, &e.Host, &method, &path, &status, &in, &ot, &ms, &action, &reason, &reqSha, &respSha, - &reqBodyState, &respBodyState, &extra, + &reqBodyState, &respBodyState, &reqTransformed, &extra, &endpoint, &rule, &approver, &approverType, &approverBy, ); err != nil { @@ -2663,6 +2673,7 @@ func readTailEvents(db *sql.DB, n int) ([]Event, error) { e.RespSha = respSha.String e.ReqBodyState = reqBodyState.String e.RespBodyState = respBodyState.String + e.ReqTransformed = reqTransformed.Bool if extra.String != "" { _ = json.Unmarshal([]byte(extra.String), &e.Facets) } @@ -2766,16 +2777,16 @@ func (s *Sink) drain() { (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_state, resp_body_state, + req_body, resp_body, req_body_state, resp_body_state, req_transformed, req_headers, resp_headers, extra, endpoint, rule, approver, approver_type, approver_by) - VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) `, e.ID, e.Ts.UnixNano(), e.Mode, e.Family, e.AgentIP, e.Host, e.Method, e.Path, e.Status, e.In, e.Out, e.Ms, e.Action, e.Reason, e.ReqSha, e.RespSha, - e.ReqBody, e.RespBody, e.ReqBodyState, e.RespBodyState, + e.ReqBody, e.RespBody, e.ReqBodyState, e.RespBodyState, e.ReqTransformed, string(rqhJSON), string(rshJSON), string(extraJSON), e.Endpoint, e.Rule, diff --git a/dashboard/src/lib/api.ts b/dashboard/src/lib/api.ts index 5b565570..bc5d10a5 100644 --- a/dashboard/src/lib/api.ts +++ b/dashboard/src/lib/api.ts @@ -542,6 +542,7 @@ export type EventRecord = { req_body?: string; resp_body?: string; req_body_state?: "complete" | "incomplete" | "aborted"; + req_transformed?: boolean; resp_body_state?: "complete" | "incomplete" | "aborted"; req_headers?: Record; resp_headers?: Record;