Add AlertRequest/NoticeServerAlert

psiphon/common/protocol/protocol.go

@@ -59,6 +59,10 @@ const (
 	PSIPHON_API_CONNECTED_REQUEST_NAME = "psiphon-connected"
 	PSIPHON_API_STATUS_REQUEST_NAME    = "psiphon-status"
 	PSIPHON_API_OSL_REQUEST_NAME       = "psiphon-osl"
+	PSIPHON_API_ALERT_REQUEST_NAME     = "psiphon-alert"
+
+	PSIPHON_API_ALERT_DISALLOWED_TRAFFIC = "disallowed-traffic"
+	PSIPHON_API_ALERT_UNSAFE_TRAFFIC     = "unsafe-traffic"
 
 	// PSIPHON_API_CLIENT_VERIFICATION_REQUEST_NAME may still be used by older Android clients
 	PSIPHON_API_CLIENT_VERIFICATION_REQUEST_NAME = "psiphon-client-verification"
@@ -384,6 +388,11 @@ type RandomStreamRequest struct {
 	DownstreamBytes int `json:"d"`
 }
 
+type AlertRequest struct {
+	Reason  string `json:"reason"`
+	Subject string `json:"subject"`
+}
+
 func DeriveSSHServerKEXPRNGSeed(obfuscatedKey string) (*prng.Seed, error) {
 	// By convention, the obfuscatedKey will often be a hex-encoded 32 byte value,
 	// but this isn't strictly required or validated, so we use SHA256 to map the

psiphon/notice.go

@@ -35,6 +35,7 @@ import (
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/buildinfo"
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/errors"
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/parameters"
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/protocol"
 	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/stacktrace"
 )
 
@@ -848,6 +849,14 @@ func NoticeApplicationParameters(keyValues parameters.KeyValues) {
 	}
 }
 
+// NoticeServerAlert reports server alerts. Each distinct server alert is
+// reported at most once per session.
+func NoticeServerAlert(alert protocol.AlertRequest) {
+	outputRepetitiveNotice(
+		"ServerAlert", fmt.Sprintf("%+v", alert), 0,
+		"ServerAlert", noticeIsDiagnostic, "reason", alert.Reason, "subject", alert.Subject)
+}
+
 type repetitiveNoticeState struct {
 	message string
 	repeats int

psiphon/server/server_test.go

@@ -926,14 +926,16 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 		t.Fatalf("error creating client controller: %s", err)
 	}
 
+	connectedServer := make(chan struct{}, 1)
 	tunnelsEstablished := make(chan struct{}, 1)
 	homepageReceived := make(chan struct{}, 1)
 	slokSeeded := make(chan struct{}, 1)
-	clientConnectedNotice := make(chan map[string]interface{}, 1)
 
 	numPruneNotices := 0
 	pruneServerEntriesNoticesEmitted := make(chan struct{}, 1)
 
+	serverAlertDisallowedNoticesEmitted := make(chan struct{}, 1)
+
 	psiphon.SetNoticeWriter(psiphon.NewNoticeReceiver(
 		func(notice []byte) {
 
@@ -946,6 +948,9 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 
 			switch noticeType {
 
+			case "ConnectedServer":
+				sendNotificationReceived(connectedServer)
+
 			case "Tunnels":
 				count := int(payload["count"].(float64))
 				if count >= numTunnels {
@@ -969,10 +974,10 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 					sendNotificationReceived(pruneServerEntriesNoticesEmitted)
 				}
 
-			case "ConnectedServer":
-				select {
-				case clientConnectedNotice <- payload:
-				default:
+			case "ServerAlert":
+				reason := payload["reason"].(string)
+				if reason == protocol.PSIPHON_API_ALERT_DISALLOWED_TRAFFIC {
+					sendNotificationReceived(serverAlertDisallowedNoticesEmitted)
 				}
 			}
 		}))
@@ -1022,7 +1027,8 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 		close(timeoutSignal)
 	}()
 
-	waitOnNotification(t, tunnelsEstablished, timeoutSignal, "tunnel establish timeout exceeded")
+	waitOnNotification(t, connectedServer, timeoutSignal, "connected server timeout exceeded")
+	waitOnNotification(t, tunnelsEstablished, timeoutSignal, "tunnel established timeout exceeded")
 	waitOnNotification(t, homepageReceived, timeoutSignal, "homepage received timeout exceeded")
 
 	expectTrafficFailure := runConfig.denyTrafficRules || (runConfig.omitAuthorization && runConfig.requireAuthorization)
@@ -1064,17 +1070,24 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 		}
 	}
 
-	// Test: await SLOK payload
+	// Test: await SLOK payload or server alert notice
+
+	time.Sleep(1 * time.Second)
 
 	if !expectTrafficFailure {
 
-		time.Sleep(1 * time.Second)
 		waitOnNotification(t, slokSeeded, timeoutSignal, "SLOK seeded timeout exceeded")
 
 		numSLOKs := psiphon.CountSLOKs()
 		if numSLOKs != expectedNumSLOKs {
 			t.Fatalf("unexpected number of SLOKs: %d", numSLOKs)
 		}
+
+	} else {
+
+		// Note: in expectTrafficFailure case, timeoutSignal may have already fired.
+
+		waitOnNotification(t, serverAlertDisallowedNoticesEmitted, nil, "")
 	}
 
 	// Test: await expected prune server entry notices
@@ -1082,12 +1095,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 	// Note: will take up to PsiphonAPIStatusRequestShortPeriodMax to emit.
 
 	if expectedNumPruneNotices > 0 {
-
-		waitOnNotification(
-			t,
-			pruneServerEntriesNoticesEmitted,
-			timeoutSignal,
-			"prune server entries timeout exceeded")
+		waitOnNotification(t, pruneServerEntriesNoticesEmitted, nil, "")
 	}
 
 	if runConfig.doDanglingTCPConn {
@@ -1110,19 +1118,13 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 	stopServer()
 	stopServer = nil
 
+	// Test: all expected server logs were emitted
+
 	// TODO: stops should be fully synchronous, but, intermittently,
 	// server_tunnel fails to appear ("missing server tunnel log")
 	// without this delay.
 	time.Sleep(100 * time.Millisecond)
 
-	// Test: all expected logs/notices were emitted
-
-	select {
-	case <-clientConnectedNotice:
-	default:
-		t.Fatalf("missing client connected notice")
-	}
-
 	select {
 	case logFields := <-serverConnectedLog:
 		err := checkExpectedLogFields(runConfig, false, false, logFields)
@@ -1926,10 +1928,14 @@ func sendNotificationReceived(c chan<- struct{}) {
 }
 
 func waitOnNotification(t *testing.T, c, timeoutSignal <-chan struct{}, timeoutMessage string) {
-	select {
-	case <-c:
-	case <-timeoutSignal:
-		t.Fatalf(timeoutMessage)
+	if timeoutSignal == nil {
+		<-c
+	} else {
+		select {
+		case <-c:
+		case <-timeoutSignal:
+			t.Fatalf(timeoutMessage)
+		}
 	}
 }
 

psiphon/server/tunnelServer.go

@@ -70,6 +70,7 @@ const (
 	MAX_AUTHORIZATIONS                    = 16
 	PRE_HANDSHAKE_RANDOM_STREAM_MAX_COUNT = 1
 	RANDOM_STREAM_MAX_BYTES               = 10485760
+	ALERT_REQUEST_QUEUE_BUFFER_SIZE       = 16
 )
 
 // TunnelServer is the main server that accepts Psiphon client
@@ -1169,6 +1170,8 @@ type sshClient struct {
 	stopTimer                            *time.Timer
 	preHandshakeRandomStreamMetrics      randomStreamMetrics
 	postHandshakeRandomStreamMetrics     randomStreamMetrics
+	sendAlertRequests                    chan protocol.AlertRequest
+	sentAlertRequests                    map[protocol.AlertRequest]bool
 }
 
 type trafficState struct {
@@ -1243,6 +1246,8 @@ func newSshClient(
 		runCtx:                 runCtx,
 		stopRunning:            stopRunning,
 		stopped:                make(chan struct{}),
+		sendAlertRequests:      make(chan protocol.AlertRequest, ALERT_REQUEST_QUEUE_BUFFER_SIZE),
+		sentAlertRequests:      make(map[protocol.AlertRequest]bool),
 	}
 
 	client.tcpTrafficState.availablePortForwardCond = sync.NewCond(new(sync.Mutex))
@@ -1653,14 +1658,21 @@ func (sshClient *sshClient) runTunnel(
 		sshClient.handleSSHRequests(requests)
 	}()
 
-	// Start OSL sender
+	// Start request senders
 
 	if sshClient.supportsServerRequests {
+
 		waitGroup.Add(1)
 		go func() {
 			defer waitGroup.Done()
 			sshClient.runOSLSender()
 		}()
+
+		waitGroup.Add(1)
+		go func() {
+			defer waitGroup.Done()
+			sshClient.runAlertSender()
+		}()
 	}
 
 	// Start the TCP port forward manager
@@ -2393,6 +2405,70 @@ func (sshClient *sshClient) sendOSLRequest() error {
 	return nil
 }
 
+// runAlertSender dequeues and sends alert requests to the client. As these
+// alerts are informational, there is no retry logic and no SSH client
+// acknowledgement (wantReply) is requested. This worker scheme allows
+// nonconcurrent components including udpgw and packet tunnel to enqueue
+// alerts without blocking their traffic processing.
+func (sshClient *sshClient) runAlertSender() {
+	for {
+		select {
+		case <-sshClient.runCtx.Done():
+			return
+
+		case request := <-sshClient.sendAlertRequests:
+			payload, err := json.Marshal(request)
+			if err != nil {
+				log.WithTraceFields(LogFields{"error": err}).Warning("Marshal failed")
+				break
+			}
+			_, _, err = sshClient.sshConn.SendRequest(
+				protocol.PSIPHON_API_ALERT_REQUEST_NAME,
+				false,
+				payload)
+			if err != nil && !isExpectedTunnelIOError(err) {
+				log.WithTraceFields(LogFields{"error": err}).Warning("SendRequest failed")
+				break
+			}
+			sshClient.Lock()
+			sshClient.sentAlertRequests[request] = true
+			sshClient.Unlock()
+		}
+	}
+}
+
+// enqueueAlertRequest enqueues an alert request to be sent to the client.
+// Only one request is sent per tunnel per protocol.AlertRequest value;
+// subsequent alerts with the same value are dropped. enqueueAlertRequest will
+// not block until the queue exceeds ALERT_REQUEST_QUEUE_BUFFER_SIZE.
+func (sshClient *sshClient) enqueueAlertRequest(request protocol.AlertRequest) {
+	sshClient.Lock()
+	if sshClient.sentAlertRequests[request] {
+		sshClient.Unlock()
+		return
+	}
+	sshClient.Unlock()
+	select {
+	case <-sshClient.runCtx.Done():
+	case sshClient.sendAlertRequests <- request:
+	}
+}
+
+func (sshClient *sshClient) enqueueDisallowedTrafficAlertRequest() {
+	sshClient.enqueueAlertRequest(protocol.AlertRequest{
+		Reason: protocol.PSIPHON_API_ALERT_DISALLOWED_TRAFFIC,
+	})
+}
+
+func (sshClient *sshClient) enqueueUnsafeTrafficAlertRequest(tags []BlocklistTag) {
+	for _, tag := range tags {
+		sshClient.enqueueAlertRequest(protocol.AlertRequest{
+			Reason:  protocol.PSIPHON_API_ALERT_UNSAFE_TRAFFIC,
+			Subject: tag.Subject,
+		})
+	}
+}
+
 func (sshClient *sshClient) rejectNewChannel(newChannel ssh.NewChannel, logMessage string) {
 
 	// We always return the reject reason "Prohibited":
@@ -2776,34 +2852,50 @@ func (sshClient *sshClient) isPortForwardPermitted(
 
 	tags := sshClient.sshServer.support.Blocklist.LookupIP(remoteIP)
 	if len(tags) > 0 {
+
 		sshClient.logBlocklistHits(remoteIP, "", tags)
+
 		if sshClient.sshServer.support.Config.BlocklistActive {
+			// Actively alert and block
+			sshClient.enqueueUnsafeTrafficAlertRequest(tags)
 			return false
 		}
 	}
 
 	// Don't lock before calling logBlocklistHits.
+	// Unlock before calling enqueueDisallowedTrafficAlertRequest/log.
+
 	sshClient.Lock()
-	defer sshClient.Unlock()
+
+	allowed := true
 
 	// Client must complete handshake before port forwards are permitted.
 	if !sshClient.handshakeState.completed {
-		return false
+		allowed = false
 	}
 
-	// Traffic rules checks.
-
-	switch portForwardType {
-	case portForwardTypeTCP:
-		if sshClient.trafficRules.AllowTCPPort(remoteIP, port) {
-			return true
-		}
-	case portForwardTypeUDP:
-		if sshClient.trafficRules.AllowUDPPort(remoteIP, port) {
-			return true
+	if allowed {
+		// Traffic rules checks.
+		switch portForwardType {
+		case portForwardTypeTCP:
+			if !sshClient.trafficRules.AllowTCPPort(remoteIP, port) {
+				allowed = false
+			}
+		case portForwardTypeUDP:
+			if !sshClient.trafficRules.AllowUDPPort(remoteIP, port) {
+				allowed = false
+			}
 		}
 	}
 
+	sshClient.Unlock()
+
+	if allowed {
+		return true
+	}
+
+	sshClient.enqueueDisallowedTrafficAlertRequest()
+
 	log.WithTraceFields(
 		LogFields{
 			"type": portForwardType,
@@ -3102,9 +3194,13 @@ func (sshClient *sshClient) handleTCPChannel(
 
 		tags := sshClient.sshServer.support.Blocklist.LookupDomain(hostToConnect)
 		if len(tags) > 0 {
+
 			sshClient.logBlocklistHits(nil, hostToConnect, tags)
+
 			if sshClient.sshServer.support.Config.BlocklistActive {
+				// Actively alert and block
 				// Note: not recording a port forward failure in this case
+				sshClient.enqueueUnsafeTrafficAlertRequest(tags)
 				sshClient.rejectNewChannel(newChannel, "port forward not permitted")
 				return
 			}

psiphon/serverApi.go

@@ -1009,6 +1009,8 @@ func HandleServerRequest(
 	switch name {
 	case protocol.PSIPHON_API_OSL_REQUEST_NAME:
 		return HandleOSLRequest(tunnelOwner, tunnel, payload)
+	case protocol.PSIPHON_API_ALERT_REQUEST_NAME:
+		return HandleAlertRequest(tunnelOwner, tunnel, payload)
 	}
 
 	return errors.Tracef("invalid request name: %s", name)
@@ -1049,3 +1051,17 @@ func HandleOSLRequest(
 
 	return nil
 }
+
+func HandleAlertRequest(
+	tunnelOwner TunnelOwner, tunnel *Tunnel, payload []byte) error {
+
+	var alertRequest protocol.AlertRequest
+	err := json.Unmarshal(payload, &alertRequest)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	NoticeServerAlert(alertRequest)
+
+	return nil
+}