From 3ada324b8a654499a428da12f84cce39b41bec35 Mon Sep 17 00:00:00 2001 From: David Hadas Date: Sat, 6 Jun 2026 14:25:00 +0300 Subject: [PATCH] feat(opa): implement singleton OPA SDK to eliminate duplicate bundle downloads Previously, the OPA plugin created separate SDK instances for inbound and outbound pipelines, resulting in: - Two bundle downloads from the bundle service - Duplicate memory usage for policy storage - Redundant OPA SDK initialization logs This commit introduces a process-wide singleton pattern that shares a single OPA SDK instance across both pipelines when they use the same bundle_url and agent_id. Key changes: 1. **Singleton OPA SDK Management** - Added `sharedSDK` struct with reference counting - Package-level `singleton` variable protected by `singletonMu` mutex - First plugin instance creates the SDK, subsequent instances reuse it - SDK is stopped only when the last reference is released 2. **OPA Logging Bridge** - Added `slogBridge` to integrate OPA SDK logs into authbridge's slog - OPA bundle downloads, parse errors, and activation now visible in logs - Tagged with "component: opa-sdk" for easy filtering 3. **SPIFFE ID Encoding** - Changed bundle resource path from `bundles/{agent-id}.tar.gz` to `bundles?spiffe={url-encoded-spiffe-id}` - Properly URL-encodes SPIFFE IDs (slashes become %2F) - Strips "spiffe://" prefix before encoding - Supports both SPIFFE ID and legacy agent ID formats 4. **Dependency Relaxation** - Removed hard `Requires: ["jwt-validation"]` dependency - Removed `RequiresAny: ["a2a-parser", "mcp-parser", "inference-parser"]` - OPA can now run independently without requiring other plugins 5. **Test Infrastructure** - Added `resetSingleton()` for test isolation - Added `newTestOPA()` helper to simplify test setup - New tests for singleton reuse, ref counting, and shutdown behavior - Updated bundle resource path tests for SPIFFE ID encoding Benefits: - Single bundle download per agent (50% reduction in network traffic) - Single OPA SDK instance (50% reduction in memory usage) - Cleaner logs with OPA SDK messages properly integrated - More flexible plugin ordering (no hard dependencies) Signed-off-by: David Hadas --- authbridge/authlib/plugins/opa/README.md | 10 +- authbridge/authlib/plugins/opa/plugin.go | 206 +++++++++++-- authbridge/authlib/plugins/opa/plugin_test.go | 283 ++++++++++++------ 3 files changed, 386 insertions(+), 113 deletions(-) diff --git a/authbridge/authlib/plugins/opa/README.md b/authbridge/authlib/plugins/opa/README.md index 8604e154f..62e9e2572 100644 --- a/authbridge/authlib/plugins/opa/README.md +++ b/authbridge/authlib/plugins/opa/README.md @@ -9,7 +9,7 @@ responses against the loaded policy using four fixed decision paths. 1. At startup the plugin reads the agent's client ID from `/shared/client-id.txt` (mounted by the kagenti-operator from a Keycloak-credentials Secret). 2. It creates an embedded OPA engine via the OPA Go SDK and configures it to - fetch `bundles/.tar.gz` from the bundle server. + fetch `bundles?spiffe=` from the bundle server. 3. The SDK downloads the bundle, activates the policy, and begins periodic polling for updates (respecting `ETag` / `If-None-Match` for lightweight 304 responses). @@ -415,9 +415,11 @@ The plugin interprets the decision as follows: ## Bundle layout -The bundle server must serve a standard OPA bundle at the path -`bundles/.tar.gz`. A minimal bundle contains a single `.rego` file -for the inbound request path: +The bundle server must serve a standard OPA bundle at +`bundles?spiffe=`. The SPIFFE ID is read from +`/shared/client-id.txt` (or the `agent_id` config field), stripped of +the `spiffe://` prefix, and URL-encoded. A minimal bundle contains a +single `.rego` file for the inbound request path: ``` bundles/my-agent.tar.gz diff --git a/authbridge/authlib/plugins/opa/plugin.go b/authbridge/authlib/plugins/opa/plugin.go index 0142d4658..fe475b2f8 100644 --- a/authbridge/authlib/plugins/opa/plugin.go +++ b/authbridge/authlib/plugins/opa/plugin.go @@ -10,9 +10,12 @@ import ( "net/http" "net/url" "strings" + "sync" "sync/atomic" + "time" "github.com/open-policy-agent/opa/sdk" + opalog "github.com/open-policy-agent/opa/v1/logging" "github.com/kagenti/kagenti-extensions/authbridge/authlib/config" "github.com/kagenti/kagenti-extensions/authbridge/authlib/pipeline" @@ -82,12 +85,92 @@ func (c *opaConfig) validate() error { return nil } +// slogBridge bridges OPA's logging.Logger interface to Go's slog, so that all +// OPA SDK internal messages (bundle downloads, parse errors, activation) surface +// in the authbridge log stream. +type slogBridge struct { + level opalog.Level + fields map[string]interface{} +} + +func newSlogBridge() *slogBridge { + return &slogBridge{level: opalog.Debug} +} + +func (b *slogBridge) attrs() []slog.Attr { + attrs := make([]slog.Attr, 0, len(b.fields)+1) + attrs = append(attrs, slog.String("component", "opa-sdk")) + for k, v := range b.fields { + attrs = append(attrs, slog.Any(k, v)) + } + return attrs +} + +func (b *slogBridge) Debug(msg string, a ...interface{}) { + if len(a) > 0 { + msg = fmt.Sprintf(msg, a...) + } + slog.LogAttrs(context.Background(), slog.LevelDebug, msg, b.attrs()...) +} + +func (b *slogBridge) Info(msg string, a ...interface{}) { + if len(a) > 0 { + msg = fmt.Sprintf(msg, a...) + } + slog.LogAttrs(context.Background(), slog.LevelInfo, msg, b.attrs()...) +} + +func (b *slogBridge) Warn(msg string, a ...interface{}) { + if len(a) > 0 { + msg = fmt.Sprintf(msg, a...) + } + slog.LogAttrs(context.Background(), slog.LevelWarn, msg, b.attrs()...) +} + +func (b *slogBridge) Error(msg string, a ...interface{}) { + if len(a) > 0 { + msg = fmt.Sprintf(msg, a...) + } + slog.LogAttrs(context.Background(), slog.LevelError, msg, b.attrs()...) +} + +func (b *slogBridge) WithFields(fields map[string]interface{}) opalog.Logger { + cp := &slogBridge{level: b.level, fields: make(map[string]interface{}, len(b.fields)+len(fields))} + for k, v := range b.fields { + cp.fields[k] = v + } + for k, v := range fields { + cp.fields[k] = v + } + return cp +} + +func (b *slogBridge) SetLevel(level opalog.Level) { b.level = level } +func (b *slogBridge) GetLevel() opalog.Level { return b.level } + // decider abstracts OPA decision-making for testability. type decider interface { Decision(ctx context.Context, options sdk.DecisionOptions) (*sdk.DecisionResult, error) Stop(ctx context.Context) } +// sharedSDK holds the process-wide singleton OPA SDK instance. Both inbound +// and outbound plugin instances share one SDK (one bundle download, one +// memory footprint) because the bundle is per-agent, not per-direction. +type sharedSDK struct { + decider decider + ready atomic.Bool + done chan struct{} + bundleURL string + agentID string + refCount int +} + +var ( + singletonMu sync.Mutex + singleton *sharedSDK +) + // OPA evaluates requests against OPA bundles downloaded from a Kagenti // Bundle Server. The bundle resource path is derived from the agent's // identity (/shared/client-id.txt). @@ -95,8 +178,7 @@ type OPA struct { cfg opaConfig inc includeSet agentID string - decider atomic.Pointer[decider] - ready atomic.Bool + shared atomic.Pointer[sharedSDK] bgCancel atomic.Pointer[context.CancelFunc] } @@ -108,8 +190,6 @@ func (p *OPA) Name() string { return "opa" } func (p *OPA) Capabilities() pipeline.PluginCapabilities { return pipeline.PluginCapabilities{ - Requires: []string{"jwt-validation"}, - RequiresAny: []string{"a2a-parser", "mcp-parser", "inference-parser"}, Description: "OPA policy enforcement for inbound and outbound requests.", } } @@ -152,7 +232,12 @@ func (p *OPA) Configure(raw json.RawMessage) error { func (p *OPA) Init(_ context.Context) error { if p.agentID != "" { - return p.startOPA() + slog.Info("opa: initializing with agent_id", "agent_id", p.agentID) + if err := p.startOPA(); err != nil { + slog.Error("opa: initialization failed", "error", err, "agent_id", p.agentID) + return err + } + return nil } if p.cfg.AgentIDFile == "" { return errors.New("opa: no agent_id or agent_id_file configured") @@ -179,24 +264,74 @@ func (p *OPA) Init(_ context.Context) error { } func (p *OPA) startOPA() error { + singletonMu.Lock() + defer singletonMu.Unlock() + + if singleton != nil { + if singleton.bundleURL != p.cfg.BundleURL { + slog.Warn("opa: redundant bundle_url for the opa sdk singleton - second config skipped", + "skipped_bundle_url", p.cfg.BundleURL, + "active_bundle_url", singleton.bundleURL) + } + singleton.refCount++ + p.shared.Store(singleton) + slog.Info("opa: reusing shared OPA SDK singleton", "agent_id", singleton.agentID) + return nil + } + cfgBytes, agentID, err := p.buildOPAConfig() if err != nil { + slog.Error("opa: failed to build OPA config", "error", err) return err } + + slog.Info("opa: starting OPA SDK with config", "config", string(cfgBytes), "agent_id", agentID) + readyCh := make(chan struct{}) + + opaLogger := newSlogBridge() + opa, err := sdk.New(context.Background(), sdk.Options{ - Config: bytes.NewReader(cfgBytes), - Ready: readyCh, + Config: bytes.NewReader(cfgBytes), + Ready: readyCh, + Logger: opaLogger, + ConsoleLogger: opaLogger, + V1Compatible: true, }) if err != nil { + slog.Error("opa: failed to initialize OPA SDK", "error", err, "agent_id", agentID) return fmt.Errorf("opa sdk.New: %w", err) } - var dec decider = opa - p.decider.Store(&dec) + + s := &sharedSDK{ + decider: opa, + done: make(chan struct{}), + bundleURL: p.cfg.BundleURL, + agentID: agentID, + refCount: 1, + } + singleton = s + p.shared.Store(s) + go func() { - <-readyCh - p.ready.Store(true) - slog.Info("opa: bundle loaded and policy activated", "agent_id", agentID) + select { + case <-readyCh: + s.ready.Store(true) + slog.Info("opa: bundle loaded and policy activated", "agent_id", agentID) + return + case <-s.done: + return + case <-time.After(30 * time.Second): + slog.Warn("opa: bundle not yet available after 30s, will keep retrying", + "agent_id", agentID, + "bundle_url", p.cfg.BundleURL) + } + select { + case <-readyCh: + s.ready.Store(true) + slog.Info("opa: bundle loaded and policy activated", "agent_id", agentID) + case <-s.done: + } }() return nil } @@ -208,8 +343,12 @@ func (p *OPA) buildOPAConfig() ([]byte, string, error) { return nil, "", errors.New("agentID is empty") } - // Escape agentID to prevent path traversal and ensure safe URL path segment - escapedAgentID := url.PathEscape(agentID) + // Strip "spiffe://" prefix if present to get the SPIFFE ID for the query parameter + spiffeID := strings.TrimPrefix(agentID, "spiffe://") + + // URL-encode SPIFFE ID for query parameter + // This is REQUIRED because SPIFFE IDs contain '/' which must be encoded as %2F + escapedSPIFFEID := url.QueryEscape(spiffeID) cfg := map[string]any{ "services": map[string]any{ @@ -220,13 +359,19 @@ func (p *OPA) buildOPAConfig() ([]byte, string, error) { "bundles": map[string]any{ "authz": map[string]any{ "service": "kagenti", - "resource": fmt.Sprintf("bundles/%s.tar.gz", escapedAgentID), + "resource": fmt.Sprintf("bundles?spiffe=%s", escapedSPIFFEID), "polling": map[string]any{ "min_delay_seconds": p.cfg.PollingMinDelay, "max_delay_seconds": p.cfg.PollingMaxDelay, }, }, }, + "decision_logs": map[string]any{ + "console": true, + }, + "logging": map[string]any{ + "level": "info", + }, } data, _ := json.Marshal(cfg) return data, agentID, nil @@ -236,14 +381,27 @@ func (p *OPA) Shutdown(ctx context.Context) error { if cancel := p.bgCancel.Swap(nil); cancel != nil { (*cancel)() } - if dec := p.decider.Load(); dec != nil { - (*dec).Stop(ctx) + s := p.shared.Swap(nil) + if s == nil { + return nil + } + singletonMu.Lock() + defer singletonMu.Unlock() + s.refCount-- + if s.refCount <= 0 { + close(s.done) + s.decider.Stop(ctx) + singleton = nil } return nil } func (p *OPA) Ready() bool { - return p.ready.Load() + s := p.shared.Load() + if s == nil { + return false + } + return s.ready.Load() } func (p *OPA) decisionPath(pctx *pipeline.Context, phase string) string { @@ -260,8 +418,8 @@ func (p *OPA) decisionPath(pctx *pipeline.Context, phase string) string { } func (p *OPA) OnRequest(ctx context.Context, pctx *pipeline.Context) pipeline.Action { - dec := p.decider.Load() - if dec == nil || !p.ready.Load() { + s := p.shared.Load() + if s == nil || !s.ready.Load() { pctx.Record(pipeline.Invocation{ Action: pipeline.ActionDeny, Reason: "opa_not_ready", @@ -271,7 +429,7 @@ func (p *OPA) OnRequest(ctx context.Context, pctx *pipeline.Context) pipeline.Ac path := p.decisionPath(pctx, "request") input := buildInput(pctx, p.inc) - result, err := (*dec).Decision(ctx, sdk.DecisionOptions{ + result, err := s.decider.Decision(ctx, sdk.DecisionOptions{ Path: path, Input: input, }) @@ -308,8 +466,8 @@ func (p *OPA) OnRequest(ctx context.Context, pctx *pipeline.Context) pipeline.Ac } func (p *OPA) OnResponse(ctx context.Context, pctx *pipeline.Context) pipeline.Action { - dec := p.decider.Load() - if dec == nil || !p.ready.Load() { + s := p.shared.Load() + if s == nil || !s.ready.Load() { pctx.Record(pipeline.Invocation{ Action: pipeline.ActionDeny, Reason: "opa_not_ready", @@ -323,7 +481,7 @@ func (p *OPA) OnResponse(ctx context.Context, pctx *pipeline.Context) pipeline.A "status_code": pctx.StatusCode, "headers": flattenHeaders(pctx.ResponseHeaders), } - result, err := (*dec).Decision(ctx, sdk.DecisionOptions{ + result, err := s.decider.Decision(ctx, sdk.DecisionOptions{ Path: path, Input: input, }) diff --git a/authbridge/authlib/plugins/opa/plugin_test.go b/authbridge/authlib/plugins/opa/plugin_test.go index 6dcbf748d..00750c1bd 100644 --- a/authbridge/authlib/plugins/opa/plugin_test.go +++ b/authbridge/authlib/plugins/opa/plugin_test.go @@ -42,6 +42,22 @@ func (c *capturingDecider) Decision(_ context.Context, opts sdk.DecisionOptions) func (c *capturingDecider) Stop(_ context.Context) {} +// newTestOPA creates an OPA plugin instance with a mock shared SDK for testing. +func newTestOPA(dec decider, inc includeSet) *OPA { + s := &sharedSDK{decider: dec, done: make(chan struct{})} + s.ready.Store(true) + p := &OPA{inc: inc} + p.shared.Store(s) + return p +} + +// resetSingleton clears the package-level singleton for test isolation. +func resetSingleton() { + singletonMu.Lock() + singleton = nil + singletonMu.Unlock() +} + // --- Configure tests --- func TestConfigure_MissingBundleURL(t *testing.T) { @@ -176,26 +192,12 @@ func TestCapabilities(t *testing.T) { p := &OPA{} caps := p.Capabilities() - expectedRequires := []string{"jwt-validation"} - if len(caps.Requires) != len(expectedRequires) { - t.Fatalf("expected Requires=%v, got %v", expectedRequires, caps.Requires) - } - for i, v := range expectedRequires { - if caps.Requires[i] != v { - t.Errorf("Requires[%d]: expected %q, got %q", i, v, caps.Requires[i]) - } - } - - expectedRequiresAny := []string{"a2a-parser", "mcp-parser", "inference-parser"} - if len(caps.RequiresAny) != len(expectedRequiresAny) { - t.Fatalf("expected RequiresAny=%v, got %v", expectedRequiresAny, caps.RequiresAny) + if len(caps.Requires) != 0 { + t.Errorf("expected no hard Requires, got %v", caps.Requires) } - for i, v := range expectedRequiresAny { - if caps.RequiresAny[i] != v { - t.Errorf("RequiresAny[%d]: expected %q, got %q", i, v, caps.RequiresAny[i]) - } + if len(caps.RequiresAny) != 0 { + t.Errorf("expected no hard RequiresAny, got %v", caps.RequiresAny) } - if caps.Description != "OPA policy enforcement for inbound and outbound requests." { t.Errorf("unexpected Description: %q", caps.Description) } @@ -899,11 +901,7 @@ func TestOnRequest_NotReady(t *testing.T) { func TestOnRequest_Allow(t *testing.T) { var dec decider = &mockDecider{result: &sdk.DecisionResult{Result: true}} - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Inbound, Method: "GET", @@ -921,11 +919,7 @@ func TestOnRequest_Allow(t *testing.T) { func TestOnRequest_Deny(t *testing.T) { var dec decider = &mockDecider{result: &sdk.DecisionResult{Result: false}} - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Inbound, Method: "GET", @@ -946,11 +940,7 @@ func TestOnRequest_Deny(t *testing.T) { func TestOnRequest_DecisionError(t *testing.T) { var dec decider = &mockDecider{err: fmt.Errorf("bundle not loaded")} - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Inbound, Method: "GET", @@ -968,11 +958,7 @@ func TestOnRequest_DecisionError(t *testing.T) { func TestOnRequest_UndefinedSkips(t *testing.T) { var dec decider = &mockDecider{err: undefinedErr()} - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Inbound, Method: "GET", @@ -994,11 +980,7 @@ func TestOnRequest_OutboundUsesCorrectPath(t *testing.T) { result: &sdk.DecisionResult{Result: true}, capture: &capturedPath, } - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Outbound, Method: "GET", @@ -1020,11 +1002,7 @@ func TestOnRequest_InboundUsesCorrectPath(t *testing.T) { result: &sdk.DecisionResult{Result: true}, capture: &capturedPath, } - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Inbound, Method: "POST", @@ -1064,11 +1042,7 @@ func TestOnResponse_NotReady(t *testing.T) { func TestOnResponse_Allow(t *testing.T) { var dec decider = &mockDecider{result: &sdk.DecisionResult{Result: true}} - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Inbound, Method: "GET", @@ -1088,11 +1062,7 @@ func TestOnResponse_Allow(t *testing.T) { func TestOnResponse_Deny(t *testing.T) { var dec decider = &mockDecider{result: &sdk.DecisionResult{Result: false}} - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Inbound, Method: "GET", @@ -1111,11 +1081,7 @@ func TestOnResponse_Deny(t *testing.T) { func TestOnResponse_UndefinedSkips(t *testing.T) { var dec decider = &mockDecider{err: undefinedErr()} - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Outbound, Method: "GET", @@ -1138,11 +1104,7 @@ func TestOnResponse_OutboundUsesCorrectPath(t *testing.T) { result: &sdk.DecisionResult{Result: true}, capture: &capturedPath, } - p := &OPA{ - inc: newIncludeSet(nil), - } - p.decider.Store(&dec) - p.ready.Store(true) + p := newTestOPA(dec, newIncludeSet(nil)) pctx := &pipeline.Context{ Direction: pipeline.Outbound, Method: "GET", @@ -1168,14 +1130,14 @@ func TestBuildOPAConfig(t *testing.T) { PollingMinDelay: 10, PollingMaxDelay: 120, }, - agentID: "my-agent", + agentID: "spiffe://mybest.com/ns/team1/sa/test-agent", } data, agentID, err := p.buildOPAConfig() if err != nil { t.Fatalf("buildOPAConfig failed: %v", err) } - if agentID != "my-agent" { - t.Errorf("expected agentID 'my-agent', got %q", agentID) + if agentID != "spiffe://mybest.com/ns/team1/sa/test-agent" { + t.Errorf("expected agentID 'spiffe://mybest.com/ns/team1/sa/test-agent', got %q", agentID) } var cfg map[string]any if err := json.Unmarshal(data, &cfg); err != nil { @@ -1188,11 +1150,13 @@ func TestBuildOPAConfig(t *testing.T) { } bundles := cfg["bundles"].(map[string]any) authz := bundles["authz"].(map[string]any) - if authz["resource"] != "bundles/my-agent.tar.gz" { - t.Errorf("expected bundles/my-agent.tar.gz, got %v", authz["resource"]) + // Verify URL-encoded SPIFFE ID in resource (without spiffe:// prefix) + expected := "bundles?spiffe=mybest.com%2Fns%2Fteam1%2Fsa%2Ftest-agent" + if authz["resource"] != expected { + t.Errorf("expected %s, got %v", expected, authz["resource"]) } } -func TestBuildOPAConfig_PathEscaping(t *testing.T) { +func TestBuildOPAConfig_SPIFFEIDEncoding(t *testing.T) { tests := []struct { name string agentID string @@ -1200,28 +1164,40 @@ func TestBuildOPAConfig_PathEscaping(t *testing.T) { wantResourcePath string }{ { - name: "valid agentID", - agentID: "my-agent", + name: "SPIFFE ID with namespace and service account", + agentID: "spiffe://mybest.com/ns/team1/sa/badass", wantErr: false, - wantResourcePath: "bundles/my-agent.tar.gz", + wantResourcePath: "bundles?spiffe=mybest.com%2Fns%2Fteam1%2Fsa%2Fbadass", }, { - name: "agentID with forward slash gets escaped", - agentID: "../evil", + name: "SPIFFE ID without prefix (legacy format)", + agentID: "mybest.com/ns/team1/sa/badass", wantErr: false, - wantResourcePath: "bundles/..%2Fevil.tar.gz", + wantResourcePath: "bundles?spiffe=mybest.com%2Fns%2Fteam1%2Fsa%2Fbadass", }, { - name: "agentID with backslash gets escaped", - agentID: "..\\evil", + name: "simple SPIFFE ID", + agentID: "spiffe://example.com/workload", wantErr: false, - wantResourcePath: "bundles/..%5Cevil.tar.gz", + wantResourcePath: "bundles?spiffe=example.com%2Fworkload", }, { - name: "agentID with special chars", + name: "SPIFFE ID with hyphens", + agentID: "spiffe://example.com/ns/my-namespace/sa/my-service", + wantErr: false, + wantResourcePath: "bundles?spiffe=example.com%2Fns%2Fmy-namespace%2Fsa%2Fmy-service", + }, + { + name: "legacy agentID without slashes", + agentID: "my-agent", + wantErr: false, + wantResourcePath: "bundles?spiffe=my-agent", + }, + { + name: "agentID with special chars gets encoded", agentID: "agent@domain.com", wantErr: false, - wantResourcePath: "bundles/agent@domain.com.tar.gz", + wantResourcePath: "bundles?spiffe=agent%40domain.com", }, { name: "empty agentID", @@ -1259,3 +1235,140 @@ func TestBuildOPAConfig_PathEscaping(t *testing.T) { }) } } + +// --- Singleton tests --- + +func TestStartOPA_SingletonReuse(t *testing.T) { + resetSingleton() + defer resetSingleton() + + // Seed the singleton with a mock decider as if the first instance started. + var dec decider = &mockDecider{result: &sdk.DecisionResult{Result: true}} + singletonMu.Lock() + singleton = &sharedSDK{ + decider: dec, + done: make(chan struct{}), + bundleURL: "http://bundle-service:8080", + agentID: "spiffe://example.com/ns/team1/sa/agent1", + refCount: 1, + } + singleton.ready.Store(true) + singletonMu.Unlock() + + // Second instance with the same bundle_url should reuse the singleton. + p := &OPA{ + cfg: opaConfig{BundleURL: "http://bundle-service:8080"}, + agentID: "spiffe://example.com/ns/team1/sa/agent1", + } + if err := p.startOPA(); err != nil { + t.Fatalf("startOPA failed: %v", err) + } + if p.shared.Load() != singleton { + t.Fatal("expected second instance to share the singleton") + } + if singleton.refCount != 2 { + t.Errorf("expected refCount=2, got %d", singleton.refCount) + } + if !p.Ready() { + t.Fatal("expected Ready() true from shared singleton") + } +} + +func TestStartOPA_SingletonDifferentBundleURL(t *testing.T) { + resetSingleton() + defer resetSingleton() + + // Seed the singleton with bundle_url "http://first-service:8080". + var dec decider = &mockDecider{result: &sdk.DecisionResult{Result: true}} + singletonMu.Lock() + singleton = &sharedSDK{ + decider: dec, + done: make(chan struct{}), + bundleURL: "http://first-service:8080", + agentID: "spiffe://example.com/ns/team1/sa/agent1", + refCount: 1, + } + singleton.ready.Store(true) + singletonMu.Unlock() + + // Second instance with a DIFFERENT bundle_url should still reuse the + // singleton (warning logged, second config skipped). + p := &OPA{ + cfg: opaConfig{BundleURL: "http://second-service:9090"}, + agentID: "spiffe://example.com/ns/team1/sa/agent1", + } + if err := p.startOPA(); err != nil { + t.Fatalf("startOPA failed: %v", err) + } + if p.shared.Load() != singleton { + t.Fatal("expected second instance to reuse singleton despite different bundle_url") + } + if singleton.refCount != 2 { + t.Errorf("expected refCount=2, got %d", singleton.refCount) + } + // The singleton should retain the FIRST bundle_url. + if singleton.bundleURL != "http://first-service:8080" { + t.Errorf("expected singleton to keep first bundle_url, got %q", singleton.bundleURL) + } +} + +func TestShutdown_RefCounting(t *testing.T) { + resetSingleton() + defer resetSingleton() + + stopped := false + var dec decider = &mockDecider{result: &sdk.DecisionResult{Result: true}} + // Override Stop to track when it's called. + trackingDec := &trackingStopDecider{decider: dec, stopped: &stopped} + singletonMu.Lock() + singleton = &sharedSDK{ + decider: trackingDec, + done: make(chan struct{}), + bundleURL: "http://bundle-service:8080", + agentID: "agent1", + refCount: 2, + } + singleton.ready.Store(true) + singletonMu.Unlock() + + p1 := &OPA{} + p1.shared.Store(singleton) + p2 := &OPA{} + p2.shared.Store(singleton) + + // First shutdown decrements refCount but does not stop the SDK. + if err := p1.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown p1 failed: %v", err) + } + if stopped { + t.Fatal("SDK should not be stopped after first Shutdown") + } + if singleton == nil { + t.Fatal("singleton should still exist after first Shutdown") + } + + // Second shutdown stops the SDK and clears the singleton. + if err := p2.Shutdown(context.Background()); err != nil { + t.Fatalf("Shutdown p2 failed: %v", err) + } + if !stopped { + t.Fatal("SDK should be stopped after last Shutdown") + } + if singleton != nil { + t.Fatal("singleton should be nil after last Shutdown") + } +} + +// trackingStopDecider wraps a decider and records when Stop is called. +type trackingStopDecider struct { + decider decider + stopped *bool +} + +func (d *trackingStopDecider) Decision(ctx context.Context, opts sdk.DecisionOptions) (*sdk.DecisionResult, error) { + return d.decider.Decision(ctx, opts) +} + +func (d *trackingStopDecider) Stop(_ context.Context) { + *d.stopped = true +}