Its logs, fetched from home

#!/bin/sh
# Creates or extends OpenSearch's users and passwords, encrypted with SOPS straight away:
# no password and no plaintext file ever lands on disk.
#   apps/logging/opensearch-security.sops.yaml            users, roles and role mappings
#   apps/logging/opensearch-dashboards-account.sops.yaml  the account Dashboards logs in with
#   apps/logging/vector-opensearch.sops.yaml              the write-only account Vector uses
#   apps/falco/falco-opensearch.sops.yaml                 the write-only account Falcosidekick uses
# Safe to run again: stored passwords and hashes are kept, only missing ones are generated,
# and a file is rewritten only when its decrypted content changes.
# It prints the admin password only when it creates one: put it in your password manager.
set -eu
cd "$(dirname "$0")/.."
SEC=apps/logging/opensearch-security.sops.yaml
DASH=apps/logging/opensearch-dashboards-account.sops.yaml
VEC=apps/logging/vector-opensearch.sops.yaml
FALCO=apps/falco/falco-opensearch.sops.yaml
# 1. What already exists. A failed decryption stops the script (set -e): it must never
#    fall through to "missing" and silently generate new passwords.
decrypt() { [ -f "$1" ] || return 0; sops -d "$1"; }
SEC_PLAIN=$(decrypt "$SEC")
DASH_PLAIN=$(decrypt "$DASH")
VEC_PLAIN=$(decrypt "$VEC")
FALCO_PLAIN=$(decrypt "$FALCO")
# field PLAINTEXT KEY: one decoded value from a decrypted Secret, empty if absent
field() { printf '%s\n' "$1" | yq ".data.\"$2\" // \"\"" | base64 --decode; }
USERS_PLAIN=$(field "$SEC_PLAIN" internal_users.yml)
# old_hash USER: that user's stored bcrypt hash, empty if absent
old_hash() { printf '%s\n' "$USERS_PLAIN" | yq ".\"$1\".hash // \"\""; }
hash() { htpasswd -bnBC 12 "" "$1" | tr -d ':\n'; }
# 2. Passwords and hashes: keep what is stored, generate what is missing.
#    A newly generated password always gets a new hash, so the two never disagree.
NEW_ADMIN_PW=""
ADMIN_HASH=$(old_hash admin)
if [ -z "$ADMIN_HASH" ]; then
  NEW_ADMIN_PW=$(openssl rand -base64 24 | tr -d '/+=')
  ADMIN_HASH=$(hash "$NEW_ADMIN_PW")
fi
DASH_PW=$(field "$DASH_PLAIN" password)
if [ -z "$DASH_PW" ]; then DASH_PW=$(openssl rand -hex 24); DASH_HASH=$(hash "$DASH_PW")
else DASH_HASH=$(old_hash kibanaserver); [ -n "$DASH_HASH" ] || DASH_HASH=$(hash "$DASH_PW"); fi
VECTOR_PW=$(field "$VEC_PLAIN" password)
if [ -z "$VECTOR_PW" ]; then VECTOR_PW=$(openssl rand -hex 24); VECTOR_HASH=$(hash "$VECTOR_PW")
else VECTOR_HASH=$(old_hash vector); [ -n "$VECTOR_HASH" ] || VECTOR_HASH=$(hash "$VECTOR_PW"); fi
FALCO_PW=$(field "$FALCO_PLAIN" ELASTICSEARCH_PASSWORD)
if [ -z "$FALCO_PW" ]; then FALCO_PW=$(openssl rand -hex 24); FALCO_HASH=$(hash "$FALCO_PW")
else FALCO_HASH=$(old_hash falco); [ -n "$FALCO_HASH" ] || FALCO_HASH=$(hash "$FALCO_PW"); fi
COOKIE=$(field "$DASH_PLAIN" cookie)
[ -n "$COOKIE" ] || COOKIE=$(openssl rand -hex 32)
# 3. OpenSearch's security configuration
INTERNAL_USERS=$(cat <<YAML
_meta:
  type: "internalusers"
  config_version: 2
admin:
  hash: "$ADMIN_HASH"
  reserved: true
  backend_roles:
    - "admin"
  description: "Human administrator"
kibanaserver:
  hash: "$DASH_HASH"
  reserved: true
  description: "OpenSearch Dashboards server"
vector:
  hash: "$VECTOR_HASH"
  reserved: false
  description: "Vector log shipper, write only"
falco:
  hash: "$FALCO_HASH"
  reserved: false
  description: "Falcosidekick, write only"
YAML
)
ROLES=$(cat <<'YAML'
_meta:
  type: "roles"
  config_version: 2
vector_writer:
  reserved: false
  cluster_permissions:
    - "cluster_monitor"
    - "indices:data/write/bulk"
  index_permissions:
    - index_patterns:
        - "fw-*"
        - "unifi-*"
        - "web-*"
      allowed_actions:
        - "create_index"
        - "write"
        - "indices:admin/mapping/put"
falco_writer:
  reserved: false
  cluster_permissions:
    - "cluster_monitor"
    - "indices:data/write/bulk"
  index_permissions:
    - index_patterns:
        - "falco-*"
      allowed_actions:
        - "create_index"
        - "write"
        - "indices:admin/mapping/put"
YAML
)
ROLES_MAPPING=$(cat <<'YAML'
_meta:
  type: "rolesmapping"
  config_version: 2
all_access:
  reserved: false
  backend_roles:
    - "admin"
kibana_server:
  reserved: true
  users:
    - "kibanaserver"
own_index:
  reserved: false
  users:
    - "*"
vector_writer:
  reserved: false
  users:
    - "vector"
falco_writer:
  reserved: false
  users:
    - "falco"
YAML
)
# 4. Into SOPS, but only where something changed. SOPS encrypts with a fresh data key
#    every time, so rewriting an unchanged secret would still change the file in Git.
norm() { yq -o=json -I=0 'sort_keys(..)'; }
put() {
  if [ -f "$1" ] && [ "$(sops -d "$1" | norm)" = "$(printf '%s\n' "$2" | norm)" ]; then
    echo "unchanged  $1"; return 0
  fi
  ENC=$(printf '%s\n' "$2" | sops --encrypt --filename-override "$1" \
        --input-type yaml --output-type yaml /dev/stdin)
  mkdir -p "$(dirname "$1")"
  printf '%s\n' "$ENC" > "$1"
  echo "written    $1"
}
put "$SEC" "$(kubectl create secret generic opensearch-security -n logging \
  --from-literal=internal_users.yml="$INTERNAL_USERS" \
  --from-literal=roles.yml="$ROLES" \
  --from-literal=roles_mapping.yml="$ROLES_MAPPING" \
  --dry-run=client -o yaml)"
put "$DASH" "$(kubectl create secret generic opensearch-dashboards-account -n logging \
  --from-literal=username=kibanaserver --from-literal=password="$DASH_PW" \
  --from-literal=cookie="$COOKIE" \
  --dry-run=client -o yaml)"
put "$VEC" "$(kubectl create secret generic vector-opensearch -n logging \
  --from-literal=username=vector --from-literal=password="$VECTOR_PW" \
  --dry-run=client -o yaml)"
put "$FALCO" "$(kubectl create secret generic falco-opensearch -n falco \
  --from-literal=ELASTICSEARCH_USERNAME=falco --from-literal=ELASTICSEARCH_PASSWORD="$FALCO_PW" \
  --dry-run=client -o yaml)"
# 5. The one password you need, only when it was just created
if [ -n "$NEW_ADMIN_PW" ]; then
  echo "New OpenSearch admin password: $NEW_ADMIN_PW"
  echo "Put it in your password manager now."
fi

# Pulls the web server's journal every 15 minutes and hands it to Vector.
# On www.azit.se, the key can only run /usr/local/sbin/logpull-journal (see homelab-site, deploy/DEPLOY.md);
# the server has no way to connect here.
apiVersion: v1
kind: Service
metadata:
  name: www-logpull-in
  namespace: logging
spec:
  type: ClusterIP                 # inside the cluster only, unlike Vector's syslog LoadBalancer
  selector:
    app.kubernetes.io/instance: vector
    app.kubernetes.io/component: Stateless-Aggregator
  ports:
    - name: www-pull
      port: 8090
      targetPort: 8090
      protocol: TCP
---
apiVersion: batch/v1
kind: CronJob
metadata:
  name: www-logpull
  namespace: logging
spec:
  schedule: "*/15 * * * *"
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 1
  failedJobsHistoryLimit: 3
  jobTemplate:
    spec:
      backoffLimit: 0             # no blind retry: the next run picks up from the server's cursor
      activeDeadlineSeconds: 600
      template:
        spec:
          restartPolicy: Never
          automountServiceAccountToken: false
          securityContext:
            seccompProfile: { type: RuntimeDefault }
          volumes:
            - name: keys
              secret: { secretName: www-logpull, defaultMode: 0400 }
          containers:
            - name: pull
              image: alpine:3.22
              securityContext: { allowPrivilegeEscalation: false }
              resources:
                requests: { cpu: 10m, memory: 32Mi }
                limits:   { cpu: 500m, memory: 128Mi }
              volumeMounts:
                - { name: keys, mountPath: /keys, readOnly: true }
              command: ["/bin/sh", "-euc"]
              args:
                - |
                  apk add --no-cache --quiet openssh-client-default curl
                  ssh -T -i /keys/id_ed25519 -o IdentitiesOnly=yes -o BatchMode=yes \
                      -o UserKnownHostsFile=/keys/known_hosts -o StrictHostKeyChecking=yes \
                      -o ConnectTimeout=20 logpull@www.azit.se > /tmp/batch.jsonl
                  echo "pulled $(wc -l < /tmp/batch.jsonl) entries"
                  [ -s /tmp/batch.jsonl ] || exit 0
                  curl -sS --fail --retry 5 --retry-all-errors --retry-delay 10 \
                       -H 'Content-Type: application/x-ndjson' --data-binary @/tmp/batch.jsonl \
                       http://www-logpull-in.logging.svc:8090/
                  echo "handed to Vector"

# 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(.)

apiVersion: source.toolkit.fluxcd.io/v1
kind: HelmRepository
metadata:
  name: vector
  namespace: logging
spec:
  interval: 24h
  url: https://helm.vector.dev
---
apiVersion: helm.toolkit.fluxcd.io/v2
kind: HelmRelease
metadata:
  name: vector
  namespace: logging
spec:
  interval: 30m
  dependsOn:
    - name: opensearch
  chart:
    spec:
      chart: vector
      version: "0.59.0"
      sourceRef:
        kind: HelmRepository
        name: vector
  values:
    role: Stateless-Aggregator
    replicas: 1
    resources:
      requests: { cpu: 100m, memory: 128Mi }
      limits:   { cpu: "1", memory: 512Mi }
    extraVolumes:
      - name: os-creds
        secret: { secretName: vector-opensearch, defaultMode: 0400 }
      - name: syslog-tls
        secret: { secretName: vector-syslog-tls, defaultMode: 0400 }
      - name: homelab-ca
        configMap: { name: homelab-ca }
      - name: hosts
        configMap: { name: vector-hosts }
      - name: www-vrl
        configMap: { name: vector-www-vrl }
    extraVolumeMounts:
      - { name: os-creds,   mountPath: /etc/vector-secrets, readOnly: true }
      - { name: syslog-tls, mountPath: /etc/vector-tls,     readOnly: true }
      - { name: homelab-ca, mountPath: /etc/homelab-ca,     readOnly: true }
      - { name: hosts,      mountPath: /etc/vector-hosts,  readOnly: true }
      - { name: www-vrl,    mountPath: /etc/vector-vrl,    readOnly: true }
    containerPorts:
      - { name: syslog-tls, containerPort: 6514, protocol: TCP }
      - { name: api,        containerPort: 8686, protocol: TCP }
    service:
      enabled: true
      type: LoadBalancer
      externalTrafficPolicy: Local
      annotations:
        metallb.io/loadBalancerIPs: 192.168.5.200
      ports:
        - { name: syslog-tls, port: 6514, targetPort: 6514, protocol: TCP }
    customConfig:
      data_dir: /vector-data-dir
      api:
        enabled: true
        address: 0.0.0.0:8686
      log_schema:
        timestamp_key: "@timestamp"
      secret:
        os:
          type: directory
          path: /etc/vector-secrets
      enrichment_tables:
        hosts:
          type: file
          file:
            path: /etc/vector-hosts/hosts.csv
            encoding:
              type: csv
          schema:
            ip: string
            hostname: string
      sources:
        firewall:
          type: syslog
          mode: tcp
          address: 0.0.0.0:6514
          tls:
            enabled: true
            crt_file: /etc/vector-tls/tls.crt
            key_file: /etc/vector-tls/tls.key
            ca_file: /etc/homelab-ca/ca.crt
            verify_certificate: true
        # journal entries from www.azit.se, posted by the www-logpull CronJob (www-logpull.yaml)
        www_pull:
          type: http_server
          address: 0.0.0.0:8090
          method: POST
          framing: { method: newline_delimited }
          decoding: { codec: json }
      transforms:
        parse_firewall:
          type: remap
          inputs: [firewall]
          source: |
            del(.host)
            del(.source_type)
            del(.version)
            del(.facility)
            del(.severity)
            del(.procid)
            .observer.hostname = del(.hostname)
            .observer.ip = del(.source_ip)
            .process.name = del(.appname)
            if .process.name == "pf" {
              .log_type = "pf"
              .event.dataset = "openbsd.pf"
              p, err = parse_regex(string!(.message), r'rule (?P<rule>[\d.]+)/\((?P<reason>[^)]*)\) (?P<action>\w+) (?P<direction>in|out) on (?P<iface>\S+): (?P<src_ip>\d+\.\d+\.\d+\.\d+)(?:\.(?P<src_port>\d+))? > (?P<dst_ip>\d+\.\d+\.\d+\.\d+)(?:\.(?P<dst_port>\d+))?: (?P<rest>.*?)(?: label=(?P<label>\S+))?$')
              if err == null {
                .rule.id = p.rule
                if p.label != null && p.label != "" { .rule.name = p.label }
                .event.reason = p.reason
                .event.action = p.action
                .network.direction = p.direction
                .source.ip = p.src_ip
                .destination.ip = p.dst_ip
                src_host, src_err = get_enrichment_table_record("hosts", {"ip": .source.ip})
                if src_err == null { .source.hostname = src_host.hostname } else if starts_with(string(.source.ip) ?? "", "192.168.") { .source.hostname = "unknown" }
                dst_host, dst_err = get_enrichment_table_record("hosts", {"ip": .destination.ip})
                if dst_err == null { .destination.hostname = dst_host.hostname } else if starts_with(string(.destination.ip) ?? "", "192.168.") { .destination.hostname = "unknown" }
                if p.src_port != null && p.src_port != "" { .source.port = to_int(p.src_port) ?? null }
                if p.dst_port != null && p.dst_port != "" { .destination.port = to_int(p.dst_port) ?? null }
                rest = string(p.rest) ?? ""
                if starts_with(rest, "udp") {
                  .network.transport = "udp"
                } else if starts_with(rest, "icmp") {
                  .network.transport = "icmp"
                } else if .destination.port != null {
                  .network.transport = "tcp"
                }
                ifname = string(p.iface) ?? ""
                .observer.ingress.interface.name = ifname
                aliases = {
                  "igc0": "WAN", "igc1": "ADMIN", "igc2": "APMGMT", "igc3": "PS5",
                  "igc4": "MEDIA", "igc5": "SERVERS", "vlan10": "TRUSTED", "vlan20": "IOT", "vlan30": "GUEST", "wg0": "VPN"
                }
                alias = get(aliases, [ifname]) ?? null
                if is_null(alias) { alias = ifname }
                .observer.ingress.interface.alias = alias
              } else {
                .tags = ["pf_parse_failure"]
              }
            } else {
              .log_type = "system"
              .event.dataset = "openbsd.system"
              if includes(["sshd", "sshd-session", "sshd-auth"], .process.name) {
                m = string!(.message)
                s, e1 = parse_regex(m, r'^(?P<outcome>Accepted|Failed) (?P<method>\S+) for (?:invalid user )?(?P<user>\S+) from (?P<ip>[0-9.]+) 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-9.]+) 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-9.]+) 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-9.]+) not allowed')
                if e4 == null { .event.category = "authentication"; .event.outcome = "failure"; .user.name = f3.user; .source.ip = f3.ip }
                if exists(.source.ip) {
                  sh, she = get_enrichment_table_record("hosts", {"ip": .source.ip})
                  if she == null { .source.hostname = sh.hostname }
                }
              }
            }
        parse_www:
          type: remap
          inputs: [www_pull]
          file: /etc/vector-vrl/parse-www.vrl
      sinks:
        opensearch:
          type: elasticsearch
          inputs: [parse_firewall]
          endpoints: ["https://opensearch-cluster-master.logging.svc:9200"]
          api_version: v8
          mode: bulk
          bulk:
            index: 'fw-{{ "{{" }} log_type {{ "}}" }}-%Y.%m.%d'
          encoding:
            except_fields: ["log_type"]
          auth:
            strategy: basic
            user: "SECRET[os.username]"
            password: "SECRET[os.password]"
          tls:
            ca_file: /etc/homelab-ca/ca.crt
            verify_certificate: true
            verify_hostname: true
        opensearch_www:
          type: elasticsearch
          inputs: [parse_www]
          endpoints: ["https://opensearch-cluster-master.logging.svc:9200"]
          api_version: v8
          mode: bulk
          bulk:
            index: "web-%Y.%m.%d"
          id_key: event.id          # the hashed journal cursor: a repeated pull overwrites, never duplicates
          auth:
            strategy: basic
            user: "SECRET[os.username]"
            password: "SECRET[os.password]"
          tls:
            ca_file: /etc/homelab-ca/ca.crt
            verify_certificate: true
            verify_hostname: true