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 53 /** 54 * A message buffer with dnstap messages queued up. It is per-worker. 55 * It has locks to synchronize. If the buffer is full, a new message 56 * cannot be added and is discarded. A thread reads the messages and sends 57 * them. 58 */ 59 struct dt_msg_queue { 60 /** lock of the buffer structure. Hold this lock to add or remove 61 * entries to the buffer. Release it so that other threads can also 62 * put messages to log, or a message can be taken out to send away 63 * by the writer thread. 64 */ 65 lock_basic_type lock; 66 /** the maximum size of the buffer, in bytes */ 67 size_t maxsize; 68 /** current size of the buffer, in bytes. data bytes of messages. 69 * If a new message make it more than maxsize, the buffer is full */ 70 size_t cursize; 71 /** list of messages. The messages are added to the back and taken 72 * out from the front. */ 73 struct dt_msg_entry* first, *last; 74 /** reference to the io thread to wakeup */ 75 struct dt_io_thread* dtio; 76 }; 77 78 /** 79 * An entry in the dt_msg_queue. contains one DNSTAP message. 80 * It is malloced. 81 */ 82 struct dt_msg_entry { 83 /** next in the list. */ 84 struct dt_msg_entry* next; 85 /** the buffer with the data to send, an encoded DNSTAP message */ 86 void* buf; 87 /** the length to send. */ 88 size_t len; 89 }; 90 91 /** 92 * Containing buffer and counter for reading DNSTAP frames. 93 */ 94 struct dt_frame_read_buf { 95 /** Buffer containing frame, except length counter(s). */ 96 void* buf; 97 /** Number of bytes written to buffer. */ 98 size_t buf_count; 99 /** Capacity of the buffer. */ 100 size_t buf_cap; 101 102 /** Frame length field. Will contain the 2nd length field for control 103 * frames. */ 104 uint32_t frame_len; 105 /** Number of bytes that have been written to the frame_length field. */ 106 size_t frame_len_done; 107 108 /** Set to 1 if this is a control frame, 0 otherwise (ie data frame). */ 109 int control_frame; 110 }; 111 112 /** 113 * IO thread that reads from the queues and writes them. 114 */ 115 struct dt_io_thread { 116 /** the thread number for the dtio thread, 117 * must be first to cast thread arg to int* in checklock code. */ 118 int threadnum; 119 /** event base, for event handling */ 120 void* event_base; 121 /** list of queues that is registered to get written */ 122 struct dt_io_list_item* io_list; 123 /** iterator point in the io_list, to pick from them in a 124 * round-robin fashion, instead of only from the first when busy. 125 * if NULL it means start at the start of the list. */ 126 struct dt_io_list_item* io_list_iter; 127 /** thread id, of the io thread */ 128 ub_thread_type tid; 129 /** if the io processing has started */ 130 int started; 131 /** ssl context for the io thread, for tls connections. type SSL_CTX* */ 132 void* ssl_ctx; 133 /** if SNI will be used for TLS connections. */ 134 int tls_use_sni; 135 136 /** file descriptor that the thread writes to */ 137 int fd; 138 /** event structure that the thread uses */ 139 void* event; 140 /** the event is added */ 141 int event_added; 142 /** event added is a write event */ 143 int event_added_is_write; 144 /** check for nonblocking connect errors on fd */ 145 int check_nb_connect; 146 /** ssl for current connection, type SSL* */ 147 void* ssl; 148 /** true if the handshake for SSL is done, 0 if not */ 149 int ssl_handshake_done; 150 /** true if briefly the SSL wants a read event, 0 if not. 151 * This happens during negotiation, we then do not want to write, 152 * but wait for a read event. */ 153 int ssl_brief_read; 154 /** true if SSL_read is waiting for a write event. Set back to 0 after 155 * single write event is handled. */ 156 int ssl_brief_write; 157 158 /** the buffer that currently getting written, or NULL if no 159 * (partial) message written now */ 160 void* cur_msg; 161 /** length of the current message */ 162 size_t cur_msg_len; 163 /** number of bytes written for the current message */ 164 size_t cur_msg_done; 165 /** number of bytes of the length that have been written, 166 * for the current message length that precedes the frame */ 167 size_t cur_msg_len_done; 168 169 /** command pipe that stops the pipe if closed. Used to quit 170 * the program. [0] is read, [1] is written to. */ 171 int commandpipe[2]; 172 /** the event to listen to the commandpipe */ 173 void* command_event; 174 /** the io thread wants to exit */ 175 int want_to_exit; 176 177 /** in stop flush, this is nonNULL and references the stop_ev */ 178 void* stop_flush_event; 179 180 /** the timer event for connection retries */ 181 void* reconnect_timer; 182 /** if the reconnect timer is added to the event base */ 183 int reconnect_is_added; 184 /** the current reconnection timeout, it is increased with 185 * exponential backoff, in msec */ 186 int reconnect_timeout; 187 188 /** If the log server is connected to over unix domain sockets, 189 * eg. a file is named that is created to log onto. */ 190 int upstream_is_unix; 191 /** if the log server is connected to over TCP. The ip address and 192 * port are used */ 193 int upstream_is_tcp; 194 /** if the log server is connected to over TLS. ip address, port, 195 * and client certificates can be used for authentication. */ 196 int upstream_is_tls; 197 198 /** Perform bidirectional Frame Streams handshake before sending 199 * messages. */ 200 int is_bidirectional; 201 /** Set if the READY control frame has been sent. */ 202 int ready_frame_sent; 203 /** Set if valid ACCEPT frame is received. */ 204 int accept_frame_received; 205 /** (partially) read frame */ 206 struct dt_frame_read_buf read_frame; 207 208 /** the file path for unix socket (or NULL) */ 209 char* socket_path; 210 /** the ip address and port number (or NULL) */ 211 char* ip_str; 212 /** is the TLS upstream authenticated by name, if nonNULL, 213 * we use the same cert bundle as used by other TLS streams. */ 214 char* tls_server_name; 215 /** are client certificates in use */ 216 int use_client_certs; 217 /** client cert files: the .key file */ 218 char* client_key_file; 219 /** client cert files: the .pem file */ 220 char* client_cert_file; 221 }; 222 223 /** 224 * IO thread list of queues list item 225 * lists a worker queue that should be looked at and sent to the log server. 226 */ 227 struct dt_io_list_item { 228 /** next in the list of buffers to inspect */ 229 struct dt_io_list_item* next; 230 /** buffer of this worker */ 231 struct dt_msg_queue* queue; 232 }; 233 234 /** 235 * Create new (empty) worker message queue. Limit set to default on max. 236 * @return NULL on malloc failure or a new queue (not locked). 237 */ 238 struct dt_msg_queue* dt_msg_queue_create(void); 239 240 /** 241 * Delete a worker message queue. It has to be unlinked from access, 242 * so it can be deleted without lock worries. The queue is emptied (deleted). 243 * @param mq: message queue. 244 */ 245 void dt_msg_queue_delete(struct dt_msg_queue* mq); 246 247 /** 248 * Submit a message to the queue. The queue is locked by the routine, 249 * the message is inserted, and then the queue is unlocked so the 250 * message can be picked up by the writer thread. 251 * @param mq: message queue. 252 * @param buf: buffer with message (dnstap contents). 253 * The buffer must have been malloced by caller. It is linked in 254 * the queue, and is free()d after use. If the routine fails 255 * the buffer is freed as well (and nothing happens, the item 256 * could not be logged). 257 * @param len: length of buffer. 258 */ 259 void dt_msg_queue_submit(struct dt_msg_queue* mq, void* buf, size_t len); 260 261 /** 262 * Create IO thread. 263 * @return new io thread object. not yet started. or NULL malloc failure. 264 */ 265 struct dt_io_thread* dt_io_thread_create(void); 266 267 /** 268 * Delete the IO thread structure. 269 * @param dtio: the io thread that is deleted. It must not be running. 270 */ 271 void dt_io_thread_delete(struct dt_io_thread* dtio); 272 273 /** 274 * Apply config to the dtio thread 275 * @param dtio: io thread, not yet started. 276 * @param cfg: config file struct. 277 * @return false on malloc failure. 278 */ 279 int dt_io_thread_apply_cfg(struct dt_io_thread* dtio, 280 struct config_file *cfg); 281 282 /** 283 * Register a msg queue to the io thread. It will be polled to see if 284 * there are messages and those then get removed and sent, when the thread 285 * is running. 286 * @param dtio: the io thread. 287 * @param mq: message queue to register. 288 * @return false on failure (malloc failure). 289 */ 290 int dt_io_thread_register_queue(struct dt_io_thread* dtio, 291 struct dt_msg_queue* mq); 292 293 /** 294 * Unregister queue from io thread. 295 * @param dtio: the io thread. 296 * @param mq: message queue. 297 */ 298 void dt_io_thread_unregister_queue(struct dt_io_thread* dtio, 299 struct dt_msg_queue* mq); 300 301 /** 302 * Start the io thread 303 * @param dtio: the io thread. 304 * @param event_base_nothr: the event base to attach the events to, in case 305 * we are running without threads. With threads, this is ignored 306 * and a thread is started to process the dnstap log messages. 307 * @param numworkers: number of worker threads. The dnstap io thread is 308 * that number +1 as the threadnumber (in logs). 309 * @return false on failure. 310 */ 311 int dt_io_thread_start(struct dt_io_thread* dtio, void* event_base_nothr, 312 int numworkers); 313 314 /** 315 * Stop the io thread 316 * @param dtio: the io thread. 317 */ 318 void dt_io_thread_stop(struct dt_io_thread* dtio); 319 320 /** callback for the dnstap reconnect, to start reconnecting to output */ 321 void dtio_reconnect_timeout_cb(int fd, short bits, void* arg); 322 323 /** callback for the dnstap events, to write to the output */ 324 void dtio_output_cb(int fd, short bits, void* arg); 325 326 /** callback for the dnstap commandpipe, to stop the dnstap IO */ 327 void dtio_cmd_cb(int fd, short bits, void* arg); 328 329 /** callback for the timer when the thread stops and wants to finish up */ 330 void dtio_stop_timer_cb(int fd, short bits, void* arg); 331 332 /** callback for the output when the thread stops and wants to finish up */ 333 void dtio_stop_ev_cb(int fd, short bits, void* arg); 334 335 /** callback for unbound-dnstap-socket */ 336 void dtio_tap_callback(int fd, short bits, void* arg); 337 338 /** callback for unbound-dnstap-socket */ 339 void dtio_mainfdcallback(int fd, short bits, void* arg); 340 341 #endif /* DTSTREAM_H */ 342