Store results/errors from individual ip/port combos (runs)

internal/engine/experiment/smtp/smtp.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/smtp"
 	"net/url"
+	"strconv"
 	"time"
 
 	"github.com/ooni/probe-cli/v3/internal/engine/experiment/urlgetter"
@@ -25,6 +26,9 @@ var (
 
 	// errInvalidScheme indicates that the scheme is invalid
 	errInvalidScheme = errors.New("scheme must be smtp(s)")
+
+	// errInvalidPort indicates that the port provided could not be parsed as an int
+	errInvalidPort = errors.New("Port number is not a valid integer")
 )
 
 const (
@@ -33,16 +37,14 @@ const (
 )
 
 // Config contains the experiment config.
-type Config struct{}
+type Config struct {
-
-type RuntimeConfig struct {
 	host       string
 	port       string
 	forced_tls bool
 	noop_count uint8
 }
 
-func config(input model.MeasurementTarget) (*RuntimeConfig, error) {
+func config(input model.MeasurementTarget) (*Config, error) {
 	if input == "" {
 		// TODO: static input data (eg. gmail/riseup..)
 		return nil, errNoInputProvided
@@ -66,11 +68,16 @@ func config(input model.MeasurementTarget) (*RuntimeConfig, error) {
 			port = "465"
 		}
 	} else {
-		// Valid port is checked by URL parsing
+		// Check that requested port is a valid integer
+		_, err := strconv.Atoi(parsed.Port())
+		if err != nil {
+			return nil, errInvalidPort
+		} else {
 			port = parsed.Port()
 		}
+	}
 
-	valid_config := RuntimeConfig{
+	valid_config := Config{
 		host:       parsed.Hostname(),
 		forced_tls: parsed.Scheme == "smtps",
 		port:       port,
@@ -81,13 +88,12 @@ func config(input model.MeasurementTarget) (*RuntimeConfig, error) {
 }
 
 // TestKeys contains the experiment results
-
 type TestKeys struct {
 	Queries []*model.ArchivalDNSLookupResult `json:"queries"`
-	Runs    map[string]*IndividualTestKeys   `json:"runs"`
+	Runs    map[string]IndividualTestKeys    `json:"runs"`
 	// Used for global failure (DNS resolution)
 	Failure string `json:"failure"`
-	// Indicates global failure or individual test failure
+	// Used to indicate global failure state
 	Failed bool `json:"failed"`
 }
 
@@ -120,9 +126,9 @@ func (m Measurer) ExperimentVersion() string {
 	return testVersion
 }
 
-// Manages sequential TCP sessions to the same hostname (over different IPs)
+// Manages sequential SMTP sessions to the same hostname (over different IPs)
-// don't use in parallel!
+// don't use in parallel because addr changed dynamically
-type TCPRunner struct {
+type SMTPRunner struct {
 	trace     *measurexlite.Trace
 	logger    model.Logger
 	ctx       context.Context
@@ -130,107 +136,35 @@ type TCPRunner struct {
 	tlsconfig *tls.Config
 	host      string
 	port      string
-	// addr is changed everytime TCPRunner.conn(addr) is called
+	// addr is changed everytime SMTPRunner.conn(addr) is called
 	addr string
 }
 
-type TCPSession struct {
+func (r *SMTPRunner) run_key() string {
-	addr     string
-	port     string
-	runner   *TCPRunner
-	tk       *IndividualTestKeys
-	tls      bool
-	raw_conn *net.Conn
-	tls_conn *net.Conn
-}
-
-func (s *TCPSession) Close() {
-	if s.tls {
-		var conn = *s.tls_conn
-		conn.Close()
-	} else {
-		var conn = *s.raw_conn
-		conn.Close()
-	}
-}
-
-func (s *TCPSession) current_conn() net.Conn {
-	if s.tls {
-		return *s.tls_conn
-	} else {
-		return *s.raw_conn
-	}
-}
-
-func (r *TCPRunner) run_key() string {
 	return net.JoinHostPort(r.addr, r.port)
 }
 
-func (r *TCPRunner) get_run() *IndividualTestKeys {
+func (r *SMTPRunner) run_error(err error) {
-	if r.tk.Runs == nil {
+	r.tk.Failed = true
-		r.tk.Runs = make(map[string]*IndividualTestKeys)
-	}
 	key := r.run_key()
-	val, exists := r.tk.Runs[key]
+	// Key is initialized in conn() no need to check here
-	if exists {
+	entry, _ := r.tk.Runs[key]
-		return val
+	entry.Failure = tracex.NewFailure(err)
-	} else {
+	r.tk.Runs[key] = entry
-		r.tk.Runs[key] = &IndividualTestKeys{}
-		return r.tk.Runs[key]
-	}
 }
 
-func (r *TCPRunner) conn(addr string, port string) (*TCPSession, bool) {
+func (r *SMTPRunner) global_error(err error) {
-	r.addr = addr
+	r.tk.Failed = true
-	run := r.get_run()
+	r.tk.Failure = *tracex.NewFailure(err)
-
-	s := new(TCPSession)
-	if !s.conn(addr, port, r, run) {
-		return nil, false
-	}
-	return s, true
 }
 
-func (r *TCPRunner) dial(addr string, port string) (net.Conn, error) {
+func (r *SMTPRunner) resolve(host string) ([]string, bool) {
-	dialer := r.trace.NewDialerWithoutResolver(r.logger)
-	conn, err := dialer.DialContext(r.ctx, "tcp", net.JoinHostPort(addr, port))
-	run := r.get_run()
-	run.TCPConnect = append(run.TCPConnect, r.trace.TCPConnects()...)
-	return conn, err
-
-}
-
-func (s *TCPSession) conn(addr string, port string, runner *TCPRunner, tk *IndividualTestKeys) bool {
-	// Initialize addr field and corresponding errors in TestKeys
-	s.addr = addr
-	s.port = port
-	s.tls = false
-	s.runner = runner
-	s.tk = tk
-
-	conn, err := runner.dial(addr, port)
-	if err != nil {
-		s.error(err)
-		return false
-	}
-	s.raw_conn = &conn
-
-	return true
-}
-
-func (s *TCPSession) error(err error) {
-	s.runner.tk.Failed = true
-	s.tk.Failure = tracex.NewFailure(err)
-	//s. = append(s.errors, tracex.NewFailure(err))
-}
-
-func (r *TCPRunner) resolve(host string) ([]string, bool) {
 	r.logger.Infof("Resolving DNS for %s", host)
 	resolver := r.trace.NewStdlibResolver(r.logger)
 	addrs, err := resolver.LookupHost(r.ctx, host)
 	r.tk.Queries = append(r.tk.Queries, r.trace.DNSLookupsFromRoundTrip()...)
 	if err != nil {
-		r.tk.Failure = *tracex.NewFailure(err)
+		r.global_error(err)
 		return []string{}, false
 	}
 	r.logger.Infof("Finished DNS for %s: %v", host, addrs)
@@ -238,70 +172,99 @@ func (r *TCPRunner) resolve(host string) ([]string, bool) {
 	return addrs, true
 }
 
-func (s *TCPSession) handshake() bool {
+func (r *SMTPRunner) get_run() IndividualTestKeys {
-	if s.tls {
+	if r.tk.Runs == nil {
-		// TLS already initialized...
+		r.tk.Runs = make(map[string]IndividualTestKeys)
-		return true
 	}
-	s.runner.logger.Infof("Starting TLS handshake with %s:%s", s.addr, s.port)
+	key := r.run_key()
-	thx := s.runner.trace.NewTLSHandshakerStdlib(s.runner.logger)
+	val, exists := r.tk.Runs[key]
-	tconn, _, err := thx.Handshake(s.runner.ctx, *s.raw_conn, s.runner.tlsconfig)
+	if exists {
-	s.tk.TLSHandshakes = append(s.tk.TLSHandshakes, s.runner.trace.FirstTLSHandshakeOrNil())
+		return val
-	if err != nil {
+	} else {
-		s.error(err)
+		return IndividualTestKeys{}
-		return false
 	}
-
-	s.tls = true
-	s.tls_conn = &tconn
-	s.runner.logger.Infof("Handshake succeeded")
-	return true
 }
 
-func (s *TCPSession) starttls(message string) bool {
+func (r *SMTPRunner) save_run(itk IndividualTestKeys) {
-	if s.tls {
+	key := r.run_key()
-		// TLS already initialized...
+	r.tk.Runs[key] = itk
-		return true
+}
+
+func (r *SMTPRunner) conn(addr string) (net.Conn, bool) {
+	// Initialize addr field and corresponding errors in TestKeys
+	r.logger.Infof("Establishing TCP to %s", addr)
+	r.addr = addr
+	run := r.get_run()
+
+	dialer := r.trace.NewDialerWithoutResolver(r.logger)
+	conn, err := dialer.DialContext(r.ctx, "tcp", net.JoinHostPort(addr, r.port))
+	run.TCPConnect = append(run.TCPConnect, r.trace.TCPConnects()...)
+	if err != nil {
+		r.run_error(err)
+		return nil, false
 	}
+	r.save_run(run)
+	return conn, true
+}
+
+func (r *SMTPRunner) handshake(conn net.Conn) (net.Conn, bool) {
+	r.logger.Infof("Starting TLS handshake with %s:%s (%s)", r.host, r.port, r.addr)
+	run := r.get_run()
+	thx := r.trace.NewTLSHandshakerStdlib(r.logger)
+	tconn, _, err := thx.Handshake(r.ctx, conn, r.tlsconfig)
+	run.TLSHandshakes = append(run.TLSHandshakes, r.trace.FirstTLSHandshakeOrNil())
+	if err != nil {
+		r.run_error(err)
+		return nil, false
+	}
+	r.save_run(run)
+	r.logger.Infof("Handshake succeeded")
+	return tconn, true
+}
+
+func (r *SMTPRunner) starttls(conn net.Conn, message string) (net.Conn, bool) {
 	if message != "" {
-		s.runner.logger.Infof("Asking for StartTLS upgrade")
+		r.logger.Infof("Asking for StartTLS upgrade")
-		s.current_conn().Write([]byte(message))
+		conn.Write([]byte(message))
 	}
-	return s.handshake()
+	tconn, success := r.handshake(conn)
+	return tconn, success
 }
 
-func (s *TCPSession) smtp(ehlo string, noop uint8) bool {
+func (r *SMTPRunner) smtp(conn net.Conn, ehlo string, noop uint8) bool {
-	// Auto-choose plaintext/TCP session
+	client, err := smtp.NewClient(conn, ehlo)
-	client, err := smtp.NewClient(s.current_conn(), ehlo)
 	if err != nil {
-		s.error(err)
+		r.run_error(err)
 		return false
 	}
 	err = client.Hello(ehlo)
 	if err != nil {
-		s.error(err)
+		r.run_error(err)
 		return false
 	}
 
 	if noop > 0 {
-		s.runner.logger.Infof("Trying to generate more no-op traffic")
+		run := r.get_run()
+		r.logger.Infof("Trying to generate more no-op traffic")
 		// TODO: noop counter per IP address
-		s.tk.NoOpCounter = 0
+		run.NoOpCounter = 0
-		for s.tk.NoOpCounter < noop {
+		for run.NoOpCounter < noop {
-			s.tk.NoOpCounter += 1
+			run.NoOpCounter += 1
-			s.runner.logger.Infof("NoOp Iteration %d", s.tk.NoOpCounter)
+			r.logger.Infof("NoOp Iteration %d", run.NoOpCounter)
 			err = client.Noop()
 			if err != nil {
-				s.error(err)
+				r.run_error(err)
 				break
 			}
 		}
 
-		if s.tk.NoOpCounter == noop {
+		r.save_run(run)
-			s.runner.logger.Infof("Successfully generated no-op traffic")
+
+		if run.NoOpCounter == noop {
+			r.logger.Infof("Successfully generated no-op traffic")
 			return true
 		} else {
-			s.runner.logger.Infof("Failed no-op traffic at iteration %d", s.tk.NoOpCounter)
+			r.logger.Infof("Failed no-op traffic at iteration %d", run.NoOpCounter)
 			return false
 		}
 	}
@@ -314,6 +277,7 @@ func (m Measurer) Run(
 	ctx context.Context, sess model.ExperimentSession,
 	measurement *model.Measurement, callbacks model.ExperimentCallbacks,
 ) error {
+
 	log := sess.Logger()
 	trace := measurexlite.NewTrace(0, measurement.MeasurementStartTimeSaved)
 
@@ -334,7 +298,7 @@ func (m Measurer) Run(
 		ServerName:         config.host,
 	}
 
-	runner := &TCPRunner{
+	runner := &SMTPRunner{
 		trace:     trace,
 		logger:    log,
 		ctx:       ctx,
@@ -342,6 +306,7 @@ func (m Measurer) Run(
 		tlsconfig: &tlsconfig,
 		host:      config.host,
 		port:      config.port,
+		addr:      "",
 	}
 
 	// First resolve DNS
@@ -351,34 +316,38 @@ func (m Measurer) Run(
 	}
 
 	for _, addr := range addrs {
-		tcp_session, success := runner.conn(addr, config.port)
+		conn, success := runner.conn(addr)
 		if !success {
-			continue
+			return nil
 		}
-		defer tcp_session.Close()
+		defer conn.Close()
 
 		if config.forced_tls {
 			// Direct TLS connection
-			if !tcp_session.handshake() {
+			tconn, success := runner.handshake(conn)
+			if !success {
 				continue
 			}
+			defer tconn.Close()
 
 			// Try EHLO + NoOps
-			if !tcp_session.smtp("localhost", config.noop_count) {
+			if !runner.smtp(tconn, "localhost", 10) {
 				continue
 			}
 		} else {
 			// StartTLS... first try plaintext EHLO
-			if !tcp_session.smtp("localhost", 0) {
+			if !runner.smtp(conn, "localhost", 0) {
 				continue
 			}
 
 			// Upgrade via StartTLS and try EHLO + NoOps
-			if !tcp_session.starttls("STARTTLS\n") {
+			tconn, success := runner.starttls(conn, "STARTTLS\n")
+			if !success {
 				continue
 			}
+			defer tconn.Close()
 
-			if !tcp_session.smtp("localhost", config.noop_count) {
+			if !runner.smtp(tconn, "localhost", 10) {
 				continue
 			}
 		}

internal/engine/experiment/smtp/smtp_test.go

@@ -1,185 +0,0 @@
-package smtp
-
-import (
-	"bufio"
-	"context"
-	"crypto/tls"
-	"errors"
-	"fmt"
-	"net"
-	"strings"
-	"testing"
-
-	"github.com/ooni/probe-cli/v3/internal/engine/mockable"
-	"github.com/ooni/probe-cli/v3/internal/model"
-)
-
-func plaintextListener() net.Listener {
-	l, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		if l, err = net.Listen("tcp6", "[::1]:0"); err != nil {
-			panic(fmt.Sprintf("httptest: failed to listen on a port: %v", err))
-		}
-	}
-	return l
-}
-
-func tlsListener(l net.Listener) net.Listener {
-	return tls.NewListener(l, &tls.Config{})
-}
-
-func listener_addr(l net.Listener) string {
-	return l.Addr().String()
-}
-
-func ValidSMTPServer(conn net.Conn) {
-	for {
-		command, err := bufio.NewReader(conn).ReadString('\n')
-		if err != nil {
-			return
-		}
-
-		if command == "" {
-		} else if command == "NOOP" {
-			conn.Write([]byte("250 2.0.0 Ok\n"))
-		} else if command == "STARTTLS" {
-			conn.Write([]byte("220 2.0.0 Ready to start TLS\n"))
-			// TODO: conn.Close does not actually close connection? or does client not detect it?
-			conn.Close()
-			return
-		} else if strings.HasPrefix(command, "EHLO") {
-			conn.Write([]byte("250 mock.example.com\n"))
-		}
-		conn.Write([]byte("\n"))
-	}
-}
-
-func TCPServer(l net.Listener) {
-	for {
-		conn, err := l.Accept()
-		if err != nil {
-			continue
-		}
-		defer conn.Close()
-		conn.Write([]byte("220 mock.example.com ESMTP (spam is not appreciated)\n"))
-		ValidSMTPServer(conn)
-	}
-}
-
-func TestMeasurer_run(t *testing.T) {
-	// runHelper is an helper function to run this set of tests.
-	runHelper := func(input string) (*model.Measurement, model.ExperimentMeasurer, error) {
-		m := NewExperimentMeasurer(Config{})
-		if m.ExperimentName() != "smtp" {
-			t.Fatal("invalid experiment name")
-		}
-		if m.ExperimentVersion() != "0.0.1" {
-			t.Fatal("invalid experiment version")
-		}
-		ctx := context.Background()
-		meas := &model.Measurement{
-			Input: model.MeasurementTarget(input),
-		}
-		sess := &mockable.Session{
-			MockableLogger: model.DiscardLogger,
-		}
-		callbacks := model.NewPrinterCallbacks(model.DiscardLogger)
-		err := m.Run(ctx, sess, meas, callbacks)
-		return meas, m, err
-	}
-
-	t.Run("with empty input", func(t *testing.T) {
-		_, _, err := runHelper("")
-		if !errors.Is(err, errNoInputProvided) {
-			t.Fatal("unexpected error", err)
-		}
-	})
-
-	t.Run("with invalid URL", func(t *testing.T) {
-		_, _, err := runHelper("\t")
-		if !errors.Is(err, errInputIsNotAnURL) {
-			t.Fatal("unexpected error", err)
-		}
-	})
-
-	t.Run("with invalid scheme", func(t *testing.T) {
-		_, _, err := runHelper("https://8.8.8.8:443/")
-		if !errors.Is(err, errInvalidScheme) {
-			t.Fatal("unexpected error", err)
-		}
-	})
-
-	t.Run("with broken TLS", func(t *testing.T) {
-		p := plaintextListener()
-		defer p.Close()
-
-		l := tlsListener(p)
-		defer l.Close()
-		addr := listener_addr(l)
-		go TCPServer(l)
-
-		meas, m, err := runHelper("smtps://" + addr)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		tk := meas.TestKeys.(*TestKeys)
-
-		for _, run := range tk.Runs {
-			for _, handshake := range run.TLSHandshakes {
-				if *handshake.Failure != "unknown_failure: remote error: tls: unrecognized name" {
-					t.Fatal("expected unrecognized_name in TLS handshake")
-				}
-			}
-
-			if run.NoOpCounter != 0 {
-				t.Fatalf("expected to not have any noops, not %d noops", run.NoOpCounter)
-			}
-		}
-
-		ask, err := m.GetSummaryKeys(meas)
-		if err != nil {
-			t.Fatal("cannot obtain summary")
-		}
-		summary := ask.(SummaryKeys)
-		if summary.IsAnomaly {
-			t.Fatal("expected no anomaly")
-		}
-	})
-
-	t.Run("with broken starttls", func(t *testing.T) {
-		l := plaintextListener()
-		defer l.Close()
-		addr := listener_addr(l)
-
-		go TCPServer(l)
-
-		meas, m, err := runHelper("smtp://" + addr)
-		if err != nil {
-			t.Fatal(err)
-		}
-
-		tk := meas.TestKeys.(*TestKeys)
-
-		for _, run := range tk.Runs {
-			for _, handshake := range run.TLSHandshakes {
-				if *handshake.Failure != "generic_timeout_error" {
-					t.Fatal("expected timeout in TLS handshake")
-				}
-			}
-
-			if run.NoOpCounter != 0 {
-				t.Fatalf("expected to not have any noops, not %d noops", run.NoOpCounter)
-			}
-		}
-
-		ask, err := m.GetSummaryKeys(meas)
-		if err != nil {
-			t.Fatal("cannot obtain summary")
-		}
-		summary := ask.(SummaryKeys)
-		if summary.IsAnomaly {
-			t.Fatal("expected no anomaly")
-		}
-	})
-}