audisp/plugins/remote/queue.c

   1 /* queue.c - a string queue implementation
   2  * Copyright 2009, 2011 Red Hat Inc., Durham, North Carolina.
   3  * All Rights Reserved.
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Lesser General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2.1 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Lesser General Public
  16  * License along with this library; if not, write to the Free Software
  17  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  18  *
  19  * Authors:
  20  *      Steve Grubb <sgrubb@redhat.com>
  21  *      Miloslav Trmač <mitr@redhat.com>
  22  */
  23
  24 #include "config.h"
  25 #include <stdio.h>
  26 #include <arpa/inet.h>
  27 #include <errno.h>
  28 #include <fcntl.h>
  29 #include <limits.h>
  30 #include <stddef.h>
  31 #include <stdint.h>
  32 #include <stdlib.h>
  33 #include <string.h>
  34 #include <sys/stat.h>
  35 #include <unistd.h>
  36 #include "queue.h"
  37
  38 struct queue
  39 {
  40         int flags;              /* Q_* */
  41         int fd;                 /* -1 if !Q_IN_FILE */
  42         /* NULL if !Q_IN_MEMORY.  [i] contains a memory copy of the queue entry
  43            "i", if known - it may be NULL even if entry exists. */
  44         unsigned char **memory;
  45         size_t num_entries;
  46         size_t entry_size;
  47         size_t queue_head;
  48         size_t queue_length;
  49         unsigned char buffer[]; /* Used only locally within q_peek() */
  50 };
  51
  52 /* Infrastructure */
  53
  54 /* Compile-time expression verification */
  55 #define verify(E) do {                          \
  56                 char verify__[(E) ? 1 : -1];    \
  57                 (void)verify__;                 \
  58         } while (0)
  59
  60 /* Like pread(), except that it handles partial reads, and returns 0 on
  61    success. */
  62 static int full_pread(int fd, void *buf, size_t size, off_t offset)
  63 {
  64         while (size != 0) {
  65                 ssize_t run, res;
  66
  67                 if (size > SSIZE_MAX)
  68                         run = SSIZE_MAX;
  69                 else
  70                         run = size;
  71                 res = pread(fd, buf, run, offset);
  72                 if (res < 0)
  73                         return -1;
  74                 if (res == 0) {
  75                         errno = ENXIO; /* Any better value? */
  76                         return -1;
  77                 }
  78                 buf = (unsigned char *)buf + res;
  79                 size -= res;
  80                 offset += res;
  81         }
  82         return 0;
  83 }
  84
  85 /* Like pwrite(), except that it handles partial writes, and returns 0 on
  86    success. */
  87 static int full_pwrite(int fd, const void *buf, size_t size, off_t offset)
  88 {
  89         while (size != 0) {
  90                 ssize_t run, res;
  91
  92                 if (size > SSIZE_MAX)
  93                         run = SSIZE_MAX;
  94                 else
  95                         run = size;
  96                 res = pwrite(fd, buf, run, offset);
  97                 if (res < 0)
  98                         return -1;
  99                 if (res == 0) {
 100                         errno = ENXIO; /* Any better value? */
 101                         return -1;
 102                 }
 103                 buf = (const unsigned char *)buf + res;
 104                 size -= res;
 105                 offset += res;
 106         }
 107         return 0;
 108 }
 109
 110 /* File format and utilities */
 111
 112 /* The mutable part of struct file_header */
 113 struct fh_state {
 114         uint32_t queue_head;    /* 0-based index of the first non-empty entry */
 115         uint32_t queue_length;  /* [0, num_entries] */
 116 };
 117
 118 /* All integer values are in network byte order (big endian) */
 119 struct file_header
 120 {
 121         uint8_t magic[14];      /* See fh_magic below */
 122         uint8_t version;        /* File format version, see FH_VERSION* below */
 123         uint8_t reserved;       /* Must be 0 */
 124         /* Total file size is (num_entries + 1) * entry_size.  This must fit
 125            into SIZE_MAX because the "len" parameter of posix_fallocate has
 126            a size_t type. */
 127         uint32_t num_entries;   /* Total number of entries allocated */
 128         uint32_t entry_size;
 129         struct fh_state s;
 130 };
 131
 132 /* Contains a '\0' byte to unambiguously mark the file as a binary file. */
 133 static const uint8_t fh_magic[14] = "\0audisp-remote";
 134 #define FH_VERSION_0 0x00
 135
 136 /* Return file position for ENTRY in Q */
 137 static size_t entry_offset (const struct queue *q, size_t entry)
 138 {
 139         return (entry + 1) * q->entry_size;
 140 }
 141
 142 /* Synchronize Q if required and return 0.
 143    On error, return -1 and set errno. */
 144 static int q_sync(struct queue *q)
 145 {
 146         if ((q->flags & Q_SYNC) == 0)
 147                 return 0;
 148         return fdatasync(q->fd);
 149 }
 150
 151 /* Sync file's fh_state with Q, q_sync (Q), and return 0.
 152    On error, return -1 and set errno. */
 153 static int sync_fh_state (struct queue *q)
 154 {
 155         struct fh_state s;
 156
 157         if (q->fd == -1)
 158                 return 0;
 159
 160         s.queue_head = htonl(q->queue_head);
 161         s.queue_length = htonl(q->queue_length);
 162         if (full_pwrite(q->fd, &s, sizeof(s), offsetof(struct file_header, s))
 163             != 0)
 164                 return -1;
 165         return q_sync(q);
 166 }
 167
 168 /* Queue implementation */
 169
 170 /* Open PATH for Q, update Q from it, and return 0.
 171    On error, return -1 and set errno; Q->fd may be set even on error. */
 172 static int q_open_file(struct queue *q, const char *path)
 173 {
 174         int open_flags, fd_flags;
 175         struct stat st;
 176         struct file_header fh;
 177
 178         open_flags = O_RDWR;
 179         if ((q->flags & Q_CREAT) != 0)
 180                 open_flags |= O_CREAT;
 181         if ((q->flags & Q_EXCL) != 0)
 182                 open_flags |= O_EXCL;
 183         q->fd = open(path, open_flags, S_IRUSR | S_IWUSR);
 184         if (q->fd == -1)
 185                 return -1;
 186
 187         fd_flags = fcntl(q->fd, F_GETFD);
 188         if (fd_flags < 0)
 189                 return -1;
 190         if (fcntl(q->fd, F_SETFD, fd_flags | FD_CLOEXEC) == -1)
 191                 return -1;
 192
 193         /* File locking in POSIX is pretty much broken... let's hope nobody
 194            attempts to open a single file twice within the same process.
 195            open() above has initialized the file offset to 0, so the lockf()
 196            below affects the whole file. */
 197         if (lockf(q->fd, F_TLOCK, 0) != 0) {
 198                 if (errno == EACCES || errno == EAGAIN)
 199                         errno = EBUSY; /* This makes more sense... */
 200                 return -1;
 201         }
 202
 203         if (fstat(q->fd, &st) != 0)
 204                 return -1;
 205         if (st.st_size == 0) {
 206                 verify(sizeof(fh.magic) == sizeof(fh_magic));
 207                 memcpy(fh.magic, fh_magic, sizeof(fh.magic));
 208                 fh.version = FH_VERSION_0;
 209                 fh.reserved = 0;
 210                 fh.num_entries = htonl(q->num_entries);
 211                 fh.entry_size = htonl(q->entry_size);
 212                 fh.s.queue_head = htonl(0);
 213                 fh.s.queue_length = htonl(0);
 214                 if (full_pwrite(q->fd, &fh, sizeof(fh), 0) != 0)
 215                         return -1;
 216                 if (q_sync(q) != 0)
 217                         return -1;
 218 #ifdef HAVE_POSIX_FALLOCATE
 219                 if (posix_fallocate(q->fd, 0,
 220                                     (q->num_entries + 1) * q->entry_size) != 0)
 221                         return -1;
 222 #endif
 223         } else {
 224                 uint32_t file_entries;
 225                 if (full_pread(q->fd, &fh, sizeof(fh), 0) != 0)
 226                         return -1;
 227                 if (memcmp(fh.magic, fh_magic, sizeof(fh.magic)) != 0
 228                     || fh.version != FH_VERSION_0 || fh.reserved != 0
 229                     || fh.entry_size != htonl(q->entry_size)) {
 230                         errno = EINVAL;
 231                         return -1;
 232                 }
 233                 file_entries = ntohl(fh.num_entries);
 234                 if (file_entries > SIZE_MAX / q->entry_size - 1
 235                     || ((uintmax_t)st.st_size
 236                         != (file_entries + 1) * q->entry_size)) {
 237                         errno = EINVAL;
 238                         return -1;
 239                 }
 240         }
 241         /* Note that this may change q->num_entries! */
 242         q->num_entries = ntohl(fh.num_entries);
 243         q->queue_head = ntohl(fh.s.queue_head);
 244         q->queue_length = ntohl(fh.s.queue_length);
 245         if (q->queue_head >= q->num_entries
 246             || q->queue_length > q->num_entries) {
 247                 errno = EINVAL;
 248                 return -1;
 249         }
 250         return 0;
 251 }
 252
 253 /* Like q_open(), but does not handle Q_RESIZE, and NUM_ENTRIES is only used
 254    when creating a new file. */
 255 static struct queue *q_open_no_resize(int q_flags, const char *path,
 256                                       size_t num_entries, size_t entry_size)
 257 {
 258         struct queue *q;
 259         int saved_errno;
 260
 261         if ((q_flags & (Q_IN_MEMORY | Q_IN_FILE)) == 0) {
 262                 errno = EINVAL;
 263                 return NULL;
 264         }
 265         if (num_entries == 0 || num_entries > UINT32_MAX
 266             || entry_size < 1 /* for trailing NUL */
 267             || entry_size < sizeof(struct file_header) /* for Q_IN_FILE */
 268             /* to allocate "struct queue" including its buffer*/
 269             || entry_size > UINT32_MAX - sizeof(struct queue)) {
 270                 errno = EINVAL;
 271                 return NULL;
 272         }
 273         if (entry_size > SIZE_MAX
 274             || num_entries > SIZE_MAX / entry_size - 1 /* for Q_IN_FILE */
 275             || num_entries > SIZE_MAX / sizeof(*q->memory)) {
 276                 errno = EINVAL;
 277                 return NULL;
 278         }
 279
 280         q = malloc(sizeof(*q) + entry_size);
 281         if (q == NULL)
 282                 return NULL;
 283         q->flags = q_flags;
 284         q->fd = -1;
 285         q->memory = NULL;
 286         q->num_entries = num_entries;
 287         q->entry_size = entry_size;
 288         q->queue_head = 0;
 289         q->queue_length = 0;
 290
 291         if ((q_flags & Q_IN_MEMORY) != 0) {
 292                 size_t sz = num_entries * sizeof(*q->memory);
 293
 294                 q->memory = malloc(sz);
 295                 if (q->memory == NULL)
 296                         goto err;
 297                 memset(q->memory, 0, sz);
 298         }
 299
 300         if ((q_flags & Q_IN_FILE) != 0 && q_open_file(q, path) != 0)
 301                 goto err;
 302
 303         return q;
 304
 305 err:
 306         saved_errno = errno;
 307         if (q->fd != -1)
 308                 close(q->fd);
 309         free(q->memory);
 310         free(q);
 311         errno = saved_errno;
 312         return NULL;
 313 }
 314
 315 void q_close(struct queue *q)
 316 {
 317         if (q->fd != -1)
 318                 close(q->fd); /* Also releases the file lock */
 319         if (q->memory != NULL) {
 320                 size_t i;
 321
 322                 for (i = 0; i < q->num_entries; i++)
 323                         free(q->memory[i]);
 324                 free(q->memory);
 325         }
 326         free(q);
 327 }
 328
 329 /* Internal use only: add DATA to Q, but don't update fh_state. */
 330 static int q_append_no_sync_fh_state(struct queue *q, const char *data)
 331 {
 332         size_t data_size, entry_index;
 333         unsigned char *copy;
 334
 335         if (q->queue_length == q->num_entries) {
 336                 errno = ENOSPC;
 337                 return -1;
 338         }
 339
 340         data_size = strlen(data) + 1;
 341         if (data_size > q->entry_size) {
 342                 errno = EINVAL;
 343                 return -1;
 344         }
 345
 346         entry_index = (q->queue_head + q->queue_length) % q->num_entries;
 347         if (q->memory != NULL) {
 348                 if (q->memory[entry_index] != NULL) {
 349                         errno = EIO; /* This is _really_ unexpected. */
 350                         return -1;
 351                 }
 352                 copy = malloc(data_size);
 353                 if (copy == NULL)
 354                         return -1;
 355                 memcpy(copy, data, data_size);
 356         } else
 357                 copy = NULL;
 358
 359         if (q->fd != -1) {
 360                 size_t offset;
 361
 362                 offset = entry_offset(q, entry_index);
 363                 if (full_pwrite(q->fd, data, data_size, offset) != 0) {
 364                         int saved_errno;
 365
 366                         saved_errno = errno;
 367                         if (copy != NULL)
 368                                 free(copy);
 369                         errno = saved_errno;
 370                         return -1;
 371                 }
 372         }
 373
 374         if (copy != NULL)
 375                 q->memory[entry_index] = copy;
 376
 377         q->queue_length++;
 378
 379         return 0;
 380 }
 381
 382 int q_append(struct queue *q, const char *data)
 383 {
 384         int r;
 385
 386         r = q_append_no_sync_fh_state(q, data);
 387         if (r != 0)
 388                 return r;
 389
 390         return sync_fh_state(q); /* Calls q_sync() */
 391 }
 392
 393 int q_peek(struct queue *q, char *buf, size_t size)
 394 {
 395         const unsigned char *data;
 396         size_t data_size;
 397
 398         if (q->queue_length == 0)
 399                 return 0;
 400
 401         if (q->memory != NULL && q->memory[q->queue_head] != NULL) {
 402                 data = q->memory[q->queue_head];
 403                 data_size = strlen((char *)data) + 1;
 404         } else if (q->fd != -1) {
 405                 const unsigned char *end;
 406
 407                 if (full_pread(q->fd, q->buffer, q->entry_size,
 408                                entry_offset(q, q->queue_head)) != 0)
 409                         return -1;
 410                 data = q->buffer;
 411                 end = memchr(q->buffer, '\0', q->entry_size);
 412                 if (end == NULL) {
 413                         /* FIXME: silently drop this entry? */
 414                         errno = EBADMSG;
 415                         return -1;
 416                 }
 417                 data_size = (end - data) + 1;
 418
 419                 if (q->memory != NULL) {
 420                         unsigned char *copy;
 421
 422                         copy = malloc(data_size);
 423                         if (copy != NULL) { /* Silently ignore failures. */
 424                                 memcpy(copy, data, data_size);
 425                                 q->memory[q->queue_head] = copy;
 426                         }
 427                 }
 428         } else {
 429                 errno = EIO; /* This is _really_ unexpected. */
 430                 return -1;
 431         }
 432
 433         if (size < data_size) {
 434                 errno = ERANGE;
 435                 return -1;
 436         }
 437         memcpy(buf, data, data_size);
 438         return data_size;
 439 }
 440
 441 /* Internal use only: drop head of Q, but don't write this into the file */
 442 static int q_drop_head_memory_only(struct queue *q)
 443 {
 444         if (q->queue_length == 0) {
 445                 errno = EINVAL;
 446                 return -1;
 447         }
 448
 449         if (q->memory != NULL) {
 450                 free(q->memory[q->queue_head]);
 451                 q->memory[q->queue_head] = NULL;
 452         }
 453
 454         q->queue_head++;
 455         if (q->queue_head == q->num_entries)
 456                 q->queue_head = 0;
 457         q->queue_length--;
 458         return 0;
 459 }
 460
 461 int q_drop_head(struct queue *q)
 462 {
 463         int r;
 464
 465         r = q_drop_head_memory_only(q);
 466         if (r != 0)
 467                 return r;
 468
 469         return sync_fh_state(q); /* Calls q_sync() */
 470 }
 471
 472 size_t q_queue_length(const struct queue *q)
 473 {
 474         return q->queue_length;
 475 }
 476
 477 struct queue *q_open(int q_flags, const char *path, size_t num_entries,
 478                      size_t entry_size)
 479 {
 480         struct queue *q, *q2;
 481         char *tmp_path, *buf;
 482         size_t path_len;
 483         int saved_errno, fd;
 484
 485         q = q_open_no_resize(q_flags, path, num_entries, entry_size);
 486         if (q == NULL || q->num_entries == num_entries)
 487                 return q;
 488
 489         if ((q->flags & Q_RESIZE) == 0) {
 490                 saved_errno = EINVAL;
 491                 goto err_errno_q;
 492         }
 493
 494         if (q->queue_length > num_entries) {
 495                 saved_errno = ENOSPC;
 496                 goto err_errno_q;
 497         }
 498
 499         buf = malloc(entry_size);
 500         if (buf == NULL) {
 501                 saved_errno = errno;
 502                 goto err_errno_q;
 503         }
 504
 505         path_len = strlen(path);
 506         tmp_path = malloc(path_len + 7);
 507         if (tmp_path == NULL) {
 508                 saved_errno = errno;
 509                 goto err_errno_buf;
 510         }
 511         memcpy(tmp_path, path, path_len);
 512         memcpy(tmp_path + path_len, "XXXXXX", 7);
 513         /* We really want tmpnam() here (safe due to the Q_EXCL below), but gcc
 514            warns on any use of tmpnam(). */
 515         fd = mkstemp(tmp_path);
 516         if (fd == -1) {
 517                 saved_errno = errno;
 518                 goto err_errno_tmp_path;
 519         }
 520         if (close(fd) != 0 || unlink(tmp_path) != 0) {
 521                 saved_errno = errno;
 522                 goto err_errno_tmp_file;
 523         }
 524
 525         q2 = q_open_no_resize(q_flags | Q_CREAT | Q_EXCL, tmp_path, num_entries,
 526                               entry_size);
 527         if (q2 == NULL) {
 528                 saved_errno = errno;
 529                 goto err_errno_tmp_file;
 530         }
 531         if (q2->num_entries != num_entries) {
 532                 errno = EIO;    /* This is _really_ unexpected. */
 533                 goto err_q2;
 534         }
 535
 536         for (;;) {
 537                 int r;
 538
 539                 r = q_peek(q, buf, entry_size);
 540                 if (r == 0)
 541                         break;
 542                 if (r < 0)
 543                         goto err_q2;
 544
 545                 if (q_append_no_sync_fh_state(q2, buf) != 0)
 546                         goto err_q2;
 547                 if (q_drop_head_memory_only(q) != 0)
 548                         goto err_q2;
 549         }
 550         if (sync_fh_state(q2) != 0)
 551                 goto err_q2;
 552
 553         if (rename(tmp_path, path) != 0)
 554                 goto err_q2;
 555
 556         q_close(q);
 557         free(buf);
 558         free(tmp_path);
 559         return q2;
 560
 561 err_q2:
 562         saved_errno = errno;
 563         q_close(q2);
 564 err_errno_tmp_file:
 565         unlink(tmp_path);
 566 err_errno_tmp_path:
 567         free(tmp_path);
 568 err_errno_buf:
 569         free(buf);
 570 err_errno_q:
 571         q_close(q);
 572         errno = saved_errno;
 573         return NULL;
 574 }