ewilliams28/releases
1
0
Fork 0
mirror of https://github.com/open-policy-agent/opa.git synced 2026-08-12 19:32:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

v1/plugins: Address race in config access (#7825)

Browse Source 
* v1/plugins: Address race in config access

I ran into this race condition on another PR:

https://github.com/open-policy-agent/opa/actions/runs/16655603110/job/47139789057

I have tried to make all manager.Config access thread-safe by adding new
getters for used values. GetConfig is regrettably based on a JSON
roundtrip deep copy of the config. This us used in tests (fine) but also
in the discovery plugin:

https://github.com/open-policy-agent/opa/blob/2d014a89bbbc307d7204817220146ffae992e838/v1/plugins/discovery/discovery.go#L122

getPluginSet is very tightly coupled to the manager.Config and because
of it's dependencies on status and the other plugins packages, it's hard
to break out.

So, for now, I think this is an improvement and worth getting a second
opinion on before more refactoring.

Signed-off-by: Charlie Egan <charlie@styra.com>

* v1/config: Use add Clone to config

This makes the use of the manager's config more thread-safe and
consistent without more API changes.

Signed-off-by: Charlie Egan <charlie@styra.com>

* topdown: Add clone() funcs for config structs

NamedValueCacheConfig.Clone, InterQueryBuiltinValueCacheConfig.Clone and
InterQueryBuiltinCacheConfig.Clone have been added.

All Clone methods return a deep copy of the struct. This is tested for
missed new fields using PopulateAllFields, a generic function that
stuffs structs with values for all fields.

Signed-off-by: Charlie Egan <charlie@styra.com>

* plugins: Clone new config

Signed-off-by: Charlie Egan <charlie@styra.com>

---------

Signed-off-by: Charlie Egan <charlie@styra.com>
This commit is contained in:
Charlie Egan
2025-08-13 12:24:16 +00:00
committed by GitHub
parent 5faa4e2e67
commit 11e52c4df6
17 changed files with 757 additions and 290 deletions
Download Patch File Download Diff File Expand all files Collapse all files
v1/config/config.go
+174 -41
View File
@@ -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:
v1/config/config_test.go
+40 -3
View File
@@ -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)
}
}
v1/plugins/bundle/plugin.go
+4 -13
View File
@@ -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 {
v1/plugins/bundle/plugin_test.go
+29 -17
View File
@@ -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)
}
v1/plugins/discovery/discovery.go
+25 -25
View File
@@ -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)
v1/plugins/discovery/discovery_test.go
+84 -69
View File
@@ -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() {
v1/plugins/plugins.go
+39 -19
View File
@@ -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.
v1/plugins/plugins_test.go
+9 -11
View File
@@ -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) {
}
v1/rego/rego.go
+2 -14
View File
@@ -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 {
v1/runtime/runtime.go
+5 -5
View File
@@ -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
}
v1/runtime/runtime_test.go
+28 -18
View File
@@ -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)
v1/sdk/opa.go
+1 -8
View File
@@ -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),
v1/server/server.go
+15 -28
View File
@@ -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 {
v1/server/server_test.go
+55 -17
View File
@@ -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())
v1/topdown/cache/cache.go
Vendored
+77
View File
@@ -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 {
v1/topdown/cache/cache_test.go
Vendored
+32 -2
View File
@@ -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")
}
}
v1/util/test/populate.go
+138
View File
@@ -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
}