Merge pull request #467 from m-lab/ipv6

Add IPv6 triple colon fix

Author: gfr10598

parser/ndt_meta.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/m-lab/etl/metrics"
 	"github.com/m-lab/etl/schema"
+	"github.com/m-lab/etl/web100"
 )
 
 // MetaFileData is the parsed info from the .meta file.
@@ -55,13 +56,13 @@ var fieldPairs = map[string]string{
 
 func handleIP(connSpec schema.Web100ValueMap, prefix string, ipString string) {
 	connSpec.SetString(prefix+"_ip", ipString)
-	if ValidateIP(ipString) != nil {
+	if web100.ValidateIP(ipString) != nil {
 		log.Printf("Failed parsing connSpec IP: %s\n", ipString)
 		metrics.WarningCount.WithLabelValues(
 			"ndt", "unknown", "failed parsing connSpec IP").Inc()
 	} else {
 		connSpec.SetString(prefix+"_ip", ipString)
-		family := ParseIPFamily(ipString)
+		family := web100.ParseIPFamily(ipString)
 		if family != -1 {
 			connSpec.SetInt64(prefix+"_af", family)
 		}

parser/ss.go

@@ -4,7 +4,6 @@ package parser
 import (
 	"bytes"
 	"errors"
-	"fmt"
 	"log"
 	"path/filepath"
 	"reflect"
@@ -106,7 +105,7 @@ func PackDataIntoSchema(ss_value map[string]string, log_time time.Time, testName
 
 	conn_spec := &schema.Web100ConnectionSpecification{
 		Local_ip:    ss_value["LocalAddress"],
-		Local_af:    ParseIPFamily(ss_value["LocalAddress"]),
+		Local_af:    web100.ParseIPFamily(ss_value["LocalAddress"]),
 		Local_port:  int64(local_port),
 		Remote_ip:   ss_value["RemAddress"],
 		Remote_port: int64(remote_port),
@@ -206,16 +205,15 @@ func (ss *SSParser) ParseAndInsert(meta map[string]bigquery.Value, testName stri
 	metrics.WorkerState.WithLabelValues(ss.TableName(), "ss").Inc()
 	defer metrics.WorkerState.WithLabelValues(ss.TableName(), "ss").Dec()
 
-	log_time, err := ExtractLogtimeFromFilename(testName)
+	logTime, err := ExtractLogtimeFromFilename(testName)
 	if err != nil {
 		return err
 	}
-	var var_names []string
 	testContent := strings.Split(string(rawContent[:]), "\n")
 	if len(testContent) < 2 {
-		return errors.New("empty test file.")
+		return errors.New("empty test file")
 	}
-	var_names, err = ParseKHeader(testContent[0])
+	varNames, err := ParseKHeader(testContent[0])
 	if err != nil {
 		metrics.ErrorCount.WithLabelValues(
 			ss.TableName(), "ss", "corrupted header").Inc()
@@ -227,34 +225,34 @@ func (ss *SSParser) ParseAndInsert(meta map[string]bigquery.Value, testName stri
 		if len(oneLine) == 0 {
 			continue
 		}
-		ss_value, err := ParseOneLine(oneLine, var_names)
+		ssValue, err := ParseOneLine(oneLine, varNames)
 		if err != nil {
 			metrics.TestCount.WithLabelValues(
 				ss.TableName(), "ss", "corrupted content").Inc()
-			return err
+			continue
 		}
-		err = ValidateIP(ss_value["LocalAddress"])
+		err = web100.ValidateIP(ssValue["LocalAddress"])
 		if err != nil {
 			metrics.TestCount.WithLabelValues(
 				ss.TableName(), "ss", "Invalid server IP").Inc()
-			return fmt.Errorf("Invalid server IP address: %s with error: %s", ss_value["LocalAddress"], err)
+			log.Printf("Invalid server IP address: %s with error: %s\n", ssValue["LocalAddress"], err)
+			continue
 		}
-		err = ValidateIP(ss_value["RemAddress"])
+		err = web100.ValidateIP(ssValue["RemAddress"])
 		if err != nil {
 			metrics.TestCount.WithLabelValues(
 				ss.TableName(), "ss", "Invalid client IP").Inc()
-			return fmt.Errorf("Invalid client IP address: %s with error: %s", ss_value["RemAddress"], err)
+			log.Printf("Invalid client IP address: %s with error: %s", ssValue["RemAddress"], err)
+			continue
 		}
-		ss_test, err := PackDataIntoSchema(ss_value, log_time, testName)
+		ssTest, err := PackDataIntoSchema(ssValue, logTime, testName)
 		if err != nil {
-			metrics.ErrorCount.WithLabelValues(
-				ss.TableName(), "ss", "corrupted data").Inc()
 			metrics.TestCount.WithLabelValues(
 				ss.TableName(), "ss", "corrupted data").Inc()
 			log.Printf("cannot pack data into sidestream schema: %v\n", err)
-			return err
+			continue
 		}
-		err = ss.inserter.InsertRow(ss_test)
+		err = ss.inserter.InsertRow(ssTest)
 		if err != nil {
 			metrics.ErrorCount.WithLabelValues(
 				ss.TableName(), "ss", "insert-err").Inc()

parser/util.go

@@ -1,9 +1,12 @@
 package web100
 
 import (
+	"errors"
 	"io"
 	"io/ioutil"
+	"net"
 	"strings"
+	"syscall"
 )
 
 // ParseWeb100Definitions reads all web100 variable definitions from tcpKis and
@@ -38,3 +41,102 @@ func ParseWeb100Definitions(tcpKis io.Reader) (map[string]string, error) {
 	}
 	return legacyNamesToNewNames, nil
 }
+
+// ParseIPFamily determines whether an IP string is v4 or v6
+func ParseIPFamily(ipStr string) int64 {
+	ip := net.ParseIP(ipStr)
+	if ip.To4() != nil {
+		return syscall.AF_INET
+	} else if ip.To16() != nil {
+		return syscall.AF_INET6
+	}
+	return -1
+}
+
+// IP validation errors.
+var (
+	ErrIPIsUnparseable   = errors.New("IP not parsable")
+	ErrIPIsUnconvertible = errors.New("IP not convertible to ipv4 or ipv6")
+	ErrIPIsZero          = errors.New("IP is zero/unspecified")
+	ErrIPIsUnroutable    = errors.New("IP is nonroutable")
+
+	ErrIPv4IsPrivate    = errors.New("private IPv4")
+	ErrIPv6IsPrivate    = errors.New("private IPv6")
+	ErrIPv4IsUnroutable = errors.New("unroutable IPv4")
+	ErrIPv6IsUnroutable = errors.New("unroutable IPv6")
+
+	ErrIPv6MultipleTripleColon = errors.New("more than one ::: in an ip address")
+	ErrIPv6QuadColon           = errors.New("IP address contains :::: ")
+)
+
+// NormalizeIPv6 fixes triple colon ::: which is produced by sidestream.
+// This error is produced by older versions of the c-web100 library, which is still
+// used by sidestream.
+func NormalizeIPv6(ipStr string) (string, error) {
+	split := strings.Split(ipStr, ":::")
+	switch len(split) {
+	case 1:
+		return ipStr, nil
+	case 2:
+		if split[1][0] == ':' {
+			return "", ErrIPv6QuadColon
+		}
+		return split[0] + "::" + split[1], nil
+	default:
+		return ipStr, ErrIPv6MultipleTripleColon
+	}
+}
+
+// ValidateIP validates (and possibly repairs) IP addresses.
+// Return nil if it is a valid IPv4 or IPv6 address (or can be repaired), non-nil otherwise.
+func ValidateIP(ipStr string) error {
+	ipStr, err := NormalizeIPv6(ipStr)
+	if err != nil {
+		return err
+	}
+	ip := net.ParseIP(ipStr)
+	if ip == nil || (ip.To4() == nil && ip.To16() == nil) {
+		return ErrIPIsUnparseable
+	}
+	if ip.To4() == nil && ip.To16() == nil {
+		return ErrIPIsUnconvertible
+	}
+	if ip.IsUnspecified() {
+		return ErrIPIsZero
+	}
+
+	if ip.IsLoopback() || ip.IsMulticast() || ip.IsLinkLocalUnicast() {
+		return ErrIPIsUnroutable
+	}
+
+	if ip.To4() != nil {
+		// Check whether it is a private IP.
+		_, private24BitBlock, _ := net.ParseCIDR("10.0.0.0/8")
+		_, private20BitBlock, _ := net.ParseCIDR("172.16.0.0/12")
+		_, private16BitBlock, _ := net.ParseCIDR("192.168.0.0/16")
+		_, private22BitBlock, _ := net.ParseCIDR("100.64.0.0/10")
+		if private24BitBlock.Contains(ip) || private20BitBlock.Contains(ip) ||
+			private16BitBlock.Contains(ip) || private22BitBlock.Contains(ip) {
+			return ErrIPv4IsPrivate
+		}
+
+		// check whether it is nonroutable IP
+		_, nonroutable1, _ := net.ParseCIDR("0.0.0.0/8")
+		_, nonroutable2, _ := net.ParseCIDR("192.0.2.0/24")
+		if nonroutable1.Contains(ip) || nonroutable2.Contains(ip) {
+			return ErrIPv4IsUnroutable
+		}
+	} else if ip.To16() != nil {
+		_, private7BitBlock, _ := net.ParseCIDR("FC00::/7")
+		if private7BitBlock.Contains(ip) {
+			return ErrIPv6IsPrivate
+		}
+
+		_, nonroutable1, _ := net.ParseCIDR("2001:db8::/32")
+		_, nonroutable2, _ := net.ParseCIDR("fec0::/10")
+		if nonroutable1.Contains(ip) || nonroutable2.Contains(ip) {
+			return ErrIPv6IsUnroutable
+		}
+	}
+	return nil
+}

parser/util_test.go

@@ -2,6 +2,8 @@ package web100_test
 
 import (
 	"bytes"
+	"fmt"
+	"syscall"
 	"testing"
 
 	"github.com/m-lab/etl/web100"
@@ -72,3 +74,47 @@ func TestParseWeb100Definitions(t *testing.T) {
 		}
 	}
 }
+
+func TestParseIPFamily(t *testing.T) {
+	if web100.ParseIPFamily("1.2.3.4") != syscall.AF_INET {
+		t.Fatalf("IPv4 address not parsed correctly.")
+	}
+	if web100.ParseIPFamily("2001:db8:0:1:1:1:1:1") != syscall.AF_INET6 {
+		t.Fatalf("IPv6 address not parsed correctly.")
+	}
+}
+
+func TestValidateIP(t *testing.T) {
+	if web100.ValidateIP("1.2.3.4") != nil {
+		fmt.Println(web100.ValidateIP("1.2.3.4"))
+		t.Fatalf("Valid IPv4 was identified as invalid.")
+	}
+	if web100.ValidateIP("2620:0:1000:2304:8053:fe91:6e2e:b4f1") != nil {
+		t.Fatalf("Valid IPv6 was identified as invalid.")
+	}
+	if web100.ValidateIP("::") == nil || web100.ValidateIP("0.0.0.0") == nil ||
+		web100.ValidateIP("abc.0.0.0") == nil || web100.ValidateIP("1.0.0.256") == nil {
+		t.Fatalf("Invalid IP was identified as valid.")
+	}
+	if web100.ValidateIP("172.16.0.1") == nil {
+		t.Fatalf("Private IP was not identified as invalid IP.")
+	}
+	if web100.ValidateIP("127.0.0.1") == nil || web100.ValidateIP("::ffff:127.0.0.1") == nil {
+		t.Fatalf("Nonroutable IP was not identified as invalid IP.")
+	}
+
+	if web100.ValidateIP("2001:668:1f:22:::81") != nil {
+		t.Fatalf("IPv6 with triple colon was not repaired.")
+	}
+	if web100.ValidateIP("2001:668:1f::::81") == nil {
+		t.Fatalf("IPv6 with quad colon was allowed.")
+	}
+}
+
+// To run benchmark...
+// go test -bench=. ./parser/...
+func BenchmarkValidateIPv4(b *testing.B) {
+	for i := 0; i < b.N; i++ {
+		_ = web100.ValidateIP("1.2.3.4")
+	}
+}