commit c3e8d2568710c9e580b7772942d7a8137ceed077
from: Martijn van Duren <martijn@openbsd.org>
date: Tue Jan 28 17:26:56 2025 UTC

If a application local session or message storage fails, send a disconnect
to smtpd, and don't call any of the callback functions.

This is a work in progress and should be considered carefully

commit - 7bc4d7c9959e45c21425a193ef38eb0034b03f95
commit + c3e8d2568710c9e580b7772942d7a8137ceed077
blob - 4d551f0f161ea841d896117f7a9a8521263adc7e
blob + 42ea7886024833958ca16a5344e7c82178f0163a
--- opensmtpd.c
+++ opensmtpd.c
@@ -41,22 +41,29 @@
 
 #define NITEMS(x) (sizeof(x) / sizeof(*x))
 
+enum disconnect {
+	DISCONNECT_FALSE = 0,
+	DISCONNECT_SERVERERROR,
+	DISCONNECT_SEND
+};
+
+struct osmtpd_session {
+	struct osmtpd_ctx ctx;		/* Must remain first element */
+	enum disconnect disconnect;
+	RB_ENTRY(osmtpd_session) entry;
+};
+
 struct osmtpd_callback {
 	enum osmtpd_type type;
 	enum osmtpd_phase phase;
 	int incoming;
-	void (*osmtpd_cb)(struct osmtpd_callback *, struct osmtpd_ctx *, char *,
-	    char *);
+	void (*osmtpd_cb)(struct osmtpd_callback *, struct osmtpd_session *,
+	    char *, char *);
 	void *cb;
 	int doregister;
 	int storereport;
 };
 
-struct osmtpd_session {
-	struct osmtpd_ctx ctx;
-	RB_ENTRY(osmtpd_session) entry;
-};
-
 static void osmtpd_register(enum osmtpd_type, enum osmtpd_phase, int, int,
     void *);
 static const char *osmtpd_typetostr(enum osmtpd_type);
@@ -64,40 +71,40 @@ static const char *osmtpd_phasetostr(enum osmtpd_phase
 static enum osmtpd_phase osmtpd_strtophase(const char *, const char *);
 static void osmtpd_newline(struct io *, int, void *);
 static void osmtpd_outevt(struct io *, int, void *);
-static void osmtpd_noargs(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_noargs(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_onearg(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_onearg(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_connect(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_connect(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_identify(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_identify(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_link_connect(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_link_connect(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
 static void osmtpd_link_disconnect(struct osmtpd_callback *,
-    struct osmtpd_ctx *, char *, char *);
-static void osmtpd_link_greeting(struct osmtpd_callback *, struct osmtpd_ctx *,
+    struct osmtpd_session *, char *, char *);
+static void osmtpd_link_greeting(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_link_identify(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_link_identify(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_link_tls(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_link_tls(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_link_auth(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_link_auth(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_tx_begin(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_tx_begin(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_tx_mail(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_tx_mail(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_tx_rcpt(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_tx_rcpt(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_tx_envelope(struct osmtpd_callback *, struct osmtpd_ctx *,
-    char *, char *);
-static void osmtpd_tx_data(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_tx_envelope(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_tx_commit(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_tx_data(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
-static void osmtpd_tx_rollback(struct osmtpd_callback *, struct osmtpd_ctx *,
+static void osmtpd_tx_commit(struct osmtpd_callback *, struct osmtpd_session *,
     char *, char *);
+static void osmtpd_tx_rollback(struct osmtpd_callback *, struct osmtpd_session *,
+    char *, char *);
 static void osmtpd_addrtoss(char *, struct sockaddr_storage *, int, char *);
 static enum osmtpd_status osmtpd_strtostatus(const char *, char *);
 static int osmtpd_session_cmp(struct osmtpd_session *, struct osmtpd_session *);
@@ -1195,8 +1202,9 @@ osmtpd_newline(struct io *io, int ev, __unused void *a
 			ctx->ctx.local_session = NULL;
 			ctx->ctx.local_message = NULL;
 			if (oncreatecb_session != NULL)
-				ctx->ctx.local_session =
-				    oncreatecb_session(&ctx->ctx);
+				if ((ctx->ctx.local_session =
+				    oncreatecb_session(&ctx->ctx)) == NULL)
+					ctx->disconnect = DISCONNECT_SERVERERROR;
 		}
 		ctx->ctx.type = type;
 		ctx->ctx.phase = phase;
@@ -1224,9 +1232,12 @@ osmtpd_newline(struct io *io, int ev, __unused void *a
 				osmtpd_errx(1, "Invalid line received: invalid "
 				    "token: %s", linedup);
 			line = end + 1;
+			if (ctx->disconnect == DISCONNECT_SERVERERROR)
+				osmtpd_filter_disconnect(&ctx->ctx,
+				    "internal server error");
 		}
 		osmtpd_callbacks[i].osmtpd_cb(&(osmtpd_callbacks[i]),
-		    &(ctx->ctx), line, linedup);
+		    ctx, line, linedup);
 	}
 	io_resume(io_stdout, IO_OUT);
 }
@@ -1245,28 +1256,28 @@ osmtpd_outevt(__unused struct io *io, int evt, __unuse
 }
 
 static void
-osmtpd_noargs(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_noargs(struct osmtpd_callback *cb, struct osmtpd_session *session,
     __unused char *params, __unused char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *);
 
-	f = cb->cb;
-	f(ctx);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx);
 }
 
 static void
-osmtpd_onearg(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx, char *line,
-    __unused char *linedup)
+osmtpd_onearg(struct osmtpd_callback *cb, struct osmtpd_session *session,
+    char *line, __unused char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *, const char *);
 
-	f = cb->cb;
-	f(ctx, line);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, line);
 }
 
 static void
-osmtpd_connect(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx, char *params,
-    char *linedup)
+osmtpd_connect(struct osmtpd_callback *cb, struct osmtpd_session *session,
+    char *params, char *linedup)
 {
 	struct sockaddr_storage ss;
 	char *hostname;
@@ -1281,28 +1292,28 @@ osmtpd_connect(struct osmtpd_callback *cb, struct osmt
 
 	osmtpd_addrtoss(address, &ss, 0, linedup);
 
-	f = cb->cb;
-	f(ctx, hostname, &ss);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, hostname, &ss);
 }
 
 static void
-osmtpd_identify(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_identify(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *identity, __unused char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *, const char *);
 
 	if (cb->storereport) {
-		free(ctx->identity);
-		if ((ctx->identity = strdup(identity)) == NULL)
+		free(session->ctx.identity);
+		if ((session->ctx.identity = strdup(identity)) == NULL)
 			osmtpd_err(1, "strdup");
 	}
 
-	f = cb->cb;
-	f(ctx, identity);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, identity);
 }
 
 static void
-osmtpd_link_connect(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_link_connect(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *params, char *linedup)
 {
 	char *end, *rdns;
@@ -1339,95 +1350,91 @@ osmtpd_link_connect(struct osmtpd_callback *cb, struct
 	params = end;
 	osmtpd_addrtoss(params, &dst, 1, linedup);
 	if (cb->storereport) {
-		if ((ctx->rdns = strdup(rdns)) == NULL)
+		if ((session->ctx.rdns = strdup(rdns)) == NULL)
 			osmtpd_err(1, "strdup");
-		ctx->fcrdns = fcrdns;
-		memcpy(&(ctx->src), &src, sizeof(ctx->src));
-		memcpy(&(ctx->dst), &dst, sizeof(ctx->dst));
+		session->ctx.fcrdns = fcrdns;
+		memcpy(&session->ctx.src, &src, sizeof(session->ctx.src));
+		memcpy(&session->ctx.dst, &dst, sizeof(session->ctx.dst));
 	}
-	if ((f = cb->cb) != NULL)
-		f(ctx, rdns, fcrdns, &src, &dst);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, rdns, fcrdns, &src, &dst);
 }
 
 static void
-osmtpd_link_disconnect(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
-    __unused char *param, __unused char *linedup)
+osmtpd_link_disconnect(struct osmtpd_callback *cb,
+    struct osmtpd_session *session, __unused char *param,
+    __unused char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *);
 	size_t i;
-	struct osmtpd_session *session, search;
 
-	if ((f = cb->cb) != NULL)
-		f(ctx);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx);
 
-	search.ctx.reqid = ctx->reqid;
-	session = RB_FIND(osmtpd_sessions, &osmtpd_sessions, &search);
-	if (session != NULL) {
-		RB_REMOVE(osmtpd_sessions, &osmtpd_sessions, session);
-		if (ondeletecb_session != NULL)
-			ondeletecb_session(ctx, session->ctx.local_session);
-		free(session->ctx.rdns);
-		free(session->ctx.identity);
-		free(session->ctx.greeting.identity);
-		free(session->ctx.ciphers);
-		free(session->ctx.username);
-		free(session->ctx.mailfrom);
-		for (i = 0; session->ctx.rcptto[i] != NULL; i++)
-			free(session->ctx.rcptto[i]);
-		free(session->ctx.rcptto);
-		free(session);
-	}
+	RB_REMOVE(osmtpd_sessions, &osmtpd_sessions, session);
+	if (ondeletecb_session != NULL && session->ctx.local_session != NULL)
+		ondeletecb_session(&session->ctx, session->ctx.local_session);
+	free(session->ctx.rdns);
+	free(session->ctx.identity);
+	free(session->ctx.greeting.identity);
+	free(session->ctx.ciphers);
+	free(session->ctx.username);
+	free(session->ctx.mailfrom);
+	for (i = 0; session->ctx.rcptto[i] != NULL; i++)
+		free(session->ctx.rcptto[i]);
+	free(session->ctx.rcptto);
+	free(session);
 }
 
 static void
-osmtpd_link_greeting(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_link_greeting(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *identity, __unused char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *, const char *);
 
 	if (cb->storereport) {
-		free(ctx->greeting.identity);
-		if ((ctx->greeting.identity = strdup(identity)) == NULL)
+		free(session->ctx.greeting.identity);
+		if ((session->ctx.greeting.identity = strdup(identity)) == NULL)
 			osmtpd_err(1, NULL);
 	}
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, identity);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, identity);
 }
 
 static void
-osmtpd_link_identify(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_link_identify(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *identity, __unused char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *, const char *);
 
 	if (cb->storereport) {
-		free(ctx->identity);
-		if ((ctx->identity = strdup(identity)) == NULL)
+		free(session->ctx.identity);
+		if ((session->ctx.identity = strdup(identity)) == NULL)
 			osmtpd_err(1, NULL);
 	}
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, identity);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, identity);
 }
 
 static void
-osmtpd_link_tls(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_link_tls(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *ciphers, __unused char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *, const char *);
 
 	if (cb->storereport) {
-		if ((ctx->ciphers = strdup(ciphers)) == NULL)
+		if ((session->ctx.ciphers = strdup(ciphers)) == NULL)
 			osmtpd_err(1, NULL);
 	}
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, ciphers);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, ciphers);
 }
 
 static void
-osmtpd_link_auth(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_link_auth(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *username, char *linedup)
 {
 	void (*f)(struct osmtpd_ctx *, const char *, enum osmtpd_auth_status);
@@ -1448,16 +1455,16 @@ osmtpd_link_auth(struct osmtpd_callback *cb, struct os
 		osmtpd_errx(1, "Invalid auth status received: %s", linedup);
 
 	if (cb->storereport && s == OSMTPD_AUTH_PASS) {
-		if ((ctx->username = strdup(username)) == NULL)
+		if ((session->ctx.username = strdup(username)) == NULL)
 			osmtpd_err(1, NULL);
 	}
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, username, s);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, username, s);
 }
 
 static void
-osmtpd_tx_begin(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_tx_begin(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *msgid, char *linedup)
 {
 	unsigned long imsgid;
@@ -1469,24 +1476,27 @@ osmtpd_tx_begin(struct osmtpd_callback *cb, struct osm
 	if ((imsgid == ULONG_MAX && errno != 0) || endptr[0] != '\0')
 		osmtpd_errx(1, "Invalid line received: invalid msgid: %s",
 		    linedup);
-	ctx->msgid = imsgid;
+	session->ctx.msgid = imsgid;
 	/* Check if we're in range */
-	if ((unsigned long) ctx->msgid != imsgid)
+	if ((unsigned long) session->ctx.msgid != imsgid)
 		osmtpd_errx(1, "Invalid line received: invalid msgid: %s",
 		    linedup);
 
 	if (!cb->storereport)
-		ctx->msgid = 0;
+		session->ctx.msgid = 0;
 
-	if (oncreatecb_message != NULL)
-		ctx->local_message = oncreatecb_message(ctx);
+	if (oncreatecb_message != NULL) {
+		session->ctx.local_message = oncreatecb_message(&session->ctx);
+		if (session->ctx.local_message == NULL)
+			session->disconnect = DISCONNECT_SERVERERROR;
+	}
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, imsgid);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, imsgid);
 }
 
 static void
-osmtpd_tx_mail(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_tx_mail(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *params, char *linedup)
 {
 	char *end, *mailfrom;
@@ -1514,7 +1524,7 @@ osmtpd_tx_mail(struct osmtpd_callback *cb, struct osmt
 		osmtpd_errx(1, "Invalid line received: missing status: %s",
 		    linedup);
 	end++[0] = '\0';
-	if (ctx->version_major == 0 && ctx->version_minor < 6) {
+	if (session->ctx.version_major == 0 && session->ctx.version_minor < 6) {
 		mailfrom = params;
 		status = osmtpd_strtostatus(end, linedup);
 	} else {
@@ -1522,16 +1532,16 @@ osmtpd_tx_mail(struct osmtpd_callback *cb, struct osmt
 		status = osmtpd_strtostatus(params, linedup);
 	}
 	if (cb->storereport) {
-		if ((ctx->mailfrom = strdup(mailfrom)) == NULL)
+		if ((session->ctx.mailfrom = strdup(mailfrom)) == NULL)
 			osmtpd_err(1, NULL);
 	}
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, msgid, mailfrom, status);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, msgid, mailfrom, status);
 }
 
 static void
-osmtpd_tx_rcpt(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_tx_rcpt(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *params, char *linedup)
 {
 	char *end, *rcptto;
@@ -1561,7 +1571,7 @@ osmtpd_tx_rcpt(struct osmtpd_callback *cb, struct osmt
 		    linedup);
 	end++[0] = '\0';
 
-	if (ctx->version_major == 0 && ctx->version_minor < 6) {
+	if (session->ctx.version_major == 0 && session->ctx.version_minor < 6) {
 		rcptto = params;
 		status = osmtpd_strtostatus(end, linedup);
 	} else {
@@ -1570,24 +1580,24 @@ osmtpd_tx_rcpt(struct osmtpd_callback *cb, struct osmt
 	}
 
 	if (cb->storereport) {
-		for (i = 0; ctx->rcptto[i] != NULL; i++)
+		for (i = 0; session->ctx.rcptto[i] != NULL; i++)
 			;
-		ctx->rcptto = reallocarray(ctx->rcptto, i + 2,
-		    sizeof(*(ctx->rcptto)));
-		if (ctx->rcptto == NULL)
+		session->ctx.rcptto = reallocarray(session->ctx.rcptto, i + 2,
+		    sizeof(*session->ctx.rcptto));
+		if (session->ctx.rcptto == NULL)
 			osmtpd_err(1, NULL);
 
-		if ((ctx->rcptto[i] = strdup(rcptto)) == NULL)
+		if ((session->ctx.rcptto[i] = strdup(rcptto)) == NULL)
 			osmtpd_err(1, NULL);
-		ctx->rcptto[i + 1] = NULL;
+		session->ctx.rcptto[i + 1] = NULL;
 	}
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, msgid, rcptto, status);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, msgid, rcptto, status);
 }
 
 static void
-osmtpd_tx_envelope(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_tx_envelope(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *params, char *linedup)
 {
 	unsigned long imsgid;
@@ -1611,19 +1621,18 @@ osmtpd_tx_envelope(struct osmtpd_callback *cb, struct 
 	params = end + 1;
 
 	evpid = strtoull(params, &end, 16);
-	if ((ctx->evpid == ULLONG_MAX && errno != 0) ||
-	    end[0] != '\0')
+	if ((session->ctx.evpid == ULLONG_MAX && errno != 0) || end[0] != '\0')
 		osmtpd_errx(1, "Invalid line received: invalid evpid: %s",
 		    linedup);
 	if (cb->storereport)
-		ctx->evpid = evpid;
+		session->ctx.evpid = evpid;
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, msgid, evpid);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, msgid, evpid);
 }
 
 static void
-osmtpd_tx_data(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_tx_data(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *params, char *linedup)
 {
 	char *end;
@@ -1645,12 +1654,12 @@ osmtpd_tx_data(struct osmtpd_callback *cb, struct osmt
 		    linedup);
 	params = end + 1;
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, msgid, osmtpd_strtostatus(params, linedup));
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, msgid, osmtpd_strtostatus(params, linedup));
 }
 
 static void
-osmtpd_tx_commit(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_tx_commit(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *params, char *linedup)
 {
 	char *end;
@@ -1679,26 +1688,26 @@ osmtpd_tx_commit(struct osmtpd_callback *cb, struct os
 		osmtpd_errx(1, "Invalid line received: invalid msg size: %s",
 		    linedup);
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, msgid, msgsz);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, msgid, msgsz);
 
-	if (ondeletecb_message != NULL) {
-		ondeletecb_message(ctx, ctx->local_message);
-		ctx->local_message = NULL;
+	if (ondeletecb_message != NULL && session->ctx.local_message != NULL) {
+		ondeletecb_message(&session->ctx, session->ctx.local_message);
+		session->ctx.local_message = NULL;
 	}
 
-	free(ctx->mailfrom);
-	ctx->mailfrom = NULL;
+	free(session->ctx.mailfrom);
+	session->ctx.mailfrom = NULL;
 
-	for (i = 0; ctx->rcptto[i] != NULL; i++)
-		free(ctx->rcptto[i]);
-	ctx->rcptto[0] = NULL;
-	ctx->evpid = 0;
-	ctx->msgid = 0;
+	for (i = 0; session->ctx.rcptto[i] != NULL; i++)
+		free(session->ctx.rcptto[i]);
+	session->ctx.rcptto[0] = NULL;
+	session->ctx.evpid = 0;
+	session->ctx.msgid = 0;
 }
 
 static void
-osmtpd_tx_rollback(struct osmtpd_callback *cb, struct osmtpd_ctx *ctx,
+osmtpd_tx_rollback(struct osmtpd_callback *cb, struct osmtpd_session *session,
     char *params, char *linedup)
 {
 	char *end;
@@ -1720,27 +1729,32 @@ osmtpd_tx_rollback(struct osmtpd_callback *cb, struct 
 		osmtpd_errx(1, "Invalid line received: invalid msgid: %s",
 		    linedup);
 
-	if ((f = cb->cb) != NULL)
-		f(ctx, msgid);
+	if ((f = cb->cb) != NULL && session->disconnect == DISCONNECT_FALSE)
+		f(&session->ctx, msgid);
 
-	if (ondeletecb_message != NULL) {
-		ondeletecb_message(ctx, ctx->local_message);
-		ctx->local_message = NULL;
+	if (ondeletecb_message != NULL && session->ctx.local_message != NULL) {
+		ondeletecb_message(&session->ctx, session->ctx.local_message);
+		session->ctx.local_message = NULL;
 	}
 
-	free(ctx->mailfrom);
-	ctx->mailfrom = NULL;
+	free(session->ctx.mailfrom);
+	session->ctx.mailfrom = NULL;
 
-	for (i = 0; ctx->rcptto[i] != NULL; i++)
-		free(ctx->rcptto[i]);
-	ctx->rcptto[0] = NULL;
-	ctx->evpid = 0;
-	ctx->msgid = 0;
+	for (i = 0; session->ctx.rcptto[i] != NULL; i++)
+		free(session->ctx.rcptto[i]);
+	session->ctx.rcptto[0] = NULL;
+	session->ctx.evpid = 0;
+	session->ctx.msgid = 0;
 }
 
 void
 osmtpd_filter_proceed(struct osmtpd_ctx *ctx)
 {
+	struct osmtpd_session *session = (struct osmtpd_session *)ctx;
+
+	if (session->disconnect == DISCONNECT_SEND)
+		return;
+
 	if (ctx->version_major == 0 && ctx->version_minor < 5)
 		io_printf(io_stdout, "filter-result|%016"PRIx64"|%016"PRIx64"|"
 		    "proceed\n", ctx->token, ctx->reqid);
@@ -1752,8 +1766,12 @@ osmtpd_filter_proceed(struct osmtpd_ctx *ctx)
 void
 osmtpd_filter_reject(struct osmtpd_ctx *ctx, int code, const char *reason, ...)
 {
+	struct osmtpd_session *session = (struct osmtpd_session *)ctx;
 	va_list ap;
 
+	if (session->disconnect == DISCONNECT_SEND)
+		return;
+
 	if (code < 200 || code > 599)
 		osmtpd_errx(1, "Invalid reject code");
 
@@ -1773,8 +1791,12 @@ void
 osmtpd_filter_reject_enh(struct osmtpd_ctx *ctx, int code, int class,
     int subject, int detail, const char *reason, ...)
 {
+	struct osmtpd_session *session = (struct osmtpd_session *)ctx;
 	va_list ap;
 
+	if (session->disconnect == DISCONNECT_SEND)
+		return;
+
 	if (code < 200 || code > 599)
 		osmtpd_errx(1, "Invalid reject code");
 	if (class < 2 || class > 5)
@@ -1801,8 +1823,12 @@ osmtpd_filter_reject_enh(struct osmtpd_ctx *ctx, int c
 void
 osmtpd_filter_disconnect(struct osmtpd_ctx *ctx, const char *reason, ...)
 {
+	struct osmtpd_session *session = (struct osmtpd_session *)ctx;
 	va_list ap;
 
+	if (session->disconnect == DISCONNECT_SEND)
+		return;
+
 	if (ctx->version_major == 0 && ctx->version_minor < 5)
 		io_printf(io_stdout, "filter-result|%016"PRIx64"|%016"PRIx64"|"
 		    "disconnect|421 ", ctx->token, ctx->reqid);
@@ -1813,14 +1839,19 @@ osmtpd_filter_disconnect(struct osmtpd_ctx *ctx, const
 	io_vprintf(io_stdout, reason, ap);
 	va_end(ap);
 	io_printf(io_stdout, "\n");
+	session->disconnect = DISCONNECT_SEND;
 }
 
 void
 osmtpd_filter_disconnect_enh(struct osmtpd_ctx *ctx, int class, int subject,
     int detail, const char *reason, ...)
 {
+	struct osmtpd_session *session = (struct osmtpd_session *)ctx;
 	va_list ap;
 
+	if (session->disconnect == DISCONNECT_SEND)
+		return;
+
 	if (class <= 2 || class >= 5)
 		osmtpd_errx(1, "Invalid enhanced status class");
 	if (subject < 0 || subject > 999)
@@ -1839,12 +1870,17 @@ osmtpd_filter_disconnect_enh(struct osmtpd_ctx *ctx, i
 	io_vprintf(io_stdout, reason, ap);
 	va_end(ap);
 	io_printf(io_stdout, "\n");
+	session->disconnect = DISCONNECT_SEND;
 }
 
 void
 osmtpd_filter_rewrite(struct osmtpd_ctx *ctx, const char *value, ...)
 {
+	struct osmtpd_session *session = (struct osmtpd_session *)ctx;
 	va_list ap;
+
+	if (session->disconnect == DISCONNECT_SEND)
+		return;
 
 	if (ctx->version_major == 0 && ctx->version_minor < 5)
 		io_printf(io_stdout, "filter-result|%016"PRIx64"|%016"PRIx64"|"
@@ -1861,7 +1897,11 @@ osmtpd_filter_rewrite(struct osmtpd_ctx *ctx, const ch
 void
 osmtpd_filter_dataline(struct osmtpd_ctx *ctx, const char *line, ...)
 {
+	struct osmtpd_session *session = (struct osmtpd_session *)ctx;
 	va_list ap;
+
+	if (session->disconnect == DISCONNECT_SEND)
+		return;
 
 	if (ctx->version_major == 0 && ctx->version_minor < 5)
 		io_printf(io_stdout, "filter-dataline|%016"PRIx64"|%016"PRIx64"|",