Implement GeoIP feature

This commit implements filterring results based on hand crafted GeoIP
database with private networks and tags.

Signed-off-by: Dinar Valeev <[email protected]>
Co-authored-by: Yury Tsarev <[email protected]>

Makefile

@@ -48,7 +48,7 @@ lint:
 	golangci-lint run
 
 build:
-	CGO_ENABLED=0 go build cmd/coredns.go
+	GOOS=linux CGO_ENABLED=0 go build cmd/coredns.go
 
 clean:
 	go clean
@@ -66,11 +66,13 @@ import-image:
 deploy-app: image import-image
 	kubectl config use-context k3d-coredns-crd
 	kubectl apply -f terratest/example/ns.yaml 
+	kubectl create -n coredns configmap geodata --from-file terratest/geogen/geoip.mmdb || true
 	kubectl apply -f https://raw.githubusercontent.com/kubernetes-sigs/external-dns/master/docs/contributing/crd-source/crd-manifest.yaml
 	helm repo add coredns https://coredns.github.io/helm
 	helm repo update
 	cd charts/coredns && helm dependency update
 	helm upgrade -i coredns -n coredns charts/coredns \
+		-f terratest/helm_values.yaml \
 		--set coredns.image.tag=${TAG}
 
 .PHONY: lincense

README.md

@@ -43,6 +43,14 @@ k8s_crd example.com {
 }
 ```
 
+## Resolving order
+In case dnsEndpoint object's target has a label of `strategy: geoip` CoreDNS `k8s_crd` plugin will respond in a special way:
+* Assuming record has multiple IPs associated with it, and DNS message comes with edns0 `CLIENT-SUBNET` option.
+* CoreDNS will compare `DC` tag for IP extracted from `CLIENT-SUBNET` option against available Endpoint.Targets
+* Return only IPs where tags match
+* If IP has no common tag, all entries are returned.
+* CoreDNS must be supplied with a specially crafted GeoIP database in MaxMind DB format and mounted as `/geoip.mmdb` Refer to `terratest/geo` for examples.
+
 ## Build
 
 ### With compile-time configuration file

gateway.go

@@ -32,7 +32,7 @@ import (
 
 const defaultSvc = "external-dns.kube-system"
 
-type lookupFunc func(indexKey string) ([]net.IP, endpoint.TTL)
+type lookupFunc func(indexKey string, clientIP net.IP) ([]net.IP, endpoint.TTL)
 
 type resourceWithIndex struct {
 	name   string
@@ -98,13 +98,30 @@ func (gw *Gateway) updateResources(newResources []string) {
 	}
 }
 
+func extractEdnsSubnet(msg *dns.Msg) net.IP {
+	edns := msg.IsEdns0()
+	if edns == nil {
+		return nil
+	}
+	for _, o := range edns.Option {
+		if o.Option() == dns.EDNS0SUBNET {
+			subnet := o.(*dns.EDNS0_SUBNET)
+			return subnet.Address
+		}
+	}
+	return nil
+}
+
 // ServeDNS implements the plugin.Handle interface.
 func (gw *Gateway) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
+	var clientIP net.IP
 	state := request.Request{W: w, Req: r}
 	log.Infof("Incoming query %s", state.QName())
 
 	qname := state.QName()
 	zone := plugin.Zones(gw.Zones).Matches(qname)
+	clientIP = extractEdnsSubnet(r)
+
 	if zone == "" {
 		log.Infof("Request %s has not matched any zones %v", qname, gw.Zones)
 		return plugin.NextOrFailure(gw.Name(), gw.Next, ctx, w, r)
@@ -140,7 +157,7 @@ func (gw *Gateway) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Ms
 	// Iterate over supported resources and lookup DNS queries
 	// Stop once we've found at least one match
 	for _, resource := range gw.Resources {
-		addrs, ttl = resource.lookup(indexKey)
+		addrs, ttl = resource.lookup(indexKey, clientIP)
 		if len(addrs) > 0 {
 			break
 		}
@@ -210,7 +227,7 @@ func (gw *Gateway) SelfAddress(state request.Request) (records []dns.RR) {
 	var addrs []net.IP
 	var ttl endpoint.TTL
 	for _, resource := range gw.Resources {
-		addrs, ttl = resource.lookup(index)
+		addrs, ttl = resource.lookup(index, net.ParseIP(state.IP()))
 		if len(addrs) > 0 {
 			break
 		}

go.mod

@@ -5,10 +5,11 @@ go 1.15
 replace go.etcd.io/etcd => go.etcd.io/etcd v0.5.0-alpha.5.0.20200425165423-262c93980547
 
 require (
-	github.com/Azure/azure-sdk-for-go v46.0.0+incompatible // indirect
 	github.com/coredns/caddy v1.1.0
 	github.com/coredns/coredns v1.8.1
+	github.com/maxmind/mmdbwriter v0.0.0-20210616205632-85bfe8b3805f
 	github.com/miekg/dns v1.1.35
+	github.com/oschwald/maxminddb-golang v1.8.0
 	k8s.io/api v0.20.2
 	k8s.io/apimachinery v0.20.2
 	k8s.io/client-go v0.20.2

kubernetes.go

@@ -33,14 +33,20 @@ import (
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	endpoint "sigs.k8s.io/external-dns/endpoint"
+
 	// "k8s.io/client-go/tools/clientcmd"
+	"github.com/oschwald/maxminddb-golang"
 )
 
 const (
 	defaultResyncPeriod   = 0
 	endpointHostnameIndex = "endpointHostname"
 )
 
+type geo struct {
+	DC string `maxminddb:"datacenter"`
+}
+
 // KubeController stores the current runtime configuration and cache
 type KubeController struct {
 	client      dnsendpoint.ExtDNSInterface
@@ -103,9 +109,9 @@ func (ctrl *KubeController) HasSynced() bool {
 func RunKubeController(ctx context.Context, c *Gateway) (*KubeController, error) {
 	config, err := rest.InClusterConfig()
 
-	//Helpful to run coredns locally
-	//kubeconfig := os.Getenv("KUBECONFIG")
-	//config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
+	// Helpful to run coredns locally
+	// kubeconfig := os.Getenv("KUBECONFIG")
+	// config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
 
 	if err != nil {
 		return nil, err
@@ -157,11 +163,17 @@ func endpointHostnameIndexFunc(obj interface{}) ([]string, error) {
 	return hostnames, nil
 }
 
-func fetchEndpointIPs(endpoints []*endpoint.Endpoint, host string) (results []net.IP, ttl endpoint.TTL) {
+func fetchEndpointIPs(endpoints []*endpoint.Endpoint, host string, ip net.IP) (results []net.IP, ttl endpoint.TTL) {
 	for _, ep := range endpoints {
 		if ep.DNSName == host {
-			results = extractEndpointIPs(ep)
 			ttl = ep.RecordTTL
+			if ep.Labels["strategy"] == "geoip" {
+				results = extractGeo(ep, ip)
+				if len(results) > 0 {
+					return
+				}
+			}
+			results = extractEndpointIPs(ep)
 		}
 	}
 	return
@@ -171,17 +183,55 @@ func extractEndpointIPs(endpoint *endpoint.Endpoint) (result []net.IP) {
 	for _, ip := range endpoint.Targets {
 		result = append(result, net.ParseIP(ip))
 	}
-	return
+	return result
+}
+func extractGeo(endpoint *endpoint.Endpoint, clientIP net.IP) (result []net.IP) {
+	db, err := maxminddb.Open("geoip.mmdb")
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer db.Close()
+
+	clientGeo := &geo{}
+	err = db.Lookup(clientIP, clientGeo)
+	if err != nil {
+		return nil
+	}
+
+	if clientGeo.DC == "" {
+		log.Infof("empty DC %+v", clientGeo)
+		return result
+	}
+
+	log.Infof("clientDC: %+v", clientGeo)
+
+	for _, ip := range endpoint.Targets {
+		geoData := &geo{}
+		log.Infof("processing IP %+v", ip)
+		err = db.Lookup(net.ParseIP(ip), geoData)
+		if err != nil {
+			log.Error(err)
+			continue
+		}
+
+		log.Infof("IP info: %+v", geoData.DC)
+		if clientGeo.DC == geoData.DC && geoData.DC != "" {
+			result = append(result, net.ParseIP(ip))
+		}
+	}
+	return result
 }
 
-func lookupEndpointIndex(ctrl cache.SharedIndexInformer) func(string) ([]net.IP, endpoint.TTL) {
-	return func(indexKey string) (result []net.IP, ttl endpoint.TTL) {
+func lookupEndpointIndex(ctrl cache.SharedIndexInformer) func(string, net.IP) ([]net.IP, endpoint.TTL) {
+	return func(indexKey string, clientIP net.IP) (result []net.IP, ttl endpoint.TTL) {
 
+		log.Infof("Index key %+v", indexKey)
 		objs, _ := ctrl.GetIndexer().ByIndex(endpointHostnameIndex, strings.ToLower(indexKey))
 		for _, obj := range objs {
 			endpoint := obj.(*endpoint.DNSEndpoint)
-			result, ttl = fetchEndpointIPs(endpoint.Spec.Endpoints, indexKey)
+			result, ttl = fetchEndpointIPs(endpoint.Spec.Endpoints, indexKey, clientIP)
 		}
+
 		return
 	}
 }

terratest/example/geo.yaml

@@ -0,0 +1,17 @@
+apiVersion: externaldns.k8s.io/v1alpha1
+kind: DNSEndpoint
+metadata:
+  name: geo-test
+  labels:
+    k8gb.absa.oss/dnstype: local
+spec:
+  endpoints:
+  - dnsName: geo.example.org
+    recordType: A
+    labels:
+      strategy: "geoip"
+    targets:
+    - 192.200.1.5
+    - 192.200.1.10
+    - 192.200.2.5
+    - 192.200.2.10

terratest/geogen/README.md

@@ -0,0 +1,3 @@
+This is an example program used to generate `geoip.mmdb` for e2e test.
+
+I order to execute it, simply run: `go run main.go`

terratest/geogen/main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"log"
+	"net"
+	"os"
+
+	"github.com/maxmind/mmdbwriter"
+	"github.com/maxmind/mmdbwriter/inserter"
+	"github.com/maxmind/mmdbwriter/mmdbtype"
+)
+
+func main() {
+	fh, err := os.Create("geoip.mmdb")
+	if err != nil {
+		log.Fatal(err)
+	}
+	writer, _ := mmdbwriter.New(
+		mmdbwriter.Options{
+			DatabaseType: "Test-IP-DB",
+			RecordSize:   24,
+			IPVersion:    4,
+		},
+	)
+	_, absaSDCNet, _ := net.ParseCIDR("192.200.1.0/24")
+	_, absa270Net, _ := net.ParseCIDR("192.200.2.0/24")
+
+	absaSDCData := mmdbtype.Map{
+		"datacenter": mmdbtype.String("site1"),
+	}
+
+	absa270Data := mmdbtype.Map{
+		"datacenter": mmdbtype.String("site2"),
+	}
+
+	if err := writer.InsertFunc(absaSDCNet, inserter.TopLevelMergeWith(absaSDCData)); err != nil {
+		log.Fatal(err)
+	}
+	if err := writer.InsertFunc(absa270Net, inserter.TopLevelMergeWith(absa270Data)); err != nil {
+		log.Fatal(err)
+	}
+
+	_, err = writer.WriteTo(fh)
+	if err != nil {
+		log.Fatal(err)
+	}
+}

terratest/helm_values.yaml

@@ -0,0 +1,9 @@
+coredns:
+  extraVolumes:
+   - name: geo-data 
+     configMap:
+       name: geodata
+  extraVolumeMounts:
+   - name: geo-data 
+     mountPath: /geoip.mmdb
+     subPath: geoip.mmdb 

terratest/test/basic_test.go

@@ -38,6 +38,7 @@ func TestBasicExample(t *testing.T) {
 
 	var coreDNSPods []corev1.Pod
 
+	clientIP := ""
 	// Path to the Kubernetes resource config we will test
 	kubeResourcePath, err := filepath.Abs("../example/dnsendpoint.yaml")
 	require.NoError(t, err)
@@ -74,27 +75,27 @@ func TestBasicExample(t *testing.T) {
 	}
 
 	t.Run("Basic type A resolve", func(t *testing.T) {
-		actualIP, err := DigIPs(t, "localhost", 1053, "host1.example.org", dns.TypeA)
+		actualIP, err := DigIPs(t, "localhost", 1053, "host1.example.org", dns.TypeA, clientIP)
 		require.NoError(t, err)
 		assert.Contains(t, actualIP, "1.2.3.4")
 	})
 
 	// check for NODATA replay on non labeled endpoints
 	t.Run("NODATA reply on non labeled endpoints", func(t *testing.T) {
-		emptyIP, err := DigIPs(t, "localhost", 1053, "host3.example.org", dns.TypeA)
+		emptyIP, err := DigIPs(t, "localhost", 1053, "host3.example.org", dns.TypeA, clientIP)
 		require.NoError(t, err)
 		assert.NotContains(t, emptyIP, "1.2.3.4")
 	})
 
 	t.Run("Validate artificial(broken) DNS doesn't break CoreDNS", func(t *testing.T) {
 		k8s.KubectlApply(t, options, brokenEndpoint)
-		_, err := DigIPs(t, "localhost", 1053, "broken1.example.org", dns.TypeA)
+		_, err := DigIPs(t, "localhost", 1053, "broken1.example.org", dns.TypeA, clientIP)
 		require.Error(t, err)
-		_, err = DigIPs(t, "localhost", 1053, "broken2.example.org", dns.TypeA)
+		_, err = DigIPs(t, "localhost", 1053, "broken2.example.org", dns.TypeA, clientIP)
 		require.Error(t, err)
 
 		// We still able to get "healthy" records
-		currentIP, err := DigIPs(t, "localhost", 1053, "host1.example.org", dns.TypeA)
+		currentIP, err := DigIPs(t, "localhost", 1053, "host1.example.org", dns.TypeA, clientIP)
 		require.NoError(t, err)
 		assert.Contains(t, currentIP, "1.2.3.4")
 	})