ctcloud: CertSpotter-backed CT cloud-resource monitor

Single stateless Go process that consumes CT logs via embedded (patched)
CertSpotter internals, classifies cloud/database/storage DNS identifiers
in-process with a zero-allocation classifier, enriches via round-robin DNS
resolvers with CNAME following, and batch-writes useful findings to PostgreSQL.

- third_party/certspotter: 2-hunk IdentifierSink patch, no stdout/JSON/files
- internal/classify: AWS/Azure/GCP/Firebase/Cloudflare R2/DigitalOcean/Supabase/OCI
- internal/dns: sharded bounded TTL cache, multi-resolver load balancing + failover
- internal/pipeline: bounded queues, dedup, direct-priority load shedding
- internal/store: pgx batch writer, Postgres CT checkpoints (no cert contents)
- internal/probe: optional non-destructive reachability checks (off by default)
- Docker: distroless non-root, read-only rootfs, no app volume, resource limits
- tests enforce 0 allocs/op on the classifier and sink hot path
This commit is contained in:
StasonJatham committed 2026-09-22 13:13:10 +02:00
1 parent 9e78a114ac
commit e8f7d6bfdf
118 files changed
+16473

No files matched your search

+345
View File
@@ -0,0 +1,345 @@
// Package pipeline wires the in-process CertSpotter identifier callback to the
// classifier, DNS enrichment and the batched PostgreSQL writer using bounded
// queues throughout.
package pipeline
import (
"context"
"net"
"sync"
"time"
"software.sslmate.com/src/certspotter"
"tea.karlcom.de/karl/bucketeer/internal/classify"
"tea.karlcom.de/karl/bucketeer/internal/dns"
"tea.karlcom.de/karl/bucketeer/internal/metrics"
"tea.karlcom.de/karl/bucketeer/internal/model"
)
// Finding is a fully enriched candidate ready to be persisted.
type Finding struct {
model.Candidate
CNAMEChain []string
IPv4 []string
IPv6 []string
}
// WriteFunc persists a batch of findings.
type WriteFunc func(ctx context.Context, batch []Finding) error
type Config struct {
DirectQueueSize int
DNSQueueSize int
DBQueueSize int
DirectWorkers int
DNSWorkers int
BatchSize int
FlushInterval time.Duration
DedupTTL time.Duration
DedupSize int
Enricher *dns.Enricher
Write WriteFunc
Metrics *metrics.Metrics
}
// Pipeline is the certificate-to-database data flow. It is safe to call
// HandleIdentifiers from the CertSpotter monitor goroutines concurrently.
type Pipeline struct {
cfg Config
directQ chan model.Candidate
dnsQ chan string
dbQ chan Finding
dedup *dns.TTLCache[bool] // hostnames already queued for DNS
seen *dns.TTLCache[bool] // provider+service+provider_hostname already written
producer sync.WaitGroup
workers sync.WaitGroup
}
func New(cfg Config) *Pipeline {
if cfg.DirectQueueSize <= 0 {
cfg.DirectQueueSize = 8192
}
if cfg.DNSQueueSize <= 0 {
cfg.DNSQueueSize = 8192
}
if cfg.DBQueueSize <= 0 {
cfg.DBQueueSize = 8192
}
if cfg.DirectWorkers <= 0 {
cfg.DirectWorkers = 2
}
if cfg.DNSWorkers <= 0 {
cfg.DNSWorkers = 16
}
if cfg.BatchSize <= 0 {
cfg.BatchSize = 500
}
if cfg.FlushInterval <= 0 {
cfg.FlushInterval = 2 * time.Second
}
if cfg.DedupTTL <= 0 {
cfg.DedupTTL = time.Hour
}
if cfg.DedupSize <= 0 {
cfg.DedupSize = 1 << 20
}
p := &Pipeline{
cfg: cfg,
directQ: make(chan model.Candidate, cfg.DirectQueueSize),
dnsQ: make(chan string, cfg.DNSQueueSize),
dbQ: make(chan Finding, cfg.DBQueueSize),
dedup: dns.NewTTLCache[bool](cfg.DedupSize, 64),
seen: dns.NewTTLCache[bool](cfg.DedupSize, 64),
}
if p.cfg.Metrics != nil {
p.cfg.Metrics.SetQueueDepth("direct", func() float64 { return float64(len(p.directQ)) })
p.cfg.Metrics.SetQueueDepth("dns", func() float64 { return float64(len(p.dnsQ)) })
p.cfg.Metrics.SetQueueDepth("db", func() float64 { return float64(len(p.dbQ)) })
}
return p
}
// Start launches the worker pools. Close must be called to drain.
func (p *Pipeline) Start(ctx context.Context) {
for i := 0; i < p.cfg.DirectWorkers; i++ {
p.producer.Add(1)
go p.directWorker(ctx)
}
for i := 0; i < p.cfg.DNSWorkers; i++ {
p.producer.Add(1)
go p.dnsWorker(ctx)
}
p.workers.Add(1)
go p.dbWorker(ctx)
}
// Close stops intake and drains all queues. It must only be called after the
// CertSpotter monitor has stopped calling HandleIdentifiers.
func (p *Pipeline) Close() {
close(p.directQ)
close(p.dnsQ)
p.producer.Wait()
close(p.dbQ)
p.workers.Wait()
}
// HandleIdentifiers implements certspotter.IdentifierSink. It is deliberately
// allocation-light and never blocks: queues are bounded and overflow is dropped.
func (p *Pipeline) HandleIdentifiers(ctx context.Context, ids *certspotter.Identifiers) {
if p.cfg.Metrics != nil {
p.cfg.Metrics.CTCertificates.Add(1)
p.cfg.Metrics.CTIdentifiers.Add(uint64(len(ids.DNSNames)))
}
for _, host := range ids.DNSNames {
p.HandleHost(host)
}
}
// HandleHost classifies a single hostname and enqueues it for direct handling
// or, when it does not match, for DNS/CNAME enrichment.
func (p *Pipeline) HandleHost(host string) {
if m, ok := classify.Classify(host); ok {
if !m.ResourceSpecific {
// Generic service endpoints (bucket name in the path) are not
// customer resources; do not create findings for them.
return
}
if p.cfg.Metrics != nil {
p.cfg.Metrics.DirectMatches.Add(1)
p.cfg.Metrics.Candidates.Add(1)
}
if !p.markSeen(host) {
return
}
cand := candidateFrom(host, host, m, model.SourceDirectCT)
select {
case p.directQ <- cand:
default:
p.drop()
}
return
}
if !classify.ShouldEnrich(host) {
return
}
if _, dup := p.dedup.Get(host); dup {
return
}
p.dedup.Set(host, true, p.cfg.DedupTTL)
select {
case p.dnsQ <- host:
default:
p.drop()
}
}
func (p *Pipeline) drop() {
if p.cfg.Metrics != nil {
p.cfg.Metrics.QueueFull.Add(1)
p.cfg.Metrics.QueueDrops.Add(1)
}
}
func candidateFrom(host, original string, m model.Match, src model.Source) model.Candidate {
return model.Candidate{
Hostname: host,
Original: original,
Region: m.Region,
Resource: m.Resource,
Provider: m.Provider,
Service: m.Service,
Source: src,
PublicSemantic: m.PublicSemantic,
ResourceSpecific: m.ResourceSpecific,
}
}
func appendIPs(f *Finding, ips []net.IP) {
for _, ip := range ips {
if v4 := ip.To4(); v4 != nil {
f.IPv4 = append(f.IPv4, v4.String())
} else {
f.IPv6 = append(f.IPv6, ip.String())
}
}
}
func (p *Pipeline) enrich(ctx context.Context, c model.Candidate) Finding {
f := Finding{Candidate: c}
if p.cfg.Enricher == nil {
return f
}
res, ok := p.cfg.Enricher.Resolve(ctx, c.Hostname)
if !ok {
return f
}
f.CNAMEChain = res.Chain
appendIPs(&f, res.IPs)
return f
}
func (p *Pipeline) directWorker(ctx context.Context) {
defer p.producer.Done()
for {
select {
case <-ctx.Done():
return
case c, ok := <-p.directQ:
if !ok {
return
}
p.pushDB(ctx, p.enrich(ctx, c))
}
}
}
func (p *Pipeline) dnsWorker(ctx context.Context) {
defer p.producer.Done()
for {
select {
case <-ctx.Done():
return
case host, ok := <-p.dnsQ:
if !ok {
return
}
if len(p.dbQ) >= cap(p.dbQ)/2 {
// Load shedding: optional CNAME enrichment is dropped before
// direct matches when the database writer is backed up.
p.drop()
continue
}
p.enrichHost(ctx, host)
}
}
}
func (p *Pipeline) enrichHost(ctx context.Context, host string) {
res, ok := p.cfg.Enricher.Resolve(ctx, host)
if !ok {
return
}
for _, target := range res.Chain[1:] {
m, matched := classify.Classify(target)
if !matched || !m.ResourceSpecific {
continue
}
if p.cfg.Metrics != nil {
p.cfg.Metrics.CNAMEMatches.Add(1)
p.cfg.Metrics.Candidates.Add(1)
}
if !p.markSeen(target) {
continue
}
f := Finding{
Candidate: candidateFrom(target, host, m, model.SourceCNAME),
CNAMEChain: res.Chain,
}
appendIPs(&f, res.IPs)
p.pushDB(ctx, f)
}
}
// markSeen dedups by hostname alone: a given hostname always classifies to the
// same provider/service, so no key concatenation (and no allocation) is needed.
func (p *Pipeline) markSeen(host string) bool {
if _, dup := p.seen.Get(host); dup {
return false
}
p.seen.Set(host, true, p.cfg.DedupTTL)
return true
}
func (p *Pipeline) pushDB(ctx context.Context, f Finding) {
select {
case p.dbQ <- f:
default:
p.drop()
}
}
func (p *Pipeline) dbWorker(ctx context.Context) {
defer p.workers.Done()
batch := make([]Finding, 0, p.cfg.BatchSize)
ticker := time.NewTicker(p.cfg.FlushInterval)
defer ticker.Stop()
flush := func() {
if len(batch) == 0 {
return
}
if p.cfg.Write != nil {
if err := p.cfg.Write(ctx, batch); err != nil {
if p.cfg.Metrics != nil {
p.cfg.Metrics.DBErrors.Add(1)
}
} else if p.cfg.Metrics != nil {
p.cfg.Metrics.DBRows.Add(uint64(len(batch)))
}
}
batch = batch[:0]
}
for {
select {
case <-ctx.Done():
flush()
return
case f, ok := <-p.dbQ:
if !ok {
flush()
return
}
batch = append(batch, f)
if len(batch) >= p.cfg.BatchSize {
flush()
}
case <-ticker.C:
flush()
}
}
}
+236
View File
@@ -0,0 +1,236 @@
package pipeline
import (
"context"
"net"
"strconv"
"sync"
"testing"
"time"
"software.sslmate.com/src/certspotter"
"tea.karlcom.de/karl/bucketeer/internal/dns"
"tea.karlcom.de/karl/bucketeer/internal/metrics"
"tea.karlcom.de/karl/bucketeer/internal/model"
)
func makeHosts(prefix, suffix string) []string {
hosts := make([]string, 256)
for i := range hosts {
hosts[i] = prefix + strconv.Itoa(i) + suffix
}
return hosts
}
// TestHandleHostNoAllocs audits the sink hot path: after warmup, classifying and
// enqueueing a hostname must not allocate.
func TestHandleHostNoAllocs(t *testing.T) {
p := New(Config{DNSQueueSize: 1 << 20, DedupSize: 1 << 20})
nonMatch := makeHosts("api", ".example.com")
direct := makeHosts("bucket", ".s3.eu-central-1.amazonaws.com")
for _, h := range nonMatch {
p.HandleHost(h)
}
for _, h := range direct {
p.HandleHost(h)
}
allocs := testing.AllocsPerRun(2000, func() {
for i := 0; i < 64; i++ {
p.HandleHost(nonMatch[i])
p.HandleHost(direct[i])
}
})
if allocs != 0 {
t.Fatalf("HandleHost allocates %.3f allocs/op, want 0", allocs)
}
}
func BenchmarkHandleHostNonMatch(b *testing.B) {
p := New(Config{DNSQueueSize: 1 << 20, DedupSize: 1 << 20})
hosts := makeHosts("api", ".example.com")
for _, h := range hosts {
p.HandleHost(h)
}
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
p.HandleHost(hosts[i&255])
}
}
func BenchmarkHandleHostDirectMatch(b *testing.B) {
p := New(Config{DirectQueueSize: 1 << 20, DedupSize: 1 << 20})
hosts := makeHosts("bucket", ".s3.eu-central-1.amazonaws.com")
for _, h := range hosts {
p.HandleHost(h)
}
b.ReportAllocs()
b.ResetTimer()
for i := 0; i < b.N; i++ {
p.HandleHost(hosts[i&255])
}
}
type fakeResolver struct {
cnames map[string]string
ips map[string][]net.IP
}
func (f *fakeResolver) LookupCNAME(ctx context.Context, host string) (string, error) {
if c, ok := f.cnames[host]; ok {
return c, nil
}
return host, nil
}
func (f *fakeResolver) LookupIP(ctx context.Context, host string) ([]net.IP, error) {
return f.ips[host], nil
}
func newEnricher(r dns.Resolver) *dns.Enricher {
return &dns.Enricher{
Resolver: r,
Cache: dns.NewTTLCache[dns.Result](1024, 8),
PositiveTTL: time.Minute,
NegativeTTL: time.Minute,
MaxDepth: 8,
}
}
type collector struct {
mu sync.Mutex
items []Finding
}
func (c *collector) write(ctx context.Context, batch []Finding) error {
c.mu.Lock()
defer c.mu.Unlock()
c.items = append(c.items, batch...)
return nil
}
func (c *collector) all() []Finding {
c.mu.Lock()
defer c.mu.Unlock()
return append([]Finding(nil), c.items...)
}
func TestDirectMatch(t *testing.T) {
col := &collector{}
m := metrics.New()
p := New(Config{
Enricher: newEnricher(&fakeResolver{}),
Write: col.write,
Metrics: m,
BatchSize: 10,
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
p.Start(ctx)
p.HandleIdentifiers(ctx, &certspotter.Identifiers{DNSNames: []string{"bucket.s3.eu-central-1.amazonaws.com"}})
time.Sleep(50 * time.Millisecond)
p.Close()
items := col.all()
if len(items) != 1 {
t.Fatalf("got %d findings, want 1 (%+v)", len(items), items)
}
f := items[0]
if f.Provider != model.ProviderAWS || f.Service != model.ServiceS3 || f.Source != model.SourceDirectCT {
t.Fatalf("unexpected finding %+v", f)
}
if m.DirectMatches.Load() != 1 {
t.Fatalf("direct matches = %d, want 1", m.DirectMatches.Load())
}
}
func TestCNAMEMatch(t *testing.T) {
col := &collector{}
r := &fakeResolver{
cnames: map[string]string{"files.example.com": "files.s3.amazonaws.com"},
ips: map[string][]net.IP{"files.example.com": {net.ParseIP("1.2.3.4")}},
}
m := metrics.New()
p := New(Config{Enricher: newEnricher(r), Write: col.write, Metrics: m, BatchSize: 10})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
p.Start(ctx)
p.HandleIdentifiers(ctx, &certspotter.Identifiers{DNSNames: []string{"files.example.com"}})
time.Sleep(50 * time.Millisecond)
p.Close()
items := col.all()
if len(items) != 1 {
t.Fatalf("got %d findings, want 1", len(items))
}
f := items[0]
if f.Source != model.SourceCNAME || f.Hostname != "files.s3.amazonaws.com" || f.Original != "files.example.com" {
t.Fatalf("unexpected finding %+v", f)
}
if len(f.IPv4) != 1 || f.IPv4[0] != "1.2.3.4" {
t.Fatalf("unexpected IPs %+v", f.IPv4)
}
if m.CNAMEMatches.Load() != 1 {
t.Fatalf("cname matches = %d, want 1", m.CNAMEMatches.Load())
}
}
func TestCNAMEMatchIPv6(t *testing.T) {
col := &collector{}
r := &fakeResolver{
cnames: map[string]string{"db.example.com": "mydb.cluster-abc.us-east-1.rds.amazonaws.com"},
ips: map[string][]net.IP{"db.example.com": {net.ParseIP("2606:4700::1111")}},
}
p := New(Config{Enricher: newEnricher(r), Write: col.write, BatchSize: 10})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
p.Start(ctx)
p.HandleHost("db.example.com")
time.Sleep(50 * time.Millisecond)
p.Close()
items := col.all()
if len(items) != 1 || len(items[0].IPv6) != 1 || items[0].IPv6[0] != "2606:4700::1111" {
t.Fatalf("unexpected IPv6 finding: %+v", items)
}
}
func TestQueueSaturationDrops(t *testing.T) {
m := metrics.New()
p := New(Config{DirectQueueSize: 1, DNSQueueSize: 1, Metrics: m})
// Intentionally do not Start: the queue stays full.
p.HandleHost("bucket.s3.amazonaws.com")
p.HandleHost("other.s3.amazonaws.com")
if m.QueueDrops.Load() == 0 {
t.Fatal("expected a drop when the direct queue is full")
}
if m.Candidates.Load() != 2 {
t.Fatalf("candidates = %d, want 2", m.Candidates.Load())
}
}
func TestDNSDedup(t *testing.T) {
m := metrics.New()
p := New(Config{DNSQueueSize: 8, Metrics: m})
p.HandleHost("files.example.com")
p.HandleHost("files.example.com")
if got := len(p.dnsQ); got != 1 {
t.Fatalf("dns queue len = %d, want 1", got)
}
}
func TestGenericEndpointNotStored(t *testing.T) {
col := &collector{}
p := New(Config{Write: col.write, BatchSize: 10})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
p.Start(ctx)
p.HandleIdentifiers(ctx, &certspotter.Identifiers{DNSNames: []string{
"objectstorage.us-ashburn-1.oraclecloud.com",
}})
time.Sleep(30 * time.Millisecond)
p.Close()
if items := col.all(); len(items) != 0 {
t.Fatalf("generic endpoint produced findings: %+v", items)
}
}