#include #include #include #include #include #include #include #include #include #include #include #include #include #include /* arbitrary */ #define MAX_CMD_LEN 1024 enum que_op { LARK_KILL, LARK_NEW, LARK_NEXT, LARK_POLL, }; #define VEC_NAME strbuf #define VEC_TYPE char #include /* doesn't use any state */ struct meta_parse_cmd {}; enum sync_state { SYNC_HEADER = 0, SYNC_SEND }; struct sync_cmd { enum sync_state state; long start; long end; char *thread; }; enum server_cmd { META_PARSE_CMD, SYNC_CMD, }; struct server_conn { int fdin; int fdout; struct sockaddr addr; socklen_t addrlen; struct strbuf strbuf; size_t nextlen; enum server_cmd cmd; union { struct meta_parse_cmd meta_parse; struct sync_cmd sync; } u; }; #define VEC_NAME server_conns #define VEC_TYPE struct server_conn #include #define VEC_NAME conn_polls #define VEC_TYPE struct pollfd #include struct server_ctx { int sock; bool oneshot; struct server_conns conns; struct conn_polls polls; int timeout; }; /* buf is sized by count. < 0 for error, 0 for no data available, > 0 for how * much data was read */ static ssize_t readx(int fd, void *buf, size_t count) { while (1) { ssize_t r = read(fd, buf, count); if (r == -1) { if (errno == EINTR) /* read interrupted, try again */ continue; if (errno == EWOULDBLOCK) /* no more data to read for now */ return 0; } /* we've read as many bytes as are currently available */ return r == 0 ? -1 : r; } /* should never be reached, but keep here to avoid warnings */ return 0; } static ssize_t writex(int fd, const void *buf, size_t count) { while (1) { ssize_t r = write(fd, buf, count); if (r == -1) { if (errno == EINTR) /* write interrupted, try again */ continue; } return r; } /* should never be reached, but keep here to avoid warnings */ return 0; } static bool strneq(const char *s1, const char *s2, size_t n) { return strncmp(s1, s2, n) == 0; } static bool streq(const char *s1, const char *s2) { return strcmp(s1, s2) == 0; } static int new_connection(struct server_ctx *ctx, int new_conn_fd) { assert(!ctx->oneshot); struct sockaddr addr; socklen_t addrlen; int ret = 0; /** @todo retry count? */ while (1) { ret = accept(new_conn_fd, &addr, &addrlen); if (ret == -1) { /* try again */ if (errno == EINTR) continue; /* some other error, throw it at the user */ return ret; } /* we have a valid fd */ break; } int fd = ret; ret = fcntl(fd, F_SETFL, O_NONBLOCK); if (ret == -1) { close(fd); return ret; } struct server_conn conn = { .fdin = fd, .fdout = fd, .addr = addr, .addrlen = addrlen, .cmd = META_PARSE_CMD, .strbuf = strbuf_create(0), .nextlen = 0 }; memset(&conn.u, 0, sizeof(conn.u)); server_conns_append(&ctx->conns, conn); return 0; } /* I'll assume that each individual subcommant has ensured all its resources are * freed by now. Invalidates the conn pointer! */ static int kill_connection(struct server_ctx *ctx, struct server_conn *conn) { close(conn->fdin); if (conn->fdout != conn->fdin) close(conn->fdout); strbuf_destroy(&conn->strbuf); assert(server_conns_len(&ctx->conns) >= 1); if (server_conns_len(&ctx->conns) == 1) { server_conns_shrink(&ctx->conns, 0); return 0; } /* overwrite our location in the vector with whaterver is at the back of * the vector */ *conn = *server_conns_pop(&ctx->conns); return 0; } /* re-entrant read, returns < 0 on error, > 0 for how * large the buffer is currently */ static ssize_t reread(struct server_conn *conn, size_t c) { assert(c != 0); struct strbuf *strbuf = &conn->strbuf; size_t n = strbuf->n; strbuf_reserve(strbuf, n + c); ssize_t r = readx(conn->fdin, strbuf->buf + n, c); if (r < 0) return r; strbuf_shrink(strbuf, n + r); return n + r; } /* read a full command, using at most max bytes. * < 0 on error (including no command found in the allotted byte count), * 0 if no command found yet, * > 0 when command is available, indicating command length (including newline) * * Note that this does not respect timeouts, and each command that uses this * should keep track of its own timing. */ static ssize_t reread_cmd(struct server_conn *conn, size_t max) { struct strbuf *strbuf = &conn->strbuf; /* previous occupancy */ size_t n = strbuf->n; ssize_t r = reread(conn, max); if (r < 0) return r; if (r == (ssize_t)n) return 0; /* parse the new bytes and see if we have an end-of-command marker */ char *buf = strbuf->buf; for (ssize_t i = n; i < r; ++i) { if (i > (ssize_t)max) return -1; if (buf[i] == '\n') return i + 1; } return 0; } /* read up to n bytes. * < 0 on error * 0 if not enough bytes yet * > 0 to indicate how many bytes are available (should always be n) */ static ssize_t reread_n(struct server_conn *conn, size_t n) { if (conn->strbuf.n >= n) return n; ssize_t r = reread(conn, n - conn->strbuf.n); if (r < 0) return r; if (r < (ssize_t)n) return 0; return r; } /* set timeout to new smallest value, taking infinite timeout (-1) into account. * */ static void set_timeout(struct server_ctx *ctx, int timeout) { if (ctx->timeout == -1) { ctx->timeout = timeout; return; } /* ctx->timeout != -1, meaning there's a timeout set and it's not * infinite, so prefer the existing value */ if (timeout == -1) return; if (timeout < ctx->timeout) ctx->timeout = timeout; } /* return 0 on success, !0 on failure */ static int parse_cmd(char *buf, char **cmd, size_t *len) { char *keystart = buf; while (*keystart && isblank(*keystart)) keystart++; /* empty string */ if (!*keystart) return -1; char *keyend = keystart + 1; while (*keyend && isalnum(*keyend)) keyend++; /* nothing after key */ if (!*keyend) return -1; /* split off cmd from rest */ *keyend = '\0'; if (cmd) *cmd = keystart; if (!len) return 0; char *lenstart = keyend + 1; while (*lenstart && isblank(*lenstart)) lenstart++; /* space after key but no len value */ if (!*lenstart) return -1; char *lenend = lenstart + 1; while (*lenend && isdigit(*lenend)) lenend++; if (!*lenend) { /* end of string, nothing after digits but there doesn't need to be */ char *check = NULL; *len = strtoull(lenstart, &check, 10); /* sanity check of sorts, if they're equal we return 0 (which we * want), otherwise -1 */ return check == lenend ? 0 : -1; } /* something following length, trailing spaces are allowed but nothing * else */ char *linend = lenend + 1; while (*linend && isblank(*linend)) linend++; /* there was other stuff besides spaces, error */ if (*linend) return -1; *lenend = '\0'; char *check = NULL; *len = strtoull(lenstart, &check, 10); return check == lenend ? 0 : -1; } static void strbuf_shift(struct strbuf *strbuf, size_t i) { assert(i <= strbuf->n); memmove(strbuf->buf, strbuf->buf + i, strbuf->n - i); strbuf_shrink(strbuf, strbuf->n - i); } static enum que_op meta_parse_cmd(struct server_conn *conn) { int r = reread_cmd(conn, MAX_CMD_LEN); if (r < 0) { syslog(LOG_DEBUG, "error while reading cmd, " "client disconnected or sent garbage data\n"); return LARK_KILL; } if (r == 0) { syslog(LOG_DEBUG, "not enough data in socket to parse cmd with\n"); return LARK_POLL; } /* r > 0, we (should) have a command and its nextlen */ char *cmd = NULL; conn->strbuf.buf[r - 1] = '\0'; if (parse_cmd(conn->strbuf.buf, &cmd, &conn->nextlen)) { syslog(LOG_DEBUG, "invalid command %s, killing connection\n", conn->strbuf.buf); return LARK_KILL; } if (strneq(conn->strbuf.buf, "SYNC", 4)) { syslog(LOG_DEBUG, "parsed SYNC command start\n"); conn->cmd = SYNC_CMD; strbuf_shift(&conn->strbuf, r); return LARK_NEW; } syslog(LOG_DEBUG, "client is rarted\n"); return LARK_KILL; } /* could also take the length of the buf? */ typedef int (*kv_cb_t)(char *key, char *value, void *data); static int parse_kv(char *buf, int (*kv_cb)(char *key, char *value, void *data), void *data) { if (!buf) return 0; char *start = buf; while (1) { char *keystart = start; /* skip leading whitespace */ while (*keystart && isblank(*keystart)) keystart++; /* end of string, nothing more to read */ if (!*keystart) return 0; /* end of blank line, skip to next */ if (*keystart == '\n') { start++; continue; } /* start of empty line just a comment, skip to next line if one exists */ if (*keystart == '#') { while (*keystart && *keystart != '\n') keystart++; /* end of string */ if (!*keystart) return 0; /* skip line-ending '\n' */ start = keystart + 1; continue; } /* start now points at the first character of a key. Next, parse * key forward a bit to check that all characters are valid */ char *keyend = keystart; while (*keyend && (isalnum(*keyend) || *keyend == '-' || *keyend == '_')) keyend++; /* key ends a line, illegal */ if (!*keyend) return -1; /* key contains illegal characters, as we didn't end up on empty * space or the '=' key */ if (!(isblank(*keyend) || *keyend == '=')) return -1; /* next, find out where value starts */ char *valstart = keyend; /* skip forward until '=' sign */ while (*valstart && *valstart != '=') valstart++; /* plain key on a line, illegal */ if (!*valstart) return -1; /* skip over '=' sign */ valstart++; /* skip forward until actual start of value, when blanks stop */ while (*valstart && isblank(*valstart)) valstart++; /* find end of value. Note that empty values are allowed, * whether they're empty by end of string or end of line or the * start of a comment*/ char *valend = valstart; char *runner = valend; while (*runner && *runner != '#' && *runner != '\n') { /* keep track of where last non-whitespace character is * since it tells us where the value ends */ if (!isblank(*runner)) valend = runner + 1; /* skip escape characters */ if (*runner == '\\') runner++; /* but don't skip end of line of end of string */ if (!*runner || *runner == '\n') break; runner++; } /* prepare start of next line */ while (*runner && *runner != '\n') runner++; if (*runner == '\n') runner++; start = runner; /* present parsed values to user */ *keyend = '\0'; *valend = '\0'; /* unescape backslashes */ long shift = 0; for (ptrdiff_t i = 0; i < valend - valstart + 1; ++i) { if (valstart[i] == '\\') { shift++; continue; } if (shift == 0) continue; valstart[i - shift] = valstart[i]; } int r = kv_cb(keystart, valstart, data); if (r) return r; } return 0; } static void destroy_sync(struct sync_cmd *cmd) { free(cmd->thread); } static int sync_kv(char *key, char *value, struct sync_cmd *cmd) { if (streq(key, "START")) { char *end = NULL; cmd->start = strtol(value, &end, 0); if (end == value || *end != '\0') { syslog(LOG_DEBUG, "not an integer value: %s = %s", key, value); return -1; } if (cmd->start < 0) { syslog(LOG_DEBUG, "illegal integer value: %ld", cmd->start); return -1; } return 0; } if (streq(key, "END")) { char *end = NULL; cmd->end = strtol(value, &end, 0); if (end == value || *end != '\0') { syslog(LOG_DEBUG, "not an integer value: %s = %s", key, value); return -1; } if (cmd->end < 0 && cmd->end != -1) { syslog(LOG_DEBUG, "illegal integer value: %ld", cmd->end); return -1; } return 0; } if (streq(key, "THREAD")) { /** @todo check validness, copy default value, etc. */ cmd->thread = strdup(value); return 0; } syslog(LOG_DEBUG, "unknown parameter to sync: %s", key); return -1; } static enum que_op sync_cmd(struct server_ctx *ctx, struct server_conn *conn) { struct sync_cmd *cmd = &conn->u.sync; switch (cmd->state) { case SYNC_HEADER: { ssize_t r = reread_n(conn, conn->nextlen); if (r < 0) { syslog(LOG_DEBUG, "client messed up connection"); return LARK_KILL; } /** @todo respect timeout of some kind? */ if (r == 0) return LARK_POLL; /* headr must include trailing newline, so we can be sure that a * valid header will remain valid if we null-terminate the * header */ conn->strbuf.buf[r - 1] = '\0'; /* parse ini-style header */ /* some default values, arguably not sensible to always sync a * whole thread but eh */ cmd->start = 0; cmd->end = -1; cmd->thread = NULL; if (parse_kv(conn->strbuf.buf, (kv_cb_t)sync_kv, cmd)) { syslog(LOG_DEBUG, "failed parsing SYNC header"); destroy_sync(cmd); return LARK_KILL; } strbuf_shift(&conn->strbuf, r); syslog(LOG_DEBUG, "got valid sync header\n"); cmd->state = SYNC_SEND; return sync_cmd(ctx, conn); } case SYNC_SEND: { syslog(LOG_DEBUG, "sending sync\n"); const char buf[] = "From: urmum@amazon.com"; ssize_t ret = writex(conn->fdout, buf, sizeof(buf)); assert(ret == sizeof(buf)); conn->cmd = META_PARSE_CMD; destroy_sync(cmd); return LARK_NEW; } } syslog(LOG_DEBUG, "sync state broken\n"); return LARK_KILL; } static int event_loop(int sock, bool oneshot) { struct server_ctx ctx = { .sock = sock, .oneshot = oneshot, .conns = server_conns_create(0), .polls = conn_polls_create(0), .timeout = -1 }; /* set up a connection */ if (oneshot) { struct server_conn conn = { .fdin = sock, .fdout = sock == STDIN_FILENO ? STDOUT_FILENO : sock, .addr = {}, .addrlen = 0, .cmd = META_PARSE_CMD, .strbuf = strbuf_create(0), .nextlen = 0 }; memset(&conn.u, 0, sizeof(conn.u)); server_conns_append(&ctx.conns, conn); } int ret = 0; while (1) { if (ctx.oneshot && server_conns_len(&ctx.conns) == 0) { ret = 0; break; } /* build up set of connections to poll */ conn_polls_reset(&ctx.polls); /* no need to poll if we're in oneshot mode, but keep pollfd * around to simplify index calculations */ struct pollfd tcp_poll = { .fd = ctx.oneshot ? -1 : ctx.sock, .events = POLLIN, .revents = 0 }; conn_polls_append(&ctx.polls, tcp_poll); foreach(server_conns, c, &ctx.conns) { struct pollfd conn_poll = { .fd = c->fdin, .events = POLLIN, .revents = 0 }; conn_polls_append(&ctx.polls, conn_poll); } /* wait for something to happen on incoming connections, or a * new connection. Eventually might also want to wait on * outgoing connections to become available for writing? */ ret = poll(ctx.polls.buf, conn_polls_len(&ctx.polls), ctx.timeout); /* reset timeout value for next iteration */ ctx.timeout = -1; if (ret == 0) { /* timeout */ if (server_conns_len(&ctx.conns) == 0) { /* nothing to do, quit for now */ break; } /* some command just timed out, handle it by fallthrough */ } if (ret == -1) { /* some kind of error */ if (errno == EINTR) { /* safe to try again */ continue; } /* exit loop due to unknown error*/ err(EXIT_FAILURE, "poll failed"); } /* otherwise, there's some event we should react to */ /* is it a new connection? */ struct pollfd *new_conn = conn_polls_at(&ctx.polls, 0); if (new_conn->revents != 0) { if (new_connection(&ctx, new_conn->fd)) err(EXIT_FAILURE, "tcp socket exploded"); new_conn->revents = 0; } for (size_t conn_idx = 0; conn_idx < server_conns_len(&ctx.conns); ++conn_idx) { struct server_conn *conn = server_conns_at(&ctx.conns, conn_idx); enum que_op op = LARK_KILL; switch (conn->cmd) { case META_PARSE_CMD: op = meta_parse_cmd(conn); break; case SYNC_CMD: op = sync_cmd(&ctx, conn); break; default: abort(); } switch (op) { case LARK_KILL: kill_connection(&ctx, conn); /* kill_connection placed last conn at this idx, * retry it */ conn_idx--; break; case LARK_NEXT: set_timeout(&ctx, 0); break; case LARK_NEW: set_timeout(&ctx, 0); /* reset union for next command */ memset(&conn->u, 0, sizeof(conn->u)); break; case LARK_POLL: /* nothing to do, timeout request should've been * set if it matters */ break; } } } conn_polls_destroy(&ctx.polls); server_conns_destroy(&ctx.conns); return ret; } int main(int argc, char *argv[argc]) { int ret = 0, opt; const char *single_conn = NULL; while ((opt = getopt(argc, argv, "s:")) != -1) { switch (opt) { case 's': single_conn = optarg; break; default: fprintf(stderr, "usage: %s [-s socket or - for stdin]\n", argv[0]); exit(EXIT_FAILURE); } } openlog(NULL, LOG_PERROR, LOG_USER); int sock = -1; if (single_conn) { if (streq(single_conn, "-")) sock = STDIN_FILENO; else sock = open(single_conn, O_RDONLY); if (sock == -1) err(EXIT_FAILURE, "failed opening single connection"); } else { int sock = socket(AF_INET, SOCK_STREAM, 0); if (sock == -1) err(EXIT_FAILURE, "failed opening tcp socket"); struct sockaddr_in addr; memset(&addr, 0, sizeof(addr)); addr.sin_family = AF_INET; addr.sin_addr.s_addr = htonl(INADDR_ANY); /* should probably come up with a reasonable port at some point */ addr.sin_port = htons(24242); if (bind(sock, (struct sockaddr *)&addr, sizeof(addr)) == -1) err(EXIT_FAILURE, "failed binding tcp socket"); if (listen(sock, 64) == -1) err(EXIT_FAILURE, "failed listening on tcp socket"); } ret = event_loop(sock, single_conn != NULL); close(sock); return ret; }