[thirdparty/systemd.git] / src / resolve / resolved-dns-stream.c

/*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/

/***
  This file is part of systemd.

  Copyright 2014 Lennart Poettering

  systemd is free software; you can redistribute it and/or modify it
  under the terms of the GNU Lesser General Public License as published by
  the Free Software Foundation; either version 2.1 of the License, or
  (at your option) any later version.

  systemd is distributed in the hope that it will be useful, but
  WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  Lesser General Public License for more details.

  You should have received a copy of the GNU Lesser General Public License
  along with systemd; If not, see <http://www.gnu.org/licenses/>.
***/

#include <netinet/tcp.h>

#include "missing.h"
#include "resolved-dns-stream.h"

#define DNS_STREAM_TIMEOUT_USEC (10 * USEC_PER_SEC)
#define DNS_STREAMS_MAX 128

static void dns_stream_stop(DnsStream *s) {
        assert(s);

        s->io_event_source = sd_event_source_unref(s->io_event_source);
        s->timeout_event_source = sd_event_source_unref(s->timeout_event_source);
        s->fd = safe_close(s->fd);
}

static int dns_stream_update_io(DnsStream *s) {
        int f = 0;

        assert(s);

        if (s->write_packet && s->n_written < sizeof(s->write_size) + s->write_packet->size)
                f |= EPOLLOUT;
        if (!s->read_packet || s->n_read < sizeof(s->read_size) + s->read_packet->size)
                f |= EPOLLIN;

        return sd_event_source_set_io_events(s->io_event_source, f);
}

static int dns_stream_complete(DnsStream *s, int error) {
        assert(s);

        dns_stream_stop(s);

        if (s->complete)
                s->complete(s, error);
        else
                dns_stream_free(s);

        return 0;
}

static int dns_stream_identify(DnsStream *s) {
        union {
                struct cmsghdr header; /* For alignment */
                uint8_t buffer[CMSG_SPACE(MAXSIZE(struct in_pktinfo, struct in6_pktinfo))
                               + EXTRA_CMSG_SPACE /* kernel appears to require extra space */];
        } control;
        struct msghdr mh = {};
        struct cmsghdr *cmsg;
        socklen_t sl;
        int r;

        assert(s);

        if (s->identified)
                return 0;

        /* Query the local side */
        s->local_salen = sizeof(s->local);
        r = getsockname(s->fd, &s->local.sa, &s->local_salen);
        if (r < 0)
                return -errno;
        if (s->local.sa.sa_family == AF_INET6 && s->ifindex <= 0)
                s->ifindex = s->local.in6.sin6_scope_id;

        /* Query the remote side */
        s->peer_salen = sizeof(s->peer);
        r = getpeername(s->fd, &s->peer.sa, &s->peer_salen);
        if (r < 0)
                return -errno;
        if (s->peer.sa.sa_family == AF_INET6 && s->ifindex <= 0)
                s->ifindex = s->peer.in6.sin6_scope_id;

        /* Check consistency */
        assert(s->peer.sa.sa_family == s->local.sa.sa_family);
        assert(IN_SET(s->peer.sa.sa_family, AF_INET, AF_INET6));

        /* Query connection meta information */
        sl = sizeof(control);
        if (s->peer.sa.sa_family == AF_INET) {
                r = getsockopt(s->fd, IPPROTO_IP, IP_PKTOPTIONS, &control, &sl);
                if (r < 0)
                        return -errno;
        } else if (s->peer.sa.sa_family == AF_INET6) {

                r = getsockopt(s->fd, IPPROTO_IPV6, IPV6_2292PKTOPTIONS, &control, &sl);
                if (r < 0)
                        return -errno;
        } else
                return -EAFNOSUPPORT;

        mh.msg_control = &control;
        mh.msg_controllen = sl;

        CMSG_FOREACH(cmsg, &mh) {

                if (cmsg->cmsg_level == IPPROTO_IPV6) {
                        assert(s->peer.sa.sa_family == AF_INET6);

                        switch (cmsg->cmsg_type) {

                        case IPV6_PKTINFO: {
                                struct in6_pktinfo *i = (struct in6_pktinfo*) CMSG_DATA(cmsg);

                                if (s->ifindex <= 0)
                                        s->ifindex = i->ipi6_ifindex;
                                break;
                        }

                        case IPV6_HOPLIMIT:
                                s->ttl = *(int *) CMSG_DATA(cmsg);
                                break;
                        }

                } else if (cmsg->cmsg_level == IPPROTO_IP) {
                        assert(s->peer.sa.sa_family == AF_INET);

                        switch (cmsg->cmsg_type) {

                        case IP_PKTINFO: {
                                struct in_pktinfo *i = (struct in_pktinfo*) CMSG_DATA(cmsg);

                                if (s->ifindex <= 0)
                                        s->ifindex = i->ipi_ifindex;
                                break;
                        }

                        case IP_TTL:
                                s->ttl = *(int *) CMSG_DATA(cmsg);
                                break;
                        }
                }
        }

        /* The Linux kernel sets the interface index to the loopback
         * device if the connection came from the local host since it
         * avoids the routing table in such a case. Let's unset the
         * interface index in such a case. */
        if (s->ifindex == LOOPBACK_IFINDEX)
                s->ifindex = 0;

        /* If we don't know the interface index still, we look for the
         * first local interface with a matching address. Yuck! */
        if (s->ifindex <= 0)
                s->ifindex = manager_find_ifindex(s->manager, s->local.sa.sa_family, s->local.sa.sa_family == AF_INET ? (union in_addr_union*) &s->local.in.sin_addr : (union in_addr_union*)  &s->local.in6.sin6_addr);

        if (s->protocol == DNS_PROTOCOL_LLMNR && s->ifindex > 0) {
                uint32_t ifindex = htobe32(s->ifindex);

                /* Make sure all packets for this connection are sent on the same interface */
                if (s->local.sa.sa_family == AF_INET) {
                        r = setsockopt(s->fd, IPPROTO_IP, IP_UNICAST_IF, &ifindex, sizeof(ifindex));
                        if (r < 0)
                                log_debug_errno(errno, "Failed to invoke IP_UNICAST_IF: %m");
                } else if (s->local.sa.sa_family == AF_INET6) {
                        r = setsockopt(s->fd, IPPROTO_IPV6, IPV6_UNICAST_IF, &ifindex, sizeof(ifindex));
                        if (r < 0)
                                log_debug_errno(errno, "Failed to invoke IPV6_UNICAST_IF: %m");
                }
        }

        s->identified = true;

        return 0;
}

static int on_stream_timeout(sd_event_source *es, usec_t usec, void *userdata) {
        DnsStream *s = userdata;

        assert(s);

        return dns_stream_complete(s, ETIMEDOUT);
}

static int on_stream_io(sd_event_source *es, int fd, uint32_t revents, void *userdata) {
        DnsStream *s = userdata;
        int r;

        assert(s);

        r = dns_stream_identify(s);
        if (r < 0)
                return dns_stream_complete(s, -r);

        if ((revents & EPOLLOUT) &&
            s->write_packet &&
            s->n_written < sizeof(s->write_size) + s->write_packet->size) {

                struct iovec iov[2];
                ssize_t ss;

                iov[0].iov_base = &s->write_size;
                iov[0].iov_len = sizeof(s->write_size);
                iov[1].iov_base = DNS_PACKET_DATA(s->write_packet);
                iov[1].iov_len = s->write_packet->size;

                IOVEC_INCREMENT(iov, 2, s->n_written);

                ss = writev(fd, iov, 2);
                if (ss < 0) {
                        if (errno != EINTR && errno != EAGAIN)
                                return dns_stream_complete(s, errno);
                } else
                        s->n_written += ss;

                /* Are we done? If so, disable the event source for EPOLLOUT */
                if (s->n_written >= sizeof(s->write_size) + s->write_packet->size) {
                        r = dns_stream_update_io(s);
                        if (r < 0)
                                return dns_stream_complete(s, -r);
                }
        }

        if ((revents & (EPOLLIN|EPOLLHUP|EPOLLRDHUP)) &&
            (!s->read_packet ||
             s->n_read < sizeof(s->read_size) + s->read_packet->size)) {

                if (s->n_read < sizeof(s->read_size)) {
                        ssize_t ss;

                        ss = read(fd, (uint8_t*) &s->read_size + s->n_read, sizeof(s->read_size) - s->n_read);
                        if (ss < 0) {
                                if (errno != EINTR && errno != EAGAIN)
                                        return dns_stream_complete(s, errno);
                        } else if (ss == 0)
                                return dns_stream_complete(s, ECONNRESET);
                        else
                                s->n_read += ss;
                }

                if (s->n_read >= sizeof(s->read_size)) {

                        if (be16toh(s->read_size) < DNS_PACKET_HEADER_SIZE)
                                return dns_stream_complete(s, EBADMSG);

                        if (s->n_read < sizeof(s->read_size) + be16toh(s->read_size)) {
                                ssize_t ss;

                                if (!s->read_packet) {
                                        r = dns_packet_new(&s->read_packet, s->protocol, be16toh(s->read_size));
                                        if (r < 0)
                                                return dns_stream_complete(s, -r);

                                        s->read_packet->size = be16toh(s->read_size);
                                        s->read_packet->ipproto = IPPROTO_TCP;
                                        s->read_packet->family = s->peer.sa.sa_family;
                                        s->read_packet->ttl = s->ttl;
                                        s->read_packet->ifindex = s->ifindex;

                                        if (s->read_packet->family == AF_INET) {
                                                s->read_packet->sender.in = s->peer.in.sin_addr;
                                                s->read_packet->sender_port = be16toh(s->peer.in.sin_port);
                                                s->read_packet->destination.in = s->local.in.sin_addr;
                                                s->read_packet->destination_port = be16toh(s->local.in.sin_port);
                                        } else {
                                                assert(s->read_packet->family == AF_INET6);
                                                s->read_packet->sender.in6 = s->peer.in6.sin6_addr;
                                                s->read_packet->sender_port = be16toh(s->peer.in6.sin6_port);
                                                s->read_packet->destination.in6 = s->local.in6.sin6_addr;
                                                s->read_packet->destination_port = be16toh(s->local.in6.sin6_port);

                                                if (s->read_packet->ifindex == 0)
                                                        s->read_packet->ifindex = s->peer.in6.sin6_scope_id;
                                                if (s->read_packet->ifindex == 0)
                                                        s->read_packet->ifindex = s->local.in6.sin6_scope_id;
                                        }
                                }

                                ss = read(fd,
                                          (uint8_t*) DNS_PACKET_DATA(s->read_packet) + s->n_read - sizeof(s->read_size),
                                          sizeof(s->read_size) + be16toh(s->read_size) - s->n_read);
                                if (ss < 0) {
                                        if (errno != EINTR && errno != EAGAIN)
                                                return dns_stream_complete(s, errno);
                                } else if (ss == 0)
                                        return dns_stream_complete(s, ECONNRESET);
                                else
                                        s->n_read += ss;
                        }

                        /* Are we done? If so, disable the event source for EPOLLIN */
                        if (s->n_read >= sizeof(s->read_size) + be16toh(s->read_size)) {
                                r = dns_stream_update_io(s);
                                if (r < 0)
                                        return dns_stream_complete(s, -r);

                                /* If there's a packet handler
                                 * installed, call that. Note that
                                 * this is optional... */
                                if (s->on_packet)
                                        return s->on_packet(s);
                        }
                }
        }

        if ((s->write_packet && s->n_written >= sizeof(s->write_size) + s->write_packet->size) &&
            (s->read_packet && s->n_read >= sizeof(s->read_size) + s->read_packet->size))
                return dns_stream_complete(s, 0);

        return 0;
}

DnsStream *dns_stream_free(DnsStream *s) {
        if (!s)
                return NULL;

        dns_stream_stop(s);

        if (s->manager) {
                LIST_REMOVE(streams, s->manager->dns_streams, s);
                s->manager->n_dns_streams--;
        }

        dns_packet_unref(s->write_packet);
        dns_packet_unref(s->read_packet);

        free(s);

        return 0;
}

DEFINE_TRIVIAL_CLEANUP_FUNC(DnsStream*, dns_stream_free);

int dns_stream_new(Manager *m, DnsStream **ret, DnsProtocol protocol, int fd) {
        static const int one = 1;
        _cleanup_(dns_stream_freep) DnsStream *s = NULL;
        int r;

        assert(m);
        assert(fd >= 0);

        if (m->n_dns_streams > DNS_STREAMS_MAX)
                return -EBUSY;

        s = new0(DnsStream, 1);
        if (!s)
                return -ENOMEM;

        s->fd = -1;
        s->protocol = protocol;

        r = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
        if (r < 0)
                return -errno;

        r = sd_event_add_io(m->event, &s->io_event_source, fd, EPOLLIN, on_stream_io, s);
        if (r < 0)
                return r;

        r = sd_event_add_time(
                        m->event,
                        &s->timeout_event_source,
                        clock_boottime_or_monotonic(),
                        now(clock_boottime_or_monotonic()) + DNS_STREAM_TIMEOUT_USEC, 0,
                        on_stream_timeout, s);
        if (r < 0)
                return r;

        LIST_PREPEND(streams, m->dns_streams, s);
        s->manager = m;
        s->fd = fd;
        m->n_dns_streams++;

        *ret = s;
        s = NULL;

        return 0;
}

int dns_stream_write_packet(DnsStream *s, DnsPacket *p) {
        assert(s);

        if (s->write_packet)
                return -EBUSY;

        s->write_packet = dns_packet_ref(p);
        s->write_size = htobe16(p->size);
        s->n_written = 0;

        return dns_stream_update_io(s);
}
Commit	Line	Data
623a4c97 LP	1	/-- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil --/
	2
	3	/***
	4	This file is part of systemd.
	5
	6	Copyright 2014 Lennart Poettering
	7
	8	systemd is free software; you can redistribute it and/or modify it
	9	under the terms of the GNU Lesser General Public License as published by
	10	the Free Software Foundation; either version 2.1 of the License, or
	11	(at your option) any later version.
	12
	13	systemd is distributed in the hope that it will be useful, but
	14	WITHOUT ANY WARRANTY; without even the implied warranty of
	15	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	16	Lesser General Public License for more details.
	17
	18	You should have received a copy of the GNU Lesser General Public License
	19	along with systemd; If not, see <http://www.gnu.org/licenses/>.
	20	***/
	21
	22	#include <netinet/tcp.h>
	23
	24	#include "missing.h"
	25	#include "resolved-dns-stream.h"
	26
	27	#define DNS_STREAM_TIMEOUT_USEC (10 * USEC_PER_SEC)
	28	#define DNS_STREAMS_MAX 128
	29
	30	static void dns_stream_stop(DnsStream *s) {
	31	assert(s);
	32
	33	s->io_event_source = sd_event_source_unref(s->io_event_source);
	34	s->timeout_event_source = sd_event_source_unref(s->timeout_event_source);
	35	s->fd = safe_close(s->fd);
	36	}
	37
	38	static int dns_stream_update_io(DnsStream *s) {
	39	int f = 0;
	40
	41	assert(s);
	42
	43	if (s->write_packet && s->n_written < sizeof(s->write_size) + s->write_packet->size)
	44	f \|= EPOLLOUT;
	45	if (!s->read_packet \|\| s->n_read < sizeof(s->read_size) + s->read_packet->size)
	46	f \|= EPOLLIN;
	47
	48	return sd_event_source_set_io_events(s->io_event_source, f);
	49	}
	50
b914e211	51	static int dns_stream_complete(DnsStream *s, int error) {
623a4c97 LP	52	assert(s);
	53
	54	dns_stream_stop(s);
	55
	56	if (s->complete)
	57	s->complete(s, error);
	58	else
	59	dns_stream_free(s);
	60
	61	return 0;
	62	}
	63
b914e211 LP	64	static int dns_stream_identify(DnsStream *s) {
	65	union {
	66	struct cmsghdr header; /* For alignment */
40a1eebd	67	uint8_t buffer[CMSG_SPACE(MAXSIZE(struct in_pktinfo, struct in6_pktinfo))
b914e211 LP	68	+ EXTRA_CMSG_SPACE /* kernel appears to require extra space */];
	69	} control;
	70	struct msghdr mh = {};
	71	struct cmsghdr *cmsg;
	72	socklen_t sl;
	73	int r;
	74
	75	assert(s);
	76
	77	if (s->identified)
	78	return 0;
	79
	80	/* Query the local side */
	81	s->local_salen = sizeof(s->local);
	82	r = getsockname(s->fd, &s->local.sa, &s->local_salen);
	83	if (r < 0)
	84	return -errno;
	85	if (s->local.sa.sa_family == AF_INET6 && s->ifindex <= 0)
	86	s->ifindex = s->local.in6.sin6_scope_id;
	87
	88	/* Query the remote side */
	89	s->peer_salen = sizeof(s->peer);
	90	r = getpeername(s->fd, &s->peer.sa, &s->peer_salen);
	91	if (r < 0)
	92	return -errno;
	93	if (s->peer.sa.sa_family == AF_INET6 && s->ifindex <= 0)
	94	s->ifindex = s->peer.in6.sin6_scope_id;
	95
	96	/* Check consistency */
	97	assert(s->peer.sa.sa_family == s->local.sa.sa_family);
	98	assert(IN_SET(s->peer.sa.sa_family, AF_INET, AF_INET6));
	99
	100	/* Query connection meta information */
	101	sl = sizeof(control);
	102	if (s->peer.sa.sa_family == AF_INET) {
	103	r = getsockopt(s->fd, IPPROTO_IP, IP_PKTOPTIONS, &control, &sl);
	104	if (r < 0)
	105	return -errno;
	106	} else if (s->peer.sa.sa_family == AF_INET6) {
	107
	108	r = getsockopt(s->fd, IPPROTO_IPV6, IPV6_2292PKTOPTIONS, &control, &sl);
	109	if (r < 0)
	110	return -errno;
	111	} else
	112	return -EAFNOSUPPORT;
	113
	114	mh.msg_control = &control;
	115	mh.msg_controllen = sl;
2a1288ff LP	116
2a1288ff LP	117	CMSG_FOREACH(cmsg, &mh) {
b914e211 LP	118
	119	if (cmsg->cmsg_level == IPPROTO_IPV6) {
	120	assert(s->peer.sa.sa_family == AF_INET6);
	121
	122	switch (cmsg->cmsg_type) {
	123
	124	case IPV6_PKTINFO: {
	125	struct in6_pktinfo i = (struct in6_pktinfo) CMSG_DATA(cmsg);
	126
	127	if (s->ifindex <= 0)
	128	s->ifindex = i->ipi6_ifindex;
	129	break;
	130	}
	131
	132	case IPV6_HOPLIMIT:
	133	s->ttl = (int ) CMSG_DATA(cmsg);
	134	break;
	135	}
	136
	137	} else if (cmsg->cmsg_level == IPPROTO_IP) {
	138	assert(s->peer.sa.sa_family == AF_INET);
	139
	140	switch (cmsg->cmsg_type) {
	141
	142	case IP_PKTINFO: {
	143	struct in_pktinfo i = (struct in_pktinfo) CMSG_DATA(cmsg);
	144
	145	if (s->ifindex <= 0)
	146	s->ifindex = i->ipi_ifindex;
	147	break;
	148	}
	149
	150	case IP_TTL:
	151	s->ttl = (int ) CMSG_DATA(cmsg);
	152	break;
	153	}
	154	}
	155	}
	156
	157	/* The Linux kernel sets the interface index to the loopback
	158	* device if the connection came from the local host since it
	159	* avoids the routing table in such a case. Let's unset the
	160	* interface index in such a case. */
a5f03596	161	if (s->ifindex == LOOPBACK_IFINDEX)
b914e211 LP	162	s->ifindex = 0;
	163
	164	/* If we don't know the interface index still, we look for the
	165	* first local interface with a matching address. Yuck! */
	166	if (s->ifindex <= 0)
	167	s->ifindex = manager_find_ifindex(s->manager, s->local.sa.sa_family, s->local.sa.sa_family == AF_INET ? (union in_addr_union) &s->local.in.sin_addr : (union in_addr_union) &s->local.in6.sin6_addr);
	168
	169	if (s->protocol == DNS_PROTOCOL_LLMNR && s->ifindex > 0) {
	170	uint32_t ifindex = htobe32(s->ifindex);
	171
	172	/* Make sure all packets for this connection are sent on the same interface */
	173	if (s->local.sa.sa_family == AF_INET) {
	174	r = setsockopt(s->fd, IPPROTO_IP, IP_UNICAST_IF, &ifindex, sizeof(ifindex));
	175	if (r < 0)
56f64d95	176	log_debug_errno(errno, "Failed to invoke IP_UNICAST_IF: %m");
b914e211 LP	177	} else if (s->local.sa.sa_family == AF_INET6) {
	178	r = setsockopt(s->fd, IPPROTO_IPV6, IPV6_UNICAST_IF, &ifindex, sizeof(ifindex));
	179	if (r < 0)
56f64d95	180	log_debug_errno(errno, "Failed to invoke IPV6_UNICAST_IF: %m");
b914e211 LP	181	}
	182	}
	183
	184	s->identified = true;
	185
	186	return 0;
	187	}
	188
623a4c97 LP	189	static int on_stream_timeout(sd_event_source es, usec_t usec, void userdata) {
	190	DnsStream *s = userdata;
	191
	192	assert(s);
	193
b914e211	194	return dns_stream_complete(s, ETIMEDOUT);
623a4c97 LP	195	}
	196
	197	static int on_stream_io(sd_event_source es, int fd, uint32_t revents, void userdata) {
	198	DnsStream *s = userdata;
	199	int r;
	200
	201	assert(s);
	202
b914e211 LP	203	r = dns_stream_identify(s);
	204	if (r < 0)
	205	return dns_stream_complete(s, -r);
	206
623a4c97 LP	207	if ((revents & EPOLLOUT) &&
	208	s->write_packet &&
	209	s->n_written < sizeof(s->write_size) + s->write_packet->size) {
	210
	211	struct iovec iov[2];
	212	ssize_t ss;
	213
	214	iov[0].iov_base = &s->write_size;
	215	iov[0].iov_len = sizeof(s->write_size);
	216	iov[1].iov_base = DNS_PACKET_DATA(s->write_packet);
	217	iov[1].iov_len = s->write_packet->size;
	218
	219	IOVEC_INCREMENT(iov, 2, s->n_written);
	220
	221	ss = writev(fd, iov, 2);
	222	if (ss < 0) {
	223	if (errno != EINTR && errno != EAGAIN)
b914e211	224	return dns_stream_complete(s, errno);
623a4c97 LP	225	} else
	226	s->n_written += ss;
	227
	228	/* Are we done? If so, disable the event source for EPOLLOUT */
	229	if (s->n_written >= sizeof(s->write_size) + s->write_packet->size) {
	230	r = dns_stream_update_io(s);
	231	if (r < 0)
b914e211	232	return dns_stream_complete(s, -r);
623a4c97 LP	233	}
	234	}
	235
	236	if ((revents & (EPOLLIN\|EPOLLHUP\|EPOLLRDHUP)) &&
	237	(!s->read_packet \|\|
	238	s->n_read < sizeof(s->read_size) + s->read_packet->size)) {
	239
	240	if (s->n_read < sizeof(s->read_size)) {
	241	ssize_t ss;
	242
	243	ss = read(fd, (uint8_t*) &s->read_size + s->n_read, sizeof(s->read_size) - s->n_read);
	244	if (ss < 0) {
	245	if (errno != EINTR && errno != EAGAIN)
b914e211	246	return dns_stream_complete(s, errno);
623a4c97	247	} else if (ss == 0)
b914e211	248	return dns_stream_complete(s, ECONNRESET);
623a4c97 LP	249	else
	250	s->n_read += ss;
	251	}
	252
	253	if (s->n_read >= sizeof(s->read_size)) {
	254
	255	if (be16toh(s->read_size) < DNS_PACKET_HEADER_SIZE)
b914e211	256	return dns_stream_complete(s, EBADMSG);
623a4c97 LP	257
	258	if (s->n_read < sizeof(s->read_size) + be16toh(s->read_size)) {
	259	ssize_t ss;
	260
	261	if (!s->read_packet) {
	262	r = dns_packet_new(&s->read_packet, s->protocol, be16toh(s->read_size));
	263	if (r < 0)
b914e211	264	return dns_stream_complete(s, -r);
623a4c97 LP	265
	266	s->read_packet->size = be16toh(s->read_size);
	267	s->read_packet->ipproto = IPPROTO_TCP;
	268	s->read_packet->family = s->peer.sa.sa_family;
	269	s->read_packet->ttl = s->ttl;
	270	s->read_packet->ifindex = s->ifindex;
	271
	272	if (s->read_packet->family == AF_INET) {
	273	s->read_packet->sender.in = s->peer.in.sin_addr;
	274	s->read_packet->sender_port = be16toh(s->peer.in.sin_port);
	275	s->read_packet->destination.in = s->local.in.sin_addr;
	276	s->read_packet->destination_port = be16toh(s->local.in.sin_port);
	277	} else {
	278	assert(s->read_packet->family == AF_INET6);
	279	s->read_packet->sender.in6 = s->peer.in6.sin6_addr;
	280	s->read_packet->sender_port = be16toh(s->peer.in6.sin6_port);
	281	s->read_packet->destination.in6 = s->local.in6.sin6_addr;
	282	s->read_packet->destination_port = be16toh(s->local.in6.sin6_port);
	283
	284	if (s->read_packet->ifindex == 0)
	285	s->read_packet->ifindex = s->peer.in6.sin6_scope_id;
	286	if (s->read_packet->ifindex == 0)
	287	s->read_packet->ifindex = s->local.in6.sin6_scope_id;
	288	}
	289	}
	290
	291	ss = read(fd,
	292	(uint8_t*) DNS_PACKET_DATA(s->read_packet) + s->n_read - sizeof(s->read_size),
	293	sizeof(s->read_size) + be16toh(s->read_size) - s->n_read);
	294	if (ss < 0) {
	295	if (errno != EINTR && errno != EAGAIN)
b914e211	296	return dns_stream_complete(s, errno);
623a4c97	297	} else if (ss == 0)
b914e211	298	return dns_stream_complete(s, ECONNRESET);
623a4c97 LP	299	else
	300	s->n_read += ss;
	301	}
	302
	303	/* Are we done? If so, disable the event source for EPOLLIN */
	304	if (s->n_read >= sizeof(s->read_size) + be16toh(s->read_size)) {
	305	r = dns_stream_update_io(s);
	306	if (r < 0)
b914e211	307	return dns_stream_complete(s, -r);
623a4c97 LP	308
	309	/* If there's a packet handler
	310	* installed, call that. Note that
	311	* this is optional... */
	312	if (s->on_packet)
	313	return s->on_packet(s);
	314	}
	315	}
	316	}
	317
	318	if ((s->write_packet && s->n_written >= sizeof(s->write_size) + s->write_packet->size) &&
	319	(s->read_packet && s->n_read >= sizeof(s->read_size) + s->read_packet->size))
b914e211	320	return dns_stream_complete(s, 0);
623a4c97 LP	321
	322	return 0;
	323	}
	324
	325	DnsStream dns_stream_free(DnsStream s) {
	326	if (!s)
	327	return NULL;
	328
	329	dns_stream_stop(s);
	330
	331	if (s->manager) {
	332	LIST_REMOVE(streams, s->manager->dns_streams, s);
	333	s->manager->n_dns_streams--;
	334	}
	335
	336	dns_packet_unref(s->write_packet);
	337	dns_packet_unref(s->read_packet);
	338
	339	free(s);
	340
	341	return 0;
	342	}
	343
	344	DEFINE_TRIVIAL_CLEANUP_FUNC(DnsStream*, dns_stream_free);
	345
	346	int dns_stream_new(Manager m, DnsStream *ret, DnsProtocol protocol, int fd) {
	347	static const int one = 1;
623a4c97	348	_cleanup_(dns_stream_freep) DnsStream *s = NULL;
623a4c97 LP	349	int r;
	350
	351	assert(m);
	352	assert(fd >= 0);
	353
	354	if (m->n_dns_streams > DNS_STREAMS_MAX)
	355	return -EBUSY;
	356
	357	s = new0(DnsStream, 1);
	358	if (!s)
	359	return -ENOMEM;
	360
	361	s->fd = -1;
	362	s->protocol = protocol;
	363
623a4c97 LP	364	r = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof(one));
	365	if (r < 0)
	366	return -errno;
	367
623a4c97 LP	368	r = sd_event_add_io(m->event, &s->io_event_source, fd, EPOLLIN, on_stream_io, s);
	369	if (r < 0)
	370	return r;
	371
9a015429 LP	372	r = sd_event_add_time(
	373	m->event,
	374	&s->timeout_event_source,
	375	clock_boottime_or_monotonic(),
	376	now(clock_boottime_or_monotonic()) + DNS_STREAM_TIMEOUT_USEC, 0,
	377	on_stream_timeout, s);
623a4c97 LP	378	if (r < 0)
	379	return r;
	380
	381	LIST_PREPEND(streams, m->dns_streams, s);
	382	s->manager = m;
	383	s->fd = fd;
	384	m->n_dns_streams++;
	385
	386	*ret = s;
	387	s = NULL;
	388
	389	return 0;
	390	}
	391
	392	int dns_stream_write_packet(DnsStream s, DnsPacket p) {
	393	assert(s);
	394
	395	if (s->write_packet)
	396	return -EBUSY;
	397
	398	s->write_packet = dns_packet_ref(p);
	399	s->write_size = htobe16(p->size);
	400	s->n_written = 0;
	401
	402	return dns_stream_update_io(s);
	403	}