text: /**
   * 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.
   *
   */
  
  /*
   * Originally written @ BBC by Graham Leggett
   * Copyright 2009-2011 British Broadcasting Corporation
   *
   */
  
  #include "apr.h"
  #include "apr_lib.h"
  #include "apr_buckets.h"
  #include "apr_file_io.h"
  #include "apr_file_info.h"
  #include "apr_hash.h"
  #include "apr_poll.h"
  #include "apr_portable.h"
  #include "apr_getopt.h"
  #include "apr_signal.h"
  #include "apr_strings.h"
  #include "apr_uuid.h"
  #if APR_HAVE_STDLIB_H
  #include <stdlib.h>
  #endif
  #if APR_HAVE_STRING_H
  #include <string.h>
  #endif
  
  #include "ap_release.h"
  
  #define DEFAULT_MAXLINES 0
  #define DEFAULT_MAXSIZE 0
  #define DEFAULT_AGE 0 * 1000 * 1000
  #define DEFAULT_PREFIX 0
  #define DEFAULT_NONBLOCK 0
  
  typedef struct file_rec
  {
      apr_pool_t *pool;
      apr_file_t *file_err;
      apr_file_t *file_in;
      apr_file_t *file_out;
      const char *directory;
      apr_bucket_alloc_t *alloc;
      apr_bucket_brigade *bb;
      apr_hash_t *request_uuids;
      apr_hash_t *response_uuids;
      apr_hash_t *filters;
      int limit;
      apr_size_t skipped_bytes;
      apr_size_t dropped_fragments;
      apr_time_t start;
      apr_time_t end;
  } file_rec;
  
  typedef struct uuid_rec
  {
      apr_pool_t *pool;
      const char *uuid;
      file_rec *file;
      apr_uint64_t count;
      apr_time_t last;
      apr_size_t offset;
      int direction;
  } uuid_rec;
  
  typedef struct filter_rec
  {
      apr_pool_t *pool;
      const char *prefix;
      apr_size_t len;
  } filter_rec;
  
  typedef struct header_rec
  {
      apr_size_t len;
      apr_time_t timestamp;
      int direction;
      char uuid[APR_UUID_FORMATTED_LENGTH + 1];
      apr_uint64_t count;
      uuid_rec *rec;
  } header_rec;
  
  static const apr_getopt_option_t
          cmdline_opts[] =
          {
                  /* commands */
                  {
                          "file",
                          'f',
                          1,
                          "   --file, -f <name>\t\t\tFile to read the firehose from.\n\t\t\t\t\tDefaults to stdin." },
                  {
                          "output-directory",
                          'd',
                          1,
                          "   --output-directory, -o <name>\tDirectory to write demuxed connections\n\t\t\t\t\tto." },
                  {
                          "uuid",
                          'u',
                          1,
                          "   --uuid, -u <uuid>\t\t\tThe UUID of the connection to\n\t\t\t\t\tdemultiplex. Can be specified more\n\t\t\t\t\tthan once." },
                  /*				{ "output-host", 'h', 1,
                   "   --output-host, -h <hostname>\tHostname to write demuxed connections to." },*/
                  /*				{
                   "speed",
                   's',
                   1,
                   "   --speed, -s <factor>\tSpeed up or slow down demuxing\n\t\t\t\tby the given factor." },*/
                  { "help", 258, 0, "   --help, -h\t\t\t\tThis help text." },
                  { "version", 257, 0,
                          "   --version\t\t\t\tDisplay the version of the program." },
                  { NULL } };
  
  #define HELP_HEADER "Usage : %s [options] [prefix1 [prefix2 ...]]\n\n" \
                      "Firehose demultiplexes the given stream of multiplexed connections, and\n" \
                      "writes each connection to a file, or to a socket as appropriate.\n" \
                      "\n" \
                      "When writing to files, each connection is placed into a dedicated file\n" \
                      "named after the UUID of the connection within the stream. Separate files\n" \
                      "will be created if requests and responses are found in the stream.\n" \
                      "\n" \
                      "If an optional prefix is specified as a parameter, connections that start\n" \
                      "with the given prefix will be included. The prefix needs to fit completely\n" \
                      "within the first fragment for a successful match to occur.\n" \
                      "\n"
  /*                    "When writing to a socket, new connections\n"
   *                    "are opened for each connection in the stream, allowing it to be possible to\n"
   *                    "'replay' traffic recorded by one server to other server.\n"
   *                    "\n\n"
   */
  #define HELP_FOOTER ""
  
  /**
   * Who are we again?
   */
  static void version(const char * const progname)
  {
      printf("%s (%s)\n", progname, AP_SERVER_VERSION);
  }
  
  /**
   * Help the long suffering end user.
   */
  static void help(const char *argv, const char * header, const char *footer,
          const apr_getopt_option_t opts[])
  {
      int i = 0;
  
      if (header) {
          printf(header, argv);
      }
  
      while (opts[i].name) {
          printf("%s\n", opts[i].description);
          i++;
      }
  
      if (footer) {
          printf("%s\n", footer);
      }
  }
  
  /**
   * Cleanup a uuid record. Removes the record from the uuid hashtable in files.
   */
  static apr_status_t cleanup_uuid_rec(void *dummy)
  {
      uuid_rec *rec = (uuid_rec *) dummy;
  
      if (rec->direction == '>') {
          apr_hash_set(rec->file->response_uuids, rec->uuid, APR_HASH_KEY_STRING,
                  NULL);
      }
      if (rec->direction == '<') {
          apr_hash_set(rec->file->request_uuids, rec->uuid, APR_HASH_KEY_STRING,
                  NULL);
      }
  
      return APR_SUCCESS;
  }
  
  /**
   * Create a uuid record, register a cleanup for it's destruction.
   */
  static apr_status_t make_uuid_rec(file_rec *file, header_rec *header,
          uuid_rec **ptr)
  {
      apr_pool_t *pool;
      uuid_rec *rec;
      apr_pool_create(&pool, file->pool);
  
      rec = apr_pcalloc(pool, sizeof(uuid_rec));
      rec->pool = pool;
      rec->file = file;
      rec->uuid = apr_pstrdup(pool, header->uuid);
      rec->count = 0;
      rec->last = header->timestamp;
      rec->direction = header->direction;
  
      if (header->direction == '>') {
          apr_hash_set(file->response_uuids, rec->uuid, APR_HASH_KEY_STRING, rec);
      }
      if (header->direction == '<') {
          apr_hash_set(file->request_uuids, rec->uuid, APR_HASH_KEY_STRING, rec);
      }
  
      apr_pool_cleanup_register(pool, rec, cleanup_uuid_rec, cleanup_uuid_rec);
  
      *ptr = rec;
      return APR_SUCCESS;
  }
  
  /**
   * Process the end of the fragment body.
   *
   * This function renames the completed stream to it's final name.
   */
  static apr_status_t finalise_body(file_rec *file, header_rec *header)
  {
      apr_status_t status;
      char *nfrom, *nto, *from, *to;
      apr_pool_t *pool;
  
      apr_pool_create(&pool, file->pool);
  
      to = apr_pstrcat(pool, header->uuid, header->direction == '>' ? ".response"
              : ".request", NULL);
      from = apr_pstrcat(pool, to, ".part", NULL);
  
      status = apr_filepath_merge(&nfrom, file->directory, from,
              APR_FILEPATH_SECUREROOT, pool);
      if (APR_SUCCESS == status) {
          status = apr_filepath_merge(&nto, file->directory, to,
                  APR_FILEPATH_SECUREROOT, pool);
          if (APR_SUCCESS == status) {
              if (APR_SUCCESS == (status = apr_file_mtime_set(nfrom, file->end, pool))) {
                  if (APR_SUCCESS != (status = apr_file_rename(nfrom, nto, pool))) {
                      apr_file_printf(
                              file->file_err,
                              "Could not rename file '%s' to '%s' for fragment write: %pm\n",
                              nfrom, nto, &status);
                  }
              }
              else {
                  apr_file_printf(
                          file->file_err,
                          "Could not set mtime on file '%s' to '%" APR_TIME_T_FMT "' for fragment write: %pm\n",
                          nfrom, file->end, &status);
              }
          }
          else {
              apr_file_printf(file->file_err,
                      "Could not merge directory '%s' with file '%s': %pm\n",
                      file->directory, to, &status);
          }
      }
      else {
          apr_file_printf(file->file_err,
                  "Could not merge directory '%s' with file '%s': %pm\n",
                  file->directory, from, &status);
      }
  
      apr_pool_destroy(pool);
  
      return status;
  }
  
  /**
   * Check if the fragment matches on of the prefixes.
   */
  static int check_prefix(file_rec *file, header_rec *header, const char *str,
          apr_size_t len)
  {
      apr_hash_index_t *hi;
      void *val;
      apr_pool_t *pool;
      int match = -1;
  
      apr_pool_create(&pool, file->pool);
  
      for (hi = apr_hash_first(pool, file->filters); hi; hi = apr_hash_next(hi)) {
          filter_rec *filter;
          apr_hash_this(hi, NULL, NULL, &val);
          filter = (filter_rec *) val;
  
          if (len > filter->len && !strncmp(filter->prefix, str, filter->len)) {
              match = 1;
              break;
          }
          match = 0;
      }
  
      apr_pool_destroy(pool);
  
      return match;
  }
  
  /**
   * Process part of the fragment body, given the header parameters.
   *
   * Currently, we append it to a file named after the UUID of the connection.
   *
   * The file is opened on demand and closed when done, so that we are
   * guaranteed never to hit a file handle limit (within reason).
   */
  static apr_status_t process_body(file_rec *file, header_rec *header,
          const char *str, apr_size_t len)
  {
      apr_status_t status;
      char *native, *name;
      apr_pool_t *pool;
      apr_file_t *handle;
  
      if (!file->start) {
          file->start = header->timestamp;
      }
      file->end = header->timestamp;
  
      apr_pool_create(&pool, file->pool);
  
      name
              = apr_pstrcat(pool, header->uuid,
                      header->direction == '>' ? ".response.part"
                              : ".request.part", NULL);
  
      status = apr_filepath_merge(&native, file->directory, name,
              APR_FILEPATH_SECUREROOT, pool);
      if (APR_SUCCESS == status) {
          if (APR_SUCCESS == (status = apr_file_open(&handle, native, APR_WRITE
                  | APR_CREATE | APR_APPEND, APR_OS_DEFAULT, pool))) {
              if (APR_SUCCESS != (status = apr_file_write_full(handle, str, len,
                      NULL))) {
                  apr_file_printf(file->file_err,
                          "Could not write fragment body to file '%s': %pm\n",
                          native, &status);
              }
          }
          else {
              apr_file_printf(file->file_err,
                      "Could not open file '%s' for fragment write: %pm\n",
                      native, &status);
          }
      }
      else {
          apr_file_printf(file->file_err,
                  "Could not merge directory '%s' with file '%s': %pm\n",
                  file->directory, name, &status);
      }
  
      apr_pool_destroy(pool);
  
      return status;
  }
  
  /**
   * Parse a chunk extension, detect overflow.
   * There are two error cases:
   *  1) If the conversion would require too many bits, a -1 is returned.
   *  2) If the conversion used the correct number of bits, but an overflow
   *     caused only the sign bit to flip, then that negative number is
   *     returned.
   * In general, any negative number can be considered an overflow error.
   */
  static apr_status_t read_hex(const char **buf, apr_uint64_t *val)
  {
      const char *b = *buf;
      apr_uint64_t chunksize = 0;
      apr_size_t chunkbits = sizeof(apr_uint64_t) * 8;
  
      if (!apr_isxdigit(*b)) {
          return APR_EGENERAL;
      }
      /* Skip leading zeros */
      while (*b == '0') {
          ++b;
      }
  
      while (apr_isxdigit(*b) && (chunkbits > 0)) {
          int xvalue = 0;
  
          if (*b >= '0' && *b <= '9') {
              xvalue = *b - '0';
          }
          else if (*b >= 'A' && *b <= 'F') {
              xvalue = *b - 'A' + 0xa;
          }
          else if (*b >= 'a' && *b <= 'f') {
              xvalue = *b - 'a' + 0xa;
          }
  
          chunksize = (chunksize << 4) | xvalue;
          chunkbits -= 4;
          ++b;
      }
      *buf = b;
      if (apr_isxdigit(*b) && (chunkbits <= 0)) {
          /* overflow */
          return APR_EGENERAL;
      }
  
      *val = chunksize;
  
      return APR_SUCCESS;
  }
  
  /**
   * Parse what might be a fragment header line.
   *
   * If the parse doesn't match for any reason, an error is returned, otherwise
   * APR_SUCCESS.
   *
   * The header structure will be filled with the header values as parsed.
   */
  static apr_status_t process_header(file_rec *file, header_rec *header,
          const char *str, apr_size_t len)
  {
      apr_uint64_t val;
      apr_status_t status;
      int i;
      apr_uuid_t raw;
      const char *end = str + len;
  
      if (APR_SUCCESS != (status = read_hex(&str, &val))) {
          return status;
      }
      header->len = val;
  
      if (!apr_isspace(*(str++))) {
          return APR_EGENERAL;
      }
  
      if (APR_SUCCESS != (status = read_hex(&str, &val))) {
          return status;
      }
      header->timestamp = val;
  
      if (!apr_isspace(*(str++))) {
          return APR_EGENERAL;
      }
  
      if (*str != '<' && *str != '>') {
          return APR_EGENERAL;
      }
      header->direction = *str;
      str++;
  
      if (!apr_isspace(*(str++))) {
          return APR_EGENERAL;
      }
  
      for (i = 0; str[i] && i < APR_UUID_FORMATTED_LENGTH; i++) {
          header->uuid[i] = str[i];
      }
      header->uuid[i] = 0;
      if (apr_uuid_parse(&raw, header->uuid)) {
          return APR_EGENERAL;
      }
      str += i;
  
      if (!apr_isspace(*(str++))) {
          return APR_EGENERAL;
      }
  
      if (APR_SUCCESS != (status = read_hex(&str, &val))) {
          return status;
      }
      header->count = val;
  
      if ((*(str++) != '\r')) {
          return APR_EGENERAL;
      }
      if ((*(str++) != '\n')) {
          return APR_EGENERAL;
      }
      if (str != end) {
          return APR_EGENERAL;
      }
  
      return APR_SUCCESS;
  }
  
  /**
   * Suck on the file/pipe, and demux any fragments on the incoming stream.
   *
   * If EOF is detected, this function returns.
   */
  static apr_status_t demux(file_rec *file)
  {
      apr_size_t len = 0;
      apr_status_t status = APR_SUCCESS;
      apr_bucket *b, *e;
      apr_bucket_brigade *bb, *obb;
      int footer = 0;
      const char *buf;
  
      bb = apr_brigade_create(file->pool, file->alloc);
      obb = apr_brigade_create(file->pool, file->alloc);
      b = apr_bucket_pipe_create(file->file_in, file->alloc);
  
      APR_BRIGADE_INSERT_HEAD(bb, b);
  
      do {
  
          /* when the pipe is closed, the pipe disappears from the brigade */
          if (APR_BRIGADE_EMPTY(bb)) {
              break;
          }
  
          status = apr_brigade_split_line(obb, bb, APR_BLOCK_READ,
                  HUGE_STRING_LEN);
  
          if (APR_SUCCESS == status || APR_EOF == status) {
              char str[HUGE_STRING_LEN];
              len = HUGE_STRING_LEN;
  
              apr_brigade_flatten(obb, str, &len);
  
              apr_brigade_cleanup(obb);
  
              if (len == HUGE_STRING_LEN) {
                  file->skipped_bytes += len;
                  continue;
              }
              else if (footer) {
                  if (len == 2 && str[0] == '\r' && str[1] == '\n') {
                      footer = 0;
                      continue;
                  }
                  file->skipped_bytes += len;
              }
              else if (len > 0) {
                  header_rec header;
                  status = process_header(file, &header, str, len);
                  if (APR_SUCCESS != status) {
                      file->skipped_bytes += len;
                      continue;
                  }
                  else {
                      int ignore = 0;
  
                      header.rec = NULL;
                      if (header.direction == '>') {
                          header.rec = apr_hash_get(file->response_uuids,
                                  header.uuid, APR_HASH_KEY_STRING);
                      }
                      if (header.direction == '<') {
                          header.rec = apr_hash_get(file->request_uuids,
                                  header.uuid, APR_HASH_KEY_STRING);
                      }
                      if (header.rec) {
                          /* does the count match what is expected? */
                          if (header.count != header.rec->count) {
                              file->dropped_fragments++;
                              ignore = 1;
                          }
                      }
                      else {
                          /* must we ignore unknown uuids? */
                          if (file->limit) {
                              ignore = 1;
                          }
  
                          /* is the counter not what we expect? */
                          else if (header.count != 0) {
                              file->skipped_bytes += len;
                              ignore = 1;
                          }
  
                          /* otherwise, make a new uuid */
                          else {
                              make_uuid_rec(file, &header, &header.rec);
                          }
                      }
  
                      if (header.len) {
                          if (APR_SUCCESS != (status = apr_brigade_partition(bb,
                                  header.len, &e))) {
                              apr_file_printf(
                                      file->file_err,
                                      "Could not read fragment body from input file: %pm\n", &status);
                              break;
                          }
                          while ((b = APR_BRIGADE_FIRST(bb)) && e != b) {
                              apr_bucket_read(b, &buf, &len, APR_READ_BLOCK);
                              if (!ignore && !header.count && !check_prefix(file,
                                      &header, buf, len)) {
                                  ignore = 1;
                              }
                              if (!ignore) {
                                  status = process_body(file, &header, buf, len);
                                  header.rec->offset += len;
                              }
                              if (ignore || APR_SUCCESS != status) {
                                  apr_bucket_delete(b);
                                  file->skipped_bytes += len;
                                  continue;
                              }
                              apr_bucket_delete(b);
                          }
                          if (!ignore) {
                              header.rec->count++;
                          }
                          footer = 1;
                          continue;
                      }
                      else {
                          /* an empty header means end-of-connection */
                          if (header.rec) {
                              if (!ignore) {
                                  if (!header.count) {
                                      status = process_body(file, &header, "", 0);
                                  }
                                  status = finalise_body(file, &header);
                              }
                              apr_pool_destroy(header.rec->pool);
                          }
                      }
  
                  }
              }
  
          }
          else {
              apr_file_printf(file->file_err,
                      "Could not read fragment header from input file: %pm\n", &status);
              break;
          }
  
      } while (1);
  
      return status;
  }
  
  /**
   * Start the application.
   */
  int main(int argc, const char * const argv[])
  {
      apr_status_t status;
      apr_pool_t *pool;
      apr_getopt_t *opt;
      int optch;
      const char *optarg;
  
      file_rec *file;
  
      /* lets get APR off the ground, and make sure it terminates cleanly */
      if (APR_SUCCESS != (status = apr_app_initialize(&argc, &argv, NULL))) {
          return 1;
      }
      atexit(apr_terminate);
  
      if (APR_SUCCESS != (status = apr_pool_create(&pool, NULL))) {
          return 1;
      }
  
  #ifdef SIGPIPE
      apr_signal_block(SIGPIPE);
  #endif
  
      file = apr_pcalloc(pool, sizeof(file_rec));
      apr_file_open_stderr(&file->file_err, pool);
      apr_file_open_stdin(&file->file_in, pool);
      apr_file_open_stdout(&file->file_out, pool);
  
      file->pool = pool;
      file->alloc = apr_bucket_alloc_create(pool);
      file->bb = apr_brigade_create(pool, file->alloc);
      file->request_uuids = apr_hash_make(pool);
      file->response_uuids = apr_hash_make(pool);
      file->filters = apr_hash_make(pool);
  
      apr_getopt_init(&opt, pool, argc, argv);
      while ((status = apr_getopt_long(opt, cmdline_opts, &optch, &optarg))
              == APR_SUCCESS) {
  
          switch (optch) {
          case 'f': {
              status = apr_file_open(&file->file_in, optarg, APR_FOPEN_READ,
                      APR_OS_DEFAULT, pool);
              if (status != APR_SUCCESS) {
                  apr_file_printf(file->file_err,
                          "Could not open file '%s' for read: %pm\n", optarg, &status);
                  return 1;
              }
              break;
          }
          case 'd': {
              apr_finfo_t finfo;
              status = apr_stat(&finfo, optarg, APR_FINFO_TYPE, pool);
              if (status != APR_SUCCESS) {
                  apr_file_printf(file->file_err,
                          "Directory '%s' could not be found: %pm\n", optarg, &status);
                  return 1;
              }
              if (finfo.filetype != APR_DIR) {
                  apr_file_printf(file->file_err,
                          "Path '%s' isn't a directory\n", optarg);
                  return 1;
              }
              file->directory = optarg;
              break;
          }
          case 'u': {
              apr_pool_t *pchild;
              uuid_rec *rec;
              apr_pool_create(&pchild, pool);
              rec = apr_pcalloc(pchild, sizeof(uuid_rec));
              rec->pool = pchild;
              rec->uuid = optarg;
              apr_hash_set(file->request_uuids, optarg, APR_HASH_KEY_STRING, rec);
              apr_hash_set(file->response_uuids, optarg, APR_HASH_KEY_STRING, rec);
              file->limit++;
              break;
          }
          case 257: {
              version(argv[0]);
              return 0;
          }
          case 258: {
              help(argv[0], HELP_HEADER, HELP_FOOTER, cmdline_opts);
              return 0;
  
          }
          }
  
      }
      if (APR_SUCCESS != status && APR_EOF != status) {
          return 1;
      }
  
      /* read filters from the command line */
      while (opt->ind < argc) {
          apr_pool_t *pchild;
          filter_rec *filter;
          apr_pool_create(&pchild, pool);
          filter = apr_pcalloc(pchild, sizeof(filter_rec));
          filter->pool = pchild;
          filter->prefix = opt->argv[opt->ind];
          filter->len = strlen(opt->argv[opt->ind]);
          apr_hash_set(file->filters, opt->argv[opt->ind], APR_HASH_KEY_STRING,
                  filter);
          opt->ind++;
      }
  
      status = demux(file);
  
      /* warn people if any non blocking writes failed */
      if (file->skipped_bytes || file->dropped_fragments) {
          apr_file_printf(
                  file->file_err,
                  "Warning: %" APR_SIZE_T_FMT " bytes skipped, %" APR_SIZE_T_FMT " fragments dropped.\n",
                  file->skipped_bytes, file->dropped_fragments);
      }
  
      if (APR_SUCCESS != status) {
          return 1;
      }
  
      return 0;
  }
  