Apply anonymization to traces before writing in triggertrace (#154)

* Add anonymization support to scamper parsers
* Anonymize traces before writing
* Update m-lab/go version
* Rename Tracetool field
* Add test case for WriteFile failure
* Correct goveralls path

Author: stephen-soltesz

.travis.yml

@@ -3,7 +3,7 @@ go:
 - 1.18
 
 before_install:
-- go get github.com/mattn/goveralls
+- go install github.com/mattn/goveralls@latest
 # Install dependencies, including test dependencies.
 - sudo apt-get update && sudo apt-get install -y scamper
 - go get -v -t ./...
@@ -18,4 +18,4 @@ script:
 after_success:
 # Coveralls
 # Upload coverage information for unit tests.
-- $HOME/gopath/bin/goveralls -coverprofile=_coverage.cov -service=travis-ci
+- $GOPATH/bin/goveralls -coverprofile=_coverage.cov -service=travis-ci

cmd/trex/main.go

@@ -156,22 +156,19 @@ func parseFile(fileName string) *parser.Scamper1 {
 		nParseErrors++
 		return nil
 	}
-	var scamper1 parser.Scamper1
-	switch p := parsedData.(type) {
-	case parser.Scamper1:
-		scamper1 = p
-	default:
+	scamper1, ok := parsedData.(*parser.Scamper1)
+	if !ok {
 		// This is an internal error because we instantiated a new MDA
 		// parser which should return Scamper1 as the concrete type.
-		fmt.Fprintf(os.Stderr, "%T: unknown datatype (expected scamper1)", p)
+		fmt.Fprintf(os.Stderr, "%T: unknown datatype (expected scamper1)", parsedData)
 		os.Exit(1)
 	}
 	nFilesParsed++
 	if len(scamper1.Tracelb.Nodes) == 0 {
 		nNoTraceroute++
 		return nil
 	}
-	return &scamper1
+	return scamper1
 }
 
 // tracerouteDuration returns the duration of the specified traceroute

go.mod

@@ -3,7 +3,7 @@ module github.com/m-lab/traceroute-caller
 go 1.18
 
 require (
-	github.com/m-lab/go v0.1.47
+	github.com/m-lab/go v0.1.55
 	github.com/m-lab/tcp-info v1.5.3
 	github.com/m-lab/uuid-annotator v0.4.7
 	github.com/prometheus/client_golang v1.12.2

go.sum

@@ -242,6 +242,8 @@ github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/m-lab/annotation-service v0.0.0-20210504151333-138bdf572368 h1:cRzgLEJxoMI0iSexWOxTZQR/gNG08ra1IoEEgwJBdgs=
 github.com/m-lab/go v0.1.47 h1:yV6RgVpiWm2BnpJcjfy4pbUkB9cz0BvBbVG64UGLiC0=
 github.com/m-lab/go v0.1.47/go.mod h1:woT26L9Hf07juZGHe7Z4WveV7MM6NS6vQaaWzRQnab4=
+github.com/m-lab/go v0.1.55 h1:cuapp1ZTaBL6QDylni+k/GVnCqf9OUdzbJM+qnkh96Q=
+github.com/m-lab/go v0.1.55/go.mod h1:O1D/EoVarJ8lZt9foANcqcKtwxHatBzUxXFFyC87aQQ=
 github.com/m-lab/tcp-info v1.5.3 h1:4IspTPcNc8D8LNRvuFnID8gDiz+hxPAtYvpKZaiGGe8=
 github.com/m-lab/tcp-info v1.5.3/go.mod h1:bkvI4qbjB6QVC2tsLSHqf5OnIYcmuLEVjo7+8YA56Kg=
 github.com/m-lab/uuid-annotator v0.4.1/go.mod h1:f/zvgcc5A3HQ1Y63HWpbBVXNcsJwQ4uRIOqsF/nyto8=

internal/triggertrace/triggertrace.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/m-lab/go/anonymize"
 	"github.com/m-lab/tcp-info/inetdiag"
 	"github.com/m-lab/traceroute-caller/hopannotation"
 	"github.com/m-lab/traceroute-caller/internal/ipcache"
@@ -50,6 +51,7 @@ type AnnotateAndArchiver interface {
 	WriteAnnotations(map[string]*annotator.ClientAnnotations, time.Time) []error
 }
 
+// TracerWriter provides the interface for issuing traces and writing results.
 type TracerWriter interface {
 	ipcache.Tracer
 	WriteFile(uuid string, t time.Time, b []byte) error
@@ -63,7 +65,8 @@ type Handler struct {
 	IPCache          FetchTracer
 	Parser           ParseTracer
 	HopAnnotator     AnnotateAndArchiver
-	Tracer           TracerWriter
+	Tracetool        TracerWriter
+	Anonymizer       anonymize.IPAnonymizer
 	done             chan struct{} // For testing.
 }
 
@@ -87,7 +90,8 @@ func NewHandler(ctx context.Context, tracetool TracerWriter, ipcCfg ipcache.Conf
 		IPCache:      ipCache,
 		Parser:       newParser,
 		HopAnnotator: hopCache,
-		Tracer:       tracetool,
+		Tracetool:    tracetool,
+		Anonymizer:   anonymize.New(anonymize.IPAnonymizationFlag),
 	}, nil
 }
 
@@ -152,8 +156,14 @@ func (h *Handler) traceAnnotateAndArchive(ctx context.Context, uuid string, dest
 		log.Printf("context %p: failed to parse traceroute output (error: %v)\n", ctx, err)
 		return
 	}
+
+	// Anonymize the parsed data in place.
+	parsedData.Anonymize(h.Anonymizer)
+	// Remarshal anonymized data for writing.
+	rawData = parsedData.MarshalJSONL()
+
 	traceStartTime := parsedData.StartTime()
-	err = h.Tracer.WriteFile(uuid, traceStartTime, rawData)
+	err = h.Tracetool.WriteFile(uuid, traceStartTime, rawData)
 	if err != nil {
 		log.Printf("context %p: failed to write trace file for uuid: %s: (error: %v)\n", ctx, uuid, err)
 	}

internal/triggertrace/triggertrace_test.go

@@ -7,6 +7,7 @@ import (
 	"log"
 	"net"
 	"os"
+	"path"
 	"sync/atomic"
 	"testing"
 	"time"
@@ -23,6 +24,7 @@ var (
 	forceParseErr      = "88.88.88.88" // force a failure parsing a traceroute output
 	forceExtractErr    = "77.77.77.77" // force a failure extracting hops
 	forceAnnotateErr   = "66.66.66.66" // force a failure annotating hops
+	forceWriteErr      = "write-err-8" // force a failure writing trace
 )
 
 func init() {
@@ -57,7 +59,12 @@ func (ft *fakeTracer) Trace(remoteIP, uuid string, t time.Time) ([]byte, error)
 }
 
 func (ft *fakeTracer) WriteFile(uuid string, t time.Time, data []byte) error {
-	return nil
+	switch uuid {
+	case forceWriteErr:
+		return errors.New("forced write error")
+	default:
+		return nil
+	}
 }
 
 func (ft *fakeTracer) CachedTrace(uuid string, t time.Time, cachedTest []byte) ([]byte, error) {
@@ -99,12 +106,12 @@ func TestNewHandler(t *testing.T) {
 	defer func() { netInterfaceAddrs = saveNetInterfaceAddrs }()
 
 	netInterfaceAddrs = fakeInterfaceAddrsBad
-	if _, err := newHandler(&fakeTracer{}); err == nil {
+	if _, err := newHandler(t, &fakeTracer{}); err == nil {
 		t.Fatalf("NewHandler() = nil, want error")
 	}
 
 	netInterfaceAddrs = fakeInterfaceAddrs
-	if _, err := newHandler(&fakeTracer{}); err != nil {
+	if _, err := newHandler(t, &fakeTracer{}); err != nil {
 		t.Fatalf("NewHandler() = %v, want nil", err)
 	}
 }
@@ -114,7 +121,7 @@ func TestOpen(t *testing.T) {
 	netInterfaceAddrs = fakeInterfaceAddrs
 	defer func() { netInterfaceAddrs = saveNetInterfaceAddrs }()
 
-	handler, err := newHandler(&fakeTracer{})
+	handler, err := newHandler(t, &fakeTracer{})
 	if err != nil {
 		t.Fatalf("NewHandler() = %v, want nil", err)
 	}
@@ -164,13 +171,14 @@ func TestClose(t *testing.T) {
 		{"bad6", "127.0.0.1", forceAnnotateErr, "00005", true, true, 1, 0},
 		{"good1", "127.0.0.1", "3.4.5.6", "00006", true, true, 1, 0},
 		{"good2", "4.5.6.7", "127.0.0.1", "00007", true, true, 1, 1},
+		{"bad7", "127.0.0.1", "192.168.33.2", forceWriteErr, true, true, 1, 0},
 	}
 	for _, test := range tests {
 		test := test
 		t.Run(test.name, func(t *testing.T) {
 			t.Parallel()
 			tracer := &fakeTracer{}
-			handler, err := newHandler(tracer)
+			handler, err := newHandler(t, tracer)
 			if err != nil {
 				t.Fatalf("NewHandler() = %v, want nil", err)
 			}
@@ -206,15 +214,15 @@ func TestClose(t *testing.T) {
 	}
 }
 
-func newHandler(tracer *fakeTracer) (*Handler, error) {
+func newHandler(t *testing.T, tracer *fakeTracer) (*Handler, error) {
 	ipcCfg := ipcache.Config{
 		EntryTimeout: 2 * time.Second,
 		ScanPeriod:   1 * time.Second,
 	}
 	annotator := &fakeAnnotator{}
 	haCfg := hopannotation.Config{
 		AnnotatorClient: annotator,
-		OutputPath:      "/tmp/annotation1",
+		OutputPath:      path.Join(t.TempDir(), "annotation1"),
 	}
 	newParser, err := parser.New("mda")
 	if err != nil {

parser/parser.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"fmt"
 	"time"
+
+	"github.com/m-lab/go/anonymize"
 )
 
 // Errors returned by parser.
@@ -51,6 +53,8 @@ type CyclestopLine struct {
 type ParsedData interface {
 	StartTime() time.Time
 	ExtractHops() []string
+	MarshalJSONL() []byte
+	Anonymize(anon anonymize.IPAnonymizer)
 }
 
 // TracerouteParser defines the interface for raw traceroute data.

parser/scamper1.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"time"
 
+	"github.com/m-lab/go/anonymize"
 	"github.com/m-lab/traceroute-caller/tracer"
 )
 
@@ -57,10 +58,12 @@ type ScamperNode struct {
 }
 
 // Scamper1 encapsulates the four lines of a traceroute:
-//   {"UUID":...}
-//   {"type":"cycle-start"...}
-//   {"type":"tracelb"...}
-//   {"type":"cycle-stop"...}
+//
+//	{"UUID":...}
+//	{"type":"cycle-start"...}
+//	{"type":"tracelb"...}
+//	{"type":"cycle-stop"...}
+//
 // Refer to scamper source code files scamper/scamper_list.h and
 // scamper/tracelb/scamper_tracelb.h for the definitions of cycle_start,
 // tracelb, and cycle_stop lines.
@@ -143,7 +146,7 @@ func (s1 *scamper1Parser) ParseRawData(rawData []byte) (ParsedData, error) {
 		return nil, fmt.Errorf("%w: %v", ErrCycleStopType, scamper1.CycleStop.Type)
 	}
 
-	return scamper1, nil
+	return &scamper1, nil
 }
 
 // StartTime returns the start time of the traceroute.
@@ -177,3 +180,42 @@ func (s1 Scamper1) ExtractHops() []string {
 	}
 	return hopStrings
 }
+
+// Anonymize looks for hops that are in the client subnet, and anonymizes them using the given anonymizer.
+func (s1 *Scamper1) Anonymize(anon anonymize.IPAnonymizer) {
+	tracelb := s1.Tracelb
+	dst := net.ParseIP(tracelb.Dst)
+	anon.IP(dst)
+	s1.Tracelb.Dst = dst.String()
+
+	for i := range tracelb.Nodes {
+		node := &tracelb.Nodes[i]
+		ip := net.ParseIP(node.Addr)
+		if anon.Contains(dst, ip) {
+			anon.IP(ip)
+			node.Addr = ip.String()
+		}
+		for j := range node.Links {
+			links := node.Links[j]
+			for k := range links {
+				link := &links[k]
+				ip = net.ParseIP(link.Addr)
+				if anon.Contains(dst, ip) {
+					anon.IP(ip)
+					link.Addr = ip.String()
+				}
+			}
+		}
+	}
+}
+
+// MarshalJSONL encodes the scamper object as JSONL.
+func (s1 Scamper1) MarshalJSONL() []byte {
+	buff := &bytes.Buffer{}
+	enc := json.NewEncoder(buff)
+	enc.Encode(s1.Metadata)
+	enc.Encode(s1.CycleStart)
+	enc.Encode(s1.Tracelb)
+	enc.Encode(s1.CycleStop)
+	return buff.Bytes()
+}