diff --git a/.gitignore b/.gitignore index 0a8bf3b..e604d74 100644 --- a/.gitignore +++ b/.gitignore @@ -33,3 +33,6 @@ version.go # Ignore output dir out + +# Local Go build cache used by sandboxed verification +.gocache/ diff --git a/LIBRARY-REFACTOR.md b/LIBRARY-REFACTOR.md new file mode 100644 index 0000000..c187587 --- /dev/null +++ b/LIBRARY-REFACTOR.md @@ -0,0 +1,54 @@ +# Refactor `dnstapir-pop` Into `cmd/pop` + Root `pop` Package + +## Summary + +Move the executable entrypoint to `cmd/pop` and convert the repo root into package `pop`. The root package exposes a library-style `Run` API, while `cmd/pop/main.go` owns process concerns such as signals, stderr reporting, and exit codes. + +## Public API And Layout + +- Keep functionality in the repo root as `package pop`; do not create a `/pop` subdirectory. +- Add: + ```go + type RunOptions struct { + Name, Version, Commit string + Args []string + Stdout, Stderr io.Writer + Reload <-chan struct{} + } + + func Run(ctx context.Context, opts RunOptions) error + ``` +- `cmd/pop/main.go` defines the ldflag-backed `name`, `version`, and `commit` vars, installs SIGINT/SIGTERM context cancellation, converts SIGHUP into `RunOptions.Reload`, calls `pop.Run`, and uses `os.Exit(1)` only on returned errors. +- Remove root globals used only by the old executable path: `Gconfig`, `mqttclientid`, and `POPExiter`. + +## Implementation Changes + +- Rename all root Go files from `package main` to `package pop`; move the old startup logic out of `main()` into `Run`. +- Update `Makefile` so `make build` runs `go build ... -o out/dnstapir-pop ./cmd/pop`; keep package/install artifacts and binary name unchanged. +- Replace process exits in `pop` with returned errors: + - `SetupLogging`, config validation/loading, MQTT setup/start, source parsing, output parsing, bootstrap setup, and policy parsing return contextual errors. + - `log.Fatal`, `os.Exit`, `panic`, `log.Panicf`, and `POPExiter` disappear from package `pop` for normal failure paths. +- Convert long-running workers to context/error style: + - `DnsEngine(ctx, *Config) error` + - `APIhandler(ctx, *Config) error` + - `ConfigUpdater(ctx, *Config) error` + - `StatusUpdater(ctx, *Config) error` + - `RefreshEngine(ctx, *Config) error` +- `Run` starts workers, watches worker errors, context cancellation, API stop requests, and reload events; it saves the RPZ serial before returning. +- Remove `MqttEngine.SetupInterruptHandler()` from the library path; shutdown should flow through `Run` and `StopEngine`. +- Update policy/RPZ helper signatures where needed so invalid list/zone formats return errors instead of exiting, and propagate those errors through RPZ generation. +- Fix the current vet-blocking format string issues encountered in `configupdater.go`, `refreshengine.go`, and `statusupdater.go` as part of making `go test ./...` meaningful after the refactor. + +## Test Plan + +- Run `gofmt` on touched Go files. +- Run `go list ./...` and confirm packages include `dnstapir-pop` and `dnstapir-pop/cmd/pop`. +- Run `go test ./...`; expected target is green after the pre-existing vet issues are fixed. +- Run `make build` and confirm `out/dnstapir-pop` is produced with ldflag metadata wired through `cmd/pop`. +- Verify with search that `POPExiter`, root-package `os.Exit`, `log.Fatal`, and normal-error `panic` usages are gone. + +## Assumptions + +- The official library entrypoint is `pop.Run`; existing exported helper names can remain, but signatures may gain `context.Context` or `error` to satisfy the strict library API. +- Config file paths, service names, package install paths, and runtime behavior stay the same unless needed to remove process exits. +- Only `cmd/pop` may terminate the process; package `pop` reports failures to its caller. diff --git a/Makefile b/Makefile index 9842ca7..8f9c4c7 100644 --- a/Makefile +++ b/Makefile @@ -8,7 +8,8 @@ OUT:=$$(pwd)/out COMMIT:=$$(cat COMMIT 2> /dev/null || git describe --dirty=+WiP --always 2> /dev/null) GOFLAGS:=-v -ldflags "-X 'main.version=$(VERSION)' -X 'main.commit=$(COMMIT)' -X 'main.name=$(PROG)'" GOOS ?= $(shell uname -s | tr A-Z a-z) -GO:=GOOS=$(GOOS) GOARCH=$(GOARCH) CGO_ENABLED=0 go +GOCACHE ?= $(OUT)/.gocache +GO:=GOOS=$(GOOS) GOARCH=$(GOARCH) CGO_ENABLED=0 GOCACHE=$(GOCACHE) go INSTALL:=install -b -c -s -p -m 0755 # For version snapshots of packages @@ -26,7 +27,7 @@ default: $(PROG) $(PROG): build build: outdir - $(GO) build $(GOFLAGS) -o $(OUT)/$(PROG) + $(GO) build $(GOFLAGS) -o $(OUT)/$(PROG) ./cmd/pop outdir: @mkdir -p $(OUT) diff --git a/apihandler.go b/apihandler.go index 9564dde..dd56234 100644 --- a/apihandler.go +++ b/apihandler.go @@ -2,12 +2,14 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "context" "crypto/tls" "encoding/gob" "encoding/json" + "errors" "fmt" "log" "net/http" @@ -246,9 +248,9 @@ func APIbootstrap(conf *Config) func(w http.ResponseWriter, r *http.Request) { defer func() { w.Header().Set("Content-Type", "application/json") me := conf.PopData.MqttEngine - me.DataMu.Lock() /* Lock because resp.TopicData needs to be accessed safely */ + me.DataMu.Lock() /* Lock because resp.TopicData needs to be accessed safely */ err := json.NewEncoder(w).Encode(resp) - me.DataMu.Unlock() + me.DataMu.Unlock() if err != nil { log.Printf("Error from json encoder: %v", err) log.Printf("resp: %v", resp) @@ -270,9 +272,9 @@ func APIbootstrap(conf *Config) func(w http.ResponseWriter, r *http.Request) { switch bp.Command { case "doubtlist-status": me := conf.PopData.MqttEngine - me.DataMu.Lock() + me.DataMu.Lock() stats := me.Stats() - me.DataMu.Unlock() + me.DataMu.Unlock() // resp.MsgCounters = stats.MsgCounters // resp.MsgTimeStamps = stats.MsgTimeStamps resp.TopicData = stats @@ -544,7 +546,7 @@ func SetupBootstrapRouter(conf *Config) *mux.Router { return r } -func walkRoutes(router *mux.Router, address string) { +func walkRoutes(router *mux.Router, address string) error { log.Printf("Defined API endpoints for router on: %s\n", address) walker := func(route *mux.Route, router *mux.Router, ancestors []*mux.Route) error { @@ -556,18 +558,20 @@ func walkRoutes(router *mux.Router, address string) { return nil } if err := router.Walk(walker); err != nil { - log.Panicf("Logging err: %s\n", err.Error()) + return fmt.Errorf("walking routes: %w", err) } - // return nil + return nil } // In practice APIdispatcher doesn't need a termination signal, as it will // just sit inside http.ListenAndServe, but we keep it for symmetry. -func APIhandler(conf *Config, done <-chan struct{}) { +func APIhandler(ctx context.Context, conf *Config) error { gob.Register(tapir.WBGlist{}) // Must register the type for gob encoding router := SetupRouter(conf) - walkRoutes(router, viper.GetString("apiserver.address")) + if err := walkRoutes(router, viper.GetString("apiserver.address")); err != nil { + return err + } log.Println("") addresses := viper.GetStringSlice("apiserver.addresses") @@ -606,47 +610,70 @@ func APIhandler(conf *Config, done <-chan struct{}) { // tls.RequireAnyClientCert, tls.RequestClientCert, tls.NoClientCert if err != nil { - POPExiter("Error creating API server tls config: %v\n", err) + return fmt.Errorf("error creating API server tls config: %w", err) } var wg sync.WaitGroup + var serveWG sync.WaitGroup + serverErrCh := make(chan error, len(addresses)+len(tlsaddresses)+len(bootstrapaddresses)+len(bootstraptlsaddresses)) + var servers []*http.Server + + startServer := func(label string, server *http.Server, serve func() error) { + servers = append(servers, server) + wg.Add(1) + serveWG.Add(1) + go func() { + defer serveWG.Done() + log.Printf("*** API: Starting %s. Listening on %s", label, server.Addr) + wg.Done() + if err := serve(); err != nil && !errors.Is(err, http.ErrServerClosed) { + select { + case serverErrCh <- fmt.Errorf("%s on %s: %w", label, server.Addr, err): + case <-ctx.Done(): + } + } + }() + } + + shutdownServers := func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + for _, server := range servers { + if err := server.Shutdown(shutdownCtx); err != nil { + log.Printf("APIhandler: error shutting down server on %s: %v", server.Addr, err) + } + } + serveWG.Wait() + } + defer shutdownServers() // log.Println("*** API: Starting API dispatcher #1. Listening on", address) if len(addresses) > 0 { for idx, address := range addresses { - wg.Add(1) - go func(wg *sync.WaitGroup) { - apiServer := &http.Server{ - Addr: address, - Handler: router, - ReadTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, - } - - log.Printf("*** API: Starting API dispatcher #%d. Listening on %s", idx+1, address) - wg.Done() - POPExiter(apiServer.ListenAndServe()) - }(&wg) + apiServer := &http.Server{ + Addr: address, + Handler: router, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + } + startServer(fmt.Sprintf("API dispatcher #%d", idx+1), apiServer, apiServer.ListenAndServe) } } if len(tlsaddresses) > 0 { if tlspossible { for idx, tlsaddress := range tlsaddresses { - wg.Add(1) - go func(wg *sync.WaitGroup) { - tlsServer := &http.Server{ - Addr: tlsaddress, - Handler: router, - TLSConfig: tlsConfig, - ReadTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, - } - log.Printf("*** API: Starting TLS API dispatcher #%d. Listening on %s", idx+1, tlsaddress) - wg.Done() - POPExiter(tlsServer.ListenAndServeTLS(certfile, keyfile)) - }(&wg) + tlsServer := &http.Server{ + Addr: tlsaddress, + Handler: router, + TLSConfig: tlsConfig, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + } + startServer(fmt.Sprintf("TLS API dispatcher #%d", idx+1), tlsServer, func() error { + return tlsServer.ListenAndServeTLS(certfile, keyfile) + }) } } else { log.Printf("*** API: APIdispatcher: Error: Cannot provide TLS service without cert and key files.\n") @@ -655,18 +682,13 @@ func APIhandler(conf *Config, done <-chan struct{}) { if len(bootstrapaddresses) > 0 { for idx, address := range bootstrapaddresses { - wg.Add(1) - go func(wg *sync.WaitGroup) { - apiServer := &http.Server{ - Addr: address, - Handler: bootstraprouter, - ReadTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, - } - log.Printf("*** API: Starting Bootstrap API dispatcher #%d. Listening on %s", idx+1, address) - wg.Done() - POPExiter(apiServer.ListenAndServe()) - }(&wg) + apiServer := &http.Server{ + Addr: address, + Handler: bootstraprouter, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + } + startServer(fmt.Sprintf("Bootstrap API dispatcher #%d", idx+1), apiServer, apiServer.ListenAndServe) } } else { log.Println("*** API: No bootstrap address specified") @@ -675,20 +697,16 @@ func APIhandler(conf *Config, done <-chan struct{}) { if len(bootstraptlsaddresses) > 0 { if tlspossible { for idx, address := range bootstraptlsaddresses { - wg.Add(1) - go func(wg *sync.WaitGroup) { - bootstrapTlsServer := &http.Server{ - Addr: address, - Handler: bootstraprouter, - TLSConfig: tlsConfig, - ReadTimeout: 10 * time.Second, - WriteTimeout: 10 * time.Second, - } - - log.Printf("*** API: Starting Bootstrap TLS API dispatcher #%d. Listening on %s", idx+1, address) - wg.Done() - POPExiter(bootstrapTlsServer.ListenAndServeTLS(certfile, keyfile)) - }(&wg) + bootstrapTlsServer := &http.Server{ + Addr: address, + Handler: bootstraprouter, + TLSConfig: tlsConfig, + ReadTimeout: 10 * time.Second, + WriteTimeout: 10 * time.Second, + } + startServer(fmt.Sprintf("Bootstrap TLS API dispatcher #%d", idx+1), bootstrapTlsServer, func() error { + return bootstrapTlsServer.ListenAndServeTLS(certfile, keyfile) + }) } } else { log.Printf("*** API: APIdispatcher: Error: Cannot provide Bootstrap TLS service without cert and key files.\n") @@ -698,7 +716,18 @@ func APIhandler(conf *Config, done <-chan struct{}) { } wg.Wait() - log.Println("API dispatcher: unclear how to stop the http server nicely.") + if len(servers) == 0 { + log.Println("API dispatcher: no API servers configured") + return nil + } + + select { + case <-ctx.Done(): + log.Println("API dispatcher: stopping") + return nil + case err := <-serverErrCh: + return err + } } func BumpSerial(conf *Config, zone string) (string, error) { diff --git a/bootstrap.go b/bootstrap.go index ad71e93..160c696 100644 --- a/bootstrap.go +++ b/bootstrap.go @@ -2,7 +2,7 @@ * Copyright (c) Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "bytes" @@ -25,22 +25,17 @@ func (td *PopData) BootstrapMqttSource(src SourceConf) (*tapir.WBGlist, error) { AuthMethod: "X-API-Key", } - cd := viper.GetString("certs.certdir") - if cd == "" { - POPExiter("BootstrapMqttSource error: missing config key: certs.certdir") - } - // cert := cd + "/" + certname - key := viper.GetString("certs.tapir-pop.key") - cert := viper.GetString("certs.tapir-pop.cert") + key := viper.GetString("certs.tapir-pop.key") + cert := viper.GetString("certs.tapir-pop.cert") tlsConfig, err := tapir.NewClientConfig(viper.GetString("certs.cacertfile"), key, cert) if err != nil { - POPExiter("BootstrapMqttSource: Error: Could not set up TLS: %v", err) + return nil, fmt.Errorf("could not set up TLS: %w", err) } // XXX: Need to verify that the server cert is valid for the bootstrap server tlsConfig.InsecureSkipVerify = true err = api.SetupTLS(tlsConfig) if err != nil { - POPExiter("BootstrapMqttSource: error setting up TLS for the API client: %v", err) + return nil, fmt.Errorf("error setting up TLS for the API client: %w", err) } bootstrapaddrs := viper.GetStringSlice("bootstrapserver.addresses") diff --git a/cmd/pop/main.go b/cmd/pop/main.go new file mode 100644 index 0000000..bc2aa88 --- /dev/null +++ b/cmd/pop/main.go @@ -0,0 +1,59 @@ +/* + * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se + */ + +package main + +import ( + "context" + "fmt" + "os" + "os/signal" + "syscall" + + pop "dnstapir-pop" +) + +/* Rewritten if building with make */ +var name = "BAD-BUILD" +var version = "BAD-BUILD" +var commit = "BAD-BUILD" + +func main() { + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + reload := make(chan struct{}, 1) + hupper := make(chan os.Signal, 1) + signal.Notify(hupper, syscall.SIGHUP) + defer signal.Stop(hupper) + + go func() { + for { + select { + case <-ctx.Done(): + return + case <-hupper: + select { + case reload <- struct{}{}: + default: + } + } + } + }() + + if err := pop.Run(ctx, pop.RunOptions{ + Name: name, + Version: version, + Commit: commit, + Args: os.Args[1:], + Stdout: os.Stdout, + Stderr: os.Stderr, + Reload: reload, + }); err != nil { + if _, writeErr := fmt.Fprintln(os.Stderr, err); writeErr != nil { + os.Exit(1) + } + os.Exit(1) + } +} diff --git a/config.go b/config.go index 2cf36ae..7806728 100644 --- a/config.go +++ b/config.go @@ -2,9 +2,10 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "fmt" "log" "time" @@ -19,7 +20,7 @@ type Config struct { ApiServer ApiserverConf DnsEngine DnsengineConf BootstrapServer BootstrapServerConf - KeyStore KeystoreConf + KeyStore KeystoreConf Sources map[string]SourceConf Policy PolicyConf Log struct { @@ -78,7 +79,7 @@ type ServerConf struct { } type KeystoreConf struct { - Path string `validate:"required,file"` + Path string `validate:"required,file"` } type SourceConf struct { @@ -88,7 +89,7 @@ type SourceConf struct { Type string `validate:"required"` Format string `validate:"required"` Source string `validate:"required"` - Immutable bool + Immutable bool Topic string ValidatorKey string Bootstrap []string @@ -141,11 +142,11 @@ func ValidateConfig(v *viper.Viper, cfgfile string) error { if v == nil { if err := viper.Unmarshal(&config); err != nil { - POPExiter("ValidateConfig: Unmarshal error: %v", err) + return fmt.Errorf("ValidateConfig: unmarshal error: %w", err) } } else { if err := v.Unmarshal(&config); err != nil { - POPExiter("ValidateConfig: Unmarshal error: %v", err) + return fmt.Errorf("ValidateConfig: unmarshal error: %w", err) } } @@ -166,7 +167,7 @@ func ValidateConfig(v *viper.Viper, cfgfile string) error { // configsections["oldsources"] = config.OldSources if err := ValidateBySection(&config, configsections, cfgfile); err != nil { - POPExiter("Config \"%s\" is missing required attributes:\n%v\n", cfgfile, err) + return fmt.Errorf("config %q is missing required attributes: %w", cfgfile, err) } return nil } @@ -183,7 +184,7 @@ func ValidateBySection(config *Config, configsections map[string]interface{}, cf } if err := validate.Struct(data); err != nil { log.Printf("ValidateBySection: data that caused validation to fail:\n%v\n", data) - POPExiter("ValidateBySection: Config %s, section %s: missing required attributes:\n%v\n", cfgfile, k, err) + return fmt.Errorf("config %s, section %s: missing required attributes: %w", cfgfile, k, err) } } return nil diff --git a/configupdater.go b/configupdater.go index 255a649..5e6e7f4 100644 --- a/configupdater.go +++ b/configupdater.go @@ -2,114 +2,131 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "context" "encoding/json" + "fmt" "log" "github.com/dnstapir/tapir" "github.com/spf13/viper" ) -func (pd *PopData) ConfigUpdater(conf *Config, stopch chan struct{}) { +func (pd *PopData) ConfigUpdater(ctx context.Context, conf *Config) error { active := viper.GetBool("tapir.config.active") if !active { pd.Logger.Printf("*** ConfigUpdater: not active, skipping") - return + return nil } // Create a new mqtt engine just for the statusupdater. me := pd.MqttEngine if me == nil { - POPExiter("ConfigUpdater: MQTT Engine not running") + return fmt.Errorf("MQTT Engine not running") } ConfigChan := make(chan tapir.MqttPkgIn, 5) configTopic := viper.GetString("tapir.config.topic") if configTopic == "" { - POPExiter("ConfigUpdater: MQTT config topic not set") + return fmt.Errorf("MQTT config topic not set") } pd.Logger.Printf("ConfigUpdater: Adding sub topic '%s' to MQTT Engine", configTopic) err := me.SubToTopic(configTopic, ConfigChan, "struct", true) // XXX: Brr. kludge. if err != nil { - POPExiter("ConfigUpdater: Error adding topic %s to MQTT Engine: %v", configTopic, err) + return fmt.Errorf("error adding topic %s to MQTT Engine: %w", configTopic, err) } pd.Logger.Printf("ConfigUpdater: Topic status for MQTT engine %s", me.Creator) log.Printf("ConfigUpdater: Starting") - for inbox := range ConfigChan { - log.Printf("ConfigUpdater: got config update message on topic %s: %v", inbox.Topic) - var gconfig tapir.GlobalConfig - err = json.Unmarshal(inbox.Payload, &gconfig) - if err != nil { - log.Printf("ConfigUpdater: error unmarshalling config update message: %v", err) - continue - } - pd.ProcessTapirGlobalConfig(gconfig) - if err != nil { - log.Printf("ConfigUpdater: error processing config update message: %v", err) + for { + select { + case <-ctx.Done(): + log.Printf("ConfigUpdater: stopping") + return nil + case inbox := <-ConfigChan: + log.Printf("ConfigUpdater: got config update message on topic %s", inbox.Topic) + var gconfig tapir.GlobalConfig + err = json.Unmarshal(inbox.Payload, &gconfig) + if err != nil { + log.Printf("ConfigUpdater: error unmarshalling config update message: %v", err) + continue + } + err = pd.ProcessTapirGlobalConfig(gconfig) + if err != nil { + log.Printf("ConfigUpdater: error processing config update message: %v", err) + } } } } -func (pd *PopData) ProcessTapirGlobalConfig(gconfig tapir.GlobalConfig) { - log.Printf("TapirProcessGlobalConfig: %+v", gconfig) +func (pd *PopData) ProcessTapirGlobalConfig(gconfig tapir.GlobalConfig) error { + log.Printf("TapirProcessGlobalConfig: %+v", gconfig) + if len(gconfig.ObservationTopics) == 0 { + return fmt.Errorf("global config has no observation topics") + } + if pd.MqttEngine == nil { + return fmt.Errorf("MQTT Engine not running") + } - // Assume there is only one topic and that it is the one we want - // TODO maybe sanitize or sanity check or something - newTopic := gconfig.ObservationTopics[0] - bootstrapServers := gconfig.Bootstrap.Servers - bootstrapUrl := gconfig.Bootstrap.BaseUrl - bootstrapKey := gconfig.Bootstrap.ApiToken + // Assume there is only one topic and that it is the one we want + // TODO maybe sanitize or sanity check or something + newTopic := gconfig.ObservationTopics[0] + bootstrapServers := gconfig.Bootstrap.Servers + bootstrapUrl := gconfig.Bootstrap.BaseUrl + bootstrapKey := gconfig.Bootstrap.ApiToken //for _, listtype := range []string{"allowlist", "denylist", "doubtlist"} { for _, wbgl := range pd.Lists["doubtlist"] { - if wbgl.Immutable || wbgl.Datasource != "mqtt" { - continue - } + if wbgl.Immutable || wbgl.Datasource != "mqtt" { + continue + } - for _, topic := range wbgl.MqttDetails.Topics { - pd.MqttEngine.RemoveTopic(topic) - break // Only one topic - } + if len(wbgl.MqttDetails.Topics) > 0 { + topic := wbgl.MqttDetails.Topics[0] + if err := pd.MqttEngine.RemoveTopic(topic); err != nil { + pd.Logger.Printf("ProcessTapirGlobalConfig: error removing previous MQTT topic %q: %v", topic, err) + } + } pd.mu.Lock() - wbgl.MqttDetails.Topics = append(wbgl.MqttDetails.Topics, newTopic.Topic) - wbgl.MqttDetails.Bootstrap = bootstrapServers - wbgl.MqttDetails.BootstrapUrl = bootstrapUrl - wbgl.MqttDetails.BootstrapKey = bootstrapKey + wbgl.MqttDetails.Topics = append(wbgl.MqttDetails.Topics, newTopic.Topic) + wbgl.MqttDetails.Bootstrap = bootstrapServers + wbgl.MqttDetails.BootstrapUrl = bootstrapUrl + wbgl.MqttDetails.BootstrapKey = bootstrapKey pd.mu.Unlock() - err := pd.MqttEngine.SubToTopic(newTopic.Topic, pd.TapirObservations, "struct", true) // XXX: Brr. kludge. - if err != nil { - POPExiter("ProcessTapirGlobalConfig: Error adding topic %s: %v", newTopic, err) - } - - src := SourceConf{ - Bootstrap: wbgl.MqttDetails.Bootstrap, - BootstrapUrl: wbgl.MqttDetails.BootstrapUrl, - BootstrapKey: wbgl.MqttDetails.BootstrapKey, - Name: wbgl.Name, - Format: wbgl.Format, - } - - if len(gconfig.Bootstrap.Servers) > 0 { - pd.Logger.Printf("ProcessTapirGlobalConfig: %d bootstrap servers advertised: %v", wbgl.Name, len(src.Bootstrap), src.Bootstrap) - tmp, err := pd.BootstrapMqttSource(src) - if err != nil { - pd.Logger.Printf("ProcessTapirGlobalConfig: Error bootstrapping MQTT source %s: %v", wbgl.Name, err) - } else { - pd.mu.Lock() - *wbgl = *tmp - pd.mu.Unlock() - } - } + err := pd.MqttEngine.SubToTopic(newTopic.Topic, pd.TapirObservations, "struct", true) // XXX: Brr. kludge. + if err != nil { + return fmt.Errorf("error adding topic %s: %w", newTopic.Topic, err) + } + + src := SourceConf{ + Bootstrap: wbgl.MqttDetails.Bootstrap, + BootstrapUrl: wbgl.MqttDetails.BootstrapUrl, + BootstrapKey: wbgl.MqttDetails.BootstrapKey, + Name: wbgl.Name, + Format: wbgl.Format, + } + + if len(gconfig.Bootstrap.Servers) > 0 { + pd.Logger.Printf("ProcessTapirGlobalConfig: %s: %d bootstrap servers advertised: %v", wbgl.Name, len(src.Bootstrap), src.Bootstrap) + tmp, err := pd.BootstrapMqttSource(src) + if err != nil { + pd.Logger.Printf("ProcessTapirGlobalConfig: Error bootstrapping MQTT source %s: %v", wbgl.Name, err) + } else { + pd.mu.Lock() + *wbgl = *tmp + pd.mu.Unlock() + } + } pd.Logger.Printf("*** DONE Processing global config") - } + } + return nil } diff --git a/dnshandler.go b/dnshandler.go index 6760572..54b04d5 100644 --- a/dnshandler.go +++ b/dnshandler.go @@ -2,12 +2,15 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "context" + "fmt" "log" "net" "strings" + "sync" "github.com/miekg/dns" "github.com/spf13/viper" @@ -18,7 +21,7 @@ import ( // var RpzZones = make(map[string]*tapir.ZoneData, 5) // func DnsEngine(scannerq chan ScanRequest, updateq chan UpdateRequest) error { -func DnsEngine(conf *Config) error { +func DnsEngine(ctx context.Context, conf *Config) error { addresses := viper.GetStringSlice("dnsengine.addresses") // verbose := viper.GetBool("dnsengine.verbose") @@ -26,24 +29,53 @@ func DnsEngine(conf *Config) error { dns.HandleFunc(".", createHandler(conf)) conf.Loggers.Dnsengine.Printf("DnsEngine: addresses: %v", addresses) + errCh := make(chan error, len(addresses)*2) + var wg sync.WaitGroup + var servers []*dns.Server for _, addr := range addresses { - for _, net := range []string{"udp", "tcp"} { - go func(addr, net string) { - conf.Loggers.Dnsengine.Printf("DnsEngine: serving on %s (%s)\n", addr, net) - server := &dns.Server{Addr: addr, Net: net} + for _, proto := range []string{"udp", "tcp"} { + server := &dns.Server{Addr: addr, Net: proto} + servers = append(servers, server) + wg.Add(1) + go func(server *dns.Server) { + defer wg.Done() + conf.Loggers.Dnsengine.Printf("DnsEngine: serving on %s (%s)\n", server.Addr, server.Net) // Must bump the buffer size of incoming UDP msgs, as updates // may be much larger then queries server.UDPSize = dns.DefaultMsgSize // 4096 if err := server.ListenAndServe(); err != nil { - conf.Loggers.Dnsengine.Printf("Failed to setup the %s server: %s\n", net, err.Error()) + select { + case errCh <- fmt.Errorf("failed to setup the %s server on %s: %w", server.Net, server.Addr, err): + case <-ctx.Done(): + } } else { - conf.Loggers.Dnsengine.Printf("DnsEngine: listening on %s/%s\n", addr, net) + conf.Loggers.Dnsengine.Printf("DnsEngine: listening on %s/%s\n", server.Addr, server.Net) } - }(addr, net) + }(server) } } - return nil + if len(servers) == 0 { + return nil + } + + shutdownServers := func() { + for _, server := range servers { + if err := server.Shutdown(); err != nil { + conf.Loggers.Dnsengine.Printf("DnsEngine: error shutting down %s/%s: %v", server.Addr, server.Net, err) + } + } + wg.Wait() + } + + select { + case <-ctx.Done(): + shutdownServers() + return nil + case err := <-errCh: + shutdownServers() + return err + } } func createHandler(conf *Config) func(w dns.ResponseWriter, r *dns.Msg) { @@ -318,7 +350,7 @@ func QueryResponder(w dns.ResponseWriter, r *dns.Msg, zd *tapir.ZoneData, qname owner = zd.Owners[zd.OwnerIndex[qname]] default: - POPExiter("Error: QueryResponder: unknown zone type: %d", zd.ZoneType) + return fmt.Errorf("unknown zone type: %d", zd.ZoneType) } var glue *tapir.RRset diff --git a/lists.go b/lists.go index c8f696b..66026af 100644 --- a/lists.go +++ b/lists.go @@ -2,11 +2,10 @@ * Copyright (c) Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "fmt" - "log" // "github.com/smhanov/dawg" "github.com/dnstapir/tapir" @@ -66,7 +65,8 @@ func (pd *PopData) Doubtlisted(name string) bool { // case "trie": // return list.Trie.Search(name) != nil default: - log.Fatalf("Unknown doubtlist format %s", list.Format) + pd.Logger.Printf("Unknown doubtlist format %s for doubtlist %s", list.Format, list.Name) + continue } } return false diff --git a/logging.go b/logging.go index 52f5fce..a858f5d 100644 --- a/logging.go +++ b/logging.go @@ -2,7 +2,7 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "fmt" @@ -14,7 +14,7 @@ import ( "gopkg.in/natefinch/lumberjack.v2" ) -func SetupLogging(conf *Config) { +func SetupLogging(conf *Config) error { logfile := viper.GetString("log.file") debug := viper.GetString("log.mode") == "debug" @@ -35,7 +35,7 @@ func SetupLogging(conf *Config) { }) fmt.Printf("TAPIR-POP standard logging to: %s\n", logfile) } else { - POPExiter("Error: standard log (key log.file) not specified") + return fmt.Errorf("standard log (key log.file) not specified") } logfile = viper.GetString("policy.logfile") @@ -43,7 +43,7 @@ func SetupLogging(conf *Config) { logfile = filepath.Clean(logfile) f, err := os.OpenFile(logfile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644) // #nosec G302 if err != nil { - POPExiter("error opening TAPIR-POP policy logfile '%s': %v", logfile, err) + return fmt.Errorf("error opening TAPIR-POP policy logfile %q: %w", logfile, err) } if debug { @@ -67,7 +67,7 @@ func SetupLogging(conf *Config) { logfile = filepath.Clean(logfile) f, err := os.OpenFile(logfile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644) // #nosec G302 if err != nil { - POPExiter("error opening TAPIR-POP dnsengine logfile '%s': %v", logfile, err) + return fmt.Errorf("error opening TAPIR-POP dnsengine logfile %q: %w", logfile, err) } if debug { @@ -91,7 +91,7 @@ func SetupLogging(conf *Config) { logfile = filepath.Clean(logfile) f, err := os.OpenFile(logfile, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0644) // #nosec G302 if err != nil { - POPExiter("error opening TAPIR-POP MQTT logfile '%s': %v", logfile, err) + return fmt.Errorf("error opening TAPIR-POP MQTT logfile %q: %w", logfile, err) } if debug { @@ -109,4 +109,5 @@ func SetupLogging(conf *Config) { log.Println("No MQTT logfile specified, using default") conf.Loggers.Mqtt = log.Default() } + return nil } diff --git a/main.go b/main.go index 2e20676..d47a2a5 100644 --- a/main.go +++ b/main.go @@ -2,45 +2,40 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "context" + "errors" "fmt" + "io" "log" - "os" - "os/signal" - - "sync" - "syscall" + "strings" "time" + "github.com/dnstapir/tapir" "github.com/google/uuid" flag "github.com/spf13/pflag" "github.com/spf13/viper" - - "github.com/dnstapir/tapir" ) -/* Rewritten if building with make */ -var name = "BAD-BUILD" -var version = "BAD-BUILD" -var commit = "BAD-BUILD" - -var POPExiter = func(args ...interface{}) { - log.Printf("POPExiter: [placeholderfunction w/o real cleanup]") - log.Printf("POPExiter: Exit message: %s", fmt.Sprintf(args[0].(string), args[1:]...)) - os.Exit(1) +type RunOptions struct { + Name string + Version string + Commit string + Args []string + Stdout io.Writer + Stderr io.Writer + Reload <-chan struct{} } func (pd *PopData) SaveRpzSerial() error { - // Save the current value of pd.Downstreams.Serial to a text file serialFile := viper.GetString("services.rpz.serialcache") if serialFile == "" { - log.Fatalf("POPExiter:No serial cache file specified") + return fmt.Errorf("no serial cache file specified") } - // serialData := []byte(fmt.Sprintf("%d", pd.Rpz.CurrentSerial)) - // err := os.WriteFile(serialFile, serialData, 0644) + serialYaml := fmt.Sprintf("current_serial: %d\n", pd.Rpz.CurrentSerial) err := os.WriteFile(serialFile, []byte(serialYaml), 0644) // #nosec G306 if err != nil { @@ -51,203 +46,247 @@ func (pd *PopData) SaveRpzSerial() error { return err } -func mainloop(conf *Config, configfile *string, pd *PopData) { - log.Println("mainloop: enter") - exit := make(chan os.Signal, 1) - signal.Notify(exit, syscall.SIGINT, syscall.SIGTERM) - hupper := make(chan os.Signal, 1) - signal.Notify(hupper, syscall.SIGHUP) - - POPExiter = func(args ...interface{}) { - var msg string - log.Printf("POPExiter: will try to clean up.") +func Run(ctx context.Context, opts RunOptions) (runErr error) { + if ctx == nil { + ctx = context.Background() + } + stdout := opts.Stdout + if stdout == nil { + stdout = os.Stdout + } + stderr := opts.Stderr + if stderr == nil { + stderr = os.Stderr + } + name := opts.Name + if name == "" { + name = "dnstapir-pop" + } + version := opts.Version + if version == "" { + version = "BAD-BUILD" + } + commit := opts.Commit + if commit == "" { + commit = "BAD-BUILD" + } - err := pd.SaveRpzSerial() - if err != nil { - log.Printf("Error saving RPZ serial: %v", err) - } + if _, err := fmt.Fprintf(stdout, "%s (TAPIR Edge Manager) version %s (%s) starting.\n", name, version, commit); err != nil { + return fmt.Errorf("writing startup message: %w", err) + } - switch args[0].(type) { - case string: - msg = fmt.Sprintf("POPExiter: Exit message: %s", - fmt.Sprintf(args[0].(string), args[1:]...)) - case error: - msg = fmt.Sprintf("POPExiter: Error message: %s", args[0].(error).Error()) + mqttClientID := "tapir-pop-" + uuid.New().String() + fs := flag.NewFlagSet(name, flag.ContinueOnError) + fs.SetOutput(stderr) + fs.BoolVarP(&tapir.GlobalCF.Debug, "debug", "d", false, "Debug mode") + fs.BoolVarP(&tapir.GlobalCF.Verbose, "verbose", "v", false, "Verbose mode") + fs.StringVar(&mqttClientID, "client-id", mqttClientID, "MQTT client id, default is a random string") + if err := fs.Parse(opts.Args); err != nil { + return err + } - default: - msg = fmt.Sprintf("POPExiter: Exit message: %v", args[0]) - } + cfgFiles, err := loadConfigFiles(stderr) + if err != nil { + return err + } - fmt.Println(msg) - log.Println(msg) + if err := ValidateConfig(nil, strings.Join(cfgFiles, ", ")); err != nil { + return fmt.Errorf("error validating config: %w", err) + } - os.Exit(1) + var conf Config + if err := viper.Unmarshal(&conf); err != nil { + return fmt.Errorf("error unmarshalling config into struct: %w", err) } - var wg sync.WaitGroup - wg.Add(1) + if err := SetupLogging(&conf); err != nil { + return err + } - go func() { - for { - // log.Println("mainloop: signal dispatcher") - select { - case <-exit: - log.Println("mainloop: Exit signal received. Cleaning up.") - err := pd.SaveRpzSerial() - if err != nil { - log.Printf("Error saving RPZ serial: %v", err) - } - // do whatever we need to do to wrap up nicely - wg.Done() - case <-hupper: - // config file to use has already been set in main() - if err := viper.ReadInConfig(); err == nil { - fmt.Fprintln(os.Stderr, "Using config file:", *configfile) - } else { - POPExiter("Could not load config %s: Error: %v", *configfile, err) - } + statusch := make(chan tapir.ComponentStatusUpdate, 10) + conf.Internal.ComponentStatusCh = statusch + conf.Internal.APIStopCh = make(chan struct{}, 1) - log.Println("mainloop: SIGHUP received. Forcing refresh of all configured zones.") - log.Printf("mainloop: Requesting refresh of all RPZ zones") - conf.PopData.RpzRefreshCh <- RpzRefresh{Name: ""} - case <-conf.Internal.APIStopCh: - log.Printf("mainloop: API instruction to stop\n") - err := pd.SaveRpzSerial() - if err != nil { - log.Printf("Error saving RPZ serial: %v", err) - } - wg.Done() - } + pd, err := NewPopData(&conf, log.Default()) + if err != nil { + return fmt.Errorf("error from NewPopData: %w", err) + } + defer func() { + if cleanupErr := cleanupPopData(pd); cleanupErr != nil { + runErr = errors.Join(runErr, cleanupErr) } }() - wg.Wait() - - log.Println("mainloop: leaving signal dispatcher") -} -var Gconfig Config -var mqttclientid string + if pd.MqttEngine == nil { + pd.mu.Lock() + err := pd.CreateMqttEngine(mqttClientID, statusch, pd.MqttLogger) + pd.mu.Unlock() + if err != nil { + return fmt.Errorf("error creating MQTT Engine: %w", err) + } + if err := pd.StartMqttEngine(pd.MqttEngine); err != nil { + return fmt.Errorf("error starting MQTT Engine: %w", err) + } + } + ctx, cancel := context.WithCancel(ctx) + defer cancel() -func main() { - fmt.Printf("%s (TAPIR Edge Manager) version %s (%s) starting.\n", name, version, commit) - // var conf Config - mqttclientid = "tapir-pop-" + uuid.New().String() - flag.BoolVarP(&tapir.GlobalCF.Debug, "debug", "d", false, "Debug mode") - flag.BoolVarP(&tapir.GlobalCF.Verbose, "verbose", "v", false, "Verbose mode") - flag.StringVarP(&mqttclientid, "client-id", "", mqttclientid, "MQTT client id, default is a random string") + workerErrCh := make(chan error, 5) + startWorker := func(name string, fn func(context.Context) error) { + go func() { + if err := fn(ctx); err != nil && !errors.Is(err, context.Canceled) { + select { + case workerErrCh <- fmt.Errorf("%s: %w", name, err): + case <-ctx.Done(): + } + } + }() + } - flag.Parse() + startWorker("config updater", func(ctx context.Context) error { + return pd.ConfigUpdater(ctx, &conf) + }) + startWorker("status updater", func(ctx context.Context) error { + return pd.StatusUpdater(ctx, &conf) + }) + startWorker("refresh engine", func(ctx context.Context) error { + return pd.RefreshEngine(ctx, &conf) + }) - var cfgFileUsed string + log.Println("*** main: Calling ParseSourcesNG()") + if err := pd.ParseSourcesNG(); err != nil { + return fmt.Errorf("error from ParseSourcesNG: %w", err) + } + log.Println("*** main: Returned from ParseSourcesNG()") - var cfgFile string - if cfgFile != "" { - viper.SetConfigFile(cfgFile) - } else { - viper.SetConfigFile(tapir.DefaultPopCfgFile) + if err := pd.ParseOutputs(); err != nil { + return fmt.Errorf("error from ParseOutputs: %w", err) } - viper.AutomaticEnv() // read in environment variables that match + startWorker("api handler", func(ctx context.Context) error { + return APIhandler(ctx, &conf) + }) + startWorker("dns engine", func(ctx context.Context) error { + return DnsEngine(ctx, &conf) + }) - // If a config file is found, read it in. - if err := viper.ReadInConfig(); err == nil { - fmt.Fprintln(os.Stderr, "Using config file:", viper.ConfigFileUsed()) - cfgFileUsed = viper.ConfigFileUsed() - } else { - POPExiter("Could not load config %s: Error: %v", tapir.DefaultPopCfgFile, err) + conf.BootTime = time.Now() + statusch <- tapir.ComponentStatusUpdate{ + Component: "main-boot", + Status: tapir.StatusOK, + Msg: "TAPIR Policy Processor started", + TimeStamp: time.Now(), } - viper.SetConfigFile(tapir.PopSourcesCfgFile) - if err := viper.MergeInConfig(); err == nil { - fmt.Fprintln(os.Stderr, "Using config file:", viper.ConfigFileUsed()) - cfgFileUsed = viper.ConfigFileUsed() - } else { - POPExiter("Could not load config %s: Error: %v", tapir.PopSourcesCfgFile, err) + + return runLoop(ctx, &conf, cfgFiles, opts.Reload, workerErrCh, stderr) +} + +func loadConfigFiles(stderr io.Writer) ([]string, error) { + viper.Reset() + viper.SetConfigFile(tapir.DefaultPopCfgFile) + viper.AutomaticEnv() + + if err := viper.ReadInConfig(); err != nil { + return nil, fmt.Errorf("could not load config %s: %w", tapir.DefaultPopCfgFile, err) } - viper.SetConfigFile(tapir.PopOutputsCfgFile) - if err := viper.MergeInConfig(); err == nil { - fmt.Fprintln(os.Stderr, "Using config file:", viper.ConfigFileUsed()) - cfgFileUsed = viper.ConfigFileUsed() - } else { - POPExiter("Could not load config %s: Error: %v", tapir.PopOutputsCfgFile, err) + cfgFiles := []string{viper.ConfigFileUsed()} + if err := printConfigFileUsed(stderr, cfgFiles[0]); err != nil { + return nil, err } - viper.SetConfigFile(tapir.PopPolicyCfgFile) - if err := viper.MergeInConfig(); err == nil { - fmt.Fprintln(os.Stderr, "Using config file:", viper.ConfigFileUsed()) - cfgFileUsed = viper.ConfigFileUsed() - } else { - POPExiter("Could not load config %s: Error: %v", tapir.PopPolicyCfgFile, err) + + for _, cfgFile := range []string{ + tapir.PopSourcesCfgFile, + tapir.PopOutputsCfgFile, + tapir.PopPolicyCfgFile, + } { + viper.SetConfigFile(cfgFile) + if err := viper.MergeInConfig(); err != nil { + return nil, fmt.Errorf("could not load config %s: %w", cfgFile, err) + } + cfgFiles = append(cfgFiles, viper.ConfigFileUsed()) + if err := printConfigFileUsed(stderr, viper.ConfigFileUsed()); err != nil { + return nil, err + } } - SetupLogging(&Gconfig) + return cfgFiles, nil +} - err := ValidateConfig(nil, cfgFileUsed) // will terminate on error - if err != nil { - POPExiter("Error validating config: %v", err) +func runLoop(ctx context.Context, conf *Config, cfgFiles []string, reload <-chan struct{}, workerErrCh <-chan error, stderr io.Writer) error { + log.Println("mainloop: enter") + defer log.Println("mainloop: leaving signal dispatcher") + + for { + select { + case <-ctx.Done(): + log.Println("mainloop: context cancelled. Cleaning up.") + return nil + case err := <-workerErrCh: + if err != nil { + return err + } + case <-reload: + if err := reloadConfig(cfgFiles, stderr); err != nil { + return err + } + log.Println("mainloop: SIGHUP received. Forcing refresh of all configured zones.") + log.Printf("mainloop: Requesting refresh of all RPZ zones") + conf.PopData.RpzRefreshCh <- RpzRefresh{Name: ""} + case <-conf.Internal.APIStopCh: + log.Printf("mainloop: API instruction to stop\n") + return nil + } } +} - err = viper.Unmarshal(&Gconfig) - if err != nil { - POPExiter("Error unmarshalling config into struct: %v", err) +func reloadConfig(cfgFiles []string, stderr io.Writer) error { + if len(cfgFiles) == 0 { + return fmt.Errorf("no config files to reload") } + viper.Reset() + viper.SetConfigFile(cfgFiles[0]) + viper.AutomaticEnv() - var stopch = make(chan struct{}, 10) - - statusch := make(chan tapir.ComponentStatusUpdate, 10) - Gconfig.Internal.ComponentStatusCh = statusch - - pd, err := NewPopData(&Gconfig, log.Default()) - if err != nil { - POPExiter("Error from NewPopData: %v", err) + if err := viper.ReadInConfig(); err != nil { + return fmt.Errorf("could not load config %s: %w", cfgFiles[0], err) + } + if err := printConfigFileUsed(stderr, viper.ConfigFileUsed()); err != nil { + return err } - if pd.MqttEngine == nil { - pd.mu.Lock() - err := pd.CreateMqttEngine(mqttclientid, statusch, pd.MqttLogger) - if err != nil { - POPExiter("Error creating MQTT Engine: %v", err) + for _, cfgFile := range cfgFiles[1:] { + viper.SetConfigFile(cfgFile) + if err := viper.MergeInConfig(); err != nil { + return fmt.Errorf("could not load config %s: %w", cfgFile, err) } - pd.mu.Unlock() - err = pd.StartMqttEngine(pd.MqttEngine) - if err != nil { - POPExiter("Error starting MQTT Engine: %v", err) + if err := printConfigFileUsed(stderr, viper.ConfigFileUsed()); err != nil { + return err } } + return nil +} - go pd.ConfigUpdater(&Gconfig, stopch) // Note that ConfigUpdater must as early as possible - go pd.StatusUpdater(&Gconfig, stopch) // Note that StatusUpdater must as early as possible - go pd.RefreshEngine(&Gconfig, stopch) - - log.Println("*** main: Calling ParseSourcesNG()") - err = pd.ParseSourcesNG() - if err != nil { - POPExiter("Error from ParseSourcesNG: %v", err) +func printConfigFileUsed(stderr io.Writer, cfgFile string) error { + if _, err := fmt.Fprintln(stderr, "Using config file:", cfgFile); err != nil { + return fmt.Errorf("writing config file notice: %w", err) } - log.Println("*** main: Returned from ParseSourcesNG()") + return nil +} - err = pd.ParseOutputs() - if err != nil { - POPExiter("Error from ParseOutputs: %v", err) +func cleanupPopData(pd *PopData) error { + if pd == nil { + return nil } - apistopper := make(chan struct{}) // - Gconfig.Internal.APIStopCh = apistopper - go APIhandler(&Gconfig, apistopper) - // go httpsserver(&conf, apistopper) - - go func() { - if err := DnsEngine(&Gconfig); err != nil { - log.Printf("Error starting DnsEngine: %v", err) + var errs []error + if err := pd.SaveRpzSerial(); err != nil { + errs = append(errs, fmt.Errorf("error saving RPZ serial: %w", err)) + } + if pd.MqttEngine != nil && pd.TapirMqttEngineRunning { + if _, err := pd.MqttEngine.StopEngine(); err != nil { + errs = append(errs, fmt.Errorf("error stopping MQTT Engine: %w", err)) } - }() - Gconfig.BootTime = time.Now() - - statusch <- tapir.ComponentStatusUpdate{ - Component: "main-boot", - Status: tapir.StatusOK, - Msg: "TAPIR Policy Processor started", - TimeStamp: time.Now(), } - - mainloop(&Gconfig, &cfgFileUsed, pd) + return errors.Join(errs...) } diff --git a/mqtt.go b/mqtt.go index 5c067ae..c50df15 100644 --- a/mqtt.go +++ b/mqtt.go @@ -2,7 +2,7 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "fmt" @@ -15,13 +15,13 @@ import ( func (pd *PopData) CreateMqttEngine(clientid string, statusch chan tapir.ComponentStatusUpdate, lg *log.Logger) error { if clientid == "" { - POPExiter("Error starting MQTT Engine: clientid not specified in config") + return fmt.Errorf("clientid not specified in config") } var err error pd.Logger.Printf("Creating MQTT Engine with clientid %s", clientid) pd.MqttEngine, err = tapir.NewMqttEngine("tapir-pop", clientid, tapir.TapirSub, statusch, lg) // sub, but no pub if err != nil { - POPExiter("Error from NewMqttEngine: %v\n", err) + return fmt.Errorf("NewMqttEngine: %w", err) } return nil } @@ -30,17 +30,19 @@ func (pd *PopData) StartMqttEngine(meng *tapir.MqttEngine) error { if pd.TapirMqttEngineRunning { return nil } + if meng == nil { + return fmt.Errorf("MQTT engine is nil") + } cmnder, outbox, inbox, err := meng.StartEngine() if err != nil { - log.Fatalf("Error from StartEngine(): %v", err) + return fmt.Errorf("StartEngine: %w", err) } pd.TapirMqttCmdCh = cmnder pd.TapirMqttPubCh = outbox pd.TapirObservations = inbox pd.TapirMqttEngineRunning = true - meng.SetupInterruptHandler() return nil } diff --git a/policy.go b/policy.go index 241e74b..0ddb4e0 100644 --- a/policy.go +++ b/policy.go @@ -2,9 +2,10 @@ * Copyright (c) 2024 Johan Stenstam, joahn.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "fmt" "log" "net" "os" @@ -34,7 +35,7 @@ func (pd *PopData) ParseOutputs() error { pd.Logger.Printf("ParseOutputs: reading outputs from %s", tapir.PopOutputsCfgFile) cfgdata, err := os.ReadFile(tapir.PopOutputsCfgFile) if err != nil { - log.Fatalf("Error from ReadFile(%s): %v", tapir.PopOutputsCfgFile, err) + return fmt.Errorf("ReadFile(%s): %w", tapir.PopOutputsCfgFile, err) } var oconf = PopOutputs{ @@ -44,7 +45,7 @@ func (pd *PopData) ParseOutputs() error { // pd.Logger.Printf("ParseOutputs: config read: %s", cfgdata) err = yaml.Unmarshal(cfgdata, &oconf) if err != nil { - log.Fatalf("Error from yaml.Unmarshal(OutputsConfig): %v", err) + return fmt.Errorf("yaml.Unmarshal(OutputsConfig): %w", err) } pd.Logger.Printf("ParseOutputs: found %d outputs", len(oconf.Outputs)) @@ -105,7 +106,7 @@ func (pd *PopData) ParseOutputs() error { // Note: we onlygethere when we know that this name is only doubtlisted // so no need tocheckfor allow- or denylisting -func (pd *PopData) ComputeRpzDoubtlistAction(name string) tapir.Action { +func (pd *PopData) ComputeRpzDoubtlistAction(name string) (tapir.Action, error) { var doubtHits = map[string]*tapir.TapirName{} for listname, list := range pd.Lists["doubtlist"] { @@ -121,13 +122,13 @@ func (pd *PopData) ComputeRpzDoubtlistAction(name string) tapir.Action { // doubtHits = append(doubtHits, v) // } default: - POPExiter("Unknown doubtlist format %s", list.Format) + return tapir.ALLOWLIST, fmt.Errorf("unknown doubtlist format %s", list.Format) } } if len(doubtHits) >= pd.Policy.Doubtlist.NumSources { pd.Policy.Logger.Printf("ComputeRpzDoubtlistAction: name %s is in %d or more sources, action is %s", name, pd.Policy.Doubtlist.NumSources, tapir.ActionToString[pd.Policy.Doubtlist.NumSourcesAction]) - return pd.Policy.Doubtlist.NumSourcesAction + return pd.Policy.Doubtlist.NumSourcesAction, nil } pd.Policy.Logger.Printf("ComputeRpzDoubtlistAction: name %s is in %d sources, not enough for action", name, len(doubtHits)) @@ -136,13 +137,13 @@ func (pd *PopData) ComputeRpzDoubtlistAction(name string) tapir.Action { if numtapirtags >= pd.Policy.Doubtlist.NumTapirTags { pd.Policy.Logger.Printf("ComputeRpzDoubtlistAction: name %s has more than %d tapir tags, action is %s", name, pd.Policy.Doubtlist.NumTapirTags, tapir.ActionToString[pd.Policy.Doubtlist.NumTapirTagsAction]) - return pd.Policy.Doubtlist.NumTapirTagsAction + return pd.Policy.Doubtlist.NumTapirTagsAction, nil } pd.Policy.Logger.Printf("ComputeRpzDoubtlistAction: name %s has %d tapir tags, not enough for action", name, numtapirtags) } pd.Policy.Logger.Printf("ComputeRpzDoubtlistAction: name %s is present in %d doubtlists, but does not trigger any action", name, len(doubtHits)) - return pd.Policy.AllowlistAction + return pd.Policy.AllowlistAction, nil } // Decision to block a doubtlisted name: @@ -166,22 +167,22 @@ func ApplyDoubtPolicy(name string, v *tapir.TapirName) string { return rpzaction } -func (pd *PopData) ComputeRpzAction(name string) tapir.Action { +func (pd *PopData) ComputeRpzAction(name string) (tapir.Action, error) { if pd.Allowlisted(name) { if pd.Debug { pd.Policy.Logger.Printf("ComputeRpzAction: name %s is doubtlisted, action is %s", name, tapir.ActionToString[pd.Policy.AllowlistAction]) } - return pd.Policy.AllowlistAction + return pd.Policy.AllowlistAction, nil } else if pd.Denylisted(name) { if pd.Debug { pd.Policy.Logger.Printf("ComputeRpzAction: name %s is denylisted, action is %s", name, tapir.ActionToString[pd.Policy.DenylistAction]) } - return pd.Policy.DenylistAction + return pd.Policy.DenylistAction, nil } else if pd.Doubtlisted(name) { if pd.Debug { pd.Policy.Logger.Printf("ComputeRpzAction: name %s is doubtlisted, needs further evaluation to determine action", name) } return pd.ComputeRpzDoubtlistAction(name) // This is not complete, only a placeholder for now. } - return tapir.ALLOWLIST + return tapir.ALLOWLIST, nil } diff --git a/reaper.go b/reaper.go index 197f7f8..009935b 100644 --- a/reaper.go +++ b/reaper.go @@ -1,7 +1,7 @@ /* * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "time" diff --git a/refreshengine.go b/refreshengine.go index 8d4752e..7976fab 100644 --- a/refreshengine.go +++ b/refreshengine.go @@ -1,9 +1,10 @@ /* * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "context" "encoding/json" "fmt" "log" @@ -44,7 +45,11 @@ type RefreshCounter struct { Downstreams []string } -func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { +// RefreshEngineInactiveError is the stable RpzCmdResponse.ErrorMsg prefix +// returned when an RPZ command is rejected because the refresh engine is disabled. +const RefreshEngineInactiveError = "refresh engine inactive" + +func (pd *PopData) RefreshEngine(ctx context.Context, conf *Config) error { var ObservationsCh = pd.TapirObservations @@ -53,9 +58,11 @@ func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { var refreshCounters = make(map[string]*RefreshCounter, 5) refreshTicker := time.NewTicker(1 * time.Second) + defer refreshTicker.Stop() reaperStart := time.Now().Truncate(pd.ReaperInterval).Add(pd.ReaperInterval) reaperTicker := time.NewTicker(pd.ReaperInterval) + defer reaperTicker.Stop() go func() { time.Sleep(time.Until(reaperStart)) @@ -64,9 +71,20 @@ func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { if !viper.GetBool("services.refreshengine.active") { log.Printf("Refresh Engine is NOT active. Zones will only be updated on receipt on Notifies.") - for range zonerefch { - // ensure that we keep reading to keep the channel open - continue + for { + select { + case <-ctx.Done(): + return nil + case <-zonerefch: + // ensure that we keep reading to keep the channel open + continue + case cmd, ok := <-rpzcmdch: + if !ok { + rpzcmdch = nil + continue + } + respondRefreshEngineInactive(cmd) + } } } else { log.Printf("RefreshEngine: Starting") @@ -88,6 +106,9 @@ func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { for { select { + case <-ctx.Done(): + log.Printf("RefreshEngine: stopping") + return nil case tpkg = <-ObservationsCh: tm := tapir.TapirMsg{} err := json.Unmarshal(tpkg.Payload, &tm) @@ -101,41 +122,24 @@ func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { tm.SrcName, len(tm.Added), len(tm.Removed)) _, err := pd.ProcessTapirUpdate(tm) if err != nil { - Gconfig.Internal.ComponentStatusCh <- tapir.ComponentStatusUpdate{ + pd.ComponentStatusCh <- tapir.ComponentStatusUpdate{ Status: tapir.StatusFail, Component: "tapir-observation", Msg: fmt.Sprintf("ProcessTapirUpdate error: %v", err), } log.Printf("RefreshEngine: Error from ProcessTapirUpdate(): %v", err) - } - Gconfig.Internal.ComponentStatusCh <- tapir.ComponentStatusUpdate{ - Status: tapir.StatusOK, - Component: "tapir-observation", - Msg: fmt.Sprintf("ProcessTapirUpdate: MQTT observation message received"), + } else { + pd.ComponentStatusCh <- tapir.ComponentStatusUpdate{ + Status: tapir.StatusOK, + Component: "tapir-observation", + Msg: "ProcessTapirUpdate: MQTT observation message received", + } } log.Printf("RefreshEngine: Tapir Observation update evaluated.") - // case "global-config": - // if !strings.HasSuffix(tpkg.Topic, "config") { - // log.Printf("RefreshEngine: received global-config message on wrong topic: %s. Ignored", tpkg.Topic) - // Gconfig.Internal.ComponentStatusCh <- tapir.ComponentStatusUpdate{ - // Status: "fail", - // Component: "mqtt-config", - // Msg: fmt.Sprintf("RefreshEngine: received global-config message on wrong topic: %s. Ignored", tpkg.Topic), - // } - // continue - // } - // pd.ProcessTapirGlobalConfig(tm) - // log.Printf("RefreshEngine: Tapir Global Config evaluated.") - // Gconfig.Internal.ComponentStatusCh <- tapir.ComponentStatusUpdate{ - // Status: "ok", - // Component: "mqtt-config", - // Msg: fmt.Sprintf("RefreshEngine: Tapir Global Config evaluated."), - // } - default: log.Printf("RefreshEngine: Tapir Message: unknown msg type: %s", tm.MsgType) - Gconfig.Internal.ComponentStatusCh <- tapir.ComponentStatusUpdate{ + pd.ComponentStatusCh <- tapir.ComponentStatusUpdate{ Status: tapir.StatusFail, Component: "mqtt-unknown", Msg: fmt.Sprintf("RefreshEngine: Tapir Message: unknown msg type: %s", tm.MsgType), @@ -272,11 +276,10 @@ func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { rc.CurRefresh-- if rc.CurRefresh <= 0 { upstream = rc.Upstream - // if rc.RRKeepFunc == nil { - // panic("RefreshEngine: keepfunc=nil") - // } if rc.RRParseFunc == nil { - panic("RefreshEngine: parsefunc=nil") + log.Printf("RefreshEngine: zone %s has nil RRParseFunc, removing refresh counter", zone) + delete(refreshCounters, zone) + continue } log.Printf("RefreshEngine: will refresh zone %s due to refresh counter", zone) @@ -333,12 +336,12 @@ func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { } resp.Msg = fmt.Sprintf("Zone %s: bumped serial from %d to %d. Notified downstreams: %v", zone, resp.OldSerial, resp.NewSerial, rc.Downstreams) - log.Printf(resp.Msg) + log.Print(resp.Msg) resp.Status = true } else { resp.Error = true resp.ErrorMsg = fmt.Sprintf("Request to bump serial for unknown zone '%s'", zone) - log.Printf(resp.ErrorMsg) + log.Print(resp.ErrorMsg) } } cmd.Result <- resp @@ -440,6 +443,27 @@ func (pd *PopData) RefreshEngine(conf *Config, stopch chan struct{}) { } } +func respondRefreshEngineInactive(cmd RpzCmdData) { + resp := RpzCmdResponse{ + Time: time.Now(), + Zone: cmd.Zone, + Domain: cmd.Domain, + Error: true, + ErrorMsg: fmt.Sprintf("%s: command %q rejected", RefreshEngineInactiveError, cmd.Command), + } + if cmd.Result == nil { + log.Printf("RefreshEngine: dropping inactive-engine response for command %q with nil result channel", cmd.Command) + return + } + + defer func() { + if r := recover(); r != nil { + log.Printf("RefreshEngine: dropping inactive-engine response for command %q on closed result channel: %v", cmd.Command, r) + } + }() + cmd.Result <- resp +} + func (pd *PopData) NotifyDownstreams() error { pd.Logger.Printf("RefreshEngine: Notifying %d downstreams for RPZ zone %s", len(pd.Downstreams), pd.Rpz.ZoneName) for _, d := range pd.Downstreams { @@ -460,27 +484,27 @@ func (pd *PopData) NotifyDownstreams() error { if err != nil { // well, we tried csu.Msg = fmt.Sprintf("Error from downstream %s on NOTIFY(%s): %v", dest, pd.Rpz.ZoneName, err) - Gconfig.Internal.ComponentStatusCh <- csu + pd.ComponentStatusCh <- csu pd.Logger.Println(csu.Msg) continue } if r.Opcode != dns.OpcodeNotify { // well, we tried csu.Msg = fmt.Sprintf("Error: not a NOTIFY response from downstream %s on NOTIFY(%s): %s", dest, pd.Rpz.ZoneName, dns.OpcodeToString[r.Opcode]) - Gconfig.Internal.ComponentStatusCh <- csu + pd.ComponentStatusCh <- csu pd.Logger.Println(csu.Msg) continue } else { if r.Rcode != dns.RcodeSuccess { csu.Msg = fmt.Sprintf("Downstream %s responded with rcode %s to NOTIFY(%s) about new SOA serial (%d)", dest, dns.RcodeToString[r.Rcode], pd.Rpz.ZoneName, pd.Rpz.Axfr.SOA.Serial) - Gconfig.Internal.ComponentStatusCh <- csu + pd.ComponentStatusCh <- csu pd.Logger.Println(csu.Msg) continue } csu.Status = tapir.StatusOK csu.Msg = fmt.Sprintf("Downstream %s responded correctly to NOTIFY(%s) about new SOA serial (%d)", dest, pd.Rpz.ZoneName, pd.Rpz.Axfr.SOA.Serial) - Gconfig.Internal.ComponentStatusCh <- csu + pd.ComponentStatusCh <- csu pd.Logger.Println(csu.Msg) } } diff --git a/rpz.go b/rpz.go index b159b98..e14c6a3 100644 --- a/rpz.go +++ b/rpz.go @@ -2,7 +2,7 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "github.com/dnstapir/tapir" @@ -67,7 +67,10 @@ func (pd *PopData) GenerateRpzAxfr() error { // pd.Logger.Printf("Doubtlisted name %s is also allowlisted. Dropped from output.", k) } else { // pd.Logger.Printf("Doubtlisted name %s is not allowlisted. Evalutate inclusion in output.", k) - action := pd.ComputeRpzAction(k) + action, err := pd.ComputeRpzAction(k) + if err != nil { + return err + } if action == tapir.ALLOWLIST { // pd.Logger.Printf("Doubtlisted name %s is not included in output.", k) } else { @@ -186,7 +189,10 @@ func (pd *PopData) GenerateRpzIxfr(data *tapir.TapirMsg) (RpzIxfr, error) { tn.Name = dns.Fqdn(tn.Name) pd.Policy.Logger.Printf("GenerateRpzIxfr: evaluating removed name %s", tn.Name) if cur, exist := pd.Rpz.Axfr.Data[tn.Name]; exist { - newAction := pd.ComputeRpzAction(tn.Name) + newAction, err := pd.ComputeRpzAction(tn.Name) + if err != nil { + return RpzIxfr{}, err + } oldAction := cur.Action if newAction != oldAction { if pd.Debug { @@ -232,7 +238,10 @@ func (pd *PopData) GenerateRpzIxfr(data *tapir.TapirMsg) (RpzIxfr, error) { tn.Name = dns.Fqdn(tn.Name) pd.Policy.Logger.Printf("GenerateRpzIxfr: evaluating added name %s", tn.Name) addtorpz = false - newAction := pd.ComputeRpzAction(tn.Name) + newAction, err := pd.ComputeRpzAction(tn.Name) + if err != nil { + return RpzIxfr{}, err + } if cur, exist := pd.Rpz.Axfr.Data[tn.Name]; exist { if newAction == tapir.ALLOWLIST { // delete from rpz diff --git a/sources.go b/sources.go index e556917..88f8643 100644 --- a/sources.go +++ b/sources.go @@ -1,9 +1,10 @@ /* * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "errors" "fmt" "log" "os" @@ -55,7 +56,7 @@ func NewPopData(conf *Config, lg *log.Logger) (*PopData, error) { err := pd.ParseOutputs() if err != nil { - POPExiter("NewPopData: Error from ParseOutputs(): %v", err) + return nil, fmt.Errorf("ParseOutputs: %w", err) } // pd.Rpz.IxfrChain = map[uint32]RpzIxfr{} @@ -69,42 +70,41 @@ func NewPopData(conf *Config, lg *log.Logger) (*PopData, error) { pd.Policy.Logger = conf.Loggers.Policy pd.Policy.AllowlistAction, err = tapir.StringToAction(viper.GetString("policy.allowlist.action")) if err != nil { - POPExiter("Error parsing allowlist policy: %v", err) + return nil, fmt.Errorf("error parsing allowlist policy: %w", err) } pd.Policy.DenylistAction, err = tapir.StringToAction(viper.GetString("policy.denylist.action")) if err != nil { - POPExiter("Error parsing denylist policy: %v", err) + return nil, fmt.Errorf("error parsing denylist policy: %w", err) } pd.Policy.Doubtlist.NumSources = viper.GetInt("policy.doubtlist.numsources.limit") if pd.Policy.Doubtlist.NumSources == 0 { - //nolint:typecheck - POPExiter("Error parsing policy: doubtlist.numsources.limit cannot be 0") + return nil, fmt.Errorf("error parsing policy: doubtlist.numsources.limit cannot be 0") } pd.Policy.Doubtlist.NumSourcesAction, err = tapir.StringToAction(viper.GetString("policy.doubtlist.numsources.action")) if err != nil { - POPExiter("Error parsing policy: %v", err) + return nil, fmt.Errorf("error parsing policy: %w", err) } pd.Policy.Doubtlist.NumTapirTags = viper.GetInt("policy.doubtlist.numtapirtags.limit") if pd.Policy.Doubtlist.NumTapirTags == 0 { - POPExiter("Error parsing policy: doubtlist.numtapirtags.limit cannot be 0") + return nil, fmt.Errorf("error parsing policy: doubtlist.numtapirtags.limit cannot be 0") } pd.Policy.Doubtlist.NumTapirTagsAction, err = tapir.StringToAction(viper.GetString("policy.doubtlist.numtapirtags.action")) if err != nil { - POPExiter("Error parsing policy: %v", err) + return nil, fmt.Errorf("error parsing policy: %w", err) } tmp := viper.GetStringSlice("policy.doubtlist.denytapir.tags") pd.Policy.Doubtlist.DenyTapirTags, err = tapir.StringsToTagMask(tmp) if err != nil { - POPExiter("Error parsing policy: %v", err) + return nil, fmt.Errorf("error parsing policy: %w", err) } pd.Policy.Doubtlist.DenyTapirAction, err = tapir.StringToAction(viper.GetString("policy.doubtlist.denytapir.action")) if err != nil { - POPExiter("Error parsing policy: %v", err) + return nil, fmt.Errorf("error parsing policy: %w", err) } // Note: We can not parse data sources here, as RefreshEngine has not yet started. @@ -159,7 +159,11 @@ func (pd *PopData) ParseSourcesNG() error { threads := 0 - var rptchan = make(chan string, 5) + type sourceResult struct { + name string + err error + } + resultCh := make(chan sourceResult, len(srcs)) for name, src := range srcs { if !*src.Active { @@ -169,15 +173,13 @@ func (pd *PopData) ParseSourcesNG() error { if pd.Debug { pd.Logger.Printf("=== ParseSourcesNG: Source: %s (%s) will be used (list type %s)", name, src.Name, src.Type) } - - var err error - threads++ go func(name string, src SourceConf, thread int) { - // defer func() { - //pd.Logger.Printf("<--Thread %d: source \"%s\" (%s) is now complete. %d remaining", thread, name, src.Source, threads) - // }() + var err error + defer func() { + resultCh <- sourceResult{name: name, err: err} + }() pd.Logger.Printf("-->Thread %d: parsing source \"%s\" (source %s)", thread, name, src.Source) newsource := tapir.WBGlist{ @@ -196,25 +198,30 @@ func (pd *PopData) ParseSourcesNG() error { pd.Logger.Printf("ParseSourcesNG: thread %d working on source \"%s\" (%s)", thread, name, src.Source) switch src.Source { case "mqtt": + if pd.MqttEngine == nil { + err = fmt.Errorf("MQTT Engine not configured") + return + } if pd.Debug { pd.Logger.Printf("ParseSourcesNG: Fetching MQTT validator key for topic %s", src.Topic) } pd.Logger.Printf("ParseSourcesNG: Adding topic '%s' to MQTT Engine", src.Topic) - err := pd.MqttEngine.SubToTopic(src.Topic, pd.TapirObservations, "struct", true) // XXX: Brr. kludge. + err = pd.MqttEngine.SubToTopic(src.Topic, pd.TapirObservations, "struct", true) // XXX: Brr. kludge. if err != nil { - POPExiter("Error adding topic %s to MQTT Engine: %v", src.Topic, err) + err = fmt.Errorf("error adding topic %s to MQTT Engine: %w", src.Topic, err) + return } pd.Logger.Printf("ParseSourcesNG: Topic data for topic %s", src.Topic) - mqttDetails := tapir.MqttDetails{ - Topics: []string{src.Topic}, - Bootstrap: src.Bootstrap, - BootstrapUrl: src.BootstrapUrl, - BootstrapKey: src.BootstrapKey, - } - newsource.MqttDetails = &mqttDetails - newsource.Immutable = src.Immutable + mqttDetails := tapir.MqttDetails{ + Topics: []string{src.Topic}, + Bootstrap: src.Bootstrap, + BootstrapUrl: src.BootstrapUrl, + BootstrapKey: src.BootstrapKey, + } + newsource.MqttDetails = &mqttDetails + newsource.Immutable = src.Immutable newsource.Format = "map" // for now if len(src.Bootstrap) > 0 { @@ -231,14 +238,13 @@ func (pd *PopData) ParseSourcesNG() error { pd.Logger.Printf("Created list [doubtlist][%s]", newsource.Name) pd.mu.Unlock() pd.Logger.Printf("*** MQTT sources are only managed via RefreshEngine.") - rptchan <- name case "file": - err = pd.ParseLocalFile(name, &newsource, rptchan) + err = pd.ParseLocalFile(name, &newsource) case "xfr": - err = pd.ParseRpzFeed(name, &newsource, rptchan) - pd.Logger.Printf("Thread %d: source \"%s\" now returned from ParseRpzFeed(). %d remaining", thread, name, threads) + err = pd.ParseRpzFeed(name, &newsource) + pd.Logger.Printf("Thread %d: source %q now returned from ParseRpzFeed()", thread, name) default: - pd.Logger.Printf("*** ParseSourcesNG: Error: unhandled source type %s", src.Source) + err = fmt.Errorf("unhandled source type %s", src.Source) } if err != nil { log.Printf("Error parsing source %s (datasource %s): %v", @@ -247,19 +253,23 @@ func (pd *PopData) ParseSourcesNG() error { }(name, src, threads) } - for { - if threads == 0 { - break - } - tmp := <-rptchan + var errs []error + for threads > 0 { + result := <-resultCh threads-- - pd.Logger.Printf("ParseSources: source \"%s\" is now complete. %d remaining", tmp, threads) + if result.err != nil { + errs = append(errs, fmt.Errorf("source %s: %w", result.name, result.err)) + } + pd.Logger.Printf("ParseSources: source \"%s\" is now complete. %d remaining", result.name, threads) + } + if len(errs) > 0 { + return errors.Join(errs...) } if pd.MqttEngine != nil && !pd.TapirMqttEngineRunning { err := pd.StartMqttEngine(pd.MqttEngine) if err != nil { - POPExiter("Error starting MQTT Engine: %v", err) + return fmt.Errorf("error starting MQTT Engine: %w", err) } } @@ -273,15 +283,14 @@ func (pd *PopData) ParseSourcesNG() error { return nil } -func (pd *PopData) ParseLocalFile(sourceid string, s *tapir.WBGlist, rpt chan string) error { +func (pd *PopData) ParseLocalFile(sourceid string, s *tapir.WBGlist) error { pd.Logger.Printf("ParseLocalFile: %s (%s)", sourceid, s.Type) var df dawg.Finder var err error s.Filename = viper.GetString(fmt.Sprintf("sources.%s.filename", sourceid)) if s.Filename == "" { - POPExiter("ParseLocalFile: source %s of type file has undefined filename", - sourceid) + return fmt.Errorf("source %s of type file has undefined filename", sourceid) } switch s.SrcFormat { @@ -291,10 +300,9 @@ func (pd *PopData) ParseLocalFile(sourceid string, s *tapir.WBGlist, rpt chan st _, err := tapir.ParseText(s.Filename, s.Names, true) if err != nil { if os.IsNotExist(err) { - POPExiter("ParseLocalFile: source %s (type file: %s) does not exist", - sourceid, s.Filename) + return fmt.Errorf("source %s (type file: %s) does not exist", sourceid, s.Filename) } - POPExiter("ParseLocalFile: error parsing file %s: %v", s.Filename, err) + return fmt.Errorf("error parsing file %s: %w", s.Filename, err) } case "csv": @@ -303,39 +311,36 @@ func (pd *PopData) ParseLocalFile(sourceid string, s *tapir.WBGlist, rpt chan st _, err := tapir.ParseCSV(s.Filename, s.Names, true) if err != nil { if os.IsNotExist(err) { - POPExiter("ParseLocalFile: source %s (type file: %s) does not exist", - sourceid, s.Filename) + return fmt.Errorf("source %s (type file: %s) does not exist", sourceid, s.Filename) } - POPExiter("ParseLocalFile: error parsing file %s: %v", s.Filename, err) + return fmt.Errorf("error parsing file %s: %w", s.Filename, err) } case "dawg": if s.Type != "allowlist" { - POPExiter("Error: source %s (file %s): DAWG is only defined for allowlists.", - sourceid, s.Filename) + return fmt.Errorf("source %s (file %s): DAWG is only defined for allowlists", sourceid, s.Filename) } pd.Logger.Printf("ParseLocalFile: loading DAWG: %s", s.Filename) df, err = dawg.Load(s.Filename) if err != nil { - POPExiter("Error from dawg.Load(%s): %v", s.Filename, err) + return fmt.Errorf("dawg.Load(%s): %w", s.Filename, err) } pd.Logger.Printf("ParseLocalFile: DAWG loaded") s.Format = "dawg" s.Dawgf = df default: - POPExiter("ParseLocalFile: SrcFormat \"%s\" is unknown.", s.SrcFormat) + return fmt.Errorf("SrcFormat %q is unknown", s.SrcFormat) } pd.mu.Lock() pd.Lists[s.Type][s.Name] = s pd.mu.Unlock() - rpt <- sourceid return nil } -func (pd *PopData) ParseRpzFeed(sourceid string, s *tapir.WBGlist, rpt chan string) error { +func (pd *PopData) ParseRpzFeed(sourceid string, s *tapir.WBGlist) error { // zone := viper.GetString(fmt.Sprintf("sources.%s.zone", sourceid)) // XXX: not the way to do it // if zone == "" { // return fmt.Errorf("Unable to load RPZ source %s, upstream zone not specified.", @@ -363,12 +368,14 @@ func (pd *PopData) ParseRpzFeed(sourceid string, s *tapir.WBGlist, rpt chan stri Resp: reRpt, } - <-reRpt + result := <-reRpt + if result.Error { + return fmt.Errorf("refreshing RPZ source %s: %s", sourceid, result.ErrorMsg) + } pd.mu.Lock() pd.Lists[s.Type][s.Name] = s pd.mu.Unlock() - rpt <- sourceid pd.Logger.Printf("ParseRpzFeed: parsing RPZ %s complete", s.RpzZoneName) return nil diff --git a/statusupdater.go b/statusupdater.go index da4e612..ccab39c 100644 --- a/statusupdater.go +++ b/statusupdater.go @@ -1,9 +1,10 @@ /* * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( + "context" "fmt" "log" "path/filepath" @@ -14,13 +15,19 @@ import ( "github.com/spf13/viper" ) -func (pd *PopData) StatusUpdater(conf *Config, stopch chan struct{}) { +func (pd *PopData) StatusUpdater(ctx context.Context, conf *Config) error { active := viper.GetBool("tapir.status.active") if !active { pd.Logger.Printf("*** StatusUpdater: not active, will just read status updates from channel and not publish anything") - for csu := range pd.ComponentStatusCh { - log.Printf("StatusUpdater: got status update message: %+v", csu) + for { + select { + case <-ctx.Done(): + log.Printf("StatusUpdater: stopping") + return nil + case csu := <-pd.ComponentStatusCh: + log.Printf("StatusUpdater: got status update message: %+v", csu) + } } } @@ -30,19 +37,14 @@ func (pd *PopData) StatusUpdater(conf *Config, stopch chan struct{}) { ComponentStatus: make(map[string]tapir.TapirComponentStatus), } - // me := pd.MqttEngine - // if me == nil { - // POPExiter("StatusUpdater: MQTT Engine not running") - // } - // Create a new mqtt engine just for the statusupdater. - // me, err := tapir.NewMqttEngine("statusupdater", viper.GetString("tapir.mqtt.clientid")+"statusupdates", tapir.TapirPub, pd.ComponentStatusCh, log.Default()) - // if err != nil { - // POPExiter("StatusUpdater: Error creating MQTT Engine: %v", err) - // } me := pd.MqttEngine + if me == nil { + return fmt.Errorf("MQTT Engine not running") + } ticker := time.NewTicker(60 * time.Second) + defer ticker.Stop() // var statusch = make(chan tapir.ComponentStatusUpdate, 10) // If any status updates arrive, print them out @@ -54,35 +56,35 @@ func (pd *PopData) StatusUpdater(conf *Config, stopch chan struct{}) { certCN, _, _, err := tapir.FetchTapirClientCert(log.Default(), pd.ComponentStatusCh) if err != nil { - POPExiter("StatusUpdater: Error fetching client certificate: %v", err) + return fmt.Errorf("error fetching client certificate: %w", err) } statusTopic, err := tapir.MqttTopic(certCN, "tapir.status.topic") if err != nil { - POPExiter("StatusUpdater: MQTT status topic not set") + return fmt.Errorf("MQTT status topic not set: %w", err) } keyfile := viper.GetString("tapir.status.signingkey") if keyfile == "" { - POPExiter("StatusUpdater: MQTT status signing key not set") + return fmt.Errorf("MQTT status signing key not set") } keyfile = filepath.Clean(keyfile) signkey, err := tapir.FetchMqttSigningKey(statusTopic, keyfile) if err != nil { - POPExiter("StatusUpdater: Error fetching MQTT signing key for topic %s: %v", statusTopic, err) + return fmt.Errorf("error fetching MQTT signing key for topic %s: %w", statusTopic, err) } pd.Logger.Printf("StatusUpdater: Adding pub topic '%s' to MQTT Engine", statusTopic) err = me.PubToTopic(statusTopic, signkey, "struct", true) // XXX: Brr. kludge. if err != nil { - POPExiter("Error adding topic %s to MQTT Engine: %v", statusTopic, err) + return fmt.Errorf("error adding topic %s to MQTT Engine: %w", statusTopic, err) } - pd.Logger.Printf("StatusUpdater: Topic status for MQTT engine %s: %+v", me.Creator) + pd.Logger.Printf("StatusUpdater: Topic status for MQTT engine %s", me.Creator) - _, outbox, _, err := me.StartEngine() - if err != nil { - POPExiter("StatusUpdater: Error starting MQTT Engine: %v", err) + outbox := pd.TapirMqttPubCh + if outbox == nil { + return fmt.Errorf("MQTT publish channel not available") } log.Printf("StatusUpdater: Starting") @@ -132,7 +134,7 @@ func (pd *PopData) StatusUpdater(conf *Config, stopch chan struct{}) { } s.ComponentStatus[csu.Component] = comp dirty = true - sur.Msg = fmt.Sprintf("StatusUpdater: %s report for known component: %s", csu.Status, csu.Component) + sur.Msg = fmt.Sprintf("StatusUpdater: %s report for known component: %s", tapir.StatusToString[csu.Status], csu.Component) default: log.Printf("StatusUpdater: %s report for unknown component: %s", tapir.StatusToString[csu.Status], csu.Component) sur.Error = true @@ -157,11 +159,11 @@ func (pd *PopData) StatusUpdater(conf *Config, stopch chan struct{}) { } default: - log.Printf("StatusUpdater: Unknown status: %s", csu.Status) + log.Printf("StatusUpdater: Unknown status: %v", csu.Status) } - case <-stopch: + case <-ctx.Done(): log.Printf("StatusUpdater: stopping") - return + return nil } } } diff --git a/structs.go b/structs.go index 391b1b2..5667309 100644 --- a/structs.go +++ b/structs.go @@ -1,7 +1,7 @@ /* * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "log" @@ -25,8 +25,8 @@ type PopData struct { ComponentStatusCh chan tapir.ComponentStatusUpdate Logger *log.Logger MqttLogger *log.Logger - DenylistedNames map[string]bool - DoubtlistedNames map[string]*tapir.TapirName + DenylistedNames map[string]bool + DoubtlistedNames map[string]*tapir.TapirName Policy PopPolicy Rpz RpzData RpzSources map[string]*tapir.ZoneData @@ -72,8 +72,8 @@ type RpzAxfr struct { type PopPolicy struct { Logger *log.Logger AllowlistAction tapir.Action - DenylistAction tapir.Action - Doubtlist DoubtlistPolicy + DenylistAction tapir.Action + Doubtlist DoubtlistPolicy } type DoubtlistPolicy struct { @@ -81,8 +81,8 @@ type DoubtlistPolicy struct { NumSourcesAction tapir.Action NumTapirTags int NumTapirTagsAction tapir.Action - DenyTapirTags tapir.TagMask - DenyTapirAction tapir.Action + DenyTapirTags tapir.TagMask + DenyTapirAction tapir.Action } // type WBGC map[string]*tapir.WBGlist diff --git a/xfr.go b/xfr.go index a8a541f..01ba008 100644 --- a/xfr.go +++ b/xfr.go @@ -2,7 +2,7 @@ * Copyright (c) 2024 Johan Stenstam, johan.stenstam@internetstiftelsen.se */ -package main +package pop import ( "fmt"