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