From 6d15b0ee64f191a3c5282058fee88bc2b99697e8 Mon Sep 17 00:00:00 2001 From: Lucas Manning Date: Sat, 15 Jan 2022 16:33:32 -0800 Subject: [PATCH] Fix packet buffer reference counting in IP fragmentation/reassembly. This change also adds a reference counting check to the fragmentation/reassembly unit tests. PiperOrigin-RevId: 422097740 --- .../network/internal/fragmentation/BUILD | 3 + .../internal/fragmentation/fragmentation.go | 8 + .../fragmentation/fragmentation_test.go | 158 +++++++++++------- .../internal/fragmentation/main_test.go | 30 ++++ .../internal/fragmentation/reassembler.go | 8 +- .../fragmentation/reassembler_test.go | 26 +++ pkg/tcpip/network/ipv4/ipv4.go | 2 + pkg/tcpip/network/ipv6/ipv6.go | 2 + 8 files changed, 175 insertions(+), 62 deletions(-) create mode 100644 pkg/tcpip/network/internal/fragmentation/main_test.go diff --git a/pkg/tcpip/network/internal/fragmentation/BUILD b/pkg/tcpip/network/internal/fragmentation/BUILD index 274f09092..c0291d803 100644 --- a/pkg/tcpip/network/internal/fragmentation/BUILD +++ b/pkg/tcpip/network/internal/fragmentation/BUILD @@ -41,10 +41,13 @@ go_test( size = "small", srcs = [ "fragmentation_test.go", + "main_test.go", "reassembler_test.go", ], library = ":fragmentation", deps = [ + "//pkg/refs", + "//pkg/refsvfs2", "//pkg/tcpip/buffer", "//pkg/tcpip/faketime", "//pkg/tcpip/network/internal/testutil", diff --git a/pkg/tcpip/network/internal/fragmentation/fragmentation.go b/pkg/tcpip/network/internal/fragmentation/fragmentation.go index 1ba4d0d36..762cd751d 100644 --- a/pkg/tcpip/network/internal/fragmentation/fragmentation.go +++ b/pkg/tcpip/network/internal/fragmentation/fragmentation.go @@ -237,6 +237,14 @@ func (f *Fragmentation) release(r *reassembler, timedOut bool) { if h := f.timeoutHandler; timedOut && h != nil { h.OnReassemblyTimeout(r.pkt) } + if r.pkt != nil { + r.pkt.DecRef() + } + for _, h := range r.holes { + if h.pkt != nil { + h.pkt.DecRef() + } + } } // releaseReassemblersLocked releases already-expired reassemblers, then diff --git a/pkg/tcpip/network/internal/fragmentation/fragmentation_test.go b/pkg/tcpip/network/internal/fragmentation/fragmentation_test.go index dadfc28cc..e3e373e45 100644 --- a/pkg/tcpip/network/internal/fragmentation/fragmentation_test.go +++ b/pkg/tcpip/network/internal/fragmentation/fragmentation_test.go @@ -1,4 +1,4 @@ -// Copyright 2018 The gVisor Authors. +// Copyright 2022 The gVisor Authors. // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -61,56 +61,56 @@ type processOutput struct { done bool } -var processTestCases = []struct { - comment string - in []processInput - out []processOutput -}{ - { - comment: "One ID", - in: []processInput{ - {id: FragmentID{ID: 0}, first: 0, last: 1, more: true, pkt: pkt(2, "01")}, - {id: FragmentID{ID: 0}, first: 2, last: 3, more: false, pkt: pkt(2, "23")}, - }, - out: []processOutput{ - {vv: buffer.VectorisedView{}, done: false}, - {vv: vv(4, "01", "23"), done: true}, - }, - }, - { - comment: "Next Header protocol mismatch", - in: []processInput{ - {id: FragmentID{ID: 0}, first: 0, last: 1, more: true, proto: 6, pkt: pkt(2, "01")}, - {id: FragmentID{ID: 0}, first: 2, last: 3, more: false, proto: 17, pkt: pkt(2, "23")}, - }, - out: []processOutput{ - {vv: buffer.VectorisedView{}, done: false}, - {vv: vv(4, "01", "23"), proto: 6, done: true}, - }, - }, - { - comment: "Two IDs", - in: []processInput{ - {id: FragmentID{ID: 0}, first: 0, last: 1, more: true, pkt: pkt(2, "01")}, - {id: FragmentID{ID: 1}, first: 0, last: 1, more: true, pkt: pkt(2, "ab")}, - {id: FragmentID{ID: 1}, first: 2, last: 3, more: false, pkt: pkt(2, "cd")}, - {id: FragmentID{ID: 0}, first: 2, last: 3, more: false, pkt: pkt(2, "23")}, - }, - out: []processOutput{ - {vv: buffer.VectorisedView{}, done: false}, - {vv: buffer.VectorisedView{}, done: false}, - {vv: vv(4, "ab", "cd"), done: true}, - {vv: vv(4, "01", "23"), done: true}, - }, - }, -} - func TestFragmentationProcess(t *testing.T) { + var processTestCases = []struct { + comment string + in []processInput + out []processOutput + }{ + { + comment: "One ID", + in: []processInput{ + {id: FragmentID{ID: 0}, first: 0, last: 1, more: true, pkt: pkt(2, "01")}, + {id: FragmentID{ID: 0}, first: 2, last: 3, more: false, pkt: pkt(2, "23")}, + }, + out: []processOutput{ + {vv: buffer.VectorisedView{}, done: false}, + {vv: vv(4, "01", "23"), done: true}, + }, + }, + { + comment: "Next Header protocol mismatch", + in: []processInput{ + {id: FragmentID{ID: 0}, first: 0, last: 1, more: true, proto: 6, pkt: pkt(2, "01")}, + {id: FragmentID{ID: 0}, first: 2, last: 3, more: false, proto: 17, pkt: pkt(2, "23")}, + }, + out: []processOutput{ + {vv: buffer.VectorisedView{}, done: false}, + {vv: vv(4, "01", "23"), proto: 6, done: true}, + }, + }, + { + comment: "Two IDs", + in: []processInput{ + {id: FragmentID{ID: 0}, first: 0, last: 1, more: true, pkt: pkt(2, "01")}, + {id: FragmentID{ID: 1}, first: 0, last: 1, more: true, pkt: pkt(2, "ab")}, + {id: FragmentID{ID: 1}, first: 2, last: 3, more: false, pkt: pkt(2, "cd")}, + {id: FragmentID{ID: 0}, first: 2, last: 3, more: false, pkt: pkt(2, "23")}, + }, + out: []processOutput{ + {vv: buffer.VectorisedView{}, done: false}, + {vv: buffer.VectorisedView{}, done: false}, + {vv: vv(4, "ab", "cd"), done: true}, + {vv: vv(4, "01", "23"), done: true}, + }, + }, + } for _, c := range processTestCases { t.Run(c.comment, func(t *testing.T) { f := NewFragmentation(minBlockSize, 1024, 512, reassembleTimeout, &faketime.NullClock{}, nil) firstFragmentProto := c.in[0].proto for i, in := range c.in { + defer in.pkt.DecRef() resPkt, proto, done, err := f.Process(in.id, in.first, in.last, in.more, in.proto, in.pkt) if err != nil { t.Fatalf("f.Process(%+v, %d, %d, %t, %d, %#v) failed: %s", @@ -180,7 +180,9 @@ func TestReassemblingTimeout(t *testing.T) { memSizeOfFrags := func(frags ...*fragment) int { var size int for _, frag := range frags { - size += pkt(len(frag.data), frag.data).MemSize() + p := pkt(len(frag.data), frag.data) + size += p.MemSize() + p.DecRef() } return size } @@ -254,7 +256,9 @@ func TestReassemblingTimeout(t *testing.T) { for _, event := range test.events { clock.Advance(event.clockAdvance) if frag := event.fragment; frag != nil { - _, _, done, err := f.Process(FragmentID{}, frag.first, frag.last, frag.more, protocol, pkt(len(frag.data), frag.data)) + p := pkt(len(frag.data), frag.data) + defer p.DecRef() + _, _, done, err := f.Process(FragmentID{}, frag.first, frag.last, frag.more, protocol, p) if err != nil { t.Fatalf("%s: f.Process failed: %s", event.name, err) } @@ -271,25 +275,41 @@ func TestReassemblingTimeout(t *testing.T) { } func TestMemoryLimits(t *testing.T) { - lowLimit := pkt(1, "0").MemSize() + p := pkt(1, "0") + defer p.DecRef() + lowLimit := p.MemSize() highLimit := 3 * lowLimit // Allow at most 3 such packets. - f := NewFragmentation(minBlockSize, highLimit, lowLimit, reassembleTimeout, &faketime.NullClock{}, nil) + // Using a manual clock here and below because the fragmentation object + // cleans up its reassemblers with a job that's scheduled with the clock + // argument. If the clock does not schedule jobs, the reassemblers are not + // released and the fragmentation object leaks packets. + c := faketime.NewManualClock() + defer c.Advance(reassembleTimeout) + f := NewFragmentation(minBlockSize, highLimit, lowLimit, reassembleTimeout, c, nil) // Send first fragment with id = 0. - if _, _, _, err := f.Process(FragmentID{ID: 0}, 0, 0, true, 0xFF, pkt(1, "0")); err != nil { + p0 := pkt(1, "0") + defer p0.DecRef() + if _, _, _, err := f.Process(FragmentID{ID: 0}, 0, 0, true, 0xFF, p0); err != nil { t.Fatal(err) } // Send first fragment with id = 1. - if _, _, _, err := f.Process(FragmentID{ID: 1}, 0, 0, true, 0xFF, pkt(1, "1")); err != nil { + p1 := pkt(1, "1") + defer p1.DecRef() + if _, _, _, err := f.Process(FragmentID{ID: 1}, 0, 0, true, 0xFF, p1); err != nil { t.Fatal(err) } // Send first fragment with id = 2. - if _, _, _, err := f.Process(FragmentID{ID: 2}, 0, 0, true, 0xFF, pkt(1, "2")); err != nil { + p2 := pkt(1, "2") + defer p2.DecRef() + if _, _, _, err := f.Process(FragmentID{ID: 2}, 0, 0, true, 0xFF, p2); err != nil { t.Fatal(err) } // Send first fragment with id = 3. This should caused id = 0 and id = 1 to be // evicted. - if _, _, _, err := f.Process(FragmentID{ID: 3}, 0, 0, true, 0xFF, pkt(1, "3")); err != nil { + p3 := pkt(1, "3") + defer p3.DecRef() + if _, _, _, err := f.Process(FragmentID{ID: 3}, 0, 0, true, 0xFF, p3); err != nil { t.Fatal(err) } @@ -305,14 +325,22 @@ func TestMemoryLimits(t *testing.T) { } func TestMemoryLimitsIgnoresDuplicates(t *testing.T) { - memSize := pkt(1, "0").MemSize() - f := NewFragmentation(minBlockSize, memSize, 0, reassembleTimeout, &faketime.NullClock{}, nil) + p0 := pkt(1, "0") + defer p0.DecRef() + memSize := p0.MemSize() + c := faketime.NewManualClock() + defer c.Advance(reassembleTimeout) + f := NewFragmentation(minBlockSize, memSize, 0, reassembleTimeout, c, nil) // Send first fragment with id = 0. - if _, _, _, err := f.Process(FragmentID{}, 0, 0, true, 0xFF, pkt(1, "0")); err != nil { + p1 := pkt(1, "0") + defer p1.DecRef() + if _, _, _, err := f.Process(FragmentID{}, 0, 0, true, 0xFF, p1); err != nil { t.Fatal(err) } // Send the same packet again. - if _, _, _, err := f.Process(FragmentID{}, 0, 0, true, 0xFF, pkt(1, "0")); err != nil { + p1dup := pkt(1, "0") + defer p1dup.DecRef() + if _, _, _, err := f.Process(FragmentID{}, 0, 0, true, 0xFF, p1dup); err != nil { t.Fatal(err) } @@ -404,8 +432,16 @@ func TestErrors(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { - f := NewFragmentation(test.blockSize, HighFragThreshold, LowFragThreshold, reassembleTimeout, &faketime.NullClock{}, nil) - _, _, done, err := f.Process(FragmentID{}, test.first, test.last, test.more, 0, pkt(len(test.data), test.data)) + p0 := pkt(len(test.data), test.data) + defer p0.DecRef() + c := faketime.NewManualClock() + defer c.Advance(reassembleTimeout) + f := NewFragmentation(test.blockSize, HighFragThreshold, LowFragThreshold, reassembleTimeout, c, nil) + resPkt, _, done, err := f.Process(FragmentID{}, test.first, test.last, test.more, 0, p0) + + if resPkt != nil { + resPkt.DecRef() + } if !errors.Is(err, test.err) { t.Errorf("got Process(_, %d, %d, %t, _, %q) = (_, _, _, %v), want = (_, _, _, %v)", test.first, test.last, test.more, test.data, err, test.err) } @@ -482,11 +518,13 @@ func TestPacketFragmenter(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { pkt := testutil.MakeRandPkt(test.transportHeaderLen, reserve, []int{test.payloadSize}, proto) + defer pkt.DecRef() originalPayload := stack.PayloadSince(pkt.TransportHeader()) var reassembledPayload buffer.VectorisedView pf := MakePacketFragmenter(pkt, test.fragmentPayloadLen, reserve) for i := 0; ; i++ { fragPkt, offset, copied, more := pf.BuildNextFragment() + defer fragPkt.DecRef() wantFragment := test.wantFragments[i] if got := pf.RemainingFragmentCount(); got != wantFragment.remaining { t.Errorf("(fragment #%d) got pf.RemainingFragmentCount() = %d, want = %d", i, got, wantFragment.remaining) @@ -538,7 +576,9 @@ func TestTimeoutHandler(t *testing.T) { ) pk1 := pkt(1, "1") + defer pk1.DecRef() pk2 := pkt(1, "2") + defer pk2.DecRef() type processParam struct { first uint16 diff --git a/pkg/tcpip/network/internal/fragmentation/main_test.go b/pkg/tcpip/network/internal/fragmentation/main_test.go new file mode 100644 index 000000000..6eb7b9795 --- /dev/null +++ b/pkg/tcpip/network/internal/fragmentation/main_test.go @@ -0,0 +1,30 @@ +// Copyright 2022 The gVisor Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fragmentation + +import ( + "os" + "testing" + + "gvisor.dev/gvisor/pkg/refs" + "gvisor.dev/gvisor/pkg/refsvfs2" +) + +func TestMain(m *testing.M) { + refs.SetLeakMode(refs.LeaksPanic) + code := m.Run() + refsvfs2.DoLeakCheck() + os.Exit(code) +} diff --git a/pkg/tcpip/network/internal/fragmentation/reassembler.go b/pkg/tcpip/network/internal/fragmentation/reassembler.go index ff6be8f0d..73873b003 100644 --- a/pkg/tcpip/network/internal/fragmentation/reassembler.go +++ b/pkg/tcpip/network/internal/fragmentation/reassembler.go @@ -135,6 +135,7 @@ func (r *reassembler) process(first, last uint16, more bool, proto uint8, pkt *s final: currentHole.final, pkt: pkt, } + pkt.IncRef() r.filled++ // For IPv6, it is possible to have different Protocol values between // fragments of a packet (because, unlike IPv4, the Protocol is not used to @@ -145,11 +146,13 @@ func (r *reassembler) process(first, last uint16, more bool, proto uint8, pkt *s // options received in the first fragment should be used - and they should // override options from following fragments. if first == 0 { + if r.pkt != nil { + r.pkt.DecRef() + } r.pkt = pkt + pkt.IncRef() r.proto = proto } - - pkt.IncRef() break } if !holeFound { @@ -167,7 +170,6 @@ func (r *reassembler) process(first, last uint16, more bool, proto uint8, pkt *s }) resPkt := r.holes[0].pkt - resPkt.DecRef() for i := 1; i < len(r.holes); i++ { stack.MergeFragment(resPkt, r.holes[i].pkt) } diff --git a/pkg/tcpip/network/internal/fragmentation/reassembler_test.go b/pkg/tcpip/network/internal/fragmentation/reassembler_test.go index cfd9f00ef..62c84485d 100644 --- a/pkg/tcpip/network/internal/fragmentation/reassembler_test.go +++ b/pkg/tcpip/network/internal/fragmentation/reassembler_test.go @@ -186,6 +186,19 @@ func TestReassemblerProcess(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { r := newReassembler(FragmentID{}, &faketime.NullClock{}) + // Emulate a call to (*Fragmentation).release(), which always happens + // after reassembler error or completion. Without release(), the + // reassembler will leak PacketBuffers. + defer func() { + for _, h := range r.holes { + if h.pkt != nil { + h.pkt.DecRef() + } + } + if r.pkt != nil { + r.pkt.DecRef() + } + }() var resPkt *stack.PacketBuffer var isDone bool for _, param := range test.params { @@ -229,5 +242,18 @@ func TestReassemblerProcess(t *testing.T) { } } }) + for _, p := range test.params { + if p.pkt != nil { + p.pkt.DecRef() + } + } + for _, w := range test.want { + if w.pkt != nil { + w.pkt.DecRef() + } + } + if test.wantPkt != nil { + test.wantPkt.DecRef() + } } } diff --git a/pkg/tcpip/network/ipv4/ipv4.go b/pkg/tcpip/network/ipv4/ipv4.go index 54be0cbe3..b2f8c5367 100644 --- a/pkg/tcpip/network/ipv4/ipv4.go +++ b/pkg/tcpip/network/ipv4/ipv4.go @@ -409,10 +409,12 @@ func (e *endpoint) handleFragments(_ *stack.Route, networkMTU uint32, pkt *stack for { fragPkt, more := buildNextFragment(&pf, networkHeader) if err := handler(fragPkt); err != nil { + fragPkt.DecRef() return n, pf.RemainingFragmentCount() + 1, err } n++ if !more { + fragPkt.DecRef() return n, pf.RemainingFragmentCount(), nil } } diff --git a/pkg/tcpip/network/ipv6/ipv6.go b/pkg/tcpip/network/ipv6/ipv6.go index 8ec8bf221..a76bc25e1 100644 --- a/pkg/tcpip/network/ipv6/ipv6.go +++ b/pkg/tcpip/network/ipv6/ipv6.go @@ -731,10 +731,12 @@ func (e *endpoint) handleFragments(r *stack.Route, networkMTU uint32, pkt *stack for { fragPkt, more := buildNextFragment(&pf, networkHeader, transProto, id) if err := handler(fragPkt); err != nil { + fragPkt.DecRef() return n, pf.RemainingFragmentCount() + 1, err } n++ if !more { + fragPkt.DecRef() return n, pf.RemainingFragmentCount(), nil } }