Deep Dive Firewall in 2026: Step-by-Step

#!/usr/bin/env bash
# scaffold.sh – Create the project directory structure
set -euo pipefail

PROJECT_ROOT="deep-dive-firewall"
mkdir -p "$PROJECT_ROOT"/{cmd/packetfilter,internal/policy,internal/monitor,charts,deploy}

# Go module initialization
cd "$PROJECT_ROOT"
go mod init github.com/yourorg/deep-dive-firewall
go mod tidy

# Install cilium/ebpf – the Go library for loading and attaching eBPF programs
go get github.com/cilium/ebpf@latest
go get github.com/cilium/ebpf/rtti@latest

# Install Python dependencies
pip install pyyaml prometheus-client asyncio uvloop 2>&1 | tail -5

# Verify kernel headers exist (required for BTF)
if [ ! -d /sys/kernel/btf/vmlinux ]; then
  echo "ERROR: BTF not available at /sys/kernel/btf/vmlinux"
  echo "Run: sudo apt install linux-headers-$(uname -r)"
  exit 1
fi

echo "Project scaffolded successfully at $PROJECT_ROOT"

#!/usr/bin/env python3
"""
policy_engine.py – Firewall policy compiler and validator.
Reads YAML firewall rules, validates them, and emits a JSON
intermediate representation consumed by the Go packet filter.

Author: Deep Dive Firewall Project
License: Apache 2.0
"""

import json
import logging
import sys
from dataclasses import dataclass, field, asdict
from enum import Enum
from ipaddress import ip_network, IPv4Network, IPv6Network
from pathlib import Path
from typing import Optional

import yaml  # PyYAML – validated at parse time

# ---------------------------------------------------------------------------
# Domain models
# ---------------------------------------------------------------------------

class Action(str, Enum):
    ALLOW = "allow"
    DENY = "deny"
    RATE_LIMIT = "rate_limit"


class Protocol(str, Enum):
    TCP = "tcp"
    UDP = "udp"
    ICMP = "icmp"
    ANY = "any"


@dataclass
class FirewallRule:
    """Single firewall rule with full metadata."""
    name: str
    priority: int                          # lower = evaluated first
    source_cidrs: list[str]                # e.g. ["10.0.0.0/8"]
    destination_ports: list[int]           # e.g. [443, 8443]
    protocol: Protocol = Protocol.ANY
    action: Action = Action.DENY
    rate_limit_pps: Optional[int] = None   # packets-per-second for rate-limit rules
    log: bool = True                        # emit a log event on match
    description: str = ""

    # ------------------------------------------------------------------
    # Validation
    # ------------------------------------------------------------------
    def validate(self) -> list[str]:
        """Return a list of validation error strings (empty = valid)."""
        errors: list[str] = []
        if not self.name.isidentifier():
            errors.append(f"Rule name '{self.name}' is not a valid identifier")
        if self.priority < 0 or self.priority > 65535:
            errors.append(f"Priority {self.priority} out of range [0, 65535]")
        # Validate every CIDR
        for cidr in self.source_cidrs:
            try:
                net: IPv4Network | IPv6Network = ip_network(cidr, strict=False)
                if net.num_addresses == 0:
                    errors.append(f"Empty CIDR: {cidr}")
            except ValueError as exc:
                errors.append(f"Invalid CIDR '{cidr}': {exc}")
        # Validate ports
        for port in self.destination_ports:
            if not (1 <= port <= 65535):
                errors.append(f"Port {port} out of range [1, 65535]")
        # Rate-limit sanity check
        if self.action == Action.RATE_LIMIT and self.rate_limit_pps is None:
            errors.append(f"Rule '{self.name}' is rate_limit but has no rate_limit_pps")
        if self.rate_limit_pps is not None and self.rate_limit_pps <= 0:
            errors.append(f"rate_limit_pps must be > 0, got {self.rate_limit_pps}")
        return errors


@dataclass
class PolicySet:
    """Complete compiled policy set ready for the agent."""
    rules: list[FirewallRule] = field(default_factory=list)
    default_action: Action = Action.DENY
    generated_at: str = ""  # ISO-8601 timestamp

    def validate(self) -> list[str]:
        errors: list[str] = []
        seen_names: set[str] = set()
        for rule in self.rules:
            errors.extend(rule.validate())
            if rule.name in seen_names:
                errors.append(f"Duplicate rule name: {rule.name}")
            seen_names.add(rule.name)
        # Ensure priority uniqueness (optional – warn, don't error)
        priorities = [r.priority for r in self.rules]
        if len(priorities) != len(set(priorities)):
            logging.warning("Duplicate priorities detected; evaluation order is deterministic but ambiguous")
        return errors


# ---------------------------------------------------------------------------
# Parser
# ---------------------------------------------------------------------------

def load_policy(path: str | Path) -> PolicySet:
    """Load YAML, validate, and return a compiled PolicySet."""
    path = Path(path)
    if not path.is_file():
        raise FileNotFoundError(f"Policy file not found: {path}")

    with open(path, "r", encoding="utf-8") as fh:
        raw = yaml.safe_load(fh)

    if not isinstance(raw, dict):
        raise ValueError("Top-level YAML must be a mapping")

    policy = PolicySet(
        default_action=Action(raw.get("default_action", "deny")),
        generated_at=raw.get("generated_at", ""),
    )

    for entry in raw.get("rules", []):
        rule = FirewallRule(
            name=entry["name"],
            priority=int(entry["priority"]),
            source_cidrs=entry.get("source_cidrs", ["0.0.0.0/0"]),
            destination_ports=[int(p) for p in entry.get("destination_ports", [])],
            protocol=Protocol(entry.get("protocol", "any")),
            action=Action(entry.get("action", "deny")),
            rate_limit_pps=entry.get("rate_limit_pps"),
            log=entry.get("log", True),
            description=entry.get("description", ""),
        )
        policy.rules.append(rule)

    # Sort by priority (ascending) so the agent can short-circuit
    policy.rules.sort(key=lambda r: r.priority)

    # Validate
    errors = policy.validate()
    if errors:
        error_block = "\n".join(f"  - {e}" for e in errors)
        raise ValueError(f"Policy validation failed:\n{error_block}")

    logging.info("Loaded %d rules from %s", len(policy.rules), path)
    return policy


def compile_to_json(policy: PolicySet, output_path: str | Path) -> None:
    """Serialize the policy set to JSON for the Go agent."""
    output_path = Path(output_path)
    payload = {
        "default_action": policy.default_action.value,
        "generated_at": policy.generated_at,
        "rules": [
            {
                "name": r.name,
                "priority": r.priority,
                "source_cidrs": r.source_cidrs,
                "destination_ports": r.destination_ports,
                "protocol": r.protocol.value,
                "action": r.action.value,
                "rate_limit_pps": r.rate_limit_pps,
                "log": r.log,
                "description": r.description,
            }
            for r in policy.rules
        ],
    }
    output_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
    logging.info("Compiled policy written to %s (%d bytes)", output_path, output_path.stat().st_size)


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]}  [output.json]", file=sys.stderr)
        sys.exit(1)

    policy_path = sys.argv[1]
    output_path = sys.argv[2] if len(sys.argv) > 2 else "policy.json"

    try:
        policy = load_policy(policy_path)
        compile_to_json(policy, output_path)
        print(f"✓ Compiled {len(policy.rules)} rules → {output_path}")
    except (ValueError, FileNotFoundError, yaml.YAMLError) as exc:
        print(f"✗ {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()

// cmd/packetfilter/main.go – eBPF packet filter agent.
// Loads CIDR-based firewall rules into an eBPF LRU hash map
// and attaches a cgroup/connect4 program to enforce them.
package main

import (
    "context"
    "encoding/binary"
    "fmt"
    "log"
    "net"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/cilium/ebpf"
    "github.com/cilium/ebpf/link"
    "github.com/cilium/ebpf/rtti"
)

// cidrKey represents the 4-byte IPv4 prefix used as the BPF map key.
type cidrKey struct {
    Prefix uint32 `json:"prefix"`
    PrefixLen uint8 `json:"prefix_len"`
    Padding  [3]byte
}

// verdictValue stores the action and metadata for a CIDR rule.
type verdictValue struct {
    Action   uint32 // 0=deny, 1=allow, 2=rate_limit
    PPSLimit uint32 // packets-per-second limit (0 = unlimited)
    Log      uint32 // 1=log, 0=silent
}

// loadCIDRMap populates the BPF hash map from a JSON policy file.
func loadCIDRMap(mapName string, policyPath string) (*ebpf.Map, error) {
    // Open the compiled eBPF object file
    objFile, err := os.Open("firewall.o")
    if err != nil {
        return nil, fmt.Errorf("opening BPF object: %w", err)
    }
    defer objFile.Close()

    obj := struct {
        CIDRMap *ebpf.Map `json:"cidr_rules"`
    }{}

    if err := rtti.LoadObjects(objFile, &obj); err != nil {
        return nil, fmt.Errorf("loading BPF objects: %w", err)
    }

    // Read the JSON policy compiled by the Python engine
    policyBytes, err := os.ReadFile(policyPath)
    if err != nil {
        return nil, fmt.Errorf("reading policy JSON: %w", err)
    }

    type rule struct {
        SourceCIDRs []string `json:"source_cidrs"`
        Action      string   `json:"action"`
        RateLimitPPS int    `json:"rate_limit_pps"`
        Log         bool    `json:"log"`
    }

    var policy struct {
        Rules []rule `json:"rules"`
    }
    if err := json.Unmarshal(policyBytes, &policy); err != nil {
        return nil, fmt.Errorf("parsing policy JSON: %w", err)
    }

    for _, r := range policy.Rules {
        for _, cidrStr := range r.SourceCIDRs {
            _, ipNet, err := net.ParseCIDR(cidrStr)
            if err != nil {
                log.Printf("WARN: skipping invalid CIDR %q: %v", cidrStr, err)
                continue
            }

            // Convert network address to host byte order uint32
            ip := binary.BigEndian.Uint32(ipNet.IP.To4())
            ones, _ := ipNet.Mask.Size()

            // Map action string to uint32 enum
            var actionVal uint32
            switch r.Action {
            case "allow":
                actionVal = 1
            case "rate_limit":
                actionVal = 2
            default:
                actionVal = 0 // deny
            }

            key := cidrKey{
                Prefix:   ip,
                PrefixLen: uint8(ones),
            }
            value := verdictValue{
                Action:   actionVal,
                PPSLimit: uint32(r.RateLimitPPS),
                Log:      boolToUint32(r.Log),
            }

            if err := obj.CIDRMap.Put(key, value); err != nil {
                return nil, fmt.Errorf("inserting CIDR %s: %w", cidrStr, err)
            }
            log.Printf("Rule: %s → action=%d pps=%d", cidrStr, actionVal, r.RateLimitPPS)
        }
    }

    return obj.CIDRMap, nil
}

func boolToUint32(b bool) uint32 {
    if b {
        return 1
    }
    return 0
}

// attachCgroupProgram loads and attaches the BPF cgroup/connect4 program.
func attachCgroupProgram(progPath string) (link.Link, error) {
    objFile, err := os.Open(progPath)
    if err != nil {
        return nil, fmt.Errorf("opening BPF object: %w", err)
    }
    defer objFile.Close()

    obj := struct {
        Connect4Prog *ebpf.Program `json:"connect4_filter"`
    }{}
    if err := rtti.LoadObjects(objFile, &obj); err != nil {
        return nil, fmt.Errorf("loading BPF programs: %w", err)
    }

    // Attach to the cgroup2 filesystem at /sys/fs/cgroup2
    l, err := link.AttachCgroup(link.CgroupOptions{
        Path:  "/sys/fs/cgroup2",
        Attach: ebpf.AttachCgroupConnect4,
        Program: obj.Connect4Prog,
    })
    if err != nil {
        return nil, fmt.Errorf("attaching cgroup/connect4: %w", err)
    }

    log.Println("cgroup/connect4 program attached successfully")
    return l, nil
}

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer stop()

    policyPath := "/etc/firewall/policy.json"
    if len(os.Args) > 1 {
        policyPath = os.Args[1]
    }

    // Step 1: Load CIDR rules into the BPF hash map
    cidrMap, err := loadCIDRMap("cidr_rules", policyPath)
    if err != nil {
        log.Fatalf("FATAL: failed to load CIDR map: %v", err)
    }
    defer cidrMap.Close()

    // Step 2: Attach the cgroup/connect4 BPF program
    cgLink, err := attachCgroupProgram("firewall.o")
    if err != nil {
        log.Fatalf("FATAL: failed to attach cgroup program: %v", err)
    }
    defer cgLink.Close()

    // Step 3: Periodically log map statistics
    ticker := time.NewTicker(30 * time.Second)
    defer ticker.Stop()

    log.Println("Firewall agent running. Press Ctrl+C to stop.")
    for {
        select {
        case <-ctx.Done():
            log.Println("Shutting down firewall agent...")
            return
        case <-ticker.C:
            stats, err := cidrMap.Stats()
            if err != nil {
                log.Printf("WARN: failed to get map stats: %v", err)
                continue
            }
            log.Printf("Map stats: entries=%d max_entries=%d", stats.Entries, stats.MaxEntries)
        }
    }
}

#!/usr/bin/env python3
"""
monitor.py – Real-time eBPF firewall event monitor.
Reads verdicts from a BPF_PERF_EVENT_ARRAY ring buffer,
enriches with container metadata, and exports Prometheus metrics.

Requires: pyyaml, prometheus-client, psutil
"""

import asyncio
import ctypes
import json
import logging
import os
import struct
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, Optional

import yaml
from prometheus_client import Counter, Gauge, start_http_server

import psutil  # For process-to-container resolution

# ---------------------------------------------------------------------------
# BPF ring buffer structures (must match the BPF program's output layout)
# ---------------------------------------------------------------------------

class VerdictEvent(ctypes.LittleEndianStructure):
    """Wire-format struct emitted by the eBPF program for each packet."""
    _pack_ = 1
    _fields_ = [
        ("timestamp_ns", ctypes.c_uint64),
        ("src_ip", ctypes.c_uint32),
        ("dst_ip", ctypes.c_uint32),
        ("dst_port", ctypes.c_uint16),
        ("ip_proto", ctypes.c_uint8),       # IPPROTO_TCP=6, IPPROTO_UDP=17
        ("verdict", ctypes.c_uint32),       # 0=deny, 1=allow, 2=rate-limited
        ("pid", ctypes.c_uint32),
        ("padding", ctypes.c_uint32),
    ]

    SRC_IP_FMT = "!I"  # Network byte order uint32

    def src_ip_str(self) -> str:
        packed = struct.pack(self.SRC_IP_FMT, self.src_ip)
        return ".".join(str(b) for b in packed)

    def dst_ip_str(self) -> str:
        packed = struct.pack(self.SRC_IP_FMT, self.dst_ip)
        return ".".join(str(b) for b in packed)

    def protocol_name(self) -> str:
        return {6: "TCP", 17: "UDP"}.get(self.ip_proto, f"PROTO_{self.ip_proto}")

    def verdict_name(self) -> str:
        return {0: "DENY", 1: "ALLOW", 2: "RATE_LIMIT"}.get(self.verdict, "UNKNOWN")


# ---------------------------------------------------------------------------
# Metrics registry
# ---------------------------------------------------------------------------

# Counter of packets by (verdict, protocol, destination_port)
PACKET_COUNT = Counter(
    "firewall_packet_total",
    "Total packets processed by the firewall",
    ["verdict", "protocol", "dst_port"],
)

# Gauge of currently active rate-limited IPs
RATE_LIMITED_IPS = Gauge(
    "firewall_rate_limited_ips",
    "Number of IPs currently under rate limiting",
)

# Histogram of packet processing latency (nanoseconds in-kernel)
PROCESSING_LATENCY = Histogram = None  # Placeholder – defined below

# Per-CIDR deny counter for anomaly detection
CIDR_DENY_COUNT: Dict[str, int] = {}
CIDR_DENY_THRESHOLD = 1000  # Alert if a single /24 exceeds this in 60s

# ---------------------------------------------------------------------------
# Container metadata resolver
# ---------------------------------------------------------------------------

@dataclass
class ContainerMeta:
    pid: int
    container_id: str
    image: str
    namespace: str
    pod: str


def resolve_container(pid: int) -> Optional[ContainerMeta]:
    """Resolve a PID to its container using /proc//cgroup.
    Returns None if the process is not in a container."""
    cgroup_path = f"/proc/{pid}/cgroup"

    if not os.path.isfile(cgroup_path):
        return None

    try:
        with open(cgroup_path, "r") as f:
            for line in f:
                # cgroup v2 format: "0::/kubepods.slice/kubepods-burstable.slice/..."
                if "kubepods" in line or "docker" in line or "containerd" in line:
                    parts = line.strip().split(":")
                    hierarchy = parts[0]
                    path = parts[2] if len(parts) > 2 else ""

                    # Extract container ID from the cgroup path
                    segments = path.split("/")
                    container_id = ""
                    for seg in reversed(segments):
                        if len(seg) >= 12 and all(c in "0123456789abcdef" for c in seg[:12]):
                            container_id = seg[:12]
                            break

                    if container_id:
                        # Attempt to enrich with Docker/containerd labels
                        image = "unknown"
                        ns = "default"
                        pod = "unknown"

                        try:
                            import subprocess
                            result = subprocess.run(
                                ["docker", "inspect", "--format", "{{.Config.Image}}", container_id],
                                capture_output=True, text=True, timeout=3
                            )
                            if result.returncode == 0:
                                image = result.stdout.strip()
                        except (subprocess.TimeoutExpired, FileNotFoundError):
                            pass

                        return ContainerMeta(
                            pid=pid,
                            container_id=container_id,
                            image=image,
                            namespace=ns,
                            pod=pod,
                        )
    except PermissionError:
        logging.warning("Permission denied reading %s – run with CAP_SYS_ADMIN", cgroup_path)

    return None


# ---------------------------------------------------------------------------
# Ring buffer consumer (asyncio-compatible)
# ---------------------------------------------------------------------------

class RingBufferConsumer:
    """Reads events from a BPF_PERF_EVENT_ARRAY ring buffer page-by-page."""

    PAGE_SIZE = 4096
    HEADER_SIZE = 16  // u64 id, u64 timestamp (bpf_perf_event_header)
    EVENT_SIZE = ctypes.sizeof(VerdictEvent)

    def __init__(self, buffer_path: str):
        self.buffer_path = Path(buffer_path)
        self.fd: Optional[int] = None
        self.running = False

    def _open(self) -> None:
        """Open the perf buffer file descriptor."""
        import ctypes.util

        # Use libbpf's perf_buffer via ctypes to avoid the Go bridge
        self.fd = os.open(self.buffer_path, os.O_RDONLY | os.O_NONBLOCK)
        if self.fd < 0:
            raise OSError(f"Failed to open perf buffer: {self.buffer_path}")

    async def consume(self, callback) -> None:
        """Continuously read pages and invoke callback with parsed events."""
        self._open()
        self.running = True
        loop = asyncio.get_running_loop()

        try:
            while self.running:
                # Read a full page from the ring buffer
                try:
                    page = await loop.run_in_executor(
                        None, os.read, self.fd, self.PAGE_SIZE
                    )
                except BlockingIOError:
                    # No data available – yield to the event loop
                    await asyncio.sleep(0.001)
                    continue
                except OSError as exc:
                    logging.error("Read error on perf buffer: %v", exc)
                    await asyncio.sleep(1)
                    continue

                if len(page) < self.HEADER_SIZE:
                    continue

                # Parse the page: skip the header, extract VerdictEvent structs
                offset = self.HEADER_SIZE
                while offset + self.EVENT_SIZE <= len(page):
                    event = VerdictEvent.from_buffer_copy(
                        page[offset:offset + self.EVENT_SIZE]
                    )
                    callback(event)
                    offset += self.EVENT_SIZE
        finally:
            os.close(self.fd)
            self.running = False

    def stop(self) -> None:
        self.running = False


# ---------------------------------------------------------------------------
# Alerting logic
# ---------------------------------------------------------------------------

async def anomaly_detector(check_interval: float = 60.0) -> None:
    """Periodically scan CIDR_DENY_COUNT and log alerts for hot IPs."""
    global CIDR_DENY_COUNT

    while True:
        await asyncio.sleep(check_interval)

        snapshot = dict(CIDR_DENY_COUNT)
        CIDR_DENY_COUNT.clear()  # Reset the window

        for cidr_prefix, count in snapshot.items():
            if count > CIDR_DENY_THRESHOLD:
                logging.warning(
                    "🚨 ANOMALY: %s received %d denied packets in %ds – possible scan",
                    cidr_prefix, count, int(check_interval)
                )
                PACKET_COUNT.labels(
                    verdict="anomaly_alert", protocol="any", dst_port="0"
                ).inc()


# ---------------------------------------------------------------------------
# Main entry point
# ---------------------------------------------------------------------------

def event_handler(event: VerdictEvent) -> None:
    """Process a single verdict event: update metrics and detect anomalies."""
    verdict_label = event.verdict_name().lower()
    proto_label = event.protocol_name().lower()
    port_label = str(event.dst_port)

    PACKET_COUNT.labels(
        verdict=verdict_label, protocol=proto_label, dst_port=port_label
    ).inc()

    # Track per-/24 deny counts for anomaly detection
    if event.verdict == 0:  # DENY
        # Mask to /24 for aggregation
        masked = event.src_ip & 0xFFFFFF00
        cidr_prefix = f"{(masked >> 24) & 0xFF}.{(masked >> 16) & 0xFF}.{(masked >> 8) & 0xFF}.0/24"
        CIDR_DENY_COUNT[cidr_prefix] = CIDR_DENY_COUNT.get(cidr_prefix, 0) + 1

    # Log every deny event for forensic analysis
    if event.verdict == 0:
        container = resolve_container(event.pid)
        container_info = f" container={container.container_id}" if container else " host-process"
        logging.info(
            "DENY: %s:%d → %s:%d proto=%s pid=%d%s",
            event.src_ip_str(), 0,  # src port not captured in this BPF program
            event.dst_ip_str(), event.dst_port,
            event.protocol_name(), event.pid, container_info,
        )


async def main() -> None:
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
    )

    # Start Prometheus metrics server on port 9091
    start_http_server(9091)
    logging.info("Prometheus metrics exposed on :9091/metrics")

    # Start the anomaly detector
    detector_task = asyncio.create_task(anomaly_detector(check_interval=60))

    # Start consuming the ring buffer
    buffer_path = os.environ.get(
        "FIREWALL_PERF_BUFFER", "/sys/fs/bpf/firewall/events"
    )
    consumer = RingBufferConsumer(buffer_path)

    try:
        logging.info("Starting firewall event monitor...")
        await consumer.consume(event_handler)
    except asyncio.CancelledError:
        logging.info("Monitor shutting down...")
    finally:
        consumer.stop()
        detector_task.cancel()


if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env bash
# compile_bpf.sh – Build BPF programs with BTF support
set -euo pipefail
clang -O2 -g -target bpf -D__TARGET_ARCH_x86 \
    -I/usr/include/bpf \
    -I./include \
    -c firewall.c \
    -o firewall.o

# Verify BTF sections exist
bpftool btf dump file firewall.o format c > /dev/null 2>&1 \&& echo "BTF OK" || echo "BTF MISSING"

package firewall.test

# Ensure the default action is always deny
test_default_deny {
    policy := json.unmarshal(input.policy_raw)
    policy.default_action == "deny"
}

# Ensure database ports are never blocked from the API subnet
test_db_access_preserved {
    policy := json.unmarshal(input.policy_raw)
    some rule in policy.rules
    rule.destination_ports[_] == 5432
    cidr in rule.source_cidrs
    startswith(cidr, "10.1.0.0/16")  # API subnet
    rule.action == "allow"
}

# Rate-limit rules must have a non-zero PPS
test_rate_limit_has_pps {
    policy := json.unmarshal(input.policy_raw)
    some rule in policy.rules
    rule.action == "rate_limit"
    rule.rate_limit_pps > 0
}

// monitor_map.go – Periodically report BPF map utilization.
package main

import (
    "fmt"
    "log"
    "time"

    "github.com/cilium/ebpf"
    "github.com/prometheus/client_golang/prometheus"
)

var mapEntries = prometheus.NewGauge(
    prometheus.GaugeOpts{
        Name: "firewall_cidr_map_entries",
        Help: "Current number of entries in the CIDR rules map",
    },
)

func init() {
    prometheus.MustRegister(mapEntries)
}

func watchMapUtilization(m *ebpf.Map, maxEntries int, interval time.Duration) {
    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    for range ticker.C {
        count := 0
        iter := m.Iter()
        for iter.Next() {
            count++
        }
        if err := iter.Err(); err != nil {
            log.Printf("WARN: map iteration error: %v", err)
            continue
        }

        mapEntries.Set(float64(count))
        utilization := float64(count) / float64(maxEntries) * 100

        if utilization > 80 {
            log.Printf("ALERT: CIDR map at %.1f%% capacity (%d/%d entries)",
                utilization, count, maxEntries)
        }
    }
}

func main() {
    // Example usage – call from your DaemonSet init container
    fmt.Println("Map utilization monitor ready.")
}