tport.c

Internet Phone, Chat, Conferencing

   * Loop until we can bind all the transports requested by the protocol to
   * the same port. 
   */
  do {
    not_supported = 1;		/* Make sure we don't loop for ever */
    pri = NULL;
    
    for (i = 0; transports[i]; i++) {
      su_addrinfo_t *ai, *res, hints[1];

      int tried = 0;

      proto = transports[i];
      error = EPROTONOSUPPORT;

      if (strcasecmp(proto, tpn->tpn_proto) != 0 && 
          strcmp(tpn->tpn_proto, tpn_any) != 0)
        continue;

      /* Resolve protocol, skip unknown transport protocols. */
      if (getprotohints(hints, proto, AI_PASSIVE) < 0)
        continue;

      hints->ai_family = family;

      if (host == NULL)
	hints->ai_flags |= AI_NUMERICHOST;

      pri = NULL;

      if (host)
	error = su_getaddrinfo(host, port, hints, &res);
      else
	error = tport_get_local_addrinfo(mr, li, port, hints, &res);

      if (error || !res) {
	if (error == EAI_SOCKTYPE)
	  SU_DEBUG_7(("%s(%p): su_getaddrinfo(%s, %s) for %s: %s\n", 
		      __func__, mr, host ? host : "\"\"", port, proto,
		      su_gai_strerror(error)));
	else
	  SU_DEBUG_3(("%s(%p): su_getaddrinfo(%s, %s) for %s: %s\n", 
		      __func__, mr, host ? host : "\"\"", port, proto,
		      su_gai_strerror(error)));
	error = ENOENT;
        continue;
      }

      p = ntohs(((su_sockaddr_t *)res->ai_addr)->su_port);

      for (ai = res; ai; ai = ai->ai_next) {
        /* Skip non-internet (AF_LOCAL) addresses */
#if SU_HAVE_IN6
        if (ai->ai_family != AF_INET && ai->ai_family != AF_INET6)
          continue;
#else
        if (ai->ai_family != AF_INET)
          continue;
#endif

        SU_DEBUG_9(("%s(%p): calling tport_socket\n", __func__, mr));

	ai->ai_socktype = hints->ai_socktype;
	ai->ai_protocol = hints->ai_protocol;

	tried = 1;

        pri = tport_listen(mr, ai, canon, proto, p, tags);

        if (pri) {
	  pri->pri_primary->tp_ident = su_strdup(pri->pri_home, tpn->tpn_ident);
	  tport_init_compression(pri, tpn->tpn_comp, tags);

#if HAVE_TLS
	  if (strcasecmp(tpn->tpn_proto, "tls") == 0) {
	    pri->pri_primary->tp_tls = tport_init_tls(tags);
	    if (!pri->pri_primary->tp_tls)
	      goto error;
	  }
#endif
	  not_supported = 0;
          if (ephemeral_port)
            port = strcpy(port0, pri->pri_primary->tp_port);
          if (ephemeral_port == -1) {
            ephemeral_port = ntohs(pri->pri_primary->tp_addr->su_port);
	    assert(ephemeral_port != 0);
	    if (p == 0)
	      p = ephemeral_port;
	  }
        }
	else {
	  error = su_errno();
	  if (error == EADDRINUSE)
	    not_supported = 0;
	}
      }

      if (host)
	su_freeaddrinfo(res);
      else
	tport_freeaddrinfo(mr, res);

      if (!pri) {
        while (*tbf)
	  tport_zap_primary(*tbf);

	if (!tried)
	  not_supported = 1;

        if (ephemeral_port != 0 && ephemeral_port != -1) {
	  while (step == 0) {
	    /* step should be relative prime to 65536 - 1024 */
	    /* 65536 - 1024 = 7 * 3 * 3 * 1024 */
	    step = (random() | 1) % (65535 - 1024);
	    if (step % 3 == 0)
	      step = (step + 2) % (65536 - 1024);
	    if (step % 7 == 0)
	      step = (step + 2) % (65536 - 1024);
	  }
	  p += step;
          if (p >= 65536) p -= (65536 - 1024);
          if (p == ephemeral_port)
            ephemeral_port = 0;
          SU_DEBUG_3(("%s(%p): cannot bind all transports to port %s, "
		      "trying %u\n", 
		      __func__, mr, port, p));
          snprintf(port0, sizeof(port0), "%u", p);
          port = port0;
        }
        break;
      }
    }
  }
  while (!pri && ephemeral_port && !not_supported);

  if (li)
    su_freelocalinfo(li);

  if (not_supported)
    error = EPROTONOSUPPORT;

  if (!pri) {
    su_seterrno(error);
    return -1;
  }

  for (pri = *tbf; pri; pri = pri->pri_next) {
    tport_launch_threadpool(pri);
  }

  return 0;

 error:
  if (li)
    su_freelocalinfo(li);
  return -1;
}


/** Convert localinfo into addrinfo */
static
int
tport_get_local_addrinfo(tport_master_t *mr, 
			 su_localinfo_t *li,
			 char const *port,
			 su_addrinfo_t const *hints,
			 su_addrinfo_t **return_ai)
{
  su_addrinfo_t *ai, **prev;
  su_sockaddr_t *su;
  unsigned long lport = 0;
  char *rest;

  if (port) {
    lport = strtoul(port, &rest, 10);
    if (lport >= 65536) {
      su_seterrno(EINVAL);
      return EAI_NONAME;
    }
  }

  prev = return_ai;

  for (; li; li = li->li_next) {
    if (hints->ai_family && hints->ai_family != li->li_family)
      continue;

    ai = calloc(1, sizeof *ai + li->li_addrlen);
    if (ai == NULL) {
      tport_freeaddrinfo(mr, *return_ai);
      return EAI_MEMORY;
    }
    
    *prev = ai, prev = &ai->ai_next;

    ai->ai_flags = AI_PASSIVE | TP_AI_ANY;
    ai->ai_family = li->li_family;
    ai->ai_socktype = hints->ai_socktype;
    ai->ai_protocol = hints->ai_protocol;
    ai->ai_addr = memcpy(ai + 1, li->li_addr, ai->ai_addrlen = li->li_addrlen);
    su = (void *)ai->ai_addr;
    su->su_port = htons(lport);
  }

  return 0;
}

static
void tport_freeaddrinfo(tport_master_t *mr, 
			su_addrinfo_t *ai)
{
  su_addrinfo_t *ai_next;

  while (ai) {
    ai_next = ai->ai_next;
    free(ai);
    ai = ai_next;
  }
}


#if HAVE_TLS
tls_t *
tport_init_tls(tagi_t *tags)
{
  char *homedir = getenv("HOME");
  char *tbf = NULL;
  char const *path = NULL;
  tls_t *tls = NULL;
  su_home_t *home = su_home_clone(NULL, sizeof *home);
  unsigned version = 1;

  if (getenv("TPORT_SSL"))
    version = 0;

  tl_gets(tags,
	  TPTAG_CERTIFICATE_REF(path),
	  TPTAG_TLS_VERSION_REF(version),
	  TAG_END());

  if (!homedir)
    homedir = "";
	   
  if (!path)
    path = tbf = su_sprintf(home, "%s/.sip/auth", homedir);
  
  if (path) {
    tls_issues_t ti = {0};
    ti.verify_depth = 2;
    ti.configured = path != tbf;
    ti.randFile = su_sprintf(home, "%s/%s", path, "tls_seed.dat");
    ti.key = su_sprintf(home, "%s/%s", path, "agent.pem");
    ti.cert = ti.key;
    ti.CAfile = su_sprintf(home, "%s/%s", path, "cafile.pem");
    ti.version = version;

    SU_DEBUG_9(("%s(%p): tls key = %s\n", __func__, home, ti.key));

    if (ti.key && ti.CAfile && ti.randFile) {
      tls = tls_init_master(&ti);
      if (tls == NULL)
	SU_DEBUG_3(("tls_init_master: %s\n", strerror(errno)));
    }
  }

  su_home_zap(home); 

  return tls;
}
#endif

static 
int tport_init_compression(tport_primary_t *pri, 
			   char const *compression, 
			   tagi_t *tl)
{
#if HAVE_SIGCOMP
  tport_master_t *mr = pri->pri_master;

  if (compression == NULL || 
      strcasecmp(compression, "sigcomp"))
    return 0;

  if (pri->pri_protocol != IPPROTO_TCP &&
      pri->pri_protocol != IPPROTO_UDP) {
    SU_DEBUG_3(("tport: no sigcomp for %s\n", pri->pri_primary->tp_protoname));
    return 0;
  }

  if (mr->mr_compartment) {
    pri->pri_primary->tp_name->tpn_comp = tport_sigcomp_name;
  }
#endif

  return 0;
}

/** Close a transport. 
 * 
 * The function tport_close() closes a socket associated with a transport
 * object.
 */
void tport_close(tport_t *self)
{
  int i;

  SU_DEBUG_5(("%s(%p): " TPN_FORMAT "\n", "tport_close", self,
	      TPN_ARGS(self->tp_name)));

  self->tp_closed = 1;
  self->tp_send_close = 3;
  self->tp_recv_close = 3;

#if HAVE_TLS
  if (self->tp_tls != NULL) {
    /* if (tport_is_primary(self)) */
      tls_free(self->tp_tls);
      /* XXX - PPe: does the tls_shutdown zap everything but socket? */
    self->tp_tls = NULL;
    su_free(self->tp_home, self->tp_tls_buffer);
  }
  else 
#endif
  if (self->tp_socket != -1)
    shutdown(self->tp_socket, 2);

  if (self->tp_index)
    su_root_deregister(self->tp_master->mr_root, self->tp_index);
  self->tp_index = 0;
#if SU_HAVE_BSDSOCK
  if (self->tp_socket != -1)
    su_close(self->tp_socket);
  self->tp_socket = -1;
#endif

  if (self->tp_params->tpp_sdwn_error && self->tp_pused)
    tport_error_report(self, -1, NULL);

#if HAVE_SIGCOMP
  if (self->tp_sigcomp->sc_cc) {
    sigcomp_compartment_unref(self->tp_sigcomp->sc_cc);
    self->tp_sigcomp->sc_cc = NULL;
  }  
#endif

  /* Zap the queued messages */
  if (self->tp_queue) {
    int N = self->tp_params->tpp_qsize;
    for (i = 0; i < N; i++) {
      if (self->tp_queue[i])
	msg_ref_destroy(self->tp_queue[i]), self->tp_queue[i] = NULL;
    }
  }

  self->tp_index = 0;
  self->tp_events = 0;
}

/** Shutdown a transport.
 *
 * The tport_shutdown() shuts down a full-duplex transport connection
 * partially or completely. If @a how is 0, the further incoming data is
 * shut down. If @a how is 1, further outgoing data is shut down. If @a how
 * is 2, both incoming and outgoing traffic is shut down.
 *
 */
int tport_shutdown(tport_t *self, int how)
{
  if (self == NULL || tport_is_primary(self))
    return -1;

  SU_DEBUG_7(("%s(%p, %d)\n", "tport_shutdown", self, how));

  if (!tport_is_tcp(self) ||
      how < 0 || 
      (how == 0 && self->tp_send_close) ||
      (how == 1 && self->tp_recv_close > 1) || 
      how >= 2) {
    tport_close(self);
    return 1;
  }

#if HAVE_TLS
  if (self->tp_tls != NULL) {
    /* XXX - send alert */
    return 0;
  }
#endif

  shutdown(self->tp_socket, how);

  if (how == 0) {
    self->tp_recv_close = 2;
    self->tp_events &= ~SU_WAIT_IN;
    if (self->tp_params->tpp_sdwn_error && self->tp_pused)
      tport_error_report(self, -1, NULL);
  }
  else if (how == 1) {
    self->tp_send_close = 2;
    self->tp_events &= ~SU_WAIT_OUT;

    if (self->tp_queue && self->tp_queue[self->tp_qhead]) {
      int i, N = self->tp_params->tpp_qsize;
      for (i = 0; i < N; i++) {
	if (self->tp_queue[i]) {
	  tport_pending_errmsg(self, self->tp_queue[i], EPIPE);
	  msg_ref_destroy(self->tp_queue[i]), self->tp_queue[i] = NULL;
	}
      }
    }
  }

  tport_events(self);

  return 0;
}

static inline
unsigned long tport_now(void)
{
  return su_now().tv_sec;
}

/** Transport timer function. */
static
void tport_tick(su_root_magic_t *magic, su_timer_t *t, tport_master_t *mr)
{
  tport_primary_t *dad;
  tport_t *tp, *tp_next;
  su_time_t now = su_now();
  int ts = su_time_ms(now);

  /* Go through all primary transports */
  for (dad = mr->mr_primaries; dad; dad = dad->pri_next) {
    if (dad->pri_protocol == IPPROTO_SCTP) {
      /* Go through all SCTP connections */

      tp = dad->pri_secondary;

      for (tp = tprb_first(tp); tp; tp = tp_next) {
	tp_next = tprb_succ(tp);
	if (tp->tp_queue && tp->tp_queue[tp->tp_qhead]) {
	  SU_DEBUG_9(("tport_tick(%p) - trying to send to %s/%s:%s\n", 
		      tp, tp->tp_protoname, tp->tp_host, tp->tp_port));
	  tport_send_queue(tp);
	}
      }      
    }

    /* Go through all secondary transports with incomplete messages */
    for (tp = tprb_first(dad->pri_secondary); tp; tp = tp_next) {
      msg_t *msg = tp->tp_msg;
      int closed;

      if (msg &&
	  tp->tp_params->tpp_timeout != UINT_MAX && 
	  tp->tp_params->tpp_timeout < ts - (int)tp->tp_time &&
	  !msg_is_streaming(msg)) {
	SU_DEBUG_5(("tport_tick(%p): incomplete message idle for %d ms\n",
		    tp, ts - (int)tp->tp_time));
	msg_set_streaming(msg, 0);
	msg_set_flags(msg, MSG_FLG_ERROR | MSG_FLG_TRUNC | MSG_FLG_TIMEOUT);
	tport_deliver(tp, msg, NULL, NULL, now);
	tp->tp_msg = NULL;
      }

      tp_next = tprb_succ(tp);

      if (tp->tp_refs)
	continue;

      closed = tport_is_closed(tp);

      if (!closed &&
	  !(tp->tp_params->tpp_idle > 0 
	    && tp->tp_params->tpp_idle < ts - (int)tp->tp_time)) {
	continue;
      }

      if (closed) {
	SU_DEBUG_5(("tport_tick(%p): closed, zapping\n", tp));
      } else {
	SU_DEBUG_5(("tport_tick(%p): unused for %d ms, closing and zapping\n",
		    tp, ts - (int)tp->tp_time));
	if (!tport_is_closed(tp))
	  tport_close(tp);
      }

      tport_zap_secondary(tp);
    }
  }

  su_timer_set(t, tport_tick