package app import ( "context" "errors" "fmt" "net/netip" "reflect" "sort" "strings" "time" "github.com/miekg/dns" "git.k8s-git-k6n.net/mats/fancy-dns/internal/api" "git.k8s-git-k6n.net/mats/fancy-dns/internal/blocklist" "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/resolver" ) // hooks builds the api.Hooks that connect the admin API to the live // components. Everything the UI can read or change goes through here. func (a *App) hooks() api.Hooks { return api.Hooks{ Store: a.cfg, Logger: a.log, Version: a.ver, StartedAt: a.startedAt, Status: a.status, Health: a.health, BlocklistState: a.blocklistState, BlocklistStats: a.blocklistStats, RefreshBlocklists: a.refreshBlocklists, ToggleBlocklist: a.toggleBlocklist, DomainCheck: a.domainCheck, Resolve: a.resolveTest, ResolveUpstream: a.resolveTestUpstream, ListDomains: a.listDomains, AddDomain: a.addDomain, DeleteDomain: a.deleteDomain, ListZones: a.listZones, GetACL: a.getACL, PutACL: a.putACL, GetUpstream: a.getUpstreams, PutUpstream: a.putUpstream, CacheStats: a.cacheStats, FlushCache: a.flushCache, Flush: a.flush, } } func (a *App) resolveTestUpstream(ctx context.Context, name, qtype, upstream string) (any, error) { if a.pool == nil { return nil, errors.New("upstream pool is not running") } up := a.pool.Get(strings.TrimSpace(upstream)) if up == nil { return nil, fmt.Errorf("unknown upstream %q", upstream) } qt := dns.TypeA if qtype != "" { var ok bool qt, ok = dns.StringToType[strings.ToUpper(strings.TrimSpace(qtype))] if !ok { return nil, fmt.Errorf("unknown record type %q", qtype) } } q := new(dns.Msg) q.SetQuestion(dns.Fqdn(name), qt) q.RecursionDesired = true started := time.Now() resp, addr, err := a.pool.Exchange(ctx, q, []string{up.Addr()}, 5*time.Second) if err != nil { return nil, err } out := map[string]any{"name": dns.Fqdn(name), "type": dns.TypeToString[qt], "rcode": "SERVFAIL", "action": "upstream", "duration_ms": time.Since(started).Milliseconds(), "upstream": addr, "group": "", "cache": "bypass", "local": false, "truncated": false, "blocked": false, "block_source": "", "answers": []string{}} if resp != nil { out["rcode"] = dns.RcodeToString[resp.Rcode] ans := []string{} for _, rr := range resp.Answer { ans = append(ans, strings.TrimSpace(rr.String())) } out["answers"] = ans } return out, nil } // health is the unauthenticated probe payload. func (a *App) health() any { status := "ok" problems := a.Degraded() switch { case len(problems) > 0: status = "degraded" } if a.dns == nil || len(a.dns.Listeners()) == 0 { status = "down" } if a.logger.Dropped() > 0 && status == "ok" { // Lines are dropped when the log stream cannot keep up. DNS is fine, but // whoever reads the log is missing events, which is worth a mention. status = "degraded" problems = append(problems, "access log is dropping lines") } return map[string]any{ "status": status, "version": a.ver, "uptime_s": int64(time.Since(a.startedAt).Seconds()), "config_version": a.cfg.Get().Version, "problems": problems, } } // status is the per-component runtime view. func (a *App) status() any { out := map[string]any{ "listeners": a.dns.Listeners(), "transports": a.dns.ByTransport(), "degraded": a.Degraded(), } if a.blocks != nil { st := a.blocks.State() if st != nil { out["blocklist"] = map[string]any{ "rules": st.Set.Len(), "sources": len(st.Sources), "duration": st.Duration, "built_at": st.Set.BuiltAt(), "errors": st.Errors, "digest": st.Digest, "has_wild": st.Set.HasWildcards(), "duplicates": st.Set.DuplicatesN(), } } } if a.local != nil { if store := a.local.Get(); store != nil { out["local_zones"] = map[string]any{ "zones": store.Zones(), "records": store.Count(), "files": store.Files(), } } } if a.cache != nil { out["cache"] = a.cacheStats() } out["access_log"] = map[string]any{ "enabled": a.logger != nil, "dropped": a.logger.Dropped(), } if a.pool != nil { out["upstream"] = a.getUpstreams() } if a.aclH != nil { out["acl"] = a.getACL() } if a.cl != nil { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) out["cluster"] = a.cl.Status(ctx) cancel() } if a.engine != nil { out["engine"] = map[string]any{ "inflight": a.engine.Inflight(), "dropped": a.engine.Dropped(), "group_stats": a.engine.GroupStats(), } } return out } // blocklistState is the full per-source view for the UI. func (a *App) blocklistState() any { if a.blocks == nil { return map[string]any{ "items": []any{}, "stats": map[string]any{"total": 0}, } } st := a.blocks.State() items := make([]blocklist.Source, 0, len(st.Sources)) for _, s := range st.Sources { items = append(items, s) } sort.Slice(items, func(i, j int) bool { return items[i].Name < items[j].Name }) return map[string]any{ "items": items, "stats": blocklistSourceStats(st), "errors": st.Errors, "digest": st.Digest, } } func blocklistSourceStats(st *blocklist.State) map[string]any { if st == nil || st.Set == nil { return map[string]any{"total": 0} } s := st.Set.Stats() return map[string]any{ "total": s.Total, "exact": s.Exact, "subdomain_rules": s.Subdomains, "root_rules": s.RootRules, "allow": s.Allow, "block": s.Block, "passthrough": s.Passthru, "drop": s.Drop, "duplicates": s.Duplicates, "sources": s.Sources, "built_at": s.BuiltAt, "sources_detail": s.PerSource, } } func (a *App) blocklistStats() any { if a.blocks == nil { return map[string]any{"total": 0} } return blocklistSourceStats(a.blocks.State()) } // invalidateBlocklistCache drops every cached answer whose verdict the new // rules might change. Without this, a name that has just become blocked keeps // serving its cached good answer until the TTL expires, which is exactly the // bug an operator hits right after clicking "add rule". func (a *App) invalidateBlocklistCache() { if a.cache == nil { return } a.cache.Flush() } // refreshBlocklists recompiles the list sources. func (a *App) refreshBlocklists(ctx context.Context, names []string) error { if a.blocks == nil { return fmt.Errorf("blocklists are disabled") } a.invalidateBlocklistCache() go func() { if len(names) == 0 { if _, err := a.blocks.ForceRefresh(ctx); err != nil { a.log.Warn("blocklist refresh failed", "error", err) } return } // The manager compiles everything in one pass, which is cheaper than // a pass per name and keeps the compiled set internally consistent. if _, err := a.blocks.ForceRefresh(ctx); err != nil { a.log.Warn("blocklist refresh failed", "error", err, "names", names) } }() // Tell the cluster leader so every instance ends up on the same content. if a.cl != nil { _ = a.cl.BroadcastRefresh(ctx, names) } return nil } // toggleBlocklist flips one source's enabled flag and persists it. 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 { if c.Block.Lists[i].Name == name { c.Block.Lists[i].Enabled = enabled found = true } } if !found { return fmt.Errorf("no blocklist named %q", name) } return nil }) if err != nil { return err } a.invalidateBlocklistCache() go func() { if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil { a.log.Warn("refresh after toggle failed", "error", rerr) } }() return nil } // 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 { // 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. next.Block = cur.Block next.Block.Lists = append([]config.ListSource(nil), cur.Block.Lists...) next.Block.LocalZones = append([]string(nil), cur.Block.LocalZones...) next.Block.BlockEntries = append([]string(nil), cur.Block.BlockEntries...) next.Block.AllowEntries = append([]string(nil), cur.Block.AllowEntries...) next.ACL = cur.ACL next.ACL.Groups = append([]config.Group(nil), cur.ACL.Groups...) next.Upstream = cur.Upstream next.Upstream.Servers = append([]config.UpstreamServer(nil), cur.Upstream.Servers...) next.Upstream.Bootstraps = append([]config.UpstreamBootstrap(nil), cur.Upstream.Bootstraps...) if err := fn(&next); err != nil { return err } if err := next.Validate(); err != nil { return err } if a.api == nil { return a.cfg.Swap(&next) } return a.api.Commit(&next) } // domainCheck explains how a name would be treated. func (a *App) domainCheck(name string) (any, error) { entry := (*blocklist.Entry)(nil) if a.blocks != nil { if set := a.blocks.Holder().Get(); set != nil { entry = set.Match(name) } } out := map[string]any{ "name": name, "query": strings.TrimSuffix(name, "."), } if entry == nil { out["matched"] = false out["action"] = "none" out["source"] = "" return out, nil } kind := "exact" pattern := entry.Name if set := a.blocks.Holder().Get(); set != nil { // Recover which of the three maps matched so the UI can say *why*. switch set.MatchPattern(name) { case blocklist.PatternSub: kind = "subdomain" pattern = "*." + entry.Name case blocklist.PatternRoot: kind = "domain" pattern = "**." + entry.Name } } out["matched"] = true out["action"] = entry.Action.String() out["source"] = entry.Source out["categories"] = entry.Categories out["hits"] = entry.Hits() out["rule"] = pattern out["matched_pattern"] = pattern out["match_kind"] = kind return out, nil } // resolveTest runs one query through the live pipeline. It goes through the // engine rather than straight to an upstream: a test that skipped the // pipeline would report a healthy upstream while the blocklist, the ACL or the // local zones were what a client actually hit. func (a *App) resolveTest(ctx context.Context, name, qtype string) (any, error) { if a.engine == nil { return nil, errors.New("the query engine is not running") } qt := dns.TypeA if qtype = strings.TrimSpace(qtype); qtype != "" { var ok bool qt, ok = dns.StringToType[strings.ToUpper(qtype)] if !ok { return nil, fmt.Errorf("unknown record type %q", qtype) } } q := new(dns.Msg) q.SetQuestion(dns.Fqdn(name), qt) q.RecursionDesired = true res := a.engine.Handle(ctx, engine.Request{ Msg: q, Remote: netip.AddrFrom4([4]byte{127, 0, 0, 1}), Transport: "api", Start: time.Now(), }) out := map[string]any{ "name": dns.Fqdn(name), "type": dns.TypeToString[qt], "rcode": "SERVFAIL", "action": string(res.Action), "duration_ms": res.Duration.Milliseconds(), "upstream": res.Upstream, "group": res.Group, "cache": res.CacheKind, "local": res.Local, "truncated": res.Truncated, "blocked": res.BlockSource != "", "block_source": res.BlockSource, "answers": []string{}, } if res.Msg == nil { out["error"] = "no response" return out, nil } out["rcode"] = dns.RcodeToString[res.Msg.Rcode] answers := make([]string, 0, len(res.Msg.Answer)) for _, rr := range res.Msg.Answer { answers = append(answers, strings.TrimSpace(rr.String())) } out["answers"] = answers return out, nil } // listDomains pages the compiled rules. // request would be too slow, so the filter runs over a bounded sample unless // the query is a prefix/exact lookup, which the LRU can answer directly. func (a *App) listDomains(q api.DomainQuery) (any, error) { if a.blocks == nil { return map[string]any{"items": []any{}, "total": 0, "page": 1, "per_page": q.PerPage}, nil } set := a.blocks.Holder().Get() if set == nil { return map[string]any{"items": []any{}, "total": 0, "page": 1, "per_page": q.PerPage}, nil } if q.Text != "" && (q.Action == "" && q.List == "") { if e := set.Lookup(q.Text); e != nil { return map[string]any{ "items": []blocklist.Entry{*e}, "total": 1, "page": 1, "per_page": q.PerPage, }, nil } } type row struct { Name string `json:"name"` Action string `json:"action"` Source string `json:"source"` Categories []string `json:"categories"` Hits uint64 `json:"hits"` Rule string `json:"rule"` } text := strings.ToLower(strings.TrimSpace(q.Text)) action := strings.ToLower(q.Action) // The cap keeps a request bounded on a large set; the UI pages anyway. const scanCap = 400000 matches := make([]row, 0, 256) total := 0 set.Each(func(pattern string, e *blocklist.Entry, kind blocklist.PatternKind) bool { if len(matches) >= scanCap { return false } if action != "" && e.Action.String() != action { return true } if q.List != "" && e.Source != q.List { return true } if text != "" && !strings.Contains(pattern, text) { return true } total++ matches = append(matches, row{ Name: pattern, Action: e.Action.String(), Source: e.Source, Categories: e.Categories, Hits: e.Hits(), // The wire form, so the console can re-add this exact rule // (with its wildcard) instead of degrading it to an exact name. Rule: kind.WireName(pattern), }) return true }) sort.Slice(matches, func(i, j int) bool { if matches[i].Hits != matches[j].Hits { return matches[i].Hits > matches[j].Hits } return matches[i].Name < matches[j].Name }) start := (q.Page - 1) * q.PerPage if start > len(matches) { start = len(matches) } end := start + q.PerPage if end > len(matches) { end = len(matches) } return map[string]any{ "items": matches[start:end], "total": total, "page": q.Page, "per_page": q.PerPage, "truncated": total > len(matches), }, nil } // addDomain appends an inline rule and persists it. func (a *App) addDomain(_ context.Context, in api.DomainInput) (any, error) { name := strings.TrimSpace(in.Name) if name == "" { return nil, fmt.Errorf("name is required") } act, ok := blocklist.ParseAction(in.Action) if !ok { return nil, fmt.Errorf("unknown action %q", in.Action) } // A bulk paste goes in as several entries; a single name goes in as one. entries := []string{} if in.Content != "" { entries = strings.Split(in.Content, "\n") } else { entries = []string{name} } err := a.mutateConfig(func(c *config.Config) error { for _, e := range entries { e = strings.TrimSpace(e) if e == "" || strings.HasPrefix(e, "#") { continue } entry := e if act == blocklist.ActionBlock { c.Block.BlockEntries = append(c.Block.BlockEntries, entry) } else { c.Block.AllowEntries = append(c.Block.AllowEntries, entry) } } return nil }) if err != nil { return nil, err } if a.blocks != nil { if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil { a.log.Warn("refresh after add failed", "error", rerr) } } // A new rule changes the verdict for the whole subtree it covers, and an // allow rule can rescue a name many other lists blocked, so the honest // thing is to drop the affected answers rather than guess. a.invalidateBlocklistCache() return map[string]any{"ok": true, "name": name, "count": len(entries)}, nil } // deleteDomain removes a rule and persists the change. func (a *App) deleteDomain(_ context.Context, name, action, list string) error { target := strings.ToLower(strings.TrimSpace(name)) removed := 0 err := a.mutateConfig(func(c *config.Config) error { strip := func(entries []string) []string { out := entries[:0] for _, e := range entries { normal := strings.ToLower(strings.TrimSpace(e)) normal = strings.TrimPrefix(strings.TrimPrefix(normal, "*."), "**.") normal = strings.TrimSuffix(normal, ".") if normal == target { removed++ continue } out = append(out, e) } return out } c.Block.BlockEntries = strip(c.Block.BlockEntries) c.Block.AllowEntries = strip(c.Block.AllowEntries) // A file-backed list is edited by editing the file, so the API only // reports that rather than pretending it removed something. if list != "" && removed == 0 { return fmt.Errorf("%q comes from list %q; edit that list instead", name, list) } if removed == 0 { return fmt.Errorf("no inline rule matches %q", name) } return nil }) if err != nil { return err } if a.blocks != nil { if _, rerr := a.blocks.RefreshAll(context.Background()); rerr != nil { a.log.Warn("refresh after delete failed", "error", rerr) } } a.invalidateBlocklistCache() return nil } // getACL returns the ACL groups. func (a *App) getACL() any { groups := a.engine.GroupStats() if a.kubeSnap.Load() != nil { inFile := map[string]bool{} for _, g := range a.cfg.Base().ACL.Groups { inFile[g.Name] = true } for i := range groups { if !inFile[groups[i].Name] && a.clusterOwned("group", groups[i].Name) { groups[i].Source = "kubernetes" } } } return map[string]any{ "groups": groups, "default_policy": a.cfg.Get().ACL.DefaultPolicy, "default_rate_limit": a.cfg.Get().ACL.DefaultRateLimit, } } // 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 != "" { c.ACL.DefaultPolicy = def } return nil }) if err != nil { return 0, err } // An allow_domains or deny_domains change alters the verdict for names // already in the cache, so the cached answers are no longer trustworthy. if a.cache != nil { a.cache.Flush() } return a.cfg.Get().Version, nil } // getUpstreams returns the pool state. func (a *App) getUpstreams() any { cfg := a.cfg.Get() // Keep forwarding targets out of the editable default-server list. defaults := a.pool.Defaults() servers := make([]resolver.Stats, 0, len(defaults)) for _, u := range defaults { servers = append(servers, u.Stats()) } return map[string]any{ "policy": cfg.Upstream.Policy, "servers": servers, // Durations on the wire are duration strings, everywhere: this is the // same spelling config.yaml uses and the same one Duration.MarshalJSON // writes for bootstraps[].refresh. It used to be a float of seconds // while the console declared the field a string. "bootstraps": cfg.Upstream.Bootstraps, "healthcheck_interval": cfg.Upstream.HealthCheck, "max_attempts": cfg.Upstream.MaxAttempts, "breaker_threshold": cfg.Upstream.BreakerThreshold, "breaker_cooldown": cfg.Upstream.BreakerCooldown, } } // putUpstream validates and persists a new upstream config. func (a *App) putUpstream(_ context.Context, up config.Upstream) (int64, error) { err := a.mutateConfig(func(c *config.Config) error { // The console edits default servers; preserve YAML-only forwarding rules. up.ForwardZones = c.Upstream.ForwardZones c.Upstream = up return nil }) if err != nil { return 0, err } // The pool rebinds itself on the config swap; do it now so the UI sees // the new servers immediately rather than after the next reload. a.pool.Rebind(up) return a.cfg.Get().Version, nil } // cacheStats returns the cache snapshot. func (a *App) cacheStats() any { if a.cache == nil { return map[string]any{"enabled": false} } st := a.cache.Stats() return map[string]any{ "enabled": st.Enabled, "size": st.Size, "entries": st.Entries, "shards": st.Shards, "min_ttl_ms": st.MinTTLMS, "max_ttl_ms": st.MaxTTLMS, "serve_stale_ms": st.ServeStaleM, "negative_ttl_ms": st.NegativeMS, "hit_rate": st.HitRate, "hits": st.Hits, "misses": st.Misses, "stale_hits": st.StaleHits, "evictions": st.Evictions, "bytes": st.Bytes, "bytes_human": st.MemoryHuman, "persisted": st.Persist, } } // flushCache empties the cache. func (a *App) flushCache() int64 { if a.cache == nil { return 0 } st := a.cache.Stats() a.cache.Flush() return int64(st.Entries) } // flush persists buffered state. func (a *App) flush() { a.logger.Flush() a.snapshotCache() }