go-otel/internal/blocker/blocker.go

// Package blocker turns the honeypot's promoted-IP feed into ingress-nginx
// `block-cidrs` entries.
//
// Safety properties, all deliberate:
//   - dry-run is the default; nothing is written unless DryRun is false;
//   - it only manages entries it added itself (tracked in an annotation on the
//     ConfigMap), so manually configured block-cidrs are preserved untouched;
//   - loopback/private/link-local addresses and an operator allowlist are never
//     blocked;
//   - blocks expire, the set is capped, growth per tick is limited, and writes
//     are rate-limited because every ConfigMap change reloads nginx.
package blocker

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"net/netip"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"k8s.io/client-go/rest"
	"k8s.io/client-go/tools/clientcmd"

	"git.tornberg.me/go-otel/internal/honeypot"
)

const AnnotationManaged = "otel-debug.k6n.net/blocked"

type Source interface {
	PromotedIPs() []honeypot.IP
}

type Config struct {
	Enabled       bool
	DryRun        bool
	Namespace     string
	ConfigMap     string
	Key           string
	Interval      time.Duration // how often the feed is evaluated
	MinWrite      time.Duration // minimum gap between ConfigMap writes
	BlockTTL      time.Duration // how long a block lasts after last sighting
	MinProbes     int           // per-window probe hits required to block
	MinPaths      int           // distinct probe paths required to block
	BroadProbes   int           // hits+paths that block regardless of MinProbes
	BroadPaths    int
	MaxEntries    int // cap on entries this blocker manages
	MaxNewPerTick int // limit on newly added entries per evaluation
	Allow         []netip.Prefix
}

func DefaultConfig() Config {
	return Config{
		DryRun: true, Namespace: "ingress-nginx", ConfigMap: "ingress-nginx-controller", Key: "block-cidrs",
		Interval: time.Minute, MinWrite: 5 * time.Minute, BlockTTL: 24 * time.Hour,
		MinProbes: 10, MinPaths: 3, BroadProbes: 100, BroadPaths: 50, MaxEntries: 500, MaxNewPerTick: 20,
	}
}

// ConfigFromEnv reads OTEL_BLOCKER_*. Enabled defaults to false and DryRun to
// true; both must be changed on purpose.
func ConfigFromEnv() Config {
	c := DefaultConfig()
	c.Enabled = strings.EqualFold(os.Getenv("OTEL_BLOCKER_ENABLED"), "true")
	if v := os.Getenv("OTEL_BLOCKER_DRY_RUN"); v != "" {
		c.DryRun = !strings.EqualFold(v, "false")
	}
	str := func(env string, dst *string) {
		if v := os.Getenv(env); v != "" {
			*dst = v
		}
	}
	num := func(env string, dst *int) {
		if n, err := strconv.Atoi(os.Getenv(env)); err == nil && n > 0 {
			*dst = n
		}
	}
	dur := func(env string, dst *time.Duration) {
		if d, err := time.ParseDuration(os.Getenv(env)); err == nil && d > 0 {
			*dst = d
		}
	}
	str("OTEL_BLOCKER_NAMESPACE", &c.Namespace)
	str("OTEL_BLOCKER_CONFIGMAP", &c.ConfigMap)
	str("OTEL_BLOCKER_KEY", &c.Key)
	num("OTEL_BLOCKER_MIN_PROBES", &c.MinProbes)
	num("OTEL_BLOCKER_MIN_PATHS", &c.MinPaths)
	num("OTEL_BLOCKER_MAX_ENTRIES", &c.MaxEntries)
	dur("OTEL_BLOCKER_INTERVAL", &c.Interval)
	dur("OTEL_BLOCKER_MIN_WRITE", &c.MinWrite)
	dur("OTEL_BLOCKER_TTL", &c.BlockTTL)
	for _, item := range strings.Split(os.Getenv("OTEL_BLOCKER_ALLOW"), ",") {
		item = strings.TrimSpace(item)
		if item == "" {
			continue
		}
		if p, err := netip.ParsePrefix(item); err == nil {
			c.Allow = append(c.Allow, p)
		} else if a, err := netip.ParseAddr(item); err == nil {
			c.Allow = append(c.Allow, netip.PrefixFrom(a, a.BitLen()))
		} else {
			log.Printf("blocker: ignoring invalid OTEL_BLOCKER_ALLOW entry %q", item)
		}
	}
	return c
}

// Start builds a client (in-cluster, or KUBECONFIG locally) and runs the
// blocker in the background. It returns nil, nil when disabled.
func Start(ctx context.Context, cfg Config, source Source) (*Blocker, error) {
	if !cfg.Enabled {
		return nil, nil
	}
	rc, err := rest.InClusterConfig()
	if err != nil {
		kubeconfig := os.Getenv("KUBECONFIG")
		if kubeconfig == "" {
			home, herr := os.UserHomeDir()
			if herr != nil {
				return nil, herr
			}
			kubeconfig = home + "/.kube/config"
		}
		if rc, err = clientcmd.BuildConfigFromFlags("", kubeconfig); err != nil {
			return nil, fmt.Errorf("blocker kubernetes config: %w", err)
		}
	}
	client, err := kubernetes.NewForConfig(rc)
	if err != nil {
		return nil, fmt.Errorf("blocker kubernetes client: %w", err)
	}
	b := New(cfg, client, source)
	go b.Run(ctx)
	return b, nil
}

type Blocker struct {
	cfg    Config
	client kubernetes.Interface
	source Source
	now    func() time.Time

	mu        sync.Mutex
	lastWrite time.Time
	lastErr   string
	last      Status
}

// Status is a snapshot for the UI and for tests.
type Status struct {
	Enabled    bool             `json:"enabled"`
	DryRun     bool             `json:"dryRun"`
	Evaluated  time.Time        `json:"evaluated"`
	Candidates []string         `json:"candidates"`
	Managed    map[string]int64 `json:"managed"` // ip -> unix expiry
	Wrote      bool             `json:"wrote"`
	LastWrite  time.Time        `json:"lastWrite"`
	Error      string           `json:"error,omitempty"`
}

// Add manually adds one public IP to the blocker-managed set. It uses the
// same expiry, allowlist, ConfigMap ownership, and dry-run rules as automatic
// blocks.
func (b *Blocker) Add(ctx context.Context, address string) (Status, error) {
	addr, err := netip.ParseAddr(strings.TrimSpace(address))
	if err != nil || honeypot.Untrackable(address) {
		return Status{}, fmt.Errorf("only public unicast IPs can be blocked")
	}
	for _, p := range b.cfg.Allow {
		if p.Contains(addr) {
			return Status{}, fmt.Errorf("IP is on the operator allowlist")
		}
	}

	b.mu.Lock()
	defer b.mu.Unlock()
	now := b.now()
	cms := b.client.CoreV1().ConfigMaps(b.cfg.Namespace)
	cm, err := cms.Get(ctx, b.cfg.ConfigMap, metav1.GetOptions{})
	if err != nil {
		return Status{}, fmt.Errorf("get configmap: %w", err)
	}
	previous := map[string]int64{}
	if raw := cm.Annotations[AnnotationManaged]; raw != "" {
		_ = json.Unmarshal([]byte(raw), &previous)
	}
	managed := map[string]int64{}
	for ip, exp := range previous {
		if exp > now.Unix() {
			managed[ip] = exp
		}
	}
	if _, exists := managed[addr.String()]; !exists && len(managed) >= b.cfg.MaxEntries {
		return Status{}, fmt.Errorf("managed block list is full")
	}
	managed[addr.String()] = now.Add(b.cfg.BlockTTL).Unix()

	status := Status{Enabled: true, DryRun: b.cfg.DryRun, Evaluated: now, Managed: managed, LastWrite: b.lastWrite}
	if b.cfg.DryRun {
		b.last = status
		return status, nil
	}
	if !b.lastWrite.IsZero() && now.Sub(b.lastWrite) < b.cfg.MinWrite {
		return status, fmt.Errorf("block list was recently updated; try again later")
	}

	current := splitCIDRs(cm.Data[b.cfg.Key])
	manual := make([]string, 0, len(current))
	for _, entry := range current {
		if _, wasOurs := previous[entry]; !wasOurs {
			manual = append(manual, entry)
		}
	}
	desired := append(manual, managedList(managed)...)
	if cm.Annotations == nil {
		cm.Annotations = map[string]string{}
	}
	if cm.Data == nil {
		cm.Data = map[string]string{}
	}
	state, _ := json.Marshal(managed)
	cm.Annotations[AnnotationManaged] = string(state)
	cm.Data[b.cfg.Key] = strings.Join(dedupSorted(desired), ",")
	if _, err := cms.Update(ctx, cm, metav1.UpdateOptions{}); err != nil {
		return Status{}, fmt.Errorf("update configmap: %w", err)
	}
	b.lastWrite, b.lastErr = now, ""
	status.Wrote, status.LastWrite = true, now
	b.last = status
	log.Printf("blocker: manually added %s (expires %s)", addr, time.Unix(managed[addr.String()], 0).UTC().Format(time.RFC3339))
	return status, nil
}

func New(cfg Config, client kubernetes.Interface, source Source) *Blocker {
	d := DefaultConfig()
	if cfg.Interval <= 0 {
		cfg.Interval = d.Interval
	}
	if cfg.MinWrite <= 0 {
		cfg.MinWrite = d.MinWrite
	}
	if cfg.BlockTTL <= 0 {
		cfg.BlockTTL = d.BlockTTL
	}
	if cfg.MaxEntries <= 0 {
		cfg.MaxEntries = d.MaxEntries
	}
	if cfg.MaxNewPerTick <= 0 {
		cfg.MaxNewPerTick = d.MaxNewPerTick
	}
	if cfg.MinProbes <= 0 {
		cfg.MinProbes = d.MinProbes
	}
	if cfg.MinPaths <= 0 {
		cfg.MinPaths = d.MinPaths
	}
	if cfg.BroadProbes <= 0 {
		cfg.BroadProbes = d.BroadProbes
	}
	if cfg.BroadPaths <= 0 {
		cfg.BroadPaths = d.BroadPaths
	}
	return &Blocker{cfg: cfg, client: client, source: source, now: time.Now}
}

// Run evaluates the feed until ctx is cancelled.
func (b *Blocker) Run(ctx context.Context) {
	log.Printf("blocker: started (dryRun=%t, %s/%s key=%s)", b.cfg.DryRun, b.cfg.Namespace, b.cfg.ConfigMap, b.cfg.Key)
	t := time.NewTicker(b.cfg.Interval)
	defer t.Stop()
	for {
		if err := b.Tick(ctx); err != nil {
			log.Printf("blocker: %v", err)
		}
		select {
		case <-ctx.Done():
			return
		case <-t.C:
		}
	}
}

// Blockable applies the policy to one promoted IP.
func (b *Blocker) Blockable(ip honeypot.IP) bool {
	if honeypot.Untrackable(ip.Address) {
		return false
	}
	addr, err := netip.ParseAddr(ip.Address)
	if err != nil {
		return false
	}
	for _, p := range b.cfg.Allow {
		if p.Contains(addr) {
			return false
		}
	}
	return (ip.Count >= b.cfg.MinProbes && ip.Paths >= b.cfg.MinPaths) ||
		(ip.Count >= b.cfg.BroadProbes && ip.Paths >= b.cfg.BroadPaths)
}

// Tick runs one evaluation. It is exported so tests and a manual trigger can
// drive it without waiting on the timer.
func (b *Blocker) Tick(ctx context.Context) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	now := b.now()
	status := Status{Enabled: true, DryRun: b.cfg.DryRun, Evaluated: now, Managed: map[string]int64{}, LastWrite: b.lastWrite}
	defer func() { b.last = status }()

	cms := b.client.CoreV1().ConfigMaps(b.cfg.Namespace)
	cm, err := cms.Get(ctx, b.cfg.ConfigMap, metav1.GetOptions{})
	if err != nil {
		b.lastErr = err.Error()
		status.Error = b.lastErr
		return fmt.Errorf("get configmap: %w", err)
	}

	previous := map[string]int64{}
	if raw := cm.Annotations[AnnotationManaged]; raw != "" {
		if err := json.Unmarshal([]byte(raw), &previous); err != nil {
			previous = map[string]int64{} // unreadable state: treat as empty rather than guess
		}
	}
	managed := map[string]int64{}
	for ip, exp := range previous {
		if exp > now.Unix() {
			managed[ip] = exp
		}
	}

	added := 0
	for _, cand := range b.source.PromotedIPs() {
		if !b.Blockable(cand) {
			continue
		}
		status.Candidates = append(status.Candidates, cand.Address)
		_, exists := managed[cand.Address]
		if !exists {
			if added >= b.cfg.MaxNewPerTick {
				continue
			}
			added++
		}
		managed[cand.Address] = now.Add(b.cfg.BlockTTL).Unix()
	}
	sort.Strings(status.Candidates)
	for len(managed) > b.cfg.MaxEntries { // drop the soonest-expiring first
		var oldest string
		for ip, exp := range managed {
			if oldest == "" || exp < managed[oldest] || (exp == managed[oldest] && ip < oldest) {
				oldest = ip
			}
		}
		delete(managed, oldest)
	}
	status.Managed = managed

	current := splitCIDRs(cm.Data[b.cfg.Key])
	manual := make([]string, 0, len(current))
	for _, entry := range current {
		if _, wasOurs := previous[entry]; !wasOurs {
			manual = append(manual, entry)
		}
	}
	desired := append([]string(nil), manual...)
	for ip := range managed {
		desired = append(desired, ip)
	}
	desired = dedupSorted(desired)
	changed := strings.Join(desired, ",") != strings.Join(dedupSorted(current), ",")

	if !changed {
		return nil
	}
	if b.cfg.DryRun {
		log.Printf("blocker[dry-run]: would set %s to %d entries (%d managed): %v", b.cfg.Key, len(desired), len(managed), managedList(managed))
		return nil
	}
	if !b.lastWrite.IsZero() && now.Sub(b.lastWrite) < b.cfg.MinWrite {
		return nil // rate-limited; the change is picked up on a later tick
	}
	state, _ := json.Marshal(managed)
	if cm.Annotations == nil {
		cm.Annotations = map[string]string{}
	}
	if cm.Data == nil {
		cm.Data = map[string]string{}
	}
	cm.Annotations[AnnotationManaged] = string(state)
	cm.Data[b.cfg.Key] = strings.Join(desired, ",")
	if len(desired) == 0 {
		delete(cm.Data, b.cfg.Key)
	}
	if _, err := cms.Update(ctx, cm, metav1.UpdateOptions{}); err != nil {
		b.lastErr = err.Error()
		status.Error = b.lastErr
		return fmt.Errorf("update configmap: %w", err)
	}
	b.lastWrite, b.lastErr = now, ""
	status.Wrote, status.LastWrite = true, now
	log.Printf("blocker: updated %s: %d entries (%d managed)", b.cfg.Key, len(desired), len(managed))
	return nil
}

// HandleStatus serves the last evaluation. Register it behind authentication.
func (b *Blocker) HandleStatus(w http.ResponseWriter, r *http.Request) {
	b.mu.Lock()
	s := b.last
	b.mu.Unlock()
	if s.Evaluated.IsZero() {
		s = Status{Enabled: true, DryRun: b.cfg.DryRun, Managed: map[string]int64{}}
	}
	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Cache-Control", "no-store")
	_ = json.NewEncoder(w).Encode(s)
}

// HandleAdd serves POST /api/blocker/ips/{ip}. Register it behind
// authentication; it never permits private or allowlisted addresses.
func (b *Blocker) HandleAdd(w http.ResponseWriter, r *http.Request) {
	status, err := b.Add(r.Context(), r.PathValue("ip"))
	if err != nil {
		code := http.StatusBadRequest
		if strings.Contains(err.Error(), "recently updated") {
			code = http.StatusTooManyRequests
		}
		http.Error(w, err.Error(), code)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	if status.DryRun {
		w.WriteHeader(http.StatusAccepted)
	}
	_ = json.NewEncoder(w).Encode(status)
}

func splitCIDRs(v string) []string {
	var out []string
	for _, p := range strings.Split(v, ",") {
		if p = strings.TrimSpace(p); p != "" {
			out = append(out, p)
		}
	}
	return out
}

func dedupSorted(in []string) []string {
	seen := map[string]struct{}{}
	out := make([]string, 0, len(in))
	for _, v := range in {
		if _, ok := seen[v]; !ok {
			seen[v] = struct{}{}
			out = append(out, v)
		}
	}
	sort.Strings(out)
	return out
}

func managedList(m map[string]int64) []string {
	out := make([]string, 0, len(m))
	for ip := range m {
		out = append(out, ip)
	}
	sort.Strings(out)
	return out
}