bucketeer/internal/discover/discover.go

// Package discover turns the domains seen in CT into candidate object-storage
// bucket names and tests them anonymously. This is the engine that actually
// finds open buckets: certificate SANs are custom domains, not bucket names, so
// we generate plausible names and ask the providers' list APIs.
//
// It is aggressively bounded: a global rate limiter caps outbound requests,
// candidate names are deduplicated with a TTL cache, and the domain intake is a
// random sample so no single domain floods the queue. Contents are never
// downloaded; only the list API is called.
package discover

import (
	"context"
	"io"
	"math/rand/v2"
	"net/http"
	"strings"
	"sync"
	"time"

	"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"
)

type Config struct {
	QPS          int // global outbound request rate
	Workers      int // concurrent candidate probes
	Queue        int // sampled domain backlog
	S3           bool
	GCS          bool
	DOSpaces     bool
	DORegions    []string
	MaxPerDomain int // cap candidates generated per domain

	Indexer *index.Indexer
	Metrics *metrics.Metrics
	// OnOpen is called for every anonymously listable bucket with its metadata
	// index. contents are never downloaded.
	OnOpen func(ctx context.Context, c model.Candidate, detail string, entries []model.Entry)
}

type Discoverer struct {
	cfg    Config
	client *http.Client
	q      *domainQueue
	tokens chan struct{}
	seen   *dns.TTLCache[bool] // candidate names already tested
	stop   chan struct{}
	wg     sync.WaitGroup
}

func New(cfg Config) *Discoverer {
	if cfg.QPS <= 0 {
		cfg.QPS = 50
	}
	if cfg.Workers <= 0 {
		cfg.Workers = 16
	}
	if cfg.Queue <= 0 {
		cfg.Queue = 65536
	}
	if cfg.MaxPerDomain <= 0 {
		cfg.MaxPerDomain = 48
	}
	return &Discoverer{
		cfg:    cfg,
		client: &http.Client{Timeout: 6 * time.Second},
		q:      newDomainQueue(cfg.Queue),
		tokens: make(chan struct{}, max(cfg.QPS, 1)),
		seen:   dns.NewTTLCache[bool](1<<20, 64),
		stop:   make(chan struct{}),
	}
}

// Submit offers a hostname to the discovery engine. Duplicate registrable
// domains and already-tested candidate names are dropped cheaply.
func (d *Discoverer) Submit(host string) {
	d.q.push(host)
}

func (d *Discoverer) Start(ctx context.Context) {
	go d.refill(ctx)
	for i := 0; i < d.cfg.Workers; i++ {
		d.wg.Add(1)
		go d.worker(ctx)
	}
}

func (d *Discoverer) Close() { close(d.stop) }
func (d *Discoverer) Wait()  { d.wg.Wait() }

// QueueLen reports the sampled domain backlog.
func (d *Discoverer) QueueLen() int { return d.q.len() }

func (d *Discoverer) refill(ctx context.Context) {
	period := time.Second / time.Duration(d.cfg.QPS)
	if period <= 0 {
		period = time.Millisecond
	}
	t := time.NewTicker(period)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-d.stop:
			return
		case <-t.C:
			select {
			case d.tokens <- struct{}{}:
			default:
			}
		}
	}
}

func (d *Discoverer) acquire(ctx context.Context) bool {
	select {
	case <-ctx.Done():
		return false
	case <-d.stop:
		return false
	case <-d.tokens:
		return true
	}
}

func (d *Discoverer) worker(ctx context.Context) {
	defer d.wg.Done()
	for {
		select {
		case <-ctx.Done():
			return
		case <-d.stop:
			return
		case <-d.q.notify:
		}
		for {
			host, ok := d.q.pop()
			if !ok {
				break
			}
			d.scanDomain(ctx, host)
		}
	}
}

func (d *Discoverer) scanDomain(ctx context.Context, host string) {
	for _, name := range candidates(host, d.cfg.MaxPerDomain) {
		if ctx.Err() != nil {
			return
		}
		d.testName(ctx, name)
	}
}

func (d *Discoverer) testName(ctx context.Context, name string) {
	if _, dup := d.seen.Get(name); dup {
		return
	}
	d.seen.Set(name, true, 12*time.Hour)
	for _, t := range d.targets(name) {
		if d.cfg.Metrics != nil {
			d.cfg.Metrics.BucketCandidates.Add(1)
		}
		if !d.acquire(ctx) {
			return
		}
		d.test(ctx, t)
	}
}

// Target is one provider listing endpoint for a candidate name.
type Target struct {
	Name     string
	Provider model.Provider
	Service  model.Service
	Hostname string // hostname for the indexer/probe candidate
	URL      string // anonymous list URL
	Region   string
}

func (d *Discoverer) targets(name string) []Target {
	var out []Target
	if d.cfg.S3 {
		out = append(out, Target{
			Name: name, Provider: model.ProviderAWS, Service: model.ServiceS3,
			Hostname: name + ".s3.amazonaws.com",
			URL:      "https://" + name + ".s3.amazonaws.com/?list-type=2&max-keys=1",
		})
	}
	if d.cfg.GCS {
		// Use the virtual-host form so each bucket has a unique hostname (the
		// path form makes every bucket share storage.googleapis.com and collide
		// on the cloud_assets unique key).
		out = append(out, Target{
			Name: name, Provider: model.ProviderGCP, Service: model.ServiceObjectStorage,
			Hostname: name + ".storage.googleapis.com",
			URL:      "https://storage.googleapis.com/" + name + "?max-keys=1",
		})
	}
	if d.cfg.DOSpaces {
		for _, region := range d.cfg.DORegions {
			host := name + "." + region + ".digitaloceanspaces.com"
			out = append(out, Target{
				Name: name, Provider: model.ProviderDigitalOcean, Service: model.ServiceObjectStorage,
				Hostname: host, Region: region,
				URL: "https://" + host + "/?list-type=2&max-keys=1",
			})
		}
	}
	return out
}

func (d *Discoverer) test(ctx context.Context, t Target) {
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, t.URL, nil)
	if err != nil {
		return
	}
	resp, err := d.client.Do(req)
	if err != nil {
		return
	}
	_, _ = io.CopyN(io.Discard, resp.Body, 512)
	finalHost := resp.Request.URL.Host
	resp.Body.Close()
	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
		return
	}
	// Listable. Use the post-redirect host (S3 global redirects to a region).
	c := model.Candidate{
		Hostname: t.Hostname, Original: t.Hostname, Resource: t.Name, Region: t.Region,
		Provider: t.Provider, Service: t.Service, ResourceSpecific: true, Source: model.SourceDiscovery,
	}
	if finalHost != "" && strings.HasSuffix(finalHost, ".amazonaws.com") {
		c.Hostname = finalHost
	}
	detail := "discovered open bucket via name enumeration (" + t.Hostname + ")"
	if d.cfg.Metrics != nil {
		d.cfg.Metrics.BucketOpen.Add(1)
	}
	if d.cfg.OnOpen == nil {
		return
	}
	var entries []model.Entry
	if d.cfg.Indexer != nil {
		if e, err := d.cfg.Indexer.List(ctx, c); err == nil {
			entries = e
		}
	}
	d.cfg.OnOpen(ctx, c, detail, entries)
}

// --- candidate generation ---

// No name guessing: candidates are exactly the labels that occur in the incoming
// hostname (the subdomain label and the registrable-domain label). Nothing is
// fabricated, permuted, or drawn from a wordlist.

// baseLabels returns plausible bucket-name stems from a hostname: the leftmost
// label and the registrable-domain label.
func baseLabels(host string) []string {
	host = strings.ToLower(strings.TrimSuffix(strings.TrimSpace(host), "."))
	labels := strings.Split(host, ".")
	if len(labels) < 2 {
		return nil
	}
	cands := []string{labels[0], labels[len(labels)-2]}
	seen := map[string]bool{}
	out := cands[:0]
	for _, c := range cands {
		if validLabel(c) && !seen[c] && !skipLabel[c] {
			seen[c] = true
			out = append(out, c)
		}
	}
	return out
}

// Only genuinely non-descriptive labels are skipped; service-ish subdomains
// (assets, cdn, media, ...) are kept because they are common bucket names.
var skipLabel = map[string]bool{
	"www": true, "api": true, "mail": true, "email": true, "smtp": true,
	"ns": true, "mx": true, "localhost": true,
}

func candidates(host string, max int) []string {
	out := baseLabels(host)
	if max > 0 && len(out) > max {
		out = out[:max]
	}
	return out
}

func validLabel(s string) bool {
	if len(s) < 3 || len(s) > 40 {
		return false
	}
	for i := 0; i < len(s); i++ {
		c := s[i]
		if !(c >= 'a' && c <= 'z' || c >= '0' && c <= '9' || c == '-') {
			return false
		}
	}
	return s[0] != '-' && s[len(s)-1] != '-'
}

func validBucketName(s string) bool {
	if len(s) < 3 || len(s) > 63 {
		return false
	}
	for i := 0; i < len(s); i++ {
		c := s[i]
		if !(c >= 'a' && c <= 'z' || c >= '0' && c <= '9' || c == '-' || c == '.') {
			return false
		}
	}
	return s[0] != '-' && s[0] != '.' && s[len(s)-1] != '-' && s[len(s)-1] != '.'
}

// --- sampled domain queue ---

type domainQueue struct {
	mu     sync.Mutex
	items  []string
	cap    int
	notify chan struct{}
}

func newDomainQueue(capacity int) *domainQueue {
	return &domainQueue{cap: capacity, notify: make(chan struct{}, 1)}
}

func (q *domainQueue) push(s string) {
	q.mu.Lock()
	if len(q.items) < q.cap {
		q.items = append(q.items, s)
	} else {
		q.items[rand.IntN(len(q.items))] = s
	}
	q.mu.Unlock()
	select {
	case q.notify <- struct{}{}:
	default:
	}
}

func (q *domainQueue) pop() (string, bool) {
	q.mu.Lock()
	n := len(q.items)
	if n == 0 {
		q.mu.Unlock()
		return "", false
	}
	i := rand.IntN(n)
	s := q.items[i]
	q.items[i] = q.items[n-1]
	q.items = q.items[:n-1]
	q.mu.Unlock()
	return s, true
}

func (q *domainQueue) len() int {
	q.mu.Lock()
	defer q.mu.Unlock()
	return len(q.items)
}