Merge ed59dcc into 49b3482

experiment/oodataformat/oodataformat.go

@@ -0,0 +1,187 @@
+// Package oodataformat contains the OONI data format.
+package oodataformat
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"net"
+	"strconv"
+	"unicode/utf8"
+
+	"github.com/ooni/netx/model"
+)
+
+// TCPConnectStatus contains the TCP connect status.
+type TCPConnectStatus struct {
+	Failure *string `json:"failure"`
+	Success bool    `json:"success"`
+}
+
+// TCPConnectEntry contains one of the entries that are part
+// of the "tcp_connect" key of a OONI report.
+type TCPConnectEntry struct {
+	IP     string           `json:"ip"`
+	Port   int              `json:"port"`
+	Status TCPConnectStatus `json:"status"`
+}
+
+// TCPConnectList is a list of TCPConnectEntry
+type TCPConnectList []TCPConnectEntry
+
+// NewTCPConnectList creates a new TCPConnectList
+func NewTCPConnectList(events [][]model.Measurement) TCPConnectList {
+	var out TCPConnectList
+	for _, roundTripEvents := range events {
+		for _, ev := range roundTripEvents {
+			if ev.Connect != nil {
+				// We assume Go is passing us legit data structs
+				ip, sport, err := net.SplitHostPort(ev.Connect.RemoteAddress)
+				if err != nil {
+					continue
+				}
+				iport, err := strconv.Atoi(sport)
+				if err != nil {
+					continue
+				}
+				if iport < 0 || iport > 65535 {
+					continue
+				}
+				out = append(out, TCPConnectEntry{
+					IP:   ip,
+					Port: iport,
+					Status: TCPConnectStatus{
+						Failure: makeFailure(ev.Connect.Error),
+						Success: ev.Connect.Error == nil,
+					},
+				})
+			}
+		}
+	}
+	return out
+}
+
+func makeFailure(err error) (s *string) {
+	if err != nil {
+		serio := err.Error()
+		s = &serio
+	}
+	return
+}
+
+// HTTPTor contains Tor information
+type HTTPTor struct {
+	ExitIP   *string `json:"exit_ip"`
+	ExitName *string `json:"exit_name"`
+	IsTor    bool    `json:"is_tor"`
+}
+
+// HTTPBody is an HTTP body. We use this helper class to define a custom
+// JSON encoder that allows us to choose the proper representation depending
+// on whether the Value field is UTF-8 or not.
+type HTTPBody struct {
+	Value string
+}
+
+// MarshalJSON marshal the body to JSON following the OONI spec that says
+// that UTF-8 bodies are represened as string and non-UTF-8 bodies are
+// instead represented as `{"format":"base64","data":"..."}`.
+func (hb HTTPBody) MarshalJSON() ([]byte, error) {
+	if utf8.ValidString(hb.Value) {
+		return json.Marshal(hb.Value)
+	}
+	er := make(map[string]string)
+	er["format"] = "base64"
+	er["data"] = base64.StdEncoding.EncodeToString([]byte(hb.Value))
+	return json.Marshal(er)
+}
+
+// HTTPRequest contains an HTTP request
+type HTTPRequest struct {
+	Body    HTTPBody          `json:"body"`
+	Headers map[string]string `json:"headers"`
+	Method  string            `json:"method"`
+	Tor     HTTPTor           `json:"tor"`
+	URL     string            `json:"url"`
+}
+
+// HTTPResponse contains an HTTP response
+type HTTPResponse struct {
+	Body    HTTPBody          `json:"body"`
+	Code    int64             `json:"code"`
+	Headers map[string]string `json:"headers"`
+}
+
+// RequestEntry is one of the entries that are part of
+// the "requests" key of a OONI report.
+type RequestEntry struct {
+	Failure  *string      `json:"failure"`
+	Request  HTTPRequest  `json:"request"`
+	Response HTTPResponse `json:"response"`
+}
+
+// RequestList is a list of RequestEntry
+type RequestList []RequestEntry
+
+// NewRequestList returns the list for "requests"
+func NewRequestList(events [][]model.Measurement) RequestList {
+	var out RequestList
+	// within the same round-trip, so proceed backwards.
+	for idx := len(events) - 1; idx >= 0; idx-- {
+		var entry RequestEntry
+		entry.Request.Headers = make(map[string]string)
+		entry.Response.Headers = make(map[string]string)
+		for _, ev := range events[idx] {
+			// Note how dividing events by round trip simplifies
+			// deciding whether there has been an error
+			if ev.Resolve != nil && ev.Resolve.Error != nil {
+				entry.Failure = makeFailure(ev.Resolve.Error)
+			}
+			if ev.Connect != nil && ev.Connect.Error != nil {
+				entry.Failure = makeFailure(ev.Connect.Error)
+			}
+			if ev.Read != nil && ev.Read.Error != nil {
+				entry.Failure = makeFailure(ev.Read.Error)
+			}
+			if ev.Write != nil && ev.Write.Error != nil {
+				entry.Failure = makeFailure(ev.Write.Error)
+			}
+			if ev.HTTPRequestHeadersDone != nil {
+				for key, values := range ev.HTTPRequestHeadersDone.Headers {
+					for _, value := range values {
+						entry.Request.Headers[key] = value
+						break
+					}
+				}
+				entry.Request.Method = ev.HTTPRequestHeadersDone.Method
+				entry.Request.URL = ev.HTTPRequestHeadersDone.URL
+				// TODO(bassosimone): do we ever send body? We should
+				// probably have an issue for this after merging.
+			}
+			if ev.HTTPResponseHeadersDone != nil {
+				for key, values := range ev.HTTPResponseHeadersDone.Headers {
+					for _, value := range values {
+						entry.Response.Headers[key] = value
+						break
+					}
+				}
+				entry.Response.Code = ev.HTTPResponseHeadersDone.StatusCode
+			}
+			if ev.HTTPResponseBodyPart != nil {
+				// Note that it's legal Go code to return bytes _and_ an
+				// error, e.g. EOF, from an io.Reader. So, we need to process
+				// the bytes anyway and then we can check the error.
+				entry.Response.Body.Value += string(ev.HTTPResponseBodyPart.Data)
+				if ev.HTTPResponseBodyPart.Error != nil &&
+					ev.HTTPResponseBodyPart.Error != io.EOF {
+					// We may see error here if we receive a bad TLS record or
+					// bad gzip data. ReadEvent only sees what happens in the
+					// network. Here we sit on top of much more stuff.
+					entry.Failure = makeFailure(ev.HTTPResponseBodyPart.Error)
+				}
+			}
+		}
+		out = append(out, entry)
+	}
+	return out
+}

experiment/oohttp/ohttp_test.go

@@ -0,0 +1,57 @@
+package oohttp
+
+import (
+	"errors"
+	"io/ioutil"
+	"net/http"
+	"testing"
+)
+
+func TestIntegration(t *testing.T) {
+	mc := NewMeasuringClient(Config{})
+	defer mc.Close()
+	client := mc.HTTPClient()
+	req, err := http.NewRequest("GET", "http://ooni.io", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	_, err = ioutil.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	measurements := mc.PopMeasurementsByRoundTrip()
+	if len(measurements) != 2 {
+		t.Fatal("expected two round trips")
+	}
+}
+
+func TestFailure(t *testing.T) {
+	mc := NewMeasuringClient(Config{})
+	defer mc.Close()
+	mc.transport = &failingRoundTripper{}
+	client := mc.HTTPClient()
+	req, err := http.NewRequest("GET", "http://ooni.io", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := client.Do(req)
+	if err == nil {
+		t.Fatal("expected an error here")
+	}
+	if resp != nil {
+		t.Fatal("expected a nil response here")
+	}
+}
+
+type failingRoundTripper struct{}
+
+func (rt *failingRoundTripper) RoundTrip(
+	req *http.Request,
+) (*http.Response, error) {
+	return nil, errors.New("mocked error")
+}

experiment/oohttp/oohttp.go

@@ -0,0 +1,108 @@
+// Package oohttp contains OONI's HTTP client.
+package oohttp
+
+import (
+	"bytes"
+	"io/ioutil"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/ooni/netx/httpx"
+	"github.com/ooni/netx/model"
+)
+
+// TODO(bassosimone): this user-agent solution is temporary and we
+// should instead select one among many user agents. We should open
+// an issue before merging this PR to address this defect.
+
+// 11.8% as of August 24, 2019 according to https://techblog.willshouse.com/2012/01/03/most-common-user-agents/
+const userAgent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.169 Safari/537.36"
+
+type roundTripMeasurements struct {
+	measurements []model.Measurement
+	mutex        sync.Mutex
+}
+
+func (rtm *roundTripMeasurements) OnMeasurement(m model.Measurement) {
+	rtm.mutex.Lock()
+	defer rtm.mutex.Unlock()
+	rtm.measurements = append(rtm.measurements, m)
+}
+
+// MeasuringClient gets you an *http.Client configured to perform
+// ooni/netx measurements during round trips.
+type MeasuringClient struct {
+	client       *http.Client
+	close        func() error
+	measurements [][]model.Measurement
+	mutex        sync.Mutex
+	rtm          *roundTripMeasurements
+	transport    http.RoundTripper
+}
+
+// Config contains the configuration
+type Config struct {
+	// CABundlePath is the path of the CA bundle to use. If empty we
+	// will be using the system default CA bundle.
+	CABundlePath string
+}
+
+// NewMeasuringClient creates a new MeasuringClient instance.
+func NewMeasuringClient(config Config) *MeasuringClient {
+	mc := new(MeasuringClient)
+	mc.client = &http.Client{
+		Transport: mc,
+	}
+	mc.rtm = new(roundTripMeasurements)
+	transport := httpx.NewTransport(time.Now(), mc.rtm)
+	transport.SetCABundle(config.CABundlePath)
+	mc.transport = transport
+	mc.close = func() error {
+		transport.CloseIdleConnections()
+		return nil
+	}
+	return mc
+}
+
+// HTTPClient returns the *http.Client you should be using.
+func (mc *MeasuringClient) HTTPClient() *http.Client {
+	return mc.client
+}
+
+// PopMeasurementsByRoundTrip returns the ooni/netx measurements organized
+// by round trip and clears the internal measurements cache.
+func (mc *MeasuringClient) PopMeasurementsByRoundTrip() [][]model.Measurement {
+	mc.mutex.Lock()
+	defer mc.mutex.Unlock()
+	out := mc.measurements
+	mc.measurements = nil
+	return out
+}
+
+// RoundTrip performs a RoundTrip.
+func (mc *MeasuringClient) RoundTrip(req *http.Request) (*http.Response, error) {
+	// Make sure we have a browser user agent for measurements.
+	req.Header.Set("User-Agent", userAgent)
+	resp, err := mc.transport.RoundTrip(req)
+	if err != nil {
+		return nil, err
+	}
+	// Fully read the response body so to see all the round trip events
+	// and have all of them inside of the c.current buffer.
+	data, err := ioutil.ReadAll(resp.Body)
+	resp.Body.Close()
+	resp.Body = ioutil.NopCloser(bytes.NewReader(data))
+	// Move events of the current round trip into the archive so that
+	// all the events we have are organized by round trip.
+	mc.mutex.Lock()
+	defer mc.mutex.Unlock()
+	mc.measurements = append(mc.measurements, mc.rtm.measurements)
+	mc.rtm.measurements = nil
+	return resp, err
+}
+
+// Close closes the resources we may have openned
+func (mc *MeasuringClient) Close() error {
+	return mc.close()
+}

experiment/telegram/telegram.go → experiment/telegram/telegram_cgo.go

@@ -1,3 +1,5 @@
+// +build cgo
+
 // Package telegram contains the Telegram network experiment.
 package telegram
 

experiment/telegram/telegram_test.go → experiment/telegram/telegram_cgo_test.go

@@ -1,3 +1,5 @@
+// +build cgo
+
 package telegram_test
 
 import (