9694612652

Committed 27 Jun 2024 10:03AM UTC coverage: 84.193% (+0.05%) from 84.143%

Build # 9694612652

Build Type

push

github

Committed by

web-flow

Commit Message

feat(oonirun): allow true JSON richer input (#1629)

Closes https://github.com/ooni/probe/issues/2759

Run Details

47 of 47 new or added lines in 8 files covered. (100.0%)

4 existing lines in 2 files now uncovered.

27845 of 33073 relevant lines covered (84.19%)

52.37 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.2

/internal/experiment/webconnectivitylte/priority.go

package webconnectivitylte

//
// Determine which connection(s) are allowed to fetch the webpage
// by giving higher priority to the system resolver, then to the
// UDP resolver, then to the DoH resolver, then to the TH.
//
// This sorting reflects the likelyhood that we will se a blockpage
// because the system resolver is the most likely to be blocked
// (e.g., in Italy). The UDP resolver is also blocked in countries
// with more censorship (e.g., in China). The DoH resolver and
// the TH have more or less the same priority here, but we needed
// to choose one of them to have higher priority.
//
// Note that this functionality is where Web Connectivity LTE
// diverges from websteps, which will instead fetch all the available
// webpages. To adhere to the Web Connectivity model, we need to
// have a single fetch per redirect. However, by allowing all the
// resolvers plus the TH to provide us with addresses, we increase
// our chances of detecting more kinds of censorship.
//
// Also note that this implementation basically makes the
// https://github.com/ooni/probe/issues/2258 issue obsolete,
// since now we're considering all resolvers.
//
// See https://github.com/ooni/probe/issues/2276.
//

import (
        "context"
        "errors"
        "fmt"
        "net"
        "time"

        "github.com/ooni/probe-cli/v3/internal/model"
        "github.com/ooni/probe-cli/v3/internal/runtimex"
)

// errNotPermittedToFetch indicates we're not permitted to fetch a webpage
// because another goroutine have already fetched that webpage.
var errNotPermittedToFetch = errors.New("webconnectivity: not permitted to fetch")

// prioritySelector selects the connection with the highest priority.
type prioritySelector struct {
        // ch is the channel used to ask for priority
        ch chan *priorityRequest

        // logger is the logger to use
        logger model.Logger

        // m contains a map from known addresses to their flags
        m map[string]int64

        // nhttps is the number of addrs resolved using DoH
        nhttps int

        // nsystem is the number of addrs resolved using the system resolver
        nsystem int

        // nudp is the nunber of addrs resolver using UDP
        nudp int

        // tk contains the TestKeys.
        tk *TestKeys

        // zeroTime is the zero time of the current measurement
        zeroTime time.Time
}

// priorityRequest is a request to get priority for fetching the webpage
// over other concurrent connections that are doing the same.
type priorityRequest struct {
        // addr is the address we're using
        addr string

        // resp is the buffered channel where the response will appear
        resp chan bool
}

// newPrioritySelector creates a new prioritySelector instance.
func newPrioritySelector(
        ctx context.Context,
        zeroTime time.Time,
        tk *TestKeys,
        logger model.Logger,
        addrs []DNSEntry,
) *prioritySelector {
        ps := &prioritySelector{
                ch:       make(chan *priorityRequest),
                logger:   logger,
                m:        map[string]int64{},
                nhttps:   0,
                nsystem:  0,
                nudp:     0,
                tk:       tk,
                zeroTime: zeroTime,
        }
        ps.log("create with %+v", addrs)
        for _, addr := range addrs {
                flags := addr.Flags
                ps.m[addr.Addr] = flags
                if (flags & DNSAddrFlagSystemResolver) != 0 {
                        ps.nsystem++
                }
                if (flags & DNSAddrFlagUDP) != 0 {
                        ps.nudp++
                }
                if (flags & DNSAddrFlagHTTPS) != 0 {
                        ps.nhttps++
                }
        }
        go ps.selector(ctx)
        return ps
}

// log formats and emits a ConnPriorityLogEntry
func (ps *prioritySelector) log(format string, v ...any) {
        ps.tk.AppendConnPriorityLogEntry(&ConnPriorityLogEntry{
                Msg: fmt.Sprintf(format, v...),
                T:   time.Since(ps.zeroTime).Seconds(),
        })
        ps.logger.Infof("prioritySelector: "+format, v...)
}

// permissionToFetch returns whether this ready-to-use connection
// is permitted to perform a round trip and fetch the webpage.
func (ps *prioritySelector) permissionToFetch(address string) bool {
        ipAddr, _, err := net.SplitHostPort(address)
        runtimex.PanicOnError(err, "net.SplitHostPort failed")
        r := &priorityRequest{
                addr: ipAddr,
                resp: make(chan bool, 1), // buffer to simplify selector() implementation
        }
        select {
        case <-time.After(10 * time.Millisecond):
                ps.log("conn %s: denied permission: timed out sending", address)
                return false
        case ps.ch <- r:
                select {
                case <-time.After(time.Second):
                        ps.log("conn %s: denied permission: timed out receiving", address)
                        return false
                case v := <-r.resp:
                        ps.log("conn %s: granted permission: %+v", address, v)
                        return v
                }
        }
}

// selector grants permission to the highest priority request that
// arrives within a reasonable time frame. This function runs into the
// background goroutine and terminates when [ctx] is done.
//
// This function implements https://github.com/ooni/probe/issues/2276.
func (ps *prioritySelector) selector(ctx context.Context) {
        // Implementation note: setting an arbitrary timeout here would
        // be ~an issue because we want this goroutine to be available in
        // case the only connections from which we could fetch a webpage
        // are the ones using TH addresses. However, we know the TH could
        // require a long time to complete due to timeouts caused by IP
        // addresses provided by the probe.
        //
        // See https://explorer.ooni.org/measurement/20220911T105037Z_webconnectivity_IT_30722_n1_ruzuQ219SmIO9SrT?input=http%3A%2F%2Festrenosli.org%2F
        // for a measurement where a too-short timeout prevented us from
        // attempting to fetch a webpage from TH-resolved addrs.
        //
        // See https://explorer.ooni.org/measurement/20220911T194527Z_webconnectivity_IT_30722_n1_jufRZGay0Db9Ge4v?input=http%3A%2F%2Festrenosli.org%2F
        // for a measurement where this issue was fixed.

        // await the first priority request
        var first *priorityRequest
        select {
        case <-ctx.Done():
                return
        case first = <-ps.ch:
        }

        // if this request is highest priority, grant permission
        if ps.isHighestPriority(first) {
                first.resp <- true // buffered channel
                return
        }

        // collect additional requests for up to extraTime, thus giving
        // a possibly higher priority connection time to establish
        const extraTime = 500 * time.Millisecond
        expired := time.NewTimer(extraTime)
        defer expired.Stop()
        requests := []*priorityRequest{first}
Loop:
        for {
                select {
                case <-expired.C:
                        break Loop
                case r := <-ps.ch:
                        requests = append(requests, r)
                case <-ctx.Done():
                        return
                }
        }

        // grant permission to the highest priority request
        highPrio := ps.findHighestPriority(requests)
        highPrio.resp <- true // buffered channel

        // deny permission to all the other inflight requests
        for _, r := range requests {
                if highPrio != r {
                        r.resp <- false // buffered channel
                }
        }
}

// findHighestPriority returns the highest priority request
func (ps *prioritySelector) findHighestPriority(reqs []*priorityRequest) *priorityRequest {
        runtimex.Assert(len(reqs) > 0, "findHighestPriority wants a non-empty reqs slice")
        for _, r := range reqs {
                if ps.isHighestPriority(r) {
                        return r
                }
        }
        return reqs[0]
}

// isHighestPriority returns whether this request is highest priority
func (ps *prioritySelector) isHighestPriority(r *priorityRequest) bool {
        // See https://github.com/ooni/probe/issues/2276
        flags := ps.m[r.addr]
        if ps.nsystem > 0 {
                if (flags & DNSAddrFlagSystemResolver) != 0 {
                        return true
                }
        } else if ps.nudp > 0 {
                if (flags & DNSAddrFlagUDP) != 0 {
                        return true
                }
        } else if ps.nhttps > 0 {
                if (flags & DNSAddrFlagHTTPS) != 0 {
                        return true
                }
        } else {
                // Happens when we only have addresses from the TH
                return true
        }
        return false
}

1	package webconnectivitylte
2
3	//
4	// Determine which connection(s) are allowed to fetch the webpage
5	// by giving higher priority to the system resolver, then to the
6	// UDP resolver, then to the DoH resolver, then to the TH.
7	//
8	// This sorting reflects the likelyhood that we will se a blockpage
9	// because the system resolver is the most likely to be blocked
10	// (e.g., in Italy). The UDP resolver is also blocked in countries
11	// with more censorship (e.g., in China). The DoH resolver and
12	// the TH have more or less the same priority here, but we needed
13	// to choose one of them to have higher priority.
14	//
15	// Note that this functionality is where Web Connectivity LTE
16	// diverges from websteps, which will instead fetch all the available
17	// webpages. To adhere to the Web Connectivity model, we need to
18	// have a single fetch per redirect. However, by allowing all the
19	// resolvers plus the TH to provide us with addresses, we increase
20	// our chances of detecting more kinds of censorship.
21	//
22	// Also note that this implementation basically makes the
23	// https://github.com/ooni/probe/issues/2258 issue obsolete,
24	// since now we're considering all resolvers.
25	//
26	// See https://github.com/ooni/probe/issues/2276.
27	//
28
29	import (
30	"context"
31	"errors"
32	"fmt"
33	"net"
34	"time"
35
36	"github.com/ooni/probe-cli/v3/internal/model"
37	"github.com/ooni/probe-cli/v3/internal/runtimex"
38	)
39
40	// errNotPermittedToFetch indicates we're not permitted to fetch a webpage
41	// because another goroutine have already fetched that webpage.
42	var errNotPermittedToFetch = errors.New("webconnectivity: not permitted to fetch")
43
44	// prioritySelector selects the connection with the highest priority.
45	type prioritySelector struct {
46	// ch is the channel used to ask for priority
47	ch chan *priorityRequest
48
49	// logger is the logger to use
50	logger model.Logger
51
52	// m contains a map from known addresses to their flags
53	m map[string]int64
54
55	// nhttps is the number of addrs resolved using DoH
56	nhttps int
57
58	// nsystem is the number of addrs resolved using the system resolver
59	nsystem int
60
61	// nudp is the nunber of addrs resolver using UDP
62	nudp int
63
64	// tk contains the TestKeys.
65	tk *TestKeys
66
67	// zeroTime is the zero time of the current measurement
68	zeroTime time.Time
69	}
70
71	// priorityRequest is a request to get priority for fetching the webpage
72	// over other concurrent connections that are doing the same.
73	type priorityRequest struct {
74	// addr is the address we're using
75	addr string
76
77	// resp is the buffered channel where the response will appear
78	resp chan bool
79	}
80
81	// newPrioritySelector creates a new prioritySelector instance.
82	func newPrioritySelector(
83	ctx context.Context,
84	zeroTime time.Time,
85	tk *TestKeys,
86	logger model.Logger,
87	addrs []DNSEntry,
88	) *prioritySelector {	54✔
89	ps := &prioritySelector{	54✔
90	ch: make(chan *priorityRequest),	54✔
91	logger: logger,	54✔
92	m: map[string]int64{},	54✔
93	nhttps: 0,	54✔
94	nsystem: 0,	54✔
95	nudp: 0,	54✔
96	tk: tk,	54✔
97	zeroTime: zeroTime,	54✔
98	}	54✔
99	ps.log("create with %+v", addrs)	54✔
100	for _, addr := range addrs {	114✔
101	flags := addr.Flags	60✔
102	ps.m[addr.Addr] = flags	60✔
103	if (flags & DNSAddrFlagSystemResolver) != 0 {	109✔
104	ps.nsystem++	49✔
105	}	49✔
106	if (flags & DNSAddrFlagUDP) != 0 {	111✔
107	ps.nudp++	51✔
108	}	51✔
109	if (flags & DNSAddrFlagHTTPS) != 0 {	99✔
110	ps.nhttps++	39✔
111	}	39✔
112	}
113	go ps.selector(ctx)	54✔
114	return ps	54✔
115	}
116
117	// log formats and emits a ConnPriorityLogEntry
118	func (ps *prioritySelector) log(format string, v ...any) {	94✔
119	ps.tk.AppendConnPriorityLogEntry(&ConnPriorityLogEntry{	94✔
120	Msg: fmt.Sprintf(format, v...),	94✔
121	T: time.Since(ps.zeroTime).Seconds(),	94✔
122	})	94✔
123	ps.logger.Infof("prioritySelector: "+format, v...)	94✔
124	}	94✔
125
126	// permissionToFetch returns whether this ready-to-use connection
127	// is permitted to perform a round trip and fetch the webpage.
128	func (ps *prioritySelector) permissionToFetch(address string) bool {	40✔
129	ipAddr, _, err := net.SplitHostPort(address)	40✔
130	runtimex.PanicOnError(err, "net.SplitHostPort failed")	40✔
131	r := &priorityRequest{	40✔
132	addr: ipAddr,	40✔
133	resp: make(chan bool, 1), // buffer to simplify selector() implementation	40✔
134	}	40✔
135	select {	40✔
UNCOV 136	case <-time.After(10 * time.Millisecond):	×
UNCOV 137	ps.log("conn %s: denied permission: timed out sending", address)	×
UNCOV 138	return false	×
139	case ps.ch <- r:	40✔
140	select {	40✔
141	case <-time.After(time.Second):	×
142	ps.log("conn %s: denied permission: timed out receiving", address)	×
143	return false	×
144	case v := <-r.resp:	40✔
145	ps.log("conn %s: granted permission: %+v", address, v)	40✔
146	return v	40✔
147	}
148	}
149	}
150
151	// selector grants permission to the highest priority request that
152	// arrives within a reasonable time frame. This function runs into the
153	// background goroutine and terminates when [ctx] is done.
154	//
155	// This function implements https://github.com/ooni/probe/issues/2276.
156	func (ps *prioritySelector) selector(ctx context.Context) {	54✔
157	// Implementation note: setting an arbitrary timeout here would	54✔
158	// be ~an issue because we want this goroutine to be available in	54✔
159	// case the only connections from which we could fetch a webpage	54✔
160	// are the ones using TH addresses. However, we know the TH could	54✔
161	// require a long time to complete due to timeouts caused by IP	54✔
162	// addresses provided by the probe.	54✔
163	//	54✔
164	// See https://explorer.ooni.org/measurement/20220911T105037Z_webconnectivity_IT_30722_n1_ruzuQ219SmIO9SrT?input=http%3A%2F%2Festrenosli.org%2F	54✔
165	// for a measurement where a too-short timeout prevented us from	54✔
166	// attempting to fetch a webpage from TH-resolved addrs.	54✔
167	//	54✔
168	// See https://explorer.ooni.org/measurement/20220911T194527Z_webconnectivity_IT_30722_n1_jufRZGay0Db9Ge4v?input=http%3A%2F%2Festrenosli.org%2F	54✔
169	// for a measurement where this issue was fixed.	54✔
170		54✔
171	// await the first priority request	54✔
172	var first *priorityRequest	54✔
173	select {	54✔
174	case <-ctx.Done():	17✔
175	return	17✔
176	case first = <-ps.ch:	37✔
177	}
178
179	// if this request is highest priority, grant permission
180	if ps.isHighestPriority(first) {	66✔
181	first.resp <- true // buffered channel	29✔
182	return	29✔
183	}	29✔
184
185	// collect additional requests for up to extraTime, thus giving
186	// a possibly higher priority connection time to establish
187	const extraTime = 500 * time.Millisecond	8✔
188	expired := time.NewTimer(extraTime)	8✔
189	defer expired.Stop()	8✔
190	requests := []*priorityRequest{first}	8✔
191	Loop:	8✔
192	for {	19✔
193	select {	11✔
194	case <-expired.C:	8✔
195	break Loop	8✔
196	case r := <-ps.ch:	3✔
197	requests = append(requests, r)	3✔
198	case <-ctx.Done():	×
199	return	×
200	}
201	}
202
203	// grant permission to the highest priority request
204	highPrio := ps.findHighestPriority(requests)	8✔
205	highPrio.resp <- true // buffered channel	8✔
206		8✔
207	// deny permission to all the other inflight requests	8✔
208	for _, r := range requests {	19✔
209	if highPrio != r {	14✔
210	r.resp <- false // buffered channel	3✔
211	}	3✔
212	}
213	}
214
215	// findHighestPriority returns the highest priority request
216	func (ps prioritySelector) findHighestPriority(reqs []priorityRequest) *priorityRequest {	8✔
217	runtimex.Assert(len(reqs) > 0, "findHighestPriority wants a non-empty reqs slice")	8✔
218	for _, r := range reqs {	19✔
219	if ps.isHighestPriority(r) {	14✔
220	return r	3✔
221	}	3✔
222	}
223	return reqs[0]	5✔
224	}
225
226	// isHighestPriority returns whether this request is highest priority
227	func (ps prioritySelector) isHighestPriority(r priorityRequest) bool {	48✔
228	// See https://github.com/ooni/probe/issues/2276	48✔
229	flags := ps.m[r.addr]	48✔
230	if ps.nsystem > 0 {	94✔
231	if (flags & DNSAddrFlagSystemResolver) != 0 {	76✔
232	return true	30✔
233	}	30✔
234	} else if ps.nudp > 0 {	4✔
235	if (flags & DNSAddrFlagUDP) != 0 {	4✔
236	return true	2✔
237	}	2✔
238	} else if ps.nhttps > 0 {	×
239	if (flags & DNSAddrFlagHTTPS) != 0 {	×
240	return true	×
241	}	×
242	} else {	×
243	// Happens when we only have addresses from the TH	×
244	return true	×
245	}	×
246	return false	16✔
247	}

ooni / probe-cli / 9694612652

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous