reporter: initial server-side capture

Lekensteyn · Lekensteyn · commit d49a36117a1f · 2017-12-24T22:45:50.000+01:00
Rename to peekableConn and wrap it in another CaptureConn instance. Move
setting of TCP Keep-Alive since a wrapped connection is not a
net.TCPConn instance.

To be done: save TLS version and success state.
diff --git a/capture_conn.go b/capture_conn.go
@@ -45,6 +45,12 @@ func (c *CaptureConn) captureFrame(b []byte, isRead bool) {
 	*c.frames = append(*c.frames, frame)
 }
 
-func (c *CaptureConn) StopCapture() {
+// StopCapture stops recording more frames. Returns true if a non-empty capture
+// was just closed and false otherwise.
+func (c *CaptureConn) StopCapture() bool {
+	if c.frames == nil || len(*c.frames) == 0 {
+		return false
+	}
 	c.frames = nil
+	return true
 }
diff --git a/reporter/capture_conn.go b/reporter/capture_conn.go
@@ -0,0 +1 @@
+../capture_conn.go
diff --git a/reporter/conn.go b/reporter/conn.go
@@ -1,12 +1,14 @@
-// A net.Conn implementation which allows for buffering the initial read such
-// that it can be peeked into without consuming it in the actual read buffer.
+// Various net.Conn implementations to support peeking or capturing data.
 package main
 
 import (
 	"net"
 	"sync"
+	"time"
 )
 
+// A net.Conn implementation which allows for buffering the initial read such
+// that it can be peeked into without consuming it in the actual read buffer.
 type conn struct {
 	net.Conn
 	readBuffer []byte
@@ -51,3 +53,18 @@ func (c *conn) Read(b []byte) (int, error) {
 	}
 	return c.Conn.Read(b)
 }
+
+type serverCaptureConn struct {
+	*CaptureConn
+	info               *ServerCapture
+	ServerCaptureReady func(*ServerCapture)
+}
+
+func (c *serverCaptureConn) Close() error {
+	err := c.CaptureConn.Close()
+	if c.CaptureConn.StopCapture() {
+		c.info.EndTime = time.Now().UTC()
+		c.ServerCaptureReady(c.info)
+	}
+	return err
+}
diff --git a/reporter/db.go b/reporter/db.go
@@ -213,3 +213,93 @@ func (model *ClientCapture) Create(tx *sql.Tx) error {
 	)
 	return err
 }
+
+// Create a new ServerCapture model. Required field: SubtestID, Frames,
+// ClientIP, ServerIP. Fields that are updated: ID, CreatedAt.
+func (model *ServerCapture) Create(tx *sql.Tx) error {
+	if model.SubtestID == 0 {
+		return errors.New("SubtestID must be initialized!")
+	}
+	if model.Frames == nil {
+		return errors.New("Frames must be initialized")
+	}
+	if model.ClientIP == nil {
+		return errors.New("client IP must be initialized")
+	}
+	if model.ServerIP == nil {
+		return errors.New("server IP must be initialized")
+	}
+	clientIP := model.ClientIP.String()
+	serverIP := model.ServerIP.String()
+	frames, err := json.Marshal(model.Frames)
+	if err != nil {
+		return err
+	}
+	err = tx.QueryRow(`
+	INSERT INTO server_captures (
+		-- id,
+		subtest_id,
+		created_at,
+		begin_time,
+		end_time,
+		actual_tls_version,
+		frames,
+		key_log,
+		has_failed,
+		client_ip,
+		server_ip
+	) VALUES (
+		--              -- id,
+		$1,             -- subtest_id,
+		now(),          -- created_at,
+		$2,             -- begin_time,
+		$3,             -- end_time,
+		$4,             -- actual_tls_version,
+		$5,             -- frames,
+		$6,             -- key_log,
+		$7,             -- has_failed,
+		$8,             -- client_ip,
+		$9              -- server_ip
+	) RETURNING
+		id,
+		created_at
+	`,
+		//&model.ID,
+		&model.SubtestID,
+		//&model.CreatedAt,
+		&model.BeginTime,
+		&model.EndTime,
+		&model.ActualTLSVersion,
+		&frames,
+		&model.KeyLog,
+		&model.HasFailed,
+		&clientIP,
+		&serverIP,
+	).Scan(
+		&model.ID,
+		&model.CreatedAt,
+	)
+	return err
+}
+
+// QuerySubtest finds SubtestID that covers the given (testID, number) pair. No
+// result is returned if the test has already concluded (this is not an error).
+func QuerySubtest(db *sql.DB, testID string, number int, mutableTestPeriodSecs int) (int, error) {
+	var subtestID int
+	err := db.QueryRow(`
+	SELECT
+		subtests.id
+	FROM subtests
+	JOIN tests
+	ON subtests.test_id = tests.id
+	WHERE
+		tests.test_id = $1 AND
+		subtests.number = $2 AND
+		is_pending AND
+		now() - created_at < $3
+	`, testID, number, mutableTestPeriodSecs).Scan(&subtestID)
+	if err != nil && err != sql.ErrNoRows {
+		return 0, err
+	}
+	return subtestID, nil
+}
diff --git a/reporter/listener.go b/reporter/listener.go
@@ -14,8 +14,8 @@ import (
 const maxHttpsQueueSize = 1024
 
 // RequestClaimer is given a hostname and should return whether the listener
-// should claim the request and if so, whether it should record the data.
-type RequestClaimer func(host string) (claimed bool, record bool)
+// should claim the request and if a ServerCapture template if it should record.
+type RequestClaimer func(host string) (claimed bool, subtestID int)
 
 type listener struct {
 	net.Listener
@@ -25,19 +25,23 @@ type listener struct {
 
 	ClaimRequest RequestClaimer
 
+	// invoked when a server capture is ready.
+	ServerCaptureReady func(*ServerCapture)
+
 	// queue for new connections, intended for reporter or test services.
 	newc chan net.Conn
 	// used to synchronize closing newc.
 	connectionsWg sync.WaitGroup
 }
 
-func newListener(ln net.Listener, initialReadTimeout time.Duration, originAddress string, claimer RequestClaimer) *listener {
+func newListener(ln net.Listener, initialReadTimeout time.Duration, originAddress string, claimer RequestClaimer, serverCaptureReady func(*ServerCapture)) *listener {
 	newc := make(chan net.Conn, maxHttpsQueueSize)
 	return &listener{
 		Listener:           ln,
 		initialReadTimeout: initialReadTimeout,
 		originAddress:      originAddress,
 		ClaimRequest:       claimer,
+		ServerCaptureReady: serverCaptureReady,
 		newc:               newc,
 	}
 }
@@ -63,9 +67,16 @@ func (ln *listener) handleConnection(c net.Conn) {
 	startTime := time.Now()
 	c.SetReadDeadline(startTime.Add(ln.initialReadTimeout))
 
+	// let dead connections eventually go away (mimic
+	// http.ListenAndServe behavior).
+	if tc, ok := c.(*net.TCPConn); ok {
+		tc.SetKeepAlive(true)
+		tc.SetKeepAlivePeriod(3 * time.Minute)
+	}
+
 	remoteAddr := c.RemoteAddr().String()
-	wrappedConn := wrapConn(c)
-	buffer, err := wrappedConn.peek(4096)
+	peekableConn := NewPeekableConn(c)
+	buffer, err := peekableConn.peek(4096)
 	if len(buffer) == 0 {
 		log.Printf("%s - failed to read a record: %v\n", remoteAddr, err)
 		return
@@ -77,17 +88,37 @@ func (ln *listener) handleConnection(c net.Conn) {
 	// server configuration.
 	c.SetReadDeadline(time.Time{})
 
-	servedByUs, _ = ln.ClaimRequest(sni)
-	// TODO handle TCP logging
+	servedByUs, subtestID := ln.ClaimRequest(sni)
 	switch {
 	case servedByUs:
-		ln.newc <- wrappedConn
-		ln.connectionsWg.Done()
+		defer ln.connectionsWg.Done()
+		if subtestID != 0 {
+			// TODO refactor this to have the logic in one place,
+			// instead of scattered through conn.go and server.go
+			serverCapture := &ServerCapture{
+				Capture: Capture{
+					SubtestID: subtestID,
+					BeginTime: startTime.UTC(),
+					Frames:    []Frame{},
+					HasFailed: true,
+				},
+				ClientIP: net.ParseIP(parseHost(remoteAddr)),
+				ServerIP: net.ParseIP(parseHost(c.LocalAddr().String())),
+			}
+			capturedConn := &serverCaptureConn{
+				CaptureConn:        NewCaptureConn(peekableConn, &serverCapture.Frames),
+				info:               serverCapture,
+				ServerCaptureReady: ln.ServerCaptureReady,
+			}
+			ln.newc <- capturedConn
+		} else {
+			ln.newc <- peekableConn
+		}
 	case ln.originAddress == "":
 		log.Printf("%s - no upstream configured", remoteAddr)
 		c.Write(tlsRecordUnrecognizedName)
 	default:
-		if err := proxyConnection(wrappedConn, ln.originAddress); err != nil {
+		if err := proxyConnection(peekableConn, ln.originAddress); err != nil {
 			log.Printf("%s - error proxying connection: %v\n", remoteAddr, err)
 		}
 	}
@@ -130,12 +161,5 @@ func (ln *listener) Accept() (net.Conn, error) {
 		return nil, http.ErrServerClosed
 	}
 
-	// let dead connections eventually go away (mimic
-	// http.ListenAndServe behavior).
-	if tc, ok := c.(*net.TCPConn); ok {
-		tc.SetKeepAlive(true)
-		tc.SetKeepAlivePeriod(3 * time.Minute)
-	}
-
 	return c, nil
 }
diff --git a/reporter/server.go b/reporter/server.go
@@ -10,6 +10,7 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -45,19 +46,91 @@ func isTestHost(host string, config *Config) bool {
 	return strings.HasSuffix(host, config.HostSuffixIPv4) || strings.HasSuffix(host, config.HostSuffixIPv6)
 }
 
-func makeIsOurHost(config *Config) RequestClaimer {
-	return func(host string) (bool, bool) {
+func makeIsOurHost(db *sql.DB, config *Config) RequestClaimer {
+	return func(host string) (bool, int) {
 		host = strings.ToLower(host)
 		if host == config.HostReporter {
 			// pass to HTTP handler, handle API requests.
-			return true, false
+			return true, 0
 		}
 		if isTestHost(host, config) {
 			// pass to HTTP handler, handling a basic response.
 			// Logging is tentatively enabled.
-			return true, true
+			return true, prepareServerCapture(db, config, host)
 		}
-		return false, false
+		return false, 0
+	}
+}
+
+func prepareServerCapture(db *sql.DB, config *Config, host string) int {
+	testID, number := parseTestHost(config, host)
+	if testID == "" {
+		log.Printf("Host \"%s\" is not a valid test domain, ignoring", host)
+		return 0
+	}
+	subtestID, err := QuerySubtest(db, testID, number, config.MutableTestPeriodSecs)
+	if err != nil {
+		log.Printf("Failed to query subtest for \"%s\": %s", host, err)
+		return 0
+	}
+	if subtestID == 0 {
+		log.Printf("Not accepting server capture for \"%s\"", host)
+		return 0
+	}
+	return subtestID
+}
+
+// parses a host name of the form "<testID>-<number><suffix>", returning the
+// TestID and subtest number. On error, the testID is empty.
+func parseTestHost(config *Config, host string) (string, int) {
+	var prefix string
+	switch {
+	case strings.HasSuffix(host, config.HostSuffixIPv4):
+		prefix = host[:len(host)-len(config.HostSuffixIPv4)]
+	case strings.HasSuffix(host, config.HostSuffixIPv6):
+		prefix = host[:len(host)-len(config.HostSuffixIPv6)]
+	default:
+		return "", 0
+	}
+
+	// testID UUID is always 36 chars followed by "-" and number.
+	if len(prefix) < 36+2 || prefix[36] != '-' {
+		return "", 0
+	}
+	testID, numberStr := prefix[0:36], prefix[37:]
+	if !ValidateUUID(testID) {
+		return "", 0
+	}
+	number, err := strconv.Atoi(numberStr)
+	if err != nil || number <= 0 {
+		return "", 0
+	}
+
+	return testID, number
+}
+
+func newServerCaptureReady(db *sql.DB) func(*ServerCapture) {
+	return func(serverCapture *ServerCapture) {
+		tx, err := db.Begin()
+		if err != nil {
+			log.Printf("Failed to begin transaction: %s", err)
+			return
+		}
+		defer func() {
+			if tx != nil {
+				tx.Rollback()
+			}
+		}()
+
+		err = serverCapture.Create(tx)
+		if err != nil {
+			log.Printf("Failed to create server capture: %s", err)
+			return
+		}
+
+		log.Printf("Stored server capture: %d", serverCapture.ID)
+		tx.Commit()
+		tx = nil
 	}
 }
 
@@ -142,7 +215,7 @@ func main() {
 		panic(err)
 	}
 	initialReadTimeout := time.Duration(config.InitialReadTimeoutSecs) * time.Second
-	wl := newListener(l, initialReadTimeout, config.OriginAddress, makeIsOurHost(config))
+	wl := newListener(l, initialReadTimeout, config.OriginAddress, makeIsOurHost(db, config), newServerCaptureReady(db))
 	go wl.Serve()
 
 	hostRouter := &hostHandler{

Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,12 @@ func (c *CaptureConn) captureFrame(b []byte, isRead bool) {`
`45`	`45`	`c.frames = append(c.frames, frame)`
`46`	`46`	`}`
`47`	`47`
`48`		`-func (c *CaptureConn) StopCapture() {`
	`48`	`+// StopCapture stops recording more frames. Returns true if a non-empty capture`
	`49`	`+// was just closed and false otherwise.`
	`50`	`+func (c *CaptureConn) StopCapture() bool {`
	`51`	`+ if c.frames == nil \|\| len(*c.frames) == 0 {`
	`52`	`+ return false`
	`53`	`+ }`
`49`	`54`	`c.frames = nil`
	`55`	`+ return true`
`50`	`56`	`}`