#!/usr/bin/env bash
#
# soc-healthcheck-node.sh — node-side collector for the UK Cyber Defence estate.
#
# Deployed to /usr/local/sbin/soc-healthcheck-node on every monitored host by the
# jump-box orchestrator (soc-healthcheck.sh). Executed via a tightly scoped
# sudoers entry — it is the ONLY thing the monitoring account may run as root.
#
# DESIGN RULE: this script EMITS FACTS AND NOTHING ELSE. It applies no
# thresholds and makes no judgements. All evaluation happens centrally on the
# jump box so that tuning lives in one file rather than nineteen.
#
# Output format: one record per line, "key<TAB>value". Keys are namespaced.
# Values are sanitised so they can never contain a tab or a newline.
#
# It is strictly READ-ONLY. It changes nothing on the host.
#
# Usage: soc-healthcheck-node <light|deep>
#
# Emily (EmilyAI) — UK Cyber Defence. All timestamps UTC.

set -uo pipefail
# NOTE: IFS is left at the shell default (space, tab, newline) ON PURPOSE.
# The hardening idiom IFS=$'\n\t' silently breaks every space-separated
# `read` in this file — /proc/loadavg and df output among them — so load
# reports 0.00 on every host and no filesystem is reported at all. Both
# failures read as perfectly healthy, which is the single worst outcome this
# suite can produce. Every expansion below is quoted instead.
IFS=$' \t\n'
umask 077
export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
export LC_ALL=C
export DEBIAN_FRONTEND=noninteractive

MODE="${1:-light}"
case "$MODE" in
    light|deep) ;;
    *) printf 'node.error\tinvalid mode: %s\n' "$MODE"; exit 2 ;;
esac

readonly OSSEC_DIR=/var/ossec
readonly GRAYLOG_TOKEN_FILE=/etc/soc-healthcheck/graylog.token
readonly FLUENTBIT_LOG=/var/log/td-agent-bit.log
readonly START_EPOCH=$(date -u +%s)

# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------

# emit <key> <value...> — sanitise and print one fact.
emit() {
    local key="$1"; shift
    local val="${*:-}"
    # Strip control characters, collapse whitespace, cap length. A value can
    # never break the record format.
    val="$(printf '%s' "$val" | tr -d '\000-\010\013\014\016-\037' | tr '\t\n\r' '   ')"
    val="${val#"${val%%[![:space:]]*}"}"
    val="${val%"${val##*[![:space:]]}"}"
    printf '%s\t%s\n' "$key" "${val:0:2000}"
}

# run <timeout_secs> <command...> — never let a check hang the pass.
run() { local t="$1"; shift; timeout --kill-after=2 "$t" "$@" 2>/dev/null; }

have() { command -v "$1" >/dev/null 2>&1; }

# keysafe <string> — make a mount point or unit name usable inside a key.
# '/' would otherwise reduce to an empty string and produce the key
# "os.disk..used_pct", so the root filesystem is named explicitly.
keysafe() {
    local s
    [[ "$1" == "/" ]] && { printf 'root'; return; }
    s="$(printf '%s' "$1" | tr -c 'A-Za-z0-9._-' '_' | sed 's/__*/_/g;s/^_//;s/_$//')"
    printf '%s' "${s:-root}"
}

# num <value> <default> — guarantee a numeric fact.
num() { case "${1:-}" in ''|*[!0-9.-]*) printf '%s' "${2:-0}" ;; *) printf '%s' "$1" ;; esac; }

# ---------------------------------------------------------------------------
# 1. identity and platform
# ---------------------------------------------------------------------------

emit node.schema 3
emit node.mode "$MODE"
emit node.collected_at "$(date -u +%Y-%m-%dT%H:%M:%SZ)"
emit node.hostname "$(hostname -f 2>/dev/null || hostname)"
emit node.script_version "1.0.0"

if [[ -r /etc/os-release ]]; then
    # shellcheck disable=SC1091
    . /etc/os-release
    emit os.id "${ID:-unknown}"
    emit os.version "${VERSION_ID:-unknown}"
    emit os.pretty "${PRETTY_NAME:-unknown}"
fi
emit os.kernel "$(uname -r)"
emit os.arch "$(uname -m)"
emit os.virt "$(run 5 systemd-detect-virt 2>/dev/null || echo unknown)"

# Uptime and boot time. A host that has rebooted since the previous pass is a
# finding in its own right — the orchestrator diffs boot_epoch across runs.
if [[ -r /proc/uptime ]]; then
    up_s="$(cut -d. -f1 /proc/uptime)"
    emit os.uptime_seconds "$(num "$up_s" 0)"
    emit os.boot_epoch "$(( START_EPOCH - $(num "$up_s" 0) ))"
fi

# ---------------------------------------------------------------------------
# 2. core OS health
# ---------------------------------------------------------------------------

# Load, normalised per core. Raw load on a 32-core box means nothing.
IFS=' ' read -r l1 l5 l15 _ < /proc/loadavg
cores="$(nproc 2>/dev/null || echo 1)"
emit os.cpu.cores "$cores"
emit os.load.1 "$(num "$l1" 0)"
emit os.load.5 "$(num "$l5" 0)"
emit os.load.15 "$(num "$l15" 0)"
emit os.load.per_core_5 "$(awk -v l="$l5" -v c="$cores" 'BEGIN{printf "%.2f", (c>0? l/c : l)}')"

# Memory and swap, from /proc/meminfo so we do not depend on `free` formatting.
if [[ -r /proc/meminfo ]]; then
    awk '
        /^MemTotal:/     {t=$2}
        /^MemAvailable:/ {a=$2}
        /^SwapTotal:/    {st=$2}
        /^SwapFree:/     {sf=$2}
        END{
            if (t>0) {
                printf "os.mem.total_kb\t%d\n", t
                printf "os.mem.available_kb\t%d\n", a
                printf "os.mem.used_pct\t%.1f\n", (t-a)*100/t
            }
            if (st>0) printf "os.swap.used_pct\t%.1f\n", (st-sf)*100/st
            else      printf "os.swap.used_pct\t0\n"
        }' /proc/meminfo
fi

# Filesystems — real ones only. Emitted per mount so the orchestrator can name
# the offending mount rather than saying "a disk is full".
disk_seen=0
while IFS=' ' read -r _fs _type size used avail pct mnt; do
    [[ -z "${mnt:-}" ]] && continue
    disk_seen=$(( disk_seen + 1 ))
    k="$(keysafe "$mnt")"
    emit "os.disk.${k}.used_pct" "${pct%\%}"
    emit "os.disk.${k}.avail_kb" "$(num "$avail" 0)"
    emit "os.disk.${k}.size_kb"  "$(num "$size" 0)"
    emit "os.disk.${k}.mount"    "$mnt"
done < <(run 10 df -PT -x tmpfs -x devtmpfs -x squashfs -x overlay 2>/dev/null | tail -n +2)
# A host with no filesystems reported is a broken measurement, not a healthy
# host, and it must never render the same as "all disks fine".
emit os.disk.count "$disk_seen"

# Inode exhaustion presents as "no space left on device" with a half-empty
# disk. It has bitten this estate before and costs one command.
while IFS=' ' read -r _fs _i _iu _if pct mnt; do
    [[ -z "${mnt:-}" ]] && continue
    pct="${pct%\%}"
    # Filesystems without inode accounting report '-'. Emitting that would
    # produce an UNKNOWN finding per mount for a filesystem that simply has
    # no inodes to exhaust — noise that would train people to ignore UNKNOWN.
    [[ "$pct" =~ ^[0-9]+$ ]] || continue
    emit "os.inode.$(keysafe "$mnt").used_pct" "$pct"
done < <(run 10 df -Pi -x tmpfs -x devtmpfs -x squashfs -x overlay 2>/dev/null | tail -n +2)

# Pending reboot and outstanding security updates.
[[ -f /var/run/reboot-required ]] && emit os.reboot_required 1 || emit os.reboot_required 0
if [[ -r /var/run/reboot-required.pkgs ]]; then
    emit os.reboot_required_pkgs "$(tr '\n' ' ' < /var/run/reboot-required.pkgs)"
fi
if [[ -x /usr/lib/update-notifier/apt-check ]]; then
    apt_out="$(run 20 /usr/lib/update-notifier/apt-check 2>&1 || true)"
    emit os.updates.total "$(num "${apt_out%%;*}" 0)"
    emit os.updates.security "$(num "${apt_out##*;}" 0)"
fi

# Failed systemd units. Note deliberately that we do NOT use `systemctl status`
# as a health check for Wazuh — see the wazuh section — but for the OS as a
# whole a failed-unit list is exactly the right instrument.
if have systemctl; then
    failed="$(run 15 systemctl list-units --state=failed --no-legend --plain 2>/dev/null | awk '{print $1}' | tr '\n' ' ')"
    emit os.systemd.failed_count "$(printf '%s' "$failed" | wc -w | tr -d ' ')"
    emit os.systemd.failed_units "${failed:-none}"
    emit os.systemd.degraded "$(run 5 systemctl is-system-running 2>/dev/null || echo unknown)"
fi

# Time synchronisation. On an estate where half the documented incidents were
# timestamp artefacts, clock drift is a first-class health check.
if have timedatectl; then
    td="$(run 5 timedatectl show -p NTPSynchronized --value 2>/dev/null || echo unknown)"
    emit os.time.ntp_synchronised "$td"
fi
if have chronyc; then
    off="$(run 5 chronyc tracking 2>/dev/null | awk -F': *' '/System time/{print $2}' | awk '{print $1}')"
    [[ -n "${off:-}" ]] && emit os.time.offset_seconds "$(num "$off" 0)"
    emit os.time.source chrony
elif [[ -e /run/systemd/timesync ]]; then
    emit os.time.source systemd-timesyncd
fi

# ---------------------------------------------------------------------------
# 3. user access
# ---------------------------------------------------------------------------
# Light mode captures the facts that must never change silently: who is root,
# who can become root, and whether sshd still refuses passwords. Deep mode adds
# the full enumeration, SSH key inventory and login history.

# Any account with UID 0 other than root is an immediate finding.
uid0="$(awk -F: '$3==0{print $1}' /etc/passwd | sort | tr '\n' ' ')"
emit users.uid0 "${uid0:-none}"
emit users.uid0_count "$(printf '%s' "$uid0" | wc -w | tr -d ' ')"

# Privilege groups, in one sorted list so the orchestrator can diff it exactly.
priv=""
for g in sudo admin wheel adm root docker lxd; do
    m="$(getent group "$g" 2>/dev/null | cut -d: -f4)"
    [[ -n "${m:-}" ]] && priv="${priv}${g}:${m} "
done
emit users.privileged_groups "${priv:-none}"

# Interactive accounts (UID >= 1000, real shell).
interactive="$(awk -F: '$3>=1000 && $3<65534 && $7 !~ /(nologin|false|sync)$/ {print $1}' /etc/passwd | sort | tr '\n' ' ')"
emit users.interactive "${interactive:-none}"
emit users.interactive_count "$(printf '%s' "$interactive" | wc -w | tr -d ' ')"

# sshd effective configuration. `sshd -T` gives the RUNNING config including
# Match blocks and defaults — reading sshd_config by hand misses both.
if have sshd; then
    sshd_t="$(run 10 sshd -T 2>/dev/null)"
    if [[ -n "${sshd_t:-}" ]]; then
        for opt in permitrootlogin passwordauthentication pubkeyauthentication \
                   permitemptypasswords challengeresponseauthentication \
                   kbdinteractiveauthentication x11forwarding allowtcpforwarding \
                   maxauthtries clientaliveinterval logingracetime port; do
            v="$(printf '%s\n' "$sshd_t" | awk -v o="$opt" '$1==o{print $2; exit}')"
            [[ -n "${v:-}" ]] && emit "users.sshd.${opt}" "$v"
        done
        emit users.sshd.config_readable 1
    else
        emit users.sshd.config_readable 0
    fi
fi

# Accounts with an empty password field — a real, if rare, catastrophe.
if [[ -r /etc/shadow ]]; then
    empty="$(awk -F: '($2==""){print $1}' /etc/shadow | tr '\n' ' ')"
    emit users.empty_password "${empty:-none}"
    locked_ct="$(awk -F: '($2 ~ /^[!*]/){c++} END{print c+0}' /etc/shadow)"
    emit users.locked_count "$locked_ct"
    emit users.shadow_readable 1
else
    emit users.shadow_readable 0
fi

if [[ "$MODE" == "deep" ]]; then
    # SSH authorised keys, per account. We emit a COUNT and a fingerprint list
    # — never the key material itself.
    keytotal=0
    while IFS=: read -r uname _ uid _ _ home _; do
        [[ "$uid" -ge 1000 || "$uid" -eq 0 ]] || continue
        ak="$home/.ssh/authorized_keys"
        [[ -r "$ak" ]] || continue
        n="$(grep -cvE '^\s*(#|$)' "$ak" 2>/dev/null || echo 0)"
        keytotal=$(( keytotal + n ))
        emit "users.keys.${uname}.count" "$n"
        if have ssh-keygen; then
            fps="$(run 10 ssh-keygen -lf "$ak" 2>/dev/null | awk '{print $2}' | tr '\n' ' ')"
            [[ -n "${fps:-}" ]] && emit "users.keys.${uname}.fingerprints" "$fps"
        fi
        # World- or group-writable authorized_keys is an authentication bypass.
        perm="$(stat -c '%a' "$ak" 2>/dev/null || echo unknown)"
        emit "users.keys.${uname}.mode" "$perm"
    done < /etc/passwd
    emit users.keys.total "$keytotal"

    # Login history. `lastlog` gives last successful login per account; never
    # logged in is as interesting as recently logged in.
    if have lastlog; then
        stale="$(run 20 lastlog -b 90 2>/dev/null | tail -n +2 | awk '$2!="**Never"{print $1}' | tr '\n' ' ')"
        emit users.lastlog.stale_90d "${stale:-none}"
        never="$(run 20 lastlog -u 1000-65000 2>/dev/null | grep -c 'Never logged in' || echo 0)"
        emit users.lastlog.never_count "$(num "$never" 0)"
    fi

    # Recent successful and failed authentications.
    if have journalctl; then
        emit users.auth.failed_24h "$(run 25 journalctl -q --since '24 hours ago' -u ssh -u sshd 2>/dev/null | grep -c 'Failed password\|Invalid user' || echo 0)"
        emit users.auth.accepted_24h "$(run 25 journalctl -q --since '24 hours ago' -u ssh -u sshd 2>/dev/null | grep -c 'Accepted ' || echo 0)"
        emit users.auth.sudo_root_24h "$(run 25 journalctl -q --since '24 hours ago' 2>/dev/null | grep -c 'sudo:.*USER=root' || echo 0)"
    fi

    # Currently logged-in sessions.
    if have who; then
        emit users.sessions "$(run 5 who 2>/dev/null | awk '{print $1"@"$5}' | sort -u | tr '\n' ' ' || echo none)"
    fi

    # sudoers files changed recently, and any NOPASSWD grant beyond our own.
    nopw="$(grep -rhE '^[^#]*NOPASSWD' /etc/sudoers /etc/sudoers.d/ 2>/dev/null | grep -v 'soc-healthcheck' | tr '\n' ';' )"
    emit users.sudoers.nopasswd "${nopw:-none}"
fi

# ---------------------------------------------------------------------------
# 4. service detection
# ---------------------------------------------------------------------------
# Detection drives everything below. Presence is judged on the FILESYSTEM, not
# on a running service — a stopped Graylog is still an installed Graylog and
# must be reported as broken rather than absent.

has_graylog=0; has_wazuh_mgr=0; has_wazuh_agent=0; has_nessus=0; has_fluentbit=0

[[ -d /etc/graylog/server || -f /usr/share/graylog-server/graylog.jar ]] && has_graylog=1
[[ -x "$OSSEC_DIR/bin/wazuh-analysisd" || -x "$OSSEC_DIR/bin/ossec-analysisd" ]] && has_wazuh_mgr=1
[[ -x "$OSSEC_DIR/bin/wazuh-agentd" || -x "$OSSEC_DIR/bin/ossec-agentd" ]] && has_wazuh_agent=1
[[ -x /opt/nessus/sbin/nessusd ]] && has_nessus=1
if have fluent-bit || have td-agent-bit \
   || [[ -f /etc/fluent-bit/fluent-bit.conf || -f /etc/td-agent-bit/td-agent-bit.conf ]]; then
    has_fluentbit=1
fi

# A host carrying the manager also carries an agent binary. Manager wins.
[[ $has_wazuh_mgr -eq 1 ]] && has_wazuh_agent=0

emit svc.graylog.installed "$has_graylog"
emit svc.wazuh_manager.installed "$has_wazuh_mgr"
emit svc.wazuh_agent.installed "$has_wazuh_agent"
emit svc.nessus.installed "$has_nessus"
emit svc.fluentbit.installed "$has_fluentbit"

svc_state() {
    local unit="$1" key="$2"
    if have systemctl; then
        emit "${key}.active" "$(run 5 systemctl is-active "$unit" 2>/dev/null || echo inactive)"
        emit "${key}.enabled" "$(run 5 systemctl is-enabled "$unit" 2>/dev/null || echo unknown)"
        local since
        since="$(run 5 systemctl show "$unit" -p ActiveEnterTimestamp --value 2>/dev/null)"
        [[ -n "${since:-}" ]] && emit "${key}.active_since" "$since"
        emit "${key}.restarts" "$(run 5 systemctl show "$unit" -p NRestarts --value 2>/dev/null || echo 0)"
    fi
}

# ---------------------------------------------------------------------------
# 5. Graylog
# ---------------------------------------------------------------------------

if [[ $has_graylog -eq 1 ]]; then
    svc_state graylog-server svc.graylog

    if [[ "$MODE" == "deep" ]]; then
        # The API is the only instrument that tells the truth. A running JVM
        # that has stopped writing to the index looks perfectly healthy to
        # systemd — that exact state blinded this estate for seven hours.
        gl_host="127.0.0.1"
        gl_port="9000"
        [[ -r /etc/graylog/server/server.conf ]] && {
            hb="$(awk -F'= *' '/^http_bind_address/{print $2}' /etc/graylog/server/server.conf | tr -d ' ')"
            [[ -n "${hb:-}" ]] && { gl_host="${hb%%:*}"; gl_port="${hb##*:}"; }
            [[ "$gl_host" == "0.0.0.0" ]] && gl_host="127.0.0.1"
        }
        base="http://${gl_host}:${gl_port}/api"

        auth=()
        if [[ -r "$GRAYLOG_TOKEN_FILE" ]]; then
            tok="$(head -n1 "$GRAYLOG_TOKEN_FILE" | tr -d '\r\n')"
            [[ -n "$tok" ]] && auth=(-u "${tok}:token")
            unset tok
            emit graylog.api.authenticated 1
        else
            emit graylog.api.authenticated 0
        fi

        gl_get() { run 20 curl -sS --max-time 15 -H 'Accept: application/json' "${auth[@]}" "$1" 2>/dev/null; }

        jrnl="$(gl_get "${base}/system/journal")"
        if [[ -n "${jrnl:-}" && "$jrnl" == *uncommitted_journal_entries* ]]; then
            emit graylog.api.reachable 1
            for f in uncommitted_journal_entries journal_size journal_size_limit \
                     append_events_per_second read_events_per_second; do
                v="$(printf '%s' "$jrnl" | sed -n "s/.*\"$f\"[[:space:]]*:[[:space:]]*\([0-9]*\).*/\1/p" | head -n1)"
                [[ -n "${v:-}" ]] && emit "graylog.journal.${f}" "$v"
            done
        else
            # An unreachable API on an installed Graylog is a CRITICAL fact,
            # and it is NOT the same as "zero results".
            emit graylog.api.reachable 0
        fi

        fails="$(gl_get "${base}/system/indexer/failures/count?since=$(date -u -d '24 hours ago' +%Y-%m-%dT%H:%M:%S.000Z)")"
        v="$(printf '%s' "$fails" | sed -n 's/.*"count"[[:space:]]*:[[:space:]]*\([0-9]*\).*/\1/p' | head -n1)"
        [[ -n "${v:-}" ]] && emit graylog.indexer.failures_24h "$v"

        defl="$(gl_get "${base}/system/deflector")"
        v="$(printf '%s' "$defl" | sed -n 's/.*"current_target"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p')"
        [[ -n "${v:-}" ]] && emit graylog.deflector.current "$v"
        printf '%s' "$defl" | grep -q '"is_up"[[:space:]]*:[[:space:]]*true' && emit graylog.deflector.is_up 1 || emit graylog.deflector.is_up 0

        health="$(gl_get "${base}/system/indexer/cluster/health")"
        v="$(printf '%s' "$health" | sed -n 's/.*"status"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p')"
        [[ -n "${v:-}" ]] && emit graylog.cluster.status "$v"
        for f in active relocating initializing unassigned; do
            v="$(printf '%s' "$health" | sed -n "s/.*\"$f\"[[:space:]]*:[[:space:]]*\([0-9]*\).*/\1/p" | head -n1)"
            [[ -n "${v:-}" ]] && emit "graylog.cluster.shards_${f}" "$v"
        done

        inputs="$(gl_get "${base}/system/inputstates")"
        emit graylog.inputs.running "$(printf '%s' "$inputs" | grep -o '"state"[[:space:]]*:[[:space:]]*"RUNNING"' | wc -l | tr -d ' ')"
        emit graylog.inputs.failed "$(printf '%s' "$inputs" | grep -o '"state"[[:space:]]*:[[:space:]]*"FAILED"' | wc -l | tr -d ' ')"

        # JVM heap pressure — the documented precursor to the node-eviction
        # class of indexer failure.
        [[ -d /var/lib/graylog-server/journal ]] && \
            emit graylog.journal.dir_bytes "$(run 20 du -sb /var/lib/graylog-server/journal 2>/dev/null | awk '{print $1}')"
    fi
fi

# ---------------------------------------------------------------------------
# 6. Wazuh manager
# ---------------------------------------------------------------------------

if [[ $has_wazuh_mgr -eq 1 ]]; then
    # DELIBERATELY NOT `systemctl status wazuh-manager`. On this estate that
    # unit reported "active (exited)" from a start in August while every core
    # daemon was dead. wazuh-control status enumerates the actual processes.
    emit wazuh.manager.check_method wazuh-control
    ctl="$(run 25 "$OSSEC_DIR/bin/wazuh-control" status 2>/dev/null)"
    if [[ -n "${ctl:-}" ]]; then
        running=0; stopped=0; stopped_list=""
        while read -r line; do
            [[ "$line" == *"is running"* ]] && { running=$((running+1)); continue; }
            if [[ "$line" == *"not running"* ]]; then
                stopped=$((stopped+1))
                stopped_list="${stopped_list}$(printf '%s' "$line" | awk '{print $1}') "
            fi
        done <<< "$ctl"
        emit wazuh.manager.daemons_running "$running"
        emit wazuh.manager.daemons_stopped "$stopped"
        emit wazuh.manager.daemons_stopped_list "${stopped_list:-none}"
    else
        emit wazuh.manager.control_readable 0
    fi

    [[ -x "$OSSEC_DIR/bin/wazuh-control" ]] && \
        emit wazuh.manager.version "$(run 10 "$OSSEC_DIR/bin/wazuh-control" info 2>/dev/null | awk -F'"' '/WAZUH_VERSION/{print $2}')"

    # alerts.json size. The orchestrator diffs this against the previous pass:
    # GROWING means analysisd is healthy and any delivery problem is the
    # shipper; FROZEN means the manager itself has stopped detecting. That one
    # comparison splits every "manager not delivering" incident in half.
    alerts="$OSSEC_DIR/logs/alerts/alerts.json"
    if [[ -r "$alerts" ]]; then
        emit wazuh.manager.alerts_json_bytes "$(stat -c '%s' "$alerts" 2>/dev/null || echo 0)"
        emit wazuh.manager.alerts_json_mtime "$(stat -c '%Y' "$alerts" 2>/dev/null || echo 0)"
    else
        emit wazuh.manager.alerts_json_readable 0
    fi

    if [[ "$MODE" == "deep" ]]; then
        # analysisd queue state. events_dropped approaching events_received
        # with every queue at 1.00 is the documented wedge: telemetry is being
        # generated and thrown away, permanently, and no restart recovers it.
        state="$OSSEC_DIR/var/run/wazuh-analysisd.state"
        if [[ -r "$state" ]]; then
            for f in events_received events_dropped events_processed \
                     event_queue_usage rule_matching_queue_usage \
                     alerts_queue_usage firewall_queue_usage \
                     statistical_queue_usage archives_queue_usage \
                     syscheck_queue_usage winevt_queue_usage dbsync_queue_usage \
                     upgrade_queue_usage; do
                v="$(awk -F"'" -v k="$f" '$0 ~ "^"k"=" {print $2}' "$state" 2>/dev/null | head -n1)"
                [[ -z "${v:-}" ]] && v="$(awk -F'=' -v k="$f" '$1==k{gsub(/[^0-9.]/,"",$2); print $2}' "$state" 2>/dev/null | head -n1)"
                [[ -n "${v:-}" ]] && emit "wazuh.analysisd.${f}" "$v"
            done
            emit wazuh.analysisd.state_readable 1
        else
            emit wazuh.analysisd.state_readable 0
        fi

        remoted="$OSSEC_DIR/var/run/wazuh-remoted.state"
        if [[ -r "$remoted" ]]; then
            for f in discarded_count queue_size total_queue_size evt_count ctrl_msg_count; do
                v="$(awk -F"'" -v k="$f" '$0 ~ "^"k"=" {print $2}' "$remoted" 2>/dev/null | head -n1)"
                [[ -n "${v:-}" ]] && emit "wazuh.remoted.${f}" "$v"
            done
        fi

        # Agent connectivity, from the manager's own view.
        if [[ -x "$OSSEC_DIR/bin/agent_control" ]]; then
            ac="$(run 30 "$OSSEC_DIR/bin/agent_control" -l 2>/dev/null)"
            if [[ -n "${ac:-}" ]]; then
                emit wazuh.agents.active "$(printf '%s' "$ac" | grep -c 'Active' || echo 0)"
                emit wazuh.agents.disconnected "$(printf '%s' "$ac" | grep -c 'Disconnected' || echo 0)"
                emit wazuh.agents.never_connected "$(printf '%s' "$ac" | grep -c 'Never connected' || echo 0)"
            fi
        fi

        # Cluster membership — a node that has silently left the cluster is the
        # shape behind several unexplained coverage windows here.
        if [[ -x "$OSSEC_DIR/bin/cluster_control" ]]; then
            emit wazuh.cluster.nodes "$(run 20 "$OSSEC_DIR/bin/cluster_control" -l 2>/dev/null | tail -n +2 | wc -l | tr -d ' ')"
        fi
        if [[ -r "$OSSEC_DIR/etc/ossec.conf" ]]; then
            emit wazuh.cluster.node_name "$(sed -n 's:.*<node_name>\(.*\)</node_name>.*:\1:p' "$OSSEC_DIR/etc/ossec.conf" | head -n1)"
            emit wazuh.cluster.disabled "$(sed -n 's:.*<disabled>\(.*\)</disabled>.*:\1:p' "$OSSEC_DIR/etc/ossec.conf" | head -n1)"
        fi

        # Errors in the manager's own log. Wazuh 4.x writes JSON as well as
        # text and the JSON file is the live one — the text file can sit frozen
        # for hours while the manager is perfectly healthy.
        for lf in "$OSSEC_DIR/logs/ossec.json" "$OSSEC_DIR/logs/ossec.log"; do
            [[ -r "$lf" ]] || continue
            emit "wazuh.manager.log.$(keysafe "$(basename "$lf")").mtime" "$(stat -c '%Y' "$lf" 2>/dev/null || echo 0)"
            emit "wazuh.manager.log.$(keysafe "$(basename "$lf")").errors_tail" \
                 "$(run 15 tail -n 2000 "$lf" 2>/dev/null | grep -ci 'ERROR\|CRITICAL' || echo 0)"
        done

        # Queue spool depth. A queue directory under /tmp is its own finding —
        # systemd-tmpfiles clears it at boot and has hung a boot for 27 minutes
        # on this estate.
        for q in "$OSSEC_DIR/queue/diff" "$OSSEC_DIR/queue/rids" "$OSSEC_DIR/queue/cluster"; do
            [[ -d "$q" ]] && emit "wazuh.queue.$(keysafe "$(basename "$q")").files" \
                "$(run 15 find "$q" -type f 2>/dev/null | wc -l | tr -d ' ')"
        done
    fi
fi

# ---------------------------------------------------------------------------
# 7. fluent-bit  (checked wherever the Wazuh manager lives)
# ---------------------------------------------------------------------------

if [[ $has_wazuh_mgr -eq 1 || $has_fluentbit -eq 1 ]]; then
    for u in fluent-bit td-agent-bit; do
        if have systemctl && run 5 systemctl cat "$u" >/dev/null 2>&1; then
            svc_state "$u" svc.fluentbit
            emit fluentbit.unit "$u"
            break
        fi
    done

    conf=""
    for c in /etc/fluent-bit/fluent-bit.conf /etc/td-agent-bit/td-agent-bit.conf; do
        [[ -r "$c" ]] && { conf="$c"; break; }
    done
    if [[ -n "${conf:-}" ]]; then
        emit fluentbit.config "$conf"
        # The DB parameter is what stops every restart seeking to EOF and
        # silently skipping everything written while it was down. Its ABSENCE
        # is the single most expensive misconfiguration documented here.
        db="$(grep -iE '^\s*DB\s' "$conf" 2>/dev/null | head -n1 | awk '{print $2}')"
        if [[ -n "${db:-}" ]]; then
            emit fluentbit.db.configured 1
            emit fluentbit.db.path "$db"
            dbdir="$(dirname "$db")"
            [[ -d "$dbdir" ]] && emit fluentbit.db.dir_exists 1 || emit fluentbit.db.dir_exists 0
            [[ -f "$db" ]] && emit fluentbit.db.file_exists 1 || emit fluentbit.db.file_exists 0
            [[ -f "$db" ]] && emit fluentbit.db.file_bytes "$(stat -c '%s' "$db" 2>/dev/null || echo 0)"
        else
            emit fluentbit.db.configured 0
        fi
        # A filesystem buffer under /tmp is cleared at boot. Flag it.
        sp="$(grep -iE '^\s*storage.path\s' "$conf" 2>/dev/null | head -n1 | awk '{print $2}')"
        [[ -n "${sp:-}" ]] && { emit fluentbit.storage.path "$sp"; [[ "$sp" == /tmp/* ]] && emit fluentbit.storage.under_tmp 1 || emit fluentbit.storage.under_tmp 0; }
        grep -qiE '^\s*Skip_Long_Lines\s+On' "$conf" 2>/dev/null && emit fluentbit.skip_long_lines 1 || emit fluentbit.skip_long_lines 0
    fi

    # fluent-bit logs ONLY on state change, and it logs to its own Log_File —
    # `journalctl -u fluent-bit` shows the startup banner and nothing else.
    # Reading the wrong source here produced a fabricated diagnosis once.
    if [[ -r "$FLUENTBIT_LOG" ]]; then
        emit fluentbit.log.path "$FLUENTBIT_LOG"
        emit fluentbit.log.mtime "$(stat -c '%Y' "$FLUENTBIT_LOG" 2>/dev/null || echo 0)"
        tailbuf="$(run 15 tail -n 3000 "$FLUENTBIT_LOG" 2>/dev/null)"
        emit fluentbit.log.errors_tail "$(printf '%s' "$tailbuf" | grep -ci '\[error\]' || echo 0)"
        emit fluentbit.log.enoent_tail "$(printf '%s' "$tailbuf" | grep -c 'errno=2' || echo 0)"
        # The observable proof that the offset DB is genuinely in use. A
        # configured-but-inert DB is indistinguishable from a working one
        # without this line.
        emit fluentbit.log.db_in_use "$(printf '%s' "$tailbuf" | grep -c 'delete unmonitored stale inodes' || echo 0)"
        unset tailbuf
    else
        emit fluentbit.log.readable 0
    fi

    if [[ "$MODE" == "deep" ]]; then
        # fluent-bit's own counters — the independent measure that settles
        # "is the shipper dropping this or is the manager not producing it".
        m="$(run 15 curl -sS --max-time 10 http://127.0.0.1:2020/api/v1/metrics 2>/dev/null)"
        if [[ -n "${m:-}" ]]; then
            emit fluentbit.metrics.reachable 1
            emit fluentbit.metrics.input_records  "$(printf '%s' "$m" | grep -o '"records"[[:space:]]*:[[:space:]]*[0-9]*' | head -n1 | grep -o '[0-9]*$')"
            emit fluentbit.metrics.output_records "$(printf '%s' "$m" | grep -o '"proc_records"[[:space:]]*:[[:space:]]*[0-9]*' | head -n1 | grep -o '[0-9]*$')"
            emit fluentbit.metrics.retries        "$(printf '%s' "$m" | grep -o '"retries"[[:space:]]*:[[:space:]]*[0-9]*' | head -n1 | grep -o '[0-9]*$')"
            emit fluentbit.metrics.retries_failed "$(printf '%s' "$m" | grep -o '"retries_failed"[[:space:]]*:[[:space:]]*[0-9]*' | head -n1 | grep -o '[0-9]*$')"
            emit fluentbit.metrics.errors         "$(printf '%s' "$m" | grep -o '"errors"[[:space:]]*:[[:space:]]*[0-9]*' | head -n1 | grep -o '[0-9]*$')"
        else
            emit fluentbit.metrics.reachable 0
        fi
        # Established outbound connections to the Graylog input.
        if have ss; then
            emit fluentbit.net.established "$(run 10 ss -Htnp state established 2>/dev/null | grep -c 'fluent-bit\|td-agent-bit' || echo 0)"
        fi
    fi
fi

# ---------------------------------------------------------------------------
# 8. Nessus
# ---------------------------------------------------------------------------

if [[ $has_nessus -eq 1 ]]; then
    svc_state nessusd svc.nessus

    emit nessus.listening "$(run 10 ss -Hltn 2>/dev/null | grep -c ':8834' || echo 0)"
    [[ -d /opt/nessus ]] && emit nessus.disk_kb "$(run 25 du -sk /opt/nessus 2>/dev/null | awk '{print $1}')"

    if [[ "$MODE" == "deep" ]]; then
        # Plugin feed age is the health check that matters. A running scanner
        # on a three-week-old feed is worse than a stopped one, because it
        # reports clean.
        if [[ -x /opt/nessus/sbin/nessuscli ]]; then
            fi_out="$(run 45 /opt/nessus/sbin/nessuscli fix --secure --list 2>/dev/null)"
            emit nessus.cli_ok 1
            v="$(run 45 /opt/nessus/sbin/nessuscli -v 2>/dev/null | head -n1)"
            [[ -n "${v:-}" ]] && emit nessus.version "$v"
        fi
        pd=/opt/nessus/lib/nessus/plugins/plugin_feed_info.inc
        if [[ -r "$pd" ]]; then
            pv="$(awk -F'"' '/PLUGIN_SET/{print $2}' "$pd" | head -n1)"
            emit nessus.plugin_set "$pv"
            # PLUGIN_SET is YYYYMMDDHHMM — convert to an age in days.
            if [[ "$pv" =~ ^[0-9]{12}$ ]]; then
                pe="$(date -u -d "${pv:0:4}-${pv:4:2}-${pv:6:2} ${pv:8:2}:${pv:10:2}" +%s 2>/dev/null || echo 0)"
                [[ "$pe" -gt 0 ]] && emit nessus.plugin_age_days "$(( (START_EPOCH - pe) / 86400 ))"
            fi
        fi
        [[ -r /opt/nessus/var/nessus/logs/nessusd.messages ]] && \
            emit nessus.log.errors_tail "$(run 15 tail -n 2000 /opt/nessus/var/nessus/logs/nessusd.messages 2>/dev/null | grep -ci 'error\|fail' || echo 0)"
    fi
fi

# ---------------------------------------------------------------------------
# 9. Wazuh agent  (only where there is no manager)
# ---------------------------------------------------------------------------

if [[ $has_wazuh_agent -eq 1 ]]; then
    emit wazuh.agent.check_method wazuh-control
    ctl="$(run 20 "$OSSEC_DIR/bin/wazuh-control" status 2>/dev/null)"
    if [[ -n "${ctl:-}" ]]; then
        emit wazuh.agent.daemons_running "$(printf '%s' "$ctl" | grep -c 'is running' || echo 0)"
        emit wazuh.agent.daemons_stopped "$(printf '%s' "$ctl" | grep -c 'not running' || echo 0)"
    fi

    # The agent's own record of whether it is talking to a manager. "Connected
    # to the server" in the log beats any service state.
    if [[ -r "$OSSEC_DIR/var/run/wazuh-agentd.state" ]]; then
        for f in status last_keepalive last_ack msg_count msg_sent; do
            v="$(awk -F"'" -v k="$f" '$0 ~ "^"k"=" {print $2}' "$OSSEC_DIR/var/run/wazuh-agentd.state" 2>/dev/null | head -n1)"
            [[ -n "${v:-}" ]] && emit "wazuh.agent.${f}" "$v"
        done
    fi

    if [[ -r "$OSSEC_DIR/etc/ossec.conf" ]]; then
        emit wazuh.agent.manager_address "$(sed -n 's:.*<address>\(.*\)</address>.*:\1:p' "$OSSEC_DIR/etc/ossec.conf" | head -n1)"
    fi
    [[ -r "$OSSEC_DIR/etc/client.keys" ]] && \
        emit wazuh.agent.registered "$([[ -s "$OSSEC_DIR/etc/client.keys" ]] && echo 1 || echo 0)"
    [[ -x "$OSSEC_DIR/bin/wazuh-control" ]] && \
        emit wazuh.agent.version "$(run 10 "$OSSEC_DIR/bin/wazuh-control" info 2>/dev/null | awk -F'"' '/WAZUH_VERSION/{print $2}')"

    if [[ "$MODE" == "deep" ]]; then
        for lf in "$OSSEC_DIR/logs/ossec.json" "$OSSEC_DIR/logs/ossec.log"; do
            [[ -r "$lf" ]] || continue
            emit "wazuh.agent.log.$(keysafe "$(basename "$lf")").mtime" "$(stat -c '%Y' "$lf" 2>/dev/null || echo 0)"
            tb="$(run 15 tail -n 1000 "$lf" 2>/dev/null)"
            emit "wazuh.agent.log.$(keysafe "$(basename "$lf")").errors_tail" "$(printf '%s' "$tb" | grep -ci 'ERROR\|CRITICAL' || echo 0)"
            emit "wazuh.agent.log.$(keysafe "$(basename "$lf")").lost_connection" "$(printf '%s' "$tb" | grep -c 'Lost connection\|Unable to connect' || echo 0)"
            unset tb
        done
    fi
fi

# ---------------------------------------------------------------------------
# 10. the install trigger
# ---------------------------------------------------------------------------
# A host with NEITHER manager NOR agent is unmonitored. The node script never
# installs anything itself — it states the fact and the orchestrator decides.

if [[ $has_wazuh_mgr -eq 0 && $has_wazuh_agent -eq 0 ]]; then
    emit wazuh.coverage none
    emit wazuh.install_required 1
else
    emit wazuh.coverage "$([[ $has_wazuh_mgr -eq 1 ]] && echo manager || echo agent)"
    emit wazuh.install_required 0
fi

emit node.duration_seconds "$(( $(date -u +%s) - START_EPOCH ))"
emit node.complete 1
exit 0
