Merge pull request #703 from rod-hynes/multi-dest-bytes

Add DestinationBytesMetricsASNs

psiphon/common/parameters/parameters.go

@@ -327,6 +327,7 @@ const (
 	RestrictDirectProviderIDsClientProbability         = "RestrictDirectProviderIDsClientProbability"
 	UpstreamProxyAllowAllServerEntrySources            = "UpstreamProxyAllowAllServerEntrySources"
 	DestinationBytesMetricsASN                         = "DestinationBytesMetricsASN"
+	DestinationBytesMetricsASNs                        = "DestinationBytesMetricsASNs"
 	DNSResolverAttemptsPerServer                       = "DNSResolverAttemptsPerServer"
 	DNSResolverAttemptsPerPreferredServer              = "DNSResolverAttemptsPerPreferredServer"
 	DNSResolverRequestTimeout                          = "DNSResolverRequestTimeout"
@@ -826,7 +827,8 @@ var defaultParameters = map[string]struct {
 
 	UpstreamProxyAllowAllServerEntrySources: {value: false},
 
-	DestinationBytesMetricsASN: {value: "", flags: serverSideOnly},
+	DestinationBytesMetricsASN:  {value: "", flags: serverSideOnly},
+	DestinationBytesMetricsASNs: {value: []string{}, flags: serverSideOnly},
 
 	DNSResolverAttemptsPerServer:                {value: 2, minimum: 1},
 	DNSResolverAttemptsPerPreferredServer:       {value: 1, minimum: 1},

psiphon/server/server_test.go

@@ -578,17 +578,32 @@ func TestBurstMonitorAndDestinationBytes(t *testing.T) {
 		})
 }
 
+func TestBurstMonitorAndLegacyDestinationBytes(t *testing.T) {
+	runServer(t,
+		&runServerConfig{
+			tunnelProtocol:           "OSSH",
+			requireAuthorization:     true,
+			doTunneledWebRequest:     true,
+			doTunneledNTPRequest:     true,
+			doDanglingTCPConn:        true,
+			doBurstMonitor:           true,
+			doLegacyDestinationBytes: true,
+			doLogHostProvider:        true,
+		})
+}
+
 func TestChangeBytesConfig(t *testing.T) {
 	runServer(t,
 		&runServerConfig{
-			tunnelProtocol:       "OSSH",
-			requireAuthorization: true,
-			doTunneledWebRequest: true,
-			doTunneledNTPRequest: true,
-			doDanglingTCPConn:    true,
-			doDestinationBytes:   true,
-			doChangeBytesConfig:  true,
-			doLogHostProvider:    true,
+			tunnelProtocol:           "OSSH",
+			requireAuthorization:     true,
+			doTunneledWebRequest:     true,
+			doTunneledNTPRequest:     true,
+			doDanglingTCPConn:        true,
+			doDestinationBytes:       true,
+			doLegacyDestinationBytes: true,
+			doChangeBytesConfig:      true,
+			doLogHostProvider:        true,
 		})
 }
 
@@ -646,34 +661,35 @@ func TestLegacyAPIEncoding(t *testing.T) {
 }
 
 type runServerConfig struct {
-	tunnelProtocol       string
-	clientTunnelProtocol string
-	passthrough          bool
-	tlsProfile           string
-	doHotReload          bool
-	doDefaultSponsorID   bool
-	denyTrafficRules     bool
-	requireAuthorization bool
-	omitAuthorization    bool
-	doTunneledWebRequest bool
-	doTunneledNTPRequest bool
-	applyPrefix          bool
-	forceFragmenting     bool
-	forceLivenessTest    bool
-	doPruneServerEntries bool
-	doDanglingTCPConn    bool
-	doPacketManipulation bool
-	doBurstMonitor       bool
-	doSplitTunnel        bool
-	limitQUICVersions    bool
-	doDestinationBytes   bool
-	doChangeBytesConfig  bool
-	doLogHostProvider    bool
-	inspectFlows         bool
-	doSteeringIP         bool
-	doTargetBrokerSpecs  bool
-	useLegacyAPIEncoding bool
-	doPersonalPairing    bool
+	tunnelProtocol           string
+	clientTunnelProtocol     string
+	passthrough              bool
+	tlsProfile               string
+	doHotReload              bool
+	doDefaultSponsorID       bool
+	denyTrafficRules         bool
+	requireAuthorization     bool
+	omitAuthorization        bool
+	doTunneledWebRequest     bool
+	doTunneledNTPRequest     bool
+	applyPrefix              bool
+	forceFragmenting         bool
+	forceLivenessTest        bool
+	doPruneServerEntries     bool
+	doDanglingTCPConn        bool
+	doPacketManipulation     bool
+	doBurstMonitor           bool
+	doSplitTunnel            bool
+	limitQUICVersions        bool
+	doDestinationBytes       bool
+	doLegacyDestinationBytes bool
+	doChangeBytesConfig      bool
+	doLogHostProvider        bool
+	inspectFlows             bool
+	doSteeringIP             bool
+	doTargetBrokerSpecs      bool
+	useLegacyAPIEncoding     bool
+	doPersonalPairing        bool
 }
 
 var (
@@ -776,7 +792,8 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 		runConfig.applyPrefix ||
 		runConfig.forceFragmenting ||
 		runConfig.doBurstMonitor ||
-		runConfig.doDestinationBytes
+		runConfig.doDestinationBytes ||
+		runConfig.doLegacyDestinationBytes
 
 	// All servers require a tactics config with valid keys.
 	tacticsRequestPublicKey, tacticsRequestPrivateKey, tacticsRequestObfuscatedKey, err :=
@@ -912,6 +929,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 			livenessTestSize,
 			runConfig.doBurstMonitor,
 			runConfig.doDestinationBytes,
+			runConfig.doLegacyDestinationBytes,
 			runConfig.applyPrefix,
 			runConfig.forceFragmenting,
 			"classic",
@@ -1181,6 +1199,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 				livenessTestSize,
 				runConfig.doBurstMonitor,
 				runConfig.doDestinationBytes,
+				runConfig.doLegacyDestinationBytes,
 				runConfig.applyPrefix,
 				runConfig.forceFragmenting,
 				"consistent",
@@ -1623,7 +1642,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 
 	if runConfig.doChangeBytesConfig {
 
-		if !runConfig.doDestinationBytes {
+		if !runConfig.doDestinationBytes || !runConfig.doLegacyDestinationBytes {
 			t.Fatalf("invalid test configuration")
 		}
 
@@ -1649,6 +1668,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 			livenessTestSize,
 			runConfig.doBurstMonitor,
 			false,
+			false,
 			runConfig.applyPrefix,
 			runConfig.forceFragmenting,
 			"consistent",
@@ -1796,6 +1816,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 		expectQUICVersion = limitQUICVersions[0]
 	}
 	expectDestinationBytesFields := runConfig.doDestinationBytes && !runConfig.doChangeBytesConfig
+	expectLegacyDestinationBytesFields := runConfig.doLegacyDestinationBytes && !runConfig.doChangeBytesConfig
 	expectMeekHTTPVersion := ""
 	if protocol.TunnelProtocolUsesMeek(runConfig.tunnelProtocol) {
 		if protocol.TunnelProtocolUsesFrontedMeek(runConfig.tunnelProtocol) {
@@ -1829,6 +1850,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 			expectUDPDataTransfer,
 			expectQUICVersion,
 			expectDestinationBytesFields,
+			expectLegacyDestinationBytesFields,
 			passthroughAddress,
 			expectMeekHTTPVersion,
 			inproxyTestConfig,
@@ -2086,6 +2108,7 @@ func checkExpectedServerTunnelLogFields(
 	expectUDPDataTransfer bool,
 	expectQUICVersion string,
 	expectDestinationBytesFields bool,
+	expectLegacyDestinationBytesFields bool,
 	expectPassthroughAddress *string,
 	expectMeekHTTPVersion string,
 	inproxyTestConfig *inproxyTestConfig,
@@ -2691,6 +2714,45 @@ func checkExpectedServerTunnelLogFields(
 		}
 	}
 
+	for _, name := range []string{
+		"asn_dest_bytes",
+		"asn_dest_bytes_up_tcp",
+		"asn_dest_bytes_down_tcp",
+		"asn_dest_bytes_up_udp",
+		"asn_dest_bytes_down_udp",
+	} {
+		if expectDestinationBytesFields && fields[name] == nil {
+			return fmt.Errorf("missing expected field '%s'", name)
+
+		} else if !expectDestinationBytesFields && fields[name] != nil {
+			return fmt.Errorf("unexpected field '%s'", name)
+		}
+	}
+
+	if expectDestinationBytesFields {
+		for _, pair := range [][]string{
+			{"asn_dest_bytes", "bytes"},
+			{"asn_dest_bytes_up_tcp", "bytes_up_tcp"},
+			{"asn_dest_bytes_down_tcp", "bytes_down_tcp"},
+			{"asn_dest_bytes_up_udp", "bytes_up_udp"},
+			{"asn_dest_bytes_down_udp", "bytes_down_udp"},
+		} {
+			if _, ok := fields[pair[0]].(map[string]any)[testGeoIPASN].(float64); !ok {
+				return fmt.Errorf("missing field entry %s: '%v'", pair[0], testGeoIPASN)
+			}
+			value0 := int64(fields[pair[0]].(map[string]any)[testGeoIPASN].(float64))
+			value1 := int64(fields[pair[1]].(float64))
+			ok := value0 == value1
+			if pair[0] == "asn_dest_bytes_up_udp" || pair[0] == "asn_dest_bytes_down_udp" || pair[0] == "asn_dest_bytes" {
+				// DNS requests are excluded from destination bytes counting
+				ok = value0 > 0 && value0 < value1
+			}
+			if !ok {
+				return fmt.Errorf("unexpected field value %s: %v != %v", pair[0], fields[pair[0]], fields[pair[1]])
+			}
+		}
+	}
+
 	for _, name := range []string{
 		"dest_bytes_asn",
 		"dest_bytes_up_tcp",
@@ -2699,15 +2761,15 @@ func checkExpectedServerTunnelLogFields(
 		"dest_bytes_down_udp",
 		"dest_bytes",
 	} {
-		if expectDestinationBytesFields && fields[name] == nil {
+		if expectLegacyDestinationBytesFields && fields[name] == nil {
 			return fmt.Errorf("missing expected field '%s'", name)
 
-		} else if !expectDestinationBytesFields && fields[name] != nil {
+		} else if !expectLegacyDestinationBytesFields && fields[name] != nil {
 			return fmt.Errorf("unexpected field '%s'", name)
 		}
 	}
 
-	if expectDestinationBytesFields {
+	if expectLegacyDestinationBytesFields {
 		name := "dest_bytes_asn"
 		if fields[name].(string) != testGeoIPASN {
 			return fmt.Errorf("unexpected field value %s: '%v'", name, fields[name])
@@ -3385,6 +3447,7 @@ func paveTacticsConfigFile(
 	livenessTestSize int,
 	doBurstMonitor bool,
 	doDestinationBytes bool,
+	doLegacyDestinationBytes bool,
 	applyOsshPrefix bool,
 	enableOsshPrefixFragmenting bool,
 	discoveryStategy string,
@@ -3406,6 +3469,7 @@ func paveTacticsConfigFile(
           %s
           %s
           %s
+          %s
           "LimitTunnelProtocols" : ["%s"],
           "FragmentorLimitProtocols" : ["%s"],
           "FragmentorProbability" : 1.0,
@@ -3490,6 +3554,13 @@ func paveTacticsConfigFile(
 	destinationBytesParameters := ""
 	if doDestinationBytes {
 		destinationBytesParameters = fmt.Sprintf(`
+          "DestinationBytesMetricsASNs" : ["%s"],
+	`, testGeoIPASN)
+	}
+
+	legacyDestinationBytesParameters := ""
+	if doLegacyDestinationBytes {
+		legacyDestinationBytesParameters = fmt.Sprintf(`
           "DestinationBytesMetricsASN" : "%s",
 	`, testGeoIPASN)
 	}
@@ -3513,6 +3584,7 @@ func paveTacticsConfigFile(
 		tacticsRequestObfuscatedKey,
 		burstParameters,
 		destinationBytesParameters,
+		legacyDestinationBytesParameters,
 		osshPrefix,
 		inproxyParametersJSON,
 		tunnelProtocol,

psiphon/server/tunnelServer.go

@@ -1809,9 +1809,7 @@ type sshClient struct {
 	sendAlertRequests                    chan protocol.AlertRequest
 	sentAlertRequests                    map[string]bool
 	peakMetrics                          peakMetrics
-	destinationBytesMetricsASN           string
-	tcpDestinationBytesMetrics           destinationBytesMetrics
-	udpDestinationBytesMetrics           destinationBytesMetrics
+	destinationBytesMetrics              map[string]*protocolDestinationBytesMetrics
 }
 
 type trafficState struct {
@@ -1941,6 +1939,11 @@ type handshakeState struct {
 	inproxyRelayLogFields   common.LogFields
 }
 
+type protocolDestinationBytesMetrics struct {
+	tcpMetrics destinationBytesMetrics
+	udpMetrics destinationBytesMetrics
+}
+
 type destinationBytesMetrics struct {
 	bytesUp   int64
 	bytesDown int64
@@ -3365,20 +3368,18 @@ func (sshClient *sshClient) logTunnel(additionalMetrics []LogFields) {
 	logFields["random_stream_downstream_bytes"] = sshClient.postHandshakeRandomStreamMetrics.downstreamBytes
 	logFields["random_stream_sent_downstream_bytes"] = sshClient.postHandshakeRandomStreamMetrics.sentDownstreamBytes
 
-	if sshClient.destinationBytesMetricsASN != "" {
+	if sshClient.destinationBytesMetrics != nil {
 
-		// Check if the configured DestinationBytesMetricsASN has changed
-		// (or been cleared). If so, don't log and discard the accumulated
-		// bytes to ensure we don't continue to record stats as previously
-		// configured.
+		// Only log destination bytes for ASNs that remain enabled in tactics.
 		//
-		// Any counts accumulated before the DestinationBytesMetricsASN change
-		// are lost. At this time we can't change
-		// sshClient.destinationBytesMetricsASN dynamically, after a tactics
-		// hot reload, as there may be destination bytes port forwards that
-		// were in place before the change, which will continue to count.
-
-		logDestBytes := true
+		// Any counts accumulated before DestinationBytesMetricsASN[s] changes
+		// are lost. At this time we can't change destination byte counting
+		// dynamically, after a tactics hot reload, as there may be
+		// destination bytes port forwards that were in place before the
+		// change, which will continue to count.
+
+		destinationBytesMetricsASNs := []string{}
+		destinationBytesMetricsASN := ""
 		if sshClient.sshServer.support.ServerTacticsParametersCache != nil {
 
 			// Target this using the client, not peer, GeoIP. In the case of
@@ -3387,24 +3388,69 @@ func (sshClient *sshClient) logTunnel(additionalMetrics []LogFields) {
 			// have transferred.
 
 			p, err := sshClient.sshServer.support.ServerTacticsParametersCache.Get(sshClient.clientGeoIPData)
-			if err != nil || p.IsNil() ||
-				sshClient.destinationBytesMetricsASN != p.String(parameters.DestinationBytesMetricsASN) {
-				logDestBytes = false
+			if err == nil && !p.IsNil() {
+				destinationBytesMetricsASNs = p.Strings(parameters.DestinationBytesMetricsASNs)
+				destinationBytesMetricsASN = p.String(parameters.DestinationBytesMetricsASN)
 			}
+			p.Close()
 		}
 
-		if logDestBytes {
-			bytesUpTCP := sshClient.tcpDestinationBytesMetrics.getBytesUp()
-			bytesDownTCP := sshClient.tcpDestinationBytesMetrics.getBytesDown()
-			bytesUpUDP := sshClient.udpDestinationBytesMetrics.getBytesUp()
-			bytesDownUDP := sshClient.udpDestinationBytesMetrics.getBytesDown()
-
-			logFields["dest_bytes_asn"] = sshClient.destinationBytesMetricsASN
-			logFields["dest_bytes_up_tcp"] = bytesUpTCP
-			logFields["dest_bytes_down_tcp"] = bytesDownTCP
-			logFields["dest_bytes_up_udp"] = bytesUpUDP
-			logFields["dest_bytes_down_udp"] = bytesDownUDP
-			logFields["dest_bytes"] = bytesUpTCP + bytesDownTCP + bytesUpUDP + bytesDownUDP
+		if destinationBytesMetricsASN != "" {
+
+			// Log any parameters.DestinationBytesMetricsASN data in the
+			// legacy log field format.
+
+			destinationBytesMetrics, ok :=
+				sshClient.destinationBytesMetrics[destinationBytesMetricsASN]
+
+			if ok {
+				bytesUpTCP := destinationBytesMetrics.tcpMetrics.getBytesUp()
+				bytesDownTCP := destinationBytesMetrics.tcpMetrics.getBytesDown()
+				bytesUpUDP := destinationBytesMetrics.udpMetrics.getBytesUp()
+				bytesDownUDP := destinationBytesMetrics.udpMetrics.getBytesDown()
+
+				logFields["dest_bytes_asn"] = destinationBytesMetricsASN
+				logFields["dest_bytes"] = bytesUpTCP + bytesDownTCP + bytesUpUDP + bytesDownUDP
+				logFields["dest_bytes_up_tcp"] = bytesUpTCP
+				logFields["dest_bytes_down_tcp"] = bytesDownTCP
+				logFields["dest_bytes_up_udp"] = bytesUpUDP
+				logFields["dest_bytes_down_udp"] = bytesDownUDP
+			}
+		}
+
+		if len(destinationBytesMetricsASNs) > 0 {
+
+			destBytes := make(map[string]int64)
+			destBytesUpTCP := make(map[string]int64)
+			destBytesDownTCP := make(map[string]int64)
+			destBytesUpUDP := make(map[string]int64)
+			destBytesDownUDP := make(map[string]int64)
+
+			for _, ASN := range destinationBytesMetricsASNs {
+
+				destinationBytesMetrics, ok :=
+					sshClient.destinationBytesMetrics[ASN]
+				if !ok {
+					continue
+				}
+
+				bytesUpTCP := destinationBytesMetrics.tcpMetrics.getBytesUp()
+				bytesDownTCP := destinationBytesMetrics.tcpMetrics.getBytesDown()
+				bytesUpUDP := destinationBytesMetrics.udpMetrics.getBytesUp()
+				bytesDownUDP := destinationBytesMetrics.udpMetrics.getBytesDown()
+
+				destBytes[ASN] = bytesUpTCP + bytesDownTCP + bytesUpUDP + bytesDownUDP
+				destBytesUpTCP[ASN] = bytesUpTCP
+				destBytesDownTCP[ASN] = bytesDownTCP
+				destBytesUpUDP[ASN] = bytesUpUDP
+				destBytesDownUDP[ASN] = bytesDownUDP
+			}
+
+			logFields["asn_dest_bytes"] = destBytes
+			logFields["asn_dest_bytes_up_tcp"] = destBytesUpTCP
+			logFields["asn_dest_bytes_down_tcp"] = destBytesDownTCP
+			logFields["asn_dest_bytes_up_udp"] = destBytesUpUDP
+			logFields["asn_dest_bytes_down_udp"] = destBytesDownUDP
 		}
 	}
 
@@ -4112,26 +4158,54 @@ func (sshClient *sshClient) setDestinationBytesMetrics() {
 		return
 	}
 
-	sshClient.destinationBytesMetricsASN = p.String(parameters.DestinationBytesMetricsASN)
+	ASNs := p.Strings(parameters.DestinationBytesMetricsASNs)
+
+	// Merge in any legacy parameters.DestinationBytesMetricsASN
+	// configuration. Data for this target will be logged using the legacy
+	// log field format; see logTunnel. If an ASN is in _both_ configuration
+	// parameters, its data will be logged in both log field formats.
+	ASN := p.String(parameters.DestinationBytesMetricsASN)
+
+	if len(ASNs) == 0 && ASN == "" {
+		return
+	}
+
+	sshClient.destinationBytesMetrics = make(map[string]*protocolDestinationBytesMetrics)
+
+	for _, ASN := range ASNs {
+		if ASN != "" {
+			sshClient.destinationBytesMetrics[ASN] = &protocolDestinationBytesMetrics{}
+		}
+	}
+
+	if ASN != "" {
+		sshClient.destinationBytesMetrics[ASN] = &protocolDestinationBytesMetrics{}
+	}
 }
 
 func (sshClient *sshClient) newDestinationBytesMetricsUpdater(portForwardType int, IPAddress net.IP) *destinationBytesMetrics {
 	sshClient.Lock()
 	defer sshClient.Unlock()
 
-	if sshClient.destinationBytesMetricsASN == "" {
+	if sshClient.destinationBytesMetrics == nil {
 		return nil
 	}
 
-	if sshClient.sshServer.support.GeoIPService.LookupISPForIP(IPAddress).ASN != sshClient.destinationBytesMetricsASN {
+	destinationASN := sshClient.sshServer.support.GeoIPService.LookupISPForIP(IPAddress).ASN
+
+	// Future enhancement: for 5 or fewer ASNs, iterate over a slice instead
+	// of using a map? See, for example, stringLookupThreshold in
+	// common/tactics.
+	metrics, ok := sshClient.destinationBytesMetrics[destinationASN]
+	if !ok {
 		return nil
 	}
 
 	if portForwardType == portForwardTypeTCP {
-		return &sshClient.tcpDestinationBytesMetrics
+		return &metrics.tcpMetrics
 	}
 
-	return &sshClient.udpDestinationBytesMetrics
+	return &metrics.udpMetrics
 }
 
 func (sshClient *sshClient) getActivityUpdaters(portForwardType int, IPAddress net.IP) []common.ActivityUpdater {