package recorder
import (
"context"
"errors"
"fmt"
"io"
"sync"
"time"
)
// AsyncBackpressurePolicy controls what AsyncRecorder does when its bounded
// queue is full.
type AsyncBackpressurePolicy uint8
const (
// AsyncBlock preserves entries by waiting for queue capacity. This is the
// default. A stalled sink can therefore delay response-body Read or Close
// calls that finalize HTTP exchanges.
AsyncBlock AsyncBackpressurePolicy = iota
// AsyncDropNewest preserves the already accepted FIFO prefix and drops the
// entry currently being recorded when the queue is full.
AsyncDropNewest
// AsyncDropOldest removes the oldest queued (not in-flight) entry to make
// room for the entry currently being recorded.
AsyncDropOldest
)
const defaultAsyncQueueCapacity = 1024
// ErrAsyncRecorderClosed is returned when Record is called after shutdown has
// begun. The entry is rejected and reported to the configured drop handler.
var ErrAsyncRecorderClosed = errors.New("recorder: async recorder is closed")
// AsyncDropReason identifies the bounded reason an entry was not delivered.
type AsyncDropReason string
const (
// AsyncDropPolicyNewest means the configured full-queue policy rejected
// the entry being recorded.
AsyncDropPolicyNewest AsyncDropReason = "policy_newest"
// AsyncDropPolicyOldest means the configured full-queue policy evicted the
// oldest queued entry.
AsyncDropPolicyOldest AsyncDropReason = "policy_oldest"
// AsyncDropTimeoutNewest means a bounded wait expired and rejected the
// entry being recorded.
AsyncDropTimeoutNewest AsyncDropReason = "timeout_newest"
// AsyncDropTimeoutOldest means a bounded wait expired and evicted the
// oldest queued entry.
AsyncDropTimeoutOldest AsyncDropReason = "timeout_oldest"
// AsyncDropClosed means shutdown rejected an entry.
AsyncDropClosed AsyncDropReason = "closed"
)
// AsyncDropHandler observes an entry that AsyncRecorder discarded. It runs
// after the queue lock is released and must be concurrency-safe and bounded.
// Returned errors are reported through AsyncRecorder's internal-error policy
// and returned by Close.
type AsyncDropHandler func(*Entry, AsyncDropReason) error
// AsyncRecorderStats is a point-in-time snapshot of queue and sink activity.
// Processed means the downstream Record call was attempted; the minimal
// Recorder interface cannot prove that a sink durably persisted an entry.
type AsyncRecorderStats struct {
Capacity int
Accepted uint64
Processed uint64
BatchesProcessed uint64
MaxBatchSize int
BlockedRecords uint64
CurrentlyBlocked int
TotalBlockTime time.Duration
MaxBlockTime time.Duration
OldestBlockAge time.Duration
DroppedNewest uint64
DroppedOldest uint64
DroppedTimeoutNewest uint64
DroppedTimeoutOldest uint64
DroppedClosed uint64
DropHandlerErrors uint64
DropHandlerPanics uint64
SinkPanics uint64
SinkErrors uint64
Pending int
InFlight int
}
// AsyncRecorderConfig defines bounded queue, batching, backpressure, callback,
// and downstream ownership behavior for NewAsyncRecorder.
type AsyncRecorderConfig struct {
// QueueCapacity is the maximum number of accepted entries awaiting delivery.
QueueCapacity int
// Backpressure controls behavior when QueueCapacity is exhausted.
Backpressure AsyncBackpressurePolicy
// CloseSink transfers io.Closer ownership of the downstream sink.
CloseSink bool
// InternalErrorMode selects the internal error policy. See the constants.
InternalErrorMode InternalErrorMode
// OnInternalError observes contained sink, callback, and close failures.
OnInternalError func(error)
// Logf is used by InternalErrorLog. Nil falls back to log.Printf.
Logf func(format string, args ...any)
// BatchSize is the maximum entries passed to RecordBatch; one disables batching.
BatchSize int
// FlushInterval bounds how long a partial batch waits; zero flushes immediately.
// A positive value requires BatchSize above one and a batch-capable sink.
FlushInterval time.Duration
// BlockTimeout bounds AsyncBlock; zero waits without a deadline. It is valid
// only with Backpressure set to AsyncBlock.
BlockTimeout time.Duration
// BlockTimeoutPolicy is the drop policy used after BlockTimeout.
BlockTimeoutPolicy AsyncBackpressurePolicy
// DropHandler observes entries discarded by policy, timeout, or close.
DropHandler AsyncDropHandler
}
// DefaultAsyncRecorderConfig returns the evidence-preserving production
// baseline, including visible internal-error logging.
func DefaultAsyncRecorderConfig() AsyncRecorderConfig {
return AsyncRecorderConfig{
QueueCapacity: defaultAsyncQueueCapacity,
Backpressure: AsyncBlock,
BatchSize: 1,
InternalErrorMode: InternalErrorLog,
}
}
type asyncRecorderState uint8
const (
asyncAccepting asyncRecorderState = iota
asyncClosing
asyncClosed
)
// AsyncRecorder decouples entry finalization from a downstream Recorder with
// one bounded FIFO queue and one worker. It preserves accepted-entry order but
// does not provide crash durability, retries, or proof of persistence.
type AsyncRecorder struct {
mu sync.Mutex
notEmpty *sync.Cond
notFull *sync.Cond
sink Recorder
queue []*Entry
head int
size int
state asyncRecorderState
policy AsyncBackpressurePolicy
closeSink bool
internalErrorMode InternalErrorMode
onInternalError func(error)
logf func(string, ...any)
now func() time.Time
batchSize int
flushWait time.Duration
blockWait time.Duration
blockDrop AsyncBackpressurePolicy
onDrop AsyncDropHandler
blocked map[uint64]time.Time
nextBlockID uint64
done chan struct{}
firstErr error
stats AsyncRecorderStats
}
// NewAsyncRecorder wraps sink with a bounded asynchronous queue. Config must
// specify a positive queue capacity and batch size; DefaultAsyncRecorderConfig
// supplies the evidence-preserving baseline of 1,024 entries, batches of one,
// and AsyncBlock. Blocking prevents queue-overflow loss during normal process
// operation, but a slow or stalled sink can delay application goroutines
// finalizing HTTP exchanges.
func NewAsyncRecorder(sink Recorder, config AsyncRecorderConfig) (*AsyncRecorder, error) {
if sink == nil {
return nil, errors.New("recorder: async recorder requires a sink")
}
if config.QueueCapacity <= 0 {
return nil, errors.New("recorder: async queue capacity must be positive")
}
if config.BatchSize <= 0 {
return nil, errors.New("recorder: async batch size must be positive")
}
if config.BatchSize > config.QueueCapacity {
return nil, errors.New("recorder: async batch size must not exceed queue capacity")
}
if config.FlushInterval < 0 {
return nil, errors.New("recorder: async flush interval must not be negative")
}
if config.FlushInterval > 0 && config.BatchSize == 1 {
return nil, errors.New("recorder: async flush interval requires batch size above one")
}
if config.BlockTimeout < 0 {
return nil, errors.New("recorder: async block timeout must not be negative")
}
if config.BlockTimeout > 0 {
if config.Backpressure != AsyncBlock {
return nil, errors.New("recorder: async block timeout requires AsyncBlock policy")
}
if config.BlockTimeoutPolicy != AsyncDropNewest && config.BlockTimeoutPolicy != AsyncDropOldest {
return nil, errors.New("recorder: async block timeout requires a drop fallback")
}
}
if config.BatchSize > 1 {
if _, ok := sink.(batchRecorder); !ok {
return nil, errors.New("recorder: async batching requires a batchRecorder sink")
}
}
switch config.Backpressure {
case AsyncBlock, AsyncDropNewest, AsyncDropOldest:
default:
return nil, fmt.Errorf("recorder: unknown async backpressure policy %d", config.Backpressure)
}
r := &AsyncRecorder{
sink: sink,
queue: make([]*Entry, config.QueueCapacity),
policy: config.Backpressure,
closeSink: config.CloseSink,
internalErrorMode: config.InternalErrorMode,
onInternalError: config.OnInternalError,
logf: config.Logf,
now: time.Now,
batchSize: config.BatchSize,
flushWait: config.FlushInterval,
blockWait: config.BlockTimeout,
blockDrop: config.BlockTimeoutPolicy,
onDrop: config.DropHandler,
blocked: make(map[uint64]time.Time),
done: make(chan struct{}),
}
r.notEmpty = sync.NewCond(&r.mu)
r.notFull = sync.NewCond(&r.mu)
go r.run()
return r, nil
}
// Record implements Recorder. With AsyncBlock it may wait for queue capacity;
// dropping policies return immediately when the queue is full. Calls made
// after Close begins return ErrAsyncRecorderClosed and are counted as
// DroppedClosed. Configured drop policies are intentional outcomes and return
// nil while remaining observable through Stats and DropHandler.
func (r *AsyncRecorder) Record(entry *Entry) error {
r.mu.Lock()
blocked := false
timedOut := false
var blockedAt time.Time
var blockID uint64
var timer *time.Timer
for r.state == asyncAccepting && r.size == len(r.queue) && r.policy == AsyncBlock {
if !blocked {
blocked = true
blockedAt = r.now()
blockID = r.nextBlockID
r.nextBlockID++
r.blocked[blockID] = blockedAt
r.stats.BlockedRecords++
r.stats.CurrentlyBlocked++
if r.blockWait > 0 {
timer = time.AfterFunc(r.blockWait, func() {
r.mu.Lock()
r.notFull.Broadcast()
r.mu.Unlock()
})
}
}
if r.blockWait > 0 && !r.now().Before(blockedAt.Add(r.blockWait)) {
timedOut = true
break
}
r.notFull.Wait()
}
if timer != nil {
timer.Stop()
}
if blocked {
d := r.now().Sub(blockedAt)
delete(r.blocked, blockID)
r.stats.CurrentlyBlocked--
r.stats.TotalBlockTime += d
if d > r.stats.MaxBlockTime {
r.stats.MaxBlockTime = d
}
}
if r.state != asyncAccepting {
r.stats.DroppedClosed++
r.mu.Unlock()
r.handleDrop(entry, AsyncDropClosed)
return ErrAsyncRecorderClosed
}
effectivePolicy := r.policy
if timedOut {
effectivePolicy = r.blockDrop
}
var dropped *Entry
var dropReason AsyncDropReason
if r.size == len(r.queue) {
switch effectivePolicy {
case AsyncDropNewest:
if timedOut {
r.stats.DroppedTimeoutNewest++
dropReason = AsyncDropTimeoutNewest
} else {
r.stats.DroppedNewest++
dropReason = AsyncDropPolicyNewest
}
r.mu.Unlock()
r.handleDrop(entry, dropReason)
return nil
case AsyncDropOldest:
dropped = r.queue[r.head]
r.queue[r.head] = nil
r.head = (r.head + 1) % len(r.queue)
r.size--
if timedOut {
r.stats.DroppedTimeoutOldest++
dropReason = AsyncDropTimeoutOldest
} else {
r.stats.DroppedOldest++
dropReason = AsyncDropPolicyOldest
}
}
}
index := (r.head + r.size) % len(r.queue)
r.queue[index] = entry
r.size++
r.stats.Accepted++
r.notEmpty.Signal()
r.mu.Unlock()
r.handleDrop(dropped, dropReason)
return nil
}
// Stats returns a concurrency-safe point-in-time snapshot.
func (r *AsyncRecorder) Stats() AsyncRecorderStats {
r.mu.Lock()
defer r.mu.Unlock()
stats := r.stats
stats.Capacity = len(r.queue)
stats.Pending = r.size
now := r.now()
for _, started := range r.blocked {
age := now.Sub(started)
if age > stats.OldestBlockAge {
stats.OldestBlockAge = age
}
}
return stats
}
func (r *AsyncRecorder) handleDrop(entry *Entry, reason AsyncDropReason) {
if entry == nil || r.onDrop == nil {
return
}
defer func() {
if recovered := recover(); recovered != nil {
r.recordDropHandlerError(fmt.Errorf("recorder: async drop handler panic (%s): %v", reason, recovered), true)
}
}()
if err := r.onDrop(entry, reason); err != nil {
r.recordDropHandlerError(fmt.Errorf("recorder: async drop handler (%s): %w", reason, err), false)
}
}
func (r *AsyncRecorder) recordDropHandlerError(err error, panicked bool) {
r.mu.Lock()
if r.firstErr == nil {
r.firstErr = err
}
if panicked {
r.stats.DropHandlerPanics++
} else {
r.stats.DropHandlerErrors++
}
r.mu.Unlock()
reportInternalError(r.internalErrorMode, r.onInternalError, r.logf, err)
}
func (r *AsyncRecorder) firstError() error {
r.mu.Lock()
defer r.mu.Unlock()
return r.firstErr
}
// Close stops accepting entries and waits for the queue and active downstream
// call to drain. If ctx expires, draining continues in the background and a
// later Close call may wait again. Active Recorder.Record calls cannot be
// cancelled because Recorder intentionally has no context-aware method.
func (r *AsyncRecorder) Close(ctx context.Context) error {
if ctx == nil {
return errors.New("recorder: async recorder close requires a context")
}
r.mu.Lock()
if r.state == asyncAccepting {
r.state = asyncClosing
r.notEmpty.Broadcast()
r.notFull.Broadcast()
}
done := r.done
r.mu.Unlock()
select {
case <-done:
return r.firstError()
case <-ctx.Done():
return ctx.Err()
}
}
func (r *AsyncRecorder) run() {
batchBuffer := make([]*Entry, r.batchSize)
for {
r.mu.Lock()
for r.size == 0 && r.state == asyncAccepting {
r.notEmpty.Wait()
}
if r.size == 0 && r.state != asyncAccepting {
r.mu.Unlock()
r.finish()
return
}
r.waitForBatchLocked()
batchSize := min(r.size, r.batchSize)
batch := batchBuffer[:batchSize]
for i := range batchSize {
batch[i] = r.queue[r.head]
r.queue[r.head] = nil
r.head = (r.head + 1) % len(r.queue)
r.size--
}
r.stats.InFlight = batchSize
r.notFull.Broadcast()
r.mu.Unlock()
r.deliverBatch(batch)
r.mu.Lock()
r.stats.Processed += uint64(batchSize)
r.stats.BatchesProcessed++
if batchSize > r.stats.MaxBatchSize {
r.stats.MaxBatchSize = batchSize
}
r.stats.InFlight = 0
r.mu.Unlock()
for i := range batch {
batch[i] = nil
}
}
}
func (r *AsyncRecorder) waitForBatchLocked() {
if r.batchSize == 1 || r.flushWait == 0 || r.size >= r.batchSize || r.state != asyncAccepting {
return
}
expired := false
timer := time.AfterFunc(r.flushWait, func() {
r.mu.Lock()
expired = true
r.notEmpty.Broadcast()
r.mu.Unlock()
})
for r.size < r.batchSize && r.state == asyncAccepting && !expired {
r.notEmpty.Wait()
}
timer.Stop()
}
func (r *AsyncRecorder) deliverBatch(entries []*Entry) {
if sink, ok := r.sink.(batchRecorder); ok && r.batchSize > 1 {
r.deliver(func() error { return sink.RecordBatch(entries) })
return
}
for _, entry := range entries {
r.deliver(func() error { return r.sink.Record(entry) })
}
}
func (r *AsyncRecorder) deliver(call func() error) {
panicked := true
defer func() {
if recovered := recover(); panicked {
r.recordError(fmt.Errorf("recorder: async sink panic: %v", recovered), true)
}
}()
err := call()
panicked = false
if err != nil {
r.recordError(fmt.Errorf("recorder: async sink: %w", err), false)
}
}
func (r *AsyncRecorder) finish() {
if r.closeSink {
if closer, ok := r.sink.(io.Closer); ok {
if err := r.callClose(closer); err != nil {
r.recordError(fmt.Errorf("recorder: close async sink: %w", err), false)
}
}
}
r.mu.Lock()
r.state = asyncClosed
close(r.done)
r.mu.Unlock()
}
func (r *AsyncRecorder) callClose(closer io.Closer) (err error) {
defer func() {
if recovered := recover(); recovered != nil {
err = fmt.Errorf("sink close panic: %v", recovered)
}
}()
return closer.Close()
}
func (r *AsyncRecorder) recordError(err error, panicked bool) {
if err == nil {
return
}
r.mu.Lock()
if r.firstErr == nil {
r.firstErr = err
}
if panicked {
r.stats.SinkPanics++
} else {
r.stats.SinkErrors++
}
r.mu.Unlock()
reportInternalError(r.internalErrorMode, r.onInternalError, r.logf, err)
}
package recorder
import (
"fmt"
"io"
"sync"
)
// BodyRedactor creates a streaming transform for one captured body. Redact
// may be called concurrently for different bodies; each returned writer is
// owned by one body and each input byte passes through it once. Close must
// flush parser state but must not close dst.
type BodyRedactor interface {
Redact(dst io.Writer, contentType string, protector BodyValueProtector) (io.WriteCloser, error)
}
// BodyValueProtector creates bounded streaming values that apply the
// transport's configured sensitive-value mode and token format. It is scoped
// to one body writer and must not be retained after that writer closes.
type BodyValueProtector interface {
NewValue() BodyValue
}
// BodyValue accepts one selected plaintext value in chunks. FinishTo writes its
// protected representation and records exactly one replacement. Encryption
// buffers only up to the configured value limit, tokenization streams through
// HMAC, and failures or oversized values fail closed to [REDACTED]. FinishTo
// may be repeated to write the same result, but a BodyValue must not receive
// more plaintext after its first FinishTo call.
type BodyValue interface {
io.Writer
FinishTo(io.Writer) error
}
type bodyRedactorFunc func(io.Writer, string, BodyValueProtector) (io.WriteCloser, error)
func (f bodyRedactorFunc) Redact(dst io.Writer, contentType string, protector BodyValueProtector) (io.WriteCloser, error) {
return f(dst, contentType, protector)
}
type writerOnly struct{ io.Writer }
type failedBodyRedactor struct{ err error }
func (w *failedBodyRedactor) Write([]byte) (int, error) { return 0, w.err }
func (w *failedBodyRedactor) Close() error { return w.err }
type protectionAwareBodyRedactorWriter struct {
inner io.WriteCloser
protector *bodyValueProtector
}
func (w *protectionAwareBodyRedactorWriter) Write(p []byte) (int, error) {
return w.inner.Write(p)
}
func (w *protectionAwareBodyRedactorWriter) Close() error {
return w.inner.Close()
}
func (w *protectionAwareBodyRedactorWriter) bodyRedactionReport() (ProtectionCounts, int64) {
return w.protector.protectionReport()
}
func (w *protectionAwareBodyRedactorWriter) bodyProtectionFailure() (error, int64) {
return w.protector.protectionFailure()
}
// safeBodyRedactorWriter normalizes custom and built-in writers to the same
// panic, short-write, and exactly-once Close behavior.
type safeBodyRedactorWriter struct {
inner io.WriteCloser
closeOnce sync.Once
closeErr error
}
func (w *safeBodyRedactorWriter) Write(p []byte) (n int, err error) {
defer func() {
if recovered := recover(); recovered != nil {
n = 0
err = fmt.Errorf("recorder: panic in body redactor Write: %v", recovered)
}
}()
n, err = w.inner.Write(p)
if err == nil && n != len(p) {
err = io.ErrShortWrite
}
return n, err
}
func (w *safeBodyRedactorWriter) Close() error {
w.closeOnce.Do(func() {
defer func() {
if recovered := recover(); recovered != nil {
w.closeErr = fmt.Errorf("recorder: panic in body redactor Close: %v", recovered)
}
}()
w.closeErr = w.inner.Close()
})
return w.closeErr
}
func (w *safeBodyRedactorWriter) bodyRedactionReport() (protection ProtectionCounts, replacements int64) {
if reporter, ok := w.inner.(interface {
bodyRedactionReport() (ProtectionCounts, int64)
}); ok {
return reporter.bodyRedactionReport()
}
return ProtectionCounts{}, 0
}
func (w *safeBodyRedactorWriter) bodyProtectionFailure() (err error, count int64) {
reporter, ok := w.inner.(interface{ bodyProtectionFailure() (error, int64) })
if !ok {
return nil, 0
}
defer func() {
if recover() != nil {
err, count = nil, 0
}
}()
err, count = reporter.bodyProtectionFailure()
return err, count
}
type auditedBodyRedactorWriter struct {
inner io.WriteCloser
audit *redactionAudit
direction BodyDirection
kind string
mu sync.Mutex
writeErr error
closeOnce sync.Once
closeErr error
}
func (w *auditedBodyRedactorWriter) Write(p []byte) (int, error) {
n, err := w.inner.Write(p)
if err != nil {
w.mu.Lock()
if w.writeErr == nil {
w.writeErr = err
}
w.mu.Unlock()
}
return n, err
}
func (w *auditedBodyRedactorWriter) Close() error {
w.closeOnce.Do(func() {
w.closeErr = w.inner.Close()
info := BodyRedactionInfo{Kind: w.kind, Outcome: BodyRedactionUnchanged}
w.mu.Lock()
writeFailed := w.writeErr != nil
w.mu.Unlock()
if writeFailed || w.closeErr != nil {
info.Outcome = BodyRedactionFailed
} else if reporter, ok := w.inner.(interface {
bodyRedactionReport() (ProtectionCounts, int64)
}); ok {
protection, replacements := reporter.bodyRedactionReport()
if replacements < 0 {
replacements = 0
}
if !protectionCountsEmpty(protection) {
copy := cloneProtectionCounts(protection)
info.Protection = ©
}
if replacements > 0 {
info.Replacements = &replacements
info.Outcome = BodyRedactionRedacted
} else {
info.Replacements = &replacements
info.Outcome = BodyRedactionUnchanged
}
}
if reporter, ok := w.inner.(interface{ bodyProtectionFailure() (error, int64) }); ok {
err, count := reporter.bodyProtectionFailure()
w.audit.addProtectionFailure(w.direction, err, count)
}
w.audit.setBody(w.direction, info)
})
return w.closeErr
}
func selectBodyRedactor(contentType string, red *redactor) (BodyRedactor, string) {
if red == nil {
return nil, ""
}
if custom := red.bodyRedactors[baseMimeType(contentType)]; custom != nil {
return custom, "custom"
}
switch {
case isMultipartFormMime(contentType) && len(red.query) > 0:
return bodyRedactorFunc(func(dst io.Writer, contentType string, protector BodyValueProtector) (io.WriteCloser, error) {
w := newMultipartStreamRedactor(dst, contentType, red.query, builtinBodyValueProtector(protector))
if w.err != nil {
return nil, w.err
}
return w, nil
}), "builtin:multipart"
case isFormMime(contentType) && len(red.query) > 0:
return bodyRedactorFunc(func(dst io.Writer, _ string, protector BodyValueProtector) (io.WriteCloser, error) {
return newFormStreamRedactor(dst, red.query, builtinBodyValueProtector(protector)), nil
}), "builtin:form"
case isJSONMime(contentType) && len(red.jsonFields) > 0:
return bodyRedactorFunc(func(dst io.Writer, _ string, protector BodyValueProtector) (io.WriteCloser, error) {
return newJSONStreamRedactor(dst, red.jsonFields, builtinBodyValueProtector(protector)), nil
}), "builtin:json"
case isXMLMime(contentType) && len(red.xmlElements) > 0:
return bodyRedactorFunc(func(dst io.Writer, _ string, protector BodyValueProtector) (io.WriteCloser, error) {
return newXMLStreamRedactor(dst, red.xmlElements, builtinBodyValueProtector(protector)), nil
}), "builtin:xml"
case !isFormMime(contentType) && !isMultipartFormMime(contentType) &&
!isJSONMime(contentType) && !isXMLMime(contentType) &&
(len(red.jsonFields) > 0 || len(red.xmlElements) > 0):
return bodyRedactorFunc(func(dst io.Writer, _ string, protector BodyValueProtector) (io.WriteCloser, error) {
return &sniffingBodyRedactor{dst: dst, red: red, protector: builtinBodyValueProtector(protector)}, nil
}), "builtin:sniff"
default:
return nil, ""
}
}
func builtinBodyValueProtector(protector BodyValueProtector) *bodyValueProtector {
return protector.(*bodyValueProtector)
}
func newBodyStreamRedactor(dst io.Writer, contentType string, red *redactor) io.WriteCloser {
selected, kind := selectBodyRedactor(contentType, red)
if selected == nil {
return nil
}
inner, err := openBodyRedactor(selected, writerOnly{dst}, contentType, red.protector)
if err != nil {
inner = &failedBodyRedactor{err: err}
} else if inner == nil {
inner = &failedBodyRedactor{err: fmt.Errorf("recorder: body redactor returned a nil writer")}
} else {
inner = &safeBodyRedactorWriter{inner: inner}
}
if red.audit != nil {
return &auditedBodyRedactorWriter{inner: inner, audit: red.audit, direction: red.direction, kind: kind}
}
return inner
}
func openBodyRedactor(redactor BodyRedactor, dst io.Writer, contentType string, protector *sensitiveValueProtector) (writer io.WriteCloser, err error) {
defer func() {
if recovered := recover(); recovered != nil {
writer = nil
err = fmt.Errorf("recorder: panic in BodyRedactor.Redact: %v", recovered)
}
}()
session := newBodyValueProtector(protector)
inner, openErr := redactor.Redact(dst, contentType, session)
if openErr != nil || inner == nil {
return inner, openErr
}
return &protectionAwareBodyRedactorWriter{inner: inner, protector: session}, nil
}
func bodyStreamRedactionEnabled(contentType string, red *redactor) bool {
selected, _ := selectBodyRedactor(contentType, red)
return selected != nil
}
package recorder
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"os"
"path/filepath"
"strings"
"sync"
"time"
)
// BodyMetadata describes the body stream a BodyWriter is created for.
type BodyMetadata struct {
// ExchangeID correlates the body with its immutable entry.
ExchangeID string
// Direction is "request" or "response".
Direction string
// ContentType is the observed media type, including parameters.
ContentType string
// SizeHint is the expected number of bytes this writer will receive
// (derived from Content-Length, already clamped to the capture limit),
// or 0 when unknown. Stores may use it to pre-allocate, but must treat
// it as untrusted: the actual stream may be shorter or longer, and a
// hostile peer can announce an absurd Content-Length.
SizeHint int64
}
// maxPreallocBytes caps how much memory a SizeHint may pre-allocate in one
// step. A lying Content-Length can therefore waste at most this much; larger
// bodies simply grow the buffer as bytes actually arrive.
const maxPreallocBytes = 4 << 20
// BodyWriter receives the captured representation of one body stream. When
// configured, the capture pipeline may decode and/or redact bytes before
// Write; it never requires a store to rewrite already-persisted content.
// Implementations must be safe for concurrent use of Write with Bytes.
type BodyWriter interface {
io.Writer
// Commit finalizes and publishes the captured representation. It is
// idempotent; Ref must remain empty until Commit succeeds.
Commit() error
// Abort closes the writer and discards its captured representation. It is
// idempotent and must not publish a Ref.
Abort() error
// Bytes returns the bytes captured so far (used when embedding content
// into the HAR document).
Bytes() ([]byte, error)
// Ref returns an opaque external reference to the stored content, or ""
// when the content lives inline in memory.
Ref() string
}
// BodyStore creates BodyWriter instances. Implementations receive the
// capture pipeline's processed representation, while BodyInfo counters and
// hashes continue to describe the original caller/wire stream.
// Implementations must be safe for concurrent use.
type BodyStore interface {
NewWriter(ctx context.Context, metadata BodyMetadata) (BodyWriter, error)
}
// entryAssetReleaser is an optional BodyStore capability for releasing all
// externally stored assets referenced by an entry that will not be retained.
type entryAssetReleaser interface {
ReleaseEntryAssets(*Entry) error
}
// MemoryBodyStore keeps captured bodies in memory. It is the default store.
type MemoryBodyStore struct{}
// NewWriter implements BodyStore. A positive SizeHint pre-sizes the buffer
// (bounded by maxPreallocBytes) so growth re-copies are avoided for bodies
// with a truthful Content-Length; a wrong hint costs at most one bounded
// allocation and never breaks the capture.
func (MemoryBodyStore) NewWriter(_ context.Context, meta BodyMetadata) (BodyWriter, error) {
w := &memoryBodyWriter{}
if hint := meta.SizeHint; hint > 0 {
if hint > maxPreallocBytes {
hint = maxPreallocBytes
}
w.buf.Grow(int(hint))
}
return w, nil
}
type memoryBodyWriter struct {
mu sync.Mutex
buf bytes.Buffer
committed bool
aborted bool
}
func (w *memoryBodyWriter) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
if w.committed || w.aborted {
return 0, os.ErrClosed
}
return w.buf.Write(p)
}
func (w *memoryBodyWriter) Commit() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.aborted {
return errors.New("recorder: commit aborted memory body writer")
}
w.committed = true
return nil
}
func (w *memoryBodyWriter) Abort() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.committed || w.aborted {
return nil
}
w.aborted = true
w.buf.Reset()
return nil
}
func (w *memoryBodyWriter) Bytes() ([]byte, error) {
w.mu.Lock()
defer w.mu.Unlock()
return append([]byte(nil), w.buf.Bytes()...), nil
}
func (w *memoryBodyWriter) Ref() string { return "" }
const (
defaultFileBodyMaxBytes = int64(1 << 30)
defaultFileBodyMaxFiles = 10_000
defaultFileBodyPartialTTL = time.Hour
fileBodyRefPrefix = "filebody:v1:"
)
// FileBodyStoreConfig defines managed disk capacity and commit durability.
type FileBodyStoreConfig struct {
// MaxBytes bounds committed and partial bytes owned by the store.
MaxBytes int64
// MaxFiles bounds committed and partial files owned by the store.
MaxFiles int
// PartialTTL controls startup cleanup of abandoned partial files. Zero
// removes every partial during startup; negative values are invalid.
PartialTTL time.Duration
// SyncOnCommit fsyncs the file and containing directory during publish. It
// does not make the separately recorded Entry crash-durable.
SyncOnCommit bool
// MaintenanceMode opens an existing store without creating directories,
// changing permissions, or recovering partial files. NewWriter is disabled;
// Open, Release, Reconcile, and Stats remain available. This makes dry-run
// reconciliation observational while preserving the normal constructor's
// path and reference validation.
MaintenanceMode bool
}
// DefaultFileBodyStoreConfig returns bounded managed-store defaults.
func DefaultFileBodyStoreConfig() FileBodyStoreConfig {
return FileBodyStoreConfig{MaxBytes: defaultFileBodyMaxBytes, MaxFiles: defaultFileBodyMaxFiles, PartialTTL: defaultFileBodyPartialTTL}
}
// FileBodyStoreStats is an atomic point-in-time lifecycle and capacity
// snapshot. Totals are process-lifetime counters except RecoveredPartials,
// which includes startup recovery.
type FileBodyStoreStats struct {
MaxBytes int64
MaxFiles int
PartialBytes int64
CommittedBytes int64
PartialFiles int
CommittedFiles int
CommittedTotal uint64
AbortedTotal uint64
ReleasedTotal uint64
QuotaRejected uint64
RecoveredPartials uint64
WriteFailures uint64
CommitFailures uint64
AbortFailures uint64
ReleaseFailures uint64
}
// ReconcileResult reports an authoritative live-reference reconciliation.
type ReconcileResult struct {
Scanned int
Released int
ReleasedBytes int64
}
// FileBodyStore spools processed body representations into a bounded managed
// directory. Writers use partial files and publish opaque references only
// after an atomic commit. Committed assets remain until explicitly released or
// removed by an authoritative reconciliation. A root must be owned by one
// process; FileBodyStore does not provide cross-process locking.
type FileBodyStore struct {
mu sync.Mutex
root string
partialDir string
assetDir string
maxBytes int64
maxFiles int
partialTTL time.Duration
syncOnCommit bool
maintenance bool
stats FileBodyStoreStats
}
// NewFileBodyStore opens a managed body store rooted at dir. Config must
// specify positive byte and file limits; DefaultFileBodyStoreConfig supplies
// the bounded baseline of 1 GiB, 10,000 files, and a one-hour partial TTL.
func NewFileBodyStore(dir string, config FileBodyStoreConfig) (*FileBodyStore, error) {
if strings.TrimSpace(dir) == "" {
return nil, errors.New("recorder: file body store directory is required")
}
if config.MaxBytes <= 0 || config.MaxFiles <= 0 || config.PartialTTL < 0 {
return nil, errors.New("recorder: invalid file body store limits")
}
root, err := filepath.Abs(dir)
if err != nil {
return nil, fmt.Errorf("recorder: resolve body store directory: %w", err)
}
s := &FileBodyStore{
root: root,
partialDir: filepath.Join(root, "partial"),
assetDir: filepath.Join(root, "assets"),
maxBytes: config.MaxBytes,
maxFiles: config.MaxFiles,
partialTTL: config.PartialTTL,
syncOnCommit: config.SyncOnCommit,
maintenance: config.MaintenanceMode,
}
if err := s.initialize(); err != nil {
return nil, err
}
return s, nil
}
// NewWriter implements BodyStore.
func (s *FileBodyStore) NewWriter(_ context.Context, _ BodyMetadata) (BodyWriter, error) {
if s == nil {
return nil, errors.New("recorder: nil file body store")
}
if s.maintenance {
return nil, errors.New("recorder: file body store is open in maintenance mode")
}
s.mu.Lock()
if s.stats.PartialFiles+s.stats.CommittedFiles >= s.maxFiles {
s.stats.QuotaRejected++
s.mu.Unlock()
return nil, errors.New("recorder: file body store file quota exceeded")
}
s.stats.PartialFiles++
s.mu.Unlock()
id := newID()
partialPath := filepath.Join(s.partialDir, id+".partial")
finalPath := filepath.Join(s.assetDir, id+".body")
f, err := os.OpenFile(partialPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
if err != nil {
s.mu.Lock()
s.stats.PartialFiles--
s.mu.Unlock()
return nil, fmt.Errorf("recorder: create body file: %w", err)
}
return &fileBodyWriter{
store: s,
f: f,
partialPath: partialPath,
finalPath: finalPath,
ref: fileBodyRefPrefix + id,
}, nil
}
type fileBodyWriter struct {
mu sync.Mutex
store *FileBodyStore
f *os.File
partialPath string
finalPath string
ref string
written int64
committed bool
aborted bool
}
func (w *fileBodyWriter) Write(p []byte) (int, error) {
w.mu.Lock()
defer w.mu.Unlock()
if w.committed || w.aborted {
return 0, os.ErrClosed
}
if err := w.store.reserveBytes(int64(len(p))); err != nil {
return 0, err
}
n, err := w.f.Write(p)
w.written += int64(n)
if n < len(p) {
w.store.releasePartialBytes(int64(len(p) - n))
if err == nil {
err = io.ErrShortWrite
}
}
if err != nil {
w.store.recordWriteFailure()
}
return n, err
}
func (w *fileBodyWriter) Commit() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.committed {
return nil
}
if w.aborted {
return errors.New("recorder: commit aborted file body writer")
}
if w.store.syncOnCommit {
if err := w.f.Sync(); err != nil {
w.store.recordCommitFailure()
_ = w.abortLocked()
return fmt.Errorf("recorder: sync body file: %w", err)
}
}
if err := w.f.Close(); err != nil {
w.store.recordCommitFailure()
_ = w.abortLocked()
return fmt.Errorf("recorder: close body file: %w", err)
}
if err := os.Rename(w.partialPath, w.finalPath); err != nil {
w.store.recordCommitFailure()
_ = w.abortLocked()
return fmt.Errorf("recorder: publish body file: %w", err)
}
if w.store.syncOnCommit {
if err := syncDirectory(w.store.assetDir); err != nil {
w.store.recordCommitFailure()
_ = os.Remove(w.finalPath)
_ = w.abortLocked()
return fmt.Errorf("recorder: sync body asset directory: %w", err)
}
}
w.committed = true
w.store.commitFile(w.written)
return nil
}
func (w *fileBodyWriter) Abort() error {
w.mu.Lock()
defer w.mu.Unlock()
return w.abortLocked()
}
func (w *fileBodyWriter) abortLocked() error {
if w.aborted || w.committed {
return nil
}
w.aborted = true
closeErr := w.f.Close()
removeErr := os.Remove(w.partialPath)
if errors.Is(removeErr, os.ErrNotExist) {
removeErr = nil
}
w.store.abortFile(w.written)
if closeErr != nil {
w.store.recordAbortFailure()
return closeErr
}
if removeErr != nil {
w.store.recordAbortFailure()
}
return removeErr
}
func (w *fileBodyWriter) Bytes() ([]byte, error) {
w.mu.Lock()
defer w.mu.Unlock()
path := w.partialPath
if w.committed {
path = w.finalPath
}
b, err := os.ReadFile(path)
if err != nil {
return nil, fmt.Errorf("recorder: read body file: %w", err)
}
return b, nil
}
func (w *fileBodyWriter) Ref() string {
w.mu.Lock()
defer w.mu.Unlock()
if !w.committed {
return ""
}
return w.ref
}
func (s *FileBodyStore) initialize() error {
if s.maintenance {
for _, dir := range []string{s.root, s.partialDir, s.assetDir} {
info, err := os.Lstat(dir)
if err != nil {
return fmt.Errorf("recorder: open body store for maintenance: %w", err)
}
if info.Mode()&os.ModeSymlink != 0 || !info.IsDir() {
return errors.New("recorder: maintenance body store paths must be directories without symbolic links")
}
}
} else {
if err := s.initializeDirectories(); err != nil {
return err
}
}
return s.scanExisting()
}
func (s *FileBodyStore) initializeDirectories() error {
if err := os.MkdirAll(s.root, 0o700); err != nil {
return fmt.Errorf("recorder: create body store directory: %w", err)
}
if err := os.MkdirAll(s.partialDir, 0o700); err != nil {
return fmt.Errorf("recorder: create body partial directory: %w", err)
}
if err := os.MkdirAll(s.assetDir, 0o700); err != nil {
return fmt.Errorf("recorder: create body asset directory: %w", err)
}
for _, dir := range []string{s.root, s.partialDir, s.assetDir} {
if err := os.Chmod(dir, 0o700); err != nil {
return fmt.Errorf("recorder: secure body store directory: %w", err)
}
}
return nil
}
func (s *FileBodyStore) scanExisting() error {
now := time.Now()
partials, err := os.ReadDir(s.partialDir)
if err != nil {
return fmt.Errorf("recorder: scan body partial directory: %w", err)
}
for _, entry := range partials {
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".partial") {
continue
}
if entry.Type()&os.ModeSymlink != 0 {
return errors.New("recorder: symlink found in body partial directory")
}
path := filepath.Join(s.partialDir, entry.Name())
info, infoErr := entry.Info()
if infoErr != nil {
return fmt.Errorf("recorder: stat partial body asset: %w", infoErr)
}
if !s.maintenance && (s.partialTTL == 0 || now.Sub(info.ModTime()) >= s.partialTTL) {
if removeErr := os.Remove(path); removeErr != nil {
return fmt.Errorf("recorder: recover partial body asset: %w", removeErr)
}
s.stats.RecoveredPartials++
continue
}
s.stats.PartialFiles++
s.stats.PartialBytes += info.Size()
}
assets, err := os.ReadDir(s.assetDir)
if err != nil {
return fmt.Errorf("recorder: scan body asset directory: %w", err)
}
for _, entry := range assets {
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".body") {
continue
}
if entry.Type()&os.ModeSymlink != 0 {
return errors.New("recorder: symlink found in body asset directory")
}
info, infoErr := entry.Info()
if infoErr != nil {
return fmt.Errorf("recorder: stat body asset: %w", infoErr)
}
s.stats.CommittedFiles++
s.stats.CommittedBytes += info.Size()
}
if s.stats.PartialBytes+s.stats.CommittedBytes > s.maxBytes ||
s.stats.PartialFiles+s.stats.CommittedFiles > s.maxFiles {
return errors.New("recorder: existing body assets exceed configured quota")
}
return nil
}
func (s *FileBodyStore) reserveBytes(n int64) error {
s.mu.Lock()
defer s.mu.Unlock()
if s.stats.PartialBytes+s.stats.CommittedBytes+n > s.maxBytes {
s.stats.QuotaRejected++
return errors.New("recorder: file body store byte quota exceeded")
}
s.stats.PartialBytes += n
return nil
}
func (s *FileBodyStore) releasePartialBytes(n int64) {
s.mu.Lock()
s.stats.PartialBytes -= n
s.mu.Unlock()
}
func (s *FileBodyStore) commitFile(size int64) {
s.mu.Lock()
s.stats.PartialFiles--
s.stats.PartialBytes -= size
s.stats.CommittedFiles++
s.stats.CommittedBytes += size
s.stats.CommittedTotal++
s.mu.Unlock()
}
func (s *FileBodyStore) abortFile(size int64) {
s.mu.Lock()
s.stats.PartialFiles--
s.stats.PartialBytes -= size
s.stats.AbortedTotal++
s.mu.Unlock()
}
func (s *FileBodyStore) recordWriteFailure() {
s.mu.Lock()
s.stats.WriteFailures++
s.mu.Unlock()
}
func (s *FileBodyStore) recordCommitFailure() {
s.mu.Lock()
s.stats.CommitFailures++
s.mu.Unlock()
}
func (s *FileBodyStore) recordAbortFailure() {
s.mu.Lock()
s.stats.AbortFailures++
s.mu.Unlock()
}
// Stats returns an atomic point-in-time lifecycle and capacity snapshot.
func (s *FileBodyStore) Stats() FileBodyStoreStats {
s.mu.Lock()
defer s.mu.Unlock()
stats := s.stats
stats.MaxBytes = s.maxBytes
stats.MaxFiles = s.maxFiles
return stats
}
// Open opens a committed opaque body reference for reading.
func (s *FileBodyStore) Open(ref string) (io.ReadCloser, error) {
path, err := s.resolveRef(ref)
if err != nil {
return nil, err
}
if _, err := regularFileInfo(path); err != nil {
return nil, fmt.Errorf("recorder: validate body asset: %w", err)
}
f, err := os.Open(path)
if err != nil {
return nil, fmt.Errorf("recorder: open body asset: %w", err)
}
return f, nil
}
// Release removes one committed body asset. Missing assets are treated as an
// idempotent success.
func (s *FileBodyStore) Release(ref string) error {
path, err := s.resolveRef(ref)
if err != nil {
return err
}
s.mu.Lock()
defer s.mu.Unlock()
info, statErr := regularFileInfo(path)
if errors.Is(statErr, os.ErrNotExist) {
return nil
}
if statErr != nil {
s.stats.ReleaseFailures++
return fmt.Errorf("recorder: stat body asset for release: %w", statErr)
}
if err := os.Remove(path); err != nil {
s.stats.ReleaseFailures++
return fmt.Errorf("recorder: release body asset: %w", err)
}
s.stats.CommittedFiles--
s.stats.CommittedBytes -= info.Size()
s.stats.ReleasedTotal++
return nil
}
// ReleaseEntryAssets releases the request and response body references owned
// by entry. Empty and duplicate references are ignored.
func (s *FileBodyStore) ReleaseEntryAssets(entry *Entry) error {
if entry == nil || entry.Recorder == nil {
return nil
}
refs := make(map[string]struct{}, 2)
if entry.Recorder.RequestBody != nil && entry.Recorder.RequestBody.Store != "" {
refs[entry.Recorder.RequestBody.Store] = struct{}{}
}
if entry.Recorder.ResponseBody != nil && entry.Recorder.ResponseBody.Store != "" {
refs[entry.Recorder.ResponseBody.Store] = struct{}{}
}
var errs []error
for ref := range refs {
if err := s.Release(ref); err != nil {
errs = append(errs, err)
}
}
return errors.Join(errs...)
}
// Reconcile removes committed assets absent from the authoritative liveRefs
// set and older than grace. Dry-run reports without deleting.
func (s *FileBodyStore) Reconcile(liveRefs []string, grace time.Duration, dryRun bool) (ReconcileResult, error) {
if grace < 0 {
return ReconcileResult{}, errors.New("recorder: reconcile grace must not be negative")
}
live := make(map[string]struct{}, len(liveRefs))
for _, ref := range liveRefs {
if _, err := s.resolveRef(ref); err != nil {
return ReconcileResult{}, err
}
live[ref] = struct{}{}
}
entries, err := os.ReadDir(s.assetDir)
if err != nil {
return ReconcileResult{}, fmt.Errorf("recorder: scan body assets for reconciliation: %w", err)
}
var result ReconcileResult
now := time.Now()
for _, entry := range entries {
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".body") {
continue
}
if entry.Type()&os.ModeSymlink != 0 {
return result, errors.New("recorder: symlink found in body asset directory")
}
result.Scanned++
id := strings.TrimSuffix(entry.Name(), ".body")
ref := fileBodyRefPrefix + id
if _, ok := live[ref]; ok {
continue
}
info, infoErr := entry.Info()
if infoErr != nil {
return result, fmt.Errorf("recorder: stat body asset for reconciliation: %w", infoErr)
}
if now.Sub(info.ModTime()) < grace {
continue
}
result.Released++
result.ReleasedBytes += info.Size()
if !dryRun {
if releaseErr := s.Release(ref); releaseErr != nil {
return result, releaseErr
}
}
}
return result, nil
}
func (s *FileBodyStore) resolveRef(ref string) (string, error) {
if !strings.HasPrefix(ref, fileBodyRefPrefix) {
return "", errors.New("recorder: invalid file body reference")
}
id := strings.TrimPrefix(ref, fileBodyRefPrefix)
if len(id) != 32 {
return "", errors.New("recorder: invalid file body reference")
}
for _, char := range id {
if !strings.ContainsRune("0123456789abcdef", char) {
return "", errors.New("recorder: invalid file body reference")
}
}
return filepath.Join(s.assetDir, id+".body"), nil
}
func syncDirectory(path string) error {
dir, err := os.Open(path)
if err != nil {
return err
}
syncErr := dir.Sync()
closeErr := dir.Close()
if syncErr != nil {
return syncErr
}
return closeErr
}
func regularFileInfo(path string) (os.FileInfo, error) {
info, err := os.Lstat(path)
if err != nil {
return nil, err
}
if !info.Mode().IsRegular() {
return nil, errors.New("body asset is not a regular file")
}
return info, nil
}
package recorder
import "io"
// byteSink caches the optional io.ByteWriter fast path once. Its reusable
// fallback buffer avoids converting each emitted byte into a fresh escaping
// []byte when the destination only implements io.Writer.
type byteSink struct {
dst io.Writer
byte io.ByteWriter
one [1]byte
}
func newByteSink(dst io.Writer) byteSink {
byteWriter, _ := dst.(io.ByteWriter)
return byteSink{dst: dst, byte: byteWriter}
}
func (w *byteSink) WriteByte(b byte) error {
if w.byte != nil {
return w.byte.WriteByte(b)
}
w.one[0] = b
n, err := w.dst.Write(w.one[:])
if err == nil && n != len(w.one) {
return io.ErrShortWrite
}
return err
}
package recorder
import (
"context"
"fmt"
"net/http"
)
// BodyDirection identifies which side of an exchange a capture decision
// applies to.
type BodyDirection string
const (
// RequestBody identifies the caller-provided request stream.
RequestBody BodyDirection = "request"
// ResponseBody identifies the response stream returned to the caller.
ResponseBody BodyDirection = "response"
)
// BodyCaptureMeta is the immutable, non-body metadata supplied to a
// BodyCapturePolicy. It excludes query values, headers, and cookies.
// StatusCode is zero for request decisions; Path is URL-escaped.
type BodyCaptureMeta struct {
Direction BodyDirection
Method string
Scheme string
Host string
Path string
StatusCode int
ContentType string
ContentEncoding string
ContentLength int64
TraceID string
RedirectIndex int
}
// BodyCaptureDecision controls capture for one request or response body.
// RedactorOverride, when non-nil, replaces the redactor selected by the
// Transport and request-scoped RedactionConfig for this body only. Nil keeps
// the existing selection; it does not disable redaction.
type BodyCaptureDecision struct {
Capture bool
Embed bool
Hash bool
MaxBodyBytes int64
RedactorOverride BodyRedactor
}
// BodyCapturePolicy decides how one body is recorded. The defaults argument
// reflects the transport's capture Config; RedactorOverride starts nil
// because RedactionConfig selection remains active unless explicitly
// overridden. Implementations may be called concurrently and must not retain
// or mutate HTTP objects.
type BodyCapturePolicy func(context.Context, BodyCaptureMeta, BodyCaptureDecision) (BodyCaptureDecision, error)
func normalizeCaptureDecision(d BodyCaptureDecision) BodyCaptureDecision {
if !d.Capture {
d.Embed = false
d.RedactorOverride = nil
}
return d
}
func decideBodyCapture(ctx context.Context, policy BodyCapturePolicy, meta BodyCaptureMeta, defaults BodyCaptureDecision) (decision BodyCaptureDecision, err error) {
defaults = normalizeCaptureDecision(defaults)
if policy == nil {
return defaults, nil
}
defer func() {
if recovered := recover(); recovered != nil {
decision = BodyCaptureDecision{}
err = fmt.Errorf("recorder: panic in BodyCapturePolicy: %v", recovered)
}
}()
decision, err = policy(ctx, meta, defaults)
if err != nil {
return BodyCaptureDecision{}, fmt.Errorf("recorder: body capture policy: %w", err)
}
return normalizeCaptureDecision(decision), nil
}
func requestCaptureMeta(req *http.Request, traceID string, redirectIndex int) BodyCaptureMeta {
meta := BodyCaptureMeta{
Direction: RequestBody,
Method: req.Method,
ContentLength: req.ContentLength,
TraceID: traceID,
RedirectIndex: redirectIndex,
}
if req.URL != nil {
meta.Scheme = req.URL.Scheme
meta.Host = req.URL.Host
meta.Path = req.URL.EscapedPath()
if meta.Path == "" {
meta.Path = "/"
}
}
meta.ContentType = req.Header.Get("Content-Type")
meta.ContentEncoding = req.Header.Get("Content-Encoding")
return meta
}
func responseCaptureMeta(req *http.Request, resp *http.Response, traceID string, redirectIndex int) BodyCaptureMeta {
meta := requestCaptureMeta(req, traceID, redirectIndex)
meta.Direction = ResponseBody
meta.StatusCode = resp.StatusCode
meta.ContentType = resp.Header.Get("Content-Type")
meta.ContentEncoding = resp.Header.Get("Content-Encoding")
meta.ContentLength = resp.ContentLength
return meta
}
package recorder
import (
"bufio"
"compress/flate"
"compress/gzip"
"compress/zlib"
"errors"
"fmt"
"io"
"sync"
)
// ContentDecoder turns a compressed body stream into its decoded form. It is
// used by the recording pipeline. With body redaction enabled, a captured
// request or response carrying a registered Content-Encoding is decoded and
// redacted before bytes reach the BodyStore. Otherwise, a fully captured
// response may be decoded when embedded in the HAR. Decoded HAR response
// content is marked by _recorder.responseBodyDecoded while bodySize, the body
// hash and body counters keep describing the encoded bytes observed by the
// caller. The live request and caller-visible response bytes are never touched.
//
// Decoders for encodings outside the standard library (brotli, zstd) are
// deliberately not bundled — the module stays dependency-free. Registering
// one is a few lines with the de-facto standard implementations. A complete,
// independently pinned and tested example is under
// docs/examples/content-decoders:
//
// import "github.com/andybalholm/brotli"
//
// config.ContentDecoders["br"] = func(r io.Reader) (io.ReadCloser, error) {
// return io.NopCloser(brotli.NewReader(r)), nil
// }
//
// import "github.com/klauspost/compress/zstd"
//
// config.ContentDecoders["zstd"] = func(r io.Reader) (io.ReadCloser, error) {
// zr, err := zstd.NewReader(r)
// if err != nil {
// return nil, err
// }
// return zr.IOReadCloser(), nil
// }
type ContentDecoder func(io.Reader) (io.ReadCloser, error)
var errDecodedBodyTooLarge = errors.New("recorder: decoded body exceeds capture limit")
// decodingRedactingBodyWriter streams encoded input through a ContentDecoder
// and then through the structured redactor before it reaches the BodyStore.
// The pipe intentionally applies backpressure: memory remains bounded by the
// decoder, parser state, and io.Pipe's synchronous handoff. As with entry
// finalization, the caller must read or close the HTTP body; otherwise the
// capture pipeline, including this worker, remains live with that body.
type decodingRedactingBodyWriter struct {
BodyWriter
pw *io.PipeWriter
done chan error
mu sync.Mutex
finalized bool
}
func newDecodingRedactingBodyWriter(dst BodyWriter, decoder ContentDecoder, mimeType string, red *redactor, limit int64) BodyWriter {
pr, pw := io.Pipe()
w := &decodingRedactingBodyWriter{BodyWriter: dst, pw: pw, done: make(chan error, 1)}
go func() {
decoded, err := decoder(pr)
if err != nil {
_ = pr.CloseWithError(err)
w.done <- fmt.Errorf("recorder: open streaming body decoder: %w", err)
return
}
buffered := bufio.NewWriterSize(dst, 32<<10)
sr := newBodyStreamRedactor(buffered, mimeType, red)
if sr == nil {
err := errors.New("recorder: selected body redactor is unavailable")
_ = decoded.Close()
_ = pr.CloseWithError(err)
w.done <- err
return
}
var copied int64
buf := make([]byte, 32<<10)
for err == nil {
var n int
n, err = decoded.Read(buf)
if n > 0 {
if limit > 0 && copied+int64(n) > limit {
err = errDecodedBodyTooLarge
break
}
copied += int64(n)
if _, writeErr := sr.Write(buf[:n]); writeErr != nil {
err = writeErr
break
}
}
}
if errors.Is(err, io.EOF) {
err = nil
}
if closeErr := decoded.Close(); err == nil {
err = closeErr
}
if closeErr := sr.Close(); err == nil {
err = closeErr
}
if flushErr := buffered.Flush(); err == nil {
err = flushErr
}
_ = pr.CloseWithError(err)
w.done <- err
}()
return w
}
func (w *decodingRedactingBodyWriter) Write(p []byte) (int, error) {
return w.pw.Write(p)
}
func (w *decodingRedactingBodyWriter) Commit() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.finalized {
return nil
}
w.finalized = true
pipeErr := w.pw.Close()
decodeErr := <-w.done
if decodeErr != nil {
_ = w.BodyWriter.Abort()
return decodeErr
}
if pipeErr != nil {
_ = w.BodyWriter.Abort()
return pipeErr
}
if err := w.BodyWriter.Commit(); err != nil {
_ = w.BodyWriter.Abort()
return err
}
return nil
}
func (w *decodingRedactingBodyWriter) Abort() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.finalized {
return nil
}
w.finalized = true
_ = w.pw.CloseWithError(errors.New("recorder: body capture aborted"))
<-w.done
return w.BodyWriter.Abort()
}
// GzipDecoder decodes gzip content using the standard library. Registered by
// default for "gzip" and "x-gzip".
func GzipDecoder(r io.Reader) (io.ReadCloser, error) {
return gzip.NewReader(r)
}
// DeflateDecoder decodes "deflate" content using the standard library.
// Registered by default. HTTP's deflate is formally zlib-wrapped (RFC 9110),
// but a number of servers send raw DEFLATE streams; the decoder sniffs the
// zlib header and handles both, like browsers do.
func DeflateDecoder(r io.Reader) (io.ReadCloser, error) {
br := bufio.NewReader(r)
hdr, err := br.Peek(2)
if err == nil && len(hdr) == 2 && hdr[0]&0x0f == 8 && (uint16(hdr[0])<<8|uint16(hdr[1]))%31 == 0 {
return zlib.NewReader(br)
}
return flate.NewReader(br), nil
}
// defaultContentDecoders returns the stdlib-only decoder set installed by
// DefaultConfig.
func defaultContentDecoders() map[string]ContentDecoder {
return map[string]ContentDecoder{
"gzip": GzipDecoder,
"x-gzip": GzipDecoder,
"deflate": DeflateDecoder,
}
}
package recorder
import (
"encoding/json"
"errors"
"fmt"
"net"
"net/http"
"net/url"
"strings"
"sync"
)
// ErrDebugStreamRecorderClosed is returned when Record is called after the
// development stream has been closed.
var ErrDebugStreamRecorderClosed = errors.New("recorder: debug stream recorder is closed")
// DebugStreamRecorderConfig controls the bounded, single-subscriber
// development stream. QueueCapacity must be positive. AllowedOrigins adds
// exact HTTP(S) browser origins to the loopback origins accepted by default.
type DebugStreamRecorderConfig struct {
QueueCapacity int
AllowedOrigins []string
}
// DefaultDebugStreamRecorderConfig returns conservative settings for local
// interactive inspection.
func DefaultDebugStreamRecorderConfig() DebugStreamRecorderConfig {
return DebugStreamRecorderConfig{QueueCapacity: 256}
}
// DebugStreamRecorderStats is a point-in-time view of the development stream.
type DebugStreamRecorderStats struct {
SubscriberActive bool
Published uint64
Dropped uint64
}
type debugStreamMessage struct {
id uint64
data []byte
}
// DebugStreamRecorder publishes finalized entries as server-sent events to
// one local subscriber. It is intended only for live local development and
// debugging: its bounded queue retains recent entries while disconnected,
// drops the oldest queued entry when full, and is not a durable evidence sink.
//
// DebugStreamRecorder is an http.Handler. The application owns the HTTP
// server and its lifecycle; ServeHTTP accepts only loopback peers. Browser
// origins are restricted to loopback unless explicitly configured.
type DebugStreamRecorder struct {
mu sync.Mutex
queue chan debugStreamMessage
nextID uint64
published uint64
dropped uint64
pendingDrops uint64
subscriber bool
closed bool
allowedOrigins map[string]struct{}
}
// NewDebugStreamRecorder creates a bounded, single-subscriber development
// stream. Use DefaultDebugStreamRecorderConfig as the baseline.
func NewDebugStreamRecorder(config DebugStreamRecorderConfig) (*DebugStreamRecorder, error) {
if config.QueueCapacity <= 0 {
return nil, errors.New("recorder: debug stream queue capacity must be positive")
}
allowedOrigins := make(map[string]struct{}, len(config.AllowedOrigins))
for i, origin := range config.AllowedOrigins {
normalized, ok := normalizeHTTPOrigin(origin)
if !ok {
return nil, fmt.Errorf("recorder: invalid debug stream allowed origin at index %d", i)
}
allowedOrigins[normalized] = struct{}{}
}
return &DebugStreamRecorder{
queue: make(chan debugStreamMessage, config.QueueCapacity),
allowedOrigins: allowedOrigins,
}, nil
}
// Record implements Recorder. Entries wait in the bounded queue until the
// subscriber receives them. When the queue is full, Record drops its oldest
// entry and reports that loss to the next subscriber as a gap event.
func (r *DebugStreamRecorder) Record(entry *Entry) error {
data, err := json.Marshal(entry)
if err != nil {
return fmt.Errorf("recorder: encode debug stream entry: %w", err)
}
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
return ErrDebugStreamRecorderClosed
}
r.nextID++
message := debugStreamMessage{id: r.nextID, data: data}
select {
case r.queue <- message:
default:
<-r.queue
r.dropped++
r.pendingDrops++
r.queue <- message
}
r.published++
return nil
}
// Stats returns a concurrency-safe snapshot.
func (r *DebugStreamRecorder) Stats() DebugStreamRecorderStats {
r.mu.Lock()
defer r.mu.Unlock()
return DebugStreamRecorderStats{
SubscriberActive: r.subscriber,
Published: r.published,
Dropped: r.dropped,
}
}
// Close disconnects the active subscriber and rejects future records. It is
// safe to call more than once.
func (r *DebugStreamRecorder) Close() error {
r.mu.Lock()
defer r.mu.Unlock()
if r.closed {
return nil
}
r.closed = true
close(r.queue)
return nil
}
// ServeHTTP streams entry and gap events. A second concurrent subscriber gets
// HTTP 409. Requests from non-loopback peers are always rejected. Browser
// origins must be loopback or explicitly listed in AllowedOrigins.
func (r *DebugStreamRecorder) ServeHTTP(w http.ResponseWriter, request *http.Request) {
if request.Method != http.MethodGet {
w.Header().Set("Allow", http.MethodGet)
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
return
}
if !isLoopbackRemote(request.RemoteAddr) {
http.Error(w, "debug stream is restricted to loopback clients", http.StatusForbidden)
return
}
origin := request.Header.Get("Origin")
if origin != "" {
if !r.isOriginAllowed(origin) {
http.Error(w, "debug stream origin is not allowed", http.StatusForbidden)
return
}
w.Header().Set("Access-Control-Allow-Origin", origin)
w.Header().Set("Vary", "Origin")
}
flusher, ok := w.(http.Flusher)
if !ok {
http.Error(w, "streaming is not supported", http.StatusInternalServerError)
return
}
queue, ok := r.subscribe()
if !ok {
http.Error(w, "a debug stream subscriber is already connected", http.StatusConflict)
return
}
defer r.unsubscribe(queue)
w.Header().Set("Content-Type", "text/event-stream")
w.Header().Set("Cache-Control", "no-cache, no-store")
w.Header().Set("X-Accel-Buffering", "no")
_, _ = fmt.Fprint(w, "event: ready\ndata: {}\n\n")
flusher.Flush()
for {
select {
case <-request.Context().Done():
return
case message, open := <-queue:
if !open {
return
}
if dropped := r.takePendingDrops(queue); dropped > 0 {
if _, err := fmt.Fprintf(w, "event: gap\ndata: {\"dropped\":%d}\n\n", dropped); err != nil {
return
}
}
if _, err := fmt.Fprintf(w, "id: %d\nevent: entry\ndata: %s\n\n", message.id, message.data); err != nil {
return
}
flusher.Flush()
}
}
}
func (r *DebugStreamRecorder) subscribe() (chan debugStreamMessage, bool) {
r.mu.Lock()
defer r.mu.Unlock()
if r.closed || r.subscriber {
return nil, false
}
r.subscriber = true
return r.queue, true
}
func (r *DebugStreamRecorder) unsubscribe(queue chan debugStreamMessage) {
r.mu.Lock()
defer r.mu.Unlock()
if r.queue == queue {
r.subscriber = false
}
}
func (r *DebugStreamRecorder) takePendingDrops(queue chan debugStreamMessage) uint64 {
r.mu.Lock()
defer r.mu.Unlock()
if r.queue != queue {
return 0
}
dropped := r.pendingDrops
r.pendingDrops = 0
return dropped
}
func isLoopbackRemote(address string) bool {
host, _, err := net.SplitHostPort(address)
if err != nil {
return false
}
ip := net.ParseIP(strings.Trim(host, "[]"))
return ip != nil && ip.IsLoopback()
}
func isLoopbackOrigin(origin string) bool {
normalized, ok := normalizeHTTPOrigin(origin)
if !ok {
return false
}
parsed, _ := url.Parse(normalized)
host := parsed.Hostname()
if strings.EqualFold(host, "localhost") {
return true
}
ip := net.ParseIP(host)
return ip != nil && ip.IsLoopback()
}
func (r *DebugStreamRecorder) isOriginAllowed(origin string) bool {
if isLoopbackOrigin(origin) {
return true
}
normalized, ok := normalizeHTTPOrigin(origin)
if !ok {
return false
}
_, ok = r.allowedOrigins[normalized]
return ok
}
func normalizeHTTPOrigin(origin string) (string, bool) {
parsed, err := url.Parse(origin)
if err != nil || parsed.User != nil || parsed.Opaque != "" || parsed.Host == "" ||
(parsed.Scheme != "http" && parsed.Scheme != "https") ||
(parsed.Path != "" && parsed.Path != "/") || parsed.RawQuery != "" || parsed.Fragment != "" {
return "", false
}
hostname := strings.ToLower(parsed.Hostname())
if hostname == "" {
return "", false
}
port := parsed.Port()
if (parsed.Scheme == "http" && port == "80") || (parsed.Scheme == "https" && port == "443") {
port = ""
}
host := hostname
if strings.Contains(hostname, ":") {
host = "[" + hostname + "]"
}
if port != "" {
host = net.JoinHostPort(hostname, port)
}
return parsed.Scheme + "://" + host, true
}
package recorder
import (
"context"
"crypto/tls"
"crypto/x509"
"errors"
"fmt"
"net"
"strings"
"syscall"
)
// Error phases. The phase names the step of the HTTP exchange in which the
// failure was observed.
const (
PhaseRequestSetup = "request_setup"
PhaseDNS = "dns"
PhaseConnect = "connect"
PhaseProxy = "proxy"
PhaseTLS = "tls"
PhaseWriteRequest = "write_request"
PhaseWriteRequestBody = "write_request_body"
PhaseWaitResponse = "wait_response"
PhaseReadResponseHeaders = "read_response_headers"
PhaseReadResponseBody = "read_response_body"
PhaseRedirect = "redirect"
PhaseContext = "context"
PhaseUnknown = "unknown"
)
// ErrorInfo is the structured _recorder.error value describing a transport or
// body-stream failure.
type ErrorInfo struct {
// Phase is one of the Phase* constants.
Phase string `json:"phase"`
// Type is the innermost observed Go error type.
Type string `json:"type"`
// Message is the configured redacted top-level error text.
Message string `json:"message"`
// Timeout and Temporary preserve matching error interface facts.
Timeout bool `json:"timeout"`
Temporary bool `json:"temporary"`
// ContextCanceled and ContextDeadlineExceeded describe the request context,
// not merely an error that happens to match a context sentinel.
ContextCanceled bool `json:"contextCanceled"`
ContextDeadlineExceeded bool `json:"contextDeadlineExceeded"`
// Cause is the configured redacted context cause, when present.
Cause string `json:"cause,omitempty"`
// UnwrapChain lists bounded Go error type names from outermost to innermost.
UnwrapChain []string `json:"unwrapChain,omitempty"`
}
const maxUnwrapDepth = 32
// unwrapChain walks the error chain (following the first branch of joined
// errors), depth-capped so hostile Unwrap implementations cannot loop us.
func unwrapChain(err error) []error {
var chain []error
for e := err; e != nil && len(chain) < maxUnwrapDepth; {
chain = append(chain, e)
switch u := e.(type) {
case interface{ Unwrap() error }:
e = u.Unwrap()
case interface{ Unwrap() []error }:
if us := u.Unwrap(); len(us) > 0 {
e = us[0]
} else {
e = nil
}
default:
e = nil
}
}
return chain
}
// newErrorInfo builds the structured error record. ctxErr is the request
// context's Err() at failure time (nil when the context was still live);
// cause is the corresponding context.Cause. The ContextCanceled /
// ContextDeadlineExceeded flags require the request context to have actually
// fired: net/http timeout errors (e.g. ResponseHeaderTimeout's
// *http.timeoutError) deliberately match context errors via errors.Is, and
// must not masquerade as a caller-initiated cancellation.
func newErrorInfo(err error, phase string, red *redactor, ctxErr, cause error) *ErrorInfo {
if err == nil {
return nil
}
chain := unwrapChain(err)
info := &ErrorInfo{
Phase: phase,
Message: red.redactError(err.Error()),
ContextCanceled: ctxErr != nil && errors.Is(err, context.Canceled),
ContextDeadlineExceeded: ctxErr != nil && errors.Is(err, context.DeadlineExceeded),
}
info.UnwrapChain = make([]string, len(chain))
for i, e := range chain {
info.UnwrapChain[i] = fmt.Sprintf("%T", e)
}
info.Type = info.UnwrapChain[len(info.UnwrapChain)-1]
for _, e := range chain {
if t, ok := e.(interface{ Timeout() bool }); ok && t.Timeout() {
info.Timeout = true
}
if t, ok := e.(interface{ Temporary() bool }); ok && t.Temporary() {
info.Temporary = true
}
}
if info.ContextDeadlineExceeded {
info.Timeout = true
}
if cause != nil {
info.Cause = red.redactError(cause.Error())
}
return info
}
// classifyPhase determines the failure phase for an error returned by the
// underlying RoundTripper. Typed matching (errors.As / errors.Is) comes
// first; the httptrace progress recorded in v is the fallback. String
// matching is a last resort only, for errors that expose no type at all
// (the redirect-loop error assembled by http.Client). ctxDone reports
// whether the request's own context had fired when the error was observed.
func classifyPhase(err error, v traceView, hasProxy, respReceived, ctxDone bool, reqBodyReadErr error) string {
if err == nil {
return ""
}
// A read failure recorded on the caller-supplied request body means the
// transport aborted while streaming the body upward.
if reqBodyReadErr != nil && !respReceived {
return PhaseWriteRequestBody
}
var dnsErr *net.DNSError
if errors.As(err, &dnsErr) {
return PhaseDNS
}
if isTLSError(err) {
return PhaseTLS
}
var opErr *net.OpError
if errors.As(err, &opErr) {
switch opErr.Op {
case "dial":
if hasProxy {
return PhaseProxy
}
return PhaseConnect
case "write":
if !v.wroteHeaders.IsZero() {
return PhaseWriteRequestBody
}
return PhaseWriteRequest
case "read":
return readPhase(v, respReceived)
}
}
if errors.Is(err, syscall.ECONNREFUSED) {
if hasProxy {
return PhaseProxy
}
return PhaseConnect
}
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
// Prefer the network step that was in flight when the context fired;
// otherwise attribute the failure to the context itself — but only
// when the request's own context really fired. Transport-internal
// timeouts (net/http's ResponseHeaderTimeout error) merely match
// context errors via errors.Is and belong to the phase the exchange
// actually stood in.
switch p := phaseFromTrace(v, respReceived); p {
case PhaseDNS, PhaseConnect, PhaseTLS:
return p
}
if ctxDone {
return PhaseContext
}
return phaseFromTrace(v, respReceived)
}
// Last resort string match: http.Client's redirect-loop error carries no
// typed error to match on.
if msg := err.Error(); strings.Contains(msg, "stopped after") && strings.Contains(msg, "redirect") {
return PhaseRedirect
}
return phaseFromTrace(v, respReceived)
}
func readPhase(v traceView, respReceived bool) string {
if respReceived {
return PhaseReadResponseBody
}
if !v.firstByte.IsZero() {
return PhaseReadResponseHeaders
}
return PhaseWaitResponse
}
// phaseFromTrace derives the failure phase from how far the exchange
// progressed according to httptrace events.
func phaseFromTrace(v traceView, respReceived bool) string {
switch {
case respReceived:
return PhaseReadResponseBody
case !v.firstByte.IsZero():
return PhaseReadResponseHeaders
case !v.wroteRequest.IsZero() || !v.wroteHeaders.IsZero():
return PhaseWaitResponse
case !v.tlsStart.IsZero() && v.tlsDone.IsZero():
return PhaseTLS
case !v.connectStart.IsZero() && v.connectDone.IsZero():
return PhaseConnect
case !v.dnsStart.IsZero() && v.dnsDone.IsZero():
return PhaseDNS
case !v.gotConn.IsZero():
return PhaseWriteRequest
case v.getConn.IsZero() && v.dnsStart.IsZero() && v.connectStart.IsZero():
return PhaseRequestSetup
default:
return PhaseUnknown
}
}
// isTLSError reports whether the chain contains a TLS or certificate
// verification error. Typed checks first; the package prefix of unexported
// types (tls alerts, net/http's tlsHandshakeTimeoutError) is the fallback.
func isTLSError(err error) bool {
var (
hostnameErr x509.HostnameError
caErr x509.UnknownAuthorityError
invalidErr x509.CertificateInvalidError
rootsErr x509.SystemRootsError
recordErr tls.RecordHeaderError
verifyErr *tls.CertificateVerificationError
)
if errors.As(err, &hostnameErr) || errors.As(err, &caErr) ||
errors.As(err, &invalidErr) || errors.As(err, &rootsErr) ||
errors.As(err, &recordErr) || errors.As(err, &verifyErr) {
return true
}
for _, e := range unwrapChain(err) {
tn := fmt.Sprintf("%T", e)
if strings.HasPrefix(tn, "tls.") || strings.HasPrefix(tn, "*tls.") ||
strings.HasPrefix(tn, "x509.") || strings.HasPrefix(tn, "*x509.") ||
strings.Contains(tn, "tlsHandshakeTimeout") {
return true
}
}
return false
}
package recorder
import (
"bytes"
"encoding/json"
"fmt"
"io"
"os"
"path/filepath"
"sync"
)
// HARFileRecorder buffers finalized entries and writes a complete HAR
// document to a file on Flush (atomically, via a temp file + rename). It is
// safe for concurrent use.
type HARFileRecorder struct {
mu sync.Mutex
path string
entries []*Entry
}
// NewHARFileRecorder creates a recorder that will write to path on Flush.
func NewHARFileRecorder(path string) *HARFileRecorder {
return &HARFileRecorder{path: path}
}
// Record implements Recorder.
func (r *HARFileRecorder) Record(e *Entry) error {
r.mu.Lock()
defer r.mu.Unlock()
r.entries = append(r.entries, e)
return nil
}
// RecordBatch processes a batch with one lock acquisition.
func (r *HARFileRecorder) RecordBatch(entries []*Entry) error {
r.mu.Lock()
defer r.mu.Unlock()
r.entries = append(r.entries, entries...)
return nil
}
// Flush writes the full HAR document collected so far. It can be called any
// number of times; each call rewrites the file atomically.
func (r *HARFileRecorder) Flush() error {
r.mu.Lock()
entries := append([]*Entry(nil), r.entries...)
r.mu.Unlock()
har := NewHAR(entries)
tmp, err := os.CreateTemp(filepath.Dir(r.path), ".recorder-*.har")
if err != nil {
return fmt.Errorf("recorder: create HAR temp file: %w", err)
}
defer func() { _ = os.Remove(tmp.Name()) }() // no-op after a successful rename
if err := har.Write(tmp); err != nil {
_ = tmp.Close()
return err
}
if err := tmp.Close(); err != nil {
return fmt.Errorf("recorder: close HAR temp file: %w", err)
}
if err := os.Rename(tmp.Name(), r.path); err != nil {
return fmt.Errorf("recorder: rename HAR file: %w", err)
}
return nil
}
// Close flushes the document; it satisfies io.Closer for defer-friendly use.
func (r *HARFileRecorder) Close() error { return r.Flush() }
// EntriesByTrace implements TraceStore: entries are buffered until Flush, so
// they remain queryable by trace ID.
func (r *HARFileRecorder) EntriesByTrace(traceID string) []*Entry {
r.mu.Lock()
defer r.mu.Unlock()
return filterTrace(r.entries, traceID)
}
// RemoveTrace implements TraceStore. Removed entries are excluded from every
// subsequent Flush.
func (r *HARFileRecorder) RemoveTrace(traceID string) int {
removed, _ := r.remove(traceID, false)
return removed
}
// TakeTrace implements TraceStore: it atomically removes and returns the
// trace's entries. The taken entries are excluded from every subsequent
// Flush — use this to split one shared recording into per-call HAR files
// (recorder.NewHAR(taken).Write(...)).
func (r *HARFileRecorder) TakeTrace(traceID string) []*Entry {
_, taken := r.remove(traceID, true)
return taken
}
func (r *HARFileRecorder) remove(traceID string, collect bool) (int, []*Entry) {
r.mu.Lock()
defer r.mu.Unlock()
kept, taken, removed := splitTrace(r.entries, traceID, collect)
r.entries = kept
return removed, taken
}
// JSONStreamRecorder writes every finalized entry immediately as one JSON
// document per line (NDJSON) to the underlying writer. This is the streaming
// export path: it deliberately does not emit the enclosing HAR wrapper, so
// the top-level HAR JSON structure is never left half-written; consumers can
// wrap the lines into a log object themselves. Safe for concurrent use.
//
// JSONStreamRecorder intentionally does not implement TraceStore: entries
// leave the process the moment they are recorded, so there is nothing left
// to query or remove. Group downstream by each line's _recorder.traceId, or
// use a retaining recorder (MemoryRecorder, HARFileRecorder) when per-trace
// access is needed.
type JSONStreamRecorder struct {
mu sync.Mutex
enc *json.Encoder
w io.Writer
}
// NewJSONStreamRecorder creates a streaming recorder writing to w.
func NewJSONStreamRecorder(w io.Writer) *JSONStreamRecorder {
return &JSONStreamRecorder{enc: json.NewEncoder(w), w: w}
}
// Record implements Recorder and reports encoding or write failures directly.
func (r *JSONStreamRecorder) Record(e *Entry) error {
r.mu.Lock()
defer r.mu.Unlock()
if err := r.enc.Encode(e); err != nil {
return fmt.Errorf("recorder: encode entry: %w", err)
}
return nil
}
// RecordBatch encodes independent NDJSON documents into one temporary buffer
// and issues one downstream Write.
func (r *JSONStreamRecorder) RecordBatch(entries []*Entry) error {
r.mu.Lock()
defer r.mu.Unlock()
if len(entries) == 0 {
return nil
}
var buffer bytes.Buffer
encoder := json.NewEncoder(&buffer)
for _, entry := range entries {
if err := encoder.Encode(entry); err != nil {
return fmt.Errorf("recorder: encode entry batch: %w", err)
}
}
n, err := r.w.Write(buffer.Bytes())
if err != nil {
return fmt.Errorf("recorder: write entry batch: %w", err)
}
if n != buffer.Len() {
return fmt.Errorf("recorder: write entry batch: %w", io.ErrShortWrite)
}
return nil
}
package recorder
import (
"bytes"
"io"
"net/url"
"strings"
"unicode/utf8"
)
const maxFormKeyBytes = 64 << 10
const formRedactedValue = "%5B" + redactedMarker + "%5D"
// formStreamRedactor preserves the application/x-www-form-urlencoded wire
// representation except for values whose decoded field name matches a
// configured query-parameter rule. Only the current raw key is buffered;
// matched values are discarded as they stream.
type formStreamRedactor struct {
dst io.Writer
bytes byteSink
fields map[string]struct{}
key []byte
keyFold []byte
inValue bool
suppress bool
err error
protected protectedValueBuffer
}
func newFormStreamRedactor(dst io.Writer, fields map[string]struct{}, protectors ...*bodyValueProtector) *formStreamRedactor {
protector := newBodyValueProtector(nil)
if len(protectors) > 0 && protectors[0] != nil {
protector = protectors[0]
}
r := &formStreamRedactor{dst: dst, bytes: newByteSink(dst), fields: fields}
r.protected.reset(protector)
return r
}
func (r *formStreamRedactor) Write(p []byte) (int, error) {
if r.err != nil {
return 0, r.err
}
for i, b := range p {
if err := r.consume(b); err != nil {
r.err = err
return i, err
}
}
return len(p), nil
}
func (r *formStreamRedactor) Close() error {
if r.err != nil {
return r.err
}
if !r.inValue && len(r.key) > 0 {
_, r.err = r.dst.Write(r.key)
r.key = nil
}
if r.err == nil && r.inValue && r.suppress {
r.err = r.emitProtected()
}
return r.err
}
func (r *formStreamRedactor) consume(b byte) error {
if !r.inValue {
switch b {
case '&':
if err := r.emitKey(); err != nil {
return err
}
return r.emitByte(b)
case '=':
r.suppress = r.keyMatches()
if err := r.emitKey(); err != nil {
return err
}
if err := r.emitByte(b); err != nil {
return err
}
r.inValue = true
if r.suppress {
r.protected.reset(r.protected.session)
if r.protected.redactImmediately() {
_, err := io.WriteString(r.dst, formRedactedValue)
return err
}
}
return nil
default:
if len(r.key) >= maxFormKeyBytes {
return errRedactionLimit
}
r.key = append(r.key, b)
return nil
}
}
if b == '&' {
if r.suppress {
if err := r.emitProtected(); err != nil {
return err
}
}
r.inValue = false
r.suppress = false
return r.emitByte(b)
}
if r.suppress {
r.protected.appendByte(b)
return nil
}
return r.emitByte(b)
}
func (r *formStreamRedactor) keyMatches() bool {
escaped := false
for _, b := range r.key {
if b == '%' || b == '+' || b >= utf8.RuneSelf {
escaped = true
break
}
}
if !escaped {
folded, _ := foldASCIIName(r.key, r.keyFold[:0])
r.keyFold = folded
_, matched := r.fields[string(folded)]
return matched
}
name := string(r.key)
if decoded, err := url.QueryUnescape(name); err == nil {
name = decoded
}
_, matched := r.fields[strings.ToLower(name)]
return matched
}
func (r *formStreamRedactor) emitProtected() error {
value := r.protected.protectedBytes()
if len(value) == 0 {
return nil
}
if bytes.Equal(value, []byte(redactedValue)) {
_, err := io.WriteString(r.dst, formRedactedValue)
return err
}
_, err := r.dst.Write(value)
return err
}
func (r *formStreamRedactor) emitKey() error {
if len(r.key) == 0 {
return nil
}
_, err := r.dst.Write(r.key)
r.key = r.key[:0]
return err
}
func (r *formStreamRedactor) emitByte(b byte) error {
return r.bytes.WriteByte(b)
}
package recorder
import (
"encoding/base64"
"encoding/json"
"fmt"
"io"
"mime"
"sort"
"strings"
"time"
"unicode/utf8"
)
const (
harVersion = "1.2"
creatorName = "github.com/mgurevin/recorder"
creatorVersion = "0.5.1"
// harTimeFormat is ISO 8601 with millisecond precision as mandated by the
// HAR 1.2 specification for startedDateTime.
harTimeFormat = "2006-01-02T15:04:05.000Z07:00"
)
// HAR is the top-level HAR 1.2 document.
type HAR struct {
Log *Log `json:"log"`
}
// Log is the HAR 1.2 log object.
type Log struct {
Version string `json:"version"`
Creator *Creator `json:"creator"`
Entries []*Entry `json:"entries"`
Comment string `json:"comment,omitempty"`
}
// Creator identifies the producing application.
type Creator struct {
Name string `json:"name"`
Version string `json:"version"`
}
// Entry is a single HAR entry: one physical HTTP exchange. Recorder-specific
// data lives only in the versioned _recorder application extension; removing
// it leaves a valid plain HAR 1.2 entry.
//
// Entries produced by Transport are immutable snapshots: neither the
// Transport nor the built-in recorders mutate an Entry after it has been
// handed to Recorder.Record.
type Entry struct {
StartedDateTime string `json:"startedDateTime"`
Time float64 `json:"time"`
Request *Request `json:"request"`
Response *Response `json:"response"`
Cache *Cache `json:"cache"`
Timings *Timings `json:"timings"`
ServerIPAddress string `json:"serverIPAddress,omitempty"`
Connection string `json:"connection,omitempty"`
Comment string `json:"comment,omitempty"`
Recorder *RecorderEntryExtension `json:"_recorder,omitempty"`
// started orders entries without re-parsing StartedDateTime.
started time.Time
}
// RecorderExtensionVersion is the frozen _recorder wire-schema version.
const RecorderExtensionVersion = "1"
// RecorderEntryExtension contains every recorder-specific HAR entry field.
type RecorderEntryExtension struct {
SchemaVersion string `json:"schemaVersion"`
TraceID string `json:"traceId,omitempty"`
ExchangeID string `json:"exchangeId,omitempty"`
RedirectIndex *int `json:"redirectIndex,omitempty"`
State string `json:"state,omitempty"`
Error *ErrorInfo `json:"error,omitempty"`
Network *NetworkInfo `json:"network,omitempty"`
TLS *TLSInfo `json:"tls,omitempty"`
Expect100 *Expect100Info `json:"expect100,omitempty"`
Informational []InformationalResponse `json:"informational,omitempty"`
RequestBody *BodyInfo `json:"requestBody,omitempty"`
ResponseBody *BodyInfo `json:"responseBody,omitempty"`
RequestTrailers []NameValuePair `json:"requestTrailers,omitempty"`
ResponseTrailers []NameValuePair `json:"responseTrailers,omitempty"`
RequestTransferEncoding []string `json:"requestTransferEncoding,omitempty"`
ResponseTransferEncoding []string `json:"responseTransferEncoding,omitempty"`
RawTrace []TraceEvent `json:"trace,omitempty"`
Redaction *RedactionInfo `json:"redaction,omitempty"`
RequestBodyEncoding string `json:"requestBodyEncoding,omitempty"`
ResponseBodyDecoded bool `json:"responseBodyDecoded,omitempty"`
}
// StartTime returns the request start time used for ordering entries.
func (e *Entry) StartTime() time.Time {
if !e.started.IsZero() {
return e.started
}
if t, err := time.Parse(time.RFC3339, e.StartedDateTime); err == nil {
return t
}
return time.Time{}
}
// NameValuePair is the HAR record for headers and query parameters.
type NameValuePair struct {
Name string `json:"name"`
Value string `json:"value"`
Comment string `json:"comment,omitempty"`
}
// Cookie is the HAR cookie record.
type Cookie struct {
Name string `json:"name"`
Value string `json:"value"`
Path string `json:"path,omitempty"`
Domain string `json:"domain,omitempty"`
Expires string `json:"expires,omitempty"`
HTTPOnly bool `json:"httpOnly,omitempty"`
Secure bool `json:"secure,omitempty"`
}
// Request is the HAR request record.
type Request struct {
Method string `json:"method"`
URL string `json:"url"`
HTTPVersion string `json:"httpVersion"`
Cookies []Cookie `json:"cookies"`
Headers []NameValuePair `json:"headers"`
QueryString []NameValuePair `json:"queryString"`
PostData *PostData `json:"postData,omitempty"`
// HeadersSize is -1: the exact serialized header bytes written to the
// wire (including headers added internally by http.Transport) are not
// observable at the RoundTripper layer.
HeadersSize int64 `json:"headersSize"`
BodySize int64 `json:"bodySize"`
Comment string `json:"comment,omitempty"`
}
// PostData is the standard HAR postData record. Binary request-body encoding
// is reported by RecorderEntryExtension.RequestBodyEncoding.
type PostData struct {
MimeType string `json:"mimeType"`
Params []PostParam `json:"params,omitempty"`
Text string `json:"text,omitempty"`
encoding string
Comment string `json:"comment,omitempty"`
}
// PostParam is a single posted form parameter.
type PostParam struct {
Name string `json:"name"`
Value string `json:"value,omitempty"`
FileName string `json:"fileName,omitempty"`
ContentType string `json:"contentType,omitempty"`
}
// Response is the HAR response record. For exchanges that failed before an
// HTTP response existed, Status is 0 and _recorder.error carries the
// failure detail.
type Response struct {
Status int `json:"status"`
StatusText string `json:"statusText"`
HTTPVersion string `json:"httpVersion"`
Cookies []Cookie `json:"cookies"`
Headers []NameValuePair `json:"headers"`
Content *Content `json:"content"`
RedirectURL string `json:"redirectURL"`
// HeadersSize is -1: wire-level header size is not observable here.
HeadersSize int64 `json:"headersSize"`
// BodySize is the payload size received on the wire, or -1 when unknown
// (body not fully read, or transparently decompressed by http.Transport
// so the compressed wire size is no longer observable).
BodySize int64 `json:"bodySize"`
Comment string `json:"comment,omitempty"`
}
// Content is the HAR content record. Size is the decoded content length when
// the decoded form is known (transparent gzip by http.Transport, or a
// configured ContentDecoder), otherwise the bytes the caller actually read.
// RecorderEntryExtension.ResponseBodyDecoded reports when Text and Size
// describe the decoded form rather than raw wire bytes.
type Content struct {
Size int64 `json:"size"`
Compression int64 `json:"compression,omitempty"`
MimeType string `json:"mimeType"`
Text string `json:"text,omitempty"`
Encoding string `json:"encoding,omitempty"`
Comment string `json:"comment,omitempty"`
decoded bool
}
// Cache is the HAR cache record. This library performs no caching, so the
// object is intentionally empty (allowed by HAR 1.2).
type Cache struct{}
// Timings is the HAR timings record. All values are milliseconds; -1 means
// the phase did not apply or could not be measured (per HAR 1.2).
type Timings struct {
Blocked float64 `json:"blocked"`
DNS float64 `json:"dns"`
Connect float64 `json:"connect"`
Send float64 `json:"send"`
Wait float64 `json:"wait"`
Receive float64 `json:"receive"`
SSL float64 `json:"ssl"`
Comment string `json:"comment,omitempty"`
}
// NetworkInfo is the _recorder.network value: connection-level facts observed
// through httptrace. These describe the physical connection this exchange
// used — when a proxy is in play (Proxy != ""), RemoteAddress, IPVersion and
// DNSAddresses therefore refer to the proxy, not the origin server: the
// origin is never observable client-side through a proxy. HTTP2 is true only
// when HTTP/2 was positively observed.
type NetworkInfo struct {
DNSAddresses []string `json:"dnsAddresses,omitempty"`
// DNSCoalesced marks that this exchange's DNS answer was shared with a
// concurrent lookup for the same host (singleflight) rather than issued
// on its own.
DNSCoalesced bool `json:"dnsCoalesced,omitempty"`
Network string `json:"network,omitempty"`
LocalAddress string `json:"localAddress,omitempty"`
RemoteAddress string `json:"remoteAddress,omitempty"`
IPVersion string `json:"ipVersion,omitempty"`
ConnectionReused bool `json:"connectionReused"`
WasIdle bool `json:"wasIdle"`
IdleTimeMS float64 `json:"idleTimeMs,omitempty"`
// Proxy is the redacted proxy URL selected by a standard http.Transport.
// Custom RoundTrippers may only expose the dialed host:port.
Proxy string `json:"proxy,omitempty"`
HTTP2 bool `json:"http2"`
// PutIdle reports whether the connection went back to the idle pool
// after this exchange. Best effort: the pool return can race with entry
// finalization, so absence means "not observed", not "did not happen".
PutIdle *PutIdleInfo `json:"putIdle,omitempty"`
}
// PutIdleInfo is the structured outcome of the connection's return to the
// keep-alive pool.
type PutIdleInfo struct {
Returned bool `json:"returned"`
Error string `json:"error,omitempty"`
}
// Expect100Info describes an "Expect: 100-continue" handshake observed on
// this exchange.
type Expect100Info struct {
// Waited reports that the transport paused before sending the body.
Waited bool `json:"waited"`
// ContinueReceived reports that the server sent 100 Continue.
ContinueReceived bool `json:"continueReceived"`
// WaitMS is the pause between waiting and the 100 arriving; omitted when
// either side of the interval was not observed.
WaitMS float64 `json:"waitMs,omitempty"`
}
// InformationalResponse is one 1xx interim response (100 Continue, 103 Early
// Hints, ...) observed before the final response. Headers are redacted with
// the same rules as final response headers.
type InformationalResponse struct {
Status int `json:"status"`
Headers []NameValuePair `json:"headers,omitempty"`
}
// TLSInfo is the _recorder.tls value built from tls.ConnectionState.
type TLSInfo struct {
Version string `json:"version"`
CipherSuite string `json:"cipherSuite"`
NegotiatedProtocol string `json:"negotiatedProtocol,omitempty"`
ServerName string `json:"serverName,omitempty"`
HandshakeComplete bool `json:"handshakeComplete"`
DidResume bool `json:"didResume"`
OCSPStapled bool `json:"ocspStapled"`
SCTCount int `json:"sctCount"`
VerifiedChains int `json:"verifiedChains"`
PeerCertificates []CertInfo `json:"peerCertificates,omitempty"`
}
// CertInfo describes one peer certificate.
type CertInfo struct {
Subject string `json:"subject"`
Issuer string `json:"issuer"`
SerialNumber string `json:"serialNumber"`
DNSNames []string `json:"dnsNames,omitempty"`
IPAddresses []string `json:"ipAddresses,omitempty"`
NotBefore string `json:"notBefore"`
NotAfter string `json:"notAfter"`
PublicKeyAlgorithm string `json:"publicKeyAlgorithm"`
SignatureAlgorithm string `json:"signatureAlgorithm"`
SHA256Fingerprint string `json:"sha256Fingerprint"`
RawDER string `json:"rawDER,omitempty"`
}
// BodyInfo describes _recorder.requestBody or responseBody and records what
// happened to a body stream.
type BodyInfo struct {
Present bool `json:"present"`
Complete bool `json:"complete"`
ClosedEarly bool `json:"closedEarly,omitempty"`
Truncated bool `json:"truncated,omitempty"`
// CapturedBytes counts bytes stored for the HAR document.
CapturedBytes int64 `json:"capturedBytes"`
// TotalBytes counts every byte that actually flowed through the stream,
// including bytes past the capture limit.
TotalBytes int64 `json:"totalBytes"`
// Hash covers TotalBytes worth of data and is only emitted when the
// stream completed (a partial-stream hash would be misleading).
Hash string `json:"hash,omitempty"`
HashAlgorithm string `json:"hashAlgorithm,omitempty"`
ReadError string `json:"readError,omitempty"`
CloseError string `json:"closeError,omitempty"`
// Store is an opaque external reference when the BodyStore keeps content
// out of memory. The referenced bytes are the captured
// representation, which may already be decoded and/or redacted.
Store string `json:"store,omitempty"`
}
// RedactionInfo summarizes actual redaction work without exposing rule names
// or original values. Counts describe values changed in the recorded copy.
type RedactionInfo struct {
Request *RedactionScopeInfo `json:"request,omitempty"`
Response *RedactionScopeInfo `json:"response,omitempty"`
Errors int64 `json:"errors,omitempty"`
RawTrace int64 `json:"rawTrace,omitempty"`
}
// RedactionScopeInfo summarizes one side of an exchange.
type RedactionScopeInfo struct {
URL int64 `json:"url,omitempty"`
Headers int64 `json:"headers,omitempty"`
QueryParameters int64 `json:"queryParameters,omitempty"`
Cookies int64 `json:"cookies,omitempty"`
Body *BodyRedactionInfo `json:"body,omitempty"`
Protection *ProtectionCounts `json:"protection,omitempty"`
}
// ProtectionCounts summarizes representation modes and fail-closed causes.
// Fallback keys are fixed recorder-defined codes and never contain errors,
// field names, key IDs, tokens, or original values.
type ProtectionCounts struct {
Redacted int64 `json:"redacted,omitempty"`
Encrypted int64 `json:"encrypted,omitempty"`
Tokenized int64 `json:"tokenized,omitempty"`
Fallbacks map[string]int64 `json:"fallbacks,omitempty"`
}
const (
BodyRedactionRedacted = "redacted"
BodyRedactionUnchanged = "unchanged"
BodyRedactionFailed = "failed"
)
// BodyRedactionInfo reports which body redactor ran and its outcome.
type BodyRedactionInfo struct {
Kind string `json:"kind"`
Outcome string `json:"outcome"`
Replacements *int64 `json:"replacements,omitempty"`
Protection *ProtectionCounts `json:"protection,omitempty"`
}
// NewHAR builds a HAR document from finished entries. Entries are ordered by
// request start time (stable, so equal timestamps keep recording order). Only
// finalized entries ever reach a Recorder, so in-flight exchanges are by
// definition absent from the export.
func NewHAR(entries []*Entry) *HAR {
sorted := make([]*Entry, len(entries))
copy(sorted, entries)
sort.SliceStable(sorted, func(i, j int) bool {
return sorted[i].StartTime().Before(sorted[j].StartTime())
})
return &HAR{Log: &Log{
Version: harVersion,
Creator: &Creator{Name: creatorName, Version: creatorVersion},
Entries: sorted,
}}
}
// Write serializes the document as indented JSON. Field order follows struct
// declaration order and header lists are sorted, so output is deterministic
// for identical entries.
func (h *HAR) Write(w io.Writer) error {
data, err := json.MarshalIndent(h, "", " ")
if err != nil {
return fmt.Errorf("recorder: marshal HAR: %w", err)
}
data = append(data, '\n')
if _, err := w.Write(data); err != nil {
return fmt.Errorf("recorder: write HAR: %w", err)
}
return nil
}
// baseMimeType strips parameters ("; charset=...") from a media type.
func baseMimeType(mimeType string) string {
if parsed, _, err := mime.ParseMediaType(mimeType); err == nil {
return parsed
}
mt, _, _ := strings.Cut(mimeType, ";")
return strings.ToLower(strings.TrimSpace(mt))
}
// isTextualMime reports whether content of the given type is stored as plain
// text in HAR (subject to it also being valid UTF-8).
func isTextualMime(mimeType string) bool {
mt := baseMimeType(mimeType)
if strings.HasPrefix(mt, "text/") {
return true
}
if strings.HasSuffix(mt, "+json") || strings.HasSuffix(mt, "+xml") {
return true
}
switch mt {
case "application/json", "application/xml", "application/javascript",
"application/ecmascript", "application/x-www-form-urlencoded",
"application/x-ndjson", "application/xhtml+xml", "image/svg+xml",
"multipart/form-data":
return true
}
return false
}
func isJSONMime(mimeType string) bool {
mt := baseMimeType(mimeType)
return mt == "application/json" || strings.HasSuffix(mt, "+json") || mt == "application/x-ndjson"
}
// isXMLMime covers generic XML plus SOAP: SOAP 1.1 uses text/xml and
// SOAP 1.2 uses application/soap+xml (matched by the +xml suffix).
func isXMLMime(mimeType string) bool {
mt := baseMimeType(mimeType)
return mt == "application/xml" || mt == "text/xml" || strings.HasSuffix(mt, "+xml")
}
func isFormMime(mimeType string) bool {
return baseMimeType(mimeType) == "application/x-www-form-urlencoded"
}
func isMultipartFormMime(mimeType string) bool {
return baseMimeType(mimeType) == "multipart/form-data"
}
// contentText renders body bytes for HAR: plain text for textual UTF-8
// content, Base64 with encoding="base64" for everything else.
func contentText(mimeType string, b []byte) (text, encoding string) {
if isTextualMime(mimeType) && utf8.Valid(b) {
return string(b), ""
}
return base64.StdEncoding.EncodeToString(b), "base64"
}
package hario
import (
"bufio"
"bytes"
"encoding/json"
"errors"
"fmt"
"io"
"net/url"
"strings"
"time"
"github.com/mgurevin/recorder"
)
const (
defaultMaxBytes = 256 << 20
defaultMaxEntries = 100_000
defaultMaxEntryBytes = 16 << 20
maxHARReadChunk = 32 << 10
)
var (
// ErrInvalidHAR reports malformed JSON or an invalid HAR/entry structure.
ErrInvalidHAR = errors.New("hario: invalid capture")
// ErrLimitExceeded reports that a configured input bound was exceeded.
ErrLimitExceeded = errors.New("hario: input limit exceeded")
errHAREntryLimit = errors.New("hario: HAR entry read limit exceeded")
)
// ReadConfig bounds untrusted capture input.
type ReadConfig struct {
// MaxBytes bounds the complete encoded HAR document or NDJSON stream.
MaxBytes int64
// MaxEntries bounds the number of decoded entries.
MaxEntries int
// MaxEntryBytes bounds one encoded HAR entry or physical NDJSON line.
MaxEntryBytes int64
}
// DefaultReadConfig returns bounded defaults suitable for local test fixtures.
func DefaultReadConfig() ReadConfig {
return ReadConfig{
MaxBytes: defaultMaxBytes,
MaxEntries: defaultMaxEntries,
MaxEntryBytes: defaultMaxEntryBytes,
}
}
// ReadHAR reads one bounded HAR 1.2 JSON document. Unknown JSON fields are
// ignored so third-party HAR extensions remain accepted.
func ReadHAR(reader io.Reader, config ReadConfig) (*recorder.HAR, error) {
stream, err := NewHARStream(reader, config)
if err != nil {
return nil, err
}
entries := make([]*recorder.Entry, 0)
for {
entry, nextErr := stream.Next()
if errors.Is(nextErr, io.EOF) {
break
}
if nextErr != nil {
return nil, nextErr
}
entries = append(entries, entry)
}
document := stream.document
document.Log.Entries = entries
return document, nil
}
// ReadNDJSON reads a bounded stream containing one recorder.Entry JSON object
// per physical line. Blank lines are ignored and the final line need not end
// with a newline.
func ReadNDJSON(reader io.Reader, config ReadConfig) ([]*recorder.Entry, error) {
stream, err := NewNDJSONStream(reader, config)
if err != nil {
return nil, err
}
entries := make([]*recorder.Entry, 0)
for {
entry, nextErr := stream.Next()
if errors.Is(nextErr, io.EOF) {
break
}
if nextErr != nil {
return nil, nextErr
}
entries = append(entries, entry)
}
return entries, nil
}
type streamFormat uint8
const (
streamFormatHAR streamFormat = iota
streamFormatNDJSON
)
// EntryStream incrementally decodes validated recorder entries. Next returns
// io.EOF after the complete input, including trailing HAR metadata, has been
// validated. EntryStream does not close or otherwise own its input reader,
// open external body assets, or retain entries returned by earlier calls.
type EntryStream struct {
format streamFormat
config ReadConfig
limited *io.LimitedReader
decoder *json.Decoder
harEntry *entryLimitReader
buffered *bufio.Reader
document *recorder.HAR
entryCount int
lineNumber int
harLogSeen bool
done bool
terminalErr error
}
// NewHARStream prepares a bounded pull stream over one HAR 1.2 document.
func NewHARStream(reader io.Reader, config ReadConfig) (*EntryStream, error) {
if reader == nil {
return nil, fmt.Errorf("%w: nil reader", ErrInvalidHAR)
}
if err := validateConfig(config); err != nil {
return nil, err
}
limited := &io.LimitedReader{R: reader, N: config.MaxBytes + 1}
entryReader := &entryLimitReader{
reader: &chunkReader{reader: limited, max: maxHARReadChunk},
}
stream := &EntryStream{
format: streamFormatHAR,
config: config,
limited: limited,
decoder: json.NewDecoder(entryReader),
harEntry: entryReader,
document: &recorder.HAR{Log: &recorder.Log{}},
}
if err := stream.openHAR(); err != nil {
return nil, stream.harError(err)
}
if consumed(config.MaxBytes, limited) {
return nil, fmt.Errorf("%w: HAR exceeds %d bytes", ErrLimitExceeded, config.MaxBytes)
}
return stream, nil
}
// NewNDJSONStream prepares a bounded pull stream over recorder.Entry objects,
// one per physical line.
func NewNDJSONStream(reader io.Reader, config ReadConfig) (*EntryStream, error) {
if reader == nil {
return nil, fmt.Errorf("%w: nil reader", ErrInvalidHAR)
}
if err := validateConfig(config); err != nil {
return nil, err
}
limited := &io.LimitedReader{R: reader, N: config.MaxBytes + 1}
return &EntryStream{
format: streamFormatNDJSON,
config: config,
limited: limited,
buffered: bufio.NewReader(limited),
}, nil
}
// Next returns the next validated entry or io.EOF after complete input
// validation. After any error, subsequent calls return the same terminal error.
func (s *EntryStream) Next() (*recorder.Entry, error) {
if s == nil {
return nil, fmt.Errorf("%w: nil entry stream", ErrInvalidHAR)
}
if s.done {
if s.terminalErr != nil {
return nil, s.terminalErr
}
return nil, io.EOF
}
var (
entry *recorder.Entry
err error
)
switch s.format {
case streamFormatHAR:
entry, err = s.nextHAR()
case streamFormatNDJSON:
entry, err = s.nextNDJSON()
}
if err != nil {
s.done = true
if !errors.Is(err, io.EOF) {
s.terminalErr = err
}
}
return entry, err
}
func (s *EntryStream) nextNDJSON() (*recorder.Entry, error) {
for {
line, readErr := readBoundedLine(s.buffered, s.config.MaxEntryBytes)
s.lineNumber++
if consumed(s.config.MaxBytes, s.limited) {
return nil, fmt.Errorf("%w: NDJSON exceeds %d bytes", ErrLimitExceeded, s.config.MaxBytes)
}
if readErr != nil && !errors.Is(readErr, io.EOF) {
return nil, fmt.Errorf("hario: read NDJSON line %d: %w", s.lineNumber, readErr)
}
trimmed := bytes.TrimSpace(line)
if len(trimmed) == 0 {
if errors.Is(readErr, io.EOF) {
return nil, io.EOF
}
continue
}
if s.entryCount == 0 {
trimmed = bytes.TrimPrefix(trimmed, []byte{0xef, 0xbb, 0xbf})
}
if s.entryCount >= s.config.MaxEntries {
return nil, fmt.Errorf("%w: NDJSON contains more than %d entries", ErrLimitExceeded, s.config.MaxEntries)
}
var entry recorder.Entry
if err := json.Unmarshal(trimmed, &entry); err != nil {
return nil, fmt.Errorf("%w: NDJSON line %d: %v", ErrInvalidHAR, s.lineNumber, err)
}
if err := validateEntry(
&entry,
fmt.Sprintf("entry %d at line %d", s.entryCount, s.lineNumber),
); err != nil {
return nil, err
}
s.entryCount++
return &entry, nil
}
}
func (s *EntryStream) openHAR() error {
if err := expectDelimiter(s.decoder, '{', "HAR document"); err != nil {
return err
}
for s.decoder.More() {
name, err := decodeFieldName(s.decoder, "HAR document")
if err != nil {
return err
}
if name != "log" {
if err := skipJSONValue(s.decoder); err != nil {
return err
}
continue
}
if s.harLogSeen {
return fmt.Errorf("%w: duplicate log field", ErrInvalidHAR)
}
s.harLogSeen = true
if err := expectDelimiter(s.decoder, '{', "log"); err != nil {
return err
}
return s.openHAREntries()
}
return fmt.Errorf("%w: log is required", ErrInvalidHAR)
}
func (s *EntryStream) openHAREntries() error {
for s.decoder.More() {
name, err := decodeFieldName(s.decoder, "log")
if err != nil {
return err
}
if name == "entries" {
return expectDelimiter(s.decoder, '[', "log.entries")
}
if err := s.decodeHARLogField(name); err != nil {
return err
}
}
return fmt.Errorf("%w: log.entries is required", ErrInvalidHAR)
}
func (s *EntryStream) nextHAR() (*recorder.Entry, error) {
if !s.decoder.More() {
if err := s.finishHAR(); err != nil {
return nil, s.harError(err)
}
return nil, io.EOF
}
if s.entryCount >= s.config.MaxEntries {
return nil, fmt.Errorf("%w: HAR contains more than %d entries", ErrLimitExceeded, s.config.MaxEntries)
}
s.harEntry.begin(s.config.MaxEntryBytes + 1)
defer s.harEntry.end()
var encoded json.RawMessage
if err := s.decoder.Decode(&encoded); err != nil {
if errors.Is(err, errHAREntryLimit) {
return nil, fmt.Errorf(
"%w: HAR entry %d exceeds %d bytes",
ErrLimitExceeded,
s.entryCount,
s.config.MaxEntryBytes,
)
}
return nil, s.harError(fmt.Errorf("%w: decode HAR entry %d: %v", ErrInvalidHAR, s.entryCount, err))
}
if consumed(s.config.MaxBytes, s.limited) {
return nil, fmt.Errorf("%w: HAR exceeds %d bytes", ErrLimitExceeded, s.config.MaxBytes)
}
if int64(len(encoded)) > s.config.MaxEntryBytes {
return nil, fmt.Errorf("%w: HAR entry %d exceeds %d bytes", ErrLimitExceeded, s.entryCount, s.config.MaxEntryBytes)
}
var entry recorder.Entry
if err := json.Unmarshal(encoded, &entry); err != nil {
return nil, fmt.Errorf("%w: decode HAR entry %d: %v", ErrInvalidHAR, s.entryCount, err)
}
if err := validateEntry(&entry, fmt.Sprintf("entry %d", s.entryCount)); err != nil {
return nil, err
}
s.entryCount++
return &entry, nil
}
func (s *EntryStream) finishHAR() error {
if err := expectDelimiter(s.decoder, ']', "log.entries"); err != nil {
return err
}
for s.decoder.More() {
name, err := decodeFieldName(s.decoder, "log")
if err != nil {
return err
}
if name == "entries" {
return fmt.Errorf("%w: duplicate log.entries field", ErrInvalidHAR)
}
if err := s.decodeHARLogField(name); err != nil {
return err
}
}
if err := expectDelimiter(s.decoder, '}', "log"); err != nil {
return err
}
for s.decoder.More() {
name, err := decodeFieldName(s.decoder, "HAR document")
if err != nil {
return err
}
if name == "log" {
return fmt.Errorf("%w: duplicate log field", ErrInvalidHAR)
}
if err := skipJSONValue(s.decoder); err != nil {
return err
}
}
if err := expectDelimiter(s.decoder, '}', "HAR document"); err != nil {
return err
}
var trailing json.RawMessage
if err := s.decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
if err == nil {
return fmt.Errorf("%w: trailing JSON document", ErrInvalidHAR)
}
return fmt.Errorf("%w: trailing content: %v", ErrInvalidHAR, err)
}
if consumed(s.config.MaxBytes, s.limited) {
return fmt.Errorf("%w: HAR exceeds %d bytes", ErrLimitExceeded, s.config.MaxBytes)
}
return validateHARMetadata(s.document)
}
func (s *EntryStream) decodeHARLogField(name string) error {
switch name {
case "version":
if err := s.decoder.Decode(&s.document.Log.Version); err != nil {
return fmt.Errorf("%w: decode log.version: %v", ErrInvalidHAR, err)
}
case "creator":
if err := s.decoder.Decode(&s.document.Log.Creator); err != nil {
return fmt.Errorf("%w: decode log.creator: %v", ErrInvalidHAR, err)
}
case "comment":
if err := s.decoder.Decode(&s.document.Log.Comment); err != nil {
return fmt.Errorf("%w: decode log.comment: %v", ErrInvalidHAR, err)
}
default:
return skipJSONValue(s.decoder)
}
return nil
}
func (s *EntryStream) harError(err error) error {
if consumed(s.config.MaxBytes, s.limited) {
return fmt.Errorf("%w: HAR exceeds %d bytes", ErrLimitExceeded, s.config.MaxBytes)
}
return err
}
func decodeFieldName(decoder *json.Decoder, path string) (string, error) {
token, err := decoder.Token()
if err != nil {
return "", fmt.Errorf("%w: decode %s field: %v", ErrInvalidHAR, path, err)
}
name, ok := token.(string)
if !ok {
return "", fmt.Errorf("%w: %s field name is invalid", ErrInvalidHAR, path)
}
return name, nil
}
func expectDelimiter(decoder *json.Decoder, want json.Delim, path string) error {
token, err := decoder.Token()
if err != nil {
return fmt.Errorf("%w: decode %s: %v", ErrInvalidHAR, path, err)
}
delimiter, ok := token.(json.Delim)
if !ok || delimiter != want {
return fmt.Errorf("%w: %s must use %q", ErrInvalidHAR, path, want)
}
return nil
}
func skipJSONValue(decoder *json.Decoder) error {
var value json.RawMessage
if err := decoder.Decode(&value); err != nil {
return fmt.Errorf("%w: decode extension: %v", ErrInvalidHAR, err)
}
return nil
}
func validateHARMetadata(document *recorder.HAR) error {
if document == nil || document.Log == nil {
return fmt.Errorf("%w: log is required", ErrInvalidHAR)
}
if document.Log.Version != "1.2" {
return fmt.Errorf("%w: log.version must be 1.2", ErrInvalidHAR)
}
if document.Log.Creator == nil || strings.TrimSpace(document.Log.Creator.Name) == "" {
return fmt.Errorf("%w: log.creator.name is required", ErrInvalidHAR)
}
return nil
}
// ValidateHAR validates the HAR container and every entry without checking
// whether captured representations are sufficient for fixture replay.
func ValidateHAR(document *recorder.HAR) error {
if document == nil {
return fmt.Errorf("%w: nil HAR document", ErrInvalidHAR)
}
if err := validateHARMetadata(document); err != nil {
return err
}
return ValidateEntries(document.Log.Entries)
}
// ValidateEntries validates format-independent HAR entry structures.
func ValidateEntries(entries []*recorder.Entry) error {
for index, entry := range entries {
if err := validateEntry(entry, fmt.Sprintf("entry %d", index)); err != nil {
return err
}
}
return nil
}
func validateConfig(config ReadConfig) error {
if config.MaxBytes <= 0 || config.MaxEntries <= 0 || config.MaxEntryBytes <= 0 {
return fmt.Errorf("%w: read limits must be positive", ErrInvalidHAR)
}
if config.MaxEntryBytes > config.MaxBytes {
return fmt.Errorf("%w: MaxEntryBytes exceeds MaxBytes", ErrInvalidHAR)
}
return nil
}
func consumed(max int64, limited *io.LimitedReader) bool {
return max+1-limited.N > max
}
func readBoundedLine(reader *bufio.Reader, max int64) ([]byte, error) {
var line bytes.Buffer
for {
fragment, err := reader.ReadSlice('\n')
if int64(line.Len()+len(fragment)) > max {
return nil, fmt.Errorf("%w: NDJSON entry exceeds %d bytes", ErrLimitExceeded, max)
}
_, _ = line.Write(fragment)
if errors.Is(err, bufio.ErrBufferFull) {
continue
}
return bytes.TrimSuffix(line.Bytes(), []byte{'\n'}), err
}
}
type chunkReader struct {
reader io.Reader
max int
}
func (r *chunkReader) Read(p []byte) (int, error) {
if len(p) > r.max {
p = p[:r.max]
}
return r.reader.Read(p)
}
type entryLimitReader struct {
reader io.Reader
remaining int64
active bool
}
func (r *entryLimitReader) begin(max int64) {
r.remaining = max
r.active = true
}
func (r *entryLimitReader) end() {
r.active = false
r.remaining = 0
}
func (r *entryLimitReader) Read(p []byte) (int, error) {
if !r.active {
return r.reader.Read(p)
}
if r.remaining <= 0 {
return 0, errHAREntryLimit
}
if int64(len(p)) > r.remaining {
p = p[:int(r.remaining)]
}
n, err := r.reader.Read(p)
r.remaining -= int64(n)
return n, err
}
func validateEntry(entry *recorder.Entry, path string) error {
if entry == nil {
return fmt.Errorf("%w: %s is nil", ErrInvalidHAR, path)
}
if _, err := time.Parse(time.RFC3339, entry.StartedDateTime); err != nil {
return fmt.Errorf("%w: %s.startedDateTime is invalid", ErrInvalidHAR, path)
}
if entry.Time < 0 {
return fmt.Errorf("%w: %s.time must not be negative", ErrInvalidHAR, path)
}
if entry.Request == nil {
return fmt.Errorf("%w: %s.request is required", ErrInvalidHAR, path)
}
if strings.TrimSpace(entry.Request.Method) == "" {
return fmt.Errorf("%w: %s.request.method is required", ErrInvalidHAR, path)
}
parsedURL, err := url.Parse(entry.Request.URL)
if err != nil || parsedURL.Scheme == "" || parsedURL.Host == "" {
return fmt.Errorf("%w: %s.request.url must be absolute", ErrInvalidHAR, path)
}
if entry.Response == nil {
return fmt.Errorf("%w: %s.response is required", ErrInvalidHAR, path)
}
if entry.Response.Status < 0 || entry.Response.Status > 999 {
return fmt.Errorf("%w: %s.response.status is invalid", ErrInvalidHAR, path)
}
if entry.Response.Content == nil {
return fmt.Errorf("%w: %s.response.content is required", ErrInvalidHAR, path)
}
if entry.Timings == nil {
return fmt.Errorf("%w: %s.timings is required", ErrInvalidHAR, path)
}
timings := []struct {
name string
value float64
}{
{"blocked", entry.Timings.Blocked},
{"dns", entry.Timings.DNS},
{"connect", entry.Timings.Connect},
{"send", entry.Timings.Send},
{"wait", entry.Timings.Wait},
{"receive", entry.Timings.Receive},
{"ssl", entry.Timings.SSL},
}
for _, timing := range timings {
if timing.value < 0 && timing.value != -1 {
return fmt.Errorf("%w: %s.timings.%s must be -1 or non-negative", ErrInvalidHAR, path, timing.name)
}
}
if entry.Recorder != nil {
if entry.Recorder.SchemaVersion != recorder.RecorderExtensionVersion {
return fmt.Errorf("%w: %s._recorder.schemaVersion is unsupported", ErrInvalidHAR, path)
}
if err := validateBodyInfo(entry.Recorder.RequestBody, path+"._recorder.requestBody"); err != nil {
return err
}
if err := validateBodyInfo(entry.Recorder.ResponseBody, path+"._recorder.responseBody"); err != nil {
return err
}
}
return nil
}
func validateBodyInfo(info *recorder.BodyInfo, path string) error {
if info == nil {
return nil
}
if info.CapturedBytes < 0 || info.TotalBytes < 0 || info.CapturedBytes > info.TotalBytes {
return fmt.Errorf("%w: %s byte counters are inconsistent", ErrInvalidHAR, path)
}
if info.Complete && info.ClosedEarly {
return fmt.Errorf("%w: %s cannot be complete and closed early", ErrInvalidHAR, path)
}
if info.Truncated && info.CapturedBytes >= info.TotalBytes {
return fmt.Errorf("%w: %s truncated counters are inconsistent", ErrInvalidHAR, path)
}
return nil
}
package hartest
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"math"
"net/http"
"net/url"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/mgurevin/recorder"
"github.com/mgurevin/recorder/hario"
)
const (
defaultMaxRequestBodyBytes = 4 << 20
defaultMaxResponseBodyBytes = 32 << 20
redactedValue = "[REDACTED]"
)
var protectedTokenPattern = regexp.MustCompile(`REC-(?:ENC|TOK)-v1\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+`)
// BodyMatchMode controls whether captured request bodies select fixtures.
type BodyMatchMode string
const (
// BodyMatchAuto compares bodies whenever either side has one.
BodyMatchAuto BodyMatchMode = "auto"
// BodyMatchExact always compares complete captured body representations.
BodyMatchExact BodyMatchMode = "exact"
// BodyMatchIgnore explicitly excludes request bodies from matching.
BodyMatchIgnore BodyMatchMode = "ignore"
)
// MatchConfig defines deterministic request matching.
type MatchConfig struct {
// Headers names the request headers that participate in matching.
Headers []string
// Body controls request-body matching.
Body BodyMatchMode
// MaxRequestBodyBytes bounds the live request body read before matching.
MaxRequestBodyBytes int64
// Normalize canonicalizes isolated live and captured request snapshots.
Normalize RequestNormalizer
}
// RequestSnapshot is an isolated request representation supplied to a
// RequestNormalizer. Mutating it never changes the live request or source
// fixture.
type RequestSnapshot struct {
Method string
URL *url.URL
Headers http.Header
// Trailers contains request trailers observed before matching.
Trailers http.Header
// Body contains the complete bounded request representation.
Body []byte
}
// RequestNormalizer canonicalizes one isolated request representation before
// deterministic matching. The same function receives the live request and
// fixture snapshots independently.
type RequestNormalizer func(*RequestSnapshot) error
// BodyOpener resolves opaque external BodyStore references. FileBodyStore
// implements this interface.
type BodyOpener interface {
Open(ref string) (io.ReadCloser, error)
}
// BodyOpenerFunc adapts a function to BodyOpener.
type BodyOpenerFunc func(string) (io.ReadCloser, error)
// Open implements BodyOpener.
func (f BodyOpenerFunc) Open(ref string) (io.ReadCloser, error) { return f(ref) }
// ProtectedValueResolver resolves one complete REC-ENC-v1 or REC-TOK-v1 token.
// resolved=false leaves the protected representation unchanged.
type ProtectedValueResolver interface {
ResolveProtectedValue(token string) (plaintext []byte, resolved bool, err error)
}
// ProtectedValueResolverFunc adapts a function to ProtectedValueResolver.
type ProtectedValueResolverFunc func(string) ([]byte, bool, error)
// ResolveProtectedValue implements ProtectedValueResolver.
func (f ProtectedValueResolverFunc) ResolveProtectedValue(token string) ([]byte, bool, error) {
return f(token)
}
// DecryptProtectedValues returns an encrypted-token resolver backed by a
// rotation-aware recorder key resolver. Tokenized values remain unresolved.
func DecryptProtectedValues(keys recorder.ProtectionKeyResolver) ProtectedValueResolver {
return ProtectedValueResolverFunc(func(token string) ([]byte, bool, error) {
if !strings.HasPrefix(token, "REC-ENC-v1.") {
return nil, false, nil
}
plaintext, err := recorder.DecryptProtectedValueWith(token, keys)
if err != nil {
return nil, false, err
}
return plaintext, true, nil
})
}
// Config defines fixture matching and optional captured-representation access.
type Config struct {
Match MatchConfig
// MaxResponseBodyBytes bounds embedded and externally opened responses.
MaxResponseBodyBytes int64
// Bodies opens opaque external body-store references.
Bodies BodyOpener
// ProtectedValues resolves encrypted or tokenized fixture values.
ProtectedValues ProtectedValueResolver
// Timing optionally replays bounded recorded latency. Its zero value
// disables timing playback.
Timing ReplayTimingConfig
}
// ReplayTimingConfig controls coarse, deterministic playback of recorded HAR
// timing evidence. Scale=0 disables playback. A positive Scale requires a
// positive MaxDelay, which bounds the total delay applied per exchange.
type ReplayTimingConfig struct {
// Scale multiplies recorded durations. For example, 0.1 replays at one
// tenth of the captured latency and 1 uses the captured duration.
Scale float64
// MaxDelay bounds the combined response-header and response-body delay for
// one exchange.
MaxDelay time.Duration
}
// DefaultConfig returns strict, bounded matching defaults.
func DefaultConfig() Config {
return Config{
Match: MatchConfig{
Headers: []string{"Content-Type"},
Body: BodyMatchAuto,
MaxRequestBodyBytes: defaultMaxRequestBodyBytes,
},
MaxResponseBodyBytes: defaultMaxResponseBodyBytes,
}
}
type fixtureEntry struct {
entry *recorder.Entry
}
// EntrySource supplies fixture candidates incrementally. NewStreamTransport
// validates every returned entry. hario EntryStream implements this interface
// for HAR and NDJSON captures and additionally validates the input envelope.
// EntrySource has no lifecycle method; its reader remains caller-owned.
type EntrySource interface {
Next() (*recorder.Entry, error)
}
// Transport implements http.RoundTripper over a finite ordered entry set.
type Transport struct {
mu sync.Mutex
config Config
entries []fixtureEntry
source EntrySource
exhausted bool
wait func(context.Context, time.Duration) error
}
// NewTransport validates entries and creates a network-free fixture transport.
func NewTransport(entries []*recorder.Entry, config Config) (*Transport, error) {
if err := validateConfig(config); err != nil {
return nil, err
}
if err := hario.ValidateEntries(entries); err != nil {
return nil, fmt.Errorf("hartest: validate entries: %w", err)
}
fixtures := make([]fixtureEntry, len(entries))
for index, entry := range entries {
fixtures[index] = fixtureEntry{entry: entry}
}
return &Transport{config: config, entries: fixtures, wait: waitContext}, nil
}
// NewStreamTransport creates a network-free fixture transport that pulls
// entries only as matching requires them. Consumed entries are released;
// unmatched entries remain available for later requests.
func NewStreamTransport(source EntrySource, config Config) (*Transport, error) {
if source == nil {
return nil, errors.New("hartest: entry source is required")
}
if err := validateConfig(config); err != nil {
return nil, err
}
return &Transport{config: config, source: source, wait: waitContext}, nil
}
// RoundTrip matches and consumes one fixture. It never sends a request to a
// network or another RoundTripper. Matching reads and closes request.Body but
// does not replace or mutate the request's Body or GetBody fields.
func (t *Transport) RoundTrip(request *http.Request) (*http.Response, error) {
if request == nil || request.URL == nil {
return nil, errors.New("hartest: request and URL are required")
}
requestBody, err := readRequestBody(request, t.config.Match.MaxRequestBodyBytes)
if err != nil {
return nil, err
}
t.mu.Lock()
locked := true
defer func() {
if locked {
t.mu.Unlock()
}
}()
var candidateErr error
nextIndex := 0
for {
for nextIndex < len(t.entries) {
fixture := &t.entries[nextIndex]
matched, err := t.matches(request, requestBody, fixture.entry)
if err != nil {
candidateErr = errors.Join(candidateErr, err)
nextIndex++
continue
}
if !matched {
nextIndex++
continue
}
timing := replayTimingPlan(fixture.entry, t.config.Timing)
response, err := t.response(request, fixture.entry, timing.body)
if err != nil {
var recorded *RecordedError
if !errors.As(err, &recorded) {
return nil, err
}
t.removeEntry(nextIndex)
t.mu.Unlock()
locked = false
if waitErr := t.wait(request.Context(), timing.beforeResponse); waitErr != nil {
return nil, waitErr
}
return nil, err
}
t.removeEntry(nextIndex)
t.mu.Unlock()
locked = false
if err := t.wait(request.Context(), timing.beforeResponse); err != nil {
_ = response.Body.Close()
return nil, err
}
return response, nil
}
loaded, err := t.loadNextEntry()
if err != nil {
return nil, err
}
if !loaded {
break
}
}
if candidateErr != nil {
return nil, fmt.Errorf("hartest: no safely matchable exchange for %s %s: %w", request.Method, safeURL(request.URL), candidateErr)
}
return nil, fmt.Errorf("hartest: no matching exchange for %s %s", request.Method, safeURL(request.URL))
}
// Verify drains a lazy source, validates its trailing input, and reports whether
// every fixture was consumed exactly once. Call it after all client requests
// have completed.
func (t *Transport) Verify() error {
if t == nil {
return errors.New("hartest: nil transport")
}
t.mu.Lock()
defer t.mu.Unlock()
unused := len(t.entries)
for {
loaded, err := t.loadNextEntry()
if err != nil {
return err
}
if !loaded {
break
}
unused++
t.entries = t.entries[:len(t.entries)-1]
}
if unused > 0 {
return fmt.Errorf("hartest: %d fixture exchanges were not consumed", unused)
}
return nil
}
func (t *Transport) loadNextEntry() (bool, error) {
if t.source == nil || t.exhausted {
return false, nil
}
entry, err := t.source.Next()
if errors.Is(err, io.EOF) {
t.exhausted = true
return false, nil
}
if err != nil {
t.exhausted = true
return false, fmt.Errorf("hartest: read fixture entry: %w", err)
}
if err := hario.ValidateEntries([]*recorder.Entry{entry}); err != nil {
t.exhausted = true
return false, fmt.Errorf("hartest: validate streamed entry: %w", err)
}
t.entries = append(t.entries, fixtureEntry{entry: entry})
return true, nil
}
func (t *Transport) removeEntry(index int) {
copy(t.entries[index:], t.entries[index+1:])
t.entries[len(t.entries)-1] = fixtureEntry{}
t.entries = t.entries[:len(t.entries)-1]
}
func validateConfig(config Config) error {
switch config.Match.Body {
case BodyMatchAuto, BodyMatchExact, BodyMatchIgnore:
default:
return fmt.Errorf("hartest: unsupported body match mode %q", config.Match.Body)
}
if config.Match.MaxRequestBodyBytes <= 0 {
return errors.New("hartest: MaxRequestBodyBytes must be positive")
}
if config.MaxResponseBodyBytes <= 0 {
return errors.New("hartest: MaxResponseBodyBytes must be positive")
}
switch {
case math.IsNaN(config.Timing.Scale) || math.IsInf(config.Timing.Scale, 0) || config.Timing.Scale < 0:
return errors.New("hartest: Timing.Scale must be finite and non-negative")
case config.Timing.Scale == 0 && config.Timing.MaxDelay != 0:
return errors.New("hartest: Timing.MaxDelay requires a positive Scale")
case config.Timing.Scale > 0 && config.Timing.MaxDelay <= 0:
return errors.New("hartest: positive Timing.Scale requires a positive MaxDelay")
}
for _, name := range config.Match.Headers {
if strings.TrimSpace(name) == "" {
return errors.New("hartest: matched header names must not be empty")
}
}
return nil
}
type replayTiming struct {
beforeResponse time.Duration
body time.Duration
}
func replayTimingPlan(entry *recorder.Entry, config ReplayTimingConfig) replayTiming {
if config.Scale == 0 || entry == nil || entry.Timings == nil {
return replayTiming{}
}
timings := entry.Timings
beforeMilliseconds := positiveMilliseconds(
timings.Blocked,
timings.DNS,
timings.Connect,
timings.Send,
timings.Wait,
)
before := scaledDelay(beforeMilliseconds, config.Scale, config.MaxDelay)
remaining := config.MaxDelay - before
body := scaledDelay(positiveMilliseconds(timings.Receive), config.Scale, remaining)
return replayTiming{beforeResponse: before, body: body}
}
func positiveMilliseconds(values ...float64) float64 {
var total float64
for _, value := range values {
if value > 0 && !math.IsNaN(value) && !math.IsInf(value, 0) {
total += value
}
}
return total
}
func scaledDelay(milliseconds, scale float64, limit time.Duration) time.Duration {
if milliseconds <= 0 || scale <= 0 || limit <= 0 {
return 0
}
scaledMilliseconds := milliseconds * scale
limitMilliseconds := float64(limit) / float64(time.Millisecond)
if scaledMilliseconds >= limitMilliseconds {
return limit
}
return time.Duration(scaledMilliseconds * float64(time.Millisecond))
}
func proportionalDelay(total time.Duration, completed, size int) time.Duration {
if total <= 0 || completed <= 0 || size <= 0 {
return 0
}
if completed >= size {
return total
}
return time.Duration(float64(total) * (float64(completed) / float64(size)))
}
func waitContext(ctx context.Context, delay time.Duration) error {
if delay <= 0 {
return nil
}
timer := time.NewTimer(delay)
defer timer.Stop()
select {
case <-timer.C:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func readRequestBody(request *http.Request, max int64) ([]byte, error) {
if request.Body == nil {
return nil, nil
}
limited := io.LimitReader(request.Body, max+1)
body, readErr := io.ReadAll(limited)
closeErr := request.Body.Close()
if int64(len(body)) > max {
return nil, fmt.Errorf("hartest: request body exceeds %d bytes", max)
}
if readErr != nil {
return nil, fmt.Errorf("hartest: read request body: %w", readErr)
}
if closeErr != nil {
return nil, fmt.Errorf("hartest: close request body: %w", closeErr)
}
return body, nil
}
func (t *Transport) matches(request *http.Request, requestBody []byte, entry *recorder.Entry) (bool, error) {
fixtureURL, err := url.Parse(entry.Request.URL)
if err != nil {
return false, errors.New("invalid fixture URL")
}
var fixtureBody []byte
if t.config.Match.Body == BodyMatchIgnore {
fixtureBody = nil
} else {
var present bool
fixtureBody, present, err = t.requestBody(entry)
if err != nil {
return false, err
}
if t.config.Match.Body == BodyMatchAuto && !present && len(requestBody) == 0 {
fixtureBody = nil
}
}
actual := &RequestSnapshot{
Method: request.Method,
URL: cloneURL(request.URL),
Headers: request.Header.Clone(),
Trailers: request.Trailer.Clone(),
Body: bytes.Clone(requestBody),
}
expected := &RequestSnapshot{
Method: entry.Request.Method,
URL: fixtureURL,
Headers: headerFromPairs(entry.Request.Headers),
Trailers: requestTrailers(entry),
Body: bytes.Clone(fixtureBody),
}
if normalizer := t.config.Match.Normalize; normalizer != nil {
if err := normalizer(actual); err != nil {
return false, errors.New("live request normalization failed")
}
if err := normalizer(expected); err != nil {
return false, errors.New("fixture request normalization failed")
}
if actual.URL == nil || expected.URL == nil {
return false, errors.New("request normalizer removed a URL")
}
}
if actual.Method != expected.Method ||
!strings.EqualFold(actual.URL.Scheme, expected.URL.Scheme) ||
!strings.EqualFold(actual.URL.Host, expected.URL.Host) ||
actual.URL.EscapedPath() != expected.URL.EscapedPath() {
return false, nil
}
matched, err := t.matchQuery(actual.URL.Query(), expected.URL.Query())
if err != nil || !matched {
return matched, err
}
matched, err = t.matchHeaders(actual.Headers, expected.Headers)
if err != nil || !matched {
return matched, err
}
matched, err = t.matchTrailers(actual.Trailers, expected.Trailers)
if err != nil || !matched {
return matched, err
}
if t.config.Match.Body == BodyMatchIgnore {
return true, nil
}
return bytes.Equal(actual.Body, expected.Body), nil
}
func (t *Transport) matchQuery(actual, expected url.Values) (bool, error) {
if len(actual) != len(expected) {
return false, nil
}
for name, expectedValues := range expected {
actualValues, ok := actual[name]
if !ok || len(actualValues) != len(expectedValues) {
return false, nil
}
resolved := make([]string, len(expectedValues))
for index, value := range expectedValues {
value, unresolved, err := t.resolveString(value)
if err != nil {
return false, errors.New("query protected-value resolution failed")
}
if unresolved || strings.Contains(value, redactedValue) {
return false, fmt.Errorf("query parameter %q cannot be matched because its fixture value is unresolved", name)
}
resolved[index] = value
}
sort.Strings(actualValues)
sort.Strings(resolved)
if !equalStrings(actualValues, resolved) {
return false, nil
}
}
return true, nil
}
func (t *Transport) matchHeaders(actual, expected http.Header) (bool, error) {
for _, name := range t.config.Match.Headers {
expectedValues := expected.Values(name)
actualValues := actual.Values(name)
if len(expectedValues) != len(actualValues) {
return false, nil
}
for index, value := range expectedValues {
resolved, unresolved, err := t.resolveString(value)
if err != nil {
return false, fmt.Errorf("matched header %q protected-value resolution failed", http.CanonicalHeaderKey(name))
}
if unresolved || strings.Contains(resolved, redactedValue) {
return false, fmt.Errorf("matched header %q has an unresolved fixture value", http.CanonicalHeaderKey(name))
}
if resolved != actualValues[index] {
return false, nil
}
}
}
return true, nil
}
func (t *Transport) matchTrailers(actual, expected http.Header) (bool, error) {
if len(actual) != len(expected) {
return false, nil
}
for name, expectedValues := range expected {
actualValues := actual.Values(name)
if len(actualValues) != len(expectedValues) {
return false, nil
}
for index, value := range expectedValues {
resolved, unresolved, err := t.resolveString(value)
if err != nil {
return false, fmt.Errorf("request trailer %q protected-value resolution failed", http.CanonicalHeaderKey(name))
}
if unresolved || strings.Contains(resolved, redactedValue) {
return false, fmt.Errorf("request trailer %q has an unresolved fixture value", http.CanonicalHeaderKey(name))
}
if resolved != actualValues[index] {
return false, nil
}
}
}
return true, nil
}
func cloneURL(source *url.URL) *url.URL {
if source == nil {
return nil
}
clone := *source
return &clone
}
func requestTrailers(entry *recorder.Entry) http.Header {
if entry.Recorder == nil {
return make(http.Header)
}
return headerFromPairs(entry.Recorder.RequestTrailers)
}
func (t *Transport) requestBody(entry *recorder.Entry) ([]byte, bool, error) {
info := bodyInfo(entry, true)
if info != nil {
if info.Truncated || info.ClosedEarly || !info.Complete {
return nil, info.Present, errors.New("request fixture body is incomplete")
}
if info.Present && info.CapturedBytes < info.TotalBytes {
return nil, true, errors.New("request fixture body was not fully captured")
}
}
var (
body []byte
present bool
err error
)
if info != nil && info.Store != "" {
body, err = t.openBody(info.Store)
present = true
} else if entry.Request.PostData != nil {
body = []byte(entry.Request.PostData.Text)
present = true
if entry.Recorder != nil && entry.Recorder.RequestBodyEncoding == "base64" {
body, err = base64.StdEncoding.Strict().DecodeString(entry.Request.PostData.Text)
}
} else if info != nil && info.Present {
return nil, true, errors.New("request fixture body was not captured")
}
if err != nil {
return nil, present, fmt.Errorf("request fixture body: %w", err)
}
body, unresolved, err := t.resolveBody(body, contentType(entry.Request.Headers))
if err != nil {
return nil, present, errors.New("request fixture body protected-value resolution failed")
}
if unresolved || bytes.Contains(body, []byte(redactedValue)) {
return nil, present, errors.New("request fixture body contains unresolved protected values")
}
return body, present, nil
}
func (t *Transport) response(
request *http.Request,
entry *recorder.Entry,
bodyDelay time.Duration,
) (*http.Response, error) {
if entry.Response.Status == 0 {
if entry.Recorder != nil && entry.Recorder.Error != nil {
return nil, &RecordedError{
Phase: entry.Recorder.Error.Phase,
Message: entry.Recorder.Error.Message,
TimedOut: entry.Recorder.Error.Timeout,
}
}
return nil, errors.New("hartest: fixture has no HTTP response")
}
body, err := t.responseBody(entry)
if err != nil {
return nil, err
}
headers, err := t.responseHeaders(entry.Response.Headers)
if err != nil {
clear(body)
return nil, err
}
if entry.Recorder != nil && entry.Recorder.ResponseBodyDecoded {
headers.Del("Content-Encoding")
headers.Del("Content-Length")
}
protoMajor, protoMinor := protocolVersion(entry.Response.HTTPVersion)
trailers := make(http.Header)
response := &http.Response{
StatusCode: entry.Response.Status,
Status: strconv.Itoa(entry.Response.Status) + " " + entry.Response.StatusText,
Proto: entry.Response.HTTPVersion,
ProtoMajor: protoMajor,
ProtoMinor: protoMinor,
Header: headers,
ContentLength: int64(len(body)),
Request: request,
Trailer: trailers,
}
var bodyErr error
if entry.Recorder != nil && entry.Recorder.ResponseBody != nil && entry.Recorder.ResponseBody.ReadError != "" {
bodyErr = errors.New("hartest: recorded response body read error")
}
var recordedTrailers []recorder.NameValuePair
if entry.Recorder != nil {
recordedTrailers = entry.Recorder.ResponseTrailers
}
response.Body = &fixtureBody{
reader: bytes.NewReader(body),
bytes: body,
readErr: bodyErr,
trailers: trailers,
recorded: recordedTrailers,
resolver: t,
context: request.Context(),
delay: bodyDelay,
wait: t.wait,
}
return response, nil
}
func (t *Transport) responseBody(entry *recorder.Entry) ([]byte, error) {
info := bodyInfo(entry, false)
if info != nil {
if info.Truncated {
return nil, errors.New("hartest: response fixture body is truncated")
}
if info.Present && info.CapturedBytes < info.TotalBytes && info.ReadError == "" {
return nil, errors.New("hartest: response fixture body was not fully captured")
}
if info.Present && info.Store == "" && entry.Response.Content.Text == "" {
return nil, errors.New("hartest: response fixture body was not captured")
}
}
if info != nil && info.Store != "" {
body, err := t.openBody(info.Store)
if err != nil {
return nil, fmt.Errorf("hartest: response fixture body: %w", err)
}
resolved, _, err := t.resolveBody(body, entry.Response.Content.MimeType)
if err != nil {
return nil, errors.New("hartest: response body protected-value resolution failed")
}
return resolved, nil
}
text := entry.Response.Content.Text
var (
body []byte
err error
)
if entry.Response.Content.Encoding == "base64" {
body, err = base64.StdEncoding.Strict().DecodeString(text)
} else {
body = []byte(text)
}
if err != nil {
return nil, fmt.Errorf("hartest: decode response body: %w", err)
}
if int64(len(body)) > t.config.MaxResponseBodyBytes {
return nil, fmt.Errorf("hartest: response fixture body exceeds %d bytes", t.config.MaxResponseBodyBytes)
}
body, _, err = t.resolveBody(body, entry.Response.Content.MimeType)
if err != nil {
return nil, errors.New("hartest: response body protected-value resolution failed")
}
return body, nil
}
func (t *Transport) openBody(ref string) ([]byte, error) {
if t.config.Bodies == nil {
return nil, errors.New("external body requires BodyOpener")
}
reader, err := t.config.Bodies.Open(ref)
if err != nil {
return nil, err
}
body, readErr := io.ReadAll(io.LimitReader(reader, t.config.MaxResponseBodyBytes+1))
closeErr := reader.Close()
if readErr != nil {
return nil, readErr
}
if closeErr != nil {
return nil, closeErr
}
if int64(len(body)) > t.config.MaxResponseBodyBytes {
return nil, fmt.Errorf("external body exceeds %d bytes", t.config.MaxResponseBodyBytes)
}
return body, nil
}
func (t *Transport) responseHeaders(pairs []recorder.NameValuePair) (http.Header, error) {
headers := make(http.Header)
for _, pair := range pairs {
value, _, err := t.resolveString(pair.Value)
if err != nil {
return nil, fmt.Errorf("hartest: response header %q protected-value resolution failed", pair.Name)
}
headers.Add(pair.Name, value)
}
return headers, nil
}
func (t *Transport) resolveString(value string) (string, bool, error) {
resolved, unresolved, err := t.resolveBytes([]byte(value), false)
return string(resolved), unresolved, err
}
func (t *Transport) resolveBody(body []byte, mimeType string) ([]byte, bool, error) {
jsonBody := strings.Contains(strings.ToLower(strings.Split(mimeType, ";")[0]), "json")
return t.resolveBytes(body, jsonBody)
}
func (t *Transport) resolveBytes(input []byte, jsonBody bool) ([]byte, bool, error) {
matches := protectedTokenPattern.FindAllIndex(input, -1)
if len(matches) == 0 {
return input, false, nil
}
if t.config.ProtectedValues == nil {
return input, true, nil
}
var output bytes.Buffer
unresolved := false
offset := 0
for _, match := range matches {
start, end := match[0], match[1]
token := string(input[start:end])
plaintext, ok, err := t.config.ProtectedValues.ResolveProtectedValue(token)
if err != nil {
clear(plaintext)
return nil, false, err
}
replaceStart, replaceEnd := start, end
if ok && jsonBody && start > 0 && end < len(input) && input[start-1] == '"' && input[end] == '"' && json.Valid(plaintext) {
replaceStart--
replaceEnd++
}
_, _ = output.Write(input[offset:replaceStart])
if ok {
_, _ = output.Write(plaintext)
} else {
unresolved = true
_, _ = output.Write(input[start:end])
}
clear(plaintext)
offset = replaceEnd
}
_, _ = output.Write(input[offset:])
return output.Bytes(), unresolved, nil
}
func headerFromPairs(pairs []recorder.NameValuePair) http.Header {
headers := make(http.Header)
for _, pair := range pairs {
headers.Add(pair.Name, pair.Value)
}
return headers
}
func contentType(pairs []recorder.NameValuePair) string {
return headerFromPairs(pairs).Get("Content-Type")
}
func bodyInfo(entry *recorder.Entry, request bool) *recorder.BodyInfo {
if entry.Recorder == nil {
return nil
}
if request {
return entry.Recorder.RequestBody
}
return entry.Recorder.ResponseBody
}
func safeURL(value *url.URL) string {
return value.Scheme + "://" + value.Host + value.EscapedPath()
}
func equalStrings(left, right []string) bool {
if len(left) != len(right) {
return false
}
for index := range left {
if left[index] != right[index] {
return false
}
}
return true
}
func protocolVersion(protocol string) (int, int) {
switch protocol {
case "HTTP/1.0":
return 1, 0
case "HTTP/1.1":
return 1, 1
case "HTTP/2", "HTTP/2.0":
return 2, 0
case "HTTP/3", "HTTP/3.0":
return 3, 0
default:
return 0, 0
}
}
// RecordedError represents a transport failure captured before an HTTP
// response existed. It preserves only evidence recorded by recorder and does
// not manufacture a concrete network or TLS error whose fields and unwrap
// chain are absent from the fixture.
type RecordedError struct {
// Phase identifies the HTTP lifecycle phase in which the recorded failure
// occurred.
Phase string
// Message is the sanitized failure message stored in the fixture.
Message string
// TimedOut reports whether the recorded failure satisfied net.Error's
// timeout contract.
TimedOut bool
}
// Error implements error.
func (e *RecordedError) Error() string {
if e.Message == "" {
return "hartest: recorded HTTP failure"
}
return "hartest: recorded HTTP failure: " + e.Message
}
// Timeout reports the recorded timeout classification.
func (e *RecordedError) Timeout() bool { return e.TimedOut }
type fixtureBody struct {
reader *bytes.Reader
bytes []byte
readErr error
trailers http.Header
recorded []recorder.NameValuePair
resolver *Transport
closed bool
context context.Context
delay time.Duration
waited time.Duration
wait func(context.Context, time.Duration) error
}
func (b *fixtureBody) Read(p []byte) (int, error) {
if b.closed {
return 0, http.ErrBodyReadAfterClose
}
if len(p) > 0 && b.reader.Len() > 0 {
readSize := min(len(p), b.reader.Len())
completed := len(b.bytes) - b.reader.Len() + readSize
target := proportionalDelay(b.delay, completed, len(b.bytes))
if delay := target - b.waited; delay > 0 {
if waitErr := b.wait(b.context, delay); waitErr != nil {
return 0, waitErr
}
b.waited = target
}
}
n, err := b.reader.Read(p)
if errors.Is(err, io.EOF) {
b.publishTrailers()
if b.readErr != nil {
err = b.readErr
b.readErr = nil
}
}
return n, err
}
func (b *fixtureBody) Close() error {
if b.closed {
return nil
}
b.closed = true
b.publishTrailers()
clear(b.bytes)
return nil
}
func (b *fixtureBody) publishTrailers() {
if b.recorded == nil {
return
}
for _, pair := range b.recorded {
value, _, err := b.resolver.resolveString(pair.Value)
if err == nil {
b.trailers.Add(pair.Name, value)
}
}
b.recorded = nil
}
package recorder
import "log"
// reportInternalError applies the shared, fail-contained Transport and
// AsyncRecorder reporting policy.
func reportInternalError(mode InternalErrorMode, onError func(error), logf func(string, ...any), err error) {
if err == nil {
return
}
if mode == InternalErrorLog {
if logf != nil {
callSafely(func() { logf("recorder: %v", err) })
} else {
log.Printf("recorder: %v", err)
}
}
if onError != nil {
callSafely(func() { onError(err) })
}
}
package recorder
import (
"bytes"
"encoding/json"
"errors"
"io"
"strings"
"unicode/utf8"
)
const (
maxJSONKeyBytes = 64 << 10
maxJSONDepth = 1024
)
var errRedactionLimit = errors.New("recorder: streaming redaction limit exceeded")
const maxBodySniffBytes = 4096
type sniffingBodyRedactor struct {
dst io.Writer
red *redactor
protector *bodyValueProtector
buf []byte
selected io.WriteCloser
plain bool
err error
}
func (s *sniffingBodyRedactor) Write(p []byte) (int, error) {
if s.err != nil {
return 0, s.err
}
if s.selected != nil {
return s.selected.Write(p)
}
if s.plain {
return s.dst.Write(p)
}
for i, b := range p {
if len(s.buf) >= maxBodySniffBytes {
s.err = errRedactionLimit
return i, s.err
}
s.buf = append(s.buf, b)
if jsonSpace(b) || (len(s.buf) <= 3 && bytes.Equal(s.buf, []byte{0xef, 0xbb, 0xbf}[:len(s.buf)])) {
continue
}
switch b {
case '{', '[':
if len(s.red.jsonFields) > 0 {
s.selected = newJSONStreamRedactor(s.dst, s.red.jsonFields, s.protector)
}
case '<':
if len(s.red.xmlElements) > 0 {
s.selected = newXMLStreamRedactor(s.dst, s.red.xmlElements, s.protector)
}
}
if s.selected == nil {
s.plain = true
_, s.err = s.dst.Write(s.buf)
} else {
_, s.err = s.selected.Write(s.buf)
}
s.buf = nil
if s.err != nil {
return i + 1, s.err
}
if i+1 < len(p) {
n, err := s.Write(p[i+1:])
return i + 1 + n, err
}
return len(p), nil
}
return len(p), nil
}
func (s *sniffingBodyRedactor) Close() error {
if s.err != nil {
return s.err
}
if s.selected != nil {
return s.selected.Close()
}
if len(s.buf) > 0 {
_, s.err = s.dst.Write(s.buf)
s.buf = nil
}
return s.err
}
type jsonContainer byte
const (
jsonRoot jsonContainer = iota
jsonObject
jsonArray
)
type jsonState byte
const (
jsonWantValue jsonState = iota
jsonWantKey
jsonWantColon
jsonWantComma
jsonDone
)
type jsonFrame struct {
kind jsonContainer
state jsonState
}
// jsonStreamRedactor preserves every non-redacted input byte. It deliberately
// redacts a recognized field even if later bytes make the document malformed:
// a streaming sink cannot roll back bytes already committed to storage.
type jsonStreamRedactor struct {
dst io.Writer
bytes byteSink
fields map[string]struct{}
stack []jsonFrame
key []byte
keyFold []byte
keyMatch bool
inString bool
keyToken bool
escaped bool
scalar bool
bom []byte
bomDone bool
suppress bool
suppressMode byte // s=string, c=composite, v=scalar
suppressDepth int
suppressQuote bool
suppressEsc bool
protected protectedValueBuffer
protectedJSON []byte
err error
}
func newJSONStreamRedactor(dst io.Writer, fields map[string]struct{}, protectors ...*bodyValueProtector) *jsonStreamRedactor {
protector := newBodyValueProtector(nil)
if len(protectors) > 0 && protectors[0] != nil {
protector = protectors[0]
}
r := &jsonStreamRedactor{
dst: dst,
bytes: newByteSink(dst),
fields: fields,
stack: []jsonFrame{{kind: jsonRoot, state: jsonWantValue}},
}
r.protected.reset(protector)
return r
}
func (r *jsonStreamRedactor) Write(p []byte) (int, error) {
if r.err != nil {
return 0, r.err
}
for i, b := range p {
if err := r.consumeWithBOM(b); err != nil {
r.err = err
return i, err
}
}
return len(p), nil
}
func (r *jsonStreamRedactor) Close() error {
if r.err == nil && len(r.bom) > 0 {
for _, b := range r.bom {
if err := r.consume(b); err != nil {
r.err = err
break
}
}
r.bom = nil
}
if r.err == nil && r.suppress {
r.err = r.emitProtected()
r.suppress = false
}
return r.err
}
func (r *jsonStreamRedactor) consumeWithBOM(b byte) error {
if r.bomDone {
return r.consume(b)
}
want := [...]byte{0xef, 0xbb, 0xbf}
if b == want[len(r.bom)] {
r.bom = append(r.bom, b)
if len(r.bom) == len(want) {
r.bomDone = true
_, err := r.dst.Write(r.bom)
r.bom = nil
return err
}
return nil
}
r.bomDone = true
for _, prefixByte := range r.bom {
if err := r.consume(prefixByte); err != nil {
return err
}
}
r.bom = nil
return r.consume(b)
}
func (r *jsonStreamRedactor) emitByte(b byte) error {
return r.bytes.WriteByte(b)
}
func jsonSpace(b byte) bool {
return b == ' ' || b == '\t' || b == '\r' || b == '\n'
}
func jsonDelimiter(b byte) bool {
return jsonSpace(b) || b == ',' || b == ']' || b == '}'
}
func (r *jsonStreamRedactor) consume(b byte) error {
if r.suppress {
done, reprocess, err := r.consumeSuppressed(b)
if err != nil {
return err
}
if !done || !reprocess {
return nil
}
}
if r.inString {
if err := r.emitByte(b); err != nil {
return err
}
if r.keyToken {
if len(r.key) >= maxJSONKeyBytes {
return errRedactionLimit
}
r.key = append(r.key, b)
}
if r.escaped {
r.escaped = false
return nil
}
if b == '\\' {
r.escaped = true
return nil
}
if b != '"' {
return nil
}
r.inString = false
if r.keyToken {
r.keyMatch = r.matchesKey()
r.key = r.key[:0]
r.keyToken = false
r.top().state = jsonWantColon
}
return nil
}
if r.scalar {
if !jsonDelimiter(b) {
return r.emitByte(b)
}
r.scalar = false
// The delimiter belongs to the containing structure.
}
f := r.top()
switch f.state {
case jsonWantKey:
if err := r.emitByte(b); err != nil {
return err
}
if jsonSpace(b) || b == '}' {
if b == '}' {
r.pop()
}
return nil
}
if b == '"' {
r.inString, r.keyToken = true, true
r.key = append(r.key[:0], b)
}
return nil
case jsonWantColon:
if err := r.emitByte(b); err != nil {
return err
}
if b == ':' {
f.state = jsonWantValue
}
return nil
case jsonWantComma:
if err := r.emitByte(b); err != nil {
return err
}
if jsonSpace(b) {
return nil
}
if b == ',' {
if f.kind == jsonObject {
f.state = jsonWantKey
} else {
f.state = jsonWantValue
}
} else if (b == '}' && f.kind == jsonObject) || (b == ']' && f.kind == jsonArray) {
r.pop()
}
return nil
case jsonDone:
if jsonSpace(b) {
return r.emitByte(b)
}
// NDJSON and JSON text sequences contain multiple top-level values.
// Reset only after the previous container/value has completed, then
// process this byte as the beginning of the next document.
f.state = jsonWantValue
return r.consume(b)
}
// jsonWantValue
if jsonSpace(b) {
return r.emitByte(b)
}
if f.kind == jsonArray && b == ']' {
if err := r.emitByte(b); err != nil {
return err
}
r.pop()
return nil
}
if f.kind == jsonObject && r.keyMatch {
r.keyMatch = false
r.markValue()
return r.startSuppression(b)
}
r.markValue()
if err := r.emitByte(b); err != nil {
return err
}
switch b {
case '{':
return r.push(jsonObject, jsonWantKey)
case '[':
return r.push(jsonArray, jsonWantValue)
case '"':
r.inString = true
default:
r.scalar = true
}
return nil
}
func (r *jsonStreamRedactor) matchesKey() bool {
if len(r.key) < 2 {
return false
}
raw := r.key[1 : len(r.key)-1]
hasUpper := false
for _, b := range raw {
if b == '\\' || b < 0x20 || b >= utf8.RuneSelf {
return r.matchesDecodedKey()
}
if b >= 'A' && b <= 'Z' {
hasUpper = true
}
}
if !hasUpper {
_, ok := r.fields[string(raw)]
return ok
}
r.keyFold = append(r.keyFold[:0], raw...)
for index, b := range r.keyFold {
if b >= 'A' && b <= 'Z' {
r.keyFold[index] = b + ('a' - 'A')
}
}
_, ok := r.fields[string(r.keyFold)]
return ok
}
func (r *jsonStreamRedactor) matchesDecodedKey() bool {
var key string
if err := json.Unmarshal(r.key, &key); err != nil {
return false
}
_, ok := r.fields[strings.ToLower(key)]
return ok
}
func (r *jsonStreamRedactor) top() *jsonFrame { return &r.stack[len(r.stack)-1] }
func (r *jsonStreamRedactor) markValue() {
f := r.top()
if f.kind == jsonRoot {
f.state = jsonDone
} else {
f.state = jsonWantComma
}
}
func (r *jsonStreamRedactor) push(kind jsonContainer, state jsonState) error {
if len(r.stack) >= maxJSONDepth {
return errRedactionLimit
}
r.stack = append(r.stack, jsonFrame{kind: kind, state: state})
return nil
}
func (r *jsonStreamRedactor) pop() {
if len(r.stack) > 1 {
r.stack = r.stack[:len(r.stack)-1]
}
}
func (r *jsonStreamRedactor) startSuppression(b byte) error {
r.suppress = true
r.protected.reset(r.protected.session)
if r.protected.redactImmediately() {
if err := r.emitJSONProtection([]byte(redactedValue)); err != nil {
return err
}
} else {
r.protected.appendByte(b)
}
switch b {
case '"':
r.suppressMode = 's'
case '{', '[':
r.suppressMode = 'c'
r.suppressDepth = 1
default:
r.suppressMode = 'v'
}
return nil
}
func (r *jsonStreamRedactor) consumeSuppressed(b byte) (done, reprocess bool, err error) {
switch r.suppressMode {
case 's':
r.protected.appendByte(b)
if r.suppressEsc {
r.suppressEsc = false
return false, false, nil
}
switch b {
case '\\':
r.suppressEsc = true
case '"':
r.suppress = false
return true, false, r.emitProtected()
}
case 'c':
r.protected.appendByte(b)
if r.suppressQuote {
if r.suppressEsc {
r.suppressEsc = false
} else if b == '\\' {
r.suppressEsc = true
} else if b == '"' {
r.suppressQuote = false
}
return false, false, nil
}
switch b {
case '"':
r.suppressQuote = true
case '{', '[':
r.suppressDepth++
if r.suppressDepth > maxJSONDepth {
return false, false, errRedactionLimit
}
case '}', ']':
r.suppressDepth--
if r.suppressDepth == 0 {
r.suppress = false
return true, false, r.emitProtected()
}
}
case 'v':
if jsonDelimiter(b) {
r.suppress = false
return true, true, r.emitProtected()
}
r.protected.appendByte(b)
}
return false, false, nil
}
func (r *jsonStreamRedactor) emitProtected() error {
value := r.protected.protectedBytes()
if len(value) == 0 {
return nil
}
return r.emitJSONProtection(value)
}
func (r *jsonStreamRedactor) emitJSONProtection(value []byte) error {
// Protected values use a fixed ASCII alphabet without JSON quote or escape
// bytes. Reusing one output buffer avoids both a per-value marshal and the
// cost of three separate writes.
r.protectedJSON = append(r.protectedJSON[:0], '"')
r.protectedJSON = append(r.protectedJSON, value...)
r.protectedJSON = append(r.protectedJSON, '"')
_, err := r.dst.Write(r.protectedJSON)
return err
}
package recorder
import (
"errors"
"io"
"sync"
)
// DefaultMemoryRecorderCapacity is the maximum number of finalized entries
// retained by NewMemoryRecorder.
const DefaultMemoryRecorderCapacity = 1024
// MemoryRecorderStats is an atomic point-in-time snapshot of retention state.
// Evicted is a lifetime counter and is not cleared by Reset.
type MemoryRecorderStats struct {
Capacity int
Retained int
Evicted uint64
}
// MemoryRecorder collects finalized entries in memory. It is safe for
// concurrent use. Entries handed to it are immutable snapshots, so the copies
// returned by Entries can be shared freely.
type MemoryRecorder struct {
mu sync.Mutex
entries []*Entry
head int
size int
evicted uint64
}
// NewMemoryRecorder returns an empty in-memory recorder retaining the newest
// DefaultMemoryRecorderCapacity finalized entries. Once full, recording a new
// entry evicts the oldest retained entry.
func NewMemoryRecorder() *MemoryRecorder {
return newMemoryRecorder(DefaultMemoryRecorderCapacity)
}
// NewMemoryRecorderWithCapacity returns an empty in-memory recorder retaining
// the newest capacity finalized entries. Capacity must be positive.
func NewMemoryRecorderWithCapacity(capacity int) (*MemoryRecorder, error) {
if capacity <= 0 {
return nil, errors.New("recorder: memory recorder capacity must be positive")
}
return newMemoryRecorder(capacity), nil
}
func newMemoryRecorder(capacity int) *MemoryRecorder {
return &MemoryRecorder{entries: make([]*Entry, capacity)}
}
// Record implements Recorder.
func (r *MemoryRecorder) Record(e *Entry) error {
r.mu.Lock()
defer r.mu.Unlock()
r.recordLocked(e)
return nil
}
// RecordBatch implements batchRecorder with one lock acquisition.
func (r *MemoryRecorder) RecordBatch(entries []*Entry) error {
r.mu.Lock()
defer r.mu.Unlock()
for _, entry := range entries {
r.recordLocked(entry)
}
return nil
}
func (r *MemoryRecorder) recordLocked(e *Entry) {
if r.size == len(r.entries) {
r.entries[r.head] = nil
r.head = (r.head + 1) % len(r.entries)
r.size--
r.evicted++
}
index := (r.head + r.size) % len(r.entries)
r.entries[index] = e
r.size++
}
// Entries returns a copy of the recorded entries in recording order.
func (r *MemoryRecorder) Entries() []*Entry {
r.mu.Lock()
defer r.mu.Unlock()
return r.entriesLocked()
}
// Snapshot atomically returns the retained entries in recording order and the
// corresponding retention statistics.
func (r *MemoryRecorder) Snapshot() ([]*Entry, MemoryRecorderStats) {
r.mu.Lock()
defer r.mu.Unlock()
return r.entriesLocked(), r.statsLocked()
}
// Stats returns an atomic point-in-time snapshot of retention state.
func (r *MemoryRecorder) Stats() MemoryRecorderStats {
r.mu.Lock()
defer r.mu.Unlock()
return r.statsLocked()
}
// Len returns the number of recorded entries.
func (r *MemoryRecorder) Len() int {
r.mu.Lock()
defer r.mu.Unlock()
return r.size
}
// Reset discards all recorded entries.
func (r *MemoryRecorder) Reset() {
r.mu.Lock()
defer r.mu.Unlock()
for i := range r.entries {
r.entries[i] = nil
}
r.head = 0
r.size = 0
}
// EntriesByTrace returns a copy of the entries belonging to the given trace
// ID (see WithTraceID), in recording order. The recorder keeps the entries.
func (r *MemoryRecorder) EntriesByTrace(traceID string) []*Entry {
r.mu.Lock()
defer r.mu.Unlock()
return filterTrace(r.entriesLocked(), traceID)
}
// RemoveTrace deletes every entry belonging to the given trace ID and
// returns how many were removed.
func (r *MemoryRecorder) RemoveTrace(traceID string) int {
removed, _ := r.remove(traceID, false)
return removed
}
// TakeTrace atomically removes and returns the entries belonging to the
// given trace ID, in recording order. Atomicity matters on a shared client:
// an entry finalized concurrently can never be lost between a query and a
// separate delete.
func (r *MemoryRecorder) TakeTrace(traceID string) []*Entry {
_, taken := r.remove(traceID, true)
return taken
}
func (r *MemoryRecorder) remove(traceID string, collect bool) (int, []*Entry) {
r.mu.Lock()
defer r.mu.Unlock()
kept, taken, removed := splitTrace(r.entriesLocked(), traceID, collect)
for i := range r.entries {
r.entries[i] = nil
}
copy(r.entries, kept)
r.head = 0
r.size = len(kept)
return removed, taken
}
func (r *MemoryRecorder) entriesLocked() []*Entry {
entries := make([]*Entry, r.size)
for i := range r.size {
entries[i] = r.entries[(r.head+i)%len(r.entries)]
}
return entries
}
func (r *MemoryRecorder) statsLocked() MemoryRecorderStats {
return MemoryRecorderStats{
Capacity: len(r.entries),
Retained: r.size,
Evicted: r.evicted,
}
}
// filterTrace returns the entries matching traceID, preserving order.
func filterTrace(entries []*Entry, traceID string) []*Entry {
var matched []*Entry
for _, e := range entries {
if e != nil && e.Recorder != nil && e.Recorder.TraceID == traceID {
matched = append(matched, e)
}
}
return matched
}
// splitTrace compacts non-matching entries to the front of the slice in
// place, optionally collecting the matching ones, and nils out the vacated
// tail so removed entries are released to the GC.
func splitTrace(entries []*Entry, traceID string, collect bool) (kept, taken []*Entry, removed int) {
kept = entries[:0]
for _, e := range entries {
if e != nil && e.Recorder != nil && e.Recorder.TraceID == traceID {
if collect {
taken = append(taken, e)
}
continue
}
kept = append(kept, e)
}
removed = len(entries) - len(kept)
for i := len(kept); i < len(entries); i++ {
entries[i] = nil
}
return kept, taken, removed
}
// HAR builds a HAR 1.2 document from the entries recorded so far, ordered by
// request start time. In-flight exchanges are absent by definition: entries
// only reach the recorder once finalized.
func (r *MemoryRecorder) HAR() *HAR { return NewHAR(r.Entries()) }
// HARForTrace builds a HAR 1.2 document containing only the entries of the
// given trace ID (the recorder keeps them; combine with TakeTrace +
// NewHAR to export-and-drop in one step).
func (r *MemoryRecorder) HARForTrace(traceID string) *HAR {
return NewHAR(r.EntriesByTrace(traceID))
}
// WriteHAR serializes the current HAR document to w as indented JSON.
func (r *MemoryRecorder) WriteHAR(w io.Writer) error { return r.HAR().Write(w) }
package recorder
import (
"errors"
"fmt"
)
// NewMultiRecorder creates a synchronous fan-out Recorder. Each call is sent
// to every recorder in argument order, even when an earlier recorder fails.
// Multiple failures are returned as one errors.Join-compatible error.
//
// Multi-recorder fan-out does not add synchronization, clone entries, recover
// panics, or own downstream lifecycle. Every supplied recorder must satisfy
// the normal concurrent-use and immutable-entry Recorder contract. Wrap the
// result in AsyncRecorder when fan-out must not run on response-finalization
// goroutines. The first recorder is mandatory; a nil recorder is a wiring
// error and causes NewMultiRecorder to panic.
func NewMultiRecorder(first Recorder, others ...Recorder) Recorder {
if first == nil {
panic("recorder: multi recorder 0 is nil")
}
for i, recorder := range others {
if recorder == nil {
panic(fmt.Sprintf("recorder: multi recorder %d is nil", i+1))
}
}
if len(others) == 0 {
return first
}
recorders := make([]Recorder, 1, len(others)+1)
recorders[0] = first
recorders = append(recorders, others...)
return &multiRecorder{recorders: recorders}
}
type multiRecorder struct {
recorders []Recorder
}
// Record delivers entry to every downstream recorder without short-circuiting.
func (r *multiRecorder) Record(entry *Entry) error {
var failures []error
for i, recorder := range r.recorders {
if err := recorder.Record(entry); err != nil {
failures = append(failures, multiRecorderError(i, recorder, err))
}
}
return errors.Join(failures...)
}
// RecordBatch preserves AsyncRecorder batching for capable downstream sinks
// and falls back to ordered Record calls for ordinary recorders.
func (r *multiRecorder) RecordBatch(entries []*Entry) error {
if len(entries) == 0 {
return nil
}
var failures []error
for i, recorder := range r.recorders {
if batch, ok := recorder.(batchRecorder); ok {
if err := batch.RecordBatch(entries); err != nil {
failures = append(failures, multiRecorderError(i, recorder, err))
}
continue
}
for _, entry := range entries {
if err := recorder.Record(entry); err != nil {
failures = append(failures, multiRecorderError(i, recorder, err))
}
}
}
return errors.Join(failures...)
}
func multiRecorderError(index int, recorder Recorder, err error) error {
return fmt.Errorf("recorder: multi recorder %d (%T): %w", index, recorder, err)
}
package recorder
import (
"bytes"
"errors"
"fmt"
"io"
"mime"
"strings"
"unicode/utf8"
)
const (
maxMultipartHeaderBytes = 64 << 10
maxMultipartPreambleBytes = 4 << 10
maxMultipartBoundaryBytes = 70
multipartFeedChunkBytes = 32 << 10
multipartCompactBytes = 4 << 10
)
var errMalformedMultipart = errors.New("recorder: malformed multipart body")
type multipartState byte
const (
multipartPreamble multipartState = iota
multipartHeaders
multipartBody
multipartEpilogue
)
// multipartStreamRedactor preserves preamble, delimiters, unmatched parts,
// and unmatched headers byte-for-byte. It buffers one part header block and
// a boundary-sized lookbehind; matched part bodies are discarded as they
// stream.
type multipartStreamRedactor struct {
dst io.Writer
fields map[string]struct{}
boundary []byte
marker []byte
lineMarker []byte
headerFold []byte
state multipartState
pending []byte
suppress bool
err error
protected protectedValueBuffer
}
func newMultipartStreamRedactor(dst io.Writer, mimeType string, fields map[string]struct{}, protectors ...*bodyValueProtector) *multipartStreamRedactor {
r := &multipartStreamRedactor{dst: dst, fields: fields, state: multipartPreamble}
protector := newBodyValueProtector(nil)
if len(protectors) > 0 && protectors[0] != nil {
protector = protectors[0]
}
r.protected.reset(protector)
mt, params, err := mime.ParseMediaType(mimeType)
boundary := params["boundary"]
if err != nil || !strings.EqualFold(mt, "multipart/form-data") || !validMultipartBoundary(boundary) {
r.err = fmt.Errorf("%w: invalid or missing boundary", errMalformedMultipart)
return r
}
r.boundary = []byte(boundary)
r.marker = []byte("--" + boundary)
r.lineMarker = append([]byte("\r\n"), r.marker...)
return r
}
func validMultipartBoundary(boundary string) bool {
if len(boundary) == 0 || len(boundary) > maxMultipartBoundaryBytes || boundary[len(boundary)-1] == ' ' {
return false
}
for i := 0; i < len(boundary); i++ {
b := boundary[i]
if (b >= 'a' && b <= 'z') || (b >= 'A' && b <= 'Z') || (b >= '0' && b <= '9') {
continue
}
if !strings.ContainsRune("'()+_,-./:=? ", rune(b)) {
return false
}
}
return true
}
func (r *multipartStreamRedactor) Write(p []byte) (int, error) {
if r.err != nil {
return 0, r.err
}
written := 0
for len(p) > 0 {
take := min(len(p), multipartFeedChunkBytes)
r.pending = append(r.pending, p[:take]...)
if err := r.process(false); err != nil {
r.err = err
return written, err
}
written += take
p = p[take:]
}
return written, nil
}
func (r *multipartStreamRedactor) Close() error {
if r.err != nil {
return r.err
}
if err := r.process(true); err != nil {
r.err = err
return err
}
switch r.state {
case multipartBody:
if !r.suppress {
_, r.err = r.dst.Write(r.pending)
} else {
r.protected.appendBytes(r.pending)
r.err = r.emitProtected()
}
r.pending = nil
if r.err == nil {
r.err = errMalformedMultipart
}
case multipartEpilogue:
_, r.err = r.dst.Write(r.pending)
r.pending = nil
default:
r.err = errMalformedMultipart
}
return r.err
}
func (r *multipartStreamRedactor) process(final bool) error {
for {
switch r.state {
case multipartPreamble:
progress, err := r.processPreamble(final)
if err != nil || !progress {
return err
}
case multipartHeaders:
progress, err := r.processHeaders()
if err != nil || !progress {
return err
}
case multipartBody:
progress, err := r.processBody(final)
if err != nil || !progress {
return err
}
case multipartEpilogue:
if len(r.pending) > 0 {
_, err := r.dst.Write(r.pending)
r.pending = nil
return err
}
return nil
}
}
}
func (r *multipartStreamRedactor) processPreamble(final bool) (bool, error) {
positions := []int{}
if bytes.HasPrefix(r.pending, r.marker) {
positions = append(positions, 0)
}
if i := bytes.Index(r.pending, r.lineMarker); i >= 0 {
positions = append(positions, i+2)
}
if len(positions) == 0 {
if len(r.pending) > maxMultipartPreambleBytes || final {
return false, errMalformedMultipart
}
return false, nil
}
pos := positions[0]
end, closing, status := r.delimiterEnd(pos, final)
if status == delimiterNeedMore {
if len(r.pending) > maxMultipartPreambleBytes+maxMultipartBoundaryBytes+8 {
return false, errMalformedMultipart
}
return false, nil
}
if status == delimiterInvalid {
return false, errMalformedMultipart
}
if _, err := r.dst.Write(r.pending[:end]); err != nil {
return false, err
}
r.consumePending(end)
if closing {
r.state = multipartEpilogue
} else {
r.state = multipartHeaders
}
return true, nil
}
func (r *multipartStreamRedactor) processHeaders() (bool, error) {
i := bytes.Index(r.pending, []byte("\r\n\r\n"))
if i < 0 {
if len(r.pending) > maxMultipartHeaderBytes {
return false, errRedactionLimit
}
return false, nil
}
end := i + 4
block := r.pending[:end]
rewritten, matched, nested, err := parseMultipartHeadersReportedWithScratch(
block,
r.fields,
r.protected.session,
&r.headerFold,
)
if err != nil {
return false, err
}
if nested && !matched {
return false, fmt.Errorf("%w: nested multipart in unmatched part", errMalformedMultipart)
}
if _, err := r.dst.Write(rewritten); err != nil {
return false, err
}
r.consumePending(end)
r.suppress = matched
if matched {
r.protected.reset(r.protected.session)
if r.protected.redactImmediately() {
if _, err := io.WriteString(r.dst, redactedValue); err != nil {
return false, err
}
}
}
r.state = multipartBody
return true, nil
}
func (r *multipartStreamRedactor) processBody(final bool) (bool, error) {
for {
i := bytes.Index(r.pending, r.lineMarker)
if i < 0 {
keep := len(r.lineMarker) - 1
if final {
return false, nil
}
if len(r.pending) <= keep {
return false, nil
}
emit := len(r.pending) - keep
if !r.suppress {
if _, err := r.dst.Write(r.pending[:emit]); err != nil {
return false, err
}
} else {
r.protected.appendBytes(r.pending[:emit])
}
r.pending = append(r.pending[:0], r.pending[emit:]...)
return false, nil
}
pos := i + 2
end, closing, status := r.delimiterEnd(pos, final)
if status == delimiterNeedMore {
if i > 0 {
if !r.suppress {
if _, err := r.dst.Write(r.pending[:i]); err != nil {
return false, err
}
} else {
r.protected.appendBytes(r.pending[:i])
}
r.pending = append(r.pending[:0], r.pending[i:]...)
}
return false, nil
}
if status == delimiterInvalid {
emit := i + 2
if !r.suppress {
if _, err := r.dst.Write(r.pending[:emit]); err != nil {
return false, err
}
}
r.pending = append(r.pending[:0], r.pending[emit:]...)
continue
}
if !r.suppress {
if _, err := r.dst.Write(r.pending[:i]); err != nil {
return false, err
}
}
if r.suppress {
r.protected.appendBytes(r.pending[:i])
if err := r.emitProtected(); err != nil {
return false, err
}
}
if _, err := r.dst.Write(r.pending[i:end]); err != nil {
return false, err
}
r.consumePending(end)
r.suppress = false
if closing {
r.state = multipartEpilogue
} else {
r.state = multipartHeaders
}
return true, nil
}
}
func (r *multipartStreamRedactor) consumePending(n int) {
remaining := len(r.pending) - n
if remaining <= multipartCompactBytes {
copy(r.pending, r.pending[n:])
r.pending = r.pending[:remaining]
return
}
r.pending = r.pending[n:]
}
func (r *multipartStreamRedactor) emitProtected() error {
value := r.protected.protectedBytes()
if len(value) == 0 {
return nil
}
_, err := r.dst.Write(value)
return err
}
type delimiterStatus byte
const (
delimiterNeedMore delimiterStatus = iota
delimiterInvalid
delimiterValid
)
func (r *multipartStreamRedactor) delimiterEnd(pos int, final bool) (end int, closing bool, status delimiterStatus) {
i := pos + len(r.marker)
if len(r.pending) < i {
return 0, false, delimiterNeedMore
}
if len(r.pending) == i+1 && r.pending[i] == '-' {
return 0, false, delimiterNeedMore
}
if len(r.pending) >= i+2 && bytes.Equal(r.pending[i:i+2], []byte("--")) {
closing = true
i += 2
}
for i < len(r.pending) && (r.pending[i] == ' ' || r.pending[i] == '\t') {
i++
}
if len(r.pending) >= i+2 && bytes.Equal(r.pending[i:i+2], []byte("\r\n")) {
return i + 2, closing, delimiterValid
}
if closing && final && i == len(r.pending) {
return i, true, delimiterValid
}
if i == len(r.pending) || (i+1 == len(r.pending) && r.pending[i] == '\r') {
return 0, false, delimiterNeedMore
}
return 0, false, delimiterInvalid
}
func parseMultipartHeaders(block []byte, fields map[string]struct{}, protectors ...*sensitiveValueProtector) ([]byte, bool, bool, error) {
protector := newSensitiveValueProtector(SensitiveValueProtection{})
if len(protectors) > 0 && protectors[0] != nil {
protector = protectors[0]
}
return parseMultipartHeadersReported(block, fields, newBodyValueProtector(protector))
}
func parseMultipartHeadersReported(block []byte, fields map[string]struct{}, protector BodyValueProtector) ([]byte, bool, bool, error) {
return parseMultipartHeadersReportedWithScratch(block, fields, protector, nil)
}
type multipartHeaderInfo struct {
disposition []byte
contentType []byte
dispositionValueStart int
dispositionValueEnd int
}
func parseMultipartHeadersReportedWithScratch(
block []byte,
fields map[string]struct{},
protector BodyValueProtector,
fold *[]byte,
) ([]byte, bool, bool, error) {
info, err := scanMultipartHeaders(block)
if err != nil {
return nil, false, false, err
}
name, hasFilename, simple := parseSimpleFormData(info.disposition)
if simple {
matched := multipartFieldMatches(fields, name, fold)
nested, err := multipartContentTypeIsNested(info.contentType)
if err != nil {
return nil, false, false, err
}
if !matched || !hasFilename {
return block, matched, nested, nil
}
}
dispType, params, err := mime.ParseMediaType(string(info.disposition))
if err != nil || !strings.EqualFold(dispType, "form-data") || params["name"] == "" {
return nil, false, false, fmt.Errorf("%w: invalid content-disposition", errMalformedMultipart)
}
matched := multipartFieldMatches(fields, []byte(params["name"]), fold)
nested, err := multipartContentTypeIsNested(info.contentType)
if err != nil {
return nil, false, false, err
}
if !matched || params["filename"] == "" {
return block, matched, nested, nil
}
value := protector.NewValue()
if _, err := io.WriteString(value, params["filename"]); err != nil {
return nil, false, false, fmt.Errorf("%w: protect filename: %v", errMalformedMultipart, err)
}
var filename strings.Builder
if err := value.FinishTo(&filename); err != nil {
return nil, false, false, fmt.Errorf("%w: protect filename: %v", errMalformedMultipart, err)
}
params["filename"] = filename.String()
formatted := mime.FormatMediaType(dispType, params)
if formatted == "" {
return nil, false, false, fmt.Errorf("%w: cannot rewrite content-disposition", errMalformedMultipart)
}
rewritten := make([]byte, 0, len(block)-len(info.disposition)+len(formatted))
rewritten = append(rewritten, block[:info.dispositionValueStart]...)
rewritten = append(rewritten, formatted...)
rewritten = append(rewritten, block[info.dispositionValueEnd:]...)
return rewritten, matched, nested, nil
}
func scanMultipartHeaders(block []byte) (multipartHeaderInfo, error) {
if len(block) < 4 || !bytes.HasSuffix(block, []byte("\r\n\r\n")) {
return multipartHeaderInfo{}, fmt.Errorf("%w: incomplete part headers", errMalformedMultipart)
}
var info multipartHeaderInfo
contentEnd := len(block) - 4
for start := 0; start < contentEnd; {
lineEnd := contentEnd
if relative := bytes.Index(block[start:contentEnd], []byte("\r\n")); relative >= 0 {
lineEnd = start + relative
}
line := block[start:lineEnd]
if len(line) == 0 || line[0] == ' ' || line[0] == '\t' {
return multipartHeaderInfo{}, fmt.Errorf("%w: folded or empty part header", errMalformedMultipart)
}
colon := bytes.IndexByte(line, ':')
if colon <= 0 {
return multipartHeaderInfo{}, fmt.Errorf("%w: invalid part header", errMalformedMultipart)
}
name := bytes.TrimSpace(line[:colon])
value := bytes.TrimSpace(line[colon+1:])
switch {
case bytes.EqualFold(name, []byte("content-disposition")):
if info.disposition != nil {
return multipartHeaderInfo{}, fmt.Errorf("%w: duplicate content-disposition", errMalformedMultipart)
}
valueStart := start + colon + 1
for valueStart < lineEnd && (block[valueStart] == ' ' || block[valueStart] == '\t') {
valueStart++
}
info.disposition = value
info.dispositionValueStart = valueStart
info.dispositionValueEnd = lineEnd
case bytes.EqualFold(name, []byte("content-type")):
if info.contentType != nil {
return multipartHeaderInfo{}, fmt.Errorf("%w: duplicate content-type", errMalformedMultipart)
}
info.contentType = value
}
start = lineEnd + 2
}
if info.disposition == nil {
return multipartHeaderInfo{}, fmt.Errorf("%w: missing content-disposition", errMalformedMultipart)
}
return info, nil
}
func parseSimpleFormData(value []byte) (name []byte, hasFilename bool, ok bool) {
position := skipMultipartWhitespace(value, 0)
end := position
for end < len(value) && value[end] != ';' {
end++
}
if !bytes.EqualFold(bytes.TrimSpace(value[position:end]), []byte("form-data")) {
return nil, false, false
}
position = end
seenName := false
seenFilename := false
for position < len(value) {
if value[position] != ';' {
return nil, false, false
}
position = skipMultipartWhitespace(value, position+1)
attributeStart := position
for position < len(value) && isMIMETokenByte(value[position]) {
position++
}
if attributeStart == position {
return nil, false, false
}
attribute := value[attributeStart:position]
position = skipMultipartWhitespace(value, position)
if position >= len(value) || value[position] != '=' {
return nil, false, false
}
position = skipMultipartWhitespace(value, position+1)
parameter, next, parsed := parseSimpleMediaParameter(value, position)
if !parsed {
return nil, false, false
}
position = skipMultipartWhitespace(value, next)
switch {
case bytes.EqualFold(attribute, []byte("name")):
if seenName || len(parameter) == 0 {
return nil, false, false
}
name = parameter
seenName = true
case bytes.EqualFold(attribute, []byte("filename")):
if seenFilename {
return nil, false, false
}
hasFilename = true
seenFilename = true
default:
// Unknown, extended, and continuation parameters need the standard
// library's duplicate, decoding, and precedence rules.
return nil, false, false
}
}
return name, hasFilename, seenName
}
func parseSimpleMediaParameter(value []byte, position int) ([]byte, int, bool) {
if position >= len(value) {
return nil, position, false
}
if value[position] == '"' {
start := position + 1
position = start
for position < len(value) && value[position] != '"' {
if value[position] == '\\' || value[position] < 0x20 || value[position] >= 0x7f {
return nil, position, false
}
position++
}
if position >= len(value) {
return nil, position, false
}
return value[start:position], position + 1, true
}
start := position
for position < len(value) && isMIMETokenByte(value[position]) {
position++
}
if start == position {
return nil, position, false
}
return value[start:position], position, true
}
func multipartFieldMatches(fields map[string]struct{}, name []byte, fold *[]byte) bool {
hasUpper := false
for _, b := range name {
if b >= utf8.RuneSelf {
_, matched := fields[strings.ToLower(string(name))]
return matched
}
if b >= 'A' && b <= 'Z' {
hasUpper = true
}
}
if !hasUpper {
_, matched := fields[string(name)]
return matched
}
if fold != nil {
folded, _ := foldASCIIName(name, (*fold)[:0])
*fold = folded
_, matched := fields[string(folded)]
return matched
}
_, matched := fields[strings.ToLower(string(name))]
return matched
}
func multipartContentTypeIsNested(value []byte) (bool, error) {
if len(value) == 0 {
return false, nil
}
if simpleMediaType(value) {
return len(value) > len("multipart/") && bytes.EqualFold(value[:len("multipart/")], []byte("multipart/")), nil
}
mediaType, _, err := mime.ParseMediaType(string(value))
if err != nil {
return false, fmt.Errorf("%w: invalid part content-type", errMalformedMultipart)
}
return strings.HasPrefix(strings.ToLower(mediaType), "multipart/"), nil
}
func simpleMediaType(value []byte) bool {
slash := 0
for _, b := range value {
if b == '/' {
slash++
continue
}
if !isMIMETokenByte(b) {
return false
}
}
return slash == 1
}
func skipMultipartWhitespace(value []byte, position int) int {
for position < len(value) && (value[position] == ' ' || value[position] == '\t') {
position++
}
return position
}
func isMIMETokenByte(b byte) bool {
if b <= 0x20 || b >= 0x7f {
return false
}
return !strings.ContainsRune(`()<>@,;:\"/[]?=`, rune(b))
}
package recorder
// InternalErrorMode controls how recorder-internal failures are reported by a
// Transport or AsyncRecorder. Protection failures are aggregated per exchange
// direction to avoid log storms. Internal errors never replace the wrapped
// HTTP result.
type InternalErrorMode int
const (
// InternalErrorIgnore does not log internal errors. A configured
// OnInternalError callback still receives them. This is the default.
InternalErrorIgnore InternalErrorMode = iota
// InternalErrorLog behaves like InternalErrorIgnore but additionally writes
// the error using Logf, or the standard log package when Logf is nil.
InternalErrorLog
)
// Config configures a Transport. The zero value disables every optional
// content/metadata capture; finalized entries still contain the core exchange
// lifecycle and body byte/completion accounting. Pass DefaultConfig explicitly
// to NewTransport for sensible production defaults.
//
// Config is copied by NewTransport. Maps, callbacks, policies, stores and other
// referenced values must be treated as immutable after construction.
type Config struct {
// CaptureRequestBody enables storing request body content. Body size and
// completion state are always tracked, even when this is false.
CaptureRequestBody bool
// CaptureResponseBody enables storing response body content.
CaptureResponseBody bool
// EmbedBodies controls whether captured body content is embedded into
// the HAR document (postData.text / content.text). When false, bodies
// are still captured into the BodyStore and the record keeps sizes,
// hashes, truncation state and the store reference
// (_recorder.requestBody/responseBody.store) — the intended production
// setting together with FileBodyStore, keeping HAR documents small
// while body content stays retrievable. Like the Capture* flags, the
// zero value disables embedding; set EmbedBodies to enable it.
EmbedBodies bool
// MaxRequestBodyBytes limits how many request body bytes are stored.
// Values <= 0 mean unlimited. The total size keeps being counted after
// the limit is reached; only content capture stops.
MaxRequestBodyBytes int64
// MaxResponseBodyBytes is the response-side equivalent of
// MaxRequestBodyBytes.
MaxResponseBodyBytes int64
// CaptureTLS enables _recorder.tls built from tls.ConnectionState.
CaptureTLS bool
// CaptureCertificates includes peer certificate details in _recorder.tls.
CaptureCertificates bool
// CaptureRawCertificates additionally embeds each certificate's raw DER
// bytes as Base64. Off by default because of size.
CaptureRawCertificates bool
// CaptureHeaders enables recording request/response headers and trailers.
CaptureHeaders bool
// CaptureCookies enables recording parsed cookies.
CaptureCookies bool
// Redaction contains global common and direction-specific selector rules
// plus trusted custom body redactors. Assign the baseline directly;
// request contexts may add more through WithRequestRedaction.
Redaction RedactionConfig
// SensitiveValueProtection controls whether values selected by configured
// redaction rules and body redactors are removed, reversibly encrypted, or
// deterministically tokenized. The zero value redacts with [REDACTED].
// Protection failures always fail closed to [REDACTED].
SensitiveValueProtection SensitiveValueProtection
// HashBodies enables hashing of body streams. The hash covers every byte
// that actually flowed, including bytes beyond the capture limit.
HashBodies bool
// BodyHashAlgorithm selects the hash: "sha256" (default), "sha1" or
// "md5". Validate rejects unknown values; unchecked configurations retain
// the fail-safe sha256 runtime fallback.
BodyHashAlgorithm string
// CaptureRawTrace stores every raw httptrace event under _recorder.trace. Event
// details pass through RedactErrorMessage before export.
CaptureRawTrace bool
// ContentDecoders maps Content-Encoding tokens (lower-case) to decoders
// used by the capture and HAR-building pipeline. With body redaction
// configured, supported encodings are decoded and redacted
// while streaming into the BodyStore; otherwise a fully captured body
// may be decoded when embedded in the HAR. DefaultConfig installs the
// stdlib-only set (gzip, x-gzip, deflate); add entries to the map for
// encodings such as brotli or zstd — see ContentDecoder for
// ready-to-paste recipes. Decoding never touches bytes read by the caller.
ContentDecoders map[string]ContentDecoder
// BodyCapturePolicy optionally overrides capture, embedding, hashing,
// limits, and body-redactor selection for each request and response body.
// Policy errors and panics fail closed to metadata-only recording.
BodyCapturePolicy BodyCapturePolicy
// HeadSamplingPolicy decides before exchange instrumentation whether to
// record fully, retain metadata only, or bypass recording entirely.
HeadSamplingPolicy HeadSamplingPolicy
// OnHeadSamplingDecision, when set, observes the effective head-sampling
// decision before instrumentation and the wrapped transport run. It also
// receives HeadSampleDrop decisions, for which no entry or completion
// callback will exist.
OnHeadSamplingDecision OnHeadSamplingDecision
// RetentionPolicy decides whether a finalized entry reaches Recorder. The
// completion callback receives the effective result after required asset
// cleanup; capture cost has already been paid.
RetentionPolicy RetentionPolicy
// BodyStore provides transactional storage for captured body bytes. Writers
// commit on body finalization and abort on retry or processing/storage
// failure. Nil means MemoryBodyStore.
BodyStore BodyStore
// InternalErrorMode selects the internal error policy. See the constants.
InternalErrorMode InternalErrorMode
// OnInternalError, when set, receives every recorder-internal error.
OnInternalError func(error)
// Logf is used by InternalErrorLog. Nil falls back to log.Printf.
Logf func(format string, args ...any)
// OnEntryCompleted, when set, is invoked with the request context and the
// finished entry every time an instrumented exchange is finalized. It runs
// after effective retention is known and before a kept entry is offered to
// Recorder.Record. The entry is borrowed for the callback duration; retaining
// ownership requires a Recorder. Discarded external assets are no longer
// readable when the callback runs.
OnEntryCompleted OnEntryCompleted
// RedactErrorMessage, when set, is applied to every error message and raw
// trace detail before export (they can contain URLs or credentials).
RedactErrorMessage func(string) string
}
// DefaultConfig returns the recommended production-safe configuration. Body
// streams are counted for lifecycle and size metadata, but their
// content is neither stored, embedded nor hashed by default. TLS metadata,
// headers and cookies remain enabled, with DefaultRedactedHeaders applied.
func DefaultConfig() Config {
return Config{
MaxRequestBodyBytes: 1 << 20,
MaxResponseBodyBytes: 1 << 20,
CaptureTLS: true,
CaptureCertificates: true,
CaptureHeaders: true,
CaptureCookies: true,
Redaction: RedactionConfig{Common: RedactionRules{
Headers: DefaultRedactedHeaders(),
}},
BodyHashAlgorithm: "sha256",
ContentDecoders: defaultContentDecoders(),
InternalErrorMode: InternalErrorLog,
}
}
// DefaultRedactedHeaders returns the headers redacted by default.
func DefaultRedactedHeaders() []string {
return []string{"Authorization", "Proxy-Authorization", "Cookie", "Set-Cookie", "X-API-Key"}
}
package recorder
import (
"errors"
"fmt"
"mime"
"sort"
"strings"
)
// Validate reports static configuration errors without performing I/O or
// invoking callbacks, policies, stores, decoders, redactors, or key providers.
// Both Config{} and DefaultConfig() are valid. Call Validate after applying
// application settings and before passing the copied configuration to
// NewTransport.
func (c Config) Validate() error {
var errs []error
errs = append(errs, validateCertificateConfig(c)...)
errs = append(errs, validateHashConfig(c)...)
errs = append(errs, validateProtectionConfig(c)...)
errs = append(errs, validateInternalErrorConfig(c)...)
errs = append(errs, validateContentDecoders(c.ContentDecoders)...)
errs = append(errs, validateRedactionConfig(c.Redaction)...)
return errors.Join(errs...)
}
func validateCertificateConfig(c Config) []error {
var errs []error
if c.CaptureCertificates && !c.CaptureTLS {
errs = append(errs, errors.New("recorder: Config.CaptureCertificates requires CaptureTLS"))
}
if c.CaptureRawCertificates && !c.CaptureCertificates {
errs = append(errs, errors.New("recorder: Config.CaptureRawCertificates requires CaptureCertificates"))
}
if c.CaptureRawCertificates && !c.CaptureTLS {
errs = append(errs, errors.New("recorder: Config.CaptureRawCertificates requires CaptureTLS"))
}
return errs
}
func validateHashConfig(c Config) []error {
switch c.BodyHashAlgorithm {
case "", "sha256", "sha1", "md5":
return nil
default:
return []error{fmt.Errorf(
"recorder: Config.BodyHashAlgorithm %q is invalid; use sha256, sha1, or md5",
c.BodyHashAlgorithm,
)}
}
}
func validateProtectionConfig(c Config) []error {
protection := c.SensitiveValueProtection
switch protection.Mode {
case "", ProtectionRedact:
return nil
case ProtectionEncrypt, ProtectionTokenize:
if protection.KeyProvider == nil {
return []error{fmt.Errorf(
"recorder: Config.SensitiveValueProtection.KeyProvider is required for mode %q",
protection.Mode,
)}
}
return nil
default:
return []error{fmt.Errorf(
"recorder: Config.SensitiveValueProtection.Mode %q is invalid",
protection.Mode,
)}
}
}
func validateInternalErrorConfig(c Config) []error {
switch c.InternalErrorMode {
case InternalErrorIgnore, InternalErrorLog:
return nil
default:
return []error{fmt.Errorf(
"recorder: Config.InternalErrorMode %d is invalid",
c.InternalErrorMode,
)}
}
}
func validateContentDecoders(decoders map[string]ContentDecoder) []error {
keys := sortedKeys(decoders)
errs := make([]error, 0)
for _, encoding := range keys {
decoder := decoders[encoding]
if !validContentCoding(encoding) {
errs = append(errs, fmt.Errorf(
"recorder: Config.ContentDecoders key %q must be one lower-case HTTP content-coding token",
encoding,
))
}
if decoder == nil {
errs = append(errs, fmt.Errorf(
"recorder: Config.ContentDecoders[%q] must not be nil",
encoding,
))
}
}
return errs
}
func validateRedactionConfig(config RedactionConfig) []error {
var errs []error
errs = append(errs, validateRedactionRules("Common", config.Common)...)
errs = append(errs, validateRedactionRules("Request", config.Request)...)
errs = append(errs, validateRedactionRules("Response", config.Response)...)
return errs
}
func validateRedactionRules(scope string, rules RedactionRules) []error {
var errs []error
errs = append(errs, validateSelectorNames(scope, "Headers", rules.Headers)...)
errs = append(errs, validateSelectorNames(scope, "QueryParameters", rules.QueryParameters)...)
errs = append(errs, validateSelectorNames(scope, "Cookies", rules.Cookies)...)
errs = append(errs, validateSelectorNames(scope, "JSONFields", rules.JSONFields)...)
errs = append(errs, validateSelectorNames(scope, "XMLElements", rules.XMLElements)...)
errs = append(errs, validateBodyRedactors(scope, rules.BodyRedactors)...)
return errs
}
func validateSelectorNames(scope, field string, names []string) []error {
var errs []error
for index, name := range names {
switch {
case strings.TrimSpace(name) == "":
errs = append(errs, fmt.Errorf(
"recorder: Config.Redaction.%s.%s[%d] must not be empty",
scope,
field,
index,
))
case name != strings.TrimSpace(name):
errs = append(errs, fmt.Errorf(
"recorder: Config.Redaction.%s.%s[%d] must not have surrounding whitespace",
scope,
field,
index,
))
}
}
return errs
}
func validateBodyRedactors(scope string, redactors map[string]BodyRedactor) []error {
keys := sortedKeys(redactors)
var errs []error
for _, mediaType := range keys {
redactor := redactors[mediaType]
base, params, err := mime.ParseMediaType(mediaType)
if err != nil || len(params) != 0 || mediaType != strings.ToLower(strings.TrimSpace(mediaType)) || base != mediaType {
errs = append(errs, fmt.Errorf(
"recorder: Config.Redaction.%s.BodyRedactors key %q must be a normalized base MIME type",
scope,
mediaType,
))
}
if redactor == nil {
errs = append(errs, fmt.Errorf(
"recorder: Config.Redaction.%s.BodyRedactors[%q] must not be nil",
scope,
mediaType,
))
}
}
return errs
}
func validContentCoding(value string) bool {
if value == "" || value == "identity" || value == "*" || value != strings.ToLower(strings.TrimSpace(value)) {
return false
}
for index := 0; index < len(value); index++ {
if !httpTokenByte(value[index]) {
return false
}
}
return true
}
func httpTokenByte(value byte) bool {
switch {
case value >= '0' && value <= '9':
return true
case value >= 'a' && value <= 'z':
return true
}
return strings.ContainsRune("!#$%&'*+-.^_`|~", rune(value))
}
func sortedKeys[T any](values map[string]T) []string {
keys := make([]string, 0, len(values))
for key := range values {
keys = append(keys, key)
}
sort.Strings(keys)
return keys
}
package recorder
import (
"context"
"crypto/rand"
"encoding/hex"
"sync/atomic"
)
// Recorder receives finalized, immutable entries. Implementations must be
// safe for concurrent use: entries arrive from whichever goroutine finishes
// reading a response body. Record reports synchronous sink failures; Transport
// contains and forwards them through its configured internal-error policy.
type Recorder interface {
Record(entry *Entry) error
}
// batchRecorder is an optional Recorder capability for sinks that can process
// multiple finalized entries more efficiently under one lock or write. The
// slice and its immutable entries are owned by the caller and must not be
// retained or mutated. AsyncRecorder discovers this capability when batching
// is explicitly configured. RecordBatch reports synchronous sink failures.
type batchRecorder interface {
Recorder
RecordBatch(entries []*Entry) error
}
// TraceStore is an optional capability for recorders that retain entries and
// can query or remove them by trace ID (see WithTraceID). The Recorder
// interface itself stays minimal — a custom recorder is still just one
// method — and callers discover the capability by type assertion:
//
// if ts, ok := rec.(TraceStore); ok {
// entries := ts.TakeTrace(id)
// }
//
// MemoryRecorder and HARFileRecorder implement TraceStore.
// JSONStreamRecorder cannot: its entries leave the process the moment they
// are recorded, so there is nothing left to query or remove.
type TraceStore interface {
// EntriesByTrace returns the entries belonging to the trace ID, in
// recording order, without removing them.
EntriesByTrace(traceID string) []*Entry
// RemoveTrace deletes the trace's entries and returns how many were
// removed.
RemoveTrace(traceID string) int
// TakeTrace atomically removes and returns the trace's entries, in
// recording order.
TakeTrace(traceID string) []*Entry
}
// Compile-time capability checks.
var (
_ TraceStore = (*MemoryRecorder)(nil)
_ TraceStore = (*HARFileRecorder)(nil)
_ batchRecorder = (*MemoryRecorder)(nil)
_ batchRecorder = (*HARFileRecorder)(nil)
_ batchRecorder = (*JSONStreamRecorder)(nil)
_ batchRecorder = (*multiRecorder)(nil)
)
// RecorderFunc adapts a function into a Recorder (the "callback recorder").
// The function must be safe for concurrent use.
type RecorderFunc func(*Entry) error
// Record implements Recorder.
func (f RecorderFunc) Record(e *Entry) error { return f(e) }
// OnEntryCompleted is invoked with the request context, finished HAR entry,
// and effective retention disposition. It runs after retention and any
// required asset release, but before a kept entry is offered to Recorder.
// The entry is borrowed only for the callback duration. External assets remain
// readable for EntryDispositionKeep; references in EntryDispositionDiscard
// entries have already been released and must not be opened.
type OnEntryCompleted func(context.Context, *Entry, EntryDisposition)
type traceCtxKey struct{}
// traceState correlates the exchanges of one logical operation. The atomic
// counter yields the redirect index: http.Client carries the original request
// context across redirect hops, so every hop sees the same state.
type traceState struct {
id string
seq atomic.Int64
}
// WithTraceID returns a context carrying the given correlation ID. Every
// exchange started under this context records the ID as _recorder.traceId and
// an incrementing redirectIndex, which links redirect chains followed by
// http.Client into one logical trace.
func WithTraceID(ctx context.Context, id string) context.Context {
return context.WithValue(ctx, traceCtxKey{}, &traceState{id: id})
}
// TraceContext is a convenience wrapper around WithTraceID that generates a
// fresh random trace ID and returns it alongside the derived context.
func TraceContext(ctx context.Context) (context.Context, string) {
id := newID()
return WithTraceID(ctx, id), id
}
// TraceIDFromContext returns the correlation ID installed by WithTraceID.
func TraceIDFromContext(ctx context.Context) (string, bool) {
if ts, ok := ctx.Value(traceCtxKey{}).(*traceState); ok {
return ts.id, true
}
return "", false
}
func traceStateFromContext(ctx context.Context) *traceState {
ts, _ := ctx.Value(traceCtxKey{}).(*traceState)
return ts
}
// newID returns a 128-bit random hex identifier.
func newID() string {
var b [16]byte
_, _ = rand.Read(b[:]) // crypto/rand.Read does not fail on supported platforms
return hex.EncodeToString(b[:])
}
package recorder
import (
"bytes"
"context"
"net/http"
"net/url"
"sort"
"strings"
)
const (
redactedMarker = "REDACTED"
redactedValue = "[" + redactedMarker + "]"
jsonRedactedValue = `"` + redactedValue + `"`
)
// redactor applies the configured rules to recorded copies during body
// capture and entry construction. It never mutates live http.Request /
// http.Response objects. Immutable after construction, safe for concurrent
// use.
type redactor struct {
headers map[string]struct{}
query map[string]struct{}
cookies map[string]struct{}
jsonFields map[string]struct{}
xmlElements map[string]struct{}
bodyRedactors map[string]BodyRedactor
protector *sensitiveValueProtector
errFn func(string) string
audit *redactionAudit
direction BodyDirection
}
func (r *redactor) withAudit(audit *redactionAudit, direction BodyDirection) *redactor {
clone := *r
clone.audit = audit
clone.direction = direction
return &clone
}
func (r *redactor) withContext(ctx context.Context, keys *protectionKeyCache) *redactor {
clone := *r
clone.protector = r.protector.withContext(ctx, keys)
return &clone
}
func newRedactor(o *Config) *redactor {
rules := effectiveRedactionRules(o.Redaction.Common, o.Redaction.Request)
return newRedactorWithRules(o, rules)
}
func newRedactorWithRules(o *Config, rules RedactionRules) *redactor {
return &redactor{
headers: lowerSet(rules.Headers),
query: lowerSet(rules.QueryParameters),
cookies: lowerSet(rules.Cookies),
jsonFields: lowerSet(rules.JSONFields),
xmlElements: lowerSet(rules.XMLElements),
errFn: o.RedactErrorMessage,
bodyRedactors: normalizedBodyRedactors(rules.BodyRedactors),
protector: newSensitiveValueProtector(o.SensitiveValueProtection),
}
}
func normalizedBodyRedactors(in map[string]BodyRedactor) map[string]BodyRedactor {
if len(in) == 0 {
return nil
}
out := make(map[string]BodyRedactor, len(in))
for mediaType, redactor := range in {
if normalized := baseMimeType(mediaType); normalized != "" && redactor != nil {
out[normalized] = redactor
}
}
return out
}
func (r *redactor) withBodyRedactor(contentType string, bodyRedactor BodyRedactor) *redactor {
if bodyRedactor == nil {
return r
}
clone := *r
clone.bodyRedactors = make(map[string]BodyRedactor, len(r.bodyRedactors)+1)
for mediaType, registered := range r.bodyRedactors {
clone.bodyRedactors[mediaType] = registered
}
clone.bodyRedactors[baseMimeType(contentType)] = bodyRedactor
return &clone
}
func (r *redactor) withRules(rules RedactionRules) *redactor {
if r == nil {
return nil
}
clone := *r
clone.headers = unionLowerSet(r.headers, rules.Headers)
clone.query = unionLowerSet(r.query, rules.QueryParameters)
clone.cookies = unionLowerSet(r.cookies, rules.Cookies)
clone.jsonFields = unionLowerSet(r.jsonFields, rules.JSONFields)
clone.xmlElements = unionLowerSet(r.xmlElements, rules.XMLElements)
clone.bodyRedactors = mergeNormalizedBodyRedactors(r.bodyRedactors, rules.BodyRedactors)
return &clone
}
func unionLowerSet(base map[string]struct{}, names []string) map[string]struct{} {
if len(names) == 0 {
return base
}
out := make(map[string]struct{}, len(base)+len(names))
for name := range base {
out[name] = struct{}{}
}
for _, name := range names {
out[strings.ToLower(name)] = struct{}{}
}
return out
}
func mergeNormalizedBodyRedactors(base map[string]BodyRedactor, additions map[string]BodyRedactor) map[string]BodyRedactor {
if len(additions) == 0 {
return base
}
out := make(map[string]BodyRedactor, len(base)+len(additions))
for mediaType, redactor := range base {
out[mediaType] = redactor
}
for mediaType, redactor := range normalizedBodyRedactors(additions) {
out[mediaType] = redactor
}
return out
}
func lowerSet(names []string) map[string]struct{} {
if len(names) == 0 {
return nil
}
set := make(map[string]struct{}, len(names))
for _, n := range names {
set[strings.ToLower(n)] = struct{}{}
}
return set
}
func (r *redactor) headerRedacted(name string) bool {
_, ok := r.headers[strings.ToLower(name)]
return ok
}
func (r *redactor) queryRedacted(name string) bool {
_, ok := r.query[strings.ToLower(name)]
return ok
}
func (r *redactor) protectString(value string) string {
protected, mode, reason, err := r.protector.protectWithError([]byte(value))
r.audit.addProtection(r.direction, mode, reason)
r.audit.addProtectionFailure(r.direction, err, 1)
return protected
}
// cookieRedacted reports whether a cookie value must be hidden: either the
// cookie name is listed, or the header that carried it (Cookie / Set-Cookie)
// is itself redacted.
func (r *redactor) cookieRedacted(name, carrierHeader string) bool {
if _, ok := r.cookies[strings.ToLower(name)]; ok {
return true
}
return r.headerRedacted(carrierHeader)
}
// headerPairs converts a header map into sorted HAR pairs, applying header
// redaction. hostValue, when non-empty, is prepended as a "Host" pair (Go
// keeps the Host header outside http.Header).
func (r *redactor) headerPairs(h http.Header, hostValue string) []NameValuePair {
pairs := make([]NameValuePair, 0, len(h)+1)
if hostValue != "" {
pairs = append(pairs, NameValuePair{Name: "Host", Value: hostValue})
}
names := make([]string, 0, len(h))
for name := range h {
names = append(names, name)
}
sort.Strings(names)
for _, name := range names {
if r.headerRedacted(name) {
var changed int64
for _, value := range h[name] {
if value != redactedValue {
changed++
}
}
r.audit.add(r.direction, "headers", changed)
pairs = append(pairs, NameValuePair{Name: name, Value: r.protectString(strings.Join(h[name], ", "))})
continue
}
for _, v := range h[name] {
pairs = append(pairs, NameValuePair{Name: name, Value: v})
}
}
return pairs
}
// redactPairs applies header redaction to an already-ordered pair list (the
// wire-order header fields reported by httptrace.WroteHeaderField).
func (r *redactor) redactPairs(pairs []NameValuePair) []NameValuePair {
out := make([]NameValuePair, len(pairs))
for i, p := range pairs {
if r.headerRedacted(p.Name) {
if p.Value != redactedValue {
r.audit.add(r.direction, "headers", 1)
}
p.Value = r.protectString(p.Value)
}
out[i] = p
}
return out
}
// queryPairs parses a raw query string preserving parameter order and
// duplicates (url.Values would lose both), applying query redaction.
func (r *redactor) queryPairs(rawQuery string) []NameValuePair {
pairs := []NameValuePair{}
if rawQuery == "" {
return pairs
}
for _, part := range strings.Split(rawQuery, "&") {
if part == "" {
continue
}
k, v, _ := strings.Cut(part, "=")
name := k
if u, err := url.QueryUnescape(k); err == nil {
name = u
}
value := v
if u, err := url.QueryUnescape(v); err == nil {
value = u
}
if r.queryRedacted(name) {
if value != redactedValue {
r.audit.add(r.direction, "query", 1)
}
value = r.protectString(value)
}
pairs = append(pairs, NameValuePair{Name: name, Value: value})
}
return pairs
}
// redactURL renders a URL with redacted userinfo password and query values.
// The original URL is never mutated.
func (r *redactor) redactURL(u *url.URL) string {
if u == nil {
return ""
}
cp := *u
if cp.User != nil {
if password, has := cp.User.Password(); has {
if password != redactedValue {
r.audit.add(r.direction, "url", 1)
}
cp.User = url.UserPassword(cp.User.Username(), r.protectString(password))
}
}
if cp.RawQuery != "" && len(r.query) > 0 {
var b strings.Builder
for i, part := range strings.Split(cp.RawQuery, "&") {
if i > 0 {
b.WriteByte('&')
}
k, value, hasEq := strings.Cut(part, "=")
name := k
if uq, err := url.QueryUnescape(k); err == nil {
name = uq
}
if hasEq && r.queryRedacted(name) {
decodedValue := value
if unescaped, err := url.QueryUnescape(value); err == nil {
decodedValue = unescaped
}
if decodedValue != redactedValue {
r.audit.add(r.direction, "url", 1)
}
b.WriteString(k)
b.WriteByte('=')
b.WriteString(url.QueryEscape(r.protectString(decodedValue)))
} else {
b.WriteString(part)
}
}
cp.RawQuery = b.String()
}
return cp.String()
}
// redactURLString applies URL redaction to absolute or relative URL text.
// Malformed values are preserved because rewriting an unparseable Location
// header could misrepresent the response.
func (r *redactor) redactURLString(raw string) string {
u, err := url.Parse(raw)
if err != nil {
return raw
}
return r.redactURL(u)
}
// urlBearingHeader reports header names whose values are URLs that can carry
// query parameters, so query redaction must reach them: Location and
// Content-Location on responses, and Referer on redirected requests — Go's
// client forwards the previous hop's URL including its query string, which
// would otherwise bypass query-parameter redaction rules.
func urlBearingHeader(name string) bool {
return strings.EqualFold(name, "Location") ||
strings.EqualFold(name, "Content-Location") ||
strings.EqualFold(name, "Referer")
}
// sanitizeURLHeaders applies URL redaction to URL-bearing header values in an
// already-redacted pair list (the list is owned by the caller and safe to
// mutate). Values that are fully redacted stay untouched.
func (r *redactor) sanitizeURLHeaders(pairs []NameValuePair) []NameValuePair {
for i := range pairs {
if pairs[i].Value != redactedValue && urlBearingHeader(pairs[i].Name) {
pairs[i].Value = r.redactURLString(pairs[i].Value)
}
}
return pairs
}
// responseHeaderPairs sanitizes URL-bearing values so query redaction cannot
// be bypassed through the response header list.
func (r *redactor) responseHeaderPairs(h http.Header) []NameValuePair {
return r.sanitizeURLHeaders(r.headerPairs(h, ""))
}
// redactJSONBody uses the same bounded streaming transformer used by body
// capture. Non-redacted bytes are emitted exactly as received.
func (r *redactor) redactJSONBody(b []byte) []byte {
if len(r.jsonFields) == 0 || len(b) == 0 {
return b
}
var out bytes.Buffer
s := newJSONStreamRedactor(&out, r.jsonFields, newBodyValueProtector(r.protector))
if _, err := s.Write(b); err != nil || s.Close() != nil {
return []byte(jsonRedactedValue)
}
return out.Bytes()
}
// redactStructuredBody applies field-level redaction appropriate for the
// body's media type. Generic/incorrect media types use the same bounded
// prefix sniffer as live body capture.
func (r *redactor) redactStructuredBody(mimeType string, b []byte) []byte {
var out bytes.Buffer
s := newBodyStreamRedactor(&out, mimeType, r)
if s == nil {
return b
}
if _, err := s.Write(b); err != nil || s.Close() != nil {
if isJSONMime(mimeType) {
return []byte(jsonRedactedValue)
}
if isFormMime(mimeType) {
return []byte(formRedactedValue)
}
if isMultipartFormMime(mimeType) {
return []byte(redactedValue)
}
return []byte(redactedValue)
}
return out.Bytes()
}
// redactXMLBody replaces the subtree inside matching elements with
// [REDACTED]. Elements match by case-insensitive local name, ignoring any
// namespace prefix, so "Password" covers <Password>, <wsse:Password>, and
// <ns2:PASSWORD> alike.
//
// The document is never re-encoded. Attribute values are NOT redacted.
func (r *redactor) redactXMLBody(b []byte) []byte {
if len(r.xmlElements) == 0 || len(b) == 0 {
return b
}
var out bytes.Buffer
s := newXMLStreamRedactor(&out, r.xmlElements, newBodyValueProtector(r.protector))
if _, err := s.Write(b); err != nil || s.Close() != nil {
return []byte(redactedValue)
}
return out.Bytes()
}
// redactError filters an error message through the configured redactor.
func (r *redactor) redactError(msg string) string {
if r.errFn != nil {
redacted := r.errFn(msg)
if redacted != msg {
r.audit.addError(1)
}
return redacted
}
return msg
}
// traceEvents returns an owned copy whose details pass through the same
// application-supplied sanitizer used for recorded error strings. Raw
// httptrace callbacks can embed dial/TLS/write errors and must not bypass the
// central redaction policy merely because CaptureRawTrace is opt-in.
func (r *redactor) traceEvents(events []TraceEvent) []TraceEvent {
if len(events) == 0 {
return nil
}
out := make([]TraceEvent, len(events))
copy(out, events)
for i := range out {
if r.errFn != nil {
redacted := r.errFn(out[i].Detail)
if redacted != out[i].Detail {
r.audit.addRawTrace(1)
}
out[i].Detail = redacted
}
}
return out
}
func (r *redactor) recordCookieRedaction() {
r.audit.add(r.direction, "cookies", 1)
}
package recorder
import "sync"
type redactionAudit struct {
mu sync.Mutex
request RedactionScopeInfo
response RedactionScopeInfo
errors int64
rawTrace int64
failures [2]protectionFailure
}
type protectionFailure struct {
first error
count int64
}
func (a *redactionAudit) scope(direction BodyDirection) *RedactionScopeInfo {
if direction == ResponseBody {
return &a.response
}
return &a.request
}
func (a *redactionAudit) add(direction BodyDirection, category string, n int64) {
if a == nil || n <= 0 {
return
}
a.mu.Lock()
defer a.mu.Unlock()
s := a.scope(direction)
switch category {
case "url":
s.URL += n
case "headers":
s.Headers += n
case "query":
s.QueryParameters += n
case "cookies":
s.Cookies += n
}
}
func (a *redactionAudit) addProtection(direction BodyDirection, mode ProtectionMode, reason string) {
if a == nil {
return
}
a.mu.Lock()
defer a.mu.Unlock()
s := a.scope(direction)
if s.Protection == nil {
s.Protection = &ProtectionCounts{}
}
switch mode {
case ProtectionEncrypt:
s.Protection.Encrypted++
case ProtectionTokenize:
s.Protection.Tokenized++
default:
s.Protection.Redacted++
}
if reason != "" {
if s.Protection.Fallbacks == nil {
s.Protection.Fallbacks = make(map[string]int64)
}
s.Protection.Fallbacks[reason]++
}
}
func (a *redactionAudit) addProtectionFailure(direction BodyDirection, err error, count int64) {
if a == nil || err == nil || count <= 0 {
return
}
index := 0
if direction == ResponseBody {
index = 1
}
a.mu.Lock()
defer a.mu.Unlock()
if a.failures[index].first == nil {
a.failures[index].first = err
}
a.failures[index].count += count
}
func (a *redactionAudit) protectionFailures() [2]protectionFailure {
if a == nil {
return [2]protectionFailure{}
}
a.mu.Lock()
defer a.mu.Unlock()
return a.failures
}
func (a *redactionAudit) addError(n int64) {
if a == nil || n <= 0 {
return
}
a.mu.Lock()
defer a.mu.Unlock()
a.errors += n
}
func (a *redactionAudit) addRawTrace(n int64) {
if a == nil || n <= 0 {
return
}
a.mu.Lock()
defer a.mu.Unlock()
a.rawTrace += n
}
func (a *redactionAudit) setBody(direction BodyDirection, body BodyRedactionInfo) {
if a == nil {
return
}
a.mu.Lock()
defer a.mu.Unlock()
copy := body
if body.Protection != nil {
protection := cloneProtectionCounts(*body.Protection)
copy.Protection = &protection
}
a.scope(direction).Body = ©
}
func (a *redactionAudit) snapshot() *RedactionInfo {
if a == nil {
return nil
}
a.mu.Lock()
defer a.mu.Unlock()
info := &RedactionInfo{Errors: a.errors, RawTrace: a.rawTrace}
if !redactionScopeEmpty(a.request) {
request := a.request
if request.Body != nil {
body := *request.Body
request.Body = &body
}
if request.Protection != nil {
protection := cloneProtectionCounts(*request.Protection)
request.Protection = &protection
}
info.Request = &request
}
if !redactionScopeEmpty(a.response) {
response := a.response
if response.Body != nil {
body := *response.Body
response.Body = &body
}
if response.Protection != nil {
protection := cloneProtectionCounts(*response.Protection)
response.Protection = &protection
}
info.Response = &response
}
if info.Request == nil && info.Response == nil && info.Errors == 0 && info.RawTrace == 0 {
return nil
}
return info
}
func redactionScopeEmpty(s RedactionScopeInfo) bool {
return s.URL == 0 && s.Headers == 0 && s.QueryParameters == 0 && s.Cookies == 0 && s.Body == nil && s.Protection == nil
}
package recorder
import (
"bufio"
"context"
"crypto/md5"
"crypto/sha1"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"hash"
"io"
"strings"
"sync"
)
// bodyCapture observes one body stream (request or response) as a tee: it
// counts every byte that flows, hashes the full stream, and stores content up
// to the configured limit in a BodyStore. It never generates reads of its
// own and never buffers the whole body for a later rewrite. Body
// redaction uses bounded parser/output buffers. A failing store or redactor
// only stops content capture — counting and the HTTP flow itself continue.
//
// bodyCapture is safe for concurrent use: the transport may still be
// streaming the request body from a background goroutine while the exchange
// is being finalized.
type bodyCapture struct {
mu sync.Mutex
ctx context.Context
store BodyStore
meta BodyMetadata
contentEncoding string
limit int64 // <= 0 means unlimited
captureContent bool
onInternal func(error)
red *redactor
decoder ContentDecoder
w BodyWriter
storeFailed bool
storedDecoded bool
storedRedacted bool
h hash.Hash
hashName string
// expected is the announced Content-Length (-1 when unknown). When a
// stream is closed after exactly expected bytes flowed, it is complete
// even without an explicit EOF read: http.Transport reads exactly
// Content-Length bytes from a request body and may never issue the
// final Read that would return io.EOF.
expected int64
finished bool
complete bool
closedEarly bool
truncated bool
captured int64
total int64
readErr error
closeErr error
}
func newBodyCapture(ctx context.Context, store BodyStore, meta BodyMetadata,
contentEncoding string, captureContent bool, limit int64, hashAlg string, hashBody bool, red *redactor, decoder ContentDecoder, onInternal func(error),
) *bodyCapture {
c := &bodyCapture{
ctx: ctx,
store: store,
meta: meta,
contentEncoding: contentEncoding,
limit: limit,
captureContent: captureContent,
onInternal: onInternal,
red: red,
decoder: decoder,
expected: -1,
}
if hashBody {
c.h, c.hashName = newBodyHash(hashAlg)
}
return c
}
func newBodyHash(alg string) (hash.Hash, string) {
switch strings.ToLower(alg) {
case "sha1":
return sha1.New(), "sha1"
case "md5":
return md5.New(), "md5"
default:
return sha256.New(), "sha256"
}
}
// observe records bytes that just flowed through the stream.
func (c *bodyCapture) observe(p []byte) {
if c == nil || len(p) == 0 {
return
}
c.mu.Lock()
if c.finished {
c.mu.Unlock()
return
}
c.total += int64(len(p))
if c.h != nil {
c.h.Write(p)
}
if !c.captureContent {
c.mu.Unlock()
return
}
take := int64(len(p))
if c.limit > 0 {
room := c.limit - c.captured
if room <= 0 {
c.truncated = true
c.mu.Unlock()
return
}
if room < take {
take = room
c.truncated = true
}
}
if c.storeFailed {
c.mu.Unlock()
return
}
var internalErr error
if c.w == nil {
enc := strings.ToLower(strings.TrimSpace(c.contentEncoding))
needsRedaction := bodyStreamRedactionEnabled(c.meta.ContentType, c.red)
if enc != "" && enc != "identity" && needsRedaction && c.decoder == nil {
c.storeFailed = true
internalErr = fmt.Errorf("recorder: no streaming decoder for redacted %s body", enc)
c.mu.Unlock()
c.internal(internalErr)
return
}
w, err := c.store.NewWriter(c.ctx, c.meta)
if err != nil {
c.storeFailed = true
internalErr = fmt.Errorf("recorder: open body store writer: %w", err)
c.mu.Unlock()
c.internal(internalErr)
return
}
c.w = w
if enc == "" || enc == "identity" {
buf := bufio.NewWriterSize(w, 32<<10)
if sr := newBodyStreamRedactor(buf, c.meta.ContentType, c.red); sr != nil {
c.w = &redactingBodyWriter{BodyWriter: w, redactor: sr, buf: buf}
c.storedRedacted = true
}
} else if needsRedaction {
c.w = newDecodingRedactingBodyWriter(w, c.decoder, c.meta.ContentType, c.red, c.limit)
c.storedDecoded = true
c.storedRedacted = true
}
}
n, err := c.w.Write(p[:take])
c.captured += int64(n)
if err != nil {
c.storeFailed = true
internalErr = fmt.Errorf("recorder: write body store: %w", err)
if abortErr := c.abortWriterLocked(); abortErr != nil {
internalErr = errors.Join(internalErr, abortErr)
}
}
c.mu.Unlock()
c.internal(internalErr)
}
type redactingBodyWriter struct {
BodyWriter
redactor io.WriteCloser
buf *bufio.Writer
mu sync.Mutex
finalized bool
}
func (w *redactingBodyWriter) Write(p []byte) (int, error) {
return w.redactor.Write(p)
}
func (w *redactingBodyWriter) Bytes() ([]byte, error) {
if err := w.buf.Flush(); err != nil {
return nil, err
}
return w.BodyWriter.Bytes()
}
func (w *redactingBodyWriter) Commit() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.finalized {
return nil
}
w.finalized = true
redactErr := w.redactor.Close()
flushErr := w.buf.Flush()
if redactErr != nil {
_ = w.BodyWriter.Abort()
return redactErr
}
if flushErr != nil {
_ = w.BodyWriter.Abort()
return flushErr
}
if err := w.BodyWriter.Commit(); err != nil {
_ = w.BodyWriter.Abort()
return err
}
return nil
}
func (w *redactingBodyWriter) Abort() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.finalized {
return nil
}
w.finalized = true
_ = w.redactor.Close()
return w.BodyWriter.Abort()
}
func (c *bodyCapture) internal(err error) {
if err != nil && c.onInternal != nil {
c.onInternal(err)
}
}
// finishComplete marks the stream as fully consumed (EOF observed).
func (c *bodyCapture) finishComplete() {
if c == nil {
return
}
c.mu.Lock()
if c.finished {
c.mu.Unlock()
return
}
c.finished = true
c.complete = true
err := c.commitWriterLocked()
c.mu.Unlock()
c.internal(err)
}
// fail records a read error terminating the stream.
func (c *bodyCapture) fail(err error) {
if c == nil {
return
}
c.mu.Lock()
if c.finished {
c.mu.Unlock()
return
}
c.finished = true
c.readErr = err
closeErr := c.commitWriterLocked()
c.mu.Unlock()
c.internal(closeErr)
}
// closed records the stream being closed; when it had not already finished,
// that is an early close. closeErr is the error returned by the underlying
// Close call, recorded regardless of stream state.
func (c *bodyCapture) closed(closeErr error) {
if c == nil {
return
}
c.mu.Lock()
if closeErr != nil {
c.closeErr = closeErr
}
if c.finished {
c.mu.Unlock()
return
}
c.finished = true
if c.expected >= 0 && c.total == c.expected {
c.complete = true
} else {
c.closedEarly = true
}
err := c.commitWriterLocked()
c.mu.Unlock()
c.internal(err)
}
// setExpected records the announced Content-Length. Values <= 0 mean unknown
// (for client requests a 0 Content-Length with a non-nil body means unknown).
func (c *bodyCapture) setExpected(n int64) {
if c == nil || n <= 0 {
return
}
c.mu.Lock()
defer c.mu.Unlock()
c.expected = n
}
func (c *bodyCapture) commitWriterLocked() error {
if c.w == nil {
return nil
}
if c.storeFailed {
return c.abortWriterLocked()
}
if err := c.w.Commit(); err != nil {
c.storeFailed = true
_ = c.w.Abort()
return fmt.Errorf("recorder: commit body store writer: %w", err)
}
return nil
}
func (c *bodyCapture) abortWriterLocked() error {
if c.w == nil {
return nil
}
if err := c.w.Abort(); err != nil {
return fmt.Errorf("recorder: abort body store writer: %w", err)
}
return nil
}
// reset restarts the capture. Used when the transport replays the request
// body via GetBody (internal retry): only the bytes of the final attempt are
// kept, matching what was actually sent on the successful exchange.
func (c *bodyCapture) reset() {
if c == nil {
return
}
c.mu.Lock()
closeErr := c.abortWriterLocked()
c.w = nil
c.storeFailed = false
c.storedDecoded = false
c.storedRedacted = false
c.finished, c.complete, c.closedEarly, c.truncated = false, false, false, false
c.captured, c.total = 0, 0
c.readErr, c.closeErr = nil, nil
if c.h != nil {
c.h.Reset()
}
c.mu.Unlock()
c.internal(closeErr)
}
// bytes returns the captured content, nil when nothing was stored.
func (c *bodyCapture) bytes() []byte {
if c == nil {
return nil
}
c.mu.Lock()
if c.w == nil || c.storeFailed {
c.mu.Unlock()
return nil
}
b, err := c.w.Bytes()
c.mu.Unlock()
if err != nil {
c.internal(fmt.Errorf("recorder: read body store: %w", err))
return nil
}
return b
}
func (c *bodyCapture) totalBytes() int64 {
if c == nil {
return 0
}
c.mu.Lock()
defer c.mu.Unlock()
return c.total
}
func (c *bodyCapture) isComplete() bool {
if c == nil {
return false
}
c.mu.Lock()
defer c.mu.Unlock()
return c.complete
}
func (c *bodyCapture) isTruncated() bool {
if c == nil {
return false
}
c.mu.Lock()
defer c.mu.Unlock()
return c.truncated
}
func (c *bodyCapture) isStoredDecoded() bool {
if c == nil {
return false
}
c.mu.Lock()
defer c.mu.Unlock()
return c.storedDecoded
}
func (c *bodyCapture) isStoredRedacted() bool {
if c == nil {
return false
}
c.mu.Lock()
defer c.mu.Unlock()
return c.storedRedacted
}
func (c *bodyCapture) readError() error {
if c == nil {
return nil
}
c.mu.Lock()
defer c.mu.Unlock()
return c.readErr
}
// info builds a requestBody/responseBody extension snapshot.
func (c *bodyCapture) info(red *redactor) *BodyInfo {
if c == nil {
return nil
}
c.mu.Lock()
defer c.mu.Unlock()
bi := &BodyInfo{
Present: true,
Complete: c.complete,
ClosedEarly: c.closedEarly,
Truncated: c.truncated,
CapturedBytes: c.captured,
TotalBytes: c.total,
}
if c.h != nil && c.complete {
bi.Hash = hex.EncodeToString(c.h.Sum(nil))
bi.HashAlgorithm = c.hashName
}
if c.readErr != nil {
bi.ReadError = red.redactError(c.readErr.Error())
}
if c.closeErr != nil {
bi.CloseError = red.redactError(c.closeErr.Error())
}
if c.w != nil {
bi.Store = c.w.Ref()
}
return bi
}
// requestBodyRecorder tees the caller-supplied request body while the
// transport streams it. Read and Close semantics of the wrapped body are
// passed through unchanged; only the bytes the transport actually read are
// recorded.
type requestBodyRecorder struct {
rc io.ReadCloser
bc *bodyCapture
ex *exchange
}
func (r *requestBodyRecorder) Read(p []byte) (int, error) {
n, err := r.rc.Read(p)
if n > 0 {
r.bc.observe(p[:n])
r.ex.setState(StateRequestBodyStreaming)
}
switch {
case err == io.EOF:
r.bc.finishComplete()
case err != nil:
r.bc.fail(err)
}
return n, err
}
func (r *requestBodyRecorder) Close() error {
err := r.rc.Close()
r.bc.closed(err)
return err
}
package recorder
import (
"context"
"net/http"
)
type requestCommentContextKey struct{}
// WithRequestComment returns a context carrying a caller-supplied annotation
// for each physical HTTP exchange started under that context. Recorder stores
// the value verbatim in the standard HAR entry.comment field; callers must not
// include secrets or other data that should be redacted.
//
// Redirect requests inherit the comment through their context. A later call
// replaces the earlier value, and an empty comment omits the HAR field.
func WithRequestComment(ctx context.Context, comment string) context.Context {
return context.WithValue(ctx, requestCommentContextKey{}, comment)
}
// RequestWithComment clones req with a context carrying a HAR entry comment.
// It returns nil when req is nil and never mutates the supplied request.
func RequestWithComment(req *http.Request, comment string) *http.Request {
if req == nil {
return nil
}
return req.WithContext(WithRequestComment(req.Context(), comment))
}
func requestCommentFromContext(ctx context.Context) string {
if ctx == nil {
return ""
}
comment, _ := ctx.Value(requestCommentContextKey{}).(string)
return comment
}
package recorder
import (
"context"
"net/http"
)
// RedactionRules adds field selectors and body redactors for one side of an
// HTTP exchange. Names are matched case-insensitively. Configurations merge
// additively, so request-scoped selectors cannot remove Transport selectors.
// A BodyRedactors entry is an explicit trusted override for its MIME type.
type RedactionRules struct {
// Headers lists header names whose values are protected. A cookie is also
// protected when its Cookie or Set-Cookie carrier header is selected.
Headers []string
// QueryParameters protects URL/query-string values, URL-encoded form
// fields, and matching multipart field/file payloads and filenames.
QueryParameters []string
// Cookies lists parsed cookie names whose values are protected.
Cookies []string
// JSONFields recursively protects matching JSON object-field values while
// preserving unaffected source bytes.
JSONFields []string
// XMLElements protects matching XML element subtrees by local name;
// namespace prefixes are ignored and attributes are not selected.
XMLElements []string
// BodyRedactors registers trusted streaming redactors by exact normalized
// base MIME type. A later merged registration wins for the same type.
BodyRedactors map[string]BodyRedactor
}
// RedactionConfig contains common and direction-specific redaction rules.
// Common rules apply to both sides. The same type configures a Transport with
// Config.Redaction and an individual request with WithRequestRedaction.
type RedactionConfig struct {
// Common applies to request and response recordings.
Common RedactionRules
// Request adds rules only to request recordings.
Request RedactionRules
// Response adds rules only to response recordings.
Response RedactionRules
}
type requestRedactionContextKey struct{}
// WithRequestRedaction returns a context carrying additional redaction rules.
// Repeated calls merge rules; for duplicate body-redactor MIME registrations,
// the most recent call wins. Input slices and maps are copied before storage.
func WithRequestRedaction(ctx context.Context, config RedactionConfig) context.Context {
current := requestRedactionFromContext(ctx)
merged := mergeRedactionConfig(current, config)
return context.WithValue(ctx, requestRedactionContextKey{}, merged)
}
// RequestWithRedaction clones req with a context carrying additional redaction
// rules. It returns nil when req is nil and never mutates the supplied request.
func RequestWithRedaction(req *http.Request, config RedactionConfig) *http.Request {
if req == nil {
return nil
}
return req.WithContext(WithRequestRedaction(req.Context(), config))
}
func requestRedactionFromContext(ctx context.Context) RedactionConfig {
if ctx == nil {
return RedactionConfig{}
}
rules, _ := ctx.Value(requestRedactionContextKey{}).(RedactionConfig)
return rules
}
func mergeRedactionConfig(left, right RedactionConfig) RedactionConfig {
return RedactionConfig{
Common: mergeRedactionRules(left.Common, right.Common),
Request: mergeRedactionRules(left.Request, right.Request),
Response: mergeRedactionRules(left.Response, right.Response),
}
}
func effectiveRedactionRules(common, directional RedactionRules) RedactionRules {
return mergeRedactionRules(common, directional)
}
func mergeRedactionRules(left, right RedactionRules) RedactionRules {
return RedactionRules{
Headers: appendCopied(left.Headers, right.Headers),
QueryParameters: appendCopied(left.QueryParameters, right.QueryParameters),
Cookies: appendCopied(left.Cookies, right.Cookies),
JSONFields: appendCopied(left.JSONFields, right.JSONFields),
XMLElements: appendCopied(left.XMLElements, right.XMLElements),
BodyRedactors: mergeBodyRedactors(left.BodyRedactors, right.BodyRedactors),
}
}
func appendCopied(left, right []string) []string {
if len(left) == 0 && len(right) == 0 {
return nil
}
out := make([]string, 0, len(left)+len(right))
out = append(out, left...)
out = append(out, right...)
return out
}
func mergeBodyRedactors(left, right map[string]BodyRedactor) map[string]BodyRedactor {
if len(left) == 0 && len(right) == 0 {
return nil
}
out := make(map[string]BodyRedactor, len(left)+len(right))
for mediaType, redactor := range left {
if normalized := baseMimeType(mediaType); normalized != "" && redactor != nil {
out[normalized] = redactor
}
}
for mediaType, redactor := range right {
if normalized := baseMimeType(mediaType); normalized != "" && redactor != nil {
out[normalized] = redactor
}
}
return out
}
package recorder
import "io"
// responseBodyRecorder wraps resp.Body. It records exactly the bytes the
// caller reads (no double read, no extra buffering beyond the capture
// writer), passes bytes, EOF and errors through unchanged, and finalizes the
// exchange record when the stream ends:
//
// - Read returning io.EOF -> record complete
// - Read returning an error -> record failed (phase read_response_body)
// - Close before EOF -> record closed early
//
// The finalize path runs at most once (guarded by the exchange), so a Close
// following EOF only adds the Close error, if any.
type responseBodyRecorder struct {
rc io.ReadCloser
bc *bodyCapture
ex *exchange
}
func (r *responseBodyRecorder) Read(p []byte) (int, error) {
n, err := r.rc.Read(p)
if n > 0 {
r.bc.observe(p[:n])
r.ex.setState(StateResponseBodyStreaming)
}
switch {
case err == io.EOF:
r.bc.finishComplete()
r.ex.finalizeComplete()
case err != nil:
r.bc.fail(err)
r.ex.finalizeBodyReadError(err)
}
return n, err
}
func (r *responseBodyRecorder) Close() error {
err := r.rc.Close()
r.bc.closed(err)
r.ex.finalizeClosed()
return err
}
package recorder
import (
"context"
"crypto/rand"
"encoding/binary"
"errors"
"fmt"
"math"
"mime"
"net/http"
"strings"
"sync/atomic"
)
// HeadSamplingDecision selects how much work the Transport performs for an
// exchange. The zero value preserves the configured full-recording behavior.
type HeadSamplingDecision uint8
const (
// HeadSampleFull applies the configured capture, hashing, trace, and
// redaction behavior.
HeadSampleFull HeadSamplingDecision = iota
// HeadSampleMetadataOnly records exchange lifecycle and core metadata
// without body content, hashing, certificates, or raw trace events.
HeadSampleMetadataOnly
// HeadSampleDrop bypasses recorder instrumentation for this exchange.
HeadSampleDrop
)
// HeadSamplingMeta is the bounded, non-secret request metadata available
// before exchange instrumentation is constructed. MIMEType excludes parameters.
type HeadSamplingMeta struct {
Method string
Scheme string
Host string
Path string
MIMEType string
ContentLength int64
SamplingKey string
TraceID string
RedirectIndex int
}
// HeadSamplingPolicy decides whether an exchange is fully recorded, reduced
// to metadata, or passed directly to the wrapped transport. Implementations
// must be fast, side-effect free, and safe for concurrent use.
type HeadSamplingPolicy func(context.Context, HeadSamplingMeta) HeadSamplingDecision
// OnHeadSamplingDecision observes the effective head-sampling decision before
// exchange instrumentation or the wrapped transport runs. It receives only
// HeadSamplingMeta and does not report an HTTP outcome. Implementations must
// be fast, concurrency-safe, and free of side effects that affect the request.
type OnHeadSamplingDecision func(context.Context, HeadSamplingMeta, HeadSamplingDecision)
// RetentionDecision controls whether a finalized entry reaches the Recorder.
type RetentionDecision uint8
const (
// RetainEntry delivers the finalized entry to the configured Recorder.
RetainEntry RetentionDecision = iota
// DiscardEntry omits the finalized entry and releases its external assets.
DiscardEntry
)
// RetentionPolicy runs before OnEntryCompleted and Recorder.Record. Capture
// cost has already been paid. Implementations must be concurrency-safe.
type RetentionPolicy func(context.Context, *Entry) RetentionDecision
// EntryDisposition reports the effective retention outcome for a finalized
// entry. It is separate from Entry.State, which describes the HTTP exchange.
type EntryDisposition uint8
const (
// EntryDispositionKeep means the entry passed retention and will be offered
// to Recorder after OnEntryCompleted returns. It does not guarantee that a
// Recorder exists or that a sink will persist the entry successfully.
EntryDispositionKeep EntryDisposition = iota
// EntryDispositionDiscard means retention selected discard, referenced
// external assets were released successfully, and Recorder will not receive
// the entry.
EntryDispositionDiscard
)
// SamplingStats is an atomic snapshot of head and tail decisions.
type SamplingStats struct {
HeadFull uint64
HeadMetadataOnly uint64
HeadDropped uint64
HeadPanics uint64
HeadInvalidDecisions uint64
Retained uint64
Discarded uint64
RetentionPanics uint64
RetentionInvalidDecisions uint64
AssetReleaseFailures uint64
}
type samplingCounters struct {
headFull atomic.Uint64
headMetadataOnly atomic.Uint64
headDropped atomic.Uint64
headPanics atomic.Uint64
headInvalidDecisions atomic.Uint64
retained atomic.Uint64
discarded atomic.Uint64
retentionPanics atomic.Uint64
retentionInvalidDecisions atomic.Uint64
assetReleaseFailures atomic.Uint64
}
type samplingKeyContextKey struct{}
// WithSamplingKey returns a context carrying a stable application sampling
// key. Rate policies prefer it over TraceID, preserving decisions across
// related requests even when no recorder trace context is installed.
func WithSamplingKey(ctx context.Context, key string) context.Context {
return context.WithValue(ctx, samplingKeyContextKey{}, key)
}
// SamplingKeyFromContext returns the key installed by WithSamplingKey.
func SamplingKeyFromContext(ctx context.Context) (string, bool) {
key, ok := ctx.Value(samplingKeyContextKey{}).(string)
return key, ok && key != ""
}
type rateHeadSampler struct {
threshold uint64
always bool
sampled HeadSamplingDecision
unsampled HeadSamplingDecision
}
// NewRateHeadSampler creates a deterministic rate policy. SamplingKey is used
// first, then TraceID. Without either, each physical exchange uses crypto/rand
// and redirect-chain consistency is not guaranteed.
func NewRateHeadSampler(fraction float64, sampled, unsampled HeadSamplingDecision) (HeadSamplingPolicy, error) {
if math.IsNaN(fraction) || fraction < 0 || fraction > 1 {
return nil, errors.New("recorder: head sampling fraction must be between 0 and 1")
}
if !validHeadSamplingDecision(sampled) || !validHeadSamplingDecision(unsampled) {
return nil, errors.New("recorder: invalid rate head sampling decision")
}
var threshold uint64
if fraction < 1 {
threshold = uint64(math.Ldexp(fraction, 64))
}
sampler := &rateHeadSampler{threshold: threshold, always: fraction == 1, sampled: sampled, unsampled: unsampled}
return sampler.sampleHead, nil
}
func (s *rateHeadSampler) sampleHead(_ context.Context, meta HeadSamplingMeta) HeadSamplingDecision {
key := meta.SamplingKey
if key == "" {
key = meta.TraceID
}
var value uint64
if key == "" {
var random [8]byte
if _, err := rand.Read(random[:]); err != nil {
return s.sampled
}
value = binary.BigEndian.Uint64(random[:])
} else {
value = stableSamplingHash(key)
}
if s.always || value < s.threshold {
return s.sampled
}
return s.unsampled
}
func stableSamplingHash(value string) uint64 {
const (
offset = uint64(14695981039346656037)
prime = uint64(1099511628211)
)
hash := offset
for i := range len(value) {
hash ^= uint64(value[i])
hash *= prime
}
// FNV-1a leaves its high bits correlated for common prefixed/sequential
// keys. Rate selection compares against the full-width threshold, so apply
// a MurmurHash3 finalizer to avalanche those correlations first.
hash ^= hash >> 33
hash *= 0xff51afd7ed558ccd
hash ^= hash >> 33
hash *= 0xc4ceb9fe1a85ec53
hash ^= hash >> 33
return hash
}
type exchangeIdentity struct {
traceID string
redirectIndex int
hasTraceState bool
samplingKey string
}
func resolveExchangeIdentity(ctx context.Context) exchangeIdentity {
identity := exchangeIdentity{}
if key, ok := SamplingKeyFromContext(ctx); ok {
identity.samplingKey = key
}
if state := traceStateFromContext(ctx); state != nil {
identity.traceID = state.id
identity.redirectIndex = int(state.seq.Add(1) - 1)
identity.hasTraceState = true
}
return identity
}
func headSamplingMeta(req *http.Request, identity exchangeIdentity) HeadSamplingMeta {
meta := HeadSamplingMeta{
Method: req.Method,
ContentLength: req.ContentLength,
SamplingKey: identity.samplingKey,
TraceID: identity.traceID,
RedirectIndex: identity.redirectIndex,
}
if req.URL != nil {
meta.Scheme = req.URL.Scheme
meta.Host = req.URL.Host
meta.Path = req.URL.EscapedPath()
if meta.Path == "" {
meta.Path = "/"
}
}
contentType := req.Header.Get("Content-Type")
if contentType != "" {
if mimeType, _, err := mime.ParseMediaType(contentType); err == nil {
meta.MIMEType = strings.ToLower(mimeType)
}
}
return meta
}
func validHeadSamplingDecision(decision HeadSamplingDecision) bool {
return decision == HeadSampleFull || decision == HeadSampleMetadataOnly || decision == HeadSampleDrop
}
func entryHasStoreReferences(entry *Entry) bool {
return entry != nil && entry.Recorder != nil &&
((entry.Recorder.RequestBody != nil && entry.Recorder.RequestBody.Store != "") ||
(entry.Recorder.ResponseBody != nil && entry.Recorder.ResponseBody.Store != ""))
}
func (t *Transport) decideHeadSampling(ctx context.Context, meta HeadSamplingMeta) (decision HeadSamplingDecision) {
policy := t.config.HeadSamplingPolicy
if policy == nil {
t.sampling.headFull.Add(1)
return HeadSampleFull
}
defer func() {
if recovered := recover(); recovered != nil {
t.sampling.headPanics.Add(1)
t.sampling.headFull.Add(1)
t.internalError(fmt.Errorf("recorder: panic in HeadSamplingPolicy: %v", recovered))
decision = HeadSampleFull
}
}()
decision = policy(ctx, meta)
if !validHeadSamplingDecision(decision) {
t.sampling.headInvalidDecisions.Add(1)
t.sampling.headFull.Add(1)
t.internalError(fmt.Errorf("recorder: invalid head sampling decision %d", decision))
return HeadSampleFull
}
switch decision {
case HeadSampleFull:
t.sampling.headFull.Add(1)
case HeadSampleMetadataOnly:
t.sampling.headMetadataOnly.Add(1)
case HeadSampleDrop:
t.sampling.headDropped.Add(1)
}
return decision
}
func (t *Transport) callOnHeadSamplingDecision(
ctx context.Context,
meta HeadSamplingMeta,
decision HeadSamplingDecision,
) {
if t.config.OnHeadSamplingDecision == nil {
return
}
defer func() {
if recovered := recover(); recovered != nil {
t.internalError(fmt.Errorf("recorder: panic in OnHeadSamplingDecision: %v", recovered))
}
}()
t.config.OnHeadSamplingDecision(ctx, meta, decision)
}
func (t *Transport) decideRetention(ctx context.Context, entry *Entry) (decision RetentionDecision) {
policy := t.config.RetentionPolicy
if policy == nil {
t.sampling.retained.Add(1)
return RetainEntry
}
defer func() {
if recovered := recover(); recovered != nil {
t.sampling.retentionPanics.Add(1)
t.sampling.retained.Add(1)
t.internalError(fmt.Errorf("recorder: panic in RetentionPolicy: %v", recovered))
decision = RetainEntry
}
}()
decision = policy(ctx, entry)
if decision != RetainEntry && decision != DiscardEntry {
t.sampling.retentionInvalidDecisions.Add(1)
t.sampling.retained.Add(1)
t.internalError(fmt.Errorf("recorder: invalid retention decision %d", decision))
return RetainEntry
}
if decision == RetainEntry {
t.sampling.retained.Add(1)
}
return decision
}
// SamplingStats returns a concurrency-safe point-in-time sampling snapshot.
func (t *Transport) SamplingStats() SamplingStats {
return SamplingStats{
HeadFull: t.sampling.headFull.Load(),
HeadMetadataOnly: t.sampling.headMetadataOnly.Load(),
HeadDropped: t.sampling.headDropped.Load(),
HeadPanics: t.sampling.headPanics.Load(),
HeadInvalidDecisions: t.sampling.headInvalidDecisions.Load(),
Retained: t.sampling.retained.Load(),
Discarded: t.sampling.discarded.Load(),
RetentionPanics: t.sampling.retentionPanics.Load(),
RetentionInvalidDecisions: t.sampling.retentionInvalidDecisions.Load(),
AssetReleaseFailures: t.sampling.assetReleaseFailures.Load(),
}
}
package recorder
import (
"context"
"crypto/aes"
"crypto/cipher"
"crypto/hmac"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"errors"
"fmt"
"hash"
"io"
"strings"
"sync"
"unicode/utf8"
)
const (
defaultMaxProtectedValueBytes = 64 << 10
maxProtectedValueBytes = 16 << 20
encryptedValuePrefix = "REC-ENC-v1."
tokenizedValuePrefix = "REC-TOK-v1."
)
// ProtectionMode controls how configured sensitive values are represented in
// recorded copies. The zero value preserves the traditional [REDACTED]
// behavior.
type ProtectionMode string
const (
// ProtectionRedact replaces every selected value with [REDACTED].
ProtectionRedact ProtectionMode = "redact"
// ProtectionEncrypt emits a reversible, key-ID-bearing AES-256-GCM token.
ProtectionEncrypt ProtectionMode = "encrypt"
// ProtectionTokenize emits a deterministic, non-reversible HMAC token.
ProtectionTokenize ProtectionMode = "tokenize"
)
// ProtectionKey is the active key used for a recorded sensitive value. ID is
// embedded in protected tokens to support rotation; it must not itself be
// sensitive. Encryption requires exactly 32 key bytes. Tokenization accepts
// 32 or more bytes.
type ProtectionKey struct {
ID string
Key []byte
}
// ProtectionKeyProvider returns the active key for the request context and
// requested mode. Recorder resolves each mode at most once per exchange, and
// the request and response share that immutable key snapshot. The function may
// be called concurrently across exchanges and must not return key material
// that callers mutate.
type ProtectionKeyProvider func(context.Context, ProtectionMode) (ProtectionKey, error)
// ProtectionKeyResolver resolves historical key material by the non-secret ID
// embedded in a protected token. The function may be called concurrently
// and must not return key material that callers mutate.
type ProtectionKeyResolver func(keyID string) (ProtectionKey, error)
// SensitiveValueProtection configures the representation of every value
// selected by built-in or custom body redactors. MaxValueBytes bounds a single
// value buffered for encryption. Tokenization streams values through HMAC
// without retaining them. Values <= 0 select the safe 64 KiB default; values
// above 16 MiB are clamped. Failures and oversized values are replaced with
// [REDACTED].
type SensitiveValueProtection struct {
// Mode selects redact, encrypt, or tokenize. The zero value redacts.
Mode ProtectionMode
// KeyProvider supplies encryption or tokenization key material.
KeyProvider ProtectionKeyProvider
// MaxValueBytes bounds one value buffered for encryption.
MaxValueBytes int
}
type sensitiveValueProtector struct {
config SensitiveValueProtection
ctx context.Context
rand func([]byte) (int, error)
keys *protectionKeyCache
}
type protectionKeyCache struct {
mu sync.Mutex
resolved map[ProtectionMode]resolvedProtectionKey
}
type resolvedProtectionKey struct {
key ProtectionKey
err error
aead cipher.AEAD
aad []byte
aeadReady bool
}
type bodyValueProtector struct {
mu sync.Mutex
protector *sensitiveValueProtector
report ProtectionCounts
replacements int64
firstErr error
failures int64
}
func newBodyValueProtector(protector *sensitiveValueProtector) *bodyValueProtector {
if protector == nil {
protector = newSensitiveValueProtector(SensitiveValueProtection{})
}
return &bodyValueProtector{protector: protector}
}
func (p *bodyValueProtector) NewValue() BodyValue {
value := &protectedValueBuffer{}
value.reset(p)
return value
}
func (p *bodyValueProtector) protectionReport() (ProtectionCounts, int64) {
p.mu.Lock()
defer p.mu.Unlock()
return cloneProtectionCounts(p.report), p.replacements
}
func (p *bodyValueProtector) protectionFailure() (error, int64) {
p.mu.Lock()
defer p.mu.Unlock()
return p.firstErr, p.failures
}
func (p *bodyValueProtector) record(mode ProtectionMode, reason string, err error) {
p.mu.Lock()
defer p.mu.Unlock()
p.replacements++
addProtectionOutcome(&p.report, mode, reason)
if err != nil {
if p.firstErr == nil {
p.firstErr = err
}
p.failures++
}
}
// protectedValueBuffer bounds the only plaintext retained by streaming
// redactors. Once the limit is crossed it forgets the accumulated value and
// remembers only that protection must fail closed.
type protectedValueBuffer struct {
session *bodyValueProtector
value []byte
tooLarge bool
emitted bool
tokenMAC hash.Hash
tokenID string
tokenBuf [256]byte
tokenLen int
tokenSum [sha256.Size]byte
cryptoBuf []byte
encoded []byte
tokenFail bool
tokenErr error
finished bool
result []byte
}
func (b *protectedValueBuffer) reset(session *bodyValueProtector) {
b.session = session
b.value = b.value[:0]
b.tooLarge = false
b.emitted = false
b.tokenID = ""
b.clearTokenBuffer()
b.tokenFail = false
b.tokenErr = nil
b.finished = false
b.result = nil
if session.protector.config.Mode == ProtectionTokenize {
key, err := session.protector.key(ProtectionTokenize)
if err != nil || len(key.Key) < 32 {
b.tokenFail = true
if err == nil {
err = errors.New("recorder: tokenization key must contain at least 32 bytes")
}
b.tokenErr = err
return
}
if b.tokenMAC == nil {
b.tokenMAC = hmac.New(sha256.New, key.Key)
} else {
b.tokenMAC.Reset()
}
b.tokenID = key.ID
}
}
func (b *protectedValueBuffer) Write(p []byte) (int, error) {
if b.finished {
return 0, errors.New("recorder: write after protected value finish")
}
b.appendBytes(p)
return len(p), nil
}
func (b *protectedValueBuffer) FinishTo(dst io.Writer) error {
result := b.protectedBytes()
if len(result) == 0 {
return nil
}
n, err := dst.Write(result)
if err == nil && n != len(result) {
return io.ErrShortWrite
}
return err
}
func (b *protectedValueBuffer) appendByte(value byte) {
if b.tooLarge || b.emitted {
return
}
if b.session.protector.config.Mode == ProtectionRedact {
return
}
if b.session.protector.config.Mode == ProtectionTokenize {
if b.tokenMAC != nil {
b.tokenBuf[b.tokenLen] = value
b.tokenLen++
if b.tokenLen == len(b.tokenBuf) {
b.flushTokenBuffer()
}
}
return
}
if len(b.value) == b.session.protector.maxValueBytes() {
b.clearValue()
b.tooLarge = true
return
}
b.value = append(b.value, value)
}
func (b *protectedValueBuffer) appendBytes(p []byte) {
if b.tooLarge || b.emitted {
return
}
if b.session.protector.config.Mode == ProtectionRedact {
return
}
if b.session.protector.config.Mode == ProtectionTokenize {
if b.tokenMAC != nil {
for len(p) > 0 {
n := copy(b.tokenBuf[b.tokenLen:], p)
b.tokenLen += n
p = p[n:]
if b.tokenLen == len(b.tokenBuf) {
b.flushTokenBuffer()
}
}
}
return
}
if len(b.value)+len(p) > b.session.protector.maxValueBytes() {
for i := range b.value {
b.value[i] = 0
}
b.value = b.value[:0]
b.tooLarge = true
return
}
b.value = append(b.value, p...)
}
func (b *protectedValueBuffer) protectedBytes() []byte {
if !b.finished {
b.finished = true
b.result, _, _ = b.finish()
}
return b.result
}
func (b *protectedValueBuffer) finish() ([]byte, ProtectionMode, string) {
if b.emitted {
return nil, ProtectionRedact, ""
}
if b.session.protector.config.Mode == ProtectionTokenize {
if b.tokenFail || b.tokenMAC == nil {
b.clearTokenBuffer()
b.session.record(ProtectionRedact, "tokenization_failed", b.tokenErr)
return b.redactedBytes(), ProtectionRedact, "tokenization_failed"
}
b.flushTokenBuffer()
sum := b.tokenMAC.Sum(b.tokenSum[:0])
value := b.encodeProtectedToken(tokenizedValuePrefix, b.tokenID, sum)
for i := range sum {
sum[i] = 0
}
b.session.record(ProtectionTokenize, "", nil)
return value, ProtectionTokenize, ""
}
if b.tooLarge {
b.session.record(ProtectionRedact, "value_too_large", nil)
return b.redactedBytes(), ProtectionRedact, "value_too_large"
}
if b.session.protector.config.Mode == ProtectionEncrypt {
value, err := b.encrypt()
b.clearValue()
if err != nil {
b.session.record(ProtectionRedact, "encryption_failed", err)
return b.redactedBytes(), ProtectionRedact, "encryption_failed"
}
b.session.record(ProtectionEncrypt, "", nil)
return value, ProtectionEncrypt, ""
}
b.clearValue()
b.session.record(ProtectionRedact, "", nil)
return b.redactedBytes(), ProtectionRedact, ""
}
func (b *protectedValueBuffer) redactedBytes() []byte {
b.encoded = append(b.encoded[:0], redactedValue...)
return b.encoded
}
func (b *protectedValueBuffer) encrypt() ([]byte, error) {
key, aead, aad, err := b.session.protector.encryption()
if err != nil {
return nil, err
}
nonceSize := aead.NonceSize()
required := nonceSize + len(b.value) + aead.Overhead()
if cap(b.cryptoBuf) < required {
b.cryptoBuf = make([]byte, nonceSize, required)
} else {
b.cryptoBuf = b.cryptoBuf[:nonceSize]
}
nonce := b.cryptoBuf[:nonceSize]
if n, err := b.session.protector.rand(nonce); err != nil {
return nil, err
} else if n != len(nonce) {
return nil, io.ErrUnexpectedEOF
}
b.cryptoBuf = aead.Seal(b.cryptoBuf, nonce, b.value, aad)
return b.encodeProtectedToken(encryptedValuePrefix, key.ID, b.cryptoBuf), nil
}
func (b *protectedValueBuffer) encodeProtectedToken(prefix, keyID string, payload []byte) []byte {
b.encoded = appendProtectedToken(b.encoded[:0], prefix, keyID, payload)
return b.encoded
}
func (b *protectedValueBuffer) flushTokenBuffer() {
if b.tokenLen == 0 {
return
}
_, _ = b.tokenMAC.Write(b.tokenBuf[:b.tokenLen])
b.clearTokenBuffer()
}
func (b *protectedValueBuffer) clearTokenBuffer() {
for i := 0; i < b.tokenLen; i++ {
b.tokenBuf[i] = 0
}
b.tokenLen = 0
}
func (b *protectedValueBuffer) clearValue() {
for i := range b.value {
b.value[i] = 0
}
b.value = b.value[:0]
}
func (b *protectedValueBuffer) redactImmediately() bool {
if b.session.protector.config.Mode == ProtectionRedact {
b.emitted = true
b.session.record(ProtectionRedact, "", nil)
return true
}
return false
}
func addProtectionOutcome(report *ProtectionCounts, mode ProtectionMode, reason string) {
switch mode {
case ProtectionEncrypt:
report.Encrypted++
case ProtectionTokenize:
report.Tokenized++
default:
report.Redacted++
}
if reason != "" {
if report.Fallbacks == nil {
report.Fallbacks = make(map[string]int64)
}
report.Fallbacks[reason]++
}
}
func cloneProtectionCounts(in ProtectionCounts) ProtectionCounts {
out := in
if len(in.Fallbacks) > 0 {
out.Fallbacks = make(map[string]int64, len(in.Fallbacks))
for reason, count := range in.Fallbacks {
out.Fallbacks[reason] = count
}
}
return out
}
func protectionCountsEmpty(in ProtectionCounts) bool {
return in.Redacted == 0 && in.Encrypted == 0 && in.Tokenized == 0 && len(in.Fallbacks) == 0
}
func newSensitiveValueProtector(config SensitiveValueProtection) *sensitiveValueProtector {
if config.Mode == "" {
config.Mode = ProtectionRedact
}
if config.MaxValueBytes <= 0 {
config.MaxValueBytes = defaultMaxProtectedValueBytes
} else if config.MaxValueBytes > maxProtectedValueBytes {
config.MaxValueBytes = maxProtectedValueBytes
}
return &sensitiveValueProtector{
config: config,
ctx: context.Background(),
rand: rand.Read,
keys: &protectionKeyCache{},
}
}
func (p *sensitiveValueProtector) withContext(ctx context.Context, keys *protectionKeyCache) *sensitiveValueProtector {
clone := *p
clone.ctx = ctx
clone.keys = keys
if clone.keys == nil {
clone.keys = &protectionKeyCache{}
}
return &clone
}
func (p *sensitiveValueProtector) maxValueBytes() int { return p.config.MaxValueBytes }
func (p *sensitiveValueProtector) protect(value []byte) (string, ProtectionMode, string) {
protected, mode, reason, _ := p.protectWithError(value)
return protected, mode, reason
}
func (p *sensitiveValueProtector) protectWithError(value []byte) (string, ProtectionMode, string, error) {
if p.config.Mode == ProtectionEncrypt && len(value) > p.maxValueBytes() {
return redactedValue, ProtectionRedact, "value_too_large", nil
}
switch p.config.Mode {
case ProtectionEncrypt:
out, err := p.encrypt(value)
if err != nil {
return redactedValue, ProtectionRedact, "encryption_failed", err
}
return out, ProtectionEncrypt, "", nil
case ProtectionTokenize:
out, err := p.tokenize(value)
if err != nil {
return redactedValue, ProtectionRedact, "tokenization_failed", err
}
return out, ProtectionTokenize, "", nil
default:
return redactedValue, ProtectionRedact, "", nil
}
}
func (p *sensitiveValueProtector) key(mode ProtectionMode) (ProtectionKey, error) {
p.keys.mu.Lock()
defer p.keys.mu.Unlock()
resolved := p.keyLocked(mode)
return resolved.key, resolved.err
}
func (p *sensitiveValueProtector) keyLocked(mode ProtectionMode) resolvedProtectionKey {
if resolved, ok := p.keys.resolved[mode]; ok {
return resolved
}
key, err := p.resolveKey(mode)
resolved := resolvedProtectionKey{key: key, err: err}
if p.keys.resolved == nil {
p.keys.resolved = make(map[ProtectionMode]resolvedProtectionKey, 1)
}
p.keys.resolved[mode] = resolved
return resolved
}
func (p *sensitiveValueProtector) encryption() (ProtectionKey, cipher.AEAD, []byte, error) {
p.keys.mu.Lock()
defer p.keys.mu.Unlock()
resolved := p.keyLocked(ProtectionEncrypt)
if resolved.err != nil {
return ProtectionKey{}, nil, nil, resolved.err
}
if resolved.aeadReady {
return resolved.key, resolved.aead, resolved.aad, resolved.err
}
if len(resolved.key.Key) != 32 {
resolved.err = errors.New("recorder: encryption key must contain exactly 32 bytes")
} else {
block, err := aes.NewCipher(resolved.key.Key)
if err != nil {
resolved.err = err
} else {
resolved.aead, resolved.err = cipher.NewGCM(block)
}
}
resolved.aad = []byte(resolved.key.ID)
resolved.aeadReady = true
p.keys.resolved[ProtectionEncrypt] = resolved
return resolved.key, resolved.aead, resolved.aad, resolved.err
}
func (p *sensitiveValueProtector) resolveKey(mode ProtectionMode) (ProtectionKey, error) {
if p.config.KeyProvider == nil {
return ProtectionKey{}, errors.New("recorder: sensitive value key provider is nil")
}
k, err := p.config.KeyProvider(p.ctx, mode)
if err != nil {
return ProtectionKey{}, err
}
if k.ID == "" || len(k.ID) > 256 || !utf8.ValidString(k.ID) {
return ProtectionKey{}, errors.New("recorder: protection key ID must be valid UTF-8 between 1 and 256 bytes")
}
return k, nil
}
func (p *sensitiveValueProtector) encrypt(value []byte) (string, error) {
k, gcm, aad, err := p.encryption()
if err != nil {
return "", err
}
nonce := make([]byte, gcm.NonceSize())
if n, err := p.rand(nonce); err != nil {
return "", err
} else if n != len(nonce) {
return "", io.ErrUnexpectedEOF
}
payload := gcm.Seal(nonce, nonce, value, aad)
return protectedToken(encryptedValuePrefix, k.ID, payload), nil
}
func (p *sensitiveValueProtector) tokenize(value []byte) (string, error) {
k, err := p.key(ProtectionTokenize)
if err != nil {
return "", err
}
if len(k.Key) < 32 {
return "", errors.New("recorder: tokenization key must contain at least 32 bytes")
}
mac := hmac.New(sha256.New, k.Key)
_, _ = mac.Write(value)
return protectedToken(tokenizedValuePrefix, k.ID, mac.Sum(nil)), nil
}
func protectedToken(prefix, keyID string, payload []byte) string {
return string(appendProtectedToken(nil, prefix, keyID, payload))
}
func appendProtectedToken(dst []byte, prefix, keyID string, payload []byte) []byte {
keyIDBytes := []byte(keyID)
keyIDLen := base64.RawURLEncoding.EncodedLen(len(keyIDBytes))
payloadLen := base64.RawURLEncoding.EncodedLen(len(payload))
required := len(prefix) + keyIDLen + 1 + payloadLen
if cap(dst) < required {
dst = make([]byte, required)
} else {
dst = dst[:required]
}
copy(dst, prefix)
base64.RawURLEncoding.Encode(dst[len(prefix):len(prefix)+keyIDLen], keyIDBytes)
dst[len(prefix)+keyIDLen] = '.'
base64.RawURLEncoding.Encode(dst[len(prefix)+keyIDLen+1:], payload)
return dst
}
// ProtectedTokenKeyID reports the non-secret key ID embedded in a supported
// protected token without decrypting or verifying its payload.
func ProtectedTokenKeyID(token string) (string, error) {
switch {
case strings.HasPrefix(token, encryptedValuePrefix):
keyID, payload, err := splitProtectedToken(token, encryptedValuePrefix)
if err == nil {
err = validateProtectedTokenPayload(payload)
}
return keyID, err
case strings.HasPrefix(token, tokenizedValuePrefix):
keyID, payload, err := splitProtectedToken(token, tokenizedValuePrefix)
if err == nil {
err = validateProtectedTokenPayload(payload)
}
return keyID, err
default:
return "", errors.New("recorder: unsupported protected token")
}
}
// DecryptProtectedValue decrypts a REC-ENC-v1 token with the matching key.
// It is useful for trusted tooling; applications should avoid persisting the
// returned plaintext.
func DecryptProtectedValue(token string, key ProtectionKey) ([]byte, error) {
kid, payload, err := parseProtectedToken(token, encryptedValuePrefix)
if err != nil {
return nil, err
}
if kid != key.ID || len(key.Key) != 32 {
return nil, errors.New("recorder: protection key does not match token")
}
block, err := aes.NewCipher(key.Key)
if err != nil {
return nil, err
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return nil, err
}
if len(payload) < gcm.NonceSize()+gcm.Overhead() {
return nil, errors.New("recorder: encrypted token payload is too short")
}
return gcm.Open(nil, payload[:gcm.NonceSize()], payload[gcm.NonceSize():], []byte(kid))
}
// DecryptProtectedValueWith resolves the key ID embedded in token and decrypts
// it. It is intended for trusted archive tooling spanning key rotations.
func DecryptProtectedValueWith(token string, resolver ProtectionKeyResolver) ([]byte, error) {
if resolver == nil {
return nil, errors.New("recorder: protection key resolver is nil")
}
keyID, payload, err := splitProtectedToken(token, encryptedValuePrefix)
if err != nil {
return nil, err
}
if err := validateProtectedTokenPayload(payload); err != nil {
return nil, err
}
key, err := resolver(keyID)
if err != nil {
return nil, fmt.Errorf("recorder: resolve protection key %q: %w", keyID, err)
}
return DecryptProtectedValue(token, key)
}
// VerifyProtectedToken reports whether value produced a REC-TOK-v1 token.
func VerifyProtectedToken(token string, value []byte, key ProtectionKey) (bool, error) {
kid, payload, err := parseProtectedToken(token, tokenizedValuePrefix)
if err != nil {
return false, err
}
if kid != key.ID || len(key.Key) < 32 {
return false, errors.New("recorder: protection key does not match token")
}
mac := hmac.New(sha256.New, key.Key)
_, _ = mac.Write(value)
return hmac.Equal(payload, mac.Sum(nil)), nil
}
// VerifyProtectedTokenWith resolves the key ID embedded in token and verifies
// value. It is intended for trusted archive tooling spanning key rotations.
func VerifyProtectedTokenWith(token string, value []byte, resolver ProtectionKeyResolver) (bool, error) {
if resolver == nil {
return false, errors.New("recorder: protection key resolver is nil")
}
keyID, payload, err := splitProtectedToken(token, tokenizedValuePrefix)
if err != nil {
return false, err
}
if err := validateProtectedTokenPayload(payload); err != nil {
return false, err
}
key, err := resolver(keyID)
if err != nil {
return false, fmt.Errorf("recorder: resolve protection key %q: %w", keyID, err)
}
return VerifyProtectedToken(token, value, key)
}
func parseProtectedToken(token, prefix string) (string, []byte, error) {
keyID, encodedPayload, err := splitProtectedToken(token, prefix)
if err != nil {
return "", nil, err
}
payload, err := base64.RawURLEncoding.DecodeString(encodedPayload)
if err != nil {
return "", nil, fmt.Errorf("recorder: malformed protected token payload: %w", err)
}
return keyID, payload, nil
}
func splitProtectedToken(token, prefix string) (string, string, error) {
rest, ok := strings.CutPrefix(token, prefix)
if !ok {
return "", "", errors.New("recorder: unsupported protected token")
}
encodedID, encodedPayload, ok := strings.Cut(rest, ".")
if !ok || encodedID == "" || encodedPayload == "" || strings.Contains(encodedPayload, ".") {
return "", "", errors.New("recorder: malformed protected token")
}
idBytes, err := base64.RawURLEncoding.DecodeString(encodedID)
if err != nil || len(idBytes) == 0 || len(idBytes) > 256 || !utf8.Valid(idBytes) {
return "", "", errors.New("recorder: malformed protected token key ID")
}
return string(idBytes), encodedPayload, nil
}
func validateProtectedTokenPayload(encodedPayload string) error {
decoder := base64.NewDecoder(base64.RawURLEncoding.Strict(), strings.NewReader(encodedPayload))
if _, err := io.Copy(io.Discard, decoder); err != nil {
return fmt.Errorf("recorder: malformed protected token payload: %w", err)
}
return nil
}
package recorder
import (
"crypto/tls"
"net/http"
"net/http/httptrace"
"net/textproto"
"strconv"
"strings"
"sync"
"time"
)
// TraceEvent is one raw httptrace event, stored under _recorder.trace
// when Config.CaptureRawTrace is enabled.
type TraceEvent struct {
Name string `json:"name"`
Time string `json:"time"`
Detail string `json:"detail,omitempty"`
}
// informational1xx is one observed 1xx interim response.
type informational1xx struct {
code int
header http.Header
}
// putIdleResult records whether the connection went back to the idle pool.
type putIdleResult struct {
returned bool
err error
}
// traceView is an immutable snapshot of everything the collector observed.
// Zero time values mean "event did not happen".
type traceView struct {
getConn time.Time
dnsStart time.Time
dnsDone time.Time
connectStart time.Time
connectDone time.Time
tlsStart time.Time
tlsDone time.Time
gotConn time.Time
wroteHeaders time.Time
wroteRequest time.Time
firstByte time.Time
wait100 time.Time
got100 time.Time
dnsAddrs []string
dnsCoalesced bool
getConnAddr string
dnsErr error
connectErr error
tlsErr error
wroteReqErr error
reused bool
wasIdle bool
idleTime time.Duration
network string
localAddr string
remoteAddr string
tlsState *tls.ConnectionState
// wroteHeaderFields are the header fields the transport actually wrote
// to the wire, in wire order (HTTP/2 includes pseudo-headers such as
// ":authority"). Values are raw; redaction happens at entry-build time.
wroteHeaderFields []NameValuePair
info1xx []informational1xx
putIdle *putIdleResult
raw []TraceEvent
}
// max1xxRecorded bounds how many interim responses one exchange stores.
const max1xxRecorded = 16
// traceCollector accumulates httptrace events for one exchange. All callback
// paths lock mu; events may arrive from transport-internal goroutines, in any
// order, and more than once (e.g. Happy-Eyeballs parallel dials) — the
// collector keeps the first start and the last successful completion of each
// step and never treats a missing event as an error.
type traceCollector struct {
mu sync.Mutex
now func() time.Time
captureRaw bool
notify func(state string)
v traceView
}
func newTraceCollector(captureRaw bool) *traceCollector {
return &traceCollector{now: time.Now, captureRaw: captureRaw}
}
// view returns a copy of the collected state, deep enough that later events
// cannot race with readers of the snapshot.
func (tc *traceCollector) view() traceView {
tc.mu.Lock()
defer tc.mu.Unlock()
v := tc.v
v.raw = append([]TraceEvent(nil), tc.v.raw...)
v.dnsAddrs = append([]string(nil), tc.v.dnsAddrs...)
v.wroteHeaderFields = append([]NameValuePair(nil), tc.v.wroteHeaderFields...)
v.info1xx = append([]informational1xx(nil), tc.v.info1xx...)
if tc.v.putIdle != nil {
cp := *tc.v.putIdle
v.putIdle = &cp
}
return v
}
// dialTarget returns the host:port the transport reported it was obtaining a
// connection for, without paying for a full snapshot copy.
func (tc *traceCollector) dialTarget() string {
tc.mu.Lock()
defer tc.mu.Unlock()
return tc.v.getConnAddr
}
// event appends a raw trace event; callers must hold tc.mu.
func (tc *traceCollector) event(name, detail string) {
if !tc.captureRaw {
return
}
tc.v.raw = append(tc.v.raw, TraceEvent{
Name: name,
Time: tc.now().UTC().Format(time.RFC3339Nano),
Detail: detail,
})
}
func (tc *traceCollector) clientTrace() *httptrace.ClientTrace {
return &httptrace.ClientTrace{
GetConn: func(hostPort string) {
tc.mu.Lock()
defer tc.mu.Unlock()
if tc.v.getConn.IsZero() {
tc.v.getConn = tc.now()
}
if tc.v.getConnAddr == "" {
tc.v.getConnAddr = hostPort
}
tc.event("GetConn", hostPort)
},
DNSStart: func(info httptrace.DNSStartInfo) {
tc.mu.Lock()
defer tc.mu.Unlock()
if tc.v.dnsStart.IsZero() {
tc.v.dnsStart = tc.now()
}
tc.event("DNSStart", info.Host)
},
DNSDone: func(info httptrace.DNSDoneInfo) {
tc.mu.Lock()
defer tc.mu.Unlock()
tc.v.dnsDone = tc.now()
tc.v.dnsCoalesced = tc.v.dnsCoalesced || info.Coalesced
for _, a := range info.Addrs {
tc.v.dnsAddrs = appendUnique(tc.v.dnsAddrs, a.IP.String())
}
detail := ""
if info.Err != nil {
tc.v.dnsErr = info.Err
detail = "error: " + info.Err.Error()
}
if info.Coalesced {
detail = strings.TrimSpace("coalesced " + detail)
}
tc.event("DNSDone", detail)
},
ConnectStart: func(network, addr string) {
tc.mu.Lock()
defer tc.mu.Unlock()
if tc.v.connectStart.IsZero() {
tc.v.connectStart = tc.now()
}
if tc.v.network == "" {
tc.v.network = network
}
tc.event("ConnectStart", network+" "+addr)
},
ConnectDone: func(network, addr string, err error) {
tc.mu.Lock()
defer tc.mu.Unlock()
if err != nil {
tc.v.connectErr = err
tc.event("ConnectDone", network+" "+addr+" error: "+err.Error())
return
}
tc.v.connectDone = tc.now()
tc.v.network = network
tc.event("ConnectDone", network+" "+addr)
},
TLSHandshakeStart: func() {
tc.mu.Lock()
defer tc.mu.Unlock()
if tc.v.tlsStart.IsZero() {
tc.v.tlsStart = tc.now()
}
tc.event("TLSHandshakeStart", "")
},
TLSHandshakeDone: func(state tls.ConnectionState, err error) {
tc.mu.Lock()
defer tc.mu.Unlock()
if err != nil {
tc.v.tlsErr = err
tc.event("TLSHandshakeDone", "error: "+err.Error())
return
}
tc.v.tlsDone = tc.now()
st := state
tc.v.tlsState = &st
tc.event("TLSHandshakeDone", state.NegotiatedProtocol)
},
GotConn: func(info httptrace.GotConnInfo) {
tc.mu.Lock()
defer tc.mu.Unlock()
tc.v.gotConn = tc.now()
// A GotConn without a Conn carries no information; do not let it
// erase details from an earlier, complete event.
if info.Conn != nil {
tc.v.reused = info.Reused
tc.v.wasIdle = info.WasIdle
tc.v.idleTime = info.IdleTime
if la := info.Conn.LocalAddr(); la != nil {
tc.v.localAddr = la.String()
if tc.v.network == "" {
tc.v.network = la.Network()
}
}
if ra := info.Conn.RemoteAddr(); ra != nil {
tc.v.remoteAddr = ra.String()
}
}
tc.event("GotConn", tc.v.remoteAddr)
},
WroteHeaderField: func(key string, values []string) {
tc.mu.Lock()
defer tc.mu.Unlock()
for _, v := range values {
tc.v.wroteHeaderFields = append(tc.v.wroteHeaderFields, NameValuePair{Name: key, Value: v})
}
tc.event("WroteHeaderField", key)
},
Wait100Continue: func() {
tc.mu.Lock()
defer tc.mu.Unlock()
if tc.v.wait100.IsZero() {
tc.v.wait100 = tc.now()
}
tc.event("Wait100Continue", "")
},
Got100Continue: func() {
tc.mu.Lock()
defer tc.mu.Unlock()
if tc.v.got100.IsZero() {
tc.v.got100 = tc.now()
}
tc.event("Got100Continue", "")
},
WroteHeaders: func() {
tc.mu.Lock()
if tc.v.wroteHeaders.IsZero() {
tc.v.wroteHeaders = tc.now()
}
tc.event("WroteHeaders", "")
notify := tc.notify
tc.mu.Unlock()
if notify != nil {
notify(StateRequestHeadersWritten)
}
},
WroteRequest: func(info httptrace.WroteRequestInfo) {
tc.mu.Lock()
defer tc.mu.Unlock()
tc.v.wroteRequest = tc.now()
detail := ""
if info.Err != nil {
tc.v.wroteReqErr = info.Err
detail = "error: " + info.Err.Error()
}
tc.event("WroteRequest", detail)
},
GotFirstResponseByte: func() {
tc.mu.Lock()
defer tc.mu.Unlock()
if tc.v.firstByte.IsZero() {
tc.v.firstByte = tc.now()
}
tc.event("GotFirstResponseByte", "")
},
Got1xxResponse: func(code int, header textproto.MIMEHeader) error {
tc.mu.Lock()
defer tc.mu.Unlock()
if len(tc.v.info1xx) < max1xxRecorded {
h := make(http.Header, len(header))
for k, vs := range header {
h[k] = append([]string(nil), vs...)
}
tc.v.info1xx = append(tc.v.info1xx, informational1xx{code: code, header: h})
}
tc.event("Got1xxResponse", strconv.Itoa(code))
return nil
},
PutIdleConn: func(err error) {
tc.mu.Lock()
defer tc.mu.Unlock()
tc.v.putIdle = &putIdleResult{returned: err == nil, err: err}
detail := ""
if err != nil {
detail = "error: " + err.Error()
}
tc.event("PutIdleConn", detail)
},
}
}
func appendUnique(list []string, s string) []string {
for _, existing := range list {
if existing == s {
return list
}
}
return append(list, s)
}
// durMS converts a duration to non-negative milliseconds.
func durMS(d time.Duration) float64 {
if d < 0 {
d = 0
}
return float64(d) / float64(time.Millisecond)
}
// msBetween returns the milliseconds between two events, -1 when either did
// not happen, and never a negative value (clock jitter between goroutines is
// clamped to 0).
func msBetween(a, b time.Time) float64 {
if a.IsZero() || b.IsZero() {
return -1
}
return durMS(b.Sub(a))
}
// computeTimings maps the collected events onto HAR timing semantics.
//
// - On a reused connection (including HTTP/2 multiplexing onto an existing
// connection) no DNS/connect/TLS work happened for this exchange, so those
// fields are -1 — not a misleading 0 — and blocked covers the wait for the
// connection from the pool.
// - Any phase whose events were not observed stays -1.
func computeTimings(v traceView, start, finish time.Time) *Timings {
t := &Timings{Blocked: -1, DNS: -1, Connect: -1, Send: -1, Wait: -1, Receive: -1, SSL: -1}
if !v.gotConn.IsZero() {
if v.reused {
t.Blocked = msBetween(start, v.gotConn)
} else {
firstNet := v.dnsStart
if firstNet.IsZero() {
firstNet = v.connectStart
}
if firstNet.IsZero() {
firstNet = v.gotConn
}
t.Blocked = msBetween(start, firstNet)
t.DNS = msBetween(v.dnsStart, v.dnsDone)
t.Connect = msBetween(v.connectStart, v.connectDone)
t.SSL = msBetween(v.tlsStart, v.tlsDone)
}
}
sendEnd := v.wroteRequest
if sendEnd.IsZero() {
sendEnd = v.wroteHeaders
}
t.Send = msBetween(v.gotConn, sendEnd)
waitStart := sendEnd
if waitStart.IsZero() {
waitStart = v.gotConn
}
t.Wait = msBetween(waitStart, v.firstByte)
t.Receive = msBetween(v.firstByte, finish)
return t
}
// Package recorder records the full life cycle of net/http client calls —
// including calls that fail at the DNS, TCP, TLS or HTTP layer — and exports
// them as standard HAR 1.2 documents enriched with one versioned _recorder
// extension. Removing that object leaves a valid plain HAR 1.2 document.
//
// The guiding principle: record what was actually observable during the call,
// as accurately and structurally as possible, without ever changing the
// behavior the caller sees. Values that cannot be measured reliably (wire
// header sizes, compressed body sizes after transparent gzip decoding, timing
// phases that did not happen) are reported as -1 or omitted — never invented.
//
// Usage:
//
// rec := recorder.NewMemoryRecorder()
// config := recorder.DefaultConfig()
// if err := config.Validate(); err != nil {
// log.Fatal(err)
// }
// client := &http.Client{
// Transport: recorder.NewTransport(http.DefaultTransport, rec, config),
// }
// resp, err := client.Get("https://example.com/")
// // ... consume resp.Body; the entry is finalized on EOF/Close ...
// _ = rec.WriteHAR(os.Stdout)
//
// A response entry is not complete when RoundTrip returns: the body has not
// been read yet. The entry reaches the Recorder when the body hits EOF, is
// closed early, or fails — or immediately, when RoundTrip itself returns an
// error. A response body that is neither fully read nor closed (a caller
// bug) never finalizes; the library deliberately uses no finalizers.
package recorder
import (
"bytes"
"context"
"crypto/sha256"
"crypto/tls"
"crypto/x509"
"encoding/base64"
"encoding/hex"
"errors"
"fmt"
"io"
"mime"
"mime/multipart"
"net"
"net/http"
"net/http/httptrace"
"net/url"
"strings"
"sync"
"time"
)
// Exchange life cycle states, recorded under _recorder.state. Only terminal states
// (completed, failed, closed_early) ever appear in exported entries, because
// entries are emitted exclusively at finalization.
const (
StateCreated = "created"
StateRequestStarted = "request_started"
StateRequestHeadersWritten = "request_headers_written"
StateRequestBodyStreaming = "request_body_streaming"
StateResponseHeadersReceived = "response_headers_received"
StateResponseBodyStreaming = "response_body_streaming"
StateCompleted = "completed"
StateFailed = "failed"
StateClosedEarly = "closed_early"
)
// Transport is an http.RoundTripper that records exchanges passing through it.
// A HeadSamplingPolicy may deliberately bypass recording for selected
// exchanges. Transport wraps a base RoundTripper (http.DefaultTransport when
// Base is nil) and never alters the request, response, or error the caller sees.
//
// Transport is safe for concurrent use by multiple goroutines provided its
// fields are immutable after construction. A zero-value Transport is not
// supported; construct one with NewTransport and an explicit Config. When a
// standard *http.Transport has a Proxy callback, NewTransport clones it once
// so the selected proxy URL can be
// observed without invoking that callback twice; configure the base before
// passing it in and close idle connections through this Transport or its
// http.Client.
type Transport struct {
base http.RoundTripper
recorder Recorder
config Config
initOnce sync.Once
red *redactor
respRed *redactor
store BodyStore
effectiveBase http.RoundTripper
sampling samplingCounters
}
// NewTransport builds an immutable Transport wrapping base. A nil base uses
// http.DefaultTransport. A nil recorder disables sink delivery while keeping
// OnEntryCompleted available.
func NewTransport(base http.RoundTripper, recorder Recorder, config Config) *Transport {
return &Transport{base: base, recorder: recorder, config: config}
}
func (t *Transport) init() {
t.initOnce.Do(func() {
t.red = newRedactor(&t.config)
t.respRed = newRedactorWithRules(&t.config,
effectiveRedactionRules(t.config.Redaction.Common, t.config.Redaction.Response))
t.store = t.config.BodyStore
if t.store == nil {
t.store = MemoryBodyStore{}
}
base := t.base
if base == nil {
base = http.DefaultTransport
}
if standard, ok := base.(*http.Transport); ok && standard.Proxy != nil {
clone := standard.Clone()
proxyFunc := clone.Proxy
clone.Proxy = func(req *http.Request) (*url.URL, error) {
proxyURL, err := proxyFunc(req)
if observation := proxyObservationFromContext(req.Context()); observation != nil {
observation.set(proxyURL)
}
return proxyURL, err
}
t.effectiveBase = clone
} else {
t.effectiveBase = base
}
})
}
func (t *Transport) wrappedTransport() http.RoundTripper {
t.init()
return t.effectiveBase
}
func (t *Transport) uninstrumentedBase() http.RoundTripper {
if t.base != nil {
return t.base
}
return http.DefaultTransport
}
type proxyObservationKey struct{}
type proxyObservation struct {
mu sync.Mutex
url *url.URL
}
func (o *proxyObservation) set(proxyURL *url.URL) {
o.mu.Lock()
defer o.mu.Unlock()
if proxyURL == nil {
o.url = nil
return
}
cp := *proxyURL
o.url = &cp
}
func (o *proxyObservation) get() *url.URL {
o.mu.Lock()
defer o.mu.Unlock()
if o.url == nil {
return nil
}
cp := *o.url
return &cp
}
func proxyObservationFromContext(ctx context.Context) *proxyObservation {
observation, _ := ctx.Value(proxyObservationKey{}).(*proxyObservation)
return observation
}
// CloseIdleConnections forwards to the wrapped transport when it supports it,
// keeping http.Client.CloseIdleConnections working through the wrapper.
func (t *Transport) CloseIdleConnections() {
if ci, ok := t.wrappedTransport().(interface{ CloseIdleConnections() }); ok {
ci.CloseIdleConnections()
}
}
// RoundTrip implements http.RoundTripper. The caller's request object is
// never mutated: recording hooks are attached to a clone. The returned
// response and error are exactly what the base transport produced, except
// that resp.Body is wrapped to observe the caller's reads.
func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
identity := resolveExchangeIdentity(req.Context())
samplingMeta := headSamplingMeta(req, identity)
sampleDecision := t.decideHeadSampling(req.Context(), samplingMeta)
t.callOnHeadSamplingDecision(req.Context(), samplingMeta, sampleDecision)
if sampleDecision == HeadSampleDrop {
return t.uninstrumentedBase().RoundTrip(req)
}
t.init()
ex := t.newExchange(req, identity, sampleDecision == HeadSampleMetadataOnly)
var err error
ex.reqDecision, err = decideBodyCapture(req.Context(), t.config.BodyCapturePolicy,
requestCaptureMeta(req, ex.traceID, ex.redirectIndex),
BodyCaptureDecision{
Capture: t.config.CaptureRequestBody,
Embed: t.config.EmbedBodies,
Hash: t.config.HashBodies,
MaxBodyBytes: t.config.MaxRequestBodyBytes,
})
if err != nil {
t.internalError(err)
}
if ex.metadataOnly {
ex.reqDecision = BodyCaptureDecision{}
}
proxySeen := &proxyObservation{}
ctx := context.WithValue(req.Context(), proxyObservationKey{}, proxySeen)
ctx = httptrace.WithClientTrace(ctx, ex.trace.clientTrace())
creq := req.Clone(ctx)
ex.req = creq
if creq.Body != nil && creq.Body != http.NoBody {
ex.reqCap = t.newCapture(ctx, ex.id, "request", creq.Header.Get("Content-Type"), creq.Header.Get("Content-Encoding"),
ex.reqDecision, creq.ContentLength, ex.red)
ex.reqCap.setExpected(creq.ContentLength)
creq.Body = &requestBodyRecorder{rc: creq.Body, bc: ex.reqCap, ex: ex}
if orig := creq.GetBody; orig != nil {
bc, exRef := ex.reqCap, ex
creq.GetBody = func() (io.ReadCloser, error) {
rc, err := orig()
if err != nil {
return nil, err
}
// The transport is replaying the body (internal retry):
// restart the capture so the record reflects the bytes of
// the attempt that actually went out.
bc.reset()
return &requestBodyRecorder{rc: rc, bc: bc, ex: exRef}, nil
}
}
}
ex.setState(StateRequestStarted)
resp, err := t.wrappedTransport().RoundTrip(creq)
ex.detectProxy(proxySeen.get(), ex.trace.dialTarget())
if err != nil {
ex.finalizeTransportError(err)
return resp, err
}
ex.onResponse(resp)
ex.respDecision, err = decideBodyCapture(ex.ctx, t.config.BodyCapturePolicy,
responseCaptureMeta(creq, resp, ex.traceID, ex.redirectIndex),
BodyCaptureDecision{
Capture: t.config.CaptureResponseBody,
Embed: t.config.EmbedBodies,
Hash: t.config.HashBodies,
MaxBodyBytes: t.config.MaxResponseBodyBytes,
})
if err != nil {
t.internalError(err)
}
if ex.metadataOnly {
ex.respDecision = BodyCaptureDecision{}
}
if resp.Body == nil {
// RoundTripper contract requires a non-nil body, but be tolerant of
// sloppy custom transports.
resp.Body = http.NoBody
}
ex.respCap = t.newCapture(ctx, ex.id, "response", resp.Header.Get("Content-Type"), resp.Header.Get("Content-Encoding"),
ex.respDecision, resp.ContentLength, ex.respRed)
if resp.StatusCode == http.StatusSwitchingProtocols {
// A 101 response ends the HTTP exchange. Its Body is the upgraded,
// bidirectional protocol stream (for example WebSocket), not an HTTP
// response body. Leave it untouched so callers retain io.ReadWriteCloser
// and do not mistake protocol frames for HAR body bytes.
ex.respCap.finishComplete()
ex.finalizeComplete()
return resp, nil
}
resp.Body = &responseBodyRecorder{rc: resp.Body, bc: ex.respCap, ex: ex}
if responseHasNoBody(creq, resp) {
// Nothing will ever be read; finalize now so callers that (legally)
// never touch the empty body still produce an entry.
ex.respCap.finishComplete()
ex.finalizeComplete()
}
return resp, nil
}
// detectProxy records the redacted URL selected by a standard http.Transport
// without evaluating its potentially stateful Proxy callback a second time.
// Custom RoundTrippers cannot expose that selection; for those, a differing
// dial target remains a host:port fallback.
func (ex *exchange) detectProxy(proxyURL *url.URL, dialed string) {
if proxyURL != nil {
ex.hasProxy = true
ex.proxyURL = ex.red.redactURL(proxyURL)
return
}
if dialed == "" || ex.req == nil || ex.req.URL == nil {
return
}
originPort := ex.req.URL.Port()
if originPort == "" {
switch ex.req.URL.Scheme {
case "http":
originPort = "80"
case "https":
originPort = "443"
}
}
originAddr := net.JoinHostPort(ex.req.URL.Hostname(), originPort)
if !strings.EqualFold(dialed, originAddr) {
ex.hasProxy = true
ex.proxyURL = dialed
}
}
func (t *Transport) newCapture(ctx context.Context, exchangeID, direction, contentType, contentEncoding string,
decision BodyCaptureDecision, contentLength int64, red *redactor,
) *bodyCapture {
// Derive the store's pre-allocation hint from Content-Length: never
// beyond what the capture limit allows, never negative, and left at 0
// (unknown) when no length was announced. The hint is advisory only —
// a stream that disagrees with it just grows or stops normally.
sizeHint := contentLength
if sizeHint < 0 {
sizeHint = 0
}
if decision.MaxBodyBytes > 0 && sizeHint > decision.MaxBodyBytes {
sizeHint = decision.MaxBodyBytes
}
meta := BodyMetadata{
ExchangeID: exchangeID,
Direction: direction,
ContentType: contentType,
SizeHint: sizeHint,
}
var decoder ContentDecoder
enc := strings.ToLower(strings.TrimSpace(contentEncoding))
if enc != "" && enc != "identity" && !strings.Contains(enc, ",") {
decoder = t.config.ContentDecoders[enc]
}
if decision.RedactorOverride != nil {
red = red.withBodyRedactor(contentType, decision.RedactorOverride)
}
return newBodyCapture(ctx, t.store, meta,
contentEncoding, decision.Capture, decision.MaxBodyBytes, t.config.BodyHashAlgorithm, decision.Hash, red, decoder, t.internalError)
}
// internalError applies the configured internal error policy. It never
// panics and never touches the HTTP flow.
func (t *Transport) internalError(err error) {
reportInternalError(t.config.InternalErrorMode, t.config.OnInternalError, t.config.Logf, err)
}
func callSafely(fn func()) {
defer func() { _ = recover() }()
fn()
}
// responseHasNoBody reports whether the response can never carry body bytes,
// so the exchange can be finalized at RoundTrip time.
func responseHasNoBody(req *http.Request, resp *http.Response) bool {
if req.Method == http.MethodHead {
return true
}
if resp.StatusCode < 200 || resp.StatusCode == http.StatusNoContent || resp.StatusCode == http.StatusNotModified {
return true
}
// ContentLength 0 means an explicit zero-length body (unknown is -1).
return resp.ContentLength == 0 && len(resp.TransferEncoding) == 0
}
// respSnapshot freezes the response metadata at RoundTrip time so later
// header mutations by the caller cannot race with entry building.
type respSnapshot struct {
present bool
status int
statusText string
proto string
headers http.Header
cookies []*http.Cookie
contentLength int64
transferEncoding []string
uncompressed bool
tls *tls.ConnectionState
}
// exchange tracks one physical HTTP exchange from RoundTrip entry to
// finalization.
type exchange struct {
t *Transport
red *redactor
respRed *redactor
audit *redactionAudit
ctx context.Context
id string
traceID string
redirectIndex int
hasTraceState bool
metadataOnly bool
hasProxy bool
proxyURL string
comment string
start time.Time
trace *traceCollector
req *http.Request // the clone handed to the base transport
reqCap *bodyCapture
respCap *bodyCapture
reqDecision BodyCaptureDecision
respDecision BodyCaptureDecision
mu sync.Mutex
state string
done bool
resp *http.Response
respSnap respSnapshot
finalizeOnce sync.Once
finish time.Time
}
func (t *Transport) newExchange(req *http.Request, identity exchangeIdentity, metadataOnly bool) *exchange {
audit := &redactionAudit{}
keys := &protectionKeyCache{}
hints := requestRedactionFromContext(req.Context())
ex := &exchange{
t: t,
red: t.red.withContext(req.Context(), keys).withRules(
effectiveRedactionRules(hints.Common, hints.Request),
).withAudit(audit, RequestBody),
respRed: t.respRed.withContext(req.Context(), keys).withRules(
effectiveRedactionRules(hints.Common, hints.Response),
).withAudit(audit, ResponseBody),
audit: audit,
ctx: req.Context(),
id: newID(),
start: time.Now(),
trace: newTraceCollector(t.config.CaptureRawTrace && !metadataOnly),
state: StateCreated,
traceID: identity.traceID,
redirectIndex: identity.redirectIndex,
hasTraceState: identity.hasTraceState,
metadataOnly: metadataOnly,
comment: requestCommentFromContext(req.Context()),
}
ex.trace.notify = ex.setState
if ex.traceID == "" {
ex.traceID = newID()
}
return ex
}
// setState advances the life cycle state; terminal states set by finalization
// are never overwritten.
func (ex *exchange) setState(s string) {
ex.mu.Lock()
defer ex.mu.Unlock()
if !ex.done {
ex.state = s
}
}
func (ex *exchange) captureHeaders() bool {
return !ex.metadataOnly && ex.t.config.CaptureHeaders
}
func (ex *exchange) captureCookies() bool {
return !ex.metadataOnly && ex.t.config.CaptureCookies
}
func (ex *exchange) captureCertificates() bool {
return !ex.metadataOnly && ex.t.config.CaptureCertificates
}
func (ex *exchange) captureRawTrace() bool {
return !ex.metadataOnly && ex.t.config.CaptureRawTrace
}
func (ex *exchange) markDone(state string) {
ex.mu.Lock()
defer ex.mu.Unlock()
ex.state = state
ex.done = true
}
// onResponse snapshots response metadata the moment the base transport
// returns it.
func (ex *exchange) onResponse(resp *http.Response) {
ex.mu.Lock()
defer ex.mu.Unlock()
ex.state = StateResponseHeadersReceived
ex.resp = resp
ex.respSnap = respSnapshot{
present: true,
status: resp.StatusCode,
statusText: statusText(resp),
proto: resp.Proto,
headers: resp.Header.Clone(),
cookies: resp.Cookies(),
contentLength: resp.ContentLength,
transferEncoding: append([]string(nil), resp.TransferEncoding...),
uncompressed: resp.Uncompressed,
tls: resp.TLS,
}
}
// contextErr returns the request context's Err(), nil while it is live.
func (ex *exchange) contextErr() error {
if ex.ctx == nil {
return nil
}
return ex.ctx.Err()
}
// contextCause returns context.Cause when the request context has been
// canceled, nil otherwise.
func (ex *exchange) contextCause() error {
if ex.ctx == nil || ex.ctx.Err() == nil {
return nil
}
return context.Cause(ex.ctx)
}
// finalizeTransportError finalizes an exchange whose RoundTrip failed without
// producing a response. Recorder failures are reported separately and never
// replace the transport error returned to the caller.
func (ex *exchange) finalizeTransportError(err error) {
ex.finalizeOnce.Do(func() {
ex.finish = time.Now()
ex.markDone(StateFailed)
v := ex.trace.view()
var reqBodyErr error
if ex.reqCap != nil {
reqBodyErr = ex.reqCap.readError()
}
ctxErr := ex.contextErr()
phase := classifyPhase(err, v, ex.hasProxy, false, ctxErr != nil, reqBodyErr)
info := newErrorInfo(err, phase, ex.red, ctxErr, ex.contextCause())
ex.emit(info)
})
}
// finalizeComplete finalizes after the response body reached EOF (or was
// known-empty at RoundTrip time).
func (ex *exchange) finalizeComplete() {
ex.finalizeOnce.Do(func() {
ex.finish = time.Now()
ex.markDone(StateCompleted)
ex.emit(nil)
})
}
// finalizeBodyReadError finalizes after a response body read failed.
func (ex *exchange) finalizeBodyReadError(err error) {
ex.finalizeOnce.Do(func() {
ex.finish = time.Now()
ex.markDone(StateFailed)
info := newErrorInfo(err, PhaseReadResponseBody, ex.red, ex.contextErr(), ex.contextCause())
ex.emit(info)
})
}
// finalizeClosed finalizes after the caller closed the body before EOF.
func (ex *exchange) finalizeClosed() {
ex.finalizeOnce.Do(func() {
ex.finish = time.Now()
state := StateClosedEarly
if ex.respCap != nil && ex.respCap.isComplete() {
state = StateCompleted
}
ex.markDone(state)
ex.emit(nil)
})
}
// emit builds the entry, applies retention and required asset cleanup, reports
// the effective disposition, and transfers kept entries to the Recorder. Each
// extension point is panic-contained independently so one cannot suppress the
// next.
func (ex *exchange) emit(errInfo *ErrorInfo) {
entry := ex.buildEntry(errInfo)
if ex.t.decideRetention(ex.ctx, entry) == DiscardEntry {
if ex.releaseDiscardedAssets(entry) {
ex.t.sampling.discarded.Add(1)
ex.callOnEntryCompleted(entry, EntryDispositionDiscard)
return
}
ex.t.sampling.retained.Add(1)
}
ex.callOnEntryCompleted(entry, EntryDispositionKeep)
ex.callRecorder(entry)
}
func (ex *exchange) callOnEntryCompleted(entry *Entry, disposition EntryDisposition) {
if ex.t.config.OnEntryCompleted == nil {
return
}
defer func() {
if recovered := recover(); recovered != nil {
ex.t.internalError(fmt.Errorf("recorder: panic in OnEntryCompleted: %v", recovered))
}
}()
ex.t.config.OnEntryCompleted(ex.ctx, entry, disposition)
}
func (ex *exchange) callRecorder(entry *Entry) {
if ex.t.recorder == nil {
return
}
defer func() {
if recovered := recover(); recovered != nil {
ex.t.internalError(fmt.Errorf("recorder: panic while recording entry: %v", recovered))
}
}()
if err := ex.t.recorder.Record(entry); err != nil {
ex.t.internalError(fmt.Errorf("recorder: record finalized entry: %w", err))
}
}
func (ex *exchange) releaseDiscardedAssets(entry *Entry) bool {
if !entryHasStoreReferences(entry) {
return true
}
releaser, ok := ex.t.store.(entryAssetReleaser)
if !ok {
ex.t.sampling.assetReleaseFailures.Add(1)
ex.t.internalError(errors.New("recorder: discarded entry body store cannot release assets; retaining entry"))
return false
}
if err := releaser.ReleaseEntryAssets(entry); err != nil {
ex.t.sampling.assetReleaseFailures.Add(1)
ex.t.internalError(fmt.Errorf("recorder: release discarded entry assets: %w", err))
return false
}
return true
}
// buildEntry assembles the immutable HAR entry snapshot.
func (ex *exchange) buildEntry(errInfo *ErrorInfo) *Entry {
v := ex.trace.view()
ex.mu.Lock()
state := ex.state
snap := ex.respSnap
var trailers http.Header
if ex.resp != nil && len(ex.resp.Trailer) > 0 {
trailers = ex.resp.Trailer.Clone()
}
ex.mu.Unlock()
proto := ex.httpVersion(v, snap)
e := &Entry{
StartedDateTime: ex.start.UTC().Format(harTimeFormat),
Time: durMS(ex.finish.Sub(ex.start)),
Cache: &Cache{},
Comment: ex.comment,
Recorder: &RecorderEntryExtension{
SchemaVersion: RecorderExtensionVersion,
TraceID: ex.traceID,
ExchangeID: ex.id,
State: state,
Error: errInfo,
},
started: ex.start,
}
if ex.hasTraceState {
idx := ex.redirectIndex
e.Recorder.RedirectIndex = &idx
}
e.Request = ex.buildRequest(v, proto)
if e.Request.PostData != nil {
e.Recorder.RequestBodyEncoding = e.Request.PostData.encoding
}
e.Response = ex.buildResponse(snap)
if e.Response.Content != nil {
e.Recorder.ResponseBodyDecoded = e.Response.Content.decoded
}
if !v.wait100.IsZero() || !v.got100.IsZero() {
e.Recorder.Expect100 = &Expect100Info{
Waited: !v.wait100.IsZero(),
ContinueReceived: !v.got100.IsZero(),
}
if ms := msBetween(v.wait100, v.got100); ms >= 0 {
e.Recorder.Expect100.WaitMS = ms
}
}
for _, ir := range v.info1xx {
rec := InformationalResponse{Status: ir.code}
if ex.captureHeaders() && len(ir.header) > 0 {
rec.Headers = ex.respRed.responseHeaderPairs(ir.header)
}
e.Recorder.Informational = append(e.Recorder.Informational, rec)
}
e.Timings = computeTimings(v, ex.start, ex.finish)
// serverIPAddress means the origin server's IP (HAR: result of DNS
// resolution). Through a proxy the TCP peer is the proxy and the origin
// IP is never observable client-side, so the field is omitted (the proxy
// address stays available under _recorder.network). It is also only written
// when the peer address really is an IP.
if v.remoteAddr != "" && !ex.hasProxy {
if host, _, err := net.SplitHostPort(v.remoteAddr); err == nil {
if ip := net.ParseIP(host); ip != nil {
e.ServerIPAddress = ip.String()
}
}
}
if v.localAddr != "" {
// HAR "connection": a unique ID of the underlying connection; the
// local port is unique per live connection, matching browser usage.
if _, port, err := net.SplitHostPort(v.localAddr); err == nil {
e.Connection = port
}
}
e.Recorder.Network = ex.buildNetwork(v, proto)
if ex.t.config.CaptureTLS {
e.Recorder.TLS = ex.buildTLS(v, snap)
}
e.Recorder.RequestBody = ex.reqCap.info(ex.red)
e.Recorder.ResponseBody = ex.respCap.info(ex.respRed)
if ex.captureRawTrace() {
e.Recorder.RawTrace = ex.red.traceEvents(v.raw)
}
if ex.captureHeaders() {
if len(ex.req.Trailer) > 0 {
e.Recorder.RequestTrailers = ex.red.headerPairs(ex.req.Trailer, "")
}
if len(trailers) > 0 {
e.Recorder.ResponseTrailers = ex.respRed.headerPairs(trailers, "")
}
}
if len(ex.req.TransferEncoding) > 0 {
e.Recorder.RequestTransferEncoding = append([]string(nil), ex.req.TransferEncoding...)
}
if len(snap.transferEncoding) > 0 {
e.Recorder.ResponseTransferEncoding = snap.transferEncoding
}
e.Recorder.Redaction = ex.audit.snapshot()
for index, failure := range ex.audit.protectionFailures() {
if failure.first == nil || failure.count == 0 {
continue
}
direction := "request"
if index == 1 {
direction = "response"
}
ex.t.internalError(fmt.Errorf("recorder: %s sensitive value protection failed for %d value(s): %w",
direction, failure.count, failure.first))
}
return e
}
// httpVersion returns the HTTP version actually observed for this exchange,
// or "" when it cannot be known. Never guessed:
//
// - a response exists -> its negotiated protocol (authoritative)
// - request headers written -> the TLS ALPN result when a handshake
// completed in this exchange; for cleartext through the standard
// *http.Transport, HTTP/1.1 (the stdlib never speaks h2c)
// - anything earlier (DNS/connect/TLS failures) -> "" — no request line
// ever reached the wire, so it has no version
func (ex *exchange) httpVersion(v traceView, snap respSnapshot) string {
if snap.present && snap.proto != "" {
return snap.proto
}
if v.wroteHeaders.IsZero() {
return ""
}
if st := v.tlsState; st != nil && st.HandshakeComplete {
if st.NegotiatedProtocol == "h2" {
return "HTTP/2.0"
}
return "HTTP/1.1"
}
if _, ok := ex.t.wrappedTransport().(*http.Transport); ok &&
ex.req.URL != nil && ex.req.URL.Scheme == "http" {
return "HTTP/1.1"
}
// Reused TLS connections (no handshake event) or custom transports:
// the protocol is not observable.
return ""
}
func (ex *exchange) buildRequest(v traceView, effectiveProto string) *Request {
req := ex.req
r := &Request{
Method: req.Method,
URL: ex.red.redactURL(req.URL),
HTTPVersion: effectiveProto,
Cookies: []Cookie{},
Headers: []NameValuePair{},
QueryString: []NameValuePair{},
HeadersSize: -1,
BodySize: 0,
}
if r.Method == "" {
r.Method = http.MethodGet
}
if req.URL != nil && !ex.metadataOnly {
r.QueryString = ex.red.queryPairs(req.URL.RawQuery)
}
if ex.captureHeaders() {
if len(v.wroteHeaderFields) > 0 {
// Prefer the headers the transport actually wrote to the wire
// (httptrace.WroteHeaderField): they include transport-added
// fields (User-Agent, Accept-Encoding, Host / :authority) in
// wire order. Redaction applies the same way, including URL
// sanitization of Referer on redirect hops.
r.Headers = ex.red.sanitizeURLHeaders(ex.red.redactPairs(v.wroteHeaderFields))
} else {
// Nothing was written (failure before the request line, or a
// custom transport without trace support): fall back to the
// caller-provided header snapshot plus the Host header.
host := req.Host
if host == "" && req.URL != nil {
host = req.URL.Host
}
r.Headers = ex.red.sanitizeURLHeaders(ex.red.headerPairs(req.Header, host))
}
}
if ex.captureCookies() {
for _, c := range req.Cookies() {
val := c.Value
if ex.red.cookieRedacted(c.Name, "cookie") {
if val != redactedValue {
ex.red.recordCookieRedaction()
}
val = ex.red.protectString(val)
}
r.Cookies = append(r.Cookies, Cookie{Name: c.Name, Value: val})
}
}
if ex.reqCap != nil {
r.BodySize = ex.reqCap.totalBytes()
if ex.reqDecision.Capture && ex.reqDecision.Embed {
if b := ex.reqCap.bytes(); len(b) > 0 {
mimeType := req.Header.Get("Content-Type")
if mimeType == "" {
mimeType = "application/octet-stream"
}
whole := ex.reqCap.isComplete() && !ex.reqCap.isTruncated()
r.PostData = ex.buildPostData(mimeType, b, whole, ex.reqCap.isStoredRedacted())
}
}
}
return r
}
func (ex *exchange) buildPostData(mimeType string, b []byte, whole, storedRedacted bool) *PostData {
pd := &PostData{MimeType: mimeType}
if whole && !storedRedacted {
b = ex.red.redactStructuredBody(mimeType, b)
}
pd.Text, pd.encoding = contentText(mimeType, b)
if whole && isFormMime(mimeType) {
// Form fields reuse the query-parameter redaction rules.
for _, p := range ex.red.queryPairs(string(b)) {
pd.Params = append(pd.Params, PostParam{Name: p.Name, Value: p.Value})
}
} else if whole && isMultipartFormMime(mimeType) {
pd.Params = multipartPostParams(mimeType, b)
}
return pd
}
func multipartPostParams(mimeType string, b []byte) []PostParam {
_, params, err := mime.ParseMediaType(mimeType)
if err != nil || params["boundary"] == "" {
return nil
}
mr := multipart.NewReader(bytes.NewReader(b), params["boundary"])
var out []PostParam
for {
part, err := mr.NextPart()
if err == io.EOF {
return out
}
if err != nil || part.FormName() == "" {
return nil
}
p := PostParam{
Name: part.FormName(),
FileName: part.FileName(),
ContentType: part.Header.Get("Content-Type"),
}
content, err := io.ReadAll(part)
if err != nil {
return nil
}
if p.FileName == "" {
p.Value = string(content)
}
out = append(out, p)
}
}
func (ex *exchange) buildResponse(snap respSnapshot) *Response {
if !snap.present {
// No HTTP response was produced; status 0 keeps the document valid
// HAR 1.2 while _recorder.error carries the failure detail.
return &Response{
Status: 0,
StatusText: "",
HTTPVersion: "",
Cookies: []Cookie{},
Headers: []NameValuePair{},
Content: &Content{Size: 0, MimeType: "x-unknown"},
RedirectURL: "",
HeadersSize: -1,
BodySize: -1,
}
}
r := &Response{
Status: snap.status,
StatusText: snap.statusText,
HTTPVersion: snap.proto,
Cookies: []Cookie{},
Headers: []NameValuePair{},
RedirectURL: ex.respRed.redactURLString(snap.headers.Get("Location")),
HeadersSize: -1,
BodySize: -1,
}
if ex.captureHeaders() {
r.Headers = ex.respRed.responseHeaderPairs(snap.headers)
}
if ex.captureCookies() {
for _, c := range snap.cookies {
val := c.Value
if ex.respRed.cookieRedacted(c.Name, "set-cookie") {
if val != redactedValue {
ex.respRed.recordCookieRedaction()
}
val = ex.respRed.protectString(val)
}
hc := Cookie{
Name: c.Name,
Value: val,
Path: c.Path,
Domain: c.Domain,
HTTPOnly: c.HttpOnly,
Secure: c.Secure,
}
if !c.Expires.IsZero() {
hc.Expires = c.Expires.UTC().Format(time.RFC3339)
}
r.Cookies = append(r.Cookies, hc)
}
}
mimeType := snap.headers.Get("Content-Type")
if mimeType == "" {
mimeType = "x-unknown"
}
content := &Content{Size: 0, MimeType: mimeType}
if ex.respCap != nil {
content.Size = ex.respCap.totalBytes()
complete := ex.respCap.isComplete()
if snap.uncompressed {
// http.Transport decompressed the stream transparently: the
// caller-visible byte count is the decoded size, and the
// compressed wire size is no longer observable -> BodySize -1.
content.decoded = true
} else if complete {
// Identity encoding, fully read: caller bytes == wire payload.
r.BodySize = content.Size
}
if ex.respDecision.Capture && ex.respDecision.Embed {
if b := ex.respCap.bytes(); len(b) > 0 {
whole := complete && !ex.respCap.isTruncated()
if whole && !snap.uncompressed {
if ex.respCap.isStoredDecoded() {
content.decoded = true
content.Size = int64(len(b))
if r.BodySize >= 0 {
content.Compression = content.Size - r.BodySize
}
} else {
// The transport did not decompress; store the decoded
// form when a decoder is registered for the encoding.
// bodySize, hash and stream counters keep the wire view.
if decoded, ok := ex.decodeBody(snap.headers.Get("Content-Encoding"), b); ok {
b = decoded
content.decoded = true
content.Size = int64(len(decoded))
if r.BodySize >= 0 {
// HAR compression = bytes saved on the wire; can
// be negative when encoding expanded the content.
content.Compression = content.Size - r.BodySize
}
}
}
}
if whole && !ex.respCap.isStoredRedacted() {
b = ex.respRed.redactStructuredBody(mimeType, b)
}
content.Text, content.Encoding = contentText(mimeType, b)
}
}
}
r.Content = content
return r
}
// decodeBody decodes fully captured compressed content for HAR embedding when
// the stored representation was not already decoded by streaming structured
// redaction. It refuses multi-step encodings ("gzip, br"), reports decoder
// failures as internal errors (the record then keeps the captured wire
// representation), and abandons decoding when the decoded form would exceed
// MaxResponseBodyBytes — a compression bomb must not inflate the recorder's
// memory beyond the configured capture budget.
func (ex *exchange) decodeBody(encoding string, b []byte) ([]byte, bool) {
enc := strings.ToLower(strings.TrimSpace(encoding))
if enc == "" || enc == "identity" || strings.Contains(enc, ",") {
return nil, false
}
dec := ex.t.config.ContentDecoders[enc]
if dec == nil {
return nil, false
}
rc, err := dec(bytes.NewReader(b))
if err != nil {
ex.t.internalError(fmt.Errorf("recorder: open %s decoder: %w", enc, err))
return nil, false
}
defer func() { _ = rc.Close() }()
limit := ex.respDecision.MaxBodyBytes
var buf bytes.Buffer
if limit > 0 {
n, err := io.Copy(&buf, io.LimitReader(rc, limit+1))
if err != nil {
ex.t.internalError(fmt.Errorf("recorder: decode %s content: %w", enc, err))
return nil, false
}
if n > limit {
return nil, false
}
} else if _, err := io.Copy(&buf, rc); err != nil {
ex.t.internalError(fmt.Errorf("recorder: decode %s content: %w", enc, err))
return nil, false
}
return buf.Bytes(), true
}
func (ex *exchange) buildNetwork(v traceView, proto string) *NetworkInfo {
n := &NetworkInfo{
DNSAddresses: v.dnsAddrs,
DNSCoalesced: v.dnsCoalesced,
Network: v.network,
LocalAddress: v.localAddr,
RemoteAddress: v.remoteAddr,
ConnectionReused: v.reused,
WasIdle: v.wasIdle,
Proxy: ex.proxyURL,
HTTP2: strings.HasPrefix(proto, "HTTP/2"),
}
if v.wasIdle {
n.IdleTimeMS = durMS(v.idleTime)
}
if v.putIdle != nil {
pi := &PutIdleInfo{Returned: v.putIdle.returned}
if v.putIdle.err != nil {
pi.Error = ex.red.redactError(v.putIdle.err.Error())
}
n.PutIdle = pi
}
if host, _, err := net.SplitHostPort(v.remoteAddr); err == nil {
if ip := net.ParseIP(host); ip != nil {
if ip.To4() != nil {
n.IPVersion = "ipv4"
} else {
n.IPVersion = "ipv6"
}
}
}
return n
}
func (ex *exchange) buildTLS(v traceView, snap respSnapshot) *TLSInfo {
st := v.tlsState
if st == nil {
st = snap.tls // reused connections see no handshake event
}
if st == nil {
return nil
}
ti := &TLSInfo{
Version: tls.VersionName(st.Version),
CipherSuite: tls.CipherSuiteName(st.CipherSuite),
NegotiatedProtocol: st.NegotiatedProtocol,
ServerName: st.ServerName,
HandshakeComplete: st.HandshakeComplete,
DidResume: st.DidResume,
OCSPStapled: len(st.OCSPResponse) > 0,
SCTCount: len(st.SignedCertificateTimestamps),
VerifiedChains: len(st.VerifiedChains),
}
if ex.captureCertificates() {
for _, cert := range st.PeerCertificates {
ti.PeerCertificates = append(ti.PeerCertificates, newCertInfo(cert, ex.t.config.CaptureRawCertificates))
}
}
return ti
}
func newCertInfo(cert *x509.Certificate, includeRaw bool) CertInfo {
sum := sha256.Sum256(cert.Raw)
ci := CertInfo{
Subject: cert.Subject.String(),
Issuer: cert.Issuer.String(),
SerialNumber: cert.SerialNumber.String(),
DNSNames: append([]string(nil), cert.DNSNames...),
NotBefore: cert.NotBefore.UTC().Format(time.RFC3339),
NotAfter: cert.NotAfter.UTC().Format(time.RFC3339),
PublicKeyAlgorithm: cert.PublicKeyAlgorithm.String(),
SignatureAlgorithm: cert.SignatureAlgorithm.String(),
SHA256Fingerprint: hex.EncodeToString(sum[:]),
}
for _, ip := range cert.IPAddresses {
ci.IPAddresses = append(ci.IPAddresses, ip.String())
}
if includeRaw {
ci.RawDER = base64.StdEncoding.EncodeToString(cert.Raw)
}
return ci
}
// statusText extracts the reason phrase from resp.Status ("200 OK" -> "OK"),
// falling back to the standard text for the code.
func statusText(resp *http.Response) string {
if resp.Status != "" {
if _, text, ok := strings.Cut(resp.Status, " "); ok {
return text
}
}
return http.StatusText(resp.StatusCode)
}
package recorder
import (
"bytes"
"io"
"strings"
"unicode/utf8"
)
const (
maxXMLMarkupBytes = 64 << 10
maxXMLDepth = 1024
)
// xmlStreamRedactor buffers only the current markup token. Ordinary character
// data is passed through immediately, or discarded while inside a redacted
// element. Namespace prefixes and all bytes outside matched subtrees survive
// unchanged.
type xmlStreamRedactor struct {
dst io.Writer
bytes byteSink
elements map[string]struct{}
markup []byte
nameFold []byte
inMarkup bool
quote byte
brackets int
suppressNames []xmlNameSpan
suppressData []byte
err error
protected protectedValueBuffer
}
type xmlNameSpan struct {
start int
end int
}
func newXMLStreamRedactor(dst io.Writer, elements map[string]struct{}, protectors ...*bodyValueProtector) *xmlStreamRedactor {
protector := newBodyValueProtector(nil)
if len(protectors) > 0 && protectors[0] != nil {
protector = protectors[0]
}
r := &xmlStreamRedactor{dst: dst, bytes: newByteSink(dst), elements: elements}
r.protected.reset(protector)
return r
}
func (r *xmlStreamRedactor) Write(p []byte) (int, error) {
if r.err != nil {
return 0, r.err
}
for i, b := range p {
if err := r.consume(b); err != nil {
r.err = err
return i, err
}
}
return len(p), nil
}
func (r *xmlStreamRedactor) Close() error {
if r.err != nil {
return r.err
}
// Preserve an incomplete markup token only when it is outside a redacted
// subtree. Inside a matched element, failing closed avoids leaking content.
if r.inMarkup && len(r.suppressNames) == 0 {
_, r.err = r.dst.Write(r.markup)
} else if len(r.suppressNames) > 0 {
r.err = r.emitProtected()
}
return r.err
}
func (r *xmlStreamRedactor) consume(b byte) error {
if !r.inMarkup {
if b == '<' {
r.inMarkup = true
r.markup = append(r.markup[:0], b)
return nil
}
if len(r.suppressNames) == 0 {
return r.bytes.WriteByte(b)
}
r.protected.appendByte(b)
return nil
}
if len(r.markup) >= maxXMLMarkupBytes {
return errRedactionLimit
}
r.markup = append(r.markup, b)
if !r.markupComplete(b) {
return nil
}
return r.finishMarkup()
}
func (r *xmlStreamRedactor) markupComplete(b byte) bool {
t := r.markup
if bytes.HasPrefix(t, []byte("<!--")) {
return bytes.HasSuffix(t, []byte("-->"))
}
if bytes.HasPrefix(t, []byte("<![CDATA[")) {
return bytes.HasSuffix(t, []byte("]]>"))
}
if bytes.HasPrefix(t, []byte("<?")) {
return bytes.HasSuffix(t, []byte("?>"))
}
if r.quote != 0 {
if b == r.quote {
r.quote = 0
}
return false
}
if b == '\'' || b == '"' {
r.quote = b
return false
}
if bytes.HasPrefix(t, []byte("<!")) {
if b == '[' {
r.brackets++
} else if b == ']' && r.brackets > 0 {
r.brackets--
}
return b == '>' && r.brackets == 0
}
return b == '>'
}
func (r *xmlStreamRedactor) finishMarkup() error {
token := r.markup
r.inMarkup = false
r.quote = 0
r.brackets = 0
kind, local, selfClosing := xmlMarkupInfo(token)
if len(r.suppressNames) > 0 {
switch kind {
case 's':
if !selfClosing {
if len(r.suppressNames) >= maxXMLDepth || len(r.suppressData)+len(local) > maxXMLMarkupBytes {
return errRedactionLimit
}
if err := r.pushSuppressName(local); err != nil {
return err
}
}
case 'e':
top := len(r.suppressNames) - 1
if !r.suppressNameEqual(local, r.suppressNames[top]) {
return nil
}
r.suppressNames = r.suppressNames[:top]
if len(r.suppressNames) == 0 {
r.suppressData = r.suppressData[:0]
if err := r.emitProtected(); err != nil {
return err
}
_, err := r.dst.Write(token)
return err
}
}
if len(r.suppressNames) > 0 {
r.protected.appendBytes(token)
}
return nil
}
if _, err := r.dst.Write(token); err != nil {
return err
}
if kind == 's' && !selfClosing {
if r.elementMatches(local) {
r.suppressNames = r.suppressNames[:0]
r.suppressData = r.suppressData[:0]
if err := r.pushSuppressName(local); err != nil {
return err
}
r.protected.reset(r.protected.session)
if r.protected.redactImmediately() {
_, err := io.WriteString(r.dst, redactedValue)
return err
}
}
}
return nil
}
func (r *xmlStreamRedactor) emitProtected() error {
value := r.protected.protectedBytes()
if len(value) == 0 {
return nil
}
_, err := r.dst.Write(value)
return err
}
func (r *xmlStreamRedactor) elementMatches(local []byte) bool {
if folded, ok := foldASCIIName(local, r.nameFold[:0]); ok {
r.nameFold = folded
_, matched := r.elements[string(folded)]
return matched
}
_, matched := r.elements[strings.ToLower(string(local))]
return matched
}
func (r *xmlStreamRedactor) pushSuppressName(local []byte) error {
start := len(r.suppressData)
if folded, ok := foldASCIIName(local, r.suppressData); ok {
r.suppressData = folded
} else {
r.suppressData = append(r.suppressData, strings.ToLower(string(local))...)
}
if len(r.suppressData) > maxXMLMarkupBytes {
r.suppressData = r.suppressData[:start]
return errRedactionLimit
}
r.suppressNames = append(r.suppressNames, xmlNameSpan{start: start, end: len(r.suppressData)})
return nil
}
func (r *xmlStreamRedactor) suppressNameEqual(local []byte, span xmlNameSpan) bool {
saved := r.suppressData[span.start:span.end]
if folded, ok := foldASCIIName(local, r.nameFold[:0]); ok {
r.nameFold = folded
return bytes.Equal(folded, saved)
}
return strings.ToLower(string(local)) == string(saved)
}
func foldASCIIName(name, dst []byte) ([]byte, bool) {
for _, b := range name {
if b >= utf8.RuneSelf {
return dst, false
}
}
for _, b := range name {
if b >= 'A' && b <= 'Z' {
b += 'a' - 'A'
}
dst = append(dst, b)
}
return dst, true
}
// xmlMarkupInfo returns s=start, e=end, or 0 for comments, CDATA, processing
// instructions and declarations.
func xmlMarkupInfo(token []byte) (kind byte, local []byte, selfClosing bool) {
if len(token) < 3 || token[0] != '<' || token[1] == '!' || token[1] == '?' {
return 0, nil, false
}
i := 1
kind = 's'
if token[i] == '/' {
kind = 'e'
i++
}
start := i
for i < len(token) {
switch token[i] {
case ' ', '\t', '\r', '\n', '/', '>':
goto nameDone
default:
i++
}
}
nameDone:
if start == i {
return 0, nil, false
}
local = token[start:i]
if colon := bytes.LastIndexByte(local, ':'); colon >= 0 {
local = local[colon+1:]
}
if kind == 's' {
j := len(token) - 2
for j >= 0 && (token[j] == ' ' || token[j] == '\t' || token[j] == '\r' || token[j] == '\n') {
j--
}
selfClosing = j >= 0 && token[j] == '/'
}
return kind, local, selfClosing
}