daemon: support restricting outgoing IP address

daemon/io.h

@@ -46,6 +46,7 @@ int tcp_bind_tls(uv_tcp_t *handle, struct sockaddr *addr);
 int tcp_bindfd(uv_tcp_t *handle, int fd);
 int tcp_bindfd_tls(uv_tcp_t *handle, int fd);
 
+/** Initialize the handle, incl. ->data = struct session * instance. type = SOCK_* */
 void io_create(uv_loop_t *loop, uv_handle_t *handle, int type);
 void io_deinit(uv_handle_t *handle);
 void io_free(uv_handle_t *handle);

daemon/worker.c

@@ -94,9 +94,17 @@ static inline void req_release(struct worker_ctx *worker, struct req *req)
 	}
 }
 
-/*! @internal Create a UDP/TCP handle */
-static uv_handle_t *ioreq_spawn(struct qr_task *task, int socktype)
+/*! @internal Create a UDP/TCP handle for an outgoing AF_INET* connection.
+ *  socktype is SOCK_* */
+static uv_handle_t *ioreq_spawn(struct qr_task *task, int socktype, sa_family_t family)
 {
+	bool precond = (socktype == SOCK_DGRAM || socktype == SOCK_STREAM)
+			&& (family == AF_INET  || family == AF_INET6);
+	if (!precond) {
+		assert(false);
+		return NULL;
+	}
+
 	if (task->pending_count >= MAX_PENDING) {
 		return NULL;
 	}
@@ -106,10 +114,30 @@ static uv_handle_t *ioreq_spawn(struct qr_task *task, int socktype)
 		return NULL;
 	}
 	io_create(task->worker->loop, handle, socktype);
+
+	/* Bind to outgoing address, according to IP v4/v6. */
+	union inaddr *addr;
+	if (family == AF_INET) {
+		addr = (union inaddr *)&task->worker->out_addr4;
+	} else {
+		addr = (union inaddr *)&task->worker->out_addr6;
+	}
+	int ret = 0;
+	if (addr->ip.sa_family != AF_UNSPEC) {
+		assert(addr->ip.sa_family == family);
+		if (socktype == SOCK_DGRAM) {
+			ret = uv_udp_bind((uv_udp_t *)handle, &addr->ip, 0);
+		} else {
+			ret = uv_tcp_bind((uv_tcp_t *)handle, &addr->ip, 0);
+		}
+	}
+
 	/* Set current handle as a subrequest type. */
 	struct session *session = handle->data;
-	session->outgoing = true;
-	int ret = array_push(session->tasks, task);
+	if (ret == 0) {
+		session->outgoing = true;
+		ret = array_push(session->tasks, task);
+	}
 	if (ret < 0) {
 		io_deinit(handle);
 		req_release(task->worker, (struct req *)handle);
@@ -575,9 +603,9 @@ static void on_timeout(uv_timer_t *req)
 static bool retransmit(struct qr_task *task)
 {
 	if (task && task->addrlist && task->addrlist_count > 0) {
-		uv_handle_t *subreq = ioreq_spawn(task, SOCK_DGRAM);
+		struct sockaddr_in6 *choice = &((struct sockaddr_in6 *)task->addrlist)[task->addrlist_turn];
+		uv_handle_t *subreq = ioreq_spawn(task, SOCK_DGRAM, choice->sin6_family);
 		if (subreq) { /* Create connection for iterative query */
-			struct sockaddr_in6 *choice = &((struct sockaddr_in6 *)task->addrlist)[task->addrlist_turn];
 			if (qr_task_send(task, subreq, (struct sockaddr *)choice, task->pktbuf) == 0) {
 				task->addrlist_turn = (task->addrlist_turn + 1) % task->addrlist_count; /* Round robin */
 				return true;
@@ -772,13 +800,15 @@ static int qr_task_step(struct qr_task *task, const struct sockaddr *packet_sour
 		if (!conn) {
 			return qr_task_step(task, NULL, NULL);
 		}
-		uv_handle_t *client = ioreq_spawn(task, sock_type);
+		const struct sockaddr *addr =
+			packet_source ? packet_source : task->addrlist;
+		uv_handle_t *client = ioreq_spawn(task, sock_type, addr->sa_family);
 		if (!client) {
 			req_release(task->worker, (struct req *)conn);
 			return qr_task_step(task, NULL, NULL);
 		}
 		conn->data = task;
-		if (uv_tcp_connect(conn, (uv_tcp_t *)client, packet_source?packet_source:task->addrlist, on_connect) != 0) {
+		if (uv_tcp_connect(conn, (uv_tcp_t *)client, addr , on_connect) != 0) {
 			req_release(task->worker, (struct req *)conn);
 			return qr_task_step(task, NULL, NULL);
 		}
@@ -1083,6 +1113,8 @@ struct worker_ctx *worker_create(struct engine *engine, knot_mm_t *pool,
 	worker->count = worker_count;
 	worker->engine = engine;
 	worker_reserve(worker, MP_FREELIST_SIZE);
+	worker->out_addr4.sin_family = AF_UNSPEC;
+	worker->out_addr6.sin6_family = AF_UNSPEC;
 	/* Register worker in Lua thread */
 	lua_pushlightuserdata(engine->L, worker);
 	lua_setglobal(engine->L, "__worker");

daemon/worker.h

@@ -78,6 +78,11 @@ struct worker_ctx {
 	int id;
 	int count;
 	unsigned tcp_pipeline_max;
+
+	/** Addresses to bind for outgoing connections or AF_UNSPEC. */
+	struct sockaddr_in out_addr4;
+	struct sockaddr_in6 out_addr6;
+
 #if __linux__
 	uint8_t wire_buf[RECVMMSG_BATCH * KNOT_WIRE_MAX_PKTSIZE];
 #else
@@ -93,6 +98,7 @@ struct worker_ctx {
 		size_t dropped;
 		size_t timeout;
 	} stats;
+
 	map_t outgoing;
 	mp_freelist_t pool_mp;
 	mp_freelist_t pool_ioreq;