From 7369da89f1774bb6c439d24f9833a07dd3d0aefc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A4=D0=B8=D0=BB=D0=B8=D0=BC=D0=BE=D0=BD=D0=B5=D0=BD?= =?UTF-8?q?=D0=BA=D0=BE=D0=B2=20=D0=9F=D0=B0=D0=B2=D0=B5=D0=BB=20=D0=90?= =?UTF-8?q?=D0=BD=D0=B0=D1=82=D0=BE=D0=BB=D1=8C=D0=B5=D0=B2=D0=B8=D1=87?= Date: Thu, 23 Jul 2026 13:09:48 +0300 Subject: [PATCH 1/5] feat: add lazy multipart form streaming MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a sequential multipart/form-data reader that exposes file payloads directly from the request body without background goroutines. Define explicit file close, stream drain, and request abort semantics while preserving the existing BindForm behavior. Signed-off-by: Филимоненков Павел Анатольевич --- multipart_stream.go | 363 +++++++++++++++++++++++++++++++++ multipart_stream_test.go | 423 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 786 insertions(+) create mode 100644 multipart_stream.go create mode 100644 multipart_stream_test.go diff --git a/multipart_stream.go b/multipart_stream.go new file mode 100644 index 00000000..60218747 --- /dev/null +++ b/multipart_stream.go @@ -0,0 +1,363 @@ +// SPDX-FileCopyrightText: Copyright 2015-2026 go-swagger maintainers +// SPDX-License-Identifier: Apache-2.0 + +package runtime + +import ( + "context" + stderrors "errors" + "fmt" + "io" + "mime/multipart" + "net/http" + "net/textproto" + "net/url" + + "github.com/go-openapi/errors" +) + +// MultipartFormStreamOption configures [NewMultipartFormStream]. +type MultipartFormStreamOption func(*multipartFormStreamConfig) + +type multipartFormStreamConfig struct { + maxBody int64 + maxFiles int + maxFilenameLen int +} + +// MultipartFormStreamMaxBody caps the total number of request-body bytes read +// by a [MultipartFormStream]. +// +// A value of 0 applies [DefaultMaxUploadBodySize]. A negative value disables +// the cap when the caller has already limited the request body upstream. +func MultipartFormStreamMaxBody(n int64) MultipartFormStreamOption { + return func(c *multipartFormStreamConfig) { c.maxBody = n } +} + +// MultipartFormStreamMaxFiles rejects a multipart stream after more than n +// file parts have been encountered. A value of 0 means no file-count cap. +func MultipartFormStreamMaxFiles(n int) MultipartFormStreamOption { + return func(c *multipartFormStreamConfig) { c.maxFiles = n } +} + +// MultipartFormStreamMaxFilenameLen rejects file parts whose filename exceeds +// n bytes. A value of 0 disables the limit. When this option is not supplied, +// [DefaultMaxUploadFilenameLength] is used. +func MultipartFormStreamMaxFilenameLen(n int) MultipartFormStreamOption { + return func(c *multipartFormStreamConfig) { c.maxFilenameLen = n } +} + +// StreamedFile exposes a file part directly from the multipart request body. +// +// Reads block until bytes arrive from the client. StreamedFile is not seekable +// and is not safe for concurrent use. Its form name, filename and MIME headers +// are available before the payload is consumed. +// +// Closing a StreamedFile drains only the unread remainder of that file part. +// Close may therefore block while the client is still uploading the current +// part. The owning [MultipartFormStream] may then advance to the next part. +type StreamedFile struct { + FieldName string + Filename string + Header textproto.MIMEHeader + part *multipart.Part + closeErr error +} + +// Read reads file payload bytes directly from the request body. +func (f *StreamedFile) Read(p []byte) (int, error) { + if f == nil || f.part == nil { + return 0, io.ErrClosedPipe + } + + return f.part.Read(p) +} + +// Close discards the unread remainder of this file part. +// +// Close does not close the underlying HTTP request body and does not consume +// subsequent multipart parts. After Close returns successfully, the parent +// MultipartFormStream may advance to the next part. +// +// Any error encountered while discarding the unread payload is returned. +func (f *StreamedFile) Close() error { + if f == nil { + return nil + } + if f.part == nil { + return f.closeErr + } + + part := f.part + f.part = nil + // multipart.Part.Close drains with io.Copy but intentionally discards the + // resulting error, so drain explicitly to preserve error propagation. + _, f.closeErr = io.Copy(io.Discard, part) + + return f.closeErr +} + +// MultipartFormStream reads multipart/form-data sequentially without parsing +// the complete request body before exposing file payloads. +// +// [MultipartFormStream.NextFile] consumes ordinary form fields until it reaches +// the next file part. Fields are appended to request.PostForm and request.Form +// as they are encountered. Consequently, fields after a file become visible +// only after the caller consumes or closes that file and advances the stream. +// +// A stream and its returned files are not safe for concurrent use. There is at +// most one active file part. Calling [MultipartFormStream.NextFile] closes and +// drains an unread active file before advancing, and may therefore block until +// the current part finishes arriving. No background goroutines are started. +// +// The caller owns the stream. Call [MultipartFormStream.Drain] to consume the +// remaining body, collect trailing fields and allow HTTP connection reuse when +// possible. Call [MultipartFormStream.Close] to abort immediately without +// draining. +type MultipartFormStream struct { + request *http.Request + reader *multipart.Reader + queryValues url.Values + current *StreamedFile + maxFiles int + maxFilenameLen int + files int + closed bool + done bool +} + +// NewMultipartFormStream creates a sequential multipart/form-data stream over +// r.Body. +// +// The constructor validates the media type and boundary but does not consume +// multipart parts. It initializes request.Form and request.PostForm in the same +// way as [http.Request.ParseForm], then populates multipart values incrementally +// as [MultipartFormStream.NextFile] advances. +// +// NewMultipartFormStream marks the request as handled by MultipartReader. +// Callers must not subsequently call [http.Request.ParseMultipartForm] or +// [BindForm] for the same request. +// +// File payloads are exposed directly from the request body and are not buffered +// in memory or temporary files. Ordinary form values are read into memory as +// they are encountered. Parts are processed in wire order. +// +// At most one StreamedFile may be active at a time. Calling +// [MultipartFormStream.NextFile] automatically closes and drains an unread +// current file before advancing. No background goroutines are started. +// +// MultipartFormStream is not safe for concurrent use. +func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) (*MultipartFormStream, error) { + cfg := multipartFormStreamConfig{ + maxFilenameLen: DefaultMaxUploadFilenameLength, + } + for _, opt := range opts { + opt(&cfg) + } + + if r == nil { + return nil, errors.NewParseError("body", "formData", "", stderrors.New("nil request")) + } + if r.Body == nil { + return nil, errors.NewParseError("body", "formData", "", stderrors.New("nil request body")) + } + + body := r.Body + if cfg.maxBody >= 0 { + maxBody := cfg.maxBody + if maxBody == 0 { + maxBody = DefaultMaxUploadBodySize + } + body = http.MaxBytesReader(nil, body, maxBody) + } + body = &contextReadCloser{ctx: r.Context(), ReadCloser: body} + r.Body = body + + reader, err := r.MultipartReader() + if err != nil { + return nil, errors.NewParseError("body", "formData", "", err) + } + if err = r.ParseForm(); err != nil { + return nil, errors.NewParseError("body", "formData", "", err) + } + + return &MultipartFormStream{ + request: r, + reader: reader, + queryValues: cloneFormValues(r.Form), + maxFiles: cfg.maxFiles, + maxFilenameLen: cfg.maxFilenameLen, + }, nil +} + +// NextFile advances through the multipart body and returns the next file part. +// Ordinary form fields encountered before that file are added to request.Form +// and request.PostForm. +// +// If the previously returned file is still open, NextFile closes and drains it +// before advancing. This may block while the client is still uploading that +// part. Any drain error is returned and the stream is aborted. +// +// NextFile returns io.EOF when no file parts remain. At that point all trailing +// ordinary fields have been collected. +func (s *MultipartFormStream) NextFile() (*StreamedFile, error) { + if s == nil { + return nil, io.ErrClosedPipe + } + if s.done { + return nil, io.EOF + } + if s.closed { + return nil, io.ErrClosedPipe + } + if err := s.closeCurrent(); err != nil { + return nil, s.abort(err) + } + + for { + part, err := s.reader.NextPart() + if err != nil { + if stderrors.Is(err, io.EOF) { + s.done = true + + return nil, io.EOF + } + + return nil, s.abort(err) + } + + fieldName := part.FormName() + filename := part.FileName() + if filename == "" { + if err := s.bindValue(part, fieldName); err != nil { + return nil, s.abort(err) + } + + continue + } + + s.files++ + if s.maxFiles > 0 && s.files > s.maxFiles { + return nil, s.abort(errors.NewParseError("body", "formData", "", + fmt.Errorf("multipart form contains %d file parts, exceeds limit %d", s.files, s.maxFiles))) + } + if err := ValidateFilenameLength(fieldName, "formData", filename, s.maxFilenameLen); err != nil { + return nil, s.abort(err) + } + + file := &StreamedFile{ + FieldName: fieldName, + Filename: filename, + Header: part.Header, + part: part, + } + s.current = file + + return file, nil + } +} + +// Drain consumes the rest of the multipart body. +// +// Unread file payloads are discarded. Non-file form fields encountered while +// draining are collected in the request form values. +// +// Drain closes the underlying request body after reaching EOF. Subsequent calls +// to NextFile return io.EOF. Drain returns any multipart parsing, payload drain +// or request-body close error. +func (s *MultipartFormStream) Drain() error { + if s == nil || s.closed { + return nil + } + + for { + file, err := s.NextFile() + if stderrors.Is(err, io.EOF) { + return s.Close() + } + if err != nil { + return stderrors.Join(err, s.Close()) + } + if err := file.Close(); err != nil { + return stderrors.Join(err, s.Close()) + } + } +} + +// Close closes the underlying HTTP request body without draining it. +// +// Close aborts further multipart processing. Call Drain instead when the +// remaining parts must be consumed, for example to collect trailing form +// fields or improve the chance of HTTP connection reuse. +func (s *MultipartFormStream) Close() error { + if s == nil || s.closed { + return nil + } + + s.closed = true + if s.current != nil { + s.current.part = nil + s.current = nil + } + + return s.request.Body.Close() +} + +func (s *MultipartFormStream) abort(err error) error { + return stderrors.Join(err, s.Close()) +} + +func (s *MultipartFormStream) closeCurrent() error { + if s.current == nil { + return nil + } + + err := s.current.Close() + s.current = nil + + return err +} + +func (s *MultipartFormStream) bindValue(part *multipart.Part, name string) error { + defer part.Close() + + value, err := io.ReadAll(part) + if err != nil { + return err + } + if name == "" { + return nil + } + + s.request.PostForm.Add(name, string(value)) + postValues := s.request.PostForm[name] + combined := make([]string, 0, len(postValues)+len(s.queryValues[name])) + combined = append(combined, postValues...) + combined = append(combined, s.queryValues[name]...) + s.request.Form[name] = combined + + return nil +} + +func cloneFormValues(values url.Values) url.Values { + cloned := make(url.Values, len(values)) + for name, entries := range values { + cloned[name] = append([]string(nil), entries...) + } + + return cloned +} + +type contextReadCloser struct { + io.ReadCloser + + ctx context.Context //nolint:containedctx // Read has no context parameter, so the wrapper must retain it +} + +func (r *contextReadCloser) Read(p []byte) (int, error) { + if err := r.ctx.Err(); err != nil { + return 0, err + } + + return r.ReadCloser.Read(p) +} diff --git a/multipart_stream_test.go b/multipart_stream_test.go new file mode 100644 index 00000000..e605250e --- /dev/null +++ b/multipart_stream_test.go @@ -0,0 +1,423 @@ +// SPDX-FileCopyrightText: Copyright 2015-2026 go-swagger maintainers +// SPDX-License-Identifier: Apache-2.0 + +package runtime + +import ( + "bytes" + "context" + stderrors "errors" + "io" + "mime/multipart" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/go-openapi/errors" + "github.com/go-openapi/testify/v2/assert" + "github.com/go-openapi/testify/v2/require" +) + +const streamedFilename = "payload.bin" + +func TestMultipartFormStreamExposesFileBeforeBodyCompletes(t *testing.T) { + pipeReader, pipeWriter := io.Pipe() + writer := multipart.NewWriter(pipeWriter) + paused := make(chan struct{}) + resume := make(chan struct{}) + writeErr := make(chan error, 1) + + go func() { + defer close(writeErr) + if err := writer.WriteField("before", "value"); err != nil { + writeErr <- err + return + } + part, err := writer.CreateFormFile(testFieldFile, streamedFilename) + if err != nil { + writeErr <- err + return + } + if _, err := io.WriteString(part, "first"); err != nil { + writeErr <- err + return + } + close(paused) + <-resume + if _, err := io.WriteString(part, "second"); err != nil { + writeErr <- err + return + } + if err := writer.Close(); err != nil { + writeErr <- err + return + } + if err := pipeWriter.Close(); err != nil { + writeErr <- err + } + }() + + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, pipeReader) + request.Header.Set(HeaderContentType, writer.FormDataContentType()) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + file, err := stream.NextFile() + require.NoError(t, err) + assert.EqualT(t, streamedFilename, file.Filename) + assert.EqualT(t, "value", request.Form.Get("before")) + + first := make([]byte, len("first")) + _, err = io.ReadFull(file, first) + require.NoError(t, err) + assert.EqualT(t, "first", string(first)) + <-paused + + readResult := make(chan struct { + data string + err error + }, 1) + go func() { + data, readErr := io.ReadAll(file) + readResult <- struct { + data string + err error + }{data: string(data), err: readErr} + }() + + select { + case result := <-readResult: + t.Fatalf("file read completed while client was paused: data=%q err=%v", result.data, result.err) + case <-time.After(100 * time.Millisecond): + } + + close(resume) + result := <-readResult + require.NoError(t, result.err) + assert.EqualT(t, "second", result.data) + require.NoError(t, stream.Drain()) + for err := range writeErr { + require.NoError(t, err) + } +} + +func TestMultipartFormStreamPreservesPartOrderAndFields(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedField{name: "before", value: "one"}, + orderedFile{field: testFieldFile1, filename: testFileFieldA, content: "AAA"}, + orderedField{name: "between", value: "two"}, + orderedFile{field: testFieldFile2, filename: testFileFieldB, content: "BBBB"}, + orderedField{name: "after", value: "three"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + first, err := stream.NextFile() + require.NoError(t, err) + assert.EqualT(t, testFieldFile1, first.FieldName) + assert.EqualT(t, "one", request.Form.Get("before")) + assert.Empty(t, request.Form.Get("between")) + content, err := io.ReadAll(first) + require.NoError(t, err) + assert.EqualT(t, "AAA", string(content)) + + second, err := stream.NextFile() + require.NoError(t, err) + assert.EqualT(t, testFieldFile2, second.FieldName) + assert.EqualT(t, "two", request.Form.Get("between")) + assert.Empty(t, request.Form.Get("after")) + content, err = io.ReadAll(second) + require.NoError(t, err) + assert.EqualT(t, "BBBB", string(content)) + + _, err = stream.NextFile() + require.ErrorIs(t, err, io.EOF) + assert.EqualT(t, "three", request.Form.Get("after")) +} + +func TestMultipartFormStreamBodyValuesPrecedeQueryValues(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedField{name: "shared", value: "body"}, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath+"?shared=query", body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + _, err = stream.NextFile() + require.NoError(t, err) + assert.Equal(t, []string{"body", "query"}, request.Form["shared"]) + assert.EqualT(t, "body", request.Form.Get("shared")) +} + +func TestMultipartFormStreamNextFileDrainsPreviousFile(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile1, filename: testFileFieldA, content: "AAAA"}, + orderedFile{field: testFieldFile2, filename: testFileFieldB, content: "BBBB"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + first, err := stream.NextFile() + require.NoError(t, err) + one := make([]byte, 1) + _, err = io.ReadFull(first, one) + require.NoError(t, err) + + second, err := stream.NextFile() + require.NoError(t, err) + assert.EqualT(t, testFileFieldB, second.Filename) + _, err = first.Read(one) + require.ErrorIs(t, err, io.ErrClosedPipe) +} + +func TestStreamedFileCloseReturnsDrainError(t *testing.T) { + pipeReader, pipeWriter := io.Pipe() + writer := multipart.NewWriter(pipeWriter) + wantErr := stderrors.New("injected body read failure") + writeDone := make(chan struct{}) + + go func() { + defer close(writeDone) + part, err := writer.CreateFormFile(testFieldFile, streamedFilename) + if err != nil { + _ = pipeWriter.CloseWithError(err) + + return + } + if _, err := io.WriteString(part, "payload"); err != nil { + _ = pipeWriter.CloseWithError(err) + + return + } + _ = pipeWriter.CloseWithError(wantErr) + }() + + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, pipeReader) + request.Header.Set(HeaderContentType, writer.FormDataContentType()) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + file, err := stream.NextFile() + require.NoError(t, err) + require.ErrorIs(t, file.Close(), wantErr) + _, err = stream.NextFile() + require.ErrorIs(t, err, wantErr) + <-writeDone +} + +func TestMultipartFormStreamDrainCollectsTrailingFieldsAndClosesBody(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + orderedField{name: "after", value: "value"}, + ) + trackedBody := &observableReadCloser{Reader: body} + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, nil) + request.Body = trackedBody + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + + _, err = stream.NextFile() + require.NoError(t, err) + require.NoError(t, stream.Drain()) + assert.EqualT(t, "value", request.Form.Get("after")) + assert.TrueT(t, trackedBody.Closed()) + _, err = stream.NextFile() + require.ErrorIs(t, err, io.EOF) +} + +func TestMultipartFormStreamCloseAbortsWithoutDraining(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + orderedField{name: "after", value: "value"}, + ) + trackedBody := &observableReadCloser{Reader: body} + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, nil) + request.Body = trackedBody + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + + file, err := stream.NextFile() + require.NoError(t, err) + require.NoError(t, stream.Close()) + assert.TrueT(t, trackedBody.Closed()) + assert.Empty(t, request.Form.Get("after")) + _, err = file.Read(make([]byte, 1)) + require.ErrorIs(t, err, io.ErrClosedPipe) +} + +func TestMultipartFormStreamBodyLimit(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile, filename: streamedFilename, content: strings.Repeat("x", 1024)}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request, MultipartFormStreamMaxBody(512)) + require.NoError(t, err) + defer stream.Close() + + file, err := stream.NextFile() + require.NoError(t, err) + _, err = io.ReadAll(file) + require.Error(t, err) + var maxBytesErr *http.MaxBytesError + require.True(t, stderrors.As(err, &maxBytesErr), "expected *http.MaxBytesError, got %T", err) +} + +func TestMultipartFormStreamMalformedBody(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + ) + truncated := body.Bytes()[:body.Len()-5] + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, bytes.NewReader(truncated)) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + file, err := stream.NextFile() + require.NoError(t, err) + _, err = io.ReadAll(file) + require.Error(t, err) +} + +func TestMultipartFormStreamContextCancellation(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + ) + ctx, cancel := context.WithCancel(context.Background()) + request := httptest.NewRequestWithContext(ctx, http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + cancel() + _, err = stream.NextFile() + require.ErrorIs(t, err, context.Canceled) +} + +func TestMultipartFormStreamLimits(t *testing.T) { + t.Run("filename", func(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile, filename: "too-long.txt", content: "payload"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request, MultipartFormStreamMaxFilenameLen(4)) + require.NoError(t, err) + defer stream.Close() + + _, err = stream.NextFile() + var parseErr *errors.ParseError + require.True(t, stderrors.As(err, &parseErr), "expected *errors.ParseError, got %T", err) + assert.EqualT(t, testFieldFile, parseErr.Name) + }) + + t.Run("files", func(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile1, filename: testFileFieldA, content: "A"}, + orderedFile{field: testFieldFile2, filename: testFileFieldB, content: "B"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request, MultipartFormStreamMaxFiles(1)) + require.NoError(t, err) + defer stream.Close() + + first, err := stream.NextFile() + require.NoError(t, err) + _, err = io.ReadAll(first) + require.NoError(t, err) + _, err = stream.NextFile() + var parseErr *errors.ParseError + require.True(t, stderrors.As(err, &parseErr), "expected *errors.ParseError, got %T", err) + }) +} + +func TestNewMultipartFormStreamRejectsNonMultipartRequest(t *testing.T) { + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, strings.NewReader("value=x")) + request.Header.Set(HeaderContentType, URLencodedFormMime) + + _, err := NewMultipartFormStream(request) + var parseErr *errors.ParseError + require.True(t, stderrors.As(err, &parseErr), "expected *errors.ParseError, got %T", err) +} + +type orderedMultipartPart interface { + writeTo(*testing.T, *multipart.Writer) +} + +type orderedField struct { + name string + value string +} + +func (f orderedField) writeTo(t *testing.T, writer *multipart.Writer) { + t.Helper() + require.NoError(t, writer.WriteField(f.name, f.value)) +} + +type orderedFile struct { + field string + filename string + content string +} + +func (f orderedFile) writeTo(t *testing.T, writer *multipart.Writer) { + t.Helper() + part, err := writer.CreateFormFile(f.field, f.filename) + require.NoError(t, err) + _, err = io.WriteString(part, f.content) + require.NoError(t, err) +} + +func orderedMultipartBody(t *testing.T, parts ...orderedMultipartPart) (*bytes.Buffer, string) { + t.Helper() + body := &bytes.Buffer{} + writer := multipart.NewWriter(body) + for _, part := range parts { + part.writeTo(t, writer) + } + require.NoError(t, writer.Close()) + + return body, writer.FormDataContentType() +} + +type observableReadCloser struct { + io.Reader + + mu sync.Mutex + closed bool +} + +func (r *observableReadCloser) Close() error { + r.mu.Lock() + defer r.mu.Unlock() + r.closed = true + + return nil +} + +func (r *observableReadCloser) Closed() bool { + r.mu.Lock() + defer r.mu.Unlock() + + return r.closed +} From 04bdf18c13157d666019941c5c4ceac3a40d5ff4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A4=D0=B8=D0=BB=D0=B8=D0=BC=D0=BE=D0=BD=D0=B5=D0=BD?= =?UTF-8?q?=D0=BA=D0=BE=D0=B2=20=D0=9F=D0=B0=D0=B2=D0=B5=D0=BB=20=D0=90?= =?UTF-8?q?=D0=BD=D0=B0=D1=82=D0=BE=D0=BB=D1=8C=D0=B5=D0=B2=D0=B8=D1=87?= Date: Thu, 23 Jul 2026 15:27:22 +0300 Subject: [PATCH 2/5] fix: align multipart streaming with net/http semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Preserve multipart query precedence, ignore nameless parts, reject multipart/mixed requests, restore a total part limit, and document request body close behavior. Signed-off-by: Филимоненков Павел Анатольевич --- multipart_stream.go | 99 ++++++++++++++++++++++++++-------------- multipart_stream_test.go | 88 +++++++++++++++++++++++++++++++++-- 2 files changed, 151 insertions(+), 36 deletions(-) diff --git a/multipart_stream.go b/multipart_stream.go index 60218747..c746b6ee 100644 --- a/multipart_stream.go +++ b/multipart_stream.go @@ -8,10 +8,10 @@ import ( stderrors "errors" "fmt" "io" + "mime" "mime/multipart" "net/http" "net/textproto" - "net/url" "github.com/go-openapi/errors" ) @@ -19,9 +19,12 @@ import ( // MultipartFormStreamOption configures [NewMultipartFormStream]. type MultipartFormStreamOption func(*multipartFormStreamConfig) +const defaultMultipartFormStreamMaxParts = 1000 + type multipartFormStreamConfig struct { maxBody int64 maxFiles int + maxParts int maxFilenameLen int } @@ -40,6 +43,13 @@ func MultipartFormStreamMaxFiles(n int) MultipartFormStreamOption { return func(c *multipartFormStreamConfig) { c.maxFiles = n } } +// MultipartFormStreamMaxParts rejects a multipart stream after more than n +// total parts have been encountered. The default is 1000, matching +// [multipart.Reader.ReadForm]. A value of 0 disables the limit. +func MultipartFormStreamMaxParts(n int) MultipartFormStreamOption { + return func(c *multipartFormStreamConfig) { c.maxParts = n } +} + // MultipartFormStreamMaxFilenameLen rejects file parts whose filename exceeds // n bytes. A value of 0 disables the limit. When this option is not supplied, // [DefaultMaxUploadFilenameLength] is used. @@ -112,16 +122,17 @@ func (f *StreamedFile) Close() error { // // The caller owns the stream. Call [MultipartFormStream.Drain] to consume the // remaining body, collect trailing fields and allow HTTP connection reuse when -// possible. Call [MultipartFormStream.Close] to abort immediately without -// draining. +// possible. Call [MultipartFormStream.Close] to stop multipart processing +// without explicitly draining the remaining parts. type MultipartFormStream struct { request *http.Request reader *multipart.Reader - queryValues url.Values current *StreamedFile maxFiles int + maxParts int maxFilenameLen int files int + parts int closed bool done bool } @@ -129,9 +140,10 @@ type MultipartFormStream struct { // NewMultipartFormStream creates a sequential multipart/form-data stream over // r.Body. // -// The constructor validates the media type and boundary but does not consume -// multipart parts. It initializes request.Form and request.PostForm in the same -// way as [http.Request.ParseForm], then populates multipart values incrementally +// The constructor accepts only multipart/form-data, validates that a non-empty +// boundary is present, but does not consume multipart parts. It initializes +// request.Form and request.PostForm in the same way as +// [http.Request.ParseForm], then populates multipart values incrementally // as [MultipartFormStream.NextFile] advances. // // NewMultipartFormStream marks the request as handled by MultipartReader. @@ -147,9 +159,12 @@ type MultipartFormStream struct { // current file before advancing. No background goroutines are started. // // MultipartFormStream is not safe for concurrent use. +// +// File names and MIME headers are supplied by the client and remain untrusted. func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) (*MultipartFormStream, error) { cfg := multipartFormStreamConfig{ maxFilenameLen: DefaultMaxUploadFilenameLength, + maxParts: defaultMultipartFormStreamMaxParts, } for _, opt := range opts { opt(&cfg) @@ -162,6 +177,21 @@ func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) return nil, errors.NewParseError("body", "formData", "", stderrors.New("nil request body")) } + contentType := r.Header.Get(HeaderContentType) + mediaType, params, err := mime.ParseMediaType(contentType) + if err != nil { + return nil, errors.NewParseError(HeaderContentType, "header", contentType, err) + } + if mediaType != MultipartFormMime { + return nil, errors.NewParseError(HeaderContentType, "header", mediaType, http.ErrNotMultipart) + } + if params["boundary"] == "" { + return nil, errors.NewParseError(HeaderContentType, "header", contentType, http.ErrMissingBoundary) + } + if err = r.ParseForm(); err != nil { + return nil, errors.NewParseError("body", "formData", "", err) + } + body := r.Body if cfg.maxBody >= 0 { maxBody := cfg.maxBody @@ -177,15 +207,12 @@ func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) if err != nil { return nil, errors.NewParseError("body", "formData", "", err) } - if err = r.ParseForm(); err != nil { - return nil, errors.NewParseError("body", "formData", "", err) - } return &MultipartFormStream{ request: r, reader: reader, - queryValues: cloneFormValues(r.Form), maxFiles: cfg.maxFiles, + maxParts: cfg.maxParts, maxFilenameLen: cfg.maxFilenameLen, }, nil } @@ -226,10 +253,24 @@ func (s *MultipartFormStream) NextFile() (*StreamedFile, error) { return nil, s.abort(err) } + s.parts++ + if s.maxParts > 0 && s.parts > s.maxParts { + return nil, s.abort(errors.NewParseError("body", "formData", "", + fmt.Errorf("multipart form contains %d parts, exceeds limit %d", s.parts, s.maxParts))) + } + fieldName := part.FormName() + if fieldName == "" { + if err = discardPart(part); err != nil { + return nil, s.abort(err) + } + + continue + } + filename := part.FileName() if filename == "" { - if err := s.bindValue(part, fieldName); err != nil { + if err = s.bindValue(part, fieldName); err != nil { return nil, s.abort(err) } @@ -278,17 +319,19 @@ func (s *MultipartFormStream) Drain() error { if err != nil { return stderrors.Join(err, s.Close()) } - if err := file.Close(); err != nil { + if err = file.Close(); err != nil { return stderrors.Join(err, s.Close()) } } } -// Close closes the underlying HTTP request body without draining it. +// Close stops multipart processing and closes the underlying HTTP request body +// without explicitly draining the remaining multipart parts. // -// Close aborts further multipart processing. Call Drain instead when the -// remaining parts must be consumed, for example to collect trailing form -// fields or improve the chance of HTTP connection reuse. +// The concrete request body may perform its own work during Close. In +// particular, a net/http server request body may discard a limited amount of +// unread data to allow connection reuse, so Close is not guaranteed to return +// immediately. Call Drain when trailing form fields must be collected. func (s *MultipartFormStream) Close() error { if s == nil || s.closed { return nil @@ -319,33 +362,23 @@ func (s *MultipartFormStream) closeCurrent() error { } func (s *MultipartFormStream) bindValue(part *multipart.Part, name string) error { - defer part.Close() - value, err := io.ReadAll(part) if err != nil { return err } - if name == "" { - return nil - } s.request.PostForm.Add(name, string(value)) - postValues := s.request.PostForm[name] - combined := make([]string, 0, len(postValues)+len(s.queryValues[name])) - combined = append(combined, postValues...) - combined = append(combined, s.queryValues[name]...) - s.request.Form[name] = combined + // Match net/http.ParseMultipartForm: query values already present in Form + // keep precedence over multipart body values, which are appended. + s.request.Form.Add(name, string(value)) return nil } -func cloneFormValues(values url.Values) url.Values { - cloned := make(url.Values, len(values)) - for name, entries := range values { - cloned[name] = append([]string(nil), entries...) - } +func discardPart(part *multipart.Part) error { + _, err := io.Copy(io.Discard, part) - return cloned + return err } type contextReadCloser struct { diff --git a/multipart_stream_test.go b/multipart_stream_test.go index e605250e..a1938976 100644 --- a/multipart_stream_test.go +++ b/multipart_stream_test.go @@ -11,6 +11,7 @@ import ( "mime/multipart" "net/http" "net/http/httptest" + "net/textproto" "strings" "sync" "testing" @@ -143,7 +144,7 @@ func TestMultipartFormStreamPreservesPartOrderAndFields(t *testing.T) { assert.EqualT(t, "three", request.Form.Get("after")) } -func TestMultipartFormStreamBodyValuesPrecedeQueryValues(t *testing.T) { +func TestMultipartFormStreamQueryValuesPrecedeBodyValues(t *testing.T) { body, contentType := orderedMultipartBody(t, orderedField{name: "shared", value: "body"}, orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, @@ -156,8 +157,36 @@ func TestMultipartFormStreamBodyValuesPrecedeQueryValues(t *testing.T) { _, err = stream.NextFile() require.NoError(t, err) - assert.Equal(t, []string{"body", "query"}, request.Form["shared"]) - assert.EqualT(t, "body", request.Form.Get("shared")) + assert.Equal(t, []string{"query", "body"}, request.Form["shared"]) + assert.EqualT(t, "query", request.Form.Get("shared")) +} + +func TestMultipartFormStreamIgnoresPartsWithoutFormName(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedRawPart{ + header: textproto.MIMEHeader{ + "Content-Disposition": {`attachment; filename="ignored.bin"`}, + }, + content: "ignored attachment", + }, + orderedRawPart{ + header: textproto.MIMEHeader{ + "Content-Disposition": {`form-data; filename="also-ignored.bin"`}, + }, + content: "ignored unnamed form part", + }, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + file, err := stream.NextFile() + require.NoError(t, err) + assert.EqualT(t, testFieldFile, file.FieldName) + assert.EqualT(t, streamedFilename, file.Filename) } func TestMultipartFormStreamNextFileDrainsPreviousFile(t *testing.T) { @@ -330,6 +359,23 @@ func TestMultipartFormStreamLimits(t *testing.T) { assert.EqualT(t, testFieldFile, parseErr.Name) }) + t.Run("parts", func(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedField{name: "first", value: "one"}, + orderedField{name: "second", value: "two"}, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + stream, err := NewMultipartFormStream(request, MultipartFormStreamMaxParts(1)) + require.NoError(t, err) + defer stream.Close() + + _, err = stream.NextFile() + var parseErr *errors.ParseError + require.True(t, stderrors.As(err, &parseErr), "expected *errors.ParseError, got %T", err) + }) + t.Run("files", func(t *testing.T) { body, contentType := orderedMultipartBody(t, orderedFile{field: testFieldFile1, filename: testFileFieldA, content: "A"}, @@ -360,10 +406,46 @@ func TestNewMultipartFormStreamRejectsNonMultipartRequest(t *testing.T) { require.True(t, stderrors.As(err, &parseErr), "expected *errors.ParseError, got %T", err) } +func TestNewMultipartFormStreamRejectsMultipartMixed(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, strings.Replace(contentType, MultipartFormMime, "multipart/mixed", 1)) + + _, err := NewMultipartFormStream(request) + var parseErr *errors.ParseError + require.True(t, stderrors.As(err, &parseErr), "expected *errors.ParseError, got %T", err) + require.ErrorIs(t, parseErr.Reason, http.ErrNotMultipart) +} + +func TestNewMultipartFormStreamRejectsEmptyBoundary(t *testing.T) { + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, strings.NewReader("body")) + request.Header.Set(HeaderContentType, `multipart/form-data; boundary=""`) + + _, err := NewMultipartFormStream(request) + var parseErr *errors.ParseError + require.True(t, stderrors.As(err, &parseErr), "expected *errors.ParseError, got %T", err) + require.ErrorIs(t, parseErr.Reason, http.ErrMissingBoundary) +} + type orderedMultipartPart interface { writeTo(*testing.T, *multipart.Writer) } +type orderedRawPart struct { + header textproto.MIMEHeader + content string +} + +func (p orderedRawPart) writeTo(t *testing.T, writer *multipart.Writer) { + t.Helper() + part, err := writer.CreatePart(p.header) + require.NoError(t, err) + _, err = io.WriteString(part, p.content) + require.NoError(t, err) +} + type orderedField struct { name string value string From d820dfd29122395fcb29b5e793b6cf93aa77ff4b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A4=D0=B8=D0=BB=D0=B8=D0=BC=D0=BE=D0=BD=D0=B5=D0=BD?= =?UTF-8?q?=D0=BA=D0=BE=D0=B2=20=D0=9F=D0=B0=D0=B2=D0=B5=D0=BB=20=D0=90?= =?UTF-8?q?=D0=BD=D0=B0=D1=82=D0=BE=D0=BB=D1=8C=D0=B5=D0=B2=D0=B8=D1=87?= Date: Thu, 23 Jul 2026 16:49:55 +0300 Subject: [PATCH 3/5] refactor: simplify multipart stream traversal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Филимоненков Павел Анатольевич --- multipart_stream.go | 182 ++++++++++++++++++++++++++++---------------- 1 file changed, 116 insertions(+), 66 deletions(-) diff --git a/multipart_stream.go b/multipart_stream.go index c746b6ee..164c7df7 100644 --- a/multipart_stream.go +++ b/multipart_stream.go @@ -228,74 +228,11 @@ func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) // NextFile returns io.EOF when no file parts remain. At that point all trailing // ordinary fields have been collected. func (s *MultipartFormStream) NextFile() (*StreamedFile, error) { - if s == nil { - return nil, io.ErrClosedPipe - } - if s.done { - return nil, io.EOF + if err := s.prepareNextFile(); err != nil { + return nil, err } - if s.closed { - return nil, io.ErrClosedPipe - } - if err := s.closeCurrent(); err != nil { - return nil, s.abort(err) - } - - for { - part, err := s.reader.NextPart() - if err != nil { - if stderrors.Is(err, io.EOF) { - s.done = true - - return nil, io.EOF - } - - return nil, s.abort(err) - } - - s.parts++ - if s.maxParts > 0 && s.parts > s.maxParts { - return nil, s.abort(errors.NewParseError("body", "formData", "", - fmt.Errorf("multipart form contains %d parts, exceeds limit %d", s.parts, s.maxParts))) - } - - fieldName := part.FormName() - if fieldName == "" { - if err = discardPart(part); err != nil { - return nil, s.abort(err) - } - continue - } - - filename := part.FileName() - if filename == "" { - if err = s.bindValue(part, fieldName); err != nil { - return nil, s.abort(err) - } - - continue - } - - s.files++ - if s.maxFiles > 0 && s.files > s.maxFiles { - return nil, s.abort(errors.NewParseError("body", "formData", "", - fmt.Errorf("multipart form contains %d file parts, exceeds limit %d", s.files, s.maxFiles))) - } - if err := ValidateFilenameLength(fieldName, "formData", filename, s.maxFilenameLen); err != nil { - return nil, s.abort(err) - } - - file := &StreamedFile{ - FieldName: fieldName, - Filename: filename, - Header: part.Header, - part: part, - } - s.current = file - - return file, nil - } + return s.readNextFile() } // Drain consumes the rest of the multipart body. @@ -394,3 +331,116 @@ func (r *contextReadCloser) Read(p []byte) (int, error) { return r.ReadCloser.Read(p) } + +func (s *MultipartFormStream) prepareNextFile() error { + if s == nil { + return io.ErrClosedPipe + } + if s.done { + return io.EOF + } + if s.closed { + return io.ErrClosedPipe + } + if err := s.closeCurrent(); err != nil { + return s.abort(err) + } + + return nil +} + +func (s *MultipartFormStream) readNextFile() (*StreamedFile, error) { + for { + part, err := s.reader.NextPart() + if err != nil { + return nil, s.handleNextPartError(err) + } + + s.parts++ + if s.maxParts > 0 && s.parts > s.maxParts { + err := errors.NewParseError( + "body", + "formData", + "", + fmt.Errorf( + "multipart form contains %d parts, exceeds limit %d", + s.parts, + s.maxParts, + ), + ) + + return nil, s.abort(err) + } + + fieldName := part.FormName() + if fieldName == "" { + if err := discardPart(part); err != nil { + return nil, s.abort(err) + } + + continue + } + + filename := part.FileName() + if filename == "" { + if err := s.bindValue(part, fieldName); err != nil { + return nil, s.abort(err) + } + + continue + } + + return s.openFile(part, fieldName, filename) + } +} + +func (s *MultipartFormStream) handleNextPartError(err error) error { + if stderrors.Is(err, io.EOF) { + s.done = true + + return io.EOF + } + + return s.abort(err) +} + +func (s *MultipartFormStream) openFile( + part *multipart.Part, + fieldName string, + filename string, +) (*StreamedFile, error) { + s.files++ + if s.maxFiles > 0 && s.files > s.maxFiles { + err := errors.NewParseError( + "body", + "formData", + "", + fmt.Errorf( + "multipart form contains %d file parts, exceeds limit %d", + s.files, + s.maxFiles, + ), + ) + + return nil, s.abort(err) + } + + if err := ValidateFilenameLength( + fieldName, + "formData", + filename, + s.maxFilenameLen, + ); err != nil { + return nil, s.abort(err) + } + + file := &StreamedFile{ + FieldName: fieldName, + Filename: filename, + Header: part.Header, + part: part, + } + s.current = file + + return file, nil +} From 3590828f064a8ede0fa9252f63764bc03ed01ff0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A4=D0=B8=D0=BB=D0=B8=D0=BC=D0=BE=D0=BD=D0=B5=D0=BD?= =?UTF-8?q?=D0=BA=D0=BE=D0=B2=20=D0=9F=D0=B0=D0=B2=D0=B5=D0=BB=20=D0=90?= =?UTF-8?q?=D0=BD=D0=B0=D1=82=D0=BE=D0=BB=D1=8C=D0=B5=D0=B2=D0=B8=D1=87?= Date: Fri, 24 Jul 2026 11:33:33 +0300 Subject: [PATCH 4/5] refactor: address multipart stream review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Филимоненков Павел Анатольевич --- form.go | 13 ++++-- multipart_stream.go | 90 ++++++++++++++++++++++++++-------------- multipart_stream_test.go | 86 ++++++++++++++++++++++++++++++++++++++ 3 files changed, 156 insertions(+), 33 deletions(-) diff --git a/form.go b/form.go index b4b36f14..213757f1 100644 --- a/form.go +++ b/form.go @@ -71,11 +71,16 @@ type FileBinder func(file multipart.File, header *multipart.FileHeader) error // behaviour) be added without breaking the signature. type BindOption func(*bindConfig) -type bindConfig struct { - maxParseMemory int64 +type multipartFormLimits struct { maxBody int64 maxFiles int maxFilenameLen int +} + +type bindConfig struct { + multipartFormLimits + + maxParseMemory int64 files []formFileSpec } @@ -198,7 +203,9 @@ func BindFormFile(name string, required bool, bind FileBinder) BindOption { // it directly as a filesystem path. func BindForm(r *http.Request, opts ...BindOption) (fatal bool, err error) { cfg := bindConfig{ - maxFilenameLen: DefaultMaxUploadFilenameLength, + multipartFormLimits: multipartFormLimits{ + maxFilenameLen: DefaultMaxUploadFilenameLength, + }, } for _, opt := range opts { opt(&cfg) diff --git a/multipart_stream.go b/multipart_stream.go index 164c7df7..05937c88 100644 --- a/multipart_stream.go +++ b/multipart_stream.go @@ -22,10 +22,9 @@ type MultipartFormStreamOption func(*multipartFormStreamConfig) const defaultMultipartFormStreamMaxParts = 1000 type multipartFormStreamConfig struct { - maxBody int64 - maxFiles int - maxParts int - maxFilenameLen int + multipartFormLimits + + maxParts int } // MultipartFormStreamMaxBody caps the total number of request-body bytes read @@ -61,7 +60,9 @@ func MultipartFormStreamMaxFilenameLen(n int) MultipartFormStreamOption { // // Reads block until bytes arrive from the client. StreamedFile is not seekable // and is not safe for concurrent use. Its form name, filename and MIME headers -// are available before the payload is consumed. +// are available before the payload is consumed. The underlying [multipart.Part] +// remains private so callers cannot bypass Close and its error-preserving drain +// semantics; Header exposes the part metadata without exposing that lifecycle. // // Closing a StreamedFile drains only the unread remainder of that file part. // Close may therefore block while the client is still uploading the current @@ -125,30 +126,33 @@ func (f *StreamedFile) Close() error { // possible. Call [MultipartFormStream.Close] to stop multipart processing // without explicitly draining the remaining parts. type MultipartFormStream struct { - request *http.Request - reader *multipart.Reader - current *StreamedFile - maxFiles int - maxParts int - maxFilenameLen int - files int - parts int - closed bool - done bool + multipartFormStreamConfig + + request *http.Request + reader *multipart.Reader + current *StreamedFile + + files int + parts int + closed bool + done bool } // NewMultipartFormStream creates a sequential multipart/form-data stream over // r.Body. // -// The constructor accepts only multipart/form-data, validates that a non-empty -// boundary is present, but does not consume multipart parts. It initializes -// request.Form and request.PostForm in the same way as -// [http.Request.ParseForm], then populates multipart values incrementally -// as [MultipartFormStream.NextFile] advances. +// For POST, PUT and PATCH requests, the constructor accepts only +// multipart/form-data, validates that a non-empty boundary is present, but does +// not consume multipart parts. For other methods, it returns an empty stream +// whose NextFile method reports io.EOF without reading the request body. +// +// The constructor initializes request.Form and request.PostForm in the same way +// as [http.Request.ParseForm], then populates multipart values incrementally as +// [MultipartFormStream.NextFile] advances. // -// NewMultipartFormStream marks the request as handled by MultipartReader. -// Callers must not subsequently call [http.Request.ParseMultipartForm] or -// [BindForm] for the same request. +// For POST, PUT and PATCH requests, NewMultipartFormStream marks the request +// as handled by MultipartReader. Callers must not subsequently call +// [http.Request.ParseMultipartForm] or [BindForm] for the same request. // // File payloads are exposed directly from the request body and are not buffered // in memory or temporary files. Ordinary form values are read into memory as @@ -163,8 +167,10 @@ type MultipartFormStream struct { // File names and MIME headers are supplied by the client and remain untrusted. func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) (*MultipartFormStream, error) { cfg := multipartFormStreamConfig{ - maxFilenameLen: DefaultMaxUploadFilenameLength, - maxParts: defaultMultipartFormStreamMaxParts, + multipartFormLimits: multipartFormLimits{ + maxFilenameLen: DefaultMaxUploadFilenameLength, + }, + maxParts: defaultMultipartFormStreamMaxParts, } for _, opt := range opts { opt(&cfg) @@ -173,6 +179,19 @@ func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) if r == nil { return nil, errors.NewParseError("body", "formData", "", stderrors.New("nil request")) } + + if !supportsMultipartFormStream(r.Method) { + if err := r.ParseForm(); err != nil { + return nil, errors.NewParseError("body", "formData", "", err) + } + + return &MultipartFormStream{ + request: r, + multipartFormStreamConfig: cfg, + done: true, + }, nil + } + if r.Body == nil { return nil, errors.NewParseError("body", "formData", "", stderrors.New("nil request body")) } @@ -209,11 +228,9 @@ func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) } return &MultipartFormStream{ - request: r, - reader: reader, - maxFiles: cfg.maxFiles, - maxParts: cfg.maxParts, - maxFilenameLen: cfg.maxFilenameLen, + request: r, + reader: reader, + multipartFormStreamConfig: cfg, }, nil } @@ -280,9 +297,22 @@ func (s *MultipartFormStream) Close() error { s.current = nil } + if s.request == nil || s.request.Body == nil { + return nil + } + return s.request.Body.Close() } +func supportsMultipartFormStream(method string) bool { + switch method { + case http.MethodPost, http.MethodPut, http.MethodPatch: + return true + default: + return false + } +} + func (s *MultipartFormStream) abort(err error) error { return stderrors.Join(err, s.Close()) } diff --git a/multipart_stream_test.go b/multipart_stream_test.go index a1938976..aba28349 100644 --- a/multipart_stream_test.go +++ b/multipart_stream_test.go @@ -309,6 +309,62 @@ func TestMultipartFormStreamBodyLimit(t *testing.T) { require.True(t, stderrors.As(err, &maxBytesErr), "expected *http.MaxBytesError, got %T", err) } +func TestMultipartFormStreamPropagatesMaxBytesHandlerError(t *testing.T) { + const limit int64 = 512 + + tests := []struct { + name string + consume func(*StreamedFile) error + }{ + { + name: "read", + consume: func(file *StreamedFile) error { + _, err := io.ReadAll(file) + + return err + }, + }, + { + name: "drain", + consume: func(file *StreamedFile) error { + return file.Close() + }, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedFile{ + field: testFieldFile, + filename: streamedFilename, + content: strings.Repeat("x", 1024), + }, + ) + + var consumeErr error + handler := http.MaxBytesHandler(http.HandlerFunc(func(_ http.ResponseWriter, request *http.Request) { + stream, err := NewMultipartFormStream(request, MultipartFormStreamMaxBody(-1)) + require.NoError(t, err) + defer stream.Close() + + file, err := stream.NextFile() + require.NoError(t, err) + consumeErr = test.consume(file) + }), limit) + + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + handler.ServeHTTP(httptest.NewRecorder(), request) + + var maxBytesErr *http.MaxBytesError + require.True(t, stderrors.As(consumeErr, &maxBytesErr), + "expected *http.MaxBytesError, got %T", consumeErr) + assert.EqualT(t, limit, maxBytesErr.Limit) + }) + } +} + func TestMultipartFormStreamMalformedBody(t *testing.T) { body, contentType := orderedMultipartBody(t, orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, @@ -397,6 +453,36 @@ func TestMultipartFormStreamLimits(t *testing.T) { }) } +func TestNewMultipartFormStreamReturnsEmptyForUnsupportedMethods(t *testing.T) { + methods := []string{ + http.MethodGet, + http.MethodHead, + http.MethodDelete, + http.MethodOptions, + } + + for _, method := range methods { + t.Run(method, func(t *testing.T) { + request := httptest.NewRequestWithContext( + t.Context(), + method, + testUploadPath+"?query=value", + strings.NewReader("this body must not be parsed as multipart"), + ) + request.Header.Set(HeaderContentType, URLencodedFormMime) + + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + _, err = stream.NextFile() + require.ErrorIs(t, err, io.EOF) + assert.EqualT(t, "value", request.Form.Get("query")) + assert.Empty(t, request.PostForm) + }) + } +} + func TestNewMultipartFormStreamRejectsNonMultipartRequest(t *testing.T) { request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, strings.NewReader("value=x")) request.Header.Set(HeaderContentType, URLencodedFormMime) From ea90c357afbfb10d315bef7b600efc19285a47f4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=A4=D0=B8=D0=BB=D0=B8=D0=BC=D0=BE=D0=BD=D0=B5=D0=BD?= =?UTF-8?q?=D0=BA=D0=BE=D0=B2=20=D0=9F=D0=B0=D0=B2=D0=B5=D0=BB=20=D0=90?= =?UTF-8?q?=D0=BD=D0=B0=D1=82=D0=BE=D0=BB=D1=8C=D0=B5=D0=B2=D0=B8=D1=87?= Date: Fri, 24 Jul 2026 16:43:47 +0300 Subject: [PATCH 5/5] feat: expose discovered multipart stream state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add snapshot accessors for multipart fields and file metadata discovered while advancing the stream. Keep NextFile as the sequential payload API while giving handlers enough state for progress tracking and application-specific validation. Signed-off-by: Филимоненков Павел Анатольевич --- multipart_stream.go | 94 +++++++++++++++++++++++++++++++++++++--- multipart_stream_test.go | 55 +++++++++++++++++++++++ 2 files changed, 142 insertions(+), 7 deletions(-) diff --git a/multipart_stream.go b/multipart_stream.go index 05937c88..76e1bf46 100644 --- a/multipart_stream.go +++ b/multipart_stream.go @@ -12,6 +12,7 @@ import ( "mime/multipart" "net/http" "net/textproto" + "net/url" "github.com/go-openapi/errors" ) @@ -75,6 +76,17 @@ type StreamedFile struct { closeErr error } +// MultipartFileInfo describes a file part discovered by [MultipartFormStream]. +// +// The payload reader is intentionally omitted. File parts remain sequential and +// are consumed through [MultipartFormStream.NextFile]. Header is a snapshot of +// the client-supplied MIME headers and must be treated as untrusted input. +type MultipartFileInfo struct { + FieldName string + Filename string + Header textproto.MIMEHeader +} + // Read reads file payload bytes directly from the request body. func (f *StreamedFile) Read(p []byte) (int, error) { if f == nil || f.part == nil { @@ -125,6 +137,11 @@ func (f *StreamedFile) Close() error { // remaining body, collect trailing fields and allow HTTP connection reuse when // possible. Call [MultipartFormStream.Close] to stop multipart processing // without explicitly draining the remaining parts. +// +// [MultipartFormStream.Fields] and [MultipartFormStream.Files] expose snapshots +// of the multipart fields and file metadata discovered so far. They never read +// ahead: fields or files after the active file become visible only after the +// stream advances. type MultipartFormStream struct { multipartFormStreamConfig @@ -132,10 +149,11 @@ type MultipartFormStream struct { reader *multipart.Reader current *StreamedFile - files int - parts int - closed bool - done bool + fields url.Values + fileInfos []MultipartFileInfo + parts int + closed bool + done bool } // NewMultipartFormStream creates a sequential multipart/form-data stream over @@ -188,6 +206,7 @@ func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) return &MultipartFormStream{ request: r, multipartFormStreamConfig: cfg, + fields: make(url.Values), done: true, }, nil } @@ -231,9 +250,46 @@ func NewMultipartFormStream(r *http.Request, opts ...MultipartFormStreamOption) request: r, reader: reader, multipartFormStreamConfig: cfg, + fields: make(url.Values), }, nil } +// Fields returns a snapshot of ordinary multipart form fields discovered so far. +// +// URL query values are not included. Repeated multipart fields preserve their +// encounter order. Fields after the active file are not visible until that file +// is consumed or closed and the stream advances. Mutating the returned values +// does not affect the stream or the request. +func (s *MultipartFormStream) Fields() url.Values { + if s == nil { + return nil + } + + return cloneMultipartValues(s.fields) +} + +// Files returns snapshots of file metadata discovered so far in wire order. +// +// The currently active file is included as soon as NextFile returns it. Payload +// readers are not retained in the index. Mutating the returned slice or MIME +// headers does not affect the stream. +func (s *MultipartFormStream) Files() []MultipartFileInfo { + if s == nil { + return nil + } + + files := make([]MultipartFileInfo, len(s.fileInfos)) + for i, file := range s.fileInfos { + files[i] = MultipartFileInfo{ + FieldName: file.FieldName, + Filename: file.Filename, + Header: cloneMultipartMIMEHeader(file.Header), + } + } + + return files +} + // NextFile advances through the multipart body and returns the next file part. // Ordinary form fields encountered before that file are added to request.Form // and request.PostForm. @@ -334,6 +390,7 @@ func (s *MultipartFormStream) bindValue(part *multipart.Part, name string) error return err } + s.fields.Add(name, string(value)) s.request.PostForm.Add(name, string(value)) // Match net/http.ParseMultipartForm: query values already present in Form // keep precedence over multipart body values, which are appended. @@ -348,6 +405,24 @@ func discardPart(part *multipart.Part) error { return err } +func cloneMultipartValues(values url.Values) url.Values { + cloned := make(url.Values, len(values)) + for name, entries := range values { + cloned[name] = append([]string(nil), entries...) + } + + return cloned +} + +func cloneMultipartMIMEHeader(header textproto.MIMEHeader) textproto.MIMEHeader { + cloned := make(textproto.MIMEHeader, len(header)) + for name, entries := range header { + cloned[name] = append([]string(nil), entries...) + } + + return cloned +} + type contextReadCloser struct { io.ReadCloser @@ -439,15 +514,15 @@ func (s *MultipartFormStream) openFile( fieldName string, filename string, ) (*StreamedFile, error) { - s.files++ - if s.maxFiles > 0 && s.files > s.maxFiles { + fileCount := len(s.fileInfos) + 1 + if s.maxFiles > 0 && fileCount > s.maxFiles { err := errors.NewParseError( "body", "formData", "", fmt.Errorf( "multipart form contains %d file parts, exceeds limit %d", - s.files, + fileCount, s.maxFiles, ), ) @@ -470,6 +545,11 @@ func (s *MultipartFormStream) openFile( Header: part.Header, part: part, } + s.fileInfos = append(s.fileInfos, MultipartFileInfo{ + FieldName: fieldName, + Filename: filename, + Header: cloneMultipartMIMEHeader(part.Header), + }) s.current = file return file, nil diff --git a/multipart_stream_test.go b/multipart_stream_test.go index aba28349..a66604e2 100644 --- a/multipart_stream_test.go +++ b/multipart_stream_test.go @@ -126,6 +126,12 @@ func TestMultipartFormStreamPreservesPartOrderAndFields(t *testing.T) { assert.EqualT(t, testFieldFile1, first.FieldName) assert.EqualT(t, "one", request.Form.Get("before")) assert.Empty(t, request.Form.Get("between")) + assert.Equal(t, []string{"one"}, stream.Fields()["before"]) + assert.Empty(t, stream.Fields()["between"]) + firstFiles := stream.Files() + require.Len(t, firstFiles, 1) + assert.EqualT(t, testFieldFile1, firstFiles[0].FieldName) + assert.EqualT(t, testFileFieldA, firstFiles[0].Filename) content, err := io.ReadAll(first) require.NoError(t, err) assert.EqualT(t, "AAA", string(content)) @@ -135,6 +141,13 @@ func TestMultipartFormStreamPreservesPartOrderAndFields(t *testing.T) { assert.EqualT(t, testFieldFile2, second.FieldName) assert.EqualT(t, "two", request.Form.Get("between")) assert.Empty(t, request.Form.Get("after")) + assert.Equal(t, []string{"one"}, stream.Fields()["before"]) + assert.Equal(t, []string{"two"}, stream.Fields()["between"]) + assert.Empty(t, stream.Fields()["after"]) + secondFiles := stream.Files() + require.Len(t, secondFiles, 2) + assert.EqualT(t, testFieldFile2, secondFiles[1].FieldName) + assert.EqualT(t, testFileFieldB, secondFiles[1].Filename) content, err = io.ReadAll(second) require.NoError(t, err) assert.EqualT(t, "BBBB", string(content)) @@ -142,6 +155,47 @@ func TestMultipartFormStreamPreservesPartOrderAndFields(t *testing.T) { _, err = stream.NextFile() require.ErrorIs(t, err, io.EOF) assert.EqualT(t, "three", request.Form.Get("after")) + assert.Equal(t, []string{"three"}, stream.Fields()["after"]) +} + +func TestMultipartFormStreamDiscoverySnapshots(t *testing.T) { + body, contentType := orderedMultipartBody(t, + orderedField{name: "tag", value: "one"}, + orderedField{name: "tag", value: "two"}, + orderedFile{field: testFieldFile, filename: streamedFilename, content: "payload"}, + ) + request := httptest.NewRequestWithContext(t.Context(), http.MethodPost, testUploadPath, body) + request.Header.Set(HeaderContentType, contentType) + + stream, err := NewMultipartFormStream(request) + require.NoError(t, err) + defer stream.Close() + + _, err = stream.NextFile() + require.NoError(t, err) + + fields := stream.Fields() + assert.Equal(t, []string{"one", "two"}, fields["tag"]) + files := stream.Files() + require.Len(t, files, 1) + assert.EqualT(t, testFieldFile, files[0].FieldName) + assert.EqualT(t, streamedFilename, files[0].Filename) + assert.NotEmpty(t, files[0].Header.Get("Content-Disposition")) + + fields.Set("tag", "changed") + files[0].Filename = "changed.bin" + files[0].Header.Set("Content-Disposition", "changed") + + assert.Equal(t, []string{"one", "two"}, stream.Fields()["tag"]) + files = stream.Files() + require.Len(t, files, 1) + assert.EqualT(t, streamedFilename, files[0].Filename) + assert.NotEqual(t, "changed", files[0].Header.Get("Content-Disposition")) + assert.Equal(t, []string{"one", "two"}, request.PostForm["tag"]) + + var nilStream *MultipartFormStream + assert.Nil(t, nilStream.Fields()) + assert.Nil(t, nilStream.Files()) } func TestMultipartFormStreamQueryValuesPrecedeBodyValues(t *testing.T) { @@ -159,6 +213,7 @@ func TestMultipartFormStreamQueryValuesPrecedeBodyValues(t *testing.T) { require.NoError(t, err) assert.Equal(t, []string{"query", "body"}, request.Form["shared"]) assert.EqualT(t, "query", request.Form.Get("shared")) + assert.Equal(t, []string{"body"}, stream.Fields()["shared"]) } func TestMultipartFormStreamIgnoresPartsWithoutFormName(t *testing.T) {