Day 83 — Threat Intelligence Aggregation

Learning Objectives

• Build a shared threat score per IP that aggregates signals from all security layers
• Implement score decay over time (old signals become less relevant)
• Connect WAF, rate limiter, and auth to the shared ThreatIntel store

Exercise: ThreatIntel Store

package threatintel

import (
    "fmt"
    "sync"
    "time"
)

type SignalType string

const (
    SignalWAFHit        SignalType = "waf_hit"
    SignalRateLimit     SignalType = "rate_limit"
    SignalAuthFailure   SignalType = "auth_failure"
    SignalScanner       SignalType = "scanner"
    SignalDDoS          SignalType = "ddos"
)

type Signal struct {
    Type   SignalType
    Weight float64
    At     time.Time
}

type IPRecord struct {
    mu      sync.Mutex
    signals []Signal
}

type ThreatIntel struct {
    mu      sync.RWMutex
    records map[string]*IPRecord
    halfLife time.Duration // score halves every halfLife
}

func New(halfLife time.Duration) *ThreatIntel {
    t := &ThreatIntel{
        records:  make(map[string]*IPRecord),
        halfLife: halfLife,
    }
    go t.cleanup()
    return t
}

// ReportSignal adds a threat signal for an IP
func (ti *ThreatIntel) ReportSignal(ip string, t SignalType, weight float64) {
    ti.mu.Lock()
    rec, ok := ti.records[ip]
    if !ok {
        rec = &IPRecord{}
        ti.records[ip] = rec
    }
    ti.mu.Unlock()

    rec.mu.Lock()
    rec.signals = append(rec.signals, Signal{Type: t, Weight: weight, At: time.Now()})
    rec.mu.Unlock()

    fmt.Printf("[THREAT] %s: signal=%s weight=%.1f\n", ip, t, weight)
}

// Score returns the current threat score for an IP (with time decay)
func (ti *ThreatIntel) Score(ip string) float64 {
    ti.mu.RLock()
    rec, ok := ti.records[ip]
    ti.mu.RUnlock()
    if !ok { return 0 }

    rec.mu.Lock()
    defer rec.mu.Unlock()

    now := time.Now()
    var total float64
    for _, s := range rec.signals {
        age := now.Sub(s.At)
        // Exponential decay: weight * 2^(-age/halfLife)
        decayFactor := math.Pow(2, -float64(age)/float64(ti.halfLife))
        total += s.Weight * decayFactor
    }
    return total
}

// Action returns the recommended action based on current score
func (ti *ThreatIntel) Action(ip string) string {
    score := ti.Score(ip)
    switch {
    case score >= 80: return "block"
    case score >= 50: return "challenge"
    case score >= 25: return "monitor"
    default:          return "allow"
    }
}

func (ti *ThreatIntel) cleanup() {
    ticker := time.NewTicker(time.Minute)
    defer ticker.Stop()
    for range ticker.C {
        cutoff := time.Now().Add(-24 * time.Hour) // clean signals older than 24h
        ti.mu.Lock()
        for ip, rec := range ti.records {
            rec.mu.Lock()
            var fresh []Signal
            for _, s := range rec.signals {
                if s.At.After(cutoff) { fresh = append(fresh, s) }
            }
            rec.signals = fresh
            if len(fresh) == 0 { delete(ti.records, ip) }
            rec.mu.Unlock()
        }
        ti.mu.Unlock()
    }
}

// Integration: ThreatIntel middleware
func (ti *ThreatIntel) Middleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ip := realClientIP(r)
        action := ti.Action(ip)
        switch action {
        case "block":
            http.Error(w, `{"error":"forbidden"}`, http.StatusForbidden)
            return
        case "challenge":
            // In production: return a PoW challenge (Day 97)
            // For now: add delay
            time.Sleep(200 * time.Millisecond)
        }
        next.ServeHTTP(w, r)
    })
}

Real-World Context

A single WAF hit might be a false positive. But 5 WAF hits + 10 rate limit violations + 3 auth failures in 10 minutes = almost certainly an attacker. The aggregated score with time decay is how Cloudflare's threat scoring works. The 10-minute half-life means an IP that cleans up its act can recover — preventing permanent false positive lockouts.