1 /* 2 * dnstap/dtstream.h - Frame Streams thread for unbound DNSTAP 3 * 4 * Copyright (c) 2020, NLnet Labs. All rights reserved. 5 * 6 * This software is open source. 7 * 8 * Redistribution and use in source and binary forms, with or without 9 * modification, are permitted provided that the following conditions 10 * are met: 11 * 12 * Redistributions of source code must retain the above copyright notice, 13 * this list of conditions and the following disclaimer. 14 * 15 * Redistributions in binary form must reproduce the above copyright notice, 16 * this list of conditions and the following disclaimer in the documentation 17 * and/or other materials provided with the distribution. 18 * 19 * Neither the name of the NLNET LABS nor the names of its contributors may 20 * be used to endorse or promote products derived from this software without 21 * specific prior written permission. 22 * 23 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 24 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 25 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR 26 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT 27 * HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, 28 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED 29 * TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 30 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 31 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 32 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 33 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 34 * 35 */ 36 37 /** 38 * \file 39 * 40 * An implementation of the Frame Streams data transport protocol for 41 * the Unbound DNSTAP message logging facility. 42 */ 43 44 #ifndef DTSTREAM_H 45 #define DTSTREAM_H 46 47 #include "util/locks.h" 48 struct dt_msg_entry; 49 struct dt_io_list_item; 50 struct dt_io_thread; 51 struct config_file; 52 struct comm_base; 53 54 /** 55 * A message buffer with dnstap messages queued up. It is per-worker. 56 * It has locks to synchronize. If the buffer is full, a new message 57 * cannot be added and is discarded. A thread reads the messages and sends 58 * them. 59 */ 60 struct dt_msg_queue { 61 /** lock of the buffer structure. Hold this lock to add or remove 62 * entries to the buffer. Release it so that other threads can also 63 * put messages to log, or a message can be taken out to send away 64 * by the writer thread. 65 */ 66 lock_basic_type lock; 67 /** the maximum size of the buffer, in bytes */ 68 size_t maxsize; 69 /** current size of the buffer, in bytes. data bytes of messages. 70 * If a new message make it more than maxsize, the buffer is full */ 71 size_t cursize; 72 /** number of messages in the queue */ 73 int msgcount; 74 /** list of messages. The messages are added to the back and taken 75 * out from the front. */ 76 struct dt_msg_entry* first, *last; 77 /** reference to the io thread to wakeup */ 78 struct dt_io_thread* dtio; 79 /** the wakeup timer for dtio, on worker event base */ 80 struct comm_timer* wakeup_timer; 81 }; 82 83 /** 84 * An entry in the dt_msg_queue. contains one DNSTAP message. 85 * It is malloced. 86 */ 87 struct dt_msg_entry { 88 /** next in the list. */ 89 struct dt_msg_entry* next; 90 /** the buffer with the data to send, an encoded DNSTAP message */ 91 void* buf; 92 /** the length to send. */ 93 size_t len; 94 }; 95 96 /** 97 * Containing buffer and counter for reading DNSTAP frames. 98 */ 99 struct dt_frame_read_buf { 100 /** Buffer containing frame, except length counter(s). */ 101 void* buf; 102 /** Number of bytes written to buffer. */ 103 size_t buf_count; 104 /** Capacity of the buffer. */ 105 size_t buf_cap; 106 107 /** Frame length field. Will contain the 2nd length field for control 108 * frames. */ 109 uint32_t frame_len; 110 /** Number of bytes that have been written to the frame_length field. */ 111 size_t frame_len_done; 112 113 /** Set to 1 if this is a control frame, 0 otherwise (ie data frame). */ 114 int control_frame; 115 }; 116 117 /** 118 * IO thread that reads from the queues and writes them. 119 */ 120 struct dt_io_thread { 121 /** the thread number for the dtio thread, 122 * must be first to cast thread arg to int* in checklock code. */ 123 int threadnum; 124 /** event base, for event handling */ 125 void* event_base; 126 /** list of queues that is registered to get written */ 127 struct dt_io_list_item* io_list; 128 /** iterator point in the io_list, to pick from them in a 129 * round-robin fashion, instead of only from the first when busy. 130 * if NULL it means start at the start of the list. */ 131 struct dt_io_list_item* io_list_iter; 132 /** thread id, of the io thread */ 133 ub_thread_type tid; 134 /** if the io processing has started */ 135 int started; 136 /** ssl context for the io thread, for tls connections. type SSL_CTX* */ 137 void* ssl_ctx; 138 /** if SNI will be used for TLS connections. */ 139 int tls_use_sni; 140 141 /** file descriptor that the thread writes to */ 142 int fd; 143 /** event structure that the thread uses */ 144 void* event; 145 /** the event is added */ 146 int event_added; 147 /** event added is a write event */ 148 int event_added_is_write; 149 /** check for nonblocking connect errors on fd */ 150 int check_nb_connect; 151 /** ssl for current connection, type SSL* */ 152 void* ssl; 153 /** true if the handshake for SSL is done, 0 if not */ 154 int ssl_handshake_done; 155 /** true if briefly the SSL wants a read event, 0 if not. 156 * This happens during negotiation, we then do not want to write, 157 * but wait for a read event. */ 158 int ssl_brief_read; 159 /** true if SSL_read is waiting for a write event. Set back to 0 after 160 * single write event is handled. */ 161 int ssl_brief_write; 162 163 /** the buffer that currently getting written, or NULL if no 164 * (partial) message written now */ 165 void* cur_msg; 166 /** length of the current message */ 167 size_t cur_msg_len; 168 /** number of bytes written for the current message */ 169 size_t cur_msg_done; 170 /** number of bytes of the length that have been written, 171 * for the current message length that precedes the frame */ 172 size_t cur_msg_len_done; 173 174 /** lock on wakeup_timer_enabled */ 175 lock_basic_type wakeup_timer_lock; 176 /** if wakeup timer is enabled in some thread */ 177 int wakeup_timer_enabled; 178 /** command pipe that stops the pipe if closed. Used to quit 179 * the program. [0] is read, [1] is written to. */ 180 int commandpipe[2]; 181 /** the event to listen to the commandpipe */ 182 void* command_event; 183 /** the io thread wants to exit */ 184 int want_to_exit; 185 186 /** in stop flush, this is nonNULL and references the stop_ev */ 187 void* stop_flush_event; 188 189 /** the timer event for connection retries */ 190 void* reconnect_timer; 191 /** if the reconnect timer is added to the event base */ 192 int reconnect_is_added; 193 /** the current reconnection timeout, it is increased with 194 * exponential backoff, in msec */ 195 int reconnect_timeout; 196 197 /** If the log server is connected to over unix domain sockets, 198 * eg. a file is named that is created to log onto. */ 199 int upstream_is_unix; 200 /** if the log server is connected to over TCP. The ip address and 201 * port are used */ 202 int upstream_is_tcp; 203 /** if the log server is connected to over TLS. ip address, port, 204 * and client certificates can be used for authentication. */ 205 int upstream_is_tls; 206 207 /** Perform bidirectional Frame Streams handshake before sending 208 * messages. */ 209 int is_bidirectional; 210 /** Set if the READY control frame has been sent. */ 211 int ready_frame_sent; 212 /** Set if valid ACCEPT frame is received. */ 213 int accept_frame_received; 214 /** (partially) read frame */ 215 struct dt_frame_read_buf read_frame; 216 217 /** the file path for unix socket (or NULL) */ 218 char* socket_path; 219 /** the ip address and port number (or NULL) */ 220 char* ip_str; 221 /** is the TLS upstream authenticated by name, if nonNULL, 222 * we use the same cert bundle as used by other TLS streams. */ 223 char* tls_server_name; 224 /** are client certificates in use */ 225 int use_client_certs; 226 /** client cert files: the .key file */ 227 char* client_key_file; 228 /** client cert files: the .pem file */ 229 char* client_cert_file; 230 }; 231 232 /** 233 * IO thread list of queues list item 234 * lists a worker queue that should be looked at and sent to the log server. 235 */ 236 struct dt_io_list_item { 237 /** next in the list of buffers to inspect */ 238 struct dt_io_list_item* next; 239 /** buffer of this worker */ 240 struct dt_msg_queue* queue; 241 }; 242 243 /** 244 * Create new (empty) worker message queue. Limit set to default on max. 245 * @param base: event base for wakeup timer. 246 * @return NULL on malloc failure or a new queue (not locked). 247 */ 248 struct dt_msg_queue* dt_msg_queue_create(struct comm_base* base); 249 250 /** 251 * Delete a worker message queue. It has to be unlinked from access, 252 * so it can be deleted without lock worries. The queue is emptied (deleted). 253 * @param mq: message queue. 254 */ 255 void dt_msg_queue_delete(struct dt_msg_queue* mq); 256 257 /** 258 * Submit a message to the queue. The queue is locked by the routine, 259 * the message is inserted, and then the queue is unlocked so the 260 * message can be picked up by the writer thread. 261 * @param mq: message queue. 262 * @param buf: buffer with message (dnstap contents). 263 * The buffer must have been malloced by caller. It is linked in 264 * the queue, and is free()d after use. If the routine fails 265 * the buffer is freed as well (and nothing happens, the item 266 * could not be logged). 267 * @param len: length of buffer. 268 */ 269 void dt_msg_queue_submit(struct dt_msg_queue* mq, void* buf, size_t len); 270 271 /** timer callback to wakeup dtio thread to process messages */ 272 void mq_wakeup_cb(void* arg); 273 274 /** 275 * Create IO thread. 276 * @return new io thread object. not yet started. or NULL malloc failure. 277 */ 278 struct dt_io_thread* dt_io_thread_create(void); 279 280 /** 281 * Delete the IO thread structure. 282 * @param dtio: the io thread that is deleted. It must not be running. 283 */ 284 void dt_io_thread_delete(struct dt_io_thread* dtio); 285 286 /** 287 * Apply config to the dtio thread 288 * @param dtio: io thread, not yet started. 289 * @param cfg: config file struct. 290 * @return false on malloc failure. 291 */ 292 int dt_io_thread_apply_cfg(struct dt_io_thread* dtio, 293 struct config_file *cfg); 294 295 /** 296 * Register a msg queue to the io thread. It will be polled to see if 297 * there are messages and those then get removed and sent, when the thread 298 * is running. 299 * @param dtio: the io thread. 300 * @param mq: message queue to register. 301 * @return false on failure (malloc failure). 302 */ 303 int dt_io_thread_register_queue(struct dt_io_thread* dtio, 304 struct dt_msg_queue* mq); 305 306 /** 307 * Unregister queue from io thread. 308 * @param dtio: the io thread. 309 * @param mq: message queue. 310 */ 311 void dt_io_thread_unregister_queue(struct dt_io_thread* dtio, 312 struct dt_msg_queue* mq); 313 314 /** 315 * Start the io thread 316 * @param dtio: the io thread. 317 * @param event_base_nothr: the event base to attach the events to, in case 318 * we are running without threads. With threads, this is ignored 319 * and a thread is started to process the dnstap log messages. 320 * @param numworkers: number of worker threads. The dnstap io thread is 321 * that number +1 as the threadnumber (in logs). 322 * @return false on failure. 323 */ 324 int dt_io_thread_start(struct dt_io_thread* dtio, void* event_base_nothr, 325 int numworkers); 326 327 /** 328 * Stop the io thread 329 * @param dtio: the io thread. 330 */ 331 void dt_io_thread_stop(struct dt_io_thread* dtio); 332 333 /** callback for the dnstap reconnect, to start reconnecting to output */ 334 void dtio_reconnect_timeout_cb(int fd, short bits, void* arg); 335 336 /** callback for the dnstap events, to write to the output */ 337 void dtio_output_cb(int fd, short bits, void* arg); 338 339 /** callback for the dnstap commandpipe, to stop the dnstap IO */ 340 void dtio_cmd_cb(int fd, short bits, void* arg); 341 342 /** callback for the timer when the thread stops and wants to finish up */ 343 void dtio_stop_timer_cb(int fd, short bits, void* arg); 344 345 /** callback for the output when the thread stops and wants to finish up */ 346 void dtio_stop_ev_cb(int fd, short bits, void* arg); 347 348 /** callback for unbound-dnstap-socket */ 349 void dtio_tap_callback(int fd, short bits, void* arg); 350 351 /** callback for unbound-dnstap-socket */ 352 void dtio_mainfdcallback(int fd, short bits, void* arg); 353 354 #endif /* DTSTREAM_H */ 355