More throttling changes in progress

- Add tunnelServer to SupportServices to make
  available to web server.
- Handhake API handler passes handshake info to
  tunnelServer/sshClient for logging, traffic rules
  filtering, and port forward enabling.
- Include handshake fields in tunnel summary log
- Require exactly one handshake API request before
  allowing port forwards.
- Replace TCPPortForwardRedirects with
  WebServerPortForwardAddress and
  WebServerPortForwardRedirectAddress: more specific
  logic ensures tunneled web server port forward is
  permitted regardless of traffic rules and handshake
  state.
- Moved shared client/serve JSON struct definitions
  to "common" package.

psiphon/common/protocol.go

@@ -76,3 +76,17 @@ func TunnelProtocolUsesMeekHTTPS(protocol string) bool {
 	return protocol == TUNNEL_PROTOCOL_FRONTED_MEEK ||
 		protocol == TUNNEL_PROTOCOL_UNFRONTED_MEEK_HTTPS
 }
+
+type HandshakeResponse struct {
+	Homepages            []string            `json:"homepages"`
+	UpgradeClientVersion string              `json:"upgrade_client_version"`
+	PageViewRegexes      []map[string]string `json:"page_view_regexes"`
+	HttpsRequestRegexes  []map[string]string `json:"https_request_regexes"`
+	EncodedServerList    []string            `json:"encoded_server_list"`
+	ClientRegion         string              `json:"client_region"`
+	ServerTimestamp      string              `json:"server_timestamp"`
+}
+
+type ConnectedResponse struct {
+	ConnectedTimestamp string `json:"connected_timestamp"`
+}

psiphon/server/api.go

@@ -133,40 +133,45 @@ func handshakeAPIRequestHandler(
 			params,
 			baseRequestParams))
 
-	// TODO: share struct definition with psiphon/serverApi.go?
-	var handshakeResponse struct {
-		Homepages            []string            `json:"homepages"`
-		UpgradeClientVersion string              `json:"upgrade_client_version"`
-		PageViewRegexes      []map[string]string `json:"page_view_regexes"`
-		HttpsRequestRegexes  []map[string]string `json:"https_request_regexes"`
-		EncodedServerList    []string            `json:"encoded_server_list"`
-		ClientRegion         string              `json:"client_region"`
-		ServerTimestamp      string              `json:"server_timestamp"`
-	}
-
-	// Ignoring errors as params are validated
+	// Note: ignoring errors as params are validated
+	sessionID, _ := getStringRequestParam(params, "client_session_id")
+	propagationChannelID, _ := getStringRequestParam(params, "propagation_channel_id")
 	sponsorID, _ := getStringRequestParam(params, "sponsor_id")
-	clientVersion, _ := getStringRequestParam(params, "client_version")
+	clientVersionStr, _ := getStringRequestParam(params, "client_version")
+	clientVersion, _ := strconv.Atoi(clientVersionStr)
 	clientPlatform, _ := getStringRequestParam(params, "client_platform")
-	clientRegion := geoIPData.Country
-
-	// Note: no guarantee that PsinetDatabase won't reload between calls
-
-	handshakeResponse.Homepages = support.PsinetDatabase.GetHomepages(
-		sponsorID, clientRegion, isMobileClientPlatform(clientPlatform))
-
-	handshakeResponse.UpgradeClientVersion = support.PsinetDatabase.GetUpgradeClientVersion(
-		clientVersion, normalizeClientPlatform(clientPlatform))
-
-	handshakeResponse.HttpsRequestRegexes = support.PsinetDatabase.GetHttpsRequestRegexes(
-		sponsorID)
-
-	handshakeResponse.EncodedServerList = support.PsinetDatabase.DiscoverServers(
-		geoIPData.DiscoveryValue)
-
-	handshakeResponse.ClientRegion = clientRegion
+	isMobile := isMobileClientPlatform(clientPlatform)
+	normalizedPlatform := normalizeClientPlatform(clientPlatform)
+
+	// Flag the SSH client as having completed its handshake. This
+	// may reselect traffic rules and starts allowing port forwards.
+
+	// TODO: in the case of SSH API requests, the actual sshClient could
+	// be passed in and used here. The session ID lookup is only strictly
+	// necessary to support web API requests.
+	err = support.TunnelServer.SetClientHandshakeState(
+		sessionID,
+		handshakeState{
+			completed:            true,
+			propagationChannelID: propagationChannelID,
+			sponsorID:            sponsorID,
+			clientVersion:        clientVersion,
+			clientPlatform:       clientPlatform,
+		})
+	if err != nil {
+		return nil, common.ContextError(err)
+	}
 
-	handshakeResponse.ServerTimestamp = common.GetCurrentTimestamp()
+	// Note: no guarantee that PsinetDatabase won't reload between database calls
+	db := support.PsinetDatabase
+	handshakeResponse := common.HandshakeResponse{
+		Homepages:            db.GetHomepages(sponsorID, geoIPData.Country, isMobile),
+		UpgradeClientVersion: db.GetUpgradeClientVersion(clientVersionStr, normalizedPlatform),
+		HttpsRequestRegexes:  db.GetHttpsRequestRegexes(sponsorID),
+		EncodedServerList:    db.DiscoverServers(geoIPData.DiscoveryValue),
+		ClientRegion:         geoIPData.Country,
+		ServerTimestamp:      common.GetCurrentTimestamp(),
+	}
 
 	responsePayload, err := json.Marshal(handshakeResponse)
 	if err != nil {
@@ -205,13 +210,10 @@ func connectedAPIRequestHandler(
 			params,
 			connectedRequestParams))
 
-	var connectedResponse struct {
-		ConnectedTimestamp string `json:"connected_timestamp"`
+	connectedResponse := common.ConnectedResponse{
+		ConnectedTimestamp: common.TruncateTimestampToHour(common.GetCurrentTimestamp()),
 	}
 
-	connectedResponse.ConnectedTimestamp =
-		common.TruncateTimestampToHour(common.GetCurrentTimestamp())
-
 	responsePayload, err := json.Marshal(connectedResponse)
 	if err != nil {
 		return nil, common.ContextError(err)
@@ -446,7 +448,7 @@ var baseRequestParams = []requestParamSpec{
 	requestParamSpec{"client_session_id", isHexDigits, requestParamOptional | requestParamNotLogged},
 	requestParamSpec{"propagation_channel_id", isHexDigits, 0},
 	requestParamSpec{"sponsor_id", isHexDigits, 0},
-	requestParamSpec{"client_version", isDigits, 0},
+	requestParamSpec{"client_version", isInt, 0},
 	requestParamSpec{"client_platform", isClientPlatform, 0},
 	requestParamSpec{"relay_protocol", isRelayProtocol, 0},
 	requestParamSpec{"tunnel_whole_device", isBooleanFlag, requestParamOptional},
@@ -730,6 +732,11 @@ func isDigits(_ *SupportServices, value string) bool {
 	})
 }
 
+func isInt(_ *SupportServices, value string) bool {
+	_, err := strconv.Atoi(value)
+	return err == nil
+}
+
 func isClientPlatform(_ *SupportServices, value string) bool {
 	return -1 == strings.IndexFunc(value, func(c rune) bool {
 		// Note: stricter than psi_web's Python string.whitespace

psiphon/server/config.go

@@ -102,6 +102,20 @@ type Config struct {
 	// authenticate itself to clients.
 	WebServerPrivateKey string
 
+	// WebServerPortForwardAddress specifies the expected network
+	// address ("<host>:<port>") specified in a client's port forward
+	// HostToConnect and PortToConnect when the client is making a
+	// tunneled connection to the web server. This address is always
+	// excepted from validation againstSSH_DISALLOWED_PORT_FORWARD_HOSTS
+	// and AllowTCPPorts/DenyTCPPorts.
+	WebServerPortForwardAddress string
+
+	// WebServerPortForwardRedirectAddress specifies an alternate
+	// destination address to be substituted and dialed instead of
+	// the original destination when the port forward destination is
+	// WebServerPortForwardAddress.
+	WebServerPortForwardRedirectAddress string
+
 	// TunnelProtocolPorts specifies which tunnel protocols to run
 	// and which ports to listen on for each protocol. Valid tunnel
 	// protocols include: "SSH", "OSSH", "UNFRONTED-MEEK-OSSH",
@@ -186,21 +200,6 @@ type Config struct {
 	// "nameserver" entry.
 	DNSResolverIPAddress string
 
-	// TCPPortForwardRedirects is a mapping from client port forward
-	// destination to an alternate destination address. When the client's
-	// port forward HostToConnect and PortToConnect matches a redirect,
-	// the redirect is substituted and dialed instead of the original
-	// destination.
-	//
-	// The redirect is applied after the original destination is
-	// validated against SSH_DISALLOWED_PORT_FORWARD_HOSTS and
-	// AllowTCPPorts/DenyTCPPorts. So the redirect may map to any
-	// otherwise prohibited destination.
-	//
-	// The redirect is applied after UDPInterceptUdpgwServerAddress is
-	// checked. So the redirect address will not be intercepted.
-	TCPPortForwardRedirects map[string]string
-
 	// LoadMonitorPeriodSeconds indicates how frequently to log server
 	// load information (number of connected clients per tunnel protocol,
 	// number of running goroutines, amount of memory allocated, etc.)
@@ -233,7 +232,7 @@ func LoadConfig(configJSON []byte) (*Config, error) {
 	}
 
 	if config.ServerIPAddress == "" {
-		return nil, errors.New("ServerIPAddress is missing from config file")
+		return nil, errors.New("ServerIPAddress is required")
 	}
 
 	if config.WebServerPort > 0 && (config.WebServerSecret == "" || config.WebServerCertificate == "" ||
@@ -243,6 +242,24 @@ func LoadConfig(configJSON []byte) (*Config, error) {
 			"Web server requires WebServerSecret, WebServerCertificate, WebServerPrivateKey")
 	}
 
+	if config.WebServerPortForwardAddress != "" {
+		if err := validateNetworkAddress(config.WebServerPortForwardAddress, false); err != nil {
+			return nil, errors.New("WebServerPortForwardAddress is invalid")
+		}
+	}
+
+	if config.WebServerPortForwardRedirectAddress != "" {
+
+		if config.WebServerPortForwardAddress == "" {
+			return nil, errors.New(
+				"WebServerPortForwardRedirectAddress requires WebServerPortForwardAddress")
+		}
+
+		if err := validateNetworkAddress(config.WebServerPortForwardRedirectAddress, false); err != nil {
+			return nil, errors.New("WebServerPortForwardRedirectAddress is invalid")
+		}
+	}
+
 	for tunnelProtocol, _ := range config.TunnelProtocolPorts {
 		if !common.Contains(common.SupportedTunnelProtocols, tunnelProtocol) {
 			return nil, fmt.Errorf("Unsupported tunnel protocol: %s", tunnelProtocol)
@@ -280,24 +297,6 @@ func LoadConfig(configJSON []byte) (*Config, error) {
 		}
 	}
 
-	validateNetworkAddress := func(address string, requireIPaddress bool) error {
-		host, portStr, err := net.SplitHostPort(address)
-		if err != nil {
-			return err
-		}
-		if requireIPaddress && net.ParseIP(host) == nil {
-			return errors.New("host must be an IP address")
-		}
-		port, err := strconv.Atoi(portStr)
-		if err != nil {
-			return err
-		}
-		if port < 0 || port > 65535 {
-			return errors.New("invalid port")
-		}
-		return nil
-	}
-
 	if config.UDPInterceptUdpgwServerAddress != "" {
 		if err := validateNetworkAddress(config.UDPInterceptUdpgwServerAddress, true); err != nil {
 			return nil, fmt.Errorf("UDPInterceptUdpgwServerAddress is invalid: %s", err)
@@ -310,20 +309,27 @@ func LoadConfig(configJSON []byte) (*Config, error) {
 		}
 	}
 
-	if config.TCPPortForwardRedirects != nil {
-		for destination, redirect := range config.TCPPortForwardRedirects {
-			if err := validateNetworkAddress(destination, false); err != nil {
-				return nil, fmt.Errorf("TCPPortForwardRedirects destination %s is invalid: %s", destination, err)
-			}
-			if err := validateNetworkAddress(redirect, false); err != nil {
-				return nil, fmt.Errorf("TCPPortForwardRedirects redirect %s is invalid: %s", redirect, err)
-			}
-		}
-	}
-
 	return &config, nil
 }
 
+func validateNetworkAddress(address string, requireIPaddress bool) error {
+	host, portStr, err := net.SplitHostPort(address)
+	if err != nil {
+		return err
+	}
+	if requireIPaddress && net.ParseIP(host) == nil {
+		return errors.New("host must be an IP address")
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		return err
+	}
+	if port < 0 || port > 65535 {
+		return errors.New("invalid port")
+	}
+	return nil
+}
+
 // GenerateConfigParams specifies customizations to be applied to
 // a generated server config.
 type GenerateConfigParams struct {
@@ -395,6 +401,9 @@ func GenerateConfig(params *GenerateConfigParams) ([]byte, []byte, []byte, error
 		}
 	}
 
+	webServerPortForwardAddress := net.JoinHostPort(
+		params.ServerIPAddress, strconv.Itoa(params.WebServerPort))
+
 	// SSH config
 
 	// TODO: use other key types: anti-fingerprint by varying params
@@ -482,6 +491,7 @@ func GenerateConfig(params *GenerateConfigParams) ([]byte, []byte, []byte, error
 		WebServerSecret:                webServerSecret,
 		WebServerCertificate:           webServerCertificate,
 		WebServerPrivateKey:            webServerPrivateKey,
+		WebServerPortForwardAddress:    webServerPortForwardAddress,
 		SSHPrivateKey:                  string(sshPrivateKey),
 		SSHServerVersion:               sshServerVersion,
 		SSHUserName:                    sshUserName,

psiphon/server/services.go

@@ -70,6 +70,8 @@ func RunServices(configJSON []byte) error {
 		return common.ContextError(err)
 	}
 
+	supportServices.TunnelServer = tunnelServer
+
 	if config.RunLoadMonitor() {
 		waitGroup.Add(1)
 		go func() {
@@ -190,6 +192,7 @@ type SupportServices struct {
 	PsinetDatabase  *psinet.Database
 	GeoIPService    *GeoIPService
 	DNSResolver     *DNSResolver
+	TunnelServer    *TunnelServer
 }
 
 // NewSupportServices initializes a new SupportServices.

psiphon/server/tunnelServer.go

@@ -198,10 +198,16 @@ func (server *TunnelServer) SelectAllClientTrafficRules() {
 	server.sshServer.selectAllClientTrafficRules()
 }
 
-// SelectClientTrafficRules resets a specified client's traffic rules
-// to use the latest server config and client state.
-func (server *TunnelServer) SelectClientTrafficRules(sessionID string) {
-	server.sshServer.selectClientTrafficRules(sessionID)
+// SetClientHandshakeState sets the handshake state -- that it completed and
+// what paramaters were passed -- in sshClient. This state is used for allowing
+// port forwards and for future traffic rule selection. SetClientHandshakeState
+// also triggers an immediate traffic rule re-selection, as the rules selected
+// upon tunnel establishment may no longer apply now that handshake values are
+// set.
+func (server *TunnelServer) SetClientHandshakeState(
+	sessionID string, state handshakeState) error {
+
+	return server.sshServer.setClientHandshakeState(sessionID, state)
 }
 
 type sshServer struct {
@@ -432,15 +438,25 @@ func (sshServer *sshServer) selectAllClientTrafficRules() {
 	}
 }
 
-func (sshServer *sshServer) selectClientTrafficRules(sessionID string) {
+func (sshServer *sshServer) setClientHandshakeState(
+	sessionID string, state handshakeState) error {
 
 	sshServer.clientsMutex.Lock()
 	client := sshServer.clients[sessionID]
 	sshServer.clientsMutex.Unlock()
 
-	if client != nil {
-		client.selectTrafficRules()
+	if client == nil {
+		return common.ContextError(errors.New("unknown session ID"))
+	}
+
+	err := client.setHandshakeState(state)
+	if err != nil {
+		return common.ContextError(err)
 	}
+
+	client.selectTrafficRules()
+
+	return nil
 }
 
 func (sshServer *sshServer) stopClients() {
@@ -593,6 +609,7 @@ type sshClient struct {
 	throttledConn           *common.ThrottledConn
 	geoIPData               GeoIPData
 	sessionID               string
+	handshakeState          handshakeState
 	udpChannel              ssh.Channel
 	trafficRules            TrafficRules
 	tcpTrafficState         trafficState
@@ -613,6 +630,14 @@ type trafficState struct {
 	totalPortForwardCount          int64
 }
 
+type handshakeState struct {
+	completed            bool
+	propagationChannelID string
+	sponsorID            string
+	clientVersion        int
+	clientPlatform       string
+}
+
 func newSshClient(
 	sshServer *sshServer, tunnelProtocol string, geoIPData GeoIPData) *sshClient {
 	return &sshClient{
@@ -734,9 +759,17 @@ func (sshClient *sshClient) stop() {
 	// it did the client may not have the opportunity to send a final
 	// request with an EOF flag set.)
 
+	// TODO: match legacy log field naming convention?
+	// TODO: log all handshake common inputs?
+
 	sshClient.Lock()
 	log.WithContextFields(
 		LogFields{
+			"handshakeCompleted":                sshClient.handshakeState.completed,
+			"propagationChannelID":              sshClient.handshakeState.propagationChannelID,
+			"sponsorID":                         sshClient.handshakeState.sponsorID,
+			"clientVersion":                     sshClient.handshakeState.clientVersion,
+			"clientPlatform":                    sshClient.handshakeState.clientPlatform,
 			"startTime":                         sshClient.activityConn.GetStartTime(),
 			"duration":                          sshClient.activityConn.GetActiveDuration(),
 			"sessionID":                         sshClient.sessionID,
@@ -854,6 +887,25 @@ func (sshClient *sshClient) handleNewPortForwardChannel(newChannel ssh.NewChanne
 	}
 }
 
+// setHandshakeState records that a client has completed a handshake API request.
+// Some parameters from the handshake request may be used in future traffic rule
+// selection. Port forwards are disallowed until a handshake is complete. The
+// handshake parameters are included in the session summary log recorded in
+// sshClient.stop().
+func (sshClient *sshClient) setHandshakeState(state handshakeState) error {
+	sshClient.Lock()
+	defer sshClient.Unlock()
+
+	// Client must only perform one handshake
+	if sshClient.handshakeState.completed {
+		return common.ContextError(errors.New("handshake already completed"))
+	}
+
+	sshClient.handshakeState = state
+
+	return nil
+}
+
 // selectTrafficRules resets the client's traffic rules based on the latest server config
 // and client state. As sshClient.trafficRules may be reset by a concurrent goroutine,
 // trafficRules must only be accessed within the sshClient mutex.
@@ -861,7 +913,9 @@ func (sshClient *sshClient) selectTrafficRules() {
 	sshClient.Lock()
 	defer sshClient.Unlock()
 
-	sshClient.trafficRules = sshClient.sshServer.support.TrafficRulesSet.GetTrafficRules(sshClient.geoIPData.Country)
+	sshClient.trafficRules = sshClient.sshServer.support.TrafficRulesSet.GetTrafficRules(
+		// TODO: sshClient.geoIPData, sshClient.handshakeState)
+		sshClient.geoIPData.Country)
 }
 
 func (sshClient *sshClient) rateLimits() common.RateLimits {
@@ -894,10 +948,13 @@ const (
 func (sshClient *sshClient) isPortForwardPermitted(
 	portForwardType int, host string, port int) bool {
 
-	// Mutex required for accessing sshClient.trafficRules
 	sshClient.Lock()
 	defer sshClient.Unlock()
 
+	if !sshClient.handshakeState.completed {
+		return false
+	}
+
 	if common.Contains(SSH_DISALLOWED_PORT_FORWARD_HOSTS, host) {
 		return false
 	}
@@ -998,7 +1055,23 @@ func (sshClient *sshClient) handleTCPChannel(
 	portToConnect int,
 	newChannel ssh.NewChannel) {
 
-	if !sshClient.isPortForwardPermitted(
+	isWebServerPortForward := false
+	config := sshClient.sshServer.support.Config
+	if config.WebServerPortForwardAddress != "" {
+		destination := net.JoinHostPort(hostToConnect, strconv.Itoa(portToConnect))
+		if destination == config.WebServerPortForwardAddress {
+			isWebServerPortForward = true
+			if config.WebServerPortForwardRedirectAddress != "" {
+				// Note: redirect format is validated when config is loaded
+				host, portStr, _ := net.SplitHostPort(config.WebServerPortForwardRedirectAddress)
+				port, _ := strconv.Atoi(portStr)
+				hostToConnect = host
+				portToConnect = port
+			}
+		}
+	}
+
+	if !isWebServerPortForward && !sshClient.isPortForwardPermitted(
 		portForwardTypeTCP, hostToConnect, portToConnect) {
 
 		sshClient.rejectNewChannel(
@@ -1006,19 +1079,6 @@ func (sshClient *sshClient) handleTCPChannel(
 		return
 	}
 
-	// Note: redirects are applied *after* isPortForwardPermitted allows the original destination
-	if sshClient.sshServer.support.Config.TCPPortForwardRedirects != nil {
-		destination := net.JoinHostPort(hostToConnect, strconv.Itoa(portToConnect))
-		if redirect, ok := sshClient.sshServer.support.Config.TCPPortForwardRedirects[destination]; ok {
-			// Note: redirect format is validated when config is loaded
-			host, portStr, _ := net.SplitHostPort(redirect)
-			port, _ := strconv.Atoi(portStr)
-			hostToConnect = host
-			portToConnect = port
-			log.WithContextFields(LogFields{"destination": destination, "redirect": redirect}).Debug("port forward redirect")
-		}
-	}
-
 	var bytesUp, bytesDown int64
 	sshClient.openedPortForward(portForwardTypeTCP)
 	defer func() {

psiphon/server/webServer.go

@@ -36,8 +36,9 @@ import (
 const WEB_SERVER_IO_TIMEOUT = 10 * time.Second
 
 type webServer struct {
-	support  *SupportServices
-	serveMux *http.ServeMux
+	support      *SupportServices
+	tunnelServer *TunnelServer
+	serveMux     *http.ServeMux
 }
 
 // RunWebServer runs a web server which supports tunneled and untunneled

psiphon/serverApi.go

@@ -167,18 +167,11 @@ func (serverContext *ServerContext) doHandshakeRequest() error {
 		}
 	}
 
-	// Note:
-	// - 'preemptive_reconnect_lifetime_milliseconds' is currently unused
+	// Legacy fields:
+	// - 'preemptive_reconnect_lifetime_milliseconds' is unused and ignored
 	// - 'ssh_session_id' is ignored; client session ID is used instead
-	var handshakeResponse struct {
-		Homepages            []string            `json:"homepages"`
-		UpgradeClientVersion string              `json:"upgrade_client_version"`
-		PageViewRegexes      []map[string]string `json:"page_view_regexes"`
-		HttpsRequestRegexes  []map[string]string `json:"https_request_regexes"`
-		EncodedServerList    []string            `json:"encoded_server_list"`
-		ClientRegion         string              `json:"client_region"`
-		ServerTimestamp      string              `json:"server_timestamp"`
-	}
+
+	var handshakeResponse common.HandshakeResponse
 	err := json.Unmarshal(response, &handshakeResponse)
 	if err != nil {
 		return common.ContextError(err)
@@ -292,9 +285,7 @@ func (serverContext *ServerContext) DoConnectedRequest() error {
 		}
 	}
 
-	var connectedResponse struct {
-		ConnectedTimestamp string `json:"connected_timestamp"`
-	}
+	var connectedResponse common.ConnectedResponse
 	err = json.Unmarshal(response, &connectedResponse)
 	if err != nil {
 		return common.ContextError(err)