felhom.eu/hub/internal/store/logbundle.go

package store

import (
	"bytes"
	"compress/gzip"
	"database/sql"
	"fmt"
	"io"
	"regexp"
	"strings"
	"time"
)

// Component log bundles (v0.46.0 observability) — the box-component (controller /
// agent) sibling of the per-app log tails. Same sovereignty posture: the operator's
// request is a PENDING flag the box consumes on its own cycle (report ACK for the
// controller, heartbeat envelope for the agent); the hub never connects in. The
// received tail is gzip-stored with a 72 h TTL; the bundler runs a token-pattern
// gate BEFORE storing — a hit stores NOTHING and flags the bundle `blocked` (fail-
// closed belt over the "keys never values" logging convention).

const (
	// LogBundleComponentController scopes by CUSTOMER (the report channel is per-customer).
	LogBundleComponentController = "controller"
	// LogBundleComponentAgent scopes by HOST (the heartbeat channel is per-host).
	LogBundleComponentAgent = "agent"

	// logBundleTTL is how long a received bundle is served before the purge sweep drops it.
	logBundleTTL = 72 * time.Hour

	// logBundleMaxRawBytes is the belt over the wire caps (both channels cap at 128 KiB):
	// oversize input keeps the NEWEST lines within this budget.
	logBundleMaxRawBytes = 1 << 20

	// logBundleKeep bounds stored bundles per (scope, component) — newest kept.
	logBundleKeep = 3
)

// secretPatterns is the bundle gate: content shaped like a credential. The logging
// conventions forbid secret VALUES in any log line, so a hit here is a violation —
// fail closed (store nothing, flag the bundle). Deliberately NOT a generic hex
// matcher: fingerprints/checksums are logged by design and are public.
var secretPatterns = []*regexp.Regexp{
	regexp.MustCompile(`-----BEGIN [A-Z ]*PRIVATE KEY-----`),
	regexp.MustCompile(`\bre_[A-Za-z0-9]{10,}`),   // Resend API key shape
	regexp.MustCompile(`PVEAPIToken=[^\s"']+:\S+`), // Proxmox token WITH secret part
	regexp.MustCompile(`(?i)\b(password|passwd|api_key|apikey|token_secret|client_secret)=[^\s"']{6,}`),
	regexp.MustCompile(`Authorization:\s*Bearer\s+\S{16,}`),
}

// findSecretPattern returns the name of the first matching pattern ("" = clean).
// A match containing "[REDACTED]" is skipped — box-side redaction deliberately
// leaves `password=[REDACTED]` shapes, which are proof of redaction, not a leak. Pure.
func findSecretPattern(lines []string) string {
	for _, l := range lines {
		for _, re := range secretPatterns {
			if m := re.FindString(l); m != "" && !strings.Contains(m, "[REDACTED]") {
				return re.String()
			}
		}
	}
	return ""
}

// LogBundleRequest is one pending operator pull intent.
type LogBundleRequest struct {
	ScopeID     string
	Component   string
	RequestedAt time.Time
}

// LogBundleMeta is a stored bundle's row without the payload.
type LogBundleMeta struct {
	ID            int
	ScopeID       string
	Component     string
	CollectedAt   time.Time
	ReceivedAt    time.Time
	SizeBytes     int64
	Blocked       bool
	BlockedReason string
}

// RequestLogBundle records (or refreshes) the operator's pending pull for one
// component. One active request per (scope, component) — a re-click refreshes.
func (s *Store) RequestLogBundle(scopeID, component string) error {
	if component != LogBundleComponentController && component != LogBundleComponentAgent {
		return fmt.Errorf("unknown log bundle component %q", component)
	}
	_, err := s.db.Exec(`
		INSERT INTO log_bundle_requests (scope_id, component, requested_at)
		VALUES (?, ?, ?)
		ON CONFLICT(scope_id, component) DO UPDATE SET requested_at = excluded.requested_at`,
		scopeID, component, time.Now().UTC())
	return err
}

// PendingLogBundleRequest reports whether a pull is pending for (scope, component)
// — the value the report ACK / heartbeat envelope advertises.
func (s *Store) PendingLogBundleRequest(scopeID, component string) (bool, error) {
	var n int
	err := s.db.QueryRow(`SELECT COUNT(1) FROM log_bundle_requests WHERE scope_id = ? AND component = ?`,
		scopeID, component).Scan(&n)
	return n > 0, err
}

// GetLogBundleRequests returns the pending requests for a scope (for the UI states).
func (s *Store) GetLogBundleRequests(scopeID string) ([]LogBundleRequest, error) {
	rows, err := s.db.Query(`SELECT scope_id, component, requested_at FROM log_bundle_requests WHERE scope_id = ?`, scopeID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []LogBundleRequest
	for rows.Next() {
		var r LogBundleRequest
		if err := rows.Scan(&r.ScopeID, &r.Component, &r.RequestedAt); err == nil {
			out = append(out, r)
		}
	}
	return out, rows.Err()
}

// SaveLogBundle stores a received component tail: secret-gate → (gzip+store |
// blocked row), prune to the newest logBundleKeep, and CLEAR the pending request
// (consume-once — same shape as SaveAppLogTail). Returns blocked=true when the
// gate fired (nothing stored beyond the flag row).
func (s *Store) SaveLogBundle(scopeID, component string, collectedAt time.Time, lines []string) (blocked bool, err error) {
	lines = capBundleLines(lines, logBundleMaxRawBytes)
	raw := strings.Join(lines, "\n")

	var gz []byte
	reason := ""
	if hit := findSecretPattern(lines); hit != "" {
		// Fail-closed: a secret-shaped token means the content is NOT stored at all.
		blocked = true
		reason = "possible secret (pattern " + hit + ")"
	} else {
		var buf bytes.Buffer
		zw := gzip.NewWriter(&buf)
		if _, err := zw.Write([]byte(raw)); err != nil {
			return false, err
		}
		if err := zw.Close(); err != nil {
			return false, err
		}
		gz = buf.Bytes()
	}

	tx, err := s.db.Begin()
	if err != nil {
		return blocked, err
	}
	defer tx.Rollback()
	if _, err := tx.Exec(`
		INSERT INTO log_bundles (scope_id, component, collected_at, received_at, size_bytes, gz, blocked, blocked_reason)
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
		scopeID, component, collectedAt, time.Now().UTC(), int64(len(raw)), gz, boolInt(blocked), reason); err != nil {
		return blocked, err
	}
	if _, err := tx.Exec(`
		DELETE FROM log_bundles
		WHERE scope_id = ? AND component = ? AND id NOT IN (
			SELECT id FROM log_bundles WHERE scope_id = ? AND component = ?
			ORDER BY id DESC LIMIT ?
		)`, scopeID, component, scopeID, component, logBundleKeep); err != nil {
		return blocked, err
	}
	// Consume-once: the fulfilled request is cleared (a blocked bundle also fulfills —
	// re-requesting the same secret-carrying ring would just block again).
	if _, err := tx.Exec(`DELETE FROM log_bundle_requests WHERE scope_id = ? AND component = ?`,
		scopeID, component); err != nil {
		return blocked, err
	}
	return blocked, tx.Commit()
}

// GetLogBundles returns a scope's stored bundle rows (meta only), newest first.
func (s *Store) GetLogBundles(scopeID string) ([]LogBundleMeta, error) {
	rows, err := s.db.Query(`
		SELECT id, scope_id, component, collected_at, received_at, size_bytes, blocked, blocked_reason
		FROM log_bundles WHERE scope_id = ? ORDER BY id DESC`, scopeID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []LogBundleMeta
	for rows.Next() {
		m, serr := scanBundleMeta(rows)
		if serr != nil {
			continue
		}
		out = append(out, m)
	}
	return out, rows.Err()
}

// GetLogBundleContent returns one bundle's meta + gunzipped lines, scoped (no
// cross-scope reads). A blocked bundle returns meta with nil lines.
func (s *Store) GetLogBundleContent(id int, scopeID string) (*LogBundleMeta, []string, error) {
	row := s.db.QueryRow(`
		SELECT id, scope_id, component, collected_at, received_at, size_bytes, blocked, blocked_reason, gz
		FROM log_bundles WHERE id = ? AND scope_id = ?`, id, scopeID)
	var m LogBundleMeta
	var blockedInt int
	var gz []byte
	if err := row.Scan(&m.ID, &m.ScopeID, &m.Component, &m.CollectedAt, &m.ReceivedAt,
		&m.SizeBytes, &blockedInt, &m.BlockedReason, &gz); err != nil {
		if err == sql.ErrNoRows {
			return nil, nil, nil
		}
		return nil, nil, err
	}
	m.Blocked = blockedInt != 0
	if m.Blocked || len(gz) == 0 {
		return &m, nil, nil
	}
	zr, err := gzip.NewReader(bytes.NewReader(gz))
	if err != nil {
		return &m, nil, err
	}
	raw, err := io.ReadAll(io.LimitReader(zr, logBundleMaxRawBytes+1))
	zr.Close()
	if err != nil {
		return &m, nil, err
	}
	return &m, strings.Split(string(raw), "\n"), nil
}

// PurgeExpiredLogBundles drops bundles older than the 72 h TTL (the honest custody
// bound — pulled logs are transient diagnostics, not an archive). now is injectable
// for the S5 test.
func (s *Store) PurgeExpiredLogBundles(now time.Time) (int, error) {
	res, err := s.db.Exec(`DELETE FROM log_bundles WHERE received_at < ?`, now.Add(-logBundleTTL).UTC())
	if err != nil {
		return 0, err
	}
	n, _ := res.RowsAffected()
	return int(n), nil
}

func scanBundleMeta(rows *sql.Rows) (LogBundleMeta, error) {
	var m LogBundleMeta
	var blockedInt int
	err := rows.Scan(&m.ID, &m.ScopeID, &m.Component, &m.CollectedAt, &m.ReceivedAt,
		&m.SizeBytes, &blockedInt, &m.BlockedReason)
	m.Blocked = blockedInt != 0
	return m, err
}

// capBundleLines keeps the NEWEST lines within the raw byte budget (belt over the
// wire caps; mirrors the box-side newest-kept semantics).
func capBundleLines(lines []string, maxBytes int) []string {
	total := 0
	start := len(lines)
	for i := len(lines) - 1; i >= 0; i-- {
		total += len(lines[i]) + 1
		if total > maxBytes {
			break
		}
		start = i
	}
	return lines[start:]
}

func boolInt(b bool) int {
	if b {
		return 1
	}
	return 0
}