source3/lib/dbwrap/dbwrap_watch.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3    Watch dbwrap record changes
   4    Copyright (C) Volker Lendecke 2012
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "includes.h"
  21 #include "system/filesys.h"
  22 #include "dbwrap/dbwrap.h"
  23 #include "dbwrap_watch.h"
  24 #include "dbwrap_open.h"
  25 #include "lib/util/util_tdb.h"
  26 #include "lib/util/tevent_ntstatus.h"
  27
  28 static struct db_context *dbwrap_record_watchers_db(void)
  29 {
  30         static struct db_context *watchers_db;
  31
  32         if (watchers_db == NULL) {
  33                 watchers_db = db_open(
  34                         NULL, lock_path("dbwrap_watchers.tdb"), 0,
  35                         TDB_CLEAR_IF_FIRST | TDB_INCOMPATIBLE_HASH,
  36                         O_RDWR|O_CREAT, 0600, DBWRAP_LOCK_ORDER_3,
  37                         DBWRAP_FLAG_NONE);
  38         }
  39         return watchers_db;
  40 }
  41
  42 static TDB_DATA dbwrap_record_watchers_key(TALLOC_CTX *mem_ctx,
  43                                            struct db_context *db,
  44                                            struct db_record *rec,
  45                                            TDB_DATA *rec_key)
  46 {
  47         const uint8_t *db_id;
  48         size_t db_id_len;
  49         TDB_DATA key, wkey;
  50
  51         dbwrap_db_id(db, &db_id, &db_id_len);
  52         key = dbwrap_record_get_key(rec);
  53
  54         wkey.dsize = sizeof(uint32_t) + db_id_len + key.dsize;
  55         wkey.dptr = talloc_array(mem_ctx, uint8_t, wkey.dsize);
  56         if (wkey.dptr == NULL) {
  57                 return make_tdb_data(NULL, 0);
  58         }
  59         SIVAL(wkey.dptr, 0, db_id_len);
  60         memcpy(wkey.dptr + sizeof(uint32_t), db_id, db_id_len);
  61         memcpy(wkey.dptr + sizeof(uint32_t) + db_id_len, key.dptr, key.dsize);
  62
  63         if (rec_key != NULL) {
  64                 rec_key->dptr = wkey.dptr + sizeof(uint32_t) + db_id_len;
  65                 rec_key->dsize = key.dsize;
  66         }
  67
  68         return wkey;
  69 }
  70
  71 static bool dbwrap_record_watchers_key_parse(
  72         TDB_DATA wkey, uint8_t **p_db_id, size_t *p_db_id_len, TDB_DATA *key)
  73 {
  74         size_t db_id_len;
  75
  76         if (wkey.dsize < sizeof(uint32_t)) {
  77                 DEBUG(1, ("Invalid watchers key\n"));
  78                 return false;
  79         }
  80         db_id_len = IVAL(wkey.dptr, 0);
  81         if (db_id_len > (wkey.dsize - sizeof(uint32_t))) {
  82                 DEBUG(1, ("Invalid watchers key, wkey.dsize=%d, "
  83                           "db_id_len=%d\n", (int)wkey.dsize, (int)db_id_len));
  84                 return false;
  85         }
  86         *p_db_id = wkey.dptr + sizeof(uint32_t);
  87         *p_db_id_len = db_id_len;
  88         key->dptr = wkey.dptr + sizeof(uint32_t) + db_id_len;
  89         key->dsize = wkey.dsize - sizeof(uint32_t) - db_id_len;
  90         return true;
  91 }
  92
  93 static NTSTATUS dbwrap_record_add_watcher(TDB_DATA w_key, struct server_id id)
  94 {
  95         struct TALLOC_CTX *frame = talloc_stackframe();
  96         struct db_context *db;
  97         struct db_record *rec;
  98         TDB_DATA value;
  99         struct server_id *ids;
 100         size_t num_ids;
 101         NTSTATUS status;
 102
 103         db = dbwrap_record_watchers_db();
 104         if (db == NULL) {
 105                 status = map_nt_error_from_unix(errno);
 106                 goto fail;
 107         }
 108         rec = dbwrap_fetch_locked(db, talloc_tos(), w_key);
 109         if (rec == NULL) {
 110                 status = map_nt_error_from_unix(errno);
 111                 goto fail;
 112         }
 113         value = dbwrap_record_get_value(rec);
 114
 115         if ((value.dsize % sizeof(struct server_id)) != 0) {
 116                 status = NT_STATUS_INTERNAL_DB_CORRUPTION;
 117                 goto fail;
 118         }
 119
 120         ids = (struct server_id *)value.dptr;
 121         num_ids = value.dsize / sizeof(struct server_id);
 122
 123         ids = talloc_array(talloc_tos(), struct server_id,
 124                            num_ids + 1);
 125         if (ids == NULL) {
 126                 status = NT_STATUS_NO_MEMORY;
 127                 goto fail;
 128         }
 129         memcpy(ids, value.dptr, value.dsize);
 130         ids[num_ids] = id;
 131         num_ids += 1;
 132
 133         status = dbwrap_record_store(
 134                 rec, make_tdb_data((uint8_t *)ids, talloc_get_size(ids)), 0);
 135 fail:
 136         TALLOC_FREE(frame);
 137         return status;
 138 }
 139
 140 static NTSTATUS dbwrap_record_del_watcher(TDB_DATA w_key, struct server_id id)
 141 {
 142         struct TALLOC_CTX *frame = talloc_stackframe();
 143         struct db_context *db;
 144         struct db_record *rec;
 145         struct server_id *ids;
 146         size_t i, num_ids;
 147         TDB_DATA value;
 148         NTSTATUS status;
 149
 150         db = dbwrap_record_watchers_db();
 151         if (db == NULL) {
 152                 status = map_nt_error_from_unix(errno);
 153                 goto fail;
 154         }
 155         rec = dbwrap_fetch_locked(db, talloc_tos(), w_key);
 156         if (rec == NULL) {
 157                 status = map_nt_error_from_unix(errno);
 158                 goto fail;
 159         }
 160         value = dbwrap_record_get_value(rec);
 161
 162         if ((value.dsize % sizeof(struct server_id)) != 0) {
 163                 status = NT_STATUS_INTERNAL_DB_CORRUPTION;
 164                 goto fail;
 165         }
 166
 167         ids = (struct server_id *)value.dptr;
 168         num_ids = value.dsize / sizeof(struct server_id);
 169
 170         for (i=0; i<num_ids; i++) {
 171                 if (serverid_equal(&id, &ids[i])) {
 172                         ids[i] = ids[num_ids-1];
 173                         value.dsize -= sizeof(struct server_id);
 174                         break;
 175                 }
 176         }
 177         if (value.dsize == 0) {
 178                 status = dbwrap_record_delete(rec);
 179                 goto done;
 180         }
 181         status = dbwrap_record_store(rec, value, 0);
 182 fail:
 183 done:
 184         TALLOC_FREE(frame);
 185         return status;
 186 }
 187
 188 static NTSTATUS dbwrap_record_get_watchers(struct db_context *db,
 189                                            struct db_record *rec,
 190                                            TALLOC_CTX *mem_ctx,
 191                                            struct server_id **p_ids,
 192                                            size_t *p_num_ids)
 193 {
 194         struct db_context *w_db;
 195         TDB_DATA key = { 0, };
 196         TDB_DATA value = { 0, };
 197         struct server_id *ids;
 198         NTSTATUS status;
 199
 200         key = dbwrap_record_watchers_key(talloc_tos(), db, rec, NULL);
 201         if (key.dptr == NULL) {
 202                 status = NT_STATUS_NO_MEMORY;
 203                 goto fail;
 204         }
 205         w_db = dbwrap_record_watchers_db();
 206         if (w_db == NULL) {
 207                 status = NT_STATUS_INTERNAL_ERROR;
 208                 goto fail;
 209         }
 210         status = dbwrap_fetch(w_db, mem_ctx, key, &value);
 211         TALLOC_FREE(key.dptr);
 212         if (!NT_STATUS_IS_OK(status)) {
 213                 goto fail;
 214         }
 215         if ((value.dsize % sizeof(struct server_id)) != 0) {
 216                 status = NT_STATUS_INTERNAL_DB_CORRUPTION;
 217                 goto fail;
 218         }
 219         ids = (struct server_id *)value.dptr;
 220         *p_ids = talloc_move(mem_ctx, &ids);
 221         *p_num_ids = value.dsize / sizeof(struct server_id);
 222         return NT_STATUS_OK;
 223 fail:
 224         TALLOC_FREE(key.dptr);
 225         TALLOC_FREE(value.dptr);
 226         return status;
 227 }
 228
 229 struct dbwrap_record_watch_state {
 230         struct tevent_context *ev;
 231         struct db_context *db;
 232         struct tevent_req *req;
 233         struct messaging_context *msg;
 234         TDB_DATA key;
 235         TDB_DATA w_key;
 236 };
 237
 238 static bool dbwrap_record_watch_filter(struct messaging_rec *rec,
 239                                        void *private_data);
 240 static void dbwrap_record_watch_done(struct tevent_req *subreq);
 241 static int dbwrap_record_watch_state_destructor(
 242         struct dbwrap_record_watch_state *state);
 243
 244 struct tevent_req *dbwrap_record_watch_send(TALLOC_CTX *mem_ctx,
 245                                             struct tevent_context *ev,
 246                                             struct db_record *rec,
 247                                             struct messaging_context *msg)
 248 {
 249         struct tevent_req *req, *subreq;
 250         struct dbwrap_record_watch_state *state;
 251         struct db_context *watchers_db;
 252         NTSTATUS status;
 253
 254         req = tevent_req_create(mem_ctx, &state,
 255                                 struct dbwrap_record_watch_state);
 256         if (req == NULL) {
 257                 return NULL;
 258         }
 259         state->db = dbwrap_record_get_db(rec);
 260         state->ev = ev;
 261         state->req = req;
 262         state->msg = msg;
 263
 264         watchers_db = dbwrap_record_watchers_db();
 265         if (watchers_db == NULL) {
 266                 tevent_req_nterror(req, map_nt_error_from_unix(errno));
 267                 return tevent_req_post(req, ev);
 268         }
 269
 270         state->w_key = dbwrap_record_watchers_key(state, state->db, rec,
 271                                                   &state->key);
 272         if (tevent_req_nomem(state->w_key.dptr, req)) {
 273                 return tevent_req_post(req, ev);
 274         }
 275
 276         subreq = messaging_filtered_read_send(
 277                 state, ev, state->msg, dbwrap_record_watch_filter, state);
 278         if (tevent_req_nomem(subreq, req)) {
 279                 return tevent_req_post(req, ev);
 280         }
 281         tevent_req_set_callback(subreq, dbwrap_record_watch_done, req);
 282
 283         status = dbwrap_record_add_watcher(
 284                 state->w_key, messaging_server_id(state->msg));
 285         if (tevent_req_nterror(req, status)) {
 286                 return tevent_req_post(req, ev);
 287         }
 288         talloc_set_destructor(state, dbwrap_record_watch_state_destructor);
 289
 290         return req;
 291 }
 292
 293 static bool dbwrap_record_watch_filter(struct messaging_rec *rec,
 294                                        void *private_data)
 295 {
 296         struct dbwrap_record_watch_state *state = talloc_get_type_abort(
 297                 private_data, struct dbwrap_record_watch_state);
 298
 299         if (rec->msg_type != MSG_DBWRAP_MODIFIED) {
 300                 return false;
 301         }
 302         if (rec->buf.length != state->w_key.dsize) {
 303                 return false;
 304         }
 305         return memcmp(rec->buf.data, state->w_key.dptr, rec->buf.length) == 0;
 306 }
 307
 308 static int dbwrap_record_watch_state_destructor(
 309         struct dbwrap_record_watch_state *s)
 310 {
 311         if (s->msg != NULL) {
 312                 dbwrap_record_del_watcher(
 313                         s->w_key, messaging_server_id(s->msg));
 314         }
 315         return 0;
 316 }
 317
 318 static void dbwrap_watch_record_stored(struct db_context *db,
 319                                        struct db_record *rec,
 320                                        void *private_data)
 321 {
 322         struct messaging_context *msg = talloc_get_type_abort(
 323                 private_data, struct messaging_context);
 324         struct server_id *ids = NULL;
 325         size_t num_ids = 0;
 326         TDB_DATA w_key = { 0, };
 327         NTSTATUS status;
 328         uint32_t i;
 329
 330         status = dbwrap_record_get_watchers(db, rec, talloc_tos(),
 331                                             &ids, &num_ids);
 332         if (NT_STATUS_EQUAL(status, NT_STATUS_NOT_FOUND)) {
 333                 goto done;
 334         }
 335         if (!NT_STATUS_IS_OK(status)) {
 336                 DEBUG(1, ("dbwrap_record_get_watchers failed: %s\n",
 337                           nt_errstr(status)));
 338                 goto done;
 339         }
 340         w_key = dbwrap_record_watchers_key(talloc_tos(), db, rec, NULL);
 341         if (w_key.dptr == NULL) {
 342                 DEBUG(1, ("dbwrap_record_watchers_key failed\n"));
 343                 goto done;
 344         }
 345
 346         for (i=0; i<num_ids; i++) {
 347                 status = messaging_send_buf(msg, ids[i], MSG_DBWRAP_MODIFIED,
 348                                             w_key.dptr, w_key.dsize);
 349                 if (!NT_STATUS_IS_OK(status)) {
 350                         char *str = procid_str_static(&ids[i]);
 351                         DEBUG(1, ("messaging_send to %s failed: %s\n",
 352                                   str, nt_errstr(status)));
 353                         TALLOC_FREE(str);
 354                 }
 355         }
 356 done:
 357         TALLOC_FREE(w_key.dptr);
 358         TALLOC_FREE(ids);
 359         return;
 360 }
 361
 362 void dbwrap_watch_db(struct db_context *db, struct messaging_context *msg)
 363 {
 364         dbwrap_set_stored_callback(db, dbwrap_watch_record_stored, msg);
 365 }
 366
 367 static void dbwrap_record_watch_done(struct tevent_req *subreq)
 368 {
 369         struct tevent_req *req = tevent_req_callback_data(
 370                 subreq, struct tevent_req);
 371         struct messaging_rec *rec;
 372         int ret;
 373
 374         ret = messaging_filtered_read_recv(subreq, talloc_tos(), &rec);
 375         TALLOC_FREE(subreq);
 376         if (ret != 0) {
 377                 tevent_req_nterror(req, map_nt_error_from_unix(ret));
 378                 return;
 379         }
 380         tevent_req_done(req);
 381 }
 382
 383 NTSTATUS dbwrap_record_watch_recv(struct tevent_req *req,
 384                                   TALLOC_CTX *mem_ctx,
 385                                   struct db_record **prec)
 386 {
 387         struct dbwrap_record_watch_state *state = tevent_req_data(
 388                 req, struct dbwrap_record_watch_state);
 389         NTSTATUS status;
 390         struct db_record *rec;
 391
 392         if (tevent_req_is_nterror(req, &status)) {
 393                 return status;
 394         }
 395         if (prec == NULL) {
 396                 return NT_STATUS_OK;
 397         }
 398         rec = dbwrap_fetch_locked(state->db, mem_ctx, state->key);
 399         if (rec == NULL) {
 400                 return NT_STATUS_INTERNAL_DB_ERROR;
 401         }
 402         *prec = rec;
 403         return NT_STATUS_OK;
 404 }
 405
 406 struct dbwrap_watchers_traverse_read_state {
 407         int (*fn)(const uint8_t *db_id, size_t db_id_len, const TDB_DATA key,
 408                   const struct server_id *watchers, size_t num_watchers,
 409                   void *private_data);
 410         void *private_data;
 411 };
 412
 413 static int dbwrap_watchers_traverse_read_callback(
 414         struct db_record *rec, void *private_data)
 415 {
 416         struct dbwrap_watchers_traverse_read_state *state =
 417                 (struct dbwrap_watchers_traverse_read_state *)private_data;
 418         uint8_t *db_id;
 419         size_t db_id_len;
 420         TDB_DATA w_key, key, w_data;
 421         int res;
 422
 423         w_key = dbwrap_record_get_key(rec);
 424         w_data = dbwrap_record_get_value(rec);
 425
 426         if (!dbwrap_record_watchers_key_parse(w_key, &db_id, &db_id_len,
 427                                               &key)) {
 428                 return 0;
 429         }
 430         if ((w_data.dsize % sizeof(struct server_id)) != 0) {
 431                 return 0;
 432         }
 433         res = state->fn(db_id, db_id_len, key,
 434                         (struct server_id *)w_data.dptr,
 435                         w_data.dsize / sizeof(struct server_id),
 436                         state->private_data);
 437         return res;
 438 }
 439
 440 void dbwrap_watchers_traverse_read(
 441         int (*fn)(const uint8_t *db_id, size_t db_id_len, const TDB_DATA key,
 442                   const struct server_id *watchers, size_t num_watchers,
 443                   void *private_data),
 444         void *private_data)
 445 {
 446         struct dbwrap_watchers_traverse_read_state state;
 447         struct db_context *db;
 448
 449         db = dbwrap_record_watchers_db();
 450         if (db == NULL) {
 451                 return;
 452         }
 453         state.fn = fn;
 454         state.private_data = private_data;
 455         dbwrap_traverse_read(db, dbwrap_watchers_traverse_read_callback,
 456                              &state, NULL);
 457 }
 458
 459 static int dbwrap_wakeall_cb(const uint8_t *db_id, size_t db_id_len,
 460                              const TDB_DATA key,
 461                              const struct server_id *watchers,
 462                              size_t num_watchers,
 463                              void *private_data)
 464 {
 465         struct messaging_context *msg = talloc_get_type_abort(
 466                 private_data, struct messaging_context);
 467         uint32_t i;
 468         DATA_BLOB blob;
 469
 470         blob.data = key.dptr;
 471         blob.length = key.dsize;
 472
 473         for (i=0; i<num_watchers; i++) {
 474                 messaging_send(msg, watchers[i], MSG_DBWRAP_MODIFIED, &blob);
 475         }
 476         return 0;
 477 }
 478
 479 void dbwrap_watchers_wakeall(struct messaging_context *msg)
 480 {
 481         dbwrap_watchers_traverse_read(dbwrap_wakeall_cb, msg);
 482 }