fwd: Improve `source' and `target' lifecycle management.

[fwd] / socket.c

/* -*-c-*-
 *
 * Socket source and target definitions
 *
 * (c) 1999 Straylight/Edgeware
 */

/*----- Licensing notice --------------------------------------------------*
 *
 * This file is part of the `fwd' port forwarder.
 *
 * `fwd' is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * `fwd' is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with `fwd'; if not, write to the Free Software Foundation,
 * Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 */

#include "fwd.h"

/*----- Data structures ---------------------------------------------------*/

/* --- Socket source options --- */

typedef struct ssource_opts {
  unsigned opt;
  unsigned conn;
  unsigned listen;
  unsigned naccept;
} ssource_opts;

static ssource_opts ssgo = { 256, 0, 5, 1 };

#define SOCKOPT_LIMIT 0u
#define SOCKOPT_NOLIMIT 1u
#define SOCKOPT_ONESHOT 2u

/* --- Socket source --- */

typedef struct ssource {
  source s;
  addr *a;
  target *t;
  addr_opts *ao;
  ssource_opts o;
  sel_file r;
} ssource;

/* --- Socket target --- */

typedef struct starget {
  target t;
  addr *a;
  addr_opts *ao;
} starget;

/* --- Socket target endpoint --- */

typedef struct stept {
  endpt e;
  conn c;
  char *desc;
} stept;

/* --- Socket source endpoint --- */

typedef struct ssept {
  endpt e;
  ssource *s;
} ssept;

/*----- Protocol table ----------------------------------------------------*/

static addr_ops *addrs[] = { &inet_ops, &un_ops, 0 };

/*----- Other persistent variables ----------------------------------------*/

static addr_opts gsao = { 0 }, gtao = { 0 };

/*----- Parsing address types ---------------------------------------------*/

/* --- @getaddrtype@ --- *
 *
 * Arguments:	@scanner *sc@ = pointer to scanner (for error reporting)
 *		@const char *p@ = pointer to protocol name
 *		@int abbrev@ = nonzero to allow abbreviations
 *
 * Returns:	Pointer to address operations table or null.
 *
 * Use:		Looks up a protocol name.  Handy when parsing addresses and
 *		other bits of configuration.  Returns null if no matching
 *		address was found.
 */

static addr_ops *getaddrtype(scanner *sc, const char *p, int abbrev)
{
  addr_ops **ops;
  addr_ops *chosen = 0;
  size_t sz = strlen(p);

  for (ops = addrs; *ops; ops++) {
    if (strncmp((*ops)->name, p, sz) == 0) {
      if ((*ops)->name[sz] == 0)
	return (*ops);
      else if (chosen && abbrev)
	error(sc, "ambiguous socket address type `%s'", p);
      chosen = *ops;
    }
  }
  if (!abbrev)
    return (0);
  return (chosen);
}

/* --- @getaddr@ --- *
 *
 * Arguments:	@scanner *sc@ = pointer to scanner to read from
 *		@unsigned type@ = address type (@ADDR_SRC@ or @ADDR_DEST@)
 *
 * Returns:	Pointer to an address successfully read.
 *
 * Use:		Reads an optionally qualified address.
 */

static addr *getaddr(scanner *sc, unsigned type)
{
  addr_ops *ops = 0;
  int abbrev = 0;

  if (sc->t == ':') {
    token(sc);
    abbrev = 1;
  }
  if (sc->t == CTOK_WORD)
    ops = getaddrtype(sc, sc->d.buf, abbrev);
  if (ops)
    token(sc);
  else if (abbrev)
    error(sc, "unknown socket address type `%s'", sc->d.buf);
  else
    ops = &inet_ops;
  if (sc->t == ':')
    token(sc);

  return (ops->read(sc, type));
}

/*----- Socket endpoints --------------------------------------------------*/

/* --- @wclose@ --- */

static void sept_wclose(endpt *e)
{
  shutdown(e->out->fd, 1);
}

/* --- @close@ (source) --- */

static void ss_listen(ssource */*ss*/);

static void ssept_close(endpt *e)
{
  ssept *ee = (ssept *)e;

  if ((ee->s->s.f&SF_ACTIVE) && ee->s->o.opt == SOCKOPT_LIMIT) {
    ee->s->o.conn++;
    if (ee->s->o.conn == 1)
      ss_listen(ee->s);
  }
  REFFD_DEC(ee->e.in);
  REFFD_DEC(ee->e.out);
  fw_dec();
  DESTROY(ee);
}

/* --- @close@ (target) --- */

static void stept_close(endpt *e)
{
  stept *ee = (stept *)e;

  if (ee->e.f & EPF_PENDING)
    conn_kill(&ee->c);
  else {
    REFFD_DEC(ee->e.in);
    REFFD_DEC(ee->e.out);
  }

  xfree(ee->desc);
  fw_dec();
  DESTROY(ee);
}

/* --- @starget_connected@ --- *
 *
 * Arguments:	@int fd@ = file descriptor now ready for use
 *		@void *p@ = pointer to an endpoint structure
 *
 * Returns:	---
 *
 * Use:		Handles successful connection of the target endpoint.
 */

void starget_connected(int fd, void *p)
{
  stept *e = p;

  if (fd == -1) {
    fw_log(-1, "[%s] connection failed: %s", e->desc, strerror(errno));
    endpt_kill(&e->e);
  } else {
    reffd *r = reffd_init(fd);
    int opt = 1;
    REFFD_INC(r);
    fdflags(fd, O_NONBLOCK, O_NONBLOCK, FD_CLOEXEC, FD_CLOEXEC);
    setsockopt(fd, SOL_SOCKET, SO_OOBINLINE, &opt, sizeof(opt));
    e->e.in = e->e.out = r;
    e->e.f &= ~EPF_PENDING;
    if (e->e.other)
      endpt_join(&e->e, e->e.other, 0);
  }
}

/* --- Socket endpoint definition --- */

static endpt_ops ssept_ops = {
  0, 0, sept_wclose, ssept_close
};

static endpt_ops stept_ops = {
  0, 0, sept_wclose, stept_close
};

/*----- Source definition -------------------------------------------------*/

/* --- @option@ --- */

static int ssource_option(source *s, scanner *sc)
{
  ssource *ss = (ssource *)s;
  ssource_opts *sso = ss ? &ss->o : &ssgo;

  CONF_BEGIN(sc, "socket", "socket source")

  /* --- Make sure the next token is a word --- */

  if (sc->t != CTOK_WORD)
    error(sc, "parse error, option keyword expected");

  /* --- Handle options at this level --- */

  if (strcmp(sc->d.buf, "conn") == 0) {
    token(sc);
    if (sc->t == '=')
      token(sc);
    if (sc->t != CTOK_WORD)
      error(sc, "parse error, expected `unlimited', `one-shot' or number");
    if (isdigit((unsigned char)sc->d.buf[0])) {
      sso->conn = atoi(sc->d.buf);
      if (sso->conn == 0)
	error(sc, "argument of `conn' must be positive");
      sso->opt = SOCKOPT_LIMIT;
      token(sc);
    } else {
      sso->conn = 0;
      sso->opt = 1 + (1 & conf_enum(sc,
				    "unlimited,one-shot,infinite",
				    ENUM_ABBREV, "`conn' option"));
    }
    CONF_ACCEPT;
  }

  if (strcmp(sc->d.buf, "listen") == 0) {
    token(sc);
    if (sc->t == '=')
      token(sc);
    if (sc->t != CTOK_WORD || !isdigit((unsigned char)sc->d.buf[0]))
      error(sc, "parse error, expected number");
    sso->listen = atoi(sc->d.buf);
    if (sso->listen == 0)
      error(sc, "argument of `listen' must be positive");
    token(sc);
    CONF_ACCEPT;
  }

  if (strcmp(sc->d.buf, "accept") == 0 ||
      strcmp(sc->d.buf, "accept-count") == 0) {
    token(sc);
    if (sc->t == '=')
      token(sc);
    if (sc->t != CTOK_WORD)
      error(sc, "parse error, expected `unlimited' or number");
    else if (isdigit((unsigned char)sc->d.buf[0])) {
      sso->naccept = atoi(sc->d.buf);
      if (sso->naccept == 0)
	error(sc, "argument of `accept-count' must be positive");
    } else {
      sso->naccept = 0;
      conf_enum(sc, "unlimited,infinite",
		ENUM_ABBREV, "`accept-count' option");
    }
    token(sc);
    CONF_ACCEPT;
  }

  if (strcmp(sc->d.buf, "logging") == 0 ||
      strcmp(sc->d.buf, "log") == 0) {
    addr_opts *ao = ss ? ss->ao : &gsao;
    token(sc);
    if (sc->t == '=')
      token(sc);
    if (conf_enum(sc, "no,yes", ENUM_ABBREV, "logging status"))
      ao->f &= ~ADDRF_NOLOG;
    else
      ao->f |= ADDRF_NOLOG;
    CONF_ACCEPT;
  }

  /* --- Pass the option around the various address types --- */

  if (ss) {
    if (ss->a->ops->option && ss->a->ops->option(sc, ss->ao, ADDR_SRC))
      CONF_ACCEPT;
  } else {
    addr_ops **a;
    for (a = addrs; *a; a++) {
      if ((*a)->option && (*a)->option(sc, 0, ADDR_GLOBAL))
	CONF_ACCEPT;
    }
  }

  /* --- Nobody understood the option --- */

  CONF_END;
}

/* --- @read@ --- */

static source *ssource_read(scanner *sc)
{
  ssource *ss;

  (void)(conf_prefix(sc, "socket") || conf_prefix(sc, "sk"));
  ss = CREATE(ssource);
  ss->s.ops = &ssource_ops;
  ss->s.ref = 1;
  ss->s.f = 0;
  ss->s.desc = 0;
  ss->t = 0;
  ss->a = getaddr(sc, ADDR_SRC);
  if (ss->a->ops->initsrcopts)
    ss->ao = ss->a->ops->initsrcopts();
  else
    ss->ao = CREATE(addr_opts);
  *ss->ao = gsao;
  ss->o = ssgo;
  return (&ss->s);
}

/* --- @ss_accept@ --- *
 *
 * Arguments:	@int fd@ = file descriptor to accept from
 *		@unsigned mode@ = what's ready with the descriptor
 *		@void *p@ = pointer to the source definition
 *
 * Returns:	---
 *
 * Use:		Accepts an incoming connection and attaches it to a target
 *		endpoint.
 */

static void ssource_destroy(source */*s*/);

static void ss_accept(int fd, unsigned mode, void *p)
{
  ssource *ss = p;
  ssept *e;
  endpt *ee;
  reffd *r;
  int acceptp = 1;
  unsigned i = 0;

  while (acceptp) {

    /* --- Make the file descriptor --- */

    {
      int opt = 1;
      if ((r = ss->a->ops->accept(fd, ss->ao, ss->s.desc)) == 0)
	return;
      setsockopt(r->fd, SOL_SOCKET, SO_OOBINLINE, &opt, sizeof(opt));
      fdflags(r->fd, O_NONBLOCK, O_NONBLOCK, FD_CLOEXEC, FD_CLOEXEC);
    }

    /* --- Make an endpoint --- */

    e = CREATE(ssept);
    e->e.ops = &ssept_ops;
    e->e.other = 0;
    e->e.f = EPF_FILE;
    e->e.t = 0;
    e->e.in = e->e.out = r;
    e->s = ss;
    REFFD_INC(r);

    /* --- Obtain the target endpoint and let rip --- */

    if ((ee = ss->t->ops->create(ss->t, ss->s.desc)) == 0) {
      REFFD_DEC(r);
      REFFD_DEC(r);
      DESTROY(e);
      return;
    }
    fw_inc();

    /* --- Note that we've done one --- */

    i++;
    if (i >= ss->o.naccept)
      acceptp = 0;

    /* --- Remove the listening socket if necessary --- */

    switch (ss->o.opt) {
      case SOCKOPT_LIMIT:
	ss->o.conn--;
	if (!ss->o.conn) {
	  if (!(ss->ao->f & ADDRF_NOLOG))
	    fw_log(-1, "[%s] maximum connections reached", ss->s.desc);
	  sel_rmfile(&ss->r);
	  close(ss->r.fd);
	  if (ss->a->ops->unbind)
	    ss->a->ops->unbind(ss->a);
	  acceptp = 0;
	}
	break;
      case SOCKOPT_NOLIMIT:
	break;
      case SOCKOPT_ONESHOT:
	sel_rmfile(&ss->r);
	close(ss->r.fd);
	if (ss->a->ops->unbind)
	  ss->a->ops->unbind(ss->a);
	source_dec(&ss->s);
	acceptp = 0;
	break;
    }

    /* --- Let everything else happen --- */

    endpt_join(&e->e, ee, ss->s.desc);
  }
}

/* --- @ss_listen@ --- *
 *
 * Arguments:	@ssource *ss@ = source to listen on
 *
 * Returns:	---
 *
 * Use:		Sets the socket to listen again, if it stopped for some
 *		reason.  This is a copy of the code in the @read@ function,
 *		because it has different (wildly different) error handling
 *		behaviour.
 */

static void ss_listen(ssource *ss)
{
  int fd;
  int opt = 1;

  if (!(ss->ao->f & ADDRF_NOLOG))
    fw_log(-1, "[%s] reattaching listener", ss->s.desc);

  /* --- Make the socket --- */

  if ((fd = ss->a->ops->bind(ss->a, ss->ao)) < 0) {
    fw_log(-1, "[%s] couldn't create socket: %s",
	   ss->s.desc, strerror(errno));
    goto fail_0;
  }

  /* --- Set it to listen for connections --- */

  setsockopt(fd, SOL_SOCKET, SO_OOBINLINE, &opt, sizeof(opt));
  fdflags(fd, O_NONBLOCK, O_NONBLOCK, FD_CLOEXEC, FD_CLOEXEC);
  if (listen(fd, ss->o.listen)) {
    fw_log(-1, "[%s] couldn't listen on socket: %s",
	   ss->s.desc, strerror(errno));
    goto fail_1;
  }

  /* --- Set the listener up again --- */

  ss->r.fd = fd;
  sel_addfile(&ss->r);
  return;

  /* --- Tidy up if it failed --- *
   *
   * I'll just remove the entire source.
   */

fail_1:
  close(fd);
fail_0:
  ss->o.conn = 0;
  source_dec(&ss->s);
}

/* --- @attach@ --- */

static void ssource_attach(source *s, scanner *sc, target *t)
{
  ssource *ss = (ssource *)s;
  int fd;
  int opt = 1;

  ss->t = t; target_inc(t);

  /* --- Initialize the description string --- */

  {
    dstr d = DSTR_INIT;
    dstr_puts(&d, "socket.");
    ss->a->ops->print(ss->a, ADDR_SRC, &d);
    dstr_puts(&d, " -> ");
    dstr_puts(&d, ss->t->desc);
    ss->s.desc = xstrdup(d.buf);
    dstr_destroy(&d);
  }

  /* --- Confirm the address --- */

  if (ss->a->ops->confirm)
    ss->a->ops->confirm(ss->a, ADDR_SRC, ss->ao);

  /* --- Initialize the socket for listening --- */

  if ((fd = ss->a->ops->bind(ss->a, ss->ao)) < 0)
    error(sc, "couldn't bind socket `%s': %s", ss->s.desc, strerror(errno));

  /* --- Set it to listen for connections --- */

  setsockopt(fd, SOL_SOCKET, SO_OOBINLINE, &opt, sizeof(opt));
  fdflags(fd, O_NONBLOCK, O_NONBLOCK, FD_CLOEXEC, FD_CLOEXEC);
  if (listen(fd, ss->o.listen)) {
    error(sc, "couldn't listen on socket `%s': %s",
	  ss->s.desc, strerror(errno));
  }

  /* --- We're ready to go now --- */

  sel_initfile(sel, &ss->r, fd, SEL_READ, ss_accept, ss);
  sel_addfile(&ss->r);
  source_add(&ss->s);
  fw_inc();
}

/* --- @shutdown@ --- */

static void ssource_shutdown(source *s)
{
  ssource *ss = (ssource *)s;

  if (ss->o.conn || ss->o.opt != SOCKOPT_LIMIT) {
    sel_rmfile(&ss->r);
    close(ss->r.fd);
    if (ss->a->ops->unbind)
      ss->a->ops->unbind(ss->a);
  }
  if (ss->a->ops->freesrcopts)
    ss->a->ops->freesrcopts(ss->ao);
  else
    DESTROY(ss->ao);
  ss->a->ops->destroy(ss->a);
  source_remove(&ss->s);
  target_dec(ss->t);
  fw_dec();
}

/* --- @destroy@ --- */

static void ssource_destroy(source *s)
{
  ssource *ss = (ssource *)s;

  xfree(ss->s.desc);
  DESTROY(ss);
  fw_dec();
}

/* --- Source definition block --- */

source_ops ssource_ops = {
  "socket",
  ssource_option, ssource_read, ssource_attach, ssource_shutdown, ssource_destroy
};

/*----- Target definition -------------------------------------------------*/

/* --- @options@ --- */

static int starget_option(target *t, scanner *sc)
{
  starget *st = (starget *)t;

  CONF_BEGIN(sc, "starget", "socket target")

  /* --- Pass the option around the various address types --- */

  if (st) {
    if (st->a->ops->option && st->a->ops->option(sc, st->ao, ADDR_DEST))
      CONF_ACCEPT;
  }
  /* We'd have done it already if it was global */

  /* --- Done --- */

  CONF_END;
}

/* --- @read@ --- */

static target *starget_read(scanner *sc)
{
  starget *st;
  dstr d = DSTR_INIT;

  (void)(conf_prefix(sc, "socket") || conf_prefix(sc, "sk"));
  st = CREATE(starget);
  st->t.ops = &starget_ops;
  st->t.ref = 1;
  st->a = getaddr(sc, ADDR_DEST);
  if (st->a->ops->inittargopts)
    st->ao = st->a->ops->inittargopts();
  else {
    st->ao = CREATE(addr_opts);
    *st->ao = gtao;
  }
  dstr_puts(&d, "socket.");
  st->a->ops->print(st->a, ADDR_DEST, &d);
  st->t.desc = xstrdup(d.buf);
  dstr_destroy(&d);
  return (&st->t);
}

/* --- @confirm@ --- */

static void starget_confirm(target *t)
{
  starget *st = (starget *)t;

  if (st->a->ops->confirm)
    st->a->ops->confirm(st->a, ADDR_DEST, st->ao);
}

/* --- @create@ --- */

static endpt *starget_create(target *t, const char *desc)
{
  starget *st = (starget *)t;
  stept *e = CREATE(stept);

  e->e.ops = &stept_ops;
  e->e.other = 0;
  e->e.f = EPF_FILE | EPF_PENDING;
  e->e.t = 0;
  e->desc = xstrdup(desc);
  if (st->a->ops->connect(st->a, st->ao, &e->c, &e->e)) {
    fw_log(-1, "[%s] couldn't connect: %s", e->desc, strerror(errno));
    DESTROY(e);
    return (0);
  }
  fw_inc();
  return (&e->e);
}

/* --- @destroy@ --- */

static void starget_destroy(target *t)
{
  starget *st = (starget *)t;
  if (st->a->ops->freetargopts)
    st->a->ops->freetargopts(st->ao);
  else
    DESTROY(st->ao);
  st->a->ops->destroy(st->a);
  xfree(st->t.desc);
  DESTROY(st);
}

/* --- Socket target definition block --- */

target_ops starget_ops = {
  "socket",
  starget_option, starget_read, starget_confirm,
  starget_create, starget_destroy
};

/*----- That's all, folks -------------------------------------------------*/