Decouple Controller from HttpProxy and SocksProxy. Use a Tunneler interface so either a Controller (tunnel pool with lifecycle management) or Tunnel (single, specific tunnel) may be used as the tunneler for proxies. This is to support prospective test scripts that will test tunnels to specific servers.

psiphon/controller.go

@@ -84,13 +84,13 @@ func NewController(config *Config) (controller *Controller) {
 // - a local SOCKS proxy that port forwards through the pool of tunnels
 // - a local HTTP proxy that port forwards through the pool of tunnels
 func (controller *Controller) Run(shutdownBroadcast <-chan struct{}) {
-	socksProxy, err := NewSocksProxy(controller)
+	socksProxy, err := NewSocksProxy(controller.config, controller)
 	if err != nil {
 		Notice(NOTICE_ALERT, "error initializing local SOCKS proxy: %s", err)
 		return
 	}
 	defer socksProxy.Close()
-	httpProxy, err := NewHttpProxy(controller)
+	httpProxy, err := NewHttpProxy(controller.config, controller)
 	if err != nil {
 		Notice(NOTICE_ALERT, "error initializing local SOCKS proxy: %s", err)
 		return
@@ -117,7 +117,7 @@ func (controller *Controller) Run(shutdownBroadcast <-chan struct{}) {
 	Notice(NOTICE_INFO, "exiting controller")
 }
 
-// SignalFailure notifies the controller than a component has failed.
+// SignalFailure notifies the controller that an associated component has failed.
 // This will terminate the controller.
 func (controller *Controller) SignalFailure() {
 	select {
@@ -441,15 +441,15 @@ func (conn *TunneledConn) Write(buffer []byte) (n int, err error) {
 	return
 }
 
-// dialWithTunnel selects an active tunnel and establishes a port forward
+// DialWithTunnel selects an active tunnel and establishes a port forward
 // connection through the selected tunnel. Failure to connect is considered
 // a port foward failure, for the purpose of monitoring tunnel health.
-func (controller *Controller) dialWithTunnel(remoteAddr string) (conn net.Conn, err error) {
+func (controller *Controller) Dial(remoteAddr string) (conn net.Conn, err error) {
 	tunnel := controller.getNextActiveTunnel()
 	if tunnel == nil {
 		return nil, ContextError(errors.New("no active tunnels"))
 	}
-	sshPortForward, err := tunnel.sshClient.Dial("tcp", remoteAddr)
+	tunnelConn, err := tunnel.Dial(remoteAddr)
 	if err != nil {
 		// TODO: conditional on type of error or error message?
 		select {
@@ -459,7 +459,7 @@ func (controller *Controller) dialWithTunnel(remoteAddr string) (conn net.Conn,
 		return nil, ContextError(err)
 	}
 	return &TunneledConn{
-			Conn:   sshPortForward,
+			Conn:   tunnelConn,
 			tunnel: tunnel},
 		nil
 }
@@ -576,7 +576,8 @@ func (controller *Controller) establishTunnelWorker() {
 			return
 		default:
 		}
-		tunnel, err := EstablishTunnel(controller, serverEntry)
+		tunnel, err := EstablishTunnel(
+			controller.config, controller.pendingConns, serverEntry)
 		if err != nil {
 			// TODO: distingush case where conn is interrupted?
 			Notice(NOTICE_INFO, "failed to connect to %s: %s", serverEntry.IpAddress, err)

psiphon/httpProxy.go

@@ -31,7 +31,7 @@ import (
 // HttpProxy is a HTTP server that relays HTTP requests through
 // the tunnel SSH client.
 type HttpProxy struct {
-	controller     *Controller
+	tunneler       Tunneler
 	listener       net.Listener
 	serveWaitGroup *sync.WaitGroup
 	httpRelay      *http.Transport
@@ -39,15 +39,15 @@ type HttpProxy struct {
 }
 
 // NewHttpProxy initializes and runs a new HTTP proxy server.
-func NewHttpProxy(controller *Controller) (proxy *HttpProxy, err error) {
+func NewHttpProxy(config *Config, tunneler Tunneler) (proxy *HttpProxy, err error) {
 	listener, err := net.Listen(
-		"tcp", fmt.Sprintf("127.0.0.1:%d", controller.config.LocalHttpProxyPort))
+		"tcp", fmt.Sprintf("127.0.0.1:%d", config.LocalHttpProxyPort))
 	if err != nil {
 		return nil, ContextError(err)
 	}
 	tunneledDialer := func(_, addr string) (conn net.Conn, err error) {
 		// TODO: connect timeout?
-		return controller.dialWithTunnel(addr)
+		return tunneler.Dial(addr)
 	}
 	// TODO: also use http.Client, with its Timeout field?
 	transport := &http.Transport{
@@ -56,7 +56,7 @@ func NewHttpProxy(controller *Controller) (proxy *HttpProxy, err error) {
 		ResponseHeaderTimeout: HTTP_PROXY_ORIGIN_SERVER_TIMEOUT,
 	}
 	proxy = &HttpProxy{
-		controller:     controller,
+		tunneler:       tunneler,
 		listener:       listener,
 		serveWaitGroup: new(sync.WaitGroup),
 		httpRelay:      transport,
@@ -187,7 +187,7 @@ func (proxy *HttpProxy) httpConnectHandler(localConn net.Conn, target string) (e
 	defer localConn.Close()
 	defer proxy.openConns.Remove(localConn)
 	proxy.openConns.Add(localConn)
-	remoteConn, err := proxy.controller.dialWithTunnel(target)
+	remoteConn, err := proxy.tunneler.Dial(target)
 	if err != nil {
 		return ContextError(err)
 	}
@@ -227,7 +227,7 @@ func (proxy *HttpProxy) serve() {
 	// Note: will be interrupted by listener.Close() call made by proxy.Close()
 	err := httpServer.Serve(proxy.listener)
 	if err != nil {
-		proxy.controller.SignalFailure()
+		proxy.tunneler.SignalFailure()
 		Notice(NOTICE_ALERT, "%s", ContextError(err))
 	}
 	Notice(NOTICE_HTTP_PROXY, "HTTP proxy stopped")

psiphon/socksProxy.go

@@ -31,7 +31,7 @@ import (
 // the tunnel SSH client and relays traffic through the port
 // forward.
 type SocksProxy struct {
-	controller     *Controller
+	tunneler       Tunneler
 	listener       *socks.SocksListener
 	serveWaitGroup *sync.WaitGroup
 	openConns      *Conns
@@ -40,14 +40,14 @@ type SocksProxy struct {
 // NewSocksProxy initializes a new SOCKS server. It begins listening for
 // connections, starts a goroutine that runs an accept loop, and returns
 // leaving the accept loop running.
-func NewSocksProxy(controller *Controller) (proxy *SocksProxy, err error) {
+func NewSocksProxy(config *Config, tunneler Tunneler) (proxy *SocksProxy, err error) {
 	listener, err := socks.ListenSocks(
-		"tcp", fmt.Sprintf("127.0.0.1:%d", controller.config.LocalSocksProxyPort))
+		"tcp", fmt.Sprintf("127.0.0.1:%d", config.LocalSocksProxyPort))
 	if err != nil {
 		return nil, ContextError(err)
 	}
 	proxy = &SocksProxy{
-		controller:     controller,
+		tunneler:       tunneler,
 		listener:       listener,
 		serveWaitGroup: new(sync.WaitGroup),
 		openConns:      new(Conns),
@@ -70,7 +70,7 @@ func (proxy *SocksProxy) socksConnectionHandler(localConn *socks.SocksConn) (err
 	defer localConn.Close()
 	defer proxy.openConns.Remove(localConn)
 	proxy.openConns.Add(localConn)
-	remoteConn, err := proxy.controller.dialWithTunnel(localConn.Req.Target)
+	remoteConn, err := proxy.tunneler.Dial(localConn.Req.Target)
 	if err != nil {
 		return ContextError(err)
 	}
@@ -92,7 +92,7 @@ func (proxy *SocksProxy) serve() {
 		if err != nil {
 			Notice(NOTICE_ALERT, "SOCKS proxy accept error: %s", err)
 			if e, ok := err.(net.Error); ok && !e.Temporary() {
-				proxy.controller.SignalFailure()
+				proxy.tunneler.SignalFailure()
 				// Fatal error, stop the proxy
 				break
 			}

psiphon/tunnel.go

@@ -31,6 +31,15 @@ import (
 	"time"
 )
 
+// Tunneler specifies the interface required by components that use a tunnel.
+// Components which use this interface may be services by a single Tunnel instance,
+// or a Controller which manages a pool of tunnels, or any other object which
+// implements Tunneler.
+type Tunneler interface {
+	Dial(remoteAddr string) (conn net.Conn, err error)
+	SignalFailure()
+}
+
 const (
 	TUNNEL_PROTOCOL_SSH            = "SSH"
 	TUNNEL_PROTOCOL_OBFUSCATED_SSH = "OSSH"
@@ -81,18 +90,18 @@ func (tunnel *Tunnel) Close() {
 // the first protocol in SupportedTunnelProtocols that's also in the
 // server capabilities is used.
 func EstablishTunnel(
-	controller *Controller, serverEntry *ServerEntry) (tunnel *Tunnel, err error) {
+	config *Config, pendingConns *Conns, serverEntry *ServerEntry) (tunnel *Tunnel, err error) {
 
 	// Select the protocol
 	var selectedProtocol string
 	// TODO: properly handle protocols (e.g. FRONTED-MEEK-OSSH) vs. capabilities (e.g., {FRONTED-MEEK, OSSH})
 	// for now, the code is simply assuming that MEEK capabilities imply OSSH capability.
-	if controller.config.TunnelProtocol != "" {
-		requiredCapability := strings.TrimSuffix(controller.config.TunnelProtocol, "-OSSH")
+	if config.TunnelProtocol != "" {
+		requiredCapability := strings.TrimSuffix(config.TunnelProtocol, "-OSSH")
 		if !Contains(serverEntry.Capabilities, requiredCapability) {
 			return nil, ContextError(fmt.Errorf("server does not have required capability"))
 		}
-		selectedProtocol = controller.config.TunnelProtocol
+		selectedProtocol = config.TunnelProtocol
 	} else {
 		// Order of SupportedTunnelProtocols is default preference order
 		for _, protocol := range SupportedTunnelProtocols {
@@ -144,9 +153,9 @@ func EstablishTunnel(
 		ConnectTimeout:             TUNNEL_CONNECT_TIMEOUT,
 		ReadTimeout:                TUNNEL_READ_TIMEOUT,
 		WriteTimeout:               TUNNEL_WRITE_TIMEOUT,
-		PendingConns:               controller.pendingConns,
-		BindToDeviceServiceAddress: controller.config.BindToDeviceServiceAddress,
-		BindToDeviceDnsServer:      controller.config.BindToDeviceDnsServer,
+		PendingConns:               pendingConns,
+		BindToDeviceServiceAddress: config.BindToDeviceServiceAddress,
+		BindToDeviceDnsServer:      config.BindToDeviceDnsServer,
 	}
 	var conn Conn
 	if useMeek {
@@ -247,6 +256,19 @@ func EstablishTunnel(
 			sshKeepAliveQuit: sshKeepAliveQuit,
 			// portForwardFailures buffer size is large enough to receive the thresold number
 			// of failure reports without blocking. Senders can drop failures without blocking.
-			portForwardFailures: make(chan int, controller.config.PortForwardFailureThreshold)},
+			portForwardFailures: make(chan int, config.PortForwardFailureThreshold)},
 		nil
 }
+
+// Dial establishes a port forward connection through the tunnel
+func (tunnel *Tunnel) Dial(remoteAddr string) (conn net.Conn, err error) {
+	// TODO: should this track port forward failures as in Controller.DialWithTunnel?
+	return tunnel.sshClient.Dial("tcp", remoteAddr)
+}
+
+// SignalFailure notifies the tunnel that an associated component has failed.
+// This will terminate the tunnel.
+func (tunnel *Tunnel) SignalFailure() {
+	Notice(NOTICE_ALERT, "tunnel received failure signal")
+	tunnel.Close()
+}