Riseup tests improvements

internal/experiment/riseupvpn/riseupvpn.go

@@ -17,7 +17,7 @@ import (
 
 const (
 	testName      = "riseupvpn"
-	testVersion   = "0.2.0"
+	testVersion   = "0.3.0"
 	eipServiceURL = "https://api.black.riseup.net:443/3/config/eip-service.json"
 	providerURL   = "https://riseup.net/provider.json"
 	geoServiceURL = "https://api.black.riseup.net:9001/json"
@@ -29,13 +29,17 @@ type EipService struct {
 	Gateways []GatewayV3
 }
 
+// Capabilities is a list of transports a gateway supports
+type Capabilities struct {
+	Transport []TransportV3
+}
+
 // GatewayV3 describes a gateway.
 type GatewayV3 struct {
-	Capabilities struct {
-		Transport []TransportV3
-	}
-	Host      string
-	IPAddress string `json:"ip_address"`
+	Capabilities Capabilities
+	Host         string
+	IPAddress    string `json:"ip_address"`
+	Location     string `json:"location"`
 }
 
 // TransportV3 describes a transport.
@@ -53,6 +57,24 @@ type GatewayConnection struct {
 	TransportType string `json:"transport_type"`
 }
 
+// GatewayLoad describes the load of a single Gateway.
+type GatewayLoad struct {
+	Host     string  `json:"host"`
+	Fullness float64 `json:"fullness"`
+	Overload bool    `json:"overload"`
+}
+
+// GeoService represents the geoService API (also known as menshen) json response
+type GeoService struct {
+	IPAddress      string        `json:"ip"`
+	Country        string        `json:"cc"`
+	City           string        `json:"city"`
+	Latitude       float64       `json:"lat"`
+	Longitude      float64       `json:"lon"`
+	Gateways       []string      `json:"gateways"`
+	SortedGateways []GatewayLoad `json:"sortedGateways"`
+}
+
 // Config contains the riseupvpn experiment config.
 type Config struct {
 	urlgetter.Config
@@ -61,7 +83,7 @@ type Config struct {
 // TestKeys contains riseupvpn test keys.
 type TestKeys struct {
 	urlgetter.TestKeys
-	APIFailure      *string             `json:"api_failure"`
+	APIFailure      []string            `json:"api_failure"`
 	APIStatus       string              `json:"api_status"`
 	CACertStatus    bool                `json:"ca_cert_status"`
 	FailingGateways []GatewayConnection `json:"failing_gateways"`
@@ -86,12 +108,13 @@ func (tk *TestKeys) UpdateProviderAPITestKeys(v urlgetter.MultiOutput) {
 	tk.Requests = append(tk.Requests, v.TestKeys.Requests...)
 	tk.TCPConnect = append(tk.TCPConnect, v.TestKeys.TCPConnect...)
 	tk.TLSHandshakes = append(tk.TLSHandshakes, v.TestKeys.TLSHandshakes...)
-	if tk.APIStatus != "ok" {
-		return // we already flipped the state
-	}
 	if v.TestKeys.Failure != nil {
-		tk.APIStatus = "blocked"
-		tk.APIFailure = v.TestKeys.Failure
+		for _, request := range v.TestKeys.Requests {
+			if request.Request.URL == eipServiceURL && request.Failure != nil {
+				tk.APIStatus = "blocked"
+			}
+		}
+		tk.APIFailure = append(tk.APIFailure, *v.TestKeys.Failure)
 		return
 	}
 }
@@ -147,11 +170,6 @@ func (tk *TestKeys) AddCACertFetchTestKeys(testKeys urlgetter.TestKeys) {
 	tk.Requests = append(tk.Requests, testKeys.Requests...)
 	tk.TCPConnect = append(tk.TCPConnect, testKeys.TCPConnect...)
 	tk.TLSHandshakes = append(tk.TLSHandshakes, testKeys.TLSHandshakes...)
-	if testKeys.Failure != nil {
-		tk.APIStatus = "blocked"
-		tk.APIFailure = tk.Failure
-		tk.CACertStatus = false
-	}
 }
 
 // Measurer performs the measurement.
@@ -204,21 +222,17 @@ func (m Measurer) Run(ctx context.Context, args *model.ExperimentArgs) error {
 			FailOnHTTPError: true,
 		}},
 	}
-	for entry := range multi.CollectOverall(ctx, inputs, 0, 50, "riseupvpn", callbacks) {
+	for entry := range multi.CollectOverall(ctx, inputs, 0, 20, "riseupvpn", callbacks) {
 		tk := entry.TestKeys
 		testkeys.AddCACertFetchTestKeys(tk)
 		if tk.Failure != nil {
-			// TODO(bassosimone,cyberta): should we update the testkeys
-			// in this case (e.g., APIFailure?)
-			// See https://github.com/ooni/probe/issues/1432.
-			return nil
-		}
-		if ok := certPool.AppendCertsFromPEM([]byte(tk.HTTPResponseBody)); !ok {
 			testkeys.CACertStatus = false
-			testkeys.APIStatus = "blocked"
-			errorValue := "invalid_ca"
-			testkeys.APIFailure = &errorValue
-			return nil
+			testkeys.APIFailure = append(testkeys.APIFailure, *tk.Failure)
+			certPool = nil
+		} else if ok := certPool.AppendCertsFromPEM([]byte(tk.HTTPResponseBody)); !ok {
+			testkeys.CACertStatus = false
+			testkeys.APIFailure = append(testkeys.APIFailure, "invalid_ca")
+			certPool = nil
 		}
 	}
 
@@ -230,22 +244,35 @@ func (m Measurer) Run(ctx context.Context, args *model.ExperimentArgs) error {
 			CertPool:        certPool,
 			Method:          "GET",
 			FailOnHTTPError: true,
+			NoTLSVerify:     !testkeys.CACertStatus,
 		}},
 		{Target: eipServiceURL, Config: urlgetter.Config{
 			CertPool:        certPool,
 			Method:          "GET",
 			FailOnHTTPError: true,
+			NoTLSVerify:     !testkeys.CACertStatus,
 		}},
 		{Target: geoServiceURL, Config: urlgetter.Config{
 			CertPool:        certPool,
 			Method:          "GET",
 			FailOnHTTPError: true,
+			NoTLSVerify:     !testkeys.CACertStatus,
 		}},
 	}
-	for entry := range multi.CollectOverall(ctx, inputs, 1, 50, "riseupvpn", callbacks) {
+
+	for entry := range multi.CollectOverall(ctx, inputs, 1, 20, "riseupvpn", callbacks) {
 		testkeys.UpdateProviderAPITestKeys(entry)
 	}
 
+	if testkeys.APIStatus == "blocked" {
+		for _, input := range inputs {
+			input.Config.Tunnel = "torsf"
+		}
+		for entry := range multi.CollectOverall(ctx, inputs, 1, 20, "riseupvpn", callbacks) {
+			testkeys.UpdateProviderAPITestKeys(entry)
+		}
+	}
+
 	// test gateways now
 	testkeys.TransportStatus = map[string]string{}
 	gateways := parseGateways(testkeys)
@@ -299,18 +326,117 @@ func generateMultiInputs(gateways []GatewayV3, transportType string) []urlgetter
 }
 
 func parseGateways(testKeys *TestKeys) []GatewayV3 {
+	var eipService *EipService = nil
+	var geoService *GeoService = nil
 	for _, requestEntry := range testKeys.Requests {
 		if requestEntry.Request.URL == eipServiceURL && requestEntry.Failure == nil {
-			// TODO(bassosimone,cyberta): is it reasonable that we discard
-			// the error when the JSON we fetched cannot be parsed?
-			// See https://github.com/ooni/probe/issues/1432
-			eipService, err := DecodeEIP3(requestEntry.Response.Body.Value)
-			if err == nil {
-				return eipService.Gateways
+			var err error = nil
+			eipService, err = DecodeEIP3(requestEntry.Response.Body.Value)
+			if err != nil {
+				testKeys.APIFailure = append(testKeys.APIFailure, "invalid_eipservice_response")
+				return nil
+			}
+		} else if requestEntry.Request.URL == geoServiceURL && requestEntry.Failure == nil {
+			var err error = nil
+			geoService, err = DecodeGeoService(requestEntry.Response.Body.Value)
+			if err != nil {
+				testKeys.APIFailure = append(testKeys.APIFailure, "invalid_geoservice_response")
 			}
 		}
 	}
-	return nil
+	return filterGateways(eipService, geoService)
+}
+
+// filterGateways selects a subset of available gateways supporting obfs4
+func filterGateways(eipService *EipService, geoService *GeoService) []GatewayV3 {
+	var result []GatewayV3 = nil
+	if eipService != nil {
+		locations := getLocationsUnderTest(eipService, geoService)
+		for _, gateway := range eipService.Gateways {
+			if !gateway.hasTransport("obfs4") ||
+				!gateway.isLocationUnderTest(locations) ||
+				geoService != nil && !geoService.isHealthyGateway(gateway) {
+				continue
+			}
+			result = append(result, gateway)
+			if len(result) == 3 {
+				return result
+			}
+		}
+	}
+	return result
+}
+
+// getLocationsUnderTest parses all gateways supporting obfs4 and returns the two locations having most obfs4 bridges
+func getLocationsUnderTest(eipService *EipService, geoService *GeoService) []string {
+	var result []string = nil
+	if eipService != nil {
+		locationMap := map[string]int{}
+		locations := []string{}
+		for _, gateway := range eipService.Gateways {
+			if !gateway.hasTransport("obfs4") {
+				continue
+			}
+			if _, ok := locationMap[gateway.Location]; !ok {
+				locations = append(locations, gateway.Location)
+			}
+			locationMap[gateway.Location] += 1
+		}
+
+		location1 := ""
+		location2 := ""
+		for _, location := range locations {
+			if locationMap[location] > locationMap[location1] {
+				location2 = location1
+				location1 = location
+			} else if locationMap[location] > locationMap[location2] {
+				location2 = location
+			}
+		}
+		if location1 != "" {
+			result = append(result, location1)
+		}
+		if location2 != "" {
+			result = append(result, location2)
+		}
+	}
+
+	return result
+}
+
+func (gateway *GatewayV3) hasTransport(s string) bool {
+	for _, transport := range gateway.Capabilities.Transport {
+		if s == transport.Type {
+			return true
+		}
+	}
+	return false
+}
+
+func (gateway *GatewayV3) isLocationUnderTest(locations []string) bool {
+	for _, location := range locations {
+		if location == gateway.Location {
+			return true
+		}
+	}
+	return false
+}
+
+func (geoService *GeoService) isHealthyGateway(gateway GatewayV3) bool {
+	if geoService.SortedGateways == nil {
+		// Earlier versions of the geoservice don't include the sorted gateway list containing the load info,
+		// so we can't say anything about the load of a gateway in that case.
+		// We assume it's an healthy location. Riseup will switch to the updated API soon *fingers crossed*
+		return true
+	}
+	for _, gatewayLoad := range geoService.SortedGateways {
+		if gatewayLoad.Host == gateway.Host {
+			return !gatewayLoad.Overload
+		}
+	}
+
+	// gateways that are not included in the geoservice should be considered unusable
+	return false
 }
 
 // DecodeEIP3 decodes eip-service.json version 3
@@ -323,6 +449,16 @@ func DecodeEIP3(body string) (*EipService, error) {
 	return &eip, nil
 }
 
+// DecodeGeoService decodes geoService  json
+func DecodeGeoService(body string) (*GeoService, error) {
+	var gs GeoService
+	err := json.Unmarshal([]byte(body), &gs)
+	if err != nil {
+		return nil, err
+	}
+	return &gs, nil
+}
+
 // NewExperimentMeasurer creates a new ExperimentMeasurer.
 func NewExperimentMeasurer(config Config) model.ExperimentMeasurer {
 	return Measurer{Config: config}