GopherProxy

/*
* xfrd-tcp.c - XFR (transfer) Daemon TCP system source file. Manages tcp conn.
*
* Copyright (c) 2001-2006, NLnet Labs. All rights reserved.
*
* See LICENSE for the license.
*
*/

#include "config.h"
#include <assert.h>
#include <errno.h>
#include <fcntl.h>
#include <unistd.h>
#include <stdlib.h>
#include <sys/uio.h>
#include "nsd.h"
#include "xfrd-tcp.h"
#include "buffer.h"
#include "packet.h"
#include "dname.h"
#include "options.h"
#include "namedb.h"
#include "xfrd.h"
#include "xfrd-disk.h"
#include "util.h"
#ifdef HAVE_TLS_1_3
#include <openssl/ssl.h>
#include <openssl/err.h>
#endif

#ifdef HAVE_TLS_1_3
void log_crypto_err(const char* str); /* in server.c */

static SSL_CTX*
create_ssl_context()
{
SSL_CTX *ctx;
unsigned char protos[] = { 3, 'd', 'o', 't' };
ctx = SSL_CTX_new(TLS_client_method());
if (!ctx) {
log_msg(LOG_ERR, "xfrd tls: Unable to create SSL ctxt");
}
else if (SSL_CTX_set_default_verify_paths(ctx) != 1) {
SSL_CTX_free(ctx);
log_msg(LOG_ERR, "xfrd tls: Unable to set default SSL verify paths");
return NULL;
}
/* Only trust 1.3 as per the specification */
else if (!SSL_CTX_set_min_proto_version(ctx, TLS1_3_VERSION)) {
SSL_CTX_free(ctx);
log_msg(LOG_ERR, "xfrd tls: Unable to set minimum TLS version 1.3");
return NULL;
}

if (SSL_CTX_set_alpn_protos(ctx, protos, sizeof(protos)) != 0) {
SSL_CTX_free(ctx);
log_msg(LOG_ERR, "xfrd tls: Unable to set ALPN protocols");
return NULL;
}
return ctx;
}

static int
tls_verify_callback(int preverify_ok, X509_STORE_CTX *ctx)
{
int err = X509_STORE_CTX_get_error(ctx);
int depth = X509_STORE_CTX_get_error_depth(ctx);

// report the specific cert error here - will need custom verify code if
// SPKI pins are supported
if (!preverify_ok)
log_msg(LOG_ERR, "xfrd tls: TLS verify failed - (%d) depth: %d error: %s",
err,
depth,
X509_verify_cert_error_string(err));
return preverify_ok;
}

static int
setup_ssl(struct xfrd_tcp_pipeline* tp, struct xfrd_tcp_set* tcp_set,
const char* auth_domain_name)
{
if (!tcp_set->ssl_ctx) {
log_msg(LOG_ERR, "xfrd tls: No TLS CTX, cannot set up XFR-over-TLS");
return 0;
}
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: setting up TLS for tls_auth domain name %s",
auth_domain_name));
tp->ssl = SSL_new((SSL_CTX*)tcp_set->ssl_ctx);
if(!tp->ssl) {
log_msg(LOG_ERR, "xfrd tls: Unable to create TLS object");
return 0;
}
SSL_set_connect_state(tp->ssl);
(void)SSL_set_mode(tp->ssl, SSL_MODE_AUTO_RETRY);
if(!SSL_set_fd(tp->ssl, tp->tcp_w->fd)) {
log_msg(LOG_ERR, "xfrd tls: Unable to set TLS fd");
SSL_free(tp->ssl);
tp->ssl = NULL;
return 0;
}

SSL_set_verify(tp->ssl, SSL_VERIFY_PEER, tls_verify_callback);
if(!SSL_set1_host(tp->ssl, auth_domain_name)) {
log_msg(LOG_ERR, "xfrd tls: TLS setting of hostname %s failed",
auth_domain_name);
SSL_free(tp->ssl);
tp->ssl = NULL;
return 0;
}
return 1;
}

static int
ssl_handshake(struct xfrd_tcp_pipeline* tp)
{
int ret;

ERR_clear_error();
ret = SSL_do_handshake(tp->ssl);
if(ret == 1) {
DEBUG(DEBUG_XFRD, 1, (LOG_INFO, "xfrd: TLS handshake successful"));
tp->handshake_done = 1;
return 1;
}
tp->handshake_want = SSL_get_error(tp->ssl, ret);
if(tp->handshake_want == SSL_ERROR_WANT_READ
|| tp->handshake_want == SSL_ERROR_WANT_WRITE)
return 1;

return 0;
}

int password_cb(char *buf, int size, int ATTR_UNUSED(rwflag), void *u)
{
strlcpy(buf, (char*)u, size);
return strlen(buf);
}

#endif

/* sort tcppipe, first on IP address, for an IPaddresss, sort on num_unused */
static int
xfrd_pipe_cmp(const void* a, const void* b)
{
const struct xfrd_tcp_pipeline* x = (struct xfrd_tcp_pipeline*)a;
const struct xfrd_tcp_pipeline* y = (struct xfrd_tcp_pipeline*)b;
int r;
if(x == y)
return 0;
if(y->key.ip_len != x->key.ip_len)
/* subtraction works because nonnegative and small numbers */
return (int)y->key.ip_len - (int)x->key.ip_len;
r = memcmp(&x->key.ip, &y->key.ip, x->key.ip_len);
if(r != 0)
return r;
/* sort that num_unused is sorted ascending, */
if(x->key.num_unused != y->key.num_unused) {
return (x->key.num_unused < y->key.num_unused) ? -1 : 1;
}
/* different pipelines are different still, even with same numunused*/
return (uintptr_t)x < (uintptr_t)y ? -1 : 1;
}

struct xfrd_tcp_set* xfrd_tcp_set_create(struct region* region, const char *tls_cert_bundle, int tcp_max, int tcp_pipeline)
{
int i;
struct xfrd_tcp_set* tcp_set = region_alloc(region,
sizeof(struct xfrd_tcp_set));
memset(tcp_set, 0, sizeof(struct xfrd_tcp_set));
tcp_set->tcp_state = NULL;
tcp_set->tcp_max = tcp_max;
tcp_set->tcp_pipeline = tcp_pipeline;
tcp_set->tcp_count = 0;
tcp_set->tcp_waiting_first = 0;
tcp_set->tcp_waiting_last = 0;
#ifdef HAVE_TLS_1_3
/* Set up SSL context */
tcp_set->ssl_ctx = create_ssl_context();
if (tcp_set->ssl_ctx == NULL)
log_msg(LOG_ERR, "xfrd: XFR-over-TLS not available");

else if (tls_cert_bundle && tls_cert_bundle[0] && SSL_CTX_load_verify_locations(
tcp_set->ssl_ctx, tls_cert_bundle, NULL) != 1) {
log_msg(LOG_ERR, "xfrd tls: Unable to set the certificate bundle file %s",
tls_cert_bundle);
}
#else
(void)tls_cert_bundle;
log_msg(LOG_INFO, "xfrd: No TLS 1.3 support - XFR-over-TLS not available");
#endif
tcp_set->tcp_state = region_alloc(region,
sizeof(*tcp_set->tcp_state)*tcp_set->tcp_max);
for(i=0; i<tcp_set->tcp_max; i++)
tcp_set->tcp_state[i] = xfrd_tcp_pipeline_create(region,
tcp_pipeline);
tcp_set->pipetree = rbtree_create(region, &xfrd_pipe_cmp);
return tcp_set;
}

static int pipeline_id_compare(const void* x, const void* y)
{
struct xfrd_tcp_pipeline_id* a = (struct xfrd_tcp_pipeline_id*)x;
struct xfrd_tcp_pipeline_id* b = (struct xfrd_tcp_pipeline_id*)y;
if(a->id < b->id)
return -1;
if(a->id > b->id)
return 1;
return 0;
}

void pick_id_values(uint16_t* array, int num, int max)
{
uint8_t inserted[65536];
int j, done;
if(num == 65536) {
/* all of them, loop and insert */
int i;
for(i=0; i<num; i++)
array[i] = (uint16_t)i;
return;
}
assert(max <= 65536);
/* This uses the Robert Floyd sampling algorithm */
/* keep track if values are already inserted, using the bitmap
* in insert array */
memset(inserted, 0, sizeof(inserted[0])*max);
done=0;
for(j = max-num; j<max; j++) {
/* random generate creates from 0..arg-1 */
int t;
if(j+1 <= 1)
t = 0;
else t = random_generate(j+1);
if(!inserted[t]) {
array[done++]=t;
inserted[t] = 1;
} else {
array[done++]=j;
inserted[j] = 1;
}
}
}

static void
clear_pipeline_entry(struct xfrd_tcp_pipeline* tp, rbnode_type* node)
{
struct xfrd_tcp_pipeline_id *n;
if(node == NULL || node == RBTREE_NULL)
return;
clear_pipeline_entry(tp, node->left);
node->left = NULL;
clear_pipeline_entry(tp, node->right);
node->right = NULL;
/* move the node into the free list */
n = (struct xfrd_tcp_pipeline_id*)node;
n->next_free = tp->pipe_id_free_list;
tp->pipe_id_free_list = n;
}

static void
xfrd_tcp_pipeline_cleanup(struct xfrd_tcp_pipeline* tp)
{
/* move entries into free list */
clear_pipeline_entry(tp, tp->zone_per_id->root);
/* clear the tree */
tp->zone_per_id->count = 0;
tp->zone_per_id->root = RBTREE_NULL;
}

static void
xfrd_tcp_pipeline_init(struct xfrd_tcp_pipeline* tp)
{
tp->key.node.key = tp;
tp->key.num_unused = tp->pipe_num;
tp->key.num_skip = 0;
tp->tcp_send_first = NULL;
tp->tcp_send_last = NULL;
xfrd_tcp_pipeline_cleanup(tp);
pick_id_values(tp->unused, tp->pipe_num, 65536);
}

struct xfrd_tcp_pipeline*
xfrd_tcp_pipeline_create(region_type* region, int tcp_pipeline)
{
int i;
struct xfrd_tcp_pipeline* tp = (struct xfrd_tcp_pipeline*)
region_alloc_zero(region, sizeof(*tp));
if(tcp_pipeline < 0)
tcp_pipeline = 0;
if(tcp_pipeline > 65536)
tcp_pipeline = 65536; /* max 16 bit ID numbers */
tp->pipe_num = tcp_pipeline;
tp->key.num_unused = tp->pipe_num;
tp->zone_per_id = rbtree_create(region, &pipeline_id_compare);
tp->pipe_id_free_list = NULL;
for(i=0; i<tp->pipe_num; i++) {
struct xfrd_tcp_pipeline_id* n = (struct xfrd_tcp_pipeline_id*)
region_alloc_zero(region, sizeof(*n));
n->next_free = tp->pipe_id_free_list;
tp->pipe_id_free_list = n;
}
tp->unused = (uint16_t*)region_alloc_zero(region,
sizeof(tp->unused[0])*tp->pipe_num);
tp->tcp_r = xfrd_tcp_create(region, QIOBUFSZ);
tp->tcp_w = xfrd_tcp_create(region, 512);
xfrd_tcp_pipeline_init(tp);
return tp;
}

static struct xfrd_zone*
xfrd_tcp_pipeline_lookup_id(struct xfrd_tcp_pipeline* tp, uint16_t id)
{
struct xfrd_tcp_pipeline_id key;
rbnode_type* n;
memset(&key, 0, sizeof(key));
key.node.key = &key;
key.id = id;
n = rbtree_search(tp->zone_per_id, &key);
if(n && n != RBTREE_NULL) {
return ((struct xfrd_tcp_pipeline_id*)n)->zone;
}
return NULL;
}

static void
xfrd_tcp_pipeline_insert_id(struct xfrd_tcp_pipeline* tp, uint16_t id,
struct xfrd_zone* zone)
{
struct xfrd_tcp_pipeline_id* n;
/* because there are tp->pipe_num preallocated entries, and we have
* only tp->pipe_num id values, the list cannot be empty now. */
assert(tp->pipe_id_free_list != NULL);
/* pick up next free xfrd_tcp_pipeline_id node */
n = tp->pipe_id_free_list;
tp->pipe_id_free_list = n->next_free;
n->next_free = NULL;
memset(&n->node, 0, sizeof(n->node));
n->node.key = n;
n->id = id;
n->zone = zone;
rbtree_insert(tp->zone_per_id, &n->node);
}

static void
xfrd_tcp_pipeline_remove_id(struct xfrd_tcp_pipeline* tp, uint16_t id)
{
struct xfrd_tcp_pipeline_id key;
rbnode_type* node;
memset(&key, 0, sizeof(key));
key.node.key = &key;
key.id = id;
node = rbtree_delete(tp->zone_per_id, &key);
if(node && node != RBTREE_NULL) {
struct xfrd_tcp_pipeline_id* n =
(struct xfrd_tcp_pipeline_id*)node;
n->next_free = tp->pipe_id_free_list;
tp->pipe_id_free_list = n;
}
}

static void
xfrd_tcp_pipeline_skip_id(struct xfrd_tcp_pipeline* tp, uint16_t id)
{
struct xfrd_tcp_pipeline_id key;
rbnode_type* n;
memset(&key, 0, sizeof(key));
key.node.key = &key;
key.id = id;
n = rbtree_search(tp->zone_per_id, &key);
if(n && n != RBTREE_NULL) {
struct xfrd_tcp_pipeline_id* zid = (struct xfrd_tcp_pipeline_id*)n;
zid->zone = TCP_NULL_SKIP;
}
}

void
xfrd_setup_packet(buffer_type* packet,
uint16_t type, uint16_t klass, const dname_type* dname, uint16_t qid)
{
/* Set up the header */
buffer_clear(packet);
ID_SET(packet, qid);
FLAGS_SET(packet, 0);
OPCODE_SET(packet, OPCODE_QUERY);
QDCOUNT_SET(packet, 1);
ANCOUNT_SET(packet, 0);
NSCOUNT_SET(packet, 0);
ARCOUNT_SET(packet, 0);
buffer_skip(packet, QHEADERSZ);

/* The question record. */
buffer_write(packet, dname_name(dname), dname->name_size);
buffer_write_u16(packet, type);
buffer_write_u16(packet, klass);
}

static socklen_t
#ifdef INET6
xfrd_acl_sockaddr(acl_options_type* acl, unsigned int port,
struct sockaddr_storage *sck)
#else
xfrd_acl_sockaddr(acl_options_type* acl, unsigned int port,
struct sockaddr_in *sck, const char* fromto)
#endif /* INET6 */
{
/* setup address structure */
#ifdef INET6
memset(sck, 0, sizeof(struct sockaddr_storage));
#else
memset(sck, 0, sizeof(struct sockaddr_in));
#endif
if(acl->is_ipv6) {
#ifdef INET6
struct sockaddr_in6* sa = (struct sockaddr_in6*)sck;
sa->sin6_family = AF_INET6;
sa->sin6_port = htons(port);
sa->sin6_addr = acl->addr.addr6;
return sizeof(struct sockaddr_in6);
#else
log_msg(LOG_ERR, "xfrd: IPv6 connection %s %s attempted but no \
INET6.", fromto, acl->ip_address_spec);
return 0;
#endif
} else {
struct sockaddr_in* sa = (struct sockaddr_in*)sck;
sa->sin_family = AF_INET;
sa->sin_port = htons(port);
sa->sin_addr = acl->addr.addr;
return sizeof(struct sockaddr_in);
}
}

socklen_t
#ifdef INET6
xfrd_acl_sockaddr_to(acl_options_type* acl, struct sockaddr_storage *to)
#else
xfrd_acl_sockaddr_to(acl_options_type* acl, struct sockaddr_in *to)
#endif /* INET6 */
{
#ifdef HAVE_TLS_1_3
unsigned int port = acl->port?acl->port:(acl->tls_auth_options?
(unsigned)atoi(TLS_PORT):(unsigned)atoi(TCP_PORT));
#else
unsigned int port = acl->port?acl->port:(unsigned)atoi(TCP_PORT);
#endif
#ifdef INET6
return xfrd_acl_sockaddr(acl, port, to);
#else
return xfrd_acl_sockaddr(acl, port, to, "to");
#endif /* INET6 */
}

socklen_t
#ifdef INET6
xfrd_acl_sockaddr_frm(acl_options_type* acl, struct sockaddr_storage *frm)
#else
xfrd_acl_sockaddr_frm(acl_options_type* acl, struct sockaddr_in *frm)
#endif /* INET6 */
{
unsigned int port = acl->port?acl->port:0;
#ifdef INET6
return xfrd_acl_sockaddr(acl, port, frm);
#else
return xfrd_acl_sockaddr(acl, port, frm, "from");
#endif /* INET6 */
}

void
xfrd_write_soa_buffer(struct buffer* packet,
const dname_type* apex, struct xfrd_soa* soa)
{
size_t rdlength_pos;
uint16_t rdlength;
buffer_write(packet, dname_name(apex), apex->name_size);

/* already in network order */
buffer_write(packet, &soa->type, sizeof(soa->type));
buffer_write(packet, &soa->klass, sizeof(soa->klass));
buffer_write(packet, &soa->ttl, sizeof(soa->ttl));
rdlength_pos = buffer_position(packet);
buffer_skip(packet, sizeof(rdlength));

/* uncompressed dnames */
buffer_write(packet, soa->prim_ns+1, soa->prim_ns[0]);
buffer_write(packet, soa->email+1, soa->email[0]);

buffer_write(packet, &soa->serial, sizeof(uint32_t));
buffer_write(packet, &soa->refresh, sizeof(uint32_t));
buffer_write(packet, &soa->retry, sizeof(uint32_t));
buffer_write(packet, &soa->expire, sizeof(uint32_t));
buffer_write(packet, &soa->minimum, sizeof(uint32_t));

/* write length of RR */
rdlength = buffer_position(packet) - rdlength_pos - sizeof(rdlength);
buffer_write_u16_at(packet, rdlength_pos, rdlength);
}

struct xfrd_tcp*
xfrd_tcp_create(region_type* region, size_t bufsize)
{
struct xfrd_tcp* tcp_state = (struct xfrd_tcp*)region_alloc(
region, sizeof(struct xfrd_tcp));
memset(tcp_state, 0, sizeof(struct xfrd_tcp));
tcp_state->packet = buffer_create(region, bufsize);
tcp_state->fd = -1;

return tcp_state;
}

static struct xfrd_tcp_pipeline*
pipeline_find(struct xfrd_tcp_set* set, xfrd_zone_type* zone)
{
rbnode_type* sme = NULL;
struct xfrd_tcp_pipeline* r;
/* smaller buf than a full pipeline with 64kb ID array, only need
* the front part with the key info, this front part contains the
* members that the compare function uses. */
struct xfrd_tcp_pipeline_key k, *key=&k;
key->node.key = key;
key->ip_len = xfrd_acl_sockaddr_to(zone->master, &key->ip);
key->num_unused = set->tcp_pipeline;
/* lookup existing tcp transfer to the master with highest unused */
if(rbtree_find_less_equal(set->pipetree, key, &sme)) {
/* exact match, strange, fully unused tcp cannot be open */
assert(0);
}
if(!sme)
return NULL;
r = (struct xfrd_tcp_pipeline*)sme->key;
/* <= key pointed at, is the master correct ? */
if(r->key.ip_len != key->ip_len)
return NULL;
if(memcmp(&r->key.ip, &key->ip, key->ip_len) != 0)
return NULL;
/* correct master, is there a slot free for this transfer? */
if(r->key.num_unused == 0)
return NULL;
return r;
}

/* remove zone from tcp waiting list */
static void
tcp_zone_waiting_list_popfirst(struct xfrd_tcp_set* set, xfrd_zone_type* zone)
{
assert(zone->tcp_waiting);
set->tcp_waiting_first = zone->tcp_waiting_next;
if(zone->tcp_waiting_next)
zone->tcp_waiting_next->tcp_waiting_prev = NULL;
else set->tcp_waiting_last = 0;
zone->tcp_waiting_next = 0;
zone->tcp_waiting = 0;
}

/* remove zone from tcp pipe write-wait list */
static void
tcp_pipe_sendlist_remove(struct xfrd_tcp_pipeline* tp, xfrd_zone_type* zone)
{
if(zone->in_tcp_send) {
if(zone->tcp_send_prev)
zone->tcp_send_prev->tcp_send_next=zone->tcp_send_next;
else tp->tcp_send_first=zone->tcp_send_next;
if(zone->tcp_send_next)
zone->tcp_send_next->tcp_send_prev=zone->tcp_send_prev;
else tp->tcp_send_last=zone->tcp_send_prev;
zone->in_tcp_send = 0;
}
}

/* remove first from write-wait list */
static void
tcp_pipe_sendlist_popfirst(struct xfrd_tcp_pipeline* tp, xfrd_zone_type* zone)
{
tp->tcp_send_first = zone->tcp_send_next;
if(tp->tcp_send_first)
tp->tcp_send_first->tcp_send_prev = NULL;
else tp->tcp_send_last = NULL;
zone->in_tcp_send = 0;
}

/* remove zone from tcp pipe ID map */
static void
tcp_pipe_id_remove(struct xfrd_tcp_pipeline* tp, xfrd_zone_type* zone,
int alsotree)
{
assert(tp->key.num_unused < tp->pipe_num && tp->key.num_unused >= 0);
if(alsotree)
xfrd_tcp_pipeline_remove_id(tp, zone->query_id);
tp->unused[tp->key.num_unused] = zone->query_id;
/* must remove and re-add for sort order in tree */
(void)rbtree_delete(xfrd->tcp_set->pipetree, &tp->key.node);
tp->key.num_unused++;
(void)rbtree_insert(xfrd->tcp_set->pipetree, &tp->key.node);
}

/* stop the tcp pipe (and all its zones need to retry) */
static void
xfrd_tcp_pipe_stop(struct xfrd_tcp_pipeline* tp)
{
struct xfrd_tcp_pipeline_id* zid;
int conn = -1;
assert(tp->key.num_unused < tp->pipe_num); /* at least one 'in-use' */
assert(tp->pipe_num - tp->key.num_unused > tp->key.num_skip); /* at least one 'nonskip' */
/* need to retry for all the zones connected to it */
/* these could use different lists and go to a different nextmaster*/
RBTREE_FOR(zid, struct xfrd_tcp_pipeline_id*, tp->zone_per_id) {
xfrd_zone_type* zone = zid->zone;
if(zone && zone != TCP_NULL_SKIP) {
assert(zone->query_id == zid->id);
conn = zone->tcp_conn;
zone->tcp_conn = -1;
zone->tcp_waiting = 0;
tcp_pipe_sendlist_remove(tp, zone);
tcp_pipe_id_remove(tp, zone, 0);
xfrd_set_refresh_now(zone);
}
}
xfrd_tcp_pipeline_cleanup(tp);
assert(conn != -1);
/* now release the entire tcp pipe */
xfrd_tcp_pipe_release(xfrd->tcp_set, tp, conn);
}

static void
tcp_pipe_reset_timeout(struct xfrd_tcp_pipeline* tp)
{
int fd = tp->handler.ev_fd;
struct timeval tv;
tv.tv_sec = xfrd->tcp_set->tcp_timeout;
tv.tv_usec = 0;
if(tp->handler_added)
event_del(&tp->handler);
memset(&tp->handler, 0, sizeof(tp->handler));
event_set(&tp->handler, fd, EV_PERSIST|EV_TIMEOUT|EV_READ|
#ifdef HAVE_TLS_1_3
( tp->ssl
? ( tp->handshake_done ? ( tp->tcp_send_first ? EV_WRITE : 0 )
: tp->handshake_want == SSL_ERROR_WANT_WRITE ? EV_WRITE : 0 )
: tp->tcp_send_first ? EV_WRITE : 0 ),
#else
( tp->tcp_send_first ? EV_WRITE : 0 ),
#endif
xfrd_handle_tcp_pipe, tp);
if(event_base_set(xfrd->event_base, &tp->handler) != 0)
log_msg(LOG_ERR, "xfrd tcp: event_base_set failed");
if(event_add(&tp->handler, &tv) != 0)
log_msg(LOG_ERR, "xfrd tcp: event_add failed");
tp->handler_added = 1;
}

/* handle event from fd of tcp pipe */
void
xfrd_handle_tcp_pipe(int ATTR_UNUSED(fd), short event, void* arg)
{
struct xfrd_tcp_pipeline* tp = (struct xfrd_tcp_pipeline*)arg;
if((event & EV_WRITE)) {
tcp_pipe_reset_timeout(tp);
if(tp->tcp_send_first) {
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: event tcp write, zone %s",
tp->tcp_send_first->apex_str));
xfrd_tcp_write(tp, tp->tcp_send_first);
}
}
if((event & EV_READ) && tp->handler_added) {
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: event tcp read"));
tcp_pipe_reset_timeout(tp);
xfrd_tcp_read(tp);
}
if((event & EV_TIMEOUT) && tp->handler_added) {
/* tcp connection timed out */
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: event tcp timeout"));
xfrd_tcp_pipe_stop(tp);
}
}

/* add a zone to the pipeline, it starts to want to write its query */
static void
pipeline_setup_new_zone(struct xfrd_tcp_set* set, struct xfrd_tcp_pipeline* tp,
xfrd_zone_type* zone)
{
/* assign the ID */
int idx;
assert(tp->key.num_unused > 0);
/* we pick a random ID, even though it is TCP anyway */
idx = random_generate(tp->key.num_unused);
zone->query_id = tp->unused[idx];
tp->unused[idx] = tp->unused[tp->key.num_unused-1];
xfrd_tcp_pipeline_insert_id(tp, zone->query_id, zone);
/* decrement unused counter, and fixup tree */
(void)rbtree_delete(set->pipetree, &tp->key.node);
tp->key.num_unused--;
(void)rbtree_insert(set->pipetree, &tp->key.node);

/* add to sendlist, at end */
zone->tcp_send_next = NULL;
zone->tcp_send_prev = tp->tcp_send_last;
zone->in_tcp_send = 1;
if(tp->tcp_send_last)
tp->tcp_send_last->tcp_send_next = zone;
else tp->tcp_send_first = zone;
tp->tcp_send_last = zone;

/* is it first in line? */
if(tp->tcp_send_first == zone) {
xfrd_tcp_setup_write_packet(tp, zone);
/* add write to event handler */
tcp_pipe_reset_timeout(tp);
}
}

void
xfrd_tcp_obtain(struct xfrd_tcp_set* set, xfrd_zone_type* zone)
{
struct xfrd_tcp_pipeline* tp;
assert(zone->tcp_conn == -1);
assert(zone->tcp_waiting == 0);

if(set->tcp_count < set->tcp_max) {
int i;
assert(!set->tcp_waiting_first);
set->tcp_count ++;
/* find a free tcp_buffer */
for(i=0; i<set->tcp_max; i++) {
if(set->tcp_state[i]->tcp_r->fd == -1) {
zone->tcp_conn = i;
break;
}
}
/** What if there is no free tcp_buffer? return; */
if (zone->tcp_conn < 0) {
return;
}

tp = set->tcp_state[zone->tcp_conn];
zone->tcp_waiting = 0;

/* stop udp use (if any) */
if(zone->zone_handler.ev_fd != -1)
xfrd_udp_release(zone);

if(!xfrd_tcp_open(set, tp, zone)) {
zone->tcp_conn = -1;
set->tcp_count --;
xfrd_set_refresh_now(zone);
return;
}
/* ip and ip_len set by tcp_open */
xfrd_tcp_pipeline_init(tp);

/* insert into tree */
(void)rbtree_insert(set->pipetree, &tp->key.node);
xfrd_deactivate_zone(zone);
xfrd_unset_timer(zone);
pipeline_setup_new_zone(set, tp, zone);
return;
}
/* check for a pipeline to the same master with unused ID */
if((tp = pipeline_find(set, zone))!= NULL) {
int i;
if(zone->zone_handler.ev_fd != -1)
xfrd_udp_release(zone);
for(i=0; i<set->tcp_max; i++) {
if(set->tcp_state[i] == tp)
zone->tcp_conn = i;
}
xfrd_deactivate_zone(zone);
xfrd_unset_timer(zone);
pipeline_setup_new_zone(set, tp, zone);
return;
}

/* wait, at end of line */
DEBUG(DEBUG_XFRD,2, (LOG_INFO, "xfrd: max number of tcp "
"connections (%d) reached.", set->tcp_max));
zone->tcp_waiting_next = 0;
zone->tcp_waiting_prev = set->tcp_waiting_last;
zone->tcp_waiting = 1;
if(!set->tcp_waiting_last) {
set->tcp_waiting_first = zone;
set->tcp_waiting_last = zone;
} else {
set->tcp_waiting_last->tcp_waiting_next = zone;
set->tcp_waiting_last = zone;
}
xfrd_deactivate_zone(zone);
xfrd_unset_timer(zone);
}

int
xfrd_tcp_open(struct xfrd_tcp_set* set, struct xfrd_tcp_pipeline* tp,
xfrd_zone_type* zone)
{
int fd, family, conn;
struct timeval tv;
assert(zone->tcp_conn != -1);

/* if there is no next master, fallback to use the first one */
/* but there really should be a master set */
if(!zone->master) {
zone->master = zone->zone_options->pattern->request_xfr;
zone->master_num = 0;
}

DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: zone %s open tcp conn to %s",
zone->apex_str, zone->master->ip_address_spec));
tp->tcp_r->is_reading = 1;
tp->tcp_r->total_bytes = 0;
tp->tcp_r->msglen = 0;
buffer_clear(tp->tcp_r->packet);
tp->tcp_w->is_reading = 0;
tp->tcp_w->total_bytes = 0;
tp->tcp_w->msglen = 0;
tp->connection_established = 0;

if(zone->master->is_ipv6) {
#ifdef INET6
family = PF_INET6;
#else
xfrd_set_refresh_now(zone);
return 0;
#endif
} else {
family = PF_INET;
}
fd = socket(family, SOCK_STREAM, IPPROTO_TCP);
if(fd == -1) {
/* squelch 'Address family not supported by protocol' at low
* verbosity levels */
if(errno != EAFNOSUPPORT || verbosity > 2)
log_msg(LOG_ERR, "xfrd: %s cannot create tcp socket: %s",
zone->master->ip_address_spec, strerror(errno));
xfrd_set_refresh_now(zone);
return 0;
}
if(fcntl(fd, F_SETFL, O_NONBLOCK) == -1) {
log_msg(LOG_ERR, "xfrd: fcntl failed: %s", strerror(errno));
close(fd);
xfrd_set_refresh_now(zone);
return 0;
}

if(xfrd->nsd->outgoing_tcp_mss > 0) {
#if defined(IPPROTO_TCP) && defined(TCP_MAXSEG)
if(setsockopt(fd, IPPROTO_TCP, TCP_MAXSEG,
(void*)&xfrd->nsd->outgoing_tcp_mss,
sizeof(xfrd->nsd->outgoing_tcp_mss)) < 0) {
log_msg(LOG_ERR, "xfrd: setsockopt(TCP_MAXSEG)"
"failed: %s", strerror(errno));
}
#else
log_msg(LOG_ERR, "setsockopt(TCP_MAXSEG) unsupported");
#endif
}

tp->key.ip_len = xfrd_acl_sockaddr_to(zone->master, &tp->key.ip);

/* bind it */
if (!xfrd_bind_local_interface(fd, zone->zone_options->pattern->
outgoing_interface, zone->master, 1)) {
close(fd);
xfrd_set_refresh_now(zone);
return 0;
}

conn = connect(fd, (struct sockaddr*)&tp->key.ip, tp->key.ip_len);
if (conn == -1 && errno != EINPROGRESS) {
log_msg(LOG_ERR, "xfrd: connect %s failed: %s",
zone->master->ip_address_spec, strerror(errno));
close(fd);
xfrd_set_refresh_now(zone);
return 0;
}
tp->tcp_r->fd = fd;
tp->tcp_w->fd = fd;

/* Check if an tls_auth name is configured which means we should try to
establish an SSL connection */
if (zone->master->tls_auth_options &&
zone->master->tls_auth_options->auth_domain_name) {
#ifdef HAVE_TLS_1_3
if (!setup_ssl(tp, set, zone->master->tls_auth_options->auth_domain_name)) {
log_msg(LOG_ERR, "xfrd: Cannot setup TLS on pipeline for %s to %s",
zone->apex_str, zone->master->ip_address_spec);
close(fd);
xfrd_set_refresh_now(zone);
return 0;
}

/* Load client certificate (if provided) */
if (zone->master->tls_auth_options->client_cert &&
zone->master->tls_auth_options->client_key) {
if (SSL_CTX_use_certificate_chain_file(set->ssl_ctx,
zone->master->tls_auth_options->client_cert) != 1) {
log_msg(LOG_ERR, "xfrd tls: Unable to load client certificate from file %s", zone->master->tls_auth_options->client_cert);
}

if (zone->master->tls_auth_options->client_key_pw) {
SSL_CTX_set_default_passwd_cb(set->ssl_ctx, password_cb);
SSL_CTX_set_default_passwd_cb_userdata(set->ssl_ctx, zone->master->tls_auth_options->client_key_pw);
}

if (SSL_CTX_use_PrivateKey_file(set->ssl_ctx, zone->master->tls_auth_options->client_key, SSL_FILETYPE_PEM) != 1) {
log_msg(LOG_ERR, "xfrd tls: Unable to load private key from file %s", zone->master->tls_auth_options->client_key);
}
}

tp->handshake_done = 0;
if(!ssl_handshake(tp)) {
if(tp->handshake_want == SSL_ERROR_SYSCALL) {
log_msg(LOG_ERR, "xfrd: TLS handshake failed "
"for %s to %s: %s", zone->apex_str,
zone->master->ip_address_spec,
strerror(errno));

} else if(tp->handshake_want == SSL_ERROR_SSL) {
char errmsg[1024];
snprintf(errmsg, sizeof(errmsg), "xfrd: "
"TLS handshake failed for %s to %s",
zone->apex_str,
zone->master->ip_address_spec);
log_crypto_err(errmsg);
} else {
log_msg(LOG_ERR, "xfrd: TLS handshake failed "
"for %s to %s with %d", zone->apex_str,
zone->master->ip_address_spec,
tp->handshake_want);
}
close(fd);
xfrd_set_refresh_now(zone);
return 0;
}
#else
log_msg(LOG_ERR, "xfrd: TLS 1.3 is not available, XFR-over-TLS is "
"not supported for %s to %s",
zone->apex_str, zone->master->ip_address_spec);
close(fd);
xfrd_set_refresh_now(zone);
return 0;
#endif
}

/* set the tcp pipe event */
if(tp->handler_added)
event_del(&tp->handler);
memset(&tp->handler, 0, sizeof(tp->handler));
event_set(&tp->handler, fd, EV_PERSIST|EV_TIMEOUT|EV_READ|
#ifdef HAVE_TLS_1_3
( !tp->ssl
|| tp->handshake_done
|| tp->handshake_want == SSL_ERROR_WANT_WRITE ? EV_WRITE : 0),
#else
EV_WRITE,
#endif
xfrd_handle_tcp_pipe, tp);
if(event_base_set(xfrd->event_base, &tp->handler) != 0)
log_msg(LOG_ERR, "xfrd tcp: event_base_set failed");
tv.tv_sec = set->tcp_timeout;
tv.tv_usec = 0;
if(event_add(&tp->handler, &tv) != 0)
log_msg(LOG_ERR, "xfrd tcp: event_add failed");
tp->handler_added = 1;
return 1;
}

void
xfrd_tcp_setup_write_packet(struct xfrd_tcp_pipeline* tp, xfrd_zone_type* zone)
{
struct xfrd_tcp* tcp = tp->tcp_w;
assert(zone->tcp_conn != -1);
assert(zone->tcp_waiting == 0);
/* start AXFR or IXFR for the zone */
if(zone->soa_disk_acquired == 0 || zone->master->use_axfr_only ||
zone->master->ixfr_disabled ||
/* if zone expired, after the first round, do not ask for
* IXFR any more, but full AXFR (of any serial number) */
(zone->state == xfrd_zone_expired && zone->round_num != 0)) {
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "request full zone transfer "
"(AXFR) for %s to %s",
zone->apex_str, zone->master->ip_address_spec));

xfrd_setup_packet(tcp->packet, TYPE_AXFR, CLASS_IN, zone->apex,
zone->query_id);
xfrd_prepare_zone_xfr(zone, TYPE_AXFR);
} else {
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "request incremental zone "
"transfer (IXFR) for %s to %s",
zone->apex_str, zone->master->ip_address_spec));

xfrd_setup_packet(tcp->packet, TYPE_IXFR, CLASS_IN, zone->apex,
zone->query_id);
xfrd_prepare_zone_xfr(zone, TYPE_IXFR);
NSCOUNT_SET(tcp->packet, 1);
xfrd_write_soa_buffer(tcp->packet, zone->apex, &zone->soa_disk);
}
if(zone->master->key_options && zone->master->key_options->tsig_key) {
xfrd_tsig_sign_request(
tcp->packet, &zone->latest_xfr->tsig, zone->master);
}
buffer_flip(tcp->packet);
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "sent tcp query with ID %d", zone->query_id));
tcp->msglen = buffer_limit(tcp->packet);
tcp->total_bytes = 0;
}

static void
tcp_conn_ready_for_reading(struct xfrd_tcp* tcp)
{
tcp->total_bytes = 0;
tcp->msglen = 0;
buffer_clear(tcp->packet);
}

#ifdef HAVE_TLS_1_3
static int
conn_write_ssl(struct xfrd_tcp* tcp, SSL* ssl)
{
int request_length;
ssize_t sent;

if(tcp->total_bytes < sizeof(tcp->msglen)) {
uint16_t sendlen = htons(tcp->msglen);
// send
request_length = sizeof(tcp->msglen) - tcp->total_bytes;
ERR_clear_error();
sent = SSL_write(ssl, (const char*)&sendlen + tcp->total_bytes,
request_length);
switch(SSL_get_error(ssl,sent)) {
case SSL_ERROR_NONE:
break;
default:
log_msg(LOG_ERR, "xfrd: generic write problem with tls");
}

if(sent == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* write would block, try later */
return 0;
} else {
return -1;
}
}

tcp->total_bytes += sent;
if(sent > (ssize_t)sizeof(tcp->msglen))
buffer_skip(tcp->packet, sent-sizeof(tcp->msglen));
if(tcp->total_bytes < sizeof(tcp->msglen)) {
/* incomplete write, resume later */
return 0;
}
assert(tcp->total_bytes >= sizeof(tcp->msglen));
}

assert(tcp->total_bytes < tcp->msglen + sizeof(tcp->msglen));

request_length = buffer_remaining(tcp->packet);
ERR_clear_error();
sent = SSL_write(ssl, buffer_current(tcp->packet), request_length);
switch(SSL_get_error(ssl,sent)) {
case SSL_ERROR_NONE:
break;
default:
log_msg(LOG_ERR, "xfrd: generic write problem with tls");
}
if(sent == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* write would block, try later */
return 0;
} else {
return -1;
}
}

buffer_skip(tcp->packet, sent);
tcp->total_bytes += sent;

if(tcp->total_bytes < tcp->msglen + sizeof(tcp->msglen)) {
/* more to write when socket becomes writable again */
return 0;
}

assert(tcp->total_bytes == tcp->msglen + sizeof(tcp->msglen));
return 1;
}
#endif

int conn_write(struct xfrd_tcp* tcp)
{
ssize_t sent;

if(tcp->total_bytes < sizeof(tcp->msglen)) {
uint16_t sendlen = htons(tcp->msglen);
#ifdef HAVE_WRITEV
struct iovec iov[2];
iov[0].iov_base = (uint8_t*)&sendlen + tcp->total_bytes;
iov[0].iov_len = sizeof(sendlen) - tcp->total_bytes;
iov[1].iov_base = buffer_begin(tcp->packet);
iov[1].iov_len = buffer_limit(tcp->packet);
sent = writev(tcp->fd, iov, 2);
#else /* HAVE_WRITEV */
sent = write(tcp->fd,
(const char*)&sendlen + tcp->total_bytes,
sizeof(tcp->msglen) - tcp->total_bytes);
#endif /* HAVE_WRITEV */

if(sent == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* write would block, try later */
return 0;
} else {
return -1;
}
}

tcp->total_bytes += sent;
if(sent > (ssize_t)sizeof(tcp->msglen))
buffer_skip(tcp->packet, sent-sizeof(tcp->msglen));
if(tcp->total_bytes < sizeof(tcp->msglen)) {
/* incomplete write, resume later */
return 0;
}
#ifdef HAVE_WRITEV
if(tcp->total_bytes == tcp->msglen + sizeof(tcp->msglen)) {
/* packet done */
return 1;
}
#endif
assert(tcp->total_bytes >= sizeof(tcp->msglen));
}

assert(tcp->total_bytes < tcp->msglen + sizeof(tcp->msglen));

sent = write(tcp->fd,
buffer_current(tcp->packet),
buffer_remaining(tcp->packet));
if(sent == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* write would block, try later */
return 0;
} else {
return -1;
}
}

buffer_skip(tcp->packet, sent);
tcp->total_bytes += sent;

if(tcp->total_bytes < tcp->msglen + sizeof(tcp->msglen)) {
/* more to write when socket becomes writable again */
return 0;
}

assert(tcp->total_bytes == tcp->msglen + sizeof(tcp->msglen));
return 1;
}

void
xfrd_tcp_write(struct xfrd_tcp_pipeline* tp, xfrd_zone_type* zone)
{
int ret;
struct xfrd_tcp* tcp = tp->tcp_w;
assert(zone->tcp_conn != -1);
assert(zone == tp->tcp_send_first);
/* see if for non-established connection, there is a connect error */
if(!tp->connection_established) {
/* check for pending error from nonblocking connect */
/* from Stevens, unix network programming, vol1, 3rd ed, p450 */
int error = 0;
socklen_t len = sizeof(error);
if(getsockopt(tcp->fd, SOL_SOCKET, SO_ERROR, &error, &len) < 0){
error = errno; /* on solaris errno is error */
}
if(error == EINPROGRESS || error == EWOULDBLOCK)
return; /* try again later */
if(error != 0) {
log_msg(LOG_ERR, "%s: Could not tcp connect to %s: %s",
zone->apex_str, zone->master->ip_address_spec,
strerror(error));
xfrd_tcp_pipe_stop(tp);
return;
}
}
#ifdef HAVE_TLS_1_3
if (tp->ssl) {
if(tp->handshake_done) {
ret = conn_write_ssl(tcp, tp->ssl);

} else if(ssl_handshake(tp)) {
tcp_pipe_reset_timeout(tp); /* reschedule */
return;

} else {
if(tp->handshake_want == SSL_ERROR_SYSCALL) {
log_msg(LOG_ERR, "xfrd: TLS handshake failed: %s",
strerror(errno));

} else if(tp->handshake_want == SSL_ERROR_SSL) {
log_crypto_err("xfrd: TLS handshake failed");
} else {
log_msg(LOG_ERR, "xfrd: TLS handshake failed "
"with value: %d", tp->handshake_want);
}
xfrd_tcp_pipe_stop(tp);
return;
}
} else
#endif
ret = conn_write(tcp);
if(ret == -1) {
log_msg(LOG_ERR, "xfrd: failed writing tcp %s", strerror(errno));
xfrd_tcp_pipe_stop(tp);
return;
}
if(tcp->total_bytes != 0 && !tp->connection_established)
tp->connection_established = 1;
if(ret == 0) {
return; /* write again later */
}
/* done writing this message */

/* remove first zone from sendlist */
tcp_pipe_sendlist_popfirst(tp, zone);

/* see if other zone wants to write; init; let it write (now) */
/* and use a loop, because 64k stack calls is a too much */
while(tp->tcp_send_first) {
/* setup to write for this zone */
xfrd_tcp_setup_write_packet(tp, tp->tcp_send_first);
/* attempt to write for this zone (if success, continue loop)*/
#ifdef HAVE_TLS_1_3
if (tp->ssl)
ret = conn_write_ssl(tcp, tp->ssl);
else
#endif
ret = conn_write(tcp);
if(ret == -1) {
log_msg(LOG_ERR, "xfrd: failed writing tcp %s", strerror(errno));
xfrd_tcp_pipe_stop(tp);
return;
}
if(ret == 0)
return; /* write again later */
tcp_pipe_sendlist_popfirst(tp, tp->tcp_send_first);
}

/* if sendlist empty, remove WRITE from event */

/* listen to READ, and not WRITE events */
assert(tp->tcp_send_first == NULL);
tcp_pipe_reset_timeout(tp);
}

#ifdef HAVE_TLS_1_3
static int
conn_read_ssl(struct xfrd_tcp* tcp, SSL* ssl)
{
ssize_t received;
/* receive leading packet length bytes */
if(tcp->total_bytes < sizeof(tcp->msglen)) {
ERR_clear_error();
received = SSL_read(ssl,
(char*) &tcp->msglen + tcp->total_bytes,
sizeof(tcp->msglen) - tcp->total_bytes);
if (received <= 0) {
int err = SSL_get_error(ssl, received);
if(err == SSL_ERROR_WANT_READ && errno == EAGAIN) {
return 0;
}
if(err == SSL_ERROR_ZERO_RETURN) {
/* EOF */
return -1;
}
if(err == SSL_ERROR_SYSCALL)
log_msg(LOG_ERR, "ssl_read returned error SSL_ERROR_SYSCALL with received %zd: %s", received, strerror(errno));
else
log_msg(LOG_ERR, "ssl_read returned error %d with received %zd", err, received);
}
if(received == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* read would block, try later */
return 0;
} else {
#ifdef ECONNRESET
if (verbosity >= 2 || errno != ECONNRESET)
#endif /* ECONNRESET */
log_msg(LOG_ERR, "tls read sz: %s", strerror(errno));
return -1;
}
} else if(received == 0) {
/* EOF */
return -1;
}
tcp->total_bytes += received;
if(tcp->total_bytes < sizeof(tcp->msglen)) {
/* not complete yet, try later */
return 0;
}

assert(tcp->total_bytes == sizeof(tcp->msglen));
tcp->msglen = ntohs(tcp->msglen);

if(tcp->msglen == 0) {
buffer_set_limit(tcp->packet, tcp->msglen);
return 1;
}
if(tcp->msglen > buffer_capacity(tcp->packet)) {
log_msg(LOG_ERR, "buffer too small, dropping connection");
return 0;
}
buffer_set_limit(tcp->packet, tcp->msglen);
}

assert(buffer_remaining(tcp->packet) > 0);
ERR_clear_error();

received = SSL_read(ssl, buffer_current(tcp->packet),
buffer_remaining(tcp->packet));

if (received <= 0) {
int err = SSL_get_error(ssl, received);
if(err == SSL_ERROR_ZERO_RETURN) {
/* EOF */
return -1;
}
if(err == SSL_ERROR_SYSCALL)
log_msg(LOG_ERR, "ssl_read returned error SSL_ERROR_SYSCALL with received %zd: %s", received, strerror(errno));
else
log_msg(LOG_ERR, "ssl_read returned error %d with received %zd", err, received);
}
if(received == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* read would block, try later */
return 0;
} else {
#ifdef ECONNRESET
if (verbosity >= 2 || errno != ECONNRESET)
#endif /* ECONNRESET */
log_msg(LOG_ERR, "tcp read %s", strerror(errno));
return -1;
}
} else if(received == 0) {
/* EOF */
return -1;
}

tcp->total_bytes += received;
buffer_skip(tcp->packet, received);

if(buffer_remaining(tcp->packet) > 0) {
/* not complete yet, wait for more */
return 0;
}

/* completed */
assert(buffer_position(tcp->packet) == tcp->msglen);
return 1;
}
#endif

int
conn_read(struct xfrd_tcp* tcp)
{
ssize_t received;
/* receive leading packet length bytes */
if(tcp->total_bytes < sizeof(tcp->msglen)) {
received = read(tcp->fd,
(char*) &tcp->msglen + tcp->total_bytes,
sizeof(tcp->msglen) - tcp->total_bytes);
if(received == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* read would block, try later */
return 0;
} else {
#ifdef ECONNRESET
if (verbosity >= 2 || errno != ECONNRESET)
#endif /* ECONNRESET */
log_msg(LOG_ERR, "tcp read sz: %s", strerror(errno));
return -1;
}
} else if(received == 0) {
/* EOF */
return -1;
}
tcp->total_bytes += received;
if(tcp->total_bytes < sizeof(tcp->msglen)) {
/* not complete yet, try later */
return 0;
}

assert(tcp->total_bytes == sizeof(tcp->msglen));
tcp->msglen = ntohs(tcp->msglen);

if(tcp->msglen == 0) {
buffer_set_limit(tcp->packet, tcp->msglen);
return 1;
}
if(tcp->msglen > buffer_capacity(tcp->packet)) {
log_msg(LOG_ERR, "buffer too small, dropping connection");
return 0;
}
buffer_set_limit(tcp->packet, tcp->msglen);
}

assert(buffer_remaining(tcp->packet) > 0);

received = read(tcp->fd, buffer_current(tcp->packet),
buffer_remaining(tcp->packet));
if(received == -1) {
if(errno == EAGAIN || errno == EINTR) {
/* read would block, try later */
return 0;
} else {
#ifdef ECONNRESET
if (verbosity >= 2 || errno != ECONNRESET)
#endif /* ECONNRESET */
log_msg(LOG_ERR, "tcp read %s", strerror(errno));
return -1;
}
} else if(received == 0) {
/* EOF */
return -1;
}

tcp->total_bytes += received;
buffer_skip(tcp->packet, received);

if(buffer_remaining(tcp->packet) > 0) {
/* not complete yet, wait for more */
return 0;
}

/* completed */
assert(buffer_position(tcp->packet) == tcp->msglen);
return 1;
}

void
xfrd_tcp_read(struct xfrd_tcp_pipeline* tp)
{
xfrd_zone_type* zone;
struct xfrd_tcp* tcp = tp->tcp_r;
int ret;
enum xfrd_packet_result pkt_result;
#ifdef HAVE_TLS_1_3
if(tp->ssl) {
if(tp->handshake_done) {
ret = conn_read_ssl(tcp, tp->ssl);

} else if(ssl_handshake(tp)) {
tcp_pipe_reset_timeout(tp); /* reschedule */
return;

} else {
if(tp->handshake_want == SSL_ERROR_SYSCALL) {
log_msg(LOG_ERR, "xfrd: TLS handshake failed: %s",
strerror(errno));

} else if(tp->handshake_want == SSL_ERROR_SSL) {
log_crypto_err("xfrd: TLS handshake failed");
} else {
log_msg(LOG_ERR, "xfrd: TLS handshake failed "
"with value: %d", tp->handshake_want);
}
xfrd_tcp_pipe_stop(tp);
return;
}
} else
#endif
ret = conn_read(tcp);
if(ret == -1) {
if(errno != 0)
log_msg(LOG_ERR, "xfrd: failed reading tcp %s", strerror(errno));
else
log_msg(LOG_ERR, "xfrd: failed reading tcp: closed");
xfrd_tcp_pipe_stop(tp);
return;
}
if(ret == 0)
return;
/* completed msg */
buffer_flip(tcp->packet);
/* see which ID number it is, if skip, handle skip, NULL: warn */
if(tcp->msglen < QHEADERSZ) {
/* too short for DNS header, skip it */
DEBUG(DEBUG_XFRD,1, (LOG_INFO,
"xfrd: tcp skip response that is too short"));
tcp_conn_ready_for_reading(tcp);
return;
}
zone = xfrd_tcp_pipeline_lookup_id(tp, ID(tcp->packet));
if(!zone || zone == TCP_NULL_SKIP) {
/* no zone for this id? skip it */
DEBUG(DEBUG_XFRD,1, (LOG_INFO,
"xfrd: tcp skip response with %s ID",
zone?"set-to-skip":"unknown"));
tcp_conn_ready_for_reading(tcp);
return;
}
assert(zone->tcp_conn != -1);

/* handle message for zone */
pkt_result = xfrd_handle_received_xfr_packet(zone, tcp->packet);
/* setup for reading the next packet on this connection */
tcp_conn_ready_for_reading(tcp);
switch(pkt_result) {
case xfrd_packet_more:
/* wait for next packet */
break;
case xfrd_packet_newlease:
/* set to skip if more packets with this ID */
xfrd_tcp_pipeline_skip_id(tp, zone->query_id);
tp->key.num_skip++;
/* fall through to remove zone from tp */
/* fallthrough */
case xfrd_packet_transfer:
if(zone->zone_options->pattern->multi_master_check) {
xfrd_tcp_release(xfrd->tcp_set, zone);
xfrd_make_request(zone);
break;
}
xfrd_tcp_release(xfrd->tcp_set, zone);
assert(zone->round_num == -1);
break;
case xfrd_packet_notimpl:
xfrd_disable_ixfr(zone);
xfrd_tcp_release(xfrd->tcp_set, zone);
/* query next server */
xfrd_make_request(zone);
break;
case xfrd_packet_bad:
case xfrd_packet_tcp:
default:
/* set to skip if more packets with this ID */
xfrd_tcp_pipeline_skip_id(tp, zone->query_id);
tp->key.num_skip++;
xfrd_tcp_release(xfrd->tcp_set, zone);
/* query next server */
xfrd_make_request(zone);
break;
}
}

void
xfrd_tcp_release(struct xfrd_tcp_set* set, xfrd_zone_type* zone)
{
int conn = zone->tcp_conn;
struct xfrd_tcp_pipeline* tp = set->tcp_state[conn];
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: zone %s released tcp conn to %s",
zone->apex_str, zone->master->ip_address_spec));
assert(zone->tcp_conn != -1);
assert(zone->tcp_waiting == 0);
zone->tcp_conn = -1;
zone->tcp_waiting = 0;

/* remove from tcp_send list */
tcp_pipe_sendlist_remove(tp, zone);
/* remove it from the ID list */
if(xfrd_tcp_pipeline_lookup_id(tp, zone->query_id) != TCP_NULL_SKIP)
tcp_pipe_id_remove(tp, zone, 1);
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: released tcp pipe now %d unused",
tp->key.num_unused));
/* if pipe was full, but no more, then see if waiting element is
* for the same master, and can fill the unused ID */
if(tp->key.num_unused == 1 && set->tcp_waiting_first) {
#ifdef INET6
struct sockaddr_storage to;
#else
struct sockaddr_in to;
#endif
socklen_t to_len = xfrd_acl_sockaddr_to(
set->tcp_waiting_first->master, &to);
if(to_len == tp->key.ip_len && memcmp(&to, &tp->key.ip, to_len) == 0) {
/* use this connection for the waiting zone */
zone = set->tcp_waiting_first;
assert(zone->tcp_conn == -1);
zone->tcp_conn = conn;
tcp_zone_waiting_list_popfirst(set, zone);
if(zone->zone_handler.ev_fd != -1)
xfrd_udp_release(zone);
xfrd_unset_timer(zone);
pipeline_setup_new_zone(set, tp, zone);
return;
}
/* waiting zone did not go to same server */
}

/* if all unused, or only skipped leftover, close the pipeline */
if(tp->key.num_unused >= tp->pipe_num || tp->key.num_skip >= tp->pipe_num - tp->key.num_unused)
xfrd_tcp_pipe_release(set, tp, conn);
}

void
xfrd_tcp_pipe_release(struct xfrd_tcp_set* set, struct xfrd_tcp_pipeline* tp,
int conn)
{
DEBUG(DEBUG_XFRD,1, (LOG_INFO, "xfrd: tcp pipe released"));
/* one handler per tcp pipe */
if(tp->handler_added)
event_del(&tp->handler);
tp->handler_added = 0;

#ifdef HAVE_TLS_1_3
/* close SSL */
if (tp->ssl) {
DEBUG(DEBUG_XFRD, 1, (LOG_INFO, "xfrd: Shutting down TLS"));
SSL_shutdown(tp->ssl);
SSL_free(tp->ssl);
tp->ssl = NULL;
}
#endif

/* fd in tcp_r and tcp_w is the same, close once */
if(tp->tcp_r->fd != -1)
close(tp->tcp_r->fd);
tp->tcp_r->fd = -1;
tp->tcp_w->fd = -1;

/* remove from pipetree */
(void)rbtree_delete(xfrd->tcp_set->pipetree, &tp->key.node);

/* a waiting zone can use the free tcp slot (to another server) */
/* if that zone fails to set-up or connect, we try to start the next
* waiting zone in the list */
while(set->tcp_count == set->tcp_max && set->tcp_waiting_first) {
/* pop first waiting process */
xfrd_zone_type* zone = set->tcp_waiting_first;
/* start it */
assert(zone->tcp_conn == -1);
zone->tcp_conn = conn;
tcp_zone_waiting_list_popfirst(set, zone);

/* stop udp (if any) */
if(zone->zone_handler.ev_fd != -1)
xfrd_udp_release(zone);
if(!xfrd_tcp_open(set, tp, zone)) {
zone->tcp_conn = -1;
xfrd_set_refresh_now(zone);
/* try to start the next zone (if any) */
continue;
}
/* re-init this tcppipe */
/* ip and ip_len set by tcp_open */
xfrd_tcp_pipeline_init(tp);

/* insert into tree */
(void)rbtree_insert(set->pipetree, &tp->key.node);
/* setup write */
xfrd_unset_timer(zone);
pipeline_setup_new_zone(set, tp, zone);
/* started a task, no need for cleanups, so return */
return;
}
/* no task to start, cleanup */
assert(!set->tcp_waiting_first);
set->tcp_count --;
assert(set->tcp_count >= 0);
}