src/journal/journald.c

   1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
   2
   3 /***
   4   This file is part of systemd.
   5
   6   Copyright 2011 Lennart Poettering
   7
   8   systemd is free software; you can redistribute it and/or modify it
   9   under the terms of the GNU General Public License as published by
  10   the Free Software Foundation; either version 2 of the License, or
  11   (at your option) any later version.
  12
  13   systemd is distributed in the hope that it will be useful, but
  14   WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16   General Public License for more details.
  17
  18   You should have received a copy of the GNU General Public License
  19   along with systemd; If not, see <http://www.gnu.org/licenses/>.
  20 ***/
  21
  22 #include <sys/epoll.h>
  23 #include <sys/socket.h>
  24 #include <errno.h>
  25 #include <sys/signalfd.h>
  26 #include <unistd.h>
  27 #include <fcntl.h>
  28 #include <sys/acl.h>
  29 #include <acl/libacl.h>
  30 #include <stddef.h>
  31 #include <sys/ioctl.h>
  32 #include <linux/sockios.h>
  33 #include <sys/statvfs.h>
  34
  35 #include "hashmap.h"
  36 #include "journal-file.h"
  37 #include "sd-daemon.h"
  38 #include "socket-util.h"
  39 #include "acl-util.h"
  40 #include "cgroup-util.h"
  41 #include "list.h"
  42 #include "journal-rate-limit.h"
  43 #include "sd-journal.h"
  44 #include "journal-internal.h"
  45
  46 #define USER_JOURNALS_MAX 1024
  47 #define STDOUT_STREAMS_MAX 4096
  48
  49 #define DEFAULT_RATE_LIMIT_INTERVAL (10*USEC_PER_SEC)
  50 #define DEFAULT_RATE_LIMIT_BURST 200
  51
  52 #define RECHECK_AVAILABLE_SPACE_USEC (30*USEC_PER_SEC)
  53
  54 #define RECHECK_VAR_AVAILABLE_USEC (30*USEC_PER_SEC)
  55
  56 #define SYSLOG_TIMEOUT_USEC (5*USEC_PER_SEC)
  57
  58 typedef struct StdoutStream StdoutStream;
  59
  60 typedef struct Server {
  61         int epoll_fd;
  62         int signal_fd;
  63         int syslog_fd;
  64         int native_fd;
  65         int stdout_fd;
  66
  67         JournalFile *runtime_journal;
  68         JournalFile *system_journal;
  69         Hashmap *user_journals;
  70
  71         uint64_t seqnum;
  72
  73         char *buffer;
  74         size_t buffer_size;
  75
  76         JournalRateLimit *rate_limit;
  77
  78         JournalMetrics runtime_metrics;
  79         JournalMetrics system_metrics;
  80
  81         bool compress;
  82
  83         uint64_t cached_available_space;
  84         usec_t cached_available_space_timestamp;
  85
  86         uint64_t var_available_timestamp;
  87
  88         LIST_HEAD(StdoutStream, stdout_streams);
  89         unsigned n_stdout_streams;
  90 } Server;
  91
  92 typedef enum StdoutStreamState {
  93         STDOUT_STREAM_TAG,
  94         STDOUT_STREAM_PRIORITY,
  95         STDOUT_STREAM_PRIORITY_PREFIX,
  96         STDOUT_STREAM_TEE_CONSOLE,
  97         STDOUT_STREAM_RUNNING
  98 } StdoutStreamState;
  99
 100 struct StdoutStream {
 101         Server *server;
 102         StdoutStreamState state;
 103
 104         int fd;
 105
 106         struct ucred ucred;
 107
 108         char *tag;
 109         int priority;
 110         bool priority_prefix:1;
 111         bool tee_console:1;
 112
 113         char buffer[LINE_MAX+1];
 114         size_t length;
 115
 116         LIST_FIELDS(StdoutStream, stdout_stream);
 117 };
 118
 119 static int server_flush_to_var(Server *s);
 120
 121 static uint64_t available_space(Server *s) {
 122         char ids[33], *p;
 123         const char *f;
 124         sd_id128_t machine;
 125         struct statvfs ss;
 126         uint64_t sum = 0, avail = 0, ss_avail = 0;
 127         int r;
 128         DIR *d;
 129         usec_t ts;
 130         JournalMetrics *m;
 131
 132         ts = now(CLOCK_MONOTONIC);
 133
 134         if (s->cached_available_space_timestamp + RECHECK_AVAILABLE_SPACE_USEC > ts)
 135                 return s->cached_available_space;
 136
 137         r = sd_id128_get_machine(&machine);
 138         if (r < 0)
 139                 return 0;
 140
 141         if (s->system_journal) {
 142                 f = "/var/log/journal/";
 143                 m = &s->system_metrics;
 144         } else {
 145                 f = "/run/log/journal/";
 146                 m = &s->runtime_metrics;
 147         }
 148
 149         assert(m);
 150
 151         p = strappend(f, sd_id128_to_string(machine, ids));
 152         if (!p)
 153                 return 0;
 154
 155         d = opendir(p);
 156         free(p);
 157
 158         if (!d)
 159                 return 0;
 160
 161         if (fstatvfs(dirfd(d), &ss) < 0)
 162                 goto finish;
 163
 164         for (;;) {
 165                 struct stat st;
 166                 struct dirent buf, *de;
 167                 int k;
 168
 169                 k = readdir_r(d, &buf, &de);
 170                 if (k != 0) {
 171                         r = -k;
 172                         goto finish;
 173                 }
 174
 175                 if (!de)
 176                         break;
 177
 178                 if (!dirent_is_file_with_suffix(de, ".journal"))
 179                         continue;
 180
 181                 if (fstatat(dirfd(d), de->d_name, &st, AT_SYMLINK_NOFOLLOW) < 0)
 182                         continue;
 183
 184                 sum += (uint64_t) st.st_blocks * (uint64_t) st.st_blksize;
 185         }
 186
 187         avail = sum >= m->max_use ? 0 : m->max_use - sum;
 188
 189         ss_avail = ss.f_bsize * ss.f_bavail;
 190
 191         ss_avail = ss_avail < m->keep_free ? 0 : ss_avail - m->keep_free;
 192
 193         if (ss_avail < avail)
 194                 avail = ss_avail;
 195
 196         s->cached_available_space = avail;
 197         s->cached_available_space_timestamp = ts;
 198
 199 finish:
 200         closedir(d);
 201
 202         return avail;
 203 }
 204
 205 static void fix_perms(JournalFile *f, uid_t uid) {
 206         acl_t acl;
 207         acl_entry_t entry;
 208         acl_permset_t permset;
 209         int r;
 210
 211         assert(f);
 212
 213         r = fchmod_and_fchown(f->fd, 0640, 0, 0);
 214         if (r < 0)
 215                 log_warning("Failed to fix access mode/rights on %s, ignoring: %s", f->path, strerror(-r));
 216
 217         if (uid <= 0)
 218                 return;
 219
 220         acl = acl_get_fd(f->fd);
 221         if (!acl) {
 222                 log_warning("Failed to read ACL on %s, ignoring: %m", f->path);
 223                 return;
 224         }
 225
 226         r = acl_find_uid(acl, uid, &entry);
 227         if (r <= 0) {
 228
 229                 if (acl_create_entry(&acl, &entry) < 0 ||
 230                     acl_set_tag_type(entry, ACL_USER) < 0 ||
 231                     acl_set_qualifier(entry, &uid) < 0) {
 232                         log_warning("Failed to patch ACL on %s, ignoring: %m", f->path);
 233                         goto finish;
 234                 }
 235         }
 236
 237         if (acl_get_permset(entry, &permset) < 0 ||
 238             acl_add_perm(permset, ACL_READ) < 0 ||
 239             acl_calc_mask(&acl) < 0) {
 240                 log_warning("Failed to patch ACL on %s, ignoring: %m", f->path);
 241                 goto finish;
 242         }
 243
 244         if (acl_set_fd(f->fd, acl) < 0)
 245                 log_warning("Failed to set ACL on %s, ignoring: %m", f->path);
 246
 247 finish:
 248         acl_free(acl);
 249 }
 250
 251 static JournalFile* find_journal(Server *s, uid_t uid) {
 252         char *p;
 253         int r;
 254         JournalFile *f;
 255         char ids[33];
 256         sd_id128_t machine;
 257
 258         assert(s);
 259
 260         /* We split up user logs only on /var, not on /run. If the
 261          * runtime file is open, we write to it exclusively, in order
 262          * to guarantee proper order as soon as we flush /run to
 263          * /var and close the runtime file. */
 264
 265         if (s->runtime_journal)
 266                 return s->runtime_journal;
 267
 268         if (uid <= 0)
 269                 return s->system_journal;
 270
 271         r = sd_id128_get_machine(&machine);
 272         if (r < 0)
 273                 return s->system_journal;
 274
 275         f = hashmap_get(s->user_journals, UINT32_TO_PTR(uid));
 276         if (f)
 277                 return f;
 278
 279         if (asprintf(&p, "/var/log/journal/%s/user-%lu.journal", sd_id128_to_string(machine, ids), (unsigned long) uid) < 0)
 280                 return s->system_journal;
 281
 282         while (hashmap_size(s->user_journals) >= USER_JOURNALS_MAX) {
 283                 /* Too many open? Then let's close one */
 284                 f = hashmap_steal_first(s->user_journals);
 285                 assert(f);
 286                 journal_file_close(f);
 287         }
 288
 289         r = journal_file_open(p, O_RDWR|O_CREAT, 0640, s->system_journal, &f);
 290         free(p);
 291
 292         if (r < 0)
 293                 return s->system_journal;
 294
 295         fix_perms(f, uid);
 296         f->metrics = s->system_metrics;
 297         f->compress = s->compress;
 298
 299         r = hashmap_put(s->user_journals, UINT32_TO_PTR(uid), f);
 300         if (r < 0) {
 301                 journal_file_close(f);
 302                 return s->system_journal;
 303         }
 304
 305         return f;
 306 }
 307
 308 static void server_vacuum(Server *s) {
 309         Iterator i;
 310         void *k;
 311         char *p;
 312         char ids[33];
 313         sd_id128_t machine;
 314         int r;
 315         JournalFile *f;
 316
 317         log_info("Rotating...");
 318
 319         if (s->runtime_journal) {
 320                 r = journal_file_rotate(&s->runtime_journal);
 321                 if (r < 0)
 322                         log_error("Failed to rotate %s: %s", s->runtime_journal->path, strerror(-r));
 323         }
 324
 325         if (s->system_journal) {
 326                 r = journal_file_rotate(&s->system_journal);
 327                 if (r < 0)
 328                         log_error("Failed to rotate %s: %s", s->system_journal->path, strerror(-r));
 329         }
 330
 331         HASHMAP_FOREACH_KEY(f, k, s->user_journals, i) {
 332                 r = journal_file_rotate(&f);
 333                 if (r < 0)
 334                         log_error("Failed to rotate %s: %s", f->path, strerror(-r));
 335                 else
 336                         hashmap_replace(s->user_journals, k, f);
 337         }
 338
 339         log_info("Vacuuming...");
 340
 341         r = sd_id128_get_machine(&machine);
 342         if (r < 0) {
 343                 log_error("Failed to get machine ID: %s", strerror(-r));
 344                 return;
 345         }
 346
 347         sd_id128_to_string(machine, ids);
 348
 349         if (s->system_journal) {
 350                 if (asprintf(&p, "/var/log/journal/%s", ids) < 0) {
 351                         log_error("Out of memory.");
 352                         return;
 353                 }
 354
 355                 r = journal_directory_vacuum(p, s->system_metrics.max_use, s->system_metrics.keep_free);
 356                 if (r < 0 && r != -ENOENT)
 357                         log_error("Failed to vacuum %s: %s", p, strerror(-r));
 358                 free(p);
 359         }
 360
 361
 362         if (s->runtime_journal) {
 363                 if (asprintf(&p, "/run/log/journal/%s", ids) < 0) {
 364                         log_error("Out of memory.");
 365                         return;
 366                 }
 367
 368                 r = journal_directory_vacuum(p, s->runtime_metrics.max_use, s->runtime_metrics.keep_free);
 369                 if (r < 0 && r != -ENOENT)
 370                         log_error("Failed to vacuum %s: %s", p, strerror(-r));
 371                 free(p);
 372         }
 373
 374         s->cached_available_space_timestamp = 0;
 375 }
 376
 377 static char *shortened_cgroup_path(pid_t pid) {
 378         int r;
 379         char *process_path, *init_path, *path;
 380
 381         assert(pid > 0);
 382
 383         r = cg_get_by_pid(SYSTEMD_CGROUP_CONTROLLER, pid, &process_path);
 384         if (r < 0)
 385                 return NULL;
 386
 387         r = cg_get_by_pid(SYSTEMD_CGROUP_CONTROLLER, 1, &init_path);
 388         if (r < 0) {
 389                 free(process_path);
 390                 return NULL;
 391         }
 392
 393         if (streq(init_path, "/"))
 394                 init_path[0] = 0;
 395
 396         if (startswith(process_path, init_path)) {
 397                 char *p;
 398
 399                 p = strdup(process_path + strlen(init_path));
 400                 if (!p) {
 401                         free(process_path);
 402                         free(init_path);
 403                         return NULL;
 404                 }
 405                 path = p;
 406         } else {
 407                 path = process_path;
 408                 process_path = NULL;
 409         }
 410
 411         free(process_path);
 412         free(init_path);
 413
 414         return path;
 415 }
 416
 417 static void dispatch_message_real(Server *s,
 418                              struct iovec *iovec, unsigned n, unsigned m,
 419                              struct ucred *ucred,
 420                              struct timeval *tv) {
 421
 422         char *pid = NULL, *uid = NULL, *gid = NULL,
 423                 *source_time = NULL, *boot_id = NULL, *machine_id = NULL,
 424                 *comm = NULL, *cmdline = NULL, *hostname = NULL,
 425                 *audit_session = NULL, *audit_loginuid = NULL,
 426                 *exe = NULL, *cgroup = NULL;
 427
 428         char idbuf[33];
 429         sd_id128_t id;
 430         int r;
 431         char *t;
 432         uid_t loginuid = 0, realuid = 0;
 433         JournalFile *f;
 434         bool vacuumed = false;
 435
 436         assert(s);
 437         assert(iovec);
 438         assert(n > 0);
 439         assert(n + 13 <= m);
 440
 441         if (ucred) {
 442                 uint32_t session;
 443                 char *path;
 444
 445                 realuid = ucred->uid;
 446
 447                 if (asprintf(&pid, "_PID=%lu", (unsigned long) ucred->pid) >= 0)
 448                         IOVEC_SET_STRING(iovec[n++], pid);
 449
 450                 if (asprintf(&uid, "_UID=%lu", (unsigned long) ucred->uid) >= 0)
 451                         IOVEC_SET_STRING(iovec[n++], uid);
 452
 453                 if (asprintf(&gid, "_GID=%lu", (unsigned long) ucred->gid) >= 0)
 454                         IOVEC_SET_STRING(iovec[n++], gid);
 455
 456                 r = get_process_comm(ucred->pid, &t);
 457                 if (r >= 0) {
 458                         comm = strappend("_COMM=", t);
 459                         if (comm)
 460                                 IOVEC_SET_STRING(iovec[n++], comm);
 461                         free(t);
 462                 }
 463
 464                 r = get_process_exe(ucred->pid, &t);
 465                 if (r >= 0) {
 466                         exe = strappend("_EXE=", t);
 467                         if (comm)
 468                                 IOVEC_SET_STRING(iovec[n++], exe);
 469                         free(t);
 470                 }
 471
 472                 r = get_process_cmdline(ucred->pid, LINE_MAX, false, &t);
 473                 if (r >= 0) {
 474                         cmdline = strappend("_CMDLINE=", t);
 475                         if (cmdline)
 476                                 IOVEC_SET_STRING(iovec[n++], cmdline);
 477                         free(t);
 478                 }
 479
 480                 r = audit_session_from_pid(ucred->pid, &session);
 481                 if (r >= 0)
 482                         if (asprintf(&audit_session, "_AUDIT_SESSION=%lu", (unsigned long) session) >= 0)
 483                                 IOVEC_SET_STRING(iovec[n++], audit_session);
 484
 485                 r = audit_loginuid_from_pid(ucred->pid, &loginuid);
 486                 if (r >= 0)
 487                         if (asprintf(&audit_loginuid, "_AUDIT_LOGINUID=%lu", (unsigned long) loginuid) >= 0)
 488                                 IOVEC_SET_STRING(iovec[n++], audit_loginuid);
 489
 490                 path = shortened_cgroup_path(ucred->pid);
 491                 if (path) {
 492                         cgroup = strappend("_SYSTEMD_CGROUP=", path);
 493                         if (cgroup)
 494                                 IOVEC_SET_STRING(iovec[n++], cgroup);
 495
 496                         free(path);
 497                 }
 498         }
 499
 500         if (tv) {
 501                 if (asprintf(&source_time, "_SOURCE_REALTIME_TIMESTAMP=%llu",
 502                              (unsigned long long) timeval_load(tv)) >= 0)
 503                         IOVEC_SET_STRING(iovec[n++], source_time);
 504         }
 505
 506         /* Note that strictly speaking storing the boot id here is
 507          * redundant since the entry includes this in-line
 508          * anyway. However, we need this indexed, too. */
 509         r = sd_id128_get_boot(&id);
 510         if (r >= 0)
 511                 if (asprintf(&boot_id, "_BOOT_ID=%s", sd_id128_to_string(id, idbuf)) >= 0)
 512                         IOVEC_SET_STRING(iovec[n++], boot_id);
 513
 514         r = sd_id128_get_machine(&id);
 515         if (r >= 0)
 516                 if (asprintf(&machine_id, "_MACHINE_ID=%s", sd_id128_to_string(id, idbuf)) >= 0)
 517                         IOVEC_SET_STRING(iovec[n++], machine_id);
 518
 519         t = gethostname_malloc();
 520         if (t) {
 521                 hostname = strappend("_HOSTNAME=", t);
 522                 if (hostname)
 523                         IOVEC_SET_STRING(iovec[n++], hostname);
 524                 free(t);
 525         }
 526
 527         assert(n <= m);
 528
 529         server_flush_to_var(s);
 530
 531 retry:
 532         f = find_journal(s, realuid == 0 ? 0 : loginuid);
 533         if (!f)
 534                 log_warning("Dropping message, as we can't find a place to store the data.");
 535         else {
 536                 r = journal_file_append_entry(f, NULL, iovec, n, &s->seqnum, NULL, NULL);
 537
 538                 if (r == -E2BIG && !vacuumed) {
 539                         log_info("Allocation limit reached.");
 540
 541                         server_vacuum(s);
 542                         vacuumed = true;
 543
 544                         log_info("Retrying write.");
 545                         goto retry;
 546                 }
 547
 548                 if (r < 0)
 549                         log_error("Failed to write entry, ignoring: %s", strerror(-r));
 550         }
 551
 552         free(pid);
 553         free(uid);
 554         free(gid);
 555         free(comm);
 556         free(exe);
 557         free(cmdline);
 558         free(source_time);
 559         free(boot_id);
 560         free(machine_id);
 561         free(hostname);
 562         free(audit_session);
 563         free(audit_loginuid);
 564         free(cgroup);
 565 }
 566
 567 static void dispatch_message(Server *s,
 568                              struct iovec *iovec, unsigned n, unsigned m,
 569                              struct ucred *ucred,
 570                              struct timeval *tv,
 571                              int priority) {
 572         int rl;
 573         char *path = NULL, *c;
 574
 575         assert(s);
 576         assert(iovec || n == 0);
 577
 578         if (n == 0)
 579                 return;
 580
 581         if (!ucred)
 582                 goto finish;
 583
 584         path = shortened_cgroup_path(ucred->pid);
 585         if (!path)
 586                 goto finish;
 587
 588         /* example: /user/lennart/3/foobar
 589          *          /system/dbus.service/foobar
 590          *
 591          * So let's cut of everything past the third /, since that is
 592          * wher user directories start */
 593
 594         c = strchr(path, '/');
 595         if (c) {
 596                 c = strchr(c+1, '/');
 597                 if (c) {
 598                         c = strchr(c+1, '/');
 599                         if (c)
 600                                 *c = 0;
 601                 }
 602         }
 603
 604         rl = journal_rate_limit_test(s->rate_limit, path, priority, available_space(s));
 605
 606         if (rl == 0) {
 607                 free(path);
 608                 return;
 609         }
 610
 611         if (rl > 1) {
 612                 int j = 0;
 613                 char suppress_message[LINE_MAX];
 614                 struct iovec suppress_iovec[15];
 615
 616                 /* Write a suppression message if we suppressed something */
 617
 618                 snprintf(suppress_message, sizeof(suppress_message), "MESSAGE=Suppressed %u messages from %s", rl - 1, path);
 619                 char_array_0(suppress_message);
 620
 621                 IOVEC_SET_STRING(suppress_iovec[j++], "PRIORITY=5");
 622                 IOVEC_SET_STRING(suppress_iovec[j++], suppress_message);
 623
 624                 dispatch_message_real(s, suppress_iovec, j, ELEMENTSOF(suppress_iovec), NULL, NULL);
 625         }
 626
 627         free(path);
 628
 629 finish:
 630         dispatch_message_real(s, iovec, n, m, ucred, tv);
 631 }
 632
 633 static void process_syslog_message(Server *s, const char *buf, struct ucred *ucred, struct timeval *tv) {
 634         char *message = NULL, *syslog_priority = NULL, *syslog_facility = NULL;
 635         struct iovec iovec[16];
 636         unsigned n = 0;
 637         int priority = LOG_USER | LOG_INFO;
 638
 639         assert(s);
 640         assert(buf);
 641
 642         parse_syslog_priority((char**) &buf, &priority);
 643         skip_syslog_date((char**) &buf);
 644
 645         if (asprintf(&syslog_priority, "PRIORITY=%i", priority & LOG_PRIMASK) >= 0)
 646                 IOVEC_SET_STRING(iovec[n++], syslog_priority);
 647
 648         if (asprintf(&syslog_facility, "SYSLOG_FACILITY=%i", LOG_FAC(priority)) >= 0)
 649                 IOVEC_SET_STRING(iovec[n++], syslog_facility);
 650
 651         message = strappend("MESSAGE=", buf);
 652         if (message)
 653                 IOVEC_SET_STRING(iovec[n++], message);
 654
 655         dispatch_message(s, iovec, n, ELEMENTSOF(iovec), ucred, tv, priority & LOG_PRIMASK);
 656
 657         free(message);
 658         free(syslog_facility);
 659         free(syslog_priority);
 660 }
 661
 662 static bool valid_user_field(const char *p, size_t l) {
 663         const char *a;
 664
 665         /* We kinda enforce POSIX syntax recommendations for
 666            environment variables here, but make a couple of additional
 667            requirements.
 668
 669            http://pubs.opengroup.org/onlinepubs/000095399/basedefs/xbd_chap08.html */
 670
 671         /* No empty field names */
 672         if (l <= 0)
 673                 return false;
 674
 675         /* Don't allow names longer than 64 chars */
 676         if (l > 64)
 677                 return false;
 678
 679         /* Variables starting with an underscore are protected */
 680         if (p[0] == '_')
 681                 return false;
 682
 683         /* Don't allow digits as first character */
 684         if (p[0] >= '0' && p[0] <= '9')
 685                 return false;
 686
 687         /* Only allow A-Z0-9 and '_' */
 688         for (a = p; a < p + l; a++)
 689                 if (!((*a >= 'A' && *a <= 'Z') ||
 690                       (*a >= '0' && *a <= '9') ||
 691                       *a == '_'))
 692                         return false;
 693
 694         return true;
 695 }
 696
 697 static void process_native_message(Server *s, const void *buffer, size_t buffer_size, struct ucred *ucred, struct timeval *tv) {
 698         struct iovec *iovec = NULL;
 699         unsigned n = 0, m = 0, j;
 700         const char *p;
 701         size_t remaining;
 702         int priority = LOG_INFO;
 703
 704         assert(s);
 705         assert(buffer || n == 0);
 706
 707         p = buffer;
 708         remaining = buffer_size;
 709
 710         while (remaining > 0) {
 711                 const char *e, *q;
 712
 713                 e = memchr(p, '\n', remaining);
 714
 715                 if (!e) {
 716                         /* Trailing noise, let's ignore it, and flush what we collected */
 717                         log_debug("Received message with trailing noise, ignoring.");
 718                         break;
 719                 }
 720
 721                 if (e == p) {
 722                         /* Entry separator */
 723                         dispatch_message(s, iovec, n, m, ucred, tv, priority);
 724                         n = 0;
 725                         priority = LOG_INFO;
 726
 727                         p++;
 728                         remaining--;
 729                         continue;
 730                 }
 731
 732                 if (*p == '.' || *p == '#') {
 733                         /* Ignore control commands for now, and
 734                          * comments too. */
 735                         remaining -= (e - p) + 1;
 736                         p = e + 1;
 737                         continue;
 738                 }
 739
 740                 /* A property follows */
 741
 742                 if (n+13 >= m) {
 743                         struct iovec *c;
 744                         unsigned u;
 745
 746                         u = MAX((n+13U) * 2U, 4U);
 747                         c = realloc(iovec, u * sizeof(struct iovec));
 748                         if (!c) {
 749                                 log_error("Out of memory");
 750                                 break;
 751                         }
 752
 753                         iovec = c;
 754                         m = u;
 755                 }
 756
 757                 q = memchr(p, '=', e - p);
 758                 if (q) {
 759                         if (valid_user_field(p, q - p)) {
 760                                 /* If the field name starts with an
 761                                  * underscore, skip the variable,
 762                                  * since that indidates a trusted
 763                                  * field */
 764                                 iovec[n].iov_base = (char*) p;
 765                                 iovec[n].iov_len = e - p;
 766                                 n++;
 767
 768                                 /* We need to determine the priority
 769                                  * of this entry for the rate limiting
 770                                  * logic */
 771                                 if (e - p == 10 &&
 772                                     memcmp(p, "PRIORITY=", 10) == 0 &&
 773                                     p[10] >= '0' &&
 774                                     p[10] <= '9')
 775                                         priority = p[10] - '0';
 776                         }
 777
 778                         remaining -= (e - p) + 1;
 779                         p = e + 1;
 780                         continue;
 781                 } else {
 782                         uint64_t l;
 783                         char *k;
 784
 785                         if (remaining < e - p + 1 + sizeof(uint64_t) + 1) {
 786                                 log_debug("Failed to parse message, ignoring.");
 787                                 break;
 788                         }
 789
 790                         memcpy(&l, e + 1, sizeof(uint64_t));
 791                         l = le64toh(l);
 792
 793                         if (remaining < e - p + 1 + sizeof(uint64_t) + l + 1 ||
 794                             e[1+sizeof(uint64_t)+l] != '\n') {
 795                                 log_debug("Failed to parse message, ignoring.");
 796                                 break;
 797                         }
 798
 799                         k = malloc((e - p) + 1 + l);
 800                         if (!k) {
 801                                 log_error("Out of memory");
 802                                 break;
 803                         }
 804
 805                         memcpy(k, p, e - p);
 806                         k[e - p] = '=';
 807                         memcpy(k + (e - p) + 1, e + 1 + sizeof(uint64_t), l);
 808
 809                         if (valid_user_field(p, e - p)) {
 810                                 iovec[n].iov_base = k;
 811                                 iovec[n].iov_len = (e - p) + 1 + l;
 812                                 n++;
 813                         } else
 814                                 free(k);
 815
 816                         remaining -= (e - p) + 1 + sizeof(uint64_t) + l + 1;
 817                         p = e + 1 + sizeof(uint64_t) + l + 1;
 818                 }
 819         }
 820
 821         dispatch_message(s, iovec, n, m, ucred, tv, priority);
 822
 823         for (j = 0; j < n; j++)
 824                 if (iovec[j].iov_base < buffer ||
 825                     (const uint8_t*) iovec[j].iov_base >= (const uint8_t*) buffer + buffer_size)
 826                         free(iovec[j].iov_base);
 827 }
 828
 829 static int stdout_stream_log(StdoutStream *s, const char *p, size_t l) {
 830         struct iovec iovec[15];
 831         char *message = NULL, *syslog_priority = NULL;
 832         unsigned n = 0;
 833         size_t tag_len;
 834         int priority;
 835
 836         assert(s);
 837         assert(p);
 838
 839         priority = s->priority;
 840
 841         if (s->priority_prefix &&
 842             l > 3 &&
 843             p[0] == '<' &&
 844             p[1] >= '0' && p[1] <= '7' &&
 845             p[2] == '>') {
 846
 847                 priority = p[1] - '0';
 848                 p += 3;
 849                 l -= 3;
 850         }
 851
 852         if (l <= 0)
 853                 return 0;
 854
 855         if (asprintf(&syslog_priority, "PRIORITY=%i", priority) >= 0)
 856                 IOVEC_SET_STRING(iovec[n++], syslog_priority);
 857
 858         tag_len = s->tag ? strlen(s->tag) + 2: 0;
 859         message = malloc(8 + tag_len + l);
 860         if (message) {
 861                 memcpy(message, "MESSAGE=", 8);
 862
 863                 if (s->tag) {
 864                         memcpy(message+8, s->tag, tag_len-2);
 865                         memcpy(message+8+tag_len-2, ": ", 2);
 866                 }
 867
 868                 memcpy(message+8+tag_len, p, l);
 869                 iovec[n].iov_base = message;
 870                 iovec[n].iov_len = 8+tag_len+l;
 871                 n++;
 872         }
 873
 874         dispatch_message(s->server, iovec, n, ELEMENTSOF(iovec), &s->ucred, NULL, priority);
 875
 876         if (s->tee_console) {
 877                 int console;
 878
 879                 console = open_terminal("/dev/console", O_WRONLY|O_NOCTTY|O_CLOEXEC);
 880                 if (console >= 0) {
 881                         n = 0;
 882                         if (s->tag) {
 883                                 IOVEC_SET_STRING(iovec[n++], s->tag);
 884                                 IOVEC_SET_STRING(iovec[n++], ": ");
 885                         }
 886
 887                         iovec[n].iov_base = (void*) p;
 888                         iovec[n].iov_len = l;
 889                         n++;
 890
 891                         IOVEC_SET_STRING(iovec[n++], (char*) "\n");
 892
 893                         writev(console, iovec, n);
 894                 }
 895         }
 896
 897         free(message);
 898         free(syslog_priority);
 899
 900         return 0;
 901 }
 902
 903 static int stdout_stream_line(StdoutStream *s, const char *p, size_t l) {
 904         assert(s);
 905         assert(p);
 906
 907         while (l > 0 && strchr(WHITESPACE, *p)) {
 908                 l--;
 909                 p++;
 910         }
 911
 912         while (l > 0 && strchr(WHITESPACE, *(p+l-1)))
 913                 l--;
 914
 915         switch (s->state) {
 916
 917         case STDOUT_STREAM_TAG:
 918
 919                 if (l > 0) {
 920                         s->tag = strndup(p, l);
 921                         if (!s->tag) {
 922                                 log_error("Out of memory");
 923                                 return -EINVAL;
 924                         }
 925                 }
 926
 927                 s->state = STDOUT_STREAM_PRIORITY;
 928                 return 0;
 929
 930         case STDOUT_STREAM_PRIORITY:
 931                 if (l != 1 || *p < '0' || *p > '7') {
 932                         log_warning("Failed to parse log priority line.");
 933                         return -EINVAL;
 934                 }
 935
 936                 s->priority = *p - '0';
 937                 s->state = STDOUT_STREAM_PRIORITY_PREFIX;
 938                 return 0;
 939
 940         case STDOUT_STREAM_PRIORITY_PREFIX:
 941                 if (l != 1 || *p < '0' || *p > '1') {
 942                         log_warning("Failed to parse priority prefix line.");
 943                         return -EINVAL;
 944                 }
 945
 946                 s->priority_prefix = *p - '0';
 947                 s->state = STDOUT_STREAM_TEE_CONSOLE;
 948                 return 0;
 949
 950         case STDOUT_STREAM_TEE_CONSOLE:
 951                 if (l != 1 || *p < '0' || *p > '1') {
 952                         log_warning("Failed to parse tee to console line.");
 953                         return -EINVAL;
 954                 }
 955
 956                 s->tee_console = *p - '0';
 957                 s->state = STDOUT_STREAM_RUNNING;
 958                 return 0;
 959
 960         case STDOUT_STREAM_RUNNING:
 961                 return stdout_stream_log(s, p, l);
 962         }
 963
 964         assert_not_reached("Unknown stream state");
 965 }
 966
 967 static int stdout_stream_scan(StdoutStream *s, bool force_flush) {
 968         char *p;
 969         size_t remaining;
 970         int r;
 971
 972         assert(s);
 973
 974         p = s->buffer;
 975         remaining = s->length;
 976         for (;;) {
 977                 char *end;
 978                 size_t skip;
 979
 980                 end = memchr(p, '\n', remaining);
 981                 if (!end) {
 982                         if (remaining >= LINE_MAX) {
 983                                 end = p + LINE_MAX;
 984                                 skip = LINE_MAX;
 985                         } else
 986                                 break;
 987                 } else
 988                         skip = end - p + 1;
 989
 990                 r = stdout_stream_line(s, p, end - p);
 991                 if (r < 0)
 992                         return r;
 993
 994                 remaining -= skip;
 995                 p += skip;
 996         }
 997
 998         if (force_flush && remaining > 0) {
 999                 r = stdout_stream_line(s, p, remaining);
1000                 if (r < 0)
1001                         return r;
1002
1003                 p += remaining;
1004                 remaining = 0;
1005         }
1006
1007         if (p > s->buffer) {
1008                 memmove(s->buffer, p, remaining);
1009                 s->length = remaining;
1010         }
1011
1012         return 0;
1013 }
1014
1015 static int stdout_stream_process(StdoutStream *s) {
1016         ssize_t l;
1017         int r;
1018
1019         assert(s);
1020
1021         l = read(s->fd, s->buffer+s->length, sizeof(s->buffer)-1-s->length);
1022         if (l < 0) {
1023
1024                 if (errno == EAGAIN)
1025                         return 0;
1026
1027                 log_warning("Failed to read from stream: %m");
1028                 return -errno;
1029         }
1030
1031         if (l == 0) {
1032                 r = stdout_stream_scan(s, true);
1033                 if (r < 0)
1034                         return r;
1035
1036                 return 0;
1037         }
1038
1039         s->length += l;
1040         r = stdout_stream_scan(s, false);
1041         if (r < 0)
1042                 return r;
1043
1044         return 1;
1045
1046 }
1047
1048 static void stdout_stream_free(StdoutStream *s) {
1049         assert(s);
1050
1051         if (s->server) {
1052                 assert(s->server->n_stdout_streams > 0);
1053                 s->server->n_stdout_streams --;
1054                 LIST_REMOVE(StdoutStream, stdout_stream, s->server->stdout_streams, s);
1055         }
1056
1057         if (s->fd >= 0) {
1058                 if (s->server)
1059                         epoll_ctl(s->server->epoll_fd, EPOLL_CTL_DEL, s->fd, NULL);
1060
1061                 close_nointr_nofail(s->fd);
1062         }
1063
1064         free(s->tag);
1065         free(s);
1066 }
1067
1068 static int stdout_stream_new(Server *s) {
1069         StdoutStream *stream;
1070         int fd, r;
1071         socklen_t len;
1072         struct epoll_event ev;
1073
1074         assert(s);
1075
1076         fd = accept4(s->stdout_fd, NULL, NULL, SOCK_NONBLOCK|SOCK_CLOEXEC);
1077         if (fd < 0) {
1078                 if (errno == EAGAIN)
1079                         return 0;
1080
1081                 log_error("Failed to accept stdout connection: %m");
1082                 return -errno;
1083         }
1084
1085         if (s->n_stdout_streams >= STDOUT_STREAMS_MAX) {
1086                 log_warning("Too many stdout streams, refusing connection.");
1087                 close_nointr_nofail(fd);
1088                 return 0;
1089         }
1090
1091         stream = new0(StdoutStream, 1);
1092         if (!stream) {
1093                 log_error("Out of memory.");
1094                 close_nointr_nofail(fd);
1095                 return -ENOMEM;
1096         }
1097
1098         stream->fd = fd;
1099
1100         len = sizeof(stream->ucred);
1101         if (getsockopt(fd, SOL_SOCKET, SO_PEERCRED, &stream->ucred, &len) < 0) {
1102                 log_error("Failed to determine peer credentials: %m");
1103                 r = -errno;
1104                 goto fail;
1105         }
1106
1107         if (shutdown(fd, SHUT_WR) < 0) {
1108                 log_error("Failed to shutdown writing side of socket: %m");
1109                 r = -errno;
1110                 goto fail;
1111         }
1112
1113         zero(ev);
1114         ev.data.ptr = stream;
1115         ev.events = EPOLLIN;
1116         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, fd, &ev) < 0) {
1117                 log_error("Failed to add stream to event loop: %m");
1118                 r = -errno;
1119                 goto fail;
1120         }
1121
1122         stream->server = s;
1123         LIST_PREPEND(StdoutStream, stdout_stream, s->stdout_streams, stream);
1124         s->n_stdout_streams ++;
1125
1126         return 0;
1127
1128 fail:
1129         stdout_stream_free(stream);
1130         return r;
1131 }
1132
1133 static int system_journal_open(Server *s) {
1134         int r;
1135         char *fn;
1136         sd_id128_t machine;
1137         char ids[33];
1138
1139         r = sd_id128_get_machine(&machine);
1140         if (r < 0)
1141                 return r;
1142
1143         sd_id128_to_string(machine, ids);
1144
1145         if (!s->system_journal) {
1146
1147                 /* First try to create the machine path, but not the prefix */
1148                 fn = strappend("/var/log/journal/", ids);
1149                 if (!fn)
1150                         return -ENOMEM;
1151                 (void) mkdir(fn, 0755);
1152                 free(fn);
1153
1154                 /* The create the system journal file */
1155                 fn = join("/var/log/journal/", ids, "/system.journal", NULL);
1156                 if (!fn)
1157                         return -ENOMEM;
1158
1159                 r = journal_file_open(fn, O_RDWR|O_CREAT, 0640, NULL, &s->system_journal);
1160                 free(fn);
1161
1162                 if (r >= 0) {
1163                         journal_default_metrics(&s->system_metrics, s->system_journal->fd);
1164
1165                         s->system_journal->metrics = s->system_metrics;
1166                         s->system_journal->compress = s->compress;
1167
1168                         fix_perms(s->system_journal, 0);
1169                 } else if (r < 0) {
1170
1171                         if (r == -ENOENT)
1172                                 r = 0;
1173                         else {
1174                                 log_error("Failed to open system journal: %s", strerror(-r));
1175                                 return r;
1176                         }
1177                 }
1178         }
1179
1180         if (!s->runtime_journal) {
1181
1182                 fn = join("/run/log/journal/", ids, "/system.journal", NULL);
1183                 if (!fn)
1184                         return -ENOMEM;
1185
1186                 if (s->system_journal) {
1187
1188                         /* Try to open the runtime journal, but only
1189                          * if it already exists, so that we can flush
1190                          * it into the system journal */
1191
1192                         r = journal_file_open(fn, O_RDWR, 0640, NULL, &s->runtime_journal);
1193                         free(fn);
1194
1195                         if (r < 0) {
1196
1197                                 if (r == -ENOENT)
1198                                         r = 0;
1199                                 else {
1200                                         log_error("Failed to open runtime journal: %s", strerror(-r));
1201                                         return r;
1202                                 }
1203                         }
1204
1205                 } else {
1206
1207                         /* OK, we really need the runtime journal, so create
1208                          * it if necessary. */
1209
1210                         (void) mkdir_parents(fn, 0755);
1211                         r = journal_file_open(fn, O_RDWR|O_CREAT, 0640, NULL, &s->runtime_journal);
1212                         free(fn);
1213
1214                         if (r < 0) {
1215                                 log_error("Failed to open runtime journal: %s", strerror(-r));
1216                                 return r;
1217                         }
1218                 }
1219
1220                 if (s->runtime_journal) {
1221                         journal_default_metrics(&s->runtime_metrics, s->runtime_journal->fd);
1222
1223                         s->runtime_journal->metrics = s->runtime_metrics;
1224                         s->runtime_journal->compress = s->compress;
1225
1226                         fix_perms(s->runtime_journal, 0);
1227                 }
1228         }
1229
1230         return r;
1231 }
1232
1233 static int server_flush_to_var(Server *s) {
1234         char path[] = "/run/log/journal/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx";
1235         Object *o = NULL;
1236         int r;
1237         sd_id128_t machine;
1238         sd_journal *j;
1239         usec_t ts;
1240
1241         assert(s);
1242
1243         if (!s->runtime_journal)
1244                 return 0;
1245
1246         ts = now(CLOCK_MONOTONIC);
1247         if (s->var_available_timestamp + RECHECK_VAR_AVAILABLE_USEC > ts)
1248                 return 0;
1249
1250         s->var_available_timestamp = ts;
1251
1252         system_journal_open(s);
1253
1254         if (!s->system_journal)
1255                 return 0;
1256
1257         r = sd_id128_get_machine(&machine);
1258         if (r < 0) {
1259                 log_error("Failed to get machine id: %s", strerror(-r));
1260                 return r;
1261         }
1262
1263         r = sd_journal_open(&j, SD_JOURNAL_RUNTIME_ONLY);
1264         if (r < 0) {
1265                 log_error("Failed to read runtime journal: %s", strerror(-r));
1266                 return r;
1267         }
1268
1269         SD_JOURNAL_FOREACH(j) {
1270                 JournalFile *f;
1271
1272                 f = j->current_file;
1273                 assert(f && f->current_offset > 0);
1274
1275                 r = journal_file_move_to_object(f, OBJECT_ENTRY, f->current_offset, &o);
1276                 if (r < 0) {
1277                         log_error("Can't read entry: %s", strerror(-r));
1278                         goto finish;
1279                 }
1280
1281                 r = journal_file_copy_entry(f, s->system_journal, o, f->current_offset, NULL, NULL, NULL);
1282                 if (r == -E2BIG) {
1283                         log_info("Allocation limit reached.");
1284
1285                         journal_file_post_change(s->system_journal);
1286                         server_vacuum(s);
1287
1288                         r = journal_file_copy_entry(f, s->system_journal, o, f->current_offset, NULL, NULL, NULL);
1289                 }
1290
1291                 if (r < 0) {
1292                         log_error("Can't write entry: %s", strerror(-r));
1293                         goto finish;
1294                 }
1295         }
1296
1297 finish:
1298         journal_file_post_change(s->system_journal);
1299
1300         journal_file_close(s->runtime_journal);
1301         s->runtime_journal = NULL;
1302
1303         if (r >= 0) {
1304                 sd_id128_to_string(machine, path + 17);
1305                 rm_rf(path, false, true, false);
1306         }
1307
1308         return r;
1309 }
1310
1311 static void forward_syslog(Server *s, const void *buffer, size_t length, struct ucred *ucred, struct timeval *tv) {
1312         struct msghdr msghdr;
1313         struct iovec iovec;
1314         struct cmsghdr *cmsg;
1315         union {
1316                 struct cmsghdr cmsghdr;
1317                 uint8_t buf[CMSG_SPACE(sizeof(struct ucred)) +
1318                             CMSG_SPACE(sizeof(struct timeval))];
1319         } control;
1320         union sockaddr_union sa;
1321
1322         assert(s);
1323
1324         zero(msghdr);
1325
1326         zero(iovec);
1327         iovec.iov_base = (void*) buffer;
1328         iovec.iov_len = length;
1329         msghdr.msg_iov = &iovec;
1330         msghdr.msg_iovlen = 1;
1331
1332         zero(sa);
1333         sa.un.sun_family = AF_UNIX;
1334         strncpy(sa.un.sun_path, "/run/systemd/syslog", sizeof(sa.un.sun_path));
1335         msghdr.msg_name = &sa;
1336         msghdr.msg_namelen = offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path);
1337
1338         zero(control);
1339         msghdr.msg_control = &control;
1340         msghdr.msg_controllen = sizeof(control);
1341
1342         cmsg = CMSG_FIRSTHDR(&msghdr);
1343         cmsg->cmsg_level = SOL_SOCKET;
1344         cmsg->cmsg_type = SCM_CREDENTIALS;
1345         cmsg->cmsg_len = CMSG_LEN(sizeof(struct ucred));
1346         memcpy(CMSG_DATA(cmsg), ucred, sizeof(struct ucred));
1347         msghdr.msg_controllen = cmsg->cmsg_len;
1348
1349         /* Forward the syslog message we received via /dev/log to
1350          * /run/systemd/syslog. Unfortunately we currently can't set
1351          * the SO_TIMESTAMP auxiliary data, and hence we don't. */
1352
1353         if (sendmsg(s->syslog_fd, &msghdr, MSG_NOSIGNAL) >= 0)
1354                 return;
1355
1356         if (errno == ESRCH) {
1357                 struct ucred u;
1358
1359                 /* Hmm, presumably the sender process vanished
1360                  * by now, so let's fix it as good as we
1361                  * can, and retry */
1362
1363                 u = *ucred;
1364                 u.pid = getpid();
1365                 memcpy(CMSG_DATA(cmsg), &u, sizeof(struct ucred));
1366
1367                 if (sendmsg(s->syslog_fd, &msghdr, MSG_NOSIGNAL) >= 0)
1368                         return;
1369         }
1370
1371         log_debug("Failed to forward syslog message: %m");
1372 }
1373
1374 static int process_event(Server *s, struct epoll_event *ev) {
1375         assert(s);
1376
1377         if (ev->data.fd == s->signal_fd) {
1378                 struct signalfd_siginfo sfsi;
1379                 ssize_t n;
1380
1381                 if (ev->events != EPOLLIN) {
1382                         log_info("Got invalid event from epoll.");
1383                         return -EIO;
1384                 }
1385
1386                 n = read(s->signal_fd, &sfsi, sizeof(sfsi));
1387                 if (n != sizeof(sfsi)) {
1388
1389                         if (n >= 0)
1390                                 return -EIO;
1391
1392                         if (errno == EINTR || errno == EAGAIN)
1393                                 return 0;
1394
1395                         return -errno;
1396                 }
1397
1398                 if (sfsi.ssi_signo == SIGUSR1) {
1399                         server_flush_to_var(s);
1400                         return 0;
1401                 }
1402
1403                 log_debug("Received SIG%s", signal_to_string(sfsi.ssi_signo));
1404                 return 0;
1405
1406         } else if (ev->data.fd == s->native_fd ||
1407                    ev->data.fd == s->syslog_fd) {
1408
1409                 if (ev->events != EPOLLIN) {
1410                         log_info("Got invalid event from epoll.");
1411                         return -EIO;
1412                 }
1413
1414                 for (;;) {
1415                         struct msghdr msghdr;
1416                         struct iovec iovec;
1417                         struct ucred *ucred = NULL;
1418                         struct timeval *tv = NULL;
1419                         struct cmsghdr *cmsg;
1420                         union {
1421                                 struct cmsghdr cmsghdr;
1422                                 uint8_t buf[CMSG_SPACE(sizeof(struct ucred)) +
1423                                             CMSG_SPACE(sizeof(struct timeval))];
1424                         } control;
1425                         ssize_t n;
1426                         int v;
1427
1428                         if (ioctl(ev->data.fd, SIOCINQ, &v) < 0) {
1429                                 log_error("SIOCINQ failed: %m");
1430                                 return -errno;
1431                         }
1432
1433                         if (v <= 0)
1434                                 return 1;
1435
1436                         if (s->buffer_size < (size_t) v) {
1437                                 void *b;
1438                                 size_t l;
1439
1440                                 l = MAX(LINE_MAX + (size_t) v, s->buffer_size * 2);
1441                                 b = realloc(s->buffer, l+1);
1442
1443                                 if (!b) {
1444                                         log_error("Couldn't increase buffer.");
1445                                         return -ENOMEM;
1446                                 }
1447
1448                                 s->buffer_size = l;
1449                                 s->buffer = b;
1450                         }
1451
1452                         zero(iovec);
1453                         iovec.iov_base = s->buffer;
1454                         iovec.iov_len = s->buffer_size;
1455
1456                         zero(control);
1457                         zero(msghdr);
1458                         msghdr.msg_iov = &iovec;
1459                         msghdr.msg_iovlen = 1;
1460                         msghdr.msg_control = &control;
1461                         msghdr.msg_controllen = sizeof(control);
1462
1463                         n = recvmsg(ev->data.fd, &msghdr, MSG_DONTWAIT);
1464                         if (n < 0) {
1465
1466                                 if (errno == EINTR || errno == EAGAIN)
1467                                         return 1;
1468
1469                                 log_error("recvmsg() failed: %m");
1470                                 return -errno;
1471                         }
1472
1473                         for (cmsg = CMSG_FIRSTHDR(&msghdr); cmsg; cmsg = CMSG_NXTHDR(&msghdr, cmsg)) {
1474
1475                                 if (cmsg->cmsg_level == SOL_SOCKET &&
1476                                     cmsg->cmsg_type == SCM_CREDENTIALS &&
1477                                     cmsg->cmsg_len == CMSG_LEN(sizeof(struct ucred)))
1478                                         ucred = (struct ucred*) CMSG_DATA(cmsg);
1479                                 else if (cmsg->cmsg_level == SOL_SOCKET &&
1480                                          cmsg->cmsg_type == SO_TIMESTAMP &&
1481                                          cmsg->cmsg_len == CMSG_LEN(sizeof(struct timeval)))
1482                                         tv = (struct timeval*) CMSG_DATA(cmsg);
1483                         }
1484
1485                         if (ev->data.fd == s->syslog_fd) {
1486                                 char *e;
1487
1488                                 e = memchr(s->buffer, '\n', n);
1489                                 if (e)
1490                                         *e = 0;
1491                                 else
1492                                         s->buffer[n] = 0;
1493
1494                                 forward_syslog(s, s->buffer, n, ucred, tv);
1495                                 process_syslog_message(s, strstrip(s->buffer), ucred, tv);
1496                         } else
1497                                 process_native_message(s, s->buffer, n, ucred, tv);
1498                 }
1499
1500                 return 1;
1501
1502         } else if (ev->data.fd == s->stdout_fd) {
1503
1504                 if (ev->events != EPOLLIN) {
1505                         log_info("Got invalid event from epoll.");
1506                         return -EIO;
1507                 }
1508
1509                 stdout_stream_new(s);
1510                 return 1;
1511
1512         } else {
1513                 StdoutStream *stream;
1514
1515                 if ((ev->events|EPOLLIN|EPOLLHUP) != (EPOLLIN|EPOLLHUP)) {
1516                         log_info("Got invalid event from epoll.");
1517                         return -EIO;
1518                 }
1519
1520                 /* If it is none of the well-known fds, it must be an
1521                  * stdout stream fd. Note that this is a bit ugly here
1522                  * (since we rely that none of the well-known fds
1523                  * could be interpreted as pointer), but nonetheless
1524                  * safe, since the well-known fds would never get an
1525                  * fd > 4096, i.e. beyond the first memory page */
1526
1527                 stream = ev->data.ptr;
1528
1529                 if (stdout_stream_process(stream) <= 0)
1530                         stdout_stream_free(stream);
1531
1532                 return 1;
1533         }
1534
1535         log_error("Unknown event.");
1536         return 0;
1537 }
1538
1539 static int open_syslog_socket(Server *s) {
1540         union sockaddr_union sa;
1541         int one, r;
1542         struct epoll_event ev;
1543         struct timeval tv;
1544
1545         assert(s);
1546
1547         if (s->syslog_fd < 0) {
1548
1549                 s->syslog_fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC, 0);
1550                 if (s->syslog_fd < 0) {
1551                         log_error("socket() failed: %m");
1552                         return -errno;
1553                 }
1554
1555                 zero(sa);
1556                 sa.un.sun_family = AF_UNIX;
1557                 strncpy(sa.un.sun_path, "/dev/log", sizeof(sa.un.sun_path));
1558
1559                 unlink(sa.un.sun_path);
1560
1561                 r = bind(s->syslog_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
1562                 if (r < 0) {
1563                         log_error("bind() failed: %m");
1564                         return -errno;
1565                 }
1566
1567                 chmod(sa.un.sun_path, 0666);
1568         }
1569
1570         one = 1;
1571         r = setsockopt(s->syslog_fd, SOL_SOCKET, SO_PASSCRED, &one, sizeof(one));
1572         if (r < 0) {
1573                 log_error("SO_PASSCRED failed: %m");
1574                 return -errno;
1575         }
1576
1577         one = 1;
1578         r = setsockopt(s->syslog_fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one));
1579         if (r < 0) {
1580                 log_error("SO_TIMESTAMP failed: %m");
1581                 return -errno;
1582         }
1583
1584         /* Since we use the same socket for forwarding this to some
1585          * other syslog implementation, make sure we don't hang
1586          * forever */
1587         timeval_store(&tv, SYSLOG_TIMEOUT_USEC);
1588         if (setsockopt(s->syslog_fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv)) < 0) {
1589                 log_error("SO_SNDTIMEO failed: %m");
1590                 return -errno;
1591         }
1592
1593         zero(ev);
1594         ev.events = EPOLLIN;
1595         ev.data.fd = s->syslog_fd;
1596         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->syslog_fd, &ev) < 0) {
1597                 log_error("Failed to add syslog server fd to epoll object: %m");
1598                 return -errno;
1599         }
1600
1601         return 0;
1602 }
1603
1604 static int open_native_socket(Server*s) {
1605         union sockaddr_union sa;
1606         int one, r;
1607         struct epoll_event ev;
1608
1609         assert(s);
1610
1611         if (s->native_fd < 0) {
1612
1613                 s->native_fd = socket(AF_UNIX, SOCK_DGRAM|SOCK_CLOEXEC, 0);
1614                 if (s->native_fd < 0) {
1615                         log_error("socket() failed: %m");
1616                         return -errno;
1617                 }
1618
1619                 zero(sa);
1620                 sa.un.sun_family = AF_UNIX;
1621                 strncpy(sa.un.sun_path, "/run/systemd/journal", sizeof(sa.un.sun_path));
1622
1623                 unlink(sa.un.sun_path);
1624
1625                 r = bind(s->native_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
1626                 if (r < 0) {
1627                         log_error("bind() failed: %m");
1628                         return -errno;
1629                 }
1630
1631                 chmod(sa.un.sun_path, 0666);
1632         }
1633
1634         one = 1;
1635         r = setsockopt(s->native_fd, SOL_SOCKET, SO_PASSCRED, &one, sizeof(one));
1636         if (r < 0) {
1637                 log_error("SO_PASSCRED failed: %m");
1638                 return -errno;
1639         }
1640
1641         one = 1;
1642         r = setsockopt(s->native_fd, SOL_SOCKET, SO_TIMESTAMP, &one, sizeof(one));
1643         if (r < 0) {
1644                 log_error("SO_TIMESTAMP failed: %m");
1645                 return -errno;
1646         }
1647
1648         zero(ev);
1649         ev.events = EPOLLIN;
1650         ev.data.fd = s->native_fd;
1651         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->native_fd, &ev) < 0) {
1652                 log_error("Failed to add native server fd to epoll object: %m");
1653                 return -errno;
1654         }
1655
1656         return 0;
1657 }
1658
1659 static int open_stdout_socket(Server *s) {
1660         union sockaddr_union sa;
1661         int r;
1662         struct epoll_event ev;
1663
1664         assert(s);
1665
1666         if (s->stdout_fd < 0) {
1667
1668                 s->stdout_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0);
1669                 if (s->stdout_fd < 0) {
1670                         log_error("socket() failed: %m");
1671                         return -errno;
1672                 }
1673
1674                 zero(sa);
1675                 sa.un.sun_family = AF_UNIX;
1676                 strncpy(sa.un.sun_path, "/run/systemd/stdout", sizeof(sa.un.sun_path));
1677
1678                 unlink(sa.un.sun_path);
1679
1680                 r = bind(s->stdout_fd, &sa.sa, offsetof(union sockaddr_union, un.sun_path) + strlen(sa.un.sun_path));
1681                 if (r < 0) {
1682                         log_error("bind() failed: %m");
1683                         return -errno;
1684                 }
1685
1686                 chmod(sa.un.sun_path, 0666);
1687
1688                 if (listen(s->stdout_fd, SOMAXCONN) < 0) {
1689                         log_error("liste() failed: %m");
1690                         return -errno;
1691                 }
1692         }
1693
1694         zero(ev);
1695         ev.events = EPOLLIN;
1696         ev.data.fd = s->stdout_fd;
1697         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->stdout_fd, &ev) < 0) {
1698                 log_error("Failed to add stdout server fd to epoll object: %m");
1699                 return -errno;
1700         }
1701
1702         return 0;
1703 }
1704
1705 static int open_signalfd(Server *s) {
1706         sigset_t mask;
1707         struct epoll_event ev;
1708
1709         assert(s);
1710
1711         assert_se(sigemptyset(&mask) == 0);
1712         sigset_add_many(&mask, SIGINT, SIGTERM, SIGUSR1, -1);
1713         assert_se(sigprocmask(SIG_SETMASK, &mask, NULL) == 0);
1714
1715         s->signal_fd = signalfd(-1, &mask, SFD_NONBLOCK|SFD_CLOEXEC);
1716         if (s->signal_fd < 0) {
1717                 log_error("signalfd(): %m");
1718                 return -errno;
1719         }
1720
1721         zero(ev);
1722         ev.events = EPOLLIN;
1723         ev.data.fd = s->signal_fd;
1724
1725         if (epoll_ctl(s->epoll_fd, EPOLL_CTL_ADD, s->signal_fd, &ev) < 0) {
1726                 log_error("epoll_ctl(): %m");
1727                 return -errno;
1728         }
1729
1730         return 0;
1731 }
1732
1733 static int server_init(Server *s) {
1734         int n, r, fd;
1735
1736         assert(s);
1737
1738         zero(*s);
1739         s->syslog_fd = s->native_fd = s->stdout_fd = s->signal_fd = s->epoll_fd = -1;
1740         s->compress = true;
1741
1742         memset(&s->system_metrics, 0xFF, sizeof(s->system_metrics));
1743         memset(&s->runtime_metrics, 0xFF, sizeof(s->runtime_metrics));
1744
1745         s->user_journals = hashmap_new(trivial_hash_func, trivial_compare_func);
1746         if (!s->user_journals) {
1747                 log_error("Out of memory.");
1748                 return -ENOMEM;
1749         }
1750
1751         s->epoll_fd = epoll_create1(EPOLL_CLOEXEC);
1752         if (s->epoll_fd < 0) {
1753                 log_error("Failed to create epoll object: %m");
1754                 return -errno;
1755         }
1756
1757         n = sd_listen_fds(true);
1758         if (n < 0) {
1759                 log_error("Failed to read listening file descriptors from environment: %s", strerror(-n));
1760                 return n;
1761         }
1762
1763         for (fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START + n; fd++) {
1764
1765                 if (sd_is_socket_unix(fd, SOCK_DGRAM, -1, "/run/systemd/native", 0) > 0) {
1766
1767                         if (s->native_fd >= 0) {
1768                                 log_error("Too many native sockets passed.");
1769                                 return -EINVAL;
1770                         }
1771
1772                         s->native_fd = fd;
1773
1774                 } else if (sd_is_socket_unix(fd, SOCK_STREAM, 1, "/run/systemd/stdout", 0) > 0) {
1775
1776                         if (s->stdout_fd >= 0) {
1777                                 log_error("Too many stdout sockets passed.");
1778                                 return -EINVAL;
1779                         }
1780
1781                         s->stdout_fd = fd;
1782
1783                 } else if (sd_is_socket_unix(fd, SOCK_DGRAM, -1, "/dev/log", 0) > 0) {
1784
1785                         if (s->syslog_fd >= 0) {
1786                                 log_error("Too many /dev/log sockets passed.");
1787                                 return -EINVAL;
1788                         }
1789
1790                         s->syslog_fd = fd;
1791
1792                 } else {
1793                         log_error("Unknown socket passed.");
1794                         return -EINVAL;
1795                 }
1796         }
1797
1798         r = open_syslog_socket(s);
1799         if (r < 0)
1800                 return r;
1801
1802         r = open_native_socket(s);
1803         if (r < 0)
1804                 return r;
1805
1806         r = open_stdout_socket(s);
1807         if (r < 0)
1808                 return r;
1809
1810         r = system_journal_open(s);
1811         if (r < 0)
1812                 return r;
1813
1814         r = open_signalfd(s);
1815         if (r < 0)
1816                 return r;
1817
1818         s->rate_limit = journal_rate_limit_new(DEFAULT_RATE_LIMIT_INTERVAL, DEFAULT_RATE_LIMIT_BURST);
1819         if (!s->rate_limit)
1820                 return -ENOMEM;
1821
1822         return 0;
1823 }
1824
1825 static void server_done(Server *s) {
1826         JournalFile *f;
1827         assert(s);
1828
1829         while (s->stdout_streams)
1830                 stdout_stream_free(s->stdout_streams);
1831
1832         if (s->system_journal)
1833                 journal_file_close(s->system_journal);
1834
1835         if (s->runtime_journal)
1836                 journal_file_close(s->runtime_journal);
1837
1838         while ((f = hashmap_steal_first(s->user_journals)))
1839                 journal_file_close(f);
1840
1841         hashmap_free(s->user_journals);
1842
1843         if (s->epoll_fd >= 0)
1844                 close_nointr_nofail(s->epoll_fd);
1845
1846         if (s->signal_fd >= 0)
1847                 close_nointr_nofail(s->signal_fd);
1848
1849         if (s->syslog_fd >= 0)
1850                 close_nointr_nofail(s->syslog_fd);
1851
1852         if (s->native_fd >= 0)
1853                 close_nointr_nofail(s->native_fd);
1854
1855         if (s->stdout_fd >= 0)
1856                 close_nointr_nofail(s->stdout_fd);
1857
1858         if (s->rate_limit)
1859                 journal_rate_limit_free(s->rate_limit);
1860
1861         free(s->buffer);
1862 }
1863
1864 int main(int argc, char *argv[]) {
1865         Server server;
1866         int r;
1867
1868         /* if (getppid() != 1) { */
1869         /*         log_error("This program should be invoked by init only."); */
1870         /*         return EXIT_FAILURE; */
1871         /* } */
1872
1873         if (argc > 1) {
1874                 log_error("This program does not take arguments.");
1875                 return EXIT_FAILURE;
1876         }
1877
1878         log_set_target(LOG_TARGET_CONSOLE);
1879         log_parse_environment();
1880         log_open();
1881
1882         umask(0022);
1883
1884         r = server_init(&server);
1885         if (r < 0)
1886                 goto finish;
1887
1888         log_debug("systemd-journald running as pid %lu", (unsigned long) getpid());
1889
1890         sd_notify(false,
1891                   "READY=1\n"
1892                   "STATUS=Processing requests...");
1893
1894         server_vacuum(&server);
1895         server_flush_to_var(&server);
1896
1897         for (;;) {
1898                 struct epoll_event event;
1899
1900                 r = epoll_wait(server.epoll_fd, &event, 1, -1);
1901                 if (r < 0) {
1902
1903                         if (errno == EINTR)
1904                                 continue;
1905
1906                         log_error("epoll_wait() failed: %m");
1907                         r = -errno;
1908                         goto finish;
1909                 } else if (r == 0)
1910                         break;
1911
1912                 r = process_event(&server, &event);
1913                 if (r < 0)
1914                         goto finish;
1915                 else if (r == 0)
1916                         break;
1917         }
1918
1919         log_debug("systemd-journald stopped as pid %lu", (unsigned long) getpid());
1920
1921 finish:
1922         sd_notify(false,
1923                   "STATUS=Shutting down...");
1924
1925         server_done(&server);
1926
1927         return r < 0 ? EXIT_FAILURE : EXIT_SUCCESS;
1928 }