// Copyright © 2023 Ory Corp // SPDX-License-Identifier: Apache-2.0 package config import ( "context" "encoding/json" "fmt" "io" "os" "path/filepath" "sync" "github.com/ghodss/yaml" "github.com/ory/herodot" "github.com/ory/x/logrusx" "github.com/ory/x/stringsx" "github.com/ory/x/urlx" "github.com/ory/x/watcherx" "github.com/pelletier/go-toml" "github.com/pkg/errors" "github.com/ory/keto/internal/namespace" ) type ( Parser func([]byte, interface{}) error NamespaceFile struct { Name string Contents []byte Parser Parser namespace *namespace.Namespace // last successfully parsed namespace. May be nil if there was a parse error } NamespaceWatcher struct { sync.RWMutex namespaces map[string]*NamespaceFile logger *logrusx.Logger target string } eventHandler interface { handleRemove(*watcherx.RemoveEvent) handleChange(*watcherx.ChangeEvent) handleError(*watcherx.ErrorEvent) } ) var _ namespace.Manager = (*NamespaceWatcher)(nil) func NewNamespaceWatcher(ctx context.Context, l *logrusx.Logger, target string) (*NamespaceWatcher, error) { nw := NamespaceWatcher{ logger: l, target: target, namespaces: make(map[string]*NamespaceFile), } return &nw, watchTarget(ctx, target, &nw, l) } func (nw *NamespaceWatcher) handleRemove(e *watcherx.RemoveEvent) { nw.Lock() defer nw.Unlock() delete(nw.namespaces, e.Source()) } func (nw *NamespaceWatcher) handleChange(e *watcherx.ChangeEvent) { // the lock is acquired before parsing to ensure that the getters are // waiting for the updated values nw.Lock() defer nw.Unlock() n := nw.readNamespaceFile(e.Reader(), e.Source()) if n == nil { return } else if n.namespace == nil { // parse failed, rolling back to previous working version if existing, ok := nw.namespaces[e.Source()]; ok { existing.Contents = n.Contents } else { nw.namespaces[e.Source()] = n } } else { nw.namespaces[e.Source()] = n } } func (nw *NamespaceWatcher) handleError(e *watcherx.ErrorEvent) { nw.logger. WithError(e). Errorf("Received error while watching namespace files at target %s.", nw.target) } func watchTarget(ctx context.Context, target string, handler eventHandler, log *logrusx.Logger) error { var ( eventCh = make(watcherx.EventChannel) watcher watcherx.Watcher ) targetUrl, err := urlx.Parse(target) if err != nil { return errors.WithStack(err) } info, err := os.Stat(targetUrl.Path) //nolint:gosec // Path is from a validated configuration URL, not user input. if err != nil { return errors.WithStack(err) } if info.IsDir() { watcher, err = watcherx.WatchDirectory(ctx, urlx.GetURLFilePath(targetUrl), eventCh) } else { watcher, err = watcherx.Watch(ctx, targetUrl, eventCh) } // this handles the watcher init error if err != nil { return err } // trigger initial load done, err := watcher.DispatchNow() if err != nil { return errors.WithStack(err) } initialEventsProcessed := make(chan struct{}) go startEventHandler(ctx, eventCh, handler, done, initialEventsProcessed, log) // wait for initial load to be done <-initialEventsProcessed return nil } func startEventHandler(ctx context.Context, eventCh watcherx.EventChannel, handler eventHandler, done <-chan int, initialEventsProcessed chan<- struct{}, log *logrusx.Logger) { initialDone := false for { select { // because we use an unbuffered chan we can be sure that at least all // initial events are handled case <-done: initialDone = true close(initialEventsProcessed) // Disable this case after it fires once. watcherx closes done when // the watcher shuts down (e.g. directoryWatcher on ctx cancel); a // closed channel is always ready, so leaving done set would re-enter // this branch and close(initialEventsProcessed) a second time (panic). done = nil case <-ctx.Done(): return case e, open := <-eventCh: if !open { return } if initialDone { log. WithField("file", e.Source()). WithField("event_type", fmt.Sprintf("%T", e)). Info("A change to a namespace file was detected.") } switch e := e.(type) { case *watcherx.RemoveEvent: handler.handleRemove(e) case *watcherx.ChangeEvent: handler.handleChange(e) case *watcherx.ErrorEvent: handler.handleError(e) default: log.Warnf("Ignored unknown event %T", e) } } } } func (nw *NamespaceWatcher) readNamespaceFile(r io.Reader, source string) *NamespaceFile { parse, err := GetParser(source) if err != nil { nw.logger. WithError(err). WithField("file_name", source). Warn("could not infer format from file extension") return nil } raw, err := io.ReadAll(r) if err != nil { nw.logger. WithError(errors.WithStack(err)). WithField("file_name", source). Error("could not read namespace file") return nil } n := namespace.Namespace{} if err := parse(raw, &n); err != nil { nw.logger. WithError(errors.WithStack(err)). WithField("file_name", source). Error("could not parse namespace file") return &NamespaceFile{Name: source, Contents: raw, Parser: parse} } return &NamespaceFile{Name: source, Contents: raw, Parser: parse, namespace: &n} } func (nw *NamespaceWatcher) GetNamespaceByName(_ context.Context, name string) (*namespace.Namespace, error) { nw.RLock() defer nw.RUnlock() for _, nsf := range nw.namespaces { if nsf.namespace != nil && nsf.namespace.Name == name { return nsf.namespace, nil } } return nil, errors.WithStack(herodot.ErrNotFound().WithErrorf( "Unknown namespace with name %s", name)) } func (nw *NamespaceWatcher) GetNamespaceByConfigID(_ context.Context, id int32) (*namespace.Namespace, error) { nw.RLock() defer nw.RUnlock() for _, nspace := range nw.namespaces { //lint:ignore SA1019 backwards compatibility //nolint:staticcheck if nspace.namespace.ID == id { return nspace.namespace, nil } } return nil, errors.WithStack(herodot.ErrNotFound().WithErrorf( "Unknown namespace with ID %d", id)) } func (nw *NamespaceWatcher) Namespaces(_ context.Context) ([]*namespace.Namespace, error) { nw.RLock() defer nw.RUnlock() nspaces := make([]*namespace.Namespace, 0, len(nw.namespaces)) for _, nsf := range nw.namespaces { if nsf.namespace != nil { nspaces = append(nspaces, nsf.namespace) } } return nspaces, nil } func (nw *NamespaceWatcher) NamespaceFiles() []*NamespaceFile { nw.RLock() defer nw.RUnlock() nsfs := make([]*NamespaceFile, 0, len(nw.namespaces)) for _, nsf := range nw.namespaces { nsfs = append(nsfs, nsf) } return nsfs } func (nw *NamespaceWatcher) ShouldReload(newValue interface{}) bool { v, ok := newValue.(string) if !ok { // the manager type changed return true } // reload if target changed return v != nw.target } func GetParser(fn string) (Parser, error) { switch ext := stringsx.SwitchExact(filepath.Ext(fn)); { case ext.AddCase(".yaml"), ext.AddCase(".yml"): return func(b []byte, i interface{}) error { return yaml.Unmarshal(b, i) }, nil case ext.AddCase(".json"): return json.Unmarshal, nil case ext.AddCase(".toml"): return toml.Unmarshal, nil default: return nil, ext.ToUnknownCaseErr() } }