Stream: the "proxy_protocol" parameter of the "listen" directive.

src/stream/ngx_stream.c

@@ -455,6 +455,7 @@ ngx_stream_add_addrs(ngx_conf_t *cf, ngx_stream_port_t *stport,
 #if (NGX_STREAM_SSL)
         addrs[i].conf.ssl = addr[i].opt.ssl;
 #endif
+        addrs[i].conf.proxy_protocol = addr[i].opt.proxy_protocol;
 
         len = ngx_sock_ntop(&addr[i].opt.sockaddr.sockaddr, addr[i].opt.socklen,
                             buf, NGX_SOCKADDR_STRLEN, 1);
@@ -504,6 +505,7 @@ ngx_stream_add_addrs6(ngx_conf_t *cf, ngx_stream_port_t *stport,
 #if (NGX_STREAM_SSL)
         addrs6[i].conf.ssl = addr[i].opt.ssl;
 #endif
+        addrs6[i].conf.proxy_protocol = addr[i].opt.proxy_protocol;
 
         len = ngx_sock_ntop(&addr[i].opt.sockaddr.sockaddr, addr[i].opt.socklen,
                             buf, NGX_SOCKADDR_STRLEN, 1);

src/stream/ngx_stream.h

@@ -27,6 +27,7 @@ typedef struct ngx_stream_session_s  ngx_stream_session_t;
 
 
 #define NGX_STREAM_OK                        200
+#define NGX_STREAM_BAD_REQUEST               400
 #define NGX_STREAM_FORBIDDEN                 403
 #define NGX_STREAM_INTERNAL_SERVER_ERROR     500
 #define NGX_STREAM_BAD_GATEWAY               502
@@ -58,6 +59,7 @@ typedef struct {
     unsigned                       reuseport:1;
 #endif
     unsigned                       so_keepalive:2;
+    unsigned                       proxy_protocol:1;
 #if (NGX_HAVE_KEEPALIVE_TUNABLE)
     int                            tcp_keepidle;
     int                            tcp_keepintvl;
@@ -72,8 +74,9 @@ typedef struct {
     ngx_stream_conf_ctx_t         *ctx;
     ngx_str_t                      addr_text;
 #if (NGX_STREAM_SSL)
-    ngx_uint_t                     ssl;    /* unsigned   ssl:1; */
+    unsigned                       ssl:1;
 #endif
+    unsigned                       proxy_protocol:1;
 } ngx_stream_addr_conf_t;
 
 typedef struct {
@@ -153,6 +156,8 @@ typedef struct {
     ngx_msec_t                     resolver_timeout;
     ngx_resolver_t                *resolver;
 
+    ngx_msec_t                     proxy_protocol_timeout;
+
     ngx_uint_t                     listen;  /* unsigned  listen:1; */
 } ngx_stream_core_srv_conf_t;
 

src/stream/ngx_stream_core_module.c

@@ -77,6 +77,13 @@ static ngx_command_t  ngx_stream_core_commands[] = {
       offsetof(ngx_stream_core_srv_conf_t, resolver_timeout),
       NULL },
 
+    { ngx_string("proxy_protocol_timeout"),
+      NGX_STREAM_MAIN_CONF|NGX_STREAM_SRV_CONF|NGX_CONF_TAKE1,
+      ngx_conf_set_msec_slot,
+      NGX_STREAM_SRV_CONF_OFFSET,
+      offsetof(ngx_stream_core_srv_conf_t, proxy_protocol_timeout),
+      NULL },
+
     { ngx_string("tcp_nodelay"),
       NGX_STREAM_MAIN_CONF|NGX_STREAM_SRV_CONF|NGX_CONF_FLAG,
       ngx_conf_set_flag_slot,
@@ -192,6 +199,7 @@ ngx_stream_core_create_srv_conf(ngx_conf_t *cf)
     cscf->file_name = cf->conf_file->file.name.data;
     cscf->line = cf->conf_file->line;
     cscf->resolver_timeout = NGX_CONF_UNSET_MSEC;
+    cscf->proxy_protocol_timeout = NGX_CONF_UNSET_MSEC;
     cscf->tcp_nodelay = NGX_CONF_UNSET;
 
     return cscf;
@@ -240,6 +248,9 @@ ngx_stream_core_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child)
         }
     }
 
+    ngx_conf_merge_msec_value(conf->proxy_protocol_timeout,
+                              prev->proxy_protocol_timeout, 5000);
+
     ngx_conf_merge_value(conf->tcp_nodelay, prev->tcp_nodelay, 1);
 
     return NGX_CONF_OK;
@@ -572,6 +583,11 @@ ngx_stream_core_listen(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
 #endif
         }
 
+        if (ngx_strcmp(value[i].data, "proxy_protocol") == 0) {
+            ls->proxy_protocol = 1;
+            continue;
+        }
+
         ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
                            "the invalid \"%V\" parameter", &value[i]);
         return NGX_CONF_ERROR;
@@ -591,6 +607,10 @@ ngx_stream_core_listen(ngx_conf_t *cf, ngx_command_t *cmd, void *conf)
         if (ls->so_keepalive) {
             return "\"so_keepalive\" parameter is incompatible with \"udp\"";
         }
+
+        if (ls->proxy_protocol) {
+            return "\"proxy_protocol\" parameter is incompatible with \"udp\"";
+        }
     }
 
     als = cmcf->listen.elts;

src/stream/ngx_stream_handler.c

@@ -13,6 +13,7 @@
 
 static void ngx_stream_close_connection(ngx_connection_t *c);
 static u_char *ngx_stream_log_error(ngx_log_t *log, u_char *buf, size_t len);
+static void ngx_stream_proxy_protocol_handler(ngx_event_t *rev);
 static void ngx_stream_init_session_handler(ngx_event_t *rev);
 static void ngx_stream_init_session(ngx_connection_t *c);
 
@@ -171,6 +172,23 @@ ngx_stream_init_connection(ngx_connection_t *c)
     rev = c->read;
     rev->handler = ngx_stream_init_session_handler;
 
+    if (addr_conf->proxy_protocol) {
+        c->log->action = "reading PROXY protocol";
+
+        rev->handler = ngx_stream_proxy_protocol_handler;
+
+        if (!rev->ready) {
+            ngx_add_timer(rev, cscf->proxy_protocol_timeout);
+
+            if (ngx_handle_read_event(rev, 0) != NGX_OK) {
+                ngx_stream_finalize_session(s,
+                                            NGX_STREAM_INTERNAL_SERVER_ERROR);
+            }
+
+            return;
+        }
+    }
+
     if (ngx_use_accept_mutex) {
         ngx_post_event(rev, &ngx_posted_events);
         return;
@@ -180,6 +198,82 @@ ngx_stream_init_connection(ngx_connection_t *c)
 }
 
 
+static void
+ngx_stream_proxy_protocol_handler(ngx_event_t *rev)
+{
+    u_char                      *p, buf[NGX_PROXY_PROTOCOL_MAX_HEADER];
+    size_t                       size;
+    ssize_t                      n;
+    ngx_err_t                    err;
+    ngx_connection_t            *c;
+    ngx_stream_session_t        *s;
+    ngx_stream_core_srv_conf_t  *cscf;
+
+    c = rev->data;
+    s = c->data;
+
+    ngx_log_debug0(NGX_LOG_DEBUG_STREAM, c->log, 0,
+                   "stream PROXY protocol handler");
+
+    if (rev->timedout) {
+        ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT, "client timed out");
+        ngx_stream_finalize_session(s, NGX_STREAM_OK);
+        return;
+    }
+
+    n = recv(c->fd, (char *) buf, sizeof(buf), MSG_PEEK);
+
+    err = ngx_socket_errno;
+
+    ngx_log_debug1(NGX_LOG_DEBUG_STREAM, c->log, 0, "recv(): %z", n);
+
+    if (n == -1) {
+        if (err == NGX_EAGAIN) {
+            rev->ready = 0;
+
+            if (!rev->timer_set) {
+                cscf = ngx_stream_get_module_srv_conf(s,
+                                                      ngx_stream_core_module);
+
+                ngx_add_timer(rev, cscf->proxy_protocol_timeout);
+            }
+
+            if (ngx_handle_read_event(rev, 0) != NGX_OK) {
+                ngx_stream_finalize_session(s,
+                                            NGX_STREAM_INTERNAL_SERVER_ERROR);
+            }
+
+            return;
+        }
+
+        ngx_connection_error(c, err, "recv() failed");
+
+        ngx_stream_finalize_session(s, NGX_STREAM_OK);
+        return;
+    }
+
+    if (rev->timer_set) {
+        ngx_del_timer(rev);
+    }
+
+    p = ngx_proxy_protocol_read(c, buf, buf + n);
+
+    if (p == NULL) {
+        ngx_stream_finalize_session(s, NGX_STREAM_BAD_REQUEST);
+        return;
+    }
+
+    size = p - buf;
+
+    if (c->recv(c, buf, size) != (ssize_t) size) {
+        ngx_stream_finalize_session(s, NGX_STREAM_INTERNAL_SERVER_ERROR);
+        return;
+    }
+
+    ngx_stream_init_session_handler(rev);
+}
+
+
 static void
 ngx_stream_init_session_handler(ngx_event_t *rev)
 {