From cfba69781a18407d5c2020c4e3f3d4fc175a6127 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Thu, 1 Jul 2021 13:56:40 +0300 Subject: Fixing multiple TLS-enabled listeners initialization. Because of the incorrect 'last' field assignment, multiple listeners with a TLS certificate did not initialize properly, which caused a router crash while establishing a connection. Test with multiple TLS listeners added. The issue was introduced in the c548e46fe516 commit. This closes #561 issue on GitHub. --- src/nxt_router.c | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index 015ae226..26b846b0 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -124,7 +124,7 @@ static void nxt_router_tls_rpc_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data); static nxt_int_t nxt_router_conf_tls_insert(nxt_router_temp_conf_t *tmcf, nxt_conf_value_t *value, nxt_socket_conf_t *skcf, - nxt_conf_value_t * conf_cmds); + nxt_conf_value_t * conf_cmds, nxt_bool_t last); #endif static void nxt_router_app_rpc_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_app_t *app); @@ -956,8 +956,6 @@ nxt_router_conf_apply(nxt_task_t *task, void *obj, void *data) tls = nxt_queue_link_data(qlk, nxt_router_tlssock_t, link); - tls->last = nxt_queue_is_empty(&tmcf->tls); - nxt_cert_store_get(task, &tls->name, tmcf->mem_pool, nxt_router_tls_rpc_handler, tls); return; @@ -1752,7 +1750,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_assert(value != NULL); ret = nxt_router_conf_tls_insert(tmcf, value, skcf, - conf_cmds); + conf_cmds, i == 0); if (nxt_slow_path(ret != NXT_OK)) { goto fail; } @@ -1761,7 +1759,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, } else { /* NXT_CONF_STRING */ ret = nxt_router_conf_tls_insert(tmcf, certificate, skcf, - conf_cmds); + conf_cmds, 1); if (nxt_slow_path(ret != NXT_OK)) { goto fail; } @@ -1856,7 +1854,7 @@ fail: static nxt_int_t nxt_router_conf_tls_insert(nxt_router_temp_conf_t *tmcf, nxt_conf_value_t *value, nxt_socket_conf_t *skcf, - nxt_conf_value_t *conf_cmds) + nxt_conf_value_t *conf_cmds, nxt_bool_t last) { nxt_router_tlssock_t *tls; @@ -1868,6 +1866,7 @@ nxt_router_conf_tls_insert(nxt_router_temp_conf_t *tmcf, tls->socket_conf = skcf; tls->conf_cmds = conf_cmds; tls->temp_conf = tmcf; + tls->last = last; nxt_conf_get_string(value, &tls->name); nxt_queue_insert_tail(&tmcf->tls, &tls->link); -- cgit From daa051e7e7266325ef38a606b3aee4377a73f0d0 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Mon, 19 Jul 2021 16:23:13 +0300 Subject: Router: fixing assertion on app thread port handle. A new application thread port message can be processed in the router after the application is removed from the router. Assertion for this case is replaced by a condition to store the new thread port until receiving the stop notification from the application process. --- src/nxt_router.c | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index 26b846b0..409f88a1 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -680,18 +680,20 @@ nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) nxt_assert(main_app_port != NULL); app = main_app_port->app; - nxt_assert(app != NULL); - nxt_thread_mutex_lock(&app->mutex); + if (nxt_fast_path(app != NULL)) { + nxt_thread_mutex_lock(&app->mutex); - /* TODO here should be find-and-add code because there can be - port waiters in port_hash */ - nxt_port_hash_add(&app->port_hash, port); - app->port_hash_count++; + /* TODO here should be find-and-add code because there can be + port waiters in port_hash */ + nxt_port_hash_add(&app->port_hash, port); + app->port_hash_count++; - nxt_thread_mutex_unlock(&app->mutex); + nxt_thread_mutex_unlock(&app->mutex); + + port->app = app; + } - port->app = app; port->main_app_port = main_app_port; nxt_port_socket_write(task, port, NXT_PORT_MSG_PORT_ACK, -1, 0, 0, NULL); -- cgit From c37ff7ed0ed06b0e928efdb217a8999ff3ff7f50 Mon Sep 17 00:00:00 2001 From: Andrey Suvorov Date: Wed, 21 Jul 2021 15:22:52 -0700 Subject: Enabling configure TLS sessions. To support TLS sessions, Unit uses the OpenSSL built-in session cache; the cache_size option defines the number sessions to store. To disable the feather, the option must be zero. --- src/nxt_router.c | 48 ++++++++++++++++++++++++++++++++++++------------ 1 file changed, 36 insertions(+), 12 deletions(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index 409f88a1..1156edb8 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -44,10 +44,10 @@ typedef struct { nxt_str_t name; nxt_socket_conf_t *socket_conf; nxt_router_temp_conf_t *temp_conf; - nxt_conf_value_t *conf_cmds; + nxt_tls_init_t *tls_init; nxt_bool_t last; - nxt_queue_link_t link; /* for nxt_socket_conf_t.tls */ + nxt_queue_link_t link; /* for nxt_socket_conf_t.tls */ } nxt_router_tlssock_t; #endif @@ -123,8 +123,8 @@ static void nxt_router_listen_socket_error(nxt_task_t *task, static void nxt_router_tls_rpc_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data); static nxt_int_t nxt_router_conf_tls_insert(nxt_router_temp_conf_t *tmcf, - nxt_conf_value_t *value, nxt_socket_conf_t *skcf, - nxt_conf_value_t * conf_cmds, nxt_bool_t last); + nxt_conf_value_t *value, nxt_socket_conf_t *skcf, nxt_tls_init_t *tls_init, + nxt_bool_t last); #endif static void nxt_router_app_rpc_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_app_t *app); @@ -1341,7 +1341,8 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_router_t *router; nxt_app_joint_t *app_joint; #if (NXT_TLS) - nxt_conf_value_t *certificate, *conf_cmds; + nxt_tls_init_t *tls_init; + nxt_conf_value_t *certificate; #endif nxt_conf_value_t *conf, *http, *value, *websocket; nxt_conf_value_t *applications, *application; @@ -1363,6 +1364,8 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, #if (NXT_TLS) static nxt_str_t certificate_path = nxt_string("/tls/certificate"); static nxt_str_t conf_commands_path = nxt_string("/tls/conf_commands"); + static nxt_str_t conf_cache_path = nxt_string("/tls/session/cache_size"); + static nxt_str_t conf_timeout_path = nxt_string("/tls/session/timeout"); #endif static nxt_str_t static_path = nxt_string("/settings/http/static"); static nxt_str_t websocket_path = nxt_string("/settings/http/websocket"); @@ -1741,7 +1744,26 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, certificate = nxt_conf_get_path(listener, &certificate_path); if (certificate != NULL) { - conf_cmds = nxt_conf_get_path(listener, &conf_commands_path); + tls_init = nxt_mp_get(tmcf->mem_pool, sizeof(nxt_tls_init_t)); + if (nxt_slow_path(tls_init == NULL)) { + return NXT_ERROR; + } + + tls_init->cache_size = 0; + tls_init->timeout = 300; + + value = nxt_conf_get_path(listener, &conf_cache_path); + if (value != NULL) { + tls_init->cache_size = nxt_conf_get_number(value); + } + + value = nxt_conf_get_path(listener, &conf_timeout_path); + if (value != NULL) { + tls_init->timeout = nxt_conf_get_number(value); + } + + tls_init->conf_cmds = nxt_conf_get_path(listener, + &conf_commands_path); if (nxt_conf_type(certificate) == NXT_CONF_ARRAY) { n = nxt_conf_array_elements_count(certificate); @@ -1752,7 +1774,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_assert(value != NULL); ret = nxt_router_conf_tls_insert(tmcf, value, skcf, - conf_cmds, i == 0); + tls_init, i == 0); if (nxt_slow_path(ret != NXT_OK)) { goto fail; } @@ -1761,7 +1783,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, } else { /* NXT_CONF_STRING */ ret = nxt_router_conf_tls_insert(tmcf, certificate, skcf, - conf_cmds, 1); + tls_init, 1); if (nxt_slow_path(ret != NXT_OK)) { goto fail; } @@ -1856,7 +1878,7 @@ fail: static nxt_int_t nxt_router_conf_tls_insert(nxt_router_temp_conf_t *tmcf, nxt_conf_value_t *value, nxt_socket_conf_t *skcf, - nxt_conf_value_t *conf_cmds, nxt_bool_t last) + nxt_tls_init_t *tls_init, nxt_bool_t last) { nxt_router_tlssock_t *tls; @@ -1865,8 +1887,8 @@ nxt_router_conf_tls_insert(nxt_router_temp_conf_t *tmcf, return NXT_ERROR; } + tls->tls_init = tls_init; tls->socket_conf = skcf; - tls->conf_cmds = conf_cmds; tls->temp_conf = tmcf; tls->last = last; nxt_conf_get_string(value, &tls->name); @@ -2467,6 +2489,8 @@ nxt_router_tls_rpc_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg, tlscf = tls->socket_conf->tls; } + tls->tls_init->conf = tlscf; + bundle = nxt_mp_get(mp, sizeof(nxt_tls_bundle_conf_t)); if (nxt_slow_path(bundle == NULL)) { goto fail; @@ -2480,8 +2504,8 @@ nxt_router_tls_rpc_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg, bundle->next = tlscf->bundle; tlscf->bundle = bundle; - ret = task->thread->runtime->tls->server_init(task, tlscf, mp, - tls->conf_cmds, tls->last); + ret = task->thread->runtime->tls->server_init(task, mp, tls->tls_init, + tls->last); if (nxt_slow_path(ret != NXT_OK)) { goto fail; } -- cgit From a3df6efc8d5994899a985080b574d9043ecd80f3 Mon Sep 17 00:00:00 2001 From: Zhidao HONG Date: Fri, 23 Jul 2021 09:14:43 +0800 Subject: Router: split nxt_http_static_conf_t from nxt_http_action_t. No functional changes. --- src/nxt_router.c | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index 1156edb8..96d62575 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -1907,7 +1907,7 @@ nxt_router_conf_process_static(nxt_task_t *task, nxt_router_conf_t *rtcf, { uint32_t next, i; nxt_mp_t *mp; - nxt_str_t *type, extension, str; + nxt_str_t *type, exten, str; nxt_int_t ret; nxt_uint_t exts; nxt_conf_value_t *mtypes_conf, *ext_conf, *value; @@ -1945,12 +1945,12 @@ nxt_router_conf_process_static(nxt_task_t *task, nxt_router_conf_t *rtcf, if (nxt_conf_type(ext_conf) == NXT_CONF_STRING) { nxt_conf_get_string(ext_conf, &str); - if (nxt_slow_path(nxt_str_dup(mp, &extension, &str) == NULL)) { + if (nxt_slow_path(nxt_str_dup(mp, &exten, &str) == NULL)) { return NXT_ERROR; } ret = nxt_http_static_mtypes_hash_add(mp, &rtcf->mtypes_hash, - &extension, type); + &exten, type); if (nxt_slow_path(ret != NXT_OK)) { return NXT_ERROR; } @@ -1965,12 +1965,12 @@ nxt_router_conf_process_static(nxt_task_t *task, nxt_router_conf_t *rtcf, nxt_conf_get_string(value, &str); - if (nxt_slow_path(nxt_str_dup(mp, &extension, &str) == NULL)) { + if (nxt_slow_path(nxt_str_dup(mp, &exten, &str) == NULL)) { return NXT_ERROR; } ret = nxt_http_static_mtypes_hash_add(mp, &rtcf->mtypes_hash, - &extension, type); + &exten, type); if (nxt_slow_path(ret != NXT_OK)) { return NXT_ERROR; } -- cgit From f3a1c1deb541784b2b0ed179514e4d5eba9fe626 Mon Sep 17 00:00:00 2001 From: Zhidao HONG Date: Sat, 24 Jul 2021 11:44:52 +0800 Subject: Router: split nxt_http_app_conf_t from nxt_http_action_t. No functional changes. --- src/nxt_router.c | 47 ++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 38 insertions(+), 9 deletions(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index 96d62575..c766c25e 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -2158,21 +2158,46 @@ nxt_router_apps_hash_use(nxt_task_t *task, nxt_router_conf_t *rtcf, int i) } +typedef struct { + nxt_app_t *app; + nxt_int_t target; +} nxt_http_app_conf_t; + nxt_int_t -nxt_router_listener_application(nxt_router_conf_t *rtcf, nxt_str_t *name, - nxt_http_action_t *action) +nxt_router_application_init(nxt_router_conf_t *rtcf, nxt_str_t *name, + nxt_str_t *target, nxt_http_action_t *action) { - nxt_app_t *app; + nxt_app_t *app; + nxt_str_t *targets; + nxt_uint_t i; + nxt_http_app_conf_t *conf; app = nxt_router_apps_hash_get(rtcf, name); - if (app == NULL) { return NXT_DECLINED; } - action->u.app.application = app; + conf = nxt_mp_get(rtcf->mem_pool, sizeof(nxt_http_app_conf_t)); + if (nxt_slow_path(conf == NULL)) { + return NXT_ERROR; + } + action->handler = nxt_http_application_handler; + action->u.conf = conf; + + conf->app = app; + + if (target != NULL && target->length != 0) { + targets = app->targets; + + for (i = 0; !nxt_strstr_eq(target, &targets[i]); i++); + + conf->target = i; + + } else { + conf->target = 0; + } return NXT_OK; } @@ -4901,13 +4926,17 @@ nxt_router_app_port_get(nxt_task_t *task, nxt_app_t *app, void nxt_router_process_http_request(nxt_task_t *task, nxt_http_request_t *r, - nxt_app_t *app) + nxt_http_action_t *action) { nxt_event_engine_t *engine; + nxt_http_app_conf_t *conf; nxt_request_rpc_data_t *req_rpc_data; + conf = action->u.conf; engine = task->thread->engine; + r->app_target = conf->target; + req_rpc_data = nxt_port_rpc_register_handler_ex(task, engine->port, nxt_router_response_ready_handler, nxt_router_response_error_handler, @@ -4938,11 +4967,11 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_http_request_t *r, r->err_work.obj = r; req_rpc_data->stream = nxt_port_rpc_ex_stream(req_rpc_data); - req_rpc_data->app = app; + req_rpc_data->app = conf->app; req_rpc_data->msg_info.body_fd = -1; req_rpc_data->rpc_cancel = 1; - nxt_router_app_use(task, app, 1); + nxt_router_app_use(task, conf->app, 1); req_rpc_data->request = r; r->req_rpc_data = req_rpc_data; @@ -4951,7 +4980,7 @@ nxt_router_process_http_request(nxt_task_t *task, nxt_http_request_t *r, r->last->completion_handler = nxt_router_http_request_done; } - nxt_router_app_port_get(task, app, req_rpc_data); + nxt_router_app_port_get(task, conf->app, req_rpc_data); nxt_router_app_prepare_request(task, req_rpc_data); } -- cgit From fa9fb29be221e0393562831a9e3bcba416652f60 Mon Sep 17 00:00:00 2001 From: Max Romanov Date: Thu, 29 Jul 2021 19:50:39 +0300 Subject: Application restart introduced. When processing a restart request, the router sends a QUIT message to all existing processes of the application. Then, a new shared application port is created to ensure that new requests won't be handled by the old processes of the application. --- src/nxt_router.c | 219 ++++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 178 insertions(+), 41 deletions(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index c766c25e..8360e75a 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -18,6 +18,8 @@ #include #include +#define NXT_SHARED_PORT_ID 0xFFFFu + typedef struct { nxt_str_t type; uint32_t processes; @@ -67,6 +69,12 @@ typedef struct { } nxt_app_rpc_t; +typedef struct { + nxt_app_joint_t *app_joint; + uint32_t generation; +} nxt_app_joint_rpc_t; + + static nxt_int_t nxt_router_prefork(nxt_task_t *task, nxt_process_t *process, nxt_mp_t *mp); static nxt_int_t nxt_router_start(nxt_task_t *task, nxt_process_data_t *data); @@ -79,6 +87,8 @@ static void nxt_router_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg); static void nxt_router_conf_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg); +static void nxt_router_app_restart_handler(nxt_task_t *task, + nxt_port_recv_msg_t *msg); static void nxt_router_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg); static void nxt_router_access_log_reopen_handler(nxt_task_t *task, @@ -281,6 +291,7 @@ static const nxt_port_handlers_t nxt_router_process_port_handlers = { .mmap = nxt_port_mmap_handler, .get_mmap = nxt_router_get_mmap_handler, .data = nxt_router_conf_data_handler, + .app_restart = nxt_router_app_restart_handler, .remove_pid = nxt_router_remove_pid_handler, .access_log = nxt_router_access_log_reopen_handler, .rpc_ready = nxt_port_rpc_handler, @@ -379,14 +390,15 @@ static void nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port, void *data) { - size_t size; - uint32_t stream; - nxt_mp_t *mp; - nxt_int_t ret; - nxt_app_t *app; - nxt_buf_t *b; - nxt_port_t *main_port; - nxt_runtime_t *rt; + size_t size; + uint32_t stream; + nxt_mp_t *mp; + nxt_int_t ret; + nxt_app_t *app; + nxt_buf_t *b; + nxt_port_t *main_port; + nxt_runtime_t *rt; + nxt_app_joint_rpc_t *app_joint_rpc; app = data; @@ -407,30 +419,29 @@ nxt_router_start_app_process_handler(nxt_task_t *task, nxt_port_t *port, *b->mem.free++ = '\0'; nxt_buf_cpystr(b, &app->conf); - nxt_router_app_joint_use(task, app->joint, 1); - - stream = nxt_port_rpc_register_handler(task, port, - nxt_router_app_port_ready, - nxt_router_app_port_error, - -1, app->joint); - - if (nxt_slow_path(stream == 0)) { - nxt_router_app_joint_use(task, app->joint, -1); - + app_joint_rpc = nxt_port_rpc_register_handler_ex(task, port, + nxt_router_app_port_ready, + nxt_router_app_port_error, + sizeof(nxt_app_joint_rpc_t)); + if (nxt_slow_path(app_joint_rpc == NULL)) { goto failed; } + stream = nxt_port_rpc_ex_stream(app_joint_rpc); + ret = nxt_port_socket_write(task, main_port, NXT_PORT_MSG_START_PROCESS, -1, stream, port->id, b); - if (nxt_slow_path(ret != NXT_OK)) { nxt_port_rpc_cancel(task, port, stream); - nxt_router_app_joint_use(task, app->joint, -1); - goto failed; } + app_joint_rpc->app_joint = app->joint; + app_joint_rpc->generation = app->generation; + + nxt_router_app_joint_use(task, app->joint, 1); + nxt_router_app_use(task, app, -1); return; @@ -504,6 +515,7 @@ nxt_router_msg_cancel(nxt_task_t *task, nxt_request_rpc_data_t *req_rpc_data) { nxt_buf_t *b, *next; nxt_bool_t cancelled; + nxt_port_t *app_port; nxt_msg_info_t *msg_info; msg_info = &req_rpc_data->msg_info; @@ -512,13 +524,20 @@ nxt_router_msg_cancel(nxt_task_t *task, nxt_request_rpc_data_t *req_rpc_data) return 0; } - cancelled = nxt_app_queue_cancel(req_rpc_data->app->shared_port->queue, - msg_info->tracking_cookie, - req_rpc_data->stream); + app_port = req_rpc_data->app_port; + + if (app_port != NULL && app_port->id == NXT_SHARED_PORT_ID) { + cancelled = nxt_app_queue_cancel(app_port->queue, + msg_info->tracking_cookie, + req_rpc_data->stream); + + if (cancelled) { + nxt_debug(task, "stream #%uD: cancelled by router", + req_rpc_data->stream); + } - if (cancelled) { - nxt_debug(task, "stream #%uD: cancelled by router", - req_rpc_data->stream); + } else { + cancelled = 0; } for (b = msg_info->buf; b != NULL; b = next) { @@ -793,6 +812,90 @@ cleanup: } +static void +nxt_router_app_restart_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg) +{ + nxt_app_t *app; + nxt_int_t ret; + nxt_str_t app_name; + nxt_port_t *port, *reply_port, *shared_port, *old_shared_port; + nxt_port_msg_type_t reply; + + reply_port = nxt_runtime_port_find(task->thread->runtime, + msg->port_msg.pid, + msg->port_msg.reply_port); + if (nxt_slow_path(reply_port == NULL)) { + nxt_alert(task, "app_restart_handler: reply port not found"); + return; + } + + app_name.length = nxt_buf_mem_used_size(&msg->buf->mem); + app_name.start = msg->buf->mem.pos; + + nxt_debug(task, "app_restart_handler: %V", &app_name); + + app = nxt_router_app_find(&nxt_router->apps, &app_name); + + if (nxt_fast_path(app != NULL)) { + shared_port = nxt_port_new(task, NXT_SHARED_PORT_ID, nxt_pid, + NXT_PROCESS_APP); + if (nxt_slow_path(shared_port == NULL)) { + goto fail; + } + + ret = nxt_port_socket_init(task, shared_port, 0); + if (nxt_slow_path(ret != NXT_OK)) { + nxt_port_use(task, shared_port, -1); + goto fail; + } + + ret = nxt_router_app_queue_init(task, shared_port); + if (nxt_slow_path(ret != NXT_OK)) { + nxt_port_write_close(shared_port); + nxt_port_read_close(shared_port); + nxt_port_use(task, shared_port, -1); + goto fail; + } + + nxt_port_write_enable(task, shared_port); + + nxt_thread_mutex_lock(&app->mutex); + + nxt_queue_each(port, &app->ports, nxt_port_t, app_link) { + + (void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, + 0, 0, NULL); + + } nxt_queue_loop; + + app->generation++; + + shared_port->app = app; + + old_shared_port = app->shared_port; + old_shared_port->app = NULL; + + app->shared_port = shared_port; + + nxt_thread_mutex_unlock(&app->mutex); + + nxt_port_close(task, old_shared_port); + nxt_port_use(task, old_shared_port, -1); + + reply = NXT_PORT_MSG_RPC_READY_LAST; + + } else { + +fail: + + reply = NXT_PORT_MSG_RPC_ERROR; + } + + nxt_port_socket_write(task, reply_port, reply, -1, msg->port_msg.stream, + 0, NULL); +} + + static void nxt_router_app_process_remove_pid(nxt_task_t *task, nxt_port_t *port, void *data) @@ -1607,7 +1710,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, app_joint->free_app_work.task = &engine->task; app_joint->free_app_work.obj = app_joint; - port = nxt_port_new(task, (nxt_port_id_t) -1, nxt_pid, + port = nxt_port_new(task, NXT_SHARED_PORT_ID, nxt_pid, NXT_PROCESS_APP); if (nxt_slow_path(port == NULL)) { return NXT_ERROR; @@ -4233,11 +4336,16 @@ static void nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data) { - nxt_app_t *app; - nxt_port_t *port; - nxt_app_joint_t *app_joint; + nxt_app_t *app; + nxt_bool_t start_process; + nxt_port_t *port; + nxt_app_joint_t *app_joint; + nxt_app_joint_rpc_t *app_joint_rpc; - app_joint = data; + nxt_assert(data != NULL); + + app_joint_rpc = data; + app_joint = app_joint_rpc->app_joint; port = msg->u.new_port; nxt_assert(app_joint != NULL); @@ -4257,14 +4365,37 @@ nxt_router_app_port_ready(nxt_task_t *task, nxt_port_recv_msg_t *msg, return; } - port->app = app; - port->main_app_port = port; - nxt_thread_mutex_lock(&app->mutex); nxt_assert(app->pending_processes != 0); app->pending_processes--; + + if (nxt_slow_path(app->generation != app_joint_rpc->generation)) { + nxt_debug(task, "new port ready for restarted app, send QUIT"); + + start_process = !task->thread->engine->shutdown + && nxt_router_app_can_start(app) + && nxt_router_app_need_start(app); + + if (start_process) { + app->pending_processes++; + } + + nxt_thread_mutex_unlock(&app->mutex); + + nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT, -1, 0, 0, NULL); + + if (start_process) { + nxt_router_start_app_process(task, app); + } + + return; + } + + port->app = app; + port->main_app_port = port; + app->processes++; nxt_port_hash_add(&app->port_hash, port); app->port_hash_count++; @@ -4318,12 +4449,16 @@ static void nxt_router_app_port_error(nxt_task_t *task, nxt_port_recv_msg_t *msg, void *data) { - nxt_app_t *app; - nxt_app_joint_t *app_joint; - nxt_queue_link_t *link; - nxt_http_request_t *r; + nxt_app_t *app; + nxt_app_joint_t *app_joint; + nxt_queue_link_t *link; + nxt_http_request_t *r; + nxt_app_joint_rpc_t *app_joint_rpc; + + nxt_assert(data != NULL); - app_joint = data; + app_joint_rpc = data; + app_joint = app_joint_rpc->app_joint; nxt_assert(app_joint != NULL); @@ -4490,7 +4625,7 @@ nxt_router_app_port_release(nxt_task_t *task, nxt_port_t *port, port->pid, port->id, (int) inc_use, (int) got_response); - if (port == app->shared_port) { + if (port->id == NXT_SHARED_PORT_ID) { nxt_thread_mutex_lock(&app->mutex); app->active_requests -= got_response + dec_requests; @@ -4860,6 +4995,8 @@ nxt_router_free_app(nxt_task_t *task, void *obj, void *data) app->shared_port->app = NULL; nxt_port_close(task, app->shared_port); nxt_port_use(task, app->shared_port, -1); + + app->shared_port = NULL; } nxt_thread_mutex_destroy(&app->mutex); -- cgit From ca373aaccd276fb412e59557a3971a8d06ada0f8 Mon Sep 17 00:00:00 2001 From: Oisin Canty Date: Thu, 12 Aug 2021 08:23:16 +0000 Subject: Router: client IP address replacement. This commit introduces the replacement of the client address based on the value of a specified HTTP header. This is intended for use when Unit is placed behind a reverse proxy like nginx or a CDN. You must specify the source addresses of the trusted proxies. This can be accomplished with any valid IP pattern supported by Unit's match block: ["10.0.0.1", "10.4.0.0/16", "!192.168.1.1"] The feature is configured per listener. The client address replacement functionality only operates when there is a source IP match and the specified header is present. Typically this would be an 'X-Forwarded-For' header. { "listeners": { "127.0.0.1:8080": { "client_ip": { "header": "X-Forwarded-For", "source": [ "10.0.0.0/8" ] }, "pass": "applications/my_app" }, } } If a request occurs and Unit receives a header like below: "X-Forwarded-For: 84.123.23.23" By default, Unit trusts the last rightmost IP in the header, so REMOTE_ADDR will be set to 84.123.23.23 if the connection originated from 10.0.0.0/8. If Unit runs behind consecutive reverse proxies and receives a header similar to the following: "X-Forwarded-For: 84.123.23.23, 10.0.0.254" You will need to enable "recursive" checking, which walks the header from last address to first and chooses the first non-trusted address it finds. { "listeners": { "127.0.0.1:8080": { "client_ip": { "header": "X-Forwarded-For", "source": [ "10.0.0.0/8" ] "recursive": true, }, "pass": "applications/my_app" }, } } If a connection from 10.0.0.0/8 occurs, the chain is walked. Here, 10.0.0.254 is also a trusted address so the client address will be replaced with 84.123.23.23. If all IP addresses in the header are trusted, the client address is set to the first address in the header: If 10.0.0.0/8 is trusted and "X-Forwarded-For: 10.0.0.3, 10.0.0.2, 10.0.0.1", the client address will be replaced with 10.0.0.3. --- src/nxt_router.c | 86 +++++++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 85 insertions(+), 1 deletion(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index 8360e75a..1aa919c2 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -107,6 +107,9 @@ static nxt_int_t nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, u_char *start, u_char *end); static nxt_int_t nxt_router_conf_process_static(nxt_task_t *task, nxt_router_conf_t *rtcf, nxt_conf_value_t *conf); +static nxt_int_t nxt_router_conf_process_client_ip(nxt_task_t *task, + nxt_router_temp_conf_t *tmcf, nxt_socket_conf_t *skcf, + nxt_conf_value_t *conf); static nxt_app_t *nxt_router_app_find(nxt_queue_t *queue, nxt_str_t *name); static nxt_int_t nxt_router_apps_hash_test(nxt_lvlhsh_query_t *lhq, void *data); @@ -1450,7 +1453,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, nxt_conf_value_t *conf, *http, *value, *websocket; nxt_conf_value_t *applications, *application; nxt_conf_value_t *listeners, *listener; - nxt_conf_value_t *routes_conf, *static_conf; + nxt_conf_value_t *routes_conf, *static_conf, *client_ip_conf; nxt_socket_conf_t *skcf; nxt_http_routes_t *routes; nxt_event_engine_t *engine; @@ -1472,6 +1475,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, #endif static nxt_str_t static_path = nxt_string("/settings/http/static"); static nxt_str_t websocket_path = nxt_string("/settings/http/websocket"); + static nxt_str_t client_ip_path = nxt_string("/client_ip"); conf = nxt_conf_json_parse(tmcf->mem_pool, start, end, NULL); if (conf == NULL) { @@ -1843,6 +1847,13 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, t->length = nxt_strlen(t->start); } + client_ip_conf = nxt_conf_get_path(listener, &client_ip_path); + ret = nxt_router_conf_process_client_ip(task, tmcf, skcf, + client_ip_conf); + if (nxt_slow_path(ret != NXT_OK)) { + return NXT_ERROR; + } + #if (NXT_TLS) certificate = nxt_conf_get_path(listener, &certificate_path); @@ -2085,6 +2096,79 @@ nxt_router_conf_process_static(nxt_task_t *task, nxt_router_conf_t *rtcf, } +static nxt_int_t +nxt_router_conf_process_client_ip(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, + nxt_socket_conf_t *skcf, nxt_conf_value_t *conf) +{ + char c; + size_t i; + nxt_mp_t *mp; + uint32_t hash; + nxt_str_t header; + nxt_conf_value_t *source_conf, *header_conf, *recursive_conf; + nxt_http_client_ip_t *client_ip; + nxt_http_route_addr_rule_t *source; + + static nxt_str_t header_path = nxt_string("/header"); + static nxt_str_t source_path = nxt_string("/source"); + static nxt_str_t recursive_path = nxt_string("/recursive"); + + if (conf == NULL) { + skcf->client_ip = NULL; + + return NXT_OK; + } + + mp = tmcf->router_conf->mem_pool; + + source_conf = nxt_conf_get_path(conf, &source_path); + header_conf = nxt_conf_get_path(conf, &header_path); + recursive_conf = nxt_conf_get_path(conf, &recursive_path); + + if (source_conf == NULL || header_conf == NULL) { + return NXT_ERROR; + } + + client_ip = nxt_mp_zget(mp, sizeof(nxt_http_client_ip_t)); + if (nxt_slow_path(client_ip == NULL)) { + return NXT_ERROR; + } + + source = nxt_http_route_addr_rule_create(task, mp, source_conf); + if (nxt_slow_path(source == NULL)) { + return NXT_ERROR; + } + + client_ip->source = source; + + nxt_conf_get_string(header_conf, &header); + + if (recursive_conf != NULL) { + client_ip->recursive = nxt_conf_get_boolean(recursive_conf); + } + + client_ip->header = nxt_str_dup(mp, NULL, &header); + if (nxt_slow_path(client_ip->header == NULL)) { + return NXT_ERROR; + } + + hash = NXT_HTTP_FIELD_HASH_INIT; + + for (i = 0; i < client_ip->header->length; i++) { + c = client_ip->header->start[i]; + hash = nxt_http_field_hash_char(hash, nxt_lowcase(c)); + } + + hash = nxt_http_field_hash_end(hash) & 0xFFFF; + + client_ip->header_hash = hash; + + skcf->client_ip = client_ip; + + return NXT_OK; +} + + static nxt_app_t * nxt_router_app_find(nxt_queue_t *queue, nxt_str_t *name) { -- cgit From 48a9399f23b9aaa2c9e5deb8013c58313e76740e Mon Sep 17 00:00:00 2001 From: Zhidao HONG Date: Thu, 12 Aug 2021 17:39:00 +0800 Subject: Introduced the generic API nxt_buf_dummy_completion(). No functional changes. --- src/nxt_router.c | 14 +++----------- 1 file changed, 3 insertions(+), 11 deletions(-) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index 1aa919c2..e0029d13 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -237,8 +237,6 @@ static void nxt_router_http_request_error(nxt_task_t *task, void *obj, static void nxt_router_http_request_done(nxt_task_t *task, void *obj, void *data); -static void nxt_router_dummy_buf_completion(nxt_task_t *task, void *obj, - void *data); static void nxt_router_app_prepare_request(nxt_task_t *task, nxt_request_rpc_data_t *req_rpc_data); static nxt_buf_t *nxt_router_prepare_msg(nxt_task_t *task, @@ -2532,7 +2530,7 @@ nxt_router_listen_socket_rpc_create(nxt_task_t *task, goto fail; } - b->completion_handler = nxt_router_dummy_buf_completion; + b->completion_handler = nxt_buf_dummy_completion; b->mem.free = nxt_cpymem(b->mem.free, skcf->listen->sockaddr, size); @@ -2763,7 +2761,7 @@ nxt_router_app_rpc_create(nxt_task_t *task, goto fail; } - b->completion_handler = nxt_router_dummy_buf_completion; + b->completion_handler = nxt_buf_dummy_completion; nxt_buf_cpystr(b, &app->name); *b->mem.free++ = '\0'; @@ -3792,7 +3790,7 @@ nxt_router_access_log_open(nxt_task_t *task, nxt_router_temp_conf_t *tmcf) goto fail; } - b->completion_handler = nxt_router_dummy_buf_completion; + b->completion_handler = nxt_buf_dummy_completion; nxt_buf_cpystr(b, &access_log->path); *b->mem.free++ = '\0'; @@ -5242,12 +5240,6 @@ nxt_router_http_request_done(nxt_task_t *task, void *obj, void *data) } -static void -nxt_router_dummy_buf_completion(nxt_task_t *task, void *obj, void *data) -{ -} - - static void nxt_router_app_prepare_request(nxt_task_t *task, nxt_request_rpc_data_t *req_rpc_data) -- cgit From e0aa132172f03fe7c31484ce7d301813b5dacb89 Mon Sep 17 00:00:00 2001 From: Andrey Suvorov Date: Tue, 17 Aug 2021 16:52:32 -0700 Subject: Added TLS session tickets support. --- src/nxt_router.c | 4 ++++ 1 file changed, 4 insertions(+) (limited to 'src/nxt_router.c') diff --git a/src/nxt_router.c b/src/nxt_router.c index e0029d13..39d375f8 100644 --- a/src/nxt_router.c +++ b/src/nxt_router.c @@ -1470,6 +1470,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, static nxt_str_t conf_commands_path = nxt_string("/tls/conf_commands"); static nxt_str_t conf_cache_path = nxt_string("/tls/session/cache_size"); static nxt_str_t conf_timeout_path = nxt_string("/tls/session/timeout"); + static nxt_str_t conf_tickets = nxt_string("/tls/session/tickets"); #endif static nxt_str_t static_path = nxt_string("/settings/http/static"); static nxt_str_t websocket_path = nxt_string("/settings/http/websocket"); @@ -1877,6 +1878,9 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf, tls_init->conf_cmds = nxt_conf_get_path(listener, &conf_commands_path); + tls_init->tickets_conf = nxt_conf_get_path(listener, + &conf_tickets); + if (nxt_conf_type(certificate) == NXT_CONF_ARRAY) { n = nxt_conf_array_elements_count(certificate); -- cgit