summaryrefslogtreecommitdiff
path: root/services
diff options
context:
space:
mode:
authorCy Schubert <cy@FreeBSD.org>2020-12-08 21:23:03 +0000
committerCy Schubert <cy@FreeBSD.org>2020-12-08 21:23:03 +0000
commitc1dbcbf2d10cd99864ab0eb44358d9875ba0c0a5 (patch)
tree8cbca8d9dc814933d2bc59b6623b792b549aac6b /services
parent4cb89f2eee3bb358f0491932ab0498b5319f4229 (diff)
Vendor import of Unbound 1.13.0.vendor/unbound/1.13.0
Security: CVE-2020-28935
Notes
svn path=/vendor/unbound/dist/; revision=368464 svn path=/vendor/unbound/1.13.0/; revision=368465; tag=vendor/unbound/1.13.0
Diffstat (limited to 'services')
-rw-r--r--services/authzone.c2
-rw-r--r--services/cache/infra.c28
-rw-r--r--services/cache/infra.h2
-rw-r--r--services/listen_dnsport.c54
-rw-r--r--services/listen_dnsport.h6
-rw-r--r--services/localzone.c2
-rw-r--r--services/mesh.c59
-rw-r--r--services/outside_network.c1087
-rw-r--r--services/outside_network.h145
-rw-r--r--services/rpz.c2
10 files changed, 1219 insertions, 168 deletions
diff --git a/services/authzone.c b/services/authzone.c
index a26d1003abe2..15be5d60c653 100644
--- a/services/authzone.c
+++ b/services/authzone.c
@@ -5387,6 +5387,7 @@ void auth_xfer_transfer_lookup_callback(void* arg, int rcode, sldns_buffer* buf,
verbose(VERB_ALGO, "auth zone %s host %s type %s transfer lookup has no answer", zname, xfr->task_transfer->lookup_target->host, (xfr->task_transfer->lookup_aaaa?"AAAA":"A"));
}
}
+ regional_free_all(temp);
} else {
if(verbosity >= VERB_ALGO) {
char zname[255+1];
@@ -6444,6 +6445,7 @@ void auth_xfer_probe_lookup_callback(void* arg, int rcode, sldns_buffer* buf,
verbose(VERB_ALGO, "auth zone %s host %s type %s probe lookup has no address", zname, xfr->task_probe->lookup_target->host, (xfr->task_probe->lookup_aaaa?"AAAA":"A"));
}
}
+ regional_free_all(temp);
} else {
if(verbosity >= VERB_ALGO) {
char zname[255+1];
diff --git a/services/cache/infra.c b/services/cache/infra.c
index c2484a9f1aa8..2d16bcd6e405 100644
--- a/services/cache/infra.c
+++ b/services/cache/infra.c
@@ -244,6 +244,7 @@ infra_create(struct config_file* cfg)
return NULL;
}
infra->host_ttl = cfg->host_ttl;
+ infra->infra_keep_probing = cfg->infra_keep_probing;
infra_dp_ratelimit = cfg->ratelimit;
infra->domain_rates = slabhash_create(cfg->ratelimit_slabs,
INFRA_HOST_STARTSIZE, cfg->ratelimit_size,
@@ -297,6 +298,7 @@ infra_adjust(struct infra_cache* infra, struct config_file* cfg)
if(!infra)
return infra_create(cfg);
infra->host_ttl = cfg->host_ttl;
+ infra->infra_keep_probing = cfg->infra_keep_probing;
infra_dp_ratelimit = cfg->ratelimit;
infra_ip_ratelimit = cfg->ip_ratelimit;
maxmem = cfg->infra_cache_numhosts * (sizeof(struct infra_key)+
@@ -445,6 +447,7 @@ infra_host(struct infra_cache* infra, struct sockaddr_storage* addr,
if(e && ((struct infra_data*)e->data)->ttl < timenow) {
/* it expired, try to reuse existing entry */
int old = ((struct infra_data*)e->data)->rtt.rto;
+ time_t tprobe = ((struct infra_data*)e->data)->probedelay;
uint8_t tA = ((struct infra_data*)e->data)->timeout_A;
uint8_t tAAAA = ((struct infra_data*)e->data)->timeout_AAAA;
uint8_t tother = ((struct infra_data*)e->data)->timeout_other;
@@ -460,6 +463,7 @@ infra_host(struct infra_cache* infra, struct sockaddr_storage* addr,
if(old >= USEFUL_SERVER_TOP_TIMEOUT) {
((struct infra_data*)e->data)->rtt.rto
= USEFUL_SERVER_TOP_TIMEOUT;
+ ((struct infra_data*)e->data)->probedelay = tprobe;
((struct infra_data*)e->data)->timeout_A = tA;
((struct infra_data*)e->data)->timeout_AAAA = tAAAA;
((struct infra_data*)e->data)->timeout_other = tother;
@@ -482,7 +486,8 @@ infra_host(struct infra_cache* infra, struct sockaddr_storage* addr,
*edns_vs = data->edns_version;
*edns_lame_known = data->edns_lame_known;
*to = rtt_timeout(&data->rtt);
- if(*to >= PROBE_MAXRTO && rtt_notimeout(&data->rtt)*4 <= *to) {
+ if(*to >= PROBE_MAXRTO && (infra->infra_keep_probing ||
+ rtt_notimeout(&data->rtt)*4 <= *to)) {
/* delay other queries, this is the probe query */
if(!wr) {
lock_rw_unlock(&e->lock);
@@ -566,18 +571,27 @@ infra_rtt_update(struct infra_cache* infra, struct sockaddr_storage* addr,
struct lruhash_entry* e = infra_lookup_nottl(infra, addr, addrlen,
nm, nmlen, 1);
struct infra_data* data;
- int needtoinsert = 0;
+ int needtoinsert = 0, expired = 0;
int rto = 1;
+ time_t oldprobedelay = 0;
if(!e) {
if(!(e = new_entry(infra, addr, addrlen, nm, nmlen, timenow)))
return 0;
needtoinsert = 1;
} else if(((struct infra_data*)e->data)->ttl < timenow) {
+ oldprobedelay = ((struct infra_data*)e->data)->probedelay;
data_entry_init(infra, e, timenow);
+ expired = 1;
}
/* have an entry, update the rtt */
data = (struct infra_data*)e->data;
if(roundtrip == -1) {
+ if(needtoinsert || expired) {
+ /* timeout on entry that has expired before the timer
+ * keep old timeout from the function caller */
+ data->rtt.rto = orig_rtt;
+ data->probedelay = oldprobedelay;
+ }
rtt_lost(&data->rtt, orig_rtt);
if(qtype == LDNS_RR_TYPE_A) {
if(data->timeout_A < TIMEOUT_COUNT_MAX)
@@ -681,7 +695,12 @@ infra_get_lame_rtt(struct infra_cache* infra,
return 0;
host = (struct infra_data*)e->data;
*rtt = rtt_unclamped(&host->rtt);
- if(host->rtt.rto >= PROBE_MAXRTO && timenow < host->probedelay
+ if(host->rtt.rto >= PROBE_MAXRTO && timenow >= host->probedelay
+ && infra->infra_keep_probing) {
+ /* single probe, keep probing */
+ if(*rtt >= USEFUL_SERVER_TOP_TIMEOUT)
+ *rtt = USEFUL_SERVER_TOP_TIMEOUT-1000;
+ } else if(host->rtt.rto >= PROBE_MAXRTO && timenow < host->probedelay
&& rtt_notimeout(&host->rtt)*4 <= host->rtt.rto) {
/* single probe for this domain, and we are not probing */
/* unless the query type allows a probe to happen */
@@ -704,7 +723,8 @@ infra_get_lame_rtt(struct infra_cache* infra,
/* see if this can be a re-probe of an unresponsive server */
/* minus 1000 because that is outside of the RTTBAND, so
* blacklisted servers stay blacklisted if this is chosen */
- if(host->rtt.rto >= USEFUL_SERVER_TOP_TIMEOUT) {
+ if(host->rtt.rto >= USEFUL_SERVER_TOP_TIMEOUT ||
+ infra->infra_keep_probing) {
lock_rw_unlock(&e->lock);
*rtt = USEFUL_SERVER_TOP_TIMEOUT-1000;
*lame = 0;
diff --git a/services/cache/infra.h b/services/cache/infra.h
index e33f2a6c04ee..14f97c4c64d3 100644
--- a/services/cache/infra.h
+++ b/services/cache/infra.h
@@ -114,6 +114,8 @@ struct infra_cache {
struct slabhash* hosts;
/** TTL value for host information, in seconds */
int host_ttl;
+ /** the hosts that are down are kept probed for recovery */
+ int infra_keep_probing;
/** hash table with query rates per name: rate_key, rate_data */
struct slabhash* domain_rates;
/** ratelimit settings for domains, struct domain_limit_data */
diff --git a/services/listen_dnsport.c b/services/listen_dnsport.c
index 3a98c264297a..d63c0e0aab00 100644
--- a/services/listen_dnsport.c
+++ b/services/listen_dnsport.c
@@ -43,6 +43,7 @@
# include <sys/types.h>
#endif
#include <sys/time.h>
+#include <limits.h>
#ifdef USE_TCP_FASTOPEN
#include <netinet/tcp.h>
#endif
@@ -81,9 +82,6 @@
/** number of queued TCP connections for listen() */
#define TCP_BACKLOG 256
-/** number of simultaneous requests a client can have */
-#define TCP_MAX_REQ_SIMULTANEOUS 32
-
#ifndef THREADS_DISABLED
/** lock on the counter of stream buffer memory */
static lock_basic_type stream_wait_count_lock;
@@ -533,7 +531,9 @@ create_udp_sock(int family, int socktype, struct sockaddr* addr,
return -1;
}
}
-# elif defined(IP_DONTFRAG)
+# elif defined(IP_DONTFRAG) && !defined(__APPLE__)
+ /* the IP_DONTFRAG option if defined in the 11.0 OSX headers,
+ * but does not work on that version, so we exclude it */
int off = 0;
if (setsockopt(s, IPPROTO_IP, IP_DONTFRAG,
&off, (socklen_t)sizeof(off)) < 0) {
@@ -1244,8 +1244,9 @@ struct listen_dnsport*
listen_create(struct comm_base* base, struct listen_port* ports,
size_t bufsize, int tcp_accept_count, int tcp_idle_timeout,
int harden_large_queries, uint32_t http_max_streams,
- char* http_endpoint, struct tcl_list* tcp_conn_limit, void* sslctx,
- struct dt_env* dtenv, comm_point_callback_type* cb, void *cb_arg)
+ char* http_endpoint, int http_notls, struct tcl_list* tcp_conn_limit,
+ void* sslctx, struct dt_env* dtenv, comm_point_callback_type* cb,
+ void *cb_arg)
{
struct listen_dnsport* front = (struct listen_dnsport*)
malloc(sizeof(struct listen_dnsport));
@@ -1295,15 +1296,19 @@ listen_create(struct comm_base* base, struct listen_port* ports,
http_max_streams, http_endpoint,
tcp_conn_limit, bufsize, front->udp_buff,
ports->ftype, cb, cb_arg);
- cp->ssl = sslctx;
+ if(http_notls && ports->ftype == listen_type_http)
+ cp->ssl = NULL;
+ else
+ cp->ssl = sslctx;
if(ports->ftype == listen_type_http) {
- if(!sslctx) {
- log_warn("HTTPS port configured, but no TLS "
+ if(!sslctx && !http_notls) {
+ log_warn("HTTPS port configured, but no TLS "
"tls-service-key or tls-service-pem "
"set");
}
#ifndef HAVE_SSL_CTX_SET_ALPN_SELECT_CB
- log_warn("Unbound is not compiled with an "
+ if(!http_notls)
+ log_warn("Unbound is not compiled with an "
"OpenSSL version supporting ALPN "
" (OpenSSL >= 1.0.2). This is required "
"to use DNS-over-HTTPS");
@@ -1402,6 +1407,7 @@ static int
resolve_ifa_name(struct ifaddrs *ifas, const char *search_ifa, char ***ip_addresses, int *ip_addresses_size)
{
struct ifaddrs *ifa;
+ void *tmpbuf;
int last_ip_addresses_size = *ip_addresses_size;
for(ifa = ifas; ifa != NULL; ifa = ifa->ifa_next) {
@@ -1466,10 +1472,12 @@ resolve_ifa_name(struct ifaddrs *ifas, const char *search_ifa, char ***ip_addres
}
verbose(4, "interface %s has address %s", search_ifa, addr_buf);
- *ip_addresses = realloc(*ip_addresses, sizeof(char *) * (*ip_addresses_size + 1));
- if(!*ip_addresses) {
+ tmpbuf = realloc(*ip_addresses, sizeof(char *) * (*ip_addresses_size + 1));
+ if(!tmpbuf) {
log_err("realloc failed: out of memory");
return 0;
+ } else {
+ *ip_addresses = tmpbuf;
}
(*ip_addresses)[*ip_addresses_size] = strdup(addr_buf);
if(!(*ip_addresses)[*ip_addresses_size]) {
@@ -1480,10 +1488,12 @@ resolve_ifa_name(struct ifaddrs *ifas, const char *search_ifa, char ***ip_addres
}
if (*ip_addresses_size == last_ip_addresses_size) {
- *ip_addresses = realloc(*ip_addresses, sizeof(char *) * (*ip_addresses_size + 1));
- if(!*ip_addresses) {
+ tmpbuf = realloc(*ip_addresses, sizeof(char *) * (*ip_addresses_size + 1));
+ if(!tmpbuf) {
log_err("realloc failed: out of memory");
return 0;
+ } else {
+ *ip_addresses = tmpbuf;
}
(*ip_addresses)[*ip_addresses_size] = strdup(search_ifa);
if(!(*ip_addresses)[*ip_addresses_size]) {
@@ -1804,8 +1814,7 @@ tcp_req_info_setup_listen(struct tcp_req_info* req)
if(!req->cp->tcp_is_reading)
wr = 1;
- if(req->num_open_req + req->num_done_req < TCP_MAX_REQ_SIMULTANEOUS &&
- !req->read_is_closed)
+ if(!req->read_is_closed)
rd = 1;
if(wr) {
@@ -2177,9 +2186,10 @@ int http2_submit_dns_response(struct http2_session* h2_session)
int ret;
nghttp2_data_provider data_prd;
char status[4];
- nghttp2_nv headers[2];
+ nghttp2_nv headers[3];
struct http2_stream* h2_stream = h2_session->c->h2_stream;
size_t rlen;
+ char rlen_str[32];
if(h2_stream->rbuffer) {
log_err("http2 submit response error: rbuffer already "
@@ -2198,6 +2208,8 @@ int http2_submit_dns_response(struct http2_session* h2_session)
}
rlen = sldns_buffer_remaining(h2_session->c->buffer);
+ snprintf(rlen_str, sizeof(rlen_str), "%u", (unsigned)rlen);
+
lock_basic_lock(&http2_response_buffer_count_lock);
if(http2_response_buffer_count + rlen > http2_response_buffer_max) {
lock_basic_unlock(&http2_response_buffer_count_lock);
@@ -2228,13 +2240,11 @@ int http2_submit_dns_response(struct http2_session* h2_session)
headers[1].valuelen = 23;
headers[1].flags = NGHTTP2_NV_FLAG_NONE;
- /*TODO be nice and add the content-length header
headers[2].name = (uint8_t*)"content-length";
headers[2].namelen = 14;
- headers[2].value =
- headers[2].valuelen =
+ headers[2].value = (uint8_t*)rlen_str;
+ headers[2].valuelen = strlen(rlen_str);
headers[2].flags = NGHTTP2_NV_FLAG_NONE;
- */
sldns_buffer_write(h2_stream->rbuffer,
sldns_buffer_current(h2_session->c->buffer),
@@ -2244,7 +2254,7 @@ int http2_submit_dns_response(struct http2_session* h2_session)
data_prd.source.ptr = h2_session;
data_prd.read_callback = http2_submit_response_read_callback;
ret = nghttp2_submit_response(h2_session->session, h2_stream->stream_id,
- headers, 2, &data_prd);
+ headers, 3, &data_prd);
if(ret) {
verbose(VERB_QUERY, "http2: set_stream_user_data failed, "
"error: %s", nghttp2_strerror(ret));
diff --git a/services/listen_dnsport.h b/services/listen_dnsport.h
index 4bbde0691fec..9d6ea2c33adf 100644
--- a/services/listen_dnsport.h
+++ b/services/listen_dnsport.h
@@ -159,6 +159,7 @@ int resolve_interface_names(struct config_file* cfg, char*** resif,
* @param harden_large_queries: whether query size should be limited.
* @param http_max_streams: maximum number of HTTP/2 streams per connection.
* @param http_endpoint: HTTP endpoint to service queries on
+ * @param http_notls: no TLS for http downstream
* @param tcp_conn_limit: TCP connection limit info.
* @param sslctx: nonNULL if ssl context.
* @param dtenv: nonNULL if dnstap enabled.
@@ -171,8 +172,9 @@ struct listen_dnsport*
listen_create(struct comm_base* base, struct listen_port* ports,
size_t bufsize, int tcp_accept_count, int tcp_idle_timeout,
int harden_large_queries, uint32_t http_max_streams,
- char* http_endpoint, struct tcl_list* tcp_conn_limit, void* sslctx,
- struct dt_env* dtenv, comm_point_callback_type* cb, void *cb_arg);
+ char* http_endpoint, int http_notls, struct tcl_list* tcp_conn_limit,
+ void* sslctx, struct dt_env* dtenv, comm_point_callback_type* cb,
+ void *cb_arg);
/**
* delete the listening structure
diff --git a/services/localzone.c b/services/localzone.c
index 6aaf0c05518c..cad46066334c 100644
--- a/services/localzone.c
+++ b/services/localzone.c
@@ -157,7 +157,7 @@ local_zone_create(uint8_t* nm, size_t len, int labs,
z->namelen = len;
z->namelabs = labs;
lock_rw_init(&z->lock);
- z->region = regional_create_custom(sizeof(struct regional));
+ z->region = regional_create_nochunk(sizeof(struct regional));
if(!z->region) {
free(z);
return NULL;
diff --git a/services/mesh.c b/services/mesh.c
index 52ff97e4a2e8..cd90509366f2 100644
--- a/services/mesh.c
+++ b/services/mesh.c
@@ -1196,6 +1196,12 @@ mesh_send_reply(struct mesh_state* m, int rcode, struct reply_info* rep,
/* Copy the client's EDNS for later restore, to make sure the edns
* compare is with the correct edns options. */
struct edns_data edns_bak = r->edns;
+ /* briefly set the replylist to null in case the
+ * meshsendreply calls tcpreqinfo sendreply that
+ * comm_point_drops because of size, and then the
+ * null stops the mesh state remove and thus
+ * reply_list modification and accounting */
+ struct mesh_reply* rlist = m->reply_list;
/* examine security status */
if(m->s.env->need_to_validate && (!(r->qflags&BIT_CD) ||
m->s.env->cfg->ignore_cd) && rep &&
@@ -1218,15 +1224,21 @@ mesh_send_reply(struct mesh_state* m, int rcode, struct reply_info* rep,
r->h2_stream->mesh_state = NULL;
}
/* send the reply */
- /* We don't reuse the encoded answer if either the previous or current
- * response has a local alias. We could compare the alias records
- * and still reuse the previous answer if they are the same, but that
- * would be complicated and error prone for the relatively minor case.
- * So we err on the side of safety. */
- if(prev && prev_buffer && prev->qflags == r->qflags &&
+ /* We don't reuse the encoded answer if:
+ * - either the previous or current response has a local alias. We could
+ * compare the alias records and still reuse the previous answer if they
+ * are the same, but that would be complicated and error prone for the
+ * relatively minor case. So we err on the side of safety.
+ * - there are registered callback functions for the given rcode, as these
+ * need to be called for each reply. */
+ if(((rcode != LDNS_RCODE_SERVFAIL &&
+ !m->s.env->inplace_cb_lists[inplace_cb_reply]) ||
+ (rcode == LDNS_RCODE_SERVFAIL &&
+ !m->s.env->inplace_cb_lists[inplace_cb_reply_servfail])) &&
+ prev && prev_buffer && prev->qflags == r->qflags &&
!prev->local_alias && !r->local_alias &&
- prev->edns.edns_present == r->edns.edns_present &&
- prev->edns.bits == r->edns.bits &&
+ prev->edns.edns_present == r->edns.edns_present &&
+ prev->edns.bits == r->edns.bits &&
prev->edns.udp_size == r->edns.udp_size &&
edns_opt_list_compare(prev->edns.opt_list, r->edns.opt_list)
== 0) {
@@ -1236,22 +1248,26 @@ mesh_send_reply(struct mesh_state* m, int rcode, struct reply_info* rep,
sldns_buffer_write_at(r_buffer, 0, &r->qid, sizeof(uint16_t));
sldns_buffer_write_at(r_buffer, 12, r->qname,
m->s.qinfo.qname_len);
+ m->reply_list = NULL;
comm_point_send_reply(&r->query_reply);
+ m->reply_list = rlist;
} else if(rcode) {
m->s.qinfo.qname = r->qname;
m->s.qinfo.local_alias = r->local_alias;
if(rcode == LDNS_RCODE_SERVFAIL) {
if(!inplace_cb_reply_servfail_call(m->s.env, &m->s.qinfo, &m->s,
- rep, rcode, &r->edns, NULL, m->s.region))
+ rep, rcode, &r->edns, &r->query_reply, m->s.region))
r->edns.opt_list = NULL;
} else {
if(!inplace_cb_reply_call(m->s.env, &m->s.qinfo, &m->s, rep, rcode,
- &r->edns, NULL, m->s.region))
+ &r->edns, &r->query_reply, m->s.region))
r->edns.opt_list = NULL;
}
error_encode(r_buffer, rcode, &m->s.qinfo, r->qid,
r->qflags, &r->edns);
+ m->reply_list = NULL;
comm_point_send_reply(&r->query_reply);
+ m->reply_list = rlist;
} else {
size_t udp_size = r->edns.udp_size;
r->edns.edns_version = EDNS_ADVERTISED_VERSION;
@@ -1261,7 +1277,7 @@ mesh_send_reply(struct mesh_state* m, int rcode, struct reply_info* rep,
m->s.qinfo.qname = r->qname;
m->s.qinfo.local_alias = r->local_alias;
if(!inplace_cb_reply_call(m->s.env, &m->s.qinfo, &m->s, rep,
- LDNS_RCODE_NOERROR, &r->edns, NULL, m->s.region) ||
+ LDNS_RCODE_NOERROR, &r->edns, &r->query_reply, m->s.region) ||
!apply_edns_options(&r->edns, &edns_bak,
m->s.env->cfg, r->query_reply.c,
m->s.region) ||
@@ -1271,13 +1287,15 @@ mesh_send_reply(struct mesh_state* m, int rcode, struct reply_info* rep,
secure))
{
if(!inplace_cb_reply_servfail_call(m->s.env, &m->s.qinfo, &m->s,
- rep, LDNS_RCODE_SERVFAIL, &r->edns, NULL, m->s.region))
+ rep, LDNS_RCODE_SERVFAIL, &r->edns, &r->query_reply, m->s.region))
r->edns.opt_list = NULL;
error_encode(r_buffer, LDNS_RCODE_SERVFAIL,
&m->s.qinfo, r->qid, r->qflags, &r->edns);
}
r->edns = edns_bak;
+ m->reply_list = NULL;
comm_point_send_reply(&r->query_reply);
+ m->reply_list = rlist;
}
/* account */
log_assert(m->s.env->mesh->num_reply_addrs > 0);
@@ -1365,20 +1383,12 @@ void mesh_query_done(struct mesh_state* mstate)
mstate->reply_list = reply_list;
} else {
struct sldns_buffer* r_buffer = r->query_reply.c->buffer;
- struct mesh_reply* rlist = mstate->reply_list;
if(r->query_reply.c->tcp_req_info) {
r_buffer = r->query_reply.c->tcp_req_info->spool_buffer;
prev_buffer = NULL;
}
- /* briefly set the replylist to null in case the
- * meshsendreply calls tcpreqinfo sendreply that
- * comm_point_drops because of size, and then the
- * null stops the mesh state remove and thus
- * reply_list modification and accounting */
- mstate->reply_list = NULL;
mesh_send_reply(mstate, mstate->s.return_rcode, rep,
r, r_buffer, prev, prev_buffer);
- mstate->reply_list = rlist;
if(r->query_reply.c->tcp_req_info) {
tcp_req_info_remove_mesh_state(r->query_reply.c->tcp_req_info, mstate);
r_buffer = NULL;
@@ -1894,7 +1904,7 @@ mesh_serve_expired_callback(void* arg)
{
struct mesh_state* mstate = (struct mesh_state*) arg;
struct module_qstate* qstate = &mstate->s;
- struct mesh_reply* r, *rlist;
+ struct mesh_reply* r;
struct mesh_area* mesh = qstate->env->mesh;
struct dns_msg* msg;
struct mesh_cb* c;
@@ -1999,15 +2009,8 @@ mesh_serve_expired_callback(void* arg)
r_buffer = r->query_reply.c->buffer;
if(r->query_reply.c->tcp_req_info)
r_buffer = r->query_reply.c->tcp_req_info->spool_buffer;
- /* briefly set the replylist to null in case the meshsendreply
- * calls tcpreqinfo sendreply that comm_point_drops because
- * of size, and then the null stops the mesh state remove and
- * thus reply_list modification and accounting */
- rlist = mstate->reply_list;
- mstate->reply_list = NULL;
mesh_send_reply(mstate, LDNS_RCODE_NOERROR, msg->rep,
r, r_buffer, prev, prev_buffer);
- mstate->reply_list = rlist;
if(r->query_reply.c->tcp_req_info)
tcp_req_info_remove_mesh_state(r->query_reply.c->tcp_req_info, mstate);
prev = r;
diff --git a/services/outside_network.c b/services/outside_network.c
index 41a1d83f1454..11951adea7bc 100644
--- a/services/outside_network.c
+++ b/services/outside_network.c
@@ -132,6 +132,52 @@ serviced_cmp(const void* key1, const void* key2)
return sockaddr_cmp(&q1->addr, q1->addrlen, &q2->addr, q2->addrlen);
}
+/** compare if the reuse element has the same address, port and same ssl-is
+ * used-for-it characteristic */
+static int
+reuse_cmp_addrportssl(const void* key1, const void* key2)
+{
+ struct reuse_tcp* r1 = (struct reuse_tcp*)key1;
+ struct reuse_tcp* r2 = (struct reuse_tcp*)key2;
+ int r;
+ /* compare address and port */
+ r = sockaddr_cmp(&r1->addr, r1->addrlen, &r2->addr, r2->addrlen);
+ if(r != 0)
+ return r;
+
+ /* compare if SSL-enabled */
+ if(r1->is_ssl && !r2->is_ssl)
+ return 1;
+ if(!r1->is_ssl && r2->is_ssl)
+ return -1;
+ return 0;
+}
+
+int
+reuse_cmp(const void* key1, const void* key2)
+{
+ int r;
+ r = reuse_cmp_addrportssl(key1, key2);
+ if(r != 0)
+ return r;
+
+ /* compare ptr value */
+ if(key1 < key2) return -1;
+ if(key1 > key2) return 1;
+ return 0;
+}
+
+int reuse_id_cmp(const void* key1, const void* key2)
+{
+ struct waiting_tcp* w1 = (struct waiting_tcp*)key1;
+ struct waiting_tcp* w2 = (struct waiting_tcp*)key2;
+ if(w1->id < w2->id)
+ return -1;
+ if(w1->id > w2->id)
+ return 1;
+ return 0;
+}
+
/** delete waiting_tcp entry. Does not unlink from waiting list.
* @param w: to delete.
*/
@@ -280,15 +326,234 @@ outnet_tcp_connect(int s, struct sockaddr_storage* addr, socklen_t addrlen)
return 1;
}
+/** log reuse item addr and ptr with message */
+static void
+log_reuse_tcp(enum verbosity_value v, const char* msg, struct reuse_tcp* reuse)
+{
+ uint16_t port;
+ char addrbuf[128];
+ if(verbosity < v) return;
+ addr_to_str(&reuse->addr, reuse->addrlen, addrbuf, sizeof(addrbuf));
+ port = ntohs(((struct sockaddr_in*)&reuse->addr)->sin_port);
+ verbose(v, "%s %s#%u fd %d", msg, addrbuf, (unsigned)port,
+ reuse->pending->c->fd);
+}
+
+/** pop the first element from the writewait list */
+static struct waiting_tcp* reuse_write_wait_pop(struct reuse_tcp* reuse)
+{
+ struct waiting_tcp* w = reuse->write_wait_first;
+ if(!w)
+ return NULL;
+ log_assert(w->write_wait_queued);
+ log_assert(!w->write_wait_prev);
+ reuse->write_wait_first = w->write_wait_next;
+ if(w->write_wait_next)
+ w->write_wait_next->write_wait_prev = NULL;
+ else reuse->write_wait_last = NULL;
+ w->write_wait_queued = 0;
+ return w;
+}
+
+/** remove the element from the writewait list */
+static void reuse_write_wait_remove(struct reuse_tcp* reuse,
+ struct waiting_tcp* w)
+{
+ if(!w)
+ return;
+ if(!w->write_wait_queued)
+ return;
+ if(w->write_wait_prev)
+ w->write_wait_prev->write_wait_next = w->write_wait_next;
+ else reuse->write_wait_first = w->write_wait_next;
+ if(w->write_wait_next)
+ w->write_wait_next->write_wait_prev = w->write_wait_prev;
+ else reuse->write_wait_last = w->write_wait_prev;
+ w->write_wait_queued = 0;
+}
+
+/** push the element after the last on the writewait list */
+static void reuse_write_wait_push_back(struct reuse_tcp* reuse,
+ struct waiting_tcp* w)
+{
+ if(!w) return;
+ log_assert(!w->write_wait_queued);
+ if(reuse->write_wait_last) {
+ reuse->write_wait_last->write_wait_next = w;
+ w->write_wait_prev = reuse->write_wait_last;
+ } else {
+ reuse->write_wait_first = w;
+ }
+ reuse->write_wait_last = w;
+ w->write_wait_queued = 1;
+}
+
+/** insert element in tree by id */
+void
+reuse_tree_by_id_insert(struct reuse_tcp* reuse, struct waiting_tcp* w)
+{
+ log_assert(w->id_node.key == NULL);
+ w->id_node.key = w;
+ rbtree_insert(&reuse->tree_by_id, &w->id_node);
+}
+
+/** find element in tree by id */
+struct waiting_tcp*
+reuse_tcp_by_id_find(struct reuse_tcp* reuse, uint16_t id)
+{
+ struct waiting_tcp key_w;
+ rbnode_type* n;
+ memset(&key_w, 0, sizeof(key_w));
+ key_w.id_node.key = &key_w;
+ key_w.id = id;
+ n = rbtree_search(&reuse->tree_by_id, &key_w);
+ if(!n) return NULL;
+ return (struct waiting_tcp*)n->key;
+}
+
+/** return ID value of rbnode in tree_by_id */
+static uint16_t
+tree_by_id_get_id(rbnode_type* node)
+{
+ struct waiting_tcp* w = (struct waiting_tcp*)node->key;
+ return w->id;
+}
+
+/** insert into reuse tcp tree and LRU, false on failure (duplicate) */
+static int
+reuse_tcp_insert(struct outside_network* outnet, struct pending_tcp* pend_tcp)
+{
+ log_reuse_tcp(VERB_CLIENT, "reuse_tcp_insert", &pend_tcp->reuse);
+ if(pend_tcp->reuse.item_on_lru_list)
+ return 1;
+ pend_tcp->reuse.node.key = &pend_tcp->reuse;
+ pend_tcp->reuse.pending = pend_tcp;
+ if(!rbtree_insert(&outnet->tcp_reuse, &pend_tcp->reuse.node)) {
+ /* this is a duplicate connection, close this one */
+ verbose(VERB_CLIENT, "reuse_tcp_insert: duplicate connection");
+ pend_tcp->reuse.node.key = NULL;
+ return 0;
+ }
+ /* insert into LRU, first is newest */
+ pend_tcp->reuse.lru_prev = NULL;
+ if(outnet->tcp_reuse_first) {
+ pend_tcp->reuse.lru_next = outnet->tcp_reuse_first;
+ outnet->tcp_reuse_first->lru_prev = &pend_tcp->reuse;
+ } else {
+ pend_tcp->reuse.lru_next = NULL;
+ outnet->tcp_reuse_last = &pend_tcp->reuse;
+ }
+ outnet->tcp_reuse_first = &pend_tcp->reuse;
+ pend_tcp->reuse.item_on_lru_list = 1;
+ return 1;
+}
+
+/** find reuse tcp stream to destination for query, or NULL if none */
+static struct reuse_tcp*
+reuse_tcp_find(struct outside_network* outnet, struct sockaddr_storage* addr,
+ socklen_t addrlen, int use_ssl)
+{
+ struct waiting_tcp key_w;
+ struct pending_tcp key_p;
+ struct comm_point c;
+ rbnode_type* result = NULL, *prev;
+ verbose(VERB_CLIENT, "reuse_tcp_find");
+ memset(&key_w, 0, sizeof(key_w));
+ memset(&key_p, 0, sizeof(key_p));
+ memset(&c, 0, sizeof(c));
+ key_p.query = &key_w;
+ key_p.c = &c;
+ key_p.reuse.pending = &key_p;
+ key_p.reuse.node.key = &key_p.reuse;
+ if(use_ssl)
+ key_p.reuse.is_ssl = 1;
+ if(addrlen > (socklen_t)sizeof(key_p.reuse.addr))
+ return NULL;
+ memmove(&key_p.reuse.addr, addr, addrlen);
+ key_p.reuse.addrlen = addrlen;
+
+ verbose(VERB_CLIENT, "reuse_tcp_find: num reuse streams %u",
+ (unsigned)outnet->tcp_reuse.count);
+ if(outnet->tcp_reuse.root == NULL ||
+ outnet->tcp_reuse.root == RBTREE_NULL)
+ return NULL;
+ if(rbtree_find_less_equal(&outnet->tcp_reuse, &key_p.reuse.node,
+ &result)) {
+ /* exact match */
+ /* but the key is on stack, and ptr is compared, impossible */
+ log_assert(&key_p.reuse != (struct reuse_tcp*)result);
+ log_assert(&key_p != ((struct reuse_tcp*)result)->pending);
+ }
+ /* not found, return null */
+ if(!result || result == RBTREE_NULL)
+ return NULL;
+ verbose(VERB_CLIENT, "reuse_tcp_find check inexact match");
+ /* inexact match, find one of possibly several connections to the
+ * same destination address, with the correct port, ssl, and
+ * also less than max number of open queries, or else, fail to open
+ * a new one */
+ /* rewind to start of sequence of same address,port,ssl */
+ prev = rbtree_previous(result);
+ while(prev && prev != RBTREE_NULL &&
+ reuse_cmp_addrportssl(prev->key, &key_p.reuse) == 0) {
+ result = prev;
+ prev = rbtree_previous(result);
+ }
+
+ /* loop to find first one that has correct characteristics */
+ while(result && result != RBTREE_NULL &&
+ reuse_cmp_addrportssl(result->key, &key_p.reuse) == 0) {
+ if(((struct reuse_tcp*)result)->tree_by_id.count <
+ MAX_REUSE_TCP_QUERIES) {
+ /* same address, port, ssl-yes-or-no, and has
+ * space for another query */
+ return (struct reuse_tcp*)result;
+ }
+ result = rbtree_next(result);
+ }
+ return NULL;
+}
+
+/** use the buffer to setup writing the query */
+static void
+outnet_tcp_take_query_setup(int s, struct pending_tcp* pend,
+ struct waiting_tcp* w)
+{
+ struct timeval tv;
+ verbose(VERB_CLIENT, "outnet_tcp_take_query_setup: setup packet to write "
+ "len %d timeout %d msec",
+ (int)w->pkt_len, w->timeout);
+ pend->c->tcp_write_pkt = w->pkt;
+ pend->c->tcp_write_pkt_len = w->pkt_len;
+ pend->c->tcp_write_and_read = 1;
+ pend->c->tcp_write_byte_count = 0;
+ pend->c->tcp_is_reading = 0;
+ comm_point_start_listening(pend->c, s, -1);
+ /* set timer on the waiting_tcp entry, this is the write timeout
+ * for the written packet. The timer on pend->c is the timer
+ * for when there is no written packet and we have readtimeouts */
+#ifndef S_SPLINT_S
+ tv.tv_sec = w->timeout/1000;
+ tv.tv_usec = (w->timeout%1000)*1000;
+#endif
+ /* if the waiting_tcp was previously waiting for a buffer in the
+ * outside_network.tcpwaitlist, then the timer is reset now that
+ * we start writing it */
+ comm_timer_set(w->timer, &tv);
+}
+
/** use next free buffer to service a tcp query */
static int
-outnet_tcp_take_into_use(struct waiting_tcp* w, uint8_t* pkt, size_t pkt_len)
+outnet_tcp_take_into_use(struct waiting_tcp* w)
{
struct pending_tcp* pend = w->outnet->tcp_free;
int s;
log_assert(pend);
- log_assert(pkt);
+ log_assert(w->pkt);
+ log_assert(w->pkt_len > 0);
log_assert(w->addrlen > 0);
+ pend->c->tcp_do_toggle_rw = 0;
+ pend->c->tcp_do_close = 0;
/* open socket */
s = outnet_get_tcp_fd(&w->addr, w->addrlen, w->outnet->tcp_mss, w->outnet->ip_dscp);
@@ -383,24 +648,65 @@ outnet_tcp_take_into_use(struct waiting_tcp* w, uint8_t* pkt, size_t pkt_len)
return 0;
}
}
- w->pkt = NULL;
w->next_waiting = (void*)pend;
- pend->id = LDNS_ID_WIRE(pkt);
w->outnet->num_tcp_outgoing++;
w->outnet->tcp_free = pend->next_free;
pend->next_free = NULL;
pend->query = w;
+ pend->reuse.outnet = w->outnet;
pend->c->repinfo.addrlen = w->addrlen;
+ pend->c->tcp_more_read_again = &pend->reuse.cp_more_read_again;
+ pend->c->tcp_more_write_again = &pend->reuse.cp_more_write_again;
+ pend->reuse.cp_more_read_again = 0;
+ pend->reuse.cp_more_write_again = 0;
memcpy(&pend->c->repinfo.addr, &w->addr, w->addrlen);
- sldns_buffer_clear(pend->c->buffer);
- sldns_buffer_write(pend->c->buffer, pkt, pkt_len);
- sldns_buffer_flip(pend->c->buffer);
- pend->c->tcp_is_reading = 0;
- pend->c->tcp_byte_count = 0;
- comm_point_start_listening(pend->c, s, -1);
+ pend->reuse.pending = pend;
+ if(pend->c->ssl)
+ pend->reuse.is_ssl = 1;
+ else pend->reuse.is_ssl = 0;
+ /* insert in reuse by address tree if not already inserted there */
+ (void)reuse_tcp_insert(w->outnet, pend);
+ reuse_tree_by_id_insert(&pend->reuse, w);
+ outnet_tcp_take_query_setup(s, pend, w);
return 1;
}
+/** Touch the lru of a reuse_tcp element, it is in use.
+ * This moves it to the front of the list, where it is not likely to
+ * be closed. Items at the back of the list are closed to make space. */
+static void
+reuse_tcp_lru_touch(struct outside_network* outnet, struct reuse_tcp* reuse)
+{
+ if(!reuse->item_on_lru_list)
+ return; /* not on the list, no lru to modify */
+ if(!reuse->lru_prev)
+ return; /* already first in the list */
+ /* remove at current position */
+ /* since it is not first, there is a previous element */
+ reuse->lru_prev->lru_next = reuse->lru_next;
+ if(reuse->lru_next)
+ reuse->lru_next->lru_prev = reuse->lru_prev;
+ else outnet->tcp_reuse_last = reuse->lru_prev;
+ /* insert at the front */
+ reuse->lru_prev = NULL;
+ reuse->lru_next = outnet->tcp_reuse_first;
+ /* since it is not first, it is not the only element and
+ * lru_next is thus not NULL and thus reuse is now not the last in
+ * the list, so outnet->tcp_reuse_last does not need to be modified */
+ outnet->tcp_reuse_first = reuse;
+}
+
+/** call callback on waiting_tcp, if not NULL */
+static void
+waiting_tcp_callback(struct waiting_tcp* w, struct comm_point* c, int error,
+ struct comm_reply* reply_info)
+{
+ if(w->cb) {
+ fptr_ok(fptr_whitelist_pending_tcp(w->cb));
+ (void)(*w->cb)(c, w->cb_arg, error, reply_info);
+ }
+}
+
/** see if buffers can be used to service TCP queries */
static void
use_free_buffer(struct outside_network* outnet)
@@ -408,25 +714,198 @@ use_free_buffer(struct outside_network* outnet)
struct waiting_tcp* w;
while(outnet->tcp_free && outnet->tcp_wait_first
&& !outnet->want_to_quit) {
+ struct reuse_tcp* reuse = NULL;
w = outnet->tcp_wait_first;
outnet->tcp_wait_first = w->next_waiting;
if(outnet->tcp_wait_last == w)
outnet->tcp_wait_last = NULL;
- if(!outnet_tcp_take_into_use(w, w->pkt, w->pkt_len)) {
- comm_point_callback_type* cb = w->cb;
- void* cb_arg = w->cb_arg;
- waiting_tcp_delete(w);
- fptr_ok(fptr_whitelist_pending_tcp(cb));
- (void)(*cb)(NULL, cb_arg, NETEVENT_CLOSED, NULL);
+ w->on_tcp_waiting_list = 0;
+ reuse = reuse_tcp_find(outnet, &w->addr, w->addrlen,
+ w->ssl_upstream);
+ if(reuse) {
+ log_reuse_tcp(VERB_CLIENT, "use free buffer for waiting tcp: "
+ "found reuse", reuse);
+ reuse_tcp_lru_touch(outnet, reuse);
+ comm_timer_disable(w->timer);
+ w->next_waiting = (void*)reuse->pending;
+ reuse_tree_by_id_insert(reuse, w);
+ if(reuse->pending->query) {
+ /* on the write wait list */
+ reuse_write_wait_push_back(reuse, w);
+ } else {
+ /* write straight away */
+ /* stop the timer on read of the fd */
+ comm_point_stop_listening(reuse->pending->c);
+ reuse->pending->query = w;
+ outnet_tcp_take_query_setup(
+ reuse->pending->c->fd, reuse->pending,
+ w);
+ }
+ } else {
+ struct pending_tcp* pend = w->outnet->tcp_free;
+ rbtree_init(&pend->reuse.tree_by_id, reuse_id_cmp);
+ pend->reuse.pending = pend;
+ memcpy(&pend->reuse.addr, &w->addr, w->addrlen);
+ pend->reuse.addrlen = w->addrlen;
+ if(!outnet_tcp_take_into_use(w)) {
+ waiting_tcp_callback(w, NULL, NETEVENT_CLOSED,
+ NULL);
+ waiting_tcp_delete(w);
+ }
+ }
+ }
+}
+
+/** add waiting_tcp element to the outnet tcp waiting list */
+static void
+outnet_add_tcp_waiting(struct outside_network* outnet, struct waiting_tcp* w)
+{
+ struct timeval tv;
+ if(w->on_tcp_waiting_list)
+ return;
+ w->next_waiting = NULL;
+ if(outnet->tcp_wait_last)
+ outnet->tcp_wait_last->next_waiting = w;
+ else outnet->tcp_wait_first = w;
+ outnet->tcp_wait_last = w;
+ w->on_tcp_waiting_list = 1;
+#ifndef S_SPLINT_S
+ tv.tv_sec = w->timeout/1000;
+ tv.tv_usec = (w->timeout%1000)*1000;
+#endif
+ comm_timer_set(w->timer, &tv);
+}
+
+/** delete element from tree by id */
+static void
+reuse_tree_by_id_delete(struct reuse_tcp* reuse, struct waiting_tcp* w)
+{
+ log_assert(w->id_node.key != NULL);
+ rbtree_delete(&reuse->tree_by_id, w);
+ w->id_node.key = NULL;
+}
+
+/** move writewait list to go for another connection. */
+static void
+reuse_move_writewait_away(struct outside_network* outnet,
+ struct pending_tcp* pend)
+{
+ /* the writewait list has not been written yet, so if the
+ * stream was closed, they have not actually been failed, only
+ * the queries written. Other queries can get written to another
+ * stream. For upstreams that do not support multiple queries
+ * and answers, the stream can get closed, and then the queries
+ * can get written on a new socket */
+ struct waiting_tcp* w;
+ if(pend->query && pend->query->error_count == 0 &&
+ pend->c->tcp_write_pkt == pend->query->pkt &&
+ pend->c->tcp_write_pkt_len == pend->query->pkt_len) {
+ /* since the current query is not written, it can also
+ * move to a free buffer */
+ if(verbosity >= VERB_CLIENT && pend->query->pkt_len > 12+2+2 &&
+ LDNS_QDCOUNT(pend->query->pkt) > 0 &&
+ dname_valid(pend->query->pkt+12, pend->query->pkt_len-12)) {
+ char buf[LDNS_MAX_DOMAINLEN+1];
+ dname_str(pend->query->pkt+12, buf);
+ verbose(VERB_CLIENT, "reuse_move_writewait_away current %s %d bytes were written",
+ buf, (int)pend->c->tcp_write_byte_count);
+ }
+ pend->c->tcp_write_pkt = NULL;
+ pend->c->tcp_write_pkt_len = 0;
+ pend->c->tcp_write_and_read = 0;
+ pend->reuse.cp_more_read_again = 0;
+ pend->reuse.cp_more_write_again = 0;
+ pend->c->tcp_is_reading = 1;
+ w = pend->query;
+ pend->query = NULL;
+ /* increase error count, so that if the next socket fails too
+ * the server selection is run again with this query failed
+ * and it can select a different server (if possible), or
+ * fail the query */
+ w->error_count ++;
+ reuse_tree_by_id_delete(&pend->reuse, w);
+ outnet_add_tcp_waiting(outnet, w);
+ }
+ while((w = reuse_write_wait_pop(&pend->reuse)) != NULL) {
+ if(verbosity >= VERB_CLIENT && w->pkt_len > 12+2+2 &&
+ LDNS_QDCOUNT(w->pkt) > 0 &&
+ dname_valid(w->pkt+12, w->pkt_len-12)) {
+ char buf[LDNS_MAX_DOMAINLEN+1];
+ dname_str(w->pkt+12, buf);
+ verbose(VERB_CLIENT, "reuse_move_writewait_away item %s", buf);
+ }
+ reuse_tree_by_id_delete(&pend->reuse, w);
+ outnet_add_tcp_waiting(outnet, w);
+ }
+}
+
+/** remove reused element from tree and lru list */
+static void
+reuse_tcp_remove_tree_list(struct outside_network* outnet,
+ struct reuse_tcp* reuse)
+{
+ verbose(VERB_CLIENT, "reuse_tcp_remove_tree_list");
+ if(reuse->node.key) {
+ /* delete it from reuse tree */
+ (void)rbtree_delete(&outnet->tcp_reuse, &reuse->node);
+ reuse->node.key = NULL;
+ }
+ /* delete from reuse list */
+ if(reuse->item_on_lru_list) {
+ if(reuse->lru_prev) {
+ /* assert that members of the lru list are waiting
+ * and thus have a pending pointer to the struct */
+ log_assert(reuse->lru_prev->pending);
+ reuse->lru_prev->lru_next = reuse->lru_next;
+ } else {
+ log_assert(!reuse->lru_next || reuse->lru_next->pending);
+ outnet->tcp_reuse_first = reuse->lru_next;
+ }
+ if(reuse->lru_next) {
+ /* assert that members of the lru list are waiting
+ * and thus have a pending pointer to the struct */
+ log_assert(reuse->lru_next->pending);
+ reuse->lru_next->lru_prev = reuse->lru_prev;
+ } else {
+ log_assert(!reuse->lru_prev || reuse->lru_prev->pending);
+ outnet->tcp_reuse_last = reuse->lru_prev;
}
+ reuse->item_on_lru_list = 0;
}
}
+/** helper function that deletes an element from the tree of readwait
+ * elements in tcp reuse structure */
+static void reuse_del_readwait_elem(rbnode_type* node, void* ATTR_UNUSED(arg))
+{
+ struct waiting_tcp* w = (struct waiting_tcp*)node->key;
+ waiting_tcp_delete(w);
+}
+
+/** delete readwait waiting_tcp elements, deletes the elements in the list */
+void reuse_del_readwait(rbtree_type* tree_by_id)
+{
+ if(tree_by_id->root == NULL ||
+ tree_by_id->root == RBTREE_NULL)
+ return;
+ traverse_postorder(tree_by_id, &reuse_del_readwait_elem, NULL);
+ rbtree_init(tree_by_id, reuse_id_cmp);
+}
+
/** decommission a tcp buffer, closes commpoint and frees waiting_tcp entry */
static void
decommission_pending_tcp(struct outside_network* outnet,
struct pending_tcp* pend)
{
+ verbose(VERB_CLIENT, "decommission_pending_tcp");
+ pend->next_free = outnet->tcp_free;
+ outnet->tcp_free = pend;
+ if(pend->reuse.node.key) {
+ /* needs unlink from the reuse tree to get deleted */
+ reuse_tcp_remove_tree_list(outnet, &pend->reuse);
+ }
+ /* free SSL structure after remove from outnet tcp reuse tree,
+ * because the c->ssl null or not is used for sorting in the tree */
if(pend->c->ssl) {
#ifdef HAVE_SSL
SSL_shutdown(pend->c->ssl);
@@ -435,11 +914,68 @@ decommission_pending_tcp(struct outside_network* outnet,
#endif
}
comm_point_close(pend->c);
- pend->next_free = outnet->tcp_free;
- outnet->tcp_free = pend;
- waiting_tcp_delete(pend->query);
+ pend->reuse.cp_more_read_again = 0;
+ pend->reuse.cp_more_write_again = 0;
+ /* unlink the query and writewait list, it is part of the tree
+ * nodes and is deleted */
pend->query = NULL;
- use_free_buffer(outnet);
+ pend->reuse.write_wait_first = NULL;
+ pend->reuse.write_wait_last = NULL;
+ reuse_del_readwait(&pend->reuse.tree_by_id);
+}
+
+/** perform failure callbacks for waiting queries in reuse read rbtree */
+static void reuse_cb_readwait_for_failure(rbtree_type* tree_by_id, int err)
+{
+ rbnode_type* node;
+ if(tree_by_id->root == NULL ||
+ tree_by_id->root == RBTREE_NULL)
+ return;
+ node = rbtree_first(tree_by_id);
+ while(node && node != RBTREE_NULL) {
+ struct waiting_tcp* w = (struct waiting_tcp*)node->key;
+ waiting_tcp_callback(w, NULL, err, NULL);
+ node = rbtree_next(node);
+ }
+}
+
+/** perform callbacks for failure and also decommission pending tcp.
+ * the callbacks remove references in sq->pending to the waiting_tcp
+ * members of the tree_by_id in the pending tcp. The pending_tcp is
+ * removed before the callbacks, so that the callbacks do not modify
+ * the pending_tcp due to its reference in the outside_network reuse tree */
+static void reuse_cb_and_decommission(struct outside_network* outnet,
+ struct pending_tcp* pend, int error)
+{
+ rbtree_type store;
+ store = pend->reuse.tree_by_id;
+ pend->query = NULL;
+ rbtree_init(&pend->reuse.tree_by_id, reuse_id_cmp);
+ pend->reuse.write_wait_first = NULL;
+ pend->reuse.write_wait_last = NULL;
+ decommission_pending_tcp(outnet, pend);
+ reuse_cb_readwait_for_failure(&store, error);
+ reuse_del_readwait(&store);
+}
+
+/** set timeout on tcp fd and setup read event to catch incoming dns msgs */
+static void
+reuse_tcp_setup_timeout(struct pending_tcp* pend_tcp)
+{
+ log_reuse_tcp(VERB_CLIENT, "reuse_tcp_setup_timeout", &pend_tcp->reuse);
+ comm_point_start_listening(pend_tcp->c, -1, REUSE_TIMEOUT);
+}
+
+/** set timeout on tcp fd and setup read event to catch incoming dns msgs */
+static void
+reuse_tcp_setup_read_and_timeout(struct pending_tcp* pend_tcp)
+{
+ log_reuse_tcp(VERB_CLIENT, "reuse_tcp_setup_readtimeout", &pend_tcp->reuse);
+ sldns_buffer_clear(pend_tcp->c->buffer);
+ pend_tcp->c->tcp_is_reading = 1;
+ pend_tcp->c->tcp_byte_count = 0;
+ comm_point_stop_listening(pend_tcp->c);
+ comm_point_start_listening(pend_tcp->c, -1, REUSE_TIMEOUT);
}
int
@@ -447,24 +983,116 @@ outnet_tcp_cb(struct comm_point* c, void* arg, int error,
struct comm_reply *reply_info)
{
struct pending_tcp* pend = (struct pending_tcp*)arg;
- struct outside_network* outnet = pend->query->outnet;
+ struct outside_network* outnet = pend->reuse.outnet;
+ struct waiting_tcp* w = NULL;
verbose(VERB_ALGO, "outnettcp cb");
- if(error != NETEVENT_NOERROR) {
+ if(error == NETEVENT_TIMEOUT) {
+ if(pend->c->tcp_write_and_read) {
+ verbose(VERB_QUERY, "outnettcp got tcp timeout "
+ "for read, ignored because write underway");
+ /* if we are writing, ignore readtimer, wait for write timer
+ * or write is done */
+ return 0;
+ } else {
+ verbose(VERB_QUERY, "outnettcp got tcp timeout %s",
+ (pend->reuse.tree_by_id.count?"for reading pkt":
+ "for keepalive for reuse"));
+ }
+ /* must be timeout for reading or keepalive reuse,
+ * close it. */
+ reuse_tcp_remove_tree_list(outnet, &pend->reuse);
+ } else if(error == NETEVENT_PKT_WRITTEN) {
+ /* the packet we want to write has been written. */
+ verbose(VERB_ALGO, "outnet tcp pkt was written event");
+ log_assert(c == pend->c);
+ log_assert(pend->query->pkt == pend->c->tcp_write_pkt);
+ log_assert(pend->query->pkt_len == pend->c->tcp_write_pkt_len);
+ pend->c->tcp_write_pkt = NULL;
+ pend->c->tcp_write_pkt_len = 0;
+ /* the pend.query is already in tree_by_id */
+ log_assert(pend->query->id_node.key);
+ pend->query = NULL;
+ /* setup to write next packet or setup read timeout */
+ if(pend->reuse.write_wait_first) {
+ verbose(VERB_ALGO, "outnet tcp setup next pkt");
+ /* we can write it straight away perhaps, set flag
+ * because this callback called after a tcp write
+ * succeeded and likely more buffer space is available
+ * and we can write some more. */
+ pend->reuse.cp_more_write_again = 1;
+ pend->query = reuse_write_wait_pop(&pend->reuse);
+ comm_point_stop_listening(pend->c);
+ outnet_tcp_take_query_setup(pend->c->fd, pend,
+ pend->query);
+ } else {
+ verbose(VERB_ALGO, "outnet tcp writes done, wait");
+ pend->c->tcp_write_and_read = 0;
+ pend->reuse.cp_more_read_again = 0;
+ pend->reuse.cp_more_write_again = 0;
+ pend->c->tcp_is_reading = 1;
+ comm_point_stop_listening(pend->c);
+ reuse_tcp_setup_timeout(pend);
+ }
+ return 0;
+ } else if(error != NETEVENT_NOERROR) {
verbose(VERB_QUERY, "outnettcp got tcp error %d", error);
+ reuse_move_writewait_away(outnet, pend);
/* pass error below and exit */
} else {
/* check ID */
- if(sldns_buffer_limit(c->buffer) < sizeof(uint16_t) ||
- LDNS_ID_WIRE(sldns_buffer_begin(c->buffer))!=pend->id) {
+ if(sldns_buffer_limit(c->buffer) < sizeof(uint16_t)) {
log_addr(VERB_QUERY,
- "outnettcp: bad ID in reply, from:",
- &pend->query->addr, pend->query->addrlen);
+ "outnettcp: bad ID in reply, too short, from:",
+ &pend->reuse.addr, pend->reuse.addrlen);
error = NETEVENT_CLOSED;
+ } else {
+ uint16_t id = LDNS_ID_WIRE(sldns_buffer_begin(
+ c->buffer));
+ /* find the query the reply is for */
+ w = reuse_tcp_by_id_find(&pend->reuse, id);
}
}
- fptr_ok(fptr_whitelist_pending_tcp(pend->query->cb));
- (void)(*pend->query->cb)(c, pend->query->cb_arg, error, reply_info);
- decommission_pending_tcp(outnet, pend);
+ if(error == NETEVENT_NOERROR && !w) {
+ /* no struct waiting found in tree, no reply to call */
+ log_addr(VERB_QUERY, "outnettcp: bad ID in reply, from:",
+ &pend->reuse.addr, pend->reuse.addrlen);
+ error = NETEVENT_CLOSED;
+ }
+ if(error == NETEVENT_NOERROR) {
+ /* add to reuse tree so it can be reused, if not a failure.
+ * This is possible if the state machine wants to make a tcp
+ * query again to the same destination. */
+ if(outnet->tcp_reuse.count < outnet->tcp_reuse_max) {
+ (void)reuse_tcp_insert(outnet, pend);
+ }
+ }
+ if(w) {
+ reuse_tree_by_id_delete(&pend->reuse, w);
+ verbose(VERB_CLIENT, "outnet tcp callback query err %d buflen %d",
+ error, (int)sldns_buffer_limit(c->buffer));
+ waiting_tcp_callback(w, c, error, reply_info);
+ waiting_tcp_delete(w);
+ }
+ verbose(VERB_CLIENT, "outnet_tcp_cb reuse after cb");
+ if(error == NETEVENT_NOERROR && pend->reuse.node.key) {
+ verbose(VERB_CLIENT, "outnet_tcp_cb reuse after cb: keep it");
+ /* it is in the reuse_tcp tree, with other queries, or
+ * on the empty list. do not decommission it */
+ /* if there are more outstanding queries, we could try to
+ * read again, to see if it is on the input,
+ * because this callback called after a successful read
+ * and there could be more bytes to read on the input */
+ if(pend->reuse.tree_by_id.count != 0)
+ pend->reuse.cp_more_read_again = 1;
+ reuse_tcp_setup_read_and_timeout(pend);
+ return 0;
+ }
+ verbose(VERB_CLIENT, "outnet_tcp_cb reuse after cb: decommission it");
+ /* no queries on it, no space to keep it. or timeout or closed due
+ * to error. Close it */
+ reuse_cb_and_decommission(outnet, pend, (error==NETEVENT_TIMEOUT?
+ NETEVENT_TIMEOUT:NETEVENT_CLOSED));
+ use_free_buffer(outnet);
return 0;
}
@@ -723,7 +1351,8 @@ outside_network_create(struct comm_base *base, size_t bufsize,
struct ub_randstate* rnd, int use_caps_for_id, int* availports,
int numavailports, size_t unwanted_threshold, int tcp_mss,
void (*unwanted_action)(void*), void* unwanted_param, int do_udp,
- void* sslctx, int delayclose, int tls_use_sni, struct dt_env* dtenv)
+ void* sslctx, int delayclose, int tls_use_sni, struct dt_env* dtenv,
+ int udp_connect)
{
struct outside_network* outnet = (struct outside_network*)
calloc(1, sizeof(struct outside_network));
@@ -761,6 +1390,9 @@ outside_network_create(struct comm_base *base, size_t bufsize,
outnet->delay_tv.tv_usec = (delayclose%1000)*1000;
}
#endif
+ if(udp_connect) {
+ outnet->udp_connect = 1;
+ }
if(numavailports == 0 || num_ports == 0) {
log_err("no outgoing ports available");
outside_network_delete(outnet);
@@ -795,6 +1427,8 @@ outside_network_create(struct comm_base *base, size_t bufsize,
outside_network_delete(outnet);
return NULL;
}
+ rbtree_init(&outnet->tcp_reuse, reuse_cmp);
+ outnet->tcp_reuse_max = num_tcp;
/* allocate commpoints */
for(k=0; k<num_ports; k++) {
@@ -958,6 +1592,17 @@ outside_network_delete(struct outside_network* outnet)
size_t i;
for(i=0; i<outnet->num_tcp; i++)
if(outnet->tcp_conns[i]) {
+ if(outnet->tcp_conns[i]->query &&
+ !outnet->tcp_conns[i]->query->
+ on_tcp_waiting_list) {
+ /* delete waiting_tcp elements that
+ * the tcp conn is working on */
+ struct pending_tcp* pend =
+ (struct pending_tcp*)outnet->
+ tcp_conns[i]->query->
+ next_waiting;
+ decommission_pending_tcp(outnet, pend);
+ }
comm_point_delete(outnet->tcp_conns[i]->c);
waiting_tcp_delete(outnet->tcp_conns[i]->query);
free(outnet->tcp_conns[i]);
@@ -972,6 +1617,10 @@ outside_network_delete(struct outside_network* outnet)
p = np;
}
}
+ /* was allocated in struct pending that was deleted above */
+ rbtree_init(&outnet->tcp_reuse, reuse_cmp);
+ outnet->tcp_reuse_first = NULL;
+ outnet->tcp_reuse_last = NULL;
if(outnet->udp_wait_first) {
struct pending* p = outnet->udp_wait_first, *np;
while(p) {
@@ -1115,13 +1764,26 @@ select_ifport(struct outside_network* outnet, struct pending* pend,
my_if = ub_random_max(outnet->rnd, num_if);
pif = &ifs[my_if];
#ifndef DISABLE_EXPLICIT_PORT_RANDOMISATION
- my_port = ub_random_max(outnet->rnd, pif->avail_total);
- if(my_port < pif->inuse) {
- /* port already open */
- pend->pc = pif->out[my_port];
- verbose(VERB_ALGO, "using UDP if=%d port=%d",
- my_if, pend->pc->number);
- break;
+ if(outnet->udp_connect) {
+ /* if we connect() we cannot reuse fds for a port */
+ if(pif->inuse >= pif->avail_total) {
+ tries++;
+ if(tries < MAX_PORT_RETRY)
+ continue;
+ log_err("failed to find an open port, drop msg");
+ return 0;
+ }
+ my_port = pif->inuse + ub_random_max(outnet->rnd,
+ pif->avail_total - pif->inuse);
+ } else {
+ my_port = ub_random_max(outnet->rnd, pif->avail_total);
+ if(my_port < pif->inuse) {
+ /* port already open */
+ pend->pc = pif->out[my_port];
+ verbose(VERB_ALGO, "using UDP if=%d port=%d",
+ my_if, pend->pc->number);
+ break;
+ }
}
/* try to open new port, if fails, loop to try again */
log_assert(pif->inuse < pif->maxout);
@@ -1138,6 +1800,17 @@ select_ifport(struct outside_network* outnet, struct pending* pend,
if(fd != -1) {
verbose(VERB_ALGO, "opened UDP if=%d port=%d",
my_if, portno);
+ if(outnet->udp_connect) {
+ /* connect() to the destination */
+ if(connect(fd, (struct sockaddr*)&pend->addr,
+ pend->addrlen) < 0) {
+ log_err_addr("udp connect failed",
+ strerror(errno), &pend->addr,
+ pend->addrlen);
+ sock_close(fd);
+ return 0;
+ }
+ }
/* grab fd */
pend->pc = outnet->unused_fds;
outnet->unused_fds = pend->pc->next;
@@ -1197,10 +1870,17 @@ randomize_and_send_udp(struct pending* pend, sldns_buffer* packet, int timeout)
log_assert(pend->pc && pend->pc->cp);
/* send it over the commlink */
- if(!comm_point_send_udp_msg(pend->pc->cp, packet,
- (struct sockaddr*)&pend->addr, pend->addrlen)) {
- portcomm_loweruse(outnet, pend->pc);
- return 0;
+ if(outnet->udp_connect) {
+ if(!comm_point_send_udp_msg(pend->pc->cp, packet, NULL, 0)) {
+ portcomm_loweruse(outnet, pend->pc);
+ return 0;
+ }
+ } else {
+ if(!comm_point_send_udp_msg(pend->pc->cp, packet,
+ (struct sockaddr*)&pend->addr, pend->addrlen)) {
+ portcomm_loweruse(outnet, pend->pc);
+ return 0;
+ }
}
/* system calls to set timeout after sending UDP to make roundtrip
@@ -1273,45 +1953,152 @@ outnet_tcptimer(void* arg)
{
struct waiting_tcp* w = (struct waiting_tcp*)arg;
struct outside_network* outnet = w->outnet;
- comm_point_callback_type* cb;
- void* cb_arg;
- if(w->pkt) {
+ verbose(VERB_CLIENT, "outnet_tcptimer");
+ if(w->on_tcp_waiting_list) {
/* it is on the waiting list */
waiting_list_remove(outnet, w);
+ waiting_tcp_callback(w, NULL, NETEVENT_TIMEOUT, NULL);
+ waiting_tcp_delete(w);
} else {
/* it was in use */
struct pending_tcp* pend=(struct pending_tcp*)w->next_waiting;
- if(pend->c->ssl) {
-#ifdef HAVE_SSL
- SSL_shutdown(pend->c->ssl);
- SSL_free(pend->c->ssl);
- pend->c->ssl = NULL;
-#endif
- }
- comm_point_close(pend->c);
- pend->query = NULL;
- pend->next_free = outnet->tcp_free;
- outnet->tcp_free = pend;
+ reuse_cb_and_decommission(outnet, pend, NETEVENT_TIMEOUT);
}
- cb = w->cb;
- cb_arg = w->cb_arg;
- waiting_tcp_delete(w);
- fptr_ok(fptr_whitelist_pending_tcp(cb));
- (void)(*cb)(NULL, cb_arg, NETEVENT_TIMEOUT, NULL);
use_free_buffer(outnet);
}
+/** close the oldest reuse_tcp connection to make a fd and struct pend
+ * available for a new stream connection */
+static void
+reuse_tcp_close_oldest(struct outside_network* outnet)
+{
+ struct pending_tcp* pend;
+ verbose(VERB_CLIENT, "reuse_tcp_close_oldest");
+ if(!outnet->tcp_reuse_last) return;
+ pend = outnet->tcp_reuse_last->pending;
+
+ /* snip off of LRU */
+ log_assert(pend->reuse.lru_next == NULL);
+ if(pend->reuse.lru_prev) {
+ outnet->tcp_reuse_last = pend->reuse.lru_prev;
+ pend->reuse.lru_prev->lru_next = NULL;
+ } else {
+ outnet->tcp_reuse_last = NULL;
+ outnet->tcp_reuse_first = NULL;
+ }
+ pend->reuse.item_on_lru_list = 0;
+
+ /* free up */
+ reuse_cb_and_decommission(outnet, pend, NETEVENT_CLOSED);
+}
+
+/** find spare ID value for reuse tcp stream. That is random and also does
+ * not collide with an existing query ID that is in use or waiting */
+uint16_t
+reuse_tcp_select_id(struct reuse_tcp* reuse, struct outside_network* outnet)
+{
+ uint16_t id = 0, curid, nextid;
+ const int try_random = 2000;
+ int i;
+ unsigned select, count, space;
+ rbnode_type* node;
+
+ /* make really sure the tree is not empty */
+ if(reuse->tree_by_id.count == 0) {
+ id = ((unsigned)ub_random(outnet->rnd)>>8) & 0xffff;
+ return id;
+ }
+
+ /* try to find random empty spots by picking them */
+ for(i = 0; i<try_random; i++) {
+ id = ((unsigned)ub_random(outnet->rnd)>>8) & 0xffff;
+ if(!reuse_tcp_by_id_find(reuse, id)) {
+ return id;
+ }
+ }
+
+ /* equally pick a random unused element from the tree that is
+ * not in use. Pick a the n-th index of an ununused number,
+ * then loop over the empty spaces in the tree and find it */
+ log_assert(reuse->tree_by_id.count < 0xffff);
+ select = ub_random_max(outnet->rnd, 0xffff - reuse->tree_by_id.count);
+ /* select value now in 0 .. num free - 1 */
+
+ count = 0; /* number of free spaces passed by */
+ node = rbtree_first(&reuse->tree_by_id);
+ log_assert(node && node != RBTREE_NULL); /* tree not empty */
+ /* see if select is before first node */
+ if(select < tree_by_id_get_id(node))
+ return select;
+ count += tree_by_id_get_id(node);
+ /* perhaps select is between nodes */
+ while(node && node != RBTREE_NULL) {
+ rbnode_type* next = rbtree_next(node);
+ if(next && next != RBTREE_NULL) {
+ curid = tree_by_id_get_id(node);
+ nextid = tree_by_id_get_id(next);
+ log_assert(curid < nextid);
+ if(curid != 0xffff && curid + 1 < nextid) {
+ /* space between nodes */
+ space = nextid - curid - 1;
+ log_assert(select >= count);
+ if(select < count + space) {
+ /* here it is */
+ return curid + 1 + (select - count);
+ }
+ count += space;
+ }
+ }
+ node = next;
+ }
+
+ /* select is after the last node */
+ /* count is the number of free positions before the nodes in the
+ * tree */
+ node = rbtree_last(&reuse->tree_by_id);
+ log_assert(node && node != RBTREE_NULL); /* tree not empty */
+ curid = tree_by_id_get_id(node);
+ log_assert(count + (0xffff-curid) + reuse->tree_by_id.count == 0xffff);
+ return curid + 1 + (select - count);
+}
+
struct waiting_tcp*
pending_tcp_query(struct serviced_query* sq, sldns_buffer* packet,
int timeout, comm_point_callback_type* callback, void* callback_arg)
{
struct pending_tcp* pend = sq->outnet->tcp_free;
+ struct reuse_tcp* reuse = NULL;
struct waiting_tcp* w;
- struct timeval tv;
- uint16_t id;
- /* if no buffer is free allocate space to store query */
+
+ verbose(VERB_CLIENT, "pending_tcp_query");
+ if(sldns_buffer_limit(packet) < sizeof(uint16_t)) {
+ verbose(VERB_ALGO, "pending tcp query with too short buffer < 2");
+ return NULL;
+ }
+
+ /* find out if a reused stream to the target exists */
+ /* if so, take it into use */
+ reuse = reuse_tcp_find(sq->outnet, &sq->addr, sq->addrlen,
+ sq->ssl_upstream);
+ if(reuse) {
+ log_reuse_tcp(VERB_CLIENT, "pending_tcp_query: found reuse", reuse);
+ log_assert(reuse->pending);
+ pend = reuse->pending;
+ reuse_tcp_lru_touch(sq->outnet, reuse);
+ }
+
+ /* if !pend but we have reuse streams, close a reuse stream
+ * to be able to open a new one to this target, no use waiting
+ * to reuse a file descriptor while another query needs to use
+ * that buffer and file descriptor now. */
+ if(!pend) {
+ reuse_tcp_close_oldest(sq->outnet);
+ pend = sq->outnet->tcp_free;
+ }
+
+ /* allocate space to store query */
w = (struct waiting_tcp*)malloc(sizeof(struct waiting_tcp)
- + (pend?0:sldns_buffer_limit(packet)));
+ + sldns_buffer_limit(packet));
if(!w) {
return NULL;
}
@@ -1319,47 +2106,76 @@ pending_tcp_query(struct serviced_query* sq, sldns_buffer* packet,
free(w);
return NULL;
}
- w->pkt = NULL;
- w->pkt_len = 0;
- id = ((unsigned)ub_random(sq->outnet->rnd)>>8) & 0xffff;
- LDNS_ID_SET(sldns_buffer_begin(packet), id);
+ w->pkt = (uint8_t*)w + sizeof(struct waiting_tcp);
+ w->pkt_len = sldns_buffer_limit(packet);
+ memmove(w->pkt, sldns_buffer_begin(packet), w->pkt_len);
+ if(reuse)
+ w->id = reuse_tcp_select_id(reuse, sq->outnet);
+ else w->id = ((unsigned)ub_random(sq->outnet->rnd)>>8) & 0xffff;
+ LDNS_ID_SET(w->pkt, w->id);
memcpy(&w->addr, &sq->addr, sq->addrlen);
w->addrlen = sq->addrlen;
w->outnet = sq->outnet;
+ w->on_tcp_waiting_list = 0;
+ w->next_waiting = NULL;
w->cb = callback;
w->cb_arg = callback_arg;
w->ssl_upstream = sq->ssl_upstream;
w->tls_auth_name = sq->tls_auth_name;
-#ifndef S_SPLINT_S
- tv.tv_sec = timeout/1000;
- tv.tv_usec = (timeout%1000)*1000;
-#endif
- comm_timer_set(w->timer, &tv);
+ w->timeout = timeout;
+ w->id_node.key = NULL;
+ w->write_wait_prev = NULL;
+ w->write_wait_next = NULL;
+ w->write_wait_queued = 0;
+ w->error_count = 0;
if(pend) {
/* we have a buffer available right now */
- if(!outnet_tcp_take_into_use(w, sldns_buffer_begin(packet),
- sldns_buffer_limit(packet))) {
- waiting_tcp_delete(w);
- return NULL;
+ if(reuse) {
+ /* reuse existing fd, write query and continue */
+ /* store query in tree by id */
+ verbose(VERB_CLIENT, "pending_tcp_query: reuse, store");
+ w->next_waiting = (void*)pend;
+ reuse_tree_by_id_insert(&pend->reuse, w);
+ /* can we write right now? */
+ if(pend->query == NULL) {
+ /* write straight away */
+ /* stop the timer on read of the fd */
+ comm_point_stop_listening(pend->c);
+ pend->query = w;
+ outnet_tcp_take_query_setup(pend->c->fd, pend,
+ w);
+ } else {
+ /* put it in the waiting list for
+ * this stream */
+ reuse_write_wait_push_back(&pend->reuse, w);
+ }
+ } else {
+ /* create new fd and connect to addr, setup to
+ * write query */
+ verbose(VERB_CLIENT, "pending_tcp_query: new fd, connect");
+ rbtree_init(&pend->reuse.tree_by_id, reuse_id_cmp);
+ pend->reuse.pending = pend;
+ memcpy(&pend->reuse.addr, &sq->addr, sq->addrlen);
+ pend->reuse.addrlen = sq->addrlen;
+ if(!outnet_tcp_take_into_use(w)) {
+ waiting_tcp_delete(w);
+ return NULL;
+ }
}
-#ifdef USE_DNSTAP
- if(sq->outnet->dtenv &&
- (sq->outnet->dtenv->log_resolver_query_messages ||
- sq->outnet->dtenv->log_forwarder_query_messages))
- dt_msg_send_outside_query(sq->outnet->dtenv, &sq->addr,
- comm_tcp, sq->zone, sq->zonelen, packet);
-#endif
} else {
/* queue up */
- w->pkt = (uint8_t*)w + sizeof(struct waiting_tcp);
- w->pkt_len = sldns_buffer_limit(packet);
- memmove(w->pkt, sldns_buffer_begin(packet), w->pkt_len);
- w->next_waiting = NULL;
- if(sq->outnet->tcp_wait_last)
- sq->outnet->tcp_wait_last->next_waiting = w;
- else sq->outnet->tcp_wait_first = w;
- sq->outnet->tcp_wait_last = w;
+ /* waiting for a buffer on the outside network buffer wait
+ * list */
+ verbose(VERB_CLIENT, "pending_tcp_query: queue to wait");
+ outnet_add_tcp_waiting(sq->outnet, w);
}
+#ifdef USE_DNSTAP
+ if(sq->outnet->dtenv &&
+ (sq->outnet->dtenv->log_resolver_query_messages ||
+ sq->outnet->dtenv->log_forwarder_query_messages))
+ dt_msg_send_outside_query(sq->outnet->dtenv, &sq->addr,
+ comm_tcp, sq->zone, sq->zonelen, packet);
+#endif
return w;
}
@@ -1477,6 +2293,7 @@ static void
waiting_list_remove(struct outside_network* outnet, struct waiting_tcp* w)
{
struct waiting_tcp* p = outnet->tcp_wait_first, *prev = NULL;
+ w->on_tcp_waiting_list = 0;
while(p) {
if(p == w) {
/* remove w */
@@ -1492,10 +2309,53 @@ waiting_list_remove(struct outside_network* outnet, struct waiting_tcp* w)
}
}
+/** reuse tcp stream, remove serviced query from stream,
+ * return true if the stream is kept, false if it is to be closed */
+static int
+reuse_tcp_remove_serviced_keep(struct waiting_tcp* w,
+ struct serviced_query* sq)
+{
+ struct pending_tcp* pend_tcp = (struct pending_tcp*)w->next_waiting;
+ verbose(VERB_CLIENT, "reuse_tcp_remove_serviced_keep");
+ /* remove the callback. let query continue to write to not cancel
+ * the stream itself. also keep it as an entry in the tree_by_id,
+ * in case the answer returns (that we no longer want), but we cannot
+ * pick the same ID number meanwhile */
+ w->cb = NULL;
+ /* see if can be entered in reuse tree
+ * for that the FD has to be non-1 */
+ if(pend_tcp->c->fd == -1) {
+ verbose(VERB_CLIENT, "reuse_tcp_remove_serviced_keep: -1 fd");
+ return 0;
+ }
+ /* if in tree and used by other queries */
+ if(pend_tcp->reuse.node.key) {
+ verbose(VERB_CLIENT, "reuse_tcp_remove_serviced_keep: in use by other queries");
+ /* do not reset the keepalive timer, for that
+ * we'd need traffic, and this is where the serviced is
+ * removed due to state machine internal reasons,
+ * eg. iterator no longer interested in this query */
+ return 1;
+ }
+ /* if still open and want to keep it open */
+ if(pend_tcp->c->fd != -1 && sq->outnet->tcp_reuse.count <
+ sq->outnet->tcp_reuse_max) {
+ verbose(VERB_CLIENT, "reuse_tcp_remove_serviced_keep: keep open");
+ /* set a keepalive timer on it */
+ if(!reuse_tcp_insert(sq->outnet, pend_tcp)) {
+ return 0;
+ }
+ reuse_tcp_setup_timeout(pend_tcp);
+ return 1;
+ }
+ return 0;
+}
+
/** cleanup serviced query entry */
static void
serviced_delete(struct serviced_query* sq)
{
+ verbose(VERB_CLIENT, "serviced_delete");
if(sq->pending) {
/* clear up the pending query */
if(sq->status == serviced_query_UDP_EDNS ||
@@ -1503,6 +2363,7 @@ serviced_delete(struct serviced_query* sq)
sq->status == serviced_query_UDP_EDNS_FRAG ||
sq->status == serviced_query_UDP_EDNS_fallback) {
struct pending* p = (struct pending*)sq->pending;
+ verbose(VERB_CLIENT, "serviced_delete: UDP");
if(p->pc)
portcomm_loweruse(sq->outnet, p->pc);
pending_delete(sq->outnet, p);
@@ -1510,14 +2371,32 @@ serviced_delete(struct serviced_query* sq)
* mesh */
outnet_send_wait_udp(sq->outnet);
} else {
- struct waiting_tcp* p = (struct waiting_tcp*)
+ struct waiting_tcp* w = (struct waiting_tcp*)
sq->pending;
- if(p->pkt == NULL) {
- decommission_pending_tcp(sq->outnet,
- (struct pending_tcp*)p->next_waiting);
+ verbose(VERB_CLIENT, "serviced_delete: TCP");
+ /* if on stream-write-waiting list then
+ * remove from waiting list and waiting_tcp_delete */
+ if(w->write_wait_queued) {
+ struct pending_tcp* pend =
+ (struct pending_tcp*)w->next_waiting;
+ verbose(VERB_CLIENT, "serviced_delete: writewait");
+ reuse_tree_by_id_delete(&pend->reuse, w);
+ reuse_write_wait_remove(&pend->reuse, w);
+ waiting_tcp_delete(w);
+ } else if(!w->on_tcp_waiting_list) {
+ struct pending_tcp* pend =
+ (struct pending_tcp*)w->next_waiting;
+ verbose(VERB_CLIENT, "serviced_delete: tcpreusekeep");
+ if(!reuse_tcp_remove_serviced_keep(w, sq)) {
+ reuse_cb_and_decommission(sq->outnet,
+ pend, NETEVENT_CLOSED);
+ use_free_buffer(sq->outnet);
+ }
+ sq->pending = NULL;
} else {
- waiting_list_remove(sq->outnet, p);
- waiting_tcp_delete(p);
+ verbose(VERB_CLIENT, "serviced_delete: tcpwait");
+ waiting_list_remove(sq->outnet, w);
+ waiting_tcp_delete(w);
}
}
}
@@ -2097,18 +2976,18 @@ outnet_serviced_query(struct outside_network* outnet,
{
struct serviced_query* sq;
struct service_callback* cb;
- struct edns_tag_addr* client_tag_addr;
+ struct edns_string_addr* client_string_addr;
if(!inplace_cb_query_call(env, qinfo, flags, addr, addrlen, zone, zonelen,
qstate, qstate->region))
return NULL;
- if((client_tag_addr = edns_tag_addr_lookup(&env->edns_tags->client_tags,
- addr, addrlen))) {
- uint16_t client_tag = htons(client_tag_addr->tag_data);
+ if((client_string_addr = edns_string_addr_lookup(
+ &env->edns_strings->client_strings, addr, addrlen))) {
edns_opt_list_append(&qstate->edns_opts_back_out,
- env->edns_tags->client_tag_opcode, 2,
- (uint8_t*)&client_tag, qstate->region);
+ env->edns_strings->client_string_opcode,
+ client_string_addr->string_len,
+ client_string_addr->string, qstate->region);
}
serviced_gen_query(buff, qinfo->qname, qinfo->qname_len, qinfo->qtype,
diff --git a/services/outside_network.h b/services/outside_network.h
index c8f6d5724a87..2fe97fa6c5c9 100644
--- a/services/outside_network.h
+++ b/services/outside_network.h
@@ -52,6 +52,7 @@ struct ub_randstate;
struct pending_tcp;
struct waiting_tcp;
struct waiting_udp;
+struct reuse_tcp;
struct infra_cache;
struct port_comm;
struct port_if;
@@ -106,6 +107,9 @@ struct outside_network {
int delayclose;
/** timeout for delayclose */
struct timeval delay_tv;
+ /** if we perform udp-connect, connect() for UDP socket to mitigate
+ * ICMP side channel leakage */
+ int udp_connect;
/** array of outgoing IP4 interfaces */
struct port_if* ip4_ifs;
@@ -154,6 +158,21 @@ struct outside_network {
size_t num_tcp;
/** number of tcp communication points in use. */
size_t num_tcp_outgoing;
+ /**
+ * tree of still-open and waiting tcp connections for reuse.
+ * can be closed and reopened to get a new tcp connection.
+ * or reused to the same destination again. with timeout to close.
+ * Entries are of type struct reuse_tcp.
+ * The entries are both active and empty connections.
+ */
+ rbtree_type tcp_reuse;
+ /** max number of tcp_reuse entries we want to keep open */
+ size_t tcp_reuse_max;
+ /** first and last(oldest) in lru list of reuse connections.
+ * the oldest can be closed to get a new free pending_tcp if needed
+ * The list contains empty connections, that wait for timeout or
+ * a new query that can use the existing connection. */
+ struct reuse_tcp* tcp_reuse_first, *tcp_reuse_last;
/** list of tcp comm points that are free for use */
struct pending_tcp* tcp_free;
/** list of tcp queries waiting for a buffer */
@@ -212,6 +231,76 @@ struct port_comm {
};
/**
+ * Reuse TCP connection, still open can be used again.
+ */
+struct reuse_tcp {
+ /** rbtree node with links in tcp_reuse tree. key is NULL when not
+ * in tree. Both active and empty connections are in the tree.
+ * key is a pointer to this structure, the members used to compare
+ * are the sockaddr and and then is-ssl bool, and then ptr value is
+ * used in case the same address exists several times in the tree
+ * when there are multiple connections to the same destination to
+ * make the rbtree items unique. */
+ rbnode_type node;
+ /** the key for the tcp_reuse tree. address of peer, ip4 or ip6,
+ * and port number of peer */
+ struct sockaddr_storage addr;
+ /** length of addr */
+ socklen_t addrlen;
+ /** also key for tcp_reuse tree, if ssl is used */
+ int is_ssl;
+ /** lru chain, so that the oldest can be removed to get a new
+ * connection when all are in (re)use. oldest is last in list.
+ * The lru only contains empty connections waiting for reuse,
+ * the ones with active queries are not on the list because they
+ * do not need to be closed to make space for others. They already
+ * service a query so the close for another query does not help
+ * service a larger number of queries. */
+ struct reuse_tcp* lru_next, *lru_prev;
+ /** true if the reuse_tcp item is on the lru list with empty items */
+ int item_on_lru_list;
+ /** the connection to reuse, the fd is non-1 and is open.
+ * the addr and port determine where the connection is going,
+ * and is key to the rbtree. The SSL ptr determines if it is
+ * a TLS connection or a plain TCP connection there. And TLS
+ * or not is also part of the key to the rbtree.
+ * There is a timeout and read event on the fd, to close it. */
+ struct pending_tcp* pending;
+ /**
+ * The more read again value pointed to by the commpoint
+ * tcp_more_read_again pointer, so that it exists after commpoint
+ * delete
+ */
+ int cp_more_read_again;
+ /**
+ * The more write again value pointed to by the commpoint
+ * tcp_more_write_again pointer, so that it exists after commpoint
+ * delete
+ */
+ int cp_more_write_again;
+ /** rbtree with other queries waiting on the connection, by ID number,
+ * of type struct waiting_tcp. It is for looking up received
+ * answers to the structure for callback. And also to see if ID
+ * numbers are unused and can be used for a new query.
+ * The write_wait elements are also in the tree, so that ID numbers
+ * can be looked up also for them. They are bool write_wait_queued. */
+ rbtree_type tree_by_id;
+ /** list of queries waiting to be written on the channel,
+ * if NULL no queries are waiting to be written and the pending->query
+ * is the query currently serviced. The first is the next in line.
+ * They are also in the tree_by_id. Once written, the are removed
+ * from this list, but stay in the tree. */
+ struct waiting_tcp* write_wait_first, *write_wait_last;
+ /** the outside network it is part of */
+ struct outside_network* outnet;
+};
+
+/** max number of queries on a reuse connection */
+#define MAX_REUSE_TCP_QUERIES 200
+/** timeout for REUSE entries in milliseconds. */
+#define REUSE_TIMEOUT 60000
+
+/**
* A query that has an answer pending for it.
*/
struct pending {
@@ -255,12 +344,15 @@ struct pending {
struct pending_tcp {
/** next in list of free tcp comm points, or NULL. */
struct pending_tcp* next_free;
- /** the ID for the query; checked in reply */
- uint16_t id;
/** tcp comm point it was sent on (and reply must come back on). */
struct comm_point* c;
/** the query being serviced, NULL if the pending_tcp is unused. */
struct waiting_tcp* query;
+ /** the pre-allocated reuse tcp structure. if ->pending is nonNULL
+ * it is in use and the connection is waiting for reuse.
+ * It is here for memory pre-allocation, and used to make this
+ * pending_tcp wait for reuse. */
+ struct reuse_tcp reuse;
};
/**
@@ -269,12 +361,27 @@ struct pending_tcp {
struct waiting_tcp {
/**
* next in waiting list.
- * if pkt==0, this points to the pending_tcp structure.
+ * if on_tcp_waiting_list==0, this points to the pending_tcp structure.
*/
struct waiting_tcp* next_waiting;
+ /** if true the item is on the tcp waiting list and next_waiting
+ * is used for that. If false, the next_waiting points to the
+ * pending_tcp */
+ int on_tcp_waiting_list;
+ /** next and prev in query waiting list for stream connection */
+ struct waiting_tcp* write_wait_prev, *write_wait_next;
+ /** true if the waiting_tcp structure is on the write_wait queue */
+ int write_wait_queued;
+ /** entry in reuse.tree_by_id, if key is NULL, not in tree, otherwise,
+ * this struct is key and sorted by ID (from waiting_tcp.id). */
+ rbnode_type id_node;
+ /** the ID for the query; checked in reply */
+ uint16_t id;
/** timeout event; timer keeps running whether the query is
* waiting for a buffer or the tcp reply is pending */
struct comm_timer* timer;
+ /** timeout in msec */
+ int timeout;
/** the outside network it is part of */
struct outside_network* outnet;
/** remote address. */
@@ -284,13 +391,14 @@ struct waiting_tcp {
/**
* The query itself, the query packet to send.
* allocated after the waiting_tcp structure.
- * set to NULL when the query is serviced and it part of pending_tcp.
- * if this is NULL, the next_waiting points to the pending_tcp.
*/
uint8_t* pkt;
/** length of query packet. */
size_t pkt_len;
- /** callback for the timeout, error or reply to the message */
+ /** callback for the timeout, error or reply to the message,
+ * or NULL if no user is waiting. the entry uses an ID number.
+ * a query that was written is no longer needed, but the ID number
+ * and a reply will come back and can be ignored if NULL */
comm_point_callback_type* cb;
/** callback user argument */
void* cb_arg;
@@ -298,6 +406,8 @@ struct waiting_tcp {
int ssl_upstream;
/** ref to the tls_auth_name from the serviced_query */
char* tls_auth_name;
+ /** the packet was involved in an error, to stop looping errors */
+ int error_count;
};
/**
@@ -421,6 +531,7 @@ struct serviced_query {
* msec to wait on timeouted udp sockets.
* @param tls_use_sni: if SNI is used for TLS connections.
* @param dtenv: environment to send dnstap events with (if enabled).
+ * @param udp_connect: if the udp_connect option is enabled.
* @return: the new structure (with no pending answers) or NULL on error.
*/
struct outside_network* outside_network_create(struct comm_base* base,
@@ -429,7 +540,8 @@ struct outside_network* outside_network_create(struct comm_base* base,
struct ub_randstate* rnd, int use_caps_for_id, int* availports,
int numavailports, size_t unwanted_threshold, int tcp_mss,
void (*unwanted_action)(void*), void* unwanted_param, int do_udp,
- void* sslctx, int delayclose, int tls_use_sni, struct dt_env *dtenv);
+ void* sslctx, int delayclose, int tls_use_sni, struct dt_env *dtenv,
+ int udp_connect);
/**
* Delete outside_network structure.
@@ -546,6 +658,19 @@ size_t outnet_get_mem(struct outside_network* outnet);
*/
size_t serviced_get_mem(struct serviced_query* sq);
+/** Pick random ID value for a tcp stream, avoids existing IDs. */
+uint16_t reuse_tcp_select_id(struct reuse_tcp* reuse,
+ struct outside_network* outnet);
+
+/** find element in tree by id */
+struct waiting_tcp* reuse_tcp_by_id_find(struct reuse_tcp* reuse, uint16_t id);
+
+/** insert element in tree by id */
+void reuse_tree_by_id_insert(struct reuse_tcp* reuse, struct waiting_tcp* w);
+
+/** delete readwait waiting_tcp elements, deletes the elements in the list */
+void reuse_del_readwait(rbtree_type* tree_by_id);
+
/** get TCP file descriptor for address, returns -1 on failure,
* tcp_mss is 0 or maxseg size to set for TCP packets. */
int outnet_get_tcp_fd(struct sockaddr_storage* addr, socklen_t addrlen, int tcp_mss, int dscp);
@@ -643,4 +768,10 @@ int pending_cmp(const void* key1, const void* key2);
/** compare function of serviced query rbtree */
int serviced_cmp(const void* key1, const void* key2);
+/** compare function of reuse_tcp rbtree in outside_network struct */
+int reuse_cmp(const void* key1, const void* key2);
+
+/** compare function of reuse_tcp tree_by_id rbtree */
+int reuse_id_cmp(const void* key1, const void* key2);
+
#endif /* OUTSIDE_NETWORK_H */
diff --git a/services/rpz.c b/services/rpz.c
index ba5dd186daad..13304652cc02 100644
--- a/services/rpz.c
+++ b/services/rpz.c
@@ -440,6 +440,8 @@ err:
respip_set_delete(r->respip_set);
if(r->taglist)
free(r->taglist);
+ if(r->region)
+ regional_destroy(r->region);
free(r);
}
return NULL;
generated by cgit v1.3 (git 2.53.0) at 2026-09-22 03:13:22 +0000