diff --git a/AGENTS.md b/AGENTS.md index 9a65875..b112d31 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,6 +32,8 @@ internal/ qlog/ bbolt query-log ring + live in-memory ring + SSE fanout metrics/ Prometheus collectors + rolling time-series rings cluster/ Valkey pub/sub, leader lease, divergence detection + kubesource/ read-only watch of the fancydns.k6n.net CRDs (DNSZone, + ClientGroup, BlockList) -> Snapshot -> config overlay api/ admin REST + SSE, auth, CORS, rate limiting app/ wiring, lifecycle, API hook implementations, memory envelope web/ go:embed of the built console @@ -103,6 +105,16 @@ Break these and you break the design, not just the test. a mirror N times. Followers serve their compiled copy. - **Blocklist changes flush the cache** rather than guessing which subtrees are affected (`invalidateBlocklistCache()` in `internal/app/hooks.go`). +- **The Kubernetes source is read-only and is an overlay, not a config.** + `config.Store` keeps `Base()` (what the YAML says) beside the effective + config (`Get()`, base + cluster objects). Anything that writes the config back + to the file — `mutateConfig`, `SetAdminAddr`, the settings editor — must start + from `Base()`, or cluster objects get copied into the YAML where they shadow + their own objects. YAML wins every collision (groups run first; a same-named + list is dropped). A cluster object is validated on its own: a bad object is + skipped and keeps its last good version, it never fails the merge. A BlockList + has no `path` and refuses `file://`, because a namespaced object must not + choose a file the runtime reads or writes. - **API keys are compared by SHA-256 hash, constant-time**, and `X-Forwarded-For` is honoured *only* from `trusted_proxies`. Honouring it from anyone lets a client spoof its identity past the ACL and the rate limiter. diff --git a/cmd/fancy-dns/main.go b/cmd/fancy-dns/main.go index 822a557..c3abf46 100644 --- a/cmd/fancy-dns/main.go +++ b/cmd/fancy-dns/main.go @@ -21,6 +21,10 @@ func main() { checkOnly = flag.Bool("check", false, "validate the configuration and exit") logLevel = flag.String("log-level", os.Getenv("FANCYDNS_LOG_LEVEL"), "override log level (debug, info, warn, error)") logFormat = flag.String("log-format", os.Getenv("FANCYDNS_LOG_FORMAT"), "override log format (text, json)") + kube = flag.Bool("kube", envBool("FANCYDNS_KUBE"), "watch Kubernetes for ClientGroup, BlockList and DNSZone objects") + kubeconfig = flag.String("kubeconfig", os.Getenv("FANCYDNS_KUBECONFIG"), "kubeconfig for -kube (default: in-cluster service account, then $KUBECONFIG, then ~/.kube/config)") + kubeContext = flag.String("kube-context", os.Getenv("FANCYDNS_KUBE_CONTEXT"), "kubeconfig context for -kube") + kubeNS = flag.String("kube-namespace", os.Getenv("FANCYDNS_KUBE_NAMESPACE"), "namespace to watch for -kube (default: the pod's own, else all)") adminAddr = flag.String("admin-addr", os.Getenv("FANCYDNS_ADMIN_ADDR"), "override the admin API listen address") ) flag.Parse() @@ -36,6 +40,12 @@ func main() { logFormat: *logFormat, adminAddr: *adminAddr, checkOnly: *checkOnly, + kube: app.KubeOptions{ + Enabled: *kube || *kubeconfig != "" || *kubeContext != "", + Kubeconfig: *kubeconfig, + Context: *kubeContext, + Namespace: *kubeNS, + }, }); err != nil { fmt.Fprintln(os.Stderr, "fancy-dns:", err) os.Exit(1) @@ -48,6 +58,7 @@ type runOptions struct { logFormat string adminAddr string checkOnly bool + kube app.KubeOptions } func run(o runOptions) error { @@ -56,6 +67,7 @@ func run(o runOptions) error { Version: app.Version, LogLevel: o.logLevel, LogFormat: o.logFormat, + Kube: o.kube, }) if err != nil { return err @@ -124,3 +136,11 @@ func envOr(key, def string) string { } return def } + +func envBool(key string) bool { + switch os.Getenv(key) { + case "1", "true", "TRUE", "True", "yes": + return true + } + return false +} diff --git a/deploy/k8s/crds/crds.yaml b/deploy/k8s/crds/crds.yaml new file mode 100644 index 0000000..5e0ae64 --- /dev/null +++ b/deploy/k8s/crds/crds.yaml @@ -0,0 +1,124 @@ +# The fancy-dns custom resources. Cluster-scoped definitions, namespaced +# objects. The runtime (`fancy-dns -kube`) watches them read-only. +# +# Field names are snake_case where they mirror config.yaml (allow_domains, +# max_concurrent, ...), so a group copied out of the YAML pastes in unchanged. +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: dnszones.fancydns.k6n.net +spec: + group: fancydns.k6n.net + scope: Namespaced + names: {kind: DNSZone, listKind: DNSZoneList, plural: dnszones, singular: dnszone, shortNames: [dnsz]} + versions: + - name: v1alpha1 + served: true + storage: true + additionalPrinterColumns: + - {name: Zone, type: string, jsonPath: .spec.zone} + - {name: Age, type: date, jsonPath: .metadata.creationTimestamp} + schema: + openAPIV3Schema: + type: object + required: [spec] + properties: + spec: + type: object + required: [zone] + properties: + zone: {type: string, minLength: 3, maxLength: 253, description: "Zone apex, e.g. home.example.com"} + ttl: {type: integer, minimum: 1, maximum: 604800, description: "Zone $TTL and SOA minimum (seconds)"} + soa: {type: string, maxLength: 512, description: "SOA rdata; omit to synthesise one from the apex NS"} + records: + type: array + maxItems: 5000 + items: + type: object + required: [name, type, value] + properties: + name: {type: string, minLength: 1, maxLength: 253, description: "Owner; @ is the apex, a name without a trailing dot is relative to the zone"} + type: {type: string, enum: [A, AAAA, CAA, CNAME, DHCID, HINFO, HTTPS, LOC, MX, NAPTR, NS, PTR, SRV, SPF, SSHFP, SVCB, TXT, URI]} + ttl: {type: integer, minimum: 1, maximum: 604800} + value: {type: string, minLength: 1, maxLength: 4096, description: "Presentation-format RDATA"} +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: clientgroups.fancydns.k6n.net +spec: + group: fancydns.k6n.net + scope: Namespaced + names: {kind: ClientGroup, listKind: ClientGroupList, plural: clientgroups, singular: clientgroup, shortNames: [cg]} + versions: + - name: v1alpha1 + served: true + storage: true + additionalPrinterColumns: + - {name: Policy, type: string, jsonPath: .spec.policy} + - {name: Priority, type: integer, jsonPath: .spec.priority} + - {name: Age, type: date, jsonPath: .metadata.creationTimestamp} + schema: + openAPIV3Schema: + type: object + required: [spec] + properties: + spec: + type: object + required: [policy, match] + properties: + priority: {type: integer, default: 0, description: "Groups are first-match-wins; lower runs first, ties break on name. YAML groups always run before these."} + comment: {type: string, maxLength: 512} + policy: {type: string, enum: [allow, block_only, deny]} + match: + type: object + properties: + cidrs: {type: array, items: {type: string}} + ptr_suffixes: {type: array, items: {type: string}} + tags: {type: array, items: {type: string}} + any: {type: boolean} + allow_domains: {type: array, items: {type: string}} + deny_domains: {type: array, items: {type: string}} + skip_blocklist: {type: boolean} + skip_cache: {type: boolean} + rate_limit: {type: integer, minimum: 0} + max_concurrent: {type: integer, minimum: 0} + log_queries: {type: boolean} + upstreams: {type: array, items: {type: string}} +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + name: blocklists.fancydns.k6n.net +spec: + group: fancydns.k6n.net + scope: Namespaced + names: {kind: BlockList, listKind: BlockListList, plural: blocklists, singular: blocklist, shortNames: [bl]} + versions: + - name: v1alpha1 + served: true + storage: true + additionalPrinterColumns: + - {name: URL, type: string, jsonPath: .spec.url} + - {name: Format, type: string, jsonPath: .spec.format} + - {name: Disabled, type: boolean, jsonPath: .spec.disabled} + - {name: Age, type: date, jsonPath: .metadata.creationTimestamp} + schema: + openAPIV3Schema: + type: object + required: [spec] + properties: + spec: + type: object + required: [url] + properties: + url: {type: string, pattern: '^https?://', description: "http(s) only; file paths are not accepted from the cluster"} + format: {type: string, enum: [auto, hosts, adblock, plain, rpz]} + mirror: {type: array, items: {type: string, pattern: '^https?://'}} + disabled: {type: boolean} + refresh: {type: string, description: "Go duration, e.g. 12h (default 24h)"} + timeout: {type: string, description: "Go duration, e.g. 30s"} + max_bytes: {type: integer, minimum: 1} + categories: {type: array, items: {type: string}} + comment: {type: string, maxLength: 512} + insecure_skip_tls_verify: {type: boolean} diff --git a/deploy/k8s/crds/examples.yaml b/deploy/k8s/crds/examples.yaml new file mode 100644 index 0000000..445f6a1 --- /dev/null +++ b/deploy/k8s/crds/examples.yaml @@ -0,0 +1,36 @@ +apiVersion: fancydns.k6n.net/v1alpha1 +kind: DNSZone +metadata: + name: home + namespace: dns +spec: + zone: home.example.com + ttl: 300 + records: + - {name: "@", type: NS, value: ns1.home.example.com.} + - {name: ns1, type: A, value: 10.0.0.53} + - {name: nas, type: A, value: 10.0.0.20} + - {name: "*.apps", type: A, value: 10.0.0.30} +--- +apiVersion: fancydns.k6n.net/v1alpha1 +kind: ClientGroup +metadata: + name: kids + namespace: dns +spec: + priority: 10 + policy: block_only + match: + cidrs: [10.0.20.0/24] + deny_domains: ["**.tiktok.com"] +--- +apiVersion: fancydns.k6n.net/v1alpha1 +kind: BlockList +metadata: + name: stevenblack + namespace: dns +spec: + url: https://raw.githubusercontent.com/StevenBlack/hosts/master/hosts + format: hosts + refresh: 24h + categories: [ads, malware] diff --git a/deploy/k8s/crds/kustomization.yaml b/deploy/k8s/crds/kustomization.yaml new file mode 100644 index 0000000..fd677dc --- /dev/null +++ b/deploy/k8s/crds/kustomization.yaml @@ -0,0 +1,8 @@ +# Apply the CRDs before anything that uses them: +# kubectl apply -f deploy/k8s/crds/crds.yaml +# kubectl apply -k deploy/k8s/crds # RBAC for the runtime +# examples.yaml is deliberately not included. +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +resources: + - rbac.yaml diff --git a/deploy/k8s/crds/rbac.yaml b/deploy/k8s/crds/rbac.yaml new file mode 100644 index 0000000..ecc0c86 --- /dev/null +++ b/deploy/k8s/crds/rbac.yaml @@ -0,0 +1,26 @@ +# Read-only access to the fancy-dns objects in one namespace, for the existing +# `fancy-dns` service account. For a runtime outside the cluster, bind the same +# Role (or a ClusterRole for -kube-namespace="") to the user in its kubeconfig. +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: fancy-dns-objects + namespace: dns +rules: + - apiGroups: [fancydns.k6n.net] + resources: [dnszones, clientgroups, blocklists] + verbs: [get, list, watch] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: fancy-dns-objects + namespace: dns +subjects: + - kind: ServiceAccount + name: fancy-dns + namespace: dns +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: fancy-dns-objects diff --git a/go.mod b/go.mod index e2416ea..4caba78 100644 --- a/go.mod +++ b/go.mod @@ -10,21 +10,63 @@ require ( github.com/redis/go-redis/v9 v9.22.0 go.etcd.io/bbolt v1.5.0 golang.org/x/net v0.59.0 + golang.org/x/oauth2 v0.36.0 gopkg.in/yaml.v3 v3.0.1 + k8s.io/apimachinery v0.37.0 + k8s.io/client-go v0.37.0 ) require ( github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/emicklei/go-restful/v3 v3.13.0 // indirect + github.com/fxamacker/cbor/v2 v2.9.1 // indirect github.com/go-jose/go-jose/v4 v4.1.4 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-openapi/jsonpointer v1.0.0 // indirect + github.com/go-openapi/jsonreference v1.0.0 // indirect + github.com/go-openapi/swag v0.27.1 // indirect + github.com/go-openapi/swag/cmdutils v0.27.1 // indirect + github.com/go-openapi/swag/conv v0.27.1 // indirect + github.com/go-openapi/swag/fileutils v0.27.1 // indirect + github.com/go-openapi/swag/jsonutils v0.27.1 // indirect + github.com/go-openapi/swag/loading v0.27.1 // indirect + github.com/go-openapi/swag/mangling v0.27.1 // indirect + github.com/go-openapi/swag/netutils v0.27.1 // indirect + github.com/go-openapi/swag/pools v0.27.1 // indirect + github.com/go-openapi/swag/stringutils v0.27.1 // indirect + github.com/go-openapi/swag/typeutils v0.27.1 // indirect + github.com/go-openapi/swag/yamlutils v0.27.1 // indirect + github.com/google/gnostic-models v0.7.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.70.1 // indirect github.com/prometheus/procfs v0.21.1 // indirect + github.com/spf13/pflag v1.0.10 // indirect + github.com/x448/float16 v0.8.4 // indirect github.com/yuin/gopher-lua v1.1.1 // indirect go.uber.org/atomic v1.11.0 // indirect - golang.org/x/oauth2 v0.36.0 // indirect + go.yaml.in/yaml/v2 v2.4.4 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/sys v0.48.0 // indirect + golang.org/x/term v0.46.0 // indirect golang.org/x/text v0.42.0 // indirect - google.golang.org/protobuf v1.36.11 // indirect + golang.org/x/time v0.15.0 // indirect + google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af // indirect + gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect + gopkg.in/inf.v0 v0.9.1 // indirect + k8s.io/api v0.37.0 // indirect + k8s.io/klog/v2 v2.140.0 // indirect + k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad // indirect + k8s.io/utils v0.0.0-20260626114624-be93311217bd // indirect + sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect + sigs.k8s.io/randfill v1.0.0 // indirect + sigs.k8s.io/structured-merge-diff/v6 v6.4.2 // indirect + sigs.k8s.io/yaml v1.6.0 // indirect ) diff --git a/go.sum b/go.sum index ee2ab72..35a5b0f 100644 --- a/go.sum +++ b/go.sum @@ -10,24 +10,84 @@ github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UF github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/coreos/go-oidc/v3 v3.21.0 h1:wZo4Q9Pum8dYEj0eMUPrqR+kvuGkeUplbLpNCkBqoWM= github.com/coreos/go-oidc/v3 v3.21.0/go.mod h1:DYCf24+ncYi+XkIH97GY1+dqoRlbaSI26KVTCI9SrY4= -github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= +github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/emicklei/go-restful/v3 v3.13.0 h1:C4Bl2xDndpU6nJ4bc1jXd+uTmYPVUwkD6bFY/oTyCes= +github.com/emicklei/go-restful/v3 v3.13.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= +github.com/fxamacker/cbor/v2 v2.9.1 h1:2rWm8B193Ll4VdjsJY28jxs70IdDsHRWgQYAI80+rMQ= +github.com/fxamacker/cbor/v2 v2.9.1/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA= github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-openapi/jsonpointer v1.0.0 h1:kR9tHqY0CtZaOPVFm622dPVNhrvYpwr4uCxgL3h1H8s= +github.com/go-openapi/jsonpointer v1.0.0/go.mod h1:Z3rw7dWu1p9IgitXCFamSlA5lmDiklEB6vkaxcNZW5Y= +github.com/go-openapi/jsonreference v1.0.0 h1:jlmTr6torcd1YgDQvSfNmRtKzYDO4FGBkrAdlAVWnpY= +github.com/go-openapi/jsonreference v1.0.0/go.mod h1:jtwdyGbJk0Xhe5Y+rwtglQP6Sb1WZST4rT32LWB+sv0= +github.com/go-openapi/swag v0.27.1 h1:VotvOLWW8q/EAxB0YdsBBGC8XYyeL1YwBj2ungAGPNg= +github.com/go-openapi/swag v0.27.1/go.mod h1:GTkJPwHfhJp6MWr4/rCh64HVI3Ofu+tcsbfjfHmTxpE= +github.com/go-openapi/swag/cmdutils v0.27.1 h1:I7sYqaWVl5mq0NEmNQkAmFDyNin9ufvMX/p2zwtQaOE= +github.com/go-openapi/swag/cmdutils v0.27.1/go.mod h1:Sm1MVFMkF6guJJ+pQqHnQA3N0j9qALV3NxzDSv6bETM= +github.com/go-openapi/swag/conv v0.27.1 h1:8wi9ZG+olmY1wXphl93EWniPtbSPkXM/feH7FgjsvrU= +github.com/go-openapi/swag/conv v0.27.1/go.mod h1:QbqMivkpKhC3g1B1GGGOJ6ANewI3S62dbzYu3Duowqs= +github.com/go-openapi/swag/fileutils v0.27.1 h1:QQqBSoi5mW4XpU85nS0mLcA+zAE6vLzrb0QkmLKf9oM= +github.com/go-openapi/swag/fileutils v0.27.1/go.mod h1:VvJFZLTZS0AI854gEQz5tk7dBESdLjiNUMSZ/th2ry8= +github.com/go-openapi/swag/jsonutils v0.27.1 h1:SVgK3i4USzCU5mibOOS/l4ea2h9UQXy7J7RNLTjuXjU= +github.com/go-openapi/swag/jsonutils v0.27.1/go.mod h1:tdlEpZqdcQ17uj6J4YdK9vd8It5qWMwjWXOs0tjpRlk= +github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1 h1:mJu3COL9WEaZVp/Kf2PRMi7tPszPEJfSr/OO75ynCs8= +github.com/go-openapi/swag/jsonutils/fixtures_test v0.27.1/go.mod h1:mofwUWx70wvskwESqRJ//k/9kURmCgyJl5m5Ppoh5kY= +github.com/go-openapi/swag/loading v0.27.1 h1:/DxUgDXKbBX4bcn7r9uEXfJyzN5XpiJmZplzQTjrRCY= +github.com/go-openapi/swag/loading v0.27.1/go.mod h1:jvGh3iA2+zyUUycB5fgJWzeHnhrpvGnJJM0RVE9ZShE= +github.com/go-openapi/swag/mangling v0.27.1 h1:yC9D0HyUE8gbP+BfmGx9+AA89ikwZTMjESK3OnnoaqA= +github.com/go-openapi/swag/mangling v0.27.1/go.mod h1:jtBE2+V+3pILxOR7Vgce+Cwp6A2PgZbvVqfNntbVs0w= +github.com/go-openapi/swag/netutils v0.27.1 h1:mICMFoS82F5TZ4Zy3cqmcQk+BFeCp3Uyq3Np7GI0/qU= +github.com/go-openapi/swag/netutils v0.27.1/go.mod h1:J+WYyFMLtvtCGqa6jLv+YNUmIKI3ZRQRrvfNDMoQoEQ= +github.com/go-openapi/swag/pools v0.27.1 h1:9LeadcMyb2GJCbXX5hVQDbZ2Lq9TL4dCs/nx1j5DO0E= +github.com/go-openapi/swag/pools v0.27.1/go.mod h1:kVQefhSK5RWuRe7BXsL8htgBPAMpN7HDGpGEknqugeE= +github.com/go-openapi/swag/stringutils v0.27.1 h1:ZXePZ0r2p1qSjo8tD3Un4vFj8+FqlCkczxDrJIhYUp8= +github.com/go-openapi/swag/stringutils v0.27.1/go.mod h1:lzRN95CxXmA03XcDWHLOb6nOMcxCqR5rGY0lOgsfRoM= +github.com/go-openapi/swag/typeutils v0.27.1 h1:KSTdFlfnse4r6dP9IrEnwMldjE+zs71UeEB3//PtVXc= +github.com/go-openapi/swag/typeutils v0.27.1/go.mod h1:Srm0xFNRZ1Y+vCxJclo5qzx8aj+1pAKda/YfFPrG0dQ= +github.com/go-openapi/swag/yamlutils v0.27.1 h1:ftxv6xvXb1E3zohUc+okZ9nSqNb9StQX/FXnKZ98sQA= +github.com/go-openapi/swag/yamlutils v0.27.1/go.mod h1:bnxFIB1qewGRiZHypXGZ3fNgf13/0HfRgnS/iZBDrOo= +github.com/go-openapi/testify/enable/yaml/v2 v2.6.0 h1:gGHwAJ0R/5jU8BEGDbfRNR3hL68dAVi84WuOApp29B0= +github.com/go-openapi/testify/enable/yaml/v2 v2.6.0/go.mod h1:tY+St1SGq4NFl0QIqdTY4aEdbChAHxhyB77XQi9iJCo= +github.com/go-openapi/testify/v2 v2.6.0 h1:5PKH2HE7YJ/LuRPQGvSxBRlFXNQhSetBLlGAgUEu3ug= +github.com/go-openapi/testify/v2 v2.6.0/go.mod h1:SgsVHtfooshd0tublTtJ50FPKhujf47YRqauXXOUxfw= +github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo= +github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/klauspost/compress v1.19.1 h1:VsB4HPswih7mmZ8WleSFQ75c/Ui1M4trX5oAsJnhSlk= github.com/klauspost/compress v1.19.1/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/miekg/dns v1.1.73 h1:uhT8nJxmTrPJYClxVxTCX+CVn6qnzSiybRk72Z6DgrE= github.com/miekg/dns v1.1.73/go.mod h1:RW2Obtfd5NZHvOFe3zYG0W8koWOQtAzyHaLo8vASBuQ= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= -github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U= +github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/prometheus/client_golang v1.24.1 h1:JnJkREXzWxUdCuPFpIWZiPispT9xVV59uiuyR2bPlnU= github.com/prometheus/client_golang v1.24.1/go.mod h1:F+oSRECHg4sse5ucfYpYDeIv/hu68Zo0uoHKetWnzcE= github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= @@ -38,8 +98,18 @@ github.com/prometheus/procfs v0.21.1 h1:GljZCt+zSTS+NZq88cyQ1LjZ+RCHp3uVuabBWA5+ github.com/prometheus/procfs v0.21.1/go.mod h1:aB55Cww9pdSJVHk0hUf0inxWyyjPogFIjmHKYgMKmtY= github.com/redis/go-redis/v9 v9.22.0 h1:laDvpYXTJtZLloinw1fA5Kqd6HAEH2XKxOkG/PDq2F0= github.com/redis/go-redis/v9 v9.22.0/go.mod h1:y2g0Wj8rQvuK0ELM+oxSudcLtC09JScs98I/X9gRWY4= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= +github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= +github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= @@ -52,6 +122,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues= golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= @@ -60,11 +132,40 @@ golang.org/x/sync v0.23.0 h1:KameEIfc1IkluZyXWLn39Wd4tURc6GbCiISGiZm2bQk= golang.org/x/sync v0.23.0/go.mod h1:sUUOizhqBxiL6pEWpqNLUiaJn1ShEbZ6BBqskPbjZm0= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= +golang.org/x/term v0.46.0 h1:3+OXuTbaKDgwk8jTi3aSLHRlmWqHEUDUtxnbFigO4YE= +golang.org/x/term v0.46.0/go.mod h1:+K02xbkittuwc0Am4abfA3Fc+XRGXkvBXNO88NCXPoc= golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI= golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E= -google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= -google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= +google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af h1:+5/Sw3GsDNlEmu7TfklWKPdQ0Ykja5VEmq2i817+jbI= +google.golang.org/protobuf v1.36.12-0.20260120151049-f2248ac996af/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/evanphx/json-patch.v4 v4.13.0 h1:czT3CmqEaQ1aanPc5SdlgQrrEIb8w/wwCvWWnfEbYzo= +gopkg.in/evanphx/json-patch.v4 v4.13.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M= +gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= +gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +k8s.io/api v0.37.0 h1:Z//Vj9N7RA/yS2sDmxyeo7h+RR4zbUrd2vrd3Z0TbB4= +k8s.io/api v0.37.0/go.mod h1:LKXgcJWMc+f4OLbP5SFR8rulEg07zZhpi/zMULiBImk= +k8s.io/apimachinery v0.37.0 h1:Np2AbDtf8x6RDHiD8T9LbKJ9gaegeVNa8yNm5FuGKm0= +k8s.io/apimachinery v0.37.0/go.mod h1:RN3nhprFSCxOi5Selxd7oMTXOe/c+ZbcE7Im+TS2zkE= +k8s.io/client-go v0.37.0 h1:nsN31fy8wBySuZ+QRnKmrjRSQLOG2rvoGN0tKd12zhQ= +k8s.io/client-go v0.37.0/go.mod h1:FcGqw+Ll/gNQiq+nPGY1Oyt9y7SgDh1d3MW3RFDEbn0= +k8s.io/klog/v2 v2.140.0 h1:Tf+J3AH7xnUzZyVVXhTgGhEKnFqye14aadWv7bzXdzc= +k8s.io/klog/v2 v2.140.0/go.mod h1:o+/RWfJ6PwpnFn7OyAG3QnO47BFsymfEfrz6XyYSSp0= +k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad h1:oXImqH8mQNk7PmvzKhmN3ddJoY6OnyM225MXwGHPm0A= +k8s.io/kube-openapi v0.0.0-20260721132016-d427ff9ee9ad/go.mod h1:0/mqHCVhlumdJ3BhCfnjSZQE037nAhNodh1/hK0T8/I= +k8s.io/utils v0.0.0-20260626114624-be93311217bd h1:Ea7fgQ5we8Y9T0OX5o0dAHzQOBRI07D/dEYRaB9ZZEs= +k8s.io/utils v0.0.0-20260626114624-be93311217bd/go.mod h1:xDxuJ0whA3d0I4mf/C4ppKHxXynQ+fxnkmQH0vTHnuk= +sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg= +sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg= +sigs.k8s.io/randfill v1.0.0 h1:JfjMILfT8A6RbawdsK2JXGBR5AQVfd+9TbzrlneTyrU= +sigs.k8s.io/randfill v1.0.0/go.mod h1:XeLlZ/jmk4i1HRopwe7/aU3H5n1zNUcX6TM94b3QxOY= +sigs.k8s.io/structured-merge-diff/v6 v6.4.2 h1:qdOxHwrl2Kaag1aQEarlYcOA9vSyGCp3CIki3aW8c4Q= +sigs.k8s.io/structured-merge-diff/v6 v6.4.2/go.mod h1:M3W8sfWvn2HhQDIbGWj3S099YozAsymCo/wrT5ohRUE= +sigs.k8s.io/yaml v1.6.0 h1:G8fkbMSAFqgEFgh4b1wmtzDnioxFCUgTZhlbj5P9QYs= +sigs.k8s.io/yaml v1.6.0/go.mod h1:796bPqUfzR/0jLAl6XjHl3Ck7MiyVv8dbTdyT3/pMf4= diff --git a/internal/api/handlers.go b/internal/api/handlers.go index f7307b3..e37ed79 100644 --- a/internal/api/handlers.go +++ b/internal/api/handlers.go @@ -626,7 +626,9 @@ func (s *Server) handleSettingsGet(w http.ResponseWriter, _ *http.Request) { fileYAML = string(b) } } - cur := s.hooks.Store.Get() + // The base, not the effective config: the settings editor writes what it + // shows back to the file, and cluster objects must not end up in the YAML. + cur := s.hooks.Store.Base() effective, err := config.Marshal(cur) if err != nil { writeError(w, http.StatusInternalServerError, "cannot render config: "+err.Error()) diff --git a/internal/app/app.go b/internal/app/app.go index e00ba87..597e9c4 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -16,6 +16,7 @@ import ( "path/filepath" "strings" "sync" + "sync/atomic" "time" "git.k8s-git-k6n.net/mats/fancy-dns/internal/acl" @@ -25,6 +26,7 @@ import ( "git.k8s-git-k6n.net/mats/fancy-dns/internal/cluster" "git.k8s-git-k6n.net/mats/fancy-dns/internal/config" "git.k8s-git-k6n.net/mats/fancy-dns/internal/engine" + "git.k8s-git-k6n.net/mats/fancy-dns/internal/kubesource" "git.k8s-git-k6n.net/mats/fancy-dns/internal/kubezones" "git.k8s-git-k6n.net/mats/fancy-dns/internal/localzone" "git.k8s-git-k6n.net/mats/fancy-dns/internal/metrics" @@ -47,6 +49,8 @@ type Options struct { // LogLevel and LogFormat override the config's logging settings. LogLevel string LogFormat string + // Kube turns on the Kubernetes source for groups, lists and zones. + Kube KubeOptions } // App is the running process. @@ -65,6 +69,10 @@ type App struct { local *localzone.Holder // zones is the Kubernetes-sourced zone client. zones *kubezones.Client + // kube watches the cluster's fancy-dns objects when enabled; kubeSnap is + // what the config overlay merges in. + kube *kubesource.Source + kubeSnap atomic.Pointer[kubesource.Snapshot] // managedZones is the console-managed zone directory: the only zone data // the admin API may rewrite. An empty directory is what makes the // console's Zones page read-only. @@ -167,6 +175,10 @@ func (a *App) Start(ctx context.Context) error { if err := a.startLogger(cfg); err != nil { return err } + if err := a.startKube(ctx); err != nil { + return err + } + cfg = a.cfg.Get() // the overlay may have changed it a.startBlocklist() a.startLocalZones(cfg) a.startUpstream(cfg) @@ -369,6 +381,9 @@ func (a *App) reloadLocalZones() { if a.managedZones.Exists() { paths = append(paths, a.managedZones.Dir()) } + if dir := a.kubeZoneDir(); dir != "" { + paths = append(paths, dir) + } if len(paths) == 0 { a.local.Set(nil) metrics.SetLocalRecords(0) @@ -866,7 +881,7 @@ func (a *App) Log() *slog.Logger { return a.log } // SetAdminAddr overrides the admin API listen address without touching the // config file, which is what a container platform's -admin-addr flag wants. func (a *App) SetAdminAddr(addr string) error { - next := a.cfg.Get() + next := a.cfg.Base() cp := *next cp.API = next.API cp.API.Listen = []string{addr} diff --git a/internal/app/harness_test.go b/internal/app/harness_test.go index 85a6441..2b9f979 100644 --- a/internal/app/harness_test.go +++ b/internal/app/harness_test.go @@ -89,7 +89,7 @@ type harness struct { // can make a realistic single-value edit instead of inventing a minimal stub. func (h *harness) OriginalConfig() string { return h.config } -func newHarness(t *testing.T, mutate func(*config.Config)) *harness { +func newHarness(t *testing.T, mutate func(*config.Config), opts ...func(*Options)) *harness { t.Helper() dir := t.TempDir() up := newTestUpstream(t) @@ -159,7 +159,11 @@ func newHarness(t *testing.T, mutate func(*config.Config)) *harness { t.Fatal(err) } - a, err := New(Options{ConfigPath: path, Version: "test"}) + opt := Options{ConfigPath: path, Version: "test"} + for _, o := range opts { + o(&opt) + } + a, err := New(opt) if err != nil { t.Fatalf("new app: %v", err) } diff --git a/internal/app/hooks.go b/internal/app/hooks.go index 2d84f6e..7b84f94 100644 --- a/internal/app/hooks.go +++ b/internal/app/hooks.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "net/netip" + "reflect" "sort" "strings" "sync" @@ -431,6 +432,9 @@ func (a *App) toggleBlocklist(name string, enabled bool) error { if a.blocks == nil { return fmt.Errorf("blocklists are disabled") } + if a.clusterOwned("list", name) { + return fmt.Errorf("blocklist %q is a Kubernetes BlockList; set spec.disabled on the object instead", name) + } err := a.mutateConfig(func(c *config.Config) error { found := false for i := range c.Block.Lists { @@ -459,7 +463,10 @@ func (a *App) toggleBlocklist(name string, enabled bool) error { // mutateConfig clones, mutates, validates and commits the config, returning // the validation error when the change is rejected. func (a *App) mutateConfig(fn func(*config.Config) error) error { - cur := a.cfg.Get() + // Start from the base, not the effective config: the effective one holds + // the cluster's groups and lists, and persisting those would copy them + // into the YAML file where they would shadow their own objects. + cur := a.cfg.Base() next := *cur // A shallow copy is not enough: the slices and maps we touch need their own // backing arrays, or a rejected change would corrupt the live config. @@ -767,6 +774,28 @@ func (a *App) getACL() any { // putACL validates and persists new ACL groups. func (a *App) putACL(_ context.Context, groups []config.Group, def string) (int64, error) { + // The console shows the effective list, so a save sends the cluster's + // groups back along with the file's. Unchanged ones are dropped here (the + // cluster still owns them); a changed one is refused, because the write + // would land in the YAML and silently shadow the object. + if snap := a.kubeSnap.Load(); snap != nil { + owned := make(map[string]config.Group, len(snap.Groups)) + for _, g := range snap.Groups { + owned[g.Name] = g + } + kept := groups[:0:0] + for _, g := range groups { + cg, ok := owned[g.Name] + if !ok { + kept = append(kept, g) + continue + } + if !reflect.DeepEqual(g, cg) { + return 0, fmt.Errorf("group %q is a Kubernetes ClientGroup; edit the object instead", g.Name) + } + } + groups = kept + } err := a.mutateConfig(func(c *config.Config) error { c.ACL.Groups = groups if def != "" { diff --git a/internal/app/kube.go b/internal/app/kube.go new file mode 100644 index 0000000..8eb622d --- /dev/null +++ b/internal/app/kube.go @@ -0,0 +1,200 @@ +package app + +import ( + "context" + "path/filepath" + "time" + + "k8s.io/client-go/dynamic" + + "git.k8s-git-k6n.net/mats/fancy-dns/internal/config" + "git.k8s-git-k6n.net/mats/fancy-dns/internal/kubesource" +) + +// KubeOptions turn on the Kubernetes source: ClientGroup, BlockList and +// DNSZone objects are watched and merged into the configuration. The runtime +// only reads; it works the same in a pod and on a workstation. +type KubeOptions struct { + Enabled bool + // Kubeconfig and Context pick credentials; both empty means the pod's + // service account when there is one, else the default kubeconfig. + Kubeconfig string + Context string + // Namespace restricts the watch; empty means the pod's own namespace + // in-cluster, else every namespace. + Namespace string + // SyncTimeout bounds how long startup waits for the first listing before + // falling back to the last saved snapshot. Zero means 30 seconds. + SyncTimeout time.Duration + // Client replaces the one built from the credentials; for tests. + Client dynamic.Interface +} + +const ( + kubeDegraded = "kubernetes: not synced, serving the last saved cluster configuration" + kubeRejected = "kubernetes: objects rejected" + kubeNoSnap = "kubernetes: not synced and no saved snapshot, cluster configuration is empty" + kubeInvalid = "kubernetes: the merged configuration is invalid, cluster objects are ignored" + kubeZoneFail = "kubernetes: cluster zones could not be written, serving the previous ones" + defaultKubeTO = 30 * time.Second +) + +func (a *App) kubeZoneDir() string { + if a.kube == nil { + return "" + } + return filepath.Join(a.DataDir(), "kube-zones") +} + +func (a *App) kubeSnapshotPath() string { return filepath.Join(a.DataDir(), "kube-snapshot.json") } + +// startKube connects to the cluster and installs the overlay. It runs before +// the components that read the configuration, so the first blocklist compile +// and the first zone load already include the cluster's objects. +func (a *App) startKube(ctx context.Context) error { + o := a.opt.Kube + if !o.Enabled { + return nil + } + ns := o.Namespace + if ns == "" { + ns = kubesource.Namespace() + } + client, from := o.Client, "injected client" + if client == nil { + var err error + client, from, err = kubesource.NewClient(kubesource.ClientOptions{Kubeconfig: o.Kubeconfig, Context: o.Context}) + if err != nil { + return err + } + } + src, err := kubesource.New(kubesource.Options{Client: client, Namespace: ns, Logger: a.log}) + if err != nil { + return err + } + a.kube = src + a.log.Info("watching Kubernetes for fancy-dns objects", "credentials", from, "namespace", ns) + + a.cfg.SetOverlay(a.kubeOverlay) + go func() { _ = src.Run(ctx) }() + + wait := o.SyncTimeout + if wait <= 0 { + wait = defaultKubeTO + } + syncCtx, cancel := context.WithTimeout(ctx, wait) + defer cancel() + if err := src.WaitSynced(syncCtx); err != nil { + // A cluster that cannot be reached must not stop DNS from coming up: + // resolve from what was last known and keep watching in the background. + snap, lerr := kubesource.Load(a.kubeSnapshotPath()) + switch { + case lerr != nil: + a.log.Error("saved cluster snapshot is unreadable", "error", lerr) + a.markDegraded(kubeNoSnap) + case snap == nil: + a.markDegraded(kubeNoSnap) + default: + a.kubeSnap.Store(snap) + a.markDegraded(kubeDegraded) + } + a.log.Warn("Kubernetes did not sync in time, continuing without it", "waited", wait.String(), "error", err) + a.applyKube(a.kubeSnap.Load(), false) + } else { + a.applyKube(src.Snapshot(), true) + } + // Later changes arrive here. Registered after the first apply so the + // startup path above is the only one that runs before the listeners exist. + src.OnChange(func(sn *kubesource.Snapshot) { a.applyKube(sn, true) }) + return nil +} + +// applyKube makes snap the cluster's contribution to the configuration. +func (a *App) applyKube(snap *kubesource.Snapshot, live bool) { + if snap == nil { + a.cfg.Reapply() + return + } + prev := a.kubeSnap.Load() + a.kubeSnap.Store(snap) + if live { + a.clearDegraded(kubeDegraded) + a.clearDegraded(kubeNoSnap) + if err := kubesource.Save(a.kubeSnapshotPath(), snap); err != nil { + a.log.Warn("could not save the cluster snapshot", "error", err) + } + if len(snap.Problems) > 0 { + a.markDegraded(kubeRejected) + } else { + a.clearDegraded(kubeRejected) + } + } + if _, err := kubesource.SyncZoneDir(a.kubeZoneDir(), snap.Zones); err != nil { + a.markDegraded(kubeZoneFail) + a.log.Error("could not write cluster zones, keeping the previous ones", "error", err) + } else { + a.clearDegraded(kubeZoneFail) + } + a.cfg.Reapply() + if prev == nil { + return + } + // Reapply reloaded the zones and rebuilt the ACL. Cached answers were + // produced under the old policy, so they go, as they do for any ACL, + // blocklist or zone edit. + a.invalidateBlocklistCache() + if a.blocks != nil { + go func() { + if _, err := a.blocks.RefreshAll(context.Background()); err != nil { + a.log.Warn("blocklist refresh after cluster change failed", "error", err) + } + }() + } +} + +// kubeOverlay is the Store overlay: the file's configuration plus the +// cluster's objects. A merge that does not validate is dropped whole and the +// file's configuration stays in force, because half of a policy is worse than +// the last whole one. +func (a *App) kubeOverlay(base *config.Config) *config.Config { + snap := a.kubeSnap.Load() + if snap == nil { + return base + } + merged, dropped := kubesource.Merge(base, snap) + for _, d := range dropped { + a.log.Warn("cluster object ignored", "object", d) + } + check := *merged + if err := check.Validate(); err != nil { + a.log.Error("cluster configuration does not validate, ignoring it", "error", err) + a.markDegraded(kubeInvalid) + return base + } + a.clearDegraded(kubeInvalid) + return merged +} + +// clusterOwned reports whether a group or list name comes from the cluster, +// and so cannot be edited through the console. +func (a *App) clusterOwned(kind, name string) bool { + snap := a.kubeSnap.Load() + if snap == nil { + return false + } + switch kind { + case "group": + for _, g := range snap.Groups { + if g.Name == name { + return true + } + } + case "list": + for _, l := range snap.Lists { + if l.Name == name { + return true + } + } + } + return false +} diff --git a/internal/app/kube_test.go b/internal/app/kube_test.go new file mode 100644 index 0000000..0d71eb4 --- /dev/null +++ b/internal/app/kube_test.go @@ -0,0 +1,118 @@ +package app + +import ( + "context" + "testing" + "time" + + "github.com/miekg/dns" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + dynfake "k8s.io/client-go/dynamic/fake" + + "git.k8s-git-k6n.net/mats/fancy-dns/internal/kubesource" +) + +func kubeObj(kind, name string, spec map[string]any) *unstructured.Unstructured { + return &unstructured.Unstructured{Object: map[string]any{ + "apiVersion": kubesource.APIGroup + "/" + kubesource.Version, + "kind": kind, + "metadata": map[string]any{"name": name, "namespace": "dns", "generation": int64(1)}, + "spec": spec, + }} +} + +func kubeClient() *dynfake.FakeDynamicClient { + return dynfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), + map[schema.GroupVersionResource]string{ + kubesource.ZoneGVR: "DNSZoneList", + kubesource.GroupGVR: "ClientGroupList", + kubesource.ListGVR: "BlockListList", + }) +} + +func eventually(t *testing.T, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if ok() { + return + } + time.Sleep(25 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", what) +} + +// TestKubeObjectsReachTheResolver is the point of the Kubernetes source: an +// object applied to the cluster changes what this resolver answers, with no +// reload, and deleting it takes the change back. +func TestKubeObjectsReachTheResolver(t *testing.T) { + client := kubeClient() + h := newHarness(t, nil, func(o *Options) { + o.Kube = KubeOptions{Enabled: true, Namespace: "dns", Client: client, SyncTimeout: 5 * time.Second} + }) + ctx := context.Background() + zones := client.Resource(kubesource.ZoneGVR).Namespace("dns") + groups := client.Resource(kubesource.GroupGVR).Namespace("dns") + + answer := func() string { + m := h.query("www.kube.test", dns.TypeA) + if len(m.Answer) == 1 { + if a, ok := m.Answer[0].(*dns.A); ok && m.Authoritative { + return a.A.String() + } + } + return "" + } + + if _, err := zones.Create(ctx, kubeObj("DNSZone", "kube", map[string]any{ + "zone": "kube.test", + "records": []any{ + map[string]any{"name": "@", "type": "NS", "value": "ns1.kube.test."}, + map[string]any{"name": "ns1", "type": "A", "value": "10.1.2.1"}, + map[string]any{"name": "www", "type": "A", "value": "10.1.2.3"}, + }, + }), metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + eventually(t, "the DNSZone to be served", func() bool { return answer() == "10.1.2.3" }) + + // The cluster's data never lands in the YAML the console writes back. + if got := len(h.app.cfg.Base().ACL.Groups); got != 0 { + t.Fatalf("base has %d groups", got) + } + + if _, err := groups.Create(ctx, kubeObj("ClientGroup", "no-loopback", map[string]any{ + "policy": "deny", "match": map[string]any{"cidrs": []any{"127.0.0.0/8"}}, + }), metav1.CreateOptions{}); err != nil { + t.Fatal(err) + } + eventually(t, "the ClientGroup to refuse the client", func() bool { + return h.query("www.example.test", dns.TypeA).Rcode == dns.RcodeRefused + }) + if len(h.app.cfg.Base().ACL.Groups) != 0 || len(h.app.cfg.Get().ACL.Groups) != 1 { + t.Fatalf("base=%d effective=%d groups", len(h.app.cfg.Base().ACL.Groups), len(h.app.cfg.Get().ACL.Groups)) + } + + // The console cannot overwrite a cluster-owned group. + if _, err := h.app.putACL(ctx, nil, ""); err != nil { + t.Fatalf("saving the console's own (empty) list: %v", err) + } + if len(h.app.cfg.Get().ACL.Groups) != 1 { + t.Fatal("saving the file's groups removed the cluster's") + } + + if err := groups.Delete(ctx, "no-loopback", metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + eventually(t, "the deleted ClientGroup to stop applying", func() bool { + return h.query("www.example.test", dns.TypeA).Rcode == dns.RcodeSuccess + }) + + if err := zones.Delete(ctx, "kube", metav1.DeleteOptions{}); err != nil { + t.Fatal(err) + } + eventually(t, "the deleted DNSZone to stop being authoritative", func() bool { return answer() == "" }) +} diff --git a/internal/config/store.go b/internal/config/store.go index b7a9e38..4d27459 100644 --- a/internal/config/store.go +++ b/internal/config/store.go @@ -10,7 +10,13 @@ import ( // in atomically, so a request never observes a half-applied config. type Store struct { cur atomic.Pointer[Config] - mu sync.Mutex // serialises writers so LastError/Version stay coherent + // base is the configuration as the file (or an API write) defined it. cur + // is base with the overlay applied; without an overlay the two are the + // same pointer. Writers that persist back to the file must start from + // base, or they would write overlay content into the file. + base atomic.Pointer[Config] + overlay atomic.Pointer[func(*Config) *Config] + mu sync.Mutex // serialises writers so LastError/Version stay coherent // path is the file the config was loaded from, used by the API to // persist mutations. path string @@ -28,9 +34,50 @@ type Store struct { func NewStore(cfg *Config, path string) *Store { s := &Store{path: path} s.cur.Store(cfg) + s.base.Store(cfg) return s } +// Base returns the configuration without the overlay: what the file says. +func (s *Store) Base() *Config { + if b := s.base.Load(); b != nil { + return b + } + return s.cur.Load() +} + +// SetOverlay installs fn, which derives the effective configuration from the +// base (a cluster source adding groups and lists, say). fn must not modify its +// argument. Call Reapply afterwards, and again whenever fn's inputs change. +func (s *Store) SetOverlay(fn func(*Config) *Config) { + if fn == nil { + s.overlay.Store(nil) + return + } + s.overlay.Store(&fn) +} + +func (s *Store) effective(base *Config) *Config { + if fn := s.overlay.Load(); fn != nil { + if out := (*fn)(base); out != nil { + return out + } + } + return base +} + +// Reapply recomputes the effective configuration from the base and swaps it +// in, running the hooks as for any other swap. +func (s *Store) Reapply() { + s.mu.Lock() + next := s.effective(s.Base()) + s.cur.Store(next) + s.generation.Add(1) + s.lastErr.Store(nil) + s.mu.Unlock() + s.runHooks(next) +} + // Get returns the current configuration snapshot. func (s *Store) Get() *Config { return s.cur.Load() } @@ -60,11 +107,13 @@ func (s *Store) Swap(cfg *Config) error { cfg.Cache.Shards = 32 } s.mu.Lock() - s.cur.Store(cfg) + s.base.Store(cfg) + eff := s.effective(cfg) + s.cur.Store(eff) s.generation.Add(1) s.lastErr.Store(nil) s.mu.Unlock() - s.runHooks(cfg) + s.runHooks(eff) return nil } @@ -84,11 +133,13 @@ func (s *Store) Reload() error { } // Preserve the working dir of the running instance so relative paths keep // resolving the same way even if the file moved. - cfg.Server.WorkingDir = s.cur.Load().Server.WorkingDir - s.cur.Store(cfg) + cfg.Server.WorkingDir = s.Base().Server.WorkingDir + s.base.Store(cfg) + eff := s.effective(cfg) + s.cur.Store(eff) s.generation.Add(1) s.lastErr.Store(nil) - s.runHooks(cfg) + s.runHooks(eff) return nil } diff --git a/internal/config/store_overlay_test.go b/internal/config/store_overlay_test.go new file mode 100644 index 0000000..9860cf2 --- /dev/null +++ b/internal/config/store_overlay_test.go @@ -0,0 +1,46 @@ +package config + +import "testing" + +func TestStoreOverlayKeepsBaseSeparate(t *testing.T) { + base := Default() + s := NewStore(base, "") + if s.Base() != base || s.Get() != base { + t.Fatal("without an overlay base and current are the same config") + } + + extra := Group{Name: "from-cluster", Policy: "deny"} + s.SetOverlay(func(b *Config) *Config { + next := *b + next.ACL.Groups = append(append([]Group(nil), b.ACL.Groups...), extra) + return &next + }) + var seen *Config + s.OnSwap(func(c *Config) { seen = c }) + gen := s.Generation() + s.Reapply() + + if got := s.Get().ACL.Groups; len(got) != 1 || got[0].Name != "from-cluster" { + t.Fatalf("effective groups = %+v", got) + } + if len(s.Base().ACL.Groups) != 0 { + t.Fatal("the overlay leaked into the base") + } + if seen != s.Get() || s.Generation() != gen+1 { + t.Fatal("Reapply must run the swap hooks and bump the generation") + } + + // A write (the API persisting a change) replaces the base; the overlay + // is applied on top of it again. + written := Default() + written.ACL.Groups = []Group{{Name: "yaml", Policy: "allow"}} + if err := s.Swap(written); err != nil { + t.Fatal(err) + } + if got := s.Get().ACL.Groups; len(got) != 2 || got[0].Name != "yaml" || got[1].Name != "from-cluster" { + t.Fatalf("after Swap = %+v", got) + } + if got := s.Base().ACL.Groups; len(got) != 1 { + t.Fatalf("base after Swap = %+v", got) + } +} diff --git a/internal/kubesource/client.go b/internal/kubesource/client.go new file mode 100644 index 0000000..5954c9e --- /dev/null +++ b/internal/kubesource/client.go @@ -0,0 +1,66 @@ +package kubesource + +import ( + "fmt" + "os" + + "k8s.io/client-go/dynamic" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" +) + +// ClientOptions say where the cluster is. +type ClientOptions struct { + // Kubeconfig is a kubeconfig path. Empty means: the in-cluster service + // account if there is one, else the standard loading rules ($KUBECONFIG, + // then ~/.kube/config). The same binary therefore runs in a pod and on a + // workstation with no flag at all. + Kubeconfig string + // Context selects a kubeconfig context; empty is the file's current one. + Context string +} + +// Namespace returns the namespace to watch when none was configured: the +// pod's own when running in-cluster, else "" (every namespace the credentials +// can list). +func Namespace() string { + b, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/namespace") + if err != nil { + return "" + } + return string(b) +} + +// NewClient builds a dynamic client and reports which source it used. +func NewClient(o ClientOptions) (dynamic.Interface, string, error) { + cfg, from, err := restConfig(o) + if err != nil { + return nil, "", err + } + // The runtime lists and watches a handful of resources; the defaults + // (5 QPS) are fine for that, but a hung apiserver must not hang us. + cfg.UserAgent = "fancy-dns" + c, err := dynamic.NewForConfig(cfg) + if err != nil { + return nil, "", fmt.Errorf("kubernetes client: %w", err) + } + return c, from, nil +} + +func restConfig(o ClientOptions) (*rest.Config, string, error) { + if o.Kubeconfig == "" && o.Context == "" { + if cfg, err := rest.InClusterConfig(); err == nil { + return cfg, "in-cluster service account", nil + } + } + rules := clientcmd.NewDefaultClientConfigLoadingRules() + if o.Kubeconfig != "" { + rules.ExplicitPath = o.Kubeconfig + } + cfg, err := clientcmd.NewNonInteractiveDeferredLoadingClientConfig( + rules, &clientcmd.ConfigOverrides{CurrentContext: o.Context}).ClientConfig() + if err != nil { + return nil, "", fmt.Errorf("no usable Kubernetes credentials (in-cluster or kubeconfig): %w", err) + } + return cfg, "kubeconfig", nil +} diff --git a/internal/kubesource/convert.go b/internal/kubesource/convert.go new file mode 100644 index 0000000..0d1c257 --- /dev/null +++ b/internal/kubesource/convert.go @@ -0,0 +1,191 @@ +package kubesource + +import ( + "fmt" + "net" + "strings" + "time" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + + "git.k8s-git-k6n.net/mats/fancy-dns/internal/config" + "git.k8s-git-k6n.net/mats/fancy-dns/internal/localzone" +) + +// maxZoneRecords bounds one DNSZone. An object is capped near 1.5 MiB by etcd +// anyway; this keeps a hostile or runaway one from costing a render. +const maxZoneRecords = 5000 + +// group is a converted ClientGroup with its ordering key. +type group struct { + cfg config.Group + priority int32 +} + +// zone is a converted DNSZone: the apex and the rendered master file. +type zone struct { + apex string + text string +} + +func spec(u *unstructured.Unstructured, into any) error { + raw, ok := u.Object["spec"].(map[string]any) + if !ok { + return fmt.Errorf("the object has no spec") + } + return runtime.DefaultUnstructuredConverter.FromUnstructured(raw, into) +} + +func convertGroup(u *unstructured.Unstructured) (group, error) { + var s ClientGroupSpec + if err := spec(u, &s); err != nil { + return group{}, err + } + g := config.Group{ + Name: u.GetName(), + Comment: s.Comment, + Match: s.Match, + Policy: s.Policy, + AllowDomains: s.AllowDomains, + DenyDomains: s.DenyDomains, + SkipBlocklist: s.SkipBlocklist, + SkipCache: s.SkipCache, + RateLimit: s.RateLimit, + MaxConcurrent: s.MaxConcurrent, + LogQueries: s.LogQueries, + Upstreams: s.Upstreams, + } + switch g.Policy { + case "allow", "deny", "block_only": + default: + return group{}, fmt.Errorf("policy must be allow, deny or block_only (got %q)", g.Policy) + } + for _, cidr := range g.Match.CIDRs { + if _, _, err := net.ParseCIDR(cidr); err != nil { + return group{}, fmt.Errorf("bad CIDR %q", cidr) + } + } + // Copy before normalising: the slice belongs to the decoded object. + g.Match.PTRSuffixes = append([]string(nil), g.Match.PTRSuffixes...) + for i, s := range g.Match.PTRSuffixes { + if !strings.HasSuffix(s, ".") { + g.Match.PTRSuffixes[i] = s + "." + } + } + if !g.Match.Any && len(g.Match.CIDRs) == 0 && len(g.Match.PTRSuffixes) == 0 && len(g.Match.Tags) == 0 { + return group{}, fmt.Errorf("match selects no clients: set cidrs, ptr_suffixes, tags or any") + } + if g.Match.Any && len(g.Match.CIDRs) == 0 && len(g.Match.PTRSuffixes) == 0 && len(g.Match.Tags) == 0 { + // config.Validate refuses this shape for every group. Refusing it here + // rejects one object instead of the whole merged configuration. + return group{}, fmt.Errorf("match.any alone is not accepted; use a catch-all in the YAML acl.groups") + } + if g.RateLimit < 0 || g.MaxConcurrent < 0 { + return group{}, fmt.Errorf("rate_limit and max_concurrent must not be negative") + } + return group{cfg: g, priority: s.Priority}, nil +} + +func convertList(u *unstructured.Unstructured) (config.ListSource, error) { + var s BlockListSpec + if err := spec(u, &s); err != nil { + return config.ListSource{}, err + } + l := config.ListSource{ + Name: u.GetName(), + Format: s.Format, + URL: s.URL, + Mirror: s.Mirror, + Enabled: !s.Disabled, + MaxBytes: s.MaxBytes, + Categories: s.Categories, + Comment: s.Comment, + InsecureSkipTLSVerify: s.InsecureSkipTLSVerify, + } + if !isHTTP(l.URL) { + return config.ListSource{}, fmt.Errorf("url must be http(s) (got %q)", l.URL) + } + for _, m := range l.Mirror { + if !isHTTP(m) { + return config.ListSource{}, fmt.Errorf("mirror must be http(s) (got %q)", m) + } + } + switch l.Format { + case "", "auto", "hosts", "adblock", "plain", "rpz": + default: + return config.ListSource{}, fmt.Errorf("unknown format %q", l.Format) + } + var err error + if l.Refresh, err = duration(s.Refresh); err != nil { + return config.ListSource{}, fmt.Errorf("refresh: %w", err) + } + if l.Timeout, err = duration(s.Timeout); err != nil { + return config.ListSource{}, fmt.Errorf("timeout: %w", err) + } + if l.MaxBytes < 0 { + return config.ListSource{}, fmt.Errorf("max_bytes must not be negative") + } + return l, nil +} + +func isHTTP(u string) bool { + return strings.HasPrefix(u, "https://") || strings.HasPrefix(u, "http://") +} + +func duration(s string) (config.Duration, error) { + if s == "" { + return 0, nil + } + d, err := time.ParseDuration(s) + if err != nil { + return 0, err + } + if d < 0 { + return 0, fmt.Errorf("must not be negative") + } + return config.Duration(d), nil +} + +func convertZone(u *unstructured.Unstructured) (zone, error) { + var s DNSZoneSpec + if err := spec(u, &s); err != nil { + return zone{}, err + } + if s.Zone == "" { + return zone{}, fmt.Errorf("spec.zone is required") + } + if len(s.Records) > maxZoneRecords { + return zone{}, fmt.Errorf("%d records exceeds the limit of %d", len(s.Records), maxZoneRecords) + } + apex, err := localzone.NormalizeOrigin(s.Zone) + if err != nil { + return zone{}, fmt.Errorf("spec.zone: %w", err) + } + var ns []string + recs := make([]localzone.Record, 0, len(s.Records)) + for _, r := range s.Records { + name := r.Name + if name == "@" { + name = apex + } + recs = append(recs, localzone.Record{Name: name, Type: r.Type, TTL: r.TTL, Value: r.Value}) + if strings.EqualFold(r.Type, "NS") && name == apex { + ns = append(ns, r.Value) + } + } + if s.SOA == "" && len(ns) == 0 { + return zone{}, fmt.Errorf("a zone needs an apex NS record (name \"@\") or an explicit soa") + } + // The generation changes exactly when the spec does, so every replica + // renders the same bytes for the same object version. + serial := uint32(u.GetGeneration()) + if serial == 0 { + serial = 1 + } + text, err := localzone.RenderZone(apex, localzone.ZoneDefaults{TTL: s.TTL, Soa: s.SOA}, recs, serial) + if err != nil { + return zone{}, err + } + return zone{apex: apex, text: text}, nil +} diff --git a/internal/kubesource/helpers_test.go b/internal/kubesource/helpers_test.go new file mode 100644 index 0000000..41c11d6 --- /dev/null +++ b/internal/kubesource/helpers_test.go @@ -0,0 +1,44 @@ +package kubesource + +import ( + "os" + "path/filepath" + "sort" + "strings" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" +) + +type schemaGVR = schema.GroupVersionResource + +var ( + metav1UpdateOptions = metav1.UpdateOptions{} + metav1DeleteOptions = metav1.DeleteOptions{} +) + +func metav1Time(sec int64) metav1.Time { return metav1.NewTime(time.Unix(sec, 0)) } + +func writeTemp(t *testing.T, z zone) []string { + t.Helper() + dir := t.TempDir() + if _, err := SyncZoneDir(dir, map[string]string{z.apex: z.text}); err != nil { + t.Fatal(err) + } + return []string{dir} +} + +func fileNames(t *testing.T, dir string) string { + t.Helper() + m, _ := filepath.Glob(filepath.Join(dir, "*")) + var out []string + for _, p := range m { + if _, err := os.Stat(p); err == nil { + out = append(out, filepath.Base(p)) + } + } + sort.Strings(out) + return strings.Join(out, ",") +} diff --git a/internal/kubesource/manifests_test.go b/internal/kubesource/manifests_test.go new file mode 100644 index 0000000..666e234 --- /dev/null +++ b/internal/kubesource/manifests_test.go @@ -0,0 +1,117 @@ +package kubesource + +import ( + "bytes" + "io" + "os" + "reflect" + "sort" + "strings" + "testing" + + "gopkg.in/yaml.v3" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" +) + +func yamlDocs(t *testing.T, path string) []map[string]any { + t.Helper() + b, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + var out []map[string]any + dec := yaml.NewDecoder(bytes.NewReader(b)) + for { + var m map[string]any + if err := dec.Decode(&m); err == io.EOF { + break + } else if err != nil { + t.Fatalf("%s: %v", path, err) + } + if m != nil { + out = append(out, m) + } + } + return out +} + +func jsonFields(typ reflect.Type) []string { + var out []string + for i := 0; i < typ.NumField(); i++ { + tag := strings.Split(typ.Field(i).Tag.Get("json"), ",")[0] + if tag != "" && tag != "-" { + out = append(out, tag) + } + } + sort.Strings(out) + return out +} + +// TestCRDSchemasMatchTheGoSpecs keeps deploy/k8s/crds/crds.yaml and the spec +// structs from drifting: a field the schema lacks is pruned by the apiserver +// before the runtime ever sees it, and a schema field the struct lacks is +// accepted and silently ignored. +func TestCRDSchemasMatchTheGoSpecs(t *testing.T) { + want := map[string]reflect.Type{ + "DNSZone": reflect.TypeOf(DNSZoneSpec{}), + "ClientGroup": reflect.TypeOf(ClientGroupSpec{}), + "BlockList": reflect.TypeOf(BlockListSpec{}), + } + seen := 0 + for _, doc := range yamlDocs(t, "../../deploy/k8s/crds/crds.yaml") { + names := doc["spec"].(map[string]any)["names"].(map[string]any) + kind := names["kind"].(string) + typ, ok := want[kind] + if !ok { + t.Errorf("unexpected CRD kind %s", kind) + continue + } + seen++ + versions := doc["spec"].(map[string]any)["versions"].([]any) + schema := versions[0].(map[string]any)["schema"].(map[string]any)["openAPIV3Schema"].(map[string]any) + props := schema["properties"].(map[string]any)["spec"].(map[string]any)["properties"].(map[string]any) + var got []string + for k := range props { + got = append(got, k) + } + sort.Strings(got) + if !reflect.DeepEqual(got, jsonFields(typ)) { + t.Errorf("%s: schema fields %v, Go fields %v", kind, got, jsonFields(typ)) + } + if plural := names["plural"].(string); plural != map[string]string{ + "DNSZone": ZoneGVR.Resource, "ClientGroup": GroupGVR.Resource, "BlockList": ListGVR.Resource, + }[kind] { + t.Errorf("%s: plural %q does not match the GVR the runtime watches", kind, plural) + } + } + if seen != len(want) { + t.Errorf("saw %d CRDs, want %d", seen, len(want)) + } +} + +// TestExampleObjectsConvert makes sure the shipped examples are accepted by +// the runtime, so the first thing an operator copies is not a rejection. +func TestExampleObjectsConvert(t *testing.T) { + n := 0 + for _, doc := range yamlDocs(t, "../../deploy/k8s/crds/examples.yaml") { + u := &unstructured.Unstructured{Object: doc} + var err error + switch u.GetKind() { + case "DNSZone": + _, err = convertZone(u) + case "ClientGroup": + _, err = convertGroup(u) + case "BlockList": + _, err = convertList(u) + default: + t.Fatalf("unknown kind %s", u.GetKind()) + } + if err != nil { + t.Errorf("%s %s: %v", u.GetKind(), u.GetName(), err) + } + n++ + } + if n != 3 { + t.Fatalf("converted %d examples, want 3", n) + } +} diff --git a/internal/kubesource/merge.go b/internal/kubesource/merge.go new file mode 100644 index 0000000..53fdfdb --- /dev/null +++ b/internal/kubesource/merge.go @@ -0,0 +1,183 @@ +package kubesource + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" + + "git.k8s-git-k6n.net/mats/fancy-dns/internal/config" +) + +func signature(s *Snapshot) string { + b, _ := json.Marshal(struct { + G []config.Group + L []config.ListSource + Z map[string]string + }{s.Groups, s.Lists, s.Zones}) + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +// Merge returns base with the cluster's groups and lists added; base is not +// modified. Precedence is deliberate: whatever the YAML declares is the +// operator's and wins. YAML groups are evaluated before cluster groups, and a +// cluster list that reuses a YAML list's name is dropped. The names of what was +// dropped are returned so the caller can say so. +func Merge(base *config.Config, snap *Snapshot) (*config.Config, []string) { + if snap == nil { + return base, nil + } + next := *base + next.ACL = base.ACL + next.ACL.Groups = append([]config.Group(nil), base.ACL.Groups...) + next.Block = base.Block + next.Block.Lists = append([]config.ListSource(nil), base.Block.Lists...) + + var dropped []string + groups := map[string]bool{} + for _, g := range base.ACL.Groups { + groups[g.Name] = true + } + for _, g := range snap.Groups { + if groups[g.Name] { + dropped = append(dropped, "ClientGroup "+g.Name+" (name used in the YAML)") + continue + } + next.ACL.Groups = append(next.ACL.Groups, g) + } + lists := map[string]bool{} + for _, l := range base.Block.Lists { + lists[l.Name] = true + } + for _, l := range snap.Lists { + if lists[l.Name] { + dropped = append(dropped, "BlockList "+l.Name+" (name used in the YAML)") + continue + } + // The same defaults config.applyDefaults gives a YAML list. + if l.Format == "" { + l.Format = "auto" + } + if l.Refresh == 0 { + l.Refresh = config.Duration(24 * time.Hour) + } + if l.Timeout <= 0 { + l.Timeout = base.Server.Timeouts.List + } + if l.MaxBytes <= 0 { + l.MaxBytes = 512 << 20 + } + next.Block.Lists = append(next.Block.Lists, l) + } + return &next, dropped +} + +// SyncZoneDir makes dir hold exactly one .zone file per zone, written +// atomically, and removes the files of zones that no longer exist. It reports +// whether anything on disk changed. Files it did not write (anything that is +// not *.zone) are left alone. +func SyncZoneDir(dir string, zones map[string]string) (bool, error) { + if err := os.MkdirAll(dir, 0o755); err != nil { + return false, fmt.Errorf("zone directory %s: %w", dir, err) + } + changed := false + want := map[string]bool{} + for apex, text := range zones { + name := zoneFile(apex) + want[name] = true + path := filepath.Join(dir, name) + if old, err := os.ReadFile(path); err == nil && string(old) == text { + continue + } + tmp, err := os.CreateTemp(dir, ".kube-zone-*") + if err != nil { + return changed, err + } + _, werr := tmp.WriteString(text) + cerr := tmp.Close() + if werr != nil || cerr != nil { + _ = os.Remove(tmp.Name()) + return changed, fmt.Errorf("write %s: %w", name, firstErr(werr, cerr)) + } + if err := os.Chmod(tmp.Name(), 0o644); err != nil { + _ = os.Remove(tmp.Name()) + return changed, err + } + if err := os.Rename(tmp.Name(), path); err != nil { + _ = os.Remove(tmp.Name()) + return changed, err + } + changed = true + } + entries, err := os.ReadDir(dir) + if err != nil { + return changed, err + } + for _, e := range entries { + if e.IsDir() || !strings.HasSuffix(e.Name(), ".zone") || want[e.Name()] { + continue + } + if err := os.Remove(filepath.Join(dir, e.Name())); err != nil { + return changed, err + } + changed = true + } + return changed, nil +} + +// zoneFile is the file name for an apex. The apex has been normalised by +// NormalizeName, so it holds only DNS name characters; the wildcard label is +// the one that is not file-safe on every platform. +func zoneFile(apex string) string { + n := strings.TrimSuffix(apex, ".") + if n == "" { + n = "root" + } + return strings.ReplaceAll(n, "*", "_wild") + ".zone" +} + +func firstErr(errs ...error) error { + for _, e := range errs { + if e != nil { + return e + } + } + return nil +} + +// Save writes the snapshot so a restart that cannot reach the apiserver still +// has the last known policy. Problems are not saved. +func Save(path string, s *Snapshot) error { + b, err := json.Marshal(s) + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return err + } + return config.WriteFileAtomic(path, b, 0o644) +} + +// Load reads a snapshot written by Save. A missing file is (nil, nil). +func Load(path string) (*Snapshot, error) { + b, err := os.ReadFile(path) + if os.IsNotExist(err) { + return nil, nil + } + if err != nil { + return nil, err + } + var s Snapshot + if err := json.Unmarshal(b, &s); err != nil { + return nil, fmt.Errorf("%s: %w", path, err) + } + if s.Zones == nil { + s.Zones = map[string]string{} + } + return &s, nil +} diff --git a/internal/kubesource/source.go b/internal/kubesource/source.go new file mode 100644 index 0000000..6c57b80 --- /dev/null +++ b/internal/kubesource/source.go @@ -0,0 +1,327 @@ +package kubesource + +import ( + "context" + "errors" + "fmt" + "log/slog" + "sort" + "sync" + "sync/atomic" + "time" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/dynamic/dynamicinformer" + "k8s.io/client-go/tools/cache" + + "git.k8s-git-k6n.net/mats/fancy-dns/internal/config" +) + +// Problem is one object the source could not use. +type Problem struct { + Kind string `json:"kind"` + Namespace string `json:"namespace"` + Name string `json:"name"` + Message string `json:"message"` + // KeptLastGood is true when an earlier valid version of the object is + // still in force. + KeptLastGood bool `json:"kept_last_good"` +} + +// Snapshot is everything the cluster currently declares, already converted and +// validated per object. It is immutable once published. +type Snapshot struct { + // Revision increases with every rebuild that changed content. + Revision uint64 `json:"revision"` + // Groups are in evaluation order. + Groups []config.Group `json:"groups"` + Lists []config.ListSource `json:"lists"` + // Zones maps an apex to its rendered master file. + Zones map[string]string `json:"zones"` + Problems []Problem `json:"problems,omitempty"` +} + +// Options configure a Source. +type Options struct { + // Client is the dynamic client; build one with NewClient. + Client dynamic.Interface + // Namespace restricts the watch; "" watches every namespace. + Namespace string + // Debounce coalesces bursts of events into one rebuild. + Debounce time.Duration + Resync time.Duration + Logger *slog.Logger +} + +// Source watches the custom resources and publishes Snapshots. +type Source struct { + opt Options + log *slog.Logger + cur atomic.Pointer[Snapshot] + trigger chan struct{} + synced chan struct{} + syncOnce sync.Once + + mu sync.Mutex + onChange []func(*Snapshot) + // good holds the last valid conversion of each object, so one edit that + // breaks an object does not remove its policy from a running resolver. + good map[string]any + sig string + rev uint64 +} + +// New returns a Source. Nothing runs until Run. +func New(opt Options) (*Source, error) { + if opt.Client == nil { + return nil, errors.New("kubesource: a client is required") + } + if opt.Debounce <= 0 { + opt.Debounce = 250 * time.Millisecond + } + if opt.Resync <= 0 { + opt.Resync = 10 * time.Minute + } + if opt.Logger == nil { + opt.Logger = slog.Default() + } + s := &Source{ + opt: opt, + log: opt.Logger.With("component", "kubesource"), + trigger: make(chan struct{}, 1), + synced: make(chan struct{}), + good: map[string]any{}, + } + return s, nil +} + +// Snapshot is the latest published snapshot, or nil before the first sync. +func (s *Source) Snapshot() *Snapshot { return s.cur.Load() } + +// OnChange registers fn to run, in order, after each snapshot that differs +// from the previous one. +func (s *Source) OnChange(fn func(*Snapshot)) { + s.mu.Lock() + s.onChange = append(s.onChange, fn) + s.mu.Unlock() +} + +// WaitSynced blocks until every informer has listed once, or ctx ends. A +// cluster without the CRDs installed never syncs, so callers bound the wait. +func (s *Source) WaitSynced(ctx context.Context) error { + select { + case <-s.synced: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +// Run starts the informers and rebuilds snapshots until ctx ends. +func (s *Source) Run(ctx context.Context) error { + factory := dynamicinformer.NewFilteredDynamicSharedInformerFactory(s.opt.Client, s.opt.Resync, s.opt.Namespace, nil) + informers := make(map[schema.GroupVersionResource]cache.SharedIndexInformer, len(allGVRs)) + handler := cache.ResourceEventHandlerFuncs{ + AddFunc: func(any) { s.poke() }, + UpdateFunc: func(_, _ any) { s.poke() }, + DeleteFunc: func(any) { s.poke() }, + } + var syncs []cache.InformerSynced + for _, gvr := range allGVRs { + inf := factory.ForResource(gvr).Informer() + if _, err := inf.AddEventHandler(handler); err != nil { + return fmt.Errorf("watch %s: %w", gvr.Resource, err) + } + _ = inf.SetWatchErrorHandler(func(_ *cache.Reflector, err error) { + s.log.Warn("watch failed, will retry", "resource", gvr.Resource, "error", err) + }) + informers[gvr] = inf + syncs = append(syncs, inf.HasSynced) + } + factory.Start(ctx.Done()) + + go func() { + if cache.WaitForCacheSync(ctx.Done(), syncs...) { + s.rebuild(informers) + s.syncOnce.Do(func() { close(s.synced) }) + } + }() + + for { + select { + case <-ctx.Done(): + factory.Shutdown() + return nil + case <-s.trigger: + } + // Debounce: let a burst (a kubectl apply of a directory) land first. + t := time.NewTimer(s.opt.Debounce) + select { + case <-ctx.Done(): + t.Stop() + factory.Shutdown() + return nil + case <-t.C: + } + select { + case <-s.synced: + s.rebuild(informers) + default: + // Not synced yet: the sync goroutine does the first rebuild, and + // a partial list must not be published as the cluster's state. + } + } +} + +func (s *Source) poke() { + select { + case s.trigger <- struct{}{}: + default: + } +} + +func (s *Source) objects(inf cache.SharedIndexInformer) []*unstructured.Unstructured { + raw := inf.GetStore().List() + out := make([]*unstructured.Unstructured, 0, len(raw)) + for _, o := range raw { + if u, ok := o.(*unstructured.Unstructured); ok && u.GetDeletionTimestamp() == nil { + out = append(out, u) + } + } + // A stable order makes duplicate resolution and the snapshot signature + // deterministic regardless of the informer's map iteration. + sort.Slice(out, func(i, j int) bool { + a, b := out[i], out[j] + if at, bt := a.GetCreationTimestamp(), b.GetCreationTimestamp(); !at.Equal(&bt) { + return at.Before(&bt) + } + if a.GetNamespace() != b.GetNamespace() { + return a.GetNamespace() < b.GetNamespace() + } + return a.GetName() < b.GetName() + }) + return out +} + +func (s *Source) rebuild(informers map[schema.GroupVersionResource]cache.SharedIndexInformer) { + s.mu.Lock() + snap := s.build( + s.objects(informers[GroupGVR]), + s.objects(informers[ListGVR]), + s.objects(informers[ZoneGVR]), + ) + sig := signature(snap) + changed := sig != s.sig + if changed { + s.sig = sig + s.rev++ + snap.Revision = s.rev + s.cur.Store(snap) + } else if prev := s.cur.Load(); prev != nil { + // Content is the same; problems may still have changed, so refresh + // them without notifying anyone. + snap.Revision = prev.Revision + s.cur.Store(snap) + } + hooks := append([]func(*Snapshot){}, s.onChange...) + s.mu.Unlock() + if !changed { + return + } + s.log.Info("cluster configuration changed", "revision", snap.Revision, + "groups", len(snap.Groups), "lists", len(snap.Lists), "zones", len(snap.Zones), + "rejected", len(snap.Problems)) + for _, fn := range hooks { + fn(snap) + } +} + +// build converts every object. Caller holds s.mu. +func (s *Source) build(groups, lists, zones []*unstructured.Unstructured) *Snapshot { + snap := &Snapshot{Zones: map[string]string{}} + live := map[string]bool{} + + problem := func(kind string, u *unstructured.Unstructured, msg string, kept bool) { + snap.Problems = append(snap.Problems, Problem{ + Kind: kind, Namespace: u.GetNamespace(), Name: u.GetName(), Message: msg, KeptLastGood: kept, + }) + s.log.Warn("rejected object", "kind", kind, "namespace", u.GetNamespace(), "name", u.GetName(), + "error", msg, "kept_last_good", kept) + } + // resolve converts one object, falling back to its last good version. + resolve := func(kind string, u *unstructured.Unstructured, conv func() (any, error)) (any, bool) { + key := kind + "/" + u.GetNamespace() + "/" + u.GetName() + live[key] = true + v, err := conv() + if err == nil { + s.good[key] = v + return v, true + } + old, ok := s.good[key] + problem(kind, u, err.Error(), ok) + return old, ok + } + + seenGroup := map[string]bool{} + var gs []group + for _, u := range groups { + v, ok := resolve("ClientGroup", u, func() (any, error) { return convertGroup(u) }) + if !ok { + continue + } + g := v.(group) + if seenGroup[g.cfg.Name] { + problem("ClientGroup", u, "another ClientGroup already uses this name", false) + continue + } + seenGroup[g.cfg.Name] = true + gs = append(gs, g) + } + sort.SliceStable(gs, func(i, j int) bool { + if gs[i].priority != gs[j].priority { + return gs[i].priority < gs[j].priority + } + return gs[i].cfg.Name < gs[j].cfg.Name + }) + for _, g := range gs { + snap.Groups = append(snap.Groups, g.cfg) + } + + seenList := map[string]bool{} + for _, u := range lists { + v, ok := resolve("BlockList", u, func() (any, error) { return convertList(u) }) + if !ok { + continue + } + l := v.(config.ListSource) + if seenList[l.Name] { + problem("BlockList", u, "another BlockList already uses this name", false) + continue + } + seenList[l.Name] = true + snap.Lists = append(snap.Lists, l) + } + + for _, u := range zones { + v, ok := resolve("DNSZone", u, func() (any, error) { return convertZone(u) }) + if !ok { + continue + } + z := v.(zone) + if _, dup := snap.Zones[z.apex]; dup { + // Objects are ordered oldest first, so the earliest claim wins. + problem("DNSZone", u, "another DNSZone already declares "+z.apex, false) + continue + } + snap.Zones[z.apex] = z.text + } + + for key := range s.good { + if !live[key] { + delete(s.good, key) + } + } + return snap +} diff --git a/internal/kubesource/source_test.go b/internal/kubesource/source_test.go new file mode 100644 index 0000000..467c9a0 --- /dev/null +++ b/internal/kubesource/source_test.go @@ -0,0 +1,296 @@ +package kubesource + +import ( + "context" + "strings" + "testing" + "time" + + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime" + dynfake "k8s.io/client-go/dynamic/fake" + + "git.k8s-git-k6n.net/mats/fancy-dns/internal/config" + "git.k8s-git-k6n.net/mats/fancy-dns/internal/localzone" + "git.k8s-git-k6n.net/mats/fancy-dns/internal/zonefile" +) + +func obj(kind, name string, spec map[string]any) *unstructured.Unstructured { + return &unstructured.Unstructured{Object: map[string]any{ + "apiVersion": APIGroup + "/" + Version, + "kind": kind, + "metadata": map[string]any{"name": name, "namespace": "dns", "generation": int64(3)}, + "spec": spec, + }} +} + +func cgroup(name string, spec map[string]any) *unstructured.Unstructured { + return obj("ClientGroup", name, spec) +} + +func fakeClient(objs ...runtime.Object) *dynfake.FakeDynamicClient { + return dynfake.NewSimpleDynamicClientWithCustomListKinds(runtime.NewScheme(), + map[schemaGVR]string{ + ZoneGVR: "DNSZoneList", + GroupGVR: "ClientGroupList", + ListGVR: "BlockListList", + }, objs...) +} + +func start(t *testing.T, objs ...runtime.Object) (*Source, *dynfake.FakeDynamicClient, chan *Snapshot) { + t.Helper() + c := fakeClient(objs...) + s, err := New(Options{Client: c, Namespace: "dns", Debounce: 10 * time.Millisecond}) + if err != nil { + t.Fatal(err) + } + ch := make(chan *Snapshot, 16) + s.OnChange(func(sn *Snapshot) { ch <- sn }) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { _ = s.Run(ctx); close(done) }() + t.Cleanup(func() { cancel(); <-done }) + return s, c, ch +} + +func next(t *testing.T, ch chan *Snapshot) *Snapshot { + t.Helper() + select { + case sn := <-ch: + return sn + case <-time.After(5 * time.Second): + t.Fatal("no snapshot published") + return nil + } +} + +var cidrMatch = map[string]any{"cidrs": []any{"10.0.0.0/8"}} + +func TestGroupsAreOrderedByPriorityThenName(t *testing.T) { + _, _, ch := start(t, + cgroup("b-kids", map[string]any{"priority": int64(10), "policy": "block_only", "match": cidrMatch}), + cgroup("a-kids", map[string]any{"priority": int64(10), "policy": "deny", "match": cidrMatch}), + cgroup("first", map[string]any{"priority": int64(1), "policy": "allow", "match": cidrMatch}), + ) + sn := next(t, ch) + var names []string + for _, g := range sn.Groups { + names = append(names, g.Name) + } + if got := strings.Join(names, ","); got != "first,a-kids,b-kids" { + t.Fatalf("order = %s", got) + } +} + +func TestRejectedObjectIsReportedAndOthersSurvive(t *testing.T) { + _, _, ch := start(t, + cgroup("good", map[string]any{"policy": "allow", "match": cidrMatch}), + cgroup("badpolicy", map[string]any{"policy": "maybe", "match": cidrMatch}), + cgroup("badcidr", map[string]any{"policy": "deny", "match": map[string]any{"cidrs": []any{"10.0.0.0/33"}}}), + cgroup("empty", map[string]any{"policy": "deny", "match": map[string]any{}}), + ) + sn := next(t, ch) + if len(sn.Groups) != 1 || sn.Groups[0].Name != "good" { + t.Fatalf("groups = %+v", sn.Groups) + } + if len(sn.Problems) != 3 { + t.Fatalf("problems = %+v", sn.Problems) + } +} + +func TestBrokenEditKeepsLastGoodVersion(t *testing.T) { + _, c, ch := start(t, cgroup("kids", map[string]any{"policy": "deny", "match": cidrMatch})) + if sn := next(t, ch); sn.Groups[0].Policy != "deny" { + t.Fatalf("first = %+v", sn.Groups) + } + // A bad edit changes nothing in force, so it publishes nothing... + bad := cgroup("kids", map[string]any{"policy": "bogus", "match": cidrMatch}) + if _, err := c.Resource(GroupGVR).Namespace("dns").Update(context.Background(), bad, metav1UpdateOptions); err != nil { + t.Fatal(err) + } + select { + case sn := <-ch: + t.Fatalf("a rejected edit published a snapshot: %+v", sn.Groups) + case <-time.After(300 * time.Millisecond): + } + // ...and a good edit goes through. + good := cgroup("kids", map[string]any{"policy": "block_only", "match": cidrMatch}) + if _, err := c.Resource(GroupGVR).Namespace("dns").Update(context.Background(), good, metav1UpdateOptions); err != nil { + t.Fatal(err) + } + if sn := next(t, ch); sn.Groups[0].Policy != "block_only" { + t.Fatalf("after good edit = %+v", sn.Groups) + } +} + +func TestDeleteRemovesGroup(t *testing.T) { + _, c, ch := start(t, cgroup("kids", map[string]any{"policy": "deny", "match": cidrMatch})) + next(t, ch) + if err := c.Resource(GroupGVR).Namespace("dns").Delete(context.Background(), "kids", metav1DeleteOptions); err != nil { + t.Fatal(err) + } + if sn := next(t, ch); len(sn.Groups) != 0 { + t.Fatalf("groups = %+v", sn.Groups) + } +} + +func TestBlockListConversion(t *testing.T) { + for name, tc := range map[string]struct { + spec map[string]any + ok bool + }{ + "https": {map[string]any{"url": "https://example.com/hosts", "format": "hosts", "refresh": "6h"}, true}, + "file url": {map[string]any{"url": "file:///etc/shadow"}, false}, + "no url": {map[string]any{"format": "hosts"}, false}, + "bad format": {map[string]any{"url": "https://e.com/x", "format": "yaml"}, false}, + "bad refresh": {map[string]any{"url": "https://e.com/x", "refresh": "soon"}, false}, + "bad mirror": {map[string]any{"url": "https://e.com/x", "mirror": []any{"file:///x"}}, false}, + } { + t.Run(name, func(t *testing.T) { + l, err := convertList(obj("BlockList", "ads", tc.spec)) + if (err == nil) != tc.ok { + t.Fatalf("err = %v", err) + } + if tc.ok && (!l.Enabled || l.Path != "" || l.Refresh != config.Duration(6*time.Hour)) { + t.Fatalf("list = %+v", l) + } + }) + } +} + +func zoneSpec(records ...any) map[string]any { + return map[string]any{"zone": "Home.Example.com", "records": records} +} + +func rec(name, typ, value string) map[string]any { + return map[string]any{"name": name, "type": typ, "value": value} +} + +func TestZoneRendersAndRoundTrips(t *testing.T) { + z, err := convertZone(obj("DNSZone", "home", zoneSpec( + rec("@", "NS", "ns1.home.example.com."), + rec("ns1", "A", "10.0.0.1"), + rec("nas", "A", "10.0.0.2"), + rec("*.apps", "A", "10.0.0.3"), + ))) + if err != nil { + t.Fatal(err) + } + if z.apex != "home.example.com." { + t.Fatalf("apex = %q", z.apex) + } + parsed, err := zonefile.Parse(strings.NewReader(z.text), zonefile.Options{Origin: z.apex, File: "t"}) + if err != nil { + t.Fatalf("rendered zone does not parse: %v\n%s", err, z.text) + } + if parsed.SOA == nil || parsed.SOA.Serial != 3 { + t.Fatalf("SOA serial should be the object generation: %+v", parsed.SOA) + } + // Same input, same bytes: replicas must agree. + again, _ := convertZone(obj("DNSZone", "home", zoneSpec( + rec("@", "NS", "ns1.home.example.com."), + rec("ns1", "A", "10.0.0.1"), + rec("nas", "A", "10.0.0.2"), + rec("*.apps", "A", "10.0.0.3"), + ))) + if again.text != z.text { + t.Fatal("rendering is not deterministic") + } + store, err := localzone.Load(writeTemp(t, z), localzone.Options{}) + if err != nil { + t.Fatal(err) + } + if store.Count() == 0 { + t.Fatal("store loaded no records") + } +} + +func TestZoneRejections(t *testing.T) { + for name, spec := range map[string]map[string]any{ + "no apex NS": zoneSpec(rec("nas", "A", "10.0.0.2")), + "bad value": zoneSpec(rec("@", "NS", "ns1.home.example.com."), rec("nas", "A", "not-an-ip")), + "outside zone": zoneSpec(rec("@", "NS", "ns1.home.example.com."), rec("x.other.org.", "A", "10.0.0.2")), + "duplicate": zoneSpec(rec("@", "NS", "ns1.home.example.com."), rec("a", "A", "10.0.0.2"), rec("a", "A", "10.0.0.2")), + "under delegated": zoneSpec(rec("@", "NS", "ns1.home.example.com."), rec("sub", "NS", "ns.else.net."), rec("x.sub", "A", "10.0.0.2")), + "soa as record": zoneSpec(rec("@", "NS", "ns1.home.example.com."), rec("@", "SOA", "a. b. 1 2 3 4 5")), + "no zone": {"records": []any{}}, + } { + t.Run(name, func(t *testing.T) { + if _, err := convertZone(obj("DNSZone", "z", spec)); err == nil { + t.Fatal("expected a rejection") + } + }) + } +} + +func TestDuplicateApexOldestWins(t *testing.T) { + a := obj("DNSZone", "a", zoneSpec(rec("@", "NS", "ns1.home.example.com."))) + b := obj("DNSZone", "b", zoneSpec(rec("@", "NS", "ns2.home.example.com."))) + a.SetCreationTimestamp(metav1Time(1000)) + b.SetCreationTimestamp(metav1Time(2000)) + _, _, ch := start(t, b, a) + sn := next(t, ch) + if len(sn.Zones) != 1 || !strings.Contains(sn.Zones["home.example.com."], "ns1.home.example.com.") { + t.Fatalf("zones = %v", sn.Zones) + } + if len(sn.Problems) != 1 || sn.Problems[0].Name != "b" { + t.Fatalf("problems = %+v", sn.Problems) + } +} + +func TestMergePrecedenceAndDefaults(t *testing.T) { + base := config.Default() + base.ACL.Groups = []config.Group{{Name: "yaml", Policy: "allow", Match: config.Match{CIDRs: []string{"192.168.0.0/16"}}}} + base.Block.Lists = []config.ListSource{{Name: "ads", URL: "https://a/x", Enabled: true}} + snap := &Snapshot{ + Groups: []config.Group{{Name: "yaml", Policy: "deny"}, {Name: "crd", Policy: "deny"}}, + Lists: []config.ListSource{{Name: "ads", URL: "https://b/x"}, {Name: "crd-list", URL: "https://c/x", Enabled: true}}, + } + merged, dropped := Merge(base, snap) + if len(merged.ACL.Groups) != 2 || merged.ACL.Groups[0].Policy != "allow" || merged.ACL.Groups[1].Name != "crd" { + t.Fatalf("groups = %+v", merged.ACL.Groups) + } + if len(merged.Block.Lists) != 2 || merged.Block.Lists[0].URL != "https://a/x" { + t.Fatalf("lists = %+v", merged.Block.Lists) + } + if got := merged.Block.Lists[1]; got.Format != "auto" || got.Refresh == 0 || got.MaxBytes == 0 { + t.Fatalf("defaults not applied: %+v", got) + } + if len(dropped) != 2 { + t.Fatalf("dropped = %v", dropped) + } + if len(base.ACL.Groups) != 1 || len(base.Block.Lists) != 1 { + t.Fatal("Merge modified its input") + } +} + +func TestSyncZoneDir(t *testing.T) { + dir := t.TempDir() + if ch, err := SyncZoneDir(dir, map[string]string{"a.example.": "x", "b.example.": "y"}); err != nil || !ch { + t.Fatalf("first sync: %v %v", ch, err) + } + if ch, err := SyncZoneDir(dir, map[string]string{"a.example.": "x", "b.example.": "y"}); err != nil || ch { + t.Fatalf("idempotent sync: %v %v", ch, err) + } + if ch, err := SyncZoneDir(dir, map[string]string{"a.example.": "x"}); err != nil || !ch { + t.Fatalf("removal sync: %v %v", ch, err) + } + if got := fileNames(t, dir); got != "a.example.zone" { + t.Fatalf("files = %s", got) + } +} + +func TestSnapshotSaveLoad(t *testing.T) { + path := t.TempDir() + "/sub/snap.json" + in := &Snapshot{Revision: 4, Groups: []config.Group{{Name: "g", Policy: "allow"}}, Zones: map[string]string{"a.": "z"}} + if err := Save(path, in); err != nil { + t.Fatal(err) + } + out, err := Load(path) + if err != nil || out.Revision != 4 || out.Groups[0].Name != "g" || out.Zones["a."] != "z" { + t.Fatalf("%+v %v", out, err) + } + if none, err := Load(path + ".missing"); none != nil || err != nil { + t.Fatalf("missing file: %v %v", none, err) + } +} diff --git a/internal/kubesource/types.go b/internal/kubesource/types.go new file mode 100644 index 0000000..798cc7b --- /dev/null +++ b/internal/kubesource/types.go @@ -0,0 +1,83 @@ +// Package kubesource turns fancy-dns custom resources (DNSZone, ClientGroup, +// BlockList) into the runtime's own configuration types. It is read-only: the +// runtime watches the cluster and never writes to it. +// +// Objects are decoded from unstructured data into the plain structs below, so +// there is no generated clientset or deepcopy code to keep in step with the +// CRD manifests in deploy/k8s/crds. +package kubesource + +import ( + "k8s.io/apimachinery/pkg/runtime/schema" + + "git.k8s-git-k6n.net/mats/fancy-dns/internal/config" +) + +// APIGroup and Version name the custom resources. +const ( + APIGroup = "fancydns.k6n.net" + Version = "v1alpha1" +) + +// The resources the runtime watches. +var ( + ZoneGVR = schema.GroupVersionResource{Group: APIGroup, Version: Version, Resource: "dnszones"} + GroupGVR = schema.GroupVersionResource{Group: APIGroup, Version: Version, Resource: "clientgroups"} + ListGVR = schema.GroupVersionResource{Group: APIGroup, Version: Version, Resource: "blocklists"} + allGVRs = []schema.GroupVersionResource{ZoneGVR, GroupGVR, ListGVR} + kindNames = map[string]string{ZoneGVR.Resource: "DNSZone", GroupGVR.Resource: "ClientGroup", ListGVR.Resource: "BlockList"} +) + +// DNSZoneSpec declares one authoritative zone. The zone's name is spec.zone, +// not metadata.name, because a DNS name is not always a legal object name. +type DNSZoneSpec struct { + Zone string `json:"zone"` + // TTL is the zone's $TTL and SOA minimum; 0 means the writer's default. + TTL uint32 `json:"ttl,omitempty"` + // SOA is the SOA rdata ("ns1.example.com. hostmaster.example.com. ..."), + // or empty to synthesise one from the first apex NS record. + SOA string `json:"soa,omitempty"` + Records []ZoneRecord `json:"records,omitempty"` +} + +// ZoneRecord is one resource record. Value is presentation-format RDATA. +type ZoneRecord struct { + Name string `json:"name"` + Type string `json:"type"` + TTL uint32 `json:"ttl,omitempty"` + Value string `json:"value"` +} + +// ClientGroupSpec mirrors config.Group. The group's name is metadata.name. +// Groups are evaluated first-match-wins, so Priority orders them: lower runs +// first, ties break on name. +type ClientGroupSpec struct { + Priority int32 `json:"priority,omitempty"` + Comment string `json:"comment,omitempty"` + Match config.Match `json:"match"` + Policy string `json:"policy"` + AllowDomains []string `json:"allow_domains,omitempty"` + DenyDomains []string `json:"deny_domains,omitempty"` + SkipBlocklist bool `json:"skip_blocklist,omitempty"` + SkipCache bool `json:"skip_cache,omitempty"` + RateLimit int `json:"rate_limit,omitempty"` + MaxConcurrent int `json:"max_concurrent,omitempty"` + LogQueries bool `json:"log_queries,omitempty"` + Upstreams []string `json:"upstreams,omitempty"` +} + +// BlockListSpec mirrors config.ListSource for a remote list. There is no path +// field on purpose: a namespace-scoped object must not be able to name a file +// the runtime reads or writes. file:// URLs are refused for the same reason. +type BlockListSpec struct { + URL string `json:"url"` + Format string `json:"format,omitempty"` + Mirror []string `json:"mirror,omitempty"` + Disabled bool `json:"disabled,omitempty"` + Refresh string `json:"refresh,omitempty"` + Timeout string `json:"timeout,omitempty"` + MaxBytes int64 `json:"max_bytes,omitempty"` + Categories []string `json:"categories,omitempty"` + Comment string `json:"comment,omitempty"` + InsecureSkipTLSVerify bool `json:"insecure_skip_tls_verify,omitempty"` +} diff --git a/internal/localzone/render.go b/internal/localzone/render.go new file mode 100644 index 0000000..da2eb1d --- /dev/null +++ b/internal/localzone/render.go @@ -0,0 +1,61 @@ +package localzone + +import ( + "fmt" + "strings" +) + +// RenderZone renders a complete zone from declared records, for sources that +// own a zone outright (a Kubernetes DNSZone) rather than editing a file in +// place. It runs the same checks the console writer does: every record goes +// through newRecord, a delegation may not shadow data, and the rendering is +// re-parsed and compared with the records before it is returned, so a renderer +// bug surfaces as an error here and not as a zone that silently stopped +// answering. +// +// serial is the SOA serial; 0 means the writer's default. A declarative source +// should pass something that changes exactly when the spec does, so two +// replicas rendering the same object produce the same bytes. +func RenderZone(origin string, def ZoneDefaults, recs []Record, serial uint32) (string, error) { + origin, err := NormalizeOrigin(origin) + if err != nil { + return "", fmt.Errorf("zone name: %w", err) + } + f, err := newFileState(origin, def) + if err != nil { + return "", err + } + for _, in := range recs { + rec, err := newRecord(origin, in.Name, in.Type, ttlOr(in.TTL, f.ttl), in.Value) + if err != nil { + return "", err + } + for _, have := range f.recs { + if have.Key() == rec.Key() && have.Value == rec.Value { + return "", fmt.Errorf("%s %s is declared twice", rec.Name, rec.Type) + } + } + f.recs = append(f.recs, rec) + } + if err := f.finish(); err != nil { + return "", err + } + if serial != 0 { + f.soa.Serial = serial + } + text := f.render() + check, err := parseFileState(origin, text) + if err != nil { + return "", fmt.Errorf("the zone file for %s did not render correctly: %w", origin, err) + } + if err := sameRecords(check.recs, f.recs); err != nil { + return "", fmt.Errorf("the zone file for %s did not render correctly: %w", origin, err) + } + return strings.Replace(text, FileHeader, ZoneSourceHeader, 1), nil +} + +// ZoneSourceHeader tops a rendered zone that an external object owns. +const ZoneSourceHeader = `; fancy-dns zone rendered from a Kubernetes DNSZone object. +; +; Regenerated whenever the object changes; edit the DNSZone, not this file. +` diff --git a/internal/qlog/qlog.go b/internal/qlog/qlog.go index da1c515..3398d29 100644 --- a/internal/qlog/qlog.go +++ b/internal/qlog/qlog.go @@ -339,7 +339,9 @@ func (l *Logger) Flush() { return } batch := l.pending - l.pending = l.pending[:0] + // Not pending[:0]: that would hand the next Log call the array this batch is + // still being read from. + l.pending = nil l.pendingMu.Unlock() err := l.db.Update(func(tx *bolt.Tx) error { diff --git a/internal/qlog/qlog_test.go b/internal/qlog/qlog_test.go index b0db0ee..7658114 100644 --- a/internal/qlog/qlog_test.go +++ b/internal/qlog/qlog_test.go @@ -836,3 +836,30 @@ func TestEntriesWithNoTimestampAreStamped(t *testing.T) { t.Errorf("TS = %v, want the injected clock %v", got[0].TS, want) } } + +// TestFlushDoesNotShareItsBatchWithNewEntries pins a race the detector finds: +// Flush took the pending slice and reset it to length zero, so entries logged +// while it was still marshalling the batch were appended over the same backing +// array. Under -race this fails without the fix; without -race it can store +// the wrong entries. +func TestFlushDoesNotShareItsBatchWithNewEntries(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.FlushInterval = time.Hour }) + for i := 0; i < 500; i++ { + l.Log(entryFor(i % 50)) // a batch big enough to take a while to store + } + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 1000; i++ { + l.Log(entryFor(i % 50)) + if i%10 == 0 { + time.Sleep(10 * time.Microsecond) + } + } + }() + for i := 0; i < 20; i++ { + l.Flush() + } + <-done + l.Flush() +}