text: /*
   * Copyright (c) 2014-2020 by Farsight Security, Inc.
   *
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  /* asprintf() does not appear on linux without this */
  #define _GNU_SOURCE
  
  #define _BSD_SOURCE
  #define _DEFAULT_SOURCE
  
  #include <sys/wait.h>
  
  #include <assert.h>
  #include <errno.h>
  #include <signal.h>
  #include <string.h>
  #include <unistd.h>
  
  #include "defs.h"
  #include "netio.h"
  #include "pdns.h"
  #include "globals.h"
  
  static void io_drain(void);
  static void fetch_reap(fetch_t);
  static void fetch_done(fetch_t);
  static void fetch_unlink(fetch_t);
  static void query_done(query_t);
  
  static writer_t writers = NULL;
  static CURLM *multi = NULL;
  static bool curl_cleanup_needed = false;
  
  const char saf_begin[] = "begin";
  const char saf_ongoing[] = "ongoing";
  const char saf_succeeded[] = "succeeded";
  const char saf_limited[] = "limited";
  const char saf_failed[] = "failed";
  
  const char *saf_valid_conds[] = {
  	saf_begin, saf_ongoing, saf_succeeded, saf_limited, saf_failed
  };
  
  /* make_curl -- perform global initializations of libcurl.
   */
  void
  make_curl(void) {
  	curl_global_init(CURL_GLOBAL_DEFAULT);
  	curl_cleanup_needed = true;
  	multi = curl_multi_init();
  	if (multi == NULL) {
  		fprintf(stderr, "%s: curl_multi_init() failed\n",
  			program_name);
  		my_exit(1);
  	}
  }
  
  /* unmake_curl -- clean up and discard libcurl's global state.
   */
  void
  unmake_curl(void) {
  	if (multi != NULL) {
  		curl_multi_cleanup(multi);
  		multi = NULL;
  	}
  	if (curl_cleanup_needed) {
  		curl_global_cleanup();
  		curl_cleanup_needed = false;
  	}
  }
  
  /* fetch -- given a url, tell libcurl to go fetch it.
   */
  void
  create_fetch(query_t query, char *url) {
  	fetch_t fetch = NULL;
  	CURLMcode res;
  
  	DEBUG(2, true, "fetch(%s)\n", url);
  	CREATE(fetch, sizeof *fetch);
  	fetch->query = query;
  	query = NULL;
  	fetch->easy = curl_easy_init();
  	if (fetch->easy == NULL) {
  		/* an error will have been output by libcurl in this case. */
  		DESTROY(fetch);
  		DESTROY(url);
  		my_exit(1);
  	}
  	fetch->url = url;
  	url = NULL;
  	curl_easy_setopt(fetch->easy, CURLOPT_URL, fetch->url);
  	if (donotverify) {
  		curl_easy_setopt(fetch->easy, CURLOPT_SSL_VERIFYPEER, 0L);
  		curl_easy_setopt(fetch->easy, CURLOPT_SSL_VERIFYHOST, 0L);
  	}
  
  	/* if user specified a prefence for IPv4 or IPv6, use it. */
  	if (curl_ipresolve != CURL_IPRESOLVE_WHATEVER)
  		curl_easy_setopt(fetch->easy,
  				 CURLOPT_IPRESOLVE, curl_ipresolve);
  
  	if (psys->auth != NULL)
  	    psys->auth(fetch);
  
  	fetch->hdrs = curl_slist_append(fetch->hdrs, jsonl_header);
  	curl_easy_setopt(fetch->easy, CURLOPT_HTTPHEADER, fetch->hdrs);
  	curl_easy_setopt(fetch->easy, CURLOPT_WRITEFUNCTION, writer_func);
  	curl_easy_setopt(fetch->easy, CURLOPT_WRITEDATA, fetch);
  	curl_easy_setopt(fetch->easy, CURLOPT_PRIVATE, fetch);
  #ifdef CURL_AT_LEAST_VERSION
  /* If CURL_AT_LEAST_VERSION is not defined then the curl is probably too old */
  #if CURL_AT_LEAST_VERSION(7,42,0)
  	/* do not allow curl to swallow /./ and /../ in our URLs */
  	curl_easy_setopt(fetch->easy, CURLOPT_PATH_AS_IS, 1L);
  #endif
  #endif /* CURL_AT_LEAST_VERSION */
  	if (debug_level >= 3)
  		curl_easy_setopt(fetch->easy, CURLOPT_VERBOSE, 1L);
  
  	fetch->query->fetch = fetch;
  
  	res = curl_multi_add_handle(multi, fetch->easy);
  	if (res != CURLM_OK) {
  		fprintf(stderr, "%s: curl_multi_add_handle() failed: %s\n",
  			program_name, curl_multi_strerror(res));
  		my_exit(1);
  	}
  }
  
  /* fetch_reap -- reap one fetch.
   */
  static void
  fetch_reap(fetch_t fetch) {
  	if (fetch->easy != NULL) {
  		curl_multi_remove_handle(multi, fetch->easy);
  		curl_easy_cleanup(fetch->easy);
  		fetch->easy = NULL;
  	}
  	if (fetch->hdrs != NULL) {
  		curl_slist_free_all(fetch->hdrs);
  		fetch->hdrs = NULL;
  	}
  	DESTROY(fetch->url);
  	DESTROY(fetch->buf);
  	DESTROY(fetch);
  }
  
  /* fetch_done -- deal with consequences of end-of-fetch.
   */
  static void
  fetch_done(fetch_t fetch) {
  	query_t query = fetch->query;
  
  	query_done(query);
  }
  
  /* fetch_unlink -- disconnect a fetch from its writer.
   */
  static void
  fetch_unlink(fetch_t fetch) {
  	assert(fetch == fetch->query->fetch);
  	fetch->query->fetch = NULL;
  	fetch->query = NULL;
  }
  
  /* writer_init -- instantiate a writer
   */
  writer_t
  writer_init(long output_limit) {
  	writer_t writer = NULL;
  
  	CREATE(writer, sizeof(struct writer));
  	writer->output_limit = output_limit;
  
  	return (writer);
  }
  
  /* query_status -- install a status code and description in a query.
   */
  void
  query_status(query_t query, const char *status, const char *message) {
  	assert((query->status == NULL) == (query->message == NULL));
  	assert(query->status == NULL);
  	query->status = strdup(status);
  	query->message = strdup(message);
  }
  
  /* writer_func -- process a block of json text.
   *
   * This function's signature must conform to write_callback() in
   * CURLOPT_WRITEFUNCTION.
   * Returns the number of bytes actually taken care of or returns
   * CURL_WRITEFUNC_PAUSE to pause this query's connection until
   * curl_easy_pause(..., CURLPAUSE_CONT) is called.
   */
  size_t
  writer_func(char *ptr, size_t size, size_t nmemb, void *blob) {
  	fetch_t fetch = (fetch_t) blob;
  	query_t query = fetch->query;
  	writer_t writer = query->writer;
  	size_t bytes = size * nmemb;
  	char *nl;
  
  	DEBUG(3, true, "writer_func(%d, %d): %d\n",
  	      (int)size, (int)nmemb, (int)bytes);
  
  	fetch->buf = realloc(fetch->buf, fetch->len + bytes);
  	memcpy(fetch->buf + fetch->len, ptr, bytes);
  	fetch->len += bytes;
  
  	/* when the fetch is a live web result, emit
  	 * !2xx errors and info payloads as reports.
  	 */
  	if (fetch->easy != NULL) {
  		if (fetch->rcode == 0)
  			curl_easy_getinfo(fetch->easy,
  					  CURLINFO_RESPONSE_CODE,
  					  &fetch->rcode);
  		if (fetch->rcode != HTTP_OK) {
  			char *message = strndup(fetch->buf, fetch->len);
  
  			/* only report the first line of data. */
  			char *eol = strpbrk(message, "\r\n");
  			if (eol != NULL)
  				*eol = '\0';
  
  			/* if the message (that's left) is just <html>,
  			 * change it to an HTTP Status code */
  			if (strcasecmp(message, "<html>") == 0) {
  				DESTROY(message);
  				if (asprintf(&message, "HTTP Status %ld",
  					     fetch->rcode) < 0)
  					my_panic(true, "asprintf");
  			}
  
  			/* only report the first response status (vs. -m). */
  			if (query->status == NULL) {
  				query_status(query,
  					     psys->status(fetch),
  					     message);
  				if (!quiet) {
  					char *url;
  
  					curl_easy_getinfo(fetch->easy,
  							CURLINFO_EFFECTIVE_URL,
  							  &url);
  					fprintf(stderr,
  						"%s: warning: "
  						"libcurl %ld [%s]\n",
  						program_name, fetch->rcode,
  						url);
  				}
  			}
  			if (!quiet)
  				fprintf(stderr, "%s: warning: libcurl: [%s]\n",
  					program_name, message);
  			DESTROY(message);
  			fetch->buf[0] = '\0';
  			fetch->len = 0;
  			return (bytes);
  		}
  	}
  
  	/* deblock. */
  	while ((nl = memchr(fetch->buf, '\n', fetch->len)) != NULL) {
  		size_t pre_len = (size_t)(nl - fetch->buf),
  			post_len = (fetch->len - pre_len) - 1;
  
  		if (writer->output_limit > 0 &&
  		    writer->count >= writer->output_limit)
  		{
  			DEBUG(9, true, "hit output limit %ld\n",
  			      writer->output_limit);
  			/* cause CURLE_WRITE_ERROR for this transfer. */
  			bytes = 0;
  			query->saf_cond = sc_we_limited;
  			/* inform io_engine() that the abort is intentional. */
  			fetch->stopped = true;
  		} else {
  			query->writer->count +=
  				data_blob(query, fetch->buf, pre_len);
  
  			switch (query->saf_cond) {
  			case sc_init:
  			case sc_begin:
  			case sc_ongoing:
  			case sc_missing:
  				break;
  			case sc_succeeded:
  			case sc_limited:
  			case sc_failed:
  			case sc_we_limited:
  				/* inform io_engine() intentional
  				 * abort. */
  				fetch->stopped = true;
  				break;
  			}
  		}
  		memmove(fetch->buf, nl + 1, post_len);
  		fetch->len = post_len;
  	}
  
  	return (bytes);
  }
  
  /* query_done -- do something with leftover buffer data when a query ends.
   */
  static void
  query_done(query_t query) {
  	DEBUG(2, true, "query_done(%s)\n", query->command);
  
  	if (!quiet) {
  		const char *msg = or_else(query->saf_msg, "");
  
  		if (query->saf_cond == sc_limited)
  			fprintf(stderr, "Query limited: %s\n", msg);
  		else if (query->saf_cond == sc_failed)
  			fprintf(stderr, "Query failed: %s\n", msg);
  		else if (query->saf_cond == sc_missing)
  			fprintf(stderr, "Query response_missing: %s\n", msg);
  		else if (query->status != NULL)
  			fprintf(stderr, "Query status: %s (%s)\n",
  				query->status, query->message);
  	}
  }
  
  /* writer_fini -- stop a writer's fetch
   */
  void
  writer_fini(writer_t writer) {
  	/* finish and close any fetches still cooking. */
  	if (writer->query != NULL) {
  		query_t query = writer->query;
  
  		/* release any buffered info. */
  		if (query->fetch != NULL) {
  			DESTROY(query->fetch->buf);
  			if (query->fetch->len != 0) {
  				fprintf(stderr,
  					"%s: warning: stranding %d octets!\n",
  					program_name, (int)query->fetch->len);
  				query->fetch->len = 0;
  			}
  
  			/* tear down any curl infrastructure on the fetch. */
  			fetch_reap(query->fetch);
  
  			query->fetch = NULL;
  		}
  		assert((query->status != NULL) == (query->message != NULL));
  		DESTROY(query->status);
  		DESTROY(query->message);
  		DESTROY(query->command);
  		DESTROY(query);
  	}
  
  	DESTROY(writer);
  }
  
  void
  unmake_writers(void) {
  	while (writers != NULL)
  		writer_fini(writers);
  }
  
  /* io_engine -- let libcurl run until there are few enough outstanding jobs.
   */
  void
  io_engine(int jobs) {
  	int still, repeats, numfds;
  
  	DEBUG(2, true, "io_engine(%d)\n", jobs);
  
  	/* let libcurl run while there are too many jobs remaining. */
  	still = 0;
  	repeats = 0;
  	while (curl_multi_perform(multi, &still) == CURLM_OK && still > jobs) {
  		DEBUG(3, true, "...waiting (still %d)\n", still);
  		numfds = 0;
  		if (curl_multi_wait(multi, NULL, 0, 0, &numfds) != CURLM_OK)
  			break;
  		if (numfds == 0) {
  			/* curl_multi_wait() can return 0 fds for no reason. */
  			if (++repeats > 1) {
  				struct timespec req, rem;
  
  				req = (struct timespec){
  					.tv_sec = 0,
  					.tv_nsec = 100*1000*1000  // 100ms
  				};
  				while (nanosleep(&req, &rem) == EINTR) {
  					/* as required by nanosleep(3). */
  					req = rem;
  				}
  			}
  		} else {
  			repeats = 0;
  		}
  		io_drain();
  	}
  	io_drain();
  }
  
  /* io_drain -- drain the response code reports.
   */
  static void
  io_drain(void) {
  	struct CURLMsg *cm;
  	int still = 0;
  
  	while ((cm = curl_multi_info_read(multi, &still)) != NULL) {
  		fetch_t fetch;
  		query_t query;
  		char *private;
  
  		curl_easy_getinfo(cm->easy_handle,
  				  CURLINFO_PRIVATE,
  				  &private);
  		fetch = (fetch_t) private;
  		query = fetch->query;
  
  		if (cm->msg == CURLMSG_DONE) {
  			if (fetch->rcode == 0)
  				curl_easy_getinfo(fetch->easy,
  						  CURLINFO_RESPONSE_CODE,
  						  &fetch->rcode);
  
  			DEBUG(2, true, "io_drain(%s) DONE rcode=%d\n",
  			      query->command, fetch->rcode);
  			DEBUG(2, true, "... saf_cond %d saf_msg %s\n",
  			      query->saf_cond,
  			      or_else(query->saf_msg, ""));
  
  			if (cm->data.result == CURLE_COULDNT_RESOLVE_HOST) {
  				fprintf(stderr,
  					"%s: warning: libcurl failed since "
  					"could not resolve host\n",
  					program_name);
  				exit_code = 1;
  			} else if (cm->data.result == CURLE_COULDNT_CONNECT) {
  				fprintf(stderr,
  					"%s: warning: libcurl failed since "
  					"could not connect\n",
  					program_name);
  				exit_code = 1;
  			} else if (cm->data.result != CURLE_OK &&
  				   !fetch->stopped)
  			{
  				fprintf(stderr,
  					"%s: warning: libcurl failed with "
  					"curl error %d (%s)\n",
  					program_name, cm->data.result,
  					curl_easy_strerror(cm->data.result));
  				exit_code = 1;
  			}
  
  			/* record emptiness as status if nothing else. */
  			if (query->writer != NULL &&
  			    query->writer->count == 0 &&
  			    query->status == NULL)
  			{
  				query_status(query,
  					     status_noerror,
  					     "no results found for query.");
  			}
  
  			fetch_done(fetch);
  			fetch_unlink(fetch);
  			fetch_reap(fetch);
  		}
  		DEBUG(3, true, "...info read (still %d)\n", still);
  	}
  }
  
  /* escape -- HTML-encode a string, in place.
   */
  void
  escape(CURL *easy, char **str) {
  	char *escaped;
  
  	if (*str == NULL)
  		return;
  	escaped = curl_easy_escape(easy, *str, (int)strlen(*str));
  	if (escaped == NULL) {
  		fprintf(stderr, "%s: curl_escape(%s) failed\n",
  			program_name, *str);
  		my_exit(1);
  	}
  	DESTROY(*str);
  	*str = strdup(escaped);
  	curl_free(escaped);
  	escaped = NULL;
  }
  