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