feat: implement riseupvpn using the DSL

Part of ooni/probe#2502

pkg/experiment/fbmessenger/measurer.go

@@ -62,7 +62,7 @@ func (m *Measurer) Run(ctx context.Context, args *model.ExperimentArgs) error {
 		args.Session.Logger(), &dsl.NullMetrics{}, args.Measurement.MeasurementStartTimeSaved)
 	defer rtx.Close()
 
-	// evaluate the function and handle exceptions
+	// evaluate the pipeline and handle exceptions
 	argument0 := dsl.NewValue(&dsl.Void{})
 	if err := dsl.Try(pipeline.Run(ctx, rtx, argument0.AsGeneric())); err != nil {
 		return err

pkg/experiment/riseupvpn/riseupvpn.go

@@ -0,0 +1,101 @@
+// Package riseupvpn implements the riseupvpn experiment.
+package riseupvpn
+
+import (
+	"context"
+	"encoding/json"
+
+	"github.com/ooni/2023-05-richer-input/pkg/dsl"
+	"github.com/ooni/probe-engine/pkg/model"
+)
+
+// NewMeasurer returns a new [Measurer] instance.
+func NewMeasurer(rawOptions json.RawMessage) *Measurer {
+	return &Measurer{rawOptions}
+}
+
+// Measurer is the riseupvpn measurer.
+type Measurer struct {
+	// RawOptions contains the raw options for this experiment.
+	RawOptions json.RawMessage
+}
+
+var _ model.ExperimentMeasurer = &Measurer{}
+
+// ExperimentName implements model.ExperimentMeasurer
+func (m *Measurer) ExperimentName() string {
+	return "riseupvpn"
+}
+
+// ExperimentVersion implements model.ExperimentMeasurer
+func (m *Measurer) ExperimentVersion() string {
+	// TODO(bassosimone): the real experiment is at version 0.2.0 and
+	// we will _probably_ be fine by saying we're at 0.4.0 since the
+	// https://github.com/ooni/probe-cli/pull/1125 PR uses 0.3.0.
+	return "0.4.0"
+}
+
+// TestKeys contains the experiment test keys.
+type TestKeys struct {
+	*dsl.Observations
+}
+
+// Run implements model.ExperimentMeasurer
+func (m *Measurer) Run(ctx context.Context, args *model.ExperimentArgs) error {
+	// parse the targets
+	var astRoot dsl.LoadableASTNode
+	if err := json.Unmarshal(m.RawOptions, &astRoot); err != nil {
+		return err
+	}
+
+	// create an AST loader
+	loader := dsl.NewASTLoader()
+
+	// create the testkeys
+	tk := &TestKeys{}
+
+	// load and make the AST runnable
+	pipeline, err := loader.Load(&astRoot)
+	if err != nil {
+		return err
+	}
+
+	// TODO(bassosimone): both fbmessenger and riseupvpn lack
+	//
+	// 1. an explicit mechanism to report the bytes sent and received, but the
+	// implicit context-based mechanism probably works;
+	//
+	// 2. a DSL-based mechanism to increment the test progress percentage.
+
+	// create the DSL runtime
+	rtx := dsl.NewMeasurexliteRuntime(
+		args.Session.Logger(), &dsl.NullMetrics{}, args.Measurement.MeasurementStartTimeSaved)
+	defer rtx.Close()
+
+	// evaluate the pipeline and handle exceptions
+	argument0 := dsl.NewValue(&dsl.Void{})
+	if err := dsl.Try(pipeline.Run(ctx, rtx, argument0.AsGeneric())); err != nil {
+		return err
+	}
+
+	// obtain the observations
+	tk.Observations = dsl.ReduceObservations(rtx.ExtractObservations()...)
+
+	// save the testkeys
+	args.Measurement.TestKeys = tk
+	return nil
+}
+
+// SummaryKeys contains summary keys for this experiment.
+//
+// Note that this structure is part of the ABI contract with ooniprobe
+// therefore we should be careful when changing it.
+type SummaryKeys struct {
+	IsAnomaly bool `json:"-"`
+}
+
+// GetSummaryKeys implements model.ExperimentMeasurer
+func (m *Measurer) GetSummaryKeys(*model.Measurement) (any, error) {
+	sk := SummaryKeys{IsAnomaly: false}
+	return sk, nil
+}

pkg/ooniprobe/runner/nettest.go

@@ -26,6 +26,7 @@ type nettestFactory = func(args *modelx.InterpreterNettestRunArguments,
 // nettestRegistry maps nettests to their constructors.
 var nettestRegistry = map[string]nettestFactory{
 	"facebook_messenger": fbmessengerNew,
+	"riseupvpn":          riseupvpnNew,
 	"signal":             signalNew,
 	"telegram":           telegramNew,
 	"urlgetter":          urlgetterNew,

pkg/ooniprobe/runner/riseupvpn.go

@@ -0,0 +1,73 @@
+package runner
+
+//
+// riseupvpn.go implements the riseupvpn nettest
+//
+
+import (
+	"context"
+	"time"
+
+	riseupvpnnew "github.com/ooni/2023-05-richer-input/pkg/experiment/riseupvpn"
+	"github.com/ooni/2023-05-richer-input/pkg/modelx"
+	"github.com/ooni/probe-engine/pkg/experiment/riseupvpn"
+	"github.com/ooni/probe-engine/pkg/model"
+)
+
+// riseupvpnNettest is the riseupvpn nettest.
+type riseupvpnNettest struct {
+	args   *modelx.InterpreterNettestRunArguments
+	config *modelx.InterpreterConfig
+	ix     *Interpreter
+}
+
+var _ nettest = &riseupvpnNettest{}
+
+// riseupvpnNew constructs a new riseupvpn instance.
+func riseupvpnNew(args *modelx.InterpreterNettestRunArguments,
+	config *modelx.InterpreterConfig, ix *Interpreter) (nettest, error) {
+	// fill the nettest struct
+	nettest := &riseupvpnNettest{
+		args:   args,
+		config: config,
+		ix:     ix,
+	}
+
+	// return to the caller
+	return nettest, nil
+}
+
+// Run implements nettest
+func (nt *riseupvpnNettest) Run(ctx context.Context) error {
+	// make sure the location didn't change
+	if err := nt.ix.location.Refresh(); err != nil {
+		return err
+	}
+
+	// save the start time
+	t0 := time.Now()
+
+	// create a new experiment instance
+	var exp model.ExperimentMeasurer
+	if nt.args.ExperimentalFlags["dsl"] {
+		exp = riseupvpnnew.NewMeasurer(nt.args.Targets)
+	} else {
+		exp = riseupvpn.NewExperimentMeasurer(riseupvpn.Config{})
+	}
+
+	// run with the given experiment and input
+	err := runExperiment(
+		ctx,
+		nt.args.Annotations,
+		newProgressEmitterNettest(nt.ix.logger, nt.ix.view),
+		exp,
+		"", // input
+		nt.ix,
+		nt.args.ReportID,
+		t0,
+		nt.config.TestHelpers,
+	)
+
+	// handle an immediate error such as a context error
+	return err
+}

pkg/x/cmd/riseupvpn/apifetch.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"net/http"
+
+	"github.com/apex/log"
+	"github.com/ooni/probe-engine/pkg/netxlite"
+	"github.com/ooni/probe-engine/pkg/runtimex"
+)
+
+// apiMustFetchCA fetches, validates and returns the CA or PANICS.
+func apiMustFetchCA() string {
+	log.Info("- fetching the CA")
+
+	resp := runtimex.Try1(http.Get("https://black.riseup.net/ca.crt"))
+	runtimex.Assert(resp.StatusCode == 200, "unexpected HTTP response status")
+	defer resp.Body.Close()
+
+	log.Infof("HTTP response: %+v", resp)
+
+	body := string(runtimex.Try1(netxlite.ReadAllContext(context.Background(), resp.Body)))
+	log.Infof("fetched CA:\n%s\n", string(body))
+	return body
+}
+
+// apiMustFetchEIPService fetches and parses the [*apiEIPService] or PANICS.
+func apiMustFetchEIPService(caCert string) *apiEIPService {
+	log.Info("- fetching eip-service.json")
+
+	// create and fill a certificate pool
+	pool := x509.NewCertPool()
+	runtimex.Assert(pool.AppendCertsFromPEM([]byte(caCert)), "AppendCertsFromPEM failed")
+
+	// create a client using a transport using the pool
+	client := &http.Client{
+		Transport: &http.Transport{
+			TLSClientConfig: &tls.Config{
+				RootCAs: pool,
+			},
+		},
+	}
+
+	// perform the HTTP round trip
+	resp := runtimex.Try1(client.Get("https://api.black.riseup.net/3/config/eip-service.json"))
+	runtimex.Assert(resp.StatusCode == 200, "unexpected HTTP response status")
+	defer resp.Body.Close()
+
+	log.Infof("HTTP response: %+v", resp)
+
+	// read the whole body
+	body := runtimex.Try1(netxlite.ReadAllContext(context.Background(), resp.Body))
+	log.Infof("fetched eip-service.json:\n%s\n", string(body))
+
+	// parse the response body
+	var eipService apiEIPService
+	runtimex.Try0(json.Unmarshal(body, &eipService))
+	return &eipService
+}

pkg/x/cmd/riseupvpn/apimodel.go

@@ -0,0 +1,63 @@
+package main
+
+//
+// Riseupvpn (and LEAP) API data model
+//
+// Code adapted from https://github.com/ooni/probe-cli/pull/1125, which was
+// originally authored by https://github.com/cyBerta.
+//
+
+import "github.com/ooni/probe-engine/pkg/runtimex"
+
+// apiEIPService is the main JSON object returned by https://api.black.riseup.net/3/config/eip-service.json.
+type apiEIPService struct {
+	Gateways []apiGatewayV3
+}
+
+// apiGatewayV3 describes a riseupvpn gateway.
+type apiGatewayV3 struct {
+	Capabilities apiCapabilities
+	Host         string
+	IPAddress    string `json:"ip_address"`
+	Location     string `json:"location"`
+}
+
+// apiCapabilities is a list of transports a gateway supports.
+type apiCapabilities struct {
+	Transport []apiTransportV3
+}
+
+// apiTransportV3 describes a transport.
+type apiTransportV3 struct {
+	Type      string
+	Protocols []string
+	Ports     []string
+	Options   map[string]string
+}
+
+// supportsTCP returns whether the transport supports TCP.
+func (txp *apiTransportV3) supportsTCP() bool {
+	return txp.supportsTransportProtocol("tcp")
+}
+
+// supportsTransportProtocol returns whether the transport uses the given
+// transport protocol, which is one of "tcp" and "udp".
+func (txp *apiTransportV3) supportsTransportProtocol(tp string) bool {
+	runtimex.Assert(tp == "tcp" || tp == "udp", "invalid transport protocol")
+	for _, protocol := range txp.Protocols {
+		if tp == protocol {
+			return true
+		}
+	}
+	return false
+}
+
+// typeIsOneOf returns whether the transport type is one of the given types.
+func (txp *apiTransportV3) typeIsOneOf(types ...string) bool {
+	for _, t := range types {
+		if txp.Type == t {
+			return true
+		}
+	}
+	return false
+}