/src/samba/source3/smbd/notifyd/notifyd.c

Source
/*
 * Unix SMB/CIFS implementation.
 *
 * Copyright (C) Volker Lendecke 2014
 *
 * This program 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 3 of the License, or
 * (at your option) any later version.
 *
 * This program 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 this program.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "replace.h"
#include <tevent.h>
#include "notifyd_private.h"
#include "lib/util/server_id.h"
#include "lib/util/data_blob.h"
#include "librpc/gen_ndr/notify.h"
#include "librpc/gen_ndr/messaging.h"
#include "librpc/gen_ndr/server_id.h"
#include "lib/dbwrap/dbwrap.h"
#include "lib/dbwrap/dbwrap_rbt.h"
#include "messages.h"
#include "tdb.h"
#include "util_tdb.h"
#include "notifyd.h"
#include "lib/util/server_id_db.h"
#include "lib/util/tevent_unix.h"
#include "lib/util/tevent_ntstatus.h"
#include "ctdbd_conn.h"
#include "ctdb_srvids.h"
#include "server_id_db_util.h"
#include "lib/util/iov_buf.h"
#include "messages_util.h"

#ifdef CLUSTER_SUPPORT
#include "ctdb_protocol.h"
#endif

struct notifyd_peer;

/*
 * All of notifyd's state
 */

struct notifyd_state {
  struct tevent_context *ev;
  struct messaging_context *msg_ctx;
  struct ctdbd_connection *ctdbd_conn;

  /*
   * Database of everything clients show interest in. Indexed by
   * absolute path. The database keys are not 0-terminated
   * to allow the critical operation, notifyd_trigger, to walk
   * the structure from the top without adding intermediate 0s.
   * The database records contain an array of
   *
   * struct notifyd_instance
   *
   * to be maintained and parsed by notifyd_parse_entry()
   */
  struct db_context *entries;

  /*
   * In the cluster case, this is the place where we store a log
   * of all MSG_SMB_NOTIFY_REC_CHANGE messages. We just 1:1
   * forward them to our peer notifyd's in the cluster once a
   * second or when the log grows too large.
   */

  struct messaging_reclog *log;

  /*
   * Array of companion notifyd's in a cluster. Every notifyd
   * broadcasts its messaging_reclog to every other notifyd in
   * the cluster. This is done by making ctdb send a message to
   * srvid CTDB_SRVID_SAMBA_NOTIFY_PROXY with destination node
   * number CTDB_BROADCAST_CONNECTED. Everybody in the cluster who
   * had called register_with_ctdbd this srvid will receive the
   * broadcasts.
   *
   * Database replication happens via these broadcasts. Also,
   * they serve as liveness indication. If a notifyd receives a
   * broadcast from an unknown peer, it will create one for this
   * srvid. Also when we don't hear anything from a peer for a
   * while, we will discard it.
   */

  struct notifyd_peer **peers;
  size_t num_peers;

  sys_notify_watch_fn sys_notify_watch;
  struct sys_notify_context *sys_notify_ctx;
};

struct notifyd_peer {
  struct notifyd_state *state;
  struct server_id pid;
  uint64_t rec_index;
  struct db_context *db;
  time_t last_broadcast;
};

static void notifyd_rec_change(struct messaging_context *msg_ctx,
             void *private_data, uint32_t msg_type,
             struct server_id src, DATA_BLOB *data);
static void notifyd_trigger(struct messaging_context *msg_ctx,
          void *private_data, uint32_t msg_type,
          struct server_id src, DATA_BLOB *data);
static void notifyd_get_db(struct messaging_context *msg_ctx,
         void *private_data, uint32_t msg_type,
         struct server_id src, DATA_BLOB *data);

#ifdef CLUSTER_SUPPORT
static void notifyd_got_db(struct messaging_context *msg_ctx,
         void *private_data, uint32_t msg_type,
         struct server_id src, DATA_BLOB *data);
static void notifyd_broadcast_reclog(struct ctdbd_connection *ctdbd_conn,
             struct server_id src,
             struct messaging_reclog *log);
#endif
static void notifyd_sys_callback(struct sys_notify_context *ctx,
         void *private_data, struct notify_event *ev,
         uint32_t filter);

#ifdef CLUSTER_SUPPORT
static struct tevent_req *notifyd_broadcast_reclog_send(
  TALLOC_CTX *mem_ctx, struct tevent_context *ev,
  struct ctdbd_connection *ctdbd_conn, struct server_id src,
  struct messaging_reclog *log);
static int notifyd_broadcast_reclog_recv(struct tevent_req *req);

static struct tevent_req *notifyd_clean_peers_send(
  TALLOC_CTX *mem_ctx, struct tevent_context *ev,
  struct notifyd_state *notifyd);
static int notifyd_clean_peers_recv(struct tevent_req *req);
#endif

static int sys_notify_watch_dummy(
  TALLOC_CTX *mem_ctx,
  struct sys_notify_context *ctx,
  const char *path,
  uint32_t *filter,
  uint32_t *subdir_filter,
  void (*callback)(struct sys_notify_context *ctx,
       void *private_data,
       struct notify_event *ev,
       uint32_t filter),
  void *private_data,
  void *handle_p)
{
  void **handle = handle_p;
  *handle = NULL;
  return 0;
}

#ifdef CLUSTER_SUPPORT
static void notifyd_broadcast_reclog_finished(struct tevent_req *subreq);
static void notifyd_clean_peers_finished(struct tevent_req *subreq);
static int notifyd_snoop_broadcast(struct tevent_context *ev,
           uint32_t src_vnn, uint32_t dst_vnn,
           uint64_t dst_srvid,
           const uint8_t *msg, size_t msglen,
           void *private_data);
#endif

struct tevent_req *notifyd_send(TALLOC_CTX *mem_ctx, struct tevent_context *ev,
        struct messaging_context *msg_ctx,
        struct ctdbd_connection *ctdbd_conn,
        sys_notify_watch_fn sys_notify_watch,
        struct sys_notify_context *sys_notify_ctx)
{
  struct tevent_req *req;
#ifdef CLUSTER_SUPPORT
  struct tevent_req *subreq;
#endif
  struct notifyd_state *state;
  struct server_id_db *names_db;
  NTSTATUS status;
  int ret;

  req = tevent_req_create(mem_ctx, &state, struct notifyd_state);
  if (req == NULL) {
    return NULL;
  }
  state->ev = ev;
  state->msg_ctx = msg_ctx;
  state->ctdbd_conn = ctdbd_conn;

  if (sys_notify_watch == NULL) {
    sys_notify_watch = sys_notify_watch_dummy;
  }

  state->sys_notify_watch = sys_notify_watch;
  state->sys_notify_ctx = sys_notify_ctx;

  state->entries = db_open_rbt(state);
  if (tevent_req_nomem(state->entries, req)) {
    return tevent_req_post(req, ev);
  }

  status = messaging_register(msg_ctx, state, MSG_SMB_NOTIFY_REC_CHANGE,
            notifyd_rec_change);
  if (tevent_req_nterror(req, status)) {
    return tevent_req_post(req, ev);
  }

  status = messaging_register(msg_ctx, state, MSG_SMB_NOTIFY_TRIGGER,
            notifyd_trigger);
  if (tevent_req_nterror(req, status)) {
    goto deregister_rec_change;
  }

  status = messaging_register(msg_ctx, state, MSG_SMB_NOTIFY_GET_DB,
            notifyd_get_db);
  if (tevent_req_nterror(req, status)) {
    goto deregister_trigger;
  }

  names_db = messaging_names_db(msg_ctx);

  ret = server_id_db_set_exclusive(names_db, "notify-daemon");
  if (ret != 0) {
    DBG_DEBUG("server_id_db_set_exclusive() failed: %s\n",
        strerror(ret));
    tevent_req_error(req, ret);
    goto deregister_get_db;
  }

  if (ctdbd_conn == NULL) {
    /*
     * No cluster around, skip the database replication
     * engine
     */
    return req;
  }

#ifdef CLUSTER_SUPPORT
  status = messaging_register(msg_ctx, state, MSG_SMB_NOTIFY_DB,
            notifyd_got_db);
  if (tevent_req_nterror(req, status)) {
    goto deregister_get_db;
  }

  state->log = talloc_zero(state, struct messaging_reclog);
  if (tevent_req_nomem(state->log, req)) {
    goto deregister_db;
  }

  subreq = notifyd_broadcast_reclog_send(
    state->log, ev, ctdbd_conn,
    messaging_server_id(msg_ctx),
    state->log);
  if (tevent_req_nomem(subreq, req)) {
    goto deregister_db;
  }
  tevent_req_set_callback(subreq,
        notifyd_broadcast_reclog_finished,
        req);

  subreq = notifyd_clean_peers_send(state, ev, state);
  if (tevent_req_nomem(subreq, req)) {
    goto deregister_db;
  }
  tevent_req_set_callback(subreq, notifyd_clean_peers_finished,
        req);

  ret = register_with_ctdbd(ctdbd_conn,
          CTDB_SRVID_SAMBA_NOTIFY_PROXY,
          notifyd_snoop_broadcast, state);
  if (ret != 0) {
    tevent_req_error(req, ret);
    goto deregister_db;
  }
#endif

  return req;

#ifdef CLUSTER_SUPPORT
deregister_db:
  messaging_deregister(msg_ctx, MSG_SMB_NOTIFY_DB, state);
#endif
deregister_get_db:
  messaging_deregister(msg_ctx, MSG_SMB_NOTIFY_GET_DB, state);
deregister_trigger:
  messaging_deregister(msg_ctx, MSG_SMB_NOTIFY_TRIGGER, state);
deregister_rec_change:
  messaging_deregister(msg_ctx, MSG_SMB_NOTIFY_REC_CHANGE, state);
  return tevent_req_post(req, ev);
}

#ifdef CLUSTER_SUPPORT

static void notifyd_broadcast_reclog_finished(struct tevent_req *subreq)
{
  struct tevent_req *req = tevent_req_callback_data(
    subreq, struct tevent_req);
  int ret;

  ret = notifyd_broadcast_reclog_recv(subreq);
  TALLOC_FREE(subreq);
  tevent_req_error(req, ret);
}

static void notifyd_clean_peers_finished(struct tevent_req *subreq)
{
  struct tevent_req *req = tevent_req_callback_data(
    subreq, struct tevent_req);
  int ret;

  ret = notifyd_clean_peers_recv(subreq);
  TALLOC_FREE(subreq);
  tevent_req_error(req, ret);
}

#endif

int notifyd_recv(struct tevent_req *req)
{
  return tevent_req_simple_recv_unix(req);
}

static bool notifyd_apply_rec_change(
  const struct server_id *client,
  const char *path, size_t pathlen,
  const struct notify_instance *chg,
  struct db_context *entries,
  sys_notify_watch_fn sys_notify_watch,
  struct sys_notify_context *sys_notify_ctx,
  struct messaging_context *msg_ctx)
{
  struct db_record *rec = NULL;
  struct notifyd_watcher watcher = {};
  struct notifyd_instance *instances = NULL;
  size_t num_instances;
  size_t i;
  struct notifyd_instance *instance = NULL;
  TDB_DATA value;
  NTSTATUS status;
  bool ok = false;
  bool new_watcher = false;

  if (pathlen == 0) {
    DBG_WARNING("pathlen==0\n");
    return false;
  }
  if (path[pathlen-1] != '\0') {
    DBG_WARNING("path not 0-terminated\n");
    return false;
  }

  DBG_DEBUG("path=%s, filter=%"PRIu32", subdir_filter=%"PRIu32", "
      "private_data=%p\n",
      path,
      chg->filter,
      chg->subdir_filter,
      chg->private_data);

  rec = dbwrap_fetch_locked(
    entries, entries,
    make_tdb_data((const uint8_t *)path, pathlen-1));

  if (rec == NULL) {
    DBG_WARNING("dbwrap_fetch_locked failed\n");
    goto fail;
  }

  num_instances = 0;
  value = dbwrap_record_get_value(rec);

  if (value.dsize != 0) {
    ok = notifyd_parse_entry(value.dptr,
           value.dsize,
           &watcher,
           NULL,
           &num_instances);
    if (!ok) {
      goto fail;
    }
  }

  /*
   * Overallocate by one instance to avoid a realloc when adding
   */
  instances = talloc_array(rec, struct notifyd_instance,
         num_instances + 1);
  if (instances == NULL) {
    DBG_WARNING("talloc failed\n");
    goto fail;
  }

  if (num_instances > 0) {
    struct notifyd_instance *tmp = NULL;
    size_t num_tmp = 0;

    ok = notifyd_parse_entry(value.dptr,
           value.dsize,
           NULL,
           &tmp,
           &num_tmp);
    if (!ok) {
      goto fail;
    }

    memcpy(instances,
           tmp,
           sizeof(struct notifyd_instance) * num_tmp);
  }

  for (i=0; i<num_instances; i++) {
    instance = &instances[i];

    if (server_id_equal(&instance->client, client) &&
        (instance->instance.private_data == chg->private_data)) {
      break;
    }
  }

  if (i < num_instances) {
    instance->instance = *chg;
  } else {
    /*
     * We've overallocated for one instance
     */
    instance = &instances[num_instances];

    *instance = (struct notifyd_instance) {
      .client = *client,
      .instance = *chg,
    };

    num_instances += 1;
  }

  /*
   * Calculate an intersection of the instances filters for the watcher.
   */
  if (instance->instance.filter > 0) {
    uint32_t filter = instance->instance.filter;

    if ((watcher.filter & filter) != filter) {
      watcher.filter |= filter;

      new_watcher = true;
    }
  }

  /*
   * Calculate an intersection of the instances subdir_filters for the
   * watcher.
   */
  if (instance->instance.subdir_filter > 0) {
    uint32_t subdir_filter = instance->instance.subdir_filter;

    if ((watcher.subdir_filter & subdir_filter) != subdir_filter) {
      watcher.subdir_filter |= subdir_filter;

      new_watcher = true;
    }
  }

  if ((instance->instance.filter == 0) &&
      (instance->instance.subdir_filter == 0)) {
    uint32_t tmp_filter = 0;
    uint32_t tmp_subdir_filter = 0;

    /* This is a delete request */
    *instance = instances[num_instances-1];
    num_instances -= 1;

    for (i = 0; i < num_instances; i++) {
      struct notifyd_instance *tmp = &instances[i];

      tmp_filter |= tmp->instance.filter;
      tmp_subdir_filter |= tmp->instance.subdir_filter;
    }

    /*
     * If the filter has changed, register a new watcher with the
     * changed filter.
     */
    if (watcher.filter != tmp_filter ||
        watcher.subdir_filter != tmp_subdir_filter)
    {
      watcher.filter = tmp_filter;
      watcher.subdir_filter = tmp_subdir_filter;

      new_watcher = true;
    }
  }

  if (new_watcher) {
    /*
     * In case we removed all notify instances, we want to remove
     * the watcher. We won't register a new one, if no filters are
     * set anymore.
     */

    TALLOC_FREE(watcher.sys_watch);

    watcher.sys_filter = watcher.filter;
    watcher.sys_subdir_filter = watcher.subdir_filter;

    /*
     * Only register a watcher if we have filter.
     */
    if (watcher.filter != 0 || watcher.subdir_filter != 0) {
      int ret = sys_notify_watch(entries,
               sys_notify_ctx,
               path,
               &watcher.sys_filter,
               &watcher.sys_subdir_filter,
               notifyd_sys_callback,
               msg_ctx,
               &watcher.sys_watch);
      if (ret != 0) {
        DBG_WARNING("sys_notify_watch for [%s] "
              "returned %s\n",
              path,
              strerror(errno));
      }
    }
  }

  DBG_DEBUG("%s has %zu instances\n", path, num_instances);

  if (num_instances == 0) {
    TALLOC_FREE(watcher.sys_watch);

    status = dbwrap_record_delete(rec);
    if (!NT_STATUS_IS_OK(status)) {
      DBG_WARNING("dbwrap_record_delete returned %s\n",
            nt_errstr(status));
      goto fail;
    }
  } else {
    struct TDB_DATA iov[2] = {
      {
        .dptr = (uint8_t *)&watcher,
        .dsize = sizeof(struct notifyd_watcher),
      },
      {
        .dptr = (uint8_t *)instances,
        .dsize = sizeof(struct notifyd_instance) *
           num_instances,
      },
    };

    status = dbwrap_record_storev(rec, iov, ARRAY_SIZE(iov), 0);
    if (!NT_STATUS_IS_OK(status)) {
      DBG_WARNING("dbwrap_record_storev returned %s\n",
            nt_errstr(status));
      goto fail;
    }
  }

  ok = true;
fail:
  TALLOC_FREE(rec);
  return ok;
}

static void notifyd_sys_callback(struct sys_notify_context *ctx,
         void *private_data, struct notify_event *ev,
         uint32_t filter)
{
  struct messaging_context *msg_ctx = talloc_get_type_abort(
    private_data, struct messaging_context);
  struct notify_trigger_msg msg;
  struct iovec iov[4];
  char slash = '/';

  msg = (struct notify_trigger_msg) {
    .when = timespec_current(),
    .action = ev->action,
    .filter = filter,
  };

  iov[0].iov_base = &msg;
  iov[0].iov_len = offsetof(struct notify_trigger_msg, path);
  iov[1].iov_base = discard_const_p(char, ev->dir);
  iov[1].iov_len = strlen(ev->dir);
  iov[2].iov_base = &slash;
  iov[2].iov_len = 1;
  iov[3].iov_base = discard_const_p(char, ev->path);
  iov[3].iov_len = strlen(ev->path)+1;

  messaging_send_iov(
    msg_ctx, messaging_server_id(msg_ctx),
    MSG_SMB_NOTIFY_TRIGGER, iov, ARRAY_SIZE(iov), NULL, 0);
}

static bool notifyd_parse_rec_change(uint8_t *buf, size_t bufsize,
             struct notify_rec_change_msg **pmsg,
             size_t *pathlen)
{
  struct notify_rec_change_msg *msg;

  if (bufsize < offsetof(struct notify_rec_change_msg, path) + 1) {
    DBG_WARNING("message too short, ignoring: %zu\n", bufsize);
    return false;
  }

  *pmsg = msg = (struct notify_rec_change_msg *)buf;
  *pathlen = bufsize - offsetof(struct notify_rec_change_msg, path);

  DBG_DEBUG("Got rec_change_msg filter=%"PRIu32", "
      "subdir_filter=%"PRIu32", private_data=%p, path=%.*s\n",
      msg->instance.filter,
      msg->instance.subdir_filter,
      msg->instance.private_data,
      (int)(*pathlen),
      msg->path);

  return true;
}

static void notifyd_rec_change(struct messaging_context *msg_ctx,
             void *private_data, uint32_t msg_type,
             struct server_id src, DATA_BLOB *data)
{
  struct notifyd_state *state = talloc_get_type_abort(
    private_data, struct notifyd_state);
  struct server_id_buf idbuf;
  struct notify_rec_change_msg *msg;
  size_t pathlen;
  bool ok;
  struct notify_instance instance;

  DBG_DEBUG("Got %zu bytes from %s\n", data->length,
      server_id_str_buf(src, &idbuf));

  ok = notifyd_parse_rec_change(data->data, data->length,
              &msg, &pathlen);
  if (!ok) {
    return;
  }

  memcpy(&instance, &msg->instance, sizeof(instance)); /* avoid SIGBUS */

  ok = notifyd_apply_rec_change(
    &src, msg->path, pathlen, &instance,
    state->entries, state->sys_notify_watch, state->sys_notify_ctx,
    state->msg_ctx);
  if (!ok) {
    DBG_DEBUG("notifyd_apply_rec_change failed, ignoring\n");
    return;
  }

  if ((state->log == NULL) || (state->ctdbd_conn == NULL)) {
    return;
  }

#ifdef CLUSTER_SUPPORT
  {

  struct messaging_rec **tmp;
  struct messaging_reclog *log;
  struct iovec iov = { .iov_base = data->data, .iov_len = data->length };

  log = state->log;

  tmp = talloc_realloc(log, log->recs, struct messaging_rec *,
           log->num_recs+1);
  if (tmp == NULL) {
    DBG_WARNING("talloc_realloc failed, ignoring\n");
    return;
  }
  log->recs = tmp;

  log->recs[log->num_recs] = messaging_rec_create(
    log->recs, src, messaging_server_id(msg_ctx),
    msg_type, &iov, 1, NULL, 0);

  if (log->recs[log->num_recs] == NULL) {
    DBG_WARNING("messaging_rec_create failed, ignoring\n");
    return;
  }

  log->num_recs += 1;

  if (log->num_recs >= 100) {
    /*
     * Don't let the log grow too large
     */
    notifyd_broadcast_reclog(state->ctdbd_conn,
           messaging_server_id(msg_ctx), log);
  }

  }
#endif
}

struct notifyd_trigger_state {
  struct messaging_context *msg_ctx;
  struct notify_trigger_msg *msg;
  bool recursive;
  bool covered_by_sys_notify;
};

static void notifyd_trigger_parser(TDB_DATA key, TDB_DATA data,
           void *private_data);

static void notifyd_trigger(struct messaging_context *msg_ctx,
          void *private_data, uint32_t msg_type,
          struct server_id src, DATA_BLOB *data)
{
  struct notifyd_state *state = talloc_get_type_abort(
    private_data, struct notifyd_state);
  struct server_id my_id = messaging_server_id(msg_ctx);
  struct notifyd_trigger_state tstate;
  const char *path;
  const char *p, *next_p;

  if (data->length < offsetof(struct notify_trigger_msg, path) + 1) {
    DBG_WARNING("message too short, ignoring: %zu\n",
          data->length);
    return;
  }
  if (data->data[data->length-1] != 0) {
    DBG_WARNING("path not 0-terminated, ignoring\n");;
    return;
  }

  tstate.msg_ctx = msg_ctx;

  tstate.covered_by_sys_notify = (src.vnn == my_id.vnn);
  tstate.covered_by_sys_notify &= !server_id_equal(&src, &my_id);

  tstate.msg = (struct notify_trigger_msg *)data->data;
  path = tstate.msg->path;

  DBG_DEBUG("Got trigger_msg action=%"PRIu32", filter=%"PRIu32", "
      "path=%s\n",
      tstate.msg->action,
      tstate.msg->filter,
      path);

  if (path[0] != '/') {
    DBG_WARNING("path %s does not start with /, ignoring\n",
          path);
    return;
  }

  for (p = strchr(path+1, '/'); p != NULL; p = next_p) {
    ptrdiff_t path_len = p - path;
    TDB_DATA key;
    uint32_t i;

    next_p = strchr(p+1, '/');
    tstate.recursive = (next_p != NULL);

    DBG_DEBUG("Trying path %.*s\n", (int)path_len, path);

    key = (TDB_DATA) { .dptr = discard_const_p(uint8_t, path),
           .dsize = path_len };

    dbwrap_parse_record(state->entries, key,
            notifyd_trigger_parser, &tstate);

    if (state->peers == NULL) {
      continue;
    }

    if (src.vnn != my_id.vnn) {
      continue;
    }

    for (i=0; i<state->num_peers; i++) {
      if (state->peers[i]->db == NULL) {
        /*
         * Inactive peer, did not get a db yet
         */
        continue;
      }
      dbwrap_parse_record(state->peers[i]->db, key,
              notifyd_trigger_parser, &tstate);
    }
  }
}

static void notifyd_send_delete(struct messaging_context *msg_ctx,
        TDB_DATA key,
        struct notifyd_instance *instance);

static void notifyd_trigger_parser(TDB_DATA key, TDB_DATA data,
           void *private_data)

{
  struct notifyd_trigger_state *tstate = private_data;
  struct notify_event_msg msg = { .action = tstate->msg->action,
          .when = tstate->msg->when };
  struct iovec iov[2];
  size_t path_len = key.dsize;
  struct notifyd_watcher watcher = {};
  struct notifyd_instance *instances = NULL;
  size_t num_instances = 0;
  size_t i;
  bool ok;

  ok = notifyd_parse_entry(data.dptr,
         data.dsize,
         &watcher,
         &instances,
         &num_instances);
  if (!ok) {
    DBG_DEBUG("Could not parse notifyd_entry\n");
    return;
  }

  DBG_DEBUG("Found %zu instances for %.*s\n",
      num_instances,
      (int)key.dsize,
      (char *)key.dptr);

  iov[0].iov_base = &msg;
  iov[0].iov_len = offsetof(struct notify_event_msg, path);
  iov[1].iov_base = tstate->msg->path + path_len + 1;
  iov[1].iov_len = strlen((char *)(iov[1].iov_base)) + 1;

  for (i=0; i<num_instances; i++) {
    struct notifyd_instance *instance = &instances[i];
    struct server_id_buf idbuf;
    uint32_t i_filter;
    NTSTATUS status;

    if (tstate->covered_by_sys_notify) {
      if (tstate->recursive) {
        i_filter = watcher.sys_subdir_filter &
             instance->instance.subdir_filter;
      } else {
        i_filter = watcher.sys_filter &
             instance->instance.filter;
      }
    } else {
      if (tstate->recursive) {
        i_filter = instance->instance.subdir_filter;
      } else {
        i_filter = instance->instance.filter;
      }
    }

    if ((i_filter & tstate->msg->filter) == 0) {
      continue;
    }

    msg.private_data = instance->instance.private_data;

    status = messaging_send_iov(
      tstate->msg_ctx, instance->client,
      MSG_PVFS_NOTIFY, iov, ARRAY_SIZE(iov), NULL, 0);

    DBG_DEBUG("messaging_send_iov to %s returned %s\n",
        server_id_str_buf(instance->client, &idbuf),
        nt_errstr(status));

    if (NT_STATUS_EQUAL(status, NT_STATUS_OBJECT_NAME_NOT_FOUND) &&
        procid_is_local(&instance->client)) {
      /*
       * That process has died
       */
      notifyd_send_delete(tstate->msg_ctx, key, instance);
      continue;
    }

    if (!NT_STATUS_IS_OK(status)) {
      DBG_WARNING("messaging_send_iov returned %s\n",
            nt_errstr(status));
    }
  }
}

/*
 * Send a delete request to ourselves to properly discard a notify
 * record for an smbd that has died.
 */

static void notifyd_send_delete(struct messaging_context *msg_ctx,
        TDB_DATA key,
        struct notifyd_instance *instance)
{
  struct notify_rec_change_msg msg = {
    .instance.private_data = instance->instance.private_data
  };
  uint8_t nul = 0;
  struct iovec iov[3];
  NTSTATUS status;

  /*
   * Send a rec_change to ourselves to delete a dead entry
   */

  iov[0] = (struct iovec) {
    .iov_base = &msg,
    .iov_len = offsetof(struct notify_rec_change_msg, path) };
  iov[1] = (struct iovec) { .iov_base = key.dptr, .iov_len = key.dsize };
  iov[2] = (struct iovec) { .iov_base = &nul, .iov_len = sizeof(nul) };

  status = messaging_send_iov(msg_ctx,
            instance->client,
            MSG_SMB_NOTIFY_REC_CHANGE,
            iov,
            ARRAY_SIZE(iov),
            NULL,
            0);

  if (!NT_STATUS_IS_OK(status)) {
    DBG_WARNING("messaging_send_iov failed: %s\n",
          nt_errstr(status));
  }
}

static void notifyd_get_db(struct messaging_context *msg_ctx,
         void *private_data, uint32_t msg_type,
         struct server_id src, DATA_BLOB *data)
{
  struct notifyd_state *state = talloc_get_type_abort(
    private_data, struct notifyd_state);
  struct server_id_buf id1, id2;
  NTSTATUS status;
  uint64_t rec_index = UINT64_MAX;
  uint8_t index_buf[sizeof(uint64_t)];
  size_t dbsize;
  uint8_t *buf;
  struct iovec iov[2];

  dbsize = dbwrap_marshall(state->entries, NULL, 0);

  buf = talloc_array(talloc_tos(), uint8_t, dbsize);
  if (buf == NULL) {
    DBG_WARNING("talloc_array(%zu) failed\n", dbsize);
    return;
  }

  dbsize = dbwrap_marshall(state->entries, buf, dbsize);

  if (dbsize != talloc_get_size(buf)) {
    DBG_DEBUG("dbsize changed: %zu->%zu\n",
        talloc_get_size(buf),
        dbsize);
    TALLOC_FREE(buf);
    return;
  }

  if (state->log != NULL) {
    rec_index = state->log->rec_index;
  }
  SBVAL(index_buf, 0, rec_index);

  iov[0] = (struct iovec) { .iov_base = index_buf,
          .iov_len = sizeof(index_buf) };
  iov[1] = (struct iovec) { .iov_base = buf,
          .iov_len = dbsize };

  DBG_DEBUG("Sending %zu bytes to %s->%s\n",
      iov_buflen(iov, ARRAY_SIZE(iov)),
      server_id_str_buf(messaging_server_id(msg_ctx), &id1),
      server_id_str_buf(src, &id2));

  status = messaging_send_iov(msg_ctx, src, MSG_SMB_NOTIFY_DB,
            iov, ARRAY_SIZE(iov), NULL, 0);
  TALLOC_FREE(buf);
  if (!NT_STATUS_IS_OK(status)) {
    DBG_WARNING("messaging_send_iov failed: %s\n",
          nt_errstr(status));
  }
}

#ifdef CLUSTER_SUPPORT

static int notifyd_add_proxy_syswatches(struct db_record *rec,
          void *private_data);

static void notifyd_got_db(struct messaging_context *msg_ctx,
         void *private_data, uint32_t msg_type,
         struct server_id src, DATA_BLOB *data)
{
  struct notifyd_state *state = talloc_get_type_abort(
    private_data, struct notifyd_state);
  struct notifyd_peer *p = NULL;
  struct server_id_buf idbuf;
  NTSTATUS status;
  int count;
  size_t i;

  for (i=0; i<state->num_peers; i++) {
    if (server_id_equal(&src, &state->peers[i]->pid)) {
      p = state->peers[i];
      break;
    }
  }

  if (p == NULL) {
    DBG_DEBUG("Did not find peer for db from %s\n",
        server_id_str_buf(src, &idbuf));
    return;
  }

  if (data->length < 8) {
    DBG_DEBUG("Got short db length %zu from %s\n", data->length,
         server_id_str_buf(src, &idbuf));
    TALLOC_FREE(p);
    return;
  }

  p->rec_index = BVAL(data->data, 0);

  p->db = db_open_rbt(p);
  if (p->db == NULL) {
    DBG_DEBUG("db_open_rbt failed\n");
    TALLOC_FREE(p);
    return;
  }

  status = dbwrap_unmarshall(p->db, data->data + 8,
           data->length - 8);
  if (!NT_STATUS_IS_OK(status)) {
    DBG_DEBUG("dbwrap_unmarshall returned %s for db %s\n",
        nt_errstr(status),
        server_id_str_buf(src, &idbuf));
    TALLOC_FREE(p);
    return;
  }

  dbwrap_traverse_read(p->db, notifyd_add_proxy_syswatches, state,
           &count);

  DBG_DEBUG("Database from %s contained %d records\n",
      server_id_str_buf(src, &idbuf),
      count);
}

static void notifyd_broadcast_reclog(struct ctdbd_connection *ctdbd_conn,
             struct server_id src,
             struct messaging_reclog *log)
{
  enum ndr_err_code ndr_err;
  uint8_t msghdr[MESSAGE_HDR_LENGTH];
  DATA_BLOB blob;
  struct iovec iov[2];
  int ret;

  if (log == NULL) {
    return;
  }

  DBG_DEBUG("rec_index=%"PRIu64", num_recs=%"PRIu32"\n",
      log->rec_index,
      log->num_recs);

  message_hdr_put(msghdr, MSG_SMB_NOTIFY_REC_CHANGES, src,
      (struct server_id) {0 });
  iov[0] = (struct iovec) { .iov_base = msghdr,
          .iov_len = sizeof(msghdr) };

  ndr_err = ndr_push_struct_blob(
    &blob, log, log,
    (ndr_push_flags_fn_t)ndr_push_messaging_reclog);
  if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
    DBG_WARNING("ndr_push_messaging_recs failed: %s\n",
          ndr_errstr(ndr_err));
    goto done;
  }
  iov[1] = (struct iovec) { .iov_base = blob.data,
          .iov_len = blob.length };

  ret = ctdbd_messaging_send_iov(
    ctdbd_conn, CTDB_BROADCAST_CONNECTED,
    CTDB_SRVID_SAMBA_NOTIFY_PROXY, iov, ARRAY_SIZE(iov));
  TALLOC_FREE(blob.data);
  if (ret != 0) {
    DBG_WARNING("ctdbd_messaging_send failed: %s\n",
          strerror(ret));
    goto done;
  }

  log->rec_index += 1;

done:
  log->num_recs = 0;
  TALLOC_FREE(log->recs);
}

struct notifyd_broadcast_reclog_state {
  struct tevent_context *ev;
  struct ctdbd_connection *ctdbd_conn;
  struct server_id src;
  struct messaging_reclog *log;
};

static void notifyd_broadcast_reclog_next(struct tevent_req *subreq);

static struct tevent_req *notifyd_broadcast_reclog_send(
  TALLOC_CTX *mem_ctx, struct tevent_context *ev,
  struct ctdbd_connection *ctdbd_conn, struct server_id src,
  struct messaging_reclog *log)
{
  struct tevent_req *req, *subreq;
  struct notifyd_broadcast_reclog_state *state;

  req = tevent_req_create(mem_ctx, &state,
        struct notifyd_broadcast_reclog_state);
  if (req == NULL) {
    return NULL;
  }
  state->ev = ev;
  state->ctdbd_conn = ctdbd_conn;
  state->src = src;
  state->log = log;

  subreq = tevent_wakeup_send(state, state->ev,
            timeval_current_ofs_msec(1000));
  if (tevent_req_nomem(subreq, req)) {
    return tevent_req_post(req, ev);
  }
  tevent_req_set_callback(subreq, notifyd_broadcast_reclog_next, req);
  return req;
}

static void notifyd_broadcast_reclog_next(struct tevent_req *subreq)
{
  struct tevent_req *req = tevent_req_callback_data(
    subreq, struct tevent_req);
  struct notifyd_broadcast_reclog_state *state = tevent_req_data(
    req, struct notifyd_broadcast_reclog_state);
  bool ok;

  ok = tevent_wakeup_recv(subreq);
  TALLOC_FREE(subreq);
  if (!ok) {
    tevent_req_oom(req);
    return;
  }

  notifyd_broadcast_reclog(state->ctdbd_conn, state->src, state->log);

  subreq = tevent_wakeup_send(state, state->ev,
            timeval_current_ofs_msec(1000));
  if (tevent_req_nomem(subreq, req)) {
    return;
  }
  tevent_req_set_callback(subreq, notifyd_broadcast_reclog_next, req);
}

static int notifyd_broadcast_reclog_recv(struct tevent_req *req)
{
  return tevent_req_simple_recv_unix(req);
}

struct notifyd_clean_peers_state {
  struct tevent_context *ev;
  struct notifyd_state *notifyd;
};

static void notifyd_clean_peers_next(struct tevent_req *subreq);

static struct tevent_req *notifyd_clean_peers_send(
  TALLOC_CTX *mem_ctx, struct tevent_context *ev,
  struct notifyd_state *notifyd)
{
  struct tevent_req *req, *subreq;
  struct notifyd_clean_peers_state *state;

  req = tevent_req_create(mem_ctx, &state,
        struct notifyd_clean_peers_state);
  if (req == NULL) {
    return NULL;
  }
  state->ev = ev;
  state->notifyd = notifyd;

  subreq = tevent_wakeup_send(state, state->ev,
            timeval_current_ofs_msec(30000));
  if (tevent_req_nomem(subreq, req)) {
    return tevent_req_post(req, ev);
  }
  tevent_req_set_callback(subreq, notifyd_clean_peers_next, req);
  return req;
}

static void notifyd_clean_peers_next(struct tevent_req *subreq)
{
  struct tevent_req *req = tevent_req_callback_data(
    subreq, struct tevent_req);
  struct notifyd_clean_peers_state *state = tevent_req_data(
    req, struct notifyd_clean_peers_state);
  struct notifyd_state *notifyd = state->notifyd;
  size_t i;
  bool ok;
  time_t now = time(NULL);

  ok = tevent_wakeup_recv(subreq);
  TALLOC_FREE(subreq);
  if (!ok) {
    tevent_req_oom(req);
    return;
  }

  i = 0;
  while (i < notifyd->num_peers) {
    struct notifyd_peer *p = notifyd->peers[i];

    if ((now - p->last_broadcast) > 60) {
      struct server_id_buf idbuf;

      /*
       * Haven't heard for more than 60 seconds. Call this
       * peer dead
       */

      DBG_DEBUG("peer %s died\n",
          server_id_str_buf(p->pid, &idbuf));
      /*
       * This implicitly decrements notifyd->num_peers
       */
      TALLOC_FREE(p);
    } else {
      i += 1;
    }
  }

  subreq = tevent_wakeup_send(state, state->ev,
            timeval_current_ofs_msec(30000));
  if (tevent_req_nomem(subreq, req)) {
    return;
  }
  tevent_req_set_callback(subreq, notifyd_clean_peers_next, req);
}

static int notifyd_clean_peers_recv(struct tevent_req *req)
{
  return tevent_req_simple_recv_unix(req);
}

static int notifyd_add_proxy_syswatches(struct db_record *rec,
          void *private_data)
{
  struct notifyd_state *state = talloc_get_type_abort(
    private_data, struct notifyd_state);
  struct db_context *db = dbwrap_record_get_db(rec);
  TDB_DATA key = dbwrap_record_get_key(rec);
  TDB_DATA value = dbwrap_record_get_value(rec);
  struct notifyd_watcher watcher = {};
  char path[key.dsize+1];
  bool ok;
  int ret;

  memcpy(path, key.dptr, key.dsize);
  path[key.dsize] = '\0';

  /* This is a remote database, we just need the watcher. */
  ok = notifyd_parse_entry(value.dptr, value.dsize, &watcher, NULL, NULL);
  if (!ok) {
    DBG_WARNING("Could not parse notifyd entry for %s\n", path);
    return 0;
  }

  watcher.sys_watch = NULL;
  watcher.sys_filter = watcher.filter;
  watcher.sys_subdir_filter = watcher.subdir_filter;

  ret = state->sys_notify_watch(db,
              state->sys_notify_ctx,
              path,
              &watcher.filter,
              &watcher.subdir_filter,
              notifyd_sys_callback,
              state->msg_ctx,
              &watcher.sys_watch);
  if (ret != 0) {
    DBG_WARNING("inotify_watch returned %s\n", strerror(errno));
  }

  memcpy(value.dptr, &watcher, sizeof(struct notifyd_watcher));

  return 0;
}

static int notifyd_db_del_syswatches(struct db_record *rec, void *private_data)
{
  TDB_DATA key = dbwrap_record_get_key(rec);
  TDB_DATA value = dbwrap_record_get_value(rec);
  struct notifyd_watcher watcher = {};
  bool ok;

  ok = notifyd_parse_entry(value.dptr, value.dsize, &watcher, NULL, NULL);
  if (!ok) {
    DBG_WARNING("Could not parse notifyd entry for %.*s\n",
          (int)key.dsize, (char *)key.dptr);
    return 0;
  }
  TALLOC_FREE(watcher.sys_watch);

  return 0;
}

static int notifyd_peer_destructor(struct notifyd_peer *p)
{
  struct notifyd_state *state = p->state;
  size_t i;

  if (p->db != NULL) {
    dbwrap_traverse_read(p->db, notifyd_db_del_syswatches,
             NULL, NULL);
  }

  for (i = 0; i<state->num_peers; i++) {
    if (p == state->peers[i]) {
      state->peers[i] = state->peers[state->num_peers-1];
      state->num_peers -= 1;
      break;
    }
  }
  return 0;
}

static struct notifyd_peer *notifyd_peer_new(
  struct notifyd_state *state, struct server_id pid)
{
  struct notifyd_peer *p, **tmp;

  tmp = talloc_realloc(state, state->peers, struct notifyd_peer *,
           state->num_peers+1);
  if (tmp == NULL) {
    return NULL;
  }
  state->peers = tmp;

  p = talloc_zero(state->peers, struct notifyd_peer);
  if (p == NULL) {
    return NULL;
  }
  p->state = state;
  p->pid = pid;

  state->peers[state->num_peers] = p;
  state->num_peers += 1;

  talloc_set_destructor(p, notifyd_peer_destructor);

  return p;
}

static void notifyd_apply_reclog(struct notifyd_peer *peer,
         const uint8_t *msg, size_t msglen)
{
  struct notifyd_state *state = peer->state;
  DATA_BLOB blob = { .data = discard_const_p(uint8_t, msg),
         .length = msglen };
  struct server_id_buf idbuf;
  struct messaging_reclog *log;
  enum ndr_err_code ndr_err;
  uint32_t i;

  if (peer->db == NULL) {
    /*
     * No db yet
     */
    return;
  }

  log = talloc(peer, struct messaging_reclog);
  if (log == NULL) {
    DBG_DEBUG("talloc failed\n");
    return;
  }

  ndr_err = ndr_pull_struct_blob_all(
    &blob, log, log,
    (ndr_pull_flags_fn_t)ndr_pull_messaging_reclog);
  if (!NDR_ERR_CODE_IS_SUCCESS(ndr_err)) {
    DBG_DEBUG("ndr_pull_messaging_reclog failed: %s\n",
        ndr_errstr(ndr_err));
    goto fail;
  }

  DBG_DEBUG("Got %"PRIu32" recs index %"PRIu64" from %s\n",
      log->num_recs,
      log->rec_index,
      server_id_str_buf(peer->pid, &idbuf));

  if (log->rec_index != peer->rec_index) {
    DBG_INFO("Got rec index %"PRIu64" from %s, "
       "expected %"PRIu64"\n",
       log->rec_index,
       server_id_str_buf(peer->pid, &idbuf),
       peer->rec_index);
    goto fail;
  }

  for (i=0; i<log->num_recs; i++) {
    struct messaging_rec *r = log->recs[i];
    struct notify_rec_change_msg *chg;
    size_t pathlen;
    bool ok;
    struct notify_instance instance;

    ok = notifyd_parse_rec_change(r->buf.data, r->buf.length,
                &chg, &pathlen);
    if (!ok) {
      DBG_INFO("notifyd_parse_rec_change failed\n");
      goto fail;
    }

    /* avoid SIGBUS */
    memcpy(&instance, &chg->instance, sizeof(instance));

    ok = notifyd_apply_rec_change(&r->src, chg->path, pathlen,
                &instance, peer->db,
                state->sys_notify_watch,
                state->sys_notify_ctx,
                state->msg_ctx);
    if (!ok) {
      DBG_INFO("notifyd_apply_rec_change failed\n");
      goto fail;
    }
  }

  peer->rec_index += 1;
  peer->last_broadcast = time(NULL);

  TALLOC_FREE(log);
  return;

fail:
  DBG_DEBUG("Dropping peer %s\n",
      server_id_str_buf(peer->pid, &idbuf));
  TALLOC_FREE(peer);
}

/*
 * Receive messaging_reclog (log of MSG_SMB_NOTIFY_REC_CHANGE
 * messages) broadcasts by other notifyds. Several cases:
 *
 * We don't know the source. This creates a new peer. Creating a peer
 * involves asking the peer for its full database. We assume ordered
 * messages, so the new database will arrive before the next broadcast
 * will.
 *
 * We know the source and the log index matches. We will apply the log
 * locally to our peer's db as if we had received it from a local
 * client.
 *
 * We know the source but the log index does not match. This means we
 * lost a message. We just drop the whole peer and wait for the next
 * broadcast, which will then trigger a fresh database pull.
 */

static int notifyd_snoop_broadcast(struct tevent_context *ev,
           uint32_t src_vnn, uint32_t dst_vnn,
           uint64_t dst_srvid,
           const uint8_t *msg, size_t msglen,
           void *private_data)
{
  struct notifyd_state *state = talloc_get_type_abort(
    private_data, struct notifyd_state);
  struct server_id my_id = messaging_server_id(state->msg_ctx);
  struct notifyd_peer *p;
  uint32_t i;
  uint32_t msg_type;
  struct server_id src, dst;
  struct server_id_buf idbuf;
  NTSTATUS status;

  if (msglen < MESSAGE_HDR_LENGTH) {
    DBG_DEBUG("Got short broadcast\n");
    return 0;
  }
  message_hdr_get(&msg_type, &src, &dst, msg);

  if (msg_type != MSG_SMB_NOTIFY_REC_CHANGES) {
    DBG_DEBUG("Got message %"PRIu32", ignoring\n", msg_type);
    return 0;
  }
  if (server_id_equal(&src, &my_id)) {
    DBG_DEBUG("Ignoring my own broadcast\n");
    return 0;
  }

  DBG_DEBUG("Got MSG_SMB_NOTIFY_REC_CHANGES from %s\n",
       server_id_str_buf(src, &idbuf));

  for (i=0; i<state->num_peers; i++) {
    if (server_id_equal(&state->peers[i]->pid, &src)) {

      DBG_DEBUG("Applying changes to peer %"PRIu32"\n", i);

      notifyd_apply_reclog(state->peers[i],
               msg + MESSAGE_HDR_LENGTH,
               msglen - MESSAGE_HDR_LENGTH);
      return 0;
    }
  }

  DBG_DEBUG("Creating new peer for %s\n",
       server_id_str_buf(src, &idbuf));

  p = notifyd_peer_new(state, src);
  if (p == NULL) {
    DBG_DEBUG("notifyd_peer_new failed\n");
    return 0;
  }

  status = messaging_send_buf(state->msg_ctx, src, MSG_SMB_NOTIFY_GET_DB,
            NULL, 0);
  if (!NT_STATUS_IS_OK(status)) {
    DBG_DEBUG("messaging_send_buf failed: %s\n",
        nt_errstr(status));
    TALLOC_FREE(p);
    return 0;
  }

  return 0;
}
#endif

Coverage Report

Created: 2026-01-16 06:47