Merge branch 'master' into staging-server

MobileLibrary/Android/SampleApps/TunneledWebView/app/src/main/java/ca/psiphon/tunneledwebview/MainActivity.java

@@ -166,26 +166,11 @@ public void run() {
     // NOTE: these are callbacks from the Psiphon Library
     //----------------------------------------------------------------------------------------------
 
-    @Override
-    public String getAppName() {
-        return "TunneledWebView Sample";
-    }
-
     @Override
     public Context getContext() {
         return this;
     }
 
-    @Override
-    public Object getVpnService() {
-        return null;
-    }
-
-    @Override
-    public Object newVpnServiceBuilder() {
-        return null;
-    }
-
     @Override
     public String getPsiphonConfig() {
         try {

MobileLibrary/Android/make.bash

@@ -48,10 +48,6 @@ fi
 mkdir -p build-tmp/psi
 unzip -o psi.aar -d build-tmp/psi
 yes | cp -f PsiphonTunnel/AndroidManifest.xml build-tmp/psi/AndroidManifest.xml
-yes | cp -f PsiphonTunnel/libs/armeabi-v7a/libtun2socks.so build-tmp/psi/jni/armeabi-v7a/libtun2socks.so
-yes | cp -f PsiphonTunnel/libs/arm64-v8a/libtun2socks.so build-tmp/psi/jni/arm64-v8a/libtun2socks.so
-yes | cp -f PsiphonTunnel/libs/x86/libtun2socks.so build-tmp/psi/jni/x86/libtun2socks.so
-yes | cp -f PsiphonTunnel/libs/x86_64/libtun2socks.so build-tmp/psi/jni/x86_64/libtun2socks.so
 mkdir -p build-tmp/psi/res/xml
 yes | cp -f PsiphonTunnel/ca_psiphon_psiphontunnel_backup_rules.xml build-tmp/psi/res/xml/ca_psiphon_psiphontunnel_backup_rules.xml
 

MobileLibrary/iOS/PsiphonTunnel/PsiphonTunnel/PsiphonTunnel.m

@@ -1199,7 +1199,7 @@ - (void)handlePsiphonNotice:(NSString * _Nonnull)noticeJSON {
     else if ([noticeType isEqualToString:@"ConnectedServerRegion"]) {
         id region = [notice valueForKeyPath:@"data.serverRegion"];
         if (![region isKindOfClass:[NSString class]]) {
-            [self logMessage:[NSString stringWithFormat: @"ActiveTunnel notice missing data.serverRegion: %@", noticeJSON]];
+            [self logMessage:[NSString stringWithFormat: @"ConnectedServerRegion notice missing data.serverRegion: %@", noticeJSON]];
             return;
         }
         if ([self.tunneledAppDelegate respondsToSelector:@selector(onConnectedServerRegion:)]) {

psiphon/common/api.go

@@ -24,6 +24,17 @@ package common
 // values are of varying types: strings, ints, arrays, structs, etc.
 type APIParameters map[string]interface{}
 
+// Add copies API parameters from b to a, skipping parameters which already
+// exist, regardless of value, in a.
+func (a APIParameters) Add(b APIParameters) {
+	for name, value := range b {
+		_, ok := a[name]
+		if !ok {
+			a[name] = value
+		}
+	}
+}
+
 // APIParameterValidator is a function that validates API parameters
 // for a particular request or context.
 type APIParameterValidator func(APIParameters) error

psiphon/common/inproxy/api.go

@@ -215,19 +215,26 @@ type ClientMetrics struct {
 // ProxyAnnounceRequest is an API request sent from a proxy to a broker,
 // announcing that it is available for a client connection. Proxies send one
 // ProxyAnnounceRequest for each available client connection. The broker will
-// match the proxy with a a client and return WebRTC connection information
+// match the proxy with a client and return WebRTC connection information
 // in the response.
 //
 // PersonalCompartmentIDs limits the clients to those that supply one of the
 // specified compartment IDs; personal compartment IDs are distributed from
 // proxy operators to client users out-of-band and provide optional access
 // control.
 //
+// When CheckTactics is set, the broker will check for new tactics or indicate
+// that the proxy's cached tactics TTL may be extended. Tactics information
+// is returned in the response TacticsPayload. To minimize broker processing
+// overhead, proxies with multiple workers should designate just one worker
+// to set CheckTactics.
+//
 // The proxy's session public key is an implicit and cryptographically
 // verified proxy ID.
 type ProxyAnnounceRequest struct {
 	PersonalCompartmentIDs []ID          `cbor:"1,keyasint,omitempty"`
 	Metrics                *ProxyMetrics `cbor:"2,keyasint,omitempty"`
+	CheckTactics           bool          `cbor:"3,keyasint,omitempty"`
 }
 
 // WebRTCSessionDescription is compatible with pion/webrtc.SessionDescription

psiphon/common/inproxy/broker.go

@@ -122,6 +122,15 @@ type BrokerConfig struct {
 	// clients. Proxies with personal compartment IDs are always allowed.
 	AllowProxy func(common.GeoIPData) bool
 
+	// PrioritizeProxy is a callback which can indicate whether proxy
+	// announcements from proxies with the specified GeoIPData and
+	// APIParameters should be prioritized in the matcher queue. Priority
+	// proxy announcements match ahead of other proxy announcements,
+	// regardless of announcement age/deadline. Priority status takes
+	// precedence over preferred NAT matching. Prioritization applies only to
+	// common compartment IDs and not personal pairing mode.
+	PrioritizeProxy func(common.GeoIPData, common.APIParameters) bool
+
 	// AllowClient is a callback which can indicate whether a client with the
 	// given GeoIP data is allowed to match with common compartment ID
 	// proxies. Clients are always allowed to match based on personal
@@ -155,6 +164,14 @@ type BrokerConfig struct {
 	// entry tags.
 	IsValidServerEntryTag func(serverEntryTag string) bool
 
+	// IsLoadLimiting is a callback which checks if the broker process is in a
+	// load limiting state, where consumed resources, including allocated
+	// system memory and CPU load, exceed determined thresholds. When load
+	// limiting is indicated, the broker will attempt to reduce load by
+	// immediately rejecting either proxy announces or client offers,
+	// depending on the state of the corresponding queues.
+	IsLoadLimiting func() bool
+
 	// PrivateKey is the broker's secure session long term private key.
 	PrivateKey SessionPrivateKey
 
@@ -233,6 +250,8 @@ func NewBroker(config *BrokerConfig) (*Broker, error) {
 			OfferLimitEntryCount:           config.MatcherOfferLimitEntryCount,
 			OfferRateLimitQuantity:         config.MatcherOfferRateLimitQuantity,
 			OfferRateLimitInterval:         config.MatcherOfferRateLimitInterval,
+
+			IsLoadLimiting: config.IsLoadLimiting,
 		}),
 
 		proxyAnnounceTimeout:       int64(config.ProxyAnnounceTimeout),
@@ -548,6 +567,8 @@ func (b *Broker) handleProxyAnnounce(
 		return nil, errors.Trace(err)
 	}
 
+	hasPersonalCompartmentIDs := len(announceRequest.PersonalCompartmentIDs) > 0
+
 	// Return MustUpgrade when the proxy's protocol version is less than the
 	// minimum required.
 	if announceRequest.Metrics.ProxyProtocolVersion < MinimumProxyProtocolVersion {
@@ -573,30 +594,33 @@ func (b *Broker) handleProxyAnnounce(
 	// proxy can store and apply the new tactics before announcing again.
 
 	var tacticsPayload []byte
-	tacticsPayload, newTacticsTag, err = b.config.GetTacticsPayload(geoIPData, apiParams)
-	if err != nil {
-		return nil, errors.Trace(err)
-	}
-
-	if tacticsPayload != nil && newTacticsTag != "" {
-		responsePayload, err := MarshalProxyAnnounceResponse(
-			&ProxyAnnounceResponse{
-				TacticsPayload: tacticsPayload,
-				NoMatch:        true,
-			})
+	if announceRequest.CheckTactics {
+		tacticsPayload, newTacticsTag, err =
+			b.config.GetTacticsPayload(geoIPData, apiParams)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
 
-		return responsePayload, nil
+		if tacticsPayload != nil && newTacticsTag != "" {
+			responsePayload, err := MarshalProxyAnnounceResponse(
+				&ProxyAnnounceResponse{
+					TacticsPayload: tacticsPayload,
+					NoMatch:        true,
+				})
+			if err != nil {
+				return nil, errors.Trace(err)
+			}
+
+			return responsePayload, nil
+		}
 	}
 
 	// AllowProxy may be used to disallow proxies from certain geolocations,
 	// such as censored locations, from announcing. Proxies with personal
 	// compartment IDs are always allowed, as they will be used only by
 	// clients specifically configured to use them.
 
-	if len(announceRequest.PersonalCompartmentIDs) == 0 &&
+	if !hasPersonalCompartmentIDs &&
 		!b.config.AllowProxy(geoIPData) {
 
 		return nil, errors.TraceNew("proxy disallowed")
@@ -608,14 +632,36 @@ func (b *Broker) handleProxyAnnounce(
 	// assigned to the same compartment.
 
 	var commonCompartmentIDs []ID
-	if len(announceRequest.PersonalCompartmentIDs) == 0 {
+	if !hasPersonalCompartmentIDs {
 		compartmentID, err := b.selectCommonCompartmentID(proxyID)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
 		commonCompartmentIDs = []ID{compartmentID}
 	}
 
+	// In the common compartment ID case, invoke the callback to check if the
+	// announcement should be prioritized.
+
+	isPriority := false
+	if b.config.PrioritizeProxy != nil && !hasPersonalCompartmentIDs {
+
+		// Limitation: Of the two return values from
+		// ValidateAndGetParametersAndLogFields, apiParams and logFields,
+		// only logFields contains fields such as max_clients
+		// and *_bytes_per_second, and so these cannot be part of any
+		// filtering performed by the PrioritizeProxy callback.
+		//
+		// TODO: include the additional fields in logFields. Since the
+		// logFields return value is the output of server.getRequestLogFields
+		// processing, it's not safe to use it directly. In addition,
+		// filtering by fields such as max_clients and *_bytes_per_second
+		// calls for range filtering, which is not yet supported in the
+		// psiphon/server.MeekServer PrioritizeProxy provider.
+
+		isPriority = b.config.PrioritizeProxy(geoIPData, apiParams)
+	}
+
 	// Await client offer.
 
 	timeout := common.ValueOrDefault(
@@ -638,6 +684,7 @@ func (b *Broker) handleProxyAnnounce(
 		proxyIP,
 		&MatchAnnouncement{
 			Properties: MatchProperties{
+				IsPriority:             isPriority,
 				CommonCompartmentIDs:   commonCompartmentIDs,
 				PersonalCompartmentIDs: announceRequest.PersonalCompartmentIDs,
 				GeoIPData:              geoIPData,
@@ -830,23 +877,19 @@ func (b *Broker) handleClientOffer(
 		return nil, errors.Trace(err)
 	}
 
+	hasPersonalCompartmentIDs := len(offerRequest.PersonalCompartmentIDs) > 0
+
 	offerSDP := offerRequest.ClientOfferSDP
 	offerSDP.SDP = string(filteredSDP)
 
 	// AllowClient may be used to disallow clients from certain geolocations
 	// from offering. Clients are always allowed to match proxies with shared
 	// personal compartment IDs.
 
-	commonCompartmentIDs := offerRequest.CommonCompartmentIDs
-
-	if !b.config.AllowClient(geoIPData) {
-
-		if len(offerRequest.PersonalCompartmentIDs) == 0 {
-			return nil, errors.TraceNew("client disallowed")
-		}
+	if !hasPersonalCompartmentIDs &&
+		!b.config.AllowClient(geoIPData) {
 
-		// Only match personal compartment IDs.
-		commonCompartmentIDs = nil
+		return nil, errors.TraceNew("client disallowed")
 	}
 
 	// Validate that the proxy destination specified by the client is a valid
@@ -884,7 +927,7 @@ func (b *Broker) handleClientOffer(
 	// personal pairing mode, to facilitate a faster no-match result and
 	// resulting broker rotation.
 	var timeout time.Duration
-	if len(offerRequest.PersonalCompartmentIDs) > 0 {
+	if hasPersonalCompartmentIDs {
 		timeout = time.Duration(atomic.LoadInt64(&b.clientOfferPersonalTimeout))
 	} else {
 		timeout = time.Duration(atomic.LoadInt64(&b.clientOfferTimeout))
@@ -901,7 +944,7 @@ func (b *Broker) handleClientOffer(
 
 	clientMatchOffer = &MatchOffer{
 		Properties: MatchProperties{
-			CommonCompartmentIDs:   commonCompartmentIDs,
+			CommonCompartmentIDs:   offerRequest.CommonCompartmentIDs,
 			PersonalCompartmentIDs: offerRequest.PersonalCompartmentIDs,
 			GeoIPData:              geoIPData,
 			NetworkType:            GetNetworkType(offerRequest.Metrics.BaseAPIParameters),

psiphon/common/inproxy/client.go

@@ -356,7 +356,11 @@ func dialClientWebRTCConn(
 
 	// Send the ClientOffer request to the broker
 
-	packedBaseParams, err := protocol.EncodePackedAPIParameters(config.BaseAPIParameters)
+	apiParams := common.APIParameters{}
+	apiParams.Add(config.BaseAPIParameters)
+	apiParams.Add(common.APIParameters(brokerCoordinator.MetricsForBrokerRequests()))
+
+	packedParams, err := protocol.EncodePackedAPIParameters(apiParams)
 	if err != nil {
 		return nil, false, errors.Trace(err)
 	}
@@ -372,7 +376,7 @@ func dialClientWebRTCConn(
 		ctx,
 		&ClientOfferRequest{
 			Metrics: &ClientMetrics{
-				BaseAPIParameters:    packedBaseParams,
+				BaseAPIParameters:    packedParams,
 				ProxyProtocolVersion: proxyProtocolVersion,
 				NATType:              config.WebRTCDialCoordinator.NATType(),
 				PortMappingTypes:     config.WebRTCDialCoordinator.PortMappingTypes(),

psiphon/common/inproxy/coordinator.go

@@ -23,6 +23,8 @@ import (
 	"context"
 	"net"
 	"time"
+
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common"
 )
 
 // RoundTripper provides a request/response round trip network transport with
@@ -170,6 +172,13 @@ type BrokerDialCoordinator interface {
 	// proxies are not well balanced across brokers.
 	BrokerClientNoMatch(roundTripper RoundTripper)
 
+	// MetricsForBrokerRequests returns the metrics, associated with the
+	// broker client instance, which are to be added to the base API
+	// parameters included in client and proxy requests sent to the broker.
+	// This includes fronting_provider_id, which varies depending on the
+	// broker client dial and isn't a fixed base API parameter value.
+	MetricsForBrokerRequests() common.LogFields
+
 	SessionHandshakeRoundTripTimeout() time.Duration
 	AnnounceRequestTimeout() time.Duration
 	AnnounceDelay() time.Duration

psiphon/common/inproxy/coordinator_test.go

@@ -46,6 +46,7 @@ type testBrokerDialCoordinator struct {
 	brokerClientRoundTripperSucceeded func(RoundTripper)
 	brokerClientRoundTripperFailed    func(RoundTripper)
 	brokerClientNoMatch               func(RoundTripper)
+	metricsForBrokerRequests          common.LogFields
 	sessionHandshakeRoundTripTimeout  time.Duration
 	announceRequestTimeout            time.Duration
 	announceDelay                     time.Duration
@@ -124,6 +125,12 @@ func (t *testBrokerDialCoordinator) BrokerClientNoMatch(roundTripper RoundTrippe
 	t.brokerClientNoMatch(roundTripper)
 }
 
+func (t *testBrokerDialCoordinator) MetricsForBrokerRequests() common.LogFields {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	return t.metricsForBrokerRequests
+}
+
 func (t *testBrokerDialCoordinator) SessionHandshakeRoundTripTimeout() time.Duration {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()