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