src/gateway/osrf_http_translator.c

   1 #include <sys/time.h>
   2 #include <sys/resource.h>
   3 #include <unistd.h>
   4 #include <strings.h>
   5 #include "apachetools.h"
   6 #include <opensrf/osrf_app_session.h>
   7 #include <opensrf/osrf_system.h>
   8 #include <opensrf/osrfConfig.h>
   9 #include <opensrf/osrf_json.h>
  10 #include <opensrf/osrf_cache.h>
  11
  12 #define MODULE_NAME "osrf_http_translator_module"
  13 #define OSRF_TRANSLATOR_CONFIG_FILE "OSRFTranslatorConfig"
  14 #define OSRF_TRANSLATOR_CONFIG_CTX "OSRFTranslatorConfigContext"
  15 #define OSRF_TRANSLATOR_CACHE_SERVER "OSRFTranslatorCacheServer"
  16
  17 #define DEFAULT_TRANSLATOR_CONFIG_CTX "gateway"
  18 #define DEFAULT_TRANSLATOR_CONFIG_FILE "/openils/conf/opensrf_core.xml"
  19 #define DEFAULT_TRANSLATOR_TIMEOUT 1200
  20 #define DEFAULT_TRANSLATOR_CACHE_SERVERS "127.0.0.1:11211"
  21
  22 #define MULTIPART_CONTENT_TYPE "multipart/x-mixed-replace;boundary=\"%s\""
  23 #define JSON_CONTENT_TYPE "text/plain"
  24 #define MAX_MSGS_PER_PACKET 256
  25 #define CACHE_TIME 300
  26
  27 #define OSRF_HTTP_HEADER_TO "X-OpenSRF-to"
  28 #define OSRF_HTTP_HEADER_XID "X-OpenSRF-xid"
  29 #define OSRF_HTTP_HEADER_FROM "X-OpenSRF-from"
  30 #define OSRF_HTTP_HEADER_THREAD "X-OpenSRF-thread"
  31 #define OSRF_HTTP_HEADER_TIMEOUT "X-OpenSRF-timeout"
  32 #define OSRF_HTTP_HEADER_SERVICE "X-OpenSRF-service"
  33 #define OSRF_HTTP_HEADER_MULTIPART "X-OpenSRF-multipart"
  34
  35
  36 char* configFile = DEFAULT_TRANSLATOR_CONFIG_FILE;
  37 char* configCtx = DEFAULT_TRANSLATOR_CONFIG_CTX;
  38 char* cacheServers = DEFAULT_TRANSLATOR_CACHE_SERVERS;
  39
  40 char* routerName = NULL;
  41 char* domainName = NULL;
  42 int osrfConnected = 0;
  43 char recipientBuf[128];
  44 char contentTypeBuf[80];
  45
  46 // for development only, writes to apache error log
  47 static void _dbg(char* s, ...) {
  48     VA_LIST_TO_STRING(s);
  49     fprintf(stderr, "%s\n", VA_BUF);
  50     fflush(stderr);
  51 }
  52
  53 // Translator struct
  54 typedef struct {
  55     request_rec* apreq;
  56     transport_client* handle;
  57     osrfList* messages;
  58     char* body;
  59     char* delim;
  60     const char* recipient;
  61     const char* service;
  62     const char* thread;
  63     const char* remoteHost;
  64     int complete;
  65     int timeout;
  66     int multipart;
  67     int connectOnly; // there is only 1 message, a CONNECT
  68     int disconnectOnly; // there is only 1 message, a DISCONNECT
  69     int connecting; // there is a connect message in this batch
  70     int disconnecting; // there is a connect message in this batch
  71     int localXid;
  72 } osrfHttpTranslator;
  73
  74
  75 static const char* osrfHttpTranslatorGetConfigFile(cmd_parms *parms, void *config, const char *arg) {
  76     configFile = (char*) arg;
  77         return NULL;
  78 }
  79 static const char* osrfHttpTranslatorGetConfigFileCtx(cmd_parms *parms, void *config, const char *arg) {
  80     configCtx = (char*) arg;
  81         return NULL;
  82 }
  83 static const char* osrfHttpTranslatorGetCacheServer(cmd_parms *parms, void *config, const char *arg) {
  84     cacheServers = (char*) arg;
  85         return NULL;
  86 }
  87
  88 /** set up the configuration handlers */
  89 static const command_rec osrfHttpTranslatorCmds[] = {
  90         AP_INIT_TAKE1( OSRF_TRANSLATOR_CONFIG_FILE, osrfHttpTranslatorGetConfigFile,
  91                         NULL, RSRC_CONF, "osrf translator config file"),
  92         AP_INIT_TAKE1( OSRF_TRANSLATOR_CONFIG_CTX, osrfHttpTranslatorGetConfigFileCtx,
  93                         NULL, RSRC_CONF, "osrf translator config file context"),
  94         AP_INIT_TAKE1( OSRF_TRANSLATOR_CACHE_SERVER, osrfHttpTranslatorGetCacheServer,
  95                         NULL, RSRC_CONF, "osrf translator cache server"),
  96     {NULL}
  97 };
  98
  99
 100 // there can only be one, so use a global static one
 101 static osrfHttpTranslator globalTranslator;
 102
 103 /*
 104  * Constructs a new translator object based on the current apache
 105  * request_rec.  Reads the request body and headers.
 106  */
 107 static osrfHttpTranslator* osrfNewHttpTranslator(request_rec* apreq) {
 108     osrfHttpTranslator* trans = &globalTranslator;
 109     trans->apreq = apreq;
 110     trans->complete = 0;
 111     trans->connectOnly = 0;
 112     trans->disconnectOnly = 0;
 113     trans->connecting = 0;
 114     trans->disconnecting = 0;
 115     trans->remoteHost = apreq->connection->remote_ip;
 116     trans->messages = NULL;
 117
 118     /* load the message body */
 119         osrfStringArray* params = apacheParseParms(apreq);
 120     trans->body = apacheGetFirstParamValue(params, "osrf-msg");
 121     osrfStringArrayFree(params);
 122
 123     /* load the request headers */
 124     if (apr_table_get(apreq->headers_in, OSRF_HTTP_HEADER_XID)) // force our log xid to match the caller
 125             osrfLogForceXid(strdup(apr_table_get(apreq->headers_in, OSRF_HTTP_HEADER_XID)));
 126
 127     trans->handle = osrfSystemGetTransportClient();
 128     trans->recipient = apr_table_get(apreq->headers_in, OSRF_HTTP_HEADER_TO);
 129     trans->service = apr_table_get(apreq->headers_in, OSRF_HTTP_HEADER_SERVICE);
 130     trans->thread = apr_table_get(apreq->headers_in, OSRF_HTTP_HEADER_THREAD); /* XXX create thread if necessary */
 131
 132     const char* timeout = apr_table_get(apreq->headers_in, OSRF_HTTP_HEADER_TIMEOUT);
 133     if(timeout)
 134         trans->timeout = atoi(timeout);
 135     else
 136         trans->timeout = DEFAULT_TRANSLATOR_TIMEOUT;
 137
 138     const char* multipart = apr_table_get(apreq->headers_in, OSRF_HTTP_HEADER_MULTIPART);
 139     if(multipart && !strcasecmp(multipart, "true"))
 140         trans->multipart = 1;
 141     else
 142         trans->multipart = 0;
 143
 144     char buf[32];
 145     snprintf(buf, sizeof(buf), "%d%ld", getpid(), time(NULL));
 146     trans->delim = md5sum(buf);
 147
 148     return trans;
 149 }
 150
 151 static void osrfHttpTranslatorFree(osrfHttpTranslator* trans) {
 152     if(!trans) return;
 153     if(trans->body)
 154         free(trans->body);
 155     if(trans->delim)
 156         free(trans->delim);
 157     osrfListFree(trans->messages);
 158 }
 159
 160 static void osrfHttpTranslatorDebug(osrfHttpTranslator* trans) {
 161     _dbg("-----------------------------------");
 162     _dbg("body = %s", trans->body);
 163     _dbg("service = %s", trans->service);
 164     _dbg("thread = %s", trans->thread);
 165     _dbg("multipart = %d", trans->multipart);
 166     _dbg("recipient = %s", trans->recipient);
 167 }
 168
 169 /**
 170  * Determines the correct recipient address based on the requested
 171  * service or recipient address.
 172  */
 173 static int osrfHttpTranslatorSetTo(osrfHttpTranslator* trans) {
 174     int stat = 0;
 175     jsonObject* sessionCache = NULL;
 176
 177     if(trans->service) {
 178         if(trans->recipient) {
 179             osrfLogError(OSRF_LOG_MARK, "Specifying both SERVICE and TO are not allowed");
 180
 181         } else {
 182             // service is specified, build a recipient address
 183             // from the router, domain, and service
 184             int size = snprintf(recipientBuf, 128, "%s@%s/%s", routerName, domainName, trans->service);
 185             recipientBuf[size] = '\0';
 186             osrfLogDebug(OSRF_LOG_MARK, "Set recipient to %s", recipientBuf);
 187             trans->recipient = recipientBuf;
 188             stat = 1;
 189         }
 190
 191     } else {
 192
 193         if(trans->recipient) {
 194             sessionCache = osrfCacheGetObject(trans->thread);
 195
 196             if(sessionCache) {
 197                 const char* ipAddr = jsonObjectGetString(jsonObjectGetKey(sessionCache, "ip"));
 198                 const char* recipient = jsonObjectGetString(jsonObjectGetKey(sessionCache, "jid"));
 199
 200                 // choosing a specific recipient address requires that the recipient and
 201                 // thread be cached on the server (so drone processes cannot be hijacked)
 202                 if(!strcmp(ipAddr, trans->remoteHost) && !strcmp(recipient, trans->recipient)) {
 203                     osrfLogDebug(OSRF_LOG_MARK, "Found cached session from host %s and recipient %s",
 204                         trans->remoteHost, trans->recipient);
 205                     stat = 1;
 206                     trans->service = apr_pstrdup(
 207                         trans->apreq->pool, jsonObjectGetString(jsonObjectGetKey(sessionCache, "service")));
 208
 209                 } else {
 210                     osrfLogError(OSRF_LOG_MARK,
 211                         "Session cache for thread %s does not match request", trans->thread);
 212                 }
 213             }  else {
 214                 osrfLogError(OSRF_LOG_MARK,
 215                     "attempt to send directly to %s without a session", trans->recipient);
 216             }
 217         } else {
 218             osrfLogError(OSRF_LOG_MARK, "No SERVICE or RECIPIENT defined");
 219         }
 220     }
 221
 222     jsonObjectFree(sessionCache);
 223     return stat;
 224 }
 225
 226 /**
 227  * Parses the request body and logs any REQUEST messages to the activity log
 228  */
 229 static int osrfHttpTranslatorParseRequest(osrfHttpTranslator* trans) {
 230     osrfMessage* msg;
 231     osrfMessage* msgList[MAX_MSGS_PER_PACKET];
 232     int numMsgs = osrf_message_deserialize(trans->body, msgList, MAX_MSGS_PER_PACKET);
 233     osrfLogDebug(OSRF_LOG_MARK, "parsed %d opensrf messages in this packet", numMsgs);
 234
 235     if(numMsgs == 0)
 236         return 0;
 237
 238     if(numMsgs == 1) {
 239         msg = msgList[0];
 240         if(msg->m_type == CONNECT) {
 241             trans->connectOnly = 1;
 242             trans->connecting = 1;
 243             return 1;
 244         }
 245         if(msg->m_type == DISCONNECT) {
 246             trans->disconnectOnly = 1;
 247             trans->disconnecting = 1;
 248             return 1;
 249         }
 250     }
 251
 252     // log request messages to the activity log
 253     int i;
 254     for(i = 0; i < numMsgs; i++) {
 255         msg = msgList[i];
 256
 257         switch(msg->m_type) {
 258
 259             case REQUEST: {
 260                 jsonObject* params = msg->_params;
 261                 growing_buffer* act = buffer_init(128);
 262                 buffer_fadd(act, "[%s] [%s] %s %s", trans->remoteHost, "", trans->service, msg->method_name);
 263
 264                 jsonObject* obj = NULL;
 265                 int i = 0;
 266                 char* str;
 267                 while((obj = jsonObjectGetIndex(params, i++))) {
 268                     str = jsonObjectToJSON(obj);
 269                     if( i == 1 )
 270                         OSRF_BUFFER_ADD(act, " ");
 271                     else
 272                         OSRF_BUFFER_ADD(act, ", ");
 273                     OSRF_BUFFER_ADD(act, str);
 274                     free(str);
 275                 }
 276                 osrfLogActivity(OSRF_LOG_MARK, "%s", act->buf);
 277                 buffer_free(act);
 278                 break;
 279             }
 280
 281             case CONNECT:
 282                 trans->connecting = 1;
 283                 break;
 284
 285             case DISCONNECT:
 286                 trans->disconnecting = 1;
 287                 break;
 288         }
 289     }
 290
 291     return 1;
 292 }
 293
 294 static int osrfHttpTranslatorCheckStatus(osrfHttpTranslator* trans, transport_message* msg) {
 295     osrfMessage* omsgList[MAX_MSGS_PER_PACKET];
 296     int numMsgs = osrf_message_deserialize(msg->body, omsgList, MAX_MSGS_PER_PACKET);
 297     osrfLogDebug(OSRF_LOG_MARK, "parsed %d response messages", numMsgs);
 298     if(numMsgs == 0) return 0;
 299
 300     osrfMessage* last = omsgList[numMsgs-1];
 301     if(last->m_type == STATUS) {
 302         if(last->status_code == OSRF_STATUS_TIMEOUT) {
 303             osrfLogDebug(OSRF_LOG_MARK, "removing cached session on request timeout");
 304             osrfCacheRemove(trans->thread);
 305             return 0;
 306         }
 307         // XXX hm, check for explicit status=COMPLETE message instead??
 308         if(last->status_code != OSRF_STATUS_CONTINUE)
 309             trans->complete = 1;
 310     }
 311
 312     return 1;
 313 }
 314
 315 static void osrfHttpTranslatorInitHeaders(osrfHttpTranslator* trans, transport_message* msg) {
 316     apr_table_set(trans->apreq->headers_out, OSRF_HTTP_HEADER_FROM, msg->sender);
 317     apr_table_set(trans->apreq->headers_out, OSRF_HTTP_HEADER_THREAD, trans->thread);
 318     if(trans->multipart) {
 319         sprintf(contentTypeBuf, MULTIPART_CONTENT_TYPE, trans->delim);
 320         contentTypeBuf[79] = '\0';
 321         osrfLogDebug(OSRF_LOG_MARK, "content type %s : %s : %s", MULTIPART_CONTENT_TYPE, trans->delim, contentTypeBuf);
 322             ap_set_content_type(trans->apreq, contentTypeBuf);
 323         ap_rprintf(trans->apreq, "--%s\n", trans->delim);
 324     } else {
 325             ap_set_content_type(trans->apreq, JSON_CONTENT_TYPE);
 326     }
 327 }
 328
 329 /**
 330  * Cache the transaction with the JID of the backend process we are talking to
 331  */
 332 static void osrfHttpTranslatorCacheSession(osrfHttpTranslator* trans, const char* jid) {
 333     jsonObject* cacheObj = jsonNewObject(NULL);
 334     jsonObjectSetKey(cacheObj, "ip", jsonNewObject(trans->remoteHost));
 335     jsonObjectSetKey(cacheObj, "jid", jsonNewObject(jid));
 336     jsonObjectSetKey(cacheObj, "service", jsonNewObject(trans->service));
 337     osrfCachePutObject((char*) trans->thread, cacheObj, CACHE_TIME);
 338 }
 339
 340
 341 /**
 342  * Writes a single chunk of multipart/x-mixed-replace content
 343  */
 344 static void osrfHttpTranslatorWriteChunk(osrfHttpTranslator* trans, transport_message* msg) {
 345     osrfLogInternal(OSRF_LOG_MARK, "sending multipart chunk %s", msg->body);
 346     ap_rprintf(trans->apreq,
 347         "Content-type: %s\n\n%s\n\n", JSON_CONTENT_TYPE, msg->body);
 348     //osrfLogInternal(OSRF_LOG_MARK, "Apache sending data: Content-type: %s\n\n%s\n\n", JSON_CONTENT_TYPE, msg->body);
 349     if(trans->complete) {
 350         ap_rprintf(trans->apreq, "--%s--\n", trans->delim);
 351         //osrfLogInternal(OSRF_LOG_MARK, "Apache sending data: --%s--\n", trans->delim);
 352     } else {
 353         ap_rprintf(trans->apreq, "--%s\n", trans->delim);
 354         //osrfLogInternal(OSRF_LOG_MARK, "Apache sending data: --%s\n", trans->delim);
 355     }
 356     ap_rflush(trans->apreq);
 357 }
 358
 359 static int osrfHttpTranslatorProcess(osrfHttpTranslator* trans) {
 360     if(trans->body == NULL)
 361         return HTTP_BAD_REQUEST;
 362
 363     if(!osrfHttpTranslatorSetTo(trans))
 364         return HTTP_BAD_REQUEST;
 365
 366     if(!osrfHttpTranslatorParseRequest(trans))
 367         return HTTP_BAD_REQUEST;
 368
 369     while(client_recv(trans->handle, 0))
 370         continue; // discard any old status messages in the recv queue
 371
 372     // send the message to the recipient
 373     transport_message* tmsg = message_init(
 374         trans->body, NULL, trans->thread, trans->recipient, NULL);
 375     message_set_osrf_xid(tmsg, osrfLogGetXid());
 376     client_send_message(trans->handle, tmsg);
 377     message_free(tmsg);
 378
 379     if(trans->disconnectOnly) {
 380         osrfLogDebug(OSRF_LOG_MARK, "exiting early on disconnect");
 381         osrfCacheRemove(trans->thread);
 382         return OK;
 383     }
 384
 385     // process the response from the opensrf service
 386     int firstWrite = 1;
 387     while(!trans->complete) {
 388         transport_message* msg = client_recv(trans->handle, trans->timeout);
 389
 390         if(trans->handle->error) {
 391             osrfLogError(OSRF_LOG_MARK, "Transport error");
 392             osrfCacheRemove(trans->thread);
 393             return HTTP_INTERNAL_SERVER_ERROR;
 394         }
 395
 396         if(msg == NULL)
 397             return HTTP_GATEWAY_TIME_OUT;
 398
 399         if(msg->is_error) {
 400             osrfLogError(OSRF_LOG_MARK, "XMPP message resulted in error code %d", msg->error_code);
 401             osrfCacheRemove(trans->thread);
 402             return HTTP_NOT_FOUND;
 403         }
 404
 405         if(!osrfHttpTranslatorCheckStatus(trans, msg))
 406             continue;
 407
 408         if(firstWrite) {
 409             osrfHttpTranslatorInitHeaders(trans, msg);
 410             if(trans->connecting)
 411                 osrfHttpTranslatorCacheSession(trans, msg->sender);
 412             firstWrite = 0;
 413         }
 414
 415         if(trans->multipart) {
 416             osrfHttpTranslatorWriteChunk(trans, msg);
 417             if(trans->connectOnly)
 418                 break;
 419         } else {
 420             if(!trans->messages)
 421                 trans->messages = osrfNewList();
 422             osrfListPush(trans->messages, msg->body);
 423
 424             if(trans->complete || trans->connectOnly) {
 425                 growing_buffer* buf = buffer_init(128);
 426                 int i;
 427                 OSRF_BUFFER_ADD(buf, osrfListGetIndex(trans->messages, 0));
 428                 for(i = 1; i < trans->messages->size; i++) {
 429                     buffer_chomp(buf); // chomp off the closing array bracket
 430                     char* body = osrfListGetIndex(trans->messages, i);
 431                     char newbuf[strlen(body)];
 432                     sprintf(newbuf, body+1); // chomp off the opening array bracket
 433                     OSRF_BUFFER_ADD_CHAR(buf, ',');
 434                     OSRF_BUFFER_ADD(buf, newbuf);
 435                 }
 436
 437                 ap_rputs(buf->buf, trans->apreq);
 438                 buffer_free(buf);
 439             }
 440         }
 441     }
 442
 443     if(trans->disconnecting) // DISCONNECT within a multi-message batch
 444         osrfCacheRemove(trans->thread);
 445
 446     return OK;
 447 }
 448
 449 static void testConnection(request_rec* r) {
 450         if(!osrfConnected || !osrfSystemGetTransportClient()) {
 451         osrfLogError(OSRF_LOG_MARK, "We're not connected to OpenSRF");
 452                 ap_log_rerror(APLOG_MARK, APLOG_ERR, 0, r, "We're not connected to OpenSRF");
 453                 usleep(100000); // .1 second to prevent process die/start overload
 454                 exit(1);
 455         }
 456 }
 457
 458 // it's dead, Jim
 459 static apr_status_t childExit(void* data) {
 460     osrf_system_shutdown();
 461     return OK;
 462 }
 463
 464 static void childInit(apr_pool_t *p, server_rec *s) {
 465         if(!osrfSystemBootstrapClientResc(configFile, configCtx, "translator")) {
 466                 ap_log_error( APLOG_MARK, APLOG_ERR, 0, s,
 467                         "Unable to Bootstrap OpenSRF Client with config %s..", configFile);
 468                 return;
 469         }
 470
 471     routerName = osrfConfigGetValue(NULL, "/router_name");
 472     domainName = osrfConfigGetValue(NULL, "/domain");
 473     const char* servers[] = {cacheServers};
 474     osrfCacheInit(servers, 1, 86400);
 475         osrfConnected = 1;
 476
 477     // at pool destroy time (= child exit time), cleanup
 478     // XXX causes us to disconnect even for clone()'d process cleanup (as in mod_cgi)
 479     //apr_pool_cleanup_register(p, NULL, childExit, apr_pool_cleanup_null);
 480 }
 481
 482 static int handler(request_rec *r) {
 483     int stat = OK;
 484         if(strcmp(r->handler, MODULE_NAME)) return DECLINED;
 485     if(r->header_only) return stat;
 486
 487         r->allowed |= (AP_METHOD_BIT << M_GET);
 488         r->allowed |= (AP_METHOD_BIT << M_POST);
 489
 490         osrfLogSetAppname("osrf_http_translator");
 491     testConnection(r);
 492
 493         osrfLogMkXid();
 494     osrfHttpTranslator* trans = osrfNewHttpTranslator(r);
 495     if(trans->body) {
 496         stat = osrfHttpTranslatorProcess(trans);
 497         //osrfHttpTranslatorDebug(trans);
 498         osrfLogInfo(OSRF_LOG_MARK, "translator resulted in status %d", stat);
 499     } else {
 500         osrfLogWarning(OSRF_LOG_MARK, "no message body to process");
 501     }
 502     osrfHttpTranslatorFree(trans);
 503         return stat;
 504 }
 505
 506
 507 static void registerHooks (apr_pool_t *p) {
 508         ap_hook_handler(handler, NULL, NULL, APR_HOOK_MIDDLE);
 509         ap_hook_child_init(childInit, NULL, NULL, APR_HOOK_MIDDLE);
 510 }
 511
 512
 513 module AP_MODULE_DECLARE_DATA osrf_http_translator_module = {
 514         STANDARD20_MODULE_STUFF,
 515     NULL,
 516         NULL,
 517     NULL,
 518         NULL,
 519     osrfHttpTranslatorCmds,
 520         registerHooks,
 521 };
 522
 523
 524
 525