Pluggable per-entry params (delay/allow)

CMakeLists.txt

@@ -84,6 +84,7 @@ add_executable(rbldnsd
         rbldnsd_dnhash.c
         rbldnsd_dnhash_fixed.c
         rbldnsd_dnset.c
+        rbldnsd_entryparams.c
         rbldnsd_generic.c
         rbldnsd_ip4set.c
         rbldnsd_ip4trie.c

rbldnsd.8

@@ -442,6 +442,43 @@ and usage examples.
 Set the base template for all individual TXT records.  See section
 "Resulting A values and TXT templates" below for more information.
 
+.SS "Per-entry parameters"
+.PP
+Some dataset types support optional per-entry parameters.  Parameters are
+introduced by a \fB@\fR token following the entry value and consist of
+key/value pairs separated by \fB:\fR or \fB;\fR.
+.PP
+Supported dataset types:
+.IP \(bu
+\fBip4trie\fR
+.IP \(bu
+\fBip6trie\fR
+.IP \(bu
+\fBdnhash\fR
+.PP
+Examples:
+.nf
+  1.2.3.0/24 :127.0.0.2:Listed @ delay=2s
+  example.tld TXT "Listed" @ ts=1730000000;delay=60s
+.fi
+.PP
+The following built-in parameters are currently recognized:
+.IP "\fBdelay=\fR\fIduration\fR"
+Delay sending the reply for a matching entry.  The value uses the same
+syntax as \fBparse_time()\fR (e.g. \fB60\fR, \fB60s\fR, \fB10m\fR, \fB1h\fR).
+.IP "\fBts=\fR\fIunix_timestamp\fR"
+UNIX timestamp in seconds.  When used together with \fBdelay\fR, the entry
+is suppressed until the cooling time is over, i.e. the entry is returned only
+when \fBnow >= ts + delay\fR.
+.IP "\fBkey=require\fR"
+Return the entry only for \fIkeyed\fR requests.  A request is considered keyed
+when the zone has an \fBaclkey\fR dataset configured and a valid key label is
+present in the queried name (\fBaclkey\fR removes that label before the actual
+dataset lookup).
+.IP "\fBkey=nodelay\fR"
+For keyed requests, bypass both reply delaying (\fBdelay\fR without \fBts\fR)
+and cooldown suppression (\fBts\fR+\fBdelay\fR) for the matching entry.
+
 .SS "ip4set Dataset"
 .PP
 A set of IP addresses or CIDR address ranges, together with A and

rbldnsd.c

@@ -1232,6 +1232,149 @@ static int do_reload(int do_fork, struct ev_loop *loop) {
 #define MSGVEC_LEN 1
 #endif
 
+struct delayed_reply {
+  ev_tstamp at;
+  int fd;
+  struct sockaddr_storage peer;
+  socklen_t peerlen;
+  unsigned char *buf;
+  size_t len;
+};
+
+static struct delayed_reply *delayed_heap;
+static unsigned delayed_heap_len;
+static unsigned delayed_heap_cap;
+static size_t delayed_bytes;
+static ev_timer delayed_timer;
+static struct ev_loop *delayed_loop;
+
+#define DELAYED_MAX_ITEMS 8192
+#define DELAYED_MAX_BYTES (32u * 1024u * 1024u)
+
+static inline void delayed_swap(unsigned a, unsigned b) {
+  struct delayed_reply tmp = delayed_heap[a];
+  delayed_heap[a] = delayed_heap[b];
+  delayed_heap[b] = tmp;
+}
+
+static void delayed_sift_up(unsigned idx) {
+  while (idx > 0) {
+    unsigned parent = (idx - 1) / 2;
+    if (delayed_heap[parent].at <= delayed_heap[idx].at) {
+      break;
+    }
+    delayed_swap(parent, idx);
+    idx = parent;
+  }
+}
+
+static void delayed_sift_down(unsigned idx) {
+  for (;;) {
+    unsigned left = idx * 2 + 1;
+    unsigned right = left + 1;
+    unsigned smallest = idx;
+    if (left < delayed_heap_len && delayed_heap[left].at < delayed_heap[smallest].at)
+      smallest = left;
+    if (right < delayed_heap_len && delayed_heap[right].at < delayed_heap[smallest].at)
+      smallest = right;
+    if (smallest == idx)
+      break;
+    delayed_swap(idx, smallest);
+    idx = smallest;
+  }
+}
+
+static void delayed_schedule_timer(struct ev_loop *loop) {
+  if (delayed_heap_len == 0) {
+    if (ev_is_active(&delayed_timer)) {
+      ev_timer_stop(loop, &delayed_timer);
+    }
+    return;
+  }
+
+  ev_tstamp now = ev_now(loop);
+  ev_tstamp after = delayed_heap[0].at > now ? (delayed_heap[0].at - now) : 0.0;
+  ev_timer_set(&delayed_timer, after, 0.0);
+  if (!ev_is_active(&delayed_timer)) {
+    ev_timer_start(loop, &delayed_timer);
+  }
+  else {
+    ev_timer_again(loop, &delayed_timer);
+  }
+}
+
+static void delayed_timer_cb(struct ev_loop *loop, ev_timer *w, int revents) {
+  (void)w;
+  (void)revents;
+
+  ev_tstamp now = ev_now(loop);
+  while (delayed_heap_len > 0 && delayed_heap[0].at <= now) {
+    struct delayed_reply r = delayed_heap[0];
+    delayed_heap[0] = delayed_heap[--delayed_heap_len];
+    if (delayed_heap_len > 0) {
+      delayed_sift_down(0);
+    }
+
+    if (r.buf && r.len) {
+      (void)sendto(r.fd, r.buf, r.len, 0, (struct sockaddr *)&r.peer, r.peerlen);
+      delayed_bytes -= r.len;
+      free(r.buf);
+    }
+  }
+
+  delayed_schedule_timer(loop);
+}
+
+static void delayed_init(struct ev_loop *loop) {
+  delayed_loop = loop;
+  ev_timer_init(&delayed_timer, delayed_timer_cb, 0.0, 0.0);
+}
+
+static int delayed_push(int fd, const struct sockaddr *peer, socklen_t peerlen,
+                        const unsigned char *buf, size_t len, unsigned delay_ms) {
+  if (delay_ms == 0 || !buf || len == 0) {
+    return 0;
+  }
+  if (delayed_heap_len >= DELAYED_MAX_ITEMS || delayed_bytes + len > DELAYED_MAX_BYTES) {
+    return 0;
+  }
+
+  if (delayed_heap_len == delayed_heap_cap) {
+    unsigned new_cap = delayed_heap_cap ? delayed_heap_cap * 2 : 128;
+    if (new_cap > DELAYED_MAX_ITEMS) {
+      new_cap = DELAYED_MAX_ITEMS;
+    }
+    struct delayed_reply *n = realloc(delayed_heap, sizeof(*n) * new_cap);
+    if (!n) {
+      return 0;
+    }
+    delayed_heap = n;
+    delayed_heap_cap = new_cap;
+  }
+
+  unsigned char *copy = malloc(len);
+  if (!copy) {
+    return 0;
+  }
+  memcpy(copy, buf, len);
+
+  struct delayed_reply *r = &delayed_heap[delayed_heap_len];
+  memset(r, 0, sizeof(*r));
+  r->fd = fd;
+  r->peerlen = peerlen;
+  memcpy(&r->peer, peer, peerlen);
+  r->buf = copy;
+  r->len = len;
+  r->at = ev_now(delayed_loop) + ((ev_tstamp)delay_ms / 1000.0);
+
+  delayed_bytes += len;
+  delayed_sift_up(delayed_heap_len);
+  delayed_heap_len++;
+
+  delayed_schedule_timer(delayed_loop);
+  return 1;
+}
+
 static int request(int fd) {
   int q;
 #ifndef NO_IPv6
@@ -1280,6 +1423,7 @@ static int request(int fd) {
 #endif
     pkt[i].p_peerlen = MSG_FIELD(msg[i], msg_namelen);
     pkt[i].p_peer = MSG_FIELD(msg[i], msg_name);
+    pkt[i].p_delay_ms = 0;
     replies_lengths[i] = replypacket(&pkt[i], q, zonelist, &qi);
 
     if (flog) {
@@ -1289,6 +1433,15 @@ static int request(int fd) {
 
   int cur_rep = 0;
   for (int i = 0; i < lim; i ++) {
+    if (replies_lengths[i] > 0 && pkt[i].p_delay_ms > 0) {
+      if (delayed_push(fd, (const struct sockaddr *)&peer_sa[i],
+                       MSG_FIELD(msg[i], msg_namelen),
+                       pkt[i].p_buf, (size_t)replies_lengths[i],
+                       pkt[i].p_delay_ms)) {
+        replies_lengths[i] = 0;
+      }
+    }
+
     if (replies_lengths[i] > 0) {
       iovs[cur_rep].iov_base = pkt[i].p_buf;
       iovs[cur_rep].iov_len = replies_lengths[i];
@@ -1547,6 +1700,7 @@ int main(int argc, char **argv) {
   }
 
   init(argc, argv, loop);
+  delayed_init(loop);
   setup_signals(loop);
   reopenlog();
   can_reload = 1;

rbldnsd.h

@@ -37,6 +37,7 @@ struct dataset;
 struct dsdata;
 struct dhdata;
 struct dsctx;
+struct dnsqinfo;
 struct sockaddr;
 struct ev_loop;
 struct ev_stat;
@@ -50,8 +51,51 @@ struct dnspacket {		/* private structure */
   const struct dataset *p_substds;
   const struct sockaddr *p_peer;/* address of the requesting client */
   unsigned p_peerlen;
+  unsigned p_delay_ms;          /* optional delay requested by entry params */
 };
 
+struct kv_pair {
+  const char *k;
+  const char *v;
+};
+
+struct kv_params {
+  unsigned n;
+  char *storage;
+  struct kv_pair kv[];
+};
+
+struct entry_meta {
+  const char *rr;
+  const struct kv_params *params;
+};
+
+struct entry_action {
+  int allow;
+  unsigned delay_ms;
+  unsigned flags;
+};
+
+#define ENTRY_ACTION_STOP   0x01u
+#define ENTRY_ACTION_NODELAY 0x02u
+
+typedef int (*entry_params_handler_t)(const struct sockaddr *requestor,
+                                      const struct dataset *ds,
+                                      const struct dnsqinfo *qinfo,
+                                      const struct kv_params *params,
+                                      struct entry_action *action);
+
+int rbldnsd_register_entry_params_handler(entry_params_handler_t cb);
+int rbldnsd_apply_entry_params(const struct sockaddr *requestor,
+                               const struct dataset *ds,
+                               const struct dnsqinfo *qinfo,
+                               const struct kv_params *params,
+                               struct entry_action *action);
+
+char *rbldnsd_split_entry_params(char *s, char **params_out);
+const struct kv_params *rbldnsd_parse_kv_params(struct mempool *mp,
+                                                struct dsctx *dsc,
+                                                const char *s);
 struct dnsquery {	/* q */
   unsigned q_type;			/* query RR type */
   unsigned q_class;			/* query class */

rbldnsd_dnhash.c

@@ -41,6 +41,7 @@ struct key {
 
 struct entry {
   const char *rr;		/* A and TXT RRs */
+  const struct kv_params *params;
 };
 
 #if defined(__SSE4_2__) && defined(__x86_64) && defined(__GNUC__)
@@ -479,6 +480,7 @@ ds_dnhash_addent(khash_t(dnhash) *h,
                 struct dnhash_bloom *bloom,
                 const unsigned char *ldn,
                 const char *rr,
+                const struct kv_params *params,
                 unsigned dnlen) {
   struct entry *e;
   khiter_t k;
@@ -495,6 +497,7 @@ ds_dnhash_addent(khash_t(dnhash) *h,
 
   e = &kh_value(h, k);
   e->rr = rr;
+  e->params = params;
 
   if (bloom && ret > 0) {
     dnhash_bloom_add(bloom, ldn, dnlen);
@@ -508,6 +511,7 @@ ds_dnhash_line(struct dataset *ds, char *s, struct dsctx *dsc) {
   struct dsdata *dsd = ds->ds_dsd;
   unsigned char dn[DNS_MAXDN];
   const char *rr;
+  const struct kv_params *params = NULL;
   unsigned char *ldn;
   unsigned dnlen, size;
   int not, iswild, isplain;
@@ -553,6 +557,11 @@ ds_dnhash_line(struct dataset *ds, char *s, struct dsctx *dsc) {
   else {
     /* else parse rest */
     SKIPSPACE(s);
+
+    char *params_s = NULL;
+    rbldnsd_split_entry_params(s, &params_s);
+    params = rbldnsd_parse_kv_params(ds->ds_mp, dsc, params_s);
+
     if (!*s || ISCOMMENT(*s)) {
       /* use default if none given */
       rr = dsd->def_rr;
@@ -581,13 +590,13 @@ ds_dnhash_line(struct dataset *ds, char *s, struct dsctx *dsc) {
     }
 
     if (!ds_dnhash_addent(dsd->wild[dnlab - 1], dsd->wild_bloom[dnlab - 1],
-                          ldn, rr, dnlen - 1)) {
+                          ldn, rr, params, dnlen - 1)) {
       return 0;
     }
   }
 
   if (isplain) {
-    if (!ds_dnhash_addent(dsd->direct, dsd->direct_bloom, ldn, rr, dnlen - 1)) {
+    if (!ds_dnhash_addent(dsd->direct, dsd->direct_bloom, ldn, rr, params, dnlen - 1)) {
       return 0;
     }
   }
@@ -713,6 +722,19 @@ ds_dnhash_query(const struct dataset *ds, const struct dnsqinfo *qi,
         dns_dntop(pkey->ldn + 1, name, sizeof(name));
       }
 
+      if (e->rr) {
+        struct entry_action act;
+        act.allow = 1;
+        act.delay_ms = 0;
+        act.flags = 0;
+        rbldnsd_apply_entry_params(pkt->p_peer, ds, qi, e->params, &act);
+        if (!act.allow) {
+          return 0;
+        }
+        if (act.delay_ms > pkt->p_delay_ms) {
+          pkt->p_delay_ms = act.delay_ms;
+        }
+      }
       addrr_a_txt(pkt, qi->qi_tflag, e->rr, name, ds);
 
       return NSQUERY_FOUND;
@@ -751,6 +773,18 @@ ds_dnhash_query(const struct dataset *ds, const struct dnsqinfo *qi,
         }
 
         if (e->rr) {
+          struct entry_action act;
+          act.allow = 1;
+          act.delay_ms = 0;
+          act.flags = 0;
+          rbldnsd_apply_entry_params(pkt->p_peer, ds, qi, e->params, &act);
+          if (!act.allow) {
+            return 0;
+          }
+          if (act.delay_ms > pkt->p_delay_ms) {
+            pkt->p_delay_ms = act.delay_ms;
+          }
+
           addrr_a_txt(pkt, qi->qi_tflag, e->rr, name, ds);
 
           return NSQUERY_FOUND;