bbcd2e2280
This diff creates a new package under netx called tracex that contains everything we need to perform measurements using events tracing and postprocessing (which is the technique with which we implement most network experiments). The general idea here is to (1) create a unique package out of all of these packages; (2) clean up the code a bit (improve tests, docs, apply more recent code patterns); (3) move the resulting code as a toplevel package inside of internal. Once this is done, netx can be further refactored to avoid subpackages and we can search for more code to salvage/refactor. See https://github.com/ooni/probe/issues/2121
422 lines
13 KiB
Go
422 lines
13 KiB
Go
// Package tor contains the tor experiment.
|
|
//
|
|
// Spec: https://github.com/ooni/spec/blob/master/nettests/ts-023-tor.md
|
|
package tor
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/ooni/probe-cli/v3/internal/atomicx"
|
|
"github.com/ooni/probe-cli/v3/internal/engine/netx/tracex"
|
|
"github.com/ooni/probe-cli/v3/internal/measurex"
|
|
"github.com/ooni/probe-cli/v3/internal/model"
|
|
"github.com/ooni/probe-cli/v3/internal/netxlite"
|
|
"github.com/ooni/probe-cli/v3/internal/runtimex"
|
|
"github.com/ooni/probe-cli/v3/internal/scrubber"
|
|
)
|
|
|
|
const (
|
|
// parallelism is the number of parallel threads we use for this experiment
|
|
parallelism = 2
|
|
|
|
// testName is the name of this experiment
|
|
testName = "tor"
|
|
|
|
// testVersion is the version of this experiment
|
|
testVersion = "0.4.0"
|
|
)
|
|
|
|
// Config contains the experiment config.
|
|
type Config struct{}
|
|
|
|
// Summary contains a summary of what happened.
|
|
type Summary struct {
|
|
Failure *string `json:"failure"`
|
|
}
|
|
|
|
// TargetResults contains the results of measuring a target.
|
|
type TargetResults struct {
|
|
Agent string `json:"agent"`
|
|
Failure *string `json:"failure"`
|
|
NetworkEvents []*measurex.ArchivalNetworkEvent `json:"network_events"`
|
|
Queries []*measurex.ArchivalDNSLookupEvent `json:"queries"`
|
|
Requests []*measurex.ArchivalHTTPRoundTripEvent `json:"requests"`
|
|
Summary map[string]Summary `json:"summary"`
|
|
TargetAddress string `json:"target_address"`
|
|
TargetName string `json:"target_name,omitempty"`
|
|
TargetProtocol string `json:"target_protocol"`
|
|
TargetSource string `json:"target_source,omitempty"`
|
|
TCPConnect []*measurex.ArchivalTCPConnect `json:"tcp_connect"`
|
|
TLSHandshakes []*measurex.ArchivalQUICTLSHandshakeEvent `json:"tls_handshakes"`
|
|
|
|
// Only for testing. We don't care about this field otherwise. We
|
|
// cannot make this private because otherwise the IP address sanitizer
|
|
// is going to panic over a private field.
|
|
DirPortCount int `json:"-"`
|
|
}
|
|
|
|
func registerExtensions(m *model.Measurement) {
|
|
tracex.ExtHTTP.AddTo(m)
|
|
tracex.ExtNetevents.AddTo(m)
|
|
tracex.ExtDNS.AddTo(m)
|
|
tracex.ExtTCPConnect.AddTo(m)
|
|
tracex.ExtTLSHandshake.AddTo(m)
|
|
}
|
|
|
|
// fillSummary fills the Summary field used by the UI.
|
|
func (tr *TargetResults) fillSummary() {
|
|
tr.Summary = make(map[string]Summary)
|
|
if len(tr.TCPConnect) < 1 {
|
|
return
|
|
}
|
|
tr.Summary[netxlite.ConnectOperation] = Summary{
|
|
Failure: tr.TCPConnect[0].Status.Failure,
|
|
}
|
|
switch tr.TargetProtocol {
|
|
case "dir_port":
|
|
// The UI currently doesn't care about this protocol
|
|
// as long as drawing a table is concerned.
|
|
tr.DirPortCount++
|
|
case "obfs4":
|
|
// We currently only perform an OBFS4 handshake, hence
|
|
// the final Failure is the handshake result
|
|
tr.Summary["handshake"] = Summary{
|
|
Failure: tr.Failure,
|
|
}
|
|
case "or_port_dirauth", "or_port":
|
|
if len(tr.TLSHandshakes) < 1 {
|
|
return
|
|
}
|
|
tr.Summary["handshake"] = Summary{
|
|
Failure: tr.TLSHandshakes[0].Failure,
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestKeys contains tor test keys.
|
|
type TestKeys struct {
|
|
DirPortTotal int64 `json:"dir_port_total"`
|
|
DirPortAccessible int64 `json:"dir_port_accessible"`
|
|
OBFS4Total int64 `json:"obfs4_total"`
|
|
OBFS4Accessible int64 `json:"obfs4_accessible"`
|
|
ORPortDirauthTotal int64 `json:"or_port_dirauth_total"`
|
|
ORPortDirauthAccessible int64 `json:"or_port_dirauth_accessible"`
|
|
ORPortTotal int64 `json:"or_port_total"`
|
|
ORPortAccessible int64 `json:"or_port_accessible"`
|
|
Targets map[string]TargetResults `json:"targets"`
|
|
}
|
|
|
|
func (tk *TestKeys) fillToplevelKeys() {
|
|
for _, value := range tk.Targets {
|
|
switch value.TargetProtocol {
|
|
case "dir_port":
|
|
tk.DirPortTotal++
|
|
if value.Failure == nil {
|
|
tk.DirPortAccessible++
|
|
}
|
|
case "obfs4":
|
|
tk.OBFS4Total++
|
|
if value.Failure == nil {
|
|
tk.OBFS4Accessible++
|
|
}
|
|
case "or_port_dirauth":
|
|
tk.ORPortDirauthTotal++
|
|
if value.Failure == nil {
|
|
tk.ORPortDirauthAccessible++
|
|
}
|
|
case "or_port":
|
|
tk.ORPortTotal++
|
|
if value.Failure == nil {
|
|
tk.ORPortAccessible++
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Measurer performs the measurement.
|
|
type Measurer struct {
|
|
config Config
|
|
fetchTorTargets func(ctx context.Context, sess model.ExperimentSession, cc string) (map[string]model.OOAPITorTarget, error)
|
|
}
|
|
|
|
// NewMeasurer creates a new Measurer
|
|
func NewMeasurer(config Config) *Measurer {
|
|
return &Measurer{
|
|
config: config,
|
|
fetchTorTargets: func(ctx context.Context, sess model.ExperimentSession, cc string) (map[string]model.OOAPITorTarget, error) {
|
|
return sess.FetchTorTargets(ctx, cc)
|
|
},
|
|
}
|
|
}
|
|
|
|
// ExperimentName implements ExperimentMeasurer.ExperiExperimentName.
|
|
func (m *Measurer) ExperimentName() string {
|
|
return testName
|
|
}
|
|
|
|
// ExperimentVersion implements ExperimentMeasurer.ExperimentVersion.
|
|
func (m *Measurer) ExperimentVersion() string {
|
|
return testVersion
|
|
}
|
|
|
|
// Run implements ExperimentMeasurer.Run.
|
|
func (m *Measurer) Run(
|
|
ctx context.Context,
|
|
sess model.ExperimentSession,
|
|
measurement *model.Measurement,
|
|
callbacks model.ExperimentCallbacks,
|
|
) error {
|
|
targets, err := m.gimmeTargets(ctx, sess)
|
|
if err != nil {
|
|
return err // fail the measurement if we cannot get any target
|
|
}
|
|
registerExtensions(measurement)
|
|
m.measureTargets(ctx, sess, measurement, callbacks, targets)
|
|
return nil
|
|
}
|
|
|
|
func (m *Measurer) gimmeTargets(
|
|
ctx context.Context, sess model.ExperimentSession,
|
|
) (map[string]model.OOAPITorTarget, error) {
|
|
ctx, cancel := context.WithTimeout(ctx, 15*time.Second)
|
|
defer cancel()
|
|
return m.fetchTorTargets(ctx, sess, sess.ProbeCC())
|
|
}
|
|
|
|
// keytarget contains a key and the related target
|
|
type keytarget struct {
|
|
key string
|
|
target model.OOAPITorTarget
|
|
}
|
|
|
|
// private returns whether a target is private. We consider private
|
|
// every target coming from a non-empty data source.
|
|
func (kt keytarget) private() bool {
|
|
return kt.target.Source != ""
|
|
}
|
|
|
|
// maybeTargetAddress returns the target address if the target is
|
|
// not private, otherwise it returns `"[scrubbed]""`.
|
|
func (kt keytarget) maybeTargetAddress() (address string) {
|
|
address = "[scrubbed]"
|
|
if !kt.private() {
|
|
address = kt.target.Address
|
|
}
|
|
return
|
|
}
|
|
|
|
func (m *Measurer) measureTargets(
|
|
ctx context.Context,
|
|
sess model.ExperimentSession,
|
|
measurement *model.Measurement,
|
|
callbacks model.ExperimentCallbacks,
|
|
targets map[string]model.OOAPITorTarget,
|
|
) {
|
|
// run measurements in parallel
|
|
var waitgroup sync.WaitGroup
|
|
rc := newResultsCollector(sess, measurement, callbacks)
|
|
waitgroup.Add(len(targets))
|
|
workch := make(chan keytarget)
|
|
for i := 0; i < parallelism; i++ {
|
|
go func(ch <-chan keytarget, total int) {
|
|
for kt := range ch {
|
|
rc.measureSingleTarget(ctx, kt, total)
|
|
waitgroup.Done()
|
|
}
|
|
}(workch, len(targets))
|
|
}
|
|
for key, target := range targets {
|
|
workch <- keytarget{key: key, target: target}
|
|
}
|
|
close(workch)
|
|
waitgroup.Wait()
|
|
// fill the measurement entry
|
|
testkeys := &TestKeys{Targets: rc.targetresults}
|
|
testkeys.fillToplevelKeys()
|
|
measurement.TestKeys = testkeys
|
|
}
|
|
|
|
type resultsCollector struct {
|
|
callbacks model.ExperimentCallbacks
|
|
completed *atomicx.Int64
|
|
flexibleConnect func(context.Context, keytarget) (*measurex.ArchivalMeasurement, *string)
|
|
measurement *model.Measurement
|
|
mu sync.Mutex
|
|
sess model.ExperimentSession
|
|
targetresults map[string]TargetResults
|
|
}
|
|
|
|
func newResultsCollector(
|
|
sess model.ExperimentSession,
|
|
measurement *model.Measurement,
|
|
callbacks model.ExperimentCallbacks,
|
|
) *resultsCollector {
|
|
rc := &resultsCollector{
|
|
callbacks: callbacks,
|
|
completed: &atomicx.Int64{},
|
|
measurement: measurement,
|
|
sess: sess,
|
|
targetresults: make(map[string]TargetResults),
|
|
}
|
|
rc.flexibleConnect = rc.defaultFlexibleConnect
|
|
return rc
|
|
}
|
|
|
|
func maybeSanitize(input TargetResults, kt keytarget) TargetResults {
|
|
if !kt.private() {
|
|
return input
|
|
}
|
|
data, err := json.Marshal(input)
|
|
runtimex.PanicOnError(err, "json.Marshal should not fail here")
|
|
// Implementation note: here we are using a strict scrubbing policy where
|
|
// we remove all IP _endpoints_, mainly for convenience, because we already
|
|
// have a well tested implementation that does that.
|
|
data = []byte(scrubber.Scrub(string(data)))
|
|
var out TargetResults
|
|
err = json.Unmarshal(data, &out)
|
|
runtimex.PanicOnError(err, "json.Unmarshal should not fail here")
|
|
return out
|
|
}
|
|
|
|
func (rc *resultsCollector) measureSingleTarget(
|
|
ctx context.Context, kt keytarget, total int,
|
|
) {
|
|
tk, failure := rc.flexibleConnect(ctx, kt)
|
|
runtimex.PanicIfNil(tk, "measurex should guarantee non-nil here")
|
|
tr := TargetResults{
|
|
Agent: "redirect",
|
|
Failure: failure,
|
|
NetworkEvents: tk.NetworkEvents,
|
|
Queries: tk.Queries,
|
|
Requests: tk.Requests,
|
|
TCPConnect: tk.TCPConnect,
|
|
TLSHandshakes: tk.TLSHandshakes,
|
|
}
|
|
tr.fillSummary()
|
|
tr = maybeSanitize(tr, kt)
|
|
rc.mu.Lock()
|
|
tr.TargetAddress = kt.maybeTargetAddress()
|
|
tr.TargetName = kt.target.Name
|
|
tr.TargetProtocol = kt.target.Protocol
|
|
tr.TargetSource = kt.target.Source
|
|
rc.targetresults[kt.key] = tr
|
|
rc.mu.Unlock()
|
|
sofar := rc.completed.Add(1)
|
|
percentage := 0.0
|
|
if total > 0 {
|
|
percentage = float64(sofar) / float64(total)
|
|
}
|
|
rc.callbacks.OnProgress(percentage, fmt.Sprintf(
|
|
"tor: access %s/%s: %s", kt.maybeTargetAddress(), kt.target.Protocol,
|
|
failureString(failure),
|
|
))
|
|
}
|
|
|
|
func maybeScrubbingLogger(input model.Logger, kt keytarget) model.Logger {
|
|
if !kt.private() {
|
|
return input
|
|
}
|
|
return &scrubber.Logger{Logger: input}
|
|
}
|
|
|
|
// defaultFlexibleConnect is the default implementation of the
|
|
// rc.flexibleConnect testable function.
|
|
//
|
|
// Arguments:
|
|
//
|
|
// - ctx is the context for timeout/cancellation;
|
|
//
|
|
// - kt contains information about the target.
|
|
//
|
|
// Returns:
|
|
//
|
|
// - tk is the measurement, which is always non nil because
|
|
// the measurex "easy" API provides this guarantee;
|
|
//
|
|
// - failure is nil or an OONI failure string.
|
|
func (rc *resultsCollector) defaultFlexibleConnect(ctx context.Context,
|
|
kt keytarget) (tk *measurex.ArchivalMeasurement, failure *string) {
|
|
mx := measurex.NewMeasurerWithDefaultSettings()
|
|
mx.Begin = rc.measurement.MeasurementStartTimeSaved
|
|
mx.Logger = maybeScrubbingLogger(rc.sess.Logger(), kt)
|
|
switch kt.target.Protocol {
|
|
case "dir_port":
|
|
URL := url.URL{
|
|
Host: kt.target.Address,
|
|
Path: "/tor/status-vote/current/consensus.z",
|
|
Scheme: "http",
|
|
}
|
|
const snapshotsize = 1 << 8 // no need to include all in report
|
|
mx.HTTPMaxBodySnapshotSize = snapshotsize
|
|
const timeout = 15 * time.Second
|
|
return mx.EasyHTTPRoundTripGET(ctx, timeout, URL.String())
|
|
case "or_port", "or_port_dirauth":
|
|
tlsConfig := measurex.NewEasyTLSConfig().InsecureSkipVerify(true)
|
|
return mx.EasyTLSConnectAndHandshake(ctx, kt.target.Address, tlsConfig)
|
|
case "obfs4":
|
|
const timeout = 15 * time.Second
|
|
return mx.EasyOBFS4ConnectAndHandshake(
|
|
ctx, timeout, kt.target.Address, rc.sess.TempDir(),
|
|
kt.target.Params)
|
|
default:
|
|
return mx.EasyTCPConnect(ctx, kt.target.Address)
|
|
}
|
|
}
|
|
|
|
// NewExperimentMeasurer creates a new ExperimentMeasurer.
|
|
func NewExperimentMeasurer(config Config) model.ExperimentMeasurer {
|
|
return NewMeasurer(config)
|
|
}
|
|
|
|
func failureString(failure *string) (s string) {
|
|
s = "success"
|
|
if failure != nil {
|
|
s = *failure
|
|
}
|
|
return
|
|
}
|
|
|
|
// 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 {
|
|
DirPortTotal int64 `json:"dir_port_total"`
|
|
DirPortAccessible int64 `json:"dir_port_accessible"`
|
|
OBFS4Total int64 `json:"obfs4_total"`
|
|
OBFS4Accessible int64 `json:"obfs4_accessible"`
|
|
ORPortDirauthTotal int64 `json:"or_port_dirauth_total"`
|
|
ORPortDirauthAccessible int64 `json:"or_port_dirauth_accessible"`
|
|
ORPortTotal int64 `json:"or_port_total"`
|
|
ORPortAccessible int64 `json:"or_port_accessible"`
|
|
IsAnomaly bool `json:"-"`
|
|
}
|
|
|
|
// GetSummaryKeys implements model.ExperimentMeasurer.GetSummaryKeys.
|
|
func (m Measurer) GetSummaryKeys(measurement *model.Measurement) (interface{}, error) {
|
|
sk := SummaryKeys{IsAnomaly: false}
|
|
tk, ok := measurement.TestKeys.(*TestKeys)
|
|
if !ok {
|
|
return sk, errors.New("invalid test keys type")
|
|
}
|
|
sk.DirPortTotal = tk.DirPortTotal
|
|
sk.DirPortAccessible = tk.DirPortAccessible
|
|
sk.OBFS4Total = tk.OBFS4Total
|
|
sk.OBFS4Accessible = tk.OBFS4Accessible
|
|
sk.ORPortDirauthTotal = tk.ORPortDirauthTotal
|
|
sk.ORPortDirauthAccessible = tk.ORPortDirauthAccessible
|
|
sk.ORPortTotal = tk.ORPortTotal
|
|
sk.ORPortAccessible = tk.ORPortAccessible
|
|
sk.IsAnomaly = ((sk.DirPortAccessible <= 0 && sk.DirPortTotal > 0) ||
|
|
(sk.OBFS4Accessible <= 0 && sk.OBFS4Total > 0) ||
|
|
(sk.ORPortDirauthAccessible <= 0 && sk.ORPortDirauthTotal > 0) ||
|
|
(sk.ORPortAccessible <= 0 && sk.ORPortTotal > 0))
|
|
return sk, nil
|
|
}
|