diff --git a/probe/endpoint/connection_tracker.go b/probe/endpoint/connection_tracker.go index 6f1facf24..e65c4d3c5 100644 --- a/probe/endpoint/connection_tracker.go +++ b/probe/endpoint/connection_tracker.go @@ -1,3 +1,5 @@ +// +build linux + package endpoint import ( @@ -5,28 +7,15 @@ import ( "time" log "github.com/sirupsen/logrus" + "github.com/typetypetype/conntrack" + "github.com/weaveworks/scope/probe/endpoint/procspy" "github.com/weaveworks/scope/probe/process" "github.com/weaveworks/scope/report" ) -// connectionTrackerConfig are the config options for the endpoint tracker. -type connectionTrackerConfig struct { - HostID string - HostName string - SpyProcs bool - UseConntrack bool - WalkProc bool - UseEbpfConn bool - ProcRoot string - BufferSize int - ProcessCache *process.CachingWalker - Scanner procspy.ConnectionScanner - DNSSnooper *DNSSnooper -} - type connectionTracker struct { - conf connectionTrackerConfig + conf ReporterConfig flowWalker flowWalker // Interface ebpfTracker *EbpfTracker reverseResolver *reverseResolver @@ -35,7 +24,7 @@ type connectionTracker struct { ebpfLastFailureTime time.Time } -func newConnectionTracker(conf connectionTrackerConfig) connectionTracker { +func newConnectionTracker(conf ReporterConfig) connectionTracker { ct := connectionTracker{ conf: conf, reverseResolver: newReverseResolver(), @@ -53,20 +42,20 @@ func newConnectionTracker(conf connectionTrackerConfig) connectionTracker { return ct } -func flowToTuple(f flow) (ft fourTuple) { +func flowToTuple(f conntrack.Conn) (ft fourTuple) { ft = fourTuple{ - f.Original.Layer3.SrcIP, - f.Original.Layer3.DstIP, - uint16(f.Original.Layer4.SrcPort), - uint16(f.Original.Layer4.DstPort), + f.Orig.Src.String(), + f.Orig.Dst.String(), + uint16(f.Orig.SrcPort), + uint16(f.Orig.DstPort), } // Handle DNAT-ed connections in the initial state - if f.Original.Layer3.DstIP != f.Reply.Layer3.SrcIP { + if !f.Orig.Dst.Equal(f.Reply.Src) { ft = fourTuple{ - f.Reply.Layer3.DstIP, - f.Reply.Layer3.SrcIP, - uint16(f.Reply.Layer4.DstPort), - uint16(f.Reply.Layer4.SrcPort), + f.Reply.Dst.String(), + f.Reply.Src.String(), + uint16(f.Reply.DstPort), + uint16(f.Reply.SrcPort), } } return ft @@ -78,7 +67,7 @@ func (t *connectionTracker) useProcfs() { t.conf.Scanner = procspy.NewConnectionScanner(t.conf.ProcessCache, t.conf.SpyProcs) } if t.flowWalker == nil { - t.flowWalker = newConntrackFlowWalker(t.conf.UseConntrack, t.conf.ProcRoot, t.conf.BufferSize) + t.flowWalker = newConntrackFlowWalker(t.conf.UseConntrack, t.conf.ProcRoot, t.conf.BufferSize, false /* natOnly */) } } @@ -118,7 +107,7 @@ func (t *connectionTracker) ReportConnections(rpt *report.Report) { // consult the flowWalker for short-lived (conntracked) connections seenTuples := map[string]fourTuple{} - t.flowWalker.walkFlows(func(f flow, alive bool) { + t.flowWalker.walkFlows(func(f conntrack.Conn, alive bool) { tuple := flowToTuple(f) seenTuples[tuple.key()] = tuple t.addConnection(rpt, false, tuple, "", nil, nil) @@ -135,10 +124,13 @@ func (t *connectionTracker) existingFlows() map[string]fourTuple { // log.Warnf("Not using conntrack: disabled") } else if err := IsConntrackSupported(t.conf.ProcRoot); err != nil { log.Warnf("Not using conntrack: not supported by the kernel: %s", err) - } else if existingFlows, err := existingConnections([]string{"--any-nat"}); err != nil { + } else if existingFlows, err := conntrack.ConnectionsSize(t.conf.BufferSize); err != nil { log.Errorf("conntrack existingConnections error: %v", err) } else { for _, f := range existingFlows { + if (f.Status & conntrack.IPS_NAT_MASK) == 0 { + continue + } tuple := flowToTuple(f) seenTuples[tuple.key()] = tuple } diff --git a/probe/endpoint/conntrack.go b/probe/endpoint/conntrack.go index 95f6f5531..155dfa41c 100644 --- a/probe/endpoint/conntrack.go +++ b/probe/endpoint/conntrack.go @@ -1,93 +1,51 @@ +// +build linux + package endpoint import ( - "bufio" - "bytes" "fmt" - "io" "io/ioutil" "path/filepath" - "strconv" - "strings" "sync" "time" - "unicode" log "github.com/sirupsen/logrus" - - "github.com/weaveworks/common/exec" + "github.com/typetypetype/conntrack" ) const ( // From https://www.kernel.org/doc/Documentation/networking/nf_conntrack-sysctl.txt eventsPath = "sys/net/netfilter/nf_conntrack_events" - - timeWait = "TIME_WAIT" - tcpProto = "tcp" - newType = "[NEW]" - updateType = "[UPDATE]" - destroyType = "[DESTROY]" + timeWait = "TIME_WAIT" + tcpClose = "CLOSE" + tcpProto = 6 ) -var ( - destroyTypeB = []byte(destroyType) - assured = []byte("[ASSURED] ") - unreplied = []byte("[UNREPLIED] ") -) - -type layer3 struct { - SrcIP string - DstIP string -} - -type layer4 struct { - SrcPort int - DstPort int - Proto string -} - -type meta struct { - Layer3 layer3 - Layer4 layer4 - ID int64 - State string -} - -type flow struct { - Type string - Original, Reply, Independent meta -} - -type conntrack struct { - Flows []flow -} - // flowWalker is something that maintains flows, and provides an accessor // method to walk them. type flowWalker interface { - walkFlows(f func(f flow, active bool)) + walkFlows(f func(conntrack.Conn, bool)) stop() } type nilFlowWalker struct{} -func (n nilFlowWalker) stop() {} -func (n nilFlowWalker) walkFlows(f func(flow, bool)) {} +func (n nilFlowWalker) stop() {} +func (n nilFlowWalker) walkFlows(f func(conntrack.Conn, bool)) {} -// conntrackWalker uses the conntrack command to track network connections and +// conntrackWalker uses conntrack (via netlink) to track network connections and // implement flowWalker. type conntrackWalker struct { sync.Mutex - cmd exec.Cmd - activeFlows map[int64]flow // active flows in state != TIME_WAIT - bufferedFlows []flow // flows coming out of activeFlows spend 1 walk cycle here + activeFlows map[uint32]conntrack.Conn // active flows in state != TIME_WAIT + bufferedFlows []conntrack.Conn // flows coming out of activeFlows spend 1 walk cycle here bufferSize int - args []string + natOnly bool quit chan struct{} } // newConntracker creates and starts a new conntracker. -func newConntrackFlowWalker(useConntrack bool, procRoot string, bufferSize int, args ...string) flowWalker { +func newConntrackFlowWalker(useConntrack bool, procRoot string, bufferSize int, natOnly bool) flowWalker { if !useConntrack { return nilFlowWalker{} } else if err := IsConntrackSupported(procRoot); err != nil { @@ -95,9 +53,9 @@ func newConntrackFlowWalker(useConntrack bool, procRoot string, bufferSize int, return nilFlowWalker{} } result := &conntrackWalker{ - activeFlows: map[int64]flow{}, + activeFlows: map[uint32]conntrack.Conn{}, bufferSize: bufferSize, - args: args, + natOnly: natOnly, quit: make(chan struct{}), } go result.loop() @@ -144,318 +102,81 @@ func (c *conntrackWalker) clearFlows() { c.bufferedFlows = append(c.bufferedFlows, f) } - c.activeFlows = map[int64]flow{} + c.activeFlows = map[uint32]conntrack.Conn{} } -func logPipe(prefix string, reader io.Reader) { - scanner := bufio.NewScanner(reader) - for scanner.Scan() { - log.Error(prefix, scanner.Text()) - } - if err := scanner.Err(); err != nil { - log.Error(prefix, err) +func (c *conntrackWalker) relevant(f conntrack.Conn) bool { + // For now, we're only interested in tcp connections - there is too much udp + // traffic going on (every container talking to dns, for example) to + // render nicely. TODO: revisit this. + if f.Orig.Proto != tcpProto { + return false } + return !(c.natOnly && (f.Status&conntrack.IPS_NAT_MASK) == 0) } func (c *conntrackWalker) run() { - // Fork another conntrack, just to capture existing connections - // for which we don't get events - existingFlows, err := existingConnections(c.args) + existingFlows, err := conntrack.ConnectionsSize(c.bufferSize) if err != nil { - log.Errorf("conntrack existingConnections error: %v", err) + log.Errorf("conntrack Connections error: %v", err) return } - for _, flow := range existingFlows { - c.handleFlow(flow, true) - } - - args := append([]string{ - "--buffer-size", strconv.Itoa(c.bufferSize), "-E", - "-o", "id", "-p", "tcp"}, c.args..., - ) - cmd := exec.Command("conntrack", args...) - stdout, err := cmd.StdoutPipe() - if err != nil { - log.Errorf("conntrack error: %v", err) - return - } - - stderr, err := cmd.StderrPipe() - if err != nil { - log.Errorf("conntrack error: %v", err) - return - } - go logPipe("conntrack stderr:", stderr) - - if err := cmd.Start(); err != nil { - log.Errorf("conntrack error: %v", err) - return - } - - defer func() { - if err := cmd.Wait(); err != nil { - log.Errorf("conntrack error: %v", err) - } - }() - c.Lock() - // We may have stopped in the mean time, - // so check to see if the channel is open - // under the lock. - select { - default: - case <-c.quit: - return + for _, flow := range existingFlows { + if c.relevant(flow) && flow.TCPState != tcpClose && flow.TCPState != timeWait { + c.activeFlows[flow.CtId] = flow + } } - c.cmd = cmd c.Unlock() - scanner := bufio.NewScanner(bufio.NewReader(stdout)) + events, stop, err := conntrack.FollowSize(c.bufferSize, conntrack.NF_NETLINK_CONNTRACK_UPDATE|conntrack.NF_NETLINK_CONNTRACK_DESTROY) + if err != nil { + log.Errorf("conntrack Follow error: %v", err) + return + } + defer log.Infof("conntrack exiting") - // Loop on the output stream + // Handle conntrack events from netlink socket for { - f, err := decodeStreamedFlow(scanner) - if err != nil { - log.Errorf("conntrack error: %v", err) + select { + case <-c.quit: + stop() return - } - c.handleFlow(f, false) - } -} - -// Get a line without [ASSURED]/[UNREPLIED] tags (it simplifies parsing) -func getUntaggedLine(scanner *bufio.Scanner) ([]byte, error) { - success := scanner.Scan() - if !success { - if err := scanner.Err(); err != nil { - return nil, err - } - return nil, io.EOF - } - line := scanner.Bytes() - // Remove [ASSURED]/[UNREPLIED] tags - line = removeInplace(line, assured) - line = removeInplace(line, unreplied) - return line, nil -} - -func removeInplace(s, sep []byte) []byte { - // TODO: See if we can get better performance - // removing multiple substrings at once (with index/suffixarray New()+Lookup()) - // Probably not worth it for only two substrings occurring once. - index := bytes.Index(s, sep) - if index < 0 { - return s - } - copy(s[index:], s[index+len(sep):]) - return s[:len(s)-len(sep)] -} - -// decodeFlowKeyValues parses the key-values from a conntrack line and updates the flow -// It only considers the following key-values: -// src=127.0.0.1 dst=127.0.0.1 sport=58958 dport=6784 src=127.0.0.1 dst=127.0.0.1 sport=6784 dport=58958 id=1595499776 -// Keys can be present twice, so the order is important. -// Conntrack could add other key-values such as secctx=, packets=, bytes=. Those are ignored. -func decodeFlowKeyValues(line []byte, f *flow) error { - var err error - for _, field := range strings.FieldsFunc(string(line), func(c rune) bool { return unicode.IsSpace(c) }) { - kv := strings.SplitN(field, "=", 2) - if len(kv) != 2 { - continue - } - key := kv[0] - value := kv[1] - firstTupleSet := f.Original.Layer4.DstPort != 0 - switch { - case key == "src": - if !firstTupleSet { - f.Original.Layer3.SrcIP = value - } else { - f.Reply.Layer3.SrcIP = value + case f, ok := <-events: + if !ok { + return } - - case key == "dst": - if !firstTupleSet { - f.Original.Layer3.DstIP = value - } else { - f.Reply.Layer3.DstIP = value + if c.relevant(f) { + c.handleFlow(f) } - - case key == "sport": - if !firstTupleSet { - f.Original.Layer4.SrcPort, err = strconv.Atoi(value) - } else { - f.Reply.Layer4.SrcPort, err = strconv.Atoi(value) - } - - case key == "dport": - if !firstTupleSet { - f.Original.Layer4.DstPort, err = strconv.Atoi(value) - } else { - f.Reply.Layer4.DstPort, err = strconv.Atoi(value) - } - - case key == "id": - f.Independent.ID, err = strconv.ParseInt(value, 10, 64) } } - - return err -} - -func decodeStreamedFlow(scanner *bufio.Scanner) (flow, error) { - var ( - // Use ints for parsing unused fields since their allocations - // are almost for free - unused [2]int - f flow - ) - - // Examples: - // " [UPDATE] udp 17 29 src=192.168.2.100 dst=192.168.2.1 sport=57767 dport=53 src=192.168.2.1 dst=192.168.2.100 sport=53 dport=57767" - // " [NEW] tcp 6 120 SYN_SENT src=127.0.0.1 dst=127.0.0.1 sport=58958 dport=6784 [UNREPLIED] src=127.0.0.1 dst=127.0.0.1 sport=6784 dport=58958 id=1595499776" - // " [UPDATE] tcp 6 120 TIME_WAIT src=10.0.2.15 dst=10.0.2.15 sport=51154 dport=4040 src=10.0.2.15 dst=10.0.2.15 sport=4040 dport=51154 [ASSURED] id=3663628160" - // " [DESTROY] tcp 6 src=172.17.0.1 dst=172.17.0.1 sport=34078 dport=53 src=172.17.0.1 dst=10.0.2.15 sport=53 dport=34078 id=3668417984" (note how the timeout and state field is missing) - - // Remove tags since they are optional and make parsing harder - line, err := getUntaggedLine(scanner) - if err != nil { - return flow{}, err - } - - line = bytes.TrimLeft(line, " ") - if bytes.HasPrefix(line, destroyTypeB) { - // Destroy events don't have a timeout or state field - _, err = fmt.Sscanf(string(line), "%s %s %d", - &f.Type, - &f.Original.Layer4.Proto, - &unused[0], - ) - } else { - _, err = fmt.Sscanf(string(line), "%s %s %d %d %s", - &f.Type, - &f.Original.Layer4.Proto, - &unused[0], - &unused[1], - &f.Independent.State, - ) - } - if err != nil { - return flow{}, fmt.Errorf("Error parsing streamed flow %q: %v ", line, err) - } - - err = decodeFlowKeyValues(line, &f) - if err != nil { - return flow{}, fmt.Errorf("Error parsing streamed flow %q: %v ", line, err) - } - - f.Reply.Layer4.Proto = f.Original.Layer4.Proto - return f, nil -} - -func existingConnections(conntrackWalkerArgs []string) ([]flow, error) { - args := append([]string{"-L", "-o", "id", "-p", "tcp"}, conntrackWalkerArgs...) - cmd := exec.Command("conntrack", args...) - stdout, err := cmd.StdoutPipe() - if err != nil { - return []flow{}, err - } - if err := cmd.Start(); err != nil { - return []flow{}, err - } - defer func() { - if err := cmd.Wait(); err != nil { - log.Errorf("conntrack existingConnections exit error: %v", err) - } - }() - - scanner := bufio.NewScanner(bufio.NewReader(stdout)) - var result []flow - for { - f, err := decodeDumpedFlow(scanner) - if err != nil { - if err == io.EOF { - break - } - log.Errorf("conntrack error: %v", err) - return result, err - } - result = append(result, f) - } - return result, nil -} - -func decodeDumpedFlow(scanner *bufio.Scanner) (flow, error) { - var ( - // Use ints for parsing unused fields since allocations - // are almost for free - unused [4]int - f flow - ) - - // Examples with different formats: - // With SELinux, there is a "secctx=" - // After "sudo sysctl net.netfilter.nf_conntrack_acct=1", there is "packets=" and "bytes=" - // - // "tcp 6 431997 ESTABLISHED src=10.32.0.1 dst=10.32.0.1 sport=50274 dport=4040 src=10.32.0.1 dst=10.32.0.1 sport=4040 dport=50274 [ASSURED] mark=0 use=1 id=407401088" - // "tcp 6 431998 ESTABLISHED src=10.0.2.2 dst=10.0.2.15 sport=49911 dport=22 src=10.0.2.15 dst=10.0.2.2 sport=22 dport=49911 [ASSURED] mark=0 use=1 id=2993966208" - // "tcp 6 108 ESTABLISHED src=172.17.0.5 dst=172.17.0.2 sport=47010 dport=80 src=172.17.0.2 dst=172.17.0.5 sport=80 dport=47010 [ASSURED] mark=0 secctx=system_u:object_r:unlabeled_t:s0 use=1 id=4001098880" - // "tcp 6 431970 ESTABLISHED src=192.168.35.116 dst=216.58.213.227 sport=49862 dport=443 packets=11 bytes=1337 src=216.58.213.227 dst=192.168.35.116 sport=443 dport=49862 packets=8 bytes=716 [ASSURED] mark=0 secctx=system_u:object_r:unlabeled_t:s0 use=1 id=943643840" - - // remove tags since they are optional and make parsing harder - line, err := getUntaggedLine(scanner) - if err != nil { - return flow{}, err - } - - _, err = fmt.Sscanf(string(line), "%s %d %d %s", &f.Original.Layer4.Proto, &unused[0], &unused[1], &f.Independent.State) - if err != nil { - return flow{}, fmt.Errorf("Error parsing dumped flow %q: %v ", line, err) - } - - err = decodeFlowKeyValues(line, &f) - if err != nil { - return flow{}, fmt.Errorf("Error parsing dumped flow %q: %v ", line, err) - } - - f.Reply.Layer4.Proto = f.Original.Layer4.Proto - return f, nil } func (c *conntrackWalker) stop() { c.Lock() defer c.Unlock() close(c.quit) - if c.cmd != nil { - c.cmd.Kill() - } } -func (c *conntrackWalker) handleFlow(f flow, forceAdd bool) { +func (c *conntrackWalker) handleFlow(f conntrack.Conn) { c.Lock() defer c.Unlock() - // For not, I'm only interested in tcp connections - there is too much udp - // traffic going on (every container talking to weave dns, for example) to - // render nicely. TODO: revisit this. - if f.Original.Layer4.Proto != tcpProto { - return - } - // Ignore flows for which we never saw an update; they are likely // incomplete or wrong. See #1462. switch { - case forceAdd || f.Type == updateType: - if f.Independent.State != timeWait { - c.activeFlows[f.Independent.ID] = f - } else if _, ok := c.activeFlows[f.Independent.ID]; ok { - delete(c.activeFlows, f.Independent.ID) + case f.MsgType == conntrack.NfctMsgUpdate: + if f.TCPState != timeWait { + c.activeFlows[f.CtId] = f + } else if _, ok := c.activeFlows[f.CtId]; ok { + delete(c.activeFlows, f.CtId) c.bufferedFlows = append(c.bufferedFlows, f) } - case f.Type == destroyType: - if active, ok := c.activeFlows[f.Independent.ID]; ok { - delete(c.activeFlows, f.Independent.ID) + case f.MsgType == conntrack.NfctMsgDestroy: + if active, ok := c.activeFlows[f.CtId]; ok { + delete(c.activeFlows, f.CtId) c.bufferedFlows = append(c.bufferedFlows, active) } } @@ -463,7 +184,7 @@ func (c *conntrackWalker) handleFlow(f flow, forceAdd bool) { // walkFlows calls f with all active flows and flows that have come and gone // since the last call to walkFlows -func (c *conntrackWalker) walkFlows(f func(flow, bool)) { +func (c *conntrackWalker) walkFlows(f func(conntrack.Conn, bool)) { c.Lock() defer c.Unlock() for _, flow := range c.activeFlows { diff --git a/probe/endpoint/conntrack_internal_test.go b/probe/endpoint/conntrack_internal_test.go deleted file mode 100644 index a2aa25397..000000000 --- a/probe/endpoint/conntrack_internal_test.go +++ /dev/null @@ -1,253 +0,0 @@ -package endpoint - -import ( - "bufio" - "io" - "strings" - "testing" - "time" - - "github.com/weaveworks/scope/test" -) - -// Obtained though conntrack -E -p tcp -o id and then tweaked -const streamedFlowsSource = `[DESTROY] tcp 6 src=10.0.0.1 dst=127.0.0.1 sport=36826 dport=28106 src=10.0.0.2 dst=127.0.0.2 sport=28107 dport=36826 [ASSURED] id=347275904 - [NEW] tcp 6 120 SYN_SENT src=10.0.0.1 dst=127.0.0.1 sport=36898 dport=28107 [UNREPLIED] src=10.0.0.2 dst=127.0.0.2 sport=28107 dport=36898 id=347275904 - [UPDATE] tcp 6 60 SYN_RECV src=10.0.0.1 dst=127.0.0.1 sport=36898 dport=28107 src=10.0.0.2 dst=127.0.0.2 sport=28107 dport=36898 id=347275904 - [UPDATE] tcp 6 432000 ESTABLISHED src=10.0.0.1 dst=127.0.0.1 sport=36898 dport=28107 src=10.0.0.2 dst=127.0.0.2 sport=28107 dport=36898 [ASSURED] id=347275904` - -var wantStreamedFlows = []flow{ - { - Type: destroyType, - Original: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.1", - DstIP: "127.0.0.1", - }, - Layer4: layer4{ - SrcPort: 36826, - DstPort: 28106, - Proto: "tcp", - }, - }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.2", - DstIP: "127.0.0.2", - }, - Layer4: layer4{ - SrcPort: 28107, - DstPort: 36826, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 347275904, - }, - }, - { - Type: newType, - Original: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.1", - DstIP: "127.0.0.1", - }, - Layer4: layer4{ - SrcPort: 36898, - DstPort: 28107, - Proto: "tcp", - }, - }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.2", - DstIP: "127.0.0.2", - }, - Layer4: layer4{ - SrcPort: 28107, - DstPort: 36898, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 347275904, - State: "SYN_SENT", - }, - }, - { - Type: updateType, - Original: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.1", - DstIP: "127.0.0.1", - }, - Layer4: layer4{ - SrcPort: 36898, - DstPort: 28107, - Proto: "tcp", - }, - }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.2", - DstIP: "127.0.0.2", - }, - Layer4: layer4{ - SrcPort: 28107, - DstPort: 36898, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 347275904, - State: "SYN_RECV", - }, - }, - { - Type: updateType, - Original: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.1", - DstIP: "127.0.0.1", - }, - Layer4: layer4{ - SrcPort: 36898, - DstPort: 28107, - Proto: "tcp", - }, - }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "10.0.0.2", - DstIP: "127.0.0.2", - }, - Layer4: layer4{ - SrcPort: 28107, - DstPort: 36898, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 347275904, - State: "ESTABLISHED", - }, - }, -} - -func testFlowDecoding(t *testing.T, source string, want []flow, decoder func(scanner *bufio.Scanner) (flow, error)) { - scanner := bufio.NewScanner(strings.NewReader(source)) - d := time.Millisecond * 100 - for _, wantFlow := range want { - haveFlow, err := decoder(scanner) - if err != nil { - t.Fatalf("Unexpected decoding error: %v", err) - } - test.Poll(t, d, wantFlow, func() interface{} { return haveFlow }) - } - - if _, err := decodeStreamedFlow(scanner); err != io.EOF { - t.Fatalf("Unexpected error value on empty input: %v", err) - } -} - -func TestStreamedFlowDecoding(t *testing.T) { - testFlowDecoding(t, streamedFlowsSource, wantStreamedFlows, decodeStreamedFlow) -} - -// Obtained through conntrack -L -p tcp -o id -// With SELinux, there is a "secctx=" -// After "sudo sysctl net.netfilter.nf_conntrack_acct=1", there is "packets=" and "bytes=" -const dumpedFlowsSource = `tcp 6 431998 ESTABLISHED src=10.0.2.2 dst=10.0.2.15 sport=49911 dport=22 src=10.0.2.15 dst=10.0.2.2 sport=22 dport=49911 [ASSURED] mark=0 use=1 id=2993966208 -tcp 6 108 ESTABLISHED src=172.17.0.5 dst=172.17.0.2 sport=47010 dport=80 src=172.17.0.2 dst=172.17.0.5 sport=80 dport=47010 [ASSURED] mark=0 secctx=system_u:object_r:unlabeled_t:s0 use=1 id=4001098880 -tcp 6 431970 ESTABLISHED src=192.168.35.116 dst=216.58.213.227 sport=49862 dport=443 packets=11 bytes=1337 src=216.58.213.227 dst=192.168.35.116 sport=443 dport=49862 packets=8 bytes=716 [ASSURED] mark=0 secctx=system_u:object_r:unlabeled_t:s0 use=1 id=943643840` - -var wantDumpedFlows = []flow{ - { - Original: meta{ - Layer3: layer3{ - SrcIP: "10.0.2.2", - DstIP: "10.0.2.15", - }, - Layer4: layer4{ - SrcPort: 49911, - DstPort: 22, - Proto: "tcp", - }, - }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "10.0.2.15", - DstIP: "10.0.2.2", - }, - Layer4: layer4{ - SrcPort: 22, - DstPort: 49911, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 2993966208, - State: "ESTABLISHED", - }, - }, - { - Original: meta{ - Layer3: layer3{ - SrcIP: "172.17.0.5", - DstIP: "172.17.0.2", - }, - Layer4: layer4{ - SrcPort: 47010, - DstPort: 80, - Proto: "tcp", - }, - }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "172.17.0.2", - DstIP: "172.17.0.5", - }, - Layer4: layer4{ - SrcPort: 80, - DstPort: 47010, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 4001098880, - State: "ESTABLISHED", - }, - }, - { - Original: meta{ - Layer3: layer3{ - SrcIP: "192.168.35.116", - DstIP: "216.58.213.227", - }, - Layer4: layer4{ - SrcPort: 49862, - DstPort: 443, - Proto: "tcp", - }, - }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "216.58.213.227", - DstIP: "192.168.35.116", - }, - Layer4: layer4{ - SrcPort: 443, - DstPort: 49862, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 943643840, - State: "ESTABLISHED", - }, - }, -} - -func TestDumpedFlowDecoding(t *testing.T) { - testFlowDecoding(t, dumpedFlowsSource, wantDumpedFlows, decodeDumpedFlow) -} diff --git a/probe/endpoint/ebpf.go b/probe/endpoint/ebpf.go index 4dd173020..2a1e1aefe 100644 --- a/probe/endpoint/ebpf.go +++ b/probe/endpoint/ebpf.go @@ -1,3 +1,5 @@ +// +build linux + package endpoint import ( diff --git a/probe/endpoint/ebpf_test.go b/probe/endpoint/ebpf_test.go index 52baae8d3..cbca96799 100644 --- a/probe/endpoint/ebpf_test.go +++ b/probe/endpoint/ebpf_test.go @@ -1,3 +1,5 @@ +// +build linux + package endpoint import ( diff --git a/probe/endpoint/nat.go b/probe/endpoint/nat.go index 52073a209..797a0d1a8 100644 --- a/probe/endpoint/nat.go +++ b/probe/endpoint/nat.go @@ -1,19 +1,24 @@ +// +build linux + package endpoint import ( + "net" "strconv" + "github.com/typetypetype/conntrack" + "github.com/weaveworks/scope/report" ) // This is our 'abstraction' of the endpoint that have been rewritten by NAT. // Original is the private IP that has been rewritten. type endpointMapping struct { - originalIP string - originalPort int + originalIP net.IP + originalPort uint16 - rewrittenIP string - rewrittenPort int + rewrittenIP net.IP + rewrittenPort uint16 } // natMapper rewrites a report to deal with NAT'd connections. @@ -25,21 +30,21 @@ func makeNATMapper(fw flowWalker) natMapper { return natMapper{fw} } -func toMapping(f flow) *endpointMapping { +func toMapping(f conntrack.Conn) *endpointMapping { var mapping endpointMapping - if f.Original.Layer3.SrcIP == f.Reply.Layer3.DstIP { + if f.Orig.Src.Equal(f.Reply.Dst) { mapping = endpointMapping{ - originalIP: f.Reply.Layer3.SrcIP, - originalPort: f.Reply.Layer4.SrcPort, - rewrittenIP: f.Original.Layer3.DstIP, - rewrittenPort: f.Original.Layer4.DstPort, + originalIP: f.Reply.Src, + originalPort: f.Reply.SrcPort, + rewrittenIP: f.Orig.Dst, + rewrittenPort: f.Orig.DstPort, } } else { mapping = endpointMapping{ - originalIP: f.Original.Layer3.SrcIP, - originalPort: f.Original.Layer4.SrcPort, - rewrittenIP: f.Reply.Layer3.DstIP, - rewrittenPort: f.Reply.Layer4.DstPort, + originalIP: f.Orig.Src, + originalPort: f.Orig.SrcPort, + rewrittenIP: f.Reply.Dst, + rewrittenPort: f.Reply.DstPort, } } @@ -49,13 +54,13 @@ func toMapping(f flow) *endpointMapping { // applyNAT duplicates Nodes in the endpoint topology of a report, based on // the NAT table. func (n natMapper) applyNAT(rpt report.Report, scope string) { - n.flowWalker.walkFlows(func(f flow, active bool) { + n.flowWalker.walkFlows(func(f conntrack.Conn, _ bool) { mapping := toMapping(f) - realEndpointPort := strconv.Itoa(mapping.originalPort) - copyEndpointPort := strconv.Itoa(mapping.rewrittenPort) - realEndpointID := report.MakeEndpointNodeID(scope, "", mapping.originalIP, realEndpointPort) - copyEndpointID := report.MakeEndpointNodeID(scope, "", mapping.rewrittenIP, copyEndpointPort) + realEndpointPort := strconv.Itoa(int(mapping.originalPort)) + copyEndpointPort := strconv.Itoa(int(mapping.rewrittenPort)) + realEndpointID := report.MakeEndpointNodeID(scope, "", mapping.originalIP.String(), realEndpointPort) + copyEndpointID := report.MakeEndpointNodeID(scope, "", mapping.rewrittenIP.String(), copyEndpointPort) node, ok := rpt.Endpoint.Nodes[realEndpointID] if !ok { diff --git a/probe/endpoint/nat_internal_test.go b/probe/endpoint/nat_internal_test.go index 51fda2605..d4f3c4354 100644 --- a/probe/endpoint/nat_internal_test.go +++ b/probe/endpoint/nat_internal_test.go @@ -1,8 +1,14 @@ +// +build linux + package endpoint import ( + "net" + "syscall" "testing" + "github.com/typetypetype/conntrack" + "github.com/weaveworks/common/mtime" "github.com/weaveworks/common/test" "github.com/weaveworks/scope/report" @@ -10,10 +16,10 @@ import ( ) type mockFlowWalker struct { - flows []flow + flows []conntrack.Conn } -func (m *mockFlowWalker) walkFlows(f func(f flow, active bool)) { +func (m *mockFlowWalker) walkFlows(f func(f conntrack.Conn, active bool)) { for _, flow := range m.flows { f(flow, true) } @@ -32,39 +38,34 @@ func TestNat(t *testing.T) { // container2 (10.0.47.2:22222), host2 (2.3.4.5:22223) -> // host1 (1.2.3.4:80), container1 (10.0.47.1:80) + c1 := net.ParseIP("10.0.47.1") + c2 := net.ParseIP("10.0.47.2") + host2 := net.ParseIP("2.3.4.5") + host1 := net.ParseIP("1.2.3.4") + // from the PoV of host1 { - f := flow{ - Type: updateType, - Original: meta{ - Layer3: layer3{ - SrcIP: "2.3.4.5", - DstIP: "1.2.3.4", - }, - Layer4: layer4{ - SrcPort: 22222, - DstPort: 80, - Proto: "tcp", - }, + f := conntrack.Conn{ + MsgType: conntrack.NfctMsgUpdate, + Orig: conntrack.Tuple{ + Src: host2, + Dst: host1, + SrcPort: 22222, + DstPort: 80, + Proto: syscall.IPPROTO_TCP, }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "10.0.47.1", - DstIP: "2.3.4.5", - }, - Layer4: layer4{ - SrcPort: 80, - DstPort: 22222, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 1, + Reply: conntrack.Tuple{ + Src: c1, + Dst: host2, + SrcPort: 80, + DstPort: 22222, + Proto: syscall.IPPROTO_TCP, }, + CtId: 1, } ct := &mockFlowWalker{ - flows: []flow{f}, + flows: []conntrack.Conn{f}, } have := report.MakeReport() @@ -88,36 +89,26 @@ func TestNat(t *testing.T) { // form the PoV of host2 { - f := flow{ - Type: updateType, - Original: meta{ - Layer3: layer3{ - SrcIP: "10.0.47.2", - DstIP: "1.2.3.4", - }, - Layer4: layer4{ - SrcPort: 22222, - DstPort: 80, - Proto: "tcp", - }, + f := conntrack.Conn{ + MsgType: conntrack.NfctMsgUpdate, + Orig: conntrack.Tuple{ + Src: c2, + Dst: host1, + SrcPort: 22222, + DstPort: 80, + Proto: syscall.IPPROTO_TCP, }, - Reply: meta{ - Layer3: layer3{ - SrcIP: "1.2.3.4", - DstIP: "2.3.4.5", - }, - Layer4: layer4{ - SrcPort: 80, - DstPort: 22223, - Proto: "tcp", - }, - }, - Independent: meta{ - ID: 2, + Reply: conntrack.Tuple{ + Src: host1, + Dst: host2, + SrcPort: 80, + DstPort: 22223, + Proto: syscall.IPPROTO_TCP, }, + CtId: 2, } ct := &mockFlowWalker{ - flows: []flow{f}, + flows: []conntrack.Conn{f}, } have := report.MakeReport() diff --git a/probe/endpoint/reporter.go b/probe/endpoint/reporter.go index 86b52dbf5..709dc0f41 100644 --- a/probe/endpoint/reporter.go +++ b/probe/endpoint/reporter.go @@ -31,13 +31,6 @@ type ReporterConfig struct { DNSSnooper *DNSSnooper } -// Reporter generates Reports containing the Endpoint topology. -type Reporter struct { - conf ReporterConfig - connectionTracker connectionTracker - natMapper natMapper -} - // SpyDuration is an exported prometheus metric var SpyDuration = prometheus.NewSummaryVec( prometheus.SummaryOpts{ @@ -50,52 +43,5 @@ var SpyDuration = prometheus.NewSummaryVec( []string{}, ) -// NewReporter creates a new Reporter that invokes procspy.Connections to -// generate a report.Report that contains every discovered (spied) connection -// on the host machine, at the granularity of host and port. That information -// is stored in the Endpoint topology. It optionally enriches that topology -// with process (PID) information. -func NewReporter(conf ReporterConfig) *Reporter { - return &Reporter{ - conf: conf, - connectionTracker: newConnectionTracker(connectionTrackerConfig{ - HostID: conf.HostID, - HostName: conf.HostName, - SpyProcs: conf.SpyProcs, - UseConntrack: conf.UseConntrack, - WalkProc: conf.WalkProc, - UseEbpfConn: conf.UseEbpfConn, - ProcRoot: conf.ProcRoot, - BufferSize: conf.BufferSize, - ProcessCache: conf.ProcessCache, - Scanner: conf.Scanner, - DNSSnooper: conf.DNSSnooper, - }), - natMapper: makeNATMapper(newConntrackFlowWalker(conf.UseConntrack, conf.ProcRoot, conf.BufferSize, "--any-nat")), - } -} - // Name of this reporter, for metrics gathering func (Reporter) Name() string { return "Endpoint" } - -// Stop stop stop -func (r *Reporter) Stop() { - r.connectionTracker.Stop() - r.natMapper.stop() - if r.conf.Scanner != nil { - r.conf.Scanner.Stop() - } -} - -// Report implements Reporter. -func (r *Reporter) Report() (report.Report, error) { - defer func(begin time.Time) { - SpyDuration.WithLabelValues().Observe(time.Since(begin).Seconds()) - }(time.Now()) - - rpt := report.MakeReport() - - r.connectionTracker.ReportConnections(&rpt) - r.natMapper.applyNAT(rpt, r.conf.HostID) - return rpt, nil -} diff --git a/probe/endpoint/reporter_linux.go b/probe/endpoint/reporter_linux.go new file mode 100644 index 000000000..dd7444c2b --- /dev/null +++ b/probe/endpoint/reporter_linux.go @@ -0,0 +1,51 @@ +// +build linux + +package endpoint + +import ( + "time" + + "github.com/weaveworks/scope/report" +) + +// Reporter generates Reports containing the Endpoint topology. +type Reporter struct { + conf ReporterConfig + connectionTracker connectionTracker + natMapper natMapper +} + +// NewReporter creates a new Reporter that invokes procspy.Connections to +// generate a report.Report that contains every discovered (spied) connection +// on the host machine, at the granularity of host and port. That information +// is stored in the Endpoint topology. It optionally enriches that topology +// with process (PID) information. +func NewReporter(conf ReporterConfig) *Reporter { + return &Reporter{ + conf: conf, + connectionTracker: newConnectionTracker(conf), + natMapper: makeNATMapper(newConntrackFlowWalker(conf.UseConntrack, conf.ProcRoot, conf.BufferSize, true /* natOnly */)), + } +} + +// Stop stop stop +func (r *Reporter) Stop() { + r.connectionTracker.Stop() + r.natMapper.stop() + if r.conf.Scanner != nil { + r.conf.Scanner.Stop() + } +} + +// Report implements Reporter. +func (r *Reporter) Report() (report.Report, error) { + defer func(begin time.Time) { + SpyDuration.WithLabelValues().Observe(time.Since(begin).Seconds()) + }(time.Now()) + + rpt := report.MakeReport() + + r.connectionTracker.ReportConnections(&rpt) + r.natMapper.applyNAT(rpt, r.conf.HostID) + return rpt, nil +} diff --git a/probe/endpoint/reporter_other.go b/probe/endpoint/reporter_other.go new file mode 100644 index 000000000..bf77b5d64 --- /dev/null +++ b/probe/endpoint/reporter_other.go @@ -0,0 +1,21 @@ +// +build !linux + +package endpoint + +import "github.com/weaveworks/scope/report" + +// Reporter dummy +type Reporter struct{} + +// NewReporter makes a dummy +func NewReporter(conf ReporterConfig) *Reporter { + return &Reporter{} +} + +// Stop dummy +func (r *Reporter) Stop() {} + +// Report implements Reporter. +func (r *Reporter) Report() (report.Report, error) { + return report.MakeReport(), nil +} diff --git a/vendor/github.com/typetypetype/conntrack/LICENSE b/vendor/github.com/typetypetype/conntrack/LICENSE new file mode 100644 index 000000000..4eb197f17 --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/LICENSE @@ -0,0 +1,22 @@ +The MIT License (MIT) + +Copyright (c) 2015 typetypetype + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. + diff --git a/vendor/github.com/typetypetype/conntrack/cli/main.go b/vendor/github.com/typetypetype/conntrack/cli/main.go new file mode 100644 index 000000000..e87687ef4 --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/cli/main.go @@ -0,0 +1,33 @@ +package main + +// Example usage + +import ( + "fmt" + "time" + + "github.com/typetypetype/conntrack" +) + +func main() { + cs, err := conntrack.Established() + if err != nil { + panic(fmt.Sprintf("Established: %s", err)) + } + fmt.Printf("Established on start:\n") + for _, cn := range cs { + fmt.Printf(" - %s\n", cn) + } + fmt.Println("") + + c, err := conntrack.New() + if err != nil { + panic(err) + } + for range time.Tick(1 * time.Second) { + fmt.Printf("Connections:\n") + for _, cn := range c.Connections() { + fmt.Printf(" - %s\n", cn) + } + } +} diff --git a/vendor/github.com/typetypetype/conntrack/client.go b/vendor/github.com/typetypetype/conntrack/client.go new file mode 100644 index 000000000..e5789f5df --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/client.go @@ -0,0 +1,466 @@ +package conntrack + +import ( + "encoding/binary" + "errors" + "fmt" + "net" + "strconv" + "syscall" + "unsafe" + + "golang.org/x/sys/unix" +) + +const ( + sizeofGenmsg = uint32(unsafe.Sizeof(unix.Nfgenmsg{})) // TODO +) + +type ConntrackListReq struct { + Header syscall.NlMsghdr + Body unix.Nfgenmsg +} + +func (c *ConntrackListReq) toWireFormat() []byte { + // adapted from syscall/NetlinkRouteRequest.toWireFormat + b := make([]byte, c.Header.Len) + *(*uint32)(unsafe.Pointer(&b[0:4][0])) = c.Header.Len + *(*uint16)(unsafe.Pointer(&b[4:6][0])) = c.Header.Type + *(*uint16)(unsafe.Pointer(&b[6:8][0])) = c.Header.Flags + *(*uint32)(unsafe.Pointer(&b[8:12][0])) = c.Header.Seq + *(*uint32)(unsafe.Pointer(&b[12:16][0])) = c.Header.Pid + b[16] = byte(c.Body.Nfgen_family) + b[17] = byte(c.Body.Version) + *(*uint16)(unsafe.Pointer(&b[18:20][0])) = c.Body.Res_id + return b +} + +func connectNetfilter(bufferSize int, groups uint32) (int, *syscall.SockaddrNetlink, error) { + s, err := syscall.Socket(syscall.AF_NETLINK, syscall.SOCK_RAW, syscall.NETLINK_NETFILTER) + if err != nil { + return 0, nil, err + } + lsa := &syscall.SockaddrNetlink{ + Family: syscall.AF_NETLINK, + Groups: groups, + } + if err := syscall.Bind(s, lsa); err != nil { + return 0, nil, err + } + if bufferSize > 0 { + // Speculatively try SO_RCVBUFFORCE which needs CAP_NET_ADMIN + if err := syscall.SetsockoptInt(s, syscall.SOL_SOCKET, syscall.SO_RCVBUFFORCE, bufferSize); err != nil { + // and if that doesn't work fall back to the ordinary SO_RCVBUF + if err := syscall.SetsockoptInt(s, syscall.SOL_SOCKET, syscall.SO_RCVBUF, bufferSize); err != nil { + return 0, nil, err + } + } + } + + return s, lsa, nil +} + +// Make syscall asking for all connections. Invoke 'cb' for each connection. +func queryAllConnections(bufferSize int, cb func(Conn)) error { + s, lsa, err := connectNetfilter(bufferSize, 0) + if err != nil { + return err + } + defer syscall.Close(s) + + msg := ConntrackListReq{ + Header: syscall.NlMsghdr{ + Len: syscall.NLMSG_HDRLEN + sizeofGenmsg, + Type: (NFNL_SUBSYS_CTNETLINK << 8) | uint16(IpctnlMsgCtGet), + Flags: syscall.NLM_F_REQUEST | syscall.NLM_F_DUMP, + Pid: 0, + Seq: 0, + }, + Body: unix.Nfgenmsg{ + Nfgen_family: syscall.AF_INET, + Version: NFNETLINK_V0, + Res_id: 0, + }, + } + wb := msg.toWireFormat() + // fmt.Printf("msg bytes: %q\n", wb) + if err := syscall.Sendto(s, wb, 0, lsa); err != nil { + return err + } + + return readMsgs(s, cb) +} + +// Stream all connections instead of query for all of them at once. +func StreamAllConnections() chan Conn { + ch := make(chan Conn, 1) + go func() { + queryAllConnections(0, func(c Conn) { + ch <- c + }) + close(ch) + }() + return ch +} + +// Lists all the connections that conntrack is tracking. +func Connections() ([]Conn, error) { + return ConnectionsSize(0) +} + +// Lists all the connections that conntrack is tracking, using specified netlink buffer size. +func ConnectionsSize(bufferSize int) ([]Conn, error) { + var conns []Conn + queryAllConnections(bufferSize, func(c Conn) { + conns = append(conns, c) + }) + return conns, nil +} + +// Established lists all established TCP connections. +func Established() ([]ConnTCP, error) { + var conns []ConnTCP + local := localIPs() + err := queryAllConnections(0, func(c Conn) { + if c.MsgType != NfctMsgUpdate { + fmt.Printf("msg isn't an update: %d\n", c.MsgType) + return + } + if c.TCPState != "ESTABLISHED" { + // fmt.Printf("state isn't ESTABLISHED: %s\n", c.TCPState) + return + } + if tc := c.ConnTCP(local); tc != nil { + conns = append(conns, *tc) + } + }) + if err != nil { + return nil, err + } + return conns, nil +} + +// Follow gives a channel with all changes. +func Follow(flags uint32) (<-chan Conn, func(), error) { + return FollowSize(0, flags) +} + +// Follow gives a channel with all changes, , using specified netlink buffer size. +func FollowSize(bufferSize int, flags uint32) (<-chan Conn, func(), error) { + var closing bool + s, _, err := connectNetfilter(bufferSize, flags) + stop := func() { + closing = true + syscall.Close(s) + } + if err != nil { + return nil, stop, err + } + + res := make(chan Conn, 1) + go func() { + defer syscall.Close(s) + if err := readMsgs(s, func(c Conn) { + // if conn.TCPState != 3 { + // // 3 is TCP established. + // continue + // } + res <- c + }); err != nil && !closing { + panic(err) + } + }() + return res, stop, nil +} + +func readMsgs(s int, cb func(Conn)) error { + rb := make([]byte, 2*syscall.Getpagesize()) +loop: + for { + nr, _, err := syscall.Recvfrom(s, rb, 0) + if err == syscall.ENOBUFS { + // ENOBUF means we miss some events here. No way around it. That's life. + cb(Conn{Err: syscall.ENOBUFS}) + continue + } else if err != nil { + return err + } + + msgs, err := syscall.ParseNetlinkMessage(rb[:nr]) + if err != nil { + return err + } + for _, msg := range msgs { + if msg.Header.Type == unix.NLMSG_ERROR { + return errors.New("NLMSG_ERROR") + } + if msg.Header.Type == unix.NLMSG_DONE { + break loop + } + + if nflnSubsysID(msg.Header.Type) != NFNL_SUBSYS_CTNETLINK { + return fmt.Errorf( + "unexpected subsys_id: %d\n", + nflnSubsysID(msg.Header.Type), + ) + } + conn, err := parsePayload(msg.Data[sizeofGenmsg:]) + if err != nil { + return err + } + + // Taken from conntrack/parse.c:__parse_message_type + switch CntlMsgTypes(nflnMsgType(msg.Header.Type)) { + case IpctnlMsgCtNew: + conn.MsgType = NfctMsgUpdate + if msg.Header.Flags&(syscall.NLM_F_CREATE|syscall.NLM_F_EXCL) > 0 { + conn.MsgType = NfctMsgNew + } + case IpctnlMsgCtDelete: + conn.MsgType = NfctMsgDestroy + } + + cb(*conn) + } + } + return nil +} + +type Tuple struct { + Proto int + Src net.IP + SrcPort uint16 + Dst net.IP + DstPort uint16 + + // ICMP stuff. + IcmpId uint16 + IcmpType uint8 + IcmpCode uint8 +} + +type Conn struct { + MsgType NfConntrackMsg + TCPState string + Status CtStatus + Orig Tuple + Reply Tuple + + // ct.mark, used to set permission type of the flow. + CtMark uint32 + + // ct.id, used to identify connections. + CtId uint32 + + // For multitenancy. + Zone uint16 + + // Flow stats. + ReplyPktLen uint64 + ReplyPktCount uint64 + OrigPktLen uint64 + OrigPktCount uint64 + + // Error, if any. + Err error +} + +// ConnTCP decides which way this connection is going and makes a ConnTCP. +func (c Conn) ConnTCP(local map[string]struct{}) *ConnTCP { + // conntrack gives us all connections, even things passing through, but it + // doesn't tell us what the local IP is. So we use `local` as a guide + // what's local. + src := c.Orig.Src.String() + dst := c.Orig.Dst.String() + _, srcLocal := local[src] + _, dstLocal := local[dst] + // If both are local we must just order things predictably. + if srcLocal && dstLocal { + srcLocal = c.Orig.SrcPort < c.Orig.DstPort + } + if srcLocal { + return &ConnTCP{ + Local: src, + LocalPort: strconv.Itoa(int(c.Orig.SrcPort)), + Remote: dst, + RemotePort: strconv.Itoa(int(c.Orig.DstPort)), + } + } + if dstLocal { + return &ConnTCP{ + Local: dst, + LocalPort: strconv.Itoa(int(c.Orig.DstPort)), + Remote: src, + RemotePort: strconv.Itoa(int(c.Orig.SrcPort)), + } + } + // Neither is local. conntrack also reports NAT connections. + return nil +} + +func parsePayload(b []byte) (*Conn, error) { + // Most of this comes from libnetfilter_conntrack/src/conntrack/parse_mnl.c + conn := &Conn{} + var attrSpace [16]Attr + attrs, err := parseAttrs(b, attrSpace[0:0]) + if err != nil { + return conn, err + } + for _, attr := range attrs { + switch CtattrType(attr.Typ) { + case CtaTupleOrig: + parseTuple(attr.Msg, &conn.Orig) + case CtaTupleReply: + parseTuple(attr.Msg, &conn.Reply) + case CtaCountersOrig: + conn.OrigPktLen, conn.OrigPktCount, _ = parseCounters(attr.Msg) + case CtaCountersReply: + conn.ReplyPktLen, conn.ReplyPktCount, _ = parseCounters(attr.Msg) + case CtaStatus: + conn.Status = CtStatus(binary.BigEndian.Uint32(attr.Msg)) + case CtaProtoinfo: + parseProtoinfo(attr.Msg, conn) + case CtaMark: + conn.CtMark = binary.BigEndian.Uint32(attr.Msg) + case CtaZone: + conn.Zone = binary.BigEndian.Uint16(attr.Msg) + case CtaId: + conn.CtId = binary.BigEndian.Uint32(attr.Msg) + } + } + return conn, nil +} + +func parseTuple(b []byte, tuple *Tuple) error { + var attrSpace [16]Attr + attrs, err := parseAttrs(b, attrSpace[0:0]) + if err != nil { + return fmt.Errorf("invalid tuple attr: %s", err) + } + for _, attr := range attrs { + // fmt.Printf("pl: %d, type: %d, multi: %t, bigend: %t\n", len(attr.Msg), attr.Typ, attr.IsNested, attr.IsNetByteorder) + switch CtattrTuple(attr.Typ) { + case CtaTupleUnspec: + // fmt.Printf("It's a tuple unspec\n") + case CtaTupleIp: + // fmt.Printf("It's a tuple IP\n") + if err := parseIP(attr.Msg, tuple); err != nil { + return err + } + case CtaTupleProto: + // fmt.Printf("It's a tuple proto\n") + parseProto(attr.Msg, tuple) + } + } + return nil +} + +func parseCounters(b []byte) (uint64, uint64, error) { + var attrSpace [16]Attr + attrs, err := parseAttrs(b, attrSpace[0:0]) + if err != nil { + return 0, 0, fmt.Errorf("invalid tuple attr: %s", err) + } + packets := uint64(0) + bytes := uint64(0) + for _, attr := range attrs { + switch CtattrCounters(attr.Typ) { + case CtaCountersPackets: + packets = binary.BigEndian.Uint64(attr.Msg) + case CtaCountersBytes: + bytes = binary.BigEndian.Uint64(attr.Msg) + } + } + return packets, bytes, nil +} + +func parseIP(b []byte, tuple *Tuple) error { + var attrSpace [16]Attr + attrs, err := parseAttrs(b, attrSpace[0:0]) + if err != nil { + return fmt.Errorf("invalid tuple attr: %s", err) + } + for _, attr := range attrs { + switch CtattrIp(attr.Typ) { + case CtaIpV4Src: + tuple.Src = make(net.IP, len(attr.Msg)) + copy(tuple.Src, attr.Msg) + case CtaIpV4Dst: + tuple.Dst = make(net.IP, len(attr.Msg)) + copy(tuple.Dst, attr.Msg) + case CtaIpV6Src: + // TODO + case CtaIpV6Dst: + // TODO + } + } + return nil +} + +func parseProto(b []byte, tuple *Tuple) error { + var attrSpace [16]Attr + attrs, err := parseAttrs(b, attrSpace[0:0]) + if err != nil { + return fmt.Errorf("invalid tuple attr: %s", err) + } + + for _, attr := range attrs { + switch CtattrL4proto(attr.Typ) { + // Protocol number. + case CtaProtoNum: + tuple.Proto = int(uint8(attr.Msg[0])) + + // TCP stuff. + case CtaProtoSrcPort: + tuple.SrcPort = binary.BigEndian.Uint16(attr.Msg) + case CtaProtoDstPort: + tuple.DstPort = binary.BigEndian.Uint16(attr.Msg) + + // ICMP stuff. + case CtaProtoIcmpId: + tuple.IcmpId = binary.BigEndian.Uint16(attr.Msg) + case CtaProtoIcmpType: + tuple.IcmpType = attr.Msg[0] + case CtaProtoIcmpCode: + tuple.IcmpCode = attr.Msg[0] + } + } + + return nil +} + +func parseProtoinfo(b []byte, conn *Conn) error { + var attrSpace [16]Attr + attrs, err := parseAttrs(b, attrSpace[0:0]) + if err != nil { + return fmt.Errorf("invalid tuple attr: %s", err) + } + for _, attr := range attrs { + switch CtattrProtoinfo(attr.Typ) { + case CtaProtoinfoTcp: + if err := parseProtoinfoTCP(attr.Msg, conn); err != nil { + return err + } + default: + // we're not interested in other protocols + } + } + return nil +} + +func parseProtoinfoTCP(b []byte, conn *Conn) error { + var attrSpace [16]Attr + attrs, err := parseAttrs(b, attrSpace[0:0]) + if err != nil { + return fmt.Errorf("invalid tuple attr: %s", err) + } + for _, attr := range attrs { + switch CtattrProtoinfoTcp(attr.Typ) { + case CtaProtoinfoTcpState: + conn.TCPState = tcpState[uint8(attr.Msg[0])] + default: + // not interested + } + } + return nil +} diff --git a/vendor/github.com/typetypetype/conntrack/conntrack.go b/vendor/github.com/typetypetype/conntrack/conntrack.go new file mode 100644 index 000000000..64cca3452 --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/conntrack.go @@ -0,0 +1,145 @@ +package conntrack + +import ( + "fmt" + "log" + "time" +) + +// ConnTCP is a connection +type ConnTCP struct { + Local string // net.IP + LocalPort string // int + Remote string // net.IP + RemotePort string // int +} + +func (c ConnTCP) String() string { + return fmt.Sprintf("%s:%s->%s:%s", c.Local, c.LocalPort, c.Remote, c.RemotePort) +} + +// ConnTrack monitors the connections. It is build with Established() and +// Follow(). +type ConnTrack struct { + connReq chan chan []ConnTCP + quit chan struct{} +} + +// New returns a ConnTrack. +func New() (*ConnTrack, error) { + c := ConnTrack{ + connReq: make(chan chan []ConnTCP), + quit: make(chan struct{}), + } + go func() { + for { + err := c.track() + select { + case <-c.quit: + return + default: + } + if err != nil { + log.Printf("conntrack: %s\n", err) + } + time.Sleep(1 * time.Second) + } + }() + + return &c, nil +} + +// Close stops all monitoring and executables. +func (c ConnTrack) Close() { + close(c.quit) +} + +// Connections returns the list of all connections seen since last time you +// called it. +func (c *ConnTrack) Connections() []ConnTCP { + r := make(chan []ConnTCP) + c.connReq <- r + return <-r +} + +// track is the main loop +func (c *ConnTrack) track() error { + // We use Follow() to keep track of conn state changes, but it doesn't give + // us the initial state. If we first look at the established connections + // and then start the follow process we might miss events. + var flags uint32 = NF_NETLINK_CONNTRACK_NEW | NF_NETLINK_CONNTRACK_UPDATE | + NF_NETLINK_CONNTRACK_DESTROY + events, stop, err := Follow(flags) + if err != nil { + return err + } + + established := map[ConnTCP]struct{}{} + cs, err := Established() + if err != nil { + return err + } + for _, c := range cs { + established[c] = struct{}{} + } + // we keep track of deleted so we can report them + deleted := map[ConnTCP]struct{}{} + + local := localIPs() + updateLocalIPs := time.Tick(time.Minute) + + for { + select { + + case <-c.quit: + stop() + return nil + + case <-updateLocalIPs: + local = localIPs() + + case e, ok := <-events: + if !ok { + return nil + } + switch { + + default: + // not interested + + case e.TCPState == "ESTABLISHED": + cn := e.ConnTCP(local) + if cn == nil { + // log.Printf("not a local connection: %+v\n", e) + continue + } + established[*cn] = struct{}{} + + case e.MsgType == NfctMsgDestroy, e.TCPState == "TIME_WAIT", e.TCPState == "CLOSE": + cn := e.ConnTCP(local) + if cn == nil { + // log.Printf("not a local connection: %+v\n", e) + continue + } + if _, ok := established[*cn]; !ok { + continue + } + delete(established, *cn) + deleted[*cn] = struct{}{} + + } + + case r := <-c.connReq: + cs := make([]ConnTCP, 0, len(established)+len(deleted)) + for c := range established { + cs = append(cs, c) + } + for c := range deleted { + cs = append(cs, c) + } + r <- cs + deleted = map[ConnTCP]struct{}{} + + } + } +} diff --git a/vendor/github.com/typetypetype/conntrack/const.go b/vendor/github.com/typetypetype/conntrack/const.go new file mode 100644 index 000000000..4e3a9eba8 --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/const.go @@ -0,0 +1,69 @@ +package conntrack + +const ( + // #defined in libnfnetlink/include/libnfnetlink/linux_nfnetlink.h + NFNL_SUBSYS_CTNETLINK = 1 + NFNETLINK_V0 = 0 + + // #defined in libnfnetlink/include/libnfnetlink/linux_nfnetlink_compat.h + NF_NETLINK_CONNTRACK_NEW = 0x00000001 + NF_NETLINK_CONNTRACK_UPDATE = 0x00000002 + NF_NETLINK_CONNTRACK_DESTROY = 0x00000004 + NF_NETLINK_CONNTRACK_EXP_NEW = 0x00000008 + NF_NETLINK_CONNTRACK_EXP_UPDATE = 0x00000010 + NF_NETLINK_CONNTRACK_EXP_DESTROY = 0x00000020 + + // #defined in libnfnetlink/include/libnfnetlink/libnfnetlink.h + NLA_F_NESTED = uint16(1 << 15) + NLA_F_NET_BYTEORDER = uint16(1 << 14) + NLA_TYPE_MASK = ^(NLA_F_NESTED | NLA_F_NET_BYTEORDER) +) + +type NfConntrackMsg int + +const ( + NfctMsgUnknown NfConntrackMsg = 0 + NfctMsgNew NfConntrackMsg = 1 << 0 + NfctMsgUpdate NfConntrackMsg = 1 << 1 + NfctMsgDestroy NfConntrackMsg = 1 << 2 +) + +// taken from libnetfilter_conntrack/src/conntrack/snprintf.c +var tcpState = []string{ + "NONE", + "SYN_SENT", + "SYN_RECV", + "ESTABLISHED", + "FIN_WAIT", + "CLOSE_WAIT", + "LAST_ACK", + "TIME_WAIT", + "CLOSE", + "LISTEN", + "MAX", + "IGNORE", +} + +// Taken from include/uapi/linux/netfilter/nf_conntrack_common.h +type CtStatus uint32 + +const ( + IPS_EXPECTED CtStatus = 1 << iota + IPS_SEEN_REPLY + IPS_ASSURED + IPS_CONFIRMED + IPS_SRC_NAT + IPS_DST_NAT + IPS_SEQ_ADJUST + IPS_SRC_NAT_DONE + IPS_DST_NAT_DONE + IPS_DYING + IPS_FIXED_TIMEOUT + IPS_TEMPLATE + IPS_UNTRACKED + IPS_HELPER + IPS_OFFLOAD + + IPS_NAT_MASK = (IPS_DST_NAT | IPS_SRC_NAT) + IPS_NAT_DONE_MASK = (IPS_DST_NAT_DONE | IPS_SRC_NAT_DONE) +) diff --git a/vendor/github.com/typetypetype/conntrack/headers.go b/vendor/github.com/typetypetype/conntrack/headers.go new file mode 100644 index 000000000..55531855e --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/headers.go @@ -0,0 +1,131 @@ +package conntrack + +// code generated by enum2go. + +type CtattrType int +const ( + CtaUnspec CtattrType = 0 + CtaTupleOrig CtattrType = 1 + CtaTupleReply CtattrType = 2 + CtaStatus CtattrType = 3 + CtaProtoinfo CtattrType = 4 + CtaHelp CtattrType = 5 + CtaNatSrc CtattrType = 6 + CtaTimeout CtattrType = 7 + CtaMark CtattrType = 8 + CtaCountersOrig CtattrType = 9 + CtaCountersReply CtattrType = 10 + CtaUse CtattrType = 11 + CtaId CtattrType = 12 + CtaNatDst CtattrType = 13 + CtaTupleMaster CtattrType = 14 + CtaNatSeqAdjOrig CtattrType = 15 + CtaNatSeqAdjReply CtattrType = 16 + CtaSecmark CtattrType = 17 + CtaZone CtattrType = 18 + CtaSecctx CtattrType = 19 + CtaTimestamp CtattrType = 20 + CtaMarkMask CtattrType = 21 + CtaLabels CtattrType = 22 + CtaLabelsMask CtattrType = 23 + CtaMax CtattrType = 24 +) + +type CtattrTuple int +const ( + CtaTupleUnspec CtattrTuple = 0 + CtaTupleIp CtattrTuple = 1 + CtaTupleProto CtattrTuple = 2 + CtaTupleMax CtattrTuple = 3 +) + +type CtattrCounters int +const ( + CtaCountersUnspec CtattrCounters = 0 + CtaCountersPackets CtattrCounters = 1 /* 64bit counters */ + CtaCountersBytes CtattrCounters = 2 /* 64bit counters */ + CtaCounters32Packets CtattrCounters = 3 /* old 32bit counters, unused */ + CtaCounters32Bytes CtattrCounters = 4 /* old 32bit counters, unused */ + CtaCountersMax CtattrCounters = 5 +) + +type CtattrIp int +const ( + CtaIpUnspec CtattrIp = 0 + CtaIpV4Src CtattrIp = 1 + CtaIpV4Dst CtattrIp = 2 + CtaIpV6Src CtattrIp = 3 + CtaIpV6Dst CtattrIp = 4 + CtaIpMax CtattrIp = 5 +) + +type CtattrL4proto int +const ( + CtaProtoUnspec CtattrL4proto = 0 + CtaProtoNum CtattrL4proto = 1 + CtaProtoSrcPort CtattrL4proto = 2 + CtaProtoDstPort CtattrL4proto = 3 + CtaProtoIcmpId CtattrL4proto = 4 + CtaProtoIcmpType CtattrL4proto = 5 + CtaProtoIcmpCode CtattrL4proto = 6 + CtaProtoIcmpv6Id CtattrL4proto = 7 + CtaProtoIcmpv6Type CtattrL4proto = 8 + CtaProtoIcmpv6Code CtattrL4proto = 9 + CtaProtoMax CtattrL4proto = 10 +) + +type CtattrProtoinfo int +const ( + CtaProtoinfoUnspec CtattrProtoinfo = 0 + CtaProtoinfoTcp CtattrProtoinfo = 1 + CtaProtoinfoDccp CtattrProtoinfo = 2 + CtaProtoinfoSctp CtattrProtoinfo = 3 + CtaProtoinfoMax CtattrProtoinfo = 4 +) + +type CtattrProtoinfoTcp int +const ( + CtaProtoinfoTcpUnspec CtattrProtoinfoTcp = 0 + CtaProtoinfoTcpState CtattrProtoinfoTcp = 1 + CtaProtoinfoTcpWscaleOriginal CtattrProtoinfoTcp = 2 + CtaProtoinfoTcpWscaleReply CtattrProtoinfoTcp = 3 + CtaProtoinfoTcpFlagsOriginal CtattrProtoinfoTcp = 4 + CtaProtoinfoTcpFlagsReply CtattrProtoinfoTcp = 5 + CtaProtoinfoTcpMax CtattrProtoinfoTcp = 6 +) + +type NfConntrackAttrGrp int +const ( + AttrGrpOrigIpv4 NfConntrackAttrGrp = 0 + AttrGrpReplIpv4 NfConntrackAttrGrp = 1 + AttrGrpOrigIpv6 NfConntrackAttrGrp = 2 + AttrGrpReplIpv6 NfConntrackAttrGrp = 3 + AttrGrpOrigPort NfConntrackAttrGrp = 4 + AttrGrpReplPort NfConntrackAttrGrp = 5 + AttrGrpIcmp NfConntrackAttrGrp = 6 + AttrGrpMasterIpv4 NfConntrackAttrGrp = 7 + AttrGrpMasterIpv6 NfConntrackAttrGrp = 8 + AttrGrpMasterPort NfConntrackAttrGrp = 9 + AttrGrpOrigCounters NfConntrackAttrGrp = 10 + AttrGrpReplCounters NfConntrackAttrGrp = 11 + AttrGrpOrigAddrSrc NfConntrackAttrGrp = 12 + AttrGrpOrigAddrDst NfConntrackAttrGrp = 13 + AttrGrpReplAddrSrc NfConntrackAttrGrp = 14 + AttrGrpReplAddrDst NfConntrackAttrGrp = 15 + AttrGrpMax NfConntrackAttrGrp = 16 +) + +type NfConntrackQuery int +const ( + NfctQCreate NfConntrackQuery = 0 + NfctQUpdate NfConntrackQuery = 1 + NfctQDestroy NfConntrackQuery = 2 + NfctQGet NfConntrackQuery = 3 + NfctQFlush NfConntrackQuery = 4 + NfctQDump NfConntrackQuery = 5 + NfctQDumpReset NfConntrackQuery = 6 + NfctQCreateUpdate NfConntrackQuery = 7 + NfctQDumpFilter NfConntrackQuery = 8 + NfctQDumpFilterReset NfConntrackQuery = 9 +) + diff --git a/vendor/github.com/typetypetype/conntrack/linux_headers.go b/vendor/github.com/typetypetype/conntrack/linux_headers.go new file mode 100644 index 000000000..250c856b6 --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/linux_headers.go @@ -0,0 +1,16 @@ +package conntrack + +// code generated by enum2go, from file "linux_nfnetlink_conntrack.h". + +type CntlMsgTypes int +const ( + IpctnlMsgCtNew CntlMsgTypes = 0 + IpctnlMsgCtGet CntlMsgTypes = 1 + IpctnlMsgCtDelete CntlMsgTypes = 2 + IpctnlMsgCtGetCtrzero CntlMsgTypes = 3 + IpctnlMsgCtGetStatsCpu CntlMsgTypes = 4 + IpctnlMsgCtGetStats CntlMsgTypes = 5 + IpctnlMsgCtGetDying CntlMsgTypes = 6 + IpctnlMsgCtGetUnconfirmed CntlMsgTypes = 7 + IpctnlMsgMax CntlMsgTypes = 8 +) diff --git a/vendor/github.com/typetypetype/conntrack/net.go b/vendor/github.com/typetypetype/conntrack/net.go new file mode 100644 index 000000000..0314379f8 --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/net.go @@ -0,0 +1,19 @@ +package conntrack + +import ( + "net" +) + +// localIPs gives all IPs we consider local. +func localIPs() map[string]struct{} { + var l = map[string]struct{}{} + if localNets, err := net.InterfaceAddrs(); err == nil { + // Not all networks are IP networks. + for _, localNet := range localNets { + if net, ok := localNet.(*net.IPNet); ok { + l[net.IP.String()] = struct{}{} + } + } + } + return l +} diff --git a/vendor/github.com/typetypetype/conntrack/netlink.go b/vendor/github.com/typetypetype/conntrack/netlink.go new file mode 100644 index 000000000..0d7e1e156 --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/netlink.go @@ -0,0 +1,21 @@ +package conntrack + +import ( + "syscall" +) + +// NFNL_MSG_TYPE +func nflnMsgType(x uint16) uint8 { + return uint8(x & 0x00ff) +} + +// NFNL_SUBSYS_ID +func nflnSubsysID(x uint16) uint8 { + return uint8((x & 0xff00) >> 8) +} + +// Round the length of a netlink route attribute up to align it +// properly. +func rtaAlignOf(attrlen int) int { + return (attrlen + syscall.RTA_ALIGNTO - 1) & ^(syscall.RTA_ALIGNTO - 1) +} diff --git a/vendor/github.com/typetypetype/conntrack/netlink_attr.go b/vendor/github.com/typetypetype/conntrack/netlink_attr.go new file mode 100644 index 000000000..2c6f947ab --- /dev/null +++ b/vendor/github.com/typetypetype/conntrack/netlink_attr.go @@ -0,0 +1,44 @@ +package conntrack + +// Netlink attr parsing. + +import ( + "encoding/binary" + "errors" +) + +const attrHdrLength = 4 + +type Attr struct { + Msg []byte + Typ int + IsNested bool + IsNetByteorder bool +} + +func parseAttrs(b []byte, attrs []Attr) ([]Attr, error) { + for len(b) >= attrHdrLength { + var attr Attr + attr, b = parseAttr(b) + attrs = append(attrs, attr) + } + if len(b) != 0 { + return nil, errors.New("leftover attr bytes") + } + return attrs, nil +} + +func parseAttr(b []byte) (Attr, []byte) { + l := binary.LittleEndian.Uint16(b[0:2]) + // length is header + payload + l -= uint16(attrHdrLength) + + typ := binary.LittleEndian.Uint16(b[2:4]) + attr := Attr{ + Msg: b[attrHdrLength : attrHdrLength+int(l)], + Typ: int(typ & NLA_TYPE_MASK), + IsNested: typ&NLA_F_NESTED > 0, + IsNetByteorder: typ&NLA_F_NET_BYTEORDER > 0, + } + return attr, b[rtaAlignOf(attrHdrLength+int(l)):] +} diff --git a/vendor/manifest b/vendor/manifest index 8f535fc0f..59af0958a 100644 --- a/vendor/manifest +++ b/vendor/manifest @@ -1903,6 +1903,14 @@ "branch": "master", "notests": true }, + { + "importpath": "github.com/typetypetype/conntrack", + "repository": "https://github.com/typetypetype/conntrack", + "vcs": "git", + "revision": "9d9dd841d4eb5f1e151e7af9926a2ddb416c6a79", + "branch": "master", + "notests": true + }, { "importpath": "github.com/uber-go/tally", "repository": "https://github.com/uber-go/tally",