base_accept.c

/* --------------------------------------------------------------------------
 *
 * License
 *
 * The contents of this file are subject to the Jabber Open Source License
 * Version 1.0 (the "License").  You may not copy or use this file, in either
 * source code or executable form, except in compliance with the License.  You
 * may obtain a copy of the License at http://www.jabber.com/license/ or at
 * http://www.opensource.org/.  
 *
 * Software distributed under the License is distributed on an "AS IS" basis,
 * WITHOUT WARRANTY OF ANY KIND, either express or implied.  See the License
 * for the specific language governing rights and limitations under the
 * License.
 *
 * Copyrights
 * 
 * Portions created by or assigned to Jabber.com, Inc. are 
 * Copyright (c) 1999-2000 Jabber.com, Inc.  All Rights Reserved.  Contact
 * information for Jabber.com, Inc. is available at http://www.jabber.com/.
 *
 * Portions Copyright (c) 1998-1999 Jeremie Miller.
 * 
 * Acknowledgements
 * 
 * Special thanks to the Jabber Open Source Contributors for their
 * suggestions and support of Jabber.
 * 
 * --------------------------------------------------------------------------*/
 
#include "jabberd.h"

extern pool jabberd__runtime;

#define A_ERROR  -1
#define A_READY   1
#define A_CLOSING 2

typedef struct queue_struct
{
  int stamp;
  xmlnode x;
  struct queue_struct *next;
} *queue, _queue;

typedef struct accept_instance_st
{
    mio m;
    int state;
    char *id;
    pool p;
    instance i;
    char *ip;
    char *secret;
    int port;
    int timeout;
    int log;
    queue q_start,q_end;
    dpacket dplast;
    pthread_mutex_t sem;
} *accept_instance, _accept_instance;


/* main thread ! */
void base_accept_kill(void *arg)
{
  accept_instance ai = (accept_instance)arg;
  log_debug(ZONE,"base accept kill");

  if (ai->state == A_READY){
    xmlnode h;
    log_debug(ZONE,"send end");

    /* send request to end transmission to wpj and other transports */
    h = xmlnode_new_tag("endconnection");
    if (ai->m != NULL) {
      mio_write(ai->m, h, NULL, -1);
      ai->state = A_CLOSING;
    }
    else
      xmlnode_free(h);
  }
}

void base_accept_queue(accept_instance ai, xmlnode x)
{
    queue q;
    if(ai == NULL || x == NULL) return;

    pthread_mutex_lock(&(ai->sem));
    q = pmalloco(xmlnode_pool(x),sizeof(_queue));
    q->stamp = time(NULL);
    q->x = x;
    q->next = NULL;

    /* is there something */
    if (ai->q_end)
      ai->q_end->next = q;
    else
      ai->q_start = q;

    ai->q_end = q;
    pthread_mutex_unlock(&(ai->sem));
}

/* Write packets to a xmlio object */
result base_accept_deliver(instance i, dpacket p, void* arg)
{
    accept_instance ai = (accept_instance)arg;

    /* Insert the message into the write_queue if we don't have a MIO socket yet.. */
    if (ai->state == A_READY || ai->state == A_CLOSING )
    {
        if(ai->dplast == p) /* don't return packets that they sent us! circular reference! */
            deliver_fail(p,"Circular Refernce Detected");
        else
            mio_write(ai->m, p->x, NULL, 0);
        return r_DONE;
    }

    base_accept_queue(ai, p->x);
    return r_DONE;
}


/* Handle incoming packets from the xstream associated with an MIO object */
void base_accept_process_xml(mio m, int state, void* arg, xmlnode x)
{
    accept_instance ai = (accept_instance)arg;
    xmlnode cur;
    queue q, q2;
    char hashbuf[41];
    int socket_size = 32768;
    int s = sizeof(int);

    log_debug(ZONE, "process XML: m:%X state:%d, arg:%X, x:%X", m, state, arg, x);

    switch(state)
    {
        case MIO_NEW:
        setsockopt(m->fd,SOL_SOCKET,SO_SNDBUF,(char *)&socket_size,s);
        setsockopt(m->fd,SOL_SOCKET,SO_RCVBUF,(char *)&socket_size,s);
        socket_size = 1;
        setsockopt(m->fd,SOL_SOCKET,SO_KEEPALIVE,(char *)&socket_size,s);
        return;

        case MIO_XML_ROOT:
            /* Ensure request namespace is correct... */
            if (j_strcmp(xmlnode_get_attrib(x, "xmlns"), "jabber:component:accept") != 0)
            {
                /* Log that the connected component sent an invalid namespace */
                log_warn(ai->i->id, "Recv'd invalid namespace. Closing connection.");
                /* Notify component with stream:error */
                mio_write(m, NULL, SERROR_NAMESPACE, -1);
                /* Close the socket and cleanup */
                mio_close(m);
                break;
            }

            /* Send header w/ proper namespace, using instance i */
            cur = xstream_header("jabber:component:accept", NULL, ai->i->id);
          
            /* Save stream ID for auth'ing later */
            ai->id = pstrdup(ai->p, xmlnode_get_attrib(cur, "id"));

          /* override if other connections */
          ai->state = A_ERROR;

            mio_write(m, NULL, xstream_header_char(cur), -1);
            xmlnode_free(cur);
            break;

        case MIO_XML_NODE:
            /* If aio has been authenticated previously, go ahead and deliver the packet */
          /* deliver packets even if closing */
              if(ai->state == A_READY || ai->state == A_CLOSING)   {
                  /* Hide 1.0 style transports etherx:* attribs */            
                  xmlnode_hide_attrib(x, "etherx:to");
                  xmlnode_hide_attrib(x, "etherx:from");
                  ai->dplast = dpacket_new(x);
                  
                  /* FOR LOG */
                  if ((ai->log >= 0) && (ai->dplast)){
                    cur = NULL;
                    if(ai->dplast->type == p_ROUTE) {
                        cur  = xmlnode_get_firstchild(ai->dplast->x);
                    }
                    else 
                        if(ai->dplast->type == p_NORM) {
                          cur = ai->dplast->x;
                        }
                    
                    if (cur){
                        if(strncmp(xmlnode_get_name(cur),"message",7) == 0) {
                          xmlnode curx;
                          __wplog->tab[ai->log]->messages++;
                          
                          if ((curx = xmlnode_get_tag(cur,"x"))
                                &&(j_strncmp(xmlnode_get_attrib(curx,"type"),"req",3)== 0))
                              __wplog->tab[ai->log]->files++;
                        }
                        else if(strncmp(xmlnode_get_name(cur),"presence",8) == 0){
                          __wplog->tab[ai->log]->presences++;
                        }
                        else if(strncmp(xmlnode_get_name(cur),"iq",2) == 0){
                          __wplog->tab[ai->log]->queries++;
                        }
                        else if(strncmp(xmlnode_get_name(cur),"iq",2) == 0){
                          __wplog->tab[ai->log]->queries++;
                        }
                    }
                  }
                  /* END LOG */
                  
                  /* xdb packet sends through socket goes to mtq */
                  if ((ai->dplast) && (ai->dplast->type == p_XDB)) {
                    /* XXX mtq_send */
                    deliver(ai->dplast, ai->i);
                    ai->dplast = NULL;
                    return;
                  }
                  else {
                    deliver(ai->dplast, ai->i);
                    ai->dplast = NULL;
                    return;
                  }
              }
              /* only other packets are handshakes */
              if(j_strcmp(xmlnode_get_name(x), "handshake") != 0) {
                  mio_write(m, NULL, "<stream:error>Must send handshake first.</stream:error>", -1);
                  mio_close(m);
                  break;
              }
              
              /* Create and check a SHA hash of this instance's password & SID */
              shahash_r(spools(xmlnode_pool(x), ai->id, ai->secret, xmlnode_pool(x)), hashbuf);
              if(j_strcmp(hashbuf, xmlnode_get_data(x)) != 0) {
                  mio_write(m, NULL, "<stream:error>Invalid handshake</stream:error>", -1);
                  mio_close(m);
                  break;
              }
              
              /* Send a handshake confirmation with our start time  */
              cur = xmlnode_new_tag("handshake");
              xmlnode_put_attrib(cur,"time",__wplog->server_start);
              mio_write(m, cur, NULL, -1);
              
              /* check for existing conenction and kill it */
            if(ai->m != NULL)  {
                    log_warn(ai->i->id, "Socket override by another connection from %s",mio_ip(m));
                    mio_write(ai->m, NULL, "<stream:error>Socket override by another connection.</stream:error>", -1);
                    mio_close(ai->m);
            }
                  
            /* hook us up! */
            ai->m = m;
            ai->state = A_READY;
                  
            /* flush old queue */
                  pthread_mutex_lock(&(ai->sem));
            q = ai->q_start;
            while(q != NULL)  {
                    q2 = q->next;
                    mio_write(m, q->x, NULL, 0);
                    q = q2;
            }
            ai->q_start = NULL;
            ai->q_end = NULL;
                  pthread_mutex_unlock(&(ai->sem));
                  
            break;
                  
      case MIO_ERROR:
        /* make sure it's the important one */
          log_debug(ZONE,"m=%d  ai->m = %d",m,ai->m);
          
            if(m != ai->m)
              return;
            
            ai->state = A_ERROR;
            
            /* clean up any tirds */
            while((cur = mio_cleanup(m)) != NULL)
              deliver_fail(dpacket_new(cur), "External Server Error");
            
            return;
            
      case MIO_CLOSED:
        /* make sure it's the important one */
        if(m != ai->m)
                return;
        
        log_debug(ZONE,"closing accepted socket");
        ai->m = NULL;
        ai->state = A_ERROR;
        
        return;
      default:
            return;
    }
    xmlnode_free(x);
}

/* check the packet queue for stale packets */
result base_accept_beat(void *arg)
{
    accept_instance ai = (accept_instance)arg;
    queue last, cur, next;
    int now = time(NULL);
    
    pthread_mutex_lock(&(ai->sem));
    cur = ai->q_start;
    while(cur != NULL)
    {
        if( (now - cur->stamp) <= ai->timeout)
        {
            last = cur;
            cur = cur->next;
            continue;
        }

        /* timed out sukkah! */
        next = cur->next;

      /* if first */
        if (ai->q_start == cur)
      {
            ai->q_start = next;

          /* first and last */
          if (ai->q_end == cur)
            ai->q_end = next; /* next must be NULL in here */
      }
      else /* if last */
        if (ai->q_end == cur)
        {
            ai->q_end = last;
        }
        else /* if not first, not last */
        {
            last->next = next;
        }

        deliver_fail(dpacket_new(cur->x),"Internal Timeout");
        cur = next;
    }
    
    pthread_mutex_unlock(&(ai->sem));
    return r_DONE;
}

result base_accept_config(instance id, xmlnode x, void *arg)
{
    char *secret = xmlnode_get_tag_data(x, "secret");
    accept_instance inst;
    int port = j_atoi(xmlnode_get_tag_data(x, "port"),0);

    if(id == NULL)
    {
        log_debug(ZONE,"base_accept_config validating configuration...");
        if (port == 0 || (xmlnode_get_tag(x,"secret") == NULL))
        {
            xmlnode_put_attrib(x,"error","<accept> requires the following subtags: <port>, and <secret>");
            return r_ERR;
        }         
        return r_PASS;
    }

    log_debug(ZONE,"base_accept_config performing configuration %s\n",xmlnode2str(x));

    /* Setup the default sink for this instance */ 
    inst              = pmalloco(id->p, sizeof(_accept_instance));
    inst->p           = id->p;
    inst->i           = id;
    inst->secret      = secret;
    inst->ip          = xmlnode_get_tag_data(x,"ip");
    inst->port        = port;
    inst->log         = j_atoi(xmlnode_get_tag_data(x, "log"),-1);
    inst->timeout     = j_atoi(xmlnode_get_tag_data(x, "timeout"),30); /* 30 sec */
    pthread_mutex_init(&(inst->sem),NULL);

    /* Start a new listening thread and associate this <listen> tag with it */
    if(mio_listen(inst->port, inst->ip, base_accept_process_xml, (void*)inst, NULL, mio_handlers_new(NULL, NULL, MIO_XML_PARSER)) == NULL)
    {
        xmlnode_put_attrib(x,"error","<accept> unable to listen on the configured ip and port");
        return r_ERR;
    }

    /* Register a packet handler and cleanup heartbeat for this instance */
    register_phandler(id, o_DELIVER, base_accept_deliver, (void *)inst);

    /* timeout check */
    register_beat(inst->timeout, base_accept_beat, (void *)inst);
    
    /* base shutdown kill */
    register_shutdown(base_accept_kill, (void *)inst);

    return r_DONE;
}


void base_accept(void)
{
    log_debug(ZONE,"base_accept loading...\n");
    register_config("accept",base_accept_config,NULL);
}