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