Merge pull request #446 from rod-hynes/master

Stats changes

.travis.yml

@@ -1,7 +1,7 @@
 language: go
 sudo: required
 go:
-- 1.9
+- 1.9.3
 addons:
   apt_packages:
     - libx11-dev

ConsoleClient/Dockerfile

@@ -22,7 +22,7 @@ RUN apt-get update -y && apt-get install -y --no-install-recommends \
   && rm -rf /var/lib/apt/lists/*
 
 # Install Go.
-ENV GOVERSION=go1.9 GOROOT=/usr/local/go GOPATH=/go PATH=$PATH:/usr/local/go/bin:/go/bin CGO_ENABLED=1
+ENV GOVERSION=go1.9.3 GOROOT=/usr/local/go GOPATH=/go PATH=$PATH:/usr/local/go/bin:/go/bin CGO_ENABLED=1
 
 RUN curl -L https://storage.googleapis.com/golang/$GOVERSION.linux-amd64.tar.gz -o /tmp/go.tar.gz \
    && tar -C /usr/local -xzf /tmp/go.tar.gz \

MobileLibrary/Android/Dockerfile

@@ -19,7 +19,7 @@ RUN apt-get update -y && apt-get install -y --no-install-recommends \
   && rm -rf /var/lib/apt/lists/*
 
 # Install Go.
-ENV GOVERSION=go1.9 GOROOT=/usr/local/go GOPATH=/go PATH=$PATH:/usr/local/go/bin:/go/bin CGO_ENABLED=1
+ENV GOVERSION=go1.9.3 GOROOT=/usr/local/go GOPATH=/go PATH=$PATH:/usr/local/go/bin:/go/bin CGO_ENABLED=1
 
 RUN curl -L https://storage.googleapis.com/golang/$GOVERSION.linux-amd64.tar.gz -o /tmp/go.tar.gz \
   && tar -C /usr/local -xzf /tmp/go.tar.gz \

MobileLibrary/iOS/build-psiphon-framework.sh

@@ -15,7 +15,7 @@ fi
 set -x -u -e
 
 # Modify this value as we use newer Go versions.
-GO_VERSION_REQUIRED="1.9"
+GO_VERSION_REQUIRED="1.9.3"
 
 # Reset the PATH to macOS default. This is mainly so we don't execute the wrong
 # gomobile executable.

README.md

@@ -20,7 +20,7 @@ Client Setup
 
 ### Build
 
-* Go 1.8 (or higher) is required.
+* Go 1.9 (or higher) is required.
 * This project builds and runs on recent versions of Windows, Linux, and Mac OS X.
 * Note that the `psiphon` package is imported using the absolute path `github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon`; without further local configuration, `go` will use this version of the code and not the local copy in the repository.
 * In this repository, run `go build` in `ConsoleClient` to make the `ConsoleClient` binary, a console Psiphon client application.

Server/Dockerfile-binary-builder

@@ -1,6 +1,6 @@
 FROM alpine:3.4
 
-ENV GOLANG_VERSION 1.9
+ENV GOLANG_VERSION 1.9.3
 ENV GOLANG_SRC_URL https://golang.org/dl/go$GOLANG_VERSION.src.tar.gz
 
 RUN set -ex \

Server/README.md

@@ -7,7 +7,7 @@ Functionality is based on the [legacy server stack](https://bitbucket.org/psipho
 
 ### Build
 Prerequisites:
- - Go 1.8 or later
+ - Go 1.9 or later
 
 Build Steps:
  - Get dependencies: `go get -d -v ./...`
@@ -15,7 +15,7 @@ Build Steps:
 
 #### MUSL `libc` build (for Alpine Linux on Docker)
 Prerequisites:
- - Go 1.8 or later
+ - Go 1.9 or later
  - Docker 1.10 or later
  - MUSL libc toolchain
 

psiphon/common/tun/nonblock_test.go

@@ -116,7 +116,7 @@ func TestNonblockingIO(t *testing.T) {
 				n, err = r.Read(data)
 			}
 			if n != 0 || err != io.EOF {
-				t.Fatalf("exected io.EOF failed")
+				t.Fatalf("expected io.EOF failed")
 			}
 		}
 

psiphon/common/tun/tun.go

@@ -342,6 +342,12 @@ type FlowActivityUpdater interface {
 type FlowActivityUpdaterMaker func(
 	upstreamHostname string, upstreamIPAddress net.IP) []FlowActivityUpdater
 
+// MetricsUpdater is a function which receives a checkpoint summary
+// of application bytes transferred through a packet tunnel.
+type MetricsUpdater func(
+	TCPApplicationBytesUp, TCPApplicationBytesDown,
+	UDPApplicationBytesUp, UDPApplicationBytesDown int64)
+
 // ClientConnected handles new client connections, creating or resuming
 // a session and returns with client packet handlers running.
 //
@@ -356,6 +362,13 @@ type FlowActivityUpdaterMaker func(
 // permitted. These callbacks must be efficient and safe for concurrent
 // calls.
 //
+// flowActivityUpdaterMaker is a callback invoked for each new packet
+// flow; it may create updaters to track flow activity.
+//
+// metricsUpdater is a callback invoked at metrics checkpoints (usually
+// when the client disconnects) with a summary of application bytes
+// transferred.
+//
 // It is safe to make concurrent calls to ClientConnected for distinct
 // session IDs. The caller is responsible for serializing calls with the
 // same session ID. Further, the caller must ensure, in the case of a client
@@ -368,7 +381,8 @@ func (server *Server) ClientConnected(
 	sessionID string,
 	transport io.ReadWriteCloser,
 	checkAllowedTCPPortFunc, checkAllowedUDPPortFunc AllowedPortChecker,
-	flowActivityUpdaterMaker FlowActivityUpdaterMaker) error {
+	flowActivityUpdaterMaker FlowActivityUpdaterMaker,
+	metricsUpdater MetricsUpdater) error {
 
 	// It's unusual to call both sync.WaitGroup.Add() _and_ Done() in the same
 	// goroutine. There's no other place to call Add() since ClientConnected is
@@ -427,9 +441,6 @@ func (server *Server) ClientConnected(
 			metrics:                  new(packetMetrics),
 			DNSResolverIPv4Addresses: append([]net.IP(nil), DNSResolverIPv4Addresses...),
 			DNSResolverIPv6Addresses: append([]net.IP(nil), server.config.GetDNSResolverIPv6Addresses()...),
-			checkAllowedTCPPortFunc:  checkAllowedTCPPortFunc,
-			checkAllowedUDPPortFunc:  checkAllowedUDPPortFunc,
-			flowActivityUpdaterMaker: flowActivityUpdaterMaker,
 			workers:                  new(sync.WaitGroup),
 		}
 
@@ -448,7 +459,13 @@ func (server *Server) ClientConnected(
 	// allocateIndex and resumeSession calls here, so interruptSession and
 	// related code must not assume resumeSession has been called.
 
-	server.resumeSession(clientSession, NewChannel(transport, MTU))
+	server.resumeSession(
+		clientSession,
+		NewChannel(transport, MTU),
+		checkAllowedTCPPortFunc,
+		checkAllowedUDPPortFunc,
+		flowActivityUpdaterMaker,
+		metricsUpdater)
 
 	return nil
 }
@@ -481,17 +498,50 @@ func (server *Server) getSession(sessionID string) *session {
 	return nil
 }
 
-func (server *Server) resumeSession(session *session, channel *Channel) {
+func (server *Server) resumeSession(
+	session *session,
+	channel *Channel,
+	checkAllowedTCPPortFunc, checkAllowedUDPPortFunc AllowedPortChecker,
+	flowActivityUpdaterMaker FlowActivityUpdaterMaker,
+	metricsUpdater MetricsUpdater) {
 
 	session.mutex.Lock()
 	defer session.mutex.Unlock()
 
+	// Performance/concurrency note: the downstream packet queue
+	// and various packet event callbacks may be accessed while
+	// the session is idle, via the runDeviceDownstream goroutine,
+	// which runs concurrent to resumeSession/interruptSession calls.
+	// Consequently, all accesses to these fields must be
+	// synchronized.
+	//
+	// Benchmarking indicates the atomic.LoadPointer mechanism
+	// outperforms a mutex; approx. 2 ns/op vs. 20 ns/op in the case
+	// of getCheckAllowedTCPPortFunc. Since these accesses occur
+	// multiple times per packet, atomic.LoadPointer is used and so
+	// each of these fields is an unsafe.Pointer in the session
+	// struct.
+
+	// Begin buffering downstream packets.
+
 	downstreamPacketQueueSize := DEFAULT_DOWNSTREAM_PACKET_QUEUE_SIZE
 	if server.config.DownstreamPacketQueueSize > 0 {
 		downstreamPacketQueueSize = server.config.DownstreamPacketQueueSize
 	}
 	downstreamPackets := NewPacketQueue(downstreamPacketQueueSize)
-	atomic.StorePointer(&session.downstreamPackets, unsafe.Pointer(downstreamPackets))
+
+	session.setDownstreamPackets(downstreamPackets)
+
+	// Set new access control, flow monitoring, and metrics
+	// callbacks; all associated with the new client connection.
+
+	session.setCheckAllowedTCPPortFunc(&checkAllowedTCPPortFunc)
+
+	session.setCheckAllowedUDPPortFunc(&checkAllowedUDPPortFunc)
+
+	session.setFlowActivityUpdaterMaker(&flowActivityUpdaterMaker)
+
+	session.setMetricsUpdater(&metricsUpdater)
 
 	session.channel = channel
 
@@ -540,22 +590,34 @@ func (server *Server) interruptSession(session *session) {
 		session.channel = nil
 	}
 
-	// Release the downstream packet buffer, so the associated
-	// memory is not consumed while no client is connected.
-	//
-	// Since runDeviceDownstream continues to run and will access
-	// session.downstreamPackets, an atomic pointer is used to
-	// synchronize access.
-	atomic.StorePointer(&session.downstreamPackets, unsafe.Pointer(nil))
+	metricsUpdater := session.getMetricsUpdater()
 
 	// interruptSession may be called for idle sessions, to ensure
 	// the session is in an expected state: in ClientConnected,
 	// and in server.Stop(); don't log in those cases.
 	if wasRunning {
 		session.metrics.checkpoint(
-			server.config.Logger, "packet_metrics", packetMetricsAll)
+			server.config.Logger,
+			metricsUpdater,
+			"packet_metrics",
+			packetMetricsAll)
 	}
 
+	// Release the downstream packet buffer, so the associated
+	// memory is not consumed while no client is connected.
+	//
+	// Since runDeviceDownstream continues to run and will access
+	// session.downstreamPackets, an atomic pointer is used to
+	// synchronize access.
+	session.setDownstreamPackets(nil)
+
+	session.setCheckAllowedTCPPortFunc(nil)
+
+	session.setCheckAllowedUDPPortFunc(nil)
+
+	session.setFlowActivityUpdaterMaker(nil)
+
+	session.setMetricsUpdater(nil)
 }
 
 func (server *Server) runSessionReaper() {
@@ -630,7 +692,7 @@ func (server *Server) runOrphanMetricsCheckpointer() {
 
 		// TODO: skip log if all zeros?
 		server.orphanMetrics.checkpoint(
-			server.config.Logger, "orphan_packet_metrics", packetMetricsRejected)
+			server.config.Logger, nil, "orphan_packet_metrics", packetMetricsRejected)
 		if done {
 			return
 		}
@@ -694,7 +756,7 @@ func (server *Server) runDeviceDownstream() {
 
 		session := s.(*session)
 
-		downstreamPackets := (*PacketQueue)(atomic.LoadPointer(&session.downstreamPackets))
+		downstreamPackets := session.getDownstreamPackets()
 
 		// No downstreamPackets buffer is maintained when no client is
 		// connected, so the packet is dropped.
@@ -811,8 +873,7 @@ func (server *Server) runClientDownstream(session *session) {
 
 	for {
 
-		downstreamPackets := (*PacketQueue)(atomic.LoadPointer(&session.downstreamPackets))
-
+		downstreamPackets := session.getDownstreamPackets()
 		// Note: downstreamPackets will not be nil, since this goroutine only
 		// runs while the session has a connected client.
 
@@ -991,6 +1052,12 @@ type session struct {
 	// (https://golang.org/pkg/sync/atomic/#pkg-note-BUG)
 	lastActivity             int64
 	lastFlowReapIndex        int64
+	checkAllowedTCPPortFunc  unsafe.Pointer
+	checkAllowedUDPPortFunc  unsafe.Pointer
+	flowActivityUpdaterMaker unsafe.Pointer
+	metricsUpdater           unsafe.Pointer
+	downstreamPackets        unsafe.Pointer
+
 	metrics                  *packetMetrics
 	sessionID                string
 	index                    int32
@@ -1002,10 +1069,6 @@ type session struct {
 	assignedIPv6Address      net.IP
 	setOriginalIPv6Address   int32
 	originalIPv6Address      net.IP
-	checkAllowedTCPPortFunc  AllowedPortChecker
-	checkAllowedUDPPortFunc  AllowedPortChecker
-	flowActivityUpdaterMaker FlowActivityUpdaterMaker
-	downstreamPackets        unsafe.Pointer
 	flows                    sync.Map
 	workers                  *sync.WaitGroup
 	mutex                    sync.Mutex
@@ -1014,6 +1077,102 @@ type session struct {
 	stopRunning              context.CancelFunc
 }
 
+func (session *session) touch() {
+	atomic.StoreInt64(&session.lastActivity, int64(monotime.Now()))
+}
+
+func (session *session) expired(idleExpiry time.Duration) bool {
+	lastActivity := monotime.Time(atomic.LoadInt64(&session.lastActivity))
+	return monotime.Since(lastActivity) > idleExpiry
+}
+
+func (session *session) setOriginalIPv4AddressIfNotSet(IPAddress net.IP) {
+	if !atomic.CompareAndSwapInt32(&session.setOriginalIPv4Address, 0, 1) {
+		return
+	}
+	// Make a copy of IPAddress; don't reference a slice of a reusable
+	// packet buffer, which will be overwritten.
+	session.originalIPv4Address = net.IP(append([]byte(nil), []byte(IPAddress)...))
+}
+
+func (session *session) getOriginalIPv4Address() net.IP {
+	if atomic.LoadInt32(&session.setOriginalIPv4Address) == 0 {
+		return nil
+	}
+	return session.originalIPv4Address
+}
+
+func (session *session) setOriginalIPv6AddressIfNotSet(IPAddress net.IP) {
+	if !atomic.CompareAndSwapInt32(&session.setOriginalIPv6Address, 0, 1) {
+		return
+	}
+	// Make a copy of IPAddress.
+	session.originalIPv6Address = net.IP(append([]byte(nil), []byte(IPAddress)...))
+}
+
+func (session *session) getOriginalIPv6Address() net.IP {
+	if atomic.LoadInt32(&session.setOriginalIPv6Address) == 0 {
+		return nil
+	}
+	return session.originalIPv6Address
+}
+
+func (session *session) setCheckAllowedTCPPortFunc(p *AllowedPortChecker) {
+	atomic.StorePointer(&session.checkAllowedTCPPortFunc, unsafe.Pointer(p))
+}
+
+func (session *session) getCheckAllowedTCPPortFunc() AllowedPortChecker {
+	p := (*AllowedPortChecker)(atomic.LoadPointer(&session.checkAllowedTCPPortFunc))
+	if p == nil {
+		return nil
+	}
+	return *p
+}
+
+func (session *session) setCheckAllowedUDPPortFunc(p *AllowedPortChecker) {
+	atomic.StorePointer(&session.checkAllowedUDPPortFunc, unsafe.Pointer(p))
+}
+
+func (session *session) getCheckAllowedUDPPortFunc() AllowedPortChecker {
+	p := (*AllowedPortChecker)(atomic.LoadPointer(&session.checkAllowedUDPPortFunc))
+	if p == nil {
+		return nil
+	}
+	return *p
+}
+
+func (session *session) setFlowActivityUpdaterMaker(p *FlowActivityUpdaterMaker) {
+	atomic.StorePointer(&session.flowActivityUpdaterMaker, unsafe.Pointer(p))
+}
+
+func (session *session) getFlowActivityUpdaterMaker() FlowActivityUpdaterMaker {
+	p := (*FlowActivityUpdaterMaker)(atomic.LoadPointer(&session.flowActivityUpdaterMaker))
+	if p == nil {
+		return nil
+	}
+	return *p
+}
+
+func (session *session) setMetricsUpdater(p *MetricsUpdater) {
+	atomic.StorePointer(&session.metricsUpdater, unsafe.Pointer(p))
+}
+
+func (session *session) getMetricsUpdater() MetricsUpdater {
+	p := (*MetricsUpdater)(atomic.LoadPointer(&session.metricsUpdater))
+	if p == nil {
+		return nil
+	}
+	return *p
+}
+
+func (session *session) setDownstreamPackets(p *PacketQueue) {
+	atomic.StorePointer(&session.downstreamPackets, unsafe.Pointer(p))
+}
+
+func (session *session) getDownstreamPackets() *PacketQueue {
+	return (*PacketQueue)(atomic.LoadPointer(&session.downstreamPackets))
+}
+
 // flowID identifies an IP traffic flow using the conventional
 // network 5-tuple. flowIDs track bidirectional flows.
 type flowID struct {
@@ -1068,46 +1227,6 @@ func (flowState *flowState) expired(idleExpiry time.Duration) bool {
 		(now.Sub(monotime.Time(atomic.LoadInt64(&flowState.lastDownstreamPacketTime))) > idleExpiry)
 }
 
-func (session *session) touch() {
-	atomic.StoreInt64(&session.lastActivity, int64(monotime.Now()))
-}
-
-func (session *session) expired(idleExpiry time.Duration) bool {
-	lastActivity := monotime.Time(atomic.LoadInt64(&session.lastActivity))
-	return monotime.Since(lastActivity) > idleExpiry
-}
-
-func (session *session) setOriginalIPv4AddressIfNotSet(IPAddress net.IP) {
-	if !atomic.CompareAndSwapInt32(&session.setOriginalIPv4Address, 0, 1) {
-		return
-	}
-	// Make a copy of IPAddress; don't reference a slice of a reusable
-	// packet buffer, which will be overwritten.
-	session.originalIPv4Address = net.IP(append([]byte(nil), []byte(IPAddress)...))
-}
-
-func (session *session) getOriginalIPv4Address() net.IP {
-	if atomic.LoadInt32(&session.setOriginalIPv4Address) == 0 {
-		return nil
-	}
-	return session.originalIPv4Address
-}
-
-func (session *session) setOriginalIPv6AddressIfNotSet(IPAddress net.IP) {
-	if !atomic.CompareAndSwapInt32(&session.setOriginalIPv6Address, 0, 1) {
-		return
-	}
-	// Make a copy of IPAddress.
-	session.originalIPv6Address = net.IP(append([]byte(nil), []byte(IPAddress)...))
-}
-
-func (session *session) getOriginalIPv6Address() net.IP {
-	if atomic.LoadInt32(&session.setOriginalIPv6Address) == 0 {
-		return nil
-	}
-	return session.originalIPv6Address
-}
-
 // isTrackingFlow checks if a flow is being tracked.
 func (session *session) isTrackingFlow(ID flowID) bool {
 
@@ -1171,10 +1290,17 @@ func (session *session) startTrackingFlow(
 		// hostname = common.ExtractHostnameFromTCPFlow(applicationData)
 	}
 
-	flowState := &flowState{
-		activityUpdaters: session.flowActivityUpdaterMaker(
+	var activityUpdaters []FlowActivityUpdater
+
+	flowActivityUpdaterMaker := session.getFlowActivityUpdaterMaker()
+	if flowActivityUpdaterMaker != nil {
+		activityUpdaters = flowActivityUpdaterMaker(
 			hostname,
-			net.IP(ID.upstreamIPAddress[:])),
+			net.IP(ID.upstreamIPAddress[:]))
+	}
+
+	flowState := &flowState{
+		activityUpdaters: activityUpdaters,
 	}
 
 	if direction == packetDirectionServerUpstream {
@@ -1245,10 +1371,12 @@ type packetMetrics struct {
 }
 
 type relayedPacketMetrics struct {
-	packetsUp   int64
-	packetsDown int64
-	bytesUp     int64
-	bytesDown   int64
+	packetsUp            int64
+	packetsDown          int64
+	bytesUp              int64
+	bytesDown            int64
+	applicationBytesUp   int64
+	applicationBytesDown int64
 }
 
 func (metrics *packetMetrics) rejectedPacket(
@@ -1271,9 +1399,9 @@ func (metrics *packetMetrics) relayedPacket(
 	direction packetDirection,
 	version int,
 	protocol internetProtocol,
-	packetLength int) {
+	packetLength, applicationDataLength int) {
 
-	var packetsMetric, bytesMetric *int64
+	var packetsMetric, bytesMetric, applicationBytesMetric *int64
 
 	if direction == packetDirectionServerUpstream ||
 		direction == packetDirectionClientUpstream {
@@ -1283,9 +1411,11 @@ func (metrics *packetMetrics) relayedPacket(
 			if protocol == internetProtocolTCP {
 				packetsMetric = &metrics.TCPIPv4.packetsUp
 				bytesMetric = &metrics.TCPIPv4.bytesUp
+				applicationBytesMetric = &metrics.TCPIPv4.applicationBytesUp
 			} else { // UDP
 				packetsMetric = &metrics.UDPIPv4.packetsUp
 				bytesMetric = &metrics.UDPIPv4.bytesUp
+				applicationBytesMetric = &metrics.UDPIPv4.applicationBytesUp
 			}
 
 		} else { // IPv6
@@ -1293,9 +1423,11 @@ func (metrics *packetMetrics) relayedPacket(
 			if protocol == internetProtocolTCP {
 				packetsMetric = &metrics.TCPIPv6.packetsUp
 				bytesMetric = &metrics.TCPIPv6.bytesUp
+				applicationBytesMetric = &metrics.TCPIPv6.applicationBytesUp
 			} else { // UDP
 				packetsMetric = &metrics.UDPIPv6.packetsUp
 				bytesMetric = &metrics.UDPIPv6.bytesUp
+				applicationBytesMetric = &metrics.UDPIPv6.applicationBytesUp
 			}
 		}
 
@@ -1306,9 +1438,11 @@ func (metrics *packetMetrics) relayedPacket(
 			if protocol == internetProtocolTCP {
 				packetsMetric = &metrics.TCPIPv4.packetsDown
 				bytesMetric = &metrics.TCPIPv4.bytesDown
+				applicationBytesMetric = &metrics.TCPIPv4.applicationBytesDown
 			} else { // UDP
 				packetsMetric = &metrics.UDPIPv4.packetsDown
 				bytesMetric = &metrics.UDPIPv4.bytesDown
+				applicationBytesMetric = &metrics.UDPIPv4.applicationBytesDown
 			}
 
 		} else { // IPv6
@@ -1316,19 +1450,18 @@ func (metrics *packetMetrics) relayedPacket(
 			if protocol == internetProtocolTCP {
 				packetsMetric = &metrics.TCPIPv6.packetsDown
 				bytesMetric = &metrics.TCPIPv6.bytesDown
+				applicationBytesMetric = &metrics.TCPIPv6.applicationBytesDown
 			} else { // UDP
 				packetsMetric = &metrics.UDPIPv6.packetsDown
 				bytesMetric = &metrics.UDPIPv6.bytesDown
+				applicationBytesMetric = &metrics.UDPIPv6.applicationBytesDown
 			}
 		}
 	}
 
-	// Note: packet length, and so bytes transferred, includes IP and TCP/UDP
-	// headers, not just payload data, as is counted in port forwarding. It
-	// makes sense to include this packet overhead, since we have to tunnel it.
-
 	atomic.AddInt64(packetsMetric, 1)
 	atomic.AddInt64(bytesMetric, int64(packetLength))
+	atomic.AddInt64(applicationBytesMetric, int64(applicationDataLength))
 }
 
 const (
@@ -1338,7 +1471,7 @@ const (
 )
 
 func (metrics *packetMetrics) checkpoint(
-	logger common.Logger, logName string, whichMetrics int) {
+	logger common.Logger, updater MetricsUpdater, logName string, whichMetrics int) {
 
 	// Report all metric counters in a single log message. Each
 	// counter is reset to 0 when added to the log.
@@ -1357,21 +1490,41 @@ func (metrics *packetMetrics) checkpoint(
 
 	if whichMetrics&packetMetricsRelayed != 0 {
 
+		var TCPApplicationBytesUp, TCPApplicationBytesDown,
+			UDPApplicationBytesUp, UDPApplicationBytesDown int64
+
 		relayedMetrics := []struct {
-			prefix  string
-			metrics *relayedPacketMetrics
+			prefix           string
+			metrics          *relayedPacketMetrics
+			updaterBytesUp   *int64
+			updaterBytesDown *int64
 		}{
-			{"tcp_ipv4_", &metrics.TCPIPv4},
-			{"tcp_ipv6_", &metrics.TCPIPv6},
-			{"udp_ipv4_", &metrics.UDPIPv4},
-			{"udp_ipv6_", &metrics.UDPIPv6},
+			{"tcp_ipv4_", &metrics.TCPIPv4, &TCPApplicationBytesUp, &TCPApplicationBytesDown},
+			{"tcp_ipv6_", &metrics.TCPIPv6, &TCPApplicationBytesUp, &TCPApplicationBytesDown},
+			{"udp_ipv4_", &metrics.UDPIPv4, &UDPApplicationBytesUp, &UDPApplicationBytesDown},
+			{"udp_ipv6_", &metrics.UDPIPv6, &UDPApplicationBytesUp, &UDPApplicationBytesDown},
 		}
 
 		for _, r := range relayedMetrics {
+
+			applicationBytesUp := atomic.SwapInt64(&r.metrics.applicationBytesUp, 0)
+			applicationBytesDown := atomic.SwapInt64(&r.metrics.applicationBytesDown, 0)
+
+			*r.updaterBytesUp += applicationBytesUp
+			*r.updaterBytesDown += applicationBytesDown
+
 			logFields[r.prefix+"packets_up"] = atomic.SwapInt64(&r.metrics.packetsUp, 0)
 			logFields[r.prefix+"packets_down"] = atomic.SwapInt64(&r.metrics.packetsDown, 0)
 			logFields[r.prefix+"bytes_up"] = atomic.SwapInt64(&r.metrics.bytesUp, 0)
 			logFields[r.prefix+"bytes_down"] = atomic.SwapInt64(&r.metrics.bytesDown, 0)
+			logFields[r.prefix+"application_bytes_up"] = applicationBytesUp
+			logFields[r.prefix+"application_bytes_down"] = applicationBytesDown
+		}
+
+		if updater != nil {
+			updater(
+				TCPApplicationBytesUp, TCPApplicationBytesDown,
+				UDPApplicationBytesUp, UDPApplicationBytesDown)
 		}
 	}
 
@@ -1726,7 +1879,7 @@ func (client *Client) Stop() {
 	client.workers.Wait()
 
 	client.metrics.checkpoint(
-		client.config.Logger, "packet_metrics", packetMetricsAll)
+		client.config.Logger, nil, "packet_metrics", packetMetricsAll)
 
 	client.config.Logger.WithContext().Info("stopped")
 }
@@ -2205,20 +2358,34 @@ func processPacket(
 
 		if protocol == internetProtocolTCP {
 
-			if checkPort == 0 ||
-				(isServer &&
-					!session.checkAllowedTCPPortFunc(net.IP(ID.upstreamIPAddress[:]), checkPort)) {
+			invalidPort := (checkPort == 0)
+
+			if !invalidPort && isServer {
+				checkAllowedTCPPortFunc := session.getCheckAllowedTCPPortFunc()
+				if checkAllowedTCPPortFunc == nil ||
+					!checkAllowedTCPPortFunc(net.IP(ID.upstreamIPAddress[:]), checkPort) {
+					invalidPort = true
+				}
+			}
 
+			if invalidPort {
 				metrics.rejectedPacket(direction, packetRejectTCPPort)
 				return false
 			}
 
 		} else if protocol == internetProtocolUDP {
 
-			if checkPort == 0 ||
-				(isServer &&
-					!session.checkAllowedUDPPortFunc(net.IP(ID.upstreamIPAddress[:]), checkPort)) {
+			invalidPort := (checkPort == 0)
+
+			if !invalidPort && isServer {
+				checkAllowedUDPPortFunc := session.getCheckAllowedUDPPortFunc()
+				if checkAllowedUDPPortFunc == nil ||
+					!checkAllowedUDPPortFunc(net.IP(ID.upstreamIPAddress[:]), checkPort) {
+					invalidPort = true
+				}
+			}
 
+			if invalidPort {
 				metrics.rejectedPacket(direction, packetRejectUDPPort)
 				return false
 			}
@@ -2341,7 +2508,7 @@ func processPacket(
 		}
 	}
 
-	metrics.relayedPacket(direction, int(version), protocol, len(packet))
+	metrics.relayedPacket(direction, int(version), protocol, len(packet), len(applicationData))
 
 	return true
 }

psiphon/common/tun/tun_test.go

@@ -108,12 +108,20 @@ func testTunneledTCP(t *testing.T, useIPv6 bool) {
 		t.Fatalf("startTestTCPServer failed: %s", err)
 	}
 
-	var counter bytesTransferredCounter
+	var flowCounter bytesTransferredCounter
+
 	flowActivityUpdaterMaker := func(_ string, _ net.IP) []FlowActivityUpdater {
-		return []FlowActivityUpdater{&counter}
+		return []FlowActivityUpdater{&flowCounter}
+	}
+
+	var metricsCounter bytesTransferredCounter
+
+	metricsUpdater := func(TCPApplicationBytesUp, TCPApplicationBytesDown, _, _ int64) {
+		metricsCounter.UpdateProgress(
+			TCPApplicationBytesUp, TCPApplicationBytesDown, 0)
 	}
 
-	testServer, err := startTestServer(useIPv6, MTU, flowActivityUpdaterMaker)
+	testServer, err := startTestServer(useIPv6, MTU, flowActivityUpdaterMaker, metricsUpdater)
 	if err != nil {
 		t.Fatalf("startTestServer failed: %s", err)
 	}
@@ -201,6 +209,8 @@ func testTunneledTCP(t *testing.T, useIPv6 bool) {
 				{"packets_down", TCP_RELAY_TOTAL_SIZE / int64(MTU)},
 				{"bytes_up", TCP_RELAY_TOTAL_SIZE},
 				{"bytes_down", TCP_RELAY_TOTAL_SIZE},
+				{"application_bytes_up", TCP_RELAY_TOTAL_SIZE},
+				{"application_bytes_down", TCP_RELAY_TOTAL_SIZE},
 			}
 
 			for _, expectedField := range expectedFields {
@@ -240,14 +250,25 @@ func testTunneledTCP(t *testing.T, useIPv6 bool) {
 	// Note: reported bytes transferred can exceed expected bytes
 	// transferred due to retransmission of packets.
 
-	upstreamBytesTransferred, downstreamBytesTransferred, _ := counter.Get()
 	expectedBytesTransferred := CONCURRENT_CLIENT_COUNT * TCP_RELAY_TOTAL_SIZE
+
+	upstreamBytesTransferred, downstreamBytesTransferred, _ := flowCounter.Get()
+	if upstreamBytesTransferred < expectedBytesTransferred {
+		t.Fatalf("unexpected flow upstreamBytesTransferred: %d; expected at least %d",
+			upstreamBytesTransferred, expectedBytesTransferred)
+	}
+	if downstreamBytesTransferred < expectedBytesTransferred {
+		t.Fatalf("unexpected flow downstreamBytesTransferred: %d; expected at least %d",
+			downstreamBytesTransferred, expectedBytesTransferred)
+	}
+
+	upstreamBytesTransferred, downstreamBytesTransferred, _ = metricsCounter.Get()
 	if upstreamBytesTransferred < expectedBytesTransferred {
-		t.Fatalf("unexpected upstreamBytesTransferred: %d; expected at least %d",
+		t.Fatalf("unexpected metrics upstreamBytesTransferred: %d; expected at least %d",
 			upstreamBytesTransferred, expectedBytesTransferred)
 	}
 	if downstreamBytesTransferred < expectedBytesTransferred {
-		t.Fatalf("unexpected downstreamBytesTransferred: %d; expected at least %d",
+		t.Fatalf("unexpected metrics downstreamBytesTransferred: %d; expected at least %d",
 			downstreamBytesTransferred, expectedBytesTransferred)
 	}
 
@@ -280,16 +301,20 @@ func (counter *bytesTransferredCounter) Get() (int64, int64, int64) {
 }
 
 type testServer struct {
-	logger       *testLogger
-	updaterMaker FlowActivityUpdaterMaker
-	tunServer    *Server
-	unixListener net.Listener
-	clientConns  *common.Conns
-	workers      *sync.WaitGroup
+	logger         *testLogger
+	updaterMaker   FlowActivityUpdaterMaker
+	metricsUpdater MetricsUpdater
+	tunServer      *Server
+	unixListener   net.Listener
+	clientConns    *common.Conns
+	workers        *sync.WaitGroup
 }
 
 func startTestServer(
-	useIPv6 bool, MTU int, updaterMaker FlowActivityUpdaterMaker) (*testServer, error) {
+	useIPv6 bool,
+	MTU int,
+	updaterMaker FlowActivityUpdaterMaker,
+	metricsUpdater MetricsUpdater) (*testServer, error) {
 
 	logger := newTestLogger(true)
 
@@ -319,12 +344,13 @@ func startTestServer(
 	}
 
 	server := &testServer{
-		logger:       logger,
-		updaterMaker: updaterMaker,
-		tunServer:    tunServer,
-		unixListener: unixListener,
-		clientConns:  new(common.Conns),
-		workers:      new(sync.WaitGroup),
+		logger:         logger,
+		updaterMaker:   updaterMaker,
+		metricsUpdater: metricsUpdater,
+		tunServer:      tunServer,
+		unixListener:   unixListener,
+		clientConns:    new(common.Conns),
+		workers:        new(sync.WaitGroup),
 	}
 
 	server.workers.Add(1)
@@ -367,7 +393,8 @@ func (server *testServer) run() {
 				signalConn,
 				checkAllowedPortFunc,
 				checkAllowedPortFunc,
-				server.updaterMaker)
+				server.updaterMaker,
+				server.metricsUpdater)
 
 			signalConn.Wait()
 

psiphon/dataStore.go

@@ -66,7 +66,6 @@ const (
 const (
 	DATA_STORE_LAST_CONNECTED_KEY           = "lastConnected"
 	DATA_STORE_LAST_SERVER_ENTRY_FILTER_KEY = "lastServerEntryFilter"
-	PERSISTENT_STAT_TYPE_TUNNEL             = tunnelStatsBucket
 	PERSISTENT_STAT_TYPE_REMOTE_SERVER_LIST = remoteServerListStatsBucket
 )
 
@@ -153,6 +152,24 @@ func InitDataStore(config *Config) (err error) {
 			return
 		}
 
+		// Cleanup obsolete tunnel (session) stats bucket, if one still exists
+
+		err = db.Update(func(tx *bolt.Tx) error {
+			tunnelStatsBucket := []byte("tunnelStats")
+			if tx.Bucket(tunnelStatsBucket) != nil {
+				err := tx.DeleteBucket(tunnelStatsBucket)
+				if err != nil {
+					NoticeAlert("DeleteBucket %s error: %s", tunnelStatsBucket, err)
+					// Continue, since this is not fatal
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			err = fmt.Errorf("initDataStore failed to create buckets: %s", err)
+			return
+		}
+
 		singleton.db = db
 
 		// The migrateServerEntries function requires the data store is
@@ -985,7 +1002,6 @@ var persistentStatStateReporting = []byte("1")
 
 var persistentStatTypes = []string{
 	PERSISTENT_STAT_TYPE_REMOTE_SERVER_LIST,
-	PERSISTENT_STAT_TYPE_TUNNEL,
 }
 
 // StorePersistentStat adds a new persistent stat record, which
@@ -1065,8 +1081,6 @@ func TakeOutUnreportedPersistentStats(maxCount int) (map[string][][]byte, error)
 
 		for _, statType := range persistentStatTypes {
 
-			stats[statType] = make([][]byte, 0)
-
 			bucket := tx.Bucket([]byte(statType))
 			cursor := bucket.Cursor()
 			for key, value := cursor.First(); key != nil; key, value = cursor.Next() {
@@ -1090,6 +1104,11 @@ func TakeOutUnreportedPersistentStats(maxCount int) (map[string][][]byte, error)
 					// Must make a copy as slice is only valid within transaction.
 					data := make([]byte, len(key))
 					copy(data, key)
+
+					if stats[statType] == nil {
+						stats[statType] = make([][]byte, 0)
+					}
+
 					stats[statType] = append(stats[statType], data)
 					count += 1
 				}

psiphon/notice.go

@@ -714,6 +714,12 @@ func NoticeServerTimestamp(timestamp string) {
 // NoticeActiveAuthorizationIDs reports the authorizations the server has accepted.
 // Each ID is a base64-encoded accesscontrol.Authorization.ID value.
 func NoticeActiveAuthorizationIDs(activeAuthorizationIDs []string) {
+
+	// Never emit 'null' instead of empty list
+	if activeAuthorizationIDs == nil {
+		activeAuthorizationIDs = make([]string, 0)
+	}
+
 	singletonNoticeLogger.outputNotice(
 		"ActiveAuthorizationIDs", 0,
 		"IDs", activeAuthorizationIDs)

psiphon/server/api.go

@@ -209,6 +209,11 @@ func handshakeAPIRequestHandler(
 		}
 	}
 
+	// Note: no guarantee that PsinetDatabase won't reload between database calls
+	db := support.PsinetDatabase
+
+	httpsRequestRegexes := db.GetHttpsRequestRegexes(sponsorID)
+
 	// Flag the SSH client as having completed its handshake. This
 	// may reselect traffic rules and starts allowing port forwards.
 
@@ -218,9 +223,10 @@ func handshakeAPIRequestHandler(
 	activeAuthorizationIDs, authorizedAccessTypes, err := support.TunnelServer.SetClientHandshakeState(
 		sessionID,
 		handshakeState{
-			completed:   true,
-			apiProtocol: apiProtocol,
-			apiParams:   copyBaseRequestParams(params),
+			completed:         true,
+			apiProtocol:       apiProtocol,
+			apiParams:         copyBaseRequestParams(params),
+			expectDomainBytes: len(httpsRequestRegexes) > 0,
 		},
 		authorizations)
 	if err != nil {
@@ -229,23 +235,24 @@ func handshakeAPIRequestHandler(
 
 	// The log comes _after_ SetClientHandshakeState, in case that call rejects
 	// the state change (for example, if a second handshake is performed)
+	//
+	// The handshake event is no longer shipped to log consumers, so this is
+	// simply a diagnostic log.
 
-	log.LogRawFieldsWithTimestamp(
+	log.WithContextFields(
 		getRequestLogFields(
-			"handshake",
+			"",
 			geoIPData,
 			authorizedAccessTypes,
 			params,
-			baseRequestParams))
+			baseRequestParams)).Info("handshake")
 
-	// Note: no guarantee that PsinetDatabase won't reload between database calls
-	db := support.PsinetDatabase
 	handshakeResponse := protocol.HandshakeResponse{
 		SSHSessionID:           sessionID,
 		Homepages:              db.GetRandomizedHomepages(sponsorID, geoIPData.Country, isMobile),
 		UpgradeClientVersion:   db.GetUpgradeClientVersion(clientVersion, normalizedPlatform),
 		PageViewRegexes:        make([]map[string]string, 0),
-		HttpsRequestRegexes:    db.GetHttpsRequestRegexes(sponsorID),
+		HttpsRequestRegexes:    httpsRequestRegexes,
 		EncodedServerList:      db.DiscoverServers(geoIPData.DiscoveryValue),
 		ClientRegion:           geoIPData.Country,
 		ServerTimestamp:        common.GetCurrentTimestamp(),
@@ -263,7 +270,8 @@ func handshakeAPIRequestHandler(
 var connectedRequestParams = append(
 	[]requestParamSpec{
 		{"session_id", isHexDigits, 0},
-		{"last_connected", isLastConnected, 0}},
+		{"last_connected", isLastConnected, 0},
+		{"establishment_duration", isIntString, requestParamOptional}},
 	baseRequestParams...)
 
 // connectedAPIRequestHandler implements the "connected" API request.
@@ -325,6 +333,8 @@ func statusAPIRequestHandler(
 		return nil, common.ContextError(err)
 	}
 
+	sessionID, _ := getStringRequestParam(params, "client_session_id")
+
 	statusData, err := getJSONObjectRequestParam(params, "statusData")
 	if err != nil {
 		return nil, common.ContextError(err)
@@ -337,27 +347,18 @@ func statusAPIRequestHandler(
 
 	logQueue := make([]LogFields, 0)
 
-	// Overall bytes transferred stats
+	// Domain bytes transferred stats
+	// Older clients may not submit this data
 
-	bytesTransferred, err := getInt64RequestParam(statusData, "bytes_transferred")
+	// Clients are expected to send host_bytes/domain_bytes stats only when
+	// configured to do so in the handshake reponse. Legacy clients may still
+	// report "(OTHER)" host_bytes when no regexes are set. Drop those stats.
+	domainBytesExpected, err := support.TunnelServer.ExpectClientDomainBytes(sessionID)
 	if err != nil {
 		return nil, common.ContextError(err)
 	}
 
-	bytesTransferredFields := getRequestLogFields(
-		"bytes_transferred",
-		geoIPData,
-		authorizedAccessTypes,
-		params,
-		statusRequestParams)
-
-	bytesTransferredFields["bytes"] = bytesTransferred
-	logQueue = append(logQueue, bytesTransferredFields)
-
-	// Domain bytes transferred stats
-	// Older clients may not submit this data
-
-	if statusData["host_bytes"] != nil {
+	if domainBytesExpected && statusData["host_bytes"] != nil {
 
 		hostBytes, err := getMapStringInt64RequestParam(statusData, "host_bytes")
 		if err != nil {
@@ -379,90 +380,6 @@ func statusAPIRequestHandler(
 		}
 	}
 
-	// Tunnel duration and bytes transferred stats
-	// Older clients may not submit this data
-
-	if statusData["tunnel_stats"] != nil {
-
-		tunnelStats, err := getJSONObjectArrayRequestParam(statusData, "tunnel_stats")
-		if err != nil {
-			return nil, common.ContextError(err)
-		}
-		for _, tunnelStat := range tunnelStats {
-
-			sessionFields := getRequestLogFields(
-				"session",
-				geoIPData,
-				authorizedAccessTypes,
-				params,
-				statusRequestParams)
-
-			sessionID, err := getStringRequestParam(tunnelStat, "session_id")
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			sessionFields["session_id"] = sessionID
-
-			tunnelNumber, err := getInt64RequestParam(tunnelStat, "tunnel_number")
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			sessionFields["tunnel_number"] = tunnelNumber
-
-			tunnelServerIPAddress, err := getStringRequestParam(tunnelStat, "tunnel_server_ip_address")
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			sessionFields["tunnel_server_ip_address"] = tunnelServerIPAddress
-
-			// Note: older clients won't send establishment_duration
-			if _, ok := tunnelStat["establishment_duration"]; ok {
-
-				strEstablishmentDuration, err := getStringRequestParam(tunnelStat, "establishment_duration")
-				if err != nil {
-					return nil, common.ContextError(err)
-				}
-				establishmentDuration, err := strconv.ParseInt(strEstablishmentDuration, 10, 64)
-				if err != nil {
-					return nil, common.ContextError(err)
-				}
-				// Client reports establishment_duration in nanoseconds; divide to get to milliseconds
-				sessionFields["establishment_duration"] = establishmentDuration / 1000000
-			}
-
-			serverHandshakeTimestamp, err := getStringRequestParam(tunnelStat, "server_handshake_timestamp")
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			sessionFields["server_handshake_timestamp"] = serverHandshakeTimestamp
-
-			strDuration, err := getStringRequestParam(tunnelStat, "duration")
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			duration, err := strconv.ParseInt(strDuration, 10, 64)
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			// Client reports duration in nanoseconds; divide to get to milliseconds
-			sessionFields["duration"] = duration / 1000000
-
-			totalBytesSent, err := getInt64RequestParam(tunnelStat, "total_bytes_sent")
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			sessionFields["total_bytes_sent"] = totalBytesSent
-
-			totalBytesReceived, err := getInt64RequestParam(tunnelStat, "total_bytes_received")
-			if err != nil {
-				return nil, common.ContextError(err)
-			}
-			sessionFields["total_bytes_received"] = totalBytesReceived
-
-			logQueue = append(logQueue, sessionFields)
-		}
-	}
-
 	// Remote server list download stats
 	// Older clients may not submit this data
 
@@ -714,7 +631,9 @@ func getRequestLogFields(
 
 	logFields := make(LogFields)
 
-	logFields["event_name"] = eventName
+	if eventName != "" {
+		logFields["event_name"] = eventName
+	}
 
 	// In psi_web, the space replacement was done to accommodate space
 	// delimited logging, which is no longer required; we retain the
@@ -760,7 +679,7 @@ func getRequestLogFields(
 			//   fields as one of two different values based on type;
 			//   we also omit port from host:port fields for now.
 			switch expectedParam.name {
-			case "client_version":
+			case "client_version", "establishment_duration":
 				intValue, _ := strconv.Atoi(strValue)
 				logFields[expectedParam.name] = intValue
 			case "meek_dial_address":

psiphon/server/tunnelServer.go

@@ -246,6 +246,14 @@ func (server *TunnelServer) GetClientHandshaked(
 	return server.sshServer.getClientHandshaked(sessionID)
 }
 
+// ExpectClientDomainBytes indicates whether the client was configured to report
+// domain bytes in its handshake response.
+func (server *TunnelServer) ExpectClientDomainBytes(
+	sessionID string) (bool, error) {
+
+	return server.sshServer.expectClientDomainBytes(sessionID)
+}
+
 // SetEstablishTunnels sets whether new tunnels may be established or not.
 // When not establishing, incoming connections are immediately closed.
 func (server *TunnelServer) SetEstablishTunnels(establish bool) {
@@ -718,6 +726,20 @@ func (sshServer *sshServer) getClientHandshaked(
 	return completed, exhausted, nil
 }
 
+func (sshServer *sshServer) expectClientDomainBytes(
+	sessionID string) (bool, error) {
+
+	sshServer.clientsMutex.Lock()
+	client := sshServer.clients[sessionID]
+	sshServer.clientsMutex.Unlock()
+
+	if client == nil {
+		return false, common.ContextError(errors.New("unknown session ID"))
+	}
+
+	return client.expectDomainBytes(), nil
+}
+
 func (sshServer *sshServer) stopClients() {
 
 	sshServer.clientsMutex.Lock()
@@ -875,6 +897,7 @@ type handshakeState struct {
 	apiProtocol           string
 	apiParams             requestJSONObject
 	authorizedAccessTypes []string
+	expectDomainBytes     bool
 }
 
 func newSshClient(
@@ -1453,12 +1476,25 @@ func (sshClient *sshClient) runTunnel(
 				return updaters
 			}
 
+			metricUpdater := func(
+				TCPApplicationBytesUp, TCPApplicationBytesDown,
+				UDPApplicationBytesUp, UDPApplicationBytesDown int64) {
+
+				sshClient.Lock()
+				sshClient.tcpTrafficState.bytesUp += TCPApplicationBytesUp
+				sshClient.tcpTrafficState.bytesDown += TCPApplicationBytesDown
+				sshClient.udpTrafficState.bytesUp += UDPApplicationBytesUp
+				sshClient.udpTrafficState.bytesDown += UDPApplicationBytesDown
+				sshClient.Unlock()
+			}
+
 			err = sshClient.sshServer.support.PacketTunnelServer.ClientConnected(
 				sshClient.sessionID,
 				packetTunnelChannel,
 				checkAllowedTCPPortFunc,
 				checkAllowedUDPPortFunc,
-				flowActivityUpdaterMaker)
+				flowActivityUpdaterMaker,
+				metricUpdater)
 			if err != nil {
 				log.WithContextFields(LogFields{"error": err}).Warning("start packet tunnel client failed")
 				sshClient.setPacketTunnelChannel(nil)
@@ -1603,6 +1639,7 @@ func (sshClient *sshClient) logTunnel(additionalMetrics LogFields) {
 		sshClient.handshakeState.apiParams,
 		baseRequestParams)
 
+	logFields["session_id"] = sshClient.sessionID
 	logFields["handshake_completed"] = sshClient.handshakeState.completed
 	logFields["start_time"] = sshClient.activityConn.GetStartTime()
 	logFields["duration"] = sshClient.activityConn.GetActiveDuration() / time.Millisecond
@@ -1754,8 +1791,11 @@ func (sshClient *sshClient) setHandshakeState(
 	// and is used to detect and prevent multiple malicious clients from reusing a
 	// single authorization (within the scope of this server).
 
-	var authorizationIDs []string
-	var authorizedAccessTypes []string
+	// authorizationIDs and authorizedAccessTypes are returned to the client and logged,
+	// respectively; initialize to empty lists so the protocol/logs don't need to handle
+	// 'null' values.
+	authorizationIDs := make([]string, 0)
+	authorizedAccessTypes := make([]string, 0)
 	var stopTime time.Time
 
 	for i, authorization := range authorizations {
@@ -1873,6 +1913,13 @@ func (sshClient *sshClient) getHandshaked() (bool, bool) {
 	return completed, exhausted
 }
 
+func (sshClient *sshClient) expectDomainBytes() bool {
+	sshClient.Lock()
+	defer sshClient.Unlock()
+
+	return sshClient.handshakeState.expectDomainBytes
+}
+
 // setTrafficRules resets the client's traffic rules based on the latest server config
 // and client properties. As sshClient.trafficRules may be reset by a concurrent
 // goroutine, trafficRules must only be accessed within the sshClient mutex.

psiphon/serverApi.go

@@ -254,6 +254,10 @@ func (serverContext *ServerContext) DoConnectedRequest() error {
 
 	params["last_connected"] = lastConnected
 
+	// serverContext.tunnel.establishDuration is nanoseconds; divide to get to milliseconds
+	params["establishment_duration"] =
+		fmt.Sprintf("%d", serverContext.tunnel.establishDuration/1000000)
+
 	var response []byte
 	if serverContext.psiphonHttpsClient == nil {
 
@@ -313,6 +317,12 @@ func (serverContext *ServerContext) DoStatusRequest(tunnel *Tunnel) error {
 		return common.ContextError(err)
 	}
 
+	// Skip the request when there's no payload to send.
+
+	if len(statusPayload) == 0 {
+		return nil
+	}
+
 	if serverContext.psiphonHttpsClient == nil {
 
 		rawMessage := json.RawMessage(statusPayload)
@@ -397,6 +407,8 @@ func makeStatusRequestPayload(
 	serverId string) ([]byte, *statusRequestPayloadInfo, error) {
 
 	transferStats := transferstats.TakeOutStatsForServer(serverId)
+	hostBytes := transferStats.GetStatsForStatusRequest()
+
 	persistentStats, err := TakeOutUnreportedPersistentStats(
 		PSIPHON_API_PERSISTENT_STATS_MAX_COUNT)
 	if err != nil {
@@ -405,21 +417,25 @@ func makeStatusRequestPayload(
 		persistentStats = nil
 		// Proceed with transferStats only
 	}
+
+	if len(hostBytes) == 0 && len(persistentStats) == 0 {
+		// There is no payload to send.
+		return nil, nil, nil
+	}
+
 	payloadInfo := &statusRequestPayloadInfo{
 		serverId, transferStats, persistentStats}
 
 	payload := make(map[string]interface{})
 
-	hostBytes, bytesTransferred := transferStats.GetStatsForStatusRequest()
 	payload["host_bytes"] = hostBytes
-	payload["bytes_transferred"] = bytesTransferred
 
-	// We're not recording these fields, but the server requires them.
+	// We're not recording these fields, but legacy servers require them.
+	payload["bytes_transferred"] = 0
 	payload["page_views"] = make([]string, 0)
 	payload["https_requests"] = make([]string, 0)
 
 	persistentStatPayloadNames := make(map[string]string)
-	persistentStatPayloadNames[PERSISTENT_STAT_TYPE_TUNNEL] = "tunnel_stats"
 	persistentStatPayloadNames[PERSISTENT_STAT_TYPE_REMOTE_SERVER_LIST] = "remote_server_list_stats"
 
 	for statType, stats := range persistentStats {
@@ -465,26 +481,19 @@ func confirmStatusRequestPayload(payloadInfo *statusRequestPayloadInfo) {
 	}
 }
 
-// RecordTunnelStat records a tunnel duration and bytes
-// sent and received for subsequent reporting and quality
-// analysis.
-//
-// Tunnel durations are precisely measured client-side
-// and reported in status requests. As the duration is
-// not determined until the tunnel is closed, tunnel
-// stats records are stored in the persistent datastore
-// and reported via subsequent status requests sent to any
-// Psiphon server.
+// RecordRemoteServerListStat records a completed common or OSL
+// remote server list resource download.
 //
-// Since the status request that reports a tunnel stats
-// record is not necessarily handled by the same server, the
-// tunnel stats records include the original server ID.
+// The RSL download event could occur when the client is unable
+// to immediately send a status request to a server, so these
+// records are stored in the persistent datastore and reported
+// via subsequent status requests sent to any Psiphon server.
 //
-// Other fields that may change between tunnel stats recording
-// and reporting include client geo data, propagation channel,
-// sponsor ID, client version. These are not stored in the
-// datastore (client region, in particular, since that would
-// create an on-disk record of user location).
+// Note that common event field values may change between the
+// stat recording and reporting include client geo data,
+// propagation channel, sponsor ID, client version. These are not
+// stored in the datastore (client region, in particular, since
+// that would create an on-disk record of user location).
 // TODO: the server could encrypt, with a nonce and key unknown to
 // the client, a blob containing this data; return it in the
 // handshake response; and the client could store and later report
@@ -493,63 +502,13 @@ func confirmStatusRequestPayload(payloadInfo *statusRequestPayloadInfo) {
 // Multiple "status" requests may be in flight at once (due
 // to multi-tunnel, asynchronous final status retry, and
 // aggressive status requests for pre-registered tunnels),
-// To avoid duplicate reporting, tunnel stats records are
+// To avoid duplicate reporting, persistent stats records are
 // "taken-out" by a status request and then "put back" in
 // case the request fails.
 //
-// Note: since tunnel stats records have a globally unique
-// identifier (sessionId + tunnelNumber), we could tolerate
-// duplicate reporting and filter our duplicates on the
-// server-side. Permitting duplicate reporting could increase
-// the velocity of reporting (for example, both the asynchronous
-// untunneled final status requests and the post-connected
-// immediate status requests could try to report the same tunnel
-// stats).
 // Duplicate reporting may also occur when a server receives and
 // processes a status request but the client fails to receive
 // the response.
-func RecordTunnelStat(
-	sessionId string,
-	tunnelNumber int64,
-	tunnelServerIpAddress string,
-	establishmentDuration string,
-	serverHandshakeTimestamp string,
-	tunnelDuration string,
-	totalBytesSent int64,
-	totalBytesReceived int64) error {
-
-	tunnelStat := struct {
-		SessionId                string `json:"session_id"`
-		TunnelNumber             int64  `json:"tunnel_number"`
-		TunnelServerIpAddress    string `json:"tunnel_server_ip_address"`
-		EstablishmentDuration    string `json:"establishment_duration"`
-		ServerHandshakeTimestamp string `json:"server_handshake_timestamp"`
-		Duration                 string `json:"duration"`
-		TotalBytesSent           int64  `json:"total_bytes_sent"`
-		TotalBytesReceived       int64  `json:"total_bytes_received"`
-	}{
-		sessionId,
-		tunnelNumber,
-		tunnelServerIpAddress,
-		establishmentDuration,
-		serverHandshakeTimestamp,
-		tunnelDuration,
-		totalBytesSent,
-		totalBytesReceived,
-	}
-
-	tunnelStatJson, err := json.Marshal(tunnelStat)
-	if err != nil {
-		return common.ContextError(err)
-	}
-
-	return StorePersistentStat(
-		PERSISTENT_STAT_TYPE_TUNNEL, tunnelStatJson)
-}
-
-// RecordRemoteServerListStat records a completed common or OSL
-// remote server list resource download. These stats use the same
-// persist-until-reported mechanism described in RecordTunnelStats.
 func RecordRemoteServerListStat(
 	url, etag string) error {
 

psiphon/transferstats/collector.go

@@ -49,18 +49,16 @@ type AccumulatedStats struct {
 
 // GetStatsForStatusRequest summarizes AccumulatedStats data as
 // required for the Psiphon Server API status request.
-func (stats AccumulatedStats) GetStatsForStatusRequest() (map[string]int64, int64) {
+func (stats AccumulatedStats) GetStatsForStatusRequest() map[string]int64 {
 
 	hostBytes := make(map[string]int64)
-	bytesTransferred := int64(0)
 
 	for hostname, hostStats := range stats.hostnameToStats {
 		totalBytes := hostStats.numBytesReceived + hostStats.numBytesSent
-		bytesTransferred += totalBytes
 		hostBytes[hostname] = totalBytes
 	}
 
-	return hostBytes, bytesTransferred
+	return hostBytes
 }
 
 // serverStats holds per-server stats.
@@ -92,14 +90,10 @@ type statsUpdate struct {
 // recordStats makes sure the given stats update is added to the global
 // collection. recentBytes are not adjusted when isPutBack is true,
 // as recentBytes aren't subject to TakeOut/PutBack.
-func recordStat(stat *statsUpdate, isPutBack bool) {
+func recordStat(stat *statsUpdate, isRecordingHostBytes, isPutBack bool) {
 	allStats.statsMutex.Lock()
 	defer allStats.statsMutex.Unlock()
 
-	if stat.hostname == "" {
-		stat.hostname = "(OTHER)"
-	}
-
 	storedServerStats := allStats.serverIDtoStats[stat.serverID]
 	if storedServerStats == nil {
 		storedServerStats = &serverStats{
@@ -108,14 +102,21 @@ func recordStat(stat *statsUpdate, isPutBack bool) {
 		allStats.serverIDtoStats[stat.serverID] = storedServerStats
 	}
 
-	storedHostStats := storedServerStats.accumulatedStats.hostnameToStats[stat.hostname]
-	if storedHostStats == nil {
-		storedHostStats = &hostStats{}
-		storedServerStats.accumulatedStats.hostnameToStats[stat.hostname] = storedHostStats
-	}
+	if isRecordingHostBytes {
 
-	storedHostStats.numBytesSent += stat.numBytesSent
-	storedHostStats.numBytesReceived += stat.numBytesReceived
+		if stat.hostname == "" {
+			stat.hostname = "(OTHER)"
+		}
+
+		storedHostStats := storedServerStats.accumulatedStats.hostnameToStats[stat.hostname]
+		if storedHostStats == nil {
+			storedHostStats = &hostStats{}
+			storedServerStats.accumulatedStats.hostnameToStats[stat.hostname] = storedHostStats
+		}
+
+		storedHostStats.numBytesSent += stat.numBytesSent
+		storedHostStats.numBytesReceived += stat.numBytesReceived
+	}
 
 	if !isPutBack {
 		storedServerStats.recentBytesSent += stat.numBytesSent
@@ -179,6 +180,10 @@ func PutBackStatsForServer(serverID string, accumulatedStats *AccumulatedStats)
 				numBytesSent:     hoststats.numBytesSent,
 				numBytesReceived: hoststats.numBytesReceived,
 			},
+			// We can set isRecordingHostBytes to true, regardless of whether there
+			// are any regexes, since there will be no host bytes to put back if they
+			// are not being recorded.
+			true,
 			true)
 	}
 }

psiphon/transferstats/conn.go

@@ -59,6 +59,16 @@ func NewConn(nextConn net.Conn, serverID string, regexps *Regexps) *Conn {
 	}
 }
 
+func (conn *Conn) isRecordingHostBytes() bool {
+
+	// When there are no regexes, no per-host bytes stats will be
+	// recorded, including no "(OTHER)" category. In this case, it's
+	// expected that there will be no data to send in any status
+	// request.
+
+	return conn.regexps != nil && len(*conn.regexps) > 0
+}
+
 // Write is called when requests are being written out through the tunnel to
 // the remote server.
 func (conn *Conn) Write(buffer []byte) (n int, err error) {
@@ -68,9 +78,12 @@ func (conn *Conn) Write(buffer []byte) (n int, err error) {
 	// Count stats before we check the error condition. It could happen that the
 	// buffer was partially written and then an error occurred.
 	if n > 0 {
+
 		// If this is the first request, try to determine the hostname to associate
-		// with this connection.
-		if atomic.CompareAndSwapInt32(&conn.firstWrite, 1, 0) {
+		// with this connection. Skip parsing when not recording per-host bytes, as
+		// the hostname isn't used in this case.
+
+		if conn.isRecordingHostBytes() && atomic.CompareAndSwapInt32(&conn.firstWrite, 1, 0) {
 
 			hostname, ok := getHostname(buffer)
 			if ok {
@@ -86,6 +99,7 @@ func (conn *Conn) Write(buffer []byte) (n int, err error) {
 			conn.hostname,
 			int64(n),
 			0},
+			conn.isRecordingHostBytes(),
 			false)
 	}
 
@@ -110,6 +124,7 @@ func (conn *Conn) Read(buffer []byte) (n int, err error) {
 		hostname,
 		0,
 		int64(n)},
+		conn.isRecordingHostBytes(),
 		false)
 
 	return

psiphon/tunnel.go

@@ -1312,62 +1312,6 @@ func (tunnel *Tunnel) operateTunnel(tunnelOwner TunnelOwner) {
 	// Always emit a final NoticeTotalBytesTransferred
 	NoticeTotalBytesTransferred(tunnel.serverEntry.IpAddress, totalSent, totalReceived)
 
-	// Tunnel does not have a serverContext when DisableApi is set.
-	if tunnel.serverContext != nil && !tunnel.IsDiscarded() {
-
-		// The stats for this tunnel will be reported via the next successful
-		// status request.
-
-		// Since client clocks are unreliable, we report the server's timestamp from
-		// the handshake response as the absolute tunnel start time. This time
-		// will be slightly earlier than the actual tunnel activation time, as the
-		// client has to receive and parse the response and activate the tunnel.
-
-		tunnelStartTime := tunnel.serverContext.serverHandshakeTimestamp
-
-		// For the tunnel duration calculation, we use the local clock. The start time
-		// is tunnel.establishedTime as recorded when the tunnel was established. For the
-		// end time, we do not use the current time as we may now be long past the
-		// actual termination time of the tunnel. For example, the host or device may
-		// have resumed after a long sleep (it's not clear that the monotonic clock service
-		// used to measure elapsed time will or will not stop during device sleep). Instead,
-		// we use the last data received time as the estimated tunnel end time.
-		//
-		// One potential issue with using the last received time is receiving data
-		// after an extended sleep because the device sleep occurred with data still in
-		// the OS socket read buffer. This is not expected to happen on Android, as the
-		// OS will wake a process when it has TCP data available to read. (For this reason,
-		// the actual long sleep issue is only with an idle tunnel; in this case the client
-		// is responsible for sending SSH keep alives but a device sleep will delay the
-		// golang SSH keep alive timer.)
-		//
-		// Idle tunnels will only read data when a SSH keep alive is sent. As a result,
-		// the last-received-time scheme can undercount tunnel durations by up to
-		// TUNNEL_SSH_KEEP_ALIVE_PERIOD_MAX for idle tunnels.
-
-		tunnelDuration := tunnel.conn.GetLastActivityMonotime().Sub(tunnel.establishedTime)
-
-		// tunnelDuration can be < 0 when tunnel.establishedTime is recorded after the
-		// last tunnel.conn.Read() succeeds. In that case, the last read would be the
-		// handshake response, so the tunnel had, essentially, no duration.
-		if tunnelDuration < 0 {
-			tunnelDuration = 0
-		}
-
-		err := RecordTunnelStat(
-			tunnel.serverContext.sessionId,
-			tunnel.serverContext.tunnelNumber,
-			tunnel.serverEntry.IpAddress,
-			fmt.Sprintf("%d", tunnel.establishDuration),
-			tunnelStartTime,
-			fmt.Sprintf("%d", tunnelDuration),
-			totalSent,
-			totalReceived)
-		if err != nil {
-			NoticeAlert("RecordTunnelStats failed: %s", common.ContextError(err))
-		}
-	}
-
 	if err == nil {
 		NoticeInfo("shutdown operate tunnel")