diff --git a/http/httphi/bench_test.go b/http/httphi/bench_test.go index 5ea8fca..837e9ef 100644 --- a/http/httphi/bench_test.go +++ b/http/httphi/bench_test.go @@ -43,7 +43,11 @@ func benchExchange(b *testing.B, conn conn) *Exchange { const bufferSize = 1024 const numHeaderCap = 2 exch := new(Exchange) - exch.Configure(make([]byte, 2*bufferSize), bufferSize, numHeaderCap, false) + exch.Configure(ExchangeConfig{ + RawBuf: make([]byte, 2*bufferSize), + RequestBufferLim: bufferSize, + NumHeaderKVCap: numHeaderCap, + }) if !exch.Acquire(conn) { b.Fatal("fresh exchange failed to acquire connection") } diff --git a/http/httphi/exchange.go b/http/httphi/exchange.go index 49063a5..df42a0b 100644 --- a/http/httphi/exchange.go +++ b/http/httphi/exchange.go @@ -51,8 +51,9 @@ type Exchange struct { type ExchangeConfig struct { RawBuf []byte RequestBufferLim int - NumHeaderCap int + NumHeaderKVCap int NormalizeOutgoingKeys bool + NoRequestBufferGrowth bool } // HijackRaw is a low-level implementation of http.Hijacker interface. @@ -95,8 +96,8 @@ func (exch *Exchange) Configure(cfg ExchangeConfig) { panic("request lim larger than buffer") } exch.rawbuf = cfg.RawBuf - exch.reqHdr.Reset(cfg.RawBuf[:0:cfg.RequestBufferLim], cfg.NumHeaderCap) - exch.reqHdr.ConfigBufferGrowth(false) + exch.reqHdr.Reset(cfg.RawBuf[:0:cfg.RequestBufferLim], cfg.NumHeaderKVCap) + exch.reqHdr.ConfigBufferGrowth(!cfg.NoRequestBufferGrowth) exch.normalizeKeys = cfg.NormalizeOutgoingKeys } diff --git a/http/httphi/exchange_test.go b/http/httphi/exchange_test.go index 6d1fa34..1296556 100644 --- a/http/httphi/exchange_test.go +++ b/http/httphi/exchange_test.go @@ -19,11 +19,18 @@ import ( func nopBackoff(consecutiveBackoffs uint) time.Duration { return lneto.BackoffFlagNop } +// defaultNumHeaderKVCap is the field table tests get unless they set their own: +// room for a realistic request, the sizing being what the test is about only +// where it says so. +const defaultNumHeaderKVCap = 32 + // newExchange returns an Exchange acquired on conn, ready to serve a request. func newExchange(t *testing.T, conn conn, cfg ExchangeConfig) *Exchange { t.Helper() exch := new(Exchange) - const numHeaderCap = 1 + if cfg.NumHeaderKVCap == 0 { + cfg.NumHeaderKVCap = defaultNumHeaderKVCap + } exch.Configure(cfg) if !exch.Acquire(conn) { t.Fatal("fresh exchange failed to acquire connection") @@ -37,7 +44,10 @@ func serve(t *testing.T, request string, mux Mux) *rwconn { const bufferSize = 1024 conn := newConn(request) conn.Hangup() // Whole request already pending, nothing more will arrive. - exch := newExchange(t, conn, bufferSize, false) + exch := newExchange(t, conn, ExchangeConfig{ + RawBuf: make([]byte, 2*bufferSize), + RequestBufferLim: bufferSize, + }) err := Handle(exch, mux, nopBackoff) if err != nil { t.Fatalf("Handle(%q): %s", request, err) @@ -152,7 +162,7 @@ func TestExchangeSetHeader(t *testing.T) { func TestExchangeSetHeaderOOM(t *testing.T) { const bufferSize = 32 conn := newConn("") - exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, bufferSize), RequestBufferLim: 64}) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*bufferSize), RequestBufferLim: bufferSize}) if exch.StageHeader("X-Big", strings.Repeat("v", 4*bufferSize)) { t.Fatal("want insufficient memory reported for oversized header value") } @@ -240,7 +250,7 @@ func TestHandleMalformedRequest(t *testing.T) { sm.Handle("/", func(ex *Exchange) { handled = true }) conn := newConn(test.request) conn.Hangup() - exch := newExchange(t, conn, 1024, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024}) if err := Handle(exch, &sm, nopBackoff); err == nil { t.Error("want error on malformed request, got nil") } @@ -281,7 +291,7 @@ var _ io.ReadWriteCloser = (*ExchangeRW)(nil) func TestExchangeRW(t *testing.T) { const body = "hello" conn := newConn("") - exch := newExchange(t, conn, 128, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*128), RequestBufferLim: 128}) var rw ExchangeRW exch.ReadWriter(&rw) @@ -302,7 +312,7 @@ func TestExchangeRW(t *testing.T) { // taken from must fail instead of reaching the next request's connection. func TestExchangeRWOutlivesExchange(t *testing.T) { conn := newConn("") - exch := newExchange(t, conn, 128, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*128), RequestBufferLim: 128}) var rw ExchangeRW exch.ReadWriter(&rw) if !rw.IsValid() { @@ -349,13 +359,19 @@ func TestExchangeReadBody(t *testing.T) { // FlushHeader appends after the last field. Buffers that fit all but the last // byte must be refused, never overrun. func TestExchangeStageOKAndFail(t *testing.T) { - const key, value = "K", "V" + // Long enough that the buffers under test clear the smallest a header + // buffer may be, while still being sized to the byte around the field. + const key, value = "X-A-Header-Field-Key", "a-header-field-value" const field = len(key) + len(value) + len(":\r\n") const numHeaderCap = 4 for _, bufLen := range []int{field + 2, field + 1, field} { conn := newConn("") exch := new(Exchange) - exch.Configure(make([]byte, bufLen), bufLen, numHeaderCap, false) + exch.Configure(ExchangeConfig{ + RawBuf: make([]byte, bufLen), + RequestBufferLim: bufLen, + NumHeaderKVCap: numHeaderCap, + }) if !exch.Acquire(conn) { t.Fatal("fresh exchange failed to acquire connection") } @@ -394,7 +410,7 @@ func TestHandleLeavesConnOpen(t *testing.T) { } { conn := newConn(request) conn.Hangup() - exch := newExchange(t, conn, 1024, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024}) Handle(exch, &sm, nopBackoff) if conn.IsClosed() { t.Errorf("Handle closed the connection for %q", request) @@ -413,7 +429,7 @@ func TestExchangeHijackOwnership(t *testing.T) { }) first := newConn("GET / HTTP/1.1\r\nHost: h\r\n\r\n") first.Hangup() - exch := newExchange(t, first, 1024, false) + exch := newExchange(t, first, ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024}) if err := Handle(exch, &sm, nopBackoff); err != nil { t.Fatal(err) } @@ -443,7 +459,7 @@ func TestHandleIdlePeerEndsOnConnDeadline(t *testing.T) { sm.Handle("/", func(ex *Exchange) { t.Error("handler must not run on partial request") }) conn := newConn("GET / HTTP") // Peer stalls mid request line, never hangs up. conn.SetDeadline(time.Now().Add(10 * time.Millisecond)) - exch := newExchange(t, conn, 1024, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024}) done := make(chan error, 1) go func() { done <- Handle(exch, &sm, nopBackoff) }() @@ -462,7 +478,7 @@ func TestHandleIdlePeerEndsOnConnDeadline(t *testing.T) { func TestExchangeWriteHeaderFlushFails(t *testing.T) { const body = "body" conn := newConn("") - exch := newExchange(t, conn, 128, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*128), RequestBufferLim: 128}) conn.FailWrites(1) // Status line write fails, body write would succeed. n, err := exch.WriteBody([]byte(body)) @@ -501,7 +517,7 @@ func TestExchangeSetHeaderInt(t *testing.T) { {value: 1, base: 37, want: "\r\n"}, // Above base 36, dropped. } { conn := newConn("") - exch := newExchange(t, conn, 256, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*256), RequestBufferLim: 256}) exch.StageHeaderInt("N", test.value, test.base) exch.WriteHeader(200) got, _ := strings.CutPrefix(conn.ViewWritten(), "HTTP/1.1 200 OK\r\n") @@ -513,7 +529,7 @@ func TestExchangeSetHeaderInt(t *testing.T) { // SetHeaderInt must format into the response buffer without allocating. func TestExchangeSetHeaderIntNoAlloc(t *testing.T) { - exch := newExchange(t, newConn(""), 256, false) + exch := newExchange(t, newConn(""), ExchangeConfig{RawBuf: make([]byte, 2*256), RequestBufferLim: 256}) allocs := testing.AllocsPerRun(100, func() { exch.StageHeaderInt("Content-Length", 1234567890, 10) }) @@ -729,7 +745,7 @@ func TestExchangeRequestParseFormSplit(t *testing.T) { sm.Handle("/f", func(exch *Exchange) { gotErr = exch.RequestParseForm(&form, make([]byte, 64), nopBackoff) }) - exch := newExchange(t, conn, 1024, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024}) if err := Handle(exch, &sm, nopBackoff); err != nil { t.Fatal(err) } @@ -801,7 +817,7 @@ func serveMultipart(t *testing.T, request string, bufSize int, skip string, segm parts, gotErr = exch.ReadMultiparts(parts, make([]byte, bufSize), newSink) }) const x = unsafe.Sizeof(http.Request{}) - exch := newExchange(t, conn, 1024, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024}) if err := Handle(exch, &sm, nopBackoff); err != nil { t.Fatal(err) } @@ -978,7 +994,7 @@ func TestHandleBrowserSizedRequest(t *testing.T) { }) conn := newConn(request) conn.Hangup() - exch := newExchange(t, conn, 8192, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*8192), RequestBufferLim: 8192}) if err := Handle(exch, &sm, nopBackoff); err != nil { t.Fatalf("serving a browser sized request: %s", err) } @@ -991,33 +1007,60 @@ func TestHandleBrowserSizedRequest(t *testing.T) { } } -// A request with more header fields than the exchange has room for must be -// answered, not dropped: the peer learns its request was too large instead of -// seeing the connection go away. -func TestHandleTooManyHeaderFields(t *testing.T) { +// A request larger than the exchange has room for must be answered, not +// dropped: the peer learns its request was too large instead of seeing the +// connection go away. Either bound may be the one it ran into, and both are +// reported to the caller. +func TestHandleRequestTooLargeAnswers431(t *testing.T) { request := "GET /echo HTTP/1.1\r\nHost: lneto.test\r\n" for i := 0; i < 512; i++ { request += "H" + strconv.Itoa(i) + ":v\r\n" } request += "\r\n" - var served bool - var sm MuxSlice - sm.Reset(1) - sm.Handle("GET /echo", func(exch *Exchange) { served = true }) - conn := newConn(request) - conn.Hangup() - exch := newExchange(t, conn, 1024, false) - err := Handle(exch, &sm, nopBackoff) - if err != nil { - t.Fatal(err) - } - if served { - t.Fatal("handler ran on a request the parser could not hold") - } - got := conn.ViewWritten() - if !strings.HasPrefix(got, "HTTP/1.1 431 ") { - t.Errorf("want a 431 answer, got %q", firstLine(got)) + for _, test := range []struct { + name string + cfg ExchangeConfig + wantErr error + }{ + { + // Room for every byte of the block, but not for its fields. The + // field table only bounds the request when growth is refused: + // otherwise it is the size the parser starts from, not a limit. + name: "more fields than the table holds", + cfg: ExchangeConfig{ + RawBuf: make([]byte, 16*1024), RequestBufferLim: 8 * 1024, + NumHeaderKVCap: 16, NoRequestBufferGrowth: true, + }, + wantErr: httpraw.ErrHeaderTooMany, + }, + { + // Room for the fields, but not for the bytes they arrive in. + name: "more bytes than the buffer holds", + cfg: ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024, NumHeaderKVCap: 1024, NoRequestBufferGrowth: true}, + wantErr: httpraw.ErrSmallHeaderBuffer, + }, + } { + t.Run(test.name, func(t *testing.T) { + var served bool + var sm MuxSlice + sm.Reset(1) + sm.Handle("GET /echo", func(exch *Exchange) { served = true }) + conn := newConn(request) + conn.Hangup() + exch := newExchange(t, conn, test.cfg) + err := Handle(exch, &sm, nopBackoff) + if !errors.Is(err, test.wantErr) { + t.Fatalf("want %v reported to the caller, got %v", test.wantErr, err) + } + if served { + t.Fatal("handler ran on a request the parser could not hold") + } + got := conn.ViewWritten() + if !strings.HasPrefix(got, "HTTP/1.1 431 ") { + t.Errorf("want a 431 answer, got %q", firstLine(got)) + } + }) } } @@ -1047,7 +1090,7 @@ func TestExchangeReadBodyDoesNotReadPastWhatArrived(t *testing.T) { // The peer is still there, waiting to be answered: a read for bytes it is // not going to send blocks, exactly as it does on a socket. conn := &blockingConn{request: "POST / HTTP/1.1\r\nHost: h\r\nContent-Length: 12\r\n\r\n" + body} - exch := newExchange(t, conn, 1024, false) + exch := newExchange(t, conn, ExchangeConfig{RawBuf: make([]byte, 2*1024), RequestBufferLim: 1024}) done := make(chan struct{}) go func() { defer close(done) diff --git a/http/httphi/mux.go b/http/httphi/mux.go index 8d66a6e..6565eee 100644 --- a/http/httphi/mux.go +++ b/http/httphi/mux.go @@ -20,6 +20,7 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error { n, err := reqhdr.ReadFromLimited(exch.rw, reqhdr.BufferFree()) if err != nil { exch.readErr = err + exch.handleError(err) return err } else if n == 0 { backoff.Do(consecutiveBackoffs) @@ -32,12 +33,7 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error { if needMore { continue // Request header split across reads, accumulate the rest. } else if err != nil { - if err == httpraw.ErrHeaderTooMany || err == httpraw.ErrSmallHeaderBuffer { - // The peer is owed an answer: no larger buffer is coming, so - // say so instead of dropping the connection, RFC 6585 5. - exch.StageHeader("Content-Length", "0") - exch.WriteHeader(int(StatusRequestHeaderFieldsTooLarge)) - } + exch.handleError(err) return err } break // Done! @@ -68,6 +64,15 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error { return nil } +func (exch *Exchange) handleError(err error) { + if err == httpraw.ErrHeaderTooMany || err == httpraw.ErrSmallHeaderBuffer || exch.reqHdr.BufferFree() == 0 { + // The peer is owed an answer: no larger buffer is coming, so + // say so instead of dropping the connection, RFC 6585 5. + exch.StageHeader("Content-Length", "0") + exch.WriteHeader(int(StatusRequestHeaderFieldsTooLarge)) + } +} + // HandlerFunc serves a single request, playing the part of http.Handler. // The exchange is only valid for the duration of the call: it is released to // the router's pool on return, so a handler must not retain it nor any slice it diff --git a/http/httphi/router.go b/http/httphi/router.go index 08a9f32..24d0ca5 100644 --- a/http/httphi/router.go +++ b/http/httphi/router.go @@ -81,7 +81,7 @@ type RouterConfig struct { // After memory is fully consumed [Exchange.StageHeader] will not append more headers. ResponseHeaderMinBufferSize int // Number of request header key/value pairs to parse before failing and returning [StatusRequestHeaderFieldsTooLarge]. - RequestNumHeaderCap int + RequestNumHeaderKVCap int // NormalizeOutgoingKeys normalizes response header field keys as they are // staged, i.e: "content-type" becomes "Content-Type". @@ -106,6 +106,7 @@ func (cfg RouterConfig) Validate() error { workerMode := cfg.workerMode() if workerMode && cfg.MaxAwaitingConns == 0 || cfg.Mux == nil || + cfg.RequestNumHeaderKVCap <= 0 || !workerMode && cfg.FixedNumGoroutines != -1 { return lneto.ErrInvalidConfig } else if cfg.Backoff == nil { @@ -157,7 +158,7 @@ func (r *Router) Configure(cfg RouterConfig) error { gen := r.gen.Load() numgoro := cfg.FixedNumGoroutines workerMode := cfg.workerMode() - r.reqNumHeaderCap = cfg.RequestNumHeaderCap + r.reqNumHeaderCap = cfg.RequestNumHeaderKVCap r.reqBuf = cfg.RequestHeaderBufferSize r.respBuf = cfg.ResponseHeaderMinBufferSize r.mux = cfg.Mux @@ -191,8 +192,9 @@ func (r *Router) Configure(cfg RouterConfig) error { r.exchs[i].Configure(ExchangeConfig{ RawBuf: r.globbuf[goff : goff+rawBuflen], RequestBufferLim: cfg.RequestHeaderBufferSize, - NumHeaderCap: cfg.RequestNumHeaderCap, + NumHeaderKVCap: cfg.RequestNumHeaderKVCap, NormalizeOutgoingKeys: cfg.NormalizeOutgoingKeys, + NoRequestBufferGrowth: true, // Hard memory limit. }) go r.goroWorker(gen, jobqueue, cfg.Backoff, cfg.Mux) } @@ -334,13 +336,16 @@ func (r *Router) getExchLocked(conn conn) (exch *Exchange) { } } - if r.numGoro == -1 { + // Unbounded mode is stored as zero, see [Router.Configure]: a router that + // spawns a goroutine per connection allocates the exchange to go with it. + if r.numGoro == 0 { exch := new(Exchange) exch.Configure(ExchangeConfig{ RawBuf: make([]byte, r.respBuf+r.reqBuf), RequestBufferLim: r.reqBuf, - NumHeaderCap: r.reqNumHeaderCap, + NumHeaderKVCap: r.reqNumHeaderCap, NormalizeOutgoingKeys: r.normalizeKeys, + NoRequestBufferGrowth: true, }) exch.Acquire(conn) // Fresh exchange, CAS cannot fail. return exch diff --git a/http/httphi/router_test.go b/http/httphi/router_test.go index 76989bb..e32c68e 100644 --- a/http/httphi/router_test.go +++ b/http/httphi/router_test.go @@ -156,6 +156,7 @@ func configSynchronousRouter(t *testing.T, router *Router, bufferSize int, mux M FixedNumGoroutines: -1, Mux: mux, RequestHeaderBufferSize: bufferSize, + RequestNumHeaderKVCap: 16, ResponseHeaderMinBufferSize: bufferSize, Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) { return lneto.BackoffFlagNop @@ -349,6 +350,7 @@ func TestRouterHandleAfterTeardown(t *testing.T) { MaxAwaitingConns: 4, Mux: &sm, RequestHeaderBufferSize: 512, + RequestNumHeaderKVCap: 16, ResponseHeaderMinBufferSize: 512, Backoff: nopBackoff, }) @@ -377,6 +379,7 @@ func TestRouterConfigureDuringWorkerHandle(t *testing.T) { MaxAwaitingConns: 4, Mux: &sm, RequestHeaderBufferSize: 512, + RequestNumHeaderKVCap: 16, ResponseHeaderMinBufferSize: 512, Backoff: nopBackoff, } diff --git a/http/httpraw/header.go b/http/httpraw/header.go index 868b449..b96d695 100644 --- a/http/httpraw/header.go +++ b/http/httpraw/header.go @@ -252,8 +252,8 @@ func (hb *headerBuf) forEach(cb func(key, value []byte) error) error { // data with Reset to parse data in-place. // If buf is nil then the current buffer is reused. There are 3 ways to use Reset: // -// h.Reset(prealloc[:0]); h.ParseBytes(httpHeader) // Tell header to use a pre-allocated buffer capacity. -// h.Reset(httpHeader); h.Parse() // Parse bytes in place with no copying. +// h.Reset(prealloc[:0], 16); h.ParseBytes(httpHeader) // Tell header to use a pre-allocated buffer capacity. +// h.Reset(httpHeader, 16); h.Parse() // Parse bytes in place with no copying. // h.Reset(nil) // Reuse buffer previously set in a call to Reset. func (h *Header) Reset(buf []byte, numHeaderCapacity int) { const persistentFlags = flagNoBufferGrow diff --git a/internet/pcap/capture.go b/internet/pcap/capture.go index 74dd8e8..f55abdb 100644 --- a/internet/pcap/capture.go +++ b/internet/pcap/capture.go @@ -768,6 +768,11 @@ func httpBodyClass(contentType, body []byte) FieldClass { return FieldClassText } +// maxCapturedHeaderFields bounds the field table a captured HTTP header is +// parsed into. Captures are read once and discarded, so the table is sized for +// a realistic request rather than for whatever the capture happens to hold. +const maxCapturedHeaderFields = 64 + func (pc *PacketBreakdown) CaptureHTTP(dst []Frame, pkt []byte, bitOffset int) ([]Frame, error) { debuglog("pcap:http:start") const httpProtocol = "HTTP" @@ -777,10 +782,10 @@ func (pc *PacketBreakdown) CaptureHTTP(dst []Frame, pkt []byte, bitOffset int) ( const asResponse = true const asRequest = false httpData := pkt[bitOffset/8:] - pc.hdr.Reset(httpData) + pc.hdr.Reset(httpData, maxCapturedHeaderFields) err := pc.hdr.Parse(asResponse) if err != nil { - pc.hdr.Reset(httpData) + pc.hdr.Reset(httpData, 0) // Field table already sized, reuse it. err = pc.hdr.Parse(asRequest) // try as request. } if err != nil {