Merge pull request #15777 from Security-Onion-Solutions/feature/postgres

Postgres integration: SOC module config + Telegraf dual-write backend

salt/firewall/map.jinja

@@ -55,4 +55,16 @@
 
 {% endif %}
 
+{# Open Postgres (5432) to minion hostgroups when Telegraf is configured to write to Postgres #}
+{% set TG_OUT = (GLOBALS.telegraf_output | default('INFLUXDB')) | upper %}
+{% if TG_OUT in ['POSTGRES', 'BOTH'] %}
+{%   if role.startswith('manager') or role == 'standalone' or role == 'eval' %}
+{%     for r in ['sensor', 'searchnode', 'heavynode', 'receiver', 'fleet', 'idh', 'desktop', 'import'] %}
+{%       if FIREWALL_DEFAULT.firewall.role[role].chain["DOCKER-USER"].hostgroups[r] is defined %}
+{%         do FIREWALL_DEFAULT.firewall.role[role].chain["DOCKER-USER"].hostgroups[r].portgroups.append('postgres') %}
+{%       endif %}
+{%     endfor %}
+{%   endif %}
+{% endif %}
+
 {% set FIREWALL_MERGED = salt['pillar.get']('firewall', FIREWALL_DEFAULT.firewall, merge=True) %}

salt/global/defaults.yaml

@@ -1,3 +1,4 @@
 global:
   pcapengine: SURICATA
-  pipeline: REDIS
+  pipeline: REDIS
+  telegraf_output: BOTH

salt/global/soc_global.yaml

@@ -65,4 +65,15 @@ global:
     description: Allows use of Endgame with Security Onion. This feature requires a license from Endgame.
     global: True
     advanced: True
+  telegraf_output:
+    description: Selects the backend(s) Telegraf writes metrics to. INFLUXDB keeps the current behavior; POSTGRES writes to the grid's Postgres instance; BOTH dual-writes for migration validation.
+    regex: ^(INFLUXDB|POSTGRES|BOTH)$
+    options:
+      - INFLUXDB
+      - POSTGRES
+      - BOTH
+    regexFailureMessage: You must enter INFLUXDB, POSTGRES, or BOTH.
+    global: True
+    advanced: True
+    helpLink: influxdb
 

salt/manager/tools/sbin/soup

@@ -362,7 +362,8 @@ preupgrade_changes() {
     # This function is to add any new pillar items if needed.
     echo "Checking to see if changes are needed."
 
-    [[ "$INSTALLEDVERSION" =~ ^2\.4\.21[0-9]+$ ]] && up_to_3.0.0   
+    [[ "$INSTALLEDVERSION" =~ ^2\.4\.21[0-9]+$ ]] && up_to_3.0.0
+    [[ "$INSTALLEDVERSION" =~ ^3\.0\.[0-9]+$ ]] && up_to_3.1.0
     true
 }
 
@@ -371,6 +372,7 @@ postupgrade_changes() {
     echo "Running post upgrade processes."
 
     [[ "$POSTVERSION" =~ ^2\.4\.21[0-9]+$ ]] && post_to_3.0.0
+    [[ "$POSTVERSION" =~ ^3\.0\.[0-9]+$ ]] && post_to_3.1.0
     true
 }
 
@@ -469,6 +471,27 @@ post_to_3.0.0() {
 
 ### 3.0.0 End ###
 
+### 3.1.0 Start ###
+
+up_to_3.1.0() {
+  INSTALLEDVERSION=3.1.0
+}
+
+post_to_3.1.0() {
+  # Provision per-minion Telegraf Postgres users for every minion known to the
+  # manager. postgres.auth iterates manage.up to generate any missing passwords;
+  # postgres.telegraf_users reconciles the roles and schemas inside the so-postgres
+  # container. Then push a telegraf state to every minion so their telegraf.conf
+  # picks up the new credentials on the first apply after soup.
+  echo "Provisioning Telegraf Postgres users for existing minions."
+  salt-call --local state.apply postgres.auth postgres.telegraf_users || true
+  salt '*' state.sls telegraf || true
+
+  POSTVERSION=3.1.0
+}
+
+### 3.1.0 End ###
+
 repo_sync() {
   echo "Sync the local repo."
   su socore -c '/usr/sbin/so-repo-sync' || fail "Unable to complete so-repo-sync."

salt/orch/telegraf_postgres_sync.sls

@@ -0,0 +1,26 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{% set MINION = salt['pillar.get']('minion_id') %}
+{% set MANAGER = salt['pillar.get']('setup:manager') or salt['grains.get']('master') %}
+
+manager_sync_telegraf_pg_users:
+  salt.state:
+    - tgt: {{ MANAGER }}
+    - sls:
+      - postgres.auth
+      - postgres.telegraf_users
+    - queue: True
+
+{% if MINION and MINION != MANAGER %}
+{{ MINION }}_apply_telegraf:
+  salt.state:
+    - tgt: {{ MINION }}
+    - sls:
+      - telegraf
+    - queue: True
+    - require:
+      - salt: manager_sync_telegraf_pg_users
+{% endif %}

salt/postgres/auth.sls

@@ -13,6 +13,24 @@
   {% set CHARS = DIGITS~LOWERCASE~UPPERCASE~SYMBOLS %}
   {% set so_postgres_user_pass = salt['pillar.get']('postgres:auth:users:so_postgres_user:pass', salt['random.get_str'](72, chars=CHARS)) %}
 
+  {# Per-minion Telegraf Postgres credentials. Merge currently-up minions with any #}
+  {# previously-known entries in pillar so existing passwords persist across runs. #}
+  {% set existing = salt['pillar.get']('postgres:auth:users', {}) %}
+  {% set up_minions = salt['saltutil.runner']('manage.up') or [] %}
+  {% set telegraf_users = {} %}
+  {% for key, entry in existing.items() %}
+    {%- if key.startswith('telegraf_') and entry.get('user') and entry.get('pass') %}
+      {%- do telegraf_users.update({key: entry}) %}
+    {%- endif %}
+  {% endfor %}
+  {% for mid in up_minions %}
+    {%- set safe = mid | replace('.','_') | replace('-','_') | lower %}
+    {%- set key = 'telegraf_' ~ safe %}
+    {%- if key not in telegraf_users %}
+      {%- do telegraf_users.update({key: {'user': 'so_telegraf_' ~ safe, 'pass': salt['random.get_str'](72, chars=CHARS)}}) %}
+    {%- endif %}
+  {% endfor %}
+
 postgres_auth_pillar:
   file.managed:
     - name: /opt/so/saltstack/local/pillar/postgres/auth.sls
@@ -25,6 +43,11 @@ postgres_auth_pillar:
               so_postgres_user:
                 user: so_postgres
                 pass: "{{ so_postgres_user_pass }}"
+              {% for key, entry in telegraf_users.items() %}
+              {{ key }}:
+                user: {{ entry.user }}
+                pass: "{{ entry.pass }}"
+              {% endfor %}
     - show_changes: False
 {% else %}
 

salt/postgres/defaults.yaml

@@ -1,5 +1,7 @@
 postgres:
   enabled: True
+  telegraf:
+    retention_days: 14
   config:
     listen_addresses: '*'
     port: 5432

salt/postgres/enabled.sls

@@ -16,6 +16,7 @@ include:
   - postgres.ssl
   - postgres.config
   - postgres.sostatus
+  - postgres.telegraf_users
 
 so-postgres:
   docker_container.running:
@@ -79,6 +80,21 @@ delete_so-postgres_so-status.disabled:
     - name: /opt/so/conf/so-status/so-status.conf
     - regex: ^so-postgres$
 
+so_telegraf_trim:
+{% if GLOBALS.telegraf_output in ['POSTGRES', 'BOTH'] %}
+  cron.present:
+{% else %}
+  cron.absent:
+{% endif %}
+    - name: /usr/sbin/so-telegraf-trim >> /opt/so/log/postgres/telegraf-trim.log 2>&1
+    - identifier: so_telegraf_trim
+    - user: root
+    - minute: '17'
+    - hour: '3'
+    - daymonth: '*'
+    - month: '*'
+    - dayweek: '*'
+
 {% else %}
 
 {{sls}}_state_not_allowed:

salt/postgres/files/init-users.sh

@@ -16,3 +16,11 @@ psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-E
     \$\$;
     GRANT ALL PRIVILEGES ON DATABASE "$POSTGRES_DB" TO "$SO_POSTGRES_USER";
 EOSQL
+
+# Bootstrap the Telegraf metrics database. Per-minion roles + schemas are
+# reconciled on every state.apply by postgres/telegraf_users.sls; this block
+# only ensures the shared database exists on first initialization.
+psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL
+    SELECT 'CREATE DATABASE so_telegraf'
+    WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = 'so_telegraf')\gexec
+EOSQL

salt/postgres/soc_postgres.yaml

@@ -0,0 +1,7 @@
+postgres:
+  telegraf:
+    retention_days:
+      description: Number of days of Telegraf metrics to keep in the so_telegraf database. Older rows are deleted nightly by so-telegraf-trim.
+      forcedType: int
+      advanced: True
+      helpLink: influxdb

salt/postgres/telegraf_users.sls

@@ -0,0 +1,49 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{% from 'allowed_states.map.jinja' import allowed_states %}
+{% if sls.split('.')[0] in allowed_states %}
+{%   from 'vars/globals.map.jinja' import GLOBALS %}
+
+{% set TG_OUT = (GLOBALS.telegraf_output | default('INFLUXDB')) | upper %}
+{% if TG_OUT in ['POSTGRES', 'BOTH'] %}
+
+{%   set users = salt['pillar.get']('postgres:auth:users', {}) %}
+{%   for key, entry in users.items() %}
+{%     if key.startswith('telegraf_') and entry.get('user') and entry.get('pass') %}
+{%       set u = entry.user %}
+{%       set p = entry.pass | replace("'", "''") %}
+
+postgres_telegraf_role_{{ u }}:
+  cmd.run:
+    - name: |
+        docker exec -i so-postgres psql -v ON_ERROR_STOP=1 -U postgres -d so_telegraf <<'EOSQL'
+        DO $$
+        BEGIN
+            IF NOT EXISTS (SELECT FROM pg_catalog.pg_roles WHERE rolname = '{{ u }}') THEN
+                EXECUTE format('CREATE ROLE %I WITH LOGIN PASSWORD %L', '{{ u }}', '{{ p }}');
+            ELSE
+                EXECUTE format('ALTER ROLE %I WITH PASSWORD %L', '{{ u }}', '{{ p }}');
+            END IF;
+        END
+        $$;
+        GRANT CONNECT ON DATABASE so_telegraf TO "{{ u }}";
+        CREATE SCHEMA IF NOT EXISTS "{{ u }}" AUTHORIZATION "{{ u }}";
+        EOSQL
+    - require:
+      - docker_container: so-postgres
+
+{%     endif %}
+{%   endfor %}
+
+{% endif %}
+
+{% else %}
+
+{{sls}}_state_not_allowed:
+  test.fail_without_changes:
+    - name: {{sls}}_state_not_allowed
+
+{% endif %}

salt/postgres/tools/sbin/so-show-stats

@@ -0,0 +1,110 @@
+#!/bin/bash
+
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+# Point-in-time host metrics from the Telegraf Postgres backend.
+# Sanity-check tool for verifying metrics are landing before the grid
+# dashboards consume them.
+
+. /usr/sbin/so-common
+
+usage() {
+  cat <<EOF
+Usage: $0 [minion_id]
+
+Shows the most recent CPU, memory, disk, and load metrics for each minion
+from the so_telegraf Postgres database. Without an argument, reports on
+every minion that has data. With a minion_id, limits output to that one.
+
+Requires: sudo, so-postgres running, global.telegraf_output set to
+POSTGRES or BOTH.
+EOF
+  exit 1
+}
+
+if [ "$(id -u)" -ne 0 ]; then
+  echo "This script must be run using sudo!"
+  exit 1
+fi
+
+case "${1:-}" in
+  -h|--help) usage ;;
+esac
+
+FILTER_MINION="${1:-}"
+
+so_psql() {
+  docker exec so-postgres psql -U postgres -d so_telegraf -At -F $'\t' "$@"
+}
+
+if ! docker exec so-postgres psql -U postgres -lqt 2>/dev/null | cut -d\| -f1 | grep -qw so_telegraf; then
+  echo "Database so_telegraf not found. Is global.telegraf_output set to POSTGRES or BOTH?"
+  exit 2
+fi
+
+# List telegraf schemas (role-per-minion naming convention: so_telegraf_<sanitized_minion_id>)
+SCHEMAS=$(so_psql -c "SELECT schema_name FROM information_schema.schemata WHERE schema_name LIKE 'so_telegraf_%' ORDER BY schema_name;")
+
+if [ -z "$SCHEMAS" ]; then
+  echo "No minion schemas found in so_telegraf."
+  exit 0
+fi
+
+print_metric() {
+  local schema="$1" table="$2" query="$3"
+  # Confirm table exists in this schema before querying
+  local exists
+  exists=$(so_psql -c "SELECT 1 FROM information_schema.tables WHERE table_schema='${schema}' AND table_name='${table}' LIMIT 1;")
+  [ -z "$exists" ] && return 0
+  so_psql -c "$query"
+}
+
+for schema in $SCHEMAS; do
+  minion="${schema#so_telegraf_}"
+  if [ -n "$FILTER_MINION" ]; then
+    # Compare against the sanitized form used in schema names
+    want=$(echo "$FILTER_MINION" | tr '.-' '_' | tr '[:upper:]' '[:lower:]')
+    [ "$minion" != "$want" ] && continue
+  fi
+
+  echo "===================================================================="
+  echo " Minion: $minion"
+  echo "===================================================================="
+
+  print_metric "$schema" "cpu" "
+    SELECT 'cpu          ' AS metric,
+           to_char(time, 'YYYY-MM-DD HH24:MI:SS') AS ts,
+           round((100 - usage_idle)::numeric, 1) || '% used'
+    FROM \"${schema}\".cpu
+    WHERE cpu = 'cpu-total'
+    ORDER BY time DESC LIMIT 1;"
+
+  print_metric "$schema" "mem" "
+    SELECT 'memory       ' AS metric,
+           to_char(time, 'YYYY-MM-DD HH24:MI:SS') AS ts,
+           round(used_percent::numeric, 1) || '% used (' ||
+           pg_size_pretty(used) || ' of ' || pg_size_pretty(total) || ')'
+    FROM \"${schema}\".mem
+    ORDER BY time DESC LIMIT 1;"
+
+  print_metric "$schema" "disk" "
+    SELECT 'disk ' || rpad(path, 8) AS metric,
+           to_char(time, 'YYYY-MM-DD HH24:MI:SS') AS ts,
+           round(used_percent::numeric, 1) || '% used (' ||
+           pg_size_pretty(used) || ' of ' || pg_size_pretty(total) || ')'
+    FROM \"${schema}\".disk
+    WHERE time = (SELECT max(time) FROM \"${schema}\".disk)
+    ORDER BY path;"
+
+  print_metric "$schema" "system" "
+    SELECT 'load         ' AS metric,
+           to_char(time, 'YYYY-MM-DD HH24:MI:SS') AS ts,
+           load1 || ' / ' || load5 || ' / ' || load15 || ' (1/5/15m)'
+    FROM \"${schema}\".system
+    ORDER BY time DESC LIMIT 1;"
+
+  echo ""
+done

salt/postgres/tools/sbin/so-telegraf-trim

@@ -0,0 +1,103 @@
+#!/bin/bash
+
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+# Deletes Telegraf metric rows older than the configured retention window from
+# every minion schema in the so_telegraf database. Intended to run daily from
+# cron. Retention comes from pillar (postgres.telegraf.retention_days),
+# defaulting to 14 days. An explicit --days argument overrides the pillar.
+
+. /usr/sbin/so-common
+
+usage() {
+  cat <<EOF
+Usage: $0 [--days N] [--dry-run]
+
+  --days N    Override retention in days (default: pillar
+              postgres.telegraf.retention_days, fallback 14)
+  --dry-run   Report what would be deleted without modifying anything
+EOF
+  exit 1
+}
+
+if [ "$(id -u)" -ne 0 ]; then
+  echo "This script must be run using sudo!"
+  exit 1
+fi
+
+DAYS=""
+DRY_RUN=0
+while [ $# -gt 0 ]; do
+  case "$1" in
+    --days) DAYS="$2"; shift 2 ;;
+    --dry-run) DRY_RUN=1; shift ;;
+    -h|--help) usage ;;
+    *) usage ;;
+  esac
+done
+
+if [ -z "$DAYS" ]; then
+  DAYS=$(salt-call --local --out=newline_values_only pillar.get postgres:telegraf:retention_days 2>/dev/null)
+fi
+if ! [[ "$DAYS" =~ ^[0-9]+$ ]] || [ "$DAYS" -lt 1 ]; then
+  DAYS=14
+fi
+
+log() {
+  echo "$(date '+%Y-%m-%d %H:%M:%S') so-telegraf-trim: $*"
+}
+
+so_psql() {
+  docker exec so-postgres psql -U postgres -d so_telegraf -At -F $'\t' "$@"
+}
+
+if ! docker exec so-postgres psql -U postgres -lqt 2>/dev/null | cut -d\| -f1 | grep -qw so_telegraf; then
+  log "Database so_telegraf not present; nothing to trim."
+  exit 0
+fi
+
+log "Trimming rows older than ${DAYS} days (dry_run=${DRY_RUN})."
+
+TOTAL_DELETED=0
+
+# One row per (schema, table) we might want to trim.
+# Column name is 'time' for all telegraf output plugin tables; skip metadata
+# tables (tag_* used for tags_as_foreign_keys).
+ROWS=$(so_psql -c "
+  SELECT table_schema || '.' || table_name
+  FROM information_schema.columns
+  WHERE column_name = 'time'
+    AND data_type IN ('timestamp with time zone', 'timestamp without time zone')
+    AND table_schema LIKE 'so_telegraf_%'
+  ORDER BY 1;")
+
+if [ -z "$ROWS" ]; then
+  log "No telegraf metric tables found."
+  exit 0
+fi
+
+for qualified in $ROWS; do
+  if [ "$DRY_RUN" -eq 1 ]; then
+    count=$(so_psql -c "SELECT count(*) FROM \"${qualified%.*}\".\"${qualified#*.}\" WHERE time < now() - interval '${DAYS} days';")
+    log "would delete ${count:-0} rows from ${qualified}"
+  else
+    # RETURNING count via a CTE so we can log how much was trimmed per table
+    deleted=$(so_psql -c "
+      WITH d AS (
+        DELETE FROM \"${qualified%.*}\".\"${qualified#*.}\"
+        WHERE time < now() - interval '${DAYS} days'
+        RETURNING 1
+      )
+      SELECT count(*) FROM d;")
+    deleted=${deleted:-0}
+    TOTAL_DELETED=$((TOTAL_DELETED + deleted))
+    [ "$deleted" -gt 0 ] && log "deleted ${deleted} rows from ${qualified}"
+  fi
+done
+
+if [ "$DRY_RUN" -eq 0 ]; then
+  log "Trim complete. Total rows deleted: ${TOTAL_DELETED}."
+fi

salt/reactor/telegraf_user_sync.sls

@@ -0,0 +1,18 @@
+# Copyright Security Onion Solutions LLC and/or licensed to Security Onion Solutions LLC under one
+# or more contributor license agreements. Licensed under the Elastic License 2.0 as shown at
+# https://securityonion.net/license; you may not use this file except in compliance with the
+# Elastic License 2.0.
+
+{# Fires on salt/auth. Only act on accepted keys — ignore pending/reject. #}
+{% if data.get('act') == 'accept' and data.get('id') %}
+
+{{ data['id'] }}_telegraf_pg_sync:
+  runner.state.orchestrate:
+    - args:
+      - mods: orch.telegraf_postgres_sync
+      - pillar:
+          minion_id: {{ data['id'] }}
+
+{% do salt.log.info('telegraf_user_sync reactor: syncing telegraf PG user for minion %s' % data['id']) %}
+
+{% endif %}

salt/salt/master.sls

@@ -62,6 +62,19 @@ engines_config:
     - name: /etc/salt/master.d/engines.conf
     - source: salt://salt/files/engines.conf
 
+reactor_config_telegraf:
+  file.managed:
+    - name: /etc/salt/master.d/reactor_telegraf.conf
+    - contents: |
+        reactor:
+          - 'salt/auth':
+            - /opt/so/saltstack/default/salt/reactor/telegraf_user_sync.sls
+    - user: root
+    - group: root
+    - mode: 644
+    - watch_in:
+      - service: salt_master_service
+
 # update the bootstrap script when used for salt-cloud
 salt_bootstrap_cloud:
   file.managed:

salt/telegraf/etc/telegraf.conf

@@ -8,6 +8,11 @@
 {%- set ZEEK_ENABLED = salt['pillar.get']('zeek:enabled', True) %}
 {%- set MDENGINE = GLOBALS.md_engine %}
 {%- set LOGSTASH_ENABLED = LOGSTASH_MERGED.enabled %}
+{%- set TG_OUT = GLOBALS.telegraf_output | upper %}
+{%- set PG_HOST = GLOBALS.manager_ip %}
+{%- set PG_SAFE = GLOBALS.minion_id | replace('.','_') | replace('-','_') | lower %}
+{%- set PG_USER = 'so_telegraf_' ~ PG_SAFE %}
+{%- set PG_PASS = salt['pillar.get']('postgres:auth:users:telegraf_' ~ PG_SAFE ~ ':pass', '') %}
 # Global tags can be specified here in key="value" format.
 [global_tags]
   role = "{{ GLOBALS.role.split('-') | last }}"
@@ -72,6 +77,7 @@
 #                            OUTPUT PLUGINS                                   #
 ###############################################################################
 
+{%- if TG_OUT in ['INFLUXDB', 'BOTH'] %}
 # Configuration for sending metrics to InfluxDB
 [[outputs.influxdb_v2]]
   urls = ["https://{{ INFLUXDBHOST }}:8086"]
@@ -85,6 +91,15 @@
    tls_key = "/etc/telegraf/telegraf.key"
   ## Use TLS but skip chain & host verification
   # insecure_skip_verify = false
+{%- endif %}
+
+{%- if TG_OUT in ['POSTGRES', 'BOTH'] %}
+# Configuration for sending metrics to PostgreSQL
+[[outputs.postgresql]]
+  connection = "host={{ PG_HOST }} port=5432 user={{ PG_USER }} password={{ PG_PASS }} dbname=so_telegraf sslmode=verify-full sslrootcert=/etc/telegraf/ca.crt"
+  schema = "{{ PG_USER }}"
+  tags_as_foreign_keys = true
+{%- endif %}
 
 ###############################################################################
 #                            PROCESSOR PLUGINS                                #

salt/vars/globals.map.jinja

@@ -24,6 +24,7 @@
     'md_engine': INIT.PILLAR.global.mdengine,
     'pcap_engine': GLOBALMERGED.pcapengine,
     'pipeline': GLOBALMERGED.pipeline,
+    'telegraf_output': GLOBALMERGED.telegraf_output,
     'so_version': INIT.PILLAR.global.soversion,
     'so_docker_gateway': DOCKERMERGED.gateway,
     'so_docker_range': DOCKERMERGED.range,