etc/afpd/fce_api.c

   1 /*
   2  * Copyright (c) 2010 Mark Williams
   3  *
   4  * File change event API for netatalk
   5  *
   6  * for every detected filesystem change a UDP packet is sent to an arbitrary list
   7  * of listeners. Each packet contains unix path of modified filesystem element,
   8  * event reason, and a consecutive event id (32 bit). Technically we are UDP client and are sending
   9  * out packets synchronuosly as they are created by the afp functions. This should not affect
  10  * performance measurably. The only delaying calls occur during initialization, if we have to
  11  * resolve non-IP hostnames to IP. All numeric data inside the packet is network byte order, so use
  12  * ntohs / ntohl to resolve length and event id. Ideally a listener receives every packet with
  13  * no gaps in event ids, starting with event id 1 and mode FCE_CONN_START followed by
  14  * data events from id 2 up to 0xFFFFFFFF, followed by 0 to 0xFFFFFFFF and so on.
  15  *
  16  * A gap or not starting with 1 mode FCE_CONN_START or receiving mode FCE_CONN_BROKEN means that
  17  * the listener has lost at least one filesystem event
  18  *
  19  * All Rights Reserved.  See COPYRIGHT.
  20  */
  21
  22 #ifdef HAVE_CONFIG_H
  23 #include "config.h"
  24 #endif /* HAVE_CONFIG_H */
  25
  26 #include <stdio.h>
  27
  28 #include <string.h>
  29 #include <stdlib.h>
  30 #include <errno.h>
  31 #include <time.h>
  32
  33
  34 #include <sys/param.h>
  35 #include <sys/socket.h>
  36 #include <netinet/in.h>
  37 #include <arpa/inet.h>
  38 #include <netdb.h>
  39
  40 #include <atalk/adouble.h>
  41 #include <atalk/vfs.h>
  42 #include <atalk/logger.h>
  43 #include <atalk/afp.h>
  44 #include <atalk/util.h>
  45 #include <atalk/cnid.h>
  46 #include <atalk/unix.h>
  47 #include <atalk/fce_api.h>
  48 #include <atalk/globals.h>
  49
  50 #include "fork.h"
  51 #include "file.h"
  52 #include "directory.h"
  53 #include "desktop.h"
  54 #include "volume.h"
  55
  56 // ONLY USED IN THIS FILE
  57 #include "fce_api_internal.h"
  58
  59 #define FCE_TRUE 1
  60 #define FCE_FALSE 0
  61
  62 /* We store our connection data here */
  63 static struct udp_entry udp_socket_list[FCE_MAX_UDP_SOCKS];
  64 static int udp_sockets = 0;
  65 static int udp_initialized = FCE_FALSE;
  66 static unsigned long fce_ev_enabled =
  67     (1 << FCE_FILE_MODIFY) |
  68     (1 << FCE_FILE_DELETE) |
  69     (1 << FCE_DIR_DELETE) |
  70     (1 << FCE_FILE_CREATE) |
  71     (1 << FCE_DIR_CREATE);
  72
  73 #define MAXIOBUF 1024
  74 static char iobuf[MAXIOBUF];
  75 static const char *skip_files[] =
  76 {
  77         ".DS_Store",
  78         NULL
  79 };
  80 static struct fce_close_event last_close_event;
  81
  82 static char *fce_event_names[] = {
  83     "",
  84     "FCE_FILE_MODIFY",
  85     "FCE_FILE_DELETE",
  86     "FCE_DIR_DELETE",
  87     "FCE_FILE_CREATE",
  88     "FCE_DIR_CREATE"
  89 };
  90
  91 /*
  92  *
  93  * Initialize network structs for any listeners
  94  * We dont give return code because all errors are handled internally (I hope..)
  95  *
  96  * */
  97 void fce_init_udp()
  98 {
  99     int rv;
 100     struct addrinfo hints, *servinfo, *p;
 101
 102     if (udp_initialized == FCE_TRUE)
 103         return;
 104
 105     memset(&hints, 0, sizeof hints);
 106     hints.ai_family = AF_UNSPEC;
 107     hints.ai_socktype = SOCK_DGRAM;
 108
 109     for (int i = 0; i < udp_sockets; i++) {
 110         struct udp_entry *udp_entry = udp_socket_list + i;
 111
 112         /* Close any pending sockets */
 113         if (udp_entry->sock != -1)
 114             close(udp_entry->sock);
 115
 116         if ((rv = getaddrinfo(udp_entry->addr, udp_entry->port, &hints, &servinfo)) != 0) {
 117             LOG(log_error, logtype_fce, "fce_init_udp: getaddrinfo(%s:%s): %s",
 118                 udp_entry->addr, udp_entry->port, gai_strerror(rv));
 119             continue;
 120         }
 121
 122         /* loop through all the results and make a socket */
 123         for (p = servinfo; p != NULL; p = p->ai_next) {
 124             if ((udp_entry->sock = socket(p->ai_family, p->ai_socktype, p->ai_protocol)) == -1) {
 125                 LOG(log_error, logtype_fce, "fce_init_udp: socket(%s:%s): %s",
 126                     udp_entry->addr, udp_entry->port, strerror(errno));
 127                 continue;
 128             }
 129             break;
 130         }
 131
 132         if (p == NULL) {
 133             LOG(log_error, logtype_fce, "fce_init_udp: no socket for %s:%s",
 134                 udp_entry->addr, udp_entry->port);
 135         }
 136         udp_entry->addrinfo = *p;
 137         memcpy(&udp_entry->addrinfo, p, sizeof(struct addrinfo));
 138         memcpy(&udp_entry->sockaddr, p->ai_addr, sizeof(struct sockaddr_storage));
 139         freeaddrinfo(servinfo);
 140     }
 141
 142     udp_initialized = FCE_TRUE;
 143 }
 144
 145 void fce_cleanup()
 146 {
 147     if (udp_initialized == FCE_FALSE )
 148         return;
 149
 150     for (int i = 0; i < udp_sockets; i++)
 151     {
 152         struct udp_entry *udp_entry = udp_socket_list + i;
 153
 154         /* Close any pending sockets */
 155         if (udp_entry->sock != -1)
 156         {
 157             close( udp_entry->sock );
 158             udp_entry->sock = -1;
 159         }
 160     }
 161     udp_initialized = FCE_FALSE;
 162 }
 163
 164 /*
 165  * Construct a UDP packet for our listeners and return packet size
 166  * */
 167 static ssize_t build_fce_packet( struct fce_packet *packet, char *path, int mode, uint32_t event_id )
 168 {
 169     size_t pathlen = 0;
 170     ssize_t data_len = 0;
 171     uint64_t *t;
 172
 173     /* Set content of packet */
 174     memcpy(packet->magic, FCE_PACKET_MAGIC, sizeof(packet->magic) );
 175     packet->version = FCE_PACKET_VERSION;
 176     packet->mode = mode;
 177
 178     packet->event_id = event_id;
 179
 180     pathlen = strlen(path); /* exclude string terminator */
 181
 182     /* This should never happen, but before we bust this server, we send nonsense, fce listener has to cope */
 183     if (pathlen >= MAXPATHLEN)
 184         pathlen = MAXPATHLEN - 1;
 185
 186     packet->datalen = pathlen;
 187
 188     /* This is the payload len. Means: the packet has len bytes more until packet is finished */
 189     data_len = FCE_PACKET_HEADER_SIZE + pathlen;
 190
 191     memcpy(packet->data, path, pathlen);
 192
 193     /* return the packet len */
 194     return data_len;
 195 }
 196
 197 /*
 198  * Handle Endianess and write into buffer w/o padding
 199  **/
 200 static void pack_fce_packet(struct fce_packet *packet, unsigned char *buf, int maxlen)
 201 {
 202     unsigned char *p = buf;
 203
 204     memcpy(p, &packet->magic[0], sizeof(packet->magic));
 205     p += sizeof(packet->magic);
 206
 207     *p = packet->version;
 208     p++;
 209
 210     *p = packet->mode;
 211     p++;
 212
 213     uint32_t *id = (uint32_t*)p;
 214     *id = htonl(packet->event_id);
 215     p += sizeof(packet->event_id);
 216
 217     uint16_t *l = ( uint16_t *)p;
 218     *l = htons(packet->datalen);
 219     p += sizeof(packet->datalen);
 220
 221     if (((p - buf) +  packet->datalen) < maxlen) {
 222         memcpy(p, &packet->data[0], packet->datalen);
 223     }
 224 }
 225
 226 /*
 227  * Send the fce information to all (connected) listeners
 228  * We dont give return code because all errors are handled internally (I hope..)
 229  * */
 230 static void send_fce_event( char *path, int mode )
 231 {
 232     static int first_event = FCE_TRUE;
 233
 234     struct fce_packet packet;
 235     void *data = &packet;
 236     static uint32_t event_id = 0; /* the unique packet couter to detect packet/data loss. Going from 0xFFFFFFFF to 0x0 is a valid increment */
 237     time_t now = time(NULL);
 238
 239     LOG(log_debug, logtype_fce, "send_fce_event: start");
 240
 241     /* initialized ? */
 242     if (first_event == FCE_TRUE) {
 243         first_event = FCE_FALSE;
 244         fce_init_udp();
 245         /* Notify listeners the we start from the beginning */
 246         send_fce_event( "", FCE_CONN_START );
 247     }
 248
 249     /* build our data packet */
 250     ssize_t data_len = build_fce_packet( &packet, path, mode, ++event_id );
 251     pack_fce_packet(&packet, iobuf, MAXIOBUF);
 252
 253     for (int i = 0; i < udp_sockets; i++)
 254     {
 255         int sent_data = 0;
 256         struct udp_entry *udp_entry = udp_socket_list + i;
 257
 258         /* we had a problem earlier ? */
 259         if (udp_entry->sock == -1)
 260         {
 261             /* We still have to wait ?*/
 262             if (now < udp_entry->next_try_on_error)
 263                 continue;
 264
 265             /* Reopen socket */
 266             udp_entry->sock = socket(udp_entry->addrinfo.ai_family,
 267                                      udp_entry->addrinfo.ai_socktype,
 268                                      udp_entry->addrinfo.ai_protocol);
 269
 270             if (udp_entry->sock == -1) {
 271                 /* failed again, so go to rest again */
 272                 LOG(log_error, logtype_fce, "Cannot recreate socket for fce UDP connection: errno %d", errno  );
 273
 274                 udp_entry->next_try_on_error = now + FCE_SOCKET_RETRY_DELAY_S;
 275                 continue;
 276             }
 277
 278             udp_entry->next_try_on_error = 0;
 279
 280             /* Okay, we have a running socket again, send server that we had a problem on our side*/
 281             data_len = build_fce_packet( &packet, "", FCE_CONN_BROKEN, 0 );
 282             pack_fce_packet(&packet, iobuf, MAXIOBUF);
 283
 284             sendto(udp_entry->sock,
 285                    iobuf,
 286                    data_len,
 287                    0,
 288                    (struct sockaddr *)&udp_entry->sockaddr,
 289                    udp_entry->addrinfo.ai_addrlen);
 290
 291             /* Rebuild our original data packet */
 292             data_len = build_fce_packet( &packet, path, mode, event_id );
 293             pack_fce_packet(&packet, iobuf, MAXIOBUF);
 294         }
 295
 296         sent_data = sendto(udp_entry->sock,
 297                            iobuf,
 298                            data_len,
 299                            0,
 300                            (struct sockaddr *)&udp_entry->sockaddr,
 301                            udp_entry->addrinfo.ai_addrlen);
 302
 303         /* Problems ? */
 304         if (sent_data != data_len) {
 305             /* Argh, socket broke, we close and retry later */
 306             LOG(log_error, logtype_fce, "send_fce_event: error sending packet to %s:%s, transfered %d of %d: %s",
 307                 udp_entry->addr, udp_entry->port, sent_data, data_len, strerror(errno));
 308
 309             close( udp_entry->sock );
 310             udp_entry->sock = -1;
 311             udp_entry->next_try_on_error = now + FCE_SOCKET_RETRY_DELAY_S;
 312         }
 313     }
 314 }
 315
 316 static int add_udp_socket(const char *target_ip, const char *target_port )
 317 {
 318     if (target_port == NULL)
 319         target_port = FCE_DEFAULT_PORT_STRING;
 320
 321     if (udp_sockets >= FCE_MAX_UDP_SOCKS) {
 322         LOG(log_error, logtype_fce, "Too many file change api UDP connections (max %d allowed)", FCE_MAX_UDP_SOCKS );
 323         return AFPERR_PARAM;
 324     }
 325
 326     udp_socket_list[udp_sockets].addr = strdup(target_ip);
 327     udp_socket_list[udp_sockets].port = strdup(target_port);
 328     udp_socket_list[udp_sockets].sock = -1;
 329     memset(&udp_socket_list[udp_sockets].addrinfo, 0, sizeof(struct addrinfo));
 330     memset(&udp_socket_list[udp_sockets].sockaddr, 0, sizeof(struct sockaddr_storage));
 331     udp_socket_list[udp_sockets].next_try_on_error = 0;
 332
 333     udp_sockets++;
 334
 335     return AFP_OK;
 336 }
 337
 338 static void save_close_event(const char *path)
 339 {
 340     time_t now = time(NULL);
 341
 342     /* Check if it's a close for the same event as the last one */
 343     if (last_close_event.time   /* is there any saved event ? */
 344         && (strcmp(path, last_close_event.path) != 0)) {
 345         /* no, so send the saved event out now */
 346         send_fce_event(last_close_event.path, FCE_FILE_MODIFY);
 347     }
 348
 349     LOG(log_debug, logtype_fce, "save_close_event: %s", path);
 350
 351     last_close_event.time = now;
 352     strncpy(last_close_event.path, path, MAXPATHLEN);
 353 }
 354
 355 /*
 356  *
 357  * Dispatcher for all incoming file change events
 358  *
 359  * */
 360 static int register_fce(const char *u_name, int is_dir, int mode)
 361 {
 362     static int first_event = FCE_TRUE;
 363
 364     AFP_ASSERT(mode >= FCE_FIRST_EVENT && mode <= FCE_LAST_EVENT);
 365
 366     LOG(log_debug, logtype_fce, "register_fce(path: %s, type: %s, event: %s",
 367         fullpathname(u_name), is_dir ? "dir" : "file", fce_event_names[mode]);
 368
 369     if (udp_sockets == 0)
 370         /* No listeners configured */
 371         return AFP_OK;
 372
 373     if (u_name == NULL)
 374         return AFPERR_PARAM;
 375
 376         /* do some initialization on the fly the first time */
 377         if (first_event) {
 378                 fce_initialize_history();
 379         first_event = FCE_FALSE;
 380         }
 381
 382         /* handle files which should not cause events (.DS_Store atc. ) */
 383         for (int i = 0; skip_files[i] != NULL; i++)
 384         {
 385                 if (!strcmp( u_name, skip_files[i]))
 386                         return AFP_OK;
 387         }
 388
 389
 390     /* FIXME: use fullpathname() for path? */
 391         char full_path_buffer[MAXPATHLEN + 1] = {""};
 392         const char *cwd = getcwdpath();
 393
 394     if (!is_dir || mode == FCE_DIR_DELETE) {
 395                 if (strlen( cwd ) + strlen( u_name) + 1 >= MAXPATHLEN) {
 396                         LOG(log_error, logtype_fce, "FCE file name too long: %s/%s", cwd, u_name );
 397                         return AFPERR_PARAM;
 398                 }
 399                 sprintf( full_path_buffer, "%s/%s", cwd, u_name );
 400         } else {
 401                 if (strlen( cwd ) >= MAXPATHLEN) {
 402                         LOG(log_error, logtype_fce, "FCE directory name too long: %s", cwd);
 403                         return AFPERR_PARAM;
 404                 }
 405                 strcpy( full_path_buffer, cwd);
 406         }
 407
 408         /* Can we ignore this event based on type or history? */
 409         if (fce_handle_coalescation(full_path_buffer, is_dir, mode)) {
 410                 LOG(log_debug9, logtype_fce, "Coalesced fc event <%d> for <%s>", mode, full_path_buffer );
 411                 return AFP_OK;
 412         }
 413
 414     if (mode & FCE_FILE_MODIFY) {
 415         save_close_event(full_path_buffer);
 416         return AFP_OK;
 417     }
 418
 419     send_fce_event( full_path_buffer, mode );
 420
 421     return AFP_OK;
 422 }
 423
 424 static void check_saved_close_events(int fmodwait)
 425 {
 426     time_t now = time(NULL);
 427
 428     /* check if configured holdclose time has passed */
 429     if (last_close_event.time && ((last_close_event.time + fmodwait) < now)) {
 430         LOG(log_debug, logtype_fce, "check_saved_close_events: sending event: %s", last_close_event.path);
 431         /* yes, send event */
 432         send_fce_event(&last_close_event.path[0], FCE_FILE_MODIFY);
 433         last_close_event.path[0] = 0;
 434         last_close_event.time = 0;
 435     }
 436 }
 437
 438 /******************** External calls start here **************************/
 439
 440 /*
 441  * API-Calls for file change api, called form outside (file.c directory.c ofork.c filedir.c)
 442  * */
 443 #ifndef FCE_TEST_MAIN
 444
 445 void fce_pending_events(AFPObj *obj)
 446 {
 447     check_saved_close_events(obj->options.fce_fmodwait);
 448 }
 449
 450 int fce_register_delete_file( struct path *path )
 451 {
 452     int ret = AFP_OK;
 453
 454     if (path == NULL)
 455         return AFPERR_PARAM;
 456
 457     if (!(fce_ev_enabled & (1 << FCE_FILE_DELETE)))
 458         return ret;
 459
 460     ret = register_fce( path->u_name, false, FCE_FILE_DELETE );
 461
 462     return ret;
 463 }
 464 int fce_register_delete_dir( char *name )
 465 {
 466     int ret = AFP_OK;
 467
 468     if (name == NULL)
 469         return AFPERR_PARAM;
 470
 471     if (!(fce_ev_enabled & (1 << FCE_DIR_DELETE)))
 472         return ret;
 473
 474     ret = register_fce( name, true, FCE_DIR_DELETE);
 475
 476     return ret;
 477 }
 478
 479 int fce_register_new_dir( struct path *path )
 480 {
 481     int ret = AFP_OK;
 482
 483     if (path == NULL)
 484         return AFPERR_PARAM;
 485
 486     if (!(fce_ev_enabled & (1 << FCE_DIR_CREATE)))
 487         return ret;
 488
 489     ret = register_fce( path->u_name, true, FCE_DIR_CREATE );
 490
 491     return ret;
 492 }
 493
 494
 495 int fce_register_new_file( struct path *path )
 496 {
 497     int ret = AFP_OK;
 498
 499     if (path == NULL)
 500         return AFPERR_PARAM;
 501
 502     if (!(fce_ev_enabled & (1 << FCE_FILE_CREATE)))
 503         return ret;
 504
 505     ret = register_fce( path->u_name, false, FCE_FILE_CREATE );
 506
 507     return ret;
 508 }
 509
 510 int fce_register_file_modification( struct ofork *ofork )
 511 {
 512     int ret = AFP_OK;
 513
 514     if (ofork == NULL)
 515         return AFPERR_PARAM;
 516
 517     if (!(fce_ev_enabled & (1 << FCE_FILE_MODIFY)))
 518         return ret;
 519
 520     ret = register_fce(of_name(ofork), false, FCE_FILE_MODIFY );
 521
 522     return ret;
 523 }
 524 #endif
 525
 526 /*
 527  *
 528  * Extern connect to afpd parameter, can be called multiple times for multiple listeners (up to MAX_UDP_SOCKS times)
 529  *
 530  * */
 531 int fce_add_udp_socket(const char *target)
 532 {
 533         const char *port = FCE_DEFAULT_PORT_STRING;
 534         char target_ip[256] = {""};
 535
 536         strncpy(target_ip, target, sizeof(target_ip) -1);
 537
 538         char *port_delim = strchr( target_ip, ':' );
 539         if (port_delim) {
 540                 *port_delim = 0;
 541                 port = port_delim + 1;
 542         }
 543         return add_udp_socket(target_ip, port);
 544 }
 545
 546 int fce_set_events(const char *events)
 547 {
 548     char *e;
 549     char *p;
 550
 551     if (events == NULL)
 552         return AFPERR_PARAM;
 553
 554     e = strdup(events);
 555
 556     fce_ev_enabled = 0;
 557
 558     for (p = strtok(e, ", "); p; p = strtok(NULL, ", ")) {
 559         if (strcmp(p, "fmod") == 0) {
 560             fce_ev_enabled |= (1 << FCE_FILE_MODIFY);
 561         } else if (strcmp(p, "fdel") == 0) {
 562             fce_ev_enabled |= (1 << FCE_FILE_DELETE);
 563         } else if (strcmp(p, "ddel") == 0) {
 564             fce_ev_enabled |= (1 << FCE_DIR_DELETE);
 565         } else if (strcmp(p, "fcre") == 0) {
 566             fce_ev_enabled |= (1 << FCE_FILE_CREATE);
 567         } else if (strcmp(p, "dcre") == 0) {
 568             fce_ev_enabled |= (1 << FCE_DIR_CREATE);
 569         }
 570     }
 571
 572     free(e);
 573
 574     return AFP_OK;
 575 }
 576
 577 #ifdef FCE_TEST_MAIN
 578
 579
 580 void shortsleep( unsigned int us )
 581 {
 582     usleep( us );
 583 }
 584 int main( int argc, char*argv[] )
 585 {
 586     int c,ret;
 587
 588     char *port = FCE_DEFAULT_PORT_STRING;
 589     char *host = "localhost";
 590     int delay_between_events = 1000;
 591     int event_code = FCE_FILE_MODIFY;
 592     char pathbuff[1024];
 593     int duration_in_seconds = 0; // TILL ETERNITY
 594     char target[256];
 595     char *path = getcwd( pathbuff, sizeof(pathbuff) );
 596
 597     // FULLSPEED TEST IS "-s 1001" -> delay is 0 -> send packets without pause
 598
 599     while ((c = getopt(argc, argv, "d:e:h:p:P:s:")) != -1) {
 600         switch(c) {
 601         case '?':
 602             fprintf(stdout, "%s: [ -p Port -h Listener1 [ -h Listener2 ...] -P path -s Delay_between_events_in_us -e event_code -d Duration ]\n", argv[0]);
 603             exit(1);
 604             break;
 605         case 'd':
 606             duration_in_seconds = atoi(optarg);
 607             break;
 608         case 'e':
 609             event_code = atoi(optarg);
 610             break;
 611         case 'h':
 612             host = strdup(optarg);
 613             break;
 614         case 'p':
 615             port = strdup(optarg);
 616             break;
 617         case 'P':
 618             path = strdup(optarg);
 619             break;
 620         case 's':
 621             delay_between_events = atoi(optarg);
 622             break;
 623         }
 624     }
 625
 626     sprintf(target, "%s:%s", host, port);
 627     if (fce_add_udp_socket(target) != 0)
 628         return 1;
 629
 630     int ev_cnt = 0;
 631     time_t start_time = time(NULL);
 632     time_t end_time = 0;
 633
 634     if (duration_in_seconds)
 635         end_time = start_time + duration_in_seconds;
 636
 637     while (1)
 638     {
 639         time_t now = time(NULL);
 640         if (now > start_time)
 641         {
 642             start_time = now;
 643             fprintf( stdout, "%d events/s\n", ev_cnt );
 644             ev_cnt = 0;
 645         }
 646         if (end_time && now >= end_time)
 647             break;
 648
 649         register_fce( path, 0, event_code );
 650         ev_cnt++;
 651
 652
 653         shortsleep( delay_between_events );
 654     }
 655 }
 656 #endif /* TESTMAIN*/