contrib/serf/buckets/deflate_buckets.c

   1 /* ====================================================================
   2  *    Licensed to the Apache Software Foundation (ASF) under one
   3  *    or more contributor license agreements.  See the NOTICE file
   4  *    distributed with this work for additional information
   5  *    regarding copyright ownership.  The ASF licenses this file
   6  *    to you under the Apache License, Version 2.0 (the
   7  *    "License"); you may not use this file except in compliance
   8  *    with the License.  You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  *    Unless required by applicable law or agreed to in writing,
  13  *    software distributed under the License is distributed on an
  14  *    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  15  *    KIND, either express or implied.  See the License for the
  16  *    specific language governing permissions and limitations
  17  *    under the License.
  18  * ====================================================================
  19  */
  20
  21 #include <apr_strings.h>
  22
  23 #include <zlib.h>
  24
  25 /* This conditional isn't defined anywhere yet. */
  26 #ifdef HAVE_ZUTIL_H
  27 #include <zutil.h>
  28 #endif
  29
  30 #include "serf.h"
  31 #include "serf_bucket_util.h"
  32
  33 /* magic header */
  34 static char deflate_magic[2] = { '\037', '\213' };
  35 #define DEFLATE_MAGIC_SIZE 10
  36 #define DEFLATE_VERIFY_SIZE 8
  37 #define DEFLATE_BUFFER_SIZE 8096
  38
  39 static const int DEFLATE_WINDOW_SIZE = -15;
  40 static const int DEFLATE_MEMLEVEL = 9;
  41
  42 typedef struct {
  43     serf_bucket_t *stream;
  44     serf_bucket_t *inflate_stream;
  45
  46     int format;                 /* Are we 'deflate' or 'gzip'? */
  47
  48     enum {
  49         STATE_READING_HEADER,   /* reading the gzip header */
  50         STATE_HEADER,           /* read the gzip header */
  51         STATE_INIT,             /* init'ing zlib functions */
  52         STATE_INFLATE,          /* inflating the content now */
  53         STATE_READING_VERIFY,   /* reading the final gzip CRC */
  54         STATE_VERIFY,           /* verifying the final gzip CRC */
  55         STATE_FINISH,           /* clean up after reading body */
  56         STATE_DONE,             /* body is done; we'll return EOF here */
  57     } state;
  58
  59     z_stream zstream;
  60     char hdr_buffer[DEFLATE_MAGIC_SIZE];
  61     unsigned char buffer[DEFLATE_BUFFER_SIZE];
  62     unsigned long crc;
  63     int windowSize;
  64     int memLevel;
  65     int bufferSize;
  66
  67     /* How much of the chunk, or the terminator, do we have left to read? */
  68     apr_size_t stream_left;
  69
  70     /* How much are we supposed to read? */
  71     apr_size_t stream_size;
  72
  73     int stream_status; /* What was the last status we read? */
  74
  75 } deflate_context_t;
  76
  77 /* Inputs a string and returns a long.  */
  78 static unsigned long getLong(unsigned char *string)
  79 {
  80     return ((unsigned long)string[0])
  81           | (((unsigned long)string[1]) << 8)
  82           | (((unsigned long)string[2]) << 16)
  83           | (((unsigned long)string[3]) << 24);
  84 }
  85
  86 serf_bucket_t *serf_bucket_deflate_create(
  87     serf_bucket_t *stream,
  88     serf_bucket_alloc_t *allocator,
  89     int format)
  90 {
  91     deflate_context_t *ctx;
  92
  93     ctx = serf_bucket_mem_alloc(allocator, sizeof(*ctx));
  94     ctx->stream = stream;
  95     ctx->stream_status = APR_SUCCESS;
  96     ctx->inflate_stream = serf_bucket_aggregate_create(allocator);
  97     ctx->format = format;
  98     ctx->crc = 0;
  99     /* zstream must be NULL'd out. */
 100     memset(&ctx->zstream, 0, sizeof(ctx->zstream));
 101
 102     switch (ctx->format) {
 103         case SERF_DEFLATE_GZIP:
 104             ctx->state = STATE_READING_HEADER;
 105             break;
 106         case SERF_DEFLATE_DEFLATE:
 107             /* deflate doesn't have a header. */
 108             ctx->state = STATE_INIT;
 109             break;
 110         default:
 111             /* Not reachable */
 112             return NULL;
 113     }
 114
 115     /* Initial size of gzip header. */
 116     ctx->stream_left = ctx->stream_size = DEFLATE_MAGIC_SIZE;
 117
 118     ctx->windowSize = DEFLATE_WINDOW_SIZE;
 119     ctx->memLevel = DEFLATE_MEMLEVEL;
 120     ctx->bufferSize = DEFLATE_BUFFER_SIZE;
 121
 122     return serf_bucket_create(&serf_bucket_type_deflate, allocator, ctx);
 123 }
 124
 125 static void serf_deflate_destroy_and_data(serf_bucket_t *bucket)
 126 {
 127     deflate_context_t *ctx = bucket->data;
 128
 129     if (ctx->state > STATE_INIT &&
 130         ctx->state <= STATE_FINISH)
 131         inflateEnd(&ctx->zstream);
 132
 133     /* We may have appended inflate_stream into the stream bucket.
 134      * If so, avoid free'ing it twice.
 135      */
 136     if (ctx->inflate_stream) {
 137         serf_bucket_destroy(ctx->inflate_stream);
 138     }
 139     serf_bucket_destroy(ctx->stream);
 140
 141     serf_default_destroy_and_data(bucket);
 142 }
 143
 144 static apr_status_t serf_deflate_read(serf_bucket_t *bucket,
 145                                       apr_size_t requested,
 146                                       const char **data, apr_size_t *len)
 147 {
 148     deflate_context_t *ctx = bucket->data;
 149     apr_status_t status;
 150     const char *private_data;
 151     apr_size_t private_len;
 152     int zRC;
 153
 154     while (1) {
 155         switch (ctx->state) {
 156         case STATE_READING_HEADER:
 157         case STATE_READING_VERIFY:
 158             status = serf_bucket_read(ctx->stream, ctx->stream_left,
 159                                       &private_data, &private_len);
 160
 161             if (SERF_BUCKET_READ_ERROR(status)) {
 162                 return status;
 163             }
 164
 165             memcpy(ctx->hdr_buffer + (ctx->stream_size - ctx->stream_left),
 166                    private_data, private_len);
 167
 168             ctx->stream_left -= private_len;
 169
 170             if (ctx->stream_left == 0) {
 171                 ctx->state++;
 172                 if (APR_STATUS_IS_EAGAIN(status)) {
 173                     *len = 0;
 174                     return status;
 175                 }
 176             }
 177             else if (status) {
 178                 *len = 0;
 179                 return status;
 180             }
 181             break;
 182         case STATE_HEADER:
 183             if (ctx->hdr_buffer[0] != deflate_magic[0] ||
 184                 ctx->hdr_buffer[1] != deflate_magic[1]) {
 185                 return SERF_ERROR_DECOMPRESSION_FAILED;
 186             }
 187             if (ctx->hdr_buffer[3] != 0) {
 188                 return SERF_ERROR_DECOMPRESSION_FAILED;
 189             }
 190             ctx->state++;
 191             break;
 192         case STATE_VERIFY:
 193         {
 194             unsigned long compCRC, compLen, actualLen;
 195
 196             /* Do the checksum computation. */
 197             compCRC = getLong((unsigned char*)ctx->hdr_buffer);
 198             if (ctx->crc != compCRC) {
 199                 return SERF_ERROR_DECOMPRESSION_FAILED;
 200             }
 201             compLen = getLong((unsigned char*)ctx->hdr_buffer + 4);
 202             /* The length in the trailer is module 2^32, so do the same for
 203                the actual length. */
 204             actualLen = ctx->zstream.total_out;
 205             actualLen &= 0xFFFFFFFF;
 206             if (actualLen != compLen) {
 207                 return SERF_ERROR_DECOMPRESSION_FAILED;
 208             }
 209             ctx->state++;
 210             break;
 211         }
 212         case STATE_INIT:
 213             zRC = inflateInit2(&ctx->zstream, ctx->windowSize);
 214             if (zRC != Z_OK) {
 215                 return SERF_ERROR_DECOMPRESSION_FAILED;
 216             }
 217             ctx->zstream.next_out = ctx->buffer;
 218             ctx->zstream.avail_out = ctx->bufferSize;
 219             ctx->state++;
 220             break;
 221         case STATE_FINISH:
 222             inflateEnd(&ctx->zstream);
 223             serf_bucket_aggregate_prepend(ctx->stream, ctx->inflate_stream);
 224             ctx->inflate_stream = 0;
 225             ctx->state++;
 226             break;
 227         case STATE_INFLATE:
 228             /* Do we have anything already uncompressed to read? */
 229             status = serf_bucket_read(ctx->inflate_stream, requested, data,
 230                                       len);
 231             if (SERF_BUCKET_READ_ERROR(status)) {
 232                 return status;
 233             }
 234             /* Hide EOF. */
 235             if (APR_STATUS_IS_EOF(status)) {
 236                 status = ctx->stream_status;
 237                 if (APR_STATUS_IS_EOF(status)) {
 238                     /* We've read all of the data from our stream, but we
 239                      * need to continue to iterate until we flush
 240                      * out the zlib buffer.
 241                      */
 242                     status = APR_SUCCESS;
 243                 }
 244             }
 245             if (*len != 0) {
 246                 return status;
 247             }
 248
 249             /* We tried; but we have nothing buffered. Fetch more. */
 250
 251             /* It is possible that we maxed out avail_out before
 252              * exhausting avail_in; therefore, continue using the
 253              * previous buffer.  Otherwise, fetch more data from
 254              * our stream bucket.
 255              */
 256             if (ctx->zstream.avail_in == 0) {
 257                 /* When we empty our inflated stream, we'll return this
 258                  * status - this allow us to eventually pass up EAGAINs.
 259                  */
 260                 ctx->stream_status = serf_bucket_read(ctx->stream,
 261                                                       ctx->bufferSize,
 262                                                       &private_data,
 263                                                       &private_len);
 264
 265                 if (SERF_BUCKET_READ_ERROR(ctx->stream_status)) {
 266                     return ctx->stream_status;
 267                 }
 268
 269                 if (!private_len && APR_STATUS_IS_EAGAIN(ctx->stream_status)) {
 270                     *len = 0;
 271                     status = ctx->stream_status;
 272                     ctx->stream_status = APR_SUCCESS;
 273                     return status;
 274                 }
 275
 276                 ctx->zstream.next_in = (unsigned char*)private_data;
 277                 ctx->zstream.avail_in = private_len;
 278             }
 279
 280             while (1) {
 281
 282                 zRC = inflate(&ctx->zstream, Z_NO_FLUSH);
 283
 284                 /* We're full or zlib requires more space. Either case, clear
 285                    out our buffer, reset, and return. */
 286                 if (zRC == Z_BUF_ERROR || ctx->zstream.avail_out == 0) {
 287                     serf_bucket_t *tmp;
 288                     ctx->zstream.next_out = ctx->buffer;
 289                     private_len = ctx->bufferSize - ctx->zstream.avail_out;
 290
 291                     ctx->crc = crc32(ctx->crc, (const Bytef *)ctx->buffer,
 292                                      private_len);
 293
 294                     /* FIXME: There probably needs to be a free func. */
 295                     tmp = SERF_BUCKET_SIMPLE_STRING_LEN((char *)ctx->buffer,
 296                                                         private_len,
 297                                                         bucket->allocator);
 298                     serf_bucket_aggregate_append(ctx->inflate_stream, tmp);
 299                     ctx->zstream.avail_out = ctx->bufferSize;
 300                     break;
 301                 }
 302
 303                 if (zRC == Z_STREAM_END) {
 304                     serf_bucket_t *tmp;
 305
 306                     private_len = ctx->bufferSize - ctx->zstream.avail_out;
 307                     ctx->crc = crc32(ctx->crc, (const Bytef *)ctx->buffer,
 308                                      private_len);
 309                     /* FIXME: There probably needs to be a free func. */
 310                     tmp = SERF_BUCKET_SIMPLE_STRING_LEN((char *)ctx->buffer,
 311                                                         private_len,
 312                                                         bucket->allocator);
 313                     serf_bucket_aggregate_append(ctx->inflate_stream, tmp);
 314
 315                     ctx->zstream.avail_out = ctx->bufferSize;
 316
 317                     /* Push back the remaining data to be read. */
 318                     tmp = serf_bucket_aggregate_create(bucket->allocator);
 319                     serf_bucket_aggregate_prepend(tmp, ctx->stream);
 320                     ctx->stream = tmp;
 321
 322                     /* We now need to take the remaining avail_in and
 323                      * throw it in ctx->stream so our next read picks it up.
 324                      */
 325                     tmp = SERF_BUCKET_SIMPLE_STRING_LEN(
 326                                         (const char*)ctx->zstream.next_in,
 327                                                      ctx->zstream.avail_in,
 328                                                      bucket->allocator);
 329                     serf_bucket_aggregate_prepend(ctx->stream, tmp);
 330
 331                     switch (ctx->format) {
 332                     case SERF_DEFLATE_GZIP:
 333                         ctx->stream_left = ctx->stream_size =
 334                             DEFLATE_VERIFY_SIZE;
 335                         ctx->state++;
 336                         break;
 337                     case SERF_DEFLATE_DEFLATE:
 338                         /* Deflate does not have a verify footer. */
 339                         ctx->state = STATE_FINISH;
 340                         break;
 341                     default:
 342                         /* Not reachable */
 343                         return APR_EGENERAL;
 344                     }
 345
 346                     break;
 347                 }
 348
 349                 /* Any other error? */
 350                 if (zRC != Z_OK) {
 351                     return SERF_ERROR_DECOMPRESSION_FAILED;
 352                 }
 353
 354                 /* As long as zRC == Z_OK, just keep looping. */
 355             }
 356             /* Okay, we've inflated.  Try to read. */
 357             status = serf_bucket_read(ctx->inflate_stream, requested, data,
 358                                       len);
 359             /* Hide EOF. */
 360             if (APR_STATUS_IS_EOF(status)) {
 361                 status = ctx->stream_status;
 362
 363                 /* If the inflation wasn't finished, return APR_SUCCESS. */
 364                 if (zRC != Z_STREAM_END)
 365                     return APR_SUCCESS;
 366
 367                 /* If our stream is finished too and all data was inflated,
 368                  * return SUCCESS so we'll iterate one more time.
 369                  */
 370                 if (APR_STATUS_IS_EOF(status)) {
 371                     /* No more data to read from the stream, and everything
 372                        inflated. If all data was received correctly, state
 373                        should have been advanced to STATE_READING_VERIFY or
 374                        STATE_FINISH. If not, then the data was incomplete
 375                        and we have an error. */
 376                     if (ctx->state != STATE_INFLATE)
 377                         return APR_SUCCESS;
 378                     else
 379                         return SERF_ERROR_DECOMPRESSION_FAILED;
 380                 }
 381             }
 382             return status;
 383         case STATE_DONE:
 384             /* We're done inflating.  Use our finished buffer. */
 385             return serf_bucket_read(ctx->stream, requested, data, len);
 386         default:
 387             /* Not reachable */
 388             return APR_EGENERAL;
 389         }
 390     }
 391
 392     /* NOTREACHED */
 393 }
 394
 395 /* ### need to implement */
 396 #define serf_deflate_readline NULL
 397 #define serf_deflate_peek NULL
 398
 399 const serf_bucket_type_t serf_bucket_type_deflate = {
 400     "DEFLATE",
 401     serf_deflate_read,
 402     serf_deflate_readline,
 403     serf_default_read_iovec,
 404     serf_default_read_for_sendfile,
 405     serf_default_read_bucket,
 406     serf_deflate_peek,
 407     serf_deflate_destroy_and_data,
 408 };