diff --git a/README.md b/README.md index 63ea974..f3ec34f 100644 --- a/README.md +++ b/README.md @@ -194,6 +194,89 @@ validator "webhook" "some_webhook_validator" { } ``` +## OPA Bundles + +The `opa` and `opa_json_patch` controllers above read a single Rego file from disk, so changing a policy means redeploying NACP. OPA *bundles* instead let NACP pull policy from a bundle server at runtime, on OPA's own refresh schedule, so rules can be shipped independently of the proxy. + +Declare one `opa_bundle` block per bundle source. `config_path` points at a regular [OPA configuration file](https://www.openpolicyagent.org/docs/latest/configuration/) — NACP passes it to the OPA SDK verbatim, so services, bundles, signing, decision logs and status settings all work as documented upstream. + +```hcl +opa_bundle "platform" { + config_path = "/local/opa-platform.yml" + + # Optional. How long startup waits for the first bundle activation. Default 30s. + ready_timeout = "30s" + # Optional. Bounds a single policy evaluation. Default 5s; "0s" to disable. + decision_timeout = "5s" + # Optional. Refuse to start unless every bundle verifies signatures. Default false. + require_signing = true +} + +validator "opa_bundle" "costcenter" { + bundle_rule { + source = "platform" # the opa_bundle id; optional when only one is configured + path = "/costcenter" # the decision to evaluate + } +} + +mutator "opa_bundle_json_patch" "add_meta" { + bundle_rule { + source = "platform" + path = "/add_meta" + } +} +``` + +Several `opa_bundle` blocks may be declared, each with its own services and signing keys — for example a platform-wide bundle alongside per-team ones. Each `bundle_rule` then names the `source` it evaluates against. + +### Decision contract + +A bundle decision path must evaluate to an **object**. Validators read `errors` and `warnings`; JSON-Patch mutators additionally read `patch`: + +```rego +package costcenter + +errors contains msg if { + not input.job.Meta.costcenter + msg := "Every job must have a costcenter metadata label" +} + +warnings contains msg if { + input.job.Priority > 75 + msg := "High priority jobs are reviewed manually" +} +``` + +Anything else fails the admission rather than being treated as "no findings" — a path pointing at a scalar (`data.costcenter.allow`), a missing decision, or an `errors` value that is not a list of strings all produce an admission error. This is deliberate: policy that is fetched over the network must fail closed when it does not say what NACP expects. + +Rego builtin errors are also fatal (`strict-builtin-errors`), so a failing `http.send` or `json.unmarshal` surfaces instead of silently making the rule undefined. + +### Signing + +Bundle policy runs with full authority over every job passing through the proxy, so whoever can answer the bundle URL can rewrite admission control. Configure [bundle signing](https://www.openpolicyagent.org/docs/latest/management-bundles/#signing) in the OPA configuration and set `require_signing = true` to make NACP refuse to start without it: + +```yaml +keys: + global_key: + algorithm: RS256 + key: ${BUNDLE_PUBLIC_KEY} + +bundles: + platform: + service: bundle_server + resource: /bundle.tar.gz + signing: + keyid: global_key +``` + +### Operating bundles + +- `GET /-/health` reports each bundle's active revision and last successful activation, returning 503 until every configured bundle has activated. Nomad's API lives under `/v1/`, so this endpoint does not shadow a proxied route. +- Failed refreshes are logged at warn level. NACP keeps enforcing the last activated bundle, so this log line is the signal that policy has gone stale. +- Every decision is logged and traced with its OPA decision ID and the active revision of each bundle, so an admission outcome can be traced back to an exact policy version. +- The `nacp.opa.decision.duration` metric records evaluation time by bundle source, decision path and outcome. +- `SIGHUP` re-reads each `config_path` and reconfigures the running instances. A configuration that fails to load leaves the previous one active. + ## More Examples Checkout the [examples](./example) folder for more examples. diff --git a/cmd/nacp/health.go b/cmd/nacp/health.go new file mode 100644 index 0000000..33e9ab2 --- /dev/null +++ b/cmd/nacp/health.go @@ -0,0 +1,58 @@ +package main + +import ( + "encoding/json" + "log/slog" + "net/http" + + "github.com/mxab/nacp/pkg/admissionctrl/opa/bundle" +) + +type healthResponse struct { + Status string `json:"status"` + Bundles []healthBundleEntry `json:"bundles,omitempty"` +} + +type healthBundleEntry struct { + Source string `json:"source"` + Bundle bundle.BundleStatus `json:"bundle"` +} + +// newHealthHandler reports whether every configured bundle has activated a +// policy. NACP keeps serving the last activated bundle when refreshes start +// failing, so this is what makes that state visible to an operator or an +// orchestrator health check instead of silently enforcing stale policy. +func newHealthHandler(bundles *bundle.Registry, logger *slog.Logger) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + response := healthResponse{Status: "ok"} + + for _, instance := range bundles.Instances() { + statuses := instance.Status() + if len(statuses) == 0 { + // The instance is up but OPA has not reported on any bundle yet. + response.Status = "unavailable" + continue + } + for _, status := range statuses { + response.Bundles = append(response.Bundles, healthBundleEntry{ + Source: instance.ID(), + Bundle: status, + }) + if !status.Activated() { + response.Status = "unavailable" + } + } + } + + code := http.StatusOK + if response.Status != "ok" { + code = http.StatusServiceUnavailable + } + + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(code) + if err := json.NewEncoder(w).Encode(response); err != nil { + logger.WarnContext(r.Context(), "Writing health response failed", "error", err) + } + }) +} diff --git a/cmd/nacp/health_test.go b/cmd/nacp/health_test.go new file mode 100644 index 0000000..5eff238 --- /dev/null +++ b/cmd/nacp/health_test.go @@ -0,0 +1,52 @@ +package main + +import ( + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/mxab/nacp/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestHealthReportsActivatedBundles(t *testing.T) { + bundles := testutil.SetupOpaRegistry(t, map[string]string{"platform": "package p"}) + handler := newHealthHandler(bundles, slog.New(slog.DiscardHandler)) + + // The bundle status listener fires asynchronously from activation, so the + // endpoint reports unavailable until OPA has told us about a revision. + var body healthResponse + require.Eventually(t, func() bool { + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/-/health", nil)) + if rec.Code != http.StatusOK { + return false + } + require.NoError(t, json.NewDecoder(rec.Body).Decode(&body)) + return true + }, 5*time.Second, 20*time.Millisecond, "health should become ok once the bundle activates") + + assert.Equal(t, "ok", body.Status) + require.Len(t, body.Bundles, 1) + assert.Equal(t, "platform", body.Bundles[0].Source) + assert.True(t, body.Bundles[0].Bundle.Activated()) +} + +func TestHealthWithoutBundlesIsOk(t *testing.T) { + bundles := testutil.SetupOpaRegistry(t, nil) + handler := newHealthHandler(bundles, slog.New(slog.DiscardHandler)) + + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/-/health", nil)) + + assert.Equal(t, http.StatusOK, rec.Code) + + var body healthResponse + require.NoError(t, json.NewDecoder(rec.Body).Decode(&body)) + assert.Equal(t, "ok", body.Status) + assert.Empty(t, body.Bundles) +} diff --git a/cmd/nacp/nacp.go b/cmd/nacp/nacp.go index d4926ef..87bda42 100644 --- a/cmd/nacp/nacp.go +++ b/cmd/nacp/nacp.go @@ -20,13 +20,13 @@ import ( "regexp" "strconv" "strings" + "syscall" "time" "github.com/mxab/nacp/pkg/admissionctrl/remoteutil" "github.com/mxab/nacp/pkg/admissionctrl/types" "github.com/mxab/nacp/pkg/logutil" nacpOtel "github.com/mxab/nacp/pkg/otel" - "github.com/open-policy-agent/opa/v1/sdk" "log/slog" @@ -36,6 +36,7 @@ import ( "github.com/mxab/nacp/pkg/admissionctrl" "github.com/mxab/nacp/pkg/admissionctrl/mutator" "github.com/mxab/nacp/pkg/admissionctrl/notation" + "github.com/mxab/nacp/pkg/admissionctrl/opa/bundle" "github.com/mxab/nacp/pkg/admissionctrl/validator" "github.com/mxab/nacp/pkg/config" "github.com/notaryproject/notation-go/dir" @@ -587,15 +588,15 @@ func run(c *config.Config) (err error) { } - opaSDK, stopOPA, err := setupOpaSDK(ctx, appLogger, c.OpaSdk) + bundles, stopBundles, err := bundle.Setup(ctx, rootFactory, c.OpaBundles) if err != nil { - return err - } - if stopOPA != nil { - defer stopOPA() + return fmt.Errorf("failed to set up OPA bundles: %w", err) } + defer stopBundles() - server, err := buildServer(c, rootFactory, opaSDK) + go watchReloadSignals(ctx, appLogger, bundles, c.OpaBundles) + + server, err := buildServer(c, rootFactory, bundles) if err != nil { return fmt.Errorf("failed to build server: %w", err) @@ -631,7 +632,7 @@ func run(c *config.Config) (err error) { } -func buildServer(c *config.Config, loggerFactory *logutil.LoggerFactory, sdk *sdk.OPA) (*http.Server, error) { +func buildServer(c *config.Config, loggerFactory *logutil.LoggerFactory, bundles *bundle.Registry) (*http.Server, error) { backend, err := url.Parse(c.Nomad.Address) if err != nil { return nil, fmt.Errorf("failed to parse nomad address: %w", err) @@ -654,12 +655,12 @@ func buildServer(c *config.Config, loggerFactory *logutil.LoggerFactory, sdk *sd proxyTransport.TLSClientConfig = nomadTlsConfig } - jobMutators, resolveTokenMutators, err := createMutators(c, loggerFactory, sdk) + jobMutators, resolveTokenMutators, err := createMutators(c, loggerFactory, bundles) if err != nil { return nil, fmt.Errorf("failed to create mutators: %w", err) } - jobValidators, resolveTokenValidators, err := createValidators(c, loggerFactory, sdk) + jobValidators, resolveTokenValidators, err := createValidators(c, loggerFactory, bundles) if err != nil { return nil, fmt.Errorf("failed to create validators: %w", err) } @@ -679,6 +680,12 @@ func buildServer(c *config.Config, loggerFactory *logutil.LoggerFactory, sdk *sd handlerFunc := NewProxyAsHandlerFunc(backend, jobHandler, loggerFactory.GetLogger("proxy-handler"), instrumentedProxyTransport) + // The Nomad API lives entirely under /v1/, so NACP's own endpoints can sit + // beside the proxied routes without shadowing any of them. + mux := http.NewServeMux() + mux.Handle("/", handlerFunc) + mux.Handle("GET /-/health", newHealthHandler(bundles, loggerFactory.GetLogger("health"))) + bind := fmt.Sprintf("%s:%d", c.Bind, c.Port) var tlsConfig *tls.Config @@ -693,7 +700,7 @@ func buildServer(c *config.Config, loggerFactory *logutil.LoggerFactory, sdk *sd server := &http.Server{ Addr: bind, TLSConfig: tlsConfig, - Handler: handlerFunc, + Handler: mux, ReadTimeout: nomadTimeout, WriteTimeout: nomadTimeout, } @@ -739,12 +746,12 @@ func createTlsConfig(caFile string, noClientCert bool) (*tls.Config, error) { return tlsConfig, nil } -func createMutators(c *config.Config, loggerFactory *logutil.LoggerFactory, opaSDK *sdk.OPA) ([]admissionctrl.JobMutator, bool, error) { +func createMutators(c *config.Config, loggerFactory *logutil.LoggerFactory, bundles *bundle.Registry) ([]admissionctrl.JobMutator, bool, error) { jobMutators := make([]admissionctrl.JobMutator, 0, len(c.Mutators)) var resolveToken bool for _, mutatorConfig := range c.Mutators { resolveToken = resolveToken || mutatorConfig.ResolveToken - jobMutator, err := createMutator(mutatorConfig, loggerFactory, opaSDK) + jobMutator, err := createMutator(mutatorConfig, loggerFactory, bundles) if err != nil { return nil, resolveToken, err } @@ -753,7 +760,9 @@ func createMutators(c *config.Config, loggerFactory *logutil.LoggerFactory, opaS return jobMutators, resolveToken, nil } -func createMutator(mutatorConfig config.Mutator, loggerFactory *logutil.LoggerFactory, opaSDK *sdk.OPA) (admissionctrl.JobMutator, error) { +// createMutator assumes config.Validate has already run, so the block each type +// requires is present. +func createMutator(mutatorConfig config.Mutator, loggerFactory *logutil.LoggerFactory, bundles *bundle.Registry) (admissionctrl.JobMutator, error) { switch mutatorConfig.Type { case "opa_json_patch": notationVerifier, err := buildVerifierIfEnabled(mutatorConfig.OpaRule.Notation, loggerFactory.GetLogger("notation_verifier")) @@ -764,21 +773,22 @@ func createMutator(mutatorConfig config.Mutator, loggerFactory *logutil.LoggerFa case "json_patch_webhook": return mutator.NewJsonPatchWebhookMutator(mutatorConfig.Name, mutatorConfig.Webhook.Endpoint, mutatorConfig.Webhook.Method, loggerFactory.GetLogger("json_patch_webhook_mutator")) case "opa_bundle_json_patch": - if mutatorConfig.OpaSdkRule == nil { - return nil, fmt.Errorf("mutator %q requires an opa_sdk_rule block", mutatorConfig.Name) + instance, err := resolveBundle(bundles, "mutator", mutatorConfig.Name, mutatorConfig.BundleRule) + if err != nil { + return nil, err } - return mutator.NewOpaBundleMutator(mutatorConfig.Name, mutatorConfig.OpaSdkRule.Path, loggerFactory.GetLogger("opa_bundle_mutator"), opaSDK) + return mutator.NewOpaBundleMutator(mutatorConfig.Name, mutatorConfig.BundleRule.Path, instance) default: return nil, fmt.Errorf("unknown mutator type %s", mutatorConfig.Type) } } -func createValidators(c *config.Config, loggerFactory *logutil.LoggerFactory, opaSDK *sdk.OPA) ([]admissionctrl.JobValidator, bool, error) { +func createValidators(c *config.Config, loggerFactory *logutil.LoggerFactory, bundles *bundle.Registry) ([]admissionctrl.JobValidator, bool, error) { jobValidators := make([]admissionctrl.JobValidator, 0, len(c.Validators)) var resolveToken bool for _, validatorConfig := range c.Validators { resolveToken = resolveToken || validatorConfig.ResolveToken - jobValidator, err := createValidator(validatorConfig, loggerFactory, opaSDK) + jobValidator, err := createValidator(validatorConfig, loggerFactory, bundles) if err != nil { return nil, resolveToken, err } @@ -787,7 +797,9 @@ func createValidators(c *config.Config, loggerFactory *logutil.LoggerFactory, op return jobValidators, resolveToken, nil } -func createValidator(validatorConfig config.Validator, loggerFactory *logutil.LoggerFactory, opaSDK *sdk.OPA) (admissionctrl.JobValidator, error) { +// createValidator assumes config.Validate has already run, so the block each +// type requires is present. +func createValidator(validatorConfig config.Validator, loggerFactory *logutil.LoggerFactory, bundles *bundle.Registry) (admissionctrl.JobValidator, error) { switch validatorConfig.Type { case "opa": notationVerifier, err := buildVerifierIfEnabled(validatorConfig.Notation, loggerFactory.GetLogger("notation_verifier")) @@ -796,10 +808,11 @@ func createValidator(validatorConfig config.Validator, loggerFactory *logutil.Lo } return validator.NewOpaValidator(validatorConfig.Name, validatorConfig.OpaRule.Filename, validatorConfig.OpaRule.Query, loggerFactory.GetLogger("opa_validator"), notationVerifier) case "opa_bundle": - if validatorConfig.OpaSdkRule == nil { - return nil, fmt.Errorf("validator %q requires an opa_sdk_rule block", validatorConfig.Name) + instance, err := resolveBundle(bundles, "validator", validatorConfig.Name, validatorConfig.BundleRule) + if err != nil { + return nil, err } - return validator.NewOpaBundleValidator(validatorConfig.Name, validatorConfig.OpaSdkRule.Path, loggerFactory.GetLogger("opa_bundle_validator"), opaSDK) + return validator.NewOpaBundleValidator(validatorConfig.Name, validatorConfig.BundleRule.Path, instance) case "webhook": return validator.NewWebhookValidator(validatorConfig.Name, validatorConfig.Webhook.Endpoint, validatorConfig.Webhook.Method, loggerFactory.GetLogger("webhook_validator")) case "notation": @@ -858,64 +871,40 @@ func buildTlsConfig(config config.NomadServerTLS) (*tls.Config, error) { } return tlsConfig, err } -func setupOpaSDK(ctx context.Context, logger *slog.Logger, opaConfig *config.OpaSdk) (*sdk.OPA, func(), error) { - if opaConfig == nil { - return nil, nil, nil +func resolveBundle(bundles *bundle.Registry, role, name string, rule *config.BundleRule) (*bundle.Instance, error) { + if rule == nil { + return nil, fmt.Errorf("%s %q requires a bundle_rule block", role, name) } - - opaSDK, err := buildOpaSdk(ctx, logger, opaConfig) + instance, err := bundles.Get(rule.Source) if err != nil { - return nil, nil, fmt.Errorf("failed to build OPA SDK: %w", err) + return nil, fmt.Errorf("%s %q: %w", role, name, err) } - - return opaSDK, func() { stopOpaSDK(opaSDK) }, nil + return instance, nil } -func buildOpaSdk(ctx context.Context, logger *slog.Logger, opaConfig *config.OpaSdk) (*sdk.OPA, error) { - return buildOpaSdkWithTimeout(ctx, logger, opaConfig, 30*time.Second) -} - -func buildOpaSdkWithTimeout(ctx context.Context, logger *slog.Logger, opaConfig *config.OpaSdk, readyTimeout time.Duration) (*sdk.OPA, error) { - logger.Info("Starting OPA SDK", "config_path", opaConfig.ConfigPath, "id", opaConfig.Id) - f, err := os.Open(opaConfig.ConfigPath) - if err != nil { - return nil, err - } - defer closeOpaConfig(f, logger) - - ready := make(chan struct{}) - readyCtx, cancel := context.WithTimeout(ctx, readyTimeout) - defer cancel() - - opaSDK, err := sdk.New(ctx, sdk.Options{ - ID: opaConfig.Id, - Config: f, - Ready: ready, - }) - if err != nil { - return nil, fmt.Errorf("failed to create OPA SDK: %w", err) - } - - logger.Info("Waiting for OPA SDK to become ready", "id", opaConfig.Id) - select { - case <-ready: - logger.Info("OPA SDK is ready", "id", opaConfig.Id) - return opaSDK, nil - case <-readyCtx.Done(): - stopOpaSDK(opaSDK) - logger.Error("OPA SDK did not become ready in time", "id", opaConfig.Id) - return nil, fmt.Errorf("OPA SDK did not become ready in time: %w", readyCtx.Err()) +// watchReloadSignals reconfigures every bundle instance on SIGHUP so that a +// changed OPA configuration (new service address, rotated signing key) can be +// picked up without dropping in-flight admission requests. +func watchReloadSignals(ctx context.Context, logger *slog.Logger, bundles *bundle.Registry, bundleConfigs []config.OpaBundle) { + if len(bundleConfigs) == 0 { + return } -} - -func stopOpaSDK(opaSDK *sdk.OPA) { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - opaSDK.Stop(shutdownCtx) -} - -func closeOpaConfig(file *os.File, logger *slog.Logger) { - if err := file.Close(); err != nil { - logger.Warn("Closing OPA SDK configuration failed", "error", err) + signals := make(chan os.Signal, 1) + signal.Notify(signals, syscall.SIGHUP) + defer signal.Stop(signals) + + for { + select { + case <-ctx.Done(): + return + case <-signals: + logger.Info("Received SIGHUP, reloading OPA bundle configuration") + if err := bundles.Reload(ctx, bundleConfigs); err != nil { + // The previous configuration stays active on failure. + logger.Error("Reloading OPA bundle configuration failed", "error", err) + continue + } + logger.Info("Reloaded OPA bundle configuration") + } } } diff --git a/cmd/nacp/nacp_test.go b/cmd/nacp/nacp_test.go index 1d08bc3..509a026 100644 --- a/cmd/nacp/nacp_test.go +++ b/cmd/nacp/nacp_test.go @@ -25,13 +25,12 @@ import ( "github.com/hashicorp/nomad/lib/file" "github.com/mxab/nacp/pkg/admissionctrl" "github.com/mxab/nacp/pkg/admissionctrl/mutator" + "github.com/mxab/nacp/pkg/admissionctrl/opa/bundle" "github.com/mxab/nacp/pkg/admissionctrl/types" "github.com/mxab/nacp/pkg/admissionctrl/validator" "github.com/mxab/nacp/pkg/config" "github.com/mxab/nacp/pkg/logutil" "github.com/mxab/nacp/testutil" - "github.com/open-policy-agent/opa/v1/sdk" - sdktest "github.com/open-policy-agent/opa/v1/sdk/test" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" @@ -790,7 +789,7 @@ func TestCreateValidators(t *testing.T) { validators: config.Validator{ Type: "opa_bundle", Name: "test", - OpaSdkRule: &config.OpaSdkRule{ + BundleRule: &config.BundleRule{ Path: "/mypolicy", }, }, @@ -798,11 +797,11 @@ func TestCreateValidators(t *testing.T) { needsOPA: true, }, { - name: "opa bundle validator without SDK", + name: "opa bundle validator without a configured bundle", validators: config.Validator{ Type: "opa_bundle", Name: "test", - OpaSdkRule: &config.OpaSdkRule{Path: "/my/policy"}, + BundleRule: &config.BundleRule{Path: "/my/policy"}, }, wantErr: true, }, @@ -826,11 +825,11 @@ func TestCreateValidators(t *testing.T) { Validators: []config.Validator{tc.validators}, } - var opaSDK *sdk.OPA + var bundles *bundle.Registry if tc.needsOPA { - opaSDK = testutil.SetupOpa(t, "package mypolicy") + bundles = testutil.SetupOpaRegistry(t, map[string]string{"test": "package mypolicy"}) } - validators, _, err := createValidators(c, discardFactory, opaSDK) + validators, _, err := createValidators(c, discardFactory, bundles) if tc.wantErr { assert.Error(t, err) @@ -847,20 +846,20 @@ func TestCreateValidators(t *testing.T) { func TestOpaBundleValidatorConfig(t *testing.T) { discardFactory, _ := logutil.NewLoggerFactory(nil, nil, false) - opaSDK := testutil.SetupOpa(t, "package configuredpath") + bundles := testutil.SetupOpaRegistry(t, map[string]string{"test": "package configuredpath"}) c := &config.Config{ Validators: []config.Validator{ { Type: "opa_bundle", Name: "test", - OpaSdkRule: &config.OpaSdkRule{ + BundleRule: &config.BundleRule{ Path: "/configuredpath", }, }, }, } - validators, _, err := createValidators(c, discardFactory, opaSDK) + validators, _, err := createValidators(c, discardFactory, bundles) require.NoError(t, err) require.Len(t, validators, 1) @@ -961,7 +960,7 @@ func TestCreateMutatators(t *testing.T) { mutators: config.Mutator{ Type: "opa_bundle_json_patch", Name: "test", - OpaSdkRule: &config.OpaSdkRule{ + BundleRule: &config.BundleRule{ Path: "/my/policy", }, }, @@ -973,7 +972,7 @@ func TestCreateMutatators(t *testing.T) { mutators: config.Mutator{ Type: "opa_bundle_json_patch", Name: "test", - OpaSdkRule: &config.OpaSdkRule{Path: "/my/policy"}, + BundleRule: &config.BundleRule{Path: "/my/policy"}, }, wantErr: true, }, @@ -1010,11 +1009,11 @@ func TestCreateMutatators(t *testing.T) { Mutators: []config.Mutator{tc.mutators}, } - var opaSDK *sdk.OPA + var bundles *bundle.Registry if tc.needsOPA { - opaSDK = testutil.SetupOpa(t, "package mypolicy") + bundles = testutil.SetupOpaRegistry(t, map[string]string{"test": "package mypolicy"}) } - mutators, _, err := createMutators(c, discardFactory, opaSDK) + mutators, _, err := createMutators(c, discardFactory, bundles) if tc.wantErr { assert.Error(t, err) @@ -1255,149 +1254,3 @@ func TestBuildConfig(t *testing.T) { }) } } - -func TestBuildOpaSdk(t *testing.T) { - - tt := []struct { - name string - - wantErr string - - configFn func(t *testing.T) string - }{ - { - name: "valid config", - - configFn: func(t *testing.T) string { - - server, err := sdktest.NewServer(sdktest.MockBundle("/bundles/bundle.tar.gz", map[string]string{ - "example.rego": `package example - - default allow = false - - `, - })) - require.NoError(t, err, "No error creating mock server") - t.Cleanup(server.Stop) - - // provide the OPA configuration which specifies - // fetching policy bundles from the mock server - // and logging decisions locally to the console - return fmt.Sprintf(`{ - "services": { - "test": { - "url": %q - } - }, - "bundles": { - "test": { - "resource": "/bundles/bundle.tar.gz" - } - }, - "decision_logs": { - "console": true - } - }`, server.URL()) - }, - }, - { - name: "invalid config", - configFn: func(t *testing.T) string { - return ` ... invalid json ... ` - }, - wantErr: "failed to create OPA SDK", - }, - } - for _, tc := range tt { - t.Run(tc.name, func(t *testing.T) { - - dir := t.TempDir() - - configPath := fmt.Sprintf("%s/opa-config.json", dir) - - err := os.WriteFile(configPath, []byte(tc.configFn(t)), 0644) - require.NoError(t, err) - - opaConfig := &config.OpaSdk{ - Id: "test", - ConfigPath: configPath, - } - - opa, err := buildOpaSdk(t.Context(), slog.Default(), opaConfig) - - if tc.wantErr != "" { - assert.Error(t, err) - assert.Contains(t, err.Error(), tc.wantErr) - return - } - assert.NoError(t, err) - assert.NotNil(t, opa) - t.Cleanup(func() { - opa.Stop(t.Context()) - }) - }) - } -} -func TestBuildOpaSdkMissingFile(t *testing.T) { - dir := t.TempDir() - configPath := fmt.Sprintf("%s/opa-config.json", dir) - - opaConfig := &config.OpaSdk{ - Id: "test", - ConfigPath: configPath, - } - - opa, err := buildOpaSdk(t.Context(), slog.Default(), opaConfig) - - assert.Error(t, err) - assert.Nil(t, opa) -} - -func TestSetupOpaSDK(t *testing.T) { - t.Run("disabled", func(t *testing.T) { - opaSDK, cleanup, err := setupOpaSDK(t.Context(), slog.Default(), nil) - require.NoError(t, err) - assert.Nil(t, opaSDK) - assert.Nil(t, cleanup) - }) - - t.Run("valid config", func(t *testing.T) { - configPath := fmt.Sprintf("%s/opa-config.json", t.TempDir()) - require.NoError(t, os.WriteFile(configPath, []byte(`{}`), 0644)) - - opaSDK, cleanup, err := setupOpaSDK(t.Context(), slog.Default(), &config.OpaSdk{Id: "test", ConfigPath: configPath}) - require.NoError(t, err) - require.NotNil(t, opaSDK) - require.NotNil(t, cleanup) - cleanup() - }) - - t.Run("invalid config", func(t *testing.T) { - configPath := fmt.Sprintf("%s/opa-config.json", t.TempDir()) - require.NoError(t, os.WriteFile(configPath, []byte(`not valid JSON`), 0644)) - - opaSDK, cleanup, err := setupOpaSDK(t.Context(), slog.Default(), &config.OpaSdk{Id: "test", ConfigPath: configPath}) - assert.ErrorContains(t, err, "failed to build OPA SDK") - assert.Nil(t, opaSDK) - assert.Nil(t, cleanup) - }) -} - -func TestBuildOpaSdkReadinessTimeout(t *testing.T) { - configPath := fmt.Sprintf("%s/opa-config.json", t.TempDir()) - configData := `{ - "services": {"test": {"url": "http://127.0.0.1:1"}}, - "bundles": {"test": {"service": "test", "resource": "/bundle.tar.gz"}} - }` - require.NoError(t, os.WriteFile(configPath, []byte(configData), 0644)) - - opaSDK, err := buildOpaSdkWithTimeout( - t.Context(), - slog.New(slog.DiscardHandler), - &config.OpaSdk{Id: "test", ConfigPath: configPath}, - time.Millisecond, - ) - - assert.ErrorContains(t, err, "OPA SDK did not become ready in time") - assert.Nil(t, opaSDK) -} diff --git a/example/demo/nacp.conf b/example/demo/nacp.conf index 7aeee87..645312c 100644 --- a/example/demo/nacp.conf +++ b/example/demo/nacp.conf @@ -77,19 +77,19 @@ mutator "json_patch_webhook" "a_remote_mutator" { {{- end }} # OPA Bundle -opa_sdk "example" { +opa_bundle "example" { config_path = "/local/opa.yml" } validator "opa_bundle" "hello_world" { - opa_sdk_rule { + bundle_rule { path = "/helloworld" } } mutator "opa_bundle_json_patch" "foobar_meta" { - opa_sdk_rule { + bundle_rule { path = "/foobar" } } diff --git a/generate.go b/generate.go index fc9f828..3fef930 100644 --- a/generate.go +++ b/generate.go @@ -1,3 +1,3 @@ package nacp -//go:generate weaver registry generate --registry ./o11y go --future ./o11y +//go:generate weaver registry generate --registry ./pkg/o11y go --future ./pkg/o11y diff --git a/go.mod b/go.mod index 875a71b..a007fa0 100644 --- a/go.mod +++ b/go.mod @@ -37,7 +37,10 @@ require ( oras.land/oras-go/v2 v2.6.0 ) -require github.com/open-policy-agent/opa v1.16.0 +require ( + github.com/open-policy-agent/opa v1.16.0 + gopkg.in/yaml.v3 v3.0.1 +) require ( dario.cat/mergo v1.0.2 // indirect @@ -216,7 +219,6 @@ require ( google.golang.org/grpc v1.80.0 // indirect google.golang.org/protobuf v1.36.11 // indirect gopkg.in/ini.v1 v1.67.1 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect kernel.org/pub/linux/libs/security/libcap/psx v1.2.77 // indirect oss.indeed.com/go/libtime v1.6.0 // indirect sigs.k8s.io/yaml v1.6.0 // indirect diff --git a/openwiki/policy-integrations.md b/openwiki/policy-integrations.md index 550d9f0..0828042 100644 --- a/openwiki/policy-integrations.md +++ b/openwiki/policy-integrations.md @@ -11,18 +11,18 @@ The HCL schema in `pkg/config/config.go` selects controller implementations at s ## Configuration model -Top-level configuration fields are `bind`, `port`, `tls`, `nomad`, repeated `validator` and `mutator` blocks, `telemetry`, and optional `opa_sdk`. Defaults are `0.0.0.0:6464`, upstream Nomad `http://localhost:4646`, info-level text logs to stdout, and disabled OTel exports. +Top-level configuration fields are `bind`, `port`, `tls`, `nomad`, repeated `validator`, `mutator` and `opa_bundle` blocks, and `telemetry`. Defaults are `0.0.0.0:6464`, upstream Nomad `http://localhost:4646`, info-level text logs to stdout, and disabled OTel exports. A controller is a labeled HCL block with a type and name. Startup wires these types: | Role | Type | Supporting block | Implementation area | | --- | --- | --- | --- | | Validator | `opa` | `opa_rule` | `pkg/admissionctrl/validator/opa_validator.go` | -| Validator | `opa_sdk` | `opa_sdk_rule` | `pkg/admissionctrl/validator/opa_bundle_validator.go` | +| Validator | `opa_bundle` | `bundle_rule` | `pkg/admissionctrl/validator/opa_bundle_validator.go` | | Validator | `webhook` | `webhook` | `pkg/admissionctrl/validator/webhook_validator.go` | | Validator | `notation` | `notation` | `pkg/admissionctrl/validator/notation_validator.go` | | Mutator | `opa_json_patch` | `opa_rule` | `pkg/admissionctrl/mutator/opa_json_patch.go` | -| Mutator | `opa_bundle_json_patch` | `opa_sdk_rule` | `pkg/admissionctrl/mutator/opa_bundle_json_patch.go` | +| Mutator | `opa_bundle_json_patch` | `bundle_rule` | `pkg/admissionctrl/mutator/opa_bundle_json_patch.go` | | Mutator | `json_patch_webhook` | `webhook` | `pkg/admissionctrl/mutator/json_patch_webhook.go` | `pkg/config/config_test.go` and its `testdata/` fixtures protect decoding/default behavior. `example/demo/nacp.conf` is the repository’s only combined configuration example; it exercises all but Notation. @@ -65,13 +65,17 @@ The embedded JSON-Patch mutator expects JSON Patch operations and applies them t An `opa_rule` may include a `notation` block. When configured, embedded OPA registers `notation_verify_image(string) -> bool`, connecting Rego decisions to the trust material described in [Notation](#notation-image-verification). -## OPA SDK and bundles +## OPA bundles -`opa_sdk "" { config_path = ... }` creates an OPA SDK instance during NACP startup. The executable waits up to 30 seconds for the SDK to become ready before serving. `opa_sdk_rule { path = ... }` selects the decision path used by either an `opa_sdk` validator or `opa_bundle_json_patch` mutator. +`opa_bundle "" { config_path = ... }` creates one OPA SDK instance per block, each with its own OPA configuration and therefore its own bundle services, signing keys and refresh schedule. Blocks are repeatable, so a platform-wide bundle can coexist with per-team ones. Optional settings are `ready_timeout` (default 30s, how long startup waits for the first activation), `decision_timeout` (default 5s, `"0s"` to inherit only the request deadline) and `require_signing`. -Both adapters pass the same payload to `sdk.OPA.Decision`. Validators consume `errors` and `warnings`; bundle mutators also consume a JSON Patch `patch`. `example/demo/nacp.conf`, `example/demo/opa.yml`, and the bundle Rego directories are the grounded reference for current decision paths and results. +`bundle_rule { source = ..., path = ... }` selects the decision an `opa_bundle` validator or `opa_bundle_json_patch` mutator evaluates. `source` names the `opa_bundle` id and may be omitted when exactly one bundle is configured. `pkg/config/config.go` validates all of this at load time — a rule naming an unknown source, or omitting `source` when several bundles exist, is a configuration error rather than a startup failure. -Bundle work was added after the embedded adapters and the latest implementation commit is explicitly a “working poc.” Treat bundle availability, refresh, and rollout behavior as an operational question to validate in your deployment; it is not a documented guarantee of this codebase. The pipeline still enforces the same ordering and error rules in [architecture](architecture.md#request-lifecycle-and-outcomes). +Both adapters go through `bundle.Instance.Decide` (`pkg/admissionctrl/opa/bundle/decide.go`), which applies the decision timeout, sets `StrictBuiltinErrors`, and parses the result with `opa.ParseDecision(..., opa.Strict)`. Strict parsing is the safety property: a decision that is not an object with list-of-string `errors`/`warnings` (and, for mutators, a JSON Patch `patch`) fails the admission instead of reading as "no findings". The embedded adapters share the same parser in `opa.Lenient` mode, which preserves their released tolerance. + +Operationally, `GET /-/health` reports each bundle's active revision and last successful activation and returns 503 until every bundle has activated; failed refreshes log at warn level while the last activated policy stays in force; decisions carry `opa.decision.id` and per-bundle revisions into logs and spans; `nacp.opa.decision.duration` records evaluation time; and `SIGHUP` reloads each `config_path`, leaving the previous configuration active if the new one fails. + +`example/demo/nacp.conf`, `example/demo/opa.yml`, and the bundle Rego directories are the grounded reference for current decision paths and results. ## Webhooks diff --git a/pkg/admissionctrl/mutator/opa_bundle_json_patch.go b/pkg/admissionctrl/mutator/opa_bundle_json_patch.go index 95c5ff0..f57bbad 100644 --- a/pkg/admissionctrl/mutator/opa_bundle_json_patch.go +++ b/pkg/admissionctrl/mutator/opa_bundle_json_patch.go @@ -4,132 +4,57 @@ import ( "context" "errors" "fmt" - "log/slog" "github.com/hashicorp/go-multierror" "github.com/hashicorp/nomad/api" "github.com/mxab/nacp/pkg/admissionctrl" "github.com/mxab/nacp/pkg/admissionctrl/mutator/jsonpatcher" + "github.com/mxab/nacp/pkg/admissionctrl/opa/bundle" "github.com/mxab/nacp/pkg/admissionctrl/types" - "github.com/open-policy-agent/opa/v1/sdk" ) type OpaBundleMutator struct { name string path string - logger *slog.Logger - opa *sdk.OPA + bundle *bundle.Instance } var _ admissionctrl.JobMutator = (*OpaBundleMutator)(nil) -func (m *OpaBundleMutator) Mutate(ctx context.Context, payload *types.Payload) (*api.Job, bool, []error, error) { - decision, err := m.opa.Decision(ctx, sdk.DecisionOptions{ - Input: payload, - Path: m.path, - }) - if err != nil { - return nil, false, nil, fmt.Errorf("failed to perform policy decision: %w", err) - } - m.logger.DebugContext(ctx, "OPA decision", slog.Any("result", decision)) - - result, ok := decision.Result.(map[string]interface{}) - if !ok { - return nil, false, nil, fmt.Errorf("policy yielded an invalid decision value: %v", decision.Result) +func NewOpaBundleMutator(name string, path string, instance *bundle.Instance) (*OpaBundleMutator, error) { + if instance == nil { + return nil, errors.New("OPA bundle is required") } - - if err := parseDecisionErrors(result["errors"]); err != nil { - return nil, false, nil, err - } - - warnings, err := parseDecisionWarnings(result["warnings"]) - if err != nil { - return nil, false, warnings, err - } - - patch, found := result["patch"] - if !found || patch == nil { - return payload.Job, false, warnings, nil - } - - job, mutated, err := applyDecisionPatch(payload.Job, patch) - return job, mutated, warnings, err -} - -func parseDecisionErrors(raw interface{}) error { - if raw == nil { - return nil - } - - entries, ok := raw.([]interface{}) - if !ok { - return fmt.Errorf("policy yielded an invalid errors value: %v", raw) - } - - var result error - for _, entry := range entries { - if entry == nil { - continue - } - message, ok := entry.(string) - if !ok { - return fmt.Errorf("policy yielded an invalid error entry value: %v", entry) - } - result = multierror.Append(result, errors.New(message)) + if path == "" { + return nil, errors.New("OPA decision path is required") } - return result + return &OpaBundleMutator{ + name: name, + path: path, + bundle: instance, + }, nil } -func parseDecisionWarnings(raw interface{}) ([]error, error) { - warnings := []error{} - if raw == nil { - return warnings, nil - } - - entries, ok := raw.([]interface{}) - if !ok { - return warnings, fmt.Errorf("policy yielded an invalid warnings value: %v", raw) +func (m *OpaBundleMutator) Mutate(ctx context.Context, payload *types.Payload) (*api.Job, bool, []error, error) { + decision, err := m.bundle.Decide(ctx, m.path, payload) + if err != nil { + return nil, false, nil, err } - for _, entry := range entries { - if entry == nil { - continue - } - message, ok := entry.(string) - if !ok { - return warnings, fmt.Errorf("policy yielded an invalid warning entry value: %v", entry) - } - warnings = append(warnings, errors.New(message)) + if len(decision.Errors) > 0 { + return nil, false, nil, multierror.Append(nil, decision.Errors...) } - return warnings, nil -} -func applyDecisionPatch(job *api.Job, raw interface{}) (*api.Job, bool, error) { - operations, ok := raw.([]interface{}) - if !ok { - return nil, false, fmt.Errorf("policy yielded an invalid patch value: %v", raw) + // A policy that produced no patch is a no-op, not an empty patch. + if !decision.HasPatch { + return payload.Job, false, decision.Warnings, nil } - result, mutated, err := jsonpatcher.PatchJob(job, operations) + job, mutated, err := jsonpatcher.PatchJob(payload.Job, decision.Patch) if err != nil { - return nil, false, fmt.Errorf("policy yielded patch failed: %w", err) + return nil, false, nil, fmt.Errorf("policy yielded patch failed: %w", err) } - return result, mutated, nil -} - -func NewOpaBundleMutator(name string, path string, logger *slog.Logger, opaSDK *sdk.OPA) (*OpaBundleMutator, error) { - if opaSDK == nil { - return nil, errors.New("OPA SDK is required") - } - if path == "" { - return nil, errors.New("OPA decision path is required") - } - return &OpaBundleMutator{ - name: name, - path: path, - logger: logger, - opa: opaSDK, - }, nil + return job, mutated, decision.Warnings, nil } func (m *OpaBundleMutator) Name() string { diff --git a/pkg/admissionctrl/mutator/opa_bundle_json_patch_test.go b/pkg/admissionctrl/mutator/opa_bundle_json_patch_test.go index 6626ea1..6cb56f1 100644 --- a/pkg/admissionctrl/mutator/opa_bundle_json_patch_test.go +++ b/pkg/admissionctrl/mutator/opa_bundle_json_patch_test.go @@ -1,7 +1,6 @@ package mutator import ( - "log/slog" "testing" "github.com/hashicorp/nomad/api" @@ -13,7 +12,7 @@ import ( func TestOpaBundleMutatorName(t *testing.T) { opa := testutil.SetupOpa(t, "package mypolicy") - mutator, err := NewOpaBundleMutator("test", "test/path", slog.Default(), opa) + mutator, err := NewOpaBundleMutator("test", "test/path", opa) require.NoError(t, err, "No error creating mutator") assert.Equal(t, "test", mutator.Name(), "Name is correct") @@ -22,11 +21,11 @@ func TestOpaBundleMutatorName(t *testing.T) { func TestNewOpaBundleMutatorValidation(t *testing.T) { opa := testutil.SetupOpa(t, "package mypolicy") - mutator, err := NewOpaBundleMutator("test", "/mypolicy", slog.Default(), nil) - assert.ErrorContains(t, err, "OPA SDK is required") + mutator, err := NewOpaBundleMutator("test", "/mypolicy", nil) + assert.ErrorContains(t, err, "OPA bundle is required") assert.Nil(t, mutator) - mutator, err = NewOpaBundleMutator("test", "", slog.Default(), opa) + mutator, err = NewOpaBundleMutator("test", "", opa) assert.ErrorContains(t, err, "OPA decision path is required") assert.Nil(t, mutator) } @@ -64,7 +63,7 @@ func TestOpaBundleMutator(t *testing.T) { expectedJob: nil, expectedMutated: false, expectedWarns: []string{}, - expectedErrs: []string{"failed to perform policy decision"}, + expectedErrs: []string{"is undefined in the active bundle"}, }, { name: "reject non-object decision", @@ -190,7 +189,7 @@ func TestOpaBundleMutator(t *testing.T) { expectedErrs: []string{"policy yielded an invalid error entry value"}, }, { - name: "handle invalid warning entry type as warning", + name: "handle invalid warning entry type as error", policy: `package mypolicy warnings = ["this is fine", 5] `, @@ -198,7 +197,7 @@ func TestOpaBundleMutator(t *testing.T) { inputJob: &api.Job{}, expectedJob: nil, expectedMutated: false, - expectedWarns: []string{"this is fine"}, + expectedWarns: []string{}, expectedErrs: []string{"policy yielded an invalid warning entry value"}, }, { @@ -223,14 +222,14 @@ func TestOpaBundleMutator(t *testing.T) { expectedJob: nil, expectedMutated: false, expectedWarns: []string{}, - expectedErrs: []string{"policy yielded patch failed"}, + expectedErrs: []string{"policy yielded an invalid patch entry value"}, }, } for _, tc := range tt { t.Run(tc.name, func(t *testing.T) { opa := testutil.SetupOpa(t, tc.policy) - mutator, err := NewOpaBundleMutator(tc.name, tc.path, slog.New(slog.DiscardHandler), opa) + mutator, err := NewOpaBundleMutator(tc.name, tc.path, opa) require.NoError(t, err, "No error creating mutator") result, mutated, warns, err := mutator.Mutate(t.Context(), &types.Payload{Job: tc.inputJob}) diff --git a/pkg/admissionctrl/mutator/opa_json_patch.go b/pkg/admissionctrl/mutator/opa_json_patch.go index 9ded6e0..30322d4 100644 --- a/pkg/admissionctrl/mutator/opa_json_patch.go +++ b/pkg/admissionctrl/mutator/opa_json_patch.go @@ -27,27 +27,24 @@ func (j *OpaJsonPatchMutator) Mutate(ctx context.Context, payload *types.Payload return nil, false, nil, err } - errors := results.GetErrors() + decision := results.Decision() - if len(errors) > 0 { - j.logger.Debug("Got errors from rule", "rule", j.Name(), "errors", errors, "job", payload.Job.ID) + if len(decision.Errors) > 0 { + j.logger.Debug("Got errors from rule", "rule", j.Name(), "errors", decision.Errors, "job", payload.Job.ID) allErrors := multierror.Append(nil) - for _, warn := range errors { + for _, warn := range decision.Errors { allErrors = multierror.Append(allErrors, fmt.Errorf("%s (%s)", warn, j.Name())) } return nil, false, nil, allErrors } - warnings := results.GetWarnings() - - if len(warnings) > 0 { - j.logger.Debug("Got warnings from rule", "rule", j.Name(), "warnings", warnings, "job", payload.Job.ID) - for _, warn := range warnings { + if len(decision.Warnings) > 0 { + j.logger.Debug("Got warnings from rule", "rule", j.Name(), "warnings", decision.Warnings, "job", payload.Job.ID) + for _, warn := range decision.Warnings { allWarnings = append(allWarnings, fmt.Errorf("%s (%s)", warn, j.Name())) } } - patchData := results.GetPatch() - patchedJob, mutated, err := jsonpatcher.PatchJob(payload.Job, patchData) + patchedJob, mutated, err := jsonpatcher.PatchJob(payload.Job, decision.Patch) if err != nil { return nil, false, nil, err } diff --git a/pkg/admissionctrl/opa/bundle/decide.go b/pkg/admissionctrl/opa/bundle/decide.go new file mode 100644 index 0000000..9f939de --- /dev/null +++ b/pkg/admissionctrl/opa/bundle/decide.go @@ -0,0 +1,108 @@ +package bundle + +import ( + "context" + "fmt" + "log/slog" + "time" + + "github.com/mxab/nacp/pkg/admissionctrl/opa" + "github.com/mxab/nacp/pkg/admissionctrl/types" + "github.com/mxab/nacp/pkg/o11y" + "github.com/open-policy-agent/opa/v1/sdk" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +// Outcome labels how a decision ended, for metrics. +const ( + outcomeAllow = "allow" + outcomeDeny = "deny" + outcomeError = "error" +) + +var decisionDuration = func() o11y.NacpOpaDecisionDuration { + instrument, err := o11y.NewNacpOpaDecisionDuration(otel.Meter("nacp.opa.bundle")) + if err != nil { + panic(err) + } + return instrument +}() + +// Decide evaluates path against this instance's active bundles and returns the +// parsed decision document. +// +// Everything the bundle adapters need in common lives here: the evaluation +// deadline, strict builtin errors, and the provenance that makes an admission +// outcome traceable back to a specific bundle revision. +func (i *Instance) Decide(ctx context.Context, path string, payload *types.Payload) (*opa.Decision, error) { + // The deadline bounds evaluation only where rego reaches a cancellation + // check (http.send, large iterations); a tight loop in a trivial rule can + // still outrun it. It is a backstop, not a hard limit. + if i.decisionTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, i.decisionTimeout) + defer cancel() + } + + span := trace.SpanFromContext(ctx) + started := time.Now() + + result, err := i.opa.Decision(ctx, sdk.DecisionOptions{ + Input: payload, + Path: path, + // A builtin that errors must fail the decision. Left off, the rule + // silently evaluates to undefined and the policy stops applying. + StrictBuiltinErrors: true, + }) + if err != nil { + i.record(ctx, path, started, outcomeError) + if sdk.IsUndefinedErr(err) { + return nil, fmt.Errorf("decision path %q is undefined in the active bundle of opa_bundle %q", path, i.id) + } + return nil, fmt.Errorf("failed to perform policy decision: %w", err) + } + + attrs := []attribute.KeyValue{ + attribute.String("opa.bundle.source", i.id), + attribute.String("opa.decision.path", path), + attribute.String("opa.decision.id", result.ID), + } + logAttrs := []any{ + slog.String("opa.bundle.source", i.id), + slog.String("opa.decision.path", path), + slog.String("opa.decision.id", result.ID), + } + for name, provenance := range result.Provenance.Bundles { + key := "opa.bundle." + name + ".revision" + attrs = append(attrs, attribute.String(key, provenance.Revision)) + logAttrs = append(logAttrs, slog.String(key, provenance.Revision)) + } + span.SetAttributes(attrs...) + + decision, err := opa.ParseDecision(result.Result, opa.Strict) + if err != nil { + i.record(ctx, path, started, outcomeError) + i.logger.DebugContext(ctx, "OPA bundle decision was not a valid decision document", logAttrs...) + return nil, err + } + + outcome := outcomeAllow + if len(decision.Errors) > 0 { + outcome = outcomeDeny + } + i.record(ctx, path, started, outcome) + i.logger.DebugContext(ctx, "OPA bundle decision", + append(logAttrs, + slog.String("opa.decision.outcome", outcome), + slog.Any("errors", decision.Errors), + slog.Any("warnings", decision.Warnings), + )...) + + return decision, nil +} + +func (i *Instance) record(ctx context.Context, path string, started time.Time, outcome string) { + decisionDuration.Record(ctx, time.Since(started).Seconds(), i.id, outcome, path) +} diff --git a/pkg/admissionctrl/opa/bundle/logging.go b/pkg/admissionctrl/opa/bundle/logging.go new file mode 100644 index 0000000..7b1d58d --- /dev/null +++ b/pkg/admissionctrl/opa/bundle/logging.go @@ -0,0 +1,71 @@ +package bundle + +import ( + "context" + "fmt" + "log/slog" + + "github.com/open-policy-agent/opa/v1/logging" +) + +// slogAdapter routes OPA's own logging through NACP's slog pipeline. +// +// This is not cosmetic. sdk.Options defaults Logger to a buffered logger which +// OPA discards once its plugins have started, so an instance built without an +// explicit logger silently swallows every bundle download failure, activation +// error and signature verification failure. +type slogAdapter struct { + logger *slog.Logger + fields []any +} + +var _ logging.Logger = (*slogAdapter)(nil) + +func newSlogAdapter(logger *slog.Logger) *slogAdapter { + return &slogAdapter{logger: logger} +} + +func (a *slogAdapter) Debug(format string, args ...any) { a.log(slog.LevelDebug, format, args...) } +func (a *slogAdapter) Info(format string, args ...any) { a.log(slog.LevelInfo, format, args...) } +func (a *slogAdapter) Warn(format string, args ...any) { a.log(slog.LevelWarn, format, args...) } +func (a *slogAdapter) Error(format string, args ...any) { a.log(slog.LevelError, format, args...) } + +func (a *slogAdapter) log(level slog.Level, format string, args ...any) { + if !a.logger.Enabled(context.Background(), level) { + return + } + message := format + if len(args) > 0 { + message = fmt.Sprintf(format, args...) + } + a.logger.Log(context.Background(), level, message, a.fields...) +} + +func (a *slogAdapter) WithFields(fields map[string]any) logging.Logger { + merged := make([]any, 0, len(a.fields)+2*len(fields)) + merged = append(merged, a.fields...) + for key, value := range fields { + merged = append(merged, slog.Any(key, value)) + } + return &slogAdapter{logger: a.logger, fields: merged} +} + +// GetLevel reports the level OPA should assume. OPA uses it to decide whether +// to emit expensive debug payloads and whether to enable rego print statements, +// so it must reflect the slog handler rather than a fixed value. +func (a *slogAdapter) GetLevel() logging.Level { + switch { + case a.logger.Enabled(context.Background(), slog.LevelDebug): + return logging.Debug + case a.logger.Enabled(context.Background(), slog.LevelInfo): + return logging.Info + case a.logger.Enabled(context.Background(), slog.LevelWarn): + return logging.Warn + default: + return logging.Error + } +} + +// SetLevel is a no-op: the level is owned by the slog handler, which NACP +// configures from telemetry.logging.level. +func (a *slogAdapter) SetLevel(logging.Level) {} diff --git a/pkg/admissionctrl/opa/bundle/registry.go b/pkg/admissionctrl/opa/bundle/registry.go new file mode 100644 index 0000000..ba30f80 --- /dev/null +++ b/pkg/admissionctrl/opa/bundle/registry.go @@ -0,0 +1,248 @@ +// Package bundle owns the OPA SDK instances that evaluate remote policy +// bundles. Each configured opa_bundle block becomes one Instance with its own +// OPA configuration, and therefore its own bundle services, signing keys and +// refresh schedule. +package bundle + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "os" + "sort" + "strings" + "sync" + "time" + + "github.com/mxab/nacp/pkg/config" + "github.com/mxab/nacp/pkg/logutil" + "github.com/open-policy-agent/opa/v1/sdk" +) + +// Instance is a single OPA SDK instance plus the NACP settings that govern how +// decisions are taken against it. +type Instance struct { + id string + opa *sdk.OPA + decisionTimeout time.Duration + logger *slog.Logger + + status *statusTracker +} + +// ID returns the opa_bundle label this instance was built from. +func (i *Instance) ID() string { return i.id } + +// Registry holds every configured bundle instance, keyed by id. +type Registry struct { + instances map[string]*Instance + ids []string +} + +// Setup builds every configured bundle instance and waits for each to activate +// its first bundle. It returns a stop function that shuts all of them down. A +// configuration without opa_bundle blocks yields an empty registry, which is a +// valid state: NACP simply has no bundle-backed controllers. +func Setup(ctx context.Context, loggerFactory *logutil.LoggerFactory, bundles []config.OpaBundle) (*Registry, func(), error) { + registry := &Registry{instances: make(map[string]*Instance, len(bundles))} + + stop := func() { + for _, instance := range registry.instances { + instance.stop() + } + } + + type result struct { + instance *Instance + err error + } + results := make([]result, len(bundles)) + + var wg sync.WaitGroup + for idx, bundleConfig := range bundles { + wg.Add(1) + go func() { + defer wg.Done() + instance, err := newInstance(ctx, loggerFactory, bundleConfig) + results[idx] = result{instance: instance, err: err} + }() + } + wg.Wait() + + var errs []error + for _, r := range results { + if r.err != nil { + errs = append(errs, r.err) + continue + } + registry.instances[r.instance.id] = r.instance + registry.ids = append(registry.ids, r.instance.id) + } + sort.Strings(registry.ids) + + if len(errs) > 0 { + // Instances that did come up still hold goroutines and HTTP clients. + stop() + return nil, nil, errors.Join(errs...) + } + + return registry, stop, nil +} + +// Get resolves the bundle a controller is bound to. An empty source resolves to +// the only configured bundle, which keeps single-bundle configurations from +// having to repeat the id on every rule. +func (r *Registry) Get(source string) (*Instance, error) { + if r == nil || len(r.instances) == 0 { + return nil, errors.New("no opa_bundle is configured") + } + if source == "" { + if len(r.ids) > 1 { + return nil, fmt.Errorf("bundle_rule.source is required, configured bundles: %s", strings.Join(r.ids, ", ")) + } + return r.instances[r.ids[0]], nil + } + instance, ok := r.instances[source] + if !ok { + return nil, fmt.Errorf("unknown bundle_rule.source %q, configured bundles: %s", source, strings.Join(r.ids, ", ")) + } + return instance, nil +} + +// Reload re-reads each bundle's OPA configuration file and reconfigures the +// running instance with it. An instance whose reload fails keeps its previous +// configuration, so a bad edit degrades to "no change" rather than to an +// admission controller with no policy. +func (r *Registry) Reload(ctx context.Context, bundleConfigs []config.OpaBundle) error { + var errs []error + for _, bundleConfig := range bundleConfigs { + instance, ok := r.instances[bundleConfig.Id] + if !ok { + errs = append(errs, fmt.Errorf("opa_bundle %q is not running", bundleConfig.Id)) + continue + } + if err := instance.reload(ctx, bundleConfig); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +func (i *Instance) reload(ctx context.Context, bundleConfig config.OpaBundle) error { + raw, err := os.ReadFile(bundleConfig.ConfigPath) + if err != nil { + return fmt.Errorf("opa_bundle %q: failed to read config_path: %w", bundleConfig.Id, err) + } + if bundleConfig.RequireSigning { + if err := verifySigningConfigured(raw); err != nil { + return fmt.Errorf("opa_bundle %q: %w", bundleConfig.Id, err) + } + } + + readyTimeout, err := bundleConfig.ResolvedReadyTimeout() + if err != nil { + return fmt.Errorf("opa_bundle %q: %w", bundleConfig.Id, err) + } + readyCtx, cancel := context.WithTimeout(ctx, readyTimeout) + defer cancel() + + if err := i.opa.Configure(readyCtx, sdk.ConfigOptions{Config: bytes.NewReader(raw)}); err != nil { + return fmt.Errorf("opa_bundle %q: failed to reconfigure: %w", bundleConfig.Id, err) + } + i.logger.Info("Reconfigured OPA bundle instance", "id", bundleConfig.Id, "config_path", bundleConfig.ConfigPath) + return nil +} + +// Instances returns every configured instance, ordered by id. +func (r *Registry) Instances() []*Instance { + if r == nil { + return nil + } + instances := make([]*Instance, 0, len(r.ids)) + for _, id := range r.ids { + instances = append(instances, r.instances[id]) + } + return instances +} + +func newInstance(ctx context.Context, loggerFactory *logutil.LoggerFactory, bundleConfig config.OpaBundle) (*Instance, error) { + logger := loggerFactory.GetLogger("opa_bundle/" + bundleConfig.Id) + + readyTimeout, err := bundleConfig.ResolvedReadyTimeout() + if err != nil { + return nil, fmt.Errorf("opa_bundle %q: %w", bundleConfig.Id, err) + } + decisionTimeout, err := bundleConfig.ResolvedDecisionTimeout() + if err != nil { + return nil, fmt.Errorf("opa_bundle %q: %w", bundleConfig.Id, err) + } + + // Read once: the same bytes are inspected for signing requirements and + // replayed on reload, so the file is never re-read behind OPA's back. + raw, err := os.ReadFile(bundleConfig.ConfigPath) + if err != nil { + return nil, fmt.Errorf("opa_bundle %q: failed to read config_path: %w", bundleConfig.Id, err) + } + if bundleConfig.RequireSigning { + if err := verifySigningConfigured(raw); err != nil { + return nil, fmt.Errorf("opa_bundle %q: %w", bundleConfig.Id, err) + } + } + + logger.Info("Starting OPA bundle instance", + "id", bundleConfig.Id, + "config_path", bundleConfig.ConfigPath, + "ready_timeout", readyTimeout, + "decision_timeout", decisionTimeout, + "require_signing", bundleConfig.RequireSigning, + ) + + instance := &Instance{ + id: bundleConfig.Id, + decisionTimeout: decisionTimeout, + logger: logger, + status: newStatusTracker(logger), + } + + ready := make(chan struct{}) + readyCtx, cancel := context.WithTimeout(ctx, readyTimeout) + defer cancel() + + // Without an explicit logger OPA buffers its own logs and then discards + // them, which hides bundle download and signature failures entirely. + opaLogger := newSlogAdapter(logger) + opaSDK, err := sdk.New(ctx, sdk.Options{ + ID: bundleConfig.Id, + Config: bytes.NewReader(raw), + Ready: ready, + Logger: opaLogger, + ConsoleLogger: opaLogger, + }) + if err != nil { + return nil, fmt.Errorf("opa_bundle %q: failed to create OPA SDK: %w", bundleConfig.Id, err) + } + instance.opa = opaSDK + instance.status.watch(opaSDK) + + logger.Info("Waiting for OPA bundle to become ready", "id", bundleConfig.Id) + select { + case <-ready: + logger.Info("OPA bundle is ready", "id", bundleConfig.Id) + return instance, nil + case <-readyCtx.Done(): + instance.stop() + logger.Error("OPA bundle did not become ready in time", "id", bundleConfig.Id) + return nil, fmt.Errorf("opa_bundle %q did not become ready in time: %w", bundleConfig.Id, readyCtx.Err()) + } +} + +func (i *Instance) stop() { + if i == nil || i.opa == nil { + return + } + shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + i.opa.Stop(shutdownCtx) +} diff --git a/pkg/admissionctrl/opa/bundle/registry_test.go b/pkg/admissionctrl/opa/bundle/registry_test.go new file mode 100644 index 0000000..2e4aa5d --- /dev/null +++ b/pkg/admissionctrl/opa/bundle/registry_test.go @@ -0,0 +1,381 @@ +package bundle + +import ( + "bytes" + "fmt" + "io" + "log/slog" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/hashicorp/nomad/api" + "github.com/mxab/nacp/pkg/admissionctrl/types" + "github.com/mxab/nacp/pkg/config" + "github.com/mxab/nacp/pkg/logutil" + sdktest "github.com/open-policy-agent/opa/v1/sdk/test" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func testPayload() *types.Payload { + id := "test-job" + return &types.Payload{Job: &api.Job{ID: &id}} +} + +func discardFactory(t *testing.T) *logutil.LoggerFactory { + t.Helper() + factory, _ := logutil.NewLoggerFactory(io.Discard, io.Discard, false) + return factory +} + +// writeConfig starts a mock bundle server serving policy and writes an OPA +// configuration pointing at it. +func writeConfig(t *testing.T, name, policy string) string { + t.Helper() + + server, err := sdktest.NewServer(sdktest.MockBundle("/bundles/bundle.tar.gz", map[string]string{ + "example.rego": policy, + })) + require.NoError(t, err) + t.Cleanup(server.Stop) + + return writeConfigFile(t, name, fmt.Sprintf(`{ + "services": {%q: {"url": %q}}, + "bundles": {%q: {"service": %q, "resource": "/bundles/bundle.tar.gz"}} + }`, name, server.URL(), name, name)) +} + +func writeConfigFile(t *testing.T, name, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), name+".json") + require.NoError(t, os.WriteFile(path, []byte(contents), 0o600)) + return path +} + +func TestSetupResolvesEachSourceToItsOwnBundle(t *testing.T) { + configs := []config.OpaBundle{ + {Id: "platform", ConfigPath: writeConfig(t, "platform", `package platformpolicy + errors = ["platform says no"]`)}, + {Id: "team", ConfigPath: writeConfig(t, "team", `package teampolicy + errors = ["team says no"]`)}, + } + + registry, stop, err := Setup(t.Context(), discardFactory(t), configs) + require.NoError(t, err) + t.Cleanup(stop) + + platform, err := registry.Get("platform") + require.NoError(t, err) + decision, err := platform.Decide(t.Context(), "/platformpolicy", testPayload()) + require.NoError(t, err) + require.Len(t, decision.Errors, 1) + assert.EqualError(t, decision.Errors[0], "platform says no") + + team, err := registry.Get("team") + require.NoError(t, err) + decision, err = team.Decide(t.Context(), "/teampolicy", testPayload()) + require.NoError(t, err) + require.Len(t, decision.Errors, 1) + assert.EqualError(t, decision.Errors[0], "team says no") + + // Each bundle only knows its own policy. + _, err = platform.Decide(t.Context(), "/teampolicy", testPayload()) + assert.ErrorContains(t, err, "is undefined in the active bundle") +} + +func TestGetSourceResolution(t *testing.T) { + single, stop, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "only", ConfigPath: writeConfig(t, "only", "package p")}, + }) + require.NoError(t, err) + t.Cleanup(stop) + + t.Run("empty source resolves to the only bundle", func(t *testing.T) { + instance, err := single.Get("") + require.NoError(t, err) + assert.Equal(t, "only", instance.ID()) + }) + + t.Run("unknown source lists the valid ids", func(t *testing.T) { + _, err := single.Get("nope") + assert.ErrorContains(t, err, `unknown bundle_rule.source "nope", configured bundles: only`) + }) + + multi, stop, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "a", ConfigPath: writeConfig(t, "a", "package p")}, + {Id: "b", ConfigPath: writeConfig(t, "b", "package p")}, + }) + require.NoError(t, err) + t.Cleanup(stop) + + t.Run("empty source is ambiguous with several bundles", func(t *testing.T) { + _, err := multi.Get("") + assert.ErrorContains(t, err, "bundle_rule.source is required, configured bundles: a, b") + }) + + t.Run("no bundles at all", func(t *testing.T) { + empty, stop, err := Setup(t.Context(), discardFactory(t), nil) + require.NoError(t, err) + t.Cleanup(stop) + + _, err = empty.Get("") + assert.ErrorContains(t, err, "no opa_bundle is configured") + }) +} + +func TestSetupErrors(t *testing.T) { + t.Run("missing config file", func(t *testing.T) { + _, _, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "missing", ConfigPath: filepath.Join(t.TempDir(), "nope.json")}, + }) + assert.ErrorContains(t, err, `opa_bundle "missing": failed to read config_path`) + }) + + t.Run("unparsable config", func(t *testing.T) { + _, _, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "bad", ConfigPath: writeConfigFile(t, "bad", "... not json ...")}, + }) + assert.ErrorContains(t, err, `opa_bundle "bad": failed to create OPA SDK`) + }) + + t.Run("readiness timeout", func(t *testing.T) { + path := writeConfigFile(t, "unreachable", `{ + "services": {"test": {"url": "http://127.0.0.1:1"}}, + "bundles": {"test": {"service": "test", "resource": "/bundle.tar.gz"}} + }`) + + _, _, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "unreachable", ConfigPath: path, ReadyTimeout: config.Ptr("1ms")}, + }) + assert.ErrorContains(t, err, `opa_bundle "unreachable" did not become ready in time`) + }) + + t.Run("one failure fails the whole setup", func(t *testing.T) { + _, _, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "good", ConfigPath: writeConfig(t, "good", "package p")}, + {Id: "bad", ConfigPath: filepath.Join(t.TempDir(), "nope.json")}, + }) + assert.ErrorContains(t, err, `opa_bundle "bad"`) + }) +} + +// TestSetupLogsBundleFailures is the regression for OPA's default logger being +// a buffered logger that is discarded once plugins start: an instance built +// without an explicit logger swallows every bundle download failure. +func TestSetupLogsBundleFailures(t *testing.T) { + var buf lockedBuffer + factory, _ := logutil.NewLoggerFactory(io.Discard, &buf, false) + + path := writeConfigFile(t, "unreachable", `{ + "services": {"test": {"url": "http://127.0.0.1:1"}}, + "bundles": {"test": {"service": "test", "resource": "/bundle.tar.gz"}} + }`) + + _, _, err := Setup(t.Context(), factory, []config.OpaBundle{ + {Id: "unreachable", ConfigPath: path, ReadyTimeout: config.Ptr("2s")}, + }) + require.Error(t, err) + + assert.Contains(t, buf.String(), "connection refused", + "the bundle download failure must reach NACP's logger, not OPA's discarded default") +} + +func TestReloadKeepsPreviousConfigOnFailure(t *testing.T) { + path := writeConfig(t, "platform", `package platformpolicy + errors = ["still here"]`) + + registry, stop, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "platform", ConfigPath: path}, + }) + require.NoError(t, err) + t.Cleanup(stop) + + // A config file that no longer parses must not take the policy down. + require.NoError(t, os.WriteFile(path, []byte("... not json ..."), 0o600)) + + err = registry.Reload(t.Context(), []config.OpaBundle{{Id: "platform", ConfigPath: path}}) + assert.ErrorContains(t, err, `opa_bundle "platform": failed to reconfigure`) + + instance, err := registry.Get("platform") + require.NoError(t, err) + decision, err := instance.Decide(t.Context(), "/platformpolicy", testPayload()) + require.NoError(t, err) + require.Len(t, decision.Errors, 1) + assert.EqualError(t, decision.Errors[0], "still here") +} + +func TestReloadUnknownBundle(t *testing.T) { + registry, stop, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "platform", ConfigPath: writeConfig(t, "platform", "package p")}, + }) + require.NoError(t, err) + t.Cleanup(stop) + + err = registry.Reload(t.Context(), []config.OpaBundle{{Id: "other", ConfigPath: "/nope"}}) + assert.ErrorContains(t, err, `opa_bundle "other" is not running`) +} + +func TestStatusReportsActiveRevision(t *testing.T) { + registry, stop, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "platform", ConfigPath: writeConfig(t, "platform", "package p")}, + }) + require.NoError(t, err) + t.Cleanup(stop) + + instance, err := registry.Get("platform") + require.NoError(t, err) + + // The status listener fires asynchronously from bundle activation. + require.Eventually(t, func() bool { + for _, status := range instance.Status() { + if status.Activated() { + return true + } + } + return false + }, 5*time.Second, 20*time.Millisecond, "bundle should report a successful activation") +} + +func TestDecisionTimeout(t *testing.T) { + registry, stop, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + { + Id: "explicit", + ConfigPath: writeConfig(t, "explicit", "package p"), + DecisionTimeout: config.Ptr("250ms"), + }, + { + Id: "default", + ConfigPath: writeConfig(t, "default", "package p"), + }, + { + Id: "unbounded", + ConfigPath: writeConfig(t, "unbounded", "package p"), + DecisionTimeout: config.Ptr("0s"), + }, + }) + require.NoError(t, err) + t.Cleanup(stop) + + explicit, err := registry.Get("explicit") + require.NoError(t, err) + assert.Equal(t, 250*time.Millisecond, explicit.decisionTimeout) + + byDefault, err := registry.Get("default") + require.NoError(t, err) + assert.Equal(t, config.DefaultBundleDecisionTimeout, byDefault.decisionTimeout) + + // An explicit zero opts out, leaving decisions bounded only by the request. + unbounded, err := registry.Get("unbounded") + require.NoError(t, err) + assert.Zero(t, unbounded.decisionTimeout) +} + +func TestSlogAdapterLevels(t *testing.T) { + var buf lockedBuffer + logger := slog.New(slog.NewTextHandler(&buf, &slog.HandlerOptions{Level: slog.LevelInfo})) + adapter := newSlogAdapter(logger) + + adapter.Debug("dropped %s", "debug") + adapter.Info("kept %s", "info") + adapter.WithFields(map[string]any{"bundle": "platform"}).Warn("with fields") + + out := buf.String() + assert.NotContains(t, out, "dropped debug") + assert.Contains(t, out, "kept info") + assert.Contains(t, out, "bundle=platform") +} + +// lockedBuffer collects log output written from OPA's background goroutines. +type lockedBuffer struct { + mtx sync.Mutex + buf bytes.Buffer +} + +func (b *lockedBuffer) Write(p []byte) (int, error) { + b.mtx.Lock() + defer b.mtx.Unlock() + return b.buf.Write(p) +} + +func (b *lockedBuffer) String() string { + b.mtx.Lock() + defer b.mtx.Unlock() + return b.buf.String() +} + +func TestVerifySigningConfigured(t *testing.T) { + tt := []struct { + name string + config string + expectErr string + }{ + { + name: "signed bundle", + config: `{ + "keys": {"global_key": {"algorithm": "RS256", "key": "-----BEGIN PUBLIC KEY-----"}}, + "bundles": {"demo": {"service": "s", "signing": {"keyid": "global_key"}}} + }`, + }, + { + name: "yaml form is accepted", + config: strings.Join([]string{ + "keys:", + " global_key:", + " algorithm: RS256", + "bundles:", + " demo:", + " signing:", + " keyid: global_key", + }, "\n"), + }, + { + name: "no bundles at all", + config: `{"services": {"s": {"url": "http://example.com"}}}`, + expectErr: "declares no bundles", + }, + { + name: "unsigned bundle", + config: `{"bundles": {"demo": {"service": "s"}}}`, + expectErr: `bundle "demo" has no signing block`, + }, + { + name: "signing without a key id", + config: `{"bundles": {"demo": {"signing": {"scope": "write"}}}}`, + expectErr: `bundle "demo" has no signing.keyid`, + }, + { + name: "key id not declared", + config: `{"keys": {"other": {}}, "bundles": {"demo": {"signing": {"keyid": "global_key"}}}}`, + expectErr: `references signing key "global_key" which is not declared in keys`, + }, + { + name: "unparsable config", + config: "... not yaml or json ...", + expectErr: "could not be parsed", + }, + } + + for _, tc := range tt { + t.Run(tc.name, func(t *testing.T) { + err := verifySigningConfigured([]byte(tc.config)) + if tc.expectErr == "" { + assert.NoError(t, err) + return + } + assert.ErrorContains(t, err, tc.expectErr) + }) + } +} + +func TestSetupRequireSigningRejectsUnsignedBundle(t *testing.T) { + path := writeConfigFile(t, "unsigned", `{"bundles": {"demo": {"service": "s"}}}`) + + _, _, err := Setup(t.Context(), discardFactory(t), []config.OpaBundle{ + {Id: "platform", ConfigPath: path, RequireSigning: true}, + }) + assert.ErrorContains(t, err, `opa_bundle "platform": require_signing is set but bundle "demo" has no signing block`) +} diff --git a/pkg/admissionctrl/opa/bundle/signing.go b/pkg/admissionctrl/opa/bundle/signing.go new file mode 100644 index 0000000..676603f --- /dev/null +++ b/pkg/admissionctrl/opa/bundle/signing.go @@ -0,0 +1,65 @@ +package bundle + +import ( + "fmt" + "sort" + + "gopkg.in/yaml.v3" +) + +// signingView is the minimal slice of an OPA configuration NACP needs in order +// to tell whether bundle signature verification is switched on. Everything else +// in the file, including the parts NACP knows nothing about, is handed to OPA +// verbatim — this only reads, it never rewrites. +type signingView struct { + Keys map[string]struct{} `yaml:"keys"` + Bundles map[string]struct { + Signing *struct { + KeyID string `yaml:"keyid"` + Scope string `yaml:"scope"` + KeyIDs string `yaml:"keyids"` + } `yaml:"signing"` + } `yaml:"bundles"` +} + +// verifySigningConfigured refuses a configuration whose bundles are downloaded +// without signature verification. Bundle policy runs with full authority over +// every job that passes through the proxy, so an unsigned bundle means whoever +// can answer the bundle URL can rewrite admission control. +// +// JSON is a subset of YAML, so this parses both forms OPA accepts. +func verifySigningConfigured(raw []byte) error { + var view signingView + if err := yaml.Unmarshal(raw, &view); err != nil { + return fmt.Errorf("require_signing is set but the OPA configuration could not be parsed: %w", err) + } + + if len(view.Bundles) == 0 { + return fmt.Errorf("require_signing is set but the OPA configuration declares no bundles") + } + + names := make([]string, 0, len(view.Bundles)) + for name := range view.Bundles { + names = append(names, name) + } + sort.Strings(names) + + for _, name := range names { + signing := view.Bundles[name].Signing + if signing == nil { + return fmt.Errorf("require_signing is set but bundle %q has no signing block", name) + } + keyID := signing.KeyID + if keyID == "" { + keyID = signing.KeyIDs + } + if keyID == "" { + return fmt.Errorf("require_signing is set but bundle %q has no signing.keyid", name) + } + if _, ok := view.Keys[keyID]; !ok { + return fmt.Errorf("require_signing is set but bundle %q references signing key %q which is not declared in keys", name, keyID) + } + } + + return nil +} diff --git a/pkg/admissionctrl/opa/bundle/status.go b/pkg/admissionctrl/opa/bundle/status.go new file mode 100644 index 0000000..d3ed4fc --- /dev/null +++ b/pkg/admissionctrl/opa/bundle/status.go @@ -0,0 +1,100 @@ +package bundle + +import ( + "log/slog" + "sync" + "time" + + "github.com/open-policy-agent/opa/v1/plugins/bundle" + "github.com/open-policy-agent/opa/v1/sdk" +) + +// BundleStatus is the part of OPA's bundle status NACP reports on. +type BundleStatus struct { + Name string `json:"name"` + ActiveRevision string `json:"active_revision,omitempty"` + LastSuccessfulActivation time.Time `json:"last_successful_activation,omitzero"` + Error string `json:"error,omitempty"` +} + +// Activated reports whether this bundle has ever been successfully activated. +// A bundle that has activated once keeps serving its last good policy even if +// the bundle server later disappears, so this is a liveness signal for startup +// rather than for staleness; staleness surfaces through Error. +func (s BundleStatus) Activated() bool { + return !s.LastSuccessfulActivation.IsZero() +} + +// statusTracker caches the latest status OPA reports for each bundle and warns +// whenever a refresh fails. Without it a bundle server that dies leaves NACP +// silently enforcing whatever policy it last managed to download. +type statusTracker struct { + logger *slog.Logger + + mtx sync.RWMutex + statuses map[string]BundleStatus +} + +func newStatusTracker(logger *slog.Logger) *statusTracker { + return &statusTracker{logger: logger, statuses: map[string]BundleStatus{}} +} + +func (t *statusTracker) watch(opaSDK *sdk.OPA) { + plugin, ok := opaSDK.Plugin(bundle.Name).(*bundle.Plugin) + if !ok || plugin == nil { + // No bundles configured for this instance (for example a purely + // inline-policy config); nothing to track. + return + } + plugin.RegisterBulkListener("nacp", t.update) +} + +func (t *statusTracker) update(statuses map[string]*bundle.Status) { + snapshot := make(map[string]BundleStatus, len(statuses)) + for name, status := range statuses { + if status == nil { + continue + } + entry := BundleStatus{ + Name: name, + ActiveRevision: status.ActiveRevision, + LastSuccessfulActivation: status.LastSuccessfulActivation, + } + if status.Message != "" { + entry.Error = status.Message + } + if entry.Error != "" || status.Code != "" { + t.logger.Warn("OPA bundle refresh failed, continuing with the last activated policy", + "bundle", name, + "code", status.Code, + "message", status.Message, + "active_revision", status.ActiveRevision, + "last_successful_activation", status.LastSuccessfulActivation, + ) + } + snapshot[name] = entry + } + + t.mtx.Lock() + defer t.mtx.Unlock() + t.statuses = snapshot +} + +func (t *statusTracker) snapshot() []BundleStatus { + t.mtx.RLock() + defer t.mtx.RUnlock() + + out := make([]BundleStatus, 0, len(t.statuses)) + for _, status := range t.statuses { + out = append(out, status) + } + return out +} + +// Status returns the latest known status of every bundle in this instance. +func (i *Instance) Status() []BundleStatus { + if i == nil || i.status == nil { + return nil + } + return i.status.snapshot() +} diff --git a/pkg/admissionctrl/opa/decision.go b/pkg/admissionctrl/opa/decision.go new file mode 100644 index 0000000..45f1bab --- /dev/null +++ b/pkg/admissionctrl/opa/decision.go @@ -0,0 +1,134 @@ +package opa + +import ( + "errors" + "fmt" +) + +// Decision is the document every NACP policy is expected to produce, no matter +// whether it is evaluated as an embedded rego file or as a rule inside a remote +// bundle: optional errors, optional warnings, and for mutators a JSON Patch. +type Decision struct { + Errors []error + Warnings []error + Patch []interface{} + // HasPatch distinguishes a policy that deliberately returned no patch from + // one that returned an empty list of operations. + HasPatch bool +} + +// ParseMode selects how strictly a decision document is interpreted. +type ParseMode int + +const ( + // Strict rejects anything that does not match the documented decision + // shape. Bundle adapters use it so that a mistyped decision path, or a + // policy that yields a scalar instead of an object, fails the admission + // instead of silently letting the job through. + Strict ParseMode = iota + // Lenient ignores values that do not match the documented shape. The + // embedded adapters use it to keep the behaviour they shipped with, where a + // missing or wrongly typed binding is treated as "nothing to report". + Lenient +) + +// ParseDecision converts the raw result of an OPA evaluation into a Decision. +func ParseDecision(raw interface{}, mode ParseMode) (*Decision, error) { + decision := &Decision{Patch: []interface{}{}} + + document, err := decisionDocument(raw, mode) + if err != nil { + return nil, err + } + if document == nil { + return decision, nil + } + + decision.Errors, err = parseMessages(document["errors"], "error", mode) + if err != nil { + return nil, err + } + decision.Warnings, err = parseMessages(document["warnings"], "warning", mode) + if err != nil { + return nil, err + } + decision.Patch, decision.HasPatch, err = parsePatch(document["patch"], mode) + if err != nil { + return nil, err + } + + return decision, nil +} + +func decisionDocument(raw interface{}, mode ParseMode) (map[string]interface{}, error) { + if document, ok := raw.(map[string]interface{}); ok { + return document, nil + } + if mode == Lenient { + return nil, nil + } + if raw == nil { + return nil, errors.New("policy decision is undefined") + } + return nil, fmt.Errorf("policy yielded an invalid decision value: %v", raw) +} + +// parseMessages reads an errors or warnings list. noun is the singular form +// used in error messages, so the plural collection reads "errors value" while a +// bad element reads "error entry value". +func parseMessages(raw interface{}, noun string, mode ParseMode) ([]error, error) { + if raw == nil { + return nil, nil + } + + entries, ok := raw.([]interface{}) + if !ok { + if mode == Lenient { + return nil, nil + } + return nil, fmt.Errorf("policy yielded an invalid %ss value: %v", noun, raw) + } + + messages := make([]error, 0, len(entries)) + for _, entry := range entries { + if entry == nil { + continue + } + message, ok := entry.(string) + if !ok { + if mode == Lenient { + // The embedded adapters have always rendered whatever the policy + // produced, so keep doing that rather than dropping the entry. + messages = append(messages, fmt.Errorf("%v", entry)) + continue + } + return nil, fmt.Errorf("policy yielded an invalid %s entry value: %v", noun, entry) + } + messages = append(messages, errors.New(message)) + } + return messages, nil +} + +func parsePatch(raw interface{}, mode ParseMode) ([]interface{}, bool, error) { + if raw == nil { + return []interface{}{}, false, nil + } + + operations, ok := raw.([]interface{}) + if !ok { + if mode == Lenient { + return []interface{}{}, false, nil + } + return nil, false, fmt.Errorf("policy yielded an invalid patch value: %v", raw) + } + + if mode == Strict { + for _, operation := range operations { + if _, ok := operation.(map[string]interface{}); !ok { + return nil, false, fmt.Errorf("policy yielded an invalid patch entry value: %v", operation) + } + } + } + + return operations, true, nil +} diff --git a/pkg/admissionctrl/opa/decision_test.go b/pkg/admissionctrl/opa/decision_test.go new file mode 100644 index 0000000..b12486c --- /dev/null +++ b/pkg/admissionctrl/opa/decision_test.go @@ -0,0 +1,178 @@ +package opa + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseDecisionStrict(t *testing.T) { + tt := []struct { + name string + raw interface{} + expectErr string + expectErrs []string + expectWarns []string + expectPatch []interface{} + expectHasPch bool + }{ + { + name: "empty document", + raw: map[string]interface{}{}, + }, + { + name: "errors", + raw: map[string]interface{}{"errors": []interface{}{"boom", "bang"}}, + expectErrs: []string{"boom", "bang"}, + }, + { + name: "warnings", + raw: map[string]interface{}{"warnings": []interface{}{"careful"}}, + expectWarns: []string{"careful"}, + }, + { + name: "patch", + raw: map[string]interface{}{"patch": []interface{}{map[string]interface{}{"op": "remove", "path": "/Meta"}}}, + expectPatch: []interface{}{map[string]interface{}{"op": "remove", "path": "/Meta"}}, + expectHasPch: true, + }, + { + name: "empty patch list is still a patch", + raw: map[string]interface{}{"patch": []interface{}{}}, + expectPatch: []interface{}{}, + expectHasPch: true, + }, + { + name: "null patch is not a patch", + raw: map[string]interface{}{"patch": nil}, + }, + { + name: "undefined decision", + raw: nil, + expectErr: "policy decision is undefined", + }, + { + name: "scalar decision does not mean allow", + raw: true, + expectErr: "policy yielded an invalid decision value", + }, + { + name: "list decision does not mean allow", + raw: []interface{}{"nope"}, + expectErr: "policy yielded an invalid decision value", + }, + { + name: "errors not a list", + raw: map[string]interface{}{"errors": 5}, + expectErr: "policy yielded an invalid errors value", + }, + { + name: "error entry not a string", + raw: map[string]interface{}{"errors": []interface{}{"fine", 5}}, + expectErr: "policy yielded an invalid error entry value", + }, + { + name: "warnings not a list", + raw: map[string]interface{}{"warnings": 5}, + expectErr: "policy yielded an invalid warnings value", + }, + { + name: "warning entry not a string", + raw: map[string]interface{}{"warnings": []interface{}{5}}, + expectErr: "policy yielded an invalid warning entry value", + }, + { + name: "patch not a list", + raw: map[string]interface{}{"patch": 5}, + expectErr: "policy yielded an invalid patch value", + }, + { + name: "patch entry not an object", + raw: map[string]interface{}{"patch": []interface{}{5}}, + expectErr: "policy yielded an invalid patch entry value", + }, + } + + for _, tc := range tt { + t.Run(tc.name, func(t *testing.T) { + decision, err := ParseDecision(tc.raw, Strict) + + if tc.expectErr != "" { + assert.ErrorContains(t, err, tc.expectErr) + assert.Nil(t, decision) + return + } + + require.NoError(t, err) + assert.Equal(t, tc.expectErrs, messages(decision.Errors)) + assert.Equal(t, tc.expectWarns, messages(decision.Warnings)) + assert.Equal(t, tc.expectHasPch, decision.HasPatch) + if tc.expectPatch != nil { + assert.Equal(t, tc.expectPatch, decision.Patch) + } + }) + } +} + +// TestParseDecisionLenient pins the tolerance the embedded adapters shipped +// with: anything that does not match the shape reports nothing rather than +// failing the request. +func TestParseDecisionLenient(t *testing.T) { + tt := []struct { + name string + raw interface{} + expectErrs []string + expectWarns []string + expectPatch []interface{} + }{ + {name: "undefined decision", raw: nil, expectPatch: []interface{}{}}, + {name: "scalar decision", raw: true, expectPatch: []interface{}{}}, + {name: "missing bindings", raw: map[string]interface{}{}, expectPatch: []interface{}{}}, + { + name: "errors not a list", + raw: map[string]interface{}{"errors": 5}, + expectPatch: []interface{}{}, + }, + { + name: "warnings not a list", + raw: map[string]interface{}{"warnings": 5}, + expectPatch: []interface{}{}, + }, + { + name: "patch not a list", + raw: map[string]interface{}{"patch": 5}, + expectPatch: []interface{}{}, + }, + { + name: "non string entries are rendered", + raw: map[string]interface{}{"errors": []interface{}{5}, "warnings": []interface{}{true}}, + expectErrs: []string{"5"}, + expectWarns: []string{"true"}, + expectPatch: []interface{}{}, + }, + } + + for _, tc := range tt { + t.Run(tc.name, func(t *testing.T) { + decision, err := ParseDecision(tc.raw, Lenient) + + require.NoError(t, err) + assert.Equal(t, tc.expectErrs, messages(decision.Errors)) + assert.Equal(t, tc.expectWarns, messages(decision.Warnings)) + assert.Equal(t, tc.expectPatch, decision.Patch) + assert.False(t, decision.HasPatch) + }) + } +} + +func messages(errs []error) []string { + if len(errs) == 0 { + return nil + } + out := make([]string, 0, len(errs)) + for _, err := range errs { + out = append(out, err.Error()) + } + return out +} diff --git a/pkg/admissionctrl/opa/opa.go b/pkg/admissionctrl/opa/opa.go index 25aeea2..62a6fad 100644 --- a/pkg/admissionctrl/opa/opa.go +++ b/pkg/admissionctrl/opa/opa.go @@ -72,31 +72,17 @@ func (q *OpaQuery) Query(ctx context.Context, payload *types2.Payload) (*OpaQuer return &OpaQueryResult{&resultSet}, nil } -func (result *OpaQueryResult) GetWarnings() []interface{} { - - rs := *result.resultSet - - warnings, ok := rs[0].Bindings["warnings"].([]interface{}) - if !ok { - return make([]interface{}, 0) - } - return warnings -} -func (result *OpaQueryResult) GetErrors() []interface{} { - +// Decision reads the query bindings through the same decision contract the +// bundle adapters use. Lenient mode keeps the tolerance the embedded adapters +// shipped with: a missing or wrongly typed binding reports nothing. +func (result *OpaQueryResult) Decision() *Decision { rs := *result.resultSet - errors, ok := rs[0].Bindings["errors"].([]interface{}) - if !ok { - return make([]interface{}, 0) - } - return errors -} -func (result *OpaQueryResult) GetPatch() []interface{} { - rs := *result.resultSet - patch, ok := rs[0].Bindings["patch"].([]interface{}) - if !ok { - return make([]interface{}, 0) + decision, err := ParseDecision(map[string]interface{}(rs[0].Bindings), Lenient) + if err != nil { + // Lenient parsing never fails, but do not let a future change here turn + // into a silent allow. + return &Decision{Errors: []error{err}, Patch: []interface{}{}} } - return patch + return decision } diff --git a/pkg/admissionctrl/opa/opa_test.go b/pkg/admissionctrl/opa/opa_test.go index 811609f..81154f0 100644 --- a/pkg/admissionctrl/opa/opa_test.go +++ b/pkg/admissionctrl/opa/opa_test.go @@ -1,4 +1,4 @@ -package opa +package opa_test import ( "context" @@ -9,6 +9,7 @@ import ( "github.com/hashicorp/nomad/api" "github.com/mxab/nacp/pkg/admissionctrl/notation" + . "github.com/mxab/nacp/pkg/admissionctrl/opa" "github.com/mxab/nacp/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -33,20 +34,24 @@ func TestOpa(t *testing.T) { assert.Nil(t, err, "No error executing query") assert.NotNil(t, result, "Result is not nil") - warnings := result.GetWarnings() - assert.Equal(t, []interface{}{"This is a warning message"}, warnings, "Warnings are correct") - - errors := result.GetErrors() - assert.Equal(t, []interface{}{"This is a error message"}, errors, "Errors are correct") - - patch := result.GetPatch() + decision := result.Decision() + assert.Equal(t, []string{"This is a warning message"}, errorMessages(decision.Warnings), "Warnings are correct") + assert.Equal(t, []string{"This is a error message"}, errorMessages(decision.Errors), "Errors are correct") assert.Equal(t, []interface{}{ map[string]interface{}{ "op": "add", "path": "/Meta", "value": map[string]interface{}{"foo": "bar"}, }, - }, patch, "Patch is correct") + }, decision.Patch, "Patch is correct") +} + +func errorMessages(errs []error) []string { + out := make([]string, 0, len(errs)) + for _, err := range errs { + out = append(out, err.Error()) + } + return out } func TestFailOnEmptyResultSet(t *testing.T) { ctx := context.Background() @@ -84,12 +89,11 @@ func TestReturnsEmptyIfNotExisting(t *testing.T) { assert.Nil(t, err, "No error executing query") assert.NotNil(t, result, "Result is not nil") - warnings := result.GetWarnings() - assert.Equal(t, []interface{}{}, warnings, "Warnings are correct") - errors := result.GetErrors() - assert.Equal(t, []interface{}{}, errors, "Errors are correct") - patch := result.GetPatch() - assert.Equal(t, []interface{}{}, patch, "Patch is correct") + decision := result.Decision() + assert.Empty(t, decision.Warnings, "Warnings are correct") + assert.Empty(t, decision.Errors, "Errors are correct") + assert.Equal(t, []interface{}{}, decision.Patch, "Patch is correct") + assert.False(t, decision.HasPatch, "No patch was produced") } @@ -113,19 +117,19 @@ func TestNotationImageValidation(t *testing.T) { name string image string verifier notation.ImageVerifier - expectedErrors []interface{} + expectedErrors []string }{ { name: "valid image", image: "validimage:latest", verifier: new(DummyVerifier), - expectedErrors: []interface{}{}, + expectedErrors: []string{}, }, { name: "invalid image", image: "invalidimage:latest", verifier: new(DummyVerifier), - expectedErrors: []interface{}{ + expectedErrors: []string{ "Image is not in valid", }, }, @@ -163,8 +167,7 @@ func TestNotationImageValidation(t *testing.T) { require.NoError(t, err, "No error executing query") require.NotNil(t, result, "Result is not nil") - errors := result.GetErrors() - assert.Equal(t, tc.expectedErrors, errors, "Errors are correct") + assert.Equal(t, tc.expectedErrors, errorMessages(result.Decision().Errors), "Errors are correct") }) } } diff --git a/pkg/admissionctrl/validator/opa_bundle_validator.go b/pkg/admissionctrl/validator/opa_bundle_validator.go index 3740b27..e18ef98 100644 --- a/pkg/admissionctrl/validator/opa_bundle_validator.go +++ b/pkg/admissionctrl/validator/opa_bundle_validator.go @@ -3,27 +3,24 @@ package validator import ( "context" "errors" - "fmt" - "log/slog" "github.com/hashicorp/go-multierror" "github.com/mxab/nacp/pkg/admissionctrl" + "github.com/mxab/nacp/pkg/admissionctrl/opa/bundle" "github.com/mxab/nacp/pkg/admissionctrl/types" - "github.com/open-policy-agent/opa/v1/sdk" ) type OpaBundleValidator struct { name string path string - logger *slog.Logger - opa *sdk.OPA + bundle *bundle.Instance } var _ admissionctrl.JobValidator = (*OpaBundleValidator)(nil) // Verify that *T implements I. -func NewOpaBundleValidator(name string, path string, logger *slog.Logger, opaSDK *sdk.OPA) (*OpaBundleValidator, error) { - if opaSDK == nil { - return nil, errors.New("OPA SDK is required") +func NewOpaBundleValidator(name string, path string, instance *bundle.Instance) (*OpaBundleValidator, error) { + if instance == nil { + return nil, errors.New("OPA bundle is required") } if path == "" { return nil, errors.New("OPA decision path is required") @@ -31,58 +28,23 @@ func NewOpaBundleValidator(name string, path string, logger *slog.Logger, opaSDK return &OpaBundleValidator{ name: name, path: path, - logger: logger, - opa: opaSDK, + bundle: instance, }, nil } -func (v *OpaBundleValidator) Validate(ctx context.Context, payload *types.Payload) (warnings []error, err error) { - - result, err := v.opa.Decision(ctx, sdk.DecisionOptions{ - Input: payload, - Path: v.path, - }) +func (v *OpaBundleValidator) Validate(ctx context.Context, payload *types.Payload) ([]error, error) { + // Decide parses the result strictly: a decision that is not the documented + // {errors, warnings} document fails the admission instead of being read as + // "the policy found nothing". + decision, err := v.bundle.Decide(ctx, v.path, payload) if err != nil { - return nil, fmt.Errorf("failed to perform policy decision: %w", err) + return nil, err } - v.logger.DebugContext(ctx, "OPA decision", slog.Any("result", result)) - - if rmap, ok := result.Result.(map[string]interface{}); ok { - if errs, found := rmap["errors"]; found { - if errlist, ok := errs.([]interface{}); ok { - - for _, e := range errlist { - if emsg, ok := e.(string); ok { - err = multierror.Append(err, errors.New(emsg)) - } else { - err = multierror.Append(err, fmt.Errorf("policy yielded an invalid error value: %v", e)) - } - } - if err != nil { - return - } - } else if errs != nil { - err = fmt.Errorf("policy yielded an invalid errors value: %v", errs) - return - } - } - if warns, found := rmap["warnings"]; found { - if warnlist, ok := warns.([]interface{}); ok { - for _, w := range warnlist { - if wmsg, ok := w.(string); ok { - warnings = append(warnings, errors.New(wmsg)) - } else { - warnings = append(warnings, fmt.Errorf("policy yielded an invalid warning value: %v", w)) - } - } - } else if warns != nil { - warnings = append(warnings, fmt.Errorf("policy yielded an invalid warnings value: %v", warns)) - } - } + if len(decision.Errors) > 0 { + return decision.Warnings, multierror.Append(nil, decision.Errors...) } - - return + return decision.Warnings, nil } func (v *OpaBundleValidator) Name() string { diff --git a/pkg/admissionctrl/validator/opa_bundle_validator_test.go b/pkg/admissionctrl/validator/opa_bundle_validator_test.go index 8d747e1..c60912e 100644 --- a/pkg/admissionctrl/validator/opa_bundle_validator_test.go +++ b/pkg/admissionctrl/validator/opa_bundle_validator_test.go @@ -1,7 +1,6 @@ package validator import ( - "log/slog" "testing" "github.com/mxab/nacp/pkg/admissionctrl/types" @@ -59,28 +58,45 @@ func TestOpaBundleValidator(t *testing.T) { policy: `package mypolicy errors = [5]`, path: "/mypolicy", - expectErrParts: []string{"policy yielded an invalid error value"}, + expectErrParts: []string{"policy yielded an invalid error entry value"}, }, { name: "handle invalid warnings value", policy: `package mypolicy warnings = 5`, - path: "/mypolicy", - expectWarns: []string{"policy yielded an invalid warnings value"}, + path: "/mypolicy", + expectErrParts: []string{"policy yielded an invalid warnings value"}, }, { - name: "handle invalid warnings value", + name: "handle invalid warning entry value", policy: `package mypolicy warnings = [5]`, - path: "/mypolicy", - expectWarns: []string{"policy yielded an invalid warning value"}, + path: "/mypolicy", + expectErrParts: []string{"policy yielded an invalid warning entry value"}, + }, + { + // A decision path that resolves to something other than the + // documented {errors, warnings, patch} document must fail the + // admission, not be read as "the policy found nothing". + name: "reject non-object decision", + policy: `package mypolicy + allow := true`, + path: "/mypolicy/allow", + expectErrParts: []string{"policy yielded an invalid decision value"}, + }, + { + name: "reject list decision", + policy: `package mypolicy + findings := ["nope"]`, + path: "/mypolicy/findings", + expectErrParts: []string{"policy yielded an invalid decision value"}, }, { name: "test invalid policy path", policy: `package mypolicy errors = ["an error message"]`, path: "/invalidpath", - expectErrParts: []string{"failed to perform policy decision"}, + expectErrParts: []string{"is undefined in the active bundle"}, }, } @@ -89,7 +105,7 @@ func TestOpaBundleValidator(t *testing.T) { job := testutil.BaseJob() opa := testutil.SetupOpa(t, tc.policy) - validator, err := NewOpaBundleValidator("testopabundlevalidator", tc.path, slog.New(slog.DiscardHandler), opa) + validator, err := NewOpaBundleValidator("testopabundlevalidator", tc.path, opa) require.NoError(t, err, "No error creating validator") @@ -115,7 +131,7 @@ func TestOpaBundleValidator(t *testing.T) { func TestBundleValidatorName(t *testing.T) { opa := testutil.SetupOpa(t, "package mypolicy") - validator, err := NewOpaBundleValidator("testopabundlevalidator", "/mypolicy", slog.New(slog.DiscardHandler), opa) + validator, err := NewOpaBundleValidator("testopabundlevalidator", "/mypolicy", opa) require.NoError(t, err, "No error creating validator") assert.Equal(t, "testopabundlevalidator", validator.Name(), "Validator name") @@ -124,11 +140,11 @@ func TestBundleValidatorName(t *testing.T) { func TestNewOpaBundleValidatorValidation(t *testing.T) { opa := testutil.SetupOpa(t, "package mypolicy") - validator, err := NewOpaBundleValidator("test", "/mypolicy", slog.Default(), nil) - assert.ErrorContains(t, err, "OPA SDK is required") + validator, err := NewOpaBundleValidator("test", "/mypolicy", nil) + assert.ErrorContains(t, err, "OPA bundle is required") assert.Nil(t, validator) - validator, err = NewOpaBundleValidator("test", "", slog.Default(), opa) + validator, err = NewOpaBundleValidator("test", "", opa) assert.ErrorContains(t, err, "OPA decision path is required") assert.Nil(t, validator) } diff --git a/pkg/admissionctrl/validator/opa_validator.go b/pkg/admissionctrl/validator/opa_validator.go index b5b786c..60de4c9 100644 --- a/pkg/admissionctrl/validator/opa_validator.go +++ b/pkg/admissionctrl/validator/opa_validator.go @@ -32,22 +32,20 @@ func (v *OpaValidator) Validate(ctx context.Context, payload *types.Payload) ([] return nil, err } - // aggregate warnings - warnings := results.GetWarnings() + decision := results.Decision() - if len(warnings) > 0 { - v.logger.Debug("Got warnings from rule", "rule", v.Name(), "warnings", warnings, "job", payload.Job.ID) - for _, warn := range warnings { + // aggregate warnings + if len(decision.Warnings) > 0 { + v.logger.Debug("Got warnings from rule", "rule", v.Name(), "warnings", decision.Warnings, "job", payload.Job.ID) + for _, warn := range decision.Warnings { allWarnings = append(allWarnings, fmt.Errorf("%s (%s)", warn, v.Name())) } } - errors := results.GetErrors() - - if len(errors) > 0 { // no errors is ok - v.logger.Debug("Got errors from rule", "rule", v.Name(), "errors", errors, "job", payload.Job.ID) + if len(decision.Errors) > 0 { // no errors is ok + v.logger.Debug("Got errors from rule", "rule", v.Name(), "errors", decision.Errors, "job", payload.Job.ID) errsForRule := &multierror.Error{} - for _, err := range errors { + for _, err := range decision.Errors { errsForRule = multierror.Append(errsForRule, fmt.Errorf("%s (%s)", err, v.Name())) } allErrs = multierror.Append(allErrs, errsForRule) diff --git a/pkg/config/config.go b/pkg/config/config.go index 23519b2..1452182 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -3,6 +3,8 @@ package config import ( "fmt" "slices" + "strings" + "time" "github.com/hashicorp/hcl/v2" "github.com/hashicorp/hcl/v2/hclsimple" @@ -22,15 +24,20 @@ type OpaRule struct { Filename string `hcl:"filename"` Notation *NotationVerifierConfig `hcl:"notation,block"` } -type OpaSdkRule struct { - Path string `hcl:"path"` + +// BundleRule selects the decision to evaluate inside a bundle. Source names the +// opa_bundle block to evaluate it against and may be omitted when exactly one +// bundle is configured. +type BundleRule struct { + Source string `hcl:"source,optional"` + Path string `hcl:"path"` } type Validator struct { Type string `hcl:"type,label"` Name string `hcl:"name,label"` OpaRule *OpaRule `hcl:"opa_rule,block"` - OpaSdkRule *OpaSdkRule `hcl:"opa_sdk_rule,block"` + BundleRule *BundleRule `hcl:"bundle_rule,block"` Webhook *Webhook `hcl:"webhook,block"` ResolveToken bool `hcl:"resolve_token,optional"` @@ -40,7 +47,7 @@ type Mutator struct { Type string `hcl:"type,label"` Name string `hcl:"name,label"` OpaRule *OpaRule `hcl:"opa_rule,block"` - OpaSdkRule *OpaSdkRule `hcl:"opa_sdk_rule,block"` + BundleRule *BundleRule `hcl:"bundle_rule,block"` Webhook *Webhook `hcl:"webhook,block"` ResolveToken bool `hcl:"resolve_token,optional"` } @@ -117,11 +124,54 @@ type Config struct { Telemetry *Telemetry `hcl:"telemetry,block"` - OpaSdk *OpaSdk `hcl:"opa_sdk,block"` + OpaBundles []OpaBundle `hcl:"opa_bundle,block"` } -type OpaSdk struct { + +// OpaBundle is one OPA SDK instance fed by its own OPA configuration file, and +// therefore its own bundle services, signing keys and refresh settings. +type OpaBundle struct { Id string `hcl:"id,label"` ConfigPath string `hcl:"config_path"` + + // ReadyTimeout bounds how long startup waits for the first bundle + // activation. Defaults to DefaultBundleReadyTimeout. + ReadyTimeout *string `hcl:"ready_timeout,optional"` + // DecisionTimeout bounds a single policy evaluation. Defaults to + // DefaultBundleDecisionTimeout; "0" inherits the request deadline. + DecisionTimeout *string `hcl:"decision_timeout,optional"` + // RequireSigning refuses to start unless every bundle in the OPA + // configuration is configured for signature verification. + RequireSigning bool `hcl:"require_signing,optional"` +} + +const ( + DefaultBundleReadyTimeout = 30 * time.Second + DefaultBundleDecisionTimeout = 5 * time.Second +) + +// ResolvedReadyTimeout parses ready_timeout, falling back to the default. +func (b OpaBundle) ResolvedReadyTimeout() (time.Duration, error) { + return parseOptionalDuration(b.ReadyTimeout, DefaultBundleReadyTimeout, "ready_timeout") +} + +// ResolvedDecisionTimeout parses decision_timeout, falling back to the default. +// A zero duration means decisions are only bounded by the request context. +func (b OpaBundle) ResolvedDecisionTimeout() (time.Duration, error) { + return parseOptionalDuration(b.DecisionTimeout, DefaultBundleDecisionTimeout, "decision_timeout") +} + +func parseOptionalDuration(raw *string, fallback time.Duration, field string) (time.Duration, error) { + if raw == nil || *raw == "" { + return fallback, nil + } + d, err := time.ParseDuration(*raw) + if err != nil { + return 0, fmt.Errorf("invalid %s %q: %w", field, *raw, err) + } + if d < 0 { + return 0, fmt.Errorf("invalid %s %q: must not be negative", field, *raw) + } + return d, nil } func DefaultConfig() *Config { @@ -176,13 +226,138 @@ func LoadConfig(name string) (*Config, error) { } } + if err := c.Validate(); err != nil { + return nil, err + } + return c, nil +} + +// Validate rejects a configuration before anything is built from it, so that a +// missing block surfaces as a config error rather than a nil dereference or a +// half-started process. +func (c *Config) Validate() error { // verify json/text out var validOuts = []string{"stdout", "stderr"} if !slices.Contains(validOuts, *c.Telemetry.Logging.SlogLogging.TextOut) { - return nil, fmt.Errorf("invalid slog text output: %s", *c.Telemetry.Logging.SlogLogging.TextOut) + return fmt.Errorf("invalid slog text output: %s", *c.Telemetry.Logging.SlogLogging.TextOut) } if !slices.Contains(validOuts, *c.Telemetry.Logging.SlogLogging.JsonOut) { - return nil, fmt.Errorf("invalid slog json output: %s", *c.Telemetry.Logging.SlogLogging.JsonOut) + return fmt.Errorf("invalid slog json output: %s", *c.Telemetry.Logging.SlogLogging.JsonOut) } - return c, nil + + bundleIds, err := c.validateBundles() + if err != nil { + return err + } + + for _, v := range c.Validators { + if err := validateController("validator", v.Type, v.Name, controllerBlocks{ + opaRule: v.OpaRule, + bundleRule: v.BundleRule, + webhook: v.Webhook, + notation: v.Notation, + }, bundleIds); err != nil { + return err + } + } + for _, m := range c.Mutators { + if err := validateController("mutator", m.Type, m.Name, controllerBlocks{ + opaRule: m.OpaRule, + bundleRule: m.BundleRule, + webhook: m.Webhook, + }, bundleIds); err != nil { + return err + } + } + return nil +} + +func (c *Config) validateBundles() ([]string, error) { + ids := make([]string, 0, len(c.OpaBundles)) + for _, b := range c.OpaBundles { + if b.Id == "" { + return nil, fmt.Errorf("opa_bundle block requires a non-empty id label") + } + if slices.Contains(ids, b.Id) { + return nil, fmt.Errorf("duplicate opa_bundle %q", b.Id) + } + if b.ConfigPath == "" { + return nil, fmt.Errorf("opa_bundle %q requires config_path", b.Id) + } + if _, err := b.ResolvedReadyTimeout(); err != nil { + return nil, fmt.Errorf("opa_bundle %q: %w", b.Id, err) + } + if _, err := b.ResolvedDecisionTimeout(); err != nil { + return nil, fmt.Errorf("opa_bundle %q: %w", b.Id, err) + } + ids = append(ids, b.Id) + } + return ids, nil +} + +type controllerBlocks struct { + opaRule *OpaRule + bundleRule *BundleRule + webhook *Webhook + notation *NotationVerifierConfig +} + +func validateController(role, typ, name string, blocks controllerBlocks, bundleIds []string) error { + required := func(present bool, block string) error { + if present { + return nil + } + return fmt.Errorf("%s %q of type %q requires a %s block", role, name, typ, block) + } + + switch typ { + case "opa", "opa_json_patch": + if err := required(blocks.opaRule != nil, "opa_rule"); err != nil { + return err + } + if blocks.opaRule.Filename == "" { + return fmt.Errorf("%s %q requires opa_rule.filename", role, name) + } + if blocks.opaRule.Query == "" { + return fmt.Errorf("%s %q requires opa_rule.query", role, name) + } + case "opa_bundle", "opa_bundle_json_patch": + if err := required(blocks.bundleRule != nil, "bundle_rule"); err != nil { + return err + } + if blocks.bundleRule.Path == "" { + return fmt.Errorf("%s %q requires bundle_rule.path", role, name) + } + if err := validateBundleSource(role, name, blocks.bundleRule.Source, bundleIds); err != nil { + return err + } + case "webhook", "json_patch_webhook": + if err := required(blocks.webhook != nil, "webhook"); err != nil { + return err + } + if blocks.webhook.Endpoint == "" { + return fmt.Errorf("%s %q requires webhook.endpoint", role, name) + } + case "notation": + if err := required(blocks.notation != nil, "notation"); err != nil { + return err + } + default: + return fmt.Errorf("unknown %s type %s", role, typ) + } + return nil +} + +func validateBundleSource(role, name, source string, bundleIds []string) error { + switch { + case len(bundleIds) == 0: + return fmt.Errorf("%s %q references a bundle but no opa_bundle block is configured", role, name) + case source == "" && len(bundleIds) > 1: + return fmt.Errorf("%s %q must set bundle_rule.source, configured bundles: %s", + role, name, strings.Join(bundleIds, ", ")) + case source != "" && !slices.Contains(bundleIds, source): + return fmt.Errorf("%s %q references unknown bundle_rule.source %q, configured bundles: %s", + role, name, source, strings.Join(bundleIds, ", ")) + } + return nil } diff --git a/pkg/config/config_test.go b/pkg/config/config_test.go index 5a2a5b3..04bb9c0 100644 --- a/pkg/config/config_test.go +++ b/pkg/config/config_test.go @@ -249,8 +249,8 @@ func TestLoadConfig(t *testing.T) { wantErr: false, }, { - name: "with opa sdk", - args: args{name: "testdata/with_opa_sdk.hcl"}, + name: "with opa bundles", + args: args{name: "testdata/with_opa_bundle.hcl"}, want: &Config{ Port: port, Bind: bind, @@ -262,8 +262,9 @@ func TestLoadConfig(t *testing.T) { { Type: "opa_bundle", Name: "some_validator", - OpaSdkRule: &OpaSdkRule{ - Path: "/my/validation/policy", + BundleRule: &BundleRule{ + Source: "platform", + Path: "/my/validation/policy", }, }, }, @@ -271,8 +272,9 @@ func TestLoadConfig(t *testing.T) { { Type: "opa_bundle_json_patch", Name: "some_mutator", - OpaSdkRule: &OpaSdkRule{ - Path: "/my/mutation/policy", + BundleRule: &BundleRule{ + Source: "team_a", + Path: "/my/mutation/policy", }, }, }, @@ -296,10 +298,18 @@ func TestLoadConfig(t *testing.T) { Enabled: false, }, }, - OpaSdk: &OpaSdk{ - - Id: "example", - ConfigPath: "/my/path/to/config.json", + OpaBundles: []OpaBundle{ + { + Id: "platform", + ConfigPath: "/my/path/to/config.json", + RequireSigning: true, + }, + { + Id: "team_a", + ConfigPath: "/my/path/to/team-a.json", + ReadyTimeout: Ptr("45s"), + DecisionTimeout: Ptr("2s"), + }, }, }, }, diff --git a/pkg/config/testdata/with_opa_bundle.hcl b/pkg/config/testdata/with_opa_bundle.hcl new file mode 100644 index 0000000..bde919d --- /dev/null +++ b/pkg/config/testdata/with_opa_bundle.hcl @@ -0,0 +1,30 @@ + +opa_bundle "platform" { + config_path = "/my/path/to/config.json" + + require_signing = true +} + +opa_bundle "team_a" { + config_path = "/my/path/to/team-a.json" + + ready_timeout = "45s" + decision_timeout = "2s" +} + +validator "opa_bundle" "some_validator" { + + bundle_rule { + source = "platform" + path = "/my/validation/policy" + } +} + +mutator "opa_bundle_json_patch" "some_mutator" { + + bundle_rule { + source = "team_a" + path = "/my/mutation/policy" + } + +} diff --git a/pkg/config/testdata/with_opa_sdk.hcl b/pkg/config/testdata/with_opa_sdk.hcl deleted file mode 100644 index 56aa45e..0000000 --- a/pkg/config/testdata/with_opa_sdk.hcl +++ /dev/null @@ -1,19 +0,0 @@ - -opa_sdk "example" { - config_path = "/my/path/to/config.json" -} - -validator "opa_bundle" "some_validator" { - - opa_sdk_rule { - path = "/my/validation/policy" - } -} - -mutator "opa_bundle_json_patch" "some_mutator" { - - opa_sdk_rule { - path = "/my/mutation/policy" - } - -} diff --git a/pkg/config/validate_test.go b/pkg/config/validate_test.go new file mode 100644 index 0000000..09706c5 --- /dev/null +++ b/pkg/config/validate_test.go @@ -0,0 +1,147 @@ +package config + +import ( + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestValidateRejectsIncompleteControllers covers the configurations that used +// to reach startup and panic on a nil block, or fail late once the OPA SDK was +// already running. +func TestValidateRejectsIncompleteControllers(t *testing.T) { + tt := []struct { + name string + mutate func(*Config) + expectErr string + }{ + { + name: "opa validator without opa_rule", + mutate: func(c *Config) { + c.Validators = []Validator{{Type: "opa", Name: "x"}} + }, + expectErr: `validator "x" of type "opa" requires a opa_rule block`, + }, + { + name: "opa validator without filename", + mutate: func(c *Config) { + c.Validators = []Validator{{Type: "opa", Name: "x", OpaRule: &OpaRule{Query: "q"}}} + }, + expectErr: `validator "x" requires opa_rule.filename`, + }, + { + name: "webhook mutator without webhook block", + mutate: func(c *Config) { + c.Mutators = []Mutator{{Type: "json_patch_webhook", Name: "x"}} + }, + expectErr: `mutator "x" of type "json_patch_webhook" requires a webhook block`, + }, + { + name: "notation validator without notation block", + mutate: func(c *Config) { + c.Validators = []Validator{{Type: "notation", Name: "x"}} + }, + expectErr: `validator "x" of type "notation" requires a notation block`, + }, + { + name: "unknown validator type", + mutate: func(c *Config) { + c.Validators = []Validator{{Type: "opa_sdk", Name: "x"}} + }, + expectErr: "unknown validator type opa_sdk", + }, + { + name: "bundle validator without bundle_rule", + mutate: func(c *Config) { + c.OpaBundles = []OpaBundle{{Id: "platform", ConfigPath: "/opa.yml"}} + c.Validators = []Validator{{Type: "opa_bundle", Name: "x"}} + }, + expectErr: `validator "x" of type "opa_bundle" requires a bundle_rule block`, + }, + { + name: "bundle validator without any configured bundle", + mutate: func(c *Config) { + c.Validators = []Validator{{Type: "opa_bundle", Name: "x", BundleRule: &BundleRule{Path: "/p"}}} + }, + expectErr: `validator "x" references a bundle but no opa_bundle block is configured`, + }, + { + name: "bundle validator must name a source when several exist", + mutate: func(c *Config) { + c.OpaBundles = []OpaBundle{ + {Id: "platform", ConfigPath: "/a.yml"}, + {Id: "team", ConfigPath: "/b.yml"}, + } + c.Validators = []Validator{{Type: "opa_bundle", Name: "x", BundleRule: &BundleRule{Path: "/p"}}} + }, + expectErr: `validator "x" must set bundle_rule.source, configured bundles: platform, team`, + }, + { + name: "bundle validator with unknown source", + mutate: func(c *Config) { + c.OpaBundles = []OpaBundle{{Id: "platform", ConfigPath: "/a.yml"}} + c.Validators = []Validator{{Type: "opa_bundle", Name: "x", BundleRule: &BundleRule{Source: "nope", Path: "/p"}}} + }, + expectErr: `validator "x" references unknown bundle_rule.source "nope", configured bundles: platform`, + }, + { + name: "duplicate bundle id", + mutate: func(c *Config) { + c.OpaBundles = []OpaBundle{ + {Id: "platform", ConfigPath: "/a.yml"}, + {Id: "platform", ConfigPath: "/b.yml"}, + } + }, + expectErr: `duplicate opa_bundle "platform"`, + }, + { + name: "bundle without config_path", + mutate: func(c *Config) { + c.OpaBundles = []OpaBundle{{Id: "platform"}} + }, + expectErr: `opa_bundle "platform" requires config_path`, + }, + { + name: "bundle with unparsable timeout", + mutate: func(c *Config) { + c.OpaBundles = []OpaBundle{{Id: "platform", ConfigPath: "/a.yml", ReadyTimeout: Ptr("soon")}} + }, + expectErr: `opa_bundle "platform": invalid ready_timeout "soon"`, + }, + } + + for _, tc := range tt { + t.Run(tc.name, func(t *testing.T) { + c := DefaultConfig() + tc.mutate(c) + assert.ErrorContains(t, c.Validate(), tc.expectErr) + }) + } +} + +func TestValidateAcceptsSingleBundleWithoutSource(t *testing.T) { + c := DefaultConfig() + c.OpaBundles = []OpaBundle{{Id: "platform", ConfigPath: "/a.yml"}} + c.Validators = []Validator{{Type: "opa_bundle", Name: "x", BundleRule: &BundleRule{Path: "/p"}}} + + assert.NoError(t, c.Validate()) +} + +func TestBundleTimeoutDefaults(t *testing.T) { + b := OpaBundle{Id: "platform", ConfigPath: "/a.yml"} + + ready, err := b.ResolvedReadyTimeout() + require.NoError(t, err) + assert.Equal(t, DefaultBundleReadyTimeout, ready) + + decision, err := b.ResolvedDecisionTimeout() + require.NoError(t, err) + assert.Equal(t, DefaultBundleDecisionTimeout, decision) + + // An explicit zero opts out of the per-decision deadline. + b.DecisionTimeout = Ptr("0s") + decision, err = b.ResolvedDecisionTimeout() + require.NoError(t, err) + assert.Zero(t, decision) +} diff --git a/pkg/o11y/common.yaml b/pkg/o11y/common.yaml index 1b54ccb..e4798fb 100644 --- a/pkg/o11y/common.yaml +++ b/pkg/o11y/common.yaml @@ -16,3 +16,22 @@ groups: The name of the mutator. stability: stable examples: ["inject_otel"] + - id: opa.bundle.source + type: string + brief: > + The id of the opa_bundle block whose OPA instance took the decision. + stability: stable + examples: ["platform"] + - id: opa.decision.path + type: string + brief: > + The bundle decision path that was evaluated. + stability: stable + examples: ["/helloworld"] + - id: opa.decision.outcome + type: string + brief: > + Whether the decision allowed the job, denied it, or could not be + evaluated. + stability: stable + examples: ["allow", "deny", "error"] diff --git a/pkg/o11y/metric.go b/pkg/o11y/metric.go index 5667af3..15d581d 100644 --- a/pkg/o11y/metric.go +++ b/pkg/o11y/metric.go @@ -188,3 +188,47 @@ func (m NacpMutatorMutationCount) Add( attribute.String("mutator.name", mutatorName), )) } + +// An instrument for recording `nacp.opa.decision.duration` +type NacpOpaDecisionDuration struct { + inst metric.Float64Histogram +} + +// Construct a new instrument for measuring `nacp.opa.decision.duration` +func NewNacpOpaDecisionDuration(m metric.Meter) (NacpOpaDecisionDuration, error) { + i, err := m.Float64Histogram( + "nacp.opa.decision.duration", + metric.WithDescription("Duration of a policy decision taken against an OPA bundle."), + metric.WithUnit("s"), + ) + if err != nil { + return NacpOpaDecisionDuration{}, err + } + return NacpOpaDecisionDuration{i}, nil +} + +// Records a new measurement. +func (m NacpOpaDecisionDuration) Record( + ctx context.Context, + value float64, + + // The id of the opa_bundle block whose OPA instance took the decision. + opaBundleSource string, + + // Whether the decision allowed the job, denied it, or could not be evaluated. + opaDecisionOutcome string, + + // The bundle decision path that was evaluated. + opaDecisionPath string, + +) { + + m.inst.Record(ctx, value, metric.WithAttributes( + + attribute.String("opa.bundle.source", opaBundleSource), + + attribute.String("opa.decision.outcome", opaDecisionOutcome), + + attribute.String("opa.decision.path", opaDecisionPath), + )) +} diff --git a/pkg/o11y/nacp.yaml b/pkg/o11y/nacp.yaml index ec48e98..7bf1e0b 100644 --- a/pkg/o11y/nacp.yaml +++ b/pkg/o11y/nacp.yaml @@ -49,3 +49,17 @@ groups: attributes: - ref: mutator.name requirement_level: required + - id: metric.nacp.opa.decision.duration + type: metric + metric_name: nacp.opa.decision.duration + stability: stable + brief: "Duration of a policy decision taken against an OPA bundle." + instrument: histogram + unit: "s" + attributes: + - ref: opa.bundle.source + requirement_level: required + - ref: opa.decision.path + requirement_level: required + - ref: opa.decision.outcome + requirement_level: required diff --git a/templates/registry/go/metric.go.j2 b/templates/registry/go/metric.go.j2 index 4c2b001..3b9f343 100644 --- a/templates/registry/go/metric.go.j2 +++ b/templates/registry/go/metric.go.j2 @@ -71,10 +71,10 @@ func {{metric.metric_name | camel_case }}AttrToAttrs(in []{{metric_name}}Attr) [ } {% endif %} -// Adds an increment to the existing count. -func (m {{ smart_title_case(metric.metric_name) }}) Add( +{{ [metric.instrument | map_text("metric_type_doc")] | comment | trim }} +func (m {{ smart_title_case(metric.metric_name) }}) {{ metric.instrument | map_text("metric_type_method") }}( ctx context.Context, - inc float64, + {{ metric.instrument | map_text("metric_type_value_name") }} float64, {% for attr in metric.attributes | required | attribute_sort %} {{ attr.brief | trim | comment }} {{ attr.name | camel_case }} {{ attr.type | map_text("attribute_type_value")}}, @@ -84,7 +84,7 @@ func (m {{ smart_title_case(metric.metric_name) }}) Add( {% endif %} ) { {# TODO - handle other instrument types. #} - m.inst.Add(ctx, inc, metric.WithAttributes( + m.inst.{{ metric.instrument | map_text("metric_type_method") }}(ctx, {{ metric.instrument | map_text("metric_type_value_name") }}, metric.WithAttributes( {% if metric.attributes | not_required | length > 0 %} append({{metric.metric_name | camel_case }}AttrToAttrs(optAttrs), {% endif %} @@ -97,8 +97,8 @@ func (m {{ smart_title_case(metric.metric_name) }}) Add( )) } {% else %} -func (m {{ smart_title_case(metric.metric_name) }}) Add(ctx context.Context, inc float64) { - (*m.instrument).Add(ctx, inc) +func (m {{ smart_title_case(metric.metric_name) }}) {{ metric.instrument | map_text("metric_type_method") }}(ctx context.Context, {{ metric.instrument | map_text("metric_type_value_name") }} float64) { + m.inst.{{ metric.instrument | map_text("metric_type_method") }}(ctx, {{ metric.instrument | map_text("metric_type_value_name") }}) } {% endif %} diff --git a/templates/registry/go/weaver.yaml b/templates/registry/go/weaver.yaml index 359b571..3240789 100644 --- a/templates/registry/go/weaver.yaml +++ b/templates/registry/go/weaver.yaml @@ -37,3 +37,13 @@ text_maps: boolean[]: "...bool" metric_type_interface: counter: Float64Counter + histogram: Float64Histogram + metric_type_method: + counter: Add + histogram: Record + metric_type_doc: + counter: Adds an increment to the existing count. + histogram: Records a new measurement. + metric_type_value_name: + counter: inc + histogram: value diff --git a/testutil/testutil.go b/testutil/testutil.go index 35f8dbf..b24b967 100644 --- a/testutil/testutil.go +++ b/testutil/testutil.go @@ -1,19 +1,21 @@ package testutil import ( - "bytes" "context" "encoding/json" "fmt" "io" "os" "path" + "path/filepath" "runtime" + "sort" "testing" + "github.com/mxab/nacp/pkg/admissionctrl/opa/bundle" "github.com/mxab/nacp/pkg/admissionctrl/types" - "github.com/open-policy-agent/opa/v1/logging" - "github.com/open-policy-agent/opa/v1/sdk" + "github.com/mxab/nacp/pkg/config" + "github.com/mxab/nacp/pkg/logutil" sdktest "github.com/open-policy-agent/opa/v1/sdk/test" "github.com/hashicorp/nomad/api" @@ -149,47 +151,80 @@ func BaseJob() *api.Job { return job } -func SetupOpa(t *testing.T, policy string) *sdk.OPA { +// SetupOpa starts a bundle server hosting policy and returns a bundle Instance +// built through the same path production uses, so tests cover the real startup +// (readiness channel, slog logger, status listener) rather than a lookalike. +func SetupOpa(t *testing.T, policy string) *bundle.Instance { + t.Helper() + return SetupOpaBundles(t, map[string]string{"test": policy})[0] +} + +// SetupOpaBundles starts one bundle server and one Instance per named policy. +func SetupOpaBundles(t *testing.T, policies map[string]string) []*bundle.Instance { + t.Helper() + + registry := SetupOpaRegistry(t, policies) + + names := make([]string, 0, len(policies)) + for name := range policies { + names = append(names, name) + } + sort.Strings(names) + + instances := make([]*bundle.Instance, 0, len(names)) + for _, name := range names { + instance, err := registry.Get(name) + require.NoError(t, err, "Bundle %q is registered", name) + instances = append(instances, instance) + } + return instances +} + +// SetupOpaRegistry builds a bundle Registry holding one Instance per named +// policy, each backed by its own mock bundle server. +func SetupOpaRegistry(t *testing.T, policies map[string]string) *bundle.Registry { + t.Helper() + + names := make([]string, 0, len(policies)) + for name := range policies { + names = append(names, name) + } + sort.Strings(names) + + configs := make([]config.OpaBundle, 0, len(names)) + for _, name := range names { + configs = append(configs, config.OpaBundle{ + Id: name, + ConfigPath: writeOpaConfig(t, name, policies[name]), + }) + } + + loggerFactory, _ := logutil.NewLoggerFactory(io.Discard, io.Discard, false) + registry, stop, err := bundle.Setup(t.Context(), loggerFactory, configs) + require.NoError(t, err, "No error setting up OPA bundles") + t.Cleanup(stop) + + return registry +} + +// writeOpaConfig starts a mock bundle server for policy and writes an OPA +// configuration pointing at it, returning the config path. +func writeOpaConfig(t *testing.T, name, policy string) string { t.Helper() - ctx := t.Context() - // create a mock HTTP bundle server server, err := sdktest.NewServer(sdktest.MockBundle("/bundles/bundle.tar.gz", map[string]string{ "example.rego": policy, })) require.NoError(t, err, "No error creating mock server") t.Cleanup(server.Stop) - // provide the OPA configuration which specifies - // fetching policy bundles from the mock server - // and logging decisions locally to the console - config := []byte(fmt.Sprintf(`{ - "services": { - "test": { - "url": %q - } - }, - "bundles": { - "test": { - "resource": "/bundles/bundle.tar.gz" - } - }, - "decision_logs": { - "console": true - } - }`, server.URL())) - - // create an instance of the OPA object - - opa, err := sdk.New(ctx, sdk.Options{ - ID: "opa-test-1", - Config: bytes.NewReader(config), - Logger: logging.New(), - }) - require.NoError(t, err, "No error creating OPA instance") - t.Cleanup(func() { - opa.Stop(ctx) - }) - - return opa + opaConfig := fmt.Sprintf(`{ + "services": {%q: {"url": %q}}, + "bundles": {%q: {"service": %q, "resource": "/bundles/bundle.tar.gz"}}, + "decision_logs": {"console": true} + }`, name, server.URL(), name, name) + + configPath := filepath.Join(t.TempDir(), name+".json") + require.NoError(t, os.WriteFile(configPath, []byte(opaConfig), 0o600), "No error writing OPA config") + return configPath }