diff options
Diffstat (limited to 'smtpd/filter.c')
| -rw-r--r-- | smtpd/filter.c | 868 |
1 files changed, 868 insertions, 0 deletions
diff --git a/smtpd/filter.c b/smtpd/filter.c new file mode 100644 index 00000000..614486b7 --- /dev/null +++ b/smtpd/filter.c @@ -0,0 +1,868 @@ +/* $OpenBSD: filter.c,v 1.25 2017/01/09 09:53:23 reyk Exp $ */ + +/* + * Copyright (c) 2011 Gilles Chehade <gilles@poolp.org> + * Copyright (c) 2012 Eric Faurot <eric@openbsd.org> + * + * Permission to use, copy, modify, and distribute this software for any + * purpose with or without fee is hereby granted, provided that the above + * copyright notice and this permission notice appear in all copies. + * + * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES + * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF + * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR + * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES + * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN + * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF + * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. + */ + +#include "includes.h" + +#include <sys/types.h> +#include <sys/queue.h> +#include <sys/tree.h> +#include <sys/socket.h> +#include <sys/wait.h> + +#include <netinet/in.h> + +#include <ctype.h> +#include <errno.h> +#include <event.h> +#include <imsg.h> +#include <inttypes.h> +#include <limits.h> +#include <resolv.h> +#include <signal.h> +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <unistd.h> + +#include "smtpd.h" +#include "log.h" + +enum { + QUERY_READY, + QUERY_RUNNING, + QUERY_DONE +}; + + +struct filter_proc { + TAILQ_ENTRY(filter_proc) entry; + struct mproc mproc; + int hooks; + int flags; + int ready; +}; + +struct filter { + TAILQ_ENTRY(filter) entry; + struct filter_proc *proc; +}; +TAILQ_HEAD(filter_lst, filter); + +TAILQ_HEAD(filter_query_lst, filter_query); +struct filter_session { + uint64_t id; + int terminate; + struct filter_lst *filters; + struct filter *fcurr; + + int error; + struct io *iev; + size_t idatalen; + FILE *ofile; + + struct filter_query *eom; +}; + +struct filter_query { + uint64_t qid; + int type; + struct filter_session *session; + + int state; + struct filter *current; + + /* current data */ + union { + struct { + struct sockaddr_storage local; + struct sockaddr_storage remote; + char hostname[HOST_NAME_MAX+1]; + } connect; + char line[LINE_MAX]; + struct mailaddr maddr; + size_t datalen; + } u; + + /* current response */ + struct { + int status; + int code; + char *response; + } smtp; +}; + +static void filter_imsg(struct mproc *, struct imsg *); +static void filter_post_event(uint64_t, int, struct filter *, struct filter *); +static struct filter_query *filter_query(struct filter_session *, int); +static void filter_drain_query(struct filter_query *); +static void filter_run_query(struct filter *, struct filter_query *); +static void filter_end_query(struct filter_query *); +static void filter_set_sink(struct filter_session *, int); +static int filter_tx(struct filter_session *, int); +static void filter_tx_io(struct io *, int, void *); + +static TAILQ_HEAD(, filter_proc) procs; +struct dict chains; + +static const char * filter_session_to_text(struct filter_session *); +static const char * filter_query_to_text(struct filter_query *); +static const char * filter_to_text(struct filter *); +static const char * filter_proc_to_text(struct filter_proc *); +static const char * query_to_str(int); +static const char * event_to_str(int); +static const char * status_to_str(int); +static const char * filterimsg_to_str(int); + +struct tree sessions; +struct tree queries; + +static void +filter_add_arg(struct filter_conf *filter, char *arg) +{ + if (filter->argc == MAX_FILTER_ARGS) { + log_warnx("warn: filter \"%s\" is full", filter->name); + fatalx("exiting"); + } + filter->argv[filter->argc++] = arg; +} + +static void +filter_extend_chain(struct filter_lst *chain, const char *name) +{ + struct filter *n; + struct filter_lst *fchain; + struct filter_conf *fconf; + int i; + + fconf = dict_xget(&env->sc_filters, name); + if (fconf->chain) { + log_debug("filter: extending with \"%s\"", name); + for (i = 0; i < fconf->argc; i++) + filter_extend_chain(chain, fconf->argv[i]); + } + else { + log_debug("filter: adding filter \"%s\"", name); + n = xcalloc(1, sizeof(*n), "filter_extend_chain"); + fchain = dict_get(&chains, name); + n->proc = TAILQ_FIRST(fchain)->proc; + TAILQ_INSERT_TAIL(chain, n, entry); + } +} + +void +filter_postfork(void) +{ + static int prepare = 0; + struct filter_conf *filter; + void *iter; + struct filter_proc *proc; + struct filter_lst *fchain; + struct filter *f; + struct mproc *p; + int done, i; + + if (prepare) + return; + prepare = 1; + + TAILQ_INIT(&procs); + dict_init(&chains); + + log_debug("filter: building simple chains..."); + + /* create all filter proc and associated chains */ + iter = NULL; + while (dict_iter(&env->sc_filters, &iter, NULL, (void **)&filter)) { + if (filter->chain) + continue; + + log_debug("filter: building simple chain \"%s\"", filter->name); + proc = xcalloc(1, sizeof(*proc), "filter_postfork"); + p = &proc->mproc; + p->handler = filter_imsg; + p->proc = PROC_FILTER; + p->name = xstrdup(filter->name, "filter_postfork"); + p->data = proc; + if (tracing & TRACE_DEBUG) + filter_add_arg(filter, "-v"); + if (foreground_log) + filter_add_arg(filter, "-d"); + if (mproc_fork(p, filter->path, filter->argv) < 0) + fatalx("filter_postfork"); + + log_debug("filter: registering proc \"%s\"", filter->name); + f = xcalloc(1, sizeof(*f), "filter_postfork"); + f->proc = proc; + + TAILQ_INSERT_TAIL(&procs, proc, entry); + fchain = xcalloc(1, sizeof(*fchain), "filter_postfork"); + TAILQ_INIT(fchain); + TAILQ_INSERT_TAIL(fchain, f, entry); + dict_xset(&chains, filter->name, fchain); + filter->done = 1; + } + + log_debug("filter: building complex chains..."); + + /* resolve all chains */ + done = 0; + while (!done) { + done = 1; + iter = NULL; + while (dict_iter(&env->sc_filters, &iter, NULL, + (void **)&filter)) { + if (filter->done) + continue; + done = 0; + filter->done = 1; + for (i = 0; i < filter->argc; i++) { + if (!dict_get(&chains, filter->argv[i])) { + filter->done = 0; + break; + } + } + if (filter->done == 0) + continue; + fchain = xcalloc(1, sizeof(*fchain), "filter_postfork"); + TAILQ_INIT(fchain); + log_debug("filter: building chain \"%s\"...", + filter->name); + for (i = 0; i < filter->argc; i++) + filter_extend_chain(fchain, filter->argv[i]); + log_debug("filter: done building chain \"%s\"", + filter->name); + dict_xset(&chains, filter->name, fchain); + } + } + log_debug("filter: done building complex chains"); + + fchain = xcalloc(1, sizeof(*fchain), "filter_postfork"); + TAILQ_INIT(fchain); + dict_xset(&chains, "<no-filter>", fchain); +} + +void +filter_configure(void) +{ + static int init = 0; + struct filter_proc *p; + + if (init) + return; + init = 1; + + tree_init(&sessions); + tree_init(&queries); + + TAILQ_FOREACH(p, &procs, entry) { + m_create(&p->mproc, IMSG_FILTER_REGISTER, 0, 0, -1); + m_add_u32(&p->mproc, FILTER_API_VERSION); + m_add_string(&p->mproc, p->mproc.name); + m_close(&p->mproc); + mproc_enable(&p->mproc); + } + + if (TAILQ_FIRST(&procs) == NULL) + smtp_configure(); +} + +void +filter_event(uint64_t id, int event) +{ + struct filter_session *s; + + if (event == EVENT_DISCONNECT) + /* On disconnect, the session is virtualy dead */ + s = tree_xpop(&sessions, id); + else + s = tree_xget(&sessions, id); + + filter_post_event(id, event, TAILQ_FIRST(s->filters), NULL); + + if (event == EVENT_DISCONNECT) { + if (s->iev) + io_free(s->iev); + if (s->ofile) + fclose(s->ofile); + free(s); + } +} + +void +filter_connect(uint64_t id, const struct sockaddr *local, + const struct sockaddr *remote, const char *host, const char *filter) +{ + struct filter_session *s; + struct filter_query *q; + + s = xcalloc(1, sizeof(*s), "filter_event"); + s->id = id; + if (filter == NULL) + filter = "<no-filter>"; + s->filters = dict_xget(&chains, filter); + tree_xset(&sessions, s->id, s); + + filter_event(id, EVENT_CONNECT); + q = filter_query(s, QUERY_CONNECT); + + memmove(&q->u.connect.local, local, SA_LEN(local)); + memmove(&q->u.connect.remote, remote, SA_LEN(remote)); + strlcpy(q->u.connect.hostname, host, sizeof(q->u.connect.hostname)); + + q->smtp.status = FILTER_OK; + q->smtp.code = 0; + q->smtp.response = NULL; + + filter_drain_query(q); +} + +void +filter_mailaddr(uint64_t id, int type, const struct mailaddr *maddr) +{ + struct filter_session *s; + struct filter_query *q; + + s = tree_xget(&sessions, id); + q = filter_query(s, type); + + strlcpy(q->u.maddr.user, maddr->user, sizeof(q->u.maddr.user)); + strlcpy(q->u.maddr.domain, maddr->domain, sizeof(q->u.maddr.domain)); + + filter_drain_query(q); +} + +void +filter_line(uint64_t id, int type, const char *line) +{ + struct filter_session *s; + struct filter_query *q; + + s = tree_xget(&sessions, id); + q = filter_query(s, type); + + if (line) + strlcpy(q->u.line, line, sizeof(q->u.line)); + + filter_drain_query(q); +} + +void +filter_eom(uint64_t id, int type, size_t datalen) +{ + struct filter_session *s; + struct filter_query *q; + + s = tree_xget(&sessions, id); + q = filter_query(s, type); + q->u.datalen = datalen; + + filter_drain_query(q); +} + +static void +filter_set_sink(struct filter_session *s, int sink) +{ + struct mproc *p; + + while (s->fcurr) { + if (s->fcurr->proc->hooks & HOOK_DATALINE) { + log_trace(TRACE_FILTERS, "filter: sending fd %d to %s", + sink, filter_to_text(s->fcurr)); + p = &s->fcurr->proc->mproc; + m_create(p, IMSG_FILTER_PIPE, 0, 0, sink); + m_add_id(p, s->id); + m_close(p); + return; + } + s->fcurr = TAILQ_PREV(s->fcurr, filter_lst, entry); + } + + log_trace(TRACE_FILTERS, "filter: chain input is %d", sink); + smtp_filter_fd(s->id, sink); +} + +void +filter_build_fd_chain(uint64_t id, int sink) +{ + struct filter_session *s; + int fd; + + s = tree_xget(&sessions, id); + s->fcurr = TAILQ_LAST(s->filters, filter_lst); + + fd = filter_tx(s, sink); + filter_set_sink(s, fd); +} + +void +filter_post_event(uint64_t id, int event, struct filter *f, struct filter *end) +{ + for(; f && f != end; f = TAILQ_NEXT(f, entry)) { + log_trace(TRACE_FILTERS, "filter: post-event event=%s filter=%s", + event_to_str(event), f->proc->mproc.name); + + m_create(&f->proc->mproc, IMSG_FILTER_EVENT, 0, 0, -1); + m_add_id(&f->proc->mproc, id); + m_add_int(&f->proc->mproc, event); + m_close(&f->proc->mproc); + } +} + +static struct filter_query * +filter_query(struct filter_session *s, int type) +{ + struct filter_query *q; + + q = xcalloc(1, sizeof(*q), "filter_query"); + q->qid = generate_uid(); + q->session = s; + q->type = type; + + q->state = QUERY_READY; + q->current = TAILQ_FIRST(s->filters); + + log_trace(TRACE_FILTERS, "filter: new query %s", query_to_str(type)); + + return (q); +} + +static void +filter_drain_query(struct filter_query *q) +{ + log_trace(TRACE_FILTERS, "filter: filter_drain_query %s", + filter_query_to_text(q)); + + /* + * The query must be passed through all filters that registered + * a hook, until one rejects it. + */ + while (q->state != QUERY_DONE) { + /* Walk over all filters */ + while (q->current) { + filter_run_query(q->current, q); + if (q->state == QUERY_RUNNING) { + log_trace(TRACE_FILTERS, + "filter: waiting for running query %s", + filter_query_to_text(q)); + return; + } + } + q->state = QUERY_DONE; + } + + /* Defer the response if the file is not closed yet. */ + if (q->type == QUERY_EOM && q->session->ofile && q->smtp.status == FILTER_OK) { + log_debug("filter: deferring eom query..."); + q->session->eom = q; + return; + } + + filter_end_query(q); +} + +static void +filter_run_query(struct filter *f, struct filter_query *q) +{ + log_trace(TRACE_FILTERS, + "filter: running filter %s for query %s", + filter_to_text(f), filter_query_to_text(q)); + + m_create(&f->proc->mproc, IMSG_FILTER_QUERY, 0, 0, -1); + m_add_id(&f->proc->mproc, q->session->id); + m_add_id(&f->proc->mproc, q->qid); + m_add_int(&f->proc->mproc, q->type); + + switch (q->type) { + case QUERY_CONNECT: + m_add_sockaddr(&f->proc->mproc, + (struct sockaddr *)&q->u.connect.local); + m_add_sockaddr(&f->proc->mproc, + (struct sockaddr *)&q->u.connect.remote); + m_add_string(&f->proc->mproc, q->u.connect.hostname); + break; + case QUERY_HELO: + m_add_string(&f->proc->mproc, q->u.line); + break; + case QUERY_MAIL: + case QUERY_RCPT: + m_add_mailaddr(&f->proc->mproc, &q->u.maddr); + break; + case QUERY_EOM: + m_add_u32(&f->proc->mproc, q->u.datalen); + break; + default: + break; + } + m_close(&f->proc->mproc); + + tree_xset(&queries, q->qid, q); + q->state = QUERY_RUNNING; +} + +static void +filter_end_query(struct filter_query *q) +{ + struct filter_session *s = q->session; + const char *response = q->smtp.response; + + log_trace(TRACE_FILTERS, "filter: filter_end_query %s", + filter_query_to_text(q)); + + if (q->type == QUERY_EOM && q->smtp.status == FILTER_OK) { + if (s->error || q->u.datalen != s->idatalen) { + response = "Internal error"; + q->smtp.code = 451; + q->smtp.status = FILTER_FAIL; + if (!s->error) + log_warnx("filter: datalen mismatch on session %" PRIx64 + ": %zu/%zu", s->id, s->idatalen, q->u.datalen); + } + } + + log_trace(TRACE_FILTERS, + "filter: query %016"PRIx64" done: " + "status=%s code=%d response=\"%s\"", + q->qid, + status_to_str(q->smtp.status), + q->smtp.code, + response); + + smtp_filter_response(s->id, q->type, q->smtp.status, q->smtp.code, + response); + free(q->smtp.response); + free(q); +} + +static void +filter_imsg(struct mproc *p, struct imsg *imsg) +{ + struct filter_proc *proc = p->data; + struct filter_session *s; + struct filter_query *q; + struct msg m; + const char *line; + uint64_t qid; + uint32_t datalen; + int type, status, code; + + if (imsg == NULL) { + log_warnx("warn: filter \"%s\" closed unexpectedly", p->name); + fatalx("exiting"); + } + + log_trace(TRACE_FILTERS, "filter: imsg %s from procfilter %s", + filterimsg_to_str(imsg->hdr.type), + filter_proc_to_text(proc)); + + switch (imsg->hdr.type) { + + case IMSG_FILTER_REGISTER: + if (proc->ready) { + log_warnx("warn: filter \"%s\" already registered", + proc->mproc.name); + exit(1); + } + + m_msg(&m, imsg); + m_get_int(&m, &proc->hooks); + m_get_int(&m, &proc->flags); + m_end(&m); + proc->ready = 1; + + log_debug("debug: filter \"%s\": hooks 0x%08x flags 0x%04x", + proc->mproc.name, proc->hooks, proc->flags); + + TAILQ_FOREACH(proc, &procs, entry) + if (!proc->ready) + return; + + smtp_configure(); + break; + + case IMSG_FILTER_RESPONSE: + m_msg(&m, imsg); + m_get_id(&m, &qid); + m_get_int(&m, &type); + if (type == QUERY_EOM) + m_get_u32(&m, &datalen); + m_get_int(&m, &status); + m_get_int(&m, &code); + if (m_is_eom(&m)) + line = NULL; + else + m_get_string(&m, &line); + m_end(&m); + + q = tree_xpop(&queries, qid); + if (q->type != type) { + log_warnx("warn: filter: type mismatch %d != %d", + q->type, type); + fatalx("exiting"); + } + q->smtp.status = status; + if (code) + q->smtp.code = code; + if (line) { + free(q->smtp.response); + q->smtp.response = xstrdup(line, "filter_imsg"); + } + q->state = (status == FILTER_OK) ? QUERY_READY : QUERY_DONE; + if (type == QUERY_EOM) + q->u.datalen = datalen; + + q->current = TAILQ_NEXT(q->current, entry); + filter_drain_query(q); + break; + + case IMSG_FILTER_PIPE: + m_msg(&m, imsg); + m_get_id(&m, &qid); + m_end(&m); + + s = tree_xget(&sessions, qid); + s->fcurr = TAILQ_PREV(s->fcurr, filter_lst, entry); + filter_set_sink(s, imsg->fd); + break; + + default: + log_warnx("warn: bad imsg from filter %s", p->name); + exit(1); + } +} + +static int +filter_tx(struct filter_session *s, int sink) +{ + int sp[2]; + + s->idatalen = 0; + s->eom = NULL; + s->error = 0; + + if (socketpair(AF_UNIX, SOCK_STREAM, PF_UNSPEC, sp) == -1) { + log_warn("warn: filter: socketpair"); + return (-1); + } + + if ((s->ofile = fdopen(sink, "w")) == NULL) { + log_warn("warn: filter: fdopen"); + close(sp[0]); + close(sp[1]); + return (-1); + } + + io_set_nonblocking(sp[0]); + io_set_nonblocking(sp[1]); + + s->iev = io_new(); + io_set_callback(s->iev, filter_tx_io, s); + io_set_fd(s->iev, sp[0]); + io_set_read(s->iev); + + return (sp[1]); +} + +static void +filter_tx_io(struct io *io, int evt, void *arg) +{ + struct filter_session *s = arg; + size_t len, n; + char *data; + + log_trace(TRACE_FILTERS, "filter: filter_tx_io(%p, %s)", + s, io_strevent(evt)); + + switch (evt) { + case IO_DATAIN: + data = io_data(s->iev); + len = io_datalen(s->iev); + + log_trace(TRACE_FILTERS, + "filter: filter_tx_io: datain (%zu) for req %016"PRIx64"", + len, s->id); + + n = fwrite(data, 1, len, s->ofile); + if (n != len) { + log_warnx("warn: filter_tx_io: fwrite %zu/%zu", n, len); + s->error = 1; + break; + } + s->idatalen += n; + io_drop(s->iev, n); + return; + + case IO_DISCONNECTED: + log_trace(TRACE_FILTERS, + "debug: filter: tx done (%zu) for req %016"PRIx64, + s->idatalen, s->id); + break; + + default: + log_warn("warn: filter_tx_io: bad evt (%d) for req %016"PRIx64, + evt, s->id); + s->error = 1; + break; + } + + io_free(s->iev); + s->iev = NULL; + fclose(s->ofile); + s->ofile = NULL; + + /* deferred eom request */ + if (s->eom) { + log_debug("filter: running eom query..."); + filter_end_query(s->eom); + } else { + log_debug("filter: eom not received yet"); + } +} + +static const char * +filter_query_to_text(struct filter_query *q) +{ + static char buf[1024]; + char tmp[1024]; + + tmp[0] = '\0'; + + switch (q->type) { + case QUERY_CONNECT: + strlcat(tmp, "=", sizeof tmp); + strlcat(tmp, ss_to_text(&q->u.connect.local), + sizeof tmp); + strlcat(tmp, " <-> ", sizeof tmp); + strlcat(tmp, ss_to_text(&q->u.connect.remote), + sizeof tmp); + strlcat(tmp, "(", sizeof tmp); + strlcat(tmp, q->u.connect.hostname, sizeof tmp); + strlcat(tmp, ")", sizeof tmp); + break; + case QUERY_MAIL: + case QUERY_RCPT: + snprintf(tmp, sizeof tmp, "=%s@%s", + q->u.maddr.user, q->u.maddr.domain); + break; + case QUERY_HELO: + snprintf(tmp, sizeof tmp, "=%s", q->u.line); + break; + default: + break; + } + snprintf(buf, sizeof buf, "%016"PRIx64"[%s%s,%s]", + q->qid, query_to_str(q->type), tmp, + filter_session_to_text(q->session)); + + return (buf); +} + +static const char * +filter_session_to_text(struct filter_session *s) +{ + static char buf[1024]; + + if (s == NULL) + return "filter_session@NULL"; + + snprintf(buf, sizeof(buf), + "filter_session@%p[datalen=%zu,eom=%p,ofile=%p]", + s, s->idatalen, s->eom, s->ofile); + + return buf; +} + +static const char * +filter_to_text(struct filter *f) +{ + static char buf[1024]; + + snprintf(buf, sizeof buf, "filter:%s", filter_proc_to_text(f->proc)); + + return (buf); +} + +static const char * +filter_proc_to_text(struct filter_proc *proc) +{ + static char buf[1024]; + + snprintf(buf, sizeof buf, "%s[hooks=0x%08x,flags=0x%04x]", + proc->mproc.name, proc->hooks, proc->flags); + + return (buf); +} + +#define CASE(x) case x : return #x + +static const char * +filterimsg_to_str(int imsg) +{ + switch (imsg) { + CASE(IMSG_FILTER_REGISTER); + CASE(IMSG_FILTER_EVENT); + CASE(IMSG_FILTER_QUERY); + CASE(IMSG_FILTER_PIPE); + CASE(IMSG_FILTER_RESPONSE); + default: + return "IMSG_FILTER_???"; + } +} + +static const char * +query_to_str(int query) +{ + switch (query) { + CASE(QUERY_CONNECT); + CASE(QUERY_HELO); + CASE(QUERY_MAIL); + CASE(QUERY_RCPT); + CASE(QUERY_DATA); + CASE(QUERY_EOM); + CASE(QUERY_DATALINE); + default: + return "QUERY_???"; + } +} + +static const char * +event_to_str(int event) +{ + switch (event) { + CASE(EVENT_CONNECT); + CASE(EVENT_RESET); + CASE(EVENT_DISCONNECT); + CASE(EVENT_TX_BEGIN); + CASE(EVENT_TX_COMMIT); + CASE(EVENT_TX_ROLLBACK); + default: + return "EVENT_???"; + } +} + +static const char * +status_to_str(int status) +{ + switch (status) { + CASE(FILTER_OK); + CASE(FILTER_FAIL); + CASE(FILTER_CLOSE); + default: + return "FILTER_???"; + } +} |