diff options
Diffstat (limited to 'sources')
| -rw-r--r-- | sources/core.c | 433 | ||||
| -rw-r--r-- | sources/core.h | 10 | ||||
| -rw-r--r-- | sources/dbi.c | 8 | ||||
| -rw-r--r-- | sources/main.c | 69 | ||||
| -rw-r--r-- | sources/ogAdmLib.h | 3 | ||||
| -rw-r--r-- | sources/ogAdmServer.c | 474 | ||||
| -rw-r--r-- | sources/rest.c | 2 | ||||
| -rw-r--r-- | sources/utils.c | 8 |
8 files changed, 534 insertions, 473 deletions
diff --git a/sources/core.c b/sources/core.c new file mode 100644 index 0000000..f7c25f5 --- /dev/null +++ b/sources/core.c @@ -0,0 +1,433 @@ +/* + * Copyright (C) 2020 Soleta Networks <info@soleta.eu> + * + * This program is free software: you can redistribute it and/or modify it under + * the terms of the GNU Affero General Public License as published by the + * Free Software Foundation, version 3. + */ + +#include "ogAdmServer.h" +#include "dbi.h" +#include "utils.h" +#include "list.h" +#include "rest.h" +#include "client.h" +#include "json.h" +#include "schedule.h" +#include <syslog.h> +#include <sys/ioctl.h> +#include <ifaddrs.h> +#include <sys/types.h> +#include <sys/stat.h> +#include <fcntl.h> +#include <jansson.h> +#include <time.h> + +static void og_client_release(struct ev_loop *loop, struct og_client *cli) +{ + if (cli->keepalive_idx >= 0) { + syslog(LOG_DEBUG, "closing keepalive connection for %s:%hu in slot %d\n", + inet_ntoa(cli->addr.sin_addr), + ntohs(cli->addr.sin_port), cli->keepalive_idx); + tbsockets[cli->keepalive_idx].cli = NULL; + } + + list_del(&cli->list); + ev_io_stop(loop, &cli->io); + close(cli->io.fd); + free(cli); +} + +static void og_client_keepalive(struct ev_loop *loop, struct og_client *cli) +{ + struct og_client *old_cli; + + old_cli = tbsockets[cli->keepalive_idx].cli; + if (old_cli && old_cli != cli) { + syslog(LOG_DEBUG, "closing old keepalive connection for %s:%hu\n", + inet_ntoa(old_cli->addr.sin_addr), + ntohs(old_cli->addr.sin_port)); + + og_client_release(loop, old_cli); + } + tbsockets[cli->keepalive_idx].cli = cli; +} + +static void og_client_reset_state(struct og_client *cli) +{ + cli->state = OG_CLIENT_RECEIVING_HEADER; + cli->buf_len = 0; +} + +static int og_client_payload_too_large(struct og_client *cli) +{ + char buf[] = "HTTP/1.1 413 Payload Too Large\r\n" + "Content-Length: 0\r\n\r\n"; + + send(og_client_socket(cli), buf, strlen(buf), 0); + + return -1; +} + +static int og_client_state_recv_hdr_rest(struct og_client *cli) +{ + char *ptr; + + ptr = strstr(cli->buf, "\r\n\r\n"); + if (!ptr) + return 0; + + cli->msg_len = ptr - cli->buf + 4; + + ptr = strstr(cli->buf, "Content-Length: "); + if (ptr) { + sscanf(ptr, "Content-Length: %i[^\r\n]", &cli->content_length); + if (cli->content_length < 0) + return -1; + cli->msg_len += cli->content_length; + } + + ptr = strstr(cli->buf, "Authorization: "); + if (ptr) + sscanf(ptr, "Authorization: %63[^\r\n]", cli->auth_token); + + return 1; +} + +static int og_client_recv(struct og_client *cli, int events) +{ + struct ev_io *io = &cli->io; + int ret; + + if (events & EV_ERROR) { + syslog(LOG_ERR, "unexpected error event from client %s:%hu\n", + inet_ntoa(cli->addr.sin_addr), + ntohs(cli->addr.sin_port)); + return 0; + } + + ret = recv(io->fd, cli->buf + cli->buf_len, + sizeof(cli->buf) - cli->buf_len, 0); + if (ret <= 0) { + if (ret < 0) { + syslog(LOG_ERR, "error reading from client %s:%hu (%s)\n", + inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port), + strerror(errno)); + } else { + syslog(LOG_DEBUG, "closed connection by %s:%hu\n", + inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); + } + return ret; + } + + return ret; +} + +static void og_client_read_cb(struct ev_loop *loop, struct ev_io *io, int events) +{ + struct og_client *cli; + int ret; + + cli = container_of(io, struct og_client, io); + + ret = og_client_recv(cli, events); + if (ret <= 0) + goto close; + + if (cli->keepalive_idx >= 0) + return; + + ev_timer_again(loop, &cli->timer); + + cli->buf_len += ret; + if (cli->buf_len >= sizeof(cli->buf)) { + syslog(LOG_ERR, "client request from %s:%hu is too long\n", + inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); + og_client_payload_too_large(cli); + goto close; + } + + switch (cli->state) { + case OG_CLIENT_RECEIVING_HEADER: + ret = og_client_state_recv_hdr_rest(cli); + if (ret < 0) + goto close; + if (!ret) + return; + + cli->state = OG_CLIENT_RECEIVING_PAYLOAD; + /* Fall through. */ + case OG_CLIENT_RECEIVING_PAYLOAD: + /* Still not enough data to process request. */ + if (cli->buf_len < cli->msg_len) + return; + + cli->state = OG_CLIENT_PROCESSING_REQUEST; + /* fall through. */ + case OG_CLIENT_PROCESSING_REQUEST: + ret = og_client_state_process_payload_rest(cli); + if (ret < 0) { + syslog(LOG_ERR, "Failed to process HTTP request from %s:%hu\n", + inet_ntoa(cli->addr.sin_addr), + ntohs(cli->addr.sin_port)); + } + if (ret < 0) + goto close; + + if (cli->keepalive_idx < 0) { + syslog(LOG_DEBUG, "server closing connection to %s:%hu\n", + inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); + goto close; + } else { + syslog(LOG_DEBUG, "leaving client %s:%hu in keepalive mode\n", + inet_ntoa(cli->addr.sin_addr), + ntohs(cli->addr.sin_port)); + og_client_keepalive(loop, cli); + og_client_reset_state(cli); + } + break; + default: + syslog(LOG_ERR, "unknown state, critical internal error\n"); + goto close; + } + return; +close: + ev_timer_stop(loop, &cli->timer); + og_client_release(loop, cli); +} + +enum og_agent_state { + OG_AGENT_RECEIVING_HEADER = 0, + OG_AGENT_RECEIVING_PAYLOAD, + OG_AGENT_PROCESSING_RESPONSE, +}; + +static int og_agent_state_recv_hdr_rest(struct og_client *cli) +{ + char *ptr; + + ptr = strstr(cli->buf, "\r\n\r\n"); + if (!ptr) + return 0; + + cli->msg_len = ptr - cli->buf + 4; + + ptr = strstr(cli->buf, "Content-Length: "); + if (ptr) { + sscanf(ptr, "Content-Length: %i[^\r\n]", &cli->content_length); + if (cli->content_length < 0) + return -1; + cli->msg_len += cli->content_length; + } + + return 1; +} + +static void og_agent_reset_state(struct og_client *cli) +{ + cli->state = OG_AGENT_RECEIVING_HEADER; + cli->buf_len = 0; + cli->content_length = 0; + memset(cli->buf, 0, sizeof(cli->buf)); +} + +static void og_agent_deliver_pending_cmd(struct og_client *cli) +{ + const struct og_cmd *cmd; + + cmd = og_cmd_find(inet_ntoa(cli->addr.sin_addr)); + if (!cmd) + return; + + og_send_request(cmd->method, cmd->type, &cmd->params, cmd->json); + cli->last_cmd_id = cmd->id; + + og_cmd_free(cmd); +} + +static void og_agent_read_cb(struct ev_loop *loop, struct ev_io *io, int events) +{ + struct og_client *cli; + int ret; + + cli = container_of(io, struct og_client, io); + + ret = og_client_recv(cli, events); + if (ret <= 0) + goto close; + + ev_timer_again(loop, &cli->timer); + + cli->buf_len += ret; + if (cli->buf_len >= sizeof(cli->buf)) { + syslog(LOG_ERR, "client request from %s:%hu is too long\n", + inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); + goto close; + } + + switch (cli->state) { + case OG_AGENT_RECEIVING_HEADER: + ret = og_agent_state_recv_hdr_rest(cli); + if (ret < 0) + goto close; + if (!ret) + return; + + cli->state = OG_AGENT_RECEIVING_PAYLOAD; + /* Fall through. */ + case OG_AGENT_RECEIVING_PAYLOAD: + /* Still not enough data to process request. */ + if (cli->buf_len < cli->msg_len) + return; + + cli->state = OG_AGENT_PROCESSING_RESPONSE; + /* fall through. */ + case OG_AGENT_PROCESSING_RESPONSE: + ret = og_agent_state_process_response(cli); + if (ret < 0) { + syslog(LOG_ERR, "Failed to process HTTP request from %s:%hu\n", + inet_ntoa(cli->addr.sin_addr), + ntohs(cli->addr.sin_port)); + goto close; + } else if (ret == 0) { + og_agent_deliver_pending_cmd(cli); + } + + syslog(LOG_DEBUG, "leaving client %s:%hu in keepalive mode\n", + inet_ntoa(cli->addr.sin_addr), + ntohs(cli->addr.sin_port)); + og_agent_reset_state(cli); + break; + default: + syslog(LOG_ERR, "unknown state, critical internal error\n"); + goto close; + } + return; +close: + ev_timer_stop(loop, &cli->timer); + og_client_release(loop, cli); +} + +static void og_client_timer_cb(struct ev_loop *loop, ev_timer *timer, int events) +{ + struct og_client *cli; + + cli = container_of(timer, struct og_client, timer); + if (cli->keepalive_idx >= 0) { + ev_timer_again(loop, &cli->timer); + return; + } + syslog(LOG_ERR, "timeout request for client %s:%hu\n", + inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); + + og_client_release(loop, cli); +} + +static void og_agent_send_refresh(struct og_client *cli) +{ + struct og_msg_params params; + int err; + + params.ips_array[0] = inet_ntoa(cli->addr.sin_addr); + params.ips_array_len = 1; + + err = og_send_request(OG_METHOD_GET, OG_CMD_REFRESH, ¶ms, NULL); + if (err < 0) { + syslog(LOG_ERR, "Can't send refresh to: %s\n", + params.ips_array[0]); + } else { + syslog(LOG_INFO, "Sent refresh to: %s\n", + params.ips_array[0]); + } +} + +/* Shut down connection if there is no complete message after 10 seconds. */ +#define OG_CLIENT_TIMEOUT 10 + +/* Agent client operation might take longer, shut down after 30 seconds. */ +#define OG_AGENT_CLIENT_TIMEOUT 30 + +int socket_rest, socket_agent_rest; + +void og_server_accept_cb(struct ev_loop *loop, struct ev_io *io, int events) +{ + struct sockaddr_in client_addr; + socklen_t addrlen = sizeof(client_addr); + struct og_client *cli; + int client_sd; + + if (events & EV_ERROR) + return; + + client_sd = accept(io->fd, (struct sockaddr *)&client_addr, &addrlen); + if (client_sd < 0) { + syslog(LOG_ERR, "cannot accept client connection\n"); + return; + } + + cli = (struct og_client *)calloc(1, sizeof(struct og_client)); + if (!cli) { + close(client_sd); + return; + } + memcpy(&cli->addr, &client_addr, sizeof(client_addr)); + if (io->fd == socket_agent_rest) + cli->keepalive_idx = 0; + else + cli->keepalive_idx = -1; + + if (io->fd == socket_rest) + cli->rest = true; + else if (io->fd == socket_agent_rest) + cli->agent = true; + + syslog(LOG_DEBUG, "connection from client %s:%hu\n", + inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); + + if (io->fd == socket_agent_rest) + ev_io_init(&cli->io, og_agent_read_cb, client_sd, EV_READ); + else + ev_io_init(&cli->io, og_client_read_cb, client_sd, EV_READ); + + ev_io_start(loop, &cli->io); + if (io->fd == socket_agent_rest) { + ev_timer_init(&cli->timer, og_client_timer_cb, + OG_AGENT_CLIENT_TIMEOUT, 0.); + } else { + ev_timer_init(&cli->timer, og_client_timer_cb, + OG_CLIENT_TIMEOUT, 0.); + } + ev_timer_start(loop, &cli->timer); + og_client_add(cli); + + if (io->fd == socket_agent_rest) { + og_agent_send_refresh(cli); + } +} + +int og_socket_server_init(const char *port) +{ + struct sockaddr_in local; + int sd, on = 1; + + sd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + if (sd < 0) { + syslog(LOG_ERR, "cannot create main socket\n"); + return -1; + } + setsockopt(sd, SOL_SOCKET, SO_REUSEPORT, &on, sizeof(int)); + + local.sin_addr.s_addr = htonl(INADDR_ANY); + local.sin_family = AF_INET; + local.sin_port = htons(atoi(port)); + + if (bind(sd, (struct sockaddr *) &local, sizeof(local)) < 0) { + close(sd); + syslog(LOG_ERR, "cannot bind socket\n"); + return -1; + } + + listen(sd, 250); + + return sd; +} diff --git a/sources/core.h b/sources/core.h new file mode 100644 index 0000000..dd5cd58 --- /dev/null +++ b/sources/core.h @@ -0,0 +1,10 @@ +#ifndef _OG_CORE_H +#define _OG_CORE_H + +extern int socket_rest, socket_agent_rest; +extern struct ev_loop *og_loop; + +int og_socket_server_init(const char *port); +void og_server_accept_cb(struct ev_loop *loop, struct ev_io *io, int events); + +#endif diff --git a/sources/dbi.c b/sources/dbi.c index cc5c5b5..6640f50 100644 --- a/sources/dbi.c +++ b/sources/dbi.c @@ -1,3 +1,11 @@ +/* + * Copyright (C) 2020 Soleta Networks <info@soleta.eu> + * + * This program is free software: you can redistribute it and/or modify it under + * the terms of the GNU Affero General Public License as published by the + * Free Software Foundation, version 3. + */ + #include "dbi.h" struct og_dbi *og_dbi_open(struct og_dbi_config *config) diff --git a/sources/main.c b/sources/main.c new file mode 100644 index 0000000..31263d6 --- /dev/null +++ b/sources/main.c @@ -0,0 +1,69 @@ +/* + * Copyright (C) 2020 Soleta Networks <info@soleta.eu> + * + * This program is free software: you can redistribute it and/or modify it under + * the terms of the GNU Affero General Public License as published by the + * Free Software Foundation, version 3. + */ + +#include "ogAdmServer.h" +#include "dbi.h" +#include "utils.h" +#include "list.h" +#include "rest.h" +#include "client.h" +#include "json.h" +#include "schedule.h" +#include "core.h" +#include <syslog.h> + +int main(int argc, char *argv[]) +{ + struct ev_io ev_io_server_rest, ev_io_agent_rest; + int i; + + og_loop = ev_default_loop(0); + + if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) + exit(EXIT_FAILURE); + + openlog("ogAdmServer", LOG_PID, LOG_DAEMON); + + if (!validacionParametros(argc, argv, 1)) // Valida parámetros de ejecución + exit(EXIT_FAILURE); + + if (!tomaConfiguracion(szPathFileCfg)) { // Toma parametros de configuracion + exit(EXIT_FAILURE); + } + + for (i = 0; i < MAXIMOS_CLIENTES; i++) { + tbsockets[i].ip[0] = '\0'; + tbsockets[i].cli = NULL; + } + + socket_rest = og_socket_server_init("8888"); + if (socket_rest < 0) + exit(EXIT_FAILURE); + + ev_io_init(&ev_io_server_rest, og_server_accept_cb, socket_rest, EV_READ); + ev_io_start(og_loop, &ev_io_server_rest); + + socket_agent_rest = og_socket_server_init("8889"); + if (socket_agent_rest < 0) + exit(EXIT_FAILURE); + + ev_io_init(&ev_io_agent_rest, og_server_accept_cb, socket_agent_rest, EV_READ); + ev_io_start(og_loop, &ev_io_agent_rest); + + if (og_dbi_schedule_get() < 0) + exit(EXIT_FAILURE); + + og_schedule_next(og_loop); + + syslog(LOG_INFO, "Waiting for connections\n"); + + while (1) + ev_loop(og_loop, 0); + + exit(EXIT_SUCCESS); +} diff --git a/sources/ogAdmLib.h b/sources/ogAdmLib.h index e5bd09e..fde24eb 100644 --- a/sources/ogAdmLib.h +++ b/sources/ogAdmLib.h @@ -115,3 +115,6 @@ char* escaparCadena(char *cadena); typeof( ((type *)0)->member ) *__mptr = (ptr); \ (type *)( (char *)__mptr - offsetof(type,member) );}) +#include <stdbool.h> + +bool tomaConfiguracion(const char *filecfg); diff --git a/sources/ogAdmServer.c b/sources/ogAdmServer.c index bcec346..c9b0b70 100644 --- a/sources/ogAdmServer.c +++ b/sources/ogAdmServer.c @@ -48,7 +48,7 @@ struct og_dbi_config dbi_config = { // true: Si el proceso es correcto // false: En caso de ocurrir algún error //________________________________________________________________________________________________________ -static bool tomaConfiguracion(const char *filecfg) +bool tomaConfiguracion(const char *filecfg) { char buf[1024], *line; char *key, *value; @@ -131,12 +131,6 @@ static bool tomaConfiguracion(const char *filecfg) #define OG_CMD_MAXLEN 64 -/* Shut down connection if there is no complete message after 10 seconds. */ -#define OG_CLIENT_TIMEOUT 10 - -/* Agent client operation might take longer, shut down after 30 seconds. */ -#define OG_AGENT_CLIENT_TIMEOUT 30 - // ________________________________________________________________________________________________________ // Función: clienteDisponible // @@ -1305,469 +1299,3 @@ bool cuestionPerfilSoftware(struct og_dbi *dbi, char *idc, char *ido, return true; } - -static void og_client_release(struct ev_loop *loop, struct og_client *cli) -{ - if (cli->keepalive_idx >= 0) { - syslog(LOG_DEBUG, "closing keepalive connection for %s:%hu in slot %d\n", - inet_ntoa(cli->addr.sin_addr), - ntohs(cli->addr.sin_port), cli->keepalive_idx); - tbsockets[cli->keepalive_idx].cli = NULL; - } - - list_del(&cli->list); - ev_io_stop(loop, &cli->io); - close(cli->io.fd); - free(cli); -} - -static void og_client_keepalive(struct ev_loop *loop, struct og_client *cli) -{ - struct og_client *old_cli; - - old_cli = tbsockets[cli->keepalive_idx].cli; - if (old_cli && old_cli != cli) { - syslog(LOG_DEBUG, "closing old keepalive connection for %s:%hu\n", - inet_ntoa(old_cli->addr.sin_addr), - ntohs(old_cli->addr.sin_port)); - - og_client_release(loop, old_cli); - } - tbsockets[cli->keepalive_idx].cli = cli; -} - -static void og_client_reset_state(struct og_client *cli) -{ - cli->state = OG_CLIENT_RECEIVING_HEADER; - cli->buf_len = 0; -} - -static int og_client_payload_too_large(struct og_client *cli) -{ - char buf[] = "HTTP/1.1 413 Payload Too Large\r\n" - "Content-Length: 0\r\n\r\n"; - - send(og_client_socket(cli), buf, strlen(buf), 0); - - return -1; -} - -static int og_client_state_recv_hdr_rest(struct og_client *cli) -{ - char *ptr; - - ptr = strstr(cli->buf, "\r\n\r\n"); - if (!ptr) - return 0; - - cli->msg_len = ptr - cli->buf + 4; - - ptr = strstr(cli->buf, "Content-Length: "); - if (ptr) { - sscanf(ptr, "Content-Length: %i[^\r\n]", &cli->content_length); - if (cli->content_length < 0) - return -1; - cli->msg_len += cli->content_length; - } - - ptr = strstr(cli->buf, "Authorization: "); - if (ptr) - sscanf(ptr, "Authorization: %63[^\r\n]", cli->auth_token); - - return 1; -} - -static int og_client_recv(struct og_client *cli, int events) -{ - struct ev_io *io = &cli->io; - int ret; - - if (events & EV_ERROR) { - syslog(LOG_ERR, "unexpected error event from client %s:%hu\n", - inet_ntoa(cli->addr.sin_addr), - ntohs(cli->addr.sin_port)); - return 0; - } - - ret = recv(io->fd, cli->buf + cli->buf_len, - sizeof(cli->buf) - cli->buf_len, 0); - if (ret <= 0) { - if (ret < 0) { - syslog(LOG_ERR, "error reading from client %s:%hu (%s)\n", - inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port), - strerror(errno)); - } else { - syslog(LOG_DEBUG, "closed connection by %s:%hu\n", - inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); - } - return ret; - } - - return ret; -} - -static void og_client_read_cb(struct ev_loop *loop, struct ev_io *io, int events) -{ - struct og_client *cli; - int ret; - - cli = container_of(io, struct og_client, io); - - ret = og_client_recv(cli, events); - if (ret <= 0) - goto close; - - if (cli->keepalive_idx >= 0) - return; - - ev_timer_again(loop, &cli->timer); - - cli->buf_len += ret; - if (cli->buf_len >= sizeof(cli->buf)) { - syslog(LOG_ERR, "client request from %s:%hu is too long\n", - inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); - og_client_payload_too_large(cli); - goto close; - } - - switch (cli->state) { - case OG_CLIENT_RECEIVING_HEADER: - ret = og_client_state_recv_hdr_rest(cli); - if (ret < 0) - goto close; - if (!ret) - return; - - cli->state = OG_CLIENT_RECEIVING_PAYLOAD; - /* Fall through. */ - case OG_CLIENT_RECEIVING_PAYLOAD: - /* Still not enough data to process request. */ - if (cli->buf_len < cli->msg_len) - return; - - cli->state = OG_CLIENT_PROCESSING_REQUEST; - /* fall through. */ - case OG_CLIENT_PROCESSING_REQUEST: - ret = og_client_state_process_payload_rest(cli); - if (ret < 0) { - syslog(LOG_ERR, "Failed to process HTTP request from %s:%hu\n", - inet_ntoa(cli->addr.sin_addr), - ntohs(cli->addr.sin_port)); - } - if (ret < 0) - goto close; - - if (cli->keepalive_idx < 0) { - syslog(LOG_DEBUG, "server closing connection to %s:%hu\n", - inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); - goto close; - } else { - syslog(LOG_DEBUG, "leaving client %s:%hu in keepalive mode\n", - inet_ntoa(cli->addr.sin_addr), - ntohs(cli->addr.sin_port)); - og_client_keepalive(loop, cli); - og_client_reset_state(cli); - } - break; - default: - syslog(LOG_ERR, "unknown state, critical internal error\n"); - goto close; - } - return; -close: - ev_timer_stop(loop, &cli->timer); - og_client_release(loop, cli); -} - -enum og_agent_state { - OG_AGENT_RECEIVING_HEADER = 0, - OG_AGENT_RECEIVING_PAYLOAD, - OG_AGENT_PROCESSING_RESPONSE, -}; - -static int og_agent_state_recv_hdr_rest(struct og_client *cli) -{ - char *ptr; - - ptr = strstr(cli->buf, "\r\n\r\n"); - if (!ptr) - return 0; - - cli->msg_len = ptr - cli->buf + 4; - - ptr = strstr(cli->buf, "Content-Length: "); - if (ptr) { - sscanf(ptr, "Content-Length: %i[^\r\n]", &cli->content_length); - if (cli->content_length < 0) - return -1; - cli->msg_len += cli->content_length; - } - - return 1; -} - -static void og_agent_reset_state(struct og_client *cli) -{ - cli->state = OG_AGENT_RECEIVING_HEADER; - cli->buf_len = 0; - cli->content_length = 0; - memset(cli->buf, 0, sizeof(cli->buf)); -} - -static void og_agent_deliver_pending_cmd(struct og_client *cli) -{ - const struct og_cmd *cmd; - - cmd = og_cmd_find(inet_ntoa(cli->addr.sin_addr)); - if (!cmd) - return; - - og_send_request(cmd->method, cmd->type, &cmd->params, cmd->json); - cli->last_cmd_id = cmd->id; - - og_cmd_free(cmd); -} - -static void og_agent_read_cb(struct ev_loop *loop, struct ev_io *io, int events) -{ - struct og_client *cli; - int ret; - - cli = container_of(io, struct og_client, io); - - ret = og_client_recv(cli, events); - if (ret <= 0) - goto close; - - ev_timer_again(loop, &cli->timer); - - cli->buf_len += ret; - if (cli->buf_len >= sizeof(cli->buf)) { - syslog(LOG_ERR, "client request from %s:%hu is too long\n", - inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); - goto close; - } - - switch (cli->state) { - case OG_AGENT_RECEIVING_HEADER: - ret = og_agent_state_recv_hdr_rest(cli); - if (ret < 0) - goto close; - if (!ret) - return; - - cli->state = OG_AGENT_RECEIVING_PAYLOAD; - /* Fall through. */ - case OG_AGENT_RECEIVING_PAYLOAD: - /* Still not enough data to process request. */ - if (cli->buf_len < cli->msg_len) - return; - - cli->state = OG_AGENT_PROCESSING_RESPONSE; - /* fall through. */ - case OG_AGENT_PROCESSING_RESPONSE: - ret = og_agent_state_process_response(cli); - if (ret < 0) { - syslog(LOG_ERR, "Failed to process HTTP request from %s:%hu\n", - inet_ntoa(cli->addr.sin_addr), - ntohs(cli->addr.sin_port)); - goto close; - } else if (ret == 0) { - og_agent_deliver_pending_cmd(cli); - } - - syslog(LOG_DEBUG, "leaving client %s:%hu in keepalive mode\n", - inet_ntoa(cli->addr.sin_addr), - ntohs(cli->addr.sin_port)); - og_agent_reset_state(cli); - break; - default: - syslog(LOG_ERR, "unknown state, critical internal error\n"); - goto close; - } - return; -close: - ev_timer_stop(loop, &cli->timer); - og_client_release(loop, cli); -} - -static void og_client_timer_cb(struct ev_loop *loop, ev_timer *timer, int events) -{ - struct og_client *cli; - - cli = container_of(timer, struct og_client, timer); - if (cli->keepalive_idx >= 0) { - ev_timer_again(loop, &cli->timer); - return; - } - syslog(LOG_ERR, "timeout request for client %s:%hu\n", - inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); - - og_client_release(loop, cli); -} - -static void og_agent_send_refresh(struct og_client *cli) -{ - struct og_msg_params params; - int err; - - params.ips_array[0] = inet_ntoa(cli->addr.sin_addr); - params.ips_array_len = 1; - - err = og_send_request(OG_METHOD_GET, OG_CMD_REFRESH, ¶ms, NULL); - if (err < 0) { - syslog(LOG_ERR, "Can't send refresh to: %s\n", - params.ips_array[0]); - } else { - syslog(LOG_INFO, "Sent refresh to: %s\n", - params.ips_array[0]); - } -} - -static int socket_rest, socket_agent_rest; - -static void og_server_accept_cb(struct ev_loop *loop, struct ev_io *io, - int events) -{ - struct sockaddr_in client_addr; - socklen_t addrlen = sizeof(client_addr); - struct og_client *cli; - int client_sd; - - if (events & EV_ERROR) - return; - - client_sd = accept(io->fd, (struct sockaddr *)&client_addr, &addrlen); - if (client_sd < 0) { - syslog(LOG_ERR, "cannot accept client connection\n"); - return; - } - - cli = (struct og_client *)calloc(1, sizeof(struct og_client)); - if (!cli) { - close(client_sd); - return; - } - memcpy(&cli->addr, &client_addr, sizeof(client_addr)); - if (io->fd == socket_agent_rest) - cli->keepalive_idx = 0; - else - cli->keepalive_idx = -1; - - if (io->fd == socket_rest) - cli->rest = true; - else if (io->fd == socket_agent_rest) - cli->agent = true; - - syslog(LOG_DEBUG, "connection from client %s:%hu\n", - inet_ntoa(cli->addr.sin_addr), ntohs(cli->addr.sin_port)); - - if (io->fd == socket_agent_rest) - ev_io_init(&cli->io, og_agent_read_cb, client_sd, EV_READ); - else - ev_io_init(&cli->io, og_client_read_cb, client_sd, EV_READ); - - ev_io_start(loop, &cli->io); - if (io->fd == socket_agent_rest) { - ev_timer_init(&cli->timer, og_client_timer_cb, - OG_AGENT_CLIENT_TIMEOUT, 0.); - } else { - ev_timer_init(&cli->timer, og_client_timer_cb, - OG_CLIENT_TIMEOUT, 0.); - } - ev_timer_start(loop, &cli->timer); - og_client_add(cli); - - if (io->fd == socket_agent_rest) { - og_agent_send_refresh(cli); - } -} - -static int og_socket_server_init(const char *port) -{ - struct sockaddr_in local; - int sd, on = 1; - - sd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); - if (sd < 0) { - syslog(LOG_ERR, "cannot create main socket\n"); - return -1; - } - setsockopt(sd, SOL_SOCKET, SO_REUSEPORT, &on, sizeof(int)); - - local.sin_addr.s_addr = htonl(INADDR_ANY); - local.sin_family = AF_INET; - local.sin_port = htons(atoi(port)); - - if (bind(sd, (struct sockaddr *) &local, sizeof(local)) < 0) { - close(sd); - syslog(LOG_ERR, "cannot bind socket\n"); - return -1; - } - - listen(sd, 250); - - return sd; -} - -struct ev_loop *og_loop; - -int main(int argc, char *argv[]) -{ - struct ev_io ev_io_server_rest, ev_io_agent_rest; - int i; - - og_loop = ev_default_loop(0); - - if (signal(SIGPIPE, SIG_IGN) == SIG_ERR) - exit(EXIT_FAILURE); - - openlog("ogAdmServer", LOG_PID, LOG_DAEMON); - - /*-------------------------------------------------------------------------------------------------------- - Validación de parámetros de ejecución y lectura del fichero de configuración del servicio - ---------------------------------------------------------------------------------------------------------*/ - if (!validacionParametros(argc, argv, 1)) // Valida parámetros de ejecución - exit(EXIT_FAILURE); - - if (!tomaConfiguracion(szPathFileCfg)) { // Toma parametros de configuracion - exit(EXIT_FAILURE); - } - - /*-------------------------------------------------------------------------------------------------------- - // Inicializa array de información de los clientes - ---------------------------------------------------------------------------------------------------------*/ - for (i = 0; i < MAXIMOS_CLIENTES; i++) { - tbsockets[i].ip[0] = '\0'; - tbsockets[i].cli = NULL; - } - /*-------------------------------------------------------------------------------------------------------- - Creación y configuración del socket del servicio - ---------------------------------------------------------------------------------------------------------*/ - - socket_rest = og_socket_server_init("8888"); - if (socket_rest < 0) - exit(EXIT_FAILURE); - - ev_io_init(&ev_io_server_rest, og_server_accept_cb, socket_rest, EV_READ); - ev_io_start(og_loop, &ev_io_server_rest); - - socket_agent_rest = og_socket_server_init("8889"); - if (socket_agent_rest < 0) - exit(EXIT_FAILURE); - - ev_io_init(&ev_io_agent_rest, og_server_accept_cb, socket_agent_rest, EV_READ); - ev_io_start(og_loop, &ev_io_agent_rest); - - if (og_dbi_schedule_get() < 0) - exit(EXIT_FAILURE); - - og_schedule_next(og_loop); - - syslog(LOG_INFO, "Waiting for connections\n"); - - while (1) - ev_loop(og_loop, 0); - - exit(EXIT_SUCCESS); -} diff --git a/sources/rest.c b/sources/rest.c index e9ee00f..35f3e03 100644 --- a/sources/rest.c +++ b/sources/rest.c @@ -22,6 +22,8 @@ #include <jansson.h> #include <time.h> +struct ev_loop *og_loop; + static TRAMA *og_msg_alloc(char *data, unsigned int len) { TRAMA *ptrTrama; diff --git a/sources/utils.c b/sources/utils.c index 4aa76f8..433a0dc 100644 --- a/sources/utils.c +++ b/sources/utils.c @@ -1,3 +1,11 @@ +/* + * Copyright (C) 2020 Soleta Networks <info@soleta.eu> + * + * This program is free software: you can redistribute it and/or modify it under + * the terms of the GNU Affero General Public License as published by the + * Free Software Foundation, version 3. + */ + #include <ctype.h> #include "utils.h" |