Meek testing and fixes (in progress)

.travis.yml

@@ -8,7 +8,7 @@ addons:
 install:
 - go get -t -d -v ./... && go build -v ./...
 script:
-- sudo go test -v ./...
+- sudo -E go test -v ./...
 - cd psiphon
 - go test -v -covermode=count -coverprofile=coverage.out
 - $HOME/gopath/bin/goveralls -coverprofile=coverage.out -service=travis-ci -repotoken $COVERALLS_TOKEN

psiphon/server/config.go

@@ -409,7 +409,7 @@ func GenerateConfig(serverIPaddress string) ([]byte, []byte, error) {
 
 	// Web server config
 
-	webServerPort := 8080
+	webServerPort := 8088
 
 	webServerSecret, err := psiphon.MakeRandomStringHex(WEB_SERVER_SECRET_BYTE_LENGTH)
 	if err != nil {
@@ -490,7 +490,7 @@ func GenerateConfig(serverIPaddress string) ([]byte, []byte, error) {
 
 	sshPort := 22
 	obfuscatedSSHPort := 53
-	meekPort := 8080
+	meekPort := 8188
 
 	config := &Config{
 		LogLevel:              "info",
@@ -517,9 +517,9 @@ func GenerateConfig(serverIPaddress string) ([]byte, []byte, error) {
 			"FRONTED-MEEK-HTTP-OSSH": 80,
 		},
 		RedisServerAddress:             "",
-		UDPForwardDNSServerAddress:     "8.8.8.8",
+		UDPForwardDNSServerAddress:     "8.8.8.8:53",
 		UDPInterceptUdpgwServerAddress: "127.0.0.1:7300",
-		MeekCookieEncryptionPrivateKey: base64.RawStdEncoding.EncodeToString(meekCookieEncryptionPrivateKey[:]),
+		MeekCookieEncryptionPrivateKey: base64.StdEncoding.EncodeToString(meekCookieEncryptionPrivateKey[:]),
 		MeekObfuscatedKey:              meekObfuscatedKey,
 		MeekCertificateCommonName:      "www.example.org",
 		MeekProhibitedHeaders:          nil,
@@ -572,7 +572,7 @@ func GenerateConfig(serverIPaddress string) ([]byte, []byte, error) {
 		Capabilities:                  capabilities,
 		Region:                        "US",
 		MeekServerPort:                meekPort,
-		MeekCookieEncryptionPublicKey: base64.RawStdEncoding.EncodeToString(meekCookieEncryptionPublicKey[:]),
+		MeekCookieEncryptionPublicKey: base64.StdEncoding.EncodeToString(meekCookieEncryptionPublicKey[:]),
 		MeekObfuscatedKey:             meekObfuscatedKey,
 		MeekFrontingHosts:             []string{serverIPaddress},
 		MeekFrontingAddresses:         []string{serverIPaddress},

psiphon/server/meek.go

@@ -30,6 +30,7 @@ import (
 	"net/http"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon"
@@ -100,6 +101,7 @@ func NewMeekServer(
 	meekServer := &MeekServer{
 		config:        config,
 		listener:      listener,
+		clientHandler: clientHandler,
 		openConns:     new(psiphon.Conns),
 		stopBroadcast: stopBroadcast,
 		sessions:      make(map[string]*meekSession),
@@ -230,7 +232,9 @@ func (server *MeekServer) ServeHTTP(responseWriter http.ResponseWriter, request
 
 	err = session.clientConn.PumpReads(request.Body)
 	if err != nil {
-		log.WithContextFields(LogFields{"error": err}).Warning("pump reads failed")
+		if err != io.EOF {
+			log.WithContextFields(LogFields{"error": err}).Warning("pump reads failed")
+		}
 		server.terminateConnection(responseWriter, request)
 		server.closeSession(sessionID)
 		return
@@ -251,7 +255,9 @@ func (server *MeekServer) ServeHTTP(responseWriter http.ResponseWriter, request
 
 	err = session.clientConn.PumpWrites(responseWriter)
 	if err != nil {
-		log.WithContextFields(LogFields{"error": err}).Warning("pump writes failed")
+		if err != io.EOF {
+			log.WithContextFields(LogFields{"error": err}).Warning("pump writes failed")
+		}
 		server.terminateConnection(responseWriter, request)
 		server.closeSession(sessionID)
 		return
@@ -572,6 +578,7 @@ type meekConn struct {
 	remoteAddr      net.Addr
 	protocolVersion int
 	closeBroadcast  chan struct{}
+	closed          int32
 	readLock        sync.Mutex
 	readyReader     chan io.Reader
 	readResult      chan error
@@ -585,6 +592,7 @@ func newMeekConn(remoteAddr net.Addr, protocolVersion int) *meekConn {
 		remoteAddr:      remoteAddr,
 		protocolVersion: protocolVersion,
 		closeBroadcast:  make(chan struct{}),
+		closed:          0,
 		readyReader:     make(chan io.Reader, 1),
 		readResult:      make(chan error, 1),
 		nextWriteBuffer: make(chan []byte, 1),
@@ -716,7 +724,9 @@ func (conn *meekConn) Write(buffer []byte) (int, error) {
 // Close closes the meekConn. This will interrupt any blocked
 // Read, Write, PumpReads, and PumpWrites.
 func (conn *meekConn) Close() error {
-	close(conn.closeBroadcast)
+	if atomic.CompareAndSwapInt32(&conn.closed, 0, 1) {
+		close(conn.closeBroadcast)
+	}
 	return nil
 }
 

psiphon/server/server_test.go

@@ -66,7 +66,7 @@ func runServer(t *testing.T, tunnelProtocol string) {
 	// create a server
 
 	serverConfigFileContents, serverEntryFileContents, err := GenerateConfig(
-		&GenerateConfigParams{})
+		"127.0.0.1")
 	if err != nil {
 		t.Fatalf("error generating server config: %s", err)
 	}
@@ -133,7 +133,7 @@ func runServer(t *testing.T, tunnelProtocol string) {
 	clientConfig.DisableRemoteServerListFetcher = true
 	clientConfig.EstablishTunnelPausePeriodSeconds = &establishTunnelPausePeriodSeconds
 	clientConfig.TargetServerEntry = string(serverEntryFileContents)
-	clientConfig.TunnelProtocol = "OSSH"
+	clientConfig.TunnelProtocol = tunnelProtocol
 	clientConfig.LocalHttpProxyPort = localHTTPProxyPort
 
 	err = psiphon.InitDataStore(clientConfig)
@@ -170,10 +170,14 @@ func runServer(t *testing.T, tunnelProtocol string) {
 			}
 		}))
 
+	controllerShutdownBroadcast := make(chan struct{})
+	controllerWaitGroup := new(sync.WaitGroup)
+	controllerWaitGroup.Add(1)
 	go func() {
-		shutdownBroadcast := make(chan struct{})
-		controller.Run(shutdownBroadcast)
+		defer controllerWaitGroup.Done()
+		controller.Run(controllerShutdownBroadcast)
 	}()
+	defer close(controllerShutdownBroadcast)
 
 	// Test: tunnels must be established within 30 seconds