modules/proxy/mod_serf.c

   1 /* Licensed to the Apache Software Foundation (ASF) under one or more
   2  * contributor license agreements.  See the NOTICE file distributed with
   3  * this work for additional information regarding copyright ownership.
   4  * The ASF licenses this file to You under the Apache License, Version 2.0
   5  * (the "License"); you may not use this file except in compliance with
   6  * the License.  You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include "httpd.h"
  18 #include "http_core.h"
  19 #include "http_config.h"
  20 #include "http_protocol.h"
  21 #include "http_log.h"
  22
  23 #include "serf.h"
  24 #include "apr_uri.h"
  25
  26 module AP_MODULE_DECLARE_DATA serf_module;
  27
  28 typedef struct {
  29     int on;
  30     apr_uri_t url;
  31 } serf_config_rec;
  32
  33 typedef struct {
  34     int rstatus;
  35     int want_ssl;
  36     int done_headers;
  37     int keep_reading;
  38     request_rec *r;
  39     serf_config_rec *conf;
  40     serf_ssl_context_t *ssl_ctx;
  41     serf_bucket_alloc_t *bkt_alloc;
  42 } s_baton_t;
  43
  44
  45 static void closed_connection(serf_connection_t *conn,
  46                               void *closed_baton,
  47                               apr_status_t why,
  48                               apr_pool_t *pool)
  49 {
  50     s_baton_t *ctx = closed_baton;
  51
  52     if (why) {
  53         /* justin says that error handling isn't done yet. hah. */
  54         /* XXXXXX: review */
  55         ap_log_rerror(APLOG_MARK, APLOG_ERR, why, ctx->r, "Closed Connection Error");
  56         ctx->rstatus = HTTP_INTERNAL_SERVER_ERROR;
  57         return;
  58     }
  59 }
  60
  61 static serf_bucket_t* conn_setup(apr_socket_t *sock,
  62                                  void *setup_baton,
  63                                  apr_pool_t *pool)
  64 {
  65     serf_bucket_t *c;
  66     s_baton_t *ctx = setup_baton;
  67
  68     c = serf_bucket_socket_create(sock, ctx->bkt_alloc);
  69     if (ctx->want_ssl) {
  70         c = serf_bucket_ssl_decrypt_create(c, ctx->ssl_ctx, ctx->bkt_alloc);
  71     }
  72
  73     return c;
  74 }
  75
  76 int copy_headers_in(void *vbaton, const char *key, const char *value)
  77 {
  78     serf_bucket_t *hdrs_bkt = (serf_bucket_t *)vbaton;
  79
  80     /* XXXXX: List of headers not to copy to serf. serf's serf_bucket_headers_setn,
  81      * doesn't actually overwrite a header if we set it once, so we need to ignore anything
  82      * we might want to toggle or combine.
  83      */
  84     switch (key[0]) {
  85     case 'a':
  86     case 'A':
  87         if (strcasecmp("Accept-Encoding", key) == 0) {
  88             return 0;
  89         }
  90         break;
  91     case 'c':
  92     case 'C':
  93         if (strcasecmp("Connection", key) == 0) {
  94             return 0;
  95         }
  96         break;
  97     case 'h':
  98     case 'H':
  99         if (strcasecmp("Host", key) == 0) {
 100             return 0;
 101         }
 102         break;
 103     case 'k':
 104     case 'K':
 105         if (strcasecmp("Keep-Alive", key) == 0) {
 106             return 0;
 107         }
 108         break;
 109     case 't':
 110     case 'T':
 111         if (strcasecmp("TE", key) == 0) {
 112             return 0;
 113         }
 114         if (strcasecmp("Trailer", key) == 0) {
 115             return 0;
 116         }
 117         break;
 118     case 'u':
 119     case 'U':
 120         if (strcasecmp("Upgrade", key) == 0) {
 121             return 0;
 122         }
 123         break;
 124     default:
 125         break;
 126     }
 127
 128     serf_bucket_headers_setn(hdrs_bkt, key, value);
 129     return 0;
 130 }
 131
 132 int copy_headers_out(void *vbaton, const char *key, const char *value)
 133 {
 134     s_baton_t *ctx = vbaton;
 135     int done = 0;
 136
 137     /* XXXXX: Special Treatment required for MANY other headers. fixme.*/
 138     switch (key[0]) {
 139     case 'c':
 140     case 'C':
 141         if (strcasecmp("Content-Type", key) == 0) {
 142             ap_set_content_type(ctx->r, value);
 143             done = 1;
 144             break;
 145         }
 146         else if (strcasecmp("Connection", key) == 0) {
 147             done = 1;
 148             break;
 149         }
 150         else if (strcasecmp("Content-Encoding", key) == 0) {
 151             done = 1;
 152             break;
 153         }
 154         else if (strcasecmp("Content-Length", key) == 0) {
 155             done = 1;
 156             break;
 157         }
 158         break;
 159     case 't':
 160     case 'T':
 161         if (strcasecmp("Transfer-Encoding", key) == 0) {
 162             done = 1;
 163             break;
 164         }
 165         break;
 166     default:
 167             break;
 168     }
 169
 170     if (!done) {
 171         apr_table_addn(ctx->r->headers_out, key, value);
 172     }
 173
 174     return 0;
 175 }
 176
 177 static serf_bucket_t* accept_response(serf_request_t *request,
 178                                       serf_bucket_t *stream,
 179                                       void *acceptor_baton,
 180                                       apr_pool_t *pool)
 181 {
 182     serf_bucket_t *c;
 183     serf_bucket_alloc_t *bkt_alloc;
 184
 185     /* get the per-request bucket allocator */
 186     bkt_alloc = serf_request_get_alloc(request);
 187
 188     /* Create a barrier so the response doesn't eat us! */
 189     c = serf_bucket_barrier_create(stream, bkt_alloc);
 190
 191     return serf_bucket_response_create(c, bkt_alloc);
 192 }
 193
 194 static apr_status_t handle_response(serf_request_t *request,
 195                                     serf_bucket_t *response,
 196                                     void *vbaton,
 197                                     apr_pool_t *pool)
 198 {
 199     apr_status_t rv;
 200     s_baton_t *ctx = vbaton;
 201     const char *data;
 202     apr_size_t len;
 203     serf_status_line sl;
 204
 205     /* XXXXXXX: Create better error message. */
 206     rv = serf_bucket_response_status(response, &sl);
 207     if (rv) {
 208         if (APR_STATUS_IS_EAGAIN(rv)) {
 209             return APR_SUCCESS;
 210         }
 211
 212         ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, ctx->r, "serf_bucket_response_status...");
 213
 214         ctx->rstatus = HTTP_INTERNAL_SERVER_ERROR;
 215
 216         return rv;
 217     }
 218
 219     /**
 220      * XXXXX: If I understood serf buckets better, it might be possible to not
 221      * copy all of the data here, and better stream it to the client.
 222      **/
 223
 224     do {
 225         rv = serf_bucket_read(response, AP_IOBUFSIZE, &data, &len);
 226
 227         if (SERF_BUCKET_READ_ERROR(rv)) {
 228             ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, ctx->r, "serf_bucket_read(response)");
 229             return rv;
 230         }
 231
 232         if (!ctx->done_headers) {
 233             serf_bucket_t *hdrs;
 234             hdrs = serf_bucket_response_get_headers(response);
 235             serf_bucket_headers_do(hdrs, copy_headers_out, ctx);
 236             ctx->done_headers = 1;
 237         }
 238
 239         /* XXXX: write to brigades and stuff. meh */
 240         ap_rwrite(data, len, ctx->r);
 241
 242         if (APR_STATUS_IS_EOF(rv)) {
 243             ctx->keep_reading = 0;
 244             return APR_EOF;
 245         }
 246
 247         /* XXXX: Should we send a flush now? */
 248         if (APR_STATUS_IS_EAGAIN(rv)) {
 249             return APR_SUCCESS;
 250         }
 251
 252     } while (1);
 253 }
 254
 255
 256 static apr_status_t setup_request(serf_request_t *request,
 257                                   void *vbaton,
 258                                   serf_bucket_t **req_bkt,
 259                                   serf_response_acceptor_t *acceptor,
 260                                   void **acceptor_baton,
 261                                   serf_response_handler_t *handler,
 262                                   void **handler_baton,
 263                                   apr_pool_t *pool)
 264 {
 265     s_baton_t *ctx = vbaton;
 266     serf_bucket_t *hdrs_bkt;
 267     serf_bucket_t *body_bkt = NULL;
 268
 269
 270     /* XXXXX: handle incoming request bodies */
 271     *req_bkt = serf_bucket_request_create(ctx->r->method, ctx->r->unparsed_uri, body_bkt,
 272                                           serf_request_get_alloc(request));
 273
 274     hdrs_bkt = serf_bucket_request_get_headers(*req_bkt);
 275
 276     apr_table_do(copy_headers_in, hdrs_bkt, ctx->r->headers_in, NULL);
 277
 278     /* XXXXXX: SerfPreserveHost on */
 279     serf_bucket_headers_setn(hdrs_bkt, "Host", ctx->conf->url.hostname);
 280
 281     serf_bucket_headers_setn(hdrs_bkt, "Accept-Encoding", "gzip");
 282
 283     if (ctx->want_ssl) {
 284         serf_bucket_alloc_t *req_alloc;
 285
 286         req_alloc = serf_request_get_alloc(request);
 287
 288         if (ctx->ssl_ctx == NULL) {
 289             *req_bkt = serf_bucket_ssl_encrypt_create(*req_bkt, NULL,
 290                                            ctx->bkt_alloc);
 291             ctx->ssl_ctx = serf_bucket_ssl_encrypt_context_get(*req_bkt);
 292         }
 293         else {
 294             *req_bkt = serf_bucket_ssl_encrypt_create(*req_bkt, ctx->ssl_ctx,
 295                                                       ctx->bkt_alloc);
 296         }
 297     }
 298
 299     *acceptor = accept_response;
 300     *acceptor_baton = ctx;
 301     *handler = handle_response;
 302     *handler_baton = ctx;
 303
 304     return APR_SUCCESS;
 305 }
 306
 307 static int drive_serf(request_rec *r, serf_config_rec *conf)
 308 {
 309     apr_status_t rv;
 310     apr_pool_t *pool = r->pool;
 311     apr_sockaddr_t *address;
 312     s_baton_t baton;
 313     /* XXXXX: make persistent/per-process or something.*/
 314     serf_context_t *serfme;
 315     serf_connection_t *conn;
 316     serf_request_t *srequest;
 317
 318     /* XXXXX: cache dns? */
 319     rv = apr_sockaddr_info_get(&address, conf->url.hostname,
 320                                APR_UNSPEC, conf->url.port, 0,
 321                                pool);
 322
 323     if (rv != APR_SUCCESS) {
 324         ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, "Unable to resolve: %s", conf->url.hostname);
 325         return HTTP_INTERNAL_SERVER_ERROR;
 326     }
 327
 328     serfme = serf_context_create(pool);
 329
 330     baton.r = r;
 331     baton.conf = conf;
 332     baton.bkt_alloc = serf_bucket_allocator_create(pool, NULL, NULL);
 333     baton.ssl_ctx = NULL;
 334     baton.rstatus = OK;
 335
 336     baton.done_headers = 0;
 337     baton.keep_reading = 1;
 338
 339     if (strcasecmp(conf->url.scheme, "https") == 0) {
 340         baton.want_ssl = 1;
 341     }
 342     else {
 343         baton.want_ssl = 0;
 344     }
 345
 346     conn = serf_connection_create(serfme, address,
 347                                   conn_setup, &baton,
 348                                   closed_connection, &baton,
 349                                   pool);
 350
 351     srequest = serf_connection_request_create(conn, setup_request,
 352                                               &baton);
 353
 354     do {
 355         rv = serf_context_run(serfme, SERF_DURATION_FOREVER, pool);
 356
 357         /* XXXX: Handle timeouts */
 358         if (APR_STATUS_IS_TIMEUP(rv)) {
 359             continue;
 360         }
 361
 362         if (rv != APR_SUCCESS) {
 363             ap_log_rerror(APLOG_MARK, APLOG_ERR, rv, r, "serf_context_run()");
 364             return HTTP_INTERNAL_SERVER_ERROR;
 365         }
 366
 367         serf_debug__closed_conn(baton.bkt_alloc);
 368     } while (baton.keep_reading);
 369
 370     return baton.rstatus;
 371 }
 372
 373 static int serf_handler(request_rec *r)
 374 {
 375     serf_config_rec *conf = ap_get_module_config(r->per_dir_config,
 376                                                  &serf_module);
 377
 378     if (conf->on == 0) {
 379         return DECLINED;
 380     }
 381
 382     return drive_serf(r, conf);
 383 }
 384
 385 static const char *add_pass(cmd_parms *cmd, void *vconf,
 386                             const char *vdest)
 387 {
 388     apr_status_t rv;
 389     serf_config_rec *conf = (serf_config_rec *) vconf;
 390
 391     rv = apr_uri_parse(cmd->pool, vdest, &conf->url);
 392
 393     if (rv != APR_SUCCESS) {
 394         return "mod_serf: Unable to parse SerfPass url.";
 395     }
 396
 397     /* XXXX: These are bugs in apr_uri_parse. Fixme. */
 398     if (!conf->url.port) {
 399         conf->url.port = apr_uri_port_of_scheme(conf->url.scheme);
 400     }
 401
 402     if (!conf->url.path) {
 403         conf->url.path = "/";
 404     }
 405
 406     conf->on = 1;
 407
 408     return NULL;
 409 }
 410
 411 static void *create_config(apr_pool_t *p, char *dummy)
 412 {
 413     serf_config_rec *new = (serf_config_rec *) apr_pcalloc(p, sizeof(serf_config_rec));
 414     new->on = 0;
 415     return new;
 416 }
 417
 418 static const command_rec serf_cmds[] =
 419 {
 420     AP_INIT_TAKE1("SerfPass", add_pass, NULL, OR_INDEXES/*making shit up*/,
 421      "A prefix and destination"),
 422     {NULL}
 423 };
 424
 425 static void register_hooks(apr_pool_t *p)
 426 {
 427     ap_hook_handler(serf_handler, NULL, NULL, APR_HOOK_FIRST);
 428 }
 429
 430 module AP_MODULE_DECLARE_DATA serf_module =
 431 {
 432     STANDARD20_MODULE_STUFF,
 433     create_config,
 434     NULL,
 435     NULL,
 436     NULL,
 437     serf_cmds,
 438     register_hooks
 439 };