Merge pull request #60 from rod-hynes/master

Controller tests; refactored Notice helpers; get Android re-integrated

.gitignore

@@ -1,6 +1,7 @@
 # Exclude test config files from source control
 psiphon_config
 psiphon.config
+controller_test.config
 psiphon.db*
 
 # Compiled Object files, Static and Dynamic libs (Shared Objects)

AndroidApp/app/src/main/java/ca/psiphon/psibot/Psiphon.java

@@ -49,12 +49,12 @@ public class Psiphon extends Psi.PsiphonProvider.Stub {
 
     // PsiphonProvider.Notice
     @Override
-    public void Notice(String message) {
-        message = message.trim();
+    public void Notice(String noticeJSON) {
+        noticeJSON = noticeJSON.trim();
 
-        android.util.Log.d("PSIPHON", message);
-        parseMessage(message);
-        Log.addEntry(message);
+        android.util.Log.d("PSIPHON", noticeJSON);
+        parseMessage(noticeJSON);
+        Log.addEntry(noticeJSON);
     }
 
     // PsiphonProvider.BindToDevice
@@ -72,8 +72,11 @@ public class Psiphon extends Psi.PsiphonProvider.Stub {
         mLocalHttpProxyPort = 0;
         mHomePages = new HashSet<String>();
 
+        // TODO: supply embedded server list
+        String embeddedServerEntryList = "";
+
         try {
-            Psi.Start(loadConfig(mVpnService), this);
+            Psi.Start(loadConfig(mVpnService), embeddedServerEntryList, this);
         } catch (Exception e) {
             throw new Utils.PsibotError("failed to start Psiphon", e);
         }
@@ -172,21 +175,24 @@ public class Psiphon extends Psi.PsiphonProvider.Stub {
         return json.toString();
     }
 
-    private synchronized void parseMessage(String message) {
-        // TODO: this is based on tentative log line formats
-        final String socksProxy = "SOCKS-PROXY-PORT ";
-        final String httpProxy = "HTTP-PROXY-PORT ";
-        final String homePage = "HOMEPAGE ";
-        final String tunnelStarted = "TUNNELS 1";
-        int index;
-        if (-1 != (index = message.indexOf(socksProxy))) {
-            mLocalSocksProxyPort = Integer.parseInt(message.substring(index + socksProxy.length()));
-        } else if (-1 != (index = message.indexOf(httpProxy))) {
-            mLocalHttpProxyPort = Integer.parseInt(message.substring(index + httpProxy.length()));
-        } else if (-1 != (index = message.indexOf(homePage))) {
-            mHomePages.add(message.substring(index + homePage.length()));
-        } else if (message.contains(tunnelStarted)) {
-            mTunnelStartedSignal.countDown();
+    private synchronized void parseMessage(String noticeJSON) {
+        try {
+            JSONObject notice = new JSONObject(noticeJSON);
+            String noticeType = notice.getString("noticeType");
+            if (noticeType.equals("Tunnels")) {
+                int count = notice.getJSONObject("data").getInt("count");
+                if (count == 1) {
+                    mTunnelStartedSignal.countDown();
+                }
+            } else if (noticeType.equals("ListeningSocksProxyPort")) {
+                mLocalSocksProxyPort = notice.getJSONObject("data").getInt("port");
+            } else if (noticeType.equals("ListeningHttpProxyPort")) {
+                mLocalHttpProxyPort = notice.getJSONObject("data").getInt("port");
+            } else if (noticeType.equals("Homepage")) {
+                mHomePages.add(notice.getJSONObject("data").getString("url"));
+            }
+        } catch (JSONException e) {
+            // Ignore notice
         }
     }
 }

AndroidApp/app/src/main/java/go/psi/Psi.java

@@ -12,7 +12,7 @@ public abstract class Psi {
     public interface PsiphonProvider extends go.Seq.Object {
         public void BindToDevice(long fileDescriptor);
         
-        public void Notice(String message);
+        public void Notice(String noticeJSON);
         
         public static abstract class Stub implements PsiphonProvider {
             static final String DESCRIPTOR = "go.psi.PsiphonProvider";
@@ -32,8 +32,8 @@ public abstract class Psi {
                     return;
                 }
                 case Proxy.CALL_Notice: {
-                    String param_message = in.readUTF16();
-                    this.Notice(param_message);
+                    String param_noticeJSON = in.readUTF16();
+                    this.Notice(param_noticeJSON);
                     return;
                 }
                 default:
@@ -63,11 +63,11 @@ public abstract class Psi {
                 Seq.send(DESCRIPTOR, CALL_BindToDevice, _in, _out);
             }
             
-            public void Notice(String message) {
+            public void Notice(String noticeJSON) {
                 go.Seq _in = new go.Seq();
                 go.Seq _out = new go.Seq();
                 _in.writeRef(ref);
-                _in.writeUTF16(message);
+                _in.writeUTF16(noticeJSON);
                 Seq.send(DESCRIPTOR, CALL_Notice, _in, _out);
             }
             
@@ -76,10 +76,11 @@ public abstract class Psi {
         }
     }
     
-    public static void Start(String configJson, PsiphonProvider provider) throws Exception {
+    public static void Start(String configJson, String embeddedServerEntryList, PsiphonProvider provider) throws Exception {
         go.Seq _in = new go.Seq();
         go.Seq _out = new go.Seq();
         _in.writeUTF16(configJson);
+        _in.writeUTF16(embeddedServerEntryList);
         _in.writeRef(provider.ref());
         Seq.send(DESCRIPTOR, CALL_Start, _in, _out);
         String _err = _out.readUTF16();

AndroidLibrary/go_psi/go_psi.go

@@ -5,7 +5,7 @@
 package go_psi
 
 import (
-	"github.com/Psiphon-Labs/psiphon-tunnel-core/AndroidLibrary/psi"
+	"."
 	"golang.org/x/mobile/bind/seq"
 )
 
@@ -23,14 +23,15 @@ func (p *proxyPsiphonProvider) BindToDevice(fileDescriptor int) {
 	seq.Transact((*seq.Ref)(p), proxyPsiphonProviderBindToDeviceCode, out)
 }
 
-func (p *proxyPsiphonProvider) Notice(message string) {
+func (p *proxyPsiphonProvider) Notice(noticeJSON string) {
 	out := new(seq.Buffer)
-	out.WriteUTF16(message)
+	out.WriteUTF16(noticeJSON)
 	seq.Transact((*seq.Ref)(p), proxyPsiphonProviderNoticeCode, out)
 }
 
 func proxy_Start(out, in *seq.Buffer) {
 	param_configJson := in.ReadUTF16()
+	param_embeddedServerEntryList := in.ReadUTF16()
 	var param_provider psi.PsiphonProvider
 	param_provider_ref := in.ReadRef()
 	if param_provider_ref.Num < 0 {
@@ -38,7 +39,7 @@ func proxy_Start(out, in *seq.Buffer) {
 	} else {
 		param_provider = (*proxyPsiphonProvider)(param_provider_ref)
 	}
-	err := psi.Start(param_configJson, param_provider)
+	err := psi.Start(param_configJson, param_embeddedServerEntryList, param_provider)
 	if err == nil {
 		out.WriteUTF16("")
 	} else {

AndroidLibrary/java_psi/go/psi/Psi.java

@@ -12,7 +12,7 @@ public abstract class Psi {
     public interface PsiphonProvider extends go.Seq.Object {
         public void BindToDevice(long fileDescriptor);
         
-        public void Notice(String message);
+        public void Notice(String noticeJSON);
         
         public static abstract class Stub implements PsiphonProvider {
             static final String DESCRIPTOR = "go.psi.PsiphonProvider";
@@ -32,8 +32,8 @@ public abstract class Psi {
                     return;
                 }
                 case Proxy.CALL_Notice: {
-                    String param_message = in.readUTF16();
-                    this.Notice(param_message);
+                    String param_noticeJSON = in.readUTF16();
+                    this.Notice(param_noticeJSON);
                     return;
                 }
                 default:
@@ -63,11 +63,11 @@ public abstract class Psi {
                 Seq.send(DESCRIPTOR, CALL_BindToDevice, _in, _out);
             }
             
-            public void Notice(String message) {
+            public void Notice(String noticeJSON) {
                 go.Seq _in = new go.Seq();
                 go.Seq _out = new go.Seq();
                 _in.writeRef(ref);
-                _in.writeUTF16(message);
+                _in.writeUTF16(noticeJSON);
                 Seq.send(DESCRIPTOR, CALL_Notice, _in, _out);
             }
             
@@ -76,10 +76,11 @@ public abstract class Psi {
         }
     }
     
-    public static void Start(String configJson, PsiphonProvider provider) throws Exception {
+    public static void Start(String configJson, String embeddedServerEntryList, PsiphonProvider provider) throws Exception {
         go.Seq _in = new go.Seq();
         go.Seq _out = new go.Seq();
         _in.writeUTF16(configJson);
+        _in.writeUTF16(embeddedServerEntryList);
         _in.writeRef(provider.ref());
         Seq.send(DESCRIPTOR, CALL_Start, _in, _out);
         String _err = _out.readUTF16();

AndroidLibrary/psi/psi.go

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2014, Psiphon Inc.
+ * Copyright (c) 2015, Psiphon Inc.
  * All rights reserved.
  *
  * This program is free software: you can redistribute it and/or modify
@@ -33,28 +33,18 @@ import (
 )
 
 type PsiphonProvider interface {
-	Notice(message string)
+	Notice(noticeJSON string)
 
 	// TODO: return 'error'; at the moment gobind doesn't
 	// work with interface function return values.
 	BindToDevice(fileDescriptor int)
 }
 
-type logRelay struct {
-	provider PsiphonProvider
-}
-
-func (lr *logRelay) Write(p []byte) (n int, err error) {
-	// TODO: buffer incomplete lines
-	lr.provider.Notice(string(p))
-	return len(p), nil
-}
-
 var controller *psiphon.Controller
 var shutdownBroadcast chan struct{}
 var controllerWaitGroup *sync.WaitGroup
 
-func Start(configJson string, provider PsiphonProvider) error {
+func Start(configJson, embeddedServerEntryList string, provider PsiphonProvider) error {
 
 	if controller != nil {
 		return fmt.Errorf("already started")
@@ -64,17 +54,32 @@ func Start(configJson string, provider PsiphonProvider) error {
 	if err != nil {
 		return fmt.Errorf("error loading configuration file: %s", err)
 	}
+	config.BindToDeviceProvider = provider
 
 	err = psiphon.InitDataStore(config)
 	if err != nil {
 		return fmt.Errorf("error initializing datastore: %s", err)
 	}
 
-	log.SetOutput(&logRelay{provider: provider})
+	psiphon.SetNoticeOutput(psiphon.NewNoticeReceiver(
+		func(notice []byte) {
+			provider.Notice(string(notice))
+		}))
 
-	config.BindToDeviceProvider = provider
+	serverEntries, err := psiphon.DecodeAndValidateServerEntryList(embeddedServerEntryList)
+	if err != nil {
+		log.Fatalf("error decoding embedded server entry list: %s", err)
+	}
+	err = psiphon.StoreServerEntries(serverEntries, false)
+	if err != nil {
+		log.Fatalf("error storing embedded server entry list: %s", err)
+	}
+
+	controller, err = psiphon.NewController(config)
+	if err != nil {
+		return fmt.Errorf("error initializing controller: %s", err)
+	}
 
-	controller = psiphon.NewController(config)
 	shutdownBroadcast = make(chan struct{})
 	controllerWaitGroup = new(sync.WaitGroup)
 	controllerWaitGroup.Add(1)

ConsoleClient/psiphonClient.go

@@ -21,6 +21,7 @@ package main
 
 import (
 	"flag"
+	"io"
 	"io/ioutil"
 	"log"
 	"os"
@@ -41,6 +42,9 @@ func main() {
 	var embeddedServerEntryListFilename string
 	flag.StringVar(&embeddedServerEntryListFilename, "serverList", "", "embedded server entry list input file")
 
+	var formatNotices bool
+	flag.BoolVar(&formatNotices, "formatNotices", false, "emit notices in human-readable format")
+
 	var profileFilename string
 	flag.StringVar(&profileFilename, "profile", "", "CPU profile output file")
 
@@ -60,18 +64,21 @@ func main() {
 		log.Fatalf("error processing configuration file: %s", err)
 	}
 
-	// Set logfile, if configured
+	// Initialize notice output; use logfile, if configured
 
+	var noticeWriter io.Writer
+	noticeWriter = os.Stderr
 	if config.LogFilename != "" {
 		logFile, err := os.OpenFile(config.LogFilename, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0600)
 		if err != nil {
 			log.Fatalf("error opening log file: %s", err)
 		}
 		defer logFile.Close()
-		psiphon.SetNoticeOutput(logFile)
-	} else {
-		psiphon.SetNoticeOutput(psiphon.NewNoticeConsoleRewriter(os.Stderr))
 	}
+	if formatNotices {
+		noticeWriter = psiphon.NewNoticeConsoleRewriter(noticeWriter)
+	}
+	psiphon.SetNoticeOutput(noticeWriter)
 
 	// Handle optional profiling parameter
 

psiphon/controller_test.go

@@ -0,0 +1,125 @@
+/*
+ * Copyright (c) 2015, Psiphon Inc.
+ * All rights reserved.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package psiphon
+
+import (
+	"io/ioutil"
+	"sync"
+	"testing"
+	"time"
+)
+
+func TestControllerRunSSH(t *testing.T) {
+	controllerRun(t, TUNNEL_PROTOCOL_SSH)
+}
+
+func TestControllerRunObfuscatedSSH(t *testing.T) {
+	controllerRun(t, TUNNEL_PROTOCOL_OBFUSCATED_SSH)
+}
+
+func TestControllerRunUnfrontedMeek(t *testing.T) {
+	controllerRun(t, TUNNEL_PROTOCOL_UNFRONTED_MEEK)
+}
+
+func TestControllerRunFrontedMeek(t *testing.T) {
+	controllerRun(t, TUNNEL_PROTOCOL_FRONTED_MEEK)
+}
+
+func controllerRun(t *testing.T, protocol string) {
+
+	configFileContents, err := ioutil.ReadFile("controller_test.config")
+	if err != nil {
+		// Skip, don't fail, if config file is not present
+		t.Skipf("error loading configuration file: %s", err)
+	}
+	config, err := LoadConfig(configFileContents)
+	if err != nil {
+		t.Errorf("error processing configuration file: %s", err)
+		t.FailNow()
+	}
+	config.TunnelProtocol = protocol
+
+	err = InitDataStore(config)
+	if err != nil {
+		t.Errorf("error initializing datastore: %s", err)
+		t.FailNow()
+	}
+
+	controller, err := NewController(config)
+	if err != nil {
+		t.Errorf("error creating controller: %s", err)
+		t.FailNow()
+	}
+
+	// Monitor notices for "Tunnels" with count > 1, the
+	// indication of tunnel establishment success
+
+	tunnelEstablished := make(chan struct{}, 1)
+	SetNoticeOutput(NewNoticeReceiver(
+		func(notice []byte) {
+			// TODO: log notices without logging server IPs:
+			// fmt.Fprintf(os.Stderr, "%s\n", string(notice))
+			count, ok := GetNoticeTunnels(notice)
+			if ok && count > 0 {
+				select {
+				case tunnelEstablished <- *new(struct{}):
+				default:
+				}
+			}
+		}))
+
+	// Run controller, which establishes tunnels
+
+	shutdownBroadcast := make(chan struct{})
+	controllerWaitGroup := new(sync.WaitGroup)
+	controllerWaitGroup.Add(1)
+	go func() {
+		defer controllerWaitGroup.Done()
+		controller.Run(shutdownBroadcast)
+	}()
+
+	// Test: tunnel must be established within 60 seconds
+
+	establishTimeout := time.NewTimer(60 * time.Second)
+
+	select {
+	case <-tunnelEstablished:
+	case <-establishTimeout.C:
+		t.Errorf("tunnel establish timeout exceeded")
+	}
+
+	close(shutdownBroadcast)
+
+	// Test: shutdown must complete within 10 seconds
+
+	shutdownTimeout := time.NewTimer(10 * time.Second)
+
+	shutdownOk := make(chan struct{}, 1)
+	go func() {
+		controllerWaitGroup.Wait()
+		shutdownOk <- *new(struct{})
+	}()
+
+	select {
+	case <-shutdownOk:
+	case <-shutdownTimeout.C:
+		t.Errorf("controller shutdown timeout exceeded")
+	}
+}

psiphon/meekConn.go

@@ -126,16 +126,18 @@ func DialMeek(
 		// In this case, host is both what is dialed and what ends up in the HTTP Host header
 		host = fmt.Sprintf("%s:%d", serverEntry.IpAddress, serverEntry.MeekServerPort)
 
-		// For unfronted meek, we let the http.Transport handle proxying, as the
-		// target server hostname has to be in the HTTP request line. Also, in this
-		// case, we don't require the proxy to support CONNECT and so we can work
-		// throigh HTTP proxies that don't support it.
-		url, err := url.Parse(fmt.Sprintf("http://%s", meekConfig.UpstreamHttpProxyAddress))
-		if err != nil {
-			return nil, ContextError(err)
+		if meekConfig.UpstreamHttpProxyAddress != "" {
+			// For unfronted meek, we let the http.Transport handle proxying, as the
+			// target server hostname has to be in the HTTP request line. Also, in this
+			// case, we don't require the proxy to support CONNECT and so we can work
+			// throigh HTTP proxies that don't support it.
+			url, err := url.Parse(fmt.Sprintf("http://%s", meekConfig.UpstreamHttpProxyAddress))
+			if err != nil {
+				return nil, ContextError(err)
+			}
+			proxyUrl = http.ProxyURL(url)
+			meekConfig.UpstreamHttpProxyAddress = ""
 		}
-		proxyUrl = http.ProxyURL(url)
-		meekConfig.UpstreamHttpProxyAddress = ""
 
 		dialer = NewTCPDialer(meekConfig)
 	}

psiphon/notice.go

@@ -20,6 +20,7 @@
 package psiphon
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -157,3 +158,78 @@ func NoticeHomepage(url string) {
 func NoticeTunnels(count int) {
 	outputNotice("Tunnels", false, "count", count)
 }
+
+type noticeObject struct {
+	NoticeType string          `json:"noticeType"`
+	Data       json.RawMessage `json:"data"`
+	Timestamp  string          `json:"timestamp"`
+}
+
+// GetNoticeTunnels receives a JSON encoded object and attempts to parse it as a Notice.
+// When the object is a Notice of type Tunnels, the count payload is returned.
+func GetNoticeTunnels(notice []byte) (count int, ok bool) {
+	var object noticeObject
+	if json.Unmarshal(notice, &object) != nil {
+		return 0, false
+	}
+	if object.NoticeType != "Tunnels" {
+		return 0, false
+	}
+	type tunnelsPayload struct {
+		Count int `json:"count"`
+	}
+	var payload tunnelsPayload
+	if json.Unmarshal(object.Data, &payload) != nil {
+		return 0, false
+	}
+	return payload.Count, true
+}
+
+// NoticeReceiver consumes a notice input stream and invokes a callback function
+// for each discrete JSON notice object byte sequence.
+type NoticeReceiver struct {
+	mutex    sync.Mutex
+	buffer   []byte
+	callback func([]byte)
+}
+
+// NewNoticeReceiver initializes a new NoticeReceiver
+func NewNoticeReceiver(callback func([]byte)) *NoticeReceiver {
+	return &NoticeReceiver{callback: callback}
+}
+
+// Write implements io.Writer.
+func (receiver *NoticeReceiver) Write(p []byte) (n int, err error) {
+	receiver.mutex.Lock()
+	defer receiver.mutex.Unlock()
+
+	receiver.buffer = append(receiver.buffer, p...)
+
+	index := bytes.Index(receiver.buffer, []byte("\n"))
+	if index == -1 {
+		return len(p), nil
+	}
+
+	notice := receiver.buffer[:index]
+	receiver.buffer = receiver.buffer[index+1:]
+
+	receiver.callback(notice)
+
+	return len(p), nil
+}
+
+// NewNoticeConsoleRewriter consumes JSON-format notice input and parses each
+// notice and rewrites in a more human-readable format more suitable for
+// console output. The data payload field is left as JSON.
+func NewNoticeConsoleRewriter(writer io.Writer) *NoticeReceiver {
+	return NewNoticeReceiver(func(notice []byte) {
+		var object noticeObject
+		_ = json.Unmarshal(notice, &object)
+		fmt.Fprintf(
+			writer,
+			"%s %s %s\n",
+			object.Timestamp,
+			object.NoticeType,
+			string(object.Data))
+	})
+}

psiphon/utils.go

@@ -20,19 +20,14 @@
 package psiphon
 
 import (
-	"bytes"
 	"crypto/rand"
 	"crypto/x509"
 	"encoding/base64"
-	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"math/big"
-	"os"
 	"runtime"
 	"strings"
-	"sync"
 	"time"
 )
 
@@ -150,48 +145,3 @@ func ContextError(err error) error {
 func IsNetworkBindError(err error) bool {
 	return strings.Contains(err.Error(), "bind: address already in use")
 }
-
-// NoticeConsoleRewriter consumes JOSN-format notice input and parses each
-// notice and rewrites in a more human-readable format more suitable for
-// console output. The data payload field is left as JSON.
-type NoticeConsoleRewriter struct {
-	mutex  sync.Mutex
-	writer io.Writer
-	buffer []byte
-}
-
-// NewNoticeConsoleRewriter initializes a new NoticeConsoleRewriter
-func NewNoticeConsoleRewriter(writer io.Writer) *NoticeConsoleRewriter {
-	return &NoticeConsoleRewriter{writer: writer}
-}
-
-// Write implements io.Writer.
-func (rewriter *NoticeConsoleRewriter) Write(p []byte) (n int, err error) {
-	rewriter.mutex.Lock()
-	defer rewriter.mutex.Unlock()
-
-	rewriter.buffer = append(rewriter.buffer, p...)
-
-	index := bytes.Index(rewriter.buffer, []byte("\n"))
-	if index == -1 {
-		return len(p), nil
-	}
-	line := rewriter.buffer[:index]
-	rewriter.buffer = rewriter.buffer[index+1:]
-
-	type NoticeObject struct {
-		NoticeType string          `json:"noticeType"`
-		Data       json.RawMessage `json:"data"`
-		Timestamp  string          `json:"timestamp"`
-	}
-
-	var noticeObject NoticeObject
-	_ = json.Unmarshal(line, &noticeObject)
-	fmt.Fprintf(os.Stderr,
-		"%s %s %s\n",
-		noticeObject.Timestamp,
-		noticeObject.NoticeType,
-		string(noticeObject.Data))
-
-	return len(p), nil
-}