god-eye/internal/modules/cloud/cloud.go

// Package cloud wraps v1 cloud detection + S3 bucket discovery.
// Drains the store, plus listens for late DNSResolved events.
package cloud

import (
	"context"
	"sync"
	"time"

	"god-eye/internal/eventbus"
	gohttp "god-eye/internal/http"
	"god-eye/internal/module"
	"god-eye/internal/scanner"
	"god-eye/internal/store"
)

const ModuleName = "cloud.detect"

type cloudModule struct{}

func Register() { module.Register(&cloudModule{}) }

func (*cloudModule) Name() string        { return ModuleName }
func (*cloudModule) Phase() module.Phase { return module.PhaseAnalysis }
func (*cloudModule) Consumes() []eventbus.EventType {
	return []eventbus.EventType{eventbus.EventDNSResolved, eventbus.EventHTTPProbed}
}
func (*cloudModule) Produces() []eventbus.EventType { return []eventbus.EventType{eventbus.EventCloudAsset} }
func (*cloudModule) DefaultEnabled() bool           { return true }

func (*cloudModule) Run(mctx module.Context) error {
	timeout := mctx.Config.Int("timeout", 5)
	client := gohttp.GetSharedClient(timeout)

	handled := make(map[string]struct{})
	var mu sync.Mutex
	shouldHandle := func(host string) bool {
		mu.Lock()
		defer mu.Unlock()
		if _, ok := handled[host]; ok {
			return false
		}
		handled[host] = struct{}{}
		return true
	}

	handle := func(host string, ips []string, cname string) {
		if !shouldHandle(host) {
			return
		}
		provider := scanner.DetectCloudProvider(ips, cname, "")
		if provider != "" {
			_ = mctx.Store.Upsert(mctx.Ctx, host, func(h *store.Host) {
				if h.CloudProvider == "" {
					h.CloudProvider = provider
				}
			})
		}

		if buckets := scanner.CheckS3BucketsWithClient(host, client); len(buckets) > 0 {
			for _, url := range buckets {
				mctx.Bus.Publish(mctx.Ctx, eventbus.CloudAssetFound{
					EventMeta: eventbus.EventMeta{At: time.Now(), Source: ModuleName, Target: host},
					Provider:  "AWS",
					Kind:      "s3-bucket",
					Name:      host,
					URL:       url,
					Status:    "accessible",
				})
			}
		}
	}

	var wg sync.WaitGroup

	// Drain: every host already in the store with an IP.
	for _, h := range mctx.Store.All(mctx.Ctx) {
		if h == nil || h.Subdomain == "" || len(h.IPs) == 0 {
			continue
		}
		h := h
		wg.Add(1)
		go func() { defer wg.Done(); handle(h.Subdomain, h.IPs, h.CNAME) }()
	}

	// Late DNSResolved events.
	sub := mctx.Bus.Subscribe(eventbus.EventDNSResolved, func(_ context.Context, e eventbus.Event) {
		ev, ok := e.(eventbus.DNSResolved)
		if !ok {
			return
		}
		wg.Add(1)
		go func() { defer wg.Done(); handle(ev.Subdomain, ev.IPs, ev.CNAME) }()
	})
	defer sub.Unsubscribe()

	select {
	case <-time.After(500 * time.Millisecond):
	case <-mctx.Ctx.Done():
	}

	wg.Wait()
	return nil
}