Repository: nats-io/nats-server (go)

Issue:
Nginx Proxy WebSocket Client IP Address
What should I do to forward the IP addresses of users logging in with Nginx Client correctly?

Architectural
```
Server Private IP Address: 10.0.0.15
Nats Server WebSocket Port: 4025
Nginx Proxy External IP Access
  -> Nats Server (WebSocket Port) Private IP
```

Nats-Client
```
Nats-Ws 1.3.0
```

Nginx Config
```
    upstream natsNodes {
        ip_hash;
        server 10.0.0.15:4025;
    }

    server {
        listen 80;
        listen 443 ssl;

        ssl_certificate /path/dynamic.pem;
        ssl_certificate_key /path/dynamic.key;

        server_name n1.dynamic.com;

        location / {
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-Proto $scheme;
            proxy_set_header Host $host;

            proxy_http_version 1.1;
            proxy_redirect off;
            proxy_pass http://natsNodes;
        }
    }
```

Nats Node Config
```
    # Server Listen
    listen: 10.0.0.15:4029

    authorization: {
        timeout: 5
        ADMIN_PERMISSION = {
            publish = ">"
            subscribe = ">"
        }
        USER_PERMISSION = {
            publish = "request"
            subscribe = ">"
        }
        users = [
            {user: admin, password: OwnAdminPassword, permissions: $ADMIN_PERMISSION}
            {user: dynamicUser, password: nil, permissions: $USER_PERMISSION}
        ]
    }

    # Cluster
    cluster {
        listen: 10.0.0.15:6922
        name: "Cluster Name"

        authorization {
            user: cUser
            password: ClusterPasswords
            timeout: 0.75
        }

        routes [
            # nats-route://cUser:ClusterPasswords@10.0.0.15:6922
            nats-route://cUserr:ClusterPasswords@10.0.0.16:6922
        ]
        cluster_advertise: 10.0.0.15
        connect_retries: 5
    }

    #Web Socket
    websocket {
        listen: "10.0.0.15:4025"
        no_tls: true
        compression: true
    }

    # Monitoring Port
    http_port: 7569

    # Configs
    # 1MB
    max_connections: 1048576
    # 5MB
    max_payload: 5242880
    # 64KB
    max_control_line: 65536

    # 2 mins
    ping_interval: "2m"
    write_deadline: "10s"

    # Logging
    debug: false
    trace: false
    logtime: true
    logfile_size_limit: 1GB
    log_file: "/var/log/nats-server.log"
```

/connz output
```
    {
        "cid": 439,
        "kind": "Client",
        "type": "websocket",
        "ip": "10.0.0.15",
        "port": 35216,
        "start": "2021-09-12T17:47:00.393533655Z",
        "last_activity": "2021-09-12T21:18:02.393267634Z",
        "rtt": "147.964595ms",
        "uptime": "3h31m15s",
        "idle": "13s",
        "pending_bytes": 0,
        "in_msgs": 0,
        "out_msgs": 1428,
        "in_bytes": 0,
        "out_bytes": 39834554,
        "subscriptions": 3,
        "lang": "nats.ws",
        "version": "1.3.0"
    },
```

Pull request description:
[IMPROVED] Websocket: added client IP from X-Forwarded-For header
This is for cases when there is a proxy (Nginx, HAProxy, etc..)
between the client and the NATS Server. If this header is present,
the first IP is the one of the originating client and will be
used as the host/IP in server's representation of the client host.

Resolves #2514

Signed-off-by: Ivan Kozlovic <ivan@synadia.com>


Patch:
diff --git a/server/client.go b/server/client.go
index aeb0d6e2..605cb8e0 100644
--- a/server/client.go
+++ b/server/client.go
@@ -610,7 +610,13 @@ func (c *client) initClient() {
 			if conn = addr.String(); conn != _EMPTY_ {
 				host, port, _ := net.SplitHostPort(conn)
 				iPort, _ := strconv.Atoi(port)
-				c.host, c.port = host, uint16(iPort)
+				c.port = uint16(iPort)
+				if c.isWebsocket() && c.ws.clientIP != _EMPTY_ {
+					c.host = c.ws.clientIP
+					conn = net.JoinHostPort(c.host, port)
+				} else {
+					c.host = host
+				}
 				// Now that we have extracted host and port, escape
 				// the string because it is going to be used in Sprintf
 				conn = strings.ReplaceAll(conn, "%", "%%")
diff --git a/server/websocket.go b/server/websocket.go
index 3dbcd554..51ab5a2c 100644
--- a/server/websocket.go
+++ b/server/websocket.go
@@ -84,6 +84,7 @@ const (
 
 	wsNoMaskingHeader       = "Nats-No-Masking"
 	wsNoMaskingValue        = "true"
+	wsXForwardedForHeader   = "X-Forwarded-For"
 	wsNoMaskingFullResponse = wsNoMaskingHeader + ": " + wsNoMaskingValue + CR_LF
 	wsPMCExtension          = "permessage-deflate" // per-message compression
 	wsPMCSrvNoCtx           = "server_no_context_takeover"
@@ -113,6 +114,7 @@ type websocket struct {
 	maskwrite  bool
 	compressor *flate.Writer
 	cookieJwt  string
+	clientIP   string
 }
 
 type srvWebsocket struct {
@@ -771,6 +773,15 @@ func (s *Server) wsUpgrade(w http.ResponseWriter, r *http.Request) (*wsUpgradeRe
 	// Server always expect "clients" to send masked payload, unless the option
 	// "no-masking" has been enabled.
 	ws := &websocket{compress: compress, maskread: !noMasking}
+
+	// Check for X-Forwarded-For header
+	if cips, ok := r.Header[wsXForwardedForHeader]; ok {
+		cip := cips[0]
+		if net.ParseIP(cip) != nil {
+			ws.clientIP = cip
+		}
+	}
+
 	if kind == CLIENT {
 		// Indicate if this is likely coming from a browser.
 		if ua := r.Header.Get("User-Agent"); ua != _EMPTY_ && strings.HasPrefix(ua, "Mozilla/") {


Test patch:
diff --git a/server/websocket_test.go b/server/websocket_test.go
index 82eeebb1..c9fe79d4 100644
--- a/server/websocket_test.go
+++ b/server/websocket_test.go
@@ -1794,7 +1794,7 @@ type testWSClientOptions struct {
 	compress, web bool
 	host          string
 	port          int
-	extraHeaders  map[string]string
+	extraHeaders  map[string][]string
 	noTLS         bool
 }
 
@@ -1818,9 +1818,16 @@ func testNewWSClient(t testing.TB, o testWSClientOptions) (net.Conn, *bufio.Read
 	if o.web {
 		req.Header.Set("User-Agent", "Mozilla/5.0")
 	}
-	if o.extraHeaders != nil {
-		for hdr, val := range o.extraHeaders {
-			req.Header.Add(hdr, val)
+	if len(o.extraHeaders) > 0 {
+		for hdr, values := range o.extraHeaders {
+			if len(values) == 0 {
+				req.Header.Set(hdr, _EMPTY_)
+				continue
+			}
+			req.Header.Set(hdr, values[0])
+			for i := 1; i < len(values); i++ {
+				req.Header.Add(hdr, values[i])
+			}
 		}
 	}
 	req.URL, _ = url.Parse("wss://" + addr)
@@ -3788,8 +3795,8 @@ func TestWSJWTCookieUser(t *testing.T) {
 			nuc:  nucSigFunc(),
 			opts: func(t *testing.T, claims *jwt.UserClaims) (testWSClientOptions, testClaimsOptions) {
 				co := cliOpts
-				co.extraHeaders = map[string]string{}
-				co.extraHeaders["Cookie"] = o.Websocket.JWTCookie + "=" + genJwt(t, claims)
+				co.extraHeaders = map[string][]string{}
+				co.extraHeaders["Cookie"] = []string{o.Websocket.JWTCookie + "=" + genJwt(t, claims)}
 				return co, testClaimsOptions{connectRequest: struct{}{}}
 			},
 			expectAnswer: "-ERR",
@@ -3799,8 +3806,8 @@ func TestWSJWTCookieUser(t *testing.T) {
 			nuc:  nucBearerFunc(),
 			opts: func(t *testing.T, claims *jwt.UserClaims) (testWSClientOptions, testClaimsOptions) {
 				co := cliOpts
-				co.extraHeaders = map[string]string{}
-				co.extraHeaders["Cookie"] = o.Websocket.JWTCookie + "=" + genJwt(t, claims)
+				co.extraHeaders = map[string][]string{}
+				co.extraHeaders["Cookie"] = []string{o.Websocket.JWTCookie + "=" + genJwt(t, claims)}
 				return co, testClaimsOptions{connectRequest: struct{}{}}
 			},
 			expectAnswer: "+OK",
@@ -3810,8 +3817,8 @@ func TestWSJWTCookieUser(t *testing.T) {
 			nuc:  nucSigFunc(),
 			opts: func(t *testing.T, claims *jwt.UserClaims) (testWSClientOptions, testClaimsOptions) {
 				co := cliOpts
-				co.extraHeaders = map[string]string{}
-				co.extraHeaders["Cookie"] = o.Websocket.JWTCookie + "=" + genJwt(t, claims)
+				co.extraHeaders = map[string][]string{}
+				co.extraHeaders["Cookie"] = []string{o.Websocket.JWTCookie + "=" + genJwt(t, claims)}
 				return co, testClaimsOptions{nuc: nucBearerFunc()}
 			},
 			expectAnswer: "+OK",
@@ -3884,6 +3891,101 @@ func TestWSReloadTLSConfig(t *testing.T) {
 	}
 }
 
+type captureClientConnectedLogger struct {
+	DummyLogger
+	ch chan string
+}
+
+func (l *captureClientConnectedLogger) Debugf(format string, v ...interface{}) {
+	msg := fmt.Sprintf(format, v...)
+	if !strings.Contains(msg, "Client connection created") {
+		return
+	}
+	select {
+	case l.ch <- msg:
+	default:
+	}
+}
+
+func TestWSXForwardedFor(t *testing.T) {
+	o := testWSOptions()
+	s := RunServer(o)
+	defer s.Shutdown()
+
+	l := &captureClientConnectedLogger{ch: make(chan string, 1)}
+	s.SetLogger(l, true, false)
+
+	for _, test := range []struct {
+		name          string
+		headers       func() map[string][]string
+		useHdrValue   bool
+		expectedValue string
+	}{
+		{"nil map", func() map[string][]string {
+			return nil
+		}, false, _EMPTY_},
+		{"empty map", func() map[string][]string {
+			return make(map[string][]string)
+		}, false, _EMPTY_},
+		{"header present empty value", func() map[string][]string {
+			m := make(map[string][]string)
+			m[wsXForwardedForHeader] = []string{}
+			return m
+		}, false, _EMPTY_},
+		{"header present invalid IP", func() map[string][]string {
+			m := make(map[string][]string)
+			m[wsXForwardedForHeader] = []string{"not a valid IP"}
+			return m
+		}, false, _EMPTY_},
+		{"header present one IP", func() map[string][]string {
+			m := make(map[string][]string)
+			m[wsXForwardedForHeader] = []string{"1.2.3.4"}
+			return m
+		}, true, "1.2.3.4"},
+		{"header present multiple IPs", func() map[string][]string {
+			m := make(map[string][]string)
+			m[wsXForwardedForHeader] = []string{"1.2.3.4", "5.6.7.8"}
+			return m
+		}, true, "1.2.3.4"},
+		{"header present IPv6", func() map[string][]string {
+			m := make(map[string][]string)
+			m[wsXForwardedForHeader] = []string{"::1"}
+			return m
+		}, true, "[::1]"},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			c, r, _ := testNewWSClient(t, testWSClientOptions{
+				host:         o.Websocket.Host,
+				port:         o.Websocket.Port,
+				extraHeaders: test.headers(),
+			})
+			defer c.Close()
+			// Send CONNECT and PING
+			wsmsg := testWSCreateClientMsg(wsBinaryMessage, 1, true, false, []byte("CONNECT {\"verbose\":false,\"protocol\":1}\r\nPING\r\n"))
+			if _, err := c.Write(wsmsg); err != nil {
+				t.Fatalf("Error sending message: %v", err)
+			}
+			// Wait for the PONG
+			if msg := testWSReadFrame(t, r); !bytes.HasPrefix(msg, []byte("PONG\r\n")) {
+				t.Fatalf("Expected PONG, got %s", msg)
+			}
+			select {
+			case d := <-l.ch:
+				ipAndColumn := fmt.Sprintf("%s:", test.expectedValue)
+				if test.useHdrValue {
+					if !strings.HasPrefix(d, ipAndColumn) {
+						t.Fatalf("Expected debug statement to start with: %q, got %q", ipAndColumn, d)
+					}
+				} else if strings.HasPrefix(d, ipAndColumn) {
+					t.Fatalf("Unexpected debug statement: %q", d)
+				}
+			case <-time.After(time.Second):
+				t.Fatal("Did not get connect debug statement")
+			}
+		})
+	}
+}
+
 // ==================================================================
 // = Benchmark tests
 // ==================================================================