orca/internal/drift/drift.go

// Package drift implements orca's drift detection subsystem
// (P10b, v0.11 milestone; R-018/R-019/R-020, REQ-103..REQ-113).
//
// The model is poll-based with a systemd Path-unit fast path for
// critical files:
//
//   - The Detector interface exposes four operations: Watch (stream
//     events via iter.Seq2[Event, error], D-017), Aggregate (read the
//     lead-side aggregated drift state), Remediate (trigger the
//     peer-side applier via orca-remediate.sh, with cooldown-on-success
//     per C4), and Acknowledge (record operator ack).
//   - DefaultDetector implements Detector against the SSH-push
//     transport (the same *sshpush.Transport used by txn / emitter).
//   - Config encodes the tiered cadence (critical 5s + systemd Path
//     units, standard 30s polling, default 60s) plus the remediation
//     policy (auto paths, require-approval paths, notify-on-remediate).
//
// The package never logs key material. slog calls carry only metadata.
package drift

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"iter"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"time"
)

type Status string

const (
	StatusModified Status = "modified"
	StatusDeleted  Status = "deleted"
	StatusCreated  Status = "created"
)

type Action string

const (
	ActionAutoRemediated Action = "auto_remediated"
	ActionReported       Action = "reported"
	ActionAcknowledged   Action = "acknowledged"
)

type ActionResult string

const (
	ActionResultSuccess ActionResult = "success"
	ActionResultFailed  ActionResult = "failed"
	ActionResultSkipped ActionResult = "skipped"
)

type Tier string

const (
	TierCritical Tier = "critical"
	TierStandard Tier = "standard"
)

type Event struct {
	EventID        string       `json:"event_id"`
	TS             time.Time    `json:"ts"`
	Host           string       `json:"host"`
	Path           string       `json:"path"`
	Status         Status       `json:"status"`
	NewSHA256      string       `json:"new_sha256"`
	LatestTxn      string       `json:"latest_txn"`
	ExpectedSHA256 string       `json:"expected_sha256"`
	DriftConfirmed bool         `json:"drift_confirmed"`
	Action         Action       `json:"action"`
	ActionResult   ActionResult `json:"action_result"`
	OperatorID     string       `json:"operator_id"`
}

type PathSpec struct {
	Tier            Tier          `json:"tier"`
	Pattern         string        `json:"pattern"`
	Interval        time.Duration `json:"interval"`
	SystemdPathUnit bool          `json:"systemd_path_unit"`
}

type RemediationPolicy struct {
	Auto                bool     `json:"auto"`
	AutoPaths           []string `json:"auto_paths"`
	RequireApproval     []string `json:"require_approval"`
	NotifyOnRemediation bool     `json:"notify_on_remediation"`
}

type PollingConfig struct {
	Enabled            bool          `json:"enabled"`
	DefaultInterval    time.Duration `json:"default_interval"`
	MaxConcurrentPeers int           `json:"max_concurrent_peers"`
}

type PathsConfig struct {
	Critical []PathSpec `json:"critical"`
	Standard []PathSpec `json:"standard"`
	Excluded []string   `json:"excluded"`
}

type Config struct {
	Polling   PollingConfig     `json:"polling"`
	Paths     PathsConfig       `json:"paths"`
	Remediate RemediationPolicy `json:"remediate"`
}

type Detector interface {
	Watch(ctx context.Context, paths []PathSpec) iter.Seq2[Event, error]
	Aggregate(ctx context.Context, leadPeer string) ([]Event, error)
	Remediate(ctx context.Context, leadPeer, path string, force bool) error
	Acknowledge(ctx context.Context, leadPeer, path string) error
}

type Transport interface {
	Exec(ctx context.Context, peer string, cmd string) ([]byte, error)
	WriteFileIdempotent(ctx context.Context, peer string, path string, content []byte, mode os.FileMode) (bool, error)
	ReadFile(ctx context.Context, peer string, path string) ([]byte, error)
}

func LeadStateDir() string {
	if p := os.Getenv("ORCA_LEAD_STATE_DIR"); p != "" {
		return p
	}
	return "/etc/orca/state"
}

func AggregatedPath() string {
	return filepath.Join(LeadStateDir(), "drift-events-aggregated.json")
}

func AcknowledgmentsPath() string {
	return filepath.Join(LeadStateDir(), "drift-acknowledgments.json")
}

func RemediatorPath() string {
	return "/usr/local/sbin/orca-remediate.sh"
}

type DefaultDetector struct {
	transport Transport
	cooldown  time.Duration
}

func NewDefaultDetector(t Transport) *DefaultDetector {
	return &DefaultDetector{transport: t, cooldown: 5 * time.Minute}
}

func (d *DefaultDetector) SetCooldown(dur time.Duration) {
	if dur > 0 {
		d.cooldown = dur
	}
}

type aggregatedDoc struct {
	TS     time.Time `json:"ts"`
	Events []Event   `json:"events"`
}

func (d *DefaultDetector) Watch(ctx context.Context, paths []PathSpec) iter.Seq2[Event, error] {
	interval := defaultWatchInterval(paths)
	if interval <= 0 {
		interval = 2 * time.Second
	}
	return func(yield func(Event, error) bool) {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		seen := make(map[string]bool)
		for {
			events, err := d.Aggregate(ctx, "")
			if err != nil {
				if !yield(Event{}, err) {
					return
				}
			} else {
				for _, e := range events {
					if !matchesPaths(e.Path, paths) {
						continue
					}
					key := e.EventID
					if key == "" {
						key = e.Host + "|" + e.Path + "|" + e.TS.Format(time.RFC3339Nano)
					}
					if seen[key] {
						continue
					}
					seen[key] = true
					if !yield(e, nil) {
						return
					}
				}
			}
			select {
			case <-ctx.Done():
				return
			case <-ticker.C:
			}
		}
	}
}

func (d *DefaultDetector) Aggregate(ctx context.Context, leadPeer string) ([]Event, error) {
	var raw []byte
	var err error
	if leadPeer == "" {
		raw, err = os.ReadFile(AggregatedPath())
		if err != nil {
			if os.IsNotExist(err) {
				return nil, nil
			}
			return nil, fmt.Errorf("drift: read aggregated: %w", err)
		}
	} else {
		raw, err = d.transport.ReadFile(ctx, leadPeer, AggregatedPath())
		if err != nil {
			return nil, fmt.Errorf("drift: read aggregated from %s: %w", leadPeer, err)
		}
	}
	if len(strings.TrimSpace(string(raw))) == 0 {
		return nil, nil
	}
	var doc aggregatedDoc
	if err := json.Unmarshal(raw, &doc); err != nil {
		return nil, fmt.Errorf("drift: parse aggregated: %w", err)
	}
	return doc.Events, nil
}

func (d *DefaultDetector) Remediate(ctx context.Context, leadPeer, path string, force bool) error {
	if leadPeer == "" {
		return fmt.Errorf("drift: remediate requires a lead peer")
	}
	if !force {
		if d.inCooldown(path) {
			slog.Info("drift remediate skipped (cooldown)", "path", path, "peer", leadPeer)
			return ErrCooldown
		}
	}
	cmd := fmt.Sprintf("bash %s %s %s %s", shellQuote(RemediatorPath()), shellQuote(leadPeer), shellQuote(""), shellQuote(path))
	out, err := d.transport.Exec(ctx, leadPeer, cmd)
	if err != nil {
		if isTransientSSH(err) {
			slog.Warn("drift remediate transient failure (no cooldown)", "path", path, "peer", leadPeer, "error", err)
			return err
		}
		return fmt.Errorf("drift: remediate %s on %s: %w (output: %s)", path, leadPeer, err, string(out))
	}
	if !force {
		d.markCooldown(path)
	}
	slog.Info("drift remediate ok", "path", path, "peer", leadPeer, "output", string(out))
	return nil
}

func (d *DefaultDetector) Acknowledge(ctx context.Context, leadPeer, path string) error {
	if leadPeer == "" {
		return fmt.Errorf("drift: acknowledge requires a lead peer")
	}
	entry := map[string]any{
		"ts":       time.Now().UTC().Format(time.RFC3339Nano),
		"path":     path,
		"operator": os.Getenv("ORCA_OPERATOR_ID"),
	}
	raw, err := json.Marshal(entry)
	if err != nil {
		return fmt.Errorf("drift: marshal ack: %w", err)
	}
	if _, err := d.transport.WriteFileIdempotent(ctx, leadPeer, AcknowledgmentsPath(), append(raw, '\n'), 0o644); err != nil {
		return fmt.Errorf("drift: write ack: %w", err)
	}
	slog.Info("drift acknowledged", "path", path, "peer", leadPeer)
	return nil
}

func (d *DefaultDetector) inCooldown(path string) bool {
	hash := pathHash(path)
	stateDir := LeadStateDir()
	cooldownDir := filepath.Join(stateDir, "remediation-cooldown")
	stampPath := filepath.Join(cooldownDir, hash)
	info, err := os.Stat(stampPath)
	if err != nil {
		return false
	}
	return time.Since(info.ModTime()) < d.cooldown
}

func (d *DefaultDetector) markCooldown(path string) {
	hash := pathHash(path)
	stateDir := LeadStateDir()
	cooldownDir := filepath.Join(stateDir, "remediation-cooldown")
	if err := os.MkdirAll(cooldownDir, 0o755); err != nil {
		slog.Warn("drift markCooldown mkdir failed", "dir", cooldownDir, "error", err)
		return
	}
	stampPath := filepath.Join(cooldownDir, hash)
	if err := os.WriteFile(stampPath, []byte(time.Now().UTC().Format(time.RFC3339Nano)), 0o644); err != nil {
		slog.Warn("drift markCooldown write failed", "path", stampPath, "error", err)
	}
}

var (
	ErrCooldown        = errors.New("drift: remediation in cooldown")
	ErrPathExcluded    = errors.New("drift: path is excluded")
	ErrOverlapCritical = errors.New("drift: path appears in both critical and excluded")
)

func LoadConfig(path string) (*Config, error) {
	if path == "" {
		return DefaultConfig(), nil
	}
	raw, err := os.ReadFile(path)
	if err != nil {
		if os.IsNotExist(err) {
			return DefaultConfig(), nil
		}
		return nil, fmt.Errorf("drift: load config %s: %w", path, err)
	}
	var cfg Config
	if err := json.Unmarshal(raw, &cfg); err != nil {
		return nil, fmt.Errorf("drift: parse config %s: %w", path, err)
	}
	applyConfigDefaults(&cfg)
	return &cfg, nil
}

func ValidateConfig(cfg *Config) error {
	if cfg == nil {
		return fmt.Errorf("drift: nil config")
	}
	if cfg.Polling.Enabled && cfg.Polling.DefaultInterval <= 0 {
		return fmt.Errorf("drift: polling enabled but default_interval is zero")
	}
	for _, p := range cfg.Paths.Critical {
		if p.Pattern == "" {
			return fmt.Errorf("drift: critical path has empty pattern")
		}
		if p.Tier == "" {
			return fmt.Errorf("drift: critical path %q has empty tier", p.Pattern)
		}
	}
	for _, p := range cfg.Paths.Standard {
		if p.Pattern == "" {
			return fmt.Errorf("drift: standard path has empty pattern")
		}
	}
	for _, ex := range cfg.Paths.Excluded {
		for _, c := range cfg.Paths.Critical {
			if pathGlobMatch(ex, c.Pattern) {
				return fmt.Errorf("drift: excluded pattern %q overlaps critical %q: %w", ex, c.Pattern, ErrOverlapCritical)
			}
		}
	}
	return nil
}

func DefaultConfig() *Config {
	cfg := &Config{
		Polling: PollingConfig{
			Enabled:            true,
			DefaultInterval:    60 * time.Second,
			MaxConcurrentPeers: 8,
		},
		Paths: PathsConfig{
			Critical: []PathSpec{
				{Tier: TierCritical, Pattern: "/etc/traefik/dynamic/orca.yml", Interval: 5 * time.Second, SystemdPathUnit: true},
				{Tier: TierCritical, Pattern: "/etc/systemd/system/orca-alloc-*.service", Interval: 5 * time.Second, SystemdPathUnit: true},
				{Tier: TierCritical, Pattern: "/etc/nftables.d/orca.nft", Interval: 5 * time.Second, SystemdPathUnit: true},
				{Tier: TierCritical, Pattern: "/etc/orca/actual/*/etc/traefik/dynamic/*", Interval: 5 * time.Second, SystemdPathUnit: true},
				{Tier: TierCritical, Pattern: "/etc/orca/actual/*/etc/systemd/system/orca-alloc-*.service", Interval: 5 * time.Second, SystemdPathUnit: true},
				{Tier: TierCritical, Pattern: "/etc/orca/actual/*/etc/sudoers.d/orca-*", Interval: 5 * time.Second, SystemdPathUnit: true},
			},
			Standard: []PathSpec{
				{Tier: TierStandard, Pattern: "/etc/orca/actual/*", Interval: 30 * time.Second},
				{Tier: TierStandard, Pattern: "/etc/sudoers.d/orca-*", Interval: 30 * time.Second},
				{Tier: TierStandard, Pattern: "/etc/systemd/system/orca-collector.{service,timer}", Interval: 30 * time.Second},
				{Tier: TierStandard, Pattern: "/etc/systemd/system/orca-aggregator.{service,timer}", Interval: 30 * time.Second},
				{Tier: TierStandard, Pattern: "/etc/systemd/system/orca-pull.{service,timer}", Interval: 30 * time.Second},
				{Tier: TierStandard, Pattern: "/etc/systemd/system/orca-drift.{service,timer}", Interval: 30 * time.Second},
			},
			Excluded: []string{
				"/etc/orca/credentials/*",
				"/run/orca/*",
				"/etc/orca/state/drift-events/*",
			},
		},
		Remediate: RemediationPolicy{
			Auto:                true,
			AutoPaths:           []string{"/etc/traefik/dynamic/*", "/etc/nftables.d/*", "/etc/sudoers.d/orca-*"},
			RequireApproval:     []string{"/etc/systemd/system/orca-alloc-*.service"},
			NotifyOnRemediation: true,
		},
	}
	applyConfigDefaults(cfg)
	return cfg
}

func applyConfigDefaults(cfg *Config) {
	if cfg.Polling.DefaultInterval == 0 {
		cfg.Polling.DefaultInterval = 60 * time.Second
	}
	if cfg.Polling.MaxConcurrentPeers == 0 {
		cfg.Polling.MaxConcurrentPeers = 8
	}
	for i := range cfg.Paths.Critical {
		if cfg.Paths.Critical[i].Tier == "" {
			cfg.Paths.Critical[i].Tier = TierCritical
		}
		if cfg.Paths.Critical[i].Interval == 0 {
			cfg.Paths.Critical[i].Interval = 5 * time.Second
		}
	}
	for i := range cfg.Paths.Standard {
		if cfg.Paths.Standard[i].Tier == "" {
			cfg.Paths.Standard[i].Tier = TierStandard
		}
		if cfg.Paths.Standard[i].Interval == 0 {
			cfg.Paths.Standard[i].Interval = 30 * time.Second
		}
	}
}

func NamespaceScope(events []Event, namespace string) []Event {
	if namespace == "" {
		var out []Event
		for _, e := range events {
			if !e.DriftConfirmed || e.Action == ActionAcknowledged {
				continue
			}
			out = append(out, e)
		}
		return out
	}
	var out []Event
	for _, e := range events {
		if !e.DriftConfirmed || e.Action == ActionAcknowledged {
			continue
		}
		if nsForPath(e.Path) == namespace {
			out = append(out, e)
		}
	}
	return out
}

func nsForPath(p string) string {
	const prefix = "/etc/orca/actual/"
	if !strings.HasPrefix(p, prefix) {
		return ""
	}
	rest := p[len(prefix):]
	if i := strings.IndexByte(rest, '/'); i >= 0 {
		return rest[:i]
	}
	return rest
}

func matchesPaths(p string, specs []PathSpec) bool {
	if len(specs) == 0 {
		return true
	}
	for _, s := range specs {
		if pathGlobMatch(s.Pattern, p) {
			return true
		}
	}
	return false
}

func defaultWatchInterval(specs []PathSpec) time.Duration {
	var min time.Duration
	for _, s := range specs {
		if s.Interval > 0 && (min == 0 || s.Interval < min) {
			min = s.Interval
		}
	}
	return min
}

func pathGlobMatch(pattern, p string) bool {
	if pattern == "" {
		return false
	}
	expanded := expandBraces(pattern)
	for _, alt := range expanded {
		if globMatchSegment(alt, p) {
			return true
		}
	}
	return false
}

func globMatchSegment(pattern, p string) bool {
	if pattern == "" {
		return false
	}
	if pattern == p {
		return true
	}
	if strings.HasSuffix(pattern, "/*") {
		prefix := pattern[:len(pattern)-2]
		if p == prefix {
			return true
		}
		return strings.HasPrefix(p, prefix+"/")
	}
	ok, err := filepath.Match(pattern, p)
	return err == nil && ok
}

func expandBraces(p string) []string {
	open := strings.IndexByte(p, '{')
	if open < 0 {
		return []string{p}
	}
	close := strings.IndexByte(p[open:], '}')
	if close < 0 {
		return []string{p}
	}
	close += open
	prefix := p[:open]
	body := p[open+1 : close]
	suffix := p[close+1:]
	alternatives := strings.Split(body, ",")
	var out []string
	for _, a := range alternatives {
		for _, tail := range expandBraces(suffix) {
			out = append(out, prefix+a+tail)
		}
	}
	return out
}

func pathHash(p string) string {
	sum := sha256.Sum256([]byte(p))
	return hex.EncodeToString(sum[:])
}

func isTransientSSH(err error) bool {
	if err == nil {
		return false
	}
	s := err.Error()
	for _, sub := range []string{"connection refused", "i/o timeout", "EOF", "no such host", "connection reset", "timeout", "deadline exceeded", "temporarily unavailable"} {
		if strings.Contains(s, sub) {
			return true
		}
	}
	return false
}

func shellQuote(s string) string {
	return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'"
}

func ParseEvent(raw []byte) (Event, error) {
	var e Event
	if err := json.Unmarshal(raw, &e); err != nil {
		return Event{}, fmt.Errorf("drift: parse event: %w", err)
	}
	return e, nil
}

func MarshalEvent(e Event) ([]byte, error) {
	return json.MarshalIndent(e, "", "  ")
}

var _ Detector = (*DefaultDetector)(nil)