Merge remote-tracking branch 'rod/master'

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()
+}