orca/scripts/orca-aggregate.sh

#!/usr/bin/env bash
# orca-aggregate.sh — lead-side aggregator (P09, C-11/C-12/C-14, REQ-107, D-237).
#
# Runs every 10s via orca-aggregate.timer. For each peer in
# /etc/orca/peers.env: SSHs in, reads /run/orca/state/<latest>.json (the
# peer's state snapshot), and merges all peer states into a single
# /etc/orca/state/cluster.json on the lead. Writes atomically (temp +
# rename). Structured logging via logger -t orca-aggregate (syslog).
#
# Drift-event aggregation extension (REQ-107, D-237): rsyncs each
# peer's /etc/orca/state/drift-events/ to a temp dir, validates each
# event's hash against /etc/orca/state/applied/<txn>/manifest.json,
# triggers orca-remediate.sh <peer> <txn> for auto-remediable paths
# (the remediation script itself lands in P10b — the call is emitted
# but the script may not exist yet), then consumes (deletes) the event
# file on the peer after processing. Aggregated drift state is written
# to /etc/orca/state/drift-events-aggregated.json. The drift section
# is skipped gracefully (with a log line) when the drift-events
# directory does not exist on the lead or on any peer.
#
# The aggregator is opt-in (C-12): the operator enables it with
# `orca cluster config --aggregator=true`, which installs this script
# + the systemd units via `orca collector start`.

set -euo pipefail

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
# shellcheck source=lib/orca-log.sh
. "$SCRIPT_DIR/lib/orca-log.sh"

ORCA_LOG_ACTOR="spiffe://orca/cli/aggregator"

PEERS_ENV="${ORCA_PEERS_ENV:-/etc/orca/peers.env}"
STATE_DIR="${ORCA_STATE_DIR:-/etc/orca/state}"
CLUSTER_JSON="$STATE_DIR/cluster.json"
DRIFT_AGG_JSON="$STATE_DIR/drift-events-aggregated.json"
REMOTE_STATE_DIR="${ORCA_REMOTE_STATE_DIR:-/run/orca/state}"
REMOTE_DRIFT_DIR="${ORCA_REMOTE_DRIFT_DIR:-/etc/orca/state/drift-events}"
APPLIED_DIR="${ORCA_APPLIED_DIR:-/etc/orca/state/applied}"
REMEDIATE_SCRIPT="${ORCA_REMEDIATE_SCRIPT:-/usr/local/sbin/orca-remediate.sh}"
SSH_OPTS="${ORCA_SSH_OPTS:--o BatchMode=yes -o StrictHostKeyChecking=accept-new -o ConnectTimeout=5}"
RSYNC_OPTS_RSYNC="${ORCA_RSYNC_OPTS_RSYNC:-}"
AGG_TMP="$(mktemp -d)"
trap 'rm -rf "$AGG_TMP"' EXIT

mkdir -p "$STATE_DIR"

read_peers() {
	if [ ! -f "$PEERS_ENV" ]; then
		return 0
	fi
	awk -F= '/^[[:space:]]*#/ {next} /^[[:space:]]*$/ {next} {sub(/^[[:space:]]*PEER_[^=]*=/,""); print}' "$PEERS_ENV"
}

merge_tmp="$AGG_TMP/merge.jsonl"
: >"$merge_tmp"

peer_count=0
while IFS= read -r peer; do
	[ -z "$peer" ] && continue
	peer_count=$((peer_count + 1))
	latest_json="$(ssh $SSH_OPTS "$peer" "ls -1 $REMOTE_STATE_DIR/*.json 2>/dev/null | sort | tail -1" 2>/dev/null || true)"
	if [ -z "$latest_json" ]; then
		orca_log_warn "aggregate" "$peer" "skipped" "no state snapshot on peer"
		continue
	fi
	snapshot="$(ssh $SSH_OPTS "$peer" "cat '$latest_json'" 2>/dev/null || true)"
	if [ -z "$snapshot" ]; then
		orca_log_warn "aggregate" "$peer" "skipped" "failed to read state snapshot"
		continue
	fi
	# REQ-131 / F11: use jq to safely construct JSON (prevents JSON
	# injection from malicious peer output). $peer is sanitized; $snapshot
	# is parsed as raw JSON by jq, so control chars can't break out.
	if command -v jq >/dev/null 2>&1; then
		snapshot_json="$(printf '%s' "$snapshot" | jq -c '.' 2>/dev/null)" || {
			orca_log_warn "aggregate" "$peer" "skipped" "peer returned invalid JSON"
			continue
		}
		peer_escaped="${peer//\"/\\\"}"
		printf '{"peer":"%s","state":%s}\n' "$peer_escaped" "$snapshot_json" >>"$merge_tmp"
	else
		# Fallback: validate $snapshot looks like JSON before interpolation.
		if ! printf '%s' "$snapshot" | grep -qE '^\s*\{.*\}\s*$'; then
			orca_log_warn "aggregate" "$peer" "skipped" "peer returned non-JSON"
			continue
		fi
		peer_escaped="${peer//\"/\\\"}"
		printf '{"peer":"%s","state":%s}\n' "$peer_escaped" "$snapshot" >>"$merge_tmp"
	fi
	orca_log_info "aggregate" "$peer" "ok" "snapshot=$latest_json"
done < <(read_peers)

ts="$(date -u +%Y-%m-%dT%H:%M:%S.%3NZ)"
{
	printf '{"ts":"%s","peers":[' "$ts"
	first=1
	while IFS= read -r line; do
		[ -z "$line" ] && continue
		if [ "$first" -eq 1 ]; then
			first=0
		else
			printf ','
		fi
		printf '%s' "$line"
	done <"$merge_tmp"
	printf ']}'
} >"$AGG_TMP/cluster.json.new"
mv "$AGG_TMP/cluster.json.new" "$CLUSTER_JSON"
orca_log_info "aggregate" "-" "ok" "peers=$peer_count cluster_json=$CLUSTER_JSON"

if [ ! -d "$REMOTE_DRIFT_DIR" ] && [ ! -d "$APPLIED_DIR" ]; then
	orca_log_info "drift" "-" "skipped" "drift-events + applied dirs absent (P10b not shipped)"
else
	drift_tmp="$AGG_TMP/drift-events"
	mkdir -p "$drift_tmp"
	drift_agg_tmp="$AGG_TMP/drift-agg.jsonl"
	: >"$drift_agg_tmp"
	while IFS= read -r peer; do
		[ -z "$peer" ] && continue
		peer_drift_dir="$drift_tmp/$peer"
		mkdir -p "$peer_drift_dir"
		if ! rsync -a --quiet $RSYNC_OPTS_RSYNC "$peer:$REMOTE_DRIFT_DIR/" "$peer_drift_dir/" 2>/dev/null; then
			orca_log_info "drift" "$peer" "skipped" "rsync failed or dir absent"
			continue
		fi
		shopt -s nullglob
		event_files=("$peer_drift_dir"/*.json)
		shopt -u nullglob
		if [ "${#event_files[@]}" -eq 0 ]; then
			orca_log_info "drift" "$peer" "ok" "no events"
			continue
		fi
		for event_file in "${event_files[@]}"; do
			event_id="$(basename "$event_file" .json)"
			path_val="$(grep -o '"path"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
			new_sha="$(grep -o '"new_sha256"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
			txn_val="$(grep -o '"latest_txn"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
			status_val="$(grep -o '"status"[[:space:]]*:[[:space:]]*"[^"]*"' "$event_file" | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
			if [ -z "$txn_val" ]; then
				orca_log_warn "drift" "$peer" "skipped" "event=$event_id no latest_txn"
				continue
			fi
			manifest="$APPLIED_DIR/$txn_val/manifest.json"
			drift_confirmed=0
			if [ -f "$manifest" ]; then
				manifest_sha="$(grep -o "\"$path_val\"[^\"]*\"sha256\"[[:space:]]*:[[:space:]]*\"[^\"]*\"" "$manifest" 2>/dev/null | grep -o '"sha256"[[:space:]]*:[[:space:]]*"[^"]*"' | head -1 | sed 's/.*: *"\([^"]*\)"/\1/')"
				if [ -n "$manifest_sha" ] && [ "$manifest_sha" != "$new_sha" ]; then
					drift_confirmed=1
				fi
			else
				orca_log_warn "drift" "$peer" "skipped" "txn=$txn_val manifest absent"
			fi
			if [ "$drift_confirmed" -eq 1 ]; then
				if [ -x "$REMEDIATE_SCRIPT" ]; then
					orca_log_info "drift" "$peer" "remediate" "txn=$txn_val path=$path_val"
					"$REMEDIATE_SCRIPT" "$peer" "$txn_val" "$path_val" \
						|| orca_log_error "drift" "$peer" "remediate-failed" "txn=$txn_val path=$path_val"
				else
					orca_log_warn "drift" "$peer" "remediate-stub" "orca-remediate.sh not installed (P10b)"
				fi
				printf '{"peer":"%s","event_id":"%s","txn":"%s","path":"%s","status":"%s","remediated":true}\n' \
					"$peer" "$event_id" "$txn_val" "$path_val" "$status_val" >>"$drift_agg_tmp"
			else
				printf '{"peer":"%s","event_id":"%s","txn":"%s","path":"%s","status":"%s","remediated":false}\n' \
					"$peer" "$event_id" "$txn_val" "$path_val" "$status_val" >>"$drift_agg_tmp"
			fi
			ssh $SSH_OPTS "$peer" "rm -f '$REMOTE_DRIFT_DIR/$event_id.json'" 2>/dev/null \
				|| orca_log_warn "drift" "$peer" "consume-failed" "event=$event_id"
		done
	done < <(read_peers)
	{
		printf '{"ts":"%s","events":[' "$ts"
		first=1
		while IFS= read -r line; do
			[ -z "$line" ] && continue
			if [ "$first" -eq 1 ]; then
				first=0
			else
				printf ','
			fi
			printf '%s' "$line"
		done <"$drift_agg_tmp"
		printf ']}'
	} >"$AGG_TMP/drift-agg.json.new"
	mv "$AGG_TMP/drift-agg.json.new" "$DRIFT_AGG_JSON"
	orca_log_info "drift" "-" "ok" "aggregated=$DRIFT_AGG_JSON"
fi

orca_log_info "aggregate" "-" "done" "peers=$peer_count"