Skip to content

Feat(opa): implement singleton OPA SDK to eliminate duplicate bundle … #491

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
davidhadas merged 1 commit into from
Jun 11, 2026
Merged

Feat(opa): implement singleton OPA SDK to eliminate duplicate bundle … #491

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions authbridge/authlib/plugins/opa/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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/<agent-id>.tar.gz` from the bundle server.
fetch `bundles?spiffe=<url-encoded-spiffe-id>` 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).
Expand Down Expand Up @@ -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/<agent-id>.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=<url-encoded-spiffe-id>`. 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
Expand Down
206 changes: 182 additions & 24 deletions authbridge/authlib/plugins/opa/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -82,21 +85,100 @@ 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).
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]
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

Expand All @@ -108,8 +190,6 @@ func (p *OPA) Name() string { return "opa" }

func (p *OPA) Capabilities() pipeline.PluginCapabilities {

@huang195 huang195 Jun 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion — Dropping Requires: [jwt-validation] and RequiresAny: [parsers] removes the pipeline's guarantee that identity claims and parsed request data are populated before OPA evaluates. A pipeline without jwt-validation was previously rejected; now OPA runs with missing claims in buildInput, which can silently weaken policy decisions. Worth confirming the default pipeline still orders jwt-validation/parsers ahead of opa, and that policies tolerate absent claims defensively.

@davidhadas davidhadas Jun 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OPA is used also in outbound where jwt-validationis missing
OPA also rely on a2a parser and mcp parser etc. but non is mandatory.
Use cases where OPA is needed although non of the above are used is not impossible.
Therefore failing the agent due to a pipeline config that has OPA without specific other plugins seems counter productive. OPA should use what it has. Rego inside may fail requests or responses inbound or outbound if info is missing on the context (leading to missing info on the input document. This is left for the admins to decide instead of failing the agent and not allowing it to start.

return pipeline.PluginCapabilities{
Requires: []string{"jwt-validation"},
RequiresAny: []string{"a2a-parser", "mcp-parser", "inference-parser"},
Description: "OPA policy enforcement for inbound and outbound requests.",
}
}
Expand Down Expand Up @@ -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")
Expand All @@ -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 {

@huang195 huang195 Jun 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion — This reuse guard compares bundleURL, but the value that actually selects the policy bundle is the agent identity (bundles?spiffe=<agentID>), and different agents share the same bundle_url (the service base). So for a same-URL/different-agent case this warning never fires and the second instance silently reuses the first agent's SDK — meaning agent B would be evaluated against agent A's policy.

That's safe today (one agent per AuthBridge process; inbound + outbound share one agentID), but the guard is checking the wrong field for the invariant it protects. Suggest keying on agentID (or asserting it matches) so that if a process ever serves more than one agent, it fails loud rather than cross-applying policy.

@davidhadas davidhadas Jun 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems that we need to keep it as is - we cant have two different urls on the config. This is what we protect here.

The entire design is for a single identity.
Agent id is not part of the config, it comes from the client id file - there is one file for one agent and therefore one id which is used for both upstream and downstream. This is what we support for now and if this changes, we ned to rethink the OPA singleton use and other aspects of the design.

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
Comment thread
davidhadas marked this conversation as resolved.
}

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):

@huang195 huang195 Jun 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit — This readiness goroutine isn't tied to Shutdown/bgCancel. If the plugin shuts down before the bundle becomes ready, the trailing <-readyCh blocks forever and the goroutine leaks. Consider selecting on a cancel/done channel alongside readyCh.

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
}
Expand All @@ -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{
Expand All @@ -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),

@huang195 huang195 Jun 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (verify before merge) — The bundle resource changes from bundles/<id>.tar.gz to bundles?spiffe=<url-encoded-id>. This is a contract change with the Kagenti Bundle Server (a separate component). If the server isn't already serving the query form, every bundle download 404s → OPA never becomes Ready → OnRequest/OnResponse deny everything. Please confirm the server supports bundles?spiffe= before merge.

Also: the plugin's own README.md:12 still documents bundles/<agent-id>.tar.gz — update it to match the new path.

(Minor: a ? embedded in OPA's bundle resource field is unusual — worth a sanity check that the OPA SDK forwards it to the service URL un-re-escaped.)

@davidhadas davidhadas Jun 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The contract indeed has intentionally changed.
The Readme need to be updated.

"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
Expand All @@ -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 {
Expand All @@ -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",
Expand All @@ -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,
})
Expand Down Expand Up @@ -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",
Expand All @@ -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,
})
Expand Down
Loading
Loading