Integrated upstreamproxy

- Change config param to "UpstreamProxyUrl", which supports
  socks4a/socks5/http proxy URLs, including auth params
- Removed existing ad hoc "HTTPS CONNECT" code
- Use upstreamproxy dialer in TCPDial for direct connect and
  fronted meek; add explicit timeout for proxy dialing in this
  case
- Use upstreamproxy transport for unfronted meek

README.md

@@ -41,7 +41,7 @@ Setup
         "ConnectionWorkerPoolSize" : 10,
         "TunnelPoolSize" : 1,
         "PortForwardFailureThreshold" : 10,
-        "UpstreamHttpProxyAddress" : ""
+        "UpstreamProxyUrl" : ""
     }
     ```
 <!--END-SAMPLE-CONFIG-->

SampleApps/Psibot/app/src/main/java/ca/psiphon/psibot/Service.java

@@ -110,10 +110,10 @@ public class Service extends VpnService
                     preferences.getString(
                             getString(R.string.preferenceTunnelProtocol),
                             getString(R.string.preferenceTunnelProtocolDefaultValue)));
-            config.put("UpstreamHttpProxyAddress",
+            config.put("UpstreamProxyUrl",
                     preferences.getString(
-                            getString(R.string.preferenceUpstreamHttpProxyAddress),
-                            getString(R.string.preferenceUpstreamHttpProxyAddressDefaultValue)));
+                            getString(R.string.preferenceUpstreamProxyUrl),
+                            getString(R.string.preferenceUpstreamProxyUrlDefaultValue)));
             config.put("LocalHttpProxyPort",
                     Integer.parseInt(
                             preferences.getString(

SampleApps/Psibot/app/src/main/res/raw/psiphon_config_stub

@@ -13,5 +13,5 @@
     "ConnectionWorkerPoolSize" : 10,
     "TunnelPoolSize" : 1,
     "PortForwardFailureThreshold" : 10,
-    "UpstreamHttpProxyAddress" : ""
+    "UpstreamProxyUrl" : ""
 }

SampleApps/Psibot/app/src/main/res/values/strings.xml

@@ -27,7 +27,7 @@
         <item>Fronted Meek</item>
     </string-array>
 
-    <string name="preference_upstream_http_proxy_address">Upstream HTTP Proxy Address</string>
+    <string name="preference_upstream_proxy_url">Upstream Proxy URL</string>
     <string name="preference_local_http_proxy_port">Local HTTP Proxy Port</string>
     <string name="preference_local_socks_proxy_port">Local SOCKS Proxy Port</string>
     <string name="preference_connection_worker_pool_size">Connection Workers</string>

SampleApps/Psibot/app/src/main/res/values/symbols.xml

@@ -22,8 +22,8 @@
     </string-array>
     <string name="preferenceTunnelProtocolDefaultValue"></string>
 
-    <string name="preferenceUpstreamHttpProxyAddress">preferenceUpstreamHttpProxyAddress</string>
-    <string name="preferenceUpstreamHttpProxyAddressDefaultValue"></string>
+    <string name="preferenceUpstreamProxyUrl">preferenceUpstreamProxyUrl</string>
+    <string name="preferenceUpstreamProxyUrlDefaultValue"></string>
 
     <string name="preferenceLocalHttpProxyPort">preferenceLocalHttpProxyPort</string>
     <string name="preferenceLocalHttpProxyPortDefaultValue">0</string>

SampleApps/Psibot/app/src/main/res/xml/preferences.xml

@@ -18,10 +18,10 @@
         android:defaultValue="@string/preferenceTunnelProtocolDefaultValue" />
 
     <EditTextPreference
-        android:key="@string/preferenceUpstreamHttpProxyAddress"
-        android:title="@string/preference_upstream_http_proxy_address"
-        android:dialogTitle="@string/preference_upstream_http_proxy_address"
-        android:defaultValue="@string/preferenceUpstreamHttpProxyAddressDefaultValue" />
+        android:key="@string/preferenceUpstreamProxyUrl"
+        android:title="@string/preference_upstream_proxy_url"
+        android:dialogTitle="@string/preference_upstream_proxy_url"
+        android:defaultValue="@string/preferenceUpstreamProxyUrlDefaultValue" />
 
     <EditTextPreference
         android:numeric="integer"

psiphon/TCPConn.go

@@ -24,13 +24,18 @@ import (
 	"net"
 	"sync"
 	"time"
+
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/upstreamproxy"
 )
 
 // TCPConn is a customized TCP connection that:
 // - can be interrupted while connecting;
+// - implements a connect timeout;
 // - implements idle read/write timeouts;
+// - uses an upstream proxy when specified, and includes
+//   upstream proxy dialing in the connect timeout;
 // - can be bound to a specific system device (for Android VpnService
-//   routing compatibility, for example).
+//   routing compatibility, for example);
 // - implements the psiphon.Conn interface
 type TCPConn struct {
 	net.Conn
@@ -44,15 +49,58 @@ type TCPConn struct {
 
 // NewTCPDialer creates a TCPDialer.
 func NewTCPDialer(config *DialConfig) Dialer {
-	return func(network, addr string) (net.Conn, error) {
+
+	dialer := func(network, addr string) (net.Conn, error) {
 		if network != "tcp" {
 			return nil, errors.New("unsupported network type in NewTCPDialer")
 		}
 		return DialTCP(addr, config)
 	}
+
+	if config.UpstreamProxyUrl != "" {
+
+		upstreamDialer := upstreamproxy.NewProxyDialFunc(
+			&upstreamproxy.UpstreamProxyConfig{
+				ForwardDialFunc: dialer,
+				ProxyURIString:  config.UpstreamProxyUrl,
+			})
+
+		dialer = func(network, addr string) (net.Conn, error) {
+
+			// The entire upstream dial is wrapped in an explicit timeout. This
+			// may include network connection read and writes when proxy auth negotation
+			// is performed.
+
+			type upstreamDialResult struct {
+				conn net.Conn
+				err  error
+			}
+			resultChannel := make(chan *upstreamDialResult, 2)
+			time.AfterFunc(config.ConnectTimeout, func() {
+				// TODO: we could "interrupt" the underlying TCPConn at this point, as
+				// it's being abandoned. But we don't have a reference to it. It's left
+				// to the outer DialConfig.PendingConns to track and clean up that TCPConn.
+				resultChannel <- &upstreamDialResult{nil, errors.New("upstreamproxy dial timeout")}
+			})
+			go func() {
+				conn, err := upstreamDialer(network, addr)
+				resultChannel <- &upstreamDialResult{conn, err}
+			}()
+			result := <-resultChannel
+
+			if _, ok := result.err.(upstreamproxy.Error); ok {
+				NoticeUpstreamProxyError(result.err)
+			}
+
+			return result.conn, result.err
+		}
+	}
+
+	return dialer
 }
 
-// TCPConn creates a new, connected TCPConn.
+// TCPConn creates a new, connected TCPConn. It uses an upstream proxy
+// when specified.
 func DialTCP(addr string, config *DialConfig) (conn *TCPConn, err error) {
 	conn, err = interruptibleTCPDial(addr, config)
 	if err != nil {

psiphon/TCPConn_unix.go

@@ -68,24 +68,9 @@ func interruptibleTCPDial(addr string, config *DialConfig) (conn *TCPConn, err e
 		}
 	}
 
-	// When using an upstream HTTP proxy, first connect to the proxy,
-	// then use HTTP CONNECT to connect to the original destination.
-	dialAddr := addr
-	if config.UpstreamHttpProxyAddress != "" {
-		dialAddr = config.UpstreamHttpProxyAddress
-
-		// Report connection errors in a notice, as user may have input
-		// invalid proxy address or credential
-		defer func() {
-			if err != nil {
-				NoticeUpstreamProxyError(err)
-			}
-		}()
-	}
-
 	// Get the remote IP and port, resolving a domain name if necessary
 	// TODO: domain name resolution isn't interruptible
-	host, strPort, err := net.SplitHostPort(dialAddr)
+	host, strPort, err := net.SplitHostPort(addr)
 	if err != nil {
 		return nil, ContextError(err)
 	}
@@ -174,15 +159,6 @@ func interruptibleTCPDial(addr string, config *DialConfig) (conn *TCPConn, err e
 
 	conn.mutex.Unlock()
 
-	// Going through upstream HTTP proxy
-	if config.UpstreamHttpProxyAddress != "" {
-		// This call can be interrupted by closing the pending conn
-		err = HttpProxyConnect(conn, addr)
-		if err != nil {
-			return nil, ContextError(err)
-		}
-	}
-
 	return conn, nil
 }
 

psiphon/TCPConn_windows.go

@@ -63,28 +63,7 @@ func interruptibleTCPDial(addr string, config *DialConfig) (conn *TCPConn, err e
 	// Call the blocking Dial in a goroutine
 	results := conn.interruptible.results
 	go func() {
-
-		// When using an upstream HTTP proxy, first connect to the proxy,
-		// then use HTTP CONNECT to connect to the original destination.
-		dialAddr := addr
-		if config.UpstreamHttpProxyAddress != "" {
-			dialAddr = config.UpstreamHttpProxyAddress
-		}
-
-		netConn, err := net.DialTimeout("tcp", dialAddr, config.ConnectTimeout)
-
-		if config.UpstreamHttpProxyAddress != "" {
-			if err == nil {
-				err = HttpProxyConnect(netConn, addr)
-			}
-			if err != nil {
-				NoticeUpstreamProxyError(err)
-			}
-		}
-		if err != nil {
-			netConn = nil
-		}
-
+		netConn, err := net.DialTimeout("tcp", addr, config.ConnectTimeout)
 		results <- &interruptibleDialResult{netConn, err}
 	}()
 

psiphon/config.go

@@ -85,7 +85,7 @@ type Config struct {
 	ConnectionWorkerPoolSize            int
 	TunnelPoolSize                      int
 	PortForwardFailureThreshold         int
-	UpstreamHttpProxyAddress            string
+	UpstreamProxyUrl                    string
 	NetworkConnectivityChecker          NetworkConnectivityChecker
 	DeviceBinder                        DeviceBinder
 	DnsServerGetter                     DnsServerGetter

psiphon/controller.go

@@ -73,10 +73,10 @@ func NewController(config *Config) (controller *Controller, err error) {
 	// used to exclude these requests and connection from VPN routing.
 	untunneledPendingConns := new(Conns)
 	untunneledDialConfig := &DialConfig{
-		UpstreamHttpProxyAddress: config.UpstreamHttpProxyAddress,
-		PendingConns:             untunneledPendingConns,
-		DeviceBinder:             config.DeviceBinder,
-		DnsServerGetter:          config.DnsServerGetter,
+		UpstreamProxyUrl: config.UpstreamProxyUrl,
+		PendingConns:     untunneledPendingConns,
+		DeviceBinder:     config.DeviceBinder,
+		DnsServerGetter:  config.DnsServerGetter,
 	}
 
 	controller = &Controller{

psiphon/meekConn.go

@@ -33,6 +33,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/upstreamproxy"
 	"golang.org/x/crypto/nacl/box"
 )
 
@@ -73,7 +74,7 @@ type MeekConn struct {
 	url                  *url.URL
 	cookie               *http.Cookie
 	pendingConns         *Conns
-	transport            *http.Transport
+	transport            transporter
 	mutex                sync.Mutex
 	isClosed             bool
 	closedSignal         chan struct{}
@@ -87,6 +88,14 @@ type MeekConn struct {
 	fullSendBuffer       chan *bytes.Buffer
 }
 
+// transporter is implemented by both http.Transport and upstreamproxy.ProxyAuthTransport.
+type transporter interface {
+	CancelRequest(req *http.Request)
+	CloseIdleConnections()
+	RegisterProtocol(scheme string, rt http.RoundTripper)
+	RoundTrip(req *http.Request) (resp *http.Response, err error)
+}
+
 // DialMeek returns an initialized meek connection. A meek connection is
 // an HTTP session which does not depend on an underlying socket connection (although
 // persistent HTTP connections are used for performance). This function does not
@@ -161,20 +170,20 @@ func DialMeek(
 				SkipVerify:     true,
 			})
 	} else {
-		// In this case, host is both what is dialed and what ends up in the HTTP Host header
+		// In the unfronted 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)
 
-		if meekConfig.UpstreamHttpProxyAddress != "" {
+		if meekConfig.UpstreamProxyUrl != "" {
 			// 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
 			// through HTTP proxies that don't support it.
-			url, err := url.Parse(fmt.Sprintf("http://%s", meekConfig.UpstreamHttpProxyAddress))
+			url, err := url.Parse(meekConfig.UpstreamProxyUrl)
 			if err != nil {
 				return nil, ContextError(err)
 			}
 			proxyUrl = http.ProxyURL(url)
-			meekConfig.UpstreamHttpProxyAddress = ""
+			meekConfig.UpstreamProxyUrl = ""
 		}
 
 		dialer = NewTCPDialer(meekConfig)
@@ -191,11 +200,21 @@ func DialMeek(
 	if err != nil {
 		return nil, ContextError(err)
 	}
-	transport := &http.Transport{
+	httpTransport := &http.Transport{
 		Proxy: proxyUrl,
 		Dial:  dialer,
 		ResponseHeaderTimeout: MEEK_ROUND_TRIP_TIMEOUT,
 	}
+	var transport transporter
+	if proxyUrl != nil {
+		// Wrap transport with a transport that can perform HTTP proxy auth negotiation
+		transport, err = upstreamproxy.NewProxyAuthTransport(httpTransport)
+		if err != nil {
+			return nil, ContextError(err)
+		}
+	} else {
+		transport = httpTransport
+	}
 
 	// The main loop of a MeekConn is run in the relay() goroutine.
 	// A MeekConn implements net.Conn concurrency semantics:
@@ -473,7 +492,7 @@ func (meek *MeekConn) readPayload(receivedPayload io.ReadCloser) (totalSize int6
 func (meek *MeekConn) roundTrip(sendPayload []byte) (receivedPayload io.ReadCloser, err error) {
 	request, err := http.NewRequest("POST", meek.url.String(), bytes.NewReader(sendPayload))
 	if err != nil {
-		return nil, err
+		return nil, ContextError(err)
 	}
 
 	if meek.frontingAddress != "" && nil == net.ParseIP(meek.frontingAddress) {

psiphon/net.go

@@ -20,12 +20,8 @@
 package psiphon
 
 import (
-	"bufio"
-	"errors"
-	"fmt"
 	"io"
 	"net"
-	"net/http"
 	"sync"
 	"time"
 
@@ -38,10 +34,16 @@ const DNS_PORT = 53
 // of a Psiphon dialer (TCPDial, MeekDial, etc.)
 type DialConfig struct {
 
-	// UpstreamHttpProxyAddress specifies an HTTP proxy to connect through
-	// (the proxy must support HTTP CONNECT). The address may be a hostname
-	// or IP address and must include a port number.
-	UpstreamHttpProxyAddress string
+	// UpstreamProxyUrl specifies a proxy to connect through.
+	// E.g., "http://proxyhost:8080"
+	//       "socks5://user:password@proxyhost:1080"
+	//       "socks4a://proxyhost:1080"
+	//       "http://NTDOMAIN\NTUser:password@proxyhost:3375"
+	//
+	// Certain tunnel protocols require HTTP CONNECT support
+	// when a HTTP proxy is specified. If CONNECT is not
+	// supported, those protocols will not connect.
+	UpstreamProxyUrl string
 
 	ConnectTimeout time.Duration
 	ReadTimeout    time.Duration
@@ -173,39 +175,6 @@ func Relay(localConn, remoteConn net.Conn) {
 	copyWaitGroup.Wait()
 }
 
-// HttpProxyConnect establishes a HTTP CONNECT tunnel to addr through
-// an established network connection to an HTTP proxy. It is assumed that
-// no payload bytes have been sent through the connection to the proxy.
-func HttpProxyConnect(rawConn net.Conn, addr string) (err error) {
-	hostname, _, err := net.SplitHostPort(addr)
-	if err != nil {
-		return ContextError(err)
-	}
-
-	// TODO: use the proxy request/response code from net/http/transport.go?
-	connectRequest := fmt.Sprintf(
-		"CONNECT %s HTTP/1.1\r\nHost: %s\r\nConnection: Keep-Alive\r\n\r\n",
-		addr, hostname)
-	_, err = rawConn.Write([]byte(connectRequest))
-	if err != nil {
-		return ContextError(err)
-	}
-
-	// Adapted from dialConn in net/http/transport.go:
-	// Read response.
-	// Okay to use and discard buffered reader here, because
-	// TLS server will not speak until spoken to.
-	response, err := http.ReadResponse(bufio.NewReader(rawConn), nil)
-	if err != nil {
-		return ContextError(err)
-	}
-	if response.StatusCode != 200 {
-		return ContextError(errors.New(response.Status))
-	}
-
-	return nil
-}
-
 // WaitForNetworkConnectivity uses a NetworkConnectivityChecker to
 // periodically check for network connectivity. It returns true if
 // no NetworkConnectivityChecker is provided (waiting is disabled)

psiphon/tunnel.go

@@ -384,13 +384,13 @@ func dialSsh(
 
 	// Create the base transport: meek or direct connection
 	dialConfig := &DialConfig{
-		UpstreamHttpProxyAddress: config.UpstreamHttpProxyAddress,
-		ConnectTimeout:           TUNNEL_CONNECT_TIMEOUT,
-		ReadTimeout:              TUNNEL_READ_TIMEOUT,
-		WriteTimeout:             TUNNEL_WRITE_TIMEOUT,
-		PendingConns:             pendingConns,
-		DeviceBinder:             config.DeviceBinder,
-		DnsServerGetter:          config.DnsServerGetter,
+		UpstreamProxyUrl: config.UpstreamProxyUrl,
+		ConnectTimeout:   TUNNEL_CONNECT_TIMEOUT,
+		ReadTimeout:      TUNNEL_READ_TIMEOUT,
+		WriteTimeout:     TUNNEL_WRITE_TIMEOUT,
+		PendingConns:     pendingConns,
+		DeviceBinder:     config.DeviceBinder,
+		DnsServerGetter:  config.DnsServerGetter,
 	}
 	if useMeek {
 		conn, err = DialMeek(serverEntry, sessionId, frontingAddress, dialConfig)