-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Capture and mux packets on all interfaces. (#12)
* Muxes packets from all interfaces
- Loading branch information
Showing
4 changed files
with
220 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| // Package muxer helps solve the problem that captures take place only on a | ||
| // per-interface basis, but tcp-info collects flow information with no reference | ||
| // to the underlying interface. | ||
| package muxer | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/google/gopacket" | ||
| "github.com/google/gopacket/layers" | ||
| "github.com/google/gopacket/pcap" | ||
| "github.com/m-lab/go/rtx" | ||
| "github.com/m-lab/packet-headers/metrics" | ||
| ) | ||
|
|
||
| func forwardPackets(ctx context.Context, in <-chan gopacket.Packet, out chan<- gopacket.Packet, wg *sync.WaitGroup) { | ||
| defer wg.Done() | ||
| metrics.InterfacesBeingCaptured.Inc() | ||
| defer metrics.InterfacesBeingCaptured.Dec() | ||
|
|
||
| for { | ||
| select { | ||
| case p, ok := <-in: | ||
| if !ok { | ||
| return | ||
| } | ||
| out <- p | ||
| case <-ctx.Done(): | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // muxPackets causes each packet on every input channel to be sent to the output channel. | ||
| func muxPackets(ctx context.Context, in []<-chan gopacket.Packet, out chan<- gopacket.Packet) { | ||
| wg := sync.WaitGroup{} | ||
| for _, inC := range in { | ||
| wg.Add(1) | ||
| go forwardPackets(ctx, inC, out, &wg) | ||
| } | ||
|
|
||
| wg.Wait() | ||
| close(out) | ||
| } | ||
|
|
||
| // PcapHandleOpener is a type to allow injection of fake packet captures to aid | ||
| // in testing. It is exactly the type of pcap.OpenLive, and in production code | ||
| // every variable of this type should be set to pcap.OpenLive. | ||
| type PcapHandleOpener func(device string, snaplen int32, promisc bool, timeout time.Duration) (handle *pcap.Handle, _ error) | ||
|
|
||
| // MustCaptureTCPOnInterfaces fires off a packet capture on every one of the | ||
| // passed-in list of interfaces, and then muxes the resulting packet streams to | ||
| // all be sent to the passed-in packets channel. | ||
| func MustCaptureTCPOnInterfaces(ctx context.Context, interfaces []string, packets chan<- gopacket.Packet, opener PcapHandleOpener, maxHeaderSize int32) { | ||
| // Capture packets on every interface. | ||
| packetCaptures := make([]<-chan gopacket.Packet, 0) | ||
| for _, iface := range interfaces { | ||
| // Open a packet capture | ||
| handle, err := opener(iface, maxHeaderSize, true, pcap.BlockForever) | ||
| rtx.Must(err, "Could not create libpcap client for %q", iface) | ||
| rtx.Must(handle.SetBPFFilter("tcp"), "Could not set up BPF filter for TCP") | ||
|
|
||
| // Stop packet capture when this function exits. | ||
| defer handle.Close() | ||
|
|
||
| // Save the packet capture channel. | ||
| packetCaptures = append(packetCaptures, gopacket.NewPacketSource(handle, layers.LinkTypeEthernet).Packets()) | ||
| } | ||
|
|
||
| // multiplex packets until all packet sources are exhausted or the context | ||
| // is cancelled. | ||
| muxPackets(ctx, packetCaptures, packets) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| package muxer | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/google/gopacket" | ||
| "github.com/google/gopacket/pcap" | ||
| "github.com/m-lab/go/rtx" | ||
| ) | ||
|
|
||
| func channelFromFile(fname string) <-chan gopacket.Packet { | ||
| // Get packets from a wireshark-produced pcap file. | ||
| handle, err := pcap.OpenOffline(fname) | ||
| rtx.Must(err, "Could not open golden pcap file %s", fname) | ||
| ps := gopacket.NewPacketSource(handle, handle.LinkType()) | ||
| return ps.Packets() | ||
| } | ||
|
|
||
| func TestMuxPacketsUntilSourceExhaustion(t *testing.T) { | ||
| // Open our two testfiles | ||
| ins := []<-chan gopacket.Packet{ | ||
| channelFromFile("../testdata/v4.pcap"), | ||
| channelFromFile("../testdata/v6.pcap"), | ||
| } | ||
| out := make(chan gopacket.Packet) | ||
| wg := sync.WaitGroup{} | ||
| wg.Add(1) | ||
| go func() { | ||
| // Mux the packets from each. | ||
| muxPackets(context.Background(), ins, out) | ||
| wg.Done() | ||
| }() | ||
| // The only way that this will close is if we exhaust all the input channels | ||
| // and they each close. So let's do that. | ||
| pcount := 0 | ||
| for range out { | ||
| pcount++ | ||
| } | ||
| wg.Wait() | ||
| // Verify that the combined flow contains the right number of packets. | ||
| if pcount != 20 { | ||
| t.Errorf("pcount should be 20, not %d", pcount) | ||
| } | ||
| } | ||
|
|
||
| func TestMuxPacketsUntilContextCancellation(t *testing.T) { | ||
| ins := []<-chan gopacket.Packet{ | ||
| make(chan gopacket.Packet), | ||
| make(chan gopacket.Packet), | ||
| } | ||
| out := make(chan gopacket.Packet) | ||
| wg := sync.WaitGroup{} | ||
| wg.Add(1) | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| go func() { | ||
| // Mux the packets from each. | ||
| muxPackets(ctx, ins, out) | ||
| wg.Done() | ||
| }() | ||
| go func() { | ||
| time.Sleep(100 * time.Millisecond) | ||
| cancel() | ||
| }() | ||
| // The input channels will never close, so only context cancellation will work. | ||
| pcount := 0 | ||
| for range out { | ||
| pcount++ | ||
| } | ||
| wg.Wait() | ||
| // If we got to here, then muxPackets terminated! Hooray! | ||
|
|
||
| // Verify that the combined flow contained no packets. | ||
| if pcount != 0 { | ||
| t.Errorf("pcount should be 0, not %d", pcount) | ||
| } | ||
|
|
||
| } | ||
|
|
||
| func fakePcapOpenLive(filename string, _ int32, _ bool, _ time.Duration) (*pcap.Handle, error) { | ||
| return pcap.OpenOffline(filename) | ||
| } | ||
|
|
||
| func TestMustCaptureOnInterfaces(t *testing.T) { | ||
| wg := sync.WaitGroup{} | ||
| packets := make(chan gopacket.Packet) | ||
| wg.Add(1) | ||
| go func() { | ||
| MustCaptureTCPOnInterfaces( | ||
| context.Background(), | ||
| []string{"../testdata/v4.pcap", "../testdata/v6.pcap"}, | ||
| packets, | ||
| fakePcapOpenLive, | ||
| 0, | ||
| ) | ||
| wg.Done() | ||
| }() | ||
|
|
||
| count := 0 | ||
| for range packets { | ||
| count++ | ||
| } | ||
| wg.Wait() | ||
| if count != 20 { | ||
| t.Errorf("Was supposed to see 20 packets, but instead saw %d", count) | ||
| } | ||
| } |