# Journal entries pulled from www.azit.se (see www-logpull.yaml), one JSON object each.
# Fields from journalctl -o json: __CURSOR, __REALTIME_TIMESTAMP (microseconds), MESSAGE,
# PRIORITY, SYSLOG_IDENTIFIER, _PID, _COMM, _SYSTEMD_UNIT, _UID, _TRANSPORT.
j = .
# Start from an empty event whose objects already exist: VRL only lets a field be set
# inside a parent it knows is an object. compact() at the end drops the ones left empty.
. = {"event": {}, "host": {}, "process": {}, "log": {"syslog": {}}, "systemd": {}, "source": {}, "url": {},
     "http": {"request": {}, "response": {"body": {}}}, "user_agent": {}, "tls": {}, "ssh": {"key": {}},
     "user": {"target": {}}}

# The cursor identifies the entry: as the document ID, a repeated pull overwrites instead of duplicating
.event.id = sha2(string(j.__CURSOR) ?? "")
."@timestamp" = from_unix_timestamp(to_int(j.__REALTIME_TIMESTAMP) ?? 0, unit: "microseconds") ?? now()
.host.name = "www.azit.se"
.message = string(j.MESSAGE) ?? "(not text)"
.process.name = string(j.SYSLOG_IDENTIFIER) ?? string(j._COMM) ?? null
.process.pid = to_int(j._PID) ?? null
.log.syslog.priority = to_int(j.PRIORITY) ?? null
.systemd.unit = string(j._SYSTEMD_UNIT) ?? null

if .process.name == "nginx_access" {
  # one request, written by nginx as JSON (log_format json_www in deploy/nginx.conf)
  .event.dataset = "nginx.access"
  .event.category = "web"
  r, err = parse_json(.message)
  if err == null && is_object(r) {
    .source.ip = r.client
    .url.domain = r.host
    .url.original = r.uri
    .http.request.method = r.method
    .http.version = r.protocol
    .http.response.status_code = to_int(r.status) ?? null
    .http.response.body.bytes = to_int(r.bytes) ?? null
    if r.referer != "" { .http.request.referrer = r.referer }
    .user_agent.original = r.agent
    if r.tls != "" { .tls.version = r.tls }
    .event.duration = to_int((to_float(r.duration) ?? 0.0) * 1000000000.0)
    .event.outcome = if (to_int(r.status) ?? 0) < 400 { "success" } else { "failure" }
  } else {
    .tags = ["nginx_parse_failure"]
  }
} else {
  .event.dataset = "linux.system"
  m = .message
  if includes(["sshd", "sshd-session", "sshd-auth"], .process.name) {
    s, e1 = parse_regex(m, r'^(?P<outcome>Accepted|Failed) (?P<method>\S+) for (?:invalid user )?(?P<user>\S+) from (?P<ip>[0-9a-fA-F.:]+) port (?P<port>\d+)(?: ssh2)?(?:: (?P<keytype>\S+) (?P<fp>SHA256:\S+))?')
    if e1 == null {
      .event.category = "authentication"
      .event.outcome = if s.outcome == "Accepted" { "success" } else { "failure" }
      .user.name = s.user
      .source.ip = s.ip
      .source.port = to_int(s.port) ?? null
      .ssh.method = s.method
      if s.fp != null && s.fp != "" { .ssh.key.fingerprint = s.fp; .ssh.key.type = s.keytype }
    }
    f1, e2 = parse_regex(m, r'^Invalid user (?P<user>\S*) from (?P<ip>[0-9a-fA-F.:]+) port')
    if e2 == null { .event.category = "authentication"; .event.outcome = "failure"; .user.name = f1.user; .source.ip = f1.ip }
    f2, e3 = parse_regex(m, r'^Connection closed by authenticating user (?P<user>\S+) (?P<ip>[0-9a-fA-F.:]+) port .*\[preauth\]')
    if e3 == null { .event.category = "authentication"; .event.outcome = "failure"; .user.name = f2.user; .source.ip = f2.ip }
    f3, e4 = parse_regex(m, r'^User (?P<user>\S+) from (?P<ip>[0-9a-fA-F.:]+) not allowed')
    if e4 == null { .event.category = "authentication"; .event.outcome = "failure"; .user.name = f3.user; .source.ip = f3.ip }
    # the pull itself logs in every 15 minutes: tagged, so dashboards and alerts can leave it out
    if .user.name == "logpull" { .tags = ["logpull"] }
  } else if .process.name == "sudo" {
    # "admin : TTY=pts/0 ; PWD=/home/admin ; USER=root ; COMMAND=/usr/bin/apt update"
    u, e5 = parse_regex(m, r'^\s*(?P<user>\S+) : .*?USER=(?P<as>\S+) ; COMMAND=(?P<cmd>.*)$')
    if e5 == null {
      .event.category = "process"
      .user.name = u.user
      .user.target.name = u.as
      .process.command_line = u.cmd
    }
  }
}

. = compact(.)