technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
28037b1f72cbff5fce053008216705baaa7dfcf6
nginx-unit/src/nxt_main_process.c
Andrew Clayton 47ff51009f Wasm: Add support for directory access. 
Due to the sandboxed nature of WebAssembly, by default WASM modules
don't have any access to the underlying filesystem.

There is however a capabilities based mechanism[0] for allowing such
access.

This adds a config option to the 'wasm' application type;
'access.filesystem' which takes an array of directory paths that are
then made available to the WASM module. This access works recursively,
i.e everything under a specific path is allowed access to.

Example config might look like

  "access" {
      "filesystem": [
          "/tmp",
          "/var/tmp"
      ]
  }

The actual mechanism used allows directories to be mapped differently in
the guest. But at the moment we don't support that and just map say /tmp
to /tmp. This can be revisited if it's something users clamour for.

Network sockets are another resource that may be controlled in this
manner, for example there is a wasi_config_preopen_socket() function,
however this requires the runtime to open the network socket then
effectively pass this through to the guest.

This is something that can be revisited in the future if users desire
it.

[0]:
<https://github.com/bytecodealliance/wasmtime/blob/main/docs/WASI-capabilities.md>

Reviewed-by: Alejandro Colomar <alx@nginx.com>
Signed-off-by: Andrew Clayton <a.clayton@nginx.com>
2023-08-17 13:09:49 +01:00

1716 lines
43 KiB
C
Raw Blame History

/*
* Copyright (C) Igor Sysoev
* Copyright (C) NGINX, Inc.
*/
#include <nxt_main.h>
#include <nxt_runtime.h>
#include <nxt_port.h>
#include <nxt_main_process.h>
#include <nxt_conf.h>
#include <nxt_router.h>
#include <nxt_port_queue.h>
#if (NXT_TLS)
#include <nxt_cert.h>
#endif
#if (NXT_HAVE_NJS)
#include <nxt_script.h>
#endif
#include <sys/mount.h>
typedef struct {
nxt_socket_t socket;
nxt_socket_error_t error;
u_char *start;
u_char *end;
} nxt_listening_socket_t;
typedef struct {
nxt_uint_t size;
nxt_conf_map_t *map;
} nxt_conf_app_map_t;
static nxt_int_t nxt_main_process_port_create(nxt_task_t *task,
nxt_runtime_t *rt);
static void nxt_main_process_title(nxt_task_t *task);
static void nxt_main_process_sigterm_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_sigquit_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_sigusr1_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_sigchld_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_signal_handler(nxt_task_t *task, void *obj,
void *data);
static void nxt_main_process_cleanup(nxt_task_t *task, nxt_process_t *process);
static void nxt_main_port_socket_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static void nxt_main_port_socket_unlink_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static nxt_int_t nxt_main_listening_socket(nxt_sockaddr_t *sa,
nxt_listening_socket_t *ls);
static void nxt_main_port_modules_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static int nxt_cdecl nxt_app_lang_compare(const void *v1, const void *v2);
static void nxt_main_process_whoami_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static void nxt_main_port_conf_store_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
static nxt_int_t nxt_main_file_store(nxt_task_t *task, const char *tmp_name,
const char *name, u_char *buf, size_t size);
static void nxt_main_port_access_log_handler(nxt_task_t *task,
nxt_port_recv_msg_t *msg);
const nxt_sig_event_t nxt_main_process_signals[] = {
nxt_event_signal(SIGHUP, nxt_main_process_signal_handler),
nxt_event_signal(SIGINT, nxt_main_process_sigterm_handler),
nxt_event_signal(SIGQUIT, nxt_main_process_sigquit_handler),
nxt_event_signal(SIGTERM, nxt_main_process_sigterm_handler),
nxt_event_signal(SIGCHLD, nxt_main_process_sigchld_handler),
nxt_event_signal(SIGUSR1, nxt_main_process_sigusr1_handler),
nxt_event_signal_end,
};
nxt_uint_t nxt_conf_ver;
static nxt_bool_t nxt_exiting;
nxt_int_t
nxt_main_process_start(nxt_thread_t *thr, nxt_task_t *task,
nxt_runtime_t *rt)
{
rt->type = NXT_PROCESS_MAIN;
if (nxt_main_process_port_create(task, rt) != NXT_OK) {
return NXT_ERROR;
}
nxt_main_process_title(task);
/*
* The discovery process will send a message processed by
* nxt_main_port_modules_handler() which starts the controller
* and router processes.
*/
return nxt_process_init_start(task, nxt_discovery_process);
}
static nxt_conf_map_t nxt_common_app_conf[] = {
{
nxt_string("type"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, type),
},
{
nxt_string("user"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, user),
},
{
nxt_string("group"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, group),
},
{
nxt_string("stdout"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, stdout_log),
},
{
nxt_string("stderr"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, stderr_log),
},
{
nxt_string("working_directory"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, working_directory),
},
{
nxt_string("environment"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, environment),
},
{
nxt_string("isolation"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, isolation),
},
{
nxt_string("limits"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, limits),
},
};
static nxt_conf_map_t nxt_common_app_limits_conf[] = {
{
nxt_string("shm"),
NXT_CONF_MAP_SIZE,
offsetof(nxt_common_app_conf_t, shm_limit),
},
{
nxt_string("requests"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, request_limit),
},
};
static nxt_conf_map_t nxt_external_app_conf[] = {
{
nxt_string("executable"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.external.executable),
},
{
nxt_string("arguments"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.external.arguments),
},
};
static nxt_conf_map_t nxt_python_app_conf[] = {
{
nxt_string("home"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.python.home),
},
{
nxt_string("path"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.python.path),
},
{
nxt_string("protocol"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.python.protocol),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.python.threads),
},
{
nxt_string("targets"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.python.targets),
},
{
nxt_string("thread_stack_size"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.python.thread_stack_size),
},
};
static nxt_conf_map_t nxt_php_app_conf[] = {
{
nxt_string("targets"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.php.targets),
},
{
nxt_string("options"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.php.options),
},
};
static nxt_conf_map_t nxt_perl_app_conf[] = {
{
nxt_string("script"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.perl.script),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.perl.threads),
},
{
nxt_string("thread_stack_size"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.perl.thread_stack_size),
},
};
static nxt_conf_map_t nxt_ruby_app_conf[] = {
{
nxt_string("script"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.ruby.script),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.ruby.threads),
},
{
nxt_string("hooks"),
NXT_CONF_MAP_STR,
offsetof(nxt_common_app_conf_t, u.ruby.hooks),
}
};
static nxt_conf_map_t nxt_java_app_conf[] = {
{
nxt_string("classpath"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.java.classpath),
},
{
nxt_string("webapp"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.java.webapp),
},
{
nxt_string("options"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.java.options),
},
{
nxt_string("unit_jars"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.java.unit_jars),
},
{
nxt_string("threads"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.java.threads),
},
{
nxt_string("thread_stack_size"),
NXT_CONF_MAP_INT32,
offsetof(nxt_common_app_conf_t, u.java.thread_stack_size),
},
};
static nxt_conf_map_t nxt_wasm_app_conf[] = {
{
nxt_string("module"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.module),
},
{
nxt_string("request_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.request_handler),
},
{
nxt_string("malloc_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.malloc_handler),
},
{
nxt_string("free_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.free_handler),
},
{
nxt_string("module_init_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.module_init_handler),
},
{
nxt_string("module_end_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.module_end_handler),
},
{
nxt_string("request_init_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.request_init_handler),
},
{
nxt_string("request_end_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.request_end_handler),
},
{
nxt_string("response_end_handler"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_common_app_conf_t, u.wasm.response_end_handler),
},
{
nxt_string("access"),
NXT_CONF_MAP_PTR,
offsetof(nxt_common_app_conf_t, u.wasm.access),
},
};
static nxt_conf_app_map_t nxt_app_maps[] = {
{ nxt_nitems(nxt_external_app_conf), nxt_external_app_conf },
{ nxt_nitems(nxt_python_app_conf), nxt_python_app_conf },
{ nxt_nitems(nxt_php_app_conf), nxt_php_app_conf },
{ nxt_nitems(nxt_perl_app_conf), nxt_perl_app_conf },
{ nxt_nitems(nxt_ruby_app_conf), nxt_ruby_app_conf },
{ nxt_nitems(nxt_java_app_conf), nxt_java_app_conf },
{ nxt_nitems(nxt_wasm_app_conf), nxt_wasm_app_conf },
};
static void
nxt_main_data_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_debug(task, "main data: %*s",
nxt_buf_mem_used_size(&msg->buf->mem), msg->buf->mem.pos);
}
static void
nxt_main_new_port_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
void *mem;
nxt_port_t *port;
nxt_port_new_port_handler(task, msg);
port = msg->u.new_port;
if (port != NULL
&& port->type == NXT_PROCESS_APP
&& msg->fd[1] != -1)
{
mem = nxt_mem_mmap(NULL, sizeof(nxt_port_queue_t),
PROT_READ | PROT_WRITE, MAP_SHARED, msg->fd[1], 0);
if (nxt_fast_path(mem != MAP_FAILED)) {
port->queue = mem;
}
nxt_fd_close(msg->fd[1]);
msg->fd[1] = -1;
}
}
static void
nxt_main_start_process_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
u_char *start, *p, ch;
size_t type_len;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *process;
nxt_app_type_t idx;
nxt_conf_value_t *conf;
nxt_process_init_t *init;
nxt_common_app_conf_t *app_conf;
rt = task->thread->runtime;
port = rt->port_by_type[NXT_PROCESS_ROUTER];
if (nxt_slow_path(port == NULL)) {
nxt_alert(task, "router port not found");
goto close_fds;
}
if (nxt_slow_path(port->pid != nxt_recv_msg_cmsg_pid(msg))) {
nxt_alert(task, "process %PI cannot start processes",
nxt_recv_msg_cmsg_pid(msg));
goto close_fds;
}
process = nxt_process_new(rt);
if (nxt_slow_path(process == NULL)) {
goto close_fds;
}
process->mem_pool = nxt_mp_create(1024, 128, 256, 32);
if (process->mem_pool == NULL) {
nxt_process_use(task, process, -1);
goto close_fds;
}
process->parent_port = rt->port_by_type[NXT_PROCESS_MAIN];
init = nxt_process_init(process);
*init = nxt_proto_process;
b = nxt_buf_chk_make_plain(process->mem_pool, msg->buf, msg->size);
if (b == NULL) {
goto failed;
}
nxt_debug(task, "main start prototype: %*s", b->mem.free - b->mem.pos,
b->mem.pos);
app_conf = nxt_mp_zalloc(process->mem_pool, sizeof(nxt_common_app_conf_t));
if (nxt_slow_path(app_conf == NULL)) {
goto failed;
}
app_conf->shared_port_fd = msg->fd[0];
app_conf->shared_queue_fd = msg->fd[1];
start = b->mem.pos;
app_conf->name.start = start;
app_conf->name.length = nxt_strlen(start);
init->name = (const char *) start;
process->name = nxt_mp_alloc(process->mem_pool, app_conf->name.length
+ sizeof("\"\" prototype") + 1);
if (nxt_slow_path(process->name == NULL)) {
goto failed;
}
p = (u_char *) process->name;
*p++ = '"';
p = nxt_cpymem(p, init->name, app_conf->name.length);
p = nxt_cpymem(p, "\" prototype", 11);
*p = '\0';
app_conf->shm_limit = 100 * 1024 * 1024;
app_conf->request_limit = 0;
start += app_conf->name.length + 1;
conf = nxt_conf_json_parse(process->mem_pool, start, b->mem.free, NULL);
if (conf == NULL) {
nxt_alert(task, "router app configuration parsing error");
goto failed;
}
rt = task->thread->runtime;
app_conf->user.start = (u_char*)rt->user_cred.user;
app_conf->user.length = nxt_strlen(rt->user_cred.user);
ret = nxt_conf_map_object(process->mem_pool, conf, nxt_common_app_conf,
nxt_nitems(nxt_common_app_conf), app_conf);
if (ret != NXT_OK) {
nxt_alert(task, "failed to map common app conf received from router");
goto failed;
}
for (type_len = 0; type_len != app_conf->type.length; type_len++) {
ch = app_conf->type.start[type_len];
if (ch == ' ' || nxt_isdigit(ch)) {
break;
}
}
idx = nxt_app_parse_type(app_conf->type.start, type_len);
if (nxt_slow_path(idx >= nxt_nitems(nxt_app_maps))) {
nxt_alert(task, "invalid app type %d received from router", (int) idx);
goto failed;
}
ret = nxt_conf_map_object(process->mem_pool, conf, nxt_app_maps[idx].map,
nxt_app_maps[idx].size, app_conf);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_alert(task, "failed to map app conf received from router");
goto failed;
}
if (app_conf->limits != NULL) {
ret = nxt_conf_map_object(process->mem_pool, app_conf->limits,
nxt_common_app_limits_conf,
nxt_nitems(nxt_common_app_limits_conf),
app_conf);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_alert(task, "failed to map app limits received from router");
goto failed;
}
}
app_conf->self = conf;
process->stream = msg->port_msg.stream;
process->data.app = app_conf;
ret = nxt_process_start(task, process);
if (nxt_fast_path(ret == NXT_OK || ret == NXT_AGAIN)) {
/* Close shared port fds only in main process. */
if (ret == NXT_OK) {
nxt_fd_close(app_conf->shared_port_fd);
nxt_fd_close(app_conf->shared_queue_fd);
}
/* Avoid fds close in caller. */
msg->fd[0] = -1;
msg->fd[1] = -1;
return;
}
failed:
nxt_process_use(task, process, -1);
port = nxt_runtime_port_find(rt, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_fast_path(port != NULL)) {
nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_ERROR,
-1, msg->port_msg.stream, 0, NULL);
}
close_fds:
nxt_fd_close(msg->fd[0]);
msg->fd[0] = -1;
nxt_fd_close(msg->fd[1]);
msg->fd[1] = -1;
}
static void
nxt_main_process_created_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_port_t *port;
nxt_process_t *process;
nxt_runtime_t *rt;
rt = task->thread->runtime;
port = nxt_runtime_port_find(rt, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_slow_path(port == NULL)) {
return;
}
process = port->process;
nxt_assert(process != NULL);
nxt_assert(process->state == NXT_PROCESS_STATE_CREATING);
#if (NXT_HAVE_LINUX_NS && NXT_HAVE_CLONE_NEWUSER)
if (nxt_is_clone_flag_set(process->isolation.clone.flags, NEWUSER)) {
if (nxt_slow_path(nxt_clone_credential_map(task, process->pid,
process->user_cred,
&process->isolation.clone)
!= NXT_OK))
{
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_ERROR,
-1, msg->port_msg.stream, 0, NULL);
return;
}
}
#endif
process->state = NXT_PROCESS_STATE_CREATED;
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_READY_LAST,
-1, msg->port_msg.stream, 0, NULL);
}
static nxt_port_handlers_t nxt_main_process_port_handlers = {
.data = nxt_main_data_handler,
.new_port = nxt_main_new_port_handler,
.process_created = nxt_main_process_created_handler,
.process_ready = nxt_port_process_ready_handler,
.whoami = nxt_main_process_whoami_handler,
.remove_pid = nxt_port_remove_pid_handler,
.start_process = nxt_main_start_process_handler,
.socket = nxt_main_port_socket_handler,
.socket_unlink = nxt_main_port_socket_unlink_handler,
.modules = nxt_main_port_modules_handler,
.conf_store = nxt_main_port_conf_store_handler,
#if (NXT_TLS)
.cert_get = nxt_cert_store_get_handler,
.cert_delete = nxt_cert_store_delete_handler,
#endif
#if (NXT_HAVE_NJS)
.script_get = nxt_script_store_get_handler,
.script_delete = nxt_script_store_delete_handler,
#endif
.access_log = nxt_main_port_access_log_handler,
.rpc_ready = nxt_port_rpc_handler,
.rpc_error = nxt_port_rpc_handler,
};
static void
nxt_main_process_whoami_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
nxt_buf_t *buf;
nxt_pid_t pid, ppid;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_process_t *pprocess;
nxt_assert(msg->port_msg.reply_port == 0);
if (nxt_slow_path(msg->buf == NULL
|| nxt_buf_used_size(msg->buf) != sizeof(nxt_pid_t)))
{
nxt_alert(task, "whoami: buffer is NULL or unexpected size");
goto fail;
}
nxt_memcpy(&ppid, msg->buf->mem.pos, sizeof(nxt_pid_t));
rt = task->thread->runtime;
pprocess = nxt_runtime_process_find(rt, ppid);
if (nxt_slow_path(pprocess == NULL)) {
nxt_alert(task, "whoami: parent process %PI not found", ppid);
goto fail;
}
pid = nxt_recv_msg_cmsg_pid(msg);
nxt_debug(task, "whoami: from %PI, parent %PI, fd %d", pid, ppid,
msg->fd[0]);
if (msg->fd[0] != -1) {
port = nxt_runtime_process_port_create(task, rt, pid, 0,
NXT_PROCESS_APP);
if (nxt_slow_path(port == NULL)) {
goto fail;
}
nxt_fd_nonblocking(task, msg->fd[0]);
port->pair[0] = -1;
port->pair[1] = msg->fd[0];
msg->fd[0] = -1;
port->max_size = 16 * 1024;
port->max_share = 64 * 1024;
port->socket.task = task;
nxt_port_write_enable(task, port);
} else {
port = nxt_runtime_port_find(rt, pid, 0);
if (nxt_slow_path(port == NULL)) {
goto fail;
}
}
if (ppid != nxt_pid) {
nxt_queue_insert_tail(&pprocess->children, &port->process->link);
}
buf = nxt_buf_mem_alloc(task->thread->engine->mem_pool,
sizeof(nxt_pid_t), 0);
if (nxt_slow_path(buf == NULL)) {
goto fail;
}
buf->mem.free = nxt_cpymem(buf->mem.free, &pid, sizeof(nxt_pid_t));
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_READY_LAST, -1,
msg->port_msg.stream, 0, buf);
fail:
if (msg->fd[0] != -1) {
nxt_fd_close(msg->fd[0]);
}
}
static nxt_int_t
nxt_main_process_port_create(nxt_task_t *task, nxt_runtime_t *rt)
{
nxt_int_t ret;
nxt_port_t *port;
nxt_process_t *process;
port = nxt_runtime_process_port_create(task, rt, nxt_pid, 0,
NXT_PROCESS_MAIN);
if (nxt_slow_path(port == NULL)) {
return NXT_ERROR;
}
process = port->process;
ret = nxt_port_socket_init(task, port, 0);
if (nxt_slow_path(ret != NXT_OK)) {
nxt_port_use(task, port, -1);
return ret;
}
/*
* A main process port. A write port is not closed
* since it should be inherited by processes.
*/
nxt_port_enable(task, port, &nxt_main_process_port_handlers);
process->state = NXT_PROCESS_STATE_READY;
return NXT_OK;
}
static void
nxt_main_process_title(nxt_task_t *task)
{
u_char *p, *end;
nxt_uint_t i;
u_char title[2048];
end = title + sizeof(title) - 1;
p = nxt_sprintf(title, end, "unit: main v" NXT_VERSION " [%s",
nxt_process_argv[0]);
for (i = 1; nxt_process_argv[i] != NULL; i++) {
p = nxt_sprintf(p, end, " %s", nxt_process_argv[i]);
}
if (p < end) {
*p++ = ']';
}
*p = '\0';
nxt_process_title(task, "%s", title);
}
static void
nxt_main_process_sigterm_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_debug(task, "sigterm handler signo:%d (%s)",
(int) (uintptr_t) obj, data);
/* TODO: fast exit. */
nxt_exiting = 1;
nxt_runtime_quit(task, 0);
}
static void
nxt_main_process_sigquit_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_debug(task, "sigquit handler signo:%d (%s)",
(int) (uintptr_t) obj, data);
/* TODO: graceful exit. */
nxt_exiting = 1;
nxt_runtime_quit(task, 0);
}
static void
nxt_main_process_sigusr1_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_mp_t *mp;
nxt_int_t ret;
nxt_uint_t n;
nxt_port_t *port;
nxt_file_t *file, *new_file;
nxt_array_t *new_files;
nxt_runtime_t *rt;
nxt_log(task, NXT_LOG_NOTICE, "signal %d (%s) recevied, %s",
(int) (uintptr_t) obj, data, "log files rotation");
rt = task->thread->runtime;
port = rt->port_by_type[NXT_PROCESS_ROUTER];
if (nxt_fast_path(port != NULL)) {
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_ACCESS_LOG,
-1, 0, 0, NULL);
}
mp = nxt_mp_create(1024, 128, 256, 32);
if (mp == NULL) {
return;
}
n = nxt_list_nelts(rt->log_files);
new_files = nxt_array_create(mp, n, sizeof(nxt_file_t));
if (new_files == NULL) {
nxt_mp_destroy(mp);
return;
}
nxt_list_each(file, rt->log_files) {
/* This allocation cannot fail. */
new_file = nxt_array_add(new_files);
new_file->name = file->name;
new_file->fd = NXT_FILE_INVALID;
new_file->log_level = NXT_LOG_ALERT;
ret = nxt_file_open(task, new_file, O_WRONLY | O_APPEND, O_CREAT,
NXT_FILE_OWNER_ACCESS);
if (ret != NXT_OK) {
goto fail;
}
} nxt_list_loop;
new_file = new_files->elts;
ret = nxt_file_stderr(&new_file[0]);
if (ret == NXT_OK) {
n = 0;
nxt_list_each(file, rt->log_files) {
nxt_port_change_log_file(task, rt, n, new_file[n].fd);
/*
* The old log file descriptor must be closed at the moment
* when no other threads use it. dup2() allows to use the
* old file descriptor for new log file. This change is
* performed atomically in the kernel.
*/
(void) nxt_file_redirect(file, new_file[n].fd);
n++;
} nxt_list_loop;
nxt_mp_destroy(mp);
return;
}
fail:
new_file = new_files->elts;
n = new_files->nelts;
while (n != 0) {
if (new_file->fd != NXT_FILE_INVALID) {
nxt_file_close(task, new_file);
}
new_file++;
n--;
}
nxt_mp_destroy(mp);
}
static void
nxt_main_process_sigchld_handler(nxt_task_t *task, void *obj, void *data)
{
int status;
nxt_int_t ret;
nxt_err_t err;
nxt_pid_t pid;
nxt_port_t *port;
nxt_queue_t children;
nxt_runtime_t *rt;
nxt_process_t *process, *child;
nxt_process_init_t init;
nxt_debug(task, "sigchld handler signo:%d (%s)",
(int) (uintptr_t) obj, data);
rt = task->thread->runtime;
for ( ;; ) {
pid = waitpid(-1, &status, WNOHANG);
if (pid == -1) {
switch (err = nxt_errno) {
case NXT_ECHILD:
return;
case NXT_EINTR:
continue;
default:
nxt_alert(task, "waitpid() failed: %E", err);
return;
}
}
nxt_debug(task, "waitpid(): %PI", pid);
if (pid == 0) {
return;
}
if (WTERMSIG(status)) {
#ifdef WCOREDUMP
nxt_alert(task, "process %PI exited on signal %d%s",
pid, WTERMSIG(status),
WCOREDUMP(status) ? " (core dumped)" : "");
#else
nxt_alert(task, "process %PI exited on signal %d",
pid, WTERMSIG(status));
#endif
} else {
nxt_trace(task, "process %PI exited with code %d",
pid, WEXITSTATUS(status));
}
process = nxt_runtime_process_find(rt, pid);
if (process != NULL) {
nxt_main_process_cleanup(task, process);
if (process->state == NXT_PROCESS_STATE_READY) {
process->stream = 0;
}
nxt_queue_init(&children);
if (!nxt_queue_is_empty(&process->children)) {
nxt_queue_add(&children, &process->children);
nxt_queue_init(&process->children);
nxt_queue_each(child, &children, nxt_process_t, link) {
port = nxt_process_port_first(child);
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_QUIT,
-1, 0, 0, NULL);
} nxt_queue_loop;
}
if (nxt_exiting) {
nxt_process_close_ports(task, process);
nxt_queue_each(child, &children, nxt_process_t, link) {
nxt_queue_remove(&child->link);
child->link.next = NULL;
nxt_process_close_ports(task, child);
} nxt_queue_loop;
if (rt->nprocesses <= 1) {
nxt_runtime_quit(task, 0);
return;
}
continue;
}
nxt_port_remove_notify_others(task, process);
nxt_queue_each(child, &children, nxt_process_t, link) {
nxt_port_remove_notify_others(task, child);
nxt_queue_remove(&child->link);
child->link.next = NULL;
nxt_process_close_ports(task, child);
} nxt_queue_loop;
init = *(nxt_process_init_t *) nxt_process_init(process);
nxt_process_close_ports(task, process);
if (init.restart) {
ret = nxt_process_init_start(task, init);
if (nxt_slow_path(ret == NXT_ERROR)) {
nxt_alert(task, "failed to restart %s", init.name);
}
}
}
}
}
static void
nxt_main_process_signal_handler(nxt_task_t *task, void *obj, void *data)
{
nxt_trace(task, "signal signo:%d (%s) recevied, ignored",
(int) (uintptr_t) obj, data);
}
static void
nxt_main_process_cleanup(nxt_task_t *task, nxt_process_t *process)
{
if (process->isolation.cleanup != NULL) {
process->isolation.cleanup(task, process);
}
if (process->isolation.cgroup_cleanup != NULL) {
process->isolation.cgroup_cleanup(task, process);
}
}
static void
nxt_main_port_socket_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
size_t size;
nxt_int_t ret;
nxt_buf_t *b, *out;
nxt_port_t *port;
nxt_sockaddr_t *sa;
nxt_port_msg_type_t type;
nxt_listening_socket_t ls;
u_char message[2048];
port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_slow_path(port == NULL)) {
return;
}
if (nxt_slow_path(port->type != NXT_PROCESS_ROUTER)) {
nxt_alert(task, "process %PI cannot create listener sockets",
msg->port_msg.pid);
return;
}
b = msg->buf;
sa = (nxt_sockaddr_t *) b->mem.pos;
/* TODO check b size and make plain */
ls.socket = -1;
ls.error = NXT_SOCKET_ERROR_SYSTEM;
ls.start = message;
ls.end = message + sizeof(message);
nxt_debug(task, "listening socket \"%*s\"",
(size_t) sa->length, nxt_sockaddr_start(sa));
ret = nxt_main_listening_socket(sa, &ls);
if (ret == NXT_OK) {
nxt_debug(task, "socket(\"%*s\"): %d",
(size_t) sa->length, nxt_sockaddr_start(sa), ls.socket);
out = NULL;
type = NXT_PORT_MSG_RPC_READY_LAST | NXT_PORT_MSG_CLOSE_FD;
} else {
size = ls.end - ls.start;
nxt_alert(task, "%*s", size, ls.start);
out = nxt_buf_mem_ts_alloc(task, task->thread->engine->mem_pool,
size + 1);
if (nxt_fast_path(out != NULL)) {
*out->mem.free++ = (uint8_t) ls.error;
out->mem.free = nxt_cpymem(out->mem.free, ls.start, size);
}
type = NXT_PORT_MSG_RPC_ERROR;
}
nxt_port_socket_write(task, port, type, ls.socket, msg->port_msg.stream,
0, out);
}
static nxt_int_t
nxt_main_listening_socket(nxt_sockaddr_t *sa, nxt_listening_socket_t *ls)
{
nxt_err_t err;
nxt_socket_t s;
const socklen_t length = sizeof(int);
static const int enable = 1;
s = socket(sa->u.sockaddr.sa_family, sa->type, 0);
if (nxt_slow_path(s == -1)) {
err = nxt_errno;
#if (NXT_INET6)
if (err == EAFNOSUPPORT && sa->u.sockaddr.sa_family == AF_INET6) {
ls->error = NXT_SOCKET_ERROR_NOINET6;
}
#endif
ls->end = nxt_sprintf(ls->start, ls->end,
"socket(\\\"%*s\\\") failed %E",
(size_t) sa->length, nxt_sockaddr_start(sa), err);
return NXT_ERROR;
}
if (setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &enable, length) != 0) {
ls->end = nxt_sprintf(ls->start, ls->end,
"setsockopt(\\\"%*s\\\", SO_REUSEADDR) failed %E",
(size_t) sa->length, nxt_sockaddr_start(sa),
nxt_errno);
goto fail;
}
#if (NXT_INET6)
if (sa->u.sockaddr.sa_family == AF_INET6) {
if (setsockopt(s, IPPROTO_IPV6, IPV6_V6ONLY, &enable, length) != 0) {
ls->end = nxt_sprintf(ls->start, ls->end,
"setsockopt(\\\"%*s\\\", IPV6_V6ONLY) failed %E",
(size_t) sa->length, nxt_sockaddr_start(sa),
nxt_errno);
goto fail;
}
}
#endif
if (bind(s, &sa->u.sockaddr, sa->socklen) != 0) {
err = nxt_errno;
#if (NXT_HAVE_UNIX_DOMAIN)
if (sa->u.sockaddr.sa_family == AF_UNIX) {
switch (err) {
case EACCES:
ls->error = NXT_SOCKET_ERROR_ACCESS;
break;
case ENOENT:
case ENOTDIR:
ls->error = NXT_SOCKET_ERROR_PATH;
break;
}
} else
#endif
{
switch (err) {
case EACCES:
ls->error = NXT_SOCKET_ERROR_PORT;
break;
case EADDRINUSE:
ls->error = NXT_SOCKET_ERROR_INUSE;
break;
case EADDRNOTAVAIL:
ls->error = NXT_SOCKET_ERROR_NOADDR;
break;
}
}
ls->end = nxt_sprintf(ls->start, ls->end, "bind(\\\"%*s\\\") failed %E",
(size_t) sa->length, nxt_sockaddr_start(sa), err);
goto fail;
}
#if (NXT_HAVE_UNIX_DOMAIN)
if (sa->u.sockaddr.sa_family == AF_UNIX
&& sa->u.sockaddr_un.sun_path[0] != '\0')
{
char *filename;
mode_t access;
nxt_thread_t *thr;
filename = sa->u.sockaddr_un.sun_path;
access = (S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH);
if (chmod(filename, access) != 0) {
ls->end = nxt_sprintf(ls->start, ls->end,
"chmod(\\\"%s\\\") failed %E",
filename, nxt_errno);
goto fail;
}
thr = nxt_thread();
nxt_runtime_listen_socket_add(thr->runtime, sa);
}
#endif
ls->socket = s;
return NXT_OK;
fail:
(void) close(s);
return NXT_ERROR;
}
static void
nxt_main_port_socket_unlink_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
#if (NXT_HAVE_UNIX_DOMAIN)
size_t i;
nxt_buf_t *b;
const char *filename;
nxt_runtime_t *rt;
nxt_sockaddr_t *sa;
nxt_listen_socket_t *ls;
b = msg->buf;
sa = (nxt_sockaddr_t *) b->mem.pos;
filename = sa->u.sockaddr_un.sun_path;
unlink(filename);
rt = task->thread->runtime;
for (i = 0; i < rt->listen_sockets->nelts; i++) {
const char *name;
ls = (nxt_listen_socket_t *) rt->listen_sockets->elts + i;
sa = ls->sockaddr;
if (sa->u.sockaddr.sa_family != AF_UNIX
|| sa->u.sockaddr_un.sun_path[0] == '\0')
{
continue;
}
name = sa->u.sockaddr_un.sun_path;
if (strcmp(name, filename) != 0) {
continue;
}
nxt_array_remove(rt->listen_sockets, ls);
break;
}
#endif
}
static nxt_conf_map_t nxt_app_lang_module_map[] = {
{
nxt_string("type"),
NXT_CONF_MAP_INT,
offsetof(nxt_app_lang_module_t, type),
},
{
nxt_string("version"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_app_lang_module_t, version),
},
{
nxt_string("file"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_app_lang_module_t, file),
},
};
static nxt_conf_map_t nxt_app_lang_mounts_map[] = {
{
nxt_string("src"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, src),
},
{
nxt_string("dst"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, dst),
},
{
nxt_string("name"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, name),
},
{
nxt_string("type"),
NXT_CONF_MAP_INT,
offsetof(nxt_fs_mount_t, type),
},
{
nxt_string("flags"),
NXT_CONF_MAP_INT,
offsetof(nxt_fs_mount_t, flags),
},
{
nxt_string("data"),
NXT_CONF_MAP_CSTRZ,
offsetof(nxt_fs_mount_t, data),
},
};
static void
nxt_main_port_modules_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
uint32_t index, jindex, nmounts;
nxt_mp_t *mp;
nxt_int_t ret;
nxt_buf_t *b;
nxt_port_t *port;
nxt_runtime_t *rt;
nxt_fs_mount_t *mnt;
nxt_conf_value_t *conf, *root, *value, *mounts;
nxt_app_lang_module_t *lang;
static nxt_str_t root_path = nxt_string("/");
static nxt_str_t mounts_name = nxt_string("mounts");
rt = task->thread->runtime;
if (msg->port_msg.pid != rt->port_by_type[NXT_PROCESS_DISCOVERY]->pid) {
nxt_alert(task, "process %PI cannot send modules", msg->port_msg.pid);
return;
}
if (nxt_exiting) {
nxt_debug(task, "ignoring discovered modules, exiting");
return;
}
port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_fast_path(port != NULL)) {
(void) nxt_port_socket_write(task, port, NXT_PORT_MSG_RPC_ERROR, -1,
msg->port_msg.stream, 0, NULL);
}
b = msg->buf;
if (b == NULL) {
return;
}
mp = nxt_mp_create(1024, 128, 256, 32);
if (mp == NULL) {
return;
}
b = nxt_buf_chk_make_plain(mp, b, msg->size);
if (b == NULL) {
return;
}
nxt_debug(task, "application languages: \"%*s\"",
b->mem.free - b->mem.pos, b->mem.pos);
conf = nxt_conf_json_parse(mp, b->mem.pos, b->mem.free, NULL);
if (conf == NULL) {
goto fail;
}
root = nxt_conf_get_path(conf, &root_path);
if (root == NULL) {
goto fail;
}
for (index = 0; /* void */ ; index++) {
value = nxt_conf_get_array_element(root, index);
if (value == NULL) {
break;
}
lang = nxt_array_zero_add(rt->languages);
if (lang == NULL) {
goto fail;
}
lang->module = NULL;
ret = nxt_conf_map_object(rt->mem_pool, value, nxt_app_lang_module_map,
nxt_nitems(nxt_app_lang_module_map), lang);
if (ret != NXT_OK) {
goto fail;
}
mounts = nxt_conf_get_object_member(value, &mounts_name, NULL);
if (mounts == NULL) {
nxt_alert(task, "missing mounts from discovery message.");
goto fail;
}
if (nxt_conf_type(mounts) != NXT_CONF_ARRAY) {
nxt_alert(task, "invalid mounts type from discovery message.");
goto fail;
}
nmounts = nxt_conf_array_elements_count(mounts);
lang->mounts = nxt_array_create(rt->mem_pool, nmounts,
sizeof(nxt_fs_mount_t));
if (lang->mounts == NULL) {
goto fail;
}
for (jindex = 0; /* */; jindex++) {
value = nxt_conf_get_array_element(mounts, jindex);
if (value == NULL) {
break;
}
mnt = nxt_array_zero_add(lang->mounts);
if (mnt == NULL) {
goto fail;
}
mnt->builtin = 1;
mnt->deps = 1;
ret = nxt_conf_map_object(rt->mem_pool, value,
nxt_app_lang_mounts_map,
nxt_nitems(nxt_app_lang_mounts_map), mnt);
if (ret != NXT_OK) {
goto fail;
}
}
nxt_debug(task, "lang %d %s \"%s\" (%d mounts)",
lang->type, lang->version, lang->file, lang->mounts->nelts);
}
qsort(rt->languages->elts, rt->languages->nelts,
sizeof(nxt_app_lang_module_t), nxt_app_lang_compare);
fail:
nxt_mp_destroy(mp);
ret = nxt_process_init_start(task, nxt_controller_process);
if (ret == NXT_OK) {
ret = nxt_process_init_start(task, nxt_router_process);
}
if (nxt_slow_path(ret == NXT_ERROR)) {
nxt_exiting = 1;
nxt_runtime_quit(task, 1);
}
}
static int nxt_cdecl
nxt_app_lang_compare(const void *v1, const void *v2)
{
int n;
const nxt_app_lang_module_t *lang1, *lang2;
lang1 = v1;
lang2 = v2;
n = lang1->type - lang2->type;
if (n != 0) {
return n;
}
n = nxt_strverscmp(lang1->version, lang2->version);
/* Negate result to move higher versions to the beginning. */
return -n;
}
static void
nxt_main_port_conf_store_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
void *p;
size_t n, size;
nxt_int_t ret;
nxt_port_t *ctl_port;
nxt_runtime_t *rt;
u_char ver[NXT_INT_T_LEN];
rt = task->thread->runtime;
ctl_port = rt->port_by_type[NXT_PROCESS_CONTROLLER];
if (nxt_slow_path(msg->port_msg.pid != ctl_port->pid)) {
nxt_alert(task, "process %PI cannot store conf", msg->port_msg.pid);
return;
}
p = MAP_FAILED;
/*
* Ancient compilers like gcc 4.8.5 on CentOS 7 wants 'size' to be
* initialized in 'cleanup' section.
*/
size = 0;
if (nxt_slow_path(msg->fd[0] == -1)) {
nxt_alert(task, "conf_store_handler: invalid shm fd");
goto error;
}
if (nxt_buf_mem_used_size(&msg->buf->mem) != sizeof(size_t)) {
nxt_alert(task, "conf_store_handler: unexpected buffer size (%d)",
(int) nxt_buf_mem_used_size(&msg->buf->mem));
goto error;
}
nxt_memcpy(&size, msg->buf->mem.pos, sizeof(size_t));
p = nxt_mem_mmap(NULL, size, PROT_READ, MAP_SHARED, msg->fd[0], 0);
nxt_fd_close(msg->fd[0]);
msg->fd[0] = -1;
if (nxt_slow_path(p == MAP_FAILED)) {
goto error;
}
nxt_debug(task, "conf_store_handler(%uz): %*s", size, size, p);
if (nxt_conf_ver != NXT_VERNUM) {
n = nxt_sprintf(ver, ver + NXT_INT_T_LEN, "%d", NXT_VERNUM) - ver;
ret = nxt_main_file_store(task, rt->ver_tmp, rt->ver, ver, n);
if (nxt_slow_path(ret != NXT_OK)) {
goto error;
}
nxt_conf_ver = NXT_VERNUM;
}
ret = nxt_main_file_store(task, rt->conf_tmp, rt->conf, p, size);
if (nxt_fast_path(ret == NXT_OK)) {
goto cleanup;
}
error:
nxt_alert(task, "failed to store current configuration");
cleanup:
if (p != MAP_FAILED) {
nxt_mem_munmap(p, size);
}
if (msg->fd[0] != -1) {
nxt_fd_close(msg->fd[0]);
msg->fd[0] = -1;
}
}
static nxt_int_t
nxt_main_file_store(nxt_task_t *task, const char *tmp_name, const char *name,
u_char *buf, size_t size)
{
ssize_t n;
nxt_int_t ret;
nxt_file_t file;
nxt_memzero(&file, sizeof(nxt_file_t));
file.name = (nxt_file_name_t *) name;
ret = nxt_file_open(task, &file, NXT_FILE_WRONLY, NXT_FILE_TRUNCATE,
NXT_FILE_OWNER_ACCESS);
if (nxt_slow_path(ret != NXT_OK)) {
return NXT_ERROR;
}
n = nxt_file_write(&file, buf, size, 0);
nxt_file_close(task, &file);
if (nxt_slow_path(n != (ssize_t) size)) {
(void) nxt_file_delete(file.name);
return NXT_ERROR;
}
return nxt_file_rename(file.name, (nxt_file_name_t *) name);
}
static void
nxt_main_port_access_log_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
u_char *path;
nxt_int_t ret;
nxt_file_t file;
nxt_port_t *port;
nxt_port_msg_type_t type;
nxt_debug(task, "opening access log file");
path = msg->buf->mem.pos;
nxt_memzero(&file, sizeof(nxt_file_t));
file.name = (nxt_file_name_t *) path;
file.log_level = NXT_LOG_ERR;
ret = nxt_file_open(task, &file, O_WRONLY | O_APPEND, O_CREAT,
NXT_FILE_OWNER_ACCESS);
type = (ret == NXT_OK) ? NXT_PORT_MSG_RPC_READY_LAST | NXT_PORT_MSG_CLOSE_FD
: NXT_PORT_MSG_RPC_ERROR;
port = nxt_runtime_port_find(task->thread->runtime, msg->port_msg.pid,
msg->port_msg.reply_port);
if (nxt_fast_path(port != NULL)) {
(void) nxt_port_socket_write(task, port, type, file.fd,
msg->port_msg.stream, 0, NULL);
}
}
Reference in New Issue View Git Blame Copy Permalink