Merge ebd257f into 1ac0e49

README.md

@@ -235,6 +235,7 @@ To install Mercure in a [Kubernetes](https://kubernetes.io) cluster, use the off
 * `READ_TIMEOUT`: maximum duration for reading the entire request, including the body, set to `0s` to disable (default), example: `2m`
 * `SUBSCRIBER_JWT_KEY`: must contain the secret key to valid subscribers' JWT, can be omited if `JWT_KEY` is set
 * `WRITE_TIMEOUT`: maximum duration before timing out writes of the response, set to `0s` to disable (default), example: `2m`
+* `USE_FORWARDED_HEADERS`: set to `1` to use the `X-Forwarded-For`, and `X-Real-IP` for the remote (client) IP address, `X-Forwarded-Proto` or `X-Forwarded-Scheme` for the scheme (http or https), `X-Forwarded-Host` for the host and the RFC 7239 `Forwarded` header, which may include both client IPs and schemes. If this option is enabled, the reverse proxy must override or remove these headers or you will be at risk.
 
 If `ACME_HOSTS` or both `CERT_FILE` and `KEY_FILE` are provided, an HTTPS server supporting HTTP/2 connection will be started.
 If not, an HTTP server will be started (**not secure**).

hub/options.go

@@ -28,6 +28,7 @@ type Options struct {
 	ReadTimeout             time.Duration
 	WriteTimeout            time.Duration
 	Compress                bool
+	UseForwardedHeaders     bool
 	Demo                    bool
 }
 
@@ -102,6 +103,7 @@ func NewOptionsFromEnv() (*Options, error) {
 		readTimeout,
 		writeTimeout,
 		os.Getenv("COMPRESS") != "0",
+		os.Getenv("USE_FORWARDED_HEADERS") == "1",
 		os.Getenv("DEMO") == "1" || os.Getenv("DEBUG") == "1",
 	}
 

hub/options_test.go

@@ -29,6 +29,7 @@ func TestNewOptionsFormNew(t *testing.T) {
 		"HEARTBEAT_INTERVAL":        "30s",
 		"READ_TIMEOUT":              "1m",
 		"WRITE_TIMEOUT":             "40s",
+		"USE_FORWARDED_HEADERS":     "1",
 	}
 	for k, v := range testEnv {
 		os.Setenv(k, v)
@@ -56,6 +57,7 @@ func TestNewOptionsFormNew(t *testing.T) {
 		40 * time.Second,
 		false,
 		true,
+		true,
 	}, opts)
 	assert.Nil(t, err)
 }

hub/server.go

@@ -114,7 +114,15 @@ func (h *Hub) chainHandlers() http.Handler {
 	} else {
 		compressHandler = corsHandler
 	}
-	secureHandler := secureMiddleware.Handler(compressHandler)
+
+	var useForwardedHeadersHandlers http.Handler
+	if h.options.UseForwardedHeaders {
+		useForwardedHeadersHandlers = handlers.ProxyHeaders(compressHandler)
+	} else {
+		useForwardedHeadersHandlers = compressHandler
+	}
+
+	secureHandler := secureMiddleware.Handler(useForwardedHeadersHandlers)
 	loggingHandler := handlers.CombinedLoggingHandler(os.Stderr, secureHandler)
 	recoveryHandler := handlers.RecoveryHandler(
 		handlers.RecoveryLogger(log.New()),

hub/server_test.go

@@ -11,12 +11,48 @@ import (
 	"testing"
 	"time"
 
+	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/stretchr/testify/assert"
 )
 
 const testURL = "http://" + testAddr + "/hub"
 const testSecureURL = "https://" + testAddr + "/hub"
 
+func TestForwardedHeaders(t *testing.T) {
+	h := createAnonymousDummy()
+	h.options.Demo = true
+	h.options.UseForwardedHeaders = true
+
+	h.Start()
+	go func() {
+		h.Serve()
+	}()
+
+	client := http.Client{Timeout: time.Duration(100 * time.Millisecond)}
+	hook := test.NewGlobal()
+
+	// loop until the web server is ready
+	var resp *http.Response
+	for resp == nil {
+		resp, _ = client.Get(testURL)
+	}
+
+	body := url.Values{"topic": {"http://example.com/test-forwarded"}, "data": {"hello"}}
+	req, _ := http.NewRequest("POST", testURL, strings.NewReader(body.Encode()))
+	req.Header.Add("X-Forwarded-For", "192.0.2.1")
+	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Add("Authorization", "Bearer "+createDummyAuthorizedJWT(h, true, []string{}))
+
+	_, err := client.Do(req)
+	if err != nil {
+		panic(err)
+	}
+
+	assert.Equal(t, "192.0.2.1", hook.LastEntry().Data["remote_addr"])
+
+	h.server.Shutdown(context.Background())
+}
+
 func TestSecurityOptions(t *testing.T) {
 	h := createAnonymousDummy()
 	h.options.Demo = true