diff --git a/v1/config/config.go b/v1/config/config.go index 62bfc65537..1912d1f38c 100644 --- a/v1/config/config.go +++ b/v1/config/config.go @@ -9,6 +9,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "os" "path/filepath" "reflect" @@ -21,6 +22,59 @@ import ( "github.com/open-policy-agent/opa/v1/version" ) +// ServerConfig represents the different server configuration options. +type ServerConfig struct { + Metrics json.RawMessage `json:"metrics,omitempty"` + + Encoding json.RawMessage `json:"encoding,omitempty"` + Decoding json.RawMessage `json:"decoding,omitempty"` +} + +// Clone creates a deep copy of ServerConfig. +func (s *ServerConfig) Clone() *ServerConfig { + if s == nil { + return nil + } + + clone := &ServerConfig{} + + if s.Encoding != nil { + clone.Encoding = make(json.RawMessage, len(s.Encoding)) + copy(clone.Encoding, s.Encoding) + } + if s.Decoding != nil { + clone.Decoding = make(json.RawMessage, len(s.Decoding)) + copy(clone.Decoding, s.Decoding) + } + if s.Metrics != nil { + clone.Metrics = make(json.RawMessage, len(s.Metrics)) + copy(clone.Metrics, s.Metrics) + } + + return clone +} + +// StorageConfig represents Config's storage options. +type StorageConfig struct { + Disk json.RawMessage `json:"disk,omitempty"` +} + +// Clone creates a deep copy of StorageConfig. +func (s *StorageConfig) Clone() *StorageConfig { + if s == nil { + return nil + } + + clone := &StorageConfig{} + + if s.Disk != nil { + clone.Disk = make(json.RawMessage, len(s.Disk)) + copy(clone.Disk, s.Disk) + } + + return clone +} + // Config represents the configuration file that OPA can be started with. type Config struct { Services json.RawMessage `json:"services,omitempty"` @@ -38,15 +92,9 @@ type Config struct { NDBuiltinCache bool `json:"nd_builtin_cache,omitempty"` PersistenceDirectory *string `json:"persistence_directory,omitempty"` DistributedTracing json.RawMessage `json:"distributed_tracing,omitempty"` - Server *struct { - Encoding json.RawMessage `json:"encoding,omitempty"` - Decoding json.RawMessage `json:"decoding,omitempty"` - Metrics json.RawMessage `json:"metrics,omitempty"` - } `json:"server,omitempty"` - Storage *struct { - Disk json.RawMessage `json:"disk,omitempty"` - } `json:"storage,omitempty"` - Extra map[string]json.RawMessage `json:"-"` + Server *ServerConfig `json:"server,omitempty"` + Storage *StorageConfig `json:"storage,omitempty"` + Extra map[string]json.RawMessage `json:"-"` } // ParseConfig returns a valid Config object with defaults injected. The id @@ -122,38 +170,6 @@ func (c Config) NDBuiltinCacheEnabled() bool { return c.NDBuiltinCache } -func (c *Config) validateAndInjectDefaults(id string) error { - - if c.DefaultDecision == nil { - s := defaultDecisionPath - c.DefaultDecision = &s - } - - _, err := ref.ParseDataPath(*c.DefaultDecision) - if err != nil { - return err - } - - if c.DefaultAuthorizationDecision == nil { - s := defaultAuthorizationDecisionPath - c.DefaultAuthorizationDecision = &s - } - - _, err = ref.ParseDataPath(*c.DefaultAuthorizationDecision) - if err != nil { - return err - } - - if c.Labels == nil { - c.Labels = map[string]string{} - } - - c.Labels["id"] = id - c.Labels["version"] = version.Version - - return nil -} - // GetPersistenceDirectory returns the configured persistence directory, or $PWD/.opa if none is configured func (c Config) GetPersistenceDirectory() (string, error) { if c.PersistenceDirectory == nil { @@ -197,6 +213,123 @@ func (c *Config) ActiveConfig() (any, error) { return result, nil } +// Clone creates a deep copy of the Config struct +func (c *Config) Clone() *Config { + if c == nil { + return nil + } + + clone := &Config{ + NDBuiltinCache: c.NDBuiltinCache, + Server: c.Server.Clone(), + Storage: c.Storage.Clone(), + Labels: maps.Clone(c.Labels), + } + + if c.Services != nil { + clone.Services = make(json.RawMessage, len(c.Services)) + copy(clone.Services, c.Services) + } + if c.Discovery != nil { + clone.Discovery = make(json.RawMessage, len(c.Discovery)) + copy(clone.Discovery, c.Discovery) + } + if c.Bundle != nil { + clone.Bundle = make(json.RawMessage, len(c.Bundle)) + copy(clone.Bundle, c.Bundle) + } + if c.Bundles != nil { + clone.Bundles = make(json.RawMessage, len(c.Bundles)) + copy(clone.Bundles, c.Bundles) + } + if c.DecisionLogs != nil { + clone.DecisionLogs = make(json.RawMessage, len(c.DecisionLogs)) + copy(clone.DecisionLogs, c.DecisionLogs) + } + if c.Status != nil { + clone.Status = make(json.RawMessage, len(c.Status)) + copy(clone.Status, c.Status) + } + if c.Keys != nil { + clone.Keys = make(json.RawMessage, len(c.Keys)) + copy(clone.Keys, c.Keys) + } + if c.Caching != nil { + clone.Caching = make(json.RawMessage, len(c.Caching)) + copy(clone.Caching, c.Caching) + } + if c.DistributedTracing != nil { + clone.DistributedTracing = make(json.RawMessage, len(c.DistributedTracing)) + copy(clone.DistributedTracing, c.DistributedTracing) + } + + if c.DefaultDecision != nil { + s := *c.DefaultDecision + clone.DefaultDecision = &s + } + if c.DefaultAuthorizationDecision != nil { + s := *c.DefaultAuthorizationDecision + clone.DefaultAuthorizationDecision = &s + } + if c.PersistenceDirectory != nil { + s := *c.PersistenceDirectory + clone.PersistenceDirectory = &s + } + + if c.Plugins != nil { + clone.Plugins = make(map[string]json.RawMessage, len(c.Plugins)) + for k, v := range c.Plugins { + if v != nil { + clone.Plugins[k] = make(json.RawMessage, len(v)) + copy(clone.Plugins[k], v) + } + } + } + + if c.Extra != nil { + clone.Extra = make(map[string]json.RawMessage, len(c.Extra)) + for k, v := range c.Extra { + if v != nil { + clone.Extra[k] = make(json.RawMessage, len(v)) + copy(clone.Extra[k], v) + } + } + } + + return clone +} + +func (c *Config) validateAndInjectDefaults(id string) error { + if c.DefaultDecision == nil { + s := defaultDecisionPath + c.DefaultDecision = &s + } + + _, err := ref.ParseDataPath(*c.DefaultDecision) + if err != nil { + return err + } + + if c.DefaultAuthorizationDecision == nil { + s := defaultAuthorizationDecisionPath + c.DefaultAuthorizationDecision = &s + } + + _, err = ref.ParseDataPath(*c.DefaultAuthorizationDecision) + if err != nil { + return err + } + + if c.Labels == nil { + c.Labels = map[string]string{} + } + + c.Labels["id"] = id + c.Labels["version"] = version.Version + + return nil +} + func removeServiceCredentials(x any) error { switch x := x.(type) { case nil: diff --git a/v1/config/config_test.go b/v1/config/config_test.go index eb25bd8bd4..dc69d9fce7 100644 --- a/v1/config/config_test.go +++ b/v1/config/config_test.go @@ -14,7 +14,9 @@ import ( "slices" "testing" + "github.com/google/go-cmp/cmp" "github.com/open-policy-agent/opa/v1/util" + "github.com/open-policy-agent/opa/v1/util/test" "github.com/open-policy-agent/opa/v1/version" ) @@ -175,7 +177,6 @@ func TestPersistDirectory(t *testing.T) { } func TestActiveConfig(t *testing.T) { - common := `"labels": { "region": "west" }, @@ -371,7 +372,6 @@ func TestActiveConfig(t *testing.T) { for name, tc := range tests { t.Run(name, func(t *testing.T) { - conf, err := ParseConfig(tc.raw, "foo") if err != nil { t.Fatal(err) @@ -403,7 +403,6 @@ func TestActiveConfig(t *testing.T) { } }) } - } func TestExtraConfigFieldsRoundtrip(t *testing.T) { @@ -438,3 +437,41 @@ bar: t.Fatalf("want %v got %v", expected, actual) } } + +func TestConfigClone(t *testing.T) { + // test nil config + var nilConfig *Config + cloned := nilConfig.Clone() + if cloned != nil { + t.Fatal("expected nil clone for nil config") + } + + // test empty config + emptyConfig := &Config{} + clonedEmpty := emptyConfig.Clone() + if clonedEmpty == nil { + t.Fatal("clone returned nil for empty config") + } + if clonedEmpty == emptyConfig { + t.Fatal("clone should be a different instance") + } + if diff := cmp.Diff(emptyConfig, clonedEmpty); diff != "" { + t.Errorf("empty clone differs:\n%s", diff) + } + + // test config with all fields populated using reflection + original := test.PopulateAllFields[Config](t) + + cloned = original.Clone() + if cloned == nil { + t.Fatal("clone returned nil") + } + + if cloned == original { + t.Fatal("clone should be different instance") + } + + if diff := cmp.Diff(original, cloned); diff != "" { + t.Errorf("clone differs:\n%s", diff) + } +} diff --git a/v1/plugins/bundle/plugin.go b/v1/plugins/bundle/plugin.go index 63d2c27694..60a9515904 100644 --- a/v1/plugins/bundle/plugin.go +++ b/v1/plugins/bundle/plugin.go @@ -247,7 +247,6 @@ func (p *Plugin) Reconfigure(ctx context.Context, config any) { p.ready = false p.manager.UpdatePluginStatus(Name, &plugins.Status{State: plugins.StateNotReady}) } - } // Loaders returns the map of bundle loaders configured on this plugin. @@ -267,7 +266,6 @@ func (p *Plugin) Trigger(ctx context.Context) error { for name, d := range downloaders { // plugin callback will also log the trigger error and include it in the bundle status err := d.Trigger(ctx) - // only return errors for TriggerMode manual as periodic bundles will be retried if err != nil { trigger := p.Config().Bundles[name].Trigger @@ -370,7 +368,6 @@ func (p *Plugin) readBundleEtagFromStore(ctx context.Context, name string) strin } func (p *Plugin) loadAndActivateBundlesFromDisk(ctx context.Context) { - persistedBundles := map[string]*bundle.Bundle{} bundles := p.getBundlesCpy() @@ -430,7 +427,6 @@ func (p *Plugin) loadAndActivateBundlesFromDisk(ctx context.Context) { } func (p *Plugin) newDownloader(name string, source *Source, bundles map[string]*Source) Loader { - if u, err := url.Parse(source.Resource); err == nil && u.Scheme == "file" { return &fileLoader{ name: name, @@ -451,8 +447,8 @@ func (p *Plugin) newDownloader(name string, source *Source, bundles map[string]* } if strings.ToLower(client.Config().Type) == "oci" { ociStorePath := filepath.Join(os.TempDir(), "opa", "oci") // use temporary folder /tmp/opa/oci - if p.manager.Config.PersistenceDirectory != nil { - ociStorePath = filepath.Join(*p.manager.Config.PersistenceDirectory, "oci") + if cfg := p.manager.GetConfig(); cfg.PersistenceDirectory != nil { + ociStorePath = filepath.Join(*cfg.PersistenceDirectory, "oci") } return download.NewOCI(conf, client, path, ociStorePath). WithCallback(callback). @@ -496,7 +492,6 @@ func (p *Plugin) oneShot(ctx context.Context, name string, u download.Update) { } func (p *Plugin) process(ctx context.Context, name string, u download.Update) { - if u.Metrics != nil { p.status[name].Metrics = u.Metrics } else { @@ -646,7 +641,7 @@ func (p *Plugin) activate(ctx context.Context, name string, b *bundle.Bundle, is isAuthzEnabled := p.manager.Info.Get(ast.StringTerm("authorization_enabled")) if ast.BooleanTerm(true).Equal(isAuthzEnabled) && ast.BooleanTerm(false).Equal(skipKnownSchemaCheck) { - authorizationDecisionRef, err := ref.ParseDataPath(*p.manager.Config.DefaultAuthorizationDecision) + authorizationDecisionRef, err := ref.ParseDataPath(*p.manager.GetConfig().DefaultAuthorizationDecision) if err != nil { return err } @@ -711,7 +706,6 @@ func (p *Plugin) configDelta(newConfig *Config) (map[string]*Source, map[string] } func (p *Plugin) saveBundleToDisk(name string, raw io.Reader) error { - bundleName := getNormalizedBundleName(name) bundleDir := filepath.Join(p.bundlePersistPath, bundleName) @@ -756,7 +750,7 @@ func (p *Plugin) log(name string) logging.Logger { } func (p *Plugin) getBundlePersistPath() (string, error) { - persistDir, err := p.manager.Config.GetPersistenceDirectory() + persistDir, err := p.manager.GetConfig().GetPersistenceDirectory() if err != nil { return "", err } @@ -818,15 +812,12 @@ func (fl *fileLoader) Start(ctx context.Context) { } func (*fileLoader) Stop(context.Context) { - } func (*fileLoader) ClearCache() { - } func (*fileLoader) SetCache(string) { - } func (fl *fileLoader) Trigger(ctx context.Context) error { diff --git a/v1/plugins/bundle/plugin_test.go b/v1/plugins/bundle/plugin_test.go index e333a94cd0..d1efadd364 100644 --- a/v1/plugins/bundle/plugin_test.go +++ b/v1/plugins/bundle/plugin_test.go @@ -711,7 +711,12 @@ func TestPluginOneShotWithAuthzSchemaVerificationNonDefaultAuthzPath(t *testing. defer manager.Stop(ctx) s := "/foo/authz/allow" - manager.Config.DefaultAuthorizationDecision = &s + cfg := manager.GetConfig() + cfg.DefaultAuthorizationDecision = &s + err := manager.Reconfigure(cfg) + if err != nil { + t.Fatal(err) + } info, err := runtime.Term(runtime.Params{Config: nil, IsAuthorizationEnabled: true}) if err != nil { @@ -3785,7 +3790,6 @@ func TestPluginActivateScopedBundle(t *testing.T) { // The test data claims a/{a1-6} where even paths are policy and // odd paths are raw JSON. if err := storage.Txn(ctx, manager.Store, storage.WriteParams, func(txn storage.Transaction) error { - externalData := map[string]any{"a": map[string]any{"a1": "x1", "a3": "x2", "a5": "x3"}} if err := manager.Store.Write(ctx, txn, storage.AddOp, storage.Path{}, externalData); err != nil { @@ -3845,7 +3849,8 @@ func TestPluginActivateScopedBundle(t *testing.T) { module = "package a.a4\n\nbar=1\n\nfunc(x) = x" b = bundle.Bundle{ - Manifest: bundle.Manifest{Revision: "quickbrownfaux-2", Roots: &[]string{"a/a3", "a/a4"}, + Manifest: bundle.Manifest{ + Revision: "quickbrownfaux-2", Roots: &[]string{"a/a3", "a/a4"}, Metadata: map[string]any{ "a": map[string]any{ "a1": "deadbeef", @@ -4156,7 +4161,6 @@ func TestPluginReconfigure(t *testing.T) { for _, stage := range stages { t.Run(stage.name, func(t *testing.T) { - plugin.Reconfigure(ctx, stage.cfg) var expectedNumBundles int @@ -4936,11 +4940,19 @@ func TestSaveBundleToDiskNewConfiguredPersistDir(t *testing.T) { manager := getTestManager() defer manager.Stop(context.Background()) - manager.Config.PersistenceDirectory = &dir + + cfg := manager.GetConfig() + cfg.PersistenceDirectory = &dir + + err := manager.Reconfigure(cfg) + if err != nil { + t.Fatal(err) + } + bundles := map[string]*Source{} plugin := New(&Config{Bundles: bundles}, manager) - err := plugin.Start(context.Background()) + err = plugin.Start(context.Background()) if err != nil { t.Fatalf("unexpected error %v", err) } @@ -5129,7 +5141,7 @@ p contains 1 if { t.Fatalf("unexpected error %v", err) } - if err := os.WriteFile(filepath.Join(bundleDir, "bundle.tar.gz"), buf.Bytes(), 0644); err != nil { + if err := os.WriteFile(filepath.Join(bundleDir, "bundle.tar.gz"), buf.Bytes(), 0o644); err != nil { t.Fatalf("unexpected error %v", err) } @@ -5209,7 +5221,15 @@ func TestConfiguredBundlePersistPath(t *testing.T) { persistPath := "/var/opa" manager := getTestManager() defer manager.Stop(context.Background()) - manager.Config.PersistenceDirectory = &persistPath + + cfg := manager.GetConfig() + cfg.PersistenceDirectory = &persistPath + + err := manager.Reconfigure(cfg) + if err != nil { + t.Fatal(err) + } + plugin := New(&Config{}, manager) path, err := plugin.getBundlePersistPath() @@ -5226,7 +5246,6 @@ func TestPluginUsingFileLoader(t *testing.T) { t.Parallel() test.WithTempFS(map[string]string{}, func(dir string) { - b := bundle.Bundle{ Data: map[string]any{}, Modules: []bundle.ModuleFile{ @@ -5397,7 +5416,6 @@ p contains 7 if { popts := ast.ParserOptions{RegoVersion: regoVersion} test.WithTempFS(map[string]string{}, func(dir string) { - b := bundle.Bundle{ Data: map[string]any{}, Modules: []bundle.ModuleFile{ @@ -5707,7 +5725,6 @@ p contains 7 if { for _, tc := range tests { t.Run(tc.note, func(t *testing.T) { test.WithTempFS(map[string]string{}, func(dir string) { - manifest := bundle.Manifest{} manifest.SetRegoVersion(tc.bundleRegoVersion) b := bundle.Bundle{ @@ -5790,7 +5807,6 @@ func TestPluginUsingDirectoryLoader(t *testing.T) { p := 7`, }, func(dir string) { - mgr := getTestManager() url := "file://" + dir @@ -5938,7 +5954,6 @@ p contains 7 if { test.WithTempFS(map[string]string{ "test.rego": tc.module, }, func(dir string) { - manager, err := plugins.New(nil, "test-instance-id", inmemtst.New(), plugins.WithParserOptions(popts)) if err != nil { t.Fatal("unexpected error:", err) @@ -6228,7 +6243,6 @@ p contains 7 if { "test.rego": tc.module, ".manifest": fmt.Sprintf(`{"rego_version": %d}`, bundleRegoVersion(tc.bundleRegoVersion)), }, func(dir string) { - managerPopts := ast.ParserOptions{RegoVersion: tc.managerRegoVersion} manager, err := plugins.New(nil, "test-instance-id", inmemtst.New(), plugins.WithParserOptions(managerPopts)) @@ -6288,7 +6302,6 @@ func TestPluginReadBundleEtagFromDiskStore(t *testing.T) { notModifiedCount := 0 s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - etag := r.Header.Get("If-None-Match") if etag == "foo" { notModifiedCount++ @@ -7395,7 +7408,7 @@ func writeTestBundleToDisk(t *testing.T, srcDir string, signed bool) bundle.Bund t.Fatalf("unexpected error %v", err) } - if err := os.WriteFile(filepath.Join(srcDir, "bundle.tar.gz"), buf.Bytes(), 0644); err != nil { + if err := os.WriteFile(filepath.Join(srcDir, "bundle.tar.gz"), buf.Bytes(), 0o644); err != nil { t.Fatalf("unexpected error %v", err) } @@ -7502,7 +7515,6 @@ func validateStoreState(ctx context.Context, t *testing.T, store storage.Store, } return nil - }); err != nil { t.Fatal(err) } diff --git a/v1/plugins/discovery/discovery.go b/v1/plugins/discovery/discovery.go index 15ea34d428..36b3c06a5b 100644 --- a/v1/plugins/discovery/discovery.go +++ b/v1/plugins/discovery/discovery.go @@ -113,13 +113,14 @@ func New(manager *plugins.Manager, opts ...func(*Discovery)) (*Discovery, error) result.logger = manager.Logger().WithFields(map[string]any{"plugin": Name}) - config, err := NewConfigBuilder().WithBytes(manager.Config.Discovery).WithServices(manager.Services()). + managerConfig := manager.GetConfig() + config, err := NewConfigBuilder().WithBytes([]byte(managerConfig.Discovery)).WithServices(manager.Services()). WithKeyConfigs(manager.PublicKeys()).Parse() if err != nil { return nil, err } else if config == nil { - if _, err := getPluginSet(result.factories, manager, manager.Config, result.metrics, result.logger, nil); err != nil { + if _, err := getPluginSet(result.factories, manager, managerConfig, result.metrics, result.logger, nil); err != nil { return nil, err } return result, nil @@ -129,8 +130,8 @@ func New(manager *plugins.Manager, opts ...func(*Discovery)) (*Discovery, error) restClient := manager.Client(config.service) if strings.ToLower(restClient.Config().Type) == "oci" { ociStorePath := filepath.Join(os.TempDir(), "opa", "oci") // use temporary folder /tmp/opa/oci - if manager.Config.PersistenceDirectory != nil { - ociStorePath = filepath.Join(*manager.Config.PersistenceDirectory, "oci") + if managerConfig.PersistenceDirectory != nil { + ociStorePath = filepath.Join(*managerConfig.PersistenceDirectory, "oci") } result.downloader = download.NewOCI(config.Config, restClient, config.path, ociStorePath). WithCallback(result.oneShot). @@ -157,7 +158,6 @@ func New(manager *plugins.Manager, opts ...func(*Discovery)) (*Discovery, error) // Start starts the dynamic discovery process if configured. func (c *Discovery) Start(ctx context.Context) error { - bundlePersistPath, err := c.getBundlePersistPath() if err != nil { return err @@ -230,7 +230,7 @@ func (c *Discovery) Unregister(name any) { } func (c *Discovery) getBundlePersistPath() (string, error) { - persistDir, err := c.manager.Config.GetPersistenceDirectory() + persistDir, err := c.manager.GetConfig().GetPersistenceDirectory() if err != nil { return "", err } @@ -239,7 +239,6 @@ func (c *Discovery) getBundlePersistPath() (string, error) { } func (c *Discovery) loadAndActivateBundleFromDisk(ctx context.Context) { - if c.config != nil && c.config.Persist { b, err := c.loadBundleFromDisk() if err != nil { @@ -293,7 +292,6 @@ func (c *Discovery) loadBundleFromDisk() (*bundleApi.Bundle, error) { } func (c *Discovery) saveBundleToDisk(raw io.Reader) error { - bundleDir := filepath.Join(c.bundlePersistPath, c.discoveryBundleDirName()) bundleFile := filepath.Join(bundleDir, "bundle.tar.gz") @@ -320,7 +318,6 @@ func saveCurrentBundleToDisk(path string, raw io.Reader) (string, error) { } func (c *Discovery) oneShot(ctx context.Context, u download.Update) { - c.processUpdate(ctx, u) if p := status.Lookup(c.manager); p != nil { @@ -405,7 +402,6 @@ func (c *Discovery) processUpdate(ctx context.Context, u download.Update) { } func (c *Discovery) reconfigure(ctx context.Context, u download.Update) error { - ps, err := c.processBundle(ctx, u.Bundle) if err != nil { return err @@ -452,7 +448,6 @@ func (c *Discovery) applyLocalPluginConfigOverride(conf *config.Config) (*config } func (c *Discovery) processBundle(ctx context.Context, b *bundleApi.Bundle) (*pluginSet, error) { - config, err := evaluateBundle(ctx, c.manager.ID, c.manager.Info, b, c.config.query) if err != nil { return nil, err @@ -474,7 +469,7 @@ func (c *Discovery) processBundle(ctx context.Context, b *bundleApi.Bundle) (*pl // Note: We don't currently support changes to the discovery // configuration. These changes are risky because errors would be // unrecoverable (without keeping track of changes and rolling back...) - config.Discovery = c.manager.Config.Discovery + config.Discovery = c.manager.GetConfig().Discovery // check for updates to the discovery service opts := c.manager.DefaultServiceOpts(config) @@ -537,7 +532,6 @@ func (c *Discovery) discoveryBundleDirName() string { } func evaluateBundle(ctx context.Context, id string, info *ast.Term, b *bundleApi.Bundle, query string) (*config.Config, error) { - modules := b.ParsedModules("discovery") compiler := ast.NewCompiler() @@ -593,11 +587,18 @@ type pluginfactory struct { config any } -func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager, config *config.Config, m metrics.Metrics, l logging.Logger, trigger *plugins.TriggerMode) (*pluginSet, error) { - +func getPluginSet( + factories map[string]plugins.Factory, + manager *plugins.Manager, + config *config.Config, + m metrics.Metrics, + l logging.Logger, + trigger *plugins.TriggerMode, +) (*pluginSet, error) { // Parse and validate plugin configurations. pluginNames := []string{} pluginFactories := []pluginfactory{} + serviceNames := manager.Services() for k := range config.Plugins { f, ok := factories[k] @@ -622,12 +623,12 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager // Parse and validate bundle/logs/status configurations. // If `bundle` was configured use that, otherwise try the new `bundles` option - bundleConfig, err := bundle.ParseConfig(config.Bundle, manager.Services()) + bundleConfig, err := bundle.ParseConfig(config.Bundle, serviceNames) if err != nil { return nil, err } if bundleConfig == nil { - bundleConfig, err = bundle.NewConfigBuilder().WithBytes(config.Bundles).WithServices(manager.Services()). + bundleConfig, err = bundle.NewConfigBuilder().WithBytes(config.Bundles).WithServices(serviceNames). WithKeyConfigs(manager.PublicKeys()).WithTriggerMode(trigger).Parse() if err != nil { return nil, err @@ -636,13 +637,13 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager manager.Logger().Warn("Deprecated 'bundle' configuration specified. Use 'bundles' instead. See https://www.openpolicyagent.org/docs/latest/configuration/#bundles") } - decisionLogsConfig, err := logs.NewConfigBuilder().WithBytes(config.DecisionLogs).WithServices(manager.Services()). + decisionLogsConfig, err := logs.NewConfigBuilder().WithBytes(config.DecisionLogs).WithServices(serviceNames). WithPlugins(pluginNames).WithTriggerMode(trigger).WithLogger(l).Parse() if err != nil { return nil, err } - statusConfig, err := status.NewConfigBuilder().WithBytes(config.Status).WithServices(manager.Services()). + statusConfig, err := status.NewConfigBuilder().WithBytes(config.Status).WithServices(serviceNames). WithPlugins(pluginNames).WithTriggerMode(trigger).Parse() if err != nil { return nil, err @@ -657,7 +658,7 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager if created { starts = append(starts, p) } else if p != nil { - reconfigs = append(reconfigs, pluginreconfig{bundleConfig, p}) + reconfigs = append(reconfigs, pluginreconfig{Config: bundleConfig, Plugin: p}) } } @@ -666,7 +667,7 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager if created { starts = append(starts, p) } else if p != nil { - reconfigs = append(reconfigs, pluginreconfig{decisionLogsConfig, p}) + reconfigs = append(reconfigs, pluginreconfig{Config: decisionLogsConfig, Plugin: p}) } } @@ -675,11 +676,11 @@ func getPluginSet(factories map[string]plugins.Factory, manager *plugins.Manager if created { starts = append(starts, p) } else if p != nil { - reconfigs = append(reconfigs, pluginreconfig{statusConfig, p}) + reconfigs = append(reconfigs, pluginreconfig{Config: statusConfig, Plugin: p}) } } - result := &pluginSet{starts, reconfigs} + result := &pluginSet{Start: starts, Reconfig: reconfigs} getCustomPlugins(manager, pluginFactories, result) @@ -708,7 +709,6 @@ func getDecisionLogsPlugin(m *plugins.Manager, config *logs.Config, metrics metr } func getStatusPlugin(m *plugins.Manager, config *status.Config, metrics metrics.Metrics) (plugin *status.Plugin, created bool) { - plugin = status.Lookup(m) if plugin == nil { @@ -724,7 +724,7 @@ func getStatusPlugin(m *plugins.Manager, config *status.Config, metrics metrics. func getCustomPlugins(manager *plugins.Manager, factories []pluginfactory, result *pluginSet) { for _, pf := range factories { if plugin := manager.Plugin(pf.name); plugin != nil { - result.Reconfig = append(result.Reconfig, pluginreconfig{pf.config, plugin}) + result.Reconfig = append(result.Reconfig, pluginreconfig{Config: pf.config, Plugin: plugin}) } else { plugin := pf.factory.New(manager, pf.config) manager.Register(pf.name, plugin) diff --git a/v1/plugins/discovery/discovery_test.go b/v1/plugins/discovery/discovery_test.go index d150b5216d..35641f0f50 100644 --- a/v1/plugins/discovery/discovery_test.go +++ b/v1/plugins/discovery/discovery_test.go @@ -54,7 +54,6 @@ func TestMain(m *testing.M) { } func TestEvaluateBundle(t *testing.T) { - sampleModule := ` package foo.bar import rego.v1 @@ -112,11 +111,9 @@ func TestEvaluateBundle(t *testing.T) { if !reflect.DeepEqual(expectedBundleConfig, parsedConfig) { t.Fatalf("Expected bundle config %v, but got %v", expectedBundleConfig, parsedConfig) } - } func TestProcessBundle(t *testing.T) { - ctx := context.Background() manager, err := plugins.New([]byte(`{ @@ -186,11 +183,9 @@ func TestProcessBundle(t *testing.T) { if err == nil { t.Fatal("Expected error but got success") } - } func TestEnvVarSubstitution(t *testing.T) { - ctx := context.Background() manager, err := plugins.New([]byte(`{ @@ -230,10 +225,11 @@ func TestEnvVarSubstitution(t *testing.T) { t.Fatalf("Expected exactly three start events but got %v", ps) } - actualConfig, err := manager.Config.ActiveConfig() + actualConfig, err := manager.GetConfig().ActiveConfig() if err != nil { t.Fatal(err) } + assertConfig(t, actualConfig, fmt.Sprintf(`{ "bundle": { "name": "test1" @@ -290,7 +286,7 @@ decision_logs := {} if { 3 == 3 } t.Fatalf("Expected exactly three start events but got %v", ps) } - actualConfig, err := manager.Config.ActiveConfig() + actualConfig, err := manager.GetConfig().ActiveConfig() if err != nil { t.Fatal(err) } @@ -329,7 +325,7 @@ decision_logs.partition_name := "bar" if { 3 == 3 } t.Fatalf("Expected exactly three start events but got %v", ps) } - actualConfig, err = manager.Config.ActiveConfig() + actualConfig, err = manager.GetConfig().ActiveConfig() if err != nil { t.Fatal(err) } @@ -375,7 +371,6 @@ decision_logs.partition_name := "bar" if { 3 == 3 } } func TestProcessBundleWithActiveConfig(t *testing.T) { - ctx := context.Background() manager, err := plugins.New([]byte(`{ @@ -431,7 +426,7 @@ func TestProcessBundleWithActiveConfig(t *testing.T) { t.Fatal(err) } - actual, err := manager.Config.ActiveConfig() + actual, err := manager.GetConfig().ActiveConfig() if err != nil { t.Fatal(err) } @@ -496,7 +491,7 @@ func TestProcessBundleWithActiveConfig(t *testing.T) { t.Fatal(err) } - actual, err = manager.Config.ActiveConfig() + actual, err = manager.GetConfig().ActiveConfig() if err != nil { t.Fatal(err) } @@ -602,7 +597,7 @@ func TestStartWithBundlePersistence(t *testing.T) { t.Fatalf("unexpected error %v", err) } - if err := os.WriteFile(filepath.Join(bundleDir, "bundle.tar.gz"), buf.Bytes(), 0644); err != nil { + if err := os.WriteFile(filepath.Join(bundleDir, "bundle.tar.gz"), buf.Bytes(), 0o644); err != nil { t.Fatalf("unexpected error %v", err) } @@ -619,7 +614,12 @@ func TestStartWithBundlePersistence(t *testing.T) { t.Fatal(err) } - manager.Config.PersistenceDirectory = &dir + cfg := manager.GetConfig() + cfg.PersistenceDirectory = &dir + err = manager.Reconfigure(cfg) + if err != nil { + t.Fatal(err) + } testPlugin := &reconfigureTestPlugin{counts: map[string]int{}} testFactory := testFactory{p: testPlugin} @@ -1370,7 +1370,12 @@ func TestSaveBundleToDiskNewConfiguredPersistDir(t *testing.T) { } // configure persistence dir instead of using the default. Discover plugin should pick this up - manager.Config.PersistenceDirectory = &dir + cfg := manager.GetConfig() + cfg.PersistenceDirectory = &dir + err = manager.Reconfigure(cfg) + if err != nil { + t.Fatal(err) + } testPlugin := &reconfigureTestPlugin{counts: map[string]int{}} testFactory := testFactory{p: testPlugin} @@ -1418,7 +1423,6 @@ func TestSaveBundleToDiskNewConfiguredPersistDir(t *testing.T) { } func TestReconfigure(t *testing.T) { - manager, err := plugins.New([]byte(`{ "labels": {"x": "y"}, "services": { @@ -1474,11 +1478,12 @@ func TestReconfigure(t *testing.T) { // Verify decision ids set expDecision := ast.MustParseTerm("data.bar.baz") expAuthzDecision := ast.MustParseTerm("data.baz.qux") - if !manager.Config.DefaultDecisionRef().Equal(expDecision.Value) { - t.Errorf("Expected default decision to be %v but got %v", expDecision, manager.Config.DefaultDecisionRef()) + cfg := manager.GetConfig() + if !cfg.DefaultDecisionRef().Equal(expDecision.Value) { + t.Errorf("Expected default decision to be %v but got %v", expDecision, cfg.DefaultDecisionRef()) } - if !manager.Config.DefaultAuthorizationDecisionRef().Equal(expAuthzDecision.Value) { - t.Errorf("Expected default authz decision to be %v but got %v", expAuthzDecision, manager.Config.DefaultAuthorizationDecisionRef()) + if !cfg.DefaultAuthorizationDecisionRef().Equal(expAuthzDecision.Value) { + t.Errorf("Expected default authz decision to be %v but got %v", expAuthzDecision, cfg.DefaultAuthorizationDecisionRef()) } // Verify plugins started @@ -1576,11 +1581,12 @@ plugins.test_plugin := v if { // Verify decision ids set expDecision := ast.MustParseTerm("data.bar.baz") expAuthzDecision := ast.MustParseTerm("data.baz.qux") - if !manager.Config.DefaultDecisionRef().Equal(expDecision.Value) { - t.Errorf("Expected default decision to be %v but got %v", expDecision, manager.Config.DefaultDecisionRef()) + cfg := manager.GetConfig() + if !cfg.DefaultDecisionRef().Equal(expDecision.Value) { + t.Errorf("Expected default decision to be %v but got %v", expDecision, cfg.DefaultDecisionRef()) } - if !manager.Config.DefaultAuthorizationDecisionRef().Equal(expAuthzDecision.Value) { - t.Errorf("Expected default authz decision to be %v but got %v", expAuthzDecision, manager.Config.DefaultAuthorizationDecisionRef()) + if !cfg.DefaultAuthorizationDecisionRef().Equal(expAuthzDecision.Value) { + t.Errorf("Expected default authz decision to be %v but got %v", expAuthzDecision, cfg.DefaultAuthorizationDecisionRef()) } // Verify plugins started @@ -1717,11 +1723,12 @@ plugins.test_plugin := v if { // Verify decision ids set expDecision := ast.MustParseTerm("data.bar.baz") expAuthzDecision := ast.MustParseTerm("data.baz.qux") - if !manager.Config.DefaultDecisionRef().Equal(expDecision.Value) { - t.Errorf("Expected default decision to be %v but got %v", expDecision, manager.Config.DefaultDecisionRef()) + cfg := manager.GetConfig() + if !cfg.DefaultDecisionRef().Equal(expDecision.Value) { + t.Errorf("Expected default decision to be %v but got %v", expDecision, cfg.DefaultDecisionRef()) } - if !manager.Config.DefaultAuthorizationDecisionRef().Equal(expAuthzDecision.Value) { - t.Errorf("Expected default authz decision to be %v but got %v", expAuthzDecision, manager.Config.DefaultAuthorizationDecisionRef()) + if !cfg.DefaultAuthorizationDecisionRef().Equal(expAuthzDecision.Value) { + t.Errorf("Expected default authz decision to be %v but got %v", expAuthzDecision, cfg.DefaultAuthorizationDecisionRef()) } // Verify plugins started @@ -1842,8 +1849,12 @@ func TestReconfigureWithLocalOverride(t *testing.T) { disco.oneShot(ctx, download.Update{Bundle: serviceBundle}) expAuthzRule := "/http/example/system/allow" - if *manager.Config.DefaultAuthorizationDecision != expAuthzRule { - t.Errorf("Expected default authorization decision %v but got %v", expAuthzRule, *manager.Config.DefaultAuthorizationDecision) + var defaultAuthzDecision string + if cfg := manager.GetConfig(); cfg.DefaultAuthorizationDecision != nil { + defaultAuthzDecision = *cfg.DefaultAuthorizationDecision + } + if defaultAuthzDecision != expAuthzRule { + t.Errorf("Expected default authorization decision %v but got %v", expAuthzRule, defaultAuthzDecision) } // `default_decision` is specified in both boot and service config. The former should take precedence. @@ -1864,8 +1875,12 @@ func TestReconfigureWithLocalOverride(t *testing.T) { } expAuthzRule = "/http/example/authz/allow" - if *manager.Config.DefaultDecision != expAuthzRule { - t.Fatalf("Expected default decision %v but got %v", expAuthzRule, *manager.Config.DefaultDecision) + var defaultDecision string + if cfg := manager.GetConfig(); cfg.DefaultDecision != nil { + defaultDecision = *cfg.DefaultDecision + } + if defaultDecision != expAuthzRule { + t.Fatalf("Expected default decision %v but got %v", expAuthzRule, defaultDecision) } // `nd_builtin_cache` is specified in both boot and service config. The former should take precedence. @@ -1885,7 +1900,7 @@ func TestReconfigureWithLocalOverride(t *testing.T) { t.Fatal("expected key \"nd_builtin_cache\" to be overridden") } - if manager.Config.NDBuiltinCache { + if manager.GetConfig().NDBuiltinCache { t.Fatal("Expected nd_builtin_cache value to be false") } @@ -1900,7 +1915,11 @@ func TestReconfigureWithLocalOverride(t *testing.T) { disco.oneShot(ctx, download.Update{Bundle: serviceBundle}) - if manager.Config.PersistenceDirectory == nil || *manager.Config.PersistenceDirectory != "test" { + var persistDir string + if cfg := manager.GetConfig(); cfg.PersistenceDirectory != nil { + persistDir = *cfg.PersistenceDirectory + } + if persistDir != "test" { t.Fatal("Unexpected update to persistence directory") } @@ -1979,7 +1998,7 @@ func TestReconfigureWithLocalOverride(t *testing.T) { } } - cacheConf, err := cache.ParseCachingConfig(manager.Config.Caching) + cacheConf, err := cache.ParseCachingConfig([]byte(manager.GetConfig().Caching)) if err != nil { t.Fatal(err) } @@ -2029,19 +2048,18 @@ func TestReconfigureWithLocalOverride(t *testing.T) { disco.oneShot(ctx, download.Update{Bundle: serviceBundle}) - var dtConfig map[string]any - err = util.Unmarshal(manager.Config.DistributedTracing, &dtConfig) - if err != nil { + var dtConfig struct { + Type string `json:"type"` + } + if err := util.Unmarshal([]byte(manager.GetConfig().DistributedTracing), &dtConfig); err != nil { t.Fatal(err) } - - ty, ok := dtConfig["type"] - if !ok { + if dtConfig.Type == "" { t.Fatal("Expected config for distributed tracing") } - if ty != "grpc" { - t.Fatalf("Expected distributed tracing \"grpc\" but got %v", ty) + if dtConfig.Type != "grpc" { + t.Fatalf("Expected distributed tracing \"grpc\" but got %v", dtConfig.Type) } } @@ -2308,7 +2326,6 @@ func TestMergeValuesAndListOverrides(t *testing.T) { } func TestReconfigureWithUpdates(t *testing.T) { - ctx := context.Background() bootConfigRaw := []byte(`{ @@ -2644,13 +2661,20 @@ func TestReconfigureWithUpdates(t *testing.T) { t.Fatalf("Unexpected error %v", err) } - if manager.Config.PersistenceDirectory == nil { - t.Fatal("Erased persistence directory configuration") - } - if manager.Config.Discovery == nil { + cfg := manager.GetConfig() + + if len(cfg.Discovery) == 0 { t.Fatal("Erased discovery plugin configuration") } + var persistDir string + if cfg.PersistenceDirectory != nil { + persistDir = *cfg.PersistenceDirectory + } + if persistDir == "" { + t.Fatal("Erased persistence directory configuration") + } + // update persistence directory in the service config and check that its boot config value is not overridden updatedBundle = makeDataBundle(14, ` { @@ -2665,13 +2689,17 @@ func TestReconfigureWithUpdates(t *testing.T) { t.Fatalf("Unexpected error %v", err) } - if manager.Config.PersistenceDirectory == nil || *manager.Config.PersistenceDirectory == "my_bundles" { + cfg = manager.GetConfig() + persistDir = "" + if cfg.PersistenceDirectory != nil { + persistDir = *cfg.PersistenceDirectory + } + if persistDir == "" || persistDir == "my_bundles" { t.Fatal("Unexpected update to persistence directory") } } func TestProcessBundleWithSigning(t *testing.T) { - ctx := context.Background() manager, err := plugins.New([]byte(`{ @@ -2762,7 +2790,6 @@ func (ts *testServer) Stop() { } func (ts *testServer) handle(w http.ResponseWriter, r *http.Request) { - var update status.UpdateRequestV1 if err := json.NewDecoder(r.Body).Decode(&update); err != nil { @@ -2775,7 +2802,6 @@ func (ts *testServer) handle(w http.ResponseWriter, r *http.Request) { } func TestStatusUpdates(t *testing.T) { - ts := testServer{t: t} ts.Start() defer ts.Stop() @@ -2922,14 +2948,13 @@ func TestStatusUpdatesFromPersistedBundlesDontDelayBoot(t *testing.T) { } discoBundleDir := filepath.Join(dir, "bundles", "config") - if err := os.MkdirAll(discoBundleDir, 0755); err != nil { + if err := os.MkdirAll(discoBundleDir, 0o755); err != nil { t.Fatal(err) } discoBundleFile, err := os.Create(filepath.Join(discoBundleDir, "bundle.tar.gz")) if err != nil { t.Fatal(err) - } defer discoBundleFile.Close() @@ -2946,14 +2971,13 @@ func TestStatusUpdatesFromPersistedBundlesDontDelayBoot(t *testing.T) { } mainBundleDir := filepath.Join(dir, "bundles", "main") - if err := os.MkdirAll(mainBundleDir, 0755); err != nil { + if err := os.MkdirAll(mainBundleDir, 0o755); err != nil { t.Fatal(err) } mainBundleFile, err := os.Create(filepath.Join(mainBundleDir, "bundle.tar.gz")) if err != nil { t.Fatal(err) - } defer mainBundleFile.Close() @@ -3016,7 +3040,6 @@ func TestStatusUpdatesFromPersistedBundlesDontDelayBoot(t *testing.T) { } func TestStatusUpdatesTimestamp(t *testing.T) { - ts := testServer{t: t} ts.Start() defer ts.Stop() @@ -3088,7 +3111,6 @@ func TestStatusUpdatesTimestamp(t *testing.T) { } func TestStatusMetricsForLogDrops(t *testing.T) { - ctx := context.Background() testLogger := test.New() @@ -3317,7 +3339,7 @@ bundle: ` manager := getTestManager(t, conf) trigger := plugins.TriggerManual - _, err := getPluginSet(nil, manager, manager.Config, nil, nil, &trigger) + _, err := getPluginSet(nil, manager, manager.GetConfig(), nil, nil, &trigger) if err != nil { t.Fatalf("Unexpected error: %s", err) } @@ -3354,7 +3376,7 @@ bundles: ` manager := getTestManager(t, conf) trigger := plugins.TriggerManual - _, err := getPluginSet(nil, manager, manager.Config, nil, nil, &trigger) + _, err := getPluginSet(nil, manager, manager.GetConfig(), nil, nil, &trigger) if err != nil { t.Fatalf("Unexpected error: %s", err) } @@ -3411,10 +3433,9 @@ bundles: for name, tc := range tests { t.Run(name, func(t *testing.T) { - manager := getTestManager(t, tc.conf) trigger := plugins.TriggerManual - _, err := getPluginSet(nil, manager, manager.Config, nil, nil, &trigger) + _, err := getPluginSet(nil, manager, manager.GetConfig(), nil, nil, &trigger) if tc.wantErr { if err == nil { @@ -3432,7 +3453,6 @@ bundles: } func TestGetPluginSetWithBadManualTriggerDecisionLogConfig(t *testing.T) { - confGood := ` services: s1: @@ -3476,10 +3496,9 @@ decision_logs: for name, tc := range tests { t.Run(name, func(t *testing.T) { - manager := getTestManager(t, tc.conf) trigger := plugins.TriggerManual - _, err := getPluginSet(nil, manager, manager.Config, nil, nil, &trigger) + _, err := getPluginSet(nil, manager, manager.GetConfig(), nil, nil, &trigger) if tc.wantErr { if err == nil { @@ -3547,10 +3566,9 @@ status: for name, tc := range tests { t.Run(name, func(t *testing.T) { - manager := getTestManager(t, tc.conf) trigger := plugins.TriggerManual - _, err := getPluginSet(nil, manager, manager.Config, nil, nil, &trigger) + _, err := getPluginSet(nil, manager, manager.GetConfig(), nil, nil, &trigger) if tc.wantErr { if err == nil { @@ -3994,7 +4012,6 @@ func newTestFixture(t *testing.T) *testFixture { } func (t *testFixture) loop(ctx context.Context) { - for { select { case stop := <-t.stopCh: @@ -4047,7 +4064,6 @@ func (t *testFixture) runQuery(ctx context.Context, query string, m metrics.Metr } func (t *testFixture) log(ctx context.Context, query string, m metrics.Metrics, result *any) error { - record := server.Info{ Timestamp: time.Now(), Path: query, @@ -4267,7 +4283,6 @@ func (t *testFixtureServer) handle(w http.ResponseWriter, r *http.Request) { } else { t.t.Fatalf("unknown path %v", r.URL.Path) } - } func (t *testFixtureServer) start() { diff --git a/v1/plugins/plugins.go b/v1/plugins/plugins.go index de914fb3a2..ca8df1ee48 100644 --- a/v1/plugins/plugins.go +++ b/v1/plugins/plugins.go @@ -177,7 +177,8 @@ type StatusListener func(status map[string]*Status) // Manager implements lifecycle management of plugins and gives plugins access // to engine-wide components like storage. type Manager struct { - Store storage.Store + Store storage.Store + // Config values should be accessed from the thread-safe GetConfig method. Config *config.Config Info *ast.Term ID string @@ -225,11 +226,15 @@ type Manager struct { bundleActivatorPlugin string } -type managerContextKey string -type managerWasmResolverKey string +type ( + managerContextKey string + managerWasmResolverKey string +) -const managerCompilerContextKey = managerContextKey("compiler") -const managerWasmResolverContextKey = managerWasmResolverKey("wasmResolvers") +const ( + managerCompilerContextKey = managerContextKey("compiler") + managerWasmResolverContextKey = managerWasmResolverKey("wasmResolvers") +) // SetCompilerOnContext puts the compiler into the storage context. Calling this // function before committing updated policies to storage allows the manager to @@ -276,7 +281,6 @@ func validateTriggerMode(mode TriggerMode) error { // ValidateAndInjectDefaultsForTriggerMode validates the trigger mode and injects default values func ValidateAndInjectDefaultsForTriggerMode(a, b *TriggerMode) (*TriggerMode, error) { - if a == nil && b != nil { err := validateTriggerMode(*b) if err != nil { @@ -438,7 +442,6 @@ func WithBundleActivatorPlugin(bundleActivatorPlugin string) func(*Manager) { // New creates a new Manager using config. func New(raw []byte, id string, store storage.Store, opts ...func(*Manager)) (*Manager, error) { - parsedConfig, err := config.ParseConfig(raw, id) if err != nil { return nil, err @@ -531,7 +534,6 @@ func New(raw []byte, id string, store storage.Store, opts ...func(*Manager)) (*M // Init returns an error if the manager could not initialize itself. Init() should // be called before Start(). Init() is idempotent. func (m *Manager) Init(ctx context.Context) error { - if m.initialized { return nil } @@ -548,7 +550,6 @@ func (m *Manager) Init(ctx context.Context) error { } err := storage.Txn(ctx, m.Store, params, func(txn storage.Transaction) error { - result, err := initload.InsertAndCompile(ctx, initload.InsertAndCompileOptions{ Store: m.Store, Txn: txn, @@ -559,7 +560,6 @@ func (m *Manager) Init(ctx context.Context) error { ParserOptions: m.parserOptions, BundleActivatorPlugin: m.bundleActivatorPlugin, }) - if err != nil { return err } @@ -575,7 +575,6 @@ func (m *Manager) Init(ctx context.Context) error { _, err = m.Store.Register(ctx, txn, storage.TriggerConfig{OnCommit: m.onCommit}) return err }) - if err != nil { if m.stop != nil { done := make(chan struct{}) @@ -594,14 +593,24 @@ func (m *Manager) Init(ctx context.Context) error { func (m *Manager) Labels() map[string]string { m.mtx.Lock() defer m.mtx.Unlock() - return m.Config.Labels + + return maps.Clone(m.Config.Labels) } // InterQueryBuiltinCacheConfig returns the configuration for the inter-query caches. func (m *Manager) InterQueryBuiltinCacheConfig() *cache.Config { m.mtx.Lock() defer m.mtx.Unlock() - return m.interQueryBuiltinCacheConfig + + return m.interQueryBuiltinCacheConfig.Clone() +} + +// GetConfig returns a deep copy of the manager's configuration. +func (m *Manager) GetConfig() *config.Config { + m.mtx.Lock() + defer m.mtx.Unlock() + + return m.Config.Clone() } // Register adds a plugin to the manager. When the manager is started, all of @@ -749,7 +758,6 @@ func (m *Manager) setWasmResolvers(rs []*wasm.Resolver) { // Start starts the manager. Init() should be called once before Start(). func (m *Manager) Start(ctx context.Context) error { - if m == nil { return nil } @@ -831,7 +839,9 @@ func (m *Manager) DefaultServiceOpts(config *config.Config) cfg.ServiceOptions { } // Reconfigure updates the configuration on the manager. -func (m *Manager) Reconfigure(config *config.Config) error { +func (m *Manager) Reconfigure(newCfg *config.Config) error { + config := newCfg.Clone() + opts := m.DefaultServiceOpts(config) keys, err := keys.ParseKeysConfig(config.Keys) @@ -862,6 +872,7 @@ func (m *Manager) Reconfigure(config *config.Config) error { // don't erase persistence directory if config.PersistenceDirectory == nil { + // update is ok since we have the lock config.PersistenceDirectory = m.Config.PersistenceDirectory } @@ -912,7 +923,6 @@ func (m *Manager) UnregisterPluginStatusListener(name string) { // listeners will be called with a copy of the new state of all // plugins. func (m *Manager) UpdatePluginStatus(pluginName string, status *Status) { - var toNotify map[string]StatusListener var statuses map[string]*Status @@ -946,7 +956,6 @@ func (m *Manager) copyPluginStatus() map[string]*Status { } func (m *Manager) onCommit(ctx context.Context, txn storage.Transaction, event storage.TriggerEvent) { - compiler := GetCompilerOnContext(event.Context) // If the context does not contain the compiler fallback to loading the @@ -974,7 +983,6 @@ func (m *Manager) onCommit(ctx context.Context, txn storage.Transaction, event s resolvers := getWasmResolversOnContext(event.Context) if resolvers != nil { m.setWasmResolvers(resolvers) - } else if event.DataChanged() { if requiresWasmResolverReload(event) { resolvers, err := bundleUtils.LoadWasmResolversFromStore(ctx, m.Store, txn, nil) @@ -1057,7 +1065,19 @@ func (m *Manager) updateWasmResolversData(ctx context.Context, event storage.Tri func (m *Manager) PublicKeys() map[string]*keys.Config { m.mtx.Lock() defer m.mtx.Unlock() - return m.keys + + if m.keys == nil { + return make(map[string]*keys.Config) + } + + result := make(map[string]*keys.Config, len(m.keys)) + for k, v := range m.keys { + if v != nil { + copied := *v + result[k] = &copied + } + } + return result } // Client returns a client for communicating with a remote service. diff --git a/v1/plugins/plugins_test.go b/v1/plugins/plugins_test.go index 1109b57edd..a9458476fd 100644 --- a/v1/plugins/plugins_test.go +++ b/v1/plugins/plugins_test.go @@ -49,7 +49,7 @@ func TestManagerCacheTriggers(t *testing.T) { t.Fatal("Listeners should not be called yet") } - err = m.Reconfigure(m.Config) + err = m.Reconfigure(m.GetConfig()) if err != nil { t.Fatalf("Unexpected error: %s", err) } @@ -87,7 +87,7 @@ func TestManagerNDCacheTriggers(t *testing.T) { t.Fatal("Listeners should not be called yet") } - err = m.Reconfigure(m.Config) + err = m.Reconfigure(m.GetConfig()) if err != nil { t.Fatalf("Unexpected error: %s", err) } @@ -202,7 +202,6 @@ func (p *testPlugin) Reconfigure(context.Context, any) { } func TestPluginManagerLazyInitBeforePluginStart(t *testing.T) { - m, err := New([]byte(`{"plugins": {"someplugin": {"enabled": true}}}`), "test", inmem.New()) if err != nil { t.Fatal(err) @@ -219,11 +218,9 @@ func TestPluginManagerLazyInitBeforePluginStart(t *testing.T) { if !mock.Started { t.Fatal("expected plugin to be started") } - } func TestPluginManagerInitBeforePluginStart(t *testing.T) { - m, err := New([]byte(`{"plugins": {"someplugin": {}}}`), "test", inmem.New()) if err != nil { t.Fatal(err) @@ -244,11 +241,9 @@ func TestPluginManagerInitBeforePluginStart(t *testing.T) { if !mock.Started { t.Fatal("expected plugin to be started") } - } func TestPluginManagerInitIdempotence(t *testing.T) { - mockStore := mock.New() m, err := New([]byte(`{"plugins": {"someplugin": {}}}`), "test", mockStore) @@ -271,7 +266,6 @@ func TestPluginManagerInitIdempotence(t *testing.T) { if len(mockStore.Transactions) != exp { t.Fatal("expected num txns to be:", exp, "but got:", len(mockStore.Transactions)) } - } func TestManagerWithCachingConfig(t *testing.T) { @@ -310,8 +304,8 @@ func TestManagerWithNDCachingConfig(t *testing.T) { } expected := true - if !m.Config.NDBuiltinCache == expected { - t.Fatalf("want %+v got %+v", expected, m.Config.NDBuiltinCache) + if cfg := m.GetConfig(); !cfg.NDBuiltinCache == expected { + t.Fatalf("want %+v got %+v", expected, cfg.NDBuiltinCache) } // config error @@ -366,7 +360,6 @@ func TestPluginManagerAuthPlugin(t *testing.T) { } func TestPluginManagerLogger(t *testing.T) { - logger := logging.Get().WithFields(map[string]any{"context": "myloggincontext"}) m, err := New([]byte(`{}`), "test", inmem.New(), Logger(logger)) @@ -436,6 +429,7 @@ func TestPluginManagerTracerProvider(t *testing.T) { t.Fatal("TracerProvider was not configured on plugin manager") } } + func TestPluginManagerServerInitialized(t *testing.T) { // Verify that ServerInitializedChannel is closed when // ServerInitialized is called. @@ -481,14 +475,18 @@ func (*myAuthPluginMock) NewClient(c rest.Config) (*http.Client, error) { 10, ), nil } + func (*myAuthPluginMock) Prepare(*http.Request) error { return nil } + func (*myAuthPluginMock) Start(context.Context) error { return nil } + func (*myAuthPluginMock) Stop(context.Context) { } + func (*myAuthPluginMock) Reconfigure(context.Context, any) { } diff --git a/v1/rego/rego.go b/v1/rego/rego.go index 3540638472..8a47d90a93 100644 --- a/v1/rego/rego.go +++ b/v1/rego/rego.go @@ -840,7 +840,6 @@ type memo struct { type memokey string func memoize(decl *Function, bctx BuiltinContext, terms []*ast.Term, ifEmpty func() (*ast.Term, error)) (*ast.Term, error) { - if !decl.Memoize { return ifEmpty() } @@ -1339,7 +1338,6 @@ func EvalMode(mode ast.CompilerEvalMode) func(r *Rego) { // New returns a new Rego object. func New(options ...func(r *Rego)) *Rego { - r := &Rego{ parsedModules: map[string]*ast.Module{}, capture: map[*ast.Expr]ast.Var{}, @@ -1411,8 +1409,8 @@ func New(options ...func(r *Rego)) *Rego { } if r.pluginMgr != nil { - for _, name := range r.pluginMgr.Plugins() { - p := r.pluginMgr.Plugin(name) + for _, pluginName := range r.pluginMgr.Plugins() { + p := r.pluginMgr.Plugin(pluginName) if p0, ok := p.(TargetPlugin); ok { r.plugins = append(r.plugins, p0) } @@ -1574,7 +1572,6 @@ func CompilePartial(yes bool) CompileOption { // Compile returns a compiled policy query. func (r *Rego) Compile(ctx context.Context, opts ...CompileOption) (*CompileResult, error) { - var cfg CompileContext for _, opt := range opts { @@ -2119,7 +2116,6 @@ func parserOptionsFromRegoVersionImport(imports []*ast.Import, popts ast.ParserO } func (r *Rego) compileModules(ctx context.Context, txn storage.Transaction, m metrics.Metrics) error { - // Only compile again if there are new modules. if len(r.bundles) > 0 || len(r.parsedModules) > 0 { @@ -2230,7 +2226,6 @@ func (r *Rego) compileQuery(query ast.Body, imports []*ast.Import, _ metrics.Met compiled, err := qc.Compile(query) return qc, compiled, err - } func (r *Rego) eval(ctx context.Context, ectx *EvalContext) (ResultSet, error) { @@ -2319,7 +2314,6 @@ func (r *Rego) eval(ctx context.Context, ectx *EvalContext) (ResultSet, error) { rs = append(rs, result) return nil }) - if err != nil { return nil, err } @@ -2392,7 +2386,6 @@ func (r *Rego) valueToQueryResult(res ast.Value, ectx *EvalContext) (ResultSet, } func (r *Rego) generateResult(qr topdown.QueryResult, ectx *EvalContext) (Result, error) { - rewritten := ectx.compiledQuery.compiler.RewrittenVars() result := newResult() @@ -2432,7 +2425,6 @@ func (r *Rego) generateResult(qr topdown.QueryResult, ectx *EvalContext) (Result } func (r *Rego) partialResult(ctx context.Context, pCfg *PrepareConfig) (PartialResult, error) { - err := r.prepare(ctx, partialResultQueryType, []extraStage{ { after: "ResolveRefs", @@ -2526,7 +2518,6 @@ func (r *Rego) partialResult(ctx context.Context, pCfg *PrepareConfig) (PartialR } func (r *Rego) partial(ctx context.Context, ectx *EvalContext) (*PartialQueries, error) { - var unknowns []*ast.Term switch { @@ -2664,7 +2655,6 @@ func (r *Rego) partial(ctx context.Context, ectx *EvalContext) (*PartialQueries, } func (r *Rego) rewriteQueryToCaptureValue(_ ast.QueryCompiler, query ast.Body) (ast.Body, error) { - checkCapture := iteration(query) || len(query) > 1 for _, expr := range query { @@ -2779,7 +2769,6 @@ type transactionCloser func(ctx context.Context, err error) error // the configured Rego object. The returned function should be used to close the txn // regardless of status. func (r *Rego) getTxn(ctx context.Context) (storage.Transaction, transactionCloser, error) { - noopCloser := func(_ context.Context, _ error) error { return nil // no-op default } @@ -2889,7 +2878,6 @@ type refResolver struct { } func iteration(x any) bool { - var stopped bool vis := ast.NewGenericVisitor(func(x any) bool { diff --git a/v1/runtime/runtime.go b/v1/runtime/runtime.go index 493c24d0eb..8e8552c031 100644 --- a/v1/runtime/runtime.go +++ b/v1/runtime/runtime.go @@ -542,7 +542,7 @@ func NewRuntime(ctx context.Context, params Params) (*Runtime, error) { // extractMetricsConfig returns the configuration for server metrics and parsing errors if any func extractMetricsConfig(config []byte, params Params) (*metrics_config.Config, error) { - var opaParsedConfig, opaParsedConfigErr = opa_config.ParseConfig(config, params.ID) + opaParsedConfig, opaParsedConfigErr := opa_config.ParseConfig(config, params.ID) if opaParsedConfigErr != nil { return nil, opaParsedConfigErr } @@ -552,8 +552,8 @@ func extractMetricsConfig(config []byte, params Params) (*metrics_config.Config, serverMetricsData = opaParsedConfig.Server.Metrics } - var configBuilder = metrics_config.NewConfigBuilder() - var metricsParsedConfig, metricsParsedConfigErr = configBuilder.WithBytes(serverMetricsData).Parse() + configBuilder := metrics_config.NewConfigBuilder() + metricsParsedConfig, metricsParsedConfigErr := configBuilder.WithBytes(serverMetricsData).Parse() if metricsParsedConfigErr != nil { return nil, fmt.Errorf("server metrics configuration parse error: %w", metricsParsedConfigErr) } @@ -661,7 +661,7 @@ func (rt *Runtime) Serve(ctx context.Context) error { // If decision_logging plugin enabled, check to see if we opted in to the ND builtins cache. if lp := logs.Lookup(rt.Manager); lp != nil { - rt.server = rt.server.WithNDBCacheEnabled(rt.Params.NDBCacheEnabled || rt.Manager.Config.NDBuiltinCacheEnabled()) + rt.server = rt.server.WithNDBCacheEnabled(rt.Params.NDBCacheEnabled || rt.Manager.GetConfig().NDBuiltinCacheEnabled()) } if rt.Params.DiagnosticAddrs != nil { @@ -1065,7 +1065,7 @@ func generateDecisionID() string { } func verifyAuthorizationPolicySchema(m *plugins.Manager) error { - authorizationDecisionRef, err := ref.ParseDataPath(*m.Config.DefaultAuthorizationDecision) + authorizationDecisionRef, err := ref.ParseDataPath(*m.GetConfig().DefaultAuthorizationDecision) if err != nil { return err } diff --git a/v1/runtime/runtime_test.go b/v1/runtime/runtime_test.go index 815ee0e302..808b4bca22 100644 --- a/v1/runtime/runtime_test.go +++ b/v1/runtime/runtime_test.go @@ -125,7 +125,7 @@ func testRuntimeProcessWatchEvents(t *testing.T, asBundle bool, readAst bool) { "hello": "world-2", } - if err := os.WriteFile(path.Join(rootDir, "some/data.json"), util.MustMarshalJSON(expected), 0644); err != nil { + if err := os.WriteFile(path.Join(rootDir, "some/data.json"), util.MustMarshalJSON(expected), 0o644); err != nil { panic(err) } @@ -222,7 +222,7 @@ func testRuntimeProcessWatchEventPolicyError(t *testing.T, asBundle bool) { default x = 2`) - if err := os.WriteFile(path.Join(rootDir, "y.rego"), newModule, 0644); err != nil { + if err := os.WriteFile(path.Join(rootDir, "y.rego"), newModule, 0o644); err != nil { t.Fatal(err) } @@ -280,7 +280,6 @@ func testRuntimeProcessWatchEventPolicyError(t *testing.T, asBundle bool) { if err != nil { t.Fatalf("Expected result to succeed before %v. Last error: %v", maxWait, err) } - }) } @@ -470,7 +469,7 @@ p contains 1 if { output.Reset() // write new policy to disk, to trigger the watcher - if err := os.WriteFile(path.Join(rootDir, "authz.rego"), []byte(tc.policy), 0644); err != nil { + if err := os.WriteFile(path.Join(rootDir, "authz.rego"), []byte(tc.policy), 0o644); err != nil { t.Fatal(err) } @@ -624,7 +623,7 @@ p contains 1 if { } // write new policy to disk, to trigger the watcher - if err := os.WriteFile(path.Join(rootDir, "authz.rego"), []byte(tc.policy), 0644); err != nil { + if err := os.WriteFile(path.Join(rootDir, "authz.rego"), []byte(tc.policy), 0o644); err != nil { t.Fatal(err) } @@ -819,7 +818,7 @@ func TestRuntimeWithAuthzSchemaVerification(t *testing.T) { input.identty = "foo" }`) - if err := os.WriteFile(path.Join(rootDir, "authz.rego"), badModule, 0644); err != nil { + if err := os.WriteFile(path.Join(rootDir, "authz.rego"), badModule, 0o644); err != nil { t.Fatal(err) } @@ -917,7 +916,6 @@ func TestCheckAuthIneffective(t *testing.T) { if !strings.Contains(stdout.String(), expected) { t.Fatalf("Expected output to contain: \"%v\" but got \"%v\"", expected, stdout.String()) } - } func TestServerInitialized(t *testing.T) { @@ -1476,26 +1474,38 @@ func TestUrlPathToConfigOverride(t *testing.T) { t.Fatal(err) } - var serviceConf map[string]any - if err = json.Unmarshal(rt.Manager.Config.Services, &serviceConf); err != nil { - t.Fatal(err) + cfg := rt.Manager.GetConfig() + + var servicesConfig map[string]map[string]any + if len(cfg.Services) > 0 { + if err := json.Unmarshal([]byte(cfg.Services), &servicesConfig); err != nil { + t.Fatal(err) + } } - cliService, ok := serviceConf["cli1"].(map[string]any) + cliService, ok := servicesConfig["cli1"] if !ok { - t.Fatal("excpected service configuration for 'cli1' service") + t.Fatal("expected service configuration for 'cli1' service") } if cliService["url"] != "https://www.example.com" { t.Error("expected cli1 service url value: 'https://www.example.com'") } - var bundleConf map[string]any - if err = json.Unmarshal(rt.Manager.Config.Bundles, &bundleConf); err != nil { - t.Fatal(err) + bundleConf := make(map[string]map[string]any) + if len(cfg.Bundles) > 0 { + var bundleConfRaw map[string]any + if err := json.Unmarshal([]byte(cfg.Bundles), &bundleConfRaw); err != nil { + t.Fatal(err) + } + for k, v := range bundleConfRaw { + if bundleMap, ok := v.(map[string]any); ok { + bundleConf[k] = bundleMap + } + } } - cliBundle, ok := bundleConf["cli1"].(map[string]any) + cliBundle, ok := bundleConf["cli1"] if !ok { t.Fatal("excpected bundle configuration for 'cli1' bundle") } @@ -1882,7 +1892,7 @@ func TestConfigHookAndNonReplacedEnvVars(t *testing.T) { hk := configHook{} cf := filepath.Join(t.TempDir(), "opa.yaml") - if err := os.WriteFile(cf, []byte("some: ${thing}\n"), 0755); err != nil { + if err := os.WriteFile(cf, []byte("some: ${thing}\n"), 0o755); err != nil { t.Fatal(err) } @@ -2110,7 +2120,7 @@ default allow := false # Reject requests by default. allow if { input.method == "POST" input.path == ["exp", "foo"] - input.body.example == "A" + input.body.example == "A" }`) rt, err := NewRuntime(ctx, params) diff --git a/v1/sdk/opa.go b/v1/sdk/opa.go index 92d75d0eb6..152a640c3c 100644 --- a/v1/sdk/opa.go +++ b/v1/sdk/opa.go @@ -143,7 +143,6 @@ func (opa *OPA) Plugin(name string) plugins.Plugin { // function is atomic. If the configuration update cannot be successfully // applied, the old configuration will remain intact. func (opa *OPA) Configure(ctx context.Context, opts ConfigOptions) error { - if err := opts.init(); err != nil { return err } @@ -205,7 +204,6 @@ func (opa *OPA) configure(ctx context.Context, bs []byte, ready chan struct{}, b }) manager.RegisterPluginStatusListener("sdk", func(status map[string]*plugins.Status) { - select { case <-ready: return @@ -283,7 +281,6 @@ func (opa *OPA) configure(ctx context.Context, bs []byte, ready chan struct{}, b // Stop closes the OPA. The OPA cannot be restarted. func (opa *OPA) Stop(ctx context.Context) { - opa.mtx.Lock() mgr := opa.state.manager opa.mtx.Unlock() @@ -295,7 +292,6 @@ func (opa *OPA) Stop(ctx context.Context) { // Decision returns a named decision. This function is threadsafe. func (opa *OPA) Decision(ctx context.Context, options DecisionOptions) (*DecisionResult, error) { - record := server.Info{ Timestamp: options.Now, Path: options.Path, @@ -394,7 +390,7 @@ func (opa *OPA) executeTransaction(ctx context.Context, record *server.Info, wor } if record.Path == "" { - record.Path = *s.manager.Config.DefaultDecision + record.Path = *s.manager.GetConfig().DefaultDecision } record.Txn, record.Error = s.manager.Store.NewTransaction(ctx, storage.TransactionParams{}) @@ -429,7 +425,6 @@ func (opa *OPA) executeTransaction(ctx context.Context, record *server.Info, wor // Note(philipc): The NDBCache is unused here, because non-deterministic // builtins are not run during partial evaluation. func (opa *OPA) Partial(ctx context.Context, options PartialOptions) (*PartialResult, error) { - if options.Mapper == nil { options.Mapper = &RawMapper{} } @@ -568,7 +563,6 @@ type evalArgs struct { } func evaluate(ctx context.Context, args evalArgs) (any, types.ProvenanceV1, ast.Value, map[string]server.BundleInfo, error) { - provenance := types.ProvenanceV1{ Version: version.Version, Vcs: version.Vcs, @@ -657,7 +651,6 @@ type partialEvalArgs struct { } func partial(ctx context.Context, args partialEvalArgs) (*rego.PartialQueries, types.ProvenanceV1, ast.Value, map[string]server.BundleInfo, error) { - provenance := types.ProvenanceV1{ Version: version.Version, Bundles: make(map[string]types.ProvenanceBundleV1), diff --git a/v1/server/server.go b/v1/server/server.go index f9225ed857..971ac28ec0 100644 --- a/v1/server/server.go +++ b/v1/server/server.go @@ -26,8 +26,6 @@ import ( "time" "github.com/open-policy-agent/opa/v1/hooks" - serverDecodingPlugin "github.com/open-policy-agent/opa/v1/plugins/server/decoding" - serverEncodingPlugin "github.com/open-policy-agent/opa/v1/plugins/server/encoding" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/trace" @@ -42,6 +40,8 @@ import ( "github.com/open-policy-agent/opa/v1/metrics" "github.com/open-policy-agent/opa/v1/plugins" bundlePlugin "github.com/open-policy-agent/opa/v1/plugins/bundle" + serverDecodingPlugin "github.com/open-policy-agent/opa/v1/plugins/server/decoding" + serverEncodingPlugin "github.com/open-policy-agent/opa/v1/plugins/server/encoding" "github.com/open-policy-agent/opa/v1/plugins/status" "github.com/open-policy-agent/opa/v1/rego" "github.com/open-policy-agent/opa/v1/server/authorizer" @@ -661,7 +661,6 @@ func (s *Server) getListenerForHTTPServer(u *url.URL, h http.Handler, t httpList } func (s *Server) getListenerForHTTPSServer(u *url.URL, h http.Handler, t httpListenerType) (Loop, httpListener, error) { - if s.cert == nil { return nil, nil, errors.New("TLS certificate required but not supplied") } @@ -764,7 +763,7 @@ func (s *Server) initHandlerAuthz(handler http.Handler) http.Handler { s.getCompiler, s.store, authorizer.Runtime(s.runtime), - authorizer.Decision(s.manager.Config.DefaultAuthorizationDecisionRef), + authorizer.Decision(s.manager.GetConfig().DefaultAuthorizationDecisionRef), authorizer.PrintHook(s.manager.PrintHook()), authorizer.EnablePrintStatements(s.manager.EnablePrintStatements()), authorizer.InterQueryCache(s.interQueryBuiltinCache), @@ -783,10 +782,10 @@ func (s *Server) initHandlerAuthz(handler http.Handler) http.Handler { // it passes the size limit down the body-reading method via the request // context. func (s *Server) initHandlerDecodingLimits(handler http.Handler) (http.Handler, error) { - var decodingRawConfig json.RawMessage - serverConfig := s.manager.Config.Server - if serverConfig != nil { - decodingRawConfig = serverConfig.Decoding + cfg := s.manager.GetConfig() + var decodingRawConfig []byte + if cfg.Server != nil { + decodingRawConfig = []byte(cfg.Server.Decoding) } decodingConfig, err := serverDecodingPlugin.NewConfigBuilder().WithBytes(decodingRawConfig).Parse() if err != nil { @@ -798,10 +797,10 @@ func (s *Server) initHandlerDecodingLimits(handler http.Handler) (http.Handler, } func (s *Server) initHandlerCompression(handler http.Handler) (http.Handler, error) { - var encodingRawConfig json.RawMessage - serverConfig := s.manager.Config.Server - if serverConfig != nil { - encodingRawConfig = serverConfig.Encoding + cfg := s.manager.GetConfig() + var encodingRawConfig []byte + if cfg.Server != nil { + encodingRawConfig = []byte(cfg.Server.Encoding) } encodingConfig, err := serverEncodingPlugin.NewConfigBuilder().WithBytes(encodingRawConfig).Parse() if err != nil { @@ -1018,7 +1017,6 @@ type bundleRevisions struct { } func getRevisions(ctx context.Context, store storage.Store, txn storage.Transaction) (bundleRevisions, error) { - var err error var br bundleRevisions br.Revisions = map[string]string{} @@ -1047,7 +1045,6 @@ func getRevisions(ctx context.Context, store storage.Store, txn storage.Transact } func (s *Server) reload(context.Context, storage.Transaction, storage.TriggerEvent) { - // NOTE(tsandall): We currently rely on the storage txn to provide // critical sections in the server. // @@ -1172,7 +1169,7 @@ func (s *Server) v0QueryPath(w http.ResponseWriter, r *http.Request, urlPath str return } - var messageType = types.MsgMissingError + messageType := types.MsgMissingError if len(s.getCompiler().GetRulesForVirtualDocument(ref)) > 0 { messageType = types.MsgFoundUndefinedError } @@ -1236,7 +1233,6 @@ func (s *Server) canEval(ctx context.Context) bool { } func (*Server) bundlesReady(pluginStatuses map[string]*plugins.Status) bool { - // Look for a discovery plugin first, if it exists and isn't ready // then don't bother with the others. // Note: use "discovery" instead of `discovery.Name` to avoid import @@ -2099,7 +2095,6 @@ func (s *Server) v1PoliciesGet(w http.ResponseWriter, r *http.Request) { } func (s *Server) v1PoliciesList(w http.ResponseWriter, r *http.Request) { - ctx := r.Context() txn, err := s.store.NewTransaction(ctx) @@ -2387,11 +2382,12 @@ func (s *Server) v1QueryPost(w http.ResponseWriter, r *http.Request) { } func (s *Server) v1ConfigGet(w http.ResponseWriter, r *http.Request) { - result, err := s.manager.Config.ActiveConfig() + result, err := s.manager.GetConfig().ActiveConfig() if err != nil { writer.ErrorAuto(w, err) return } + writer.JSONOK(w, types.ConfigResponseV1{Result: &result}, pretty(r)) } @@ -2407,7 +2403,6 @@ func (s *Server) v1StatusGet(w http.ResponseWriter, r *http.Request) { } func (s *Server) checkPolicyIDScope(ctx context.Context, txn storage.Transaction, id string) error { - bs, err := s.store.GetPolicy(ctx, txn, id) if err != nil { return err @@ -2422,7 +2417,6 @@ func (s *Server) checkPolicyIDScope(ctx context.Context, txn storage.Transaction } func (s *Server) checkPolicyPackageScope(ctx context.Context, txn storage.Transaction, pkg *ast.Package) error { - path, err := pkg.Path.Ptr() if err != nil { return err @@ -2448,7 +2442,6 @@ func (s *Server) getMetrics(r *http.Request) metrics.Metrics { } func (s *Server) checkPathScope(ctx context.Context, txn storage.Transaction, path storage.Path) error { - names, err := bundle.ReadBundleNamesFromStore(ctx, s.store, txn) if err != nil { if !storage.IsNotFound(err) { @@ -2547,7 +2540,6 @@ func (s *Server) abortAuto(ctx context.Context, txn storage.Transaction, w http. } func (s *Server) loadModules(ctx context.Context, txn storage.Transaction) (map[string]*ast.Module, error) { - ids, err := s.store.ListPolicies(ctx, txn) if err != nil { return nil, err @@ -2647,7 +2639,6 @@ func parseRefQuery(str string) (ast.Body, error) { } func (*Server) prepareV1PatchSlice(root string, ops []types.PatchV1) (result []patchImpl, err error) { - root = "/" + strings.Trim(root, "/") for _, op := range ops { @@ -2700,7 +2691,6 @@ func (s *Server) generateDecisionID() string { } func (s *Server) getProvenance(br bundleRevisions) *types.ProvenanceV1 { - p := &types.ProvenanceV1{ Version: version.Version, Vcs: version.Vcs, @@ -2730,7 +2720,7 @@ func (s *Server) hasLegacyBundle(br bundleRevisions) bool { func (s *Server) generateDefaultDecisionPath() string { // Assume the path is safe to transition back to a url - p, _ := s.manager.Config.DefaultDecisionRef().Ptr() + p, _ := s.manager.GetConfig().DefaultDecisionRef().Ptr() return p } @@ -2824,7 +2814,6 @@ func getBoolParam(url *url.URL, name string, ifEmpty bool) bool { } func getStringSliceParam(url *url.URL, name string) []string { - p, ok := url.Query()[name] if !ok { return nil @@ -2860,7 +2849,6 @@ func getExplain(url *url.URL, zero types.ExplainModeV1) types.ExplainModeV1 { } func readInputV0(r *http.Request) (ast.Value, *any, error) { - parsed, ok := authorizer.GetBodyOnContext(r.Context()) if ok { v, err := ast.InterfaceToValue(parsed) @@ -2902,7 +2890,6 @@ func readInputGetV1(str string) (ast.Value, *any, error) { } func readInputPostV1(r *http.Request) (ast.Value, *any, error) { - parsed, ok := authorizer.GetBodyOnContext(r.Context()) if ok { if obj, ok := parsed.(map[string]any); ok { diff --git a/v1/server/server_test.go b/v1/server/server_test.go index 9bfc7abdec..55c147dd82 100644 --- a/v1/server/server_test.go +++ b/v1/server/server_test.go @@ -1750,9 +1750,7 @@ func TestDataV1Metrics(t *testing.T) { func TestConfigV1(t *testing.T) { t.Parallel() - f := newFixture(t) - - c := []byte(`{"services": { + c := `{"services": { "acmecorp": { "url": "https://example.com/control-plane-api/v1", "credentials": {"bearer": {"token": "test"}} @@ -1763,21 +1761,16 @@ func TestConfigV1(t *testing.T) { }, "keys": { "global_key": { - "algorithm": HS256, + "algorithm": "HS256", "key": "secret" } - }}`) + }}` - conf, err := config.ParseConfig(c, "foo") - if err != nil { - t.Fatal(err) - } - - f.server.manager.Config = conf + f := newFixtureWithConfig(t, c) expected := map[string]any{ "result": map[string]any{ - "labels": map[string]any{"id": "foo", "version": version.Version, "region": "west"}, + "labels": map[string]any{"id": "test", "version": version.Version, "region": "west"}, "keys": map[string]any{"global_key": map[string]any{"algorithm": "HS256"}}, "services": map[string]any{"acmecorp": map[string]any{"url": "https://example.com/control-plane-api/v1"}}, "default_authorization_decision": "/system/authz/allow", @@ -1792,23 +1785,61 @@ func TestConfigV1(t *testing.T) { if err := f.v1(http.MethodGet, "/config", "", 200, string(bs)); err != nil { t.Fatal(err) } +} +func TestConfigV1WithInvalidConfig(t *testing.T) { + t.Parallel() + + // build some invalid config to forcibly load badServicesConfig := []byte(`{ "services": { "acmecorp": ["foo"] } }`) - conf, err = config.ParseConfig(badServicesConfig, "foo") + conf, err := config.ParseConfig(badServicesConfig, "foo") if err != nil { t.Fatal(err) } - f.server.manager.Config = conf + // create a new server and manager + ctx := context.Background() + server := New(). + WithAddresses([]string{"localhost:8182"}). + WithStore(inmem.New()) + + m, err := plugins.New([]byte{}, "test", server.store) + if err != nil { + t.Fatal(err) + } + + // NOTE: This is the only place we update the manager config directly. + // We do this to create an invalid configuration that would be impossible + // to set through normal Reconfigure call. This is done without + // starting/running the manager to be thread-safe. The manager does not + // need to be running in this test as the server just needs to access the + // manager config value. + m.Config = conf + + server = server.WithManager(m) + if err := m.Start(ctx); err != nil { + t.Fatal(err) + } + + server, err = server.Init(ctx) + if err != nil { + t.Fatal(err) + } + + f := &fixture{ + server: server, + recorder: httptest.NewRecorder(), + t: t, + } if err := f.v1(http.MethodGet, "/config", "", 500, `{ - "code": "internal_error", - "message": "type assertion error"}`); err != nil { + "code": "internal_error", + "message": "type assertion error"}`); err != nil { t.Fatal(err) } } @@ -4772,7 +4803,14 @@ func TestUnversionedPost(t *testing.T) { // update the default decision path s := "http/authz" - f.server.manager.Config.DefaultDecision = &s + + cfg := f.server.manager.GetConfig() + cfg.DefaultDecision = &s + + err := f.server.manager.Reconfigure(cfg) + if err != nil { + t.Fatal(err) + } f.reset() f.server.Handler.ServeHTTP(f.recorder, post()) diff --git a/v1/topdown/cache/cache.go b/v1/topdown/cache/cache.go index 60f38aaba2..d514bed787 100644 --- a/v1/topdown/cache/cache.go +++ b/v1/topdown/cache/cache.go @@ -43,12 +43,40 @@ type Config struct { InterQueryBuiltinValueCache InterQueryBuiltinValueCacheConfig `json:"inter_query_builtin_value_cache"` } +// Clone creates a deep copy of Config. +func (c *Config) Clone() *Config { + if c == nil { + return nil + } + + return &Config{ + InterQueryBuiltinCache: *c.InterQueryBuiltinCache.Clone(), + InterQueryBuiltinValueCache: *c.InterQueryBuiltinValueCache.Clone(), + } +} + // NamedValueCacheConfig represents the configuration of a named cache that built-in functions can utilize. // A default configuration to be used if not explicitly configured can be registered using RegisterDefaultInterQueryBuiltinValueCacheConfig. type NamedValueCacheConfig struct { MaxNumEntries *int `json:"max_num_entries,omitempty"` } +// Clone creates a deep copy of NamedValueCacheConfig. +func (n *NamedValueCacheConfig) Clone() *NamedValueCacheConfig { + if n == nil { + return nil + } + + clone := &NamedValueCacheConfig{} + + if n.MaxNumEntries != nil { + maxEntries := *n.MaxNumEntries + clone.MaxNumEntries = &maxEntries + } + + return clone +} + // InterQueryBuiltinValueCacheConfig represents the configuration of the inter-query value cache that built-in functions can utilize. // MaxNumEntries - max number of cache entries type InterQueryBuiltinValueCacheConfig struct { @@ -56,6 +84,29 @@ type InterQueryBuiltinValueCacheConfig struct { NamedCacheConfigs map[string]*NamedValueCacheConfig `json:"named,omitempty"` } +// Clone creates a deep copy of InterQueryBuiltinValueCacheConfig. +func (i *InterQueryBuiltinValueCacheConfig) Clone() *InterQueryBuiltinValueCacheConfig { + if i == nil { + return nil + } + + clone := &InterQueryBuiltinValueCacheConfig{} + + if i.MaxNumEntries != nil { + maxEntries := *i.MaxNumEntries + clone.MaxNumEntries = &maxEntries + } + + if i.NamedCacheConfigs != nil { + clone.NamedCacheConfigs = make(map[string]*NamedValueCacheConfig, len(i.NamedCacheConfigs)) + for k, v := range i.NamedCacheConfigs { + clone.NamedCacheConfigs[k] = v.Clone() + } + } + + return clone +} + // InterQueryBuiltinCacheConfig represents the configuration of the inter-query cache that built-in functions can utilize. // MaxSizeBytes - max capacity of cache in bytes // ForcedEvictionThresholdPercentage - capacity usage in percentage after which forced FIFO eviction starts @@ -66,6 +117,32 @@ type InterQueryBuiltinCacheConfig struct { StaleEntryEvictionPeriodSeconds *int64 `json:"stale_entry_eviction_period_seconds,omitempty"` } +// Clone creates a deep copy of InterQueryBuiltinCacheConfig. +func (i *InterQueryBuiltinCacheConfig) Clone() *InterQueryBuiltinCacheConfig { + if i == nil { + return nil + } + + clone := &InterQueryBuiltinCacheConfig{} + + if i.MaxSizeBytes != nil { + maxSize := *i.MaxSizeBytes + clone.MaxSizeBytes = &maxSize + } + + if i.ForcedEvictionThresholdPercentage != nil { + threshold := *i.ForcedEvictionThresholdPercentage + clone.ForcedEvictionThresholdPercentage = &threshold + } + + if i.StaleEntryEvictionPeriodSeconds != nil { + period := *i.StaleEntryEvictionPeriodSeconds + clone.StaleEntryEvictionPeriodSeconds = &period + } + + return clone +} + // ParseCachingConfig returns the config for the inter-query cache. func ParseCachingConfig(raw []byte) (*Config, error) { if raw == nil { diff --git a/v1/topdown/cache/cache_test.go b/v1/topdown/cache/cache_test.go index 5437f1f482..98ca55f52b 100644 --- a/v1/topdown/cache/cache_test.go +++ b/v1/topdown/cache/cache_test.go @@ -12,6 +12,7 @@ import ( "time" "github.com/open-policy-agent/opa/v1/ast" + "github.com/open-policy-agent/opa/v1/util/test" ) func TestParseCachingConfig(t *testing.T) { @@ -477,7 +478,6 @@ func TestConcurrentInsert(t *testing.T) { cacheValue2 := newInterQueryCacheValue(ast.String("bar2"), 5) cache.Insert(ast.String("foo2"), cacheValue2) - }() } wg.Wait() @@ -773,7 +773,6 @@ func TestCancelNewInterQueryCacheWithContext(t *testing.T) { if fetchedCacheValue, found := cache.Get(ast.StringTerm("foo").Value); !found { t.Fatalf("Expected cache entry with value %v for foo, found %v", cacheValue, fetchedCacheValue) } - } func TestUpdateConfig(t *testing.T) { @@ -853,3 +852,34 @@ func (p testInterQueryCacheValue) SizeInBytes() int64 { func (p testInterQueryCacheValue) Clone() (InterQueryCacheValue, error) { return &testInterQueryCacheValue{value: p.value, size: p.size}, nil } + +func TestConfigClone(t *testing.T) { + // test nil config + var nilConfig *Config + cloned := nilConfig.Clone() + if cloned != nil { + t.Fatal("expected nil clone for nil config") + } + + // test config with all fields populated using reflection + original := test.PopulateAllFields[Config](t) + + cloned = original.Clone() + if cloned == nil { + t.Fatal("clone returned nil") + } + + if cloned == original { + t.Fatal("clone should be different instance") + } + + if !reflect.DeepEqual(original, cloned) { + t.Errorf("clone differs from original") + } + + // Test that modifying the clone doesn't affect the original + *cloned.InterQueryBuiltinCache.MaxSizeBytes = 999 + if *original.InterQueryBuiltinCache.MaxSizeBytes == 999 { + t.Errorf("modifying clone affected original") + } +} diff --git a/v1/util/test/populate.go b/v1/util/test/populate.go new file mode 100644 index 0000000000..4e6c28428e --- /dev/null +++ b/v1/util/test/populate.go @@ -0,0 +1,138 @@ +// Copyright 2025 The OPA Authors. All rights reserved. +// Use of this source code is governed by an Apache2 +// license that can be found in the LICENSE file. + +package test + +import ( + "encoding/json" + "fmt" + "reflect" + "testing" +) + +// PopulateAllFields uses reflection to populate all fields of a struct with test data. +// This is useful for testing code that must handle all fields but when new fields +// might be added and missed. It must be possible to set all fields, or the helper +// will fail until the fields are supported. +// Caveats: only supports types needed at time of implementation, will not work +// on recursive structs. +func PopulateAllFields[T any](t *testing.T) *T { + t.Helper() + + var instance T + instancePtr := &instance + instanceType := reflect.TypeOf(instance) + instanceValue := reflect.ValueOf(instancePtr).Elem() + + populateStruct(t, instanceType, instanceValue) + + return instancePtr +} + +func populateStruct(t *testing.T, structType reflect.Type, structValue reflect.Value) { + t.Helper() + + for i := range structType.NumField() { + field := structType.Field(i) + fieldValue := structValue.Field(i) + + if !fieldValue.CanSet() { + continue + } + + if !populateDefaultTypes(t, field.Type, fieldValue, i) { + t.Fatalf("Unknown field type %s for field %s - update PopulateAllFields()", field.Type, field.Name) + } + } +} + +func populateDefaultTypes(t *testing.T, fieldType reflect.Type, fieldValue reflect.Value, index int) bool { + t.Helper() + + switch fieldType.Kind() { + case reflect.Slice: + if fieldType == reflect.TypeOf(json.RawMessage{}) { + fieldValue.Set(reflect.ValueOf([]byte(fmt.Sprintf(`{"test": "bar-%d"}`, index)))) + return true + } + + case reflect.Ptr: + switch fieldType.Elem().Kind() { + case reflect.String: + testString := fmt.Sprintf("test-value-%d", index) + fieldValue.Set(reflect.ValueOf(&testString)) + + return true + case reflect.Int: + testInt := 100 + index // unique value per field + fieldValue.Set(reflect.ValueOf(&testInt)) + + return true + case reflect.Int64: + testInt64 := int64(200 + index) // unique value per field + fieldValue.Set(reflect.ValueOf(&testInt64)) + + return true + case reflect.Struct: + newStruct := reflect.New(fieldType.Elem()) + populateStruct(t, fieldType.Elem(), newStruct.Elem()) + fieldValue.Set(newStruct) + + return true + } + + case reflect.Bool: + fieldValue.SetBool(true) + + return true + + case reflect.Struct: + populateStruct(t, fieldType, fieldValue) + + return true + + case reflect.Map: + switch { + case fieldType.Key().Kind() == reflect.String && fieldType.Elem().Kind() == reflect.String: + fieldValue.Set(reflect.ValueOf(map[string]string{ + "env": fmt.Sprintf("test-%d", index), + "version": fmt.Sprintf("1.%d", index), + })) + + return true + + case fieldType.Key().Kind() == reflect.String && + fieldType.Elem() == reflect.TypeOf(json.RawMessage{}): + + fieldValue.Set(reflect.ValueOf(map[string]json.RawMessage{ + "key1": []byte(fmt.Sprintf(`{"test": "bar-%d"}`, index)), + "key2": []byte(fmt.Sprintf(`{"foo": "baz-%d"}`, index)), + })) + + return true + + case fieldType.Key().Kind() == reflect.String && + fieldType.Elem().Kind() == reflect.Ptr && + fieldType.Elem().Elem().Kind() == reflect.Struct: + + elemType := fieldType.Elem().Elem() + + mapVal := reflect.MakeMap(fieldType) + + for _, key := range []string{"test1", "test2"} { + newElem := reflect.New(elemType) + + populateStruct(t, elemType, newElem.Elem()) + + mapVal.SetMapIndex(reflect.ValueOf(key), newElem) + } + + fieldValue.Set(mapVal) + + return true + } + } + + return false +}