diff options
Diffstat (limited to 'src/rpc')
| -rw-r--r-- | src/rpc/PROTOCOL | 2 | ||||
| -rw-r--r-- | src/rpc/deps-exe/skabus-rpc-client | 1 | ||||
| -rw-r--r-- | src/rpc/deps-exe/skabus-rpcc | 4 | ||||
| -rw-r--r-- | src/rpc/deps-exe/skabus-rpccctl | 4 | ||||
| -rw-r--r-- | src/rpc/deps-exe/skabus-rpcd | 2 | ||||
| -rw-r--r-- | src/rpc/skabus-rpc-client.c | 120 | ||||
| -rw-r--r-- | src/rpc/skabus-rpcc.c | 619 | ||||
| -rw-r--r-- | src/rpc/skabus-rpcc.h | 18 | ||||
| -rw-r--r-- | src/rpc/skabus-rpccctl.c | 148 | ||||
| -rw-r--r-- | src/rpc/skabus-rpccctl.h | 36 | ||||
| -rw-r--r-- | src/rpc/skabus-rpcd.c | 71 | ||||
| -rw-r--r-- | src/rpc/skabus_rpccctl.c | 121 |
12 files changed, 1108 insertions, 38 deletions
diff --git a/src/rpc/PROTOCOL b/src/rpc/PROTOCOL index 216c9bd..bd9ee64 100644 --- a/src/rpc/PROTOCOL +++ b/src/rpc/PROTOCOL @@ -1,3 +1,4 @@ + Protocol between a client - using skabus_rpc_*() - and skabus-rpcd. Registering (1st connection) (client -> server) @@ -79,4 +80,3 @@ and contains an idstr instead of an ifname. iflen : 1 ifname : iflen '\0' : 1 - diff --git a/src/rpc/deps-exe/skabus-rpc-client b/src/rpc/deps-exe/skabus-rpc-client new file mode 100644 index 0000000..e7187fe --- /dev/null +++ b/src/rpc/deps-exe/skabus-rpc-client @@ -0,0 +1 @@ +-lskarnet diff --git a/src/rpc/deps-exe/skabus-rpcc b/src/rpc/deps-exe/skabus-rpcc new file mode 100644 index 0000000..615a1db --- /dev/null +++ b/src/rpc/deps-exe/skabus-rpcc @@ -0,0 +1,4 @@ +${LIBSKABUS} +-lskarnet +${SOCKET_LIB} +${TAINNOW_LIB} diff --git a/src/rpc/deps-exe/skabus-rpccctl b/src/rpc/deps-exe/skabus-rpccctl new file mode 100644 index 0000000..0f82e05 --- /dev/null +++ b/src/rpc/deps-exe/skabus-rpccctl @@ -0,0 +1,4 @@ +skabus_rpccctl.o +-lskarnet +${SOCKET_LIB} +${TAINNOW_LIB} diff --git a/src/rpc/deps-exe/skabus-rpcd b/src/rpc/deps-exe/skabus-rpcd index a4a573f..e8bcb88 100644 --- a/src/rpc/deps-exe/skabus-rpcd +++ b/src/rpc/deps-exe/skabus-rpcd @@ -4,5 +4,5 @@ skabus_rpcd_query.o libskabus.a.xyzzy -ls6 -lskarnet -${TAINNOW_LIB} ${SOCKET_LIB} +${TAINNOW_LIB} diff --git a/src/rpc/skabus-rpc-client.c b/src/rpc/skabus-rpc-client.c new file mode 100644 index 0000000..9cf9efe --- /dev/null +++ b/src/rpc/skabus-rpc-client.c @@ -0,0 +1,120 @@ +/* ISC license. */ + +#include <skalibs/types.h> +#include <skalibs/sgetopt.h> +#include <skalibs/strerr2.h> +#include <skalibs/djbunix.h> +#include <s6/config.h> +#include <skabus/config.h> +#include "skabus-rpcc.h" + +#define USAGE "skabus-rpc-client [ -v verbosity ] [ -d | -D ] [ -1 ] [ -c maxconn ] [ -b backlog ] [ -t timeout ] [ -T lameducktimeout ] [ -C pmprog:sep:flags ] [ -y ifname:ifprog:sep:flags ... ] rpccpath rpcdpath clientname" +#define dieusage() strerr_dieusage(100, USAGE) + +int main (int argc, char const *const *argv, char const *const *envp) +{ + unsigned int verbosity = 1 ; + int flag1 = 0 ; + int flagreuse = 1 ; + unsigned int maxconn = 0 ; + unsigned int backlog = (unsigned int)-1 ; + unsigned int timeout = 0 ; + unsigned int ltimeout = 0 ; + PROG = "skabus-rpc-client" ; + unsigned int ifn = 1 ; + char const *interfaces[SKABUS_RPCC_INTERFACES_MAX] = { 0 } ; + { + subgetopt_t l = SUBGETOPT_ZERO ; + for (;;) + { + register int opt = subgetopt_r(argc, argv, "Dd1v:c:b:t:T:C:y:", &l) ; + if (opt == -1) break ; + switch (opt) + { + case 'D' : flagreuse = 0 ; break ; + case 'd' : flagreuse = 1 ; break ; + case '1' : flag1 = 1 ; break ; + case 'v' : if (!uint0_scan(l.arg, &verbosity)) dieusage() ; break ; + case 'c' : if (!uint0_scan(l.arg, &maxconn)) dieusage() ; if (!maxconn) maxconn = 1 ; break ; + case 'b' : if (!uint0_scan(l.arg, &backlog)) dieusage() ; break ; + case 't' : if (!uint0_scan(l.arg, &timeout)) dieusage() ; break ; + case 'T' : if (!uint0_scan(l.arg, <imeout)) dieusage() ; break ; + case 'C' : interfaces[0] = l.arg ; break ; + case 'y' : + { + if (ifn >= SKABUS_RPCC_INTERFACES_MAX) dietoomanyinterfaces() ; + if (!l.arg[0]) dieemptyifname() ; + interfaces[ifn++] = l.arg ; + break ; + } + default : dieusage() ; + } + } + argc -= l.ind ; argv += l.ind ; + if (argc < 3) dieusage() ; + } + + { + unsigned int m = 0, pos = 0 ; + char fmt[UINT_FMT * 5] ; + char const *newargv[22 + 2 * ifn] ; + newargv[m++] = S6_EXTBINPREFIX "s6-ipcserver-socketbinder" ; + if (!flagreuse) newargv[m++] = "-D" ; + newargv[m++] = "-a" ; + newargv[m++] = "0700" ; + if (backlog != (unsigned int)-1) + { + newargv[m++] = "-b" ; + newargv[m++] = fmt + pos ; + pos += uint_fmt(fmt + pos, backlog) ; + fmt[pos++] = 0 ; + } + newargv[m++] = "--" ; + newargv[m++] = *argv++ ; + newargv[m++] = SKABUS_BINPREFIX "skabus-rpcc" ; + if (verbosity != 1) + { + newargv[m++] = "-v" ; + newargv[m++] = fmt + pos ; + pos += uint_fmt(fmt + pos, verbosity) ; + fmt[pos++] = 0 ; + } + if (flag1) newargv[m++] = "-1" ; + if (maxconn) + { + newargv[m++] = "-c" ; + newargv[m++] = fmt + pos ; + pos += uint_fmt(fmt + pos, maxconn) ; + fmt[pos++] = 0 ; + } + if (timeout) + { + newargv[m++] = "-t" ; + newargv[m++] = fmt + pos ; + pos += uint_fmt(fmt + pos, timeout) ; + fmt[pos++] = 0 ; + } + if (ltimeout) + { + newargv[m++] = "-T" ; + newargv[m++] = fmt + pos ; + pos += uint_fmt(fmt + pos, timeout) ; + fmt[pos++] = 0 ; + } + if (interfaces[0]) + { + newargv[m++] = "-X" ; + newargv[m++] = interfaces[0] ; + } + for (unsigned int i = 1 ; i < ifn ; i++) + { + newargv[m++] = "-y" ; + newargv[m++] = interfaces[i] ; + } + newargv[m++] = "--" ; + newargv[m++] = *argv++ ; + newargv[m++] = *argv++ ; + newargv[m++] = 0 ; + xpathexec_run(newargv[0], newargv, envp) ; + } +} diff --git a/src/rpc/skabus-rpcc.c b/src/rpc/skabus-rpcc.c new file mode 100644 index 0000000..dec5501 --- /dev/null +++ b/src/rpc/skabus-rpcc.c @@ -0,0 +1,619 @@ +/* ISC license. */ + +#include <sys/types.h> +#include <sys/stat.h> +#include <unistd.h> +#include <stdint.h> +#include <fcntl.h> +#include <limits.h> +#include <regex.h> +#include <errno.h> +#include <signal.h> +#include <skalibs/posixplz.h> +#include <skalibs/uint32.h> +#include <skalibs/uint64.h> +#include <skalibs/types.h> +#include <skalibs/bytestr.h> +#include <skalibs/allreadwrite.h> +#include <skalibs/error.h> +#include <skalibs/cdb.h> +#include <skalibs/strerr2.h> +#include <skalibs/selfpipe.h> +#include <skalibs/sig.h> +#include <skalibs/stralloc.h> +#include <skalibs/bufalloc.h> +#include <skalibs/djbunix.h> +#include <skalibs/sgetopt.h> +#include <skalibs/tai.h> +#include <skalibs/iopause.h> +#include <skalibs/env.h> +#include <skalibs/webipc.h> +#include <skalibs/gensetdyn.h> +#include <skalibs/genqdyn.h> +#include <skalibs/skamisc.h> +#include <execline/config.h> +#include <skabus/rpc.h> +#include "skabus-rpcc.h" + +#define USAGE "skabus-rpcc [ -v verbosity ] [ -1 ] [ -c maxconn ] [ -t timeout ] [ -T lameducktimeout ] [ -C pmprog:sep:flags ] [ -y ifname:ifprog:sep:flags ... ] rpcdpath clientname" +#define dieusage() strerr_dieusage(100, USAGE) ; + +static tain_t answertto = TAIN_INFINITE_RELATIVE ; + +static unsigned int verbosity = 1 ; +static int cont = 1 ; +static tain_t lameduckdeadline = TAIN_INFINITE_RELATIVE ; +static skabus_rpc_t a = SKABUS_RPC_ZERO ; + +static void quit (void) +{ + if (cont) + { + cont = 0 ; + tain_add_g(&lameduckdeadline, &lameduckdeadline) ; + } +} + + + /* Interfaces */ + +typedef struct interface_s interface_t, *interface_t_ref ; +struct interface_s +{ + uint32_t next ; + unsigned int xindex[2] ; + stralloc storage ; + pid_t pid ; + uint32_t id ; + buffer in ; + stralloc insa ; + bufalloc out ; + tain_t tto ; + tain_t deadline ; + genqdyn serialq ; + char sep[2] ; + char buf[SKABUS_RPCC_BUFSIZE] ; +} ; + +static gensetdyn interfaces = GENSETDYN_INIT(interface_t, 1, 0, 1) ; +#define numinterfaces (gensetdyn_n(&interfaces) - 1) +#define INTERFACE(i) GENSETDYN_P(interface_t, &interfaces, (i)) +static uint32_t interface_sentinel ; + +static inline void interface_free (interface_t *y) +{ + fd_close(bufalloc_fd(&y->out)) ; + fd_close(buffer_fd(&y->in)) ; + genqdyn_free(&y->serialq) ; + bufalloc_free(&y->out) ; + stralloc_free(&y->insa) ; + stralloc_free(&y->storage) ; + kill(y->pid, SIGTERM) ; +} + +static inline void interface_delete (uint32_t i, uint32_t prev) +{ + interface_t *y = INTERFACE(i) ; + INTERFACE(prev)->next = y->next ; + interface_free(y) ; + gensetdyn_delete(&interfaces, i) ; +} + +static int rclient_function (skabus_rpc_t *a, skabus_rpc_rinfo_t const *info, unixmessage_t const *m, void *aux) +{ + tain_t deadline ; + interface_t *y = aux ; + if (m->nfds) { unixmessage_drop(m) ; return (errno = EPROTO, 0) ; } + if (cont) + { + if (!genqdyn_push(&y->serialq, &info->serial)) return 0 ; + if (!bufalloc_put(&y->out, m->s, m->len)) goto err ; + if (!bufalloc_put(&y->out, &y->sep[0], 1)) goto berr ; + tain_add_g(&deadline, &y->tto) ; + if (tain_less(&deadline, &y->deadline)) y->deadline = deadline ; + return 1 ; + } + else + { + tain_uint(&deadline, 2) ; + tain_add_g(&deadline, &deadline) ; + return skabus_rpc_reply_g(a, info->serial, ESRCH, "", 0, &deadline) ; + } + berr: + bufalloc_unput(&y->out, m->len) ; + err: + genqdyn_unpush(&y->serialq) ; + return 0 ; +} + +static int rclient_cancel_function (uint64_t serial, char reason, void *aux) +{ + (void)serial ; + (void)reason ; + (void)aux ; + return 1 ; +} + +static int interface_add (char const *ifname, char const *ifprog, char const *re, unsigned int milli, char const *sep) +{ + uint32_t yy ; + interface_t *y ; + int fd[2] ; + tain_t deadline ; + skabus_rpc_interface_t ifbody = { .f = &rclient_function, .cancelf = &rclient_cancel_function, .aux = y } + char const *argv[4] = { EXECLINE_EXTBINPREFIX "execlineb", "-c", ifprog, 0 } ; + pid_t pid = child_spawn2(argv[0], argv, (char const *const *)environ, fd) ; + if (!pid) return 0 ; + if (!gensetdyn_new(&interfaces, &yy)) goto err ; + y = INTERFACE(yy) ; + y->storage = stralloc_zero ; + if (!stralloc_catb(&y->storage, ifname, strlen(ifname)+1)) goto err ; + tain_uint(&deadline, 2) ; + tain_add_g(&deadline, &deadline) ; + if (!skabus_rpc_interface_register_g(a, &y->id, ifname, &ifbody, re, &deadline)) goto ferr ; + y->pid = pid ; + buffer_init(&y->in, &buffer_read, fd[0], y->buf, SKABUS_RPCC_BUFSIZE) ; + y->insa = stralloc_zero ; + bufalloc_init(&y->out, &fd_write, fd[1]) ; + tain_from_millisecs(&y->tto, milli) ; + tain_add_g(&y->deadline, tain_infinite_relative) ; + genqdyn_init(&y->serialq, sizeof(uint64_t), 3, 8) ; + y->sep[0] = sep[0] ; + y->sep[1] = sep[1] ; + y->next = INTERFACE(interface_sentinel)->next ; + INTERFACE(interface_sentinel)->next = yy ; + return 1 ; + + ferr: + stralloc_free(&y->storage) ; + err: + fd_close(fd[1]) ; + fd_close(fd[0]) ; + return 0 ; +} + +static inline int interface_lookup_by_name (char const *ifname, uint32_t *n, uint32_t *nprev) +{ + uint32_t i = interface_sentinel->next, prev = interface_sentinel ; + while (i != interface_sentinel) + { + interface_t *y = INTERFACE(i) ; + if (!strcmp(y->storage.s, ifname)) + { + *n = i ; + *nprev = prev ; + return 1 ; + } + prev = i ; + i = y->next ; + } + return 0 ; +} + +static int interface_remove (char const *ifname) +{ + tain_t deadline ; + uint32_t i, prev, id ; + if (!interface_lookup_by_name(ifname, &i, &prev)) return 0 ; + id = INTERFACE(i)->id ; + interface_delete(i, prev) ; + tain_uint(&deadline, 2) ; + tain_add_g(&deadline, &deadline) ; + return skabus_rpc_interface_unregister_g(a, INTERFACE(i)->id, &deadline) ; +} + +static inline int interface_prepare_iopause (uint32_t i, tain_t *deadline, iopause_fd *x, uint32_t *j) +{ + interface_t *y = INTERFACE(i) ; + if (tain_less(&y->deadline, deadline)) *deadline = y->deadline ; + if (!bufalloc_isempty(&y->out)) + { + x[*j].fd = bufalloc_fd(&y->out) ; + x[*j].events = IOPAUSE_WRITE ; + y->xindex[0] = (*j)++ ; + } + else y->xindex[0] = 0 ; + if (genqdyn_n(&y->serialq)) + { + x[*j].fd = buffer_fd(&y->in) ; + x[*j].events = IOPAUSE_READ ; + y->xindex[1] = (*j)++ ; + } + else y->xindex[1] = 0 ; + return y->xindex[0] || y->xindex[1] ; +} + +static inline int interface_event (iopause_fd const *x, uint32_t i) +{ + interface_t *y = INTERFACE(i) ; + if (y->xindex[1] && x[y->xindex[1]].revents & IOPAUSE_READ) + { + if (!genqdyn_n(&y->serialq)) return 0 ; + while (genqdyn_n(&y->serialq)) + { + tain_t deadline ; + int r = sanitize_read(skagetln(&y->in, &y->insa, y->sep[1])) ; + if (r < 0) return 0 ; + if (!r) break ; + tain_uint(&deadline, 2) ; + tain_add_g(&deadline, &deadline) ; + if (!skabus_rpc_reply(a, *GENQDYN_PEEK(uint64_t, &y->serialq), 0, y->insa.s, y->insa.len - 1, &deadline)) return 0 ; + genqdyn_pop(&y->serialq) ; + } + } + if (y->xindex[0] && x[y->xindex[0]].revents & IOPAUSE_WRITE) + { + if (!bufalloc_flush(&y->out) && !error_isagain(errno)) return 0 ; + } +} + + + +static int query (char const *ifname, char const *q, tain_t const *limit, uint64_t *qid) +{ + uint64_t id ; + tain_t deadline ; + tain_uint(&deadline, 2) ; + tain_add_g(&deadline, &deadline) ; + id = skabus_rpc_send_g(a, ifname, q, strlen(q), limit, &deadline) ; + if (!id) return 0 ; + *qid = id ; + return 1 ; +} + + + /* Clients */ + +typedef struct client_s client_t, *client_t_ref ; +struct client_s +{ + uint32_t next ; + uint32_t xindex ; + textmessage_sender_t out ; + textmessage_receiver_t in ; + tain_t deadline ; + uint64_t query ; +} ; + +static gensetdyn clients = GENSETDYN_INIT(client_t, 1, 0, 1) ; +static uint32_t client_sentinel ; +#define CLIENT(i) GENSETDYN_P(client_t, &clients, (i)) +#define numclients (gensetdyn_n(&clients) - 1) + +static inline void client_free (client_t *c) +{ + fd_close(textmessage_sender_fd(&c->out)) ; + textmessage_sender_free(&c->out) ; + textmessage_receiver_free(&c->in) ; +} + +static inline void client_delete (uint32_t cc, uint32_t prev) +{ + client_t *c = CLIENT(cc) ; + CLIENT(prev)->next = c->next ; + client_free(c) ; + gensetdyn_delete(clients, cc) ; +} + +static void client_setdeadline (client_t *c) +{ + tain_t blah ; + tain_half(&blah, &tain_infinite_relative) ; + tain_add_g(&blah, &blah) ; + if (tain_less(&blah, &c->deadline)) + tain_add_g(&c->deadline, &answertto) ; +} + +static inline int client_prepare_iopause (uint32_t i, tain_t *deadline, iopause_fd *x, uint32_t *j) +{ + client_t *c = CLIENT(i) ; + if (tain_less(&c->deadline, deadline)) *deadline = c->deadline ; + if (!textmessage_sender_isempty(&c->out) || !textmessage_receiver_isempty(&c->in) || (cont && !textmessage_receiver_isfull(&c->in))) + { + x[*j].fd = textmessage_sender_fd(&c->out) ; + x[*j].events = (!textmessage_receiver_isempty(&c->in) || (cont && !textmessage_receiver_isfull(&c->in)) ? IOPAUSE_READ : 0) | (!textmessage_sender_isempty(&c->out) ? IOPAUSE_WRITE : 0) ; + c->xindex = (*j)++ ; + } + else c->xindex = 0 ; + return !!c->xindex ; +} + +static inline int client_flush (uint32_t i, iopause_fd const *x) +{ + client_t *c = CLIENT(i) ; + if (c->xindex && (x[c->xindex].revents & IOPAUSE_WRITE)) + { + if (textmessage_sender_flush(&c->out)) + tain_add_g(&c->deadline, &tain_infinite_relative) ; + else if (!error_isagain(errno)) return 0 ; + } + return 1 ; +} + +static inline int client_add (uint32_t *cc, int fd) +{ + + *cc = genset_new(clients) ; + client_t *c = CLIENT(*cc) ; + tain_add_g(&c->deadline, &answertto) ; + textmessage_sender_init(&c->out, fd) ; + textmessage_receiver_init(&c->in, fd) ; + c->curquery = 0 ; + c->next = CLIENT(sentinel)->next ; + CLIENT(sentinel)->next = *cc ; +} + +static int answer (uint32_t cc, char e) +{ + client_t *c = CLIENT(cc) ; + if (!textmessage_put(&c->out, &e, 1)) return 0 ; + client_setdeadline(c) ; + return 1 ; +} + +static inline void handle_signals (void) +{ + for (;;) switch (selfpipe_read()) + { + case -1 : strerr_diefu1sys(111, "selfpipe_read()") ; + case 0 : return ; + case SIGTERM : quit() ; break ; + case SIGCHLD : + { + } + break ; + default : break ; + } +} + +static int do_interface_register (uint32_t cc, char const *s, size_t len) +{ + uint32_t ifproglen, relen ; + unsigned char ifnamelen ; + if (len < 15) return (errno = EPROTO, 0) ; + ifnamelen = s[0] ; + uint32_unpack_big(s + 1, &ifproglen) ; + uint32_unpack_big(s + 5, &relen) ; + if (len != 12 + ifnamelen + ifproglen + relen) return (errno = EPROTO, 0) ; + if (s[9 + ifnamelen] || s[10 + ifnamelen + ifproglen] || s[11 + ifnamelen + ifproglen + relen]) return (errno = EPROTO, 0) ; + if (!interface_add(s + 9, s + 10 + ifnamelen, s + 11 + ifnamelen + ifproglen)) return answer(cc, errno) ; + return answer(cc, 0) ; +} + +static int do_interface_unregister (uint32_t cc, char const *s, size_t len) +{ + unsigned char ifnamelen ; + if (len < 3) return (errno = EPROTO, 0) ; + ifnamelen = s[0] ; + if ((len != ifnamelen + 2) || s[ifnamelen+1]) return (errno = EPROTO, 0) ; + if (!interface_remove(s+1)) return answer(cc, errno) ; + return answer(cc, 0) ; +} + +static int do_query (uint32_t cc, char const *s, size_t len) +{ + client_t *c = CLIENT(cc) ; + tain_t limit ; + uint32_t querylen, timeout ; + unsigned char ifnamelen ; + if (len < 12) return (errno = EPROTO, 0) ; + ifnamelen = s[0] ; + uint32_unpack_big(s + 1, &timeout) ; + uint32_unpack_big(s + 5, &querylen) ; + if (len != 11 + ifnamelen + querylen || s[9 + ifnamelen] || s[10 + ifnamelen + querylen]) return (errno = EPROTO, 0) ; + if (timeout) tain_from_millisecs(&limit, timeout) ; + else limit = tain_infinite_relative ; + tain_add_g(&limit, &limit) ; + if (!query(s + 9, s + 10 + ifnamelen, &limit, &c->query)) return answer(cc, errno) ; + return answer(cc, 0) ; +} + +static int do_quit (uint32_t cc, char const *s, size_t len) +{ + if (len) return (errno = EPROTO, 0) ; + quit() ; + (void)s ; + return answer(cc, 0) ; +} + +static int do_error (uint32_t cc, char const *s, size_t len) +{ + (void)cc ; + (void)s ; + (void)len ; + return (errno = EPROTO, 0) ; +} + +typedef int parsefunc_t (uint32_t, char const *, size_t) ; +typedef parsefunc_t *parsefunc_t_ref ; + +static int parse_client_protocol (struct iovec const *v, void *p) +{ + static parsefunc_t_ref const f[5] = + { + &do_interface_register, + &do_interface_unregister, + &do_query, + &do_quit, + &do_error + } ; + if (!v->iov_len) return (errno = EPROTO, 0) ; + return *f[byte_chr("IiQ.", 4, *(char const *)v->iov_base)])(*(uint32_t *)p, (char const *)v->iov_base + 1, v->iov_len - 1) ; +} + +static void removeclient (uint32_t *i, uint32_t j) +{ + if (verbosity >= 2) + { + char fmt[UINT32_FMT] ; + fmt[uint32_fmt(fmt, *i)] = 0 ; + strerr_warni2sys("removing client ", fmt) ; + } + client_remove(*i, j) ; + *i = j ; +} + +int main (int argc, char const *const *argv, char const *const *envp) +{ + tain_t deadline ; + int spfd ; + int flag1 = 0 ; + uint32_t maxconn = 8 ; + unsigned int ifn = 1 ; + PROG = "skabus-rpcc" ; + + { + subgetopt_t l = SUBGETOPT_ZERO ; + unsigned int t = 0, T = 0 ; + for (;;) + { + int opt = subgetopt_r(argc, argv, "v:1t:T:c:C:y:n:", &l) ; + if (opt == -1) break ; + switch (opt) + { + case 'v' : if (!uint0_scan(l.arg, &verbosity)) dieusage() ; break ; + case '1' : flag1 = 1 ; break ; + case 't' : if (!uint0_scan(l.arg, &t)) dieusage() ; break ; + case 'T' : if (!uint0_scan(l.arg, &T)) dieusage() ; break ; + case 'c' : if (!uint320_scan(l.arg, &maxconn)) dieusage() ; break ; + case 'C' : interfaces[0] = l.arg ; break ; + case 'y' : + { + if (ifn >= SKABUS_RPCC_INTERFACES_MAX) dietoomanyinterfaces() ; + if (!l.arg[0]) dieemptyifname() ; + interfaces[ifn++] = l.arg ; + break ; + } + default : dieusage() ; + } + } + argc -= l.ind ; argv += l.ind ; + if (t) tain_from_millisecs(&answertto, t) ; + if (T) tain_from_millisecs(&lameduckdeadline, T) ; + } + if (argc < 2) dieusage() ; + if (maxconn > SKABUS_RPCC_MAX) maxconn = SKABUS_RPCC_MAX ; + if (!maxconn) maxconn = 1 ; + { + struct stat st ; + if (fstat(0, &st) < 0) strerr_diefu1sys(111, "fstat stdin") ; + if (!S_ISSOCK(st.st_mode)) strerr_dief1x(100, "stdin is not a socket") ; + } + if (flag1) + { + if (fcntl(1, F_GETFD) < 0) + strerr_dief1sys(100, "called with option -1 but stdout said") ; + } + else close(1) ; + spfd = selfpipe_init() ; + if (spfd < 0) strerr_diefu1sys(111, "selfpipe_init") ; + if (sig_ignore(SIGPIPE) < 0) strerr_diefu1sys(111, "ignore SIGPIPE") ; + { + sigset_t set ; + sigemptyset(&set) ; + sigaddset(&set, SIGTERM) ; + sigaddset(&set, SIGHUP) ; + if (selfpipe_trapset(&set) < 0) strerr_diefu1sys(111, "trap signals") ; + } + + tain_now_g() ; + tain_add_g(&deadline, &answertto) ; + + { + genset clientinfo ; + client_t clientstorage[1+maxconn] ; + uint32_t clientfreelist[1+maxconn] ; + iopause_fd x[3 + (maxconn << 1) + (SKABUS_RPCC_INTERFACES_MAX) << 1] ; + GENSET_init(&clientinfo, client_t, clientstorage, clientfreelist, 1+maxconn) ; + sentinel = genset_new(&clientinfo) ; + clientstorage[sentinel].next = sentinel ; + clients = &clientinfo ; + x[0].fd = spfd ; x[0].events = IOPAUSE_READ ; + x[1].fd = 0 ; + + if (flag1) + { + fd_write(1, "\n", 1) ; + fd_close(1) ; + } + + for (;;) + { + int r = 1 ; + uint32_t i = clientstorage[sentinel].next, j = 2 ; + query_get_mindeadline(&deadline) ; + if (!cont && tain_less(&lameduckdeadline, &deadline)) deadline = lameduckdeadline ; + if (queries_pending()) r = 0 ; + + x[1].events = (cont && (numconn < maxconn)) ? IOPAUSE_READ : 0 ; + for (; i != sentinel ; i = clientstorage[i].next) + if (client_prepare_iopause(i, &deadline, x, &j, cont)) r = 0 ; + if (!cont && r) break ; + + r = iopause_g(x, j, &deadline) ; + if (r < 0) strerr_diefu1sys(111, "iopause") ; + + + /* Timeout */ + + if (!r) + { + if (!cont && !tain_future(&lameduckdeadline)) return 1 ; + for (;;) + { + if (!query_lookup_by_mindeadline(&i)) break ; + if (tain_future(&QUERY(i)->deadline)) break ; + query_fail(i, ETIMEDOUT) ; + } + errno = ETIMEDOUT ; + for (i = clientstorage[sentinel].next, j = sentinel ; i != sentinel ; j = i, i = clientstorage[i].next) + if (!tain_future(&clientstorage[i].deadline)) removeclient(&i, j) ; + continue ; + } + + + /* Signal */ + + if (x[0].revents & IOPAUSE_READ) handle_signals() ; + + + /* Event */ + + for (j = sentinel, i = clientstorage[sentinel].next ; i != sentinel ; j = i, i = clientstorage[i].next) + if (!client_flush(i, x)) removeclient(&i, j) ; + + for (j = sentinel, i = clientstorage[sentinel].next ; i != sentinel ; j = i, i = clientstorage[i].next) + switch(client_read(i, x)) + { + case 0 : errno = 0 ; + case -1 : + case -2 : + { + removeclient(&i, j) ; + break ; + } + case 1 : break ; + default : X() ; + } + + + /* New connection */ + + if (x[1].revents & IOPAUSE_READ) + { + uint32_t flags = 0 ; + uid_t uid ; + gid_t gid ; + regex_t idstr_re, interfaces_re ; + int fd = ipc_accept_nb(x[1].fd, 0, 0, 0) ; + if (fd < 0) + if (!error_isagain(errno)) strerr_diefu1sys(111, "accept") ; + else continue ; + else if (!new_connection(fd, &uid, &gid, &idstr_re, &interfaces_re, &flags)) + fd_close(fd) ; + else client_add(&i, &idstr_re, &interfaces_re, uid, gid, fd, flags) ; + } + } + } + return 0 ; +} diff --git a/src/rpc/skabus-rpcc.h b/src/rpc/skabus-rpcc.h new file mode 100644 index 0000000..9f3557f --- /dev/null +++ b/src/rpc/skabus-rpcc.h @@ -0,0 +1,18 @@ +/* ISC license. */ + +#ifndef SKABUS_RPCC_H +#define SKABUS_RPCC_H + +#include <skalibs/strerr2.h> + +#define SKABUS_RPCC_BUFSIZE 4095 +#define SKABUS_RPCC_MAX 256 +#define SKABUS_RPCC_INTERFACES_MAX 64 +#define SKABUS_RPCC_QUERIES_MAX 1024 + +#define X() strerr_dief1x(101, "unexpected error - please submit a bug-report.") +#define dietoomanyinterfaces() strerr_dief1x(100, "too many interface definitions") +#define dieemptyifname() strerr_dief1x(100, "an interface name may not be empty") +#define dieemptyifprog() strerr_dief1x(100, "an interface program may not be empty") + +#endif diff --git a/src/rpc/skabus-rpccctl.c b/src/rpc/skabus-rpccctl.c new file mode 100644 index 0000000..af9ed72 --- /dev/null +++ b/src/rpc/skabus-rpccctl.c @@ -0,0 +1,148 @@ +/* ISC license. */ + +#include <stdint.h> +#include <string.h> +#include <skalibs/types.h> +#include <skalibs/sgetopt.h> +#include <skalibs/strerr2.h> +#include <skalibs/tai.h> +#include <skalibs/buffer.h> +#include <skalibs/stralloc.h> +#include "skabus-rpccctl.h" + +#define USAGE "skabus-rpccctl [ -t timeout ] help|interface|interface-remove|query args..." +#define dieusage() strerr_dieusage(100, USAGE) + +static tain_t deadline ; +static skabus_rpcc_t a = SKABUS_RPCC_ZERO ; + +static void add_interface (char const *rpccpath, char const *ifname, char const *ifprog, char const *re) +{ + if (!skabus_rpcc_start_g(&a, rpccpath, &deadline)) + strerr_diefu2sys(111, "start session with skabus-rpcc instance at ", rpccpath) ; + if (!skabus_rpcc_interface_register_g(&a, ifname, ifprog, re, &deadline)) + strerr_diefu6sys(111, "register interface ", ifname, " with body ", ifprog, " and regex ", re) ; + skabus_rpcc_end(&a) ; +} + +static void remove_interface (char const *rpccpath, char const *ifname) +{ + if (!skabus_rpcc_start_g(&a, rpccpath, &deadline)) + strerr_diefu2sys(111, "start session with skabus-rpcc instance at ", rpccpath) ; + if (!skabus_rpcc_interface_unregister_g(&a, ifname, &deadline)) + strerr_diefu2sys(111, "unregister interface ", ifname) ; + skabus_rpcc_end(&a) ; +} + +static void query (char const *rpccpath, char const *ifname, char const *query, uint32_t timeout) +{ + stralloc sa = STRALLOC_ZERO ; + if (!skabus_rpcc_start_g(&a, rpccpath, &deadline)) + strerr_diefu2sys(111, "start session with skabus-rpcc instance at ", rpccpath) ; + if (!skabus_rpcc_query_g(&a, &sa, ifname, query, timeout, &deadline)) + strerr_diefu2sys(111, "send query to interface ", ifname) ; + skabus_rpcc_end(&a) ; + if (buffer_putflush(buffer_1, sa.s, sa.len) < 0) + strerr_diefu1sys(111, "write to stdout") ; + stralloc_free(&sa) ; +} + +static void quit (char const *rpccpath) +{ + if (!skabus_rpcc_start_g(&a, rpccpath, &deadline)) + strerr_diefu2sys(111, "start session with skabus-rpcc instance at ", rpccpath) ; + if (!skabus_rpcc_quit_g(&a, &deadline)) + strerr_diefu1sys(111, "send quit command") ; + skabus_rpcc_end(&a) ; +} + +static inline void print_help (void) +{ + static char const *help = +"skabus-rpccctl help\n" +"skabus-rpccctl [ -t timeout ] interface rpccpath interface-name interface-program clientid-regex\n" +"skabus-rpccctl [ -t timeout ] interface-remove rpccpath interface-name\n" +"skabus-rpccctl [ -t timeout ] [ -T limit ] query rpccpath interface-name query\n" +"skabus-rpccctl quit\n" ; + if (buffer_putsflush(buffer_1, help) < 0) + strerr_diefu1sys(111, "write to stdout") ; +} + +static inline unsigned int lookup (char const *const *table, char const *command) +{ + unsigned int i = 0 ; + for (; table[i] ; i++) if (!strcmp(command, table[i])) break ; + return i ; +} + +static inline unsigned int parse_command (char const *command) +{ + static char const *const command_table[13] = + { + "help", + "interface", + "interface-remove", + "query", + "quit", + 0 + } ; + unsigned int i = lookup(command_table, command) ; + if (!command_table[i]) dieusage() ; + return i ; +} + +int main (int argc, char const *const *argv) +{ + unsigned int what ; + uint32_t timeout = 0 ; + PROG = "skabus-rpccctl" ; + { + unsigned int t = 0 ; + subgetopt_t l = SUBGETOPT_ZERO ; + for (;;) + { + int opt = subgetopt_r(argc, argv, "t:T:", &l) ; + if (opt == -1) break ; + switch (opt) + { + case 't' : if (!uint0_scan(l.arg, &t)) dieusage() ; break ; + case 'T' : if (!uint320_scan(l.arg, &timeout)) dieusage() ; break ; + default : dieusage() ; + } + } + argc -= l.ind ; argv += l.ind ; + if (t) tain_from_millisecs(&deadline, t) ; + else deadline = tain_infinite_relative ; + } + + if (!argc) dieusage() ; + + tain_now_g() ; + tain_add_g(&deadline, &deadline) ; + + what = parse_command(argv[0]) ; + switch (what) + { + case 0 : + print_help() ; + break ; + case 1 : + if (argc < 5) dieusage() ; + add_interface(argv[2], argv[3], argv[4], argv[5]) ; + break ; + case 2 : + if (argc < 3) dieusage() ; + remove_interface(argv[2], argv[3]) ; + break ; + case 3 : + if (argc < 4) dieusage() ; + query(argv[2], argv[3], argv[4], timeout) ; + break ; + case 4 : + if (argc < 2) dieusage() ; + quit(argv[2]) ; + break ; + default : dieusage() ; + } + return 0 ; +} diff --git a/src/rpc/skabus-rpccctl.h b/src/rpc/skabus-rpccctl.h new file mode 100644 index 0000000..fe5b620 --- /dev/null +++ b/src/rpc/skabus-rpccctl.h @@ -0,0 +1,36 @@ +/* ISC license. */ + +#ifndef SKABUS_RPCCCTL_H +#define SKABUS_RPCCCTL_H + +#include <stdint.h> +#include <skalibs/tai.h> +#include <skalibs/stralloc.h> +#include <skalibs/textmessage.h> + +typedef struct skabus_rpcc_s skabus_rpcc_t, *skabus_rpcc_t_ref ; +struct skabus_rpcc_s +{ + textmessage_sender_t out ; + textmessage_receiver_t in ; +} ; +#define SKABUS_RPCC_ZERO { .in = TEXTMESSAGE_RECEIVER_ZERO, .out = TEXTMESSAGE_SENDER_ZERO } + +extern int skabus_rpcc_start (skabus_rpcc_t *, char const *, tain_t const *, tain_t *) ; +#define skabus_rpcc_start_g(a, name, deadline) skabus_rpcc_start(a, name, (deadline), &STAMP) + +extern void skabus_rpcc_end (skabus_rpcc_t *) ; + +extern int skabus_rpcc_interface_register (skabus_rpcc_t *, char const *, char const *, char const *, tain_t const *, tain_t *) ; +#define skabus_rpcc_interface_register_g(a, ifname, ifprog, re, deadline) skabus_rpcc_interface_register(a, ifname, ifprog, re, (deadline), &STAMP) + +extern int skabus_rpcc_interface_unregister (skabus_rpcc_t *, char const *, tain_t const *, tain_t *) ; +#define skabus_rpcc_interface_unregister_g(a, ifname, deadline) skabus_rpcc_interface_unregister(a, ifname, (deadline), &STAMP) + +extern int skabus_rpcc_query (skabus_rpcc_t *, stralloc *, char const *, char const *, uint32_t, tain_t const *, tain_t *) ; +#define skabus_rpcc_query_g(a, reply, ifname, query, timeout, deadline) skabus_rpcc_query(a, reply, ifname, query, timeout, (deadline), &STAMP) + +extern int skabus_rpcc_quit (skabus_rpcc_t *, tain_t const *, tain_t *) ; +#define skabus_rpcc_quit_g(a, deadline) skabus_rpcc_quit(a, (deadline), &STAMP) + +#endif diff --git a/src/rpc/skabus-rpcd.c b/src/rpc/skabus-rpcd.c index 26692f2..167cf76 100644 --- a/src/rpc/skabus-rpcd.c +++ b/src/rpc/skabus-rpcd.c @@ -32,7 +32,7 @@ #include <skabus/rpc.h> #include "skabus-rpcd.h" -#define USAGE "skabus-rpcd [ -v verbosity ] [ -1 ] [ -d | -D ] [ -c maxconn ] [ -t timeout ] [ -T lameducktimeout ] [ -i rulesdir | -x rulesfile ] [ -S | -s ] [ -J | -j ]" +#define USAGE "skabus-rpcd [ -v verbosity ] [ -1 ] [ -c maxconn ] [ -t timeout ] [ -T lameducktimeout ] [ -i rulesdir | -x rulesfile ] [ -S | -s ] [ -J | -j ]" #define dieusage() strerr_dieusage(100, USAGE) ; tain_t answertto = TAIN_INFINITE_RELATIVE ; @@ -45,7 +45,6 @@ static unsigned int rulestype = 0 ; static char const *rules = 0 ; static int cdbfd = -1 ; static struct cdb cdbmap = CDB_ZERO ; -static int flagidstrpub = 0, flaginterfacespub = 0 ; static inline void handle_signals (void) { @@ -274,9 +273,9 @@ static int makere (regex_t *re, char const *s, char const *var) return 0 ; } -static void defaultre (regex_t *re, int flag) +static void defaultre (regex_t *re, unsigned int pubflag) { - char const *s = flag ? ".*" : ".^" ; + char const *s = pubflag ? ".*" : ".^" ; int r = regcomp(re, s, REG_EXTENDED | REG_NOSUB) ; if (r) { @@ -286,44 +285,45 @@ static void defaultre (regex_t *re, int flag) } } -static inline int parse_env (char const *const *envp, regex_t *idstr_re, regex_t *interfaces_re, uint32_t *flags) +static inline int parse_env (char const *const *envp, regex_t *idstr_re, regex_t *interfaces_re, uint32_t *flags, unsigned int *donep) { uint32_t fl = 0 ; - int idstr_done = 0, interfaces_done = 0 ; + unsigned int done = 0 ; for (; *envp ; envp++) { if (str_start(*envp, "SKABUS_RPC_QSENDFDS=")) fl |= 1 ; if (str_start(*envp, "SKABUS_RPC_RSENDFDS=")) fl |= 2 ; - if (!idstr_done) + if (!(done & 1)) { - idstr_done = makere(idstr_re, *envp, "SKABUS_RPC_ID_REGEX") ; - if (idstr_done < 0) + int r = makere(idstr_re, *envp, "SKABUS_RPC_ID_REGEX") ; + if (r < 0) { - if (interfaces_done) regfree(interfaces_re) ; + if (done & 2) regfree(interfaces_re) ; return 0 ; } + else if (r) done |= 1 ; } - if (!interfaces_done) + if (!(done & 2)) { - interfaces_done = makere(interfaces_re, *envp, "SKABUS_RPC_INTERFACES_REGEX") ; - if (interfaces_done < 0) + int r = makere(interfaces_re, *envp, "SKABUS_RPC_INTERFACES_REGEX") ; + if (r < 0) { - if (idstr_done) regfree(idstr_re) ; + if (done & 1) regfree(idstr_re) ; return 0 ; } + else if (r) done |= 2 ; } - if (idstr_done && interfaces_done) return 1 ; } - if (!idstr_done) defaultre(idstr_re, flagidstrpub) ; - if (!interfaces_done) defaultre(interfaces_re, flaginterfacespub) ; *flags = fl ; + *donep = done ; return 1 ; } -static inline int new_connection (int fd, uid_t *uid, gid_t *gid, regex_t *idstr_re, regex_t *interfaces_re, uint32_t *flags) +static inline int new_connection (int fd, uid_t *uid, gid_t *gid, regex_t *idstr_re, regex_t *interfaces_re, uint32_t *flags, unsigned int pubflags) { s6_accessrules_params_t params = S6_ACCESSRULES_PARAMS_ZERO ; s6_accessrules_result_t result = S6_ACCESSRULES_ERROR ; + unsigned int done = 0 ; if (getpeereid(fd, uid, gid) < 0) { @@ -347,17 +347,13 @@ static inline int new_connection (int fd, uid_t *uid, gid_t *gid, regex_t *idstr strerr_warnw1sys("error while checking rules") ; return 0 ; } - if (params.exec.s) + if (params.exec.len && verbosity) { - stralloc_free(¶ms.exec) ; - if (verbosity) - { - char fmtuid[UID_FMT] ; - char fmtgid[GID_FMT] ; - fmtuid[uid_fmt(fmtuid, *uid)] = 0 ; - fmtgid[gid_fmt(fmtgid, *gid)] = 0 ; - strerr_warnw4x("unused exec string in rules for uid ", fmtuid, " gid ", fmtgid) ; - } + char fmtuid[UID_FMT] ; + char fmtgid[GID_FMT] ; + fmtuid[uid_fmt(fmtuid, *uid)] = 0 ; + fmtgid[gid_fmt(fmtgid, *gid)] = 0 ; + strerr_warnw4x("unused exec string in rules for uid ", fmtuid, " gid ", fmtgid) ; } if (params.env.s) { @@ -366,18 +362,20 @@ static inline int new_connection (int fd, uid_t *uid, gid_t *gid, regex_t *idstr if (!env_make(envp, n, params.env.s, params.env.len)) { if (verbosity) strerr_warnwu1sys("env_make") ; - stralloc_free(¶ms.env) ; + s6_accessrules_params_free(¶ms) ; return 0 ; } envp[n] = 0 ; - if (!parse_env(envp, idstr_re, interfaces_re, flags)) + if (!parse_env(envp, idstr_re, interfaces_re, flags, &done)) { if (verbosity) strerr_warnwu1sys("parse_env") ; s6_accessrules_params_free(¶ms) ; return 0 ; } - s6_accessrules_params_free(¶ms) ; } + s6_accessrules_params_free(¶ms) ; + if (!(done & 1)) defaultre(idstr_re, pubflags & 1) ; + if (!(done & 2)) defaultre(interfaces_re, pubflags & 2) ; return 1 ; } @@ -386,6 +384,7 @@ int main (int argc, char const *const *argv, char const *const *envp) int spfd ; int flag1 = 0 ; uint32_t maxconn = 64 ; + unsigned int pubflags = 0 ; PROG = "skabus-rpcd" ; { @@ -398,10 +397,10 @@ int main (int argc, char const *const *argv, char const *const *envp) switch (opt) { case 'v' : if (!uint0_scan(l.arg, &verbosity)) dieusage() ; break ; - case 'S' : flagidstrpub = 0 ; break ; - case 's' : flagidstrpub = 1 ; break ; - case 'J' : flaginterfacespub = 0 ; break ; - case 'j' : flaginterfacespub = 1 ; break ; + case 'S' : pubflags &= ~1U ; break ; + case 's' : pubflags |= 1U ; break ; + case 'J' : pubflags &= ~2U ; break ; + case 'j' : pubflags |= 2U ; break ; case '1' : flag1 = 1 ; break ; case 'i' : rules = l.arg ; rulestype = 1 ; break ; case 'x' : rules = l.arg ; rulestype = 2 ; break ; @@ -539,7 +538,7 @@ int main (int argc, char const *const *argv, char const *const *envp) if (fd < 0) if (!error_isagain(errno)) strerr_diefu1sys(111, "accept") ; else continue ; - else if (!new_connection(fd, &uid, &gid, &idstr_re, &interfaces_re, &flags)) + else if (!new_connection(fd, &uid, &gid, &idstr_re, &interfaces_re, &flags, pubflags)) fd_close(fd) ; else client_add(&i, &idstr_re, &interfaces_re, uid, gid, fd, flags) ; } diff --git a/src/rpc/skabus_rpccctl.c b/src/rpc/skabus_rpccctl.c new file mode 100644 index 0000000..a57f3be --- /dev/null +++ b/src/rpc/skabus_rpccctl.c @@ -0,0 +1,121 @@ + /* ISC license. */ + +#include <sys/uio.h> +#include <stdint.h> +#include <string.h> +#include <errno.h> +#include <stdlib.h> +#include <limits.h> +#include <skalibs/uint32.h> +#include <skalibs/error.h> +#include <skalibs/stralloc.h> +#include <skalibs/djbunix.h> +#include <skalibs/webipc.h> +#include <skalibs/textmessage.h> +#include <skabus/rpc.h> +#include "skabus-rpccctl.h" + +int skabus_rpcc_start (skabus_rpcc_t *a, char const *path, tain_t const *deadline, tain_t *stamp) +{ + int fd = ipc_stream_nb() ; + if (fd < 0) return 0 ; + if (!ipc_timed_connect(fd, path, deadline, stamp)) + { + fd_close(fd) ; + return 0 ; + } + textmessage_sender_init(&a->out, fd) ; + textmessage_receiver_init(&a->in, fd) ; + return 1 ; +} + +void skabus_rpcc_end (skabus_rpcc_t *a) +{ + fd_close(textmessage_sender_fd(&a->out)) ; + textmessage_sender_free(&a->out) ; + textmessage_receiver_free(&a->in) ; +} + +int skabus_rpcc_interface_register (skabus_rpcc_t *a, char const *ifname, char const *ifprog, char const *re, tain_t const *deadline, tain_t *stamp) +{ + size_t ifnamelen, ifproglen, relen ; + char *ifprogfn = realpath(ifprog, 0) ; + if (!ifprogfn) return 0 ; + ifnamelen = strlen(ifname) ; + ifproglen = strlen(ifprogfn) ; + relen = strlen(re) ; + if (ifnamelen > SKABUS_RPC_INTERFACE_MAXLEN || ifproglen > PATH_MAX || relen > SKABUS_RPC_RE_MAXLEN) goto terr ; + { + char buf[9] ; + struct iovec v[5] = + { + { .iov_base = "I", .iov_len = 1 }, + { .iov_base = buf, .iov_len = 9 }, + { .iov_base = ifname, .iov_len = ifnamelen + 1 }, + { .iov_base = ifprogfn, .iov_len = ifproglen + 1 }, + { .iov_base = re, .iov_len = relen + 1 } + } ; + buf[0] = (unsigned char)ifnamelen ; + uint32_pack_big(buf + 1, ifproglen) ; + uint32_pack_big(buf + 5, relen) ; + if (!textmessage_timed_commandv(&a->out, v, 5, deadline, stamp)) goto err ; + } + return 1 ; + + terr: + errno = ENAMETOOLONG ; + err: + free(ifprogfn) ; + return 0 ; +} + +int skabus_rpcc_interface_unregister (skabus_rpcc_t *a, char const *ifname, tain_t const *deadline, tain_t *stamp) +{ + size_t ifnamelen = strlen(ifname) ; + if (ifnamelen > SKABUS_RPC_INTERFACE_MAXLEN) return (errno = ENAMETOOLONG, 0) ; + { + unsigned char c = ifnamelen ; + struct iovec v[3] = + { + { .iov_base = "i", .iov_len = 1 }, + { .iov_base = &c, .iov_len = 1 }, + { .iov_base = ifname, .iov_len = ifnamelen + 1 } + } ; + if (!textmessage_timed_commandv(&a->out, v, 3, deadline, stamp)) return 0 ; + } + return 1 ; +} + +int skabus_rpcc_query (skabus_rpcc_t *a, stralloc *reply, char const *ifname, char const *query, uint32_t timeout, tain_t const *deadline, tain_t *stamp) +{ + size_t ifnamelen = strlen(ifname) ; + size_t querylen = strlen(query) ; + if (ifnamelen > SKABUS_RPC_INTERFACE_MAXLEN || querylen > UINT32_MAX) return (errno = ENAMETOOLONG, 0) ; + { + char buf[9] ; + struct iovec v[4] = + { + { .iov_base = "Q", .iov_len = 1 }, + { .iov_base = buf, .iov_len = 59 }, + { .iov_base = ifname, .iov_len = ifnamelen + 1 }, + { .iov_base = query, .iov_len = querylen + 1 }, + } ; + buf[0] = ifnamelen ; + uint32_pack_big(buf + 1, querylen) ; + uint32_pack_big(buf + 5, timeout) ; + if (!textmessage_timed_sendv(&a->out, v, 4)) return 0 ; + } + { + struct iovec v ; + if (!textmessage_timed_receive(&a->in, &v, deadline, stamp)) return 0 ; + if (!v.iov_len) return (errno = EPROTO, 0) ; + if (*(unsigned char *)v.iov_base) return (errno = *(unsigned char)v.iov_base, 0) ; + if (!stralloc_catb(reply, (char *)v.iov_base + 1, v.iov_len - 1)) return 0 ; + } + return 1 ; +} + +int skabus_rpcc_quit (skabus_rpcc_t *a, tain_t const *deadline, tain_t *stamp) +{ + return textmessage_timed_command(&a->out, ".", 1, deadline, stamp) ; +} |