bucket discovery engine: CT domain -> candidate names -> listing index

CMS SANs are custom domains, not bucket names, so matching SANs finds almost
no buckets. This adds internal/discover: from each non-matching hostname it
derives candidate names (domain label + subdomain label + ~40 suffixes) plus a
built-in wordlist, and anonymously tests listing on S3, GCS and (optionally)
DigitalOcean Spaces. Every 200 is indexed (keys/sizes/dates only).

Bounded hard: global BUCKET_SCAN_QPS rate limit, 12h per-name dedup, random
sampled domain queue, MaxPerDomain cap. Never downloads contents.

Fix: GCS candidates now use the virtual-host form
(<bucket>.storage.googleapis.com) so buckets no longer collide on the
cloud_assets unique key; previously they all shared storage.googleapis.com and
overwrote each other.

Live: 30 open buckets in ~5 min (e.g. estimator-dev, image-prod,
update-media, mos-images S3; several GCS buckets), 11k+ objects indexed.

cmd/ctcloud/main.go

@@ -20,6 +20,7 @@ import (
 	"software.sslmate.com/src/certspotter/monitor"
 
 	"tea.karlcom.de/karl/bucketeer/internal/config"
+	"tea.karlcom.de/karl/bucketeer/internal/discover"
 	"tea.karlcom.de/karl/bucketeer/internal/dns"
 	"tea.karlcom.de/karl/bucketeer/internal/index"
 	"tea.karlcom.de/karl/bucketeer/internal/metrics"
@@ -145,6 +146,35 @@ func run() error {
 	}
 	scanner := &scan.Scanner{Prober: scanProber, Indexer: indexer, Enricher: enricher, Store: st, BackfillWorkers: backfillWorkers}
 
+	var discoverer *discover.Discoverer
+	if cfg.BucketScanEnabled {
+		discoverer = discover.New(discover.Config{
+			QPS: cfg.BucketScanQPS, Workers: cfg.BucketScanWorkers, Queue: cfg.BucketScanQueue,
+			S3: true, GCS: true, DOSpaces: cfg.BucketScanDO,
+			DORegions: []string{"nyc3", "sfo3", "fra1", "ams3", "sgp1"},
+			Indexer:   indexer, Metrics: m,
+			OnOpen: func(uctx context.Context, c model.Candidate, detail string, entries []model.Entry) {
+				if _, err := st.EnsureAsset(uctx, c); err != nil {
+					return
+				}
+				_ = st.UpdateExposure(uctx, c.Provider.String(), c.Service.String(), c.Hostname,
+					model.ExposurePublicRead, detail, true, true)
+				if n, err := st.ReplaceEntries(uctx, c.Provider.String(), c.Service.String(), c.Hostname, entries); err == nil && n > 0 {
+					m.EntriesIndexed.Add(uint64(n))
+				}
+				slog.Info("open bucket discovered", "event", "open_bucket",
+					"host", c.Hostname, "provider", c.Provider.String(), "service", c.Service.String(), "entries", len(entries))
+			},
+		})
+		discoverer.Start(ctx)
+		for _, name := range discover.Wordlist() {
+			discoverer.SubmitName(name)
+		}
+		slog.Info("bucket name discovery enabled", "event", "startup",
+			"qps", cfg.BucketScanQPS, "workers", cfg.BucketScanWorkers, "do_spaces", cfg.BucketScanDO,
+			"wordlist", len(discover.Wordlist()))
+	}
+
 	var prober *probe.Prober
 	if cfg.ActiveProbing {
 		pcfg := probe.Config{
@@ -184,6 +214,10 @@ func run() error {
 		return err
 	}
 
+	var discoverField interface{ Submit(host string) }
+	if discoverer != nil {
+		discoverField = discoverer
+	}
 	p := pipeline.New(pipeline.Config{
 		DirectQueueSize: cfg.DirectQueueSize,
 		DNSQueueSize:    cfg.DNSQueueSize,
@@ -196,6 +230,7 @@ func run() error {
 		Enricher:        enricher,
 		Write:           write,
 		Metrics:         m,
+		Discover:        discoverField,
 	})
 	p.Start(ctx)
 
@@ -237,6 +272,10 @@ func run() error {
 			prober.Close()
 			prober.Wait()
 		}
+		if discoverer != nil {
+			discoverer.Close()
+			discoverer.Wait()
+		}
 		close(drained)
 	}()
 	select {
@@ -339,6 +378,10 @@ func logStats(ctx context.Context, m *metrics.Metrics, p *pipeline.Pipeline, st
 					"probe_public", m.ProbePublic.Load(),
 				)
 			}
+			attrs = append(attrs,
+				"bucket_candidates", m.BucketCandidates.Load(),
+				"bucket_open", m.BucketOpen.Load(),
+			)
 			slog.Info("periodic stats", attrs...)
 		}
 	}