bucketeer/cmd/ctcloud/main.go

// Command ctcloud continuously consumes Certificate Transparency logs through
// embedded CertSpotter internals, classifies cloud/database/storage resources
// in-process and persists only useful, enriched findings to PostgreSQL.
package main

import (
	"context"
	"errors"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"os/signal"
	"runtime"
	"strings"
	"sync/atomic"
	"syscall"
	"time"

	"software.sslmate.com/src/certspotter/monitor"

	"tea.karlcom.de/karl/bucketeer/internal/config"
	"tea.karlcom.de/karl/bucketeer/internal/dns"
	"tea.karlcom.de/karl/bucketeer/internal/index"
	"tea.karlcom.de/karl/bucketeer/internal/metrics"
	"tea.karlcom.de/karl/bucketeer/internal/model"
	"tea.karlcom.de/karl/bucketeer/internal/pipeline"
	"tea.karlcom.de/karl/bucketeer/internal/probe"
	"tea.karlcom.de/karl/bucketeer/internal/scan"
	"tea.karlcom.de/karl/bucketeer/internal/store"
	"tea.karlcom.de/karl/bucketeer/internal/webui"
)

func main() {
	if err := run(); err != nil {
		slog.Error("fatal", "error", err)
		os.Exit(1)
	}
}

func levelFromString(s string) slog.Level {
	switch s {
	case "debug":
		return slog.LevelDebug
	case "warn":
		return slog.LevelWarn
	case "error":
		return slog.LevelError
	default:
		return slog.LevelInfo
	}
}

func run() error {
	cfg, err := config.Load()
	if err != nil {
		return err
	}
	opts := &slog.HandlerOptions{Level: levelFromString(cfg.LogLevel)}
	var handler slog.Handler = slog.NewTextHandler(os.Stderr, opts)
	if cfg.LogFormat == "json" {
		handler = slog.NewJSONHandler(os.Stderr, opts) // machine/LLM friendly
	}
	slog.SetDefault(slog.New(handler))
	if strings.Contains(cfg.DatabaseURL, "sslmode=disable") {
		slog.Warn("insecure database transport", "event", "config_warning",
			"detail", "DATABASE_URL has sslmode=disable; only use this on a private network")
	}
	slog.Info("configuration",
		"event", "startup",
		"database_host", redactURLHost(cfg.DatabaseURL),
		"metrics_addr", cfg.MetricsAddr,
		"ui_auth", cfg.UIUser != "",
		"active_probing", cfg.ActiveProbing,
		"log_list_source", cfg.LogListSource,
		"start_at_end", cfg.StartAtEnd,
		"dns_workers", cfg.DNSWorkers,
		"direct_queue", cfg.DirectQueueSize,
		"dns_queue", cfg.DNSQueueSize,
		"db_queue", cfg.DBQueueSize,
		"db_batch_size", cfg.DBBatchSize,
		"db_flush_interval", cfg.DBFlushInterval.String(),
	)

	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	m := metrics.New()

	st, err := store.Open(ctx, cfg.DatabaseURL, cfg.DBMaxConns)
	if err != nil {
		return err
	}
	defer st.Close()
	st.SetMetrics(m)
	if err := st.Migrate(ctx); err != nil {
		return err
	}
	slog.Info("database ready")

	dnsServers := cfg.DNSServers
	var resolver *dns.ServerPool
	switch {
	case len(dnsServers) == 0:
		// Default: a curated set of reliable public resolvers. Warm it too, so a
		// resolver that is unreachable from this network is cooled down instead
		// of costing a timeout on every random pick.
		resolver = dns.NewServerPool(dns.CuratedServers())
		warmCtx, cancel := context.WithTimeout(ctx, 8*time.Second)
		healthy := resolver.Warm(warmCtx, "example.com", 16, 64)
		cancel()
		slog.Info("dns resolvers configured", "count", resolver.Len(), "healthy", healthy, "set", "curated")
	case len(dnsServers) == 1 && dnsServers[0] == "fresh":
		resolver = dns.NewServerPool(append(dns.CuratedServers(), dns.EmbeddedServers()...))
		warmCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
		healthy := resolver.Warm(warmCtx, "example.com", 32, 256)
		cancel()
		slog.Info("dns resolvers warmed", "total", resolver.Len(), "healthy", healthy, "set", "fresh")
	default:
		resolver = dns.NewServerPool(dnsServers)
		slog.Info("dns resolvers configured", "count", resolver.Len(), "set", "custom")
	}

	enricher := &dns.Enricher{
		Resolver:    resolver,
		Cache:       dns.NewTTLCache[dns.Result](cfg.DNSCacheSize, 64),
		PositiveTTL: cfg.DNSPositiveTTL,
		NegativeTTL: cfg.DNSNegativeTTL,
		Timeout:     cfg.DNSTimeout,
		MaxDepth:    cfg.DNSMaxCNAMEdepth,
		Metrics:     m,
	}

	// One indexer is shared by the CT prober and the on-demand scanner.
	indexer := index.New(cfg.IndexMaxObjects, 2*cfg.ProbeTimeout)

	// Scanner prober: a separate, never-started Prober used synchronously by the
	// on-demand scan endpoint (Probe is stateless).
	scanProber := probe.New(probe.Config{Workers: 1, Timeout: cfg.ProbeTimeout, Deep: cfg.ProbeDeep, Metrics: m})
	// Backfill does one DB write per target, so keep its concurrency near the
	// pool size to avoid starving the live pipeline and the UI.
	backfillWorkers := cfg.ProbeWorkers
	if backfillWorkers > 16 {
		backfillWorkers = 16
	}
	scanner := &scan.Scanner{Prober: scanProber, Indexer: indexer, Enricher: enricher, Store: st, BackfillWorkers: backfillWorkers}

	var prober *probe.Prober
	if cfg.ActiveProbing {
		pcfg := probe.Config{
			Workers: cfg.ProbeWorkers,
			Timeout: cfg.ProbeTimeout,
			Queue:   cfg.ProbeQueueSize,
			Deep:    cfg.ProbeDeep,
			Metrics: m,
			Update: func(uctx context.Context, c model.Candidate, r probe.Result) {
				_ = st.UpdateExposure(uctx, c.Provider.String(), c.Service.String(), c.Hostname,
					r.Exposure, r.Detail, r.Listable, r.ObjectReadable)
			},
		}
		if cfg.IndexEnabled {
			pcfg.Indexer = indexer
			pcfg.OnIndex = func(uctx context.Context, c model.Candidate, entries []model.Entry) {
				if n, err := st.ReplaceEntries(uctx, c.Provider.String(), c.Service.String(), c.Hostname, entries); err != nil {
					slog.Warn("index write failed", "event", "index_error", "host", c.Hostname, "error", err)
				} else if n > 0 {
					slog.Info("indexed readable resource", "event", "indexed",
						"host", c.Hostname, "provider", c.Provider.String(), "service", c.Service.String(), "entries", n)
				}
			}
		}
		prober = probe.New(pcfg)
		prober.Start(ctx)
		slog.Info("active probing enabled", "event", "startup", "indexing", cfg.IndexEnabled, "index_max_objects", cfg.IndexMaxObjects)
	}

	write := func(wctx context.Context, batch []pipeline.Finding) error {
		err := st.WriteFindings(wctx, batch)
		if err == nil && prober != nil {
			for i := range batch {
				prober.Submit(batch[i].Candidate)
			}
		}
		return err
	}

	p := pipeline.New(pipeline.Config{
		DirectQueueSize: cfg.DirectQueueSize,
		DNSQueueSize:    cfg.DNSQueueSize,
		DBQueueSize:     cfg.DBQueueSize,
		DNSWorkers:      cfg.DNSWorkers,
		BatchSize:       cfg.DBBatchSize,
		FlushInterval:   cfg.DBFlushInterval,
		DedupTTL:        cfg.DedupeTTL,
		DedupSize:       cfg.DedupSize,
		Enricher:        enricher,
		Write:           write,
		Metrics:         m,
	})
	p.Start(ctx)

	var monitorReady atomic.Bool
	monitorCfg := &monitor.Config{
		LogListSource:       cfg.LogListSource,
		State:               st,
		StartAtEnd:          cfg.StartAtEnd,
		Verbose:             cfg.LogLevel == "debug",
		HealthCheckInterval: 24 * time.Hour,
		IdentifierSink:      p,
	}

	monitorDone := make(chan error, 1)
	go func() {
		monitorReady.Store(true)
		monitorDone <- monitor.Run(ctx, monitorCfg)
	}()

	ui := webui.New(st, webui.Config{User: cfg.UIUser, Pass: cfg.UIPass, Scanner: scanner})
	httpSrv := startHTTP(cfg.MetricsAddr, m, st, ui, &monitorReady)
	go logStats(ctx, m, p, st, resolver, prober, cfg.StatsInterval)

	slog.Info("ctcloud started")
	err = <-monitorDone
	if err != nil && !errors.Is(err, context.Canceled) {
		slog.Error("ct monitor stopped", "error", err)
	}

	// Graceful shutdown: stop intake, drain in-flight work, flush, then exit.
	slog.Info("shutting down")
	shutdownCtx, cancel := context.WithTimeout(context.Background(), cfg.ShutdownTimeout)
	defer cancel()

	drained := make(chan struct{})
	go func() {
		p.Close()
		if prober != nil {
			prober.Close()
			prober.Wait()
		}
		close(drained)
	}()
	select {
	case <-drained:
	case <-shutdownCtx.Done():
		slog.Warn("shutdown timed out before pipeline drained")
	}

	if err := httpSrv.Shutdown(shutdownCtx); err != nil {
		slog.Warn("http shutdown", "error", err)
	}
	slog.Info("stopped", "certificates", m.CTCertificates.Load(), "candidates", m.Candidates.Load(), "rows", m.DBRows.Load())
	return nil
}

func startHTTP(addr string, m *metrics.Metrics, st *store.Store, ui http.Handler, ready *atomic.Bool) *http.Server {
	mux := http.NewServeMux()
	mux.Handle("/metrics", m.Handler())
	mux.Handle("/", ui)
	mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("ok\n"))
	})
	mux.HandleFunc("/readyz", func(w http.ResponseWriter, r *http.Request) {
		ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
		defer cancel()
		if !ready.Load() {
			http.Error(w, "ct monitor not initialized", http.StatusServiceUnavailable)
			return
		}
		if err := st.Ping(ctx); err != nil {
			http.Error(w, "database unreachable", http.StatusServiceUnavailable)
			return
		}
		w.WriteHeader(http.StatusOK)
		_, _ = w.Write([]byte("ready\n"))
	})
	srv := &http.Server{
		Addr:              addr,
		Handler:           mux,
		ReadHeaderTimeout: 5 * time.Second,
		ReadTimeout:       10 * time.Second,
		WriteTimeout:      10 * time.Second,
		IdleTimeout:       60 * time.Second,
	}
	go func() {
		if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
			slog.Error("metrics server", "error", err)
		}
	}()
	return srv
}

// logStats emits one structured record per interval with a stable key
// vocabulary, so it can be parsed by an LLM or a log pipeline. It never logs per
// certificate; per-certificate hot-path events are only counted.
func logStats(ctx context.Context, m *metrics.Metrics, p *pipeline.Pipeline, st *store.Store, resolver *dns.ServerPool, prober *probe.Prober, interval time.Duration) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			var ms runtime.MemStats
			runtime.ReadMemStats(&ms)
			direct, dnsQ, dbQ := p.Depths()
			dbTotal, dbIdle, dbAcquired := st.PoolStats()
			attrs := []any{
				"event", "stats",
				"certificates", m.CTCertificates.Load(),
				"identifiers", m.CTIdentifiers.Load(),
				"direct_matches", m.DirectMatches.Load(),
				"cname_matches", m.CNAMEMatches.Load(),
				"candidates", m.Candidates.Load(),
				"dns_queries", m.DNSQueries.Load(),
				"dns_cache_hits", m.DNSCacheHits.Load(),
				"dns_errors", m.DNSErrors.Load(),
				"dns_resolvers", resolver.Len(),
				"ct_errors", m.CTErrors.Load(),
				"ct_malformed", m.MalformedCT.Load(),
				"db_rows", m.DBRows.Load(),
				"db_errors", m.DBErrors.Load(),
				"db_pool_total", dbTotal,
				"db_pool_idle", dbIdle,
				"db_pool_acquired", dbAcquired,
				"queue_direct", direct,
				"queue_dns", dnsQ,
				"queue_db", dbQ,
				"queue_drops", m.QueueDrops.Load(),
				"goroutines", runtime.NumGoroutine(),
				"mem_sys_bytes", ms.Sys,
				"mem_heap_bytes", ms.HeapInuse,
				"gc_cycles", ms.NumGC,
			}
			if prober != nil {
				attrs = append(attrs,
					"probe_queue", prober.QueueLen(),
					"probe_attempts", m.ProbeAttempts.Load(),
					"probe_public", m.ProbePublic.Load(),
				)
			}
			slog.Info("periodic stats", attrs...)
		}
	}
}

func redactURLHost(raw string) string {
	u, err := url.Parse(raw)
	if err != nil || u.Host == "" {
		return "unparseable"
	}
	return u.Host
}